From 5fcdefd52e3ec8718897ce4caf1402fdc9889f72 Mon Sep 17 00:00:00 2001 From: Duncan Date: Mon, 17 Aug 2026 19:58:09 -0400 Subject: [PATCH] fix(workflows): align relay reply threading with ingest resolver Pass-1 review fixes for workflow reply_in_thread: - Metadata-less parents: a workflow reply onto a marked-but-unindexed parent recovers the parent's root/reply ancestry from its NIP-10 tags (depth 2) instead of assuming top-level (depth 1). Extracts the ancestry derivation into a shared helper used by both the client (resolve_nip10_thread_meta) and workflow (resolve_relay_reply_thread_meta) resolvers so the two cannot diverge. - trigger_is_reply now requires a NIP-10 reply marker, matching ingest's treatment of (root=Some, reply=None) as top-level. A lone root marker no longer counts as a reply. - A relay-built threaded reply now pushes the live kind:39005 thread-summary overlay after insert, exactly as the human ingest path does, so desktops update the root's badge without refetching. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- crates/buzz-relay/src/handlers/ingest.rs | 105 +++++++++----- crates/buzz-relay/src/workflow_sink.rs | 156 +++++++++++++++++++++ crates/buzz-test-client/tests/e2e_relay.rs | 107 ++++++++++++++ crates/buzz-workflow/src/lib.rs | 52 ++++++- 4 files changed, 379 insertions(+), 41 deletions(-) diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index 8c37e977d..1de9824d2 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -807,46 +807,18 @@ pub(crate) async fn resolve_nip10_thread_meta( (effective_root, root_ts, depth) } None => { - let parent_root = parent_event - .event - .tags - .iter() - .find_map(|t| { - let parts = t.as_slice(); - if parts.len() >= 4 && parts[0] == "e" && parts[3] == "root" { - hex::decode(&parts[1]).ok().filter(|b| b.len() == 32) - } else { - None - } - }) - .or_else(|| { - parent_event.event.tags.iter().find_map(|t| { - let parts = t.as_slice(); - if parts.len() >= 4 && parts[0] == "e" && parts[3] == "reply" { - hex::decode(&parts[1]).ok().filter(|b| b.len() == 32) - } else { - None - } - }) - }) - .unwrap_or_else(|| parent_bytes.clone()); + let (parent_root, root_created, depth) = derive_ancestry_from_parent_tags( + community_id, + &parent_event.event, + &parent_bytes, + parent_created, + state, + ) + .await; if client_root_bytes != parent_root { return Err("root tag does not match thread ancestry".to_string()); } - let depth = if parent_root == parent_bytes { 1 } else { 2 }; - let root_created = if parent_root != parent_bytes { - if let Ok(Some(root_ev)) = - state.db.get_event_by_id(community_id, &parent_root).await - { - chrono::DateTime::from_timestamp(root_ev.event.created_at.as_secs() as i64, 0) - .unwrap_or(parent_created) - } else { - parent_created - } - } else { - parent_created - }; (parent_root, root_created, depth) } }; @@ -872,6 +844,52 @@ pub(crate) async fn resolve_nip10_thread_meta( })) } +/// Recover a reply's thread ancestry from its *parent's* NIP-10 tags when the +/// parent has **no** `thread_metadata` row (legacy or not-yet-indexed events). +/// +/// The parent's own marked `root` (else `reply`) `e` tag names the thread root; +/// absent both, the parent is itself top-level and is its own root. Depth is 1 +/// when the parent is the root and 2 otherwise — a reply to a nested-but- +/// unindexed parent must not be mistaken for a top-level reply. +/// +/// Shared by [`resolve_nip10_thread_meta`] (client path) and +/// [`resolve_relay_reply_thread_meta`] (workflow path) so the two cannot +/// diverge. Returns `(root_event_id, root_event_created_at, depth)`. +async fn derive_ancestry_from_parent_tags( + community_id: CommunityId, + parent_event: &Event, + parent_bytes: &[u8], + parent_created: chrono::DateTime, + state: &AppState, +) -> (Vec, chrono::DateTime, i32) { + let marked_ancestor = |marker: &str| { + parent_event.tags.iter().find_map(|t| { + let parts = t.as_slice(); + if parts.len() >= 4 && parts[0] == "e" && parts[3] == marker { + hex::decode(&parts[1]).ok().filter(|b| b.len() == 32) + } else { + None + } + }) + }; + let parent_root = marked_ancestor("root") + .or_else(|| marked_ancestor("reply")) + .unwrap_or_else(|| parent_bytes.to_vec()); + + if parent_root.as_slice() == parent_bytes { + (parent_root, parent_created, 1) + } else { + let root_created = + if let Ok(Some(root_ev)) = state.db.get_event_by_id(community_id, &parent_root).await { + chrono::DateTime::from_timestamp(root_ev.event.created_at.as_secs() as i64, 0) + .unwrap_or(parent_created) + } else { + parent_created + }; + (parent_root, root_created, 2) + } +} + /// Resolved thread ancestry for a relay-built reply (workflow path). /// /// Carries the parent and root identifiers plus the reply's depth, so the @@ -977,8 +995,19 @@ pub(crate) async fn resolve_relay_reply_thread_meta( }; (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), + // No metadata row ⇒ recover the parent's ancestry from its own NIP-10 + // tags. A marked (but not-yet-indexed) nested parent yields depth 2, not + // a false top-level depth 1. + None => { + derive_ancestry_from_parent_tags( + community_id, + &parent_event.event, + &parent_bytes, + parent_created, + state, + ) + .await + } }; if depth > 100 { diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index d07ca7d1d..05fac180a 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/crates/buzz-relay/src/workflow_sink.rs @@ -404,6 +404,20 @@ impl ActionSink for RelayActionSink { None, ) .await; + + // A threaded reply changed its thread's counters — push a fresh + // relay-signed kind:39005 so subscribed clients update badge + // counts without refetching the head window, exactly as the + // ingest path does after a reply insert. Fan-out-only and + // best-effort; skipped for top-level (non-reply) messages. + if let Some(owned) = &thread_meta_owned { + crate::handlers::side_effects::emit_live_thread_summary( + &tenant, + &state, + channel_uuid, + owned.root_event_id.clone(), + ); + } } Ok(event_id_hex) @@ -872,6 +886,148 @@ mod integration_tests { assert_eq!(meta.root_event_id.as_deref(), Some(root_bytes.as_slice())); } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn workflow_reply_to_metadata_less_nested_parent_recovers_depth_2() { + // A parent that carries NIP-10 root/reply markers but has NO + // thread_metadata row (legacy or not-yet-indexed) must be recognized as + // nested: the workflow reply threads at depth 2 onto the parent's own + // root, not a false top-level depth 1. + let state = test_state().await; + + let author = nostr::Keys::generate(); + let author_hex = author.public_key().to_hex(); + + let host = format!("wf-legacy-{}.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-legacy", + ChannelType::Stream, + ChannelVisibility::Open, + None, + &author.public_key().to_bytes(), + None, + ) + .await + .expect("create channel"); + + let channel_hex = channel.id.to_string(); + + // A top-level root message, inserted WITHOUT any thread metadata row. + let root_event = EventBuilder::new(Kind::from(KIND_STREAM_MESSAGE as u16), "root") + .tags([Tag::parse(["h", &channel_hex]).expect("h tag")]) + .sign_with_keys(&author) + .expect("sign root"); + let root_hex = root_event.id.to_hex(); + state + .db + .insert_event(community, &root_event, Some(channel.id)) + .await + .expect("insert root"); + + // A nested parent that marks its root/reply — but, crucially, is stored + // with NO thread_metadata row (the legacy/unindexed case F1 addresses). + let parent_event = + EventBuilder::new(Kind::from(KIND_STREAM_MESSAGE as u16), "nested parent") + .tags([ + Tag::parse(["h", &channel_hex]).expect("h tag"), + Tag::parse(["e", &root_hex, "", "root"]).expect("root tag"), + Tag::parse(["e", &root_hex, "", "reply"]).expect("reply tag"), + ]) + .sign_with_keys(&author) + .expect("sign parent"); + let parent_hex = parent_event.id.to_hex(); + state + .db + .insert_event(community, &parent_event, Some(channel.id)) + .await + .expect("insert parent"); + assert!( + state + .db + .get_thread_metadata_by_event(community, parent_event.id.as_bytes()) + .await + .expect("query parent meta") + .is_none(), + "test premise: the nested parent must have no thread_metadata row" + ); + + // A workflow reply onto the metadata-less nested parent. + let reply_hex = RelayActionSink::new(&state) + .send_message( + community, + &channel_hex, + "workflow reply", + &author_hex, + Some(&parent_hex), + ) + .await + .expect("send reply"); + + let reply_id_bytes = nostr::EventId::from_hex(&reply_hex) + .expect("reply id") + .as_bytes() + .to_vec(); + 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, 2, + "reply to a marked-but-unindexed nested parent is depth 2, not top-level" + ); + let root_bytes = nostr::EventId::from_hex(&root_hex) + .expect("root id") + .as_bytes() + .to_vec(); + let parent_bytes = parent_event.id.as_bytes().to_vec(); + assert_eq!( + meta.root_event_id.as_deref(), + Some(root_bytes.as_slice()), + "root recovered from the parent's own NIP-10 markers" + ); + assert_eq!( + meta.parent_event_id.as_deref(), + Some(parent_bytes.as_slice()) + ); + + // The reply's own NIP-10 e-tags point root→the recovered root, + // reply→the immediate parent (matching the ingest resolver). + let stored = state + .db + .get_event_by_id(community, &reply_id_bytes) + .await + .expect("query reply") + .expect("reply persisted"); + let marker = |m: &str| -> Option { + 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(parent_hex.as_str())); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn workflow_reply_to_missing_parent_errors() { diff --git a/crates/buzz-test-client/tests/e2e_relay.rs b/crates/buzz-test-client/tests/e2e_relay.rs index 882cbabfe..18ea05825 100644 --- a/crates/buzz-test-client/tests/e2e_relay.rs +++ b/crates/buzz-test-client/tests/e2e_relay.rs @@ -2661,6 +2661,113 @@ async fn test_reply_ingest_pushes_live_thread_summary() { client.disconnect().await.expect("disconnect"); } +/// F3 (workflow path): a `message_posted` workflow whose `send_message` action +/// has `reply_in_thread: true` posts a threaded reply to the triggering +/// top-level message — and that relay-built reply must push the same live +/// kind:39005 thread-summary overlay the human ingest path does, so desktops +/// update the root's badge without refetching. Also exercises F2's semantics: +/// the `trigger_is_reply == false` filter must fire on the top-level message. +#[tokio::test] +#[ignore] +async fn test_workflow_reply_in_thread_pushes_live_thread_summary() { + let url = relay_url(); + let http = relay_http_url(); + let keys = Keys::generate(); + let pubkey_hex = keys.public_key().to_hex(); + let channel = create_test_channel(&keys).await; + + // A message_posted workflow that replies in-thread, but only to NEW + // top-level messages (`trigger_is_reply == false`) — so it cannot recurse + // on the reply it just posted. + let yaml = format!( + "name: reply-bot\n\ + description: F3 live probe\n\ + trigger:\n\ + \x20 on: message_posted\n\ + \x20 filter: \"trigger_is_reply == false\"\n\ + steps:\n\ + \x20 - id: step1\n\ + \x20 name: Reply\n\ + \x20 action: send_message\n\ + \x20 text: \"auto-reply\"\n\ + \x20 reply_in_thread: true\n" + ); + let def = EventBuilder::new(Kind::Custom(30620), yaml) + .tags([ + Tag::parse(["d", &Uuid::new_v4().to_string()]).unwrap(), + Tag::parse(["h", channel.as_str()]).unwrap(), + Tag::parse(["name", "reply-bot"]).unwrap(), + ]) + .sign_with_keys(&keys) + .expect("sign workflow def"); + let client = reqwest::Client::new(); + let resp = client + .post(format!("{http}/events")) + .header("X-Pubkey", &pubkey_hex) + .header("Content-Type", "application/json") + .body(serde_json::to_string(&def).unwrap()) + .send() + .await + .expect("submit workflow def"); + let body: serde_json::Value = resp.json().await.expect("parse def response"); + assert!( + body["accepted"].as_bool().unwrap_or(false), + "workflow def not accepted: {body}" + ); + + // Live 39005 subscription for the channel, shaped like the desktop window + // store's. + let mut ws = BuzzTestClient::connect(&url, &keys).await.expect("connect"); + let sid = sub_id("wf-live-summary"); + let filter = Filter::new() + .kind(Kind::Custom(39005)) + .custom_tags(SingleLetterTag::lowercase(Alphabet::H), [channel.as_str()]); + ws.subscribe(&sid, vec![filter]).await.expect("subscribe"); + ws.collect_until_eose(&sid, Duration::from_secs(5)) + .await + .expect("EOSE"); + + // Post a top-level message — the workflow fires and posts a threaded reply. + let root = EventBuilder::new(Kind::Custom(9), "trigger me") + .tags([Tag::parse(["h", channel.as_str()]).unwrap()]) + .sign_with_keys(&keys) + .expect("sign root"); + let root_id = root.id; + let ok = ws.send_event(root).await.expect("send root"); + assert!(ok.accepted, "root rejected: {}", ok.message); + + // The workflow reply's 39005 overlay must arrive and target the root with a + // reply_count of 1 — proving the relay-built reply pushed the live summary. + let summary = loop { + match ws + .recv_event(Duration::from_secs(10)) + .await + .expect("recv 39005 for workflow reply") + { + RelayMessage::Event { event, .. } if event.kind == Kind::Custom(39005) => break *event, + _ => continue, + } + }; + let root_tag_val = summary + .tags + .iter() + .find(|t| t.as_slice().first().map(String::as_str) == Some("e")) + .and_then(|t| t.content().map(str::to_string)) + .expect("summary carries root e-tag"); + assert_eq!( + root_tag_val, + root_id.to_hex(), + "workflow-reply summary targets the triggering top-level message as root" + ); + let content: serde_json::Value = serde_json::from_str(&summary.content).expect("JSON"); + assert_eq!( + content["reply_count"], 1, + "workflow threaded reply counted up: {content}" + ); + + ws.disconnect().await.expect("disconnect"); +} + /// Read a member's authoritative role from the relay-signed kind:39002 member /// list. The relay's own view of membership, not the client's — a kind:9000 can /// be `accepted` (stored) while its membership side effect fails, so asserting diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index bc068191b..3bc0e409c 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -1022,12 +1022,14 @@ pub fn build_trigger_context(event: &buzz_core::StoredEvent) -> executor::Trigge } /// 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. +/// `reply`. A `root` marker alone is not enough: the ingest resolver treats +/// `(root=Some, reply=None)` as top-level (see `resolve_nip10_thread_meta`), so +/// a `message_posted` filter of `trigger_is_reply == false` fires on those and +/// on messages with no NIP-10 markers at all. 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") + parts.len() >= 4 && parts[0] == "e" && parts[3] == "reply" }) } @@ -1603,6 +1605,50 @@ steps: assert!(ctx.is_reply, "message with reply/root e-tags is a reply"); } + #[test] + fn build_trigger_context_is_reply_true_for_reply_only_marker() { + // A NIP-10 `reply` marker without a `root` marker (the fallback ingest + // treats as `root == reply`) is still a threaded reply. + use nostr::{EventBuilder, Keys, Kind, Tag}; + use uuid::Uuid; + let parent = Keys::generate(); + let parent_event = EventBuilder::new(Kind::Custom(9), "parent") + .sign_with_keys(&parent) + .expect("sign parent"); + let keys = Keys::generate(); + let event = EventBuilder::new(Kind::Custom(9), "reply only") + .tags([Tag::parse(["e", &parent_event.id.to_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, "a lone `reply` marker is a reply"); + } + + #[test] + fn build_trigger_context_is_reply_false_for_root_only_marker() { + // Ingest treats `(root=Some, reply=None)` as top-level, so + // `event_is_reply` must too — otherwise `trigger_is_reply == false` + // would skip a message the relay stored as a new top-level post. + use nostr::{EventBuilder, Keys, Kind, Tag}; + use uuid::Uuid; + let root = Keys::generate(); + let root_event = EventBuilder::new(Kind::Custom(9), "root") + .sign_with_keys(&root) + .expect("sign root"); + let keys = Keys::generate(); + let event = EventBuilder::new(Kind::Custom(9), "root marker only") + .tags([Tag::parse(["e", &root_event.id.to_hex(), "", "root"]).expect("root 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, + "a lone `root` marker is top-level to ingest, not 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