hatchery_node/

browser_capture.rs

//! Linux dig2browser capture inside the node (feature `d2b`).
//!
//! Owns a headless Chromium via dig2browser, captures viewport frames (JPEG via
//! CDP `Page.captureScreenshot`, PNG fallback), and applies dig2-wire input.
//! Transport is the restream G4aControl sidecar — this module never dials HQ.

#![cfg(feature = "d2b")]

use dig2browser::browser::{
    BrowserPreference, BrowserProfile, LaunchConfig, StealthBrowser, StealthConfig, StealthPage,
};
use dig2browser::detect::args::RendererSandboxMode;
use dig2browser::stealth::LocaleProfile;
use hatchery_core::{
    decode_browser_input, BrowserFrameRegion, BrowserFrameTrace, BrowserInput, BrowserMouseButton, BrowserStreamEvent,
};
use serde_json::json;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::{mpsc, oneshot, Mutex};

const DEFAULT_URL: &str = "about:blank";
const CAPTURE_EVERY: Duration = Duration::from_millis(250);
const META_EVERY: Duration = Duration::from_millis(1000);
const VIEW_W: u32 = 1280;
const VIEW_H: u32 = 720;

pub struct BrowserSessionHandle {
    pub session: String,
    input_tx: mpsc::UnboundedSender<BrowserInput>,
    /// Wake the capture loop to push a full frame (after uplink reconnect).
    kick_tx: mpsc::UnboundedSender<()>,
    stop_tx: Option<oneshot::Sender<()>>,
    join: Option<tokio::task::JoinHandle<()>>,
}

impl BrowserSessionHandle {
    pub fn push_input(&self, input: BrowserInput) {
        let _ = self.input_tx.send(input);
    }

    pub fn kick_frame(&self) {
        let _ = self.kick_tx.send(());
    }

    pub async fn stop(mut self) {
        if let Some(tx) = self.stop_tx.take() {
            let _ = tx.send(());
        }
        if let Some(join) = self.join.take() {
            let _ = join.await;
        }
    }
}

/// Process-wide sink the active restream bridge installs. Browser sessions always
/// send through this so frames survive bridge reconnect (noise decrypt / eof).
#[derive(Clone, Default)]
pub struct BrowserEventSink {
    inner: Arc<Mutex<Option<mpsc::UnboundedSender<BrowserStreamEvent>>>>,
}

impl BrowserEventSink {
    pub async fn set(&self, tx: mpsc::UnboundedSender<BrowserStreamEvent>) {
        *self.inner.lock().await = Some(tx);
    }

    pub async fn clear(&self) {
        *self.inner.lock().await = None;
    }

    pub async fn send(&self, ev: BrowserStreamEvent) {
        let mut guard = self.inner.lock().await;
        let live = guard.is_some();
        match &ev {
            BrowserStreamEvent::Frame { session, bytes, .. } => {
                tracing::debug!(%session, bytes = bytes.len(), sink_live = live, "d2b frame → sink");
            }
            BrowserStreamEvent::Started { session, url, .. } => {
                tracing::debug!(%session, %url, sink_live = live, "d2b started → sink");
            }
            _ => {}
        }
        if let Some(tx) = guard.as_ref() {
            if tx.send(ev).is_err() {
                tracing::warn!("d2b sink send failed (bridge rx dropped)");
            }
        } else {
            tracing::warn!("d2b sink unbound; event dropped");
        }
    }
}

/// Registry of live browser sessions on this node.
pub struct BrowserSessionMap {
    inner: Mutex<std::collections::BTreeMap<String, BrowserSessionHandle>>,
    sink: BrowserEventSink,
}

impl Default for BrowserSessionMap {
    fn default() -> Self {
        Self {
            inner: Mutex::new(std::collections::BTreeMap::new()),
            sink: BrowserEventSink::default(),
        }
    }
}

impl BrowserSessionMap {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn sink(&self) -> BrowserEventSink {
        self.sink.clone()
    }

    /// Bridge just came up: point sessions at its event channel and force a full frame.
    pub async fn bind_bridge(&self, tx: mpsc::UnboundedSender<BrowserStreamEvent>) {
        self.sink.set(tx).await;
        let guard = self.inner.lock().await;
        for h in guard.values() {
            h.kick_frame();
        }
    }

    pub async fn start(
        &self,
        node_id: impl Into<String>,
        session: String,
        url: String,
        profile_dir: PathBuf,
        event_tx: mpsc::UnboundedSender<BrowserStreamEvent>,
    ) -> Result<(), String> {
        hatchery_core::validate_browser_session_id(&session)?;
        // Always refresh the live sink (HQ may Start on a fresh bridge).
        self.sink.set(event_tx).await;
        let node_id = node_id.into();
        let mut guard = self.inner.lock().await;
        if guard.contains_key(&session) {
            // Re-open from HQ (panel reopened / reload): reuse the running browser.
            self.sink
                .send(BrowserStreamEvent::Started {
                    node_id,
                    session: session.clone(),
                    url: url.clone(),
                })
                .await;
            if let Some(h) = guard.get(&session) {
                h.kick_frame();
                if !url.trim().is_empty() {
                    h.push_input(BrowserInput::Navigate { url });
                }
            }
            return Ok(());
        }
        let handle = spawn_session(
            node_id,
            session.clone(),
            if url.trim().is_empty() {
                DEFAULT_URL.to_owned()
            } else {
                url
            },
            profile_dir,
            self.sink.clone(),
        )?;
        guard.insert(session, handle);
        Ok(())
    }

    pub async fn input(&self, session: &str, bytes: &[u8]) -> Result<(), String> {
        let input = decode_browser_input(bytes)?;
        let guard = self.inner.lock().await;
        let Some(handle) = guard.get(session) else {
            return Err("browser session not found".into());
        };
        handle.push_input(input);
        Ok(())
    }

    pub async fn stop(&self, session: &str) -> Result<(), String> {
        let mut guard = self.inner.lock().await;
        let Some(handle) = guard.remove(session) else {
            return Err("browser session not found".into());
        };
        drop(guard);
        handle.stop().await;
        Ok(())
    }

    pub async fn stop_all(&self) {
        let mut guard = self.inner.lock().await;
        let handles: Vec<_> = std::mem::take(&mut *guard).into_values().collect();
        drop(guard);
        for h in handles {
            h.stop().await;
        }
    }
}

fn spawn_session(
    node_id: String,
    session: String,
    start_url: String,
    profile_dir: PathBuf,
    sink: BrowserEventSink,
) -> Result<BrowserSessionHandle, String> {
    let (input_tx, input_rx) = mpsc::unbounded_channel();
    let (kick_tx, kick_rx) = mpsc::unbounded_channel();
    let (stop_tx, stop_rx) = oneshot::channel();
    let session_for_task = session.clone();
    let sink_err = sink.clone();
    let join = tokio::spawn(async move {
        if let Err(err) = run_session(
            node_id.clone(),
            session_for_task.clone(),
            start_url,
            profile_dir,
            input_rx,
            kick_rx,
            stop_rx,
            sink.clone(),
        )
        .await
        {
            sink_err
                .send(BrowserStreamEvent::Error {
                    node_id: node_id.clone(),
                    session: session_for_task.clone(),
                    message: err,
                })
                .await;
        }
        sink_err
            .send(BrowserStreamEvent::Stopped {
                node_id,
                session: session_for_task,
                reason: "closed".into(),
            })
            .await;
    });
    Ok(BrowserSessionHandle {
        session,
        input_tx,
        kick_tx,
        stop_tx: Some(stop_tx),
        join: Some(join),
    })
}

