hypercall_admin/monitoring/

integrity.rs

//! System integrity admin endpoints (margin engine + orderbook).

use axum::{extract::State, http::StatusCode, response::IntoResponse};
use rust_decimal::Decimal;
use rust_decimal_macros::dec;
use serde::Serialize;
use sonic_rs::json;
use std::collections::HashMap;
use tracing::warn;

use crate::state::AdminState;
use hypercall_runtime_api::sonic_json::SonicJson;
use hypercall_types::perp_underlying;

/// Detailed integrity report for the margin engine.
#[derive(Debug, Serialize, utoipa::ToSchema)]
pub struct IntegrityReport {
    /// ISO 8601 timestamp of when the report was generated
    pub timestamp: String,
    /// Overall system health status
    pub overall_status: IntegrityStatus,
    /// Individual integrity checks performed
    pub checks: Vec<IntegrityCheck>,
    /// Summary of ledger balances
    pub ledger_summary: LedgerSummary,
    /// Summary of open positions
    pub position_summary: PositionSummary,
    /// Summary of settlements
    pub settlement_summary: SettlementSummary,
}

/// System health status level
#[derive(Debug, Clone, Copy, Serialize, PartialEq, utoipa::ToSchema)]
#[serde(rename_all = "UPPERCASE")]
pub enum IntegrityStatus {
    /// All systems operating normally
    Healthy,
    /// Minor issues detected, monitoring recommended
    Warning,
    /// Significant issues detected, investigation required
    Critical,
}

/// Individual integrity check result
#[derive(Debug, Serialize, utoipa::ToSchema)]
pub struct IntegrityCheck {
    /// Name of the check performed
    pub name: String,
    /// Status result of the check
    pub status: IntegrityStatus,
    /// Human-readable description of the result
    pub message: String,
    /// Additional details (varies by check type)
    #[schema(value_type = Option<Object>)]
    pub details: Option<sonic_rs::Value>,
}

/// Summary of ledger balances across all accounts
#[derive(Debug, Serialize, utoipa::ToSchema)]
pub struct LedgerSummary {
    /// Total DB-projected balance across all accounts (USD), unavailable when engine-owned.
    pub total_balance: Option<String>,
    /// Current-balance authority for this report.
    pub authority_state: String,
    /// Number of accounts with balances
    pub account_count: i64,
    /// Total number of fills processed
    pub total_fills: i64,
    /// Total trading volume (USD)
    pub total_volume: String,
}

/// Summary of open positions across all accounts
#[derive(Debug, Serialize, utoipa::ToSchema)]
pub struct PositionSummary {
    /// Total open interest across all instruments
    pub total_open_interest: String,
    /// Number of unique instruments with positions
    pub unique_instruments: i64,
    /// List of symbols with position imbalances (should be empty)
    pub net_position_imbalances: Vec<PositionImbalance>,
}

/// Position imbalance details for a symbol
#[derive(Debug, Clone, Serialize, PartialEq, utoipa::ToSchema)]
pub struct PositionImbalance {
    /// Option symbol
    pub symbol: String,
    /// Net position (should be zero for balanced book)
    pub net_position: String,
    /// Severity of the imbalance
    pub severity: IntegrityStatus,
}

#[derive(Debug, Clone, Serialize, PartialEq)]
struct PerpExposure {
    symbol: String,
    net_position: String,
}

#[derive(Debug, Clone, PartialEq)]
struct PositionBalanceEvaluation {
    imbalances: Vec<PositionImbalance>,
    excluded_perp_exposures: Vec<PerpExposure>,
    total_imbalance: Decimal,
    status: IntegrityStatus,
}

fn position_balance_severity(abs_net: Decimal) -> IntegrityStatus {
    if abs_net > dec!(0.01) {
        IntegrityStatus::Critical
    } else if abs_net > dec!(0.0001) {
        IntegrityStatus::Warning
    } else {
        IntegrityStatus::Healthy
    }
}

