hypercall/read_cache/

portfolio.rs

use super::greeks::GreeksCache;
use super::tier::TierCache;
use hypercall_types::api_models::{
    MarginSummary, Portfolio, PortfolioGreeksAggregate, Position, PositionGreeksLeg,
    PositionWithMetrics, SpanMarginSummary,
};
use hypercall_types::portfolio_greeks::{
    build_net_position_quantities, calculate_portfolio_greeks,
};
use hypercall_types::position_metrics::{enrich_position_metrics, PositionMarginMetrics};

use crate::portfolio::canonical_perp_symbol;
use crate::portfolio::{
    HypercorePositionUpdate, PortfolioBalance, PortfolioChange, PortfolioService,
    PortfolioServiceImpl,
};
use crate::rsm::margin_service::{ExtendedRiskGrid, ScenarioPnl, SpanMarginService};
use crate::rsm::portfolio_margin::RiskAccountBuilder;
use crate::rsm::MarginMode;
use crate::shared::order_types::ParsedSymbol;
use crate::snapshot::{
    DbPortfolioSnapshotLoader, SnapshotLoader, Snapshotable, SyncState, SyncStatus,
};
use crate::standard_margin::{
    PositionMarginContribution, StandardAccountBuilder, StandardMarginService,
};
use crate::types::MarginDetails;
use anyhow::Result;
use futures::future::join_all;
use hypercall_db_diesel::DatabaseHandler;
use hypercall_runtime_api::OrderSnapshotProvider;
use hypercall_runtime_api::RuntimeOrderSummary;
use hypercall_types::ws_protocol::PortfolioUpdate;
use hypercall_types::{to_contract_units_decimal, to_human_readable_decimal, WalletAddress};
use hypercall_types::{EngineMessage, PositionExpiredMessage};
use rust_decimal::prelude::FromPrimitive;
use rust_decimal::Decimal;
use rust_decimal_macros::dec;
use std::collections::{BTreeSet, HashMap};
use std::str::FromStr;
use std::sync::Arc;
use tokio::sync::{mpsc, Mutex, OwnedMutexGuard, RwLock};

/// Notification work collected under the projection barrier and dispatched
/// after the barrier is released so that WebSocket sends don't block other
/// state mutations.
pub(crate) enum PendingNotifications {
    Fill { fill: hypercall_types::Fill },
    Changes(Vec<PortfolioChange>),
}

type SubscriberId = u64;
pub const ENGINE_COMMAND_SNAPSHOT_STREAM: &str = "engine_command";

#[derive(Debug, Clone)]
pub struct WalletMarginSnapshot {
    pub mode: MarginMode,
    pub span_margin: SpanMarginSummary,
    pub margin_summary: MarginSummary,
    pub total_margin_used: Decimal,
    pub available_balance: Decimal,
    /// Per-symbol IM/MM contributions for Standard mode.
    /// None for Portfolio (SPAN) mode because portfolio-level scenario risk
    /// is not additively attributable per position.
    pub standard_position_contributions: Option<HashMap<String, PositionMarginContribution>>,
    /// Per-symbol current option marks for Standard mode.
    /// None for Portfolio (SPAN) mode.
    pub standard_option_marks: Option<HashMap<String, Decimal>>,
}

#[derive(Debug, Clone)]
pub struct PmRiskGridData {
    pub margin_details: MarginDetails,
    pub position_details: MarginDetails,
    pub scenario_pnls: Vec<ScenarioPnl>,
    pub extended_grid: ExtendedRiskGrid,
}

pub struct PortfolioCache {
    service: Arc<PortfolioServiceImpl>,
    db: Arc<DatabaseHandler>,
    settlement: Arc<dyn hypercall_db::SettlementWriter>,
    last_processed_seq: Arc<RwLock<i64>>,
    projection_barrier: Arc<Mutex<()>>,
    subscribers: Arc<
        RwLock<
            HashMap<WalletAddress, HashMap<SubscriberId, mpsc::UnboundedSender<PortfolioUpdate>>>,
        >,
    >,
    next_subscriber_id: Arc<RwLock<SubscriberId>>,
    /// Optional SPAN margin service for real-time margin updates
    span_margin_service: Arc<RwLock<Option<Arc<SpanMarginService>>>>,
    /// Optional Greeks cache for spot price lookups
    greeks_cache: Arc<RwLock<Option<Arc<GreeksCache>>>>,
    /// Optional RiskAccountBuilder for computing open orders IM
    risk_account_builder: Arc<RwLock<Option<Arc<RiskAccountBuilder>>>>,
    /// Optional tier cache for margin mode lookups.
    tier_cache: Arc<RwLock<Option<Arc<TierCache>>>>,
    /// Optional order snapshot provider for open-order calculations.
    order_snapshot: Arc<RwLock<Option<Arc<dyn OrderSnapshotProvider>>>>,
    /// Optional standard margin service for Standard mode calculations.
    standard_margin_service: Arc<RwLock<Option<Arc<StandardMarginService>>>>,
    /// Optional standard account builder for Standard mode account assembly.
    standard_account_builder: Arc<RwLock<Option<Arc<StandardAccountBuilder>>>>,
    /// Portfolio sync status for readiness gating.
    /// API/engine should not report "ready" until this is Ready.
    sync_status: Arc<SyncStatus>,
}

impl PortfolioCache {
    pub fn new(db: Arc<DatabaseHandler>) -> Self {
        let settlement: Arc<dyn hypercall_db::SettlementWriter> = db.clone();
        Self {
            service: Arc::new(PortfolioServiceImpl::new()),
            settlement,
            db,
            last_processed_seq: Arc::new(RwLock::new(0)),
            projection_barrier: Arc::new(Mutex::new(())),
            subscribers: Arc::new(RwLock::new(HashMap::new())),
            next_subscriber_id: Arc::new(RwLock::new(0)),
            span_margin_service: Arc::new(RwLock::new(None)),
            greeks_cache: Arc::new(RwLock::new(None)),
            risk_account_builder: Arc::new(RwLock::new(None)),
            tier_cache: Arc::new(RwLock::new(None)),
            order_snapshot: Arc::new(RwLock::new(None)),
            standard_margin_service: Arc::new(RwLock::new(None)),
            standard_account_builder: Arc::new(RwLock::new(None)),
            sync_status: Arc::new(SyncStatus::new()),
        }
    }

    /// Get the sync status for readiness checks.
    ///
    /// API endpoints and WebSocket handlers should check this before
    /// serving requests that depend on portfolio state.
    pub fn sync_status(&self) -> Arc<SyncStatus> {
        self.sync_status.clone()
    }

    pub async fn lock_projection_barrier(&self) -> OwnedMutexGuard<()> {
        self.projection_barrier.clone().lock_owned().await
    }

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

    /// Get the current sync state.
    pub fn sync_state(&self) -> SyncState {
        self.sync_status.state()
    }

    /// Mark portfolio as catching up (called after snapshot restore).
    pub fn set_catching_up(&self) {
        self.sync_status.set_catching_up();
        tracing::info!("Portfolio sync status: CatchingUp");
    }

    /// Mark portfolio as ready (called after catchup replay completes).
    pub fn set_ready(&self) {
        self.sync_status.set_ready();
        tracing::info!("Portfolio sync status: Ready");
    }

    /// Set margin-related dependencies for shared REST/WS margin calculations.
    pub async fn set_margin_dependencies(
        &self,
        span_margin_service: Arc<SpanMarginService>,
        greeks_cache: Arc<GreeksCache>,
        risk_account_builder: Arc<RiskAccountBuilder>,
        tier_cache: Arc<TierCache>,
        order_snapshot: Arc<dyn OrderSnapshotProvider>,
        standard_margin_service: Arc<StandardMarginService>,
        standard_account_builder: Arc<StandardAccountBuilder>,
    ) {
        let mut svc = self.span_margin_service.write().await;
        *svc = Some(span_margin_service);
        drop(svc);

        let mut cache = self.greeks_cache.write().await;
        *cache = Some(greeks_cache);
        drop(cache);

        let mut builder = self.risk_account_builder.write().await;
        *builder = Some(risk_account_builder);
        drop(builder);

        let mut tier = self.tier_cache.write().await;
        *tier = Some(tier_cache);
        drop(tier);

        let mut orders = self.order_snapshot.write().await;
        *orders = Some(order_snapshot);
        drop(orders);

        let mut standard_svc = self.standard_margin_service.write().await;
        *standard_svc = Some(standard_margin_service);
        drop(standard_svc);

        let mut standard_builder = self.standard_account_builder.write().await;
        *standard_builder = Some(standard_account_builder);
    }

    pub async fn compute_wallet_margin_snapshot(
        &self,
        wallet: &WalletAddress,
    ) -> anyhow::Result<WalletMarginSnapshot> {
        let margin_mode = self.margin_mode_for_wallet(wallet).await?;
        self.compute_wallet_margin_snapshot_for_mode(wallet, margin_mode)
            .await
    }

    async fn compute_wallet_margin_snapshot_for_mode(
        &self,
        wallet: &WalletAddress,
        margin_mode: MarginMode,
    ) -> anyhow::Result<WalletMarginSnapshot> {
        match margin_mode {
            MarginMode::Portfolio => self.compute_portfolio_mode_snapshot(wallet).await,
            MarginMode::Standard => self.compute_standard_mode_snapshot(wallet).await,
        }
    }

    async fn compute_portfolio_mode_snapshot(
        &self,
        wallet: &WalletAddress,
    ) -> anyhow::Result<WalletMarginSnapshot> {
        self.prune_applied_settlement_ghost_positions(wallet)
            .await?;

        let margin_svc = self.span_margin_service.read().await;
        let risk_builder = self.risk_account_builder.read().await;

        let (margin_service, risk_account_builder) =
            match (margin_svc.as_ref(), risk_builder.as_ref()) {
                (Some(m), Some(r)) => (m.clone(), r.clone()),
                _ => {
                    return Err(anyhow::anyhow!(
                        "Portfolio margin dependencies are not fully configured"
                    ));
                }
            };
        drop(margin_svc);
        drop(risk_builder);

        let snapshot = risk_account_builder.build_snapshot(wallet).await?;
        let market_state =
            risk_account_builder.resolve_market_state(&snapshot, margin_service.config())?;
        let position_snapshot = snapshot.without_open_orders();

        let margin_details = margin_service
            .compute_margin_from_snapshot(&snapshot, market_state.clone())
            .map_err(|e| anyhow::anyhow!("SPAN margin failed for {}: {}", wallet, e))?;
        let position_details = margin_service
            .compute_margin_from_snapshot(&position_snapshot, market_state)
            .map_err(|e| {
                anyhow::anyhow!("Position-only SPAN margin failed for {}: {}", wallet, e)
            })?;

        let position_im = position_details.initial_margin_required;
        let open_orders_im = (margin_details.initial_margin_required
            - position_details.initial_margin_required)
            .max(Decimal::ZERO);
        let position_mm = position_details.maintenance_margin_required;
        let total_margin_used = position_im + open_orders_im;
        let available_balance = (margin_details.equity - total_margin_used).max(dec!(0));

        let span_margin = SpanMarginSummary {
            equity: margin_details.equity,
            initial_margin_required: position_im,
            maintenance_margin_required: position_mm,
            open_orders_initial_margin: open_orders_im,
            option_margin_required: position_details.scanning_risk,
            scanning_risk: position_details.scanning_risk,
            option_floor: position_details.option_floor,
            gamma_overlay: position_details.gamma_overlay,
            hypercore_margin_required: dec!(0),
        };

        let margin_summary = MarginSummary {
            mode: MarginMode::Portfolio.as_str().to_string(),
            equity: margin_details.equity,
            position_im,
            open_orders_im,
            initial_margin: margin_details.equity - position_im - open_orders_im,
            maintenance_margin: margin_details.equity - position_mm,
            open_orders_premium_reserved: None,
        };

        Ok(WalletMarginSnapshot {
            mode: MarginMode::Portfolio,
            span_margin,
            margin_summary,
            total_margin_used,
            available_balance,
            standard_position_contributions: None,
            standard_option_marks: None,
        })
    }

    pub async fn compute_pm_risk_grid_data(
        &self,
        wallet: &WalletAddress,
    ) -> anyhow::Result<PmRiskGridData> {
        let margin_mode = self.margin_mode_for_wallet(wallet).await?;
        if matches!(margin_mode, MarginMode::Standard) {
            return Err(anyhow::anyhow!(
                "compute_pm_risk_grid_data called for Standard-mode wallet {}",
                wallet
            ));
        }

        self.prune_applied_settlement_ghost_positions(wallet)
            .await?;

        let margin_svc = self.span_margin_service.read().await;
        let risk_builder = self.risk_account_builder.read().await;

        let (margin_service, risk_account_builder) =
            match (margin_svc.as_ref(), risk_builder.as_ref()) {
                (Some(m), Some(r)) => (m.clone(), r.clone()),
                _ => {
                    return Err(anyhow::anyhow!(
                        "Portfolio margin dependencies are not fully configured"
                    ));
                }
            };
        drop(margin_svc);
        drop(risk_builder);

        let snapshot = risk_account_builder.build_snapshot(wallet).await?;
        let market_state =
            risk_account_builder.resolve_market_state(&snapshot, margin_service.config())?;
        let position_snapshot = snapshot.without_open_orders();

        let margin_details = margin_service
            .compute_margin_from_snapshot(&snapshot, market_state.clone())
            .map_err(|e| anyhow::anyhow!("SPAN margin failed: {}", e))?;
        let position_details = margin_service
            .compute_margin_from_snapshot(&position_snapshot, market_state.clone())
            .map_err(|e| anyhow::anyhow!("Position-only SPAN margin failed: {}", e))?;
        let scenario_pnls = margin_service
            .compute_risk_grid_from_snapshot(&snapshot, market_state.clone())
            .map_err(|e| anyhow::anyhow!("Risk grid failed: {}", e))?;
        let extended_grid = margin_service
            .compute_extended_risk_grid_from_snapshot(&snapshot, market_state)
            .map_err(|e| anyhow::anyhow!("Extended risk grid failed: {}", e))?;

        Ok(PmRiskGridData {
            margin_details,
            position_details,
            scenario_pnls,
            extended_grid,
        })
    }

