fix(workflows): retire deleted definitions atomically

crates/buzz-db/src/lib.rs

@@ -50,7 +50,7 @@ pub use store::{
     admin_moderation, allowlist, api_token, archived_identities, channel, channel_members,
     community, deletion, dm, event, feed, git_repo, moderation, partition, product_feedback, push,
     reaction, relay_admin_actions, relay_invite, relay_members, relay_operators, reminder,
-    replaceable, thread, usage, user, workflow,
+    replaceable, thread, usage, user, workflow, workflow_deletion,
 };
 
 pub use allowlist::AllowlistEntry;

crates/buzz-db/src/store/mod.rs

@@ -54,3 +54,8 @@ pub mod usage;
 pub mod user;
 /// Workflow, run, and approval persistence.
 pub mod workflow;
+/// Atomic workflow deletion and definition retirement.
+pub mod workflow_deletion;
+
+#[cfg(test)]
+mod workflow_deletion_postgres_tests;

crates/buzz-db/src/store/workflow_deletion.rs

@@ -0,0 +1,79 @@
+//! Transaction-bound deletion of executable workflows and signed definitions.
+
+use std::collections::BTreeSet;
+
+use buzz_core::{kind::KIND_WORKFLOW_DEF, CommunityId};
+use chrono::{DateTime, Utc};
+use sqlx::{Postgres, Transaction};
+use uuid::Uuid;
+
+use crate::{DbError, Result};
+
+/// Retire a workflow coordinate under the same lock used by definition writes.
+///
+/// The caller must authorize the coordinate owner and commit the signed deletion
+/// in this transaction. A newer live definition preserves the executable row;
+/// missing rows are allowed so retries can repair definitions left by old relays.
+pub async fn delete_in_transaction(
+    tx: &mut Transaction<'_, Postgres>,
+    community: CommunityId,
+    owner: &[u8],
+    workflow_id: Uuid,
+    coordinate: &str,
+    deleted_at_secs: i64,
+) -> Result<Vec<Uuid>> {
+    let cutoff = DateTime::<Utc>::from_timestamp(deleted_at_secs, 0)
+        .ok_or(DbError::InvalidTimestamp(deleted_at_secs))?;
+    let lock_key = super::replaceable::event_replacement_lock_key(
+        community,
+        KIND_WORKFLOW_DEF as i32,
+        owner,
+        Some(coordinate.as_bytes()),
+    );
+    sqlx::query("SELECT pg_advisory_xact_lock($1)")
+        .bind(lock_key)
+        .execute(&mut **tx)
+        .await?;
+
+    let newer_definition: 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.as_uuid())
+    .bind(KIND_WORKFLOW_DEF as i32)
+    .bind(owner)
+    .bind(coordinate)
+    .bind(cutoff)
+    .fetch_one(&mut **tx)
+    .await?;
+
+    let mut channels = BTreeSet::new();
+    if !newer_definition {
+        let removed: Option<Option<Uuid>> = sqlx::query_scalar(
+            "DELETE FROM workflows WHERE community_id = $1 AND id = $2 AND owner_pubkey = $3 \
+             RETURNING channel_id",
+        )
+        .bind(community.as_uuid())
+        .bind(workflow_id)
+        .bind(owner)
+        .fetch_optional(&mut **tx)
+        .await?;
+        channels.extend(removed.flatten());
+    }
+
+    let definition_channels: Vec<Option<Uuid>> = sqlx::query_scalar(
+        "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 RETURNING channel_id",
+    )
+    .bind(community.as_uuid())
+    .bind(KIND_WORKFLOW_DEF as i32)
+    .bind(owner)
+    .bind(coordinate)
+    .bind(cutoff)
+    .fetch_all(&mut **tx)
+    .await?;
+    channels.extend(definition_channels.into_iter().flatten());
+    Ok(channels.into_iter().collect())
+}

crates/buzz-db/src/store/workflow_deletion_postgres_tests.rs

