minerva 0.2.0

extern crate alloc;

use alloc::collections::{BTreeMap, BTreeSet};
use core::num::NonZeroU64;

use super::{Cut, Dot, VersionVector};

mod algebra;
mod reads;
mod station;
mod wire;

pub(in crate::metis) use wire::{HAVE_SET_STATION_LEN, HAVE_SET_WIRE_HEADER_LEN};
pub use wire::{HaveSetDecodeBudget, HaveSetDecodeError};

use station::StationDots;

/// The exact have-set: per station, exactly which *dots* (1-based event indexes)
/// have been received, however disordered the arrival.
///
/// A [`VersionVector`] read as a cut can only say "everything below here",
/// while a disordered receipt stream produces an arbitrary subset of the dot
/// space. `DotSet` represents that subset exactly. It exposes the two cuts
/// that bracket it. [`floor`](Self::floor) is the greatest gap-free prefix,
/// and a genuine cut by construction. [`high_water`](Self::high_water) is the
/// per-station maximum, and a liveness horizon. [`holes`](Self::holes) names
/// the difference: the dots a repair protocol must fetch (PRD 0012).
///
/// # Why the floor exists
///
/// A `Stability` report is a cut claim the tracker cannot verify, and the
/// natural *wrong* report is the high-water a replica already gossips: over
/// holes it is an upper bound of a cut, and a meet over upper bounds
/// over-claims stability in the unrecoverable direction. `floor()` removes
/// the trap by construction. It cannot over-claim, whatever the arrival
/// order. In-order receipt does not need this type; it exists for the receipt
/// shape that breaks bare counters.
///
/// # Dots are 1-based; their meaning is the caller's
///
/// A dot is a [`Dot`]: a station and its `k`-th event with `k >= 1`. The
/// type carries that law, so no door re-checks it (ruling R-91); a 0-based
/// consumer maps `dot = sequence + 1`.
///
/// The caller defines what a dot means: received, applied, or persisted. This
/// is the standing vector-meaning hazard. A
/// [`Stability`](crate::metis::Stability) report must be the floor of an
/// *application-delivery* have-set, never a receipt-level one.
///
/// # Totality
///
/// Every operation is total and panic-free, counters saturating at the
/// [`u64`] ceiling. Insertion is idempotent and order-independent, and the
/// representation is canonical, so any insertion history of the same set
/// yields structurally equal values and `Eq`/`Hash` are honest. `merge` is
/// set union, the join of the have-set lattice, which keeps the type pure
/// fold-and-read material with no seriality hazard.
///
/// # The exceptions set is unbounded
///
/// In-order arrival keeps the out-of-order exceptions empty, so the structure
/// costs what a bare counter costs until a source is mid-reorder. The
/// exceptions then grow with every unfilled hole span, without bound. Bounding
/// them is caller policy, watched through
/// [`exceptions_len`](Self::exceptions_len).
#[derive(Clone, Debug, Default, PartialEq, Eq, Hash)]
pub struct DotSet {
    /// Per station, the received dots in compact form; a station with no dots
    /// is absent (canonical form). Visible inside the dot-set module tree so
    /// the codec, reads, and algebra folds use one representation.
    pub(in crate::metis::dot_set) stations: BTreeMap<u32, StationDots>,
}

/// Allocation-free summary of one nonempty station row in a [`DotSet`].
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub struct DotSetStation {
    station: u32,
    floor: u64,
    high_water: u64,
}

impl DotSetStation {
    /// Station identity for this row.
    #[must_use]
    pub const fn station(self) -> u32 {
        self.station
    }

    /// Greatest gap-free dot held for this station.
    #[must_use]
    pub const fn floor(self) -> u64 {
        self.floor
    }

    /// Greatest dot held for this station, including sparse receipt.
    #[must_use]
    pub const fn high_water(self) -> u64 {
        self.high_water
    }

    /// Span above the gap-free floor and through the high water.
    #[must_use]
    pub const fn forward_gap(self) -> u64 {
        self.high_water.saturating_sub(self.floor)
    }
}

impl DotSet {
    /// Creates an empty have-set: nothing received from anyone, both bracketing
    /// cuts at bottom.
    #[must_use]
    pub const fn new() -> Self {
        Self {
            stations: BTreeMap::new(),
        }
    }

