//! The published replay log (`replay.json`, SPEC §8.1), read through the
//! reference library. `conformance/verify_replay.py` reads the same file without it.

use jcp_grant::standing::MintLedger;
use jcp_receipt::{extract, jcr1, read_stream, replay_ledger, replay_mints, verify_log, ReceiptCore};
use jcp_runtime::Runtime;
use jcp_schema::ContentId;
use serde_json::Value;
use std::path::PathBuf;

fn pack_file(name: &str) -> Value {
    let path = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../../conformance/v1").join(name);
    serde_json::from_slice(&std::fs::read(path).expect("read pack file")).expect("the pack file parses")
}

fn steward_and_key() -> (String, [u8; 32]) {
    let m = pack_file("meta.json");
    let key = hex::decode(m["pubkey_hex"].as_str().unwrap()).unwrap().try_into().unwrap();
    (m["steward"].as_str().unwrap().to_string(), key)
}

fn envelope(entry: &Value) -> Vec<u8> {
    hex::decode(entry["envelope_hex"].as_str().unwrap()).unwrap()
}

fn entries(doc: &Value) -> Vec<Vec<u8>> {
    doc["entries"].as_array().expect("the log lists entries").iter().map(envelope).collect()
}

/// What an entry publishes for one dimension: `debits` and `tainted` in
/// joules, `money` in minor units.
fn published(entry: &Value, dim: &str) -> Vec<(ContentId, u64)> {
    let unit = if dim == "money" { "minor" } else { "uj" };
    entry[dim]
        .as_array()
        .unwrap_or_else(|| panic!("the entry publishes {dim}"))
        .iter()
        .map(|d| (ContentId::from_hex(d["grant_hex"].as_str().unwrap()).unwrap(), d[unit].as_u64().unwrap()))
        .collect()
}

/// A receipt read in every dimension, as the pack publishes it.
fn reads(core: &ReceiptCore) -> [Vec<(ContentId, u64)>; 3] {
    [core.debits(), core.money_debits(), core.tainted_accruals()]
}

fn publishes(entry: &Value) -> [Vec<(ContentId, u64)>; 3] {
    [published(entry, "debits"), published(entry, "money"), published(entry, "tainted")]
}

/// What an entry publishes a standing grant's mint ledger moved by: the grant,
/// its fires, and the budget they minted.
fn published_mints(entry: &Value) -> Vec<(ContentId, u64, u64)> {
    entry["mints"]
        .as_array()
        .expect("the entry publishes mints")
        .iter()
        .map(|d| {
            let grant = ContentId::from_hex(d["grant_hex"].as_str().unwrap()).unwrap();
            (grant, d["fires"].as_u64().unwrap(), d["uj"].as_u64().unwrap())
        })
        .collect()
}

/// What a receipt adds to a mint ledger: one fire and its `minted_uj`, on a firing.
fn reads_mints(core: &ReceiptCore) -> Vec<(ContentId, u64, u64)> {
    core.minted_uj.filter(|_| core.is_mint()).map(|uj| (core.grant_id, 1, uj)).into_iter().collect()
}

/// The log verifies end to end, and each core is the published wire form.
#[test]
fn the_published_log_is_one_verified_chain() {
    let doc = pack_file("replay.json");
    let (steward, key) = steward_and_key();
    let log = entries(&doc);
    assert!(!log.is_empty(), "a wrong path must not look like a pass");
    assert_eq!(verify_log(&log, &steward, &key), Ok(log.len()));
    for (env, entry) in log.iter().zip(doc["entries"].as_array().unwrap()) {
        let payload = jcr1::verify_cose_sign1(env, &steward, &key).unwrap();
        let core: Value = jcp_schema::from_canonical_cbor(&payload).unwrap();
        assert_eq!(core, entry["wire"], "{}", entry["what"]);
    }
}

