hypercall/read_cache/

instruments_registry.rs

use anyhow::Result;
use async_trait::async_trait;
use rust_decimal::Decimal;
use rust_decimal_macros::dec;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::RwLock;
use tracing::{debug, error, info, warn};

use crate::messaging::EventBusTrait;
use crate::shared::option_token_address::derive_option_token_address;
use crate::shared::topics::TOPIC_MARKET_UPDATES;
use crate::snapshot::{SnapshotError, Snapshotable, SyncStatus};
use chrono::{DateTime, Utc};
use hypercall_types::api_models::{Instrument, InstrumentStatus};
use hypercall_types::{
    EngineMessage, Market, MarketUpdateMessage, MarketUpdateStatus, OptionType as MessageOptionType,
};
use hypercall_types::{TradingModes, WalletAddress};

/// Default interval for periodic DB refresh (60 seconds)
const DEFAULT_REFRESH_INTERVAL_SECS: u64 = 60;

/// Serializable instrument state for snapshots.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct InstrumentSnapshotState {
    pub instrument_id: i32,
    pub id: String,
    pub underlying: String,
    pub strike: Decimal,
    pub expiry: u64,
    pub option_type: String,
    #[serde(default)]
    pub option_token_address: Option<WalletAddress>,
    pub mark_iv: Option<Decimal>,
    pub volume_24h: Decimal,
    pub open_interest: Decimal,
    pub updated_at: i64, // Unix timestamp for serialization
    #[serde(default)]
    pub trading_mode: String,
    #[serde(default = "default_active_status")]
    pub status: String,
}

fn default_active_status() -> String {
    "ACTIVE".to_string()
}

impl From<&Instrument> for InstrumentSnapshotState {
    fn from(inst: &Instrument) -> Self {
        Self {
            instrument_id: inst.instrument_id,
            id: inst.id.clone(),
            underlying: inst.underlying.clone(),
            strike: inst.strike,
            expiry: inst.expiry,
            option_type: inst.option_type.clone(),
            option_token_address: inst.option_token_address,
            mark_iv: inst.mark_iv,
            volume_24h: inst.volume_24h,
            open_interest: inst.open_interest,
            updated_at: inst.updated_at.timestamp(),
            trading_mode: inst.trading_mode.as_db_str(),
            status: inst.status.as_db_str().to_string(),
        }
    }
}

impl InstrumentSnapshotState {
    pub fn into_instrument(self) -> Instrument {
        let state = self;
        let mut status = InstrumentStatus::from_db_str(&state.status).unwrap_or_else(|| {
            panic!(
                "CRITICAL_FAILURE: Unknown instrument status '{}' for {} in snapshot. Persisted data is corrupt.",
                state.status, state.id
            )
        });
        // If the stored status claims Active but the expiry has already passed,
        // override to ExpiredPendingPrice. This prevents stale snapshots (or old
        // snapshots without a status field) from briefly serving expired instruments
        // until the engine or periodic refresh corrects the state.
        if status.is_active() && is_expiry_passed(state.expiry) {
            status = InstrumentStatus::ExpiredPendingPrice;
        }
        Instrument {
            instrument_id: state.instrument_id,
            id: state.id,
            underlying: state.underlying,
            strike: state.strike,
            expiry: state.expiry,
            option_type: state.option_type,
            option_token_address: state.option_token_address,
            mark_iv: state.mark_iv,
            volume_24h: state.volume_24h,
            open_interest: state.open_interest,
            updated_at: DateTime::from_timestamp(state.updated_at, 0).unwrap_or_else(Utc::now),
            status,
            trading_mode: TradingModes::from_db_str(&state.trading_mode),
        }
    }
}

/// Binary serialization helpers for instrument snapshots.
pub mod serialization {
    use super::{default_active_status, InstrumentSnapshotState, SnapshotError};
    use rust_decimal::Decimal;
    use serde::Deserialize;

    pub fn serialize(state: &InstrumentSnapshotState) -> Result<Vec<u8>, SnapshotError> {
        bincode::serialize(state).map_err(|e| SnapshotError::Serialization(e.to_string()))
    }

