//! The reference provider: the open provider protocol around this crate's
//! exact simulator, served over HTTP ([`serve_http`]) or on standard input
//! and output ([`serve_stdio`]).
//!
//! It is what the tests run against and the template to copy: a provider for
//! any device is this file with [`Reference::handle`]'s simulation replaced by
//! the device. Everything else it does is what the protocol asks of every
//! provider:
//!
//! - `describe` names the device: its qubits, its native gates, its
//!   connectivity, a calibration snapshot (here a hash of its configuration,
//!   with each qubit's readout error and each pair's error), and its billing;
//! - `estimate` prices a program before it is sent (free, or a price per shot
//!   or per job, as configured, in exact decimals), or, configured with no
//!   price set, answers that it is not free and sets no price for it (the
//!   unpriced form, section 5.2), after checking it as `submit` would;
//! - `submit` checks the program against the device (its basis, its qubits,
//!   its pairs, and on a star device the resonator used in turn) and runs it:
//!   the state vector in the core's fixed point, each `move` the exchange it
//!   is on every state a program in turn reaches,
//!   sampled with the seed the message gives (1 when it gives none), each
//!   measured bit then flipped with the configured readout error;
//! - `status`, `result` and `cancel` answer for the jobs it holds, a job
//!   reading `running` for the configured number of status checks first;
//! - with a credential configured (as its BLAKE3 hash, so the credential
//!   itself is on no command line), every message without it is refused
//!   `unauthorized`;
//! - it takes energy-landscape jobs (`wai.quantum.landscape/1` and
//!   `wai.quantum.landscape/2`, the protocol's §11) on the extension's CPU reference,
//!   unless configured not to, or to take one version alone: `estimate` counts a job's
//!   operations and memory as the extension does, and refuses a request its reader
//!   refuses with the reader's own sentence and line; a job of a version its
//!   description does not list is `unsupported`; `submit` runs it; `result` gives the
//!   extension's answer, of the request's version. With no qubits it takes those
//!   alone, and refuses every program.

use crate::open_provider::{CREDENTIAL_ENV, PROTOCOL, WireProgram, native_gates, program_from_json};
use crate::basis_label;
use serde_json::{Value, json};
use std::collections::BTreeMap;
use std::io::{BufRead, BufReader, Read, Write};
use std::net::{SocketAddr, TcpListener, TcpStream};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use wai_quantum::quantum::MAX_QUBITS;
use wai_quantum::quantum_toolchain::from_qasm;
use wai_quantum::quantum_toolchain::native::{Basis, Coupling, Dyadic, NativeGate, resonator_discipline};

/// The most shots one job may ask for.
pub const MAX_SHOTS: u64 = 1_000_000;
/// The most bytes a message may hold over HTTP.
const MAX_BODY: usize = 16 << 20;

/// What a price counts.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Per {
    Shot,
    Job,
}

/// The device the reference provider declares and simulates.
#[derive(Clone, Debug, PartialEq)]
pub struct Config {
    /// The provider's name for itself.
    pub name: String,
    pub qubits: u16,
    pub basis: Basis,
    /// Every pair connected, rather than a line.
    pub all_to_all: bool,
    /// A star: the qubits round one computational resonator (the place after
    /// the last qubit), each coupled to it by `cz` and each moving into it,
    /// rather than a line. Only for a basis whose two-qubit gate is `cz`.
    pub star: bool,
    /// `None` for free (unless `unpriced`); else the price of each unit, in
    /// exact decimals, and its currency.
    pub price: Option<(String, String, Per)>,
    /// Not free, and no price set: `billing` says `{"free": false, "priced":
    /// false}` (section 5.1), and each estimate answers the unpriced form
    /// (section 5.2), stating a landscape job's counts and no amount. Not with a
    /// `price`.
    pub unpriced: bool,
    /// The free use offered, as text.
    pub free_tier: Option<String>,
    /// Status checks a job reads `running` before it is `done`.
    pub polls: u32,
    /// The chance each measured bit is read flipped.
    pub readout_error: f64,
    /// The BLAKE3 hash of the credential every message must carry.
    pub credential_blake3: Option<[u8; 32]>,
    /// Whether it takes energy-landscape jobs (§11), on the extension's CPU reference.
    pub landscape: bool,
    /// The versions of the landscape protocol it takes, where it takes landscape jobs:
    /// both by default, as its description's `landscape.protocols` then lists them.
    pub landscape_protocols: Vec<String>,
}

impl Default for Config {
    fn default() -> Self {
        Config { name: "reference".into(), qubits: 5, basis: Basis::RzSxXCz, all_to_all: false, star: false, price: None, unpriced: false, free_tier: None, polls: 0, readout_error: 0.0, credential_blake3: None, landscape: true, landscape_protocols: LANDSCAPE_PROTOCOLS.map(String::from).to_vec() }
    }
}

/// A credential's hash, as [`Config::credential_blake3`] holds it, in hex.
pub fn credential_hash(credential: &str) -> String {
    hex(blake3::hash(credential.as_bytes()).as_bytes())
}

fn hex(b: &[u8]) -> String {
    b.iter().map(|x| format!("{x:02x}")).collect()
}

fn unhex32(s: &str) -> Option<[u8; 32]> {
    if s.len() != 64 {
        return None;
    }
    let mut out = [0u8; 32];
    for (i, b) in out.iter_mut().enumerate() {
        *b = u8::from_str_radix(s.get(2 * i..2 * i + 2)?, 16).ok()?;
    }
    Some(out)
}

