//! Retiring receipts under a seal (log-seal §5.4): for every cut of every
//! log, what a restart restores from the compacted log is what it restores
//! from the whole one, and the seals after the cut still read.

use ed25519_dalek::SigningKey;
use jcp_grant::issue;
use jcp_logseal::compact::{compact, frontier, resign, root_extending, sums_of, verify_compacted, Compacted};
use jcp_logseal::{merkle_root, seal_log, seal_stating, HybridKey, Policy, SealError, Statement, StewardKeys, Trust};
use jcp_receipt::{FlowSeal, ReceiptLog};
use jcp_runtime::{HoldId, Price, Runtime};
use jcp_schema::flow::FlowFacts;
use jcp_schema::{Capability, ContentId, Did, Energy, Frame, FrameKind, Funds, Grant, SignedGrant, Trust as FrameTrust};
use serde_json::{json, Value};

const STEWARD: &str = "did:example:compacted";

fn sk(s: u8) -> SigningKey {
    SigningKey::from_bytes(&[s; 32])
}
fn did(s: u8) -> Did {
    Did::from_verifying_key(&sk(s).verifying_key())
}
fn key() -> HybridKey {
    HybridKey::new(sk(7), [9; 32])
}
fn keys() -> StewardKeys {
    StewardKeys { ed: key().ed_public(), pq: key().pq_public().to_vec() }
}
fn receipt_key() -> [u8; 32] {
    sk(7).verifying_key().to_bytes()
}
fn usd(n: u64) -> Price {
    Price::new("USD", n)
}

fn grant(n: u8) -> SignedGrant {
    issue(
        &sk(1),
        Grant {
            issuer: did(1),
            subject: did(2),
            capability: Capability::new(ContentId::of_bytes(b"compacted"), ["pay"]),
            budget_uj: 1_000_000,
            budget_funds: Some(Funds::new("USD", 100_000)),
            caveats: Vec::new(),
            expiry: 0,
            parent: None,
            delegate_depth: None,
            nonce: [n; 16],
        },
    )
    .unwrap()
}
fn frame(sg: &SignedGrant, n: u64) -> Frame {
    Frame {
        kind: FrameKind::ToolCall,
        trust: FrameTrust::Control,
        grant_id: sg.grant.id(),
        action: "pay".into(),
        resource: ContentId::of_bytes(b"compacted"),
        payload_hash: ContentId::of_bytes(format!("work:{n}").as_bytes()),
        model: None,
    }
}

struct Rng(u64);
impl Rng {
    fn below(&mut self, n: u64) -> u64 {
        self.0 ^= self.0 << 13;
        self.0 ^= self.0 >> 7;
        self.0 ^= self.0 << 17;
        self.0 % n.max(1)
    }
}

/// An open hold and what the log knows of it.
struct Open {
    grant: usize,
    id: HoldId,
    receipt: ContentId,
    captures: Vec<(ContentId, u64)>,
}

/// A log of the whole priced lifecycle, at random: decisions, holds with
/// deadlines, captures, reversals, settles, reclaims and refunds, over three
/// grants.
fn random_log(seed: u64, steps: usize) -> Vec<Vec<u8>> {
    let mut rng = Rng(seed.wrapping_mul(0x9E37_79B9_7F4A_7C15) | 1);
    let grants = [grant(1), grant(2), grant(3)];
    let facts = FlowFacts::bottom();
    let clean = FlowSeal::clean();
    let mut rt = Runtime::new();
    let mut log = ReceiptLog::new(sk(7), STEWARD);
    let mut open: Vec<Open> = Vec::new();
    for step in 0..steps {
        let now = 10 + step as u64;
        let g = rng.below(3) as usize;
        let sg = &grants[g];
        let f = frame(sg, step as u64);
        match rng.below(7) {
            0 => {
                let d = rt.decide_priced(&f, sg, did(2), Energy::measured(1 + rng.below(50)), &usd(1 + rng.below(300)), now);
                log.append(&f, &sg.grant, &d);
            }
            1 | 2 => {
                let (d, h) = rt.hold_priced_until(&f, sg, did(2), now, &facts, &usd(100 + rng.below(900)), now + 5 + rng.below(20));
                if let Some(h) = h {
                    log.append_hold(&f, &sg.grant, &d, &clean, h.uj, h.minor);
                    open.push(Open { grant: g, id: h.id, receipt: log.head(), captures: Vec::new() });
                }
            }
            3 if !open.is_empty() => {
                let o = rng.below(open.len() as u64) as usize;
                let h = &mut open[o];
                let sg = &grants[h.grant];
                let amount = 1 + rng.below(400);
                let d = rt.capture_priced(&sg.grant, h.id, Energy::measured(1 + rng.below(30)), &usd(amount), now, &facts);
                log.append_capture(&frame(sg, step as u64), &sg.grant, &d, &clean, h.receipt);
                if d.is_allow() {
                    h.captures.push((log.head(), amount));
                }
            }
            4 if !open.is_empty() => {
                let o = rng.below(open.len() as u64) as usize;
                let h = open.remove(o);
                let sg = &grants[h.grant];
                let d = rt.settle_priced(&sg.grant, h.id, Energy::measured(rng.below(20)), &usd(rng.below(200)), &facts);
                log.append_settle(&frame(sg, step as u64), &sg.grant, &d, &clean, h.receipt);
                if !d.is_allow() {
                    open.push(h);
                }
            }
            5 if !open.is_empty() => {
                let o = rng.below(open.len() as u64) as usize;
                let h = &open[o];
                let sg = &grants[h.grant];
                if rng.below(2) == 0 {
                    let (d, _) = rt.reclaim(&sg.grant, h.id, now + rng.below(30));
                    log.append_reclaim(&frame(sg, step as u64), &sg.grant, &d, h.receipt);
                } else if let Some(&(capture, amount)) = h.captures.last() {
                    let back = 1 + rng.below(amount);
                    let d = rt.reverse_capture(&sg.grant, h.id, &usd(back));
                    log.append_reversal(&frame(sg, step as u64), &sg.grant, &d, capture, back);
                }
            }
            _ => {
                let d = rt.refund(&sg.grant, did(1), &usd(1 + rng.below(100)));
                log.append_refund(&f, &sg.grant, &d, 1 + rng.below(100));
            }
        }
    }
    log.entries().to_vec()
}

