Merge pull request #395 from code0-tech/#381-update-flow-export

added update and delete to flow export

Author: raphael-goetz

src/sagittarius/flow_service_client_impl/dev_export.rs

@@ -2,44 +2,73 @@
 //! JSON file, so a developer can inspect what Sagittarius last synced
 //! without needing NATS/KV tooling. Only called when running with
 //! `environment == DEVELOPMENT`.
+//!
+//! The destination is `config.static_config.flow_path` — the same path
+//! static mode loads from at startup — so a development export can be fed
+//! straight back in as a static-mode fallback.
 
 use std::path::Path;
 
 use tokio::fs;
-use tucana::shared::Flows;
+use tucana::shared::{Flows, ValidationFlow};
 
-const EXPORT_PATH: &str = "flowExport.json";
-const EXPORT_TMP_PATH: &str = "flowExport.json.tmp";
+/// Reads the current export at `path`, treating a missing or unparsable
+/// file as an empty set so callers can always fall back to a fresh export.
+async fn read_flows(path: &str) -> Flows {
+    match fs::read(path).await {
+        Ok(bytes) => match serde_json::from_slice(&bytes) {
+            Ok(flows) => flows,
+            Err(error) => {
+                log::warn!(
+                    "Failed to parse existing development flow export path={} error={:?}; starting from an empty export",
+                    path,
+                    error
+                );
+                Flows::default()
+            }
+        },
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Flows::default(),
+        Err(error) => {
+            log::warn!(
+                "Failed to read existing development flow export path={} error={}; starting from an empty export",
+                path,
+                error
+            );
+            Flows::default()
+        }
+    }
+}
 
-/// Writes `flows` to `flowExport.json`, replacing any previous export.
+/// Writes `flows` to `path`, replacing any previous export.
 ///
-/// Goes through a temp file plus rename so a reader never observes a
-/// partially written export. If the rename fails because the destination
-/// already exists, the destination is removed and the rename retried once
-/// before giving up.
-pub(super) async fn overwrite(flows: Flows) {
-    let json = match serde_json::to_vec_pretty(&flows) {
+/// Goes through a `path`-adjacent temp file plus rename so a reader never
+/// observes a partially written export. If the rename fails because the
+/// destination already exists, the destination is removed and the rename
+/// retried once before giving up. Returns whether the write succeeded.
+async fn write(path: &str, flows: &Flows) -> bool {
+    let json = match serde_json::to_vec_pretty(flows) {
         Ok(bytes) => bytes,
         Err(error) => {
             log::error!(
                 "Failed to serialize development flow export flow_count={} error={:?}",
                 flows.flows.len(),
                 error
             );
-            return;
+            return false;
         }
     };
 
-    let final_path = Path::new(EXPORT_PATH);
-    let tmp_path = Path::new(EXPORT_TMP_PATH);
+    let final_path = Path::new(path);
+    let tmp_path_buf = format!("{path}.tmp");
+    let tmp_path = Path::new(&tmp_path_buf);
 
     if let Err(error) = fs::write(tmp_path, &json).await {
         log::error!(
             "Failed to write development flow export path={} error={}",
             tmp_path.display(),
             error
         );
-        return;
+        return false;
     }
 
     if let Err(error) = fs::rename(tmp_path, final_path).await {
@@ -75,13 +104,61 @@ pub(super) async fn overwrite(flows: Flows) {
                     cleanup_error
                 );
             }
-            return;
+            return false;
         }
     }
 
-    log::info!(
-        "Exported {} flows to {}",
-        flows.flows.len(),
-        final_path.display()
-    );
+    true
+}
+
+/// Overwrites `path` with `flows` wholesale.
+pub(super) async fn overwrite(path: &str, flows: Flows) {
+    let flow_count = flows.flows.len();
+    if write(path, &flows).await {
+        log::info!("Exported {} flows to {}", flow_count, path);
+    }
+}
+
+/// Adds or replaces a single flow in the export at `path`, read-modify-write.
+pub(super) async fn upsert_flow(path: &str, flow: ValidationFlow) {
+    let mut flows = read_flows(path).await;
+    let flow_id = flow.flow_id;
+
+    match flows.flows.iter_mut().find(|f| f.flow_id == flow_id) {
+        Some(existing) => *existing = flow,
+        None => flows.flows.push(flow),
+    }
+
+    if write(path, &flows).await {
+        log::info!(
+            "Upserted flow_id={} into development flow export path={}",
+            flow_id,
+            path
+        );
+    }
+}
+
+/// Removes a single flow from the export at `path` by id, read-modify-write.
+/// Logs a warning (without failing) if no matching flow was found.
+pub(super) async fn remove_flow(path: &str, flow_id: i64) {
+    let mut flows = read_flows(path).await;
+    let original_count = flows.flows.len();
+    flows.flows.retain(|f| f.flow_id != flow_id);
+
+    if flows.flows.len() == original_count {
+        log::warn!(
+            "Development flow export had no flow matching flow_id={} to remove path={}",
+            flow_id,
+            path
+        );
+        return;
+    }
+
+    if write(path, &flows).await {
+        log::info!(
+            "Removed flow_id={} from development flow export path={}",
+            flow_id,
+            path
+        );
+    }
 }

