hypercall/runtime/tasks/

bbo_snapshot.rs

use std::collections::{HashMap, VecDeque};
use std::sync::Arc;
use std::time::Duration;

use chrono::Utc;
use metrics::counter;
use rust_decimal::prelude::FromPrimitive;
use rust_decimal::Decimal;
use tokio::time::{interval_at, Instant, MissedTickBehavior};
use tracing::{error, info, warn};

use crate::read_cache::instruments_registry::InstrumentsCache;
use hypercall_db::{AnalyticsWriter, BboReferenceData, NewBboSnapshotInput};
use hypercall_runtime_api::{QuoteProvider, SnapshotBookQuote};

const SECONDS_PER_DAY: i64 = 86_400;

/// We only need ~24h of data for the reference ask lookup, but keep a small
/// buffer so that edge-of-window queries still resolve correctly.
const CACHE_RETENTION_SECS: i64 = 25 * 3600; // 25 hours

#[derive(Debug, Clone)]
pub struct BboSnapshotTaskConfig {
    pub interval_secs: u64,
    pub retention_days: i64,
}

impl Default for BboSnapshotTaskConfig {
    fn default() -> Self {
        Self {
            interval_secs: 300,
            retention_days: 7,
        }
    }
}

impl BboSnapshotTaskConfig {
    pub fn from_config(config: &crate::backend_config::BboSnapshotRuntimeConfig) -> Self {
        let interval_secs = config.interval_secs;
        let retention_days = config.retention_days;
        assert!(
            interval_secs > 0,
            "background_tasks.bbo_snapshot.interval_secs must be > 0"
        );
        assert!(
            retention_days > 0,
            "background_tasks.bbo_snapshot.retention_days must be > 0"
        );

        Self {
            interval_secs,
            retention_days,
        }
    }

    fn retention_secs(&self) -> i64 {
        self.retention_days
            .checked_mul(SECONDS_PER_DAY)
            .expect("BBO snapshot retention days overflowed seconds conversion")
    }
}

#[derive(Clone)]
pub struct BboSnapshotService {
    instruments_cache: Arc<InstrumentsCache>,
    quote_provider: Arc<dyn QuoteProvider>,
    db: Arc<dyn AnalyticsWriter>,
    config: BboSnapshotTaskConfig,
    in_memory_snapshots: Arc<tokio::sync::RwLock<HashMap<String, VecDeque<CachedBboSnapshot>>>>,
}

#[derive(Debug, Clone)]
struct CachedBboSnapshot {
    snapshot_ts: i64,
    best_ask: Decimal,
}

impl BboSnapshotService {
    pub fn new(
        instruments_cache: Arc<InstrumentsCache>,
        quote_provider: Arc<dyn QuoteProvider>,
        db: Arc<dyn AnalyticsWriter>,
        config: BboSnapshotTaskConfig,
    ) -> Self {
        Self {
            instruments_cache,
            quote_provider,
            db,
            config,
            in_memory_snapshots: Arc::new(tokio::sync::RwLock::new(HashMap::new())),
        }
    }

    /// Bootstrap the in-memory cache from DB so that `price_change` is
    /// available immediately after restart (no 24h gap).
    pub async fn bootstrap_from_db(&self) {
        let cutoff_ts = Utc::now().timestamp() - CACHE_RETENTION_SECS;
        match self.db.load_bbo_snapshots_since(cutoff_ts).await {
            Ok(rows) => {
                let mut cache = self.in_memory_snapshots.write().await;
                for row in &rows {
                    let series = cache
                        .entry(row.symbol.clone())
                        .or_insert_with(VecDeque::new);
                    series.push_back(CachedBboSnapshot {
                        snapshot_ts: row.snapshot_ts,
                        best_ask: row.best_ask,
                    });
                }
                info!(
                    symbols = cache.len(),
                    rows = rows.len(),
                    "BBO snapshot cache bootstrapped from DB"
                );
            }
            Err(e) => {
                warn!(
                    error = %e,
                    "Failed to bootstrap BBO cache from DB; price_change will be null until snapshots accumulate"
                );
            }
        }
    }

