//! Tier-3 cross-shard unified reads suite (m12p4, the m11p6 L4 follow-up; REAL
//! multi-process cluster with PARTIAL shard placement).
//!
//! The m11p6 read fan-out (`scatter_merge`) is corpus-complete ONLY when a node
//! hosts a replica of EVERY shard group (full placement / `S=1`). Under PARTIAL
//! placement a node misses the groups it does not host, so a `/feed` on it would
//! be local-shard-only. m12p4 closes that: the gateway runs its LOCAL scatter,
//! then fans out to the groups it does NOT host (`forward_candidates`, internal
//! `?shard=g` hop) and merges. Two pillars:
//!
//! 1. **Cross-node fan-out makes a partial-node read corpus-complete** — items
//!    written across all groups are ALL returned by a `/feed` served from a node
//!    that hosts a strict subset of those groups (it reaches the missing groups
//!    over HTTP). The same `/feed` on a node hosting a different subset returns
//!    the SAME corpus — coverage is placement-independent.
//!
//! 2. **An unreachable missing group degrades, never hard-fails** — kill the
//!    sole node hosting a group, and a partial node's `/feed` still returns the
//!    items from the reachable groups (a partial, non-empty page) instead of a
//!    5xx — the honest-degraded contract.
//!
//! Run: `cargo test -p tidal-server --features cluster-e2e --test cluster_cross_shard_reads -- --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;

use std::collections::BTreeSet;
use std::time::Duration;

use support::multiproc::{MultiProcCluster, convergence_budget};
use tidaldb::{replication::shard::ShardRouter, schema::EntityId};

/// Fast election so a single-replica group self-elects promptly and a killed
/// node's failover (pillar 2's reachable groups) completes inside the budget.
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";

const NODES: usize = 3;
const SHARDS: usize = 3;
const ITEMS: u64 = 60;

/// Partial placement (m12p4): a "ring" where each group has RF=2 (so the engine
/// builds its WAL ship feed — a single-replica group has no peer and is rejected)
/// but each NODE hosts a STRICT SUBSET (2 of the 3 groups). Every node therefore
/// misses exactly one group and MUST cross-shard fan out to be corpus-complete.
///
/// * group 0: nodes [0, 1]  (leader 0)   — node 2 misses it
/// * group 1: nodes [1, 2]  (leader 1)   — node 0 misses it
/// * group 2: nodes [2, 0]  (leader 2)   — node 1 misses it
///
/// So node 0 hosts {0, 2} and misses group 1; node 1 hosts {0, 1} and misses
/// group 2; node 2 hosts {1, 2} and misses group 0.
fn partial_placement() -> Vec<Vec<usize>> {
    vec![vec![0, 1], vec![1, 2], vec![2, 0]]
}

/// Write `1..=ITEMS` items (each with a `view` signal so it ranks in `for_you`)
/// through `gateway`'s `/items` + `/signals` — the gateway hash-routes each to
/// its owning group's leader (forwarding when the gateway does not host it). All
/// items land on SOME group; the set spans all `SHARDS` groups by the router hash.
fn seed_corpus(cluster: &MultiProcCluster, gateway: usize) -> BTreeSet<u64> {
    let mut written = BTreeSet::new();
    for e in 1..=ITEMS {
        let item = cluster.post(
            gateway,
            "/items",
            &serde_json::json!({ "entity_id": e, "metadata": { "title": format!("item {e}") } }),
        );
        assert!(
            item.status().is_success(),
            "POST /items for entity {e} should route + apply (status {})",
            item.status()
        );
        let sig = cluster.post(
            gateway,
            "/signals",
            &serde_json::json!({ "entity_id": e, "signal": "view", "weight": 1.0 }),
        );
        assert!(
            sig.status().is_success(),
            "POST /signals for entity {e} should route + apply (status {})",
            sig.status()
        );
        written.insert(e);
    }
    written
}

/// The set of entity ids a `/feed` on `gateway` returned (best-effort: returns
/// the parsed item `entity_id`s; panics if the read itself failed).
fn feed_entities(cluster: &MultiProcCluster, gateway: usize, limit: u32) -> BTreeSet<u64> {
    let body = cluster.get_json(gateway, &format!("/feed?profile=for_you&limit={limit}"));
    body["items"]
        .as_array()
        .unwrap_or(&Vec::new())
        .iter()
        .filter_map(|it| it["entity_id"].as_u64())
        .collect()
}

/// Which groups the written corpus actually spans (by the gateway router hash) —
/// the test only asserts cross-shard completeness if the corpus genuinely touches
/// a group some partial node does not host.
fn groups_touched(written: &BTreeSet<u64>) -> BTreeSet<u16> {
    let router = ShardRouter::hash(SHARDS as u16).expect("build shard router");
    written
        .iter()
        .map(|&e| router.route(EntityId::new(e)).0)
        .collect()
}

