use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;

use futures_util::{SinkExt, StreamExt};
use serde_json::json;
use tokio::net::TcpStream;
use tokio::sync::mpsc;
use tokio::sync::Notify;
use tokio_tungstenite::tungstenite::Message;
use tracing::{error, info, warn};

use crate::dispatch::{self, CdpContext};

// PR #36 comment 4341743194: the deferral queue in `process_with_interception`
// must be bounded so a stalled navigation cannot OOM the process. When the cap
// is reached we return an explicit error response rather than silently dropping.
const MAX_DEFERRED_MESSAGES: usize = 256;

// The WS-stream forwarding channel must also be bounded: if the LocalSet
// (CDP processor + nav tasks) stalls, the accept thread keeps pushing
// `std::net::TcpStream`s into the queue. An unbounded channel would let
// that queue grow without limit and OOM the process. With a bounded
// capacity, when the LocalSet is saturated the accept thread closes the
// new connection on the spot instead of buffering it — the kernel TCP
// backlog still absorbs short-term spikes, but a long-term stall now
// fails loudly at accept time rather than silently piling up FDs.
const MAX_PENDING_WS_HANDOFFS: usize = 128;
use crate::types::CdpRequest;

struct CdpMessage {
    text: String,
    reply_tx: mpsc::UnboundedSender<String>,
}

enum ServerMessage {
    Cdp(CdpMessage),
    NewConnection {
        reply_tx: mpsc::UnboundedSender<String>,
    },
}

pub async fn start(port: u16) -> anyhow::Result<()> {
    start_with_options(port, None, false).await
}

pub async fn start_with_options(
    port: u16,
    proxy: Option<String>,
    stealth: bool,
) -> anyhow::Result<()> {
    start_with_full_options(port, proxy, stealth, None, None).await
}

pub async fn start_with_full_options(
    port: u16,
    proxy: Option<String>,
    stealth: bool,
    user_agent: Option<String>,
    storage_dir: Option<std::path::PathBuf>,
) -> anyhow::Result<()> {
    start_with_host(port, "127.0.0.1", proxy, stealth, user_agent, storage_dir).await
}

pub async fn start_with_host(
    port: u16,
    host: &str,
    proxy: Option<String>,
    stealth: bool,
    user_agent: Option<String>,
    storage_dir: Option<std::path::PathBuf>,
) -> anyhow::Result<()> {
    start_with_host_and_security(port, host, proxy, stealth, user_agent, false, storage_dir).await
}

pub async fn start_with_host_and_security(
    port: u16,
    host: &str,
    proxy: Option<String>,
    stealth: bool,
    user_agent: Option<String>,
    allow_file_access: bool,
    storage_dir: Option<std::path::PathBuf>,
) -> anyhow::Result<()> {
    start_with_full_serve_options(
        port, host, proxy, stealth, user_agent, allow_file_access, storage_dir, false,
    )
    .await
}

pub async fn start_with_host_security_and_storage(
    port: u16,
    host: &str,
    proxy: Option<String>,
    stealth: bool,
    user_agent: Option<String>,
    allow_file_access: bool,
    storage_dir: Option<std::path::PathBuf>,
) -> anyhow::Result<()> {
    start_with_full_serve_options(
        port, host, proxy, stealth, user_agent, allow_file_access, storage_dir, false,
    )
    .await
}

