From 4569a6838ae26c5a46d92e08a0da32d3a9ce17df Mon Sep 17 00:00:00 2001 From: Raphael Date: Mon, 27 Jul 2026 14:38:27 +0200 Subject: [PATCH] feat: added update and delete to flow export --- .../flow_service_client_impl/dev_export.rs | 117 +++++++++++++++--- .../flow_service_client_impl/mod.rs | 20 ++- src/startup/dynamic_mode.rs | 2 + 3 files changed, 117 insertions(+), 22 deletions(-) diff --git a/src/sagittarius/flow_service_client_impl/dev_export.rs b/src/sagittarius/flow_service_client_impl/dev_export.rs index 5ed66b4..e0d7603 100644 --- a/src/sagittarius/flow_service_client_impl/dev_export.rs +++ b/src/sagittarius/flow_service_client_impl/dev_export.rs @@ -2,23 +2,51 @@ //! 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!( @@ -26,12 +54,13 @@ pub(super) async fn overwrite(flows: Flows) { 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!( @@ -39,7 +68,7 @@ pub(super) async fn overwrite(flows: Flows) { 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 + ); + } } diff --git a/src/sagittarius/flow_service_client_impl/mod.rs b/src/sagittarius/flow_service_client_impl/mod.rs index 7ba3848..1fe790f 100644 --- a/src/sagittarius/flow_service_client_impl/mod.rs +++ b/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, 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, env: String, token: String, + flow_export_path: String, channel: Channel, sagittarius_ready: Arc, action_config_tx: broadcast::Sender, @@ -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; diff --git a/src/startup/dynamic_mode.rs b/src/startup/dynamic_mode.rs index 732b28a..e20bf63 100644 --- a/src/startup/dynamic_mode.rs +++ b/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(),