//! Holdings — `extensions/holdings.md`, the record `wai.delivery.holdings`: a node's signed
//! statement of which objects of which groups it holds now, for its neighbours.
//!
//! A record (`WHD1`) lists, for each group a node holds objects of, the group's link hash, its
//! leaf count and either `whole` or a bitmap of the objects held ([`Holdings::read`], §4). This
//! module holds records to an instant and to the group receipts a verifier has ([`verdict`],
//! §5.1 to §5.3), answers whether a record holds an object ([`holds`], §5.4) and derives which
//! wanted objects each neighbour holds and which must come from the origin ([`plan`], §5.4).
//!
//! # Rules enforced here rather than only documented
//!
//! - A holding has one form: a form-0 entry sets at least one bit, not every bit, and none past
//!   its leaf count, so two records stating the same holdings are the same bytes.
//! - Nothing here is evidence of an object's bytes: an object a neighbour serves is verified
//!   under its receipts, whoever serves it (§6).

use std::collections::BTreeMap;

use ed25519_dalek::{Signer, SigningKey};
use serde_json::Value;
use sha2::{Digest, Sha256};

use crate::jwp_delivery::GroupReceipt;

pub mod corpus;

/// The codes holdings §7 lists, in its order.
pub const CODES: &[&str] = &[
    MALFORMED,
    SIGNATURE_INVALID,
    NOT_YET,
    EXPIRED,
    LEAF_COUNT_MISMATCH,
    CONTRADICTED,
    OBJECT_NOT_HELD,
    OBJECT_OUT_OF_RANGE,
    GROUP_NOT_ADVERTISED,
];

pub const MALFORMED: &str = "holdings_malformed";
pub const SIGNATURE_INVALID: &str = "holdings_signature_invalid";
pub const NOT_YET: &str = "holdings_not_yet";
pub const EXPIRED: &str = "holdings_expired";
pub const LEAF_COUNT_MISMATCH: &str = "holdings_leaf_count_mismatch";
pub const CONTRADICTED: &str = "holdings_contradicted";
pub const OBJECT_NOT_HELD: &str = "object_not_held";
pub const OBJECT_OUT_OF_RANGE: &str = "object_out_of_range";
pub const GROUP_NOT_ADVERTISED: &str = "group_not_advertised";

/// §4.1: the magic.
pub const MAGIC: &[u8; 4] = b"WHD1";
/// §4.3: what the signature covers, before the record's bytes.
pub const DOMAIN: &[u8] = b"wai:holdings\x01";
/// §4.4: the most bytes a record has.
pub const MAX_LEN: usize = 1 << 20;
/// 2^53 − 1: the greatest instant and leaf count a record states.
pub const MAX_INT: u64 = (1 << 53) - 1;

/// What a holder holds of one group (§4.2).
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Held {
    /// Form 0: the objects whose bits are set.
    Bitmap(Vec<u8>),
    /// Form 1: every object.
    Whole,
}

/// One entry of a record (§4.2).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Entry {
    pub group: [u8; 32],
    pub leaf_count: u64,
    pub held: Held,
}

impl Entry {
    /// The entry that holds the objects `held` of a group of `leaf_count`, in its one form:
    /// whole when it holds them all. `None` when it holds none.
    pub fn of(group: [u8; 32], leaf_count: u64, held: &[u64]) -> Option<Entry> {
        let mut bits = vec![0u8; leaf_count.div_ceil(8) as usize];
        for &i in held.iter().filter(|&&i| i < leaf_count) {
            bits[(i / 8) as usize] |= 0x80 >> (i % 8);
        }
        let set = (0..leaf_count).filter(|&i| bit(&bits, i)).count() as u64;
        match set {
            0 => None,
            n if n == leaf_count => Some(Entry { group, leaf_count, held: Held::Whole }),
            _ => Some(Entry { group, leaf_count, held: Held::Bitmap(bits) }),
        }
    }

    /// Whether object `i` is held, and why not.
    pub fn holds(&self, i: u64) -> Result<(), &'static str> {
        if i >= self.leaf_count {
            return Err(OBJECT_OUT_OF_RANGE);
        }
        match &self.held {
            Held::Whole => Ok(()),
            Held::Bitmap(b) if bit(b, i) => Ok(()),
            Held::Bitmap(_) => Err(OBJECT_NOT_HELD),
        }
    }
}

fn bit(b: &[u8], i: u64) -> bool {
    b.get((i / 8) as usize).is_some_and(|x| x & (0x80 >> (i % 8)) != 0)
}

