hypercall_api/handlers/

admin.rs

#[cfg(feature = "test-endpoints")]
use std::time::Instant;

#[cfg(feature = "test-endpoints")]
use rust_decimal_macros::dec;
#[cfg(feature = "test-endpoints")]
use utoipa::ToSchema;

use crate::sonic_json::SonicJson;
use crate::{
    error::ApiError,
    middleware::SignerContext,
    models::{
        DeleteUserTierRequest, GetUserTierQuery, SetUserTierRequest, UserTierData, UserTierResponse,
    },
};
use axum::extract::{Query, State};
use axum::http::StatusCode;
use hypercall_db::UserTierUpdate;
use hypercall_engine::command::{
    AccruePmSettlementInterestCommand, JournalPmRecoveryPlanCommand,
    MarkPmRecoveryActionSubmittedCommand, PmRecoveryActionResult, PmRecoveryExternalKind,
    PmRecoveryPlanCommand, ResolvePmRecoveryActionCommand, SetPmSettlementPoolConfigCommand,
};
use hypercall_margin::portfolio::PmSettlementPoolConfig;
#[cfg(feature = "test-endpoints")]
use hypercall_runtime_api::increment_pending_requests;
#[cfg(feature = "test-endpoints")]
use hypercall_types::utils::get_timestamp_millis;
#[cfg(feature = "test-endpoints")]
use hypercall_types::Side;
#[cfg(feature = "test-endpoints")]
use hypercall_types::{OrderAction, OrderActionMessage, OrderInfo, OrderUpdateStatus, TimeInForce};
use serde::{Deserialize, Serialize};

use super::AppState;

#[derive(Debug, Deserialize)]
pub struct SetPmSettlementPoolConfigRequest {
    pub request_id: uuid::Uuid,
    pub input_digest: String,
    pub underlying: String,
    pub config_version: u32,
    pub config: PmSettlementPoolConfig,
    pub timestamp_ms: u64,
}

#[derive(Debug, Deserialize)]
pub struct AccruePmSettlementInterestRequest {
    pub request_id: uuid::Uuid,
    pub input_digest: String,
    pub wallet: hypercall_types::WalletAddress,
    pub underlying: String,
    pub to_ms: i64,
    pub timestamp_ms: u64,
}

#[derive(Debug, Deserialize)]
pub struct JournalPmRecoveryPlanRequest {
    pub request_id: uuid::Uuid,
    pub input_digest: String,
    pub plan: PmRecoveryPlanCommand,
    pub timestamp_ms: u64,
}

#[derive(Debug, Deserialize)]
pub struct MarkPmRecoveryActionSubmittedRequest {
    pub request_id: uuid::Uuid,
    pub input_digest: String,
    pub wallet: hypercall_types::WalletAddress,
    pub plan_id: String,
    pub action_index: u32,
    pub attempt: u32,
    pub external_id: String,
    pub external_kind: PmRecoveryExternalKind,
    pub timestamp_ms: u64,
}

#[derive(Debug, Deserialize)]
pub struct ResolvePmRecoveryActionRequest {
    pub request_id: uuid::Uuid,
    pub input_digest: String,
    pub wallet: hypercall_types::WalletAddress,
    pub plan_id: String,
    pub action_index: u32,
    pub attempt: u32,
    pub result: PmRecoveryActionResult,
    pub recovered_usdc: rust_decimal::Decimal,
    pub liability_reduction_usdc: rust_decimal::Decimal,
    pub result_external_id: Option<String>,
    pub timestamp_ms: u64,
}

#[derive(Debug, Serialize)]
pub struct PmRecoveryProjectionResponse {
    pub plans: Vec<hypercall_db::PmRecoveryPlanProjection>,
    pub actions: Vec<hypercall_db::PmRecoveryActionProjection>,
}

#[derive(Debug, Serialize)]
pub struct PmSettlementGateStateResponse {
    pub portfolio_margin_pool_enabled: bool,
    pub allowlist_count: usize,
    pub allowlist_wallets: Vec<hypercall_types::WalletAddress>,
    pub pools: Vec<PmSettlementGatePoolState>,
    pub account_blockers: PmSettlementAccountBlockers,
    pub recovery_actions: PmSettlementRecoveryActionGateState,
    pub projection_freshness: PmSettlementProjectionFreshness,
    pub expansion_ready: PmSettlementExpansionReadiness,
}

#[derive(Debug, Serialize)]
pub struct PmSettlementGatePoolState {
    pub underlying: String,
    pub pool_available_usdc: rust_decimal::Decimal,
    pub pool_target_usdc: rust_decimal::Decimal,
    pub pool_capacity_usdc: rust_decimal::Decimal,
    pub pool_utilization: Option<rust_decimal::Decimal>,
    pub active_timing_bridge_usdc: rust_decimal::Decimal,
    pub active_settlement_debt_usdc: rust_decimal::Decimal,
    pub normal_utilization_cap: Option<rust_decimal::Decimal>,
    pub crisis_utilization_cap: Option<rust_decimal::Decimal>,
    pub below_target: bool,
    pub utilization_unavailable: bool,
    pub above_crisis_cap: bool,
    pub projection_seq: i64,
    pub last_engine_command_id: i64,
}

#[derive(Debug, Default, Serialize)]
pub struct PmSettlementAccountBlockers {
    pub total_accounts: usize,
    pub bridged_accounts: usize,
    pub debt_accounts: usize,
    pub overdue_bridge_accounts: usize,
    pub active_recovery_accounts: usize,
    pub active_bridge_usdc: rust_decimal::Decimal,
    pub active_debt_usdc: rust_decimal::Decimal,
}

#[derive(Debug, Default, Serialize)]
pub struct PmSettlementRecoveryActionGateState {
    pub total_actions: usize,
    pub planned_actions: usize,
    pub submitted_actions: usize,
    pub terminal_actions: usize,
}

