hypercall/liquidator/

cache.rs

//! In-memory cache for liquidation states.
//!
//! Provides fast synchronous access to liquidation state for order blocking
//! in the unified engine, with async methods for updates.

use super::state::{
    state_str, AccountLiquidationStatus, FullLiquidationMetadata, LiquidatedMetadata,
    LiquidationState, PartialLiquidationMetadata,
};
use crate::rsm::MarginMode;
use hypercall_db::LiquidationReader;
use hypercall_types::WalletAddress;
use rust_decimal::Decimal;
use serde_json::Value as JsonValue;
use std::collections::HashMap;
use std::sync::{Arc, RwLock};
use tracing::{debug, info, warn};

/// Default liquidation state for accounts not in the cache.
const DEFAULT_STATE: LiquidationState = LiquidationState::Healthy;

/// In-memory cache for liquidation states.
///
/// Provides fast synchronous reads for the hot path (order validation)
/// and async writes for state transitions.
///
/// Uses `std::sync::RwLock` to allow blocking reads in synchronous contexts.
pub struct LiquidationCache {
    /// Cached liquidation states by wallet.
    states: Arc<RwLock<HashMap<WalletAddress, AccountLiquidationStatus>>>,
}

impl LiquidationCache {
    /// Create a new empty cache.
    pub fn new() -> Self {
        Self {
            states: Arc::new(RwLock::new(HashMap::new())),
        }
    }

    // ========== Synchronous Methods (for unified engine hot path) ==========

    /// Get liquidation state synchronously (blocking).
    ///
    /// Blocks until the lock is available, then returns the state.
    /// Returns Healthy if wallet not found in cache.
    /// This is the hot path for order validation.
    pub fn get_state_sync(&self, wallet: &WalletAddress) -> LiquidationState {
        let cache = self.states.read().expect("RwLock poisoned");
        cache
            .get(wallet)
            .map(|s| s.state.clone())
            .unwrap_or(DEFAULT_STATE)
    }

    /// Check if wallet is in pre-liquidation (synchronous).
    pub fn is_pre_liquidation_sync(&self, wallet: &WalletAddress) -> bool {
        self.get_state_sync(wallet).is_pre_liquidation()
    }

    /// Check if risk-increasing orders should be blocked (synchronous).
    pub fn should_block_risk_increasing_sync(&self, wallet: &WalletAddress) -> bool {
        self.get_state_sync(wallet).should_block_risk_increasing()
    }

    // ========== Async Methods (for watcher and API) ==========
    //
    // Note: These methods are async for API compatibility but use std::sync::RwLock
    // internally, which blocks synchronously. This is acceptable because the lock
    // hold time is very short (just HashMap operations).

    /// Get full liquidation status.
    pub async fn get_status(&self, wallet: &WalletAddress) -> Option<AccountLiquidationStatus> {
        let cache = self.states.read().expect("RwLock poisoned");
        cache.get(wallet).cloned()
    }

    /// Get liquidation state.
    pub async fn get_state(&self, wallet: &WalletAddress) -> LiquidationState {
        let cache = self.states.read().expect("RwLock poisoned");
        cache
            .get(wallet)
            .map(|s| s.state.clone())
            .unwrap_or(DEFAULT_STATE)
    }

    /// Get all wallets currently being tracked.
    pub async fn get_all_wallets(&self) -> Vec<WalletAddress> {
        let cache = self.states.read().expect("RwLock poisoned");
        cache.keys().cloned().collect()
    }

    /// Get all wallets in a specific state.
    pub async fn get_wallets_in_state(&self, state_type: &str) -> Vec<WalletAddress> {
        let cache = self.states.read().expect("RwLock poisoned");
        cache
            .iter()
            .filter(|(_, status)| status.state.as_str() == state_type)
            .map(|(wallet, _)| *wallet)
            .collect()
    }

    /// Get count of wallets in each state.
    pub async fn get_state_counts(&self) -> HashMap<String, usize> {
        let cache = self.states.read().expect("RwLock poisoned");
        let mut counts: HashMap<String, usize> = HashMap::new();

        for status in cache.values() {
            *counts.entry(status.state.as_str().to_string()).or_insert(0) += 1;
        }

        counts
    }