    /// Directly seed the in-memory cache. Used by integration tests that insert
    /// BBO rows via `upsert_bbo_snapshots` and need the handler to see them.
    pub async fn seed_snapshots(&self, snapshots: &[NewBboSnapshotInput]) {
        let mut cache = self.in_memory_snapshots.write().await;
        for snapshot in snapshots {
            let series = cache
                .entry(snapshot.symbol.clone())
                .or_insert_with(VecDeque::new);

            // Insert in sorted order using binary search
            let pos = series
                .binary_search_by_key(&snapshot.snapshot_ts, |s| s.snapshot_ts)
                .unwrap_or_else(|pos| pos);

            if pos < series.len() && series[pos].snapshot_ts == snapshot.snapshot_ts {
                // Duplicate timestamp -- update in place
                series[pos].best_ask = snapshot.best_ask;
            } else {
                series.insert(
                    pos,
                    CachedBboSnapshot {
                        snapshot_ts: snapshot.snapshot_ts,
                        best_ask: snapshot.best_ask,
                    },
                );
            }
        }
    }

    pub async fn get_reference_asks(
        &self,
        symbols: &[String],
        cutoff_ts: i64,
    ) -> HashMap<String, BboReferenceData> {
        if symbols.is_empty() {
            return HashMap::new();
        }

        let snapshots = self.in_memory_snapshots.read().await;
        build_reference_asks_from_cache(&snapshots, symbols, cutoff_ts)
    }

    pub fn start_with_shutdown(
        self: Arc<Self>,
        mut shutdown_rx: tokio::sync::broadcast::Receiver<()>,
    ) -> tokio::task::JoinHandle<()> {
        tokio::spawn(async move {
            info!(
                "Starting BBO snapshot service (interval_secs={}, retention_days={})",
                self.config.interval_secs, self.config.retention_days
            );

            self.bootstrap_from_db().await;

            let mut ticker = aligned_interval_secs(self.config.interval_secs);
            let mut consecutive_write_failures = 0_u64;
            let mut consecutive_cleanup_failures = 0_u64;

            loop {
                tokio::select! {
                    _ = shutdown_rx.recv() => {
                        info!("BBO snapshot service received shutdown signal");
                        break;
                    }
                    _ = ticker.tick() => {
                        let now_secs = Utc::now().timestamp();
                        self.capture_and_persist(now_secs, &mut consecutive_write_failures).await;
                        self.cleanup_retention(now_secs, &mut consecutive_cleanup_failures).await;
                    }
                }
            }

            info!("BBO snapshot service stopped");
        })
    }

    async fn capture_and_persist(&self, now_secs: i64, consecutive_failures: &mut u64) {
        let snapshot_ts = align_timestamp_secs(now_secs, self.config.interval_secs);
        let active_symbols: Vec<String> = self
            .instruments_cache
            .get_all()
            .await
            .into_iter()
            .filter(|inst| inst.status.is_active())
            .map(|inst| inst.id)
            .collect();

        counter!("ht_bbo_snapshot_rows_total", "status" => "attempted")
            .increment(active_symbols.len() as u64);

        let snapshots = build_snapshot_rows(&active_symbols, self.quote_provider.as_ref());
        let skipped = active_symbols.len().saturating_sub(snapshots.len());
        counter!("ht_bbo_snapshot_rows_total", "status" => "skipped_invalid")
            .increment(skipped as u64);

        if snapshots.is_empty() {
            return;
        }

        // Write to DB first -- only update in-memory cache on success so that
        // the cache never contains data that wasn't persisted.
        match self.db.upsert_bbo_snapshots(snapshot_ts, &snapshots).await {
            Ok(written_rows) => {
                *consecutive_failures = 0;
                counter!("ht_bbo_snapshot_rows_total", "status" => "written")
                    .increment(written_rows as u64);

                self.update_in_memory_snapshots(snapshot_ts, &snapshots)
                    .await;
            }
            Err(e) => {
                *consecutive_failures += 1;
                counter!("ht_bbo_snapshot_rows_total", "status" => "write_error").increment(1);
                if *consecutive_failures <= 3 || (*consecutive_failures).is_multiple_of(10) {
                    error!(
                        failures = *consecutive_failures,
                        snapshot_ts,
                        error = %e,
                        "Failed to persist BBO snapshot batch"
                    );
                }
            }
        }
    }