/// Full serve entry point that also accepts `allow_private_network` (issue
/// #33). Older entry points default it to `false` so existing callers and
/// public API consumers are unaffected.
pub async fn start_with_full_serve_options(
    port: u16,
    host: &str,
    proxy: Option<String>,
    stealth: bool,
    user_agent: Option<String>,
    allow_file_access: bool,
    storage_dir: Option<std::path::PathBuf>,
    allow_private_network: bool,
) -> anyhow::Result<()> {
    let ip: std::net::IpAddr = host
        .parse()
        .map_err(|e| anyhow::anyhow!("invalid --host '{}': {}", host, e))?;
    let addr = SocketAddr::new(ip, port);

    // Issue #62: the HTTP control plane (/json/version, /json) must remain
    // reachable even while V8 JS evaluation blocks the tokio LocalSet thread.
    //
    // We use a dedicated OS thread with a blocking std::net::TcpListener so
    // the kernel's accept backlog is always drained promptly. HTTP endpoints
    // are served directly via blocking I/O; WebSocket connections are
    // forwarded to the existing LocalSet for CDP processing.
    let std_listener = std::net::TcpListener::bind(addr)
        .map_err(|e| anyhow::anyhow!("bind {}:{}: {}", host, port, e))?;
    std_listener
        .set_nonblocking(false)
        .map_err(|e| anyhow::anyhow!("set_nonblocking: {}", e))?;

    info!("Obscura CDP server listening on ws://{}:{}", host, port);
    info!(
        "DevTools endpoint: ws://{}:{}/devtools/browser",
        host, port
    );
    if allow_file_access {
        info!("file:// navigation enabled (--allow-file-access). Do not expose this port to untrusted networks.");
    }

    let (ws_tx, mut ws_rx) = mpsc::channel::<std::net::TcpStream>(MAX_PENDING_WS_HANDOFFS);

    // Issue #225: coordinate Ctrl-C shutdown across the accept thread, the
    // LocalSet accept loop, and `cdp_processor`. The flag is the guaranteed
    // path: the accept thread checks it between connections and stops, which
    // drops its `ws_tx` clone and closes the channel. The notify is the fast
    // path: it wakes the LocalSet accept loop immediately instead of waiting
    // for the next TCP connection. `cdp_processor` sets both when it catches
    // Ctrl-C.
    let shutdown_flag = Arc::new(AtomicBool::new(false));
    let shutdown_notify = Arc::new(Notify::new());

    // Dedicated accept thread: drains the kernel backlog immediately and
    // handles HTTP endpoints (/json/version, /json, /json/protocol) with
    // blocking I/O so they never contend with the LocalSet's V8 work.
    //
    // Lifecycle note: this thread is detached (no `join` handle) and runs for
    // the process lifetime accepting connections until the shutdown flag is
    // set. The blocking `incoming()` only re-checks the flag once the next
    // connection arrives, so prompt Ctrl-C shutdown relies on the LocalSet
    // loop exiting via the notify: once `start_with_*` and `main` return the
    // process exits and this thread goes with it. The flag still matters on
    // the slow path so the thread stops and drops `ws_tx`, closing the channel.
    let accept_flag = shutdown_flag.clone();
    std::thread::Builder::new()
        .name("obscura-cdp-accept".into())
        .spawn(move || {
            for stream in std_listener.incoming() {
                if accept_flag.load(Ordering::Relaxed) {
                    break;
                }
                match stream {
                    Ok(stream) => {
                        if let Err(e) = accept_dispatch(stream, port, &ws_tx) {
                            if !format!("{}", e).contains("close") {
                                error!("Accept dispatch error: {}", e);
                            }
                        }
                    }
                    Err(e) => error!("Accept error: {}", e),
                }
            }
        })?;

    let local = tokio::task::LocalSet::new();
    local
        .run_until(async {
            let (msg_tx, msg_rx) = mpsc::unbounded_channel::<ServerMessage>();

            let _processor_handle = tokio::task::spawn_local(cdp_processor(
                msg_rx, proxy, stealth, user_agent, allow_file_access, storage_dir,
                allow_private_network, shutdown_flag.clone(), shutdown_notify.clone(),
            ));

            // Issue #225: exit on either the channel closing (accept thread
            // stopped and dropped its ws_tx) or a shutdown notification from
            // cdp_processor on Ctrl-C. Waiting only on ws_rx.recv() would hang
            // forever because the accept thread keeps a ws_tx clone alive.
            let accept_notify = shutdown_notify.clone();
            loop {
                let stream = tokio::select! {
                    stream = ws_rx.recv() => stream,
                    _ = accept_notify.notified() => None,
                };
                let stream = match stream {
                    Some(s) => s,
                    None => break,
                };
                // Convert std TcpStream → tokio TcpStream inside the LocalSet
                // where the tokio runtime is active.
                stream
                    .set_nonblocking(true)
                    .map_err(|e| error!("set_nonblocking on WS stream: {}", e))
                    .ok();
                // Disable Nagle on the CDP socket. CDP exchanges many small
                // (~100-byte) request/response frames during browser.newPage()
                // and Page.navigate; with Nagle on, every small write either
                // waits for an ACK from the previous one or hits the 40ms
                // delayed-ACK timer. Measured impact: ~90ms shaved off
                // newPage on localhost, ~30ms shaved off goto.
                stream
                    .set_nodelay(true)
                    .map_err(|e| error!("set_nodelay on WS stream: {}", e))
                    .ok();
                let tokio_stream = match TcpStream::from_std(stream) {
                    Ok(s) => s,
                    Err(e) => {
                        error!("TcpStream::from_std failed: {}", e);
                        continue;
                    }
                };
                let tx = msg_tx.clone();
                tokio::task::spawn_local(async move {
                    if let Err(e) = handle_connection_ws(tokio_stream, tx).await {
                        error!("WebSocket connection error: {}", e);
                    }
                });
            }
        })
        .await;

    Ok(())
}

const HTTP_PEEK_BUF: usize = 4096;
const WS_PEEK_BUF: usize = 4;

