//! `invariant` — coordinación de verdad: I-confluence sobre el grafo (T8c, §C–§F).
//!
//! Sin invariante, un CRDT nunca coordina (T8a/b). La coordinación aparece cuando
//! hay un **invariante global `I`** que la fusión no preserva: los estados válidos
//! `V = {s : I(s)}` no son cerrados bajo join. Ejemplo canónico (Bailis et al.,
//! VLDB 2014): **saldo ≥ 0**.
//!
//! Modelo: `bound` = saldo inicial; cada réplica `i` retira `withdrawals[i]`. El
//! estado es un `GCounter` grow-only de retiros (merge = máximo por casilla). El
//! invariante sobre un estado reconciliado `s` es `bound ≥ Σ retiros(s)`.
//!
//! **`AccountConfig` es un DETECTOR reactivo por-estado** (T8c): funde estados ya
//! divergidos y comprueba el merge. T9 §A lo degrada de "consejero de
//! coordinación" a alarma: tiene un **punto ciego** (una restricción global
//! codificada por aristas desaparece al particionar el grafo — ver `COMPARISON.md`
//! y el test `t9_punto_ciego...`). La **evitación** real vive en `escrow` sobre la
//! restricción global `GlobalInvariant` de abajo, no aquí.

use std::collections::{BTreeMap, BTreeSet};

use crate::cell::GCounter;
use crate::native;
use crate::nerve::ReplicaId;
use crate::oracle::OracleVerdict;
use crate::verdict::{Knot, Verdict};

/// Una **restricción global** `valor ≥ floor` (T9 §D): incide sobre TODAS las
/// réplicas que tocan el recurso, compartan arista o no. No es edge-local — ese
/// fue el Error 2 de T8. El escrow reparte su `slack` en presupuestos por réplica.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct GlobalInvariant {
    /// `K` en `valor ≥ K`.
    pub floor: i64,
    /// Valor inicial del recurso.
    pub initial: i64,
}

impl GlobalInvariant {
    /// Crea la restricción `valor ≥ floor` con valor inicial `initial`.
    pub fn new(initial: i64, floor: i64) -> Self {
        Self { floor, initial }
    }

    /// El *slack* `S = valor − K`: lo que se puede gastar sin romper `I`.
    pub fn slack(&self) -> i64 {
        self.initial - self.floor
    }

    /// ¿El valor dado satisface el invariante?
    pub fn holds(&self, value: i64) -> bool {
        value >= self.floor
    }
}

/// Una instancia del programa "retiros sobre una cuenta con saldo ≥ 0".
#[derive(Debug, Clone)]
pub struct AccountConfig {
    /// Nombre del caso.
    pub name: String,
    /// Número de réplicas.
    pub replicas: usize,
    /// Saldo inicial `B₀` (la cota del invariante).
    pub bound: u64,
    /// Retiro concurrente de cada réplica.
    pub withdrawals: Vec<u64>,
    /// Grafo de compartición (quién reconcilia con quién).
    pub edges: Vec<(usize, usize)>,
}

impl AccountConfig {
    /// Construye una instancia (con `withdrawals.len() == replicas`).
    pub fn new(
        name: impl Into<String>,
        bound: u64,
        withdrawals: Vec<u64>,
        edges: Vec<(usize, usize)>,
    ) -> Self {
        Self {
            name: name.into(),
            replicas: withdrawals.len(),
            bound,
            withdrawals,
            edges,
        }
    }

    /// Estado local de cada réplica: un `GCounter` con su propia casilla puesta a
    /// su retiro (grow-only; solo incrementa lo suyo).
    fn seeds(&self) -> Vec<GCounter> {
        (0..self.replicas)
            .map(|i| {
                let mut slots = vec![0u64; self.replicas];
                slots[i] = self.withdrawals[i];
                GCounter { slots }
            })
            .collect()
    }

    /// Estado reconciliado por el orden nativo (merge por componente conexa).
    pub fn reconcile(&self) -> Vec<GCounter> {
        native::reconcile(&self.edges, &self.seeds())
    }

