feat(cli): add buzz messages subscribe for streaming channel events

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 |

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<F>(
+        &self,
+        filter: serde_json::Value,
+        idle_timeout_secs: u64,
+        mut on_event: F,
+    ) -> Result<std::convert::Infallible, CliError>
+    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<BlobDescriptor, CliError> {
@@ -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<serde_json::Value> = 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<serde_json::Value> = events.iter().map(normalize_event).collect();
     serde_json::to_string(&normalized).unwrap_or_default()
 }
 

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<u64> {
+    let parsed: Vec<u64> = 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<u64> = 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<i64>,
+    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,

crates/buzz-cli/src/lib.rs

@@ -460,6 +460,28 @@ pub enum MessagesCmd {
         #[arg(long)]
         kinds: Option<String>,
     },
+    /// 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 <UUID>\n  buzz messages subscribe --channel <UUID> --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<String>,
+        /// Unix timestamp — only stream events at or after this time [default: now]
+        #[arg(long)]
+        since: Option<i64>,
+        /// 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),