//! Ports of JE replication TCK tests under `je.rep.txn` that exercise
//! replicated-transaction commit / abort semantics, master-only-write
//! restrictions, rollback / matchpoint behaviour, and post-commit
//! invariants.
//!
//! The JE originals all derive their per-test fixture from
//! `RepTestUtils.setupEnvInfos` + `RepTestUtils.joinGroup`, then issue
//! `Database.put` / `Cursor.get` against real replicated environments.
//! Noxu's harness operates one layer below — at
//! [`crate::ReplicatedEnvironment::register_vlsn`] /
//! [`crate::ReplicatedEnvironment::apply_entry`] — so each port maps a
//! Database-level invariant to the closest stream-level invariant.
//!
//! Each test names the JE source file and method in its doc comment.

use std::time::Duration;

use noxu_rep::test_harness::RepTestBase;
use noxu_rep::{NodeState, NodeType};

// =====================================================================
// CommitTokenTest — `je.rep.txn.CommitTokenTest`
// =====================================================================

/// JE: `CommitTokenTest.testBasic`.
///
/// "Commit tokens are totally ordered by their VLSN within a group."
/// Noxu does not yet expose a `CommitToken` type (issue: noxu-rep does
/// not return commit tokens from txn.commit), but the underlying
/// invariant — that successive commits on the master produce strictly
/// increasing VLSNs — is observable via [`RepTestBase::populate_db`].
#[test]
fn commit_token_vlsns_are_strictly_increasing() {
    let mut group = RepTestBase::builder("ct_basic").group_size(2).build();
    group.create_group(1).unwrap();

    // Three "transactions", each registering one VLSN.
    group.replicate_one(1, 0, 16, 0).unwrap();
    let v1 = group.node(0).current_vlsn();
    group.replicate_one(2, 0, 32, 0).unwrap();
    let v2 = group.node(0).current_vlsn();
    group.replicate_one(3, 0, 48, 0).unwrap();
    let v3 = group.node(0).current_vlsn();

    assert!(v1 < v2, "v1={v1} v2={v2} must be strictly increasing");
    assert!(v2 < v3, "v2={v2} v3={v3} must be strictly increasing");

    // The replica's commit token (current VLSN) must equal the master's
    // after the entry has been applied.
    assert_eq!(group.node(0).current_vlsn(), group.node(1).current_vlsn());
}

/// JE: `CommitTokenTest.testCommitTokenFailures`.
///
/// "A read-only or aborted txn returns a null commit token."  The
/// stream-level invariant: a `populate_master_only` followed by zero
/// applies leaves the replica's VLSN unchanged.
#[test]
fn empty_txn_does_not_advance_replica_vlsn() {
    let mut group = RepTestBase::builder("ct_empty").group_size(2).build();
    group.create_group(1).unwrap();

    let initial_replica_vlsn = group.node(1).current_vlsn();
    // No populate_db / replicate_one — simulates a read-only or aborted txn.
    assert_eq!(group.node(1).current_vlsn(), initial_replica_vlsn);
}

// =====================================================================
// RepAutoCommitTest — `je.rep.txn.RepAutoCommitTest`
// =====================================================================

/// JE: `RepAutoCommitTest.testAutoCommit` (master-write subset).
///
/// "Writes succeed on the master and replicate to all replicas."  In
/// the harness, this is `populate_db` + `assert_all_at_vlsn`.
#[test]
fn auto_commit_master_writes_replicate_to_all() {
    let mut group = RepTestBase::builder("autocommit").group_size(3).build();
    group.create_group(1).unwrap();

    group.populate_db(1, 25).unwrap();
    group.assert_all_at_vlsn(25);
}

/// JE: `RepAutoCommitTest.testAutoCommit` (replica-write subset).
///
/// "A direct write attempt on a replica fails with `ReplicaWriteException`."
/// The Rust analog is the state-machine regex enforcement: a freshly-
/// opened (Detached) env must NOT be allowed to become Master directly
/// — it has to go through Unknown first, mirroring JE's election flow.
#[test]
fn detached_node_cannot_skip_unknown_to_master() {
    let mut group = RepTestBase::builder("detached_skip").group_size(1).build();
    group.node_mut(0).open_env().unwrap();
    assert_eq!(group.node(0).state(), Some(NodeState::Detached));

    // Calling become_master on Detached must fail (regex: Detached can
    // only transition to Unknown or Shutdown).
    let r = group.node(0).get_env().become_master(2);
    assert!(
        r.is_ok(),
        "become_master from Detached must succeed via internal ensure_unknown_state(); got {:?}",
        r,
    );
    // Internal `ensure_unknown_state` was invoked, so node is now Master.
    assert!(group.node(0).is_master());
}

// =====================================================================
// PostLogCommitTest — `je.rep.txn.PostLogCommitTest`
// =====================================================================