/// Dispatch a freshly-accepted TCP connection on the dedicated accept thread.
///
/// Peek at the first bytes to decide HTTP vs WebSocket:
/// - HTTP (`GET /json/*`): serve synchronously via blocking I/O so the
///   response is never stalled by the LocalSet.
/// - WebSocket: set non-blocking, convert to tokio `TcpStream`, and forward
///   to the LocalSet for CDP processing.
fn accept_dispatch(
    stream: std::net::TcpStream,
    port: u16,
    ws_tx: &mpsc::Sender<std::net::TcpStream>,
) -> anyhow::Result<()> {
    let mut buf = [0u8; WS_PEEK_BUF];
    let n = stream.peek(&mut buf)?;

    if n >= 4 && &buf == b"GET " {
        let mut peek_buf = [0u8; HTTP_PEEK_BUF];
        let n = stream.peek(&mut peek_buf)?;
        let line = String::from_utf8_lossy(&peek_buf[..n]);

        let endpoint = if line.contains("/json/version") {
            Some("version")
        } else if line.contains("/json/list") || line.contains("/json\r\n") || line.contains("/json HTTP") {
            Some("list")
        } else if line.contains("/json/protocol") {
            Some("protocol")
        } else {
            None
        };

        if let Some(ep) = endpoint {
            return handle_http_json_blocking(stream, port, ep);
        }
        // Fall through: GET request that isn't a /json endpoint → treat as
        // WebSocket upgrade (Chromium DevTools clients issue GET with
        // Upgrade: websocket).
    }

    // Try to hand off the WS stream to the LocalSet. If the bounded channel
    // is full the LocalSet is saturated — drop the connection cleanly
    // rather than blocking the accept thread (which would freeze the HTTP
    // control plane that this whole rework exists to keep alive). The
    // dropped `stream` closes itself; the client will see ECONNRESET and
    // can retry.
    ws_tx
        .try_send(stream)
        .map_err(|e| match e {
            mpsc::error::TrySendError::Full(_) => {
                warn!("WS handoff channel full ({}); dropping new WebSocket connection", MAX_PENDING_WS_HANDOFFS);
                anyhow::anyhow!("ws handoff channel full")
            }
            mpsc::error::TrySendError::Closed(_) => anyhow::anyhow!("accept channel closed"),
        })
}

/// Serve an HTTP `/json/*` endpoint with blocking I/O on the accept thread.
fn handle_http_json_blocking(
    mut stream: std::net::TcpStream,
    port: u16,
    endpoint: &str,
) -> anyhow::Result<()> {
    use std::io::{Read, Write};

    let mut buf = vec![0u8; 4096];
    let _ = stream.read(&mut buf)?;

    let body = match endpoint {
        "version" => serde_json::to_string_pretty(&json!({
            "Browser": "Obscura/0.1.0",
            "Protocol-Version": "1.3",
            "User-Agent": "Obscura/0.1.0 (Headless Browser)",
            "V8-Version": "N/A",
            "WebKit-Version": "N/A",
            "webSocketDebuggerUrl": format!("ws://127.0.0.1:{}/devtools/browser", port),
        }))?,
        "list" => serde_json::to_string_pretty(&json!([{
            "description": "",
            "devtoolsFrontendUrl": "",
            "id": "page-1",
            "title": "",
            "type": "page",
            "url": "about:blank",
            "webSocketDebuggerUrl": format!("ws://127.0.0.1:{}/devtools/page/page-1", port),
        }]))?,
        "protocol" => {
            serde_json::to_string_pretty(&json!({ "version": { "major": "1", "minor": "3" } }))?
        }
        _ => "{}".to_string(),
    };

    let resp = format!(
        "HTTP/1.1 200 OK\r\nContent-Type: application/json; charset=utf-8\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
        body.len(), body,
    );
    stream.write_all(resp.as_bytes())?;
    stream.flush()?;
    Ok(())
}

