//! The third-party hop record (extensions/route-record.md §6.2): what a router sent a third
//! party for one hop, what came back, and what the provider bills, signed by the router and
//! bound to the hop's attempt and receipt.
//!
//! It holds no secret. The key the hop ran on is never stated, only whose account it is; an
//! exchange has no member for a header, since a key travels in one; and a URL is stated without
//! its query, since a provider's signed download link is a bearer token. [`url_without_query`]
//! and [`kept_response`] make a URL and a response what the record may state.

use ed25519_dalek::{Signer, SigningKey};
use serde_json::{json, Value};

use super::{at, canonical, digest, exactly, hex, int, output_commitment, read_document, receipt_id, tok, unhex, unit, Reason, Refusal, ERRORS, MAX_OUTPUTS};

pub const KIND: &str = "wai-route-third-party-hop";
pub const VERSION: u64 = 1;
/// The domain a record is signed under (§6.2).
pub const SIGN_DOMAIN: &[u8] = b"wai:route-third-party-hop\x01";
/// The most exchanges a record states.
pub const MAX_EXCHANGES: usize = 1024;
/// The hop ran on the router's operator's own account with the provider.
pub const OPERATOR: &str = "operator";
/// The hop ran on the account of the job's owner.
pub const CUSTOMER: &str = "customer";

/// The members of a record (§6.2).
pub const MEMBERS: [&str; 20] = [
    "kind", "version", "job", "hop", "endpoint", "provider", "receipt", "origin", "account", "provider_job", "outcome", "error", "exchanges", "outputs", "usage", "charge", "start",
    "end", "signer", "sig",
];
/// The members of an exchange (§6.2).
pub const EXCHANGE_MEMBERS: [&str; 8] = ["method", "url", "start", "end", "sent", "status", "received", "json"];

/// What a verification reads of a record that holds (§7.3).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Read {
    pub hop: u64,
    pub digest: [u8; 32],
    /// [`OPERATOR`] or [`CUSTOMER`].
    pub account: String,
    /// Its `usage`: `null` or the object.
    pub usage: Value,
    /// Its `charge`: `null` or the object.
    pub charge: Value,
}

/// A URL as a record states it: cut at its first `?` or `#`.
pub fn url_without_query(url: &str) -> &str {
    url.split(['?', '#']).next().unwrap_or("")
}

fn is_url(s: &str) -> bool {
    s.starts_with("https://") || s.starts_with("http://")
}

/// A JSON response as a record keeps it (§6.2): each number that is not an integer 0 to 2⁵³ − 1
/// written as its decimal text, and each string that begins `https://` or `http://` cut at its
/// first `?` or `#`, a member's name among them.
pub fn kept_response(v: &Value) -> Value {
    let kept = |s: &str| if is_url(s) { url_without_query(s).to_string() } else { s.to_string() };
    match v {
        Value::String(s) => Value::String(kept(s)),
        Value::Number(n) if n.as_u64().is_none_or(|u| u > crate::jcs::MAX_INTEGER) => Value::String(n.to_string()),
        Value::Array(a) => Value::Array(a.iter().map(kept_response).collect()),
        Value::Object(o) => Value::Object(o.iter().map(|(k, v)| (kept(k), kept_response(v))).collect()),
        other => other.clone(),
    }
}

/// The bytes a record is signed over (§6.2): the domain, then the canonical form of the record
/// without `sig`.
pub fn signing_bytes(record: &Value) -> Vec<u8> {
    let mut r = record.clone();
    if let Some(o) = r.as_object_mut() {
        o.remove("sig");
    }
    let mut m = SIGN_DOMAIN.to_vec();
    m.extend_from_slice(canonical(&r).as_bytes());
    m
}

/// Sign a record body with the router's `key`: sets `kind`, `version`, `signer` and `sig`.
pub fn sign(mut body: Value, key: &SigningKey) -> Value {
    let o = body.as_object_mut().expect("a record is an object");
    o.insert("kind".into(), json!(KIND));
    o.insert("version".into(), json!(VERSION));
    o.insert("signer".into(), json!(hex(&key.verifying_key().to_bytes())));
    o.insert("sig".into(), json!(""));
    let sig = key.sign(&signing_bytes(&body));
    body["sig"] = json!(hex(&sig.to_bytes()));
    body
}