    /// Update or insert liquidation status.
    pub async fn set_status(&self, status: AccountLiquidationStatus) {
        let wallet = status.wallet;
        let state_str = status.state.as_str();

        let mut cache = self.states.write().expect("RwLock poisoned");
        cache.insert(wallet, status);

        debug!("Updated liquidation status for {}: {}", wallet, state_str);
    }

    /// Update health metrics for a wallet.
    pub async fn update_health(
        &self,
        wallet: &WalletAddress,
        equity: Decimal,
        mm_required: Decimal,
        timestamp: u64,
    ) {
        let mut cache = self.states.write().expect("RwLock poisoned");
        if let Some(status) = cache.get_mut(wallet) {
            status.update_health(equity, mm_required, timestamp);
        }
    }

    /// Transition wallet to pre-liquidation state.
    ///
    /// Returns the previous state for logging/events.
    pub async fn enter_pre_liquidation(
        &self,
        wallet: &WalletAddress,
        metadata: PartialLiquidationMetadata,
        timestamp: u64,
    ) -> Option<LiquidationState> {
        let mut cache = self.states.write().expect("RwLock poisoned");
        if let Some(status) = cache.get_mut(wallet) {
            let previous = status.state.clone();
            status.enter_pre_liquidation(metadata, timestamp);
            info!(
                "Wallet {} entered pre-liquidation (shortfall: {})",
                wallet,
                status.shortfall()
            );
            return Some(previous);
        }
        None
    }

    /// Transition wallet back to healthy state.
    ///
    /// Returns the previous state for logging/events.
    pub async fn recover_to_healthy(
        &self,
        wallet: &WalletAddress,
        timestamp: u64,
    ) -> Option<LiquidationState> {
        let mut cache = self.states.write().expect("RwLock poisoned");
        if let Some(status) = cache.get_mut(wallet) {
            let previous = status.state.clone();
            if previous.is_pre_liquidation() {
                status.recover_to_healthy(timestamp);
                info!("Wallet {} recovered to healthy state", wallet);
                return Some(previous);
            }
        }
        None
    }

    /// Transition wallet to in-liquidation state.
    ///
    /// Returns the previous state for logging/events.
    pub async fn enter_liquidation(
        &self,
        wallet: &WalletAddress,
        metadata: FullLiquidationMetadata,
        timestamp: u64,
    ) -> Option<LiquidationState> {
        let mut cache = self.states.write().expect("RwLock poisoned");
        if let Some(status) = cache.get_mut(wallet) {
            let previous = status.state.clone();
            let auction_id = metadata.auction_id.clone();
            status.enter_liquidation(metadata, timestamp);
            info!(
                "Wallet {} entered liquidation, auction_id={}",
                wallet, auction_id
            );
            return Some(previous);
        }
        None
    }

    /// Transition wallet to liquidated state.
    ///
    /// Returns the previous state for logging/events.
    pub async fn complete_liquidation(
        &self,
        wallet: &WalletAddress,
        metadata: LiquidatedMetadata,
        timestamp: u64,
    ) -> Option<LiquidationState> {
        let mut cache = self.states.write().expect("RwLock poisoned");
        if let Some(status) = cache.get_mut(wallet) {
            let previous = status.state.clone();
            status.complete_liquidation(metadata, timestamp);
            info!("Wallet {} liquidation completed", wallet);
            return Some(previous);
        }
        None
    }

    /// Remove a wallet from tracking (e.g., after liquidation completed).
    pub async fn remove(&self, wallet: &WalletAddress) {
        let mut cache = self.states.write().expect("RwLock poisoned");
        cache.remove(wallet);
        debug!("Removed wallet {} from liquidation cache", wallet);
    }

