//! `escrow` — evitación de coordinación por presupuestos (T9, §B).
//!
//! Método de escrow (O'Neil 1986) como Bounded Counter CRDT (Balegas et al.,
//! SRDS 2015). Para un invariante global `valor ≥ K`:
//!
//! 1. El *slack* `S = valor − K` es lo gastable sin romper `I`.
//! 2. Se **reparte** `S` en presupuestos por réplica `b_i` con `Σ b_i ≤ S`.
//! 3. Mientras `r_i` tenga presupuesto, **gasta localmente sin coordinar** — y es
//!    correcto **incluso particionada**, porque `b_i` es suyo.
//! 4. La única coordinación es el **rebalanceo edge-local** al agotar `b_i` (T9b).
//!
//! **El certificado.** Si `spent_i ≤ b_i` en cada réplica, entonces
//! `valor = inicial − Σ spent_i ≥ inicial − Σ b_i ≥ inicial − S = K`, para **toda
//! ejecución, orden y partición**. No se descubre tarde: se garantiza por
//! construcción. Eso es evitación real, no la detección reactiva de T8 (§A).
//!
//! **Honestidad (§B):** escrow no vence a CALM; *transforma* la operación
//! ("retirar sin límite" → "retirar dentro de presupuesto") para que sea
//! I-confluente, al costo de un rebalanceo ocasional.

use crate::invariant::GlobalInvariant;

/// El resultado de intentar gastar localmente.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SpendResult {
    /// Dentro de presupuesto: gasto local, **cero coordinación**.
    Local,
    /// Presupuesto agotado: hace falta rebalanceo edge-local (T9b).
    NeedsRebalance {
        /// Cuánto falta para cubrir el gasto pedido.
        deficit: i64,
    },
}

/// El presupuesto repartido: la 0-cocadena `b` sobre las réplicas.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Budget {
    /// Presupuesto restante por réplica.
    pub per_replica: Vec<i64>,
}

impl Budget {
    /// Presupuesto a partir de un reparto explícito.
    pub fn new(per_replica: Vec<i64>) -> Self {
        Self { per_replica }
    }

    /// Reparte el `slack` en `n` presupuestos proporcionales a `weights`, con
    /// división entera para garantizar `Σ b_i ≤ slack` (reparto siempre válido).
    pub fn split(inv: &GlobalInvariant, weights: &[u64]) -> Self {
        let s = inv.slack().max(0) as u128;
        let total: u128 = weights.iter().map(|&w| w as u128).sum();
        let per_replica = if total == 0 {
            vec![0; weights.len()]
        } else {
            // ⌊w_i · S / total⌋ ⟹ la suma nunca excede S.
            weights.iter().map(|&w| ((w as u128 * s) / total) as i64).collect()
        };
        Self { per_replica }
    }

    /// El certificado del reparto (§B): `Σ b_i ≤ slack` y todo `b_i ≥ 0`.
    pub fn is_valid_split(&self, inv: &GlobalInvariant) -> bool {
        self.per_replica.iter().all(|&b| b >= 0)
            && self.per_replica.iter().sum::<i64>() <= inv.slack()
    }

    /// Presupuesto restante de la réplica `r`.
    pub fn remaining(&self, r: usize) -> i64 {
        self.per_replica[r]
    }

    /// Gasto local de `amount` (≥ 0) en la réplica `r`. Si cabe, descuenta y
    /// devuelve `Local`; si no, **no gasta** y devuelve `NeedsRebalance` con el
    /// déficit. Nunca declara "libre global" a ciegas.
    pub fn try_spend(&mut self, r: usize, amount: i64) -> SpendResult {
        debug_assert!(amount >= 0, "los gastos son no negativos");
        if amount <= self.per_replica[r] {
            self.per_replica[r] -= amount;
            SpendResult::Local
        } else {
            SpendResult::NeedsRebalance {
                deficit: amount - self.per_replica[r],
            }
        }
    }