async fn run_session(
    node_id: String,
    session: String,
    start_url: String,
    profile_dir: PathBuf,
    mut input_rx: mpsc::UnboundedReceiver<BrowserInput>,
    mut kick_rx: mpsc::UnboundedReceiver<()>,
    mut stop_rx: oneshot::Receiver<()>,
    sink: BrowserEventSink,
) -> Result<(), String> {
    std::fs::create_dir_all(&profile_dir).map_err(|e| e.to_string())?;
    // Shared node profile: one Chrome for every session, each session in its own
    // window (a background tab never paints). Otherwise one Chrome per session.
    let shared = shared::profile().is_some_and(|p| p == profile_dir);
    let (owned, mut page) = if shared {
        (None, shared::open_page(&profile_dir).await?)
    } else {
        ensure_profile_accept_languages(&profile_dir);
        let browser = StealthBrowser::launch_with(launch_config(&profile_dir), stealth_config())
            .await
            .map_err(|e| e.to_string())?;
        let page = browser.new_page("about:blank").await.map_err(|e| e.to_string())?;
        (Some(browser), page)
    };
    let _ = page.set_viewport(VIEW_W, VIEW_H, 1.0).await;
    apply_browser_locale(&page).await;
    // Headful: keep our page in front (a background tab never paints).
    let _ = page.cdp_call("Page.bringToFront", None).await;
    let _ = page
        .cdp_call("Emulation.setFocusEmulationEnabled", Some(json!({ "enabled": true })))
        .await;
    let _ = page.cdp_call("Page.navigate", Some(json!({ "url": start_url }))).await;
    sink
        .send(BrowserStreamEvent::Started {
            node_id: node_id.clone(),
            session: session.clone(),
            url: start_url.clone(),
        })
        .await;
    // First paint immediately — don't wait for screencast (HQ shows white otherwise).
    let mut force_kick = true;

    let opener_id = page
        .cdp_call("Target.getTargetInfo", None)
        .await
        .ok()
        .and_then(|v| v["targetInfo"]["targetId"].as_str().map(str::to_owned))
        .unwrap_or_default();

    // True screencast over a side CDP connection to our page target (push frames,
    // ack each); falls back to polling captureScreenshot if it can't be set up.
    let (sc_tx, mut sc_rx) = mpsc::channel::<Vec<u8>>(4);
    let mut screencast = match screencast::start(&profile_dir, &page, sc_tx).await {
        Ok(h) => Some(h),
        Err(e) => {
            tracing::warn!(%session, error = %e, "d2b screencast unavailable; polling");
            None
        }
    };

    // OAuth / window.open: Chrome opens a real second window we must stream, or
    // the parent spinner waits forever (Continue with Google, etc.).
    let (mut popup_rx, popup_watch) = match popups::start(&profile_dir, opener_id.clone()).await {
        Ok(v) => v,
        Err(e) => {
            tracing::warn!(%session, error = %e, "d2b popup watch unavailable");
            let (_t, r) = mpsc::unbounded_channel();
            (r, tokio::spawn(async {}))
        }
    };
    // Parent page parked here while we stream a popup; popup_id is the active child.
    let mut parent_page: Option<StealthPage> = None;
    let mut popup_id: Option<String> = None;
    // Dedupe Opened events (create + infoChanged both fire).
    let mut seen_popup: std::collections::HashSet<String> = std::collections::HashSet::new();

    let mut tick = tokio::time::interval(if screencast.is_some() { META_EVERY } else { CAPTURE_EVERY });
    tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
    let mut last_hash: u64 = 0;
    let (mut url, mut title) = (start_url.clone(), String::new());
    let mut input_n: u64 = 0;
    // (input_n, received, cdp_ms, dispatched)
    let mut pending: Option<(u64, std::time::Instant, u32, std::time::Instant, u32)> = None;
    let mut last_frame: Option<Vec<u8>> = None;
    let mut delta = delta::Delta::default();
    // Set on every stream switch: a static page may never emit a screencast frame,
    // so the next tick pushes one captured frame (+ fresh url/title) to HQ.
    let mut meta_fail: u32 = 0;
    macro_rules! restore_parent {
        () => {
            if let Some(p) = parent_page.take() {
                page = p;
                popup_id = None;
                let _ = page.cdp_call("Page.bringToFront", None).await;
                if let Some(h) = screencast.take() { h.abort(); }
                let (tx, rx) = mpsc::channel::<Vec<u8>>(4);
                sc_rx = rx;
                screencast = screencast::start(&profile_dir, &page, tx).await.ok();
                delta = delta::Delta::default();
                last_frame = None;
                last_hash = 0;
                url.clear();
                title.clear();
                force_kick = true;
                tick.reset_immediately();
            }
        };
    }
    loop {
        let frame: Option<(Vec<u8>, &'static str)> = tokio::select! {
            _ = &mut stop_rx => break,
            _ = kick_rx.recv() => {
                force_kick = true;
                tick.reset_immediately();
                None
            }
            pe = popup_rx.recv() => {
                match pe {
                    Some(popups::Ev::Opened { target_id, url: purl }) => {
                        if popup_id.as_deref() == Some(target_id.as_str()) || !seen_popup.insert(target_id.clone()) {
                            None
                        } else if parent_page.is_some() {
                            // Already showing a popup; ignore nested ones for now.
                            tracing::info!(%session, %target_id, "d2b popup ignored (already in one)");
                            None
                        } else {
                            tracing::info!(%session, %target_id, url = %purl, "d2b popup opened; switching stream");
                            match attach_target(shared, &owned, &profile_dir, &target_id).await {
                                Ok(child) => {
                                    let _ = child.set_viewport(VIEW_W, VIEW_H, 1.0).await;
                                    apply_browser_locale(&child).await;
                                    let _ = child.cdp_call("Page.bringToFront", None).await;
                                    let _ = child
                                        .cdp_call("Emulation.setFocusEmulationEnabled", Some(json!({ "enabled": true })))
                                        .await;
                                    let old = std::mem::replace(&mut page, child);
                                    parent_page = Some(old);
                                    popup_id = Some(target_id);
                                    if let Some(h) = screencast.take() { h.abort(); }
                                    let (tx, rx) = mpsc::channel::<Vec<u8>>(4);
                                    sc_rx = rx;
                                    screencast = screencast::start(&profile_dir, &page, tx).await.ok();
                                    delta = delta::Delta::default();
                                    last_frame = None;
                                    last_hash = 0;
                                    force_kick = true;
                                    tick.reset_immediately();
                                    None
                                }
                                Err(e) => {
                                    tracing::warn!(%session, error = %e, "d2b popup attach failed");
                                    None
                                }
                            }
                        }
                    }
                    Some(popups::Ev::Closed { target_id }) => {
                        seen_popup.remove(&target_id);
                        if popup_id.as_deref() == Some(target_id.as_str()) {
                            tracing::info!(%session, %target_id, "d2b popup closed; restoring parent");
                            restore_parent!();
                        }
                        None
                    }
                    None => None,
                }
            }
            f = sc_rx.recv(), if screencast.is_some() => f.map(|b| (b, "screencast")),
            _ = tick.tick() => {
                if screencast.is_some() {
                    // Screencast pushes pixels; the tick only refreshes url/title.
                    match tokio::time::timeout(Duration::from_millis(1500), page_meta(&page)).await {
                        Ok(Ok((u, ti))) => {
                            meta_fail = 0;
                            let changed = u != url || ti != title;
                            url = u; title = ti;
                            if force_kick || last_frame.is_none() {
                                // No screencast frame since the switch: capture one so HQ
                                // leaves the old (popup) picture and title.
                                match tokio::time::timeout(Duration::from_secs(3), capture_frame(&page)).await {
                                    Ok(Ok(bytes)) => { force_kick = false; Some((bytes, "poll")) }
                                    Ok(Err(e)) => { tracing::warn!(%session, error = %e, "d2b force_kick capture failed"); None }
                                    Err(_) => { tracing::warn!(%session, "d2b force_kick capture timeout"); None }
                                }
                            } else if changed {
                                last_frame.clone().map(|b| (b, "meta"))
                            } else { None }
                        }
                        _ if popup_id.is_some() && { meta_fail += 1; meta_fail >= 3 } => {
                            meta_fail = 0;
                            // Popup target died without a targetDestroyed (redirect/close race).
                            tracing::info!(%session, "d2b popup unresponsive; restoring parent");
                            if let Some(id) = popup_id.clone() { seen_popup.remove(&id); }
                            restore_parent!();
                            None
                        }
                        _ => None,
                    }
                } else {
                    if let Ok(Ok((u, ti))) = tokio::time::timeout(Duration::from_secs(3), page_meta(&page)).await { url = u; title = ti; }
                    match tokio::time::timeout(Duration::from_secs(5), capture_frame(&page)).await {
                        Ok(Ok(bytes)) => {
                            let hash = { use std::hash::{Hash, Hasher}; let mut h = std::collections::hash_map::DefaultHasher::new(); bytes.hash(&mut h); h.finish() };
                            if hash != last_hash { last_hash = hash; Some((bytes, "poll")) } else { None }
                        }
                        _ => None,
                    }
                }
            }
            input = input_rx.recv() => {
                let Some(input) = input else { break };
                let received = std::time::Instant::now();
                input_n += 1;
                let queue_ms = 0u32;
                if matches!(&input, BrowserInput::Navigate { url } if url.is_empty()) {
                    // Re-open: resend the last frame so a new viewer paints at once.
                    last_frame.clone().map(|b| (b, "replay"))
                } else {
                    // Address-bar navigate while an OAuth popup is up: close the popup
                    // and drive the parent (operator expects the bar to leave the login).
                    let input = if matches!(&input, BrowserInput::Navigate { url } if !url.is_empty())
                        && popup_id.is_some()
                    {
                        if let Some(id) = popup_id.clone() {
                            tracing::info!(%session, %id, "d2b navigate closes popup first");
                            let _ = page.cdp_call("Target.closeTarget", Some(json!({ "targetId": id }))).await;
                            seen_popup.remove(&id);
                            restore_parent!();
                        }
                        input
                    } else {
                        input
                    };
                    let copy = matches!(&input, BrowserInput::Key { key } if keys::is_copy(key));
                    let sel = if copy { keys::selection(&page).await } else { None };
                    if let Some(text) = sel.filter(|t| !t.is_empty()) {
                        sink.send(BrowserStreamEvent::Clipboard { node_id: node_id.clone(), session: session.clone(), text }).await;
                    }
                    match apply_input(&page, &input).await {
                        Err(err) => { tracing::warn!(%session, error = %err, "browser input failed"); None }
                        Ok(()) => {
                            let cdp_ms = received.elapsed().as_millis() as u32;
                            pending = Some((input_n, received, cdp_ms, std::time::Instant::now(), queue_ms));
                            if screencast.is_none() {
                                last_hash = 0;
                                tick.reset_immediately();
                            }
                            None
                        }
                    }
                }
            }
        };
        let Some((bytes, source)) = frame else { continue };
        // Screencast frames become deltas (changed rectangle only) when small enough:
        // a few KB fit one TCP window on every long hop instead of several RTTs.
        let (bytes, region) = if source == "screencast" {
            match delta.encode(&bytes) {
                delta::Out::Unchanged => continue,
                delta::Out::Full => (bytes, None),
                delta::Out::Region(b, r) => (b, Some(r)),
            }
        } else if source == "replay" || source == "meta" {
            (delta.full_jpeg().unwrap_or(bytes), None)
        } else {
            (bytes, None)
        };
        if region.is_none() && (source == "screencast" || source == "poll") {
            last_frame = Some(bytes.clone());
        }
        let trace = if source == "screencast" || source == "poll" {
            pending.take().map(|(n, received, cdp_ms, dispatched, queue_ms)| BrowserFrameTrace {
                input_n: n,
                queue_ms,
                cdp_ms,
                to_frame_ms: dispatched.elapsed().as_millis() as u32,
                node_ms: received.elapsed().as_millis() as u32,
                source: if region.is_some() { "screencast-delta".to_owned() } else { source.to_owned() },
            })
        } else { None };
        sink
            .send(BrowserStreamEvent::Frame {
                node_id: node_id.clone(),
                session: session.clone(),
                bytes,
                width: VIEW_W,
                height: VIEW_H,
                url: url.clone(),
                title: title.clone(),
                trace,
                region,
            })
            .await;
    }
    popup_watch.abort();
    if let Some(h) = screencast { h.abort(); }
    // Close an open popup first, then the parent page we stream.
    if let Some(id) = popup_id.take() {
        let _ = page.cdp_call("Target.closeTarget", Some(json!({ "targetId": id }))).await;
        if let Some(parent) = parent_page.take() {
            page = parent;
        }
    }
    match owned {
        Some(browser) => { let _ = browser.close().await; }
        None => { let _ = page.cdp_call("Page.close", None).await; }
    }
    Ok(())
}