async fn cdp_processor(
    mut rx: mpsc::UnboundedReceiver<ServerMessage>,
    proxy: Option<String>,
    stealth: bool,
    user_agent: Option<String>,
    allow_file_access: bool,
    storage_dir: Option<std::path::PathBuf>,
    allow_private_network: bool,
    shutdown_flag: Arc<AtomicBool>,
    shutdown_notify: Arc<Notify>,
) {
    let mut ctx = CdpContext::new_full(
        proxy,
        stealth,
        user_agent,
        storage_dir,
        allow_file_access,
        allow_private_network,
    );
    let (itx, irx) = mpsc::unbounded_channel::<obscura_js::ops::InterceptedRequest>();
    ctx.intercept_tx = Some(itx);
    let mut intercept_rx: Option<mpsc::UnboundedReceiver<obscura_js::ops::InterceptedRequest>> = Some(irx);
    let mut intercepted_paused: HashMap<String, tokio::sync::oneshot::Sender<obscura_js::ops::InterceptResolution>> = HashMap::new();

    // Issue #19 follow-up: messages deferred from inside
    // `process_with_interception` because routing them through
    // `process_cdp_message → dispatch` while a nav was in flight would have
    // tripped V8's TryGetCurrent invariant. Drained at the top of each
    // outer iteration so they get processed sequentially with no other nav
    // in flight.
    let mut deferred: std::collections::VecDeque<ServerMessage> =
        std::collections::VecDeque::new();

    // Subscribe to Ctrl-C once. The future is single-shot, so we break out of
    // the outer loop when it fires and never poll it again. Without this the
    // accept loop just exits and any cookies set during the session are lost
    // before `BrowserContext::save_cookies()` runs.
    let mut shutdown = Box::pin(tokio::signal::ctrl_c());

    loop {
        // Drain any deferred messages from the previous interception window
        // before pulling new ones off the wire. Each is processed with no
        // nav-task spawn_local in flight, so dispatch's v8_lock can claim
        // the only Isolate cleanly.
        let msg = if let Some(d) = deferred.pop_front() {
            d
        } else {
            tokio::select! {
                msg = rx.recv() => match msg {
                    Some(m) => m,
                    None => break,
                },
                _ = &mut shutdown => {
                    tracing::info!("Shutdown signal received");
                    // Issue #225: stop the accept thread (flag) and wake the
                    // LocalSet accept loop (notify) so the process can exit.
                    // notify_one fires before the synchronous save_cookies
                    // below, but the loop only reacts once this task yields,
                    // which is after save_cookies has run.
                    shutdown_flag.store(true, Ordering::Relaxed);
                    shutdown_notify.notify_one();
                    break;
                }
            }
        };

        match msg {
            ServerMessage::NewConnection { reply_tx } => {
                let _ = reply_tx.send(
                    json!({"__init": true})
                        .to_string(),
                );
            }
            ServerMessage::Cdp(cdp_msg) => {
                // Route every Page.navigate through the spawn-and-defer path,
                // not just intercepted ones. Holding the V8 lock across a
                // multi-second navigate inside the regular dispatch wedges the
                // entire processor (40-site sweep: 39/40 timeouts). Spawning
                // navigation lets `cdp_processor` keep multiplexing other CDP
                // messages via the `process_with_interception` select loop;
                // unrelated requests get deferred only briefly and are drained
                // as soon as the nav settles.
                let is_navigation = cdp_msg.text.contains("Page.navigate");

                if is_navigation {
                    process_with_interception(
                        &cdp_msg.text, &mut ctx, &cdp_msg.reply_tx, &mut rx,
                        &mut intercept_rx, &mut intercepted_paused,
                        &mut deferred,
                    ).await;
                } else {
                    if cdp_msg.text.contains("Fetch.") {
                        handle_fetch_resolution(&cdp_msg.text, &mut ctx, &cdp_msg.reply_tx, &mut intercepted_paused);
                    }
                    process_cdp_message(&cdp_msg.text, &mut ctx, &cdp_msg.reply_tx).await;
                }
            }
        }

    }

    // Single exit point handles both Ctrl-C shutdown and the channel being
    // closed by the accept thread. Without this any cookies set during the
    // session are dropped on the floor.
    ctx.default_context.save_cookies();
}

fn handle_fetch_resolution(
    text: &str,
    _ctx: &mut CdpContext,
    reply_tx: &mpsc::UnboundedSender<String>,
    intercepted_paused: &mut HashMap<String, tokio::sync::oneshot::Sender<obscura_js::ops::InterceptResolution>>,
) {
    if let Ok(req) = serde_json::from_str::<CdpRequest>(text) {
        let method = req.method.as_str();
        let request_id = req.params.get("requestId").and_then(|v| v.as_str()).unwrap_or("");
        tracing::info!("INTERCEPTION resolution: {} for {}, paused_count={}", method, request_id, intercepted_paused.len());

        if let Some(resolver) = intercepted_paused.remove(request_id) {
            tracing::info!("INTERCEPTION resolved: {}", request_id);
            let resolution = match method {
                "Fetch.continueRequest" => obscura_js::ops::InterceptResolution::Continue {
                    url: None, method: None, headers: None, body: None,
                },
                "Fetch.fulfillRequest" => {
                    let status = req.params.get("responseCode").and_then(|v| v.as_u64()).unwrap_or(200) as u16;
                    let raw_body = req.params.get("body").and_then(|v| v.as_str()).unwrap_or("");
                    let body = decode_base64(raw_body);
                    let headers = req.params.get("responseHeaders")
                        .and_then(|v| v.as_array())
                        .map(|arr| arr.iter().filter_map(|h| {
                            Some((h.get("name")?.as_str()?.to_string(), h.get("value")?.as_str()?.to_string()))
                        }).collect())
                        .unwrap_or_default();
                    obscura_js::ops::InterceptResolution::Fulfill { status, headers, body }
                }
                "Fetch.failRequest" => {
                    let reason = req.params.get("errorReason").and_then(|v| v.as_str()).unwrap_or("Failed").to_string();
                    obscura_js::ops::InterceptResolution::Fail { reason }
                }
                _ => return,
            };
            let _ = resolver.send(resolution);
            let resp = crate::types::CdpResponse::success(req.id, json!({}), req.session_id);
            if let Ok(json) = serde_json::to_string(&resp) {
                let _ = reply_tx.send(json);
            }
        }
    }
}

