//! SEAMLESS-UPDATES W9 (releases#305) — the node-local HTTP listener is a
//! BRAIN resource, so `spt daemon refresh` (and a brain-only apply) re-runs the
//! serving code with no daemon flip.
//!
//! The rig is field-shaped: a REAL `spt daemon run` broker + supervised brain
//! on a FIXED loopback port (the operator's shape, not the rig's ephemeral
//! posture — an ephemeral rebind lands on a new port and would hide the very
//! rebind this measures). The listener's OWNING PROCESS is read from the OS:
//! before the refresh it must be the brain, never the broker; after it, the
//! NEW brain, on the SAME port, with the broker pid unchanged. At base (serving
//! in the broker) the owner is the broker pid on both sides of the refresh, so
//! this is red at base by construction.
//!
//! The port the product REPORTS (`spt docs url` → seed control → DocsStatus)
//! must be the port the new generation announced
//! (REQ-WEB-BOUND-PORT-GENERATION-SCOPED).
//!
//! The rebind gap is MEASURED by a prober hammering the port across the
//! refresh with `Connection: close` requests (the server closes first, so
//! server-side TIME_WAIT is live on the port when the brain dies — the case
//! where SO_REUSEADDR semantics matter). Each sample is an INTERVAL, not an
//! instant — a refused connect on Windows loopback takes most of a second —
//! so the gap is reported as a bracket: LOWER = the down time the failed
//! samples prove, UPPER = last success START before the outage to first
//! success END after it. It is REPORTED, never gated on a budget: a product
//! constant raced under load is a false red.
//!
//! HEAVY at birth: it spawns a real `spt daemon run` tree and cycles its brain.
//
// [int->REQ-WEB-SERVING-IN-BRAIN]
// [int->REQ-WEB-BOUND-PORT-GENERATION-SCOPED]

use std::io::{Read as _, Write as _};
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

mod common;
use common::CommandNoWindowExt;

const MARKER: &str = "W9-SERVING-IN-BRAIN-MARKER";

/// Scoped pid-tree kill (cleanup) — never machine-wide.
fn kill_pid(pid: u32) {
    #[cfg(windows)]
    let _ = Command::new("taskkill")
        .no_window()
        .args(["/PID", &pid.to_string(), "/F", "/T"])
        .output();
    #[cfg(unix)]
    let _ = Command::new("kill").args(["-9", &pid.to_string()]).output();
}

/// `(pid, generation)` out of `brain.ready`; `None` until it parses.
fn read_ready(path: &Path) -> Option<(u32, u64)> {
    let v: serde_json::Value = serde_json::from_str(&std::fs::read_to_string(path).ok()?).ok()?;
    Some((v.get("pid")?.as_u64()? as u32, v.get("generation")?.as_u64()?))
}

/// Poll `brain.ready` until it holds a pid different from `was`.
fn wait_ready_not(path: &Path, was: Option<u32>, budget: Duration) -> Option<(u32, u64)> {
    let deadline = Instant::now() + budget;
    while Instant::now() < deadline {
        if let Some((pid, generation)) = read_ready(path) {
            if was != Some(pid) {
                return Some((pid, generation));
            }
        }
        std::thread::sleep(Duration::from_millis(50));
    }
    None
}

/// A loopback port free right now (bind 0, read, drop). The daemon is then
/// told to bind it as a FIXED port — the field shape.
fn free_port() -> u16 {
    let listener = std::net::TcpListener::bind(("127.0.0.1", 0)).expect("probe a free port");
    listener.local_addr().expect("local addr").port()
}

/// One `Connection: close` GET; `Some((status, body))` or `None` on any
/// connect/read failure (the prober's "down" sample).
fn try_get(port: u16, path: &str, timeout: Duration) -> Option<(u16, String)> {
    let address = std::net::SocketAddr::from(([127, 0, 0, 1], port));
    let mut stream = std::net::TcpStream::connect_timeout(&address, timeout).ok()?;
    stream.set_read_timeout(Some(timeout)).ok()?;
    stream.set_write_timeout(Some(timeout)).ok()?;
    let request = format!("GET {path} HTTP/1.1\r\nHost: localhost:{port}\r\nConnection: close\r\n\r\n");
    stream.write_all(request.as_bytes()).ok()?;
    let mut raw = Vec::new();
    stream.read_to_end(&mut raw).ok()?;
    let split = raw.windows(4).position(|w| w == b"\r\n\r\n")?;
    let headers = String::from_utf8_lossy(&raw[..split]).into_owned();
    let status = headers.split_whitespace().nth(1)?.parse().ok()?;
    Some((status, String::from_utf8_lossy(&raw[split + 4..]).into_owned()))
}

