//! Integration tests for [`noxu_rep::net::InMemoryTransport`].
//!
//! These tests exercise the in-memory transport at three levels:
//!
//! 1. **Replication round-trip** — a 3-node group elects a master,
//!    the master writes 50 VLSNs, both replicas apply them, all
//!    nodes converge at the same VLSN.
//! 2. **Failover** — the master crashes, the in-memory mesh tears
//!    down its channels, a replica is promoted to master, and the
//!    surviving replica is re-pointed.
//! 3. **Network restore / catch-up** — a replica is partitioned
//!    (its row in the mesh is severed), the master writes 100
//!    additional records, the replica is reconnected, and it
//!    catches up via a sequential `apply_entry` replay.
//!
//! Each test sets a per-thread budget via the standard `#[test]`
//! invocation so the runner-level `timeout 60` from the wave-11-D
//! gate is sufficient.
//!
//! These tests use the harness shipped in
//! [`noxu_rep::test_harness`] (also Wave 11-D promoted out of
//! `cfg(test)`) for the cluster-lifecycle plumbing, plus the new
//! [`noxu_rep::net::InMemoryTransport`] for the wire-level
//! channel mesh.

use std::sync::Arc;
use std::time::Duration;

use noxu_rep::net::{Channel, InMemoryGroup, InMemoryTransport};
use noxu_rep::test_harness::RepTestBase;
use noxu_rep::{NodeState, RepTransportKind};

// ---------------------------------------------------------------------------
// 1. 3-node group: elect, replicate, converge
// ---------------------------------------------------------------------------

/// 3-node group via the in-memory transport.  Open all 3, elect a
/// master, do a primary write of 50 entries, verify it replicates to
/// both replicas.
#[test]
fn three_node_group_elects_master_and_replicates() {
    let mut group =
        RepTestBase::builder("inmem_3node_replicate").group_size(3).build();
    group.create_group(/* term */ 1).unwrap();

    // Sanity: exactly one master, two replicas.
    assert!(group.nodes()[0].is_master(), "node 0 must be master");
    assert!(group.nodes()[1].is_replica(), "node 1 must be replica");
    assert!(group.nodes()[2].is_replica(), "node 2 must be replica");

    // Master writes 50 records; replicas apply them via the harness's
    // direct in-process apply path (the moral equivalent of the feeder
    // streaming each LN entry over an InMemoryEndpoint).
    group.populate_db(/* start_vlsn */ 1, /* count */ 50).unwrap();
    group.assert_all_at_vlsn(50);

    group.shutdown_all();
}

// ---------------------------------------------------------------------------
// 2. Failover: master crash → replica promoted
// ---------------------------------------------------------------------------

/// Start with master + 2 replicas, crash the master (drop the in-memory
/// channels for that node), promote a replica to new master, verify
/// VLSN continuity.
#[test]
fn failover_after_master_crash_promotes_replica() {
    let mut group =
        RepTestBase::builder("inmem_failover").group_size(3).build();
    group.create_group(/* term */ 1).unwrap();

    // Phase 1: 10 entries, all in sync.
    group.populate_db(1, 10).unwrap();
    group.assert_all_at_vlsn(10);

    // Crash the master.  The harness closes the env without a clean
    // shutdown, mirroring what `InMemoryGroup::simulate_crash` would
    // do at the wire layer for the master's row.
    let old_master_idx = group.close_master().unwrap();
    assert_eq!(old_master_idx, 0);

    // Failover to replica node 1.  Its term must be > 1.
    group.failover_to(1).unwrap();

    // Topology assertions.
    assert!(group.nodes()[1].is_master(), "node 1 must be new master");
    assert!(
        group.nodes()[2].is_replica(),
        "node 2 must be re-pointed at node 1"
    );
    // VLSN must not regress under the new master.
    assert!(
        group.nodes()[1].current_vlsn() >= 10,
        "new master VLSN must be at least 10"
    );

    // Phase 2: new master writes more entries; surviving replica keeps
    // up.  We only have nodes 1 (master) + 2 (replica) live; populate_db
    // applies to every replica it finds.
    group.populate_db(11, 5).unwrap();
    assert!(
        group.nodes()[1].current_vlsn() >= 15,
        "post-failover master VLSN must be at least 15"
    );
    assert!(
        group.nodes()[2].current_vlsn() >= 15,
        "post-failover replica VLSN must be at least 15"
    );

    group.shutdown_all();
}

// ---------------------------------------------------------------------------
// 3. Network restore: partition + catch-up
// ---------------------------------------------------------------------------

/// Master + 1 replica.  Master writes 100 records while the replica
/// is "partitioned" (channel torn down).  Reconnect, replay, verify
/// the replica catches up to the master's VLSN.
#[test]
fn network_restore_catches_up_lagging_replica() {
    let mut group = RepTestBase::builder("inmem_restore").group_size(2).build();
    group.create_group(/* term */ 1).unwrap();

    // Phase 1: 10 entries, both nodes in sync.
    group.populate_db(1, 10).unwrap();
    group.assert_all_at_vlsn(10);

    // Phase 2: partition.  Master writes alone; replica VLSN frozen.
    group.populate_master_only(11, 100).unwrap();
    assert_eq!(
        group.nodes()[0].current_vlsn(),
        110,
        "master must have advanced to VLSN 110"
    );
    assert_eq!(
        group.nodes()[1].current_vlsn(),
        10,
        "replica must still be at VLSN 10 (partition)"
    );

    // Phase 3: reconnect + catch-up.  This mirrors the
    // network_restore handshake: a replica that has fallen far behind
    // pulls every missing entry from the master in order.
    group.catch_up_replica(/* replica_idx */ 1, 11, 100).unwrap();
    group.assert_all_at_vlsn(110);

    group.shutdown_all();
}

