//! Delivery settlement — `wai.delivery.settlement` ([extensions/settlement.md]).
//!
//! Delivery paid on what the party that received it signed. The downstream party of a
//! delivery, a next hop or a sink, signs an **acknowledgement** (`WDA1`) of what it received
//! in a window: per object class, the objects by content hash under jwp-receipts §2.2's tree,
//! the byte range of each, and their byte count. A hop whose downstream party does not sign
//! states that delivery in a **declaration**, the same record signed by the hop, which is
//! counted apart and billed never. A **range proof** (`WBR1`) shows that given bytes are a
//! range of an object, over BLAKE3's chunk tree, without the rest of it. A retrieval that
//! costs money is billed only to a grant whose proof of possession the request carried. Per
//! party and per period, a **settlement statement** (`WSS1`) states the acknowledged bytes
//! by object class and payer, the declared bytes, the party's energy by acquisition class
//! from its per-hop records ([`crate::live_metering`]), and the amount under a tariff in
//! integer micro-units; [`verify`] recomputes every figure from the records.
//!
//! What this module implements, by section of the extension:
//!
//! - §3, §4: the acknowledgement and the declaration, their entries, items, rules and chains;
//! - §5: the range proof ([`RangeProof`]): building, reading and verifying it, and audits;
//! - §6: the checks of a grant and its proof on an entry;
//! - §7: the statement, its lines, its tariff and its recomputation ([`figures`]), and its
//!   attribution lines from a [`Basis`] (§7.8, [`attribution_lines`]);
//! - §8: verification ([`verify`]).
//!
//! The corpus `live-conformance/settlement/` is written by [`corpus::generate`] and checked
//! by [`corpus::verify`] here, and by `live-conformance/settlement_verify.py`, which shares
//! no code with this module.
//!
//! [extensions/settlement.md]: ../../extensions/settlement.md

use std::collections::BTreeMap;

use blake3::hazmat::{merge_subtrees_non_root, merge_subtrees_root, ChainingValue, HasherExt, Mode};
use ed25519_dalek::{Signer, SigningKey};
use serde_json::{json, Value};
use sha2::{Digest, Sha256};

use crate::jwp_hash::{content_hash, DOMAIN_OBJECT};
use crate::live_events as events;
use crate::live_metering::{self as metering, HopRecord, Total};
use crate::live_room::{self as room, Descriptor, HeldGrant, Proof};

pub mod corpus;

/// The record this extension registers.
pub const CAPABILITY: &str = "wai.delivery.settlement";
/// The magic of an acknowledgement or a declaration (§3.1).
pub const ACK_MAGIC: [u8; 4] = *b"WDA1";
/// The magic of a range proof (§5.2).
pub const RANGE_MAGIC: [u8; 4] = *b"WBR1";
/// The magic of a settlement statement (§7.1).
pub const STATEMENT_MAGIC: [u8; 4] = *b"WSS1";
/// The version of every record in this revision.
pub const VERSION: u8 = 1;
/// The domain of a record's signature (§3.5).
pub const ACK_DOMAIN: &[u8] = b"wai:settle-ack\x01";
/// The domain of an item list's digest (§3.2).
pub const ITEMS_DOMAIN: &[u8] = b"wai:settle-items\x01";
/// The domain of a statement's signature (§7.4).
pub const STATEMENT_DOMAIN: &[u8] = b"wai:settle-statement\x01";
/// jwp-receipts §2.2's legacy leaf and node domains.
pub const LEAF_DOMAIN: &[u8] = b"moq-jwp:merkle-leaf\x01";
pub const NODE_DOMAIN: &[u8] = b"moq-jwp:merkle-node\x01";
/// An entry's length (§3.2).
pub const ENTRY_LEN: usize = 141;
/// An item's length (§3.4).
pub const ITEM_LEN: usize = 48;
/// A line's length (§7.2).
pub const LINE_LEN: usize = 90;
/// A record's fixed fields before its entries, the entry count included (§3.1).
pub const ACK_FIXED: usize = 121;
/// A range proof's fixed fields (§5.2).
pub const RANGE_FIXED: usize = 62;
/// BLAKE3's chunk (§5.1).
pub const CHUNK: u64 = 1024;
/// The length of jwp-receipts §2.1's domain, which starts `X` (§5.1).
pub const DOMAIN_LEN: u64 = 15;
/// Prices are per 10^9 bytes (§7.3).
pub const PRICE_BYTES: u128 = 1_000_000_000;
/// The object classes (§3.3).
pub const CLASSES: usize = 4;

/// A record's mode (§3.1).
pub mod mode {
    /// Signed by the downstream party.
    pub const ACK: u8 = 1;
    /// Signed by the upstream party (§4).
    pub const DECLARATION: u8 = 2;
}

/// The downstream party's role (§3.1).
pub mod role {
    pub const HOP: u8 = 1;
    pub const SINK: u8 = 2;
}

/// The object classes (§3.3).
pub mod class {
    pub const LIVE: u8 = 1;
    pub const FETCH: u8 = 2;
    pub const FILL: u8 = 3;
    pub const ANCILLARY: u8 = 4;

    /// The class's name.
    pub fn name(c: u8) -> &'static str {
        match c {
            LIVE => "live",
            FETCH => "fetch",
            FILL => "fill",
            ANCILLARY => "ancillary",
            _ => "unknown",
        }
    }

    /// A retrieval class (2 or 3), billed only to a grant (§6).
    pub fn is_retrieval(c: u8) -> bool {
        c == FETCH || c == FILL
    }
}

/// A line's kind (§7.2).
pub mod line_kind {
    /// A delivery line.
    pub const DELIVERY: u8 = 1;
    /// An attribution line: a payee that is not a hop, paid from a basis (§7.8).
    pub const ATTRIBUTION: u8 = 2;
}

// ---------------------------------------------------------------------------------------
// Reasons (§11)
// ---------------------------------------------------------------------------------------

/// Why a record, a range proof or a statement is refused (§11).
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub enum Reason {
    /// A code of live-metering §9 (and, through it, of live-rooms §11 and live-events §9).
    Metering(metering::Reason),
    AckInvalid,
    SelfAcknowledged,
    ItemsInvalid,
    ItemsMismatch,
    AckWrongParty,
    AckOutsidePeriod,
    AckOverlap,
    AckChainBroken,
    DeclaredAndAcknowledged,
    RangeInvalid,
    RangeOutsideObject,
    RangeProofMismatch,
    RangeNotAcknowledged,
    StatementInvalid,
    PaidWithoutGrant,
    BytesNotAcknowledged,
    StatementMismatch,
    HopNotParty,
    SpanOutsidePeriod,
}

impl From<metering::Reason> for Reason {
    fn from(r: metering::Reason) -> Self {
        Reason::Metering(r)
    }
}

impl From<room::Reason> for Reason {
    fn from(r: room::Reason) -> Self {
        Reason::Metering(metering::Reason::Room(r))
    }
}

const fn ev(r: events::Reason) -> Reason {
    Reason::Metering(metering::Reason::Room(room::Reason::Events(r)))
}

const fn rm(r: room::Reason) -> Reason {
    Reason::Metering(metering::Reason::Room(r))
}

const TRUNCATED: Reason = ev(events::Reason::Truncated);
const MALFORMED: Reason = ev(events::Reason::Malformed);
const UNKNOWN_VERSION: Reason = ev(events::Reason::UnknownVersion);
const BAD_KEY: Reason = ev(events::Reason::BadKey);
const BAD_SIGNATURE: Reason = ev(events::Reason::BadSignature);
const RECORD_MISSING: Reason = Reason::Metering(metering::Reason::RecordMissing);
const CLASSES_SUMMED: Reason = Reason::Metering(metering::Reason::ClassesSummed);

impl Reason {
    /// The codes this extension adds, in §11's order.
    pub const ALL: [Reason; 19] = [
        Reason::AckInvalid,
        Reason::SelfAcknowledged,
        Reason::ItemsInvalid,
        Reason::ItemsMismatch,
        Reason::AckWrongParty,
        Reason::AckOutsidePeriod,
        Reason::AckOverlap,
        Reason::AckChainBroken,
        Reason::DeclaredAndAcknowledged,
        Reason::RangeInvalid,
        Reason::RangeOutsideObject,
        Reason::RangeProofMismatch,
        Reason::RangeNotAcknowledged,
        Reason::StatementInvalid,
        Reason::PaidWithoutGrant,
        Reason::BytesNotAcknowledged,
        Reason::StatementMismatch,
        Reason::HopNotParty,
        Reason::SpanOutsidePeriod,
    ];

    /// The codes of other extensions this one raises (§11's last paragraph), beside every code
    /// a descriptor, a grant and a per-hop record raise.
    pub const REUSED: [&'static str; 15] = [
        "truncated",
        "malformed",
        "unknown_version",
        "bad_key",
        "bad_signature",
        "record_missing",
        "classes_summed",
        "outside_grant_window",
        "grant_wrong_principal",
        "seek_not_granted",
        "pop_malformed",
        "pop_wrong_grant",
        "pop_wrong_key",
        "pop_bad_signature",
        "pop_replayed",
    ];