    pub fn deserialize(data: &[u8]) -> Result<InstrumentSnapshotState, SnapshotError> {
        use tracing::warn;

        // Helper: bincode::deserialize already validates ALL bytes are consumed
        // (returns Err if trailing bytes remain), so truncated V4 data cannot
        // silently succeed as V3. We additionally log on every fallback so
        // schema drift is visible in metrics.

        // V4 (current): all fields including status
        if let Ok(state) = bincode::deserialize::<InstrumentSnapshotState>(data) {
            return Ok(state);
        }

        // V3: has trading_mode but no status
        #[derive(Deserialize)]
        struct V3 {
            instrument_id: i32,
            id: String,
            underlying: String,
            strike: Decimal,
            expiry: u64,
            option_type: String,
            option_token_address: Option<hypercall_types::WalletAddress>,
            mark_iv: Option<Decimal>,
            volume_24h: Decimal,
            open_interest: Decimal,
            updated_at: i64,
            trading_mode: String,
        }
        if let Ok(v) = bincode::deserialize::<V3>(data) {
            warn!(
                instrument_id = v.instrument_id,
                id = %v.id,
                "Deserialized instrument snapshot from V3 format (missing status field)"
            );
            return Ok(InstrumentSnapshotState {
                instrument_id: v.instrument_id,
                id: v.id,
                underlying: v.underlying,
                strike: v.strike,
                expiry: v.expiry,
                option_type: v.option_type,
                option_token_address: v.option_token_address,
                mark_iv: v.mark_iv,
                volume_24h: v.volume_24h,
                open_interest: v.open_interest,
                updated_at: v.updated_at,
                trading_mode: v.trading_mode,
                status: default_active_status(),
            });
        }

        // V2: has option_token_address but no trading_mode or status
        #[derive(Deserialize)]
        struct V2 {
            instrument_id: i32,
            id: String,
            underlying: String,
            strike: Decimal,
            expiry: u64,
            option_type: String,
            option_token_address: Option<hypercall_types::WalletAddress>,
            mark_iv: Option<Decimal>,
            volume_24h: Decimal,
            open_interest: Decimal,
            updated_at: i64,
        }
        if let Ok(v) = bincode::deserialize::<V2>(data) {
            warn!(
                instrument_id = v.instrument_id,
                id = %v.id,
                "Deserialized instrument snapshot from V2 format (missing trading_mode + status)"
            );
            return Ok(InstrumentSnapshotState {
                instrument_id: v.instrument_id,
                id: v.id,
                underlying: v.underlying,
                strike: v.strike,
                expiry: v.expiry,
                option_type: v.option_type,
                option_token_address: v.option_token_address,
                mark_iv: v.mark_iv,
                volume_24h: v.volume_24h,
                open_interest: v.open_interest,
                updated_at: v.updated_at,
                trading_mode: hypercall_types::TradingModes::ORDERBOOK.as_db_str(),
                status: default_active_status(),
            });
        }

        // V1 (oldest): no option_token_address, no trading_mode, no status
        #[derive(Deserialize)]
        struct V1 {
            instrument_id: i32,
            id: String,
            underlying: String,
            strike: Decimal,
            expiry: u64,
            option_type: String,
            mark_iv: Option<Decimal>,
            volume_24h: Decimal,
            open_interest: Decimal,
            updated_at: i64,
        }
        let v = bincode::deserialize::<V1>(data).map_err(|e| {
            SnapshotError::Serialization(format!(
                "Failed to deserialize instrument snapshot (tried v4/v3/v2/v1, {} bytes): {}",
                data.len(),
                e
            ))
        })?;
        warn!(
            instrument_id = v.instrument_id,
            id = %v.id,
            "Deserialized instrument snapshot from V1 format (missing option_token_address + trading_mode + status)"
        );
        Ok(InstrumentSnapshotState {
            instrument_id: v.instrument_id,
            id: v.id,
            underlying: v.underlying,
            strike: v.strike,
            expiry: v.expiry,
            option_type: v.option_type,
            option_token_address: None,
            mark_iv: v.mark_iv,
            volume_24h: v.volume_24h,
            open_interest: v.open_interest,
            updated_at: v.updated_at,
            trading_mode: hypercall_types::TradingModes::ORDERBOOK.as_db_str(),
            status: default_active_status(),
        })
    }
}

/// Check whether an expiry Unix timestamp has already passed.
fn is_expiry_passed(expiry_ts: u64) -> bool {
    let now = Utc::now().timestamp() as u64;
    expiry_ts <= now
}

/// Cache for fast instrument lookups
/// Provides O(1) access to instruments by symbol, underlying, expiry, and instrument id
pub struct InstrumentsCache {
    /// Primary storage: symbol -> Instrument
    instruments: Arc<RwLock<HashMap<String, Instrument>>>,

    /// Index: underlying -> Vec<symbol>
    by_underlying: Arc<RwLock<HashMap<String, Vec<String>>>>,

    /// Index: (underlying, expiry) -> Vec<symbol>
    by_underlying_expiry: Arc<RwLock<HashMap<(String, u64), Vec<String>>>>,