async fn attach_target(
    shared_profile: bool,
    owned: &Option<StealthBrowser>,
    profile_dir: &std::path::Path,
    target_id: &str,
) -> Result<StealthPage, String> {
    if shared_profile {
        shared::attach(profile_dir, target_id).await
    } else {
        let b = owned.as_ref().ok_or("no browser for attach")?;
        b.attach_page(target_id).await.map_err(|e| e.to_string())
    }
}


fn launch_config(profile_dir: &std::path::Path) -> LaunchConfig {
    // Do not pass a second --accept-lang here: dig2browser already sets one from the
    // primary locale, and two flags make Chrome mangle q-values. Full en+ru list is
    // applied via profile intl.accept_languages + Network.setExtraHTTPHeaders.
    let extra_args = vec![
        "--force-device-scale-factor=1".into(),
        "--disable-dev-shm-usage".into(),
        // Several session windows paint at once in one shared Chrome.
        "--disable-backgrounding-occluded-windows".into(),
        "--disable-renderer-backgrounding".into(),
        "--disable-background-timer-throttling".into(),
    ];
    LaunchConfig {
        // Headful on Xvfb when a DISPLAY is present (less bot-detection); else headless.
        headless: std::env::var("HATCHERY_D2B_HEADFUL").map(|v| v != "1").unwrap_or(true)
            || std::env::var_os("DISPLAY").is_none(),
        window_size: (VIEW_W, VIEW_H),
        profile: BrowserProfile::Persistent(profile_dir.to_path_buf()),
        browser_pref: BrowserPreference::ChromeOnly,
        renderer_sandbox: RendererSandboxMode::CompatibilityDisabled,
        extra_args,
        ..LaunchConfig::default()
    }
}

/// Full Accept-Language header value, e.g. `en-US,en;q=0.9,ru;q=0.8`.
/// Unset in OSS; host sets via `HATCHERY_BROWSER_ACCEPT_LANGUAGES`.
fn browser_accept_languages() -> Option<String> {
    std::env::var("HATCHERY_BROWSER_ACCEPT_LANGUAGES")
        .ok()
        .map(|v| v.trim().to_owned())
        .filter(|v| !v.is_empty())
}

/// Language tags without q-values for `navigator.languages`.
fn browser_language_tags() -> Vec<String> {
    let Some(raw) = browser_accept_languages() else { return Vec::new() };
    raw.split(',')
        .map(|p| p.split(';').next().unwrap_or(p).trim().to_owned())
        .filter(|t| !t.is_empty())
        .collect()
}

