mod common;

use betex::{
    book::{
        BookEvent, BookEventEnvelope, BookMarketState, EventMetadata,
        protocol::{Command, CommandKind, MarketState},
    },
    config::{Config, JournalConfig},
    disruptor::Envelope,
    engine::{
        core::{EngineBuilder, EngineEvent},
        journal::JournalReader,
        journaler::JournalHandler,
    },
    types::{MarketId, unix_epoch},
};
use common::{
    assert_corrupt_record, assert_incomplete_tx_after_cursor, assert_invalid_tx_framing,
    assert_schema_mismatch, assert_sequence_gap, two_runner_market_config,
};

const ENGINE_WAL_DB_NAME: &str = "walw";

/// Create a test envelope with the given seq, tx_id, tx_len, tx_ix.
fn make_envelope(seq: u64, tx_id: u64, tx_len: u16, tx_ix: u16) -> 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: BookMarketState::Open,
                reason: String::new(),
            },
        },
        response_cb: None,
        tx_id,
        tx_len,
        tx_ix,
    }
}

/// Create a single-event transaction envelope at the given seq.
fn make_single(seq: u64) -> Envelope<BookEventEnvelope> {
    make_envelope(seq, seq, 1, 0)
}

fn test_config() -> JournalConfig {
    JournalConfig {
        segment_map_size_bytes: 8 * 1024 * 1024,
        max_segment_events: 5_000,
        validate_on_startup: false, // we validate explicitly in tests
        prune_before_seq_on_startup: None,
        clear_trash_on_startup: true,
    }
}

fn test_path(name: &str) -> std::path::PathBuf {
    common::unique_path_under_target("journal", name)
}

fn trash_path(root: &std::path::Path) -> std::path::PathBuf {
    root.join("trash")
}

fn list_segment_dirs_for_test(root: &std::path::Path) -> Vec<String> {
    let mut dirs = Vec::new();
    if let Ok(read_dir) = std::fs::read_dir(root) {
        for entry in read_dir.flatten() {
            if entry.file_type().map(|ty| ty.is_dir()).unwrap_or(false) {
                let name = entry.file_name().to_string_lossy().into_owned();
                if name.starts_with("seg_") {
                    dirs.push(name);
                }
            }
        }
    }
    dirs.sort();
    dirs
}

fn write_raw_segment_event(path: &std::path::Path, dir_name: &str, seq: u64) {
    write_raw_segment_event_with(path, dir_name, "wal", seq, 1);
}

fn write_raw_segment_bytes_with(
    path: &std::path::Path,
    dir_name: &str,
    db_name: &str,
    seq: u64,
    schema_version: u64,
    bytes: &[u8],
) {
    let seg_dir = path.join(dir_name);
    std::fs::create_dir_all(&seg_dir).unwrap();

    let cfg = test_config();
    let env = unsafe {
        heed::EnvOpenOptions::new()
            .map_size(cfg.segment_map_size_bytes)
            .max_dbs(8)
            .open(&seg_dir)
            .unwrap()
    };

    let mut wtxn = env.write_txn().unwrap();
    let log: heed::Database<heed::types::U64<heed::byteorder::BigEndian>, heed::types::Bytes> =
        env.create_database(&mut wtxn, Some(db_name)).unwrap();
    let meta: heed::Database<heed::types::Bytes, heed::types::U64<heed::byteorder::BigEndian>> =
        env.create_database(&mut wtxn, Some("meta")).unwrap();
    meta.put(&mut wtxn, b"schema_version", &schema_version)
        .unwrap();
    log.put(&mut wtxn, &seq, bytes).unwrap();
    wtxn.commit().unwrap();
}

fn write_raw_segment_event_with(
    path: &std::path::Path,
    dir_name: &str,
    db_name: &str,
    seq: u64,
    schema_version: u64,
) {
    let envelope = make_single(seq);
    let bytes =
        rkyv::api::high::to_bytes::<betex::disruptor::traits::RkyvError>(&envelope).unwrap();
    write_raw_segment_bytes_with(
        path,
        dir_name,
        db_name,
        seq,
        schema_version,
        bytes.as_slice(),
    );
}

struct NoopHandler;

impl disrupt_rs::EventHandler<EngineEvent> for NoopHandler {
    fn on_event(
        &mut self,
        _event: &EngineEvent,
        _sequence: disrupt_rs::Sequence,
        _end_of_batch: bool,
    ) {
    }
}

fn submit_state_change(
    handle: &betex::engine::core::EngineHandler,
    market_id: MarketId,
    state: MarketState,
) {
    handle
        .submit(Command {
            correlation_id: None,
            metadata: None,
            market_id,
            kind: CommandKind::SetMarketState { state },
        })
        .expect("submit state change");
}

// ============================================================================
// Tests
// ============================================================================

#[test]
fn single_segment_write_and_recover() {
    let path = test_path("single_seg_write_recover");
    if common::is_lmdb_unsupported("") {
        return;
    }

    {
        let journal = JournalHandler::open(&path, "wal", test_config()).unwrap();
        let mut wtxn = journal.write_txn().unwrap();
        for seq in 0..10 {
            journal.put_envelope(&mut wtxn, &make_single(seq)).unwrap();
        }
        wtxn.commit().unwrap();
    }

    // Reopen and recover
    let journal = JournalHandler::open(&path, "wal", test_config()).unwrap();
    let mut recovered = Vec::new();
    journal
        .recover_from::<BookEventEnvelope, _>(None, |env, eob| {
            let seq: u64 = env.seq.into();
            recovered.push((seq, eob));
            Ok(())
        })
        .unwrap();

    assert_eq!(recovered.len(), 10);
    for (i, (seq, _)) in recovered.iter().enumerate() {
        assert_eq!(*seq, i as u64);
    }
    // Last event should have eob=true
    assert!(recovered.last().unwrap().1);
}

#[test]
fn journal_reader_head_seq_ignores_incomplete_tail() {
    let path = test_path("journal_reader_head_seq");

    {
        let journal = JournalHandler::open(&path, ENGINE_WAL_DB_NAME, test_config()).unwrap();
        let mut wtxn = journal.write_txn().unwrap();
        journal.put_envelope(&mut wtxn, &make_single(1)).unwrap();
        journal
            .put_envelope(&mut wtxn, &make_envelope(2, 2, 2, 0))
            .unwrap();
        wtxn.commit().unwrap();
    }

    let reader = JournalReader::open(&path, test_config()).unwrap();
    assert_eq!(reader.head_seq().unwrap(), Some(1));

    let mut recovered = Vec::new();
    reader
        .recover_from(None, |event, end_of_batch| {
            recovered.push((
                event.seq,
                event.tx_id,
                event.tx_len,
                event.tx_ix,
                event.payload.market_seq,
                event.response_cb.is_none(),
                end_of_batch,
            ));
            Ok(())
        })
        .unwrap();

    assert_eq!(recovered, vec![(1, 1, 1, 0, 1, true, true)]);
}