    async fn compute_standard_mode_snapshot(
        &self,
        wallet: &WalletAddress,
    ) -> anyhow::Result<WalletMarginSnapshot> {
        let standard_svc_guard = self.standard_margin_service.read().await;
        let standard_builder_guard = self.standard_account_builder.read().await;
        let order_snapshot_guard = self.order_snapshot.read().await;
        let greeks_cache_guard = self.greeks_cache.read().await;

        let (standard_margin_service, standard_account_builder, order_snapshot, greeks_cache) =
            match (
                standard_svc_guard.as_ref(),
                standard_builder_guard.as_ref(),
                order_snapshot_guard.as_ref(),
                greeks_cache_guard.as_ref(),
            ) {
                (Some(svc), Some(builder), Some(snapshot), Some(greeks)) => (
                    svc.clone(),
                    builder.clone(),
                    snapshot.clone(),
                    greeks.clone(),
                ),
                _ => {
                    return Err(anyhow::anyhow!(
                        "Standard margin dependencies are not fully configured"
                    ));
                }
            };
        drop(standard_svc_guard);
        drop(standard_builder_guard);
        drop(order_snapshot_guard);
        drop(greeks_cache_guard);

        // Refresh UPNL before building the account — compute_standard_mode_snapshot
        // is called independently of get_portfolio, so positions may have stale UPNL.
        if let Err(e) = self.refresh_live_market_prices_for_wallet(wallet).await {
            tracing::warn!(
                "Standard margin snapshot: could not refresh prices for {}: {}",
                wallet,
                e
            );
        }

        let standard_account = standard_account_builder.build(wallet).await?;
        let standard_position_contributions =
            standard_margin_service.compute_position_margin_breakdown(&standard_account);
        let standard_option_marks: HashMap<String, Decimal> = standard_account
            .option_positions
            .iter()
            .map(|p| (p.symbol.clone(), p.mark_price))
            .collect();
        let cash = standard_account.usdc_balance;
        let base_result = standard_margin_service.compute_margin(&standard_account);

        let mut hypothetical_account = standard_account.clone();
        let open_sell_positions =
            Self::snapshot_open_sell_positions(wallet, order_snapshot.as_ref())?;
        let missing_open_sell_spot_underlyings: BTreeSet<String> = open_sell_positions
            .iter()
            .filter(|pos| pos.position.spot_price == dec!(0))
            .map(|pos| pos.position.underlying.clone())
            .collect();
        let mut missing_open_sell_spot_prices = HashMap::new();
        if !missing_open_sell_spot_underlyings.is_empty() {
            let fetches =
                missing_open_sell_spot_underlyings
                    .into_iter()
                    .map(|underlying| {
                        let greeks_cache = greeks_cache.clone();
                        async move {
                            let spot = greeks_cache.get_spot_price(&underlying).await.ok_or_else(
                                || {
                                    anyhow::anyhow!(
                                        "Missing spot price for {} in standard margin calculation",
                                        underlying
                                    )
                                },
                            )?;
                            let spot_decimal = Decimal::from_f64(spot).ok_or_else(|| {
                                anyhow::anyhow!("Invalid spot price conversion for {}", underlying)
                            })?;
                            Ok::<(String, Decimal), anyhow::Error>((underlying, spot_decimal))
                        }
                    });

            for result in join_all(fetches).await {
                let (underlying, spot_decimal) = result?;
                missing_open_sell_spot_prices.insert(underlying, spot_decimal);
            }
        }

        for mut pos in open_sell_positions {
            if pos.position.spot_price == dec!(0) {
                pos.position.spot_price = *missing_open_sell_spot_prices
                    .get(&pos.position.underlying)
                    .ok_or_else(|| {
                        anyhow::anyhow!(
                            "Missing prefetched spot price for {} in standard margin calculation",
                            pos.position.underlying
                        )
                    })?;
            }
            hypothetical_account.option_positions.push(pos.position);
        }

        let hypothetical_result = standard_margin_service.compute_margin(&hypothetical_account);
        let open_orders_im =
            (hypothetical_result.position_im - base_result.position_im).max(dec!(0));
        let premium_reserved = Self::snapshot_open_buy_premium(wallet, order_snapshot.as_ref())?;
        let total_margin_used = base_result.position_im + open_orders_im;
        let available_balance = (cash - total_margin_used - premium_reserved).max(dec!(0));

        let span_margin = SpanMarginSummary {
            equity: base_result.equity,
            initial_margin_required: base_result.position_im,
            maintenance_margin_required: base_result.position_mm,
            open_orders_initial_margin: open_orders_im,
            option_margin_required: base_result.position_im,
            scanning_risk: dec!(0),
            option_floor: dec!(0),
            gamma_overlay: dec!(0),
            hypercore_margin_required: dec!(0),
        };

        let margin_summary = MarginSummary {
            mode: MarginMode::Standard.as_str().to_string(),
            equity: base_result.equity,
            position_im: base_result.position_im,
            open_orders_im,
            initial_margin: base_result.equity
                - base_result.position_im
                - open_orders_im
                - premium_reserved,
            maintenance_margin: base_result.equity - base_result.position_mm,
            open_orders_premium_reserved: Some(premium_reserved),
        };

        Ok(WalletMarginSnapshot {
            mode: MarginMode::Standard,
            span_margin,
            margin_summary,
            total_margin_used,
            available_balance,
            standard_position_contributions: Some(standard_position_contributions),
            standard_option_marks: Some(standard_option_marks),
        })
    }

    fn snapshot_open_sell_positions(
        wallet: &WalletAddress,
        order_snapshot: &dyn OrderSnapshotProvider,
    ) -> anyhow::Result<Vec<hypercall_engine::order_index::OpenSellPositionInfo>> {
        let mut positions = Vec::new();
        for order in order_snapshot.get_open_orders_for_wallet(wallet) {
            if !matches!(order.side, hypercall_types::Side::Sell) {
                continue;
            }
            let parsed = ParsedSymbol::from_symbol(&order.symbol).map_err(|e| {
                anyhow::anyhow!(
                    "Invalid option symbol '{}' in open sell order {} for {}: {}",
                    order.symbol,
                    order.order_id,
                    wallet,
                    e
                )
            })?;
            if order.remaining_size <= dec!(0) {
                continue;
            }

            let premium = order.price * order.remaining_size;
            let expiry_date_str = format!("{}", parsed.expiry);
            let expiry_code = expiry_date_str.parse::<u64>().map_err(|e| {
                anyhow::anyhow!(
                    "Invalid expiry '{}' in open sell order {} for {} ({}): {}",
                    expiry_date_str,
                    order.order_id,
                    wallet,
                    order.symbol,
                    e
                )
            })?;
            let expiry_ts =
                hypercall_types::expiry_date_to_timestamp_checked(&parsed.underlying, expiry_code)
                    .map_err(|e| {
                    anyhow::anyhow!(
                        "Invalid expiry timestamp in open sell order {} for {} ({}): {}",
                        order.order_id,
                        wallet,
                        order.symbol,
                        e
                    )
                })? as i64;

            let position = hypercall_margin::OptionPosition {
                symbol: order.symbol.clone(),
                underlying: parsed.underlying.clone(),
                expiry_ts,
                strike: parsed.strike,
                is_call: matches!(parsed.option_type, crate::types::OptionType::Call),
                size: -order.remaining_size,
                mark_price: order.price,
                entry_price: order.price,
                spot_price: dec!(0),
            };

            positions
                .push(hypercall_engine::order_index::OpenSellPositionInfo { premium, position });
        }

        Ok(positions)
    }

    fn snapshot_open_buy_premium(
        wallet: &WalletAddress,
        order_snapshot: &dyn OrderSnapshotProvider,
    ) -> anyhow::Result<Decimal> {
        order_snapshot
            .get_open_orders_for_wallet(wallet)
            .into_iter()
            .filter(|order: &RuntimeOrderSummary| matches!(order.side, hypercall_types::Side::Buy))
            .filter(|order| order.remaining_size > dec!(0))
            .try_fold(dec!(0), |acc, order| {
                ParsedSymbol::from_symbol(&order.symbol).map_err(|e| {
                    anyhow::anyhow!(
                        "Invalid option symbol '{}' in open buy order {} for {}: {}",
                        order.symbol,
                        order.order_id,
                        wallet,
                        e
                    )
                })?;
                Ok(acc + (order.price * order.remaining_size))
            })
    }

    /// Compute and publish margin update for a wallet to all subscribers.
    /// This is called after fills to provide real-time margin updates.
    pub async fn publish_margin_update(&self, wallet: &WalletAddress) {
        if let Err(error) = self.try_publish_margin_update(wallet).await {
            tracing::error!("Failed to publish margin update for {}: {}", wallet, error);
        }
    }

    async fn try_publish_margin_update(&self, wallet: &WalletAddress) -> anyhow::Result<()> {
        let margin_mode = self.margin_mode_for_wallet(wallet).await?;

        // Refresh market prices before computing the margin snapshot so equity
        // and margin values reflect the current mark, not stale in-memory UPNL.
        match margin_mode {
            MarginMode::Portfolio => {
                self.refresh_live_market_prices_for_wallet(wallet)
                    .await
                    .map_err(|error| {
                        anyhow::anyhow!(
                            "PM margin update unavailable due to strict repricing failure: {}",
                            error
                        )
                    })?;
            }
            MarginMode::Standard => {
                if let Err(error) = self.refresh_live_market_prices_for_wallet(wallet).await {
                    tracing::warn!(
                        "Skipping live repricing before standard margin update for {}: {}",
                        wallet,
                        error
                    );
                }
            }
        }

        let snapshot = self
            .compute_wallet_margin_snapshot_for_mode(wallet, margin_mode)
            .await
            .map_err(|error| {
                anyhow::anyhow!(
                    "margin snapshot calculation failed for {}: {}",
                    wallet,
                    error
                )
            })?;

        // Notify subscribers
        self.notify_subscribers(
            wallet,
            PortfolioUpdate::MarginUpdate {
                span_margin: snapshot.span_margin,
                total_margin_used: snapshot.total_margin_used,
                available_balance: snapshot.available_balance,
                timestamp: chrono::Utc::now().timestamp(),
            },
        )
        .await;

        Ok(())
    }

    async fn prune_applied_settlement_ghost_positions(
        &self,
        wallet: &WalletAddress,
    ) -> anyhow::Result<usize> {
        let option_symbols = match self.service.get_portfolio_balance(wallet).await {
            Some(balance) => balance
                .positions
                .keys()
                .filter(|symbol| ParsedSymbol::from_symbol(symbol).is_ok())
                .cloned()
                .collect::<Vec<_>>(),
            None => return Ok(0),
        };

        if option_symbols.is_empty() {
            return Ok(0);
        }

        let settlement = self.settlement.clone();
        let wallet_owned = *wallet;
        let settled_symbols = tokio::task::spawn_blocking(move || {
            settlement.get_applied_settlement_symbols_sync(&wallet_owned, &option_symbols)
        })
        .await
        .map_err(|e| anyhow::anyhow!("spawn_blocking join error: {}", e))?
        .map_err(|error| {
            anyhow::anyhow!(
                "Failed to query applied settlements for {}: {}",
                wallet,
                error
            )
        })?;

        for symbol in &settled_symbols {
            tracing::warn!(
                wallet = %wallet,
                symbol,
                "Removing settled ghost position from portfolio projection"
            );
            metrics::counter!("ht_portfolio_ghost_positions_repaired_total").increment(1);
            self.service.remove_expired_position(wallet, symbol).await;
        }

        Ok(settled_symbols.len())
    }

    /// Compute and publish margin updates for all wallets with active portfolio subscribers.
    ///
    /// Returns the number of unique wallets attempted.
    pub async fn publish_margin_updates_for_subscribers(&self) -> usize {
        let wallets: Vec<WalletAddress> = {
            let subscribers = self.subscribers.read().await;
            subscribers.keys().cloned().collect()
        };

        for wallet in &wallets {
            self.publish_margin_update(wallet).await;
        }

        wallets.len()
    }

    async fn has_portfolio_subscribers(&self, wallet: &WalletAddress) -> bool {
        let subscribers = self.subscribers.read().await;
        subscribers
            .get(wallet)
            .map(|account_subs| !account_subs.is_empty())
            .unwrap_or(false)
    }

    async fn compute_wallet_live_greeks_snapshot(
        &self,
        wallet: &WalletAddress,
    ) -> anyhow::Result<(Vec<PositionGreeksLeg>, Option<PortfolioGreeksAggregate>)> {
        let greeks_cache = self
            .greeks_cache
            .read()
            .await
            .as_ref()
            .cloned()
            .ok_or_else(|| anyhow::anyhow!("GreeksCache not configured"))?;

        let live_portfolio = self.service.get_portfolio(wallet).await;
        let live_positions = live_portfolio
            .positions
            .into_iter()
            .map(|p| (p.position.symbol, p.position.amount));
        let net_quantities = build_net_position_quantities(live_positions, &[])
            .map_err(|e| anyhow::anyhow!("Invalid position state for greeks update: {}", e))?;

        if net_quantities.is_empty() {
            return Ok((Vec::new(), None));
        }

        let mut contract_greeks = HashMap::with_capacity(net_quantities.len());
        for symbol in net_quantities.keys() {
            let greeks = greeks_cache.get_greeks(symbol).await.map_err(|e| {
                anyhow::anyhow!(
                    "Failed to fetch greeks for symbol {} while publishing WS update: {}",
                    symbol,
                    e
                )
            })?;
            contract_greeks.insert(symbol.clone(), greeks);
        }

        let response = calculate_portfolio_greeks(*wallet, &net_quantities, &contract_greeks)
            .map_err(|e| anyhow::anyhow!("Failed to compute portfolio greeks: {}", e))?;

        Ok((response.per_leg, response.aggregate))
    }

    /// Compute and publish a portfolio Greeks update for a wallet.
    pub async fn publish_greeks_update(&self, wallet: &WalletAddress) {
        if !self.has_portfolio_subscribers(wallet).await {
            return;
        }

        let (per_leg, aggregate) = match self.compute_wallet_live_greeks_snapshot(wallet).await {
            Ok(snapshot) => snapshot,
            Err(e) => {
                tracing::debug!(
                    "Skipping greeks update publish for {} due to calculation error: {}",
                    wallet,
                    e
                );
                return;
            }
        };

        self.notify_subscribers(
            wallet,
            PortfolioUpdate::GreeksUpdate {
                per_leg,
                aggregate,
                timestamp: chrono::Utc::now().timestamp(),
            },
        )
        .await;
    }

