diff --git a/Cargo.lock b/Cargo.lock index 5e9c5aade3d..3e025aa46b1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -984,6 +984,7 @@ dependencies = [ "clap", "diffy", "dirs", + "futures-util", "hex", "infer", "nostr 0.44.7", @@ -996,6 +997,7 @@ dependencies = [ "tempfile", "thiserror 2.0.18", "tokio", + "tokio-tungstenite 0.29.0", "url", "uuid", ] diff --git a/crates/buzz-cli/Cargo.toml b/crates/buzz-cli/Cargo.toml index 59d1bb2cee6..5e300075d45 100644 --- a/crates/buzz-cli/Cargo.toml +++ b/crates/buzz-cli/Cargo.toml @@ -93,3 +93,6 @@ tempfile = "3" axum = { workspace = true } # `test-util` enables paused-time control for deterministic timeout tests tokio = { workspace = true, features = ["macros", "rt-multi-thread", "test-util"] } +# Local authenticated WebSocket relay fixtures for subscription tests +tokio-tungstenite = { workspace = true } +futures-util = { workspace = true } diff --git a/crates/buzz-cli/TESTING.md b/crates/buzz-cli/TESTING.md index b7fa06d2031..b0d8d838159 100644 --- a/crates/buzz-cli/TESTING.md +++ b/crates/buzz-cli/TESTING.md @@ -222,6 +222,11 @@ buzz messages thread --channel "$CHANNEL_ID" --event "$REPLY_ID" | jq . buzz messages thread \ --link "buzz://message?channel=$CHANNEL_ID&id=$REPLY_ID&thread=$EVENT_ID" | jq . +# messages wait — returns the next matching reply immediately, or null on timeout +SINCE=$(date +%s) +buzz messages wait --channel "$CHANNEL_ID" --event "$EVENT_ID" \ + --author "" --since "$SINCE" --timeout 5 | jq . + # messages search buzz messages search --query "Hello" | jq . buzz messages search --query "CLI test" --limit 5 | jq . @@ -571,6 +576,7 @@ buzz channels delete --channel "$FORUM_ID" | jq . | 6 | `messages thread` | ☐ | | | 7 | `messages search` | ☐ | With limit | | 8 | `messages vote` | ☐ | Up and down | +| 8a | `messages wait` | ☐ | WebSocket reply and null timeout | | 9 | `channels list` | ☐ | With visibility, member | | 10 | `channels get` | ☐ | | | 11 | `channels create` | ☐ | Stream and forum | diff --git a/crates/buzz-cli/src/client.rs b/crates/buzz-cli/src/client.rs index 76d0e6fb959..a3608ead15c 100644 --- a/crates/buzz-cli/src/client.rs +++ b/crates/buzz-cli/src/client.rs @@ -569,6 +569,67 @@ impl BuzzClient { &self.relay_url } + /// Wait for the first event matching a Nostr subscription filter. + /// + /// The full connect, NIP-42 authentication, subscription, and receive + /// sequence is bounded by `timeout_duration`. A normal timeout returns + /// `Ok(None)`; transport and relay subscription failures remain errors so + /// callers can reconcile through the HTTP query path. + pub async fn wait_for_event( + &self, + filter: &serde_json::Value, + timeout_duration: Duration, + ) -> Result, CliError> { + if timeout_duration.is_zero() { + return Ok(None); + } + + let ws_url = to_ws_url(&self.relay_url); + let subscription_id = format!("buzz-cli-wait-{}", uuid::Uuid::new_v4()); + let wait = async { + let mut connection = buzz_ws_client::NostrWsConnection::connect_authenticated( + &ws_url, + &self.keys, + self.auth_tag.as_ref(), + ) + .await + .map_err(|error| CliError::Other(error.to_string()))?; + + connection + .send_raw(&serde_json::json!(["REQ", subscription_id, filter])) + .await + .map_err(|error| CliError::Other(error.to_string()))?; + + loop { + match connection.next_event(timeout_duration).await { + Ok(buzz_ws_client::RelayMessage::Event { + subscription_id: received_subscription_id, + event, + }) if received_subscription_id == subscription_id => { + let _ = connection.disconnect().await; + return Ok(Some(*event)); + } + Ok(buzz_ws_client::RelayMessage::Closed { + subscription_id: closed_subscription_id, + message, + }) if closed_subscription_id == subscription_id => { + return Err(CliError::Other(format!( + "relay closed subscription {closed_subscription_id}: {message}" + ))); + } + Ok(_) => {} + Err(buzz_ws_client::WsClientError::Timeout) => return Ok(None), + Err(error) => return Err(CliError::Other(error.to_string())), + } + } + }; + + match tokio::time::timeout(timeout_duration, wait).await { + Ok(result) => result, + Err(_) => Ok(None), + } + } + /// Return the owner pubkey carried by the NIP-OA auth tag, if any. /// /// The auth tag is `["auth", owner_pubkey, conditions, sig]`; the @@ -2332,6 +2393,103 @@ mod tests { normalize_events, BuzzClient, }; use nostr::{EventBuilder, Keys, Kind, Tag}; + use std::time::Duration; + + #[tokio::test] + async fn wait_for_event_authenticates_subscribes_and_returns_live_match() { + use futures_util::{SinkExt, StreamExt}; + use tokio_tungstenite::{accept_async, tungstenite::Message}; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let channel_id = "123e4567-e89b-12d3-a456-426614174000"; + let root_event_id = "a".repeat(64); + let event = EventBuilder::new(Kind::Custom(9), "live reply") + .tags([ + Tag::parse(["h", channel_id]).unwrap(), + Tag::parse(["e", root_event_id.as_str(), "", "root"]).unwrap(), + ]) + .sign_with_keys(&Keys::generate()) + .unwrap(); + let expected_event_id = event.id; + let expected_filter = serde_json::json!({ + "kinds": [9], + "#h": [channel_id], + "#e": [root_event_id], + }); + let server_filter = expected_filter.clone(); + + let server = tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + let mut websocket = accept_async(stream).await.unwrap(); + websocket + .send(Message::Text( + serde_json::json!(["AUTH", "test-challenge"]) + .to_string() + .into(), + )) + .await + .unwrap(); + + let auth_text = websocket + .next() + .await + .unwrap() + .unwrap() + .into_text() + .unwrap(); + let auth: serde_json::Value = serde_json::from_str(&auth_text).unwrap(); + let auth_event_id = auth[1]["id"].as_str().unwrap(); + websocket + .send(Message::Text( + serde_json::json!(["OK", auth_event_id, true, ""]) + .to_string() + .into(), + )) + .await + .unwrap(); + + let request_text = websocket + .next() + .await + .unwrap() + .unwrap() + .into_text() + .unwrap(); + let request: serde_json::Value = serde_json::from_str(&request_text).unwrap(); + assert_eq!(request[0], "REQ"); + assert_eq!(request[2], server_filter); + let subscription_id = request[1].as_str().unwrap(); + websocket + .send(Message::Text( + serde_json::json!(["EOSE", subscription_id]) + .to_string() + .into(), + )) + .await + .unwrap(); + websocket + .send(Message::Text( + serde_json::json!(["EVENT", subscription_id, event]) + .to_string() + .into(), + )) + .await + .unwrap(); + }); + + let client = + BuzzClient::new(format!("http://{address}"), Keys::generate(), None, None).unwrap(); + let received = client + .wait_for_event(&expected_filter, Duration::from_secs(2)) + .await + .unwrap() + .unwrap(); + + assert_eq!(received.id, expected_event_id); + assert_eq!(received.content, "live reply"); + server.await.unwrap(); + } #[test] fn normalize_events_preserves_the_complete_signed_event_shape() { diff --git a/crates/buzz-cli/src/commands/messages.rs b/crates/buzz-cli/src/commands/messages.rs index 9f41fbf751c..a155cfbd3d0 100644 --- a/crates/buzz-cli/src/commands/messages.rs +++ b/crates/buzz-cli/src/commands/messages.rs @@ -1,5 +1,6 @@ use buzz_sdk::{DeleteMessageOptions, DiffMeta, ThreadRef, VoteDirection}; use nostr::PublicKey; +use std::time::Duration; use uuid::Uuid; use crate::client::{normalize_events, normalize_write_response, BuzzClient}; @@ -464,6 +465,68 @@ pub async fn cmd_get_thread( Ok(()) } +fn build_wait_reply_filter( + channel_id: &str, + root_event_id: &str, + author_hex: &str, + since: i64, +) -> serde_json::Value { + serde_json::json!({ + "kinds": [9, 40002], + "#h": [channel_id], + "#e": [root_event_id], + "authors": [author_hex], + "since": since, + }) +} + +fn format_wait_event( + event: nostr::Event, + format: &crate::OutputFormat, +) -> Result { + let raw = serde_json::to_value(event) + .map_err(|error| CliError::Other(format!("failed to serialize event: {error}")))?; + let normalized = normalize_events(&[raw]); + let formatted = format_events(&normalized, format); + let mut events: Vec = serde_json::from_str(&formatted) + .map_err(|error| CliError::Other(format!("failed to format event: {error}")))?; + let event = events + .pop() + .ok_or_else(|| CliError::Other("relay returned an empty event".into()))?; + + serde_json::to_string(&event) + .map_err(|error| CliError::Other(format!("failed to encode event output: {error}"))) +} + +pub async fn cmd_wait_for_reply( + client: &BuzzClient, + channel_id: &str, + root_event_id: &str, + author: &str, + since: i64, + timeout_seconds: u64, + format: &crate::OutputFormat, +) -> Result<(), CliError> { + validate_uuid(channel_id)?; + validate_hex64(root_event_id)?; + if since < 0 { + return Err(CliError::Usage("--since must be zero or greater".into())); + } + let author_hex = PublicKey::parse(author) + .map_err(|_| CliError::Usage(format!("invalid --author pubkey: {author}")))? + .to_hex(); + let filter = build_wait_reply_filter(channel_id, root_event_id, &author_hex, since); + + match client + .wait_for_event(&filter, Duration::from_secs(timeout_seconds)) + .await? + { + Some(event) => println!("{}", format_wait_event(event, format)?), + None => println!("null"), + } + Ok(()) +} + pub async fn cmd_search( client: &BuzzClient, query: Option<&str>, @@ -1031,6 +1094,13 @@ pub async fn dispatch( ) .await } + MessagesCmd::Wait { + channel, + event, + author, + since, + timeout, + } => cmd_wait_for_reply(client, &channel, &event, &author, since, timeout, format).await, MessagesCmd::Search { query, author, @@ -1056,16 +1126,16 @@ pub async fn dispatch( #[cfg(test)] mod tests { use super::{ - channel_id_from_event, cmd_get_thread, event_mention_pubkeys, find_root_from_tags, - format_events, match_profiles_by_name, merge_message_mentions, missing_members, - normalize_explicit_mentions, parse_member_pubkeys, resolve_names_to_pubkeys, - resolve_thread_target, thread_ref_from_event, thread_ref_from_parent_tags, BuzzClient, - CliError, Uuid, + build_wait_reply_filter, channel_id_from_event, cmd_get_thread, event_mention_pubkeys, + find_root_from_tags, format_events, format_wait_event, match_profiles_by_name, + merge_message_mentions, missing_members, normalize_explicit_mentions, parse_member_pubkeys, + resolve_names_to_pubkeys, resolve_thread_target, thread_ref_from_event, + thread_ref_from_parent_tags, BuzzClient, CliError, Uuid, }; use buzz_sdk::mentions::{ extract_at_mentions_with_known, extract_at_names, match_names_to_profiles, MentionProfile, }; - use nostr::Keys; + use nostr::{EventBuilder, Keys, Kind}; use serde_json::json; const ID_A: &str = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; @@ -1105,6 +1175,35 @@ mod tests { ); } + #[test] + fn wait_reply_filter_is_scoped_to_thread_author_and_time() { + let channel = "123e4567-e89b-12d3-a456-426614174000"; + let filter = build_wait_reply_filter(channel, ID_A, PK_VALID_A, 1_777_777_777); + + assert_eq!(filter["kinds"], json!([9, 40002])); + assert_eq!(filter["#h"], json!([channel])); + assert_eq!(filter["#e"], json!([ID_A])); + assert_eq!(filter["authors"], json!([PK_VALID_A])); + assert_eq!(filter["since"], json!(1_777_777_777)); + } + + #[test] + fn wait_reply_uses_the_message_compact_contract() { + let event = EventBuilder::new(Kind::Custom(9), "live reply") + .sign_with_keys(&Keys::generate()) + .unwrap(); + + let output: serde_json::Value = + serde_json::from_str(&format_wait_event(event, &crate::OutputFormat::Compact).unwrap()) + .unwrap(); + + assert_eq!(output["content"], "live reply"); + assert!(output.get("id").is_some()); + assert!(output.get("created_at").is_some()); + assert!(output.get("pubkey").is_none()); + assert!(output.get("kind").is_none()); + } + #[tokio::test] async fn malformed_channel_is_rejected_before_thread_fetch() { let client = diff --git a/crates/buzz-cli/src/lib.rs b/crates/buzz-cli/src/lib.rs index d0155970fa2..d8482384f98 100644 --- a/crates/buzz-cli/src/lib.rs +++ b/crates/buzz-cli/src/lib.rs @@ -501,6 +501,27 @@ pub enum MessagesCmd { #[arg(long)] depth_limit: Option, }, + /// Wait for the next reply from one author in a message thread + #[command( + after_help = "Example:\n buzz messages wait --channel --event --author --since --timeout 300" + )] + Wait { + /// Channel UUID containing the thread + #[arg(long)] + channel: String, + /// Thread root event ID (64-char hex) + #[arg(long)] + event: String, + /// Reply author pubkey (hex or npub) + #[arg(long)] + author: String, + /// Unix timestamp — ignore replies created before this time + #[arg(long)] + since: i64, + /// Maximum seconds to wait before returning null + #[arg(long, value_parser = clap::value_parser!(u64).range(1..=86400))] + timeout: u64, + }, /// Full-text search across messages #[command( after_help = "Examples:\n buzz messages search --query checkout\n buzz messages search --author npub1... --since 1783497600\n buzz messages search --author Aaron --query checkout --limit 20" @@ -2196,6 +2217,60 @@ mod tests { .is_err()); } + #[test] + fn messages_wait_requires_a_bounded_complete_target() { + let channel = "123e4567-e89b-12d3-a456-426614174000"; + let event = "a".repeat(64); + let author = "b".repeat(64); + let valid = [ + "buzz", + "messages", + "wait", + "--channel", + channel, + "--event", + event.as_str(), + "--author", + author.as_str(), + "--since", + "1777777777", + "--timeout", + "300", + ]; + + assert!(Cli::try_parse_from(valid).is_ok()); + assert!(Cli::try_parse_from([ + "buzz", + "messages", + "wait", + "--channel", + channel, + "--event", + event.as_str(), + "--author", + author.as_str(), + "--since", + "1777777777", + ]) + .is_err()); + assert!(Cli::try_parse_from([ + "buzz", + "messages", + "wait", + "--channel", + channel, + "--event", + event.as_str(), + "--author", + author.as_str(), + "--since", + "1777777777", + "--timeout", + "0", + ]) + .is_err()); + } + #[test] fn set_status_clear_rejects_text_and_emoji() { for extra in [["--text", "busy"], ["--emoji", "🎶"]] { @@ -2306,7 +2381,8 @@ mod tests { "send", "send-diff", "thread", - "vote" + "vote", + "wait" ] ); assert_eq!( @@ -2440,7 +2516,7 @@ mod tests { ("feed", 1), ("issues", 6), ("media", 1), - ("messages", 8), + ("messages", 9), ("pack", 2), ("patches", 4), ("pr", 5),