/// A decimal amount times a whole number, exactly: `"0.0003" × 1000 = "0.3"`.
pub fn decimal_times(amount: &str, n: u64) -> Option<String> {
    let (int, frac) = amount.split_once('.').unwrap_or((amount, ""));
    if int.is_empty() || int.len() + frac.len() > 30 || !int.bytes().chain(frac.bytes()).all(|b| b.is_ascii_digit()) {
        return None;
    }
    let digits: u128 = format!("{int}{frac}").parse().ok()?;
    let product = digits.checked_mul(u128::from(n))?.to_string();
    let scale = frac.len();
    let padded = format!("{product:0>width$}", width = scale + 1);
    let (i, f) = padded.split_at(padded.len() - scale);
    let f = f.trim_end_matches('0');
    Some(if f.is_empty() { i.to_string() } else { format!("{i}.{f}") })
}

/// One job the provider holds.
struct Job {
    /// An energy-landscape job's answer, in place of counts.
    landscape: Option<Value>,
    counts: BTreeMap<String, u64>,
    shots: u64,
    polls_left: u32,
    cancelled: bool,
    charged: Option<String>,
}

/// The reference provider: see the module documentation.
pub struct Reference {
    config: Config,
    jobs: Mutex<BTreeMap<String, Job>>,
    next: AtomicU64,
}

fn error(code: &str, message: impl Into<String>) -> Value {
    json!({ "ok": false, "error": { "code": code, "message": message.into() } })
}

/// The `splitmix64` step, for the readout errors' draws.
fn splitmix64(state: &mut u64) -> u64 {
    *state = state.wrapping_add(0x9e37_79b9_7f4a_7c15);
    let mut z = *state;
    z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9);
    z = (z ^ (z >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb);
    z ^ (z >> 31)
}

impl Reference {
    /// A provider declaring `config`'s device, or why the configuration is
    /// not one.
    pub fn new(config: Config) -> Result<Self, String> {
        if (config.qubits == 0 && !config.landscape) || config.qubits > 1024 {
            return Err(format!("{} qubits; the reference device has 1 to 1024, or none where it takes energy-landscape jobs alone", config.qubits));
        }
        if !(0.0..=0.5).contains(&config.readout_error) {
            return Err("a readout error is a probability from 0 to 0.5".into());
        }
        if config.basis == Basis::RzSxXCzRxRzz {
            return Err("rzsxxczrxrzz is a built-in client's basis: a client of the protocol compiles to rzsxxcz on a device that lists its gates (section 4)".into());
        }
        if config.star && (config.all_to_all || !config.basis.takes_resonators()) {
            return Err("a star device runs move and cz on its resonator, in rzsxxcz or prxcz, and is not also all to all".into());
        }
        if let Some((price, currency, _)) = &config.price
            && (!crate::open_provider::is_decimal(price) || currency.is_empty())
        {
            return Err(format!("{price} {currency}: a price is a decimal amount and a currency"));
        }
        if config.unpriced && config.price.is_some() {
            return Err("a device that sets no price has no price: --unpriced or --price, not both".into());
        }
        Ok(Reference { config, jobs: Mutex::new(BTreeMap::new()), next: AtomicU64::new(1) })
    }

    pub fn config(&self) -> &Config {
        &self.config
    }

    fn connected(&self, a: u16, b: u16) -> bool {
        let c = &self.config;
        if c.star {
            // Each qubit with the resonator, at index `qubits`.
            return a != b && (a == c.qubits) != (b == c.qubits) && a.min(b) < c.qubits && a.max(b) <= c.qubits;
        }
        a != b && a < c.qubits && b < c.qubits && (c.all_to_all || a.abs_diff(b) == 1)
    }

    /// The device's connectivity, its resonator among it on a star.
    pub fn coupling(&self) -> Coupling {
        let n = self.config.qubits;
        if self.config.star {
            let pairs: Vec<(u16, u16)> = (0..n).map(|q| (q, n)).collect();
            return Coupling::with_resonators(n + 1, &pairs, &[n], &pairs).expect("a star is a valid map");
        }
        let pairs: Vec<(u16, u16)> = (0..n).flat_map(|a| (a + 1..n).map(move |b| (a, b))).filter(|&(a, b)| self.connected(a, b)).collect();
        Coupling::new(n, &pairs).expect("a line or a complete map is valid")
    }

    /// The device's name: `line-5`, `all-5`, `star-5`; `landscape` for a device that
    /// takes energy-landscape jobs alone.
    pub fn device(&self) -> String {
        let c = &self.config;
        if c.qubits == 0 {
            return "landscape".into();
        }
        format!("{}-{}", if c.star { "star" } else if c.all_to_all { "all" } else { "line" }, c.qubits)
    }

    /// The calibration snapshot: a hash of everything the device's behaviour
    /// depends on.
    pub fn calibration_id(&self) -> String {
        let c = &self.config;
        let mut text = format!("{}|{}|{}|{}|{}", c.qubits, basis_label(c.basis), c.all_to_all, c.readout_error.to_bits(), c.name);
        if c.star {
            text.push_str("|star");
        }
        format!("reference-{}", &hex(blake3::hash(text.as_bytes()).as_bytes())[..12])
    }

