hypercall/rsm/

engine_snapshot.rs

use crate::vol_oracle::vol_surface_cache::VolatilitySurface;
use async_trait::async_trait;
use hypercall_runtime_api::{
    AgentAuthProvider, BookSnapshotState, OrderSnapshotProvider, QuoteProvider, QuoteSnapshot,
    RuntimeOrderSummary, SnapshotBookQuote,
};
use hypercall_types::api_models::Order as ApiOrder;
use hypercall_types::WalletAddress;
use rust_decimal::prelude::ToPrimitive;
use rust_decimal_macros::dec;
use serde::{Deserialize, Serialize};
use sha3::{Digest, Keccak256};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};

use arc_swap::ArcSwap;
use hypercall_recovery::RestartStateComponent;

use super::ledger::{BalanceLedger, BalanceProvider, LedgerError};
use super::restart_components::{EngineRecoveryCapture, PersistentEngineStateComponent};
use hypercall_engine::{EngineOrderIndex, OrderBook, OrderSummary};

/// Read-only snapshot of engine orderbook state, published via ArcSwap
/// for lock-free reads by API handlers.
#[derive(Clone, Debug)]
pub struct EngineSnapshot {
    pub quotes: HashMap<String, SnapshotBookQuote>,
    pub orders_by_wallet: HashMap<WalletAddress, Vec<OrderSummary>>,
    pub balance_ledger: BalanceLedger,
    /// All order IDs from orderbooks with their wallet addresses.
    /// Sourced directly from orderbook data, not the order index,
    /// so it includes orders placed before the index was built.
    pub all_orderbook_orders: Vec<(u64, WalletAddress)>,
    /// Agent authorizations: owner wallet → list of (agent wallet, optional expiry ms).
    pub agent_authorizations: HashMap<WalletAddress, Vec<(WalletAddress, Option<u64>)>>,
    pub snapshot_published: bool,
    pub l2_seq: i64,
    pub published_at: Instant,
    pub engine_state_digest: EngineStateDigest,
}

impl EngineSnapshot {
    pub fn is_agent_authorized(&self, wallet: &WalletAddress, agent: &WalletAddress) -> bool {
        if wallet == agent {
            return true;
        }
        let now_ms = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_millis() as u64)
            .unwrap_or(0);
        self.agent_authorizations
            .get(wallet)
            .and_then(|agents| {
                agents.iter().find_map(|(a, expires)| {
                    if a != agent {
                        return None;
                    }
                    match expires {
                        None => Some(true),
                        Some(ts) => Some(*ts > now_ms),
                    }
                })
            })
            .unwrap_or(false)
    }
}

/// Deterministic digest of engine-owned state at a published snapshot boundary.
///
/// This intentionally excludes freshness-only process state such as quote
/// staleness timestamps. It is meant for primary-vs-standby parity checks when
/// both sides report the same sequence.
#[derive(Clone, Debug, Serialize, Deserialize, utoipa::ToSchema)]
pub struct EngineStateDigest {
    pub l2_seq: i64,
    pub next_order_id: u64,
    pub next_trade_id: u64,
    pub overall_digest: String,
    pub orders_digest: String,
    pub orders_count: usize,
    pub positions_digest: String,
    pub positions_count: usize,
    pub cash_digest: String,
    pub cash_wallet_count: usize,
    pub markets_digest: String,
    pub expired_instruments_count: usize,
    pub trading_modes_count: usize,
    pub prices_digest: String,
    pub spot_price_count: usize,
    pub iv_surface_count: usize,
    pub iv_source_timestamps_digest: String,
    pub iv_source_timestamp_count: usize,
    pub perp_positions_digest: String,
    pub perp_positions_count: usize,
    pub hypercore_equity_digest: String,
    pub hypercore_equity_count: usize,
    pub mmp_digest: String,
    pub mmp_state_count: usize,
    pub mmp_enabled_digest: String,
    pub mmp_enabled_count: usize,
    pub liquidation_states_digest: String,
    pub liquidation_state_count: usize,
    pub wallet_margin_modes_digest: String,
    pub wallet_margin_mode_count: usize,
    pub wallet_trading_limits_digest: String,
    pub wallet_trading_limits_count: usize,
    pub wallet_tiers_digest: String,
    pub wallet_tier_count: usize,
    pub deposit_watermarks_digest: String,
    pub deposit_watermark_count: usize,
    pub agent_auth_digest: String,
    pub agent_auth_count: usize,
    pub nonce_sets_digest: String,
    pub nonce_signer_count: usize,
    pub pm_settlement_digest: String,
    pub pm_settlement_pool_count: usize,
    pub pm_settlement_account_count: usize,
    pub pm_settlement_request_count: usize,
}

impl EngineStateDigest {
    pub fn empty() -> Self {
        let empty = hex_digest(|_| {});
        Self {
            l2_seq: 0,
            next_order_id: 0,
            next_trade_id: 0,
            overall_digest: empty.clone(),
            orders_digest: empty.clone(),
            orders_count: 0,
            positions_digest: empty.clone(),
            positions_count: 0,
            cash_digest: empty.clone(),
            cash_wallet_count: 0,
            markets_digest: empty.clone(),
            expired_instruments_count: 0,
            trading_modes_count: 0,
            prices_digest: empty.clone(),
            spot_price_count: 0,
            iv_surface_count: 0,
            iv_source_timestamps_digest: empty.clone(),
            iv_source_timestamp_count: 0,
            perp_positions_digest: empty.clone(),
            perp_positions_count: 0,
            hypercore_equity_digest: empty.clone(),
            hypercore_equity_count: 0,
            mmp_digest: empty.clone(),
            mmp_state_count: 0,
            mmp_enabled_digest: empty.clone(),
            mmp_enabled_count: 0,
            liquidation_states_digest: empty.clone(),
            liquidation_state_count: 0,
            wallet_margin_modes_digest: empty.clone(),
            wallet_margin_mode_count: 0,
            wallet_trading_limits_digest: empty.clone(),
            wallet_trading_limits_count: 0,
            wallet_tiers_digest: empty.clone(),
            wallet_tier_count: 0,
            deposit_watermarks_digest: empty.clone(),
            deposit_watermark_count: 0,
            agent_auth_digest: empty.clone(),
            agent_auth_count: 0,
            nonce_sets_digest: empty,
            nonce_signer_count: 0,
            pm_settlement_digest: hex_digest(|_| {}),
            pm_settlement_pool_count: 0,
            pm_settlement_account_count: 0,
            pm_settlement_request_count: 0,
        }
    }