fn evaluate_position_balance(
    net_positions_by_symbol: &HashMap<String, Decimal>,
) -> PositionBalanceEvaluation {
    let mut imbalances = Vec::new();
    let mut excluded_perp_exposures = Vec::new();
    let mut total_imbalance = dec!(0);

    for (symbol, net) in net_positions_by_symbol {
        if perp_underlying(symbol).is_some() {
            if *net != Decimal::ZERO {
                excluded_perp_exposures.push(PerpExposure {
                    symbol: symbol.clone(),
                    net_position: net.to_string(),
                });
            }
            continue;
        }

        let abs_net = net.abs();
        total_imbalance += abs_net;
        let severity = position_balance_severity(abs_net);

        if severity != IntegrityStatus::Healthy {
            imbalances.push(PositionImbalance {
                symbol: symbol.clone(),
                net_position: net.to_string(),
                severity,
            });
        }
    }

    imbalances.sort_by(|a, b| a.symbol.cmp(&b.symbol));
    excluded_perp_exposures.sort_by(|a, b| a.symbol.cmp(&b.symbol));

    PositionBalanceEvaluation {
        imbalances,
        excluded_perp_exposures,
        total_imbalance,
        status: position_balance_severity(total_imbalance),
    }
}

/// Summary of settlement processing
#[derive(Debug, Serialize, utoipa::ToSchema)]
pub struct SettlementSummary {
    /// Total number of settlements
    pub total: i64,
    /// Number of settlements applied
    pub applied: i64,
    /// Number of settlements pending
    pub pending: i64,
    /// Total value of settlements (USD)
    pub total_value: String,
}

