//! Bearer receipts — `extensions/bearer-receipts.md`, the record `wai.bearer.receipts`: one
//! receipt on every bearer.
//!
//! A publisher's group and object receipts are the same bytes on a two-way relay chain, IP
//! multicast and a bearer with no return path. What differs is what can be said about
//! delivery, and that is said beside the receipt, in two claim classes whose bodies and
//! context checks are `crate::claim_receipt::bearer`'s: the publisher's `transmission`
//! claim, which declares the bearer and attests transmission, and a sink's
//! `reconciliation` claim, which attests the objects it received by a root under the
//! group's own tree. This module derives what a verifier may state about one group from
//! them ([`verdict`], §6), the upper bound on a receiver's share of a shared transmission's
//! energy (§7), and confirmation on a bearer with no return path ([`offline`], §8).
//!
//! # Rules enforced here rather than only documented
//!
//! - A group's object verdicts are computed before, and independently of, any declaration
//!   or reconciliation: they are the same on every bearer.
//! - No status is "delivered". A group is `published`, `transmitted` or `reconciled`, and a
//!   statement that it was delivered to *k* sinks holds only when *k* distinct keys
//!   reconciled every object of it.
//! - Two declarations of one bearer instance that differ set both aside; a reconciliation
//!   over an undeclared or contradicted instance is counted nowhere; a key is counted once
//!   per instance, its later reconciliations joined to its first.
//! - The per-receiver figure is a bound, rounded up, stated only where the transmission was
//!   metered, never as a measurement and never as 0.
//! - A key log that arrives on the bearer is held and anchors nothing.

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

use serde_json::Value;

use crate::claim_receipt::{Bearer, ClaimBody, ClaimReceipt, Reason as ClaimReason, Rejection};
use crate::jwp_delivery::{DeliveryError, GroupReceipt, ObjectReceipt};
use crate::key_log::{self, Anchor, AnchorPath, Anchoring, Item, Revision};

pub mod corpus;

/// The codes bearer-receipts §10 lists, in its order.
pub const CODES: &[&str] = &[
    "transmission_not_transmitted",
    "transmission_subject_mismatch",
    "transmission_signer_mismatch",
    "reconciliation_not_received",
    "reconciliation_bitmap_malformed",
    "reconciliation_subject_mismatch",
    "reconciliation_by_publisher",
    "reconciliation_leaf_count_mismatch",
    "reconciliation_objects_not_held",
    "reconciliation_root_mismatch",
    BEARER_CONTRADICTED,
    BEARER_NOT_DECLARED,
    SINK_NOT_CONFIRMED,
    DUPLICATE_SINK,
    DELIVERY_NOT_RECONCILED,
    DELIVERY_EXCEEDS_RECONCILED,
    ANCHOR_ON_BEARER,
];

pub const BEARER_CONTRADICTED: &str = "bearer_contradicted";
pub const BEARER_NOT_DECLARED: &str = "bearer_not_declared";
pub const SINK_NOT_CONFIRMED: &str = "sink_not_confirmed";
pub const DUPLICATE_SINK: &str = "reconciliation_duplicate_sink";
pub const DELIVERY_NOT_RECONCILED: &str = "delivery_not_reconciled";
pub const DELIVERY_EXCEEDS_RECONCILED: &str = "delivery_exceeds_reconciled";
pub const ANCHOR_ON_BEARER: &str = "anchor_on_bearer";

/// What a verifier may state about a group (§6.3).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Status {
    /// No declaration stands.
    Published,
    /// A declaration stands, and no reconciliation is counted.
    Transmitted,
    /// A reconciliation is counted.
    Reconciled,
}

impl Status {
    pub fn label(self) -> &'static str {
        match self {
            Status::Published => "published",
            Status::Transmitted => "transmitted",
            Status::Reconciled => "reconciled",
        }
    }
}