/// Each entry reads as what the runtime's ledgers moved by when it was
/// sealed: its joules, its money and its tainted accrual. A hold is read
/// against the whole log, so the log is read as one stream: a hold no entry
/// settles reads as what a restart holds, the figure the runtime bounded it by.
#[test]
fn each_published_entry_reads_as_its_ledger_movement() {
    let doc = pack_file("replay.json");
    let (steward, key) = steward_and_key();
    let log = entries(&doc);
    let read = read_stream(log.iter().map(Vec::as_slice), &steward, &key);
    assert_eq!(read.len(), log.len());
    for ((entry, ledger), env) in doc["entries"].as_array().unwrap().iter().zip(read).zip(&log) {
        let l = ledger.expect("every published entry verifies");
        assert_eq!([l.uj, l.minor, l.tainted_uj], publishes(entry), "{}", entry["what"]);
        let core = extract(env, &steward, &key).unwrap();
        assert_eq!(reads_mints(&core), published_mints(entry), "{}", entry["what"]);
    }
}

/// The firings in the log rebuild each standing grant's mint ledger the
/// firings left, and the child's call between them is no firing.
#[test]
fn replaying_the_published_log_rebuilds_the_mint_ledger() {
    let doc = pack_file("replay.json");
    let (steward, key) = steward_and_key();
    let log = entries(&doc);
    let minted = replay_mints(log.iter().map(Vec::as_slice), &steward, &key);
    let listed = doc["mints_after"].as_object().expect("the log publishes mints_after");
    assert!(!listed.is_empty(), "a wrong path must not look like a pass");
    for (grant, want) in listed {
        let id = ContentId::from_hex(grant).unwrap();
        let rebuilt = MintLedger::of_firings(minted.iter().filter(|(g, _)| *g == id).map(|(_, uj)| *uj));
        let want = MintLedger { fires: want["fires"].as_u64().unwrap(), spent_uj: want["uj"].as_u64().unwrap() };
        assert_eq!(rebuilt, want, "{grant}");
        let mut stale = MintLedger { fires: 1, spent_uj: 0 };
        stale.restore(&rebuilt);
        assert_eq!(stale, want, "a stale copy restored beside the stream");
    }
    assert_eq!(minted.len() as u64, listed.values().map(|v| v["fires"].as_u64().unwrap()).sum::<u64>());
}

/// The holds in the log: the one an entry settles reads as nothing alone,
/// and the two nothing settles are what `unresolved_holds` names.
#[test]
fn the_published_holds_read_as_the_log_settles_them() {
    let doc = pack_file("replay.json");
    let (steward, key) = steward_and_key();
    let log = entries(&doc);
    let holds: Vec<usize> = (0..log.len()).filter(|&i| extract(&log[i], &steward, &key).unwrap().is_hold()).collect();
    assert_eq!(holds.len(), 3, "a settled hold, an unsettled one, and an unsettled tainted one");
    let unresolved = jcp_receipt::unresolved_holds(log.iter().map(Vec::as_slice), &steward, &key);
    assert_eq!(unresolved, holds[1..].iter().map(|&i| ContentId::of_bytes(&log[i])).collect::<Vec<_>>());
    let settle = extract(&log[holds[0] + 1], &steward, &key).unwrap();
    assert_eq!(settle.settles, Some(ContentId::of_bytes(&log[holds[0]])));
    for &i in &holds {
        let core = extract(&log[i], &steward, &key).unwrap();
        assert!(core.debits().is_empty() && core.tainted_accruals().is_empty(), "a hold alone debits nothing");
    }
}

