External/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-08-18 06:50:31 +02:00
Code Issues Packages Projects Releases Wiki Activity

fix(workflows): align relay reply threading with ingest resolver

Browse Source 
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 <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
Duncan
2026-08-17 19:58:09 -04:00
co-authored by Will Pfleger
parent ecbf20a66f
commit 5fcdefd52e
4 changed files with 379 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/buzz-relay/src/handlers/ingest.rs
+67 -38
View File
@@ -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<Utc>,
state: &AppState,
) -> (Vec<u8>, chrono::DateTime<Utc>, 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 {
crates/buzz-relay/src/workflow_sink.rs
+156
View File
@@ -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<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(parent_hex.as_str()));
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn workflow_reply_to_missing_parent_errors() {
crates/buzz-test-client/tests/e2e_relay.rs
+107
View File
@@ -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
crates/buzz-workflow/src/lib.rs
+49 -3
View File
@@ -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