diff --git a/desktop/scripts/check-file-sizes.mjs b/desktop/scripts/check-file-sizes.mjs index a9c853258..a174ce430 100644 --- a/desktop/scripts/check-file-sizes.mjs +++ b/desktop/scripts/check-file-sizes.mjs @@ -42,11 +42,11 @@ const overrides = new Map([ ["src/features/channels/useUnreadChannels.ts", 715], // NIP-RS read marker tracking + participated/authored/followed thread ID sets + localStorage persistence + catch-up REQ with thread activity collection + thread reply activity feed items + mutedRootIds denylist with localStorage persistence + muteThread/unmuteThread callbacks ["src/features/notifications/hooks.ts", 535], // notification settings + feed notification lifecycle + profile batch resolution + truncated-pubkey guard + badge state ["src/features/home/ui/HomeView.tsx", 505], // inbox/feed orchestration + thread context + reply/delete flow + NIP-RS read-state projection wiring (useHomeInboxReadState) - ["src/features/messages/hooks.ts", 540], // message query/mutation hooks + optimistic updates + isEncryptedChannel routing (NIP-17 serverless DMs/private channels) + ["src/features/messages/hooks.ts", 640], // + serverless multi-relay routing (send/history/live-subscribe via Rust pool to avoid single-relay split-brain) + message query/mutation hooks + optimistic updates + isEncryptedChannel routing (NIP-17 serverless DMs/private channels) ["src/features/messages/ui/MessageComposer.tsx", 800], // media upload handlers (paste, drop, dialog) + channelId reset effect + edit mode (pre-fill, save, cancel, escape) + composer autofocus (#572) + Sprout code-block paste branch (round-trips copy-button output as a literal codeBlock so Markdown can't reshape it) + scroll-to-bottom on multi-line paste (#619) + Slack-style attachment-editable edits: seed pendingImeta from edit target, stash/restore user's draft pendingImeta across edit-mode entry/exit, re-append imeta markdown lines on edit-submit so renderer draws them ["src/features/settings/ui/SettingsView.tsx", 600], ["src/features/sidebar/ui/AppSidebar.tsx", 860], // channels + forums creation forms + Pulse nav - ["src/shared/api/relayClientSession.ts", 1140], // + NIP-17 encrypted channels (fetchEncryptedHistory/subscribeToChannel decrypt gift wraps via decryptGiftWrap) + serverless mode (setServerless, connect() skips AUTH-wait, late-challenge answering) // durable websocket session manager with reconnect/replay/recovery state + sendTypingIndicator + fetchChannelHistoryBefore + subscribeToChannelLive (huddle TTS) + subscribeToHuddleEvents (huddle indicator) + disconnect() for workspace switch teardown + fetchEvents/subscribeLive/publishEvent for NIP-RS read state + publishUserStatus/subscribeToUserStatusUpdates (NIP-38) + ConnectionState plumbing & stall-watchdog wiring for half-open WS detection (Warp orange-icon case) + terminal session latch (auth rejection no longer racing back to reconnecting) — emitter + watchdog + reconnect policy logic extracted to relayConnectionStateEmitter.ts / relayStallWatchdog.ts / relayReconnectPolicy.ts + ["src/shared/api/relayClientSession.ts", 1180], // + serverless diagnostics logging // + NIP-17 encrypted channels (fetchEncryptedHistory/subscribeToChannel decrypt gift wraps via decryptGiftWrap) + serverless mode (setServerless, connect() skips AUTH-wait, late-challenge answering) // durable websocket session manager with reconnect/replay/recovery state + sendTypingIndicator + fetchChannelHistoryBefore + subscribeToChannelLive (huddle TTS) + subscribeToHuddleEvents (huddle indicator) + disconnect() for workspace switch teardown + fetchEvents/subscribeLive/publishEvent for NIP-RS read state + publishUserStatus/subscribeToUserStatusUpdates (NIP-38) + ConnectionState plumbing & stall-watchdog wiring for half-open WS detection (Warp orange-icon case) + terminal session latch (auth rejection no longer racing back to reconnecting) — emitter + watchdog + reconnect policy logic extracted to relayConnectionStateEmitter.ts / relayStallWatchdog.ts / relayReconnectPolicy.ts ["src-tauri/src/migration.rs", 1010], // worktree shared-agent-data symlink sync (SHARED_AGENT_FILES + SHARED_AGENT_DIRS symlink-to-canonical + sibling pack migration) + mcp_command provider reconciliation + persona_pack_path reconciliation + tests ["src-tauri/src/commands/media.rs", 730], // ffmpeg video transcode + poster frame extraction + run_ffmpeg_with_timeout (find_ffmpeg via resolve_command, is_video_file, transcode_to_mp4, extract_poster_frame, transcode_and_extract_poster) + spawn_blocking wrappers + tests ["src-tauri/src/commands/agents.rs", 881], // remote agent lifecycle routing (local + provider branches) + scope enforcement + persona pack metadata wiring + mcp_toolsets field + NIP-OA auth_tag in deploy payload @@ -69,7 +69,7 @@ const overrides = new Map([ ["src/features/channels/ui/AddChannelBotDialog.tsx", 690], // provider mode: Run on selector, trust warning, probe effect, single-agent enforcement, provider warnings display + RespondTo field + reuse guardrail ["src/features/settings/ui/ChannelTemplatesSettingsCard.tsx", 850], // template CRUD card + TemplateFormDialog (persona/team chip selectors + provider assignments + canvas template) + TemplateTeamSelector + ProviderAssignments + ProviderRow ["src/shared/api/types.ts", 650], // ... + AcpProviderCatalogEntry + AcpProvider (narrowed subtype) + InstallRuntimeResult + RespondToMode + respondTo/respondToAllowlist on ManagedAgent/Create/Update inputs - ["src-tauri/src/commands/channels.rs", 670], // serverless membership now carries roles (creator=owner) in kind:39002 p-tags + serverless membership read-modify-write of kind:39002 (join/leave/add/remove publish member list directly) + create_channel branches serverless (publish 39000+39002 directly) vs server (kind 9007 command) + channel CRUD commands // create_channel branches serverless (publish 39000+39002 directly) vs server (kind 9007 command) + channel CRUD commands + ["src-tauri/src/commands/channels.rs", 800], // serverless membership now carries roles (creator=owner) in kind:39002 p-tags + serverless membership read-modify-write of kind:39002 (join/leave/add/remove publish member list directly) + create_channel branches serverless (publish 39000+39002 directly) vs server (kind 9007 command) + channel CRUD commands // create_channel branches serverless (publish 39000+39002 directly) vs server (kind 9007 command) + channel CRUD commands ["src-tauri/src/events.rs", 1110], // + serverless channel delete (NIP-09 kind 5 over 39000/39002 coords) + role-aware serverless members builder (owner role) + tests // + self-p-tag regression tests (allow_self_tagging on 39000/39002) // serverless addressable builders (build_channel_metadata_serverless 39000 + build_channel_members_serverless 39002) + their tests + event builders + build_huddle_guidelines (kind:48106) + post_event_raw transport helper + participant p-tag on join/leave + NIP-43 relay admin builders (add/remove/change-role) + check_relay_role + DM/presence/workflow command builders + NIP-IA identity-archive builders (9035/9036) + .allow_self_tagging() guards (nostr 0.44 strips self-`p` by default; self-archive/unarchive needs it preserved) + spec vector 1 layout test ["src-tauri/src/huddle/mod.rs", 1020], // huddle state machine + Tauri commands + sync protocol doc; state/relay/pipeline extracted + emit_huddle_state_changed wiring ["src-tauri/src/huddle/models.rs", 950], // model download manager for Parakeet TDT-CTC STT + Pocket TTS with streaming downloads + SHA-256 verification + Rust-native tar extraction + version manifest + atomic swap + hot-start signaling + MODEL_LICENSE.txt sidecar (fail-closed readiness) + idempotent legacy Moonshine dir cleanup + tts_readiness_requires_license_sidecar test + Mary (VCTK p333) reference voice attribution block @@ -78,10 +78,10 @@ const overrides = new Map([ ["src-tauri/src/huddle/preprocessing.rs", 670], // TTS text preprocessing pipeline + unified split_sentences + int_to_words 0-999999 + URL trailing punctuation preservation + 23 unit tests ["src-tauri/src/huddle/relay_api.rs", 520], // audio relay recv task + per-peer frame counting for remote human TTS interrupt + NIP-98 channel member query ["src-tauri/src/huddle/tts.rs", 1380], // TTS pipeline + session warmup + cancel/shutdown handling + apply_fade_out (fade-out only — leading fade removed 2026-05-18 after onset-attenuation regression measured in examples/pocket_onset_probe.rs) + FIRST_APPEND_LEAD_IN_SAMPLES + build_sentence_append_plan (pure helper enforcing the lead-in fires exactly once per utterance, not per sentence — see lead_in_pad_fires_exactly_once_per_utterance regression test) + normalize_for_playback (per-sentence peak normalization to -3 dBFS ceiling with MAX_GAIN cap) + 30 unit tests (18 interrupt + 5 fade-out + 1 first-append-lead-in + 3 build-sentence-append-plan + 6 normalize) - ["src-tauri/src/relay.rs", 510], // +4 lines for NIP-OA auth tag injection in profile sync (build_profile_event) + verification test + ["src-tauri/src/relay.rs", 540], // + serverless query/submit diagnostics logging // +4 lines for NIP-OA auth tag injection in profile sync (build_profile_event) + verification test ["src-tauri/src/commands/pairing.rs", 600], // NIP-AB pairing actor: 3 Tauri commands + background WS task + NIP-42 auth + NIP-43 probe + event parsing helpers - ["src-tauri/src/lib.rs", 770], // +4 lines for PairingHandle managed state + 3 pairing command registrations + parse_message_deep_link helper extracted with 6 unit tests covering empty-param filter regression + mod migration + sync_shared_agent_data/reconcile_provider_mcp_commands/reconcile_persona_pack_paths calls on launch + SIGINT/SIGTERM/SIGHUP signal handlers for agent process cleanup - ["src/shared/api/tauri.ts", 1250], // pairing command wrappers + applyWorkspace + NIP-44 encrypt/decrypt wrappers + observer_url field + relay member API functions (list/get/add/remove/change-role) + prevent sleep + AcpProviderCatalogEntry raw types + fromRawAcpProviderCatalogEntry converter + installAcpRuntime + decryptGiftWrap (NIP-17 serverless encrypted channels) + ["src-tauri/src/lib.rs", 790], // + serverless message commands (query/subscribe/unsubscribe_channel_messages) registration // +4 lines for PairingHandle managed state + 3 pairing command registrations + parse_message_deep_link helper extracted with 6 unit tests covering empty-param filter regression + mod migration + sync_shared_agent_data/reconcile_provider_mcp_commands/reconcile_persona_pack_paths calls on launch + SIGINT/SIGTERM/SIGHUP signal handlers for agent process cleanup + ["src/shared/api/tauri.ts", 1300], // + serverless multi-relay message wrappers (queryChannelMessages, subscribe/unsubscribeChannelMessages) // pairing command wrappers + applyWorkspace + NIP-44 encrypt/decrypt wrappers + observer_url field + relay member API functions (list/get/add/remove/change-role) + prevent sleep + AcpProviderCatalogEntry raw types + fromRawAcpProviderCatalogEntry converter + installAcpRuntime + decryptGiftWrap (NIP-17 serverless encrypted channels) ]); async function walkFiles(directory) { diff --git a/desktop/src-tauri/src/commands/channels.rs b/desktop/src-tauri/src/commands/channels.rs index 6def95308..be67114eb 100644 --- a/desktop/src-tauri/src/commands/channels.rs +++ b/desktop/src-tauri/src/commands/channels.rs @@ -142,17 +142,27 @@ pub async fn get_channels(state: State<'_, AppState>) -> Result 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 = 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, + limit: usize, + until: Option, + state: State<'_, AppState>, +) -> Result, 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:` 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, + app: tauri::AppHandle, + state: State<'_, AppState>, +) -> Result { + 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::(); + 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; diff --git a/desktop/src-tauri/src/commands/channels_tests.rs b/desktop/src-tauri/src/commands/channels_tests.rs index 00bea5e83..612937227 100644 --- a/desktop/src-tauri/src/commands/channels_tests.rs +++ b/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 = 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::(); + 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; +} diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs index f97732830..1467d1442 100644 --- a/desktop/src-tauri/src/lib.rs +++ b/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, diff --git a/desktop/src-tauri/src/relay.rs b/desktop/src-tauri/src/relay.rs index b7116f074..79876fff0 100644 --- a/desktop/src-tauri/src/relay.rs +++ b/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 diff --git a/desktop/src-tauri/src/ws_pool.rs b/desktop/src-tauri/src/ws_pool.rs index 85261ef2e..d7f75ba9a 100644 --- a/desktop/src-tauri/src/ws_pool.rs +++ b/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>, + /// Live subscriptions → the relays they're open on (for CLOSE on teardown). + live_subs: std::sync::Mutex>>, } 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, + ) -> 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, 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)); diff --git a/desktop/src/features/messages/hooks.ts b/desktop/src/features/messages/hooks.ts index cb86a01ab..79d712cec 100644 --- a/desktop/src/features/messages/hooks.ts +++ b/desktop/src/features/messages/hooks.ts @@ -13,12 +13,16 @@ import { resolveReplyRootId, } from "@/features/messages/lib/threading"; import { relayClient } from "@/shared/api/relayClient"; +import { listen } from "@tauri-apps/api/event"; import { addReaction, deleteMessage, editMessage, + queryChannelMessages, removeReaction, sendChannelMessage, + subscribeChannelMessages, + unsubscribeChannelMessages, } from "@/shared/api/tauri"; import type { Channel, Identity, RelayEvent } from "@/shared/api/types"; @@ -40,10 +44,31 @@ function isEncryptedChannel(channel: Channel | null): boolean { // from the on-render overlay. import { applyEditTagOverlay } from "@/features/messages/lib/applyEditTagOverlay.mjs"; import { + CHANNEL_EVENT_KINDS, KIND_STREAM_MESSAGE, KIND_SYSTEM_MESSAGE, } from "@/shared/constants/kinds"; +/** + * Fetch channel history. In serverless mode, non-encrypted channels go through + * the Rust multi-relay pool (`queryChannelMessages`) so reads hit the same + * relay set as writes — the live-WS path is single-relay and split-brains. + * Encrypted channels and server mode keep the existing live-WS path. + */ +async function fetchHistoryForChannel( + channel: Channel, + limit: number, +): Promise { + if (isActiveWorkspaceServerless() && !isEncryptedChannel(channel)) { + return queryChannelMessages(channel.id, [...CHANNEL_EVENT_KINDS], limit); + } + return relayClient.fetchChannelHistory( + channel.id, + limit, + isEncryptedChannel(channel), + ); +} + type MessageQueryContext = { optimisticId: string; previousMessages: RelayEvent[]; @@ -146,10 +171,9 @@ export function useChannelMessagesQuery(channel: Channel | null) { throw new Error("No channel selected."); } - const history = await relayClient.fetchChannelHistory( - channel.id, + const history = await fetchHistoryForChannel( + channel, CHANNEL_HISTORY_LIMIT, - isEncryptedChannel(channel), ); const currentMessages = queryClient.getQueryData(queryKey) ?? []; @@ -171,18 +195,17 @@ export function useChannelSubscription(channel: Channel | null) { const channelType = channel?.channelType ?? null; const encrypted = isEncryptedChannel(channel); const syncLatestHistory = useEffectEvent(async () => { - if (!channelId) { + if (!channel) { return; } - const history = await relayClient.fetchChannelHistory( - channelId, + const history = await fetchHistoryForChannel( + channel, CHANNEL_HISTORY_LIMIT, - encrypted, ); queryClient.setQueryData( - channelMessagesKey(channelId), + channelMessagesKey(channel.id), (current = []) => { const mergedHistory = normalizeTimelineMessages([ ...current, @@ -233,6 +256,47 @@ export function useChannelSubscription(channel: Channel | null) { let isDisposed = false; let cleanup: (() => Promise) | undefined; + + // Serverless (non-encrypted): subscribe across ALL relays via the Rust + // pool, listening for `serverless-event:` Tauri events. This is + // standard Nostr realtime without the single-relay split-brain of the live + // WS. Encrypted channels keep the live-WS gift-wrap path. + if (isActiveWorkspaceServerless() && !encrypted) { + let unlisten: (() => void) | undefined; + let subId: string | undefined; + void (async () => { + unlisten = await listen( + `serverless-event:${channelId}`, + (event) => { + if (!isDisposed) { + appendMessage(event.payload); + } + }, + ); + if (isDisposed) { + unlisten(); + return; + } + subId = await subscribeChannelMessages(channelId, [ + ...CHANNEL_EVENT_KINDS, + ]); + // Initial backfill so existing messages show immediately. + void syncLatestHistory().catch(() => {}); + })().catch((error) => { + console.error( + "Failed serverless channel subscription", + channelId, + error, + ); + }); + + return () => { + isDisposed = true; + if (unlisten) unlisten(); + if (subId) void unsubscribeChannelMessages(subId); + }; + } + const disposeReconnectListener = relayClient.subscribeToReconnects(() => { void syncLatestHistory().catch((error) => { if (!isDisposed) { @@ -373,6 +437,36 @@ export function useSendMessageMutation( }; } + // Serverless: route through the Rust command, which publishes over the + // multi-relay connection pool (publish-to-all, succeed-if-any-accepts). + // The live-WS path (relayClient.sendMessage) uses a single relay with no + // pool/fallback, so it trips public-relay rate limits with no recovery. + if (isActiveWorkspaceServerless()) { + const result = await sendChannelMessage( + channel.id, + content, + null, + undefined, + mentionPubkeys, + ); + return { + id: result.eventId, + pubkey: identity.pubkey, + created_at: result.createdAt, + kind: KIND_STREAM_MESSAGE, + tags: [ + ["h", channel.id], + ["p", identity.pubkey], + ...normalizeMentionPubkeys( + mentionPubkeys ?? [], + identity.pubkey, + ).map((pk) => ["p", pk]), + ], + content: content.trim(), + sig: "", + }; + } + return relayClient.sendMessage( channel.id, content, diff --git a/desktop/src/shared/api/relayClientSession.ts b/desktop/src/shared/api/relayClientSession.ts index bcc82f11e..2c73a7c56 100644 --- a/desktop/src/shared/api/relayClientSession.ts +++ b/desktop/src/shared/api/relayClientSession.ts @@ -531,16 +531,31 @@ export class RelayClient { this.relayUrl = await getRelayWsUrl(); } + console.info( + `[relay] connecting live WS to ${this.relayUrl} (serverless=${this.serverless})`, + ); + const generation = ++this.connectionGeneration; this.onMessageChannel = new Channel((message) => { void this.handleWsMessage(message, generation); }); - this.wsId = await invoke("plugin:websocket|connect", { - url: this.relayUrl, - onMessage: this.onMessageChannel, - config: {}, - }); + try { + this.wsId = await invoke("plugin:websocket|connect", { + url: this.relayUrl, + onMessage: this.onMessageChannel, + config: {}, + }); + console.info( + `[relay] live WS connected to ${this.relayUrl} (id=${this.wsId})`, + ); + } catch (error) { + console.error( + `[relay] live WS connect FAILED to ${this.relayUrl}:`, + error, + ); + throw error; + } if (this.serverless) { // Generic public relays don't require (or send) a NIP-42 AUTH challenge. @@ -666,9 +681,15 @@ export class RelayClient { private async sendRaw(payload: unknown[]) { if (this.wsId === null) { + console.error( + `[relay] sendRaw FAILED — socket not connected; payload=${payload[0]}`, + ); throw new Error("Relay socket is not connected."); } + console.debug( + `[relay] → ${payload[0]} ${JSON.stringify(payload).slice(0, 160)}`, + ); await invoke("plugin:websocket|send", { id: this.wsId, message: { @@ -728,9 +749,15 @@ export class RelayClient { timeoutMessage: string, sendErrorMessage: string, ) { + console.info( + `[relay] publishEvent kind=${event.kind} id=${event.id.slice(0, 8)} → ${this.relayUrl}`, + ); return new Promise((resolve, reject) => { const timeout = window.setTimeout(() => { this.pendingEvents.delete(event.id); + console.warn( + `[relay] publishEvent TIMEOUT id=${event.id.slice(0, 8)} (no OK in 8s) — ${timeoutMessage}`, + ); reject(new Error(timeoutMessage)); }, 8_000); @@ -868,8 +895,14 @@ export class RelayClient { private handleEvent(subId: string, event: RelayEvent) { const subscription = this.subscriptions.get(subId); if (!subscription) { + console.debug( + `[relay] ← EVENT for UNKNOWN sub ${subId} (kind ${event.kind})`, + ); return; } + console.debug( + `[relay] ← EVENT sub=${subId.slice(0, 12)} kind=${event.kind} (${subscription.mode})`, + ); if (subscription.mode === "history") { subscription.events.push(event); @@ -921,6 +954,9 @@ export class RelayClient { } private handleOk(eventId: string, success: boolean, message: string) { + console.info( + `[relay] OK id=${eventId.slice(0, 8)} accepted=${success} msg=${JSON.stringify(message)}`, + ); if (this.authRequest && this.authRequest.pendingEventId === eventId) { window.clearTimeout(this.authRequest.timeout); const authRequest = this.authRequest; diff --git a/desktop/src/shared/api/tauri.ts b/desktop/src/shared/api/tauri.ts index 1b741cb1a..06ef94310 100644 --- a/desktop/src/shared/api/tauri.ts +++ b/desktop/src/shared/api/tauri.ts @@ -719,6 +719,45 @@ export async function getEventById(eventId: string): Promise { return JSON.parse(eventJson) as RelayEvent; } +/** + * Fetch channel message history over the multi-relay pool (serverless mode). + * Reads from the same relay set used for writes so reads/writes converge + * (the live-WS path is single-relay and split-brains in serverless mode). + */ +export async function queryChannelMessages( + channelId: string, + kinds: number[], + limit: number, + until?: number, +): Promise { + return invokeTauri("query_channel_messages", { + channelId, + kinds, + limit, + until: until ?? null, + }); +} + +/** + * Open a persistent live subscription for a channel across ALL relays + * (serverless). New events are emitted as `serverless-event:` Tauri + * events. Returns the subscription id for teardown. Standard Nostr realtime — + * subscribe to every relay at once and merge. + */ +export async function subscribeChannelMessages( + channelId: string, + kinds: number[], +): Promise { + return invokeTauri("subscribe_channel_messages", { + channelId, + kinds, + }); +} + +export async function unsubscribeChannelMessages(subId: string): Promise { + await invokeTauri("unsubscribe_channel_messages", { subId }); +} + export async function sendChannelMessage( channelId: string, content: string,