orderbook-rs 0.9.0

//! Deterministic replay engine for event journals.
//!
//! [`ReplayEngine`] reads a sequence of [`SequencerEvent`]s from a [`Journal`]
//! and re-applies each command to a fresh [`OrderBook`], producing an
//! identical final state. This enables disaster recovery, audit compliance,
//! and state verification.

use super::error::JournalError;
use super::journal::Journal;
use super::types::{SequencerCommand, SequencerEvent, SequencerResult};
use crate::orderbook::clock::Clock;
use crate::orderbook::fees::FeeSchedule;
use crate::orderbook::stp::STPMode;
use crate::orderbook::{OrderBook, OrderBookError, OrderBookSnapshot};
use serde::{Deserialize, Serialize};
use std::marker::PhantomData;
use std::sync::Arc;
use thiserror::Error;

/// Book configuration injected into a fresh [`OrderBook`] before replay so
/// that a journal produced by a **non-default-config** book reconstructs to
/// the same structure.
///
/// The plain [`ReplayEngine::replay_from`] / [`ReplayEngine::replay_from_with_clock`]
/// entry points build the target book with all configuration left at its
/// defaults (`tick_size` / `lot_size` / `min_order_size` / `max_order_size`
/// = `None`, `stp_mode` = [`STPMode::None`], `fee_schedule` = `None`). A book
/// that used any of these — for example a `MarketOrderByAmount` that rounds
/// per level under a `lot_size`, a self-cross prevented live by STP, or fees —
/// would replay into a **structurally different** book, so `snapshots_match`
/// can fail at verify and the recovered state would be wrong.
///
/// To recover such a book deterministically, carry the original configuration
/// alongside the journal (it is the same set of fields persisted in
/// [`OrderBookSnapshotPackage`](crate::OrderBookSnapshot)'s package form) and
/// replay through a `*_with_config` variant
/// ([`ReplayEngine::replay_from_with_config`] /
/// [`ReplayEngine::replay_from_with_clock_and_config`]).
///
/// The configuration is supplied by the **caller** — replay does not read it
/// from the journal, so the on-disk journal format is unchanged and
/// `ORDERBOOK_SNAPSHOT_FORMAT_VERSION` is not bumped.
///
/// `Default` yields the all-defaults configuration, equivalent to the plain
/// replay entry points.
#[derive(Debug, Clone, Default)]
pub struct ReplayBookConfig {
    /// Fee schedule the source book used, or `None` for no fees. Applied via
    /// [`OrderBook::set_fee_schedule`].
    pub fee_schedule: Option<FeeSchedule>,

    /// Self-trade prevention mode the source book used. [`STPMode::None`]
    /// (the default) disables STP. Applied via [`OrderBook::set_stp_mode`].
    pub stp_mode: STPMode,

    /// Tick size (minimum price increment) the source book used, or `None`
    /// for no tick validation. Applied via [`OrderBook::set_tick_size_opt`].
    pub tick_size: Option<u128>,

    /// Lot size (minimum quantity increment) the source book used, or `None`
    /// for no lot validation / rounding. Applied via
    /// [`OrderBook::set_lot_size_opt`].
    pub lot_size: Option<u64>,

    /// Minimum order size the source book used, or `None` for no minimum.
    /// Applied via [`OrderBook::set_min_order_size`] only when `Some`.
    pub min_order_size: Option<u64>,

    /// Maximum order size the source book used, or `None` for no maximum.
    /// Applied via [`OrderBook::set_max_order_size`] only when `Some`.
    pub max_order_size: Option<u64>,
}

impl ReplayBookConfig {
    /// Creates a [`ReplayBookConfig`] from its six fields.
    ///
    /// Equivalent to building the struct with public-field syntax; provided so
    /// callers can construct the carrier without naming every field at the call
    /// site. Use [`ReplayBookConfig::default`] for the all-defaults case.
    ///
    /// # Arguments
    ///
    /// * `fee_schedule` — fee schedule the source book used, or `None`
    /// * `stp_mode` — self-trade prevention mode the source book used
    /// * `tick_size` — tick size the source book used, or `None`
    /// * `lot_size` — lot size the source book used, or `None`
    /// * `min_order_size` — minimum order size the source book used, or `None`
    /// * `max_order_size` — maximum order size the source book used, or `None`
    #[must_use]
    pub fn new(
        fee_schedule: Option<FeeSchedule>,
        stp_mode: STPMode,
        tick_size: Option<u128>,
        lot_size: Option<u64>,
        min_order_size: Option<u64>,
        max_order_size: Option<u64>,
    ) -> Self {
        Self {
            fee_schedule,
            stp_mode,
            tick_size,
            lot_size,
            min_order_size,
            max_order_size,
        }
    }

    /// Applies this configuration to a freshly-constructed `book` in place,
    /// before any journal events are replayed into it.
    ///
    /// `fee_schedule`, `stp_mode`, `tick_size`, and `lot_size` are applied
    /// unconditionally (a `None` / [`STPMode::None`] value resets the field to
    /// its default, which is a no-op on a fresh book). `min_order_size` and
    /// `max_order_size` are applied only when `Some`, mirroring the existing
    /// `set_min_order_size` / `set_max_order_size` setters which take a bare
    /// value rather than an `Option`.
    fn apply_to<T>(&self, book: &mut OrderBook<T>)
    where
        T: Serialize + for<'de> Deserialize<'de> + Clone + Send + Sync + Default + 'static,
    {
        book.set_fee_schedule(self.fee_schedule);
        book.set_stp_mode(self.stp_mode);
        book.set_tick_size_opt(self.tick_size);
        book.set_lot_size_opt(self.lot_size);
        if let Some(min) = self.min_order_size {
            book.set_min_order_size(min);
        }
        if let Some(max) = self.max_order_size {
            book.set_max_order_size(max);
        }
    }
}

