minerva 0.2.0

//! Local data-plane writes, hygiene, and delta construction.

extern crate alloc;

use alloc::boxed::Box;
use alloc::vec;
use alloc::vec::Vec;

use crate::kairos::{Clock, Kairos, TickCounter};
use crate::metis::{Anchor, Dot, DotSet, Dotted, Locus, Metatheses, Metathesis, Retired, Rhapsody};

use super::super::{Note, OldDelta};
use super::{Replica, Text, fold_native};

impl Replica {
    /// Mints this station's next old-plane identity. The counter is
    /// pre-incremented from a count, so it is one-based by construction
    /// and never the non-dot zero (ruling R-91).
    fn next_dot(&mut self) -> Dot {
        self.counter += 1;
        Dot::from_parts(self.id, self.counter).expect("a pre-incremented counter is nonzero")
    }
    // ---- local edits -----------------------------------------------------

    /// Inserts one element at visual position `at` (clamped) of the plane
    /// this replica currently edits. Returns the minted dot.
    pub fn insert_visible(&mut self, at: usize, out: &mut Vec<Note>) -> Dot {
        if self.transition.is_some() {
            return self.insert_native(at, out);
        }
        let dot = self.next_dot();
        let (delta, _) = weave_delta(self.text.store(), &self.clock, dot, at);
        self.commit_old(vec![dot], OldDelta::Text(Box::new(delta)), out);
        dot
    }

    /// Inserts at visual position `at` under the *writer discipline*, the
    /// coordination-free retirement license's writer half (the S224 arm,
    /// reading the excision brief's H1 verdict against the R-41
    /// counterexample): the anchor is chosen in the *oath plane*, a clone
    /// of the current plane condensed against this station's own published
    /// acknowledgements as a trusted witness, so a weave minted after the
    /// station acknowledged a removal never anchors the removed dot (nor
    /// anything the enacted acknowledgement sweeps with it). The
    /// acknowledgement thereby becomes a quiescence oath, the QSBR shape
    /// from safe memory reclamation: every anchoring weave minted BEFORE
    /// the oath rides inside the acknowledgement's carried floor (the cut
    /// gate delivers it everywhere before the acknowledgement counts), and
    /// every weave minted AFTER the oath is rebound here, so once the meet
    /// covers a dot no write can ever strand on its excision, and the
    /// cut-gated meet alone licenses condense on local evidence
    /// (`properties::a_disciplined_fleet_condenses_on_local_evidence_and_converges`).
    ///
    /// The binding set is `acked`, this station's own published oath,
    /// never the gathered meet: the meet lags at exactly the wrong moment
    /// (a peer can hold the full meet and excise while the gathering is
    /// still in flight toward this station), pinned as
    /// `boundary::writer::the_discipline_binds_on_own_acknowledgement_not_the_meet`.
    ///
    /// The rank is minted above the REAL bucket's head (tombstones
    /// included), so the weave reads adjacent to the rebound anchor in
    /// every skeleton, condensed or not. The rebinding is the discipline's
    /// price, paid whether or not anyone ever excises: leaving a retired
    /// tombstone's subtree forfeits the sibling rank contest a same-bucket
    /// weave would have run
    /// (`boundary::writer::the_discipline_prices_placement_not_convergence`).
    pub fn insert_disciplined(&mut self, at: usize, out: &mut Vec<Note>) -> Dot {
        assert!(
            self.transition.is_none(),
            "replica {}: the disciplined weave is an old-plane operation",
            self.id
        );
        let mut oath = self.text.store().clone();
        let _ = oath.condense(&Retired::trust(self.acked.clone()));
        let dot = self.next_dot();
        let (delta, _) = weave_delta_disciplined(self.text.store(), &oath, &self.clock, dot, at);
        self.commit_old(vec![dot], OldDelta::Text(Box::new(delta)), out);
        dot
    }

    /// Deletes the element at visual position `at` of the current plane,
    /// minting a born-dead marker dot beside the superseded target (the
    /// dotted-delete duty). Returns the deleted dot, or `None` when the
    /// plane is empty.
    pub fn delete_visible(&mut self, at: usize, out: &mut Vec<Note>) -> Option<Dot> {
        if self.transition.is_some() {
            return self.delete_native(at, out);
        }
        let order = self.text.store().order();
        if order.is_empty() {
            return None;
        }
        let target = order[at % order.len()];
        let marker = self.next_dot();
        let delta = removal_delta(marker, target);
        self.commit_old(vec![marker], OldDelta::Text(Box::new(delta)), out);
        Some(target)
    }

