jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-27 04:30:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source
This commit is contained in:
Will Pfleger
2026-06-05 16:26:30 -04:00
committed by GitHub
parent 2f50011bdd
commit ef98ae942a
13 changed files with 635 additions and 399 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
AGENTS.md
View File

@@ -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

16
Cargo.lock generated
View File

@@ -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"

2
Cargo.toml
View File

@@ -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`

3
crates/sprout-cli/Cargo.toml
View File

@@ -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" }

37
crates/sprout-cli/src/client.rs
View File

@@ -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
// ---------------------------------------------------------------------------

11
crates/sprout-cli/src/commands/users.rs
View File

@@ -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.
/// The CLI uses the HTTP bridge (POST /events) which rejects ephemeral kinds.
/// This will fail until the CLI gains a WS publish path. The kind is correct
/// per the protocol spec (KIND_PRESENCE_UPDATE = 20001).
/// Kind 20001 is ephemeral and only accepted via WebSocket connections. This
/// method connects to the relay over WS, performs NIP-42 authentication, and
/// publishes the event directly — bypassing the HTTP bridge.
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(())
}

1
crates/sprout-test-client/Cargo.toml
View File

@@ -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 }

439
crates/sprout-test-client/src/lib.rs
View File

@@ -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, RelayUrl, Tag};
use nostr::{Event, EventBuilder, Filter, Keys, Kind, 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.
#[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, 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}"
))),
}
}
use sprout_ws_client::NostrWsConnection;
pub use sprout_ws_client::{parse_relay_message, OkResponse, RelayMessage, WsClientError};
/// 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,
buffer: VecDeque<RelayMessage>,
pending_challenge: Option<String>,
relay_url: String,
inner: NostrWsConnection,
}
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
.parse::<url::Url>()
.map_err(|e| TestClientError::Url(e.to_string()))?;
let (ws, _response) = connect_async(parsed.as_str())
.await
.map_err(TestClientError::WebSocket)?;
let inner = NostrWsConnection::connect(url).await?;
debug!("connected to relay at {url}");
Ok(Self {
ws,
buffer: VecDeque::new(),
pending_challenge: None,
relay_url: url.to_string(),
})
Ok(Self { inner })
}
/// 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?;
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");
self.inner.authenticate(keys, None).await?;
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();
self.send_raw(&json!(["EVENT", event])).await?;
self.wait_for_ok(&event_id, Duration::from_secs(10)).await
Ok(self.inner.send_event(event).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() {
return Ok(msg);
}
self.recv_one(timeout_dur).await
Ok(self.inner.next_event(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);
let mut found_eose = false;
for msg in old_buffer {
if found_eose {
self.buffer.push_back(msg);
continue;
loop {
let remaining = deadline
.checked_duration_since(tokio::time::Instant::now())
.unwrap_or(Duration::ZERO);
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> {
self.ws.close(None).await?;
pub async fn disconnect(self) -> Result<(), TestClientError> {
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() {

17
crates/sprout-ws-client/Cargo.toml Normal file
View File

@@ -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 }

280
crates/sprout-ws-client/src/connection.rs Normal file
View File

@@ -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
}

51
crates/sprout-ws-client/src/error.rs Normal file
View File

@@ -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())
}
}

9
crates/sprout-ws-client/src/lib.rs Normal file
View File

@@ -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};

167
crates/sprout-ws-client/src/message.rs Normal file
View File

@@ -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()))
}