//! Tier-3 MULTI-PROCESS cluster UAT (m8p10 task 04).
//!
//! Each test spins a [`MultiProcCluster`]: ONE `tidal-server cluster --region`
//! OS process per region, peering over real gRPC and forwarding over real HTTP.
//! Unlike `cluster_e2e.rs` (every region in one process), convergence here is
//! verified against EVERY follower PROCESS's own `/cluster/status/local`, so a
//! passing assertion proves a write crossed the process boundary.
//!
//! Covered UAT steps (the rest land in tasks 05/06):
//!
//! * step 1 — cross-process replication, feed parity to 1e-6, per-signal
//!   write→applied p99 < 2s;
//! * step 2 — SIGKILL the leader, promote a survivor, forwarded write to the
//!   other survivor, zero data loss, failover < 10s;
//! * step 5 — region-routing flip with zero read downtime.
//!
//! Plus write-forwarding and aggregated-status proofs over real processes.
//!
//! ```bash
//! cargo test -p tidal-server --features cluster-e2e --test cluster_multiproc -- --nocapture
//! ```
#![cfg(feature = "cluster-e2e")]
// Tier-3 harness allows, mirroring `cluster_routes.rs`: `unwrap` on known-good
// fixtures is idiomatic test noise; the lossy numeric casts are the same
// pervasive-and-intentional scoring/percentile math the crate config documents;
// `items_after_statements` is for the per-test `const` budgets declared next to
// the code they bound (clearer than hoisting them above the harness setup).
#![allow(
    clippy::unwrap_used,
    clippy::missing_panics_doc,
    clippy::too_many_lines,
    clippy::cast_precision_loss,
    clippy::cast_possible_truncation,
    clippy::cast_sign_loss,
    clippy::items_after_statements
)]

mod support;

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

use support::multiproc::{MultiProcCluster, convergence_budget, seed_items_and_embeddings};

/// The leader index in every test (region 0 = `us-east`).
const LEADER: usize = 0;

/// Fetch a node's local-region feed as a sorted `(entity_id, score)` vector.
fn feed_pairs(
    cluster: &MultiProcCluster,
    idx: usize,
    profile: &str,
    limit: u32,
) -> Vec<(u64, f64)> {
    let body = cluster.get_json(idx, &format!("/feed?profile={profile}&limit={limit}"));
    let mut pairs: Vec<(u64, f64)> = body["items"]
        .as_array()
        .unwrap_or(&Vec::new())
        .iter()
        .map(|it| {
            (
                it["entity_id"].as_u64().unwrap(),
                it["score"].as_f64().unwrap(),
            )
        })
        .collect();
    pairs.sort_by_key(|(id, _)| *id);
    pairs
}

/// Assert two feed views carry the SAME items with scores equal to 1e-6.
fn assert_feed_parity(label: &str, a: &[(u64, f64)], b: &[(u64, f64)]) {
    assert_eq!(
        a.iter().map(|(id, _)| *id).collect::<Vec<_>>(),
        b.iter().map(|(id, _)| *id).collect::<Vec<_>>(),
        "{label}: feed item sets differ"
    );
    for ((id_a, score_a), (_, score_b)) in a.iter().zip(b.iter()) {
        assert!(
            (score_a - score_b).abs() <= 1e-6,
            "{label}: score for item {id_a} differs: {score_a} vs {score_b}"
        );
    }
}

/// The p99 of a latency sample set (nearest-rank, sorted ascending).
fn p99(samples: &mut [Duration]) -> Duration {
    assert!(!samples.is_empty(), "p99 of empty sample set");
    samples.sort_unstable();
    // Nearest-rank: ceil(0.99 * n) - 1, clamped into range.
    let idx = (((samples.len() as f64) * 0.99).ceil() as usize)
        .saturating_sub(1)
        .min(samples.len() - 1);
    samples[idx]
}

// ── UAT step 1: replication under 2s + feed parity ──────────────────────────────