/// A statement offered about a group (§6.4).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Assertion {
    /// Transmitted over a bearer of this class.
    Transmitted(Bearer),
    /// Delivered to this many sinks, at least 1.
    Delivered(u64),
}

/// An object receipt the verifier holds, with the envelope's bytes where it holds them.
#[derive(Clone, Debug)]
pub struct HeldObject {
    pub receipt: Value,
    pub envelope: Option<Vec<u8>>,
}

/// The inputs of §6.1.
#[derive(Clone, Debug, Default)]
pub struct Inputs {
    /// The group receipt, in its JSON form (jwp-receipts §2.4.1).
    pub group: Value,
    pub objects: Vec<HeldObject>,
    /// The transmission and reconciliation claims, each signer's in the order of its chain.
    pub claims: Vec<String>,
    /// For a verifier that is a sink: the bearer class and instance it received over.
    pub sink: Option<(Bearer, String)>,
    pub assertion: Option<Assertion>,
    /// For a relying party that counts only keys it has confirmed: those keys.
    pub count_only: Option<Vec<[u8; 32]>>,
}

/// What became of one claim (§6.2 steps 3–5).
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum ClaimVerdict {
    /// Refused at a step of jwp-receipts §3.1.
    Refused(Rejection),
    /// A transmission claim that stands.
    Declared,
    /// A reconciliation counted for its instance.
    Counted,
    /// A second reconciliation of a key for one instance: its bits joined to the first's.
    Joined,
    /// Accepted, and set aside for the reason given.
    SetAside(&'static str),
}

impl ClaimVerdict {
    pub fn label(&self) -> String {
        match self {
            ClaimVerdict::Refused(r) => format!("reject step={} reason={}", r.step, r.reason.label()),
            ClaimVerdict::Declared => "declared".into(),
            ClaimVerdict::Counted => "counted".into(),
            ClaimVerdict::Joined => format!("joined reason={DUPLICATE_SINK}"),
            ClaimVerdict::SetAside(r) => format!("set-aside reason={r}"),
        }
    }
}

/// One declaration that stands, with its count and its bound (§6.2 step 6, §7).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct BearerLine {
    pub bearer_id: String,
    pub bearer: Bearer,
    /// Distinct keys counted for this instance.
    pub sinks: u64,
    /// The transmission's figure and its class's label, where it was metered.
    pub energy: Option<(u64, String)>,
    /// ⌈joules_micro / max(1, sinks)⌉, where it was metered; never 0 for an unmetered one.
    pub per_receiver_upper_bound: Option<u64>,
}

/// §6's verdict over one group.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Verdict {
    /// Each object: `delivered-verified`, `inclusion-verified`, or `reject reason=…`.
    pub objects: Vec<String>,
    pub claims: Vec<ClaimVerdict>,
    pub status: Status,
    pub sinks: u64,
    pub whole: u64,
    /// For each object, the distinct keys whose joined bits hold it.
    pub received_by: Vec<u64>,
    pub bearers: Vec<BearerLine>,
    /// §6.5, for a verifier that is a sink.
    pub sink: Option<&'static str>,
    /// §6.4: `Ok` when the statement holds.
    pub assertion: Option<Result<(), &'static str>>,
}

/// §7: ⌈joules / max(1, n)⌉.
pub fn per_receiver_upper_bound(joules_micro: u64, n: u64) -> u64 {
    joules_micro.div_ceil(n.max(1))
}

fn bit(bits: &[bool], i: usize) -> bool {
    bits.get(i).copied().unwrap_or(false)
}

/// A bearer instance that stands, as §6.2 step 5 counts reconciliations over it: a publisher's
/// declaration, which sends every object, or an instance another extension adds beside them
/// (cache-authority §7.2 step 4), which sends the objects it names.
#[derive(Clone, Debug)]
pub(crate) struct Instance {
    pub bearer: Bearer,
    /// The key that sent over it. A reconciliation over an instance signed by this key is set
    /// aside `self_acknowledged` (settlement §3.6); a publisher's own is refused before, at
    /// jwp-receipts §3.1 step 6.
    pub transmitter: [u8; 32],
}