    /// The `describe` answer.
    pub fn describe(&self) -> Value {
        let c = &self.config;
        let qubits: Vec<Value> = (0..c.qubits).map(|q| json!({ "qubit": q, "readout_error": c.readout_error })).collect();
        let mut calibration = json!({ "id": self.calibration_id(), "qubits": qubits });
        let connectivity = if c.qubits == 0 {
            json!([])
        } else if c.star {
            let pairs: Vec<Value> = (0..c.qubits).map(|q| json!([q, c.qubits])).collect();
            calibration["edges"] = Value::Array((0..c.qubits).map(|q| json!({ "qubits": [q, c.qubits], "error": 0.0 })).collect());
            Value::Array(pairs)
        } else if c.all_to_all {
            json!("all")
        } else {
            let pairs: Vec<Value> = (1..c.qubits).map(|q| json!([q - 1, q])).collect();
            calibration["edges"] = Value::Array((1..c.qubits).map(|q| json!({ "qubits": [q - 1, q], "error": 0.0 })).collect());
            Value::Array(pairs)
        };
        let mut billing = match &c.price {
            None if c.unpriced => json!({ "free": false, "priced": false }),
            None => json!({ "free": true }),
            Some((price, currency, per)) => json!({ "unit": if *per == Per::Shot { "shot" } else { "job" }, "price": price, "currency": currency }),
        };
        if let Some(t) = &c.free_tier {
            billing["free_tier"] = json!(t);
        }
        let mut native: Vec<&str> = native_gates(c.basis).to_vec();
        if c.star {
            native.push("move");
        }
        let mut d = json!({
            "ok": true,
            "protocol": PROTOCOL,
            "provider": c.name,
            "device": self.device(),
            "provenance": "simulator",
            "simulator": "the wai-quantum exact simulator: the state vector in fixed point, sampled with the seed given",
            "qubits": c.qubits + u16::from(c.star),
            "native_gates": native,
            "connectivity": connectivity,
            "calibration": calibration,
            "billing": billing,
            "max_shots": MAX_SHOTS,
            "seeded": true,
        });
        if c.star {
            d["resonators"] = json!([c.qubits]);
            d["moves"] = Value::Array((0..c.qubits).map(|q| json!([q, c.qubits])).collect());
        }
        if c.landscape {
            // `protocol` is the earliest version it takes, and `protocols` every one (§11.1).
            let earliest = LANDSCAPE_PROTOCOLS.into_iter().find(|p| c.landscape_protocols.iter().any(|x| x == p)).unwrap_or(LANDSCAPE_PROTOCOLS[0]);
            d["landscape"] = json!({ "protocol": earliest, "protocols": c.landscape_protocols, "ops": LANDSCAPE_OPS, "members": [{ "id": "cpu:reference", "class": "cpu" }] });
        }
        if c.qubits == 0 {
            d["native_gates"] = json!([]);
            d["provenance"] = json!("emulator");
            d.as_object_mut().expect("an object").remove("simulator");
            d["emulator"] = json!("the energy-landscape extension's CPU reference: the settling model's kernels, in software, on this machine");
        }
        d
    }

    /// An energy-landscape job's `estimate` or `submit` (§11.2): read as the extension
    /// reads it, its bounds checked, and its operations and memory counted; then, for a
    /// `submit`, run on the extension's CPU reference.
    fn landscape(&self, m: &Value, submit: bool) -> Value {
        use wai_quantum::engine::landscape as el;
        let (op, request) = match el::read_message(m) {
            Ok(x) => x,
            Err((code, why)) => return error(code, why),
        };
        if !self.config.landscape {
            return error("unsupported", "this device takes no energy-landscape job: its description offers none");
        }
        // The version the request names, the first where it names none: one the reader
        // does not read is refused in its sentence, and one this device's description
        // does not list is unsupported (quantum-providers §11.2).
        let version = match wai_quantum::quantum_landscape::job::version_of(&request) {
            Ok(v) => v.protocol(),
            Err(r) => return el::refused(&r),
        };
        if !self.config.landscape_protocols.iter().any(|p| p == version) {
            return error("unsupported", format!("this device takes energy-landscape jobs of {}, as its description says, and this job is of {version}", self.config.landscape_protocols.join(" and ")));
        }
        let (operations, bytes) = match el::counted(op, &request, wai_quantum::quantum_landscape::MAX_JOB_BYTES) {
            Ok(x) => x,
            Err(r) => return el::refused(&r),
        };
        let cost = self.config.price.as_ref().map(|(price, currency, _)| (price.clone(), currency.clone()));
        if !submit {
            let mut a = match &cost {
                None if self.config.unpriced => json!({ "ok": true, "free": false, "priced": false }),
                None => json!({ "ok": true, "free": true }),
                Some((amount, currency)) => json!({ "ok": true, "free": false, "amount": amount, "currency": currency, "unit": "job" }),
            };
            a["seconds"] = json!(0);
            a["operations"] = json!(operations);
            a["bytes"] = json!(bytes);
            return a;
        }
        let answer = match wai_quantum::quantum_landscape::job::call(op, &request) {
            Ok(a) => a,
            Err(r) => return el::refused(&r),
        };
        let id = format!("ref-{}", self.next.fetch_add(1, Ordering::Relaxed));
        self.jobs.lock().unwrap_or_else(|p| p.into_inner()).insert(id.clone(), Job { landscape: Some(answer), counts: BTreeMap::new(), shots: 0, polls_left: self.config.polls, cancelled: false, charged: cost.map(|c| c.0) });
        json!({ "ok": true, "job": id })
    }

