//! m11p4 exit gates — automatic failover, fencing, bounded churn — over a
//! REAL 3-process cluster (tier 3, `cluster-e2e` feature).
//!
//! The three gates from docs/roadmap-to-cluster.md §4/m11p4:
//!
//! 1. **Auto-failover, zero acked loss** (`mp_auto_failover_*`): SIGKILL the
//!    leader under `ack=quorum` load with ZERO operator verbs → a survivor is
//!    elected and writes resume in <10s, and the m11p3 ledger invariants
//!    (frontier + content) hold on the new leader, across repeated rounds
//!    with pseudo-random kill points.
//! 2. **Fencing under partition + restart** (`mp_fenced_ex_leader_*`): the
//!    leader is partitioned away (real TCP severs), the survivors elect, the
//!    old leader RESTARTS while still partitioned — and cannot accept a
//!    single write (its durable boot state forbids self-leadership, the
//!    §1.4-1 fix), then rejoins as a follower on heal.
//! 3. **Bounded churn** (`mp_flapping_links_bounded_churn`): repeated
//!    sever/heal cycles on the leader's links produce bounded elections (the
//!    pre-vote absorbs flaps; terms never explode) and the cluster converges
//!    to exactly one leader that serves quorum writes.
//!
//! Election timings are tuned fast (500–1000ms timeouts) so the suite stays
//! within the tier-3 budget; the production defaults scale the same
//! machinery up, not a different protocol.
#![cfg(feature = "cluster-e2e")]
#![allow(clippy::unwrap_used, clippy::significant_drop_tightening)]

mod support;

use std::sync::{
    Arc,
    atomic::{AtomicBool, Ordering},
};
use std::time::{Duration, Instant};

use support::{
    multiproc::{ClusterOptions, MultiProcCluster, convergence_budget},
    partition::{ProxyController, proxied_rewrite},
};

/// All three roster regions, so every directed edge gets its own relay and a
/// node can be isolated in BOTH directions (inbound edges via
/// `region(r).sever_all()` cut what reaches it; outbound edges via
/// `edge(r, peer)` cut its own heartbeats/ships — without the outbound cut a
/// "partitioned" LEADER keeps resetting every follower's election timer).
const ALL_REGIONS: [&str; 3] = ["us-east", "eu-west", "ap-south"];

/// Fully isolate `region` from `peers` (both directions, gRPC + HTTP).
fn isolate(proxies: &ProxyController, region: &str, peers: &[&str]) {
    proxies.region(region).sever_all();
    for peer in peers {
        proxies.edge(region, peer).sever_all();
    }
}

/// Undo [`isolate`].
fn rejoin(proxies: &ProxyController, region: &str, peers: &[&str]) {
    proxies.region(region).heal_all();
    for peer in peers {
        proxies.edge(region, peer).heal_all();
    }
}

/// The fast election block every test in this suite runs with. Constraint:
/// `lease (350) + heartbeat (100) < timeout_min (500)` — the C2 bound.
const FAST_ELECTION_YAML: &str = "election:\n  heartbeat_interval_ms: 100\n  election_timeout_min_ms: 500\n  election_timeout_max_ms: 1000\n  leader_lease_ms: 350";

/// The exit gate's failover budget: detect → elect → writes resume.
const FAILOVER_BUDGET: Duration = Duration::from_secs(10);

/// A unique all-alpha search token for an entity id (mirrors the m11p3
/// ledger checker's probe).
fn item_token(entity_id: u64) -> String {
    let mut token = String::from("elx");
    for d in entity_id.to_string().bytes() {
        token.push(char::from(b'a' + (d - b'0')));
    }
    token
}

/// POST with `x-tidal-ack` through a dedicated client. `Some(seq)` only for
/// a 2xx carrying `x-tidal-seq` — the ledger's definition of "acknowledged".
fn post_acked(
    client: &reqwest::blocking::Client,
    base: &str,
    path: &str,
    ack: &str,
    body: &serde_json::Value,
) -> Option<u64> {
    let resp = client
        .post(format!("{base}{path}"))
        .header("x-tidal-ack", ack)
        .json(body)
        .send()
        .ok()?;
    if !resp.status().is_success() {
        return None;
    }
    resp.headers()
        .get("x-tidal-seq")?
        .to_str()
        .ok()?
        .parse()
        .ok()
}