/// §6.2 steps 5 and 6: what became of each accepted reconciliation, the bits each counted key
/// holds over each instance, and the counts joined per key.
#[derive(Clone, Debug, Default)]
pub(crate) struct Count {
    /// For each accepted reconciliation, by its position among the claims: counted, joined, or
    /// set aside.
    pub verdicts: Vec<(usize, ClaimVerdict)>,
    /// The bits each counted key holds, per instance: (key, `bearer_id`).
    pub bits: BTreeMap<([u8; 32], String), Vec<bool>>,
    /// Distinct keys counted, over every instance.
    pub sinks: u64,
    /// Keys whose bits, joined across every instance, hold every object.
    pub whole: u64,
    /// For each object, the distinct keys whose joined bits hold it.
    pub received_by: Vec<u64>,
}

impl Count {
    /// Distinct keys counted for the instance `id`.
    pub fn instance_sinks(&self, id: &str) -> u64 {
        self.bits.keys().filter(|(_, bid)| bid == id).count() as u64
    }

    /// The distinct keys whose bits over the instance `id` alone hold object `j`.
    #[cfg(feature = "cache_authority")]
    pub fn instance_received(&self, id: &str, j: u64) -> u64 {
        self.bits.iter().filter(|((_, bid), b)| bid == id && bit(b, j as usize)).count() as u64
    }
}

/// §6.2 steps 5 and 6 over the accepted claims of a group of `n` objects: the instances that
/// stand by `bearer_id`, and those whose declarations were set aside (`contradicted`). A
/// reconciliation over a contradicted instance, over none that stands, of another class than its
/// instance's, signed by its instance's transmitter, or by a key outside `count_only` when that is
/// given, is set aside; otherwise it is counted, and a key's second over one instance is joined.
pub(crate) fn count(
    n: u64,
    accepted: &[Option<ClaimReceipt>],
    standing: &BTreeMap<String, Instance>,
    contradicted: &BTreeSet<String>,
    count_only: Option<&[[u8; 32]]>,
) -> Count {
    let n = n as usize;
    let mut out = Count::default();
    for (i, c) in accepted.iter().enumerate() {
        let Some(c) = c else { continue };
        let Some(r) = c.reconciliation() else { continue };
        let aside = if contradicted.contains(&r.bearer_id) {
            Some(BEARER_CONTRADICTED)
        } else {
            match standing.get(&r.bearer_id) {
                None => Some(BEARER_NOT_DECLARED),
                Some(d) if d.bearer != r.bearer => Some(BEARER_CONTRADICTED),
                Some(d) if d.transmitter == c.signer_pubkey => Some(SELF_ACKNOWLEDGED),
                Some(_) => match count_only {
                    Some(keys) if !keys.contains(&c.signer_pubkey) => Some(SINK_NOT_CONFIRMED),
                    _ => None,
                },
            }
        };
        if let Some(reason) = aside {
            out.verdicts.push((i, ClaimVerdict::SetAside(reason)));
            continue;
        }
        let k = (c.signer_pubkey, r.bearer_id.clone());
        out.verdicts.push((i, if out.bits.contains_key(&k) { ClaimVerdict::Joined } else { ClaimVerdict::Counted }));
        let b = out.bits.entry(k).or_insert_with(|| vec![false; n]);
        for j in r.indices() {
            b[j as usize] = true;
        }
    }
    let mut joined: BTreeMap<[u8; 32], Vec<bool>> = BTreeMap::new();
    for ((key, _), b) in &out.bits {
        let into = joined.entry(*key).or_insert_with(|| vec![false; n]);
        for (x, y) in into.iter_mut().zip(b) {
            *x |= *y;
        }
    }
    out.sinks = joined.len() as u64;
    out.whole = joined.values().filter(|b| b.iter().all(|x| *x)).count() as u64;
    out.received_by = (0..n).map(|j| joined.values().filter(|b| bit(b, j)).count() as u64).collect();
    out
}

