//! Tier-3 quorum-ack suite (m11p3, REAL multi-process cluster).
//!
//! Two pillars:
//!
//! 1. **Quorum semantics under partition** — `ack=quorum` writes succeed
//!    while a majority is reachable, fail fast (retryable 503 naming the
//!    laggards) when it is not, never disturb `ack=leader` traffic, and
//!    recover after heal.
//! 2. **The ledger checker (the m11p3 exit gate)** — SIGKILL the leader
//!    under concurrent `ack=quorum` load, across many distinct kill points,
//!    and prove ZERO acknowledged-write loss: every write the client saw a
//!    2xx + `x-tidal-seq` for is present on the promoted survivor.
//!
//!    The proof is two-layered per kill point:
//!    - **Frontier**: a quorum ack for seqno S means some follower's
//!      CONTIGUOUS applied frontier reached S (durably — m11p3 acks are
//!      post-apply). The operator rule "promote the max-applied survivor"
//!      therefore guarantees the promoted node holds EVERY acked seqno:
//!      `max(acked seq) <= max(survivor applied_events)` is asserted before
//!      the promote.
//!    - **Content**: every acked ITEM is found via `/search` on the new
//!      leader (the frontier can't lie about data it doesn't have, but this
//!      catches a frontier that lies about data it has).
//!
//!    Kill-point count: `TIDAL_QUORUM_KILLPOINTS` (default 8 for CI; the
//!    exit-gate run is 100 — see docs/planning/milestone-11/phase-3.md for
//!    the recorded run).
//!
//! Run: `cargo test -p tidal-server --features cluster-e2e --test cluster_quorum -- --nocapture`

#![cfg(feature = "cluster-e2e")]
#![allow(
    clippy::unwrap_used,
    clippy::expect_used,
    clippy::panic,
    clippy::cast_possible_truncation,
    clippy::cast_precision_loss,
    clippy::too_many_lines
)]

mod support;

/// This suite validates the m11p3 quorum mechanics and the MANUAL failover
/// drill (operator promote of the max-applied survivor). Auto-election is
/// pinned OFF so the m11p4 failure detector cannot race the drill — the
/// automatic path has its own exit-gate suite (`cluster_election.rs`).
const LEGACY_ELECTION_YAML: &str = "election:\n  auto_election: false";

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

use support::{
    invariants::{AckLedger, post_acked},
    multiproc::{BREAKER_RESET, ClusterOptions, MultiProcCluster, convergence_budget},
    partition::proxied_rewrite,
};

const LEADER: usize = 0;

/// CI-default kill points; the exit-gate run sets `TIDAL_QUORUM_KILLPOINTS=100`.
fn killpoints() -> usize {
    std::env::var("TIDAL_QUORUM_KILLPOINTS")
        .ok()
        .and_then(|v| v.parse().ok())
        .filter(|&n| n > 0)
        .unwrap_or(8)
}