    pub fn from_ctx(ctx: &crate::rsm::engine_deps::EngineCtx, l2_seq: i64) -> Self {
        let mut orders_count = 0usize;
        let orders_digest = hex_digest(|h| {
            let mut orderbooks: Vec<_> = ctx.orderbooks.iter().collect();
            orderbooks.sort_by(|a, b| a.0.cmp(b.0));
            for (symbol, book) in orderbooks {
                update_str(h, symbol);
                let mut orders = book.get_all_orders();
                orders.sort_by_key(|order| order.order_id);
                update_u64(h, orders.len() as u64);
                orders_count += orders.len();
                for order in orders {
                    update_u64(h, order.order_id);
                    update_decimal(h, &order.price);
                    update_decimal(h, &order.quantity);
                    update_str(h, &format!("{:?}", order.side));
                    update_wallet(h, &order.wallet);
                    update_u64(h, order.timestamp);
                    update_opt_str(h, order.client_id.as_deref());
                    update_bool(h, order.mmp_enabled);
                    update_decimal(h, &order.original_size);
                }
            }
        });

        let positions_digest = hex_digest(|h| {
            let mut positions: Vec<_> = ctx.engine_positions.iter().collect();
            positions.sort_by(|((wallet_a, symbol_a), _), ((wallet_b, symbol_b), _)| {
                wallet_a.cmp(wallet_b).then_with(|| symbol_a.cmp(symbol_b))
            });
            for ((wallet, symbol), position) in positions {
                update_wallet(h, wallet);
                update_str(h, symbol);
                update_decimal(h, &position.quantity);
                update_decimal(h, &position.entry_price);
            }
        });

        let cash_digest = hex_digest(|h| {
            let cash = ctx.balance_ledger.sorted_nonzero_entries();
            for (wallet, balance) in cash {
                update_wallet(h, &wallet);
                update_decimal(h, &balance);
            }
        });

        let markets_digest = hex_digest(|h| {
            let mut expired: Vec<_> = ctx.expired_instruments.iter().collect();
            expired.sort_by(|a, b| a.0.cmp(b.0));
            for (symbol, is_expired) in expired {
                update_str(h, symbol);
                update_bool(h, *is_expired);
            }

            let mut trading_modes: Vec<_> = ctx.instrument_trading_modes.iter().collect();
            trading_modes.sort_by(|a, b| a.0.cmp(b.0));
            for (symbol, mode) in trading_modes {
                update_str(h, symbol);
                update_str(h, &format!("{:?}", mode));
            }
        });

        let prices_digest = hex_digest(|h| {
            let mut spot_prices: Vec<_> = ctx.spot_prices.iter().collect();
            spot_prices.sort_by(|a, b| a.0.cmp(b.0));
            for (underlying, price) in spot_prices {
                update_str(h, underlying);
                update_decimal(h, price);
            }

            let mut iv_surfaces: Vec<_> = ctx.iv_surfaces.iter().collect();
            iv_surfaces.sort_by(|a, b| a.0.cmp(b.0));
            for (underlying, surface) in iv_surfaces {
                update_str(h, underlying);
                update_volatility_surface(h, surface);
            }
        });

        let iv_source_timestamps_digest = hex_digest(|h| {
            let mut timestamps: Vec<_> = ctx.iv_source_timestamps.iter().collect();
            timestamps.sort_by(|a, b| a.0.cmp(b.0));
            for (underlying, timestamp) in timestamps {
                update_str(h, underlying);
                update_i64(h, *timestamp);
            }
        });

        let perp_positions_digest = hex_digest(|h| {
            let mut positions: Vec<_> = ctx.deps.perp_positions.iter().collect();
            positions.sort_by(|((account_a, coin_a), _), ((account_b, coin_b), _)| {
                account_a.cmp(account_b).then_with(|| coin_a.cmp(coin_b))
            });
            for ((account, coin), position) in positions {
                update_str(h, account);
                update_str(h, coin);
                update_str(h, &position.coin);
                update_f64(h, position.size);
                update_opt_f64(h, position.entry_price);
                update_f64(h, position.position_value);
                update_f64(h, position.unrealized_pnl);
                update_f64(h, position.margin_used);
                update_opt_f64(h, position.liquidation_price);
            }
        });

        let hypercore_equity_digest = hex_digest(|h| {
            let mut equities: Vec<_> = ctx.deps.hypercore_account_equity.iter().collect();
            equities.sort_by_key(|(wallet, _)| **wallet);
            for (wallet, equity) in equities {
                update_wallet(h, wallet);
                update_decimal(h, equity);
                let timestamp = ctx.deps.hypercore_equity_timestamps.get(wallet);
                update_opt_u64(h, timestamp.copied());
            }

            let mut timestamp_only_wallets: Vec<_> = ctx
                .deps
                .hypercore_equity_timestamps
                .iter()
                .filter(|(wallet, _)| !ctx.deps.hypercore_account_equity.contains_key(wallet))
                .collect();
            timestamp_only_wallets.sort_by_key(|(wallet, _)| **wallet);
            for (wallet, timestamp) in timestamp_only_wallets {
                update_wallet(h, wallet);
                update_u64(h, *timestamp);
            }
        });

        let mmp_digest = hex_digest(|h| {
            let mut states: Vec<_> = ctx.mmp_state.iter().collect();
            states.sort_by(|((wallet_a, currency_a), _), ((wallet_b, currency_b), _)| {
                wallet_a
                    .cmp(wallet_b)
                    .then_with(|| currency_a.cmp(currency_b))
            });
            for ((wallet, currency), state) in states {
                update_wallet(h, wallet);
                update_str(h, currency);
                update_bool(h, state.enabled);
                update_i64(h, state.interval_ms);
                update_i64(h, state.frozen_time_ms);
                update_opt_f64(h, state.qty_limit);
                update_opt_f64(h, state.delta_limit);
                update_opt_f64(h, state.vega_limit);
                update_u64(h, state.cumulative_qty);
                update_f64(h, state.cumulative_delta);
                update_f64(h, state.cumulative_vega);
                update_opt_u64(h, state.frozen_until);
                update_u64(h, state.fills.len() as u64);
                for fill in &state.fills {
                    update_u64(h, fill.timestamp_ms);
                    update_u64(h, fill.quantity);
                    update_f64(h, fill.delta);
                    update_f64(h, fill.vega);
                }
            }
        });

        let mmp_enabled_digest = hex_digest(|h| {
            let mut configs: Vec<_> = ctx.deps.mmp_enabled.iter().collect();
            configs.sort_by(|((wallet_a, currency_a), _), ((wallet_b, currency_b), _)| {
                wallet_a
                    .cmp(wallet_b)
                    .then_with(|| currency_a.cmp(currency_b))
            });
            for ((wallet, currency), enabled) in configs {
                update_wallet(h, wallet);
                update_str(h, currency);
                update_bool(h, *enabled);
            }
        });

        let liquidation_states_digest = hex_digest(|h| {
            let mut states: Vec<_> = ctx.deps.liquidation_states.iter().collect();
            states.sort_by_key(|(wallet, _)| **wallet);
            for (wallet, state) in states {
                update_wallet(h, wallet);
                update_str(h, &format!("{:?}", state));
            }
        });

        let wallet_margin_modes_digest = hex_digest(|h| {
            let mut modes: Vec<_> = ctx.deps.wallet_margin_modes.iter().collect();
            modes.sort_by_key(|(wallet, _)| **wallet);
            for (wallet, mode) in modes {
                update_wallet(h, wallet);
                update_str(h, &format!("{:?}", mode));
            }
        });

        let wallet_trading_limits_digest = hex_digest(|h| {
            update_trading_limits(h, &ctx.deps.default_trading_limits);
            let mut limits: Vec<_> = ctx.deps.wallet_trading_limits.iter().collect();
            limits.sort_by_key(|(wallet, _)| **wallet);
            for (wallet, trading_limits) in limits {
                update_wallet(h, wallet);
                update_trading_limits(h, trading_limits);
            }
        });

        let wallet_tiers_digest = hex_digest(|h| {
            let mut tiers: Vec<_> = ctx.deps.wallet_tiers.iter().collect();
            tiers.sort_by_key(|(wallet, _)| **wallet);
            for (wallet, tier) in tiers {
                update_wallet(h, wallet);
                update_str(h, tier);
            }
        });

        let deposit_watermarks_digest = hex_digest(|h| {
            let mut watermarks: Vec<_> = ctx.deposit_update_watermarks.iter().collect();
            watermarks.sort_by_key(|(wallet, _)| **wallet);
            for (wallet, watermark) in watermarks {
                update_wallet(h, wallet);
                update_opt_u64(h, watermark.sequence);
                update_u64(h, watermark.timestamp_ms);
                update_decimal(h, &watermark.balance_after);
            }
            update_len(h, ctx.applied_deposit_source_event_hashes.len());
            for source_hash in &ctx.applied_deposit_source_event_hashes {
                h.update(source_hash.as_slice());
            }
        });

        let agent_auth_count: usize = ctx.agent_authorizations.values().map(|s| s.len()).sum();
        let agent_auth_digest = hex_digest(|h| {
            let mut wallets: Vec<_> = ctx.agent_authorizations.iter().collect();
            wallets.sort_by_key(|(w, _)| **w);
            for (wallet, agents) in wallets {
                update_wallet(h, wallet);
                let mut sorted_agents: Vec<_> = agents.iter().collect();
                sorted_agents.sort_by_key(|(a, _)| *a);
                update_u64(h, sorted_agents.len() as u64);
                for (agent, expires) in sorted_agents {
                    update_wallet(h, agent);
                    update_opt_u64(h, *expires);
                }
            }
        });

        let nonce_signer_count = ctx.nonce_sets.len();
        let nonce_sets_digest = hex_digest(|h| {
            let mut signers: Vec<_> = ctx.nonce_sets.iter().collect();
            signers.sort_by_key(|(w, _)| **w);
            for (signer, set) in signers {
                update_wallet(h, signer);
                update_u64(h, set.count() as u64);
                for &nonce in set.iter() {
                    update_u64(h, nonce);
                }
            }
        });

        let pm_settlement_digest = hex_digest(|h| {
            update_pm_settlement_state(h, &ctx.pm_settlement_state);
        });

        let next_order_id = ctx.next_order_id;
        let next_trade_id = ctx.next_trade_id;
        let overall_digest = hex_digest(|h| {
            update_i64(h, l2_seq);
            update_u64(h, next_order_id);
            update_u64(h, next_trade_id);
            update_str(h, &orders_digest);
            update_str(h, &positions_digest);
            update_str(h, &cash_digest);
            update_str(h, &markets_digest);
            update_str(h, &prices_digest);
            update_str(h, &iv_source_timestamps_digest);
            update_str(h, &perp_positions_digest);
            update_str(h, &hypercore_equity_digest);
            update_str(h, &mmp_digest);
            update_str(h, &mmp_enabled_digest);
            update_str(h, &liquidation_states_digest);
            update_str(h, &wallet_margin_modes_digest);
            update_str(h, &wallet_trading_limits_digest);
            update_str(h, &wallet_tiers_digest);
            update_str(h, &deposit_watermarks_digest);
            update_str(h, &agent_auth_digest);
            update_str(h, &nonce_sets_digest);
            update_str(h, &pm_settlement_digest);
        });

        Self {
            l2_seq,
            next_order_id,
            next_trade_id,
            overall_digest,
            orders_digest,
            orders_count,
            positions_digest,
            positions_count: ctx.engine_positions.len(),
            cash_digest,
            cash_wallet_count: ctx.balance_ledger.len(),
            markets_digest,
            expired_instruments_count: ctx.expired_instruments.len(),
            trading_modes_count: ctx.instrument_trading_modes.len(),
            prices_digest,
            spot_price_count: ctx.spot_prices.len(),
            iv_surface_count: ctx.iv_surfaces.len(),
            iv_source_timestamps_digest,
            iv_source_timestamp_count: ctx.iv_source_timestamps.len(),
            perp_positions_digest,
            perp_positions_count: ctx.deps.perp_positions.len(),
            hypercore_equity_digest,
            hypercore_equity_count: ctx.deps.hypercore_account_equity.len(),
            mmp_digest,
            mmp_state_count: ctx.mmp_state.len(),
            mmp_enabled_digest,
            mmp_enabled_count: ctx.deps.mmp_enabled.len(),
            liquidation_states_digest,
            liquidation_state_count: ctx.deps.liquidation_states.len(),
            wallet_margin_modes_digest,
            wallet_margin_mode_count: ctx.deps.wallet_margin_modes.len(),
            wallet_trading_limits_digest,
            wallet_trading_limits_count: ctx.deps.wallet_trading_limits.len(),
            wallet_tiers_digest,
            wallet_tier_count: ctx.deps.wallet_tiers.len(),
            deposit_watermarks_digest,
            deposit_watermark_count: ctx.deposit_update_watermarks.len(),
            agent_auth_digest,
            agent_auth_count,
            nonce_sets_digest,
            nonce_signer_count,
            pm_settlement_digest,
            pm_settlement_pool_count: ctx.pm_settlement_state.pools.len(),
            pm_settlement_account_count: ctx.pm_settlement_state.accounts.len(),
            pm_settlement_request_count: ctx.pm_settlement_state.idempotency.len(),
        }
    }
}

