diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 9b26876747..e171ae322e 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -3750,6 +3750,27 @@ impl Db { workflow::delete_workflow_for_owner(&self.pool, community_id, id, owner_pubkey).await } + /// Atomically delete an owned workflow and tombstone its kind-30620 + /// definition, respecting the deletion event's NIP-09 timestamp. + pub async fn delete_workflow_and_definition_for_owner( + &self, + community_id: CommunityId, + id: Uuid, + owner_pubkey: &[u8], + d_tag: &str, + deletion_created_at_secs: i64, + ) -> Result { + workflow::delete_workflow_and_definition_for_owner( + &self.pool, + community_id, + id, + owner_pubkey, + d_tag, + deletion_created_at_secs, + ) + .await + } + /// Find a workflow by owner pubkey and name within a community. Used for /// NIP-09 a-tag deletion where the d-tag is the workflow name (not UUID). pub async fn find_workflow_by_owner_and_name( diff --git a/crates/buzz-db/src/workflow.rs b/crates/buzz-db/src/workflow.rs index 7a2396c1fd..db5d783503 100644 --- a/crates/buzz-db/src/workflow.rs +++ b/crates/buzz-db/src/workflow.rs @@ -15,7 +15,7 @@ use sha2::{Digest, Sha256}; use sqlx::{PgPool, Row}; use uuid::Uuid; -use buzz_core::CommunityId; +use buzz_core::{kind::KIND_WORKFLOW_DEF, CommunityId}; use crate::error::{DbError, Result}; @@ -187,6 +187,21 @@ pub struct WorkflowRecord { pub updated_at: DateTime, } +/// Result of atomically deleting a workflow and tombstoning its definition. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum WorkflowDeleteOutcome { + /// The workflow row, definition event, or both were deleted. + Deleted { + /// Channel whose workflow cache must be invalidated, when a workflow + /// row was present. + channel_id: Option, + /// Whether the live kind-30620 definition event was tombstoned. + definition_deleted: bool, + }, + /// The deletion event predates the live definition and was ignored. + Stale, +} + /// A single execution of a workflow. #[derive(Debug, Clone)] pub struct WorkflowRunRecord { @@ -787,6 +802,88 @@ pub async fn delete_workflow_for_owner( } } +/// Atomically delete an owned workflow row and tombstone its live definition. +/// +/// The definition coordinate is `(kind:30620, owner_pubkey, d_tag)`. The event +/// is tombstoned only when it is no newer than `deletion_created_at_secs`, per +/// NIP-09 ordering. If a newer live definition exists, the transaction returns +/// [`WorkflowDeleteOutcome::Stale`] without deleting the workflow row. This +/// keeps the query-visible definition and executable record consistent across +/// crashes and retries. +pub async fn delete_workflow_and_definition_for_owner( + pool: &PgPool, + community_id: CommunityId, + id: Uuid, + owner_pubkey: &[u8], + d_tag: &str, + deletion_created_at_secs: i64, +) -> Result { + let deletion_created_at = DateTime::from_timestamp(deletion_created_at_secs, 0) + .ok_or(DbError::InvalidTimestamp(deletion_created_at_secs))?; + let mut tx = pool.begin().await?; + + let definition_deleted = sqlx::query( + "UPDATE events SET deleted_at = NOW() \ + WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 \ + AND deleted_at IS NULL AND created_at <= $5", + ) + .bind(community_id.as_uuid()) + .bind(KIND_WORKFLOW_DEF as i32) + .bind(owner_pubkey) + .bind(d_tag) + .bind(deletion_created_at) + .execute(&mut *tx) + .await? + .rows_affected() + > 0; + + if !definition_deleted { + let newer_definition_exists: bool = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM events \ + WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 \ + AND deleted_at IS NULL AND created_at > $5)", + ) + .bind(community_id.as_uuid()) + .bind(KIND_WORKFLOW_DEF as i32) + .bind(owner_pubkey) + .bind(d_tag) + .bind(deletion_created_at) + .fetch_one(&mut *tx) + .await?; + + if newer_definition_exists { + tx.rollback().await?; + return Ok(WorkflowDeleteOutcome::Stale); + } + } + + let row = sqlx::query( + "DELETE FROM workflows WHERE community_id = $1 AND id = $2 AND owner_pubkey = $3 \ + RETURNING channel_id", + ) + .bind(community_id.as_uuid()) + .bind(id) + .bind(owner_pubkey) + .fetch_optional(&mut *tx) + .await?; + let workflow_deleted = row.is_some(); + let channel_id: Option = match row { + Some(row) => row.try_get("channel_id")?, + None => None, + }; + + if !workflow_deleted && !definition_deleted { + tx.rollback().await?; + return Err(DbError::NotFound(format!("workflow {id}"))); + } + + tx.commit().await?; + Ok(WorkflowDeleteOutcome::Deleted { + channel_id, + definition_deleted, + }) +} + // -- Workflow Run CRUD -------------------------------------------------------- /// Insert a new workflow run. Returns the new run's UUID. @@ -2081,6 +2178,174 @@ mod tests { .expect("insert workflow"); } + async fn insert_workflow_definition_event( + pool: &PgPool, + community: CommunityId, + owner: &[u8], + d_tag: &str, + created_at: DateTime, + ) { + let event_id = Uuid::new_v4().as_bytes().repeat(2); + let tags = serde_json::json!([["d", d_tag]]); + sqlx::query( + "INSERT INTO events \ + (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, d_tag) \ + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, NOW(), $9)", + ) + .bind(community.as_uuid()) + .bind(event_id) + .bind(owner) + .bind(created_at) + .bind(KIND_WORKFLOW_DEF as i32) + .bind(tags) + .bind("{}") + .bind(vec![0u8; 64]) + .bind(d_tag) + .execute(pool) + .await + .expect("insert workflow definition event"); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn workflow_delete_atomically_tombstones_definition() { + let pool = setup_pool().await; + let community = make_community(&pool).await; + let workflow_id = Uuid::new_v4(); + let channel_id = Uuid::new_v4(); + let owner = vec![0xb2; 32]; + let d_tag = workflow_id.to_string(); + let definition_created_at = Utc.timestamp_opt(1_700_000_000, 0).unwrap(); + + insert_workflow_with_ids(&pool, community, workflow_id, channel_id, "wf-delete").await; + insert_workflow_definition_event(&pool, community, &owner, &d_tag, definition_created_at) + .await; + + let outcome = delete_workflow_and_definition_for_owner( + &pool, + community, + workflow_id, + &owner, + &d_tag, + definition_created_at.timestamp() + 1, + ) + .await + .expect("atomic workflow deletion"); + assert_eq!( + outcome, + WorkflowDeleteOutcome::Deleted { + channel_id: Some(channel_id), + definition_deleted: true, + } + ); + assert!(matches!( + get_workflow(&pool, community, workflow_id).await, + Err(DbError::NotFound(_)) + )); + + let live_definition_count: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM events \ + WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 \ + AND deleted_at IS NULL", + ) + .bind(community.as_uuid()) + .bind(KIND_WORKFLOW_DEF as i32) + .bind(&owner) + .bind(&d_tag) + .fetch_one(&pool) + .await + .expect("count live definitions"); + assert_eq!(live_definition_count, 0); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn stale_workflow_delete_preserves_definition_and_record() { + let pool = setup_pool().await; + let community = make_community(&pool).await; + let workflow_id = Uuid::new_v4(); + let channel_id = Uuid::new_v4(); + let owner = vec![0xb2; 32]; + let d_tag = workflow_id.to_string(); + let definition_created_at = Utc.timestamp_opt(1_700_000_100, 0).unwrap(); + + insert_workflow_with_ids(&pool, community, workflow_id, channel_id, "wf-stale").await; + insert_workflow_definition_event(&pool, community, &owner, &d_tag, definition_created_at) + .await; + + let outcome = delete_workflow_and_definition_for_owner( + &pool, + community, + workflow_id, + &owner, + &d_tag, + definition_created_at.timestamp() - 1, + ) + .await + .expect("stale workflow deletion"); + assert_eq!(outcome, WorkflowDeleteOutcome::Stale); + assert!(get_workflow(&pool, community, workflow_id).await.is_ok()); + + let live_definition_count: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM events \ + WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 \ + AND deleted_at IS NULL", + ) + .bind(community.as_uuid()) + .bind(KIND_WORKFLOW_DEF as i32) + .bind(&owner) + .bind(&d_tag) + .fetch_one(&pool) + .await + .expect("count live definitions"); + assert_eq!(live_definition_count, 1); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn workflow_delete_commits_global_row_when_definition_is_already_absent() { + let pool = setup_pool().await; + let community = make_community(&pool).await; + let owner = vec![0xc3; 32]; + ensure_user(&pool, community, &owner) + .await + .expect("ensure owner"); + let workflow_id = create_workflow( + &pool, + community, + None, + &owner, + "global-workflow", + r#"{"trigger":{"on":"schedule"},"steps":[]}"#, + &[0u8; 32], + ) + .await + .expect("create global workflow"); + let d_tag = workflow_id.to_string(); + + let outcome = delete_workflow_and_definition_for_owner( + &pool, + community, + workflow_id, + &owner, + &d_tag, + Utc::now().timestamp(), + ) + .await + .expect("delete global workflow without definition event"); + assert_eq!( + outcome, + WorkflowDeleteOutcome::Deleted { + channel_id: None, + definition_deleted: false, + } + ); + assert!(matches!( + get_workflow(&pool, community, workflow_id).await, + Err(DbError::NotFound(_)) + )); + } + /// Issue 4 (workflow identity): the same workflow UUID and channel UUID can /// exist in communities A and B (PK `(community_id, id)`). A request-scoped /// `get_workflow` / `list_enabled_channel_workflows` MUST return only the diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 660a55fef3..9cb70a7018 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -2094,7 +2094,8 @@ async fn handle_a_tag_deletion( .map_err(|_| anyhow::anyhow!("invalid kind in a-tag"))?; let pubkey_hex = parts[1]; let d_tag = parts[2]; - let actor_bytes = effective_message_author(event, &state.relay_keypair.public_key()); + let target_owner_bytes = hex::decode(pubkey_hex) + .map_err(|e| anyhow::anyhow!("invalid pubkey hex in a-tag {pubkey_hex}: {e}"))?; match kind_num { // kind:30350 revocation is exclusively a higher-generation inactive replacement. @@ -2102,40 +2103,90 @@ async fn handle_a_tag_deletion( tracing::debug!(d_tag, "NIP-09 deletion ignored for push lease"); } buzz_core::kind::KIND_WORKFLOW_DEF => { + let deletion_created_at = event.created_at.as_secs() as i64; // Try UUID first (workflow_id); fall back to name-based lookup. if let Ok(wf_id) = uuid::Uuid::parse_str(d_tag) { - let channel_id = state + let outcome = state .db - .delete_workflow_for_owner(tenant.community(), wf_id, &actor_bytes) + .delete_workflow_and_definition_for_owner( + tenant.community(), + wf_id, + &target_owner_bytes, + d_tag, + deletion_created_at, + ) .await .map_err(|e| anyhow::anyhow!("failed to delete workflow {wf_id}: {e}"))?; - if let Some(channel_id) = channel_id { - state - .workflow_engine - .invalidate_channel_workflows(tenant.community(), channel_id); + match outcome { + buzz_db::workflow::WorkflowDeleteOutcome::Deleted { + channel_id, + definition_deleted, + } => { + if let Some(channel_id) = channel_id { + state + .workflow_engine + .invalidate_channel_workflows(tenant.community(), channel_id); + } + tracing::info!( + workflow_id = %wf_id, + definition_deleted, + "Workflow and definition deleted via NIP-09 a-tag (UUID)" + ); + } + buzz_db::workflow::WorkflowDeleteOutcome::Stale => { + tracing::info!( + workflow_id = %wf_id, + "Ignored stale NIP-09 workflow deletion" + ); + } } - tracing::info!(workflow_id = %wf_id, "Workflow deleted via NIP-09 a-tag (UUID)"); } else { // Name-based lookup match state .db - .find_workflow_by_owner_and_name(tenant.community(), &actor_bytes, d_tag) + .find_workflow_by_owner_and_name(tenant.community(), &target_owner_bytes, d_tag) .await { Ok(Some(wf)) => { - let channel_id = state + let outcome = state .db - .delete_workflow_for_owner(tenant.community(), wf.id, &actor_bytes) + .delete_workflow_and_definition_for_owner( + tenant.community(), + wf.id, + &target_owner_bytes, + d_tag, + deletion_created_at, + ) .await .map_err(|e| { anyhow::anyhow!("failed to delete workflow {}: {e}", wf.id) })?; - if let Some(channel_id) = channel_id { - state - .workflow_engine - .invalidate_channel_workflows(tenant.community(), channel_id); + match outcome { + buzz_db::workflow::WorkflowDeleteOutcome::Deleted { + channel_id, + definition_deleted, + } => { + if let Some(channel_id) = channel_id { + state.workflow_engine.invalidate_channel_workflows( + tenant.community(), + channel_id, + ); + } + tracing::info!( + workflow_id = %wf.id, + name = d_tag, + definition_deleted, + "Workflow and definition deleted via NIP-09 a-tag (name)" + ); + } + buzz_db::workflow::WorkflowDeleteOutcome::Stale => { + tracing::info!( + workflow_id = %wf.id, + name = d_tag, + "Ignored stale NIP-09 workflow deletion" + ); + } } - tracing::info!(workflow_id = %wf.id, name = d_tag, "Workflow deleted via NIP-09 a-tag (name)"); } Ok(None) => { tracing::warn!( @@ -2150,11 +2201,11 @@ async fn handle_a_tag_deletion( } // Generic NIP-33 (parameterized-replaceable) soft-delete by coordinate. // - // Listed after the workflow branch so workflow's bespoke deletion - // (which doesn't soft-delete the `events` row by design — that's a - // separate concern) takes precedence. For every other addressable - // kind, including kind:30023 (NIP-23 long-form), we soft-delete the - // live row matching `(kind, pubkey, d_tag)` so REQs stop returning it. + // Listed after the workflow branch because workflow deletion atomically + // tombstones its definition alongside the executable row. For every + // other addressable kind, including kind:30023 (NIP-23 long-form), we + // soft-delete the live row matching `(kind, pubkey, d_tag)` so REQs stop + // returning it. // See https://github.com/block/sprout/issues/714. k if is_parameterized_replaceable(k) => { let pubkey_bytes = match hex::decode(pubkey_hex) {