    /// Compute and publish greeks updates for all wallets with active portfolio subscribers.
    ///
    /// Returns the number of unique wallets attempted.
    pub async fn publish_greeks_updates_for_subscribers(&self) -> usize {
        let wallets: Vec<WalletAddress> = {
            let subscribers = self.subscribers.read().await;
            subscribers.keys().cloned().collect()
        };

        for wallet in &wallets {
            self.publish_greeks_update(wallet).await;
        }

        wallets.len()
    }

    /// Get the underlying PortfolioService for wiring into other components.
    ///
    /// This returns the same service instance that PortfolioCache uses,
    /// allowing UnifiedEngine to read executed state from the canonical source.
    pub fn get_service(&self) -> Arc<PortfolioServiceImpl> {
        self.service.clone()
    }

    /// Per-position repricing: update each position's market price individually.
    ///
    /// Reprice all wallets that have positions. Called once at startup after all
    /// dependencies (greeks cache, oracles) are wired, to fix stale UPNL=0 from
    /// journal replay.
    pub async fn reprice_all_wallets(&self) -> usize {
        let wallets: Vec<WalletAddress> = {
            let all = self.service.all_portfolios().await;
            all.into_iter()
                .filter(|(_, balance)| !balance.positions.is_empty())
                .map(|(wallet, _)| wallet)
                .collect()
        };
        let total = wallets.len();
        let mut repriced = 0;
        for wallet in &wallets {
            match self.refresh_live_market_prices_for_wallet(wallet).await {
                Ok(n) if n > 0 => {
                    repriced += 1;
                    tracing::debug!("Repriced {} symbols for wallet {}", n, wallet);
                }
                Ok(_) => {}
                Err(e) => {
                    tracing::debug!("Could not reprice wallet {}: {}", wallet, e);
                }
            }
        }
        tracing::info!(
            "Startup repricing: {}/{} wallets with positions repriced",
            repriced,
            total
        );
        repriced
    }

    /// Unlike the old batch approach (`refresh_wallet_theoretical_option_upnl`), this
    /// continues on individual failures so one instrument's IV failure doesn't poison
    /// the entire wallet's UPNL.
    async fn refresh_live_market_prices_for_wallet(
        &self,
        wallet: &WalletAddress,
    ) -> anyhow::Result<usize> {
        let _ = self
            .prune_applied_settlement_ghost_positions(wallet)
            .await?;

        let portfolio_balance = match self.service.get_portfolio_balance(wallet).await {
            Some(portfolio_balance) => portfolio_balance,
            None => return Ok(0),
        };
        if portfolio_balance.positions.is_empty() {
            return Ok(0);
        }

        let greeks_cache = match self.greeks_cache.read().await.as_ref().cloned() {
            Some(greeks_cache) => greeks_cache,
            None => {
                return Err(anyhow::anyhow!(
                    "Greeks cache unavailable for portfolio repricing"
                ))
            }
        };

        let wallet_for_errors = wallet.to_string();
        let mut symbols: Vec<String> = portfolio_balance
            .positions
            .keys()
            .filter(|s| portfolio_perp_underlying(s).is_none())
            .cloned()
            .collect();
        symbols.sort();
        let fetches = symbols.into_iter().map(|symbol| {
            let greeks_cache = greeks_cache.clone();
            let wallet_for_errors = wallet_for_errors.clone();
            async move {
                let market_price = if ParsedSymbol::from_symbol(&symbol).is_ok() {
                    greeks_cache
                        .get_theoretical_price(&symbol)
                        .await
                        .map_err(|error| {
                            anyhow::anyhow!(
                                "Portfolio repricing unavailable for {} on {}: {}",
                                symbol,
                                wallet_for_errors,
                                error
                            )
                        })?
                } else {
                    return Err(anyhow::anyhow!(
                        "Portfolio repricing unavailable for {} on {}: unsupported symbol",
                        symbol,
                        wallet_for_errors
                    ));
                };

                let market_price_decimal =
                    Decimal::from_f64_retain(market_price).ok_or_else(|| {
                        anyhow::anyhow!(
                            "Invalid repriced market price for {} on {}: {}",
                            symbol,
                            wallet_for_errors,
                            market_price
                        )
                    })?;
                Ok::<(String, Decimal), anyhow::Error>((symbol, market_price_decimal))
            }
        });

        let mut market_prices = HashMap::new();
        for result in join_all(fetches).await {
            match result {
                Ok((symbol, market_price_decimal)) => {
                    market_prices.insert(symbol, market_price_decimal);
                }
                Err(e) => {
                    tracing::warn!("refresh_live_market_prices: wallet={}, error={}", wallet, e);
                    return Err(e);
                }
            }
        }

        if market_prices.is_empty() {
            return Ok(0);
        }

        let updated_symbols = market_prices.len();
        self.service.update_market_prices(market_prices).await;
        Ok(updated_symbols)
    }

    async fn margin_mode_for_wallet(&self, wallet: &WalletAddress) -> anyhow::Result<MarginMode> {
        let tier_cache = self
            .tier_cache
            .read()
            .await
            .as_ref()
            .cloned()
            .ok_or_else(|| anyhow::anyhow!("TierCache not configured"))?;
        tier_cache.get_margin_mode(wallet).await
    }

    /// Publish repriced positions for a subscribed wallet.
    pub async fn publish_position_updates(&self, wallet: &WalletAddress) {
        if !self.has_portfolio_subscribers(wallet).await {
            return;
        }

        let positions = match self.build_enriched_positions(wallet).await {
            Ok(positions) => positions,
            Err(error) => {
                tracing::debug!(
                    "Skipping position update publish for {} due to calculation error: {}",
                    wallet,
                    error
                );
                return;
            }
        };

        let timestamp = chrono::Utc::now().timestamp();
        for position in positions {
            self.notify_subscribers(
                wallet,
                PortfolioUpdate::PositionUpdate {
                    position,
                    timestamp,
                },
            )
            .await;
        }
    }

    /// Publish repriced position updates for every wallet with active subscribers.
    pub async fn publish_position_updates_for_subscribers(&self) -> usize {
        let wallets: Vec<WalletAddress> = {
            let subscribers = self.subscribers.read().await;
            subscribers.keys().cloned().collect()
        };

        for wallet in &wallets {
            self.publish_position_updates(wallet).await;
        }

        wallets.len()
    }

    async fn build_enriched_positions(
        &self,
        wallet: &WalletAddress,
    ) -> anyhow::Result<Vec<PositionWithMetrics>> {
        // Use per-position repricing (graceful degradation) instead of batch
        if let Err(error) = self.refresh_live_market_prices_for_wallet(wallet).await {
            tracing::warn!(
                "Failed to refresh live market prices for {} while building positions: {}",
                wallet,
                error
            );
        }
        let mut portfolio = self.service.get_portfolio(wallet).await;
        let margin_snapshot = match self.compute_wallet_margin_snapshot(wallet).await {
            Ok(snapshot) => snapshot,
            Err(e) => {
                tracing::warn!(
                    "Margin snapshot unavailable for {} while building WS positions: {}",
                    wallet,
                    e
                );
                return Ok(portfolio.positions);
            }
        };
        let mode = margin_snapshot.mode;
        portfolio.margin_mode = mode.as_str().to_string();
        portfolio.margin_summary = Some(margin_snapshot.margin_summary);
        portfolio.span_margin = Some(margin_snapshot.span_margin);
        portfolio.available_balance = margin_snapshot.available_balance;
        portfolio.total_margin_used = margin_snapshot.total_margin_used;
        enrich_position_metrics(
            mode,
            margin_snapshot
                .standard_position_contributions
                .map(|contributions| {
                    contributions
                        .into_iter()
                        .map(|(symbol, contribution)| {
                            (
                                symbol,
                                PositionMarginMetrics {
                                    initial_margin: contribution.initial_margin,
                                    maintenance_margin: contribution.maintenance_margin,
                                },
                            )
                        })
                        .collect()
                }),
            margin_snapshot.standard_option_marks,
            &mut portfolio,
        )
        .map_err(|status| {
            anyhow::anyhow!(
                "Failed to enrich position metrics for {}: status={}",
                wallet,
                status
            )
        })?;
        Ok(portfolio.positions)
    }

    async fn build_enriched_position_update(
        &self,
        wallet: &WalletAddress,
        symbol: &str,
    ) -> anyhow::Result<PositionWithMetrics> {
        let positions = self.build_enriched_positions(wallet).await?;
        if let Some(position) = positions.into_iter().find(|p| p.position.symbol == symbol) {
            return Ok(position);
        }

        Ok(PositionWithMetrics {
            position: Position {
                wallet_address: *wallet,
                symbol: symbol.to_string(),
                amount: dec!(0),
                entry_price: dec!(0),
                margin_posted: dec!(0),
                realized_pnl: dec!(0),
                unrealized_pnl: dec!(0),
                updated_at: chrono::Utc::now(),
            },
            notional_value: dec!(0),
            maintenance_margin: dec!(0),
            liquidation_price: dec!(0),
            margin_ratio: dec!(0),
        })
    }

    /// Initialize portfolios from the latest snapshot and return the next
    /// `engine_commands.command_id` that must be replayed into memory.
    pub async fn initialize(&self) -> Result<i64> {
        let loader = DbPortfolioSnapshotLoader::new(self.db.clone());

        match loader.load_latest() {
            Ok(Some((snapshot_id, state))) => {
                if let Some(next_command_id) = state
                    .offsets
                    .get(ENGINE_COMMAND_SNAPSHOT_STREAM)
                    .and_then(|partitions| partitions.get(&0))
                    .copied()
                {
                    self.service
                        .clear_all()
                        .await
                        .map_err(|e| anyhow::anyhow!("Failed clearing portfolio state: {}", e))?;
                    for (wallet, balance) in state.states {
                        self.service.restore(&wallet, balance).await.map_err(|e| {
                            anyhow::anyhow!(
                                "Failed restoring portfolio state for {}: {}",
                                wallet,
                                e
                            )
                        })?;
                    }

                    tracing::info!(
                        "Restored portfolio from snapshot id={} with journal replay boundary command_id={}",
                        snapshot_id,
                        next_command_id
                    );
                    self.sync_status.set_catching_up();
                    tracing::info!(
                        "Portfolio sync status: CatchingUp (snapshot restored, awaiting journal replay)"
                    );
                    Ok(next_command_id)
                } else {
                    tracing::warn!(
                        "Ignoring legacy portfolio snapshot id={} without {} boundary, rebuilding from journal",
                        snapshot_id,
                        ENGINE_COMMAND_SNAPSHOT_STREAM
                    );
                    self.service.clear_all().await.map_err(|e| {
                        anyhow::anyhow!("Failed clearing legacy portfolio state: {}", e)
                    })?;
                    self.sync_status.set_catching_up();
                    Ok(1)
                }
            }
            Ok(None) => {
                tracing::info!("No portfolio snapshots found, rebuilding from journal");
                self.service
                    .clear_all()
                    .await
                    .map_err(|e| anyhow::anyhow!("Failed clearing portfolio state: {}", e))?;
                self.sync_status.set_catching_up();
                tracing::info!(
                    "Portfolio sync status: CatchingUp (no snapshot, replaying journal from genesis)"
                );
                Ok(1)
            }
            Err(e) => {
                tracing::error!("Failed to restore portfolio from snapshot: {}", e);
                Err(anyhow::anyhow!("Snapshot restore failed: {}", e))
            }
        }
    }

    /// Handle hypercore position updates from the HypercorePositionService.
    ///
    /// Routes to set_hypercore_position (for snapshots) or apply_hypercore_position_update (for diffs)
    /// based on the snapshot flag.
    pub async fn handle_hypercore_position_update(&self, update: HypercorePositionUpdate) {
        if update.snapshot {
            self.service.set_hypercore_position(&update).await;
        } else {
            self.service.apply_hypercore_position_update(&update).await;
        }

        // Optionally notify WebSocket subscribers of position update
        let wallet_normalized = update.account.to_lowercase();
        let timestamp = chrono::Utc::now().timestamp();
        let position_symbol = canonical_perp_symbol(&update.coin);

        if let Ok(wallet) = WalletAddress::from_str(&wallet_normalized) {
            if !self.has_portfolio_subscribers(&wallet).await {
                return;
            }

            let position = match self
                .build_enriched_position_update(&wallet, &position_symbol)
                .await
            {
                Ok(position) => position,
                Err(e) => {
                    tracing::error!(
                        "Failed to build hypercore position update for {}/{}: {}",
                        wallet,
                        update.coin,
                        e
                    );
                    return;
                }
            };
            self.notify_subscribers(
                &wallet,
                PortfolioUpdate::PositionUpdate {
                    position,
                    timestamp,
                },
            )
            .await;
            self.publish_margin_update(&wallet).await;
            self.publish_greeks_update(&wallet).await;
        }
    }

    /// Handle incoming engine messages to update portfolio state.
    ///
    /// Fill handling is memory-only. Durable `fills`, `trades`, and fill-driven
    /// ledger rows are owned by the journal persistence path.
    pub async fn handle_engine_message(&self, message: EngineMessage, seq: i64) {
        // Acquire barrier for state mutation only; release before notifications.
        let pending_notifications = {
            let _guard = self.lock_projection_barrier().await;
            self.handle_engine_message_under_barrier(message, seq).await
        };
        // Barrier is released here. Send notifications without holding it.
        self.send_pending_notifications(pending_notifications).await;
    }

    pub(crate) async fn handle_engine_message_under_barrier(
        &self,
        message: EngineMessage,
        seq: i64,
    ) -> PendingNotifications {
        match &message {
            EngineMessage::OrderFilled { fill, .. } => {
                self.apply_fill_state_mutation(fill, seq).await;
                PendingNotifications::Fill { fill: fill.clone() }
            }
            EngineMessage::PositionExpired(ref expiry_msg) => {
                let change = self
                    .apply_position_expired_state_mutation(expiry_msg, Some(seq))
                    .await;
                PendingNotifications::Changes(vec![change])
            }
            _ => {
                let changes = match self.service.apply_event(&message).await {
                    Ok(c) => c,
                    Err(e) => {
                        panic!("CRITICAL: Portfolio event processing failed: {}", e);
                    }
                };

                let mut last_seq = self.last_processed_seq.write().await;
                *last_seq = seq;
                drop(last_seq);

                PendingNotifications::Changes(changes)
            }
        }
    }