/// settlement §3.6's code: a party's statement that it received what it sent.
pub(crate) const SELF_ACKNOWLEDGED: &str = "self_acknowledged";

/// §6.2 steps 1 and 2: the group under jwp-receipts §3 steps 1–2 (the error, when it fails),
/// the object receipts that read, and each object's verdict in the order given, the same
/// whatever the bearer.
pub(crate) fn group_and_objects(group: &Value, held: &[HeldObject]) -> Result<(GroupReceipt, Vec<ObjectReceipt>, Vec<String>), DeliveryError> {
    // Step 1.
    let group = GroupReceipt::from_json(group)?;
    group.verify()?;
    // Step 2: the same whatever the bearer.
    let mut receipts: Vec<ObjectReceipt> = Vec::new();
    let objects: Vec<String> = held
        .iter()
        .map(|o| {
            let r = match ObjectReceipt::from_json(&o.receipt) {
                Ok(r) => r,
                Err(e) => return format!("reject reason={}", e.reason.label()),
            };
            let v = r.verify_against(&group, o.envelope.as_deref());
            receipts.push(r);
            match v {
                Ok(ev) => ev.label().to_owned(),
                Err(e) => format!("reject reason={}", e.reason.label()),
            }
        })
        .collect();
    Ok((group, receipts, objects))
}