#[derive(Debug, Default, Serialize)]
pub struct PmSettlementProjectionFreshness {
    pub projection_rows: usize,
    pub max_projection_seq: i64,
    pub max_engine_command_id: i64,
}

#[derive(Debug, Serialize)]
pub struct PmSettlementExpansionReadiness {
    pub allowlist_configured: bool,
    pub pool_facts_available: bool,
    pub projections_present: bool,
    pub no_pool_below_target: bool,
    pub no_pool_above_crisis_cap: bool,
    pub no_account_debt: bool,
    pub no_active_recovery_accounts: bool,
    pub no_overdue_bridge: bool,
    pub no_planned_recovery_actions: bool,
    pub no_submitted_recovery_actions: bool,
    pub ready_for_single_wallet_smoke: bool,
    pub ready_for_allowlist_expansion: bool,
}

// ===== MMP (Market Maker Protection) Handlers =====

/// Set or update MMP configuration
#[utoipa::path(
    post,
    path = "/mmp/config",
    request_body = SetMmpConfigRequest,
    responses(
        (status = 200, description = "MMP config updated", body = MmpConfigData),
        (status = 403, description = "Wallet mismatch"),
        (status = 500, description = "Internal server error")
    ),
    security(("eip712_signature" = [])),
    tag = "MMP"
)]
pub async fn set_mmp_config(
    State(state): State<AppState>,
    signer_ctx: SignerContext,
    SonicJson(request): SonicJson<crate::models::SetMmpConfigRequest>,
) -> Result<SonicJson<crate::models::ApiResponse<crate::models::MmpConfigData>>, ApiError> {
    tracing::info!(
        "Setting MMP config for wallet: {} (signed by: {}), currency: {}",
        signer_ctx.wallet_address,
        signer_ctx.signer_address,
        request.currency
    );

    // Validate wallet matches signer
    if request.wallet != signer_ctx.wallet_address {
        tracing::warn!("Wallet mismatch in MMP config request");
        return Err(ApiError::forbidden(
            "Wallet mismatch: signer does not match request wallet",
        ));
    }

    // Create config from request
    let new_config = hypercall_db::MmpConfigRecord {
        wallet_address: request.wallet,
        currency: request.currency.clone(),
        interval_ms: request.interval_ms,
        frozen_time_ms: request.frozen_time_ms,
        qty_limit: request.qty_limit,
        delta_limit: request.delta_limit,
        vega_limit: request.vega_limit,
        enabled: true,
        created_at: None,
        updated_at: None,
    };

    // Save to MMP cache (which handles DB persistence)
    state.mmp_cache.set_config(new_config).await.map_err(|e| {
        tracing::error!("Failed to save MMP config: {}", e);
        ApiError::internal_error("Failed to save MMP config")
    })?;

    let config_data = crate::models::MmpConfigData {
        wallet_address: request.wallet,
        currency: request.currency,
        interval_ms: request.interval_ms,
        frozen_time_ms: request.frozen_time_ms,
        qty_limit: request.qty_limit,
        delta_limit: request.delta_limit,
        vega_limit: request.vega_limit,
        enabled: true,
    };

    Ok(SonicJson(crate::models::ApiResponse::success(config_data)))
}

/// Get MMP configuration(s) for a wallet
#[utoipa::path(
    get,
    path = "/mmp/config",
    params(
        ("wallet" = String, Query, description = "Wallet address"),
        ("currency" = Option<String>, Query, description = "Currency filter")
    ),
    responses(
        (status = 200, description = "MMP configuration(s)", body = MmpConfigResponse),
        (status = 500, description = "Internal server error")
    ),
    tag = "MMP"
)]
pub async fn get_mmp_config(
    State(state): State<AppState>,
    Query(query): Query<crate::models::GetMmpConfigQuery>,
) -> Result<SonicJson<crate::models::MmpConfigResponse>, ApiError> {
    tracing::info!("Getting MMP config for wallet: {}", query.wallet);

    let configs = if let Some(currency) = query.currency {
        // Get specific currency config
        if let Some(config) = state.mmp_cache.get_config(&query.wallet, &currency).await {
            vec![config]
        } else {
            vec![]
        }
    } else {
        // Get all configs for wallet
        state.mmp_cache.get_configs_for_wallet(&query.wallet).await
    };

    let data: Vec<crate::models::MmpConfigData> = configs
        .into_iter()
        .map(|c| crate::models::MmpConfigData {
            wallet_address: c.wallet_address,
            currency: c.currency,
            interval_ms: c.interval_ms,
            frozen_time_ms: c.frozen_time_ms,
            qty_limit: c.qty_limit,
            delta_limit: c.delta_limit,
            vega_limit: c.vega_limit,
            enabled: c.enabled,
        })
        .collect();

    Ok(SonicJson(crate::models::MmpConfigResponse {
        success: true,
        data,
    }))
}

/// Delete (disable) MMP configuration
#[utoipa::path(
    delete,
    path = "/mmp/config",
    request_body = DeleteMmpConfigRequest,
    responses(
        (status = 200, description = "MMP config deleted"),
        (status = 403, description = "Wallet mismatch"),
        (status = 500, description = "Internal server error")
    ),
    security(("eip712_signature" = [])),
    tag = "MMP"
)]
pub async fn delete_mmp_config(
    State(state): State<AppState>,
    signer_ctx: SignerContext,
    SonicJson(request): SonicJson<crate::models::DeleteMmpConfigRequest>,
) -> Result<SonicJson<crate::models::ApiResponse<String>>, ApiError> {
    tracing::info!(
        "Deleting MMP config for wallet: {} (signed by: {}), currency: {}",
        signer_ctx.wallet_address,
        signer_ctx.signer_address,
        request.currency
    );

    // Validate wallet matches signer
    if request.wallet != signer_ctx.wallet_address {
        tracing::warn!("Wallet mismatch in delete MMP config request");
        return Err(ApiError::forbidden(
            "Wallet mismatch: signer does not match request wallet",
        ));
    }

    state
        .mmp_cache
        .delete_config(&request.wallet, &request.currency)
        .await
        .map_err(|e| {
            tracing::error!("Failed to delete MMP config: {}", e);
            ApiError::internal_error("Failed to delete MMP config")
        })?;

    Ok(SonicJson(crate::models::ApiResponse::success(
        "MMP config deleted successfully".to_string(),
    )))
}