    async fn cleanup_retention(&self, now_secs: i64, consecutive_failures: &mut u64) {
        let cutoff_ts = now_secs - self.config.retention_secs();
        // Prune in-memory cache more aggressively -- we only need ~25h of data.
        let cache_cutoff = now_secs - CACHE_RETENTION_SECS;
        self.prune_in_memory_snapshots(cache_cutoff).await;
        match self.db.delete_bbo_snapshots_older_than(cutoff_ts).await {
            Ok(deleted_rows) => {
                *consecutive_failures = 0;
                counter!("ht_bbo_snapshot_retention_rows_deleted_total")
                    .increment(deleted_rows as u64);
            }
            Err(e) => {
                *consecutive_failures += 1;
                counter!("ht_bbo_snapshot_retention_cleanup_total", "status" => "error")
                    .increment(1);
                if *consecutive_failures <= 3 || (*consecutive_failures).is_multiple_of(10) {
                    error!(
                        failures = *consecutive_failures,
                        cutoff_ts,
                        error = %e,
                        "Failed to cleanup BBO snapshots retention"
                    );
                }
            }
        }
    }

    async fn update_in_memory_snapshots(
        &self,
        snapshot_ts: i64,
        snapshots: &[NewBboSnapshotInput],
    ) {
        let mut cache = self.in_memory_snapshots.write().await;
        for snapshot in snapshots {
            let series = cache
                .entry(snapshot.symbol.clone())
                .or_insert_with(VecDeque::new);

            match series.back() {
                Some(last) if last.snapshot_ts == snapshot_ts => {
                    // Same timestamp as the tail -- update in place.
                    series.back_mut().unwrap().best_ask = snapshot.best_ask;
                }
                Some(last) if last.snapshot_ts > snapshot_ts => {
                    // Out-of-order arrival -- insert in sorted position.
                    let pos = series
                        .binary_search_by_key(&snapshot_ts, |s| s.snapshot_ts)
                        .unwrap_or_else(|pos| pos);
                    series.insert(
                        pos,
                        CachedBboSnapshot {
                            snapshot_ts,
                            best_ask: snapshot.best_ask,
                        },
                    );
                }
                _ => {
                    series.push_back(CachedBboSnapshot {
                        snapshot_ts,
                        best_ask: snapshot.best_ask,
                    });
                }
            }
        }
    }

    async fn prune_in_memory_snapshots(&self, cutoff_ts: i64) {
        let mut cache = self.in_memory_snapshots.write().await;
        cache.retain(|_, series| {
            while series
                .front()
                .is_some_and(|snapshot| snapshot.snapshot_ts < cutoff_ts)
            {
                let _ = series.pop_front();
            }
            !series.is_empty()
        });
    }
}

fn build_reference_asks_from_cache(
    snapshots: &HashMap<String, VecDeque<CachedBboSnapshot>>,
    symbols: &[String],
    cutoff_ts: i64,
) -> HashMap<String, BboReferenceData> {
    let mut references = HashMap::new();

    for symbol in symbols {
        if let Some(series) = snapshots.get(symbol) {
            if series.is_empty() {
                continue;
            }

            // Binary search for the last entry at or before cutoff_ts.
            // partition_point returns the first index where snapshot_ts > cutoff_ts,
            // so the entry just before it (if any) is the latest <= cutoff_ts.
            let partition = series.partition_point(|entry| entry.snapshot_ts <= cutoff_ts);
            let latest_before = if partition > 0 {
                Some(&series[partition - 1])
            } else {
                None
            };
            // Only use a real 24h-ago reference. If there's no snapshot
            // before the cutoff, the instrument is too new for a meaningful
            // price_change — skip it rather than comparing against a
            // minutes-old price that produces absurd percentages.
            if let Some(row) = latest_before {
                references.insert(
                    symbol.clone(),
                    BboReferenceData {
                        reference_ask: row.best_ask,
                        reference_ts: row.snapshot_ts,
                        used_earliest_fallback: false,
                    },
                );
            }
        }
    }

    references
}

fn build_snapshot_rows(
    symbols: &[String],
    quote_provider: &dyn QuoteProvider,
) -> Vec<NewBboSnapshotInput> {
    symbols
        .iter()
        .filter_map(|symbol| {
            quote_provider
                .get_quote(symbol)
                .and_then(|quote| snapshot_row_from_quote(symbol, &quote))
        })
        .collect()
}