    /// El valor del recurso implicado por unos gastos `spent_i` (para el
    /// certificado): `inicial − Σ spent_i`.
    pub fn value_after(inv: &GlobalInvariant, spent: &[i64]) -> i64 {
        inv.initial - spent.iter().sum::<i64>()
    }
}

/// **Presupuesto con procedencia** (T10, §C.1 — la vía Balegas): todo es monótono
/// creciente, así que la fusión es un CRDT y **no hay carreras**. Sustituye al
/// `Budget` mutable de T9 (`b_i` leído-y-escrito), que era seguro solo en
/// snapshots congelados y secuenciales.
///
/// El presupuesto disponible es una **función derivada** de dos historiales que
/// solo crecen: `granted[from][to]` (derechos cedidos) y `spent[i]` (gasto).
///
/// **El teorema (§C):** `Σ avail = Σ b0 − Σ spent`, invariante bajo cualquier
/// entrelazado, porque cada `granted[j][i]` entra `+` en `i` y `−` en `j` y se
/// cancela. Imposible fabricar presupuesto por fusión; imposible perderlo por un
/// crash a medio propagar (un `grant` vive entero en `granted[from][to]`: al
/// re-fusionar, o llegó completo o no llegó).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EscrowState {
    /// Reparto inicial (constante); `Σ b0 ≤ slack`.
    pub b0: Vec<i64>,
    /// `granted[from][to]`: derechos transferidos, solo crece. Fila `from`
    /// escrita solo por la réplica `from` (single-writer ⟹ sin conflicto).
    pub granted: Vec<Vec<i64>>,
    /// `spent[i]`: gasto de la réplica `i`, solo crece.
    pub spent: Vec<i64>,
}

impl EscrowState {
    /// Estado inicial a partir de un reparto `b0` (sin transferencias ni gasto).
    pub fn new(b0: Vec<i64>) -> Self {
        let n = b0.len();
        Self {
            b0,
            granted: vec![vec![0; n]; n],
            spent: vec![0; n],
        }
    }

    /// Número de réplicas.
    pub fn replicas(&self) -> usize {
        self.b0.len()
    }

    /// Presupuesto disponible de `i` = inicial + recibido − cedido − gastado.
    pub fn avail(&self, i: usize) -> i64 {
        let recibido: i64 = (0..self.replicas()).map(|j| self.granted[j][i]).sum();
        let cedido: i64 = self.granted[i].iter().sum();
        self.b0[i] + recibido - cedido - self.spent[i]
    }

    /// Presupuesto total disponible. Por el teorema de §C es exactamente
    /// `Σ b0 − Σ spent` — los `granted` se cancelan y **no cambian el total**.
    pub fn total_avail(&self) -> i64 {
        self.b0.iter().sum::<i64>() - self.spent.iter().sum::<i64>()
    }

    /// La otra vía de calcular el total (sumando cada `avail`), que DEBE coincidir
    /// con `total_avail`: la prueba de la cancelación de `granted`.
    pub fn sum_avail(&self) -> i64 {
        (0..self.replicas()).map(|i| self.avail(i)).sum()
    }

    /// La réplica `from` cede `amount` (≥0) a `to`. Guardado por su `avail` local
    /// (single-writer): nunca cede más de lo que tiene ⟹ imposible fabricar en el
    /// origen. Devuelve `false` si no le alcanza.
    pub fn grant(&mut self, from: usize, to: usize, amount: i64) -> bool {
        debug_assert!(amount >= 0);
        if amount <= self.avail(from) {
            self.granted[from][to] += amount;
            true
        } else {
            false
        }
    }

    /// La réplica `i` gasta `amount` (≥0) si le queda `avail`. Monótono.
    pub fn spend(&mut self, i: usize, amount: i64) -> bool {
        debug_assert!(amount >= 0);
        if amount <= self.avail(i) {
            self.spent[i] += amount;
            true
        } else {
            false
        }
    }

