diff --git a/crates/buzz-cli/src/commands/workflows.rs b/crates/buzz-cli/src/commands/workflows.rs index ac3f876dc2..a0c20710da 100644 --- a/crates/buzz-cli/src/commands/workflows.rs +++ b/crates/buzz-cli/src/commands/workflows.rs @@ -181,7 +181,7 @@ async fn current_workflow_revision( /// Trigger a workflow — sign and submit a kind:46020 event. /// /// When `inputs` is provided, it is parsed as a JSON object and used as the -/// event content (MCP parity). When omitted, the event content is `{}`. +/// event content (MCP parity). When omitted, the event content is empty. pub async fn cmd_trigger_workflow( client: &BuzzClient, workflow_id: &str, @@ -190,37 +190,28 @@ pub async fn cmd_trigger_workflow( let wf_uuid = parse_uuid(workflow_id)?; let revision = current_workflow_revision(client, workflow_id).await?; + let builder = workflow_trigger_builder(wf_uuid, &revision, inputs)?; + let event = client.sign_event(builder)?; + let resp = client.submit_event(event).await?; + println!("{}", normalize_write_response(&resp)); + Ok(()) +} + +fn workflow_trigger_builder( + workflow_id: uuid::Uuid, + revision: &str, + inputs: Option<&str>, +) -> Result { if let Some(raw) = inputs { - // Parse and validate it is a JSON object, then build the event manually - // so we can embed the inputs as the event content. let parsed: serde_json::Value = serde_json::from_str(raw) .map_err(|e| CliError::Usage(format!("--inputs is not valid JSON: {e}")))?; - if !parsed.is_object() { - return Err(CliError::Usage("--inputs must be a JSON object".into())); - } - let content = serde_json::to_string(&parsed).unwrap_or_default(); - use nostr::{EventBuilder, Kind, Tag}; - let tags = vec![ - Tag::parse(["d", &wf_uuid.to_string()]) - .map_err(|e| CliError::Other(format!("tag error: {e}")))?, - Tag::parse(["e", revision.as_str()]) - .map_err(|e| CliError::Other(format!("tag error: {e}")))?, - ]; - let builder = EventBuilder::new( - Kind::Custom(buzz_sdk::kind::KIND_WORKFLOW_TRIGGER as u16), - &content, - ) - .tags(tags); - let event = client.sign_event(builder)?; - let resp = client.submit_event(event).await?; - println!("{}", normalize_write_response(&resp)); + let object = parsed + .as_object() + .ok_or_else(|| CliError::Usage("--inputs must be a JSON object".into()))?; + buzz_sdk::build_workflow_trigger_with_inputs(workflow_id, revision, object).map_err(sdk_err) } else { - let builder = buzz_sdk::build_workflow_trigger(wf_uuid, &revision).map_err(sdk_err)?; - let event = client.sign_event(builder)?; - let resp = client.submit_event(event).await?; - println!("{}", normalize_write_response(&resp)); + buzz_sdk::build_workflow_trigger(workflow_id, revision).map_err(sdk_err) } - Ok(()) } /// Approve or deny a workflow step — sign and submit a kind:46030 (grant) or 46031 (deny) event. @@ -275,3 +266,32 @@ pub async fn dispatch(cmd: crate::WorkflowsCmd, client: &BuzzClient) -> Result<( } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn trigger_builders_preserve_inputs_and_distinguish_same_second_operations() { + let keys = nostr::Keys::generate(); + let workflow_id = uuid::Uuid::new_v4(); + let revision = "ab".repeat(32); + let timestamp = nostr::Timestamp::from(1_700_000_000); + for inputs in [None, Some(r#"{"message":"hello"}"#)] { + let build = || { + workflow_trigger_builder(workflow_id, &revision, inputs) + .unwrap() + .custom_created_at(timestamp) + .sign_with_keys(&keys) + .unwrap() + }; + let first = build(); + let second = build(); + assert_ne!(first.id, second.id); + assert_eq!(first.content, inputs.unwrap_or("")); + } + for invalid in ["not-json", "[]", "null", "1"] { + assert!(workflow_trigger_builder(workflow_id, &revision, Some(invalid)).is_err()); + } + } +} diff --git a/crates/buzz-db/src/store/channel.rs b/crates/buzz-db/src/store/channel.rs index 12065c3a06..18800fb80a 100644 --- a/crates/buzz-db/src/store/channel.rs +++ b/crates/buzz-db/src/store/channel.rs @@ -21,10 +21,11 @@ pub use buzz_core::channel::{ChannelType, ChannelVisibility, MemberRole}; // Keep the established channel module paths compatible while membership SQL // and invariants live in their dedicated store module. pub use crate::channel_members::{ - acquire_channel_membership_lock, add_member, get_accessible_channel_ids, get_accessible_channels, get_bot_members, - get_member_count, get_member_counts_bulk, get_member_role, get_member_role_in_transaction, get_members, get_members_bulk, - get_users_bulk, is_member, list_large_channel_rosters_needing_reconciliation, - lock_member_snapshot, membership_pairs, remove_member, verify_channel_roster_fence_behavior, + acquire_channel_membership_lock, add_member, get_accessible_channel_ids, + get_accessible_channels, get_bot_members, get_member_count, get_member_counts_bulk, + get_member_role, get_member_role_in_transaction, get_members, get_members_bulk, get_users_bulk, + is_member, list_large_channel_rosters_needing_reconciliation, lock_member_snapshot, + membership_pairs, remove_member, verify_channel_roster_fence_behavior, verify_channel_roster_fence_catalog, AccessibleChannel, BotChannelEntry, BotMemberRecord, LargeChannelRoster, LockedMemberSnapshot, MemberRecord, UserRecord, }; diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index 8443f2872f..92bde63857 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/crates/buzz-db/src/store/channel_members.rs @@ -1519,6 +1519,29 @@ impl Db { } } +/// Get an active member role using the caller's transaction. +/// +/// Authorization callers must acquire [`acquire_channel_membership_lock`] first +/// and hold the transaction through the dependent mutation. +pub async fn get_member_role_in_transaction( + tx: &mut Transaction<'_, Postgres>, + community_id: CommunityId, + channel_id: Uuid, + pubkey: &[u8], +) -> Result> { + let row = sqlx::query( + "SELECT cm.role::text AS role FROM channel_members cm \ + JOIN channels c ON cm.community_id = c.community_id AND cm.channel_id = c.id AND c.deleted_at IS NULL \ + WHERE cm.community_id = $1 AND cm.channel_id = $2 AND cm.pubkey = $3 AND cm.removed_at IS NULL", + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .bind(pubkey) + .fetch_optional(&mut **tx) + .await?; + Ok(row.map(|r| r.try_get("role")).transpose()?) +} + #[cfg(test)] mod postgres_tests { use super::*; @@ -3210,27 +3233,3 @@ mod postgres_tests { drop_scratch_db(&admin, pool, &scratch_name).await; } } - -/// Get an active member role using the caller's transaction. -/// -/// Authorization callers must acquire [`acquire_channel_membership_lock`] first -/// and hold the transaction through the dependent mutation. -pub async fn get_member_role_in_transaction( - tx: &mut Transaction<'_, Postgres>, - community_id: CommunityId, - channel_id: Uuid, - pubkey: &[u8], -) -> Result> { - let row = sqlx::query( - "SELECT cm.role::text AS role FROM channel_members cm \ - JOIN channels c ON cm.community_id = c.community_id AND cm.channel_id = c.id AND c.deleted_at IS NULL \ - WHERE cm.community_id = $1 AND cm.channel_id = $2 AND cm.pubkey = $3 AND cm.removed_at IS NULL", - ) - .bind(community_id.as_uuid()) - .bind(channel_id) - .bind(pubkey) - .fetch_optional(&mut **tx) - .await?; - Ok(row.map(|r| r.try_get("role")).transpose()?) -} - diff --git a/crates/buzz-db/src/store/workflow.rs b/crates/buzz-db/src/store/workflow.rs index 45c42a7d29..6cded51f1b 100644 --- a/crates/buzz-db/src/store/workflow.rs +++ b/crates/buzz-db/src/store/workflow.rs @@ -906,6 +906,24 @@ pub async fn create_workflow_run( Ok(id) } +/// Find the committed result of a manual trigger within its tenant and workflow. +/// The command event and run commit atomically, so a retry can recover this ID. +pub async fn get_workflow_run_id_by_trigger( + pool: &PgPool, + community_id: CommunityId, + workflow_id: Uuid, + trigger_event_id: &[u8], +) -> Result> { + Ok(sqlx::query_scalar( + "SELECT id FROM workflow_runs WHERE community_id = $1 AND workflow_id = $2 AND trigger_event_id = $3", + ) + .bind(community_id.as_uuid()) + .bind(workflow_id) + .bind(trigger_event_id) + .fetch_optional(pool) + .await?) +} + /// Fetch a single workflow run by ID, scoped to its community. pub async fn get_workflow_run( pool: &PgPool, @@ -1352,6 +1370,24 @@ pub async fn find_by_owner_and_name( // -- Run and approval Db API -------------------------------------------------- impl Db { + /// Recover the committed run ID for an exact manual-trigger retry. + #[datastore_span(name = "get_workflow_run_id_by_trigger", system = "postgresql")] + pub async fn get_workflow_run_id_by_trigger( + &self, + community_id: CommunityId, + workflow_id: Uuid, + trigger_event_id: &[u8], + ) -> Result> { + crate::workflow::get_workflow_run_id_by_trigger( + &self.pool, + community_id, + workflow_id, + trigger_event_id, + ) + .await + } + + /// Fetch and share-lock one workflow on an existing transaction. #[datastore_span(name = "get_workflow_for_share_in_transaction", system = "postgresql")] pub async fn get_workflow_for_share_in_transaction( diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 399e120e26..fc14f0c78a 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -991,11 +991,17 @@ async fn handle_workflow_trigger( // triggers as global events leaks workflow IDs to unrelated relay members. let mut tx = match persist_command_event(&state.db, tenant, event, workflow.channel_id).await? { PersistResult::Duplicate => { - return Ok(IngestResult { - event_id: event.id.to_hex(), - accepted: true, - message: "duplicate: already processed".into(), - }); + // A lost response does not lose the result: event and run committed + // together. Read from the primary, including after a concurrent retry. + let run_id = state + .db + .get_workflow_run_id_by_trigger(community_id, workflow_id, event.id.as_bytes()) + .await + .map_err(|e| IngestError::Internal(format!("error: recover workflow run: {e}")))? + .ok_or_else(|| { + IngestError::Internal("error: stored workflow trigger has no run".into()) + })?; + return Ok(workflow_trigger_response(event, run_id)); } PersistResult::Inserted(tx) => tx, }; @@ -1116,8 +1122,12 @@ async fn handle_workflow_trigger( .await; }); - // 6. Return response - Ok(IngestResult { + // 6. Return the same result shape for initial delivery and exact replay. + Ok(workflow_trigger_response(event, run_id)) +} + +fn workflow_trigger_response(event: &Event, run_id: Uuid) -> IngestResult { + IngestResult { event_id: event.id.to_hex(), accepted: true, message: format!( @@ -1126,7 +1136,7 @@ async fn handle_workflow_trigger( "run_id": run_id.to_string(), }) ), - }) + } } /// Enforce the approver_spec field against the requesting pubkey. @@ -1739,12 +1749,30 @@ mod postgres_tests { keys: &Keys, workflow_id: Uuid, revision: &str, + ) -> Event { + workflow_trigger_event_at( + keys, + workflow_id, + revision, + Timestamp::now().as_secs(), + &Uuid::new_v4().to_string(), + ) + } + + fn workflow_trigger_event_at( + keys: &Keys, + workflow_id: Uuid, + revision: &str, + created_at: u64, + request_id: &str, ) -> Event { EventBuilder::new(Kind::Custom(KIND_WORKFLOW_TRIGGER as u16), "") .tags(vec![ Tag::parse(["d", workflow_id.to_string().as_str()]).expect("d tag"), Tag::parse(["e", revision]).expect("revision tag"), + Tag::parse(["request-id", request_id]).expect("request identity"), ]) + .custom_created_at(Timestamp::from(created_at)) .sign_with_keys(keys) .expect("sign workflow trigger") } @@ -2042,44 +2070,7 @@ mod postgres_tests { )); } - #[tokio::test] - #[ignore = "requires Postgres"] - async fn concurrent_human_owner_manual_triggers_do_not_starve_one_connection_pool() { - let (state, tenant, human, _agent, workflow_id, revision) = - manual_trigger_test_context().await; - let triggers = (0..8) - .map(|_| workflow_trigger_event(&human, workflow_id, &revision)) - .collect::>(); - - let results = tokio::time::timeout(std::time::Duration::from_secs(8), async { - let mut tasks = tokio::task::JoinSet::new(); - for trigger in triggers { - let state = Arc::clone(&state); - let tenant = tenant.clone(); - let auth = http_auth(&human); - tasks.spawn(async move { - handle_workflow_trigger(&tenant, &state, &trigger, &auth).await - }); - } - let mut results = Vec::new(); - while let Some(result) = tasks.join_next().await { - results.push(result.expect("trigger task must not panic")); - } - results - }) - .await - .expect("concurrent triggers must drain rather than pool-starve"); - - assert_eq!(results.len(), 8); - for result in results { - assert!( - result - .expect("concurrent human-owner trigger must succeed") - .accepted, - "manual trigger should be accepted" - ); - } - } + mod trigger_delivery; #[test] fn workflow_revision_parser_accepts_create_and_valid_update() { diff --git a/crates/buzz-relay/src/handlers/command_executor/tests/trigger_delivery.rs b/crates/buzz-relay/src/handlers/command_executor/tests/trigger_delivery.rs new file mode 100644 index 0000000000..7617cf77bf --- /dev/null +++ b/crates/buzz-relay/src/handlers/command_executor/tests/trigger_delivery.rs @@ -0,0 +1,205 @@ +use super::*; +use std::collections::{HashMap, HashSet}; + +fn response_run_id(result: &IngestResult) -> Uuid { + assert!(result.accepted); + let response: serde_json::Value = serde_json::from_str( + result + .message + .strip_prefix("response:") + .expect("new run response"), + ) + .expect("response JSON"); + Uuid::parse_str(response["run_id"].as_str().expect("run ID")).expect("run UUID") +} + +async fn assert_persisted_trigger_runs( + state: &AppState, + tenant: &TenantContext, + workflow_id: Uuid, + revision: &Event, + triggers: &[Event], + run_ids: &HashSet, +) { + let runs = state + .db + .list_workflow_runs(tenant.community(), workflow_id, 100) + .await + .expect("persisted runs"); + assert_eq!( + runs.len(), + triggers.len(), + "exactly one run per signed operation" + ); + assert_eq!( + run_ids.len(), + triggers.len(), + "distinct acknowledged run IDs" + ); + assert_eq!( + runs.iter().map(|run| run.id).collect::>(), + *run_ids + ); + let trigger_ids = triggers + .iter() + .map(|trigger| trigger.id.as_bytes().to_vec()) + .collect::>(); + assert_eq!( + trigger_ids.len(), + triggers.len(), + "fixture requests must be distinct" + ); + assert_eq!( + runs.iter() + .map(|run| run.trigger_event_id.clone().expect("trigger association")) + .collect::>(), + trigger_ids + ); + for run in runs { + assert_eq!( + run.definition_event_id.as_deref(), + Some(revision.id.as_bytes().as_slice()) + ); + } + for trigger in triggers { + assert_eq!( + state + .db + .get_workflow_run_id_by_trigger( + tenant.community(), + Uuid::new_v4(), + trigger.id.as_bytes(), + ) + .await + .expect("workflow-scoped lookup"), + None + ); + let stored = state + .db + .get_event_by_id(tenant.community(), trigger.id.as_bytes()) + .await + .expect("trigger remains stored"); + assert!( + stored.is_some(), + "later triggers must not replace earlier operations" + ); + } +} + +async fn deliver_and_replay(same_second: bool) { + let (state, tenant, human, _agent, workflow_id, revision) = manual_trigger_test_context().await; + let now = Timestamp::now().as_secs(); + let mut triggers = vec![ + workflow_trigger_event_at(&human, workflow_id, &revision.id.to_hex(), now, "first"), + workflow_trigger_event_at( + &human, + workflow_id, + &revision.id.to_hex(), + if same_second { now } else { now - 1 }, + "second", + ), + ]; + if same_second { + // A lower ID wins a NIP-33 timestamp tie. Deliver it first so accidental + // coordinate replacement would incorrectly suppress the second request. + triggers.sort_by_key(|event| event.id); + } + assert_ne!(triggers[0].id, triggers[1].id); + assert_eq!( + triggers[0].created_at == triggers[1].created_at, + same_second + ); + let mut run_ids = HashSet::new(); + let mut responses = HashMap::new(); + for trigger in &triggers { + let result = handle_workflow_trigger(&tenant, &state, trigger, &http_auth(&human)) + .await + .expect("distinct trigger accepted"); + assert!(run_ids.insert(response_run_id(&result))); + responses.insert(trigger.id, result.message); + } + // Replay both, including the earlier operation after another has committed. + for trigger in &triggers { + let replay = handle_workflow_trigger(&tenant, &state, trigger, &http_auth(&human)) + .await + .expect("exact replay accepted"); + assert!(replay.accepted); + assert_eq!( + replay.message, responses[&trigger.id], + "retry recovers the original result" + ); + } + assert_persisted_trigger_runs(&state, &tenant, workflow_id, &revision, &triggers, &run_ids) + .await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn manual_triggers_delivered_newer_then_older_create_distinct_runs() { + deliver_and_replay(false).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn manual_triggers_in_one_second_create_distinct_runs() { + deliver_and_replay(true).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn concurrent_human_owner_manual_triggers_do_not_starve_one_connection_pool() { + let (state, tenant, human, _agent, workflow_id, revision) = manual_trigger_test_context().await; + let now = Timestamp::now().as_secs(); + let triggers = (0..8) + .map(|i| { + workflow_trigger_event_at( + &human, + workflow_id, + &revision.id.to_hex(), + now, + &i.to_string(), + ) + }) + .collect::>(); + let mut deliveries = triggers.clone(); + // Concurrent exact retries must not create additional runs either. + deliveries.extend_from_slice(&triggers[..2]); + let results = tokio::time::timeout(std::time::Duration::from_secs(8), async { + let mut tasks = tokio::task::JoinSet::new(); + for trigger in deliveries { + let state = Arc::clone(&state); + let tenant = tenant.clone(); + let auth = http_auth(&human); + tasks.spawn( + async move { handle_workflow_trigger(&tenant, &state, &trigger, &auth).await }, + ); + } + let mut results = Vec::new(); + while let Some(result) = tasks.join_next().await { + results.push( + result + .expect("trigger task must not panic") + .expect("trigger succeeds"), + ); + } + results + }) + .await + .expect("concurrent triggers must drain rather than pool-starve"); + assert_eq!(results.len(), 10); + assert!(results.iter().all(|result| result.accepted)); + let mut acknowledged = HashMap::new(); + for result in &results { + let run_id = response_run_id(result); + if let Some(previous) = acknowledged.insert(result.event_id.clone(), run_id) { + assert_eq!( + previous, run_id, + "concurrent retry returns the original run" + ); + } + } + assert_eq!(acknowledged.len(), 8); + let run_ids = acknowledged.into_values().collect::>(); + assert_persisted_trigger_runs(&state, &tenant, workflow_id, &revision, &triggers, &run_ids) + .await; +} diff --git a/crates/buzz-sdk/src/builders.rs b/crates/buzz-sdk/src/builders.rs index 4d7599cc84..c6845ece8d 100644 --- a/crates/buzz-sdk/src/builders.rs +++ b/crates/buzz-sdk/src/builders.rs @@ -1659,16 +1659,41 @@ pub fn build_workflow_delete( } /// Build a workflow trigger event (kind 46020) bound to an exact signed revision. +/// +/// Each call creates a new operation, even within the same second. Retain and +/// resubmit the signed event when retrying that operation; do not rebuild it. pub fn build_workflow_trigger( workflow_id: Uuid, definition_event_id: &str, +) -> Result { + workflow_trigger_builder(workflow_id, definition_event_id, "") +} + +/// Build a distinct workflow trigger operation with JSON object inputs. +/// Retry by resubmitting the signed event, as with [`build_workflow_trigger`]. +pub fn build_workflow_trigger_with_inputs( + workflow_id: Uuid, + definition_event_id: &str, + inputs: &serde_json::Map, +) -> Result { + let content = serde_json::Value::Object(inputs.clone()).to_string(); + check_content(&content, 64 * 1024)?; + workflow_trigger_builder(workflow_id, definition_event_id, &content) +} + +fn workflow_trigger_builder( + workflow_id: Uuid, + definition_event_id: &str, + content: &str, ) -> Result { let revision = check_hex_exact(definition_event_id, 64, "definition_event_id")?; let tags = vec![ tag(&["d", &workflow_id.to_string()])?, tag(&["e", &revision])?, + // Event IDs hash the unsigned fields, not the randomized signature. + tag(&["request-id", &Uuid::new_v4().to_string()])?, ]; - Ok(EventBuilder::new(Kind::Custom(KIND_WORKFLOW_TRIGGER as u16), "").tags(tags)) + Ok(EventBuilder::new(Kind::Custom(KIND_WORKFLOW_TRIGGER as u16), content).tags(tags)) } /// Build a workflow approval event — kind 46030 (grant) or 46031 (deny). @@ -4127,6 +4152,34 @@ mod tests { assert!(build_workflow_trigger(wid, "not-an-event-id").is_err()); } + #[test] + fn workflow_trigger_invocations_in_one_second_have_distinct_ids() { + let keys = nostr::Keys::generate(); + let workflow_id = uuid(); + let revision = "ab".repeat(32); + let timestamp = nostr::Timestamp::from(1_700_000_000); + let first = build_workflow_trigger(workflow_id, &revision).unwrap(); + let replay = first + .clone() + .custom_created_at(timestamp) + .sign_with_keys(&keys) + .unwrap(); + let first = first + .custom_created_at(timestamp) + .sign_with_keys(&keys) + .unwrap(); + let second = build_workflow_trigger(workflow_id, &revision) + .unwrap() + .custom_created_at(timestamp) + .sign_with_keys(&keys) + .unwrap(); + assert_eq!(first.id, replay.id, "retry the same built operation"); + assert_ne!( + first.id, second.id, + "new invocations must not become replays" + ); + } + #[test] fn workflow_approval_grant() { let hash = "a".repeat(64); diff --git a/desktop/src-tauri/src/events/workflows.rs b/desktop/src-tauri/src/events/workflows.rs index b73f8fca4f..8d29639c31 100644 --- a/desktop/src-tauri/src/events/workflows.rs +++ b/desktop/src-tauri/src/events/workflows.rs @@ -36,12 +36,10 @@ pub fn build_workflow_trigger( workflow_id: &str, definition_event_id: &str, ) -> Result { - EventId::from_hex(definition_event_id).map_err(|_| "invalid workflow revision".to_string())?; - let tags = vec![ - tag(vec!["d", workflow_id])?, - tag(vec!["e", definition_event_id])?, - ]; - Ok(EventBuilder::new(Kind::Custom(46020), "").tags(tags)) + let workflow_id = + uuid::Uuid::parse_str(workflow_id).map_err(|_| "invalid workflow id".to_string())?; + buzz_sdk_pkg::build_workflow_trigger(workflow_id, definition_event_id) + .map_err(|error| error.to_string()) } /// Kind 46030 — grant an approval token (with optional note). @@ -55,3 +53,24 @@ pub fn build_approval_deny(token: &str, note: Option<&str>) -> Result