#[test]
fn journal_reader_head_seq_allows_pruned_prefix() {
    let path = test_path("journal_reader_head_seq_pruned_prefix");

    {
        let mut journal = JournalHandler::open(&path, ENGINE_WAL_DB_NAME, test_config()).unwrap();

        let mut wtxn = journal.write_txn().unwrap();
        journal.put_envelope(&mut wtxn, &make_single(1)).unwrap();
        wtxn.commit().unwrap();

        journal.rotate(2).unwrap();
        let mut wtxn = journal.write_txn().unwrap();
        journal.put_envelope(&mut wtxn, &make_single(2)).unwrap();
        wtxn.commit().unwrap();

        journal.rotate(3).unwrap();
        let mut wtxn = journal.write_txn().unwrap();
        journal.put_envelope(&mut wtxn, &make_single(3)).unwrap();
        wtxn.commit().unwrap();
    }

    let startup_cfg = JournalConfig {
        prune_before_seq_on_startup: Some(2),
        clear_trash_on_startup: false,
        ..test_config()
    };
    {
        let mut journal = JournalHandler::open(&path, ENGINE_WAL_DB_NAME, startup_cfg).unwrap();
        journal.apply_startup_maintenance().unwrap();
    }

    let reader = JournalReader::open(&path, test_config()).unwrap();
    assert_eq!(reader.head_seq().unwrap(), Some(3));
}

#[test]
fn journal_reader_recover_range_skips_partial_edges() {
    let path = test_path("journal_reader_recover_range");

    {
        let journal = JournalHandler::open(&path, ENGINE_WAL_DB_NAME, test_config()).unwrap();
        let mut wtxn = journal.write_txn().unwrap();
        journal
            .put_envelope(&mut wtxn, &make_envelope(1, 10, 2, 0))
            .unwrap();
        journal
            .put_envelope(&mut wtxn, &make_envelope(2, 10, 2, 1))
            .unwrap();
        journal.put_envelope(&mut wtxn, &make_single(3)).unwrap();
        journal
            .put_envelope(&mut wtxn, &make_envelope(4, 11, 2, 0))
            .unwrap();
        journal
            .put_envelope(&mut wtxn, &make_envelope(5, 11, 2, 1))
            .unwrap();
        wtxn.commit().unwrap();
    }

    let reader = JournalReader::open(&path, test_config()).unwrap();
    let mut recovered = Vec::new();
    reader
        .recover_range(Some(1), 4, |event, end_of_batch| {
            recovered.push((event.seq, event.tx_id, end_of_batch));
            Ok(())
        })
        .unwrap();

    assert_eq!(recovered, vec![(3, 3, true)]);
}

#[test]
fn journal_reader_recover_range_rejects_left_edge_gap() {
    let path = test_path("journal_reader_recover_range_left_edge_gap");

    {
        let journal = JournalHandler::open(&path, ENGINE_WAL_DB_NAME, test_config()).unwrap();
        let mut wtxn = journal.write_txn().unwrap();
        journal.put_envelope(&mut wtxn, &make_single(5)).unwrap();
        wtxn.commit().unwrap();
    }

    let reader = JournalReader::open(&path, test_config()).unwrap();
    let err = reader
        .recover_range(Some(1), 5, |_event, _end_of_batch| Ok(()))
        .unwrap_err();
    assert_sequence_gap(&err, 2, 5);
}

#[test]
fn journal_reader_recover_from_allows_mid_tx_cursor() {
    let path = test_path("journal_reader_recover_from_mid_tx");

    {
        let journal = JournalHandler::open(&path, ENGINE_WAL_DB_NAME, test_config()).unwrap();
        let mut wtxn = journal.write_txn().unwrap();
        journal
            .put_envelope(&mut wtxn, &make_envelope(1, 10, 3, 0))
            .unwrap();
        journal
            .put_envelope(&mut wtxn, &make_envelope(2, 10, 3, 1))
            .unwrap();
        journal
            .put_envelope(&mut wtxn, &make_envelope(3, 10, 3, 2))
            .unwrap();
        journal.put_envelope(&mut wtxn, &make_single(4)).unwrap();
        wtxn.commit().unwrap();
    }

    let reader = JournalReader::open(&path, test_config()).unwrap();
    let mut recovered = Vec::new();
    reader
        .recover_from(Some(1), |event, end_of_batch| {
            recovered.push((event.seq, event.tx_id, event.tx_ix, end_of_batch));
            Ok(())
        })
        .unwrap();

    assert_eq!(
        recovered,
        vec![(2, 10, 1, false), (3, 10, 2, false), (4, 4, 0, true)]
    );
}

#[test]
fn journal_reader_recover_from_ignores_gap_before_cursor() {
    let path = test_path("journal_reader_ignore_gap_before_cursor");

    {
        let journal = JournalHandler::open(&path, ENGINE_WAL_DB_NAME, test_config()).unwrap();
        let mut wtxn = journal.write_txn().unwrap();
        journal.put_envelope(&mut wtxn, &make_single(1)).unwrap();
        journal.put_envelope(&mut wtxn, &make_single(3)).unwrap();
        journal.put_envelope(&mut wtxn, &make_single(10)).unwrap();
        wtxn.commit().unwrap();
    }

    let reader = JournalReader::open(&path, test_config()).unwrap();
    let mut recovered = Vec::new();
    reader
        .recover_from(Some(9), |event, end_of_batch| {
            recovered.push((event.seq, end_of_batch));
            Ok(())
        })
        .unwrap();

    assert_eq!(recovered, vec![(10, true)]);
}

#[test]
fn journal_reader_rejects_gap_after_partial_left_edge() {
    let path = test_path("journal_reader_partial_left_edge_gap");

    {
        let journal = JournalHandler::open(&path, ENGINE_WAL_DB_NAME, test_config()).unwrap();
        let mut wtxn = journal.write_txn().unwrap();
        journal
            .put_envelope(&mut wtxn, &make_envelope(2, 10, 2, 1))
            .unwrap();
        journal.put_envelope(&mut wtxn, &make_single(4)).unwrap();
        wtxn.commit().unwrap();
    }

    let reader = JournalReader::open(&path, test_config()).unwrap();

    let head_err = reader.head_seq().unwrap_err();
    assert_sequence_gap(&head_err, 1, 2);

    let recover_err = reader
        .recover_from(None, |_event, _end_of_batch| Ok(()))
        .unwrap_err();
    assert_sequence_gap(&recover_err, 1, 2);

    let range_err = reader
        .recover_range(Some(1), 4, |_event, _end_of_batch| Ok(()))
        .unwrap_err();
    assert_sequence_gap(&range_err, 3, 4);
}

