//! `fencing` — la frontera del clon (T11c, §D): lo que NINGÚN sistema
//! coordination-free puede cerrar.
//!
//! F2 (réplica-por-dispositivo, T11a) resuelve *dos dispositivos distintos*.
//! Queda el caso duro: **una celda con dos escritores vivos** — un dispositivo
//! restaurado a dos copias (VM snapshot, proceso duplicado) que comparten
//! `device_id` y celda y **no se comunican**. Cada clon puede gastar la celda
//! entera → doble gasto.
//!
//! **Teorema de imposibilidad (`BOUNDARY.md`).** Ningún mecanismo local detecta un
//! clon silencioso antes de que los clones se comuniquen. Es el mismo CAP que el
//! sobregiro particionado, ahora dentro de una celda.
//!
//! **Lo mejor posible — fencing por época, enforced en el merge:** cada arranque
//! de un dispositivo incrementa una **época** monótona **device-local** (evita
//! consenso). El gasto se indexa por `(device_id, epoch)`. Al fusionar, dos épocas
//! para un mismo `device_id` delatan un clon: la de época menor es obsoleta. No se
//! **previene** el doble gasto (imposible), se **acota** (a una celda por clon no
//! detectado) y se **detecta** al reconectar: de "silencioso e ilimitado" a
//! "acotado y delatado".

use std::collections::BTreeMap;

/// Identidad de un dispositivo (una celda de presupuesto).
pub type DeviceId = u64;

/// Resultado de fusionar dos vistas de una misma celda-dispositivo.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum MergeOutcome {
    /// Un único linaje: fusión limpia.
    Ok,
    /// Dos linajes con el mismo `device_id`: clon detectado. La época menor es el
    /// clon obsoleto; sus gastos post-bifurcación quedan aislados.
    CloneDetected {
        /// El dispositivo bifurcado.
        device_id: DeviceId,
        /// Época obsoleta (el clon).
        stale_epoch: u64,
        /// Época viva (el linaje que continúa).
        live_epoch: u64,
    },
}

/// El gasto de una celda-dispositivo, indexado por época (fencing).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FencedCell {
    /// El dispositivo dueño de la celda.
    pub device_id: DeviceId,
    /// Presupuesto de la celda (`b0`).
    pub budget: i64,
    /// Época viva actual (monótona, device-local, persistida al arrancar).
    pub epoch: u64,
    /// Gasto por época: `epoch -> spent`. Cada época solo crece (CRDT por época).
    pub spent: BTreeMap<u64, i64>,
}

impl FencedCell {
    /// Celda nueva en la época 0, sin gasto.
    pub fn new(device_id: DeviceId, budget: i64) -> Self {
        Self {
            device_id,
            budget,
            epoch: 0,
            spent: BTreeMap::new(),
        }
    }

    /// Arranque del dispositivo: incrementa y persiste la época (device-local).
    pub fn boot(&mut self) -> u64 {
        self.epoch += 1;
        self.epoch
    }

    /// Gasta `amount` (≥0) en la época viva actual.
    pub fn spend(&mut self, amount: i64) {
        debug_assert!(amount >= 0);
        *self.spent.entry(self.epoch).or_insert(0) += amount;
    }

    /// Épocas con gasto no nulo (los linajes presentes).
    fn linajes(&self) -> Vec<u64> {
        self.spent.iter().filter(|(_, &s)| s > 0).map(|(&e, _)| e).collect()
    }

    /// Gasto total reclamado por TODOS los linajes (incluye el doble gasto de un
    /// clon; puede exceder el presupuesto).
    pub fn total_claimed(&self) -> i64 {
        self.spent.values().sum()
    }

    /// Gasto del linaje **vivo** (la época mayor): lo que se propaga como legítimo
    /// hacia el resto de la malla. Aísla los linajes obsoletos.
    pub fn reconciled_spent(&self) -> i64 {
        self.spent.iter().next_back().map(|(_, &s)| s).unwrap_or(0)
    }

    /// Daño acotado: el sobregiro de la celda por el clon, `max(0, total − budget)`.
    /// Por el fencing está acotado a **una celda** (cada clon gasta ≤ su presupuesto).
    pub fn overspend(&self) -> i64 {
        (self.total_claimed() - self.budget).max(0)
    }

