betex 0.4.0

//! High-performance internal book data structures.
//!
//! The public book API is event-sourced and deterministic. Internally, we want:
//! - O(1) cancel/replace by `OrderId`
//! - O(1) removal from per-price FIFO queues
//! - Fixed-size per-runner tick ladders (Betfair has 350 ticks)
//! - Deterministic iteration when emitting multi-event commands (close/remove-runner/etc.)
//!
//! This module provides:
//! - `OrderStore`: arena-backed storage keyed by `OrderId` with both fast and sorted indices
//! - `RunnerBook`: ladder-indexed levels with an intrusive FIFO per tick

use crate::book::common::types::{BookOrder, PriceSize, RunnerPrices};
use crate::book::protocol::command::Side;
use crate::book::protocol::reject::RejectReason;
use crate::types::{Money, OddsX10000, OrderId, RunnerId};
use serde::{Deserialize, Deserializer, Serialize, Serializer, de::Error as _};
use slab::Slab;
use std::collections::BTreeMap;
use std::ops::Bound;
use tracing::error;

const TICK_COUNT: usize = crate::types::odds::TICK_LADDER.len();

pub(crate) type OrderKey = usize;

#[derive(Debug, Clone, Copy, Default)]
struct Links {
    prev: Option<OrderKey>,
    next: Option<OrderKey>,
}

#[derive(Debug, Clone)]
struct OrderSlot {
    order: BookOrder,
    links: Links,
    in_level: bool,
    level_side: Side,
    level_tick: u16,
}

impl OrderSlot {
    fn try_new(order: BookOrder) -> Result<Self, RejectReason> {
        let Some(tick) = order.price.tick_index() else {
            return Err(RejectReason::InvalidOdds);
        };
        Ok(Self {
            level_side: order.info.side,
            level_tick: tick as u16,
            order,
            links: Links::default(),
            in_level: false,
        })
    }
}

/// Arena-backed order storage.
///
/// - Hot lookups: `hashbrown::HashMap<OrderId, OrderKey>`
/// - Deterministic iteration: `BTreeMap<OrderId, OrderKey>`
///
/// Serialize/deserialize as `BTreeMap<OrderId, BookOrder>` for stable snapshots.
#[derive(Debug, Clone, Default)]
pub(crate) struct OrderStore {
    slab: Slab<OrderSlot>,
    by_id: hashbrown::HashMap<OrderId, OrderKey>,
    by_id_sorted: BTreeMap<OrderId, OrderKey>,
}

impl OrderStore {
    pub fn with_capacity(capacity: usize) -> Self {
        Self {
            slab: Slab::with_capacity(capacity),
            by_id: hashbrown::HashMap::with_capacity(capacity),
            by_id_sorted: BTreeMap::new(),
        }
    }

    pub fn get_key(&self, order_id: &OrderId) -> Option<OrderKey> {
        self.by_id.get(order_id).copied()
    }

    pub fn contains(&self, order_id: &OrderId) -> bool {
        self.by_id.contains_key(order_id)
    }

    pub fn get(&self, order_id: &OrderId) -> Option<&BookOrder> {
        let key = self.get_key(order_id)?;
        Some(&self.slab.get(key)?.order)
    }

    pub fn get_mut(&mut self, order_id: &OrderId) -> Option<&mut BookOrder> {
        let key = self.get_key(order_id)?;
        Some(&mut self.slab.get_mut(key)?.order)
    }

    fn slot(&self, key: OrderKey) -> &OrderSlot {
        &self.slab[key]
    }

    fn slot_mut(&mut self, key: OrderKey) -> &mut OrderSlot {
        &mut self.slab[key]
    }

    pub fn order_by_key(&self, key: OrderKey) -> &BookOrder {
        &self.slab[key].order
    }

    pub fn stored_tick(&self, key: OrderKey) -> usize {
        self.slab[key].level_tick as usize
    }

    pub fn in_level_by_key(&self, key: OrderKey) -> bool {
        self.slab[key].in_level
    }

