diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index b37dcedff8c..e7ec2a0f000 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -3330,6 +3330,7 @@ impl Db { #[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 +3338,10 @@ impl Db { name: &str, definition_json: &str, definition_hash: &[u8], + definition_event_id: &[u8], ) -> Result<()> { workflow::upsert_workflow( - &self.pool, + tx, community_id, id, channel_id, @@ -3347,10 +3349,27 @@ impl Db { name, definition_json, definition_hash, + definition_event_id, ) .await } + /// List workflows that have not yet captured an exact signed revision. + #[datastore_span(name = "list_legacy_workflows", system = "postgresql")] + pub async fn list_legacy_workflows(&self) -> Result> { + workflow::list_legacy_workflows(&self.pool).await + } + + /// Compare-and-set an exact revision onto an unchanged legacy workflow snapshot. + #[datastore_span(name = "bind_legacy_workflow_revision", system = "postgresql")] + pub async fn bind_legacy_workflow_revision( + &self, + workflow: &workflow::WorkflowRecord, + definition_event_id: &[u8], + ) -> Result { + workflow::bind_legacy_workflow_revision(&self.pool, workflow, definition_event_id).await + } + /// Fetch a single workflow by ID, scoped to its community. #[datastore_span(name = "get_workflow", system = "postgresql")] pub async fn get_workflow( @@ -3548,6 +3567,7 @@ impl Db { &self, community_id: CommunityId, workflow_id: Uuid, + definition_event_id: Option<&[u8]>, trigger_event_id: Option<&[u8]>, trigger_context: Option<&serde_json::Value>, ) -> Result { @@ -3555,6 +3575,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 c91577b7607..25301395b3e 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -649,7 +649,7 @@ mod tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 34); + assert_eq!(migrations.len(), 35); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); assert!(migrations[0] @@ -1112,6 +1112,24 @@ mod tests { assert!(heartbeat_vacuum.contains("vacuum_truncate = false")); assert!(desired_schema.contains("vacuum_truncate = false")); + // Workflow revision capture is additive: nullable 32-byte event IDs on + // both the materialized definition and run, with identical fresh-schema + // constraints and no backfill hidden in startup migration state. + assert_eq!(migrations[34].version, 35); + let workflow_revision_binding = migrations[34].sql.as_str(); + assert!(workflow_revision_binding.contains("ALTER TABLE workflows")); + assert!(workflow_revision_binding.contains("ALTER TABLE workflow_runs")); + assert!(workflow_revision_binding.contains("octet_length(definition_event_id) = 32")); + assert!(!workflow_revision_binding + .to_ascii_lowercase() + .contains("update ")); + assert_eq!( + desired_schema + .matches("octet_length(definition_event_id) = 32") + .count(), + 2 + ); + // pgschema intentionally reconciles DDL, not seed DML or table storage // parameters. Its post-apply reconciliation must restore and verify // both parts of the live heartbeat contract for fresh bootstraps. diff --git a/crates/buzz-db/src/workflow.rs b/crates/buzz-db/src/workflow.rs index e970e978aaf..39c7f7303a9 100644 --- a/crates/buzz-db/src/workflow.rs +++ b/crates/buzz-db/src/workflow.rs @@ -177,6 +177,9 @@ 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. + /// NULL is retained for workflows created before revision capture. + 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 +204,9 @@ pub struct WorkflowRunRecord { pub community_id: CommunityId, /// The workflow definition that was executed. pub workflow_id: Uuid, + /// Exact owner-signed kind:30620 revision selected when this run was created. + /// NULL is retained when the workflow has no captured revision yet. + 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 +320,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, + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, community_id: CommunityId, id: Uuid, channel_id: Option, @@ -322,16 +328,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 +353,8 @@ pub async fn upsert_workflow( .bind(channel_id) .bind(definition_json) .bind(definition_hash) - .fetch_optional(pool) + .bind(definition_event_id) + .fetch_optional(&mut **tx) .await?; if row.is_none() { @@ -370,7 +379,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 +394,61 @@ pub async fn get_workflow( row_to_workflow_record(row) } +/// List workflows that predate exact signed revision capture. +pub async fn list_legacy_workflows(pool: &PgPool) -> Result> { + let rows = 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 definition_event_id IS NULL + ORDER BY community_id, created_at, id + "#, + ) + .fetch_all(pool) + .await?; + + rows.into_iter().map(row_to_workflow_record).collect() +} + +/// Bind a legacy workflow to a provenance-checked signed definition. +/// +/// The validated workflow snapshot is part of the compare-and-set. This keeps +/// a legacy writer in a rolling deployment from changing the materialized +/// workflow between reconciliation's proof check and this bind. +pub async fn bind_legacy_workflow_revision( + pool: &PgPool, + workflow: &WorkflowRecord, + definition_event_id: &[u8], +) -> Result { + let result = sqlx::query( + r#" + UPDATE workflows + SET definition_event_id = $8 + WHERE community_id = $1 + AND id = $2 + AND owner_pubkey = $3 + AND channel_id IS NOT DISTINCT FROM $4 + AND definition = $5 + AND definition_hash = $6 + AND updated_at = $7 + AND definition_event_id IS NULL + "#, + ) + .bind(workflow.community_id.as_uuid()) + .bind(workflow.id) + .bind(&workflow.owner_pubkey) + .bind(workflow.channel_id) + .bind(&workflow.definition) + .bind(&workflow.definition_hash) + .bind(workflow.updated_at) + .bind(definition_event_id) + .execute(pool) + .await?; + + Ok(result.rows_affected() == 1) +} + /// List workflows for a channel, ordered newest first. /// /// `limit` is capped at [`LIST_MAX_LIMIT`]. Pass `None` to use [`LIST_DEFAULT_LIMIT`]. @@ -401,7 +465,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 +496,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 +524,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 @@ -802,6 +866,7 @@ pub async fn create_workflow_run( pool: &PgPool, community_id: CommunityId, workflow_id: Uuid, + definition_event_id: Option<&[u8]>, trigger_event_id: Option<&[u8]>, trigger_context: Option<&serde_json::Value>, ) -> Result { @@ -810,13 +875,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 +899,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 +931,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 +1249,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 +1269,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 +1315,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 @@ -1381,6 +1449,7 @@ mod tests { owner_pubkey: vec![0xab; 32], channel_id: Some(channel_id), definition: def.clone(), + definition_event_id: None, definition_hash: vec![0x01, 0x02, 0x03, 0x04], status: WorkflowStatus::Active, enabled: true, @@ -1411,6 +1480,7 @@ mod tests { owner_pubkey: vec![0x00; 32], channel_id: None, definition: serde_json::json!({}), + definition_event_id: None, definition_hash: vec![], status: WorkflowStatus::Active, enabled: true, @@ -1433,6 +1503,7 @@ mod tests { owner_pubkey: vec![0x01; 32], channel_id: None, definition: serde_json::json!({}), + definition_event_id: None, definition_hash: vec![0xAA], status: WorkflowStatus::Active, enabled: true, @@ -1462,6 +1533,7 @@ mod tests { owner_pubkey: vec![], channel_id: None, definition: serde_json::json!({}), + definition_event_id: None, definition_hash: vec![], status: status.clone(), enabled: true, @@ -1482,6 +1554,7 @@ mod tests { owner_pubkey: vec![], channel_id: None, definition: serde_json::json!({}), + definition_event_id: None, definition_hash: vec![], status: WorkflowStatus::Active, enabled: false, @@ -1505,6 +1578,7 @@ mod tests { id, community_id: CommunityId::from_uuid(Uuid::new_v4()), workflow_id, + definition_event_id: None, status: RunStatus::Running, trigger_event_id: Some(trigger_event_id.clone()), current_step: 2, @@ -1536,6 +1610,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 +1635,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 +1668,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 +1692,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, @@ -1841,6 +1919,202 @@ mod tests { (workflow_id, community) } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn legacy_revision_binding_is_idempotent_and_never_rewrites_runs() { + let pool = setup_pool().await; + let community = make_community(&pool).await; + let (workflow_id, _) = make_workflow_in(&pool, community).await; + let run_id = create_workflow_run(&pool, community, workflow_id, None, None, None) + .await + .expect("create legacy run"); + let first_revision = [0x31; 32]; + let competing_revision = [0x32; 32]; + + let workflow = get_workflow(&pool, community, workflow_id) + .await + .expect("read legacy workflow"); + assert!( + bind_legacy_workflow_revision(&pool, &workflow, &first_revision) + .await + .expect("bind revision") + ); + assert!( + !bind_legacy_workflow_revision(&pool, &workflow, &competing_revision) + .await + .expect("repeat binding") + ); + + assert_eq!( + get_workflow(&pool, community, workflow_id) + .await + .expect("read workflow") + .definition_event_id + .as_deref(), + Some(first_revision.as_slice()) + ); + assert!(get_workflow_run(&pool, community, run_id) + .await + .expect("read historical run") + .definition_event_id + .is_none()); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn legacy_revision_binding_rejects_a_rewritten_snapshot() { + let pool = setup_pool().await; + let community = make_community(&pool).await; + let (workflow_id, _) = make_workflow_in(&pool, community).await; + let snapshot = get_workflow(&pool, community, workflow_id) + .await + .expect("read legacy workflow snapshot"); + + let rewritten_definition = r#"{"trigger":{"on":"schedule"},"steps":[{"id":"changed","action":"send_message","text":"new"}]}"#; + let rewritten_hash = [0x77; 32]; + sqlx::query( + r#" + UPDATE workflows + SET definition = $3::jsonb, definition_hash = $4, updated_at = NOW() + WHERE community_id = $1 AND id = $2 + "#, + ) + .bind(community.as_uuid()) + .bind(workflow_id) + .bind(rewritten_definition) + .bind(rewritten_hash.as_slice()) + .execute(&pool) + .await + .expect("simulate legacy writer rewrite"); + + assert!( + !bind_legacy_workflow_revision(&pool, &snapshot, &[0x41; 32]) + .await + .expect("reject stale snapshot") + ); + let current = get_workflow(&pool, community, workflow_id) + .await + .expect("read rewritten workflow"); + assert!(current.definition_event_id.is_none()); + assert_eq!(current.definition_hash, rewritten_hash); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn legacy_revision_binding_rejects_an_equal_definition_rewrite() { + let pool = setup_pool().await; + let community = make_community(&pool).await; + let (workflow_id, _) = make_workflow_in(&pool, community).await; + let snapshot = get_workflow(&pool, community, workflow_id) + .await + .expect("read legacy workflow snapshot"); + + sqlx::query( + r#" + UPDATE workflows + SET definition = definition, + definition_hash = definition_hash, + updated_at = updated_at + INTERVAL '1 microsecond' + WHERE community_id = $1 AND id = $2 + "#, + ) + .bind(community.as_uuid()) + .bind(workflow_id) + .execute(&pool) + .await + .expect("simulate equal-materialization legacy rewrite"); + + assert!( + !bind_legacy_workflow_revision(&pool, &snapshot, &[0x42; 32]) + .await + .expect("reject stale generation") + ); + assert!(get_workflow(&pool, community, workflow_id) + .await + .expect("read rewritten workflow") + .definition_event_id + .is_none()); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn revision_capture_is_additive_and_transactional() { + let pool = setup_pool().await; + let community = make_community(&pool).await; + let (workflow_id, _) = make_workflow_in(&pool, community).await; + let legacy = get_workflow(&pool, community, workflow_id) + .await + .expect("read legacy workflow"); + assert!(legacy.definition_event_id.is_none()); + + let revision = [0x42; 32]; + let legacy_run = create_workflow_run(&pool, community, workflow_id, None, None, None) + .await + .expect("create compatible legacy run"); + let bound_run = + create_workflow_run(&pool, community, workflow_id, Some(&revision), None, None) + .await + .expect("create revision-bound run"); + assert!(get_workflow_run(&pool, community, legacy_run) + .await + .expect("read legacy run") + .definition_event_id + .is_none()); + assert_eq!( + get_workflow_run(&pool, community, bound_run) + .await + .expect("read bound run") + .definition_event_id + .as_deref(), + Some(revision.as_slice()) + ); + + let mut tx = pool.begin().await.expect("begin revision update"); + upsert_workflow( + &mut tx, + community, + workflow_id, + legacy.channel_id, + &legacy.owner_pubkey, + &legacy.name, + &legacy.definition.to_string(), + &legacy.definition_hash, + &revision, + ) + .await + .expect("capture revision"); + tx.rollback().await.expect("roll back revision update"); + assert!(get_workflow(&pool, community, workflow_id) + .await + .expect("read rolled-back workflow") + .definition_event_id + .is_none()); + + let mut tx = pool.begin().await.expect("begin committed update"); + upsert_workflow( + &mut tx, + community, + workflow_id, + legacy.channel_id, + &legacy.owner_pubkey, + &legacy.name, + &legacy.definition.to_string(), + &legacy.definition_hash, + &revision, + ) + .await + .expect("capture committed revision"); + tx.commit().await.expect("commit revision update"); + assert_eq!( + get_workflow(&pool, community, workflow_id) + .await + .expect("read revision-bound workflow") + .definition_event_id + .as_deref(), + Some(revision.as_slice()) + ); + } + /// Confinement: a duplicate workflow UUID existing in both community A and /// community B must claim independently. Claiming `(A, id, t)` must NOT /// consume `(B, id, t)` — B's identical instant stays claimable, and the @@ -1963,7 +2237,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, None, None, None) .await .expect("create run ok"); @@ -1992,7 +2266,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, None, None, None) .await .expect("create second run ok"); let reattached = @@ -2273,10 +2547,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, None, 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, None, 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 5dbb2aaf50c..57b65320758 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -2115,7 +2115,13 @@ pub async fn workflow_webhook( let run_id = state .db - .create_workflow_run(community_id, id, None, trigger_ctx_json.as_ref()) + .create_workflow_run( + community_id, + id, + workflow.definition_event_id.as_deref(), + None, + trigger_ctx_json.as_ref(), + ) .await .map_err(|e| super::internal_error(&format!("db error: {e}")))?; diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 007db43ffd9..ed92fd07fdb 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -20,7 +20,8 @@ use uuid::Uuid; use buzz_core::kind::*; use buzz_core::tenant::{CommunityId, TenantContext}; use buzz_datastore_tracing::datastore_span; -use buzz_db::workflow::{ApprovalStatus, RunStatus}; +use buzz_db::event::EventQuery; +use buzz_db::workflow::{ApprovalStatus, RunStatus, WorkflowRecord}; use buzz_db::DbError; use buzz_workflow::executor::TriggerContext; @@ -92,12 +93,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). +/// Most domain mutations still execute on the connection pool rather than this +/// transaction, so their handlers rely on idempotency if event commit fails. +/// Workflow definition ingest is the exception: it materializes the workflow +/// revision on this same transaction so the signed event and revision pointer +/// commit or roll back together. #[datastore_span(name = "persist_command_event", system = "postgresql")] async fn persist_command_event( db: &buzz_db::Db, @@ -294,6 +294,103 @@ fn compute_definition_hash(json_str: &str) -> Vec { Sha256::digest(json_str.as_bytes()).to_vec() } +/// Outcome counts from one legacy workflow reconciliation pass. +#[derive(Debug, Default, PartialEq, Eq)] +pub struct LegacyWorkflowReconcileReport { + /// Legacy workflow rows examined. + pub examined: usize, + /// Rows bound to an exact signed definition. + pub bound: usize, + /// Rows left on the compatibility fallback because provenance was incomplete. + pub unprovable: usize, + /// Rows concurrently bound by another writer. + pub raced: usize, +} + +fn legacy_definition_matches(workflow: &WorkflowRecord, event: &Event) -> bool { + let workflow_id = workflow.id.to_string(); + let workflow_channel_id = workflow.channel_id.map(|channel| channel.to_string()); + if event.kind.as_u16() as u32 != KIND_WORKFLOW_DEF + || event.pubkey.to_bytes().as_slice() != workflow.owner_pubkey.as_slice() + || extract_d_tag(event).as_deref() != Some(workflow_id.as_str()) + || extract_h_tag(event).as_deref() != workflow_channel_id.as_deref() + { + return false; + } + + let Ok((_, canonical_json)) = buzz_workflow::WorkflowEngine::parse_yaml(&event.content) else { + return false; + }; + let Ok(mut definition) = serde_json::from_str::(&canonical_json) else { + return false; + }; + + // Webhook secrets were generated by the relay and never existed in the + // owner-signed event. Reuse the stored secret before comparing the exact + // materialized JSON and its legacy hash. + if let Some(secret) = webhook_secret::extract_secret(&workflow.definition) { + webhook_secret::inject_secret(&mut definition, &secret); + } + let Ok(materialized_json) = serde_json::to_string(&definition) else { + return false; + }; + + definition == workflow.definition + && compute_definition_hash(&materialized_json) == workflow.definition_hash +} + +/// Reconcile legacy workflow rows only when the live signed NIP-33 head proves +/// the exact materialized definition. Historical runs deliberately remain NULL: +/// the current head cannot prove which earlier revision a completed run used. +pub async fn reconcile_legacy_workflow_revisions( + db: &buzz_db::Db, +) -> Result { + let workflows = db.list_legacy_workflows().await?; + let mut report = LegacyWorkflowReconcileReport { + examined: workflows.len(), + ..Default::default() + }; + + for workflow in workflows { + let Some(channel_id) = workflow.channel_id else { + report.unprovable += 1; + continue; + }; + let mut query = EventQuery::for_community(workflow.community_id); + // Workflow definitions written before exact revision capture were + // retained as channel-less rows even though their signed `h` tag was + // channel-scoped. Search both historical and current storage shapes; + // legacy_definition_matches remains the authority for the signed tag. + query.channel_ids = Some(vec![channel_id]); + query.channel_ids_include_global = true; + query.kinds = Some(vec![KIND_WORKFLOW_DEF as i32]); + query.pubkey = Some(workflow.owner_pubkey.clone()); + query.d_tag = Some(workflow.id.to_string()); + query.limit = Some(1); + + let mut events = db.query_events(&query).await?; + let Some(stored) = events.pop() else { + report.unprovable += 1; + continue; + }; + if !legacy_definition_matches(&workflow, &stored.event) { + report.unprovable += 1; + continue; + } + + if db + .bind_legacy_workflow_revision(&workflow, stored.event.id.as_bytes()) + .await? + { + report.bound += 1; + } else { + report.raced += 1; + } + } + + Ok(report) +} + async fn handle_dm_open( tenant: &TenantContext, state: &Arc, @@ -736,8 +833,8 @@ 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 signed definition and materialized revision atomically. + 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 +866,7 @@ async fn handle_workflow_def( state .db .upsert_workflow( + &mut tx, community_id, workflow_id, Some(channel_id), @@ -776,6 +874,7 @@ async fn handle_workflow_def( &workflow_name, &definition_json_final, &hash, + event.id.as_bytes(), ) .await .map_err(|e| match e { @@ -785,17 +884,15 @@ 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. + // Commit before cache invalidation so a concurrent refill can observe the new revision. tx.commit() .await .map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?; + 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(), @@ -915,6 +1012,7 @@ async fn handle_workflow_trigger( .create_workflow_run( community_id, workflow_id, + workflow.definition_event_id.as_deref(), Some(&event_id_bytes), trigger_ctx_json.as_ref(), ) @@ -1408,7 +1506,7 @@ mod tests { } EventBuilder::new( Kind::Custom(KIND_WORKFLOW_DEF as u16), - format!("name: {name}\ntrigger:\n on: message_posted\nsteps: []\n"), + format!("name: {name}\ntrigger:\n on: message_posted\nsteps:\n - id: notify\n action: send_message\n text: hello\n"), ) .tags(tags) .custom_created_at(Timestamp::from(created_at)) @@ -1425,6 +1523,151 @@ mod tests { } } + fn legacy_workflow_for_event(event: &Event, channel_id: Uuid) -> WorkflowRecord { + let (_, canonical_json) = buzz_workflow::WorkflowEngine::parse_yaml(&event.content) + .expect("canonical workflow definition"); + let definition: serde_json::Value = + serde_json::from_str(&canonical_json).expect("definition json"); + let materialized_json = + serde_json::to_string(&definition).expect("production materialized json"); + WorkflowRecord { + id: Uuid::parse_str(extract_d_tag(event).as_deref().expect("d tag")) + .expect("workflow UUID"), + community_id: CommunityId::from_uuid(Uuid::new_v4()), + name: "legacy".to_string(), + owner_pubkey: event.pubkey.to_bytes().to_vec(), + channel_id: Some(channel_id), + definition_hash: compute_definition_hash(&materialized_json), + definition, + definition_event_id: None, + status: buzz_db::workflow::WorkflowStatus::Active, + enabled: true, + created_at: Utc::now(), + updated_at: Utc::now(), + } + } + + #[test] + fn legacy_revision_match_requires_exact_semantic_provenance() { + let keys = Keys::generate(); + let workflow_id = Uuid::new_v4(); + let event = workflow_event( + &keys, + workflow_id, + Timestamp::now().as_secs(), + None, + "legacy", + ); + let channel_id = Uuid::parse_str(extract_h_tag(&event).as_deref().expect("h tag")) + .expect("channel UUID"); + let workflow = legacy_workflow_for_event(&event, channel_id); + assert!(legacy_definition_matches(&workflow, &event)); + + let mut wrong_hash = workflow.clone(); + wrong_hash.definition_hash = vec![0x55; 32]; + assert!(!legacy_definition_matches(&wrong_hash, &event)); + + let mut wrong_channel = workflow.clone(); + wrong_channel.channel_id = Some(Uuid::new_v4()); + assert!(!legacy_definition_matches(&wrong_channel, &event)); + + let mut wrong_owner = workflow; + wrong_owner.owner_pubkey = Keys::generate().public_key().to_bytes().to_vec(); + assert!(!legacy_definition_matches(&wrong_owner, &event)); + } + + #[test] + fn legacy_webhook_secret_is_reapplied_before_matching() { + let keys = Keys::generate(); + let workflow_id = Uuid::new_v4(); + let channel_id = Uuid::new_v4(); + let event = EventBuilder::new( + Kind::Custom(KIND_WORKFLOW_DEF as u16), + "name: hook\ntrigger:\n on: webhook\nsteps:\n - id: notify\n action: send_message\n text: hello\n", + ) + .tags([ + 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(&keys) + .expect("workflow event"); + let mut workflow = legacy_workflow_for_event(&event, channel_id); + webhook_secret::inject_secret(&mut workflow.definition, "legacy-secret"); + workflow.definition_hash = compute_definition_hash( + &serde_json::to_string(&workflow.definition).expect("materialized definition"), + ); + + assert!(legacy_definition_matches(&workflow, &event)); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn reconciliation_finds_historical_null_channel_event_and_binds_it() { + let (db, tenant) = persistence_test_context().await; + let keys = Keys::generate(); + let owner = keys.public_key().to_bytes(); + db.ensure_user(tenant.community(), owner.as_slice()) + .await + .expect("ensure workflow owner"); + let channel = db + .create_channel( + tenant.community(), + "legacy-workflow", + buzz_db::channel::ChannelType::Stream, + buzz_db::channel::ChannelVisibility::Private, + None, + owner.as_slice(), + None, + ) + .await + .expect("create workflow channel"); + let content = "name: legacy\ntrigger:\n on: message_posted\nsteps:\n - id: notify\n action: send_message\n text: hello\n"; + let (_, canonical_json) = buzz_workflow::WorkflowEngine::parse_yaml(content) + .expect("canonical workflow definition"); + let definition: serde_json::Value = + serde_json::from_str(&canonical_json).expect("production definition value"); + let definition_json_final = + serde_json::to_string(&definition).expect("production materialized definition"); + let workflow_id = db + .create_workflow( + tenant.community(), + Some(channel.id), + owner.as_slice(), + "legacy", + &definition_json_final, + &compute_definition_hash(&definition_json_final), + ) + .await + .expect("create legacy workflow"); + let event = EventBuilder::new(Kind::Custom(KIND_WORKFLOW_DEF as u16), content) + .tags([ + 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(&keys) + .expect("sign historical definition"); + + let (_, inserted) = db + .insert_event(tenant.community(), &event, None) + .await + .expect("store historical channel-less event"); + assert!(inserted); + + let report = reconcile_legacy_workflow_revisions(&db) + .await + .expect("reconcile legacy workflow"); + assert_eq!(report.bound, 1); + assert_eq!(report.raced, 0); + assert_eq!( + db.get_workflow(tenant.community(), workflow_id) + .await + .expect("read reconciled workflow") + .definition_event_id + .as_deref(), + Some(event.id.as_bytes().as_slice()) + ); + } + #[test] fn workflow_revision_parser_accepts_create_and_valid_update() { let revision = [0x42; 32]; diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 3e39b22f8b2..a11418a2ac2 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -363,6 +363,21 @@ async fn main() -> anyhow::Result<()> { Err(e) => error!("Failed to backfill d_tags: {e}"), } + // Bind legacy workflow rows only when the retained signed definition head + // exactly proves the materialized JSON. Runs are intentionally untouched: + // a present-day head cannot prove which historical revision they executed. + match buzz_relay::handlers::command_executor::reconcile_legacy_workflow_revisions(&db).await { + Ok(report) if report.examined > 0 => info!( + examined = report.examined, + bound = report.bound, + unprovable = report.unprovable, + raced = report.raced, + "Reconciled provenance-safe legacy workflow revisions" + ), + Ok(_) => {} + Err(e) => error!("Failed to reconcile legacy workflow revisions: {e}"), + } + let audit = if config.audit_enabled { let audit_pool = sqlx::postgres::PgPoolOptions::new() .max_connections(5) diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index bceb6d8bd8d..4fc4f9d5be8 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -407,6 +407,7 @@ impl WorkflowEngine { .create_workflow_run( community_id, workflow.id, + workflow.definition_event_id.as_deref(), Some(&trigger_event_id_bytes), Some(&trigger_ctx_json), ) @@ -669,6 +670,7 @@ impl WorkflowEngine { .create_workflow_run( community_id, workflow.id, + workflow.definition_event_id.as_deref(), None, // no trigger event for cron trigger_ctx_json.as_ref(), ) diff --git a/migrations/0035_workflow_revision_binding.sql b/migrations/0035_workflow_revision_binding.sql new file mode 100644 index 00000000000..799e18dae22 --- /dev/null +++ b/migrations/0035_workflow_revision_binding.sql @@ -0,0 +1,8 @@ +-- Preserve exact signed workflow revisions without changing legacy execution. +-- Existing rows remain NULL until an explicit, provenance-safe migration. +ALTER TABLE workflows + ADD COLUMN definition_event_id BYTEA + CHECK (definition_event_id IS NULL OR octet_length(definition_event_id) = 32); +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 ce0e10e9329..81ece02ab09 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -368,6 +368,9 @@ CREATE TABLE workflows ( channel_id UUID, definition JSONB NOT NULL, definition_hash BYTEA NOT NULL, + 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 +391,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,