//! The open provider protocol end to end, against the reference provider
//! over both transports: a local process (`wai-qpu reference --stdio`) and
//! HTTP (the reference served on 127.0.0.1, reached with `curl`). Every
//! operation, the conformance vectors, the reference's pinned counts, the
//! person's registry with its credential in a file only they can read, the
//! refusal of a run that is not free until it is confirmed, receipts that
//! verify with the engine's verifier, and no credential in anything a run
//! returns.
//!
//! No request leaves this machine and nothing here spends money: the priced
//! runs are the reference's, whose price is a number in its configuration.

use serde_json::{Value, json};
use std::sync::Arc;
use std::time::Duration;
use wai_quantum::ed25519_dalek::SigningKey;
use wai_quantum::engine::job::verify;
use wai_quantum_qpu::conformance::{self, Outcome};
use wai_quantum_qpu::job::{self, JobError, Target};
use wai_quantum_qpu::open_provider::{Channel, Http, OpenProvider, PROTOCOL, Process};
use wai_quantum_qpu::reference::{self, Config, Per, Reference};
use wai_quantum_qpu::registry::{Credential, Entry, Kind, Registry, Stores};
use wai_quantum_qpu::transport::Curl;
use wai_quantum_qpu::{Provider, QpuSampler, Secret, Wait};

const SECRET: &str = "wai-test-credential-5f1c9e";
const BELL: &str = "OPENQASM 3.0; qubit[2] q; bit[2] c; h q[0]; cx q[0], q[1]; c = measure q;";

fn bin() -> String {
    env!("CARGO_BIN_EXE_wai-qpu").to_string()
}

fn have_curl() -> bool {
    std::process::Command::new("curl").arg("--version").output().is_ok()
}

fn no_wait() -> Wait {
    Wait { interval: Duration::ZERO, polls: 20 }
}

fn key() -> SigningKey {
    SigningKey::from_bytes(&[21; 32])
}

/// The reference as a process, with `args` after `reference --stdio`.
fn process(args: &[&str], credential: Option<&str>) -> Process {
    let mut command = vec![bin(), "reference".into(), "--stdio".into()];
    command.extend(args.iter().map(|a| a.to_string()));
    Process::new(command, credential.map(Secret::new)).unwrap().with_timeout(Duration::from_secs(60))
}

/// How long [`Running`] waits for a process it has killed to be reaped. `SIGKILL` ends a
/// process at once unless it is stuck in the kernel; a drop that waited on such a one without
/// a bound would hang the test binary instead of failing it.
const REAPED_WITHIN: Duration = Duration::from_secs(10);

/// A process a test started, killed and reaped when it is dropped: at the end of the test, or
/// while a failed assertion unwinds through it. A [`std::process::Child`] is not killed when it
/// is dropped, so a test that failed before its last line left the reference it served over
/// HTTP running after the test binary had ended, holding its port, one more for each failed
/// run. It reads as the child it holds.
struct Running(std::process::Child);

impl Running {
    /// Kills the process and reaps it, waiting up to [`REAPED_WITHIN`]: whether it was reaped.
    /// A process already reaped is not signalled again (`Child::kill` does nothing then, as its
    /// pid may be another process's by now).
    fn stop(&mut self) -> bool {
        let _ = self.0.kill();
        let t = std::time::Instant::now();
        loop {
            match self.0.try_wait() {
                Ok(Some(_)) => return true,
                Ok(None) if t.elapsed() < REAPED_WITHIN => std::thread::sleep(Duration::from_millis(10)),
                _ => return false,
            }
        }
    }
}

impl std::ops::Deref for Running {
    type Target = std::process::Child;

    fn deref(&self) -> &std::process::Child {
        &self.0
    }
}

impl std::ops::DerefMut for Running {
    fn deref_mut(&mut self) -> &mut std::process::Child {
        &mut self.0
    }
}

impl Drop for Running {
    fn drop(&mut self) {
        if !self.stop() {
            eprintln!("process {} was killed and not reaped within {} s", self.0.id(), REAPED_WITHIN.as_secs());
        }
    }
}

/// The reference served over HTTP in this process, and its endpoint.
fn http(config: Config) -> String {
    let at = reference::spawn_http(Arc::new(Reference::new(config).unwrap()), "127.0.0.1:0").unwrap();
    format!("http://{at}/v1/provider")
}