/// The whole log restores the ledgers the runtime ended with, grant by grant,
/// in every dimension, and a grant the pack lists nothing for holds nothing.
#[test]
fn replaying_the_published_log_restores_the_ledger() {
    let doc = pack_file("replay.json");
    let (steward, key) = steward_and_key();
    let log = entries(&doc);
    let mut rt = Runtime::new();
    rt.restore_ledger(replay_ledger(log.iter().map(Vec::as_slice), &steward, &key));

    let held = |total: &str, g: &ContentId| match total {
        "spent_after" => rt.spent(g),
        "money_after" => rt.spent_minor(g),
        _ => rt.spent_tainted(g),
    };
    let budgets = doc["grants"].as_object().unwrap();
    for total in ["spent_after", "money_after", "tainted_after"] {
        let listed = doc[total].as_object().unwrap_or_else(|| panic!("the log publishes {total}"));
        assert!(!listed.is_empty(), "{total}");
        for g in budgets.values() {
            let id = g["id_hex"].as_str().unwrap();
            let want = listed.get(id).map_or(0, |v| v.as_u64().unwrap());
            assert_eq!(held(total, &ContentId::from_hex(id).unwrap()), want, "{total} {id}");
        }
    }
    for g in budgets.values() {
        let id = ContentId::from_hex(g["id_hex"].as_str().unwrap()).unwrap();
        assert!(rt.spent(&id) <= g["budget_uj"].as_u64().unwrap(), "replay passed a joule ceiling on {id:?}");
        if let Some(minor) = g["budget_funds"]["minor"].as_u64() {
            assert!(rt.spent_minor(&id) <= minor, "replay passed a money ceiling on {id:?}");
        }
    }
}

/// The log keeps each form the reading rules distinguish, so no rule can stop
/// being exercised by a later regeneration.
#[test]
fn the_published_log_holds_each_form() {
    let doc = pack_file("replay.json");
    let wires: Vec<&Value> = doc["entries"].as_array().unwrap().iter().map(|e| &e["wire"]).collect();
    let denial = |w: &&&Value| w["verdict"] == "deny";
    let stated = |w: &Value| w.get("debited_uj").and_then(Value::as_u64);
    assert!(wires.iter().any(|w| w["verdict"] == "allow" && stated(w).is_none()), "a plain allow");
    assert!(wires.iter().filter(denial).any(|w| stated(w).is_some_and(|uj| uj > 0)), "a denial that debited");
    assert!(wires.iter().filter(denial).any(|w| stated(w) == Some(0)), "a refusal stating its zero");
    assert!(wires.iter().filter(denial).any(|w| stated(w).is_none()), "a refusal stating nothing");
    assert!(wires.iter().any(|w| stated(w).is_some_and(|uj| uj > 0 && uj < w["cost_uj"].as_u64().unwrap())), "a cap");
    assert!(wires.iter().any(|w| w.get("chain_ids").is_some()), "a chained receipt");
    assert!(wires.iter().any(|w| w.get("chain_debited_uj").is_some()), "links debited differently");
    assert!(wires.iter().any(|w| w.get("debited_minor").is_some()), "a money debit");
    assert!(wires.iter().any(|w| w.get("chain_debited_minor").is_some()), "links debited money differently");
    let flow = |w: &&&Value| w.get("flow").is_some();
    assert!(wires.iter().filter(flow).any(|w| w["verdict"] == "allow" && w.get("tainted_uj").is_none()), "an accrual its flow block says");
    assert!(wires.iter().filter(flow).any(|w| w["verdict"] == "deny" && w["cost_uj"].as_u64() > Some(0)), "a tainted refusal with a cost");
    assert!(wires.iter().any(|w| w.get("tainted_uj").is_some() && w.get("flow").is_none()), "an accrual stated");
    let hold = |w: &&&Value| w.get("held_uj").is_some() && w["verdict"] == "allow";
    assert_eq!(wires.iter().filter(hold).count(), 3, "holds: settled, unsettled, unsettled and tainted");
    assert!(wires.iter().filter(hold).any(|w| w.get("flow").is_some()), "a tainted hold");
    assert!(wires.iter().any(|w| w.get("settles").is_some()), "a settle");
    let firing = |w: &&&Value| w.get("minted_uj").is_some() && w["verdict"] == "allow";
    assert_eq!(wires.iter().filter(firing).count(), 2, "two firings of one standing grant");
    let standing = wires.iter().find(firing).map(|w| w["grant_id"].clone()).unwrap();
    assert!(
        wires.iter().any(|w| w.get("minted_uj").is_none() && w.get("chain_ids").is_some_and(|c| c[0] == standing)),
        "a call under a child a firing minted"
    );
    assert!(wires.iter().all(|w| w.get("seq").is_some() && w.get("prev").is_some()), "a log's links");
}