    /// Embeds a *witnessed* cut as its gap-free have-set: every dot at or below
    /// the cut is held, nothing above, no holes.
    ///
    /// The public witnessed embedding, and the only public way to assert
    /// gap-freedom into a have-set. The vector inside a [`Cut`] is gap-free by
    /// construction, because [`Cut`]'s constructor set is closed over honest
    /// sources. Embedding it therefore cannot over-claim. The bare-vector
    /// `from_cut` (`pub(crate)`) leaves a hole open, and this door closes it:
    /// `from_cut` takes the caller's word that the vector *means* a cut, so an
    /// upper-bound-over-holes walks straight in. Round-trips exactly:
    /// `from_witnessed(&c).floor() == *c.as_vector()`.
    ///
    /// The bare-vector embedding is `pub(crate)`, so no outside caller can
    /// launder an unwitnessed vector into a gap-free have-set: the door is
    /// this `Cut`-taking form alone.
    ///
    /// ```compile_fail
    /// use minerva::metis::{DotSet, VersionVector};
    /// let v = VersionVector::new();
    /// // `from_cut` is `pub(crate)`: the bare-vector embedding is unreachable
    /// // from outside, which is exactly the laundering path this closes.
    /// let laundered = DotSet::from_cut(&v);
    /// ```
    #[must_use]
    pub fn from_witnessed(cut: &Cut) -> Self {
        Self::from_cut(cut.as_vector())
    }

    /// Embeds a genuine cut as its gap-free have-set: every dot at or below the
    /// cut is held, nothing above, no holes.
    ///
    /// The lossless direction of the pair (`from_cut(&v).floor() == v`); the
    /// projection back is [`floor`](Self::floor). The caller asserts the
    /// vector's *meaning* is a cut (a delivered prefix), not a high-water that
    /// may cover holes; embedding an upper bound here is exactly the over-claim
    /// this type exists to prevent, which is why this form is `pub(crate)`: the
    /// public door is [`from_witnessed`](Self::from_witnessed), whose [`Cut`]
    /// argument carries the gap-freedom proof. This bare-vector form stays
    /// in-crate for the closure theorems that legitimately hold an
    /// already-witnessed vector ([`Cut::to_have_set`], the lattice property
    /// tests) and never launders an unwitnessed vector across the crate
    /// boundary.
    #[must_use]
    pub(crate) fn from_cut(cut: &VersionVector) -> Self {
        Self {
            stations: cut
                .iter()
                .map(|(station, count)| {
                    (
                        station,
                        StationDots {
                            floor: count,
                            above: BTreeSet::new(),
                        },
                    )
                })
                .collect(),
        }
    }

    /// Records receipt of a dot, returning whether it was newly recorded
    /// (`false` for a duplicate).
    ///
    /// The refusal is duplicate-only: the non-dot `0` is unrepresentable in
    /// [`Dot`], which carries the one-based law on the type (ruling R-91).
    ///
    /// Idempotent and order-independent: any insertion history of the same dots
    /// yields an equal `DotSet`. A dot at `floor + 1` advances the floor and
    /// absorbs any exception run it makes contiguous; any other dot above the
    /// floor parks in the exceptions until the gap below it fills.
    ///
    /// The returned `bool` is safely ignorable: the have-set is grow-only
    /// and idempotent, so a `false` is a harmless duplicate receipt, not a
    /// refused write (contrast [`DotFun::insert`](crate::metis::DotFun::insert)).
    pub fn insert(&mut self, dot: Dot) -> bool {
        let counter = dot.counter();
        let slot = self.stations.entry(dot.station()).or_default();
        if slot.contains(counter) {
            return false;
        }
        if counter == slot.floor.saturating_add(1) {
            slot.floor = counter;
            slot.absorb();
        } else {
            let _ = slot.above.insert(counter);
        }
        true
    }

