use betex::{
    book::protocol::command::{Command, CommandKind, Side},
    book::{BookEvent, BookEventEnvelope, EventMetadata},
    config::Config,
    disruptor::Envelope,
    engine::{
        core::EngineBuilder,
        journaler::JournalHandler,
        root::EngineRoot,
        state::{EngineState, Market},
    },
    types::*,
};
mod common;
use common::{
    assert_cursor_beyond_recoverable_head, assert_invalid_tx_framing, assert_recovery_incomplete,
    assert_sequence_gap, is_lmdb_eprem, multi_runner_test_config as test_market_cfg,
};

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

fn temp_wal_path(test_name: &str) -> std::path::PathBuf {
    common::unique_path_under_target("wal", test_name)
}

fn make_wal_event(seq: u64) -> Envelope<BookEventEnvelope> {
    Envelope {
        seq,
        payload: BookEventEnvelope {
            market_id: MarketId(1),
            market_name: String::new(),
            market_seq: seq,
            timestamp: unix_epoch(),
            metadata: EventMetadata::default(),
            event: BookEvent::MarketStateChanged {
                to: betex::book::BookMarketState::Open,
                reason: String::new(),
            },
        },
        response_cb: None,
        tx_id: seq,
        tx_len: 1,
        tx_ix: 0,
    }
}

fn make_wal_event_with_framing(seq: u64, tx_len: u16, tx_ix: u16) -> Envelope<BookEventEnvelope> {
    Envelope {
        tx_len,
        tx_ix,
        ..make_wal_event(seq)
    }
}

fn make_wal_event_for_tx(
    seq: u64,
    tx_id: u64,
    tx_len: u16,
    tx_ix: u16,
) -> Envelope<BookEventEnvelope> {
    Envelope {
        tx_id,
        tx_len,
        tx_ix,
        ..make_wal_event(seq)
    }
}

struct SnapshottingHandler {
    root: Arc<Mutex<EngineRoot>>,
    done_tx: Option<flume::Sender<()>>,
}

impl disrupt_rs::EventHandler<betex::engine::core::EngineEvent> for SnapshottingHandler {
    fn on_event(
        &mut self,
        event: &betex::engine::core::EngineEvent,
        _sequence: disrupt_rs::Sequence,
        end_of_batch: bool,
    ) {
        let mut root = self.root.lock();
        root.assert_next_seq(event.seq);
        root.apply(std::slice::from_ref(&event.payload));
        root.observe_tx_id(event.tx_id);

        if end_of_batch && let Some(tx) = self.done_tx.take() {
            let _ = tx.send(());
        }
    }
}

struct TxIdCaptureHandler {
    first_tx_id: Arc<Mutex<Option<u64>>>,
    done_tx: Option<flume::Sender<()>>,
}

impl disrupt_rs::EventHandler<betex::engine::core::EngineEvent> for TxIdCaptureHandler {
    fn on_event(
        &mut self,
        event: &betex::engine::core::EngineEvent,
        _sequence: disrupt_rs::Sequence,
        end_of_batch: bool,
    ) {
        let mut first = self.first_tx_id.lock();
        if first.is_none() {
            *first = Some(event.tx_id);
        }
        if end_of_batch && let Some(tx) = self.done_tx.take() {
            let _ = tx.send(());
        }
    }
}

struct SeqCaptureHandler {
    seen: Arc<Mutex<Vec<u64>>>,
}

impl disrupt_rs::EventHandler<betex::engine::core::EngineEvent> for SeqCaptureHandler {
    fn on_event(
        &mut self,
        event: &betex::engine::core::EngineEvent,
        _sequence: disrupt_rs::Sequence,
        _end_of_batch: bool,
    ) {
        self.seen.lock().push(event.seq);
    }
}