/// Write `intl.accept_languages` into the Chrome profile Preferences (no cookie touch).
/// Must run while Chrome is not holding the profile.
fn ensure_profile_accept_languages(profile_dir: &std::path::Path) {
    let Some(al) = browser_accept_languages() else { return };
    let pref = profile_dir.join("Default").join("Preferences");
    let Ok(raw) = std::fs::read_to_string(&pref) else {
        // Fresh profile: Chrome will create Preferences; seed a minimal file.
        let _ = std::fs::create_dir_all(profile_dir.join("Default"));
        let seed = serde_json::json!({ "intl": { "accept_languages": al } });
        let _ = std::fs::write(&pref, seed.to_string());
        return;
    };
    let Ok(mut v) = serde_json::from_str::<serde_json::Value>(&raw) else { return };
    let obj = v.as_object_mut().unwrap();
    let intl = obj
        .entry("intl".to_owned())
        .or_insert_with(|| json!({}))
        .as_object_mut();
    let Some(intl) = intl else { return };
    intl.insert("accept_languages".into(), json!(al));
    if let Ok(out) = serde_json::to_string(&v) {
        let _ = std::fs::write(&pref, out);
    }
}

/// Locale / timezone for dig2browser. Driven only by env so OSS stays agnostic:
/// `HATCHERY_BROWSER_LANG` (BCP-47, e.g. en-US) and `HATCHERY_BROWSER_TZ`
/// (IANA, e.g. America/New_York). Unset → dig2browser defaults (en-US, no TZ override).
/// When TZ is set we also export it into this process so Chrome inherits it.
fn stealth_config() -> StealthConfig {
    let mut cfg = StealthConfig::default();
    let lang = std::env::var("HATCHERY_BROWSER_LANG").ok().filter(|v| !v.is_empty());
    let tz = std::env::var("HATCHERY_BROWSER_TZ").ok().filter(|v| !v.is_empty());
    if lang.is_none() && tz.is_none() {
        return cfg;
    }
    if let Some(ref t) = tz {
        // Chrome Date/Intl without CDP still see the process TZ.
        std::env::set_var("TZ", t);
    }
    cfg.locale = LocaleProfile {
        locale: lang.unwrap_or_else(|| cfg.locale.locale.clone()),
        timezone: tz.or(cfg.locale.timezone),
    };
    cfg
}

/// CDP locale/timezone on a target (parent or popup). dig2browser already applies
/// Emulation.setTimezoneOverride via stealth; we also force setLocaleOverride when
/// HATCHERY_BROWSER_LANG is set. When ACCEPT_LANGUAGES is set, re-assert
/// navigator.languages after dig2's single-locale JS override.
async fn apply_browser_locale(page: &StealthPage) {
    if let Ok(lang) = std::env::var("HATCHERY_BROWSER_LANG") {
        if !lang.is_empty() {
            let _ = page
                .cdp_call("Emulation.setLocaleOverride", Some(json!({ "locale": lang })))
                .await;
        }
    }
    if let Ok(tz) = std::env::var("HATCHERY_BROWSER_TZ") {
        if !tz.is_empty() {
            let _ = page
                .cdp_call(
                    "Emulation.setTimezoneOverride",
                    Some(json!({ "timezoneId": tz })),
                )
                .await;
        }
    }
    if let Some(al) = browser_accept_languages() {
        // dig2's --accept-lang is en-only; force the full en+ru header on this target.
        let _ = page.cdp_call("Network.enable", None).await;
        let _ = page
            .cdp_call(
                "Network.setExtraHTTPHeaders",
                Some(json!({ "headers": { "Accept-Language": al } })),
            )
            .await;
    }
    let tags = browser_language_tags();
    if tags.is_empty() {
        return;
    }
    let primary = tags.first().cloned().unwrap_or_else(|| "en-US".into());
    let tags_json = serde_json::to_string(&tags).unwrap_or_else(|_| "[]".into());
    let src = format!(
        r#"(function(){{
  const tags = {tags_json};
  try {{
    Object.defineProperty(Navigator.prototype, 'languages', {{
      get: () => Object.freeze(tags.slice()), configurable: true
    }});
    Object.defineProperty(Navigator.prototype, 'language', {{
      get: () => {primary:?}, configurable: true
    }});
  }} catch (_) {{}}
}})();"#
    );
    let _ = page
        .cdp_call(
            "Page.addScriptToEvaluateOnNewDocument",
            Some(json!({ "source": src })),
        )
        .await;
    let _ = page
        .eval(&format!(
            "(() => {{ try {{ Object.defineProperty(navigator,'languages',{{get:()=>Object.freeze({tags_json}),configurable:true}}); Object.defineProperty(navigator,'language',{{get:()=>{primary:?},configurable:true}}); }} catch(_){{}} return [navigator.language, navigator.languages]; }})()"
        ))
        .await;
}

/// One persistent browser profile shared by every session of this node.
/// Set by `HATCHERY_BROWSER_PROFILE` (a user-data-dir path); cookies and logins
/// survive node restarts and are the same for all sessions and CLIs.
pub mod shared {
    use super::*;
    use std::sync::OnceLock;

    pub fn profile() -> Option<PathBuf> {
        std::env::var_os("HATCHERY_BROWSER_PROFILE")
            .filter(|v| !v.is_empty())
            .map(PathBuf::from)
    }

