hypercall/rsm/unified_engine/

rfq_handler.rs

use super::*;
use crate::rsm::apply::MmpFillUpdate;
use hypercall_runtime_api::{RfqExecuteCommand, RfqExecuteRequest, RfqExecuteResult};
use hypercall_types::RFQ_SELF_TRADE_REJECTION_REASON;
use hypercall_types::{to_contract_units_decimal, Fill, FillSource, Side};
use hypercall_types::{EngineMessage, RfqFillLeg, RfqFillMessage};
use metrics::histogram;
use std::time::Instant;

/// Output of `plan_rfq_execution` — the planned execution as a list of
/// `EngineMessage`s plus a fill_id. The caller (the journal pipeline) is
/// responsible for actually persisting and applying these events.
///
/// Building this is **pure**: no DB writes, no portfolio mutations, no
/// WS emissions. Validation that can fail (instruments, MMP, margin,
/// premium-vs-legs cross-check) happens here and returns `Failed` before
/// any events are produced. Once the plan exists, journaling +
/// projection are unconditional.
pub(crate) struct RfqExecutionPlan {
    pub fill_id: String,
    pub events: Vec<EngineMessage>,
    pub mmp_updates: Vec<MmpFillUpdate>,
}

impl UnifiedEngine {
    /// Handle an RFQ execution request within the engine's single-threaded loop.
    ///
    /// The flow:
    /// 1. `plan_rfq_execution` validates the command and produces a list of
    ///    `EngineMessage` events (per-leg `OrderFilled` + summary `RfqFilled`)
    ///    along with post-commit MMP updates. No state mutation, no DB
    ///    writes, no WS emissions yet.
    /// 2. `process_rfq_journaled` runs the events through the same journal
    ///    pipeline as `process_order_journaled`: serialize → batch send →
    ///    wait for journal commit ACK → apply portfolio projection under
    ///    barrier → emit events to WS → ACK the caller.
    /// 3. Post-journal MMP updates fire after the projection barrier
    ///    releases — they're side effects that don't need durability.
    pub(crate) async fn handle_rfq_execute(&mut self, request: RfqExecuteRequest) {
        let handle_start = Instant::now();
        let cmd = &request.command;
        let rfq_span = tracing::info_span!(
            "engine.rfq_execute",
            rfq_id = %cmd.rfq_id,
            taker = %cmd.taker_wallet,
            qp = %cmd.qp_wallet,
            request_id = %cmd.request_id,
        );
        let _guard = rfq_span.enter();

        // Idempotency fast-path: check before apply() to avoid
        // re-planning and double-advancing the hash chain.
        if !cmd.request_id.is_empty() {
            if let Some(cached_fill_id) = self.lookup_cached_rfq_response(&cmd.request_id).await {
                debug!(
                    "RFQ idempotency hit for rfq_id={}, request_id={}, cached fill_id={}",
                    cmd.rfq_id, cmd.request_id, cached_fill_id
                );
                let _ = request.response_tx.send(RfqExecuteResult::Success {
                    fill_id: cached_fill_id,
                });
                return;
            }
        }

        let timestamp = cmd.timestamp_ms;
        if timestamp == 0 || cmd.fill_id.is_empty() {
            let _ = request.response_tx.send(RfqExecuteResult::Failed {
                reason: "RFQ command missing deterministic timestamp_ms or fill_id".to_string(),
            });
            return;
        }
        let env = crate::rsm::apply::CommandEnvelope::new(
            timestamp,
            crate::rsm::apply::EngineCommand::RfqExecute(cmd.clone()),
        );
        let apply_start = Instant::now();
        let apply_output = match self.apply(env) {
            Ok(output) => output,
            Err(crate::rsm::unified_engine::apply_interface::EngineError::Rejected(reason)) => {
                warn!(
                    rfq_id = %cmd.rfq_id,
                    reason = %reason,
                    "RFQ rejected by engine (nonce or admission)"
                );
                let _ = request
                    .response_tx
                    .send(RfqExecuteResult::Failed { reason });
                return;
            }
            Err(e) => {
                panic!(
                    "JOURNAL_FATAL: apply() failed for RFQ rfq_id={}: {}",
                    cmd.rfq_id, e
                )
            }
        };
        histogram!("ht_rfq_apply_seconds").record(apply_start.elapsed().as_secs_f64());

        // Extract plan result from apply output
        let rfq_plan = apply_output.rfq_plan.unwrap_or_else(|| {
            panic!(
                "JOURNAL_FATAL: apply() returned no rfq_plan for RFQ rfq_id={}",
                cmd.rfq_id
            )
        });

        match rfq_plan {
            Err(failure) => {
                warn!(
                    "RFQ execution rejected at planning: rfq_id={}, reason={}",
                    cmd.rfq_id,
                    match &failure {
                        RfqExecuteResult::Failed { reason } => reason.as_str(),
                        _ => "unknown",
                    }
                );
                let _ = request.response_tx.send(failure);
            }
            Ok(plan) => {
                info!(
                    "RFQ planned: rfq_id={}, fill_id={}, legs={}, premium={}",
                    cmd.rfq_id,
                    plan.fill_id,
                    cmd.legs.len(),
                    cmd.net_premium,
                );

                self.process_rfq_journaled(
                    request,
                    plan.fill_id,
                    apply_output.events,
                    apply_output.balance_updates,
                    plan.mmp_updates,
                )
                .await;

                histogram!("ht_rfq_handle_total_seconds")
                    .record(handle_start.elapsed().as_secs_f64());
            }
        }
    }

