//! m11p5 — kind-4 `MembershipRecord` on the one replicated log (data layer).
//!
//! Proves the WAL-level membership machinery end to end through a real
//! persistent `TidalDb` in cluster mode:
//!
//! - a leader appends a kind-4 record, the `ClusterMembership` cell folds it,
//!   and the seqno is durable;
//! - reopening the data dir re-derives the LATEST roster from the surviving log
//!   (recovery fold — the highest version wins, no merge logic);
//! - the append path refuses version regressions and non-cluster mode.
//!
//! Consumer logic (quorum / election reconfigure) is NOT exercised here — this
//! is the data layer only, exactly the m11p5 A3 task slice.

#![allow(clippy::unwrap_used)]

use tidaldb::{
    TempTidalHome, TidalDb,
    db::config::{NodeConfig, NodeRole},
    replication::shard::ShardId,
    schema::{DecaySpec, EntityKind, SchemaBuilder, Window},
    testing::{CrashInjector, CrashPoint, crash_injector::run_with_crash},
    wal::format::{MemberEntry, MemberRole, MembershipRecord},
};

use std::time::Duration;

fn make_schema() -> tidaldb::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::AllTime])
        .velocity(false)
        .add();
    builder.build().unwrap()
}

/// Open a PERSISTENT cluster node (non-empty `peer_shards` → `replicate_blobs`,
/// the WAL becomes the one replicated log so kind-4 records can ride it).
fn open_cluster_node(home: &TempTidalHome, schema: tidaldb::schema::Schema) -> TidalDb {
    TidalDb::builder()
        .with_data_dir(home.path())
        .with_schema(schema)
        .with_cluster(NodeConfig {
            role: NodeRole::Single,
            shard_id: ShardId(0),
            peer_shards: vec![ShardId(1)],
            ..NodeConfig::default()
        })
        .open()
        .expect("persistent cluster node opens")
}

fn member(id: u16, name: &str, role: MemberRole) -> MemberEntry {
    MemberEntry {
        id,
        name: name.to_string(),
        grpc_addr: format!("tidaldb-{id}.peers.svc.cluster.local:9500"),
        http_addr: format!("http://tidaldb-{id}.peers.svc.cluster.local:9501"),
        role,
    }
}

fn roster_v1() -> MembershipRecord {
    MembershipRecord {
        version: 1,
        term: 2,
        members: vec![
            member(0, "us-east", MemberRole::Voter),
            member(1, "eu-west", MemberRole::Voter),
            member(2, "ap-south", MemberRole::Voter),
        ],
    }
}

/// A v2 record that adds a learner (the join → learner conf-change shape).
fn roster_v2() -> MembershipRecord {
    MembershipRecord {
        version: 2,
        term: 2,
        members: vec![
            member(0, "us-east", MemberRole::Voter),
            member(1, "eu-west", MemberRole::Voter),
            member(2, "ap-south", MemberRole::Voter),
            member(3, "us-west", MemberRole::Learner),
        ],
    }
}

/// A leader appends a membership record; the cell folds it and the seqno is
/// non-zero (a real durable WAL slot, never the dedup sentinel).
#[test]
fn append_folds_the_cell_and_is_durable() {
    let home = TempTidalHome::new().unwrap();
    let db = open_cluster_node(&home, make_schema());

    // Topology era: no kind-4 record applied yet.
    assert_eq!(db.cluster_membership(), None);

    let seq = db.append_membership_record(roster_v1()).unwrap();
    assert!(
        seq >= 1,
        "a membership record consumes a real seqno, got {seq}"
    );

    let (version, term, members) = db.cluster_membership().expect("cell folded the record");
    assert_eq!(version, 1);
    assert_eq!(term, 2);
    assert_eq!(members, roster_v1().members);
}

