Fix Android WebXDC realtime: bi-directional WS, subscribe race, cleanup

Multiple fixes for intermittent one-way data flow on Android:

1. Subscribe-before-connect race fix (realtime.rs):
   Move peer connect() AFTER subscribe_with_opts() so the gossip
   actor has the topic registered when connections deliver messages.

2. Bi-directional WebSocket receive (rt_ws.rs, realtime.rs):
   WS was send-only. Now incoming gossip data is pushed back through
   the WS connection, bypassing JNI evaluateJavascript starvation.

3. Bootstrap peers in preconnect (commands.rs):
   Pass cached + persisted peer addresses to join_channel as bootstrap
   instead of joining with 0 peers and racing with try_add_peer.

4. Android overlay cleanup (commands.rs):
   miniapp_close and stale instance cleanup now properly tear down
   realtime channels, session peers, and WS senders.

5. Android CSP (MiniAppWebViewClient.kt):
   Added ws://127.0.0.1:* to connect-src and wasm-unsafe-eval to
   script-src to match desktop CSP.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: JSKitty

src-tauri/gen/android/app/src/main/java/io/vectorapp/miniapp/MiniAppWebViewClient.kt

@@ -45,7 +45,7 @@ class MiniAppWebViewClient(
          *
          * This matches the desktop implementation for consistency.
          */
-        private const val CSP = """default-src 'self' http://webxdc.localhost; style-src 'self' http://webxdc.localhost 'unsafe-inline' blob:; font-src 'self' http://webxdc.localhost data: blob:; script-src 'self' http://webxdc.localhost 'unsafe-inline' 'unsafe-eval' blob:; connect-src 'self' http://webxdc.localhost ipc: data: blob:; img-src 'self' http://webxdc.localhost data: blob:; media-src 'self' http://webxdc.localhost data: blob:; webrtc 'block'"""
+        private const val CSP = """default-src 'self' http://webxdc.localhost; style-src 'self' http://webxdc.localhost 'unsafe-inline' blob:; font-src 'self' http://webxdc.localhost data: blob:; script-src 'self' http://webxdc.localhost 'unsafe-inline' 'unsafe-eval' 'wasm-unsafe-eval' blob:; connect-src 'self' http://webxdc.localhost ipc: data: blob: ws://127.0.0.1:*; img-src 'self' http://webxdc.localhost data: blob:; media-src 'self' http://webxdc.localhost data: blob:; webrtc 'block'"""
 
         /**
          * Permissions Policy that denies all sensitive APIs by default.

src-tauri/src/android/miniapp_jni.rs

@@ -459,7 +459,8 @@ pub extern "C" fn Java_io_vectorapp_miniapp_MiniAppIpc_joinRealtimeChannelNative
 
         // Join the channel with mpsc event target
         let event_target = crate::miniapps::realtime::EventTarget::MpscSender(tx);
-        let is_rejoin = match iroh.join_channel(topic, vec![], Some(event_target), Some(app_for_join.clone()), miniapp_id_for_join.clone()).await {
+        let ws_targets = Some(state.realtime.ws_senders.clone());
+        let is_rejoin = match iroh.join_channel(topic, vec![], Some(event_target), Some(app_for_join.clone()), miniapp_id_for_join.clone(), ws_targets).await {
             Ok((rejoin, _)) => {
                 if rejoin {
                     log_info!("[WEBXDC] Android: Re-joined existing channel for topic: {}", topic_encoded_for_join);
@@ -1279,6 +1280,17 @@ fn generate_android_webxdc_bridge(self_addr: &str, self_name: &str) -> String {
                         rtWs.binaryType = 'arraybuffer';
                         rtWs.onclose = function() {{ rtWs = null; }};
                         rtWs.onerror = function() {{ try {{ rtWs.close(); }} catch(e) {{}} rtWs = null; }};
+                        // Bi-directional: receive gossip data via WS (bypasses JNI starvation)
+                        rtWs.onmessage = function(ev) {{
+                            var fn_listener = window.__miniapp_realtime_listener;
+                            if (fn_listener && ev.data) {{
+                                // Data arrives as base91 string in binary frame
+                                var bytes = new Uint8Array(ev.data);
+                                var str = '';
+                                for (var i = 0; i < bytes.length; i++) str += String.fromCharCode(bytes[i]);
+                                fn_listener(new Uint8Array(b91d(str)));
+                            }}
+                        }};
                         // Detect stuck CONNECTING state (some WebViews silently block WS)
                         setTimeout(function() {{
                             if (rtWs && rtWs.readyState === 0) {{

src-tauri/src/miniapps/commands.rs

@@ -696,8 +696,9 @@ pub async fn miniapp_open(
     href: Option<String>,
     topic_id: Option<String>,
 ) -> Result<(), Error> {
+    log_info!("[WEBXDC] miniapp_open called: chat={}, msg={}", chat_id, message_id);
     let path = PathBuf::from(&file_path);
-    
+
     // Generate unique ID from file hash
     let id = format!("miniapp_{:x}", md5_hash(&file_path));
     // For marketplace apps (empty chat/message), use the app id as the window label
@@ -724,8 +725,29 @@ pub async fn miniapp_open(
                 }
                 return Ok(());
             } else {
-                // Overlay was closed, clean up state
-                log_warn!("Instance exists but overlay closed, cleaning up: {}", existing_label);
+                // Overlay was closed but state was never cleaned up.
+                // Do full teardown: realtime channel, session peers, instance.
+                log_warn!("Instance exists but overlay closed, full cleanup: {}", existing_label);
+
+                let channel_state = state.remove_realtime_channel(&existing_label).await;
+                if let Some(channel) = channel_state {
+                    let topic_encoded = super::realtime::encode_topic_id(&channel.topic);
+                    if let Ok(iroh) = state.realtime.get_or_init().await {
+                        if let Err(e) = iroh.leave_channel(channel.topic, &existing_label).await {
+                            log_warn!("[WEBXDC] Stale cleanup: leave_channel failed: {}", e);
+                        }
+                    }
+                    if let Some(my_pk) = crate::MY_PUBLIC_KEY.get() {
+                        if let Ok(my_npub) = my_pk.to_bech32() {
+                            state.remove_session_peer(&channel.topic, &my_npub).await;
+                        }
+                    }
+                    let chat_id_clone = chat_id.clone();
+                    tokio::spawn(async move {
+                        crate::commands::realtime::send_webxdc_peer_left(chat_id_clone, topic_encoded).await;
+                    });
+                }
+                state.realtime.ws_senders.write().unwrap_or_else(|e| e.into_inner()).remove(&existing_label);
                 state.remove_instance(&existing_label).await;
             }
         }
@@ -833,11 +855,32 @@ pub async fn miniapp_open(
                 Err(e) => { log_warn!("[WEBXDC] Preconnect: Iroh init failed: {e}"); return; }
             };
 
-            // Create gossip channel with NO event target. Incoming data is BUFFERED
-            // (not dropped) until joinRealtimeChannel sets the target and flushes.
-            // This lets us form the gossip mesh NOW so peers are connected before
-            // the game starts its election/handshake.
-            if let Err(e) = iroh.join_channel(topic, vec![], None, Some(app_pc.clone()), label_pc.clone()).await {
+            // Collect any cached peer addresses (from advertisements that arrived
+            // before we opened) + persisted peers from the DB to use as bootstrap.
+            let mut bootstrap_peers: Vec<iroh::EndpointAddr> = Vec::new();
+
+            // Cached from recent Nostr advertisements
+            let cached = state.take_peer_addrs(&topic).await;
+            bootstrap_peers.extend(cached);
+
+            // Persisted from DB
+            let my_npub = crate::MY_PUBLIC_KEY.get()
+                .and_then(|pk| nostr_sdk::prelude::ToBech32::to_bech32(pk).ok())
+                .unwrap_or_default();
+            if let Ok(records) = crate::db::get_active_peer_advertisements(&topic_encoded, &my_npub) {
+                for record in &records {
+                    if let Ok(addr) = super::realtime::decode_node_addr(&record.node_addr_encoded) {
+                        bootstrap_peers.push(addr);
+                    }
+                }
+            }
+
+            log_info!("[WEBXDC] Preconnect: joining with {} bootstrap peers", bootstrap_peers.len());
+
+            // Create gossip channel with bootstrap peers and NO event target.
+            // Incoming data is BUFFERED (not dropped) until joinRealtimeChannel
+            // sets the target and flushes.
+            if let Err(e) = iroh.join_channel(topic, bootstrap_peers, None, Some(app_pc.clone()), label_pc.clone(), None).await {
                 log_warn!("[WEBXDC] Preconnect: join_channel failed: {e}");
                 return;
             }
@@ -1163,6 +1206,36 @@ pub async fn miniapp_close(
             }
         }
 
+        // Full teardown: remove channel state, leave gossip, clean up
+        // (Desktop does this in WindowEvent::Destroyed, but Android has no
+        // Tauri window, so we must do it here explicitly)
+        let channel_state = state.remove_realtime_channel(&label).await;
+        if let Some(channel) = channel_state {
+            let topic_encoded = super::realtime::encode_topic_id(&channel.topic);
+
+            if let Ok(iroh) = state.realtime.get_or_init().await {
+                if let Err(e) = iroh.leave_channel(channel.topic, &label).await {
+                    log_warn!("[WEBXDC] Failed to leave channel on close: {}", e);
+                }
+            }
+
+            // Remove ourselves from session peers
+            if let Some(my_pk) = crate::MY_PUBLIC_KEY.get() {
+                if let Ok(my_npub) = my_pk.to_bech32() {
+                    state.remove_session_peer(&channel.topic, &my_npub).await;
+                }
+            }
+
+            // Send peer-left via Nostr
+            let chat_id_clone = chat_id.clone();
+            tokio::spawn(async move {
+                crate::commands::realtime::send_webxdc_peer_left(chat_id_clone, topic_encoded).await;
+            });
+        }
+
+        // Clean up WS sender
+        state.realtime.ws_senders.write().unwrap_or_else(|e| e.into_inner()).remove(&label);
+
         state.remove_instance(&label).await;
     }
 
@@ -1306,7 +1379,8 @@ pub async fn miniapp_join_realtime_channel(
 
     // Create the gossip channel WITH the event target — no data can be dropped
     let event_target = EventTarget::TauriChannel(channel);
-    let (is_rejoin, _) = iroh.join_channel(topic, vec![], Some(event_target), Some(app.clone()), label.to_string()).await
+    let ws_targets = Some(state.realtime.ws_senders.clone());
+    let (is_rejoin, _) = iroh.join_channel(topic, vec![], Some(event_target), Some(app.clone()), label.to_string(), ws_targets).await
         .map_err(|e| Error::RealtimeError(e.to_string()))?;
 
     let topic_encoded_clone = topic_encoded.clone();

src-tauri/src/miniapps/realtime.rs

@@ -254,6 +254,7 @@ impl IrohState {
         event_target: Option<EventTarget>,
         app_handle: Option<AppHandle>,
         label: String,
+        ws_event_targets: Option<Arc<std::sync::RwLock<HashMap<String, tokio::sync::mpsc::Sender<Vec<u8>>>>>>,
     ) -> Result<(bool, Option<oneshot::Receiver<()>>)> {
         let mut channels = self.channels.write().await;
 
@@ -265,6 +266,15 @@ impl IrohState {
                 let mut state = channel_state.event_target.write().unwrap_or_else(|e| { log_error!("[WEBXDC] RwLock poisoned — recovering"); e.into_inner() });
                 state.set_target(target); // Flushes buffered events from preconnect phase
             }
+            // Wire up WS sender for bi-directional receive (if WS is connected)
+            if let Some(ref senders) = ws_event_targets {
+                let map = senders.read().unwrap_or_else(|e| e.into_inner());
+                if let Some(ws_tx) = map.get(&label) {
+                    let mut state = channel_state.event_target.write().unwrap_or_else(|e| e.into_inner());
+                    state.set_ws_sender(ws_tx.clone());
+                    log_info!("[WEBXDC] RT WS bi-directional enabled for: {label}");
+                }
+            }
             return Ok((true, None));
         }
 
@@ -276,7 +286,20 @@ impl IrohState {
             peer_ids.len()
         );
 
-        // Connect to peers so gossip can discover them
+        // DON'T manually connect + handle_connection here — that creates
+        // connections BEFORE the topic subscription exists, causing a race
+        // where messages arrive for an unregistered topic and get lost.
+        // Instead, connect AFTER subscribing, so the gossip actor has the
+        // topic registered when the connection delivers messages.
+
+        let (join_tx, join_rx) = oneshot::channel();
+
+        let gossip_topic = self
+            .gossip
+            .subscribe_with_opts(topic, JoinOptions::with_bootstrap(peer_ids))
+            .await?;
+
+        // NOW connect — topic subscription is registered, safe to receive
         for peer_addr in &peers {
             if !peer_addr.addrs.is_empty() {
                 let addr = peer_addr.clone();
@@ -294,19 +317,22 @@ impl IrohState {
                 });
             }
         }
-
-        let (join_tx, join_rx) = oneshot::channel();
-
-        let gossip_topic = self
-            .gossip
-            .subscribe_with_opts(topic, JoinOptions::with_bootstrap(peer_ids))
-            .await?;
         let (gossip_sender, gossip_receiver) = gossip_topic.split();
 
         // Create shared event target for the subscribe loop (buffers events if target is None)
         let shared_event_target: SharedEventTarget = Arc::new(std::sync::RwLock::new(EventTargetState::new(event_target)));
         let shared_target_clone = shared_event_target.clone();
 
+        // Wire up WS sender for bi-directional receive (if WS is connected)
+        if let Some(ref senders) = ws_event_targets {
+            let map = senders.read().unwrap_or_else(|e| e.into_inner());
+            if let Some(ws_tx) = map.get(&label) {
+                let mut state = shared_event_target.write().unwrap_or_else(|e| e.into_inner());
+                state.set_ws_sender(ws_tx.clone());
+                log_info!("[WEBXDC] RT WS bi-directional enabled for: {label}");
+            }
+        }
+
         // Create shared peer count
         let shared_peer_count: SharedPeerCount = Arc::new(AtomicUsize::new(0));
         let peer_count_clone = shared_peer_count.clone();
@@ -391,11 +417,12 @@ impl IrohState {
 
         log_trace!("[WEBXDC] add_peer: Connecting to peer {}", peer_addr.id);
 
-        // Connect to the peer and hand the connection to gossip
+        // Connect and hand to gossip, then join_peers.
+        // Topic subscription already exists (channel is in the map),
+        // so the connection won't race with topic registration.
         let conn = self.endpoint.connect(peer_addr, GOSSIP_ALPN).await?;
         self.gossip.handle_connection(conn).await?;
 
-        // Join the peer to the existing gossip topic
         let channels = self.channels.read().await;
         if let Some(channel_state) = channels.get(topic) {
             channel_state.sender.join_peers(vec![peer.id]).await?;
@@ -459,6 +486,9 @@ impl IrohState {
         // 1. Remove fast-path SendHandle (drops its GossipSender clone)
         self.send_handles.write().unwrap_or_else(|e| { log_error!("[WEBXDC] RwLock poisoned — recovering"); e.into_inner() }).remove(label);
 
+        // Remove WS sender for this label (ws_senders is on RealtimeManager,
+        // but we're on IrohState — caller handles this separately)
+
         if let Some(channel) = self.channels.write().await.remove(&topic) {
             // 2. Drop the ChannelState's sender explicitly (don't wait for implicit drop)
             drop(channel.sender);
@@ -467,7 +497,7 @@ impl IrohState {
             channel.subscribe_loop.abort();
             let _ = channel.subscribe_loop.await;
 
-            // 4. Small yield to let the runtime fully clean up dropped tasks
+            // 4. Yield to let the gossip actor process the quit
             tokio::task::yield_now().await;
 
             log_info!("Left realtime channel {:?}", topic);
@@ -515,19 +545,42 @@ pub enum EventTarget {
 pub(crate) struct EventTargetState {
     target: Option<EventTarget>,
     buffer: Vec<RealtimeEvent>,
+    /// Optional WebSocket sender for bi-directional WS (bypasses JNI on Android).
+    /// When set, Data events are sent directly through WS instead of the normal target.
+    ws_sender: Option<tokio::sync::mpsc::Sender<Vec<u8>>>,
 }
 
 impl EventTargetState {
     fn new(target: Option<EventTarget>) -> Self {
-        Self { target, buffer: Vec::new() }
+        Self { target, buffer: Vec::new(), ws_sender: None }
     }
 
-    /// Send an event, buffering if no target is set yet
+    /// Register a WS sender for bi-directional receive. Data events bypass
+    /// the normal target (JNI on Android) and go straight through WebSocket.
+    pub fn set_ws_sender(&mut self, sender: tokio::sync::mpsc::Sender<Vec<u8>>) {
+        self.ws_sender = Some(sender);
+    }
+
+    pub fn clear_ws_sender(&mut self) {
+        self.ws_sender = None;
+    }
+
+    /// Send an event, buffering if no target is set yet.
+    /// Data events are routed through WebSocket when available (bypasses JNI on Android).
     fn send(&mut self, event: RealtimeEvent) -> bool {
+        // If WS sender is available and this is a Data event, send via WS directly.
+        // This bypasses the JNI/evaluateJavascript path that gets starved by WASM.
+        if let Some(ref ws_tx) = self.ws_sender {
+            if let RealtimeEvent::Data(ref b91_data) = event {
+                // Send raw base91 string as binary WS frame
+                let _ = ws_tx.try_send(b91_data.as_bytes().to_vec());
+                return true;
+            }
+        }
+
         if let Some(ref target) = self.target {
             Self::deliver(target, event)
         } else {
-            // Buffer up to 256 events (prevents unbounded memory if target is never set)
             if self.buffer.len() < 256 {
                 self.buffer.push(event);
             }
@@ -764,6 +817,9 @@ pub struct RealtimeManager {
     /// Fast-path send handles — owned here so the WS server can start
     /// before IrohState exists (critical for Android JNI timing).
     send_handles: Arc<std::sync::RwLock<HashMap<String, SendHandle>>>,
+    /// Map of window_label → WS sender for bi-directional receive.
+    /// WS handler registers sender on connect, join_channel wires it into the event target.
+    pub(crate) ws_senders: Arc<std::sync::RwLock<HashMap<String, tokio::sync::mpsc::Sender<Vec<u8>>>>>,
 }
 
 impl RealtimeManager {
@@ -773,6 +829,7 @@ impl RealtimeManager {
             relay_url,
             ws_info: std::sync::OnceLock::new(),
             send_handles: Arc::new(std::sync::RwLock::new(HashMap::new())),
+            ws_senders: Arc::new(std::sync::RwLock::new(HashMap::new())),
         }
     }
 
@@ -831,6 +888,7 @@ impl RealtimeManager {
 
         // Spawn accept loop on the MAIN Tauri runtime (survives JNI temp runtime)
         let send_handles = self.send_handles.clone();
+        let ws_senders = self.ws_senders.clone();
         tauri::async_runtime::spawn(async move {
             // Convert std listener to tokio listener on the main runtime
             let listener = match tokio::net::TcpListener::from_std(std_listener) {
@@ -840,7 +898,7 @@ impl RealtimeManager {
                     return;
                 }
             };
-            super::rt_ws::run_accept_loop(listener, token, send_handles).await;
+            super::rt_ws::run_accept_loop(listener, token, send_handles, ws_senders).await;
         });
     }