#[test]
fn tx_id_continues_from_snapshot_when_no_wal_events_after_snapshot() -> anyhow::Result<()> {
    let wal_path = temp_wal_path("tx-id-persists");
    let _ = std::fs::remove_dir_all(&wal_path);

    let cfg = Config::default();
    let market_id = MarketId(1);
    let market_cfg = test_market_cfg(market_id);

    // Projection-style snapshot root that we persist across restarts.
    let mut state = EngineState::new(cfg.clone());
    state.markets.insert(
        market_id,
        Market::from_config_with_cfg(&state.cfg, market_cfg.clone()),
    );
    let snapshot_root = Arc::new(Mutex::new(EngineRoot::new_with(state, 0)));

    // Run 1: submit one tx and produce a snapshot root at the same seq as the WAL.
    let (snap_done_tx, snap_done_rx) = flume::bounded::<()>(1);
    let handler = SnapshottingHandler {
        root: Arc::clone(&snapshot_root),
        done_tx: Some(snap_done_tx),
    };

    let mut engine = match EngineBuilder::with_markets(cfg.clone(), [market_cfg.clone()], &wal_path)
    {
        Ok(b) => b
            .register_handler(
                betex::engine::consumers::ResponseCallbackHandler::default(),
                betex::HandlerRecovery::LiveOnly,
                None,
            )
            .register_handler(handler, betex::HandlerRecovery::LiveOnly, None)
            .build()?,
        Err(e) if is_lmdb_eprem(&e) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(e) => return Err(e),
    };

    let cmd1 = Command {
        correlation_id: Some(CorrelationId(1.to_string())),
        metadata: None,
        market_id,
        kind: CommandKind::PlaceOrder {
            runner_id: RunnerId(1),
            account_id: AccountId::from(1),
            client_order_id: None,
            side: Side::Yes,
            odds: OddsX10000(20000),
            stake: Money(100),
            persistence: betex::book::Persistence::Persist,
            time_in_force: betex::book::TimeInForce::Gtc,
        },
    };

    let handle = engine.handle();
    let resp1 = handle.submit_with_response(cmd1)?;
    let _ = resp1.recv_timeout(Duration::from_secs(2))?;
    snap_done_rx.recv_timeout(Duration::from_secs(2))?;
    engine.stop()?;

    let snap = snapshot_root.lock().clone();
    let prev_next_tx_id = snap.next_tx_id();

    // Run 2: restart from the snapshot root. There are no WAL events after the snapshot's seq,
    // so the engine must not reset tx_id back to 0.
    let (tx_done_tx, tx_done_rx) = flume::bounded::<()>(1);
    let first_tx_id = Arc::new(Mutex::new(None));
    let tx_capture = TxIdCaptureHandler {
        first_tx_id: Arc::clone(&first_tx_id),
        done_tx: Some(tx_done_tx),
    };

    let mut engine = match EngineBuilder::from_root(snap, cfg.clone(), &wal_path) {
        Ok(b) => b
            .register_handler(
                betex::engine::consumers::ResponseCallbackHandler::default(),
                betex::HandlerRecovery::LiveOnly,
                None,
            )
            .register_handler(tx_capture, betex::HandlerRecovery::LiveOnly, None)
            .build()?,
        Err(e) if is_lmdb_eprem(&e) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(e) => return Err(e),
    };

    let cmd2 = Command {
        correlation_id: Some(CorrelationId(2.to_string())),
        metadata: None,
        market_id,
        kind: CommandKind::PlaceOrder {
            runner_id: RunnerId(1),
            account_id: AccountId::from(1),
            client_order_id: None,
            side: Side::No,
            odds: OddsX10000(20000),
            stake: Money(100),
            persistence: betex::book::Persistence::Persist,
            time_in_force: betex::book::TimeInForce::Gtc,
        },
    };

    let handle = engine.handle();
    let resp2 = handle.submit_with_response(cmd2)?;
    let _ = resp2.recv_timeout(Duration::from_secs(2))?;
    tx_done_rx.recv_timeout(Duration::from_secs(2))?;
    engine.stop()?;

    assert_eq!(*first_tx_id.lock(), Some(prev_next_tx_id));

    let _ = std::fs::remove_dir_all(&wal_path);
    Ok(())
}