#[test]
fn journal_reader_rejects_complete_partial_left_edge_for_exact_reads() {
    let path = test_path("journal_reader_partial_left_edge_exact");

    {
        let journal = JournalHandler::open(&path, ENGINE_WAL_DB_NAME, test_config()).unwrap();
        let mut wtxn = journal.write_txn().unwrap();
        journal
            .put_envelope(&mut wtxn, &make_envelope(2, 10, 3, 1))
            .unwrap();
        journal
            .put_envelope(&mut wtxn, &make_envelope(3, 10, 3, 2))
            .unwrap();
        journal.put_envelope(&mut wtxn, &make_single(4)).unwrap();
        wtxn.commit().unwrap();
    }

    let reader = JournalReader::open(&path, test_config()).unwrap();

    let head_err = reader.head_seq().unwrap_err();
    assert_sequence_gap(&head_err, 1, 2);

    let recover_err = reader
        .recover_from(None, |_event, _end_of_batch| Ok(()))
        .unwrap_err();
    assert_sequence_gap(&recover_err, 1, 2);

    let mut recovered = Vec::new();
    reader
        .recover_range(None, 4, |event, end_of_batch| {
            recovered.push((event.seq, end_of_batch));
            Ok(())
        })
        .unwrap();
    assert_eq!(recovered, vec![(4, true)]);
}

#[test]
fn journal_reader_recover_from_rejects_after_seq_zero_partial_left_edge() {
    let path = test_path("journal_reader_after_seq_zero_partial_left_edge");

    {
        let journal = JournalHandler::open(&path, ENGINE_WAL_DB_NAME, test_config()).unwrap();
        let mut wtxn = journal.write_txn().unwrap();
        journal
            .put_envelope(&mut wtxn, &make_envelope(1, 10, 3, 1))
            .unwrap();
        journal
            .put_envelope(&mut wtxn, &make_envelope(2, 10, 3, 2))
            .unwrap();
        journal.put_envelope(&mut wtxn, &make_single(3)).unwrap();
        wtxn.commit().unwrap();
    }

    let reader = JournalReader::open(&path, test_config()).unwrap();
    let err = reader
        .recover_from(Some(0), |_event, _end_of_batch| Ok(()))
        .unwrap_err();
    assert_sequence_gap(&err, 0, 1);
}

#[test]
fn journal_reader_recover_from_rejects_missing_first_batch_after_cursor() {
    let path = test_path("journal_reader_recover_from_after_cursor_gap");

    {
        let journal = JournalHandler::open(&path, ENGINE_WAL_DB_NAME, test_config()).unwrap();
        let mut wtxn = journal.write_txn().unwrap();
        journal
            .put_envelope(&mut wtxn, &make_envelope(1, 10, 2, 0))
            .unwrap();
        journal
            .put_envelope(&mut wtxn, &make_envelope(2, 10, 2, 1))
            .unwrap();
        journal.put_envelope(&mut wtxn, &make_single(5)).unwrap();
        wtxn.commit().unwrap();
    }

    let reader = JournalReader::open(&path, test_config()).unwrap();
    let err = reader
        .recover_from(Some(2), |_event, _end_of_batch| Ok(()))
        .unwrap_err();
    assert_sequence_gap(&err, 3, 5);
}

#[test]
fn journal_reader_recover_from_rejects_broken_cursor_with_later_records() {
    let path = test_path("journal_reader_recover_from_broken_cursor");

    {
        let journal = JournalHandler::open(&path, ENGINE_WAL_DB_NAME, test_config()).unwrap();
        let mut wtxn = journal.write_txn().unwrap();
        journal
            .put_envelope(&mut wtxn, &make_envelope(1, 10, 3, 0))
            .unwrap();
        journal.put_envelope(&mut wtxn, &make_single(2)).unwrap();
        wtxn.commit().unwrap();
    }

    let reader = JournalReader::open(&path, test_config()).unwrap();
    let err = reader
        .recover_from(Some(1), |_event, _end_of_batch| Ok(()))
        .unwrap_err();
    assert_incomplete_tx_after_cursor(&err, 1, 10, 2);
}

#[test]
fn journal_reader_recover_from_rejects_invalid_cursor_even_with_later_records() {
    let path = test_path("journal_reader_recover_from_invalid_cursor");

    {
        let journal = JournalHandler::open(&path, ENGINE_WAL_DB_NAME, test_config()).unwrap();
        let mut wtxn = journal.write_txn().unwrap();
        journal
            .put_envelope(&mut wtxn, &make_envelope(1, 10, 1, 1))
            .unwrap();
        journal.put_envelope(&mut wtxn, &make_single(2)).unwrap();
        wtxn.commit().unwrap();
    }

    let reader = JournalReader::open(&path, test_config()).unwrap();
    let err = reader
        .recover_from(Some(1), |_event, _end_of_batch| Ok(()))
        .unwrap_err();
    assert_invalid_tx_framing(&err, 1, 10, 1, 1);
}

#[test]
fn journal_reader_recover_from_rejects_cursor_inside_incomplete_tail() {
    let path = test_path("journal_reader_recover_from_incomplete_tail");

    {
        let journal = JournalHandler::open(&path, ENGINE_WAL_DB_NAME, test_config()).unwrap();
        let mut wtxn = journal.write_txn().unwrap();
        for seq in 1..=8 {
            journal.put_envelope(&mut wtxn, &make_single(seq)).unwrap();
        }
        journal
            .put_envelope(&mut wtxn, &make_envelope(9, 99, 3, 0))
            .unwrap();
        journal
            .put_envelope(&mut wtxn, &make_envelope(10, 99, 3, 1))
            .unwrap();
        wtxn.commit().unwrap();
    }

    let reader = JournalReader::open(&path, test_config()).unwrap();
    let err = reader
        .recover_from(Some(10), |_event, _end_of_batch| Ok(()))
        .unwrap_err();
    assert_incomplete_tx_after_cursor(&err, 10, 99, 11);
}