    /// The program in `m` and its shots, checked against the device.
    fn program_of(&self, m: &Value) -> Result<(WireProgram, u64), Value> {
        let p = program_from_json(m.get("program").unwrap_or(&Value::Null)).map_err(|e| error("invalid", e))?;
        let shots = m.get("shots").and_then(Value::as_u64).ok_or_else(|| error("invalid", "`shots` is a whole number"))?;
        if shots == 0 || shots > MAX_SHOTS {
            return Err(error("invalid", format!("{shots} shots; this device takes 1 to {MAX_SHOTS}")));
        }
        if p.basis != self.config.basis {
            return Err(error("refused", format!("this device runs {}, not {}", native_gates(self.config.basis).join(" "), native_gates(p.basis).join(" "))));
        }
        for (i, g) in p.gates.iter().enumerate() {
            let ok = match *g {
                NativeGate::Cz(a, b) | NativeGate::Zz(a, b, _) | NativeGate::Xx(a, b, _) | NativeGate::Ms(a, b, ..) => self.connected(a, b),
                NativeGate::Move(q, r) => self.config.star && r == self.config.qubits && q < r,
                NativeGate::Rz(q, _) | NativeGate::Sx(q) | NativeGate::X(q) | NativeGate::Prx(q, _, _) | NativeGate::Gpi(q, _) | NativeGate::Gpi2(q, _) => q < self.config.qubits,
            };
            if !ok {
                return Err(error("refused", format!("gate {i} is on qubits this device does not have, or on a pair it does not connect")));
            }
        }
        // A resonator used out of turn: a move it cannot make, or a state left in it.
        if self.config.star
            && let Err(e) = resonator_discipline(&p.gates, &self.coupling())
        {
            return Err(error("refused", e.to_string()));
        }
        if let Some(&q) = p.measure.iter().find(|&&q| q >= self.config.qubits) {
            return Err(error("refused", format!("qubit {q} is not one of this device's {}", self.config.qubits)));
        }
        Ok((p, shots))
    }

    /// The cost of `shots` shots, or `None` when free.
    fn cost(&self, shots: u64) -> Option<(String, String)> {
        let (price, currency, per) = self.config.price.as_ref()?;
        let amount = match per {
            Per::Shot => decimal_times(price, shots)?,
            Per::Job => price.clone(),
        };
        Some((amount, currency.clone()))
    }

    /// `shots` shots of `p` with `seed`: counts keyed by the program's bits,
    /// the highest first.
    pub fn simulate(&self, p: &WireProgram, shots: u64, seed: u64) -> Result<BTreeMap<String, u64>, String> {
        let mut used: Vec<u16> = p.measure.clone();
        for g in &p.gates {
            let (qs, k) = g.qubits();
            used.extend_from_slice(&qs[..k]);
        }
        used.sort_unstable();
        used.dedup();
        if used.len() > usize::from(MAX_QUBITS) {
            return Err(format!("the program touches {} qubits; the simulator holds {MAX_QUBITS}", used.len()));
        }
        let idx = |q: u16| used.binary_search(&q).expect("every touched qubit is listed");
        // The gates the reader takes, each native gate by its definition
        // (wai/extensions/quantum-providers.md §4).
        let phased = |i: usize, theta: Dyadic, phi: Dyadic| format!("rz({}) q[{i}];\nrx({}) q[{i}];\nrz({}) q[{i}];\n", (-phi).qasm(), theta.qasm(), phi.qasm());
        let pi = Dyadic::new(1, 1).expect("a half turn");
        let half_pi = Dyadic::new(1, 2).expect("a quarter turn");
        let mut text = format!("OPENQASM 3.0;\nqubit[{}] q;\n", used.len());
        for g in &p.gates {
            text.push_str(&match *g {
                NativeGate::Rz(q, a) => format!("rz({}) q[{}];\n", a.qasm(), idx(q)),
                NativeGate::Sx(q) => format!("sx q[{}];\n", idx(q)),
                NativeGate::X(q) => format!("x q[{}];\n", idx(q)),
                NativeGate::Prx(q, t, f) => phased(idx(q), t, f),
                NativeGate::Gpi(q, f) => phased(idx(q), pi, f),
                NativeGate::Gpi2(q, f) => phased(idx(q), half_pi, f),
                NativeGate::Cz(a, b) => format!("cz q[{}],q[{}];\n", idx(a), idx(b)),
                NativeGate::Zz(a, b, t) => {
                    let (a, b) = (idx(a), idx(b));
                    format!("cx q[{a}],q[{b}];\nrz({}) q[{b}];\ncx q[{a}],q[{b}];\n", t.qasm())
                }
                // exp(−iπθ·X⊗X) = rxx(2πθ).
                NativeGate::Xx(a, b, t) => format!("rxx({}) q[{}],q[{}];\n", t.qasm(), idx(a), idx(b)),
                // ms(φ0, φ1, θ): rxx(2πθ) between each qubit's frames.
                NativeGate::Ms(a, b, p0, p1, t) => wai_quantum::quantum_toolchain::native::ms_text(idx(a), idx(b), p0, p1, t),
                // The exchange: what a move is on every state a program in turn reaches.
                NativeGate::Move(q, r) => format!("swap q[{}],q[{}];\n", idx(q), idx(r)),
            });
        }
        let c = from_qasm(&text).map_err(|e| format!("the simulator cannot read the program: {e}"))?;
        let per_index = c.simulate().map_err(|e| e.to_string())?.sample_shots(seed, shots);
        let bits = usize::from(p.bits);
        let reads: Vec<usize> = p.measure.iter().map(|&q| idx(q)).collect();
        let mut flips = seed ^ 0x5265_6164_6f75_7421;
        let threshold = (self.config.readout_error * (1u64 << 53) as f64) as u64;
        let mut counts: BTreeMap<String, u64> = BTreeMap::new();
        for (x, &k) in per_index.iter().enumerate() {
            if k == 0 {
                continue;
            }
            let word: Vec<u8> = (0..bits).map(|i| ((x >> reads[i]) & 1) as u8).collect();
            if threshold == 0 {
                let key: String = (0..bits).rev().map(|i| char::from(b'0' + word[i])).collect();
                *counts.entry(key).or_insert(0) += k;
                continue;
            }
            for _ in 0..k {
                let key: String = (0..bits).rev().map(|i| char::from(b'0' + (word[i] ^ u8::from(splitmix64(&mut flips) >> 11 < threshold)))).collect();
                *counts.entry(key).or_insert(0) += 1;
            }
        }
        Ok(counts)
    }