/// Pillar 1: a `/feed` served from a node hosting a STRICT SUBSET of the groups
/// returns items spanning ALL groups (the cross-node fan-out), and the same read
/// on a differently-placed node returns the SAME corpus.
#[test]
fn mp_partial_placement_feed_spans_all_groups() {
    let cluster = MultiProcCluster::start_sharded_partial(
        NODES,
        &partial_placement(),
        Some(FAST_ELECTION_YAML),
    );
    let _ = cluster.wait_shard_leaders_agreed_partial(convergence_budget());

    // Seed the whole corpus through node 0's gateway (it forwards each write to
    // the owning group's leader — including group 1, which node 0 does not host).
    let written = seed_corpus(&cluster, 0);
    let touched = groups_touched(&written);
    assert_eq!(
        touched,
        (0..SHARDS as u16).collect(),
        "test corpus must touch EVERY group so each partial node misses real data — got {touched:?}"
    );

    // Let the cross-group write forwards + signal applies settle on every leader.
    std::thread::sleep(Duration::from_millis(800));

    let router = ShardRouter::hash(SHARDS as u16).expect("build shard router");

    // Node 0 hosts {0, 2} and MISSES group 1 — its group-1 items are reachable
    // ONLY by the cross-shard fan-out to a node hosting group 1.
    let on_node0 = feed_entities(&cluster, 0, ITEMS as u32);
    let group1_items: BTreeSet<u64> = written
        .iter()
        .copied()
        .filter(|&e| router.route(EntityId::new(e)).0 == 1)
        .collect();
    assert!(
        !group1_items.is_empty(),
        "precondition: some written items hash to group 1 (node 0's missing group)"
    );
    let returned_group1: BTreeSet<u64> = on_node0.intersection(&group1_items).copied().collect();
    assert_eq!(
        returned_group1, group1_items,
        "node 0's /feed must return EVERY group-1 item via the cross-shard fan-out \
         (missing items ⇒ the read was local-shard-only): returned {returned_group1:?} of {group1_items:?}"
    );
    // The full corpus is covered (the page holds them all).
    assert_eq!(
        on_node0, written,
        "node 0's cross-shard /feed must cover the WHOLE corpus"
    );

    // Placement-independence: node 1 hosts {0, 1} and MISSES group 2, yet returns
    // the SAME whole corpus (its missing group differs from node 0's).
    let on_node1 = feed_entities(&cluster, 1, ITEMS as u32);
    assert_eq!(
        on_node1, written,
        "node 1's cross-shard /feed (different missing group) must cover the same whole corpus"
    );
}

/// Pillar 2: killing the sole node hosting a group degrades a partial node's
/// `/feed` to a partial (non-empty) page — never a hard failure.
#[test]
fn mp_partial_placement_feed_degrades_when_group_unreachable() {
    let mut cluster = MultiProcCluster::start_sharded_partial(
        NODES,
        &partial_placement(),
        Some(FAST_ELECTION_YAML),
    );
    let _ = cluster.wait_shard_leaders_agreed_partial(convergence_budget());

    let written = seed_corpus(&cluster, 0);
    std::thread::sleep(Duration::from_millis(800));

    // Sanity: before the kill, node 0's cross-shard feed covers the whole corpus.
    let before = feed_entities(&cluster, 0, ITEMS as u32);
    assert_eq!(before, written, "pre-kill /feed must be corpus-complete");

    // Group 1 lives on nodes {1, 2}; node 0 hosts groups {0, 2}. Kill BOTH of
    // group 1's replicas → group 1 is unreachable from node 0's fan-out, while
    // node 0's LOCAL replicas of groups 0 and 2 still serve reads (a read needs no
    // leader). Node 0 is the lone survivor for groups 0 and 2.
    cluster.kill_hard(1);
    cluster.kill_hard(2);
    // Let the connect-fail / breaker surface (the fan-out hop to group 1 errors).
    std::thread::sleep(Duration::from_secs(2));

    // The degraded read MUST still succeed (HTTP 200) and return the reachable
    // groups' items — never a 5xx, never an empty page.
    let resp = cluster.get(0, &format!("/feed?profile=for_you&limit={ITEMS}"));
    assert!(
        resp.status().is_success(),
        "a missing-group outage must DEGRADE the read, not fail it (status {})",
        resp.status()
    );
    let body: serde_json::Value = resp.json().expect("feed body is JSON");
    let returned: BTreeSet<u64> = body["items"]
        .as_array()
        .expect("items array")
        .iter()
        .filter_map(|it| it["entity_id"].as_u64())
        .collect();

    let router = ShardRouter::hash(SHARDS as u16).expect("build shard router");
    // Groups 0 and 2 are hosted locally on node 0 → reachable. Group 1 → gone.
    let reachable: BTreeSet<u64> = written
        .iter()
        .copied()
        .filter(|&e| {
            let g = router.route(EntityId::new(e)).0;
            g == 0 || g == 2
        })
        .collect();
    let group1: BTreeSet<u64> = written
        .iter()
        .copied()
        .filter(|&e| router.route(EntityId::new(e)).0 == 1)
        .collect();

    assert!(
        !returned.is_empty(),
        "degraded /feed must still return the reachable groups' items, not an empty page"
    );
    // Every reachable-group item is still present (node 0's local scatter is fine).
    assert!(
        reachable.is_subset(&returned),
        "degraded /feed must still cover groups 0+2 (locally hosted): missing {:?}",
        reachable.difference(&returned).collect::<Vec<_>>()
    );
    // The unreachable group's items are gone — degraded, not magically present.
    assert!(
        returned.is_disjoint(&group1),
        "group 1 is unreachable (both replicas killed), so its items cannot appear"
    );
    // m12p4 review fix (CRITICAL): the degradation must be VISIBLE on the wire,
    // not server-log-only — the client must be able to tell a partial page from a
    // complete one. A degraded read carries a non-empty `unavailable_shards`.
    let unavailable = body["unavailable_shards"]
        .as_array()
        .expect("degraded /feed must surface `unavailable_shards` on the wire (not silent)");
    assert!(
        !unavailable.is_empty(),
        "a degraded read must NAME the unreachable group(s) so the client knows the page is partial"
    );
}