#[test]
fn journal_reader_rejects_sequence_gaps() {
    let path = test_path("journal_reader_sequence_gap");

    {
        let journal = JournalHandler::open(&path, ENGINE_WAL_DB_NAME, test_config()).unwrap();
        let mut wtxn = journal.write_txn().unwrap();
        journal.put_envelope(&mut wtxn, &make_single(1)).unwrap();
        journal.put_envelope(&mut wtxn, &make_single(3)).unwrap();
        wtxn.commit().unwrap();
    }

    let reader = JournalReader::open(&path, test_config()).unwrap();
    let head_err = reader.head_seq().unwrap_err();
    assert_sequence_gap(&head_err, 2, 3);

    let recover_err = reader
        .recover_from(None, |_event, _end_of_batch| Ok(()))
        .unwrap_err();
    assert_sequence_gap(&recover_err, 2, 3);
}

#[test]
fn journal_reader_rejects_schema_mismatch() {
    let path = test_path("journal_reader_schema_mismatch");

    write_raw_segment_event_with(&path, "seg_00000000000000000000", ENGINE_WAL_DB_NAME, 1, 2);

    let reader = JournalReader::open(&path, test_config()).unwrap();
    let err = reader.head_seq().unwrap_err();
    assert_schema_mismatch(&err, 1, 2);
}

#[test]
fn journal_reader_rejects_invalid_tx_framing_across_public_reads() {
    let path = test_path("journal_reader_invalid_tx_framing");

    {
        let journal = JournalHandler::open(&path, ENGINE_WAL_DB_NAME, test_config()).unwrap();
        let mut wtxn = journal.write_txn().unwrap();
        journal
            .put_envelope(&mut wtxn, &make_envelope(1, 55, 1, 1))
            .unwrap();
        wtxn.commit().unwrap();
    }

    let reader = JournalReader::open(&path, test_config()).unwrap();

    let head_err = reader.head_seq().unwrap_err();
    assert_invalid_tx_framing(&head_err, 1, 55, 1, 1);

    let recover_err = reader
        .recover_from(None, |_event, _end_of_batch| Ok(()))
        .unwrap_err();
    assert_invalid_tx_framing(&recover_err, 1, 55, 1, 1);

    let range_err = reader
        .recover_range(None, 1, |_event, _end_of_batch| Ok(()))
        .unwrap_err();
    assert_invalid_tx_framing(&range_err, 1, 55, 1, 1);
}

#[test]
fn journal_reader_rejects_corrupt_record_across_public_reads() {
    let path = test_path("journal_reader_corrupt_record");

    write_raw_segment_bytes_with(
        &path,
        "seg_00000000000000000000",
        ENGINE_WAL_DB_NAME,
        1,
        1,
        &[0xFF, 0x00, 0x12],
    );

    let reader = JournalReader::open(&path, test_config()).unwrap();

    let head_err = reader.head_seq().unwrap_err();
    assert_corrupt_record(&head_err, 1);

    let recover_err = reader
        .recover_from(None, |_event, _end_of_batch| Ok(()))
        .unwrap_err();
    assert_corrupt_record(&recover_err, 1);

    let range_err = reader
        .recover_range(None, 1, |_event, _end_of_batch| Ok(()))
        .unwrap_err();
    assert_corrupt_record(&range_err, 1);
}

#[test]
fn journal_reader_open_is_side_effect_free() {
    let path = test_path("journal_reader_side_effect_free");
    let _ = std::fs::remove_dir_all(&path);

    {
        let mut journal = JournalHandler::open(&path, ENGINE_WAL_DB_NAME, test_config()).unwrap();
        let mut wtxn = journal.write_txn().unwrap();
        journal.put_envelope(&mut wtxn, &make_single(1)).unwrap();
        wtxn.commit().unwrap();
        journal.rotate(2).unwrap();
        let mut wtxn = journal.write_txn().unwrap();
        journal.put_envelope(&mut wtxn, &make_single(2)).unwrap();
        wtxn.commit().unwrap();
    }

    let trash = trash_path(&path);
    std::fs::create_dir_all(&trash).unwrap();
    let trash_marker = trash.join("keep-me");
    std::fs::write(&trash_marker, b"x").unwrap();

    let reader_cfg = JournalConfig {
        prune_before_seq_on_startup: Some(1),
        clear_trash_on_startup: true,
        ..test_config()
    };
    let reader = JournalReader::open(&path, reader_cfg).unwrap();
    assert_eq!(reader.head_seq().unwrap(), Some(2));

    let segments = list_segment_dirs_for_test(&path);
    assert_eq!(segments.len(), 2);
    assert!(trash_marker.exists());
}

#[test]
fn multi_segment_recovery_across_boundaries() {
    let path = test_path("multi_seg_recovery");

    let mut journal = JournalHandler::open(&path, "wal", test_config()).unwrap();

    // Write 5 events to segment 0
    {
        let mut wtxn = journal.write_txn().unwrap();
        for seq in 0..5 {
            journal.put_envelope(&mut wtxn, &make_single(seq)).unwrap();
        }
        wtxn.commit().unwrap();
    }

    // Rotate
    journal.rotate(5).unwrap();

    // Write 5 more events to segment 1
    {
        let mut wtxn = journal.write_txn().unwrap();
        for seq in 5..10 {
            journal.put_envelope(&mut wtxn, &make_single(seq)).unwrap();
        }
        wtxn.commit().unwrap();
    }

    // Recover all
    let mut recovered = Vec::new();
    journal
        .recover_from::<BookEventEnvelope, _>(None, |env, _eob| {
            let seq: u64 = env.seq.into();
            recovered.push(seq);
            Ok(())
        })
        .unwrap();

    assert_eq!(recovered.len(), 10);
    for (i, seq) in recovered.iter().enumerate() {
        assert_eq!(*seq, i as u64);
    }
}

#[test]
fn rotation_mid_stream_contiguous() {
    let path = test_path("rotation_contiguous");

    let mut journal = JournalHandler::open(&path, "wal", test_config()).unwrap();

    // Write events 0..5 to segment 0
    {
        let mut wtxn = journal.write_txn().unwrap();
        for seq in 0..5 {
            journal.put_envelope(&mut wtxn, &make_single(seq)).unwrap();
        }
        wtxn.commit().unwrap();
    }

    journal.rotate(5).unwrap();

    // Write events 5..10 to segment 1
    {
        let mut wtxn = journal.write_txn().unwrap();
        for seq in 5..10 {
            journal.put_envelope(&mut wtxn, &make_single(seq)).unwrap();
        }
        wtxn.commit().unwrap();
    }

    let segments = journal.list_segments().unwrap();
    assert_eq!(segments.len(), 2);
    assert_eq!(segments[0].start_seq, Some(0));
    assert_eq!(segments[0].end_seq, Some(4));
    assert_eq!(segments[1].start_seq, Some(5));
    assert_eq!(segments[1].end_seq, Some(9));

    // Validate contiguity
    journal.validate_segments().unwrap();
}

