mirror of
https://github.com/block/buzz.git
synced 2026-08-18 06:50:31 +02:00
feat(workflows): reply in-thread from send_message action
The send_message workflow action could only post new top-level channel messages. Add reply_in_thread so a message_posted workflow can reply in the triggering message's thread, and expose trigger_is_reply so a filter can restrict the workflow to top-level messages only. Schema rejects reply_in_thread on schedule/webhook triggers, which carry no message to reply to. The relay sink resolves parent/root/depth from the known trigger event (mirroring the ingest thread rules), emits NIP-10 root/reply e-tags, and persists real thread metadata instead of the hardcoded top-level params. The desktop step card gains a checkbox shown only for message-based triggers, round-tripped through the YAML form. Co-authored-by: Will Pfleger <pfleger.will@gmail.com> Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
Duncan
@@ -872,6 +872,128 @@ pub(crate) async fn resolve_nip10_thread_meta(
|
||||
}))
|
||||
}
|
||||
|
||||
/// Resolved thread ancestry for a relay-built reply (workflow path).
|
||||
///
|
||||
/// Carries the parent and root identifiers plus the reply's depth, so the
|
||||
/// caller can both emit matching NIP-10 `root`/`reply` tags and persist thread
|
||||
/// metadata for the signed reply event.
|
||||
pub(crate) struct ReplyAncestry {
|
||||
pub parent_event_id: Vec<u8>,
|
||||
pub parent_event_created_at: chrono::DateTime<Utc>,
|
||||
pub root_event_id: Vec<u8>,
|
||||
pub root_event_created_at: chrono::DateTime<Utc>,
|
||||
pub depth: i32,
|
||||
}
|
||||
|
||||
impl ReplyAncestry {
|
||||
/// Root event ID as lowercase hex, for the NIP-10 `root` tag.
|
||||
pub fn root_hex(&self) -> String {
|
||||
hex::encode(&self.root_event_id)
|
||||
}
|
||||
|
||||
/// Parent event ID as lowercase hex, for the NIP-10 `reply` tag.
|
||||
pub fn parent_hex(&self) -> String {
|
||||
hex::encode(&self.parent_event_id)
|
||||
}
|
||||
|
||||
/// Build the DB thread-metadata params for the signed reply event.
|
||||
pub fn into_thread_meta(
|
||||
self,
|
||||
reply_event_id: Vec<u8>,
|
||||
reply_created_at: chrono::DateTime<Utc>,
|
||||
channel_id: Uuid,
|
||||
) -> ThreadMetadataOwned {
|
||||
ThreadMetadataOwned {
|
||||
event_id: reply_event_id,
|
||||
event_created_at: reply_created_at,
|
||||
channel_id,
|
||||
parent_event_id: self.parent_event_id,
|
||||
parent_event_created_at: self.parent_event_created_at,
|
||||
root_event_id: self.root_event_id,
|
||||
root_event_created_at: self.root_event_created_at,
|
||||
depth: self.depth,
|
||||
broadcast: false,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Resolve thread ancestry for a reply built by the relay (workflow path).
|
||||
///
|
||||
/// Unlike [`resolve_nip10_thread_meta`], which validates client-supplied NIP-10
|
||||
/// `e` tags, this derives ancestry from a known `parent_hex` (the triggering
|
||||
/// event) and *computes* the correct root and depth. Enforces the same-channel
|
||||
/// invariant and the depth limit that the ingest path applies.
|
||||
pub(crate) async fn resolve_relay_reply_thread_meta(
|
||||
community_id: CommunityId,
|
||||
parent_hex: &str,
|
||||
channel_id: Uuid,
|
||||
state: &AppState,
|
||||
) -> Result<ReplyAncestry, String> {
|
||||
let parent_bytes =
|
||||
hex::decode(parent_hex).map_err(|_| "invalid parent event ID hex".to_string())?;
|
||||
|
||||
let (parent_event_result, parent_meta_result) = tokio::join!(
|
||||
state.db.get_event_by_id(community_id, &parent_bytes),
|
||||
state
|
||||
.db
|
||||
.get_thread_metadata_by_event(community_id, &parent_bytes),
|
||||
);
|
||||
|
||||
let parent_event = parent_event_result
|
||||
.map_err(|e| format!("db error looking up parent: {e}"))?
|
||||
.ok_or_else(|| "reply parent not found".to_string())?;
|
||||
|
||||
match parent_event.channel_id {
|
||||
Some(parent_ch) if parent_ch != channel_id => {
|
||||
return Err("parent event belongs to a different channel".to_string());
|
||||
}
|
||||
None => return Err("parent event has no channel association".to_string()),
|
||||
_ => {}
|
||||
}
|
||||
|
||||
let parent_created =
|
||||
chrono::DateTime::from_timestamp(parent_event.event.created_at.as_secs() as i64, 0)
|
||||
.unwrap_or_else(Utc::now);
|
||||
|
||||
let parent_meta =
|
||||
parent_meta_result.map_err(|e| format!("db error looking up thread metadata: {e}"))?;
|
||||
|
||||
// Root = parent's root if the parent is itself a reply, else the parent.
|
||||
// Depth = parent depth + 1 (a direct reply to a top-level message is depth 1).
|
||||
let (root_bytes, root_created, depth) = match parent_meta {
|
||||
Some(meta) => {
|
||||
let effective_root = meta.root_event_id.unwrap_or_else(|| parent_bytes.clone());
|
||||
let root_ts = if effective_root == parent_bytes {
|
||||
parent_created
|
||||
} else if let Ok(Some(root_ev)) = state
|
||||
.db
|
||||
.get_event_by_id(community_id, &effective_root)
|
||||
.await
|
||||
{
|
||||
chrono::DateTime::from_timestamp(root_ev.event.created_at.as_secs() as i64, 0)
|
||||
.unwrap_or(parent_created)
|
||||
} else {
|
||||
parent_created
|
||||
};
|
||||
(effective_root, root_ts, meta.depth + 1)
|
||||
}
|
||||
// No metadata row ⇒ parent is a top-level message; it is its own root.
|
||||
None => (parent_bytes.clone(), parent_created, 1),
|
||||
};
|
||||
|
||||
if depth > 100 {
|
||||
return Err("thread depth limit exceeded".to_string());
|
||||
}
|
||||
|
||||
Ok(ReplyAncestry {
|
||||
parent_event_id: parent_bytes,
|
||||
parent_event_created_at: parent_created,
|
||||
root_event_id: root_bytes,
|
||||
root_event_created_at: root_created,
|
||||
depth,
|
||||
})
|
||||
}
|
||||
|
||||
/// Count all `e` tags regardless of content validity.
|
||||
fn count_e_tags(event: &Event) -> usize {
|
||||
event
|
||||
|
||||
@@ -176,10 +176,12 @@ impl ActionSink for RelayActionSink {
|
||||
channel_id: &str,
|
||||
text: &str,
|
||||
author_pubkey: &str,
|
||||
reply_to: Option<&str>,
|
||||
) -> Pin<Box<dyn Future<Output = Result<String, ActionSinkError>> + Send + '_>> {
|
||||
let channel_id = channel_id.to_owned();
|
||||
let text = text.to_owned();
|
||||
let author_pubkey = author_pubkey.to_owned();
|
||||
let reply_to = reply_to.map(str::to_owned);
|
||||
|
||||
Box::pin(async move {
|
||||
// 0. Upgrade weak reference — fails only during shutdown.
|
||||
@@ -271,6 +273,39 @@ impl ActionSink for RelayActionSink {
|
||||
.map_err(|e| ActionSinkError::EventBuild(format!("workflow owner tag: {e}")))?,
|
||||
];
|
||||
|
||||
// Resolve thread ancestry when this is a threaded reply, so the
|
||||
// built event carries NIP-10 `root`/`reply` e-tags and persists real
|
||||
// thread metadata (matching the ingest path) instead of top-level.
|
||||
let reply_ancestry = match reply_to.as_deref() {
|
||||
Some(parent_hex) => Some(
|
||||
crate::handlers::ingest::resolve_relay_reply_thread_meta(
|
||||
tenant.community(),
|
||||
parent_hex,
|
||||
channel_uuid,
|
||||
&state,
|
||||
)
|
||||
.await
|
||||
.map_err(ActionSinkError::InvalidInput)?,
|
||||
),
|
||||
None => None,
|
||||
};
|
||||
|
||||
// NIP-10 e-tags for the thread. Marked `root`/`reply` so clients and
|
||||
// the ingest resolver read the ancestry the same way. When the reply
|
||||
// is directly under the root, both markers point at the same event.
|
||||
if let Some(ancestry) = &reply_ancestry {
|
||||
let root_hex = ancestry.root_hex();
|
||||
let parent_hex = ancestry.parent_hex();
|
||||
tags.push(
|
||||
Tag::parse(["e", &root_hex, "", "root"])
|
||||
.map_err(|e| ActionSinkError::EventBuild(format!("root e tag: {e}")))?,
|
||||
);
|
||||
tags.push(
|
||||
Tag::parse(["e", &parent_hex, "", "reply"])
|
||||
.map_err(|e| ActionSinkError::EventBuild(format!("reply e tag: {e}")))?,
|
||||
);
|
||||
}
|
||||
|
||||
// Resolve `@Name` mentions to channel-member pubkeys and append a
|
||||
// `p` tag for each (skipping the author, already tagged above). A
|
||||
// resolution failure must not drop the message, so log and proceed
|
||||
@@ -326,17 +361,24 @@ 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 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,
|
||||
broadcast: false,
|
||||
// Threaded replies persist the resolved parent/root/depth; a
|
||||
// non-reply workflow message stays top-level (depth=0, no parent).
|
||||
let thread_meta_owned = reply_ancestry.map(|ancestry| {
|
||||
ancestry.into_thread_meta(event_id_bytes.clone(), event_created_at, channel_uuid)
|
||||
});
|
||||
let thread_meta = Some(match &thread_meta_owned {
|
||||
Some(owned) => owned.as_params(),
|
||||
None => 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,
|
||||
broadcast: false,
|
||||
},
|
||||
});
|
||||
|
||||
let (stored_event, was_inserted) = state
|
||||
@@ -681,6 +723,7 @@ mod integration_tests {
|
||||
&channel.id.to_string(),
|
||||
"heads up @Robby — please take a look",
|
||||
&author_hex,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect("send_message");
|
||||
@@ -726,4 +769,153 @@ mod integration_tests {
|
||||
so consumers never infer ownership from p-tag order"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn workflow_reply_in_thread_threads_onto_parent() {
|
||||
let state = test_state().await;
|
||||
|
||||
let author = nostr::Keys::generate();
|
||||
let author_hex = author.public_key().to_hex();
|
||||
|
||||
let host = format!("wf-thread-{}.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:?}"),
|
||||
};
|
||||
|
||||
let channel = state
|
||||
.db
|
||||
.create_channel(
|
||||
community,
|
||||
"wf-thread",
|
||||
ChannelType::Stream,
|
||||
ChannelVisibility::Open,
|
||||
None,
|
||||
&author.public_key().to_bytes(),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect("create channel");
|
||||
|
||||
let sink = RelayActionSink::new(&state);
|
||||
|
||||
// 1. A top-level workflow message becomes the thread root.
|
||||
let root_hex = sink
|
||||
.send_message(
|
||||
community,
|
||||
&channel.id.to_string(),
|
||||
"root message",
|
||||
&author_hex,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect("send root");
|
||||
|
||||
// 2. A reply_in_thread message threads onto it.
|
||||
let reply_hex = sink
|
||||
.send_message(
|
||||
community,
|
||||
&channel.id.to_string(),
|
||||
"threaded reply",
|
||||
&author_hex,
|
||||
Some(&root_hex),
|
||||
)
|
||||
.await
|
||||
.expect("send reply");
|
||||
|
||||
// The reply event carries NIP-10 root+reply e-tags pointing at the root.
|
||||
let reply_id_bytes = nostr::EventId::from_hex(&reply_hex)
|
||||
.expect("reply id")
|
||||
.as_bytes()
|
||||
.to_vec();
|
||||
let stored = state
|
||||
.db
|
||||
.get_event_by_id(community, &reply_id_bytes)
|
||||
.await
|
||||
.expect("query reply")
|
||||
.expect("reply persisted");
|
||||
let marker = |m: &str| -> Option<String> {
|
||||
stored.event.tags.iter().find_map(|t| {
|
||||
let p = t.as_slice();
|
||||
if p.len() >= 4 && p[0] == "e" && p[3] == m {
|
||||
Some(p[1].clone())
|
||||
} else {
|
||||
None
|
||||
}
|
||||
})
|
||||
};
|
||||
assert_eq!(marker("root").as_deref(), Some(root_hex.as_str()));
|
||||
assert_eq!(marker("reply").as_deref(), Some(root_hex.as_str()));
|
||||
|
||||
// Thread metadata reflects a depth-1 reply parented on the root.
|
||||
let meta = state
|
||||
.db
|
||||
.get_thread_metadata_by_event(community, &reply_id_bytes)
|
||||
.await
|
||||
.expect("query meta")
|
||||
.expect("reply has thread metadata");
|
||||
assert_eq!(
|
||||
meta.depth, 1,
|
||||
"direct reply to a top-level message is depth 1"
|
||||
);
|
||||
let root_bytes = nostr::EventId::from_hex(&root_hex)
|
||||
.expect("root id")
|
||||
.as_bytes()
|
||||
.to_vec();
|
||||
assert_eq!(meta.parent_event_id.as_deref(), Some(root_bytes.as_slice()));
|
||||
assert_eq!(meta.root_event_id.as_deref(), Some(root_bytes.as_slice()));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn workflow_reply_to_missing_parent_errors() {
|
||||
let state = test_state().await;
|
||||
let author = nostr::Keys::generate();
|
||||
let author_hex = author.public_key().to_hex();
|
||||
let host = format!("wf-missing-{}.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:?}"),
|
||||
};
|
||||
let channel = state
|
||||
.db
|
||||
.create_channel(
|
||||
community,
|
||||
"wf-missing",
|
||||
ChannelType::Stream,
|
||||
ChannelVisibility::Open,
|
||||
None,
|
||||
&author.public_key().to_bytes(),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect("create channel");
|
||||
|
||||
let unknown = nostr::Keys::generate().public_key().to_hex();
|
||||
let err = RelayActionSink::new(&state)
|
||||
.send_message(
|
||||
community,
|
||||
&channel.id.to_string(),
|
||||
"orphan reply",
|
||||
&author_hex,
|
||||
Some(&unknown),
|
||||
)
|
||||
.await
|
||||
.expect_err("reply to a non-existent parent must fail");
|
||||
assert!(
|
||||
matches!(err, ActionSinkError::InvalidInput(_)),
|
||||
"expected InvalidInput, got {err:?}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -57,6 +57,9 @@ 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)
|
||||
/// - `reply_to`: when `Some(event_id_hex)`, the message is posted as a
|
||||
/// threaded reply to that event (NIP-10 root/reply tags + real thread
|
||||
/// metadata); when `None`, it is a top-level channel message.
|
||||
///
|
||||
/// Returns the event ID hex string on success.
|
||||
fn send_message(
|
||||
@@ -65,5 +68,6 @@ pub trait ActionSink: Send + Sync {
|
||||
channel_id: &str,
|
||||
text: &str,
|
||||
author_pubkey: &str,
|
||||
reply_to: Option<&str>,
|
||||
) -> Pin<Box<dyn Future<Output = Result<String, ActionSinkError>> + Send + '_>>;
|
||||
}
|
||||
|
||||
@@ -37,6 +37,10 @@ pub struct TriggerContext {
|
||||
pub emoji: String,
|
||||
/// Event ID of the triggering message (hex string).
|
||||
pub message_id: String,
|
||||
/// True when the triggering event is itself a threaded reply (carries a
|
||||
/// NIP-10 `reply`/`root` marker e-tag). Lets a `message_posted` filter
|
||||
/// select only top-level messages via `trigger_is_reply == false`.
|
||||
pub is_reply: bool,
|
||||
/// Arbitrary webhook body fields (webhook trigger).
|
||||
pub webhook_fields: HashMap<String, String>,
|
||||
}
|
||||
@@ -213,6 +217,7 @@ fn apply_filter(value: String, filter: &str) -> Result<String, WorkflowError> {
|
||||
/// | `trigger.timestamp` | `trigger_timestamp` |
|
||||
/// | `trigger.emoji` | `trigger_emoji` |
|
||||
/// | `trigger.message_id` | `trigger_message_id` |
|
||||
/// | `trigger.is_reply` | `trigger_is_reply` (bool) |
|
||||
/// | `steps.STEP_ID.output.FIELD` | `steps_STEP_ID_output_FIELD` |
|
||||
///
|
||||
/// Also registers string helper functions that the `cron` crate's `evalexpr` v11
|
||||
@@ -300,6 +305,14 @@ pub fn build_eval_context(
|
||||
.map_err(|e| WorkflowError::ConditionError(e.to_string()))?;
|
||||
}
|
||||
|
||||
// `trigger_is_reply` is boolean (not a string field), so a filter can read
|
||||
// `trigger_is_reply == false` to fire only on top-level messages.
|
||||
ctx.set_value(
|
||||
"trigger_is_reply".into(),
|
||||
Value::Boolean(trigger_ctx.is_reply),
|
||||
)
|
||||
.map_err(|e| WorkflowError::ConditionError(e.to_string()))?;
|
||||
|
||||
for (step_id, output) in step_outputs {
|
||||
if let JsonValue::Object(map) = output {
|
||||
for (field, val) in map {
|
||||
@@ -403,9 +416,14 @@ pub fn resolve_step_templates(
|
||||
};
|
||||
|
||||
match &step.action {
|
||||
SendMessage { text, channel } => Ok(SendMessage {
|
||||
SendMessage {
|
||||
text,
|
||||
channel,
|
||||
reply_in_thread,
|
||||
} => Ok(SendMessage {
|
||||
text: t(text)?,
|
||||
channel: t_opt(channel)?,
|
||||
reply_in_thread: *reply_in_thread,
|
||||
}),
|
||||
SendDm { to, text } => Ok(SendDm {
|
||||
to: t(to)?,
|
||||
@@ -546,7 +564,11 @@ pub async fn dispatch_action(
|
||||
let result = serving_write
|
||||
.protect(async {
|
||||
match action {
|
||||
SendMessage { text, channel } => {
|
||||
SendMessage {
|
||||
text,
|
||||
channel,
|
||||
reply_in_thread,
|
||||
} => {
|
||||
// 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
|
||||
@@ -577,16 +599,38 @@ pub async fn dispatch_action(
|
||||
)?;
|
||||
let owner_pubkey_hex = hex::encode(&workflow.owner_pubkey);
|
||||
|
||||
// Thread the reply onto the triggering message when requested.
|
||||
// The trigger must carry the event to reply to; schema
|
||||
// validation already forbids `reply_in_thread` on triggers
|
||||
// that have no message, so an empty id here is a real fault.
|
||||
let reply_to = if *reply_in_thread {
|
||||
if trigger_ctx.message_id.is_empty() {
|
||||
return Err(WorkflowError::InvalidDefinition(
|
||||
"SendMessage: reply_in_thread is set but the trigger has no message_id to reply to".into(),
|
||||
));
|
||||
}
|
||||
Some(trigger_ctx.message_id.as_str())
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
info!(
|
||||
run_id = %run_id,
|
||||
step = step_id,
|
||||
channel = %channel_id,
|
||||
reply_in_thread = *reply_in_thread,
|
||||
"SendMessage → {channel_id}: {text}"
|
||||
);
|
||||
|
||||
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,
|
||||
reply_to,
|
||||
)
|
||||
.await
|
||||
.map_err(WorkflowError::from)?;
|
||||
|
||||
@@ -1266,6 +1310,7 @@ mod tests {
|
||||
timestamp: "1700000000".to_owned(),
|
||||
emoji: "fire".to_owned(),
|
||||
message_id: "event-id-hex".to_owned(),
|
||||
is_reply: false,
|
||||
webhook_fields: HashMap::new(),
|
||||
}
|
||||
}
|
||||
@@ -1385,6 +1430,56 @@ mod tests {
|
||||
assert!(!result);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn condition_trigger_is_reply_selects_top_level_only() {
|
||||
// The top-level-only filter from the feature's use case.
|
||||
let mut ctx = make_trigger();
|
||||
|
||||
ctx.is_reply = false;
|
||||
assert!(
|
||||
evaluate_condition("trigger_is_reply == false", &ctx, &HashMap::new())
|
||||
.await
|
||||
.unwrap(),
|
||||
"top-level message should pass the filter"
|
||||
);
|
||||
|
||||
ctx.is_reply = true;
|
||||
assert!(
|
||||
!evaluate_condition("trigger_is_reply == false", &ctx, &HashMap::new())
|
||||
.await
|
||||
.unwrap(),
|
||||
"threaded reply should be filtered out"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn resolve_step_templates_carries_reply_in_thread() {
|
||||
let ctx = make_trigger();
|
||||
let step = Step {
|
||||
id: "reply".to_owned(),
|
||||
name: None,
|
||||
if_expr: None,
|
||||
timeout_secs: None,
|
||||
action: ActionDef::SendMessage {
|
||||
text: "hi {{trigger.author}}".to_owned(),
|
||||
channel: None,
|
||||
reply_in_thread: true,
|
||||
},
|
||||
};
|
||||
let resolved = resolve_step_templates(&step, &ctx, &HashMap::new()).unwrap();
|
||||
match resolved {
|
||||
ActionDef::SendMessage {
|
||||
text,
|
||||
reply_in_thread,
|
||||
..
|
||||
} => {
|
||||
assert_eq!(text, "hi abc123def456");
|
||||
assert!(reply_in_thread, "reply_in_thread must survive resolution");
|
||||
}
|
||||
other => panic!("unexpected action: {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn condition_or_expression() {
|
||||
let ctx = make_trigger(); // text contains "P1"
|
||||
|
||||
@@ -1016,10 +1016,21 @@ pub fn build_trigger_context(event: &buzz_core::StoredEvent) -> executor::Trigge
|
||||
timestamp: event.event.created_at.as_secs().to_string(),
|
||||
emoji,
|
||||
message_id,
|
||||
is_reply: event_is_reply(&event.event),
|
||||
webhook_fields: HashMap::new(),
|
||||
}
|
||||
}
|
||||
|
||||
/// True when an event is a threaded reply — it carries a NIP-10 `e` tag marked
|
||||
/// `reply` or `root`. A top-level message has neither, so a `message_posted`
|
||||
/// filter of `trigger_is_reply == false` fires only on new top-level messages.
|
||||
fn event_is_reply(event: &nostr::Event) -> bool {
|
||||
event.tags.iter().any(|tag| {
|
||||
let parts = tag.as_slice();
|
||||
parts.len() >= 4 && parts[0] == "e" && (parts[3] == "reply" || parts[3] == "root")
|
||||
})
|
||||
}
|
||||
|
||||
/// Pure authority decision for [`WorkflowEngine::check_owner_authority`].
|
||||
///
|
||||
/// `role` is the owner's *current* active role in the workflow's channel
|
||||
@@ -1564,6 +1575,53 @@ steps:
|
||||
// Non-reaction events have empty emoji.
|
||||
assert_eq!(ctx.emoji, "");
|
||||
assert!(ctx.webhook_fields.is_empty());
|
||||
// A top-level message (no e-tags) is not a reply.
|
||||
assert!(!ctx.is_reply);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn build_trigger_context_is_reply_true_for_threaded_message() {
|
||||
use nostr::{EventBuilder, Keys, Kind, Tag};
|
||||
use uuid::Uuid;
|
||||
let root = Keys::generate();
|
||||
let root_event = EventBuilder::new(Kind::Custom(9), "root")
|
||||
.tags([])
|
||||
.sign_with_keys(&root)
|
||||
.expect("sign root");
|
||||
let root_hex = root_event.id.to_hex();
|
||||
|
||||
let keys = Keys::generate();
|
||||
let event = EventBuilder::new(Kind::Custom(9), "a threaded reply")
|
||||
.tags([
|
||||
Tag::parse(["e", &root_hex, "", "root"]).expect("root tag"),
|
||||
Tag::parse(["e", &root_hex, "", "reply"]).expect("reply tag"),
|
||||
])
|
||||
.sign_with_keys(&keys)
|
||||
.expect("sign");
|
||||
let stored = buzz_core::StoredEvent::new(event, Some(Uuid::new_v4()));
|
||||
let ctx = build_trigger_context(&stored);
|
||||
assert!(ctx.is_reply, "message with reply/root e-tags is a reply");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn build_trigger_context_is_reply_false_for_unmarked_e_tag() {
|
||||
// A bare `e` tag with no NIP-10 marker (e.g. a plain mention/quote) is
|
||||
// not treated as a thread reply — only `reply`/`root` markers count.
|
||||
use nostr::{EventBuilder, Keys, Kind, Tag};
|
||||
use uuid::Uuid;
|
||||
let other = Keys::generate();
|
||||
let other_event = EventBuilder::new(Kind::Custom(9), "other")
|
||||
.tags([])
|
||||
.sign_with_keys(&other)
|
||||
.expect("sign");
|
||||
let keys = Keys::generate();
|
||||
let event = EventBuilder::new(Kind::Custom(9), "quotes another")
|
||||
.tags([Tag::parse(["e", &other_event.id.to_hex()]).expect("bare e tag")])
|
||||
.sign_with_keys(&keys)
|
||||
.expect("sign");
|
||||
let stored = buzz_core::StoredEvent::new(event, Some(Uuid::new_v4()));
|
||||
let ctx = build_trigger_context(&stored);
|
||||
assert!(!ctx.is_reply, "unmarked e-tag must not count as a reply");
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -97,6 +97,11 @@ pub enum ActionDef {
|
||||
/// Optional channel UUID override. Must be a valid UUID string.
|
||||
#[serde(default)]
|
||||
channel: Option<String>,
|
||||
/// Reply to the triggering message in its thread instead of posting a
|
||||
/// new top-level message. Only valid for message-based triggers, which
|
||||
/// carry a triggering event to reply to.
|
||||
#[serde(default)]
|
||||
reply_in_thread: bool,
|
||||
},
|
||||
/// Send a direct message to a user.
|
||||
SendDm {
|
||||
@@ -205,6 +210,34 @@ impl WorkflowDef {
|
||||
}
|
||||
}
|
||||
|
||||
// `reply_in_thread` requires a triggering message to reply to. Schedule
|
||||
// and webhook triggers have none, so reject the combination at
|
||||
// definition time rather than failing silently at run time.
|
||||
let trigger_has_message = matches!(
|
||||
self.trigger,
|
||||
TriggerDef::MessagePosted { .. }
|
||||
| TriggerDef::ReactionAdded { .. }
|
||||
| TriggerDef::DiffPosted { .. }
|
||||
);
|
||||
if !trigger_has_message {
|
||||
for step in &self.steps {
|
||||
if matches!(
|
||||
step.action,
|
||||
ActionDef::SendMessage {
|
||||
reply_in_thread: true,
|
||||
..
|
||||
}
|
||||
) {
|
||||
return Err(WorkflowError::InvalidDefinition(format!(
|
||||
"step '{}': reply_in_thread requires a message-based trigger \
|
||||
(message_posted, reaction_added, or diff_posted); \
|
||||
schedule and webhook triggers have no message to reply to",
|
||||
step.id
|
||||
)));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if let TriggerDef::Schedule { cron, interval } = &self.trigger {
|
||||
if cron.is_none() && interval.is_none() {
|
||||
return Err(WorkflowError::InvalidDefinition(
|
||||
@@ -454,6 +487,78 @@ mod tests {
|
||||
assert!(matches!(err, WorkflowError::InvalidDefinition(_)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reply_in_thread_defaults_false_and_round_trips() {
|
||||
// Absent field defaults to false.
|
||||
let yaml = "name: Auto Reply\ntrigger:\n on: message_posted\nsteps:\n - id: s1\n action: send_message\n text: hi\n";
|
||||
let (def, _) = parse_yaml(yaml).expect("parse failed");
|
||||
match &def.steps[0].action {
|
||||
ActionDef::SendMessage {
|
||||
reply_in_thread, ..
|
||||
} => assert!(!reply_in_thread, "should default to false"),
|
||||
other => panic!("unexpected action: {other:?}"),
|
||||
}
|
||||
|
||||
// Explicit true parses, and survives a JSON round-trip.
|
||||
let yaml = "name: Auto Reply\ntrigger:\n on: message_posted\nsteps:\n - id: s1\n action: send_message\n text: hi\n reply_in_thread: true\n";
|
||||
let (def, _) = parse_yaml(yaml).expect("parse failed");
|
||||
match &def.steps[0].action {
|
||||
ActionDef::SendMessage {
|
||||
reply_in_thread, ..
|
||||
} => assert!(reply_in_thread),
|
||||
other => panic!("unexpected action: {other:?}"),
|
||||
}
|
||||
let json = serde_json::to_string(&def).expect("serialize");
|
||||
let reparsed: WorkflowDef = serde_json::from_str(&json).expect("json round-trip");
|
||||
assert!(matches!(
|
||||
&reparsed.steps[0].action,
|
||||
ActionDef::SendMessage {
|
||||
reply_in_thread: true,
|
||||
..
|
||||
}
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn validate_accepts_reply_in_thread_on_message_triggers() {
|
||||
for on in ["message_posted", "reaction_added", "diff_posted"] {
|
||||
let yaml = format!(
|
||||
"name: Auto Reply\ntrigger:\n on: {on}\nsteps:\n - id: s1\n action: send_message\n text: hi\n reply_in_thread: true\n"
|
||||
);
|
||||
parse_yaml(&yaml)
|
||||
.unwrap_or_else(|e| panic!("reply_in_thread should be valid on {on}: {e}"));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn validate_rejects_reply_in_thread_on_schedule_trigger() {
|
||||
let yaml = "name: Bad\ntrigger:\n on: schedule\n cron: '0 9 * * 1-5'\nsteps:\n - id: s1\n action: send_message\n text: hi\n reply_in_thread: true\n";
|
||||
let err = parse_yaml(yaml).unwrap_err();
|
||||
match &err {
|
||||
WorkflowError::InvalidDefinition(msg) => {
|
||||
assert!(
|
||||
msg.contains("reply_in_thread"),
|
||||
"expected reply_in_thread in: {msg}"
|
||||
);
|
||||
}
|
||||
other => panic!("expected InvalidDefinition, got: {other}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn validate_rejects_reply_in_thread_on_webhook_trigger() {
|
||||
let yaml = "name: Bad\ntrigger:\n on: webhook\nsteps:\n - id: s1\n action: send_message\n text: hi\n channel: 00000000-0000-0000-0000-000000000000\n reply_in_thread: true\n";
|
||||
let err = parse_yaml(yaml).unwrap_err();
|
||||
assert!(matches!(err, WorkflowError::InvalidDefinition(_)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn validate_allows_reply_in_thread_false_on_schedule() {
|
||||
// Explicit `false` on a schedule trigger is fine — no message needed.
|
||||
let yaml = "name: OK\ntrigger:\n on: schedule\n cron: '0 9 * * 1-5'\nsteps:\n - id: s1\n action: send_message\n text: hi\n reply_in_thread: false\n";
|
||||
parse_yaml(yaml).expect("reply_in_thread: false on schedule should be valid");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn enabled_defaults_to_true() {
|
||||
let yaml = "name: Test\ntrigger:\n on: webhook\nsteps:\n - id: s1\n action: delay\n duration: 1m\n";
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import { Trash2 } from "lucide-react";
|
||||
|
||||
import { Button } from "@/shared/ui/button";
|
||||
import { Checkbox } from "@/shared/ui/checkbox";
|
||||
import { Input } from "@/shared/ui/input";
|
||||
import { Textarea } from "@/shared/ui/textarea";
|
||||
import { FieldLabel, FormSelect } from "./workflowFormPrimitives";
|
||||
@@ -112,6 +113,21 @@ function StepConfigFields({
|
||||
</p>
|
||||
) : null}
|
||||
</div>
|
||||
{triggerType !== "webhook" && triggerType !== "schedule" ? (
|
||||
<div className="flex items-center gap-2">
|
||||
<Checkbox
|
||||
checked={step.replyInThread === true}
|
||||
disabled={disabled}
|
||||
id={`${prefix}-reply-in-thread`}
|
||||
onCheckedChange={(checked) =>
|
||||
onUpdate({ ...step, replyInThread: checked === true })
|
||||
}
|
||||
/>
|
||||
<label className="text-xs" htmlFor={`${prefix}-reply-in-thread`}>
|
||||
Reply to triggering message in thread
|
||||
</label>
|
||||
</div>
|
||||
) : null}
|
||||
</div>
|
||||
);
|
||||
case "send_dm":
|
||||
|
||||
@@ -0,0 +1,63 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { test } from "node:test";
|
||||
|
||||
import {
|
||||
formStateToYaml,
|
||||
yamlToFormState,
|
||||
DEFAULT_FORM_STATE,
|
||||
} from "./workflowFormTypes.ts";
|
||||
|
||||
function sendMessageState(overrides) {
|
||||
return {
|
||||
...DEFAULT_FORM_STATE,
|
||||
name: "Auto Reply",
|
||||
trigger: { on: "message_posted", filter: "trigger_is_reply == false" },
|
||||
steps: [
|
||||
{
|
||||
id: "step_1",
|
||||
action: "send_message",
|
||||
text: "pre-written reply",
|
||||
...overrides,
|
||||
},
|
||||
],
|
||||
};
|
||||
}
|
||||
|
||||
test("reply_in_thread is emitted only when the checkbox is on", () => {
|
||||
const withReply = formStateToYaml(sendMessageState({ replyInThread: true }));
|
||||
assert.match(withReply, /reply_in_thread: true/);
|
||||
|
||||
const withoutReply = formStateToYaml(
|
||||
sendMessageState({ replyInThread: false }),
|
||||
);
|
||||
assert.doesNotMatch(withoutReply, /reply_in_thread/);
|
||||
|
||||
const unset = formStateToYaml(sendMessageState({}));
|
||||
assert.doesNotMatch(unset, /reply_in_thread/);
|
||||
});
|
||||
|
||||
test("reply_in_thread round-trips YAML -> form -> YAML", () => {
|
||||
const yaml = formStateToYaml(sendMessageState({ replyInThread: true }));
|
||||
const parsed = yamlToFormState(yaml);
|
||||
assert.equal(parsed.ok, true);
|
||||
assert.equal(parsed.state.steps[0].replyInThread, true);
|
||||
|
||||
const reserialized = formStateToYaml(parsed.state);
|
||||
assert.match(reserialized, /reply_in_thread: true/);
|
||||
});
|
||||
|
||||
test("absent reply_in_thread parses as false", () => {
|
||||
const yaml = [
|
||||
"name: No Reply",
|
||||
"trigger:",
|
||||
" on: message_posted",
|
||||
"steps:",
|
||||
" - id: step_1",
|
||||
" action: send_message",
|
||||
" text: hi",
|
||||
"",
|
||||
].join("\n");
|
||||
const parsed = yamlToFormState(yaml);
|
||||
assert.equal(parsed.ok, true);
|
||||
assert.equal(parsed.state.steps[0].replyInThread, false);
|
||||
});
|
||||
@@ -43,6 +43,7 @@ export type StepFormState = {
|
||||
duration?: string;
|
||||
text?: string;
|
||||
channel?: string;
|
||||
replyInThread?: boolean;
|
||||
to?: string;
|
||||
url?: string;
|
||||
method?: string;
|
||||
@@ -141,6 +142,7 @@ function actionFieldsForStep(step: StepFormState): Record<string, unknown> {
|
||||
case "send_message":
|
||||
if (step.text) fields.text = step.text;
|
||||
if (step.channel) fields.channel = step.channel;
|
||||
if (step.replyInThread) fields.reply_in_thread = true;
|
||||
break;
|
||||
case "send_dm":
|
||||
if (step.to) fields.to = step.to;
|
||||
@@ -274,6 +276,7 @@ export function yamlToFormState(
|
||||
duration: step.duration as string | undefined,
|
||||
text: step.text as string | undefined,
|
||||
channel: step.channel as string | undefined,
|
||||
replyInThread: step.reply_in_thread === true,
|
||||
to: step.to as string | undefined,
|
||||
url: step.url as string | undefined,
|
||||
method: step.method as string | undefined,
|
||||
|
||||
Reference in New Issue
Block a user