/// 3 processes; seed items+embeddings on the leader (broadcast), write 100 `view`
/// signals to the leader, and measure per-signal write→follower-applied latency by
/// polling EACH follower's own `/cluster/status/local`. Assert: both followers
/// converge, p99 of the per-signal applied latency < 2s, and each follower's local
/// feed matches the leader's feed to 1e-6 (cross-process replication is correct,
/// not just present).
#[test]
fn mp_uat_step1_replication_under_2s() {
    let cluster = Arc::new(MultiProcCluster::start(3));
    let followers = [1usize, 2usize];

    const ITEMS: u64 = 20;
    const SIGNALS: u64 = 100;

    seed_items_and_embeddings(&cluster, LEADER, ITEMS);

    // Per-signal write instants, published by the writer AS each write returns so
    // the concurrent pollers can attribute an applied-observation to the right
    // write the moment it lands (not after the whole batch — that conflation is
    // what inflates the earliest signals' apparent latency).
    let write_at: Arc<Mutex<Vec<Instant>>> =
        Arc::new(Mutex::new(Vec::with_capacity(SIGNALS as usize)));
    let writer_done = Arc::new(AtomicBool::new(false));

    // Spawn ONE poller per follower BEFORE writing. Each records, per signal index
    // k (1-based), the first instant it observed `applied >= k` on its follower.
    let pollers: Vec<_> = followers
        .iter()
        .map(|&f| {
            let cluster = Arc::clone(&cluster);
            let writer_done = Arc::clone(&writer_done);
            thread::spawn(move || -> Vec<Instant> {
                // observed_at[k-1] = first instant follower reported applied >= k.
                let mut observed_at: Vec<Option<Instant>> = vec![None; SIGNALS as usize];
                let mut max_seen: u64 = 0;
                let deadline = Instant::now() + convergence_budget();
                loop {
                    if let Some(st) = cluster.local_status(f) {
                        let applied = st["applied_events"].as_u64().unwrap_or(0).min(SIGNALS);
                        let now = Instant::now();
                        while max_seen < applied {
                            observed_at[max_seen as usize] = Some(now);
                            max_seen += 1;
                        }
                        if max_seen >= SIGNALS && st["lag_events"].as_u64().unwrap_or(u64::MAX) == 0
                        {
                            break;
                        }
                    }
                    assert!(
                        Instant::now() <= deadline,
                        "follower {f} did not apply all {SIGNALS} signals: applied={max_seen}"
                    );
                    // Tight poll so the observed-applied instant tracks reality; the
                    // writer-done flag lets us avoid spinning after writes complete.
                    if writer_done.load(Ordering::Relaxed) && max_seen >= SIGNALS {
                        break;
                    }
                    thread::sleep(Duration::from_millis(2));
                }
                observed_at
                    .into_iter()
                    .map(|o| o.expect("every signal observed applied"))
                    .collect()
            })
        })
        .collect();

    // Write SIGNALS view signals to the leader; publish each write instant. The
    // signal entity cycles over the seeded items so every item accrues weight.
    for n in 1..=SIGNALS {
        let entity_id = ((n - 1) % ITEMS) + 1;
        let t = Instant::now();
        let resp = cluster.post(
            LEADER,
            "/signals",
            &serde_json::json!({ "entity_id": entity_id, "signal": "view", "weight": 1.0 }),
        );
        assert_eq!(
            resp.status().as_u16(),
            204,
            "leader /signals must 204: {}",
            resp.status()
        );
        write_at.lock().unwrap().push(t);
    }
    writer_done.store(true, Ordering::Relaxed);

    // Join the pollers and compute per-signal write->applied latency. Each
    // follower contributes SIGNALS samples.
    let write_at = write_at.lock().unwrap().clone();
    assert_eq!(write_at.len(), SIGNALS as usize);
    let mut latencies: Vec<Duration> = Vec::with_capacity((SIGNALS as usize) * followers.len());
    for poller in pollers {
        let observed = poller.join().expect("poller thread");
        assert_eq!(observed.len(), SIGNALS as usize);
        for (k, applied_at) in observed.iter().enumerate() {
            latencies.push(applied_at.saturating_duration_since(write_at[k]));
        }
    }

    // Belt-and-suspenders: the harness-level converged check against EVERY follower.
    cluster.wait_converged_all(convergence_budget());

    let p99_latency = p99(&mut latencies);
    let max_latency = latencies.iter().copied().max().unwrap_or_default();
    assert!(
        p99_latency < Duration::from_secs(2),
        "replication p99 was {p99_latency:?} (max {max_latency:?}); SLA is < 2s"
    );
    println!(
        "[step1] {} signals x {} followers: p99 write->applied = {:?}, max = {:?}",
        SIGNALS,
        followers.len(),
        p99_latency,
        max_latency
    );

    // Feed parity: each follower's LOCAL feed equals the leader's feed to 1e-6.
    let leader_feed = feed_pairs(&cluster, LEADER, "trending", ITEMS as u32);
    assert!(
        !leader_feed.is_empty(),
        "leader feed must rank seeded items"
    );
    for &f in &followers {
        let follower_feed = feed_pairs(&cluster, f, "trending", ITEMS as u32);
        assert_feed_parity(
            &format!("leader vs {}", cluster.region_name(f)),
            &leader_feed,
            &follower_feed,
        );
    }
    println!(
        "[step1] feed parity verified: {} items match leader<->follower scores to 1e-6",
        leader_feed.len()
    );
}

