//! `wai.delivery.revenue_share` (extensions/revenue-share.md): the committed corpus
//! `live-conformance/share/vectors.json` regenerates byte for byte from its seeds and the ledger
//! corpus, and is the only file there besides its README; every vector gets the outcome it states;
//! every reason the extension adds and every code it reuses refuses a vector; §11 lists exactly the
//! reasons the reference reports; and the worked example of §12: a channel's windows airing two
//! owners' programmes with mid-roll breaks, impressions from a room's break-form receipts and from
//! streams stitched by others, a contract, the statements, and their recomputation by a party
//! holding only the public records.
#![cfg(feature = "revenue_share")]

use std::collections::{BTreeMap, BTreeSet};
use std::path::PathBuf;

use serde_json::{Value, json};
use wai::revenue_share::{self as share, Contract, Reason, Statement, corpus};

fn wai_dir() -> PathBuf {
    PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("..")
}

fn read(p: &str) -> String {
    std::fs::read_to_string(wai_dir().join(p)).unwrap_or_else(|e| panic!("{p}: {e}"))
}

fn committed() -> String {
    read("live-conformance/share/vectors.json")
}

fn doc() -> Value {
    serde_json::from_str(&committed()).expect("the corpus is JSON")
}

#[test]
fn the_committed_corpus_regenerates_from_its_seeds_and_the_ledger_corpus() {
    let generated = corpus::generate(&wai_dir()).expect("the corpus generates");
    assert!(committed() == generated, "live-conformance/share/vectors.json differs from `wai_revenue_share_vectors gen`");
    let mut files: Vec<String> = std::fs::read_dir(wai_dir().join("live-conformance/share")).unwrap().flatten().map(|e| e.file_name().to_string_lossy().into_owned()).collect();
    files.sort();
    assert_eq!(files, ["README.md", "vectors.json"]);
}

#[test]
fn every_vector_gets_its_stated_outcome() {
    let fails = corpus::verify(&committed());
    assert!(fails.is_empty(), "{}", fails.join("\n"));
}

/// Every reason this extension adds and every code it reuses refuses a vector, and the receipt
/// ledger refuses one at its own place.
#[test]
fn the_corpus_covers_every_reason_and_case() {
    let d = doc();
    let vectors = d["vectors"].as_array().unwrap();
    let seen: BTreeSet<&str> = vectors.iter().filter_map(|v| v["outcome"]["reason"].as_str()).collect();
    for r in Reason::ALL {
        assert!(seen.contains(r.code()), "nothing in the corpus is refused {r}");
    }
    for r in Reason::REUSED {
        assert!(seen.contains(r), "nothing in the corpus is refused {r}");
    }
    let get = |n: &str| vectors.iter().find(|v| v["name"] == n).unwrap_or_else(|| panic!("no vector {n}"))["outcome"].clone();
    assert_eq!(get("ledger_refused")["at"], json!(["ledger", 0, "invoice"]), "the receipt ledger's refusal at its place");
    assert_eq!(get("other_contract_offered")["at"][0], json!("contracts"));
    for n in ["owner_a_october", "owner_b_october", "operator_october", "owner_b_following", "owner_a_between", "owner_a_nothing_billed"] {
        assert_eq!(get(n)["outcome"], "accept", "{n}");
    }
}

/// §11's table names exactly the codes the reference adds, in its order.
#[test]
fn section_11_lists_exactly_the_reference_codes() {
    let text = read("extensions/revenue-share.md");
    let at = text.find("\n## 11. Reasons").expect("§11");
    let end = text[at + 1..].find("\n## ").map_or(text.len(), |e| at + 1 + e);
    let listed: Vec<&str> = text[at..end].lines().filter_map(|l| l.strip_prefix("| `")).filter_map(|l| l.split('`').next()).collect();
    let codes: Vec<&str> = Reason::ALL.iter().map(|r| r.code()).collect();
    assert_eq!(listed, codes);
}