fn snapshot_row_from_quote(symbol: &str, quote: &SnapshotBookQuote) -> Option<NewBboSnapshotInput> {
    let best_bid = quote.best_bid?;
    let best_ask = quote.best_ask?;
    if best_bid <= 0.0 || best_ask <= 0.0 {
        return None;
    }

    let best_bid_dec = Decimal::from_f64(best_bid)?;
    let best_ask_dec = Decimal::from_f64(best_ask)?;

    let best_bid_size = quote
        .best_bid_size
        .filter(|size| *size > 0.0)
        .and_then(Decimal::from_f64);
    let best_ask_size = quote
        .best_ask_size
        .filter(|size| *size > 0.0)
        .and_then(Decimal::from_f64);

    Some(NewBboSnapshotInput {
        symbol: symbol.to_string(),
        best_bid: best_bid_dec,
        best_ask: best_ask_dec,
        best_bid_size,
        best_ask_size,
        snapshot_ts: 0,
    })
}

fn aligned_interval_secs(interval_secs: u64) -> tokio::time::Interval {
    let now_secs = Utc::now().timestamp();
    let next_secs = next_bucket_start_secs(now_secs, interval_secs);
    let delay_secs = (next_secs - now_secs).max(0) as u64;

    let start_at = Instant::now() + Duration::from_secs(delay_secs);
    let mut ticker = interval_at(start_at, Duration::from_secs(interval_secs));
    ticker.set_missed_tick_behavior(MissedTickBehavior::Skip);
    ticker
}

fn align_timestamp_secs(timestamp_secs: i64, interval_secs: u64) -> i64 {
    let interval = i64::try_from(interval_secs).expect("interval_secs exceeded i64::MAX");
    timestamp_secs - (timestamp_secs % interval)
}

fn next_bucket_start_secs(timestamp_secs: i64, interval_secs: u64) -> i64 {
    align_timestamp_secs(timestamp_secs, interval_secs)
        + i64::try_from(interval_secs).expect("interval_secs exceeded i64::MAX")
}

#[async_trait::async_trait]
impl hypercall_api::caches::options_summary::BboReferenceAskReader for BboSnapshotService {
    async fn get_reference_asks(
        &self,
        symbols: &[String],
        cutoff_ts: i64,
    ) -> HashMap<String, BboReferenceData> {
        self.get_reference_asks(symbols, cutoff_ts).await
    }
}