    pub(crate) async fn handle_order_filled_under_barrier(
        &self,
        fill: &hypercall_types::Fill,
        seq: i64,
    ) {
        self.apply_fill_state_mutation(fill, seq).await;
    }

    pub(crate) async fn handle_position_expired_under_barrier(
        &self,
        expiry_msg: &PositionExpiredMessage,
        seq: Option<i64>,
    ) -> PendingNotifications {
        let change = self
            .apply_position_expired_state_mutation(expiry_msg, seq)
            .await;
        PendingNotifications::Changes(vec![change])
    }

    pub(crate) async fn handle_replayed_position_expired_projection_under_barrier(
        &self,
        expiry_msg: &PositionExpiredMessage,
    ) -> PendingNotifications {
        self.service
            .remove_expired_position(&expiry_msg.wallet_address, &expiry_msg.symbol)
            .await;

        tracing::info!(
            "PortfolioCache: Applied replayed expiry projection for {}/{}",
            expiry_msg.wallet_address,
            expiry_msg.symbol,
        );

        let (position_change, total_margin_used) = self
            .get_position_and_margin_for_notification(
                &expiry_msg.wallet_address,
                &expiry_msg.symbol,
            )
            .await;

        PendingNotifications::Changes(vec![PortfolioChange {
            wallet: expiry_msg.wallet_address,
            position_changes: vec![position_change],
            balance_change: None,
            total_margin_used,
        }])
    }

    pub(crate) async fn handle_option_custody_delta(
        &self,
        wallet: WalletAddress,
        symbol: String,
        quantity_delta: Decimal,
        seq: Option<i64>,
    ) {
        let pending = {
            let _guard = self.lock_projection_barrier().await;
            self.handle_option_custody_delta_under_barrier(wallet, symbol, quantity_delta, seq)
                .await
        };
        self.send_pending_notifications(pending).await;
    }

    pub(crate) async fn handle_option_custody_delta_under_barrier(
        &self,
        wallet: WalletAddress,
        symbol: String,
        quantity_delta: Decimal,
        seq: Option<i64>,
    ) -> PendingNotifications {
        let change = self
            .service
            .apply_option_custody_delta(&wallet, &symbol, quantity_delta)
            .await;
        if let Some(seq) = seq {
            let mut last_seq = self.last_processed_seq.write().await;
            *last_seq = seq;
        }
        PendingNotifications::Changes(vec![change])
    }

    pub async fn replay_option_custody_delta(
        &self,
        wallet: WalletAddress,
        symbol: String,
        quantity_delta: Decimal,
        seq: i64,
    ) {
        let _guard = self.lock_projection_barrier().await;
        self.handle_option_custody_delta_under_barrier(wallet, symbol, quantity_delta, Some(seq))
            .await;
    }

    /// Send fill notifications. Call AFTER releasing the
    /// projection barrier so these non-critical operations don't block other
    /// state mutations.
    pub(crate) async fn send_fill_notifications(&self, fill: &hypercall_types::Fill) {
        let timestamp = chrono::Utc::now().timestamp();

        let (taker_pos, taker_margin) = self
            .get_position_and_margin_for_notification(&fill.taker_wallet_address, &fill.symbol)
            .await;
        let taker_change = PortfolioChange {
            wallet: fill.taker_wallet_address,
            position_changes: vec![taker_pos],
            balance_change: None,
            total_margin_used: taker_margin,
        };
        self.send_change_notifications(&taker_change, timestamp)
            .await;

        let (maker_pos, maker_margin) = self
            .get_position_and_margin_for_notification(&fill.maker_wallet_address, &fill.symbol)
            .await;
        let maker_change = PortfolioChange {
            wallet: fill.maker_wallet_address,
            position_changes: vec![maker_pos],
            balance_change: None,
            total_margin_used: maker_margin,
        };
        self.send_change_notifications(&maker_change, timestamp)
            .await;
    }

    pub async fn replay_journal_fill(&self, fill: &hypercall_types::Fill, seq: i64) {
        let _guard = self.lock_projection_barrier().await;
        self.apply_fill_state_mutation(fill, seq).await;
    }

    /// Apply fill state mutations to portfolio memory. This is the barrier-protected
    /// part of fill processing: position updates and seq advancement.
    /// Does NOT send notifications -- callers should use
    /// `send_fill_notifications` after releasing the barrier.
    async fn apply_fill_state_mutation(&self, fill: &hypercall_types::Fill, seq: i64) {
        let taker_side_orderbook = fill.taker_side;
        let maker_side = match fill.taker_side {
            hypercall_types::Side::Buy => hypercall_types::Side::Sell,
            hypercall_types::Side::Sell => hypercall_types::Side::Buy,
        };
        let size_human = to_human_readable_decimal(&fill.symbol, fill.size);

        self.service
            .apply_fill_to_memory_both_sides(
                &fill.taker_wallet_address,
                &fill.maker_wallet_address,
                &fill.symbol,
                &taker_side_orderbook,
                &maker_side,
                fill.price,
                size_human,
            )
            .await;

        let mut last_seq = self.last_processed_seq.write().await;
        *last_seq = seq;
    }

    async fn apply_position_expired_state_mutation(
        &self,
        expiry_msg: &PositionExpiredMessage,
        seq: Option<i64>,
    ) -> PortfolioChange {
        let settlement = self.settlement.clone();
        let expiry_msg_for_repair = expiry_msg.clone();
        let settlement_outcome = tokio::task::spawn_blocking(move || {
            settlement.try_apply_settlement_sync(
                &expiry_msg_for_repair.wallet_address,
                &expiry_msg_for_repair.symbol,
                expiry_msg_for_repair.position_size,
                expiry_msg_for_repair.settlement_price,
                expiry_msg_for_repair.settlement_value,
                expiry_msg_for_repair.margin_mode,
                expiry_msg_for_repair.timestamp as i64,
                expiry_msg_for_repair.settlement_entry_price,
                expiry_msg_for_repair.cost_basis,
                expiry_msg_for_repair.net_pnl,
            )
        })
        .await
        .unwrap_or_else(|error| {
            panic!(
                "CRITICAL: settlement repair task join failed for {}/{}: {}",
                expiry_msg.wallet_address, expiry_msg.symbol, error
            )
        })
        .unwrap_or_else(|error| {
            metrics::counter!("ht_settlement_apply_failures_total").increment(1);
            panic!(
                "CRITICAL: failed to confirm or repair settlement accounting for {}/{}: {}",
                expiry_msg.wallet_address, expiry_msg.symbol, error
            )
        });

        if settlement_outcome.newly_persisted {
            tracing::warn!(
                wallet = %expiry_msg.wallet_address,
                symbol = %expiry_msg.symbol,
                "Repaired missing settlement payout before portfolio projection"
            );
        } else {
            let confirmed_applied = self
                .settlement
                .is_settlement_ledger_applied_sync(&expiry_msg.wallet_address, &expiry_msg.symbol)
                .unwrap_or_else(|error| {
                    panic!(
                        "CRITICAL: DB error checking repaired settlement state for {}/{}: {}. \
                         Panicking to allow replay.",
                        expiry_msg.wallet_address, expiry_msg.symbol, error
                    )
                });
            if !confirmed_applied {
                metrics::counter!("ht_settlement_apply_failures_total").increment(1);
                panic!(
                    "CRITICAL: settlement repair completed but ledger_applied is still false for {}/{}",
                    expiry_msg.wallet_address, expiry_msg.symbol
                )
            }
        }

        self.service
            .remove_expired_position(&expiry_msg.wallet_address, &expiry_msg.symbol)
            .await;

        tracing::info!(
            "PortfolioCache: Applied settled expiry projection for {}/{}",
            expiry_msg.wallet_address,
            expiry_msg.symbol,
        );
        metrics::counter!("ht_settlement_apply_success_total").increment(1);

        if let Some(seq) = seq {
            let mut last_seq = self.last_processed_seq.write().await;
            *last_seq = seq;
        }

        let (position_change, total_margin_used) = self
            .get_position_and_margin_for_notification(
                &expiry_msg.wallet_address,
                &expiry_msg.symbol,
            )
            .await;

        PortfolioChange {
            wallet: expiry_msg.wallet_address,
            position_changes: vec![position_change],
            balance_change: None,
            total_margin_used,
        }
    }

    /// Dispatch pending notification work that was collected under the barrier.
    pub(crate) async fn send_pending_notifications(&self, pending: PendingNotifications) {
        match pending {
            PendingNotifications::Fill { fill } => self.send_fill_notifications(&fill).await,
            PendingNotifications::Changes(changes) => {
                let timestamp = chrono::Utc::now().timestamp();
                for change in &changes {
                    self.send_change_notifications(change, timestamp).await;
                }
            }
        }
    }

    /// Send portfolio update notifications from a PortfolioChange struct.
    ///
    /// This is the centralized notification method - all portfolio state changes
    /// should go through here to ensure consistent WebSocket updates.
    async fn send_change_notifications(&self, change: &PortfolioChange, timestamp: i64) {
        // Send position updates for each changed position
        for pos_change in &change.position_changes {
            // If margin worked for the fill, it must work for the notification.
            // Crash instead of silently dropping the PositionUpdate (causes WS inconsistency).
            let position = self
                .build_enriched_position_update(&change.wallet, &pos_change.symbol)
                .await
                .expect("build_enriched_position_update failed after successful fill — margin state is inconsistent");
            self.notify_subscribers(
                &change.wallet,
                PortfolioUpdate::PositionUpdate {
                    position,
                    timestamp,
                },
            )
            .await;
        }

        // Send balance update
        self.notify_subscribers(
            &change.wallet,
            PortfolioUpdate::BalanceUpdate {
                total_margin_used: change.total_margin_used,
                timestamp,
            },
        )
        .await;

        self.publish_greeks_update(&change.wallet).await;
    }

    /// Get position state and margin for a wallet/symbol after a fill.
    ///
    /// Returns both values from a single portfolio read to avoid redundant lock acquisitions.
    /// Used to construct PortfolioChange for notification after apply_fill_to_memory.
    async fn get_position_and_margin_for_notification(
        &self,
        wallet: &WalletAddress,
        symbol: &str,
    ) -> (crate::portfolio::PositionChange, Decimal) {
        if let Some(balance) = self.service.get_portfolio_balance(wallet).await {
            let margin = balance.total_margin_used;
            if let Some(pos) = balance.positions.get(symbol) {
                return (
                    crate::portfolio::PositionChange {
                        symbol: symbol.to_string(),
                        amount: pos.amount,
                        entry_price: pos.entry_price,
                        margin_posted: pos.margin_posted,
                        realized_pnl: pos.realized_pnl,
                        unrealized_pnl: pos.unrealized_pnl,
                    },
                    margin,
                );
            }
            // Position closed but portfolio exists
            return (
                crate::portfolio::PositionChange {
                    symbol: symbol.to_string(),
                    amount: dec!(0),
                    entry_price: dec!(0),
                    margin_posted: dec!(0),
                    realized_pnl: dec!(0),
                    unrealized_pnl: dec!(0),
                },
                margin,
            );
        }
        // Portfolio doesn't exist
        (
            crate::portfolio::PositionChange {
                symbol: symbol.to_string(),
                amount: dec!(0),
                entry_price: dec!(0),
                margin_posted: dec!(0),
                realized_pnl: dec!(0),
                unrealized_pnl: dec!(0),
            },
            dec!(0),
        )
    }

    /// Subscribe to portfolio updates for a specific account
    pub async fn subscribe(
        &self,
        account: WalletAddress,
    ) -> (SubscriberId, mpsc::UnboundedReceiver<PortfolioUpdate>) {
        let (tx, rx) = mpsc::unbounded_channel();

        // Generate a new subscriber ID
        let mut next_id = self.next_subscriber_id.write().await;
        let subscriber_id = *next_id;
        *next_id += 1;
        drop(next_id);

        let positions = match self.build_enriched_positions(&account).await {
            Ok(positions) => positions,
            Err(e) => {
                tracing::error!(
                    "Failed to build initial portfolio snapshot for {}: {}",
                    account,
                    e
                );
                return (subscriber_id, rx);
            }
        };

        // Add subscriber to the map
        let mut subscribers = self.subscribers.write().await;
        subscribers
            .entry(account)
            .or_insert_with(HashMap::new)
            .insert(subscriber_id, tx.clone());
        drop(subscribers);

        let _ = tx.send(PortfolioUpdate::Initial {
            positions,
            timestamp: chrono::Utc::now().timestamp(),
        });

        // Trigger margin update so subscriber gets equity/margin immediately
        // (publish_margin_update is a no-op if margin services aren't configured)
        self.publish_margin_update(&account).await;
        self.publish_greeks_update(&account).await;

        (subscriber_id, rx)
    }

    /// Unsubscribe a specific subscriber
    pub async fn unsubscribe(&self, account: &WalletAddress, subscriber_id: SubscriberId) {
        let mut subscribers = self.subscribers.write().await;
        if let Some(account_subs) = subscribers.get_mut(account) {
            account_subs.remove(&subscriber_id);
            if account_subs.is_empty() {
                subscribers.remove(account);
            }
        }
    }

    /// Return the number of active subscribers for a wallet.
    pub async fn subscriber_count_for_wallet(&self, account: &WalletAddress) -> usize {
        let subscribers = self.subscribers.read().await;
        subscribers
            .get(account)
            .map(|account_subs| account_subs.len())
            .unwrap_or(0)
    }

    /// Notify subscribers of portfolio updates
    async fn notify_subscribers(&self, account: &WalletAddress, update: PortfolioUpdate) {
        let failed_subscribers: Vec<SubscriberId> = {
            let subscribers = self.subscribers.read().await;
            let Some(account_subs) = subscribers.get(account) else {
                return;
            };

            account_subs
                .iter()
                .filter_map(|(subscriber_id, tx)| {
                    tx.send(update.clone()).err().map(|_| *subscriber_id)
                })
                .collect()
        };

        if failed_subscribers.is_empty() {
            return;
        }

        let mut subscribers = self.subscribers.write().await;
        if let Some(account_subs) = subscribers.get_mut(account) {
            for subscriber_id in failed_subscribers {
                account_subs.remove(&subscriber_id);
            }

            if account_subs.is_empty() {
                subscribers.remove(account);
            }
        }
    }