/// The pid that owns the LISTENING socket on `127.0.0.1:<port>`, read from
/// the OS — the one observation that tells brain-owned from broker-owned.
fn listener_owner(port: u16) -> Option<u32> {
    #[cfg(windows)]
    {
        let out = Command::new("netstat").no_window().args(["-ano", "-p", "TCP"]).output().ok()?;
        let text = String::from_utf8_lossy(&out.stdout).into_owned();
        let local = format!("127.0.0.1:{port}");
        text.lines().find_map(|line| {
            let cols: Vec<&str> = line.split_whitespace().collect();
            (cols.len() == 5 && cols[1] == local && cols[3] == "LISTENING")
                .then(|| cols[4].parse().ok())
                .flatten()
        })
    }
    #[cfg(unix)]
    {
        let out = Command::new("ss").args(["-ltnpH"]).output().ok()?;
        let text = String::from_utf8_lossy(&out.stdout).into_owned();
        let local = format!("127.0.0.1:{port}");
        text.lines()
            .find(|line| line.split_whitespace().nth(3) == Some(local.as_str()))
            .and_then(|line| line.split("pid=").nth(1))
            .and_then(|rest| rest.split(|c: char| !c.is_ascii_digit()).next())
            .and_then(|pid| pid.parse().ok())
    }
}

/// `spt docs url` — the product's reported URL (DocsStatus over seed control).
fn reported_docs_url(spt_bin: &Path, home: &Path) -> Option<String> {
    let mut cmd = Command::new(spt_bin);
    cmd.no_window().args(["docs", "url"]).env("SPT_HOME", home);
    let out = common::output_bounded(cmd, Duration::from_secs(20));
    out.status
        .success()
        .then(|| String::from_utf8_lossy(&out.stdout).trim().to_owned())
}

/// Poll the reported URL until it names `port` (or the budget ends); returns
/// the last answer seen, for the panel.
fn wait_reported_port(spt_bin: &Path, home: &Path, port: u16, budget: Duration) -> Option<String> {
    let deadline = Instant::now() + budget;
    let needle = format!("//localhost:{port}/");
    let mut last = None;
    while Instant::now() < deadline {
        last = reported_docs_url(spt_bin, home);
        if last.as_deref().is_some_and(|url| url.contains(&needle)) {
            return last;
        }
        std::thread::sleep(Duration::from_millis(200));
    }
    last
}

