minerva 0.2.0

extern crate alloc;

use alloc::collections::BTreeSet;
use alloc::vec::Vec;

use crate::metis::VersionVector;
use crate::metis::dot::Dot;
use crate::metis::dot_set::HAVE_SET_WIRE_HEADER_LEN;
use thiserror::Error;

use super::super::SealedEpoch;
use super::super::arrival::Admission;
use super::super::gate::{EpochAddress, InvalidEpochAddress};
use super::error::WireCutError;
use super::shared::{
    WIRE_ADDRESS_LEN, WIRE_DOT_LEN, cut_embedding_len, decode_cut, encode_address_into,
    encode_cut_into, encode_dot_into, read_address_parts, read_dot, read_u32,
};

/// Version tag for the canonical seal wire encoding.
const SEAL_WIRE_V1: u8 = 0x01;
/// Version tag for the admission-bearing sibling encoding (PRD 0028 R7,
/// ruling R-89): the version-1 layout followed by one admission section.
/// Emitted exactly when the record carries an admission, so each version
/// keeps its own canonical bijection; a version-1 reader refuses it as
/// [`SealDecodeError::UnknownVersion`], which is the fail-closed shed
/// guard. An admission never drops silently out of a spelling.
const SEAL_WIRE_V2: u8 = 0x02;
/// Version, winner, two empty sets, and an empty join.
pub(super) const SEAL_WIRE_MIN_LEN: usize = 1 + WIRE_ADDRESS_LEN + 4 + 4 + HAVE_SET_WIRE_HEADER_LEN;

/// The default admission-joiner ceiling: generous for any one boundary's
/// growth, small enough to bound the decoder's amplification, and
/// independent of the candidate ceiling (the two counts are unrelated).
/// Override with [`SealDecodeBudget::with_max_joiners`].
const SEAL_DEFAULT_MAX_JOINERS: usize = 1024;

/// Ceilings for what a [`SealRecord`] decoder may materialize.
///
/// Each retained set is checked independently in the frame's candidate
/// unit; a version-2 admission's joiners are checked against their own
/// ceiling. The unbudgeted door remains bounded by the input bytes.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub struct SealDecodeBudget {
    max_candidates: usize,
    max_joiners: usize,
}

impl SealDecodeBudget {
    /// Allows at most `max_candidates` rows in each retained set, with
    /// the default admission-joiner ceiling.
    #[must_use]
    pub const fn new(max_candidates: usize) -> Self {
        Self {
            max_candidates,
            max_joiners: SEAL_DEFAULT_MAX_JOINERS,
        }
    }

    /// Replaces the admission-joiner ceiling.
    #[must_use]
    pub const fn with_max_joiners(self, max_joiners: usize) -> Self {
        Self {
            max_joiners,
            ..self
        }
    }

    /// Maximum rows per retained set.
    #[must_use]
    pub const fn max_candidates(self) -> usize {
        self.max_candidates
    }

    /// Maximum joiners in a version-2 admission section.
    #[must_use]
    pub const fn max_joiners(self) -> usize {
        self.max_joiners
    }
}

