fix(cli): publish ephemeral events over WebSocket via sprout-ws-client (#876)

AGENTS.md

@@ -61,6 +61,7 @@ crates/
   sprout-cli          # Agent-first CLI
   sprout-sdk          # Typed Nostr event builders
   sprout-admin        # Operator CLI for relay administration
+  sprout-ws-client    # Shared NIP-42 WebSocket client (connect, auth, publish)
   sprout-test-client  # Integration test client and E2E test suite
   sprig               # All-in-one harness bundling ACP, agent, and dev MCP
 

Cargo.lock

@@ -7558,6 +7558,7 @@ dependencies = [
  "sprout-core",
  "sprout-persona",
  "sprout-sdk",
+ "sprout-ws-client",
  "thiserror 2.0.18",
  "tokio",
  "url",
@@ -7845,6 +7846,7 @@ dependencies = [
  "serde_json",
  "sha2 0.11.0",
  "sprout-core",
+ "sprout-ws-client",
  "thiserror 2.0.18",
  "tokio",
  "tokio-tungstenite 0.29.0",
@@ -7876,6 +7878,20 @@ dependencies = [
  "uuid",
 ]
 
+[[package]]
+name = "sprout-ws-client"
+version = "0.1.0"
+dependencies = [
+ "futures-util",
+ "nostr",
+ "serde_json",
+ "thiserror 2.0.18",
+ "tokio",
+ "tokio-tungstenite 0.29.0",
+ "tracing",
+ "url",
+]
+
 [[package]]
 name = "sqlx"
 version = "0.9.0"

Cargo.toml

@@ -12,6 +12,7 @@ members = [
     "crates/sprig",
     "crates/sprout-proxy",
     "crates/sprout-test-client",
+    "crates/sprout-ws-client",
     "crates/sprout-admin",
     "crates/sprout-workflow",
     "crates/sprout-media",
@@ -116,6 +117,7 @@ sprout-proxy = { path = "crates/sprout-proxy" }
 sprout-workflow = { path = "crates/sprout-workflow" }
 sprout-media = { path = "crates/sprout-media" }
 sprout-sdk = { path = "crates/sprout-sdk" }
+sprout-ws-client = { path = "crates/sprout-ws-client" }
 
 # CI profile — release-grade codegen for the relay so e2e tests hit a
 # realistic binary, not an unoptimised debug build.  Inherits `release`

crates/sprout-cli/Cargo.toml

@@ -62,3 +62,6 @@ url = { workspace = true }
 
 # Persona pack parsing, validation, and resolution
 sprout-persona = { path = "../sprout-persona" }
+
+# WebSocket client — ephemeral event publish (kind:20001 is WS-only on the relay)
+sprout-ws-client = { path = "../sprout-ws-client" }

crates/sprout-cli/src/client.rs

@@ -283,6 +283,36 @@ impl SproutClient {
         self.handle_response(resp).await
     }
 
+    // -----------------------------------------------------------------------
+    // WebSocket publish (ephemeral events)
+    // -----------------------------------------------------------------------
+
+    /// Publish an ephemeral event via WebSocket with NIP-42 authentication.
+    ///
+    /// The relay rejects ephemeral kinds (20000–29999) over HTTP. Delegates to
+    /// `sprout_ws_client::publish_event` which handles connect, NIP-42 auth,
+    /// EVENT send, OK wait, and graceful close.
+    pub async fn publish_ephemeral_event(&self, event: nostr::Event) -> Result<String, CliError> {
+        let ws_url = to_ws_url(&self.relay_url);
+        let ok =
+            sprout_ws_client::publish_event(&ws_url, event, &self.keys, self.auth_tag.as_ref(), 10)
+                .await
+                .map_err(|e| CliError::Other(e.to_string()))?;
+
+        if !ok.accepted {
+            return Err(CliError::Relay {
+                status: 400,
+                body: ok.message,
+            });
+        }
+        Ok(serde_json::json!({
+            "event_id": ok.event_id,
+            "accepted": true,
+            "message": ok.message,
+        })
+        .to_string())
+    }
+
     // -----------------------------------------------------------------------
     // File upload (Blossom protocol)
     // -----------------------------------------------------------------------
@@ -441,6 +471,13 @@ pub fn normalize_relay_url(url: &str) -> String {
         .to_string()
 }
 
+/// Convert an HTTP(S) relay base URL back to a WebSocket URL for NIP-01 connections.
+fn to_ws_url(http_url: &str) -> String {
+    http_url
+        .replace("https://", "wss://")
+        .replace("http://", "ws://")
+}
+
 // ---------------------------------------------------------------------------
 // Output normalization helpers
 // ---------------------------------------------------------------------------

crates/sprout-cli/src/commands/users.rs

@@ -276,17 +276,16 @@ pub async fn cmd_get_presence(client: &SproutClient, pubkeys_csv: &str) -> Resul
     Ok(())
 }
 
-/// Set presence status — sign and submit a kind:20001 presence update event.
+/// Set presence status — sign and submit a kind:20001 presence update event via WebSocket.
 ///
-/// NOTE: Kind 20001 is ephemeral and only accepted via WebSocket connections.
+/// Kind 20001 is ephemeral and only accepted via WebSocket connections. This
-/// The CLI uses the HTTP bridge (POST /events) which rejects ephemeral kinds.
+/// method connects to the relay over WS, performs NIP-42 authentication, and
-/// This will fail until the CLI gains a WS publish path. The kind is correct
+/// publishes the event directly — bypassing the HTTP bridge.
-/// per the protocol spec (KIND_PRESENCE_UPDATE = 20001).
 pub async fn cmd_set_presence(client: &SproutClient, status: &str) -> Result<(), CliError> {
     let builder = sprout_sdk::build_presence_update(status).map_err(crate::validate::sdk_err)?;
     let event = client.sign_event(builder)?;
 
-    let resp = client.submit_event(event).await?;
+    let resp = client.publish_ephemeral_event(event).await?;
     println!("{}", normalize_write_response(&resp));
     Ok(())
 }

crates/sprout-test-client/Cargo.toml

@@ -10,6 +10,7 @@ description = "Integration test client and E2E test suite for Sprout"
 [dependencies]
 anyhow = { workspace = true }
 sprout-core = { workspace = true }
+sprout-ws-client = { workspace = true }
 nostr = { workspace = true }
 tokio = { workspace = true }
 tokio-tungstenite = { workspace = true }

crates/sprout-test-client/src/lib.rs

@@ -3,167 +3,26 @@
 
 //! Minimal NIP-01 WebSocket test client for the Sprout relay.
 
-use std::collections::VecDeque;
 use std::time::Duration;
 
-use futures_util::{SinkExt, StreamExt};
+use nostr::{Event, EventBuilder, Filter, Keys, Kind, Tag};
-use nostr::{Event, EventBuilder, Filter, Keys, Kind, RelayUrl, Tag};
 use serde_json::{json, Value};
 use thiserror::Error;
-use tokio::time::timeout;
-use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream};
 use tracing::debug;
 
-/// A message received from a Nostr relay.
+use sprout_ws_client::NostrWsConnection;
-#[derive(Debug, Clone)]
+pub use sprout_ws_client::{parse_relay_message, OkResponse, RelayMessage, WsClientError};
-pub enum RelayMessage {
-    /// An event matching an active subscription.
-    Event {
-        /// The subscription ID this event belongs to.
-        subscription_id: String,
-        /// The Nostr event payload.
-        event: Box<Event>,
-    },
-    /// Acknowledgement of a published event.
-    Ok(OkResponse),
-    /// End-of-stored-events marker for a subscription.
-    Eose {
-        /// The subscription ID that has reached end-of-stored-events.
-        subscription_id: String,
-    },
-    /// The relay closed a subscription, usually with an error.
-    Closed {
-        /// The subscription ID that was closed.
-        subscription_id: String,
-        /// Human-readable reason for the closure.
-        message: String,
-    },
-    /// A human-readable notice from the relay.
-    Notice {
-        /// The notice text.
-        message: String,
-    },
-    /// A NIP-42 authentication challenge from the relay.
-    Auth {
-        /// The challenge string to sign.
-        challenge: String,
-    },
-}
-
-/// The relay's response to a published event (NIP-01 `OK` message).
-#[derive(Debug, Clone)]
-pub struct OkResponse {
-    /// Hex-encoded ID of the event that was acknowledged.
-    pub event_id: String,
-    /// Whether the relay accepted the event.
-    pub accepted: bool,
-    /// Human-readable reason string (empty when accepted without comment).
-    pub message: String,
-}
-
-/// Parse a raw relay text frame into a typed [`RelayMessage`].
-#[allow(clippy::result_large_err)]
-pub fn parse_relay_message(text: &str) -> Result<RelayMessage, TestClientError> {
-    let arr: Vec<Value> = serde_json::from_str(text)?;
-
-    let msg_type = arr
-        .first()
-        .and_then(|v| v.as_str())
-        .ok_or_else(|| TestClientError::UnexpectedMessage(text.to_string()))?;
-
-    match msg_type {
-        "EVENT" => {
-            let sub_id = arr
-                .get(1)
-                .and_then(|v| v.as_str())
-                .ok_or_else(|| TestClientError::UnexpectedMessage(text.to_string()))?
-                .to_string();
-            let event: Event = serde_json::from_value(
-                arr.get(2)
-                    .cloned()
-                    .ok_or_else(|| TestClientError::UnexpectedMessage(text.to_string()))?,
-            )?;
-            Ok(RelayMessage::Event {
-                subscription_id: sub_id,
-                event: Box::new(event),
-            })
-        }
-        "OK" => {
-            let event_id = arr
-                .get(1)
-                .and_then(|v| v.as_str())
-                .ok_or_else(|| TestClientError::UnexpectedMessage(text.to_string()))?
-                .to_string();
-            let accepted = arr.get(2).and_then(|v| v.as_bool()).unwrap_or(false);
-            let message = arr
-                .get(3)
-                .and_then(|v| v.as_str())
-                .unwrap_or("")
-                .to_string();
-            Ok(RelayMessage::Ok(OkResponse {
-                event_id,
-                accepted,
-                message,
-            }))
-        }
-        "EOSE" => {
-            let sub_id = arr
-                .get(1)
-                .and_then(|v| v.as_str())
-                .ok_or_else(|| TestClientError::UnexpectedMessage(text.to_string()))?
-                .to_string();
-            Ok(RelayMessage::Eose {
-                subscription_id: sub_id,
-            })
-        }
-        "CLOSED" => {
-            let sub_id = arr
-                .get(1)
-                .and_then(|v| v.as_str())
-                .ok_or_else(|| TestClientError::UnexpectedMessage(text.to_string()))?
-                .to_string();
-            let message = arr
-                .get(2)
-                .and_then(|v| v.as_str())
-                .unwrap_or("")
-                .to_string();
-            Ok(RelayMessage::Closed {
-                subscription_id: sub_id,
-                message,
-            })
-        }
-        "NOTICE" => {
-            let message = arr
-                .get(1)
-                .and_then(|v| v.as_str())
-                .unwrap_or("")
-                .to_string();
-            Ok(RelayMessage::Notice { message })
-        }
-        "AUTH" => {
-            let challenge = arr
-                .get(1)
-                .and_then(|v| v.as_str())
-                .ok_or_else(|| TestClientError::UnexpectedMessage(text.to_string()))?
-                .to_string();
-            Ok(RelayMessage::Auth { challenge })
-        }
-        other => Err(TestClientError::UnexpectedMessage(format!(
-            "unknown message type: {other}"
-        ))),
-    }
-}
 
 /// Errors returned by [`SproutTestClient`] operations.
 #[derive(Debug, Error)]
 pub enum TestClientError {
     /// A WebSocket transport error occurred.
     #[error("WebSocket error: {0}")]
-    WebSocket(#[from] tokio_tungstenite::tungstenite::Error),
+    WebSocket(tokio_tungstenite::tungstenite::Error),
 
     /// A JSON serialization or deserialization error occurred.
     #[error("JSON error: {0}")]
-    Json(#[from] serde_json::Error),
+    Json(serde_json::Error),
 
     /// Failed to build a Nostr event.
     #[error("Nostr event builder error: {0}")]
@@ -198,20 +57,32 @@ pub enum TestClientError {
     NoAuthChallenge,
 }
 
+impl From<WsClientError> for TestClientError {
+    fn from(e: WsClientError) -> Self {
+        match e {
+            WsClientError::WebSocket(e) => TestClientError::WebSocket(e),
+            WsClientError::Json(e) => TestClientError::Json(e),
+            WsClientError::EventBuilder(s) => TestClientError::EventBuilder(s),
+            WsClientError::Url(s) => TestClientError::Url(s),
+            WsClientError::Timeout => TestClientError::Timeout,
+            WsClientError::ConnectionClosed => TestClientError::ConnectionClosed,
+            WsClientError::UnexpectedMessage(s) => TestClientError::UnexpectedMessage(s),
+            WsClientError::AuthFailed(s) => TestClientError::AuthFailed(s),
+            WsClientError::EventRejected(s) => TestClientError::EventRejected(s),
+            WsClientError::NoAuthChallenge => TestClientError::NoAuthChallenge,
+        }
+    }
+}
+
 impl From<nostr::event::builder::Error> for TestClientError {
     fn from(e: nostr::event::builder::Error) -> Self {
         TestClientError::EventBuilder(e.to_string())
     }
 }
 
-type WsStream = WebSocketStream<MaybeTlsStream<tokio::net::TcpStream>>;
-
 /// WebSocket test client for integration testing against a running Sprout relay.
 pub struct SproutTestClient {
-    ws: WsStream,
+    inner: NostrWsConnection,
-    buffer: VecDeque<RelayMessage>,
-    pending_challenge: Option<String>,
-    relay_url: String,
 }
 
 impl SproutTestClient {
@@ -224,50 +95,20 @@ impl SproutTestClient {
 
     /// Connects to the relay at `url` without performing authentication.
     pub async fn connect_unauthenticated(url: &str) -> Result<Self, TestClientError> {
-        let parsed = url
+        let inner = NostrWsConnection::connect(url).await?;
-            .parse::<url::Url>()
-            .map_err(|e| TestClientError::Url(e.to_string()))?;
-
-        let (ws, _response) = connect_async(parsed.as_str())
-            .await
-            .map_err(TestClientError::WebSocket)?;
-
         debug!("connected to relay at {url}");
-
+        Ok(Self { inner })
-        Ok(Self {
-            ws,
-            buffer: VecDeque::new(),
-            pending_challenge: None,
-            relay_url: url.to_string(),
-        })
     }
 
     /// Performs NIP-42 authentication using `keys` against the connected relay.
     pub async fn authenticate(&mut self, keys: &Keys) -> Result<(), TestClientError> {
-        let challenge = self.wait_for_auth_challenge(Duration::from_secs(5)).await?;
+        self.inner.authenticate(keys, None).await?;
-
-        let relay_url =
-            RelayUrl::parse(&self.relay_url).map_err(|e| TestClientError::Url(e.to_string()))?;
-
-        let auth_event = EventBuilder::auth(&challenge, relay_url).sign_with_keys(keys)?;
-        let event_id = auth_event.id.to_hex();
-
-        self.send_raw(&json!(["AUTH", auth_event])).await?;
-
-        let ok = self.wait_for_ok(&event_id, Duration::from_secs(5)).await?;
-        if !ok.accepted {
-            return Err(TestClientError::AuthFailed(ok.message));
-        }
-
-        debug!("NIP-42 authentication successful");
         Ok(())
     }
 
     /// Sends a signed event to the relay and waits for the OK response.
     pub async fn send_event(&mut self, event: Event) -> Result<OkResponse, TestClientError> {
-        let event_id = event.id.to_hex();
+        Ok(self.inner.send_event(event).await?)
-        self.send_raw(&json!(["EVENT", event])).await?;
-        self.wait_for_ok(&event_id, Duration::from_secs(10)).await
     }
 
     /// Builds and sends a text message event to `channel_id` using the given `kind`.
@@ -296,14 +137,16 @@ impl SproutTestClient {
         msg.push(json!("REQ"));
         msg.push(json!(sub_id));
         for f in filters {
-            msg.push(serde_json::to_value(&f)?);
+            msg.push(serde_json::to_value(&f).map_err(WsClientError::Json)?);
         }
-        self.send_raw(&Value::Array(msg)).await
+        self.inner.send_raw(&Value::Array(msg)).await?;
+        Ok(())
     }
 
     /// Sends a CLOSE message to cancel the subscription identified by `sub_id`.
     pub async fn close_subscription(&mut self, sub_id: &str) -> Result<(), TestClientError> {
-        self.send_raw(&json!(["CLOSE", sub_id])).await
+        self.inner.send_raw(&json!(["CLOSE", sub_id])).await?;
+        Ok(())
     }
 
     /// Receives the next relay message, waiting up to `timeout_dur`.
@@ -311,10 +154,7 @@ impl SproutTestClient {
         &mut self,
         timeout_dur: Duration,
     ) -> Result<RelayMessage, TestClientError> {
-        if let Some(msg) = self.buffer.pop_front() {
+        Ok(self.inner.next_event(timeout_dur).await?)
-            return Ok(msg);
-        }
-        self.recv_one(timeout_dur).await
     }
 
     /// Collects all events for `sub_id` until EOSE is received, waiting up to `timeout_dur`.
@@ -326,14 +166,16 @@ impl SproutTestClient {
         let deadline = tokio::time::Instant::now() + timeout_dur;
         let mut events = Vec::new();
 
-        let old_buffer = std::mem::take(&mut self.buffer);
+        loop {
-        let mut found_eose = false;
+            let remaining = deadline
-        for msg in old_buffer {
+                .checked_duration_since(tokio::time::Instant::now())
-            if found_eose {
+                .unwrap_or(Duration::ZERO);
-                self.buffer.push_back(msg);
+
-                continue;
+            if remaining.is_zero() {
+                return Err(TestClientError::Timeout);
             }
-            match msg {
+
+            match self.inner.next_event(remaining).await? {
                 RelayMessage::Event {
                     subscription_id,
                     event,
@@ -341,213 +183,24 @@ impl SproutTestClient {
                     events.push(*event);
                 }
                 RelayMessage::Eose { subscription_id } if subscription_id == sub_id => {
-                    found_eose = true;
+                    return Ok(events);
                 }
-                other => self.buffer.push_back(other),
-            }
-        }
-        if found_eose {
-            return Ok(events);
-        }
-
-        loop {
-            let remaining = deadline
-                .checked_duration_since(tokio::time::Instant::now())
-                .unwrap_or(Duration::ZERO);
-
-            if remaining.is_zero() {
-                return Err(TestClientError::Timeout);
-            }
-
-            let raw = timeout(remaining, self.ws.next())
-                .await
-                .map_err(|_| TestClientError::Timeout)?
-                .ok_or(TestClientError::ConnectionClosed)?
-                .map_err(TestClientError::WebSocket)?;
-
-            match raw {
-                Message::Text(text) => {
-                    let msg = parse_relay_message(&text)?;
-                    match msg {
-                        RelayMessage::Event {
-                            subscription_id,
-                            event,
-                        } if subscription_id == sub_id => {
-                            events.push(*event);
-                        }
-                        RelayMessage::Eose { subscription_id } if subscription_id == sub_id => {
-                            return Ok(events);
-                        }
-                        RelayMessage::Auth { ref challenge } => {
-                            self.pending_challenge = Some(challenge.clone());
-                            self.buffer.push_back(msg);
-                        }
-                        other => self.buffer.push_back(other),
-                    }
-                }
-                Message::Ping(data) => {
-                    self.ws.send(Message::Pong(data)).await?;
-                }
-                Message::Close(_) => return Err(TestClientError::ConnectionClosed),
                 _ => {}
             }
         }
     }
 
     /// Closes the WebSocket connection gracefully.
-    pub async fn disconnect(mut self) -> Result<(), TestClientError> {
+    pub async fn disconnect(self) -> Result<(), TestClientError> {
-        self.ws.close(None).await?;
+        self.inner.disconnect().await?;
         Ok(())
     }
-
-    async fn send_raw(&mut self, value: &Value) -> Result<(), TestClientError> {
-        let text = serde_json::to_string(value)?;
-        debug!("→ relay: {text}");
-        self.ws.send(Message::Text(text.into())).await?;
-        Ok(())
-    }
-
-    async fn recv_one(&mut self, timeout_dur: Duration) -> Result<RelayMessage, TestClientError> {
-        if let Some(msg) = self.buffer.pop_front() {
-            return Ok(msg);
-        }
-
-        loop {
-            let raw = timeout(timeout_dur, self.ws.next())
-                .await
-                .map_err(|_| TestClientError::Timeout)?
-                .ok_or(TestClientError::ConnectionClosed)?
-                .map_err(TestClientError::WebSocket)?;
-
-            match raw {
-                Message::Text(text) => {
-                    let msg = parse_relay_message(&text)?;
-                    if let RelayMessage::Auth { ref challenge } = msg {
-                        self.pending_challenge = Some(challenge.clone());
-                    }
-                    return Ok(msg);
-                }
-                Message::Ping(data) => {
-                    self.ws.send(Message::Pong(data)).await?;
-                }
-                Message::Close(_) => return Err(TestClientError::ConnectionClosed),
-                _ => {}
-            }
-        }
-    }
-
-    async fn wait_for_auth_challenge(
-        &mut self,
-        timeout_dur: Duration,
-    ) -> Result<String, TestClientError> {
-        if let Some(challenge) = self.pending_challenge.take() {
-            return Ok(challenge);
-        }
-
-        if let Some(idx) = self
-            .buffer
-            .iter()
-            .position(|m| matches!(m, RelayMessage::Auth { .. }))
-        {
-            match self.buffer.remove(idx).unwrap() {
-                RelayMessage::Auth { challenge } => return Ok(challenge),
-                _ => unreachable!(),
-            }
-        }
-
-        let deadline = tokio::time::Instant::now() + timeout_dur;
-
-        loop {
-            let remaining = deadline
-                .checked_duration_since(tokio::time::Instant::now())
-                .unwrap_or(Duration::ZERO);
-
-            if remaining.is_zero() {
-                return Err(TestClientError::NoAuthChallenge);
-            }
-
-            let raw = timeout(remaining, self.ws.next())
-                .await
-                .map_err(|_| TestClientError::NoAuthChallenge)?
-                .ok_or(TestClientError::ConnectionClosed)?
-                .map_err(TestClientError::WebSocket)?;
-
-            match raw {
-                Message::Text(text) => {
-                    let msg = parse_relay_message(&text)?;
-                    match msg {
-                        RelayMessage::Auth { challenge } => return Ok(challenge),
-                        other => self.buffer.push_back(other),
-                    }
-                }
-                Message::Ping(data) => {
-                    self.ws.send(Message::Pong(data)).await?;
-                }
-                Message::Close(_) => return Err(TestClientError::ConnectionClosed),
-                _ => {}
-            }
-        }
-    }
-
-    async fn wait_for_ok(
-        &mut self,
-        event_id: &str,
-        timeout_dur: Duration,
-    ) -> Result<OkResponse, TestClientError> {
-        let deadline = tokio::time::Instant::now() + timeout_dur;
-
-        if let Some(idx) = self
-            .buffer
-            .iter()
-            .position(|m| matches!(m, RelayMessage::Ok(ok) if ok.event_id == event_id))
-        {
-            match self.buffer.remove(idx).unwrap() {
-                RelayMessage::Ok(ok) => return Ok(ok),
-                _ => unreachable!(),
-            }
-        }
-
-        loop {
-            let remaining = deadline
-                .checked_duration_since(tokio::time::Instant::now())
-                .unwrap_or(Duration::ZERO);
-
-            if remaining.is_zero() {
-                return Err(TestClientError::Timeout);
-            }
-
-            let raw = timeout(remaining, self.ws.next())
-                .await
-                .map_err(|_| TestClientError::Timeout)?
-                .ok_or(TestClientError::ConnectionClosed)?
-                .map_err(TestClientError::WebSocket)?;
-
-            match raw {
-                Message::Text(text) => {
-                    let msg = parse_relay_message(&text)?;
-                    match msg {
-                        RelayMessage::Ok(ok) if ok.event_id == event_id => return Ok(ok),
-                        RelayMessage::Auth { ref challenge } => {
-                            self.pending_challenge = Some(challenge.clone());
-                            self.buffer.push_back(msg);
-                        }
-                        other => self.buffer.push_back(other),
-                    }
-                }
-                Message::Ping(data) => {
-                    self.ws.send(Message::Pong(data)).await?;
-                }
-                Message::Close(_) => return Err(TestClientError::ConnectionClosed),
-                _ => {}
-            }
-        }
-    }
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
-    use nostr::Keys;
+    use nostr::{EventBuilder, Keys, Kind, RelayUrl, Tag};
 
     #[test]
     fn parse_relay_messages() {

crates/sprout-ws-client/Cargo.toml

@@ -0,0 +1,17 @@
+[package]
+name = "sprout-ws-client"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+repository.workspace = true
+
+[dependencies]
+nostr = { workspace = true }
+tokio = { workspace = true }
+tokio-tungstenite = { workspace = true }
+futures-util = { workspace = true }
+serde_json = { workspace = true }
+thiserror = { workspace = true }
+url = { workspace = true }
+tracing = { workspace = true }

crates/sprout-ws-client/src/connection.rs

@@ -0,0 +1,280 @@
+use std::collections::VecDeque;
+use std::time::Duration;
+
+use futures_util::{SinkExt, StreamExt};
+use nostr::{Event, Keys, Tag};
+use serde_json::{json, Value};
+use tokio::time::timeout;
+use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream};
+use tracing::debug;
+
+use crate::error::WsClientError;
+use crate::message::{build_auth_event, parse_relay_message, OkResponse, RelayMessage};
+
+type WsStream = WebSocketStream<MaybeTlsStream<tokio::net::TcpStream>>;
+
+/// A NIP-42-capable WebSocket connection to a Nostr relay.
+pub struct NostrWsConnection {
+    ws: WsStream,
+    buffer: VecDeque<RelayMessage>,
+    pending_challenge: Option<String>,
+    relay_url: String,
+}
+
+impl NostrWsConnection {
+    /// Connects to the relay at `url` and performs NIP-42 authentication with `keys`.
+    ///
+    /// Pass `auth_tag` to include a NIP-OA authorization tag in the AUTH event.
+    pub async fn connect_authenticated(
+        url: &str,
+        keys: &Keys,
+        auth_tag: Option<&Tag>,
+    ) -> Result<Self, WsClientError> {
+        let mut conn = Self::connect(url).await?;
+        conn.authenticate(keys, auth_tag).await?;
+        Ok(conn)
+    }
+
+    /// Connects to the relay at `url` without performing authentication.
+    pub async fn connect(url: &str) -> Result<Self, WsClientError> {
+        let parsed = url
+            .parse::<url::Url>()
+            .map_err(|e| WsClientError::Url(e.to_string()))?;
+
+        let (ws, _response) = connect_async(parsed.as_str())
+            .await
+            .map_err(WsClientError::WebSocket)?;
+
+        debug!("connected to relay at {url}");
+
+        Ok(Self {
+            ws,
+            buffer: VecDeque::new(),
+            pending_challenge: None,
+            relay_url: url.to_string(),
+        })
+    }
+
+    /// Performs NIP-42 authentication using `keys` against the connected relay.
+    ///
+    /// Pass `auth_tag` to include a NIP-OA authorization tag in the AUTH event.
+    pub async fn authenticate(
+        &mut self,
+        keys: &Keys,
+        auth_tag: Option<&Tag>,
+    ) -> Result<(), WsClientError> {
+        let challenge = self.wait_for_auth_challenge(Duration::from_secs(5)).await?;
+
+        let auth_event = build_auth_event(&challenge, &self.relay_url, keys, auth_tag)?;
+        let event_id = auth_event.id.to_hex();
+
+        self.send_raw(&json!(["AUTH", auth_event])).await?;
+
+        let ok = self.wait_for_ok(&event_id, Duration::from_secs(5)).await?;
+        if !ok.accepted {
+            return Err(WsClientError::AuthFailed(ok.message));
+        }
+
+        debug!("NIP-42 authentication successful");
+        Ok(())
+    }
+
+    /// Sends a signed event to the relay and waits for the OK response.
+    pub async fn send_event(&mut self, event: Event) -> Result<OkResponse, WsClientError> {
+        let event_id = event.id.to_hex();
+        self.send_raw(&json!(["EVENT", event])).await?;
+        self.wait_for_ok(&event_id, Duration::from_secs(10)).await
+    }
+
+    /// Receives the next relay message, waiting up to `timeout_dur`.
+    pub async fn next_event(
+        &mut self,
+        timeout_dur: Duration,
+    ) -> Result<RelayMessage, WsClientError> {
+        if let Some(msg) = self.buffer.pop_front() {
+            return Ok(msg);
+        }
+        self.recv_one(timeout_dur).await
+    }
+
+    /// Closes the WebSocket connection gracefully.
+    pub async fn disconnect(mut self) -> Result<(), WsClientError> {
+        self.ws.close(None).await?;
+        Ok(())
+    }
+
+    /// Sends a raw JSON value as a WebSocket text frame.
+    pub async fn send_raw(&mut self, value: &Value) -> Result<(), WsClientError> {
+        let text = serde_json::to_string(value)?;
+        debug!("→ relay: {text}");
+        self.ws.send(Message::Text(text.into())).await?;
+        Ok(())
+    }
+
+    async fn recv_one(&mut self, timeout_dur: Duration) -> Result<RelayMessage, WsClientError> {
+        if let Some(msg) = self.buffer.pop_front() {
+            return Ok(msg);
+        }
+
+        loop {
+            let raw = timeout(timeout_dur, self.ws.next())
+                .await
+                .map_err(|_| WsClientError::Timeout)?
+                .ok_or(WsClientError::ConnectionClosed)?
+                .map_err(WsClientError::WebSocket)?;
+
+            match raw {
+                Message::Text(text) => {
+                    let msg = parse_relay_message(&text)?;
+                    if let RelayMessage::Auth { ref challenge } = msg {
+                        self.pending_challenge = Some(challenge.clone());
+                    }
+                    return Ok(msg);
+                }
+                Message::Ping(data) => {
+                    self.ws.send(Message::Pong(data)).await?;
+                }
+                Message::Close(_) => return Err(WsClientError::ConnectionClosed),
+                _ => {}
+            }
+        }
+    }
+
+    async fn wait_for_auth_challenge(
+        &mut self,
+        timeout_dur: Duration,
+    ) -> Result<String, WsClientError> {
+        if let Some(challenge) = self.pending_challenge.take() {
+            return Ok(challenge);
+        }
+
+        if let Some(idx) = self
+            .buffer
+            .iter()
+            .position(|m| matches!(m, RelayMessage::Auth { .. }))
+        {
+            match self.buffer.remove(idx).unwrap() {
+                RelayMessage::Auth { challenge } => return Ok(challenge),
+                _ => unreachable!(),
+            }
+        }
+
+        let deadline = tokio::time::Instant::now() + timeout_dur;
+
+        loop {
+            let remaining = deadline
+                .checked_duration_since(tokio::time::Instant::now())
+                .unwrap_or(Duration::ZERO);
+
+            if remaining.is_zero() {
+                return Err(WsClientError::NoAuthChallenge);
+            }
+
+            let raw = timeout(remaining, self.ws.next())
+                .await
+                .map_err(|_| WsClientError::NoAuthChallenge)?
+                .ok_or(WsClientError::ConnectionClosed)?
+                .map_err(WsClientError::WebSocket)?;
+
+            match raw {
+                Message::Text(text) => {
+                    let msg = parse_relay_message(&text)?;
+                    match msg {
+                        RelayMessage::Auth { challenge } => {
+                            if challenge.len() > 1024 {
+                                return Err(WsClientError::AuthFailed(
+                                    "challenge exceeds 1024 bytes".into(),
+                                ));
+                            }
+                            return Ok(challenge);
+                        }
+                        other => self.buffer.push_back(other),
+                    }
+                }
+                Message::Ping(data) => {
+                    self.ws.send(Message::Pong(data)).await?;
+                }
+                Message::Close(_) => return Err(WsClientError::ConnectionClosed),
+                _ => {}
+            }
+        }
+    }
+
+    async fn wait_for_ok(
+        &mut self,
+        event_id: &str,
+        timeout_dur: Duration,
+    ) -> Result<OkResponse, WsClientError> {
+        let deadline = tokio::time::Instant::now() + timeout_dur;
+
+        if let Some(idx) = self
+            .buffer
+            .iter()
+            .position(|m| matches!(m, RelayMessage::Ok(ok) if ok.event_id == event_id))
+        {
+            match self.buffer.remove(idx).unwrap() {
+                RelayMessage::Ok(ok) => return Ok(ok),
+                _ => unreachable!(),
+            }
+        }
+
+        loop {
+            let remaining = deadline
+                .checked_duration_since(tokio::time::Instant::now())
+                .unwrap_or(Duration::ZERO);
+
+            if remaining.is_zero() {
+                return Err(WsClientError::Timeout);
+            }
+
+            let raw = timeout(remaining, self.ws.next())
+                .await
+                .map_err(|_| WsClientError::Timeout)?
+                .ok_or(WsClientError::ConnectionClosed)?
+                .map_err(WsClientError::WebSocket)?;
+
+            match raw {
+                Message::Text(text) => {
+                    let msg = parse_relay_message(&text)?;
+                    match msg {
+                        RelayMessage::Ok(ok) if ok.event_id == event_id => return Ok(ok),
+                        RelayMessage::Auth { ref challenge } => {
+                            self.pending_challenge = Some(challenge.clone());
+                            self.buffer.push_back(msg);
+                        }
+                        other => self.buffer.push_back(other),
+                    }
+                }
+                Message::Ping(data) => {
+                    self.ws.send(Message::Pong(data)).await?;
+                }
+                Message::Close(_) => return Err(WsClientError::ConnectionClosed),
+                _ => {}
+            }
+        }
+    }
+}
+
+/// One-shot helper: connect, authenticate, send one event, disconnect.
+///
+/// Establishes a fresh WebSocket connection, completes NIP-42 authentication,
+/// publishes `event`, waits for the relay's OK response, then closes the
+/// connection. The entire operation is bounded by `timeout_secs`.
+pub async fn publish_event(
+    relay_url: &str,
+    event: Event,
+    keys: &Keys,
+    auth_tag: Option<&Tag>,
+    timeout_secs: u64,
+) -> Result<OkResponse, WsClientError> {
+    let result = tokio::time::timeout(Duration::from_secs(timeout_secs), async {
+        let mut conn = NostrWsConnection::connect(relay_url).await?;
+        conn.authenticate(keys, auth_tag).await?;
+        let ok = conn.send_event(event).await?;
+        let _ = conn.disconnect().await;
+        Ok::<_, WsClientError>(ok)
+    })
+    .await
+    .map_err(|_| WsClientError::Timeout)?;
+    result
+}

crates/sprout-ws-client/src/error.rs

@@ -0,0 +1,51 @@
+use thiserror::Error;
+
+/// Errors returned by [`crate::NostrWsConnection`] and related operations.
+#[derive(Debug, Error)]
+pub enum WsClientError {
+    /// A WebSocket transport error occurred.
+    #[error("WebSocket error: {0}")]
+    WebSocket(#[from] tokio_tungstenite::tungstenite::Error),
+
+    /// A JSON serialization or deserialization error occurred.
+    #[error("JSON error: {0}")]
+    Json(#[from] serde_json::Error),
+
+    /// Failed to build a Nostr event.
+    #[error("Nostr event builder error: {0}")]
+    EventBuilder(String),
+
+    /// Failed to parse a URL.
+    #[error("URL parse error: {0}")]
+    Url(String),
+
+    /// The relay did not respond within the expected time.
+    #[error("Timeout waiting for relay message")]
+    Timeout,
+
+    /// The WebSocket connection was closed before the operation completed.
+    #[error("Connection closed unexpectedly")]
+    ConnectionClosed,
+
+    /// The relay sent a message that was not expected at this point.
+    #[error("Unexpected relay message: {0}")]
+    UnexpectedMessage(String),
+
+    /// NIP-42 authentication was rejected by the relay.
+    #[error("Authentication failed: {0}")]
+    AuthFailed(String),
+
+    /// The relay rejected the submitted event.
+    #[error("Event rejected by relay: {0}")]
+    EventRejected(String),
+
+    /// No NIP-42 AUTH challenge was received from the relay.
+    #[error("No AUTH challenge received from relay")]
+    NoAuthChallenge,
+}
+
+impl From<nostr::event::builder::Error> for WsClientError {
+    fn from(e: nostr::event::builder::Error) -> Self {
+        WsClientError::EventBuilder(e.to_string())
+    }
+}

crates/sprout-ws-client/src/lib.rs

@@ -0,0 +1,9 @@
+#![deny(unsafe_code)]
+
+pub mod connection;
+pub mod error;
+pub mod message;
+
+pub use connection::{publish_event, NostrWsConnection};
+pub use error::WsClientError;
+pub use message::{build_auth_event, parse_relay_message, OkResponse, RelayMessage};

crates/sprout-ws-client/src/message.rs

@@ -0,0 +1,167 @@
+use nostr::{Event, EventBuilder, Keys, RelayUrl, Tag};
+use serde_json::Value;
+
+use crate::error::WsClientError;
+
+/// A message received from a Nostr relay.
+#[derive(Debug, Clone)]
+pub enum RelayMessage {
+    /// An event matching an active subscription.
+    Event {
+        /// The subscription ID this event belongs to.
+        subscription_id: String,
+        /// The Nostr event payload.
+        event: Box<Event>,
+    },
+    /// Acknowledgement of a published event.
+    Ok(OkResponse),
+    /// End-of-stored-events marker for a subscription.
+    Eose {
+        /// The subscription ID that has reached end-of-stored-events.
+        subscription_id: String,
+    },
+    /// The relay closed a subscription, usually with an error.
+    Closed {
+        /// The subscription ID that was closed.
+        subscription_id: String,
+        /// Human-readable reason for the closure.
+        message: String,
+    },
+    /// A human-readable notice from the relay.
+    Notice {
+        /// The notice text.
+        message: String,
+    },
+    /// A NIP-42 authentication challenge from the relay.
+    Auth {
+        /// The challenge string to sign.
+        challenge: String,
+    },
+}
+
+/// The relay's response to a published event (NIP-01 `OK` message).
+#[derive(Debug, Clone)]
+pub struct OkResponse {
+    /// Hex-encoded ID of the event that was acknowledged.
+    pub event_id: String,
+    /// Whether the relay accepted the event.
+    pub accepted: bool,
+    /// Human-readable reason string (empty when accepted without comment).
+    pub message: String,
+}
+
+/// Parse a raw relay text frame into a typed [`RelayMessage`].
+#[allow(clippy::result_large_err)]
+pub fn parse_relay_message(text: &str) -> Result<RelayMessage, WsClientError> {
+    let arr: Vec<Value> = serde_json::from_str(text)?;
+
+    let msg_type = arr
+        .first()
+        .and_then(|v| v.as_str())
+        .ok_or_else(|| WsClientError::UnexpectedMessage(text.to_string()))?;
+
+    match msg_type {
+        "EVENT" => {
+            let sub_id = arr
+                .get(1)
+                .and_then(|v| v.as_str())
+                .ok_or_else(|| WsClientError::UnexpectedMessage(text.to_string()))?
+                .to_string();
+            let event: Event = serde_json::from_value(
+                arr.get(2)
+                    .cloned()
+                    .ok_or_else(|| WsClientError::UnexpectedMessage(text.to_string()))?,
+            )?;
+            Ok(RelayMessage::Event {
+                subscription_id: sub_id,
+                event: Box::new(event),
+            })
+        }
+        "OK" => {
+            let event_id = arr
+                .get(1)
+                .and_then(|v| v.as_str())
+                .ok_or_else(|| WsClientError::UnexpectedMessage(text.to_string()))?
+                .to_string();
+            let accepted = arr.get(2).and_then(|v| v.as_bool()).unwrap_or(false);
+            let message = arr
+                .get(3)
+                .and_then(|v| v.as_str())
+                .unwrap_or("")
+                .to_string();
+            Ok(RelayMessage::Ok(OkResponse {
+                event_id,
+                accepted,
+                message,
+            }))
+        }
+        "EOSE" => {
+            let sub_id = arr
+                .get(1)
+                .and_then(|v| v.as_str())
+                .ok_or_else(|| WsClientError::UnexpectedMessage(text.to_string()))?
+                .to_string();
+            Ok(RelayMessage::Eose {
+                subscription_id: sub_id,
+            })
+        }
+        "CLOSED" => {
+            let sub_id = arr
+                .get(1)
+                .and_then(|v| v.as_str())
+                .ok_or_else(|| WsClientError::UnexpectedMessage(text.to_string()))?
+                .to_string();
+            let message = arr
+                .get(2)
+                .and_then(|v| v.as_str())
+                .unwrap_or("")
+                .to_string();
+            Ok(RelayMessage::Closed {
+                subscription_id: sub_id,
+                message,
+            })
+        }
+        "NOTICE" => {
+            let message = arr
+                .get(1)
+                .and_then(|v| v.as_str())
+                .unwrap_or("")
+                .to_string();
+            Ok(RelayMessage::Notice { message })
+        }
+        "AUTH" => {
+            let challenge = arr
+                .get(1)
+                .and_then(|v| v.as_str())
+                .ok_or_else(|| WsClientError::UnexpectedMessage(text.to_string()))?
+                .to_string();
+            Ok(RelayMessage::Auth { challenge })
+        }
+        other => Err(WsClientError::UnexpectedMessage(format!(
+            "unknown message type: {other}"
+        ))),
+    }
+}
+
+/// Builds a NIP-42 AUTH event, optionally injecting a NIP-OA auth tag.
+///
+/// The `auth_tag` parameter allows callers to attach a workspace-scoped
+/// authorization tag (e.g. `["auth", "<token>"]`) alongside the standard
+/// relay and challenge tags required by NIP-42.
+pub fn build_auth_event(
+    challenge: &str,
+    relay_url: &str,
+    keys: &Keys,
+    auth_tag: Option<&Tag>,
+) -> Result<Event, WsClientError> {
+    let url = RelayUrl::parse(relay_url).map_err(|e| WsClientError::Url(e.to_string()))?;
+    let builder = EventBuilder::auth(challenge, url);
+    let builder = if let Some(tag) = auth_tag {
+        builder.tags([tag.clone()])
+    } else {
+        builder
+    };
+    builder
+        .sign_with_keys(keys)
+        .map_err(|e| WsClientError::EventBuilder(e.to_string()))
+}