betex 0.4.0

//! Shared helpers for the market close process.
//!
//! ## Close Process Overview
//!
//! When a market closes, all resting orders must be cancelled. To avoid
//! emitting unbounded events in a single transaction, the close process
//! uses **batched cancellation**:
//!
//! 1. `CloseMarket` command starts the close, cancelling up to `batch_max_events` orders
//! 2. If orders remain, market enters `Closing` state with [`CloseProcessState`]
//! 3. `ContinueCloseMarket` commands process additional batches
//! 4. Final batch transitions market to `Closed` state
//!
//! ## Cursor Semantics
//!
//! The `cursor_after` field in [`CloseProcessState`] and [`BookEvent::MarketOrdersSettled`]
//! tracks pagination through the order set:
//!
//! - `None` on the first batch (start from beginning)
//! - `Some(order_id)` indicates the last processed order
//! - Subsequent batches resume from orders greater than the cursor
//!
//! This cursor-based pagination ensures **idempotent recovery**: replaying
//! events that have already been applied is safe because the cursor prevents
//! re-processing orders.
//!
//! ## Recovery Guarantees
//!
//! The close process is designed for crash recovery:
//!
//! - **No double cancellation**: Orders before the cursor are skipped on replay
//! - **Deterministic ordering**: Orders are processed in BTreeMap order (by OrderId)
//! - **Progress tracking**: `chunks_done` and `cancelled_total` allow monitoring
//!
//! ## Event Budget
//!
//! The `batch_max_events` parameter controls how many order cancellations can
//! occur per batch. The minimum is [`MIN_CLOSE_BATCH_EVENTS`] (2) to ensure
//! at least the state-change event fits. Typical values are 1000-4096.

use crate::book::common::events::{BookEvent, BookEventEnvelope};
use crate::book::common::types::{BookMarketState, BookOrder, CloseProcessState};
use crate::types::OrderId;

// Account for MarketStateChanged + MarketOrdersSettled events
pub const MIN_CLOSE_BATCH_EVENTS: u16 = 2;
pub const DEFAULT_CLOSE_BATCH_EVENTS: u16 = 4096;

/// Encode the close batch size into the market-state-change reason so the close process can be
/// initialized deterministically during event application (no command-time mutation).
pub fn close_start_reason(batch_max_events: u16) -> String {
    format!("CLOSE_START:{batch_max_events}")
}

pub fn parse_close_start_batch_max_events(reason: &str) -> Option<u16> {
    reason
        .strip_prefix("CLOSE_START:")
        .and_then(|s| s.parse::<u16>().ok())
}

#[inline]
pub fn is_live_order(o: &BookOrder) -> bool {
    matches!(
        o.info.state,
        crate::book::common::types::BookOrderState::ExecutableUnmatched
            | crate::book::common::types::BookOrderState::ExecutablePartiallyMatched
    )
}

pub fn count_live_orders<'a>(orders: impl Iterator<Item = (OrderId, &'a BookOrder)>) -> u64 {
    orders.filter(|(_, o)| is_live_order(o)).count() as u64
}

pub fn close_start_batch(
    batch_max_events: u16,
    mut state_change: impl FnMut(BookMarketState, &str, &mut Vec<BookEventEnvelope>),
    emit: impl Fn(BookEvent) -> BookEventEnvelope,
    mut cancel_chunk: impl FnMut(Option<OrderId>, usize) -> (Vec<OrderId>, Option<OrderId>, bool, u64),
) -> Vec<BookEventEnvelope> {
    // Determine whether the close can complete in this initial batch before choosing the state
    // transition(s). This ensures the emitted event count never exceeds `batch_max_events`.
    //
    // Events emitted by this function:
    // - Non-final start batch: MarketStateChanged(to=Closing) + MarketOrdersSettled (2)
    // - Final start batch: MarketStateChanged(to=Closed) + MarketOrdersSettled (2)
    let cancel_budget = batch_max_events as usize;
    let (cancelled_order_ids, cursor_after, done, _cancelled_count) =
        cancel_chunk(None, cancel_budget);

    let mut events = Vec::new();

    if done {
        // Close can complete without entering the intermediate Closing state.
        state_change(BookMarketState::Closed, "CLOSE_DONE", &mut events);
        events.push(emit(BookEvent::MarketOrdersSettled {
            cursor_after: None, // Final batch
            order_ids: cancelled_order_ids,
            is_final: true,
        }));
    } else {
        // Enter Closing and persist the batch size in the reason for deterministic close-process init.
        let reason = close_start_reason(batch_max_events);
        state_change(BookMarketState::Closing, &reason, &mut events);
        events.push(emit(BookEvent::MarketOrdersSettled {
            cursor_after,
            order_ids: cancelled_order_ids,
            is_final: false,
        }));
    }

    events
}

pub fn close_continue_batch(
    proc_state: CloseProcessState,
    mut state_change: impl FnMut(BookMarketState, &str, &mut Vec<BookEventEnvelope>),
    emit: impl Fn(BookEvent) -> BookEventEnvelope,
    mut cancel_chunk: impl FnMut(Option<OrderId>, usize) -> (Vec<OrderId>, Option<OrderId>, bool, u64),
) -> Vec<BookEventEnvelope> {
    let mut events = Vec::new();

    let cancel_budget = proc_state.batch_max_events as usize; // No overhead
    let (cancelled_order_ids, cursor_after, done, _cancelled_count) =
        cancel_chunk(proc_state.cursor_after, cancel_budget);

    if done {
        state_change(BookMarketState::Closed, "CLOSE_DONE", &mut events);
        events.push(emit(BookEvent::MarketOrdersSettled {
            cursor_after: None, // Final batch
            order_ids: cancelled_order_ids,
            is_final: true,
        }));
    } else {
        events.push(emit(BookEvent::MarketOrdersSettled {
            cursor_after,
            order_ids: cancelled_order_ids,
            is_final: false,
        }));
    }

    events
}