    /// The answer to one message, `credential` the one it carried.
    pub fn handle(&self, m: &Value, credential: Option<&str>) -> Value {
        if let Some(want) = &self.config.credential_blake3 {
            match credential {
                Some(c) if blake3::hash(c.as_bytes()).as_bytes() == want => {}
                _ => return error("unauthorized", "this provider takes a credential, and the message carries none or another"),
            }
        }
        match m.get("protocol").and_then(Value::as_str) {
            Some(PROTOCOL) => {}
            Some(other) => return error("unsupported_version", format!("this provider speaks {PROTOCOL}, not {other}")),
            None => return error("invalid", "a message names its protocol"),
        }
        let job_id = || m.get("job").and_then(Value::as_str).map(String::from).ok_or_else(|| error("invalid", "name the `job`"));
        match m.get("op").and_then(Value::as_str).unwrap_or("") {
            "describe" => self.describe(),
            "estimate" | "submit" if m.get("landscape").is_some() => self.landscape(m, m["op"] == "submit"),
            "estimate" => match self.program_of(m) {
                Ok((_, shots)) => match self.cost(shots) {
                    // The unpriced form (section 5.2): the program checked as `submit`
                    // checks it, and no amount.
                    None if self.config.unpriced => json!({ "ok": true, "free": false, "priced": false, "seconds": 0 }),
                    None => json!({ "ok": true, "free": true, "seconds": 0 }),
                    Some((amount, currency)) => json!({
                        "ok": true, "free": false, "amount": amount, "currency": currency,
                        "unit": if self.config.price.as_ref().is_some_and(|p| p.2 == Per::Shot) { "shot" } else { "job" },
                        "seconds": 0,
                    }),
                },
                Err(e) => e,
            },
            "submit" => {
                let (p, shots) = match self.program_of(m) {
                    Ok(x) => x,
                    Err(e) => return e,
                };
                let seed = m.get("seed").and_then(Value::as_u64).unwrap_or(1);
                let counts = match self.simulate(&p, shots, seed) {
                    Ok(c) => c,
                    Err(e) => return error("refused", e),
                };
                let id = format!("ref-{}", self.next.fetch_add(1, Ordering::Relaxed));
                let charged = self.cost(shots).map(|(a, _)| a);
                self.jobs.lock().unwrap_or_else(|p| p.into_inner()).insert(id.clone(), Job { landscape: None, counts, shots, polls_left: self.config.polls, cancelled: false, charged });
                json!({ "ok": true, "job": id })
            }
            op @ ("status" | "result" | "cancel") => {
                let id = match job_id() {
                    Ok(id) => id,
                    Err(e) => return e,
                };
                let mut jobs = self.jobs.lock().unwrap_or_else(|p| p.into_inner());
                let Some(job) = jobs.get_mut(&id) else { return error("not_found", format!("no job {id}")) };
                match op {
                    "status" => {
                        if job.cancelled {
                            json!({ "ok": true, "status": "cancelled" })
                        } else if job.polls_left > 0 {
                            job.polls_left -= 1;
                            json!({ "ok": true, "status": "running" })
                        } else {
                            json!({ "ok": true, "status": "done" })
                        }
                    }
                    "result" => {
                        if job.cancelled {
                            return error("refused", format!("job {id} was cancelled"));
                        }
                        if job.polls_left > 0 {
                            return error("refused", format!("job {id} has not finished"));
                        }
                        // An energy-landscape job's answer, as the extension gives it (§11.2).
                        if let Some(answer) = &job.landscape {
                            let mut a = json!({ "ok": true, "landscape": answer });
                            if let (Some(amount), Some((_, currency, _))) = (&job.charged, &self.config.price) {
                                a["charged"] = json!({ "amount": amount, "currency": currency });
                            }
                            return a;
                        }
                        // What a provider states of its counts and of how it ran the
                        // program (§5.5): shots drawn one by one, and the gates
                        // simulated as written, with nothing a device would add.
                        let mut a = json!({
                            "ok": true,
                            "counts": job.counts,
                            "shots": job.shots,
                            "calibration": self.calibration_id(),
                            "counts_source": "shots",
                            "execution": {
                                "mitigation": "none", "twirling": "none", "decoupling": "none", "remapping": "none", "optimisation": "none",
                                "verbatim": true,
                                "note": "the reference simulates the program's gates as written, on the qubits named, and corrects nothing"
                            },
                        });
                        if let (Some(amount), Some((_, currency, _))) = (&job.charged, &self.config.price) {
                            a["charged"] = json!({ "amount": amount, "currency": currency });
                        }
                        a
                    }
                    _ => {
                        if job.polls_left == 0 && !job.cancelled {
                            json!({ "ok": true, "status": "done" })
                        } else {
                            job.cancelled = true;
                            json!({ "ok": true, "status": "cancelled" })
                        }
                    }
                }
            }
            "" => error("invalid", "a message names its `op`"),
            other => error("unsupported", format!("no operation {other:?}; the operations are describe, estimate, submit, status, result and cancel")),
        }
    }
}