    fn cell() -> &'static Mutex<Option<StealthBrowser>> {
        static CELL: OnceLock<Mutex<Option<StealthBrowser>>> = OnceLock::new();
        CELL.get_or_init(|| Mutex::new(None))
    }

    /// A Chrome left over from a previous node run still holds the profile lock.
    fn kill_stale(dir: &std::path::Path) {
        let want = format!("--user-data-dir={}", dir.display());
        let Ok(rd) = std::fs::read_dir("/proc") else { return };
        for entry in rd.flatten() {
            let Ok(pid) = entry.file_name().to_string_lossy().parse::<i32>() else { continue };
            let Ok(raw) = std::fs::read(entry.path().join("cmdline")) else { continue };
            let args: Vec<&[u8]> = raw.split(|b| *b == 0).collect();
            if args.iter().any(|a| *a == want.as_bytes()) {
                let _ = std::process::Command::new("kill").arg(pid.to_string()).status();
            }
        }
        std::thread::sleep(Duration::from_millis(500));
        for f in ["SingletonLock", "SingletonSocket", "SingletonCookie"] {
            let _ = std::fs::remove_file(dir.join(f));
        }
    }

    pub async fn open_page(dir: &std::path::Path) -> Result<StealthPage, String> {
        let mut g = cell().lock().await;
        let mut last = String::new();
        for _ in 0..2 {
            if g.is_none() {
                kill_stale(dir);
                ensure_profile_accept_languages(dir);
                let b = StealthBrowser::launch_with(launch_config(dir), stealth_config())
                    .await
                    .map_err(|e| e.to_string())?;
                *g = Some(b);
            }
            let b = g.as_ref().expect("browser set");
            match new_window_page(b, dir).await {
                Ok(p) => return Ok(p),
                Err(e) => {
                    tracing::warn!(error = %e, "shared browser unusable; relaunching");
                    last = e;
                    if let Some(old) = g.take() { let _ = old.close().await; }
                }
            }
        }
        Err(last)
    }

    /// Attach to an existing target (OAuth popup / window.open) in the shared Chrome.
    pub async fn attach(dir: &std::path::Path, target_id: &str) -> Result<StealthPage, String> {
        let g = cell().lock().await;
        let b = g.as_ref().ok_or("shared browser not running")?;
        // Ensure this is still the profile we expect (cell is process-global).
        let _ = dir;
        b.attach_page(target_id).await.map_err(|e| e.to_string())
    }

    /// Target.createTarget{newWindow} on the browser endpoint, then attach.
    async fn new_window_page(b: &StealthBrowser, dir: &std::path::Path) -> Result<StealthPage, String> {
        use futures_util::{SinkExt, StreamExt};
        use tokio_tungstenite::tungstenite::Message;
        let port = super::screencast::devtools_port(dir).ok_or("devtools port not found")?;
        let path = browser_ws_path(port).await?;
        let (mut ws, _) = tokio_tungstenite::connect_async(format!("ws://127.0.0.1:{port}{path}"))
            .await
            .map_err(|e| e.to_string())?;
        ws.send(Message::text(json!({ "id": 1, "method": "Target.createTarget",
            "params": { "url": "about:blank", "newWindow": true } }).to_string()))
            .await
            .map_err(|e| e.to_string())?;
        let target = tokio::time::timeout(Duration::from_secs(10), async {
            while let Some(Ok(m)) = ws.next().await {
                let Message::Text(t) = m else { continue };
                let Ok(v) = serde_json::from_str::<serde_json::Value>(&t) else { continue };
                if v["id"] == 1 {
                    return v["result"]["targetId"].as_str().map(str::to_owned).ok_or_else(|| t.to_string());
                }
            }
            Err("browser endpoint closed".to_string())
        })
        .await
        .map_err(|_| "createTarget timed out".to_string())??;
        let _ = ws.close(None).await;
        b.attach_page(&target).await.map_err(|e| e.to_string())
    }

    /// `/devtools/browser/<id>` from the DevTools HTTP endpoint (`/json/version`).
    async fn browser_ws_path(port: u16) -> Result<String, String> {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        let mut s = tokio::net::TcpStream::connect(("127.0.0.1", port)).await.map_err(|e| e.to_string())?;
        // Chrome keeps the connection open: read headers, then Content-Length bytes.
        s.write_all(b"GET /json/version HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n").await.map_err(|e| e.to_string())?;
        let mut buf = Vec::new();
        let body = tokio::time::timeout(Duration::from_secs(5), async {
            let mut chunk = [0u8; 4096];
            loop {
                let n = s.read(&mut chunk).await.map_err(|e| e.to_string())?;
                if n == 0 { return Err("devtools closed".to_string()) }
                buf.extend_from_slice(&chunk[..n]);
                let text = String::from_utf8_lossy(&buf).into_owned();
                if let Some((head, body)) = text.split_once("\r\n\r\n") {
                    let len = head.lines()
                        .find_map(|l| l.to_ascii_lowercase().strip_prefix("content-length:").and_then(|v| v.trim().parse::<usize>().ok()))
                        .unwrap_or(0);
                    if body.len() >= len { return Ok(body.to_owned()) }
                }
            }
        })
        .await
        .map_err(|_| "devtools /json/version timed out".to_string())??;
        let body = body.as_str();
        let v: serde_json::Value = serde_json::from_str(body).map_err(|e| e.to_string())?;
        let url = v["webSocketDebuggerUrl"].as_str().ok_or("no webSocketDebuggerUrl")?;
        url.find("/devtools/browser").map(|i| url[i..].to_owned()).ok_or_else(|| "bad webSocketDebuggerUrl".into())
    }
}

async fn page_meta(page: &StealthPage) -> Result<(String, String), String> {
    let v = page.eval("[location.href, document.title]").await.map_err(|e| e.to_string())?;
    let a = v.as_array().cloned().unwrap_or_default();
    Ok((
        a.first().and_then(|x| x.as_str()).unwrap_or_default().to_owned(),
        a.get(1).and_then(|x| x.as_str()).unwrap_or_default().to_owned(),
    ))
}

/// Delta frames: diff each decoded frame against the previous one in 32px tiles
/// and send only the changed rectangle when it is a small part of the screen.
mod delta {
    use super::BrowserFrameRegion;
    use image::{codecs::jpeg::JpegEncoder, ImageReader, RgbImage};

    const TILE: u32 = 32;
    #[derive(Default)]
    pub struct Delta { prev: Option<RgbImage> }
    pub enum Out { Unchanged, Full, Region(Vec<u8>, BrowserFrameRegion) }

    impl Delta {
        pub fn encode(&mut self, jpeg: &[u8]) -> Out {
            let Ok(img) = ImageReader::new(std::io::Cursor::new(jpeg)).with_guessed_format().map_err(|_| ()).and_then(|r| r.decode().map_err(|_| ())) else { return Out::Full };
            let cur = img.to_rgb8();
            let Some(prev) = self.prev.as_ref().filter(|p| p.dimensions() == cur.dimensions()) else {
                self.prev = Some(cur);
                return Out::Full;
            };
            let (w, h) = cur.dimensions();
            let (mut x0, mut y0, mut x1, mut y1) = (u32::MAX, u32::MAX, 0u32, 0u32);
            let (pr, cr) = (prev.as_raw(), cur.as_raw());
            for ty in (0..h).step_by(TILE as usize) {
                for tx in (0..w).step_by(TILE as usize) {
                    let tw = TILE.min(w - tx); let th = TILE.min(h - ty);
                    let mut changed = false;
                    for yy in ty..ty + th {
                        let a = ((yy * w + tx) * 3) as usize; let b = a + (tw * 3) as usize;
                        // JPEG noise tolerance: a tile changes when any byte moves > 24.
                        if pr[a..b].iter().zip(&cr[a..b]).any(|(p, c)| p.abs_diff(*c) > 24) { changed = true; break; }
                    }
                    if changed { x0 = x0.min(tx); y0 = y0.min(ty); x1 = x1.max(tx + tw); y1 = y1.max(ty + th); }
                }
            }
            if x0 == u32::MAX { return Out::Unchanged; }
            let (rw, rh) = (x1 - x0, y1 - y0);
            if (rw * rh) as u64 * 2 > (w * h) as u64 {
                self.prev = Some(cur);
                return Out::Full;
            }
            let crop = image::imageops::crop_imm(&cur, x0, y0, rw, rh).to_image();
            let mut out = Vec::new();
            if JpegEncoder::new_with_quality(&mut out, 70).encode_image(&crop).is_err() { self.prev = Some(cur); return Out::Full; }
            // Base the next diff on what the viewer now shows (prev + this region).
            let mut next = prev.clone();
            image::imageops::replace(&mut next, &crop, x0 as i64, y0 as i64);
            self.prev = Some(next);
            Out::Region(out, BrowserFrameRegion { x: x0, y: y0, w: rw, h: rh, src_w: w, src_h: h })
        }

        /// Current composite as a full JPEG (for a viewer that just attached).
        pub fn full_jpeg(&self) -> Option<Vec<u8>> {
            let img = self.prev.as_ref()?;
            let mut out = Vec::new();
            JpegEncoder::new_with_quality(&mut out, 60).encode_image(img).ok()?;
            Some(out)
        }
    }
}

/// Watch the browser target tree for window.open / OAuth popups belonging to our page.
mod popups {
    use super::*;
    use futures_util::{SinkExt, StreamExt};
    use tokio_tungstenite::tungstenite::Message;

    pub enum Ev {
        /// A page target opened by `opener` (our session page).
        Opened { target_id: String, url: String },
        /// A page target we were watching went away (window.close).
        Closed { target_id: String },
    }

