minerva 0.2.0

use super::{Dot, DotSet, DotStore, Dotted, Received};

mod condense;

/// A station's composer over one causal pair: the [causal-store](Dotted) write
/// flow with its two disciplines made *structural* rather than remembered.
///
/// [`Dotted<S>`] is the kernel object, and its write flow is two *sequencing*
/// obligations a hand-roller can forget:
///
/// 1. *Merge before the next dot.* Assignment reads the never-forgetting
///    context, so a delta must be merged locally before the next dot is
///    assigned, or the same dot is re-issued and every peer whose context
///    already covers it silently drops the second write: the *doomed dot*
///    trap.
/// 2. *Cover the store.* A write that supersedes builds its delta through
///    [`try_new`](Dotted::try_new), whose context must contain every store
///    dot.
///
/// [`compose`](Self::compose) closes both in one call: it assigns the fresh
/// dot, assembles the delta context as *(superseded) union (the store's own
/// dots)* so coverage holds by construction, merges into the held state, and
/// returns the delta for shipping. Since the held state moves on every write,
/// the next call already carries the last dot. Freshness is a property of the
/// flow rather than a rule the caller keeps.
///
/// # What this is not
///
/// No transport, no persistence, no policy. The facade never chooses what to
/// write or what to supersede, so add-wins versus remove-wins stays exactly
/// which dots the caller passes to [`compose`](Self::compose) and
/// [`retract`](Self::retract). Persisting a delta before shipping it is the
/// caller's discipline, with [`adopt`](Self::adopt) as the resume half. This
/// is the causal-pair analogue of [`Producer`](super::Producer).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Composer<S> {
    /// The station this composer writes as: every fresh dot is minted here.
    station: u32,
    /// The held causal pair, updated by every local write so the next dot
    /// assignment reads a context that already carries the last one.
    state: Dotted<S>,
}

impl<S: DotStore + Clone> Composer<S> {
    /// A fresh composer for `station` over the bottom pair (nothing held,
    /// nothing seen).
    #[must_use]
    pub fn new(station: u32) -> Self {
        Self {
            station,
            state: Dotted::new(),
        }
    }

    /// Resumes a composer for `station` over persisted `state`.
    ///
    /// The caller's durability is the caller's: this is the resume half of the
    /// persist-then-send discipline (PRD 0015), and the `state` it adopts is
    /// whatever the caller durably kept. Assignment continues fresh off the
    /// adopted context.
    #[must_use]
    pub const fn adopt(station: u32, state: Dotted<S>) -> Self {
        Self { station, state }
    }

    /// The station this composer writes as.
    #[must_use]
    pub const fn station(&self) -> u32 {
        self.station
    }

    /// The held causal pair: this composer's current full state.
    #[must_use]
    pub const fn state(&self) -> &Dotted<S> {
        &self.state
    }

    /// The whole write flow in one step, returning the dot it minted beside the
    /// delta.
    ///
    /// Assigns the next fresh dot and hands it to `weave`, which returns the
    /// delta's store and the dots this write supersedes --- empty for a pure
    /// add, the observed old dots for a register-style update. `compose` then
    /// builds the covering context itself, merges the delta into the held
    /// state, and returns the minted dot beside the delta.
    ///
    /// # The returned dot is the receipt
    ///
    /// A consumer keying an out-of-band payload channel by this dot reads it
    /// *from the receipt*, never by re-deriving it through
    /// [`next_dot`](Dotted::next_dot) before the call: a merge landing
    /// between that re-derivation and `compose` keys the payload under a
    /// doomed dot. `compose` never inspects content, only the store's own
    /// `dots()`.
    ///
    /// # Run writes ride the same call
    ///
    /// A bulk write needs no second facade. Dots are per-station counters and
    /// this composer is its station's one writer, so every dot at or past the
    /// receipt is fresh by the context invariant: a `weave` closure recording
    /// a whole run from the receipt dot composes exactly like a single-dot
    /// write, and the one merge keeps the next assignment fresh past the
    /// whole run.
    ///
    /// # Why the refusal arm is unreachable
    ///
    /// The context is seeded from `superseded` and then every dot the store
    /// *reports* is inserted, so [`try_new`](Dotted::try_new)'s coverage
    /// check --- reading that same enumeration --- can find nothing
    /// uncovered. Coverage holds for any store, honest or not.
    ///
    /// A store that lies by under-reporting its support cannot be caught
    /// here, precisely because `try_new` trusts the enumeration `compose`
    /// trusted: the lie rides through, and the delta covers exactly the dots
    /// the store admits to. That is the honest outcome --- a store the crate
    /// cannot see into is one it cannot police, and the trait laws are the
    /// store author's contract. The `Err` arm is a total fallback the crate's
    /// no-`unwrap` rule requires, degrading to the pure-testimony delta over
    /// `superseded`; its unreachability is pinned by
    /// `test_compose_over_a_lying_store_rides_the_lie_through`.
    pub fn compose(&mut self, weave: impl FnOnce(Dot) -> (S, DotSet)) -> (Dot, Dotted<S>) {
        let dot = self.state.next_dot(self.station);
        let (store, superseded) = weave(dot);
        let delta = Self::assemble(store, superseded);
        self.state.merge_from(&delta);
        (dot, delta)
    }