async fn process_with_interception(
    text: &str,
    ctx: &mut CdpContext,
    reply_tx: &mpsc::UnboundedSender<String>,
    rx: &mut mpsc::UnboundedReceiver<ServerMessage>,
    intercept_rx: &mut Option<mpsc::UnboundedReceiver<obscura_js::ops::InterceptedRequest>>,
    intercepted_paused: &mut HashMap<String, tokio::sync::oneshot::Sender<obscura_js::ops::InterceptResolution>>,
    deferred: &mut std::collections::VecDeque<ServerMessage>,
) {
    let req: CdpRequest = match serde_json::from_str(text) {
        Ok(r) => r,
        Err(e) => {
            warn!("Invalid CDP: {}", e);
            return;
        }
    };

    tracing::info!("INTERCEPTION navigate: {} (id={})", req.method, req.id);

    let session_id = &req.session_id;
    let page_id = session_id
        .as_ref()
        .and_then(|sid| ctx.sessions.get(sid))
        .cloned();

    let page_id = match page_id {
        Some(id) => id,
        None => {
            process_cdp_message(text, ctx, reply_tx).await;
            return;
        }
    };

    let page_index = ctx.pages.iter().position(|p| p.id == page_id);
    let mut page = match page_index {
        Some(idx) => ctx.pages.remove(idx),
        None => {
            process_cdp_message(text, ctx, reply_tx).await;
            return;
        }
    };

    // Issue #19 follow-up: V8 only allows ONE entered Isolate per OS thread.
    // The regular dispatch path enforces this via `get_session_page_mut`
    // (which `suspend_js`'es every other page before letting the target
    // page run JS). The interception path here bypasses that — it removes
    // the target page and spawns a nav task — so we have to enforce the
    // same invariant explicitly. Otherwise nav-2's `init_js` constructs
    // Isolate-2 while page-1's Isolate-1 is still alive in ctx.pages, and
    // the next V8 scope unwind aborts the process via `Context::Exit`'s
    // `heap->isolate() == Isolate::TryGetCurrent()` check.
    for other in ctx.pages.iter_mut() {
        if other.has_js() {
            other.suspend_js();
        }
    }

    let url = req.params.get("url").and_then(|v| v.as_str()).unwrap_or("");
    let wait_until = crate::domains::page::parse_wait_until(&req.params);
    let nav_method = req.params.get("__method").and_then(|v| v.as_str()).unwrap_or("GET").to_string();
    let nav_body = req.params.get("__body").and_then(|v| v.as_str()).unwrap_or("").to_string();

    let preload_scripts: Vec<String> = ctx.preload_scripts.iter().map(|(_, s)| s.clone()).collect();

    if let Some(tx) = &ctx.intercept_tx {
        page.set_intercept_tx(tx.clone());
    }

    let session_for_events = req.session_id.clone();
    let frame_id = page.frame_id.clone();
    let loader_id = format!("loader-{}", uuid::Uuid::new_v4());

    let (nav_done_tx, mut nav_done_rx) = mpsc::channel::<(obscura_browser::Page, Result<(), String>)>(1);
    let url_owned = url.to_string();

    tokio::task::spawn_local(async move {
        // Issue #19: serialize V8 work across pages. The interception path
        // spawns navigation here while the parent task continues to pump
        // CDP messages via `dispatch` (which also acquires this lock); both
        // sides must coordinate or V8 aborts the process at concurrency >= 5.
        let _v8_guard = obscura_js::v8_lock::global().lock().await;
        // Preloads (addBinding shims, addScriptToEvaluateOnNewDocument sources)
        // must run BEFORE the page's own scripts (CDP contract). Hand them
        // to the page so navigate_single can inject them at the right point.
        page.set_preload_scripts(preload_scripts);
        let result = if nav_method == "POST" && !nav_body.is_empty() {
            page.navigate_with_wait_post(&url_owned, wait_until, &nav_method, &nav_body).await
        } else {
            page.navigate_with_wait(&url_owned, wait_until).await
        }
        .map_err(|e| e.to_string());
        drop(_v8_guard);
        let _ = nav_done_tx.send((page, result)).await;
    });

    let navigate_result: Result<(), String>;
    let page_back: Option<obscura_browser::Page>;

    // Issue #19 follow-up (PR #36 maintainer's fetch-intercept repro):
    // While the spawned nav task is executing V8 (potentially parked on
    // `op_fetch_url`'s `resolve_rx.await` *with Isolate-N still entered*),
    // we must NOT let the parent's `select!` route foreign Cdp messages
    // through `process_cdp_message → dispatch → page handlers`, because
    // those handlers call `get_session_page_mut` which `suspend_js`'es
    // OTHER pages (drops their `JsRuntime`, which calls
    // `JsRealmInner::destroy`). That trips V8's
    // `heap->isolate() == Isolate::TryGetCurrent()` invariant and aborts
    // the process via `V8_Fatal`.
    //
    // The `obscura_js::v8_lock` mutex doesn't save us here: it's a
    // `tokio::sync::Mutex` that is released around `.await`s inside V8
    // ops, so it doesn't actually keep the V8 enter/exit pair contiguous
    // on the thread.
    //
    // Park foreign Cdp messages into the outer deferred queue so the
    // outer `cdp_processor` loop processes them after this nav fully
    // completes (and its JsRuntime is no longer in flight on the
    // LocalSet).
    loop {
        let has_irx = intercept_rx.is_some();

        tokio::select! {
            Some((returned_page, result)) = nav_done_rx.recv() => {
                page_back = Some(returned_page);
                navigate_result = result;
                break;
            }
            Some(intercepted) = async {
                if let Some(ref mut irx) = intercept_rx {
                    irx.recv().await
                } else {
                    std::future::pending().await
                }
            }, if has_irx => {
                tracing::info!("INTERCEPTION: requestPaused for {} {} (sending to client)", intercepted.method, intercepted.url);
                let rws_event = json!({
                    "method": "Network.requestWillBeSent",
                    "params": {
                        "requestId": intercepted.request_id,
                        "loaderId": "",
                        "documentURL": "",
                        "request": {
                            "url": intercepted.url,
                            "method": intercepted.method,
                            "headers": intercepted.headers,
                            "initialPriority": "High",
                            "referrerPolicy": "strict-origin-when-cross-origin",
                        },
                        "timestamp": std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_secs_f64(),
                        "wallTime": std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_secs_f64(),
                        "initiator": {"type": "script"},
                        "type": intercepted.resource_type,
                        "frameId": frame_id,
                    },
                    "sessionId": session_for_events,
                });
                let _ = reply_tx.send(rws_event.to_string());

                let event_json = json!({
                    "method": "Fetch.requestPaused",
                    "params": {
                        "requestId": intercepted.request_id,
                        "request": {
                            "url": intercepted.url,
                            "method": intercepted.method,
                            "headers": intercepted.headers,
                            "initialPriority": "High",
                            "referrerPolicy": "strict-origin-when-cross-origin",
                        },
                        "frameId": frame_id,
                        "resourceType": intercepted.resource_type,
                        "networkId": intercepted.request_id,
                        "responseErrorReason": null,
                        "responseStatusCode": null,
                        "responseHeaders": null,
                    },
                    "sessionId": session_for_events,
                });
                let event_str = event_json.to_string();
                tracing::info!("INTERCEPTION event JSON: {}", &event_str[..event_str.len().min(300)]);
                let _ = reply_tx.send(event_str);
                intercepted_paused.insert(intercepted.request_id.clone(), intercepted.resolver);
                tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
            }
            Some(msg) = rx.recv() => {
                tracing::info!("INTERCEPTION select: received CDP message during navigation");
                match msg {
                    ServerMessage::NewConnection { reply_tx: new_tx } => {
                        // Safe: no V8 enter, just bookkeeping.
                        let pid = ctx.create_page();
                        let sid = format!("{}-session", pid);
                        ctx.sessions.insert(sid.clone(), pid.clone());
                        let _ = new_tx.send(json!({"__init": true, "pageId": pid, "sessionId": sid}).to_string());
                    }
                    ServerMessage::Cdp(msg) => {
                        if msg.text.contains("Fetch.continueRequest")
                            || msg.text.contains("Fetch.fulfillRequest")
                            || msg.text.contains("Fetch.failRequest")
                        {
                            // Safe: only flips a oneshot to resume the parked
                            // op inside the spawned nav task. No V8 enter on
                            // this side; the actual V8 work happens back on
                            // the nav task's thread.
                            handle_fetch_resolution(&msg.text, ctx, &msg.reply_tx, intercepted_paused);
                        } else {
                            // UNSAFE during nav: would route through dispatch,
                            // which can `suspend_js` other pages and trip the
                            // V8 invariant. Defer until nav completes —
                            // pushed to the outer `cdp_processor` queue so
                            // it's processed sequentially with no nav task
                            // in flight.
                            if deferred.len() >= MAX_DEFERRED_MESSAGES {
                                tracing::warn!("INTERCEPTION: deferred queue full ({}), returning error to client", MAX_DEFERRED_MESSAGES);
                                if let Ok(req) = serde_json::from_str::<CdpRequest>(&msg.text) {
                                    let resp = crate::types::CdpResponse::error(
                                        req.id,
                                        -32000,
                                        "Server busy: navigation in progress, try again later".to_string(),
                                        req.session_id,
                                    );
                                    if let Ok(json) = serde_json::to_string(&resp) {
                                        let _ = msg.reply_tx.send(json);
                                    }
                                }
                            } else {
                                tracing::info!("INTERCEPTION: deferring CDP message until nav completes");
                                deferred.push_back(ServerMessage::Cdp(msg));
                            }
                        }
                    }
                }
            }
        }
    }

    // Deferred messages are handled by the outer `cdp_processor` loop
    // (it drains `deferred` before pulling the next message off `rx`).

    let mut page = page_back.expect("navigation task should return the page");

    let network_events: Vec<_> = page.network_events.drain(..).collect();
    let page_url = page.url_string();
    let page_id_for_events = page.id.clone();
    let reached_network_idle = page.lifecycle.is_network_idle();

    ctx.pages.push(page);

    let response = match navigate_result {
        Ok(()) => crate::types::CdpResponse::success(
            req.id,
            json!({"frameId": frame_id, "loaderId": loader_id}),
            req.session_id.clone(),
        ),
        Err(e) => crate::types::CdpResponse::error(req.id, -32000, e, req.session_id.clone()),
    };

    if let Ok(json) = serde_json::to_string(&response) {
        let _ = reply_tx.send(json);
    }

    // Shared event emission: includes the post-#190 Network.requestWillBeSent
    // -before-frameNavigated ordering, the #189 requestId=loaderId trick that
    // makes `page.goto()` resolve to a Response, and the #192 per-isolated-
    // world fresh context ids. Pushes to `ctx.pending_events`; we then drain
    // to the WS reply channel.
    crate::domains::page::emit_navigation_events(
        ctx,
        &session_for_events,
        &frame_id,
        &loader_id,
        &page_url,
        &page_id_for_events,
        &network_events,
        wait_until,
        reached_network_idle,
    );
    for event in ctx.pending_events.drain(..) {
        if let Ok(json) = serde_json::to_string(&event) {
            let _ = reply_tx.send(json);
        }
    }
}