    /// The reason's code.
    pub fn code(self) -> &'static str {
        match self {
            Reason::Metering(r) => r.code(),
            Reason::AckInvalid => "ack_invalid",
            Reason::SelfAcknowledged => "self_acknowledged",
            Reason::ItemsInvalid => "items_invalid",
            Reason::ItemsMismatch => "items_mismatch",
            Reason::AckWrongParty => "ack_wrong_party",
            Reason::AckOutsidePeriod => "ack_outside_period",
            Reason::AckOverlap => "ack_overlap",
            Reason::AckChainBroken => "ack_chain_broken",
            Reason::DeclaredAndAcknowledged => "declared_and_acknowledged",
            Reason::RangeInvalid => "range_invalid",
            Reason::RangeOutsideObject => "range_outside_object",
            Reason::RangeProofMismatch => "range_proof_mismatch",
            Reason::RangeNotAcknowledged => "range_not_acknowledged",
            Reason::StatementInvalid => "statement_invalid",
            Reason::PaidWithoutGrant => "paid_without_grant",
            Reason::BytesNotAcknowledged => "bytes_not_acknowledged",
            Reason::StatementMismatch => "statement_mismatch",
            Reason::HopNotParty => "hop_not_party",
            Reason::SpanOutsidePeriod => "span_outside_period",
        }
    }
}

impl std::fmt::Display for Reason {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(self.code())
    }
}

impl std::error::Error for Reason {}

// ---------------------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------------------

fn sha256(parts: &[&[u8]]) -> [u8; 32] {
    let mut h = Sha256::new();
    for p in parts {
        h.update(p);
    }
    h.finalize().into()
}

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

fn verify_sig(key: &[u8; 32], domain: &[u8], body: &[u8], sig: &[u8; 64]) -> Result<(), Reason> {
    let mut m = domain.to_vec();
    m.extend_from_slice(body);
    crate::sig::verify_strict(key, &m, sig).map_err(|e| match e {
        crate::sig::SigError::BadKey => BAD_KEY,
        crate::sig::SigError::BadSignature => BAD_SIGNATURE,
    })
}

fn sign(key: &SigningKey, domain: &[u8], body: &[u8]) -> [u8; 64] {
    let mut m = domain.to_vec();
    m.extend_from_slice(body);
    key.sign(&m).to_bytes()
}

/// The digest of given bytes (§8.2): SHA-256 over all but their last 64.
pub fn given_digest(b: &[u8]) -> [u8; 32] {
    sha256(&[&b[..b.len().saturating_sub(64)]])
}

/// The digest of a proof record (§8.2): SHA-256 over all of it.
pub fn proof_digest(b: &[u8]) -> [u8; 32] {
    sha256(&[b])
}

/// A cursor; running out is `truncated`.
struct Rd<'a> {
    b: &'a [u8],
    p: usize,
}

impl<'a> Rd<'a> {
    fn take(&mut self, n: usize) -> Result<&'a [u8], Reason> {
        let e = self.p.checked_add(n).filter(|&e| e <= self.b.len()).ok_or(TRUNCATED)?;
        let s = &self.b[self.p..e];
        self.p = e;
        Ok(s)
    }
    fn u8(&mut self) -> Result<u8, Reason> {
        Ok(self.take(1)?[0])
    }
    fn u16(&mut self) -> Result<u16, Reason> {
        Ok(u16::from_le_bytes(self.take(2)?.try_into().expect("2 bytes")))
    }
    fn u32(&mut self) -> Result<u32, Reason> {
        Ok(u32::from_le_bytes(self.take(4)?.try_into().expect("4 bytes")))
    }
    fn u64(&mut self) -> Result<u64, Reason> {
        Ok(u64::from_le_bytes(self.take(8)?.try_into().expect("8 bytes")))
    }
    fn arr<const N: usize>(&mut self) -> Result<[u8; N], Reason> {
        Ok(self.take(N)?.try_into().expect("N bytes"))
    }
    fn done(&self) -> bool {
        self.p == self.b.len()
    }
}

/// The head every record of this extension starts with: the magic, then version 1.
fn read_head(b: &[u8], magic: &[u8; 4]) -> Result<(), Reason> {
    if b.len() < 4 {
        return Err(TRUNCATED);
    }
    if &b[..4] != magic {
        return Err(MALFORMED);
    }
    if b.len() < 5 {
        return Err(TRUNCATED);
    }
    if b[4] != VERSION {
        return Err(UNKNOWN_VERSION);
    }
    Ok(())
}

// ---------------------------------------------------------------------------------------
// §2, §3.2: the tree root and the item list
// ---------------------------------------------------------------------------------------

/// jwp-receipts §2.2's legacy leaf of a content hash.
pub fn leaf(content_hash: &[u8; 32]) -> [u8; 32] {
    let mut h = blake3::Hasher::new();
    h.update(LEAF_DOMAIN);
    h.update(content_hash);
    *h.finalize().as_bytes()
}

fn node(l: &[u8; 32], r: &[u8; 32]) -> [u8; 32] {
    let mut h = blake3::Hasher::new();
    h.update(NODE_DOMAIN);
    h.update(l);
    h.update(r);
    *h.finalize().as_bytes()
}

/// The tree root of a list of content hashes (§2): the last node of an odd level paired with
/// itself, a single leaf its own root, and zero for no leaf.
pub fn tree_root(hashes: &[[u8; 32]]) -> [u8; 32] {
    if hashes.is_empty() {
        return [0; 32];
    }
    let mut level: Vec<[u8; 32]> = hashes.iter().map(leaf).collect();
    while level.len() > 1 {
        if level.len() % 2 == 1 {
            level.push(*level.last().expect("non-empty"));
        }
        level = level.as_chunks::<2>().0.iter().map(|p| node(&p[0], &p[1])).collect();
    }
    level[0]
}

/// An item (§3.4): a byte range of one object.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Item {
    pub content_hash: [u8; 32],
    pub start: u64,
    pub end: u64,
}

impl Item {
    /// The whole of `envelope`.
    pub fn whole(envelope: &[u8]) -> Item {
        Item { content_hash: content_hash(envelope), start: 0, end: envelope.len() as u64 }
    }

    /// Its byte count.
    pub fn bytes(&self) -> u64 {
        self.end - self.start
    }
}

/// An item list's bytes (§3.4).
pub fn item_list(items: &[Item]) -> Vec<u8> {
    let mut o = Vec::with_capacity(ITEM_LEN * items.len());
    for i in items {
        o.extend_from_slice(&i.content_hash);
        o.extend_from_slice(&i.start.to_le_bytes());
        o.extend_from_slice(&i.end.to_le_bytes());
    }
    o
}

/// An item list's digest (§3.2).
pub fn items_digest(list: &[u8]) -> [u8; 32] {
    sha256(&[ITEMS_DOMAIN, list])
}

/// Read an item list (§3.6): items of 48 bytes, each range not empty, no two sharing a byte
/// of one object; otherwise `items_invalid`.
pub fn read_items(b: &[u8]) -> Result<Vec<Item>, Reason> {
    if !b.len().is_multiple_of(ITEM_LEN) {
        return Err(Reason::ItemsInvalid);
    }
    let items: Vec<Item> = b
        .as_chunks::<ITEM_LEN>()
        .0
        .iter()
        .map(|c| Item {
            content_hash: c[..32].try_into().expect("32 bytes"),
            start: u64::from_le_bytes(c[32..40].try_into().expect("8 bytes")),
            end: u64::from_le_bytes(c[40..48].try_into().expect("8 bytes")),
        })
        .collect();
    if items.iter().any(|i| i.start >= i.end) {
        return Err(Reason::ItemsInvalid);
    }
    let mut sorted: Vec<&Item> = items.iter().collect();
    sorted.sort_by_key(|i| (i.content_hash, i.start, i.end));
    if sorted.windows(2).any(|w| w[0].content_hash == w[1].content_hash && w[1].start < w[0].end) {
        return Err(Reason::ItemsInvalid);
    }
    Ok(items)
}

// ---------------------------------------------------------------------------------------
// §3 The acknowledgement
// ---------------------------------------------------------------------------------------

/// An entry (§3.2).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Entry {
    pub class: u8,
    pub grant: [u8; 32],
    pub proof: [u8; 32],
    pub objects: u32,
    pub bytes: u64,
    pub root: [u8; 32],
    pub items: [u8; 32],
}

impl Entry {
    /// The entry of `items`, of `class`, under `grant` and `proof` (zero for none).
    pub fn of(class: u8, grant: [u8; 32], proof: [u8; 32], items: &[Item]) -> Entry {
        let hashes: Vec<[u8; 32]> = items.iter().map(|i| i.content_hash).collect();
        Entry {
            class,
            grant,
            proof,
            objects: items.len() as u32,
            bytes: items.iter().map(Item::bytes).sum(),
            root: tree_root(&hashes),
            items: items_digest(&item_list(items)),
        }
    }