    /// Look up a cached RFQ response by request_id. Returns
    /// `Some(fill_id)` if the journal has already recorded this
    /// request, `None` otherwise.
    ///
    /// Two-step check:
    /// 1. Fast path: in-memory `known_request_ids` cache. Skips DB
    ///    entirely if the uuid isn't in the cache.
    /// 2. Slow path: query the journal writer for the full record and
    ///    decode the cached fill_id from `response_data`.
    ///
    /// Returns `None` on any decode/lookup failure rather than
    /// panicking — a stale cache entry should fall through to the
    /// normal execution path, not crash the engine.
    async fn lookup_cached_rfq_response(&self, request_id: &str) -> Option<String> {
        let req_uuid = uuid::Uuid::parse_str(request_id).ok()?;
        if !self.known_request_ids.contains(&req_uuid) {
            return None;
        }
        let writer = self.journal_writer.as_ref()?.clone();
        let lookup_uuid = req_uuid;
        let existing = tokio::task::spawn_blocking(move || writer.get_by_request_id(&lookup_uuid))
            .await
            .ok()?
            .ok()??;
        // response_data layout: [version_byte=1][rmp(String fill_id)]
        let bytes = existing.response_data.as_ref()?;
        if bytes.len() < 2 {
            return None;
        }
        let fill_id: String = rmp_serde::from_slice(&bytes[1..]).ok()?;
        Some(fill_id)
    }

    /// Serialise an RFQ response (fill_id) for journal storage. Wire
    /// format mirrors the orderbook command format: [version=1][rmp].
    pub(super) fn encode_rfq_response_data(fill_id: &str) -> Vec<u8> {
        let mut buf = Vec::with_capacity(64);
        buf.push(1u8);
        rmp_serde::encode::write_named(&mut buf, &fill_id).expect("encode fill_id");
        buf
    }

