refactor(cdp): migrate all scanners to in-process transport; drop TCP DevTools port

app/src-tauri/src/cdp/conn.rs

@@ -1,101 +1,35 @@
 //! [`CdpConn`] — per-attach handle on top of the in-process CDP transport.
 //!
 //! Wraps an [`Arc<WebviewCdpTransport>`](super::in_process::WebviewCdpTransport)
-//! with the same `call` / `pump_events` surface the scanners and the
-//! per-account session opener were already using. The previous
-//! WebSocket-backed implementation (one socket per attach) is gone for
-//! new code paths; all attaches for a given webview now share the same
+//! with the same `call` / `pump_events` surface scanners and the per-account
+//! session opener use. All attaches for a given webview share the same
 //! in-process channel, and a [`CdpConn`] is just a cheap session-scoped
 //! view.
-//!
-//! For backward compatibility with per-scanner duplicated implementations
-//! that still attach via the TCP loopback DevTools port (whatsapp,
-//! slack, telegram, wechat, meet_video — see issue follow-up), the
-//! [`CdpConn::open_ws`] legacy constructor keeps the old `tungstenite`
-//! WebSocket path alive. Callers should migrate to
-//! [`super::conn_for_account`] over time.
 
-use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::Duration;
 
-use futures_util::{SinkExt, StreamExt};
-use serde_json::{json, Value};
+use serde_json::Value;
 use tokio::sync::broadcast::error::RecvError;
-use tokio::sync::oneshot;
-use tokio_tungstenite::{connect_async, tungstenite::Message};
 
 use super::in_process::{EventFrame, WebviewCdpTransport};
 