    /// **Fusión con enforcement de fencing (§D).** Une los gastos por época
    /// (máximo, monótono por época) y, si aparecen dos linajes, delata el clon.
    pub fn merge(&mut self, other: &FencedCell) -> MergeOutcome {
        debug_assert_eq!(self.device_id, other.device_id, "mismo dispositivo");
        self.budget = self.budget.max(other.budget);
        for (&e, &s) in &other.spent {
            let cur = self.spent.entry(e).or_insert(0);
            *cur = (*cur).max(s);
        }
        let linajes = self.linajes();
        if linajes.len() >= 2 {
            MergeOutcome::CloneDetected {
                device_id: self.device_id,
                stale_epoch: linajes[0],                 // BTreeMap ordenado: menor
                live_epoch: *linajes.last().unwrap(),    // mayor
            }
        } else {
            MergeOutcome::Ok
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use proptest::prelude::*;

    /// §G.3 — detección de clon. Un `device_id` se bifurca en dos épocas; cada
    /// clon gasta su celda entera aislado; al reconectar (merge) se delata,
    /// el daño está acotado a una celda, y el linaje vivo no propaga el doble gasto.
    #[test]
    fn clon_detectado_acotado_y_delatado() {
        // Celda de 50. Clon A (época 1) gasta 50; clon B (época 2) gasta 50.
        let mut a = FencedCell::new(7, 50);
        a.epoch = 1;
        a.spend(50);
        let mut b = FencedCell::new(7, 50);
        b.epoch = 2;
        b.spend(50);

        let outcome = a.merge(&b);
        assert_eq!(
            outcome,
            MergeOutcome::CloneDetected { device_id: 7, stale_epoch: 1, live_epoch: 2 },
            "el clon se delata al reconectar, no pasa silencioso"
        );
        // El doble gasto ocurrió (100 reclamado de una celda de 50)...
        assert_eq!(a.total_claimed(), 100);
        // ...pero está ACOTADO a una celda: sobregiro = 50 = un presupuesto.
        assert_eq!(a.overspend(), 50);
        assert!(a.overspend() <= a.budget, "daño acotado a una celda");
        // El linaje vivo (época 2) es lo único que se propaga como legítimo.
        assert_eq!(a.reconciled_spent(), 50);
    }

    /// §G.4 — fuente monótona (una sola época): merge limpio, comportamiento max.
    #[test]
    fn fuente_monotona_una_epoca_es_ok() {
        let mut a = FencedCell::new(7, 50);
        a.boot(); // época 1
        a.spend(30);
        let mut b = a.clone();
        b.spend(40); // misma época, crece (monótono)

        assert_eq!(a.merge(&b), MergeOutcome::Ok);
        assert_eq!(a.reconciled_spent(), 70, "max por época: 30 y 40 se suman en el linaje");
        assert_eq!(a.overspend(), 20, "70 − 50; no es un clon, es gasto real de la celda");
    }

    /// §G.4 (dientes) — el clon **falla ruidosamente**: nunca es un `Ok` silencioso.
    #[test]
    fn el_clon_nunca_pasa_como_ok() {
        let mut c = FencedCell::new(9, 100);
        c.epoch = 3;
        c.spend(60);
        let mut d = FencedCell::new(9, 100);
        d.epoch = 5;
        d.spend(80);
        assert!(matches!(c.merge(&d), MergeOutcome::CloneDetected { .. }));
    }

    proptest! {
        /// El daño de un clon está SIEMPRE acotado a una celda: dos clones, cada
        /// uno gastando ≤ su presupuesto, dan un sobregiro ≤ una celda, y el clon
        /// se detecta si ambos gastaron.
        #[test]
        fn dano_de_clon_acotado_a_una_celda(
            budget in 1i64..200,
            ea in 1u64..10,
            eb in 1u64..10,
            sa in 0i64..200,
            sb in 0i64..200,
        ) {
            prop_assume!(ea != eb); // dos épocas distintas = clon potencial
            let (sa, sb) = (sa.min(budget), sb.min(budget)); // cada clon ≤ su celda
            let mut a = FencedCell::new(1, budget);
            a.epoch = ea;
            a.spend(sa);
            let mut b = FencedCell::new(1, budget);
            b.epoch = eb;
            b.spend(sb);

            let outcome = a.merge(&b);
            let es_clon = matches!(outcome, MergeOutcome::CloneDetected { .. });
            if sa > 0 && sb > 0 {
                prop_assert!(es_clon, "dos linajes con gasto deben delatar el clon");
            }
            prop_assert!(a.overspend() <= budget, "daño acotado a una celda");
        }
    }
}