/// §6.2: the verdict over one group. A group that fails jwp-receipts §3 steps 1–2 is the
/// error, and nothing is derived from it.
pub fn verdict(inp: &Inputs) -> Result<Verdict, DeliveryError> {
    let (group, receipts, objects) = group_and_objects(&inp.group, &inp.objects)?;
    let refs: Vec<&ObjectReceipt> = receipts.iter().collect();

    // Step 3.
    let mut held: BTreeMap<[u8; 32], ClaimReceipt> = BTreeMap::new();
    let mut claims: Vec<ClaimVerdict> = Vec::with_capacity(inp.claims.len());
    let mut accepted: Vec<Option<ClaimReceipt>> = Vec::with_capacity(inp.claims.len());
    for wire in &inp.claims {
        let checked = (|| {
            let c = ClaimReceipt::from_json(wire)?;
            c.verify()?;
            let previous = held.insert(c.signer_pubkey, c.clone());
            c.check_parent(previous.as_ref())?;
            match &c.body {
                ClaimBody::Transmission(_) => c.check_transmission_against(&group)?,
                ClaimBody::Reconciliation(_) => c.check_reconciliation_against(&group, &refs)?,
                _ => return Err(Rejection { step: 6, reason: ClaimReason::ContextMismatch }),
            }
            Ok(c)
        })();
        match checked {
            Ok(c) => {
                claims.push(ClaimVerdict::Declared);
                accepted.push(Some(c));
            }
            Err(r) => {
                claims.push(ClaimVerdict::Refused(r));
                accepted.push(None);
            }
        }
    }

    // Step 4: the declarations, by instance.
    let mut by_instance: BTreeMap<String, Vec<usize>> = BTreeMap::new();
    for (i, c) in accepted.iter().enumerate() {
        if let Some(t) = c.as_ref().and_then(ClaimReceipt::transmission) {
            by_instance.entry(t.bearer_id.clone()).or_default().push(i);
        }
    }
    let mut standing: BTreeMap<String, usize> = BTreeMap::new();
    let mut contradicted: BTreeSet<String> = BTreeSet::new();
    for (id, idx) in &by_instance {
        let first = accepted[idx[0]].as_ref().and_then(ClaimReceipt::transmission).expect("a transmission");
        if idx.iter().all(|&i| accepted[i].as_ref().and_then(ClaimReceipt::transmission) == Some(first)) {
            standing.insert(id.clone(), idx[0]);
        } else {
            contradicted.insert(id.clone());
            for &i in idx {
                claims[i] = ClaimVerdict::SetAside(BEARER_CONTRADICTED);
            }
        }
    }
    let declaration = |id: &str| standing.get(id).and_then(|&i| accepted[i].as_ref()).and_then(ClaimReceipt::transmission);

    // Steps 5 and 6: the reconciliations, in order, and the counts.
    let instances: BTreeMap<String, Instance> = standing
        .keys()
        .map(|id| {
            let d = declaration(id).expect("a declaration that stands");
            (id.clone(), Instance { bearer: d.bearer, transmitter: group.signer_pubkey })
        })
        .collect();
    let counted = count(group.leaf_count, &accepted, &instances, &contradicted, inp.count_only.as_deref());
    for (i, v) in &counted.verdicts {
        claims[*i] = v.clone();
    }
    let (sinks, whole, received_by) = (counted.sinks, counted.whole, counted.received_by.clone());
    let bearers = standing
        .iter()
        .map(|(id, &i)| {
            let t = accepted[i].as_ref().and_then(ClaimReceipt::transmission).expect("a transmission");
            let k = counted.instance_sinks(id);
            let energy = t.energy.as_ref().map(|e| (e.joules_micro, e.energy_provenance.clone()));
            BearerLine {
                bearer_id: id.clone(),
                bearer: t.bearer,
                sinks: k,
                per_receiver_upper_bound: energy.as_ref().map(|(j, _)| per_receiver_upper_bound(*j, k)),
                energy,
            }
        })
        .collect::<Vec<_>>();
    let status = if sinks > 0 {
        Status::Reconciled
    } else if !standing.is_empty() {
        Status::Transmitted
    } else {
        Status::Published
    };

    // §6.5.
    let sink = inp.sink.as_ref().map(|(b, id)| {
        if contradicted.contains(id) {
            return BEARER_CONTRADICTED;
        }
        match declaration(id) {
            None => BEARER_NOT_DECLARED,
            Some(d) if d.bearer != *b => BEARER_CONTRADICTED,
            Some(_) => "declared",
        }
    });
    // §6.4.
    let assertion = inp.assertion.map(|a| match a {
        Assertion::Transmitted(b) => {
            if bearers.iter().any(|l| l.bearer == b) {
                Ok(())
            } else {
                Err(BEARER_NOT_DECLARED)
            }
        }
        Assertion::Delivered(k) => {
            if whole == 0 {
                Err(DELIVERY_NOT_RECONCILED)
            } else if whole < k {
                Err(DELIVERY_EXCEEDS_RECONCILED)
            } else {
                Ok(())
            }
        }
    });
    Ok(Verdict { objects, claims, status, sinks, whole, received_by, bearers, sink, assertion })
}

// ───────────────────────────── §8: no return path ─────────────────────────────

/// How a key log reached the sink (§8.1, §8.2).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Via {
    /// Provisioned in advance by the relying party.
    Pinned,
    /// Arrived over the bearer.
    Bearer,
}

impl Via {
    pub fn label(self) -> &'static str {
        match self {
            Via::Pinned => "pinned",
            Via::Bearer => "bearer",
        }
    }
}

/// A key-log document the sink holds, and how it came to hold it.
#[derive(Clone, Debug)]
pub struct LogInput {
    pub origin: String,
    pub via: Via,
    pub log: String,
}

/// The item a sink confirms (§8.3).
#[derive(Clone, Debug)]
pub enum OfflineItem {
    /// A group receipt and the object receipts under it, in JSON.
    Group { group: Value, objects: Vec<Value> },
    /// A claim's wire form.
    Claim(String),
}

/// §8.3's outcome: each log's, and the item's confirmation with its figure's where it has
/// one.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OfflineVerdict {
    pub logs: Vec<String>,
    pub confirmation: String,
    pub figure: Option<String>,
}