// ── UAT step 2: leader crash + failover under 10s ───────────────────────────────

/// 3 processes; seed + converge; SIGKILL the leader; promote `eu-west` via the
/// OTHER survivor (`ap-south`); wait for both survivors to agree on the new leader;
/// POST a signal to the non-leader survivor (proving it forwards to the NEW leader,
/// 204). Assert: pre-crash data still served on both survivors (no data loss),
/// promote→first-successful-write elapsed < 10s, and post-failover feed parity.
#[test]
fn mp_uat_step2_leader_crash_failover_under_10s() {
    let mut cluster = MultiProcCluster::start(3);
    let eu_west = 1usize;
    let ap_south = 2usize;

    const ITEMS: u64 = 12;

    // Seed and converge before the crash.
    seed_items_and_embeddings(&cluster, LEADER, ITEMS);
    for entity_id in 1..=ITEMS {
        let resp = cluster.post(
            LEADER,
            "/signals",
            &serde_json::json!({ "entity_id": entity_id, "signal": "view", "weight": entity_id as f64 }),
        );
        assert_eq!(resp.status().as_u16(), 204);
    }
    cluster.wait_converged_all(convergence_budget());

    // Pre-crash feed snapshot on a survivor (eu-west), to prove no data loss after.
    let pre_crash_feed = feed_pairs(&cluster, eu_west, "trending", ITEMS as u32);
    assert!(
        !pre_crash_feed.is_empty(),
        "pre-crash survivor feed must rank seeded items"
    );

    // CRASH the leader (real SIGKILL — no graceful drain).
    cluster.kill_hard(LEADER);

    // Promote eu-west by calling promote ON ap-south (the OTHER survivor). Time
    // the failover window from this instant to the first successful write.
    let failover_start = Instant::now();
    let new_leader = cluster.region_name(eu_west).to_string();
    let resp = cluster.post(
        ap_south,
        "/cluster/promote",
        &serde_json::json!({ "region": new_leader }),
    );
    assert_eq!(
        resp.status().as_u16(),
        200,
        "promote on survivor must 200: {}",
        resp.status()
    );
    let body: serde_json::Value = resp.json().unwrap();
    assert_eq!(body["leader"].as_str(), Some(new_leader.as_str()));
    // The dead old leader is EXPECTED to land in `failed` (it cannot ack the fan-out).
    println!(
        "[step2] promote fan-out: acked={:?} failed={:?}",
        body["acked"], body["failed"]
    );

    // Both LIVE survivors must agree eu-west now leads.
    cluster.wait_leader_agreed(&new_leader, Duration::from_secs(10));

    // POST a signal to the NON-leader survivor (ap-south) — it must forward to the
    // NEW leader (eu-west) transparently and 204. Retry within the failover budget
    // so a just-promoted leader that has not finished settling does not flake.
    let write_deadline = failover_start + Duration::from_secs(10);
    let failover_elapsed = loop {
        let resp = cluster.post(
            ap_south,
            "/signals",
            &serde_json::json!({ "entity_id": 1, "signal": "like", "weight": 3.0 }),
        );
        if resp.status().as_u16() == 204 {
            break failover_start.elapsed();
        }
        assert!(
            Instant::now() <= write_deadline,
            "no successful write to the new leader within 10s of failover: last status {}",
            resp.status()
        );
        thread::sleep(Duration::from_millis(50));
    };
    assert!(
        failover_elapsed < Duration::from_secs(10),
        "failover (promote -> first successful write) was {failover_elapsed:?}; SLA is < 10s"
    );
    println!("[step2] failover (promote -> first write) = {failover_elapsed:?}");

    // No data loss: both survivors still serve the pre-crash items, and their
    // feeds remain in parity with the pre-crash snapshot's item set.
    for &survivor in &[eu_west, ap_south] {
        let feed = feed_pairs(&cluster, survivor, "trending", ITEMS as u32);
        let item_set: Vec<u64> = feed.iter().map(|(id, _)| *id).collect();
        let pre_set: Vec<u64> = pre_crash_feed.iter().map(|(id, _)| *id).collect();
        assert_eq!(
            item_set,
            pre_set,
            "{}: post-failover feed lost pre-crash items",
            cluster.region_name(survivor)
        );
    }
    // Survivor feeds agree with each other (replicated state, identical ranking).
    let eu_feed = feed_pairs(&cluster, eu_west, "trending", ITEMS as u32);
    let ap_feed = feed_pairs(&cluster, ap_south, "trending", ITEMS as u32);
    assert_feed_parity("eu-west vs ap-south (post-failover)", &eu_feed, &ap_feed);
    println!(
        "[step2] no data loss: {} pre-crash items still served on both survivors",
        eu_feed.len()
    );
}