-/// Timeout applied to a single request/response round-trip in the
-/// legacy WebSocket transport. Long enough to cover a cold-attach on a
-/// sluggish machine; the in-process transport uses
-/// [`crate::cdp::CALL_TIMEOUT`] (also 35s) for symmetry.
-const LEGACY_CALL_TIMEOUT: Duration = Duration::from_secs(35);
-
-/// Internal: legacy WebSocket dispatch state.
-struct LegacyWs {
-    sink: futures_util::stream::SplitSink<
-        tokio_tungstenite::WebSocketStream<
-            tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
-        >,
-        Message,
-    >,
-    stream: futures_util::stream::SplitStream<
-        tokio_tungstenite::WebSocketStream<
-            tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
-        >,
-    >,
-    next_id: i64,
-    pending: HashMap<i64, oneshot::Sender<Result<Value, String>>>,
-}
-
-enum Backend {
-    InProcess(Arc<WebviewCdpTransport>),
-    LegacyWs(LegacyWs),
-}
-
-/// Per-attach CDP handle. Internally either wraps an
-/// `Arc<WebviewCdpTransport>` (in-process channel) or a tungstenite
-/// `WebSocketStream` (legacy TCP loopback). The session_id filter is
-/// per-handle so concurrent attachers don't see each other's events.
+/// Per-attach CDP handle. Wraps an `Arc<WebviewCdpTransport>` and
+/// filters incoming events by `session_id` so concurrent attachers
+/// don't see each other's events.
 pub struct CdpConn {
-    backend: Backend,
+    transport: Arc<WebviewCdpTransport>,
     label: String,
 }
 
 impl CdpConn {
     /// Wrap an already-installed in-process transport. Callers obtain
     /// the transport from the per-app [`super::CdpRegistry`]
-    /// (`app.state()`) — typically via
-    /// [`super::conn_for_account`].
+    /// (`app.state()`) — typically via [`super::conn_for_account`] or
+    /// [`super::conn_for_label`].
     pub fn new(transport: Arc<WebviewCdpTransport>) -> Self {
         let label = transport.label().to_string();
-        Self {
-            backend: Backend::InProcess(transport),
-            label,
-        }
-    }
-
-    /// Legacy: open a CDP connection over the loopback TCP WebSocket
-    /// exposed by `--remote-debugging-port`. Kept for the per-scanner
-    /// duplicated implementations (whatsapp, slack, telegram, wechat,
-    /// meet_video) that have not yet migrated to the in-process
-    /// channel. New code paths should use [`super::conn_for_account`].
-    pub async fn open_ws(ws_url: &str) -> Result<Self, String> {
-        let (ws, _resp) = connect_async(ws_url)
-            .await
-            .map_err(|e| format!("ws connect: {e}"))?;
-        let (sink, stream) = ws.split();
-        Ok(Self {
-            backend: Backend::LegacyWs(LegacyWs {
-                sink,
-                stream,
-                next_id: 1,
-                pending: HashMap::new(),
-            }),
-            label: format!("ws:{ws_url}"),
-        })
+        Self { transport, label }
     }
 
     /// Setup-phase request/response: sends a JSON-RPC call and awaits
@@ -108,10 +42,22 @@ impl CdpConn {
         params: Value,
         session_id: Option<&str>,
     ) -> Result<Value, String> {
-        match &mut self.backend {
-            Backend::InProcess(transport) => transport.call(method, params, session_id).await,
-            Backend::LegacyWs(ws) => legacy_ws_call(ws, method, params, session_id).await,
-        }
+        self.transport.call(method, params, session_id).await
+    }
+
+    /// Same as [`call`](Self::call) but with a caller-supplied response
+    /// timeout. Slack's IDB batch serialisation can run past the default
+    /// 35s; other callers should stick with `call`.
+    pub async fn call_with_timeout(
+        &mut self,
+        method: &str,
+        params: Value,
+        session_id: Option<&str>,
+        timeout: Duration,
+    ) -> Result<Value, String> {
+        self.transport
+            .call_with_timeout(method, params, session_id, timeout)
+            .await
     }
 
     /// Subscribe to the transport's event stream and dispatch every
@@ -123,137 +69,47 @@ impl CdpConn {
     /// so we drop events belonging to other sessions.
     ///
     /// Returns when the channel closes (the transport has been
-    /// forgotten / ws shut down) or on an unrecoverable error.
-    /// `Lagged` is treated as a continuation signal — the caller's idle
-    /// watchdog will eventually time out the session and the outer
-    /// reconnect loop re-attaches.
+    /// forgotten) or on an unrecoverable error. `Lagged` is logged and
+    /// treated as a continuation signal: the pump keeps draining rather
+    /// than tearing down the session, so a burst that overflows
+    /// `EVENT_CHANNEL_CAP` drops the skipped frames without re-syncing.
+    /// Long-lived consumers (e.g. the Discord scanner) do not yet have an
+    /// idle watchdog to time out a stalled/destroyed page target and force
+    /// a re-attach — that self-healing path is tracked as a fast-follow in
+    /// #3693.
     pub async fn pump_events<F>(&mut self, session_id: &str, mut on_event: F) -> Result<(), String>
     where
         F: FnMut(&str, &Value),
     {
-        match &mut self.backend {
-            Backend::InProcess(transport) => {
-                let mut rx = transport.subscribe_events();
-                loop {
-                    match rx.recv().await {
-                        Ok(EventFrame {
-                            method,
-                            params,
-                            session_id: evt_session,
-                        }) => {
-                            if !evt_session.is_empty() && evt_session != session_id {
-                                continue;
-                            }
-                            on_event(&method, &params);
-                        }
-                        Err(RecvError::Lagged(skipped)) => {
-                            log::warn!(
-                                "[cdp][{}] event channel lagged skipped={} session_id={}",
-                                self.label,
-                                skipped,
-                                session_id
-                            );
-                            continue;
-                        }
-                        Err(RecvError::Closed) => return Ok(()),
+        let mut rx = self.transport.subscribe_events();
+        loop {
+            match rx.recv().await {
+                Ok(EventFrame {
+                    method,
+                    params,
+                    session_id: evt_session,
+                }) => {
+                    if !evt_session.is_empty() && evt_session != session_id {
+                        continue;
                     }
+                    on_event(&method, &params);
+                }
+                Err(RecvError::Lagged(skipped)) => {
+                    log::warn!(
+                        "[cdp][{}] event channel lagged skipped={} session_id={}",
+                        self.label,
+                        skipped,
+                        session_id
+                    );
+                    continue;
                 }
+                Err(RecvError::Closed) => return Ok(()),
             }
-            Backend::LegacyWs(ws) => legacy_ws_pump_events(ws, session_id, on_event).await,
         }
     }
 
-    /// Diagnostic helper — webview label (in-process) or
-    /// `"ws:<url>"` (legacy WS) this connection is bound to.
+    /// Diagnostic helper — webview label this connection is bound to.
     pub fn label(&self) -> &str {
         &self.label
     }
 }
-
-async fn legacy_ws_call(
-    ws: &mut LegacyWs,
-    method: &str,
-    params: Value,
-    session_id: Option<&str>,
-) -> Result<Value, String> {
-    let id = ws.next_id;
-    ws.next_id += 1;
-    let mut req = json!({ "id": id, "method": method, "params": params });
-    if let Some(s) = session_id {
-        req["sessionId"] = json!(s);
-    }
-    let body = serde_json::to_string(&req).map_err(|e| format!("encode: {e}"))?;
-    ws.sink
-        .send(Message::Text(body))
-        .await
-        .map_err(|e| format!("ws send: {e}"))?;
-    loop {
-        let msg = tokio::time::timeout(LEGACY_CALL_TIMEOUT, ws.stream.next())
-            .await
-            .map_err(|_| format!("ws read timeout (method={method})"))?
-            .ok_or_else(|| format!("ws closed (method={method})"))?
-            .map_err(|e| format!("ws recv: {e}"))?;
-        let text = match msg {
-            Message::Text(t) => t,
-            Message::Binary(_) | Message::Ping(_) | Message::Pong(_) | Message::Frame(_) => {
-                continue
-            }
-            Message::Close(_) => return Err("ws closed".into()),
-        };
-        let v: Value = serde_json::from_str(&text).map_err(|e| format!("decode: {e}"))?;
-        if v.get("id").and_then(|x| x.as_i64()) != Some(id) {
-            continue;
-        }
-        if let Some(err) = v.get("error") {
-            return Err(format!("cdp error: {err}"));
-        }
-        return Ok(v.get("result").cloned().unwrap_or(Value::Null));
-    }
-}
-
-async fn legacy_ws_pump_events<F>(
-    ws: &mut LegacyWs,
-    session_id: &str,
-    mut on_event: F,
-) -> Result<(), String>
-where
-    F: FnMut(&str, &Value),
-{
-    loop {
-        let msg = ws
-            .stream
-            .next()
-            .await
-            .ok_or_else(|| "ws closed".to_string())?
-            .map_err(|e| format!("ws recv: {e}"))?;
-        let text = match msg {
-            Message::Text(t) => t,
-            Message::Binary(_) | Message::Ping(_) | Message::Pong(_) | Message::Frame(_) => {
-                continue
-            }
-            Message::Close(_) => return Ok(()),
-        };
-        let v: Value = match serde_json::from_str(&text) {
-            Ok(v) => v,
-            Err(_) => continue,
-        };
-        if let Some(id) = v.get("id").and_then(|x| x.as_i64()) {
-            if let Some(tx) = ws.pending.remove(&id) {
-                let res = if let Some(err) = v.get("error") {
-                    Err(format!("cdp error: {err}"))
-                } else {
-                    Ok(v.get("result").cloned().unwrap_or(Value::Null))
-                };
-                let _ = tx.send(res);
-            }
-            continue;
-        }
-        let method = v.get("method").and_then(|x| x.as_str()).unwrap_or("");
-        let evt_session = v.get("sessionId").and_then(|x| x.as_str()).unwrap_or("");
-        if !evt_session.is_empty() && evt_session != session_id {
-            continue;
-        }
-        let params = v.get("params").cloned().unwrap_or(Value::Null);
-        on_event(method, &params);
-    }
-}

app/src-tauri/src/cdp/in_process.rs

@@ -90,6 +90,22 @@ impl WebviewCdpTransport {
         method: &str,
         params: Value,
         session_id: Option<&str>,
+    ) -> Result<Value, String> {
+        self.call_with_timeout(method, params, session_id, CALL_TIMEOUT)
+            .await
+    }
+
+    /// Same as [`call`](Self::call) but with a caller-supplied response
+    /// timeout. Slack's IDB serialise batches can legitimately exceed the
+    /// default 35s budget; the canonical 60s timeout in that module flows
+    /// through here without forcing every other caller to opt into a
+    /// longer wait.
+    pub async fn call_with_timeout(
+        self: &Arc<Self>,
+        method: &str,
+        params: Value,
+        session_id: Option<&str>,
+        timeout: Duration,
     ) -> Result<Value, String> {
         let id = {
             let mut n = self.next_id.lock().expect("next_id mutex poisoned");
@@ -134,7 +150,7 @@ impl WebviewCdpTransport {
             return Err(e);
         }
 
-        match tokio::time::timeout(CALL_TIMEOUT, rx).await {
+        match tokio::time::timeout(timeout, rx).await {
             Ok(Ok(res)) => res,
             Ok(Err(_)) => {
                 // oneshot dropped — transport torn down between dispatch and
@@ -266,19 +282,29 @@ pub fn set_cef_app_handle(app: AppHandle<tauri::Cef>) {
 /// cold-boot race against a scanner that started before
 /// `webview_accounts::open` finished). Callers back off and retry.
 pub fn install_for_account(account_id: &str) -> Result<Arc<WebviewCdpTransport>, String> {
+    install_for_label(&format!("acct_{account_id}"))
+}
+
+/// Install (or look up) the in-process CDP transport for any
+/// CEF-backed webview, keyed by its concrete label. Generic counterpart
+/// of [`install_for_account`] for webviews that aren't account scanners
+/// — e.g. Meet call windows labelled `meet-call-{request_id}`.
+///
+/// Returns `Err` when the webview hasn't been created yet (cold-boot
+/// race). Callers back off and retry.
+pub fn install_for_label(label: &str) -> Result<Arc<WebviewCdpTransport>, String> {
     let app = CEF_APP_HANDLE
         .get()
         .ok_or_else(|| "cdp::set_cef_app_handle has not been called yet".to_string())?;
-    let label = format!("acct_{account_id}");
     let registry_state = app
         .try_state::<CdpRegistry>()
         .ok_or_else(|| "CdpRegistry not managed by app".to_string())?;
     let registry = registry_state.inner();
-    if let Some(existing) = registry.by_label(&label) {
+    if let Some(existing) = registry.by_label(label) {
         return Ok(existing);
     }
     let webview = app
-        .get_webview(&label)
+        .get_webview(label)
         .ok_or_else(|| format!("no webview for label={label}"))?;
     install_for_webview(registry, webview)
 }