/// Errors that can occur during journal replay.
#[derive(Debug, Error)]
pub enum ReplayError {
    /// The journal contains no events to replay.
    #[error("journal is empty — nothing to replay")]
    EmptyJournal,

    /// The requested starting sequence number exceeds the journal's last entry.
    #[error("invalid from_sequence {from_sequence}: journal last sequence is {last_sequence}")]
    InvalidSequence {
        /// The sequence number requested.
        from_sequence: u64,
        /// The last sequence number in the journal.
        last_sequence: u64,
    },

    /// A gap was detected between expected and found sequence numbers.
    #[error("sequence gap detected: expected {expected}, found {found}")]
    SequenceGap {
        /// The expected next sequence number.
        expected: u64,
        /// The actual sequence number found.
        found: u64,
    },

    /// The protocol sequence counter overflowed `u64` while advancing.
    ///
    /// Unreachable at any realistic journal length, but advancing the counter
    /// with a checked add (rather than a saturating one) keeps gap detection
    /// correct at the boundary instead of silently stalling `expected_seq`.
    #[error("replay sequence counter overflowed u64 at sequence {at}")]
    SequenceOverflow {
        /// The sequence number that could not be advanced past.
        at: u64,
    },

    /// An OrderBook operation failed during replay.
    #[error("order book error during replay at sequence {sequence_num}: {source}")]
    OrderBookError {
        /// The sequence number of the event that caused the error.
        sequence_num: u64,
        /// The underlying error.
        #[source]
        source: OrderBookError,
    },

    /// The replayed state does not match the expected snapshot.
    #[error("snapshot mismatch: replayed state diverges from expected snapshot")]
    SnapshotMismatch,

