jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-01 06:51:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

style(workflow): rustfmt the multi-tenant fix set

Browse Source 
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 <tlongwell@block.xyz>
Signed-off-by: Tyler Longwell <tlongwell@block.xyz>
This commit is contained in:
npub1qyvc0c5kl4gqv2fd97fsk46tu378sqgy35vc83rvgfwne90sel7s0ed67d
2026-06-27 15:26:18 -04:00
parent ea687ccb33
commit 08a0399923
11 changed files with 136 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
crates/buzz-admin/src/main.rs
View File

@@ -189,7 +189,8 @@ async fn cmd_remove_member(pubkey_arg: String, role_filter: Option<String>) -> 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 {

10
crates/buzz-core/src/tenant.rs
View File

@@ -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");
}

25
crates/buzz-db/src/lib.rs
View File

@@ -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<Option<String>> {
pub async fn lookup_community_host(&self, community_id: CommunityId) -> Result<Option<String>> {
let row = sqlx::query(
r#"
SELECT host
@@ -1598,8 +1595,13 @@ impl Db {
workflow_id: Uuid,
scheduled_for: chrono::DateTime<chrono::Utc>,
) -> Result<Option<workflow::ScheduledWorkflowFireClaim>> {
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<bool> {
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).

22
crates/buzz-db/src/relay_members.rs
View File

@@ -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<bool> {
pub async fn is_relay_member(pool: &PgPool, community: CommunityId, pubkey: &str) -> Result<bool> {
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<Vec<RelayMember>> {
pub async fn list_relay_members(pool: &PgPool, community: CommunityId) -> Result<Vec<RelayMember>> {
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);

83
crates/buzz-db/src/workflow.rs
View File

@@ -936,8 +936,15 @@ pub async fn update_approval(
note: Option<&str>,
) -> Result<bool> {
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,

4
crates/buzz-relay/src/api/bridge.rs
View File

@@ -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((

4
crates/buzz-relay/src/api/mod.rs
View File

@@ -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,

4
crates/buzz-relay/src/handlers/command_executor.rs
View File

@@ -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

2
crates/buzz-relay/src/main.rs
View File

@@ -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!(

9
crates/buzz-workflow/src/executor.rs
View File

@@ -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;

34
crates/buzz-workflow/src/lib.rs
View File

@@ -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<DateTime<Utc>>,
now: DateTime<Utc>,
) -> 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<Utc>, workflow_id: Uuid) -> Option<DateTime<Utc>> {
fn interval_fire_instant(
dur: &str,
now: DateTime<Utc>,
workflow_id: Uuid,
) -> Option<DateTime<Utc>> {
match executor::parse_duration_secs(dur) {
Ok(interval_secs) if interval_secs > 0 => {
let secs = interval_secs as i64;