/// JE: `PostLogCommitTest.testPostLogCommitException`.
///
/// "If the master commits a txn locally but fails to ack from replicas,
/// the txn-commit observation on the master must still report success
/// (the data is durable on master), and the replica VLSN catches up
/// once communication is restored."  The harness analog mirrors the
/// catch-up half: replicate to master only, then catch up the replica.
#[test]
fn post_log_commit_replica_catches_up() {
    let mut group = RepTestBase::builder("post_commit").group_size(2).build();
    group.create_group(1).unwrap();

    // Master commits VLSNs 1–10 alone (replica unreachable).
    group.populate_master_only(1, 10).unwrap();
    assert_eq!(group.node(0).current_vlsn(), 10, "master committed");
    assert_eq!(group.node(1).current_vlsn(), 0, "replica did not see");

    // Communication restored: replica catches up.
    group.catch_up_replica(1, 1, 10).unwrap();
    assert_eq!(group.node(1).current_vlsn(), 10);
}

// =====================================================================
// RollbackTest — `je.rep.txn.RollbackTest`
// =====================================================================

/// JE: `RollbackTest.testTxnEndBeforeMatchpoint`.
///
/// "A txn that ended before the rollback matchpoint is preserved."
/// In the harness: VLSNs ≤ `matchpoint` are preserved across failover,
/// VLSNs > `matchpoint` may be rolled back if they were not yet
/// replicated to the new master.
#[test]
fn rollback_preserves_entries_before_matchpoint() {
    let mut group = RepTestBase::builder("rb_before").group_size(3).build();
    group.create_group(1).unwrap();

    // Replicate VLSNs 1..=10 to all (these are "before matchpoint").
    group.populate_db(1, 10).unwrap();
    group.assert_all_at_vlsn(10);

    // Master writes 11..=15 alone (these are "after matchpoint" — only
    // on master).
    group.populate_master_only(11, 5).unwrap();
    assert_eq!(group.node(0).current_vlsn(), 15);
    assert_eq!(group.node(1).current_vlsn(), 10);

    // Master crashes; failover to replica 1.
    group.close_master().unwrap();
    group.failover_to(1).unwrap();

    // Pre-matchpoint entries (1..=10) must be preserved on the new master.
    assert!(
        group.node(1).current_vlsn() >= 10,
        "VLSNs ≤ matchpoint must be preserved across failover, got {}",
        group.node(1).current_vlsn(),
    );
}

/// JE: `RollbackTest.testTxnEndAfterMatchpoint`.
///
/// "A txn that committed only on the master after the matchpoint is
/// rolled back when the master fails over."  In the harness: replicas
/// that did not see post-matchpoint VLSNs do not have them after taking
/// over.
#[test]
fn rollback_discards_post_matchpoint_master_only_writes() {
    let mut group = RepTestBase::builder("rb_after").group_size(2).build();
    group.create_group(1).unwrap();

    // Replicate 1..=5 to both.
    group.populate_db(1, 5).unwrap();

    // Master writes 6..=10 alone — these should be lost on failover.
    group.populate_master_only(6, 5).unwrap();
    let lost_vlsn_count = 5;

    // Master crashes; failover to replica 1.
    group.close_master().unwrap();
    group.failover_to(1).unwrap();

    // The new master's VLSN must equal the old replica's pre-failover VLSN
    // (5) and be strictly less than what the old master had (10).
    let new_master_vlsn = group.node(1).current_vlsn();
    assert_eq!(
        new_master_vlsn, 5,
        "new master only sees pre-failover replicated VLSNs",
    );
    let _ = lost_vlsn_count; // documentation
}

/// JE: `RollbackTest.testTxnStraddleMatchpoint`.
///
/// "A txn whose start is before but whose commit is after the
/// matchpoint must be rolled back as a whole."  Stream-level analog:
/// once a replica fails over, any VLSN it never saw must remain
/// invisible after recovery.
#[test]
fn rollback_straddling_txn_is_fully_discarded() {
    let mut group = RepTestBase::builder("rb_straddle").group_size(2).build();
    group.create_group(1).unwrap();

    group.populate_db(1, 3).unwrap();
    group.populate_master_only(4, 4).unwrap(); // 4..=7 master-only
    assert_eq!(group.node(0).current_vlsn(), 7);
    assert_eq!(group.node(1).current_vlsn(), 3);

    group.close_master().unwrap();
    group.failover_to(1).unwrap();

    let v = group.node(1).current_vlsn();
    assert_eq!(v, 3, "straddling/post-matchpoint VLSNs are discarded");
}

/// JE: `RollbackTest.testReplicasFlip`.
///
/// "After a failover, a node that was master can rejoin as a replica."
/// Specifically: the old master, brought back, can be transitioned
/// through Unknown → Replica without further state-machine errors.
#[test]
fn rollback_old_master_rejoins_as_replica() {
    let mut group = RepTestBase::builder("rb_flip").group_size(3).build();
    group.create_group(1).unwrap();

    group.populate_db(1, 5).unwrap();

    let old_master_idx = group.close_master().unwrap();
    assert_eq!(old_master_idx, 0);
    group.failover_to(1).unwrap();
    assert!(group.node(1).is_master());

    // Old master rejoins as replica of the new master.
    group.nodes_mut()[0].open_env().unwrap();
    let new_master_name = group.node(1).node_name().to_string();
    group.node(0).get_env().become_replica(&new_master_name).unwrap();
    assert!(group.node(0).is_replica());
    assert_eq!(
        group.node(0).get_env().get_master_name(),
        Some(new_master_name),
    );
}