// ── UAT step 5: tenant region-routing flip with zero read downtime ──────────────

/// 3 processes; seed + converge so the data is present in every region. A
/// background thread hammers `GET /feed?region=us-east` (forwarded to us-east's
/// process) through a stable gateway node. We then flip the pin to
/// `?region=eu-west` (routing config only — replication already placed the data).
/// Assert: ZERO failed reads across the flip window, and the new region returns
/// the identical item set.
#[test]
fn mp_uat_step5_tenant_routing_flip() {
    let cluster = Arc::new(MultiProcCluster::start(3));
    let gateway = 2usize; // ap-south as the stable tenant gateway
    let us_east = cluster.region_name(0).to_string();
    let eu_west = cluster.region_name(1).to_string();

    const ITEMS: u64 = 10;
    seed_items_and_embeddings(&cluster, LEADER, ITEMS);
    for entity_id in 1..=ITEMS {
        let resp = cluster.post(
            LEADER,
            "/signals",
            &serde_json::json!({ "entity_id": entity_id, "signal": "view", "weight": entity_id as f64 }),
        );
        assert_eq!(resp.status().as_u16(), 204);
    }
    cluster.wait_converged_all(convergence_budget());

    // The region currently pinned for tenant reads; flipped mid-flight.
    let pinned: Arc<std::sync::RwLock<String>> = Arc::new(std::sync::RwLock::new(us_east.clone()));
    let stop = Arc::new(AtomicBool::new(false));
    let failures = Arc::new(AtomicU64::new(0));
    let reads = Arc::new(AtomicU64::new(0));

    let reader_cluster = Arc::clone(&cluster);
    let reader_pinned = Arc::clone(&pinned);
    let reader_stop = Arc::clone(&stop);
    let reader_failures = Arc::clone(&failures);
    let reader_reads = Arc::clone(&reads);
    let reader = thread::spawn(move || {
        let client = reader_cluster.client();
        while !reader_stop.load(Ordering::Relaxed) {
            let region = reader_pinned.read().unwrap().clone();
            let url = format!(
                "{}/feed?profile=trending&limit={ITEMS}&region={region}",
                reader_cluster.node(gateway)
            );
            match client.get(&url).timeout(Duration::from_secs(3)).send() {
                Ok(resp) if resp.status().is_success() => {
                    // A forwarded read must return a non-empty ranked feed.
                    let body: serde_json::Value = resp.json().unwrap_or(serde_json::Value::Null);
                    if body["items"].as_array().is_none_or(Vec::is_empty) {
                        reader_failures.fetch_add(1, Ordering::Relaxed);
                    }
                }
                _ => {
                    reader_failures.fetch_add(1, Ordering::Relaxed);
                }
            }
            reader_reads.fetch_add(1, Ordering::Relaxed);
            thread::sleep(Duration::from_millis(10));
        }
    });

    // Let the loop establish a baseline against us-east, then FLIP the pin to
    // eu-west (config-only — no data movement) and let it run against the new
    // region. Bounded settle windows; the failure count is the correctness gate.
    let baseline_reads = reads.load(Ordering::Relaxed);
    let flip_deadline = Instant::now() + Duration::from_secs(5);
    while reads.load(Ordering::Relaxed) < baseline_reads + 20 {
        assert!(
            Instant::now() <= flip_deadline,
            "reader loop did not progress"
        );
        thread::sleep(Duration::from_millis(10));
    }
    {
        let mut p = pinned.write().unwrap();
        *p = eu_west.clone();
    }
    let post_flip_target = reads.load(Ordering::Relaxed) + 20;
    let after_deadline = Instant::now() + Duration::from_secs(5);
    while reads.load(Ordering::Relaxed) < post_flip_target {
        assert!(
            Instant::now() <= after_deadline,
            "reader loop stalled after flip"
        );
        thread::sleep(Duration::from_millis(10));
    }

    stop.store(true, Ordering::Relaxed);
    reader.join().unwrap();

    let total_reads = reads.load(Ordering::Relaxed);
    let total_failures = failures.load(Ordering::Relaxed);
    assert_eq!(
        total_failures, 0,
        "tenant routing flip had {total_failures} failed reads across {total_reads} reads"
    );
    println!(
        "[step5] {total_reads} forwarded reads across the us-east -> eu-west flip, 0 failures"
    );

    // The new region returns the identical item set as the old (same replicated data).
    let us_feed: Vec<u64> = read_feed_items(&cluster, gateway, &us_east, ITEMS);
    let eu_feed: Vec<u64> = read_feed_items(&cluster, gateway, &eu_west, ITEMS);
    assert_eq!(
        us_feed, eu_feed,
        "post-flip region must serve the identical item set"
    );
    println!(
        "[step5] us-east and eu-west serve identical {} items",
        us_feed.len()
    );
}

