//! Web module tests: bus semantics, token gate, and a full server boot on an
//! ephemeral port with a fake (unreachable) node.

use crate::config::{Cluster, Config, Node};
use crate::web::events::{Event, build_snapshot};
use crate::web::{AppState, router};
use std::collections::HashMap;
use std::net::SocketAddr;

pub(crate) async fn call(addr: SocketAddr, path: &str) -> reqwest::Response {
    reqwest::Client::new()
        .get(format!("http://{addr}{path}"))
        .send()
        .await
        .unwrap()
}

/// Config with one node at 127.0.0.1:2222 — connection refused instantly,
/// machine-independent (no real SSH, no hardware).
pub(crate) fn fake_config() -> Config {
    Config {
        ssh_user: "u".into(),
        ssh_key_path: "/nonexistent/key".into(),
        clusters: vec![Cluster {
            name: "fake".into(),
            master_ib_ip: None,
            nodes: vec![Node {
                name: "n0".into(),
                ssh_ip: "127.0.0.1:2222".into(),
                ib_ip: None,
                rank: 0,
                host_key: None,
                entity: None,
            }],
            extra_entities: vec![],
        }],
        nccl: HashMap::new(),
        hf_cache_dir: None,
        init_port: 25000,
        before_serve: HashMap::new(),
        exec_aliases: HashMap::new(),
        home_assistant: None,
        mikrotik: None,
        global_mods_dir: None,
        drop_caches: true,
        mount_aliases: HashMap::new(),
        recipe_sources: Vec::new(),
        recipe_repos: Vec::new(),
        docker_image_sources: Vec::new(),
        cluster_sources: Vec::new(),
        registry: None,
    }
}

#[path = "tests_auth.rs"]
mod auth;

pub(crate) fn app_state(token: Option<&str>) -> AppState {
    app_state_with(fake_config(), token)
}

pub(crate) fn app_state_with(config: Config, token: Option<&str>) -> AppState {
    let (tx, _) = tokio::sync::broadcast::channel(16);
    AppState {
        config: std::sync::Arc::new(config),
        bus: tx,
        token: token.map(String::from),
        sessions: std::sync::Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())),
        last: std::sync::Arc::new(std::sync::Mutex::new(None)),
        jobs: std::sync::Arc::new(std::sync::Mutex::new(Vec::new())),
        log_attaches: std::sync::Arc::new(std::sync::Mutex::new(Vec::new())),
    }
}

pub(crate) async fn boot() -> (SocketAddr, std::sync::Arc<tokio::task::JoinHandle<()>>) {
    boot_with(fake_config()).await
}

pub(crate) async fn boot_with(
    config: Config,
) -> (SocketAddr, std::sync::Arc<tokio::task::JoinHandle<()>>) {
    boot_app(app_state_with(config, None)).await
}

async fn boot_app(app: AppState) -> (SocketAddr, std::sync::Arc<tokio::task::JoinHandle<()>>) {
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let handle = tokio::spawn(async move {
        axum::serve(listener, router(app)).await.unwrap();
    });
    (addr, std::sync::Arc::new(handle))
}

#[tokio::test]
async fn api_state_reports_unreachable_node() {
    let (addr, handle) = boot().await;
    let body: serde_json::Value = reqwest::get(format!("http://{addr}/api/state"))
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert_eq!(body["clusters"].as_array().unwrap().len(), 1);
    assert_eq!(body["clusters"][0]["name"], "fake");
    assert_eq!(body["clusters"][0]["nodes"][0]["reachable"], false);
    assert_eq!(body["clusters"][0]["free"], 0);
    handle.abort();
}