/// A holdings record (§4).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Holdings {
    pub holder: [u8; 32],
    pub as_of_ms: u64,
    pub expires_ms: u64,
    pub entries: Vec<Entry>,
    pub sig: [u8; 64],
}

impl Holdings {
    /// A record of `entries`, signed by `key`; the entries sorted by group. Not checked: a
    /// corpus writes records that break the rules too.
    pub fn seal(key: &SigningKey, as_of_ms: u64, expires_ms: u64, mut entries: Vec<Entry>) -> Holdings {
        entries.sort_by_key(|e| e.group);
        let mut h = Holdings { holder: key.verifying_key().to_bytes(), as_of_ms, expires_ms, entries, sig: [0; 64] };
        h.sig = key.sign(&[DOMAIN, &h.signed_bytes()].concat()).to_bytes();
        h
    }

    /// The bytes from `magic` through the last entry: what the digest and the signature cover.
    pub fn signed_bytes(&self) -> Vec<u8> {
        let mut b = Vec::new();
        b.extend_from_slice(MAGIC);
        b.push(1);
        b.extend_from_slice(&self.holder);
        b.extend_from_slice(&self.as_of_ms.to_be_bytes());
        b.extend_from_slice(&self.expires_ms.to_be_bytes());
        b.extend_from_slice(&(self.entries.len() as u32).to_be_bytes());
        for e in &self.entries {
            b.extend_from_slice(&e.group);
            b.extend_from_slice(&e.leaf_count.to_be_bytes());
            match &e.held {
                Held::Bitmap(bits) => {
                    b.push(0);
                    b.extend_from_slice(bits);
                }
                Held::Whole => b.push(1),
            }
        }
        b
    }

    pub fn to_bytes(&self) -> Vec<u8> {
        [self.signed_bytes(), self.sig.to_vec()].concat()
    }

    /// §4.3: SHA-256 over the signed bytes.
    pub fn digest(&self) -> [u8; 32] {
        Sha256::digest(self.signed_bytes()).into()
    }

    /// §4.4: read `bytes` as one record, its rules, then its signature.
    pub fn read(bytes: &[u8]) -> Result<Holdings, &'static str> {
        let h = Holdings::parse(bytes).ok_or(MALFORMED)?;
        crate::sig::verify_strict(&h.holder, &[DOMAIN, &h.signed_bytes()].concat(), &h.sig).map_err(|_| SIGNATURE_INVALID)?;
        Ok(h)
    }

    /// The layout and rules 1 to 5, in the order the bytes are read.
    fn parse(bytes: &[u8]) -> Option<Holdings> {
        if bytes.len() > MAX_LEN {
            return None;
        }
        let mut r = Reader { b: bytes, at: 0 };
        if r.take(4)? != MAGIC || r.take(1)? != [1] {
            return None;
        }
        let holder: [u8; 32] = r.take(32)?.try_into().ok()?;
        let as_of_ms = r.u64()?;
        let expires_ms = r.u64()?;
        if as_of_ms >= expires_ms || expires_ms > MAX_INT {
            return None;
        }
        let n = u32::from_be_bytes(r.take(4)?.try_into().ok()?);
        if n == 0 {
            return None;
        }
        let mut entries: Vec<Entry> = Vec::new();
        for _ in 0..n {
            let group: [u8; 32] = r.take(32)?.try_into().ok()?;
            if entries.last().is_some_and(|e| e.group >= group) {
                return None;
            }
            let leaf_count = r.u64()?;
            if leaf_count == 0 || leaf_count > MAX_INT {
                return None;
            }
            let held = match r.take(1)?[0] {
                0 => {
                    let bits = r.take(leaf_count.div_ceil(8) as usize)?.to_vec();
                    let past = (leaf_count..bits.len() as u64 * 8).any(|i| bit(&bits, i));
                    let set = (0..leaf_count).filter(|&i| bit(&bits, i)).count() as u64;
                    if past || set == 0 || set == leaf_count {
                        return None;
                    }
                    Held::Bitmap(bits)
                }
                1 => Held::Whole,
                _ => return None,
            };
            entries.push(Entry { group, leaf_count, held });
        }
        let sig: [u8; 64] = r.take(64)?.try_into().ok()?;
        (r.at == bytes.len()).then_some(Holdings { holder, as_of_ms, expires_ms, entries, sig })
    }

    /// The entry for `group`, if the record has one.
    pub fn entry(&self, group: &[u8; 32]) -> Option<&Entry> {
        self.entries.binary_search_by(|e| e.group.cmp(group)).ok().map(|i| &self.entries[i])
    }
}