    /// Componentes conexas del grafo (union-find). Camino independiente del merge.
    fn components(&self) -> Vec<usize> {
        let mut parent: Vec<usize> = (0..self.replicas).collect();
        fn find(parent: &mut [usize], x: usize) -> usize {
            let mut r = x;
            while parent[r] != r {
                r = parent[r];
            }
            r
        }
        for &(a, b) in &self.edges {
            let ra = find(&mut parent, a);
            let rb = find(&mut parent, b);
            if ra != rb {
                parent[ra] = rb;
            }
        }
        (0..self.replicas).map(|i| find(&mut parent, i)).collect()
    }

    /// **Veredicto del haz (§C):** difunde al punto fijo y comprueba `∈ V`. Corre
    /// libre si ninguna componente reconciliada sobregira; si alguna lo hace,
    /// localiza las réplicas culpables (las que retiraron) y sus aristas.
    pub fn haz_verdict(&self) -> Verdict {
        let fixpoint = self.reconcile();

        // Vértices cuyo estado reconciliado viola el invariante (sobregiro).
        let mut malos = BTreeSet::new();
        for (v, state) in fixpoint.iter().enumerate() {
            if self.bound < state.value() {
                malos.insert(v);
            }
        }

        if malos.is_empty() {
            return Verdict::RunsFree;
        }

        let replicas: Vec<ReplicaId> = malos
            .iter()
            .filter(|&&v| self.withdrawals[v] > 0)
            .map(|&v| ReplicaId(v))
            .collect();
        let edges: Vec<usize> = self
            .edges
            .iter()
            .enumerate()
            .filter(|(_, &(a, b))| malos.contains(&a) || malos.contains(&b))
            .map(|(i, _)| i)
            .collect();

        Verdict::NeedsCoordination {
            knots: vec![Knot { replicas, edges }],
        }
    }

    /// El veredicto del haz como `OracleVerdict` (para comparar).
    pub fn haz_coordination(&self) -> OracleVerdict {
        match self.haz_verdict() {
            Verdict::RunsFree => OracleVerdict::CoordinationFree,
            Verdict::NeedsCoordination { .. } => OracleVerdict::NeedsCoordination,
        }
    }

    /// **CALM sintáctico por-programa:** el retiro es una operación no monótona
    /// (decrementa el saldo), así que el check crudo condena el programa entero,
    /// sin mirar valores → siempre `coordina`. Es la respuesta a batir (§D).
    pub fn calm_syntactic(&self) -> OracleVerdict {
        OracleVerdict::NeedsCoordination
    }

    /// **Oráculo de I-confluence independiente** (§D): por un camino que NO usa la
    /// difusión — suma aritmética de los retiros por componente vía union-find — y
    /// comprueba `Σ ≤ bound`. El haz debe igualarlo en toda instancia (corrección
    /// y soundness §E.2).
    pub fn i_confluence_oracle(&self) -> OracleVerdict {
        let comp = self.components();
        let mut sums: BTreeMap<usize, u64> = BTreeMap::new();
        for (&c, &w) in comp.iter().zip(&self.withdrawals) {
            *sums.entry(c).or_default() += w;
        }
        if sums.values().all(|&s| s <= self.bound) {
            OracleVerdict::CoordinationFree
        } else {
            OracleVerdict::NeedsCoordination
        }
    }