/// Reset MMP state (clear fills and unfreeze)
#[utoipa::path(
    post,
    path = "/mmp/reset",
    request_body = ResetMmpRequest,
    responses(
        (status = 200, description = "MMP state reset"),
        (status = 403, description = "Wallet mismatch"),
        (status = 500, description = "Internal server error")
    ),
    security(("eip712_signature" = [])),
    tag = "MMP"
)]
pub async fn reset_mmp(
    State(state): State<AppState>,
    signer_ctx: SignerContext,
    SonicJson(request): SonicJson<crate::models::ResetMmpRequest>,
) -> Result<SonicJson<crate::models::ApiResponse<String>>, ApiError> {
    tracing::info!(
        "Resetting MMP for wallet: {} (signed by: {}), currency: {}",
        signer_ctx.wallet_address,
        signer_ctx.signer_address,
        request.currency
    );

    // Validate wallet matches signer
    if request.wallet != signer_ctx.wallet_address {
        tracing::warn!("Wallet mismatch in reset MMP request");
        return Err(ApiError::forbidden(
            "Wallet mismatch: signer does not match request wallet",
        ));
    }

    state
        .mmp_cache
        .reset_mmp(&request.wallet, &request.currency)
        .await;

    Ok(SonicJson(crate::models::ApiResponse::success(
        "MMP state reset successfully".to_string(),
    )))
}

// ===== User Tier Handlers =====

/// Get user tier for a wallet (public endpoint)
#[utoipa::path(
    get,
    path = "/user/tier",
    params(
        ("wallet" = String, Query, description = "Wallet address")
    ),
    responses(
        (status = 200, description = "User tier data", body = UserTierResponse),
        (status = 500, description = "Internal server error")
    ),
    tag = "Admin"
)]
pub async fn get_user_tier(
    State(state): State<AppState>,
    Query(query): Query<GetUserTierQuery>,
) -> Result<SonicJson<UserTierResponse>, ApiError> {
    tracing::info!("Getting user tier for wallet: {}", query.wallet);

    let tier = state
        .tier_cache
        .get_tier(&query.wallet)
        .await
        .unwrap_or_else(|| UserTierData {
            wallet_address: query.wallet,
            tier: "tier2".to_string(), // Default is tier2 (unrestricted)
        });

    Ok(SonicJson(UserTierResponse {
        success: true,
        data: tier,
    }))
}

/// Set user tier (admin only - requires signature)
#[utoipa::path(
    post,
    path = "/user/tier",
    request_body = SetUserTierRequest,
    responses(
        (status = 200, description = "User tier updated", body = UserTierData),
        (status = 400, description = "Invalid tier value"),
        (status = 500, description = "Internal server error")
    ),
    security(("eip712_signature" = [])),
    tag = "Admin"
)]
pub async fn set_user_tier(
    State(state): State<AppState>,
    signer_ctx: SignerContext,
    SonicJson(request): SonicJson<SetUserTierRequest>,
) -> Result<SonicJson<crate::models::ApiResponse<UserTierData>>, ApiError> {
    // TODO: Add admin check - verify signer is in admin list
    // For now, just validate signature is present

    tracing::info!(
        "Setting user tier for wallet: {} (signed by: {}), tier: {}",
        request.wallet,
        signer_ctx.signer_address,
        request.tier
    );

    // Validate tier value
    if request.tier != "tier1" && request.tier != "tier2" {
        tracing::error!("Invalid tier value: {}", request.tier);
        return Err(ApiError::bad_request(format!(
            "Invalid tier value: {}. Must be 'tier1' or 'tier2'",
            request.tier
        )));
    }

    let previous_tier = state
        .tier_cache
        .get_tier_record(&request.wallet)
        .await
        .map_err(|e| {
            tracing::error!("Failed to load current user tier: {}", e);
            ApiError::internal_error("Failed to load current user tier")
        })?;

    let new_tier = UserTierUpdate {
        wallet_address: request.wallet,
        tier: request.tier.clone(),
        margin_mode: None,             // Keep existing margin_mode
        version: None,                 // Keep existing version
        max_open_orders: None,         // Use default
        max_open_positions: None,      // Use default
        orders_per_minute: None,       // Use default
        cancels_per_minute: None,      // Use default
        api_requests_per_minute: None, // Use default
    };

    state.tier_cache.set_tier(new_tier).await.map_err(|e| {
        tracing::error!("Failed to save user tier: {}", e);
        ApiError::internal_error("Failed to save user tier")
    })?;

    let wallet = request.wallet;
    if let Err(e) = super::submit_tier_update_command(&state, wallet).await {
        tracing::error!("Failed to apply user tier update in engine: {}", e);
        if let Err(rollback_err) = state
            .tier_cache
            .restore_tier_record(&wallet, previous_tier.as_ref())
            .await
        {
            tracing::error!(
                wallet = %wallet,
                error = %rollback_err,
                "Failed to roll back tier after engine apply failure"
            );
        }
        return Err(ApiError::internal_error("Failed to apply user tier update"));
    }

    let tier_data = UserTierData {
        wallet_address: wallet,
        tier: request.tier,
    };

    Ok(SonicJson(crate::models::ApiResponse::success(tier_data)))
}

