//! 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(),
        docker_image_sources: Vec::new(),
        cluster_sources: Vec::new(),
        registry: None,
    }
}

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

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),
        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
}

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"));
    handle.abort();
}

#[tokio::test]
async fn token_gate_enforces_when_set() {
    let app = app_state(Some("s3cret"));
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(async move {
        axum::serve(listener, router(app)).await.unwrap();
    });
    let base = format!("http://{addr}/api/state");
    // No token → 401.
    assert_eq!(reqwest::get(&base).await.unwrap().status(), 401);
    // Wrong token → 401.
    assert_eq!(
        reqwest::get(format!("{base}?token=wrong"))
            .await
            .unwrap()
            .status(),
        401
    );
    // Right token (query form — what SSE clients use) → 200.
    assert_eq!(
        reqwest::get(format!("{base}?token=s3cret"))
            .await
            .unwrap()
            .status(),
        200
    );
}

#[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 {
        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 {
        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"),
    }
}

#[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;
    }
    let (addr, _h) = boot_with(playwright_config()).await;
    // tokio::process (not std::process::output — that would block this
    // runtime's only thread while the axum server runs on it → deadlock).
    let out = tokio::process::Command::new("deno")
        .arg("run")
        .arg("-A")
        .arg("src/web/ui.playwright.mjs")
        .env("SM121_WEB_URL", format!("http://{addr}"))
        .output()
        .await
        .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}"
    );
    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.
fn playwright_config() -> Config {
    let mut cfg = fake_config();
    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
}