#[test]
fn handler_recovery_replays_from_explicit_cursor_during_build() -> anyhow::Result<()> {
    let wal_path = temp_wal_path("handler-recovery-from-cursor");
    let _ = std::fs::remove_dir_all(&wal_path);

    let cfg = Config::default();
    let market_id = MarketId(1);
    let market_cfg = test_market_cfg(market_id);

    let mut state = EngineState::new(cfg.clone());
    state.markets.insert(
        market_id,
        Market::from_config_with_cfg(&state.cfg, market_cfg.clone()),
    );
    let snapshot_root = Arc::new(Mutex::new(EngineRoot::new_with(state, 0)));

    let (snap_done_tx, snap_done_rx) = flume::bounded::<()>(1);
    let handler = SnapshottingHandler {
        root: Arc::clone(&snapshot_root),
        done_tx: Some(snap_done_tx),
    };

    let mut engine = match EngineBuilder::with_markets(cfg.clone(), [market_cfg.clone()], &wal_path)
    {
        Ok(b) => b
            .register_handler(
                betex::engine::consumers::ResponseCallbackHandler::default(),
                betex::HandlerRecovery::LiveOnly,
                None,
            )
            .register_handler(handler, betex::HandlerRecovery::LiveOnly, None)
            .build()?,
        Err(e) if is_lmdb_eprem(&e) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(e) => return Err(e),
    };

    let cmd = Command {
        correlation_id: Some(CorrelationId(3.to_string())),
        metadata: None,
        market_id,
        kind: CommandKind::PlaceOrder {
            runner_id: RunnerId(1),
            account_id: AccountId::from(1),
            client_order_id: None,
            side: Side::Yes,
            odds: OddsX10000(20000),
            stake: Money(100),
            persistence: betex::book::Persistence::Persist,
            time_in_force: betex::book::TimeInForce::Gtc,
        },
    };

    let handle = engine.handle();
    let resp = handle.submit_with_response(cmd)?;
    let _ = resp.recv_timeout(Duration::from_secs(2))?;
    snap_done_rx.recv_timeout(Duration::from_secs(2))?;
    engine.stop()?;

    let snap = snapshot_root.lock().clone();
    let seen = Arc::new(Mutex::new(Vec::new()));
    let capture = SeqCaptureHandler {
        seen: Arc::clone(&seen),
    };

    let mut engine = match EngineBuilder::from_root(snap, cfg, &wal_path) {
        Ok(b) => b
            .register_handler(capture, betex::HandlerRecovery::AfterSeq(0), None)
            .build()?,
        Err(e) if is_lmdb_eprem(&e) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(e) => return Err(e),
    };

    assert_eq!(*seen.lock(), vec![1]);

    engine.stop()?;
    let _ = std::fs::remove_dir_all(&wal_path);
    Ok(())
}

#[test]
fn handler_recovery_allows_mid_tx_cursor_during_build() -> anyhow::Result<()> {
    let wal_path = temp_wal_path("handler-recovery-mid-tx-cursor");
    let _ = std::fs::remove_dir_all(&wal_path);

    let cfg = Config::default();

    match JournalHandler::open_engine(&wal_path, cfg.journal.clone()) {
        Ok(journal) => {
            let mut wtxn = journal.write_txn()?;
            journal.put_envelope(&mut wtxn, &make_wal_event_for_tx(1, 77, 3, 0))?;
            journal.put_envelope(&mut wtxn, &make_wal_event_for_tx(2, 77, 3, 1))?;
            journal.put_envelope(&mut wtxn, &make_wal_event_for_tx(3, 77, 3, 2))?;
            journal.put_envelope(&mut wtxn, &make_wal_event(4))?;
            wtxn.commit()?;
        }
        Err(e) if is_lmdb_eprem(&e) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(e) => return Err(e),
    }

    let market_id = MarketId(1);
    let market_cfg = test_market_cfg(market_id);
    let mut state = EngineState::new(cfg.clone());
    state.markets.insert(
        market_id,
        Market::from_config_with_cfg(&state.cfg, market_cfg),
    );
    let snapshot_root = EngineRoot::new_with(state, 4);
    let seen = Arc::new(Mutex::new(Vec::new()));
    let capture = SeqCaptureHandler {
        seen: Arc::clone(&seen),
    };

    let mut engine = match EngineBuilder::from_root(snapshot_root, cfg, &wal_path) {
        Ok(builder) => builder
            .register_handler(capture, betex::HandlerRecovery::AfterSeq(1), None)
            .build()?,
        Err(e) if is_lmdb_eprem(&e) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(e) => return Err(e),
    };

    assert_eq!(*seen.lock(), vec![2, 3, 4]);

    engine.stop()?;
    let _ = std::fs::remove_dir_all(&wal_path);
    Ok(())
}