/// Decode failure for a [`SealRecord`] wire frame.
#[non_exhaustive]
#[derive(Error, Debug, Clone, Copy, PartialEq, Eq)]
pub enum SealDecodeError {
    /// The leading version byte is not recognized.
    #[error("unknown seal wire version: {0:#04x}")]
    UnknownVersion(u8),
    /// The input is too short for its counts or carries trailing bytes.
    #[error("unexpected seal frame length: expected {expected}, found {found}")]
    UnexpectedLength {
        /// Required lower bound, or exact length for trailing input.
        expected: usize,
        /// Supplied input length.
        found: usize,
    },
    /// The winner or a candidate address is invalid.
    #[error("seal address: {0}")]
    Address(#[from] InvalidEpochAddress),
    /// The candidate set is not in canonical ascending order.
    #[error("seal candidates not ascending: {found} after {previous}")]
    NonAscendingCandidates {
        /// Last well-ordered address.
        previous: EpochAddress,
        /// Out-of-order address.
        found: EpochAddress,
    },
    /// A ledger row carries the non-dot counter zero.
    #[error("seal ledger dot at station {station} has the non-dot counter zero")]
    ZeroLedgerDot {
        /// Station attached to the malformed row.
        station: u32,
    },
    /// The declaration-dot ledger is not in canonical ascending order.
    #[error("seal ledger not ascending: {found} after {previous}")]
    NonAscendingLedger {
        /// Last well-ordered ledger dot.
        previous: Dot,
        /// Out-of-order ledger dot.
        found: Dot,
    },
    /// The candidate count exceeds the caller's budget.
    #[error("seal frame declares {count} candidates, past the budget's {budget}")]
    TooManyCandidates {
        /// Declared candidate count.
        count: u64,
        /// Caller budget.
        budget: u64,
    },
    /// The ledger count exceeds the caller's budget.
    #[error("seal frame declares {count} ledger dots, past the budget's {budget}")]
    TooManyLedgerDots {
        /// Declared ledger-dot count.
        count: u64,
        /// Caller budget.
        budget: u64,
    },
    /// The sealed-join embedding refused.
    #[error("sealed join: {0}")]
    Cut(#[from] WireCutError),
    /// A version-2 frame with an empty joiner set: an admission that
    /// admits nobody is not a boundary, and the version-1 spelling is the
    /// canonical one for an admissionless record.
    #[error("seal admission carries no joiners")]
    EmptyAdmission,
    /// The admission's joiner set is not in canonical ascending order.
    #[error("seal joiners not ascending: {found} after {previous}")]
    NonAscendingJoiners {
        /// Last well-ordered joiner.
        previous: u32,
        /// Out-of-order joiner.
        found: u32,
    },
    /// The joiner count exceeds the caller's admission-joiner ceiling.
    #[error("seal frame declares {count} joiners, past the budget's {budget}")]
    TooManyJoiners {
        /// Declared joiner count.
        count: u64,
        /// Caller budget.
        budget: u64,
    },
    /// The admission's base is not the record's own predecessor
    /// generation: a boundary rides exactly the record sealed one past
    /// its base (PRD 0028 R1), so any other base is a claim no seal can
    /// make.
    #[error("seal admission base at generation {base}, record at {record}")]
    AdmissionBaseMismatch {
        /// The admission base's generation.
        base: u64,
        /// The record's generation.
        record: u64,
    },
}

type DecodedSet<'a, T> = (BTreeSet<T>, &'a [u8]);

/// Reads a version-2 frame's admission section: the base address, then a
/// strictly ascending, nonempty joiner set, re-proving the one relation a
/// seal fixes. The base is the record's own predecessor generation.
/// Joiners have their own ceiling, independent of the candidate one (the
/// two counts are unrelated); the unbudgeted door stays bounded by the
/// input bytes.
fn read_admission<'a>(
    bytes: &[u8],
    rest: &'a [u8],
    declaration: EpochAddress,
    budget: Option<SealDecodeBudget>,
) -> Result<(Admission, &'a [u8]), SealDecodeError> {
    let short = |rest: &[u8]| SealDecodeError::UnexpectedLength {
        expected: bytes.len() + (WIRE_ADDRESS_LEN + 4).saturating_sub(rest.len()),
        found: bytes.len(),
    };
    let ((base_generation, base_dot), rest) =
        read_address_parts(rest).ok_or_else(|| short(rest))?;
    let base = EpochAddress::try_from_parts(base_generation, base_dot)?;
    if base_generation.checked_add(1) != Some(declaration.generation()) {
        return Err(SealDecodeError::AdmissionBaseMismatch {
            base: base_generation,
            record: declaration.generation(),
        });
    }
    let (count, mut rest) = read_u32(rest).ok_or_else(|| short(rest))?;
    if count == 0 {
        return Err(SealDecodeError::EmptyAdmission);
    }
    if (count as usize).saturating_mul(4) > rest.len() {
        return Err(short(rest));
    }
    if let Some(budget) = budget
        && count as usize > budget.max_joiners()
    {
        return Err(SealDecodeError::TooManyJoiners {
            count: u64::from(count),
            budget: budget.max_joiners() as u64,
        });
    }
    let mut joiners = BTreeSet::new();
    let mut previous: Option<u32> = None;
    for _ in 0..count {
        let (joiner, tail) = read_u32(rest).ok_or_else(|| short(rest))?;
        if previous.is_some_and(|previous| joiner <= previous) {
            return Err(SealDecodeError::NonAscendingJoiners {
                previous: previous.unwrap_or(0),
                found: joiner,
            });
        }
        previous = Some(joiner);
        let _ = joiners.insert(joiner);
        rest = tail;
    }
    let admission =
        Admission::new(base, joiners).map_err(|_refusal| SealDecodeError::EmptyAdmission)?;
    Ok((admission, rest))
}

