From 035826cae9b985765019f2f4a10d1904e7086680 Mon Sep 17 00:00:00 2001 From: wayne-o Date: Wed, 29 Jul 2026 13:46:36 +0200 Subject: [PATCH 1/2] feat(workflow): expose message thread context Signed-off-by: wayne-o --- crates/buzz-workflow/src/executor.rs | 49 +++++++++++++++++++ crates/buzz-workflow/src/lib.rs | 72 ++++++++++++++++++++++++++++ 2 files changed, 121 insertions(+) diff --git a/crates/buzz-workflow/src/executor.rs b/crates/buzz-workflow/src/executor.rs index e30541377e..fa57f6c800 100644 --- a/crates/buzz-workflow/src/executor.rs +++ b/crates/buzz-workflow/src/executor.rs @@ -37,6 +37,10 @@ pub struct TriggerContext { pub emoji: String, /// Event ID of the triggering message (hex string). pub message_id: String, + /// Root event ID for a threaded message, or the message's own ID when top-level. + pub root_message_id: String, + /// Immediate parent event ID for a reply, empty when the message is top-level. + pub parent_message_id: String, /// Arbitrary webhook body fields (webhook trigger). pub webhook_fields: HashMap, } @@ -54,6 +58,8 @@ impl TriggerContext { "timestamp" => Some(&self.timestamp), "emoji" => Some(&self.emoji), "message_id" => Some(&self.message_id), + "root_message_id" => Some(&self.root_message_id), + "parent_message_id" => Some(&self.parent_message_id), other => self.webhook_fields.get(other).map(|s| s.as_str()), } } @@ -65,6 +71,7 @@ impl TriggerContext { /// - `| truncate(N)` — truncate to N characters /// - `| npub` — encode a hex pubkey as its full bech32 `npub` (non-pubkey /// values pass through unchanged); `truncate_pubkey` is a legacy alias +/// - `| json` — encode a string as a JSON string literal /// /// Unknown `{{keys}}` are left as literal text (no error, no substitution). pub fn resolve_template( @@ -195,6 +202,11 @@ fn apply_filter(value: String, filter: &str) -> Result { return Ok(value); } + if filter == "json" { + return serde_json::to_string(&value) + .map_err(|error| WorkflowError::TemplateError(error.to_string())); + } + Err(WorkflowError::TemplateError(format!( "unknown filter: {filter}" ))) @@ -213,6 +225,8 @@ fn apply_filter(value: String, filter: &str) -> Result { /// | `trigger.timestamp` | `trigger_timestamp` | /// | `trigger.emoji` | `trigger_emoji` | /// | `trigger.message_id` | `trigger_message_id` | +/// | `trigger.root_message_id` | `trigger_root_message_id` | +/// | `trigger.parent_message_id` | `trigger_parent_message_id` | /// | `steps.STEP_ID.output.FIELD` | `steps_STEP_ID_output_FIELD` | /// /// Also registers string helper functions that the `cron` crate's `evalexpr` v11 @@ -293,6 +307,14 @@ pub fn build_eval_context( ("trigger_timestamp", trigger_ctx.timestamp.as_str()), ("trigger_emoji", trigger_ctx.emoji.as_str()), ("trigger_message_id", trigger_ctx.message_id.as_str()), + ( + "trigger_root_message_id", + trigger_ctx.root_message_id.as_str(), + ), + ( + "trigger_parent_message_id", + trigger_ctx.parent_message_id.as_str(), + ), ]; for (name, val) in &trigger_fields { @@ -1229,6 +1251,8 @@ mod tests { timestamp: "1700000000".to_owned(), emoji: "fire".to_owned(), message_id: "event-id-hex".to_owned(), + root_message_id: "root-event-id-hex".to_owned(), + parent_message_id: "parent-event-id-hex".to_owned(), webhook_fields: HashMap::new(), } } @@ -1247,6 +1271,31 @@ mod tests { assert_eq!(out, "By abc123def456"); } + #[test] + fn resolve_trigger_thread_ids() { + let ctx = make_trigger(); + let out = resolve_template( + "{{trigger.root_message_id}}/{{trigger.parent_message_id}}", + &ctx, + &HashMap::new(), + ) + .unwrap(); + assert_eq!(out, "root-event-id-hex/parent-event-id-hex"); + } + + #[test] + fn json_filter_escapes_human_text() { + let mut ctx = make_trigger(); + ctx.text = "He said \"hello\"\nthen left.".to_owned(); + let out = resolve_template( + "{\"message\":{{trigger.text | json}}}", + &ctx, + &HashMap::new(), + ) + .unwrap(); + assert_eq!(out, "{\"message\":\"He said \\\"hello\\\"\\nthen left.\"}"); + } + #[test] fn resolve_step_output() { let ctx = make_trigger(); diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index 7aaa3d1702..f45ee7ca11 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -952,6 +952,8 @@ async fn should_fire_workflow( /// - `emoji` — for `KIND_REACTION` events, the content is the emoji; otherwise empty /// - `message_id` — for reactions, the target message's event ID (from `e` tag); /// for all other events, the event's own ID +/// - `root_message_id` — thread root for replies, otherwise the event's own ID +/// - `parent_message_id` — immediate parent for replies, otherwise empty pub fn build_trigger_context(event: &buzz_core::StoredEvent) -> executor::TriggerContext { let kind_u32 = event_kind_u32(&event.event); let content = event.event.content.clone(); @@ -1008,6 +1010,32 @@ pub fn build_trigger_context(event: &buzz_core::StoredEvent) -> executor::Trigge event.event.id.to_hex() }; + let mut thread_root = None; + let mut thread_parent = None; + if kind_u32 != KIND_REACTION { + for tag in event.event.tags.iter() { + let parts = tag.as_slice(); + if parts.first().map(String::as_str) != Some("e") || parts.len() < 4 { + continue; + } + + let id = &parts[1]; + if id.len() != 64 || !id.chars().all(|c| c.is_ascii_hexdigit()) { + continue; + } + + match parts[3].as_str() { + "root" => thread_root = Some(id.clone()), + "reply" => thread_parent = Some(id.clone()), + _ => {} + } + } + } + let root_message_id = thread_root + .or_else(|| thread_parent.clone()) + .unwrap_or_else(|| message_id.clone()); + let parent_message_id = thread_parent.unwrap_or_default(); + executor::TriggerContext { text: content, author, @@ -1018,6 +1046,8 @@ pub fn build_trigger_context(event: &buzz_core::StoredEvent) -> executor::Trigge timestamp: event.event.created_at.as_secs().to_string(), emoji, message_id, + root_message_id, + parent_message_id, webhook_fields: HashMap::new(), } } @@ -1563,11 +1593,53 @@ steps: assert_eq!(ctx.channel_id, stored.channel_id.unwrap().to_string()); assert_eq!(ctx.timestamp, stored.event.created_at.as_secs().to_string()); assert_eq!(ctx.message_id, stored.event.id.to_hex()); + assert_eq!(ctx.root_message_id, stored.event.id.to_hex()); + assert_eq!(ctx.parent_message_id, ""); // Non-reaction events have empty emoji. assert_eq!(ctx.emoji, ""); assert!(ctx.webhook_fields.is_empty()); } + #[test] + fn build_trigger_context_direct_reply_uses_parent_as_root() { + use nostr::{EventBuilder, EventId, Keys, Kind, Tag}; + use uuid::Uuid; + + let parent_id = EventId::from_byte_array([0x31; 32]).to_hex(); + let event = EventBuilder::new(Kind::Custom(9), "direct reply") + .tags([Tag::parse(["e", &parent_id, "", "reply"]).unwrap()]) + .sign_with_keys(&Keys::generate()) + .expect("sign"); + let stored = buzz_core::StoredEvent::new(event, Some(Uuid::new_v4())); + + let ctx = build_trigger_context(&stored); + + assert_eq!(ctx.root_message_id, parent_id); + assert_eq!(ctx.parent_message_id, parent_id); + } + + #[test] + fn build_trigger_context_nested_reply_preserves_root_and_parent() { + use nostr::{EventBuilder, EventId, Keys, Kind, Tag}; + use uuid::Uuid; + + let root_id = EventId::from_byte_array([0x41; 32]).to_hex(); + let parent_id = EventId::from_byte_array([0x42; 32]).to_hex(); + let event = EventBuilder::new(Kind::Custom(9), "nested reply") + .tags([ + Tag::parse(["e", &root_id, "", "root"]).unwrap(), + Tag::parse(["e", &parent_id, "", "reply"]).unwrap(), + ]) + .sign_with_keys(&Keys::generate()) + .expect("sign"); + let stored = buzz_core::StoredEvent::new(event, Some(Uuid::new_v4())); + + let ctx = build_trigger_context(&stored); + + assert_eq!(ctx.root_message_id, root_id); + assert_eq!(ctx.parent_message_id, parent_id); + } + #[test] fn build_trigger_context_reaction_event() { let (stored, target_id_hex) = make_reaction_event(); From 3e86a6df6dbf51bafbbc250e7f0c53ffc364d63c Mon Sep 17 00:00:00 2001 From: wayne-o Date: Wed, 29 Jul 2026 14:20:46 +0200 Subject: [PATCH 2/2] feat(workflow): support thread-root messages Signed-off-by: wayne-o --- crates/buzz-relay/src/workflow_sink.rs | 62 ++++++++++++++++++++++--- crates/buzz-workflow/src/action_sink.rs | 2 + crates/buzz-workflow/src/executor.rs | 48 +++++++++++++++++-- crates/buzz-workflow/src/schema.rs | 16 +++++++ 4 files changed, 119 insertions(+), 9 deletions(-) diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index 97c31c2561..0eee603761 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/crates/buzz-relay/src/workflow_sink.rs @@ -176,10 +176,12 @@ impl ActionSink for RelayActionSink { channel_id: &str, text: &str, author_pubkey: &str, + thread_root: Option<&str>, ) -> Pin> + Send + '_>> { let channel_id = channel_id.to_owned(); let text = text.to_owned(); let author_pubkey = author_pubkey.to_owned(); + let thread_root = thread_root.map(str::to_owned); Box::pin(async move { // 0. Upgrade weak reference — fails only during shutdown. @@ -235,6 +237,46 @@ impl ActionSink for RelayActionSink { )); } + let thread_root = if let Some(root_hex) = thread_root { + let root_id = nostr::EventId::from_hex(&root_hex).map_err(|e| { + ActionSinkError::InvalidInput(format!("invalid thread root event ID: {e}")) + })?; + let root_bytes = root_id.as_bytes().to_vec(); + let root_event = state + .db + .get_event_by_id(tenant.community(), &root_bytes) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))? + .ok_or_else(|| { + ActionSinkError::InvalidInput(format!( + "thread root event {root_hex} was not found" + )) + })?; + if root_event.channel_id != Some(channel_uuid) { + return Err(ActionSinkError::InvalidInput( + "thread root belongs to a different channel".into(), + )); + } + let root_meta = state + .db + .get_thread_metadata_by_event(tenant.community(), &root_bytes) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))?; + if root_meta.is_some_and(|meta| meta.parent_event_id.is_some()) { + return Err(ActionSinkError::InvalidInput( + "thread_root must identify a top-level message".into(), + )); + } + let root_created = chrono::DateTime::from_timestamp( + root_event.event.created_at.as_secs() as i64, + 0, + ) + .unwrap_or_else(Utc::now); + Some((root_id.to_hex(), root_bytes, root_created)) + } else { + None + }; + let author_pubkey = nostr::PublicKey::from_hex(&author_pubkey).map_err(|e| { ActionSinkError::InvalidInput(format!("invalid author pubkey: {e}")) })?; @@ -265,6 +307,12 @@ impl ActionSink for RelayActionSink { Tag::parse(["buzz:workflow", "true"]) .map_err(|e| ActionSinkError::EventBuild(format!("workflow tag: {e}")))?, ]; + if let Some((root_hex, _, _)) = &thread_root { + tags.push( + Tag::parse(["e", root_hex, "", "reply"]) + .map_err(|e| ActionSinkError::EventBuild(format!("thread tag: {e}")))?, + ); + } // Resolve `@Name` mentions to channel-member pubkeys and append a // `p` tag for each (skipping the author, already tagged above). A @@ -321,16 +369,17 @@ impl ActionSink for RelayActionSink { ); // 4. Persist event with thread metadata (matches REST handler path). - // Workflow messages are always top-level: depth=0, no parent/root. + let parent_event_id = thread_root.as_ref().map(|(_, bytes, _)| bytes.as_slice()); + let parent_created_at = thread_root.as_ref().map(|(_, _, created)| *created); let thread_meta = Some(buzz_db::event::ThreadMetadataParams { event_id: &event_id_bytes, event_created_at, channel_id: channel_uuid, - parent_event_id: None, - parent_event_created_at: None, - root_event_id: None, - root_event_created_at: None, - depth: 0, + parent_event_id, + parent_event_created_at: parent_created_at, + root_event_id: parent_event_id, + root_event_created_at: parent_created_at, + depth: i32::from(parent_event_id.is_some()), broadcast: false, }); @@ -676,6 +725,7 @@ mod integration_tests { &channel.id.to_string(), "heads up @Robby — please take a look", &author_hex, + None, ) .await .expect("send_message"); diff --git a/crates/buzz-workflow/src/action_sink.rs b/crates/buzz-workflow/src/action_sink.rs index 0c6002e74e..eee5d48340 100644 --- a/crates/buzz-workflow/src/action_sink.rs +++ b/crates/buzz-workflow/src/action_sink.rs @@ -57,6 +57,7 @@ pub trait ActionSink: Send + Sync { /// - `text`: message body (must not be empty/whitespace-only) /// - `author_pubkey`: hex-encoded pubkey of the workflow owner (used for /// the `p` attribution tag; the relay keypair signs the event) + /// - `thread_root`: optional top-level event ID to post a direct reply under /// /// Returns the event ID hex string on success. fn send_message( @@ -65,5 +66,6 @@ pub trait ActionSink: Send + Sync { channel_id: &str, text: &str, author_pubkey: &str, + thread_root: Option<&str>, ) -> Pin> + Send + '_>>; } diff --git a/crates/buzz-workflow/src/executor.rs b/crates/buzz-workflow/src/executor.rs index fa57f6c800..b9c2ce733f 100644 --- a/crates/buzz-workflow/src/executor.rs +++ b/crates/buzz-workflow/src/executor.rs @@ -425,9 +425,14 @@ pub fn resolve_step_templates( }; match &step.action { - SendMessage { text, channel } => Ok(SendMessage { + SendMessage { + text, + channel, + thread_root, + } => Ok(SendMessage { text: t(text)?, channel: t_opt(channel)?, + thread_root: t_opt(thread_root)?, }), SendDm { to, text } => Ok(SendDm { to: t(to)?, @@ -549,7 +554,11 @@ pub async fn dispatch_action( use ActionDef::*; match action { - SendMessage { text, channel } => { + SendMessage { + text, + channel, + thread_root, + } => { // Look up workflow metadata for destination validation and // attribution, scoped to the run's community — the same run/workflow // UUID may exist in another community, so a bare-id lookup could @@ -589,7 +598,13 @@ pub async fn dispatch_action( let event_id = engine .action_sink()? - .send_message(community_id, &channel_id, text, &owner_pubkey_hex) + .send_message( + community_id, + &channel_id, + text, + &owner_pubkey_hex, + thread_root.as_deref(), + ) .await .map_err(WorkflowError::from)?; @@ -1283,6 +1298,33 @@ mod tests { assert_eq!(out, "root-event-id-hex/parent-event-id-hex"); } + #[test] + fn resolve_send_message_thread_root_template() { + let step = Step { + id: "reply".to_owned(), + name: None, + if_expr: None, + timeout_secs: None, + action: ActionDef::SendMessage { + text: "{{trigger.text}}".to_owned(), + channel: None, + thread_root: Some("{{trigger.root_message_id}}".to_owned()), + }, + }; + + let resolved = resolve_step_templates(&step, &make_trigger(), &HashMap::new()).unwrap(); + + match resolved { + ActionDef::SendMessage { + text, thread_root, .. + } => { + assert_eq!(text, "P1 incident in production"); + assert_eq!(thread_root.as_deref(), Some("root-event-id-hex")); + } + other => panic!("unexpected action: {other:?}"), + } + } + #[test] fn json_filter_escapes_human_text() { let mut ctx = make_trigger(); diff --git a/crates/buzz-workflow/src/schema.rs b/crates/buzz-workflow/src/schema.rs index 9bc79aa48b..8dd4aa125e 100644 --- a/crates/buzz-workflow/src/schema.rs +++ b/crates/buzz-workflow/src/schema.rs @@ -97,6 +97,9 @@ pub enum ActionDef { /// Optional channel UUID override. Must be a valid UUID string. #[serde(default)] channel: Option, + /// Optional top-level event ID. When set, post directly into that thread. + #[serde(default)] + thread_root: Option, }, /// Send a direct message to a user. SendDm { @@ -298,6 +301,19 @@ mod tests { assert_eq!(reparsed.name, def.name); } + #[test] + fn parse_send_message_thread_root() { + let yaml = "name: Thread reply\ntrigger:\n on: webhook\nsteps:\n - id: reply\n action: send_message\n text: '{{trigger.message}}'\n thread_root: '{{trigger.root_message_id}}'\n"; + let (def, _) = parse_yaml(yaml).expect("parse failed"); + + match &def.steps[0].action { + ActionDef::SendMessage { thread_root, .. } => { + assert_eq!(thread_root.as_deref(), Some("{{trigger.root_message_id}}")); + } + other => panic!("unexpected action: {other:?}"), + } + } + #[test] fn parse_reaction_added_trigger() { let yaml = "name: Triage\ntrigger:\n on: reaction_added\n emoji: clipboard\nsteps:\n - id: ack\n action: add_reaction\n emoji: eyes\n";