#[test]
fn handler_recovery_rejects_after_seq_zero_partial_left_edge() -> anyhow::Result<()> {
    let wal_path = temp_wal_path("handler-recovery-after-seq-zero-partial-left-edge");
    let _ = std::fs::remove_dir_all(&wal_path);

    let cfg = Config::default();

    match JournalHandler::open_engine(&wal_path, cfg.journal.clone()) {
        Ok(journal) => {
            let mut wtxn = journal.write_txn()?;
            journal.put_envelope(&mut wtxn, &make_wal_event_for_tx(1, 77, 3, 1))?;
            journal.put_envelope(&mut wtxn, &make_wal_event_for_tx(2, 77, 3, 2))?;
            journal.put_envelope(&mut wtxn, &make_wal_event(3))?;
            wtxn.commit()?;
        }
        Err(e) if is_lmdb_eprem(&e) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(e) => return Err(e),
    }

    let market_id = MarketId(1);
    let market_cfg = test_market_cfg(market_id);
    let mut state = EngineState::new(cfg.clone());
    state.markets.insert(
        market_id,
        Market::from_config_with_cfg(&state.cfg, market_cfg),
    );
    let snapshot_root = EngineRoot::new_with(state, 3);
    let capture = SeqCaptureHandler {
        seen: Arc::new(Mutex::new(Vec::new())),
    };

    let err = match EngineBuilder::from_root(snapshot_root, cfg, &wal_path) {
        Ok(builder) => builder
            .register_handler(capture, betex::HandlerRecovery::AfterSeq(0), None)
            .build()
            .err()
            .expect("partial left edge after seq 0 should fail engine build"),
        Err(e) if is_lmdb_eprem(&e) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(e) => return Err(e),
    };

    assert_sequence_gap(&err, 0, 1);

    let _ = std::fs::remove_dir_all(&wal_path);
    Ok(())
}

#[test]
fn handler_recovery_rejects_gapped_wal_during_build() -> anyhow::Result<()> {
    let wal_path = temp_wal_path("handler-recovery-sequence-gap");
    let _ = std::fs::remove_dir_all(&wal_path);

    let cfg = Config::default();

    match JournalHandler::open_engine(&wal_path, cfg.journal.clone()) {
        Ok(journal) => {
            let mut wtxn = journal.write_txn()?;
            journal.put_envelope(&mut wtxn, &make_wal_event(1))?;
            journal.put_envelope(&mut wtxn, &make_wal_event(3))?;
            wtxn.commit()?;
        }
        Err(e) if is_lmdb_eprem(&e) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(e) => return Err(e),
    }

    let market_id = MarketId(1);
    let market_cfg = test_market_cfg(market_id);
    let mut state = EngineState::new(cfg.clone());
    state.markets.insert(
        market_id,
        Market::from_config_with_cfg(&state.cfg, market_cfg),
    );
    let snapshot_root = EngineRoot::new_with(state, 3);
    let capture = SeqCaptureHandler {
        seen: Arc::new(Mutex::new(Vec::new())),
    };

    let err = match EngineBuilder::from_root(snapshot_root, cfg, &wal_path) {
        Ok(builder) => builder
            .register_handler(capture, betex::HandlerRecovery::AfterSeq(0), None)
            .build()
            .err()
            .expect("gapped WAL should fail engine build"),
        Err(e) if is_lmdb_eprem(&e) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(e) => return Err(e),
    };

    assert_sequence_gap(&err, 2, 3);

    let _ = std::fs::remove_dir_all(&wal_path);
    Ok(())
}

