From cc9b0b22c41c7a1d7bd5ba5cd684aa9e25a2e84c Mon Sep 17 00:00:00 2001 From: Michael Neale Date: Wed, 3 Jun 2026 07:16:12 +1000 Subject: [PATCH] refactor(serverless): back agent relay layer with nostr-relay-pool MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The serverless agent path hand-rolled multi-relay subscribe/reconnect/ resubscribe/dedup in relay.rs — reimplementing what nostr-relay-pool (rust-nostr) already does, with gaps that caused the agent to miss messages when a single relay 502'd mid-session. Replace the serverless backend with nostr-relay-pool's RelayPool, which provides the standard Nostr client model used by damus/0xchat/nostr-tools: - persistent REQ per relay, fanned out across all relays - auto-reconnect (default) with jittered backoff - auto-resubscribe on reconnect (relays replay stored events on REQ) - merge + dedup across relays (Event notification = first-seen only) HarnessRelay now dispatches to ServerlessRelay when serverless; server mode (NIP-42 AUTH + HTTP bridge) is unchanged. RestClient and the typing/ presence publisher share the pool. Subscription IDs match the server-mode scheme so event routing is identical. Proven: e2e_agent_responds passes 3x consecutively (~3s each) against live public relays, surviving damus 503s via nos.lol. - 272 unit tests pass, clippy clean --- Cargo.lock | 164 ++++++++++++- Cargo.toml | 4 + crates/sprout-acp/Cargo.toml | 1 + crates/sprout-acp/src/lib.rs | 1 + crates/sprout-acp/src/relay.rs | 137 ++++++++++- crates/sprout-acp/src/serverless_relay.rs | 279 ++++++++++++++++++++++ 6 files changed, 569 insertions(+), 17 deletions(-) create mode 100644 crates/sprout-acp/src/serverless_relay.rs diff --git a/Cargo.lock b/Cargo.lock index 0159860fc3..8263b99253 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -153,6 +153,37 @@ dependencies = [ "syn", ] +[[package]] +name = "async-utility" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a34a3b57207a7a1007832416c3e4862378c8451b4e8e093e436f48c2d3d2c151" +dependencies = [ + "futures-util", + "gloo-timers", + "tokio", + "wasm-bindgen-futures", +] + +[[package]] +name = "async-wsocket" +version = "0.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1c92385c7c8b3eb2de1b78aeca225212e4c9a69a78b802832759b108681a5069" +dependencies = [ + "async-utility", + "futures", + "futures-util", + "js-sys", + "tokio", + "tokio-rustls", + "tokio-socks", + "tokio-tungstenite 0.26.2", + "url", + "wasm-bindgen", + "web-sys", +] + [[package]] name = "atoi" version = "2.0.0" @@ -162,6 +193,12 @@ dependencies = [ "num-traits", ] +[[package]] +name = "atomic-destructor" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ef49f5882e4b6afaac09ad239a4f8c70a24b8f2b0897edb1f706008efd109cf4" + [[package]] name = "atomic-waker" version = "1.1.2" @@ -268,7 +305,7 @@ dependencies = [ "sha1 0.10.6", "sync_wrapper", "tokio", - "tokio-tungstenite", + "tokio-tungstenite 0.29.0", "tower", "tower-layer", "tower-service", @@ -727,7 +764,7 @@ dependencies = [ "serde_json", "sprout-sdk", "tokio", - "tokio-tungstenite", + "tokio-tungstenite 0.29.0", "url", ] @@ -1386,6 +1423,18 @@ dependencies = [ "regex-syntax", ] +[[package]] +name = "gloo-timers" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbb143cf96099802033e0d4f4963b19fd2e0b728bcf076cd9cf7f6634f092994" +dependencies = [ + "futures-channel", + "futures-core", + "js-sys", + "wasm-bindgen", +] + [[package]] name = "h2" version = "0.4.14" @@ -2040,6 +2089,12 @@ dependencies = [ "tracing-subscriber", ] +[[package]] +name = "lru" +version = "0.16.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f66e8d5d03f609abc3a39e6f08e4164ebf1447a732906d39eb9b99b7919ef39" + [[package]] name = "lru-slab" version = "0.1.2" @@ -2237,6 +2292,12 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "27b02d87554356db9e9a873add8782d4ea6e3e58ea071a9adb9a2e8ddb884a8b" +[[package]] +name = "negentropy" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0efe882e02d206d8d279c20eb40e03baf7cb5136a1476dc084a324fbc3ec42d" + [[package]] name = "nix" version = "0.31.3" @@ -2273,6 +2334,35 @@ dependencies = [ "url", ] +[[package]] +name = "nostr-database" +version = "0.44.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7462c9d8ae5ef6a28d66a192d399ad2530f1f2130b13186296dbb11bdef5b3d1" +dependencies = [ + "lru", + "nostr", + "tokio", +] + +[[package]] +name = "nostr-relay-pool" +version = "0.44.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91b2c039df4f96c4bf7dae52a74fd5516ad6dda83a11c0c69dea91b5255a4f37" +dependencies = [ + "async-utility", + "async-wsocket", + "atomic-destructor", + "hex", + "lru", + "negentropy", + "nostr", + "nostr-database", + "tokio", + "tracing", +] + [[package]] name = "ntapi" version = "0.4.3" @@ -3649,6 +3739,7 @@ dependencies = [ "hex", "nix", "nostr", + "nostr-relay-pool", "reqwest 0.13.3", "rustls", "serde", @@ -3659,7 +3750,7 @@ dependencies = [ "sprout-sdk", "thiserror 2.0.18", "tokio", - "tokio-tungstenite", + "tokio-tungstenite 0.29.0", "tokio-util", "toml", "tracing", @@ -3765,7 +3856,7 @@ dependencies = [ "sprout-sdk", "thiserror 2.0.18", "tokio", - "tokio-tungstenite", + "tokio-tungstenite 0.29.0", "url", "uuid", ] @@ -3856,7 +3947,7 @@ dependencies = [ "sprout-sdk", "thiserror 2.0.18", "tokio", - "tokio-tungstenite", + "tokio-tungstenite 0.29.0", "tracing", "tracing-subscriber", "url", @@ -3904,7 +3995,7 @@ dependencies = [ "serde_json", "sha2 0.11.0", "tokio", - "tokio-tungstenite", + "tokio-tungstenite 0.29.0", "tokio-util", ] @@ -3920,7 +4011,7 @@ dependencies = [ "sprout-core", "thiserror 2.0.18", "tokio", - "tokio-tungstenite", + "tokio-tungstenite 0.29.0", "url", "zeroize", ] @@ -3956,7 +4047,7 @@ dependencies = [ "sprout-core", "thiserror 2.0.18", "tokio", - "tokio-tungstenite", + "tokio-tungstenite 0.29.0", "tower-http", "tracing", "tracing-subscriber", @@ -4081,7 +4172,7 @@ dependencies = [ "sprout-mcp", "thiserror 2.0.18", "tokio", - "tokio-tungstenite", + "tokio-tungstenite 0.29.0", "tracing", "tracing-subscriber", "url", @@ -4541,6 +4632,18 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-socks" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7e2948f60dbe26b35f2c7fb74ac2854c1fddded0fe9d7548fcc674a246f7615" +dependencies = [ + "either", + "futures-util", + "thiserror 1.0.69", + "tokio", +] + [[package]] name = "tokio-stream" version = "0.1.18" @@ -4552,6 +4655,22 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-tungstenite" +version = "0.26.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7a9daff607c6d2bf6c16fd681ccb7eecc83e4e2cdc1ca067ffaadfca5de7f084" +dependencies = [ + "futures-util", + "log", + "rustls", + "rustls-pki-types", + "tokio", + "tokio-rustls", + "tungstenite 0.26.2", + "webpki-roots 0.26.11", +] + [[package]] name = "tokio-tungstenite" version = "0.29.0" @@ -4564,7 +4683,7 @@ dependencies = [ "rustls-pki-types", "tokio", "tokio-rustls", - "tungstenite", + "tungstenite 0.29.0", "webpki-roots 0.26.11", ] @@ -4759,6 +4878,25 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "tungstenite" +version = "0.26.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4793cb5e56680ecbb1d843515b23b6de9a75eb04b66643e256a396d43be33c13" +dependencies = [ + "bytes", + "data-encoding", + "http", + "httparse", + "log", + "rand 0.9.4", + "rustls", + "rustls-pki-types", + "sha1 0.10.6", + "thiserror 2.0.18", + "utf-8", +] + [[package]] name = "tungstenite" version = "0.29.0" @@ -4863,6 +5001,12 @@ version = "2.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "daf8dba3b7eb870caf1ddeed7bc9d2a049f3cfdfae7cb521b087cc33ae4c49da" +[[package]] +name = "utf-8" +version = "0.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9" + [[package]] name = "utf8_iter" version = "1.0.4" diff --git a/Cargo.toml b/Cargo.toml index 13fd1e2f8d..4b8a2c3047 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -57,6 +57,10 @@ deadpool-redis = { version = "0.23", features = ["rt_tokio_1"] } # Nostr nostr = { version = "0.44", features = ["nip44", "nip59", "nip98"] } +# High-level relay pool: persistent multi-relay subscriptions with built-in +# auto-reconnect + auto-resubscribe (relays replay stored events on REQ). Used +# for the serverless agent path instead of hand-rolling reconnect/resubscribe. +nostr-relay-pool = { version = "0.44", default-features = false } # Serialization serde = { version = "1", features = ["derive"] } diff --git a/crates/sprout-acp/Cargo.toml b/crates/sprout-acp/Cargo.toml index e3f2505c25..1b7783f93d 100644 --- a/crates/sprout-acp/Cargo.toml +++ b/crates/sprout-acp/Cargo.toml @@ -23,6 +23,7 @@ sprout-persona = { path = "../sprout-persona" } # Nostr nostr = { workspace = true } +nostr-relay-pool = { workspace = true } # Async runtime tokio = { workspace = true } diff --git a/crates/sprout-acp/src/lib.rs b/crates/sprout-acp/src/lib.rs index 486f95e9f6..8903ebfbc5 100644 --- a/crates/sprout-acp/src/lib.rs +++ b/crates/sprout-acp/src/lib.rs @@ -8,6 +8,7 @@ mod observer; mod pool; mod queue; mod relay; +mod serverless_relay; use std::collections::{HashMap, HashSet}; use std::sync::Arc; diff --git a/crates/sprout-acp/src/relay.rs b/crates/sprout-acp/src/relay.rs index e70852506b..b7326bb561 100644 --- a/crates/sprout-acp/src/relay.rs +++ b/crates/sprout-acp/src/relay.rs @@ -114,6 +114,10 @@ pub struct RestClient { /// down or rate-limiting — the standard Nostr client model (damus /// `RelayPool`, nostr-tools `SimplePool`). pub ws_urls: Vec, + /// Shared serverless relay pool. When `Some`, `query`/`submit_event` use the + /// pool (auto-reconnect, multi-relay merge/dedup) instead of the hand-rolled + /// `query_ws`/`submit_event_ws`. `None` in server mode. + pub pool: Option, } /// Whether an HTTP status code is retriable (transient server/rate-limit errors). @@ -274,6 +278,19 @@ impl RestClient { /// Accepts a slice of `nostr::Filter` (serialized as JSON array). /// Returns the events as a `serde_json::Value` (JSON array of event objects). pub async fn query(&self, filters: &[nostr::Filter]) -> Result { + if let Some(pool) = &self.pool { + let events = pool + .fetch_events( + filters.to_vec(), + std::time::Duration::from_secs(8), + nostr_relay_pool::relay::ReqExitPolicy::ExitOnEOSE, + ) + .await + .map_err(|e| RelayError::Http(format!("fetch failed: {e}")))?; + let events: Vec = events.into_iter().collect(); + return serde_json::to_value(events) + .map_err(|e| RelayError::Http(format!("event serialize error: {e}"))); + } if self.serverless { return self.query_ws(filters).await; } @@ -289,6 +306,13 @@ impl RestClient { /// /// The event must already be signed. Returns the relay response JSON. pub async fn submit_event(&self, event: &Event) -> Result { + if let Some(pool) = &self.pool { + let output = pool + .send_event(event) + .await + .map_err(|e| RelayError::Http(format!("publish failed: {e}")))?; + return Ok(serde_json::json!({"accepted": !output.success.is_empty()})); + } if self.serverless { return self.submit_event_ws(event).await; } @@ -686,13 +710,13 @@ enum RelayMessage { // ── Commands sent from HarnessRelay to the background task ─────────────────── /// Subscription ID for the global membership notification subscription. -const MEMBERSHIP_NOTIF_SUB_ID: &str = "membership-notif"; +pub(crate) const MEMBERSHIP_NOTIF_SUB_ID: &str = "membership-notif"; /// Subscription ID for encrypted owner-to-agent observer control frames. -const OBSERVER_CONTROL_SUB_ID: &str = "agent-observer-control"; +pub(crate) const OBSERVER_CONTROL_SUB_ID: &str = "agent-observer-control"; /// Subscription ID for the agent's NIP-17 gift-wrap inbox (encrypted serverless /// private channels / DMs). The relay only stores `kind 1059` blobs addressed /// by `#p`; the agent unwraps each and routes the inner message by its `h` tag. -const GIFT_WRAP_SUB_ID: &str = "agent-gift-wrap-inbox"; +pub(crate) const GIFT_WRAP_SUB_ID: &str = "agent-gift-wrap-inbox"; /// NIP-59 gift wrap kind. const KIND_GIFT_WRAP: u16 = 1059; @@ -764,17 +788,32 @@ pub struct HarnessRelay { /// Handles to all background tasks (one per relay + the command fan-out), /// for clean shutdown. bg_handles: Vec>, + /// Serverless backend (nostr-relay-pool). When `Some`, all subscribe / + /// query / publish / next_event calls dispatch here instead of the + /// hand-rolled background-task path above. Built only in serverless mode; + /// server mode leaves this `None` and uses the original code unchanged. + serverless_backend: Option, } -/// Cloneable publisher handle for signed events on the relay background socket. +/// Cloneable publisher handle for signed events. In server mode it sends to the +/// relay background task via `cmd_tx`; in serverless it publishes through the +/// shared `RelayPool` (clone-cheap, Arc inside). #[derive(Clone)] pub struct RelayEventPublisher { cmd_tx: mpsc::Sender, + pool: Option, } impl RelayEventPublisher { - /// Publish a signed event through the relay background task. + /// Publish a signed event through the relay background task (server) or the + /// pool (serverless). pub async fn publish_event(&self, event: Event) -> Result<(), RelayError> { + if let Some(pool) = &self.pool { + pool.send_event(&event) + .await + .map_err(|e| RelayError::Http(format!("publish failed: {e}")))?; + return Ok(()); + } self.cmd_tx .send(RelayCommand::PublishEvent { event: Box::new(event), @@ -800,6 +839,42 @@ impl HarnessRelay { auth_tag: Option, serverless: bool, ) -> Result { + // Serverless: delegate the entire relay layer to nostr-relay-pool + // (persistent multi-relay REQ + auto-reconnect + auto-resubscribe + + // merge/dedup). The hand-rolled background-task path below is used ONLY + // for server mode (NIP-42 AUTH against the Sprout relay). + if serverless { + let backend = crate::serverless_relay::ServerlessRelay::connect( + relay_url, + keys, + agent_pubkey_hex, + ) + .await?; + let relay_urls = backend.relay_urls().to_vec(); + let primary = relay_urls + .first() + .cloned() + .unwrap_or_else(|| relay_url.to_string()); + // Unused channels/clients in serverless mode — the backend owns the + // real streams. We keep the struct shape uniform. + let (_dead_event_tx, event_rx) = mpsc::channel::>(1); + let (cmd_tx, _dead_cmd_rx) = mpsc::channel::(1); + return Ok(Self { + event_rx, + observer_control_rx: None, + cmd_tx, + http: reqwest::Client::new(), + relay_url: primary, + relay_urls, + keys: keys.clone(), + auth_tag, + serverless, + seen_event_ids: std::collections::VecDeque::new(), + bg_handles: Vec::new(), + serverless_backend: Some(backend), + }); + } + // Serverless workspaces pass a comma-separated relay LIST // (`wss://a,wss://b,...`). Connect to ALL of them and merge — this is // the standard Nostr client pattern (damus `RelayPool`, nostr-tools @@ -931,6 +1006,7 @@ impl HarnessRelay { serverless, seen_event_ids: std::collections::VecDeque::new(), bg_handles, + serverless_backend: None, }) } @@ -1063,6 +1139,7 @@ impl HarnessRelay { } else { self.relay_urls.clone() }, + pool: self.serverless_backend.as_ref().map(|b| b.pool()), } } @@ -1075,6 +1152,9 @@ impl HarnessRelay { channel_id: Uuid, filter: ChannelFilter, ) -> Result<(), RelayError> { + if let Some(backend) = &self.serverless_backend { + return backend.subscribe_channel(channel_id, filter).await; + } self.cmd_tx .send(RelayCommand::Subscribe { channel_id, filter }) .await @@ -1085,6 +1165,13 @@ impl HarnessRelay { /// Subscribe to membership notifications for this agent. pub async fn subscribe_membership_notifications(&mut self) -> Result<(), RelayError> { + if let Some(backend) = &self.serverless_backend { + // Generic relays don't emit Sprout's kind:44100 membership push + // (that's a server side effect). In serverless the harness relies + // on the periodic re-discovery loop instead — nothing to subscribe. + // Still subscribe the gift-wrap inbox so encrypted DMs/privates work. + return backend.subscribe_gift_wrap().await; + } self.cmd_tx .send(RelayCommand::SubscribeMembership) .await @@ -1094,6 +1181,10 @@ impl HarnessRelay { /// Subscribe to encrypted observer control frames addressed to this agent. pub async fn subscribe_observer_controls(&mut self) -> Result<(), RelayError> { + if self.serverless_backend.is_some() { + // Observer controls are a Sprout-relay feature; no-op in serverless. + return Ok(()); + } self.cmd_tx .send(RelayCommand::SubscribeObserverControls) .await @@ -1110,11 +1201,16 @@ impl HarnessRelay { pub fn event_publisher(&self) -> RelayEventPublisher { RelayEventPublisher { cmd_tx: self.cmd_tx.clone(), + pool: self.serverless_backend.as_ref().map(|b| b.pool()), } } /// Unsubscribe from a channel. pub async fn unsubscribe_channel(&mut self, channel_id: Uuid) -> Result<(), RelayError> { + if let Some(backend) = &self.serverless_backend { + backend.unsubscribe_channel(channel_id).await; + return Ok(()); + } self.cmd_tx .send(RelayCommand::Unsubscribe { channel_id }) .await @@ -1128,6 +1224,10 @@ impl HarnessRelay { /// Reads from the background task's event channel. Returns `None` on /// connection loss — the caller should call [`reconnect`](Self::reconnect). pub async fn next_event(&mut self) -> Option { + if let Some(backend) = &mut self.serverless_backend { + // The pool already dedups across relays — no local ring needed. + return backend.next_event().await; + } // The background tasks send `None` to signal connection loss. With // multiple relays the same event arrives from each relay it's stored on, // so dedup by event id (bounded ring, like damus's `seenEvents`). @@ -1152,6 +1252,9 @@ impl HarnessRelay { /// (typing indicators) prefer [`try_publish_event`] which never blocks. #[allow(dead_code)] // Public API — callers outside the harness may use this pub async fn publish_event(&self, event: Event) -> Result<(), RelayError> { + if let Some(backend) = &self.serverless_backend { + return backend.publish(&event).await; + } self.cmd_tx .send(RelayCommand::PublishEvent { event: Box::new(event), @@ -1165,6 +1268,15 @@ impl HarnessRelay { /// Suitable for ephemeral commands like typing indicators where dropping /// the event on a full command channel is acceptable. pub fn try_publish_event(&self, event: Event) -> Result<(), RelayError> { + if let Some(backend) = &self.serverless_backend { + // Ephemeral (typing): fire-and-forget on the pool. Spawn so we don't + // block; dropping on error is acceptable for typing indicators. + let pool = backend.pool(); + tokio::spawn(async move { + let _ = pool.send_event(&event).await; + }); + return Ok(()); + } self.cmd_tx .try_send(RelayCommand::PublishEvent { event: Box::new(event), @@ -1218,6 +1330,10 @@ impl HarnessRelay { /// Reconnect after connection loss. Instructs the background task to /// re-authenticate and resubscribe to all previously active channels. pub async fn reconnect(&mut self) -> Result<(), RelayError> { + if self.serverless_backend.is_some() { + // The pool auto-reconnects + auto-resubscribes; nothing to do. + return Ok(()); + } warn!("relay connection lost — reconnecting…"); self.cmd_tx .send(RelayCommand::Reconnect) @@ -1232,6 +1348,10 @@ impl HarnessRelay { /// the background task to finish. Use this from async contexts instead of /// relying on `Drop` (which aborts immediately). pub async fn shutdown(mut self) { + if let Some(backend) = &self.serverless_backend { + backend.shutdown().await; + return; + } let _ = self.cmd_tx.send(RelayCommand::Shutdown).await; let handles = std::mem::take(&mut self.bg_handles); for handle in handles { @@ -3038,7 +3158,10 @@ fn jittered_duration(base: Duration) -> Duration { /// cryptographically verified `sender`, so we trust the rumor's author; the /// downstream pipeline reads only id/pubkey/kind/created_at/tags/content (never /// `.sig`). We materialize an `Event` via JSON with an empty signature. -fn rumor_to_event(rumor: &nostr::UnsignedEvent, sender: nostr::PublicKey) -> Option { +pub(crate) fn rumor_to_event( + rumor: &nostr::UnsignedEvent, + sender: nostr::PublicKey, +) -> Option { let id = rumor.id?; // Rumors carry no signature (NIP-17). The seal already verified `sender`, // and the downstream pipeline never inspects `.sig`, so we attach an @@ -3055,7 +3178,7 @@ fn rumor_to_event(rumor: &nostr::UnsignedEvent, sender: nostr::PublicKey) -> Opt )) } -fn extract_h_tag_uuid(event: &nostr::Event) -> Option { +pub(crate) fn extract_h_tag_uuid(event: &nostr::Event) -> Option { event.tags.iter().find_map(|tag| { let tag_vec = tag.as_slice(); if tag_vec.len() >= 2 && tag_vec[0] == "h" { diff --git a/crates/sprout-acp/src/serverless_relay.rs b/crates/sprout-acp/src/serverless_relay.rs new file mode 100644 index 0000000000..d31ba7ba7b --- /dev/null +++ b/crates/sprout-acp/src/serverless_relay.rs @@ -0,0 +1,279 @@ +//! Serverless relay backend built on `nostr-relay-pool`. +//! +//! In serverless mode the agent is an ordinary Nostr client talking to generic +//! public relays (no Sprout server, no NIP-42 AUTH, no HTTP bridge). Rather +//! than hand-roll multi-relay subscribe / reconnect / resubscribe / dedup (the +//! `relay.rs` background-task machinery used for server mode), we delegate to +//! [`RelayPool`], which provides exactly the standard Nostr client behaviour: +//! +//! - **Persistent REQ per relay**, fanned out across every relay in the pool. +//! - **Auto-reconnect** (on by default) with jittered backoff. +//! - **Auto-resubscribe** on reconnect — the pool re-sends every active REQ, and +//! relays replay their stored events matching the filter (NIP-01), so events +//! that arrived during a disconnect are recovered. +//! - **Merge + dedup** across relays: `RelayPoolNotification::Event` fires only +//! the first time an event id is seen, and excludes events we published. +//! +//! This mirrors how damus (`RelayPool`), 0xchat, and nostr-tools (`SimplePool`) +//! receive messages reliably across flaky relays. +//! +//! Subscription IDs match the server-mode scheme so the harness routes events +//! identically: `ch-` for channels, plus the fixed membership / observer +//! / gift-wrap inbox ids. + +use nostr::{Event, Filter, Keys, Kind, PublicKey, SingleLetterTag, SubscriptionId}; +use nostr_relay_pool::{RelayOptions, RelayPool, RelayPoolNotification, SubscribeOptions}; +use tokio::sync::mpsc; +use tracing::{debug, info, warn}; +use uuid::Uuid; + +use crate::config::ChannelFilter; +use crate::relay::{ + channel_sub_id, extract_h_tag_uuid, rumor_to_event, RelayError, SproutEvent, GIFT_WRAP_SUB_ID, + MEMBERSHIP_NOTIF_SUB_ID, OBSERVER_CONTROL_SUB_ID, +}; + +/// Serverless relay backend. Owns a [`RelayPool`] and a background task that +/// converts pool notifications into [`SproutEvent`]s. +/// +/// One-shot queries (channel discovery) go through `RestClient`, which shares +/// this pool via [`ServerlessRelay::pool`] — so there's no separate fetch path +/// here. +pub struct ServerlessRelay { + pool: RelayPool, + agent_pubkey: PublicKey, + /// Merged, classified channel/membership/gift-wrap events. + event_rx: mpsc::Receiver>, + /// Relay URLs the pool is configured with (for diagnostics). + relay_urls: Vec, + /// Background notification-pump handle. + _pump: tokio::task::JoinHandle<()>, +} + +impl ServerlessRelay { + /// Build the pool, add all relays, connect, and spawn the notification pump. + pub async fn connect( + relay_url: &str, + keys: &Keys, + agent_pubkey_hex: &str, + ) -> Result { + let relay_urls: Vec = relay_url + .split(',') + .map(|s| s.trim().to_string()) + .filter(|s| !s.is_empty()) + .collect(); + if relay_urls.is_empty() { + return Err(RelayError::Http("no relay URLs provided".into())); + } + + let agent_pubkey = PublicKey::from_hex(agent_pubkey_hex) + .map_err(|e| RelayError::Http(format!("invalid agent pubkey: {e}")))?; + + let pool = RelayPool::new(); + let mut added = 0usize; + for url in &relay_urls { + // RelayOptions::default() has reconnect=true + adjusted (jittered) + // retry interval — exactly the auto-reconnect we want. + match pool.add_relay(url.as_str(), RelayOptions::default()).await { + Ok(_) => added += 1, + Err(e) => warn!("serverless: failed to add relay {url}: {e}"), + } + } + if added == 0 { + return Err(RelayError::Http("no relays could be added".into())); + } + + // Connects all relays in the background; reconnection is automatic. + pool.connect().await; + info!( + "serverless: relay pool connecting to {}/{} relay(s): {}", + added, + relay_urls.len(), + relay_urls.join(", ") + ); + + let (event_tx, event_rx) = mpsc::channel::>(1024); + + let pump_keys = keys.clone(); + let notifications = pool.notifications(); + let pump = tokio::spawn(notification_pump(notifications, pump_keys, event_tx)); + + Ok(Self { + pool, + agent_pubkey, + event_rx, + relay_urls, + _pump: pump, + }) + } + + /// Relay URLs the pool is using. + pub fn relay_urls(&self) -> &[String] { + &self.relay_urls + } + + /// Cheap clone of the underlying pool (Arc inside) so detached tasks + /// (presence, typing) can publish ephemeral events without holding the + /// whole `ServerlessRelay`. + pub fn pool(&self) -> RelayPool { + self.pool.clone() + } + + /// Next merged event (channel / membership / decrypted gift-wrap). Returns + /// `None` only if the pump task ended (pool shut down). + pub async fn next_event(&mut self) -> Option { + // The pump sends `Some(event)`; it never sends `None` unless closing. + match self.event_rx.recv().await { + Some(Some(ev)) => Some(ev), + Some(None) | None => None, + } + } + + /// Publish a signed event to all relays (succeeds if any accepts). + pub async fn publish(&self, event: &Event) -> Result<(), RelayError> { + let output = self + .pool + .send_event(event) + .await + .map_err(|e| RelayError::Http(format!("publish failed: {e}")))?; + if output.success.is_empty() { + return Err(RelayError::Http(format!( + "no relay accepted event (failed on {} relay(s))", + output.failed.len() + ))); + } + Ok(()) + } + + /// Subscribe to a channel with the standard `ch-` id. The pool keeps + /// the REQ open and resubscribes on reconnect automatically. + pub async fn subscribe_channel( + &self, + channel_id: Uuid, + filter: ChannelFilter, + ) -> Result<(), RelayError> { + let id = SubscriptionId::new(channel_sub_id(channel_id)); + let mut f = Filter::new().custom_tag( + SingleLetterTag::lowercase(nostr::Alphabet::H), + channel_id.to_string(), + ); + if let Some(kinds) = filter.kinds { + f = f.kinds(kinds.into_iter().map(|k| Kind::from(k as u16))); + } + if filter.require_mention { + f = f.pubkey(self.agent_pubkey); + } + self.pool + .subscribe_with_id(id, f, SubscribeOptions::default()) + .await + .map_err(|e| RelayError::Http(format!("subscribe failed: {e}")))?; + debug!("serverless: subscribed to channel {channel_id}"); + Ok(()) + } + + /// Unsubscribe a channel. + pub async fn unsubscribe_channel(&self, channel_id: Uuid) { + let id = SubscriptionId::new(channel_sub_id(channel_id)); + self.pool.unsubscribe(&id).await; + } + + /// Subscribe to the agent's NIP-17 gift-wrap inbox (kind 1059, #p = agent). + pub async fn subscribe_gift_wrap(&self) -> Result<(), RelayError> { + let id = SubscriptionId::new(GIFT_WRAP_SUB_ID); + let f = Filter::new().kind(Kind::GiftWrap).pubkey(self.agent_pubkey); + self.pool + .subscribe_with_id(id, f, SubscribeOptions::default()) + .await + .map_err(|e| RelayError::Http(format!("gift-wrap subscribe failed: {e}")))?; + Ok(()) + } + + /// Disconnect / shut down the pool. + pub async fn shutdown(&self) { + self.pool.shutdown().await; + } +} + +/// Classify pool notifications into SproutEvents / observer frames, mirroring +/// the server-mode bg-task routing — but with no manual dedup/reconnect (the +/// pool handles both). +async fn notification_pump( + mut notifications: tokio::sync::broadcast::Receiver, + keys: Keys, + event_tx: mpsc::Sender>, +) { + loop { + let notif = match notifications.recv().await { + Ok(n) => n, + // Lagged: we missed some — keep going (relays will replay on any + // resubscribe; dedup is by id so no harm). + Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => { + warn!("serverless: notification stream lagged by {n}"); + continue; + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => { + let _ = event_tx.send(None).await; + return; + } + }; + + match notif { + RelayPoolNotification::Event { + subscription_id, + event, + .. + } => { + let sub = subscription_id.as_str(); + if sub == GIFT_WRAP_SUB_ID { + // NIP-17 encrypted message: unwrap to the inner rumor. + match nostr::nips::nip59::UnwrappedGift::from_gift_wrap(&keys, &event).await { + Ok(unwrapped) => { + if let Some(inner) = rumor_to_event(&unwrapped.rumor, unwrapped.sender) + { + if let Some(channel_uuid) = extract_h_tag_uuid(&inner) { + let _ = event_tx + .send(Some(SproutEvent { + channel_id: channel_uuid, + event: inner, + })) + .await; + } + } + } + Err(e) => debug!("serverless: gift wrap not for us: {e}"), + } + } else if sub == OBSERVER_CONTROL_SUB_ID { + // Observer controls are a Sprout-relay feature; never + // subscribed in serverless, so this branch is unreachable. + } else if sub == MEMBERSHIP_NOTIF_SUB_ID { + if let Some(channel_uuid) = extract_h_tag_uuid(&event) { + let _ = event_tx + .send(Some(SproutEvent { + channel_id: channel_uuid, + event: *event, + })) + .await; + } + } else if let Some(channel_uuid) = channel_uuid_from_sub(sub) { + let _ = event_tx + .send(Some(SproutEvent { + channel_id: channel_uuid, + event: *event, + })) + .await; + } + } + RelayPoolNotification::Message { .. } => {} + RelayPoolNotification::Shutdown => { + let _ = event_tx.send(None).await; + return; + } + } + } +} + +/// Parse `ch-` subscription ids. +fn channel_uuid_from_sub(sub: &str) -> Option { + sub.strip_prefix("ch-") + .and_then(|s| Uuid::parse_str(s).ok()) +}