fix(serverless): standard multi-relay Nostr — fix split-brain reads, query flood, single-relay live subscription

The serverless transport was behaving like it talked to the bespoke Sprout
server instead of a standard multi-relay Nostr client. Three real bugs:

1. Split-brain: messages were sent over the multi-relay pool (landing on
   whichever relay accepts, e.g. nos.lol when damus rate-limits) but READ over
   a single-relay live WS (damus only) — so your own message was invisible.
   Fix: in serverless, send + history + LIVE SUBSCRIPTION all go through the
   Rust multi-relay pool. New persistent pool subscribe()/unsubscribe() keeps a
   REQ open on ALL relays at once and merges+dedups (standard Nostr, like damus
   SimplePool). New commands: query_channel_messages, subscribe/unsubscribe_
   channel_messages; events streamed to the UI via serverless-event:<channel>.

2. Query flood: get_channels ran {kinds:[39000],limit:5000} (discover ALL
   network channels) + a 500-id member-count batch on a public relay. Fix:
   serverless skips network-wide discovery; only your member channels are
   listed (4 scoped queries).

3. Diagnostics: added [serverless]/[pool] backend logging and [relay] webview
   console logging so failures are visible, not silent.

Proven with live tests against damus+nos.lol: multi-relay publish survives
per-relay rate-limit, and the live subscription delivers a message published
to a different relay than the naive primary (the split-brain fix).

desktop/src-tauri/src/commands/channels.rs

@@ -142,17 +142,27 @@ pub async fn get_channels(state: State<'_, AppState>) -> Result<Vec<ChannelInfo>
         Vec::new()
     };
 
-    // Step 3: fetch ALL open channel metadata so the channel browser can show
+    // Step 3: fetch open channel metadata so the channel browser can show
     // discoverable channels the user hasn't joined yet. The relay's access
     // control allows reading kind:39000 for open channels regardless of membership.
-    let open_meta_events = query_relay(
-        &state,
-        &[serde_json::json!({
-            "kinds": [39000],
-            "limit": 5000,
-        })],
-    )
-    .await?;
+    //
+    // SERVERLESS: skip this. A generic public relay has no Sprout-specific
+    // notion of "our" channels — an unfiltered kind:39000 query returns the
+    // ENTIRE network's channels (hundreds of unrelated test channels from
+    // damus/nos.lol), which is slow and floods the sidebar with junk. In
+    // serverless mode you only see channels you're a member of (Step 1/2).
+    let open_meta_events = if state.is_serverless() {
+        Vec::new()
+    } else {
+        query_relay(
+            &state,
+            &[serde_json::json!({
+                "kinds": [39000],
+                "limit": 5000,
+            })],
+        )
+        .await?
+    };
 
     // Merge: member channels (marked as member) + open channels (not yet joined).
     let member_d_tags: std::collections::HashSet<String> = meta_events
@@ -649,6 +659,101 @@ pub async fn leave_channel(channel_id: String, state: State<'_, AppState>) -> Re
     Ok(())
 }
 