#[test]
fn handler_recovery_rejects_missing_first_batch_after_cursor() -> anyhow::Result<()> {
    let wal_path = temp_wal_path("handler-recovery-after-cursor-gap");
    let _ = std::fs::remove_dir_all(&wal_path);

    let cfg = Config::default();

    match JournalHandler::open_engine(&wal_path, cfg.journal.clone()) {
        Ok(journal) => {
            let mut wtxn = journal.write_txn()?;
            journal.put_envelope(&mut wtxn, &make_wal_event(1))?;
            journal.put_envelope(&mut wtxn, &make_wal_event(2))?;
            journal.put_envelope(&mut wtxn, &make_wal_event(5))?;
            wtxn.commit()?;
        }
        Err(e) if is_lmdb_eprem(&e) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(e) => return Err(e),
    }

    let market_id = MarketId(1);
    let market_cfg = test_market_cfg(market_id);
    let mut state = EngineState::new(cfg.clone());
    state.markets.insert(
        market_id,
        Market::from_config_with_cfg(&state.cfg, market_cfg),
    );
    let snapshot_root = EngineRoot::new_with(state, 2);
    let capture = SeqCaptureHandler {
        seen: Arc::new(Mutex::new(Vec::new())),
    };

    let err = match EngineBuilder::from_root(snapshot_root, cfg, &wal_path) {
        Ok(builder) => builder
            .register_handler(capture, betex::HandlerRecovery::AfterSeq(2), None)
            .build()
            .err()
            .expect("missing first batch after cursor should fail engine build"),
        Err(e) if is_lmdb_eprem(&e) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(e) => return Err(e),
    };

    assert_sequence_gap(&err, 3, 5);

    let _ = std::fs::remove_dir_all(&wal_path);
    Ok(())
}

#[test]
fn handler_recovery_ignores_gap_before_snapshot_cursor() -> anyhow::Result<()> {
    let wal_path = temp_wal_path("handler-recovery-ignore-gap-before-snapshot");
    let _ = std::fs::remove_dir_all(&wal_path);

    let cfg = Config::default();

    match JournalHandler::open_engine(&wal_path, cfg.journal.clone()) {
        Ok(journal) => {
            let mut wtxn = journal.write_txn()?;
            journal.put_envelope(&mut wtxn, &make_wal_event(1))?;
            journal.put_envelope(&mut wtxn, &make_wal_event(3))?;
            journal.put_envelope(&mut wtxn, &make_wal_event(10))?;
            journal.put_envelope(&mut wtxn, &make_wal_event(11))?;
            wtxn.commit()?;
        }
        Err(e) if is_lmdb_eprem(&e) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(e) => return Err(e),
    }

    let market_id = MarketId(1);
    let market_cfg = test_market_cfg(market_id);
    let mut state = EngineState::new(cfg.clone());
    let mut market = Market::from_config_with_cfg(&state.cfg, market_cfg);
    market.last_market_seq = 9;
    state.markets.insert(market_id, market);
    let snapshot_root = EngineRoot::new_with(state, 9);
    let seen = Arc::new(Mutex::new(Vec::new()));
    let capture = SeqCaptureHandler {
        seen: Arc::clone(&seen),
    };

    let mut engine = match EngineBuilder::from_root(snapshot_root, cfg, &wal_path) {
        Ok(builder) => builder
            .register_handler(capture, betex::HandlerRecovery::AfterSeq(9), None)
            .build()?,
        Err(e) if is_lmdb_eprem(&e) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(e) => return Err(e),
    };

    assert_eq!(*seen.lock(), vec![10, 11]);

    engine.stop()?;
    let _ = std::fs::remove_dir_all(&wal_path);
    Ok(())
}