    pub async fn settle_and_complete_liquidation(
        &self,
        wallet: &WalletAddress,
        auction_id: &str,
        liquidator: &WalletAddress,
        settlement_value: Decimal,
        timestamp: u64,
    ) -> Option<LiquidationState> {
        info!(
            target: "liquidation::onchain",
            wallet = %wallet,
            auction_id = %auction_id,
            liquidator = %liquidator,
            settlement_value = %settlement_value,
            "ONCHAIN_CALL: LiquidationContract.settleAuction(auction_id={}, liquidator={}, settlement_value={})",
            auction_id, liquidator, settlement_value
        );

        info!(
            target: "liquidation::onchain",
            wallet = %wallet,
            auction_id = %auction_id,
            "ONCHAIN_CALL: RootCommitter.commitMarginRoot() - updating merkle root post-liquidation"
        );

        let Some(cached_auction_id) = self
            .get_state(wallet)
            .await
            .auction_id()
            .map(str::to_string)
        else {
            return None;
        };
        if cached_auction_id != auction_id {
            warn!(
                wallet = %wallet,
                requested_auction_id = %auction_id,
                cached_auction_id = %cached_auction_id,
                "Refusing to complete liquidation with mismatched auction ids"
            );
            return None;
        }

        self.complete_liquidation(
            wallet,
            LiquidatedMetadata {
                auction_id: cached_auction_id,
                completed_at: timestamp,
                winner: Some(*liquidator),
                bonus: Decimal::ZERO,
                tx_hash: None,
            },
            timestamp,
        )
        .await
    }

    /// Initialize status for a wallet if not already present.
    pub async fn init_if_absent(
        &self,
        wallet: WalletAddress,
        margin_mode: MarginMode,
        equity: Decimal,
        mm_required: Decimal,
        timestamp: u64,
    ) {
        let mut cache = self.states.write().expect("RwLock poisoned");
        cache.entry(wallet).or_insert_with(|| {
            AccountLiquidationStatus::healthy(wallet, margin_mode, equity, mm_required, timestamp)
        });
    }

    /// Clear all entries (for testing).
    #[cfg(test)]
    pub async fn clear(&self) {
        let mut cache = self.states.write().expect("RwLock poisoned");
        cache.clear();
    }

    // ========== Database Persistence Methods ==========

    /// Load all liquidation states from the database into the cache.
    ///
    /// This should be called during startup to restore state.
    pub async fn load_from_db(
        &self,
        handler: &hypercall_db_diesel::DieselDb,
    ) -> anyhow::Result<usize> {
        let records = handler.get_all_liquidation_states().await?;
        let count = records.len();

        let mut cache = self.states.write().expect("RwLock poisoned");
        for record in records {
            let wallet = record.wallet_address;
            let status = Self::record_to_status(record).map_err(|error| {
                anyhow::anyhow!(
                    "failed to parse persisted liquidation state for {}: {}",
                    wallet,
                    error
                )
            })?;
            debug!(
                "Loaded liquidation state for {}: {}",
                status.wallet,
                status.state.as_str()
            );
            cache.insert(status.wallet, status);
        }

        info!("Loaded {} liquidation states from database", count);
        Ok(count)
    }