/// GET /monitoring/integrity - Comprehensive integrity check
///
/// Returns detailed integrity report including:
/// - Position balance verification (longs should equal shorts)
/// - Ledger totals
/// - Settlement status
/// - Any detected anomalies
///
/// **Authentication**: Requires `X-Admin-Key` header (performance protection).
/// These endpoints execute synchronous database queries.
#[utoipa::path(
    get,
    path = "/monitoring/integrity",
    responses(
        (status = 200, description = "Integrity report", body = IntegrityReport),
        (status = 401, description = "Invalid or missing X-Admin-Key header")
    ),
    tag = "Monitoring",
    security(("admin_key" = []))
)]
pub async fn integrity_check(State(app_state): State<AdminState>) -> impl IntoResponse {
    let timestamp = chrono::Utc::now().to_rfc3339();
    let mut checks = Vec::new();
    let mut overall_status = IntegrityStatus::Healthy;

    // Get sync DB handler (early-return if not configured; reads go through async db)
    let _db_handler = match app_state.sync_db.as_ref() {
        Some(h) => h.clone(),
        None => {
            return (
                StatusCode::SERVICE_UNAVAILABLE,
                SonicJson(json!({
                    "error": "Database handler not available"
                })),
            )
                .into_response();
        }
    };

    // Get position data from portfolio cache (not DB - positions table doesn't exist)
    let portfolios = app_state.portfolio_cache.get_all_portfolios().await;

    // Calculate net positions by symbol from cache
    let mut net_positions_by_symbol: std::collections::HashMap<String, Decimal> =
        std::collections::HashMap::new();
    let mut oi_by_underlying: std::collections::HashMap<String, Decimal> =
        std::collections::HashMap::new();

    for summary in portfolios.values() {
        for (symbol, pos) in &summary.positions {
            // Net position for zero-sum check
            *net_positions_by_symbol.entry(symbol.clone()).or_default() += pos.amount;

            // Open interest by underlying (sum of absolute positions)
            if let Some(underlying) = symbol.split('-').next() {
                *oi_by_underlying.entry(underlying.to_string()).or_default() += pos.amount.abs();
            }
        }
    }

    // Run integrity checks via async persistence pool.
    let integrity: &dyn hypercall_db::IntegrityReader = app_state.db.as_ref();
    let db_results = integrity.get_integrity_query_results().await;

    let engine_digest = app_state.engine_state_digest_provider.engine_state_digest();
    let fill_volume_result = db_results.fill_volume;
    let settlement_result = db_results.settlement_stats;

    // ===== Check 0: Database Connectivity =====
    // Track DB query failures - don't silently zero them out (fail loudly principle)
    let mut db_errors: Vec<String> = Vec::new();

    if let Err(e) = &fill_volume_result {
        db_errors.push(format!("fill_volume: {}", e));
    }
    if let Err(e) = &settlement_result {
        db_errors.push(format!("settlement_stats: {}", e));
    }

    if !db_errors.is_empty() {
        overall_status = IntegrityStatus::Critical;
        checks.push(IntegrityCheck {
            name: "Database Connectivity".to_string(),
            status: IntegrityStatus::Critical,
            message: format!("{} database queries failed", db_errors.len()),
            details: Some(json!({ "failed_queries": db_errors })),
        });
        warn!("INTEGRITY CHECK: Database queries failed: {:?}", db_errors);
    }

    // ===== Check 1: Position Balance (Zero-Sum) =====
    //
    // HyperCore perp positions are external venue exposure imported into the
    // portfolio cache for margin. They are not expected to net to zero inside
    // Hypercall, so the matching-engine zero-sum invariant only applies to
    // option symbols.
    let position_balance = evaluate_position_balance(&net_positions_by_symbol);
    let position_check_status = position_balance.status;

    if position_check_status == IntegrityStatus::Critical {
        overall_status = IntegrityStatus::Critical;
    } else if position_check_status == IntegrityStatus::Warning
        && overall_status != IntegrityStatus::Critical
    {
        overall_status = IntegrityStatus::Warning;
    }

    checks.push(IntegrityCheck {
        name: "Position Balance (Zero-Sum)".to_string(),
        status: position_check_status,
        message: if position_balance.imbalances.is_empty() {
            if position_balance.excluded_perp_exposures.is_empty() {
                "All option positions are balanced (sum of longs equals sum of shorts)".to_string()
            } else {
                format!(
                    "All option positions are balanced; {} external perp exposure symbol(s) excluded from zero-sum check",
                    position_balance.excluded_perp_exposures.len()
                )
            }
        } else {
            format!(
                "{} option symbol(s) have position imbalance totaling {}",
                position_balance.imbalances.len(),
                position_balance.total_imbalance
            )
        },
        details: if position_balance.imbalances.is_empty()
            && position_balance.excluded_perp_exposures.is_empty()
        {
            None
        } else if position_balance.excluded_perp_exposures.is_empty() {
            Some(sonic_rs::to_value(&position_balance.imbalances).unwrap_or_default())
        } else if position_balance.imbalances.is_empty() {
            Some(json!({
                "excluded_perp_exposures": position_balance.excluded_perp_exposures.clone(),
            }))
        } else {
            Some(json!({
                "imbalances": position_balance.imbalances.clone(),
                "excluded_perp_exposures": position_balance.excluded_perp_exposures.clone(),
            }))
        },
    });

    // ===== Check 2: Settlement Status =====
    let settlement_check_status = match &settlement_result {
        Ok((_, _, pending, _)) => {
            if *pending > 10 {
                IntegrityStatus::Critical
            } else if *pending > 0 {
                IntegrityStatus::Warning
            } else {
                IntegrityStatus::Healthy
            }
        }
        Err(_) => IntegrityStatus::Critical, // Already reported in DB check
    };

    let (settlement_total, settlement_applied, settlement_pending, settlement_value) =
        settlement_result
            .as_ref()
            .map(|(t, a, p, v)| (*t, *a, *p, *v))
            .unwrap_or((0, 0, 0, dec!(0)));

    if settlement_check_status == IntegrityStatus::Critical && settlement_result.is_ok() {
        overall_status = IntegrityStatus::Critical;
    } else if settlement_check_status == IntegrityStatus::Warning
        && overall_status != IntegrityStatus::Critical
    {
        overall_status = IntegrityStatus::Warning;
    }

    checks.push(IntegrityCheck {
        name: "Settlement Processing".to_string(),
        status: settlement_check_status,
        message: if settlement_result.is_err() {
            "Unable to check - database query failed".to_string()
        } else {
            format!(
                "{} total settlements, {} applied, {} pending",
                settlement_total, settlement_applied, settlement_pending
            )
        },
        details: None,
    });

    // ===== Check 4: Settlement Balance Authority =====
    checks.push(IntegrityCheck {
        name: "Settlement Balance Authority".to_string(),
        status: IntegrityStatus::Healthy,
        message:
            "Settlement cash authority is EngineSnapshot.balance_ledger; DB settlement rows are audit evidence"
                .to_string(),
        details: None,
    });

    // ===== Check 3: Engine Balance Ledger =====
    // Saturate rather than panic: this is a read-only monitoring endpoint, and a
    // wallet count above i64::MAX is not physically reachable, so a clamp keeps
    // the endpoint serving instead of crashing the server.
    let account_count = i64::try_from(engine_digest.cash_wallet_count).unwrap_or(i64::MAX);
    let (fill_count, fill_volume) = fill_volume_result
        .as_ref()
        .map(|(c, v)| (*c, *v))
        .unwrap_or((0, dec!(0)));

    checks.push(IntegrityCheck {
        name: "Engine Balance Ledger".to_string(),
        status: IntegrityStatus::Healthy,
        message: format!(
            "Engine balance ledger has {} nonzero wallet(s); DB current-balance projection is write-only/downstream",
            account_count
        ),
        details: Some(json!({
            "cash_digest": engine_digest.cash_digest
        })),
    });

    // ===== Check 5: WAL Unreplicated Tail Size =====
    {
        use hypercall_journal::checkpoint::{checkpoint_path_for, read_checkpoint};

        let wal_path = app_state.runtime_config.wal_path.clone();
        let checkpoint_path = checkpoint_path_for(&wal_path);

        // Use async metadata so a slow filesystem (NFS, overloaded disk) cannot
        // stall the Tokio runtime. read_checkpoint reads a tiny fixed-size header,
        // so it stays synchronous.
        let wal_metadata = tokio::fs::metadata(&wal_path).await;
        let wal_check = match (wal_metadata, read_checkpoint(&checkpoint_path)) {
            (Ok(meta), Ok(checkpoint)) => {
                let file_size = meta.len();
                let unreplicated = file_size.saturating_sub(checkpoint.wal_offset);
                let unreplicated_mb = unreplicated as f64 / (1024.0 * 1024.0);

                const WARNING_THRESHOLD: u64 = 100 * 1024 * 1024; // 100 MB
                const CRITICAL_THRESHOLD: u64 = 1024 * 1024 * 1024; // 1 GB

                let status = if unreplicated > CRITICAL_THRESHOLD {
                    IntegrityStatus::Critical
                } else if unreplicated > WARNING_THRESHOLD {
                    IntegrityStatus::Warning
                } else {
                    IntegrityStatus::Healthy
                };

                if status == IntegrityStatus::Critical {
                    overall_status = IntegrityStatus::Critical;
                } else if status == IntegrityStatus::Warning
                    && overall_status != IntegrityStatus::Critical
                {
                    overall_status = IntegrityStatus::Warning;
                }

                IntegrityCheck {
                    name: "WAL Unreplicated Tail".to_string(),
                    status,
                    message: format!(
                        "Unreplicated tail: {:.1} MB (file: {} bytes, checkpoint: {} bytes)",
                        unreplicated_mb, file_size, checkpoint.wal_offset
                    ),
                    details: Some(json!({
                        "wal_file_bytes": file_size,
                        "checkpoint_offset_bytes": checkpoint.wal_offset,
                        "unreplicated_bytes": unreplicated,
                        "unreplicated_mb": format!("{:.1}", unreplicated_mb),
                    })),
                }
            }
            (Err(e), _) if e.kind() == std::io::ErrorKind::NotFound => {
                // WAL file doesn't exist - journaling is disabled
                IntegrityCheck {
                    name: "WAL Unreplicated Tail".to_string(),
                    status: IntegrityStatus::Healthy,
                    message: "WAL file not present (journaling may be disabled)".to_string(),
                    details: None,
                }
            }
            (Err(e), _) => {
                // Permission error, broken mount, etc. - flag it
                if overall_status != IntegrityStatus::Critical {
                    overall_status = IntegrityStatus::Warning;
                }
                IntegrityCheck {
                    name: "WAL Unreplicated Tail".to_string(),
                    status: IntegrityStatus::Warning,
                    message: format!("Unable to stat WAL file: {}", e),
                    details: None,
                }
            }
            (Ok(_), Err(e)) => {
                if overall_status != IntegrityStatus::Critical {
                    overall_status = IntegrityStatus::Warning;
                }
                IntegrityCheck {
                    name: "WAL Unreplicated Tail".to_string(),
                    status: IntegrityStatus::Warning,
                    message: format!("Unable to read checkpoint: {}", e),
                    details: None,
                }
            }
        };
        checks.push(wal_check);
    }

    // ===== Calculate Open Interest =====
    let total_oi: Decimal = oi_by_underlying.values().copied().sum();

    let unique_instruments = net_positions_by_symbol.len() as i64;

    // Build response
    let report = IntegrityReport {
        timestamp,
        overall_status,
        checks,
        ledger_summary: LedgerSummary {
            total_balance: None,
            authority_state: "engine_owned_not_db_aggregated".to_string(),
            account_count,
            total_fills: fill_count,
            total_volume: format!("{:.2}", fill_volume),
        },
        position_summary: PositionSummary {
            total_open_interest: format!("{:.4}", total_oi),
            unique_instruments,
            net_position_imbalances: position_balance.imbalances,
        },
        settlement_summary: SettlementSummary {
            total: settlement_total,
            applied: settlement_applied,
            pending: settlement_pending,
            total_value: format!("{:.2}", settlement_value),
        },
    };

    (StatusCode::OK, SonicJson(report)).into_response()
}

