From 08a0399923fa10205a91c413ff0479a83a5b198c Mon Sep 17 00:00:00 2001 From: npub1qyvc0c5kl4gqv2fd97fsk46tu378sqgy35vc83rvgfwne90sel7s0ed67d <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@sprout-oss.stage.blox.sqprod.co> Date: Sat, 27 Jun 2026 15:26:18 -0400 Subject: [PATCH] style(workflow): rustfmt the multi-tenant fix set Apply cargo fmt to the 11 files touched by the 9 reviewed multi-tenant fix commits. CI Rust Lint (just fmt-check) flagged formatting drift my local clippy --lib --tests run did not exercise; base a7d3817481 was fmt-clean, so this only normalizes lines my own commits introduced. No logic change. Co-authored-by: Tyler Longwell Signed-off-by: Tyler Longwell --- crates/buzz-admin/src/main.rs | 3 +- crates/buzz-core/src/tenant.rs | 10 ++- crates/buzz-db/src/lib.rs | 25 ++++-- crates/buzz-db/src/relay_members.rs | 22 ++--- crates/buzz-db/src/workflow.rs | 83 +++++++++++++++---- crates/buzz-relay/src/api/bridge.rs | 4 +- crates/buzz-relay/src/api/mod.rs | 4 +- .../src/handlers/command_executor.rs | 4 +- crates/buzz-relay/src/main.rs | 2 +- crates/buzz-workflow/src/executor.rs | 9 +- crates/buzz-workflow/src/lib.rs | 34 ++++---- 11 files changed, 136 insertions(+), 64 deletions(-) diff --git a/crates/buzz-admin/src/main.rs b/crates/buzz-admin/src/main.rs index fd6075c25d..f04960f65e 100644 --- a/crates/buzz-admin/src/main.rs +++ b/crates/buzz-admin/src/main.rs @@ -189,7 +189,8 @@ async fn cmd_remove_member(pubkey_arg: String, role_filter: Option) -> R db.remove_relay_member_if_role(tenant.community(), &pubkey_hex, role) .await } else { - db.remove_relay_member(tenant.community(), &pubkey_hex).await + db.remove_relay_member(tenant.community(), &pubkey_hex) + .await }; match result { diff --git a/crates/buzz-core/src/tenant.rs b/crates/buzz-core/src/tenant.rs index 81317a8bc0..f7894a6999 100644 --- a/crates/buzz-core/src/tenant.rs +++ b/crates/buzz-core/src/tenant.rs @@ -248,8 +248,14 @@ mod tests { fn relay_url_authority_collapses_default_ports() { // Default ports collapse to the bare host, matching how an inbound // `Host` header for the same deployment normalizes. - assert_eq!(relay_url_authority("wss://relay.example:443"), "relay.example"); - assert_eq!(relay_url_authority("ws://relay.example:80"), "relay.example"); + assert_eq!( + relay_url_authority("wss://relay.example:443"), + "relay.example" + ); + assert_eq!( + relay_url_authority("ws://relay.example:80"), + "relay.example" + ); assert_eq!(relay_url_authority("wss://relay.example"), "relay.example"); } diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 2876d33ec1..04a4dfd8dd 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -268,10 +268,7 @@ impl Db { /// fan out under *that* community rather than the deployment default. The /// community is authoritative; the host is read back for labelling only and /// is never used to re-derive the community. - pub async fn lookup_community_host( - &self, - community_id: CommunityId, - ) -> Result> { + pub async fn lookup_community_host(&self, community_id: CommunityId) -> Result> { let row = sqlx::query( r#" SELECT host @@ -1598,8 +1595,13 @@ impl Db { workflow_id: Uuid, scheduled_for: chrono::DateTime, ) -> Result> { - workflow::claim_scheduled_workflow_fire(&self.pool, community_id, workflow_id, scheduled_for) - .await + workflow::claim_scheduled_workflow_fire( + &self.pool, + community_id, + workflow_id, + scheduled_for, + ) + .await } /// Fetch the latest claimed schedule instant for interval trigger anchoring. @@ -1794,8 +1796,15 @@ impl Db { approver_pubkey: Option<&[u8]>, note: Option<&str>, ) -> Result { - workflow::update_approval(&self.pool, community_id, token, status, approver_pubkey, note) - .await + workflow::update_approval( + &self.pool, + community_id, + token, + status, + approver_pubkey, + note, + ) + .await } /// Update an approval by its already-hashed token (no re-hashing). diff --git a/crates/buzz-db/src/relay_members.rs b/crates/buzz-db/src/relay_members.rs index ce363b3a6a..e4999ac1ba 100644 --- a/crates/buzz-db/src/relay_members.rs +++ b/crates/buzz-db/src/relay_members.rs @@ -28,11 +28,7 @@ pub struct RelayMember { } /// Returns `true` if `pubkey` (64-char hex) is a member of `community`. -pub async fn is_relay_member( - pool: &PgPool, - community: CommunityId, - pubkey: &str, -) -> Result { +pub async fn is_relay_member(pool: &PgPool, community: CommunityId, pubkey: &str) -> Result { let row = sqlx::query("SELECT 1 FROM relay_members WHERE community_id = $1 AND pubkey = $2") .bind(community.as_uuid()) .bind(pubkey) @@ -70,10 +66,7 @@ pub async fn get_relay_member( } /// Returns all relay members of `community` ordered by `created_at` ascending. -pub async fn list_relay_members( - pool: &PgPool, - community: CommunityId, -) -> Result> { +pub async fn list_relay_members(pool: &PgPool, community: CommunityId) -> Result> { let rows = sqlx::query( "SELECT pubkey, role, added_by, created_at, updated_at \ FROM relay_members WHERE community_id = $1 ORDER BY created_at ASC", @@ -315,12 +308,11 @@ pub async fn backfill_from_allowlist(pool: &PgPool, community: CommunityId) -> R // Only backfill if this community's relay_members is empty — once it has // rows (from a previous backfill or manual admin commands), we must not // re-add members that were intentionally removed. - let has_members: bool = sqlx::query_scalar( - "SELECT EXISTS (SELECT 1 FROM relay_members WHERE community_id = $1)", - ) - .bind(community.as_uuid()) - .fetch_one(pool) - .await?; + let has_members: bool = + sqlx::query_scalar("SELECT EXISTS (SELECT 1 FROM relay_members WHERE community_id = $1)") + .bind(community.as_uuid()) + .fetch_one(pool) + .await?; if has_members { return Ok(0); diff --git a/crates/buzz-db/src/workflow.rs b/crates/buzz-db/src/workflow.rs index b0332b2b7c..0e00a76ea4 100644 --- a/crates/buzz-db/src/workflow.rs +++ b/crates/buzz-db/src/workflow.rs @@ -936,8 +936,15 @@ pub async fn update_approval( note: Option<&str>, ) -> Result { let token_hash = hash_approval_token(token); - update_approval_by_stored_hash(pool, community_id, &token_hash, status, approver_pubkey, note) - .await + update_approval_by_stored_hash( + pool, + community_id, + &token_hash, + status, + approver_pubkey, + note, + ) + .await } /// Update an approval by its already-hashed token value. @@ -1961,10 +1968,22 @@ mod tests { // Same workflow UUID and channel UUID in both communities. let shared_workflow_id = Uuid::new_v4(); let shared_channel_id = Uuid::new_v4(); - insert_workflow_with_ids(&pool, community_a, shared_workflow_id, shared_channel_id, "wf-A") - .await; - insert_workflow_with_ids(&pool, community_b, shared_workflow_id, shared_channel_id, "wf-B") - .await; + insert_workflow_with_ids( + &pool, + community_a, + shared_workflow_id, + shared_channel_id, + "wf-A", + ) + .await; + insert_workflow_with_ids( + &pool, + community_b, + shared_workflow_id, + shared_channel_id, + "wf-B", + ) + .await; // Scoped get returns each community's own row, never the other's. let from_a = get_workflow(&pool, community_a, shared_workflow_id) @@ -1973,9 +1992,15 @@ mod tests { let from_b = get_workflow(&pool, community_b, shared_workflow_id) .await .expect("B's workflow exists"); - assert_eq!(from_a.community_id, community_a, "A lookup must resolve A's row"); + assert_eq!( + from_a.community_id, community_a, + "A lookup must resolve A's row" + ); assert_eq!(from_a.name, "wf-A"); - assert_eq!(from_b.community_id, community_b, "B lookup must resolve B's row"); + assert_eq!( + from_b.community_id, community_b, + "B lookup must resolve B's row" + ); assert_eq!(from_b.name, "wf-B"); // A workflow that exists ONLY in B must be NotFound under A. @@ -1993,7 +2018,11 @@ mod tests { let listed_a = list_enabled_channel_workflows(&pool, community_a, shared_channel_id) .await .expect("list A"); - assert_eq!(listed_a.len(), 1, "A's channel listing must contain exactly A's workflow"); + assert_eq!( + listed_a.len(), + 1, + "A's channel listing must contain exactly A's workflow" + ); assert_eq!(listed_a[0].community_id, community_a); assert_eq!(listed_a[0].name, "wf-A"); } @@ -2010,10 +2039,22 @@ mod tests { let community_a = make_community(&pool).await; let community_b = make_community(&pool).await; let shared_workflow_id = Uuid::new_v4(); - insert_workflow_with_ids(&pool, community_a, shared_workflow_id, Uuid::new_v4(), "wf-A") - .await; - insert_workflow_with_ids(&pool, community_b, shared_workflow_id, Uuid::new_v4(), "wf-B") - .await; + insert_workflow_with_ids( + &pool, + community_a, + shared_workflow_id, + Uuid::new_v4(), + "wf-A", + ) + .await; + insert_workflow_with_ids( + &pool, + community_b, + shared_workflow_id, + Uuid::new_v4(), + "wf-B", + ) + .await; delete_workflow(&pool, community_a, shared_workflow_id) .await @@ -2094,8 +2135,12 @@ mod tests { .expect("create approval B"); // Scoped read returns each community's own approval (its own run id). - let read_a = get_approval(&pool, community_a, token).await.expect("read A"); - let read_b = get_approval(&pool, community_b, token).await.expect("read B"); + let read_a = get_approval(&pool, community_a, token) + .await + .expect("read A"); + let read_b = get_approval(&pool, community_b, token) + .await + .expect("read B"); assert_eq!(read_a.run_id, run_a, "A read must resolve A's approval"); assert_eq!(read_b.run_id, run_b, "B read must resolve B's approval"); @@ -2113,8 +2158,12 @@ mod tests { .expect("grant A"); assert!(granted, "A's approval must be granted"); - let after_a = get_approval(&pool, community_a, token).await.expect("re-read A"); - let after_b = get_approval(&pool, community_b, token).await.expect("re-read B"); + let after_a = get_approval(&pool, community_a, token) + .await + .expect("re-read A"); + let after_b = get_approval(&pool, community_b, token) + .await + .expect("re-read B"); assert_eq!(after_a.status, ApprovalStatus::Granted, "A is now granted"); assert_eq!( after_b.status, diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 0279d79255..bc80195b9a 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -1102,7 +1102,9 @@ pub async fn workflow_webhook( None, ) .await; - engine.finalize_run(community_id, run_id, result, None).await; + engine + .finalize_run(community_id, run_id, result, None) + .await; }); Ok(( diff --git a/crates/buzz-relay/src/api/mod.rs b/crates/buzz-relay/src/api/mod.rs index 856c167112..10a88002ba 100644 --- a/crates/buzz-relay/src/api/mod.rs +++ b/crates/buzz-relay/src/api/mod.rs @@ -86,9 +86,7 @@ pub mod relay_members { .db .is_relay_member(community, &owner_hex) .await - .map_err(|e| { - format!("relay membership check (owner) failed: {e}") - })?; + .map_err(|e| format!("relay membership check (owner) failed: {e}"))?; if owner_is_member { debug!( agent = %pubkey_hex, diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 2bd881f739..0a7ab09d3e 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -809,7 +809,9 @@ async fn handle_workflow_trigger( None, ) .await; - engine.finalize_run(community_id, run_id, result, None).await; + engine + .finalize_run(community_id, run_id, result, None) + .await; }); // 6. Return response diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index ed64859bb7..e14bff9022 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -591,7 +591,7 @@ async fn main() -> anyhow::Result<()> { ) .await { - Ok(true) => {} // We won the claim — proceed to publish. + Ok(true) => {} // We won the claim — proceed to publish. Ok(false) => continue, // Another pod claimed it; no side effect here. Err(e) => { warn!( diff --git a/crates/buzz-workflow/src/executor.rs b/crates/buzz-workflow/src/executor.rs index e98a1e70af..feceec07b7 100644 --- a/crates/buzz-workflow/src/executor.rs +++ b/crates/buzz-workflow/src/executor.rs @@ -1149,7 +1149,14 @@ async fn execute_steps( .unwrap_or(engine.config.default_timeout_secs); let dispatch_result = tokio::time::timeout( std::time::Duration::from_secs(timeout_secs), - dispatch_action(&step.id, &resolved_action, engine, community_id, run_id, trigger_ctx), + dispatch_action( + &step.id, + &resolved_action, + engine, + community_id, + run_id, + trigger_ctx, + ), ) .await; diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index 1bd6f2aca2..2befec6240 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -334,7 +334,9 @@ impl WorkflowEngine { let result = executor::execute_run(&engine, community_id, run_id, &def_clone, &ctx_clone) .await; - engine.finalize_run(community_id, run_id, result, None).await; + engine + .finalize_run(community_id, run_id, result, None) + .await; }); } @@ -366,14 +368,7 @@ impl WorkflowEngine { last: Option>, now: DateTime, ) -> bool { - interval_prefilter_should_fire( - &self.last_fired, - community_id, - workflow_id, - dur, - last, - now, - ) + interval_prefilter_should_fire(&self.last_fired, community_id, workflow_id, dur, last, now) } /// Background loop for scheduled (cron/interval) triggers. @@ -614,10 +609,17 @@ impl WorkflowEngine { let def_clone = def.clone(); let ctx_clone = trigger_ctx.clone(); tokio::spawn(async move { - let result = - executor::execute_run(&engine, community_id, run_id, &def_clone, &ctx_clone) - .await; - engine.finalize_run(community_id, run_id, result, None).await; + let result = executor::execute_run( + &engine, + community_id, + run_id, + &def_clone, + &ctx_clone, + ) + .await; + engine + .finalize_run(community_id, run_id, result, None) + .await; }); } @@ -676,7 +678,11 @@ fn cron_fire_instant( /// so they collide on one `(community, workflow, scheduled_for)` claim — only /// one wins and creates the run. Returns `None` if the duration is unparseable /// or non-positive (the caller skips firing). -fn interval_fire_instant(dur: &str, now: DateTime, workflow_id: Uuid) -> Option> { +fn interval_fire_instant( + dur: &str, + now: DateTime, + workflow_id: Uuid, +) -> Option> { match executor::parse_duration_secs(dur) { Ok(interval_secs) if interval_secs > 0 => { let secs = interval_secs as i64;