    fn encode(&self, o: &mut Vec<u8>) {
        o.push(self.class);
        o.extend_from_slice(&self.grant);
        o.extend_from_slice(&self.proof);
        o.extend_from_slice(&self.objects.to_le_bytes());
        o.extend_from_slice(&self.bytes.to_le_bytes());
        o.extend_from_slice(&self.root);
        o.extend_from_slice(&self.items);
    }

    fn read(r: &mut Rd) -> Result<Entry, Reason> {
        Ok(Entry { class: r.u8()?, grant: r.arr()?, proof: r.arr()?, objects: r.u32()?, bytes: r.u64()?, root: r.arr()?, items: r.arr()? })
    }

    fn order(&self) -> (u8, [u8; 32], [u8; 32]) {
        (self.class, self.grant, self.proof)
    }

    /// §3.6's rules on one entry.
    fn rules_hold(&self) -> bool {
        let class_ok = (1..=CLASSES as u8).contains(&self.class);
        let sums = self.objects >= 1 && self.bytes >= self.objects as u64;
        let grant_proof = if class::is_retrieval(self.class) { (self.grant == [0; 32]) == (self.proof == [0; 32]) } else { self.proof == [0; 32] };
        class_ok && sums && grant_proof
    }

    /// §3.6: whether `items`, read, recompute against this entry.
    pub fn recomputes(&self, items: &[Item]) -> bool {
        let sum: u128 = items.iter().map(|i| i.bytes() as u128).sum();
        let hashes: Vec<[u8; 32]> = items.iter().map(|i| i.content_hash).collect();
        items.len() as u64 == self.objects as u64
            && sum == self.bytes as u128
            && tree_root(&hashes) == self.root
            && items_digest(&item_list(items)) == self.items
    }
}

/// An acknowledgement or a declaration (§3.1).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Ack {
    pub mode: u8,
    pub role: u8,
    pub upstream: [u8; 32],
    pub downstream: [u8; 32],
    pub start: u64,
    pub end: u64,
    pub prev: [u8; 32],
    pub entries: Vec<Entry>,
    pub sig: [u8; 64],
}

impl Ack {
    /// The record without `sig`.
    pub fn body(&self) -> Vec<u8> {
        let mut o = Vec::with_capacity(ACK_FIXED + ENTRY_LEN * self.entries.len());
        o.extend_from_slice(&ACK_MAGIC);
        o.push(VERSION);
        o.push(self.mode);
        o.push(self.role);
        o.extend_from_slice(&self.upstream);
        o.extend_from_slice(&self.downstream);
        o.extend_from_slice(&self.start.to_le_bytes());
        o.extend_from_slice(&self.end.to_le_bytes());
        o.extend_from_slice(&self.prev);
        o.extend_from_slice(&u16::try_from(self.entries.len()).expect("at most 65535 entries").to_le_bytes());
        for e in &self.entries {
            e.encode(&mut o);
        }
        o
    }

    pub fn to_bytes(&self) -> Vec<u8> {
        let mut o = self.body();
        o.extend_from_slice(&self.sig);
        o
    }

    /// Its digest (§3.5).
    pub fn digest(&self) -> [u8; 32] {
        sha256(&[&self.body()])
    }

    /// The key that signs it: `downstream` in an acknowledgement, `upstream` in a declaration.
    pub fn signer(&self) -> [u8; 32] {
        if self.mode == mode::DECLARATION { self.upstream } else { self.downstream }
    }

    /// Signed with `key`, which should be [`Ack::signer`]'s.
    pub fn sign(mut self, key: &SigningKey) -> Ack {
        self.sig = sign(key, ACK_DOMAIN, &self.body());
        self
    }

    /// §3.6: read a record, hold it to its rules, then its signature.
    pub fn parse(b: &[u8]) -> Result<Ack, Reason> {
        read_head(b, &ACK_MAGIC)?;
        let mut r = Rd { b, p: 5 };
        let (m, rl) = (r.u8()?, r.u8()?);
        let (upstream, downstream) = (r.arr()?, r.arr()?);
        let (start, end) = (r.u64()?, r.u64()?);
        let prev = r.arr()?;
        let n = r.u16()? as usize;
        let mut entries = Vec::with_capacity(n.min(b.len() / ENTRY_LEN));
        for _ in 0..n {
            entries.push(Entry::read(&mut r)?);
        }
        let sig = r.arr()?;
        if !r.done() {
            return Err(MALFORMED);
        }
        let a = Ack { mode: m, role: rl, upstream, downstream, start, end, prev, entries, sig };
        let rules = (a.mode == mode::ACK || a.mode == mode::DECLARATION)
            && (a.role == role::HOP || a.role == role::SINK)
            && a.start < a.end
            && a.upstream != [0; 32]
            && (a.mode == mode::DECLARATION || a.downstream != [0; 32])
            && !a.entries.is_empty()
            && a.entries.windows(2).all(|w| w[0].order() < w[1].order())
            && a.entries.iter().all(Entry::rules_hold);
        if !rules {
            return Err(Reason::AckInvalid);
        }
        if a.mode == mode::ACK && a.downstream == a.upstream {
            return Err(Reason::SelfAcknowledged);
        }
        verify_sig(&a.signer(), ACK_DOMAIN, &b[..b.len() - 64], &a.sig)?;
        Ok(a)
    }
}

// ---------------------------------------------------------------------------------------
// §5 Range proofs
// ---------------------------------------------------------------------------------------

/// `left(n)` (§5.1): the left child of a subtree of `n` > 1024 bytes.
fn left_len(n: u64) -> u64 {
    let q = (n - 1) / CHUNK;
    CHUNK << (63 - q.leading_zeros())
}

/// The non-root chaining value of the subtree `x` whose first chunk is at byte offset
/// `start` of the whole input.
fn subtree_cv(x: &[u8], start: u64) -> ChainingValue {
    let mut h = blake3::Hasher::new();
    h.set_input_offset(start);
    h.update(x);
    h.finalize_non_root()
}

/// The covered chunks of a range (§5.2), as offsets into `X`: (c0, c1), for an object of
/// `envelope_len` bytes. `None` when `X`'s length does not fit in a u64.
pub fn covered(envelope_len: u64, start: u64, end: u64) -> Option<(u64, u64)> {
    let l = DOMAIN_LEN.checked_add(envelope_len)?;
    let c0 = (DOMAIN_LEN as u128 + start as u128) / CHUNK as u128 * CHUNK as u128;
    let c1 = (DOMAIN_LEN as u128 + end as u128).div_ceil(CHUNK as u128) * CHUNK as u128;
    Some((c0 as u64, c1.min(l as u128) as u64))
}

/// A range proof (§5.2).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RangeProof {
    pub content_hash: [u8; 32],
    pub envelope_len: u64,
    pub start: u64,
    pub end: u64,
    pub cvs: Vec<[u8; 32]>,
    /// The covered bytes that are envelope (§5.2).
    pub bytes: Vec<u8>,
}

/// What a range proof that passes establishes (§5.5, §8.3).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct RangeFacts {
    pub start: u64,
    pub end: u64,
    pub envelope_len: u64,
    pub covered: (u64, u64),
    /// Whether the proof's last covered chunk is the object's last: a tail proof.
    pub length_established: bool,
}

/// The producer's walk: the chaining values of the subtrees wholly outside [c0, c1), in the
/// order [`RangeProof::verify`] consumes them.
fn collect_cvs(x: &[u8], s: u64, n: u64, c0: u64, c1: u64, out: &mut Vec<[u8; 32]>) {
    if s + n <= c0 || s >= c1 {
        out.push(subtree_cv(&x[s as usize..(s + n) as usize], s));
    } else if !(c0 <= s && s + n <= c1) {
        let l = left_len(n);
        collect_cvs(x, s, l, c0, c1, out);
        collect_cvs(x, s + l, n - l, c0, c1, out);
    }
}

/// §5.4's `cv(s, n)` over the covered chunks `xc`, which start at `c0` of `X`.
fn node_cv(xc: &[u8], s: u64, n: u64, c0: u64, c1: u64, it: &mut dyn Iterator<Item = [u8; 32]>) -> Result<ChainingValue, Reason> {
    if s + n <= c0 || s >= c1 {
        it.next().ok_or(Reason::RangeProofMismatch)
    } else if c0 <= s && s + n <= c1 {
        Ok(subtree_cv(&xc[(s - c0) as usize..(s - c0 + n) as usize], s))
    } else if n <= CHUNK {
        // A single chunk lies wholly inside or wholly outside the covered chunks (§5.4).
        Err(Reason::RangeProofMismatch)
    } else {
        let l = left_len(n);
        let left = node_cv(xc, s, l, c0, c1, it)?;
        let right = node_cv(xc, s + l, n - l, c0, c1, it)?;
        Ok(merge_subtrees_non_root(&left, &right, Mode::Hash))
    }
}