    /// Moves the element at visual position `from` to sit visually after
    /// the element at `to_after` (`None` for the document head), as a
    /// movement testimony in the current old plane. The local assert
    /// carries the recipe's deferral duty: an adopted replica mints no
    /// movement testimonies while its window is open, because the consignment
    /// re-spells the topology movement acceptance depends on and a
    /// window-time verdict can flip at the seal (the module note's duty
    /// list; movement intents queue caller-side and mint after the seal).
    /// A laggard lawfully moves old-plane while peers hold a window open,
    /// and the seal consignment carries the judged outcome across.
    pub fn move_visible(
        &mut self,
        from: usize,
        to_after: Option<usize>,
        out: &mut Vec<Note>,
    ) -> Option<Dot> {
        assert!(
            self.transition.is_none(),
            "replica {}: old-plane moves are a pre-adoption operation",
            self.id
        );
        let order = self.text.store().recension(self.moves.store()).order();
        if order.is_empty() {
            return None;
        }
        let target = order[from % order.len()];
        let after = to_after
            .map(|index| order[index % order.len()])
            .filter(|&dot| dot != target);
        // The anchor and the movement target are payload seats: they keep
        // the raw coordinate spelling this slice leaves untouched (R-91).
        let anchor = after.map_or(Anchor::Origin, |dot| Anchor::After(dot.into()));
        let rank = self.rank_above(anchor);
        let dot = self.next_dot();
        let delta = Dotted::from_store(Metatheses::singleton(
            dot,
            Metathesis {
                target: target.into(),
                to: Locus { anchor, rank },
            },
        ));
        self.commit_old(vec![dot], OldDelta::Moves(delta), out);
        Some(dot)
    }
    /// Excises retired order tombstones against this replica's honestly
    /// gathered witness (the generation's cut-gated retirement meet).
    /// Local hygiene, not protocol: nothing is broadcast, `order()` and
    /// the recension are invariant (the anchor pins guarantee the latter),
    /// and a peer that condensed at a different time converges again at
    /// the next equal-witness condense. Pre-window only, like every
    /// old-plane operation. Returns how many loci were excised.
    pub fn condense(&mut self) -> usize {
        assert!(
            self.transition.is_none(),
            "replica {}: condense is an old-plane operation",
            self.id
        );
        let retired = self.retirement.retired();
        let mut store = self.text.store().clone();
        let excised = store.condense(&retired);
        let context = self.text.context().clone();
        self.text = Text::try_new(store, context)
            .expect("condense trims only the skeleton, so the context still covers the store");
        excised
    }

    /// Retires movement testimonies whose target is *causally removed*
    /// here (its dot in the text context, its locus gone: excised by a
    /// condense, or a removal that outran the weave), as a *dotted*
    /// old-plane delta: a born-dead marker dot beside the superseded
    /// testimony dots, so the retirement is classifiable against every
    /// future epoch boundary and its arrival raises floors like any event
    /// (the dotted-delete duty applied to the named consumer's in-plane
    /// condense retract). A locally absent locus alone is NOT orphanhood:
    /// a testimony can outrun its target's weave through the reordering
    /// fabric, and retiring it would erase a valid in-flight move
    /// everywhere, convergent but lossy (the review-caught hazard,
    /// pinned as `boundary::retirement::an_outrun_testimony_is_not_an_orphan`); the
    /// context test is the removal evidence, because context covering an
    /// undelivered weave means the element arrives already superseded.
    /// Retiring the testimony
    /// releases its anchor pin, so the next acknowledgement round lets the
    /// anchor's tombstone excise: pin, retract, unpin, excise is the whole
    /// hygiene lifecycle. Returns the minted marker, or `None` when no
    /// testimony is orphaned.
    pub fn retire_orphans(&mut self, out: &mut Vec<Note>) -> Option<Dot> {
        assert!(
            self.transition.is_none(),
            "replica {}: orphan retirement is an old-plane operation",
            self.id
        );
        let mut orphaned = DotSet::new();
        for (dot, testimony) in self.moves.store() {
            // The payload-to-identity funnel (R-91): a raw target naming
            // the non-dot zero is placed nowhere and covered by nothing,
            // which is exactly what the two reads below used to answer.
            let Ok(target) = Dot::try_from(testimony.target) else {
                continue;
            };
            if self.text.store().locus(target).is_none() && self.text.context().contains(target) {
                let _ = orphaned.insert(dot);
            }
        }
        let _ = orphaned.dots().next()?;
        let marker = self.next_dot();
        let mut context = orphaned;
        let _ = context.insert(marker);
        self.commit_old(
            vec![marker],
            OldDelta::Moves(Dotted::from_context(context)),
            out,
        );
        Some(marker)
    }

    fn rank_above(&self, anchor: Anchor) -> Kairos {
        let top = {
            let store = self.text.store();
            store
                .children_of(anchor)
                .next()
                .and_then(|dot| store.locus(dot))
                .map(|locus| locus.rank)
        };
        if let Some(rank) = top {
            self.clock.observe(rank);
        }
        self.clock.now(0u16)
    }