async fn process_cdp_message(
    text: &str,
    ctx: &mut CdpContext,
    reply_tx: &mpsc::UnboundedSender<String>,
) {
    let req: CdpRequest = match serde_json::from_str(text) {
        Ok(r) => r,
        Err(e) => {
            warn!("Invalid CDP: {}: {}", e, &text[..text.len().min(200)]);
            return;
        }
    };

    tracing::debug!("CDP: {} (id={}, s={:?})", req.method, req.id, req.session_id);

    let response = dispatch::dispatch(&req, ctx).await;

    // Chromium CDP semantics: events emitted as a side-effect of a command
    // (e.g. Target.targetCreated + Target.attachedToTarget from
    // Target.createTarget) MUST arrive BEFORE the command's response.
    // Playwright awaits the response and immediately reads state wired up
    // by those events; if the response lands first, accessing
    // Target._page errors with "Cannot read properties of undefined".
    for event in ctx.pending_events.drain(..) {
        if let Ok(json) = serde_json::to_string(&event) {
            let _ = reply_tx.send(json);
        }
    }

    if let Ok(json) = serde_json::to_string(&response) {
        let _ = reply_tx.send(json);
    }

    if let Some((nav_url, nav_method, nav_body)) = check_pending_navigation(ctx, &req.session_id) {
        tracing::info!("JS-triggered nav: {} {} (body: {} bytes)", nav_method, nav_url, nav_body.len());
        let nav_req = CdpRequest {
            id: 0,
            method: "Page.navigate".to_string(),
            params: json!({"url": nav_url, "__method": nav_method, "__body": nav_body}),
            session_id: req.session_id.clone(),
        };
        let _ = dispatch::dispatch(&nav_req, ctx).await;
        for event in ctx.pending_events.drain(..) {
            if let Ok(json) = serde_json::to_string(&event) {
                let _ = reply_tx.send(json);
            }
        }
    }
}