@@ -0,0 +1,196 @@
+use super::workflow_deletion::delete_in_transaction;
+use buzz_core::CommunityId;
+use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp};
+use sqlx::PgPool;
+use uuid::Uuid;
+
+struct Fixture {
+    pool: PgPool,
+    community: CommunityId,
+    keys: Keys,
+    id: Uuid,
+    timestamp: i64,
+}
+
+impl Fixture {
+    async fn new() -> Self {
+        let pool = PgPool::connect(&crate::test_support::database_url())
+            .await
+            .unwrap();
+        let community = CommunityId::from_uuid(Uuid::new_v4());
+        sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
+            .bind(community.as_uuid())
+            .bind(format!("deletion-{}.example", community.as_uuid()))
+            .execute(&pool)
+            .await
+            .unwrap();
+        let keys = Keys::generate();
+        let f = Self {
+            pool,
+            community,
+            keys,
+            id: Uuid::new_v4(),
+            timestamp: Timestamp::now().as_secs() as i64,
+        };
+        f.seed(f.community).await;
+        f
+    }
+
+    async fn seed(&self, community: CommunityId) {
+        crate::user::ensure_user(&self.pool, community, &self.keys.public_key().to_bytes())
+            .await
+            .unwrap();
+        crate::workflow::upsert_workflow(
+            &self.pool,
+            community,
+            self.id,
+            None,
+            &self.keys.public_key().to_bytes(),
+            "deletion-test",
+            "{}",
+            &[0; 32],
+        )
+        .await
+        .unwrap();
+        let event = EventBuilder::new(Kind::Custom(30620), "definition")
+            .tags([Tag::parse(["d", &self.id.to_string()]).unwrap()])
+            .custom_created_at(Timestamp::from(self.timestamp as u64))
+            .sign_with_keys(&self.keys)
+            .unwrap();
+        crate::event::insert_event(&self.pool, community, &event, None)
+            .await
+            .unwrap();
+    }
+
+    async fn counts(&self, community: CommunityId) -> (i64, i64) {
+        sqlx::query_as(
+            "SELECT (SELECT count(*) FROM workflows WHERE community_id = $1 AND id = $2), \
+             (SELECT count(*) FROM events WHERE community_id = $1 AND kind = 30620 \
+              AND d_tag = $3 AND deleted_at IS NULL)",
+        )
+        .bind(community.as_uuid())
+        .bind(self.id)
+        .bind(self.id.to_string())
+        .fetch_one(&self.pool)
+        .await
+        .unwrap()
+    }
+
+    async fn delete(&self, owner: &[u8], timestamp: i64, commit: bool) {
+        let mut tx = self.pool.begin().await.unwrap();
+        delete_in_transaction(
+            &mut tx,
+            self.community,
+            owner,
+            self.id,
+            &self.id.to_string(),
+            timestamp,
+        )
+        .await
+        .unwrap();
+        if commit {
+            tx.commit().await.unwrap();
+        } else {
+            tx.rollback().await.unwrap();
+        }
+    }
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn rollback_preserves_both_rows_and_commit_retires_both() {
+    let f = Fixture::new().await;
+    f.delete(&f.keys.public_key().to_bytes(), f.timestamp, false)
+        .await;
+    assert_eq!(f.counts(f.community).await, (1, 1));
+    f.delete(&f.keys.public_key().to_bytes(), f.timestamp, true)
+        .await;
+    assert_eq!(f.counts(f.community).await, (0, 0));
+    assert!(
+        crate::workflow::claim_scheduled_workflow_fire(
+            &f.pool,
+            f.community,
+            f.id,
+            chrono::Utc::now()
+        )
+        .await
+        .unwrap()
+        .is_none(),
+        "deleted workflow cannot acquire a scheduled execution claim"
+    );
+    f.delete(&f.keys.public_key().to_bytes(), f.timestamp, true)
+        .await;
+    assert_eq!(f.counts(f.community).await, (0, 0));
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn missing_runtime_still_retires_orphan_definition() {
+    let f = Fixture::new().await;
+    sqlx::query("DELETE FROM workflows WHERE community_id = $1 AND id = $2")
+        .bind(f.community.as_uuid())
+        .bind(f.id)
+        .execute(&f.pool)
+        .await
+        .unwrap();
+    assert_eq!(f.counts(f.community).await, (0, 1));
+    f.delete(&f.keys.public_key().to_bytes(), f.timestamp, true)
+        .await;
+    assert_eq!(f.counts(f.community).await, (0, 0));
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn owner_timestamp_and_community_boundaries_preserve_unrelated_rows() {
+    let f = Fixture::new().await;
+    let other = Fixture::new().await;
+    f.seed(other.community).await;
+    f.delete(&other.keys.public_key().to_bytes(), f.timestamp, true)
+        .await;
+    assert_eq!(f.counts(f.community).await, (1, 1));
+    f.delete(&f.keys.public_key().to_bytes(), f.timestamp - 1, true)
+        .await;
+    assert_eq!(f.counts(f.community).await, (1, 1));
+    f.delete(&f.keys.public_key().to_bytes(), f.timestamp, true)
+        .await;
+    assert_eq!(f.counts(f.community).await, (0, 0));
+    assert_eq!(f.counts(other.community).await, (1, 1));
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn deletion_waits_for_the_definition_writer_coordinate_lock() {
+    let f = Fixture::new().await;
+    let mut writer = f.pool.begin().await.unwrap();
+    let lock = super::replaceable::event_replacement_lock_key(
+        f.community,
+        30620,
+        &f.keys.public_key().to_bytes(),
+        Some(f.id.to_string().as_bytes()),
+    );
+    sqlx::query("SELECT pg_advisory_xact_lock($1)")
+        .bind(lock)
+        .execute(&mut *writer)
+        .await
+        .unwrap();
+    let owner = f.keys.public_key().to_bytes();
+    let coordinate = f.id.to_string();
+    let mut tx = f.pool.begin().await.unwrap();
+    {
+        let deletion =
+            delete_in_transaction(&mut tx, f.community, &owner, f.id, &coordinate, f.timestamp);
+        tokio::pin!(deletion);
+        assert!(
+            tokio::time::timeout(std::time::Duration::from_millis(100), &mut deletion)
+                .await
+                .is_err()
+        );
+        writer.commit().await.unwrap();
+        tokio::time::timeout(std::time::Duration::from_secs(5), deletion)
+            .await
+            .unwrap()
+            .unwrap();
+    }
+    tx.commit().await.unwrap();
+    assert_eq!(f.counts(f.community).await, (0, 0));
+}

crates/buzz-relay/src/handlers/ingest.rs

@@ -3144,7 +3144,12 @@ async fn ingest_event_inner(
         });
     }
 
-    let (stored_event, was_inserted) = if buzz_core::kind::is_replaceable(kind_u32) {
+    let workflow_deletion =
+        super::workflow_deletion::persist(tenant, state, &event, channel_id).await?;
+    let workflow_deletion_applied = workflow_deletion.is_some();
+    let (stored_event, was_inserted) = if let Some(result) = workflow_deletion {
+        result
+    } else if buzz_core::kind::is_replaceable(kind_u32) {
         // NIP-16 replaceable event — atomic replace with stale-write protection.
         // channel_id is None for global kinds (0, 1, 3) due to step 5b above.
         state
@@ -3211,7 +3216,7 @@ async fn ingest_event_inner(
         });
     }
 
-    if crate::handlers::side_effects::is_side_effect_kind(kind_u32) {
+    if !workflow_deletion_applied && crate::handlers::side_effects::is_side_effect_kind(kind_u32) {
         if let Err(e) =
             crate::handlers::side_effects::handle_side_effects(tenant, kind_u32, &event, state)
                 .await

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_deletion;
 
 /// Extract an optional TTL (in seconds) from a Nostr event's `ttl` tag,
 /// applying the server-side override when configured.