/// Serves the protocol on `input` and `output`, one message and one answer a
/// line, until `input` ends. `credential` is the one the process was given
/// (the [`CREDENTIAL_ENV`] variable).
pub fn serve_stdio(r: &Reference, input: impl BufRead, mut output: impl Write, credential: Option<&str>) -> std::io::Result<()> {
    for line in input.lines() {
        let line = line?;
        if line.trim().is_empty() {
            continue;
        }
        let reply = match serde_json::from_str::<Value>(&line) {
            Ok(m) => r.handle(&m, credential),
            Err(e) => error("invalid", format!("not JSON: {e}")),
        };
        writeln!(output, "{reply}")?;
        output.flush()?;
    }
    Ok(())
}

/// A line of at most `max` bytes.
fn bounded_line(r: &mut impl BufRead, max: u64) -> std::io::Result<String> {
    let mut s = String::new();
    r.by_ref().take(max).read_line(&mut s)?;
    Ok(s)
}

fn respond(mut stream: &TcpStream, status: &str, body: &Value) -> std::io::Result<()> {
    let b = body.to_string();
    write!(stream, "HTTP/1.1 {status}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{b}", b.len())?;
    stream.flush()
}

fn serve_connection(r: &Reference, stream: TcpStream) -> std::io::Result<()> {
    stream.set_read_timeout(Some(Duration::from_secs(30)))?;
    let mut reader = BufReader::new(stream.try_clone()?);
    let request_line = bounded_line(&mut reader, 8192)?;
    let method = request_line.split_whitespace().next().unwrap_or("").to_string();
    let (mut length, mut credential) = (None::<usize>, None::<String>);
    for _ in 0..100 {
        let h = bounded_line(&mut reader, 8192)?;
        let h = h.trim_end();
        if h.is_empty() {
            break;
        }
        if let Some((name, value)) = h.split_once(':') {
            let value = value.trim();
            if name.eq_ignore_ascii_case("content-length") {
                length = value.parse().ok();
            } else if name.eq_ignore_ascii_case("authorization") {
                credential = value.strip_prefix("Bearer ").map(String::from);
            }
        }
    }
    if method != "POST" {
        return respond(&stream, "405 Method Not Allowed", &error("invalid", "the protocol's messages are POSTed"));
    }
    let Some(length) = length.filter(|&l| l <= MAX_BODY) else { return respond(&stream, "413 Content Too Large", &error("invalid", format!("a message has a Content-Length of at most {MAX_BODY} bytes"))) };
    let mut body = vec![0u8; length];
    reader.read_exact(&mut body)?;
    let reply = match serde_json::from_slice::<Value>(&body) {
        Ok(m) => r.handle(&m, credential.as_deref()),
        Err(e) => error("invalid", format!("not JSON: {e}")),
    };
    respond(&stream, "200 OK", &reply)
}

/// Serves the protocol over HTTP on `listener`, each connection on a thread
/// of its own, until the process ends.
pub fn serve_http(r: Arc<Reference>, listener: TcpListener) -> std::io::Result<()> {
    for stream in listener.incoming() {
        let Ok(stream) = stream else { continue };
        let r = r.clone();
        std::thread::spawn(move || {
            let _ = serve_connection(&r, stream);
        });
    }
    Ok(())
}

/// [`serve_http`] on `addr` in a thread of its own: the address it listens on.
pub fn spawn_http(r: Arc<Reference>, addr: &str) -> std::io::Result<SocketAddr> {
    let listener = TcpListener::bind(addr)?;
    let at = listener.local_addr()?;
    std::thread::spawn(move || serve_http(r, listener));
    Ok(at)
}

/// The command's usage.
pub const USAGE: &str = "usage: reference --stdio | --http ADDR [--qubits N] [--basis rzsxxcz|prxcz|gpizz|prxrzxx|gpims] [--all-to-all | --star] [--price AMOUNT --currency CODE [--per shot|job] | --unpriced] [--free-tier TEXT] [--polls N] [--readout-error P] [--credential-blake3 HEX] [--name NAME] [--no-landscape | --landscape-protocol 1|2 ...]";

/// The versions of the landscape protocol the reference takes unless told otherwise:
/// both, the earliest first (§11.1).
pub const LANDSCAPE_PROTOCOLS: [&str; 2] = [wai_quantum::quantum_landscape::PROTOCOL, wai_quantum::quantum_landscape::PROTOCOL_2];

/// The energy-landscape operations the reference takes: all four (§11.1).
pub const LANDSCAPE_OPS: [&str; 4] = wai_quantum::engine::landscape::OPS;

/// How the command serves.
#[derive(Clone, Debug, PartialEq)]
pub enum Serve {
    Stdio,
    Http(String),
}