/// **For every cut, the compacted log restores what the whole log does,**
/// and a seal written after the cut still reads against it.
#[test]
fn every_cut_of_every_log_restores_what_the_whole_log_restores() {
    let mut carried_any = 0;
    // Each cut signs two hybrid seals and a compaction and reads them back,
    // so the count is held to what a debug build does in seconds.
    for seed in 0..24u64 {
        let entries = random_log(seed, 40);
        if entries.is_empty() {
            continue;
        }
        let whole = sums_of(&entries, STEWARD, &receipt_key());
        let mut rng = Rng(seed + 1000);
        for _ in 0..3 {
            let cut = 1 + rng.below(entries.len() as u64) as usize;
            let at = seal_log(&entries[..cut], STEWARD, &key(), None).unwrap();
            let log = compact(&entries, &at, STEWARD, &key(), &receipt_key()).unwrap();
            carried_any += log.carried.len();
            let later = if cut < entries.len() { vec![seal_log(&entries, STEWARD, &key(), Some(&at)).unwrap()] } else { Vec::new() };
            for trust in [Trust::ReceiptsAndSeals, Trust::SealsAlone] {
                let reading = verify_compacted(&log, &later, STEWARD, &keys(), &receipt_key(), trust, Policy::RequireSealed)
                    .unwrap_or_else(|e| panic!("seed {seed} cut {cut}: {e}"));
                assert_eq!(reading.ledger, whole, "seed {seed} cut {cut}: the restart differs");
                assert_eq!((reading.standing.entries, reading.standing.sealed), (entries.len(), entries.len()));
            }
            assert!(log.carried.len() + log.after.len() <= entries.len());
        }
    }
    assert!(carried_any > 0, "no cut ever carried an open hold: the test proves less than it says");
}

/// The frontier makes every root a whole tree would, from any cut.
#[test]
fn a_frontier_and_what_follows_make_every_later_root() {
    let ids: Vec<ContentId> = (0..40u32).map(|i| ContentId::of_bytes(&i.to_le_bytes())).collect();
    for k in 1..=ids.len() {
        let f = frontier(&ids[..k]);
        assert_eq!(f.len(), (k as u64).count_ones() as usize);
        for m in k..=ids.len() {
            assert_eq!(root_extending(k as u64, &f, &ids[k..], m as u64), Some(merkle_root(&ids[..m])), "{k} → {m}");
        }
        assert_eq!(root_extending(k as u64, &f[1..], &ids[k..], k as u64), None, "a frontier short a subtree");
    }
}