impl EngineSnapshot {
    /// Build a snapshot from the engine's live orderbooks and order index.
    pub fn build(
        orderbooks: &HashMap<String, OrderBook>,
        order_index: &EngineOrderIndex,
        l2_seq: i64,
    ) -> Self {
        Self::build_with_digest(
            orderbooks,
            order_index,
            BalanceLedger::new(),
            HashMap::new(),
            l2_seq,
            EngineStateDigest::empty(),
        )
    }

    /// Build a snapshot from the full engine context, including parity digest state.
    pub fn build_with_ctx(ctx: &crate::rsm::engine_deps::EngineCtx, l2_seq: i64) -> Self {
        let agent_authorizations: HashMap<WalletAddress, Vec<(WalletAddress, Option<u64>)>> = ctx
            .agent_authorizations
            .iter()
            .map(|(wallet, agents)| {
                let mut sorted: Vec<(WalletAddress, Option<u64>)> =
                    agents.iter().map(|(a, e)| (*a, *e)).collect();
                sorted.sort_by_key(|(a, _)| *a);
                (*wallet, sorted)
            })
            .collect();

        Self::build_with_digest(
            &ctx.orderbooks,
            &ctx.order_index,
            ctx.balance_ledger.clone(),
            agent_authorizations,
            l2_seq,
            PersistentEngineStateComponent::digest(&EngineRecoveryCapture::for_digest(ctx, l2_seq)),
        )
    }