/// A named part of the corpus.
fn part<'a>(d: &'a Value, kind: &str, name: &str) -> &'a Value {
    d[kind].get(name).unwrap_or_else(|| panic!("no {kind} {name}"))
}

/// The inputs of October under the contract `contract`, from the corpus's named parts.
fn october(d: &Value, contract: &str) -> Value {
    let ledger = |n: &str| {
        let l = part(d, "ledgers", n);
        let sources: Vec<Value> = l["sources"].as_array().unwrap().iter().map(|s| part(d, "sources", s.as_str().unwrap()).clone()).collect();
        json!({"tariff": l["tariff"], "invoice": l["invoice"], "previous": l["previous"], "sources": sources})
    };
    json!({
        "contract": part(d, "contracts", contract),
        "contracts": [part(d, "contracts", "next_quarter")],
        "runs": [part(d, "runs", "oct_16"), part(d, "runs", "oct_25")],
        "ledgers": [ledger("day_16"), ledger("day_25")],
    })
}

fn unhex(s: &str) -> Vec<u8> {
    (0..s.len()).step_by(2).map(|i| u8::from_str_radix(&s[i..i + 2], 16).unwrap()).collect()
}

/// The worked example (§12). The operator computes and signs October's statements of owner A,
/// owner B and itself from the channel's two windows and two invoices; a party holding only the
/// public records receives each verification document as text, verifies it, and recomputes every
/// line from the attributed impressions and the contract alone.
#[test]
fn the_worked_example_recomputes_from_the_public_records() {
    let d = doc();
    let inputs = october(&d, "preceding");
    let (oct_start, oct_end) = (1_790_812_800_000_000u64, 1_793_491_200_000_000u64);
    let operator = corpus::key("operator");
    let parties = [("owner-a", corpus::key("owner-a")), ("owner-b", corpus::key("owner-b")), ("operator", operator.clone())];
    // Each split and class: the revenue each statement states, and the parts of its parties.
    let mut revenue: BTreeMap<(u16, u8), u64> = BTreeMap::new();
    let mut parts: BTreeMap<(u16, u8), u64> = BTreeMap::new();
    for (who, k) in &parties {
        let (s, document) = share::create(&inputs, &operator, k.verifying_key().to_bytes(), oct_start, oct_end).expect("the statement");
        // The committed corpus holds this statement, byte for byte.
        let name = match *who {
            "owner-a" => "owner_a_october",
            "owner-b" => "owner_b_october",
            _ => "operator_october",
        };
        let v = d["vectors"].as_array().unwrap().iter().find(|v| v["name"] == name).unwrap();
        assert_eq!(v["statement"].as_str().map(unhex), Some(s.to_bytes()), "{name}");

        // What the party receives: the document, as text. It holds no key but its own.
        let text = serde_json::to_string(&document).unwrap();
        let received: Value = serde_json::from_str(&text).unwrap();
        let a = share::verify(&received).expect("the statement verifies");
        let contract = Contract::parse(&unhex(received["contract"].as_str().unwrap())).expect("the contract is in force");
        let statement = Statement::read(&unhex(received["statement"].as_str().unwrap())).expect("the statement reads");
        assert_eq!(statement, s);

        // Recompute every line from the attributed impressions and the contract.
        let mut groups: BTreeMap<(u16, u8), (u64, u64)> = BTreeMap::new();
        for i in &a.evidence.impressions {
            let g = groups.entry((i.split, i.class)).or_default();
            g.0 += 1;
            g.1 += i.price;
        }
        let me = k.verifying_key().to_bytes();
        let mut total = 0;
        let mut lines = statement.lines.iter();
        for ((split, class), (n, rev)) in &groups {
            let ps = contract.parties(*split);
            let Some(at) = ps.iter().position(|(p, _)| *p == me) else {
                continue;
            };
            let division = share::divide(*rev, &ps.iter().map(|p| p.1).collect::<Vec<_>>());
            assert_eq!(division.iter().sum::<u64>(), *rev, "the parts of a split sum to its revenue");
            let l = lines.next().expect("a line for every split and class the party shares in");
            assert_eq!((l.split, l.class, l.impressions, l.revenue, l.ppm, l.amount), (*split, *class, *n, *rev, ps[at].1, division[at]));
            total += l.amount;
            assert_eq!(*revenue.entry((*split, *class)).or_insert(*rev), *rev, "every statement states one revenue for a split and class");
            *parts.entry((*split, *class)).or_default() += l.amount;
        }
        assert!(lines.next().is_none(), "no line the impressions do not give");
        assert_eq!(statement.total, total);

        // The join itself: each impression placed on the program-as-run.
        let assets: Vec<(Option<&str>, u8)> = a.evidence.impressions.iter().map(|i| (i.asset.as_deref(), i.class)).collect();
        use share::class::{BEACONS, PRESENTED, UNPLACED};
        let mut want = vec![(Some("a/ep-101"), BEACONS); 3];
        want.extend(vec![(Some("b/doc-7"), PRESENTED); 4]);
        // The 25th: two ads of the first record at 12:02:00 and 12:02:15 (the ad slot, by the rule
        // preceding), the second record's at 12:02:00, the form-1 stream's at 12:20:00, and three
        // presentations of DASH events, which state no time.
        want.extend([(Some("a/ep-102"), PRESENTED), (Some("a/ep-102"), PRESENTED), (Some("a/ep-102"), PRESENTED), (Some("b/doc-8"), PRESENTED)]);
        want.extend(vec![(None, UNPLACED); 3]);
        assert_eq!(assets, want);
    }
    // The three statements' parts of each split and class sum to its revenue.
    assert_eq!(parts, revenue);
    assert_eq!(revenue, BTreeMap::from([((0, 1), 3 * 1501), ((0, 2), 3 * 903), ((1, 1), 5 * 1501), ((2, 3), 3 * 1501)]));
}