    /// Convert a database record to an AccountLiquidationStatus.
    pub(crate) fn record_to_status(
        record: hypercall_db::LiquidationStateRecord,
    ) -> anyhow::Result<AccountLiquidationStatus> {
        let updated_at = persisted_updated_at_ms(&record)?;
        let margin_mode: MarginMode = record
            .margin_mode
            .parse()
            .map_err(|_| anyhow::anyhow!("Unknown margin mode: {}", record.margin_mode))?;

        let state = match record.state.as_str() {
            state_str::HEALTHY => LiquidationState::Healthy,
            state_str::PRE_LIQUIDATION => {
                let mm_shortfall = record
                    .mm_shortfall
                    .unwrap_or_else(|| (record.mm_required - record.equity).max(Decimal::ZERO));
                LiquidationState::PreLiquidation(PartialLiquidationMetadata {
                    entered_at: required_i64(record.entered_pre_liq_at, "entered_pre_liq_at")?,
                    mm_shortfall,
                    target_equity: required_decimal(record.target_equity, "target_equity")?,
                    escalation_deadline: required_i64(
                        record.escalation_deadline,
                        "escalation_deadline",
                    )?,
                    last_reprice_at: optional_i64(record.last_reprice_at),
                    active_order_request_ids: match record.partial_order_request_ids.as_ref() {
                        Some(_) => json_string_array(
                            record.partial_order_request_ids.as_ref(),
                            "partial_order_request_ids",
                        )?,
                        None => Vec::new(),
                    },
                    active_order_client_ids: match record.partial_order_client_ids.as_ref() {
                        Some(_) => json_string_array(
                            record.partial_order_client_ids.as_ref(),
                            "partial_order_client_ids",
                        )?,
                        None => Vec::new(),
                    },
                    bonus_bps: {
                        let raw = required_i32(record.partial_bonus_bps, "partial_bonus_bps")?;
                        if raw < 0 {
                            warn!("partial_bonus_bps is negative ({}), clamping to 0", raw);
                            0u32
                        } else {
                            raw as u32
                        }
                    },
                    pending_full_auction_id: record.auction_id,
                    pending_full_request_id: record.request_id,
                    pending_full_tx_hash: record.tx_hash,
                    pending_full_margin_needed: record.margin_needed,
                })
            }
            state_str::IN_LIQUIDATION => LiquidationState::InLiquidation(FullLiquidationMetadata {
                auction_id: required_string(record.auction_id, "auction_id")?,
                request_id: record.request_id,
                tx_hash: record.tx_hash,
                started_at: required_i64(record.auction_started_at, "auction_started_at")?,
                chain_start_time: optional_i64(record.chain_start_time),
                margin_needed: required_decimal(record.margin_needed, "margin_needed")?,
                stop_request_id: record.stop_request_id,
                stop_tx_hash: record.stop_tx_hash,
            }),
            state_str::LIQUIDATED => LiquidationState::Liquidated(LiquidatedMetadata {
                auction_id: required_string(record.auction_id, "auction_id")?,
                completed_at: required_i64(record.liquidated_at, "liquidated_at")?,
                winner: record.resolved_winner,
                bonus: required_decimal(record.resolved_bonus, "resolved_bonus")?,
                tx_hash: record.resolution_tx_hash,
            }),
            _ => return Err(anyhow::anyhow!("Unknown state: {}", record.state)),
        };

        Ok(AccountLiquidationStatus {
            wallet: record.wallet_address,
            state,
            margin_mode,
            equity: record.equity,
            mm_required: record.mm_required,
            maintenance_margin: record.maintenance_margin,
            updated_at,
        })
    }
}

fn required_decimal(value: Option<Decimal>, field: &str) -> anyhow::Result<Decimal> {
    value.ok_or_else(|| anyhow::anyhow!("missing required liquidation field '{}'", field))
}

fn required_i64(value: Option<i64>, field: &str) -> anyhow::Result<u64> {
    let raw =
        value.ok_or_else(|| anyhow::anyhow!("missing required liquidation field '{}'", field))?;
    u64::try_from(raw)
        .map_err(|_| anyhow::anyhow!("negative value in liquidation field '{}': {}", field, raw))
}

fn required_i32(value: Option<i32>, field: &str) -> anyhow::Result<i32> {
    value.ok_or_else(|| anyhow::anyhow!("missing required liquidation field '{}'", field))
}

fn optional_i64(value: Option<i64>) -> Option<u64> {
    value.map(|raw| {
        u64::try_from(raw).unwrap_or_else(|_| {
            panic!(
                "STATE_CORRUPTION: negative persisted liquidation timestamp/block value {}",
                raw
            )
        })
    })
}

fn persisted_updated_at_ms(record: &hypercall_db::LiquidationStateRecord) -> anyhow::Result<u64> {
    if let Some(updated_at_ms) = optional_i64(record.updated_at_ms) {
        return Ok(updated_at_ms);
    }

    if let Some(updated_at) = record.updated_at {
        return Ok(updated_at.and_utc().timestamp_millis() as u64);
    }

    if let Some(created_at) = record.created_at {
        return Ok(created_at.and_utc().timestamp_millis() as u64);
    }

    anyhow::bail!(
        "missing liquidation updated_at_ms and database timestamps for wallet {}",
        record.wallet_address
    )
}

fn required_string(value: Option<String>, field: &str) -> anyhow::Result<String> {
    let value =
        value.ok_or_else(|| anyhow::anyhow!("missing required liquidation field '{}'", field))?;
    if value.trim().is_empty() {
        anyhow::bail!("liquidation field '{}' must not be empty", field);
    }
    Ok(value)
}