/// The command's arguments read: how to serve, and the device.
pub fn parse_args(args: &[String]) -> Result<(Serve, Config), String> {
    let mut serve = None;
    let mut c = Config::default();
    let (mut price, mut currency, mut per) = (None, None, Per::Shot);
    let mut versions: Vec<String> = Vec::new();
    let mut it = args.iter();
    while let Some(flag) = it.next() {
        let mut value = || it.next().cloned().ok_or_else(|| format!("{flag} needs a value\n{USAGE}"));
        match flag.as_str() {
            "--stdio" => serve = Some(Serve::Stdio),
            "--http" => serve = Some(Serve::Http(value()?)),
            "--qubits" => c.qubits = value()?.parse().map_err(|_| "--qubits takes a number")?,
            "--basis" => c.basis = crate::open_provider::basis_named(&value()?).ok_or("--basis is rzsxxcz, prxcz, gpizz, prxrzxx or gpims")?,
            "--all-to-all" => c.all_to_all = true,
            "--star" => c.star = true,
            "--price" => price = Some(value()?),
            "--currency" => currency = Some(value()?),
            "--per" => {
                per = match value()?.as_str() {
                    "shot" => Per::Shot,
                    "job" => Per::Job,
                    _ => return Err("--per is shot or job".into()),
                }
            }
            "--free-tier" => c.free_tier = Some(value()?),
            "--unpriced" => c.unpriced = true,
            "--polls" => c.polls = value()?.parse().map_err(|_| "--polls takes a number")?,
            "--readout-error" => c.readout_error = value()?.parse().map_err(|_| "--readout-error takes a probability")?,
            "--credential-blake3" => c.credential_blake3 = Some(unhex32(&value()?).ok_or("--credential-blake3 takes 64 hex digits")?),
            "--name" => c.name = value()?,
            "--no-landscape" => c.landscape = false,
            "--landscape-protocol" => {
                let named = match value()?.as_str() {
                    "1" | wai_quantum::quantum_landscape::PROTOCOL => LANDSCAPE_PROTOCOLS[0],
                    "2" | wai_quantum::quantum_landscape::PROTOCOL_2 => LANDSCAPE_PROTOCOLS[1],
                    _ => return Err("--landscape-protocol is 1 or 2".into()),
                };
                if !versions.iter().any(|v| v == named) {
                    versions.push(named.to_string());
                }
            }
            other => return Err(format!("unknown flag {other}\n{USAGE}")),
        }
    }
    if !versions.is_empty() {
        c.landscape_protocols = versions;
    }
    match (price, currency) {
        (Some(p), Some(cur)) => c.price = Some((p, cur, per)),
        (None, None) => {}
        _ => return Err("--price and --currency go together".into()),
    }
    Ok((serve.ok_or_else(|| format!("give --stdio or --http ADDR\n{USAGE}"))?, c))
}

