//! m8p10 task 03: cross-process route tests.
//!
//! Builds 3-node in-process clusters — distinct `RegionClusterState`s pointing at
//! each other's REAL loopback gRPC + HTTP addresses — and drives them over real
//! HTTP (axum on loopback) + real `GrpcTransport` replication + real reqwest
//! forwarding. No OS processes (tier-3 OS-process coverage is tasks 04/05); these
//! run in the default test build like `cluster_region.rs` / `cluster_grpc.rs`.
//!
//! They prove the node behaves as one coherent cluster: writes to ANY node land
//! on the leader and replicate; promote propagates; `/cluster/status` aggregates
//! every region with `reachable` honesty; `/cluster/reconcile` converges both
//! sides and reports timing (idempotent on repeat); `/sharded/*` fans out across
//! processes with honest degraded semantics and routes writes to the owner.
#![allow(
    clippy::unwrap_used,
    clippy::missing_panics_doc,
    clippy::too_many_lines,
    clippy::doc_markdown,
    clippy::cast_precision_loss
)]

use std::{
    net::{SocketAddr, TcpListener},
    sync::Arc,
    time::{Duration, Instant},
};

use tidal_server::cluster::{
    ElectionSpec, RegionClusterState, RegionSpec, ReplicationSpec, TimeoutsSpec, TopologySpec,
    WalSpec, build_region_router,
};
use tidaldb::schema::{DecaySpec, EntityKind, Schema, SchemaBuilder, Window};

/// A `view` signal (decayed) + a `hide` hard-negative signal (so `/hardnegs`
/// resolves) — the same schema shape `cluster_region.rs` uses.
fn region_schema() -> Schema {
    let mut builder = SchemaBuilder::new();
    let _ = builder
        .signal(
            "view",
            EntityKind::Item,
            DecaySpec::Exponential {
                half_life: Duration::from_secs(7 * 24 * 3600),
            },
        )
        .windows(&[Window::OneHour])
        .velocity(false)
        .add();
    let _ = builder
        .signal("hide", EntityKind::Item, DecaySpec::Permanent)
        .velocity(false)
        .add();
    builder.build().unwrap()
}

fn free_addr() -> SocketAddr {
    TcpListener::bind("127.0.0.1:0")
        .unwrap()
        .local_addr()
        .unwrap()
}

/// A 3-region topology where every region declares real loopback grpc + http
/// addresses pointing at the in-test binds. Region 0 (`us-east`) is the leader.
struct Cluster3 {
    names: [String; 3],
    grpc: [SocketAddr; 3],
    http: [SocketAddr; 3],
}

impl Cluster3 {
    fn new() -> Self {
        Self {
            names: ["us-east".into(), "eu-west".into(), "ap-south".into()],
            grpc: [free_addr(), free_addr(), free_addr()],
            http: [free_addr(), free_addr(), free_addr()],
        }
    }

    fn topology(&self) -> TopologySpec {
        TopologySpec {
            regions: (0..3)
                .map(|i| RegionSpec {
                    name: self.names[i].clone(),
                    grpc_addr: Some(self.grpc[i].to_string()),
                    grpc_bind: None,
                    http_addr: Some(self.http[i].to_string()),
                    grpc_tls: None,
                    metrics_addr: None,
                })
                .collect(),
            leader: self.names[0].clone(),
            write_workers: None,
            timeouts: TimeoutsSpec::default(),
            replication: ReplicationSpec::default(),
            wal: WalSpec::default(),
            election: ElectionSpec::default(),
        }
    }

    fn base(&self, i: usize) -> String {
        format!("http://{}", self.http[i])
    }
}

/// A node's persistent data dir (m11p2). MUST be created before — and held
/// past — anything that can hold the node: a dir deleted while its node runs
/// wedges fjall's flush worker on `NotFound` and the node's checkpoint thread
/// polls forever.
fn region_dir() -> tempfile::TempDir {
    tempfile::tempdir().expect("create per-region data dir")
}

/// Build one region node off the reactor (GrpcTransport::new blocks on its own
/// runtime, so it must run on a plain thread).
fn build_region(
    topology: TopologySpec,
    region: &str,
    dir: &tempfile::TempDir,
) -> RegionClusterState {
    let region = region.to_string();
    let data_dir = dir.path().to_path_buf();
    std::thread::spawn(move || {
        RegionClusterState::new(
            &topology,
            &region,
            region_schema(),
            Vec::new(),
            Some(data_dir),
            0,
        )
    })
    .join()
    .unwrap()
    .expect("region node builds with real gRPC transport")
}