    /// Records receipt of a contiguous dot run `first .. first + len` in one
    /// write: the bulk-ingest form of [`insert`](Self::insert), agreeing
    /// with it exactly (inserting the same dots one at a time yields an
    /// equal set; duplicates are absorbed).
    ///
    /// `first` is a [`NonZeroU64`] because a run of dots starts at a dot: the
    /// old `first.max(1)` clamp is the type now (ruling R-91). The station is
    /// a separate parameter because a run lives inside one station's counters.
    ///
    /// Cost is the compact form's own: a run touching `floor + 1` extends
    /// the floor in one step and absorbs any exceptions it reaches (the
    /// in-order ingest shape, `O(log + absorbed)`), and a run parked above
    /// a gap enters the exceptions as a range extension. A run whose end
    /// would pass the `u64::MAX` dot ceiling is truncated at the ceiling
    /// (the per-dot form can name no dot past it either).
    pub(crate) fn insert_run(&mut self, station: u32, first: NonZeroU64, len: u32) {
        if len == 0 {
            return;
        }
        let start = first.get();
        // The run's last dot, inclusive; saturation truncates at the ceiling.
        // `start >= 1` and `len >= 1`, so the range is never inverted.
        let last = start.saturating_add(u64::from(len) - 1);
        let slot = self.stations.entry(station).or_default();
        if start <= slot.floor.saturating_add(1) {
            if last > slot.floor {
                slot.floor = last;
                // Exceptions the raised floor swallowed leave `above`; the
                // canonical form keeps every exception strictly above it.
                // At the dot ceiling the checked add is load-bearing: a
                // saturating split point would RETAIN an exception at
                // `u64::MAX` beside a `u64::MAX` floor, a duplicate
                // representation the per-dot form absorbs (the S212 gate
                // review's finding; reachable through a witnessed-cut
                // seeded floor, not through inserts alone).
                match slot.floor.checked_add(1) {
                    Some(next) => slot.above = slot.above.split_off(&next),
                    None => slot.above.clear(),
                }
                slot.absorb();
            }
        } else {
            slot.above.extend(start..=last);
        }
    }

    /// Removes one dot, returning whether it was held; the *store-role* write.
    ///
    /// As a receipt-side have-set the type is grow-only by charter (PRD 0012),
    /// which is why this is `pub(crate)`: it exists for the presence-*store*
    /// role alone (the [`DotStore`](crate::metis::DotStore) instance), whose
    /// survivor law shrinks a store in place
    /// (`causal_merge_from`, S182). Total and canonical-form-preserving: a
    /// mid-prefix removal retracts that station's floor and parks the rest of
    /// the old prefix as exceptions (`O(floor - dot)` once, the compact form's
    /// honest price), and a station emptied by removal drops its row.
    pub(crate) fn remove(&mut self, dot: Dot) -> bool {
        let station = dot.station();
        let Some(slot) = self.stations.get_mut(&station) else {
            return false;
        };
        let removed = slot.remove(dot.counter());
        if slot.floor == 0 && slot.above.is_empty() {
            let _ = self.stations.remove(&station);
        }
        removed
    }

    /// Whether the dot has been received. Total.
    #[must_use]
    pub fn contains(&self, dot: Dot) -> bool {
        self.stations
            .get(&dot.station())
            .is_some_and(|slot| slot.contains(dot.counter()))
    }

    /// The *floor*: the greatest genuine cut contained in the have-set, per
    /// station the longest gap-free prefix.
    ///
    /// Order-theoretically the have-set's *interior* (the greatest down-set it
    /// contains). Gap-free by construction, so it is the honest
    /// [`Stability::report`](crate::metis::Stability::report) input (PRD 0011
    /// R8's cut claim) and can never over-claim, whatever the arrival order.
    /// Monotone non-decreasing under [`insert`](Self::insert) and
    /// [`merge`](Self::merge), but only *laxly* compatible with union:
    /// `a.merge(&b).floor()` dominates `a.floor().merge(&b.floor())` and is
    /// strictly greater when the two sides fill each other's holes, which is
    /// why exchanging have-sets beats exchanging floors.
    #[must_use]
    pub fn floor(&self) -> VersionVector {
        let mut cut = VersionVector::new();
        for (&station, slot) in &self.stations {
            cut.observe(station, slot.floor);
        }
        cut
    }

    /// The *high water*: per station, the greatest received dot.
    ///
    /// This is the cover of the have-set's down-closure, the least down-set
    /// containing it. It is a liveness quantity. Use it for "who is ahead"
    /// anti-entropy reads, where over-claiming costs only a phantom pull.
    /// Never use it as a stability report: over holes it is an upper bound of
    /// a cut, and meeting upper bounds over-claims, which is the unsafe
    /// direction. Unlike the floor it is a join homomorphism:
    /// `a.merge(&b).high_water() == a.high_water().merge(&b.high_water())`.
    /// Always dominates [`floor`](Self::floor); the gap between the two is
    /// this replica's reorder window.
    #[must_use]
    pub fn high_water(&self) -> VersionVector {
        let mut bound = VersionVector::new();
        for (&station, slot) in &self.stations {
            bound.observe(station, slot.high_water());
        }
        bound
    }
}