minerva 0.2.0

//! The coalesced child plane (crdt-vision arc 11 phase four): the maintained
//! sibling index's cost shell.
//!
//! The contract form is the S116 child index, `BTreeMap<Anchor, Vec<Dot>>`:
//! per anchor, its children in the sibling order. On honest editing traffic
//! that map is almost entirely chain interiors, each element the sole child
//! of its own predecessor, so nearly every entry is a map node plus a
//! heap-allocated one-element vector restating what the run-coalesced
//! identity plane already carries: a stepped slot *means* "anchored `After`
//! its predecessor" by the chain law itself (`apply_chain_step`), and an
//! explicit slot's anchor is one column read away. So the plane is the
//! index for exactly those buckets, and this shell stores the rest.
//!
//! One representation invariant carries every read:
//!
//! * an anchor with an **explicit bucket** holds ALL of its children there,
//!   sorted in the sibling order (rank descending, dot ascending on ties);
//! * an anchor with **no explicit bucket** has either no child at all or
//!   exactly its *chain child*, the same-station successor dot anchored
//!   `After` it, derivable from the identity plane in constant time in
//!   the document ([`IdentityPlane::anchor_of`]: a station-map probe plus
//!   page arithmetic, never a rank walk-back).
//!
//! The invariant is maintained, never checked per read: an insert under an
//! anchor whose implicit chain child is incumbent materializes that child
//! into a fresh explicit bucket first (the plane's own
//! step-to-explicit conversion, one plane over); a removal that leaves a
//! bucket spelling only what the plane can spell drops the bucket. Like
//! the plane's stranded explicit entries, an explicit bucket a rebuild
//! would leave implicit is layout, not value (equality never reads this
//! coordinate), and the dual-form agreement property pins every read
//! against the contract form regardless of spelling history.
//!
//! Chains are `After`-shaped by the chain law, so `Origin` and `Before`
//! buckets are always explicit; and the chain-child probe takes the
//! successor through `checked_add`, so an anchor at the `u64::MAX` dot
//! ceiling simply has no derivable child (the S200 overflow lesson,
//! applied at design time).

extern crate alloc;

use alloc::collections::BTreeMap;
use alloc::vec::Vec;

use super::identity::IdentityPlane;
use super::placement::{Anchor, Dot};
use crate::metis::dot::RawDot;

/// The sibling order between two woven dots sharing an anchor: rank
/// DESCENDING (the newest write nearest its anchor, the RGA rule, PRD 0017
/// R4), ties broken by the dot ascending so the order is total. Reads the
/// plane, so both dots must be woven.
pub(super) fn sibling_cmp(plane: &IdentityPlane, a: Dot, b: Dot) -> core::cmp::Ordering {
    let (ra, rb) = (
        plane.get(&a).expect("left sibling has a locus").rank,
        plane.get(&b).expect("right sibling has a locus").rank,
    );
    rb.cmp(&ra).then_with(|| a.cmp(&b))
}

/// The derivable chain child of `anchor`: the same-station successor dot,
/// woven and anchored `After` its predecessor. `None` for `Origin` and
/// `Before` anchors (chains are `After`-shaped), at the `u64::MAX` dot
/// ceiling (a successor probe past it would overflow, so the refusal is
/// the honest verdict), and wherever the successor is absent or anchored
/// elsewhere. Constant in the document (a station-map probe plus page
/// arithmetic): the anchor read never materializes a rank.
fn chain_child(plane: &IdentityPlane, anchor: Anchor) -> Option<Dot> {
    let Anchor::After(RawDot {
        station,
        counter: index,
    }) = anchor
    else {
        return None;
    };
    // The successor of any coordinate is one-based, so the crossing back
    // into identity space never refuses; `ok()?` keeps the read total
    // rather than asserting it (ruling R-91).
    let successor = Dot::from_parts(station, index.checked_add(1)?).ok()?;
    (plane.anchor_of(&successor) == Some(anchor)).then_some(successor)
}

/// Whether `dot` is its own anchor's chain child, off the locus already in
/// hand (the enumeration passes read `(dot, locus)` pairs, so probing the
/// plane again would be redundant).
fn is_chain_child(dot: Dot, anchor: Anchor) -> bool {
    matches!(
        anchor,
        Anchor::After(RawDot { station, counter: index })
            if station == dot.station() && index.checked_add(1) == Some(dot.counter())
    )
}

/// One anchor's children in the sibling order: a view over either spelling.
/// Never empty (a childless anchor has no bucket at all).
#[derive(Clone, Copy, Debug)]
pub(super) enum Bucket<'a> {
    /// The stored exceptional bucket, sorted.
    Explicit(&'a [Dot]),
    /// The plane-derived chain child, the sole member.
    Implicit(Dot),
}

impl<'a> Bucket<'a> {
    /// The stored head: the highest-ranked child, the rank to beat.
    pub(super) const fn first(&self) -> Dot {
        match self {
            Self::Explicit(bucket) => bucket[0],
            Self::Implicit(dot) => *dot,
        }
    }