/// Serve `router` on `addr` using `rt`; returns once the listener is bound.
fn serve(rt: &tokio::runtime::Runtime, router: axum::Router, addr: SocketAddr) {
    let listener = rt
        .block_on(tokio::net::TcpListener::bind(addr))
        .unwrap_or_else(|e| panic!("bind {addr}: {e}"));
    rt.spawn(async move {
        let _ = axum::serve(listener, router).await;
    });
}

/// A 3-node serving handle: the runtime, the cluster spec, and an HTTP client.
struct Serving {
    /// Held for the test's lifetime so the axum servers keep running; dropped at
    /// test end, which stops the runtime and tears the nodes down.
    _rt: tokio::runtime::Runtime,
    /// Per-node data-dir guards (m11p2: persistent WAL). Declared AFTER the
    /// runtime so the nodes (owned by its tasks) drop BEFORE the dirs delete
    /// — a deleted-dir teardown wedges fjall's flush retry loop forever.
    _dirs: Vec<tempfile::TempDir>,
    cluster: Cluster3,
    client: reqwest::blocking::Client,
}

impl Serving {
    /// Build + serve all three region nodes on loopback.
    fn start_all() -> Self {
        Self::start(&[0, 1, 2])
    }

    /// Build + serve only the named regions (by index), so a test can leave one
    /// node DOWN to exercise unreachable/degraded paths.
    fn start(indices: &[usize]) -> Self {
        let rt = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(3)
            .enable_all()
            .build()
            .unwrap();
        let cluster = Cluster3::new();
        // Dirs FIRST: they must outlive every node (see `region_dir`).
        let dirs: Vec<tempfile::TempDir> = indices.iter().map(|_| region_dir()).collect();
        for (slot, &i) in indices.iter().enumerate() {
            let node = build_region(cluster.topology(), &cluster.names[i], &dirs[slot]);
            serve(
                &rt,
                build_region_router(Arc::new(node), None),
                cluster.http[i],
            );
        }
        Self {
            _rt: rt,
            _dirs: dirs,
            cluster,
            client: reqwest::blocking::Client::new(),
        }
    }

    fn base(&self, i: usize) -> String {
        self.cluster.base(i)
    }
}

/// Poll `pred(json)` against `GET {base}{path}` until it holds or the deadline.
fn poll_json(
    client: &reqwest::blocking::Client,
    url: &str,
    pred: impl Fn(&serde_json::Value) -> bool,
) -> serde_json::Value {
    let deadline = Instant::now() + Duration::from_secs(8);
    loop {
        let body: serde_json::Value = client.get(url).send().unwrap().json().unwrap();
        if pred(&body) {
            return body;
        }
        assert!(
            Instant::now() <= deadline,
            "predicate not met within 8s for {url}: {body}"
        );
        std::thread::sleep(Duration::from_millis(25));
    }
}

fn post_json(
    client: &reqwest::blocking::Client,
    url: &str,
    body: &serde_json::Value,
) -> reqwest::blocking::Response {
    client.post(url).json(body).send().unwrap()
}

// ── Tests ─────────────────────────────────────────────────────────────────────

/// A signal POSTed to a FOLLOWER is forwarded to the leader (204), durably
/// applied there, and replicates to every node over gRPC.
#[test]
fn forwarded_write_lands_on_leader_and_replicates() {
    let s = Serving::start_all();
    let client = &s.client;

    // Seed items on the LEADER (m11p2: items ride the replicated log — the
    // kind-1 WAL record reaches every follower; no HTTP broadcast). Then POST
    // signals to a FOLLOWER (eu-west, index 1) — these must forward to the
    // leader.
    for i in 1..=6u64 {
        let resp = post_json(
            client,
            &format!("{}/items", s.base(0)),
            &serde_json::json!({ "entity_id": i, "metadata": { "title": format!("item {i}") } }),
        );
        assert_eq!(resp.status().as_u16(), 201, "leader /items");

        // POST /signals to a FOLLOWER → forwarded to leader → 204.
        let resp = post_json(
            client,
            &format!("{}/signals", s.base(1)),
            &serde_json::json!({ "entity_id": i, "signal": "view", "weight": 1.0 }),
        );
        assert_eq!(
            resp.status().as_u16(),
            204,
            "follower /signals must forward to leader and 204: {}",
            resp.status()
        );
    }

    // Every follower converges (applied reaches 6, lag 0) via the WAL relay.
    for i in [1usize, 2] {
        poll_json(
            client,
            &format!("{}/cluster/status/local", s.base(i)),
            |st| {
                st["applied_events"].as_u64().unwrap_or(0) >= 6
                    && st["lag_events"].as_u64().unwrap_or(u64::MAX) == 0
            },
        );
    }

    // The followers' broadcast items are present and the feed ranks them.
    for i in 0..3 {
        let feed: serde_json::Value = client
            .get(format!("{}/feed?profile=trending&limit=6", s.base(i)))
            .send()
            .unwrap()
            .json()
            .unwrap();
        assert!(
            !feed["items"].as_array().unwrap().is_empty(),
            "node {i} feed must rank replicated items"
        );
    }
}