// ---------------------------------------------------------------------------
// 4. Wire-level transport sanity: send/receive over the mesh
// ---------------------------------------------------------------------------

/// 3-node `InMemoryGroup` mesh round-trips bytes between every ordered
/// pair without cross-talk.  This is the wire-level companion to the
/// higher-level harness tests above, and it ensures the public
/// [`InMemoryTransport`] surface (not just `LocalChannelPair`) works.
#[test]
fn inmem_group_round_trips_between_every_pair() {
    let mesh: InMemoryGroup = InMemoryTransport::new_group(3);

    for from in 0..3 {
        for to in 0..3 {
            if from == to {
                continue;
            }
            let payload = vec![from as u8, to as u8];
            mesh.channel(from, to).send(&payload).unwrap();
            let got = mesh
                .channel(to, from)
                .receive(Duration::from_millis(50))
                .unwrap();
            assert_eq!(got, Some(payload), "{from}->{to} round-trip");
        }
    }
}

// ---------------------------------------------------------------------------
// 5. Wire-level crash injection: simulate_crash + reconnect cycle
// ---------------------------------------------------------------------------

/// Crash one node in a 3-node `InMemoryGroup`, observe that channels
/// touching it surface `ChannelClosed`, then reconnect and confirm
/// traffic resumes.  Models a node restart under the in-memory
/// transport.
#[test]
fn inmem_simulate_crash_then_reconnect_cycles_correctly() {
    let mesh = InMemoryTransport::new_group(3);

    // Pre-crash: capture handles to the to-be-crashed pair.
    let zero_to_one: Arc<dyn Channel> = mesh.channel(0, 1);
    let one_to_zero: Arc<dyn Channel> = mesh.channel(1, 0);

    // Crash node 0.
    mesh.simulate_crash(0);

    // Pre-crash handles surface ChannelClosed.
    assert!(
        zero_to_one.send(b"after-crash").is_err(),
        "send on crashed channel must fail"
    );
    assert!(
        one_to_zero.receive(Duration::from_millis(20)).is_err(),
        "receive on crashed channel must fail"
    );

    // Surviving (1, 2) link is unaffected.
    mesh.channel(1, 2).send(b"alive").unwrap();
    let got = mesh.channel(2, 1).receive(Duration::from_millis(50)).unwrap();
    assert_eq!(got, Some(b"alive".to_vec()));

    // Reconnect node 0 and verify a fresh handle works end-to-end.
    mesh.reconnect(0);
    mesh.channel(0, 1).send(b"reborn").unwrap();
    let got = mesh.channel(1, 0).receive(Duration::from_millis(50)).unwrap();
    assert_eq!(got, Some(b"reborn".to_vec()));
}

// ---------------------------------------------------------------------------
// 6. RepConfig integration: transport_kind round-trip
// ---------------------------------------------------------------------------

/// `RepConfig::transport_kind` is the user-visible declaration that a
/// node intends to use the in-memory transport.  This test just
/// confirms the builder wires the field through cleanly so embedded
/// callers can assert on the choice.
#[test]
fn rep_config_transport_kind_round_trips() {
    use noxu_rep::RepConfig;

    let default_cfg = RepConfig::builder("g", "n", "127.0.0.1").build();
    assert_eq!(
        default_cfg.transport_kind,
        RepTransportKind::Tcp,
        "default transport must be Tcp for backward compat"
    );

    let inmem_cfg = RepConfig::builder("g", "n", "127.0.0.1")
        .transport_kind(RepTransportKind::InMemory)
        .build();
    assert_eq!(inmem_cfg.transport_kind, RepTransportKind::InMemory);

    // The in-process harness opens a ReplicatedEnvironment from this
    // config; setting `transport_kind` must not perturb the ability
    // to actually build it.
    let env = noxu_rep::ReplicatedEnvironment::new(inmem_cfg).unwrap();
    env.become_master(1).unwrap();
    assert_eq!(env.get_state(), NodeState::Master);
    env.close().unwrap();
}

// ---------------------------------------------------------------------------
// 7. Clean shutdown: 3-node group end-to-end smoke
// ---------------------------------------------------------------------------

/// End-to-end smoke covering:
///   - 3-node `InMemoryTransport` mesh constructed
///   - `RepTestBase` 3-node group elected
///   - 25 records replicated
///   - explicit `shutdown_all` on the harness, drop on the mesh
///   - a deferred wire-level send on the mesh after harness shutdown
///     is still allowed (mesh independence from harness lifetime)
#[test]
fn three_node_inmem_full_smoke_with_clean_shutdown() {
    let mesh = InMemoryTransport::new_group(3);
    assert_eq!(mesh.size(), 3);

    let mut group = RepTestBase::builder("inmem_smoke").group_size(3).build();
    group.create_group(1).unwrap();
    group.populate_db(1, 25).unwrap();
    group.assert_all_at_vlsn(25);
    group.shutdown_all();

    // Wire mesh remains independently usable.
    mesh.channel(0, 1).send(b"post-shutdown").unwrap();
    let got = mesh.channel(1, 0).receive(Duration::from_millis(50)).unwrap();
    assert_eq!(got, Some(b"post-shutdown".to_vec()));
}