src/sagittarius/flow_service_client_impl/mod.rs

@@ -4,7 +4,8 @@
 //! along the way.
 //!
 //! - [`flow_store`] applies the sync operations (delete/replace/update) to the KV store.
-//! - [`dev_export`] mirrors the synced flows to a local JSON file, development only.
+//! - [`dev_export`] mirrors the synced flows to a local JSON file, development only,
+//!   updating it incrementally for single-flow updates/deletes and wholesale on replace.
 
 mod dev_export;
 mod flow_store;
@@ -38,6 +39,9 @@ pub struct SagittariusFlowClient {
     client: FlowServiceClient<Channel>,
     env: String,
     token: String,
+    /// Path to mirror synced flows to when running in development, same
+    /// path static mode loads its fallback export from.
+    flow_export_path: String,
     /// Flipped to `true` once the sync stream is established, so other
     /// components can hold off on work that depends on Sagittarius state
     /// actually being loaded.
@@ -50,6 +54,7 @@ impl SagittariusFlowClient {
         store: Arc<async_nats::jetstream::kv::Store>,
         env: String,
         token: String,
+        flow_export_path: String,
         channel: Channel,
         sagittarius_ready: Arc<AtomicBool>,
         action_config_tx: broadcast::Sender<tucana::shared::ModuleConfigurations>,
@@ -61,6 +66,7 @@ impl SagittariusFlowClient {
             client,
             env,
             token,
+            flow_export_path,
             sagittarius_ready,
             action_config_tx,
         }
@@ -83,6 +89,11 @@ impl SagittariusFlowClient {
         match data {
             Data::DeletedFlowId(id) => {
                 log::debug!("Applying flow deletion flow_id={}", id);
+
+                if self.is_development() {
+                    dev_export::remove_flow(&self.flow_export_path, id).await;
+                }
+
                 let deleted_count = flow_store::delete_flow(&self.store, id).await;
 
                 if deleted_count == 0 {
@@ -99,6 +110,11 @@ impl SagittariusFlowClient {
             }
             Data::UpdatedFlow(flow) => {
                 let flow_id = flow.flow_id;
+
+                if self.is_development() {
+                    dev_export::upsert_flow(&self.flow_export_path, flow.clone()).await;
+                }
+
                 let (key, result) = flow_store::store_flow(&self.store, &flow).await;
                 match result {
                     Ok(()) => {
@@ -124,7 +140,7 @@ impl SagittariusFlowClient {
                 );
 
                 if self.is_development() {
-                    dev_export::overwrite(flows.clone()).await;
+                    dev_export::overwrite(&self.flow_export_path, flows.clone()).await;
                 }
 
                 let (purged_count, stored_count) = flow_store::replace_all(&self.store, flows).await;

src/startup/dynamic_mode.rs

@@ -80,6 +80,7 @@ pub async fn run(
 
     let backend_url_for_flow = config.dynamic_config.backend_url.clone();
     let runtime_token_for_flow = config.dynamic_config.backend_token.clone();
+    let flow_export_path_for_flow = config.static_config.flow_path.clone();
     let sagittarius_ready_for_flow = app_readiness.sagittarius_ready.clone();
 
     let env = match config.environment {
@@ -146,6 +147,7 @@ pub async fn run(
                 kv_for_flow.clone(),
                 env.clone(),
                 runtime_token_for_flow.clone(),
+                flow_export_path_for_flow.clone(),
                 ch,
                 sagittarius_ready_for_flow.clone(),
                 action_config_tx.clone(),