    fn build_with_digest(
        orderbooks: &HashMap<String, OrderBook>,
        order_index: &EngineOrderIndex,
        balance_ledger: BalanceLedger,
        agent_authorizations: HashMap<WalletAddress, Vec<(WalletAddress, Option<u64>)>>,
        l2_seq: i64,
        engine_state_digest: EngineStateDigest,
    ) -> Self {
        let mut quotes = HashMap::with_capacity(orderbooks.len());

        for (symbol, book) in orderbooks {
            let best_bid = book.get_best_bid().and_then(|d| d.to_f64());
            let best_ask = book.get_best_ask().and_then(|d| d.to_f64());

            let bid_depth = book.get_bid_depth();
            let ask_depth = book.get_ask_depth();

            let best_bid_size = bid_depth.first().and_then(|(_, sz)| sz.to_f64());
            let best_ask_size = ask_depth.first().and_then(|(_, sz)| sz.to_f64());

            let mid = match (best_bid, best_ask) {
                (Some(b), Some(a)) if b > 0.0 && a > 0.0 => Some((b + a) / 2.0),
                (Some(b), None) if b > 0.0 => Some(b),
                (None, Some(a)) if a > 0.0 => Some(a),
                _ => None,
            };

            let bids: Vec<(f64, f64)> = bid_depth
                .iter()
                .filter_map(|(p, s)| Some((p.to_f64()?, s.to_f64()?)))
                .collect();

            let asks: Vec<(f64, f64)> = ask_depth
                .iter()
                .filter_map(|(p, s)| Some((p.to_f64()?, s.to_f64()?)))
                .collect();

            quotes.insert(
                symbol.clone(),
                SnapshotBookQuote {
                    best_bid,
                    best_bid_size,
                    best_ask,
                    best_ask_size,
                    mid,
                    bids,
                    asks,
                },
            );
        }

        let orders_by_wallet = order_index.snapshot_orders();

        // Collect all orders directly from orderbooks (not order index).
        // This captures orders that predate the order index (e.g., orphans
        // from previous MM instances that were loaded from the WAL snapshot).
        let total_orderbook_orders = orderbooks.values().map(OrderBook::order_count).sum();
        let mut all_orderbook_orders = Vec::with_capacity(total_orderbook_orders);
        for book in orderbooks.values() {
            book.append_order_wallets(&mut all_orderbook_orders);
        }

        Self {
            quotes,
            orders_by_wallet,
            balance_ledger,
            all_orderbook_orders,
            agent_authorizations,
            snapshot_published: true,
            l2_seq,
            published_at: Instant::now(),
            engine_state_digest,
        }
    }

    /// Create an empty snapshot (used at startup before first publish).
    pub fn empty() -> Self {
        Self {
            quotes: HashMap::new(),
            orders_by_wallet: HashMap::new(),
            balance_ledger: BalanceLedger::new(),
            all_orderbook_orders: Vec::new(),
            agent_authorizations: HashMap::new(),
            snapshot_published: false,
            l2_seq: 0,
            published_at: Instant::now(),
            engine_state_digest: EngineStateDigest::empty(),
        }
    }
}

fn hex_digest(build: impl FnOnce(&mut Keccak256)) -> String {
    let mut hasher = Keccak256::new();
    build(&mut hasher);
    format!("0x{}", hex::encode(hasher.finalize()))
}

fn update_len(hasher: &mut Keccak256, len: usize) {
    hasher.update((len as u64).to_le_bytes());
}

fn update_str(hasher: &mut Keccak256, value: &str) {
    update_len(hasher, value.len());
    hasher.update(value.as_bytes());
}

fn update_uuid(hasher: &mut Keccak256, value: &uuid::Uuid) {
    hasher.update(value.as_bytes());
}