    pub fn insert(
        &mut self,
        order_id: OrderId,
        order: BookOrder,
    ) -> Result<OrderKey, RejectReason> {
        debug_assert_eq!(order_id, order.info.order_id);
        let key = self.slab.insert(OrderSlot::try_new(order)?);
        self.by_id.insert(order_id, key);
        self.by_id_sorted.insert(order_id, key);
        Ok(key)
    }

    pub fn iter_sorted(&self) -> impl Iterator<Item = (OrderId, &BookOrder)> {
        self.by_id_sorted
            .iter()
            .map(|(&oid, &key)| (oid, &self.slab[key].order))
    }

    pub fn iter_sorted_from(
        &self,
        cursor_after: Option<OrderId>,
    ) -> impl Iterator<Item = (OrderId, &BookOrder)> {
        let range = match cursor_after {
            Some(c) => (Bound::Excluded(c), Bound::Unbounded),
            None => (Bound::Unbounded, Bound::Unbounded),
        };
        self.by_id_sorted
            .range(range)
            .map(|(&oid, &key)| (oid, &self.slab[key].order))
    }

    pub fn iter_keys_sorted(&self) -> impl Iterator<Item = (OrderId, OrderKey, &BookOrder)> {
        self.by_id_sorted
            .iter()
            .map(|(&oid, &key)| (oid, key, &self.slab[key].order))
    }

    pub fn is_in_level(&self, order_id: &OrderId) -> bool {
        let Some(&key) = self.by_id.get(order_id) else {
            return false;
        };
        self.slab[key].in_level
    }

    pub fn remove(&mut self, order_id: &OrderId) -> Option<BookOrder> {
        let key = self.by_id.remove(order_id)?;
        self.by_id_sorted.remove(order_id);
        Some(self.slab.remove(key).order)
    }
}

#[cfg(test)]
impl OrderStore {
    pub(crate) fn debug_slab_len(&self) -> usize {
        self.slab.len()
    }

    pub(crate) fn debug_slab_capacity(&self) -> usize {
        self.slab.capacity()
    }
}

impl Serialize for OrderStore {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        use serde::ser::SerializeMap;
        let mut map = serializer.serialize_map(Some(self.by_id_sorted.len()))?;
        for (&oid, &key) in &self.by_id_sorted {
            map.serialize_entry(&oid, &self.slab[key].order)?;
        }
        map.end()
    }
}

impl<'de> Deserialize<'de> for OrderStore {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        let orders = BTreeMap::<OrderId, BookOrder>::deserialize(deserializer)?;
        let mut store = Self {
            slab: Slab::with_capacity(orders.len()),
            by_id: hashbrown::HashMap::with_capacity(orders.len()),
            by_id_sorted: BTreeMap::new(),
        };
        for (oid, order) in orders {
            let slot = OrderSlot::try_new(order)
                .map_err(|err| D::Error::custom(format!("invalid order: {err:?}")))?;
            let key = store.slab.insert(slot);
            store.by_id.insert(oid, key);
            store.by_id_sorted.insert(oid, key);
        }
        Ok(store)
    }
}

#[derive(Debug, Clone, Copy, Default)]
struct Bitset350 {
    words: [u64; 6],
}

impl Bitset350 {
    fn set(&mut self, idx: usize) {
        let w = idx / 64;
        let b = idx % 64;
        self.words[w] |= 1u64 << b;
    }

    fn clear(&mut self, idx: usize) {
        let w = idx / 64;
        let b = idx % 64;
        self.words[w] &= !(1u64 << b);
    }

    fn next_set_from(&self, start: usize) -> Option<usize> {
        if start >= TICK_COUNT {
            return None;
        }
        let mut w = start / 64;
        let mut bits = self.words[w] & (!0u64 << (start % 64));
        loop {
            if bits != 0 {
                return Some(w * 64 + bits.trailing_zeros() as usize);
            }
            w += 1;
            if w >= self.words.len() {
                return None;
            }
            bits = self.words[w];
        }
    }

    fn prev_set_from(&self, start: usize) -> Option<usize> {
        let start = start.min(TICK_COUNT.saturating_sub(1));
        let mut w = start / 64;
        let mut bits = self.words[w] & (!0u64 >> (63 - (start % 64)));
        loop {
            if bits != 0 {
                return Some(w * 64 + (63 - bits.leading_zeros() as usize));
            }
            if w == 0 {
                return None;
            }
            w -= 1;
            bits = self.words[w];
        }
    }
}

