From 2e5602a442780e32aea8fab3c8ba2b34198e24c7 Mon Sep 17 00:00:00 2001 From: Riley Crane Date: Thu, 3 Sep 2026 15:43:44 -0700 Subject: [PATCH] =?UTF-8?q?accumulator:=20thread=20selections=20=E2=80=94?= =?UTF-8?q?=20anchor=20+=20all=20descendant=20replies?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Selection gains a threads clause: 64-hex event ids, each selecting itself plus every descendant reply at any depth, resolved by a recursive CTE over the mirror's new parent column (NIP-10 e-tag linkage). channels and threads union as scope; authors/kinds/window intersect on top. Legacy mirrors gain the column in place, backfilled once from the raw JSON already on disk — no relay re-fetch. GET /events/{id} now returns parent + thread_root for pickers. One-off comparison recipe documented in the README. Co-Authored-By: Claude Fable 5 Signed-off-by: Riley Crane --- crates/buzz-accumulator/README.md | 30 +- crates/buzz-accumulator/src/daemon/folds.rs | 4 + crates/buzz-accumulator/src/daemon/http.rs | 10 +- crates/buzz-accumulator/src/daemon/store.rs | 411 +++++++++++++++++++- crates/buzz-accumulator/src/daemon/sync.rs | 2 + crates/buzz-accumulator/src/selection.rs | 54 ++- crates/buzz-accumulator/ui/src/App.tsx | 1 + crates/buzz-accumulator/ui/src/api.ts | 3 + 8 files changed, 497 insertions(+), 18 deletions(-) diff --git a/crates/buzz-accumulator/README.md b/crates/buzz-accumulator/README.md index a7126814ad..40f880bd19 100644 --- a/crates/buzz-accumulator/README.md +++ b/crates/buzz-accumulator/README.md @@ -10,9 +10,16 @@ Signals — raw relay events, never rewritten — are the substrate. On top of them: - **Selection** — a frozen-or-live description of a signal set: who × what × - when (`channels`, `authors`, `kinds`, plus the selection's own `since` / - `until_exclusive`). A pinned `until_exclusive` **freezes** it; an open end - means **live** — "and whatever comes next". + when (`channels`, `authors`, `threads`, `kinds`, plus the selection's own + `since` / `until_exclusive`). A pinned `until_exclusive` **freezes** it; an + open end means **live** — "and whatever comes next". + `threads` holds event ids: each anchor selects **itself plus every + descendant reply at any depth** (an anchor need not be a thread root — a + mid-thread id selects that sub-subtree). `channels` and `threads` **union** + as scope; `authors` and `kinds` intersect on top. Subtrees are resolved + from the mirror's reply linkage (`parent` column, NIP-10 `e` tags) at + materialization time, so a live thread selection keeps picking up new + replies. - **Fold** — name + selection + model + instructions: a **factory for artifacts**. Each run computes `artifact' = fold(artifact, new_signals)`. A frozen selection makes the fold run until its set is covered, then it is @@ -92,7 +99,7 @@ so any non-loopback bind must add authentication first. |---|---| | `GET /status` | Connection, backfill, mirror counts. | | `GET /channels` | Discovered channels. | -| `GET /events/{id}` | One mirrored event (the citation-chip endpoint). | +| `GET /events/{id}` | One mirrored event (the citation-chip endpoint), with `parent` + `thread_root` reply lineage. | | `POST /select/preview` | `{selection, since?, until_exclusive?}` → count, size, daily rhythm. $0. | | `POST /select/events` | Same body + `limit`, `after` keyset cursor → the actual events, paged. | | `GET /folds` | All fold specs. | @@ -117,6 +124,21 @@ the key follows (the mirror only ever contains those), and "participated in since X" is `{authors: [], since: X}`. Resolve, then create the fold. +### One-off artifacts (compare a thread to a channel) + +A one-off is just a frozen fold you never touch again — no permanent digest +chain required first: + +1. `GET /events/{id}` on any message in the thread → `thread_root`. +2. `PUT /folds/compare--` with a **frozen** selection and a + purpose-specific prompt: + `{selection: {threads: [""], channels: [""], + until_exclusive: }, model, instructions}`. +3. `POST .../preflight` ($0) to see size/cost, then `POST .../run` until + `coverage.complete` (large sets chunk across runs, honestly). +4. Read the artifact; the fold is done forever — every later preflight says + *cached*. + ## Layout - `src/` — pure engine (selection, spec, plan/complete run, estimate, diff --git a/crates/buzz-accumulator/src/daemon/folds.rs b/crates/buzz-accumulator/src/daemon/folds.rs index bfe083030f..8116104b84 100644 --- a/crates/buzz-accumulator/src/daemon/folds.rs +++ b/crates/buzz-accumulator/src/daemon/folds.rs @@ -337,6 +337,7 @@ mod tests { created_at: 100, content: "hello world".into(), raw: "{}".into(), + parent: None, }]) .await .expect("seed signal"); @@ -446,6 +447,7 @@ mod tests { created_at: 5_000, content: "after the freeze".into(), raw: "{}".into(), + parent: None, }]) .await .expect("late event"); @@ -488,6 +490,7 @@ mod tests { created_at: 1_000 + i as i64, content: "m".repeat(5_000), raw: "{}".into(), + parent: None, }) .collect(); store.upsert_events(&backlog).await.expect("seed backlog"); @@ -569,6 +572,7 @@ mod tests { created_at: 9_000, content: "the new arrival".into(), raw: "{}".into(), + parent: None, }]) .await .expect("late event"); diff --git a/crates/buzz-accumulator/src/daemon/http.rs b/crates/buzz-accumulator/src/daemon/http.rs index c531ea88cd..3be2729a98 100644 --- a/crates/buzz-accumulator/src/daemon/http.rs +++ b/crates/buzz-accumulator/src/daemon/http.rs @@ -247,7 +247,14 @@ async fn get_event( .await? .ok_or_else(|| ApiError(StatusCode::NOT_FOUND, format!("event not found: {id}")))?; let names = s.store.names(&BTreeSet::from([sig.pubkey.clone()])).await?; - Ok(Json(event_json(&sig, names.get(&sig.pubkey)))) + let mut body = event_json(&sig, names.get(&sig.pubkey)); + // Reply lineage for thread pickers: the direct parent and the topmost + // mirrored ancestor. `thread_root == id` means the event starts a thread. + if let Some((parent, root)) = s.store.thread_lineage(&sig.id).await? { + body["parent"] = json!(parent); + body["thread_root"] = json!(root); + } + Ok(Json(body)) } fn event_json(sig: &crate::Signal, author_name: Option<&String>) -> Value { @@ -656,6 +663,7 @@ mod tests { created_at: ts, content: format!("msg-{id_char}"), raw: "{}".into(), + parent: None, } } diff --git a/crates/buzz-accumulator/src/daemon/store.rs b/crates/buzz-accumulator/src/daemon/store.rs index 339807cf11..3d149f6c47 100644 --- a/crates/buzz-accumulator/src/daemon/store.rs +++ b/crates/buzz-accumulator/src/daemon/store.rs @@ -32,6 +32,8 @@ pub struct StoredEvent { pub content: String, /// Full raw event JSON as received from the relay. pub raw: String, + /// Direct reply parent (NIP-10 `e` tag), 64-hex, if the event is a reply. + pub parent: Option, } /// A discovered channel and its backfill bookkeeping. @@ -84,12 +86,18 @@ CREATE TABLE IF NOT EXISTS events ( kind INTEGER NOT NULL, created_at INTEGER NOT NULL, content TEXT NOT NULL, - raw TEXT NOT NULL + raw TEXT NOT NULL, + parent TEXT ); CREATE INDEX IF NOT EXISTS idx_events_channel_time ON events(channel, created_at); CREATE INDEX IF NOT EXISTS idx_events_kind_time ON events(kind, created_at); CREATE INDEX IF NOT EXISTS idx_events_author ON events(pubkey); +CREATE TABLE IF NOT EXISTS meta ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL +); + CREATE TABLE IF NOT EXISTS channels ( id TEXT PRIMARY KEY, name TEXT, @@ -143,7 +151,73 @@ impl Store { .connect_with(options) .await?; sqlx::raw_sql(SCHEMA).execute(&pool).await?; - Ok(Self { pool }) + let store = Self { pool }; + store.migrate_parent_column().await?; + Ok(store) + } + + /// Adds the `parent` column to mirrors created before threads existed and + /// backfills it from the raw event JSON already on disk — no relay + /// re-fetch, no reset. Idempotent: the backfill runs once, guarded by a + /// meta flag. + async fn migrate_parent_column(&self) -> Result<(), sqlx::Error> { + let has_parent = sqlx::query( + "SELECT COUNT(*) AS n FROM pragma_table_info('events') WHERE name = 'parent'", + ) + .fetch_one(&self.pool) + .await? + .get::("n") + > 0; + if !has_parent { + sqlx::raw_sql("ALTER TABLE events ADD COLUMN parent TEXT") + .execute(&self.pool) + .await?; + } + // The index lives here, not in SCHEMA: on a legacy mirror the column + // must exist before the index can. + sqlx::raw_sql("CREATE INDEX IF NOT EXISTS idx_events_parent ON events(parent)") + .execute(&self.pool) + .await?; + let done = sqlx::query("SELECT value FROM meta WHERE key = 'parent_backfill'") + .fetch_optional(&self.pool) + .await? + .map(|r| r.get::("value")); + if done.as_deref() == Some("done") { + return Ok(()); + } + // Page by id keyset so a large mirror is never fully in memory. + let mut last_id = String::new(); + loop { + let rows = + sqlx::query("SELECT id, raw FROM events WHERE id > ? ORDER BY id LIMIT 2000") + .bind(&last_id) + .fetch_all(&self.pool) + .await?; + if rows.is_empty() { + break; + } + let mut tx = self.pool.begin().await?; + for r in &rows { + let id: String = r.get("id"); + let raw: String = r.get("raw"); + if let Some(parent) = parent_from_raw(&raw) { + sqlx::query("UPDATE events SET parent = ? WHERE id = ?") + .bind(parent) + .bind(&id) + .execute(&mut *tx) + .await?; + } + last_id = id; + } + tx.commit().await?; + } + sqlx::query( + "INSERT INTO meta (key, value) VALUES ('parent_backfill', 'done') + ON CONFLICT(key) DO UPDATE SET value = excluded.value", + ) + .execute(&self.pool) + .await?; + Ok(()) } // ---- events ---- @@ -154,8 +228,8 @@ impl Store { let mut tx = self.pool.begin().await?; for ev in events { let done = sqlx::query( - "INSERT OR IGNORE INTO events (id, channel, pubkey, kind, created_at, content, raw) - VALUES (?, ?, ?, ?, ?, ?, ?)", + "INSERT OR IGNORE INTO events (id, channel, pubkey, kind, created_at, content, raw, parent) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)", ) .bind(&ev.id) .bind(&ev.channel) @@ -164,6 +238,7 @@ impl Store { .bind(ev.created_at) .bind(&ev.content) .bind(&ev.raw) + .bind(&ev.parent) .execute(&mut *tx) .await?; inserted += done.rows_affected(); @@ -253,6 +328,49 @@ impl Store { Ok(row.map(row_to_signal)) } + /// Reply lineage of one event: its direct parent and its thread root — + /// the topmost *mirrored* ancestor (an event whose own parent is absent + /// or unknown to the mirror). Returns `None` when the event itself is + /// unknown. Cycle-safe. + pub async fn thread_lineage( + &self, + id: &str, + ) -> Result, String)>, sqlx::Error> { + let rows = sqlx::query( + "WITH RECURSIVE up(id, parent) AS ( + SELECT id, parent FROM events WHERE id = ? + UNION + SELECT e.id, e.parent FROM events e JOIN up ON e.id = up.parent + ) + SELECT id, parent FROM up", + ) + .bind(id) + .fetch_all(&self.pool) + .await?; + let chain: BTreeMap> = rows + .into_iter() + .map(|r| { + ( + r.get::("id"), + r.get::, _>("parent"), + ) + }) + .collect(); + if !chain.contains_key(id) { + return Ok(None); + } + let parent = chain.get(id).cloned().flatten(); + let mut root = id.to_string(); + let mut seen = BTreeSet::new(); + while let Some(Some(p)) = chain.get(&root) { + if !chain.contains_key(p) || !seen.insert(p.clone()) { + break; + } + root = p.clone(); + } + Ok(Some((parent, root))) + } + // ---- channels ---- /// Registers a channel if unknown, seeding the backfill cursor at @@ -519,24 +637,54 @@ impl Store { } /// Base signal query: columns + window + selection predicates, no ordering. +/// +/// Scope rule: `channels` and `threads` union — an event is in scope when it +/// is in a listed channel OR in the subtree (anchor + descendants at any +/// depth) of a listed thread anchor. `authors` and `kinds` intersect on top. +/// The subtree is resolved by a recursive CTE over the mirror's `parent` +/// links; `UNION` (not `UNION ALL`) makes it terminate even on a pathological +/// parent cycle. fn signal_query( selection: &Selection, since: i64, until_exclusive: i64, ) -> QueryBuilder { - let mut qb: QueryBuilder = QueryBuilder::new( + let mut qb: QueryBuilder = QueryBuilder::new(""); + if !selection.threads.is_empty() { + qb.push("WITH RECURSIVE thread_scope(id) AS (SELECT id FROM events WHERE id IN ("); + let mut sep = qb.separated(", "); + for t in &selection.threads { + sep.push_bind(t); + } + qb.push(") UNION SELECT e.id FROM events e JOIN thread_scope ts ON e.parent = ts.id) "); + } + qb.push( "SELECT id, channel, pubkey, kind, created_at, content FROM events WHERE created_at >= ", ); qb.push_bind(since); qb.push(" AND created_at < "); qb.push_bind(until_exclusive); - if !selection.channels.is_empty() { - qb.push(" AND channel IN ("); - let mut sep = qb.separated(", "); - for c in &selection.channels { - sep.push_bind(c); + match (selection.channels.is_empty(), selection.threads.is_empty()) { + (false, false) => { + qb.push(" AND (channel IN ("); + let mut sep = qb.separated(", "); + for c in &selection.channels { + sep.push_bind(c); + } + qb.push(") OR id IN (SELECT id FROM thread_scope))"); } - qb.push(")"); + (false, true) => { + qb.push(" AND channel IN ("); + let mut sep = qb.separated(", "); + for c in &selection.channels { + sep.push_bind(c); + } + qb.push(")"); + } + (true, false) => { + qb.push(" AND id IN (SELECT id FROM thread_scope)"); + } + (true, true) => {} } if !selection.authors.is_empty() { qb.push(" AND pubkey IN ("); @@ -555,6 +703,46 @@ fn signal_query( qb } +/// Extracts the direct reply parent from raw event JSON (used by the one-time +/// backfill; live ingest parses typed tags in `sync`). +fn parent_from_raw(raw: &str) -> Option { + let value: serde_json::Value = serde_json::from_str(raw).ok()?; + let tags: Vec> = value + .get("tags")? + .as_array()? + .iter() + .filter_map(|t| { + t.as_array().map(|items| { + items + .iter() + .filter_map(|v| v.as_str().map(str::to_string)) + .collect() + }) + }) + .collect(); + parent_from_tag_rows(&tags) +} + +/// NIP-10 direct-parent resolution over `e` tags: prefer the `reply`-marked +/// tag; else the `root`-marked one (a direct reply to the root carries only +/// that); else the last positional `e` tag. +pub(crate) fn parent_from_tag_rows(tags: &[Vec]) -> Option { + let e_tags: Vec<&Vec> = tags + .iter() + .filter(|t| t.first().map(String::as_str) == Some("e") && t.get(1).is_some()) + .collect(); + let marked = |m: &str| { + e_tags + .iter() + .find(|t| t.get(3).map(String::as_str) == Some(m)) + .and_then(|t| t.get(1)) + }; + marked("reply") + .or_else(|| marked("root")) + .or_else(|| e_tags.last().and_then(|t| t.get(1))) + .map(|id| id.trim().to_ascii_lowercase()) +} + fn row_to_signal(r: sqlx::sqlite::SqliteRow) -> Signal { Signal { id: r.get("id"), @@ -579,9 +767,21 @@ mod tests { created_at: ts, content: format!("content-{id}"), raw: "{}".into(), + parent: None, } } + fn reply(id: &str, parent: &str, channel: Option<&str>, ts: i64) -> StoredEvent { + StoredEvent { + parent: Some(parent.repeat(64 / parent.len().max(1))), + ..ev(id, channel, 9, ts) + } + } + + fn full_id(short: &str) -> String { + short.repeat(64 / short.len().max(1)) + } + fn selection(channels: &[&str]) -> Selection { Selection { channels: channels.iter().map(|s| s.to_string()).collect(), @@ -621,6 +821,195 @@ mod tests { assert_eq!(got[0].channel.as_deref(), Some("cha")); } + /// A three-deep thread in `cha` (root → reply → reply-to-reply), a + /// bystander in `cha`, and an event in `chb`. + async fn thread_fixture(store: &Store) { + store + .upsert_events(&[ + ev("a1", Some("cha"), 9, 100), // thread root + reply("b2", "a1", Some("cha"), 110), // depth 1 + reply("c3", "b2", Some("cha"), 120), // depth 2 (sub-thread) + ev("d4", Some("cha"), 9, 130), // same channel, not in thread + ev("e5", Some("chb"), 9, 140), // other channel + ]) + .await + .expect("insert fixture"); + } + + #[tokio::test] + async fn thread_selection_takes_anchor_plus_all_descendants() { + let store = Store::open(":memory:").await.expect("open"); + thread_fixture(&store).await; + let sel = Selection { + threads: vec![full_id("a1")], + ..Selection::default() + }; + let got = store.query_signals(&sel, 0, 1_000).await.expect("query"); + let ids: Vec = got.iter().map(|s| s.id.clone()).collect(); + assert_eq!(ids, vec![full_id("a1"), full_id("b2"), full_id("c3")]); + // A mid-thread anchor selects only that sub-subtree. + let sel = Selection { + threads: vec![full_id("b2")], + ..Selection::default() + }; + let got = store.query_signals(&sel, 0, 1_000).await.expect("query"); + let ids: Vec = got.iter().map(|s| s.id.clone()).collect(); + assert_eq!(ids, vec![full_id("b2"), full_id("c3")]); + } + + #[tokio::test] + async fn threads_and_channels_union_while_filters_intersect() { + let store = Store::open(":memory:").await.expect("open"); + thread_fixture(&store).await; + // Riley's comparison shape: thread A + all of channel B. + let sel = Selection { + threads: vec![full_id("a1")], + channels: vec!["chb".into()], + ..Selection::default() + }; + let got = store.query_signals(&sel, 0, 1_000).await.expect("query"); + let ids: Vec = got.iter().map(|s| s.id.clone()).collect(); + assert_eq!( + ids, + vec![full_id("a1"), full_id("b2"), full_id("c3"), full_id("e5")] + ); + // The window still clamps subtree members. + let got = store.query_signals(&sel, 105, 115).await.expect("query"); + let ids: Vec = got.iter().map(|s| s.id.clone()).collect(); + assert_eq!(ids, vec![full_id("b2")]); + // Authors intersect: a different author matches nothing. + let sel = Selection { + threads: vec![full_id("a1")], + authors: vec!["f".repeat(64)], + ..Selection::default() + }; + assert!(store + .query_signals(&sel, 0, 1_000) + .await + .expect("query") + .is_empty()); + // An anchor the mirror has never seen selects nothing (not an error). + let sel = Selection { + threads: vec!["9".repeat(64)], + ..Selection::default() + }; + assert!(store + .query_signals(&sel, 0, 1_000) + .await + .expect("query") + .is_empty()); + } + + #[tokio::test] + async fn thread_lineage_resolves_parent_and_root() { + let store = Store::open(":memory:").await.expect("open"); + thread_fixture(&store).await; + let (parent, root) = store + .thread_lineage(&full_id("c3")) + .await + .expect("query") + .expect("known event"); + assert_eq!(parent.as_deref(), Some(full_id("b2").as_str())); + assert_eq!(root, full_id("a1")); + let (parent, root) = store + .thread_lineage(&full_id("a1")) + .await + .expect("query") + .expect("known event"); + assert_eq!(parent, None); + assert_eq!(root, full_id("a1")); + assert!(store + .thread_lineage(&"9".repeat(64)) + .await + .expect("query") + .is_none()); + } + + #[tokio::test] + async fn legacy_mirror_gains_parent_column_and_backfills_from_raw() { + // A pre-threads mirror: no `parent` column, no meta table, raw JSON on + // disk. Opening the store must migrate it in place — no relay fetch. + let path = std::env::temp_dir().join(format!( + "accumulator-parent-migration-{}.db", + std::process::id() + )); + let path_str = path.to_str().expect("utf8 temp path").to_string(); + let _ = std::fs::remove_file(&path); + { + let options = SqliteConnectOptions::new() + .filename(&path_str) + .create_if_missing(true); + let pool = SqlitePoolOptions::new() + .max_connections(1) + .connect_with(options) + .await + .expect("open legacy"); + sqlx::raw_sql( + "CREATE TABLE events ( + id TEXT PRIMARY KEY, channel TEXT, pubkey TEXT NOT NULL, + kind INTEGER NOT NULL, created_at INTEGER NOT NULL, + content TEXT NOT NULL, raw TEXT NOT NULL + )", + ) + .execute(&pool) + .await + .expect("legacy schema"); + let root = full_id("a1"); + let raw_reply = format!( + r#"{{"id":"{}","tags":[["h","cha"],["e","{root}","","reply"]]}}"#, + full_id("b2") + ); + for (id, raw) in [(root.clone(), "{}".to_string()), (full_id("b2"), raw_reply)] { + sqlx::query( + "INSERT INTO events (id, channel, pubkey, kind, created_at, content, raw) + VALUES (?, 'cha', ?, 9, 100, 'x', ?)", + ) + .bind(&id) + .bind("a".repeat(64)) + .bind(&raw) + .execute(&pool) + .await + .expect("legacy insert"); + } + pool.close().await; + } + let store = Store::open(&path_str).await.expect("migrating open"); + let sel = Selection { + threads: vec![full_id("a1")], + ..Selection::default() + }; + let got = store.query_signals(&sel, 0, 1_000).await.expect("query"); + let ids: Vec = got.iter().map(|s| s.id.clone()).collect(); + assert_eq!(ids, vec![full_id("a1"), full_id("b2")]); + let _ = std::fs::remove_file(&path); + } + + #[test] + fn parent_resolution_prefers_reply_then_root_then_last_positional() { + let root = full_id("a1"); + let mid = full_id("b2"); + let tag = |name: &str, rest: &[&str]| -> Vec { + std::iter::once(name) + .chain(rest.iter().copied()) + .map(str::to_string) + .collect() + }; + // Marked reply wins over marked root. + let tags = vec![ + tag("e", &[&root, "", "root"]), + tag("e", &[&mid, "", "reply"]), + ]; + assert_eq!(parent_from_tag_rows(&tags), Some(mid.clone())); + // Only a root marker: that IS the direct parent. + let tags = vec![tag("e", &[&root, "", "root"])]; + assert_eq!(parent_from_tag_rows(&tags), Some(root.clone())); + // Unmarked positional tags: last one is the parent (NIP-10 legacy). + let tags = vec![tag("e", &[&root]), tag("e", &[&mid])]; + assert_eq!(parent_from_tag_rows(&tags), Some(mid.clone())); + // No e tags at all. + assert_eq!(parent_from_tag_rows(&[tag("h", &["cha"])]), None); + } + #[tokio::test] async fn artifact_pk_is_a_version_fence() { let store = Store::open(":memory:").await.expect("open"); diff --git a/crates/buzz-accumulator/src/daemon/sync.rs b/crates/buzz-accumulator/src/daemon/sync.rs index 582ea3279f..f0be0e4d38 100644 --- a/crates/buzz-accumulator/src/daemon/sync.rs +++ b/crates/buzz-accumulator/src/daemon/sync.rs @@ -568,6 +568,7 @@ async fn verify_and_map(events: Vec) -> Vec { /// Distills a verified relay event into its storage row. pub(crate) fn event_to_stored(ev: &nostr::Event) -> StoredEvent { + let tag_rows: Vec> = ev.tags.iter().map(|t| t.as_slice().to_vec()).collect(); StoredEvent { id: ev.id.to_hex(), channel: tag_value(ev, "h").map(str::to_string), @@ -576,6 +577,7 @@ pub(crate) fn event_to_stored(ev: &nostr::Event) -> StoredEvent { created_at: ev.created_at.as_secs() as i64, content: ev.content.clone(), raw: serde_json::to_string(ev).unwrap_or_default(), + parent: super::store::parent_from_tag_rows(&tag_rows), } } diff --git a/crates/buzz-accumulator/src/selection.rs b/crates/buzz-accumulator/src/selection.rs index 7758001c1d..96880fdde3 100644 --- a/crates/buzz-accumulator/src/selection.rs +++ b/crates/buzz-accumulator/src/selection.rs @@ -25,6 +25,13 @@ pub struct Selection { /// Author pubkeys (hex). #[serde(default, skip_serializing_if = "Vec::is_empty")] pub authors: Vec, + /// Thread anchors: 64-hex event ids, each selecting itself plus every + /// descendant reply at any depth. Joins `channels` as a scope union — an + /// event is in scope when it is in a listed channel OR a listed subtree. + /// An anchor need not be a thread root: a mid-thread id selects that + /// sub-subtree. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub threads: Vec, /// Event kinds; empty means [`DEFAULT_SIGNAL_KINDS`]. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub kinds: Vec, @@ -48,15 +55,20 @@ impl Selection { for a in &mut self.authors { *a = a.trim().to_ascii_lowercase(); } + for t in &mut self.threads { + *t = t.trim().to_ascii_lowercase(); + } self.channels.sort(); self.channels.dedup(); self.authors.sort(); self.authors.dedup(); + self.threads.sort(); + self.threads.dedup(); self.kinds.sort_unstable(); self.kinds.dedup(); - if self.channels.is_empty() && self.authors.is_empty() { + if self.channels.is_empty() && self.authors.is_empty() && self.threads.is_empty() { return Err(Error::InvalidSpec( - "selection must name at least one channel or author".into(), + "selection must name at least one channel, author, or thread".into(), )); } if let Some(c) = self.channels.iter().find(|c| !is_uuid(c)) { @@ -69,6 +81,11 @@ impl Selection { "author {a:?} is not a 64-hex pubkey" ))); } + if let Some(t) = self.threads.iter().find(|t| !is_hex_id(t)) { + return Err(Error::InvalidSpec(format!( + "thread {t:?} is not a 64-hex event id" + ))); + } if let Some(k) = self.kinds.iter().find(|k| **k > u16::MAX as u32) { return Err(Error::InvalidSpec(format!( "kind {k} is outside the event-kind range" @@ -137,6 +154,11 @@ fn is_hex_pubkey(s: &str) -> bool { s.len() == 64 && s.bytes().all(|c| matches!(c, b'0'..=b'9' | b'a'..=b'f')) } +/// Lowercase 64-hex event id (same alphabet as a pubkey). +fn is_hex_id(s: &str) -> bool { + is_hex_pubkey(s) +} + /// Deterministically materialize fetched signals: `created_at ASC, id ASC`, /// duplicates (same id) removed. /// @@ -261,6 +283,34 @@ mod tests { assert_eq!(s.resolve_window(Some(150), Some(180), 10_000), (150, 180)); } + #[test] + fn threads_clause_canonicalizes_and_satisfies_the_scope_rule() { + let root = "F".repeat(64); + let mut s = Selection { + threads: vec![root.clone(), root.to_ascii_lowercase()], + ..Selection::default() + }; + s.canonicalize().expect("threads alone are a valid scope"); + assert_eq!(s.threads, vec![root.to_ascii_lowercase()]); + + let mut bad = Selection { + threads: vec!["not-an-id".into()], + ..Selection::default() + }; + assert!(bad.canonicalize().is_err()); + } + + #[test] + fn selection_json_without_threads_still_loads_and_compares_equal() { + // Specs saved before the threads clause must keep loading and stay + // cache-compatible with a threads-less selection. + let legacy = format!(r#"{{"channels":["{CH_A}"]}}"#); + let loaded: Selection = serde_json::from_str(&legacy).expect("deserialize"); + assert_eq!(loaded, channels(&[CH_A])); + let out = serde_json::to_string(&loaded).expect("serialize"); + assert!(!out.contains("threads"), "empty threads must not serialize"); + } + #[test] fn selection_json_without_window_still_loads_as_live() { // Specs saved before the when moved into the selection must keep diff --git a/crates/buzz-accumulator/ui/src/App.tsx b/crates/buzz-accumulator/ui/src/App.tsx index 1fe9ee3729..028bfb87f5 100644 --- a/crates/buzz-accumulator/ui/src/App.tsx +++ b/crates/buzz-accumulator/ui/src/App.tsx @@ -24,6 +24,7 @@ function fromUrl(): { selection: Selection; window: TimeWindow } { selection: { channels: list("channels"), authors: list("authors"), + threads: list("threads"), kinds: list("kinds") .map((k) => Number.parseInt(k, 10)) .filter((k) => Number.isFinite(k) && k >= 0), diff --git a/crates/buzz-accumulator/ui/src/api.ts b/crates/buzz-accumulator/ui/src/api.ts index f240975641..da69eb899f 100644 --- a/crates/buzz-accumulator/ui/src/api.ts +++ b/crates/buzz-accumulator/ui/src/api.ts @@ -3,6 +3,8 @@ export type Selection = { channels: string[]; authors: string[]; + /** Thread anchors: each event id selects itself + all descendant replies. */ + threads: string[]; kinds: number[]; /** Selection's own window (who × what × when). A pinned `until_exclusive` * freezes the selection; open ends mean live ("and whatever comes next"). */ @@ -180,6 +182,7 @@ function normalizeSelection(s: Partial | undefined): Selection { return { channels: s?.channels ?? [], authors: s?.authors ?? [], + threads: s?.threads ?? [], kinds: s?.kinds ?? [], since: s?.since, until_exclusive: s?.until_exclusive,