/// Delete user tier (admin only)
#[utoipa::path(
    delete,
    path = "/user/tier",
    request_body = DeleteUserTierRequest,
    responses(
        (status = 200, description = "User tier deleted"),
        (status = 500, description = "Internal server error")
    ),
    security(("eip712_signature" = [])),
    tag = "Admin"
)]
pub async fn delete_user_tier(
    State(state): State<AppState>,
    signer_ctx: SignerContext,
    SonicJson(request): SonicJson<DeleteUserTierRequest>,
) -> Result<SonicJson<crate::models::ApiResponse<String>>, ApiError> {
    // TODO: Add admin check

    tracing::info!(
        "Resetting user tier for wallet: {} (signed by: {})",
        request.wallet,
        signer_ctx.signer_address
    );

    let previous_tier = state
        .tier_cache
        .get_tier_record(&request.wallet)
        .await
        .map_err(|e| {
            tracing::error!("Failed to load current user tier: {}", e);
            ApiError::internal_error("Failed to load current user tier")
        })?;

    state
        .tier_cache
        .delete_tier(&request.wallet)
        .await
        .map_err(|e| {
            tracing::error!("Failed to reset user tier: {}", e);
            ApiError::internal_error("Failed to reset user tier")
        })?;

    if let Err(e) = super::submit_tier_update_command(&state, request.wallet).await {
        tracing::error!("Failed to apply user tier deletion in engine: {}", e);
        if let Err(rollback_err) = state
            .tier_cache
            .restore_tier_record(&request.wallet, previous_tier.as_ref())
            .await
        {
            tracing::error!(
                wallet = %request.wallet,
                error = %rollback_err,
                "Failed to roll back tier after engine apply failure"
            );
        }
        return Err(ApiError::internal_error(
            "Failed to apply user tier deletion",
        ));
    }

    Ok(SonicJson(crate::models::ApiResponse::success(
        "User tier reset successfully (reverted to default)".to_string(),
    )))
}

// =============================================================================
// Cancel All Orders Endpoint (Testnet Only)
// =============================================================================

#[cfg(feature = "test-endpoints")]
/// Response for cancel_all_orders endpoint
#[derive(Debug, Serialize, ToSchema)]
pub struct CancelAllOrdersResponse {
    /// Total MM orders found to cancel
    pub total_mm_orders: usize,
    /// Message indicating background processing started
    pub message: String,
}

