//! The whole command line from a workstation (spec §8.6, spectra#21 and
//! #23): the daemon on one machine, the owner, their agents and every
//! command on another, with no ledger file on the workstation at all.

mod common;

use common::{add_agent, delegate, hermetic, run, text, Serving};
use serde_json::{json, Value};
use spectra_core::keys::KeyPair;
use std::path::Path;
use std::process::Stdio;
use std::time::Duration;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};

fn json_of(d: &Path, args: &[&str]) -> Value {
    let o = run(d, args);
    assert!(o.status.success(), "{args:?}: {}", text(&o));
    serde_json::from_slice(&o.stdout)
        .unwrap_or_else(|e| panic!("{args:?}: one JSON document ({e}): {}", String::from_utf8_lossy(&o.stdout)))
}

/// A server in `s`, claimed by the owner whose home is `c/owner`.
fn connected(s: &Path, c: &Path) -> (Serving, u16) {
    KeyPair::create_home(&c.join("owner"), "Andy McCall", "").unwrap();
    let owner = KeyPair::load_home(&c.join("owner")).unwrap();
    assert!(run(s, &["init", "--owner", &owner.key.to_hex(), "--name", "earth"]).status.success());
    let (srv, port) = Serving::start_tls(s);
    let url = format!("https://127.0.0.1:{port}");
    assert_eq!(run(c, &["connect", &url]).status.code(), Some(1));
    delegate(&c.join("owner"), &KeyPair::instance(&s.join("data"), "earth").unwrap());
    let o = run(c, &["connect", &url]);
    assert!(o.status.success(), "{}", text(&o));
    (srv, port)
}

#[tokio::test]
async fn every_command_works_from_another_machine() {
    let (sd, cd) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
    let (s, c) = (sd.path(), cd.path());
    let (srv, _port) = connected(s, c);

    std::fs::create_dir_all(c.join("oxygen")).unwrap();
    let o = run(c, &["project", "add", c.join("oxygen").to_str().unwrap()]);
    assert!(o.status.success(), "{}", text(&o));
    add_agent(c, "claude-oxygen", &[]);
    add_agent(c, "spare", &[]);

    let id = json_of(c, &["identity", "show", "--json"]);
    assert_eq!(id["instance"]["name"], "earth");
    assert_eq!(id["instance"]["delegated"], true, "{id}");
    let agents = json_of(c, &["agent", "list", "--json"]);
    let claude = agents.as_array().unwrap().iter().find(|a| a["name"] == "claude-oxygen").unwrap();
    assert_eq!(claude["state"], "ok", "{claude}");
    assert_eq!(claude["display"], "claude-oxygen · Andy McCall's agent");
    let projects = json_of(c, &["project", "list", "--json"]);
    assert_eq!(projects[0]["name"], "oxygen");
    assert!(projects[0]["agents"].as_array().unwrap().contains(&json!("claude-oxygen")));
    let o = run(c, &["agent", "disable", "spare"]);
    assert!(o.status.success(), "{}", text(&o));

    assert_eq!(json_of(c, &["say", "--json", "-c", "general", "the release is tomorrow"])["channel"], "#general");
    assert!(json_of(c, &["inbox", "--json"]).is_array());
    assert_eq!(json_of(c, &["search", "--json", "release"]).as_array().unwrap().len(), 1);
    let brief: Value = serde_json::from_slice(&run(c, &["api", "GET", "/api/v1/brief"]).stdout).unwrap();
    assert_eq!(brief["actor"], "human");

    // An agent's shim on the workstation talks to the server.
    let mut cmd = tokio::process::Command::from(hermetic(c));
    let home = c.join("agents/claude-oxygen");
    cmd.args(["mcp", "--agent", "claude-oxygen", "--home", home.to_str().unwrap()])
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .kill_on_drop(true);
    let mut shim = cmd.spawn().unwrap();
    let mut stdin = shim.stdin.take().unwrap();
    let mut lines = BufReader::new(shim.stdout.take().unwrap()).lines();
    for v in [
        json!({ "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": { "name": "t", "version": "1" } } }),
        json!({ "jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": { "name": "brief", "arguments": {} } }),
    ] {
        stdin.write_all(format!("{v}\n").as_bytes()).await.unwrap();
    }
    stdin.flush().await.unwrap();
    let mut brief = Value::Null;
    for _ in 0..2 {
        let line = tokio::time::timeout(Duration::from_secs(20), lines.next_line()).await.unwrap().unwrap().unwrap();
        let v: Value = serde_json::from_str(&line).unwrap();
        if v["id"] == 2 {
            brief = v;
        }
    }
    assert!(brief["result"]["content"][0]["text"].as_str().unwrap().contains("You are claude-oxygen · Andy McCall's agent"), "{brief}");
    drop(shim);

    // The ledger stays on the server: verify asks it; check and rebuild refuse here.
    let v = json_of(c, &["admin", "verify", "--json"]);
    assert!(v["findings"].as_array().unwrap().is_empty(), "{v}");
    let o = run(c, &["admin", "check"]);
    assert!(!o.status.success() && text(&o).contains("runs on the server"), "{}", text(&o));
    assert!(!c.join("data").exists(), "the workstation has no data directory");
    srv.stop();

    let o = run(s, &["admin", "verify"]);
    assert!(o.status.success(), "{}", text(&o));
}

