Pup - the AI's own browser

//! The event pump: one thread per connected browser reading every CDP event and keeping the
//! per-tab state the verbs report (console errors, exceptions, pending JS dialogs, URL/title
//! changes, closed targets). Node had this spread over page.on handlers per tab; here it is one
//! subscription on the browser socket, routed by the flattened sessionId.

use crate::TabState;
use pup_cdp::Cdp;
use pup_session::Sessions;
use serde_json::{json, Value};
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::Duration;

pub type TabStates = Arc<Mutex<HashMap<String, TabState>>>;

type Hook = Box<dyn Fn(&str, &str, &Value) + Send + Sync>;
static HOOK: Mutex<Option<Arc<Hook>>> = Mutex::new(None);
pub fn set_hook(h: Hook) { *HOOK.lock().unwrap() = Some(Arc::new(h)); }

const ERROR_CAP: usize = 300;

pub fn start_pump(profile: String, cdp: Cdp, sessions: Arc<Sessions>, states: TabStates) {
    let _ = cdp.send(None, "Target.setDiscoverTargets", json!({ "discover": true }), Duration::from_secs(5));
    let rx = cdp.subscribe(None, None);
    let pump_started = std::time::Instant::now();
    std::thread::Builder::new().name(format!("cdp-events-{profile}")).spawn(move || {
        while let Ok(ev) = rx.recv() {
            let sid = ev.session_id.clone().unwrap_or_default();
            match ev.method.as_str() {
                "Runtime.exceptionThrown" => {
                    let d = &ev.params["exceptionDetails"];
                    let text = d["exception"]["description"].as_str().or(d["text"].as_str()).unwrap_or("uncaught exception").to_string();
                    push_error(&states, &sid, json!({ "type": "exception", "text": text, "ts": pup_core::now_ms() }));
                }
                "Runtime.consoleAPICalled" if ev.params["type"] == json!("error") => {
                    let text = ev.params["args"].as_array().map(|a| a.iter().map(|x| x["value"].as_str().map(|s| s.to_string()).or_else(|| x["description"].as_str().map(|s| s.to_string())).unwrap_or_else(|| x["value"].to_string())).collect::<Vec<_>>().join(" ")).unwrap_or_default();
                    push_error(&states, &sid, json!({ "type": "console.error", "text": text, "ts": pup_core::now_ms() }));
                }
                "Page.javascriptDialogOpening" => {
                    let mut st = states.lock().unwrap();
                    st.entry(sid.clone()).or_default().dialog = Some(json!({ "type": ev.params["type"], "message": ev.params["message"], "defaultValue": ev.params["defaultPrompt"], "url": ev.params["url"], "since": pup_core::now_ms() }));
                    pup_core::log(format!("[dialog] {} dialog opened on session {}: {}", ev.params["type"].as_str().unwrap_or("?"), &sid[..sid.len().min(8)], ev.params["message"].as_str().unwrap_or("").chars().take(80).collect::<String>()));
                }
                "Runtime.bindingCalled" | "Page.screencastFrame" => { let h = HOOK.lock().unwrap().clone(); if let Some(h) = h { h(&ev.method, &sid, &ev.params); } }
                "Page.javascriptDialogClosed" => { states.lock().unwrap().entry(sid.clone()).or_default().dialog = None; }
                "Page.frameNavigated" if ev.params["frame"]["parentId"].is_null() => {
                    let url = ev.params["frame"]["url"].as_str().unwrap_or("").to_string();
                    states.lock().unwrap().entry(sid.clone()).or_default().url = Some(url.clone());
                    for s in sessions.all() { if s.tabs.iter().any(|t| t.cdp_session == sid) { sessions.with_mut(&s.id, |x| { if let Some(t) = x.tabs.iter_mut().find(|t| t.cdp_session == sid) { t.url = url.clone(); } }); } }
                }
                "Target.targetInfoChanged" => {
                    let info = &ev.params["targetInfo"];
                    let tid = info["targetId"].as_str().unwrap_or("").to_string();
                    let (url, title) = (info["url"].as_str().unwrap_or("").to_string(), info["title"].as_str().unwrap_or("").to_string());
                    for s in sessions.all() { if s.tabs.iter().any(|t| t.target_id == tid) { sessions.with_mut(&s.id, |x| { if let Some(t) = x.tabs.iter_mut().find(|t| t.target_id == tid) { t.url = url.clone(); t.title = title.clone(); } }); } }
                }
                // Chrome replays targetCreated for every EXISTING target right after setDiscoverTargets; those are
                // recovery's to sort out (2.0.454: the replay was adopted as "user-opened tabs" at boot and swallowed
                // the named sessions' targets). Only targets created after the pump has settled are adopted.
                "Target.targetCreated" if ev.params["targetInfo"]["type"] == json!("page") && pump_started.elapsed() > Duration::from_secs(5) => {
                    // 2.0.453: a tab the USER opened in a pup window (Ctrl+T, a link, Chrome restoring a discarded
                    // tab) becomes a tab of that window's session, keyed by the CDP Browser window id. Deferred
                    // 400 ms so pup's own births (which attach their target themselves) win the race.
                    let tid = ev.params["targetInfo"]["targetId"].as_str().unwrap_or("").to_string();
                    let url = ev.params["targetInfo"]["url"].as_str().unwrap_or("").to_string();
                    if !tid.is_empty() && !url.starts_with("chrome-extension://") && !url.starts_with("devtools://") {
                        let (cdp2, sessions2) = (cdp.clone(), sessions.clone());
                        std::thread::spawn(move || {
                            std::thread::sleep(Duration::from_millis(1500));   // 2.0.467: pup's own opens are tracked within milliseconds now; this margin covers a slow attach
                            if sessions2.all().iter().any(|s| s.tabs.iter().any(|t| t.target_id == tid)) { return; }
                            let Ok((wid, _)) = cdp2.window_for_target(&tid) else { return };
                            let Some(owner) = sessions2.all().into_iter().find(|s| s.window_id == Some(wid) && s.hwnd.is_some()) else { return };
                            // 2.0.473: don't adopt into a window that is mid-restore or just relaunched — the tab is
                            // pup's own reopen or Chrome auto-restoring the killed session's page, not a user tab.
                            if crate::recovery::restoring(&owner.id) || crate::recovery::recently_relaunched(&owner.id, 15_000) {
                                pup_core::log(format!("[tabs] \"{}\": ignored a target that appeared during its restore ({})", owner.id, &url[..url.len().min(80)]));
                                return;
                            }
                            // 2.0.478: never adopt a URL the window already has a tab for. Chrome spawns an extra
                            // page target for a page pup already opened (a prerender, a second target for the same
                            // URL), and adopting it added a duplicate tab that pup then could not drive — the whole
                            // window read 0-live after a restart (isolated single-restart, ~1 in 5). A genuine
                            // second tab of the same URL is rare and not worth that failure.
                            let base = |u: &str| u.split('?').next().unwrap_or("").trim_end_matches('/').to_string();
                            if !base(&url).is_empty() && owner.tabs.iter().any(|t| base(&t.url) == base(&url)) {
                                pup_core::log(format!("[tabs] \"{}\": ignored a duplicate target for a URL it already has ({})", owner.id, &url[..url.len().min(80)]));
                                return;
                            }
                            let mut s = owner.clone();
                            match crate::attach_tab(&mut s, &cdp2, &tid) {
                                Ok(tab_id) => { s.tabless_since_ms = 0; pup_core::log(format!("[tabs] \"{}\": adopted a tab opened outside pup as {tab_id} ({})", s.id, &url[..url.len().min(80)])); sessions2.insert(s.clone()); sessions2.save_file(&s); }
                                Err(e) => pup_core::log(format!("[tabs] \"{}\": could not adopt target {tid}: {}", s.id, e["error"].as_str().unwrap_or("?"))),
                            }
                        });
                    }
                }
                "Target.targetDestroyed" => {
                    let tid = ev.params["targetId"].as_str().unwrap_or("").to_string();
                    for s in sessions.all() {
                        if let Some(pos) = s.tabs.iter().position(|t| t.target_id == tid) {
                            let left = sessions.with_mut(&s.id, |x| { let t = x.tabs.remove(pos); states.lock().unwrap().remove(&t.cdp_session); if x.active_tab >= x.tabs.len() && !x.tabs.is_empty() { x.active_tab = x.tabs.len() - 1; } x.tabs.len() }).unwrap_or(0);
                            // 2.0.453: a destroyed LAST tab no longer kills the session on the spot. After a sleep, a Memory
                            // Saver discard or a renderer restart Chrome destroys and recreates a tab's target while the OS
                            // window stays on screen (John, 2026-09-10: the window was there, pup had dropped it, the
                            // toolbar pen did nothing). Keep the window session; the health tick re-attaches a page
                            // target in the same Browser window or removes the session once the OS window is gone.
                            if left == 0 { pup_core::log(format!("[health] session \"{}\": its last tab target was destroyed outside pup; keeping the window session while Chrome may replace it", s.id)); sessions.with_mut(&s.id, |x| x.tabless_since_ms = pup_core::now_ms()); }
                            if let Some(x) = sessions.get(&s.id) { sessions.save_file(&x); }
                        }
                    }
                }
                _ => {}
            }
        }
        pup_core::log(format!("[events] pump for profile \"{profile}\" ended (browser connection closed)"));
        crate::recovery::kick_health();   // 2.0.470: recover the windows now, not at the next 15 s tick
    }).ok();
}

fn push_error(states: &TabStates, sid: &str, e: Value) {
    let mut st = states.lock().unwrap();
    let t = st.entry(sid.to_string()).or_default();
    t.errors.push(e);
    if t.errors.len() > ERROR_CAP { let n = t.errors.len() - ERROR_CAP; t.errors.drain(0..n); }
}