/// Promote a NON-leader (eu-west) via that follower; the promote fans out so all
/// three nodes agree on the new leader, and a write to the OLD leader now forwards
/// to the new one (transparently 204).
#[test]
fn promote_fans_out() {
    let s = Serving::start_all();
    let client = &s.client;

    // Promote eu-west (index 1) by calling promote ON eu-west; it fans out.
    let resp = post_json(
        client,
        &format!("{}/cluster/promote", s.base(1)),
        &serde_json::json!({ "region": s.cluster.names[1] }),
    );
    assert_eq!(resp.status().as_u16(), 200);
    let body: serde_json::Value = resp.json().unwrap();
    assert_eq!(body["leader"].as_str(), Some(s.cluster.names[1].as_str()));
    assert!(body.get("acked").is_some(), "promote must report fan-out");

    // All three nodes agree the leader is eu-west (status/local view).
    for i in 0..3 {
        let st = poll_json(
            client,
            &format!("{}/cluster/status/local", s.base(i)),
            |st| st["leader"].as_str() == Some(s.cluster.names[1].as_str()),
        );
        assert_eq!(st["leader"].as_str(), Some(s.cluster.names[1].as_str()));
    }

    // A write to the OLD leader (us-east, index 0) now forwards to the new leader
    // (eu-west) and 204s — transparent leader forwarding after a promote.
    let resp = post_json(
        client,
        &format!("{}/signals", s.base(0)),
        &serde_json::json!({ "entity_id": 1, "signal": "view", "weight": 1.0 }),
    );
    assert_eq!(
        resp.status().as_u16(),
        204,
        "write to demoted node must forward to the new leader: {}",
        resp.status()
    );
}

/// `GET /cluster/status` on any node aggregates EVERY region; with one node DOWN
/// that region reports `reachable: false`, `partitioned: true`, worst-case lag.
#[test]
fn status_aggregates_all_regions() {
    // Start only nodes 0 and 1; ap-south (index 2) is DOWN.
    let s = Serving::start(&[0, 1]);
    let client = &s.client;

    // Seed a couple of signals so the leader's relay seqno is non-zero.
    for i in 1..=3u64 {
        let _ = post_json(
            client,
            &format!("{}/signals", s.base(0)),
            &serde_json::json!({ "entity_id": i, "signal": "view", "weight": 1.0 }),
        );
    }

    // Aggregate from the leader: 3 regions, ap-south unreachable.
    let status = poll_json(client, &format!("{}/cluster/status", s.base(0)), |st| {
        let regions = st["regions"].as_array();
        regions.is_some_and(|r| r.len() == 3) && st["relay_log_len"].as_u64().unwrap_or(0) >= 3
    });
    assert_eq!(status["leader"].as_str(), Some(s.cluster.names[0].as_str()));
    let relay_len = status["relay_log_len"].as_u64().unwrap();
    assert!(relay_len >= 3, "leader relay len: {relay_len}");

    let regions = status["regions"].as_array().unwrap();
    let row = |name: &str| {
        regions
            .iter()
            .find(|r| r["name"].as_str() == Some(name))
            .unwrap()
    };

    // The leader row: reachable, zero lag.
    let leader_row = row(&s.cluster.names[0]);
    assert_eq!(leader_row["reachable"].as_bool(), Some(true));
    assert_eq!(leader_row["lag_events"].as_u64(), Some(0));

    // The live follower (eu-west): reachable.
    let follower_row = row(&s.cluster.names[1]);
    assert_eq!(follower_row["reachable"].as_bool(), Some(true));

    // The DOWN region (ap-south): unreachable, partitioned, applied 0, worst-case
    // lag = leader_last_seq.
    let down_row = row(&s.cluster.names[2]);
    assert_eq!(
        down_row["reachable"].as_bool(),
        Some(false),
        "down region unreachable: {down_row}"
    );
    assert_eq!(down_row["partitioned"].as_bool(), Some(true));
    assert_eq!(down_row["applied_events"].as_u64(), Some(0));
    assert_eq!(
        down_row["lag_events"].as_u64(),
        Some(relay_len),
        "worst-case lag = leader_last_seq"
    );
}