    /// **Fusión CRDT**: máximo componente a componente en `granted` y `spent`
    /// (los historiales solo crecen). Conmutativa, asociativa, idempotente. No
    /// puede subir `total_avail`: `Σ spent` solo crece ⟹ el total solo baja.
    pub fn merge(&mut self, other: &EscrowState) {
        debug_assert_eq!(self.b0, other.b0, "mismo reparto inicial");
        let n = self.replicas();
        for i in 0..n {
            self.spent[i] = self.spent[i].max(other.spent[i]);
            for j in 0..n {
                self.granted[i][j] = self.granted[i][j].max(other.granted[i][j]);
            }
        }
    }
}

/// Una **identidad** como **conjunto de celdas-dispositivo** (T11a, §C, cura de
/// F2). No se modela la identidad como una réplica: un usuario es un conjunto de
/// dispositivos, cada uno una réplica del esquema de T10 con su propia celda de
/// presupuesto y su propio `spent`. Así la disciplina single-writer se cumple a
/// nivel de **celda-dispositivo**, no de identidad: dos dispositivos gastan sin
/// serializar, sin carrera, sin fencing (F2 es un no-problema por diseño).
///
/// El invariante (el recurso, p.ej. saldo ≥ 0) es de la identidad; su
/// disponibilidad y su certificado se computan sobre SUS dispositivos.
#[derive(Debug, Clone)]
pub struct Identity {
    /// Índices de las celdas-dispositivo de esta identidad en el `EscrowState`.
    pub devices: Vec<usize>,
    /// El invariante del recurso de esta identidad.
    pub inv: GlobalInvariant,
}

impl Identity {
    /// Crea una identidad con sus dispositivos y su invariante.
    pub fn new(devices: Vec<usize>, inv: GlobalInvariant) -> Self {
        Self { devices, inv }
    }

    /// Presupuesto disponible de la identidad = Σ del `avail` de sus dispositivos.
    /// Los `grant` **entre** dispositivos de la identidad se cancelan ⟹ solo lo
    /// mueven `b0` y `spent`: `total_avail = Σ b0[d] − Σ spent[d]`.
    pub fn total_avail(&self, state: &EscrowState) -> i64 {
        self.devices.iter().map(|&d| state.avail(d)).sum()
    }

    /// Valor del recurso de la identidad: `inicial − Σ spent[d]` sobre sus celdas.
    pub fn value(&self, state: &EscrowState) -> i64 {
        let gastado: i64 = self.devices.iter().map(|&d| state.spent[d]).sum();
        self.inv.initial - gastado
    }