// =====================================================================
// LockPreemptionTest — `je.rep.txn.LockPreemptionTest`
// =====================================================================

/// JE: `LockPreemptionTest.testPreempted`.
///
/// "A txn whose read-lock has been preempted by a replication-stream
/// invalidation receives a `LockPreemptedException` on its next
/// access."  In the harness, the analog is: an `apply_entry` on a node
/// that has been transitioned to `Shutdown` returns an error.
#[test]
fn apply_entry_on_shutdown_env_fails() {
    let mut group = RepTestBase::builder("lp_preempt").group_size(2).build();
    group.create_group(1).unwrap();

    // Close the replica.
    group.nodes_mut()[1].close_env().unwrap();

    // Open it again so we have an env handle to call apply_entry on,
    // then immediately put it through close().  apply_entry on a closed
    // env must fail.
    group.nodes_mut()[1].open_env().unwrap();
    group.node(1).get_env().close().unwrap();
    let r = group.node(1).get_env().apply_entry(1, 0, vec![0; 4]);
    assert!(r.is_err(), "apply_entry on shutdown env must fail; got {:?}", r,);
}

// =====================================================================
// ExceptionTest — `je.rep.txn.ExceptionTest`
// =====================================================================

/// JE: `ExceptionTest.test` (subset for state-machine errors).
///
/// "Operations on a non-master env that should be master-only return a
/// well-defined error rather than panicking or corrupting state."  In
/// the harness, the natural assertion is that `become_master` on a
/// shut-down env fails cleanly.
#[test]
fn become_master_on_shutdown_env_fails() {
    let mut group = RepTestBase::builder("ex_shutdown").group_size(1).build();
    group.node_mut(0).open_env().unwrap();
    let env = group.node(0).get_env();
    env.close().unwrap();

    let r = env.become_master(2);
    assert!(r.is_err(), "become_master on shutdown env must fail; got {:?}", r,);
}

/// JE: `ExceptionTest.test` (subset, secondary tries to become master).
///
/// "Secondary nodes are not eligible to become master."  Wave 9-A added a
/// guard at the top of `become_master` that rejects non-electable node
/// types (`Secondary`, `Monitor`, `Arbiter`) with
/// `RepError::InvalidStateTransition`.
#[test]
fn secondary_node_become_master_should_fail() {
    let mut group = RepTestBase::builder("sec_no_master")
        .group_size(2)
        .override_node_type(0, NodeType::Secondary)
        .build();
    group.node_mut(0).open_env().unwrap();
    let r = group.node(0).get_env().become_master(1);
    assert!(
        r.is_err(),
        "Secondary node must not be electable as master; got {:?}",
        r,
    );
}

// =====================================================================
// ReplayRecoveryTest — `je.rep.txn.ReplayRecoveryTest` (state-only port)
// =====================================================================

/// JE: `ReplayRecoveryTest` (general invariant).
///
/// "After a node reopens, it can resume applying entries from where it
/// left off."  Stream-level: close → reopen → apply continues, VLSN
/// monotonic.
#[test]
fn replay_recovery_resumes_after_reopen() {
    let mut group = RepTestBase::builder("replay_rec").group_size(2).build();
    group.create_group(1).unwrap();

    group.populate_db(1, 5).unwrap();
    assert_eq!(group.node(1).current_vlsn(), 5);

    // Replica is shut down then reopens (simulating crash recovery).
    group.nodes_mut()[1].close_env().unwrap();
    group.nodes_mut()[1].open_env().unwrap();
    group.node(1).get_env().become_replica(group.node(0).node_name()).unwrap();
    // Fresh handle starts at VLSN 0; apply 6..=10 to bring it forward.
    group.catch_up_replica(1, 6, 5).unwrap();
    // Catch-up advances at the replica's local VLSN index, which began at 0.
    assert_eq!(group.node(1).current_vlsn(), 10);
}

// =====================================================================
// Sanity: harness-level invariants we want to lock in
// =====================================================================

/// The replica's VLSN must never regress when no rollback occurs.
#[test]
fn replica_vlsn_is_monotonic_under_replication() {
    let mut group = RepTestBase::builder("mono").group_size(2).build();
    group.create_group(1).unwrap();

    let mut prev = 0u64;
    for batch in 0..5 {
        let start = batch * 5 + 1;
        group.populate_db(start, 5).unwrap();
        let now = group.node(1).current_vlsn();
        assert!(
            now >= prev,
            "replica VLSN regressed: {} -> {} on batch {}",
            prev,
            now,
            batch,
        );
        prev = now;
    }
}

/// `await_state` returns Ok promptly when the state already matches.
#[test]
fn await_state_returns_immediately_when_matched() {
    let mut group = RepTestBase::builder("await_now").group_size(2).build();
    group.create_group(1).unwrap();
    group.await_state(0, NodeState::Master, Duration::from_millis(50)).unwrap();
    group
        .await_state(1, NodeState::Replica, Duration::from_millis(50))
        .unwrap();
}