/// The recipes panel is mounted under /recipes: the page, its own SSE
/// stream and the /r/ viewer all live behind the prefix.
#[tokio::test]
async fn recipes_panel_is_mounted_under_prefix() {
    let (addr, handle) = boot().await;
    let page = reqwest::get(format!("http://{addr}/recipes"))
        .await
        .unwrap()
        .text()
        .await
        .unwrap();
    // its own live stream resolves behind the prefix (headers arrive at once)
    let sse = reqwest::get(format!("http://{addr}/recipes/events"))
        .await
        .unwrap();
    assert_eq!(sse.status(), 200);
    assert!(
        page.contains("auth(\"/recipes/events\")"),
        "prefix not applied"
    );
    assert!(!page.contains("auth(\"/events\")"));
    // the dashboard's own routes still resolve
    let ui = reqwest::get(format!("http://{addr}/"))
        .await
        .unwrap()
        .text()
        .await
        .unwrap();
    assert!(ui.contains("recwin") || ui.contains("Recipes"));
    // Recipe logs open a dashboard WINDOW (postMessage bridge), not an inline
    // stream: the report page posts {sm121:"openLogs"}, the dashboard listens
    // and opens its logs window. No inline /api/logs fetch in the report page.
    assert!(
        page.contains("sm121: \"openLogs\"") && !page.contains("fetch(\"/api/logs\""),
        "report page should postMessage for logs, not stream inline"
    );
    assert!(
        ui.contains("openLogsForRecipe") && ui.contains("sm121 === \"openLogs\""),
        "dashboard should handle the recipes iframe logs message"
    );
    handle.abort();
}

#[test]
fn jobs_persist_and_mark_interrupted() {
    use crate::web::jobs;
    let app = app_state(None);
    let dir = std::env::temp_dir().join(format!("sm121-jobs-test-{}", std::process::id()));
    std::fs::create_dir_all(&dir).unwrap();
    let path = dir.join("jobs.json");
    // a registry snapshot with a running job: the process died mid-flight
    std::fs::write(
        &path,
        r#"[{"id":"j9","kind":"serve","cluster":"c","recipe":"r","status":"running","message":"started","started":1},
            {"id":"j8","kind":"power","cluster":"c","recipe":null,"status":"ok","message":"on","started":2}]"#,
    )
    .unwrap();
    jobs::load_from(Some(&path), &app);
    let jobs = app.jobs.lock().unwrap().clone();
    std::fs::remove_dir_all(&dir).unwrap();
    assert_eq!(jobs.len(), 2);
    let running = jobs.iter().find(|j| j.id == "j9").unwrap();
    assert_eq!(
        running.status,
        jobs::JobStatus::Failed,
        "running -> failed on boot"
    );
    assert!(
        running.message.contains("runlock"),
        "the ledger hint: {}",
        running.message
    );
    let done = jobs.iter().find(|j| j.id == "j8").unwrap();
    assert_eq!(done.status, jobs::JobStatus::Ok);
}

#[test]
fn stale_from_snapshot_picks_unreachable_nodes() {
    let app = app_state(None);
    use crate::web::api::stale_from_snapshot;
    let node = |name: &str, ip: &str, reachable: bool| crate::web::events::NodeSnap {
        name: name.into(),
        ssh_ip: ip.into(),
        reachable,
        occupied: false,
        containers: Vec::new(),
        rank: 0,
        ha: None,
    };
    let snap = crate::web::events::Snapshot {
        version: String::new(),
        ts: 1,
        clusters: vec![crate::web::events::ClusterSnap {
            name: "c".into(),
            up: false,
            switch_off: false,
            soft_off: Vec::new(),
            free: 0,
            nodes: vec![node("a", "1.1.1.1", true), node("b", "1.1.1.2", false)],
        }],
        jobs: Vec::new(),
        topo: Default::default(),
        attaches: Vec::new(),
    };
    *app.last.lock().unwrap() = Some(snap);
    assert_eq!(stale_from_snapshot(&app, "c"), vec!["b".to_string()]);
    assert!(stale_from_snapshot(&app, "other").is_empty());
}