/// Reads the strictly ascending retired-candidate set under its optional
/// caller budget.
fn read_candidate_set<'a>(
    bytes: &[u8],
    rest: &'a [u8],
    budget: Option<SealDecodeBudget>,
) -> Result<DecodedSet<'a, EpochAddress>, SealDecodeError> {
    let short = |rest: &[u8], needed: u64| SealDecodeError::UnexpectedLength {
        expected: usize::try_from(
            (u64::try_from(bytes.len() - rest.len()).unwrap_or(u64::MAX)).saturating_add(needed),
        )
        .unwrap_or(usize::MAX),
        found: bytes.len(),
    };
    let (count, rest) = read_u32(rest).ok_or_else(|| short(rest, 4))?;
    if let Some(budget) = budget
        && u64::from(count) > budget.max_candidates() as u64
    {
        return Err(SealDecodeError::TooManyCandidates {
            count: u64::from(count),
            budget: budget.max_candidates() as u64,
        });
    }

    let reserved = 4 + HAVE_SET_WIRE_HEADER_LEN;
    let backable = rest.len().saturating_sub(reserved) / WIRE_ADDRESS_LEN;
    if count as usize > backable {
        let floor = u64::from(count)
            .saturating_mul(WIRE_ADDRESS_LEN as u64)
            .saturating_add(reserved as u64);
        return Err(short(rest, floor));
    }

    let mut candidates = BTreeSet::new();
    let mut previous: Option<EpochAddress> = None;
    let mut rest = rest;
    for _ in 0..count {
        let ((generation, dot), tail) =
            read_address_parts(rest).ok_or_else(|| short(rest, WIRE_ADDRESS_LEN as u64))?;
        let candidate = EpochAddress::try_from_parts(generation, dot)?;
        if let Some(previous) = previous
            && candidate <= previous
        {
            return Err(SealDecodeError::NonAscendingCandidates {
                previous,
                found: candidate,
            });
        }
        previous = Some(candidate);
        let _ = candidates.insert(candidate);
        rest = tail;
    }
    Ok((candidates, rest))
}

/// Reads the strictly ascending protocol declaration-dot ledger under its
/// optional caller budget.
fn read_ledger_set<'a>(
    bytes: &[u8],
    rest: &'a [u8],
    budget: Option<SealDecodeBudget>,
) -> Result<DecodedSet<'a, Dot>, SealDecodeError> {
    let short = |rest: &[u8], needed: u64| SealDecodeError::UnexpectedLength {
        expected: usize::try_from(
            (u64::try_from(bytes.len() - rest.len()).unwrap_or(u64::MAX)).saturating_add(needed),
        )
        .unwrap_or(usize::MAX),
        found: bytes.len(),
    };
    let (count, rest) = read_u32(rest).ok_or_else(|| short(rest, 4))?;
    if let Some(budget) = budget
        && u64::from(count) > budget.max_candidates() as u64
    {
        return Err(SealDecodeError::TooManyLedgerDots {
            count: u64::from(count),
            budget: budget.max_candidates() as u64,
        });
    }

    let backable = rest.len().saturating_sub(HAVE_SET_WIRE_HEADER_LEN) / WIRE_DOT_LEN;
    if count as usize > backable {
        let floor = u64::from(count)
            .saturating_mul(WIRE_DOT_LEN as u64)
            .saturating_add(HAVE_SET_WIRE_HEADER_LEN as u64);
        return Err(short(rest, floor));
    }

    let mut protocol = BTreeSet::new();
    let mut previous: Option<Dot> = None;
    let mut rest = rest;
    for _ in 0..count {
        let (raw, tail) = read_dot(rest).ok_or_else(|| short(rest, WIRE_DOT_LEN as u64))?;
        // The counter-zero refusal stays exactly here, before the identity
        // exists; the order against the ascending check is unchanged, and
        // [`Dot`] then carries the law (ruling R-91).
        let Ok(dot) = Dot::from_parts(raw.0, raw.1) else {
            return Err(SealDecodeError::ZeroLedgerDot { station: raw.0 });
        };
        if let Some(previous) = previous
            && dot <= previous
        {
            return Err(SealDecodeError::NonAscendingLedger {
                previous,
                found: dot,
            });
        }
        previous = Some(dot);
        let _ = protocol.insert(dot);
        rest = tail;
    }
    Ok((protocol, rest))
}