    /// Discover page targets that our `opener` already opened (e.g. Google OAuth
    /// before we started watching) and watch for new ones.
    pub async fn start(
        profile_dir: &std::path::Path,
        opener_id: String,
    ) -> Result<(mpsc::UnboundedReceiver<Ev>, tokio::task::JoinHandle<()>), String> {
        let port = screencast::devtools_port(profile_dir).ok_or("devtools port not found")?;
        let path = browser_ws_path(port).await?;
        let (mut ws, _) = tokio_tungstenite::connect_async(format!("ws://127.0.0.1:{port}{path}"))
            .await
            .map_err(|e| e.to_string())?;
        let (tx, rx) = mpsc::unbounded_channel();
        // Enable target discovery; also pull current targets so an already-open
        // OAuth window is adopted immediately.
        ws.send(Message::text(
            json!({ "id": 1, "method": "Target.setDiscoverTargets", "params": { "discover": true } })
                .to_string(),
        ))
        .await
        .map_err(|e| e.to_string())?;
        ws.send(Message::text(json!({ "id": 2, "method": "Target.getTargets" }).to_string()))
            .await
            .map_err(|e| e.to_string())?;
        let join = tokio::spawn(async move {
            while let Some(Ok(msg)) = ws.next().await {
                let Message::Text(text) = msg else { continue };
                let Ok(v) = serde_json::from_str::<serde_json::Value>(&text) else { continue };
                if v["id"] == 2 {
                    if let Some(arr) = v["result"]["targetInfos"].as_array() {
                        for t in arr {
                            maybe_open(&tx, &opener_id, t);
                        }
                    }
                    continue;
                }
                match v["method"].as_str() {
                    Some("Target.targetCreated") => {
                        maybe_open(&tx, &opener_id, &v["params"]["targetInfo"]);
                    }
                    Some("Target.targetDestroyed") => {
                        if let Some(id) = v["params"]["targetId"].as_str() {
                            let _ = tx.send(Ev::Closed {
                                target_id: id.to_owned(),
                            });
                        }
                    }
                    Some("Target.targetInfoChanged") => {
                        // URL fills in after creation; surface Opened again with the URL.
                        maybe_open(&tx, &opener_id, &v["params"]["targetInfo"]);
                    }
                    _ => {}
                }
            }
        });
        Ok((rx, join))
    }

    fn maybe_open(tx: &mpsc::UnboundedSender<Ev>, opener_id: &str, info: &serde_json::Value) {
        let ty = info["type"].as_str().unwrap_or("");
        if ty != "page" && ty != "webview" {
            return;
        }
        let Some(tid) = info["targetId"].as_str() else { return };
        let opener = info["openerId"].as_str().unwrap_or("");
        if opener != opener_id {
            return;
        }
        let url = info["url"].as_str().unwrap_or("").to_owned();
        let _ = tx.send(Ev::Opened {
            target_id: tid.to_owned(),
            url,
        });
    }

    async fn browser_ws_path(port: u16) -> Result<String, String> {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        let mut s = tokio::net::TcpStream::connect(("127.0.0.1", port))
            .await
            .map_err(|e| e.to_string())?;
        s.write_all(b"GET /json/version HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n")
            .await
            .map_err(|e| e.to_string())?;
        let mut buf = Vec::new();
        let body = tokio::time::timeout(Duration::from_secs(5), async {
            let mut chunk = [0u8; 4096];
            loop {
                let n = s.read(&mut chunk).await.map_err(|e| e.to_string())?;
                if n == 0 {
                    return Err("devtools closed".to_string());
                }
                buf.extend_from_slice(&chunk[..n]);
                let text = String::from_utf8_lossy(&buf).into_owned();
                if let Some((head, body)) = text.split_once("\r\n\r\n") {
                    let len = head
                        .lines()
                        .find_map(|l| {
                            l.to_ascii_lowercase()
                                .strip_prefix("content-length:")
                                .and_then(|v| v.trim().parse::<usize>().ok())
                        })
                        .unwrap_or(0);
                    if body.len() >= len {
                        return Ok(body.to_owned());
                    }
                }
            }
        })
        .await
        .map_err(|_| "devtools /json/version timed out".to_string())??;
        let v: serde_json::Value = serde_json::from_str(&body).map_err(|e| e.to_string())?;
        let url = v["webSocketDebuggerUrl"]
            .as_str()
            .ok_or("no webSocketDebuggerUrl")?;
        url.find("/devtools/browser")
            .map(|i| url[i..].to_owned())
            .ok_or_else(|| "bad webSocketDebuggerUrl".into())
    }
}

/// Side CDP websocket to our page target for Page.startScreencast.
mod screencast {
    use super::*;
    use futures_util::{SinkExt, StreamExt};
    use tokio_tungstenite::tungstenite::Message;

    /// DevTools port of the Chrome that owns `profile_dir`: DevToolsActivePort if
    /// present, else the --remote-debugging-port of the process using that profile.
    pub(super) fn devtools_port(profile_dir: &std::path::Path) -> Option<u16> {
        if let Ok(s) = std::fs::read_to_string(profile_dir.join("DevToolsActivePort")) {
            if let Some(p) = s.lines().next().and_then(|l| l.trim().parse().ok()) { return Some(p); }
        }
        let want = format!("--user-data-dir={}", profile_dir.display());
        for entry in std::fs::read_dir("/proc").ok()?.flatten() {
            let Ok(raw) = std::fs::read(entry.path().join("cmdline")) else { continue };
            let args: Vec<String> = raw.split(|b| *b == 0).map(|a| String::from_utf8_lossy(a).into_owned()).collect();
            if !args.iter().any(|a| a == &want) { continue; }
            if let Some(p) = args.iter().find_map(|a| a.strip_prefix("--remote-debugging-port=").and_then(|v| v.parse().ok())) {
                return Some(p);
            }
        }
        None
    }

    /// Max JPEG bytes per frame so one frame fits a single restream message.
    const BUDGET: usize = 40 * 1024;

    pub async fn start(
        profile_dir: &std::path::Path,
        page: &StealthPage,
        out: mpsc::Sender<Vec<u8>>,
    ) -> Result<tokio::task::JoinHandle<()>, String> {
        let port = devtools_port(profile_dir).ok_or("devtools port not found")?;
        let info = page.cdp_call("Target.getTargetInfo", None).await.map_err(|e| e.to_string())?;
        let target = info["targetInfo"]["targetId"].as_str().ok_or("no target id")?.to_owned();
        let url = format!("ws://127.0.0.1:{port}/devtools/page/{target}");
        let (ws, _) = tokio_tungstenite::connect_async(&url).await.map_err(|e| e.to_string())?;
        let (mut tx, mut rx) = ws.split();
        let mut quality: u32 = 50;
        let mut max_w: u32 = VIEW_W;
        let mut id: u64 = 1;
        let start_msg = |id: u64, q: u32, w: u32| json!({ "id": id, "method": "Page.startScreencast",
            "params": { "format": "jpeg", "quality": q, "maxWidth": w, "maxHeight": w * VIEW_H / VIEW_W, "everyNthFrame": 1 } }).to_string();
        tx.send(Message::text(json!({ "id": id, "method": "Page.enable" }).to_string())).await.map_err(|e| e.to_string())?;
        id += 1;
        tx.send(Message::text(start_msg(id, quality, max_w))).await.map_err(|e| e.to_string())?;
        Ok(tokio::spawn(async move {
            while let Some(Ok(msg)) = rx.next().await {
                let Message::Text(text) = msg else { continue };
                let Ok(v) = serde_json::from_str::<serde_json::Value>(&text) else { continue };
                if v["method"] != "Page.screencastFrame" { continue; }
                let sid = v["params"]["sessionId"].clone();
                // Ack first so Chrome can start producing the next frame immediately.
                id += 1;
                let _ = tx.send(Message::text(json!({ "id": id, "method": "Page.screencastFrameAck", "params": { "sessionId": sid } }).to_string())).await;
                let Some(b64) = v["params"]["data"].as_str() else { continue };
                use base64::Engine;
                let Ok(bytes) = base64::engine::general_purpose::STANDARD.decode(b64) else { continue };
                if bytes.len() > BUDGET {
                    // Too big for one restream message: step quality, then size, and restart.
                    if quality > 25 { quality -= 15 } else if max_w > 640 { max_w = max_w * 3 / 4 } else { continue }
                    id += 1;
                    let _ = tx.send(Message::text(json!({ "id": id, "method": "Page.stopScreencast" }).to_string())).await;
                    id += 1;
                    let _ = tx.send(Message::text(start_msg(id, quality, max_w))).await;
                    continue;
                }
                // Keep latency low: if the consumer is behind, drop rather than queue.
                let _ = out.try_send(bytes);
            }
        }))
    }
}