#[tokio::test]
async fn token_gate_covers_recipes_panel() {
    let app = app_state(Some("s3cret"));
    *app.last.lock().unwrap() = Some(crate::web::events::Snapshot {
        version: String::new(),
        ts: 1,
        clusters: Vec::new(),
        jobs: Vec::new(),
        topo: Default::default(),
        attaches: Vec::new(),
    });
    let (addr, handle) = boot_app(app).await;
    let base = format!("http://{addr}");
    // bare requests are rejected on every panel route...
    for path in ["/recipes", "/recipes/events"] {
        let st = reqwest::get(format!("{base}{path}"))
            .await
            .unwrap()
            .status();
        assert_eq!(st, 401, "{path} must be gated");
    }
    // ...and the session cookie (issued by POST /api/login) opens them —
    // the iframe and the report page's SSE ride the cookie, not the token.
    let client = reqwest::Client::new();
    let res = client
        .post(format!("{base}/api/login"))
        .header("authorization", "Bearer s3cret")
        .send()
        .await
        .unwrap();
    assert_eq!(res.status(), 204);
    let cookie = res
        .headers()
        .get("set-cookie")
        .unwrap()
        .to_str()
        .unwrap()
        .split(';')
        .next()
        .unwrap()
        .to_string();
    let page = client
        .get(format!("{base}/recipes"))
        .header("cookie", &cookie)
        .send()
        .await
        .unwrap();
    assert_eq!(page.status(), 200, "the recipes page opens with the cookie");
    let sse = client
        .get(format!("{base}/recipes/events"))
        .header("cookie", &cookie)
        .send()
        .await
        .unwrap();
    assert_eq!(sse.status(), 200);
    drop(sse);
    handle.abort();
}

#[tokio::test]
async fn sse_replays_cached_snapshot_on_connect() {
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    // Nothing is broadcast after this point: the frame must come from the
    // cached snapshot, otherwise a fresh client waits a whole poll tick.
    let app = app_state(None);
    *app.last.lock().unwrap() = Some(crate::web::events::Snapshot {
        version: String::new(),
        ts: 1,
        clusters: Vec::new(),
        jobs: Vec::new(),
        topo: Default::default(),
        attaches: Vec::new(),
    });
    let (addr, handle) = boot_app(app).await;
    let mut sock = tokio::net::TcpStream::connect(addr).await.unwrap();
    sock.write_all(b"GET /events HTTP/1.1\r\nHost: x\r\n\r\n")
        .await
        .unwrap();
    let mut buf = vec![0u8; 4096];
    let n = sock.read(&mut buf).await.unwrap();
    let text = String::from_utf8_lossy(&buf[..n]);
    assert!(
        text.contains("\"type\":\"state\""),
        "no replayed frame: {text}"
    );
    handle.abort();
}

#[tokio::test]
async fn bus_delivers_state_events() {
    let (tx, mut rx) = tokio::sync::broadcast::channel::<Event>(8);
    let snap = crate::web::events::Snapshot {
        version: String::new(),
        ts: 1,
        clusters: Vec::new(),
        jobs: Vec::new(),
        topo: Default::default(),
        attaches: Vec::new(),
    };
    tx.send(Event::State {
        ts: snap.ts,
        snapshot: snap.clone(),
    })
    .unwrap();
    let got = rx.recv().await.unwrap();
    match got {
        Event::State { ts, snapshot } => {
            assert_eq!(ts, 1);
            assert!(snapshot.clusters.is_empty());
        }
        Event::Job { .. } => panic!("unexpected job event"),
        Event::Log { .. } => panic!("unexpected log event"),
    }
}

/// The dashboard header carries a small version badge under the title; the
/// version itself is baked in at image build (`SM121_VERSION`) and reaches the
/// page through the state snapshot, so a plain `cargo build` shows `dev`.
#[test]
fn header_shows_the_build_version() {
    use crate::web::events::build_snapshot;
    let snap = build_snapshot(
        &fake_config(),
        &crate::web::probe::Fleet {
            nodes: std::collections::HashMap::new(),
            switch_off: std::collections::HashMap::new(),
            soft_off: std::collections::HashMap::new(),
            ha: std::collections::HashMap::new(),
        },
        &[],
        &[],
        &std::collections::HashMap::new(),
    );
    assert_eq!(snap.version, crate::build_info::version());
    let json = serde_json::to_string(&snap).unwrap();
    assert!(
        json.contains("\"version\""),
        "state JSON has no version: {json}"
    );
    let page = super::UI_HTML;
    for needle in ["id=\"ver\"", "header .ver", "v\" + snap.version"] {
        assert!(page.contains(needle), "dashboard page missing {needle}");
    }
}