/// A sealed epoch's record in wire shape (S284; the PRD 0025 kind-8
/// payload): the winning address, two distinct retained sets, and sealed
/// join.
///
/// The decoded value is a peer claim. Live receivers compare it with
/// their locally derived [`SealedEpoch`]; bootstrap installation remains
/// gated by [`Epochs::bootstrap`](crate::metis::Epochs::bootstrap).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct SealRecord {
    declaration: EpochAddress,
    candidates: BTreeSet<EpochAddress>,
    protocol: BTreeSet<Dot>,
    sealed_join: VersionVector,
    admission: Option<Admission>,
}

impl SealRecord {
    /// Spells a locally derived seal as a report record. Total: an
    /// admission-bearing record spells as the version-2 sibling frame, so
    /// nothing this door accepts can shed a membership boundary.
    #[must_use]
    pub fn from_sealed(sealed: &SealedEpoch) -> Self {
        Self {
            declaration: sealed.declaration,
            candidates: sealed.candidates.clone(),
            protocol: sealed.protocol.clone(),
            sealed_join: sealed.sealed_join.clone(),
            admission: sealed.admission.clone(),
        }
    }

    /// The membership boundary this record carries, or `None`.
    #[must_use]
    pub const fn admission(&self) -> Option<&Admission> {
        self.admission.as_ref()
    }

    /// The winning epoch address.
    #[must_use]
    pub const fn declaration(&self) -> EpochAddress {
        self.declaration
    }

    /// The sealed join of the adoption reports.
    #[must_use]
    pub const fn sealed_join(&self) -> &VersionVector {
        &self.sealed_join
    }