impl RangeProof {
    /// The proof of bytes `start`..`end` of `envelope`.
    pub fn build(envelope: &[u8], start: u64, end: u64) -> RangeProof {
        assert!(start < end && end <= envelope.len() as u64, "a range of the object");
        let mut x = DOMAIN_OBJECT.to_vec();
        x.extend_from_slice(envelope);
        let l = x.len() as u64;
        let (c0, c1) = covered(envelope.len() as u64, start, end).expect("an object in memory");
        let mut cvs = Vec::new();
        if !(c0 == 0 && c1 == l) {
            let lt = left_len(l);
            collect_cvs(&x, 0, lt, c0, c1, &mut cvs);
            collect_cvs(&x, lt, l - lt, c0, c1, &mut cvs);
        }
        let bytes = x[c0.max(DOMAIN_LEN) as usize..c1 as usize].to_vec();
        RangeProof { content_hash: content_hash(envelope), envelope_len: envelope.len() as u64, start, end, cvs, bytes }
    }

    pub fn to_bytes(&self) -> Vec<u8> {
        let mut o = Vec::with_capacity(RANGE_FIXED + 32 * self.cvs.len() + self.bytes.len());
        o.extend_from_slice(&RANGE_MAGIC);
        o.push(VERSION);
        o.extend_from_slice(&self.content_hash);
        o.extend_from_slice(&self.envelope_len.to_le_bytes());
        o.extend_from_slice(&self.start.to_le_bytes());
        o.extend_from_slice(&self.end.to_le_bytes());
        o.push(u8::try_from(self.cvs.len()).expect("at most 255 chaining values"));
        for cv in &self.cvs {
            o.extend_from_slice(cv);
        }
        o.extend_from_slice(&self.bytes);
        o
    }

    /// §5.3: read a range proof.
    pub fn read(b: &[u8]) -> Result<RangeProof, Reason> {
        read_head(b, &RANGE_MAGIC)?;
        if b.len() < RANGE_FIXED {
            return Err(TRUNCATED);
        }
        let mut r = Rd { b, p: 5 };
        let content_hash = r.arr()?;
        let (envelope_len, start, end) = (r.u64()?, r.u64()?, r.u64()?);
        let n = r.u8()? as usize;
        if start >= end {
            return Err(Reason::RangeInvalid);
        }
        if end > envelope_len {
            return Err(Reason::RangeOutsideObject);
        }
        let (c0, c1) = covered(envelope_len, start, end).ok_or(Reason::RangeOutsideObject)?;
        let want = RANGE_FIXED as u128 + 32 * n as u128 + (c1 - c0.max(DOMAIN_LEN)) as u128;
        if (b.len() as u128) < want {
            return Err(TRUNCATED);
        }
        if (b.len() as u128) > want {
            return Err(MALFORMED);
        }
        let cvs = (0..n).map(|_| r.arr()).collect::<Result<Vec<[u8; 32]>, Reason>>()?;
        let bytes = b[r.p..].to_vec();
        Ok(RangeProof { content_hash, envelope_len, start, end, cvs, bytes })
    }

    /// The covered chunks (§5.2) as offsets into `X`.
    pub fn covered(&self) -> (u64, u64) {
        covered(self.envelope_len, self.start, self.end).expect("checked when read")
    }

    /// §5.4: check this proof against the content hash `h`, and against the item it answers
    /// an audit of (§5.6) where there is one.
    pub fn verify(&self, h: &[u8; 32], audit: Option<&Item>) -> Result<RangeFacts, Reason> {
        if self.content_hash != *h {
            return Err(Reason::RangeProofMismatch);
        }
        if let Some(i) = audit
            && (i.content_hash != *h || self.start < i.start || self.end > i.end)
        {
            return Err(Reason::RangeNotAcknowledged);
        }
        let l = DOMAIN_LEN + self.envelope_len;
        let (c0, c1) = self.covered();
        let mut xc = Vec::with_capacity((c1 - c0) as usize);
        if c0 == 0 {
            xc.extend_from_slice(DOMAIN_OBJECT);
        }
        xc.extend_from_slice(&self.bytes);
        let root = if c0 == 0 && c1 == l {
            if !self.cvs.is_empty() {
                return Err(Reason::RangeProofMismatch);
            }
            *blake3::hash(&xc).as_bytes()
        } else {
            let mut it = self.cvs.iter().copied();
            let lt = left_len(l);
            let left = node_cv(&xc, 0, lt, c0, c1, &mut it)?;
            let right = node_cv(&xc, lt, l - lt, c0, c1, &mut it)?;
            if it.next().is_some() {
                return Err(Reason::RangeProofMismatch);
            }
            *merge_subtrees_root(&left, &right, Mode::Hash).as_bytes()
        };
        if root != *h {
            return Err(Reason::RangeProofMismatch);
        }
        Ok(RangeFacts { start: self.start, end: self.end, envelope_len: self.envelope_len, covered: (c0, c1), length_established: c1 == l })
    }
}

/// §5.3 then §5.4 over the bytes of a range proof.
pub fn check_range(b: &[u8], h: &[u8; 32], audit: Option<&Item>) -> Result<RangeFacts, Reason> {
    RangeProof::read(b)?.verify(h, audit)
}

// ---------------------------------------------------------------------------------------
// §7 The settlement statement
// ---------------------------------------------------------------------------------------

/// A line (§7.2).
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
pub struct Line {
    pub kind: u8,
    pub payee: [u8; 32],
    pub class: u8,
    pub payer: [u8; 32],
    pub objects: u64,
    pub bytes: u64,
    pub amount: u64,
}

impl Line {
    fn encode(&self, o: &mut Vec<u8>) {
        o.push(self.kind);
        o.extend_from_slice(&self.payee);
        o.push(self.class);
        o.extend_from_slice(&self.payer);
        o.extend_from_slice(&self.objects.to_le_bytes());
        o.extend_from_slice(&self.bytes.to_le_bytes());
        o.extend_from_slice(&self.amount.to_le_bytes());
    }

    fn read(r: &mut Rd) -> Result<Line, Reason> {
        Ok(Line { kind: r.u8()?, payee: r.arr()?, class: r.u8()?, payer: r.arr()?, objects: r.u64()?, bytes: r.u64()?, amount: r.u64()? })
    }

    fn key(&self) -> (u8, [u8; 32], u8, [u8; 32]) {
        (self.kind, self.payee, self.class, self.payer)
    }

    /// A delivery line of a retrieval class whose payer is zero (§7.2).
    pub fn is_free(&self) -> bool {
        self.kind == line_kind::DELIVERY && class::is_retrieval(self.class) && self.payer == [0; 32]
    }

    /// An attribution line (§7.8).
    pub fn is_attribution(&self) -> bool {
        self.kind == line_kind::ATTRIBUTION
    }
}

/// A tariff (§7.1): a currency, and a price per object class in millionths of a minor unit
/// per 10^9 bytes.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Tariff {
    pub currency: [u8; 3],
    pub prices: [u64; CLASSES],
}

impl Tariff {
    /// No tariff.
    pub fn none() -> Tariff {
        Tariff { currency: [0; 3], prices: [0; CLASSES] }
    }

    /// The amount of `bytes` of class `c` (§7.3), in integers wide enough.
    pub fn amount(&self, c: u8, bytes: u128) -> u128 {
        bytes * self.prices[(c - 1) as usize] as u128 / PRICE_BYTES
    }
}

/// A settlement statement (§7.1).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Statement {
    pub party: [u8; 32],
    pub start: u64,
    pub end: u64,
    pub records: Vec<[u8; 32]>,
    pub hops: Vec<[u8; 32]>,
    pub tariff: Tariff,
    /// Acknowledged (objects, bytes) per object class.
    pub classes: [(u64, u64); CLASSES],
    /// Declared: (declarations, objects, bytes).
    pub unacknowledged: (u32, u64, u64),
    pub unmetered: u16,
    pub totals: Vec<Total>,
    pub lines: Vec<Line>,
    pub amount: u64,
    pub sig: [u8; 64],
}

fn read_list(r: &mut Rd) -> Result<Vec<[u8; 32]>, Reason> {
    let n = r.u16()? as usize;
    (0..n).map(|_| r.arr()).collect()
}

fn put_list(o: &mut Vec<u8>, l: &[[u8; 32]]) {
    o.extend_from_slice(&u16::try_from(l.len()).expect("at most 65535").to_le_bytes());
    for d in l {
        o.extend_from_slice(d);
    }
}

