//! BLOCKER 6 crash-recovery integration test: a follower is a DURABLE replica.
//!
//! Before this fix, a follower applied replicated events only to its in-memory
//! signal ledger (`apply_wal_event` bypasses the WAL) and never persisted its
//! per-shard applied high-water-mark (`to_checkpoint_bytes`/`from_checkpoint_bytes`
//! were dead, and `ReplicationState::new` pinned every shard at 0 on every
//! startup). A follower crash therefore silently lost every replicated event
//! since process start, and on restart the follower was indistinguishable from a
//! fresh node (`applied_seqno = 0`) — making failover "promote highest replayed
//! seqno" meaningless and guaranteeing double-counting on every re-shipped
//! segment.
//!
//! The fix has two halves, both proven here end-to-end on a real persistent
//! follower with a real on-disk WAL + fjall store:
//!
//! 1. **WAL-first apply (Option B):** the receiver now applies each replicated
//!    event via `apply_replicated_event`, which appends it to the follower's OWN
//!    WAL (blocking until fsync) before mutating in-memory state. The follower's
//!    normal WAL replay on open reconstructs the applied aggregate — no acked
//!    replicated state is lost across a crash.
//! 2. **Persisted high-water-mark:** the per-shard `applied_seqno` is checkpointed
//!    (periodically + on shutdown + via `force_replication_checkpoint`) under
//!    `Tag::ReplicationState` and restored on open, so the idempotency boundary
//!    survives a restart and a re-shipped already-applied segment is a no-op.
//!
//! The crash is a REAL crash: `run_with_crash` panic-unwinds out of the follower's
//! shutdown checkpoint (no graceful shutdown completes), leaving on disk exactly
//! the crash-consistent checkpoint that `force_replication_checkpoint` landed
//! before the crash. The lock file releases as the db drops during the unwind.

#![allow(
    clippy::too_many_lines,
    clippy::unwrap_used,
    clippy::cast_possible_truncation,
    clippy::doc_markdown
)]

use std::{sync::Arc, time::Duration};

use tidaldb::{
    TempTidalHome, TidalDb,
    db::config::{NodeConfig, NodeRole},
    replication::{
        RegionId, WalSegmentId,
        shard::ShardId,
        state::ReplicationState,
        transport::{Transport, TransportError, WalSegmentPayload},
    },
    schema::{DecaySpec, EntityId, EntityKind, SchemaBuilder, Window},
    signals::{NoopWalWriter, SignalLedger},
    testing::{CrashInjector, CrashPoint, crash_injector::run_with_crash},
    wal::format::batch::{EventRecord, encode_batch},
};

// ── Test transport: a channel the test pushes shipped segments into ──────────

struct ChannelTransport {
    rx: crossbeam::channel::Receiver<WalSegmentPayload>,
}

impl Transport for ChannelTransport {
    fn send_segment(
        &self,
        _to: ShardId,
        _payload: WalSegmentPayload,
    ) -> Result<(), TransportError> {
        Ok(())
    }
    fn recv_segment(&self) -> Option<WalSegmentPayload> {
        self.rx.recv().ok()
    }
    fn local_shard(&self) -> ShardId {
        ShardId::SINGLE
    }
}

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().expect("schema must be valid")
}

/// Signal type id for "view" (deterministic; first signal alphabetically = 0).
fn view_type_id(schema: &tidaldb::schema::Schema) -> u8 {
    let ledger = SignalLedger::new(schema.clone(), Box::new(NoopWalWriter));
    ledger.resolve_signal_type("view").unwrap().as_u16() as u8
}

/// Open a PERSISTENT follower against `home`.
fn open_persistent_follower(home: &TempTidalHome, schema: tidaldb::schema::Schema) -> TidalDb {
    TidalDb::builder()
        .with_data_dir(home.path())
        .with_schema(schema)
        .with_cluster(NodeConfig {
            role: NodeRole::Follower,
            ..NodeConfig::default()
        })
        .open()
        .expect("persistent follower should open")
}