/// `/cluster/reconcile` exchanges CRDT snapshots and converges both sides: a hide
/// recorded ONLY on us-east becomes visible on eu-west after reconcile, the timing
/// fields are reported, and a second reconcile is a no-op (idempotent).
#[test]
fn reconcile_converges_and_times() {
    let s = Serving::start(&[0, 1]);
    let client = &s.client;

    // Record a hide on us-east (the leader) only. (Hardnegs converge via reconcile,
    // not the WAL relay, so eu-west does not see it yet.)
    let resp = post_json(
        client,
        &format!("{}/hardnegs", s.base(0)),
        &serde_json::json!({ "user_id": 42, "item_id": 7 }),
    );
    assert_eq!(
        resp.status().as_u16(),
        204,
        "hardneg on leader: {}",
        resp.status()
    );

    // Reconcile eu-west WITH us-east: eu-west drives the exchange, applying
    // us-east's pre-merge snapshot (which carries the hide).
    let resp = post_json(
        client,
        &format!("{}/cluster/reconcile", s.base(1)),
        &serde_json::json!({ "region": s.cluster.names[0] }),
    );
    assert_eq!(resp.status().as_u16(), 200, "reconcile: {}", resp.status());
    let body: serde_json::Value = resp.json().unwrap();
    assert_eq!(body["ok"].as_bool(), Some(true));
    assert_eq!(body["region"].as_str(), Some(s.cluster.names[0].as_str()));
    // Both timing fields are present (u64, possibly 0 on a fast localhost merge).
    assert!(
        body["local_elapsed_ms"].is_u64(),
        "local_elapsed_ms reported: {body}"
    );
    assert!(
        body["remote_elapsed_ms"].is_u64(),
        "remote_elapsed_ms reported: {body}"
    );
    let first_ops = body["ops_applied"].as_u64().unwrap();
    assert!(
        first_ops >= 1,
        "first reconcile must apply the hide: {body}"
    );

    // Capture eu-west's hardneg state after the first reconcile via take_snapshot
    // (exposed over the internal snapshot route): the hide for (42, 7) is present.
    let hardneg_present = |base: &str| -> bool {
        // Drive the internal snapshot route with an EMPTY remote snapshot (a no-op
        // merge) and read back the node's pre-merge snapshot, which lists every
        // hard-negative register it holds.
        let resp = client
            .post(format!("{base}/cluster/reconcile/snapshot"))
            .header("x-tidal-internal", "1")
            .json(&serde_json::json!({ "signal_states": [], "hardneg_registers": [] }))
            .send()
            .unwrap();
        assert_eq!(resp.status().as_u16(), 200, "snapshot route");
        let body: serde_json::Value = resp.json().unwrap();
        body["pre_merge_snapshot"]["hardneg_registers"]
            .as_array()
            .is_some_and(|regs| {
                regs.iter().any(|r| {
                    // wire form: [user_id, item_id, register]
                    r.get(0).and_then(serde_json::Value::as_u64) == Some(42)
                        && r.get(1).and_then(serde_json::Value::as_u64) == Some(7)
                })
            })
    };
    assert!(
        hardneg_present(&s.base(1)),
        "after reconcile, eu-west must hold the hide for (42, 7)"
    );

    // A second reconcile is idempotent IN EFFECT: the LWW plan re-resolves the
    // same hide (the engine re-stamps Hide registers at now() per snapshot, so the
    // op count is not 0 — that is the engine's honest semantics), but the
    // converged STATE is unchanged: the hide is still present on both sides, never
    // dropped or duplicated.
    let resp = post_json(
        client,
        &format!("{}/cluster/reconcile", s.base(1)),
        &serde_json::json!({ "region": s.cluster.names[0] }),
    );
    assert_eq!(resp.status().as_u16(), 200);
    assert!(
        hardneg_present(&s.base(1)) && hardneg_present(&s.base(0)),
        "second reconcile must leave both sides converged (hide present on both)"
    );
}