/// Poll the LIVE nodes for an elected leader: a node whose local status
/// reports `role == "leader"` at a term above `after_term`. Returns
/// `(node_idx, term, elapsed)`.
fn await_elected_leader(
    cluster: &MultiProcCluster,
    candidates: &[usize],
    after_term: u64,
    budget: Duration,
) -> (usize, u64, Duration) {
    let started = Instant::now();
    let deadline = started + budget;
    loop {
        for &idx in candidates {
            if let Some(status) = cluster.local_status(idx) {
                let term = status["term"].as_u64().unwrap_or(0);
                if status["role"].as_str() == Some("leader") && term > after_term {
                    return (idx, term, started.elapsed());
                }
            }
        }
        assert!(
            Instant::now() < deadline,
            "no leader elected among {candidates:?} within {budget:?} \
             (terms must move past {after_term})"
        );
        std::thread::sleep(Duration::from_millis(50));
    }
}

/// Write through `base` until one `ack=quorum` write succeeds (or the
/// deadline passes). Returns the first acked seq.
fn await_write_resumes(
    client: &reqwest::blocking::Client,
    base: &str,
    entity_id: u64,
    deadline: Instant,
) -> u64 {
    loop {
        if let Some(seq) = post_acked(
            client,
            base,
            "/signals",
            "quorum",
            &serde_json::json!({ "entity_id": entity_id, "signal": "view", "weight": 1.0 }),
        ) {
            return seq;
        }
        assert!(
            Instant::now() < deadline,
            "quorum writes did not resume before the failover budget expired"
        );
        std::thread::sleep(Duration::from_millis(50));
    }
}