    /// ¿Se preserva el invariante de la identidad?
    pub fn invariant_holds(&self, state: &EscrowState) -> bool {
        self.inv.holds(self.value(state))
    }
}

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

    /// El caso 60/60 de T8, ahora bien tratado (§F): con presupuestos [50,50]
    /// (reparto válido de S=100), una réplica **particionada** que intenta retirar
    /// 60 recibe `NeedsRebalance`, NO `Local`. Muere el falso-verde del detector.
    #[test]
    fn t9_particionado_60_es_bloqueado_no_libre() {
        let inv = GlobalInvariant::new(100, 0); // saldo ≥ 0, slack 100
        let mut budget = Budget::new(vec![50, 50]);
        assert!(budget.is_valid_split(&inv));

        // La réplica 0, aislada, intenta retirar 60 (> su presupuesto de 50).
        let res = budget.try_spend(0, 60);
        assert_eq!(res, SpendResult::NeedsRebalance { deficit: 10 });
        // El veredicto de escrow NO es "libre": pide rebalanceo (localizado).
        assert!(matches!(Verdict::via_escrow(0, &res), Verdict::NeedsCoordination { .. }));
        // Y no gastó: el presupuesto sigue intacto (el gasto queda pendiente).
        assert_eq!(budget.remaining(0), 50);
    }

    /// Gasto dentro de presupuesto: local y descuenta.
    #[test]
    fn gasto_dentro_de_presupuesto_es_local() {
        let mut budget = Budget::new(vec![50, 50]);
        assert_eq!(budget.try_spend(0, 30), SpendResult::Local);
        assert_eq!(budget.remaining(0), 20);
        assert!(matches!(Verdict::via_escrow(0, &SpendResult::Local), Verdict::RunsFree));
    }

    fn split_valido(inv: &GlobalInvariant, raw: &[u64]) -> Budget {
        let b = Budget::split(inv, raw);
        assert!(b.is_valid_split(inv), "split debe ser válido por construcción");
        b
    }

    proptest! {
        /// **EL CERTIFICADO (§F, el corazón de T9a).** Para todo reparto válido y
        /// TODA combinación de gastos dentro de presupuesto —incluidas particiones
        /// arbitrarias y todas las réplicas gastando su máximo a la vez— el
        /// invariante global `valor ≥ K` se preserva. Se cuantifica sobre todos
        /// los estados alcanzables, no sobre un merge concreto.
        #[test]
        fn certificado_valor_nunca_baja_del_piso(
            initial in 0i64..500,
            floor in -100i64..100,
            pesos in prop::collection::vec(0u64..50, 1..6),
            gasto_raw in prop::collection::vec(0u64..600, 1..6),
        ) {
            prop_assume!(initial >= floor);              // slack ≥ 0
            let inv = GlobalInvariant::new(initial, floor);
            let budget = split_valido(&inv, &pesos);

            // Gastos arbitrarios pero acotados por el presupuesto de cada réplica.
            let spent: Vec<i64> = budget
                .per_replica
                .iter()
                .zip(gasto_raw.iter().chain(std::iter::repeat(&0)))
                .map(|(&b, &g)| (g as i64).min(b))       // spent_i ≤ b_i
                .collect();

            // El certificado: valor = inicial − Σ spent ≥ floor. Para cualquier
            // subconjunto que sincronice (partición) el valor es aún mayor.
            let valor = Budget::value_after(&inv, &spent);
            prop_assert!(inv.holds(valor), "valor {} < piso {}", valor, inv.floor);
        }
    }

    // ---- T10a: presupuesto con procedencia (EscrowState) ----

    /// **El modo de fallo de §B, PREVENIDO en el origen.** `B` tiene 30. Cede 20 a
    /// `A`; el segundo intento de ceder 20 a `C` **falla** (solo le quedan 10),
    /// porque `grant` está guardado por el `avail` local. Nada de fabricar 20 por
    /// una lectura-y-escritura descuidada. `Σ avail` se conserva en 30.
    #[test]
    fn grant_guardado_previene_la_fabricacion() {
        let mut s = EscrowState::new(vec![0, 30, 0]);
        assert!(s.grant(1, 0, 20), "B cede 20 a A");
        assert!(!s.grant(1, 2, 20), "B ya no puede ceder otros 20 (solo le quedan 10)");
        assert!(s.grant(1, 2, 10), "pero sí los 10 restantes");
        assert_eq!(s.avail(0), 20);
        assert_eq!(s.avail(1), 0);
        assert_eq!(s.avail(2), 10);
        assert_eq!(s.total_avail(), 30, "Σ avail conservado: cero fabricación");
    }

    /// Crash a mitad de transferencia: un `grant` aplicado en una copia y no
    /// propagado, al re-fusionar tarde da el mismo estado. Ni fabrica ni pierde.
    #[test]
    fn crash_a_mitad_de_transferencia_es_seguro() {
        let base = EscrowState::new(vec![30, 0]);
        // Copia 1: B0 concede 20 a la réplica 1 (el "grant en tránsito").
        let mut c1 = base.clone();
        assert!(c1.grant(0, 1, 20));
        // Copia 2 (crashea sin ver el grant): fusiona el resto tal cual.
        let mut c2 = base.clone();
        c2.merge(&base);
        // El grant llega tarde: c2 re-fusiona con c1.
        c2.merge(&c1);
        assert_eq!(c2, c1, "el grant llega entero o no llega; nunca a medias");
        assert_eq!(c2.total_avail(), 30);
    }

    /// **§G.2 — réplica por dispositivo (F2).** Una identidad con dos
    /// dispositivos (celdas `[50,50]`, cuenta 100). Ambos gastan concurrentemente
    /// SU celda sin serializar; el gasto de uno **no toca** el `avail` del otro;
    /// el `total_avail` de la identidad es `Σ celdas − Σ spent`, invariante.
    #[test]
    fn t11a_dos_dispositivos_gastan_sin_pisarse() {
        let inv = GlobalInvariant::new(100, 0);
        let id = Identity::new(vec![0, 1], inv);
        let mut state = EscrowState::new(vec![50, 50]);

        assert!(state.spend(0, 40));
        assert!(state.spend(1, 30));
        assert_eq!(state.avail(1), 20, "el gasto de 0 no tocó el avail de 1");
        assert_eq!(id.total_avail(&state), 30, "Σ celdas − Σ spent = 100 − 70");
        assert!(id.invariant_holds(&state));

        // Cada celda tiene un único escritor lógico: 0 no puede gastar la de 1.
        assert!(!state.spend(0, 20), "0 ya solo tiene 10 en su propia celda");
    }

    /// Rebalanceo **entre dispositivos** de una identidad = un `grant`, que
    /// conserva el `total_avail` (T10a) — el rebalanceo edge-local de T10b.
    #[test]
    fn t11a_rebalanceo_entre_dispositivos_conserva() {
        let inv = GlobalInvariant::new(100, 0);
        let id = Identity::new(vec![0, 1], inv);
        let mut state = EscrowState::new(vec![0, 50]);
        let antes = id.total_avail(&state);

        assert!(state.grant(1, 0, 20), "el dispositivo 1 cede 20 al 0 (seco)");
        assert_eq!(state.avail(0), 20);
        assert_eq!(state.avail(1), 30);
        assert_eq!(id.total_avail(&state), antes, "el rebalanceo conserva");
    }

    /// Identidades disjuntas no se pisan (F2): la disponibilidad de una identidad
    /// depende **solo** de sus dispositivos; la actividad de otra no la afecta.
    #[test]
    fn t11a_identidades_disjuntas_aisladas() {
        let inv = GlobalInvariant::new(100, 0);
        let a = Identity::new(vec![0, 1], inv);
        let b = Identity::new(vec![2, 3], inv);
        let mut state = EscrowState::new(vec![50, 50, 50, 50]);

        // Toda la actividad ocurre en la identidad B.
        assert!(state.spend(2, 40));
        assert!(state.grant(3, 2, 10));

        assert_eq!(a.total_avail(&state), 100, "A intacta");
        assert!(a.invariant_holds(&state));
        assert!(b.invariant_holds(&state));
    }

    fn escrow_st() -> impl Strategy<Value = EscrowState> {
        (1usize..=4).prop_flat_map(|n| {
            let b0 = prop::collection::vec(0i64..50, n);
            let granted = prop::collection::vec(prop::collection::vec(0i64..30, n), n);
            let spent = prop::collection::vec(0i64..30, n);
            (b0, granted, spent).prop_map(|(b0, granted, spent)| EscrowState { b0, granted, spent })
        })
    }

    /// Dos estados con el mismo reparto inicial (para poder fusionarlos).
    fn par_de_escrow() -> impl Strategy<Value = (EscrowState, EscrowState)> {
        escrow_st().prop_flat_map(|a| {
            let n = a.replicas();
            let granted = prop::collection::vec(prop::collection::vec(0i64..30, n), n);
            let spent = prop::collection::vec(0i64..30, n);
            let b0 = a.b0.clone();
            (Just(a), (granted, spent)).prop_map(move |(a, (g, s))| {
                let b = EscrowState { b0: b0.clone(), granted: g, spent: s };
                (a, b)
            })
        })
    }

    /// Fusiona todas las copias en una vista global (el estado "verdadero").
    fn merge_all(copies: &[EscrowState], b0: &[i64]) -> EscrowState {
        let mut g = EscrowState::new(b0.to_vec());
        for c in copies {
            g.merge(c);
        }
        g
    }

    /// Escenario dinámico: invariante, reparto por pesos, y una secuencia de
    /// eventos `(op, x, y, monto)` a interpretar como spend/grant/merge.
    #[allow(clippy::type_complexity)]
    fn escenario() -> impl Strategy<Value = (GlobalInvariant, Vec<u64>, Vec<(u8, usize, usize, i64)>)> {
        (2usize..=4, 0i64..200, -50i64..50).prop_flat_map(|(n, slack, floor)| {
            let inv = GlobalInvariant::new(floor + slack, floor);
            let pesos = prop::collection::vec(0u64..20, n);
            let evento = (0u8..3, 0..n, 0..n, 0i64..80);
            let eventos = prop::collection::vec(evento, 0..40);
            (Just(inv), pesos, eventos)
        })
    }

    proptest! {
        /// **§F.5 — EL CERTIFICADO DINÁMICO (el teorema de T10).** Entrelaza gastos,
        /// transferencias (`grant`) concurrentes, crashes (grants sin propagar) y
        /// particiones (merges en orden arbitrario) sobre copias por réplica. En
        /// **todo momento** `valor = inicial − Σ spent ≥ K`, y el presupuesto total
        /// nunca se fabrica (`total_avail ≥ 0`). Subsume el certificado estático de
        /// T9: ya no vale solo para snapshots congelados, sino bajo concurrencia.
        #[test]
        fn certificado_dinamico(( inv, pesos, eventos) in escenario()) {
            let n = pesos.len();
            // Reparto válido: Σ b0 ≤ slack (mismo split proporcional de T9a).
            let b0 = Budget::split(&inv, &pesos).per_replica;
            // Una copia local por réplica (vistas divergibles bajo partición).
            let mut copies = vec![EscrowState::new(b0.clone()); n];

            for (op, x, y, amount) in eventos {
                let i = x % n;
                let j = y % n;
                match op % 3 {
                    // Gasto local, guardado por el avail local (single-writer índice i).
                    0 => { copies[i].spend(i, amount); }
                    // Transferencia: la réplica i cede desde SU fila, guardada.
                    1 => { copies[i].grant(i, j, amount); }
                    // Sincronización: i fusiona la copia de j (orden arbitrario = partición/crash).
                    _ => { let otra = copies[j].clone(); copies[i].merge(&otra); }
                }

                // El teorema, tras cada evento: la vista global no fabrica ni sobregira.
                let global = merge_all(&copies, &b0);
                prop_assert!(global.total_avail() >= 0, "presupuesto fabricado: total_avail < 0");
                let valor = inv.initial - global.spent.iter().sum::<i64>();
                prop_assert!(inv.holds(valor), "valor {} < piso {}", valor, inv.floor);
            }
        }

        /// **§F.1 — el total no se puede fabricar.** `Σ avail` (sumando cada réplica)
        /// SIEMPRE es igual a `Σ b0 − Σ spent`: los `granted` se cancelan, pase lo
        /// que pase con las transferencias. Este es el teorema que mata el §B.
        #[test]
        fn total_avail_es_invariante_a_los_granted(s in escrow_st()) {
            prop_assert_eq!(s.sum_avail(), s.total_avail());
            prop_assert_eq!(s.total_avail(), s.b0.iter().sum::<i64>() - s.spent.iter().sum::<i64>());
        }

        /// **§F.2 — `merge` es un CRDT:** conmutativo e idempotente.
        #[test]
        fn merge_conmutativo_e_idempotente((a, b) in par_de_escrow()) {
            let mut ab = a.clone(); ab.merge(&b);
            let mut ba = b.clone(); ba.merge(&a);
            prop_assert_eq!(&ab, &ba, "merge conmutativo");
            let mut abb = ab.clone(); abb.merge(&b);
            prop_assert_eq!(&abb, &ab, "merge idempotente");
        }

        /// **§F.1 (fusión) — `merge` nunca sube el total.** `Σ spent` solo crece por
        /// max ⟹ `total_avail` solo puede bajar. Imposible fabricar por fusión.
        #[test]
        fn merge_nunca_fabrica_presupuesto((a, b) in par_de_escrow()) {
            let mut m = a.clone(); m.merge(&b);
            prop_assert!(m.total_avail() <= a.total_avail());
            prop_assert!(m.total_avail() <= b.total_avail());
        }
    }
}