/// `/sharded/feed` fans out across processes; with one node DOWN the response is
/// `degraded: true`, names the unavailable shard, and still merges results from
/// the live shards.
#[test]
fn sharded_feed_degrades_honestly() {
    // Start nodes 0 and 1; ap-south (index 2) is DOWN.
    let s = Serving::start(&[0, 1]);
    let client = &s.client;

    // Seed items + signals on BOTH live nodes' local stores via /sharded writes so
    // each shard owns its slice. (Sharded writes route to the owner; a write whose
    // owner is the DOWN node will 503, which we tolerate — we only need the live
    // shards populated for the read fan-out.)
    for i in 1..=12u64 {
        let _ = post_json(
            client,
            &format!("{}/sharded/items", s.base(0)),
            &serde_json::json!({ "entity_id": i, "metadata": { "title": format!("item {i}") } }),
        );
        let _ = post_json(
            client,
            &format!("{}/sharded/signals", s.base(0)),
            &serde_json::json!({ "entity_id": i, "signal": "view", "weight": i as f64 }),
        );
    }

    // Sharded feed from the leader, generous per-shard deadline so the LIVE shards
    // always answer and only the DOWN shard degrades.
    let feed: serde_json::Value = poll_json(
        client,
        &format!(
            "{}/sharded/feed?profile=trending&limit=12&deadline_ms=2000",
            s.base(0)
        ),
        |f| f["scatter_gather"]["degraded"].as_bool() == Some(true),
    );
    let sg = &feed["scatter_gather"];
    assert_eq!(
        sg["degraded"].as_bool(),
        Some(true),
        "must be degraded: {feed}"
    );
    let unavailable = sg["unavailable_shards"].as_array().unwrap();
    assert!(
        unavailable
            .iter()
            .any(|u| u.as_str() == Some(s.cluster.names[2].as_str())),
        "unavailable_shards must name the down region: {sg}"
    );
    // Still returns merged results from the live shards — never an error.
    assert!(
        !feed["items"].as_array().unwrap().is_empty(),
        "degraded sharded feed must still return live-shard results: {feed}"
    );
}

/// A `/sharded/items` write whose `ShardRouter` owner is a REMOTE region is
/// forwarded to that region and becomes readable via the owner's local feed.
#[test]
fn sharded_write_routes_to_owner() {
    let s = Serving::start_all();
    let client = &s.client;

    // Find an entity id whose engine ShardRouter owner is NOT the leader (so the
    // write must forward). Probe the entity_shard mapping the server uses by
    // checking which region's local feed the item lands on after a /sharded write.
    // The engine hash is deterministic; iterate ids until one owns to region 1 or 2.
    let mut routed = false;
    for entity_id in 1..=200u64 {
        let resp = post_json(
            client,
            &format!("{}/sharded/items", s.base(0)),
            &serde_json::json!({ "entity_id": entity_id, "metadata": { "title": format!("e{entity_id}") } }),
        );
        // 201 = created (locally or forwarded to a reachable owner).
        if resp.status().as_u16() != 201 {
            continue;
        }
        // Add a signal too so the item is rankable on its owner.
        let _ = post_json(
            client,
            &format!("{}/sharded/signals", s.base(0)),
            &serde_json::json!({ "entity_id": entity_id, "signal": "view", "weight": 5.0 }),
        );

        // Check each non-leader region's LOCAL feed: if the item appears there, it
        // was routed to that owner (a remote region), proving sharded forwarding.
        for owner in [1usize, 2] {
            let feed: serde_json::Value = client
                .get(format!("{}/feed?profile=trending&limit=50", s.base(owner)))
                .send()
                .unwrap()
                .json()
                .unwrap();
            let on_owner = feed["items"]
                .as_array()
                .unwrap()
                .iter()
                .any(|it| it["entity_id"].as_u64() == Some(entity_id));
            if on_owner {
                routed = true;
                break;
            }
        }
        if routed {
            break;
        }
    }
    assert!(
        routed,
        "no sharded item routed to a remote region within 200 ids — sharded forwarding broken"
    );
}