#[test]
fn empty_trailing_segment_after_crash() {
    let path = test_path("empty_trailing");

    // Create segment 0 with data, then rotate (creating empty segment 1)
    {
        let mut journal = JournalHandler::open(&path, "wal", test_config()).unwrap();
        let mut wtxn = journal.write_txn().unwrap();
        for seq in 0..5 {
            journal.put_envelope(&mut wtxn, &make_single(seq)).unwrap();
        }
        wtxn.commit().unwrap();
        journal.rotate(5).unwrap();
        // Don't write anything to the new segment — simulate crash
    }

    // Reopen
    let journal = JournalHandler::open(&path, "wal", test_config()).unwrap();

    // Recover should get events from segment 0
    let mut recovered = Vec::new();
    journal
        .recover_from::<BookEventEnvelope, _>(None, |env, _eob| {
            let seq: u64 = env.seq.into();
            recovered.push(seq);
            Ok(())
        })
        .unwrap();

    assert_eq!(recovered.len(), 5);

    // Validate should pass (empty trailing segment is OK)
    journal.validate_segments().unwrap();
}

#[test]
fn recovery_from_mid_segment() {
    let path = test_path("recovery_mid_seg");

    let mut journal = JournalHandler::open(&path, "wal", test_config()).unwrap();

    // Write 5 events to segment 0
    {
        let mut wtxn = journal.write_txn().unwrap();
        for seq in 0..5 {
            journal.put_envelope(&mut wtxn, &make_single(seq)).unwrap();
        }
        wtxn.commit().unwrap();
    }

    journal.rotate(5).unwrap();

    // Write 5 events to segment 1
    {
        let mut wtxn = journal.write_txn().unwrap();
        for seq in 5..10 {
            journal.put_envelope(&mut wtxn, &make_single(seq)).unwrap();
        }
        wtxn.commit().unwrap();
    }

    // Recover from seq 7 (should get events 8, 9)
    let mut recovered = Vec::new();
    journal
        .recover_from::<BookEventEnvelope, _>(Some(7), |env, _eob| {
            let seq: u64 = env.seq.into();
            recovered.push(seq);
            Ok(())
        })
        .unwrap();

    assert_eq!(recovered.len(), 2);
    assert_eq!(recovered[0], 8);
    assert_eq!(recovered[1], 9);
}

#[test]
fn recovery_from_mid_tx_cursor_replays_remaining_events() {
    let path = test_path("recovery_mid_tx_cursor");

    let journal = JournalHandler::open(&path, "wal", test_config()).unwrap();

    {
        let mut wtxn = journal.write_txn().unwrap();
        journal
            .put_envelope(&mut wtxn, &make_envelope(1, 42, 3, 0))
            .unwrap();
        journal
            .put_envelope(&mut wtxn, &make_envelope(2, 42, 3, 1))
            .unwrap();
        journal
            .put_envelope(&mut wtxn, &make_envelope(3, 42, 3, 2))
            .unwrap();
        journal.put_envelope(&mut wtxn, &make_single(4)).unwrap();
        wtxn.commit().unwrap();
    }

    let mut recovered = Vec::new();
    journal
        .recover_from::<BookEventEnvelope, _>(Some(1), |env, eob| {
            let seq: u64 = env.seq.into();
            let tx_id: u64 = env.tx_id.into();
            let tx_ix: u16 = env.tx_ix.into();
            recovered.push((seq, tx_id, tx_ix, eob));
            Ok(())
        })
        .unwrap();

    assert_eq!(
        recovered,
        vec![(2, 42, 1, false), (3, 42, 2, false), (4, 4, 0, true)]
    );
}

#[test]
fn recovery_from_cursor_ignores_gap_before_cursor() {
    let path = test_path("recovery_ignore_gap_before_cursor");

    let journal = JournalHandler::open(&path, "wal", test_config()).unwrap();

    {
        let mut wtxn = journal.write_txn().unwrap();
        journal.put_envelope(&mut wtxn, &make_single(1)).unwrap();
        journal.put_envelope(&mut wtxn, &make_single(3)).unwrap();
        journal.put_envelope(&mut wtxn, &make_single(10)).unwrap();
        wtxn.commit().unwrap();
    }

    let mut recovered = Vec::new();
    journal
        .recover_from::<BookEventEnvelope, _>(Some(9), |env, eob| {
            let seq: u64 = env.seq.into();
            recovered.push((seq, eob));
            Ok(())
        })
        .unwrap();

    assert_eq!(recovered, vec![(10, true)]);
}

#[test]
fn tx_sealed_at_segment_boundary() {
    let path = test_path("tx_sealed_boundary");

    let mut journal = JournalHandler::open(&path, "wal", test_config()).unwrap();

    // Write a 3-event transaction in segment 0
    {
        let mut wtxn = journal.write_txn().unwrap();
        for ix in 0..3u16 {
            let env = make_envelope(ix as u64, 100, 3, ix);
            journal.put_envelope(&mut wtxn, &env).unwrap();
        }
        wtxn.commit().unwrap();
    }

    journal.rotate(3).unwrap();

    // Write a 2-event transaction in segment 1
    {
        let mut wtxn = journal.write_txn().unwrap();
        for ix in 0..2u16 {
            let env = make_envelope(3 + ix as u64, 101, 2, ix);
            journal.put_envelope(&mut wtxn, &env).unwrap();
        }
        wtxn.commit().unwrap();
    }

    // Recover both transactions
    let mut recovered = Vec::new();
    journal
        .recover_from::<BookEventEnvelope, _>(None, |env, _eob| {
            let seq: u64 = env.seq.into();
            recovered.push(seq);
            Ok(())
        })
        .unwrap();

    assert_eq!(recovered.len(), 5);
    assert_eq!(recovered, vec![0, 1, 2, 3, 4]);
}