/// Crossed orderbook details for debugging
#[derive(Debug, Serialize, utoipa::ToSchema)]
pub struct CrossedOrderbook {
    /// Symbol (e.g., "BTC-20260131-100000-C")
    pub symbol: String,
    /// Best bid price
    pub best_bid: f64,
    /// Best ask price
    pub best_ask: f64,
    /// Spread (ask - bid, negative if crossed)
    pub spread: f64,
    /// Age of last update in seconds
    pub age_seconds: i64,
    /// Last L2 update sequence number
    pub last_l2_seq: Option<i64>,
    /// Number of bid price levels
    pub bid_levels: usize,
    /// Number of ask price levels
    pub ask_levels: usize,
    /// Top 10 bid levels (price, size)
    pub top_bids: Vec<(f64, f64)>,
    /// Top 10 ask levels (price, size)
    pub top_asks: Vec<(f64, f64)>,
}

/// Orderbook integrity report
#[derive(Debug, Serialize, utoipa::ToSchema)]
pub struct OrderbookIntegrityReport {
    /// Timestamp of the report
    pub timestamp: String,
    /// Total number of orderbooks in cache
    pub total_orderbooks: usize,
    /// Number of crossed orderbooks (bid >= ask)
    pub crossed_count: usize,
    /// Number of stale orderbooks (>60s since last update)
    pub stale_count: usize,
    /// Details of each crossed orderbook
    pub crossed_orderbooks: Vec<CrossedOrderbook>,
}