/// m11p3 quorum semantics over a REAL 3-process cluster with real TCP
/// partitions:
///
/// - healthy: `ack=quorum` 204/201 + seq header; the leader's `commit_index`
///   tracks the writes;
/// - ONE follower severed: quorum (2 of 3) still commits via the other;
/// - BOTH followers severed: quorum 503s fast naming the laggards while
///   `ack=leader` writes keep succeeding (the knob's cost is the caller's
///   choice, never the deployment's);
/// - healed: quorum commits again.
#[test]
fn mp_quorum_writes_gate_and_recover_under_partition() {
    let (rewrite, proxies) = proxied_rewrite(&["eu-west", "ap-south"]);
    let cluster = MultiProcCluster::start_with(
        ClusterOptions::new(3)
            .with_topology_extra(LEGACY_ELECTION_YAML)
            .with_rewrite(rewrite),
    );
    let client = reqwest::blocking::Client::builder()
        .timeout(Duration::from_secs(8))
        .build()
        .unwrap();
    let leader_base = cluster.node(LEADER);

    // ── Healthy majority: quorum writes commit ──────────────────────────────
    let seq = post_acked(
        &client,
        &leader_base,
        "/items",
        "quorum",
        &serde_json::json!({ "entity_id": 1, "metadata": { "title": "quorum one" } }),
    )
    .expect("healthy-cluster quorum item write must ack");
    let view_seq = post_acked(
        &client,
        &leader_base,
        "/signals",
        "quorum",
        &serde_json::json!({ "entity_id": 1, "signal": "view", "weight": 1.0 }),
    )
    .expect("healthy-cluster quorum signal write must ack");
    assert!(view_seq > seq, "one log: signal follows the item");
    let status = cluster.local_status(LEADER).unwrap();
    assert!(
        status["commit_index"].as_u64().unwrap() >= view_seq,
        "a quorum ack is at or below the commit index: {status}"
    );
    println!("[quorum] healthy: item seq={seq}, view seq={view_seq} committed");

    // ── One follower down: 2-of-3 majority still commits ────────────────────
    proxies.region("ap-south").sever_grpc();
    let seq = post_acked(
        &client,
        &leader_base,
        "/signals",
        "quorum",
        &serde_json::json!({ "entity_id": 1, "signal": "view", "weight": 2.0 }),
    )
    .expect("quorum must survive a single-follower outage (majority intact)");
    println!("[quorum] ap-south severed: quorum still commits (seq={seq})");

    // ── Both followers down: quorum 503s naming the laggards ────────────────
    proxies.region("eu-west").sever_grpc();
    let resp = client
        .post(format!("{leader_base}/signals"))
        .header("x-tidal-ack", "quorum")
        .json(&serde_json::json!({ "entity_id": 1, "signal": "view", "weight": 3.0 }))
        .send()
        .unwrap();
    assert_eq!(
        resp.status().as_u16(),
        503,
        "no majority reachable: the quorum write must fail fast"
    );
    let body: serde_json::Value = resp.json().unwrap();
    assert_eq!(body["retryable"].as_bool(), Some(true));
    let laggards: Vec<&str> = body["laggards"]
        .as_array()
        .unwrap()
        .iter()
        .map(|v| v.as_str().unwrap())
        .collect();
    assert!(
        laggards.contains(&"eu-west") || laggards.contains(&"ap-south"),
        "the 503 names the lagging followers: {body}"
    );
    // The leader-ack contract is untouched by the followers' outage.
    let resp = client
        .post(format!("{leader_base}/signals"))
        .header("x-tidal-ack", "leader")
        .json(&serde_json::json!({ "entity_id": 1, "signal": "view", "weight": 4.0 }))
        .send()
        .unwrap();
    assert_eq!(
        resp.status().as_u16(),
        204,
        "ack=leader writes keep succeeding through a follower outage"
    );
    println!("[quorum] both severed: quorum 503 named {laggards:?}; leader-ack still 204");

    // ── Heal: quorum recovers (drive through the breaker window) ────────────
    proxies.region("eu-west").heal_all();
    proxies.region("ap-south").heal_all();
    for region in ["eu-west", "ap-south"] {
        let resp = client
            .post(format!("{leader_base}/cluster/heal"))
            .json(&serde_json::json!({ "region": region }))
            .send()
            .unwrap();
        assert_eq!(resp.status().as_u16(), 200);
    }
    let deadline = Instant::now() + BREAKER_RESET + convergence_budget();
    let mut healed_seq = None;
    while healed_seq.is_none() {
        assert!(
            Instant::now() <= deadline,
            "healed quorum writes must commit within the breaker+convergence budget"
        );
        healed_seq = post_acked(
            &client,
            &leader_base,
            "/signals",
            "quorum",
            &serde_json::json!({ "entity_id": 1, "signal": "view", "weight": 5.0 }),
        );
        if healed_seq.is_none() {
            // Re-issue the heal: the breaker can swallow the first post-heal
            // ships (the documented runbook loop).
            for region in ["eu-west", "ap-south"] {
                let _ = client
                    .post(format!("{leader_base}/cluster/heal"))
                    .json(&serde_json::json!({ "region": region }))
                    .send();
            }
            std::thread::sleep(Duration::from_millis(250));
        }
    }
    println!(
        "[quorum] healed: quorum commits again (seq={})",
        healed_seq.unwrap()
    );
}