#[test]
fn handler_recovery_allows_cursor_ahead_of_snapshot_within_recoverable_head() -> anyhow::Result<()>
{
    let wal_path = temp_wal_path("handler-recovery-ahead-of-snapshot");
    let _ = std::fs::remove_dir_all(&wal_path);

    let cfg = Config::default();

    match JournalHandler::open_engine(&wal_path, cfg.journal.clone()) {
        Ok(journal) => {
            let mut wtxn = journal.write_txn()?;
            for seq in 1..=12 {
                journal.put_envelope(&mut wtxn, &make_wal_event(seq))?;
            }
            wtxn.commit()?;
        }
        Err(e) if is_lmdb_eprem(&e) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(e) => return Err(e),
    }

    let market_id = MarketId(1);
    let market_cfg = test_market_cfg(market_id);
    let mut state = EngineState::new(cfg.clone());
    let mut market = Market::from_config_with_cfg(&state.cfg, market_cfg.clone());
    market.last_market_seq = 8;
    state.markets.insert(market_id, market);

    let seen = Arc::new(Mutex::new(Vec::new()));
    let mut engine = match EngineBuilder::from_root(EngineRoot::new_with(state, 8), cfg, &wal_path)
    {
        Ok(builder) => builder
            .register_handler(
                SeqCaptureHandler {
                    seen: Arc::clone(&seen),
                },
                betex::HandlerRecovery::AfterSeq(10),
                None,
            )
            .build()?,
        Err(e) if is_lmdb_eprem(&e) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(e) => return Err(e),
    };

    assert_eq!(*seen.lock(), vec![11, 12]);

    engine.stop()?;
    let _ = std::fs::remove_dir_all(&wal_path);
    Ok(())
}

#[test]
fn handler_recovery_rejects_cursor_beyond_recoverable_head() -> anyhow::Result<()> {
    let wal_path = temp_wal_path("handler-recovery-beyond-recoverable-head");
    let _ = std::fs::remove_dir_all(&wal_path);

    let cfg = Config::default();

    match JournalHandler::open_engine(&wal_path, cfg.journal.clone()) {
        Ok(journal) => {
            let mut wtxn = journal.write_txn()?;
            for seq in 1..=12 {
                journal.put_envelope(&mut wtxn, &make_wal_event(seq))?;
            }
            wtxn.commit()?;
        }
        Err(e) if is_lmdb_eprem(&e) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(e) => return Err(e),
    }

    let market_id = MarketId(1);
    let market_cfg = test_market_cfg(market_id);
    let mut state = EngineState::new(cfg.clone());
    let mut market = Market::from_config_with_cfg(&state.cfg, market_cfg.clone());
    market.last_market_seq = 8;
    state.markets.insert(market_id, market);

    let err = match EngineBuilder::from_root(
        EngineRoot::new_with(state.clone(), 8),
        cfg.clone(),
        &wal_path,
    ) {
        Ok(builder) => builder
            .register_handler(
                SeqCaptureHandler {
                    seen: Arc::new(Mutex::new(Vec::new())),
                },
                betex::HandlerRecovery::AfterSeq(13),
                None,
            )
            .build()
            .err()
            .expect("cursor beyond recoverable head should fail engine build"),
        Err(e) if is_lmdb_eprem(&e) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(e) => return Err(e),
    };
    assert_cursor_beyond_recoverable_head(&err, 13, 12);

    let err = match EngineBuilder::from_root(EngineRoot::new_with(state, 8), cfg, &wal_path) {
        Ok(builder) => builder
            .register_handler(
                SeqCaptureHandler {
                    seen: Arc::new(Mutex::new(Vec::new())),
                },
                betex::HandlerRecovery::AfterSeq(8),
                None,
            )
            .register_handler(
                SeqCaptureHandler {
                    seen: Arc::new(Mutex::new(Vec::new())),
                },
                betex::HandlerRecovery::AfterSeq(13),
                None,
            )
            .build()
            .err()
            .expect("any handler beyond recoverable head should fail engine build"),
        Err(e) if is_lmdb_eprem(&e) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(e) => return Err(e),
    };
    assert_cursor_beyond_recoverable_head(&err, 13, 12);

    let _ = std::fs::remove_dir_all(&wal_path);
    Ok(())
}

