//! A contract on the free job service, over HTTP, with its worker: what meets it
//! runs, its reasoning sealed; what does not is refused with every route's
//! reason; the service never reaches a provider; and a deadline it can hold a run
//! to, it holds.

use serde_json::{Value, json};
use std::io::{Read, Write};
use std::net::TcpStream;
use std::path::Path;
use std::process::{Command, Stdio};
use std::time::{Duration, Instant};

mod common;

use common::Running;

fn http(port: u16, method: &str, path: &str, body: Option<&Value>) -> (u16, Value) {
    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(120))).unwrap();
    let text = body.map(Value::to_string).unwrap_or_default();
    write!(s, "{method} {path} HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{text}", text.len()).unwrap();
    let mut out = String::new();
    s.read_to_string(&mut out).unwrap();
    let status = out.split_whitespace().nth(1).and_then(|c| c.parse().ok()).unwrap_or(0);
    let body = out.split_once("\r\n\r\n").map(|(_, b)| b.to_string()).unwrap_or_default();
    (status, serde_json::from_str(&body).unwrap_or(Value::Null))
}

fn start(dir: &Path) -> (Running, u16) {
    start_with(dir, "0")
}

fn start_with(dir: &Path, energy: &str) -> (Running, u16) {
    let mut cmd = Command::new(env!("CARGO_BIN_EXE_wai-quantum-service"));
    cmd.arg("serve");
    launch(cmd, dir, energy)
}

/// `cmd`, the service, started over the store at `dir`, once it answers: its process and
/// the port it bound. It binds a port the system chooses and says which in its log, as
/// tests/depth.rs's services do. A port found free here and handed to it raced: the
/// listener that finds it is inherited, until its exec, by a child another test forks at
/// that moment, so this test's first connection could reach that copy and be reset, or
/// refused once it closed, before the service had bound the port (CI's Linux job, where
/// the pid 1 test's probe forks as the other two start).
fn launch(mut cmd: Command, dir: &Path, energy: &str) -> (Running, u16) {
    let path = dir.join("service.log");
    cmd.env("WAI_QSVC_ADDR", "127.0.0.1:0")
        .env("WAI_QSVC_STORE", dir.join("store"))
        .env("WAI_QSVC_KEY", dir.join("signing.key"))
        .env("WAI_QSVC_JOB_GB", "2")
        .env("WAI_QSVC_BUDGET_GB", "4")
        .env("WAI_QSVC_WORKERS", "1")
        .env("WAI_QSVC_ENERGY", energy)
        .stdout(Stdio::null())
        .stderr(std::fs::File::create(&path).expect("the service's log"));
    let mut child = Running::new(cmd.spawn().expect("the service starts"));
    let log = || std::fs::read_to_string(&path).unwrap_or_default();
    let t = Instant::now();
    let port = loop {
        if let Some(p) = log().lines().find_map(|l| l.split(" on 127.0.0.1:").nth(1)?.split(',').next()?.trim().parse::<u16>().ok()) {
            break p;
        }
        if let Ok(Some(status)) = child.try_wait() {
            panic!("the service ended as it started ({status}): {}", log());
        }
        assert!(t.elapsed() < Duration::from_secs(60), "the service did not start in 60 s: {}", log());
        std::thread::sleep(Duration::from_millis(50));
    };
    while TcpStream::connect(("127.0.0.1", port)).is_err() || http(port, "GET", "/v1/health", None).0 != 200 {
        if let Ok(Some(status)) = child.try_wait() {
            panic!("the service ended ({status}): {}", log());
        }
        assert!(t.elapsed() < Duration::from_secs(60), "the service did not answer: {}", log());
        std::thread::sleep(Duration::from_millis(50));
    }
    (child, port)
}

fn until_done(port: u16, id: &str) -> Value {
    let t = Instant::now();
    loop {
        let (_, j) = http(port, "GET", &format!("/v1/jobs/{id}"), None);
        if ["done", "failed", "cancelled", "interrupted"].contains(&j["status"].as_str().unwrap_or("")) {
            return j;
        }
        assert!(t.elapsed() < Duration::from_secs(120), "{id} did not finish: {j}");
        std::thread::sleep(Duration::from_millis(50));
    }
}