    /// ¿Es una **discrepancia buena** (versión fuerte, §D instancia A)? El haz
    /// corre libre mientras el CALM sintáctico coordina — con `I` preservado.
    pub fn es_discrepancia_buena(&self) -> bool {
        self.haz_coordination() == OracleVerdict::CoordinationFree
            && self.calm_syntactic() == OracleVerdict::NeedsCoordination
    }
}

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

    /// **El premio (§D, instancia A):** dos réplicas retiran 10 cada una de una
    /// cuenta de 100. Los valores caben (20 ≤ 100) → el haz corre libre. Pero el
    /// retiro es no monótono → CALM sintáctico coordina. Discrepancia buena: el
    /// haz es más fino que el check por-programa. Testigo para `STRONG_RESULT.md`.
    #[test]
    fn discrepancia_buena_existe() {
        let a = AccountConfig::new("retiros-que-caben", 100, vec![10, 10], vec![(0, 1)]);
        assert_eq!(a.haz_coordination(), OracleVerdict::CoordinationFree);
        assert_eq!(a.calm_syntactic(), OracleVerdict::NeedsCoordination);
        assert_eq!(a.i_confluence_oracle(), OracleVerdict::CoordinationFree);
        assert!(a.es_discrepancia_buena());
    }

    /// Instancia B: los retiros sobregiran (60 + 60 > 100) → el haz coordina, y
    /// coincide con I-confluence. Además localiza a las dos réplicas culpables.
    #[test]
    fn sobregiro_coordina_y_localiza() {
        let a = AccountConfig::new("retiros-que-sobregiran", 100, vec![60, 60], vec![(0, 1)]);
        assert_eq!(a.haz_coordination(), OracleVerdict::NeedsCoordination);
        assert_eq!(a.i_confluence_oracle(), OracleVerdict::NeedsCoordination);
        match a.haz_verdict() {
            Verdict::NeedsCoordination { knots } => {
                assert_eq!(knots[0].replicas, vec![ReplicaId(0), ReplicaId(1)]);
            }
            Verdict::RunsFree => panic!("debería coordinar"),
        }
    }

    /// **El PUNTO CIEGO del detector** (T9 §A, Error 2). Lo que T8c celebraba como
    /// "la topología importa" es en realidad un falso-verde peligroso: con los
    /// mismos retiros 60/60 sobre 100, el detector dice `libre` al **particionar**
    /// —el caso MÁS peligroso— porque, sin arista, no ve el acoplamiento del
    /// invariante GLOBAL. No es que sea seguro; es que el haz está ciego. La cura
    /// (escrow) bloquea correctamente ese caso: ver `escrow::tests` y `COMPARISON.md`.
    #[test]
    fn t9_punto_ciego_del_detector_particionado() {
        let desconectado = AccountConfig::new("particion", 100, vec![60, 60], vec![]);
        let conectado = AccountConfig::new("conjunto", 100, vec![60, 60], vec![(0, 1)]);
        // Conectado detecta el sobregiro...
        assert_eq!(conectado.haz_coordination(), OracleVerdict::NeedsCoordination);
        // ...pero particionado lo declara "libre": ESO es el bug que motiva T9.
        assert_eq!(desconectado.haz_coordination(), OracleVerdict::CoordinationFree);
    }

    /// Barrido determinista que **caza** discrepancias buenas: hay muchas, no una
    /// casualidad. Prueba de que la versión fuerte es robusta.
    #[test]
    fn la_caza_encuentra_muchas_discrepancias() {
        let mut encontradas = 0;
        for total in 0..=90u64 {
            for w0 in 0..=total {
                let a = AccountConfig::new("barrido", 100, vec![w0, total - w0], vec![(0, 1)]);
                if a.es_discrepancia_buena() {
                    encontradas += 1;
                }
            }
        }
        assert!(encontradas > 100, "la discrepancia buena es robusta: {encontradas}");
    }

    fn instancia() -> impl Strategy<Value = AccountConfig> {
        (2usize..=5, 1u64..=50).prop_flat_map(|(n, bound)| {
            let ws = prop::collection::vec(0u64..=40, n);
            let arista = (0..n, 0..n).prop_filter("sin lazos", |(a, b)| a != b);
            let es = prop::collection::vec(arista, 0..8);
            (Just(bound), ws, es)
                .prop_map(|(bound, ws, es)| AccountConfig::new("prop", bound, ws, es))
        })
    }

    proptest! {
        /// **Corrección (§D):** el haz iguala al oráculo de I-confluence en TODA
        /// instancia. Dos caminos independientes (difusión de merges vs suma por
        /// componente) que coinciden siempre.
        #[test]
        fn haz_iguala_i_confluence(a in instancia()) {
            prop_assert_eq!(a.haz_coordination(), a.i_confluence_oracle());
        }

        /// **Soundness (§E.2, no se negocia):** el haz NUNCA dice "corre libre"
        /// cuando alguna componente reconciliada viola el invariante.
        #[test]
        fn soundness_el_haz_nunca_miente(a in instancia()) {
            if a.haz_coordination() == OracleVerdict::CoordinationFree {
                for state in a.reconcile() {
                    prop_assert!(a.bound >= state.value(), "libre ⟹ invariante preservado");
                }
            }
        }
    }
}