#[test]
fn contiguity_gap_detected() {
    let path = test_path("contiguity_gap");

    // Manually create two segments with a gap
    // Segment 0: events 0..5
    {
        let journal = JournalHandler::open(&path, "wal", test_config()).unwrap();
        let mut wtxn = journal.write_txn().unwrap();
        for seq in 0..5 {
            journal.put_envelope(&mut wtxn, &make_single(seq)).unwrap();
        }
        wtxn.commit().unwrap();
    }

    // Create segment starting at 7 (gap: should be 5)
    // We do this by creating the directory and writing to it directly
    {
        let seg_dir = path.join("seg_000000000007");
        std::fs::create_dir_all(&seg_dir).unwrap();
        let cfg = test_config();
        let env = unsafe {
            heed::EnvOpenOptions::new()
                .map_size(cfg.segment_map_size_bytes)
                .max_dbs(8)
                .open(&seg_dir)
                .unwrap()
        };
        let mut wtxn = env.write_txn().unwrap();
        let log: heed::Database<heed::types::U64<heed::byteorder::BigEndian>, heed::types::Bytes> =
            env.create_database(&mut wtxn, Some("wal")).unwrap();
        let meta: heed::Database<heed::types::Bytes, heed::types::U64<heed::byteorder::BigEndian>> =
            env.create_database(&mut wtxn, Some("meta")).unwrap();
        meta.put(&mut wtxn, b"schema_version", &1u64).unwrap();

        // Write event at seq 7
        let envelope = make_single(7);
        let bytes =
            rkyv::api::high::to_bytes::<betex::disruptor::traits::RkyvError>(&envelope).unwrap();
        log.put(&mut wtxn, &7u64, bytes.as_slice()).unwrap();
        wtxn.commit().unwrap();
    }

    // Now open and validate
    let journal = JournalHandler::open(&path, "wal", test_config()).unwrap();
    let result = journal.validate_segments();
    assert!(result.is_err());
    let err_msg = result.unwrap_err().to_string();
    assert!(
        err_msg.contains("contiguity gap"),
        "Expected contiguity gap error, got: {err_msg}"
    );
}

#[test]
fn delete_segments_preserves_active() {
    let path = test_path("delete_preserves_active");

    let mut journal = JournalHandler::open(&path, "wal", test_config()).unwrap();

    // Segment 0: events 0..5
    {
        let mut wtxn = journal.write_txn().unwrap();
        for seq in 0..5 {
            journal.put_envelope(&mut wtxn, &make_single(seq)).unwrap();
        }
        wtxn.commit().unwrap();
    }

    journal.rotate(5).unwrap();

    // Segment 1: events 5..10
    {
        let mut wtxn = journal.write_txn().unwrap();
        for seq in 5..10 {
            journal.put_envelope(&mut wtxn, &make_single(seq)).unwrap();
        }
        wtxn.commit().unwrap();
    }

    journal.rotate(10).unwrap();

    // Segment 2 (active): events 10..15
    {
        let mut wtxn = journal.write_txn().unwrap();
        for seq in 10..15 {
            journal.put_envelope(&mut wtxn, &make_single(seq)).unwrap();
        }
        wtxn.commit().unwrap();
    }

    // Delete segments where end_seq <= 9
    let deleted = journal.prune_segments_before(9).unwrap();
    assert_eq!(deleted.len(), 2);

    let segments = journal.list_segments().unwrap();
    assert_eq!(segments.len(), 1);
    assert_eq!(segments[0].start_seq, Some(10));
}

#[test]
fn delete_segments_partial() {
    let path = test_path("delete_partial");

    let mut journal = JournalHandler::open(&path, "wal", test_config()).unwrap();

    // Segment 0: events 0..5
    {
        let mut wtxn = journal.write_txn().unwrap();
        for seq in 0..5 {
            journal.put_envelope(&mut wtxn, &make_single(seq)).unwrap();
        }
        wtxn.commit().unwrap();
    }

    journal.rotate(5).unwrap();

    // Segment 1: events 5..10
    {
        let mut wtxn = journal.write_txn().unwrap();
        for seq in 5..10 {
            journal.put_envelope(&mut wtxn, &make_single(seq)).unwrap();
        }
        wtxn.commit().unwrap();
    }

    journal.rotate(10).unwrap();

    // Segment 2 (active): events 10..15
    {
        let mut wtxn = journal.write_txn().unwrap();
        for seq in 10..15 {
            journal.put_envelope(&mut wtxn, &make_single(seq)).unwrap();
        }
        wtxn.commit().unwrap();
    }

    // Delete only segments where end_seq <= 4
    let deleted = journal.prune_segments_before(4).unwrap();
    assert_eq!(deleted.len(), 1);

    let segments = journal.list_segments().unwrap();
    assert_eq!(segments.len(), 2);
    assert_eq!(segments[0].start_seq, Some(5));
    assert_eq!(segments[1].start_seq, Some(10));
}

#[test]
fn delete_segments_moves_to_trash() {
    let path = test_path("delete_to_trash");

    let mut journal = JournalHandler::open(&path, "wal", test_config()).unwrap();

    // Segment 0: events 0..5
    {
        let mut wtxn = journal.write_txn().unwrap();
        for seq in 0..5 {
            journal.put_envelope(&mut wtxn, &make_single(seq)).unwrap();
        }
        wtxn.commit().unwrap();
    }

    journal.rotate(5).unwrap();

    // Segment 1: events 5..10
    {
        let mut wtxn = journal.write_txn().unwrap();
        for seq in 5..10 {
            journal.put_envelope(&mut wtxn, &make_single(seq)).unwrap();
        }
        wtxn.commit().unwrap();
    }

    journal.rotate(10).unwrap();

    // Segment 2 (active): events 10..15
    {
        let mut wtxn = journal.write_txn().unwrap();
        for seq in 10..15 {
            journal.put_envelope(&mut wtxn, &make_single(seq)).unwrap();
        }
        wtxn.commit().unwrap();
    }

    let moved = journal.prune_segments_before(9).unwrap();
    assert_eq!(moved.len(), 2);

    let trash = trash_path(&path);
    assert!(trash.exists());
    for dir_name in moved {
        assert!(trash.join(&dir_name).exists());
        assert!(!path.join(&dir_name).exists());
    }

    let segments = journal.list_segments().unwrap();
    assert_eq!(segments.len(), 1);
    assert_eq!(segments[0].start_seq, Some(10));
}