    /// Index: instrument id -> symbol
    by_instrument_id: Arc<RwLock<HashMap<i32, String>>>,

    /// Last processed market update sequence number
    last_market_update_seq: Arc<RwLock<i64>>,

    /// Sequence captured during last list_all() call for atomic snapshot consistency.
    /// This ensures snapshot_offsets() returns the same seq that was captured with the state.
    last_snapshot_seq: Arc<RwLock<i64>>,

    /// Sync status for readiness tracking
    sync_status: Arc<SyncStatus>,

    /// Flag to track if we've received at least one update (for readiness)
    has_received_update: AtomicBool,
}

impl Default for InstrumentsCache {
    fn default() -> Self {
        Self::new()
    }
}

impl InstrumentsCache {
    /// Create a new empty instruments cache
    pub fn new() -> Self {
        Self {
            instruments: Arc::new(RwLock::new(HashMap::new())),
            by_underlying: Arc::new(RwLock::new(HashMap::new())),
            by_underlying_expiry: Arc::new(RwLock::new(HashMap::new())),
            by_instrument_id: Arc::new(RwLock::new(HashMap::new())),
            last_market_update_seq: Arc::new(RwLock::new(0)),
            last_snapshot_seq: Arc::new(RwLock::new(0)),
            sync_status: Arc::new(SyncStatus::new()),
            has_received_update: AtomicBool::new(false),
        }
    }

    /// Get the sync status for readiness checks.
    pub fn sync_status(&self) -> Arc<SyncStatus> {
        self.sync_status.clone()
    }

    /// Check if the cache is ready to serve requests.
    pub fn is_ready(&self) -> bool {
        self.sync_status.is_ready()
    }