/// Recovery fold: two records durable in the WAL, then a hard CRASH (no
/// graceful shutdown — so no compaction runs over the control records), then
/// reopen. Recovery re-derives the LATEST version (full snapshots, no merge
/// logic; the highest version below the durable frontier wins).
///
/// A crash — not a clean `close()` — is the faithful test: it exercises exactly
/// the path that re-derives the roster from the surviving log on restart, and
/// it sidesteps the shutdown-time WAL compaction that (absent the §2.1
/// retention pin, a later m11p5 stage) would reclaim a segment holding only
/// control records.
#[test]
fn recovery_re_derives_the_latest_roster_after_crash() {
    let home = TempTidalHome::new().unwrap();
    let schema = make_schema();

    // The FIRST CheckpointPreFlush passes (our explicit force_replication_checkpoint
    // lands the WAL marker durably at the records' seq); the SECOND (the shutdown
    // ledger checkpoint) fires the crash, so no compaction runs.
    let injector = CrashInjector::new(CrashPoint::CheckpointPreFlush, 1);
    let outcome = run_with_crash(&injector, || {
        let db = open_cluster_node(&home, schema.clone());
        db.append_membership_record(roster_v1()).unwrap();
        db.append_membership_record(roster_v2()).unwrap();
        // The live cell already holds v2 before the crash.
        assert_eq!(db.cluster_membership().unwrap().0, 2);

        // Land a crash-consistent on-disk checkpoint (WAL marker at the records'
        // seq); the membership records' seqnos are durable in the WAL.
        db.force_replication_checkpoint().unwrap();

        // CRASH during the shutdown ledger checkpoint — nothing past here, and
        // no compaction, reaches disk.
        let _ = db.close();
    });
    assert!(
        matches!(outcome, Err(CrashPoint::CheckpointPreFlush)),
        "the crash injector must have fired during shutdown, got {outcome:?}"
    );

    // Reopen the crashed node: recovery scans the surviving log and re-derives
    // the highest version. v1 sits below v2 in the log; the monotonic-by-version
    // advance makes "latest wins" without any delta merge.
    let db = open_cluster_node(&home, schema);
    let (version, term, members) = db
        .cluster_membership()
        .expect("recovery re-derived the roster from the surviving log");
    assert_eq!(version, 2, "the highest-version record wins after recovery");
    assert_eq!(term, 2);
    assert_eq!(members, roster_v2().members);
    assert_eq!(
        members.len(),
        4,
        "the learner is present in the recovered roster"
    );
}

/// A version that does not strictly advance the applied cell is refused — the
/// cell advances only forward, so a regressing or re-appended version could
/// never fold and must fail loudly rather than journal a silent no-op.
#[test]
fn version_regression_is_refused() {
    let home = TempTidalHome::new().unwrap();
    let db = open_cluster_node(&home, make_schema());

    db.append_membership_record(roster_v2()).unwrap(); // version 2

    // Re-appending the same version is refused.
    let same = db.append_membership_record(roster_v2()).unwrap_err();
    assert!(same.to_string().contains("does not advance"), "got: {same}");

    // A lower version is refused.
    let lower = db.append_membership_record(roster_v1()).unwrap_err();
    assert!(
        lower.to_string().contains("does not advance"),
        "got: {lower}"
    );

    // The cell still holds v2 (no failed append mutated it).
    assert_eq!(db.cluster_membership().unwrap().0, 2);
}

/// A structurally-invalid roster is refused before it enters the log (it would
/// fail apply identically on every follower).
#[test]
fn invalid_roster_is_refused() {
    let home = TempTidalHome::new().unwrap();
    let db = open_cluster_node(&home, make_schema());

    let dup = MembershipRecord {
        version: 1,
        term: 1,
        members: vec![
            member(5, "a", MemberRole::Voter),
            member(5, "b", MemberRole::Voter),
        ],
    };
    let err = db.append_membership_record(dup).unwrap_err();
    assert!(
        err.to_string().contains("duplicate member id"),
        "got: {err}"
    );
    assert_eq!(db.cluster_membership(), None, "no record entered the log");
}

/// A non-cluster (standalone) node refuses the append: the WAL is not a
/// replicated log there.
#[test]
fn standalone_node_refuses_membership_append() {
    let home = TempTidalHome::new().unwrap();
    let db = TidalDb::builder()
        .with_data_dir(home.path())
        .with_schema(make_schema())
        .open()
        .expect("standalone node opens");

    let err = db.append_membership_record(roster_v1()).unwrap_err();
    assert!(err.to_string().contains("cluster mode"), "got: {err}");
}