betex 0.20.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)
    }
}

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) {
        if idx >= TICK_COUNT {
            return;
        }
        let w = idx / 64;
        let b = idx % 64;
        self.words[w] |= 1u64 << b;
    }

    fn clear(&mut self, idx: usize) {
        if idx >= TICK_COUNT {
            return;
        }
        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)
    }
}

#[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 ordered by bettor value:
        // highest odds first for backers.
        let mut tick = self.lay.best_desc();
        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,
                });
            }
            if t == 0 {
                break;
            }
            tick = self.lay.next_desc_from(t - 1);
        }

        // available_to_lay shows BACK liquidity ordered by bettor value:
        // lowest odds first for layers.
        let mut tick = self.back.best_asc();
        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,
                });
            }
            tick = self.back.next_asc_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 tests {
    use super::*;

    #[test]
    fn bitset350_set_out_of_range_does_not_panic() {
        let mut b = Bitset350::default();
        let res = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| b.set(400)));
        assert!(res.is_ok(), "set should ignore out-of-range indices");
        assert_eq!(b.next_set_from(0), None);
    }

    #[test]
    fn bitset350_clear_out_of_range_does_not_panic() {
        let mut b = Bitset350::default();
        b.set(10);
        let res = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| b.clear(400)));
        assert!(res.is_ok(), "clear should ignore out-of-range indices");
        assert_eq!(b.next_set_from(0), Some(10));
    }
}