fn update_opt_str(hasher: &mut Keccak256, value: Option<&str>) {
    update_bool(hasher, value.is_some());
    if let Some(value) = value {
        update_str(hasher, value);
    }
}

fn update_bool(hasher: &mut Keccak256, value: bool) {
    hasher.update([value as u8]);
}

fn update_u64(hasher: &mut Keccak256, value: u64) {
    hasher.update(value.to_le_bytes());
}

fn update_opt_u64(hasher: &mut Keccak256, value: Option<u64>) {
    update_bool(hasher, value.is_some());
    if let Some(value) = value {
        update_u64(hasher, value);
    }
}

fn update_i64(hasher: &mut Keccak256, value: i64) {
    hasher.update(value.to_le_bytes());
}

fn update_opt_i64(hasher: &mut Keccak256, value: Option<i64>) {
    update_bool(hasher, value.is_some());
    if let Some(value) = value {
        update_i64(hasher, value);
    }
}

fn update_i32(hasher: &mut Keccak256, value: i32) {
    hasher.update(value.to_le_bytes());
}

fn update_f64(hasher: &mut Keccak256, value: f64) {
    hasher.update(value.to_bits().to_le_bytes());
}

fn update_opt_f64(hasher: &mut Keccak256, value: Option<f64>) {
    update_bool(hasher, value.is_some());
    if let Some(value) = value {
        update_f64(hasher, value);
    }
}

fn update_decimal(hasher: &mut Keccak256, value: &rust_decimal::Decimal) {
    update_str(hasher, &value.normalize().to_string());
}

fn update_wallet(hasher: &mut Keccak256, wallet: &WalletAddress) {
    update_str(hasher, &wallet.to_string());
}

fn update_u32(hasher: &mut Keccak256, value: u32) {
    hasher.update(value.to_le_bytes());
}

fn update_opt_decimal(hasher: &mut Keccak256, value: Option<&rust_decimal::Decimal>) {
    update_bool(hasher, value.is_some());
    if let Some(value) = value {
        update_decimal(hasher, value);
    }
}

fn update_pm_settlement_config(
    hasher: &mut Keccak256,
    config: &hypercall_margin::portfolio::PmSettlementPoolConfig,
) {
    update_decimal(hasher, &config.target_short_oi_notional_multiplier);
    update_decimal(hasher, &config.utilization_kink);
    update_decimal(hasher, &config.apr_at_kink);
    update_decimal(hasher, &config.max_apr);
    update_decimal(hasher, &config.normal_utilization_cap);
    update_decimal(hasher, &config.crisis_utilization_cap);
    update_i64(hasher, config.bridge_window_ms);
    update_u32(hasher, config.policy_version);
}

fn update_pm_settlement_state(
    hasher: &mut Keccak256,
    state: &crate::rsm::portfolio_margin::settlement_state::PmSettlementState,
) {
    update_u64(hasher, state.pools.len() as u64);
    for (underlying, pool) in &state.pools {
        update_str(hasher, underlying);
        update_str(hasher, &pool.underlying);
        update_decimal(hasher, &pool.pool_available_usdc);
        update_decimal(hasher, &pool.pool_target_usdc);
        update_decimal(hasher, &pool.active_timing_bridge_usdc);
        update_decimal(hasher, &pool.active_settlement_debt_usdc);
        update_u32(hasher, pool.config_version);
        update_u32(hasher, pool.policy_version);
        update_bool(hasher, pool.config.is_some());
        if let Some(config) = &pool.config {
            update_pm_settlement_config(hasher, config);
        }
        update_opt_decimal(hasher, pool.utilization.as_ref());
        update_u64(hasher, pool.updated_at_ms);
    }

    update_u64(hasher, state.accounts.len() as u64);
    for (key, account) in &state.accounts {
        update_wallet(hasher, &key.wallet);
        update_str(hasher, &key.underlying);
        update_wallet(hasher, &account.wallet);
        update_str(hasher, &account.underlying);
        update_decimal(hasher, &account.bridge_principal_usdc);
        update_decimal(hasher, &account.debt_principal_usdc);
        update_decimal(hasher, &account.accrued_interest_usdc);
        update_i64(hasher, account.last_interest_accrual_ms);
        update_u32(hasher, account.policy_version);
        update_opt_i64(hasher, account.bridge_deadline_ms);
        update_str(hasher, &format!("{:?}", account.status));
        update_opt_str(hasher, account.active_recovery_plan_id.as_deref());
        update_u64(hasher, account.updated_at_ms);
    }

    update_u64(hasher, state.events.len() as u64);
    for (key, event) in &state.events {
        update_wallet(hasher, &key.wallet);
        update_str(hasher, &key.market_id);
        update_i64(hasher, key.expiry_ts_ms);
        update_str(hasher, &key.margin_mode);
        update_u64(hasher, key.settlement_event_sequence);
        update_wallet(hasher, &event.event_key.wallet);
        update_str(hasher, &event.event_key.market_id);
        update_i64(hasher, event.event_key.expiry_ts_ms);
        update_str(hasher, &event.event_key.margin_mode);
        update_u64(hasher, event.event_key.settlement_event_sequence);
        update_str(hasher, &event.input_digest);
        update_str(hasher, &event.output_digest);
        update_str(hasher, &event.status);
        update_uuid(hasher, &event.request_id);
        update_str(hasher, &event.underlying);
        update_str(hasher, &event.event_type);
        update_decimal(hasher, &event.amount_usdc);
    }

    update_u64(hasher, state.idempotency.len() as u64);
    for (request_id, record) in &state.idempotency {
        update_str(hasher, request_id);
        update_uuid(hasher, &record.request_id);
        update_str(hasher, &record.input_digest);
        update_str(hasher, &record.output_digest);
        update_str(hasher, &record.command_type);
        update_u64(hasher, record.applied_at_ms);
    }

    update_u64(hasher, state.recovery_plans.len() as u64);
    for (plan_id, plan) in &state.recovery_plans {
        update_str(hasher, plan_id);
        update_str(hasher, &plan.plan_id);
        update_wallet(hasher, &plan.wallet);
        update_str(hasher, &plan.underlying);
        update_str(hasher, &plan.trigger);
        update_str(hasher, &plan.reason);
        update_u32(hasher, plan.policy_version);
        update_u32(hasher, plan.recovery_priority_version);
        update_str(hasher, &plan.status);
        update_str(hasher, &plan.input_digest);
        update_decimal(hasher, &plan.target_reduction_usdc);
        update_decimal(hasher, &plan.expected_usdc_recovered);
        update_decimal(hasher, &plan.expected_obligation_reduced);
        update_decimal(hasher, &plan.expected_impact_usdc);
        update_opt_decimal(hasher, plan.post_plan_utilization.as_ref());
        update_u64(hasher, plan.actions.len() as u64);
        for action in &plan.actions {
            update_u32(hasher, action.action_index);
            update_str(hasher, &action.action_type);
            update_str(hasher, &action.target);
            update_str(hasher, &action.action_payload);
            update_str(hasher, &action.status);
            update_decimal(hasher, &action.expected_usdc_recovered);
            update_decimal(hasher, &action.expected_obligation_reduced);
            update_decimal(hasher, &action.expected_impact_usdc);
            update_u32(hasher, action.attempt);
            update_opt_str(hasher, action.submitted_external_id.as_deref());
            update_opt_str(hasher, action.external_kind.as_deref());
            update_opt_str(hasher, action.result.as_deref());
            update_opt_str(hasher, action.result_external_id.as_deref());
            update_decimal(hasher, &action.recovered_usdc);
            update_decimal(hasher, &action.liability_reduction_usdc);
            update_u64(hasher, action.updated_at_ms);
        }
        update_u64(hasher, plan.updated_at_ms);
    }
}