impl Statement {
    /// The statement without `sig`.
    pub fn body(&self) -> Vec<u8> {
        let mut o = Vec::new();
        o.extend_from_slice(&STATEMENT_MAGIC);
        o.push(VERSION);
        o.extend_from_slice(&self.party);
        o.extend_from_slice(&self.start.to_le_bytes());
        o.extend_from_slice(&self.end.to_le_bytes());
        put_list(&mut o, &self.records);
        put_list(&mut o, &self.hops);
        o.extend_from_slice(&self.tariff.currency);
        for p in self.tariff.prices {
            o.extend_from_slice(&p.to_le_bytes());
        }
        for (n, b) in self.classes {
            o.extend_from_slice(&n.to_le_bytes());
            o.extend_from_slice(&b.to_le_bytes());
        }
        o.extend_from_slice(&self.unacknowledged.0.to_le_bytes());
        o.extend_from_slice(&self.unacknowledged.1.to_le_bytes());
        o.extend_from_slice(&self.unacknowledged.2.to_le_bytes());
        o.extend_from_slice(&self.unmetered.to_le_bytes());
        o.push(u8::try_from(self.totals.len()).expect("at most 255 totals"));
        for t in &self.totals {
            o.push(t.tag);
            o.extend_from_slice(&t.figures.to_le_bytes());
            o.extend_from_slice(&t.joules.to_le_bytes());
        }
        o.extend_from_slice(&u16::try_from(self.lines.len()).expect("at most 65535 lines").to_le_bytes());
        for l in &self.lines {
            l.encode(&mut o);
        }
        o.extend_from_slice(&self.amount.to_le_bytes());
        o
    }

    pub fn to_bytes(&self) -> Vec<u8> {
        let mut o = self.body();
        o.extend_from_slice(&self.sig);
        o
    }

    /// Its digest (§7.4).
    pub fn digest(&self) -> [u8; 32] {
        sha256(&[&self.body()])
    }

    /// Signed with `key`; `party` is left as it is.
    pub fn sign(mut self, key: &SigningKey) -> Statement {
        self.sig = sign(key, STATEMENT_DOMAIN, &self.body());
        self
    }

    /// §7.5: read a statement, hold it to its rules, then its signature.
    pub fn parse(b: &[u8]) -> Result<Statement, Reason> {
        read_head(b, &STATEMENT_MAGIC)?;
        let mut r = Rd { b, p: 5 };
        let party = r.arr()?;
        let (start, end) = (r.u64()?, r.u64()?);
        let records = read_list(&mut r)?;
        let hops = read_list(&mut r)?;
        let currency = r.arr()?;
        let mut prices = [0u64; CLASSES];
        for p in &mut prices {
            *p = r.u64()?;
        }
        let mut classes = [(0u64, 0u64); CLASSES];
        for c in &mut classes {
            *c = (r.u64()?, r.u64()?);
        }
        let unacknowledged = (r.u32()?, r.u64()?, r.u64()?);
        let unmetered = r.u16()?;
        let m = r.u8()? as usize;
        let mut totals = Vec::with_capacity(m);
        for _ in 0..m {
            totals.push(Total { tag: r.u8()?, figures: r.u32()?, joules: r.u64()? });
        }
        let p = r.u16()? as usize;
        let mut lines = Vec::with_capacity(p.min(b.len() / LINE_LEN));
        for _ in 0..p {
            lines.push(Line::read(&mut r)?);
        }
        let amount = r.u64()?;
        let sig = r.arr()?;
        if !r.done() {
            return Err(MALFORMED);
        }
        let s = Statement { party, start, end, records, hops, tariff: Tariff { currency, prices }, classes, unacknowledged, unmetered, totals, lines, amount, sig };
        if s.totals.iter().any(|t| t.tag == 0) {
            return Err(CLASSES_SUMMED);
        }
        let ascending = |l: &[[u8; 32]]| l.windows(2).all(|w| w[0] < w[1]);
        let tariff_ok = if s.tariff.currency == [0; 3] { s.tariff.prices.iter().all(|&p| p == 0) } else { s.tariff.currency.iter().all(u8::is_ascii_uppercase) };
        let rules = s.start < s.end
            && !s.records.is_empty()
            && ascending(&s.records)
            && ascending(&s.hops)
            && tariff_ok
            && s.totals.iter().all(|t| (2..=5).contains(&t.tag) && t.figures > 0)
            && s.totals.windows(2).all(|w| w[0].tag < w[1].tag)
            && s.lines.iter().all(|l| {
                let kind_ok = match l.kind {
                    line_kind::DELIVERY => l.payee == s.party,
                    line_kind::ATTRIBUTION => l.payee != [0; 32] && l.payer != [0; 32],
                    _ => false,
                };
                kind_ok && (1..=CLASSES as u8).contains(&l.class) && l.objects > 0
            })
            && s.lines.windows(2).all(|w| w[0].key() < w[1].key());
        if !rules {
            return Err(Reason::StatementInvalid);
        }
        verify_sig(&s.party, STATEMENT_DOMAIN, &b[..b.len() - 64], &s.sig)?;
        Ok(s)
    }
}

/// A basis an attribution line is paid from (§7.8; influence §6.1), as the verifier of the
/// attribution accepted it.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Basis {
    /// The output's content hash.
    pub output: [u8; 32],
    /// The whole, in units.
    pub whole: u128,
    /// The payable buckets.
    pub payees: Vec<BasisPayee>,
}

/// A payable bucket of a basis: a key, its units and its tariff (§7.8).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct BasisPayee {
    pub key: [u8; 32],
    pub units: u128,
    /// Millionths of a minor unit of the statement's currency per delivery.
    pub per_use: u64,
    /// Parts per million of the revenue.
    pub share_ppm: u64,
}

/// Attribution lines per (payee, class, basis): (objects, bytes, amount) (§7.8).
pub type AttributionLines = BTreeMap<([u8; 32], u8, [u8; 32]), (u128, u128, u128)>;
/// A basis's output's deliveries per (basis, class): (deliveries, bytes) (§7.8).
pub type Deliveries = BTreeMap<([u8; 32], u8), (u128, u128)>;

/// The figures of §7.6, recomputed, in integers wide enough not to overflow.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Figures {
    /// Per (class, payer): (objects, bytes, amount).
    pub lines: BTreeMap<(u8, [u8; 32]), (u128, u128, u128)>,
    /// The attribution lines (§7.8).
    pub attribution: AttributionLines,
    /// The deliveries of each basis's output (§7.8).
    pub deliveries: Deliveries,
    pub classes: [(u128, u128); CLASSES],
    pub unacknowledged: (u128, u128, u128),
    pub unmetered: u128,
    /// Per tag: (figures, joules).
    pub totals: BTreeMap<u8, (u128, u128)>,
    pub amount: u128,
}

/// §7.6, §7.7: the figures of a statement whose records are `acks` and whose hop records are
/// `hops`, under `tariff`.
pub fn figures(acks: &[Ack], hops: &[HopRecord], tariff: &Tariff) -> Figures {
    let mut lines: BTreeMap<(u8, [u8; 32]), (u128, u128, u128)> = BTreeMap::new();
    let mut unack = (0u128, 0u128, 0u128);
    for a in acks {
        if a.mode == mode::DECLARATION {
            unack.0 += 1;
            for e in &a.entries {
                unack.1 += e.objects as u128;
                unack.2 += e.bytes as u128;
            }
            continue;
        }
        for e in &a.entries {
            let payer = if class::is_retrieval(e.class) { e.grant } else { a.downstream };
            let l = lines.entry((e.class, payer)).or_default();
            l.0 += e.objects as u128;
            l.1 += e.bytes as u128;
        }
    }
    let mut classes = [(0u128, 0u128); CLASSES];
    let mut amount = 0u128;
    for ((c, payer), l) in lines.iter_mut() {
        let free = class::is_retrieval(*c) && *payer == [0; 32];
        l.2 = if free { 0 } else { tariff.amount(*c, l.1) };
        classes[(*c - 1) as usize].0 += l.0;
        classes[(*c - 1) as usize].1 += l.1;
        amount += l.2;
    }
    let mut totals: BTreeMap<u8, (u128, u128)> = BTreeMap::new();
    let mut unmetered = 0u128;
    for h in hops {
        let tag = metering::tag_of(h.class.as_ref());
        if tag == 0 {
            unmetered += 1;
        } else {
            let t = totals.entry(tag).or_default();
            t.0 += 1;
            t.1 += h.joules as u128;
        }
    }
    Figures { lines, attribution: BTreeMap::new(), deliveries: BTreeMap::new(), classes, unacknowledged: unack, unmetered, totals, amount }
}

/// ⌊R × s × u / (10^6 × T)⌋ with u ≤ T, exactly: with y = R × s = a × d + r, d = 10^6 × T,
/// it is a × u + ⌊r × u / d⌋, and a × u ≤ y / 10^6 and r × u < d × T hold in a u128 for every
/// figure §7.8 admits.
pub fn share_of(revenue: u128, share_ppm: u64, units: u128, whole: u128) -> u128 {
    let y = revenue * share_ppm as u128;
    let d = 1_000_000 * whole;
    let (a, r) = (y / d, y % d);
    a * units + r * units / d
}

