From acff9cc2c5dc3ffd5d3136739aeed0ac009c6fe0 Mon Sep 17 00:00:00 2001 From: Michael Morale Date: Sat, 1 Aug 2026 21:45:28 -0500 Subject: [PATCH 1/2] Fix workflow agent wake and deletion Implement relay-authored workflow DMs and atomically remove workflow definitions with their executable records. Co-authored-by: Michael Morale Signed-off-by: Michael Morale --- crates/buzz-db/src/lib.rs | 21 ++ crates/buzz-db/src/workflow.rs | 267 ++++++++++++++++- .../buzz-relay/src/handlers/side_effects.rs | 93 ++++-- crates/buzz-relay/src/workflow_sink.rs | 277 +++++++++++++++++- crates/buzz-workflow/src/action_sink.rs | 16 + crates/buzz-workflow/src/executor.rs | 43 ++- 6 files changed, 690 insertions(+), 27 deletions(-) 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) { diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index 97c31c2561..4e1f9f153b 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/crates/buzz-relay/src/workflow_sink.rs @@ -8,7 +8,7 @@ use std::future::Future; use std::pin::Pin; use std::sync::{Arc, Weak}; -use buzz_core::kind::KIND_STREAM_MESSAGE; +use buzz_core::kind::{KIND_MEMBER_ADDED_NOTIFICATION, KIND_STREAM_MESSAGE}; use buzz_core::tenant::CommunityId; use buzz_workflow::action_sink::{ActionSink, ActionSinkError}; use chrono::Utc; @@ -17,6 +17,9 @@ use tracing::info; use uuid::Uuid; use crate::handlers::event::dispatch_persistent_event; +use crate::handlers::side_effects::{ + emit_group_discovery_events, emit_membership_notification, publish_dm_visibility_snapshot, +}; use crate::state::AppState; /// Resolves `@Name` mentions in workflow message text to the pubkeys of the @@ -362,6 +365,191 @@ impl ActionSink for RelayActionSink { Ok(event_id_hex) }) } + + fn send_dm( + &self, + community_id: CommunityId, + recipient_pubkey: &str, + text: &str, + author_pubkey: &str, + ) -> Pin> + Send + '_>> { + let recipient_pubkey = recipient_pubkey.to_owned(); + let text = text.to_owned(); + let author_pubkey = author_pubkey.to_owned(); + + Box::pin(async move { + let state = self + .state + .upgrade() + .ok_or_else(|| ActionSinkError::Database("relay is shutting down".into()))?; + + let host = state + .db + .lookup_community_host(community_id) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))? + .ok_or_else(|| { + ActionSinkError::Database(format!( + "workflow run community {community_id} is not mapped to a host" + )) + })?; + let tenant = buzz_core::tenant::TenantContext::resolved(community_id, host); + + if text.trim().is_empty() { + return Err(ActionSinkError::EmptyContent); + } + + let recipient = nostr::PublicKey::from_hex(&recipient_pubkey).map_err(|e| { + ActionSinkError::InvalidInput(format!("invalid recipient pubkey: {e}")) + })?; + let author = nostr::PublicKey::from_hex(&author_pubkey).map_err(|e| { + ActionSinkError::InvalidInput(format!("invalid author pubkey: {e}")) + })?; + let recipient_bytes = recipient.to_bytes(); + let recipient_hex = recipient.to_hex(); + let author_hex = author.to_hex(); + let relay_bytes = state.relay_keypair.public_key().to_bytes(); + + if recipient_bytes == relay_bytes { + return Err(ActionSinkError::InvalidInput( + "workflow DM recipient cannot be the relay identity".into(), + )); + } + + if state + .db + .get_relay_member(tenant.community(), &recipient_hex) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))? + .is_none() + { + return Err(ActionSinkError::InvalidInput( + "workflow DM recipient is not a relay member".into(), + )); + } + + // Workflow automation DMs are relay↔recipient threads. This gives + // self-targeting workflows two distinct channel participants while + // retaining the workflow owner as explicit message attribution. + let (dm_channel, was_created) = state + .db + .open_dm( + tenant.community(), + &[recipient_bytes.as_slice()], + relay_bytes.as_slice(), + ) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))?; + let dm_channel_id = dm_channel.id; + + state + .db + .unhide_dm( + tenant.community(), + dm_channel_id, + recipient_bytes.as_slice(), + ) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))?; + + if was_created { + metrics::counter!( + "buzz_channels_created_total", + "community" => tenant.host().to_owned(), + "type" => "dm" + ) + .increment(1); + + for participant in [&relay_bytes[..], &recipient_bytes[..]] { + state.invalidate_membership(&tenant, dm_channel_id, participant); + emit_membership_notification( + &tenant, + &state, + dm_channel_id, + participant, + relay_bytes.as_slice(), + KIND_MEMBER_ADDED_NOTIFICATION, + ) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))?; + } + } + + // Emit discovery on every send so a retry heals a missed initial + // discovery event. Refresh visibility after the explicit unhide. + emit_group_discovery_events(&tenant, &state, dm_channel_id) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))?; + publish_dm_visibility_snapshot(&tenant, &state, recipient_bytes.as_slice()) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))?; + + let dm_channel_id_text = dm_channel_id.to_string(); + let tags = vec![ + Tag::parse(["h", &dm_channel_id_text]) + .map_err(|e| ActionSinkError::EventBuild(format!("h tag: {e}")))?, + Tag::parse(["p", &recipient_hex]) + .map_err(|e| ActionSinkError::EventBuild(format!("p tag: {e}")))?, + Tag::parse(["actor", &author_hex]) + .map_err(|e| ActionSinkError::EventBuild(format!("actor tag: {e}")))?, + Tag::parse(["buzz:workflow", "true"]) + .map_err(|e| ActionSinkError::EventBuild(format!("workflow tag: {e}")))?, + ]; + let event = EventBuilder::new(Kind::from(KIND_STREAM_MESSAGE as u16), &text) + .tags(tags) + .sign_with_keys(&state.relay_keypair) + .map_err(|e| ActionSinkError::EventBuild(format!("signing: {e}")))?; + let event_id_hex = event.id.to_hex(); + let event_id_bytes = event.id.as_bytes().to_vec(); + let event_created_at = + chrono::DateTime::from_timestamp(event.created_at.as_secs() as i64, 0) + .unwrap_or_else(Utc::now); + let thread_meta = Some(buzz_db::event::ThreadMetadataParams { + event_id: &event_id_bytes, + event_created_at, + channel_id: dm_channel_id, + parent_event_id: None, + parent_event_created_at: None, + root_event_id: None, + root_event_created_at: None, + depth: 0, + broadcast: false, + }); + + let (stored_event, was_inserted) = state + .db + .insert_event_with_thread_metadata( + tenant.community(), + &event, + Some(dm_channel_id), + thread_meta, + ) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))?; + + if was_inserted { + let _ = dispatch_persistent_event( + &tenant, + &state, + &stored_event, + KIND_STREAM_MESSAGE, + &author_hex, + None, + ) + .await; + } + + info!( + event_id = %event_id_hex, + channel_id = %dm_channel_id, + recipient = %recipient_hex, + author = %author_hex, + "Workflow SendDm: delivered relay-authored DM" + ); + + Ok(event_id_hex) + }) + } } #[cfg(test)] @@ -708,4 +896,91 @@ mod integration_tests { "mentioned member {agent_hex} must be p-tagged so it wakes; got {p_tag_targets:?}" ); } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn workflow_send_dm_creates_relay_dm_and_p_tags_recipient() { + let state = test_state().await; + let author = nostr::Keys::generate(); + let author_hex = author.public_key().to_hex(); + let recipient = nostr::Keys::generate(); + let recipient_hex = recipient.public_key().to_hex(); + let recipient_bytes = recipient.public_key().to_bytes(); + let relay_bytes = state.relay_keypair.public_key().to_bytes(); + + let host = format!("wf-dm-{}.example", uuid::Uuid::new_v4().simple()); + let community = match state + .db + .create_community_with_owner(&host, &author_hex) + .await + .expect("create community") + { + CreateCommunityWithOwnerResult::Created(rec) => rec.id, + other => panic!("expected fresh community, got {other:?}"), + }; + state + .db + .ensure_user(community, recipient_bytes.as_slice()) + .await + .expect("ensure recipient user"); + state + .db + .add_relay_member(community, &recipient_hex, "member", Some(&author_hex)) + .await + .expect("add recipient relay membership"); + + let sink = RelayActionSink::new(&state); + let event_id_hex = sink + .send_dm( + community, + &recipient_hex, + "run the read-only repository monitor", + &author_hex, + ) + .await + .expect("send_dm"); + + let dms = state + .db + .list_dms_for_user(community, recipient_bytes.as_slice(), 100, None) + .await + .expect("list recipient DMs"); + assert_eq!(dms.len(), 1, "expected one relay automation DM"); + let dm = &dms[0]; + let participant_pubkeys: Vec<&[u8]> = dm + .participants + .iter() + .map(|participant| participant.pubkey.as_slice()) + .collect(); + assert!(participant_pubkeys.contains(&recipient_bytes.as_slice())); + assert!(participant_pubkeys.contains(&relay_bytes.as_slice())); + + let id_bytes = nostr::EventId::from_hex(&event_id_hex) + .expect("event id") + .as_bytes() + .to_vec(); + let stored = state + .db + .get_event_by_id(community, &id_bytes) + .await + .expect("query event") + .expect("event persisted"); + assert_eq!(stored.event.pubkey, state.relay_keypair.public_key()); + assert_eq!(stored.channel_id, Some(dm.channel_id)); + + let has_tag = |kind: &str, value: &str| { + stored.event.tags.iter().any(|tag| { + let parts = tag.as_slice(); + parts.first().map(String::as_str) == Some(kind) + && parts.get(1).map(String::as_str) == Some(value) + }) + }; + assert!(has_tag("h", &dm.channel_id.to_string())); + assert!( + has_tag("p", &recipient_hex), + "recipient must be p-tagged so ACP mention-gated wake can fire" + ); + assert!(has_tag("actor", &author_hex)); + assert!(has_tag("buzz:workflow", "true")); + } } diff --git a/crates/buzz-workflow/src/action_sink.rs b/crates/buzz-workflow/src/action_sink.rs index 0c6002e74e..a9b91eaf1a 100644 --- a/crates/buzz-workflow/src/action_sink.rs +++ b/crates/buzz-workflow/src/action_sink.rs @@ -66,4 +66,20 @@ pub trait ActionSink: Send + Sync { text: &str, author_pubkey: &str, ) -> Pin> + Send + '_>>; + + /// Send a direct message to one relay member on behalf of a workflow owner. + /// + /// The relay creates or reuses its two-party automation DM with + /// `recipient_pubkey`, signs the message with the relay key, and carries the + /// workflow owner in an `actor` attribution tag. The recipient is always + /// explicitly `p`-tagged so mention-gated agent runtimes can wake reliably. + /// + /// Returns the event ID hex string on success. + fn send_dm( + &self, + community_id: CommunityId, + recipient_pubkey: &str, + text: &str, + author_pubkey: &str, + ) -> Pin> + Send + '_>>; } diff --git a/crates/buzz-workflow/src/executor.rs b/crates/buzz-workflow/src/executor.rs index e30541377e..3ecc7133a0 100644 --- a/crates/buzz-workflow/src/executor.rs +++ b/crates/buzz-workflow/src/executor.rs @@ -577,10 +577,45 @@ pub async fn dispatch_action( }))) } - SendDm { to, text: _ } => { - warn!(run_id = %run_id, step = step_id, "SendDm not yet implemented (to={to})"); - // TODO (WF-07): emit DM event. - Err(WorkflowError::NotImplemented("SendDm".into())) + SendDm { to, text } => { + let wf_run = engine + .db + .get_workflow_run(community_id, run_id) + .await + .map_err(|e| { + WorkflowError::WebhookError(format!( + "SendDm: failed to load workflow run {run_id}: {e}" + )) + })?; + let workflow = engine + .db + .get_workflow(community_id, wf_run.workflow_id) + .await + .map_err(|e| { + WorkflowError::WebhookError(format!( + "SendDm: failed to load workflow {}: {e}", + wf_run.workflow_id + )) + })?; + let owner_pubkey_hex = hex::encode(&workflow.owner_pubkey); + + info!( + run_id = %run_id, + step = step_id, + recipient = %to, + "SendDm: delivering workflow DM" + ); + + let event_id = engine + .action_sink()? + .send_dm(community_id, to, text, &owner_pubkey_hex) + .await + .map_err(WorkflowError::from)?; + + Ok(StepResult::Completed(serde_json::json!({ + "sent": true, + "event_id": event_id, + }))) } SetChannelTopic { topic: _ } => { From 327628f2be472fd93e692be5b46cb39ac27771ee Mon Sep 17 00:00:00 2001 From: Michael Morale Date: Sat, 1 Aug 2026 22:00:04 -0500 Subject: [PATCH 2/2] chore(workflow): defer send_dm to existing proposal Narrow the upstream pull request to the unique atomic workflow deletion repair; send_dm remains under discussion in PR #2614 and issue #2628. Co-authored-by: Michael Morale Signed-off-by: Michael Morale --- crates/buzz-relay/src/workflow_sink.rs | 277 +----------------------- crates/buzz-workflow/src/action_sink.rs | 16 -- crates/buzz-workflow/src/executor.rs | 43 +--- 3 files changed, 5 insertions(+), 331 deletions(-) diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index 4e1f9f153b..97c31c2561 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/crates/buzz-relay/src/workflow_sink.rs @@ -8,7 +8,7 @@ use std::future::Future; use std::pin::Pin; use std::sync::{Arc, Weak}; -use buzz_core::kind::{KIND_MEMBER_ADDED_NOTIFICATION, KIND_STREAM_MESSAGE}; +use buzz_core::kind::KIND_STREAM_MESSAGE; use buzz_core::tenant::CommunityId; use buzz_workflow::action_sink::{ActionSink, ActionSinkError}; use chrono::Utc; @@ -17,9 +17,6 @@ use tracing::info; use uuid::Uuid; use crate::handlers::event::dispatch_persistent_event; -use crate::handlers::side_effects::{ - emit_group_discovery_events, emit_membership_notification, publish_dm_visibility_snapshot, -}; use crate::state::AppState; /// Resolves `@Name` mentions in workflow message text to the pubkeys of the @@ -365,191 +362,6 @@ impl ActionSink for RelayActionSink { Ok(event_id_hex) }) } - - fn send_dm( - &self, - community_id: CommunityId, - recipient_pubkey: &str, - text: &str, - author_pubkey: &str, - ) -> Pin> + Send + '_>> { - let recipient_pubkey = recipient_pubkey.to_owned(); - let text = text.to_owned(); - let author_pubkey = author_pubkey.to_owned(); - - Box::pin(async move { - let state = self - .state - .upgrade() - .ok_or_else(|| ActionSinkError::Database("relay is shutting down".into()))?; - - let host = state - .db - .lookup_community_host(community_id) - .await - .map_err(|e| ActionSinkError::Database(e.to_string()))? - .ok_or_else(|| { - ActionSinkError::Database(format!( - "workflow run community {community_id} is not mapped to a host" - )) - })?; - let tenant = buzz_core::tenant::TenantContext::resolved(community_id, host); - - if text.trim().is_empty() { - return Err(ActionSinkError::EmptyContent); - } - - let recipient = nostr::PublicKey::from_hex(&recipient_pubkey).map_err(|e| { - ActionSinkError::InvalidInput(format!("invalid recipient pubkey: {e}")) - })?; - let author = nostr::PublicKey::from_hex(&author_pubkey).map_err(|e| { - ActionSinkError::InvalidInput(format!("invalid author pubkey: {e}")) - })?; - let recipient_bytes = recipient.to_bytes(); - let recipient_hex = recipient.to_hex(); - let author_hex = author.to_hex(); - let relay_bytes = state.relay_keypair.public_key().to_bytes(); - - if recipient_bytes == relay_bytes { - return Err(ActionSinkError::InvalidInput( - "workflow DM recipient cannot be the relay identity".into(), - )); - } - - if state - .db - .get_relay_member(tenant.community(), &recipient_hex) - .await - .map_err(|e| ActionSinkError::Database(e.to_string()))? - .is_none() - { - return Err(ActionSinkError::InvalidInput( - "workflow DM recipient is not a relay member".into(), - )); - } - - // Workflow automation DMs are relay↔recipient threads. This gives - // self-targeting workflows two distinct channel participants while - // retaining the workflow owner as explicit message attribution. - let (dm_channel, was_created) = state - .db - .open_dm( - tenant.community(), - &[recipient_bytes.as_slice()], - relay_bytes.as_slice(), - ) - .await - .map_err(|e| ActionSinkError::Database(e.to_string()))?; - let dm_channel_id = dm_channel.id; - - state - .db - .unhide_dm( - tenant.community(), - dm_channel_id, - recipient_bytes.as_slice(), - ) - .await - .map_err(|e| ActionSinkError::Database(e.to_string()))?; - - if was_created { - metrics::counter!( - "buzz_channels_created_total", - "community" => tenant.host().to_owned(), - "type" => "dm" - ) - .increment(1); - - for participant in [&relay_bytes[..], &recipient_bytes[..]] { - state.invalidate_membership(&tenant, dm_channel_id, participant); - emit_membership_notification( - &tenant, - &state, - dm_channel_id, - participant, - relay_bytes.as_slice(), - KIND_MEMBER_ADDED_NOTIFICATION, - ) - .await - .map_err(|e| ActionSinkError::Database(e.to_string()))?; - } - } - - // Emit discovery on every send so a retry heals a missed initial - // discovery event. Refresh visibility after the explicit unhide. - emit_group_discovery_events(&tenant, &state, dm_channel_id) - .await - .map_err(|e| ActionSinkError::Database(e.to_string()))?; - publish_dm_visibility_snapshot(&tenant, &state, recipient_bytes.as_slice()) - .await - .map_err(|e| ActionSinkError::Database(e.to_string()))?; - - let dm_channel_id_text = dm_channel_id.to_string(); - let tags = vec![ - Tag::parse(["h", &dm_channel_id_text]) - .map_err(|e| ActionSinkError::EventBuild(format!("h tag: {e}")))?, - Tag::parse(["p", &recipient_hex]) - .map_err(|e| ActionSinkError::EventBuild(format!("p tag: {e}")))?, - Tag::parse(["actor", &author_hex]) - .map_err(|e| ActionSinkError::EventBuild(format!("actor tag: {e}")))?, - Tag::parse(["buzz:workflow", "true"]) - .map_err(|e| ActionSinkError::EventBuild(format!("workflow tag: {e}")))?, - ]; - let event = EventBuilder::new(Kind::from(KIND_STREAM_MESSAGE as u16), &text) - .tags(tags) - .sign_with_keys(&state.relay_keypair) - .map_err(|e| ActionSinkError::EventBuild(format!("signing: {e}")))?; - let event_id_hex = event.id.to_hex(); - let event_id_bytes = event.id.as_bytes().to_vec(); - let event_created_at = - chrono::DateTime::from_timestamp(event.created_at.as_secs() as i64, 0) - .unwrap_or_else(Utc::now); - let thread_meta = Some(buzz_db::event::ThreadMetadataParams { - event_id: &event_id_bytes, - event_created_at, - channel_id: dm_channel_id, - parent_event_id: None, - parent_event_created_at: None, - root_event_id: None, - root_event_created_at: None, - depth: 0, - broadcast: false, - }); - - let (stored_event, was_inserted) = state - .db - .insert_event_with_thread_metadata( - tenant.community(), - &event, - Some(dm_channel_id), - thread_meta, - ) - .await - .map_err(|e| ActionSinkError::Database(e.to_string()))?; - - if was_inserted { - let _ = dispatch_persistent_event( - &tenant, - &state, - &stored_event, - KIND_STREAM_MESSAGE, - &author_hex, - None, - ) - .await; - } - - info!( - event_id = %event_id_hex, - channel_id = %dm_channel_id, - recipient = %recipient_hex, - author = %author_hex, - "Workflow SendDm: delivered relay-authored DM" - ); - - Ok(event_id_hex) - }) - } } #[cfg(test)] @@ -896,91 +708,4 @@ mod integration_tests { "mentioned member {agent_hex} must be p-tagged so it wakes; got {p_tag_targets:?}" ); } - - #[tokio::test] - #[ignore = "requires Postgres"] - async fn workflow_send_dm_creates_relay_dm_and_p_tags_recipient() { - let state = test_state().await; - let author = nostr::Keys::generate(); - let author_hex = author.public_key().to_hex(); - let recipient = nostr::Keys::generate(); - let recipient_hex = recipient.public_key().to_hex(); - let recipient_bytes = recipient.public_key().to_bytes(); - let relay_bytes = state.relay_keypair.public_key().to_bytes(); - - let host = format!("wf-dm-{}.example", uuid::Uuid::new_v4().simple()); - let community = match state - .db - .create_community_with_owner(&host, &author_hex) - .await - .expect("create community") - { - CreateCommunityWithOwnerResult::Created(rec) => rec.id, - other => panic!("expected fresh community, got {other:?}"), - }; - state - .db - .ensure_user(community, recipient_bytes.as_slice()) - .await - .expect("ensure recipient user"); - state - .db - .add_relay_member(community, &recipient_hex, "member", Some(&author_hex)) - .await - .expect("add recipient relay membership"); - - let sink = RelayActionSink::new(&state); - let event_id_hex = sink - .send_dm( - community, - &recipient_hex, - "run the read-only repository monitor", - &author_hex, - ) - .await - .expect("send_dm"); - - let dms = state - .db - .list_dms_for_user(community, recipient_bytes.as_slice(), 100, None) - .await - .expect("list recipient DMs"); - assert_eq!(dms.len(), 1, "expected one relay automation DM"); - let dm = &dms[0]; - let participant_pubkeys: Vec<&[u8]> = dm - .participants - .iter() - .map(|participant| participant.pubkey.as_slice()) - .collect(); - assert!(participant_pubkeys.contains(&recipient_bytes.as_slice())); - assert!(participant_pubkeys.contains(&relay_bytes.as_slice())); - - let id_bytes = nostr::EventId::from_hex(&event_id_hex) - .expect("event id") - .as_bytes() - .to_vec(); - let stored = state - .db - .get_event_by_id(community, &id_bytes) - .await - .expect("query event") - .expect("event persisted"); - assert_eq!(stored.event.pubkey, state.relay_keypair.public_key()); - assert_eq!(stored.channel_id, Some(dm.channel_id)); - - let has_tag = |kind: &str, value: &str| { - stored.event.tags.iter().any(|tag| { - let parts = tag.as_slice(); - parts.first().map(String::as_str) == Some(kind) - && parts.get(1).map(String::as_str) == Some(value) - }) - }; - assert!(has_tag("h", &dm.channel_id.to_string())); - assert!( - has_tag("p", &recipient_hex), - "recipient must be p-tagged so ACP mention-gated wake can fire" - ); - assert!(has_tag("actor", &author_hex)); - assert!(has_tag("buzz:workflow", "true")); - } } diff --git a/crates/buzz-workflow/src/action_sink.rs b/crates/buzz-workflow/src/action_sink.rs index a9b91eaf1a..0c6002e74e 100644 --- a/crates/buzz-workflow/src/action_sink.rs +++ b/crates/buzz-workflow/src/action_sink.rs @@ -66,20 +66,4 @@ pub trait ActionSink: Send + Sync { text: &str, author_pubkey: &str, ) -> Pin> + Send + '_>>; - - /// Send a direct message to one relay member on behalf of a workflow owner. - /// - /// The relay creates or reuses its two-party automation DM with - /// `recipient_pubkey`, signs the message with the relay key, and carries the - /// workflow owner in an `actor` attribution tag. The recipient is always - /// explicitly `p`-tagged so mention-gated agent runtimes can wake reliably. - /// - /// Returns the event ID hex string on success. - fn send_dm( - &self, - community_id: CommunityId, - recipient_pubkey: &str, - text: &str, - author_pubkey: &str, - ) -> Pin> + Send + '_>>; } diff --git a/crates/buzz-workflow/src/executor.rs b/crates/buzz-workflow/src/executor.rs index 3ecc7133a0..e30541377e 100644 --- a/crates/buzz-workflow/src/executor.rs +++ b/crates/buzz-workflow/src/executor.rs @@ -577,45 +577,10 @@ pub async fn dispatch_action( }))) } - SendDm { to, text } => { - let wf_run = engine - .db - .get_workflow_run(community_id, run_id) - .await - .map_err(|e| { - WorkflowError::WebhookError(format!( - "SendDm: failed to load workflow run {run_id}: {e}" - )) - })?; - let workflow = engine - .db - .get_workflow(community_id, wf_run.workflow_id) - .await - .map_err(|e| { - WorkflowError::WebhookError(format!( - "SendDm: failed to load workflow {}: {e}", - wf_run.workflow_id - )) - })?; - let owner_pubkey_hex = hex::encode(&workflow.owner_pubkey); - - info!( - run_id = %run_id, - step = step_id, - recipient = %to, - "SendDm: delivering workflow DM" - ); - - let event_id = engine - .action_sink()? - .send_dm(community_id, to, text, &owner_pubkey_hex) - .await - .map_err(WorkflowError::from)?; - - Ok(StepResult::Completed(serde_json::json!({ - "sent": true, - "event_id": event_id, - }))) + SendDm { to, text: _ } => { + warn!(run_id = %run_id, step = step_id, "SendDm not yet implemented (to={to})"); + // TODO (WF-07): emit DM event. + Err(WorkflowError::NotImplemented("SendDm".into())) } SetChannelTopic { topic: _ } => {