fn update_trading_limits(
    hasher: &mut Keccak256,
    limits: &hypercall_types::api_models::TradingLimits,
) {
    update_i32(hasher, limits.max_open_orders);
    update_i32(hasher, limits.max_open_positions);
    update_i32(hasher, limits.orders_per_minute);
    update_i32(hasher, limits.cancels_per_minute);
    update_i32(hasher, limits.api_requests_per_minute);
}

fn update_volatility_surface(hasher: &mut Keccak256, surface: &VolatilitySurface) {
    let mut strike_points = surface.export_all_points();
    strike_points.sort_by(|a, b| {
        a.expiry
            .cmp(&b.expiry)
            .then_with(|| a.strike.total_cmp(&b.strike))
            .then_with(|| a.iv.total_cmp(&b.iv))
    });
    update_u64(hasher, strike_points.len() as u64);
    for point in strike_points {
        update_f64(hasher, point.strike);
        update_i64(hasher, point.expiry);
        update_f64(hasher, point.iv);
    }

    let mut atm_vols = surface.export_atm_vols();
    atm_vols.sort_by_key(|(expiry, _)| *expiry);
    update_u64(hasher, atm_vols.len() as u64);
    for (expiry, iv) in atm_vols {
        update_i64(hasher, expiry);
        update_f64(hasher, iv);
    }

    let mut delta_curves = surface.export_delta_curves();
    delta_curves.sort_by_key(|curve| curve.expiry);
    update_u64(hasher, delta_curves.len() as u64);
    for mut curve in delta_curves {
        update_i64(hasher, curve.expiry);
        curve.points.sort_by(|a, b| {
            a.delta
                .total_cmp(&b.delta)
                .then_with(|| a.iv.total_cmp(&b.iv))
        });
        update_u64(hasher, curve.points.len() as u64);
        for point in curve.points {
            update_f64(hasher, point.delta);
            update_f64(hasher, point.iv);
        }
    }
}

pub trait EngineStateDigestProvider: Send + Sync {
    fn engine_state_digest(&self) -> EngineStateDigest;
}

/// Production implementation of QuoteProvider backed by ArcSwap<EngineSnapshot>.
///
/// The engine publishes snapshots into the ArcSwap; this provider loads them
/// lock-free for API reads.
pub struct SnapshotQuoteProvider {
    snapshot: Arc<ArcSwap<EngineSnapshot>>,
}

impl SnapshotQuoteProvider {
    pub fn new(snapshot: Arc<ArcSwap<EngineSnapshot>>) -> Self {
        Self { snapshot }
    }

    /// Get the underlying ArcSwap handle (needed by the engine to publish into).
    pub fn snapshot_handle(&self) -> &Arc<ArcSwap<EngineSnapshot>> {
        &self.snapshot
    }

    pub fn balance_ledger_sync_snapshot(&self) -> (crate::rsm::ledger::BalanceLedger, u64) {
        let snapshot = self.snapshot.load();
        (
            snapshot.balance_ledger.clone(),
            snapshot.balance_ledger.last_balance_update_seq(),
        )
    }
}

impl QuoteProvider for SnapshotQuoteProvider {
    fn get_quote(&self, symbol: &str) -> Option<SnapshotBookQuote> {
        self.snapshot.load().quotes.get(symbol).cloned()
    }

    fn get_quote_with_seq(&self, symbol: &str) -> (Option<SnapshotBookQuote>, i64) {
        let snap = self.snapshot.load();
        (snap.quotes.get(symbol).cloned(), snap.l2_seq)
    }

    fn book_snapshot_state(&self, symbol: &str) -> BookSnapshotState {
        let snap = self.snapshot.load();
        if !snap.snapshot_published {
            return BookSnapshotState::NotReady {
                l2_seq: snap.l2_seq,
            };
        }

        BookSnapshotState::Ready {
            quote: snap
                .quotes
                .get(symbol)
                .cloned()
                .unwrap_or_else(SnapshotBookQuote::empty),
            l2_seq: snap.l2_seq,
        }
    }

    fn all_quotes(&self) -> HashMap<String, SnapshotBookQuote> {
        self.snapshot.load().quotes.clone()
    }

    fn l2_seq(&self) -> i64 {
        self.snapshot.load().l2_seq
    }

    fn snapshot(&self) -> QuoteSnapshot {
        let snap = self.snapshot.load();
        QuoteSnapshot {
            quotes: snap.quotes.clone(),
            l2_seq: snap.l2_seq,
        }
    }

    fn staleness(&self) -> Duration {
        self.snapshot.load().published_at.elapsed()
    }
}

impl EngineStateDigestProvider for SnapshotQuoteProvider {
    fn engine_state_digest(&self) -> EngineStateDigest {
        self.snapshot.load().engine_state_digest.clone()
    }
}

