diff --git a/crates/buzz-core/src/kind.rs b/crates/buzz-core/src/kind.rs index 3c6f1d5913..32f296cc27 100644 --- a/crates/buzz-core/src/kind.rs +++ b/crates/buzz-core/src/kind.rs @@ -437,6 +437,9 @@ pub const KIND_THREAD_SUMMARY: u32 = 39005; /// content = `{has_more, next_cursor}`. The only authority on exhaustion — /// clients must not infer `has_more` from row counts. pub const KIND_WINDOW_BOUNDS: u32 = 39006; +/// Thread page bounds overlay: `e` tag = root event id, content = +/// `{has_more, next_cursor}`. Synthesized at query time and never stored. +pub const KIND_THREAD_BOUNDS: u32 = 39007; /// Workflow definition (parameterized replaceable, d=workflow_uuid). pub const KIND_WORKFLOW_DEF: u32 = 30620; @@ -692,6 +695,7 @@ pub const ALL_KINDS: &[u32] = &[ KIND_NIP29_GROUP_ROLES, KIND_THREAD_SUMMARY, KIND_WINDOW_BOUNDS, + KIND_THREAD_BOUNDS, KIND_PRESENCE_UPDATE, KIND_TYPING_INDICATOR, KIND_HUDDLE_REACTION, @@ -836,6 +840,7 @@ pub const fn is_relay_only_kind(kind: u32) -> bool { | KIND_DM_VISIBILITY | KIND_THREAD_SUMMARY | KIND_WINDOW_BOUNDS + | KIND_THREAD_BOUNDS ) } @@ -864,6 +869,7 @@ const _: () = assert!(is_parameterized_replaceable(KIND_DM_VISIBILITY)); // 3062 const _: () = assert!(is_parameterized_replaceable(KIND_PROJECT)); // 30621 ∈ 30000–39999 const _: () = assert!(is_parameterized_replaceable(KIND_THREAD_SUMMARY)); // 39005 ∈ 30000–39999 const _: () = assert!(is_parameterized_replaceable(KIND_WINDOW_BOUNDS)); // 39006 ∈ 30000–39999 +const _: () = assert!(is_parameterized_replaceable(KIND_THREAD_BOUNDS)); // 39007 ∈ 30000–39999 // Compile-time: NIP-34 parameterized replaceable kinds are in the correct range. const _: () = assert!( diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 1ba0909bbf..349bd40a15 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -2910,7 +2910,45 @@ impl Db { depth_limit: Option, limit: u32, cursor: Option<&[u8]>, - ) -> Result> { + ) -> Result { + self.get_thread_replies_ordered( + community_id, + root_event_id, + depth_limit, + limit, + cursor, + thread::ThreadOrder::Oldest, + ) + .await + } + + /// Fetch replies under a root event in the requested traversal order. + pub async fn get_thread_replies_ordered( + &self, + community_id: CommunityId, + root_event_id: &[u8], + depth_limit: Option, + limit: u32, + cursor: Option<&[u8]>, + order: thread::ThreadOrder, + ) -> Result { + if order == thread::ThreadOrder::Newest { + // Descending pages can contain rows newer than their tail cursor; + // the ascending fence proof cannot establish that shape safely. + // Keep newest-first traversal writer-authoritative. + Self::record_route("thread_newest", "writer", "predicate"); + return thread::get_thread_replies_ordered( + &self.pool, + community_id, + root_event_id, + depth_limit, + limit, + cursor, + order, + ) + .await; + } + let (path, predicate): (&'static str, RoutePredicate) = match cursor { Some(_) => ( "thread_cursor", @@ -2923,29 +2961,34 @@ impl Db { if let RouteDecision::Replica(mut tx, entry, reason) = self.route_read(path, predicate).await { - match thread::get_thread_replies_on( + match thread::get_thread_replies_ordered_on( &mut tx, community_id, root_event_id, depth_limit, limit, cursor, + order, ) .await { - Ok(replies) => { + Ok(page) => { if cursor.is_none() { // Predicate A: bounded-stale head page, served as proved. Self::record_route(path, "replica", reason); - return Ok(replies); + return Ok(page); } - let full = replies.len() >= limit as usize; - let below_fence = replies - .last() - .is_some_and(|tail| tail.created_at <= entry.fence_wall); - if full && below_fence { + // Only a non-terminal page whose raw scan cursor is at or + // below the proved wall can be trusted from a replica. A + // delivered-event tail is not authoritative: corrupt rows + // may make it empty or older than the actual scan position. + let below_fence = page + .next_cursor + .as_ref() + .is_some_and(|(ts, _)| *ts <= entry.fence_wall); + if page.has_more && below_fence { Self::record_route(path, "replica", reason); - return Ok(replies); + return Ok(page); } // Candidate terminal page, or page reaching above the // proved wall — verify against the writer. Recorded as @@ -2965,13 +3008,14 @@ impl Db { } } } - thread::get_thread_replies( + thread::get_thread_replies_ordered( &self.pool, community_id, root_event_id, depth_limit, limit, cursor, + order, ) .await } diff --git a/crates/buzz-db/src/thread.rs b/crates/buzz-db/src/thread.rs index 007677e258..f00e796508 100644 --- a/crates/buzz-db/src/thread.rs +++ b/crates/buzz-db/src/thread.rs @@ -15,6 +15,46 @@ use crate::{error::Result, event::row_to_stored_event}; // -- Structs ------------------------------------------------------------------ +/// Traversal order for a thread page. +#[derive(Debug, Clone, Copy, Default, Eq, PartialEq)] +pub enum ThreadOrder { + /// Preserve the historical oldest-to-newest traversal. + #[default] + Oldest, + /// Traverse newest-to-oldest so recent replies can be presented first. + Newest, +} + +/// One authoritative page of thread replies. +#[derive(Debug, Clone)] +pub struct ThreadPage { + /// Reconstructable replies from the scanned raw rows. + pub replies: Vec, + /// Whether another raw row exists beyond this page. + pub has_more: bool, + /// Composite scan cursor `(created_at, event_id)` of the last raw row + /// retained for this page. Captured before reconstruction, so corrupt rows + /// cannot stall or prematurely terminate traversal. `Some` iff `has_more`. + pub next_cursor: Option<(DateTime, Vec)>, +} + +impl std::ops::Deref for ThreadPage { + type Target = [ThreadReply]; + + fn deref(&self) -> &Self::Target { + &self.replies + } +} + +impl IntoIterator for ThreadPage { + type Item = ThreadReply; + type IntoIter = std::vec::IntoIter; + + fn into_iter(self) -> Self::IntoIter { + self.replies.into_iter() + } +} + /// A single reply within a thread, joined with event content. #[derive(Debug, Clone)] pub struct ThreadReply { @@ -349,15 +389,38 @@ pub async fn get_thread_replies( depth_limit: Option, limit: u32, cursor: Option<&[u8]>, -) -> Result> { +) -> Result { + get_thread_replies_ordered( + pool, + community_id, + root_event_id, + depth_limit, + limit, + cursor, + ThreadOrder::Oldest, + ) + .await +} + +/// Fetch a thread page in the requested traversal order. +pub async fn get_thread_replies_ordered( + pool: &PgPool, + community_id: CommunityId, + root_event_id: &[u8], + depth_limit: Option, + limit: u32, + cursor: Option<&[u8]>, + order: ThreadOrder, +) -> Result { let mut conn = pool.acquire().await?; - get_thread_replies_on( + get_thread_replies_ordered_on( &mut conn, community_id, root_event_id, depth_limit, limit, cursor, + order, ) .await } @@ -366,14 +429,15 @@ pub async fn get_thread_replies( /// runs the page on the exact reader connection whose heartbeat observation /// proved coverage (the proof is connection-local; a different pooled /// session may sit at a different replay position). -pub(crate) async fn get_thread_replies_on( +pub(crate) async fn get_thread_replies_ordered_on( conn: &mut sqlx::PgConnection, community_id: CommunityId, root_event_id: &[u8], depth_limit: Option, limit: u32, cursor: Option<&[u8]>, -) -> Result> { + order: ThreadOrder, +) -> Result { // Decode cursor bytes -> keyset (timestamp, optional event_id) for the // WHERE condition. Layout: 8-byte BE i64 seconds, then the raw event_id. // An 8-byte-only cursor is legacy timestamp-only paging (no tiebreak). @@ -430,25 +494,37 @@ pub(crate) async fn get_thread_replies_on( } match &cursor_key { Some((_, Some(_))) => { - // Composite keyset: strict row comparison with an event_id tiebreak - // so same-second replies paginate without gaps or duplicates. + // Composite keyset with an event_id tiebreak so same-second replies + // paginate without gaps or duplicates in either direction. let ts_idx = param_idx; let id_idx = param_idx + 1; + let op = match order { + ThreadOrder::Oldest => ">", + ThreadOrder::Newest => "<", + }; sql.push_str(&format!( - " AND (tm.event_created_at, tm.event_id) > (${ts_idx}, ${id_idx})" + " AND (tm.event_created_at, tm.event_id) {op} (${ts_idx}, ${id_idx})" )); param_idx += 2; } Some((_, None)) => { // Legacy timestamp-only cursor (no tiebreak). - sql.push_str(&format!(" AND tm.event_created_at > ${param_idx}")); + let op = match order { + ThreadOrder::Oldest => ">", + ThreadOrder::Newest => "<", + }; + sql.push_str(&format!(" AND tm.event_created_at {op} ${param_idx}")); param_idx += 1; } None => {} } + let direction = match order { + ThreadOrder::Oldest => "ASC", + ThreadOrder::Newest => "DESC", + }; sql.push_str(&format!( - " ORDER BY tm.event_created_at ASC, tm.event_id ASC LIMIT ${param_idx}" + " ORDER BY tm.event_created_at {direction}, tm.event_id {direction} LIMIT ${param_idx}" )); let mut q = sqlx::query(sqlx::AssertSqlSafe(sql)) @@ -467,9 +543,23 @@ pub(crate) async fn get_thread_replies_on( } None => {} } - q = q.bind(limit as i32); + // One extra raw row is the server's authoritative evidence of continuation. + q = q.bind(limit as i64 + 1); - let rows = q.fetch_all(&mut *conn).await?; + let mut rows = q.fetch_all(&mut *conn).await?; + let has_more = rows.len() > limit as usize; + rows.truncate(limit as usize); + let next_cursor = if has_more { + match rows.last() { + Some(row) => Some(( + row.try_get::, _>("event_created_at")?, + row.try_get::, _>("event_id")?, + )), + None => None, + } + } else { + None + }; let mut replies = Vec::with_capacity(rows.len()); for row in rows { @@ -506,7 +596,11 @@ pub(crate) async fn get_thread_replies_on( }); } - Ok(replies) + Ok(ThreadPage { + replies, + has_more, + next_cursor, + }) } /// Fetch aggregated thread stats for a single event, plus up to 10 participant pubkeys. @@ -1215,6 +1309,37 @@ mod tests { let mut expected_sorted = expected_ids.clone(); expected_sorted.sort(); assert_eq!(unique, expected_sorted, "paged set != inserted tied set"); + + // The optional reverse traversal must converge over the same tied set + // while issuing cursors from the last raw row of each descending page. + let mut newest_ids = Vec::new(); + let mut newest_cursor: Option> = None; + loop { + let page = get_thread_replies_ordered( + &pool, + community, + root.id.as_bytes(), + Some(10), + page_limit, + newest_cursor.as_deref(), + ThreadOrder::Newest, + ) + .await + .expect("fetch newest page"); + newest_ids.extend(page.replies.iter().map(|reply| reply.event_id.clone())); + if !page.has_more { + assert!(page.next_cursor.is_none()); + break; + } + let (ts, id) = page.next_cursor.expect("has_more implies cursor"); + let mut next = ts.timestamp().to_be_bytes().to_vec(); + next.extend_from_slice(&id); + newest_cursor = Some(next); + } + let mut newest_unique = newest_ids; + newest_unique.sort(); + newest_unique.dedup(); + assert_eq!(newest_unique, expected_sorted); } /// Nested replies (depth >= 2) must be reachable in a subtree read. Every @@ -1417,7 +1542,15 @@ mod tests { .expect("insert root event"); // Two replies under the same root: one stays valid, one we corrupt. - let good = make_stream_event(&author, "good"); + // Pin ordering so the corrupt row is the entire first raw page and the + // valid row is beyond it. This exercises zero delivered events with + // `has_more = true`, not merely skip-and-continue within one page. + let bad_secs = root.created_at.as_secs() + 1; + let good_secs = bad_secs + 1; + let good = EventBuilder::new(Kind::Custom(9), "good") + .custom_created_at(nostr::Timestamp::from(good_secs)) + .sign_with_keys(&author) + .expect("sign good reply"); let good_id = good.id.to_hex(); let good_created_at = event_created_at(&good); insert_event_with_thread_metadata( @@ -1440,7 +1573,10 @@ mod tests { .await .expect("insert good reply"); - let bad = make_stream_event(&author, "bad"); + let bad = EventBuilder::new(Kind::Custom(9), "bad") + .custom_created_at(nostr::Timestamp::from(bad_secs)) + .sign_with_keys(&author) + .expect("sign bad reply"); let bad_created_at = event_created_at(&bad); insert_event_with_thread_metadata( &pool, @@ -1474,15 +1610,40 @@ mod tests { .rows_affected(); assert_eq!(rows_changed, 1, "expected to corrupt exactly one row"); - let replies = get_thread_replies(&pool, community, root.id.as_bytes(), Some(10), 10, None) + let first = get_thread_replies(&pool, community, root.id.as_bytes(), Some(10), 1, None) .await - .expect("fetch thread replies must succeed despite a corrupt row"); + .expect("fetch corrupt first page"); + assert!( + first.replies.is_empty(), + "corrupt raw row must not be delivered" + ); + assert!( + first.has_more, + "the valid row beyond the corrupt page remains" + ); + let (cursor_ts, cursor_id) = first + .next_cursor + .expect("has_more page must expose its raw scan cursor"); + assert_eq!(cursor_ts, bad_created_at); + assert_eq!(cursor_id, bad.id.as_bytes()); - // The corrupt reply is skipped; the valid one survives. The whole - // query does NOT 500 on a single unreconstructable row. - assert_eq!(replies.len(), 1); - assert_eq!(replies[0].stored_event.event.id.to_hex(), good_id); - assert_eq!(replies[0].stored_event.event.content, "good"); + let mut cursor = cursor_ts.timestamp().to_be_bytes().to_vec(); + cursor.extend_from_slice(&cursor_id); + let second = get_thread_replies( + &pool, + community, + root.id.as_bytes(), + Some(10), + 1, + Some(&cursor), + ) + .await + .expect("fetch valid second page"); + assert!(!second.has_more); + assert!(second.next_cursor.is_none()); + assert_eq!(second.replies.len(), 1); + assert_eq!(second.replies[0].stored_event.event.id.to_hex(), good_id); + assert_eq!(second.replies[0].stored_event.event.content, "good"); } /// Insert one top-level event (root metadata, broadcast) into a channel. diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index dfce484494..10532afdd6 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -286,6 +286,19 @@ fn extract_depth_limit(raw: &Value) -> Option { .and_then(|n| u32::try_from(n).ok()) } +/// Optional thread traversal order. Unknown or absent values preserve the +/// historical oldest-first behavior. +fn extract_thread_order(raw: &Value) -> buzz_db::thread::ThreadOrder { + match raw + .get("thread_order") + .or_else(|| raw.get("threadOrder")) + .and_then(Value::as_str) + { + Some("newest") => buzz_db::thread::ThreadOrder::Newest, + _ => buzz_db::thread::ThreadOrder::Oldest, + } +} + /// Extract a thread pagination cursor from the raw filter JSON. /// /// The desktop pages `get_thread_replies` forward with a keyset cursor derived @@ -1179,21 +1192,24 @@ async fn query_events_authed( let limit = filter .limit .unwrap_or(100) - .min(BRIDGE_THREAD_MAX_LIMIT as usize) as u32; + .min(BRIDGE_THREAD_MAX_LIMIT as usize) + .max(1) as u32; let thread_cursor = extract_thread_cursor(raw); - let thread_replies = state + let thread_order = extract_thread_order(raw); + let thread_page = state .db - .get_thread_replies( + .get_thread_replies_ordered( tenant.community(), &root_bytes, Some(depth), limit, thread_cursor.as_deref(), + thread_order, ) .await .map_err(|e| internal_error(&format!("thread query error: {e}")))?; - for reply in thread_replies { + for reply in thread_page.replies { let se = reply.stored_event; if !event_in_accessible_channel(&se, &accessible_channels) { continue; @@ -1208,6 +1224,45 @@ async fn query_events_authed( events.push(v); } } + + // Relay-signed, query-time pagination authority. The cursor describes + // the raw scan position, not the last event that happened to survive + // reconstruction or authorization filtering. + let next_cursor = thread_page.next_cursor.as_ref().map(|(ts, id)| { + serde_json::json!({ + "created_at": ts.timestamp(), + "id": hex::encode(id), + }) + }); + let content = serde_json::json!({ + "has_more": thread_page.has_more, + "next_cursor": next_cursor, + }); + let order = match thread_order { + buzz_db::thread::ThreadOrder::Oldest => "oldest", + buzz_db::thread::ThreadOrder::Newest => "newest", + }; + let request_cursor = thread_cursor + .as_deref() + .map(hex::encode) + .unwrap_or_else(|| "head".to_owned()); + let tags = vec![ + nostr::Tag::parse(["e", root_hex]) + .map_err(|e| internal_error(&format!("thread bounds e tag: {e}")))?, + nostr::Tag::parse(["d", &format!("{root_hex}:{order}:{request_cursor}")]) + .map_err(|e| internal_error(&format!("thread bounds d tag: {e}")))?, + ]; + let overlay = nostr::EventBuilder::new( + nostr::Kind::Custom(buzz_core::kind::KIND_THREAD_BOUNDS as u16), + content.to_string(), + ) + .tags(tags) + .sign_with_keys(&state.relay_keypair) + .map_err(|e| internal_error(&format!("thread bounds sign: {e}")))?; + events.push( + serde_json::to_value(&overlay) + .map_err(|e| internal_error(&format!("thread bounds serialize: {e}")))?, + ); handled.insert(idx); } @@ -3072,6 +3127,28 @@ mod tests { assert_eq!(extract_depth_limit(&raw), Some(3)); } + #[test] + fn extract_thread_order_defaults_oldest_and_accepts_both_field_styles() { + use buzz_db::thread::ThreadOrder; + + assert_eq!( + extract_thread_order(&serde_json::json!({})), + ThreadOrder::Oldest + ); + assert_eq!( + extract_thread_order(&serde_json::json!({ "thread_order": "newest" })), + ThreadOrder::Newest + ); + assert_eq!( + extract_thread_order(&serde_json::json!({ "threadOrder": "newest" })), + ThreadOrder::Newest + ); + assert_eq!( + extract_thread_order(&serde_json::json!({ "thread_order": "future" })), + ThreadOrder::Oldest + ); + } + #[test] fn extract_thread_cursor_valid() { // Timestamp-only cursor: 8-byte BE seconds, no tiebreak id.