fn op(ch: &dyn Channel, op: &str, fields: Value) -> Value {
    let mut m = json!({ "protocol": PROTOCOL, "op": op });
    for (k, v) in fields.as_object().unwrap() {
        m[k] = v.clone();
    }
    ch.exchange(&m).unwrap()
}

/// Every operation on one channel: describe, estimate, submit, status,
/// result and cancel, each as the protocol states it.
fn every_operation(ch: &dyn Channel) {
    let d = op(ch, "describe", json!({}));
    assert_eq!((d["ok"].as_bool(), d["protocol"].as_str(), d["qubits"].as_u64()), (Some(true), Some(PROTOCOL), Some(5)), "{d}");
    // A program the device runs, compiled here.
    let provider = OpenProvider::new(ch);
    let sampler = QpuSampler::new(&provider).unwrap();
    let c = wai_quantum::qasm::parse(BELL).unwrap().dyadic().unwrap();
    let program = sampler.compile(&c).unwrap();
    let wire = wai_quantum_qpu::open_provider::program_json(&program);
    let e = op(ch, "estimate", json!({ "program": wire, "shots": 100 }));
    assert_eq!((e["ok"].as_bool(), e["free"].as_bool()), (Some(true), Some(true)), "{e}");
    let s = op(ch, "submit", json!({ "program": wire, "shots": 100, "seed": 3 }));
    let id = s["job"].as_str().unwrap_or_else(|| panic!("{s}")).to_string();
    assert_eq!(op(ch, "status", json!({ "job": id }))["status"], "done");
    let r = op(ch, "result", json!({ "job": id }));
    let counts = r["counts"].as_object().unwrap();
    assert_eq!(counts.values().map(|v| v.as_u64().unwrap()).sum::<u64>(), 100);
    assert!(counts.keys().all(|k| k == "00" || k == "11"), "{r}");
    assert_eq!(r["calibration"], d["calibration"]["id"]);
    // A finished job cannot be cancelled: it stays done.
    assert_eq!(op(ch, "cancel", json!({ "job": id }))["status"], "done");
    assert_eq!(op(ch, "result", json!({ "job": "nope" }))["error"]["code"], "not_found");
}

#[test]
fn every_operation_over_a_local_process() {
    every_operation(&process(&[], None));
}

#[test]
fn every_operation_over_http() {
    if !have_curl() {
        println!("no curl on this host: the HTTP transport runs where curl is");
        return;
    }
    every_operation(&Http::new(Curl::new(30), http(Config::default()), None).unwrap());
}

/// The conformance vectors pass on the reference, every check run.
fn conforms(ch: &dyn Channel) {
    let r = conformance::check_open(ch, &serde_json::Map::new(), false, no_wait());
    assert!(r.passed(), "{}", r.text());
    assert_eq!(r.count(Outcome::Skip), 0, "a free provider runs every check: {}", r.text());
    // Fifteen messages, the description, six programs estimated, two
    // refusals, and six programs run with their cancels; and its
    // energy-landscape jobs (section 11): the offer, thirteen messages (three of them
    // the second version's), and twelve jobs estimated and answered, six of each
    // version of the landscape protocol.
    assert_eq!(r.count(Outcome::Pass), 15 + 1 + 6 + 2 + 6 * 2 + 1 + 13 + 12 * 2, "{}", r.text());
}

#[test]
fn the_reference_conforms_over_both_transports() {
    conforms(&process(&[], None));
    if have_curl() {
        conforms(&Http::new(Curl::new(30), http(Config::default()), None).unwrap());
    }
    // And in each other basis, all to all.
    for basis in ["prxcz", "gpizz"] {
        conforms(&process(&["--basis", basis, "--all-to-all"], None));
    }
}

#[test]
fn a_priced_provider_runs_no_job_in_its_conformance_check_without_leave() {
    let ch = process(&["--price", "0.01", "--currency", "USD", "--per", "job"], None);
    let r = conformance::check_open(&ch, &serde_json::Map::new(), false, no_wait());
    assert!(r.passed(), "{}", r.text());
    let skips: Vec<_> = r.checks.iter().filter(|c| c.outcome == Outcome::Skip).collect();
    for name in ["the programs run", "landscape: the jobs run"] {
        assert!(skips.iter().any(|s| s.name == name && s.detail.contains("0.01 USD")), "{name}: {skips:?}");
    }
    assert!(!r.checks.iter().any(|c| c.name.ends_with(": runs")));
    let r = conformance::check_open(&ch, &serde_json::Map::new(), true, no_wait());
    assert!(r.passed() && r.count(Outcome::Skip) == 0, "{}", r.text());
}