    /// Get snapshot offsets for the market updates topic.
    /// Returns the last processed sequence number as an offset.
    pub fn snapshot_offsets(&self) -> Result<HashMap<String, HashMap<i32, i64>>, SnapshotError> {
        // Use the snapshot seq captured during list_all() for consistency
        let seq = tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current()
                .block_on(async { *self.last_snapshot_seq.read().await })
        });

        let mut offsets = HashMap::new();
        let mut partition_offsets = HashMap::new();
        // Store next offset to process (seq + 1)
        partition_offsets.insert(0, seq + 1);
        offsets.insert(TOPIC_MARKET_UPDATES.to_string(), partition_offsets);
        Ok(offsets)
    }

    /// Apply snapshot offsets after restore.
    pub async fn apply_snapshot_offsets(
        &self,
        offsets: HashMap<String, HashMap<i32, i64>>,
    ) -> Result<(), SnapshotError> {
        if let Some(topic_offsets) = offsets.get(TOPIC_MARKET_UPDATES) {
            if let Some(&offset) = topic_offsets.get(&0) {
                // offset is next_offset_to_apply, so last processed is offset - 1
                let mut seq = self.last_market_update_seq.write().await;
                *seq = offset.saturating_sub(1);
            }
        }
        Ok(())
    }

    /// Update the last processed sequence number.
    pub async fn update_seq(&self, seq: i64) {
        let mut last_seq = self.last_market_update_seq.write().await;
        if seq > *last_seq {
            *last_seq = seq;
        }
    }

    /// Initialize the cache by loading instruments from DB and starting event listener (legacy, no shutdown support).
    /// For graceful shutdown support, use `initialize_with_shutdown`.
    pub async fn initialize(
        self: Arc<Self>,
        diesel_db: &Arc<dyn hypercall_db::BootstrapReader>,
        event_bus: Arc<dyn EventBusTrait>,
    ) -> Result<()> {
        let (tx, rx) = tokio::sync::broadcast::channel::<()>(1);
        std::mem::forget(tx);
        self.initialize_with_shutdown(diesel_db, event_bus, rx)
            .await
    }

    pub async fn initialize_with_shutdown(
        self: Arc<Self>,
        diesel_db: &Arc<dyn hypercall_db::BootstrapReader>,
        _event_bus: Arc<dyn EventBusTrait>,
        shutdown_rx: tokio::sync::broadcast::Receiver<()>,
    ) -> Result<()> {
        self.load_instruments_from_db(diesel_db.as_ref()).await?;

        self.sync_status.set_ready();
        self.has_received_update.store(true, Ordering::SeqCst);

        self.clone()
            .start_periodic_refresh(diesel_db.clone(), shutdown_rx)
            .await;
        info!("InstrumentsCache initialized with periodic DB refresh");
        Ok(())
    }

    async fn start_periodic_refresh(
        self: Arc<Self>,
        diesel_db: Arc<dyn hypercall_db::BootstrapReader>,
        mut shutdown_rx: tokio::sync::broadcast::Receiver<()>,
    ) {
        let refresh_interval = DEFAULT_REFRESH_INTERVAL_SECS;
        info!(
            "Starting InstrumentsCache periodic refresh (interval={}s)",
            refresh_interval
        );

        let cache = self.clone();
        let handle = tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_secs(refresh_interval));
            interval.tick().await;
            info!("InstrumentsCache periodic refresh task started");

            loop {
                tokio::select! {
                    result = shutdown_rx.recv() => {
                        info!(?result, "InstrumentsCache periodic refresh stopping");
                        break;
                    }
                    _ = interval.tick() => {
                        let count_before = cache.len().await;
                        match cache.reload_from_db(diesel_db.as_ref()).await {
                            Ok(()) => {
                                let count_after = cache.len().await;
                                info!(count_before, count_after, "InstrumentsCache periodic refresh completed");
                            }
                            Err(e) => {
                                error!("InstrumentsCache periodic refresh failed: {}", e);
                            }
                        }
                    }
                }
            }
        });

        tokio::spawn(async move {
            match handle.await {
                Ok(()) => info!("InstrumentsCache periodic refresh task exited normally"),
                Err(e) => error!("InstrumentsCache periodic refresh task died: {}", e),
            }
        });
    }

    /// Initialize the cache with snapshot support.
    ///
    /// Flow:
    /// 1. Try to load from snapshot (if available)
    /// 2. If snapshot exists: restore state, catchup from offsets, mark as ready
    /// 3. If no snapshot: load from DB as fallback, mark as ready
    /// 4. Start event listener for live updates
    pub async fn initialize_with_snapshot<L>(
        self: Arc<Self>,
        diesel_db: &Arc<dyn hypercall_db::BootstrapReader>,
        event_bus: Arc<dyn EventBusTrait>,
        loader: &L,
        shutdown_rx: tokio::sync::broadcast::Receiver<()>,
    ) -> Result<Option<HashMap<String, HashMap<i32, i64>>>>
    where
        L: crate::snapshot::SnapshotLoader<Key = String, State = InstrumentSnapshotState>,
    {
        use crate::snapshot::bootstrap_from_snapshot;

        let offsets = match bootstrap_from_snapshot(loader, self.as_ref()).await {
            Ok(Some((snapshot_id, offsets))) => {
                let instruments_count = self.len().await;
                if instruments_count == 0 {
                    warn!(
                        snapshot_id,
                        "Instruments snapshot restored 0 entries - falling back to database"
                    );
                    self.load_instruments_from_db(diesel_db.as_ref()).await?;
                    self.has_received_update.store(true, Ordering::SeqCst);
                    None
                } else {
                    self.apply_snapshot_offsets(offsets.clone()).await?;
                    self.has_received_update.store(true, Ordering::SeqCst);
                    info!(
                        snapshot_id,
                        instruments_count, "InstrumentsCache restored from snapshot"
                    );
                    Some(offsets)
                }
            }
            Ok(None) => {
                info!("No instruments snapshot found, loading from database");
                self.load_instruments_from_db(diesel_db.as_ref()).await?;
                self.has_received_update.store(true, Ordering::SeqCst);
                None
            }
            Err(e) => {
                let error_msg = e.to_string();
                warn!(error = %error_msg, "Failed to restore instruments snapshot, falling back to database");
                self.load_instruments_from_db(diesel_db.as_ref()).await?;
                self.has_received_update.store(true, Ordering::SeqCst);
                None
            }
        };

        if offsets.is_none() {
            self.sync_status.set_ready();
        }

        self.clone()
            .start_event_listeners_with_shutdown(event_bus, shutdown_rx.resubscribe())
            .await?;

        self.clone()
            .start_periodic_refresh(diesel_db.clone(), shutdown_rx)
            .await;

        let watchdog_cache = self.clone();
        let watchdog_db = diesel_db.clone();
        tokio::spawn(async move {
            for delay_secs in [30u64, 60] {
                tokio::time::sleep(Duration::from_secs(delay_secs)).await;
                let count = watchdog_cache.len().await;
                if count > 0 {
                    info!(
                        count,
                        delay_secs, "InstrumentsCache watchdog: cache is populated, exiting"
                    );
                    return;
                }
                warn!(
                    delay_secs,
                    "InstrumentsCache watchdog: cache still empty, reloading from DB"
                );
                match watchdog_cache.reload_from_db(watchdog_db.as_ref()).await {
                    Ok(()) => {
                        let new_count = watchdog_cache.len().await;
                        info!(
                            new_count,
                            delay_secs, "InstrumentsCache watchdog: DB reload complete"
                        );
                        if new_count > 0 {
                            return;
                        }
                    }
                    Err(e) => {
                        error!(error = %e, delay_secs, "InstrumentsCache watchdog: DB reload failed");
                    }
                }
            }
            warn!("InstrumentsCache watchdog: cache still empty after all retries");
        });

        Ok(offsets)
    }

    async fn load_instruments_from_db(
        &self,
        diesel_db: &dyn hypercall_db::BootstrapReader,
    ) -> Result<()> {
        use hypercall_types::expiry_date_to_timestamp;

        let db_count: i64 = diesel_db.get_instrument_count().await.unwrap_or(-1);

        if db_count < 0 {
            error!("Failed to count instruments in database - table may not exist");
        }

        let records = diesel_db.get_all_instruments().await?;

        // Convert from InstrumentRecord to API Instrument type, then build fresh indexes.
        // We fully rebuild off-lock and only swap after success, so transient DB
        // failures cannot poison the live cache with an empty state.
        let mut instruments: Vec<Instrument> = records
            .into_iter()
            .map(|r| {
                let expiry_ts = expiry_date_to_timestamp(&r.underlying, r.expiry as u64);
                let option_type_str = match r.option_type {
                    hypercall_types::OptionType::Call => "call".to_string(),
                    hypercall_types::OptionType::Put => "put".to_string(),
                };
                Instrument {
                    instrument_id: r.instrument_numeric_id,
                    id: r.id,
                    underlying: r.underlying,
                    strike: r.strike,
                    expiry: expiry_ts as u64,
                    option_type: option_type_str,
                    option_token_address: r.option_token_address,
                    mark_iv: None,
                    volume_24h: Decimal::ZERO,
                    open_interest: Decimal::ZERO,
                    updated_at: Utc::now(),
                    status: r.status,
                    trading_mode: TradingModes::from_db_str(&r.trading_mode),
                }
            })
            .collect();

        // Override status for instruments whose expiry has passed but whose DB
        // status hasn't been updated yet (startup race with the engine).
        for inst in &mut instruments {
            if inst.status.is_active() && is_expiry_passed(inst.expiry) {
                inst.status = InstrumentStatus::ExpiredPendingPrice;
            }
        }
        let mut rebuilt_instruments_map = HashMap::with_capacity(instruments.len());
        let mut rebuilt_by_underlying_map: HashMap<String, Vec<String>> = HashMap::new();
        let mut rebuilt_by_expiry_map: HashMap<(String, u64), Vec<String>> = HashMap::new();
        let mut rebuilt_by_id_map = HashMap::with_capacity(instruments.len());

        for instrument in instruments {
            let symbol = instrument.id.clone();
            let underlying = instrument.underlying.clone();
            let expiry = instrument.expiry;
            let instrument_id = instrument.instrument_id;

            rebuilt_instruments_map.insert(symbol.clone(), instrument);
            rebuilt_by_underlying_map
                .entry(underlying.clone())
                .or_default()
                .push(symbol.clone());
            rebuilt_by_expiry_map
                .entry((underlying, expiry))
                .or_default()
                .push(symbol.clone());
            rebuilt_by_id_map.insert(instrument_id, symbol);
        }

        let mut instruments_map = self.instruments.write().await;
        let mut by_underlying_map = self.by_underlying.write().await;
        let mut by_expiry_map = self.by_underlying_expiry.write().await;
        let mut by_id_map = self.by_instrument_id.write().await;

        *instruments_map = rebuilt_instruments_map;
        *by_underlying_map = rebuilt_by_underlying_map;
        *by_expiry_map = rebuilt_by_expiry_map;
        *by_id_map = rebuilt_by_id_map;

        let total_instruments = instruments_map.len();
        let total_underlyings = by_underlying_map.len();
        let underlyings: Vec<String> = by_underlying_map.keys().cloned().collect();

        info!(
            db_count,
            total_instruments,
            total_underlyings,
            ?underlyings,
            "InstrumentsCache loaded from database"
        );

        if total_instruments == 0 && db_count > 0 {
            warn!(
                db_count,
                "Database has instruments but none loaded - possible conversion error"
            );
        }

        Ok(())
    }

    /// Start event listeners for market updates with shutdown support.
    /// Uses `subscribe_with_offsets` to track offsets for snapshot consistency.
    async fn start_event_listeners_with_shutdown(
        self: Arc<Self>,
        event_bus: Arc<dyn EventBusTrait>,
        mut shutdown_rx: tokio::sync::broadcast::Receiver<()>,
    ) -> Result<()> {
        info!("Starting event listener for InstrumentsCache");

        let topics = vec![TOPIC_MARKET_UPDATES.to_string()];
        let mut receiver = event_bus
            .subscribe_with_offsets(topics)
            .await
            .map_err(|e| anyhow::anyhow!("Failed to subscribe to event topics: {}", e))?;

        // Spawn task to listen for market updates
        tokio::spawn(async move {
            loop {
                tokio::select! {
                    _ = shutdown_rx.recv() => {
                        info!("InstrumentsCache event listener received shutdown signal");
                        break;
                    }
                    maybe_envelope = receiver.recv() => {
                        match maybe_envelope {
                            Some(envelope) => {
                                let offset = envelope.offset;

                                match envelope.message {
                                    EngineMessage::MarketUpdate(market_update) => {
                                        if let Err(e) = self.handle_market_update(market_update).await {
                                            error!("Error handling market update: {}", e);
                                        } else if let Some(off) = offset {
                                            // Only advance seq after successful processing
                                            // so failed messages are retried on restart.
                                            self.update_seq(off).await;
                                        }
                                    }
                                    other => {
                                        debug!("InstrumentsCache ignoring non-MarketUpdate message: {:?}", std::mem::discriminant(&other));
                                    }
                                }
                            }
                            None => break, // Channel closed
                        }
                    }
                }
            }
            info!("InstrumentsCache event listener stopped");
        });

        info!("Event listener started for InstrumentsCache");
        Ok(())
    }

    /// Handle market update messages (create, delete, expire)
    /// Public to allow catchup replay from saved offsets.
    pub async fn handle_market_update(&self, message: MarketUpdateMessage) -> Result<()> {
        match message.status {
            MarketUpdateStatus::MarketCreated | MarketUpdateStatus::MarketAlreadyExists => {
                self.add_instrument(&message.market).await?;
                info!("Added instrument to cache: {}", message.market.symbol);
            }
            MarketUpdateStatus::MarketDeleted
            | MarketUpdateStatus::MarketExpired
            | MarketUpdateStatus::MarketPendingSettlement => {
                self.remove_instrument(&message.market.symbol).await?;
                info!("Removed instrument from cache: {}", message.market.symbol);
            }
            MarketUpdateStatus::MarketCreationFailed | MarketUpdateStatus::MarketDeletionFailed => {
                // Market lifecycle update failed, cache remains unchanged.
                debug!(
                    "Market lifecycle update failed for {}: not mutating cache",
                    message.market.symbol
                );
            }
        }
        Ok(())
    }

    /// Add a new instrument to the cache
    async fn add_instrument(&self, market: &Market) -> Result<()> {
        use hypercall_types::expiry_date_to_timestamp;

        let symbol = market.symbol.clone();
        let underlying = market.underlying.clone();
        // market.expiry is YYYYMMDD code from CatalogManager, convert to Unix timestamp
        let expiry = expiry_date_to_timestamp(&underlying, market.expiry) as u64;

        // Generate instrument_id (temporary until persisted IDs are plumbed through)
        let instruments_map = self.instruments.read().await;
        let max_id = instruments_map
            .values()
            .map(|i| i.instrument_id)
            .max()
            .unwrap_or(0);
        drop(instruments_map);

        let option_type = match market.option_type {
            MessageOptionType::Call => "call".to_string(),
            MessageOptionType::Put => "put".to_string(),
        };
        let option_token_address = Some(derive_option_token_address(
            &underlying,
            market.expiry,
            market.strike,
            &option_type,
        )?);

        let instrument = Instrument {
            instrument_id: max_id + 1,
            id: symbol.clone(),
            underlying: underlying.clone(),
            strike: market.strike,
            expiry,
            option_type,
            option_token_address,
            mark_iv: None,
            volume_24h: dec!(0),
            open_interest: dec!(0),
            updated_at: Utc::now(),
            status: InstrumentStatus::Active,
            trading_mode: TradingModes::ORDERBOOK,
        };
        let instrument_id = instrument.instrument_id;

        // Insert into primary storage
        let mut instruments_map = self.instruments.write().await;
        instruments_map.insert(symbol.clone(), instrument);
        drop(instruments_map);

        // Update by_underlying index
        let mut by_underlying_map = self.by_underlying.write().await;
        by_underlying_map
            .entry(underlying.clone())
            .or_insert_with(Vec::new)
            .push(symbol.clone());
        drop(by_underlying_map);

        // Update by_underlying_expiry index
        let mut by_expiry_map = self.by_underlying_expiry.write().await;
        by_expiry_map
            .entry((underlying, expiry))
            .or_insert_with(Vec::new)
            .push(symbol.clone());
        drop(by_expiry_map);

        let mut by_id_map = self.by_instrument_id.write().await;
        by_id_map.insert(instrument_id, symbol.clone());

        Ok(())
    }

    /// Remove an instrument from the cache
    async fn remove_instrument(&self, symbol: &str) -> Result<()> {
        // Get instrument details before removing
        let instruments_map = self.instruments.read().await;
        let instrument = match instruments_map.get(symbol) {
            Some(i) => i.clone(),
            None => return Ok(()), // Already removed
        };
        drop(instruments_map);

        let underlying = instrument.underlying.clone();
        let expiry = instrument.expiry;
        let instrument_id = instrument.instrument_id;

        // Remove from primary storage
        let mut instruments_map = self.instruments.write().await;
        instruments_map.remove(symbol);
        drop(instruments_map);

        // Remove from by_underlying index
        let mut by_underlying_map = self.by_underlying.write().await;
        if let Some(symbols) = by_underlying_map.get_mut(&underlying) {
            symbols.retain(|s| s != symbol);
            if symbols.is_empty() {
                by_underlying_map.remove(&underlying);
            }
        }
        drop(by_underlying_map);

        // Remove from by_underlying_expiry index
        let mut by_expiry_map = self.by_underlying_expiry.write().await;
        if let Some(symbols) = by_expiry_map.get_mut(&(underlying.clone(), expiry)) {
            symbols.retain(|s| s != symbol);
            if symbols.is_empty() {
                by_expiry_map.remove(&(underlying, expiry));
            }
        }
        drop(by_expiry_map);

        let mut by_id_map = self.by_instrument_id.write().await;
        by_id_map.remove(&instrument_id);

        Ok(())
    }

    /// Get instrument by symbol
    pub async fn get_by_symbol(&self, symbol: &str) -> Option<Instrument> {
        let instruments = self.instruments.read().await;
        instruments.get(symbol).cloned()
    }

    /// Synchronously check whether an instrument's trading mode includes RFQ.
    /// Returns false for unknown instruments so callers default to rejecting.
    pub fn allows_rfq(&self, symbol: &str) -> bool {
        self.instruments
            .try_read()
            .map(|guard| {
                guard
                    .get(symbol)
                    .map(|i| i.trading_mode.allows_rfq())
                    .unwrap_or(false)
            })
            .unwrap_or(false)
    }

    /// Get all instruments for a given underlying
    pub async fn get_by_underlying(&self, underlying: &str) -> Vec<Instrument> {
        let by_underlying = self.by_underlying.read().await;
        let available: Vec<&String> = by_underlying.keys().collect();
        let cache_total = self.instruments.read().await.len();

        let symbols = match by_underlying.get(underlying) {
            Some(s) => s.clone(),
            None => {
                warn!(
                    underlying,
                    ?available,
                    cache_total,
                    "No instruments found for underlying"
                );
                return Vec::new();
            }
        };
        drop(by_underlying);

        let instruments = self.instruments.read().await;
        let result: Vec<Instrument> = symbols
            .iter()
            .filter_map(|symbol| instruments.get(symbol).cloned())
            .collect();

        result
    }

    /// Get all instruments for a given underlying and expiry
    pub async fn get_by_underlying_and_expiry(
        &self,
        underlying: &str,
        expiry: u64,
    ) -> Vec<Instrument> {
        let by_expiry = self.by_underlying_expiry.read().await;
        let symbols = match by_expiry.get(&(underlying.to_string(), expiry)) {
            Some(s) => s.clone(),
            None => return Vec::new(),
        };
        drop(by_expiry);

        let instruments = self.instruments.read().await;
        symbols
            .iter()
            .filter_map(|symbol| instruments.get(symbol).cloned())
            .collect()
    }

    /// Get instrument by instrument id
    pub async fn get_by_instrument_id(&self, instrument_id: i32) -> Option<Instrument> {
        let by_id = self.by_instrument_id.read().await;
        let symbol = match by_id.get(&instrument_id) {
            Some(symbol) => symbol.clone(),
            None => return None,
        };
        drop(by_id);

        let instruments = self.instruments.read().await;
        instruments.get(&symbol).cloned()
    }

    /// Get all instruments (for debugging/admin)
    pub async fn get_all(&self) -> Vec<Instrument> {
        let instruments = self.instruments.read().await;
        instruments.values().cloned().collect()
    }

    /// Get count of cached instruments
    pub async fn len(&self) -> usize {
        let instruments = self.instruments.read().await;
        instruments.len()
    }

    /// Check if cache is empty
    pub async fn is_empty(&self) -> bool {
        let instruments = self.instruments.read().await;
        instruments.is_empty()
    }

    pub async fn reload_from_db(
        &self,
        diesel_db: &dyn hypercall_db::BootstrapReader,
    ) -> Result<()> {
        self.load_instruments_from_db(diesel_db).await
    }

    /// Add an instrument directly to the cache (used during snapshot restore).
    async fn add_instrument_direct(&self, instrument: Instrument) -> Result<()> {
        let symbol = instrument.id.clone();
        let underlying = instrument.underlying.clone();
        let expiry = instrument.expiry;
        let instrument_id = instrument.instrument_id;

        // Insert into primary storage
        let mut instruments_map = self.instruments.write().await;
        instruments_map.insert(symbol.clone(), instrument);
        drop(instruments_map);

        // Update by_underlying index
        let mut by_underlying_map = self.by_underlying.write().await;
        by_underlying_map
            .entry(underlying.clone())
            .or_insert_with(Vec::new)
            .push(symbol.clone());
        drop(by_underlying_map);

        // Update by_underlying_expiry index
        let mut by_expiry_map = self.by_underlying_expiry.write().await;
        by_expiry_map
            .entry((underlying, expiry))
            .or_insert_with(Vec::new)
            .push(symbol.clone());
        drop(by_expiry_map);

        let mut by_id_map = self.by_instrument_id.write().await;
        by_id_map.insert(instrument_id, symbol);

        Ok(())
    }
}

