feat: serve thread windows over websocket req

Co-authored-by: Kalvin Chau <kalvin@block.xyz>
Signed-off-by: am <6e30cd56c30e030cd31bb0939b94a7c257c9a09d5ba2d92cf2735da45629f248@buzz.block.builderlab.xyz>

Signed-off-by: Kalvin Chau <kalvin@block.xyz>

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(|_| {

crates/buzz-relay/src/api/bridge/thread_window.rs

@@ -87,17 +87,21 @@ fn append(events: &mut Vec<Value>, 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<Item = &'a Request>,
+    allowed_channels: Option<&[uuid::Uuid]>,
 ) -> Result<Vec<Value>, 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::<HashSet<_>>() != current.iter().collect::<HashSet<_>>() {

crates/buzz-relay/src/connection.rs

@@ -902,6 +902,7 @@ async fn handle_text_message(text: String, conn: Arc<ConnectionState>, 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<ConnectionState>, 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),

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<Filter>,
+    raw_filters: Vec<Value>,
     before_ids: Vec<Option<Vec<u8>>>,
     conn: Arc<ConnectionState>,
     state: Arc<AppState>,
@@ -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<Item = &'a buzz_core::thread_window::Request>,
+    allowed_channels: Option<&[uuid::Uuid]>,
+    reader: &PublicKey,
+    conn: &ConnectionState,
+    state: &AppState,
+) -> Result<bool, crate::api::bridge::thread_window::Error> {
+    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<Value>,
+) {
+    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

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<Filter>,
+        /// Raw filters retained because `nostr::Filter` drops extension fields.
+        raw_filters: Vec<Value>,
         /// Optional per-filter composite cursor tiebreaks from raw extension fields.
         before_ids: Vec<Option<Vec<u8>>>,
     },
@@ -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!([

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::<Vec<_>>(),
+        expected.iter().map(|event| event.id).collect::<Vec<_>>(),
+        "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.

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.