/// One HTTP/1.1 exchange with the proxy: status, head, body.
fn http(port: u16, method: &str, path: &str, headers: &[(&str, &str)], body: &str) -> (u16, String, String) {
    use std::io::{Read, Write};
    let mut s = std::net::TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(10))).unwrap();
    let mut req = format!("{method} {path} HTTP/1.1\r\nconnection: close\r\ncontent-length: {}\r\n", body.len());
    if !headers.iter().any(|(k, _)| k.eq_ignore_ascii_case("host")) {
        req.push_str(&format!("host: 127.0.0.1:{port}\r\n"));
    }
    for (k, v) in headers {
        req.push_str(&format!("{k}: {v}\r\n"));
    }
    req.push_str("\r\n");
    req.push_str(body);
    s.write_all(req.as_bytes()).unwrap();
    let mut out = Vec::new();
    let _ = s.read_to_end(&mut out);
    let text = String::from_utf8_lossy(&out).into_owned();
    let (head, body) = text.split_once("\r\n\r\n").unwrap_or((&text, ""));
    let status = head.split_whitespace().nth(1).and_then(|c| c.parse().ok()).unwrap_or(0);
    (status, head.to_string(), body.to_string())
}

#[tokio::test]
async fn radar_through_the_loopback_proxy() {
    use std::io::BufRead;
    let (sd, cd) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
    let (s, c) = (sd.path(), cd.path());
    let (srv, _port) = connected(s, c);

    let mut open = hermetic(c).args(["open", "--no-browser", "--port", "0"]).stdout(Stdio::piped()).spawn().unwrap();
    let out = open.stdout.take().unwrap();
    let (tx, rx) = std::sync::mpsc::channel();
    std::thread::spawn(move || {
        for line in std::io::BufReader::new(out).lines().map_while(Result::ok) {
            if let Some(u) = line.split_whitespace().find(|w| w.contains("/signin?code=")) {
                let _ = tx.send(u.to_string());
            }
        }
    });
    let link = rx.recv_timeout(Duration::from_secs(10)).expect("spectra open printed a link");
    let rest = link.strip_prefix("http://127.0.0.1:").expect("a loopback link");
    let (port, path) = rest.split_once('/').unwrap();
    let port: u16 = port.parse().unwrap();

    // Signing in through the proxy sets the session cookie.
    let (status, head, _) = http(port, "GET", &format!("/{path}"), &[], "");
    assert_eq!(status, 303, "{head}");
    let cookie = head
        .lines()
        .find_map(|l| {
            l.to_ascii_lowercase()
                .starts_with("set-cookie:")
                .then(|| l.split_once(':').unwrap().1.trim().split(';').next().unwrap().to_string())
        })
        .expect("a session cookie");

    // Live and a channel load; the channel's form carries the CSRF token.
    let (status, _, live) = http(port, "GET", "/", &[("cookie", &cookie)], "");
    assert_eq!(status, 200);
    assert!(live.contains("Spectra"), "{live}");
    let (status, _, page) = http(port, "GET", "/c/general", &[("cookie", &cookie)], "");
    assert_eq!(status, 200);
    let csrf = page.split("name=\"_csrf\" value=\"").nth(1).and_then(|r| r.split('"').next()).expect("a CSRF token").to_string();

    // A post from the proxy's own page goes through, as the owner.
    let origin = format!("http://127.0.0.1:{port}");
    let form = format!("_csrf={csrf}&body=hello+from+radar+over+the+proxy");
    let hdrs = [("cookie", cookie.as_str()), ("origin", origin.as_str()), ("content-type", "application/x-www-form-urlencoded")];
    let (status, head, _) = http(port, "POST", "/c/general", &hdrs, &form);
    assert_eq!(status, 303, "{head}");
    let hits = json_of(c, &["search", "--json", "proxy"]);
    assert_eq!(hits.as_array().unwrap().len(), 1, "{hits}");
    assert_eq!(hits[0]["author"], "human");

    // Another site cannot use the proxy: not by Origin, nor by Host.
    let evil = [("cookie", cookie.as_str()), ("origin", "http://evil.example"), ("content-type", "application/x-www-form-urlencoded")];
    assert_eq!(http(port, "POST", "/c/general", &evil, &form).0, 403);
    assert_eq!(http(port, "GET", "/", &[("host", "evil.example"), ("cookie", &cookie)], "").0, 403);

    open.kill().ok();
    open.wait().ok();
    srv.stop();
    let o = run(s, &["admin", "verify"]);
    assert!(o.status.success(), "{}", text(&o));
}