    /// Validate the command and build the planned event list. Pure: no
    /// DB writes, no portfolio mutations, no WS emissions. Used by
    /// `handle_rfq_execute` for the live path AND by replay paths that
    /// need to re-derive events without re-emitting.
    pub(crate) fn plan_rfq_execution(
        &mut self,
        cmd: &RfqExecuteCommand,
    ) -> Result<RfqExecutionPlan, RfqExecuteResult> {
        if cmd.timestamp_ms == 0 {
            return Err(RfqExecuteResult::Failed {
                reason: "RFQ command missing timestamp_ms".to_string(),
            });
        }
        if cmd.fill_id.is_empty() {
            return Err(RfqExecuteResult::Failed {
                reason: "RFQ command missing fill_id".to_string(),
            });
        }

        if cmd.taker_wallet == cmd.qp_wallet {
            return Err(RfqExecuteResult::Failed {
                reason: RFQ_SELF_TRADE_REJECTION_REASON.to_string(),
            });
        }

        // 1. Validate instruments exist and allow RFQ trading
        for leg in &cmd.legs {
            if self.ctx.expired_instruments.get(&leg.instrument) == Some(&true)
                || self
                    .expiry_manager
                    .is_instrument_expired(&leg.instrument, &self.ctx.orderbooks)
            {
                return Err(RfqExecuteResult::Failed {
                    reason: format!("Instrument has expired: {}", leg.instrument),
                });
            }
            if !self.ctx.orderbooks.contains_key(&leg.instrument) {
                return Err(RfqExecuteResult::Failed {
                    reason: format!("Invalid symbol: {}", leg.instrument),
                });
            }
            match self.ctx.instrument_trading_modes.get(&leg.instrument) {
                None => {
                    return Err(RfqExecuteResult::Failed {
                        reason: format!("Instrument not found: {}", leg.instrument),
                    });
                }
                Some(mode) if !mode.allows_rfq() => {
                    return Err(RfqExecuteResult::Failed {
                        reason: format!(
                            "Instrument {} is orderbook-only and does not accept RFQ trades",
                            leg.instrument
                        ),
                    });
                }
                Some(_) => {}
            }
        }

        // 2. Check MMP frozen for QP across every underlying in the
        //    command. A QP can be frozen on BTC but not ETH, so a BTC-leg
        //    RFQ must reject while an ETH-leg RFQ passes. Walk all unique
        //    underlyings and fail on the first frozen one.
        //    Uses engine-internal mmp_state (synchronous, no async reads).
        {
            let now = cmd.timestamp_ms;
            let mut checked_underlyings: std::collections::HashSet<String> =
                std::collections::HashSet::new();
            for leg in &cmd.legs {
                let Some(underlying) = leg.instrument.split('-').next().map(|s| s.to_string())
                else {
                    continue;
                };
                if !checked_underlyings.insert(underlying.clone()) {
                    continue;
                }
                let mmp_key = (cmd.qp_wallet, underlying.clone());
                if let Some(mmp_state) = self.ctx.mmp_state.get(&mmp_key) {
                    if mmp_state.is_frozen(now) {
                        return Err(RfqExecuteResult::Failed {
                            reason: format!("Quote provider MMP is frozen for {}", underlying),
                        });
                    }
                }
            }
        }

        // 3a. Cash coverage check on the signed net premium.
        //
        //     SPAN margin doesn't enforce cash floors: long-only option
        //     positions have IM = 0 because the loss is capped at the
        //     premium paid, but we never debit the premium from cash in
        //     the hypothetical portfolio. That means a wallet with 10
        //     USDC could pass SPAN for a 95 USDC long straddle. The
        //     downstream fill would then fail at ledger update time,
        //     which is the wrong place to find out.
        //
        //     We refuse at plan time instead: whichever wallet owes
        //     `abs(net_premium)` must have that much liquid cash. If the
        //     balance_ledger must be the source of truth for this pre-check.
        {
            use rust_decimal_macros::dec;

            if cmd.net_premium < dec!(0) {
                let taker_debit = -cmd.net_premium;
                let taker_cash = self
                    .ctx
                    .balance_ledger
                    .get(&cmd.taker_wallet)
                    .copied()
                    .unwrap_or(dec!(0));
                tracing::debug!(
                    request_id = %cmd.request_id,
                    payer_wallet = %cmd.taker_wallet,
                    payer_role = "taker",
                    payer_cash = %taker_cash,
                    required_debit = %taker_debit,
                    "RFQ balance_ledger premium pre-check"
                );
                if taker_cash < taker_debit {
                    tracing::warn!(
                        request_id = %cmd.request_id,
                        payer_wallet = %cmd.taker_wallet,
                        payer_role = "taker",
                        payer_cash = %taker_cash,
                        required_debit = %taker_debit,
                        "Rejecting RFQ: insufficient balance_ledger cash for premium debit"
                    );
                    return Err(RfqExecuteResult::Failed {
                        reason: format!(
                            "Taker cash {} insufficient to cover net premium debit {}",
                            taker_cash, taker_debit
                        ),
                    });
                }
            } else if cmd.net_premium > dec!(0) {
                let qp_debit = cmd.net_premium;
                let qp_cash = self
                    .ctx
                    .balance_ledger
                    .get(&cmd.qp_wallet)
                    .copied()
                    .unwrap_or(dec!(0));
                tracing::debug!(
                    request_id = %cmd.request_id,
                    payer_wallet = %cmd.qp_wallet,
                    payer_role = "qp",
                    payer_cash = %qp_cash,
                    required_debit = %qp_debit,
                    "RFQ balance_ledger premium pre-check"
                );
                if qp_cash < qp_debit {
                    tracing::warn!(
                        request_id = %cmd.request_id,
                        payer_wallet = %cmd.qp_wallet,
                        payer_role = "qp",
                        payer_cash = %qp_cash,
                        required_debit = %qp_debit,
                        "Rejecting RFQ: insufficient balance_ledger cash for premium debit"
                    );
                    return Err(RfqExecuteResult::Failed {
                        reason: format!(
                            "QP cash {} insufficient to cover net premium debit {}",
                            qp_cash, qp_debit
                        ),
                    });
                }
            }
            // net_premium == 0 (e.g. a perfect risk reversal) → no cash
            // debit on either side, skip the cash check.
        }

        // 3b. Margin check — build one OrderInfo per leg for the taker
        //     and one opposite-side OrderInfo per leg for the QP, then
        //     score each wallet as a single hypothetical portfolio via
        //     `check_margin_for_orders`. This is the whole reason we
        //     can't loop per-leg: a hedged collar (short call + long put)
        //     is healthy on the combined portfolio even though the short
        //     leg alone would fail margin.
        let taker_orders: Vec<OrderInfo> = cmd
            .legs
            .iter()
            .map(|leg| OrderInfo {
                symbol: leg.instrument.clone(),
                side: leg.taker_side,
                price: leg.price,
                size: to_contract_units_decimal(&leg.instrument, leg.size),
                tif: hypercall_types::TimeInForce::IOC,
                client_id: None,
                order_id: None,
                is_perp: false,
                underlying: leg.instrument.split('-').next().map(|s| s.to_string()),
                reduce_only: Some(false),
                nonce: None,
                signature: None,
                mmp_enabled: false,
                builder_code_address: cmd.builder_code_address,
            })
            .collect();

        for order in &taker_orders {
            if let Err(reason) = self.check_order_limits(&cmd.taker_wallet, order) {
                return Err(RfqExecuteResult::Failed {
                    reason: format!("Taker order limit failed: {}", reason),
                });
            }
            if let Err(reason) = LiquidationManager::check_preliquidation_order_allowed(
                &self.ctx.deps,
                &self.ctx.engine_positions,
                &cmd.taker_wallet,
                order,
            ) {
                return Err(RfqExecuteResult::Failed {
                    reason: format!("Taker pre-liquidation check failed: {}", reason),
                });
            }
            if let Err(reason) = self.margin_manager.check_tier_restrictions(
                &self.ctx.deps,
                &self.ctx.engine_positions,
                &self.ctx.balance_ledger,
                &cmd.taker_wallet,
                order,
                &self.ctx.order_index,
            ) {
                return Err(RfqExecuteResult::Failed {
                    reason: format!("Taker tier restriction failed: {}", reason),
                });
            }
        }

        if let Err(reason) = self.margin_manager.check_margin_for_orders(
            &self.ctx.deps,
            &self.ctx.engine_positions,
            &self.ctx.balance_ledger,
            &cmd.taker_wallet,
            &taker_orders,
            &self.ctx.order_index,
        ) {
            return Err(RfqExecuteResult::Failed {
                reason: format!("Taker margin check failed: {}", reason),
            });
        }

        // QP takes the opposite side of every leg.
        let qp_orders: Vec<OrderInfo> = cmd
            .legs
            .iter()
            .map(|leg| {
                let qp_side = match leg.taker_side {
                    Side::Buy => Side::Sell,
                    Side::Sell => Side::Buy,
                };
                OrderInfo {
                    symbol: leg.instrument.clone(),
                    side: qp_side,
                    price: leg.price,
                    size: to_contract_units_decimal(&leg.instrument, leg.size),
                    tif: hypercall_types::TimeInForce::IOC,
                    client_id: None,
                    order_id: None,
                    is_perp: false,
                    underlying: leg.instrument.split('-').next().map(|s| s.to_string()),
                    reduce_only: Some(false),
                    nonce: None,
                    signature: None,
                    mmp_enabled: false,
                    builder_code_address: None,
                }
            })
            .collect();

        for order in &qp_orders {
            if let Err(reason) = self.check_order_limits(&cmd.qp_wallet, order) {
                return Err(RfqExecuteResult::Failed {
                    reason: format!("QP order limit failed: {}", reason),
                });
            }
            if let Err(reason) = LiquidationManager::check_preliquidation_order_allowed(
                &self.ctx.deps,
                &self.ctx.engine_positions,
                &cmd.qp_wallet,
                order,
            ) {
                return Err(RfqExecuteResult::Failed {
                    reason: format!("QP pre-liquidation check failed: {}", reason),
                });
            }
            if let Err(reason) = self.margin_manager.check_tier_restrictions(
                &self.ctx.deps,
                &self.ctx.engine_positions,
                &self.ctx.balance_ledger,
                &cmd.qp_wallet,
                order,
                &self.ctx.order_index,
            ) {
                return Err(RfqExecuteResult::Failed {
                    reason: format!("QP tier restriction failed: {}", reason),
                });
            }
        }

        if let Err(reason) = self.check_pm_settlement_pool_gate_for_order_groups(&[
            (&cmd.taker_wallet, &taker_orders),
            (&cmd.qp_wallet, &qp_orders),
        ]) {
            return Err(RfqExecuteResult::Failed {
                reason: format!("RFQ PM settlement gate failed: {}", reason),
            });
        }

        if let Err(reason) = self.margin_manager.check_margin_for_quote_provider_orders(
            &self.ctx.deps,
            &self.ctx.engine_positions,
            &self.ctx.balance_ledger,
            &cmd.qp_wallet,
            &qp_orders,
            &self.ctx.order_index,
        ) {
            return Err(RfqExecuteResult::Failed {
                reason: format!("QP margin check failed: {}", reason),
            });
        }

        // 4. Validate that the signed `net_premium` (which the taker's
        //    EIP-712 signature committed to) matches the legs the QP sent,
        //    INCLUDING the sign. Without this check, a malformed or
        //    malicious quote where `net_premium` diverges from the legs
        //    would let the taker authorize one amount while the engine
        //    debits/credits another.
        //
        //    Unit + sign convention (matches `rfq_responder::price_rfq`):
        //    - `leg.size` on `RfqExecuteLeg` is in HUMAN-READABLE contracts
        //      (same as the frontend submits), NOT contract units. So the
        //      leg contribution is `leg.price * leg.size` with no unit
        //      divisor.
        //    - Taker buys → premium contribution is negative (taker pays).
        //      Taker sells → positive (taker receives). So the signed
        //      `net_premium` matches `sum(leg.price * leg.size * sign)`.
        let leg_implied_premium: rust_decimal::Decimal = cmd
            .legs
            .iter()
            .map(|l| {
                let magnitude = l.price * l.size;
                match l.taker_side {
                    Side::Buy => -magnitude,
                    Side::Sell => magnitude,
                }
            })
            .sum();
        if leg_implied_premium != cmd.net_premium {
            return Err(RfqExecuteResult::Failed {
                reason: format!(
                    "Leg-implied premium {} does not match signed net_premium {} for rfq_id={}",
                    leg_implied_premium, cmd.net_premium, cmd.rfq_id
                ),
            });
        }

        let plan = self.build_rfq_execution_plan_unchecked(cmd);

        Ok(plan)
    }