#[cfg(feature = "test-endpoints")]
/// Cancel all market-maker orders across all wallets (testnet only)
///
/// Cleans up orphan orders left by previous MM instances that used
/// ephemeral wallets. Only available when `modes.enable_test_endpoints=true`.
///
/// Identifies MM orders by `client_id` matching the `cli_\d+` pattern
/// (from `wallet_client.rs`: `format!("cli_{}", nonce)`). User orders
/// placed via the frontend are left untouched.
#[utoipa::path(
    post,
    path = "/test/cancel-all-orders",
    responses(
        (status = 200, description = "MM orders cancelled", body = ApiResponse<CancelAllOrdersResponse>),
        (status = 403, description = "Not in testnet mode"),
        (status = 500, description = "Internal server error")
    ),
    tag = "testnet"
)]
pub async fn cancel_all_orders(
    State(state): State<AppState>,
) -> Result<SonicJson<crate::models::ApiResponse<CancelAllOrdersResponse>>, ApiError> {
    let all_orders = state.order_snapshot.get_all_orders();

    // Diagnostic: also check quote provider for symbols with quotes but no orders
    {
        let all_quotes = state.quote_provider.all_quotes();
        let order_symbols: std::collections::HashSet<String> =
            all_orders.iter().map(|(o, _)| o.symbol.clone()).collect();
        let mut quoted_no_orders = Vec::new();
        for (symbol, quote) in &all_quotes {
            let has_depth = !quote.bids.is_empty() || !quote.asks.is_empty();
            if has_depth && !order_symbols.contains(symbol) {
                quoted_no_orders.push(format!(
                    "{}(bids={},asks={})",
                    symbol,
                    quote.bids.len(),
                    quote.asks.len()
                ));
            }
        }
        if !quoted_no_orders.is_empty() {
            tracing::warn!(
                count = quoted_no_orders.len(),
                symbols = ?quoted_no_orders,
                "cancel_all_orders: symbols have orderbook depth but NO orders in snapshot"
            );
        }
        tracing::info!(
            total_orders = all_orders.len(),
            total_quoted_symbols = all_quotes.len(),
            "cancel_all_orders: snapshot state"
        );
    }

    // For orders with unknown client_id (None), look up the DB for the real value.
    // This handles orders restored from WAL snapshots that lost metadata.
    let unknown_order_ids: Vec<i64> = all_orders
        .iter()
        .filter(|(o, _)| o.client_id.is_none())
        .filter_map(|(o, _)| i64::try_from(o.order_id).ok())
        .collect();

    let db_client_ids: std::collections::HashMap<i64, Option<String>> = if !unknown_order_ids
        .is_empty()
    {
        {
            let analytics: &dyn hypercall_db::AnalyticsReader = state.db.as_ref();
            match analytics
                .get_client_ids_by_order_ids(&unknown_order_ids)
                .await
            {
                Ok(map) => {
                    tracing::info!(
                        unknown_count = unknown_order_ids.len(),
                        db_found = map.len(),
                        "cancel_all_orders: resolved client_ids from DB"
                    );
                    map
                }
                Err(e) => {
                    tracing::warn!(
                        error = %e,
                        "cancel_all_orders: DB lookup failed, falling back to in-memory client_id"
                    );
                    std::collections::HashMap::new()
                }
            }
        }
    } else {
        std::collections::HashMap::new()
    };

    // Determine effective client_id: prefer in-memory, fall back to DB.
    let is_mm_order = |order: &hypercall_runtime_api::RuntimeOrderSummary| -> bool {
        let effective_client_id = order.client_id.clone().or_else(|| {
            // Try DB lookup for orders with unknown client_id
            i64::try_from(order.order_id)
                .ok()
                .and_then(|id| db_client_ids.get(&id))
                .cloned()
                .flatten()
        });

        match &effective_client_id {
            Some(cid) => cid.starts_with("cli_") && cid[4..].bytes().all(|b| b.is_ascii_digit()),
            None => {
                // No client_id in memory OR DB — could be a frontend order
                // placed without client_id (valid input) or a very old order.
                // Do NOT cancel — only cancel orders positively identified as MM.
                false
            }
        }
    };

    // Log orders that DON'T match the MM heuristic
    for (order, wallet) in &all_orders {
        if !is_mm_order(order) {
            tracing::info!(
                order_id = order.order_id,
                symbol = %order.symbol,
                client_id = ?order.client_id,
                wallet = %wallet,
                "cancel_all_orders: skipping non-MM order"
            );
        }
    }

    // Filter to MM orders using DB-backed client_id resolution
    let mm_orders: Vec<_> = all_orders
        .into_iter()
        .filter(|(order, _wallet)| is_mm_order(order))
        .collect();

    let total = mm_orders.len();
    tracing::info!(
        "cancel_all_orders: found {} MM/orphan orders to cancel, spawning background task",
        total
    );

    let order_sender = state.order_sender.clone();
    tokio::spawn(async move {
        let mut succeeded = 0usize;
        let mut failed = 0usize;

        for (i, (order, wallet)) in mm_orders.iter().enumerate() {
            let order_id = order.order_id;
            let order_info = OrderInfo {
                symbol: String::new(),
                price: dec!(0),
                size: dec!(0),
                side: Side::Buy,
                tif: TimeInForce::GTC,
                client_id: None,
                order_id: Some(order_id),
                is_perp: false,
                underlying: None,
                reduce_only: None,
                nonce: None,
                signature: None,
                mmp_enabled: false,
                builder_code_address: None,
            };

            let order_action_msg = OrderActionMessage {
                timestamp: get_timestamp_millis(),
                info: order_info,
                action: OrderAction::CancelOrder,
                wallet: *wallet,
                api_wallet_address: Some(*wallet),
                mmp_triggered: false,
                request_id: Some(uuid::Uuid::now_v7().to_string()),
            };

            let (response_tx, mut response_rx) = tokio::sync::mpsc::channel(1);

            let engine_request = hypercall_runtime_api::UnifiedEngineRequest {
                message: order_action_msg,
                response_tx,
                enqueued_at: Instant::now(),
                #[cfg(feature = "otel-tracing")]
                trace_context: None,
            };

            increment_pending_requests();
            if order_sender.send(engine_request).await.is_err() {
                tracing::error!(
                    "cancel_all_orders: failed to send cancel for order {}",
                    order_id
                );
                failed += 1;
                continue;
            }

            match tokio::time::timeout(std::time::Duration::from_secs(5), response_rx.recv()).await
            {
                Ok(Some(resp)) if resp.status == OrderUpdateStatus::Canceled => {
                    succeeded += 1;
                }
                Ok(Some(resp)) => {
                    tracing::warn!(
                        "cancel_all_orders: order {} got status {:?}",
                        order_id,
                        resp.status
                    );
                    failed += 1;
                }
                _ => {
                    tracing::warn!(
                        "cancel_all_orders: timeout/no response for order {}",
                        order_id
                    );
                    failed += 1;
                }
            }

            if (i + 1) % 1000 == 0 {
                tracing::info!(
                    "cancel_all_orders: progress {}/{} (succeeded={}, failed={})",
                    i + 1,
                    total,
                    succeeded,
                    failed
                );
            }
        }

        tracing::info!(
            "cancel_all_orders: done. sent={}, succeeded={}, failed={}",
            total,
            succeeded,
            failed
        );
    });

    Ok(SonicJson(crate::models::ApiResponse::success(
        CancelAllOrdersResponse {
            total_mm_orders: total,
            message: format!("Spawned background task to cancel {} MM orders", total),
        },
    )))
}

fn ensure_pm_settlement_pool_enabled(state: &AppState) -> Result<(), ApiError> {
    if state.runtime_config.portfolio_margin_pool_enabled {
        Ok(())
    } else {
        Err(ApiError::new(
            StatusCode::SERVICE_UNAVAILABLE,
            "portfolio_margin_pool_disabled",
            "portfolio_margin_pool_enabled is false; PM settlement pool mutations are disabled",
        ))
    }
}

fn ensure_pm_settlement_wallet_allowed(
    state: &AppState,
    wallet: &hypercall_types::WalletAddress,
) -> Result<(), ApiError> {
    if state
        .runtime_config
        .portfolio_margin_settlement_allowlist
        .contains(wallet)
    {
        Ok(())
    } else {
        Err(ApiError::new(
            StatusCode::FORBIDDEN,
            "portfolio_margin_wallet_not_allowlisted",
            "wallet is not allowlisted for PM settlement pool recovery commands",
        ))
    }
}

async fn submit_pm_settlement_command(
    state: &AppState,
    command: hypercall_engine::command::EngineCommand,
) -> Result<(), ApiError> {
    let sender = state
        .pm_settlement_admin_sender
        .as_ref()
        .ok_or_else(|| ApiError::internal_error("PM settlement admin channel is not configured"))?;
    let (tx, rx) = tokio::sync::oneshot::channel();
    sender
        .send(hypercall_runtime_api::PmSettlementAdminRequest {
            command,
            applied_tx: tx,
        })
        .await
        .map_err(|_| ApiError::internal_error("PM settlement admin channel is closed"))?;
    tokio::time::timeout(super::ENGINE_RESPONSE_TIMEOUT, rx)
        .await
        .map_err(|_| ApiError::gateway_timeout("PM settlement command timed out"))?
        .map_err(|_| ApiError::internal_error("PM settlement command acknowledgement dropped"))?
        .map_err(ApiError::bad_request)
}