/// Gate 1: kill the leader under `ack=quorum` load — zero operator verbs —
/// across repeated rounds. Every round asserts: a survivor is elected and
/// writes resume within the 10s budget; the new leader's durable frontier
/// covers every acknowledged seq (INVARIANT A); every acknowledged item is
/// searchable on the new leader (INVARIANT B). The killed node restarts and
/// rejoins (reseeded if it quarantined with a divergent suffix — writes that
/// were leader-staged but never quorum-acked).
// One linear multi-round drill (load -> kill -> elect -> invariants ->
// rejoin); splitting it would scatter the round's ordering rules.
#[allow(clippy::too_many_lines)]
#[test]
fn mp_auto_failover_writes_resume_zero_acked_loss() {
    let rounds: usize = std::env::var("TIDAL_ELECTION_KILLPOINTS")
        .ok()
        .and_then(|v| v.parse().ok())
        .filter(|&n| n > 0)
        .unwrap_or(5);
    let mut cluster = MultiProcCluster::start_with(
        ClusterOptions::new(3).with_topology_extra(FAST_ELECTION_YAML),
    );
    let client = reqwest::blocking::Client::builder()
        .timeout(Duration::from_secs(3))
        .build()
        .unwrap();

    let mut entity_cursor: u64 = 1;
    let mut current_leader = 0usize;
    let mut last_term = 0u64;

    for round in 0..rounds {
        // ── Load: two writer threads against the current leader. ──────────
        let stop = Arc::new(AtomicBool::new(false));
        let leader_base = cluster.node(current_leader);
        let mut writers = Vec::new();
        for w in 0..2u64 {
            let stop = Arc::clone(&stop);
            let base = leader_base.clone();
            let first_entity = entity_cursor + w * 10_000;
            writers.push(std::thread::spawn(move || {
                let client = reqwest::blocking::Client::builder()
                    .timeout(Duration::from_secs(3))
                    .build()
                    .unwrap();
                let mut acked: Vec<(u64, u64)> = Vec::new();
                let mut entity = first_entity;
                while !stop.load(Ordering::Acquire) {
                    let item_seq = post_acked(
                        &client,
                        &base,
                        "/items",
                        "quorum",
                        &serde_json::json!({
                            "entity_id": entity,
                            "metadata": { "title": item_token(entity) },
                        }),
                    );
                    let view_seq = post_acked(
                        &client,
                        &base,
                        "/signals",
                        "quorum",
                        &serde_json::json!({
                            "entity_id": entity, "signal": "view", "weight": 1.0
                        }),
                    );
                    if let Some(seq) = item_seq {
                        acked.push((entity, seq.max(view_seq.unwrap_or(0))));
                    }
                    entity += 1;
                }
                acked
            }));
        }

        // Pseudo-random kill point per round (reproducible).
        std::thread::sleep(Duration::from_millis(150 + (round as u64 * 97) % 400));
        cluster.kill_hard(current_leader);
        stop.store(true, Ordering::Release);
        let mut ledger: Vec<(u64, u64)> = Vec::new();
        for w in writers {
            ledger.extend(w.join().expect("writer thread"));
        }
        let max_acked_seq = ledger.iter().map(|&(_, s)| s).max().unwrap_or(0);

        // ── ZERO operator verbs: the survivors elect on their own. ─────────
        let survivors: Vec<usize> = (0..3).filter(|&i| i != current_leader).collect();
        let (new_leader, new_term, elapsed) =
            await_elected_leader(&cluster, &survivors, last_term, FAILOVER_BUDGET);
        // Writes must RESUME (not just leadership exist) inside the budget.
        let resume_deadline = Instant::now() + FAILOVER_BUDGET;
        let _ = await_write_resumes(
            &client,
            &cluster.node(new_leader),
            900_000 + round as u64,
            resume_deadline,
        );
        println!(
            "round {round}: leader {current_leader} killed -> {new_leader} elected at \
             term {new_term} in {elapsed:?}; {} acked writes (max seq {max_acked_seq})",
            ledger.len()
        );

        // ── INVARIANT A (frontier): the elected leader's ELECTION-TIME
        // position — in the killed leader's stream numbering, the only
        // numbering the acked seqs live in — covers every acknowledged
        // write (the vote restriction guarantees it). The leader's own
        // `last_seq` is its NEW stream's numbering and is NOT comparable.
        let status = cluster.local_status(new_leader).expect("leader status");
        let prev_term = status["prev_log_term"].as_u64().unwrap();
        let prev_seq = status["prev_log_seq"].as_u64().unwrap();
        assert_eq!(
            prev_term, last_term,
            "round {round}: the elected leader's election-time tail term must be \
             the killed leader's term (same stream numbering as the acked seqs)"
        );
        assert!(
            prev_seq >= max_acked_seq,
            "round {round}: elected leader's election-time frontier {prev_seq} is \
             below an acknowledged seq {max_acked_seq} — acked-write loss"
        );

        // ── INVARIANT B (content): every acked item is searchable on the
        // new leader (the text index auto-commits within ~2s; allow 10).
        let search_deadline = Instant::now() + Duration::from_secs(10);
        for &(entity, _) in &ledger {
            let token = item_token(entity);
            loop {
                let found: serde_json::Value = client
                    .get(format!(
                        "{}/search?query={token}&limit=5",
                        cluster.node(new_leader)
                    ))
                    .send()
                    .unwrap()
                    .json()
                    .unwrap();
                let hit = found["items"]
                    .as_array()
                    .is_some_and(|r| r.iter().any(|x| x["entity_id"].as_u64() == Some(entity)));
                if hit {
                    break;
                }
                assert!(
                    Instant::now() < search_deadline,
                    "round {round}: acked item {entity} (token {token}) not found on \
                     the elected leader — acked-write loss"
                );
                std::thread::sleep(Duration::from_millis(200));
            }
        }

        // ── Bring the killed node back for the next round. A divergent
        // suffix (leader-staged, never quorum-acked writes) legitimately
        // quarantines — the documented recovery is a reseed.
        cluster.restart(current_leader, &[]);
        let rejoin_deadline = Instant::now() + convergence_budget() + Duration::from_secs(10);
        loop {
            let status = cluster.local_status(current_leader);
            let quarantined = status
                .as_ref()
                .and_then(|s| s["quarantined"].as_bool())
                .unwrap_or(false);
            if quarantined {
                println!(
                    "round {round}: restarted node {current_leader} quarantined \
                     (divergent suffix) — reseeding, the documented recovery"
                );
                cluster.kill_hard(current_leader);
                cluster.wipe_data_dir(current_leader);
                cluster.restart(current_leader, &[]);
            }
            let caught_up = cluster.local_status(current_leader).is_some_and(|s| {
                s["term"].as_u64().unwrap_or(0) >= new_term
                    && s["role"].as_str() == Some("follower")
                    && s["lag_events"].as_u64() == Some(0)
            });
            if caught_up {
                break;
            }
            assert!(
                Instant::now() < rejoin_deadline,
                "round {round}: killed node {current_leader} did not rejoin/converge"
            );
            std::thread::sleep(Duration::from_millis(200));
        }

        current_leader = new_leader;
        last_term = new_term;
        entity_cursor += 100_000;
    }
}