/// §7.8: the attribution lines of the bases `bases` (each a digest and the basis), over the
/// acknowledgements `acks` (mode 1 counts; declarations do not) whose entries' items are `items`,
/// under `tariff`; and each basis's deliveries per class.
pub fn attribution_lines(acks: &[Ack], items: &[Vec<Vec<Item>>], tariff: &Tariff, bases: &[([u8; 32], &Basis)]) -> (AttributionLines, Deliveries) {
    let mut lines = BTreeMap::new();
    let mut deliveries = BTreeMap::new();
    for (digest, b) in bases {
        for c in 1..=CLASSES as u8 {
            let (mut d, mut y, mut paid) = (0u128, 0u128, 0u128);
            for (a, lists) in acks.iter().zip(items) {
                if a.mode != mode::ACK {
                    continue;
                }
                let mut held = false;
                for (e, list) in a.entries.iter().zip(lists) {
                    if e.class != c {
                        continue;
                    }
                    let free = class::is_retrieval(c) && e.grant == [0; 32];
                    for it in list.iter().filter(|it| it.content_hash == b.output) {
                        held = true;
                        y += it.bytes() as u128;
                        if !free {
                            paid += it.bytes() as u128;
                        }
                    }
                }
                d += held as u128;
            }
            if d == 0 {
                continue;
            }
            deliveries.insert((*digest, c), (d, y));
            let revenue = paid * tariff.prices[(c - 1) as usize] as u128 / PRICE_BYTES;
            for p in &b.payees {
                let l = lines.entry((p.key, c, *digest)).or_insert((d, y, 0u128));
                l.2 += p.per_use as u128 * d + share_of(revenue, p.share_ppm, p.units, b.whole);
            }
        }
    }
    (lines, deliveries)
}

/// The revenue of a basis's output in class `c` (§7.8): ⌊paid bytes × price / 10^9⌋.
pub fn revenue(acks: &[Ack], items: &[Vec<Vec<Item>>], tariff: &Tariff, output: &[u8; 32], c: u8) -> u128 {
    let mut paid = 0u128;
    for (a, lists) in acks.iter().zip(items) {
        if a.mode != mode::ACK {
            continue;
        }
        for (e, list) in a.entries.iter().zip(lists) {
            if e.class == c && !(class::is_retrieval(c) && e.grant == [0; 32]) {
                paid += list.iter().filter(|it| it.content_hash == *output).map(|it| it.bytes() as u128).sum::<u128>();
            }
        }
    }
    paid * tariff.prices[(c - 1) as usize] as u128 / PRICE_BYTES
}

/// The payers of a statement's attribution lines, in the order the lines first name them
/// (§8.2 step 6).
pub fn basis_order(s: &Statement) -> Vec<[u8; 32]> {
    let mut out: Vec<[u8; 32]> = Vec::new();
    for l in s.lines.iter().filter(|l| l.is_attribution()) {
        if !out.contains(&l.payer) {
            out.push(l.payer);
        }
    }
    out
}

/// Add the attribution lines of `bases` to a statement computed by [`compute`] (§7.8), over its
/// records and their item lists; unsigned. Panics where a figure does not fit its field.
pub fn with_attribution(mut s: Statement, records: &[(Ack, Vec<Vec<Item>>)], bases: &[([u8; 32], &Basis)]) -> Statement {
    let acks: Vec<Ack> = records.iter().map(|(a, _)| a.clone()).collect();
    let items: Vec<Vec<Vec<Item>>> = records.iter().map(|(_, i)| i.clone()).collect();
    let (lines, _) = attribution_lines(&acks, &items, &s.tariff, bases);
    let u = |x: u128| u64::try_from(x).expect("a figure that fits");
    s.lines.retain(|l| !l.is_attribution());
    for ((payee, c, basis), (n, b, a)) in lines {
        s.lines.push(Line { kind: line_kind::ATTRIBUTION, payee, class: c, payer: basis, objects: u(n), bytes: u(b), amount: u(a) });
    }
    s.lines.sort_by_key(Line::key);
    s
}

/// The statement of `party` for the period, settling `records` (each with its digest) and
/// `hops`, under `tariff`; unsigned. Panics where a figure does not fit its field.
pub fn compute(party: [u8; 32], start: u64, end: u64, records: &[([u8; 32], Ack)], hops: &[([u8; 32], HopRecord)], tariff: Tariff) -> Statement {
    let acks: Vec<Ack> = records.iter().map(|(_, a)| a.clone()).collect();
    let hrs: Vec<HopRecord> = hops.iter().map(|(_, h)| h.clone()).collect();
    let f = figures(&acks, &hrs, &tariff);
    let u = |x: u128| u64::try_from(x).expect("a figure that fits");
    let mut rec: Vec<[u8; 32]> = records.iter().map(|(d, _)| *d).collect();
    rec.sort();
    let mut hd: Vec<[u8; 32]> = hops.iter().map(|(d, _)| *d).collect();
    hd.sort();
    let mut lines: Vec<Line> = f
        .lines
        .iter()
        .map(|((c, payer), (n, b, a))| Line { kind: line_kind::DELIVERY, payee: party, class: *c, payer: *payer, objects: u(*n), bytes: u(*b), amount: u(*a) })
        .collect();
    lines.sort_by_key(Line::key);
    Statement {
        party,
        start,
        end,
        records: rec,
        hops: hd,
        tariff,
        classes: f.classes.map(|(n, b)| (u(n), u(b))),
        unacknowledged: (u32::try_from(f.unacknowledged.0).expect("fits"), u(f.unacknowledged.1), u(f.unacknowledged.2)),
        unmetered: u16::try_from(f.unmetered).expect("fits"),
        totals: f.totals.iter().map(|(&tag, &(n, j))| Total { tag, figures: u32::try_from(n).expect("fits"), joules: u(j) }).collect(),
        lines,
        amount: u(f.amount),
        sig: [0; 64],
    }
}

/// §7.6: compare a statement with its recomputed figures, in the order of that section.
pub fn check_figures(s: &Statement, f: &Figures) -> Result<(), Reason> {
    let delivery: Vec<&Line> = s.lines.iter().filter(|l| !l.is_attribution()).collect();
    let attribution: Vec<&Line> = s.lines.iter().filter(|l| l.is_attribution()).collect();
    // 1. A free line bills nothing.
    if s.lines.iter().any(|l| l.is_free() && l.amount != 0) {
        return Err(Reason::PaidWithoutGrant);
    }
    // 2. No line bills what no acknowledgement covers.
    for l in &delivery {
        match f.lines.get(&(l.class, l.payer)) {
            None => return Err(Reason::BytesNotAcknowledged),
            Some(&(n, b, _)) if l.objects as u128 > n || l.bytes as u128 > b => return Err(Reason::BytesNotAcknowledged),
            _ => {}
        }
    }
    for l in &attribution {
        match f.deliveries.get(&(l.payer, l.class)) {
            None => return Err(Reason::BytesNotAcknowledged),
            Some(&(n, b)) if l.objects as u128 > n || l.bytes as u128 > b => return Err(Reason::BytesNotAcknowledged),
            _ => {}
        }
    }
    // 3. A total summed across classes.
    let classes: Vec<u8> = f.totals.keys().copied().collect();
    for t in &s.totals {
        if let Some(&(_, sum)) = f.totals.get(&t.tag)
            && t.joules as u128 != sum
        {
            let others: Vec<u8> = classes.iter().copied().filter(|&x| x != t.tag).collect();
            let summed = (1u32..(1 << others.len())).any(|mask| {
                let extra: u128 = others.iter().enumerate().filter(|(i, _)| mask & (1 << i) != 0).map(|(_, x)| f.totals[x].1).sum();
                extra + sum == t.joules as u128
            });
            if summed {
                return Err(CLASSES_SUMMED);
            }
        }
    }
    // 4. Anything else.
    let same_lines = delivery.len() == f.lines.len()
        && delivery.iter().all(|l| f.lines.get(&(l.class, l.payer)) == Some(&(l.objects as u128, l.bytes as u128, l.amount as u128)))
        && attribution.len() == f.attribution.len()
        && attribution.iter().all(|l| f.attribution.get(&(l.payee, l.class, l.payer)) == Some(&(l.objects as u128, l.bytes as u128, l.amount as u128)));
    let same_classes = s.classes.iter().zip(f.classes.iter()).all(|(a, b)| (a.0 as u128, a.1 as u128) == *b);
    let u = s.unacknowledged;
    let same_unack = (u.0 as u128, u.1 as u128, u.2 as u128) == f.unacknowledged;
    let same_totals = s.totals.len() == f.totals.len() && s.totals.iter().all(|t| f.totals.get(&t.tag) == Some(&(t.figures as u128, t.joules as u128)));
    if !(same_lines && same_classes && same_unack && s.unmetered as u128 == f.unmetered && same_totals && s.amount as u128 == f.amount) {
        return Err(Reason::StatementMismatch);
    }
    Ok(())
}

// ---------------------------------------------------------------------------------------
// §8 Verification
// ---------------------------------------------------------------------------------------

