hypercall_api/websocket/

mod.rs

pub mod event_forwarder;
pub mod rfq;

use crate::boundary::read_models::PortfolioCacheApi;
use crate::options_chain::{parse_expiry_date_to_code, OptionsChainOptionTypeFilter};
use crate::rfq::handler_state::RfqHandlerState;
use crate::rfq::indicative_quote_cache::IndicativeQuoteCache;
use axum::{
    extract::{
        ws::{close_code, CloseFrame, Message, WebSocket, WebSocketUpgrade},
        Query, State,
    },
    response::Response,
};
use dashmap::DashMap;
use futures::{SinkExt, StreamExt};
use hypercall_types::api_models::OptionsChainStrikeRow;
pub use hypercall_types::ws_protocol::{
    IndexPriceEntry, PortfolioUpdate, WsCandleUpdate, WsCompetitionFinalStanding,
    WsCompetitionFinalStats, WsCompetitionGapUpdate, WsCompetitionPnlStanding,
    WsCompetitionPnlSummary, WsCompetitionRankChange, WsCompetitionUpdate, WsFillUpdate,
    WsIndexPriceUpdate, WsIndicativeMarketData, WsLiquidationStateChange, WsMarketUpdate,
    WsMessage, WsOptionsChainUpdate, WsOrderMessage, WsOrderResult, WsOrderbookUpdate,
    WsPositionExpired, WsRfqLegRequest, WsRfqQuoteEntry, WsRfqQuotes, WsRfqStatusUpdate,
    WsTradeUpdate,
};
use hypercall_types::{Side, WalletAddress};
use metrics::{counter, gauge, histogram};
use rust_decimal::Decimal;
use serde::Deserialize;
use std::collections::HashSet;
use std::str::FromStr;
use std::sync::Arc;
use tokio::sync::{mpsc, RwLock};
use tokio::time::{self, Duration, Instant, MissedTickBehavior};
use tracing::{debug, warn};

/// Channels that require authentication before subscribing.
/// These channels contain wallet-specific data and filter messages by authenticated wallet.
const AUTHENTICATED_CHANNELS: &[&str] = &[
    "order_updates",
    "fills",
    "portfolio",
    "liquidation",
    "competition",
    "competition_engagement",
    "rfq",
];
const DEFAULT_WS_HEARTBEAT_INTERVAL: Duration = Duration::from_secs(20);
const DEFAULT_WS_PONG_TIMEOUT: Duration = Duration::from_secs(60);

fn validate_ws_heartbeat_config(
    heartbeat_interval_ms: u64,
    pong_timeout_ms: u64,
) -> (Duration, Duration) {
    if heartbeat_interval_ms == 0 || pong_timeout_ms == 0 {
        warn!("WebSocket heartbeat settings must be positive, using defaults");
        return (DEFAULT_WS_HEARTBEAT_INTERVAL, DEFAULT_WS_PONG_TIMEOUT);
    }

    let heartbeat_interval = Duration::from_millis(heartbeat_interval_ms);
    let pong_timeout = Duration::from_millis(pong_timeout_ms);

    if pong_timeout < heartbeat_interval {
        warn!(
            heartbeat_interval_ms,
            pong_timeout_ms,
            "WebSocket pong timeout must be greater than or equal to heartbeat interval, using defaults"
        );
        return (DEFAULT_WS_HEARTBEAT_INTERVAL, DEFAULT_WS_PONG_TIMEOUT);
    }

    (heartbeat_interval, pong_timeout)
}

pub fn ws_heartbeat_config(
    api_config: &hypercall_config::ApiRuntimeConfig,
) -> (Duration, Duration) {
    validate_ws_heartbeat_config(
        api_config.ws_heartbeat_interval_ms,
        api_config.ws_pong_timeout_ms,
    )
}