#[derive(Debug, Clone, Copy)]
pub(crate) struct Level {
    pub head: Option<OrderKey>,
    pub tail: Option<OrderKey>,
    pub total_remaining: Money,
}

impl Default for Level {
    fn default() -> Self {
        Self {
            head: None,
            tail: None,
            total_remaining: Money::zero(),
        }
    }
}

#[derive(Debug, Clone)]
pub(crate) struct SideBook {
    pub levels: [Level; TICK_COUNT],
    mask: Bitset350,
}

impl Default for SideBook {
    fn default() -> Self {
        Self {
            levels: std::array::from_fn(|_| Level::default()),
            mask: Bitset350::default(),
        }
    }
}

impl SideBook {
    fn level_mut(&mut self, tick: usize) -> &mut Level {
        &mut self.levels[tick]
    }

    fn level(&self, tick: usize) -> &Level {
        &self.levels[tick]
    }

    pub fn best_asc(&self) -> Option<usize> {
        self.mask.next_set_from(0).filter(|&i| i < TICK_COUNT)
    }

    pub fn best_desc(&self) -> Option<usize> {
        self.mask
            .prev_set_from(TICK_COUNT - 1)
            .filter(|&i| i < TICK_COUNT)
    }

    pub fn next_asc_from(&self, start: usize) -> Option<usize> {
        self.mask.next_set_from(start).filter(|&i| i < TICK_COUNT)
    }

    pub fn next_desc_from(&self, start: usize) -> Option<usize> {
        self.mask.prev_set_from(start).filter(|&i| i < TICK_COUNT)
    }

    /// Iterate over non-empty tick indices in ascending order.
    #[allow(dead_code)]
    pub fn ticks_asc(&self) -> TickIterAsc<'_> {
        TickIterAsc {
            book: self,
            current: self.best_asc(),
        }
    }

    /// Iterate over non-empty tick indices in descending order.
    #[allow(dead_code)]
    pub fn ticks_desc(&self) -> TickIterDesc<'_> {
        TickIterDesc {
            book: self,
            current: self.best_desc(),
        }
    }
}

/// Iterator over non-empty tick indices in ascending order.
#[allow(dead_code)]
pub struct TickIterAsc<'a> {
    book: &'a SideBook,
    current: Option<usize>,
}

impl Iterator for TickIterAsc<'_> {
    type Item = usize;

    fn next(&mut self) -> Option<Self::Item> {
        let tick = self.current?;
        self.current = self.book.next_asc_from(tick + 1);
        Some(tick)
    }
}

/// Iterator over non-empty tick indices in descending order.
#[allow(dead_code)]
pub struct TickIterDesc<'a> {
    book: &'a SideBook,
    current: Option<usize>,
}

impl Iterator for TickIterDesc<'_> {
    type Item = usize;

    fn next(&mut self) -> Option<Self::Item> {
        let tick = self.current?;
        self.current = if tick == 0 {
            None
        } else {
            self.book.next_desc_from(tick - 1)
        };
        Some(tick)
    }
}

#[derive(Debug, Clone, Default)]
pub(crate) struct RunnerBook {
    pub back: SideBook,
    pub lay: SideBook,
}

impl RunnerBook {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn runner_prices(&self, runner_id: RunnerId, depth: usize) -> RunnerPrices {
        let mut result = RunnerPrices {
            runner_id,
            available_to_back: Vec::with_capacity(depth),
            available_to_lay: Vec::with_capacity(depth),
        };

        // available_to_back shows LAY liquidity (ascending)
        let mut tick = self.lay.best_asc();
        while let Some(t) = tick {
            if result.available_to_back.len() >= depth {
                break;
            }
            let level = self.lay.level(t);
            if level.total_remaining.0 > 0 {
                result.available_to_back.push(PriceSize {
                    price: OddsX10000(crate::types::odds::TICK_LADDER[t]),
                    size: level.total_remaining,
                });
            }
            tick = self.lay.next_asc_from(t + 1);
        }