#[test]
fn startup_clears_trash() {
    let path = test_path("startup_clears_trash");

    {
        let mut journal = JournalHandler::open(&path, "wal", test_config()).unwrap();

        // Segment 0: events 0..5
        {
            let mut wtxn = journal.write_txn().unwrap();
            for seq in 0..5 {
                journal.put_envelope(&mut wtxn, &make_single(seq)).unwrap();
            }
            wtxn.commit().unwrap();
        }

        journal.rotate(5).unwrap();

        // Segment 1: events 5..10
        {
            let mut wtxn = journal.write_txn().unwrap();
            for seq in 5..10 {
                journal.put_envelope(&mut wtxn, &make_single(seq)).unwrap();
            }
            wtxn.commit().unwrap();
        }

        journal.rotate(10).unwrap();

        // Segment 2 (active): events 10..15
        {
            let mut wtxn = journal.write_txn().unwrap();
            for seq in 10..15 {
                journal.put_envelope(&mut wtxn, &make_single(seq)).unwrap();
            }
            wtxn.commit().unwrap();
        }

        let moved = journal.prune_segments_before(9).unwrap();
        assert_eq!(moved.len(), 2);
        assert!(trash_path(&path).exists());
    }

    let trash = trash_path(&path);
    // open() now runs apply_startup_maintenance, so trash is cleared on open
    let journal = JournalHandler::open(&path, "wal", test_config()).unwrap();
    if trash.exists() {
        assert_eq!(std::fs::read_dir(&trash).unwrap().count(), 0);
    }

    let segments = journal.list_segments().unwrap();
    assert_eq!(segments.len(), 1);
    assert_eq!(segments[0].start_seq, Some(10));
}

#[test]
fn startup_prunes_before_sequence_then_clears_trash() {
    let path = test_path("startup_prune_before_seq");

    let setup_config = JournalConfig {
        segment_map_size_bytes: 8 * 1024 * 1024,
        max_segment_events: 5_000,
        validate_on_startup: false,
        prune_before_seq_on_startup: None,
        clear_trash_on_startup: false,
    };

    {
        let mut journal = JournalHandler::open(&path, "wal", setup_config).unwrap();

        // Segment 0: events 0..5
        {
            let mut wtxn = journal.write_txn().unwrap();
            for seq in 0..5 {
                journal.put_envelope(&mut wtxn, &make_single(seq)).unwrap();
            }
            wtxn.commit().unwrap();
        }

        journal.rotate(5).unwrap();

        // Segment 1: events 5..10
        {
            let mut wtxn = journal.write_txn().unwrap();
            for seq in 5..10 {
                journal.put_envelope(&mut wtxn, &make_single(seq)).unwrap();
            }
            wtxn.commit().unwrap();
        }

        journal.rotate(10).unwrap();

        // Segment 2 (active): events 10..15
        {
            let mut wtxn = journal.write_txn().unwrap();
            for seq in 10..15 {
                journal.put_envelope(&mut wtxn, &make_single(seq)).unwrap();
            }
            wtxn.commit().unwrap();
        }
    }

    // On startup: prune sealed segments <=9 into trash, then delete trash.
    let startup_config = JournalConfig {
        segment_map_size_bytes: 8 * 1024 * 1024,
        max_segment_events: 5_000,
        validate_on_startup: false,
        prune_before_seq_on_startup: Some(9),
        clear_trash_on_startup: true,
    };
    let trash = trash_path(&path);
    std::fs::create_dir_all(&trash).unwrap();
    let trash_marker = trash.join("keep-me");
    std::fs::write(&trash_marker, b"x").unwrap();

    // open() now runs apply_startup_maintenance: prune + trash cleanup on open
    let journal = JournalHandler::open(&path, "wal", startup_config).unwrap();

    let segments = journal.list_segments().unwrap();
    assert_eq!(segments.len(), 1);
    assert_eq!(segments[0].start_seq, Some(10));

    assert!(!trash_marker.exists());
    if trash.exists() {
        assert_eq!(std::fs::read_dir(&trash).unwrap().count(), 0);
    }
}

#[test]
fn segment_info_correct() {
    let path = test_path("segment_info");

    let mut journal = JournalHandler::open(&path, "wal", test_config()).unwrap();

    // Segment 0: events 0..5
    {
        let mut wtxn = journal.write_txn().unwrap();
        for seq in 0..5 {
            journal.put_envelope(&mut wtxn, &make_single(seq)).unwrap();
        }
        wtxn.commit().unwrap();
    }

    journal.rotate(5).unwrap();

    // Segment 1: events 5..8
    {
        let mut wtxn = journal.write_txn().unwrap();
        for seq in 5..8 {
            journal.put_envelope(&mut wtxn, &make_single(seq)).unwrap();
        }
        wtxn.commit().unwrap();
    }

    let segments = journal.list_segments().unwrap();
    assert_eq!(segments.len(), 2);

    assert_eq!(segments[0].start_seq, Some(0));
    assert_eq!(segments[0].end_seq, Some(4));
    assert_eq!(segments[0].event_count, 5);
    assert!(segments[0].file_size_bytes > 0);

    assert_eq!(segments[1].start_seq, Some(5));
    assert_eq!(segments[1].end_seq, Some(7));
    assert_eq!(segments[1].event_count, 3);
    assert!(segments[1].file_size_bytes > 0);
}

#[test]
fn auto_rotate_max_segment_events() {
    let path = test_path("auto_rotate");

    let config = JournalConfig {
        segment_map_size_bytes: 8 * 1024 * 1024,
        max_segment_events: 5,
        validate_on_startup: false,
        prune_before_seq_on_startup: None,
        clear_trash_on_startup: true,
    };

    let mut journal = JournalHandler::open(&path, "wal", config).unwrap();
    let mut next_seq = 0u64;

    // Write 12 single-event txs, checking should_rotate between each
    for _ in 0..12 {
        {
            let mut wtxn = journal.write_txn().unwrap();
            journal
                .put_envelope(&mut wtxn, &make_single(next_seq))
                .unwrap();
            wtxn.commit().unwrap();
        }

        if journal.should_rotate() {
            next_seq += 1;
            journal.rotate(next_seq).unwrap();
        } else {
            next_seq += 1;
        }
    }

    let segments = journal.list_segments().unwrap();
    // With max_segment_events=5, 12 events should produce 3 segments:
    // seg0: 5 events, seg1: 5 events, seg2: 2 events
    assert_eq!(segments.len(), 3);
    assert_eq!(segments[0].event_count, 5);
    assert_eq!(segments[1].event_count, 5);
    assert_eq!(segments[2].event_count, 2);
}

#[test]
fn multiple_rotations_sequential() {
    let path = test_path("multi_rotations");

    let mut journal = JournalHandler::open(&path, "wal", test_config()).unwrap();
    let mut next_seq = 0u64;

    // Rotate 5 times with 3 events between each
    for _ in 0..5 {
        {
            let mut wtxn = journal.write_txn().unwrap();
            for _ in 0..3 {
                journal
                    .put_envelope(&mut wtxn, &make_single(next_seq))
                    .unwrap();
                next_seq += 1;
            }
            wtxn.commit().unwrap();
        }
        journal.rotate(next_seq).unwrap();
    }

    // Write 2 more events to the final active segment
    {
        let mut wtxn = journal.write_txn().unwrap();
        for _ in 0..2 {
            journal
                .put_envelope(&mut wtxn, &make_single(next_seq))
                .unwrap();
            next_seq += 1;
        }
        wtxn.commit().unwrap();
    }

    // Validate
    journal.validate_segments().unwrap();

    // Recover all events
    let mut recovered = Vec::new();
    journal
        .recover_from::<BookEventEnvelope, _>(None, |env, _eob| {
            let seq: u64 = env.seq.into();
            recovered.push(seq);
            Ok(())
        })
        .unwrap();

    // 5 * 3 + 2 = 17 events total
    assert_eq!(recovered.len(), 17);
    for (i, seq) in recovered.iter().enumerate() {
        assert_eq!(*seq, i as u64);
    }
}