#[test]
fn a_provider_that_breaks_the_protocol_fails_its_check() {
    // One that answers every unknown thing ok, and bills in floats.
    let r = Arc::new(Reference::new(Config::default()).unwrap());
    let sloppy = wai_quantum_qpu::open_provider::Local(move |m: &Value| {
        let mut a = r.handle(m, None);
        if a["ok"] == false {
            a = json!({ "ok": true });
        }
        if m["op"] == "describe" {
            a["billing"] = json!({ "unit": "shot", "price": 0.01, "currency": "USD" });
        }
        a
    });
    let report = conformance::check_open(&sloppy, &serde_json::Map::new(), false, no_wait());
    assert!(!report.passed());
    let failed: Vec<&str> = report.checks.iter().filter(|c| c.outcome == Outcome::Fail).map(|c| c.name.as_str()).collect();
    assert!(failed.contains(&"an unknown operation is unsupported") && failed.contains(&"the description is complete"), "{failed:?}");
}

#[test]
fn the_reference_gives_its_pinned_counts_on_every_platform() {
    let v = conformance::vectors();
    let pins = &v["reference"]["counts"];
    let r = Arc::new(Reference::new(Config::default()).unwrap());
    let ch = wai_quantum_qpu::open_provider::Local(move |m: &Value| r.handle(m, None));
    let provider = OpenProvider::new(&ch);
    let sampler = QpuSampler::new(&provider).unwrap();
    let mut got = serde_json::Map::new();
    for p in v["programs"].as_array().unwrap() {
        let prog = wai_quantum::qasm::parse(p["qasm"].as_str().unwrap()).unwrap();
        let (counts, _) = sampler.run_seeded(&prog.dyadic().unwrap(), p["shots"].as_u64().unwrap(), Some(1)).unwrap();
        let mut keyed = serde_json::Map::new();
        for (x, &k) in counts.iter().enumerate() {
            if k > 0 {
                keyed.insert(prog.bitstring(|q| x >> q & 1 == 1), json!(k));
            }
        }
        got.insert(p["name"].as_str().unwrap().to_string(), Value::Object(keyed));
    }
    assert_eq!(pins, &Value::Object(got.clone()), "the reference's counts at seed 1 are not the pinned ones; these are:\n{}", Value::Object(got));
}

fn priced() -> Vec<&'static str> {
    vec!["--price", "0.0003", "--currency", "USD"]
}

#[test]
fn a_run_that_is_not_free_is_refused_until_confirmed_and_carries_its_cost() {
    let t = Target::new("lab", Kind::Process, Box::new(OpenProvider::new(process(&priced(), None))), None);
    let req = json!({ "qasm": BELL, "shots": 1000, "provider": "lab" });
    let p = job::plan(&t, &req).unwrap();
    assert!(!p.free);
    assert_eq!((p.json["estimate"]["amount"].as_str(), p.json["estimate"]["currency"].as_str(), p.json["estimate"]["unit"].as_str()), (Some("0.3"), Some("USD"), Some("shot")), "{}", p.json);
    assert_eq!(p.json["billing"], json!({ "free": false, "unit": "shot", "price": "0.0003", "currency": "USD" }));
    match job::run(&t, &req, false, &key(), no_wait(), |_| {}) {
        Err(JobError::Unconfirmed { message, .. }) => assert!(message.contains("0.3 USD") && message.contains("confirm"), "{message}"),
        other => panic!("ran without a confirmation: {other:?}"),
    }
    let body = job::run(&t, &req, true, &key(), no_wait(), |_| {}).unwrap();
    assert!(verify(&body).unwrap().0);
    let cost = &body["result"]["cost"];
    assert_eq!((cost["confirmed"].as_bool(), cost["estimate"]["amount"].as_str(), cost["charged"]["amount"].as_str()), (Some(true), Some("0.3"), Some("0.3")), "{cost}");
}