impl AgentAuthProvider for SnapshotQuoteProvider {
    fn is_agent_authorized(&self, wallet: &WalletAddress, agent: &WalletAddress) -> bool {
        self.snapshot.load().is_agent_authorized(wallet, agent)
    }

    fn get_authorized_agents(&self, wallet: &WalletAddress) -> Vec<WalletAddress> {
        let now_ms = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_millis() as u64)
            .unwrap_or(0);
        self.snapshot
            .load()
            .agent_authorizations
            .get(wallet)
            .map(|agents| {
                agents
                    .iter()
                    .filter(|(_, expires)| expires.map_or(true, |ts| ts > now_ms))
                    .map(|(a, _)| *a)
                    .collect()
            })
            .unwrap_or_default()
    }
}

impl OrderSnapshotProvider for SnapshotQuoteProvider {
    fn get_open_orders_for_wallet(&self, wallet: &WalletAddress) -> Vec<RuntimeOrderSummary> {
        self.snapshot
            .load()
            .orders_by_wallet
            .get(wallet)
            .cloned()
            .map(|orders| orders.into_iter().map(Into::into).collect())
            .unwrap_or_default()
    }

    fn get_all_orders(&self) -> Vec<(RuntimeOrderSummary, WalletAddress)> {
        let snap = self.snapshot.load();
        // Prefer orders_by_wallet (has client_id for filtering),
        // but fall back to all_orderbook_orders for orphans not in the index.
        let mut from_index: Vec<_> = snap
            .orders_by_wallet
            .iter()
            .flat_map(|(wallet, orders)| {
                orders
                    .iter()
                    .cloned()
                    .map(move |o| (RuntimeOrderSummary::from(o), *wallet))
            })
            .collect();

        // Add any orderbook orders not already in the index
        let indexed_ids: std::collections::HashSet<u64> =
            from_index.iter().map(|(o, _)| o.order_id).collect();
        for (order_id, wallet) in &snap.all_orderbook_orders {
            if !indexed_ids.contains(order_id) {
                from_index.push((
                    RuntimeOrderSummary {
                        order_id: *order_id,
                        symbol: String::new(),
                        side: hypercall_types::Side::Buy,
                        price: rust_decimal::Decimal::ZERO,
                        original_size: rust_decimal::Decimal::ZERO,
                        remaining_size: rust_decimal::Decimal::ZERO,
                        is_perp: false,
                        mmp_enabled: false,
                        client_id: None, // unknown — will NOT match cli_* filter
                        created_at: 0,
                    },
                    *wallet,
                ));
            }
        }
        from_index
    }
}

#[async_trait]
impl BalanceProvider for SnapshotQuoteProvider {
    async fn get_balance(
        &self,
        wallet: &WalletAddress,
    ) -> Result<rust_decimal::Decimal, LedgerError> {
        let snapshot = self.snapshot.load();
        let balance = snapshot.balance_ledger.balance(wallet);
        tracing::trace!(
            wallet = %wallet,
            balance = %balance,
            l2_seq = snapshot.l2_seq,
            cash_wallet_count = snapshot.engine_state_digest.cash_wallet_count,
            "Read balance from EngineSnapshot balance_ledger"
        );
        Ok(balance)
    }
}

/// Adapter that exposes snapshot open orders through the `OpenOrdersSource` trait.
///
/// This lets `RiskAccountBuilder` consume engine-truth open orders from
/// the engine snapshot.
pub struct SnapshotOpenOrdersSource {
    order_snapshot: Arc<dyn OrderSnapshotProvider>,
}

impl SnapshotOpenOrdersSource {
    pub fn new(order_snapshot: Arc<dyn OrderSnapshotProvider>) -> Self {
        Self { order_snapshot }
    }
}

#[async_trait]
impl crate::rsm::portfolio_margin::risk_account_builder::OpenOrdersSource
    for SnapshotOpenOrdersSource
{
    async fn get_open_orders(&self, wallet: &WalletAddress) -> Vec<ApiOrder> {
        self.order_snapshot
            .get_open_orders_for_wallet(wallet)
            .into_iter()
            .map(|summary| {
                let filled_size = (summary.original_size - summary.remaining_size).max(dec!(0));
                let status = if summary.remaining_size < summary.original_size {
                    Some("partially_filled".to_string())
                } else {
                    Some("open".to_string())
                };

                ApiOrder {
                    order_id: i64::try_from(summary.order_id)
                        .expect("Engine order_id exceeded i64::MAX"),
                    wallet_address: *wallet,
                    symbol: summary.symbol,
                    side: format!("{:?}", summary.side),
                    price: summary.price,
                    size: summary.original_size,
                    // EngineOrderIndex currently tracks only resting open orders,
                    // which are normalized to GTC semantics in this read model.
                    tif: "gtc".to_string(),
                    status,
                    created_at: summary.created_at,
                    updated_at: None,
                    filled_size: Some(filled_size),
                    mmp_enabled: summary.mmp_enabled,
                }
            })
            .collect()
    }
}

/// Mock QuoteProvider for tests — returns configurable quotes.
#[cfg(any(test, feature = "test-utils"))]
pub struct MockQuoteProvider {
    pub quotes: std::sync::RwLock<HashMap<String, SnapshotBookQuote>>,
    pub l2_seq: std::sync::atomic::AtomicI64,
    pub created_at: Instant,
}

#[cfg(any(test, feature = "test-utils"))]
impl Default for MockQuoteProvider {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(any(test, feature = "test-utils"))]
impl MockQuoteProvider {
    pub fn new() -> Self {
        Self {
            quotes: std::sync::RwLock::new(HashMap::new()),
            l2_seq: std::sync::atomic::AtomicI64::new(0),
            created_at: Instant::now(),
        }
    }

    pub fn set_quote(&self, symbol: &str, quote: SnapshotBookQuote) {
        self.quotes
            .write()
            .unwrap()
            .insert(symbol.to_string(), quote);
    }

    pub fn set_l2_seq(&self, seq: i64) {
        self.l2_seq.store(seq, std::sync::atomic::Ordering::SeqCst);
    }
}

#[cfg(any(test, feature = "test-utils"))]
impl QuoteProvider for MockQuoteProvider {
    fn get_quote(&self, symbol: &str) -> Option<SnapshotBookQuote> {
        self.quotes.read().unwrap().get(symbol).cloned()
    }