        // available_to_lay shows BACK liquidity (descending)
        let mut tick = self.back.best_desc();
        while let Some(t) = tick {
            if result.available_to_lay.len() >= depth {
                break;
            }
            let level = self.back.level(t);
            if level.total_remaining.0 > 0 {
                result.available_to_lay.push(PriceSize {
                    price: OddsX10000(crate::types::odds::TICK_LADDER[t]),
                    size: level.total_remaining,
                });
            }
            if t == 0 {
                break;
            }
            tick = self.back.next_desc_from(t - 1);
        }

        result
    }

    pub fn insert_tail(&mut self, orders: &mut OrderStore, key: OrderKey, remaining: Money) {
        let (tick, side) = {
            let s = orders.slot(key);
            (s.level_tick as usize, s.level_side)
        };
        let book = match side {
            Side::Yes => &mut self.back,
            Side::No => &mut self.lay,
        };
        if book.levels[tick].head.is_none() {
            book.mask.set(tick);
            book.levels[tick].head = Some(key);
            book.levels[tick].tail = Some(key);
        } else {
            let Some(tail) = book.levels[tick].tail else {
                error!(tick, "tail missing for non-empty level");
                return;
            };
            orders.slot_mut(tail).links.next = Some(key);
            orders.slot_mut(key).links.prev = Some(tail);
            book.levels[tick].tail = Some(key);
        }
        book.levels[tick].total_remaining =
            Money(book.levels[tick].total_remaining.0 + remaining.0);
        let slot = orders.slot_mut(key);
        slot.in_level = true;
        slot.links.next = None;
    }

    pub fn unlink(&mut self, orders: &mut OrderStore, key: OrderKey, remaining: Money) {
        let (tick, side, prev, next) = {
            let slot = orders.slot(key);
            (
                slot.level_tick as usize,
                slot.level_side,
                slot.links.prev,
                slot.links.next,
            )
        };
        let book = match side {
            Side::Yes => &mut self.back,
            Side::No => &mut self.lay,
        };
        let became_empty = {
            let level = book.level_mut(tick);

            if level.head == Some(key) {
                level.head = next;
            }
            if level.tail == Some(key) {
                level.tail = prev;
            }
            if let Some(p) = prev {
                orders.slot_mut(p).links.next = next;
            }
            if let Some(n) = next {
                orders.slot_mut(n).links.prev = prev;
            }

            level.total_remaining = Money(level.total_remaining.0.saturating_sub(remaining.0));
            level.head.is_none()
        };
        if became_empty {
            book.mask.clear(tick);
        }

        let slot = orders.slot_mut(key);
        slot.links = Links::default();
        slot.in_level = false;
    }

    pub fn level_total_remaining(&self, side: Side, tick: usize) -> Money {
        let book = match side {
            Side::Yes => &self.back,
            Side::No => &self.lay,
        };
        book.level(tick).total_remaining
    }

    pub fn level_head(&self, side: Side, tick: usize) -> Option<OrderKey> {
        let book = match side {
            Side::Yes => &self.back,
            Side::No => &self.lay,
        };
        book.level(tick).head
    }

    pub fn best_tick_asc(&self, side: Side) -> Option<usize> {
        let book = match side {
            Side::Yes => &self.back,
            Side::No => &self.lay,
        };
        book.best_asc()
    }

    pub fn best_tick_desc(&self, side: Side) -> Option<usize> {
        let book = match side {
            Side::Yes => &self.back,
            Side::No => &self.lay,
        };
        book.best_desc()
    }

    pub fn next_tick_asc_from(&self, side: Side, start: usize) -> Option<usize> {
        let book = match side {
            Side::Yes => &self.back,
            Side::No => &self.lay,
        };
        book.next_asc_from(start)
    }

    pub fn next_tick_desc_from(&self, side: Side, start: usize) -> Option<usize> {
        let book = match side {
            Side::Yes => &self.back,
            Side::No => &self.lay,
        };
        book.next_desc_from(start)
    }

