diff --git a/crates/buzz-cli/README.md b/crates/buzz-cli/README.md index a2dcdce6d2..a28f0ccbd9 100644 --- a/crates/buzz-cli/README.md +++ b/crates/buzz-cli/README.md @@ -107,6 +107,7 @@ stored rules in `validation_error` so an owner can remove and repair them. | | `edit` | Edit a message you sent | | | `delete` | Delete a message | | | `get` | List messages in a channel | +| | `subscribe` | Stream new messages live over WebSocket, one JSON object per line | | | `thread` | Get a message thread | | | `search` | Full-text search, filterable by author | | | `vote` | Vote on a forum post | diff --git a/crates/buzz-cli/src/client.rs b/crates/buzz-cli/src/client.rs index d0dd2677a9..7c86149d46 100644 --- a/crates/buzz-cli/src/client.rs +++ b/crates/buzz-cli/src/client.rs @@ -1095,6 +1095,87 @@ impl BuzzClient { .to_string()) } + /// Hold an authenticated WebSocket subscription open and hand every matching + /// event to `on_event` as the relay pushes it. + /// + /// This never returns `Ok`: a subscription that is working is a subscription + /// that has not finished. Every return is a reason the stream stopped, which + /// is the only thing a caller can act on — a reader that cannot tell "quiet + /// channel" from "dead socket" is worse than a poller, because it believes + /// it is listening. + /// + /// `idle_timeout_secs` is what makes that distinction possible. The relay + /// heartbeats every 30 s (`buzz_relay::connection::heartbeat_loop`) and + /// `NostrWsConnection` answers each Ping without surfacing it, so the read + /// deadline is only reached when nothing at all arrived — a half-dead socket, + /// not a quiet room. Keep the value a comfortable multiple of 30. + pub async fn subscribe_events( + &self, + filter: serde_json::Value, + idle_timeout_secs: u64, + mut on_event: F, + ) -> Result + where + F: FnMut(&nostr::Event) -> Result<(), CliError>, + { + use buzz_ws_client::{NostrWsConnection, RelayMessage, WsClientError}; + + let ws_url = to_ws_url(&self.relay_url); + let mut conn = + NostrWsConnection::connect_authenticated(&ws_url, &self.keys, self.auth_tag.as_ref()) + .await + .map_err(|e| CliError::Other(format!("{ws_url}: {e}")))?; + + let sub_id = format!("buzz-cli-{}", uuid::Uuid::new_v4()); + conn.send_raw(&serde_json::json!(["REQ", sub_id, filter])) + .await + .map_err(|e| CliError::Other(format!("REQ failed: {e}")))?; + + let idle = std::time::Duration::from_secs(idle_timeout_secs); + loop { + match conn.next_event(idle).await { + Ok(RelayMessage::Event { + subscription_id, + event, + }) => { + if subscription_id == sub_id { + on_event(&event)?; + } + } + // EOSE means stored events are done and live delivery starts; + // COUNT and OK cannot arrive on this connection but are not worth + // tearing a working subscription down for. + Ok(RelayMessage::Eose { .. }) + | Ok(RelayMessage::Count { .. }) + | Ok(RelayMessage::Ok(_)) => {} + Ok(RelayMessage::Notice { message }) => { + eprintln!("relay notice: {message}"); + } + // A re-issued AUTH challenge means the relay stopped treating this + // connection as authenticated; the subscription is no longer + // delivering anything, so reconnecting is the only repair. + Ok(RelayMessage::Auth { .. }) => { + return Err(CliError::Other( + "relay re-issued an AUTH challenge — session no longer authenticated" + .into(), + )); + } + Ok(RelayMessage::Closed { message, .. }) => { + return Err(CliError::Other(format!( + "relay closed the subscription: {message}" + ))); + } + Err(WsClientError::Timeout) => { + return Err(CliError::Other(format!( + "no relay traffic for {idle_timeout_secs}s, not even a heartbeat — \ + treating the socket as dead" + ))); + } + Err(e) => return Err(CliError::Other(e.to_string())), + } + } + } + /// Upload a file to the relay's Blossom endpoint. /// Returns a BlobDescriptor on success. pub async fn upload_file(&self, file_path: &str) -> Result { @@ -1302,22 +1383,27 @@ fn to_ws_url(http_url: &str) -> String { .replace("http://", "ws://") } +/// Normalize one raw event JSON object into the shape every read path emits: +/// `{id, pubkey, kind, content, created_at, tags}`. +/// +/// Shared by the HTTP read path (`normalize_events`) and the WebSocket stream +/// (`messages subscribe`) so a consumer can dedupe across both without knowing +/// which one delivered a given event. +pub fn normalize_event(e: &serde_json::Value) -> serde_json::Value { + serde_json::json!({ + "id": e.get("id").and_then(|v| v.as_str()).unwrap_or(""), + "pubkey": e.get("pubkey").and_then(|v| v.as_str()).unwrap_or(""), + "kind": e.get("kind").and_then(|v| v.as_u64()).unwrap_or(0), + "content": e.get("content").and_then(|v| v.as_str()).unwrap_or(""), + "created_at": e.get("created_at").and_then(|v| v.as_u64()).unwrap_or(0), + "tags": e.get("tags").cloned().unwrap_or(serde_json::json!([])), + }) +} + /// Normalize raw event JSON array into consistent shape. /// Each event becomes: {id, pubkey, kind, content, created_at, tags} pub fn normalize_events(events: &[serde_json::Value]) -> String { - let normalized: Vec = events - .iter() - .map(|e| { - serde_json::json!({ - "id": e.get("id").and_then(|v| v.as_str()).unwrap_or(""), - "pubkey": e.get("pubkey").and_then(|v| v.as_str()).unwrap_or(""), - "kind": e.get("kind").and_then(|v| v.as_u64()).unwrap_or(0), - "content": e.get("content").and_then(|v| v.as_str()).unwrap_or(""), - "created_at": e.get("created_at").and_then(|v| v.as_u64()).unwrap_or(0), - "tags": e.get("tags").cloned().unwrap_or(serde_json::json!([])), - }) - }) - .collect(); + let normalized: Vec = events.iter().map(normalize_event).collect(); serde_json::to_string(&normalized).unwrap_or_default() } diff --git a/crates/buzz-cli/src/commands/messages.rs b/crates/buzz-cli/src/commands/messages.rs index 40a9ae80b5..00059bbe8b 100644 --- a/crates/buzz-cli/src/commands/messages.rs +++ b/crates/buzz-cli/src/commands/messages.rs @@ -1,8 +1,10 @@ +use std::io::Write; + use buzz_sdk::{DeleteMessageOptions, DiffMeta, ThreadRef, VoteDirection}; use nostr::PublicKey; use uuid::Uuid; -use crate::client::{normalize_events, normalize_write_response, BuzzClient}; +use crate::client::{normalize_event, normalize_events, normalize_write_response, BuzzClient}; use crate::error::CliError; use crate::validate::{ infer_language, parse_event_id, parse_uuid, read_or_stdin, truncate_diff, @@ -350,6 +352,24 @@ fn format_events(normalized: &str, format: &crate::OutputFormat) -> String { } } +/// The event kinds a channel read returns when the caller does not name any: +/// chat plus the channel-scoped system events a reader expects to see. +const CHANNEL_READ_KINDS: [u64; 5] = [9, 40002, 40008, 45001, 45003]; + +/// Parse `--kinds 9,1984` into a filter list, falling back to the default read +/// set. Shared by the HTTP read and the WebSocket stream so `messages subscribe` +/// delivers exactly what `messages get` would have returned. +fn channel_read_kinds(kinds: Option<&str>) -> Vec { + let parsed: Vec = kinds + .map(|k| k.split(',').filter_map(|s| s.trim().parse().ok()).collect()) + .unwrap_or_default(); + if parsed.is_empty() { + CHANNEL_READ_KINDS.to_vec() + } else { + parsed + } +} + pub async fn cmd_get_messages( client: &BuzzClient, channel_id: &str, @@ -363,19 +383,11 @@ pub async fn cmd_get_messages( let limit = limit.unwrap_or(50).min(200); let mut filter = serde_json::json!({ - "kinds": [9, 40002, 40008, 45001, 45003], + "kinds": channel_read_kinds(kinds), "#h": [channel_id], "limit": limit }); - // If specific kinds requested, override - if let Some(k) = kinds { - let kind_list: Vec = k.split(',').filter_map(|s| s.trim().parse().ok()).collect(); - if !kind_list.is_empty() { - filter["kinds"] = serde_json::json!(kind_list); - } - } - if let Some(b) = before { filter["until"] = serde_json::json!(b); } @@ -391,6 +403,91 @@ pub async fn cmd_get_messages( Ok(()) } +/// Relay heartbeat period (`buzz_relay::connection::heartbeat_loop`). The idle +/// deadline has to clear several of these or a healthy quiet channel looks dead. +const RELAY_HEARTBEAT_SECS: u64 = 30; + +/// Stream a channel over a held-open authenticated WebSocket, one normalized +/// event per line on stdout, until the connection stops delivering. +/// +/// Deliberately not `--format`-aware. The output is a transport for a program +/// that reads one line at a time and acts on it, not a rendering for a person, +/// and a pretty table cannot be consumed a line at a time. +pub async fn cmd_subscribe_messages( + client: &BuzzClient, + channel_id: &str, + kinds: Option<&str>, + since: Option, + idle_timeout: u64, + reconnect_after: u64, +) -> Result<(), CliError> { + validate_uuid(channel_id)?; + if idle_timeout <= RELAY_HEARTBEAT_SECS { + return Err(CliError::Usage(format!( + "--idle-timeout {idle_timeout} is not above the relay's {RELAY_HEARTBEAT_SECS}s \ + heartbeat, so a healthy connection would be torn down as dead; use 90 or more" + ))); + } + + // Default to "from now". History is what `messages get` is for, and a + // subscriber that replays the backlog on every reconnect turns a flapping + // network into a flood. + let since = since.unwrap_or_else(|| { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs() as i64) + .unwrap_or(0) + }); + + let filter = serde_json::json!({ + "kinds": channel_read_kinds(kinds), + "#h": [channel_id], + "since": since, + }); + + let mut out = std::io::stdout().lock(); + let mut reader_gone = false; + + let stream = client.subscribe_events(filter, idle_timeout, |event| { + let raw = serde_json::to_value(event) + .map_err(|e| CliError::Other(format!("cannot serialize event: {e}")))?; + let line = normalize_event(&raw).to_string(); + // Flush every line. The consumer is a reader blocked on this pipe, so + // a line sitting in a buffer is an undelivered message — the exact + // failure this command exists to remove. + if let Err(e) = writeln!(out, "{line}").and_then(|()| out.flush()) { + if e.kind() == std::io::ErrorKind::BrokenPipe { + reader_gone = true; + } + return Err(CliError::Other(format!("stdout: {e}"))); + } + Ok(()) + }); + + // A healthy subscription is indistinguishable from a subscription the relay + // has quietly stopped matching against: both are silent, and both heartbeat. + // Ending a good connection on a schedule is what gives the supervisor its + // chance to re-read over HTTP and find out which one this was. + let stopped = if reconnect_after == 0 { + stream.await + } else { + match tokio::time::timeout(std::time::Duration::from_secs(reconnect_after), stream).await { + Ok(reason) => reason, + Err(_) => Err(CliError::Other(format!( + "scheduled re-subscribe after {reconnect_after}s" + ))), + } + }; + + match stopped { + Ok(never) => match never {}, + // The reader closed the pipe. That is the reader's decision, not a + // failure of the stream, and must not be reported as one. + Err(_) if reader_gone => Ok(()), + Err(reason) => Err(reason), + } +} + pub async fn cmd_get_thread( client: &BuzzClient, channel_id: &str, @@ -962,6 +1059,23 @@ pub async fn dispatch( ) .await } + MessagesCmd::Subscribe { + channel, + kinds, + since, + idle_timeout, + reconnect_after, + } => { + cmd_subscribe_messages( + client, + &channel, + kinds.as_deref(), + since, + idle_timeout, + reconnect_after, + ) + .await + } MessagesCmd::Thread { channel, event, diff --git a/crates/buzz-cli/src/lib.rs b/crates/buzz-cli/src/lib.rs index f745e7b280..c56fe5b015 100644 --- a/crates/buzz-cli/src/lib.rs +++ b/crates/buzz-cli/src/lib.rs @@ -460,6 +460,28 @@ pub enum MessagesCmd { #[arg(long)] kinds: Option, }, + /// Stream new messages from a channel over a live WebSocket, one JSON object + /// per line, until the connection drops + #[command( + after_help = "Holds an authenticated NIP-42 connection open and prints each matching\nevent as it arrives — no polling, no interval.\n\nOutput is newline-delimited JSON on stdout, one event per line, in the same\nshape 'messages get' returns, flushed immediately so a reader blocked on a\nline wakes the moment the relay pushes one. Diagnostics go to stderr.\n\nIt never exits 0. Every exit is a reason the stream stopped, so a supervisor\ncan reconnect and backfill the gap with 'messages get --since' instead of\nsitting on a dead socket believing the channel is quiet.\n\nExamples:\n buzz messages subscribe --channel \n buzz messages subscribe --channel --kinds 9 --since 1783497600" + )] + Subscribe { + /// Channel UUID + #[arg(long)] + channel: String, + /// Comma-separated event kinds to stream [default: the same set 'messages get' reads] + #[arg(long)] + kinds: Option, + /// Unix timestamp — only stream events at or after this time [default: now] + #[arg(long)] + since: Option, + /// Give up when the relay sends nothing at all, heartbeats included, for this many seconds + #[arg(long, default_value_t = 90)] + idle_timeout: u64, + /// Stop after this many seconds even while healthy, so a supervisor can backfill over HTTP and re-subscribe [0 disables] + #[arg(long, default_value_t = 300)] + reconnect_after: u64, + }, /// Get a message thread (replies to a root message) Thread { /// Channel UUID @@ -2161,6 +2183,7 @@ mod tests { "search", "send", "send-diff", + "subscribe", "thread", "vote" ] @@ -2295,7 +2318,7 @@ mod tests { ("feed", 1), ("issues", 4), ("media", 1), - ("messages", 8), + ("messages", 9), ("pack", 2), ("patches", 4), ("pr", 5),