/// A published debit changed in place fails the signature, and the log breaks
/// at that entry.
#[test]
fn a_debit_changed_on_a_published_receipt_is_refused() {
    let doc = pack_file("replay.json");
    let (steward, key) = steward_and_key();
    let log = entries(&doc);
    for member in ["debited_uj", "debited_minor", "tainted_uj", "held_uj", "settles", "minted_uj"] {
        let stated: Vec<usize> = doc["entries"]
            .as_array()
            .unwrap()
            .iter()
            .enumerate()
            .filter(|(_, e)| e["wire"].get(member).is_some())
            .map(|(i, _)| i)
            .collect();
        assert!(!stated.is_empty(), "{member}");
        // The key as CBOR text: its length in the header, then the bytes.
        let name = [&[0x60 | member.len() as u8][..], member.as_bytes()].concat();
        for i in stated {
            let mut changed = log.clone();
            let at = changed[i].windows(name.len()).position(|w| w == name).unwrap() + name.len();
            changed[i][at] ^= 0x01;
            assert!(extract(&changed[i], &steward, &key).is_err(), "{member} on entry {i}");
            assert_eq!(verify_log(&changed, &steward, &key), Err(i));
        }
    }
}

/// An entry cannot leave the log: its `prev` ties the next one to it.
#[test]
fn removing_a_published_entry_breaks_the_log() {
    let doc = pack_file("replay.json");
    let (steward, key) = steward_and_key();
    let mut log = entries(&doc);
    let breach = doc["entries"]
        .as_array()
        .unwrap()
        .iter()
        .position(|e| e["wire"]["verdict"] == "deny" && e["wire"].get("debited_uj").and_then(Value::as_u64).is_some_and(|uj| uj > 0))
        .expect("the log holds a breach");
    log.remove(breach);
    assert_eq!(verify_log(&log, &steward, &key), Err(breach));
}

/// The published receipts in the form sealed before a member verify and state
/// none of it. Without `debited_uj` a breach and a refusal read alike, as the
/// most either can have debited; without `debited_minor` or `tainted_uj` a
/// receipt reads as no money and no accrual, which it cannot show.
#[test]
fn the_published_receipts_without_a_member_read_as_published() {
    let doc = pack_file("replay.json");
    let (steward, key) = steward_and_key();
    let legacy = doc["legacy"].as_array().expect("the pack lists them");
    let before = |m: &str| legacy.iter().filter(|e| e["sealed_before"] == m).count();
    assert_eq!((before("debited_uj"), before("debited_minor"), before("tainted_uj"), before("settles")), (2, 1, 1, 1));
    for entry in legacy {
        let core = extract(&envelope(entry), &steward, &key).expect("still readable");
        let what = &entry["what"];
        assert_eq!(reads(&core), publishes(entry), "{what}");
        match entry["sealed_before"].as_str().unwrap() {
            "debited_uj" => {
                assert_eq!((core.debited_uj, core.verdict.as_str()), (None, "deny"), "{what}");
                assert_eq!(core.debits()[0].1, core.cost_uj);
            }
            "debited_minor" => assert_eq!(core.debited_minor, None, "{what}"),
            "tainted_uj" => assert_eq!(core.tainted_uj, None, "{what}"),
            "settles" => assert_eq!(core.settles, None, "{what}"),
            other => panic!("an unknown member {other}"),
        }
    }
}