/// A recomputation is not a re-assertion: every figure of a statement is held to the records.
#[test]
fn a_statement_that_states_another_figure_is_refused() {
    let d = doc();
    let inputs = october(&d, "preceding");
    let operator = corpus::key("operator");
    let (s, document) = share::create(&inputs, &operator, corpus::key("owner-a").verifying_key().to_bytes(), 1_790_812_800_000_000, 1_793_491_200_000_000).unwrap();
    for (what, f, reason) in [
        ("a revenue raised", Box::new(|s: &mut Statement| s.lines[0].revenue += 1) as Box<dyn Fn(&mut Statement)>, "share_statement_mismatch"),
        ("a ppm changed", Box::new(|s: &mut Statement| s.lines[0].ppm += 1), "share_statement_mismatch"),
        ("an invoice dropped", Box::new(|s: &mut Statement| {
            s.invoices.pop();
        }), "share_statement_mismatch"),
        ("the total raised", Box::new(|s: &mut Statement| s.total += 1), "share_statement_mismatch"),
        ("a class moved", Box::new(|s: &mut Statement| s.lines[1].class = 3), "line_unsourced"),
    ] {
        let mut x = s.clone();
        f(&mut x);
        let mut doc = document.clone();
        doc["statement"] = json!(x.sign(&operator).to_bytes().iter().map(|b| format!("{b:02x}")).collect::<String>());
        assert_eq!(share::verify(&doc).err().map(|e| e.0), Some(reason), "{what}");
    }
}