struct Reader<'a> {
    b: &'a [u8],
    at: usize,
}

impl<'a> Reader<'a> {
    fn take(&mut self, n: usize) -> Option<&'a [u8]> {
        let s = self.b.get(self.at..self.at.checked_add(n)?)?;
        self.at += n;
        Some(s)
    }
    fn u64(&mut self) -> Option<u64> {
        Some(u64::from_be_bytes(self.take(8)?.try_into().ok()?))
    }
}

/// §5.4: whether a record holds object `i` of `group`, and why not.
pub fn holds(h: &Holdings, group: &[u8; 32], i: u64) -> Result<(), &'static str> {
    h.entry(group).ok_or(GROUP_NOT_ADVERTISED)?.holds(i)
}

/// What became of one record at an instant (§5.1 to §5.3).
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum RecordVerdict {
    /// Refused, for the reason given.
    Refused(&'static str),
    /// It stands: its holder's latest at the instant.
    Stands,
    /// A record of its holder with a later `as_of_ms` stands.
    Superseded,
    /// Set aside: another of its holder's records at the same instant differs.
    SetAside(&'static str),
}

impl RecordVerdict {
    pub fn label(&self) -> String {
        match self {
            RecordVerdict::Refused(r) => format!("reject reason={r}"),
            RecordVerdict::Stands => "stands".into(),
            RecordVerdict::Superseded => "superseded".into(),
            RecordVerdict::SetAside(r) => format!("set-aside reason={r}"),
        }
    }
}

/// §5.1 to §5.3: the records, as bytes, at `at_ms`, against the group receipts given (those
/// that fail jwp-receipts §3 steps 1 and 2 are not held). Returns each record's verdict, in
/// order, and the records that stand, by their position.
pub fn verdict(records: &[Vec<u8>], receipts: &[Value], at_ms: u64) -> (Vec<RecordVerdict>, BTreeMap<usize, Holdings>) {
    let held: BTreeMap<[u8; 32], u64> = receipts
        .iter()
        .filter_map(|v| GroupReceipt::from_json(v).ok())
        .filter(|g| g.verify().is_ok())
        .map(|g| (g.link_hash(), g.leaf_count))
        .collect();
    let mut out = Vec::with_capacity(records.len());
    let mut read: BTreeMap<usize, Holdings> = BTreeMap::new();
    for (k, bytes) in records.iter().enumerate() {
        let v = Holdings::read(bytes).and_then(|h| {
            if at_ms < h.as_of_ms {
                return Err(NOT_YET);
            }
            if at_ms >= h.expires_ms {
                return Err(EXPIRED);
            }
            if h.entries.iter().any(|e| held.get(&e.group).is_some_and(|&n| n != e.leaf_count)) {
                return Err(LEAF_COUNT_MISMATCH);
            }
            Ok(h)
        });
        match v {
            Ok(h) => {
                read.insert(k, h);
                out.push(RecordVerdict::Stands);
            }
            Err(r) => out.push(RecordVerdict::Refused(r)),
        }
    }
    // §5.3: each holder's latest; two latest that differ set both aside.
    let mut by_holder: BTreeMap<[u8; 32], Vec<usize>> = BTreeMap::new();
    for (k, h) in &read {
        by_holder.entry(h.holder).or_default().push(*k);
    }
    for ks in by_holder.values() {
        let latest = ks.iter().map(|k| read[k].as_of_ms).max().expect("a holder has a record");
        let at_latest: Vec<usize> = ks.iter().copied().filter(|k| read[k].as_of_ms == latest).collect();
        let contradicted = at_latest.iter().any(|k| read[k].digest() != read[&at_latest[0]].digest());
        for k in ks {
            if read[k].as_of_ms < latest {
                out[*k] = RecordVerdict::Superseded;
            } else if contradicted {
                out[*k] = RecordVerdict::SetAside(CONTRADICTED);
            }
        }
    }
    let standing = read.into_iter().filter(|(k, _)| out[*k] == RecordVerdict::Stands).collect();
    (out, standing)
}

/// §5.4: for each wanted object (a group's link hash and a position), the holders whose record
/// stands and holds it, in the asker's order of `standing`; and the wanted objects no holder
/// holds, by position in `wanted`. A holder's records that stand are one (§5.3).
pub fn plan(standing: &[&Holdings], wanted: &[([u8; 32], u64)]) -> (Vec<Vec<[u8; 32]>>, Vec<usize>) {
    let mut holders: Vec<Vec<[u8; 32]>> = Vec::with_capacity(wanted.len());
    let mut from_origin = Vec::new();
    for (w, (group, i)) in wanted.iter().enumerate() {
        let mut who: Vec<[u8; 32]> = Vec::new();
        for h in standing {
            if holds(h, group, *i).is_ok() && !who.contains(&h.holder) {
                who.push(h.holder);
            }
        }
        if who.is_empty() {
            from_origin.push(w);
        }
        holders.push(who);
    }
    (holders, from_origin)
}

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

    fn key(seed: u8) -> SigningKey {
        SigningKey::from_bytes(&[seed; 32])
    }

    #[test]
    fn an_entry_has_one_form() {
        let g = [7; 32];
        assert_eq!(Entry::of(g, 9, &[0, 1, 2, 3, 4, 5, 6, 7, 8]).unwrap().held, Held::Whole);
        assert_eq!(Entry::of(g, 9, &[0, 8]).unwrap().held, Held::Bitmap(vec![0x80, 0x80]));
        assert!(Entry::of(g, 9, &[]).is_none());
        assert!(Entry::of(g, 9, &[9, 20]).is_none(), "positions past the group are not held");
    }

    #[test]
    fn a_record_reads_back_and_every_rule_refuses() {
        let k = key(3);
        let e = vec![Entry::of([2; 32], 9, &[0, 8]).unwrap(), Entry::of([1; 32], 4, &[0, 1, 2, 3]).unwrap()];
        let h = Holdings::seal(&k, 1_000, 2_000, e);
        assert_eq!(h.entries[0].group, [1; 32], "sorted by group");
        let b = h.to_bytes();
        assert_eq!(Holdings::read(&b), Ok(h.clone()));
        assert_eq!(Holdings::read(&b[..b.len() - 1]), Err(MALFORMED), "short");
        assert_eq!(Holdings::read(&[b.clone(), vec![0]].concat()), Err(MALFORMED), "trailing");
        assert_eq!(Holdings::read(&vec![0; MAX_LEN + 1]), Err(MALFORMED), "past the bound");
        let mut bad = b.clone();
        bad[100] ^= 1;
        assert!(Holdings::read(&bad).is_err());
        let mut sig = b.clone();
        let last = sig.len() - 1;
        sig[last] ^= 1;
        assert_eq!(Holdings::read(&sig), Err(SIGNATURE_INVALID));
        let same_time = Holdings::seal(&k, 2_000, 2_000, h.entries.clone());
        assert_eq!(Holdings::read(&same_time.to_bytes()), Err(MALFORMED), "expires not after as_of");
    }

    #[test]
    fn queries_and_a_plan() {
        let a = Holdings::seal(&key(3), 1, 9, vec![Entry::of([1; 32], 4, &[0, 1, 2, 3]).unwrap()]);
        let b = Holdings::seal(&key(4), 1, 9, vec![Entry::of([1; 32], 4, &[2]).unwrap(), Entry::of([2; 32], 2, &[1]).unwrap()]);
        assert_eq!(holds(&b, &[1; 32], 2), Ok(()));
        assert_eq!(holds(&b, &[1; 32], 1), Err(OBJECT_NOT_HELD));
        assert_eq!(holds(&b, &[1; 32], 4), Err(OBJECT_OUT_OF_RANGE));
        assert_eq!(holds(&a, &[2; 32], 0), Err(GROUP_NOT_ADVERTISED));
        let (who, origin) = plan(&[&b, &a], &[([1; 32], 2), ([1; 32], 0), ([2; 32], 0), ([3; 32], 0)]);
        assert_eq!(who[0], vec![b.holder, a.holder], "in the asker's order");
        assert_eq!(who[1], vec![a.holder]);
        assert_eq!(origin, vec![2, 3]);
    }

    /// §4.1's magic is the registry's: one row names it, a magic of this text's §4.1, and no
    /// other wire identifier of any kind is the same four letters in any case.
    #[test]
    fn the_magic_is_registered_once_and_shared_with_nothing() {
        use crate::codecs::registry::{WireKind, WIRE_IDENTIFIERS};
        let m = std::str::from_utf8(MAGIC).unwrap();
        let rows: Vec<_> = WIRE_IDENTIFIERS.iter().filter(|w| w.name.eq_ignore_ascii_case(m)).collect();
        assert_eq!(rows.len(), 1, "{m}: the registry has one row for it");
        assert_eq!((rows[0].kind, rows[0].document, rows[0].section), (WireKind::Magic, "extensions/holdings.md", "§4.1"));
    }
}
