diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index 5901512504..d134a8aa93 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -702,7 +702,7 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 44); + assert_eq!(migrations.len(), 45); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); assert!(migrations[0] @@ -1368,6 +1368,35 @@ mod postgres_tests { assert!(include_str!("../../../../schema/schema.sql").contains("error_code TEXT")); } + #[test] + fn workflow_deletion_cutoff_matches_desired_schema_without_legacy_backfill() { + let migration = MIGRATOR + .iter() + .find(|m| m.version == 45) + .expect("migration 45"); + let schema = include_str!("../../../../schema/schema.sql"); + let table = |sql: &str| { + split_sql_statements(sql) + .into_iter() + .find(|statement| { + normalize_sql(statement).starts_with("create table workflow_deletions") + }) + .map(|statement| normalize_sql(&statement)) + .expect("deletion table") + }; + assert_eq!(table(migration.sql.as_ref()), table(schema)); + let statements = split_sql_statements(migration.sql.as_ref()); + assert_eq!( + statements.len(), + 2, + "forward proof must never be inferred from legacy data" + ); + assert_eq!( + normalize_sql(&statements[1]), + "select attach_community_write_fence('workflow_deletions')" + ); + } + #[test] fn push_match_trigger_is_narrowed_to_message_kinds_additively() { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); @@ -1822,6 +1851,11 @@ mod postgres_tests { let mut expected_fences = migration.fence_attachments.clone(); expected_fences.remove("product_feedback"); expected_fences.remove("rate_limit_violations"); + let workflow_lifecycle = MIGRATOR + .iter() + .find(|migration| migration.version == 45) + .expect("embedded workflow lifecycle migration"); + expected_fences.extend(surface(workflow_lifecycle.sql.as_ref()).fence_attachments); assert_eq!( expected_fences, schema.fence_attachments, "write-fence attachment targets differ after recovery policy" diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index 0e184e00d8..dd316b313d 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -81,12 +81,14 @@ pub const EXPECTED_SCOPED_TABLES: &[&str] = &[ "thread_metadata", "users", "workflow_approvals", + "workflow_deletions", "workflow_runs", "workflows", ]; /// Foreign-key-safe child-before-parent order for the PostgreSQL purge. pub const PURGE_SCOPED_TABLES: &[&str] = &[ + "workflow_deletions", "workflow_approvals", "scheduled_workflow_fires", "workflow_runs", @@ -4997,3 +4999,7 @@ mod postgres_tests { .expect("drop probe database"); } } + +#[cfg(test)] +#[path = "deletion/workflow_postgres_tests.rs"] +mod workflow_postgres_tests; diff --git a/crates/buzz-db/src/store/deletion/workflow_postgres_tests.rs b/crates/buzz-db/src/store/deletion/workflow_postgres_tests.rs new file mode 100644 index 0000000000..dbc6a18e67 --- /dev/null +++ b/crates/buzz-db/src/store/deletion/workflow_postgres_tests.rs @@ -0,0 +1,158 @@ +//! The forward workflow proof is tenant data, including during whole-community purge. +use super::*; +use buzz_core::channel::{ChannelType, ChannelVisibility}; + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn desired_workflow_deletion_proofs_are_fenced_and_purged() { + check_workflow_deletion_proofs("desired").await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn migration_schema_workflow_deletion_proofs_are_fenced_and_purged() { + check_workflow_deletion_proofs("migration").await; +} + +async fn check_workflow_deletion_proofs(mode: &str) { + assert_eq!(std::env::var("BUZZ_TEST_SCHEMA_MODE").as_deref(), Ok(mode)); + let pool = PgPool::connect(&crate::test_support::database_url()) + .await + .expect("isolated database"); + let db = Db::from_pool(pool.clone()); + if mode == "migration" { + db.migrate().await.expect("all migrations"); + } + let store = db.deletion_store(); + let mut tenants = Vec::new(); + for label in ["target", "control"] { + let host = format!("workflow-purge-{label}-{}.example", Uuid::new_v4()); + let community = db + .ensure_configured_community(&host) + .await + .expect("community") + .id; + let channel = db + .create_channel( + community, + "workflows", + ChannelType::Stream, + ChannelVisibility::Private, + None, + &[7; 32], + None, + ) + .await + .expect("channel") + .id; + sqlx::query( + "INSERT INTO workflow_deletions \ + (community_id, owner_pubkey, workflow_id, channel_id, deleted_through, event_id) \ + VALUES ($1, $2, $3, $4, now(), $5)", + ) + .bind(community.as_uuid()) + .bind(vec![7_u8; 32]) + .bind(Uuid::new_v4()) + .bind(channel) + .bind(vec![8_u8; 32]) + .execute(&pool) + .await + .expect("proof"); + tenants.push((community, host)); + } + let (target, host) = &tenants[0]; + let schema = store + .inventory_schema(*target) + .await + .expect("catalog includes proof"); + assert_eq!(schema.row_counts["workflow_deletions"], 1); + assert!(schema + .fenced_tables + .contains(&"workflow_deletions".to_string())); + let storage = StorageManifest { + version: 5, + prefixes: ["_meta", "_uploads", "repos"] + .into_iter() + .map(|prefix| { + let (keys_digest, object_count) = KeyStreamDigest::new().finish(); + PrefixManifest { + prefix: format!("{prefix}/{target}/"), + keys_digest, + object_count, + total_bytes: 0, + } + }) + .collect(), + }; + let request = store + .submit(host, "test-operator", None) + .await + .expect("request"); + store + .freeze_inventory( + request.id, + &FrozenInventory { + schema, + storage: storage.clone(), + }, + ) + .await + .expect("freeze"); + store + .approve(request.id, "test-approver", None) + .await + .expect("approve"); + let claim = store + .claim_specific(request.id, "test-executor", DEFAULT_LEASE_DURATION) + .await + .expect("claim") + .expect("won claim"); + store.begin_quiescing(&claim.lease).await.expect("quiesce"); + let generation = store.fence(&claim.lease).await.expect("fence"); + let token = LeaseToken { + fence_generation: Some(generation), + ..claim.lease + }; + let error = sqlx::query( + "UPDATE workflow_deletions SET deleted_through = now() WHERE community_id = $1", + ) + .bind(target.as_uuid()) + .execute(&pool) + .await + .expect_err("proof updates must be fenced"); + assert!( + error.to_string().contains("community write fenced"), + "{error}" + ); + store + .freeze_destructive_storage_manifest(&token, &storage) + .await + .expect("storage"); + store.mark_drained(&token).await.expect("drain"); + store + .mark_bindings_removed(&token, serde_json::json!({"keys": 0})) + .await + .expect("bindings"); + let counts = store + .purge_postgres(&token) + .await + .expect("purge child before channel"); + assert_eq!(counts["workflow_deletions"], 1); + store + .mark_cache_purged(&token, serde_json::json!({"keys": 0})) + .await + .expect("cache"); + store + .verify_postgres_logically_deleted(&token) + .await + .expect("logical absence"); + let remaining: Vec = sqlx::query_scalar("SELECT community_id FROM workflow_deletions") + .fetch_all(&pool) + .await + .expect("remaining proofs"); + assert_eq!( + remaining, + vec![*tenants[1].0.as_uuid()], + "control tenant stays intact" + ); +} diff --git a/crates/buzz-db/src/store/workflow.rs b/crates/buzz-db/src/store/workflow.rs index 3ceed9ea32..c03f3eb60e 100644 --- a/crates/buzz-db/src/store/workflow.rs +++ b/crates/buzz-db/src/store/workflow.rs @@ -7,12 +7,15 @@ //! - Approval tokens are stored as SHA-256 hashes (never plaintext). //! - All list queries have a bounded LIMIT to prevent unbounded scans. +/// Transactional workflow-coordinate lifecycle operations. +pub mod lifecycle; + use std::fmt; use std::str::FromStr; use chrono::{DateTime, Utc}; use sha2::{Digest, Sha256}; -use sqlx::{PgPool, Row}; +use sqlx::{Executor, PgPool, Postgres, Row, Transaction}; use uuid::Uuid; use buzz_core::CommunityId; @@ -324,6 +327,55 @@ pub async fn upsert_workflow( name: &str, definition_json: &str, definition_hash: &[u8], +) -> Result<()> { + upsert_workflow_on( + pool, + community_id, + id, + channel_id, + owner_pubkey, + name, + definition_json, + definition_hash, + ) + .await +} + +/// Upsert the runtime projection in the same transaction as its signed definition. +#[allow(clippy::too_many_arguments)] +pub async fn upsert_workflow_in_transaction( + tx: &mut Transaction<'_, Postgres>, + community_id: CommunityId, + id: Uuid, + channel_id: Option, + owner_pubkey: &[u8], + name: &str, + definition_json: &str, + definition_hash: &[u8], +) -> Result<()> { + upsert_workflow_on( + &mut **tx, + community_id, + id, + channel_id, + owner_pubkey, + name, + definition_json, + definition_hash, + ) + .await +} + +#[allow(clippy::too_many_arguments)] +async fn upsert_workflow_on<'e>( + executor: impl Executor<'e, Database = Postgres>, + community_id: CommunityId, + id: Uuid, + channel_id: Option, + owner_pubkey: &[u8], + name: &str, + definition_json: &str, + definition_hash: &[u8], ) -> Result<()> { let row = sqlx::query( r#" @@ -347,7 +399,7 @@ pub async fn upsert_workflow( .bind(channel_id) .bind(definition_json) .bind(definition_hash) - .fetch_optional(pool) + .fetch_optional(executor) .await?; if row.is_none() { @@ -369,6 +421,23 @@ pub async fn get_workflow( pool: &PgPool, community_id: CommunityId, id: Uuid, +) -> Result { + get_workflow_on(pool, community_id, id).await +} + +/// Read workflow state after taking its coordinate lock, without leaving the event transaction. +pub async fn get_workflow_in_transaction( + tx: &mut Transaction<'_, Postgres>, + community_id: CommunityId, + id: Uuid, +) -> Result { + get_workflow_on(&mut **tx, community_id, id).await +} + +async fn get_workflow_on<'e>( + executor: impl Executor<'e, Database = Postgres>, + community_id: CommunityId, + id: Uuid, ) -> Result { let row = sqlx::query( r#" @@ -380,7 +449,7 @@ pub async fn get_workflow( ) .bind(community_id.as_uuid()) .bind(id) - .fetch_optional(pool) + .fetch_optional(executor) .await? .ok_or_else(|| DbError::NotFound(format!("workflow {id}")))?; diff --git a/crates/buzz-db/src/store/workflow/lifecycle.rs b/crates/buzz-db/src/store/workflow/lifecycle.rs new file mode 100644 index 0000000000..1c466eb580 --- /dev/null +++ b/crates/buzz-db/src/store/workflow/lifecycle.rs @@ -0,0 +1,166 @@ +//! Forward-only workflow deletion proofs, serialized with NIP-33 replacement. +//! One fixed-size row per deleted coordinate; no workflow content or secrets. + +use buzz_core::{kind::KIND_WORKFLOW_DEF, CommunityId}; +use chrono::{DateTime, Utc}; +use sqlx::{Postgres, Transaction}; +use uuid::Uuid; + +use crate::{observability, replaceable::event_replacement_lock_key, Result}; + +/// Signed live head and its server-resolved channel. +#[derive(Debug, sqlx::FromRow)] +pub struct Head { + /// Signed timestamp, not server receipt time. + pub created_at: DateTime, + /// Exact signed definition ID. + pub id: Vec, + /// Stored channel scope; absent legacy scope is not authoritative. + pub channel_id: Option, +} + +/// Latest committed forward deletion for this coordinate. +#[derive(Debug, sqlx::FromRow)] +pub struct Deletion { + /// Actual channel resolved by the successful transaction. + pub channel_id: Uuid, + /// Older/equal unseen saves must not revive this coordinate. + pub deleted_through: DateTime, + /// Exact deletion whose atomic transaction committed. + pub event_id: Vec, +} + +/// Take the same lock as canonical kind-30620 replacement. The caller owns the transaction. +pub async fn lock_coordinate( + tx: &mut Transaction<'_, Postgres>, + community: CommunityId, + owner: &[u8], + id: Uuid, +) -> Result<()> { + let d_tag = id.to_string(); + let key = event_replacement_lock_key( + community, + KIND_WORKFLOW_DEF as i32, + owner, + Some(d_tag.as_bytes()), + ); + observability::observe_advisory_lock( + observability::LockType::Replacement, + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(key) + .execute(&mut **tx), + ) + .await?; + Ok(()) +} + +/// Read the live definition under the coordinate lock. +pub async fn head( + tx: &mut Transaction<'_, Postgres>, + community: CommunityId, + owner: &[u8], + id: Uuid, +) -> Result> { + Ok(sqlx::query_as( + "SELECT created_at, id, channel_id FROM events WHERE community_id=$1 AND kind=$2 \ + AND pubkey=$3 AND d_tag=$4 AND deleted_at IS NULL ORDER BY created_at DESC, id ASC LIMIT 1", + ).bind(community.as_uuid()).bind(KIND_WORKFLOW_DEF as i32).bind(owner).bind(id.to_string()) + .fetch_optional(&mut **tx).await?) +} + +/// Read the trusted forward cutoff. Accepted legacy deletion events are never consulted. +pub async fn deletion( + tx: &mut Transaction<'_, Postgres>, + community: CommunityId, + owner: &[u8], + id: Uuid, +) -> Result> { + Ok(sqlx::query_as( + "SELECT channel_id, deleted_through, event_id FROM workflow_deletions \ + WHERE community_id=$1 AND owner_pubkey=$2 AND workflow_id=$3", + ) + .bind(community.as_uuid()) + .bind(owner) + .bind(id) + .fetch_optional(&mut **tx) + .await?) +} + +/// Recognize exact stored events, including soft-deleted versions, without reapplying them. +pub async fn event_seen( + tx: &mut Transaction<'_, Postgres>, + community: CommunityId, + event: &nostr::Event, +) -> Result { + Ok(sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM events WHERE community_id=$1 AND id=$2 AND created_at=$3)", + ) + .bind(community.as_uuid()) + .bind(event.id.as_bytes().as_slice()) + .bind( + DateTime::from_timestamp(event.created_at.as_secs() as i64, 0).ok_or( + crate::DbError::InvalidTimestamp(event.created_at.as_secs() as i64), + )?, + ) + .fetch_one(&mut **tx) + .await?) +} + +/// Mutate both projections and advance the proof in the caller's transaction. +/// The caller must hold the coordinate lock, validate owner/channel and timestamp, +/// and insert the deletion event in this same transaction before committing. +pub async fn delete_in_transaction( + tx: &mut Transaction<'_, Postgres>, + community: CommunityId, + owner: &[u8], + id: Uuid, + channel: Uuid, + event: &nostr::Event, +) -> Result<()> { + let through = DateTime::from_timestamp(event.created_at.as_secs() as i64, 0).ok_or( + crate::DbError::InvalidTimestamp(event.created_at.as_secs() as i64), + )?; + sqlx::query("DELETE FROM workflows WHERE community_id=$1 AND id=$2 AND owner_pubkey=$3 AND channel_id=$4") + .bind(community.as_uuid()).bind(id).bind(owner).bind(channel).execute(&mut **tx).await?; + sqlx::query( + "UPDATE events SET deleted_at=NOW() WHERE community_id=$1 AND kind=$2 AND pubkey=$3 \ + AND d_tag=$4 AND channel_id=$5 AND created_at <= $6 AND deleted_at IS NULL", + ) + .bind(community.as_uuid()) + .bind(KIND_WORKFLOW_DEF as i32) + .bind(owner) + .bind(id.to_string()) + .bind(channel) + .bind(through) + .execute(&mut **tx) + .await?; + sqlx::query("INSERT INTO workflow_deletions (community_id, owner_pubkey, workflow_id, channel_id, deleted_through, event_id) \ + VALUES ($1,$2,$3,$4,$5,$6) ON CONFLICT (community_id, owner_pubkey, workflow_id) \ + DO UPDATE SET deleted_through=EXCLUDED.deleted_through, event_id=EXCLUDED.event_id \ + WHERE workflow_deletions.deleted_through < EXCLUDED.deleted_through") + .bind(community.as_uuid()).bind(owner).bind(id).bind(channel).bind(through) + .bind(event.id.as_bytes().as_slice()).execute(&mut **tx).await?; + Ok(()) +} + +/// Lock the active member and channel rows until the lifecycle transaction ends. +/// Returns the actual role; open-channel visibility is not workflow-author authority. +pub async fn channel_role( + tx: &mut Transaction<'_, Postgres>, + community: CommunityId, + channel: Uuid, + owner: &[u8], +) -> Result> { + Ok(sqlx::query_scalar( + "SELECT cm.role::text FROM channel_members cm JOIN channels c \ + ON c.community_id=cm.community_id AND c.id=cm.channel_id \ + WHERE cm.community_id=$1 AND cm.channel_id=$2 AND cm.pubkey=$3 \ + AND cm.removed_at IS NULL AND c.deleted_at IS NULL AND c.archived_at IS NULL \ + FOR SHARE OF c, cm", + ) + .bind(community.as_uuid()) + .bind(channel) + .bind(owner) + .fetch_optional(&mut **tx) + .await?) +} diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 074f6b391d..5323e74010 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -92,12 +92,13 @@ 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 +/// NOTE: Other domain mutations (open_dm, 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). +/// operations (open_dm, hide_dm, update_approval). Workflow definitions instead +/// compose their runtime projection in this transaction. #[datastore_span(name = "persist_command_event", system = "postgresql")] async fn persist_command_event( db: &buzz_db::Db, @@ -130,6 +131,53 @@ async fn persist_command_event( } let kind = event.kind.as_u16() as i32; + if kind == KIND_WORKFLOW_DEF as i32 { + use buzz_db::workflow::lifecycle; + let id = Uuid::parse_str(d_tag) + .map_err(|_| IngestError::Rejected("invalid: bad workflow_id format".into()))?; + // UUID aliases otherwise share a runtime row but not a replacement lock. + if d_tag != id.to_string() + || event + .tags + .iter() + .filter(|tag| tag.kind().to_string() == "d") + .count() + != 1 + { + return Err(IngestError::Rejected( + "invalid: workflow d tag must be one canonical UUID".into(), + )); + } + let owner = event.pubkey.to_bytes(); + lifecycle::lock_coordinate(&mut tx, tenant.community(), &owner, id) + .await + .map_err(|e| { + IngestError::Internal(format!("error: workflow coordinate lock: {e}")) + })?; + if lifecycle::event_seen(&mut tx, tenant.community(), event) + .await + .map_err(|e| IngestError::Internal(format!("error: workflow replay lookup: {e}")))? + { + return Ok(PersistResult::Duplicate); + } + if let Some(deletion) = lifecycle::deletion(&mut tx, tenant.community(), &owner, id) + .await + .map_err(|e| { + IngestError::Internal(format!("error: workflow deletion lookup: {e}")) + })? + { + if event.created_at.as_secs() as i64 <= deletion.deleted_through.timestamp() { + return Err(IngestError::Rejected( + "conflict: workflow save predates a committed deletion".into(), + )); + } + if channel_id != Some(deletion.channel_id) { + return Err(IngestError::Rejected( + "forbidden: workflow belongs to a different channel".into(), + )); + } + } + } let (expected_revision, revision_error) = match parse_expected_workflow_revision( kind, extract_tag(event, "expected-revision").as_deref(), @@ -657,44 +705,65 @@ async fn handle_workflow_def( let workflow_id = Uuid::parse_str(&workflow_id_str) .map_err(|_| IngestError::Rejected("invalid: bad workflow_id format".into()))?; - // 2. Validate caller has channel access (minimum: is a member) - let is_member = state - .is_member_cached(tenant.community(), channel_id, &self_bytes) - .await - .map_err(|e| IngestError::Internal(format!("error: membership check: {e}")))?; - if !is_member { - return Err(IngestError::Rejected( - "forbidden: not a member of this channel".into(), - )); - } - // 3. Parse YAML from event.content let (def, definition_json_str) = buzz_workflow::WorkflowEngine::parse_yaml(&event.content) .map_err(|e| IngestError::Rejected(format!("invalid: workflow YAML parse error: {e}")))?; let workflow_name = extract_tag(event, "name").unwrap_or_else(|| def.name.clone()); - // SEC-006: definitions with exfiltration-capable actions (call_webhook) - // require elevated channel authority to save — plain membership is not - // enough, because the workflow will forward channel content outward with - // the owner's standing authority. Fail-closed on lookup errors. - if def.requires_elevated_authority() { - let role = state - .db - .get_member_role(tenant.community(), channel_id, &self_bytes) - .await - .map_err(|e| IngestError::Internal(format!("error: role check: {e}")))?; - if !matches!(role.as_deref(), Some("owner") | Some("admin")) { - return Err(IngestError::Rejected( - "forbidden: workflows with call_webhook actions require the owner or admin role" - .into(), - )); - } - } - let mut definition_json: serde_json::Value = serde_json::from_str(&definition_json_str) .map_err(|e| IngestError::Internal(format!("error: json parse of definition: {e}")))?; - let existing_workflow = match state.db.get_workflow(tenant.community(), workflow_id).await { + let channel_tags: Vec<_> = event + .tags + .iter() + .filter(|tag| tag.kind().to_string() == "h") + .collect(); + if channel_tags.len() != 1 + || channel_tags[0].as_slice().len() != 2 + || channel_id_str != channel_id.to_string() + { + return Err(IngestError::Rejected( + "invalid: workflow h tag must be one canonical channel UUID".into(), + )); + } + super::ingest::check_token_channel_access(auth, channel_id).map_err(IngestError::AuthFailed)?; + + // Persist the command event — returns open transaction + let mut tx = match persist_command_event(&state.db, tenant, event, None).await? { + PersistResult::Duplicate => { + return Ok(IngestResult { + event_id: event.id.to_hex(), + accepted: true, + message: "duplicate: already processed".into(), + }); + } + PersistResult::Inserted(tx) => tx, + }; + + let role = buzz_db::workflow::lifecycle::channel_role( + &mut tx, + tenant.community(), + channel_id, + &self_bytes, + ) + .await + .map_err(|e| IngestError::Internal(format!("error: workflow channel authority: {e}")))? + .ok_or_else(|| { + IngestError::Rejected("forbidden: workflow requires active channel membership".into()) + })?; + if def.requires_elevated_authority() && !matches!(role.as_str(), "owner" | "admin") { + return Err(IngestError::Rejected( + "forbidden: workflows with call_webhook actions require the owner or admin role".into(), + )); + } + + let existing_workflow = match buzz_db::workflow::get_workflow_in_transaction( + &mut tx, + tenant.community(), + workflow_id, + ) + .await + { Ok(workflow) => { if workflow.owner_pubkey != self_bytes || workflow.channel_id != Some(channel_id) { return Err(IngestError::Rejected( @@ -737,66 +806,36 @@ 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? { - PersistResult::Duplicate => { - return Ok(IngestResult { - event_id: event.id.to_hex(), - accepted: true, - message: "duplicate: already processed".into(), - }); - } - PersistResult::Inserted(tx) => tx, - }; - - // 4. Execute: upsert by the NIP-33 d-tag UUID. A retry updates the same - // row instead of creating another enabled workflow that would fan out on - // every matching event. The workflow's community is the request's - // server-bound tenant — never re-derived from the (client-supplied) channel - // id. `community_of_channel(channel_id)` is ambiguous when the same channel - // UUID exists in two communities and could mint the workflow under the wrong - // tenant; `tenant.community()` is the authoritative owner. We then verify the - // channel actually exists *inside that community* (scoped `get_channel`), - // which fails closed if the client named a channel that belongs to a - // different community — the same guarantee the `(community_id, channel_id)` - // composite FK enforces on insert, surfaced here as a clean rejection. + // The tenant is host-bound; channel_role verified and locked this channel in it. let community_id = tenant.community(); - state - .db - .get_channel_for_event_write(community_id, channel_id) - .await - .map_err(|_| IngestError::Rejected("invalid: workflow channel not found".into()))?; + buzz_db::workflow::upsert_workflow_in_transaction( + &mut tx, + community_id, + workflow_id, + Some(channel_id), + &self_bytes, + &workflow_name, + &definition_json_final, + &hash, + ) + .await + .map_err(|e| match e { + DbError::AccessDenied(_) => IngestError::Rejected( + "forbidden: workflow belongs to a different owner or channel".into(), + ), + other => IngestError::Internal(format!("error: db upsert_workflow: {other}")), + })?; - state - .db - .upsert_workflow( - community_id, - workflow_id, - Some(channel_id), - &self_bytes, - &workflow_name, - &definition_json_final, - &hash, - ) - .await - .map_err(|e| match e { - DbError::AccessDenied(_) => IngestError::Rejected( - "forbidden: workflow belongs to a different owner or channel".into(), - ), - 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. + // Both signed intent and runtime projection become visible together. tx.commit() .await .map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?; + // Invalidate only after commit so a concurrent read cannot repopulate the old definition. + 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(), diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index ee1d0312be..64c8ed2b29 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -774,7 +774,10 @@ pub(crate) async fn check_channel_membership( } } -fn check_token_channel_access(auth: &IngestAuth, channel_id: Uuid) -> Result<(), String> { +pub(super) fn check_token_channel_access( + auth: &IngestAuth, + channel_id: Uuid, +) -> Result<(), String> { if let Some(allowed) = auth.channel_ids() { if !allowed.contains(&channel_id) { return Err("restricted: token does not have access to this channel".to_string()); @@ -2285,7 +2288,7 @@ async fn ingest_event_inner( // Command kinds are routed AFTER signature verification, timestamp check, // pubkey/auth match, and scope validation — never before. - if buzz_core::kind::is_command_kind(kind_u32) { + if buzz_core::kind::is_command_kind(kind_u32) && kind_u32 != KIND_WORKFLOW_DEF { return super::command_executor::handle_command(tenant, state, event, auth).await; } @@ -2398,6 +2401,19 @@ async fn ingest_event_inner( } } + // Workflow saves are content writes and must not bypass the durable ban/timeout gate. + if kind_u32 == KIND_WORKFLOW_DEF { + return super::command_executor::handle_command(tenant, state, event, auth).await; + } + + // Canonical workflow deletion owns storage and both projections atomically. + // Keep it after common signature, principal, scope and moderation checks, + // but before the generic address path (which has no resolved channel). + if let Some(coordinate) = super::workflow_lifecycle::deletion_coordinate(&event)? { + return super::workflow_lifecycle::delete(tenant, state, tracer, &event, &auth, coordinate) + .await; + } + let mut channel_id = if kind_u32 == KIND_REACTION { match derive_reaction_channel(tenant.community(), &state.db, &event).await { ReactionChannelResult::Channel(ch_id) => Some(ch_id), diff --git a/crates/buzz-relay/src/handlers/mod.rs b/crates/buzz-relay/src/handlers/mod.rs index 2f4aa00b59..f326aa4d9b 100644 --- a/crates/buzz-relay/src/handlers/mod.rs +++ b/crates/buzz-relay/src/handlers/mod.rs @@ -40,6 +40,7 @@ pub mod report_resolution; pub mod req; /// NIP-29 and NIP-25 side-effect handlers. pub mod side_effects; +mod workflow_lifecycle; /// Extract an optional TTL (in seconds) from a Nostr event's `ttl` tag, /// applying the server-side override when configured. diff --git a/crates/buzz-relay/src/handlers/workflow_lifecycle.rs b/crates/buzz-relay/src/handlers/workflow_lifecycle.rs new file mode 100644 index 0000000000..a76d14f6e1 --- /dev/null +++ b/crates/buzz-relay/src/handlers/workflow_lifecycle.rs @@ -0,0 +1,284 @@ +//! Canonical workflow deletion: author/channel validation and a single commit. + +use super::ingest::{IngestAuth, IngestError, IngestResult}; +use crate::state::AppState; +use buzz_core::{kind::KIND_DELETION, TenantContext}; +use buzz_db::{ + workflow::{self, lifecycle}, + DbError, +}; +use nostr::Event; +use std::sync::Arc; +use uuid::Uuid; + +pub(super) struct Coordinate { + owner: [u8; 32], + id: Uuid, +} + +fn rejected(message: &str) -> IngestError { + IngestError::Rejected(message.into()) +} +fn db_error(error: DbError) -> IngestError { + IngestError::Internal(format!("error: workflow lifecycle: {error}")) +} + +/// Route only canonical UUID workflow a-deletes; other kind-5 operations retain their contract. +pub(super) fn deletion_coordinate(event: &Event) -> Result, IngestError> { + if u32::from(event.kind.as_u16()) != KIND_DELETION { + return Ok(None); + } + let targets: Vec<_> = event + .tags + .iter() + .filter(|t| matches!(t.kind().to_string().as_str(), "a" | "e")) + .collect(); + let Some(value) = targets.iter().find_map(|t| { + (t.kind().to_string() == "a") + .then(|| t.content()) + .flatten() + .filter(|v| v.starts_with("30620:")) + }) else { + return Ok(None); + }; + let parts: Vec<_> = value.split(':').collect(); + if parts.len() != 3 { + return Err(rejected("invalid: malformed workflow deletion coordinate")); + } + let Ok(id) = Uuid::parse_str(parts[2]) else { + return Ok(None); + }; // legacy name-based path + if targets.len() != 1 || targets[0].as_slice().len() != 2 || parts[2] != id.to_string() { + return Err(rejected( + "invalid: workflow deletion requires exactly one canonical UUID coordinate", + )); + } + let owner = hex::decode(parts[1]) + .ok() + .and_then(|bytes| <[u8; 32]>::try_from(bytes).ok()) + .ok_or_else(|| rejected("invalid: malformed workflow owner"))?; + if parts[1] != hex::encode(owner) { + return Err(rejected("invalid: noncanonical workflow owner")); + } + Ok(Some(Coordinate { owner, id })) +} + +pub(super) async fn delete( + tenant: &TenantContext, + state: &Arc, + tracer: &Arc, + event: &Event, + auth: &IngestAuth, + coordinate: Coordinate, +) -> Result { + let Coordinate { owner, id } = coordinate; + if owner != auth.pubkey().to_bytes() || owner != event.pubkey.to_bytes() { + return Err(rejected( + "forbidden: only the workflow author can delete it", + )); + } + let community = tenant.community(); + let mut tx = state + .db + .begin_event_write_transaction() + .await + .map_err(db_error)?; + buzz_deletion::store(&state.db) + .guard_transaction(&mut tx, community) + .await + .map_err(|_| rejected("restricted: community writes are fenced"))?; + lifecycle::lock_coordinate(&mut tx, community, &owner, id) + .await + .map_err(db_error)?; + let head = lifecycle::head(&mut tx, community, &owner, id) + .await + .map_err(db_error)?; + let cutoff = lifecycle::deletion(&mut tx, community, &owner, id) + .await + .map_err(db_error)?; + let runtime = match workflow::get_workflow_in_transaction(&mut tx, community, id).await { + Ok(row) => Some(row), + Err(DbError::NotFound(_)) => None, + Err(error) => return Err(db_error(error)), + }; + if runtime + .as_ref() + .is_some_and(|row| row.owner_pubkey != owner) + { + return Err(rejected("forbidden: workflow belongs to a different owner")); + } + let channel = head + .as_ref() + .and_then(|h| h.channel_id) + .or_else(|| runtime.as_ref().and_then(|r| r.channel_id)) + .or_else(|| cutoff.as_ref().map(|d| d.channel_id)) + .ok_or_else(|| rejected("invalid: workflow not found"))?; + if head.as_ref().is_some_and(|h| h.channel_id != Some(channel)) + || runtime + .as_ref() + .is_some_and(|r| r.channel_id != Some(channel)) + || cutoff.as_ref().is_some_and(|d| d.channel_id != channel) + { + return Err(rejected("conflict: workflow lifecycle is unverified")); + } + let h_tags: Vec<_> = event + .tags + .iter() + .filter(|tag| tag.kind().to_string() == "h") + .collect(); + if h_tags.len() > 1 + || h_tags.first().is_some_and(|tag| { + tag.as_slice().len() != 2 || tag.content() != Some(channel.to_string().as_str()) + }) + { + return Err(rejected( + "forbidden: workflow deletion channel does not match", + )); + } + super::ingest::check_token_channel_access(auth, channel).map_err(IngestError::AuthFailed)?; + if lifecycle::channel_role(&mut tx, community, channel, &owner) + .await + .map_err(db_error)? + .is_none() + { + return Err(rejected( + "forbidden: workflow requires active channel membership", + )); + } + // Proof is forward-only; an old accepted event is not evidence that its side effects completed. + if lifecycle::event_seen(&mut tx, community, event) + .await + .map_err(db_error)? + { + if cutoff + .as_ref() + .is_some_and(|d| d.event_id == event.id.as_bytes().as_slice()) + { + emit_success(tracer, tenant, event, auth, channel, false); + return Ok(IngestResult { + event_id: event.id.to_hex(), + accepted: true, + message: "duplicate: workflow deletion previously committed".into(), + }); + } + return Err(rejected( + "conflict: previous workflow deletion is unverified; refresh required", + )); + } + if head.is_some() != runtime.is_some() { + return Err(rejected("conflict: workflow lifecycle is unverified")); + } + let timestamp = event.created_at.as_secs() as i64; + if head + .as_ref() + .is_some_and(|h| h.created_at.timestamp() > timestamp) + || cutoff + .as_ref() + .is_some_and(|d| d.deleted_through.timestamp() >= timestamp) + { + return Err(rejected( + "conflict: workflow deletion is stale; refresh required", + )); + } + let (stored, inserted) = + buzz_db::event::insert_event_in_transaction(&mut tx, community, event, Some(channel)) + .await + .map_err(db_error)?; + if !inserted { + return Err(rejected( + "conflict: previous workflow deletion is unverified; refresh required", + )); + } + lifecycle::delete_in_transaction(&mut tx, community, &owner, id, channel, event) + .await + .map_err(db_error)?; + tx.commit().await.map_err(|e| db_error(e.into()))?; + state + .workflow_engine + .invalidate_channel_workflows(community, channel); + super::event::dispatch_persistent_event( + tenant, + state, + &stored, + KIND_DELETION, + &event.pubkey.to_hex(), + None, + ) + .await; + emit_success(tracer, tenant, event, auth, channel, true); + Ok(IngestResult { + event_id: event.id.to_hex(), + accepted: true, + message: format!( + "response:{}", + serde_json::json!({"workflow_id":id,"deleted":true,"lifecycle_version":1}) + ), + }) +} + +fn emit_success( + tracer: &Arc, + tenant: &TenantContext, + event: &Event, + auth: &IngestAuth, + channel: Uuid, + inserted: bool, +) { + use crate::conformance::{ + channel_label, claimed_community_from_event, emit, msg_id_label, state_for_request, + TraceAction, + }; + let msg_id = msg_id_label(event.id.as_bytes()); + let channel = channel_label(channel); + let claimed_community = claimed_community_from_event(event); + let action = if inserted { + TraceAction::WriteInsert { + msg_id, + channel, + claimed_community, + } + } else { + TraceAction::WriteDuplicate { + msg_id, + channel, + claimed_community, + } + }; + emit(tracer, action, state_for_request(tenant, auth.pubkey())); +} + +#[cfg(test)] +mod postgres_tests; + +#[cfg(test)] +mod tests { + use super::*; + use buzz_core::kind::KIND_WORKFLOW_DEF; + use nostr::{EventBuilder, Keys, Kind, Tag}; + #[test] + fn routes_only_one_canonical_workflow_coordinate() { + let keys = Keys::generate(); + let id = Uuid::new_v4(); + let address = format!("{KIND_WORKFLOW_DEF}:{}:{id}", keys.public_key()); + let event = |tags| { + EventBuilder::new(Kind::EventDeletion, "") + .tags(tags) + .sign_with_keys(&keys) + .expect("fixture event") + }; + let a = Tag::parse(["a", &address]).expect("a tag"); + assert!(deletion_coordinate(&event(vec![a.clone()])) + .expect("valid") + .is_some()); + assert!(deletion_coordinate(&event(vec![a.clone(), a])).is_err()); + let alias = format!("{KIND_WORKFLOW_DEF}:{}:{}", keys.public_key(), id.simple()); + assert!( + deletion_coordinate(&event(vec![Tag::parse(["a", &alias]).expect("alias")])).is_err() + ); + assert!( + deletion_coordinate(&event(vec![Tag::parse(["e", &"a".repeat(64)]).expect("e")])) + .expect("generic") + .is_none() + ); + } +} diff --git a/crates/buzz-relay/src/handlers/workflow_lifecycle/postgres_tests.rs b/crates/buzz-relay/src/handlers/workflow_lifecycle/postgres_tests.rs new file mode 100644 index 0000000000..ba7b9369d8 --- /dev/null +++ b/crates/buzz-relay/src/handlers/workflow_lifecycle/postgres_tests.rs @@ -0,0 +1,571 @@ +//! Real ingest + PostgreSQL lifecycle tests. The nextest wrapper owns a database per test. +use super::super::ingest::ingest_event; +use super::*; +use buzz_auth::Scope; +use buzz_core::{ + channel::{ChannelType, ChannelVisibility}, + kind::KIND_WORKFLOW_DEF, +}; +use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + +struct Fixture { + state: Arc, + pool: sqlx::PgPool, + tenant: TenantContext, + keys: Keys, + channel: Uuid, + id: Uuid, + now: u64, + _cache: tempfile::TempDir, +} +impl Fixture { + async fn new() -> Self { + let url = + std::env::var("BUZZ_TEST_DATABASE_URL").expect("isolated PostgreSQL URL required"); + assert_eq!( + std::env::var("BUZZ_TEST_SCHEMA_MODE").as_deref(), + Ok("desired") + ); + let pool = sqlx::PgPool::connect(&url) + .await + .expect("connect isolated database"); + let db = buzz_db::Db::from_pool(pool.clone()); + let host = format!("workflow-{}.example", Uuid::new_v4()); + let community = db + .ensure_configured_community(&host) + .await + .expect("community") + .id; + let tenant = TenantContext::resolved(community, host); + let keys = Keys::generate(); + let channel = db + .create_channel( + community, + "workflow-test", + ChannelType::Stream, + ChannelVisibility::Private, + None, + keys.public_key().as_bytes(), + None, + ) + .await + .expect("channel") + .id; + let mut config = crate::config::Config::from_env().expect("config"); + config.database_url = url; + // Deliberately no live Redis, relay, S3, subscriber or workflow executor. + // The post-commit broadcast may fail; durable completion must not depend on it. + config.redis_url = "redis://127.0.0.1:1".into(); + config.require_relay_membership = false; + let cache = tempfile::tempdir().expect("cache dir"); + config.git_pack_cache_path = cache.path().to_path_buf(); + let redis = deadpool_redis::Config::from_url(&config.redis_url) + .create_pool(Some(deadpool_redis::Runtime::Tokio1)) + .expect("redis pool"); + let pubsub = Arc::new( + buzz_pubsub::PubSubManager::new(&config.redis_url, redis.clone()) + .await + .expect("pubsub"), + ); + let engine = Arc::new(buzz_workflow::WorkflowEngine::new( + db.clone(), + buzz_workflow::WorkflowConfig::default(), + )); + let auth = buzz_auth::AuthService::new(config.auth.clone()); + let search = buzz_search::SearchService::new(pool.clone()); + let media = buzz_media::MediaStorage::new(&config.media).expect("media config"); + let (state, _) = AppState::new( + config, + db, + redis, + None, + pubsub, + auth, + search, + engine, + Keys::generate(), + media, + ); + Self { + state: Arc::new(state), + pool, + tenant, + keys, + channel, + id: Uuid::new_v4(), + now: Timestamp::now().as_secs(), + _cache: cache, + } + } + fn auth(&self) -> IngestAuth { + IngestAuth::Nip42 { + pubkey: self.keys.public_key(), + scopes: vec![Scope::MessagesWrite], + channel_ids: None, + conn_id: Uuid::new_v4(), + } + } + fn save(&self, time: u64, name: &str) -> Event { + self.sign(Kind::Custom(KIND_WORKFLOW_DEF as u16), time, + &format!("name: {name}\nenabled: false\ntrigger:\n on: message_posted\nsteps:\n - id: wait\n action: delay\n duration: 1s\n"), + vec![vec!["d".into(), self.id.to_string()], vec!["h".into(), self.channel.to_string()]]) + } + fn delete(&self, time: u64) -> Event { + self.sign( + Kind::EventDeletion, + time, + "", + vec![vec![ + "a".into(), + format!("30620:{}:{}", self.keys.public_key(), self.id), + ]], + ) + } + fn sign(&self, kind: Kind, time: u64, content: &str, tags: Vec>) -> Event { + EventBuilder::new(kind, content) + .tags(tags.into_iter().map(|t| Tag::parse(t).expect("tag"))) + .custom_created_at(Timestamp::from(time)) + .sign_with_keys(&self.keys) + .expect("signed fixture") + } + async fn send(&self, event: &Event) -> Result { + ingest_event(&self.state, &self.tenant, event.clone(), self.auth()).await + } + async fn live(&self) -> (Option>, Option, Option>) { + let mut tx = self + .state + .db + .begin_event_write_transaction() + .await + .expect("tx"); + let owner = self.keys.public_key().to_bytes(); + let head = lifecycle::head(&mut tx, self.tenant.community(), &owner, self.id) + .await + .expect("head"); + let cutoff = lifecycle::deletion(&mut tx, self.tenant.community(), &owner, self.id) + .await + .expect("cutoff"); + let runtime = + match workflow::get_workflow_in_transaction(&mut tx, self.tenant.community(), self.id) + .await + { + Ok(row) => Some(row.name), + Err(DbError::NotFound(_)) => None, + Err(e) => panic!("runtime: {e}"), + }; + (head.map(|h| h.id), runtime, cutoff.map(|d| d.event_id)) + } + async fn seen(&self, event: &Event) -> bool { + let mut tx = self + .state + .db + .begin_event_write_transaction() + .await + .expect("tx"); + lifecycle::event_seen(&mut tx, self.tenant.community(), event) + .await + .expect("seen") + } +} +fn reject(result: Result, prefix: &str) { + match result { + Err(IngestError::Rejected(message) | IngestError::AuthFailed(message)) => { + assert!(message.starts_with(prefix), "{message}") + } + Err(other) => panic!("expected {prefix}, got {other:?}"), + Ok(result) => panic!("expected {prefix}, got success {}", result.message), + } +} + +#[tokio::test] +#[ignore = "requires PostgreSQL"] +async fn ingest_save_delete_replay_and_both_timestamp_orders() { + let f = Fixture::new().await; + let create = f.save(f.now, "first"); + assert!(f.send(&create).await.expect("create").accepted); + assert_eq!( + f.live().await, + ( + Some(create.id.to_bytes().to_vec()), + Some("first".into()), + None + ) + ); + let stale = f.delete(f.now - 1); + reject(f.send(&stale).await, "conflict:"); + assert!(!f.seen(&stale).await); + // NIP-09 includes the same second; the arrival-order reversal below must agree. + let delete = f.delete(f.now); + let result = f.send(&delete).await.expect("delete"); + assert!(result.message.contains("\"deleted\":true")); + assert_eq!( + f.live().await, + (None, None, Some(delete.id.to_bytes().to_vec())) + ); + assert!(f + .send(&delete) + .await + .expect("verified replay") + .message + .starts_with("duplicate:")); + assert!(f + .send(&create) + .await + .expect("exact save replay") + .message + .starts_with("duplicate:")); + for time in [f.now - 1, f.now] { + let delayed = f.save(time, "unseen delayed save"); + reject(f.send(&delayed).await, "conflict:"); + assert!(!f.seen(&delayed).await); + } + let newer = f.save(f.now + 1, "newer"); + assert!( + f.send(&newer) + .await + .expect("intentional recreation") + .accepted + ); + assert!(f + .send(&delete) + .await + .expect("old delete replay") + .message + .starts_with("duplicate:")); + assert_eq!( + f.live().await, + ( + Some(newer.id.to_bytes().to_vec()), + Some("newer".into()), + Some(delete.id.to_bytes().to_vec()) + ) + ); +} + +#[tokio::test] +#[ignore = "requires PostgreSQL"] +async fn ingest_rolls_back_both_projections_and_proof_on_commit_failure() { + let f = Fixture::new().await; + let create = f.save(f.now, "first"); + f.send(&create).await.expect("create"); + let before = f.live().await; + // A deferred runtime trigger fails at COMMIT, after the real save upsert. + // The former pool-based upsert would fail here before the event commit; the + // delete control additionally catches an event/proof committed before runtime work. + sqlx::raw_sql("CREATE FUNCTION fail_workflow_commit() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'fixture commit failure'; END $$; CREATE CONSTRAINT TRIGGER fail_workflow_commit AFTER INSERT OR UPDATE OR DELETE ON workflows DEFERRABLE INITIALLY DEFERRED FOR EACH ROW EXECUTE FUNCTION fail_workflow_commit();") + .execute(&f.pool).await.expect("fault trigger"); + let update = f.save(f.now + 1, "failed update"); + assert!(matches!( + f.send(&update).await, + Err(IngestError::Internal(_)) + )); + assert_eq!(f.live().await, before); + assert!(!f.seen(&update).await); + let delete = f.delete(f.now + 2); + assert!(matches!( + f.send(&delete).await, + Err(IngestError::Internal(_)) + )); + assert_eq!(f.live().await, before); + assert!(!f.seen(&delete).await); + sqlx::raw_sql("DROP TRIGGER fail_workflow_commit ON workflows") + .execute(&f.pool) + .await + .expect("clear fault"); + f.send(&update) + .await + .expect("same event retry after rollback"); + f.send(&delete) + .await + .expect("same delete retry after rollback"); + assert_eq!( + f.live().await, + (None, None, Some(delete.id.to_bytes().to_vec())) + ); +} + +#[tokio::test] +#[ignore = "requires PostgreSQL"] +async fn ingest_denies_wrong_channel_token_owner_revocation_and_legacy_completion() { + let f = Fixture::new().await; + let create = f.save(f.now, "first"); + f.send(&create).await.expect("create"); + let before = f.live().await; + let delete = f.delete(f.now + 1); + let mut scoped = f.auth(); + if let IngestAuth::Nip42 { channel_ids, .. } = &mut scoped { + *channel_ids = Some(vec![Uuid::new_v4()]); + } + reject( + ingest_event(&f.state, &f.tenant, delete.clone(), scoped.clone()).await, + "restricted:", + ); + reject( + ingest_event(&f.state, &f.tenant, f.save(f.now + 1, "restricted"), scoped).await, + "restricted:", + ); + let wrong_h = f.sign( + Kind::EventDeletion, + f.now + 1, + "", + vec![ + vec![ + "a".into(), + format!("30620:{}:{}", f.keys.public_key(), f.id), + ], + vec!["h".into(), Uuid::new_v4().to_string()], + ], + ); + reject(f.send(&wrong_h).await, "forbidden:"); + let foreign = Keys::generate(); + let wrong_owner = f.sign( + Kind::EventDeletion, + f.now + 1, + "", + vec![vec![ + "a".into(), + format!("30620:{}:{}", foreign.public_key(), f.id), + ]], + ); + reject(f.send(&wrong_owner).await, "forbidden:"); + // An old accepted generic tombstone is not a forward completion proof. + let mut tx = f + .state + .db + .begin_event_write_transaction() + .await + .expect("tx"); + buzz_db::event::insert_event_in_transaction( + &mut tx, + f.tenant.community(), + &delete, + Some(f.channel), + ) + .await + .expect("legacy event"); + tx.commit().await.expect("legacy commit"); + reject(f.send(&delete).await, "conflict:"); + assert_eq!(f.live().await, before); + sqlx::query( + "UPDATE channel_members SET removed_at=NOW() WHERE community_id=$1 AND channel_id=$2", + ) + .bind(f.tenant.community().as_uuid()) + .bind(f.channel) + .execute(&f.pool) + .await + .expect("revoke"); + reject(f.send(&f.delete(f.now + 2)).await, "forbidden:"); + reject(f.send(&f.save(f.now + 2, "revoked")).await, "forbidden:"); + assert_eq!(f.live().await, before); +} + +#[tokio::test] +#[ignore = "requires PostgreSQL"] +async fn ingest_save_event_commit_failure_cannot_leave_pool_runtime_upsert() { + let f = Fixture::new().await; + let create = f.save(f.now, "first"); + f.send(&create).await.expect("create"); + let before = f.live().await; + // Events are partitioned: install the deferred failure on the real partition. + sqlx::raw_sql("CREATE FUNCTION fail_event_commit() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'fixture event commit failure'; END $$; DO $$ DECLARE partition regclass; BEGIN FOR partition IN SELECT inhrelid::regclass FROM pg_inherits WHERE inhparent='events'::regclass LOOP EXECUTE format('CREATE CONSTRAINT TRIGGER fail_event_commit AFTER INSERT ON %s DEFERRABLE INITIALLY DEFERRED FOR EACH ROW EXECUTE FUNCTION fail_event_commit()', partition); END LOOP; END $$;") + .execute(&f.pool).await.expect("event fault trigger"); + let update = f.save(f.now + 1, "must roll back runtime too"); + assert!(matches!( + f.send(&update).await, + Err(IngestError::Internal(_)) + )); + assert_eq!(f.live().await, before); + assert!(!f.seen(&update).await); +} + +#[tokio::test] +#[ignore = "requires PostgreSQL"] +async fn ingest_concurrent_save_delete_share_coordinate_lock_and_keep_newer_head() { + let f = Fixture::new().await; + f.send(&f.save(f.now, "first")).await.expect("create"); + let mut blocker = f + .state + .db + .begin_event_write_transaction() + .await + .expect("blocker"); + lifecycle::lock_coordinate( + &mut blocker, + f.tenant.community(), + f.keys.public_key().as_bytes(), + f.id, + ) + .await + .expect("coordinate lock"); + let save = f.save(f.now + 2, "newer wins"); + let delete = f.delete(f.now + 1); + let (save_result, delete_result) = tokio::join!(async { f.send(&save).await }, async { + let delete_future = f.send(&delete); + let release = async { + tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + let waiting: i64 = sqlx::query_scalar("SELECT count(*) FROM pg_locks WHERE locktype='advisory' AND NOT granted AND database=(SELECT oid FROM pg_database WHERE datname=current_database())") + .fetch_one(&f.pool).await.expect("lock waiters"); + if waiting >= 2 { break; } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }).await.expect("both production paths must wait on the coordinate lock"); + blocker.commit().await.expect("release lock"); + }; + let (result, ()) = tokio::join!(delete_future, release); + result + }); + assert!(save_result.expect("save").accepted); + match delete_result { + Ok(r) => assert!(r.accepted), + Err(IngestError::Rejected(m)) => assert!(m.starts_with("conflict:")), + other => panic!("unexpected delete: {}", other.is_ok()), + } + let (head, runtime, _) = f.live().await; + assert_eq!(head, Some(save.id.to_bytes().to_vec())); + assert_eq!(runtime, Some("newer wins".into())); +} + +#[tokio::test] +#[ignore = "requires PostgreSQL"] +async fn ingest_rejects_legacy_split_missing_archived_and_cross_community_targets() { + let f = Fixture::new().await; + reject(f.send(&f.delete(f.now)).await, "invalid:"); + let create = f.save(f.now, "first"); + f.send(&create).await.expect("create"); + let other_host = format!("other-{}.example", Uuid::new_v4()); + let other_community = f + .state + .db + .ensure_configured_community(&other_host) + .await + .expect("other") + .id; + let other = TenantContext::resolved(other_community, other_host); + reject( + ingest_event(&f.state, &other, f.delete(f.now + 1), f.auth()).await, + "invalid:", + ); + reject( + ingest_event( + &f.state, + &other, + f.save(f.now + 1, "wrong tenant"), + f.auth(), + ) + .await, + "forbidden:", + ); + let before = f.live().await; + sqlx::query("UPDATE channels SET archived_at=NOW() WHERE community_id=$1 AND id=$2") + .bind(f.tenant.community().as_uuid()) + .bind(f.channel) + .execute(&f.pool) + .await + .expect("archive"); + reject(f.send(&f.delete(f.now + 1)).await, "forbidden:"); + reject(f.send(&f.save(f.now + 1, "archived")).await, "forbidden:"); + assert_eq!(f.live().await, before); + sqlx::query("UPDATE channels SET archived_at=NULL WHERE community_id=$1 AND id=$2") + .bind(f.tenant.community().as_uuid()) + .bind(f.channel) + .execute(&f.pool) + .await + .expect("unarchive"); + sqlx::query("DELETE FROM workflows WHERE community_id=$1 AND id=$2") + .bind(f.tenant.community().as_uuid()) + .bind(f.id) + .execute(&f.pool) + .await + .expect("legacy split"); + reject(f.send(&f.delete(f.now + 1)).await, "conflict:"); + assert_eq!( + f.live().await, + (Some(create.id.to_bytes().to_vec()), None, None) + ); +} + +#[tokio::test] +#[ignore = "requires PostgreSQL"] +async fn ingest_ban_timeout_and_schema_fence_are_enforced() { + let f = Fixture::new().await; + f.send(&f.save(f.now, "first")).await.expect("create"); + for (banned, until) in [ + (true, None), + ( + false, + Some(chrono::Utc::now() + chrono::Duration::minutes(5)), + ), + ] { + sqlx::query("INSERT INTO community_bans (community_id,pubkey,banned,muted_until,actor_pubkey) VALUES ($1,$2,$3,$4,$2) ON CONFLICT (community_id,pubkey) DO UPDATE SET banned=$3,muted_until=$4") + .bind(f.tenant.community().as_uuid()).bind(f.keys.public_key().as_bytes().as_slice()).bind(banned).bind(until).execute(&f.pool).await.expect("restriction"); + let prefix = if banned { "blocked:" } else { "restricted:" }; + reject(f.send(&f.save(f.now + 1, "blocked")).await, prefix); + reject(f.send(&f.delete(f.now + 1)).await, prefix); + } + let attached: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM pg_trigger WHERE tgrelid='workflow_deletions'::regclass AND NOT tgisinternal AND tgfoid='enforce_community_write_fence()'::regprocedure)") + .fetch_one(&f.pool).await.expect("live fence catalog"); + assert!( + attached, + "desired-state deletion proof table must have a real write fence" + ); +} + +#[tokio::test] +#[ignore = "requires PostgreSQL"] +async fn ingest_concurrent_webhook_saves_preserve_one_secret_and_cas_revision() { + let f = Fixture::new().await; + f.send(&f.save(f.now, "first")).await.expect("create"); + let webhook = |time, name| { + let save = f.save(time, name); + f.sign( + save.kind, + time, + &save.content.replace("on: message_posted", "on: webhook"), + save.tags.iter().map(|t| t.as_slice().to_vec()).collect(), + ) + }; + let first = webhook(f.now + 1, "webhook first"); + let second = webhook(f.now + 2, "webhook second"); + let (a, b) = tokio::join!(f.send(&first), f.send(&second)); + let receipts: Vec = [a.expect("first outcome"), b.expect("second outcome")] + .iter() + .filter_map(|r| { + r.message + .strip_prefix("response:") + .map(|s| serde_json::from_str(s).expect("receipt")) + }) + .collect(); + let secrets: Vec<_> = receipts + .iter() + .filter_map(|v| v.get("webhook_secret").and_then(|v| v.as_str())) + .collect(); + assert_eq!( + secrets.len(), + 1, + "only the first committed webhook transition gets a secret" + ); + let row = f + .state + .db + .get_workflow(f.tenant.community(), f.id) + .await + .expect("runtime"); + assert_eq!( + crate::webhook_secret::extract_secret(&row.definition).as_deref(), + Some(secrets[0]) + ); + assert_eq!(row.name, "webhook second"); + let stale = f.sign( + first.kind, + f.now + 3, + &first.content, + vec![ + vec!["h".into(), f.channel.to_string()], + vec!["d".into(), f.id.to_string()], + vec!["expected-revision".into(), first.id.to_hex()], + ], + ); + reject(f.send(&stale).await, "conflict:"); + assert_eq!(f.live().await.0, Some(second.id.to_bytes().to_vec())); +} diff --git a/migrations/0045_workflow_deletion_cutoff.sql b/migrations/0045_workflow_deletion_cutoff.sql new file mode 100644 index 0000000000..cad6f6b988 --- /dev/null +++ b/migrations/0045_workflow_deletion_cutoff.sql @@ -0,0 +1,13 @@ +-- Forward-only proof of atomic canonical workflow deletion. Never backfill from +-- legacy kind-5 events: their side effects may not have committed. +CREATE TABLE workflow_deletions ( + community_id UUID NOT NULL REFERENCES communities(id), + owner_pubkey BYTEA NOT NULL CHECK (octet_length(owner_pubkey) = 32), + workflow_id UUID NOT NULL, + channel_id UUID NOT NULL, + deleted_through TIMESTAMPTZ NOT NULL, + event_id BYTEA NOT NULL CHECK (octet_length(event_id) = 32), + PRIMARY KEY (community_id, owner_pubkey, workflow_id), + FOREIGN KEY (community_id, channel_id) REFERENCES channels(community_id, id) +); +SELECT attach_community_write_fence('workflow_deletions'); diff --git a/schema/schema.sql b/schema/schema.sql index 0950812562..a5f108e348 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -382,6 +382,21 @@ CREATE INDEX idx_workflows_channel_active ON workflows (community_id, channel_id -- side effects run under the owning tenant's context (Lane0 contract §4a.5). CREATE INDEX idx_workflows_enabled ON workflows (enabled, status) WHERE enabled; +-- ── Workflow deletion cutoff ────────────────────────────────────────────────── + +-- Forward-only proof of atomic canonical workflow deletion. Never backfill from +-- legacy kind-5 events: their side effects may not have committed. +CREATE TABLE workflow_deletions ( + community_id UUID NOT NULL REFERENCES communities(id), + owner_pubkey BYTEA NOT NULL CHECK (octet_length(owner_pubkey) = 32), + workflow_id UUID NOT NULL, + channel_id UUID NOT NULL, + deleted_through TIMESTAMPTZ NOT NULL, + event_id BYTEA NOT NULL CHECK (octet_length(event_id) = 32), + PRIMARY KEY (community_id, owner_pubkey, workflow_id), + FOREIGN KEY (community_id, channel_id) REFERENCES channels(community_id, id) +); + -- ── Workflow runs ───────────────────────────────────────────────────────────── CREATE TABLE workflow_runs ( @@ -1752,6 +1767,7 @@ SELECT attach_community_write_fence('thread_metadata'); SELECT attach_community_write_fence('users'); SELECT attach_community_write_fence('workflow_approvals'); SELECT attach_community_write_fence('workflow_runs'); +SELECT attach_community_write_fence('workflow_deletions'); SELECT attach_community_write_fence('workflows'); -- ── Relay operator/moderator roster ──────────────────────────────────────────