/// THE m11p3 EXIT GATE: SIGKILL the leader under `ack=quorum` load at many
/// distinct kill points; the ledger checker proves zero acknowledged loss on
/// the promoted (max-applied) survivor every time. See the module docs for
/// the two-layer proof.
#[test]
fn mp_quorum_ledger_zero_acked_loss_across_killpoints() {
    let rounds = killpoints();
    println!("[ledger] running {rounds} leader-kill points (TIDAL_QUORUM_KILLPOINTS to widen)");

    for round in 0..rounds {
        let mut opts = ClusterOptions::new(3).with_topology_extra(LEGACY_ELECTION_YAML);
        opts.log = "info".into();
        let mut cluster = MultiProcCluster::start_with(opts);
        let leader_base = cluster.node(LEADER);
        let gateway_bases = [
            cluster.node(LEADER),
            cluster.node(1), // forwarded quorum writes through a follower
        ];

        // ── Concurrent quorum writers; each builds an AckLedger of the writes
        // the client saw acknowledged (the shared first-class checker). ──────
        let stop = Arc::new(AtomicBool::new(false));
        let id_base = 1_000 * (round as u64 + 1);
        let mut writers = Vec::new();
        for (w, base) in gateway_bases.iter().enumerate() {
            let base = base.clone();
            let stop = Arc::clone(&stop);
            writers.push(std::thread::spawn(move || {
                let client = reqwest::blocking::Client::builder()
                    .timeout(Duration::from_secs(4))
                    .build()
                    .unwrap();
                let mut ledger = AckLedger::new();
                let mut n = 0u64;
                while !stop.load(Ordering::Acquire) {
                    let entity_id = id_base + (w as u64) * 500 + n;
                    n += 1;
                    // An un-acked item (leader dying/dead, forward failed, or
                    // quorum timeout) is owed nothing by contract — skip it.
                    let _ = ledger.write_item_and_view(&client, &base, "quorum", entity_id);
                }
                ledger.writes().to_vec()
            }));
        }

        // Pseudo-random kill point: spread across boot-warm, mid-burst, and
        // saturated states deterministically (reproducible per round).
        let kill_after = Duration::from_millis(120 + (round as u64 * 97) % 480);
        std::thread::sleep(kill_after);
        cluster.kill_hard(LEADER);
        stop.store(true, Ordering::Release);
        let _ = client_drain(&leader_base); // flush any half-open socket
        let mut ledger = AckLedger::new();
        for w in writers {
            ledger.extend(w.join().expect("writer thread").into_iter());
        }
        let max_acked_seq = ledger.max_acked_seq();

        // ── INVARIANT A (frontier): no acked seqno above the max-applied
        // survivor's contiguous durable frontier — the shared checker also
        // returns the survivor the operator rule says to promote. ───────────
        let ctx = format!("round {round}");
        let (chosen, chosen_applied) =
            ledger.assert_frontier_covers_acks(&cluster, &[1usize, 2usize], &ctx);

        // m11p4: promote is a FENCED transfer — the election's up-to-date
        // restriction can refuse a target that fell behind between this
        // test's status sample and the vote (in-flight ships keep applying
        // for a moment after the kill). The operator drill is to promote the
        // OTHER survivor in that case; the zero-acked-loss invariants hold
        // for whichever node the election admits.
        let mut new_leader = cluster.region_name(chosen).to_string();
        let resp = cluster.post(
            chosen,
            "/cluster/promote",
            &serde_json::json!({ "region": new_leader }),
        );
        if resp.status().as_u16() != 200 {
            let other = [1usize, 2usize]
                .into_iter()
                .find(|&idx| idx != chosen)
                .expect("two survivors");
            println!(
                "[ledger] round {round}: promote of {new_leader} refused (it fell \
                 behind the other survivor); promoting the other"
            );
            new_leader = cluster.region_name(other).to_string();
            let retry = cluster.post(
                other,
                "/cluster/promote",
                &serde_json::json!({ "region": new_leader }),
            );
            assert_eq!(retry.status().as_u16(), 200, "round {round}: promote retry");
        }
        cluster.wait_leader_agreed(&new_leader, Duration::from_secs(10));

        // ── INVARIANT B (content): every acked item is present on the new
        // leader (the shared checker polls past the text index's 2s commit). ─
        let winner_idx = (0..3)
            .find(|&i| cluster.region_name(i) == new_leader)
            .expect("winner index");
        ledger.assert_items_present(&cluster, winner_idx, Duration::from_secs(10), &ctx);
        println!(
            "[ledger] round {round}: kill@{kill_after:?} → {} acked writes (max seq \
             {max_acked_seq}) all present on {new_leader} (applied {chosen_applied})",
            ledger.len()
        );
        drop(cluster);
    }
}

/// Issue one throwaway GET so a dead leader's half-open client sockets are
/// observed closed before the ledger math (keeps the teardown deterministic
/// on macOS, where a killed process's sockets can linger in the client pool).
fn client_drain(base: &str) -> Option<()> {
    let client = reqwest::blocking::Client::builder()
        .timeout(Duration::from_millis(300))
        .build()
        .ok()?;
    let _ = client.get(format!("{base}/health/live")).send();
    Some(())
}