/// Sign a record again over whatever it states, keeping its `signer`: for the corpus's controls.
#[doc(hidden)]
pub fn resign(mut body: Value, key: &SigningKey) -> Value {
    body["sig"] = json!("");
    let sig = key.sign(&signing_bytes(&body));
    body["sig"] = json!(hex(&sig.to_bytes()));
    body
}

fn text(v: &Value, max: usize) -> bool {
    v.as_str().is_some_and(|s| !s.is_empty() && s.len() <= max)
}

fn lowercase_hex(s: &str, digits: usize) -> bool {
    s.len() == digits && s.bytes().all(|c| c.is_ascii_digit() || (b'a'..=b'f').contains(&c))
}

/// `https://`, then a host, then optionally `:` and a port, and nothing else.
fn origin(s: &str) -> bool {
    let Some(rest) = s.strip_prefix("https://") else { return false };
    let (host, port) = match rest.split_once(':') {
        Some((h, p)) => (h, Some(p)),
        None => (rest, None),
    };
    !host.is_empty()
        && host.len() <= 253
        && host.bytes().all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == b'.' || c == b'-')
        && port.is_none_or(|p| (1..=5).contains(&p.len()) && p.bytes().all(|c| c.is_ascii_digit()))
}

fn url(s: &str) -> bool {
    let rest = s.strip_prefix("https://").or_else(|| s.strip_prefix("http://"));
    rest.is_some_and(|r| !r.is_empty() && r.len() <= 2040) && !s.contains(['?', '#'])
}

/// Whether a kept response holds a link's query: a string, a member's name among them, that begins
/// `https://` or `http://` and holds a `?` or a `#`.
fn keeps_a_query(v: &Value) -> bool {
    let bad = |s: &str| is_url(s) && s.contains(['?', '#']);
    match v {
        Value::String(s) => bad(s),
        Value::Array(a) => a.iter().any(keeps_a_query),
        Value::Object(o) => o.iter().any(|(k, v)| bad(k) || keeps_a_query(v)),
        _ => false,
    }
}

/// `sha256` of no bytes.
const NOTHING: &str = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";

fn body_form(v: &Value) -> Option<(&str, u64)> {
    let o = exactly(v, &["sha256", "bytes"])?;
    let d = o["sha256"].as_str().filter(|s| lowercase_hex(s, 64))?;
    Some((d, int(&o["bytes"])?))
}

/// An exchange of §6.2, after `previous`, the start of the one before it, or the record's start.
fn exchange_form(x: &Value, previous: u64, to: u64) -> Option<u64> {
    let o = exactly(x, &EXCHANGE_MEMBERS)?;
    let m = o["method"].as_str()?;
    if m.is_empty() || m.len() > 16 || !m.bytes().all(|c| c.is_ascii_uppercase()) || !o["url"].as_str().is_some_and(url) {
        return None;
    }
    let (start, end) = (int(&o["start"])?, int(&o["end"])?);
    if start > end || end > to || start < previous {
        return None;
    }
    body_form(&o["sent"])?;
    let status = int(&o["status"])?;
    if status != 0 && !(100..=599).contains(&status) {
        return None;
    }
    let (received, bytes) = body_form(&o["received"])?;
    if status == 0 && (received != NOTHING || bytes != 0 || !o["json"].is_null()) {
        return None;
    }
    if keeps_a_query(&o["json"]) {
        return None;
    }
    Some(start)
}

fn usage_form(v: &Value) -> bool {
    exactly(v, &["unit", "quantity", "source"]).is_some_and(|o| o["unit"].as_str().is_some_and(unit) && int(&o["quantity"]).is_some() && text(&o["source"], 512))
}

fn charge_form(v: &Value) -> bool {
    exactly(v, &["amount", "unit", "source"]).is_some_and(|o| text(&o["amount"], 64) && text(&o["unit"], 64) && text(&o["source"], 512))
}