    fn insert_native(&mut self, at: usize, out: &mut Vec<Note>) -> Dot {
        let transition = self.transition.as_mut().expect("routed by phase");
        transition.native_counter += 1;
        let dot = Dot::from_parts(self.id, transition.native_counter)
            .expect("a pre-incremented counter is nonzero");
        let (delta, stamp) = weave_delta(transition.native_text.store(), &self.clock, dot, at);
        let delta = OldDelta::Text(Box::new(delta));
        let dots = vec![dot];
        fold_native(transition, &dots, &delta, Some(stamp));
        let note = Note::Native {
            epoch: transition.adopted.address(),
            stamp,
            dots,
            delta,
        };
        self.emit_mint(out, note);
        self.poll(out);
        dot
    }

    fn delete_native(&mut self, at: usize, out: &mut Vec<Note>) -> Option<Dot> {
        let stamp = self.clock.now(0u16);
        let transition = self.transition.as_mut().expect("routed by phase");
        let order = transition.native_text.store().order();
        if order.is_empty() {
            return None;
        }
        let target = order[at % order.len()];
        transition.native_counter += 1;
        let marker = Dot::from_parts(self.id, transition.native_counter)
            .expect("a pre-incremented counter is nonzero");
        let delta = OldDelta::Text(Box::new(removal_delta(marker, target)));
        let dots = vec![marker];
        fold_native(transition, &dots, &delta, Some(stamp));
        let note = Note::Native {
            epoch: transition.adopted.address(),
            stamp,
            dots,
            delta,
        };
        self.emit_mint(out, note);
        self.poll(out);
        Some(target)
    }

    fn commit_old(&mut self, dots: Vec<Dot>, delta: OldDelta, out: &mut Vec<Note>) {
        for &dot in &dots {
            let _ = self.have.insert(dot);
            let _ = self.recorded.insert(dot);
        }
        self.fold_current(&delta);
        self.log.push((dots.clone(), delta.clone()));
        let note = Note::Old {
            generation: self.generation,
            dots,
            delta,
        };
        self.emit_mint(out, note);
        self.poll(out);
    }
}

/// Builds one weave delta at visual slot `at` (clamped) of `plane`,
/// following the studio's rank discipline: observe the slot's incumbent
/// top child so the new rank mints above it.
fn weave_delta(
    plane: &Rhapsody,
    clock: &Clock<TickCounter>,
    dot: Dot,
    at: usize,
) -> (Text, Kairos) {
    let order = plane.order();
    let after = if order.is_empty() {
        None
    } else {
        at.checked_sub(1).map(|index| order[index % order.len()])
    };
    let anchor = plane.anchor_for_visual_insert(after.map(Into::into));
    let top = plane
        .children_of(anchor)
        .next()
        .and_then(|child| plane.locus(child))
        .map(|locus| locus.rank);
    if let Some(rank) = top {
        clock.observe(rank);
    }
    let rank = clock.now(0u16);
    let mut delta = Rhapsody::new();
    assert!(delta.weave(dot, Locus { anchor, rank }));
    (Text::from_store(delta), rank)
}

/// The disciplined weave delta: the visual slot comes from the real
/// plane (`order()` is condense-invariant, so the oath plane reads the
/// same visible document), the ANCHOR from the oath plane (the real
/// plane with this station's published acknowledgements enacted, so a
/// sworn-off tombstone is never named and the descent lands exactly
/// where a fully excised peer's own descent would), and the RANK above
/// the real bucket's head (a retired tombstone still in this bucket may
/// outrank the oath bucket's head; minting above the real head keeps the
/// weave adjacent to its anchor in every skeleton, condensed or not).
fn weave_delta_disciplined(
    plane: &Rhapsody,
    oath: &Rhapsody,
    clock: &Clock<TickCounter>,
    dot: Dot,
    at: usize,
) -> (Text, Kairos) {
    let order = plane.order();
    let after = if order.is_empty() {
        None
    } else {
        at.checked_sub(1).map(|index| order[index % order.len()])
    };
    let anchor = oath.anchor_for_visual_insert(after.map(Into::into));
    let top = plane
        .children_of(anchor)
        .next()
        .and_then(|child| plane.locus(child))
        .map(|locus| locus.rank);
    if let Some(rank) = top {
        clock.observe(rank);
    }
    let rank = clock.now(0u16);
    let mut delta = Rhapsody::new();
    assert!(delta.weave(dot, Locus { anchor, rank }));
    (Text::from_store(delta), rank)
}

/// A dotted delete: a born-dead marker beside the superseded target, so
/// the removal is classifiable against every future epoch boundary.
fn removal_delta(marker: Dot, target: Dot) -> Text {
    let mut context = DotSet::new();
    let _ = context.insert(marker);
    let _ = context.insert(target);
    Dotted::from_context(context)
}