    /// The register write, superseding a set computed from the *held* store
    /// inside the facade.
    ///
    /// A register-style update mints a fresh dot and displaces the values
    /// already held, so its supersede-set is a function of the current state. A
    /// caller that reads that set *before* [`compose`](Self::compose) opens a
    /// read-then-compose window: a merge landing in between changes what the
    /// store holds, and the pre-read supersede-set no longer matches (a TOCTOU
    /// on the very set that decides which old dots this write displaces).
    /// `compose_super` closes it by taking the supersede-set as a closure over
    /// the held store, computed at the moment of the write:
    /// [`DotFun::observed`](crate::metis::DotFun::observed) is the canonical
    /// argument for a bare register, and any read that names exactly the dots
    /// the write displaces is admissible. `weave` returns only the store; the
    /// covering context and the merge are the facade's, exactly as
    /// [`compose`](Self::compose)'s. Returns the minted dot beside the delta,
    /// the same receipt (see [`compose`](Self::compose)).
    pub fn compose_super(
        &mut self,
        weave: impl FnOnce(Dot) -> S,
        superseded: impl FnOnce(&S) -> DotSet,
    ) -> (Dot, Dotted<S>) {
        let dot = self.state.next_dot(self.station);
        let displaced = superseded(self.state.store());
        let store = weave(dot);
        let delta = Self::assemble(store, displaced);
        self.state.merge_from(&delta);
        (dot, delta)
    }

    /// Assembles a covered delta from a write's `store` and the dots it
    /// `superseded`, the shared body of [`compose`](Self::compose) and
    /// [`compose_super`](Self::compose_super).
    ///
    /// Covers by construction: seeds the context with everything the write
    /// supersedes, then folds in every dot the store reports, so
    /// [`try_new`](Dotted::try_new) finds nothing uncovered and the success path
    /// is total (the refusal arm is an unreachable documented fallback, pinned
    /// by `test_compose_over_a_lying_store_rides_the_lie_through`; a store that
    /// under-reports its support rides the lie through as the honest degraded
    /// outcome, because a store the crate cannot see into is a store it cannot
    /// police).
    fn assemble(store: S, superseded: DotSet) -> Dotted<S> {
        let mut context = superseded.clone();
        for held in store.dots() {
            let _ = context.insert(held);
        }
        Dotted::try_new(store, context).unwrap_or_else(|_| Dotted::from_context(superseded))
    }

    /// The observed-remove flow: a pure-testimony delta superseding exactly
    /// `superseded`, merged into the held state and returned for shipping.
    ///
    /// The delta is [`from_context`](Dotted::from_context) of `superseded`: a
    /// bottom store testifying to having seen those dots, so merging it (here
    /// and at every peer) supersedes exactly them. Which dots to pass is the
    /// caller's semantics: the dots observed under a key for an observed
    /// remove, or dots not yet observed for a remove-wins posture. `Composer`
    /// does not choose.
    pub fn retract(&mut self, superseded: DotSet) -> Dotted<S> {
        let delta = Dotted::from_context(superseded);
        self.state.merge_from(&delta);
        delta
    }

    /// Learns a peer's delta or full state (they are one type) `from` a named
    /// station: merges it into the held state and testifies to what it merged.
    ///
    /// The read side of the flow, the mirror of the deltas the write side
    /// ships. Because [`merge`](Dotted::merge) is a semilattice join, absorbing
    /// deltas in any order, grouping, and multiplicity converges, and a full
    /// state is just a delta a fresh peer has never seen.
    ///
    /// The returned [`Received`] is the receipt of that merge: it pairs `from`
    /// with the delta's own context, the dots the sender demonstrably held (a
    /// peer cannot ship what it has not seen). Feeding it to
    /// [`Purview::note`](crate::metis::Purview::note) raises the sender's row by
    /// exactly the context just received, so the over-claim and misattribution
    /// hazards the tracker used to guard by prose are unrepresentable: the
    /// evidence carries its own peer and its own proven context.
    pub fn absorb(&mut self, from: u32, delta: &Dotted<S>) -> Received {
        self.state.merge_from(delta);
        Received::witness(from, delta.context().clone())
    }

    /// What this composer owes a peer whose context is `peer_context`: the delta
    /// that converges the peer to this composer's state (delegates to
    /// [`delta_for`](Dotted::delta_for)).
    ///
    /// The store fragment is only what the peer lacks and the context carries
    /// every removal whole, so shipping this and merging it is exactly as good
    /// as shipping the whole state (PRD 0015 open question 3).
    #[must_use]
    pub fn owed_to(&self, peer_context: &DotSet) -> Dotted<S> {
        self.state.delta_for(peer_context)
    }

    /// [`owed_to`](Self::owed_to) under a recording-possession witness: the
    /// owed delta when the peer also states which recording state it holds
    /// (delegates to [`delta_for_witnessed`](Dotted::delta_for_witnessed),
    /// whose rustdoc owns the law and the witness's trust shape).
    ///
    /// The read a steady-state anti-entropy loop should run: against a peer
    /// that ships its recording have-set beside its context, the fragment is
    /// bounded by what is actually owed instead of carrying a
    /// recording-carrying store's plane whole per probe (the arc 10
    /// measurement: document-sized fragments at one owed keystroke, and at
    /// none).
    #[must_use]
    pub fn owed_to_witnessed(&self, peer_context: &DotSet, peer_recording: &DotSet) -> Dotted<S> {
        self.state.delta_for_witnessed(peer_context, peer_recording)
    }
}