/// Read a `?region=`-forwarded feed and return its item ids (sorted).
fn read_feed_items(
    cluster: &MultiProcCluster,
    gateway: usize,
    region: &str,
    limit: u64,
) -> Vec<u64> {
    let body = cluster.get_json(
        gateway,
        &format!("/feed?profile=trending&limit={limit}&region={region}"),
    );
    let mut ids: Vec<u64> = body["items"]
        .as_array()
        .unwrap_or(&Vec::new())
        .iter()
        .map(|it| it["entity_id"].as_u64().unwrap())
        .collect();
    ids.sort_unstable();
    ids
}

// ── Write forwarding over real processes ────────────────────────────────────────

/// A `POST /signals` to a FOLLOWER process must forward to the leader (204) and
/// replicate to EVERY follower process via the WAL relay.
#[test]
fn mp_write_forwarding() {
    let cluster = MultiProcCluster::start(3);
    let follower = 1usize;

    seed_items_and_embeddings(&cluster, LEADER, 6);

    // POST signals to the FOLLOWER — each must forward to the leader and 204.
    for entity_id in 1..=6u64 {
        let resp = cluster.post(
            follower,
            "/signals",
            &serde_json::json!({ "entity_id": entity_id, "signal": "view", "weight": 1.0 }),
        );
        assert_eq!(
            resp.status().as_u16(),
            204,
            "follower /signals must forward to leader and 204: {}",
            resp.status()
        );
    }

    // Converges on EVERY follower process.
    cluster.wait_converged_all(convergence_budget());

    // Every node ranks the replicated items.
    for idx in 0..cluster.len() {
        let body = cluster.get_json(idx, "/feed?profile=trending&limit=6");
        assert!(
            !body["items"].as_array().unwrap().is_empty(),
            "node {} must rank replicated items",
            cluster.region_name(idx)
        );
    }
    println!("[forwarding] follower-forwarded signals converged on all 3 processes");
}

// ── Aggregated status over real processes ───────────────────────────────────────

/// `GET /cluster/status` on EACH of the 3 processes reports the same leader, all 3
/// regions, and every region reachable.
#[test]
fn mp_status_aggregation() {
    let cluster = MultiProcCluster::start(3);
    let leader_name = cluster.region_name(LEADER).to_string();

    // Seed a few signals so the leader's relay seqno is non-zero, then converge.
    seed_items_and_embeddings(&cluster, LEADER, 3);
    for entity_id in 1..=3u64 {
        let resp = cluster.post(
            LEADER,
            "/signals",
            &serde_json::json!({ "entity_id": entity_id, "signal": "view", "weight": 1.0 }),
        );
        assert_eq!(resp.status().as_u16(), 204);
    }
    cluster.wait_converged_all(convergence_budget());

    for idx in 0..cluster.len() {
        // Aggregation polls peers with a tight budget; retry until every peer
        // is observed reachable (a just-converged peer may miss one poll window).
        let deadline = Instant::now() + Duration::from_secs(10);
        loop {
            let body = cluster.get_json(idx, "/cluster/status");
            let regions = body["regions"].as_array().cloned().unwrap_or_default();
            let leader_ok = body["leader"].as_str() == Some(leader_name.as_str());
            let three_regions = regions.len() == 3;
            let all_reachable = regions
                .iter()
                .all(|r| r["reachable"].as_bool() == Some(true));
            if leader_ok && three_regions && all_reachable {
                println!(
                    "[status] node {} aggregates leader={} regions=3 all-reachable",
                    cluster.region_name(idx),
                    leader_name
                );
                break;
            }
            assert!(
                Instant::now() <= deadline,
                "node {} aggregated status not fully reachable in time: {body}",
                cluster.region_name(idx)
            );
            thread::sleep(Duration::from_millis(100));
        }
    }
}