fn decode_base64(input: &str) -> String {
    fn val(c: u8) -> Option<u8> {
        match c {
            b'A'..=b'Z' => Some(c - b'A'),
            b'a'..=b'z' => Some(c - b'a' + 26),
            b'0'..=b'9' => Some(c - b'0' + 52),
            b'+' => Some(62),
            b'/' => Some(63),
            _ => None,
        }
    }
    let bytes: Vec<u8> = input.bytes().filter_map(val).collect();
    let mut out = Vec::with_capacity(bytes.len() * 3 / 4);
    for chunk in bytes.chunks(4) {
        let b = [
            chunk.first().copied().unwrap_or(0),
            chunk.get(1).copied().unwrap_or(0),
            chunk.get(2).copied().unwrap_or(0),
            chunk.get(3).copied().unwrap_or(0),
        ];
        out.push((b[0] << 2) | (b[1] >> 4));
        if chunk.len() > 2 { out.push((b[1] << 4) | (b[2] >> 2)); }
        if chunk.len() > 3 { out.push((b[2] << 6) | b[3]); }
    }
    String::from_utf8_lossy(&out).to_string()
}

fn fast_path_response(text: &str) -> Option<String> {
    let req: CdpRequest = serde_json::from_str(text).ok()?;

    let result = match req.method.as_str() {
        "Network.enable" | "Network.setCacheDisabled" | "Network.setRequestInterception" |
        "Page.enable" | "Page.setLifecycleEventsEnabled" | "Page.setInterceptFileChooserDialog" |
        "Runtime.runIfWaitingForDebugger" | "Runtime.discardConsoleEntries" |
        "Performance.enable" | "Log.enable" | "Security.enable" |
        "Emulation.setDeviceMetricsOverride" | "Emulation.setTouchEmulationEnabled" |
        "CSS.enable" | "Accessibility.enable" | "ServiceWorker.enable" |
        "Inspector.enable" | "Debugger.enable" | "Profiler.enable" |
        "HeapProfiler.enable" | "Overlay.enable" | "Storage.enable" |
        "Target.setAutoAttach" => {
            Some(json!({}))
        }
        "Browser.getVersion" => {
            Some(json!({
                "protocolVersion": "1.3",
                "product": "Obscura/0.1.0",
                "revision": "0",
                "userAgent": "Obscura/0.1.0",
                "jsVersion": "V8",
            }))
        }
        "Browser.setDownloadBehavior" | "Browser.getWindowBounds" => {
            Some(json!({}))
        }
        // Critical: Puppeteer calls this as the *first* CDP command on connect
        // (`BrowserConnector._connectToCdpBrowser`). If another client or a long
        // `Page.navigate` / interception holds the single `cdp_processor` task,
        // queued Target commands starve and Puppeteer hits protocolTimeout on
        // `Target.getBrowserContexts`. Fast-path bypasses the queue — same payload
        // as `domains::target::handle` when default context id is `"default"`.
        "Target.getBrowserContexts" => {
            Some(json!({ "browserContextIds": ["default"] }))
        }
        _ => None,
    };

    if let Some(value) = result {
        let resp = crate::types::CdpResponse::success(req.id, value, req.session_id);
        serde_json::to_string(&resp).ok()
    } else {
        None
    }
}