pub async fn set_pm_settlement_pool_config(
    State(state): State<AppState>,
    SonicJson(request): SonicJson<SetPmSettlementPoolConfigRequest>,
) -> Result<SonicJson<crate::models::ApiResponse<String>>, ApiError> {
    ensure_pm_settlement_pool_enabled(&state)?;
    submit_pm_settlement_command(
        &state,
        hypercall_engine::command::EngineCommand::SetPmSettlementPoolConfig(
            SetPmSettlementPoolConfigCommand {
                request_id: request.request_id,
                input_digest: request.input_digest,
                underlying: request.underlying,
                config_version: request.config_version,
                config: request.config,
                timestamp_ms: request.timestamp_ms,
            },
        ),
    )
    .await?;
    Ok(SonicJson(crate::models::ApiResponse::success(
        "PM settlement pool config command accepted".to_string(),
    )))
}

pub async fn accrue_pm_settlement_interest(
    State(state): State<AppState>,
    SonicJson(request): SonicJson<AccruePmSettlementInterestRequest>,
) -> Result<SonicJson<crate::models::ApiResponse<String>>, ApiError> {
    ensure_pm_settlement_pool_enabled(&state)?;
    ensure_pm_settlement_wallet_allowed(&state, &request.wallet)?;
    submit_pm_settlement_command(
        &state,
        hypercall_engine::command::EngineCommand::AccruePmSettlementInterest(
            AccruePmSettlementInterestCommand {
                request_id: request.request_id,
                input_digest: request.input_digest,
                wallet: request.wallet,
                underlying: request.underlying,
                to_ms: request.to_ms,
                timestamp_ms: request.timestamp_ms,
            },
        ),
    )
    .await?;
    Ok(SonicJson(crate::models::ApiResponse::success(
        "PM settlement interest accrual command accepted".to_string(),
    )))
}

pub async fn journal_pm_recovery_plan(
    State(state): State<AppState>,
    SonicJson(request): SonicJson<JournalPmRecoveryPlanRequest>,
) -> Result<SonicJson<crate::models::ApiResponse<String>>, ApiError> {
    ensure_pm_settlement_pool_enabled(&state)?;
    ensure_pm_settlement_wallet_allowed(&state, &request.plan.wallet)?;
    submit_pm_settlement_command(
        &state,
        hypercall_engine::command::EngineCommand::JournalPmRecoveryPlan(
            JournalPmRecoveryPlanCommand {
                request_id: request.request_id,
                input_digest: request.input_digest,
                plan: request.plan,
                timestamp_ms: request.timestamp_ms,
            },
        ),
    )
    .await?;
    Ok(SonicJson(crate::models::ApiResponse::success(
        "PM recovery plan journal command accepted".to_string(),
    )))
}

pub async fn mark_pm_recovery_action_submitted(
    State(state): State<AppState>,
    SonicJson(request): SonicJson<MarkPmRecoveryActionSubmittedRequest>,
) -> Result<SonicJson<crate::models::ApiResponse<String>>, ApiError> {
    ensure_pm_settlement_pool_enabled(&state)?;
    ensure_pm_settlement_wallet_allowed(&state, &request.wallet)?;
    submit_pm_settlement_command(
        &state,
        hypercall_engine::command::EngineCommand::MarkPmRecoveryActionSubmitted(
            MarkPmRecoveryActionSubmittedCommand {
                request_id: request.request_id,
                input_digest: request.input_digest,
                wallet: request.wallet,
                plan_id: request.plan_id,
                action_index: request.action_index,
                attempt: request.attempt,
                external_id: request.external_id,
                external_kind: request.external_kind,
                timestamp_ms: request.timestamp_ms,
            },
        ),
    )
    .await?;
    Ok(SonicJson(crate::models::ApiResponse::success(
        "PM recovery action submission command accepted".to_string(),
    )))
}

pub async fn resolve_pm_recovery_action(
    State(state): State<AppState>,
    SonicJson(request): SonicJson<ResolvePmRecoveryActionRequest>,
) -> Result<SonicJson<crate::models::ApiResponse<String>>, ApiError> {
    ensure_pm_settlement_pool_enabled(&state)?;
    ensure_pm_settlement_wallet_allowed(&state, &request.wallet)?;
    submit_pm_settlement_command(
        &state,
        hypercall_engine::command::EngineCommand::ResolvePmRecoveryAction(
            ResolvePmRecoveryActionCommand {
                request_id: request.request_id,
                input_digest: request.input_digest,
                wallet: request.wallet,
                plan_id: request.plan_id,
                action_index: request.action_index,
                attempt: request.attempt,
                result: request.result,
                recovered_usdc: request.recovered_usdc,
                liability_reduction_usdc: request.liability_reduction_usdc,
                result_external_id: request.result_external_id,
                timestamp_ms: request.timestamp_ms,
            },
        ),
    )
    .await?;
    Ok(SonicJson(crate::models::ApiResponse::success(
        "PM recovery action resolution command accepted".to_string(),
    )))
}

pub async fn get_pm_settlement_pools(
    State(state): State<AppState>,
) -> Result<
    SonicJson<crate::models::ApiResponse<Vec<hypercall_db::PmSettlementPoolProjection>>>,
    ApiError,
> {
    ensure_pm_settlement_pool_enabled(&state)?;
    let rows = state.db.list_pm_settlement_pools().await.map_err(|error| {
        tracing::error!("Failed to read PM settlement pool projections: {}", error);
        ApiError::internal_error("Failed to read PM settlement pool projections")
    })?;
    Ok(SonicJson(crate::models::ApiResponse::success(rows)))
}

pub async fn get_pm_settlement_accounts(
    State(state): State<AppState>,
) -> Result<
    SonicJson<crate::models::ApiResponse<Vec<hypercall_db::PmSettlementAccountProjection>>>,
    ApiError,