async fn capture_frame(page: &StealthPage) -> Result<Vec<u8>, String> {
    // JPEG via CDP, sized to fit one Noise transport message (64 KiB incl. envelope):
    // step quality down, then scale, until under FRAME_BUDGET. Oversized frames used to
    // kill the restream uplink ("noise: input error") on busy pages.
    const FRAME_BUDGET: usize = 40 * 1024;
    let mut bytes = Vec::new();
    'fit: for scale in [1.0f64, 0.75, 0.5] {
        for q in [60u32, 45, 32, 22, 14] {
            let mut params = json!({ "format": "jpeg", "quality": q });
            if scale < 1.0 {
                params["clip"] = json!({ "x": 0, "y": 0, "width": VIEW_W, "height": VIEW_H, "scale": scale });
            }
            let v = page.cdp_call("Page.captureScreenshot", Some(params)).await.map_err(|e| e.to_string())?;
            let b64 = v.get("data").and_then(|x| x.as_str()).ok_or("no jpeg data")?;
            use base64::Engine;
            bytes = base64::engine::general_purpose::STANDARD.decode(b64).map_err(|e| e.to_string())?;
            if bytes.len() <= FRAME_BUDGET {
                break 'fit;
            }
        }
    }
    if bytes.len() > FRAME_BUDGET {
        return Err(format!("frame too large ({} bytes)", bytes.len()));
    }
    Ok(bytes)
}

async fn apply_input(page: &StealthPage, input: &BrowserInput) -> Result<(), String> {
    match input {
        BrowserInput::Click { x, y, button } => {
            // Raw CDP, viewport CSS pixels (DPR 1): move, press, release — a real
            // trusted click that also reaches cross-origin iframes (e.g. Turnstile).
            let b = match button {
                BrowserMouseButton::Left => "left",
                BrowserMouseButton::Right => "right",
                BrowserMouseButton::Middle => "middle",
            };
            tracing::info!(x = *x, y = *y, button = b, "d2b click");
            page.cdp_call(
                "Input.dispatchMouseEvent",
                Some(json!({ "type": "mouseMoved", "x": x, "y": y, "button": "none" })),
            )
            .await
            .map_err(|e| e.to_string())?;
            for ty in ["mousePressed", "mouseReleased"] {
                page.cdp_call(
                    "Input.dispatchMouseEvent",
                    Some(json!({ "type": ty, "x": x, "y": y, "button": b, "buttons": if ty == "mousePressed" { 1 } else { 0 }, "clickCount": 1 })),
                )
                .await
                .map_err(|e| e.to_string())?;
            }
        }
        BrowserInput::MouseDown { x, y, button } | BrowserInput::MouseUp { x, y, button } => {
            let down = matches!(input, BrowserInput::MouseDown { .. });
            let b = match button {
                BrowserMouseButton::Left => "left",
                BrowserMouseButton::Right => "right",
                BrowserMouseButton::Middle => "middle",
            };
            let count = mouse::click_count(down, *x, *y);
            if down {
                page.cdp_call("Input.dispatchMouseEvent", Some(json!({ "type": "mouseMoved", "x": x, "y": y, "button": "none" })))
                    .await
                    .map_err(|e| e.to_string())?;
            }
            page.cdp_call(
                "Input.dispatchMouseEvent",
                Some(json!({ "type": if down { "mousePressed" } else { "mouseReleased" }, "x": x, "y": y,
                    "button": b, "buttons": if down { mouse::bit(b) } else { 0 }, "clickCount": count })),
            )
            .await
            .map_err(|e| e.to_string())?;
            mouse::set_held(if down { mouse::bit(b) } else { 0 });
        }
        BrowserInput::MouseMove { x, y } => {
            let held = mouse::held();
            let b = match held { 1 => "left", 2 => "right", 4 => "middle", _ => "none" };
            page.cdp_call("Input.dispatchMouseEvent", Some(json!({ "type": "mouseMoved", "x": x, "y": y, "button": b, "buttons": held })))
                .await
                .map_err(|e| e.to_string())?;
        }
        BrowserInput::Scroll { x, y, dx, dy } => {
            page.wheel(*x, *y, *dx, *dy)
                .await
                .map_err(|e| e.to_string())?
        }
        BrowserInput::Key { key } => keys::press(page, key).await?,
        BrowserInput::Text { text } => {
            // Prefer CDP insertText; fall back to per-character key_press.
            match page
                .cdp_call("Input.insertText", Some(json!({ "text": text })))
                .await
            {
                Ok(_) => {}
                Err(_) => {
                    for ch in text.chars() {
                        page.key_press(&ch.to_string())
                            .await
                            .map_err(|e| e.to_string())?;
                    }
                }
            }
        }
        BrowserInput::Navigate { url } if url.is_empty() => {}
        BrowserInput::Navigate { url } => {
            // Raw Page.navigate returns once the request commits; frames keep
            // streaming while the page loads (goto blocked the loop on load).
            // Cap the wait so a hung CDP cannot freeze the capture loop.
            tracing::info!(%url, "d2b navigate");
            let _ = page.cdp_call("Page.bringToFront", None).await;
            match tokio::time::timeout(
                Duration::from_secs(5),
                page.cdp_call("Page.navigate", Some(json!({ "url": url }))),
            )
            .await
            {
                Ok(Ok(r)) => {
                    if let Some(err) = r.get("errorText").and_then(|v| v.as_str()) {
                        if !err.is_empty() {
                            return Err(format!("navigate: {err}"));
                        }
                    }
                }
                Ok(Err(e)) => return Err(e.to_string()),
                Err(_) => tracing::warn!(%url, "d2b navigate timed out; continuing"),
            }
        }
    }
    Ok(())
}

/// Mouse state for press/drag/release (one operator drives a page at a time).
mod mouse {
    use std::sync::atomic::{AtomicU32, Ordering};
    use std::sync::Mutex;
    static HELD: AtomicU32 = AtomicU32::new(0);
    static LAST: Mutex<Option<(std::time::Instant, f64, f64, u32)>> = Mutex::new(None);

    pub fn bit(b: &str) -> u32 { match b { "left" => 1, "right" => 2, "middle" => 4, _ => 0 } }
    pub fn held() -> u32 { HELD.load(Ordering::Relaxed) }
    pub fn set_held(v: u32) { HELD.store(v, Ordering::Relaxed) }

    /// Double / triple click: presses close in time and place count up.
    pub fn click_count(down: bool, x: f64, y: f64) -> u32 {
        let mut g = LAST.lock().unwrap_or_else(|e| e.into_inner());
        let now = std::time::Instant::now();
        let prev = g.filter(|(t, px, py, _)| now.duration_since(*t).as_millis() < 500 && (px - x).abs() < 6.0 && (py - y).abs() < 6.0);
        if !down { return prev.map(|p| p.3).unwrap_or(1) }
        let n = prev.map(|p| p.3 % 3 + 1).unwrap_or(1);
        *g = Some((now, x, y, n));
        n
    }
}