    pub fn iter_level_keys<'a>(
        &'a self,
        side: Side,
        tick: usize,
        orders: &'a OrderStore,
    ) -> LevelKeyIter<'a> {
        LevelKeyIter {
            orders,
            cur: self.level_head(side, tick),
        }
    }

    pub fn decrement_level_total(&mut self, side: Side, tick: usize, delta: Money) {
        let book = match side {
            Side::Yes => &mut self.back,
            Side::No => &mut self.lay,
        };
        let level = book.level_mut(tick);
        level.total_remaining = Money(level.total_remaining.0.saturating_sub(delta.0));
    }
}

pub(crate) struct LevelKeyIter<'a> {
    orders: &'a OrderStore,
    cur: Option<OrderKey>,
}

impl<'a> Iterator for LevelKeyIter<'a> {
    type Item = OrderKey;

    fn next(&mut self) -> Option<Self::Item> {
        let cur = self.cur?;
        self.cur = self.orders.slab[cur].links.next;
        Some(cur)
    }
}

#[cfg(test)]
mod proptest_tests {
    use super::*;
    use crate::book::common::types::{BookOrder, BookOrderInfo, BookOrderState};
    use crate::book::protocol::command::Persistence;
    use crate::types::{AccountId, unix_epoch};
    use proptest::prelude::*;

    fn arb_order_id() -> impl Strategy<Value = OrderId> {
        (1u64..=10000).prop_map(OrderId)
    }

    fn arb_runner_id() -> impl Strategy<Value = RunnerId> {
        (1u32..=10).prop_map(RunnerId)
    }

    fn arb_side() -> impl Strategy<Value = Side> {
        prop_oneof![Just(Side::Yes), Just(Side::No)]
    }

    fn arb_tick_index() -> impl Strategy<Value = usize> {
        0..TICK_COUNT
    }

    fn arb_stake() -> impl Strategy<Value = Money> {
        (1i64..=1_000_000).prop_map(Money)
    }

    fn make_test_order(
        order_id: OrderId,
        runner_id: RunnerId,
        side: Side,
        tick: usize,
    ) -> BookOrder {
        let price = OddsX10000(crate::types::odds::TICK_LADDER[tick]);
        BookOrder {
            info: BookOrderInfo {
                order_id,
                account_id: AccountId(1),
                side,
                state: BookOrderState::ExecutableUnmatched,
                created_at: unix_epoch(),
                last_updated_at: unix_epoch(),
            },
            runner_id,
            price,
            stake: Money(1000),
            matched: Money(0),
            persistence: Persistence::Persist,
        }
    }