    /// The stored tail: the lowest-ranked child.
    pub(super) const fn last(&self) -> Dot {
        match self {
            Self::Explicit(bucket) => bucket[bucket.len() - 1],
            Self::Implicit(dot) => *dot,
        }
    }

    /// The number of children in the bucket (the vector-shift width a
    /// mutation of this bucket pays, the collation's charge unit).
    pub(super) const fn len(&self) -> usize {
        match self {
            Self::Explicit(bucket) => bucket.len(),
            Self::Implicit(_) => 1,
        }
    }

    /// The child at stored position `at`.
    pub(super) fn get(&self, at: usize) -> Option<Dot> {
        match self {
            Self::Explicit(bucket) => bucket.get(at).copied(),
            Self::Implicit(dot) => (at == 0).then_some(*dot),
        }
    }

    /// Every child in stored order.
    pub(super) fn iter(&self) -> BucketIter<'a> {
        match self {
            Self::Explicit(bucket) => BucketIter::Slice(bucket.iter()),
            Self::Implicit(dot) => BucketIter::One(Some(*dot)),
        }
    }

    /// The children strictly after stored position `at`.
    pub(super) fn suffix(&self, at: usize) -> BucketIter<'a> {
        match self {
            Self::Explicit(bucket) => BucketIter::Slice(bucket[at + 1..].iter()),
            Self::Implicit(_) => BucketIter::One(None),
        }
    }

    /// The children strictly before stored position `at`.
    pub(super) fn prefix(&self, at: usize) -> BucketIter<'a> {
        match self {
            Self::Explicit(bucket) => BucketIter::Slice(bucket[..at].iter()),
            Self::Implicit(_) => BucketIter::One(None),
        }
    }
}

/// A [`Bucket`] iterator over either spelling, yielding dots by value in
/// stored order (double-ended, so the walk's read-order reversals cost
/// nothing new).
#[derive(Clone, Debug)]
pub(super) enum BucketIter<'a> {
    /// An explicit bucket's slice.
    Slice(core::slice::Iter<'a, Dot>),
    /// Zero or one derived child.
    One(Option<Dot>),
}

impl BucketIter<'_> {
    /// The empty iteration (a childless anchor).
    pub(super) const fn empty() -> Self {
        Self::One(None)
    }
}

impl Iterator for BucketIter<'_> {
    type Item = Dot;

    fn next(&mut self) -> Option<Dot> {
        match self {
            Self::Slice(iter) => iter.next().copied(),
            Self::One(dot) => dot.take(),
        }
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        let len = match self {
            Self::Slice(iter) => iter.len(),
            Self::One(dot) => usize::from(dot.is_some()),
        };
        (len, Some(len))
    }
}

impl DoubleEndedIterator for BucketIter<'_> {
    fn next_back(&mut self) -> Option<Dot> {
        match self {
            Self::Slice(iter) => iter.next_back().copied(),
            Self::One(dot) => dot.take(),
        }
    }
}

impl ExactSizeIterator for BucketIter<'_> {}
impl core::iter::FusedIterator for BucketIter<'_> {}

/// A removal's consequence for its bucket's stored tail (the
/// region-endpoint replacement read `condense` consumes: a parent's
/// last-child edge in the order thread's dynamic forests must move exactly
/// when the stored tail does).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum TailChange {
    /// The removed dot was not the stored tail; the endpoint edge stands.
    Unchanged,
    /// The removed dot was the stored tail; the new tail, `None` when the
    /// bucket emptied.
    Replaced(Option<Dot>),
}

/// The coalesced child plane: the exceptional sibling buckets, everything
/// else derived from the identity plane per read.
#[derive(Clone, Debug, Default)]
pub(super) struct ChildPlane {
    /// Every anchor whose children the plane cannot spell alone: more than
    /// one child, a lone child that is not the chain child, or `Origin` /
    /// `Before` anchors (never derivable). Holds ALL of its anchor's
    /// children, sorted; never stored empty.
    explicit: BTreeMap<Anchor, Vec<Dot>>,
}

impl ChildPlane {
    /// The empty plane.
    pub(super) const fn new() -> Self {
        Self {
            explicit: BTreeMap::new(),
        }
    }