/// A room a verifier is given (§8.1): its descriptor and its grants in order.
#[derive(Clone, Debug, Default)]
pub struct RoomInput {
    pub descriptor: Vec<u8>,
    pub grants: Vec<Vec<u8>>,
}

/// What a statement verifier is given (§8.1).
#[derive(Clone, Debug, Default)]
pub struct Inputs {
    /// Records by digest.
    pub records: BTreeMap<[u8; 32], Vec<u8>>,
    /// Item lists by record digest, one per entry.
    pub items: BTreeMap<[u8; 32], Vec<Vec<u8>>>,
    /// Proof records by SHA-256.
    pub proofs: BTreeMap<[u8; 32], Vec<u8>>,
    /// Per-hop records by digest.
    pub hops: BTreeMap<[u8; 32], Vec<u8>>,
    pub rooms: Vec<RoomInput>,
    /// The bases its attribution lines name, by digest, each verified (§8.1; influence §6.2).
    pub bases: BTreeMap<[u8; 32], Basis>,
}

/// An accepted statement (§8.3).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Accepted {
    pub digest: [u8; 32],
    pub statement: Statement,
    /// `whole`, `partial` or `none` (§7.7).
    pub coverage: &'static str,
}

/// A refusal: the reason, and where (§8.2), as a JSON array.
pub type Refusal = (Reason, Value);

/// A room as a verifier holds it.
struct HeldRoom {
    descriptor: Descriptor,
    id: [u8; 32],
    grants: Vec<HeldGrant>,
}

fn grant_check(a: &Ack, e: &Entry, rooms: &[HeldRoom], proofs: &BTreeMap<[u8; 32], Vec<u8>>) -> Result<(), Reason> {
    // §6.3 step 1.
    let g = rooms.iter().flat_map(|r| r.grants.iter()).find(|h| h.digest == e.grant).ok_or(RECORD_MISSING)?;
    // Step 2: good throughout the window.
    let (nbf, exp) = (g.grant.nbf as u128 * 1_000_000, g.grant.exp as u128 * 1_000_000);
    if !(nbf <= a.start as u128 && a.end as u128 <= exp) {
        return Err(rm(room::Reason::OutsideGrantWindow));
    }
    if !class::is_retrieval(e.class) {
        // Step 3.
        if g.grant.principal != a.downstream {
            return Err(rm(room::Reason::GrantWrongPrincipal));
        }
        return Ok(());
    }
    // Step 4: live-rooms §10.5.4 steps 2 to 5.
    let pb = proofs.get(&e.proof).ok_or(RECORD_MISSING)?;
    let p = Proof::read(pb).map_err(|_| rm(room::Reason::PopMalformed))?;
    if p.grant != g.digest {
        return Err(rm(room::Reason::PopWrongGrant));
    }
    if p.key != g.grant.principal {
        return Err(rm(room::Reason::PopWrongKey));
    }
    let mut m = room::PROOF_DOMAIN.to_vec();
    m.extend_from_slice(&p.body());
    crate::sig::verify_strict(&p.key, &m, &p.sig).map_err(|_| rm(room::Reason::PopBadSignature))?;
    // Step 5.
    if e.class == class::FETCH && g.grant.caps.powers & room::power::SEEK == 0 {
        return Err(rm(room::Reason::SeekNotGranted));
    }
    Ok(())
}

fn overlaps(a: (u64, u64), b: (u64, u64)) -> bool {
    a.0 < b.1 && b.0 < a.1
}

/// §8.2: verify a statement against its inputs.
pub fn verify(statement: &[u8], i: &Inputs) -> Result<Accepted, Refusal> {
    // Step 1: the statement.
    let s = Statement::parse(statement).map_err(|r| (r, json!(["statement"])))?;
    // Step 2: the rooms.
    let mut rooms: Vec<HeldRoom> = Vec::new();
    for ri in &i.rooms {
        let d = Descriptor::parse(&ri.descriptor).map_err(|r| (Reason::from(r), json!(["room", hex(&given_digest(&ri.descriptor))])))?;
        let id = d.room_id();
        let grants = room::check_grants(&d, &id, &ri.grants).map_err(|(k, r)| (Reason::from(r), json!(["room", hex(&id), "grant", k])))?;
        rooms.push(HeldRoom { descriptor: d, id, grants });
    }
    // Step 3: each record.
    let mut acks: Vec<Ack> = Vec::with_capacity(s.records.len());
    let mut item_lists: Vec<Vec<Vec<Item>>> = Vec::with_capacity(s.records.len());
    for (k, d) in s.records.iter().enumerate() {
        let at = || json!(["record", k]);
        let b = i.records.get(d).ok_or((RECORD_MISSING, at()))?;
        let a = Ack::parse(b).map_err(|r| (r, at()))?;
        if a.upstream != s.party {
            return Err((Reason::AckWrongParty, at()));
        }
        if !(s.start <= a.start && a.end <= s.end) {
            return Err((Reason::AckOutsidePeriod, at()));
        }
        let lists = i.items.get(d);
        let mut read = Vec::with_capacity(a.entries.len());
        for (j, e) in a.entries.iter().enumerate() {
            let at = || json!(["record", k, "entry", j]);
            let list = lists.and_then(|l| l.get(j)).ok_or((RECORD_MISSING, at()))?;
            let items = read_items(list).map_err(|r| (r, at()))?;
            if !e.recomputes(&items) {
                return Err((Reason::ItemsMismatch, at()));
            }
            if e.grant != [0; 32] {
                grant_check(&a, e, &rooms, &i.proofs).map_err(|r| (r, at()))?;
            }
            read.push(items);
        }
        acks.push(a);
        item_lists.push(read);
    }
    // Step 4: the chains.
    let digests = &s.records;
    let mut chains: BTreeMap<(u8, [u8; 32], [u8; 32]), Vec<usize>> = BTreeMap::new();
    for (k, a) in acks.iter().enumerate() {
        chains.entry((a.mode, a.upstream, a.downstream)).or_default().push(k);
    }
    let mut before: BTreeMap<usize, usize> = BTreeMap::new();
    for ks in chains.values_mut() {
        ks.sort_by_key(|&k| (acks[k].start, k));
        for w in ks.windows(2) {
            before.insert(w[1], w[0]);
        }
    }
    for (k, a) in acks.iter().enumerate() {
        if let Some(&p) = before.get(&k) {
            if acks[p].end > a.start {
                return Err((Reason::AckOverlap, json!(["record", k])));
            }
            if a.prev != digests[p] {
                return Err((Reason::AckChainBroken, json!(["record", k])));
            }
        }
    }
    let all_items = |k: usize| -> Vec<Item> { item_lists[k].iter().flatten().copied().collect() };
    for (k, a) in acks.iter().enumerate() {
        if a.mode != mode::DECLARATION || a.downstream == [0; 32] {
            continue;
        }
        let declared = all_items(k);
        for (q, b) in acks.iter().enumerate() {
            if b.mode != mode::ACK || b.downstream != a.downstream || !overlaps((a.start, a.end), (b.start, b.end)) {
                continue;
            }
            let acked = all_items(q);
            if declared.iter().any(|x| acked.iter().any(|y| x.content_hash == y.content_hash && overlaps((x.start, x.end), (y.start, y.end)))) {
                return Err((Reason::DeclaredAndAcknowledged, json!(["record", k])));
            }
        }
    }
    let mut named: BTreeMap<[u8; 32], usize> = BTreeMap::new();
    for (k, a) in acks.iter().enumerate() {
        for (j, e) in a.entries.iter().enumerate() {
            if e.proof == [0; 32] {
                continue;
            }
            match named.get(&e.proof) {
                Some(&q) if q != k => return Err((rm(room::Reason::PopReplayed), json!(["record", k, "entry", j]))),
                Some(_) => {}
                None => {
                    named.insert(e.proof, k);
                }
            }
        }
    }
    // Step 5: each hop record.
    let mut hops: Vec<HopRecord> = Vec::with_capacity(s.hops.len());
    for (k, d) in s.hops.iter().enumerate() {
        let at = || json!(["hop", k]);
        let b = i.hops.get(d).ok_or((RECORD_MISSING, at()))?;
        let named_room: [u8; 32] = b.get(5..37).and_then(|x| x.try_into().ok()).unwrap_or([0; 32]);
        let h = HopRecord::parse(b, &named_room).map_err(|r| (Reason::from(r), at()))?;
        let r = rooms.iter().find(|r| r.id == h.room).ok_or((RECORD_MISSING, at()))?;
        if h.hop != s.party {
            return Err((Reason::HopNotParty, at()));
        }
        let t0 = r.descriptor.t0_us as u128;
        if !(s.start as u128 <= t0 + h.start as u128 && t0 + h.end as u128 <= s.end as u128) {
            return Err((Reason::SpanOutsidePeriod, at()));
        }
        hops.push(h);
    }
    // Step 6: the bases.
    let mut bases: Vec<([u8; 32], &Basis)> = Vec::new();
    for (k, d) in basis_order(&s).iter().enumerate() {
        let b = i.bases.get(d).ok_or((RECORD_MISSING, json!(["basis", k])))?;
        bases.push((*d, b));
    }
    // Step 7: the figures.
    let mut f = figures(&acks, &hops, &s.tariff);
    (f.attribution, f.deliveries) = attribution_lines(&acks, &item_lists, &s.tariff, &bases);
    check_figures(&s, &f).map_err(|r| (r, json!(["statement"])))?;
    let coverage = if hops.is_empty() {
        "none"
    } else if s.unmetered == 0 {
        "whole"
    } else {
        "partial"
    };
    Ok(Accepted { digest: s.digest(), statement: s, coverage })
}