    proptest! {
        #![proptest_config(ProptestConfig::with_cases(100))]

        /// Verify that HashMap, BTreeMap, and slab always agree on order presence.
        #[test]
        fn indices_stay_in_sync(
            order_ids in prop::collection::vec(arb_order_id(), 1..50),
            runner_ids in prop::collection::vec(arb_runner_id(), 1..50),
            sides in prop::collection::vec(arb_side(), 1..50),
            ticks in prop::collection::vec(arb_tick_index(), 1..50),
            remove_indices in prop::collection::vec(0usize..50, 0..25),
        ) {
            let mut store = OrderStore::default();
            let mut inserted = Vec::new();

            // Insert orders
            let n = order_ids.len().min(runner_ids.len()).min(sides.len()).min(ticks.len());
            for i in 0..n {
                let order = make_test_order(order_ids[i], runner_ids[i], sides[i], ticks[i]);
                if !store.contains(&order_ids[i]) {
                    let _ = store.insert(order_ids[i], order);
                    inserted.push(order_ids[i]);
                }
            }

            // Verify indices agree after insertions
            for &oid in &inserted {
                prop_assert!(store.by_id.contains_key(&oid));
                prop_assert!(store.by_id_sorted.contains_key(&oid));
                let key = store.get_key(&oid).unwrap();
                prop_assert!(store.slab.contains(key));
            }

            // Remove some orders
            for &idx in &remove_indices {
                if idx < inserted.len() {
                    let oid = inserted[idx];
                    store.remove(&oid);
                }
            }

            // Verify indices still agree
            for (&oid, &key) in &store.by_id {
                prop_assert!(store.by_id_sorted.contains_key(&oid));
                prop_assert!(store.slab.contains(key));
            }
            for (&oid, &key) in &store.by_id_sorted {
                prop_assert!(store.by_id.contains_key(&oid));
                prop_assert!(store.slab.contains(key));
            }
            prop_assert_eq!(store.by_id.len(), store.by_id_sorted.len());
        }

        /// Verify slab capacity doesn't grow unbounded with insert/remove cycles.
        #[test]
        fn slab_reuse_bounded(
            ops in prop::collection::vec(
                prop_oneof![
                    arb_order_id().prop_map(|id| (true, id)),
                    arb_order_id().prop_map(|id| (false, id)),
                ],
                1..200
            ),
        ) {
            let mut store = OrderStore::default();
            let mut max_capacity = 0usize;
            let mut current_count = 0usize;

            for (is_insert, order_id) in ops {
                if is_insert {
                    if !store.contains(&order_id) {
                        let order =
                            make_test_order(order_id, RunnerId(1), Side::Yes, 100);
                        let _ = store.insert(order_id, order);
                        current_count += 1;
                    }
                } else {
                    if store.remove(&order_id).is_some() {
                        current_count = current_count.saturating_sub(1);
                    }
                }
                max_capacity = max_capacity.max(store.debug_slab_capacity());
            }

            // Slab capacity should not exceed 2x the maximum concurrent count
            let max_concurrent = store.debug_slab_len().max(current_count);
            // Account for slab's growth strategy (typically doubles)
            prop_assert!(
                max_capacity <= (max_concurrent + 1).next_power_of_two() * 2,
                "Capacity {} too large for max concurrent {} orders",
                max_capacity,
                max_concurrent
            );
        }

        /// Verify level totals equal the sum of order remainings.
        #[test]
        fn level_totals_consistent(
            ticks in prop::collection::vec(arb_tick_index(), 1..30),
            sides in prop::collection::vec(arb_side(), 1..30),
            stakes in prop::collection::vec(arb_stake(), 1..30),
        ) {
            let mut store = OrderStore::default();
            let mut runner_book = RunnerBook::new();

            let n = ticks.len().min(sides.len()).min(stakes.len());
            for i in 0..n {
                let order_id = OrderId(i as u64 + 1);
                let mut order = make_test_order(order_id, RunnerId(1), sides[i], ticks[i]);
                order.stake = stakes[i];
                let key = store.insert(order_id, order).expect("valid order insert");
                runner_book.insert_tail(&mut store, key, stakes[i]);
            }

            // Check level totals
            for tick in 0..TICK_COUNT {
                for side in [Side::Yes, Side::No] {
                    let level_total = runner_book.level_total_remaining(side, tick);
                    let sum: i64 = runner_book
                        .iter_level_keys(side, tick, &store)
                        .map(|key| store.order_by_key(key).stake.0)
                        .sum();
                    prop_assert_eq!(
                        level_total.0, sum,
                        "Level total mismatch at tick {} {:?}: {} vs {}",
                        tick, side, level_total.0, sum
                    );
                }
            }
        }

        /// Verify orders at the same price iterate in insertion (FIFO) order.
        #[test]
        fn fifo_ordering_preserved(
            order_count in 2usize..20,
            tick in arb_tick_index(),
            side in arb_side(),
        ) {
            let mut store = OrderStore::default();
            let mut runner_book = RunnerBook::new();
            let mut insertion_order = Vec::new();

            for i in 0..order_count {
                let order_id = OrderId(i as u64 + 1);
                let order = make_test_order(order_id, RunnerId(1), side, tick);
                let key = store.insert(order_id, order).expect("valid order insert");
                runner_book.insert_tail(&mut store, key, Money(1000));
                insertion_order.push(order_id);
            }

            // Iterate and verify FIFO order
            let iterated: Vec<OrderId> = runner_book
                .iter_level_keys(side, tick, &store)
                .map(|key| store.order_by_key(key).info.order_id)
                .collect();

            prop_assert_eq!(
                insertion_order, iterated,
                "FIFO order not preserved"
            );
        }
    }
}