/// §8.3. An item that fails its own verification is the error: `refuse reason=…` for a
/// group or an object, `reject step=… reason=…` for a claim.
pub fn offline(item: &OfflineItem, logs: &[LogInput]) -> Result<OfflineVerdict, String> {
    // Step 1.
    let (item, figure) = match item {
        OfflineItem::Group { group, objects } => {
            let refuse = |e: DeliveryError| format!("refuse reason={}", e.reason.label());
            let g = GroupReceipt::from_json(group).map_err(refuse)?;
            let os = objects.iter().map(ObjectReceipt::from_json).collect::<Result<Vec<_>, _>>().map_err(refuse)?;
            let refs: Vec<&ObjectReceipt> = os.iter().collect();
            (Item::from_group(&g, &refs).map_err(refuse)?, None)
        }
        OfflineItem::Claim(wire) => {
            let reject = |r: Rejection| format!("reject step={} reason={}", r.step, r.reason.label());
            let c = ClaimReceipt::from_json(wire).map_err(reject)?;
            c.verify().map_err(reject)?;
            (Item::from_claim(&c), Item::figure_of_claim(&c))
        }
    };
    // Step 2: the provisioned logs, anchored as pinned. A revision held twice is held once.
    let mut revisions: Vec<Revision> = Vec::new();
    let hold = |revisions: &mut Vec<Revision>, revs: Vec<Revision>| {
        for r in revs {
            if !revisions.iter().any(|h| h.revision_hash() == r.revision_hash()) {
                revisions.push(r);
            }
        }
    };
    let mut anchors: Vec<Anchor> = Vec::new();
    let mut outcomes: Vec<Option<String>> = vec![None; logs.len()];
    for (i, l) in logs.iter().enumerate().filter(|(_, l)| l.via == Via::Pinned) {
        outcomes[i] = Some(match key_log::anchor_log(&l.origin, &l.log, AnchorPath::Pinned, None) {
            Anchoring::Anchored(a) => {
                hold(&mut revisions, key_log::read_log(&l.origin, &l.log).expect("an anchored log reads"));
                anchors.push(a);
                "anchored".into()
            }
            Anchoring::NotAnchored(r) => format!("not-anchored reason={}", r.label()),
            Anchoring::Replaced => "not-anchored reason=key_log_replaced".into(),
        });
    }
    // Step 3: the logs that arrived on the bearer, held and never anchored.
    for (i, l) in logs.iter().enumerate().filter(|(_, l)| l.via == Via::Bearer) {
        outcomes[i] = Some(match key_log::read_log(&l.origin, &l.log) {
            Ok(revs) => {
                hold(&mut revisions, revs);
                if anchors.iter().any(|a| a.origin == l.origin) { "held".into() } else { ANCHOR_ON_BEARER.into() }
            }
            Err(r) => format!("unread reason={}", r.label()),
        });
    }
    // Step 4.
    let confirmation = key_log::confirm(&item, &revisions, &anchors, None).verdict();
    let figure = figure.map(|f| key_log::confirm(&f, &revisions, &anchors, None).verdict());
    Ok(OfflineVerdict { logs: outcomes.into_iter().map(|o| o.expect("every log has an outcome")).collect(), confirmation, figure })
}

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

    #[test]
    fn the_bound_rounds_up_and_never_divides_by_zero() {
        assert_eq!(per_receiver_upper_bound(9_000_000, 0), 9_000_000);
        assert_eq!(per_receiver_upper_bound(9_000_000, 1), 9_000_000);
        assert_eq!(per_receiver_upper_bound(10_000_000, 3), 3_333_334);
        assert_eq!(per_receiver_upper_bound(9_000_000, 3), 3_000_000);
        assert_eq!(per_receiver_upper_bound(1, 1_000), 1, "a bound of a metered figure is never 0");
    }

    #[test]
    fn the_codes_are_listed_once() {
        let set: BTreeSet<&str> = CODES.iter().copied().collect();
        assert_eq!(set.len(), CODES.len());
    }
}
