diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 625eb50737f..a6a402840d5 100644 --- a/crates/buzz-db/src/lib.rs +++ b/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; diff --git a/crates/buzz-db/src/store/mod.rs b/crates/buzz-db/src/store/mod.rs index 1fa1273eb0f..2988dd0581a 100644 --- a/crates/buzz-db/src/store/mod.rs +++ b/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; diff --git a/crates/buzz-db/src/store/workflow_deletion.rs b/crates/buzz-db/src/store/workflow_deletion.rs new file mode 100644 index 00000000000..5513aac059d --- /dev/null +++ b/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> { + let cutoff = DateTime::::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> = 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> = 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()) +} diff --git a/crates/buzz-db/src/store/workflow_deletion_postgres_tests.rs b/crates/buzz-db/src/store/workflow_deletion_postgres_tests.rs new file mode 100644 index 00000000000..007dee8b0f2 --- /dev/null +++ b/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)); +} diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index ee1d0312be9..62e7155a1d1 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/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 diff --git a/crates/buzz-relay/src/handlers/mod.rs b/crates/buzz-relay/src/handlers/mod.rs index 2f4aa00b595..fb50d2fe20e 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_deletion; /// 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_deletion.rs b/crates/buzz-relay/src/handlers/workflow_deletion.rs new file mode 100644 index 00000000000..4dc5ef02fbb --- /dev/null +++ b/crates/buzz-relay/src/handlers/workflow_deletion.rs @@ -0,0 +1,82 @@ +//! Canonical workflow deletes commit their signed event and effects together. + +use std::sync::Arc; + +use buzz_core::{kind::KIND_DELETION, kind::KIND_WORKFLOW_DEF, tenant::TenantContext, StoredEvent}; +use nostr::{Event, TagKind}; +use uuid::Uuid; + +use super::ingest::IngestError; +use crate::state::AppState; + +/// Persist an already-authorized canonical workflow deletion, if applicable. +/// Legacy name coordinates retain their existing handler. +pub(super) async fn persist( + tenant: &TenantContext, + state: &Arc, + event: &Event, + channel_id: Option, +) -> Result, IngestError> { + if event.kind.as_u16() as u32 != KIND_DELETION { + return Ok(None); + } + let Some(address) = event.tags.iter().find_map(|tag| { + (tag.kind() == TagKind::a()) + .then(|| tag.content()) + .flatten() + }) else { + return Ok(None); + }; + let mut parts = address.splitn(3, ':'); + if parts.next().and_then(|kind| kind.parse::().ok()) != Some(KIND_WORKFLOW_DEF) { + return Ok(None); + } + let owner_hex = parts.next().unwrap_or_default(); + let coordinate = parts.next().unwrap_or_default(); + let Ok(workflow_id) = Uuid::parse_str(coordinate) else { + return Ok(None); + }; + let owner = hex::decode(owner_hex) + .map_err(|_| IngestError::Rejected("invalid: workflow coordinate owner".into()))?; + + let mut tx = state + .db + .begin_event_write_transaction() + .await + .map_err(database_error)?; + buzz_deletion::store(&state.db) + .guard_transaction(&mut tx, tenant.community()) + .await + .map_err(|error| { + IngestError::Rejected(format!("restricted: community writes are fenced: {error}")) + })?; + let channels = buzz_db::workflow_deletion::delete_in_transaction( + &mut tx, + tenant.community(), + &owner, + workflow_id, + coordinate, + event.created_at.as_secs() as i64, + ) + .await + .map_err(database_error)?; + // Do not skip effects for an exact replay: older relays may have accepted + // the deletion while leaving a live definition behind. + let stored = + buzz_db::event::insert_event_in_transaction(&mut tx, tenant.community(), event, channel_id) + .await + .map_err(database_error)?; + tx.commit().await.map_err(|error| { + IngestError::Internal(format!("error: committing workflow deletion: {error}")) + })?; + for channel in channels { + state + .workflow_engine + .invalidate_channel_workflows(tenant.community(), channel); + } + Ok(Some(stored)) +} + +fn database_error(error: buzz_db::DbError) -> IngestError { + IngestError::Internal(format!("error: persisting workflow deletion: {error}")) +} diff --git a/crates/buzz-test-client/tests/e2e_workflow_deletion.rs b/crates/buzz-test-client/tests/e2e_workflow_deletion.rs new file mode 100644 index 00000000000..e55ade1ad7f --- /dev/null +++ b/crates/buzz-test-client/tests/e2e_workflow_deletion.rs @@ -0,0 +1,164 @@ +//! Real ingest/query regressions. Use only an isolated relay and test accounts. +//! Set RELAY_URL, BUZZ_TEST_OWNER_PRIVATE_KEY, BUZZ_TEST_MEMBER_PRIVATE_KEY, +//! and BUZZ_TEST_CHANNEL_ID, then run this ignored test target. + +use base64::{engine::general_purpose::STANDARD, Engine}; +use nostr::hashes::{sha256, Hash}; +use nostr::nips::nip98::{HttpData, HttpMethod}; +use nostr::{Event, EventBuilder, JsonUtil, Keys, Kind, Tag, Timestamp}; +use serde_json::{json, Value}; +use uuid::Uuid; + +struct Fixture { + url: String, + owner: Keys, + member: Keys, + channel: Uuid, +} + +impl Fixture { + fn new() -> Self { + Self { + url: std::env::var("RELAY_URL") + .expect("isolated RELAY_URL") + .replace("ws://", "http://") + .replace("wss://", "https://") + .trim_end_matches('/') + .to_owned(), + owner: Keys::parse(&std::env::var("BUZZ_TEST_OWNER_PRIVATE_KEY").unwrap()).unwrap(), + member: Keys::parse(&std::env::var("BUZZ_TEST_MEMBER_PRIVATE_KEY").unwrap()).unwrap(), + channel: std::env::var("BUZZ_TEST_CHANNEL_ID") + .unwrap() + .parse() + .unwrap(), + } + } + + async fn post(&self, keys: &Keys, path: &str, body: Value) -> Value { + let url = format!("{}{path}", self.url); + let body = body.to_string(); + let http = HttpData::new(url.parse().unwrap(), HttpMethod::POST) + .payload(sha256::Hash::hash(body.as_bytes())); + let auth = EventBuilder::http_auth(http) + .tag(Tag::parse(["nonce", &Uuid::new_v4().to_string()]).unwrap()) + .sign_with_keys(keys) + .unwrap(); + reqwest::Client::new() + .post(url) + .header( + "Authorization", + format!("Nostr {}", STANDARD.encode(auth.as_json())), + ) + .header("Content-Type", "application/json") + .body(body) + .send() + .await + .unwrap() + .json() + .await + .unwrap() + } + + async fn submit(&self, keys: &Keys, event: &Event) -> Value { + self.post(keys, "/events", serde_json::to_value(event).unwrap()) + .await + } + + async fn accepted(&self, event: &Event) { + let response = self.submit(&self.owner, event).await; + assert_eq!(response["accepted"], true, "{response}"); + } + + async fn query(&self, filter: Value) -> Vec { + serde_json::from_value(self.post(&self.owner, "/query", json!([filter])).await).unwrap() + } + + async fn definition(&self, id: Uuid) -> Vec { + self.query(json!({"kinds": [30620], "#d": [id.to_string()]})) + .await + } + + fn deletion(&self, id: Uuid, timestamp: Timestamp) -> Event { + buzz_sdk::build_workflow_delete(&self.owner.public_key().to_hex(), id) + .unwrap() + .custom_created_at(timestamp) + .sign_with_keys(&self.owner) + .unwrap() + } + + async fn create(&self, id: Uuid, timestamp: Timestamp) { + let event = buzz_sdk::build_workflow_def(self.channel, id, + "name: delete-regression\ntrigger:\n on: schedule\n cron: '0 0 1 1 *'\nsteps:\n - id: marker\n action: send_message\n text: isolated deletion regression\n") + .unwrap().custom_created_at(timestamp).sign_with_keys(&self.owner).unwrap(); + self.accepted(&event).await; + assert_eq!(self.definition(id).await.len(), 1); + } +} + +#[tokio::test] +#[ignore = "requires isolated relay and two test accounts"] +async fn deletion_removes_get_list_and_execution_and_replay_is_safe() { + let f = Fixture::new(); + let id = Uuid::new_v4(); + let now = Timestamp::now(); + f.create(id, now).await; + + // A different authenticated channel member cannot delete this coordinate. + let deletion = f.deletion(id, now); + let forged = EventBuilder::new(Kind::Custom(5), "") + .tags(deletion.tags.clone()) + .sign_with_keys(&f.member) + .unwrap(); + let rejected = f.submit(&f.member, &forged).await; + assert_ne!(rejected["accepted"], true, "{rejected}"); + assert_eq!(f.definition(id).await.len(), 1); + + f.accepted(&deletion).await; + for _ in 0..2 { + assert!( + f.definition(id).await.is_empty(), + "get must omit deleted definition" + ); + let list = f + .query(json!({"kinds": [30620], "#h": [f.channel.to_string()]})) + .await; + assert!( + list.iter() + .all(|event| event.tags.identifier() != Some(id.to_string().as_str())), + "list must omit deleted definition" + ); + let trigger = buzz_sdk::build_workflow_trigger(id) + .unwrap() + .sign_with_keys(&f.owner) + .unwrap(); + let response = f.submit(&f.owner, &trigger).await; + assert_ne!( + response["accepted"], true, + "deleted workflow must not execute: {response}" + ); + assert!( + response.to_string().contains("workflow not found"), + "{response}" + ); + f.accepted(&deletion).await; + } +} + +#[tokio::test] +#[ignore = "requires isolated relay and two test accounts"] +async fn stale_deletion_preserves_newer_definition_and_runtime() { + let f = Fixture::new(); + let id = Uuid::new_v4(); + let now = Timestamp::now(); + f.create(id, now).await; + f.accepted(&f.deletion(id, Timestamp::from(now.as_secs() - 1))) + .await; + assert_eq!(f.definition(id).await.len(), 1); + let trigger = buzz_sdk::build_workflow_trigger(id) + .unwrap() + .sign_with_keys(&f.owner) + .unwrap(); + f.accepted(&trigger).await; + f.accepted(&f.deletion(id, now)).await; + assert!(f.definition(id).await.is_empty()); +}