fn json_string_array(value: Option<&JsonValue>, field: &str) -> anyhow::Result<Vec<String>> {
    let Some(value) = value else {
        return Ok(Vec::new());
    };
    let array = value.as_array().ok_or_else(|| {
        anyhow::anyhow!(
            "liquidation field '{}' must be a JSON array of strings",
            field
        )
    })?;
    array
        .iter()
        .map(|entry| {
            entry.as_str().map(ToOwned::to_owned).ok_or_else(|| {
                anyhow::anyhow!(
                    "liquidation field '{}' must contain only string values",
                    field
                )
            })
        })
        .collect()
}

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

#[cfg(test)]
mod tests {
    use super::*;
    use chrono::Utc;
    use rust_decimal_macros::dec;

    fn test_wallet() -> WalletAddress {
        "0x1234567890123456789012345678901234567890"
            .parse()
            .unwrap()
    }

    fn test_wallet_2() -> WalletAddress {
        "0xabcdefabcdefabcdefabcdefabcdefabcdefabcd"
            .parse()
            .unwrap()
    }

    fn partial_metadata(shortfall: Decimal) -> PartialLiquidationMetadata {
        PartialLiquidationMetadata {
            entered_at: 1_500,
            mm_shortfall: shortfall,
            target_equity: dec!(5_500),
            escalation_deadline: 61_500,
            last_reprice_at: Some(1_750),
            active_order_request_ids: vec!["req-1".to_string()],
            active_order_client_ids: vec!["liq-1".to_string()],
            bonus_bps: 100,
            pending_full_auction_id: None,
            pending_full_request_id: None,
            pending_full_tx_hash: None,
            pending_full_margin_needed: None,
        }
    }

    #[tokio::test]
    async fn test_empty_cache_returns_healthy() {
        let cache = LiquidationCache::new();
        let wallet = test_wallet();

        // Async method
        assert!(cache.get_state(&wallet).await.is_healthy());

        // Sync method
        assert!(cache.get_state_sync(&wallet).is_healthy());
        assert!(!cache.should_block_risk_increasing_sync(&wallet));
    }

    #[tokio::test]
    async fn test_set_and_get_status() {
        let cache = LiquidationCache::new();
        let wallet = test_wallet();

        let status = AccountLiquidationStatus::healthy(
            wallet,
            MarginMode::Standard,
            dec!(10000),
            dec!(5000),
            1000,
        );

        cache.set_status(status).await;

        let retrieved = cache.get_status(&wallet).await.unwrap();
        assert!(retrieved.state.is_healthy());
        assert_eq!(retrieved.equity, dec!(10000));
    }

    #[tokio::test]
    async fn test_pre_liquidation_transition() {
        let cache = LiquidationCache::new();
        let wallet = test_wallet();

        // Initialize with underwater account
        let status = AccountLiquidationStatus::healthy(
            wallet,
            MarginMode::Standard,
            dec!(4000),
            dec!(5000),
            1000,
        );
        cache.set_status(status).await;

        // Transition to pre-liquidation
        let previous = cache
            .enter_pre_liquidation(&wallet, partial_metadata(dec!(1000)), 2000)
            .await;
        assert!(previous.is_some());
        assert!(previous.unwrap().is_healthy());

        // Verify state
        assert!(cache.is_pre_liquidation_sync(&wallet));
        assert!(cache.should_block_risk_increasing_sync(&wallet));
    }

    #[tokio::test]
    async fn test_recovery_transition() {
        let cache = LiquidationCache::new();
        let wallet = test_wallet();

        // Initialize and enter pre-liquidation
        let mut status = AccountLiquidationStatus::healthy(
            wallet,
            MarginMode::Standard,
            dec!(4000),
            dec!(5000),
            1000,
        );
        status.enter_pre_liquidation(partial_metadata(dec!(1000)), 1500);
        cache.set_status(status).await;

        assert!(cache.is_pre_liquidation_sync(&wallet));

        // Recover
        let previous = cache.recover_to_healthy(&wallet, 2000).await;
        assert!(previous.is_some());
        assert!(previous.unwrap().is_pre_liquidation());

        // Verify healthy
        assert!(!cache.is_pre_liquidation_sync(&wallet));
        assert!(!cache.should_block_risk_increasing_sync(&wallet));
    }