+/// Fetch channel message history over the multi-relay pool (serverless).
+///
+/// The live-WS read path (`relayClient`) connects to a single relay, which
+/// split-brains against the multi-relay *write* path (a message published to
+/// nos.lol is invisible to a read subscription on damus). This command queries
+/// the same relay set used for writes and merges/dedups results, so reads and
+/// writes converge. Returns events as JSON (the same shape the live WS yields).
+#[tauri::command]
+pub async fn query_channel_messages(
+    channel_id: String,
+    kinds: Vec<u16>,
+    limit: usize,
+    until: Option<u64>,
+    state: State<'_, AppState>,
+) -> Result<Vec<serde_json::Value>, String> {
+    let mut filter = serde_json::json!({
+        "kinds": kinds,
+        "#h": [channel_id],
+        "limit": limit,
+    });
+    if let Some(u) = until {
+        filter["until"] = serde_json::json!(u);
+    }
+    let events = query_relay(&state, &[filter]).await?;
+    Ok(events
+        .iter()
+        .map(|ev| serde_json::to_value(ev).unwrap_or(serde_json::Value::Null))
+        .filter(|v| !v.is_null())
+        .collect())
+}
+
+/// Open a persistent live subscription for a channel across ALL relays
+/// (serverless). Each new matching event is emitted to the frontend as a
+/// `serverless-event:<channel_id>` Tauri event. Returns the subscription id;
+/// pass it to `unsubscribe_channel_messages` to tear down.
+///
+/// This gives standard Nostr realtime in serverless mode: we subscribe to every
+/// relay at once and merge, so a message that landed on relay B (because relay A
+/// rate-limited the write) still streams back live — no polling, no split-brain.
+#[tauri::command]
+pub async fn subscribe_channel_messages(
+    channel_id: String,
+    kinds: Vec<u16>,
+    app: tauri::AppHandle,
+    state: State<'_, AppState>,
+) -> Result<String, String> {
+    use tauri::Emitter;
+
+    let relay_urls = crate::relay::relay_ws_urls_with_override(&state);
+    let keys = {
+        let guard = state.keys.lock().map_err(|e| e.to_string())?;
+        guard.clone()
+    };
+    let filter = serde_json::json!({
+        "kinds": kinds,
+        "#h": [channel_id],
+        "since": chrono::Utc::now().timestamp(),
+        "limit": 0,
+    });
+
+    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<nostr::Event>();
+    let sub_id = state
+        .relay_pool
+        .subscribe(&relay_urls, &keys, filter, tx)
+        .await;
+
+    // Forward events to the frontend, deduping by id (the same event arrives
+    // from multiple relays). The task ends when the sender is dropped on
+    // unsubscribe / pool clear.
+    let event_name = format!("serverless-event:{channel_id}");
+    tokio::spawn(async move {
+        let mut seen = std::collections::HashSet::new();
+        while let Some(ev) = rx.recv().await {
+            if !seen.insert(ev.id.to_hex()) {
+                continue;
+            }
+            if let Ok(v) = serde_json::to_value(&ev) {
+                let _ = app.emit(&event_name, v);
+            }
+        }
+    });
+
+    Ok(sub_id)
+}
+
+/// Tear down a live subscription opened by `subscribe_channel_messages`.
+#[tauri::command]
+pub async fn unsubscribe_channel_messages(
+    sub_id: String,
+    state: State<'_, AppState>,
+) -> Result<(), String> {
+    state.relay_pool.unsubscribe(&sub_id).await;
+    Ok(())
+}
+
 #[cfg(test)]
 #[path = "channels_tests.rs"]
 mod tests;

desktop/src-tauri/src/commands/channels_tests.rs

@@ -410,3 +410,88 @@ async fn serverless_burst_no_rate_limit() {
     }
     eprintln!("✅ 6 sequential publishes reused one pooled connection — no rate limit");
 }
+
+// ── Multi-relay live subscription (the split-brain fix) ──────────────────────
+//
+// Reproduces the exact bug: a message published to relay B must still reach a
+// subscriber, because the subscription is open on ALL relays at once (standard
+// Nostr — like damus/SimplePool), not just one. We:
+//   1. open a pool live subscription across [damus, nos.lol] for a channel,
+//   2. publish a kind-9 message to that channel (lands on whichever accepts),
+//   3. assert the subscription delivers it within a few seconds.
+//
+// Run with:
+//   cargo test --manifest-path desktop/src-tauri/Cargo.toml \
+//     serverless_live_subscription_multi_relay -- --ignored --nocapture
+#[tokio::test]
+#[ignore = "network: hits live public relays"]
+async fn serverless_live_subscription_multi_relay() {
+    use crate::app_state::build_app_state;
+    use crate::relay::submit_event;
+    use std::sync::atomic::Ordering;
+
+    let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
+
+    let keys = nostr::Keys::generate();
+    let relays: Vec<String> = std::env::var("RELAY_URL")
+        .unwrap_or_else(|_| "wss://relay.damus.io,wss://nos.lol".to_string())
+        .split(',')
+        .map(|s| s.trim().to_string())
+        .collect();
+
+    let state = build_app_state();
+    *state.keys.lock().unwrap() = keys.clone();
+    *state.relay_url_override.lock().unwrap() = Some(relays.join(","));
+    state.serverless.store(true, Ordering::Relaxed);
+
+    let channel = uuid::Uuid::new_v4();
+    let secret = format!("live-sub-{}", &channel.to_string()[..8]);
+
+    // 1. Open a live subscription across all relays.
+    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<nostr::Event>();
+    let sub_id = state
+        .relay_pool
+        .subscribe(
+            &relays,
+            &keys,
+            serde_json::json!({
+                "kinds": [9],
+                "#h": [channel.to_string()],
+                "since": chrono::Utc::now().timestamp() - 5,
+            }),
+            tx,
+        )
+        .await;
+    eprintln!("opened live sub {sub_id} across {relays:?}");
+
+    // Let the REQs register on all relays.
+    tokio::time::sleep(std::time::Duration::from_secs(2)).await;
+
+    // 2. Publish a message (multi-relay fanout — lands wherever accepts).
+    let builder =
+        crate::events::build_message(channel, &secret, None, &[], &[]).expect("build message");
+    let resp = submit_event(builder, &state).await.expect("publish");
+    eprintln!(
+        "published: accepted={} msg={:?}",
+        resp.accepted, resp.message
+    );
+
+    // 3. The live subscription must deliver it (from whichever relay stored it).
+    let got = tokio::time::timeout(std::time::Duration::from_secs(15), async {
+        while let Some(ev) = rx.recv().await {
+            if ev.content == secret {
+                return Some(ev);
+            }
+        }
+        None
+    })
+    .await
+    .expect("timed out waiting for live event — split-brain not fixed");
+
+    let got = got.expect("subscription channel closed before event arrived");
+    assert_eq!(got.content, secret);
+    assert_eq!(got.pubkey, keys.public_key());
+    eprintln!("✅ live subscription delivered the message across relays — no split-brain");
+
+    state.relay_pool.unsubscribe(&sub_id).await;
+}