#[async_trait]
impl Snapshotable for InstrumentsCache {
    type Key = String;
    type State = InstrumentSnapshotState;

    /// Returns a cloned snapshot of all instrument entries.
    ///
    /// CRITICAL: We must capture the state AND the sequence atomically.
    /// This ensures snapshot_offsets() returns the same seq that was captured with the state.
    async fn list_all(&self) -> Result<HashMap<Self::Key, Self::State>, SnapshotError> {
        // Acquire BOTH locks to ensure atomic snapshot
        let instruments = self.instruments.read().await;
        let seq = *self.last_market_update_seq.read().await;

        // Store the captured seq for snapshot_offsets() to use
        {
            let mut snapshot_seq = self.last_snapshot_seq.write().await;
            *snapshot_seq = seq;
        }

        let mut states = HashMap::with_capacity(instruments.len());
        for (symbol, instrument) in instruments.iter() {
            states.insert(symbol.clone(), InstrumentSnapshotState::from(instrument));
        }

        Ok(states)
    }

    async fn restore(&self, key: &Self::Key, state: Self::State) -> Result<(), SnapshotError> {
        let instrument: Instrument = state.into_instrument();

        // Verify key matches
        if &instrument.id != key {
            return Err(SnapshotError::Serialization(format!(
                "Key mismatch: expected {}, got {}",
                key, instrument.id
            )));
        }

        self.add_instrument_direct(instrument)
            .await
            .map_err(|e| SnapshotError::DbError(e.to_string()))
    }