#[test]
fn daemon_refresh_rebinds_the_http_listener_in_the_new_brain() {
    // ── (1) Isolated, NET-LESS SPT_HOME with a landed docs page. ──
    let home = tempfile::tempdir().unwrap();
    std::env::set_var("SPT_HOME", home.path());
    let identity_dir = home.path().join("identity");
    std::fs::create_dir_all(&identity_dir).unwrap();
    std::fs::write(identity_dir.join("node.key"), "not-a-valid-seed").unwrap();
    let docs = home.path().join("docs");
    std::fs::create_dir_all(&docs).unwrap();
    std::fs::write(docs.join("index.html"), format!("<html>{MARKER}</html>")).unwrap();

    let spt_bin = PathBuf::from(env!("CARGO_BIN_EXE_spt"));
    let port = free_port();

    // ── (2) The REAL daemon, on a FIXED docs port. The rig's ephemeral
    //    posture is deliberately NOT set: it would put the new brain on a new
    //    port and hide the rebind. The port was free a moment ago and is this
    //    tree's alone; a collision shows as DOCS_SERVER_BIND_FAIL in the panel.
    let daemon_log = home.path().join("daemon.stderr.log");
    let log_file = std::fs::File::create(&daemon_log).expect("create daemon stderr log");
    let mut broker: Child = Command::new(&spt_bin)
        .no_window()
        .args(["daemon", "run"])
        .env("SPT_HOME", home.path())
        .env(spt_daemon::docshost::DOCS_PORT_ENV, port.to_string())
        .env_remove(spt_daemon::docshost::TEST_EPHEMERAL_ADVISORY_PORTS_ENV)
        .stdout(Stdio::null())
        .stderr(Stdio::from(log_file))
        .spawn()
        .expect("spawn spt daemon run (broker process)");
    let broker_pid = broker.id();
    let ready_path = home.path().join("brain.ready");
    let panel = || common::daemon_stderr_panel(&daemon_log);

    let before = wait_ready_not(&ready_path, None, Duration::from_secs(60));
    let Some((brain_before, gen_before)) = before else {
        let _ = broker.kill();
        let _ = broker.wait();
        panic!("PRECONDITION: brain never came up.\n{}", panel());
    };

    // ── (3) BEFORE: the product reports this port, the page serves, and the
    //    OS says the BRAIN owns the listener. ──
    let url_before = wait_reported_port(&spt_bin, home.path(), port, Duration::from_secs(30));
    let path = url_before
        .as_deref()
        .and_then(|url| url.split_once(&format!("localhost:{port}")).map(|(_, p)| p.to_owned()));
    let page_before = path.as_deref().and_then(|p| try_get(port, p, Duration::from_secs(10)));
    let owner_before = listener_owner(port);

    // ── (4) The prober: one sample every ~5 ms across the refresh. ──
    let probe_path = path.clone().unwrap_or_else(|| "/".to_owned());
    let stop = Arc::new(AtomicBool::new(false));
    // (attempt start, attempt end, served 200)
    let samples: Arc<Mutex<Vec<(Instant, Instant, bool)>>> = Arc::new(Mutex::new(Vec::new()));
    let prober = {
        let stop = Arc::clone(&stop);
        let samples = Arc::clone(&samples);
        std::thread::spawn(move || {
            while !stop.load(Ordering::Relaxed) {
                let started = Instant::now();
                let ok = try_get(port, &probe_path, Duration::from_millis(500))
                    .is_some_and(|(status, _)| status == 200);
                samples.lock().unwrap().push((started, Instant::now(), ok));
                std::thread::sleep(Duration::from_millis(5));
            }
        })
    };
    std::thread::sleep(Duration::from_millis(300));

    // ── (5) THE VERB. ──
    let refresh_sent = Instant::now();
    let refresh = {
        let mut cmd = Command::new(&spt_bin);
        cmd.no_window().args(["daemon", "refresh"]).env("SPT_HOME", home.path());
        common::output_bounded(cmd, Duration::from_secs(30))
    };
    let cycled = wait_ready_not(&ready_path, Some(brain_before), Duration::from_secs(60));
    let url_after = wait_reported_port(&spt_bin, home.path(), port, Duration::from_secs(30));
    // Let the prober see the new listener settle before stopping it.
    let settle = Instant::now() + Duration::from_secs(10);
    while Instant::now() < settle {
        let last_ok = samples.lock().unwrap().last().is_some_and(|(s, _, ok)| *ok && *s > refresh_sent);
        if last_ok && cycled.is_some() {
            break;
        }
        std::thread::sleep(Duration::from_millis(50));
    }
    std::thread::sleep(Duration::from_millis(300));
    stop.store(true, Ordering::Relaxed);
    let _ = prober.join();

    // ── (6) AFTER. ──
    let page_after = path.as_deref().and_then(|p| try_get(port, p, Duration::from_secs(10)));
    let owner_after = listener_owner(port);
    let broker_alive = broker.try_wait().map(|st| st.is_none()).unwrap_or(false);

    // The rebind gap, as a bracket over the one outage the refresh causes:
    // the failed samples after the last pre-outage success. LOWER: the
    // listener was closed at some moment inside the first AND the last failed
    // interval, so it was down at least last.start - first.end. UPPER: it was
    // serving at some moment inside the bracketing successes, so it was down
    // at most next_ok.end - prev_ok.start.
    let samples = samples.lock().unwrap().clone();
    let pre_refresh_failures = samples.iter().filter(|(_, end, ok)| *end < refresh_sent && !ok).count();
    let first_fail = samples.iter().position(|(start, _, ok)| *start >= refresh_sent && !ok);
    let (gap_lower, gap_upper, failed_samples, never_recovered) = match first_fail {
        None => (Duration::ZERO, Duration::ZERO, 0, false),
        Some(first) => {
            let recovered = samples[first..].iter().position(|(_, _, ok)| *ok).map(|i| first + i);
            let last_fail = recovered.map_or(samples.len() - 1, |r| r - 1);
            let lower = samples[last_fail].0.saturating_duration_since(samples[first].1);
            let upper = match (first.checked_sub(1), recovered) {
                (Some(prev), Some(next)) => samples[next].1 - samples[prev].0,
                _ => Duration::MAX,
            };
            (lower, upper, last_fail + 1 - first, recovered.is_none())
        }
    };
    let refresh_stdout = String::from_utf8_lossy(&refresh.stdout).into_owned();
    let refresh_stderr = String::from_utf8_lossy(&refresh.stderr).into_owned();
    let daemon_panel = panel();
    eprintln!(
        "=== W9 serving-in-brain: os={} port={port} broker_pid={broker_pid} \
         brain_before={brain_before}/g{gen_before} cycled={cycled:?} \
         owner_before={owner_before:?} owner_after={owner_after:?} broker_alive={broker_alive} \
         url_before={url_before:?} url_after={url_after:?} \
         samples={} pre_refresh_failures={pre_refresh_failures} failed_samples={failed_samples} \
         REBIND_GAP_MS=[{}, {}] never_recovered={never_recovered} ===\n\
         --- refresh stdout ---\n{refresh_stdout}\n--- refresh stderr ---\n{refresh_stderr}\n{daemon_panel}",
        std::env::consts::OS,
        samples.len(),
        gap_lower.as_millis(),
        gap_upper.as_millis(),
    );

    // ── (7) Reap the tree SCOPED before asserting. ──
    let _ = {
        let mut cmd = Command::new(&spt_bin);
        cmd.no_window().args(["daemon", "stop", "--force"]).env("SPT_HOME", home.path());
        common::output_bounded(cmd, Duration::from_secs(20))
    };
    if let Some((pid, _)) = cycled {
        kill_pid(pid);
    }
    kill_pid(brain_before);
    let _ = broker.kill();
    let _ = broker.wait();

    // ── ASSERTIONS ──
    assert!(
        page_before.as_ref().is_some_and(|(status, body)| *status == 200 && body.contains(MARKER)),
        "PRECONDITION: the landed docs page must serve before the refresh \
         (reported {url_before:?}, got {page_before:?}).\n{daemon_panel}"
    );
    assert_eq!(
        owner_before,
        Some(brain_before),
        "REQ-WEB-SERVING-IN-BRAIN: the loopback listener must be owned by the BRAIN \
         (pid {brain_before}), not the broker (pid {broker_pid}).\n{daemon_panel}"
    );
    assert!(
        refresh.status.success(),
        "`spt daemon refresh` must exit 0 (stdout: {refresh_stdout}; stderr: {refresh_stderr}).\n{daemon_panel}"
    );
    let (brain_after, gen_after) = cycled.unwrap_or_else(|| {
        panic!("PRECONDITION: the brain must cycle on refresh (pid stayed {brain_before}).\n{daemon_panel}")
    });
    assert!(gen_after > gen_before, "the brain generation must advance");
    assert!(
        broker_alive,
        "the broker (pid {broker_pid}) must survive the refresh — no daemon flip.\n{daemon_panel}"
    );
    assert_eq!(
        owner_after,
        Some(brain_after),
        "REQ-WEB-SERVING-IN-BRAIN: after refresh the SAME port must be owned by the NEW \
         brain (pid {brain_after}) — the serving code re-ran from the respawned image.\n{daemon_panel}"
    );
    assert!(
        url_after.as_deref().is_some_and(|url| url.contains(&format!("//localhost:{port}/"))),
        "REQ-WEB-BOUND-PORT-GENERATION-SCOPED: DocsStatus must report the port the new \
         generation announced (got {url_after:?}).\n{daemon_panel}"
    );
    assert!(
        page_after.as_ref().is_some_and(|(status, body)| *status == 200 && body.contains(MARKER)),
        "the page must serve from the new brain (got {page_after:?}).\n{daemon_panel}"
    );
    assert!(
        !never_recovered,
        "the prober never saw the port serve again after the refresh.\n{daemon_panel}"
    );
}