#[test]
fn a_provider_run_is_sealed_with_what_it_ran_and_verifies() {
    let t = Target::new("lab", Kind::Process, Box::new(OpenProvider::new(process(&["--polls", "3", "--readout-error", "0.05"], None))), None);
    let req = json!({ "qasm": BELL, "shots": 2000, "seed": 9, "provider": "lab" });
    let body = job::run(&t, &req, false, &key(), no_wait(), |_| {}).unwrap();
    let (ok, signer) = verify(&body).unwrap();
    assert!(ok, "{body}");
    assert_eq!(signer.as_str(), Some(key().verifying_key().as_bytes().iter().map(|b| format!("{b:02x}")).collect::<String>().as_str()));
    let r = &body["result"];
    // A simulator's counts are simulated, not measured; the reference says it
    // drew them one shot at a time and ran the gates as written.
    assert_eq!((r["method"].as_str(), r["accuracy"]["class"].as_str(), r["counts_source"].as_str()), (Some("provider"), Some("simulated"), Some("shots")));
    assert_eq!((r["execution"]["stated_by"].as_str(), r["execution"]["verbatim"].as_bool(), r["execution"]["mitigation"].as_str()), (Some("provider"), Some(true), Some("none")), "{}", r["execution"]);
    assert_eq!(r["job"]["polls"], 3);
    assert_eq!((r["provider"]["name"].as_str(), r["provider"]["protocol"].as_str(), r["provider"]["provenance"].as_str()), (Some("lab"), Some(PROTOCOL), Some("simulator:the wai-quantum exact simulator: the state vector in fixed point, sampled with the seed given")));
    assert!(r["calibration"].as_str().unwrap().starts_with("reference-"));
    assert_eq!(r["compiled"]["program"].as_str().unwrap().len(), 64);
    // The readout error moves the counts off the exact distribution, by
    // about twice the error on two bits: well above what 2000 shots leave.
    let d = r["ideal"]["total_variation_distance"].as_f64().unwrap();
    let floor = r["ideal"]["sampling_tvd_expected"].as_f64().unwrap();
    assert!(d > 0.06 && d < 0.14 && floor < 0.02, "{}", r["ideal"]);
    // Any change to what it ran breaks the seal.
    for path in ["/result/calibration", "/result/compiled/program", "/result/provider/protocol", "/request/seed"] {
        let mut forged = body.clone();
        *forged.pointer_mut(path).unwrap() = json!("x");
        assert!(!verify(&forged).unwrap().0, "{path}");
    }
}

/// Everything a run returns, as text.
fn all_text(t: &Target, req: &Value) -> String {
    let mut out = String::new();
    out.push_str(&job::plan(t, req).map(|p| p.json.to_string()).unwrap_or_else(|e| e.to_string()));
    out.push_str(&match job::run(t, req, true, &key(), no_wait(), |_| {}) {
        Ok(b) => b.to_string(),
        Err(e) => format!("{e} {e:?}"),
    });
    let r = conformance::check_builtin(t, false, no_wait());
    out.push_str(&r.text());
    out
}

#[test]
fn a_credential_reaches_the_provider_and_nothing_a_run_returns() {
    let hash = reference::credential_hash(SECRET);
    // Over a process: in its environment, never on its command line.
    let ch = process(&["--credential-blake3", &hash], Some(SECRET));
    assert!(!ch.label().contains(SECRET));
    let t = Target::new("lab", Kind::Process, Box::new(OpenProvider::new(ch)), Some(Secret::new(SECRET)));
    let req = json!({ "qasm": BELL, "shots": 100, "provider": "lab" });
    let text = all_text(&t, &req);
    assert!(text.contains("\"receipt\""), "the run did not go through: {text}");
    assert!(!text.contains(SECRET), "the credential is in a run's output");
    // Without it, the provider refuses, and the refusal names no credential.
    let t = Target::new("lab", Kind::Process, Box::new(OpenProvider::new(process(&["--credential-blake3", &hash], None))), None);
    let e = job::plan(&t, &req).err().unwrap().to_string();
    assert!(e.contains("unauthorized"), "{e}");
    // Over HTTP: in the Authorization header, and a provider that quotes it
    // back has it taken out.
    if have_curl() {
        let c = Config { credential_blake3: Some(blake3::hash(SECRET.as_bytes()).into()), ..Config::default() };
        let t = Target::new("lab", Kind::Http, Box::new(OpenProvider::new(Http::new(Curl::new(30), http(c), Some(Secret::new(SECRET))).unwrap())), Some(Secret::new(SECRET)));
        let text = all_text(&t, &req);
        assert!(text.contains("\"receipt\"") && !text.contains(SECRET), "{}", text.replace(SECRET, "<the credential>"));
        let quotes = Arc::new(Reference::new(Config::default()).unwrap());
        let echo = wai_quantum_qpu::open_provider::Local(move |m: &Value| if m["op"] == "submit" { json!({ "ok": false, "error": { "code": "refused", "message": format!("bad token {SECRET}") } }) } else { quotes.handle(m, None) });
        let t = Target::new("lab", Kind::Http, Box::new(OpenProvider::new(echo)), Some(Secret::new(SECRET)));
        let e = job::run(&t, &req, true, &key(), no_wait(), |_| {}).err().unwrap().to_string();
        assert!(e.contains("bad token …") && !e.contains(SECRET), "{e}");
    }
}