/// GET /monitoring/orderbooks - List all orderbooks with crossed state detection
///
/// **Authentication**: Requires `X-Admin-Key` header (performance protection).
#[utoipa::path(
    get,
    path = "/monitoring/orderbooks",
    responses(
        (status = 200, description = "Orderbook integrity report", body = OrderbookIntegrityReport),
        (status = 401, description = "Invalid or missing X-Admin-Key header")
    ),
    tag = "Monitoring",
    security(("admin_key" = []))
)]
pub async fn orderbook_integrity(State(app_state): State<AdminState>) -> impl IntoResponse {
    let quotes = app_state.quote_provider.all_quotes();
    let staleness = app_state.quote_provider.staleness();
    let l2_seq = app_state.quote_provider.l2_seq();
    let now = chrono::Utc::now();

    let mut crossed_orderbooks = Vec::new();
    let age_seconds = staleness.as_secs() as i64;

    for (symbol, quote) in &quotes {
        if let (Some(bid), Some(ask)) = (quote.best_bid, quote.best_ask) {
            if bid > 0.0 && ask > 0.0 && bid >= ask {
                crossed_orderbooks.push(CrossedOrderbook {
                    symbol: symbol.clone(),
                    best_bid: bid,
                    best_ask: ask,
                    spread: ask - bid,
                    age_seconds,
                    last_l2_seq: Some(l2_seq),
                    bid_levels: quote.bids.len(),
                    ask_levels: quote.asks.len(),
                    top_bids: quote.bids.iter().take(10).cloned().collect(),
                    top_asks: quote.asks.iter().take(10).cloned().collect(),
                });
            }
        }
    }

    // Sort by spread (most negative/crossed first)
    crossed_orderbooks.sort_by(|a, b| {
        a.spread
            .partial_cmp(&b.spread)
            .unwrap_or(std::cmp::Ordering::Equal)
    });

    // Stale if no snapshot update in 30+ seconds
    let stale_count = if age_seconds > 30 { quotes.len() } else { 0 };

    let report = OrderbookIntegrityReport {
        timestamp: now.to_rfc3339(),
        total_orderbooks: quotes.len(),
        crossed_count: crossed_orderbooks.len(),
        stale_count,
        crossed_orderbooks,
    };

    (StatusCode::OK, SonicJson(report)).into_response()
}

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

    #[test]
    fn position_balance_excludes_external_perp_exposure() {
        let positions = HashMap::from([
            ("BTC-PERP".to_string(), dec!(-0.031)),
            ("ETH-20260501-2000-C".to_string(), dec!(0)),
        ]);

        let result = evaluate_position_balance(&positions);

        assert_eq!(result.status, IntegrityStatus::Healthy);
        assert!(result.imbalances.is_empty());
        assert_eq!(result.total_imbalance, dec!(0));
        assert_eq!(
            result.excluded_perp_exposures,
            vec![PerpExposure {
                symbol: "BTC-PERP".to_string(),
                net_position: "-0.031".to_string(),
            }]
        );
    }

    #[test]
    fn position_balance_flags_option_imbalance_even_with_perp_exposure() {
        let positions = HashMap::from([
            ("BTC-PERP".to_string(), dec!(-0.031)),
            ("ETH-20260501-2000-C".to_string(), dec!(0.02)),
        ]);

        let result = evaluate_position_balance(&positions);

        assert_eq!(result.status, IntegrityStatus::Critical);
        assert_eq!(result.total_imbalance, dec!(0.02));
        assert_eq!(
            result.imbalances,
            vec![PositionImbalance {
                symbol: "ETH-20260501-2000-C".to_string(),
                net_position: "0.02".to_string(),
                severity: IntegrityStatus::Critical,
            }]
        );
        assert_eq!(result.excluded_perp_exposures.len(), 1);
    }

    #[test]
    fn position_balance_preserves_warning_threshold_for_small_option_imbalance() {
        let positions = HashMap::from([("ETH-20260501-2000-C".to_string(), dec!(0.001))]);

        let result = evaluate_position_balance(&positions);

        assert_eq!(result.status, IntegrityStatus::Warning);
        assert_eq!(result.imbalances[0].severity, IntegrityStatus::Warning);
    }
}