#[test]
fn handler_recovery_rejects_invalid_tx_framing_during_build() -> anyhow::Result<()> {
    let wal_path = temp_wal_path("handler-recovery-invalid-tx-framing");
    let _ = std::fs::remove_dir_all(&wal_path);

    let cfg = Config::default();

    match JournalHandler::open_engine(&wal_path, cfg.journal.clone()) {
        Ok(journal) => {
            let mut wtxn = journal.write_txn()?;
            journal.put_envelope(&mut wtxn, &make_wal_event_with_framing(1, 1, 1))?;
            wtxn.commit()?;
        }
        Err(e) if is_lmdb_eprem(&e) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(e) => return Err(e),
    }

    let market_id = MarketId(1);
    let market_cfg = test_market_cfg(market_id);
    let mut state = EngineState::new(cfg.clone());
    state.markets.insert(
        market_id,
        Market::from_config_with_cfg(&state.cfg, market_cfg),
    );
    let snapshot_root = EngineRoot::new_with(state, 0);
    let capture = SeqCaptureHandler {
        seen: Arc::new(Mutex::new(Vec::new())),
    };

    let err = match EngineBuilder::from_root(snapshot_root, cfg, &wal_path) {
        Ok(builder) => builder
            .register_handler(capture, betex::HandlerRecovery::AfterSeq(0), None)
            .build()
            .err()
            .expect("invalid tx framing should fail engine build"),
        Err(e) if is_lmdb_eprem(&e) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(e) => return Err(e),
    };

    assert_invalid_tx_framing(&err, 1, 1, 1, 1);

    let _ = std::fs::remove_dir_all(&wal_path);
    Ok(())
}

#[test]
fn handler_recovery_requires_retained_wal_through_snapshot() -> anyhow::Result<()> {
    let wal_path = temp_wal_path("handler-recovery-requires-snapshot-head");
    let _ = std::fs::remove_dir_all(&wal_path);

    let cfg = Config::default();

    match JournalHandler::open_engine(&wal_path, cfg.journal.clone()) {
        Ok(mut journal) => {
            let mut wtxn = journal.write_txn()?;
            for seq in 1..=10 {
                journal.put_envelope(&mut wtxn, &make_wal_event(seq))?;
            }
            wtxn.commit()?;
            journal.rotate(11)?;
            let moved = journal.prune_segments_before(10)?;
            assert_eq!(moved.len(), 1);
        }
        Err(e) if is_lmdb_eprem(&e) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(e) => return Err(e),
    }

    let market_id = MarketId(1);
    let market_cfg = test_market_cfg(market_id);
    let mut state = EngineState::new(cfg.clone());
    state.markets.insert(
        market_id,
        Market::from_config_with_cfg(&state.cfg, market_cfg.clone()),
    );

    let err = match EngineBuilder::from_root(
        EngineRoot::new_with(state.clone(), 10),
        cfg.clone(),
        &wal_path,
    ) {
        Ok(builder) => builder
            .register_handler(
                SeqCaptureHandler {
                    seen: Arc::new(Mutex::new(Vec::new())),
                },
                betex::HandlerRecovery::AfterSeq(0),
                None,
            )
            .build()
            .err()
            .expect("missing retained WAL before snapshot head should fail engine build"),
        Err(e) if is_lmdb_eprem(&e) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(e) => return Err(e),
    };

    assert_recovery_incomplete(&err, 10, 0);

    let mut engine = match EngineBuilder::from_root(EngineRoot::new_with(state, 10), cfg, &wal_path)
    {
        Ok(builder) => builder
            .register_handler(
                SeqCaptureHandler {
                    seen: Arc::new(Mutex::new(Vec::new())),
                },
                betex::HandlerRecovery::AfterSeq(10),
                None,
            )
            .build()?,
        Err(e) if is_lmdb_eprem(&e) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(e) => return Err(e),
    };

    engine.stop()?;
    let _ = std::fs::remove_dir_all(&wal_path);
    Ok(())
}