    /// Journal read error during replay.
    #[error("journal error during replay: {0}")]
    JournalError(#[from] JournalError),
}

/// Stateless replay engine that reconstructs [`OrderBook`] state from a [`Journal`].
///
/// All methods are associated functions (no `&self` receiver) — `ReplayEngine`
/// holds no state itself. Use it as a namespace for replay operations.
pub struct ReplayEngine<T> {
    _phantom: PhantomData<T>,
}

impl<T> ReplayEngine<T>
where
    T: Serialize + for<'de> Deserialize<'de> + Clone + Send + Sync + Default + 'static,
{
    /// Replays all events from `from_sequence` onwards onto a fresh [`OrderBook`].
    ///
    /// Returns the reconstructed book and the sequence number of the last
    /// event applied. Only successful commands (non-`Rejected` results) are
    /// replayed — rejected events are skipped without error.
    ///
    /// For deterministic replay with a custom clock, see
    /// [`Self::replay_from_with_clock`].
    ///
    /// # Configuration
    ///
    /// This entry point builds the target book with **all configuration at its
    /// defaults** (`tick_size` / `lot_size` / `min_order_size` /
    /// `max_order_size` = `None`, `stp_mode` = [`STPMode::None`],
    /// `fee_schedule` = `None`). It is therefore only valid for replaying a
    /// journal that was produced by a **default-config** book. A book that used
    /// tick / lot / STP / fees must be replayed through
    /// [`Self::replay_from_with_config`] (or
    /// [`Self::replay_from_with_clock_and_config`]) with the matching
    /// [`ReplayBookConfig`], or the reconstructed state will diverge from the
    /// original (and `snapshots_match` will report a mismatch).
    ///
    /// # Arguments
    ///
    /// * `journal` — the event source
    /// * `from_sequence` — first sequence number to include (inclusive); pass `0` for full replay
    /// * `symbol` — symbol used to create the fresh OrderBook
    ///
    /// # Errors
    ///
    /// - [`ReplayError::EmptyJournal`] if the journal has no events
    /// - [`ReplayError::InvalidSequence`] if `from_sequence` > last journal sequence
    /// - [`ReplayError::OrderBookError`] if a command fails unexpectedly during replay
    /// - [`ReplayError::JournalError`] if reading from the journal fails
    #[must_use = "replay result carries the reconstructed book and the last applied sequence"]
    pub fn replay_from(
        journal: &impl Journal<T>,
        from_sequence: u64,
        symbol: &str,
    ) -> Result<(OrderBook<T>, u64), ReplayError> {
        Self::replay_from_with_progress(journal, from_sequence, symbol, |_, _| {})
    }

    /// Replays events with a progress callback invoked after each applied event.
    ///
    /// The callback receives `(events_applied: u64, current_sequence: u64)`.
    /// Useful for long replays where progress reporting is needed.
    ///
    /// For deterministic replay with a custom clock, see
    /// [`Self::replay_from_with_clock`].
    ///
    /// # Arguments
    ///
    /// * `journal` — the event source
    /// * `from_sequence` — first sequence number to include; pass `0` for full replay
    /// * `symbol` — symbol for the fresh OrderBook
    /// * `progress` — callback invoked after each event: `(events_applied, sequence_num)`
    ///
    /// # Errors
    ///
    /// Same as [`replay_from`](Self::replay_from).
    #[must_use = "replay result carries the reconstructed book and the last applied sequence"]
    pub fn replay_from_with_progress(
        journal: &impl Journal<T>,
        from_sequence: u64,
        symbol: &str,
        progress: impl Fn(u64, u64),
    ) -> Result<(OrderBook<T>, u64), ReplayError> {
        let last_seq = match journal.last_sequence() {
            Some(seq) => seq,
            None => return Err(ReplayError::EmptyJournal),
        };

        if from_sequence > last_seq {
            return Err(ReplayError::InvalidSequence {
                from_sequence,
                last_sequence: last_seq,
            });
        }

        let book = OrderBook::new(symbol);
        let last_applied_seq = Self::replay_into(&book, journal, from_sequence, progress)?;
        Ok((book, last_applied_seq))
    }

    /// Like [`Self::replay_from`] but injects a caller-supplied
    /// [`ReplayBookConfig`] into the fresh book **before** any events are
    /// replayed.
    ///
    /// This is the entry point for recovering a **non-default-config** book:
    /// the configuration (fees, STP, tick / lot / min / max order size) is
    /// applied to the target book so that a journal produced under that
    /// configuration reconstructs to the same structure and passes
    /// `snapshots_match` against the original. The configuration is supplied by
    /// the caller — it is not read from the journal, so the journal format is
    /// unchanged.
    ///
    /// For byte-identical timestamp reproduction (e.g. replay tests, or
    /// disaster-recovery that must match engine-assigned timestamps), use
    /// [`Self::replay_from_with_clock_and_config`].
    ///
    /// # Arguments
    ///
    /// * `journal` — the event source
    /// * `from_sequence` — first sequence number to include (inclusive); pass `0` for full replay
    /// * `symbol` — symbol used to create the fresh OrderBook
    /// * `config` — configuration the source book used, applied before replay
    ///
    /// # Errors
    ///
    /// Same as [`replay_from`](Self::replay_from).
    #[must_use = "replay result carries the reconstructed book and the last applied sequence"]
    pub fn replay_from_with_config(
        journal: &impl Journal<T>,
        from_sequence: u64,
        symbol: &str,
        config: &ReplayBookConfig,
    ) -> Result<(OrderBook<T>, u64), ReplayError> {
        let last_seq = match journal.last_sequence() {
            Some(seq) => seq,
            None => return Err(ReplayError::EmptyJournal),
        };

        if from_sequence > last_seq {
            return Err(ReplayError::InvalidSequence {
                from_sequence,
                last_sequence: last_seq,
            });
        }

        let mut book = OrderBook::new(symbol);
        config.apply_to(&mut book);
        let last_applied_seq = Self::replay_into(&book, journal, from_sequence, |_, _| {})?;
        Ok((book, last_applied_seq))
    }

    /// Like [`Self::replay_from`] but injects a caller-supplied [`Clock`] into
    /// the reconstructed book.
    ///
    /// This is the canonical entry point for byte-identical replay tests and
    /// disaster-recovery pipelines that must reproduce engine-assigned
    /// timestamps deterministically. Pass a
    /// [`crate::orderbook::clock::StubClock`] for test and proptest-driven
    /// replay, or a [`crate::orderbook::clock::MonotonicClock`] for
    /// production disaster-recovery where wall-clock timestamps are
    /// acceptable.
    ///
    /// # Configuration
    ///
    /// Like [`Self::replay_from`], this builds the target book with **all
    /// configuration at its defaults** and is only valid for a default-config
    /// source book. To recover a book that used tick / lot / STP / fees
    /// deterministically, use [`Self::replay_from_with_clock_and_config`] with
    /// the matching [`ReplayBookConfig`].
    ///
    /// # Arguments
    ///
    /// * `journal` — the event source
    /// * `from_sequence` — first sequence number to include (inclusive); pass `0` for full replay
    /// * `symbol` — symbol used to create the fresh OrderBook
    /// * `clock` — pre-constructed clock shared across the reconstructed book
    ///
    /// # Errors
    ///
    /// Same as [`replay_from`](Self::replay_from).
    #[must_use = "replay result carries the reconstructed book and the last applied sequence"]
    pub fn replay_from_with_clock(
        journal: &impl Journal<T>,
        from_sequence: u64,
        symbol: &str,
        clock: Arc<dyn Clock>,
    ) -> Result<(OrderBook<T>, u64), ReplayError> {
        Self::replay_from_with_clock_and_progress(journal, from_sequence, symbol, clock, |_, _| {})
    }

    /// Like [`Self::replay_from_with_progress`] plus clock injection.
    ///
    /// Equivalent to [`Self::replay_from_with_clock`] but forwards each
    /// successfully-applied event to a progress callback. Useful for long
    /// replays where progress reporting is needed and byte-identical
    /// timestamp reproduction is required — the canonical entry point for
    /// byte-identical replay tests and disaster-recovery pipelines that must
    /// reproduce engine-assigned timestamps deterministically.
    ///
    /// # Arguments
    ///
    /// * `journal` — the event source
    /// * `from_sequence` — first sequence number to include; pass `0` for full replay
    /// * `symbol` — symbol for the fresh OrderBook
    /// * `clock` — pre-constructed clock shared across the reconstructed book
    /// * `progress` — callback invoked after each event: `(events_applied, sequence_num)`
    ///
    /// # Errors
    ///
    /// Same as [`replay_from`](Self::replay_from).
    #[must_use = "replay result carries the reconstructed book and the last applied sequence"]
    pub fn replay_from_with_clock_and_progress(
        journal: &impl Journal<T>,
        from_sequence: u64,
        symbol: &str,
        clock: Arc<dyn Clock>,
        progress: impl Fn(u64, u64),
    ) -> Result<(OrderBook<T>, u64), ReplayError> {
        let last_seq = match journal.last_sequence() {
            Some(seq) => seq,
            None => return Err(ReplayError::EmptyJournal),
        };

        if from_sequence > last_seq {
            return Err(ReplayError::InvalidSequence {
                from_sequence,
                last_sequence: last_seq,
            });
        }

        let book = OrderBook::with_clock(symbol, clock);
        let last_applied_seq = Self::replay_into(&book, journal, from_sequence, progress)?;
        Ok((book, last_applied_seq))
    }

    /// Like [`Self::replay_from_with_clock`] but also injects a caller-supplied
    /// [`ReplayBookConfig`] into the fresh book **before** any events are
    /// replayed.
    ///
    /// This is the canonical entry point for byte-identical, deterministic
    /// recovery of a **non-default-config** book: the injected [`Clock`]
    /// reproduces engine-assigned timestamps and the [`ReplayBookConfig`]
    /// reproduces the structural configuration (fees, STP, tick / lot / min /
    /// max order size), so the reconstructed book passes `snapshots_match`
    /// against the original. The configuration is supplied by the caller — it
    /// is not read from the journal, so the journal format is unchanged.
    ///
    /// # Arguments
    ///
    /// * `journal` — the event source
    /// * `from_sequence` — first sequence number to include (inclusive); pass `0` for full replay
    /// * `symbol` — symbol used to create the fresh OrderBook
    /// * `clock` — pre-constructed clock shared across the reconstructed book
    /// * `config` — configuration the source book used, applied before replay
    ///
    /// # Errors
    ///
    /// Same as [`replay_from`](Self::replay_from).
    #[must_use = "replay result carries the reconstructed book and the last applied sequence"]
    pub fn replay_from_with_clock_and_config(
        journal: &impl Journal<T>,
        from_sequence: u64,
        symbol: &str,
        clock: Arc<dyn Clock>,
        config: &ReplayBookConfig,
    ) -> Result<(OrderBook<T>, u64), ReplayError> {
        let last_seq = match journal.last_sequence() {
            Some(seq) => seq,
            None => return Err(ReplayError::EmptyJournal),
        };

        if from_sequence > last_seq {
            return Err(ReplayError::InvalidSequence {
                from_sequence,
                last_sequence: last_seq,
            });
        }

        let mut book = OrderBook::with_clock(symbol, clock);
        config.apply_to(&mut book);
        let last_applied_seq = Self::replay_into(&book, journal, from_sequence, |_, _| {})?;
        Ok((book, last_applied_seq))
    }

    /// Shared replay loop. Applies events from `journal` starting at
    /// `from_sequence` to the already-constructed `book`, reporting
    /// per-event progress via `progress`, and returns the last applied
    /// sequence number.
    ///
    /// Does not construct the book and does not perform the
    /// `EmptyJournal` / `InvalidSequence` pre-checks — those remain the
    /// responsibility of the public entry points so that the distinction
    /// between "the journal is empty" and "the journal exists but
    /// contains no matching range" is preserved.
    fn replay_into(
        book: &OrderBook<T>,
        journal: &impl Journal<T>,
        from_sequence: u64,
        progress: impl Fn(u64, u64),
    ) -> Result<u64, ReplayError> {
        let mut last_applied_seq = 0u64;
        let mut count = 0u64;
        let mut expected_seq = from_sequence;

        let iter = journal.read_from(from_sequence)?;

        for entry_result in iter {
            let entry = entry_result?;
            let event = &entry.event;

            // Gap detection
            if event.sequence_num != expected_seq {
                return Err(ReplayError::SequenceGap {
                    expected: expected_seq,
                    found: event.sequence_num,
                });
            }

            // Advance `expected_seq` before applying so gap detection stays
            // correct even if the event is a rejected no-op. `last_applied_seq`,
            // `count`, and `progress` track only events that actually mutate
            // the book — consistent with the "events applied" / "last applied
            // sequence" contract on the public entry points.
            let applied = !matches!(event.result, SequencerResult::Rejected { .. });
            Self::apply_event(book, event)?;
            // Protocol counter: a saturating add would silently stop advancing
            // `expected_seq` at the u64 ceiling and mask a real gap, so use a
            // checked add and surface a typed overflow error instead (per the
            // no-saturating-on-protocol-counters rule).
            expected_seq = expected_seq
                .checked_add(1)
                .ok_or(ReplayError::SequenceOverflow { at: expected_seq })?;

            if applied {
                last_applied_seq = event.sequence_num;
                count = count
                    .checked_add(1)
                    .ok_or(ReplayError::SequenceOverflow { at: count })?;
                progress(count, last_applied_seq);
            }
        }

        Ok(last_applied_seq)
    }

    /// Replays the full journal and compares the result to an expected snapshot.
    ///
    /// Returns `Ok(true)` if the replayed state matches, `Ok(false)` if it
    /// diverges. The comparison uses [`snapshots_match`] which checks symbol,
    /// bid price levels, and ask price levels.
    ///
    /// # Errors
    ///
    /// - [`ReplayError::EmptyJournal`] if the journal has no events
    /// - [`ReplayError::OrderBookError`] if replay fails
    /// - [`ReplayError::JournalError`] if reading from the journal fails
    pub fn verify(
        journal: &impl Journal<T>,
        expected_snapshot: &OrderBookSnapshot,
    ) -> Result<bool, ReplayError> {
        let (book, _) = Self::replay_from(journal, 0, &expected_snapshot.symbol)?;
        let actual = book.create_snapshot(usize::MAX);
        Ok(snapshots_match(&actual, expected_snapshot))
    }

    /// Applies a single sequencer event to the given book.
    ///
    /// Events with `Rejected` results are skipped — they represent commands
    /// that failed at write time and must not be re-applied during replay.
    fn apply_event(book: &OrderBook<T>, event: &SequencerEvent<T>) -> Result<(), ReplayError> {
        // Skip events whose original execution was rejected.
        if matches!(event.result, SequencerResult::Rejected { .. }) {
            return Ok(());
        }

        match &event.command {
            SequencerCommand::AddOrder(order) => {
                book.add_order(order.clone())
                    .map_err(|e| ReplayError::OrderBookError {
                        sequence_num: event.sequence_num,
                        source: e,
                    })?;
            }
            SequencerCommand::CancelOrder(id) => {
                book.cancel_order(*id)
                    .map_err(|e| ReplayError::OrderBookError {
                        sequence_num: event.sequence_num,
                        source: e,
                    })?;
            }
            SequencerCommand::UpdateOrder(update) => {
                book.update_order(*update)
                    .map_err(|e| ReplayError::OrderBookError {
                        sequence_num: event.sequence_num,
                        source: e,
                    })?;
            }
            SequencerCommand::MarketOrder { id, quantity, side } => {
                book.submit_market_order(*id, *quantity, *side)
                    .map_err(|e| ReplayError::OrderBookError {
                        sequence_num: event.sequence_num,
                        source: e,
                    })?;
            }
            SequencerCommand::MarketOrderByAmount { id, amount, side } => {
                book.submit_market_order_by_amount(*id, *amount, *side)
                    .map_err(|e| ReplayError::OrderBookError {
                        sequence_num: event.sequence_num,
                        source: e,
                    })?;
            }
            SequencerCommand::CancelAll => {
                let _ = book.cancel_all_orders();
            }
            SequencerCommand::CancelBySide { side } => {
                let _ = book.cancel_orders_by_side(*side);
            }
            SequencerCommand::CancelByUser { user_id } => {
                let _ = book.cancel_orders_by_user(*user_id);
            }
            SequencerCommand::CancelByPriceRange {
                side,
                min_price,
                max_price,
            } => {
                let _ = book.cancel_orders_by_price_range(*side, *min_price, *max_price);
            }
        }

        Ok(())
    }
}

/// Compares two [`OrderBookSnapshot`]s for structural equality.
///
/// Two snapshots are considered equal when:
/// - `symbol` is identical
/// - The sorted bid price levels match (by price, **visible** quantity, **hidden**
///   quantity, and **order count**)
/// - The sorted ask price levels match (same four fields)
///
/// This is the equality oracle for replay correctness, so it compares the full
/// per-level state — not just visible quantity. Comparing visible quantity alone
/// would be a subset check that misses divergences in hidden iceberg / reserve
/// liquidity or in how the same visible depth is split across orders at a price
/// (#102).
///
/// Timestamps are intentionally excluded from comparison because replayed
/// books may be created at a different wall-clock time than the original.
#[must_use]
pub fn snapshots_match(actual: &OrderBookSnapshot, expected: &OrderBookSnapshot) -> bool {
    if actual.symbol != expected.symbol {
        return false;
    }

    // Compare bids sorted by price descending (highest bid first)
    let mut actual_bids: Vec<_> = actual.bids.iter().collect();
    let mut expected_bids: Vec<_> = expected.bids.iter().collect();
    actual_bids.sort_by_key(|b| std::cmp::Reverse(b.price()));
    expected_bids.sort_by_key(|b| std::cmp::Reverse(b.price()));

    if actual_bids.len() != expected_bids.len() {
        return false;
    }
    for (a, b) in actual_bids.iter().zip(expected_bids.iter()) {
        if a.price() != b.price()
            || a.visible_quantity() != b.visible_quantity()
            || a.hidden_quantity() != b.hidden_quantity()
            || a.order_count() != b.order_count()
        {
            return false;
        }
    }

    // Compare asks sorted by price ascending (lowest ask first)
    let mut actual_asks: Vec<_> = actual.asks.iter().collect();
    let mut expected_asks: Vec<_> = expected.asks.iter().collect();
    actual_asks.sort_by_key(|l| l.price());
    expected_asks.sort_by_key(|l| l.price());

    if actual_asks.len() != expected_asks.len() {
        return false;
    }
    for (a, b) in actual_asks.iter().zip(expected_asks.iter()) {
        if a.price() != b.price()
            || a.visible_quantity() != b.visible_quantity()
            || a.hidden_quantity() != b.hidden_quantity()
            || a.order_count() != b.order_count()
        {
            return false;
        }
    }

    true
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::orderbook::clock::{MonotonicClock, StubClock};
    use crate::orderbook::sequencer::InMemoryJournal;
    use crate::orderbook::trade::TradeResult;
    use pricelevel::{
        Hash32, Id, MatchResult, OrderType, Price, Quantity, Side, TimeInForce, TimestampMs,
    };

    fn make_add_event(seq: u64, id: Id, price: u128, qty: u64, side: Side) -> SequencerEvent<()> {
        let order = OrderType::Standard {
            id,
            price: Price::new(price),
            quantity: Quantity::new(qty),
            side,
            time_in_force: TimeInForce::Gtc,
            user_id: Hash32::zero(),
            timestamp: TimestampMs::new(0),
            extra_fields: (),
        };
        SequencerEvent {
            sequence_num: seq,
            timestamp_ns: 0,
            command: SequencerCommand::AddOrder(order),
            result: SequencerResult::OrderAdded { order_id: id },
        }
    }

    /// #102: `snapshots_match` is the replay equality oracle and must compare the
    /// full per-level state — not just visible quantity. Two snapshots that differ
    /// only in hidden quantity or order count must NOT be reported equal.
    #[test]
    fn test_snapshots_match_compares_hidden_quantity_and_order_count() {
        fn lvl(
            price: u128,
            visible: u64,
            hidden: u64,
            count: usize,
        ) -> pricelevel::PriceLevelSnapshot {
            serde_json::from_value(serde_json::json!({
                "price": price,
                "visible_quantity": visible,
                "hidden_quantity": hidden,
                "order_count": count,
                "orders": []
            }))
            .expect("valid snapshot JSON")
        }

        let base = OrderBookSnapshot {
            symbol: "TEST".to_string(),
            timestamp: 0,
            bids: vec![lvl(100, 10, 5, 2)],
            asks: Vec::new(),
        };
        assert!(
            snapshots_match(&base, &base.clone()),
            "identical snapshots must match"
        );

        let diff_hidden = OrderBookSnapshot {
            symbol: "TEST".to_string(),
            timestamp: 0,
            bids: vec![lvl(100, 10, 7, 2)],
            asks: Vec::new(),
        };
        assert!(
            !snapshots_match(&base, &diff_hidden),
            "a hidden-quantity divergence must not be reported equal"
        );

        let diff_count = OrderBookSnapshot {
            symbol: "TEST".to_string(),
            timestamp: 0,
            bids: vec![lvl(100, 10, 5, 3)],
            asks: Vec::new(),
        };
        assert!(
            !snapshots_match(&base, &diff_count),
            "an order-count divergence must not be reported equal"
        );
    }

    /// #126: the protocol sequence counter advances with `checked_add`, so at
    /// the `u64` boundary it surfaces a typed `SequenceOverflow` instead of
    /// silently stalling `expected_seq` (which would mask a real gap).
    #[test]
    fn test_replay_sequence_counter_overflow_is_a_typed_error() {
        let journal: InMemoryJournal<()> = InMemoryJournal::new();
        // A single event at the very top of the sequence space.
        let ev = make_add_event(u64::MAX, Id::new_uuid(), 100, 10, Side::Buy);
        assert!(journal.append(&ev).is_ok());

        // Replaying from u64::MAX applies the event, then advancing the
        // expected-sequence counter past u64::MAX must overflow with a typed
        // error rather than saturating.
        match ReplayEngine::<()>::replay_from(&journal, u64::MAX, "TEST") {
            Err(ReplayError::SequenceOverflow { at }) => assert_eq!(at, u64::MAX),
            Err(other) => panic!("expected SequenceOverflow {{ at: u64::MAX }}, got {other:?}"),
            Ok(_) => panic!("advancing past u64::MAX must error"),
        }
    }

    #[test]
    fn test_replay_from_with_clock_uses_injected_clock() {
        let journal: InMemoryJournal<()> = InMemoryJournal::new();
        for (seq, price) in [(0u64, 100u128), (1, 101), (2, 102)] {
            let ev = make_add_event(seq, Id::new_uuid(), price, 10, Side::Buy);
            assert!(journal.append(&ev).is_ok());
        }

        let clock: Arc<dyn Clock> = Arc::new(StubClock::starting_at(42_000));
        let result = ReplayEngine::<()>::replay_from_with_clock(&journal, 0, "TEST", clock);
        assert!(result.is_ok(), "replay_from_with_clock should succeed");
        let (book, last_seq) = result.expect("replay succeeded");
        assert_eq!(last_seq, 2);

        // The injected StubClock was seeded at 42_000. After the book has
        // been constructed, any ticks the replay consumed have advanced the
        // counter — so the next tick must be >= 42_000.
        let now = book.clock().now_millis();
        assert!(
            now.as_u64() >= 42_000,
            "expected injected clock value, got {}",
            now.as_u64()
        );
    }

    #[test]
    fn test_replay_from_with_clock_preserves_behavior_of_replay_from() {
        // Journal shared across both replays.
        let journal: InMemoryJournal<()> = InMemoryJournal::new();
        let ids: Vec<Id> = (0..3).map(|_| Id::new_uuid()).collect();
        let events = [
            make_add_event(0, ids[0], 100, 5, Side::Buy),
            make_add_event(1, ids[1], 101, 7, Side::Buy),
            make_add_event(2, ids[2], 105, 3, Side::Sell),
        ];
        for ev in &events {
            assert!(journal.append(ev).is_ok());
        }

        let (book_plain, last_seq_plain) = ReplayEngine::<()>::replay_from(&journal, 0, "TEST")
            .expect("plain replay should succeed");

        let clock: Arc<dyn Clock> = Arc::new(MonotonicClock);
        let (book_with_clock, last_seq_with_clock) =
            ReplayEngine::<()>::replay_from_with_clock(&journal, 0, "TEST", clock)
                .expect("clock-aware replay should succeed");

        assert_eq!(last_seq_plain, last_seq_with_clock);
        assert_eq!(last_seq_plain, 2);

        let snap_plain = book_plain.create_snapshot(usize::MAX);
        let snap_with_clock = book_with_clock.create_snapshot(usize::MAX);
        assert!(
            snapshots_match(&snap_plain, &snap_with_clock),
            "snapshots must match across replay variants"
        );
    }

    #[test]
    fn test_replay_from_with_clock_propagates_sequence_gap() {
        let journal: InMemoryJournal<()> = InMemoryJournal::new();
        // Sequences 0, 1, 2, then jump to 4 (gap at 3).
        let events = [
            make_add_event(0, Id::new_uuid(), 100, 1, Side::Buy),
            make_add_event(1, Id::new_uuid(), 101, 1, Side::Buy),
            make_add_event(2, Id::new_uuid(), 102, 1, Side::Buy),
            make_add_event(4, Id::new_uuid(), 104, 1, Side::Buy),
        ];
        for ev in &events {
            assert!(journal.append(ev).is_ok());
        }

        let clock: Arc<dyn Clock> = Arc::new(StubClock::new());
        let result = ReplayEngine::<()>::replay_from_with_clock(&journal, 0, "TEST", clock);

        match result {
            Err(ReplayError::SequenceGap { expected, found }) => {
                assert_eq!(expected, 3);
                assert_eq!(found, 4);
            }
            Err(other) => panic!(
                "expected SequenceGap {{ expected: 3, found: 4 }}, got {:?}",
                other
            ),
            Ok(_) => panic!("expected SequenceGap {{ expected: 3, found: 4 }}, got Ok(_)"),
        }
    }

    #[test]
    fn test_replay_market_order_by_amount_matches_live_book() {
        // Build a journal: seed an ask wall, then take it with a
        // notional market order. Replay against a fresh book and
        // require the resulting snapshot to match the live one — proves
        // the additive variant dispatches identically to the live path.
        let journal: InMemoryJournal<()> = InMemoryJournal::new();
        let mut seq = 0u64;

        // Three asks at 100, 101, 102 — each size 10.
        for price in [100u128, 101, 102] {
            let ev = make_add_event(seq, Id::new_uuid(), price, 10, Side::Sell);
            assert!(journal.append(&ev).is_ok());
            seq += 1;
        }

        // Notional buy: $1500 sweeps 10@100 + 10@101 = $2010 total — but
        // we cap at $1500 so only 10@100 + 4@101 (=$1404) lands. The
        // residual $96 is dust < 1*101 = 101 still — actually it can buy
        // 0 more at 101 → stop short of the third level. Exact behavior
        // doesn't matter for this test; what matters is replay parity.
        let taker_id = Id::new_uuid();
        let ev = SequencerEvent::<()> {
            sequence_num: seq,
            timestamp_ns: 0,
            command: SequencerCommand::MarketOrderByAmount {
                id: taker_id,
                amount: 1_500,
                side: Side::Buy,
            },
            // Result is informational for replay — replay re-executes
            // the command against a fresh book. Use TradeExecuted with an
            // empty match-result so the journal entry stays semantically
            // consistent with a market-by-amount taker (and is not skipped
            // by the Rejected branch in `replay_from`).
            result: SequencerResult::TradeExecuted {
                trade_result: TradeResult::new(
                    "TEST".to_string(),
                    MatchResult::new(taker_id, Quantity::new(0)),
                ),
            },
        };
        assert!(journal.append(&ev).is_ok());

        // Drive the live book through the same sequence so we have a
        // ground-truth snapshot.
        let live_book: crate::OrderBook<()> = crate::OrderBook::new("TEST");
        for price in [100u128, 101, 102] {
            live_book
                .add_order(OrderType::Standard {
                    id: Id::new_uuid(),
                    price: Price::new(price),
                    quantity: Quantity::new(10),
                    side: Side::Sell,
                    time_in_force: TimeInForce::Gtc,
                    user_id: Hash32::zero(),
                    timestamp: TimestampMs::new(0),
                    extra_fields: (),
                })
                .expect("seed ask");
        }
        // Note: live_book seeds with fresh UUIDs, so `snapshots_match`
        // compares the structural per-level fields (price, visible/hidden
        // quantity, order count) rather than order ids. Use the same notional
        // amount so the residual book state matches.
        let _ = live_book.match_market_order_by_amount(taker_id, 1_500, Side::Buy);

        // Replay journal into a fresh book.
        let (replayed, last_seq) =
            ReplayEngine::<()>::replay_from(&journal, 0, "TEST").expect("replay must succeed");
        assert_eq!(last_seq, seq);

        let live_snap = live_book.create_snapshot(usize::MAX);
        let replayed_snap = replayed.create_snapshot(usize::MAX);
        assert!(
            snapshots_match(&live_snap, &replayed_snap),
            "live and replayed snapshots must match after notional market order"
        );
    }

    /// #101: `replay_from_with_config` applies every config field to the fresh
    /// book before replaying. A `Default` config leaves the book at defaults;
    /// a populated config is reflected field-for-field.
    #[test]
    fn test_replay_from_with_config_applies_every_field() {
        let journal: InMemoryJournal<()> = InMemoryJournal::new();
        let ev = make_add_event(0, Id::new_uuid(), 100, 10, Side::Buy);
        assert!(journal.append(&ev).is_ok());

        // Default config => all-defaults book.
        let (book, _) = ReplayEngine::<()>::replay_from_with_config(
            &journal,
            0,
            "TEST",
            &ReplayBookConfig::default(),
        )
        .expect("default config replay");
        assert_eq!(book.fee_schedule(), None);
        assert_eq!(book.stp_mode(), STPMode::None);
        assert_eq!(book.tick_size(), None);
        assert_eq!(book.lot_size(), None);
        assert_eq!(book.min_order_size(), None);
        assert_eq!(book.max_order_size(), None);

        // Populated config => reflected on the reconstructed book. Price 100 is
        // on the 10-tick grid and qty 10 is a 5-lot multiple within [1, 1000].
        let fee = FeeSchedule::new(-2, 5);
        let config = ReplayBookConfig::new(
            Some(fee),
            STPMode::None,
            Some(10),
            Some(5),
            Some(1),
            Some(1_000),
        );
        let (book, _) = ReplayEngine::<()>::replay_from_with_config(&journal, 0, "TEST", &config)
            .expect("populated config replay");
        assert_eq!(book.fee_schedule(), Some(fee));
        assert_eq!(book.tick_size(), Some(10));
        assert_eq!(book.lot_size(), Some(5));
        assert_eq!(book.min_order_size(), Some(1));
        assert_eq!(book.max_order_size(), Some(1_000));
    }

    /// #101: the `*_with_config` variants share the pre-checks of the plain
    /// entry points — an empty journal is `EmptyJournal`, an out-of-range
    /// `from_sequence` is `InvalidSequence`.
    #[test]
    fn test_replay_with_config_pre_checks_match_plain_variants() {
        let empty: InMemoryJournal<()> = InMemoryJournal::new();
        assert!(matches!(
            ReplayEngine::<()>::replay_from_with_config(
                &empty,
                0,
                "TEST",
                &ReplayBookConfig::default()
            ),
            Err(ReplayError::EmptyJournal)
        ));

        let journal: InMemoryJournal<()> = InMemoryJournal::new();
        let ev = make_add_event(0, Id::new_uuid(), 100, 10, Side::Buy);
        assert!(journal.append(&ev).is_ok());
        let clock: Arc<dyn Clock> = Arc::new(StubClock::new());
        match ReplayEngine::<()>::replay_from_with_clock_and_config(
            &journal,
            5,
            "TEST",
            clock,
            &ReplayBookConfig::default(),
        ) {
            Err(ReplayError::InvalidSequence {
                from_sequence,
                last_sequence,
            }) => {
                assert_eq!(from_sequence, 5);
                assert_eq!(last_sequence, 0);
            }
            Err(other) => panic!("expected InvalidSequence, got {other:?}"),
            Ok(_) => panic!("expected InvalidSequence, got Ok(_)"),
        }
    }

    #[test]
    fn test_market_order_by_amount_command_serde_json_roundtrip() {
        let cmd: SequencerCommand<()> = SequencerCommand::MarketOrderByAmount {
            id: Id::new_uuid(),
            amount: 12_345_678,
            side: Side::Buy,
        };
        let json = serde_json::to_vec(&cmd).expect("serialize");
        let decoded: SequencerCommand<()> = serde_json::from_slice(&json).expect("deserialize");
        match decoded {
            SequencerCommand::MarketOrderByAmount { amount, side, .. } => {
                assert_eq!(amount, 12_345_678);
                assert_eq!(side, Side::Buy);
            }
            other => panic!("expected MarketOrderByAmount, got {other:?}"),
        }
    }

    #[cfg(feature = "bincode")]
    #[test]
    fn test_market_order_by_amount_command_bincode_roundtrip() {
        use bincode::config::standard;
        use bincode::serde::{decode_from_slice, encode_to_vec};
        let cmd: SequencerCommand<()> = SequencerCommand::MarketOrderByAmount {
            id: Id::new_uuid(),
            amount: 999_999,
            side: Side::Sell,
        };
        let bytes = encode_to_vec(&cmd, standard()).expect("encode");
        let (decoded, n) =
            decode_from_slice::<SequencerCommand<()>, _>(&bytes, standard()).expect("decode");
        assert_eq!(n, bytes.len());
        match decoded {
            SequencerCommand::MarketOrderByAmount { amount, side, .. } => {
                assert_eq!(amount, 999_999);
                assert_eq!(side, Side::Sell);
            }
            other => panic!("expected MarketOrderByAmount, got {other:?}"),
        }
    }
}