/// Gate 2: partition the leader away with real TCP severs, let the survivors
/// elect, RESTART the old leader while still partitioned — it must boot as a
/// follower (durable election state, never the topology file) and cannot
/// accept a single write; on heal it rejoins the new term as a follower.
#[test]
fn mp_fenced_ex_leader_restart_cannot_write() {
    let (rewrite, proxies) = proxied_rewrite(&ALL_REGIONS);
    let mut opts = ClusterOptions::new(3)
        .with_topology_extra(FAST_ELECTION_YAML)
        .with_rewrite(rewrite);
    opts.log = "info".into();
    let mut cluster = MultiProcCluster::start_with(opts);
    let client = reqwest::blocking::Client::builder()
        .timeout(Duration::from_secs(2))
        .build()
        .unwrap();

    // Baseline data, fully converged BEFORE the partition so the old leader
    // carries no divergent suffix (this gate is about fencing, not reseed).
    for entity in 1..=5u64 {
        post_acked(
            &client,
            &cluster.node(0),
            "/signals",
            "quorum",
            &serde_json::json!({ "entity_id": entity, "signal": "view", "weight": 1.0 }),
        )
        .expect("baseline quorum write");
    }
    cluster.wait_converged_all(convergence_budget());

    // ── Partition the leader away (BOTH directions); the survivors elect. ──
    isolate(&proxies, "us-east", &["eu-west", "ap-south"]);
    let (new_leader, new_term, elapsed) =
        await_elected_leader(&cluster, &[1, 2], 0, FAILOVER_BUDGET);
    println!("survivors elected node {new_leader} at term {new_term} in {elapsed:?}");
    // The new leadership serves quorum writes (2 of 3 replicas).
    let _ = await_write_resumes(
        &client,
        &cluster.node(new_leader),
        700_001,
        Instant::now() + FAILOVER_BUDGET,
    );

    // ── Restart the old leader while STILL partitioned. ────────────────────
    cluster.kill_hard(0);
    cluster.restart(0, &[]);

    // The §1.4-1 assertion: across a multi-second window, the restarted
    // ex-leader REFUSES every write — its durable state boots it as a
    // follower and the topology file's `leader: us-east` is dead weight.
    let fence_window = Instant::now() + Duration::from_secs(3);
    let mut attempts = 0u32;
    while Instant::now() < fence_window {
        let accepted = post_acked(
            &client,
            &cluster.node(0),
            "/signals",
            "leader",
            &serde_json::json!({ "entity_id": 700_100, "signal": "view", "weight": 1.0 }),
        );
        assert!(
            accepted.is_none(),
            "the restarted, partitioned ex-leader ACCEPTED a write (seq {accepted:?}) — \
             the §1.4-1 split-brain hole is open"
        );
        attempts += 1;
        std::thread::sleep(Duration::from_millis(100));
    }
    let status = cluster.local_status(0).expect("ex-leader serves status");
    assert_eq!(
        status["is_leader"].as_bool(),
        Some(false),
        "restarted ex-leader must not claim leadership: {status}"
    );
    println!("fencing held across {attempts} write attempts: {status}");

    // ── Heal: the ex-leader joins the new term as a follower and converges.
    rejoin(&proxies, "us-east", &["eu-west", "ap-south"]);
    let rejoin_deadline = Instant::now() + convergence_budget() + Duration::from_secs(30);
    loop {
        let status = cluster.local_status(0).expect("status");
        assert_ne!(
            status["quarantined"].as_bool(),
            Some(true),
            "a fully-converged-then-partitioned ex-leader must rejoin CLEAN \
             (no divergent suffix existed): {status}"
        );
        if status["term"].as_u64().unwrap_or(0) >= new_term
            && status["role"].as_str() == Some("follower")
            && status["lag_events"].as_u64() == Some(0)
        {
            break;
        }
        assert!(
            Instant::now() < rejoin_deadline,
            "healed ex-leader did not rejoin term {new_term} and converge: {status}"
        );
        std::thread::sleep(Duration::from_millis(200));
    }
}

