diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 4714ce1343..19bbf9997c 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -1232,7 +1232,13 @@ async fn query_events_authed( } return tokio::time::timeout( thread_window::DEADLINE, - thread_window::query_batch(state, tenant, &pubkey, thread_windows.iter().flatten()), + thread_window::query_batch( + state, + tenant, + &pubkey, + thread_windows.iter().flatten(), + None, + ), ) .await .map_err(|_| { diff --git a/crates/buzz-relay/src/api/bridge/thread_window.rs b/crates/buzz-relay/src/api/bridge/thread_window.rs index 793a541c0d..b866bb99cc 100644 --- a/crates/buzz-relay/src/api/bridge/thread_window.rs +++ b/crates/buzz-relay/src/api/bridge/thread_window.rs @@ -87,17 +87,21 @@ fn append(events: &mut Vec, budget: &mut Budget, event: &nostr::Event) -> /// Authorize the entire batch against one writer access set, then refresh it /// once before releasing any output. A later window must never suppress only /// its own rows while releasing an earlier window built before revocation. -pub(super) async fn query_batch<'a>( +pub(crate) async fn query_batch<'a>( state: &AppState, tenant: &TenantContext, reader: &nostr::PublicKey, requests: impl IntoIterator, + allowed_channels: Option<&[uuid::Uuid]>, ) -> Result, Error> { - let accessible = state + let mut accessible = state .db .get_accessible_channel_ids(tenant.community(), &reader.to_bytes()) .await .map_err(|e| database_error("access", e))?; + if let Some(allowed) = allowed_channels { + accessible.retain(|channel| allowed.contains(channel)); + } let mut budget = Budget::default(); let mut events = Vec::new(); for request in requests { @@ -105,11 +109,14 @@ pub(super) async fn query_batch<'a>( events.extend(query(state, tenant, reader, request, &accessible, &mut budget).await?); } } - let current = state + let mut current = state .db .get_accessible_channel_ids(tenant.community(), &reader.to_bytes()) .await .map_err(|e| database_error("final access", e))?; + if let Some(allowed) = allowed_channels { + current.retain(|channel| allowed.contains(channel)); + } // Grants can expose auxiliary events omitted from the original closure; // revocations can invalidate earlier windows or their cross-channel aux. if accessible.iter().collect::>() != current.iter().collect::>() { diff --git a/crates/buzz-relay/src/connection.rs b/crates/buzz-relay/src/connection.rs index 12a450861e..95fd5a34a9 100644 --- a/crates/buzz-relay/src/connection.rs +++ b/crates/buzz-relay/src/connection.rs @@ -902,6 +902,7 @@ async fn handle_text_message(text: String, conn: Arc, state: Ar ClientMessage::Req { sub_id, filters, + raw_filters, before_ids, } => { let conn = Arc::clone(&conn); @@ -919,7 +920,15 @@ async fn handle_text_message(text: String, conn: Arc, state: Ar let span = tracing::info_span!("ws.req", conn_id = %conn.conn_id, sub_id = %sub_id); tokio::spawn( async move { - handlers::req::handle_req(sub_id, filters, before_ids, conn, state).await; + handlers::req::handle_req( + sub_id, + filters, + raw_filters, + before_ids, + conn, + state, + ) + .await; drop(permit); } .instrument(span), diff --git a/crates/buzz-relay/src/handlers/req.rs b/crates/buzz-relay/src/handlers/req.rs index 97d252cb7e..f0e6601d2d 100644 --- a/crates/buzz-relay/src/handlers/req.rs +++ b/crates/buzz-relay/src/handlers/req.rs @@ -15,7 +15,8 @@ use buzz_core::tenant::TenantContext; use buzz_db::EventQuery; use buzz_pubsub::EventTopic; use hex; -use nostr::Filter; +use nostr::{Filter, PublicKey}; +use serde_json::Value; use buzz_auth::Scope; @@ -52,6 +53,7 @@ const _: () = assert!(FILTER_QUERY_CONCURRENCY >= 2 && FILTER_QUERY_CONCURRENCY pub async fn handle_req( sub_id: String, filters: Vec, + raw_filters: Vec, before_ids: Vec>>, conn: Arc, state: Arc, @@ -94,6 +96,69 @@ pub async fn handle_req( } }; + let thread_windows = match crate::api::bridge::thread_window::parse(&raw_filters) { + Ok(windows) => windows, + Err((status, body)) => { + close_thread_window_error(&conn, &sub_id, status, &body); + return; + } + }; + if thread_windows.iter().any(Option::is_some) { + if thread_windows.iter().any(Option::is_none) { + conn.send(RelayMessage::closed( + &sub_id, + "invalid: thread_window cannot mix with other query modes", + )); + return; + } + let reader = match PublicKey::from_slice(&pubkey_bytes) { + Ok(reader) => reader, + Err(error) => { + warn!(conn_id = %conn_id, %error, "Authenticated reader key became invalid"); + conn.send(RelayMessage::closed( + &sub_id, + "error: invalid authenticated reader", + )); + return; + } + }; + // A finite REQ replaces any live subscription with the same id before + // its first row can be sent. Release its fan-out topic as CLOSE does. + conn.subscriptions.lock().await.remove(&sub_id); + if let Some(replaced) = state.sub_registry.remove_subscription(conn_id, &sub_id) { + release_subscription_topics(&state, &conn.tenant, &replaced.scope).await; + } + let result = tokio::time::timeout( + crate::api::bridge::thread_window::DEADLINE, + serve_thread_windows( + &sub_id, + thread_windows.iter().flatten(), + token_channel_ids.as_deref(), + &reader, + &conn, + &state, + ), + ) + .await; + match result { + Ok(Ok(true)) => { + conn.send(RelayMessage::eose(&sub_id)); + } + Ok(Ok(false)) => { + // A failed enqueue is not a complete page. Do not send EOSE: + // the client must retry rather than accept truncated bounds. + } + Ok(Err((status, body))) => close_thread_window_error(&conn, &sub_id, status, &body), + Err(_) => { + conn.send(RelayMessage::closed( + &sub_id, + "error: thread window deadline exceeded", + )); + } + } + return; + } + let channel_id = extract_channel_id_from_filters(&filters); let requested_channel_ids = match extract_channel_ids_from_filters_limited(&filters) { Ok(ids) => ids, @@ -493,6 +558,49 @@ pub async fn handle_req( ); } +async fn serve_thread_windows<'a>( + sub_id: &str, + requests: impl Iterator, + allowed_channels: Option<&[uuid::Uuid]>, + reader: &PublicKey, + conn: &ConnectionState, + state: &AppState, +) -> Result { + let events = crate::api::bridge::thread_window::query_batch( + state, + &conn.tenant, + reader, + requests, + allowed_channels, + ) + .await?; + for event in events { + if !conn.send(serde_json::json!(["EVENT", sub_id, event]).to_string()) { + return Ok(false); + } + } + Ok(true) +} + +fn close_thread_window_error( + conn: &ConnectionState, + sub_id: &str, + status: axum::http::StatusCode, + body: &axum::Json, +) { + let detail = body + .0 + .get("error") + .and_then(Value::as_str) + .unwrap_or("thread window failed"); + let prefix = if status.is_client_error() { + "invalid" + } else { + "error" + }; + conn.send(RelayMessage::closed(sub_id, &format!("{prefix}: {detail}"))); +} + /// FTS candidate hits fetched per page. Pages are always full regardless of /// the requested limit — post-filtering discards an unpredictable share of /// hits, so the scan fetches candidates in full pages rather than sizing diff --git a/crates/buzz-relay/src/protocol.rs b/crates/buzz-relay/src/protocol.rs index 5832a72a87..d9ddd7fd47 100644 --- a/crates/buzz-relay/src/protocol.rs +++ b/crates/buzz-relay/src/protocol.rs @@ -22,6 +22,8 @@ pub enum ClientMessage { sub_id: String, /// The filters that determine which events are delivered. filters: Vec, + /// Raw filters retained because `nostr::Filter` drops extension fields. + raw_filters: Vec, /// Optional per-filter composite cursor tiebreaks from raw extension fields. before_ids: Vec>>, }, @@ -108,6 +110,12 @@ impl ClientMessage { let before_ids = filter_values .iter() .map(|value| { + // Thread-mode cursor validation belongs to the REQ handler, + // where invalid input can terminate this subscription with + // CLOSED rather than leaving a bare NOTICE. + if value.get("thread_window").is_some_and(|v| v == true) { + return Ok(None); + } let Some(raw) = value.get("before_id") else { return Ok(None); }; @@ -137,6 +145,7 @@ impl ClientMessage { Ok(ClientMessage::Req { sub_id, filters, + raw_filters: filter_values.to_vec(), before_ids, }) } @@ -341,6 +350,38 @@ mod tests { } } + #[test] + fn parse_req_preserves_thread_window_extension_fields() { + let raw = serde_json::json!([ + "REQ", + "thread-window", + { + "thread_window": true, + "#h": [uuid::Uuid::nil()], + "#e": ["ab".repeat(32)], + "kinds": [9], + "depth_limit": 42, + "include_aux": true, + } + ]) + .to_string(); + + match ClientMessage::parse(&raw).unwrap() { + ClientMessage::Req { + filters, + raw_filters, + .. + } => { + assert_eq!(filters.len(), 1); + assert_eq!(raw_filters.len(), 1); + assert_eq!(raw_filters[0]["thread_window"], true); + assert_eq!(raw_filters[0]["depth_limit"], 42); + assert_eq!(raw_filters[0]["include_aux"], true); + } + _ => panic!("expected Req"), + } + } + #[test] fn parse_req_composite_cursor_preserves_filter_alignment() { let raw = serde_json::json!([ diff --git a/crates/buzz-test-client/tests/e2e_relay.rs b/crates/buzz-test-client/tests/e2e_relay.rs index a801c1aaae..5e354f4459 100644 --- a/crates/buzz-test-client/tests/e2e_relay.rs +++ b/crates/buzz-test-client/tests/e2e_relay.rs @@ -2719,6 +2719,86 @@ async fn test_private_channel_member_cannot_grant_admin() { member_client.disconnect().await.expect("disconnect member"); } +/// WebSocket REQ preserves thread-window extension fields and serves the same +/// newest-first row + signed-bounds contract as POST /query. +#[tokio::test] +#[ignore] +async fn test_thread_window_req_returns_rows_bounds_and_eose() { + let url = relay_url(); + let keys = Keys::generate(); + let channel = create_test_channel(&keys).await; + let mut client = BuzzTestClient::connect(&url, &keys).await.expect("connect"); + + let root = EventBuilder::new(Kind::Custom(9), "thread-window root") + .tags([Tag::parse(["h", channel.as_str()]).unwrap()]) + .sign_with_keys(&keys) + .expect("sign root"); + let root_id = root.id; + let ok = client.send_event(root).await.expect("send root"); + assert!(ok.accepted, "root rejected: {}", ok.message); + + let mut expected = Vec::new(); + for content in ["older reply", "newer reply"] { + let reply = EventBuilder::new(Kind::Custom(9), content) + .tags([ + Tag::parse(["h", channel.as_str()]).unwrap(), + Tag::parse(["e", &root_id.to_hex(), "", "reply"]).unwrap(), + ]) + .sign_with_keys(&keys) + .expect("sign reply"); + expected.push(reply.clone()); + let ok = client.send_event(reply).await.expect("send reply"); + assert!(ok.accepted, "reply rejected: {}", ok.message); + } + expected.sort_by(|left, right| { + right + .created_at + .cmp(&left.created_at) + .then_with(|| left.id.cmp(&right.id)) + }); + + let sid = sub_id("thread-window"); + client + .send_raw(&serde_json::json!([ + "REQ", + sid, + { + "thread_window": true, + "#h": [channel], + "#e": [root_id.to_hex()], + "kinds": [9], + "limit": 10, + } + ])) + .await + .expect("send thread-window REQ"); + let events = client + .collect_until_eose(&sid, Duration::from_secs(10)) + .await + .expect("collect thread-window response through EOSE"); + + let rows: Vec<_> = events + .iter() + .filter(|event| event.kind == Kind::Custom(9)) + .collect(); + assert_eq!( + rows.iter().map(|event| event.id).collect::>(), + expected.iter().map(|event| event.id).collect::>(), + "thread rows must use created_at DESC, id ASC order" + ); + let bounds: Vec<_> = events + .iter() + .filter(|event| event.kind == Kind::Custom(39007)) + .collect(); + assert_eq!(bounds.len(), 1, "served window must have one bounds event"); + let content: serde_json::Value = + serde_json::from_str(&bounds[0].content).expect("bounds content JSON"); + assert_eq!(content["has_more"], false); + assert_eq!(content["next_cursor"], serde_json::Value::Null); + + client.disconnect().await.expect("disconnect"); +} + /// Live badge counts: every thread mutation pushes a fresh relay-signed /// kind:39005 recount to channel subscribers — a reply counts up, deleting /// that reply counts back down — without any window refetch. diff --git a/docs/nips/NIP-CW.md b/docs/nips/NIP-CW.md index 9e6ea14e1b..d9648e20ea 100644 --- a/docs/nips/NIP-CW.md +++ b/docs/nips/NIP-CW.md @@ -6,7 +6,7 @@ Channel and Thread Windows `draft` `optional` `relay` -**Depends on**: NIP-01 (basic event format, filters), NIP-11 (relay information document), NIP-29 (relay-based groups), NIP-98 (HTTP auth) +**Depends on**: NIP-01 (basic event format, filters), NIP-11 (relay information document), NIP-29 (relay-based groups), NIP-42 (WebSocket auth), NIP-98 (HTTP auth) ## Abstract @@ -39,7 +39,7 @@ This NIP does not change ingest, storage, or fan-out. Rows returned in a window This NIP does not define around-target retrieval or cross-page snapshot isolation. It defines protocol compatibility and fallback rules, but the Buzz thread-mode implementation ships no client opt-in or fallback implementation. Thread mode starts at the newest reply and continues toward older replies. -This NIP does not require WebSocket REQ support. A relay MAY serve window filters only on an HTTP query surface and ignore the extension fields on REQ (see §Degradation). +A relay MAY serve window filters on HTTP, WebSocket REQ, or both. A client uses the bounds overlay—not transport selection—to determine whether the extension was honored (see §Degradation). ## Terminology @@ -218,8 +218,9 @@ legacy thread continuation fields and MUST NOT be reused as such. ## Thread Mode `thread_window: true` requests a newest-first page of replies to one root. It is -served through Buzz's NIP-98-authenticated `POST /query`; channel mode, -`kind:39006`, and the legacy oldest-first thread path are unchanged. The response +served through Buzz's NIP-98-authenticated `POST /query` and NIP-42-authenticated +WebSocket `REQ`; channel mode, `kind:39006`, and the legacy oldest-first thread +path are unchanged. The response is a flat array of reply events, optional auxiliary events, and exactly one relay-signed `kind:39007` bounds event. This specification adds no endpoint, subscription, or around-target query. @@ -342,7 +343,7 @@ client opt-in or fallback implementation. The channel-mode extension fields in this NIP are *additional* keys on a standard filter, and clients and relays that do not implement it need no changes: -- **Extension-unaware relay**: a tolerant filter parser (one that ignores unknown keys, as common NIP-01 implementations do) serves the filter as a plain `kinds` + `#h` query — a complete, correct, standard event stream. A strict parser may instead reject the filter outright. Both are safe: neither produces a wrong-but-plausible top-level timeline. A client MUST treat *either* signal — a response with no valid `kind:39006`, or an error/unsupported-filter response — as a downgrade, and fall back by reissuing a clean standard filter with all extension keys removed and assembling threads client-side. (Buzz's own WebSocket REQ path is such a tolerant parser: the filter deserializer drops the extension fields, so a window filter on REQ serves the standard query.) +- **Extension-unaware relay**: a tolerant filter parser (one that ignores unknown keys, as common NIP-01 implementations do) serves the filter as a plain `kinds` + `#h` query — a complete, correct, standard event stream. A strict parser may instead reject the filter outright. Both are safe: neither produces a wrong-but-plausible top-level timeline. A client MUST treat *either* signal — a response with no valid `kind:39006`, or an error/unsupported-filter response — as a downgrade, and fall back by reissuing a clean standard filter with all extension keys removed and assembling threads client-side. (A relay may implement the extension on one query transport but not another; clients detect support independently on each transport.) - **Extension-unaware client**: never sends `top_level`, never sees an overlay kind, and observes a completely standard relay. A relay implementing this NIP MAY advertise it in its NIP-11 relay information document; the discovery mechanism is out of scope for this NIP. A client needs no advertisement to probe safely: send one head window request and apply the downgrade rule above — the presence of a valid `kind:39006` is the capability signal. @@ -377,6 +378,7 @@ A channel-mode client with neither an authenticated transport nor a verifiable r - **NIP-01**: Supplies the filter grammar this NIP extends and the parameterized-replaceable semantics overlays lean on. (Degradation safety comes from this NIP's explicit downgrade-and-retry rule, not from assuming universal unknown-field tolerance.) - **NIP-29**: Supplies the channel model (`h` tags, group-scoped reads) windows are scoped by. - **NIP-50** and relay-side search: sibling precedent — a relay-computed view requested through extended filter fields, invisible to relays that do not implement it. +- **NIP-42**: Authenticates the WebSocket REQ surface Buzz serves thread windows on. - **NIP-98**: Authenticates the HTTP query surface Buzz serves windows on. - **NIP-11**: Names the relay identity that signs overlays and the natural place to advertise support.