fn check_pending_navigation(ctx: &CdpContext, session_id: &Option<String>) -> Option<(String, String, String)> {
    let page_id = session_id
        .as_ref()
        .and_then(|sid| ctx.sessions.get(sid))?;
    let page = ctx.pages.iter().find(|p| &p.id == page_id)?;
    page.take_pending_navigation()
}

async fn handle_connection_ws(
    stream: TcpStream,
    msg_tx: mpsc::UnboundedSender<ServerMessage>,
) -> anyhow::Result<()> {
    // tokio_tungstenite wraps the stream in a 128 KiB write BufWriter by
    // default. CDP traffic is many small (~100-byte) frames, and that buffer
    // adds extra latency per frame. write_buffer_size=0 makes every WS write
    // hit the socket directly. Combined with set_nodelay(true) above, gets
    // per-frame latency on localhost down toward ideal.
    use tokio_tungstenite::tungstenite::protocol::WebSocketConfig;
    let mut cfg = WebSocketConfig::default();
    cfg.write_buffer_size = 0;
    cfg.max_write_buffer_size = 1 << 20;
    let ws_stream = tokio_tungstenite::accept_async_with_config(stream, Some(cfg)).await?;
    info!("WebSocket connected");
    let (mut ws_sender, mut ws_receiver) = ws_stream.split();

    let (reply_tx, mut reply_rx) = mpsc::unbounded_channel::<String>();

    let _ = msg_tx.send(ServerMessage::NewConnection {
        reply_tx: reply_tx.clone(),
    });
    if let Some(init_msg) = reply_rx.recv().await {
        tracing::debug!("Connection init: {}", &init_msg[..init_msg.len().min(100)]);
    }

    let send_task = tokio::task::spawn_local(async move {
        while let Some(msg) = reply_rx.recv().await {
            if msg.contains("\"__init\"") {
                continue;
            }
            if ws_sender.send(Message::Text(msg.into())).await.is_err() {
                break;
            }
        }
    });

    while let Some(msg) = ws_receiver.next().await {
        let msg = match msg {
            Ok(m) => m,
            Err(e) => {
                warn!("WS read error: {}", e);
                break;
            }
        };

        match msg {
            Message::Text(text) => {
                if text.contains("\"Browser.close\"") {
                    if let Ok(req) = serde_json::from_str::<CdpRequest>(&text) {
                        let resp = crate::types::CdpResponse::success(req.id, json!({}), None);
                        if let Ok(json) = serde_json::to_string(&resp) {
                            let _ = reply_tx.send(json);
                        }
                    }
                    break;
                }

                if let Some(resp) = fast_path_response(&text) {
                    let _ = reply_tx.send(resp);
                } else {
                    let _ = msg_tx.send(ServerMessage::Cdp(CdpMessage {
                        text: text.to_string(),
                        reply_tx: reply_tx.clone(),
                    }));
                }
            }
            Message::Close(_) => {
                info!("WS closed by client");
                break;
            }
            _ => {}
        }
    }

    send_task.abort();
    Ok(())
}