#[test]
fn configurable_map_size() {
    let path1 = test_path("map_size_2mb");
    let path2 = test_path("map_size_4mb");

    // Open with 2MB map size
    let config_2mb = JournalConfig {
        segment_map_size_bytes: 2 * 1024 * 1024,
        max_segment_events: 5_000,
        validate_on_startup: false,
        prune_before_seq_on_startup: None,
        clear_trash_on_startup: true,
    };
    let journal = JournalHandler::open(&path1, "wal", config_2mb).unwrap();
    {
        let mut wtxn = journal.write_txn().unwrap();
        for seq in 0..10 {
            journal.put_envelope(&mut wtxn, &make_single(seq)).unwrap();
        }
        wtxn.commit().unwrap();
    }

    // Open with 4MB map size
    let config_4mb = JournalConfig {
        segment_map_size_bytes: 4 * 1024 * 1024,
        max_segment_events: 5_000,
        validate_on_startup: false,
        prune_before_seq_on_startup: None,
        clear_trash_on_startup: true,
    };
    let journal = JournalHandler::open(&path2, "wal", config_4mb).unwrap();
    {
        let mut wtxn = journal.write_txn().unwrap();
        for seq in 0..10 {
            journal.put_envelope(&mut wtxn, &make_single(seq)).unwrap();
        }
        wtxn.commit().unwrap();
    }

    // Verify both work
    let mut recovered = Vec::new();
    journal
        .recover_from::<BookEventEnvelope, _>(None, |env, _eob| {
            let seq: u64 = env.seq.into();
            recovered.push(seq);
            Ok(())
        })
        .unwrap();
    assert_eq!(recovered.len(), 10);
}

#[test]
fn mixed_width_segment_names_sorted_numerically() {
    let path = test_path("mixed_width_sort");

    // Legacy width (12 digits) and larger width (13 digits) names.
    write_raw_segment_event(&path, "seg_999999999999", 999_999_999_999);
    write_raw_segment_event(&path, "seg_1000000000000", 1_000_000_000_000);

    // Opening should choose the numerically latest segment as active.
    let journal = JournalHandler::open(&path, "wal", test_config()).unwrap();
    let mut wtxn = journal.write_txn().unwrap();
    journal
        .put_envelope(&mut wtxn, &make_single(1_000_000_000_001))
        .unwrap();
    wtxn.commit().unwrap();

    let segments = journal.list_segments().unwrap();
    assert_eq!(segments.len(), 2);
    assert_eq!(segments[0].start_seq, Some(999_999_999_999));
    assert_eq!(segments[0].event_count, 1);
    assert_eq!(segments[1].start_seq, Some(1_000_000_000_000));
    assert_eq!(segments[1].event_count, 2);
    assert_eq!(segments[1].end_seq, Some(1_000_000_000_001));
}

#[test]
fn restart_with_new_disruptor_preserves_rotation_sequence_contiguity() {
    let wal_path = test_path("journal-rotation-restart-seq");
    let _ = std::fs::remove_dir_all(&wal_path);

    let cfg = Config {
        ring_size_pow2: 64,
        journal: JournalConfig {
            segment_map_size_bytes: 8 * 1024 * 1024,
            max_segment_events: 2,
            validate_on_startup: true,
            prune_before_seq_on_startup: None,
            clear_trash_on_startup: true,
        },
        ..Default::default()
    };

    // First run.
    let mut engine = match EngineBuilder::with_markets(
        cfg.clone(),
        [two_runner_market_config(MarketId(1))],
        &wal_path,
    ) {
        Ok(builder) => builder
            .register_handler(NoopHandler, betex::HandlerRecovery::LiveOnly, None)
            .build()
            .expect("start engine"),
        Err(e) => {
            if common::is_lmdb_unsupported(&e) {
                eprintln!("skipping LMDB WAL test: {e}");
                return;
            }
            panic!("builder failed: {e}");
        }
    };

    let handle = engine.handle();
    for state in [
        MarketState::Suspended,
        MarketState::Open,
        MarketState::Suspended,
        MarketState::Open,
        MarketState::Suspended,
        MarketState::Open,
    ] {
        submit_state_change(&handle, MarketId(1), state);
    }
    std::thread::sleep(std::time::Duration::from_millis(100));
    engine.stop().expect("stop first engine");

    // Restart with a fresh disruptor pipeline and write more events.
    let mut engine = match EngineBuilder::with_markets(
        cfg.clone(),
        [two_runner_market_config(MarketId(1))],
        &wal_path,
    ) {
        Ok(builder) => builder
            .register_handler(NoopHandler, betex::HandlerRecovery::LiveOnly, None)
            .build()
            .expect("restart engine"),
        Err(e) => {
            if common::is_lmdb_unsupported(&e) {
                eprintln!("skipping LMDB WAL test: {e}");
                return;
            }
            panic!("builder restart failed: {e}");
        }
    };

    let handle = engine.handle();
    for state in [
        MarketState::Suspended,
        MarketState::Open,
        MarketState::Suspended,
        MarketState::Open,
    ] {
        submit_state_change(&handle, MarketId(1), state);
    }
    std::thread::sleep(std::time::Duration::from_millis(100));
    engine.stop().expect("stop second engine");

    // Validate segments and recovery invariants after restart + auto-rotation.
    let journal = JournalHandler::open_engine(&wal_path, cfg.journal.clone()).expect("open wal");
    journal
        .validate_segments()
        .expect("segments should be contiguous after restart");

    let mut recovered = Vec::new();
    journal
        .recover_from::<BookEventEnvelope, _>(None, |env, _eob| {
            recovered.push(u64::from(env.seq));
            Ok(())
        })
        .expect("recover events");

    assert_eq!(recovered.len(), 15);
    assert_eq!(recovered.first().copied(), Some(1));
    assert_eq!(recovered.last().copied(), Some(15));
    for pair in recovered.windows(2) {
        assert_eq!(pair[1], pair[0] + 1);
    }

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