#[test]
fn the_registry_runs_a_provider_with_its_credential_in_a_file_only_its_owner_reads() {
    let dir = std::env::temp_dir().join(format!("wai-qpu-providers-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&dir);
    let stores = Stores::files(dir.join("credentials"));
    let mut reg = Registry::load(dir.join("quantum-providers.json")).unwrap();
    let (credential, note) = stores.keep("lab", &Secret::new(SECRET)).unwrap();
    assert_eq!(credential, Credential::File);
    assert!(note.contains("only you can read"));
    let entry = Entry {
        name: "lab".into(),
        kind: Kind::Process,
        endpoint: None,
        command: vec![bin(), "reference".into(), "--stdio".into(), "--credential-blake3".into(), reference::credential_hash(SECRET)],
        settings: serde_json::Map::new(),
        credential,
    };
    reg.add(entry, false).unwrap();
    reg.save().unwrap();
    let list = std::fs::read_to_string(reg.path()).unwrap();
    assert!(!list.contains(SECRET), "the list holds the credential");
    let reg = Registry::load(reg.path()).unwrap();
    let t = wai_quantum_qpu::registry::open(reg.get("lab").unwrap(), &stores).unwrap();
    let body = job::run(&t, &json!({ "qasm": BELL, "shots": 64, "provider": "lab" }), false, &key(), no_wait(), |_| {}).unwrap();
    assert!(verify(&body).unwrap().0);
    assert!(!body.to_string().contains(SECRET));
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        assert_eq!(std::fs::metadata(dir.join("credentials").join("lab")).unwrap().permissions().mode() & 0o777, 0o600);
    }
    let _ = std::fs::remove_dir_all(&dir);
}

#[cfg(unix)]
#[test]
fn a_process_that_dies_or_hangs_is_an_error_not_a_wait() {
    let gone = Process::new(vec!["false".into()], None).unwrap();
    let e = gone.exchange(&json!({ "protocol": PROTOCOL, "op": "describe" })).unwrap_err().to_string();
    assert!(e.contains("ended without an answer") || e.contains("did not take the message"), "{e}");
    let hangs = Process::new(vec!["sleep".into(), "30".into()], None).unwrap().with_timeout(Duration::from_millis(300));
    let e = hangs.exchange(&json!({ "protocol": PROTOCOL, "op": "describe" })).unwrap_err().to_string();
    assert!(e.contains("no answer within"), "{e}");
    let missing = Process::new(vec!["wai-qpu-no-such-program".into()], None).unwrap();
    assert!(missing.exchange(&json!({})).is_err());
}

#[test]
fn the_reference_command_serves_http_and_says_where() {
    if !have_curl() {
        return;
    }
    use std::io::BufRead;
    // Held as it starts, so one that never says where it listens is killed too.
    let mut child = Running(std::process::Command::new(bin()).args(["reference", "--http", "127.0.0.1:0", "--qubits", "3"]).stdout(std::process::Stdio::piped()).spawn().unwrap());
    let mut line = String::new();
    std::io::BufReader::new(child.stdout.take().unwrap()).read_line(&mut line).unwrap();
    let url = line.trim().strip_prefix("listening ").unwrap_or_else(|| panic!("{line}")).to_string();
    let h = Http::new(Curl::new(30), url, None).unwrap();
    let d = op(&h, "describe", json!({}));
    let _ = child.kill();
    let _ = child.wait();
    assert_eq!(d["qubits"], 3, "{d}");
    // A provider of the open protocol is a fabric sampler like any other.
    let p = OpenProvider::new(Http::new(Curl::new(30), http(Config { price: Some(("1".into(), "EUR".into(), Per::Job)), ..Config::default() }), None).unwrap());
    let e = p.estimate(&QpuSampler::new(&p).unwrap().compile(&wai_quantum_qpu::probe_circuit(2)).unwrap(), 10).unwrap().unwrap();
    assert_eq!((e.free, e.cost.unwrap().amount.as_str()), (false, "1"));
}