    pub async fn process_test_fill(
        &self,
        account: &WalletAddress,
        option_id: &str,
        side: &str,
        price: Decimal,
        volume: Decimal,
    ) {
        use hypercall_types::Fill;
        use hypercall_types::Side;

        let side_enum = match side {
            "buy" => Side::Buy,
            "sell" => Side::Sell,
            _ => return,
        };

        let size_contract_units = to_contract_units_decimal(option_id, volume);

        let fill = Fill {
            trade_id: 0,
            taker_order_id: 0,
            maker_order_id: 0,
            symbol: option_id.to_string(),
            price,
            size: size_contract_units,
            taker_side: side_enum,
            taker_wallet_address: *account,
            maker_wallet_address: WalletAddress::from(alloy::primitives::Address::ZERO),
            fee: dec!(0),
            is_taker: true,
            timestamp: 0,
            builder_code_address: None,
            builder_code_fee: None,
            source: Default::default(),
            taker_realized_pnl: None,
            maker_realized_pnl: None,
            underlying_notional: None,
        };

        self.service
            .apply_event(&EngineMessage::OrderFilled {
                accounting: hypercall_engine::FillAccounting::from_fill(&fill),
                fill,
            })
            .await
            .unwrap();
    }

    /// Get portfolio for a specific account
    pub async fn get_portfolio(&self, account: &WalletAddress) -> Result<Portfolio> {
        if let Err(error) = self.refresh_live_market_prices_for_wallet(account).await {
            tracing::warn!(
                "Failed to refresh live market prices for {} before REST portfolio read: {}",
                account,
                error
            );
        }
        Ok(self.service.get_portfolio(account).await)
    }

    pub async fn get_portfolio_fail_closed_pm(&self, account: &WalletAddress) -> Result<Portfolio> {
        self.refresh_live_market_prices_for_wallet(account)
            .await
            .map_err(|error| anyhow::anyhow!("PM portfolio read unavailable: {}", error))?;
        Ok(self.service.get_portfolio(account).await)
    }

    pub async fn has_live_position_symbol(&self, wallet: &WalletAddress, symbol: &str) -> bool {
        self.service
            .get_portfolio_balance(wallet)
            .await
            .map(|balance| balance.positions.contains_key(symbol))
            .unwrap_or(false)
    }

    /// Check if a portfolio exists for an account.
    pub async fn has_portfolio(&self, account: &WalletAddress) -> bool {
        self.service.get_portfolio_balance(account).await.is_some()
    }

    /// Update market prices for unrealized PnL calculations
    pub async fn update_market_prices(&self, prices: HashMap<String, Decimal>) {
        // Delegate to service
        self.service.update_market_prices(prices).await;
    }

    /// Get the last processed portfolio sequence marker.
    pub async fn get_last_processed_seq(&self) -> i64 {
        *self.last_processed_seq.read().await
    }

    /// Capture current `(seq, portfolios)` for tests and diagnostics.
    pub async fn capture_snapshot_state(&self) -> (i64, HashMap<WalletAddress, PortfolioBalance>) {
        let _guard = self.lock_projection_barrier().await;
        let seq = *self.last_processed_seq.read().await;
        let portfolios = self.service.all_portfolios().await;
        (seq, portfolios)
    }

    pub async fn capture_portfolios_under_barrier(
        &self,
        _barrier_proof: &OwnedMutexGuard<()>,
    ) -> HashMap<WalletAddress, PortfolioBalance> {
        self.service.all_portfolios().await
    }

    /// Get all portfolios with margin info (for metrics collection).
    ///
    /// Returns a map of wallet address to portfolio summary including positions and margin.
    pub async fn get_all_portfolios(&self) -> HashMap<WalletAddress, PortfolioSummary> {
        let portfolios = self.service.all_portfolios().await;
        let mut result = HashMap::new();

        for (wallet, balance) in portfolios {
            // Convert positions to summary format
            let positions: HashMap<String, PositionSummary> = balance
                .positions
                .iter()
                .map(|(symbol, pos)| {
                    (
                        symbol.clone(),
                        PositionSummary {
                            symbol: pos.symbol.clone(),
                            amount: pos.amount,
                            entry_price: pos.entry_price,
                            realized_pnl: pos.realized_pnl,
                            unrealized_pnl: pos.unrealized_pnl,
                        },
                    )
                })
                .collect();

            result.insert(
                wallet,
                PortfolioSummary {
                    positions,
                    margin_info: None, // Margin is computed on-demand, not cached
                },
            );
        }

        result
    }
}

/// Summary of a portfolio for metrics collection.
#[derive(Debug, Clone)]
pub struct PortfolioSummary {
    pub positions: HashMap<String, PositionSummary>,
    pub margin_info: Option<MarginInfo>,
}

/// Summary of a position for metrics collection.
#[derive(Debug, Clone)]
pub struct PositionSummary {
    pub symbol: String,
    pub amount: Decimal,
    pub entry_price: Decimal,
    pub realized_pnl: Decimal,
    pub unrealized_pnl: Decimal,
}

/// Summary of margin info for metrics collection.
#[derive(Debug, Clone)]
pub struct MarginInfo {
    pub equity: Decimal,
    pub initial_margin: Decimal,
    pub maintenance_margin: Decimal,
}

use crate::shared::order_types::perp_underlying as portfolio_perp_underlying;

#[cfg(test)]
mod tests {
    use super::*;
    use crate::messaging::{EventBusTrait, MockEventBus};
    use crate::price_oracle::hyperliquid_oracle::{
        HyperliquidMarkPriceOracle, HyperliquidOracleConfig,
    };
    use crate::read_cache::tier::TierCache;
    use crate::rsm::engine_snapshot::{
        MockOrderSnapshotProvider, MockQuoteProvider, SnapshotOpenOrdersSource,
    };
    use crate::rsm::ledger::InMemoryLedger;
    use crate::rsm::margin_service::SpanMarginService;
    use crate::rsm::MarginMode;
    use crate::snapshot::{DbPortfolioSnapshotWriter, SnapshotWriter};
    use crate::standard_margin::StandardMarginService;
    use crate::types::{Config, Scenario, ScenarioType};
    use crate::vol_oracle::FixedTestRiskVolOracle;
    use hypercall_engine::FeeConfig;
    use hypercall_runtime_api::{OrderSnapshotProvider, QuoteProvider, SnapshotBookQuote};
    use hypercall_types::Fill;
    use hypercall_types::Side;
    use sqlx::postgres::{PgConnectOptions, PgPoolOptions};
    use std::collections::{HashMap, HashSet};
    use testcontainers::runners::AsyncRunner;
    use testcontainers::ContainerAsync;
    use testcontainers::ImageExt;
    use testcontainers_modules::postgres::Postgres;
    use tokio::time::{sleep, timeout, Duration};
    use uuid::Uuid;

    use hypercall_types::wallet_address::test_wallet;

    struct TestContext {
        cache: Arc<PortfolioCache>,
        diesel_db: Arc<hypercall_db_diesel::DieselDb>,
        sqlx_connect_options: PgConnectOptions,
        _container: Option<ContainerAsync<Postgres>>,
    }

    fn normalize_test_database_url(base_database_url: &str, database_name: Option<&str>) -> String {
        let mut parts = base_database_url.splitn(2, '?');
        let base = parts
            .next()
            .expect("DATABASE_URL must contain a URL base component");
        let normalized_base = match parts.next() {
            Some(query) => {
                let mut seen_keys = HashSet::new();
                let normalized_query = query
                    .replace('?', "&")
                    .split('&')
                    .filter(|pair| !pair.is_empty())
                    .filter(|pair| {
                        let key = pair.split('=').next().unwrap_or_default();
                        seen_keys.insert(key.to_string())
                    })
                    .collect::<Vec<_>>()
                    .join("&");
                if normalized_query.is_empty() {
                    base.to_string()
                } else {
                    format!("{base}?{normalized_query}")
                }
            }
            None => base.to_string(),
        };

        let Some(database_name) = database_name else {
            return normalized_base;
        };

        let (normalized_base, normalized_query) = match normalized_base.split_once('?') {
            Some((base, query)) => (base, Some(query)),
            None => (normalized_base.as_str(), None),
        };
        let (base_prefix, _) = normalized_base
            .rsplit_once('/')
            .expect("DATABASE_URL must include a database path");
        let rewritten_base = format!("{base_prefix}/{database_name}");
        match normalized_query {
            Some(query) if !query.is_empty() => format!("{rewritten_base}?{query}"),
            _ => rewritten_base,
        }
    }

    async fn try_setup_test_cache_from_shared_database(
        base_database_url: &str,
    ) -> Result<TestContext, String> {
        let database_name = format!("portfolio_cache_test_{}", Uuid::now_v7().simple());
        let admin_database_url = normalize_test_database_url(base_database_url, None);
        let admin_pool = PgPoolOptions::new()
            .max_connections(1)
            .min_connections(1)
            .acquire_timeout(Duration::from_secs(30))
            .connect(&admin_database_url)
            .await
            .map_err(|error| {
                format!("failed to connect to shared PostgreSQL test database: {error}")
            })?;
        sqlx::query(&format!("CREATE DATABASE \"{}\"", database_name))
            .execute(&admin_pool)
            .await
            .map_err(|error| {
                format!("failed to create isolated PostgreSQL test database: {error}")
            })?;
        admin_pool.close().await;

        let database_url = normalize_test_database_url(base_database_url, Some(&database_name));
        let sqlx_connect_options = PgConnectOptions::from_str(&database_url).map_err(|error| {
            format!("isolated PostgreSQL test database URL must parse: {error}")
        })?;
        let diesel_handler = Arc::new(
            DatabaseHandler::new(&database_url)
                .map_err(|error| format!("failed to initialize DieselEventHandler: {error:#}"))?,
        );
        let cache = Arc::new(PortfolioCache::new(diesel_handler));
        let _ = cache
            .initialize()
            .await
            .map_err(|error| format!("failed to initialize portfolio cache: {error}"))?;

        let diesel_db = Arc::new(
            hypercall_db_diesel::DieselDb::new_no_tls(&database_url, 2)
                .await
                .map_err(|error| format!("failed to initialize DieselDb: {error:#}"))?,
        );

        Ok(TestContext {
            cache,
            diesel_db,
            sqlx_connect_options,
            _container: None,
        })
    }

    async fn setup_test_cache() -> TestContext {
        if let Ok(base_database_url) = std::env::var("TEST_DATABASE_URL") {
            match try_setup_test_cache_from_shared_database(&base_database_url).await {
                Ok(context) => return context,
                Err(error) => {
                    eprintln!(
                        "shared PostgreSQL test database unavailable for portfolio_cache tests: {error}; falling back to testcontainer"
                    );
                }
            }
        }

        // Start PostgreSQL container with the new async API
        // Match deployed PostgreSQL 16 lanes.
        // testcontainers-modules defaults to postgres:11-alpine, where
        // `ALTER TYPE ... ADD VALUE` fails inside a transaction — PG12 lifted
        // that restriction, so pinning 16-alpine keeps migrations (including
        // the enum-extending ones in migrations/2026-04-11-000001 and
        // migrations/2026-04-12-000002) runnable under diesel's default
        // per-migration transaction.
        //
        // NOTE: `with_tag` returns `ContainerRequest<Postgres>` via the
        // `ImageExt` trait, which doesn't expose the Postgres-specific
        // builders (`with_db_name`, `with_user`, ...). Call it last.
        let postgres_image = Postgres::default()
            .with_db_name("test_db")
            .with_user("test_user")
            .with_password("test_password")
            .with_startup_timeout(Duration::from_secs(120))
            .with_tag("16-alpine");

        let container = postgres_image
            .start()
            .await
            .expect("Failed to start PostgreSQL container");
        let port = container
            .get_host_port_ipv4(5432)
            .await
            .expect("Failed to get port");

        let database_url = format!(
            "postgresql://test_user:test_password@127.0.0.1:{}/test_db",
            port
        );

        // CI can expose the mapped port before Postgres is ready to accept
        // migrations. Poll the database directly instead of sleeping a fixed
        // amount of time.
        let mut last_ready_error = None;
        for _ in 0..60 {
            match PgPoolOptions::new()
                .max_connections(1)
                .min_connections(1)
                .acquire_timeout(Duration::from_secs(5))
                .connect(&database_url)
                .await
            {
                Ok(pool) => match sqlx::query("SELECT 1").execute(&pool).await {
                    Ok(_) => {
                        pool.close().await;
                        last_ready_error = None;
                        break;
                    }
                    Err(error) => {
                        last_ready_error = Some(format!("readiness query failed: {error}"));
                        pool.close().await;
                    }
                },
                Err(error) => {
                    last_ready_error = Some(format!("failed to connect to postgres: {error}"));
                }
            }

            sleep(Duration::from_secs(1)).await;
        }

        if let Some(error) = last_ready_error {
            panic!("PostgreSQL container failed readiness checks: {error}");
        }

        // Create diesel handler which will run migrations. Even after the
        // readiness query succeeds, Postgres can still be finishing startup on
        // slower CI machines, so give migrations a short retry window.
        let mut diesel_handler = None;
        let mut last_handler_error = None;
        for _ in 0..10 {
            match DatabaseHandler::new(&database_url) {
                Ok(handler) => {
                    diesel_handler = Some(Arc::new(handler));
                    last_handler_error = None;
                    break;
                }
                Err(error) => {
                    last_handler_error =
                        Some(format!("failed to create diesel handler: {error:#}"));
                    sleep(Duration::from_secs(1)).await;
                }
            }
        }

        let diesel_handler = diesel_handler.unwrap_or_else(|| {
            panic!(
                "PostgreSQL container never became migration-ready: {}",
                last_handler_error.unwrap_or_else(|| "unknown error".to_string())
            )
        });

        let cache = Arc::new(PortfolioCache::new(diesel_handler));

        // Initialize returns the next engine command_id to replay - we ignore it for these tests
        let _ = cache.initialize().await.unwrap();

        let diesel_db = Arc::new(
            hypercall_db_diesel::DieselDb::new(&database_url, 2)
                .await
                .expect("DieselDb must initialize for test context"),
        );

        TestContext {
            cache,
            diesel_db,
            sqlx_connect_options: PgConnectOptions::from_str(&database_url)
                .expect("Test container database URL must parse for sqlx"),
            _container: Some(container),
        }
    }