> {
    ensure_pm_settlement_pool_enabled(&state)?;
    let rows = state
        .db
        .list_pm_settlement_accounts()
        .await
        .map_err(|error| {
            tracing::error!(
                "Failed to read PM settlement account projections: {}",
                error
            );
            ApiError::internal_error("Failed to read PM settlement account projections")
        })?;
    Ok(SonicJson(crate::models::ApiResponse::success(rows)))
}

pub async fn get_pm_settlement_events(
    State(state): State<AppState>,
) -> Result<
    SonicJson<crate::models::ApiResponse<Vec<hypercall_db::PmSettlementEventProjection>>>,
    ApiError,
> {
    ensure_pm_settlement_pool_enabled(&state)?;
    let rows = state
        .db
        .list_pm_settlement_events()
        .await
        .map_err(|error| {
            tracing::error!("Failed to read PM settlement event projections: {}", error);
            ApiError::internal_error("Failed to read PM settlement event projections")
        })?;
    Ok(SonicJson(crate::models::ApiResponse::success(rows)))
}

pub async fn get_pm_settlement_interest_events(
    State(state): State<AppState>,
) -> Result<
    SonicJson<crate::models::ApiResponse<Vec<hypercall_db::PmSettlementInterestEventProjection>>>,
    ApiError,
> {
    ensure_pm_settlement_pool_enabled(&state)?;
    let rows = state
        .db
        .list_pm_settlement_interest_events()
        .await
        .map_err(|error| {
            tracing::error!(
                "Failed to read PM settlement interest event projections: {}",
                error
            );
            ApiError::internal_error("Failed to read PM settlement interest event projections")
        })?;
    Ok(SonicJson(crate::models::ApiResponse::success(rows)))
}

pub async fn get_pm_settlement_repayment_events(
    State(state): State<AppState>,
) -> Result<
    SonicJson<crate::models::ApiResponse<Vec<hypercall_db::PmSettlementRepaymentEventProjection>>>,
    ApiError,
> {
    ensure_pm_settlement_pool_enabled(&state)?;
    let rows = state
        .db
        .list_pm_settlement_repayment_events()
        .await
        .map_err(|error| {
            tracing::error!(
                "Failed to read PM settlement repayment event projections: {}",
                error
            );
            ApiError::internal_error("Failed to read PM settlement repayment event projections")
        })?;
    Ok(SonicJson(crate::models::ApiResponse::success(rows)))
}

pub async fn get_pm_recovery_projections(
    State(state): State<AppState>,
) -> Result<SonicJson<crate::models::ApiResponse<PmRecoveryProjectionResponse>>, ApiError> {
    ensure_pm_settlement_pool_enabled(&state)?;
    let plans = state.db.list_pm_recovery_plans().await.map_err(|error| {
        tracing::error!("Failed to read PM recovery plan projections: {}", error);
        ApiError::internal_error("Failed to read PM recovery plan projections")
    })?;
    let actions = state.db.list_pm_recovery_actions().await.map_err(|error| {
        tracing::error!("Failed to read PM recovery action projections: {}", error);
        ApiError::internal_error("Failed to read PM recovery action projections")
    })?;
    Ok(SonicJson(crate::models::ApiResponse::success(
        PmRecoveryProjectionResponse { plans, actions },
    )))
}