/// Shared map for the restream bridge (process-global for the node binary).
pub fn shared_map() -> Arc<BrowserSessionMap> {
    use std::sync::OnceLock;
    static MAP: OnceLock<Arc<BrowserSessionMap>> = OnceLock::new();
    MAP.get_or_init(|| Arc::new(BrowserSessionMap::new())).clone()
}

/// Keyboard over raw CDP: "Control+Shift+a" style chords, named keys with their
/// virtual key codes (Backspace, Delete, arrows...), editing commands for the
/// usual shortcuts, and browser navigation keys (reload, back, forward).
mod keys {
    use super::*;

    pub struct Chord<'a> { pub mods: u32, pub key: &'a str }

    pub fn parse(combo: &str) -> Chord<'_> {
        let mut mods = 0;
        let mut rest = combo;
        loop {
            let Some((head, tail)) = rest.split_once('+') else { break };
            if tail.is_empty() { break }
            let bit = match head { "Alt" => 1, "Control" | "Ctrl" => 2, "Meta" => 4, "Shift" => 8, _ => break };
            mods |= bit;
            rest = tail;
        }
        Chord { mods, key: rest }
    }

    /// (code, windowsVirtualKeyCode, text)
    pub fn def(key: &str) -> (String, u32, Option<String>) {
        let named = |code: &str, vk: u32, text: Option<&str>| (code.to_owned(), vk, text.map(str::to_owned));
        match key {
            "Backspace" => named("Backspace", 8, None),
            "Tab" => named("Tab", 9, None),
            "Enter" => named("Enter", 13, Some("\r")),
            "Escape" => named("Escape", 27, None),
            "PageUp" => named("PageUp", 33, None),
            "PageDown" => named("PageDown", 34, None),
            "End" => named("End", 35, None),
            "Home" => named("Home", 36, None),
            "ArrowLeft" => named("ArrowLeft", 37, None),
            "ArrowUp" => named("ArrowUp", 38, None),
            "ArrowRight" => named("ArrowRight", 39, None),
            "ArrowDown" => named("ArrowDown", 40, None),
            "Insert" => named("Insert", 45, None),
            "Delete" => named("Delete", 46, None),
            k if k.len() > 1 && k.starts_with('F') && k[1..].parse::<u32>().is_ok_and(|n| (1..=12).contains(&n)) => {
                let n: u32 = k[1..].parse().unwrap_or(1);
                (k.to_owned(), 111 + n, None)
            }
            k => {
                let mut it = k.chars();
                match (it.next(), it.next()) {
                    (Some(c), None) if c.is_ascii_alphabetic() => {
                        let u = c.to_ascii_uppercase();
                        (format!("Key{u}"), u as u32, Some(k.to_owned()))
                    }
                    (Some(c), None) if c.is_ascii_digit() => (format!("Digit{c}"), c as u32, Some(k.to_owned())),
                    (Some(' '), None) => ("Space".into(), 32, Some(" ".into())),
                    (Some(_), None) => (String::new(), 0, Some(k.to_owned())),
                    _ => (String::new(), 0, None),
                }
            }
        }
    }

    pub fn commands(mods: u32, key: &str) -> Vec<&'static str> {
        if mods & 6 == 0 { return vec![] }
        match (key.to_ascii_lowercase().as_str(), mods & 8 != 0) {
            ("a", _) => vec!["selectAll"],
            ("c", _) => vec!["copy"],
            ("x", _) => vec!["cut"],
            ("v", _) => vec!["paste"],
            ("z", false) => vec!["undo"],
            ("z", true) | ("y", _) => vec!["redo"],
            _ => vec![],
        }
    }

    pub fn is_copy(combo: &str) -> bool {
        let c = parse(combo);
        c.mods & 6 != 0 && matches!(c.key, "c" | "C" | "x" | "X" | "Insert")
    }

    /// Selected text of the page (inputs and textareas included), capped.
    pub async fn selection(page: &StealthPage) -> Option<String> {
        let v = page.eval("(() => { const a = document.activeElement; let t = ''; \
            if (a && (a.tagName === 'INPUT' || a.tagName === 'TEXTAREA') && typeof a.selectionStart === 'number') t = a.value.slice(a.selectionStart, a.selectionEnd); \
            if (!t) t = String(window.getSelection() || ''); return t.slice(0, 100000); })()").await.ok()?;
        v.as_str().map(str::to_owned)
    }

    enum Nav { Reload, Back, Forward }

    fn nav(c: &Chord<'_>) -> Option<Nav> {
        match (c.key, c.mods) {
            ("BrowserRefresh", _) | ("F5", _) => Some(Nav::Reload),
            ("r" | "R", m) if m & 6 != 0 => Some(Nav::Reload),
            ("BrowserBack", _) => Some(Nav::Back),
            ("ArrowLeft", 1) => Some(Nav::Back),
            ("BrowserForward", _) => Some(Nav::Forward),
            ("ArrowRight", 1) => Some(Nav::Forward),
            _ => None,
        }
    }

    async fn history(page: &StealthPage, step: i64) -> Result<(), String> {
        let h = page.cdp_call("Page.getNavigationHistory", None).await.map_err(|e| e.to_string())?;
        let i = h["currentIndex"].as_i64().unwrap_or(0) + step;
        let entries = h["entries"].as_array().cloned().unwrap_or_default();
        let Some(e) = usize::try_from(i).ok().and_then(|i| entries.get(i)) else { return Ok(()) };
        page.cdp_call("Page.navigateToHistoryEntry", Some(json!({ "entryId": e["id"] })))
            .await
            .map_err(|e| e.to_string())?;
        Ok(())
    }

    pub async fn press(page: &StealthPage, combo: &str) -> Result<(), String> {
        let c = parse(combo);
        if matches!(c.key, "Shift" | "Control" | "Alt" | "Meta" | "CapsLock" | "") { return Ok(()) }
        if let Some(n) = nav(&c) {
            return match n {
                Nav::Reload => page.cdp_call("Page.reload", None).await.map(|_| ()).map_err(|e| e.to_string()),
                Nav::Back => history(page, -1).await,
                Nav::Forward => history(page, 1).await,
            };
        }
        let (code, vk, text) = def(c.key);
        let typed = if c.mods & 7 == 0 { text } else { None };
        let mut down = json!({
            "type": if typed.is_some() { "keyDown" } else { "rawKeyDown" },
            "modifiers": c.mods, "key": c.key, "code": code,
            "windowsVirtualKeyCode": vk, "nativeVirtualKeyCode": vk,
        });
        if let Some(t) = &typed { down["text"] = json!(t); down["unmodifiedText"] = json!(t); }
        let cmds = commands(c.mods, c.key);
        if !cmds.is_empty() { down["commands"] = json!(cmds); }
        page.cdp_call("Input.dispatchKeyEvent", Some(down)).await.map_err(|e| e.to_string())?;
        page.cdp_call("Input.dispatchKeyEvent", Some(json!({
            "type": "keyUp", "modifiers": c.mods, "key": c.key, "code": code,
            "windowsVirtualKeyCode": vk, "nativeVirtualKeyCode": vk,
        }))).await.map_err(|e| e.to_string())?;
        Ok(())
    }

    #[cfg(test)]
    mod tests {
        use super::*;
        #[test]
        fn chords_parse() {
            let c = parse("Control+Shift+a"); assert_eq!((c.mods, c.key), (10, "a"));
            let c = parse("Control++"); assert_eq!((c.mods, c.key), (2, "+"));
            let c = parse("Backspace"); assert_eq!((c.mods, c.key), (0, "Backspace"));
            assert_eq!(def("Delete").1, 46);
            assert_eq!(def("a").0, "KeyA");
            assert_eq!(commands(2, "a"), vec!["selectAll"]);
        }
    }
}