/// Check if a channel requires authentication.
fn requires_authentication(channel: &str) -> bool {
    AUTHENTICATED_CHANNELS.contains(&channel)
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct OptionsChainSubscription {
    symbols: Option<Vec<String>>,
    /// UTC calendar date code in YYYYMMDD form (e.g. 20260131), matched
    /// against the UTC date of instrument expiry timestamps so one filter
    /// spans underlyings with different expiry times of day.
    expiry: Option<u64>,
    option_type: OptionsChainOptionTypeFilter,
}

impl Default for OptionsChainSubscription {
    fn default() -> Self {
        Self {
            symbols: None,
            expiry: None,
            option_type: OptionsChainOptionTypeFilter::Both,
        }
    }
}

fn normalize_symbols(symbols: Option<Vec<String>>) -> Option<Vec<String>> {
    let values = symbols?;
    let mut deduped: Vec<String> = values
        .into_iter()
        .map(|value| value.trim().to_ascii_uppercase())
        .filter(|value| !value.is_empty())
        .collect();

    if deduped.is_empty() {
        return None;
    }

    deduped.sort();
    deduped.dedup();
    Some(deduped)
}

fn parse_options_chain_subscription(
    symbols: Option<Vec<String>>,
    expiry: Option<String>,
    option_type: Option<String>,
) -> Result<OptionsChainSubscription, String> {
    let expiry = match expiry {
        Some(value) if value.trim().is_empty() => None,
        Some(value) => Some(parse_expiry_date_to_code(value.trim())?),
        None => None,
    };

    Ok(OptionsChainSubscription {
        symbols: normalize_symbols(symbols),
        expiry,
        option_type: OptionsChainOptionTypeFilter::parse(option_type.as_deref())?,
    })
}

fn options_chain_update_symbols(update: &WsOptionsChainUpdate) -> HashSet<String> {
    match update {
        WsOptionsChainUpdate::Upsert { row, .. } => {
            let mut symbols = HashSet::new();
            if let Some(call) = &row.call {
                symbols.insert(call.symbol.to_ascii_uppercase());
            }
            if let Some(put) = &row.put {
                symbols.insert(put.symbol.to_ascii_uppercase());
            }
            symbols
        }
        WsOptionsChainUpdate::Remove { symbol, .. } => HashSet::from([symbol.to_ascii_uppercase()]),
    }
}

/// UTC calendar date code (YYYYMMDD) of an expiry timestamp in seconds.
fn expiry_timestamp_to_code(expiry_ts_secs: u64) -> Option<u64> {
    use chrono::Datelike;
    let timestamp = i64::try_from(expiry_ts_secs).ok()?;
    let date = chrono::DateTime::from_timestamp(timestamp, 0)?.date_naive();
    Some((date.year() as u64) * 10_000 + (date.month() as u64) * 100 + date.day() as u64)
}

fn options_chain_update_matches(
    update: &WsOptionsChainUpdate,
    subscription: &OptionsChainSubscription,
) -> bool {
    if let Some(expected_expiry_code) = subscription.expiry {
        let update_expiry = match update {
            WsOptionsChainUpdate::Upsert { expiry, .. } => *expiry,
            WsOptionsChainUpdate::Remove { expiry, .. } => *expiry,
        };
        // The subscription filter is a calendar date (YYYYMMDD). Compare on
        // the UTC date of the update's expiry timestamp so the filter works
        // across underlyings with different expiry times of day.
        if expiry_timestamp_to_code(update_expiry) != Some(expected_expiry_code) {
            return false;
        }
    }

    if let Some(symbols_filter) = &subscription.symbols {
        let symbols_set: HashSet<String> = symbols_filter.iter().cloned().collect();
        let update_symbols = options_chain_update_symbols(update);
        if symbols_set.is_disjoint(&update_symbols) {
            return false;
        }
    }

    match update {
        WsOptionsChainUpdate::Upsert { row, .. } => {
            let has_call = row.call.is_some();
            let has_put = row.put.is_some();
            match subscription.option_type {
                OptionsChainOptionTypeFilter::Both => has_call || has_put,
                OptionsChainOptionTypeFilter::Call => has_call,
                OptionsChainOptionTypeFilter::Put => has_put,
            }
        }
        WsOptionsChainUpdate::Remove { option_type, .. } => {
            subscription.option_type.allows_option_type(option_type)
        }
    }
}

fn apply_option_type_filter_to_row(
    mut row: OptionsChainStrikeRow,
    option_type: OptionsChainOptionTypeFilter,
) -> Option<OptionsChainStrikeRow> {
    if row.call.is_some() && row.put.is_some() {
        warn!(
            strike = row.strike,
            "options_chain upsert contained both call and put legs; pruning by subscription filter"
        );
    }

    match option_type {
        OptionsChainOptionTypeFilter::Both => {}
        OptionsChainOptionTypeFilter::Call => row.put = None,
        OptionsChainOptionTypeFilter::Put => row.call = None,
    }

    if row.call.is_none() && row.put.is_none() {
        return None;
    }

    Some(row)
}

fn apply_options_chain_filters_to_update(
    update: &WsOptionsChainUpdate,
    option_type_filter: OptionsChainOptionTypeFilter,
) -> Option<WsOptionsChainUpdate> {
    match update {
        WsOptionsChainUpdate::Upsert {
            currency,
            expiry,
            row,
            timestamp,
        } => {
            let row = apply_option_type_filter_to_row(row.clone(), option_type_filter)?;

            Some(WsOptionsChainUpdate::Upsert {
                currency: currency.clone(),
                expiry: *expiry,
                row,
                timestamp: *timestamp,
            })
        }
        WsOptionsChainUpdate::Remove {
            currency,
            expiry,
            strike,
            option_type,
            symbol,
            timestamp,
        } => {
            if !option_type_filter.allows_option_type(option_type) {
                return None;
            }

            Some(WsOptionsChainUpdate::Remove {
                currency: currency.clone(),
                expiry: *expiry,
                strike: *strike,
                option_type: option_type.clone(),
                symbol: symbol.clone(),
                timestamp: *timestamp,
            })
        }
    }
}

fn effective_option_type_for_matching_subscriptions(
    subscriptions: &[OptionsChainSubscription],
    update: &WsOptionsChainUpdate,
) -> Option<OptionsChainOptionTypeFilter> {
    let mut has_call = false;
    let mut has_put = false;
    let mut has_both = false;

    for subscription in subscriptions {
        if !options_chain_update_matches(update, subscription) {
            continue;
        }

        match subscription.option_type {
            OptionsChainOptionTypeFilter::Call => has_call = true,
            OptionsChainOptionTypeFilter::Put => has_put = true,
            OptionsChainOptionTypeFilter::Both => has_both = true,
        }
    }

    if has_both || (has_call && has_put) {
        Some(OptionsChainOptionTypeFilter::Both)
    } else if has_call {
        Some(OptionsChainOptionTypeFilter::Call)
    } else if has_put {
        Some(OptionsChainOptionTypeFilter::Put)
    } else {
        None
    }
}

type ClientId = uuid::Uuid;

struct Client {
    sender: mpsc::UnboundedSender<Arc<WsMessage>>,
    subscriptions: Vec<String>,
    options_chain_subscriptions: Vec<OptionsChainSubscription>,
    /// When set, only forward order_updates/fills whose symbol matches one of
    /// these underlyings (e.g. "BTC", "ETH"). When empty, all orders are forwarded.
    order_update_symbols: HashSet<String>,
    /// When set, only forward orderbook/trades updates whose full instrument
    /// symbol matches. When empty, all updates are forwarded.
    public_channel_symbols: HashSet<String>,
    authenticated: bool,
    wallet_address: Option<WalletAddress>,
}

#[derive(Clone)]
pub struct PubSubManager {
    clients: Arc<DashMap<ClientId, Client>>,
    channel_subscribers: Arc<DashMap<String, Vec<ClientId>>>,
}

impl Default for PubSubManager {
    fn default() -> Self {
        Self::new()
    }
}

impl PubSubManager {
    pub fn new() -> Self {
        Self {
            clients: Arc::new(DashMap::new()),
            channel_subscribers: Arc::new(DashMap::new()),
        }
    }

    pub fn add_client(
        &self,
        client_id: ClientId,
        sender: mpsc::UnboundedSender<Arc<WsMessage>>,
        authenticated: bool,
        wallet_address: Option<WalletAddress>,
    ) {
        let client = Client {
            sender,
            subscriptions: Vec::new(),
            options_chain_subscriptions: Vec::new(),
            order_update_symbols: HashSet::new(),
            public_channel_symbols: HashSet::new(),
            authenticated,
            wallet_address,
        };
        let prev = self.clients.insert(client_id, client);
        // Only increment metrics if this is a new client (not a replacement)
        if prev.is_none() {
            gauge!("ht_ws_connections_active").increment(1.0);
            counter!("ht_ws_connections_total").increment(1);
        }
    }

    /// Set the wallet address on a connected client (message-based auth).
    pub fn authenticate_client(
        &self,
        client_id: ClientId,
        wallet_address: WalletAddress,
    ) -> Result<(), String> {
        let mut client = self.clients.get_mut(&client_id).ok_or("Client not found")?;
        client.authenticated = true;
        client.wallet_address = Some(wallet_address);
        Ok(())
    }

    pub fn remove_client(&self, client_id: ClientId) {
        if let Some((_, client)) = self.clients.remove(&client_id) {
            // Only decrement metrics if client actually existed
            gauge!("ht_ws_connections_active").decrement(1.0);
            counter!("ht_ws_disconnections_total").increment(1);

            for channel in &client.subscriptions {
                if let Some(mut subs) = self.channel_subscribers.get_mut(channel) {
                    subs.retain(|id| id != &client_id);
                    if subs.is_empty() {
                        // Drop the mutable ref before removing the key
                        drop(subs);
                        self.channel_subscribers.remove(channel);
                    }
                }
                gauge!("ht_ws_channel_subscribers", "channel" => channel.clone()).decrement(1.0);
            }
        }
    }

    pub fn subscribe(
        &self,
        client_id: ClientId,
        channel: String,
        symbols: Option<Vec<String>>,
        expiry: Option<String>,
        option_type: Option<String>,
    ) -> Result<(), String> {
        // Get mutable access to client for auth check + subscription update
        let mut client = self.clients.get_mut(&client_id).ok_or("Client not found")?;

        // Check authentication for protected channels
        if requires_authentication(&channel) && !client.authenticated {
            return Err("Authentication required for this channel".to_string());
        }

        if channel == "options_chain" {
            let subscription =
                parse_options_chain_subscription(symbols.clone(), expiry, option_type)?;

            if !client.options_chain_subscriptions.contains(&subscription) {
                client.options_chain_subscriptions.push(subscription);
            }
        }

        // For order_updates and fills: track symbol filters.
        // Symbols are underlyings (e.g. "BTC", "ETH") extracted from the
        // full instrument symbol (e.g. "BTC-20260131-100000-C" -> "BTC").
        if (channel == "order_updates" || channel == "fills") && symbols.is_some() {
            if let Some(syms) = &symbols {
                for sym in syms {
                    let normalized = sym.trim().to_uppercase();
                    if !normalized.is_empty() {
                        // Accept both full instrument names and bare underlyings.
                        // Extract the underlying (first segment before '-').
                        let underlying = normalized
                            .split('-')
                            .next()
                            .unwrap_or(&normalized)
                            .to_string();
                        client.order_update_symbols.insert(underlying);
                    }
                }
            }
        }

        // For public market-data channels: track exact instrument symbol filters.
        if (channel == "orderbook" || channel == "trades" || channel == "indicative_market_data")
            && symbols.is_some()
        {
            if let Some(syms) = &symbols {
                for sym in syms {
                    let normalized = sym.trim().to_uppercase();
                    if !normalized.is_empty() {
                        client.public_channel_symbols.insert(normalized);
                    }
                }
            }
        }

        // Check if already subscribed to prevent duplicates
        if client.subscriptions.contains(&channel) {
            return Ok(());
        }

        // Add to client's subscriptions
        client.subscriptions.push(channel.clone());

        // Drop client ref before touching channel_subscribers (separate DashMap)
        drop(client);

        // Add to channel subscribers (with duplicate check)
        let mut subs = self.channel_subscribers.entry(channel.clone()).or_default();
        if !subs.contains(&client_id) {
            subs.push(client_id);
        }

        gauge!("ht_ws_channel_subscribers", "channel" => channel).set(subs.len() as f64);

        Ok(())
    }

    pub fn unsubscribe(
        &self,
        client_id: ClientId,
        channel: String,
        symbols: Option<Vec<String>>,
        expiry: Option<String>,
        option_type: Option<String>,
    ) -> Result<(), String> {
        let mut client = self.clients.get_mut(&client_id).ok_or("Client not found")?;

        // For order_updates and fills: remove specific symbol filters,
        // or clear all filters if no symbols specified.
        if channel == "order_updates" || channel == "fills" {
            if let Some(syms) = &symbols {
                for sym in syms {
                    let normalized = sym.trim().to_uppercase();
                    let underlying = normalized
                        .split('-')
                        .next()
                        .unwrap_or(&normalized)
                        .to_string();
                    client.order_update_symbols.remove(&underlying);
                }
                // If there are still symbol filters, don't fully unsubscribe
                // from the channel - just acknowledge the symbol removal.
                if !client.order_update_symbols.is_empty() {
                    return Ok(());
                }
            } else {
                // No symbols specified - clear all filters (full unsubscribe)
                client.order_update_symbols.clear();
            }
        }

        // For public market-data channels: remove specific instrument symbol filters.
        if channel == "orderbook" || channel == "trades" || channel == "indicative_market_data" {
            if let Some(syms) = &symbols {
                for sym in syms {
                    let normalized = sym.trim().to_uppercase();
                    client.public_channel_symbols.remove(&normalized);
                }
                if !client.public_channel_symbols.is_empty() {
                    return Ok(());
                }
            } else {
                client.public_channel_symbols.clear();
            }
        }

        if channel == "options_chain" {
            let normalized_symbols = normalize_symbols(symbols);
            let normalized_expiry = expiry.and_then(|value| {
                let trimmed = value.trim();
                if trimmed.is_empty() {
                    None
                } else {
                    Some(trimmed.to_string())
                }
            });
            let normalized_option_type = option_type.and_then(|value| {
                let trimmed = value.trim();
                if trimmed.is_empty() {
                    None
                } else {
                    Some(trimmed.to_string())
                }
            });

            if normalized_symbols.is_none()
                && normalized_expiry.is_none()
                && normalized_option_type.is_none()
            {
                client.options_chain_subscriptions.clear();
            } else {
                let subscription = parse_options_chain_subscription(
                    normalized_symbols,
                    normalized_expiry,
                    normalized_option_type,
                )?;
                client
                    .options_chain_subscriptions
                    .retain(|existing| existing != &subscription);
            }

            if !client.options_chain_subscriptions.is_empty() {
                return Ok(());
            }
        }

        // Remove from client's subscriptions
        client.subscriptions.retain(|ch| ch != &channel);

        // Drop client ref before touching channel_subscribers
        drop(client);

        // Remove from channel subscribers
        let mut subscriber_count = 0;
        if let Some(mut subs) = self.channel_subscribers.get_mut(&channel) {
            subs.retain(|id| id != &client_id);
            subscriber_count = subs.len();
            if subs.is_empty() {
                drop(subs);
                self.channel_subscribers.remove(&channel);
            }
        }

        gauge!("ht_ws_channel_subscribers", "channel" => channel).set(subscriber_count as f64);

        Ok(())
    }

    pub fn publish_to_channel(&self, channel: &str, message: WsMessage) {
        debug!(
            "Publishing ws message to channel: {:?}: {:?}",
            channel, message
        );
        counter!("ht_ws_messages_sent_total", "channel" => channel.to_string()).increment(1);

        let start = Instant::now();
        let msg = Arc::new(message);
        let mut fan_out: u64 = 0;

        if let Some(subscriber_ids) = self.channel_subscribers.get(channel) {
            let is_authenticated = requires_authentication(channel);

            for client_id in subscriber_ids.iter() {
                if let Some(client) = self.clients.get(client_id) {
                    if is_authenticated {
                        let should_send = match msg.as_ref() {
                            WsMessage::OrderUpdate(order) => {
                                let wallet_match =
                                    client.wallet_address.as_ref() == Some(&order.wallet_address);
                                let symbol_match = client.order_update_symbols.is_empty()
                                    || order
                                        .request
                                        .symbol
                                        .split('-')
                                        .next()
                                        .map(|u| {
                                            client.order_update_symbols.contains(&u.to_uppercase())
                                        })
                                        .unwrap_or(true);
                                wallet_match && symbol_match
                            }
                            WsMessage::Fill(fill) => {
                                let wallet_match =
                                    client.wallet_address.as_ref() == Some(&fill.wallet_address);
                                let symbol_match = client.order_update_symbols.is_empty()
                                    || fill
                                        .symbol
                                        .split('-')
                                        .next()
                                        .map(|u| {
                                            client.order_update_symbols.contains(&u.to_uppercase())
                                        })
                                        .unwrap_or(true);
                                wallet_match && symbol_match
                            }
                            WsMessage::PositionExpired(expired) => {
                                client.wallet_address.as_ref() == Some(&expired.wallet_address)
                            }
                            WsMessage::LiquidationStateChange(state_change) => {
                                client.wallet_address.as_ref() == Some(&state_change.wallet_address)
                            }
                            WsMessage::CompetitionUpdate(update) => {
                                client.wallet_address.as_ref() == Some(&update.wallet_address)
                            }
                            WsMessage::CompetitionPnlSummary(summary) => {
                                client.wallet_address.as_ref() == Some(&summary.wallet_address)
                            }
                            WsMessage::CompetitionFinalStats(final_stats) => {
                                client.wallet_address.as_ref() == Some(&final_stats.wallet_address)
                            }
                            WsMessage::CompetitionRankChange(change) => {
                                client.wallet_address.as_ref() == Some(&change.wallet_address)
                            }
                            WsMessage::CompetitionGapUpdate(gap) => {
                                client.wallet_address.as_ref() == Some(&gap.wallet_address)
                            }
                            WsMessage::CompetitionFinalStanding(final_standing) => {
                                client.wallet_address.as_ref()
                                    == Some(&final_standing.wallet_address)
                            }
                            WsMessage::RfqQuotes(rfq) => {
                                client.wallet_address.as_ref() == Some(&rfq.taker_wallet)
                            }
                            WsMessage::RfqStatusUpdate(rfq) => {
                                client.wallet_address.as_ref() == Some(&rfq.taker_wallet)
                            }
                            WsMessage::RfqAcceptResult {
                                taker_wallet: Some(ref tw),
                                ..
                            } => client.wallet_address.as_ref() == Some(tw),
                            _ => true,
                        };

                        if should_send {
                            let _ = client.sender.send(Arc::clone(&msg));
                            fan_out += 1;
                        }
                    } else if channel == "options_chain" {
                        if let WsMessage::OptionsChainUpdate(update) = msg.as_ref() {
                            let effective_option_type =
                                effective_option_type_for_matching_subscriptions(
                                    &client.options_chain_subscriptions,
                                    update,
                                );
                            if let Some(option_type) = effective_option_type {
                                if let Some(filtered_update) =
                                    apply_options_chain_filters_to_update(update, option_type)
                                {
                                    let _ = client.sender.send(Arc::new(
                                        WsMessage::OptionsChainUpdate(filtered_update),
                                    ));
                                    fan_out += 1;
                                }
                            }
                        }
                    } else {
                        // Public channel - apply symbol filter if set
                        let passes_filter = if client.public_channel_symbols.is_empty() {
                            true
                        } else {
                            match msg.as_ref() {
                                WsMessage::OrderbookUpdate(update) => client
                                    .public_channel_symbols
                                    .contains(&update.symbol.to_uppercase()),
                                WsMessage::Trade(trade) => client
                                    .public_channel_symbols
                                    .contains(&trade.symbol.to_uppercase()),
                                WsMessage::IndicativeMarketData(update) => client
                                    .public_channel_symbols
                                    .contains(&update.instrument.to_uppercase()),
                                _ => true,
                            }
                        };
                        if passes_filter {
                            let _ = client.sender.send(Arc::clone(&msg));
                            fan_out += 1;
                        }
                    }
                }
            }
        }

        let duration_seconds = start.elapsed().as_secs_f64();
        histogram!("ht_ws_publish_fan_out", "channel" => channel.to_string())
            .record(fan_out as f64);
        histogram!("ht_ws_publish_duration_seconds", "channel" => channel.to_string())
            .record(duration_seconds);
    }

    pub async fn active_channels_with_prefix(&self, prefix: &str) -> Vec<String> {
        let mut channels: Vec<String> = self
            .channel_subscribers
            .iter()
            .map(|entry| entry.key().clone())
            .filter(|channel| channel.starts_with(prefix))
            .collect();
        channels.sort_unstable();
        channels
    }

    pub fn send_to_client(&self, client_id: ClientId, message: WsMessage) {
        if let Some(client) = self.clients.get(&client_id) {
            let _ = client.sender.send(Arc::new(message));
        }
    }

    pub fn publish_orderbook_update(&self, update: WsOrderbookUpdate) {
        self.publish_to_channel("orderbook", WsMessage::OrderbookUpdate(update));
    }

    pub fn publish_order_update(&self, order: WsOrderMessage) {
        self.publish_to_channel("order_updates", WsMessage::OrderUpdate(order));
    }

    pub fn publish_fill(&self, fill: WsFillUpdate) {
        self.publish_to_channel("fills", WsMessage::Fill(fill));
    }

    pub fn publish_market_update(&self, update: WsMarketUpdate) {
        self.publish_to_channel("market_updates", WsMessage::MarketUpdate(update));
    }

    pub fn publish_trade(&self, trade: WsTradeUpdate) {
        self.publish_to_channel("trades", WsMessage::Trade(trade));
    }

    pub fn publish_options_chain_update(&self, update: WsOptionsChainUpdate) {
        self.publish_to_channel("options_chain", WsMessage::OptionsChainUpdate(update));
    }

    pub fn publish_index_prices(&self, update: WsIndexPriceUpdate) {
        self.publish_to_channel("index_prices", WsMessage::IndexPriceUpdate(update));
    }

    pub fn publish_indicative_market_data(&self, update: WsIndicativeMarketData) {
        self.publish_to_channel(
            "indicative_market_data",
            WsMessage::IndicativeMarketData(update),
        );
    }

    pub fn publish_position_expired(&self, expired: WsPositionExpired) {
        self.publish_to_channel("portfolio", WsMessage::PositionExpired(expired));
    }

    pub fn publish_liquidation_state_change(&self, state_change: WsLiquidationStateChange) {
        self.publish_to_channel(
            "liquidation",
            WsMessage::LiquidationStateChange(state_change),
        );
    }

    pub async fn publish_competition_update(&self, update: WsCompetitionUpdate) {
        self.publish_to_channel("competition", WsMessage::CompetitionUpdate(update));
    }

    pub async fn publish_competition_pnl_summary(&self, summary: WsCompetitionPnlSummary) {
        self.publish_to_channel("competition", WsMessage::CompetitionPnlSummary(summary));
    }

    pub async fn publish_competition_final_stats(&self, final_stats: WsCompetitionFinalStats) {
        self.publish_to_channel("competition", WsMessage::CompetitionFinalStats(final_stats));
    }

    pub async fn publish_competition_rank_change(&self, update: WsCompetitionRankChange) {
        self.publish_to_channel(
            "competition_engagement",
            WsMessage::CompetitionRankChange(update),
        );
    }

    pub async fn publish_competition_gap_update(&self, update: WsCompetitionGapUpdate) {
        self.publish_to_channel(
            "competition_engagement",
            WsMessage::CompetitionGapUpdate(update),
        );
    }

    pub async fn publish_competition_final_standing(&self, update: WsCompetitionFinalStanding) {
        self.publish_to_channel(
            "competition_engagement",
            WsMessage::CompetitionFinalStanding(update),
        );
    }

    pub fn publish_rfq_quotes(&self, quotes: WsRfqQuotes) {
        self.publish_to_channel("rfq", WsMessage::RfqQuotes(quotes));
    }

    /// Return wallets subscribed to the given channel (authenticated wallets only).
    pub async fn subscribed_wallets(&self, channel: &str) -> Vec<WalletAddress> {
        let mut wallets = HashSet::new();

        if let Some(subscriber_ids) = self.channel_subscribers.get(channel) {
            for client_id in subscriber_ids.iter() {
                if let Some(client) = self.clients.get(client_id) {
                    if let Some(wallet) = client.wallet_address {
                        wallets.insert(wallet);
                    }
                }
            }
        }

        wallets.into_iter().collect()
    }
}

impl crate::rfq::qp_ws_state::RfqWebsocketPublisher for PubSubManager {
    fn publish_indicative_market_data(&self, update: WsIndicativeMarketData) {
        PubSubManager::publish_indicative_market_data(self, update);
    }

    fn publish_rfq_quotes(&self, quotes: WsRfqQuotes) {
        PubSubManager::publish_rfq_quotes(self, quotes);
    }

    fn publish_to_channel(&self, channel: &str, message: WsMessage) {
        PubSubManager::publish_to_channel(self, channel, message);
    }
}

#[derive(Clone)]
pub struct WsState {
    pub pubsub: PubSubManager,
    pub indicative_cache: Option<Arc<IndicativeQuoteCache>>,
    pub portfolio_cache: Option<Arc<dyn PortfolioCacheApi>>,
    pub competition_service: Option<Arc<hypercall_competition::CompetitionService>>,
    pub heartbeat_config: (Duration, Duration),
    pub rfq_handler_state: Option<RfqHandlerState>,
    pub order_sender: Option<mpsc::Sender<hypercall_runtime_api::UnifiedEngineRequest>>,
    pub trading_halt: Option<Arc<RwLock<crate::trading_halt::TradingHaltState>>>,
    pub agent_auth: Arc<dyn hypercall_runtime_api::AgentAuthProvider>,
    pub signing_chain_id: u64,
}

#[derive(Deserialize)]
pub struct WsQuery {
    pub wallet: Option<String>,
}

pub async fn websocket_handler(
    ws: WebSocketUpgrade,
    State(state): State<WsState>,
    Query(query): Query<WsQuery>,
) -> Response {
    // WebSocket authentication modes:
    // - No query params => public streams only
    // - ?wallet=0x... => authenticated for that wallet, allows private streams
    let (authenticated, wallet_address) = if let Some(wallet) = query.wallet {
        // Wallet-based authentication for EIP-712 flow
        // Parse wallet string to WalletAddress
        match WalletAddress::from_str(&wallet) {
            Ok(addr) => (true, Some(addr)),
            Err(_) => {
                tracing::warn!(
                    "Invalid wallet address format in WebSocket query: {}",
                    wallet
                );
                (false, None)
            }
        }
    } else {
        // Public streams only
        (false, None)
    };

    ws.on_upgrade(move |socket| handle_socket(socket, state, authenticated, wallet_address))
}

async fn handle_socket(
    socket: WebSocket,
    state: WsState,
    authenticated: bool,
    wallet_address: Option<WalletAddress>,
) {
    let (heartbeat_interval, pong_timeout) = state.heartbeat_config;
    let (mut sender, mut receiver) = socket.split();
    let client_id = uuid::Uuid::new_v4();

    let (tx, mut rx) = mpsc::unbounded_channel::<Arc<WsMessage>>();

    // Add client to pubsub manager
    let mut wallet_addr_clone = wallet_address;
    let mut authenticated = authenticated;
    state
        .pubsub
        .add_client(client_id, tx.clone(), authenticated, wallet_address);

    // Task for portfolio updates if authenticated
    let mut portfolio_task = None;
    let mut portfolio_subscription: Option<(WalletAddress, u64)> = None;
    if authenticated && wallet_addr_clone.is_some() && state.portfolio_cache.is_some() {
        // This will be set up when client subscribes to portfolio channel
    }

    let mut heartbeat = time::interval(heartbeat_interval);
    heartbeat.set_missed_tick_behavior(MissedTickBehavior::Delay);
    heartbeat.tick().await;
    let mut last_pong = Instant::now();

    loop {
        tokio::select! {
            maybe_msg = rx.recv() => {
                let Some(msg) = maybe_msg else {
                    break;
                };

                if let Ok(json) = sonic_rs::to_string(msg.as_ref()) {
                    debug!("Sending message to client: {:?}", msg);
                    if sender.send(Message::Text(json)).await.is_err() {
                        break;
                    }
                }
            }
            maybe_frame = receiver.next() => {
                let Some(frame_result) = maybe_frame else {
                    break;
                };

                let Ok(frame) = frame_result else {
                    break;
                };

                match frame {
                    Message::Text(text) => {
                        let deser_result = sonic_rs::from_str::<WsMessage>(&text);
                        if deser_result.is_err() && (text.contains("SubmitAutoExecuteRfq") || text.contains("SubmitRfq")) {
                            tracing::warn!(
                                client_id = ?client_id,
                                error = %deser_result.as_ref().unwrap_err(),
                                msg_len = text.len(),
                                "Failed to deserialize RFQ WS message"
                            );
                        }
                        if let Ok(ws_msg) = deser_result {
                            match ws_msg {
                                WsMessage::Subscribe {
                                    channel,
                                    symbols,
                                    expiry,
                                    option_type,
                                } => {
                                    match state.pubsub.subscribe(
                                        client_id,
                                        channel.clone(),
                                        symbols.clone(),
                                        expiry,
                                        option_type,
                                    ) {
                                        Ok(_) => {
                                            if channel == "portfolio"
                                                && authenticated
                                                && portfolio_subscription.is_none()
                                            {
                                                if let Some(ref wallet) = wallet_addr_clone {
                                                    if let Some(ref cache) = state.portfolio_cache {
                                                        let (subscriber_id, mut portfolio_rx) =
                                                            cache.subscribe(*wallet).await;
                                                        let tx_clone = tx.clone();
                                                        portfolio_subscription =
                                                            Some((*wallet, subscriber_id));
                                                        portfolio_task =
                                                            Some(tokio::spawn(async move {
                                                                while let Some(update) =
                                                                    portfolio_rx.recv().await
                                                                {
                                                                    if tx_clone
                                                                        .send(Arc::new(
                                                                            WsMessage::PortfolioUpdate(
                                                                                update,
                                                                            ),
                                                                        ))
                                                                        .is_err()
                                                                    {
                                                                        break;
                                                                    }
                                                                }
                                                            }));
                                                    }
                                                }
                                            }

                                            state.pubsub.send_to_client(
                                                client_id,
                                                WsMessage::Subscribed {
                                                    channel: channel.clone(),
                                                },
                                            );

                                            if channel == "indicative_market_data" {
                                                if let Some(ref cache) = state.indicative_cache {
                                                    let symbol_filter = normalize_symbols(symbols);
                                                    let mut snapshot_count = 0u64;
                                                    for quote in cache.get_all_aggregates() {
                                                        if let Some(filter) = &symbol_filter {
                                                            if !filter.contains(
                                                                &quote.instrument.to_ascii_uppercase(),
                                                            ) {
                                                                continue;
                                                            }
                                                        }
                                                        snapshot_count += 1;
                                                        state
                                                            .pubsub
                                                            .send_to_client(
                                                                client_id,
                                                                WsMessage::IndicativeMarketData(
                                                                    WsIndicativeMarketData {
                                                                        instrument: quote.instrument,
                                                                        best_bid: Some(quote.best_bid),
                                                                        best_ask: Some(quote.best_ask),
                                                                        bid_iv: None,
                                                                        ask_iv: None,
                                                                        indicative_bid_size: quote
                                                                            .indicative_bid_size
                                                                            .into(),
                                                                        indicative_ask_size: quote
                                                                            .indicative_ask_size
                                                                            .into(),
                                                                        num_providers: quote
                                                                            .num_providers,
                                                                        timestamp: quote.updated_at,
                                                                    },
                                                                ),
                                                            );
                                                    }
                                                    counter!(
                                                        "ht_ws_indicative_snapshot_sent_total"
                                                    )
                                                    .increment(snapshot_count);
                                                }
                                            }

                                            if channel == "competition" && authenticated {
                                                if let (Some(wallet), Some(service)) = (
                                                    wallet_addr_clone,
                                                    state.competition_service.as_ref(),
                                                ) {
                                                    match service
                                                        .build_competition_pnl_summary_for_wallet(
                                                            wallet,
                                                            chrono::Utc::now().timestamp_millis(),
                                                        )
                                                        .await
                                                    {
                                                        Ok(summary) => {
                                                            state.pubsub.send_to_client(
                                                                client_id,
                                                                WsMessage::CompetitionPnlSummary(
                                                                    summary,
                                                                ),
                                                            );
                                                        }
                                                        Err(err) => {
                                                            state.pubsub.send_to_client(
                                                                client_id,
                                                                WsMessage::Error {
                                                                    message: format!(
                                                                        "failed to build competition summary: {}",
                                                                        err
                                                                    ),
                                                                },
                                                            );
                                                        }
                                                    }
                                                }
                                            }
                                        }
                                        Err(err) => {
                                            state.pubsub.send_to_client(
                                                client_id,
                                                WsMessage::Error { message: err },
                                            );
                                        }
                                    }
                                }
                                WsMessage::Unsubscribe {
                                    channel,
                                    symbols,
                                    expiry,
                                    option_type,
                                } => {
                                    match state.pubsub.unsubscribe(
                                        client_id,
                                        channel.clone(),
                                        symbols,
                                        expiry,
                                        option_type,
                                    ) {
                                        Ok(_) => {
                                            if channel == "portfolio" {
                                                if let Some(task) = portfolio_task.take() {
                                                    task.abort();
                                                }
                                                if let Some((wallet, subscriber_id)) =
                                                    portfolio_subscription.take()
                                                {
                                                    if let Some(ref cache) = state.portfolio_cache {
                                                        cache.unsubscribe(&wallet, subscriber_id).await;
                                                    }
                                                }
                                            }

                                            state.pubsub.send_to_client(
                                                client_id,
                                                WsMessage::Unsubscribed { channel },
                                            );
                                        }
                                        Err(err) => {
                                            state.pubsub.send_to_client(
                                                client_id,
                                                WsMessage::Error { message: err },
                                            );
                                        }
                                    }
                                }
                                WsMessage::Authenticate { wallet } => {
                                    match WalletAddress::from_str(&wallet) {
                                        Ok(addr) => {
                                            match state.pubsub.authenticate_client(client_id, addr) {
                                                Ok(_) => {
                                                    wallet_addr_clone = Some(addr);
                                                    authenticated = true;
                                                    state.pubsub.send_to_client(
                                                        client_id,
                                                        WsMessage::Authenticated {
                                                            wallet: wallet.clone(),
                                                        },
                                                    );
                                                }
                                                Err(err) => {
                                                    state.pubsub.send_to_client(
                                                        client_id,
                                                        WsMessage::Error { message: err },
                                                    );
                                                }
                                            }
                                        }
                                        Err(_) => {
                                            state.pubsub.send_to_client(
                                                client_id,
                                                WsMessage::Error {
                                                    message: format!(
                                                        "Invalid wallet address: {}",
                                                        wallet
                                                    ),
                                                },
                                            );
                                        }
                                    }
                                }
                                WsMessage::SubmitRfq {
                                    rfq_id,
                                    legs,
                                    wallet_address: req_wallet,
                                    nonce,
                                    signature,
                                } => {
                                    if !authenticated {
                                        state.pubsub.send_to_client(
                                            client_id,
                                            WsMessage::Error {
                                                message: "Not authenticated".to_string(),
                                            },
                                        );
                                    } else if let Some(ref rfq_state) = state.rfq_handler_state {
                                        // Convert WsRfqLegRequest -> RfqLegRequest
                                        let typed_legs: Vec<hypercall_types::RfqLegRequest> = legs
                                            .iter()
                                            .map(|l| hypercall_types::RfqLegRequest {
                                                instrument: l.instrument.clone(),
                                                side: l.side,
                                                size: l.size.clone(),
                                            })
                                            .collect();

                                        let wallet = match WalletAddress::from_str(&req_wallet) {
                                            Ok(w) => w,
                                            Err(_) => {
                                                state.pubsub.send_to_client(
                                                    client_id,
                                                    WsMessage::Error {
                                                        message: format!(
                                                            "Invalid wallet address: {}",
                                                            req_wallet
                                                        ),
                                                    },
                                                );
                                                continue;
                                            }
                                        };

                                        match super::handlers::rfq::submit_rfq_inner(
                                            &rfq_state.rfq_manager,
                                            rfq_state.agent_auth.as_ref(),
                                            &rfq_id,
                                            &typed_legs,
                                            wallet,
                                            nonce,
                                            &signature,
                                            state.signing_chain_id,
                                        )
                                        .await
                                        {
                                            Ok(record) => {
                                                state.pubsub.send_to_client(
                                                    client_id,
                                                    WsMessage::RfqStatusUpdate(WsRfqStatusUpdate {
                                                        rfq_id: record.rfq_id.to_string(),
                                                        status: format!("{:?}", record.status),
                                                        taker_wallet: record.taker_wallet,
                                                    }),
                                                );
                                            }
                                            Err(err) => {
                                                state.pubsub.send_to_client(
                                                    client_id,
                                                    WsMessage::Error {
                                                        message: format!(
                                                            "RFQ submission failed: {}",
                                                            err
                                                        ),
                                                    },
                                                );
                                            }
                                        }
                                    } else {
                                        state.pubsub.send_to_client(
                                            client_id,
                                            WsMessage::Error {
                                                message: "RFQ system not available".to_string(),
                                            },
                                        );
                                    }
                                }
                                WsMessage::SubmitAutoExecuteRfq {
                                    rfq_id,
                                    legs,
                                    wallet_address: req_wallet,
                                    limit_price,
                                    nonce,
                                    signature,
                                } => {
                                    tracing::info!(
                                        client_id = ?client_id,
                                        rfq_id = %rfq_id,
                                        limit_price = %limit_price,
                                        authenticated,
                                        "WS SubmitAutoExecuteRfq received"
                                    );
                                    if !authenticated {
                                        tracing::warn!(client_id = ?client_id, rfq_id = %rfq_id, "SubmitAutoExecuteRfq rejected: not authenticated");
                                        state.pubsub.send_to_client(
                                            client_id,
                                            WsMessage::Error {
                                                message: "Not authenticated".to_string(),
                                            },
                                        );
                                    } else if let Some(ref rfq_state) = state.rfq_handler_state {
                                        let typed_legs: Vec<hypercall_types::RfqLegRequest> = legs
                                            .iter()
                                            .map(|l| hypercall_types::RfqLegRequest {
                                                instrument: l.instrument.clone(),
                                                side: l.side,
                                                size: l.size.clone(),
                                            })
                                            .collect();

                                        let wallet = match WalletAddress::from_str(&req_wallet) {
                                            Ok(w) => w,
                                            Err(_) => {
                                                state.pubsub.send_to_client(
                                                    client_id,
                                                    WsMessage::Error {
                                                        message: format!(
                                                            "Invalid wallet address: {}",
                                                            req_wallet
                                                        ),
                                                    },
                                                );
                                                continue;
                                            }
                                        };

                                        match super::handlers::rfq::submit_auto_execute_rfq_inner(
                                            &rfq_state.rfq_manager,
                                            rfq_state.agent_auth.as_ref(),
                                            &rfq_id,
                                            &typed_legs,
                                            wallet,
                                            &limit_price,
                                            nonce,
                                            &signature,
                                            state.signing_chain_id,
                                        )
                                        .await
                                        {
                                            Ok(record) => {
                                                state.pubsub.send_to_client(
                                                    client_id,
                                                    WsMessage::RfqStatusUpdate(WsRfqStatusUpdate {
                                                        rfq_id: record.rfq_id.to_string(),
                                                        status: format!("{:?}", record.status),
                                                        taker_wallet: record.taker_wallet,
                                                    }),
                                                );
                                            }
                                            Err(err) => {
                                                state.pubsub.send_to_client(
                                                    client_id,
                                                    WsMessage::Error {
                                                        message: format!(
                                                            "Auto-execute RFQ submission failed: {}",
                                                            err
                                                        ),
                                                    },
                                                );
                                            }
                                        }
                                    } else {
                                        state.pubsub.send_to_client(
                                            client_id,
                                            WsMessage::Error {
                                                message: "RFQ system not available".to_string(),
                                            },
                                        );
                                    }
                                }
                                // TODO(rate-limit): WS PlaceOrder and AcceptRfqQuote bypass
                                // the HTTP `write_route_rate_limit_middleware`. The
                                // `RateLimitCache::check_and_increment` API is callable
                                // outside middleware, but `WsState` does not currently
                                // hold a `RateLimitCache` reference. Adding one requires
                                // threading it through `WsState` and plumbing the cache
                                // into the integrated server builder -- scope for a
                                // follow-up PR.
                                WsMessage::AcceptRfqQuote {
                                    rfq_id,
                                    quote_id,
                                    wallet_address: req_wallet,
                                    nonce,
                                    signature,
                                } => {
                                    if !authenticated {
                                        state.pubsub.send_to_client(
                                            client_id,
                                            WsMessage::Error {
                                                message: "Not authenticated".to_string(),
                                            },
                                        );
                                    } else if let Some(ref rfq_state) = state.rfq_handler_state {
                                        let wallet = match WalletAddress::from_str(&req_wallet) {
                                            Ok(w) => w,
                                            Err(_) => {
                                                state.pubsub.send_to_client(
                                                    client_id,
                                                    WsMessage::Error {
                                                        message: format!(
                                                            "Invalid wallet address: {}",
                                                            req_wallet
                                                        ),
                                                    },
                                                );
                                                continue;
                                            }
                                        };

                                        match super::handlers::rfq::accept_rfq_quote_inner(
                                            &rfq_state.rfq_manager,
                                            rfq_state.agent_auth.as_ref(),
                                            &rfq_id,
                                            &quote_id,
                                            wallet,
                                            nonce,
                                            &signature,
                                            state.signing_chain_id,
                                        )
                                        .await
                                        {
                                            Ok(response) => {
                                                state.pubsub.send_to_client(
                                                    client_id,
                                                    WsMessage::RfqAcceptResult {
                                                        rfq_id: response.rfq_id,
                                                        quote_id: response.quote_id,
                                                        status: response.status.as_str().to_string(),
                                                        fill_id: Some(response.fill_id),
                                                        // send_to_client is already targeted; no
                                                        // channel-level filtering needed.
                                                        taker_wallet: None,
                                                    },
                                                );
                                            }
                                            Err(err) => {
                                                state.pubsub.send_to_client(
                                                    client_id,
                                                    WsMessage::Error {
                                                        message: format!(
                                                            "RFQ accept failed: {}",
                                                            err
                                                        ),
                                                    },
                                                );
                                            }
                                        }
                                    } else {
                                        state.pubsub.send_to_client(
                                            client_id,
                                            WsMessage::Error {
                                                message: "RFQ system not available".to_string(),
                                            },
                                        );
                                    }
                                }
                                WsMessage::PlaceOrder {
                                    wallet: req_wallet,
                                    symbol,
                                    side,
                                    size,
                                    price,
                                    tif,
                                    client_id: req_client_id,
                                    nonce,
                                    signature,
                                    mmp_enabled,
                                    builder_code_address,
                                } => {
                                    if !authenticated {
                                        state.pubsub.send_to_client(
                                            client_id,
                                            WsMessage::Error {
                                                message: "Not authenticated".to_string(),
                                            },
                                        );
                                    } else if let Some(ref order_sender) = state.order_sender {
                                        match handle_ws_place_order(
                                            &state,
                                            order_sender,
                                            &req_wallet,
                                            &symbol,
                                            &side,
                                            &size,
                                            &price,
                                            tif.as_deref(),
                                            req_client_id.as_deref(),
                                            nonce,
                                            &signature,
                                            mmp_enabled.unwrap_or(false),
                                            builder_code_address.as_deref(),
                                        )
                                        .await
                                        {
                                            Ok(result) => {
                                                state.pubsub.send_to_client(
                                                    client_id,
                                                    WsMessage::OrderResult(result),
                                                );
                                            }
                                            Err(err) => {
                                                state.pubsub.send_to_client(
                                                    client_id,
                                                    WsMessage::Error {
                                                        message: format!(
                                                            "Order placement failed: {}",
                                                            err
                                                        ),
                                                    },
                                                );
                                            }
                                        }
                                    } else {
                                        state.pubsub.send_to_client(
                                            client_id,
                                            WsMessage::Error {
                                                message: "Order system not available"
                                                    .to_string(),
                                            },
                                        );
                                    }
                                }
                                _ => {}
                            }
                        }
                    }
                    Message::Ping(payload) => {
                        counter!("ht_ws_ping_received_total").increment(1);
                        if sender.send(Message::Pong(payload)).await.is_err() {
                            break;
                        }
                    }
                    Message::Pong(_) => {
                        last_pong = Instant::now();
                        counter!("ht_ws_pongs_received_total").increment(1);
                    }
                    Message::Close(_) => break,
                    _ => {}
                }
            }
            _ = heartbeat.tick() => {
                if Instant::now().duration_since(last_pong) >= pong_timeout {
                    counter!("ht_ws_heartbeat_timeouts_total").increment(1);
                    let _ = sender
                        .send(Message::Close(Some(CloseFrame {
                            code: close_code::POLICY,
                            reason: "pong timeout".into(),
                        })))
                        .await;
                    break;
                }

                counter!("ht_ws_pings_sent_total").increment(1);
                if sender.send(Message::Ping(Vec::new())).await.is_err() {
                    break;
                }
            }
        }
    }

    // Clean up
    if let Some(task) = portfolio_task.take() {
        task.abort();
    }
    if let Some((wallet, subscriber_id)) = portfolio_subscription.take() {
        if let Some(ref cache) = state.portfolio_cache {
            cache.unsubscribe(&wallet, subscriber_id).await;
        }
    }
    state.pubsub.remove_client(client_id);
}

/// Timeout for waiting for engine response on WS order placement (10 seconds).
const WS_ENGINE_RESPONSE_TIMEOUT: Duration = Duration::from_secs(10);

/// Handle a PlaceOrder message received via WebSocket.
///
/// Replicates the core logic of the REST `place_order` / `process_single_order`
/// handlers: signature recovery, agent authorization, validation, engine
/// dispatch, and response mapping.
#[allow(clippy::too_many_arguments)]
async fn handle_ws_place_order(
    state: &WsState,
    order_sender: &mpsc::Sender<hypercall_runtime_api::UnifiedEngineRequest>,
    wallet_str: &str,
    symbol: &str,
    side_str: &str,
    size_str: &str,
    price_str: &str,
    tif_str: Option<&str>,
    client_id: Option<&str>,
    nonce: u64,
    signature: &str,
    mmp_enabled: bool,
    builder_code_address: Option<&str>,
) -> Result<WsOrderResult, String> {
    use hypercall_auth::SignatureRecovery;
    use hypercall_runtime_api::{increment_pending_requests, UnifiedEngineRequest};
    use hypercall_types::utils::get_timestamp_millis;
    use hypercall_types::{
        to_contract_units_decimal, validate_price_precision, ParsedOptionSymbol,
        MAX_PRICE_SIGNIFICANT_FIGURES,
    };
    use hypercall_types::{OrderAction, OrderActionMessage, OrderInfo, TimeInForce};
    use rust_decimal::prelude::ToPrimitive;
    use rust_decimal_macros::dec;

    // Parse wallet address
    let wallet_address = WalletAddress::from_str(wallet_str)
        .map_err(|_| format!("Invalid wallet address: {}", wallet_str))?;

    // Parse side
    let side = match side_str.to_lowercase().as_str() {
        "buy" => Side::Buy,
        "sell" => Side::Sell,
        other => return Err(format!("Invalid side: {}", other)),
    };

    // Parse TIF
    let tif = match tif_str.unwrap_or("gtc").to_lowercase().as_str() {
        "gtc" => TimeInForce::GTC,
        "ioc" => TimeInForce::IOC,
        "fok" => TimeInForce::FOK,
        other => return Err(format!("Invalid tif: {}", other)),
    };

    // Recover signer from EIP-712 signature
    let tif_label = match tif {
        TimeInForce::GTC => "gtc",
        TimeInForce::IOC => "ioc",
        TimeInForce::FOK => "fok",
    };
    let signer_address = SignatureRecovery::recover_place_order_signer(
        wallet_address,
        symbol,
        &format!("{:?}", side),
        size_str,
        price_str,
        tif_label,
        client_id.unwrap_or(""),
        nonce,
        signature,
        state.signing_chain_id,
    )
    .map_err(|e| format!("Signature verification failed: {}", e))?;
    let signer_wallet = WalletAddress::from(signer_address);

    let is_authorized = state
        .agent_auth
        .is_agent_authorized(&wallet_address, &signer_wallet);
    if !is_authorized {
        return Err(format!(
            "Unauthorized: signer {} not authorized for wallet {}",
            signer_wallet, wallet_address
        ));
    }

    // Validate the symbol
    let _parsed_symbol =
        ParsedOptionSymbol::from_symbol(symbol).map_err(|e| format!("Invalid symbol: {}", e))?;

    // Check trading halt
    if let Some(ref trading_halt) = state.trading_halt {
        let halt_state = trading_halt.read().await;
        if let Some(reason) = halt_state.blocked_reason(symbol) {
            return Err(format!(
                "{}. Order placement is disabled while trading is halted.",
                reason
            ));
        }
    }

    // Parse and validate price
    let price =
        Decimal::from_str(price_str).map_err(|_| format!("Invalid price format: {}", price_str))?;
    if price <= dec!(0) {
        return Err("Price must be positive".to_string());
    }
    let price_f64 = price
        .to_f64()
        .ok_or_else(|| format!("Price {} cannot be represented as f64", price))?;
    validate_price_precision(price_f64, MAX_PRICE_SIGNIFICANT_FIGURES)
        .map_err(|e| format!("Invalid price precision: {}", e))?;

    // Parse and validate size
    let size =
        Decimal::from_str(size_str).map_err(|_| format!("Invalid size format: {}", size_str))?;
    if size <= dec!(0) {
        return Err("Size must be positive".to_string());
    }

    // Parse optional builder code address
    let builder_code_wallet = match builder_code_address {
        Some(addr) => Some(
            WalletAddress::from_str(addr)
                .map_err(|_| format!("Invalid builder_code_address: {}", addr))?,
        ),
        None => None,
    };

    // Build OrderInfo
    let order_info = OrderInfo {
        symbol: symbol.to_string(),
        price,
        size: to_contract_units_decimal(symbol, size),
        side,
        tif,
        client_id: client_id.map(|s| s.to_string()),
        order_id: None,
        is_perp: false,
        underlying: None,
        reduce_only: None,
        nonce: None,
        signature: None,
        mmp_enabled,
        builder_code_address: builder_code_wallet,
    };

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

    // Create response channel
    let (response_tx, mut response_rx) = mpsc::channel(1);

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

    // Send to engine
    increment_pending_requests();
    order_sender
        .send(engine_request)
        .await
        .map_err(|_| "Failed to send order to engine".to_string())?;

    // Wait for engine response
    let response = match tokio::time::timeout(WS_ENGINE_RESPONSE_TIMEOUT, response_rx.recv()).await
    {
        Ok(Some(resp)) => resp,
        Ok(None) => return Err("No response from engine".to_string()),
        Err(_) => return Err("Timeout waiting for engine response".to_string()),
    };

    Ok(WsOrderResult {
        order_id: response.order_id,
        status: format!("{:?}", response.status),
        symbol: response.info.symbol.clone(),
        side: format!("{:?}", response.info.side),
        price: response.info.price.to_string(),
        // Echo back the human-readable size the client submitted, not the
        // engine's internal contract units (which are multiplied by 1e8).
        size: size_str.to_string(),
        reason: response.reason.clone(),
    })
}

// Helper module for UUID
pub mod uuid {
    use std::sync::atomic::{AtomicU64, Ordering};

    static COUNTER: AtomicU64 = AtomicU64::new(0);

    #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
    pub struct Uuid(u64);

    impl Uuid {
        pub fn new_v4() -> Self {
            Uuid(COUNTER.fetch_add(1, Ordering::SeqCst))
        }
    }
}

#[cfg(test)]
mod tests {
    use super::{
        validate_ws_heartbeat_config, ws_heartbeat_config, PubSubManager,
        DEFAULT_WS_HEARTBEAT_INTERVAL, DEFAULT_WS_PONG_TIMEOUT,
    };
    use hypercall_types::ws_protocol::WsIndicativeMarketData;
    use tokio::time::Duration;

    #[test]
    fn test_ws_heartbeat_config_accepts_valid_custom_values() {
        let (heartbeat_interval, pong_timeout) = validate_ws_heartbeat_config(250, 750);

        assert_eq!(heartbeat_interval, Duration::from_millis(250));
        assert_eq!(pong_timeout, Duration::from_millis(750));
    }

    #[test]
    fn test_ws_heartbeat_config_rejects_zero_interval() {
        let (heartbeat_interval, pong_timeout) = validate_ws_heartbeat_config(0, 60_000);

        assert_eq!(heartbeat_interval, DEFAULT_WS_HEARTBEAT_INTERVAL);
        assert_eq!(pong_timeout, DEFAULT_WS_PONG_TIMEOUT);
    }

    #[test]
    fn test_ws_heartbeat_config_rejects_timeout_shorter_than_interval() {
        let (heartbeat_interval, pong_timeout) = validate_ws_heartbeat_config(500, 400);

        assert_eq!(heartbeat_interval, DEFAULT_WS_HEARTBEAT_INTERVAL);
        assert_eq!(pong_timeout, DEFAULT_WS_PONG_TIMEOUT);
    }

    #[test]
    fn test_ws_heartbeat_config_uses_defaults_without_runtime_config() {
        let (heartbeat_interval, pong_timeout) =
            ws_heartbeat_config(&hypercall_config::ApiRuntimeConfig::default());

        assert_eq!(heartbeat_interval, DEFAULT_WS_HEARTBEAT_INTERVAL);
        assert_eq!(pong_timeout, DEFAULT_WS_PONG_TIMEOUT);
    }

    #[test]
    fn test_indicative_market_data_honors_symbol_filter() {
        let pubsub = PubSubManager::new();
        let client_id = super::uuid::Uuid::new_v4();
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();

        pubsub.add_client(client_id, tx, false, None);
        pubsub
            .subscribe(
                client_id,
                "indicative_market_data".to_string(),
                Some(vec!["BTC-20260131-100000-C".to_string()]),
                None,
                None,
            )
            .expect("indicative subscription should succeed");

        pubsub.publish_indicative_market_data(WsIndicativeMarketData {
            instrument: "ETH-20260131-100000-C".to_string(),
            best_bid: Some(dec!(10)),
            bid_iv: None,
            ask_iv: None,
            best_ask: Some(dec!(11)),
            indicative_bid_size: Some(dec!(1)),
            indicative_ask_size: Some(dec!(1)),
            num_providers: 1,
            timestamp: 1,
        });
        assert!(rx.try_recv().is_err());

        pubsub.publish_indicative_market_data(WsIndicativeMarketData {
            instrument: "BTC-20260131-100000-C".to_string(),
            best_bid: Some(dec!(20)),
            bid_iv: None,
            ask_iv: None,
            best_ask: Some(dec!(21)),
            indicative_bid_size: Some(dec!(2)),
            indicative_ask_size: Some(dec!(2)),
            num_providers: 1,
            timestamp: 2,
        });

        let message = rx
            .try_recv()
            .expect("matching indicative quote should be delivered");
        match message.as_ref() {
            WsMessage::IndicativeMarketData(update) => {
                assert_eq!(update.instrument, "BTC-20260131-100000-C");
                assert_eq!(update.best_bid, Some(dec!(20)));
                assert_eq!(update.best_ask, Some(dec!(21)));
            }
            other => panic!("expected IndicativeMarketData, got {other:?}"),
        }
        assert!(rx.try_recv().is_err());
    }

    // --- WS message serialization round-trip tests ---

    use super::{
        WsMessage, WsOrderResult, WsRfqLegRequest, WsRfqQuoteEntry, WsRfqQuotes, WsRfqStatusUpdate,
    };
    use hypercall_types::{Side, WalletAddress};
    use rust_decimal_macros::dec;
    use std::str::FromStr;

    #[test]
    fn test_submit_rfq_roundtrip() {
        let msg = WsMessage::SubmitRfq {
            rfq_id: "test-123".to_string(),
            legs: vec![WsRfqLegRequest {
                instrument: "ETH-20260425-2400-C".to_string(),
                side: Side::Buy,
                size: "10".to_string(),
            }],
            wallet_address: "0x1234567890abcdef1234567890abcdef12345678".to_string(),
            nonce: 42,
            signature: "0xsig".to_string(),
        };
        let json = serde_json::to_string(&msg).unwrap();
        assert!(json.contains("\"type\":\"SubmitRfq\""));
        assert!(json.contains("\"rfq_id\":\"test-123\""));
        assert!(json.contains("\"nonce\":42"));

        let parsed: WsMessage = serde_json::from_str(&json).unwrap();
        match parsed {
            WsMessage::SubmitRfq {
                rfq_id,
                legs,
                wallet_address,
                nonce,
                signature,
            } => {
                assert_eq!(rfq_id, "test-123");
                assert_eq!(legs.len(), 1);
                assert_eq!(legs[0].instrument, "ETH-20260425-2400-C");
                assert_eq!(legs[0].size, "10");
                assert_eq!(wallet_address, "0x1234567890abcdef1234567890abcdef12345678");
                assert_eq!(nonce, 42);
                assert_eq!(signature, "0xsig");
            }
            other => panic!("Expected SubmitRfq, got {:?}", other),
        }
    }

    #[test]
    fn test_accept_rfq_quote_roundtrip() {
        let msg = WsMessage::AcceptRfqQuote {
            rfq_id: "rfq-abc".to_string(),
            quote_id: "quote-xyz".to_string(),
            wallet_address: "0xdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef".to_string(),
            nonce: 99,
            signature: "0xdeadsig".to_string(),
        };
        let json = serde_json::to_string(&msg).unwrap();
        assert!(json.contains("\"type\":\"AcceptRfqQuote\""));
        assert!(json.contains("\"rfq_id\":\"rfq-abc\""));
        assert!(json.contains("\"quote_id\":\"quote-xyz\""));

        let parsed: WsMessage = serde_json::from_str(&json).unwrap();
        match parsed {
            WsMessage::AcceptRfqQuote {
                rfq_id,
                quote_id,
                wallet_address,
                nonce,
                signature,
            } => {
                assert_eq!(rfq_id, "rfq-abc");
                assert_eq!(quote_id, "quote-xyz");
                assert_eq!(wallet_address, "0xdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef");
                assert_eq!(nonce, 99);
                assert_eq!(signature, "0xdeadsig");
            }
            other => panic!("Expected AcceptRfqQuote, got {:?}", other),
        }
    }

    #[test]
    fn test_place_order_roundtrip() {
        let msg = WsMessage::PlaceOrder {
            wallet: "0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".to_string(),
            symbol: "BTC-20260131-100000-C".to_string(),
            side: "Buy".to_string(),
            size: "1.5".to_string(),
            price: "5000".to_string(),
            tif: Some("gtc".to_string()),
            client_id: Some("my-order-1".to_string()),
            nonce: 1000,
            signature: "0xordersig".to_string(),
            mmp_enabled: Some(true),
            builder_code_address: None,
        };
        let json = serde_json::to_string(&msg).unwrap();
        assert!(json.contains("\"type\":\"PlaceOrder\""));
        assert!(json.contains("\"symbol\":\"BTC-20260131-100000-C\""));
        assert!(json.contains("\"side\":\"Buy\""));
        assert!(json.contains("\"price\":\"5000\""));
        // builder_code_address should be omitted via skip_serializing_if
        assert!(!json.contains("builder_code_address"));

        let parsed: WsMessage = serde_json::from_str(&json).unwrap();
        match parsed {
            WsMessage::PlaceOrder {
                wallet,
                symbol,
                side,
                size,
                price,
                tif,
                client_id,
                nonce,
                signature,
                mmp_enabled,
                builder_code_address,
            } => {
                assert_eq!(wallet, "0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa");
                assert_eq!(symbol, "BTC-20260131-100000-C");
                assert_eq!(side, "Buy");
                assert_eq!(size, "1.5");
                assert_eq!(price, "5000");
                assert_eq!(tif, Some("gtc".to_string()));
                assert_eq!(client_id, Some("my-order-1".to_string()));
                assert_eq!(nonce, 1000);
                assert_eq!(signature, "0xordersig");
                assert_eq!(mmp_enabled, Some(true));
                assert!(builder_code_address.is_none());
            }
            other => panic!("Expected PlaceOrder, got {:?}", other),
        }
    }

    #[test]
    fn test_place_order_minimal_fields_roundtrip() {
        // Test with only required fields (no tif, client_id, mmp_enabled, builder_code_address)
        let json = r#"{"type":"PlaceOrder","wallet":"0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","symbol":"ETH-20260228-3000-P","side":"Sell","size":"2","price":"100","nonce":1,"signature":"0xsig"}"#;
        let parsed: WsMessage = serde_json::from_str(json).unwrap();
        match parsed {
            WsMessage::PlaceOrder {
                tif,
                client_id,
                mmp_enabled,
                builder_code_address,
                ..
            } => {
                assert!(tif.is_none());
                assert!(client_id.is_none());
                assert!(mmp_enabled.is_none());
                assert!(builder_code_address.is_none());
            }
            other => panic!("Expected PlaceOrder, got {:?}", other),
        }
    }

    #[test]
    fn test_rfq_quotes_response_serialization() {
        let msg = WsMessage::RfqQuotes(WsRfqQuotes {
            rfq_id: "rfq-001".to_string(),
            quotes: vec![WsRfqQuoteEntry {
                quote_id: "q-1".to_string(),
                net_premium: dec!(150.50),
                expires_at: 1700000000,
            }],
            status: "quoted".to_string(),
            taker_wallet: WalletAddress::from_str("0xbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb")
                .unwrap(),
        });
        let json = serde_json::to_string(&msg).unwrap();
        assert!(json.contains("\"type\":\"RfqQuotes\""));
        assert!(json.contains("\"rfq_id\":\"rfq-001\""));
        assert!(json.contains("\"quote_id\":\"q-1\""));
        assert!(json.contains("\"status\":\"quoted\""));

        // Verify it round-trips
        let parsed: WsMessage = serde_json::from_str(&json).unwrap();
        match parsed {
            WsMessage::RfqQuotes(quotes) => {
                assert_eq!(quotes.rfq_id, "rfq-001");
                assert_eq!(quotes.quotes.len(), 1);
                assert_eq!(quotes.quotes[0].quote_id, "q-1");
                assert_eq!(quotes.status, "quoted");
            }
            other => panic!("Expected RfqQuotes, got {:?}", other),
        }
    }

    #[test]
    fn test_rfq_accept_result_serialization() {
        let msg = WsMessage::RfqAcceptResult {
            rfq_id: "rfq-002".to_string(),
            quote_id: "q-2".to_string(),
            status: "filled".to_string(),
            fill_id: Some("fill-abc".to_string()),
            taker_wallet: None,
        };
        let json = serde_json::to_string(&msg).unwrap();
        assert!(json.contains("\"type\":\"RfqAcceptResult\""));
        assert!(json.contains("\"rfq_id\":\"rfq-002\""));
        assert!(json.contains("\"fill_id\":\"fill-abc\""));
        // taker_wallet is skip_serializing so must NOT appear in JSON
        assert!(!json.contains("taker_wallet"));

        let parsed: WsMessage = serde_json::from_str(&json).unwrap();
        match parsed {
            WsMessage::RfqAcceptResult {
                rfq_id,
                quote_id,
                status,
                fill_id,
                ..
            } => {
                assert_eq!(rfq_id, "rfq-002");
                assert_eq!(quote_id, "q-2");
                assert_eq!(status, "filled");
                assert_eq!(fill_id, Some("fill-abc".to_string()));
            }
            other => panic!("Expected RfqAcceptResult, got {:?}", other),
        }
    }

    #[test]
    fn test_order_result_serialization() {
        let msg = WsMessage::OrderResult(WsOrderResult {
            order_id: Some(12345),
            status: "Open".to_string(),
            symbol: "BTC-20260131-100000-C".to_string(),
            side: "Buy".to_string(),
            price: "5000".to_string(),
            size: "1000000".to_string(),
            reason: None,
        });
        let json = serde_json::to_string(&msg).unwrap();
        assert!(json.contains("\"type\":\"OrderResult\""));
        assert!(json.contains("\"order_id\":12345"));
        assert!(json.contains("\"status\":\"Open\""));
        assert!(json.contains("\"symbol\":\"BTC-20260131-100000-C\""));
        // reason should be omitted via skip_serializing_if
        assert!(!json.contains("\"reason\""));

        let parsed: WsMessage = serde_json::from_str(&json).unwrap();
        match parsed {
            WsMessage::OrderResult(result) => {
                assert_eq!(result.order_id, Some(12345));
                assert_eq!(result.status, "Open");
                assert_eq!(result.symbol, "BTC-20260131-100000-C");
                assert_eq!(result.side, "Buy");
                assert!(result.reason.is_none());
            }
            other => panic!("Expected OrderResult, got {:?}", other),
        }
    }

    #[test]
    fn test_order_result_with_rejection_reason() {
        let msg = WsMessage::OrderResult(WsOrderResult {
            order_id: None,
            status: "Rejected".to_string(),
            symbol: "ETH-20260228-3000-P".to_string(),
            side: "Sell".to_string(),
            price: "100".to_string(),
            size: "500000".to_string(),
            reason: Some("Insufficient margin".to_string()),
        });
        let json = serde_json::to_string(&msg).unwrap();
        assert!(json.contains("\"reason\":\"Insufficient margin\""));
        assert!(json.contains("\"order_id\":null"));

        let parsed: WsMessage = serde_json::from_str(&json).unwrap();
        match parsed {
            WsMessage::OrderResult(result) => {
                assert!(result.order_id.is_none());
                assert_eq!(result.reason, Some("Insufficient margin".to_string()));
            }
            other => panic!("Expected OrderResult, got {:?}", other),
        }
    }

    #[test]
    fn test_rfq_status_update_serialization() {
        let msg = WsMessage::RfqStatusUpdate(WsRfqStatusUpdate {
            rfq_id: "rfq-status-test".to_string(),
            status: "expired".to_string(),
            taker_wallet: WalletAddress::from_str("0xcccccccccccccccccccccccccccccccccccccccc")
                .unwrap(),
        });
        let json = serde_json::to_string(&msg).unwrap();
        assert!(json.contains("\"type\":\"RfqStatusUpdate\""));
        assert!(json.contains("\"status\":\"expired\""));

        let parsed: WsMessage = serde_json::from_str(&json).unwrap();
        match parsed {
            WsMessage::RfqStatusUpdate(update) => {
                assert_eq!(update.rfq_id, "rfq-status-test");
                assert_eq!(update.status, "expired");
            }
            other => panic!("Expected RfqStatusUpdate, got {:?}", other),
        }
    }
}