    fn get_quote_with_seq(&self, symbol: &str) -> (Option<SnapshotBookQuote>, i64) {
        let quotes = self.quotes.read().unwrap();
        let seq = self.l2_seq.load(std::sync::atomic::Ordering::SeqCst);
        (quotes.get(symbol).cloned(), seq)
    }

    fn book_snapshot_state(&self, symbol: &str) -> BookSnapshotState {
        let quotes = self.quotes.read().unwrap();
        let seq = self.l2_seq.load(std::sync::atomic::Ordering::SeqCst);
        if seq <= 0 {
            return BookSnapshotState::NotReady { l2_seq: seq };
        }

        BookSnapshotState::Ready {
            quote: quotes
                .get(symbol)
                .cloned()
                .unwrap_or_else(SnapshotBookQuote::empty),
            l2_seq: seq,
        }
    }

    fn all_quotes(&self) -> HashMap<String, SnapshotBookQuote> {
        self.quotes.read().unwrap().clone()
    }

    fn l2_seq(&self) -> i64 {
        self.l2_seq.load(std::sync::atomic::Ordering::SeqCst)
    }

    fn snapshot(&self) -> QuoteSnapshot {
        QuoteSnapshot {
            quotes: self.quotes.read().unwrap().clone(),
            l2_seq: self.l2_seq.load(std::sync::atomic::Ordering::SeqCst),
        }
    }

    fn staleness(&self) -> Duration {
        self.created_at.elapsed()
    }
}

#[cfg(any(test, feature = "test-utils"))]
impl EngineStateDigestProvider for MockQuoteProvider {
    fn engine_state_digest(&self) -> EngineStateDigest {
        EngineStateDigest::empty()
    }
}

/// Mock AgentAuthProvider for tests — self-signing only (no agents authorized).
#[cfg(any(test, feature = "test-utils"))]
pub struct MockAgentAuthProvider;

#[cfg(any(test, feature = "test-utils"))]
impl AgentAuthProvider for MockAgentAuthProvider {
    fn is_agent_authorized(&self, wallet: &WalletAddress, agent: &WalletAddress) -> bool {
        wallet == agent
    }

    fn get_authorized_agents(&self, _wallet: &WalletAddress) -> Vec<WalletAddress> {
        Vec::new()
    }
}

/// Mock OrderSnapshotProvider for tests — returns empty orders by default.
#[cfg(any(test, feature = "test-utils"))]
pub struct MockOrderSnapshotProvider;

#[cfg(any(test, feature = "test-utils"))]
impl OrderSnapshotProvider for MockOrderSnapshotProvider {
    fn get_open_orders_for_wallet(&self, _wallet: &WalletAddress) -> Vec<RuntimeOrderSummary> {
        Vec::new()
    }

    fn get_all_orders(&self) -> Vec<(RuntimeOrderSummary, WalletAddress)> {
        Vec::new()
    }
}

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

    #[test]
    fn empty_engine_state_digest_is_stable_and_non_ambiguous() {
        let a = EngineStateDigest::empty();
        let b = EngineStateDigest::empty();

        assert_eq!(a.overall_digest, b.overall_digest);
        assert_eq!(a.orders_digest, b.orders_digest);
        assert!(a.overall_digest.starts_with("0x"));
        assert_eq!(a.overall_digest.len(), 66);
        assert_eq!(a.orders_count, 0);
        assert_eq!(a.positions_count, 0);
        assert_eq!(a.cash_wallet_count, 0);
    }

    #[test]
    fn snapshot_quote_provider_exposes_latest_engine_state_digest() {
        let snapshot = Arc::new(ArcSwap::from_pointee(EngineSnapshot::empty()));
        let provider = SnapshotQuoteProvider::new(snapshot.clone());

        let mut next = EngineSnapshot::empty();
        next.engine_state_digest.next_order_id = 42;
        next.engine_state_digest.overall_digest =
            "0x1111111111111111111111111111111111111111111111111111111111111111".to_string();
        snapshot.store(Arc::new(next));

        let digest = provider.engine_state_digest();
        assert_eq!(digest.next_order_id, 42);
        assert_eq!(
            digest.overall_digest,
            "0x1111111111111111111111111111111111111111111111111111111111111111"
        );
    }

    #[test]
    fn snapshot_quote_provider_treats_published_seq_zero_as_ready_empty_book() {
        let snapshot = Arc::new(ArcSwap::from_pointee(EngineSnapshot::empty()));
        let provider = SnapshotQuoteProvider::new(snapshot.clone());

        assert!(matches!(
            provider.book_snapshot_state("SPCX-20260611-140-C"),
            BookSnapshotState::NotReady { l2_seq: 0 }
        ));

        snapshot.store(Arc::new(EngineSnapshot::build(
            &HashMap::new(),
            &EngineOrderIndex::new(),
            0,
        )));

        match provider.book_snapshot_state("SPCX-20260611-140-C") {
            BookSnapshotState::Ready { quote, l2_seq } => {
                assert_eq!(l2_seq, 0);
                assert!(quote.is_empty_book());
            }
            BookSnapshotState::NotReady { l2_seq } => {
                panic!("published seq-zero snapshot should be ready, got {l2_seq}")
            }
        }
    }

    #[test]
    fn volatility_surface_digest_tracks_pricing_values() {
        let mut surface = VolatilitySurface::new();
        surface.insert(100_000.0, 1_800_000_000, 0.65);
        surface.set_atm_vol(1_800_000_000, 0.62);
        surface.set_delta_iv(1_800_000_000, 0.25, 0.70);

        let base_digest = hex_digest(|h| update_volatility_surface(h, &surface));

        let mut changed_strike_point = surface.clone();
        changed_strike_point.insert(100_000.0, 1_800_000_000, 0.66);
        assert_ne!(
            base_digest,
            hex_digest(|h| update_volatility_surface(h, &changed_strike_point))
        );

        let mut changed_atm = surface.clone();
        changed_atm.set_atm_vol(1_800_000_000, 0.63);
        assert_ne!(
            base_digest,
            hex_digest(|h| update_volatility_surface(h, &changed_atm))
        );

        let mut changed_delta = surface;
        changed_delta.set_delta_iv(1_800_000_000, 0.25, 0.71);
        assert_ne!(
            base_digest,
            hex_digest(|h| update_volatility_surface(h, &changed_delta))
        );
    }
}