pub async fn get_pm_settlement_gate_state(
    State(state): State<AppState>,
) -> Result<SonicJson<crate::models::ApiResponse<PmSettlementGateStateResponse>>, ApiError> {
    ensure_pm_settlement_pool_enabled(&state)?;

    let pools = state.db.list_pm_settlement_pools().await.map_err(|error| {
        tracing::error!(
            "Failed to read PM settlement pool projections for gate state: {}",
            error
        );
        ApiError::internal_error("Failed to read PM settlement pool projections")
    })?;
    let pool_gate_counts = state
        .db
        .pm_settlement_pool_gate_counts()
        .await
        .map_err(|error| {
            tracing::error!("Failed to read PM settlement pool gate counts: {}", error);
            ApiError::internal_error("Failed to read PM settlement pool gate counts")
        })?;
    let accounts = state
        .db
        .list_pm_settlement_accounts()
        .await
        .map_err(|error| {
            tracing::error!(
                "Failed to read PM settlement account projections for gate state: {}",
                error
            );
            ApiError::internal_error("Failed to read PM settlement account projections")
        })?;
    let recovery_plans = state.db.list_pm_recovery_plans().await.map_err(|error| {
        tracing::error!(
            "Failed to read PM recovery plan projections for gate state: {}",
            error
        );
        ApiError::internal_error("Failed to read PM recovery plan projections")
    })?;
    let recovery_action_rows = state.db.list_pm_recovery_actions().await.map_err(|error| {
        tracing::error!(
            "Failed to read PM recovery action projections for gate state: {}",
            error
        );
        ApiError::internal_error("Failed to read PM recovery action projections")
    })?;
    let now_ms = chrono::Utc::now().timestamp_millis();
    let account_gate_counts = state
        .db
        .pm_settlement_account_gate_counts(now_ms)
        .await
        .map_err(|error| {
            tracing::error!(
                "Failed to read PM settlement account gate counts: {}",
                error
            );
            ApiError::internal_error("Failed to read PM settlement account gate counts")
        })?;
    let recovery_action_gate_counts =
        state
            .db
            .pm_recovery_action_gate_counts()
            .await
            .map_err(|error| {
                tracing::error!("Failed to read PM recovery action gate counts: {}", error);
                ApiError::internal_error("Failed to read PM recovery action gate counts")
            })?;

    let gate_pools = pools
        .iter()
        .map(|pool| {
            let below_target = pool.pool_available_usdc < pool.pool_target_usdc;
            let utilization_unavailable = pool.pool_utilization.is_none();
            let above_crisis_cap = match (pool.pool_utilization, pool.crisis_utilization_cap) {
                (Some(utilization), Some(cap)) => utilization > cap,
                _ => false,
            };
            PmSettlementGatePoolState {
                underlying: pool.underlying.clone(),
                pool_available_usdc: pool.pool_available_usdc,
                pool_target_usdc: pool.pool_target_usdc,
                pool_capacity_usdc: pool.pool_capacity_usdc,
                pool_utilization: pool.pool_utilization,
                active_timing_bridge_usdc: pool.active_timing_bridge_usdc,
                active_settlement_debt_usdc: pool.active_settlement_debt_usdc,
                normal_utilization_cap: pool.normal_utilization_cap,
                crisis_utilization_cap: pool.crisis_utilization_cap,
                below_target,
                utilization_unavailable,
                above_crisis_cap,
                projection_seq: pool.projection_seq,
                last_engine_command_id: pool.last_engine_command_id,
            }
        })
        .collect::<Vec<_>>();

    let account_blockers = pm_settlement_account_blockers(account_gate_counts);
    let recovery_actions = pm_settlement_recovery_action_gate_state(recovery_action_gate_counts);

    let projection_freshness = pm_settlement_projection_freshness(
        &pools,
        &accounts,
        &recovery_plans,
        &recovery_action_rows,
    );
    let pool_facts_available = pool_gate_counts.total_pools > 0
        && pool_gate_counts.missing_utilization_pools == 0
        && pool_gate_counts.missing_crisis_cap_pools == 0;
    let no_pool_below_target = pool_gate_counts.below_target_pools == 0;
    let no_pool_above_crisis_cap = pool_gate_counts.above_crisis_cap_pools == 0;
    let no_account_debt = account_blockers.debt_accounts == 0;
    let no_active_recovery_accounts = account_blockers.active_recovery_accounts == 0;
    let no_overdue_bridge = account_blockers.overdue_bridge_accounts == 0;
    let no_planned_recovery_actions = recovery_actions.planned_actions == 0;
    let no_submitted_recovery_actions = recovery_actions.submitted_actions == 0;
    let allowlist_configured = !state
        .runtime_config
        .portfolio_margin_settlement_allowlist
        .is_empty();
    let projections_present = projection_freshness.projection_rows > 0;
    let ready_for_single_wallet_smoke =
        allowlist_configured && pool_facts_available && projections_present;
    let ready_for_allowlist_expansion = ready_for_single_wallet_smoke
        && no_pool_below_target
        && no_pool_above_crisis_cap
        && no_account_debt
        && no_active_recovery_accounts
        && no_overdue_bridge
        && no_planned_recovery_actions
        && no_submitted_recovery_actions;

    Ok(SonicJson(crate::models::ApiResponse::success(
        PmSettlementGateStateResponse {
            portfolio_margin_pool_enabled: true,
            allowlist_count: state
                .runtime_config
                .portfolio_margin_settlement_allowlist
                .len(),
            allowlist_wallets: state
                .runtime_config
                .portfolio_margin_settlement_allowlist
                .clone(),
            pools: gate_pools,
            account_blockers,
            recovery_actions,
            projection_freshness,
            expansion_ready: PmSettlementExpansionReadiness {
                allowlist_configured,
                pool_facts_available,
                projections_present,
                no_pool_below_target,
                no_pool_above_crisis_cap,
                no_account_debt,
                no_active_recovery_accounts,
                no_overdue_bridge,
                no_planned_recovery_actions,
                no_submitted_recovery_actions,
                ready_for_single_wallet_smoke,
                ready_for_allowlist_expansion,
            },
        },
    )))
}

fn count_to_usize(count: i64) -> usize {
    usize::try_from(count).expect("Postgres COUNT returned an invalid count")
}

fn pm_settlement_account_blockers(
    counts: hypercall_db::PmSettlementAccountGateCounts,
) -> PmSettlementAccountBlockers {
    PmSettlementAccountBlockers {
        total_accounts: count_to_usize(counts.total_accounts),
        bridged_accounts: count_to_usize(counts.bridged_accounts),
        debt_accounts: count_to_usize(counts.debt_accounts),
        overdue_bridge_accounts: count_to_usize(counts.overdue_bridge_accounts),
        active_recovery_accounts: count_to_usize(counts.active_recovery_accounts),
        active_bridge_usdc: counts.active_bridge_usdc,
        active_debt_usdc: counts.active_debt_usdc,
    }
}

fn pm_settlement_recovery_action_gate_state(
    counts: hypercall_db::PmRecoveryActionGateCounts,
) -> PmSettlementRecoveryActionGateState {
    PmSettlementRecoveryActionGateState {
        total_actions: count_to_usize(counts.total_actions),
        planned_actions: count_to_usize(counts.planned_actions),
        submitted_actions: count_to_usize(counts.submitted_actions),
        terminal_actions: count_to_usize(counts.terminal_actions),
    }
}

fn pm_settlement_projection_freshness(
    pools: &[hypercall_db::PmSettlementPoolProjection],
    accounts: &[hypercall_db::PmSettlementAccountProjection],
    plans: &[hypercall_db::PmRecoveryPlanProjection],
    actions: &[hypercall_db::PmRecoveryActionProjection],
) -> PmSettlementProjectionFreshness {
    let projection_rows = pools.len() + accounts.len() + plans.len() + actions.len();
    let max_projection_seq = pools
        .iter()
        .map(|row| row.projection_seq)
        .chain(accounts.iter().map(|row| row.projection_seq))
        .chain(plans.iter().map(|row| row.projection_seq))
        .chain(actions.iter().map(|row| row.projection_seq))
        .max()
        .unwrap_or_default();
    let max_engine_command_id = pools
        .iter()
        .map(|row| row.last_engine_command_id)
        .chain(accounts.iter().map(|row| row.last_engine_command_id))
        .chain(plans.iter().map(|row| row.engine_command_id))
        .chain(actions.iter().map(|row| row.engine_command_id))
        .max()
        .unwrap_or_default();

    PmSettlementProjectionFreshness {
        projection_rows,
        max_projection_seq,
        max_engine_command_id,
    }
}