    /// Build RFQ fill events without re-running live acceptance checks.
    ///
    /// This is used by standby NATS replay after the primary has already
    /// accepted and journaled the RFQ. Replay must not reject a command that
    /// the primary accepted, and it must not block on live-only validation.
    pub(crate) fn build_rfq_execution_plan_unchecked(
        &mut self,
        cmd: &RfqExecuteCommand,
    ) -> RfqExecutionPlan {
        let fill_id = cmd.fill_id.clone();
        let timestamp = cmd.timestamp_ms;

        let mut events: Vec<EngineMessage> = Vec::with_capacity(cmd.legs.len() + 1);
        let mut rfq_fill_legs: Vec<RfqFillLeg> = Vec::with_capacity(cmd.legs.len());
        let mut mmp_updates: Vec<MmpFillUpdate> = Vec::with_capacity(cmd.legs.len());

        for leg in &cmd.legs {
            let trade_id = self.ctx.next_trade_id;
            self.ctx.next_trade_id += 1;

            // Convert human-readable size to contract units (internal format)
            let size_in_contract_units = to_contract_units_decimal(&leg.instrument, leg.size);
            let price_f64 = leg
                .price
                .to_f64()
                .expect("CRITICAL_FAILURE: RFQ leg price is not representable as f64");
            let size_u64 = size_in_contract_units.trunc().to_u64().expect(
                "CRITICAL_FAILURE: RFQ leg size is not representable as u64 contract units",
            );
            let fee_calc = self.fee_service.get_fees(
                &cmd.qp_wallet.as_hex(),
                &cmd.taker_wallet.as_hex(),
                price_f64,
                size_u64,
                cmd.builder_code_address.as_ref(),
            );
            let taker_fee = rust_decimal::Decimal::from_f64_retain(fee_calc.taker_fee)
                .expect("CRITICAL_FAILURE: RFQ taker fee is not representable as Decimal");

            let mut fill = Fill {
                trade_id,
                taker_order_id: 0,
                maker_order_id: 0,
                symbol: leg.instrument.clone(),
                price: leg.price,
                size: size_in_contract_units,
                taker_side: leg.taker_side,
                taker_wallet_address: cmd.taker_wallet,
                maker_wallet_address: cmd.qp_wallet,
                fee: taker_fee,
                is_taker: true,
                timestamp,
                builder_code_address: cmd.builder_code_address,
                builder_code_fee: fee_calc.builder_code_fee,
                source: FillSource::Rfq {
                    rfq_id: cmd.rfq_id.clone(),
                    quote_id: cmd.quote_id.clone(),
                },
                taker_realized_pnl: None,
                maker_realized_pnl: None,
                underlying_notional: None,
            };
            self.attach_fill_underlying_notional(&mut fill);

            events.push(EngineMessage::OrderFilled {
                accounting: hypercall_engine::FillAccounting::zero(fill.trade_id),
                fill: fill.clone(),
            });

            rfq_fill_legs.push(RfqFillLeg {
                instrument: leg.instrument.clone(),
                taker_side: leg.taker_side,
                price: leg.price,
                size: leg.size,
            });

            if let Some(underlying) = leg.instrument.split('-').next() {
                mmp_updates.push(MmpFillUpdate {
                    qp_wallet: cmd.qp_wallet,
                    underlying: underlying.to_string(),
                    fill,
                    timestamp_ms: timestamp,
                });
            }
        }

        // Append the summary RfqFilled event last so consumers see the
        // per-leg fills before the rollup.
        events.push(EngineMessage::RfqFilled(RfqFillMessage {
            fill_id: fill_id.clone(),
            rfq_id: cmd.rfq_id.clone(),
            quote_id: cmd.quote_id.clone(),
            taker_wallet: cmd.taker_wallet,
            qp_wallet: cmd.qp_wallet,
            legs: rfq_fill_legs,
            net_premium: cmd.net_premium,
            taker_accept_signature: cmd.taker_signature.clone(),
            timestamp,
        }));

        RfqExecutionPlan {
            fill_id,
            events,
            mmp_updates,
        }
    }
}