/// The command: serve the reference provider as `args` say, until standard
/// input ends (`--stdio`) or the process is ended (`--http`, which first
/// prints `listening http://ADDR`).
pub fn main(args: &[String]) -> Result<(), String> {
    let (serve, config) = parse_args(args)?;
    let r = Reference::new(config)?;
    match serve {
        Serve::Stdio => {
            let credential = std::env::var(CREDENTIAL_ENV).ok();
            serve_stdio(&r, std::io::stdin().lock(), std::io::stdout().lock(), credential.as_deref()).map_err(|e| e.to_string())
        }
        Serve::Http(addr) => {
            let listener = TcpListener::bind(&addr).map_err(|e| format!("{addr}: {e}"))?;
            let at = listener.local_addr().map_err(|e| e.to_string())?;
            println!("listening http://{at}");
            std::io::stdout().flush().map_err(|e| e.to_string())?;
            serve_http(Arc::new(r), listener).map_err(|e| e.to_string())
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

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

    #[test]
    fn decimals_multiply_exactly() {
        assert_eq!(decimal_times("0.0003", 1000).as_deref(), Some("0.3"));
        assert_eq!(decimal_times("1.25", 4).as_deref(), Some("5"));
        assert_eq!(decimal_times("7", 3).as_deref(), Some("21"));
        assert_eq!(decimal_times("0.001", 1).as_deref(), Some("0.001"));
        assert_eq!(decimal_times("0.10", 0).as_deref(), Some("0"));
        assert_eq!(decimal_times("x", 1), None);
    }

    #[test]
    fn the_device_is_described_and_a_bad_message_is_refused_with_its_code() {
        let r = Reference::new(Config::default()).unwrap();
        let d = r.handle(&msg("describe", json!({})), None);
        assert_eq!((d["ok"].as_bool(), d["qubits"].as_u64(), d["billing"]["free"].as_bool()), (Some(true), Some(5), Some(true)), "{d}");
        assert_eq!(d["connectivity"], json!([[0, 1], [1, 2], [2, 3], [3, 4]]));
        let code = |m: Value| r.handle(&m, None)["error"]["code"].as_str().unwrap_or("").to_string();
        assert_eq!(code(json!({ "op": "describe" })), "invalid");
        assert_eq!(code(json!({ "protocol": "wai.quantum.provider/2", "op": "describe" })), "unsupported_version");
        assert_eq!(code(msg("teleport", json!({}))), "unsupported");
        assert_eq!(code(msg("status", json!({ "job": "nope" }))), "not_found");
        assert_eq!(code(msg("submit", json!({ "program": {}, "shots": 1 }))), "invalid");
    }

    #[test]
    fn a_credential_is_checked_by_its_hash() {
        let r = Reference::new(Config { credential_blake3: unhex32(&credential_hash("s3cret-value")), ..Config::default() }).unwrap();
        assert_eq!(r.handle(&msg("describe", json!({})), None)["error"]["code"], "unauthorized");
        assert_eq!(r.handle(&msg("describe", json!({})), Some("other"))["error"]["code"], "unauthorized");
        assert_eq!(r.handle(&msg("describe", json!({})), Some("s3cret-value"))["ok"], true);
    }

    #[test]
    fn the_arguments_read_and_a_bad_one_is_named() {
        let a = |s: &str| parse_args(&s.split_whitespace().map(String::from).collect::<Vec<_>>());
        let (s, c) = a("--http 127.0.0.1:0 --qubits 3 --basis gpizz --all-to-all --price 0.01 --currency USD --per job --polls 2").unwrap();
        assert_eq!((s, c.qubits, c.basis, c.all_to_all, c.price, c.polls), (Serve::Http("127.0.0.1:0".into()), 3, Basis::GpiZz, true, Some(("0.01".into(), "USD".into(), Per::Job)), 2));
        assert!(a("--qubits 3").unwrap_err().contains("--stdio"));
        assert!(a("--stdio --price 1").unwrap_err().contains("together"));
        assert!(a("--stdio --bogus").unwrap_err().contains("unknown flag"));
        assert!(Reference::new(Config { qubits: 0, landscape: false, ..Config::default() }).is_err());
        // With no qubits it takes energy-landscape jobs alone, and says so.
        let alone = Reference::new(Config { qubits: 0, ..Config::default() }).unwrap();
        let d = alone.handle(&msg("describe", json!({})), None);
        assert_eq!((d["qubits"].as_u64(), d["native_gates"].clone(), d["connectivity"].clone(), d["provenance"].as_str()), (Some(0), json!([]), json!([]), Some("emulator")), "{d}");
        assert_eq!(d["landscape"]["ops"], json!(["run", "sweep", "compare", "replay"]));
        assert_eq!((d["landscape"]["protocol"].clone(), d["landscape"]["protocols"].clone()), (json!(LANDSCAPE_PROTOCOLS[0]), json!(LANDSCAPE_PROTOCOLS)));
        let (_, c) = a("--stdio --no-landscape").unwrap();
        assert!(!c.landscape);
        // One version alone, by its number or its name: its description lists it, a job of
        // the other is unsupported, and a protocol the reader does not read is refused.
        let second = json!({ "landscape": { "op": "run", "request": { "program": "vars 3\nexactly 1 0 1 2\nsolve\n", "protocol": LANDSCAPE_PROTOCOLS[1] } } });
        let first = json!({ "landscape": { "op": "run", "request": { "program": "vars 2\nagree 0 1\nminimize\n" } } });
        let (_, c) = a("--stdio --landscape-protocol 1 --landscape-protocol 1").unwrap();
        assert_eq!(c.landscape_protocols, [LANDSCAPE_PROTOCOLS[0]]);
        let one = Reference::new(c).unwrap();
        assert_eq!(one.describe()["landscape"]["protocols"], json!([LANDSCAPE_PROTOCOLS[0]]));
        assert_eq!((one.handle(&msg("estimate", second.clone()), None)["error"]["code"].as_str(), one.handle(&msg("estimate", first.clone()), None)["ok"].as_bool()), (Some("unsupported"), Some(true)));
        assert_eq!(one.handle(&msg("submit", second.clone()), None)["error"]["code"], "unsupported");
        let (_, c) = a("--stdio --landscape-protocol wai.quantum.landscape/2").unwrap();
        let two = Reference::new(c).unwrap();
        assert_eq!((two.describe()["landscape"]["protocol"].clone(), two.describe()["landscape"]["protocols"].clone()), (json!(LANDSCAPE_PROTOCOLS[1]), json!([LANDSCAPE_PROTOCOLS[1]])));
        assert_eq!((two.handle(&msg("estimate", second.clone()), None)["ok"].as_bool(), two.handle(&msg("estimate", first), None)["error"]["code"].as_str()), (Some(true), Some("unsupported")));
        assert!(a("--stdio --landscape-protocol 3").unwrap_err().contains("1 or 2"));
        let mut other = second;
        other["landscape"]["request"]["protocol"] = json!("another/9");
        assert_eq!(two.handle(&msg("estimate", other), None)["error"]["code"], "refused");
        assert!(Reference::new(Config { price: Some(("1e3".into(), "USD".into(), Per::Shot)), ..Config::default() }).is_err());
        // Not free and no price set: its billing and its estimates say so, and a price
        // beside it is refused.
        let (_, c) = a("--stdio --unpriced").unwrap();
        let unpriced = Reference::new(c).unwrap();
        assert_eq!(unpriced.describe()["billing"], json!({ "free": false, "priced": false }));
        let program = json!({ "format": "wai.native/1", "basis": "rzsxxcz", "bits": 1, "gates": [{ "gate": "x", "qubits": [0] }], "measure": [{ "qubit": 0, "bit": 0 }] });
        assert_eq!(unpriced.handle(&msg("estimate", json!({ "program": program, "shots": 10 })), None), json!({ "ok": true, "free": false, "priced": false, "seconds": 0 }));
        assert_eq!(unpriced.handle(&msg("estimate", json!({ "program": program, "shots": 0 })), None)["error"]["code"], "invalid");
        let l = unpriced.handle(&msg("estimate", json!({ "landscape": { "op": "run", "request": { "program": "vars 2\nagree 0 1\nminimize\n" } } })), None);
        assert!(l["free"] == false && l["priced"] == false && l["operations"].is_u64() && l["bytes"].is_u64() && l.get("amount").is_none(), "{l}");
        assert!(a("--stdio --unpriced --price 1 --currency USD").is_ok_and(|(_, c)| Reference::new(c).is_err()));
    }
}