/// The four operations from the command line (extensions/revenue-share.md §10): a contract from
/// its terms, countersigned by each owner until it is in force; a statement created from the
/// corpus's October and verified; and a statement changed after signing, refused with its reason
/// as the code at exit 1.
#[test]
fn the_command_line_creates_countersigns_and_verifies() {
    use std::process::Command;
    let d = doc();
    let dir = std::env::temp_dir().join(format!("wai-share-cli-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&dir);
    std::fs::create_dir_all(&dir).unwrap();
    let wai = |args: &[&str]| Command::new(env!("CARGO_BIN_EXE_wai")).args(args).arg("--json").current_dir(&dir).output().unwrap();
    let envelope = |o: &std::process::Output| -> Value { serde_json::from_slice(&o.stdout).unwrap_or_else(|_| panic!("{}", String::from_utf8_lossy(&o.stdout))) };
    let hexs = |b: &[u8]| b.iter().map(|x| format!("{x:02x}")).collect::<String>();
    for k in ["operator", "owner-a", "owner-b"] {
        std::fs::write(dir.join(format!("{k}.key")), hexs(&corpus::key(k).to_bytes())).unwrap();
    }
    let pk = |k: &str| hexs(&corpus::key(k).verifying_key().to_bytes());
    let (a, b) = (pk("owner-a"), pk("owner-b"));
    let terms = json!({
        "start": 0, "end": 10, "currency": "EUR", "channel": "fast-7", "channel_signer": "22".repeat(32), "rule": "following", "between": 1,
        "carriers": [{"room": "11".repeat(32)}, {"stitcher": "33".repeat(32)}], "owners": [b, a],
        "splits": [{"operator": 400_000, "owners": {a.clone(): 600_000}}, {"operator": 0, "owners": {a.clone(): 500_000, b.clone(): 500_000}}],
        "assets": {"b/doc-1": 1, "a/ep-1": 0},
    });
    std::fs::write(dir.join("terms.json"), terms.to_string()).unwrap();
    let o = wai(&["contract", "create", "terms.json", "--key", "operator.key", "--out", "c.wrc"]);
    assert!(o.status.success(), "{}", String::from_utf8_lossy(&o.stderr));
    assert_eq!(envelope(&o)["data"]["in_force"], false);
    let o = wai(&["contract", "accept", "c.wrc", "--key", "owner-a.key", "--out", "c1.wrc"]);
    assert_eq!(envelope(&o)["data"]["in_force"], false, "one owner of two");
    let o = wai(&["contract", "accept", "c1.wrc", "--key", "owner-b.key", "--out", "c2.wrc"]);
    assert!(o.status.success());
    assert_eq!(envelope(&o)["data"]["in_force"], true);
    assert!(Contract::parse(&std::fs::read(dir.join("c2.wrc")).unwrap()).is_ok());
    // A key the contract does not name countersigns nothing.
    let o = wai(&["contract", "accept", "c2.wrc", "--key", "operator.key", "--out", "c3.wrc"]);
    assert_eq!(o.status.code(), Some(1));
    assert!(!dir.join("c3.wrc").exists());

    std::fs::write(dir.join("inputs.json"), october(&d, "preceding").to_string()).unwrap();
    let o = wai(&["statement", "create", "inputs.json", "--party", &a, "--start", "1790812800000000", "--end", "1793491200000000", "--key", "operator.key", "--out", "a.json"]);
    assert!(o.status.success(), "{}", String::from_utf8_lossy(&o.stdout));
    let v = envelope(&o);
    assert_eq!(v["data"]["outcome"]["total"], 1501 + 903 + 2251);
    let o = wai(&["statement", "verify", "a.json"]);
    assert!(o.status.success());
    assert_eq!(envelope(&o)["data"]["outcome"], "accept");
    // The statement's total changed after signing.
    let mut doc: Value = serde_json::from_slice(&std::fs::read(dir.join("a.json")).unwrap()).unwrap();
    let mut s = Statement::read(&unhex(doc["statement"].as_str().unwrap())).unwrap();
    s.total += 1;
    doc["statement"] = json!(hexs(&s.sign(&corpus::key("operator")).to_bytes()));
    std::fs::write(dir.join("b.json"), doc.to_string()).unwrap();
    let o = wai(&["statement", "verify", "b.json"]);
    assert_eq!(o.status.code(), Some(1));
    assert_eq!(envelope(&o)["error"]["code"], "share_statement_mismatch");
    let _ = std::fs::remove_dir_all(&dir);
}