From 47424548e0e6d0549380d42b79215c100b82783e Mon Sep 17 00:00:00 2001 From: Sean Gearin Date: Thu, 23 Jul 2026 16:20:33 -0400 Subject: [PATCH] feat(workflow): implement the send_dm action (WF-07) Signed-off-by: Sean Gearin --- crates/buzz-relay/src/workflow_sink.rs | 324 +++++++++++++++++++++++- crates/buzz-workflow/src/action_sink.rs | 24 ++ crates/buzz-workflow/src/executor.rs | 146 ++++++++++- 3 files changed, 488 insertions(+), 6 deletions(-) diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index 97c31c25611..52d308d6b1e 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/crates/buzz-relay/src/workflow_sink.rs @@ -8,15 +8,19 @@ 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; use nostr::{EventBuilder, Kind, Tag}; -use tracing::info; +use tracing::{info, warn}; use uuid::Uuid; use crate::handlers::event::dispatch_persistent_event; +use crate::handlers::side_effects::{ + emit_group_discovery_events, emit_membership_notification, emit_system_message, + publish_dm_visibility_snapshot, +}; use crate::state::AppState; /// Resolves `@Name` mentions in workflow message text to the pubkeys of the @@ -362,6 +366,214 @@ 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 { + // 0. Upgrade weak reference — fails only during shutdown. + let state = self + .state + .upgrade() + .ok_or_else(|| ActionSinkError::Database("relay is shutting down".into()))?; + + // Same tenancy rule as send_message: the run carries its owning + // community; never re-derive it from the deployment default. + 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); + + // 1. Validate content is not empty/whitespace-only + if text.trim().is_empty() { + return Err(ActionSinkError::EmptyContent); + } + + // 2. Parse and validate pubkeys + let author_pubkey = nostr::PublicKey::from_hex(&author_pubkey).map_err(|e| { + ActionSinkError::InvalidInput(format!("invalid author pubkey: {e}")) + })?; + let recipient_pubkey = nostr::PublicKey::from_hex(&recipient_pubkey).map_err(|e| { + ActionSinkError::InvalidInput(format!("invalid recipient pubkey: {e}")) + })?; + if recipient_pubkey == author_pubkey { + return Err(ActionSinkError::InvalidInput( + "SendDm: recipient is the workflow owner".into(), + )); + } + let author_bytes = author_pubkey.to_bytes().to_vec(); + let recipient_bytes = recipient_pubkey.to_bytes().to_vec(); + let author_pubkey_hex = author_pubkey.to_hex(); + let recipient_pubkey_hex = recipient_pubkey.to_hex(); + + // 3. Find or create the owner↔recipient DM channel — the same + // idempotent open path as the kind:41010 command handler + // (`handle_dm_open`), whose post-open side effects are mirrored + // below. + let (channel, was_created) = state + .db + .open_dm( + tenant.community(), + &[recipient_bytes.as_slice()], + &author_bytes, + ) + .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 pk in [&author_bytes, &recipient_bytes] { + state.invalidate_membership(&tenant, channel.id, pk); + } + + if let Err(e) = emit_system_message( + &tenant, + &state, + channel.id, + serde_json::json!({ + "type": "dm_created", + "actor": author_pubkey_hex, + "participants": [author_pubkey_hex, recipient_pubkey_hex], + }), + ) + .await + { + warn!("workflow DM open: system message failed: {e}"); + } + + if let Err(e) = emit_group_discovery_events(&tenant, &state, channel.id).await { + warn!(channel = %channel.id, "workflow DM open: discovery emission failed: {e}"); + } + + for participant in [&author_bytes, &recipient_bytes] { + if let Err(e) = emit_membership_notification( + &tenant, + &state, + channel.id, + participant, + &author_bytes, + KIND_MEMBER_ADDED_NOTIFICATION, + ) + .await + { + warn!("workflow DM open: membership notification failed: {e}"); + } + } + } else { + // Re-open cleared the owner's hidden_at; refresh their NIP-DV + // snapshot so the DM reappears in the sidebar. + if let Err(e) = publish_dm_visibility_snapshot(&tenant, &state, &author_bytes).await + { + warn!("workflow DM re-open: visibility snapshot failed: {e}"); + } + } + + // 4. Build kind:9 Nostr event + // - Signed by relay keypair (event.pubkey = relay pubkey) + // - `p` tag attributes the message to the workflow owner + // - `p` tag addresses the recipient (agent wake and push + // delivery are `p`-tag gated) + // - `h` tag scopes to the DM channel + // - `buzz:workflow` tag prevents recursive workflow triggering + let channel_id_canonical = channel.id.to_string(); + let tags = vec![ + Tag::parse(["p", &author_pubkey_hex]) + .map_err(|e| ActionSinkError::EventBuild(format!("p tag: {e}")))?, + Tag::parse(["p", &recipient_pubkey_hex]) + .map_err(|e| ActionSinkError::EventBuild(format!("recipient p tag: {e}")))?, + Tag::parse(["h", &channel_id_canonical]) + .map_err(|e| ActionSinkError::EventBuild(format!("h tag: {e}")))?, + Tag::parse(["buzz:workflow", "true"]) + .map_err(|e| ActionSinkError::EventBuild(format!("workflow tag: {e}")))?, + ]; + + let kind = Kind::from(KIND_STREAM_MESSAGE as u16); + let event = EventBuilder::new(kind, &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 kind_u32 = KIND_STREAM_MESSAGE; + + let event_created_at = { + let ts = event.created_at.as_secs() as i64; + chrono::DateTime::from_timestamp(ts, 0).unwrap_or_else(Utc::now) + }; + + // Deliberately no message body in the log — DMs are private. + info!( + event_id = %event_id_hex, + channel_id = %channel_id_canonical, + author = %author_pubkey, + recipient = %recipient_pubkey, + "Workflow SendDm: posting kind {kind_u32} event" + ); + + // 5. Persist event with thread metadata (matches send_message). + // Workflow DMs are always top-level: depth=0, no parent/root. + let thread_meta = Some(buzz_db::event::ThreadMetadataParams { + event_id: &event_id_bytes, + event_created_at, + channel_id: 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(channel.id), + thread_meta, + ) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))?; + + // 6. Post-persist side effects (fan-out, search, audit) + // Only if actually inserted (idempotency guard). + if was_inserted { + let _ = dispatch_persistent_event( + &tenant, + &state, + &stored_event, + kind_u32, + &author_pubkey_hex, + None, + ) + .await; + } + + Ok(event_id_hex) + }) + } } #[cfg(test)] @@ -708,4 +920,112 @@ 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_opens_channel_and_p_tags_recipient() { + let state = test_state().await; + + let owner = nostr::Keys::generate(); + let owner_hex = owner.public_key().to_hex(); + let recipient = nostr::Keys::generate(); + let recipient_hex = recipient.public_key().to_hex(); + + let host = format!("wf-dm-{}.example", uuid::Uuid::new_v4().simple()); + let community = match state + .db + .create_community_with_owner(&host, &owner_hex) + .await + .expect("create community") + { + CreateCommunityWithOwnerResult::Created(rec) => rec.id, + other => panic!("expected fresh community, got {other:?}"), + }; + + let sink = RelayActionSink::new(&state); + let event_id_hex = sink + .send_dm(community, &recipient_hex, "workflow says hi", &owner_hex) + .await + .expect("send_dm"); + + // The emitted event must land in a channel_type='dm' channel whose + // participant set is exactly {owner, recipient}. + 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"); + + let h_tag_channel = stored + .event + .tags + .iter() + .find(|t| t.as_slice().first().map(|s| s.as_str()) == Some("h")) + .and_then(|t| t.as_slice().get(1).cloned()) + .expect("h tag present"); + let channel_id = uuid::Uuid::parse_str(&h_tag_channel).expect("h tag is a channel UUID"); + let channel = state + .db + .get_channel(community, channel_id) + .await + .expect("dm channel exists"); + assert_eq!(channel.channel_type, "dm"); + assert_eq!(channel.visibility, "private"); + + let p_tag_targets: Vec<&str> = stored + .event + .tags + .iter() + .filter(|t| t.as_slice().first().map(|s| s.as_str()) == Some("p")) + .filter_map(|t| t.as_slice().get(1).map(|s| s.as_str())) + .collect(); + assert!( + p_tag_targets.contains(&owner_hex.as_str()), + "owner must be attributed via p tag; got {p_tag_targets:?}" + ); + assert!( + p_tag_targets.contains(&recipient_hex.as_str()), + "recipient must be p-tagged (wake/push are p-tag gated); got {p_tag_targets:?}" + ); + + // A second DM to the same recipient must reuse the same channel + // (open_dm is idempotent on the participant set). + let second_event_id = sink + .send_dm(community, &recipient_hex, "again", &owner_hex) + .await + .expect("second send_dm"); + let second_id_bytes = nostr::EventId::from_hex(&second_event_id) + .expect("event id") + .as_bytes() + .to_vec(); + let second_stored = state + .db + .get_event_by_id(community, &second_id_bytes) + .await + .expect("query second event") + .expect("second event persisted"); + let second_channel = second_stored + .event + .tags + .iter() + .find(|t| t.as_slice().first().map(|s| s.as_str()) == Some("h")) + .and_then(|t| t.as_slice().get(1).cloned()) + .expect("h tag present"); + assert_eq!( + second_channel, h_tag_channel, + "repeat DMs must reuse the same DM channel" + ); + + // Self-DM is rejected — the participant set would collapse to one. + let err = sink + .send_dm(community, &owner_hex, "hi me", &owner_hex) + .await + .expect_err("self-DM must be rejected"); + assert!(matches!(err, ActionSinkError::InvalidInput(_))); + } } diff --git a/crates/buzz-workflow/src/action_sink.rs b/crates/buzz-workflow/src/action_sink.rs index 0c6002e74eb..18586024375 100644 --- a/crates/buzz-workflow/src/action_sink.rs +++ b/crates/buzz-workflow/src/action_sink.rs @@ -66,4 +66,28 @@ pub trait ActionSink: Send + Sync { text: &str, author_pubkey: &str, ) -> Pin> + Send + '_>>; + + /// Send a direct message from the workflow owner to a recipient. + /// + /// - `community_id`: the server-resolved community that owns the workflow + /// run driving this side effect (same tenancy rules as [`send_message`]). + /// - `recipient_pubkey`: hex-encoded pubkey of the DM recipient + /// - `text`: message body (must not be empty/whitespace-only) + /// - `author_pubkey`: hex-encoded pubkey of the workflow owner + /// + /// The relay finds or creates the owner↔recipient DM channel (the same + /// idempotent path as a kind:41010 DM-open command) and emits a kind:9 + /// message into it, signed by the relay keypair with `p` tags attributing + /// the owner and addressing the recipient. + /// + /// Returns the event ID hex string on success. + /// + /// [`send_message`]: ActionSink::send_message + 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 a029b44622c..43b7ceb58bd 100644 --- a/crates/buzz-workflow/src/executor.rs +++ b/crates/buzz-workflow/src/executor.rs @@ -451,6 +451,26 @@ pub fn resolve_step_templates( } } +/// Normalize the `SendDm` `to` field into a lowercase hex pubkey. +/// +/// Accepts a 64-char hex pubkey or a bech32 `npub` (the `| npub` template +/// filter produces the latter), mirroring how `resolve_send_message_channel` +/// canonicalizes its destination before the sink call. Rejects empty or +/// malformed values — including unresolved `{{...}}` placeholders, which +/// `resolve_template` leaves as literal text. +fn resolve_dm_recipient(to: &str) -> Result { + let to = to.trim(); + if to.is_empty() { + return Err(WorkflowError::InvalidDefinition( + "SendDm: recipient 'to' is empty".into(), + )); + } + let pubkey = nostr::PublicKey::parse(to).map_err(|e| { + WorkflowError::InvalidDefinition(format!("SendDm: invalid recipient pubkey '{to}': {e}")) + })?; + Ok(pubkey.to_hex()) +} + /// Result of dispatching a single step action. #[derive(Debug)] pub enum StepResult { @@ -577,10 +597,51 @@ 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 } => { + // Same community-scoped run/workflow lookup as SendMessage: the + // owner pubkey defines the DM participant set, so loading the + // wrong community's row would DM on behalf of the wrong owner. + 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 recipient = resolve_dm_recipient(to)?; + let owner_pubkey_hex = hex::encode(&workflow.owner_pubkey); + + // Unlike SendMessage, the body is deliberately not logged — DMs + // are private and relay logs are not. + info!( + run_id = %run_id, + step = step_id, + to = %recipient, + "SendDm → {recipient}" + ); + + let event_id = engine + .action_sink()? + .send_dm(community_id, &recipient, text, &owner_pubkey_hex) + .await + .map_err(WorkflowError::from)?; + + Ok(StepResult::Completed(serde_json::json!({ + "sent": true, + "event_id": event_id, + }))) } SetChannelTopic { topic: _ } => { @@ -1831,4 +1892,81 @@ mod tests { .expect("override should be accepted"); assert_eq!(resolved, override_channel_id.to_string()); } + + const RECIPIENT_HEX: &str = "e17e5abf7b1dbd363f0ed6fbda2455609727b2555428dea251388c542cd2f03f"; + const RECIPIENT_NPUB: &str = "npub1u9l940mmrk7nv0cw6maa5fz4vztj0vj42s5dagj38zx9gtxj7qls94fpux"; + + #[test] + fn send_dm_recipient_accepts_hex_pubkey() { + let resolved = resolve_dm_recipient(RECIPIENT_HEX).expect("hex pubkey should be accepted"); + assert_eq!(resolved, RECIPIENT_HEX); + } + + #[test] + fn send_dm_recipient_accepts_npub_and_normalizes_to_hex() { + // The `| npub` template filter can hand us a bech32 pubkey; the sink + // expects hex, so the resolver must normalize. + let resolved = resolve_dm_recipient(RECIPIENT_NPUB).expect("npub should be accepted"); + assert_eq!(resolved, RECIPIENT_HEX); + } + + #[test] + fn send_dm_recipient_normalizes_uppercase_hex() { + let resolved = resolve_dm_recipient(&RECIPIENT_HEX.to_uppercase()) + .expect("uppercase hex should be accepted"); + assert_eq!(resolved, RECIPIENT_HEX); + } + + #[test] + fn send_dm_recipient_trims_whitespace() { + let padded = format!(" {RECIPIENT_HEX}\n"); + let resolved = resolve_dm_recipient(&padded).expect("padded hex should be accepted"); + assert_eq!(resolved, RECIPIENT_HEX); + } + + #[test] + fn send_dm_recipient_rejects_empty_and_whitespace() { + for input in ["", " ", "\t\n"] { + let err = resolve_dm_recipient(input).unwrap_err(); + assert!(matches!(err, WorkflowError::InvalidDefinition(_))); + } + } + + #[test] + fn send_dm_recipient_rejects_malformed_values() { + // Includes an unresolved template placeholder: `resolve_template` + // leaves unknown `{{keys}}` as literal text, which must not silently + // become a DM destination. + for input in ["not-a-pubkey", "abc123", "{{trigger.author}}"] { + let err = resolve_dm_recipient(input).unwrap_err(); + assert!( + matches!(err, WorkflowError::InvalidDefinition(_)), + "expected InvalidDefinition for {input:?}" + ); + } + } + + #[test] + fn resolve_step_templates_resolves_send_dm_fields() { + let mut ctx = make_trigger(); + ctx.author = RECIPIENT_HEX.to_owned(); + let step: Step = serde_yaml::from_str( + r#" +id: notify +action: send_dm +to: "{{trigger.author}}" +text: "You said: {{trigger.text}}" +"#, + ) + .expect("step should parse"); + let resolved = + resolve_step_templates(&step, &ctx, &HashMap::new()).expect("templates should resolve"); + match resolved { + ActionDef::SendDm { to, text } => { + assert_eq!(to, RECIPIENT_HEX); + assert_eq!(text, "You said: P1 incident in production"); + } + other => panic!("expected SendDm, got {other:?}"), + } + } }