/// Build a WAL segment payload carrying `entities` as "view" events, with the
/// batch's first WAL seq = `first_seq` (so its last seq = first_seq + N - 1).
fn build_segment(
    schema: &tidaldb::schema::Schema,
    entities: &[u64],
    first_seq: u64,
) -> WalSegmentPayload {
    let type_id = view_type_id(schema);
    let events: Vec<EventRecord> = entities
        .iter()
        .enumerate()
        .map(|(i, &e)| EventRecord::signal(e, type_id, 1.0, 1_000_000_000 + i as u64))
        .collect();
    let bytes = encode_batch(&events, first_seq, first_seq).unwrap();
    WalSegmentPayload {
        id: WalSegmentId::new(RegionId::SINGLE, ShardId::SINGLE, first_seq),
        bytes,
        event_count: entities.len() as u64,
        // Last WAL seq = first_seq + N - 1 (matches the doc comment above).
        leader_last_seq: first_seq + entities.len() as u64 - 1,
    }
}

/// Bounded poll for the follower's applied seqno to reach `expected`.
fn wait_for_applied(state: &ReplicationState, expected: u64) {
    let deadline = std::time::Instant::now() + Duration::from_secs(5);
    loop {
        if state.applied_seqno(ShardId::SINGLE) == Some(expected) {
            return;
        }
        assert!(
            std::time::Instant::now() < deadline,
            "follower did not reach applied seqno {expected} within 5s (saw {:?})",
            state.applied_seqno(ShardId::SINGLE),
        );
        std::thread::sleep(Duration::from_millis(1));
    }
}

/// The full BLOCKER 6 contract on a real persistent follower across a hard crash.
#[test]
fn follower_replicated_state_and_hwm_survive_a_crash() {
    let home = TempTidalHome::new().unwrap();
    let schema = make_schema();
    let type_name = "view";

    // ── Phase 1: open follower, replicate a segment, land a crash-consistent
    //    checkpoint, then CRASH (no graceful shutdown). ────────────────────────
    //
    // fire_after_n = 1: the FIRST CheckpointPreFlush (inside the explicit
    // force_replication_checkpoint below) passes so the crash-consistent
    // checkpoint actually lands on disk; the SECOND (the follower's shutdown
    // signal-ledger checkpoint) fires the crash. The follower has no cohort
    // entries, so only the two signal-ledger checkpoints reach this crash point.
    let injector = CrashInjector::new(CrashPoint::CheckpointPreFlush, 1);
    let outcome = run_with_crash(&injector, || {
        let follower = open_persistent_follower(&home, schema.clone());
        let state = follower.replication_state().clone();

        // Wire a channel transport and start the real receiver thread.
        let (tx, rx) = crossbeam::channel::bounded(8);
        let transport = Arc::new(ChannelTransport { rx });
        follower.start_replication(Arc::clone(&transport)).unwrap();

        // Leader ships a segment carrying WAL seqs 1..=3 (three "view" events).
        tx.send(build_segment(&schema, &[10, 20, 30], 1)).unwrap();
        wait_for_applied(&state, 3);

        // The follower's in-memory aggregate reflects all three replicated views.
        for e in [10u64, 20, 30] {
            assert_eq!(
                follower
                    .read_windowed_count(EntityId::new(e), type_name, Window::AllTime)
                    .unwrap(),
                1,
                "follower must have applied replicated view for entity {e}"
            );
        }
        assert_eq!(state.applied_seqno(ShardId::SINGLE), Some(3));

        // Land a crash-consistent on-disk checkpoint: ledger aggregate + WAL
        // marker + replication high-water-mark, all at the same WAL seq. This is
        // the durable state a crash must preserve.
        follower.force_replication_checkpoint().unwrap();

        // Drop the transport sender so the receiver thread can exit cleanly when
        // shutdown_inner joins it (the crash fires AFTER that join, during the
        // signal-ledger checkpoint, so it is a true post-join crash).
        drop(tx);

        // CRASH: close() panic-unwinds at CheckpointPreFlush during the final
        // signal-ledger checkpoint. Nothing past the force_replication_checkpoint
        // above reaches disk. The Result is irrelevant — the injector panic
        // unwinds out of close(), so run_with_crash yields Err(CrashPoint).
        let _ = follower.close();
    });
    assert!(
        matches!(outcome, Err(CrashPoint::CheckpointPreFlush)),
        "the crash injector must have fired during the follower's shutdown checkpoint, \
         got {outcome:?}"
    );

    // ── Phase 2: reopen the crashed follower. Prove the three BLOCKER 6
    //    guarantees. ────────────────────────────────────────────────────────────
    let follower = open_persistent_follower(&home, schema.clone());

    // (a) Replicated signal state is intact after reopen (recovered from the
    //     ledger checkpoint + the follower's OWN WAL replay — Option B). Without
    //     the WAL-first apply, these counts would be 0 (in-memory only, lost on
    //     crash).
    for e in [10u64, 20, 30] {
        assert_eq!(
            follower
                .read_windowed_count(EntityId::new(e), type_name, Window::AllTime)
                .unwrap(),
            1,
            "replicated view for entity {e} must survive the crash (durable follower replica)"
        );
    }

    // (b) The applied high-water-mark survived — NOT reset to 0. A restarted
    //     follower that reported applied_seqno 0 would be indistinguishable from a
    //     fresh node, making failover-by-highest-replayed-seqno meaningless.
    let restored_state = follower.replication_state().clone();
    assert_eq!(
        restored_state.applied_seqno(ShardId::SINGLE),
        Some(3),
        "applied high-water-mark must be restored from the checkpoint, not reset to 0"
    );

    // (c) Re-shipping the SAME already-applied segment is an idempotent no-op:
    //     the restored high-water-mark gates it, so counts do not double.
    let (tx2, rx2) = crossbeam::channel::bounded(8);
    let transport2 = Arc::new(ChannelTransport { rx: rx2 });
    follower.start_replication(Arc::clone(&transport2)).unwrap();

    tx2.send(build_segment(&schema, &[10, 20, 30], 1)).unwrap();
    // Give the receiver time to observe-and-skip the duplicate segment. The HWM
    // stays at 3, so we poll for stability rather than an advance.
    std::thread::sleep(Duration::from_millis(50));

    assert_eq!(
        restored_state.applied_seqno(ShardId::SINGLE),
        Some(3),
        "re-shipped already-applied segment must not advance the high-water-mark"
    );
    for e in [10u64, 20, 30] {
        assert_eq!(
            follower
                .read_windowed_count(EntityId::new(e), type_name, Window::AllTime)
                .unwrap(),
            1,
            "re-shipping an already-applied segment must NOT double-count entity {e}"
        );
    }

    // A genuinely NEW segment (WAL seqs 4..=5) still advances and applies once,
    // proving the follower is a live replica, not frozen at its restored boundary.
    tx2.send(build_segment(&schema, &[40, 50], 4)).unwrap();
    wait_for_applied(&restored_state, 5);
    for e in [40u64, 50] {
        assert_eq!(
            follower
                .read_windowed_count(EntityId::new(e), type_name, Window::AllTime)
                .unwrap(),
            1,
            "a new post-restart segment must apply once for entity {e}"
        );
    }

    drop(tx2);
    follower.close().unwrap();
}