/// The JSON facts of an accepted statement (§8.3), as the corpus states them.
pub fn accepted_json(a: &Accepted) -> Value {
    let s = &a.statement;
    let currency = if s.tariff.currency == [0; 3] { Value::Null } else { Value::String(String::from_utf8(s.tariff.currency.to_vec()).expect("capitals")) };
    let classes: serde_json::Map<String, Value> =
        (1..=CLASSES as u8).map(|c| (class::name(c).to_string(), json!({"objects": s.classes[(c - 1) as usize].0, "bytes": s.classes[(c - 1) as usize].1}))).collect();
    let lines: Vec<Value> = s
        .lines
        .iter()
        .filter(|l| !l.is_attribution())
        .map(|l| json!({"class": class::name(l.class), "payer": hex(&l.payer), "objects": l.objects, "bytes": l.bytes, "amount": l.amount, "free": l.is_free()}))
        .collect();
    let attribution: Vec<Value> = s
        .lines
        .iter()
        .filter(|l| l.is_attribution())
        .map(|l| json!({"payee": hex(&l.payee), "class": class::name(l.class), "basis": hex(&l.payer), "objects": l.objects, "bytes": l.bytes, "amount": l.amount}))
        .collect();
    let totals: Vec<Value> = s.totals.iter().map(|t| json!({"class": metering::tag_label(t.tag), "figures": t.figures, "joules": t.joules})).collect();
    let mut out = json!({
        "outcome": "accept",
        "digest": hex(&a.digest),
        "currency": currency,
        "classes": classes,
        "lines": lines,
        "unacknowledged": {"declarations": s.unacknowledged.0, "objects": s.unacknowledged.1, "bytes": s.unacknowledged.2},
        "coverage": a.coverage,
        "unmetered": s.unmetered,
        "totals": totals,
        "amount": s.amount,
    });
    if !attribution.is_empty() {
        out["attribution"] = Value::Array(attribution);
    }
    out
}

/// The JSON outcome of a statement's verification.
pub fn outcome_json(r: &Result<Accepted, Refusal>) -> Value {
    match r {
        Ok(a) => accepted_json(a),
        Err((reason, at)) => json!({"outcome": "refuse", "reason": reason.code(), "at": at}),
    }
}

/// The JSON outcome of a range proof's check (§8.3).
pub fn range_json(r: &Result<RangeFacts, Reason>) -> Value {
    match r {
        Ok(f) => json!({
            "outcome": "accept",
            "start": f.start,
            "end": f.end,
            "envelope_len": f.envelope_len,
            "covered": [f.covered.0, f.covered.1],
            "length_established": f.length_established,
        }),
        Err(reason) => json!({"outcome": "refuse", "reason": reason.code()}),
    }
}

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

    fn bytes(n: usize, seed: u8) -> Vec<u8> {
        (0..n).map(|i| (i as u32).wrapping_mul(2_654_435_761).to_le_bytes()[i % 4] ^ seed).collect()
    }

    /// Every range of objects of many lengths, around every chunk boundary, proves and
    /// verifies, and its root is the content hash BLAKE3 computes over the whole object.
    #[test]
    fn range_proofs_verify_against_the_content_hash() {
        for n in [1usize, 15, 1008, 1009, 1010, 2033, 2034, 3000, 4096, 5121, 9000, 20_000] {
            let env = bytes(n, 7);
            let h = content_hash(&env);
            let mut points: Vec<u64> = vec![0, 1, n as u64 / 2, n as u64];
            for k in 0..=(n as u64 + 15) / CHUNK {
                for d in [0i64, -1, 1] {
                    let p = (k * CHUNK) as i64 - 15 + d;
                    if p >= 0 && p <= n as i64 {
                        points.push(p as u64);
                    }
                }
            }
            points.sort();
            points.dedup();
            for &s in &points {
                for &e in &points {
                    if s >= e {
                        continue;
                    }
                    let p = RangeProof::build(&env, s, e);
                    let b = p.to_bytes();
                    let facts = check_range(&b, &h, None).unwrap_or_else(|r| panic!("n {n} [{s}, {e}): {r}"));
                    assert_eq!(facts.length_established, facts.covered.1 == 15 + n as u64);
                    assert_eq!(RangeProof::read(&b).unwrap(), p);
                }
            }
        }
    }

    /// A changed byte, a dropped, extra or changed chaining value, and another object's
    /// bytes each fail.
    #[test]
    fn a_tampered_range_proof_fails() {
        let env = bytes(9_000, 3);
        let other = bytes(9_000, 4);
        let h = content_hash(&env);
        let p = RangeProof::build(&env, 3_000, 4_500);
        assert!(!p.cvs.is_empty());
        let mut q = p.clone();
        q.bytes[10] ^= 1;
        assert_eq!(q.verify(&h, None), Err(Reason::RangeProofMismatch));
        let mut q = p.clone();
        q.cvs.pop();
        assert_eq!(q.verify(&h, None), Err(Reason::RangeProofMismatch));
        let mut q = p.clone();
        q.cvs.push([0; 32]);
        assert_eq!(q.verify(&h, None), Err(Reason::RangeProofMismatch));
        let mut q = p.clone();
        q.cvs[0][0] ^= 1;
        assert_eq!(q.verify(&h, None), Err(Reason::RangeProofMismatch));
        let mut q = p.clone();
        q.bytes = RangeProof::build(&other, 3_000, 4_500).bytes;
        assert_eq!(q.verify(&h, None), Err(Reason::RangeProofMismatch));
        let mut b = p.to_bytes();
        b.push(0);
        assert_eq!(RangeProof::read(&b), Err(MALFORMED));
        let b = p.to_bytes();
        assert_eq!(RangeProof::read(&b[..b.len() - 1]), Err(TRUNCATED));
    }

    /// The tree root is jwp-receipts §2.2's: a single leaf is its own root, and an odd level
    /// pairs its last node with itself.
    #[test]
    fn tree_root_is_the_group_tree() {
        let h = |i: u8| [i; 32];
        assert_eq!(tree_root(&[]), [0; 32]);
        assert_eq!(tree_root(&[h(1)]), leaf(&h(1)));
        let (a, b, c) = (leaf(&h(1)), leaf(&h(2)), leaf(&h(3)));
        assert_eq!(tree_root(&[h(1), h(2), h(3)]), node(&node(&a, &b), &node(&c, &c)));
    }

    /// A record round-trips, a self-acknowledgement is refused, and a retrieval entry with a
    /// grant and no proof breaks the rules.
    #[test]
    fn records_read_and_refuse() {
        let k = SigningKey::from_bytes(&[7; 32]);
        let up = SigningKey::from_bytes(&[8; 32]).verifying_key().to_bytes();
        let items = [Item { content_hash: [1; 32], start: 0, end: 100 }];
        let a = Ack { mode: mode::ACK, role: role::SINK, upstream: up, downstream: k.verifying_key().to_bytes(), start: 0, end: 10, prev: [0; 32], entries: vec![Entry::of(class::LIVE, [0; 32], [0; 32], &items)], sig: [0; 64] }.sign(&k);
        assert_eq!(Ack::parse(&a.to_bytes()).unwrap(), a);
        let me = Ack { upstream: k.verifying_key().to_bytes(), ..a.clone() }.sign(&k);
        assert_eq!(Ack::parse(&me.to_bytes()), Err(Reason::SelfAcknowledged));
        let bad = Ack { entries: vec![Entry::of(class::FETCH, [2; 32], [0; 32], &items)], ..a.clone() }.sign(&k);
        assert_eq!(Ack::parse(&bad.to_bytes()), Err(Reason::AckInvalid));
        assert_eq!(read_items(&item_list(&[items[0], Item { start: 50, end: 60, ..items[0] }])), Err(Reason::ItemsInvalid));
        assert!(read_items(&item_list(&[items[0], Item { start: 100, end: 160, ..items[0] }])).is_ok());
    }

    /// An amount is rounded down once per line, and a free line bills nothing.
    #[test]
    fn amounts_round_down_per_line() {
        let t = Tariff { currency: *b"USD", prices: [2_000_000, 5_000_000, 8_000_000, 1_000_000] };
        assert_eq!(t.amount(class::LIVE, 1_234_567), 2_469);
        assert_eq!(t.amount(class::FILL, 999), 7);
        assert_eq!(Tariff::none().amount(class::FETCH, 1 << 40), 0);
    }
}