    /// The sibling bucket of `anchor`, `None` when it has no child.
    pub(super) fn bucket<'a>(
        &'a self,
        plane: &IdentityPlane,
        anchor: Anchor,
    ) -> Option<Bucket<'a>> {
        if let Some(bucket) = self.explicit.get(&anchor) {
            debug_assert!(
                !bucket.is_empty(),
                "an explicit bucket is never stored empty"
            );
            return Some(Bucket::Explicit(bucket));
        }
        chain_child(plane, anchor).map(Bucket::Implicit)
    }

    /// The children of `anchor` in stored order; empty when it has none.
    pub(super) fn iter<'a>(&'a self, plane: &IdentityPlane, anchor: Anchor) -> BucketIter<'a> {
        self.bucket(plane, anchor)
            .map_or(BucketIter::empty(), |bucket| bucket.iter())
    }

    /// Records a freshly woven `dot` under `anchor` at its one sorted
    /// position. Runs AFTER the plane insert: the sibling comparison reads
    /// the just-inserted locus, and the implicit spelling *is* the plane's
    /// own entry, so a chain child arriving alone stores nothing here. A
    /// sibling arriving beside an incumbent implicit child materializes
    /// the incumbent first (the plane's step-to-explicit conversion, one
    /// coordinate over).
    pub(super) fn insert(&mut self, plane: &IdentityPlane, anchor: Anchor, dot: Dot) {
        if let Some(bucket) = self.explicit.get_mut(&anchor) {
            let pos = bucket.partition_point(|&other| sibling_cmp(plane, other, dot).is_lt());
            bucket.insert(pos, dot);
            return;
        }
        match chain_child(plane, anchor) {
            Some(candidate) if candidate == dot => {}
            Some(candidate) => {
                let mut bucket = alloc::vec![candidate];
                let pos = bucket.partition_point(|&other| sibling_cmp(plane, other, dot).is_lt());
                bucket.insert(pos, dot);
                let _ = self.explicit.insert(anchor, bucket);
            }
            None => {
                let _ = self.explicit.insert(anchor, alloc::vec![dot]);
            }
        }
    }

    /// Removes an excised `dot` from `anchor`'s bucket, reporting what
    /// happened to the stored tail. Runs AFTER the plane removal: an
    /// implicit child's departure is its plane entry's own, so there is
    /// nothing stored to drop. A bucket the plane can spell alone again
    /// (its sole survivor is the chain child) de-materializes, so churn
    /// does not strand singleton buckets the way it cannot help stranding
    /// wider ones.
    pub(super) fn remove(&mut self, plane: &IdentityPlane, anchor: Anchor, dot: Dot) -> TailChange {
        let Some(bucket) = self.explicit.get_mut(&anchor) else {
            // No explicit bucket: the dot was the implicit chain child,
            // whole bucket and plane entry both, so its removal emptied
            // the bucket by construction.
            debug_assert_eq!(
                chain_child(plane, anchor),
                None,
                "an implicit bucket holds exactly the removed chain child"
            );
            return TailChange::Replaced(None);
        };
        let was_last = bucket.last() == Some(&dot);
        if let Some(pos) = bucket.iter().position(|&child| child == dot) {
            let _ = bucket.remove(pos);
        }
        let new_last = bucket.last().copied();
        if bucket.len() <= 1 && new_last == chain_child(plane, anchor) {
            let _ = self.explicit.remove(&anchor);
        }
        if was_last {
            TailChange::Replaced(new_last)
        } else {
            TailChange::Unchanged
        }
    }

    /// Materializes the child plane from a full identity plane. Two
    /// passes: the first groups every child that is not its own anchor's
    /// chain child; the second pulls the chain child into any bucket that
    /// went explicit anyway (the invariant: an explicit bucket holds ALL
    /// of its anchor's children), then sorts. `O(n)` over the plane plus
    /// the sort on the exceptional buckets only.
    pub(super) fn build(plane: &IdentityPlane) -> Self {
        let mut explicit: BTreeMap<Anchor, Vec<Dot>> = BTreeMap::new();
        for (dot, locus) in plane.iter() {
            if !is_chain_child(dot, locus.anchor) {
                explicit.entry(locus.anchor).or_default().push(dot);
            }
        }
        for (&anchor, bucket) in &mut explicit {
            if let Some(candidate) = chain_child(plane, anchor) {
                bucket.push(candidate);
            }
            bucket.sort_unstable_by(|a, b| sibling_cmp(plane, *a, *b));
        }
        Self { explicit }
    }

    /// Every anchor with children, paired with its LAST stored child (the
    /// region-endpoint registration read), in no particular order: the
    /// explicit buckets off the map, the implicit ones derived by one
    /// plane pass.
    pub(super) fn last_children<'a>(
        &'a self,
        plane: &'a IdentityPlane,
    ) -> impl Iterator<Item = (Anchor, Dot)> + 'a {
        let explicit = self.explicit.iter().map(|(&anchor, bucket)| {
            let last = *bucket
                .last()
                .expect("an explicit bucket is never stored empty");
            (anchor, last)
        });
        let implicit = plane.iter().filter_map(move |(dot, locus)| {
            (is_chain_child(dot, locus.anchor) && !self.explicit.contains_key(&locus.anchor))
                .then_some((locus.anchor, dot))
        });
        explicit.chain(implicit)
    }

    /// Exceptional buckets held (the coalescing accounting read: on honest
    /// chained traffic this is the run-boundary count, not the element
    /// count).
    #[cfg(test)]
    pub(super) fn explicit_buckets(&self) -> usize {
        self.explicit.len()
    }

    /// Children stored explicitly (every other woven element rides as its
    /// own plane entry).
    #[cfg(test)]
    pub(super) fn explicit_children(&self) -> usize {
        self.explicit.values().map(Vec::len).sum()
    }
}

#[cfg(test)]
mod tests;