    async fn clear_all(&self) -> Result<(), SnapshotError> {
        self.instruments.write().await.clear();
        self.by_underlying.write().await.clear();
        self.by_underlying_expiry.write().await.clear();
        self.by_instrument_id.write().await.clear();
        *self.last_market_update_seq.write().await = 0;
        self.has_received_update.store(false, Ordering::SeqCst);
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn reload_from_db_keeps_existing_cache_when_db_load_fails() {
        let cache = InstrumentsCache::new();

        cache
            .add_instrument_direct(Instrument {
                instrument_id: 1,
                id: "BTC-20270101-100000-C".to_string(),
                underlying: "BTC".to_string(),
                strike: dec!(100000),
                expiry: 1798761600,
                option_type: "call".to_string(),
                option_token_address: None,
                mark_iv: None,
                volume_24h: dec!(0),
                open_interest: dec!(0),
                updated_at: Utc::now(),
                status: InstrumentStatus::Active,
                trading_mode: TradingModes::ORDERBOOK,
            })
            .await
            .expect("seed instrument should be inserted");

        let before = cache.len().await;
        assert_eq!(before, 1, "expected seeded cache size");

        // Invalid endpoint forces DB load failure. On failure, cache must remain
        // unchanged instead of being cleared.
        let bad_db = hypercall_db_diesel::DieselDb::new_no_tls(
            "postgresql://invalid:invalid@127.0.0.1:1/invalid",
            1,
        )
        .await
        .expect("lazy pool should be constructed");

        let reload_result = cache.reload_from_db(&bad_db).await;
        assert!(
            reload_result.is_err(),
            "reload should fail against an invalid DB endpoint"
        );

        let after = cache.len().await;
        assert_eq!(
            after, before,
            "failed reload must not clear previously cached instruments"
        );
    }
}