    #[tokio::test]
    async fn test_get_wallets_in_state() {
        let cache = LiquidationCache::new();
        let wallet1 = test_wallet();
        let wallet2 = test_wallet_2();

        // Add one healthy, one pre-liquidation
        let status1 = AccountLiquidationStatus::healthy(
            wallet1,
            MarginMode::Standard,
            dec!(10000),
            dec!(5000),
            1000,
        );
        cache.set_status(status1).await;

        let mut status2 = AccountLiquidationStatus::healthy(
            wallet2,
            MarginMode::Portfolio,
            dec!(4000),
            dec!(5000),
            1000,
        );
        status2.enter_pre_liquidation(partial_metadata(dec!(1000)), 1500);
        cache.set_status(status2).await;

        // Get wallets by state
        let healthy = cache.get_wallets_in_state("Healthy").await;
        assert_eq!(healthy.len(), 1);
        assert_eq!(healthy[0], wallet1);

        let pre_liq = cache.get_wallets_in_state("PreLiquidation").await;
        assert_eq!(pre_liq.len(), 1);
        assert_eq!(pre_liq[0], wallet2);
    }

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

        // Add various states
        for i in 0..5 {
            let wallet: WalletAddress = format!("0x{:040x}", i).parse().unwrap();
            let status = AccountLiquidationStatus::healthy(
                wallet,
                MarginMode::Standard,
                dec!(10000),
                dec!(5000),
                1000,
            );
            cache.set_status(status).await;
        }

        // Put 2 in pre-liquidation
        for i in 5..7 {
            let wallet: WalletAddress = format!("0x{:040x}", i).parse().unwrap();
            let mut status = AccountLiquidationStatus::healthy(
                wallet,
                MarginMode::Standard,
                dec!(4000),
                dec!(5000),
                1000,
            );
            status.enter_pre_liquidation(partial_metadata(dec!(1000)), 1500);
            cache.set_status(status).await;
        }

        let counts = cache.get_state_counts().await;
        assert_eq!(counts.get("Healthy"), Some(&5));
        assert_eq!(counts.get("PreLiquidation"), Some(&2));
    }

    #[tokio::test]
    async fn test_init_if_absent() {
        let cache = LiquidationCache::new();
        let wallet = test_wallet();

        // First init creates entry
        cache
            .init_if_absent(wallet, MarginMode::Standard, dec!(10000), dec!(5000), 1000)
            .await;

        let status = cache.get_status(&wallet).await.unwrap();
        assert_eq!(status.equity, dec!(10000));

        // Second init doesn't overwrite
        cache
            .init_if_absent(wallet, MarginMode::Standard, dec!(20000), dec!(5000), 2000)
            .await;

        let status = cache.get_status(&wallet).await.unwrap();
        assert_eq!(status.equity, dec!(10000)); // Still original value
    }

    #[test]
    fn test_record_to_status_requires_partial_bonus_bps() {
        let now = Utc::now().naive_utc();
        let error = LiquidationCache::record_to_status(hypercall_db::LiquidationStateRecord {
            wallet_address: test_wallet(),
            state: state_str::PRE_LIQUIDATION.to_string(),
            margin_mode: MarginMode::Standard.as_str().to_string(),
            equity: dec!(4000),
            mm_required: dec!(5000),
            maintenance_margin: dec!(-1000),
            liquidation_mode: Some("partial".to_string()),
            target_equity: Some(dec!(6000)),
            entered_pre_liq_at: Some(2_000),
            mm_shortfall: Some(dec!(1000)),
            escalation_deadline: Some(62_000),
            last_reprice_at: Some(3_000),
            partial_order_request_ids: Some(JsonValue::Array(vec![JsonValue::String(
                "req-1".to_string(),
            )])),
            partial_order_client_ids: Some(JsonValue::Array(vec![JsonValue::String(
                "liq-1".to_string(),
            )])),
            partial_bonus_bps: None,
            auction_id: None,
            request_id: None,
            tx_hash: None,
            auction_started_at: None,
            chain_start_time: None,
            margin_needed: None,
            stop_request_id: None,
            stop_tx_hash: None,
            liquidated_at: None,
            resolved_winner: None,
            resolved_bonus: None,
            resolution_tx_hash: None,
            last_observed_block: None,
            updated_at_ms: Some(3_500),
            created_at: Some(now),
            updated_at: Some(now),
        })
        .unwrap_err();

        assert!(error
            .to_string()
            .contains("missing required liquidation field 'partial_bonus_bps'"));
    }
}