/// Gate 3: flapping links produce BOUNDED churn. Short flaps (below the
/// election timeout) are absorbed by the pre-vote/lease machinery; long
/// flaps elect; terms never explode; the cluster converges to exactly one
/// leader that serves quorum writes.
#[test]
fn mp_flapping_links_bounded_churn() {
    let (rewrite, proxies) = proxied_rewrite(&ALL_REGIONS);
    let mut cluster = MultiProcCluster::start_with(
        ClusterOptions::new(3)
            .with_topology_extra(FAST_ELECTION_YAML)
            .with_rewrite(rewrite),
    );
    let _ = &mut cluster;
    let client = reqwest::blocking::Client::builder()
        .timeout(Duration::from_secs(2))
        .build()
        .unwrap();

    // Short flaps: sever 250ms (below the 500ms timeout floor), heal 400ms.
    // The lease refusals + timer resets must absorb these without elections.
    for _ in 0..4 {
        isolate(&proxies, "us-east", &["eu-west", "ap-south"]);
        std::thread::sleep(Duration::from_millis(250));
        rejoin(&proxies, "us-east", &["eu-west", "ap-south"]);
        std::thread::sleep(Duration::from_millis(400));
    }
    // Long flaps: sever past the timeout so real elections happen, then heal.
    for _ in 0..3 {
        isolate(&proxies, "us-east", &["eu-west", "ap-south"]);
        std::thread::sleep(Duration::from_millis(1_500));
        rejoin(&proxies, "us-east", &["eu-west", "ap-south"]);
        std::thread::sleep(Duration::from_millis(800));
    }

    // Convergence: exactly one leader, agreed term, bounded churn.
    let deadline = Instant::now() + Duration::from_secs(15);
    let (leaders, term) = loop {
        let statuses: Vec<serde_json::Value> =
            (0..3).filter_map(|i| cluster.local_status(i)).collect();
        let leaders: Vec<usize> = statuses
            .iter()
            .enumerate()
            .filter(|(_, s)| s["role"].as_str() == Some("leader"))
            .map(|(i, _)| i)
            .collect();
        let terms: Vec<u64> = statuses
            .iter()
            .map(|s| s["term"].as_u64().unwrap_or(0))
            .collect();
        let agreed = terms.iter().max() == terms.iter().min();
        if statuses.len() == 3 && leaders.len() == 1 && agreed {
            break (leaders, terms[0]);
        }
        assert!(
            Instant::now() < deadline,
            "cluster did not converge to one leader on one term: \
             leaders={leaders:?} terms={terms:?}"
        );
        std::thread::sleep(Duration::from_millis(100));
    };
    // Bounded churn: 3 long flaps + slop can justify a handful of terms,
    // never dozens (a term explosion = pre-vote regression / livelock).
    assert!(
        term <= 15,
        "term {term} after 7 flaps — election churn is unbounded"
    );
    println!("converged: leader node {} at term {term}", leaders[0]);

    // The survivor of all that chaos still serves quorum writes.
    let _ = await_write_resumes(
        &client,
        &cluster.node(leaders[0]),
        800_001,
        Instant::now() + FAILOVER_BUDGET,
    );
}