    /// The candidate addresses, winner included, ascending.
    pub fn candidates(&self) -> impl Iterator<Item = EpochAddress> + '_ {
        self.candidates.iter().copied()
    }

    /// The protocol declaration-dot ledger, ascending.
    pub fn declaration_dots(&self) -> impl Iterator<Item = Dot> + '_ {
        self.protocol.iter().copied()
    }

    /// Encodes this record as its canonical wire frame: version 1, or the
    /// version-2 sibling exactly when an admission rides.
    #[must_use]
    pub fn to_bytes(&self) -> Vec<u8> {
        let mut out = Vec::with_capacity(self.encoded_len());
        out.push(if self.admission.is_some() {
            SEAL_WIRE_V2
        } else {
            SEAL_WIRE_V1
        });
        encode_address_into(&mut out, self.declaration);
        let candidate_count = u32::try_from(self.candidates.len()).unwrap_or(u32::MAX);
        out.extend_from_slice(&candidate_count.to_be_bytes());
        for candidate in &self.candidates {
            encode_address_into(&mut out, *candidate);
        }
        let ledger_count = u32::try_from(self.protocol.len()).unwrap_or(u32::MAX);
        out.extend_from_slice(&ledger_count.to_be_bytes());
        for &dot in &self.protocol {
            encode_dot_into(&mut out, dot);
        }
        encode_cut_into(&mut out, &self.sealed_join);
        if let Some(admission) = &self.admission {
            encode_address_into(&mut out, admission.base());
            let joiner_count = u32::try_from(admission.joiners().count()).unwrap_or(u32::MAX);
            out.extend_from_slice(&joiner_count.to_be_bytes());
            for joiner in admission.joiners() {
                out.extend_from_slice(&joiner.to_be_bytes());
            }
        }
        out
    }

    /// The exact frame length, computed allocation-free.
    #[must_use]
    pub fn encoded_len(&self) -> usize {
        1 + WIRE_ADDRESS_LEN
            + 4
            + self.candidates.len() * WIRE_ADDRESS_LEN
            + 4
            + self.protocol.len() * WIRE_DOT_LEN
            + cut_embedding_len(&self.sealed_join)
            + self.admission.as_ref().map_or(0, |admission| {
                WIRE_ADDRESS_LEN + 4 + admission.joiners().count() * 4
            })
    }

    /// Decodes one canonical frame under the input-length ceiling.
    ///
    /// # Errors
    ///
    /// Returns [`SealDecodeError`] for malformed structure or content.
    pub fn from_bytes(bytes: &[u8]) -> Result<Self, SealDecodeError> {
        let (record, tail) = Self::from_prefix(bytes)?;
        Self::reject_tail(bytes, tail)?;
        Ok(record)
    }

    /// Decodes one canonical frame under a caller-owned retained-set
    /// budget.
    ///
    /// # Errors
    ///
    /// Returns [`SealDecodeError`] for budget, structure, or content
    /// refusals.
    pub fn from_bytes_with_budget(
        bytes: &[u8],
        budget: SealDecodeBudget,
    ) -> Result<Self, SealDecodeError> {
        let (record, tail) = Self::from_prefix_with_budget(bytes, Some(budget))?;
        Self::reject_tail(bytes, tail)?;
        Ok(record)
    }

    /// Decodes a self-delimiting frame prefix.
    ///
    /// # Errors
    ///
    /// Returns [`SealDecodeError`] for malformed structure or content.
    pub fn from_prefix(bytes: &[u8]) -> Result<(Self, &[u8]), SealDecodeError> {
        Self::from_prefix_with_budget(bytes, None)
    }

    /// Decode core shared with the lineage-proof frame.
    pub(super) fn from_prefix_with_budget(
        bytes: &[u8],
        budget: Option<SealDecodeBudget>,
    ) -> Result<(Self, &[u8]), SealDecodeError> {
        if bytes.len() < SEAL_WIRE_MIN_LEN {
            return Err(SealDecodeError::UnexpectedLength {
                expected: SEAL_WIRE_MIN_LEN,
                found: bytes.len(),
            });
        }
        let version = bytes[0];
        if version != SEAL_WIRE_V1 && version != SEAL_WIRE_V2 {
            return Err(SealDecodeError::UnknownVersion(version));
        }
        let ((generation, dot), rest) =
            read_address_parts(&bytes[1..]).ok_or(SealDecodeError::UnexpectedLength {
                expected: SEAL_WIRE_MIN_LEN,
                found: bytes.len(),
            })?;
        let declaration = EpochAddress::try_from_parts(generation, dot)?;
        let (candidates, rest) = read_candidate_set(bytes, rest, budget)?;
        let (protocol, rest) = read_ledger_set(bytes, rest, budget)?;
        let (sealed_join, tail) = decode_cut(rest)?;
        let (admission, tail) = if version == SEAL_WIRE_V2 {
            let (admission, tail) = read_admission(bytes, tail, declaration, budget)?;
            (Some(admission), tail)
        } else {
            (None, tail)
        };
        Ok((
            Self {
                declaration,
                candidates,
                protocol,
                sealed_join,
                admission,
            },
            tail,
        ))
    }

    const fn reject_tail(bytes: &[u8], tail: &[u8]) -> Result<(), SealDecodeError> {
        if tail.is_empty() {
            Ok(())
        } else {
            Err(SealDecodeError::UnexpectedLength {
                expected: bytes.len() - tail.len(),
                found: bytes.len(),
            })
        }
    }
}