#[async_trait::async_trait]
impl crate::shared::service::Service for BboSnapshotService {
    fn name(&self) -> &'static str {
        "BboSnapshotService"
    }

    fn owner(&self) -> crate::shared::service::ServiceOwner {
        crate::shared::service::ServiceOwner::Api
    }

    async fn run(
        self: std::sync::Arc<Self>,
        shutdown: crate::shared::ShutdownRx,
    ) -> anyhow::Result<()> {
        self.start_with_shutdown(shutdown)
            .await
            .map_err(|e| anyhow::anyhow!("BboSnapshotService task failed: {:?}", e))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::rsm::engine_snapshot::MockQuoteProvider;
    use hypercall_runtime_api::SnapshotBookQuote;
    use std::collections::VecDeque;

    #[test]
    fn test_align_timestamp_secs() {
        let ts = 1_700_000_123_i64;
        let aligned = align_timestamp_secs(ts, 300);
        assert_eq!(aligned % 300, 0);
        assert!(aligned <= ts);
        assert!((ts - aligned) < 300);
    }

    #[test]
    fn test_build_snapshot_rows_filters_invalid_quotes() {
        let quote_provider = MockQuoteProvider::new();
        quote_provider.set_quote(
            "BTC-VALID",
            SnapshotBookQuote {
                best_bid: Some(100.0),
                best_bid_size: Some(1.0),
                best_ask: Some(101.0),
                best_ask_size: Some(2.0),
                mid: Some(100.5),
                bids: vec![(100.0, 1.0)],
                asks: vec![(101.0, 2.0)],
            },
        );
        quote_provider.set_quote(
            "BTC-NO-BID",
            SnapshotBookQuote {
                best_bid: None,
                best_bid_size: None,
                best_ask: Some(101.0),
                best_ask_size: Some(2.0),
                mid: Some(101.0),
                bids: vec![],
                asks: vec![(101.0, 2.0)],
            },
        );
        quote_provider.set_quote(
            "BTC-NON-POSITIVE",
            SnapshotBookQuote {
                best_bid: Some(0.0),
                best_bid_size: Some(1.0),
                best_ask: Some(101.0),
                best_ask_size: Some(2.0),
                mid: Some(50.5),
                bids: vec![(0.0, 1.0)],
                asks: vec![(101.0, 2.0)],
            },
        );

        let symbols = vec![
            "BTC-VALID".to_string(),
            "BTC-NO-BID".to_string(),
            "BTC-NON-POSITIVE".to_string(),
            "BTC-MISSING".to_string(),
        ];

        let rows = build_snapshot_rows(&symbols, &quote_provider);
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0].symbol, "BTC-VALID");
        assert_eq!(rows[0].best_bid.to_string(), "100");
        assert_eq!(rows[0].best_ask.to_string(), "101");
    }

    #[test]
    fn test_retention_secs_default() {
        let cfg = BboSnapshotTaskConfig::default();
        assert_eq!(cfg.retention_secs(), 7 * SECONDS_PER_DAY);
    }

    #[tokio::test]
    async fn test_get_reference_asks_uses_latest_before_cutoff_skips_too_new() {
        let mut snapshots = HashMap::new();
        snapshots.insert(
            "BTC-1".to_string(),
            VecDeque::from(vec![
                CachedBboSnapshot {
                    snapshot_ts: 100,
                    best_ask: Decimal::from(10),
                },
                CachedBboSnapshot {
                    snapshot_ts: 200,
                    best_ask: Decimal::from(20),
                },
                CachedBboSnapshot {
                    snapshot_ts: 300,
                    best_ask: Decimal::from(30),
                },
            ]),
        );
        // ETH-1 only has a snapshot AFTER the cutoff — too new for a
        // meaningful 24h reference, so it should be skipped entirely.
        snapshots.insert(
            "ETH-1".to_string(),
            VecDeque::from(vec![CachedBboSnapshot {
                snapshot_ts: 500,
                best_ask: Decimal::from(50),
            }]),
        );

        let refs = build_reference_asks_from_cache(
            &snapshots,
            &["BTC-1".to_string(), "ETH-1".to_string()],
            250,
        );

        let btc = refs.get("BTC-1").expect("missing BTC reference");
        assert_eq!(btc.reference_ts, 200);
        assert_eq!(btc.reference_ask, Decimal::from(20));
        assert!(!btc.used_earliest_fallback);

        // ETH-1 should NOT have a reference — no snapshot before cutoff
        assert!(
            !refs.contains_key("ETH-1"),
            "instruments with no pre-cutoff snapshot should not get a reference"
        );
    }

    #[tokio::test]
    async fn test_exact_cutoff_boundary_returns_that_entry() {
        let mut snapshots = HashMap::new();
        snapshots.insert(
            "SYM".to_string(),
            VecDeque::from(vec![
                CachedBboSnapshot {
                    snapshot_ts: 100,
                    best_ask: Decimal::from(10),
                },
                CachedBboSnapshot {
                    snapshot_ts: 200,
                    best_ask: Decimal::from(20),
                },
            ]),
        );

        // cutoff_ts == snapshot_ts should match (<=)
        let refs = build_reference_asks_from_cache(&snapshots, &["SYM".to_string()], 200);
        let entry = refs
            .get("SYM")
            .expect("should find entry at exact boundary");
        assert_eq!(entry.reference_ts, 200);
        assert_eq!(entry.reference_ask, Decimal::from(20));
        assert!(!entry.used_earliest_fallback);
    }

    #[tokio::test]
    async fn test_duplicate_timestamp_upsert() {
        let mut snapshots = HashMap::new();
        snapshots.insert(
            "SYM".to_string(),
            VecDeque::from(vec![CachedBboSnapshot {
                snapshot_ts: 100,
                best_ask: Decimal::from(10),
            }]),
        );

        // Insert duplicate timestamp via update_in_memory_snapshots path
        let service_snapshots = Arc::new(tokio::sync::RwLock::new(snapshots));

        // Simulate the update logic directly
        {
            let mut cache = service_snapshots.write().await;
            let series = cache.get_mut("SYM").unwrap();
            // Same timestamp as back -- should update
            assert_eq!(series.back().unwrap().snapshot_ts, 100);
            series.back_mut().unwrap().best_ask = Decimal::from(99);
            assert_eq!(series.len(), 1);
            assert_eq!(series.back().unwrap().best_ask, Decimal::from(99));
        }
    }

    #[tokio::test]
    async fn test_prune_boundary() {
        let mut snapshots = HashMap::new();
        snapshots.insert(
            "SYM".to_string(),
            VecDeque::from(vec![
                CachedBboSnapshot {
                    snapshot_ts: 100,
                    best_ask: Decimal::from(10),
                },
                CachedBboSnapshot {
                    snapshot_ts: 200,
                    best_ask: Decimal::from(20),
                },
                CachedBboSnapshot {
                    snapshot_ts: 300,
                    best_ask: Decimal::from(30),
                },
            ]),
        );

        let cache = Arc::new(tokio::sync::RwLock::new(snapshots));

        // Prune with cutoff_ts = 200: entries with ts < 200 should be removed
        {
            let mut c = cache.write().await;
            c.retain(|_, series| {
                while series.front().is_some_and(|s| s.snapshot_ts < 200) {
                    let _ = series.pop_front();
                }
                !series.is_empty()
            });
        }

        let c = cache.read().await;
        let series = c.get("SYM").unwrap();
        assert_eq!(series.len(), 2);
        assert_eq!(series[0].snapshot_ts, 200); // ts=200 is NOT pruned (< 200 is false)
        assert_eq!(series[1].snapshot_ts, 300);
    }

    #[tokio::test]
    async fn test_seed_snapshots_inserts_in_sorted_order() {
        let _instruments_cache = Arc::new(InstrumentsCache::new());
        let _quote_provider: Arc<dyn QuoteProvider> = Arc::new(MockQuoteProvider::new());
        // We need a DieselDb but seed_snapshots doesn't use it -- construct with
        // a dummy URL that won't actually be connected to in this test.
        // Instead, just test the cache logic directly.
        let cache: Arc<tokio::sync::RwLock<HashMap<String, VecDeque<CachedBboSnapshot>>>> =
            Arc::new(tokio::sync::RwLock::new(HashMap::new()));

        // Simulate seed_snapshots logic
        let snapshots = vec![
            NewBboSnapshotInput {
                symbol: "A".to_string(),
                best_bid: Decimal::from(1),
                best_ask: Decimal::from(30),
                best_bid_size: None,
                best_ask_size: None,
                snapshot_ts: 300,
            },
            NewBboSnapshotInput {
                symbol: "A".to_string(),
                best_bid: Decimal::from(1),
                best_ask: Decimal::from(10),
                best_bid_size: None,
                best_ask_size: None,
                snapshot_ts: 100,
            },
            NewBboSnapshotInput {
                symbol: "A".to_string(),
                best_bid: Decimal::from(1),
                best_ask: Decimal::from(20),
                best_bid_size: None,
                best_ask_size: None,
                snapshot_ts: 200,
            },
        ];

        {
            let mut c = cache.write().await;
            for snapshot in &snapshots {
                let series = c
                    .entry(snapshot.symbol.clone())
                    .or_insert_with(VecDeque::new);
                let pos = series
                    .binary_search_by_key(&snapshot.snapshot_ts, |s| s.snapshot_ts)
                    .unwrap_or_else(|pos| pos);
                if pos < series.len() && series[pos].snapshot_ts == snapshot.snapshot_ts {
                    series[pos].best_ask = snapshot.best_ask;
                } else {
                    series.insert(
                        pos,
                        CachedBboSnapshot {
                            snapshot_ts: snapshot.snapshot_ts,
                            best_ask: snapshot.best_ask,
                        },
                    );
                }
            }
        }

        let c = cache.read().await;
        let series = c.get("A").unwrap();
        assert_eq!(series.len(), 3);
        assert_eq!(series[0].snapshot_ts, 100);
        assert_eq!(series[1].snapshot_ts, 200);
        assert_eq!(series[2].snapshot_ts, 300);
    }
}