From 08020bebca345084a96c1e256e44ee8f12a6fc3c Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Tue, 25 Aug 2026 22:27:01 -0400 Subject: [PATCH] feat(workflows): bind owner triggers to exact signed revisions Reconcile the complete Node A observable behavior onto current main without carrying its historical merge topology. Co-authored-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> Signed-off-by: Logan Johnson --- crates/buzz-cli/src/commands/workflows.rs | 27 +- crates/buzz-db/src/channel.rs | 26 +- crates/buzz-db/src/event.rs | 24 + crates/buzz-db/src/lib.rs | 53 +- crates/buzz-db/src/migration.rs | 17 +- crates/buzz-db/src/workflow.rs | 290 ++++++++- crates/buzz-relay/src/api/bridge.rs | 59 +- crates/buzz-relay/src/api/workflows.rs | 66 +- .../src/handlers/command_executor.rs | 571 ++++++++++++++---- crates/buzz-relay/src/router.rs | 4 + crates/buzz-sdk/src/builders.rs | 17 +- .../tests/e2e_workflow_agent_owner.rs | 205 +++++++ crates/buzz-workflow/src/executor.rs | 4 + crates/buzz-workflow/src/lib.rs | 287 +++++++-- desktop/src-tauri/src/commands/workflows.rs | 8 +- desktop/src-tauri/src/events.rs | 25 + desktop/src-tauri/src/events/workflows.rs | 13 +- .../0033_workflow_definition_event_id.sql | 6 + .../0034_workflow_run_definition_event_id.sql | 5 + schema/schema.sql | 8 + 20 files changed, 1486 insertions(+), 229 deletions(-) create mode 100644 crates/buzz-test-client/tests/e2e_workflow_agent_owner.rs create mode 100644 migrations/0033_workflow_definition_event_id.sql create mode 100644 migrations/0034_workflow_run_definition_event_id.sql diff --git a/crates/buzz-cli/src/commands/workflows.rs b/crates/buzz-cli/src/commands/workflows.rs index 0028dfc7663..ac3f876dc2f 100644 --- a/crates/buzz-cli/src/commands/workflows.rs +++ b/crates/buzz-cli/src/commands/workflows.rs @@ -162,6 +162,22 @@ pub async fn cmd_delete_workflow(client: &BuzzClient, workflow_id: &str) -> Resu Ok(()) } +async fn current_workflow_revision( + client: &BuzzClient, + workflow_id: &str, +) -> Result { + let response = client + .get_authed(&format!("/workflows/{workflow_id}/revision")) + .await?; + let event: serde_json::Value = serde_json::from_str(&response) + .map_err(|e| CliError::Other(format!("invalid workflow revision response: {e}")))?; + event + .get("id") + .and_then(|id| id.as_str()) + .map(str::to_owned) + .ok_or_else(|| CliError::NotFound(format!("workflow {workflow_id} not found"))) +} + /// 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 @@ -172,6 +188,7 @@ pub async fn cmd_trigger_workflow( inputs: Option<&str>, ) -> Result<(), CliError> { let wf_uuid = parse_uuid(workflow_id)?; + let revision = current_workflow_revision(client, workflow_id).await?; if let Some(raw) = inputs { // Parse and validate it is a JSON object, then build the event manually @@ -183,8 +200,12 @@ pub async fn cmd_trigger_workflow( } 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}")))?]; + 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, @@ -194,7 +215,7 @@ pub async fn cmd_trigger_workflow( let resp = client.submit_event(event).await?; println!("{}", normalize_write_response(&resp)); } else { - let builder = buzz_sdk::build_workflow_trigger(wf_uuid).map_err(sdk_err)?; + 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)); diff --git a/crates/buzz-db/src/channel.rs b/crates/buzz-db/src/channel.rs index 98790e3d623..77eefefa6fc 100644 --- a/crates/buzz-db/src/channel.rs +++ b/crates/buzz-db/src/channel.rs @@ -473,7 +473,7 @@ pub async fn verify_channel_roster_fence_behavior(pool: &sqlx::PgPool) -> Result /// Take the per-channel membership lock. MUST be the first statement in the /// transaction that then reads roles/owner counts and writes membership, so the /// whole check-then-write sequence is atomic against a concurrent one. -async fn acquire_channel_membership_lock( +pub async fn acquire_channel_membership_lock( tx: &mut Transaction<'_, Postgres>, community_id: CommunityId, channel_id: Uuid, @@ -1872,6 +1872,30 @@ pub async fn get_member_role( Ok(row.map(|r| r.try_get("role")).transpose()?) } +/// Get an active member role using the caller's transaction. +/// +/// Callers that authorize a workflow run must first acquire +/// [`acquire_channel_membership_lock`] and hold the transaction through run +/// commit, serializing authorization with membership revocation. +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()?) +} + /// Archive ephemeral channels whose TTL deadline has passed. /// /// Returns the `(community_id, host, channel_id)` list that was archived. Idempotent — the diff --git a/crates/buzz-db/src/event.rs b/crates/buzz-db/src/event.rs index 136bcce26b5..151331cd6d5 100644 --- a/crates/buzz-db/src/event.rs +++ b/crates/buzz-db/src/event.rs @@ -1032,6 +1032,30 @@ pub async fn get_event_by_id( } } +/// Fetches a single non-deleted event by ID on the caller's transaction. +/// +/// Use this when a transaction already owns the connection so the lookup cannot +/// block waiting for another pool connection or observe a different snapshot. +pub async fn get_event_by_id_in_transaction( + tx: &mut Transaction<'_, Postgres>, + community_id: CommunityId, + id_bytes: &[u8], +) -> Result> { + let row = sqlx::query( + "SELECT id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id \ + FROM events WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL ORDER BY created_at DESC LIMIT 1", + ) + .bind(community_id.as_uuid()) + .bind(id_bytes) + .fetch_optional(&mut **tx) + .await?; + + match row { + Some(r) => row_to_stored_event(r), + None => Ok(None), + } +} + /// Fetches the latest global (non-channel, `channel_id IS NULL`) replaceable event /// for a (kind, pubkey) pair. /// diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index b37dcedff8c..e4dc5f44ed9 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -1419,6 +1419,17 @@ impl Db { event::get_event_by_id(&self.pool, community_id, id_bytes).await } + /// Fetches a single non-deleted event by its raw ID bytes on the caller's transaction. + #[datastore_span(name = "get_event_by_id_in_transaction", system = "postgresql")] + pub async fn get_event_by_id_in_transaction( + &self, + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + community_id: CommunityId, + id_bytes: &[u8], + ) -> Result> { + event::get_event_by_id_in_transaction(tx, community_id, id_bytes).await + } + /// Fetches a single event by its raw ID bytes, **including soft-deleted rows**. #[datastore_span(name = "get_event_by_id_including_deleted", system = "postgresql")] pub async fn get_event_by_id_including_deleted( @@ -3325,11 +3336,12 @@ impl Db { .await } - /// Insert or update a workflow using its NIP-33 `d`-tag UUID. + /// Atomically insert or update a workflow using its NIP-33 `d`-tag UUID. #[allow(clippy::too_many_arguments)] #[datastore_span(name = "upsert_workflow", system = "postgresql")] pub async fn upsert_workflow( &self, + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, community_id: CommunityId, id: Uuid, channel_id: Option, @@ -3337,9 +3349,10 @@ impl Db { name: &str, definition_json: &str, definition_hash: &[u8], + definition_event_id: &[u8], ) -> Result<()> { workflow::upsert_workflow( - &self.pool, + tx.as_mut(), community_id, id, channel_id, @@ -3347,6 +3360,7 @@ impl Db { name, definition_json, definition_hash, + definition_event_id, ) .await } @@ -3542,12 +3556,46 @@ impl Db { workflow::find_by_owner_and_name(&self.pool, community_id, owner_pubkey, name).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( + &self, + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + community_id: CommunityId, + id: Uuid, + ) -> Result { + workflow::get_workflow_for_share_in_transaction(tx, community_id, id).await + } + + /// Create a new workflow run on an existing transaction. + #[datastore_span(name = "create_workflow_run_in_transaction", system = "postgresql")] + pub async fn create_workflow_run_in_transaction( + &self, + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + community_id: CommunityId, + workflow_id: Uuid, + definition_event_id: &[u8], + trigger_event_id: Option<&[u8]>, + trigger_context: Option<&serde_json::Value>, + ) -> Result { + workflow::create_workflow_run_in_transaction( + tx, + community_id, + workflow_id, + definition_event_id, + trigger_event_id, + trigger_context, + ) + .await + } + /// Create a new workflow run. #[datastore_span(name = "create_workflow_run", system = "postgresql")] pub async fn create_workflow_run( &self, community_id: CommunityId, workflow_id: Uuid, + definition_event_id: &[u8], trigger_event_id: Option<&[u8]>, trigger_context: Option<&serde_json::Value>, ) -> Result { @@ -3555,6 +3603,7 @@ impl Db { &self.pool, community_id, workflow_id, + definition_event_id, trigger_event_id, trigger_context, ) diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index 9df02c8abf9..1508197a793 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -645,7 +645,7 @@ mod tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 32); + assert_eq!(migrations.len(), 34); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); assert!(migrations[0] @@ -1072,6 +1072,21 @@ mod tests { assert!(roster_fence.contains("snapshot_members IS DISTINCT FROM canonical_members")); assert!(roster_fence.contains("ERRCODE = '23514'")); + // Exact workflow-definition revision persistence is additive and keeps + // pre-migration rows nullable until their next signed definition write. + assert_eq!(migrations[32].version, 33); + let workflow_revision = migrations[32].sql.as_str(); + assert!(workflow_revision.contains("ADD COLUMN definition_event_id BYTEA")); + assert!(workflow_revision.contains("octet_length(definition_event_id) = 32")); + + // Runs durably inherit the exact signed revision they execute. Existing + // runs remain nullable and execution/resume must fail them closed. + assert_eq!(migrations[33].version, 34); + let run_revision = migrations[33].sql.as_str(); + assert!(run_revision.contains("ALTER TABLE workflow_runs")); + assert!(run_revision.contains("ADD COLUMN definition_event_id BYTEA")); + assert!(run_revision.contains("octet_length(definition_event_id) = 32")); + // Fresh desired-state bootstrap must install the identical executable // fence as migration 0032. CI and isolated relay startup use schema.sql // without running migrations, so drift reopens rolling-deploy races. diff --git a/crates/buzz-db/src/workflow.rs b/crates/buzz-db/src/workflow.rs index e970e978aaf..b06352c333c 100644 --- a/crates/buzz-db/src/workflow.rs +++ b/crates/buzz-db/src/workflow.rs @@ -177,6 +177,8 @@ pub struct WorkflowRecord { pub definition: serde_json::Value, /// SHA-256 hash of the canonical definition JSON. pub definition_hash: Vec, + /// Exact owner-signed kind:30620 event that materialized this revision. + pub definition_event_id: Option>, /// Current lifecycle status of the workflow definition. pub status: WorkflowStatus, /// Whether the workflow will fire on matching events. @@ -201,6 +203,11 @@ pub struct WorkflowRunRecord { pub community_id: CommunityId, /// The workflow definition that was executed. pub workflow_id: Uuid, + /// Exact owner-signed kind:30620 revision this run executes. + /// + /// NULL is retained only for runs created before the revision-binding + /// migration; execution and resume paths must fail those rows closed. + pub definition_event_id: Option>, /// Current execution status of this run. pub status: RunStatus, /// Raw event ID bytes that triggered this run, if any. @@ -314,7 +321,7 @@ pub async fn create_workflow( /// cross-channel overwrite primitive while still making retries idempotent. #[allow(clippy::too_many_arguments)] pub async fn upsert_workflow( - pool: &PgPool, + conn: &mut sqlx::PgConnection, community_id: CommunityId, id: Uuid, channel_id: Option, @@ -322,16 +329,18 @@ pub async fn upsert_workflow( name: &str, definition_json: &str, definition_hash: &[u8], + definition_event_id: &[u8], ) -> Result<()> { let row = sqlx::query( r#" INSERT INTO workflows - (community_id, id, name, owner_pubkey, channel_id, definition, definition_hash, status, enabled) - VALUES ($1, $2, $3, $4, $5, $6::jsonb, $7, 'active', TRUE) + (community_id, id, name, owner_pubkey, channel_id, definition, definition_hash, definition_event_id, status, enabled) + VALUES ($1, $2, $3, $4, $5, $6::jsonb, $7, $8, 'active', TRUE) ON CONFLICT (community_id, id) DO UPDATE SET name = EXCLUDED.name, definition = EXCLUDED.definition, definition_hash = EXCLUDED.definition_hash, + definition_event_id = EXCLUDED.definition_event_id, updated_at = NOW() WHERE workflows.owner_pubkey = EXCLUDED.owner_pubkey AND workflows.channel_id IS NOT DISTINCT FROM EXCLUDED.channel_id @@ -345,7 +354,8 @@ pub async fn upsert_workflow( .bind(channel_id) .bind(definition_json) .bind(definition_hash) - .fetch_optional(pool) + .bind(definition_event_id) + .fetch_optional(conn) .await?; if row.is_none() { @@ -370,7 +380,7 @@ pub async fn get_workflow( ) -> Result { let row = sqlx::query( r#" - SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, + SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, definition_event_id, status::text AS status, enabled, created_at, updated_at FROM workflows WHERE community_id = $1 AND id = $2 @@ -385,6 +395,34 @@ pub async fn get_workflow( row_to_workflow_record(row) } +/// Fetch and lock one workflow on the caller's transaction. +/// +/// The shared row lock is held through commit. Definition replacement takes an +/// update lock on the same row, so callers can validate an exact revision and +/// create dependent rows without a replacement committing between those steps. +pub async fn get_workflow_for_share_in_transaction( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + community_id: CommunityId, + id: Uuid, +) -> Result { + let row = sqlx::query( + r#" + SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, definition_event_id, + status::text AS status, enabled, created_at, updated_at + FROM workflows + WHERE community_id = $1 AND id = $2 + FOR SHARE + "#, + ) + .bind(community_id.as_uuid()) + .bind(id) + .fetch_optional(&mut **tx) + .await? + .ok_or_else(|| DbError::NotFound(format!("workflow {id}")))?; + + row_to_workflow_record(row) +} + /// List workflows for a channel, ordered newest first. /// /// `limit` is capped at [`LIST_MAX_LIMIT`]. Pass `None` to use [`LIST_DEFAULT_LIMIT`]. @@ -401,7 +439,7 @@ pub async fn list_channel_workflows( let rows = sqlx::query( r#" - SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, + SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, definition_event_id, status::text AS status, enabled, created_at, updated_at FROM workflows WHERE community_id = $1 AND channel_id = $2 @@ -432,7 +470,7 @@ pub async fn list_enabled_channel_workflows( ) -> Result> { let rows = sqlx::query( r#" - SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, + SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, definition_event_id, status::text AS status, enabled, created_at, updated_at FROM workflows WHERE community_id = $1 @@ -460,7 +498,7 @@ pub async fn list_enabled_channel_workflows( pub async fn list_all_enabled_workflows(pool: &PgPool) -> Result> { let rows = sqlx::query( r#" - SELECT w.id, w.community_id, w.name, w.owner_pubkey, w.channel_id, w.definition, w.definition_hash, + SELECT w.id, w.community_id, w.name, w.owner_pubkey, w.channel_id, w.definition, w.definition_hash, w.definition_event_id, w.status::text AS status, w.enabled, w.created_at, w.updated_at FROM workflows w JOIN communities c ON c.id = w.community_id @@ -793,6 +831,39 @@ pub async fn delete_workflow_for_owner( // -- Workflow Run CRUD -------------------------------------------------------- +/// Insert a new workflow run on the caller's transaction. +/// +/// Command handlers use this with the transaction that persisted the trigger +/// event so both rows commit or roll back together. +pub async fn create_workflow_run_in_transaction( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + community_id: CommunityId, + workflow_id: Uuid, + definition_event_id: &[u8], + trigger_event_id: Option<&[u8]>, + trigger_context: Option<&serde_json::Value>, +) -> Result { + let id = Uuid::new_v4(); + + sqlx::query( + r#" + INSERT INTO workflow_runs + (community_id, id, workflow_id, definition_event_id, status, trigger_event_id, current_step, execution_trace, trigger_context) + VALUES ($1, $2, $3, $4, 'pending', $5, 0, '[]', $6) + "#, + ) + .bind(community_id.as_uuid()) + .bind(id) + .bind(workflow_id) + .bind(definition_event_id) + .bind(trigger_event_id) + .bind(trigger_context) + .execute(&mut **tx) + .await?; + + Ok(id) +} + /// Insert a new workflow run. Returns the new run's UUID. /// /// `trigger_context` is the serialized `TriggerContext` for this run. It is stored @@ -802,6 +873,7 @@ pub async fn create_workflow_run( pool: &PgPool, community_id: CommunityId, workflow_id: Uuid, + definition_event_id: &[u8], trigger_event_id: Option<&[u8]>, trigger_context: Option<&serde_json::Value>, ) -> Result { @@ -810,13 +882,14 @@ pub async fn create_workflow_run( sqlx::query( r#" INSERT INTO workflow_runs - (community_id, id, workflow_id, status, trigger_event_id, current_step, execution_trace, trigger_context) - VALUES ($1, $2, $3, 'pending', $4, 0, '[]', $5) + (community_id, id, workflow_id, definition_event_id, status, trigger_event_id, current_step, execution_trace, trigger_context) + VALUES ($1, $2, $3, $4, 'pending', $5, 0, '[]', $6) "#, ) .bind(community_id.as_uuid()) .bind(id) .bind(workflow_id) + .bind(definition_event_id) .bind(trigger_event_id) .bind(trigger_context) .execute(pool) @@ -833,7 +906,7 @@ pub async fn get_workflow_run( ) -> Result { let row = sqlx::query( r#" - SELECT community_id, id, workflow_id, status::text AS status, trigger_event_id, current_step, + SELECT community_id, id, workflow_id, definition_event_id, status::text AS status, trigger_event_id, current_step, execution_trace, trigger_context, started_at, completed_at, error_message, error_code, created_at FROM workflow_runs WHERE community_id = $1 AND id = $2 @@ -865,7 +938,7 @@ pub async fn list_workflow_runs_page( let limit = limit.clamp(1, LIST_MAX_LIMIT); let rows = sqlx::query( r#" - SELECT community_id, id, workflow_id, status::text AS status, trigger_event_id, current_step, + SELECT community_id, id, workflow_id, definition_event_id, status::text AS status, trigger_event_id, current_step, execution_trace, trigger_context, started_at, completed_at, error_message, error_code, created_at FROM workflow_runs WHERE community_id = $1 AND workflow_id = $2 @@ -1183,6 +1256,7 @@ fn row_to_workflow_record(row: sqlx::postgres::PgRow) -> Result channel_id, definition: row.try_get("definition")?, definition_hash: row.try_get("definition_hash")?, + definition_event_id: row.try_get("definition_event_id")?, status, enabled, created_at: row.try_get("created_at")?, @@ -1202,6 +1276,7 @@ fn row_to_run_record(row: sqlx::postgres::PgRow) -> Result { id, community_id: CommunityId::from_uuid(community_id), workflow_id, + definition_event_id: row.try_get("definition_event_id")?, status, trigger_event_id: row.try_get("trigger_event_id")?, current_step: row.try_get("current_step")?, @@ -1247,7 +1322,7 @@ pub async fn find_by_owner_and_name( ) -> Result> { let row = sqlx::query( r#" - SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, + SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, definition_event_id, status::text AS status, enabled, created_at, updated_at FROM workflows WHERE community_id = $1 AND owner_pubkey = $2 AND name = $3 @@ -1382,6 +1457,7 @@ mod tests { channel_id: Some(channel_id), definition: def.clone(), definition_hash: vec![0x01, 0x02, 0x03, 0x04], + definition_event_id: None, status: WorkflowStatus::Active, enabled: true, created_at: now, @@ -1412,6 +1488,7 @@ mod tests { channel_id: None, definition: serde_json::json!({}), definition_hash: vec![], + definition_event_id: None, status: WorkflowStatus::Active, enabled: true, created_at: now, @@ -1434,6 +1511,7 @@ mod tests { channel_id: None, definition: serde_json::json!({}), definition_hash: vec![0xAA], + definition_event_id: None, status: WorkflowStatus::Active, enabled: true, created_at: now, @@ -1463,6 +1541,7 @@ mod tests { channel_id: None, definition: serde_json::json!({}), definition_hash: vec![], + definition_event_id: None, status: status.clone(), enabled: true, created_at: now, @@ -1483,6 +1562,7 @@ mod tests { channel_id: None, definition: serde_json::json!({}), definition_hash: vec![], + definition_event_id: None, status: WorkflowStatus::Active, enabled: false, created_at: now, @@ -1505,6 +1585,7 @@ mod tests { id, community_id: CommunityId::from_uuid(Uuid::new_v4()), workflow_id, + definition_event_id: Some(vec![0x42; 32]), status: RunStatus::Running, trigger_event_id: Some(trigger_event_id.clone()), current_step: 2, @@ -1536,6 +1617,7 @@ mod tests { id: Uuid::new_v4(), community_id: CommunityId::from_uuid(Uuid::new_v4()), workflow_id: Uuid::new_v4(), + definition_event_id: None, status: RunStatus::Pending, trigger_event_id: None, current_step: 0, @@ -1560,6 +1642,7 @@ mod tests { id: Uuid::new_v4(), community_id: CommunityId::from_uuid(Uuid::new_v4()), workflow_id: Uuid::new_v4(), + definition_event_id: None, status: RunStatus::Failed, trigger_event_id: None, current_step: 1, @@ -1592,6 +1675,7 @@ mod tests { id: Uuid::new_v4(), community_id: CommunityId::from_uuid(Uuid::new_v4()), workflow_id: Uuid::new_v4(), + definition_event_id: None, status: RunStatus::Completed, trigger_event_id: None, current_step: 2, @@ -1615,6 +1699,7 @@ mod tests { id: Uuid::new_v4(), community_id: CommunityId::from_uuid(Uuid::new_v4()), workflow_id: Uuid::new_v4(), + definition_event_id: None, status: RunStatus::Pending, trigger_event_id: None, current_step: 0, @@ -1963,7 +2048,7 @@ mod tests { .expect("claim wins"); // Create the run the won claim is responsible for, then attach it. - let run_id = create_workflow_run(&pool, community, workflow_id, None, None) + let run_id = create_workflow_run(&pool, community, workflow_id, &[0x42; 32], None, None) .await .expect("create run ok"); @@ -1992,7 +2077,7 @@ mod tests { // A second attach is a no-op: the `workflow_run_id IS NULL` guard means // an already-linked claim is never re-pointed to a different run. - let other_run = create_workflow_run(&pool, community, workflow_id, None, None) + let other_run = create_workflow_run(&pool, community, workflow_id, &[0x42; 32], None, None) .await .expect("create second run ok"); let reattached = @@ -2129,6 +2214,177 @@ mod tests { .expect("insert workflow"); } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn stale_revision_after_replacement_creates_no_run() { + let pool = setup_pool().await; + let community = make_community(&pool).await; + let workflow_id = Uuid::new_v4(); + insert_workflow_with_ids( + &pool, + community, + workflow_id, + Uuid::new_v4(), + "revision-race", + ) + .await; + let revision_a = vec![0xa1u8; 32]; + let revision_b = vec![0xb2u8; 32]; + sqlx::query( + "UPDATE workflows SET definition_event_id = $3 \ + WHERE community_id = $1 AND id = $2", + ) + .bind(community.as_uuid()) + .bind(workflow_id) + .bind(revision_a.as_slice()) + .execute(&pool) + .await + .expect("install revision A"); + + // B wins before stale trigger A enters its commit transaction. + sqlx::query( + "UPDATE workflows SET definition_event_id = $3 \ + WHERE community_id = $1 AND id = $2", + ) + .bind(community.as_uuid()) + .bind(workflow_id) + .bind(revision_b.as_slice()) + .execute(&pool) + .await + .expect("replace with revision B"); + + let mut trigger = pool.begin().await.expect("begin stale trigger"); + let current = get_workflow_for_share_in_transaction(&mut trigger, community, workflow_id) + .await + .expect("lock current workflow"); + assert_ne!( + current.definition_event_id.as_deref(), + Some(revision_a.as_slice()) + ); + trigger.rollback().await.expect("reject stale trigger"); + + let runs: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM workflow_runs WHERE community_id = $1 AND workflow_id = $2", + ) + .bind(community.as_uuid()) + .bind(workflow_id) + .fetch_one(&pool) + .await + .expect("count runs after stale rejection"); + assert_eq!(runs, 0, "stale revision must not create a workflow run"); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn workflow_run_transaction_rolls_back_and_retry_creates_exactly_one_run() { + let pool = setup_pool().await; + let community = make_community(&pool).await; + let workflow_id = Uuid::new_v4(); + insert_workflow_with_ids( + &pool, + community, + workflow_id, + Uuid::new_v4(), + "atomic-trigger", + ) + .await; + let trigger_event_id = vec![0x7au8; 32]; + let trigger_pubkey = vec![0x7bu8; 32]; + let trigger_sig = vec![0x7cu8; 64]; + + let mut aborted = pool.begin().await.expect("begin aborted transaction"); + sqlx::query( + "INSERT INTO events \ + (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at) \ + VALUES ($1, $2, $3, NOW(), 46020, '[]'::jsonb, '', $4, NOW())", + ) + .bind(community.as_uuid()) + .bind(trigger_event_id.as_slice()) + .bind(trigger_pubkey.as_slice()) + .bind(trigger_sig.as_slice()) + .execute(&mut *aborted) + .await + .expect("insert trigger event before abort"); + create_workflow_run_in_transaction( + &mut aborted, + community, + workflow_id, + &[0x42; 32], + Some(&trigger_event_id), + None, + ) + .await + .expect("insert run before abort"); + aborted.rollback().await.expect("roll back run"); + + let after_abort_event: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM events WHERE community_id = $1 AND id = $2") + .bind(community.as_uuid()) + .bind(&trigger_event_id) + .fetch_one(&pool) + .await + .expect("count rolled-back trigger events"); + assert_eq!( + after_abort_event, 0, + "aborted trigger must leave no event row" + ); + let after_abort: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM workflow_runs WHERE community_id = $1 AND trigger_event_id = $2", + ) + .bind(community.as_uuid()) + .bind(&trigger_event_id) + .fetch_one(&pool) + .await + .expect("count rolled-back runs"); + assert_eq!(after_abort, 0, "aborted trigger must leave no run row"); + + let mut retry = pool.begin().await.expect("begin retry transaction"); + sqlx::query( + "INSERT INTO events \ + (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at) \ + VALUES ($1, $2, $3, NOW(), 46020, '[]'::jsonb, '', $4, NOW())", + ) + .bind(community.as_uuid()) + .bind(trigger_event_id.as_slice()) + .bind(trigger_pubkey.as_slice()) + .bind(trigger_sig.as_slice()) + .execute(&mut *retry) + .await + .expect("insert trigger event on retry"); + create_workflow_run_in_transaction( + &mut retry, + community, + workflow_id, + &[0x42; 32], + Some(&trigger_event_id), + None, + ) + .await + .expect("insert retry run"); + retry.commit().await.expect("commit retry run"); + + let after_retry_event: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM events WHERE community_id = $1 AND id = $2") + .bind(community.as_uuid()) + .bind(&trigger_event_id) + .fetch_one(&pool) + .await + .expect("count committed trigger events"); + assert_eq!( + after_retry_event, 1, + "retry must create exactly one event row" + ); + let after_retry: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM workflow_runs WHERE community_id = $1 AND trigger_event_id = $2", + ) + .bind(community.as_uuid()) + .bind(&trigger_event_id) + .fetch_one(&pool) + .await + .expect("count committed retry runs"); + assert_eq!(after_retry, 1, "retry must create exactly one run row"); + } + /// Issue 4 (workflow identity): the same workflow UUID and channel UUID can /// exist in communities A and B (PK `(community_id, id)`). A request-scoped /// `get_workflow` / `list_enabled_channel_workflows` MUST return only the @@ -2273,10 +2529,10 @@ mod tests { insert_workflow_with_ids(&pool, community_a, workflow_id, channel_id, "wf-A").await; insert_workflow_with_ids(&pool, community_b, workflow_id, Uuid::new_v4(), "wf-B").await; - let run_a = create_workflow_run(&pool, community_a, workflow_id, None, None) + let run_a = create_workflow_run(&pool, community_a, workflow_id, &[0x42; 32], None, None) .await .expect("run A"); - let run_b = create_workflow_run(&pool, community_b, workflow_id, None, None) + let run_b = create_workflow_run(&pool, community_b, workflow_id, &[0x42; 32], None, None) .await .expect("run B"); diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 2545cd819dd..50f6a74c102 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -2111,52 +2111,41 @@ pub async fn workflow_webhook( .await .map_err(|_| not_found("workflow not found"))?; + let definition_event_id = workflow + .definition_event_id + .as_deref() + .ok_or_else(|| not_found("workflow not found"))?; let run_id = state .db - .create_workflow_run(community_id, id, None, trigger_ctx_json.as_ref()) + .create_workflow_run( + community_id, + id, + definition_event_id, + None, + trigger_ctx_json.as_ref(), + ) .await .map_err(|e| super::internal_error(&format!("db error: {e}")))?; // Spawn workflow execution asynchronously. let engine = Arc::clone(&state.workflow_engine); - let db = state.db.clone(); - let def_value = workflow.definition.clone(); let trigger_ctx_clone = trigger_ctx.clone(); tokio::spawn(async move { - let def: buzz_workflow::WorkflowDef = match serde_json::from_value(def_value) { - Ok(d) => d, - Err(e) => { - tracing::error!("webhook: failed to parse definition: {e}"); - if let Err(db_err) = db - .update_workflow_run( - community_id, - run_id, - buzz_db::workflow::RunStatus::Failed, - 0, - &serde_json::json!([]), - Some(buzz_db::workflow::WorkflowRunFailure { - code: "invalid_definition", - message: &format!("definition parse error: {e}"), - }), - ) - .await - { - tracing::error!("webhook: failed to mark run as failed: {db_err}"); - } - return; + let result = match engine.load_run_definition(community_id, run_id).await { + Ok((_, definition)) => { + buzz_workflow::executor::execute_from_step( + &engine, + community_id, + run_id, + &definition, + &trigger_ctx_clone, + 0, + None, + ) + .await } + Err(error) => Err((error, buzz_workflow::error::PartialProgress::default())), }; - - let result = buzz_workflow::executor::execute_from_step( - &engine, - community_id, - run_id, - &def, - &trigger_ctx_clone, - 0, - None, - ) - .await; engine .finalize_run(community_id, run_id, result, None) .await; diff --git a/crates/buzz-relay/src/api/workflows.rs b/crates/buzz-relay/src/api/workflows.rs index a3d5a6c729e..70d4bfcbd8f 100644 --- a/crates/buzz-relay/src/api/workflows.rs +++ b/crates/buzz-relay/src/api/workflows.rs @@ -46,6 +46,7 @@ async fn authorize_workflow_read( path: &str, raw_query: Option<&str>, workflow_id: Uuid, + allow_immutable_owner: bool, ) -> Result)> { let raw_host = headers .get(axum::http::header::HOST) @@ -97,15 +98,59 @@ async fn authorize_workflow_read( .await .map_err(|error| internal_error(&format!("workflow channel access lookup: {error}")))?; if !accessible.contains(&channel_id) { - return Err(api_error( - StatusCode::FORBIDDEN, - "workflow is not accessible", - )); + let controls = allow_immutable_owner + && (workflow.owner_pubkey == pubkey_bytes + || state + .db + .is_agent_owner(tenant.community(), &workflow.owner_pubkey, &pubkey_bytes) + .await + .map_err(|error| internal_error(&format!("workflow owner lookup: {error}")))?); + if !controls { + return Err(api_error( + StatusCode::FORBIDDEN, + "workflow is not accessible", + )); + } } Ok(tenant) } +/// `GET /workflows/{workflow_id}/revision` — current signed revision for an +/// authorized channel reader or the managed agent's immutable human owner. +/// +/// This narrow endpoint does not grant channel visibility; it returns only the +/// exact owner-signed definition event needed to construct a revision-bound +/// manual trigger. +pub async fn workflow_revision( + State(state): State>, + Path(workflow_id): Path, + headers: HeaderMap, +) -> Result, (StatusCode, Json)> { + let path = format!("/workflows/{workflow_id}/revision"); + let tenant = authorize_workflow_read(&state, &headers, &path, None, workflow_id, true).await?; + let workflow = state + .db + .get_workflow(tenant.community(), workflow_id) + .await + .map_err(|_| api_error(StatusCode::NOT_FOUND, "workflow not found"))?; + let revision = workflow.definition_event_id.as_deref().ok_or_else(|| { + api_error( + StatusCode::CONFLICT, + "owner-signed workflow revision is unavailable", + ) + })?; + let event = state + .db + .get_event_by_id(tenant.community(), revision) + .await + .map_err(|error| internal_error(&format!("get workflow revision: {error}")))? + .ok_or_else(|| api_error(StatusCode::CONFLICT, "workflow revision is unavailable"))?; + Ok(Json(serde_json::to_value(event.event).map_err( + |error| internal_error(&format!("serialize workflow revision: {error}")), + )?)) +} + /// `GET /workflows/{workflow_id}/runs` — one authorized, keyset-paginated page. pub async fn workflow_runs( State(state): State>, @@ -129,8 +174,15 @@ pub async fn workflow_runs( } let path = format!("/workflows/{workflow_id}/runs"); - let tenant = - authorize_workflow_read(&state, &headers, &path, raw_query.as_deref(), workflow_id).await?; + let tenant = authorize_workflow_read( + &state, + &headers, + &path, + raw_query.as_deref(), + workflow_id, + false, + ) + .await?; let mut rows = state .db .list_workflow_runs_page( @@ -169,7 +221,7 @@ pub async fn run_approvals( headers: HeaderMap, ) -> Result, (StatusCode, Json)> { let path = format!("/workflows/{workflow_id}/runs/{run_id}/approvals"); - let tenant = authorize_workflow_read(&state, &headers, &path, None, workflow_id).await?; + let tenant = authorize_workflow_read(&state, &headers, &path, None, workflow_id, false).await?; let run = state .db diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 007db43ffd9..08b0946a6d2 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -92,12 +92,11 @@ enum PersistResult { /// If the event is a duplicate (ON CONFLICT DO NOTHING), the transaction is /// rolled back and `PersistResult::Duplicate` is returned — no mutations needed. /// -/// NOTE: Domain mutations (open_dm, upsert_workflow, etc.) execute on the -/// connection pool, NOT inside this transaction. The pattern is idempotent but -/// not strictly atomic: if a mutation succeeds but commit fails, the mutation -/// persists without the event record. On retry, the event INSERT succeeds -/// (no conflict), and the mutation re-executes — which is safe for idempotent -/// operations (open_dm, hide_dm, update_approval, upsert_workflow). +/// NOTE: Most domain mutations still execute on the connection pool rather +/// than in this transaction. Workflow-definition ingestion is the exception: +/// its materialized workflow row and exact signed revision are written through +/// this transaction so the event and revision binding commit atomically. +/// Other operations remain idempotent but not strictly atomic. #[datastore_span(name = "persist_command_event", system = "postgresql")] async fn persist_command_event( db: &buzz_db::Db, @@ -736,8 +735,9 @@ async fn handle_workflow_def( .map_err(|e| IngestError::Internal(format!("error: json serialize: {e}")))?; let hash = compute_definition_hash(&definition_json_final); - // Persist the command event — returns open transaction - let tx = match persist_command_event(&state.db, tenant, event, None).await? { + // Persist the command event — returns the transaction that will also own + // the materialized workflow revision update. + let mut tx = match persist_command_event(&state.db, tenant, event, Some(channel_id)).await? { PersistResult::Duplicate => { return Ok(IngestResult { event_id: event.id.to_hex(), @@ -769,6 +769,7 @@ async fn handle_workflow_def( state .db .upsert_workflow( + &mut tx, community_id, workflow_id, Some(channel_id), @@ -776,6 +777,7 @@ async fn handle_workflow_def( &workflow_name, &definition_json_final, &hash, + event.id.as_bytes(), ) .await .map_err(|e| match e { @@ -785,17 +787,18 @@ async fn handle_workflow_def( other => IngestError::Internal(format!("error: db upsert_workflow: {other}")), })?; - // Drop the trigger-path cache entry so the new/updated definition fires on - // the next matching event instead of after the cache TTL. - state - .workflow_engine - .invalidate_channel_workflows(community_id, channel_id); - // Commit the event transaction after the idempotent workflow upsert succeeds. tx.commit() .await .map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?; + // Invalidate only after commit. Invalidating while the new row is still + // invisible lets a concurrent trigger refill the cache with the old + // definition and retain it until TTL expiry. + state + .workflow_engine + .invalidate_channel_workflows(community_id, channel_id); + // 5. Return response let mut resp = serde_json::json!({ "workflow_id": workflow_id.to_string(), @@ -811,6 +814,101 @@ async fn handle_workflow_def( }) } +async fn caller_controls_workflow( + state: &Arc, + community_id: CommunityId, + workflow_owner: &[u8], + caller: &[u8], +) -> Result { + if workflow_owner == caller { + return Ok(true); + } + + state + .db + .is_agent_owner(community_id, workflow_owner, caller) + .await + .map_err(|e| IngestError::Internal(format!("error: workflow owner check: {e}"))) +} + +fn exact_tag_value<'a>(event: &'a Event, name: &str) -> Option<&'a str> { + let mut values = event.tags.iter().filter_map(|tag| { + (tag.kind().to_string() == name) + .then(|| tag.content()) + .flatten() + }); + let value = values.next()?; + values.next().is_none().then_some(value) +} + +async fn verify_workflow_revision( + state: &Arc, + mut tx: Option<&mut sqlx::Transaction<'_, sqlx::Postgres>>, + community_id: CommunityId, + workflow: &buzz_db::workflow::WorkflowRecord, + requested_revision: &[u8], +) -> Result<(), IngestError> { + let Some(persisted_revision) = workflow.definition_event_id.as_deref() else { + return Err(IngestError::Rejected( + "invalid: owner-signed workflow revision is unavailable".into(), + )); + }; + if persisted_revision != requested_revision { + return Err(IngestError::Rejected( + "conflict: workflow revision does not match current definition".into(), + )); + } + + let stored = match tx.as_mut() { + Some(tx) => { + state + .db + .get_event_by_id_in_transaction(tx, community_id, persisted_revision) + .await + } + None => { + state + .db + .get_event_by_id(community_id, persisted_revision) + .await + } + } + .map_err(|e| IngestError::Internal(format!("error: workflow revision lookup: {e}")))? + .ok_or_else(|| IngestError::Rejected("invalid: signed workflow revision not found".into()))?; + let definition_event = &stored.event; + let workflow_id = workflow.id.to_string(); + let workflow_channel_id = workflow.channel_id.map(|id| id.to_string()); + if definition_event.id.as_bytes() != persisted_revision + || !definition_event.verify_id() + || !definition_event.verify_signature() + || definition_event.kind.as_u16() as u32 != KIND_WORKFLOW_DEF + || definition_event.pubkey.to_bytes().as_slice() != workflow.owner_pubkey + || exact_tag_value(definition_event, "d") != Some(workflow_id.as_str()) + || workflow_channel_id.is_none() + || exact_tag_value(definition_event, "h") != workflow_channel_id.as_deref() + || stored.channel_id != workflow.channel_id + { + return Err(IngestError::Rejected( + "invalid: signed workflow revision binding mismatch".into(), + )); + } + + let (_, signed_json) = buzz_workflow::WorkflowEngine::parse_yaml(&definition_event.content) + .map_err(|_| { + IngestError::Rejected("invalid: signed workflow revision is malformed".into()) + })?; + let signed_definition: serde_json::Value = + serde_json::from_str(&signed_json).map_err(|_| { + IngestError::Rejected("invalid: signed workflow revision is malformed".into()) + })?; + if signed_definition != webhook_secret::strip_secret(&workflow.definition) { + return Err(IngestError::Rejected( + "invalid: signed workflow revision differs from materialized definition".into(), + )); + } + Ok(()) +} + async fn handle_workflow_trigger( tenant: &TenantContext, state: &Arc, @@ -819,13 +917,21 @@ async fn handle_workflow_trigger( ) -> Result { let self_bytes = auth.pubkey().to_bytes().to_vec(); - // 1. Extract workflow reference from `d` tag or `e` tag - let workflow_id_str = extract_d_tag(event) - .or_else(|| extract_e_tag(event)) - .ok_or_else(|| { - IngestError::Rejected("invalid: missing workflow reference (d or e tag)".into()) - })?; - let workflow_id = Uuid::parse_str(&workflow_id_str) + // 1. Bind the command to both the workflow UUID and one exact signed revision. + let workflow_id_str = exact_tag_value(event, "d").ok_or_else(|| { + IngestError::Rejected("invalid: expected exactly one workflow d tag".into()) + })?; + let revision_hex = exact_tag_value(event, "e").ok_or_else(|| { + IngestError::Rejected("invalid: expected exactly one workflow revision e tag".into()) + })?; + let requested_revision = hex::decode(revision_hex) + .map_err(|_| IngestError::Rejected("invalid: bad workflow revision event id".into()))?; + if requested_revision.len() != 32 { + return Err(IngestError::Rejected( + "invalid: bad workflow revision event id".into(), + )); + } + let workflow_id = Uuid::parse_str(workflow_id_str) .map_err(|_| IngestError::Rejected("invalid: bad workflow_id format".into()))?; // 2. Validate workflow exists — scoped to the caller's community. The same @@ -839,14 +945,20 @@ async fn handle_workflow_trigger( .await .map_err(|_| IngestError::Rejected("invalid: workflow not found".into()))?; - // 3. Manual triggers execute with the workflow owner's authority, so only - // the owner may start them. Channel membership alone is insufficient: a - // member could otherwise invoke another user's webhook or message actions. - if workflow.owner_pubkey != self_bytes { + // 3. Manual triggers execute with the workflow owner's authority. Permit + // that principal and, for a managed agent, its immutable human owner. + // Channel membership alone remains insufficient. + if !caller_controls_workflow(state, community_id, &workflow.owner_pubkey, &self_bytes).await? { return Err(IngestError::Rejected( "forbidden: not authorized to trigger this workflow".into(), )); } + // Managed-agent ownership is immutable. Carry the authorized workflow + // principal across the transaction boundary so no pool-backed ownership + // lookup is attempted while the command transaction holds its connection. + let authorized_workflow_owner = workflow.owner_pubkey.clone(); + + verify_workflow_revision(state, None, community_id, &workflow, &requested_revision).await?; // SEC-006: manual triggers must honor the workflow's lifecycle state and // recheck the owner's *current* channel authority before creating a run. @@ -876,7 +988,7 @@ async fn handle_workflow_trigger( // Persist the command event under the workflow channel even though the // trigger event itself only carries the workflow UUID. Storing channel // triggers as global events leaks workflow IDs to unrelated relay members. - let tx = match persist_command_event(&state.db, tenant, event, workflow.channel_id).await? { + 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(), @@ -887,6 +999,57 @@ async fn handle_workflow_trigger( PersistResult::Inserted(tx) => tx, }; + // Serialize the final authority check and run commit with channel + // membership writers. If revocation commits first we observe no role; if + // this lock wins, revocation cannot commit until this run is durable. + buzz_db::channel::acquire_channel_membership_lock(&mut tx, community_id, wf_channel_id) + .await + .map_err(|e| IngestError::Internal(format!("error: membership lock: {e}")))?; + + // Re-read the workflow under a row lock on the same transaction that will + // commit the trigger event and run. Definition replacement updates this row, + // so it cannot commit between this exact-revision check and our commit. A + // replacement that won first is observed here and rejected as stale. + let workflow = state + .db + .get_workflow_for_share_in_transaction(&mut tx, community_id, workflow_id) + .await + .map_err(|_| IngestError::Rejected("invalid: workflow not found".into()))?; + if workflow.owner_pubkey != authorized_workflow_owner { + return Err(IngestError::Rejected( + "conflict: workflow owner changed while trigger was being processed".into(), + )); + } + verify_workflow_revision( + state, + Some(&mut tx), + community_id, + &workflow, + &requested_revision, + ) + .await?; + if !workflow.enabled || workflow.status != buzz_db::workflow::WorkflowStatus::Active { + return Err(IngestError::Rejected( + "forbidden: workflow is disabled or inactive".into(), + )); + } + let role = buzz_db::channel::get_member_role_in_transaction( + &mut tx, + community_id, + wf_channel_id, + &workflow.owner_pubkey, + ) + .await + .map_err(|e| IngestError::Internal(format!("error: owner authority lookup: {e}")))?; + if !matches!( + (role.as_deref(), def.requires_elevated_authority()), + (Some(_), false) | (Some("owner" | "admin"), true) + ) { + return Err(IngestError::Rejected( + "forbidden: not authorized to trigger this workflow".into(), + )); + } + // 4. Execute: create workflow run let mut trigger_ctx = TriggerContext { channel_id: workflow @@ -894,6 +1057,7 @@ async fn handle_workflow_trigger( .map(|id| id.to_string()) .unwrap_or_default(), author: hex::encode(&self_bytes), + definition_event_id: revision_hex.to_owned(), ..Default::default() }; if !event.content.is_empty() { @@ -912,9 +1076,11 @@ async fn handle_workflow_trigger( let event_id_bytes = event.id.as_bytes().to_vec(); let run_id = state .db - .create_workflow_run( + .create_workflow_run_in_transaction( + &mut tx, community_id, workflow_id, + &requested_revision, Some(&event_id_bytes), trigger_ctx_json.as_ref(), ) @@ -928,44 +1094,23 @@ async fn handle_workflow_trigger( // 5. Spawn workflow execution let engine = Arc::clone(&state.workflow_engine); - let db = state.db.clone(); - let def_value = workflow.definition.clone(); let trigger_ctx_clone = trigger_ctx.clone(); tokio::spawn(async move { - let def: buzz_workflow::WorkflowDef = match serde_json::from_value(def_value) { - Ok(d) => d, - Err(e) => { - tracing::error!("workflow_trigger: failed to parse definition: {e}"); - if let Err(db_err) = db - .update_workflow_run( - community_id, - run_id, - RunStatus::Failed, - 0, - &serde_json::json!([]), - Some(buzz_db::workflow::WorkflowRunFailure { - code: "invalid_definition", - message: &format!("definition parse error: {e}"), - }), - ) - .await - { - tracing::error!("workflow_trigger: failed to mark run as failed: {db_err}"); - } - return; + let result = match engine.load_run_definition(community_id, run_id).await { + Ok((_, definition)) => { + buzz_workflow::executor::execute_from_step( + &engine, + community_id, + run_id, + &definition, + &trigger_ctx_clone, + 0, + None, + ) + .await } + Err(error) => Err((error, buzz_workflow::error::PartialProgress::default())), }; - - let result = buzz_workflow::executor::execute_from_step( - &engine, - community_id, - run_id, - &def, - &trigger_ctx_clone, - 0, - None, - ) - .await; engine .finalize_run(community_id, run_id, result, None) .await; @@ -1277,15 +1422,38 @@ async fn resume_workflow_after_approval( workflow_id: Uuid, resume_index: usize, ) { - let run = match db.get_workflow_run(community_id, run_id).await { - Ok(r) => r, + let (run, def) = match engine.load_run_definition(community_id, run_id).await { + Ok(bound) => bound, Err(e) => { - tracing::error!("resume_workflow: failed to fetch run {run_id}: {e}"); + tracing::error!( + "resume_workflow: failed to load run-bound definition for {run_id}: {e}" + ); + if let Ok(run) = db.get_workflow_run(community_id, run_id).await { + let _ = db + .update_workflow_run( + community_id, + run_id, + RunStatus::Failed, + run.current_step, + &run.execution_trace, + Some(buzz_db::workflow::WorkflowRunFailure { + code: "invalid_definition_revision", + message: &e.to_string(), + }), + ) + .await; + } return; } }; - // Guard: only resume runs that are actually waiting for approval + if run.workflow_id != workflow_id { + tracing::error!( + "resume_workflow: approval workflow {workflow_id} does not match run {}", + run.workflow_id + ); + return; + } if run.status != RunStatus::WaitingApproval { tracing::warn!( "resume_workflow: run {run_id} has status '{}', expected 'waiting_approval'", @@ -1294,40 +1462,6 @@ async fn resume_workflow_after_approval( return; } - let workflow = match db.get_workflow(community_id, workflow_id).await { - Ok(w) => w, - Err(e) => { - tracing::error!("resume_workflow: failed to fetch workflow {workflow_id}: {e}"); - return; - } - }; - - let def: buzz_workflow::WorkflowDef = match serde_json::from_value(workflow.definition.clone()) - { - Ok(d) => d, - Err(e) => { - tracing::error!("resume_workflow: failed to parse workflow definition: {e}"); - if let Err(db_err) = db - .update_workflow_run( - community_id, - run_id, - RunStatus::Failed, - run.current_step, - &run.execution_trace, - Some(buzz_db::workflow::WorkflowRunFailure { - code: "invalid_definition", - message: &format!("definition parse error: {e}"), - }), - ) - .await - { - tracing::error!("resume_workflow: failed to mark run as failed: {db_err}"); - } - return; - } - }; - - // Reconstruct step_outputs from execution trace for template resolution let mut initial_outputs: std::collections::HashMap = std::collections::HashMap::new(); if let Some(trace_arr) = run.execution_trace.as_array() { @@ -1341,14 +1475,11 @@ async fn resume_workflow_after_approval( } } - // Restore trigger context for {{trigger.*}} templates let trigger_ctx: TriggerContext = run .trigger_context .as_ref() .and_then(|v| serde_json::from_value(v.clone()).ok()) .unwrap_or_default(); - - // Execute remaining steps let existing_trace = run.execution_trace.as_array().cloned(); let result = buzz_workflow::executor::execute_from_step( &engine, @@ -1374,7 +1505,10 @@ mod tests { let url = std::env::var("BUZZ_TEST_DATABASE_URL") .or_else(|_| std::env::var("DATABASE_URL")) .unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".to_string()); - let pool = sqlx::PgPool::connect(&url) + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .acquire_timeout(std::time::Duration::from_secs(1)) + .connect(&url) .await .expect("connect workflow persistence test database"); let db = buzz_db::Db::from_pool(pool); @@ -1390,6 +1524,164 @@ mod tests { (db, TenantContext::resolved(community, host)) } + async fn manual_trigger_test_context() -> (Arc, TenantContext, Keys, Uuid, Event) { + use buzz_core::channel::{ChannelType, ChannelVisibility}; + + let url = std::env::var("BUZZ_TEST_DATABASE_URL") + .or_else(|_| std::env::var("DATABASE_URL")) + .unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".to_string()); + let setup_pool = sqlx::PgPool::connect(&url) + .await + .expect("connect workflow trigger setup database"); + let setup_db = buzz_db::Db::from_pool(setup_pool.clone()); + setup_db + .migrate() + .await + .expect("migrate workflow trigger test database"); + + let host = format!("workflow-trigger-{}.example", Uuid::new_v4().simple()); + let community = setup_db + .ensure_configured_community(&host) + .await + .expect("create workflow trigger test community") + .id; + let tenant = TenantContext::resolved(community, host.clone()); + let human = Keys::generate(); + let agent = Keys::generate(); + let human_bytes = human.public_key().to_bytes(); + let agent_bytes = agent.public_key().to_bytes(); + setup_db + .ensure_user(community, &human_bytes) + .await + .expect("ensure human owner"); + setup_db + .ensure_user(community, &agent_bytes) + .await + .expect("ensure managed agent"); + assert!(setup_db + .set_agent_owner(community, &agent_bytes, &human_bytes) + .await + .expect("set immutable agent owner")); + let channel = setup_db + .create_channel( + community, + "manual-trigger-pool", + ChannelType::Stream, + ChannelVisibility::Open, + None, + &agent_bytes, + None, + ) + .await + .expect("create workflow channel"); + let workflow_id = Uuid::new_v4(); + let definition = EventBuilder::new( + Kind::Custom(KIND_WORKFLOW_DEF as u16), + concat!( + "name: manual-trigger-pool\n", + "trigger:\n on: message_posted\n", + "steps:\n - id: send\n action: send_message\n text: done\n", + ), + ) + .tags(vec![ + Tag::parse(["d", workflow_id.to_string().as_str()]).expect("d tag"), + Tag::parse(["h", channel.id.to_string().as_str()]).expect("h tag"), + ]) + .sign_with_keys(&agent) + .expect("sign workflow definition"); + let (_, definition_json) = buzz_workflow::WorkflowEngine::parse_yaml(&definition.content) + .expect("parse signed workflow definition"); + let definition_hash = compute_definition_hash(&definition_json); + let mut tx = setup_db + .begin_transaction() + .await + .expect("begin workflow seed"); + buzz_db::event::insert_event_in_transaction( + &mut tx, + community, + &definition, + Some(channel.id), + ) + .await + .expect("persist signed workflow definition"); + setup_db + .upsert_workflow( + &mut tx, + community, + workflow_id, + Some(channel.id), + &agent_bytes, + "manual-trigger-pool", + &definition_json, + &definition_hash, + definition.id.as_bytes(), + ) + .await + .expect("materialize signed workflow"); + tx.commit().await.expect("commit signed workflow"); + + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .acquire_timeout(std::time::Duration::from_secs(1)) + .connect(&url) + .await + .expect("connect one-connection workflow trigger pool"); + let db = buzz_db::Db::from_pool(pool.clone()); + let mut config = crate::config::Config::from_env().expect("config from env"); + config.database_url = url; + config.redis_url = "redis://127.0.0.1:1".to_string(); + config.relay_url = format!("wss://{host}"); + config.require_relay_membership = false; + let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) + .create_pool(Some(deadpool_redis::Runtime::Tokio1)) + .expect("redis pool config"); + let pubsub = Arc::new( + buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) + .await + .expect("pubsub manager"), + ); + let audit = buzz_audit::AuditService::new(pool.clone()); + let auth = buzz_auth::AuthService::new(config.auth.clone()); + let search = buzz_search::SearchService::new(pool); + let workflow_engine = Arc::new(buzz_workflow::WorkflowEngine::new( + db.clone(), + buzz_workflow::WorkflowConfig::default(), + )); + let media_storage = buzz_media::MediaStorage::new(&config.media).expect("media storage"); + let (state, _audit_shutdown) = AppState::new( + config, + db, + redis_pool, + audit, + pubsub, + auth, + search, + workflow_engine, + Keys::generate(), + media_storage, + ); + setup_pool.close().await; + (Arc::new(state), tenant, human, workflow_id, definition) + } + + fn workflow_trigger_event(keys: &Keys, workflow_id: Uuid, revision: &Event) -> 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.id.to_hex().as_str()]).expect("revision tag"), + ]) + .sign_with_keys(keys) + .expect("sign workflow trigger") + } + + fn http_auth(keys: &Keys) -> IngestAuth { + IngestAuth::Http { + pubkey: keys.public_key(), + scopes: vec![buzz_auth::Scope::MessagesWrite], + auth_method: super::super::ingest::HttpAuthMethod::Nip98, + } + } + fn workflow_event( keys: &Keys, workflow_id: Uuid, @@ -1425,6 +1717,60 @@ mod tests { } } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn human_owner_manual_trigger_completes_with_one_connection() { + let (state, tenant, human, workflow_id, revision) = manual_trigger_test_context().await; + let trigger = workflow_trigger_event(&human, workflow_id, &revision); + + let result = tokio::time::timeout( + std::time::Duration::from_secs(3), + handle_workflow_trigger(&tenant, &state, &trigger, &http_auth(&human)), + ) + .await + .expect("human-owner trigger must not wait for a second pool connection") + .expect("human-owner trigger must succeed"); + assert!(result.message.contains("\"run_id\"")); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn concurrent_human_owner_manual_triggers_do_not_starve_one_connection_pool() { + let (state, tenant, human, 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" + ); + } + } + #[test] fn workflow_revision_parser_accepts_create_and_valid_update() { let revision = [0x42; 32]; @@ -1473,6 +1819,10 @@ mod tests { let workflow_id = Uuid::new_v4(); let created_at = Timestamp::now().as_secs(); let create = workflow_event(&keys, workflow_id, created_at, None, "create"); + let channel_id = Uuid::parse_str( + exact_tag_value(&create, "h").expect("workflow definition channel tag"), + ) + .expect("workflow definition channel UUID"); let missing_revision = hex::encode([0x24; 32]); let missing_revision_update = workflow_event( @@ -1493,15 +1843,22 @@ mod tests { if message == "conflict: workflow revision does not exist" )); - let PersistResult::Inserted(tx) = persist_command_event(&db, &tenant, &create, None) - .await - .expect("persist create") + let PersistResult::Inserted(tx) = + persist_command_event(&db, &tenant, &create, Some(channel_id)) + .await + .expect("persist create") else { panic!("first create must insert"); }; tx.commit().await.expect("commit create"); + let stored_create = db + .get_event_by_id(tenant.community(), create.id.as_bytes()) + .await + .expect("load persisted workflow definition") + .expect("persisted workflow definition"); + assert_eq!(stored_create.channel_id, Some(channel_id)); assert!(matches!( - persist_command_event(&db, &tenant, &create, None) + persist_command_event(&db, &tenant, &create, Some(channel_id)) .await .expect("replay create"), PersistResult::Duplicate diff --git a/crates/buzz-relay/src/router.rs b/crates/buzz-relay/src/router.rs index fc5396407c1..0d9487ec3d7 100644 --- a/crates/buzz-relay/src/router.rs +++ b/crates/buzz-relay/src/router.rs @@ -75,6 +75,10 @@ pub fn build_router(state: Arc) -> Router { // Relay-owned third-party GIF metadata proxy (NIP-98 auth). .route(api::gifs::SEARCH_PATH, post(api::gifs::search)) .route(api::gifs::SHARE_PATH, post(api::gifs::share)) + .route( + "/workflows/{workflow_id}/revision", + get(api::workflows::workflow_revision), + ) .route( "/workflows/{workflow_id}/runs", get(api::workflows::workflow_runs), diff --git a/crates/buzz-sdk/src/builders.rs b/crates/buzz-sdk/src/builders.rs index 71c0f1e73db..c8fdc83223e 100644 --- a/crates/buzz-sdk/src/builders.rs +++ b/crates/buzz-sdk/src/builders.rs @@ -1640,9 +1640,16 @@ pub fn build_workflow_delete( build_delete_addressable(KIND_WORKFLOW_DEF, author_pubkey, &workflow_id.to_string()) } -/// Build a workflow trigger event (kind 46020). -pub fn build_workflow_trigger(workflow_id: Uuid) -> Result { - let tags = vec![tag(&["d", &workflow_id.to_string()])?]; +/// Build a workflow trigger event (kind 46020) bound to an exact signed revision. +pub fn build_workflow_trigger( + workflow_id: Uuid, + definition_event_id: &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])?, + ]; Ok(EventBuilder::new(Kind::Custom(KIND_WORKFLOW_TRIGGER as u16), "").tags(tags)) } @@ -4010,9 +4017,11 @@ mod tests { #[test] fn workflow_trigger_happy_path() { let wid = uuid(); - let ev = sign(build_workflow_trigger(wid).unwrap()); + let ev = sign(build_workflow_trigger(wid, &"ab".repeat(32)).unwrap()); assert_eq!(ev.kind.as_u16(), 46020); assert!(has_tag(&ev, "d", &wid.to_string())); + assert!(has_tag(&ev, "e", &"ab".repeat(32))); + assert!(build_workflow_trigger(wid, "not-an-event-id").is_err()); } #[test] diff --git a/crates/buzz-test-client/tests/e2e_workflow_agent_owner.rs b/crates/buzz-test-client/tests/e2e_workflow_agent_owner.rs new file mode 100644 index 00000000000..d93609b0ea1 --- /dev/null +++ b/crates/buzz-test-client/tests/e2e_workflow_agent_owner.rs @@ -0,0 +1,205 @@ +//! End-to-end authorization and revision-binding coverage for agent-owned workflows. +//! +//! Run against a local relay with: +//! `cargo test -p buzz-test-client --test e2e_workflow_agent_owner -- --ignored` + +use buzz_sdk::nip_oa; +use buzz_test_client::BuzzTestClient; +use nostr::{EventBuilder, Keys, Kind, Tag}; + +fn relay_url() -> String { + std::env::var("RELAY_URL").unwrap_or_else(|_| "ws://localhost:3000".to_string()) +} + +fn workflow_yaml(text: &str) -> String { + format!( + "name: Agent-owned workflow\n\ + trigger:\n\ + \x20 on: webhook\n\ + steps:\n\ + \x20 - id: notify\n\ + \x20 action: send_message\n\ + \x20 text: {text}\n" + ) +} + +async fn connect_agent_with_owner(agent: &Keys, owner: &Keys) -> BuzzTestClient { + let tag_json = + nip_oa::compute_auth_tag(owner, &agent.public_key(), "").expect("compute NIP-OA auth tag"); + let auth_tag = nip_oa::parse_auth_tag(&tag_json).expect("parse NIP-OA auth tag"); + let mut client = BuzzTestClient::connect_unauthenticated(&relay_url()) + .await + .expect("connect agent"); + client + .authenticate_with_nip_oa(agent, &auth_tag) + .await + .expect("authenticate agent with NIP-OA"); + client +} + +fn trigger(keys: &Keys, workflow_id: uuid::Uuid, revision: &str) -> nostr::Event { + buzz_sdk::build_workflow_trigger(workflow_id, revision) + .expect("build workflow trigger") + .sign_with_keys(keys) + .expect("sign workflow trigger") +} + +#[tokio::test] +#[ignore] +async fn agent_and_human_owner_trigger_exact_revision_but_unrelated_member_cannot() { + let owner = Keys::generate(); + let agent = Keys::generate(); + let unrelated = Keys::generate(); + let channel_id = uuid::Uuid::new_v4(); + let workflow_id = uuid::Uuid::new_v4(); + + // NIP-OA authentication materializes the immutable community-scoped + // agent→owner relationship used by workflow authorization. + let mut agent_client = connect_agent_with_owner(&agent, &owner).await; + + let create_channel = EventBuilder::new(Kind::Custom(9007), "") + .tags([ + Tag::parse(["h", &channel_id.to_string()]).unwrap(), + Tag::parse(["name", "workflow-agent-owner-e2e"]).unwrap(), + Tag::parse(["channel_type", "stream"]).unwrap(), + Tag::parse(["visibility", "open"]).unwrap(), + ]) + .sign_with_keys(&agent) + .unwrap(); + let response = agent_client + .send_event(create_channel) + .await + .expect("create channel"); + assert!(response.accepted, "channel rejected: {}", response.message); + + let definition = + buzz_sdk::build_workflow_def(channel_id, workflow_id, &workflow_yaml("owner-triggered")) + .expect("build workflow definition") + .sign_with_keys(&agent) + .expect("sign workflow definition"); + let revision = definition.id.to_hex(); + let response = agent_client + .send_event(definition) + .await + .expect("define workflow"); + assert!( + response.accepted, + "definition rejected: {}", + response.message + ); + + let add_unrelated = EventBuilder::new(Kind::Custom(9000), "") + .tags([ + Tag::parse(["h", &channel_id.to_string()]).unwrap(), + Tag::parse(["p", &unrelated.public_key().to_hex()]).unwrap(), + ]) + .sign_with_keys(&agent) + .expect("sign add-member event"); + let response = agent_client + .send_event(add_unrelated) + .await + .expect("add unrelated member"); + assert!( + response.accepted, + "add member rejected: {}", + response.message + ); + + let response = agent_client + .send_event(trigger(&agent, workflow_id, &revision)) + .await + .expect("send agent trigger"); + assert!( + response.accepted, + "agent trigger rejected: {}", + response.message + ); + + let mut unrelated_client = BuzzTestClient::connect(&relay_url(), &unrelated) + .await + .expect("connect unrelated user"); + let response = unrelated_client + .send_event(trigger(&unrelated, workflow_id, &revision)) + .await + .expect("send unrelated trigger"); + assert!(!response.accepted, "unrelated member triggered workflow"); + assert!(response.message.contains("not authorized to trigger")); + + let mut owner_client = BuzzTestClient::connect(&relay_url(), &owner) + .await + .expect("connect human owner"); + let response = owner_client + .send_event(trigger(&owner, workflow_id, &revision)) + .await + .expect("send owner trigger"); + assert!( + response.accepted, + "owner trigger rejected: {}", + response.message + ); + + let missing_revision = EventBuilder::new(Kind::Custom(46020), "") + .tags([Tag::parse(["d", &workflow_id.to_string()]).unwrap()]) + .sign_with_keys(&owner) + .unwrap(); + let response = owner_client + .send_event(missing_revision) + .await + .expect("send unbound trigger"); + assert!(!response.accepted, "trigger without revision was accepted"); + assert!(response.message.contains("workflow revision e tag")); + + let wrong_revision = "ab".repeat(32); + let response = owner_client + .send_event(trigger(&owner, workflow_id, &wrong_revision)) + .await + .expect("send wrong-revision trigger"); + assert!(!response.accepted, "wrong revision was accepted"); + assert!(response + .message + .contains("does not match current definition")); + + // A valid old revision becomes stale immediately after an agent-signed update. + // NIP-33 revisions use second-granularity timestamps. Advance past the + // creation second so this update is newer regardless of the event-ID + // tie-breaker for equal timestamps. + tokio::time::sleep(std::time::Duration::from_secs(1)).await; + let update = buzz_sdk::build_workflow_update( + channel_id, + workflow_id, + &workflow_yaml("updated"), + &revision, + ) + .expect("build workflow update") + .sign_with_keys(&agent) + .expect("sign workflow update"); + let current_revision = update.id.to_hex(); + let response = agent_client + .send_event(update) + .await + .expect("update workflow"); + assert!(response.accepted, "update rejected: {}", response.message); + + let response = owner_client + .send_event(trigger(&owner, workflow_id, &revision)) + .await + .expect("send stale trigger"); + assert!(!response.accepted, "stale revision was accepted"); + assert!(response + .message + .contains("does not match current definition")); + + let response = owner_client + .send_event(trigger(&owner, workflow_id, ¤t_revision)) + .await + .expect("send current trigger"); + assert!( + response.accepted, + "current revision rejected: {}", + response.message + ); + + agent_client.disconnect().await.ok(); + unrelated_client.disconnect().await.ok(); + owner_client.disconnect().await.ok(); +} diff --git a/crates/buzz-workflow/src/executor.rs b/crates/buzz-workflow/src/executor.rs index 5c712dcff7c..9aa9d4c3a2e 100644 --- a/crates/buzz-workflow/src/executor.rs +++ b/crates/buzz-workflow/src/executor.rs @@ -43,6 +43,9 @@ pub struct TriggerContext { pub is_reply: bool, /// Arbitrary webhook body fields (webhook trigger). pub webhook_fields: HashMap, + /// Exact owner-signed kind:30620 definition revision executed by this run. + #[serde(default)] + pub definition_event_id: String, } impl TriggerContext { @@ -1312,6 +1315,7 @@ mod tests { message_id: "event-id-hex".to_owned(), is_reply: false, webhook_fields: HashMap::new(), + definition_event_id: String::new(), } } diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index bceb6d8bd8d..b8445c9b952 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -44,7 +44,9 @@ use std::collections::HashMap; use std::sync::Arc; use std::sync::OnceLock; -use buzz_core::kind::{event_kind_u32, is_workflow_execution_kind, KIND_REACTION}; +use buzz_core::kind::{ + event_kind_u32, is_workflow_execution_kind, KIND_REACTION, KIND_WORKFLOW_DEF, +}; use buzz_core::tenant::CommunityId; use buzz_db::workflow::RunStatus; use buzz_db::Db; @@ -122,6 +124,62 @@ impl WorkflowEngine { } } + /// Load and verify the exact owner-signed definition bound to a run. + /// + /// The mutable `workflows` row supplies only immutable identity/channel + /// binding. Definition content always comes from the run's signed event; + /// legacy runs without a revision fail closed. + pub async fn load_run_definition( + &self, + community_id: CommunityId, + run_id: Uuid, + ) -> Result<(buzz_db::workflow::WorkflowRunRecord, WorkflowDef), WorkflowError> { + let run = self.db.get_workflow_run(community_id, run_id).await?; + let revision = run.definition_event_id.as_deref().ok_or_else(|| { + WorkflowError::InvalidDefinition( + "workflow run has no owner-signed definition revision".into(), + ) + })?; + let workflow = self.db.get_workflow(community_id, run.workflow_id).await?; + let stored = self + .db + .get_event_by_id_including_deleted(community_id, revision) + .await? + .ok_or_else(|| { + WorkflowError::InvalidDefinition( + "workflow run definition event is unavailable".into(), + ) + })?; + let event = &stored.event; + let workflow_id = run.workflow_id.to_string(); + let channel_id = workflow.channel_id.map(|id| id.to_string()); + let exact_tag = |name: &str| { + let mut values = event.tags.iter().filter_map(|tag| { + (tag.kind().to_string() == name) + .then(|| tag.content()) + .flatten() + }); + let value = values.next(); + value.filter(|_| values.next().is_none()) + }; + if event.id.as_bytes() != revision + || !event.verify_id() + || !event.verify_signature() + || event_kind_u32(event) != KIND_WORKFLOW_DEF + || event.pubkey.to_bytes().as_slice() != workflow.owner_pubkey + || exact_tag("d") != Some(workflow_id.as_str()) + || channel_id.is_none() + || exact_tag("h") != channel_id.as_deref() + || stored.channel_id != workflow.channel_id + { + return Err(WorkflowError::InvalidDefinition( + "workflow run definition event binding mismatch".into(), + )); + } + let (definition, _) = Self::parse_yaml(&event.content)?; + Ok((run, definition)) + } + /// Drop the cached enabled-workflow list for a channel. /// /// Must be called after any write to a workflow's trigger eligibility or @@ -407,6 +465,13 @@ impl WorkflowEngine { .create_workflow_run( community_id, workflow.id, + match workflow.definition_event_id.as_deref() { + Some(revision) => revision, + None => { + tracing::warn!(workflow_id = %workflow.id, "Skipping workflow — signed revision unavailable"); + continue; + } + }, Some(&trigger_event_id_bytes), Some(&trigger_ctx_json), ) @@ -426,13 +491,22 @@ impl WorkflowEngine { ); let engine = Arc::clone(self); - 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; + let result = match engine.load_run_definition(community_id, run_id).await { + Ok((_, definition)) => { + executor::execute_run( + &engine, + community_id, + run_id, + &definition, + &ctx_clone, + ) + .await + } + Err(error) => Err((error, PartialProgress::default())), + }; engine .finalize_run(community_id, run_id, result, None) .await; @@ -669,6 +743,13 @@ impl WorkflowEngine { .create_workflow_run( community_id, workflow.id, + match workflow.definition_event_id.as_deref() { + Some(revision) => revision, + None => { + tracing::warn!(workflow_id = %workflow.id, "Cron tick: signed revision unavailable"); + continue; + } + }, None, // no trigger event for cron trigger_ctx_json.as_ref(), ) @@ -721,17 +802,21 @@ impl WorkflowEngine { ); let engine = Arc::clone(self); - 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; + let result = match engine.load_run_definition(community_id, run_id).await { + Ok((_, definition)) => { + executor::execute_run( + &engine, + community_id, + run_id, + &definition, + &ctx_clone, + ) + .await + } + Err(error) => Err((error, PartialProgress::default())), + }; engine .finalize_run(community_id, run_id, result, None) .await; @@ -999,6 +1084,7 @@ pub fn build_trigger_context(event: &buzz_core::StoredEvent) -> executor::Trigge message_id, is_reply: event_is_reply(&event.event), webhook_fields: HashMap::new(), + definition_event_id: String::new(), } } @@ -1974,17 +2060,25 @@ steps: "enabled": true, }) .to_string(); - let workflow_id = db - .create_workflow( - community, - Some(channel_id), - &member, - "sec006-event", - &def_json, - &[0u8; 32], - ) - .await - .expect("create workflow"); + // Seed through the production insert path (`upsert_workflow`), which + // binds a signed definition revision: run creation now skips any + // workflow without one (legacy NULL-revision rows fail closed). + let workflow_id = Uuid::new_v4(); + let mut tx = db.begin_transaction().await.expect("begin tx"); + db.upsert_workflow( + &mut tx, + community, + workflow_id, + Some(channel_id), + &member, + "sec006-event", + &def_json, + &[0u8; 32], + &[0x42u8; 32], + ) + .await + .expect("create workflow"); + tx.commit().await.expect("commit workflow"); let engine = Arc::new(WorkflowEngine::new(db.clone(), WorkflowConfig::default())); @@ -2039,29 +2133,41 @@ steps: }) .to_string(); - // Same definition, two owners: plain member vs channel owner. - let wf_member = db - .create_workflow( - community, - Some(channel_id), - &member, - "hook-member", - &def_json, - &[0u8; 32], - ) - .await - .expect("create member workflow"); - let wf_owner = db - .create_workflow( - community, - Some(channel_id), - &creator, - "hook-owner", - &def_json, - &[1u8; 32], - ) - .await - .expect("create owner workflow"); + // Same definition, two owners: plain member vs channel owner. Seed + // through `upsert_workflow` so each row carries a bound signed + // revision (run creation skips NULL-revision rows fail-closed). + let wf_member = Uuid::new_v4(); + let mut tx = db.begin_transaction().await.expect("begin member tx"); + db.upsert_workflow( + &mut tx, + community, + wf_member, + Some(channel_id), + &member, + "hook-member", + &def_json, + &[0u8; 32], + &[0x42u8; 32], + ) + .await + .expect("create member workflow"); + tx.commit().await.expect("commit member workflow"); + let wf_owner = Uuid::new_v4(); + let mut tx = db.begin_transaction().await.expect("begin owner tx"); + db.upsert_workflow( + &mut tx, + community, + wf_owner, + Some(channel_id), + &creator, + "hook-owner", + &def_json, + &[1u8; 32], + &[0x43u8; 32], + ) + .await + .expect("create owner workflow"); + tx.commit().await.expect("commit owner workflow"); let engine = Arc::new(WorkflowEngine::new(db.clone(), WorkflowConfig::default())); engine @@ -2087,4 +2193,89 @@ steps: "channel owner's call_webhook workflow fires" ); } + + /// A run is bound to the exact signed revision persisted at creation. + /// Replacing the definition (NIP-33) soft-deletes the old kind-30620 + /// event, but historical runs must keep loading their bound revision — + /// `load_run_definition` reads through soft deletion. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn run_definition_loads_after_revision_soft_deleted() { + let db = setup_db().await; + let owner_keys = nostr::Keys::generate(); + let owner = owner_keys.public_key().to_bytes().to_vec(); + let member = nostr::Keys::generate().public_key().to_bytes().to_vec(); + let (community, channel_id) = setup_channel(&db, &owner, &member).await; + + let workflow_id = Uuid::new_v4(); + let yaml = concat!( + "name: revision-history\n", + "trigger:\n on: message_posted\n", + "steps:\n - id: s1\n action: send_message\n text: hi\n", + ); + let definition_event = + nostr::EventBuilder::new(nostr::Kind::Custom(KIND_WORKFLOW_DEF as u16), yaml) + .tags([ + nostr::Tag::parse(["d", &workflow_id.to_string()]).expect("d tag"), + nostr::Tag::parse(["h", &channel_id.to_string()]).expect("h tag"), + ]) + .sign_with_keys(&owner_keys) + .expect("sign definition"); + db.insert_event(community, &definition_event, Some(channel_id)) + .await + .expect("store definition event"); + + let (def, _) = WorkflowEngine::parse_yaml(yaml).expect("parse yaml"); + let def_json = serde_json::to_string(&def).expect("serialize definition"); + let mut tx = db.begin_transaction().await.expect("begin tx"); + db.upsert_workflow( + &mut tx, + community, + workflow_id, + Some(channel_id), + &owner, + "revision-history", + &def_json, + &[0u8; 32], + definition_event.id.as_bytes(), + ) + .await + .expect("create workflow"); + tx.commit().await.expect("commit workflow"); + + let run_id = db + .create_workflow_run( + community, + workflow_id, + definition_event.id.as_bytes(), + None, + None, + ) + .await + .expect("create run"); + + let engine = Arc::new(WorkflowEngine::new(db.clone(), WorkflowConfig::default())); + + // Bound revision loads while the event is live. + engine + .load_run_definition(community, run_id) + .await + .expect("load definition before deletion"); + + // Definition replacement soft-deletes the superseded revision event. + assert!( + db.soft_delete_event(community, definition_event.id.as_bytes()) + .await + .expect("soft delete revision"), + "revision event must exist to be soft-deleted" + ); + + // The historical run still resolves its exact signed revision. + let (run, loaded) = engine + .load_run_definition(community, run_id) + .await + .expect("load definition after soft deletion"); + assert_eq!(run.id, run_id); + assert_eq!(loaded.name, "revision-history"); + } } diff --git a/desktop/src-tauri/src/commands/workflows.rs b/desktop/src-tauri/src/commands/workflows.rs index c4e5d38c8ba..31aa948b93a 100644 --- a/desktop/src-tauri/src/commands/workflows.rs +++ b/desktop/src-tauri/src/commands/workflows.rs @@ -323,7 +323,13 @@ pub async fn trigger_workflow( workflow_id: String, state: State<'_, AppState>, ) -> Result { - let builder = events::build_workflow_trigger(&workflow_id)?; + // Resolve the current signed definition at trigger time. The relay binds + // authorization and execution to this exact revision and rejects a stale + // result if an update races this command. + let revision: nostr::Event = + get_relay_json(&state, &format!("/workflows/{workflow_id}/revision")).await?; + let definition_event_id = revision.id.to_hex(); + let builder = events::build_workflow_trigger(&workflow_id, &definition_event_id)?; let result = submit_event(builder, &state).await?; trigger_wire_from_message(workflow_id, &result.message) } diff --git a/desktop/src-tauri/src/events.rs b/desktop/src-tauri/src/events.rs index 1828b3f5605..8c57e0d9767 100644 --- a/desktop/src-tauri/src/events.rs +++ b/desktop/src-tauri/src/events.rs @@ -769,6 +769,31 @@ pub use workflows::{ mod tests { use super::*; use nostr::Keys; + #[test] + fn workflow_trigger_binds_exact_definition_revision() { + let workflow_id = Uuid::new_v4().to_string(); + let revision = "ab".repeat(32); + let event = build_workflow_trigger(&workflow_id, &revision) + .expect("build workflow trigger") + .sign_with_keys(&Keys::generate()) + .expect("sign workflow trigger"); + let tags: Vec> = event + .tags + .iter() + .map(|tag| tag.as_slice().to_vec()) + .collect(); + + assert_eq!(event.kind, Kind::Custom(46020)); + assert_eq!( + tags, + vec![ + vec!["d".to_string(), workflow_id.clone()], + vec!["e".to_string(), revision.clone()], + ] + ); + assert!(build_workflow_trigger(&workflow_id, "not-an-event-id").is_err()); + } + #[test] fn channel_builders_reject_hash_only_names() { let channel_id = Uuid::new_v4(); diff --git a/desktop/src-tauri/src/events/workflows.rs b/desktop/src-tauri/src/events/workflows.rs index 8615f73851f..b73f8fca4f8 100644 --- a/desktop/src-tauri/src/events/workflows.rs +++ b/desktop/src-tauri/src/events/workflows.rs @@ -31,9 +31,16 @@ pub fn build_workflow_delete( Ok(EventBuilder::new(Kind::Custom(5), "").tags(tags)) } -/// Kind 46020 — trigger a workflow run by id. -pub fn build_workflow_trigger(workflow_id: &str) -> Result { - let tags = vec![tag(vec!["d", workflow_id])?]; +/// Kind 46020 — trigger a workflow run by id, bound to one exact definition revision. +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)) } diff --git a/migrations/0033_workflow_definition_event_id.sql b/migrations/0033_workflow_definition_event_id.sql new file mode 100644 index 00000000000..c733a83850c --- /dev/null +++ b/migrations/0033_workflow_definition_event_id.sql @@ -0,0 +1,6 @@ +-- Bind each materialized workflow row to the exact owner-signed kind:30620 +-- revision that produced it. Existing rows remain nullable until re-saved; +-- revision-bound execution fails closed when the revision is unavailable. +ALTER TABLE workflows + ADD COLUMN definition_event_id BYTEA + CHECK (definition_event_id IS NULL OR octet_length(definition_event_id) = 32); diff --git a/migrations/0034_workflow_run_definition_event_id.sql b/migrations/0034_workflow_run_definition_event_id.sql new file mode 100644 index 00000000000..b89222de2dd --- /dev/null +++ b/migrations/0034_workflow_run_definition_event_id.sql @@ -0,0 +1,5 @@ +-- Bind each new workflow run to the exact signed definition it executes. +-- Existing rows remain NULL and all resume/execution paths fail them closed. +ALTER TABLE workflow_runs + ADD COLUMN definition_event_id BYTEA + CHECK (definition_event_id IS NULL OR octet_length(definition_event_id) = 32); diff --git a/schema/schema.sql b/schema/schema.sql index 6e14e6be1bf..f2230e7747c 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -368,6 +368,11 @@ CREATE TABLE workflows ( channel_id UUID, definition JSONB NOT NULL, definition_hash BYTEA NOT NULL, + -- Exact owner-signed kind:30620 revision that materialized this row. + -- Nullable only for pre-0033 rows; revision-bound execution fails closed until re-saved. + definition_event_id BYTEA CHECK ( + definition_event_id IS NULL OR octet_length(definition_event_id) = 32 + ), status workflow_status NOT NULL DEFAULT 'active', enabled BOOLEAN NOT NULL DEFAULT TRUE, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), @@ -388,6 +393,9 @@ CREATE TABLE workflow_runs ( community_id UUID NOT NULL REFERENCES communities(id), id UUID NOT NULL DEFAULT gen_random_uuid(), workflow_id UUID NOT NULL, + definition_event_id BYTEA CHECK ( + definition_event_id IS NULL OR octet_length(definition_event_id) = 32 + ), status run_status NOT NULL DEFAULT 'pending', trigger_event_id BYTEA, current_step INT NOT NULL DEFAULT 0,