minerva 0.2.0

extern crate alloc;

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

use crate::metis::{Dot, DotSet};

use super::super::d;
use super::super::editor::{Replica, gossip_text};
use super::probes::framing_probe;
use super::readings::Readings;

/// Runs the measured session and reads every quantity. Scaled to stay
/// debug-build fast; the scale is recorded in the report. The shape (type,
/// delete, steady gossip) is the 10k-char session the slice describes.
pub(super) fn run_session(typed_target: usize, delete_target: usize) -> Readings {
    // Whole-state anti-entropy ships each station's dots contiguously, so the
    // steady-state reorder window stays empty. The synthetic context below pins
    // that the reads still report a real partial-transport gap.
    const GOSSIP_PERIOD: usize = 7;

    let roster = [1u32, 2u32, 3u32];
    let mut replicas: Vec<Replica> = roster.iter().map(|&s| Replica::new(s, &roster)).collect();

    // Each replica keeps its own cursor and types a growing single-author chain;
    // periodic directed gossip interleaves peer dots with local ones.
    let mut anchors: Vec<Option<Dot>> = alloc::vec![None; replicas.len()];
    let mut typed_dots: Vec<Dot> = Vec::new();
    let mut typed = 0usize;
    let mut round = 0usize;
    while typed < typed_target {
        for idx in 0..replicas.len() {
            if typed >= typed_target {
                break;
            }
            let ch = char::from(b'a' + u8::try_from(round % 26).unwrap_or(0));
            let dot = replicas[idx].insert(ch, anchors[idx]);
            anchors[idx] = Some(dot);
            typed_dots.push(dot);
            typed += 1;
            if typed.is_multiple_of(GOSSIP_PERIOD) {
                let next = (idx + 1) % replicas.len();
                let (from, into) = split_two(&mut replicas, idx, next);
                gossip_text(from, into);
            }
        }
        round += 1;
    }

    let exceptions: Vec<usize> = replicas
        .iter()
        .map(|r| r.text_context().exceptions_len())
        .collect();
    let holes: Vec<u64> = replicas
        .iter()
        .map(|r| r.text_context().hole_count())
        .collect();
    let full_frame_bytes = replicas[0].text_context().to_bytes().len();

    let (demo_exceptions, demo_holes) = {
        let mut ctx = DotSet::new();
        let _ = ctx.insert(d(9, 1));
        let _ = ctx.insert(d(9, 5));
        let _ = ctx.insert(d(9, 6));
        (ctx.exceptions_len(), ctx.hole_count())
    };

    // Delete a spread of typed dots at their author replicas. The stride is
    // forced coprime to the roster size so the churn spans every station.
    let station_index: BTreeMap<u32, usize> =
        roster.iter().enumerate().map(|(i, &s)| (s, i)).collect();
    let mut deleted_per_station: BTreeMap<u32, usize> = BTreeMap::new();
    let mut deleted = 0usize;
    if !typed_dots.is_empty() {
        let mut stride = (typed_dots.len() / delete_target.max(1)).max(1);
        if stride.is_multiple_of(roster.len()) {
            stride += 1;
        }
        let mut i = 0usize;
        while i < typed_dots.len() && deleted < delete_target {
            let dot = typed_dots[i];
            if let Some(&idx) = station_index.get(&dot.station()) {
                replicas[idx].delete(dot);
                *deleted_per_station.entry(dot.station()).or_insert(0) += 1;
                deleted += 1;
            }
            i += stride;
        }
    }
    assert_eq!(
        deleted_per_station.len(),
        roster.len(),
        "the delete churn spans every replica, not just one",
    );

    converge(&mut replicas);
    let visible_len = replicas[0].visible_len();
    let skeleton_len = replicas[0].skeleton_len();
    for replica in &replicas {
        assert_eq!(
            replica.visible_len(),
            visible_len,
            "all replicas agree on the visible size after convergence",
        );
    }

    let framing = framing_probe();

    Readings {
        visible_len,
        skeleton_len,
        per_keystroke_dots: framing.per_keystroke_dots,
        per_keystroke_bytes: framing.per_keystroke_bytes,
        per_keystroke_context_bytes: framing.per_keystroke_context_bytes,
        per_keystroke_floored_context_bytes: framing.per_keystroke_floored_context_bytes,
        batch_dots: framing.batch_dots,
        batch_bytes: framing.batch_bytes,
        exceptions,
        holes,
        demo_exceptions,
        demo_holes,
        full_frame_bytes,
        typed,
        deleted,
    }
}

/// Mutable access to two distinct replicas for a gossip step.
fn split_two(replicas: &mut [Replica], from: usize, into: usize) -> (&Replica, &mut Replica) {
    assert_ne!(from, into);
    if from < into {
        let (left, right) = replicas.split_at_mut(into);
        (&left[from], &mut right[0])
    } else {
        let (left, right) = replicas.split_at_mut(from);
        (&right[0], &mut left[into])
    }
}

/// Gossips every directed pair until the replicas' visible sizes agree. The loop
/// is bounded by the roster size and exits early on agreement.
fn converge(replicas: &mut [Replica]) {
    for _ in 0..replicas.len() {
        for i in 0..replicas.len() {
            for j in 0..replicas.len() {
                if i != j {
                    let (from, into) = split_two(replicas, i, j);
                    gossip_text(from, into);
                }
            }
        }
        let first = replicas[0].visible_len();
        if replicas.iter().all(|r| r.visible_len() == first) {
            break;
        }
    }
}