/// A scene with every kind of carried receipt, for the refusals below.
fn scene() -> (Vec<Vec<u8>>, Vec<u8>, Compacted) {
    let sg = grant(1);
    let facts = FlowFacts::bottom();
    let clean = FlowSeal::clean();
    let mut rt = Runtime::new();
    let mut log = ReceiptLog::new(sk(7), STEWARD);
    let f = frame(&sg, 0);
    let d = rt.decide_priced(&f, &sg, did(2), Energy::measured(5), &usd(50), 1);
    log.append(&f, &sg.grant, &d);
    // a hold settled before the cut: retired
    let (d, h) = rt.hold_priced_until(&f, &sg, did(2), 2, &facts, &usd(300), 100);
    let h = h.unwrap();
    log.append_hold(&f, &sg.grant, &d, &clean, h.uj, h.minor);
    let done = log.head();
    let d = rt.settle_priced(&sg.grant, h.id, Energy::measured(3), &usd(120), &facts);
    log.append_settle(&f, &sg.grant, &d, &clean, done);
    // a hold captured before the cut and settled after it: carried
    let (d, h2) = rt.hold_priced_until(&f, &sg, did(2), 3, &facts, &usd(500), 100);
    let h2 = h2.unwrap();
    log.append_hold(&f, &sg.grant, &d, &clean, h2.uj, h2.minor);
    let live = log.head();
    let d = rt.capture_priced(&sg.grant, h2.id, Energy::measured(4), &usd(200), 4, &facts);
    log.append_capture(&f, &sg.grant, &d, &clean, live);
    let d = rt.refund(&sg.grant, did(1), &usd(30));
    log.append_refund(&f, &sg.grant, &d, 30);
    let cut = log.entries().len();
    let d = rt.settle_priced(&sg.grant, h2.id, Energy::measured(2), &usd(40), &facts);
    log.append_settle(&f, &sg.grant, &d, &clean, live);
    let entries = log.entries().to_vec();
    let at = seal_log(&entries[..cut], STEWARD, &key(), None).unwrap();
    let later = seal_log(&entries, STEWARD, &key(), Some(&at)).unwrap();
    let compacted = compact(&entries, &at, STEWARD, &key(), &receipt_key()).unwrap();
    (entries, later, compacted)
}

fn read(log: &Compacted, later: &[u8]) -> Result<jcp_logseal::compact::CompactedReading, SealError> {
    verify_compacted(log, &[later.to_vec()], STEWARD, &keys(), &receipt_key(), Trust::SealsAlone, Policy::RequireSealed)
}

#[test]
fn the_open_hold_is_carried_and_the_settled_one_retired() {
    let (entries, later, log) = scene();
    assert_eq!(log.carried.iter().map(|c| c.index).collect::<Vec<_>>(), [3, 4], "the open hold and its capture");
    assert_eq!(read(&log, &later).unwrap().ledger, sums_of(&entries, STEWARD, &receipt_key()));
}

#[test]
fn a_compaction_that_disagrees_with_itself_or_its_log_is_refused() {
    let (_, later, log) = scene();
    let code = |l: &Compacted| read(l, &later).map(|_| ()).map_err(|e| e.code());

    // sums changed under the old signatures
    let mut v: Value = jcp_schema::from_canonical_cbor(&log.compaction).unwrap();
    let g = v["retired"]["minor"].as_object().unwrap().keys().next().unwrap().clone();
    v["retired"]["minor"][&g] = json!(1);
    let altered = Compacted { compaction: jcp_receipt::jcr1::canonical_cbor(&v).unwrap(), ..log.clone() };
    assert_eq!(code(&altered), Err("seal_signature_invalid"));

    // a carried receipt left out
    let short = Compacted { carried: log.carried[..1].to_vec(), ..log.clone() };
    assert_eq!(code(&short), Err("compaction_mismatch"));

    // a carried receipt with another's path
    let mut swapped = log.clone();
    swapped.carried[0].path = swapped.carried[1].path.clone();
    assert_eq!(code(&swapped), Err("compaction_mismatch"));

    // the steward retiring the open hold and re-signing: the settle after the
    // cut names a hold the log no longer has
    let unclosed = Compacted {
        compaction: resign(&log.compaction, |r| { r.insert("carried".into(), json!([])); }, &key()).unwrap(),
        carried: Vec::new(),
        ..log.clone()
    };
    assert_eq!(code(&unclosed), Err("compaction_unclosed"));

    // a frontier that does not make the seal's root
    let bent = Compacted {
        compaction: resign(&log.compaction, |r| { r["frontier"][0] = json!(ContentId::of_bytes(b"x").to_hex()); }, &key()).unwrap(),
        ..log.clone()
    };
    assert_eq!(code(&bent), Err("compaction_mismatch"));

    // a later seal whose head is the log's and whose root is not: only the
    // frontier extended by the receipts after the cut can tell
    let (entries, _, _) = scene();
    let ids: Vec<ContentId> = entries.iter().map(|e| ContentId::of_bytes(e)).collect();
    let n = ids.len();
    let false_root = Statement { count: n as u64, head: ids[n - 1], root: merkle_root(&ids[..n - 1]), prev: Some(ContentId::of_bytes(&log.seal)), next_key: None };
    let liar = seal_stating(STEWARD, &false_root, &key()).unwrap();
    assert_eq!(read(&log, &liar).map(|_| ()).map_err(|e| e.code()), Err("seal_root_mismatch"));

    // a receipt after the cut taken away: the later seal no longer reads
    let cut_short = Compacted { after: Vec::new(), ..log.clone() };
    assert_eq!(code(&cut_short), Err("log_truncated"));

    // signed by someone holding the steward's Ed25519 key and not its
    // post-quantum one
    let forged = Compacted { compaction: resign(&log.compaction, |_| {}, &HybridKey::new(sk(7), [66; 32])).unwrap(), ..log };
    assert_eq!(code(&forged), Err("pq_signature_invalid"));
}