/// The service, and the directory of its store and its log, removed once it has ended.
struct Service(Running, std::path::PathBuf);

impl Drop for Service {
    fn drop(&mut self) {
        self.0.stop();
        let _ = std::fs::remove_dir_all(&self.1);
    }
}

const BELL: &str = "OPENQASM 3.0;\nqubit[2] q;\nbit[2] c;\nh q[0];\ncx q[0], q[1];\nc = measure q;\n";

#[test]
fn a_contract_is_met_or_refused_and_the_service_never_reaches_a_provider() {
    let dir = std::env::temp_dir().join(format!("wai-qsvc-contract-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&dir);
    std::fs::create_dir_all(&dir).unwrap();
    let (child, port) = start(&dir);
    let _service = Service(child, dir.clone());

    // Met: its reasoning sealed, and the receipt held to it.
    let req = json!({ "qasm": BELL, "shots": 200, "contract": { "exactness": "exact", "deadline_seconds": 60, "providers": "allowed" } });
    let (code, made) = http(port, "POST", "/v1/jobs", Some(&req));
    assert_eq!(code, 202, "{made}");
    let job = until_done(port, made["id"].as_str().unwrap());
    assert_eq!(job["status"], "done", "{job}");
    let r = &job["result"];
    assert_eq!(r["planner"]["choice"], "method:stabilizer", "{}", r["planner"]);
    let c = wai_quantum::engine::job::check(&job);
    assert!(c.valid(), "{:?}", c.reasons);
    assert_eq!(r["planner"]["host"]["providers"], json!([]), "the service brings none");
    assert!(r["planner"]["routes"].as_array().unwrap().iter().all(|x| x["kind"] == "local"), "{}", r["planner"]);
    assert_eq!(r["planner"]["kept"]["deadline"]["kept"], true, "{}", r["planner"]["kept"]);
    assert_eq!(r["planner"]["stable"]["choice"], false, "providers allowed makes a choice the host's");
    // The same again, what a host that reaches none owes: run for this job alone.
    let plain = json!({ "qasm": BELL, "shots": 200 });
    let (_, made) = http(port, "POST", "/v1/jobs", Some(&plain));
    let plain_job = until_done(port, made["id"].as_str().unwrap());
    assert_eq!(plain_job["result"]["counts"], r["counts"], "the same seed, the same method, the same counts");
    assert!(plain_job["result"].get("planner").is_none());

    // A dry run shows every route, and the refused one every reason.
    let (_, dry) = http(port, "POST", "/v1/jobs", Some(&json!({ "qasm": BELL, "dry_run": true, "contract": { "exactness": "exact" } })));
    assert_eq!((dry["accepted"].as_bool(), dry["method"].as_str(), dry["planner"]["choice"].as_str()), (Some(true), Some("stabilizer"), Some("method:stabilizer")), "{dry}");
    let (_, dry) = http(port, "POST", "/v1/jobs", Some(&json!({ "qasm": BELL, "dry_run": true, "contract": { "exactness": "measured", "providers": "allowed" } })));
    assert_eq!((dry["accepted"].as_bool(), dry["refusal"]["code"].as_str()), (Some(false), Some("J071")), "{dry}");
    assert!(dry["planner"]["routes"].as_array().unwrap().iter().all(|x| x["rejected"]["code"] == "class" || x["rejected"]["code"] == "cannot"), "{dry}");

    // Asked for a device, for a provider by name, or for energy it has not metered: refused.
    for (req, code_want) in [
        (json!({ "qasm": BELL, "contract": { "exactness": "measured", "providers": "allowed" } }), "J071"),
        (json!({ "qasm": BELL, "provider": "lab", "contract": { "providers": "allowed" } }), "J002"),
        (json!({ "qasm": BELL, "contract": { "max_joules": 1000 } }), "J071"),
        (json!({ "qasm": BELL, "contract": { "exact": true } }), "J070"),
        (json!({ "programs": [BELL, BELL], "contract": {} }), "J073"),
    ] {
        let (code, body) = http(port, "POST", "/v1/jobs", Some(&req));
        assert_eq!(code, 400, "{req}: {body}");
        assert_eq!(body["code"], code_want, "{req}: {body}");
    }

    // A run past its deadline is stopped: this one's nominal time counts its state,
    // and its million shots are what takes the time. The service looks at the
    // worker again at the deadline itself, not at its next round 200 ms on, so a
    // run that would end within that round is stopped all the same.
    let slow = json!({ "qasm": "OPENQASM 3.0;\nqubit[24] q;\nbit[24] c;\nrx(0.3) q[0];\nc = measure q;\n", "method": "statevector", "shots": 1000000, "contract": { "deadline_seconds": 0.05 } });
    let (code, made) = http(port, "POST", "/v1/jobs", Some(&slow));
    assert_eq!(code, 202, "{made}");
    let job = until_done(port, made["id"].as_str().unwrap());
    assert_eq!(job["status"], "failed", "{job}");
    assert!(job["error"].as_str().unwrap().ends_with("[J074]"), "{job}");
    assert!(job.get("result").is_none(), "nothing was sealed: {job}");
}

/// A ceiling is held while the job runs: with the service's meter on, a live reading
/// of the worker stops it at the ceiling, and the receipt says so. Where nothing
/// reads the worker, the route is refused: that is the meter off, in the test above.
#[test]
fn a_job_is_stopped_at_its_energy_ceiling_and_its_receipt_says_so() {
    let dir = std::env::temp_dir().join(format!("wai-qsvc-ceiling-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&dir);
    std::fs::create_dir_all(&dir).unwrap();
    let (mut service, port) = start_with(&dir, "1");
    let (_, about) = http(port, "GET", "/v1", None);
    if about["energy"]["meter"] == "unmetered" || about["energy"].get("meter").is_none() {
        println!("nothing reads a worker on this host: {}", about["energy"]);
        let _ = service.kill();
        let _ = service.wait();
        if cfg!(target_os = "macos") {
            panic!("a Mac reads its worker's energy, so this test may not pass by skipping");
        }
        return;
    }
    let slow = json!({ "qasm": "OPENQASM 3.0;\nqubit[24] q;\nbit[24] c;\nrx(0.3) q[0];\nc = measure q;\n", "method": "statevector", "shots": 1000000, "contract": { "max_joules": 0.01 } });
    let (code, made) = http(port, "POST", "/v1/jobs", Some(&slow));
    // A host whose worker nothing reads refuses the route, with why.
    if code == 400 && made["code"] == "J071" && made["error"].as_str().is_some_and(|e| e.contains("reads no energy")) {
        let _ = service.kill();
        let _ = service.wait();
        if cfg!(target_os = "macos") {
            panic!("a Mac reads its worker's energy, so this test may not pass by skipping: {made}");
        }
        return;
    }
    assert_eq!(code, 202, "{made}");
    let job = until_done(port, made["id"].as_str().unwrap());
    let _ = service.kill();
    let _ = service.wait();
    let _ = std::fs::remove_dir_all(&dir);
    assert_eq!(job["status"], "done", "{job}");
    let r = &job["result"];
    assert_eq!(r["stopped"]["at"], "max_joules", "{r}");
    assert!(r["stopped"]["joules"].as_f64().unwrap() >= 0.01 && r.get("counts").is_none(), "{r}");
    assert_eq!(r["planner"]["kept"]["energy"]["kept"], false, "{}", r["planner"]["kept"]);
    assert_eq!(r["accuracy"]["class"], "none");
    // What the killed worker took in all is sealed, with the longest time between reads:
    // no less than the reading that stopped it, within the uncertainty the figure declares.
    // That reading is the counter as it stood (nanojoules, on a Mac); the figure is whole
    // microjoules rounded from the counter read after the kill, and its declared standard
    // uncertainty is at least that rounding's, half a microjoule either way (0.5/√3 µJ).
    // A worker killed before its counter moved again reads the same nanojoules at its
    // exit, and the figure rounded down from them was less than the reading itself
    // (macOS CI, 04db5802a and e52bf1154; 118 of 698 runs on the M3 Ultra, by at most
    // 0.497 µJ).
    let stopped_j = r["stopped"]["joules"].as_f64().unwrap();
    let whole_uj = r["energy"]["joules_micro"].as_u64();
    let declared_uj = whole_uj.zip(r["energy"]["energy_uncertainty"]["relative_ppm"].as_u64()).map(|(uj, ppm)| uj as f64 * ppm as f64 / 1e6);
    eprintln!("stopped at {stopped_j} J; the whole figure {whole_uj:?} µJ, its standard uncertainty {declared_uj:?} µJ");
    assert!(
        whole_uj.zip(declared_uj).is_some_and(|(uj, u)| uj as f64 + 3f64.sqrt() * u >= stopped_j * 1e6),
        "the whole figure is less than the reading that stopped the worker ({stopped_j} J), past its declared uncertainty: {}",
        r["energy"]
    );
    assert!(r["stopped"]["read"].as_str().unwrap().contains("the longest time between two reads"), "{}", r["stopped"]);
    let body = json!({ "request": job["request"], "result": r });
    assert!(wai_quantum::engine::job::check(&body).valid(), "{:?}", wai_quantum::engine::job::check(&body).reasons);
}

/// The service as a namespace's first process (pid 1), as a container runs it with
/// no init: its workers are its children like any others, and run.
///
/// It needs a user and a pid namespace made for this user. Where the kernel refuses
/// them, the test does not run, and says so on standard error, past the harness's
/// capture, in a line that starts `SKIPPED the_service_runs_its_jobs_as_pid_1`: the
/// harness still reports it `ok`, so a workflow that needs it run looks for that line.
/// The kernel refuses with `EPERM` in a container with no `CAP_SYS_ADMIN` under a
/// seccomp profile that allows `unshare` only with it (Docker's default, which CI's
/// Linux job runs under), or where unprivileged user namespaces are off; with `ENOSPC`
/// where `user.max_user_namespaces` is 0; with `EINVAL` where the kernel has no such
/// namespaces. Any other failure of the probe (no `unshare`, an option it lacks) fails
/// the test. `--kill-child` ends the service when `unshare` is killed: without it the
/// service, pid 1 of a namespace of its own, outlives the test.
#[cfg(target_os = "linux")]
#[test]
fn the_service_runs_its_jobs_as_pid_1() {
    let probe = Command::new("unshare")
        .args(["--user", "--pid", "--fork", "--kill-child", "true"])
        .env("LC_ALL", "C")
        .stdin(Stdio::null())
        .stdout(Stdio::null())
        .stderr(Stdio::piped())
        .output()
        .expect("unshare (util-linux) starts");
    if !probe.status.success() {
        let why = String::from_utf8_lossy(&probe.stderr).trim().to_string();
        let refused = ["unshare failed: Operation not permitted", "unshare failed: No space left on device", "unshare failed: Invalid argument"];
        assert!(refused.iter().any(|r| why.contains(r)), "unshare --user --pid failed ({}), and not with the kernel's refusal of the namespaces: {why}", probe.status);
        let _ = writeln!(std::io::stderr(), "SKIPPED the_service_runs_its_jobs_as_pid_1: the kernel made no user and pid namespace for this user ({why}), so the service was not run as pid 1 here");
        return;
    }
    let dir = std::env::temp_dir().join(format!("wai-qsvc-pid1-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&dir);
    std::fs::create_dir_all(&dir).unwrap();
    let mut cmd = Command::new("unshare");
    cmd.args(["--user", "--pid", "--fork", "--kill-child"]).arg(env!("CARGO_BIN_EXE_wai-quantum-service")).arg("serve");
    let (mut service, port) = launch(cmd, &dir, "0");
    let bell = json!({ "qasm": "OPENQASM 3.0; qubit[2] q; bit[2] c; h q[0]; cx q[0], q[1]; c = measure q;", "shots": 100 });
    let (code, done) = http(port, "POST", "/v1/run", Some(&bell));
    let _ = service.kill();
    let _ = service.wait();
    let _ = std::fs::remove_dir_all(&dir);
    assert_eq!((code, done["status"].as_str()), (200, Some("done")), "{done}");
}