    async fn attach_test_greeks_cache(context: &TestContext, symbol: &str) {
        let raw_event_bus = Arc::new(MockEventBus::new().expect("Failed to create mock event bus"));
        raw_event_bus.clone().start_processing().await;
        let event_bus: Arc<dyn EventBusTrait> = raw_event_bus.clone();

        {
            use diesel::prelude::*;
            use diesel::sql_types::{BigInt, Numeric, Text};

            let pool = context.cache.db.pool();
            let mut conn = pool.get().expect("Failed to get Diesel connection");
            diesel::sql_query(
                "INSERT INTO instruments (id, underlying, strike, expiry, option_type)
                 VALUES ($1, $2, $3, $4, $5)
                 ON CONFLICT (id) DO NOTHING",
            )
            .bind::<Text, _>(symbol)
            .bind::<Text, _>("BTC")
            .bind::<Numeric, _>(dec!(50000))
            .bind::<BigInt, _>(20261231_i64)
            .bind::<Text, _>("call")
            .execute(&mut conn)
            .expect("Failed to insert test instrument");
        }

        let oracle_config = HyperliquidOracleConfig {
            api_url: "https://api.hyperliquid-testnet.xyz/info".to_string(),
            poll_interval_ms: 60_000,
            risk_free_rate: 0.05,
            symbol: "BTC".to_string(),
            twap_window_seconds: 300,
            oracle_writer: None,
            max_memory_samples: 100,
            ws_feed: None,
            price_notify: None,
            min_settlement_samples: 500,
        };
        let btc_oracle = Arc::new(
            HyperliquidMarkPriceOracle::new(oracle_config).expect("Failed to create BTC oracle"),
        );
        btc_oracle.set_spot_price_for_testing(50_000.0).await;

        let mut oracles = HashMap::new();
        oracles.insert("BTC".to_string(), btc_oracle);

        let quote_provider = Arc::new(MockQuoteProvider::new());
        quote_provider.set_quote(
            symbol,
            SnapshotBookQuote {
                best_bid: Some(10_000.0),
                best_bid_size: Some(5.0),
                best_ask: Some(11_000.0),
                best_ask_size: Some(5.0),
                mid: Some(10_500.0),
                bids: vec![(10_000.0, 5.0)],
                asks: vec![(11_000.0, 5.0)],
            },
        );
        let quote_provider_trait: Arc<dyn QuoteProvider> = quote_provider.clone();
        let (_shutdown_tx, shutdown_rx) = tokio::sync::broadcast::channel::<()>(1);

        let greeks_cache = GreeksCache::new(
            context.diesel_db.as_ref(),
            event_bus,
            oracles,
            shutdown_rx,
            quote_provider_trait,
        )
        .await
        .expect("Failed to create GreeksCache");
        greeks_cache
            .set_vol_oracle(Arc::new(FixedTestRiskVolOracle::new(0.5)))
            .await;

        let mut greeks_ready = false;
        let mut last_error = String::new();
        for _ in 0..20 {
            match greeks_cache.get_greeks(symbol).await {
                Ok(_) => {
                    greeks_ready = true;
                    break;
                }
                Err(e) => {
                    last_error = e.to_string();
                }
            }
            sleep(Duration::from_millis(50)).await;
        }
        assert!(
            greeks_ready,
            "expected test GreeksCache to have greeks for {}, last error: {}",
            symbol, last_error
        );

        *context.cache.greeks_cache.write().await = Some(greeks_cache);
    }

    /// Insert a trade record to satisfy the foreign key constraint on fills.
    async fn insert_trade_for_fill(context: &TestContext, fill: &hypercall_types::Fill) {
        use diesel::prelude::*;
        use diesel::sql_types::{BigInt, Binary, Numeric, Text};
        use rust_decimal_macros::dec;

        let pool = context.cache.db.pool();
        let mut conn = pool.get().expect("Failed to get connection");

        diesel::sql_query(
            "INSERT INTO trades (trade_id, symbol, price, size, maker_address, taker_address, maker_fee, taker_fee, timestamp)
             VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
             ON CONFLICT (trade_id) DO NOTHING"
        )
        .bind::<BigInt, _>(fill.trade_id as i64)
        .bind::<Text, _>(&fill.symbol)
        .bind::<Numeric, _>(fill.price)
        .bind::<Numeric, _>(fill.size)
        .bind::<Binary, _>(&fill.maker_wallet_address)
        .bind::<Binary, _>(&fill.taker_wallet_address)
        .bind::<Numeric, _>(dec!(0)) // maker_fee
        .bind::<Numeric, _>(fill.fee) // taker_fee
        .bind::<BigInt, _>(fill.timestamp as i64)
        .execute(&mut conn)
        .expect("Failed to insert trade");
    }

    async fn insert_applied_settlement(
        context: &TestContext,
        wallet: WalletAddress,
        symbol: &str,
        position_size: Decimal,
        settlement_entry_price: Decimal,
        cost_basis: Decimal,
        net_pnl: Decimal,
    ) {
        let settlement_price = dec!(1000);
        let settlement_value = position_size * settlement_price;
        let outcome = context
            .cache
            .settlement
            .try_apply_settlement_sync(
                &wallet,
                symbol,
                position_size,
                settlement_price,
                settlement_value,
                MarginMode::Standard,
                2,
                Some(settlement_entry_price),
                Some(cost_basis),
                Some(net_pnl),
            )
            .expect("Failed to persist applied settlement");
        assert!(outcome.newly_persisted);
    }

    async fn attach_test_pm_margin_dependencies(
        context: &TestContext,
        wallet: WalletAddress,
    ) -> Arc<TierCache> {
        attach_test_greeks_cache(context, "BTC-20261231-50000-C").await;
        let greeks_cache = context
            .cache
            .greeks_cache
            .read()
            .await
            .as_ref()
            .cloned()
            .expect("greeks cache should be configured");

        let ledger: Arc<dyn crate::rsm::Ledger + Send + Sync> = Arc::new(InMemoryLedger::new());
        let portfolio_service: Arc<dyn crate::portfolio::PortfolioService + Send + Sync> =
            context.cache.get_service();
        let order_snapshot: Arc<dyn OrderSnapshotProvider> = Arc::new(MockOrderSnapshotProvider);
        let snapshot_open_orders = Arc::new(SnapshotOpenOrdersSource::new(order_snapshot.clone()));
        let risk_account_builder = Arc::new(crate::rsm::portfolio_margin::RiskAccountBuilder::new(
            ledger.clone(),
            portfolio_service.clone(),
            snapshot_open_orders,
            greeks_cache.clone(),
        ));
        let standard_margin_service = Arc::new(StandardMarginService::new());
        let standard_account_builder =
            Arc::new(crate::standard_margin::StandardAccountBuilder::new(
                ledger,
                portfolio_service,
                greeks_cache.clone(),
            ));
        let tier_cache = Arc::new(
            TierCache::new(context.cache.db.clone()).expect("tier cache should initialize"),
        );
        tier_cache
            .set_margin_mode(&wallet, MarginMode::Portfolio)
            .await
            .expect("wallet should be set to PM");

        let span_margin_service = Arc::new(SpanMarginService::new_for_tests(Config {
            risk_free_rate: 0.05,
            base_volatility: 0.8,
            base_skew: 0.0,
            base_excess_kurtosis: 0.0,
            delta_threshold: 0.0001,
            strike_match_tolerance: 0.01,
            expiry_match_tolerance_years: 0.001,
            scenarios: vec![
                Scenario {
                    scenario_type: ScenarioType::SpotChange,
                    value: 0.15,
                },
                Scenario {
                    scenario_type: ScenarioType::SpotChange,
                    value: -0.15,
                },
            ],
            allow_standard_margin_shorts: false,
            fee_config: FeeConfig::default(),
        }));

        context
            .cache
            .set_margin_dependencies(
                span_margin_service,
                greeks_cache,
                risk_account_builder,
                tier_cache.clone(),
                order_snapshot,
                standard_margin_service,
                standard_account_builder,
            )
            .await;

        tier_cache
    }

    #[tokio::test]
    async fn test_get_portfolio_fail_closed_pm_requires_repricing_inputs() {
        let context = setup_test_cache().await;
        let wallet = test_wallet(31);

        context
            .cache
            .service
            .restore(
                &wallet,
                PortfolioBalance {
                    positions: HashMap::from([(
                        "BTC".to_string(),
                        crate::portfolio::PositionData {
                            symbol: "BTC".to_string(),
                            amount: dec!(1),
                            entry_price: dec!(48000),
                            margin_posted: dec!(0),
                            realized_pnl: dec!(0),
                            unrealized_pnl: dec!(0),
                        },
                    )]),
                    total_margin_used: dec!(0),
                },
            )
            .await
            .expect("restore should succeed");

        let err = context
            .cache
            .get_portfolio_fail_closed_pm(&wallet)
            .await
            .expect_err("PM REST reads must fail closed without repricing inputs");
        assert!(
            err.to_string().contains("PM portfolio read unavailable"),
            "unexpected error: {}",
            err
        );
    }

    #[tokio::test]
    async fn test_get_portfolio_fail_closed_pm_skips_perp_repricing() {
        let context = setup_test_cache().await;
        let wallet = test_wallet(32);

        context
            .cache
            .service
            .restore(
                &wallet,
                PortfolioBalance {
                    positions: HashMap::from([(
                        "BTC-PERP".to_string(),
                        crate::portfolio::PositionData {
                            symbol: "BTC-PERP".to_string(),
                            amount: dec!(1),
                            entry_price: dec!(48000),
                            margin_posted: dec!(0),
                            realized_pnl: dec!(0),
                            unrealized_pnl: dec!(750),
                        },
                    )]),
                    total_margin_used: dec!(0),
                },
            )
            .await
            .expect("restore should succeed");

        attach_test_greeks_cache(&context, "BTC-20260331-50000-C").await;

        let portfolio = context
            .cache
            .get_portfolio_fail_closed_pm(&wallet)
            .await
            .expect("PM REST reads should succeed — perps are skipped during repricing");
        let btc_position = portfolio
            .positions
            .iter()
            .find(|position| position.position.symbol == "BTC-PERP")
            .expect("BTC-PERP position should be present");
        // HyperCore UPNL is authoritative — not repriced from spot
        assert_eq!(btc_position.position.unrealized_pnl, dec!(750));
    }

    #[tokio::test]
    async fn test_get_portfolio_fail_closed_pm_repairs_applied_settlement_ghost_position() {
        let context = setup_test_cache().await;
        let wallet = test_wallet(33);
        let symbol = "BTC-20260403-34000-C";

        context
            .cache
            .service
            .restore(
                &wallet,
                PortfolioBalance {
                    positions: HashMap::from([(
                        symbol.to_string(),
                        crate::portfolio::PositionData {
                            symbol: symbol.to_string(),
                            amount: dec!(1),
                            entry_price: dec!(1500),
                            margin_posted: dec!(0),
                            realized_pnl: dec!(0),
                            unrealized_pnl: dec!(0),
                        },
                    )]),
                    total_margin_used: dec!(0),
                },
            )
            .await
            .expect("restore should succeed");
        insert_applied_settlement(
            &context,
            wallet,
            symbol,
            dec!(1),
            dec!(0),
            dec!(0),
            dec!(1000),
        )
        .await;

        let portfolio = context
            .cache
            .get_portfolio_fail_closed_pm(&wallet)
            .await
            .expect("PM portfolio should self-heal settled ghost positions");

        assert!(
            portfolio
                .positions
                .iter()
                .all(|position| position.position.symbol != symbol),
            "settled ghost symbol should be removed from REST portfolio"
        );
        let internal = context
            .cache
            .service
            .get_portfolio_balance(&wallet)
            .await
            .expect("wallet should still exist");
        assert!(
            !internal.positions.contains_key(symbol),
            "settled ghost symbol should be removed from in-memory portfolio"
        );
    }

    #[tokio::test]
    async fn test_get_portfolio_fail_closed_pm_keeps_fail_closed_without_settlement_proof() {
        let context = setup_test_cache().await;
        let wallet = test_wallet(34);
        let symbol = "BTC-20260403-34000-C";

        context
            .cache
            .service
            .restore(
                &wallet,
                PortfolioBalance {
                    positions: HashMap::from([(
                        symbol.to_string(),
                        crate::portfolio::PositionData {
                            symbol: symbol.to_string(),
                            amount: dec!(1),
                            entry_price: dec!(1500),
                            margin_posted: dec!(0),
                            realized_pnl: dec!(0),
                            unrealized_pnl: dec!(0),
                        },
                    )]),
                    total_margin_used: dec!(0),
                },
            )
            .await
            .expect("restore should succeed");

        let err = context
            .cache
            .get_portfolio_fail_closed_pm(&wallet)
            .await
            .expect_err("PM portfolio should stay fail-closed without settlement proof");
        assert!(
            err.to_string().contains("PM portfolio read unavailable"),
            "unexpected error: {}",
            err
        );
    }

    #[tokio::test]
    async fn test_compute_wallet_margin_snapshot_repairs_applied_settlement_ghost_position() {
        let context = setup_test_cache().await;
        let wallet = test_wallet(35);
        let symbol = "BTC-20260403-34000-C";

        context
            .cache
            .service
            .restore(
                &wallet,
                PortfolioBalance {
                    positions: HashMap::from([(
                        symbol.to_string(),
                        crate::portfolio::PositionData {
                            symbol: symbol.to_string(),
                            amount: dec!(1),
                            entry_price: dec!(1500),
                            margin_posted: dec!(0),
                            realized_pnl: dec!(0),
                            unrealized_pnl: dec!(0),
                        },
                    )]),
                    total_margin_used: dec!(0),
                },
            )
            .await
            .expect("restore should succeed");
        insert_applied_settlement(
            &context,
            wallet,
            symbol,
            dec!(1),
            dec!(0),
            dec!(0),
            dec!(1000),
        )
        .await;
        let _tier_cache = attach_test_pm_margin_dependencies(&context, wallet).await;

        let snapshot = context
            .cache
            .compute_wallet_margin_snapshot(&wallet)
            .await
            .expect("PM margin snapshot should self-heal settled ghost positions");

        assert_eq!(snapshot.mode, MarginMode::Portfolio);
        assert_eq!(snapshot.total_margin_used, dec!(0));
        assert_eq!(snapshot.available_balance, dec!(0));
        let internal = context
            .cache
            .service
            .get_portfolio_balance(&wallet)
            .await
            .expect("wallet should still exist");
        assert!(
            !internal.positions.contains_key(symbol),
            "settled ghost symbol should be removed before PM margin snapshot"
        );
    }