/// Focused regression guard: `ReplicationState::to/from_checkpoint_bytes` is no
/// longer dead — it round-trips through the durable store on a real reopen.
///
/// This is the narrowest proof of the persisted-high-water-mark half: open a
/// follower, advance the high-water-mark by replicating a segment, clean-shutdown
/// (which persists the HWM), reopen, and assert the HWM is restored (not 0).
#[test]
fn replication_high_water_mark_round_trips_through_clean_restart() {
    let home = TempTidalHome::new().unwrap();
    let schema = make_schema();

    {
        let follower = open_persistent_follower(&home, schema.clone());
        let state = follower.replication_state().clone();
        let (tx, rx) = crossbeam::channel::bounded(8);
        let transport = Arc::new(ChannelTransport { rx });
        follower.start_replication(Arc::clone(&transport)).unwrap();

        tx.send(build_segment(&schema, &[1, 2, 3, 4], 1)).unwrap();
        wait_for_applied(&state, 4);
        assert_eq!(state.applied_seqno(ShardId::SINGLE), Some(4));

        drop(tx);
        follower.close().unwrap(); // shutdown persists the HWM checkpoint.
    }

    let follower = open_persistent_follower(&home, schema);
    assert_eq!(
        follower.replication_state().applied_seqno(ShardId::SINGLE),
        Some(4),
        "applied high-water-mark must be restored after a clean restart, not reset to 0"
    );
    follower.close().unwrap();
}