fn outputs_form(v: &Value) -> Option<Vec<[u8; 32]>> {
    let a = v.as_array()?;
    let mut names = std::collections::BTreeSet::new();
    let mut hashes = Vec::with_capacity(a.len());
    for e in a {
        let pair = e.as_array().filter(|p| p.len() == 2)?;
        let (name, h) = (pair[0].as_str()?, pair[1].as_str()?);
        if name.is_empty() || name.len() > 255 || name.contains('/') || !names.insert(name) {
            return None;
        }
        let h: [u8; 32] = unhex(h)?;
        if hashes.contains(&h) {
            return None;
        }
        hashes.push(h);
    }
    Some(hashes)
}

/// A record of §6.2's members and forms, a kept response holding no link's query.
pub fn form(r: &Value) -> bool {
    let Some(o) = exactly(r, &MEMBERS) else { return false };
    let (Some(start), Some(end)) = (int(&o["start"]), int(&o["end"])) else { return false };
    let Some(outputs) = outputs_form(&o["outputs"]) else { return false };
    let shape = match o["outcome"].as_str() {
        Some("served") => o["error"].is_null() && (1..=MAX_OUTPUTS).contains(&outputs.len()) && usage_form(&o["usage"]),
        Some("failed") => {
            exactly(&o["error"], &["class", "code"]).is_some_and(|e| e["class"].as_str().is_some_and(|c| ERRORS.contains(&c)) && text(&e["code"], 256))
                && outputs.is_empty()
                && (o["usage"].is_null() || usage_form(&o["usage"]))
        }
        _ => false,
    };
    let Some(exchanges) = o["exchanges"].as_array() else { return false };
    let mut previous = start;
    for x in exchanges {
        match exchange_form(x, previous, end) {
            Some(s) => previous = s,
            None => return false,
        }
    }
    // `start` at most `end` follows: an exchange lies between them, and there is one.
    shape
        && o["version"] == json!(VERSION)
        && (1..=MAX_EXCHANGES).contains(&exchanges.len())
        && tok(&o["job"]).is_some()
        && int(&o["hop"]).is_some_and(|h| h >= 1)
        && tok(&o["endpoint"]).is_some()
        && tok(&o["provider"]).is_some()
        && o["receipt"].as_str().is_some_and(receipt_id)
        && o["origin"].as_str().is_some_and(origin)
        && matches!(o["account"].as_str(), Some(OPERATOR | CUSTOMER))
        && (o["provider_job"].is_null() || text(&o["provider_job"], 256))
        && (o["charge"].is_null() || charge_form(&o["charge"]))
        && o["signer"].as_str().and_then(unhex::<32>).is_some()
        && o["sig"].as_str().and_then(unhex::<64>).is_some()
}

/// §7.2 step 9: each record read, its signature checked and held to its hop. `receipts` are the
/// hop receipts step 8 has already checked.
pub(super) fn check(rec: &Value, texts: &[&str], receipts: &[Value]) -> Result<Vec<Read>, Refusal> {
    let mut read = Vec::new();
    let mut seen = std::collections::BTreeSet::new();
    for (i, t) in texts.iter().enumerate() {
        let place = |m: Option<&str>| -> Vec<Value> {
            match m {
                Some(m) => at(&[json!("third_party"), json!(i), json!(m)]),
                None => at(&[json!("third_party"), json!(i)]),
            }
        };
        let r = read_document(t).ok_or_else(|| Refusal::new(Reason::Malformed, place(None)))?;
        if r.get("kind").and_then(Value::as_str) != Some(KIND) {
            return Err(Refusal::new(Reason::Malformed, place(None)));
        }
        if r.get("version").and_then(int).is_some_and(|v| v != VERSION) {
            return Err(Refusal::new(Reason::UnknownVersion, place(Some("version"))));
        }
        if !form(&r) {
            return Err(Refusal::new(Reason::ThirdPartyInvalid, place(None)));
        }
        let signer: [u8; 32] = unhex(r["signer"].as_str().expect("a key")).expect("a key");
        let sig: [u8; 64] = unhex(r["sig"].as_str().expect("a signature")).expect("a signature");
        crate::sig::verify_strict(&signer, &signing_bytes(&r), &sig).map_err(|e| {
            let reason = match e {
                crate::sig::SigError::BadKey => Reason::BadKey,
                crate::sig::SigError::BadSignature => Reason::BadSignature,
            };
            Refusal::new(reason, place(Some("sig")))
        })?;
        bind(rec, &r, &mut seen, &place)?;
        delivered(rec, &r, receipts).then_some(()).ok_or_else(|| Refusal::new(Reason::ThirdPartyOutputMismatch, place(Some("outputs"))))?;
        read.push(Read {
            hop: r["hop"].as_u64().expect("a hop"),
            digest: digest(&r),
            account: r["account"].as_str().expect("an account").to_string(),
            usage: r["usage"].clone(),
            charge: r["charge"].clone(),
        });
    }
    read.sort_by_key(|x| x.hop);
    Ok(read)
}