    #[tokio::test]
    async fn test_portfolio_initialization() {
        let context = setup_test_cache().await;
        let cache = context.cache.clone();

        // Initialize with empty database - replay starts from command_id 1
        let next_command_id = cache.initialize().await.unwrap();
        assert_eq!(next_command_id, 1);

        // Should have no portfolios
        let portfolios = cache.service.all_portfolios().await;
        assert_eq!(portfolios.len(), 0);
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_snapshot_and_recovery() {
        let context = setup_test_cache().await;
        let cache = context.cache.clone();

        // Create fills via engine messages
        let fill1 = Fill {
            trade_id: 1,
            taker_order_id: 100,
            maker_order_id: 101,
            symbol: "BTC-CALL-100000".to_string(),
            price: dec!(1000),
            size: to_contract_units_decimal("BTC-CALL-100000", dec!(10)),
            taker_side: Side::Buy,
            taker_wallet_address: test_wallet(1),
            maker_wallet_address: test_wallet(200),
            fee: dec!(0),
            is_taker: true,
            timestamp: 0,
            builder_code_address: None,
            builder_code_fee: None,
            source: Default::default(),
            taker_realized_pnl: None,
            maker_realized_pnl: None,
            underlying_notional: None,
        };

        let fill2 = Fill {
            trade_id: 2,
            taker_order_id: 102,
            maker_order_id: 103,
            symbol: "ETH-PUT-3000".to_string(),
            price: dec!(50),
            size: to_contract_units_decimal("ETH-PUT-3000", dec!(5)),
            taker_side: Side::Sell,
            taker_wallet_address: test_wallet(2),
            maker_wallet_address: test_wallet(200),
            fee: dec!(0),
            is_taker: true,
            timestamp: 0,
            builder_code_address: None,
            builder_code_fee: None,
            source: Default::default(),
            taker_realized_pnl: None,
            maker_realized_pnl: None,
            underlying_notional: None,
        };

        insert_trade_for_fill(&context, &fill1).await;
        cache
            .handle_engine_message(
                EngineMessage::OrderFilled {
                    accounting: hypercall_engine::FillAccounting::from_fill(&fill1),
                    fill: fill1,
                },
                1,
            )
            .await;
        insert_trade_for_fill(&context, &fill2).await;
        cache
            .handle_engine_message(
                EngineMessage::OrderFilled {
                    accounting: hypercall_engine::FillAccounting::from_fill(&fill2),
                    fill: fill2,
                },
                2,
            )
            .await;

        // Verify seq was set correctly by handle_engine_message
        assert_eq!(
            cache.get_last_processed_seq().await,
            2,
            "Seq should be 2 after processing 2 fills"
        );

        // Take snapshot using the standalone writer with barrier-aware capture
        let get_offsets = move || {
            let mut offsets = HashMap::new();
            offsets.insert(
                ENGINE_COMMAND_SNAPSHOT_STREAM.to_string(),
                HashMap::from([(0, 1)]),
            );
            Ok(offsets)
        };
        // Use barrier-aware capture for crash consistency
        let cache_for_capture = cache.clone();
        let capture_snapshot = move || {
            tokio::task::block_in_place(|| {
                tokio::runtime::Handle::current().block_on(async {
                    let (_seq, portfolios) = cache_for_capture.capture_snapshot_state().await;
                    let mut offsets = HashMap::new();
                    offsets.insert(
                        ENGINE_COMMAND_SNAPSHOT_STREAM.to_string(),
                        HashMap::from([(0, 1)]),
                    );
                    Ok((portfolios, offsets))
                })
            })
        };

        let writer =
            DbPortfolioSnapshotWriter::new(cache.db.clone(), cache.get_service(), get_offsets)
                .with_capture_snapshot(capture_snapshot);
        writer.take_snapshot().unwrap();

        // Create new cache instance with same DB
        let cache2 = Arc::new(PortfolioCache::new(cache.db.clone()));
        let next_command_id = cache2.initialize().await.unwrap();
        assert_eq!(next_command_id, 1);

        // Verify portfolios were recovered
        let portfolio1 = cache2.get_portfolio(&test_wallet(1)).await.unwrap();
        assert_eq!(portfolio1.positions.len(), 1);
        assert!(cache2.has_portfolio(&test_wallet(1)).await);

        let portfolio2 = cache2.get_portfolio(&test_wallet(2)).await.unwrap();
        assert_eq!(portfolio2.positions.len(), 1);
        assert!(cache2.has_portfolio(&test_wallet(2)).await);
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_replay_journal_fill_applies_multiple_fills_from_one_command() {
        let context = setup_test_cache().await;
        let cache = context.cache.clone();
        let symbol = "BTC-CALL-100000".to_string();
        let taker_wallet = test_wallet(1);
        let maker_wallet = test_wallet(200);

        let fill1 = Fill {
            trade_id: 10,
            taker_order_id: 100,
            maker_order_id: 101,
            symbol: symbol.clone(),
            price: dec!(1000),
            size: to_contract_units_decimal(&symbol, dec!(0.1)),
            taker_side: Side::Buy,
            taker_wallet_address: taker_wallet,
            maker_wallet_address: maker_wallet,
            fee: dec!(0),
            is_taker: true,
            timestamp: 0,
            builder_code_address: None,
            builder_code_fee: None,
            source: Default::default(),
            taker_realized_pnl: None,
            maker_realized_pnl: None,
            underlying_notional: None,
        };
        let fill2 = Fill {
            trade_id: 11,
            taker_order_id: 100,
            maker_order_id: 102,
            symbol: symbol.clone(),
            price: dec!(1000),
            size: to_contract_units_decimal(&symbol, dec!(0.9)),
            taker_side: Side::Buy,
            taker_wallet_address: taker_wallet,
            maker_wallet_address: maker_wallet,
            fee: dec!(0),
            is_taker: true,
            timestamp: 0,
            builder_code_address: None,
            builder_code_fee: None,
            source: Default::default(),
            taker_realized_pnl: None,
            maker_realized_pnl: None,
            underlying_notional: None,
        };

        insert_trade_for_fill(&context, &fill1).await;
        insert_trade_for_fill(&context, &fill2).await;

        cache.replay_journal_fill(&fill1, 42).await;
        cache.replay_journal_fill(&fill2, 42).await;

        let balance = cache
            .service
            .get_portfolio_balance(&taker_wallet)
            .await
            .expect("taker portfolio should exist after replay");
        assert_eq!(balance.positions.get(&symbol).unwrap().amount, dec!(1.0));
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_snapshot_recovery_repairs_applied_settlement_ghost_position_on_read() {
        let context = setup_test_cache().await;
        let wallet = test_wallet(36);
        let symbol = "BTC-20260403-34000-C";

        context
            .cache
            .service
            .restore(
                &wallet,
                PortfolioBalance {
                    positions: HashMap::from([(
                        symbol.to_string(),
                        crate::portfolio::PositionData {
                            symbol: symbol.to_string(),
                            amount: dec!(1),
                            entry_price: dec!(1500),
                            margin_posted: dec!(0),
                            realized_pnl: dec!(0),
                            unrealized_pnl: dec!(0),
                        },
                    )]),
                    total_margin_used: dec!(0),
                },
            )
            .await
            .expect("restore should succeed");
        insert_applied_settlement(
            &context,
            wallet,
            symbol,
            dec!(1),
            dec!(0),
            dec!(0),
            dec!(1000),
        )
        .await;

        let cache = context.cache.clone();
        let get_offsets = move || {
            let mut offsets = HashMap::new();
            offsets.insert(
                ENGINE_COMMAND_SNAPSHOT_STREAM.to_string(),
                HashMap::from([(0, 1)]),
            );
            Ok(offsets)
        };
        let cache_for_capture = cache.clone();
        let capture_snapshot = move || {
            tokio::task::block_in_place(|| {
                tokio::runtime::Handle::current().block_on(async {
                    let (_seq, portfolios) = cache_for_capture.capture_snapshot_state().await;
                    let mut offsets = HashMap::new();
                    offsets.insert(
                        ENGINE_COMMAND_SNAPSHOT_STREAM.to_string(),
                        HashMap::from([(0, 1)]),
                    );
                    Ok((portfolios, offsets))
                })
            })
        };

        let writer =
            DbPortfolioSnapshotWriter::new(cache.db.clone(), cache.get_service(), get_offsets)
                .with_capture_snapshot(capture_snapshot);
        writer.take_snapshot().expect("snapshot should succeed");

        let cache2 = Arc::new(PortfolioCache::new(cache.db.clone()));
        let next_command_id = cache2
            .initialize()
            .await
            .expect("initialize should succeed");
        assert_eq!(next_command_id, 1);

        let portfolio = cache2
            .get_portfolio(&wallet)
            .await
            .expect("portfolio read should self-heal settled ghost position after restore");
        assert!(
            portfolio
                .positions
                .iter()
                .all(|position| position.position.symbol != symbol),
            "settled ghost symbol should be removed after snapshot recovery"
        );
    }

    #[tokio::test]
    async fn test_portfolio_subscription() {
        let context = setup_test_cache().await;
        let cache = context.cache.clone();

        // Subscribe to portfolio updates
        let (_subscriber_id, mut rx) = cache.subscribe(test_wallet(1)).await;

        // Should receive initial snapshot
        let update = rx.recv().await.unwrap();
        match update {
            PortfolioUpdate::Initial { positions, .. } => {
                assert_eq!(positions.len(), 0);
            }
            _ => panic!("Expected Initial update"),
        }

        // Process a fill via engine message
        use hypercall_types::Fill;

        let fill = Fill {
            trade_id: 1,
            taker_order_id: 100,
            maker_order_id: 101,
            symbol: "BTC-CALL-100000".to_string(),
            price: dec!(1000),
            size: to_contract_units_decimal("BTC-CALL-100000", dec!(10)),
            taker_side: Side::Buy,
            taker_wallet_address: test_wallet(1),
            maker_wallet_address: test_wallet(200),
            fee: dec!(0),
            is_taker: true,
            timestamp: 0,
            builder_code_address: None,
            builder_code_fee: None,
            source: Default::default(),
            taker_realized_pnl: None,
            maker_realized_pnl: None,
            underlying_notional: None,
        };

        insert_trade_for_fill(&context, &fill).await;
        cache
            .handle_engine_message(
                EngineMessage::OrderFilled {
                    accounting: hypercall_engine::FillAccounting::from_fill(&fill),
                    fill,
                },
                1,
            )
            .await;

        // Should receive position update
        let update = rx.recv().await.unwrap();
        match update {
            PortfolioUpdate::PositionUpdate { position, .. } => {
                assert_eq!(position.position.symbol, "BTC-CALL-100000");
                assert_eq!(position.position.amount, dec!(10));
                assert_eq!(position.position.realized_pnl, dec!(0));
                assert_eq!(position.position.unrealized_pnl, dec!(0));
            }
            _ => panic!("Expected PositionUpdate"),
        }

        // Should receive balance update
        let update = rx.recv().await.unwrap();
        match update {
            PortfolioUpdate::BalanceUpdate {
                total_margin_used, ..
            } => {
                // Margin used tracked from positions
                let _ = total_margin_used;
            }
            _ => panic!("Expected BalanceUpdate"),
        }
    }

    #[tokio::test]
    async fn test_position_expired_notifies_subscribers_after_projection_update() {
        let context = setup_test_cache().await;
        let cache = context.cache.clone();
        let wallet = test_wallet(1);
        let symbol = "BTC-20261231-100000-C".to_string();

        let (_subscriber_id, mut rx) = cache.subscribe(wallet).await;
        let initial = rx.recv().await.unwrap();
        assert!(matches!(initial, PortfolioUpdate::Initial { .. }));

        let fill = Fill {
            trade_id: 1,
            taker_order_id: 100,
            maker_order_id: 101,
            symbol: symbol.clone(),
            price: dec!(1000),
            size: to_contract_units_decimal(&symbol, dec!(10)),
            taker_side: Side::Buy,
            taker_wallet_address: wallet,
            maker_wallet_address: test_wallet(200),
            fee: dec!(0),
            is_taker: true,
            timestamp: 0,
            builder_code_address: None,
            builder_code_fee: None,
            source: hypercall_types::FillSource::Orderbook,
            taker_realized_pnl: None,
            maker_realized_pnl: None,
            underlying_notional: None,
        };

        insert_trade_for_fill(&context, &fill).await;
        cache
            .handle_engine_message(
                EngineMessage::OrderFilled {
                    accounting: hypercall_engine::FillAccounting::from_fill(&fill),
                    fill,
                },
                1,
            )
            .await;

        let _ = rx.recv().await.unwrap();
        let _ = rx.recv().await.unwrap();

        insert_applied_settlement(
            &context,
            wallet,
            &symbol,
            dec!(10),
            dec!(1000),
            dec!(10000),
            dec!(0),
        )
        .await;
        cache
            .handle_engine_message(
                EngineMessage::PositionExpired(PositionExpiredMessage {
                    wallet_address: wallet,
                    margin_mode: MarginMode::Standard,
                    symbol: symbol.clone(),
                    position_size: dec!(10),
                    settlement_price: dec!(1000),
                    settlement_value: dec!(10000),
                    settlement_entry_price: Some(dec!(1000)),
                    cost_basis: Some(dec!(10000)),
                    net_pnl: Some(dec!(0)),
                    timestamp: 2,
                }),
                2,
            )
            .await;

        match rx.recv().await.unwrap() {
            PortfolioUpdate::PositionUpdate { position, .. } => {
                assert_eq!(position.position.symbol, symbol);
                assert_eq!(position.position.amount, dec!(0));
            }
            other => panic!("Expected PositionUpdate after expiry, got {:?}", other),
        }

        match rx.recv().await.unwrap() {
            PortfolioUpdate::BalanceUpdate {
                total_margin_used, ..
            } => {
                assert_eq!(total_margin_used, dec!(0));
            }
            other => panic!("Expected BalanceUpdate after expiry, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn test_subscription_receives_greeks_snapshot_update() {
        let context = setup_test_cache().await;
        let cache = context.cache.clone();
        attach_test_greeks_cache(&context, "BTC-20261231-50000-C").await;

        let (_subscriber_id, mut rx) = cache.subscribe(test_wallet(1)).await;
        let initial = timeout(Duration::from_secs(2), rx.recv())
            .await
            .expect("timed out waiting for initial update")
            .expect("channel closed");
        assert!(matches!(initial, PortfolioUpdate::Initial { .. }));

        let greeks_update = timeout(Duration::from_secs(2), rx.recv())
            .await
            .expect("timed out waiting for greeks snapshot")
            .expect("channel closed");
        match greeks_update {
            PortfolioUpdate::GreeksUpdate {
                per_leg, aggregate, ..
            } => {
                assert!(per_leg.is_empty(), "expected empty per-leg greeks");
                assert!(aggregate.is_none(), "expected null aggregate greeks");
            }
            other => panic!("expected GreeksUpdate, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn test_fill_triggers_greeks_update() {
        let context = setup_test_cache().await;
        let cache = context.cache.clone();
        let symbol = "BTC-20261231-50000-C";
        attach_test_greeks_cache(&context, symbol).await;

        let (_subscriber_id, mut rx) = cache.subscribe(test_wallet(1)).await;
        let _ = timeout(Duration::from_secs(2), rx.recv())
            .await
            .expect("timed out waiting for initial update")
            .expect("channel closed");
        let _ = timeout(Duration::from_secs(2), rx.recv())
            .await
            .expect("timed out waiting for greeks snapshot")
            .expect("channel closed");

        use hypercall_types::Fill;

        let fill = Fill {
            trade_id: 1001,
            taker_order_id: 100,
            maker_order_id: 101,
            symbol: symbol.to_string(),
            price: dec!(1000),
            size: to_contract_units_decimal(symbol, dec!(2)),
            taker_side: Side::Buy,
            taker_wallet_address: test_wallet(1),
            maker_wallet_address: test_wallet(200),
            fee: dec!(0),
            is_taker: true,
            timestamp: 0,
            builder_code_address: None,
            builder_code_fee: None,
            source: Default::default(),
            taker_realized_pnl: None,
            maker_realized_pnl: None,
            underlying_notional: None,
        };

        insert_trade_for_fill(&context, &fill).await;
        cache
            .handle_engine_message(
                EngineMessage::OrderFilled {
                    accounting: hypercall_engine::FillAccounting::from_fill(&fill),
                    fill,
                },
                10,
            )
            .await;

        let mut saw_position_update = false;
        let mut saw_greeks = false;
        for _ in 0..10 {
            let update = timeout(Duration::from_secs(5), rx.recv())
                .await
                .expect("timed out waiting for update")
                .expect("channel closed");
            match update {
                PortfolioUpdate::PositionUpdate { position, .. } => {
                    assert_eq!(position.position.symbol, symbol);
                    assert!(
                        position.position.unrealized_pnl != dec!(0),
                        "option UPNL should be repriced from theoretical mark"
                    );
                    saw_position_update = true;
                }
                PortfolioUpdate::GreeksUpdate {
                    per_leg, aggregate, ..
                } => {
                    assert!(!per_leg.is_empty(), "expected non-empty per-leg greeks");
                    assert_eq!(per_leg[0].symbol, symbol);
                    assert!(aggregate.is_some(), "expected aggregate greeks");
                    saw_greeks = true;
                    if saw_position_update {
                        break;
                    }
                }
                _ => {}
            }
        }

        assert!(
            saw_position_update,
            "expected a PositionUpdate with theoretical repriced UPNL after fill"
        );
        assert!(saw_greeks, "expected a GreeksUpdate after fill");
    }

    #[tokio::test]
    async fn test_get_portfolio_succeeds_when_theoretical_mark_missing() {
        let context = setup_test_cache().await;
        let cache = context.cache.clone();
        attach_test_greeks_cache(&context, "BTC-20261231-50000-C").await;

        let symbol_missing_mark = "BTC-20261231-51000-C";
        let fill = Fill {
            trade_id: 2001,
            taker_order_id: 100,
            maker_order_id: 101,
            symbol: symbol_missing_mark.to_string(),
            price: dec!(1000),
            size: to_contract_units_decimal(symbol_missing_mark, dec!(1)),
            taker_side: Side::Buy,
            taker_wallet_address: test_wallet(1),
            maker_wallet_address: test_wallet(200),
            fee: dec!(0),
            is_taker: true,
            timestamp: 0,
            builder_code_address: None,
            builder_code_fee: None,
            source: Default::default(),
            taker_realized_pnl: None,
            maker_realized_pnl: None,
            underlying_notional: None,
        };

        insert_trade_for_fill(&context, &fill).await;
        cache
            .handle_engine_message(
                EngineMessage::OrderFilled {
                    accounting: hypercall_engine::FillAccounting::from_fill(&fill),
                    fill,
                },
                11,
            )
            .await;

        // Should succeed even when theoretical marks are missing (best-effort repricing)
        let portfolio = cache
            .get_portfolio(&test_wallet(1))
            .await
            .expect("portfolio should succeed even with missing theoretical marks");
        assert!(!portfolio.positions.is_empty());
    }

    #[tokio::test]
    async fn test_get_portfolio_reprices_option_upnl_with_theoretical_mark() {
        let context = setup_test_cache().await;
        let cache = context.cache.clone();
        let symbol = "BTC-20261231-50000-C";
        attach_test_greeks_cache(&context, symbol).await;

        let fill = Fill {
            trade_id: 2002,
            taker_order_id: 100,
            maker_order_id: 101,
            symbol: symbol.to_string(),
            price: dec!(1000),
            size: to_contract_units_decimal(symbol, dec!(1)),
            taker_side: Side::Buy,
            taker_wallet_address: test_wallet(1),
            maker_wallet_address: test_wallet(200),
            fee: dec!(0),
            is_taker: true,
            timestamp: 0,
            builder_code_address: None,
            builder_code_fee: None,
            source: Default::default(),
            taker_realized_pnl: None,
            maker_realized_pnl: None,
            underlying_notional: None,
        };

        insert_trade_for_fill(&context, &fill).await;
        cache
            .handle_engine_message(
                EngineMessage::OrderFilled {
                    accounting: hypercall_engine::FillAccounting::from_fill(&fill),
                    fill,
                },
                12,
            )
            .await;

        let portfolio = cache
            .get_portfolio(&test_wallet(1))
            .await
            .expect("portfolio fetch should succeed");
        let position = portfolio
            .positions
            .iter()
            .find(|p| p.position.symbol == symbol)
            .expect("position should exist");
        assert!(
            position.position.unrealized_pnl != dec!(0),
            "REST portfolio should expose theoretical repriced UPNL"
        );
    }

    #[tokio::test]
    async fn test_multiple_subscribers() {
        let context = setup_test_cache().await;
        let cache = context.cache.clone();

        // Multiple subscribers for same account
        let (_subscriber_id_1, mut rx1) = cache.subscribe(test_wallet(1)).await;
        let (_subscriber_id_2, mut rx2) = cache.subscribe(test_wallet(1)).await;

        // Both should receive initial snapshot
        let _ = rx1.recv().await.unwrap();
        let _ = rx2.recv().await.unwrap();

        // Process a fill via engine message
        use hypercall_types::Fill;

        let fill = Fill {
            trade_id: 2,
            taker_order_id: 200,
            maker_order_id: 201,
            symbol: "BTC-CALL-100000".to_string(),
            price: dec!(1000),
            size: to_contract_units_decimal("BTC-CALL-100000", dec!(10)),
            taker_side: Side::Buy,
            taker_wallet_address: test_wallet(1),
            maker_wallet_address: test_wallet(200),
            fee: dec!(0),
            is_taker: true,
            timestamp: 0,
            builder_code_address: None,
            builder_code_fee: None,
            source: Default::default(),
            taker_realized_pnl: None,
            maker_realized_pnl: None,
            underlying_notional: None,
        };

        insert_trade_for_fill(&context, &fill).await;
        cache
            .handle_engine_message(
                EngineMessage::OrderFilled {
                    accounting: hypercall_engine::FillAccounting::from_fill(&fill),
                    fill,
                },
                2,
            )
            .await;

        // Both should receive updates
        let update1 = rx1.recv().await.unwrap();
        let update2 = rx2.recv().await.unwrap();

        match (update1, update2) {
            (
                PortfolioUpdate::PositionUpdate { position: pos1, .. },
                PortfolioUpdate::PositionUpdate { position: pos2, .. },
            ) => {
                assert_eq!(pos1.position.realized_pnl, dec!(0));
                assert_eq!(pos1.position.unrealized_pnl, dec!(0));
                assert_eq!(pos2.position.realized_pnl, dec!(0));
                assert_eq!(pos2.position.unrealized_pnl, dec!(0));
            }
            _ => panic!("Expected both subscribers to receive PositionUpdate"),
        }
    }

    #[tokio::test]
    async fn test_seq_updated_after_apply() {
        let context = setup_test_cache().await;
        let cache = context.cache.clone();

        // Initially seq should be 0
        assert_eq!(cache.get_last_processed_seq().await, 0);

        // Process a fill
        use hypercall_types::Fill;

        let fill = Fill {
            trade_id: 42,
            taker_order_id: 100,
            maker_order_id: 101,
            symbol: "BTC-CALL-100000".to_string(),
            price: dec!(1000),
            size: to_contract_units_decimal("BTC-CALL-100000", dec!(10)),
            taker_side: Side::Buy,
            taker_wallet_address: test_wallet(1),
            maker_wallet_address: test_wallet(200),
            fee: dec!(0),
            is_taker: true,
            timestamp: 0,
            builder_code_address: None,
            builder_code_fee: None,
            source: Default::default(),
            taker_realized_pnl: None,
            maker_realized_pnl: None,
            underlying_notional: None,
        };

        insert_trade_for_fill(&context, &fill).await;
        cache
            .handle_engine_message(
                EngineMessage::OrderFilled {
                    accounting: hypercall_engine::FillAccounting::from_fill(&fill),
                    fill,
                },
                42,
            )
            .await;

        // Seq should now be 42
        assert_eq!(cache.get_last_processed_seq().await, 42);

        // Portfolio should reflect the fill
        let portfolio = cache.get_portfolio(&test_wallet(1)).await.unwrap();
        assert_eq!(portfolio.positions.len(), 1);
        assert_eq!(portfolio.positions[0].position.amount, dec!(10));
    }

    #[tokio::test]
    async fn test_capture_snapshot_state_is_consistent() {
        let context = setup_test_cache().await;
        let cache = context.cache.clone();

        // Process two fills
        use hypercall_types::Fill;

        let fill1 = Fill {
            trade_id: 1,
            taker_order_id: 100,
            maker_order_id: 101,
            symbol: "BTC-CALL-100000".to_string(),
            price: dec!(1000),
            size: to_contract_units_decimal("BTC-CALL-100000", dec!(10)),
            taker_side: Side::Buy,
            taker_wallet_address: test_wallet(1),
            maker_wallet_address: test_wallet(200),
            fee: dec!(0),
            is_taker: true,
            timestamp: 0,
            builder_code_address: None,
            builder_code_fee: None,
            source: Default::default(),
            taker_realized_pnl: None,
            maker_realized_pnl: None,
            underlying_notional: None,
        };

        insert_trade_for_fill(&context, &fill1).await;
        cache
            .handle_engine_message(
                EngineMessage::OrderFilled {
                    accounting: hypercall_engine::FillAccounting::from_fill(&fill1),
                    fill: fill1,
                },
                1,
            )
            .await;

        // Capture snapshot state
        let (seq, portfolios) = cache.capture_snapshot_state().await;

        // The snapshot should be consistent: seq matches state
        assert_eq!(seq, 1);
        assert!(portfolios.contains_key(&test_wallet(1)));
        let balance = portfolios.get(&test_wallet(1)).unwrap();
        assert!(balance.positions.contains_key("BTC-CALL-100000"));
        assert_eq!(
            balance.positions.get("BTC-CALL-100000").unwrap().amount,
            dec!(10)
        );
    }

    #[test]
    fn test_normalize_test_database_url_repairs_duplicate_query_separator() {
        let normalized = normalize_test_database_url(
            "postgresql://user:pass@localhost:5432/test_db?sslmode=disable?sslmode=require",
            Some("isolated_db"),
        );

        assert_eq!(
            normalized,
            "postgresql://user:pass@localhost:5432/isolated_db?sslmode=disable"
        );
    }

    #[test]
    fn test_normalize_test_database_url_preserves_authority_verbatim() {
        let normalized = normalize_test_database_url(
            "postgresql://user:pa%40ss@shared-db.internal:5432/test_db?sslmode=require",
            Some("isolated_db"),
        );

        assert_eq!(
            normalized,
            "postgresql://user:pa%40ss@shared-db.internal:5432/isolated_db?sslmode=require"
        );
    }

    #[tokio::test]
    async fn test_publish_margin_updates_for_subscribers_counts_unique_wallets() {
        let context = setup_test_cache().await;
        let cache = context.cache.clone();

        // Two subscribers on wallet A.
        let (_subscriber_id_a1, mut rx_a1) = cache.subscribe(test_wallet(1)).await;
        let (_subscriber_id_a2, mut rx_a2) = cache.subscribe(test_wallet(1)).await;
        // One subscriber on wallet B.
        let (_subscriber_id_b1, mut rx_b1) = cache.subscribe(test_wallet(2)).await;

        // Drain Initial messages.
        let _ = rx_a1.recv().await.unwrap();
        let _ = rx_a2.recv().await.unwrap();
        let _ = rx_b1.recv().await.unwrap();

        let attempted = cache.publish_margin_updates_for_subscribers().await;
        assert_eq!(attempted, 2, "should attempt unique wallets only");
    }

    #[tokio::test]
    async fn test_publish_margin_updates_for_subscribers_empty() {
        let context = setup_test_cache().await;
        let cache = context.cache.clone();

        let attempted = cache.publish_margin_updates_for_subscribers().await;
        assert_eq!(attempted, 0, "no subscribers should attempt zero wallets");
    }

    #[tokio::test]
    async fn test_dropped_subscriber_is_pruned_on_notify() {
        let context = setup_test_cache().await;
        let cache = context.cache.clone();
        let wallet = test_wallet(1);

        let (_subscriber_id, mut rx) = cache.subscribe(wallet).await;
        let _ = rx.recv().await.unwrap();
        drop(rx);

        cache
            .notify_subscribers(
                &wallet,
                PortfolioUpdate::BalanceUpdate {
                    total_margin_used: dec!(0),
                    timestamp: chrono::Utc::now().timestamp(),
                },
            )
            .await;

        let subscribers = cache.subscribers.read().await;
        assert!(
            subscribers.get(&wallet).is_none(),
            "dropped subscriber should be removed after send failure"
        );
    }
}