diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8c79231a792..c1f1a80b13d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -707,6 +707,16 @@ jobs: --run-ignored ignored-only env: BUZZ_TEST_DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz + - name: Manual workflow trigger PostgreSQL tests + # Pin delivery identity, result-preserving replay, authorization, and + # exact-revision execution; infra-free unit runs ignore these cases. + run: | + cargo nextest run \ + --archive-file target/ci/backend-integration-tests.tar.zst \ + -E 'package(buzz-relay) and (test(/^handlers::command_executor::tests::/) or test(/revoked_immutable_owner_receives_only_revision_id/))' \ + --run-ignored ignored-only + env: + BUZZ_TEST_DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz - name: Database pressure observability PostgreSQL tests # Explicit pool acquisition and advisory-lock metrics require real # Postgres and are ignored by the infrastructure-free unit-test job. diff --git a/crates/buzz-cli/src/commands/workflows.rs b/crates/buzz-cli/src/commands/workflows.rs index 0028dfc7663..a0c20710da0 100644 --- a/crates/buzz-cli/src/commands/workflows.rs +++ b/crates/buzz-cli/src/commands/workflows.rs @@ -162,44 +162,56 @@ 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 -/// event content (MCP parity). When omitted, the event content is `{}`. +/// event content (MCP parity). When omitted, the event content is empty. pub async fn cmd_trigger_workflow( client: &BuzzClient, workflow_id: &str, inputs: Option<&str>, ) -> Result<(), CliError> { let wf_uuid = parse_uuid(workflow_id)?; + let revision = current_workflow_revision(client, workflow_id).await?; + + let builder = workflow_trigger_builder(wf_uuid, &revision, inputs)?; + let event = client.sign_event(builder)?; + let resp = client.submit_event(event).await?; + println!("{}", normalize_write_response(&resp)); + Ok(()) +} +fn workflow_trigger_builder( + workflow_id: uuid::Uuid, + revision: &str, + inputs: Option<&str>, +) -> Result { if let Some(raw) = inputs { - // Parse and validate it is a JSON object, then build the event manually - // so we can embed the inputs as the event content. let parsed: serde_json::Value = serde_json::from_str(raw) .map_err(|e| CliError::Usage(format!("--inputs is not valid JSON: {e}")))?; - if !parsed.is_object() { - return Err(CliError::Usage("--inputs must be a JSON object".into())); - } - let content = serde_json::to_string(&parsed).unwrap_or_default(); - use nostr::{EventBuilder, Kind, Tag}; - let tags = vec![Tag::parse(["d", &wf_uuid.to_string()]) - .map_err(|e| CliError::Other(format!("tag error: {e}")))?]; - let builder = EventBuilder::new( - Kind::Custom(buzz_sdk::kind::KIND_WORKFLOW_TRIGGER as u16), - &content, - ) - .tags(tags); - let event = client.sign_event(builder)?; - let resp = client.submit_event(event).await?; - println!("{}", normalize_write_response(&resp)); + let object = parsed + .as_object() + .ok_or_else(|| CliError::Usage("--inputs must be a JSON object".into()))?; + buzz_sdk::build_workflow_trigger_with_inputs(workflow_id, revision, object).map_err(sdk_err) } else { - let builder = buzz_sdk::build_workflow_trigger(wf_uuid).map_err(sdk_err)?; - let event = client.sign_event(builder)?; - let resp = client.submit_event(event).await?; - println!("{}", normalize_write_response(&resp)); + buzz_sdk::build_workflow_trigger(workflow_id, revision).map_err(sdk_err) } - Ok(()) } /// Approve or deny a workflow step — sign and submit a kind:46030 (grant) or 46031 (deny) event. @@ -254,3 +266,32 @@ pub async fn dispatch(cmd: crate::WorkflowsCmd, client: &BuzzClient) -> Result<( } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn trigger_builders_preserve_inputs_and_distinguish_same_second_operations() { + let keys = nostr::Keys::generate(); + let workflow_id = uuid::Uuid::new_v4(); + let revision = "ab".repeat(32); + let timestamp = nostr::Timestamp::from(1_700_000_000); + for inputs in [None, Some(r#"{"message":"hello"}"#)] { + let build = || { + workflow_trigger_builder(workflow_id, &revision, inputs) + .unwrap() + .custom_created_at(timestamp) + .sign_with_keys(&keys) + .unwrap() + }; + let first = build(); + let second = build(); + assert_ne!(first.id, second.id); + assert_eq!(first.content, inputs.unwrap_or("")); + } + for invalid in ["not-json", "[]", "null", "1"] { + assert!(workflow_trigger_builder(workflow_id, &revision, Some(invalid)).is_err()); + } + } +} diff --git a/crates/buzz-db/src/store/channel.rs b/crates/buzz-db/src/store/channel.rs index a93ffb36c6a..d8d07556c1d 100644 --- a/crates/buzz-db/src/store/channel.rs +++ b/crates/buzz-db/src/store/channel.rs @@ -21,10 +21,11 @@ pub use buzz_core::channel::{ChannelType, ChannelVisibility, MemberRole}; // Keep the established channel module paths compatible while membership SQL // and invariants live in their dedicated store module. pub use crate::channel_members::{ - add_member, get_accessible_channel_ids, get_accessible_channels, get_bot_members, - get_member_count, get_member_counts_bulk, get_member_role, get_members, get_members_bulk, - get_users_bulk, is_member, list_large_channel_rosters_needing_reconciliation, - lock_member_snapshot, membership_pairs, remove_member, verify_channel_roster_fence_behavior, + acquire_channel_membership_lock, add_member, get_accessible_channel_ids, + get_accessible_channels, get_bot_members, get_member_count, get_member_counts_bulk, + get_member_role, get_member_role_in_transaction, get_members, get_members_bulk, get_users_bulk, + is_member, list_large_channel_rosters_needing_reconciliation, lock_member_snapshot, + membership_pairs, remove_member, verify_channel_roster_fence_behavior, verify_channel_roster_fence_catalog, AccessibleChannel, BotChannelEntry, BotMemberRecord, LargeChannelRoster, LockedMemberSnapshot, MemberRecord, UserRecord, }; diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index f0fd3332acd..a3a1db42083 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/crates/buzz-db/src/store/channel_members.rs @@ -173,7 +173,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, @@ -1378,6 +1378,29 @@ impl Db { } } +/// Get an active member role using the caller's transaction. +/// +/// Authorization callers must acquire [`acquire_channel_membership_lock`] first +/// and hold the transaction through the dependent mutation. +pub async fn get_member_role_in_transaction( + tx: &mut Transaction<'_, Postgres>, + community_id: CommunityId, + channel_id: Uuid, + pubkey: &[u8], +) -> Result> { + let row = sqlx::query( + "SELECT cm.role::text AS role FROM channel_members cm \ + JOIN channels c ON cm.community_id = c.community_id AND cm.channel_id = c.id AND c.deleted_at IS NULL \ + WHERE cm.community_id = $1 AND cm.channel_id = $2 AND cm.pubkey = $3 AND cm.removed_at IS NULL", + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .bind(pubkey) + .fetch_optional(&mut **tx) + .await?; + Ok(row.map(|r| r.try_get("role")).transpose()?) +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index 60e6b05ef9b..8821da30f45 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -1026,6 +1026,26 @@ pub async fn get_event_by_id( } } +/// Fetch a single live event by ID on the caller's transaction. +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. /// @@ -1349,6 +1369,17 @@ pub async fn insert_event_with_thread_metadata( } impl Db { + /// Fetch a live event by ID using 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> { + crate::event::get_event_by_id_in_transaction(tx, community_id, id_bytes).await + } + /// Inserts an event. Returns `(StoredEvent, was_inserted)` — `false` on duplicate. #[datastore_span(name = "insert_event", system = "postgresql")] pub async fn insert_event( diff --git a/crates/buzz-db/src/store/workflow.rs b/crates/buzz-db/src/store/workflow.rs index 6b8dcba1cd0..dd19e06020b 100644 --- a/crates/buzz-db/src/store/workflow.rs +++ b/crates/buzz-db/src/store/workflow.rs @@ -409,6 +409,33 @@ pub async fn get_workflow( row_to_workflow_record(row) } +/// Fetch and share-lock one workflow on an existing transaction. +/// +/// Definition replacement updates this row, so holding this lock through run +/// creation keeps revision validation and the dependent run atomic. +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`]. @@ -817,6 +844,34 @@ pub async fn delete_workflow_for_owner( // -- Workflow Run CRUD -------------------------------------------------------- +/// Insert a new exact-revision workflow run on the caller's transaction. +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 @@ -851,6 +906,24 @@ pub async fn create_workflow_run( Ok(id) } +/// Find the committed result of a manual trigger within its tenant and workflow. +/// The command event and run commit atomically, so a retry can recover this ID. +pub async fn get_workflow_run_id_by_trigger( + pool: &PgPool, + community_id: CommunityId, + workflow_id: Uuid, + trigger_event_id: &[u8], +) -> Result> { + Ok(sqlx::query_scalar( + "SELECT id FROM workflow_runs WHERE community_id = $1 AND workflow_id = $2 AND trigger_event_id = $3", + ) + .bind(community_id.as_uuid()) + .bind(workflow_id) + .bind(trigger_event_id) + .fetch_optional(pool) + .await?) +} + /// Fetch a single workflow run by ID, scoped to its community. pub async fn get_workflow_run( pool: &PgPool, @@ -1297,6 +1370,56 @@ pub async fn find_by_owner_and_name( // -- Run and approval Db API -------------------------------------------------- impl Db { + /// Recover the committed run ID for an exact manual-trigger retry. + #[datastore_span(name = "get_workflow_run_id_by_trigger", system = "postgresql")] + pub async fn get_workflow_run_id_by_trigger( + &self, + community_id: CommunityId, + workflow_id: Uuid, + trigger_event_id: &[u8], + ) -> Result> { + crate::workflow::get_workflow_run_id_by_trigger( + &self.pool, + community_id, + workflow_id, + trigger_event_id, + ) + .await + } + + /// Fetch and share-lock one workflow on an existing transaction. + #[datastore_span(name = "get_workflow_for_share_in_transaction", system = "postgresql")] + pub async fn get_workflow_for_share_in_transaction( + &self, + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + community_id: CommunityId, + id: Uuid, + ) -> Result { + crate::workflow::get_workflow_for_share_in_transaction(tx, community_id, id).await + } + + /// Create an exact-revision 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 { + crate::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( diff --git a/crates/buzz-relay/src/api/workflows.rs b/crates/buzz-relay/src/api/workflows.rs index a3d5a6c729e..7b969141c79 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,57 @@ 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 +/// revision event ID needed to construct a revision-bound manual trigger; the +/// signed definition remains subject to normal channel-read authorization. +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::json!({ "id": event.event.id.to_hex() }))) +} + /// `GET /workflows/{workflow_id}/runs` — one authorized, keyset-paginated page. pub async fn workflow_runs( State(state): State>, @@ -129,8 +172,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 +219,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 @@ -228,8 +278,72 @@ fn approval_json(approval: &buzz_db::workflow::ApprovalRecord) -> Value { #[cfg(test)] mod tests { + use std::sync::Arc; + + use axum::{ + body::{to_bytes, Body}, + http::{header, Request, StatusCode}, + }; + use nostr::{EventBuilder, Keys, Kind, Tag}; + use sha2::Digest; + use tower::ServiceExt; + use super::*; + const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; // sadscan:disable np.postgres.1 + + async fn workflow_test_state(host: &str) -> Arc { + let database_url = std::env::var("BUZZ_TEST_DATABASE_URL") + .or_else(|_| std::env::var("DATABASE_URL")) + .unwrap_or_else(|_| TEST_DB_URL.to_string()); + let redis_url = + std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".to_string()); + let mut config = crate::config::Config::from_env().expect("config from env"); + config.database_url = database_url.clone(); + config.redis_url = redis_url.clone(); + config.relay_url = format!("wss://{host}"); + config.require_auth_token = false; + config.require_relay_membership = false; + + let pool = sqlx::PgPool::connect(&database_url) + .await + .expect("connect workflow API test database"); + // The harness prepares the schema (CI uses pgschema, not SQLx history). + let db = buzz_db::Db::from_pool(pool.clone()); + db.ensure_configured_community(host) + .await + .expect("create workflow API test community"); + let redis_pool = deadpool_redis::Config::from_url(&redis_url) + .create_pool(Some(deadpool_redis::Runtime::Tokio1)) + .expect("redis pool config"); + let pubsub = Arc::new( + buzz_pubsub::PubSubManager::new(&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, + ); + Arc::new(state) + } + #[test] fn request_path_preserves_signed_query_verbatim() { assert_eq!( @@ -261,4 +375,138 @@ mod tests { assert!(wire.get("token").is_none()); assert_eq!(wire["approval_ref"], hex::encode([0xab; 32])); } + + #[tokio::test] + #[ignore = "requires Postgres and Redis"] + async fn revoked_immutable_owner_receives_only_revision_id() { + use buzz_core::channel::{ChannelType, ChannelVisibility, MemberRole}; + + let host = format!("workflow-revision-{}.example", Uuid::new_v4().simple()); + let state = workflow_test_state(&host).await; + let tenant = state + .db + .ensure_configured_community(&host) + .await + .expect("load workflow API test community"); + let community = tenant.id; + let owner = Keys::generate(); + let agent = Keys::generate(); + let owner_bytes = owner.public_key().to_bytes(); + let agent_bytes = agent.public_key().to_bytes(); + state + .db + .ensure_user(community, &owner_bytes) + .await + .expect("ensure immutable owner"); + state + .db + .ensure_user(community, &agent_bytes) + .await + .expect("ensure managed agent"); + assert!(state + .db + .set_agent_owner(community, &agent_bytes, &owner_bytes) + .await + .expect("set immutable owner")); + let channel = state + .db + .create_channel( + community, + "revision-secret-boundary", + ChannelType::Stream, + ChannelVisibility::Private, + None, + &agent_bytes, + None, + ) + .await + .expect("create workflow channel"); + state + .db + .add_member( + community, + channel.id, + &owner_bytes, + MemberRole::Member, + Some(&agent_bytes), + ) + .await + .expect("add immutable owner to workflow channel"); + + let workflow_id = Uuid::new_v4(); + let secret = format!("secret-{}", Uuid::new_v4().simple()); + let yaml = format!( + "name: guarded\ntrigger:\n on: webhook\nsteps:\n - id: call\n action: call_webhook\n url: https://example.invalid\n headers:\n Authorization: Bearer {secret}\n body: '{secret}'\n" + ); + let revision = EventBuilder::new(Kind::Custom(30620), &yaml) + .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 secret-bearing workflow revision"); + let (_, definition_json) = + buzz_workflow::WorkflowEngine::parse_yaml(&yaml).expect("parse workflow YAML"); + let definition_hash = sha2::Sha256::digest(definition_json.as_bytes()); + let mut tx = state + .db + .begin_transaction() + .await + .expect("begin workflow seed"); + buzz_db::event::insert_event_in_transaction( + &mut tx, + community, + &revision, + Some(channel.id), + ) + .await + .expect("persist signed workflow revision"); + state + .db + .upsert_workflow( + &mut tx, + community, + workflow_id, + Some(channel.id), + &agent_bytes, + "guarded", + &definition_json, + definition_hash.as_slice(), + revision.id.as_bytes(), + ) + .await + .expect("materialize workflow revision"); + tx.commit().await.expect("commit workflow seed"); + state + .db + .remove_member(community, channel.id, &owner_bytes, &agent_bytes) + .await + .expect("revoke immutable owner's channel access"); + let tenant_context = TenantContext::resolved(community, host.clone()); + state.invalidate_membership(&tenant_context, channel.id, &owner_bytes); + + let path = format!("/workflows/{workflow_id}/revision"); + let response = crate::router::build_router(Arc::clone(&state)) + .oneshot( + Request::builder() + .method("GET") + .uri(&path) + .header(header::HOST, &host) + .header("x-pubkey", owner.public_key().to_hex()) + .body(Body::empty()) + .expect("workflow revision request"), + ) + .await + .expect("workflow revision response"); + assert_eq!(response.status(), StatusCode::OK); + let bytes = to_bytes(response.into_body(), 1024 * 1024) + .await + .expect("read workflow revision response"); + let body: Value = serde_json::from_slice(&bytes).expect("revision response JSON"); + assert_eq!(body, serde_json::json!({ "id": revision.id.to_hex() })); + let serialized = String::from_utf8(bytes.to_vec()).expect("UTF-8 response"); + assert!(!serialized.contains(&secret)); + assert!(body.get("content").is_none()); + assert!(body.get("tags").is_none()); + } } diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 5ce845af5b4..a5521b9162b 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -92,11 +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. /// -/// 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. +/// 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,7 +736,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 signed definition and materialized revision atomically. + // 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 { @@ -787,11 +788,14 @@ async fn handle_workflow_def( other => IngestError::Internal(format!("error: db upsert_workflow: {other}")), })?; - // Commit before cache invalidation so a concurrent refill can observe the new revision. + // 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); @@ -811,6 +815,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 +918,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 +946,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,17 +989,74 @@ 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(), - accepted: true, - message: "duplicate: already processed".into(), - }); + // A lost response does not lose the result: event and run committed + // together. Read from the primary, including after a concurrent retry. + let run_id = state + .db + .get_workflow_run_id_by_trigger(community_id, workflow_id, event.id.as_bytes()) + .await + .map_err(|e| IngestError::Internal(format!("error: recover workflow run: {e}")))? + .ok_or_else(|| { + IngestError::Internal("error: stored workflow trigger has no run".into()) + })?; + return Ok(workflow_trigger_response(event, run_id)); } PersistResult::Inserted(tx) => tx, }; + // 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 @@ -912,10 +1082,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, - workflow.definition_event_id.as_deref(), + &requested_revision, Some(&event_id_bytes), trigger_ctx_json.as_ref(), ) @@ -929,51 +1100,34 @@ 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; }); - // 6. Return response - Ok(IngestResult { + // 6. Return the same result shape for initial delivery and exact replay. + Ok(workflow_trigger_response(event, run_id)) +} + +fn workflow_trigger_response(event: &Event, run_id: Uuid) -> IngestResult { + IngestResult { event_id: event.id.to_hex(), accepted: true, message: format!( @@ -982,7 +1136,7 @@ async fn handle_workflow_trigger( "run_id": run_id.to_string(), }) ), - }) + } } /// Enforce the approver_spec field against the requesting pubkey. @@ -1295,19 +1449,12 @@ 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, + let (run, def) = match engine.load_run_definition(community_id, run_id).await { + Ok(loaded) => loaded, Err(e) => { - tracing::error!("resume_workflow: failed to parse workflow definition: {e}"); + tracing::error!( + "resume_workflow: failed to load signed definition for run {run_id}: {e}" + ); if let Err(db_err) = db .update_workflow_run( community_id, @@ -1317,7 +1464,7 @@ async fn resume_workflow_after_approval( &run.execution_trace, Some(buzz_db::workflow::WorkflowRunFailure { code: "invalid_definition", - message: &format!("definition parse error: {e}"), + message: &format!("signed run definition unavailable: {e}"), }), ) .await @@ -1328,6 +1475,30 @@ async fn resume_workflow_after_approval( } }; + if run.workflow_id != workflow_id { + tracing::error!( + "resume_workflow: approval workflow {workflow_id} does not match run workflow {}", + run.workflow_id + ); + if let Err(e) = db + .update_workflow_run( + community_id, + run_id, + RunStatus::Failed, + run.current_step, + &run.execution_trace, + Some(buzz_db::workflow::WorkflowRunFailure { + code: "approval_binding_mismatch", + message: "approval does not belong to the workflow run", + }), + ) + .await + { + tracing::error!("resume_workflow: failed to mark mismatched run as failed: {e}"); + } + return; + } + // Reconstruct step_outputs from execution trace for template resolution let mut initial_outputs: std::collections::HashMap = std::collections::HashMap::new(); @@ -1371,17 +1542,46 @@ mod tests { use super::*; use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + #[derive(Default)] + struct RecordingActionSink { + messages: std::sync::Mutex>, + } + + impl buzz_workflow::ActionSink for RecordingActionSink { + fn send_message( + &self, + _community_id: CommunityId, + _channel_id: &str, + text: &str, + _author_pubkey: &str, + _reply_to: Option<&str>, + ) -> std::pin::Pin< + Box< + dyn std::future::Future> + + Send + + '_, + >, + > { + self.messages + .lock() + .expect("recording action sink lock") + .push(text.to_string()); + Box::pin(async { Ok("recorded-event".to_string()) }) + } + } + async fn persistence_test_context() -> (buzz_db::Db, TenantContext) { 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"); + // The harness prepares the schema (CI uses pgschema, not SQLx history). let db = buzz_db::Db::from_pool(pool); - db.migrate() - .await - .expect("migrate workflow persistence test database"); let host = format!("workflow-cas-{}.example", Uuid::new_v4().simple()); let community = db .ensure_configured_community(&host) @@ -1391,6 +1591,197 @@ mod tests { (db, TenantContext::resolved(community, host)) } + async fn manual_trigger_test_context() -> (Arc, TenantContext, Keys, 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"); + // The harness prepares the schema (CI uses pgschema, not SQLx history). + let setup_db = buzz_db::Db::from_pool(setup_pool.clone()); + + 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: approval\n action: request_approval\n from: '@owner'\n message: approve\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, + agent, + workflow_id, + definition, + ) + } + + fn workflow_trigger_event_for_revision( + keys: &Keys, + workflow_id: Uuid, + revision: &str, + ) -> Event { + workflow_trigger_event_at( + keys, + workflow_id, + revision, + Timestamp::now().as_secs(), + &Uuid::new_v4().to_string(), + ) + } + + fn workflow_trigger_event_at( + keys: &Keys, + workflow_id: Uuid, + revision: &str, + created_at: u64, + request_id: &str, + ) -> Event { + EventBuilder::new(Kind::Custom(KIND_WORKFLOW_TRIGGER as u16), "") + .tags(vec![ + Tag::parse(["d", workflow_id.to_string().as_str()]).expect("d tag"), + Tag::parse(["e", revision]).expect("revision tag"), + Tag::parse(["request-id", request_id]).expect("request identity"), + ]) + .custom_created_at(Timestamp::from(created_at)) + .sign_with_keys(keys) + .expect("sign workflow trigger") + } + + fn workflow_trigger_event(keys: &Keys, workflow_id: Uuid, revision: &Event) -> Event { + workflow_trigger_event_for_revision(keys, workflow_id, &revision.id.to_hex()) + } + + 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, @@ -1426,6 +1817,254 @@ mod tests { } } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn human_owner_manual_trigger_completes_with_one_connection() { + let (state, tenant, human, agent, 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"); + let run_id = Uuid::parse_str( + serde_json::from_str::( + result + .message + .strip_prefix("response:") + .expect("workflow trigger response prefix"), + ) + .expect("workflow trigger response JSON")["run_id"] + .as_str() + .expect("workflow trigger run id"), + ) + .expect("workflow trigger run UUID"); + let (_, loaded_definition) = state + .workflow_engine + .load_run_definition(tenant.community(), run_id) + .await + .expect("manual execution must load its exact signed revision"); + assert_eq!(loaded_definition.name, "manual-trigger-pool"); + + let agent_trigger = workflow_trigger_event(&agent, workflow_id, &revision); + let agent_result = + handle_workflow_trigger(&tenant, &state, &agent_trigger, &http_auth(&agent)) + .await + .expect("workflow principal must be able to trigger its own workflow"); + assert!(agent_result.message.contains("\"run_id\"")); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn approval_resume_executes_the_run_bound_signed_revision() { + let (state, tenant, _human, agent, workflow_id, revision_a) = + manual_trigger_test_context().await; + let community_id = tenant.community(); + let db = state.db.clone(); + let trigger_context = serde_json::to_value(TriggerContext { + channel_id: exact_tag_value(&revision_a, "h") + .unwrap_or_default() + .to_string(), + ..TriggerContext::default() + }) + .expect("serialize trigger context"); + let run_id = db + .create_workflow_run( + community_id, + workflow_id, + Some(revision_a.id.as_bytes()), + None, + Some(&trigger_context), + ) + .await + .expect("create revision A run"); + db.update_workflow_run( + community_id, + run_id, + RunStatus::WaitingApproval, + 0, + &serde_json::json!([{ + "step_id": "approval", + "output": {"approved": true} + }]), + None, + ) + .await + .expect("suspend revision A run for approval"); + + let channel_id = Uuid::parse_str(exact_tag_value(&revision_a, "h").expect("channel tag")) + .expect("channel UUID"); + let revision_b = EventBuilder::new( + Kind::Custom(KIND_WORKFLOW_DEF as u16), + concat!( + "name: revision-b\n", + "trigger:\n on: message_posted\n", + "steps:\n", + " - id: approval\n action: request_approval\n from: '@owner'\n message: approve\n", + " - id: after\n action: send_message\n text: revision B\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 revision B"); + let (_, definition_b_json) = buzz_workflow::WorkflowEngine::parse_yaml(&revision_b.content) + .expect("parse revision B"); + let definition_b_hash = compute_definition_hash(&definition_b_json); + let mut tx = db + .begin_transaction() + .await + .expect("begin revision B update"); + buzz_db::event::insert_event_in_transaction( + &mut tx, + community_id, + &revision_b, + Some(channel_id), + ) + .await + .expect("persist revision B"); + db.upsert_workflow( + &mut tx, + community_id, + workflow_id, + Some(channel_id), + &agent.public_key().to_bytes(), + "revision-b", + &definition_b_json, + &definition_b_hash, + revision_b.id.as_bytes(), + ) + .await + .expect("materialize revision B"); + tx.commit().await.expect("commit revision B update"); + + let sink = Arc::new(RecordingActionSink::default()); + state.workflow_engine.set_action_sink(sink.clone()); + resume_workflow_after_approval( + Arc::clone(&state.workflow_engine), + db.clone(), + community_id, + run_id, + workflow_id, + 1, + ) + .await; + + let resumed = db + .get_workflow_run(community_id, run_id) + .await + .expect("load resumed run"); + assert_eq!(resumed.status, RunStatus::Completed); + assert_eq!( + sink.messages + .lock() + .expect("recorded messages lock") + .as_slice(), + ["done"], + "approval resume must execute revision A, never current revision B" + ); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn approval_resume_fails_closed_without_a_signed_run_revision() { + let (state, tenant, _human, _agent, workflow_id, _revision) = + manual_trigger_test_context().await; + let community_id = tenant.community(); + let db = state.db.clone(); + let run_id = db + .create_workflow_run(community_id, workflow_id, None, None, None) + .await + .expect("create legacy revisionless run"); + db.update_workflow_run( + community_id, + run_id, + RunStatus::WaitingApproval, + 0, + &serde_json::json!([]), + None, + ) + .await + .expect("suspend revisionless run"); + + resume_workflow_after_approval( + Arc::clone(&state.workflow_engine), + db.clone(), + community_id, + run_id, + workflow_id, + 1, + ) + .await; + + let failed = db + .get_workflow_run(community_id, run_id) + .await + .expect("load failed run"); + assert_eq!(failed.status, RunStatus::Failed); + assert_eq!(failed.error_code.as_deref(), Some("invalid_definition")); + assert!(failed + .error_message + .as_deref() + .is_some_and(|message| message.contains("no owner-signed definition revision"))); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn manual_trigger_rejects_non_owner_and_stale_or_missing_revision() { + let (state, tenant, human, _agent, workflow_id, revision) = + manual_trigger_test_context().await; + let stranger = Keys::generate(); + let unauthorized = workflow_trigger_event(&stranger, workflow_id, &revision); + let unauthorized_error = + match handle_workflow_trigger(&tenant, &state, &unauthorized, &http_auth(&stranger)) + .await + { + Err(error) => error, + Ok(_) => panic!("channel membership must not grant manual trigger authority"), + }; + assert!(matches!( + unauthorized_error, + IngestError::Rejected(ref message) + if message == "forbidden: not authorized to trigger this workflow" + )); + + let stale = workflow_trigger_event_for_revision(&human, workflow_id, &"42".repeat(32)); + let stale_error = + match handle_workflow_trigger(&tenant, &state, &stale, &http_auth(&human)).await { + Err(error) => error, + Ok(_) => panic!("a stale signed revision must be rejected"), + }; + assert!(matches!( + stale_error, + IngestError::Rejected(ref message) + if message == "conflict: workflow revision does not match current definition" + )); + + let missing = EventBuilder::new(Kind::Custom(KIND_WORKFLOW_TRIGGER as u16), "") + .tag(Tag::parse(["d", workflow_id.to_string().as_str()]).expect("d tag")) + .sign_with_keys(&human) + .expect("sign revision-less trigger"); + let missing_error = + match handle_workflow_trigger(&tenant, &state, &missing, &http_auth(&human)).await { + Err(error) => error, + Ok(_) => panic!("a revision-less trigger must fail closed"), + }; + assert!(matches!( + missing_error, + IngestError::Rejected(ref message) + if message == "invalid: expected exactly one workflow revision e tag" + )); + } + + mod trigger_delivery; + #[test] fn workflow_revision_parser_accepts_create_and_valid_update() { let revision = [0x42; 32]; @@ -1474,6 +2113,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( @@ -1494,15 +2137,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/handlers/command_executor/tests/trigger_delivery.rs b/crates/buzz-relay/src/handlers/command_executor/tests/trigger_delivery.rs new file mode 100644 index 00000000000..7617cf77bfb --- /dev/null +++ b/crates/buzz-relay/src/handlers/command_executor/tests/trigger_delivery.rs @@ -0,0 +1,205 @@ +use super::*; +use std::collections::{HashMap, HashSet}; + +fn response_run_id(result: &IngestResult) -> Uuid { + assert!(result.accepted); + let response: serde_json::Value = serde_json::from_str( + result + .message + .strip_prefix("response:") + .expect("new run response"), + ) + .expect("response JSON"); + Uuid::parse_str(response["run_id"].as_str().expect("run ID")).expect("run UUID") +} + +async fn assert_persisted_trigger_runs( + state: &AppState, + tenant: &TenantContext, + workflow_id: Uuid, + revision: &Event, + triggers: &[Event], + run_ids: &HashSet, +) { + let runs = state + .db + .list_workflow_runs(tenant.community(), workflow_id, 100) + .await + .expect("persisted runs"); + assert_eq!( + runs.len(), + triggers.len(), + "exactly one run per signed operation" + ); + assert_eq!( + run_ids.len(), + triggers.len(), + "distinct acknowledged run IDs" + ); + assert_eq!( + runs.iter().map(|run| run.id).collect::>(), + *run_ids + ); + let trigger_ids = triggers + .iter() + .map(|trigger| trigger.id.as_bytes().to_vec()) + .collect::>(); + assert_eq!( + trigger_ids.len(), + triggers.len(), + "fixture requests must be distinct" + ); + assert_eq!( + runs.iter() + .map(|run| run.trigger_event_id.clone().expect("trigger association")) + .collect::>(), + trigger_ids + ); + for run in runs { + assert_eq!( + run.definition_event_id.as_deref(), + Some(revision.id.as_bytes().as_slice()) + ); + } + for trigger in triggers { + assert_eq!( + state + .db + .get_workflow_run_id_by_trigger( + tenant.community(), + Uuid::new_v4(), + trigger.id.as_bytes(), + ) + .await + .expect("workflow-scoped lookup"), + None + ); + let stored = state + .db + .get_event_by_id(tenant.community(), trigger.id.as_bytes()) + .await + .expect("trigger remains stored"); + assert!( + stored.is_some(), + "later triggers must not replace earlier operations" + ); + } +} + +async fn deliver_and_replay(same_second: bool) { + let (state, tenant, human, _agent, workflow_id, revision) = manual_trigger_test_context().await; + let now = Timestamp::now().as_secs(); + let mut triggers = vec![ + workflow_trigger_event_at(&human, workflow_id, &revision.id.to_hex(), now, "first"), + workflow_trigger_event_at( + &human, + workflow_id, + &revision.id.to_hex(), + if same_second { now } else { now - 1 }, + "second", + ), + ]; + if same_second { + // A lower ID wins a NIP-33 timestamp tie. Deliver it first so accidental + // coordinate replacement would incorrectly suppress the second request. + triggers.sort_by_key(|event| event.id); + } + assert_ne!(triggers[0].id, triggers[1].id); + assert_eq!( + triggers[0].created_at == triggers[1].created_at, + same_second + ); + let mut run_ids = HashSet::new(); + let mut responses = HashMap::new(); + for trigger in &triggers { + let result = handle_workflow_trigger(&tenant, &state, trigger, &http_auth(&human)) + .await + .expect("distinct trigger accepted"); + assert!(run_ids.insert(response_run_id(&result))); + responses.insert(trigger.id, result.message); + } + // Replay both, including the earlier operation after another has committed. + for trigger in &triggers { + let replay = handle_workflow_trigger(&tenant, &state, trigger, &http_auth(&human)) + .await + .expect("exact replay accepted"); + assert!(replay.accepted); + assert_eq!( + replay.message, responses[&trigger.id], + "retry recovers the original result" + ); + } + assert_persisted_trigger_runs(&state, &tenant, workflow_id, &revision, &triggers, &run_ids) + .await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn manual_triggers_delivered_newer_then_older_create_distinct_runs() { + deliver_and_replay(false).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn manual_triggers_in_one_second_create_distinct_runs() { + deliver_and_replay(true).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn concurrent_human_owner_manual_triggers_do_not_starve_one_connection_pool() { + let (state, tenant, human, _agent, workflow_id, revision) = manual_trigger_test_context().await; + let now = Timestamp::now().as_secs(); + let triggers = (0..8) + .map(|i| { + workflow_trigger_event_at( + &human, + workflow_id, + &revision.id.to_hex(), + now, + &i.to_string(), + ) + }) + .collect::>(); + let mut deliveries = triggers.clone(); + // Concurrent exact retries must not create additional runs either. + deliveries.extend_from_slice(&triggers[..2]); + let results = tokio::time::timeout(std::time::Duration::from_secs(8), async { + let mut tasks = tokio::task::JoinSet::new(); + for trigger in deliveries { + let state = Arc::clone(&state); + let tenant = tenant.clone(); + let auth = http_auth(&human); + tasks.spawn( + async move { handle_workflow_trigger(&tenant, &state, &trigger, &auth).await }, + ); + } + let mut results = Vec::new(); + while let Some(result) = tasks.join_next().await { + results.push( + result + .expect("trigger task must not panic") + .expect("trigger succeeds"), + ); + } + results + }) + .await + .expect("concurrent triggers must drain rather than pool-starve"); + assert_eq!(results.len(), 10); + assert!(results.iter().all(|result| result.accepted)); + let mut acknowledged = HashMap::new(); + for result in &results { + let run_id = response_run_id(result); + if let Some(previous) = acknowledged.insert(result.event_id.clone(), run_id) { + assert_eq!( + previous, run_id, + "concurrent retry returns the original run" + ); + } + } + assert_eq!(acknowledged.len(), 8); + let run_ids = acknowledged.into_values().collect::>(); + assert_persisted_trigger_runs(&state, &tenant, workflow_id, &revision, &triggers, &run_ids) + .await; +} diff --git a/crates/buzz-relay/src/router.rs b/crates/buzz-relay/src/router.rs index dd0fde6fdcd..cce756581c2 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..b58cff1c74d 100644 --- a/crates/buzz-sdk/src/builders.rs +++ b/crates/buzz-sdk/src/builders.rs @@ -1640,10 +1640,42 @@ 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()])?]; - Ok(EventBuilder::new(Kind::Custom(KIND_WORKFLOW_TRIGGER as u16), "").tags(tags)) +/// Build a workflow trigger event (kind 46020) bound to an exact signed revision. +/// +/// Each call creates a new operation, even within the same second. Retain and +/// resubmit the signed event when retrying that operation; do not rebuild it. +pub fn build_workflow_trigger( + workflow_id: Uuid, + definition_event_id: &str, +) -> Result { + workflow_trigger_builder(workflow_id, definition_event_id, "") +} + +/// Build a distinct workflow trigger operation with JSON object inputs. +/// Retry by resubmitting the signed event, as with [`build_workflow_trigger`]. +pub fn build_workflow_trigger_with_inputs( + workflow_id: Uuid, + definition_event_id: &str, + inputs: &serde_json::Map, +) -> Result { + let content = serde_json::Value::Object(inputs.clone()).to_string(); + check_content(&content, 64 * 1024)?; + workflow_trigger_builder(workflow_id, definition_event_id, &content) +} + +fn workflow_trigger_builder( + workflow_id: Uuid, + definition_event_id: &str, + content: &str, +) -> Result { + let revision = check_hex_exact(definition_event_id, 64, "definition_event_id")?; + let tags = vec![ + tag(&["d", &workflow_id.to_string()])?, + tag(&["e", &revision])?, + // Event IDs hash the unsigned fields, not the randomized signature. + tag(&["request-id", &Uuid::new_v4().to_string()])?, + ]; + Ok(EventBuilder::new(Kind::Custom(KIND_WORKFLOW_TRIGGER as u16), content).tags(tags)) } /// Build a workflow approval event — kind 46030 (grant) or 46031 (deny). @@ -4010,9 +4042,39 @@ 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] + fn workflow_trigger_invocations_in_one_second_have_distinct_ids() { + let keys = nostr::Keys::generate(); + let workflow_id = uuid(); + let revision = "ab".repeat(32); + let timestamp = nostr::Timestamp::from(1_700_000_000); + let first = build_workflow_trigger(workflow_id, &revision).unwrap(); + let replay = first + .clone() + .custom_created_at(timestamp) + .sign_with_keys(&keys) + .unwrap(); + let first = first + .custom_created_at(timestamp) + .sign_with_keys(&keys) + .unwrap(); + let second = build_workflow_trigger(workflow_id, &revision) + .unwrap() + .custom_created_at(timestamp) + .sign_with_keys(&keys) + .unwrap(); + assert_eq!(first.id, replay.id, "retry the same built operation"); + assert_ne!( + first.id, second.id, + "new invocations must not become replays" + ); } #[test] diff --git a/crates/buzz-test-client/tests/conformance_multitenant.rs b/crates/buzz-test-client/tests/conformance_multitenant.rs index 4c8c8904ac5..70740c9889a 100644 --- a/crates/buzz-test-client/tests/conformance_multitenant.rs +++ b/crates/buzz-test-client/tests/conformance_multitenant.rs @@ -1778,7 +1778,31 @@ mod workflows { .to_string() } - /// Fire a workflow by id on `http_base`'s community (kind:46020, `d`=id). + /// Fetch the current owner-signed workflow revision from `http_base`'s + /// community. Manual triggers must name this exact event id. + async fn workflow_revision(http_base: &str, keys: &Keys, workflow_id: &str) -> String { + let resp = reqwest::Client::new() + .get(format!("{http_base}/workflows/{workflow_id}/revision")) + .header("X-Pubkey", keys.public_key().to_hex()) + .send() + .await + .unwrap_or_else(|e| panic!("GET workflow revision from {http_base} failed: {e}")); + let status = resp.status(); + let body = resp.text().await.expect("read workflow revision body"); + assert!( + status.is_success(), + "workflow revision lookup against {http_base} failed with {status}: {body}" + ); + let event: serde_json::Value = serde_json::from_str(&body) + .unwrap_or_else(|e| panic!("parse workflow revision JSON: {e} ({body:?})")); + event["id"] + .as_str() + .unwrap_or_else(|| panic!("workflow revision response missing event id: {event}")) + .to_string() + } + + /// Fire an exact workflow revision on `http_base`'s community + /// (kind:46020, `d`=workflow id, `e`=signed revision id). /// Returns a normalized `{accepted, message}` body so the caller can assert /// on the *wire-observable* accept/reject and message. The HTTP bridge maps /// `IngestError::Rejected` to HTTP 400 + `{error}` while the WS door maps the @@ -1788,9 +1812,13 @@ mod workflows { http_base: &str, keys: &Keys, workflow_id: &str, + revision: &str, ) -> serde_json::Value { let event = EventBuilder::new(Kind::Custom(KIND_WORKFLOW_TRIGGER), "") - .tags(vec![Tag::parse(["d", workflow_id]).unwrap()]) + .tags(vec![ + Tag::parse(["d", workflow_id]).unwrap(), + Tag::parse(["e", revision]).unwrap(), + ]) .sign_with_keys(keys) .unwrap(); @@ -1837,9 +1865,11 @@ mod workflows { /// This deliberately removes "not a member of U in B" as an alternate /// cause of the B rejection — K *is* a member of U in B. /// 2. Define a workflow in `U` under **A** (kind:30620). The server - /// generates `W` and returns it. `W` is an A-community row. - /// 3. Fire `W` under host **B** (kind:46020, `d`=W) as K. Must be - /// rejected — `accepted == false` and the generic `workflow not found` + /// generates `W` and returns it. Retrieve its current signed revision + /// `R` through A's authenticated revision endpoint. `W` and `R` are + /// A-community records. + /// 3. Fire `W`/`R` under host **B** (kind:46020, `d`=W, `e`=R) as K. Must + /// be rejected — `accepted == false` and the generic `workflow not found` /// message — because `get_workflow(B_community, W)` finds nothing: `W` /// exists only in A. K's membership of U-in-B is irrelevant; the /// lookup never reaches the membership check. @@ -1893,11 +1923,17 @@ mod workflows { uuid::Uuid::parse_str(&workflow_id).is_ok(), "server-generated workflow_id must be a UUID, got {workflow_id:?}" ); + let revision = workflow_revision(&http_a, &keys, &workflow_id).await; + assert_eq!( + revision.len(), + 64, + "signed workflow revision must be a 32-byte event id" + ); - // (3) Fire W under host B as K. Must fail closed: W is an A-community - // row, and get_workflow(B_community, W) finds nothing. K is a member of - // U in B, so a leak here is the community fence failing, not membership. - let b_resp = trigger_workflow(&http_b, &keys, &workflow_id).await; + // (3) Fire W/R under host B as K. Keep the trigger grammar otherwise + // valid with A's real signed revision so only tenant-scoped workflow + // resolution can explain the rejection. + let b_resp = trigger_workflow(&http_b, &keys, &workflow_id, &revision).await; assert_eq!( b_resp["accepted"].as_bool(), Some(false), @@ -1915,7 +1951,7 @@ mod workflows { // proves the B rejection is community confinement, not an // untriggerable workflow, and exercises the same-community happy path // through the fence under test. - let a_resp = trigger_workflow(&http_a, &keys, &workflow_id).await; + let a_resp = trigger_workflow(&http_a, &keys, &workflow_id, &revision).await; assert_eq!( a_resp["accepted"].as_bool(), Some(true), diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index 4fc4f9d5be8..816db94015c 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 diff --git a/desktop/src-tauri/src/commands/workflows.rs b/desktop/src-tauri/src/commands/workflows.rs index c4e5d38c8ba..563d7dd1bc0 100644 --- a/desktop/src-tauri/src/commands/workflows.rs +++ b/desktop/src-tauri/src/commands/workflows.rs @@ -1,6 +1,6 @@ use std::collections::HashSet; -use serde::Serialize; +use serde::{Deserialize, Serialize}; use serde_json::Value; use tauri::State; @@ -323,7 +323,20 @@ pub async fn trigger_workflow( workflow_id: String, state: State<'_, AppState>, ) -> Result { - let builder = events::build_workflow_trigger(&workflow_id)?; + #[derive(Deserialize)] + struct WorkflowRevisionWire { + id: String, + } + + // 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. The revision endpoint returns + // only this identifier so immutable ownership never grants definition-read + // access after channel membership is revoked. + let revision: WorkflowRevisionWire = + get_relay_json(&state, &format!("/workflows/{workflow_id}/revision")).await?; + let definition_event_id = revision.id; + 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/workflows.rs b/desktop/src-tauri/src/events/workflows.rs index 8615f73851f..8d29639c315 100644 --- a/desktop/src-tauri/src/events/workflows.rs +++ b/desktop/src-tauri/src/events/workflows.rs @@ -31,10 +31,15 @@ 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])?]; - Ok(EventBuilder::new(Kind::Custom(46020), "").tags(tags)) +/// 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 { + let workflow_id = + uuid::Uuid::parse_str(workflow_id).map_err(|_| "invalid workflow id".to_string())?; + buzz_sdk_pkg::build_workflow_trigger(workflow_id, definition_event_id) + .map_err(|error| error.to_string()) } /// Kind 46030 — grant an approval token (with optional note). @@ -48,3 +53,24 @@ pub fn build_approval_deny(token: &str, note: Option<&str>) -> Result