#[test]
fn handler_recovery_prunes_obsolete_segments_before_validation() -> anyhow::Result<()> {
    let wal_path = temp_wal_path("handler-recovery-prune-before-validate");
    let _ = std::fs::remove_dir_all(&wal_path);

    let setup_cfg = Config::default();
    match JournalHandler::open_engine(&wal_path, setup_cfg.journal.clone()) {
        Ok(mut journal) => {
            let mut wtxn = journal.write_txn()?;
            journal.put_envelope(&mut wtxn, &make_wal_event(1))?;
            wtxn.commit()?;
            journal.rotate(10)?;
            let mut wtxn = journal.write_txn()?;
            journal.put_envelope(&mut wtxn, &make_wal_event(10))?;
            journal.put_envelope(&mut wtxn, &make_wal_event(11))?;
            wtxn.commit()?;
        }
        Err(e) if is_lmdb_eprem(&e) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(e) => return Err(e),
    }

    let mut cfg = Config::default();
    cfg.journal.validate_on_startup = true;
    cfg.journal.prune_before_seq_on_startup = Some(9);
    cfg.journal.clear_trash_on_startup = true;

    let market_id = MarketId(1);
    let market_cfg = test_market_cfg(market_id);
    let mut state = EngineState::new(cfg.clone());
    let mut market = Market::from_config_with_cfg(&state.cfg, market_cfg);
    market.last_market_seq = 9;
    state.markets.insert(market_id, market);

    let seen = Arc::new(Mutex::new(Vec::new()));
    let capture = SeqCaptureHandler {
        seen: Arc::clone(&seen),
    };

    let mut engine = match EngineBuilder::from_root(EngineRoot::new_with(state, 9), cfg, &wal_path)
    {
        Ok(builder) => builder
            .register_handler(capture, betex::HandlerRecovery::AfterSeq(9), None)
            .build()?,
        Err(e) if is_lmdb_eprem(&e) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(e) => return Err(e),
    };

    assert_eq!(*seen.lock(), vec![10, 11]);

    engine.stop()?;
    let _ = std::fs::remove_dir_all(&wal_path);
    Ok(())
}

#[test]
fn handler_recovery_replays_before_snapshot_with_startup_pruning() -> anyhow::Result<()> {
    let wal_path = temp_wal_path("handler-recovery-startup-pruning");
    let _ = std::fs::remove_dir_all(&wal_path);

    let setup_cfg = Config::default();
    match JournalHandler::open_engine(&wal_path, setup_cfg.journal.clone()) {
        Ok(mut journal) => {
            let mut wtxn = journal.write_txn()?;
            for seq in 1..=10 {
                journal.put_envelope(&mut wtxn, &make_wal_event(seq))?;
            }
            wtxn.commit()?;
            journal.rotate(11)?;
        }
        Err(e) if is_lmdb_eprem(&e) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(e) => return Err(e),
    }

    let mut cfg = Config::default();
    cfg.journal.prune_before_seq_on_startup = Some(10);
    cfg.journal.clear_trash_on_startup = true;

    let market_id = MarketId(1);
    let market_cfg = test_market_cfg(market_id);
    let mut state = EngineState::new(cfg.clone());
    state.markets.insert(
        market_id,
        Market::from_config_with_cfg(&state.cfg, market_cfg),
    );

    let seen = Arc::new(Mutex::new(Vec::new()));
    let capture = SeqCaptureHandler {
        seen: Arc::clone(&seen),
    };

    let mut engine = match EngineBuilder::from_root(EngineRoot::new_with(state, 10), cfg, &wal_path)
    {
        Ok(builder) => builder
            .register_handler(capture, betex::HandlerRecovery::AfterSeq(0), None)
            .build()?,
        Err(e) if is_lmdb_eprem(&e) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(e) => return Err(e),
    };

    assert_eq!(*seen.lock(), (1..=10).collect::<Vec<_>>());

    engine.stop()?;

    let trash = wal_path.join("trash");
    if trash.exists() {
        assert_eq!(std::fs::read_dir(&trash)?.count(), 0);
    }

    let _ = std::fs::remove_dir_all(&wal_path);
    Ok(())
}