desktop/src-tauri/src/lib.rs

@@ -531,6 +531,9 @@ pub fn run() {
             nip44_encrypt_to_self,
             nip44_decrypt_from_self,
             get_channels,
+            query_channel_messages,
+            subscribe_channel_messages,
+            unsubscribe_channel_messages,
             create_channel,
             open_dm,
             hide_dm,

desktop/src-tauri/src/relay.rs

@@ -171,7 +171,21 @@ pub async fn query_relay(
     // Serverless mode: no HTTP bridge. Query the generic relay over WS.
     if state.is_serverless() {
         let relay_urls = relay_ws_urls_with_override(state);
-        return crate::ws_relay::query_relay_ws(state, &relay_urls, filters).await;
+        eprintln!(
+            "sprout-desktop: [serverless] query → {relay_urls:?} filters={}",
+            serde_json::to_string(filters).unwrap_or_default()
+        );
+        let r = crate::ws_relay::query_relay_ws(state, &relay_urls, filters).await;
+        match &r {
+            Ok(events) => {
+                eprintln!(
+                    "sprout-desktop: [serverless] query OK {} event(s)",
+                    events.len()
+                )
+            }
+            Err(e) => eprintln!("sprout-desktop: [serverless] query ERR: {e}"),
+        }
+        return r;
     }
 
     let url = format!("{}/query", relay_api_base_url_with_override(state));
@@ -340,7 +354,16 @@ pub async fn submit_event(
     // Serverless mode: no HTTP bridge. Publish to the generic relay over WS.
     if state.is_serverless() {
         let relay_urls = relay_ws_urls_with_override(state);
-        return crate::ws_relay::submit_event_ws(builder, state, &relay_urls).await;
+        eprintln!("sprout-desktop: [serverless] submit_event → {relay_urls:?}");
+        let r = crate::ws_relay::submit_event_ws(builder, state, &relay_urls).await;
+        match &r {
+            Ok(resp) => eprintln!(
+                "sprout-desktop: [serverless] submit_event OK accepted={} msg={:?}",
+                resp.accepted, resp.message
+            ),
+            Err(e) => eprintln!("sprout-desktop: [serverless] submit_event ERR: {e}"),
+        }
+        return r;
     }
 
     // All synchronous work (signing) must complete before any .await

desktop/src-tauri/src/ws_pool.rs

@@ -95,6 +95,8 @@ pub struct RelayPool {
     /// retried. Lets a burst of queries skip a dead relay instead of each
     /// paying the full connect timeout.
     failed: std::sync::Mutex<HashMap<String, std::time::Instant>>,
+    /// Live subscriptions → the relays they're open on (for CLOSE on teardown).
+    live_subs: std::sync::Mutex<HashMap<String, Vec<String>>>,
 }
 
 impl RelayPool {
@@ -224,6 +226,66 @@ impl RelayPool {
         Ok(events)
     }
 
+    /// Open a PERSISTENT live subscription across all given relays. Every
+    /// matching EVENT from any relay is forwarded to `sink` (deduped by id).
+    /// The REQ stays open until [`RelayPool::unsubscribe`] is called with the
+    /// returned sub id. This is how serverless gets standard Nostr realtime:
+    /// subscribe to every relay at once and merge, so a message published to
+    /// relay B (because relay A rate-limited the write) still streams back.
+    pub async fn subscribe(
+        &self,
+        relay_urls: &[String],
+        keys: &Keys,
+        filter: serde_json::Value,
+        sink: mpsc::UnboundedSender<nostr::Event>,
+    ) -> String {
+        let sub_id = format!("live-{}", uuid::Uuid::new_v4());
+        let req_json = serde_json::Value::Array(vec![
+            serde_json::Value::String("REQ".into()),
+            serde_json::Value::String(sub_id.clone()),
+            filter,
+        ])
+        .to_string();
+
+        for url in relay_urls {
+            // Best-effort: a dead relay just doesn't contribute events.
+            let conn = match self.get(url, keys).await {
+                Ok(c) => c,
+                Err(_) => continue,
+            };
+            {
+                let mut d = conn.dispatch.lock().await;
+                d.queries.insert(
+                    sub_id.clone(),
+                    PendingQuery {
+                        events_tx: sink.clone(),
+                        done_tx: None, // never completes — live stream
+                    },
+                );
+            }
+            let _ = conn.send(req_json.clone()).await;
+        }
+        // Track which relays this sub is on so we can CLOSE them later.
+        self.live_subs
+            .lock()
+            .unwrap()
+            .insert(sub_id.clone(), relay_urls.to_vec());
+        sub_id
+    }
+
+    /// Close a live subscription on all relays it was opened on.
+    pub async fn unsubscribe(&self, sub_id: &str) {
+        let urls = self.live_subs.lock().unwrap().remove(sub_id);
+        let Some(urls) = urls else { return };
+        let close_json = serde_json::json!(["CLOSE", sub_id]).to_string();
+        for url in urls {
+            if let Some(conn) = self.live(&url) {
+                conn.dispatch.lock().await.queries.remove(sub_id);
+                let _ = conn.send(close_json.clone()).await;
+            }
+        }
+    }
+
     /// Publish a signed event over the pooled connection and await OK.
     pub async fn publish(
         &self,
@@ -249,11 +311,18 @@ impl RelayPool {
         let event_json = serde_json::json!(["EVENT", event]).to_string();
         if let Err(e) = conn.send(event_json).await {
             conn.dispatch.lock().await.publishes.remove(&event_id);
+            eprintln!("sprout-desktop: [pool] {relay_url} send EVENT failed: {e}");
             return Err(e);
         }
 
         match tokio::time::timeout(PUBLISH_TIMEOUT, ok_rx).await {
-            Ok(Ok(resp)) => Ok(resp),
+            Ok(Ok(resp)) => {
+                eprintln!(
+                    "sprout-desktop: [pool] {relay_url} OK accepted={} msg={:?}",
+                    resp.accepted, resp.message
+                );
+                Ok(resp)
+            }
             // Channel dropped (reader died) — clean up and report.
             Ok(Err(_)) => {
                 conn.dispatch.lock().await.publishes.remove(&event_id);
@@ -278,15 +347,18 @@ impl RelayPool {
     pub fn clear(&self) {
         self.conns.lock().unwrap().clear();
         self.failed.lock().unwrap().clear();
+        self.live_subs.lock().unwrap().clear();
     }
 }
 
 /// Establish a connection and spawn its background reader task.
 async fn connect(relay_url: &str, keys: &Keys) -> Result<Arc<Conn>, String> {
+    eprintln!("sprout-desktop: [pool] connecting to {relay_url}");
     let (ws, _) = tokio::time::timeout(CONNECT_TIMEOUT, connect_async(relay_url))
         .await
         .map_err(|_| format!("relay {relay_url} connection timed out"))?
         .map_err(|e| format!("relay connection failed: {e}"))?;
+    eprintln!("sprout-desktop: [pool] connected to {relay_url}");
     let (write, read) = ws.split();
     let dispatch = Arc::new(Mutex::new(Dispatch::default()));
     let alive = Arc::new(std::sync::atomic::AtomicBool::new(true));