#[tokio::test]
async fn snapshot_serializes_with_type_tag() {
    let ev = Event::State {
        ts: 7,
        snapshot: build_snapshot(
            &fake_config(),
            &crate::web::probe::Fleet {
                nodes: std::collections::HashMap::new(),
                switch_off: std::collections::HashMap::new(),
                soft_off: std::collections::HashMap::new(),
                ha: std::collections::HashMap::new(),
            },
            &[],
            &[],
            &std::collections::HashMap::new(),
        ),
    };
    let json = serde_json::to_value(&ev).unwrap();
    assert_eq!(json["type"], "state");
    assert_eq!(json["ts"], 7);
    assert!(json["snapshot"].is_object());
}

// (P2/P3/P4 API round-trip tests moved to api_tests.rs — 500-line cap)

/// Click-through UI tests: boots the real server on an ephemeral port and
/// drives the embedded ui.html with Playwright (deno, npm-free).
/// Skipped when deno or the chromium binary is unavailable.
#[tokio::test]
async fn ui_playwright_checks() {
    if std::process::Command::new("deno")
        .arg("--version")
        .output()
        .is_err()
    {
        eprintln!("SKIP ui_playwright_checks: deno not available");
        return;
    }
    // The recipes panel needs at least one row: a fixture recipe source
    // (the launcher repo no longer carries a local `recipes/`).
    let fx = std::env::temp_dir().join(format!("sm121-ui-fixture-{}", std::process::id()));
    std::fs::create_dir_all(fx.join("ztest-ui-fixture")).unwrap();
    std::fs::write(
        fx.join("ztest-ui-fixture/recipe.toml"),
        "name = \"ztest-ui-fixture\"\nengine = \"vllm\"\n",
    )
    .unwrap();
    let (addr, _h) = boot_with(playwright_config(Some(&fx))).await;
    // Serialize with the report viewer's Playwright suite (shared browser).
    // The lock + deno run live in spawn_blocking: the std guard must not
    // be held across an await, and std::process::output must not block
    // this runtime's only thread (the axum server runs on it).
    let url = format!("http://{addr}");
    let out = tokio::task::spawn_blocking(move || {
        let _pw = crate::ui_test_lock::playwright_session();
        std::process::Command::new("deno")
            .arg("run")
            .arg("-A")
            .arg(concat!(
                env!("CARGO_MANIFEST_DIR"),
                "/src/web/ui.playwright.mjs"
            ))
            .env("SM121_WEB_URL", url)
            .output()
    })
    .await
    .unwrap()
    .unwrap();
    let stdout = String::from_utf8_lossy(&out.stdout);
    let stderr = String::from_utf8_lossy(&out.stderr);
    if stdout.contains("SKIP") {
        eprintln!("SKIP ui_playwright_checks:\n{stdout}");
        return;
    }
    assert!(
        out.status.success(),
        "playwright UI checks failed:\n{stdout}\n{stderr}"
    );
    let _ = std::fs::remove_dir_all(&fx);
    eprintln!("{stdout}");
}

/// Two offline clusters (both nodes on refused loopback ports) — the
/// fleet the Playwright UI checks render: chip strip with two chips,
/// down-cluster action bar, type-the-name power flow. `recipe_dir` (a dir
/// holding `<name>/recipe.toml` fixtures) feeds the recipes panel — the
/// launcher repo no longer carries a local `recipes/`, so the test seeds
/// its own.
fn playwright_config(recipe_dir: Option<&std::path::Path>) -> Config {
    let mut cfg = fake_config();
    if let Some(d) = recipe_dir {
        cfg.recipe_sources.push(d.display().to_string());
    }
    cfg.clusters.push(Cluster {
        name: "fake2".into(),
        master_ib_ip: None,
        nodes: vec![Node {
            name: "n1".into(),
            ssh_ip: "127.0.0.1:2223".into(),
            ib_ip: None,
            rank: 0,
            host_key: None,
            entity: None,
        }],
        extra_entities: vec![],
    });
    cfg
}