/// The attempt a record's `hop` names.
fn attempt<'a>(rec: &'a Value, hop: u64) -> Option<&'a Value> {
    let i = usize::try_from(hop.checked_sub(1)?).ok()?;
    rec["attempts"].as_array().expect("a list").get(i)
}

fn bind(rec: &Value, r: &Value, seen: &mut std::collections::BTreeSet<u64>, place: &dyn Fn(Option<&str>) -> Vec<Value>) -> Result<(), Refusal> {
    let refuse = |m: &str| Refusal::new(Reason::ThirdPartyMismatch, place(Some(m)));
    for m in ["signer", "job"] {
        if r[m] != rec[m] {
            return Err(refuse(m));
        }
    }
    let hop = r["hop"].as_u64().expect("a hop");
    let a = attempt(rec, hop).ok_or_else(|| refuse("hop"))?;
    if !seen.insert(hop) {
        return Err(refuse("hop"));
    }
    for m in ["endpoint", "receipt", "outcome"] {
        if r[m] != a[m] {
            return Err(refuse(m));
        }
    }
    if r["outcome"] == "failed" && r["error"]["class"] != a["error"] {
        return Err(refuse("error"));
    }
    let c = rec["candidates"].as_array().expect("a list").iter().find(|c| c["endpoint"] == r["endpoint"]).ok_or_else(|| refuse("endpoint"))?;
    if r["provider"] != c["provider"] {
        return Err(refuse("provider"));
    }
    if !r["usage"].is_null() && r["usage"]["unit"] != c["unit"] {
        return Err(refuse("usage"));
    }
    if r["start"].as_u64() < a["start"].as_u64() {
        return Err(refuse("start"));
    }
    if r["end"].as_u64() > a["end"].as_u64() {
        return Err(refuse("end"));
    }
    Ok(())
}

/// What a served record names is what its hop delivered (§6.2): its outputs' commitment is its
/// receipt's `subject.output`, to which §6 has held the attempt's `outputs` or `root`, and in a
/// record of version 3 their number is the attempt's `count`, which the receipt does not name. A
/// failed record names nothing.
fn delivered(rec: &Value, r: &Value, receipts: &[Value]) -> bool {
    if r["outcome"] != "served" {
        return true;
    }
    let hashes: Vec<[u8; 32]> = outputs_form(&r["outputs"]).expect("its form was read");
    let root = hex(&output_commitment(&hashes));
    let a = attempt(rec, r["hop"].as_u64().expect("a hop")).expect("its binding was read");
    let receipt = receipts.iter().find(|x| x.get("receipt_id") == Some(&a["receipt"])).expect("step 8 found it");
    receipt.get("subject").and_then(|s| s.get("output")).and_then(Value::as_str) == Some(root.as_str())
        && a.get("delivered").is_none_or(|d| d["count"].as_u64() == Some(hashes.len() as u64))
}

/// A read record's outcome (§7.3).
pub fn read_json(r: &Read) -> Value {
    json!({"hop": r.hop, "digest": hex(&r.digest), "account": r.account, "usage": r.usage, "charge": r.charge})
}
