rusty_feeder/exchange/bybit/futures/

provider.rs

//! Bybit Futures market provider implementation
//!
//! High-performance provider for Bybit Futures markets with WebSocket streaming
//! and REST API support. Optimized for HFT applications with nanosecond precision.

use rusty_common::collections::FxHashMap;
use smartstring::alias::String;
use std::sync::Arc;
use std::time::{Duration, Instant};

use super::data::{MarketTradeResponse, orderbook::OrderbookResponse};
use anyhow::{Context, Result, anyhow};
use async_trait::async_trait;
use parking_lot::RwLock;
use quanta::Clock;
use reqwest::header::{HeaderMap, HeaderValue};
use rusty_common::json::Value;
use rusty_common::websocket::{Message, WebSocketClient, WebSocketConfig, WebSocketError};
use rusty_model::{
    data::{
        book_snapshot::OrderBookSnapshot, market_trade::MarketTrade,
        simd_orderbook::SharedSimdOrderBook,
    },
    instruments::{Instrument, InstrumentId},
    venues::Venue,
};
use simd_json::prelude::{ValueAsArray, ValueAsScalar, ValueObjectAccess};
use smallvec::SmallVec;
use tokio::sync::{mpsc, watch};
use tokio::task::JoinHandle;

use crate::provider::prelude::*;

// Constants for Bybit Futures API
const BYBIT_FUTURES_API_URL: &str = "https://api.bybit.com";
const BYBIT_FUTURES_WS_URL: &str = "wss://stream.bybit.com/v5/public/linear";
const _BYBIT_FUTURES_TESTNET_API_URL: &str = "https://api-testnet.bybit.com";
const _BYBIT_FUTURES_TESTNET_WS_URL: &str = "wss://stream-testnet.bybit.com/v5/public/linear";

// Rate limits for Bybit Futures API
const BYBIT_FUTURES_RATE_LIMITS: &[RateLimit] = &[
    RateLimit {
        limit_type: "REQUEST_WEIGHT",
        interval: "MINUTE",
        interval_num: 1,
        limit: 120,
    },
    RateLimit {
        limit_type: "ORDERS",
        interval: "MINUTE",
        interval_num: 1,
        limit: 100,
    },
];

/// Bybit Futures instrument implementation
#[derive(Debug, Clone)]
pub struct BybitFuturesInstrument {
    /// Unique instrument identifier
    pub id: InstrumentId,
}

impl Instrument for BybitFuturesInstrument {
    fn id(&self) -> InstrumentId {
        self.id.clone()
    }

    fn symbol(&self) -> String {
        self.id.symbol.clone()
    }

    fn venue(&self) -> Venue {
        self.id.venue
    }

    fn as_any(&self) -> &dyn std::any::Any {
        self
    }

    fn clone_box(&self) -> Box<dyn Instrument> {
        Box::new(self.clone())
    }
}

/// Subscription message for Bybit Futures WebSocket
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct SubscriptionMessage {
    op: String,
    args: Vec<String>,
}

impl SubscriptionMessage {
    /// Create subscription for trades
    fn trades(symbols: Vec<String>) -> Self {
        let args = symbols
            .into_iter()
            .map(|s| format!("publicTrade.{s}").into())
            .collect();

        Self {
            op: "subscribe".into(),
            args,
        }
    }

    /// Create subscription for orderbook
    fn orderbook(symbols: Vec<String>) -> Self {
        let args = symbols
            .into_iter()
            .map(|s| format!("orderbook.50.{s}").into())
            .collect();

        Self {
            op: "subscribe".into(),
            args,
        }
    }
}

/// Bybit Futures market data provider
///
/// High-performance provider for Bybit Futures markets with WebSocket streaming
/// and REST API support. Optimized for HFT applications with nanosecond precision.
#[derive(Debug)]
pub struct BybitFuturesProvider {
    /// Connection configuration
    config: ConnectionConfig,

    /// WebSocket connection status
    connection_status: Arc<RwLock<ConnectionState>>,

    /// Connection statistics
    stats: Arc<RwLock<ConnectionStats>>,

    /// Shared clock for time synchronization
    clock: Clock,

    /// Active subscriptions
    subscriptions: Arc<RwLock<FxHashMap<String, watch::Sender<bool>>>>,

    /// Task handles for WebSocket connections
    ws_handles: Arc<RwLock<FxHashMap<String, JoinHandle<()>>>>,

    /// Last connection attempt timestamp
    last_connection_attempt: Arc<RwLock<Instant>>,

    /// HTTP client for REST API requests
    http_client: reqwest::Client,
}

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

impl BybitFuturesProvider {
    /// Create a new Bybit Futures provider with default configuration
    #[inline]
    #[must_use]
    pub fn new() -> Self {
        Self::with_config(None)
    }

    /// Create a new Bybit Futures provider with custom configuration
    #[inline]
    #[must_use]
    pub fn with_config(config: Option<ConnectionConfig>) -> Self {
        let mut default_config = ConnectionConfig::default();

        // Set default WebSocket configuration
        default_config.websocket_config.base_url = BYBIT_FUTURES_WS_URL.into();
        default_config.websocket_config.ping_interval_milliseconds = 20000; // 20 seconds
        default_config.websocket_config.use_compression = false;

        // Set default REST API configuration
        default_config.rest_config.base_url = BYBIT_FUTURES_API_URL.into();
        default_config.rest_config.timeout_milliseconds = 5000; // 5 seconds

        // Use custom config if provided
        let config = config.unwrap_or(default_config);
        let clock = config.clock.clone();

        // Create HTTP client with appropriate settings
        let mut headers = HeaderMap::new();
        headers.insert(
            "User-Agent",
            HeaderValue::from_str(&config.rest_config.user_agent)
                .unwrap_or_else(|_| HeaderValue::from_static("RustyHFT/1.0")),
        );

        let http_client = reqwest::Client::builder()
            .timeout(Duration::from_millis(
                config.rest_config.timeout_milliseconds,
            ))
            .connect_timeout(Duration::from_millis(
                config.rest_config.timeout_milliseconds / 2,
            ))
            .pool_max_idle_per_host(config.rest_config.connection_pool_size)
            .pool_idle_timeout(Duration::from_millis(
                config.rest_config.keep_alive_milliseconds,
            ))
            .default_headers(headers)
            .build()
            .unwrap_or_default();

        Self {
            config,
            connection_status: Arc::new(RwLock::new(ConnectionState::Disconnected)),
            stats: Arc::new(RwLock::new(ConnectionStats::default())),
            clock,
            subscriptions: Arc::new(RwLock::new(FxHashMap::default())),
            ws_handles: Arc::new(RwLock::new(FxHashMap::default())),
            last_connection_attempt: Arc::new(RwLock::new(Instant::now())),
            http_client,
        }
    }

    /// Helper function to update connection statistics upon message reception
    #[inline]
    fn update_receive_stats(
        stats: Arc<RwLock<ConnectionStats>>,
        message_size: usize,
        local_time: u64,
    ) {
        let mut s = stats.write();
        s.messages_received += 1;
        s.bytes_received += message_size as u64;
        s.last_message_time = local_time;
    }

    /// Helper function to create WebSocket request with proper headers
    fn create_websocket_config(
        url: String,
        use_compression: bool,
        ping_interval_milliseconds: u64,
        timeout_milliseconds: u64,
    ) -> WebSocketConfig {
        WebSocketConfig::builder(rusty_common::types::Exchange::Bybit, url.to_string())
            .ping_interval(Duration::from_millis(ping_interval_milliseconds))
            .timeout(Duration::from_millis(timeout_milliseconds))
            .compression(if use_compression {
                rusty_common::websocket::CompressionConfig::default()
            } else {
                rusty_common::websocket::CompressionConfig::disabled()
            })
            .build()
    }

    /// Connect to the WebSocket API
    #[inline]
    async fn connect(&self) -> Result<()> {
        if *self.connection_status.read() == ConnectionState::Connected {
            return Ok(());
        }

        // Check if we need to respect reconnection backoff
        {
            let now = Instant::now();
            let last_attempt = *self.last_connection_attempt.read();
            let backoff_ms = self
                .config
                .websocket_config
                .reconnect
                .backoff_initial_milliseconds;

            if now.duration_since(last_attempt) < Duration::from_millis(backoff_ms) {
                // Wait until backoff period is over
                tokio::time::sleep(Duration::from_millis(backoff_ms)).await;
            }

            // Update last attempt time
            *self.last_connection_attempt.write() = Instant::now();
        }

        *self.connection_status.write() = ConnectionState::Connecting;

        // Note: Connection is now handled by individual trade/orderbook connections
        // using WebSocketClient pattern. This method is kept for compatibility.
        *self.connection_status.write() = ConnectionState::Connected;
        self.stats.write().connected_time = self.clock.raw();
        Ok(())
    }

    /// Create a WebSocket connection for trade data
    #[inline]
    async fn create_trade_connection(
        &self,
        symbols: SmallVec<[String; 8]>,
        tx: mpsc::Sender<MarketTradeResponse>,
        subscription: SubscriptionMessage,
    ) -> Result<JoinHandle<()>> {
        let connection_id = format!("trade-{}", symbols.join(","));

        // Create stop signal
        let (stop_tx, stop_rx) = watch::channel(false);
        self.subscriptions
            .write()
            .insert(connection_id.clone().into(), stop_tx);

        // Clone parameters for the task
        let url = self.config.websocket_config.base_url.clone();
        let clock = self.clock.clone();
        let timeout_milliseconds = self.config.websocket_config.timeout_milliseconds;
        let ping_interval_milliseconds = self.config.websocket_config.ping_interval_milliseconds;
        let stats = self.stats.clone();
        let connection_status = self.connection_status.clone();
        let use_compression = self.config.websocket_config.use_compression;

        // Spawn a task for the WebSocket connection
        let handle = tokio::spawn(async move {
            loop {
                // Check for stop signal
                if *stop_rx.borrow() {
                    break;
                }

                // Create WebSocket configuration
                let ws_config = Self::create_websocket_config(
                    url.clone(),
                    use_compression,
                    ping_interval_milliseconds,
                    timeout_milliseconds,
                );

                // Create WebSocket client
                let mut client = WebSocketClient::new(ws_config);

                // Update connection status
                *connection_status.write() = ConnectionState::Connecting;

                // Create a simple message handler
                let handler = BybitFuturesMessageHandler::new_trade_handler(
                    clock.clone(),
                    stats.clone(),
                    tx.clone(),
                    subscription.clone(),
                );

                // Run the WebSocket client
                if let Err(e) = client.run(handler).await {
                    log::error!("WebSocket client error: {e}");
                    *connection_status.write() = ConnectionState::Error;
                }

                // Check for stop signal before reconnecting
                if *stop_rx.borrow() {
                    break;
                }

                // Reconnect after a delay
                tokio::time::sleep(Duration::from_millis(1000)).await;
            }
        });

        Ok(handle)
    }

    /// Create a WebSocket connection for orderbook data
    #[inline]
    async fn create_orderbook_connection(
        &self,
        symbols: SmallVec<[String; 8]>,
        tx: mpsc::Sender<OrderbookResponse>,
        subscription: SubscriptionMessage,
    ) -> Result<JoinHandle<()>> {
        let connection_id = format!("orderbook-{}", symbols.join(","));

        // Create stop signal
        let (stop_tx, stop_rx) = watch::channel(false);
        self.subscriptions
            .write()
            .insert(connection_id.clone().into(), stop_tx);

        // Clone parameters for the task
        let url = self.config.websocket_config.base_url.clone();
        let clock = self.clock.clone();
        let timeout_milliseconds = self.config.websocket_config.timeout_milliseconds;
        let ping_interval_milliseconds = self.config.websocket_config.ping_interval_milliseconds;
        let stats = self.stats.clone();
        let connection_status = self.connection_status.clone();
        let use_compression = self.config.websocket_config.use_compression;

        // Spawn a task for the WebSocket connection
        let handle = tokio::spawn(async move {
            loop {
                // Check for stop signal
                if *stop_rx.borrow() {
                    break;
                }

                // Create WebSocket configuration
                let ws_config = Self::create_websocket_config(
                    url.clone(),
                    use_compression,
                    ping_interval_milliseconds,
                    timeout_milliseconds,
                );

                // Create WebSocket client
                let mut client = WebSocketClient::new(ws_config);

                // Update connection status
                *connection_status.write() = ConnectionState::Connecting;

                // Create a simple message handler
                let handler = BybitFuturesMessageHandler::new_orderbook_handler(
                    clock.clone(),
                    stats.clone(),
                    tx.clone(),
                    subscription.clone(),
                );

                // Run the WebSocket client
                if let Err(e) = client.run(handler).await {
                    log::error!("WebSocket client error: {e}");
                    *connection_status.write() = ConnectionState::Error;
                }

                // Check for stop signal before reconnecting
                if *stop_rx.borrow() {
                    break;
                }

                // Reconnect after a delay
                tokio::time::sleep(Duration::from_millis(1000)).await;
            }
        });

        Ok(handle)
    }

    /// Fetch orderbook snapshot for a symbol
    async fn fetch_orderbook_snapshot(
        &self,
        symbol: &str,
        depth: Option<u32>,
    ) -> Result<OrderBookSnapshot> {
        let limit = depth.unwrap_or(50);
        let url = format!(
            "{}/v5/market/orderbook?category=linear&symbol={}&limit={}",
            self.config.rest_config.base_url, symbol, limit
        );

        let response = self.http_client.get(&url).send().await?;

        if !response.status().is_success() {
            let error_text = response.text().await?;
            return Err(anyhow!(
                "Failed to fetch orderbook snapshot: {}",
                error_text
            ));
        }

        // Parse the response using simd_json
        let bytes = response
            .bytes()
            .await
            .context("Failed to get response bytes")?;
        let mut bytes_vec = bytes.to_vec();
        let response_json: simd_json::OwnedValue =
            simd_json::from_slice(&mut bytes_vec).context("Failed to parse JSON")?;

        // Extract data
        let result = response_json
            .get("result")
            .ok_or_else(|| anyhow!("Invalid response format"))?;

        let update_id = result.get("u").and_then(|v| v.as_u64()).unwrap_or(0);

        let bids = result
            .get("b")
            .and_then(|b| b.as_array())
            .ok_or_else(|| anyhow!("Missing bids"))?;

        let asks = result
            .get("a")
            .and_then(|a| a.as_array())
            .ok_or_else(|| anyhow!("Missing asks"))?;

        // Create OrderBookSnapshot with current timestamp (nanoseconds since epoch)
        let timestamp = self.clock.raw();
        let instrument_id = InstrumentId::new(symbol, Venue::Bybit);

        let mut order_book_snapshot =
            OrderBookSnapshot::new_empty(instrument_id, timestamp, update_id);

        // Parse bids
        for bid in bids {
            if let Some(bid_array) = bid.as_array()
                && bid_array.len() >= 2
                && let (Some(price_str), Some(quantity_str)) =
                    (bid_array[0].as_str(), bid_array[1].as_str())
                && let (Ok(price), Ok(quantity)) = (
                    price_str.parse::<rust_decimal::Decimal>(),
                    quantity_str.parse::<rust_decimal::Decimal>(),
                )
            {
                order_book_snapshot.add_bid(price, quantity);
            }
        }

        // Parse asks
        for ask in asks {
            if let Some(ask_array) = ask.as_array()
                && ask_array.len() >= 2
                && let (Some(price_str), Some(quantity_str)) =
                    (ask_array[0].as_str(), ask_array[1].as_str())
                && let (Ok(price), Ok(quantity)) = (
                    price_str.parse::<rust_decimal::Decimal>(),
                    quantity_str.parse::<rust_decimal::Decimal>(),
                )
            {
                order_book_snapshot.add_ask(price, quantity);
            }
        }

        Ok(order_book_snapshot)
    }
}

#[async_trait]
impl Provider for BybitFuturesProvider {
    type TradeMessage = MarketTradeResponse;
    type DepthMessage = OrderbookResponse;
    type InstrumentMessage = Value;

    fn name(&self) -> &'static str {
        "Bybit Futures"
    }

    fn venue(&self) -> Venue {
        Venue::Bybit
    }

    fn config(&self) -> &ConnectionConfig {
        &self.config
    }

    async fn init(&mut self) -> Result<()> {
        self.connect().await
    }

    async fn subscribe_trades(
        &self,
        symbols: SmallVec<[String; 8]>,
    ) -> Result<mpsc::Receiver<Self::TradeMessage>> {
        // Ensure connection is established
        self.connect().await?;

        // Create channel for trade messages
        let (tx, rx) = mpsc::channel(1024);

        // Create subscription message
        let subscription = SubscriptionMessage::trades(symbols.to_vec());
        let connection_id = format!("trade-{}", symbols.join(","));

        // Create WebSocket connection and handle
        let handle = self
            .create_trade_connection(symbols, tx, subscription)
            .await?;

        // Save task handle
        self.ws_handles.write().insert(connection_id.into(), handle);

        Ok(rx)
    }

    async fn unsubscribe_trades(&self) -> Result<()> {
        // Find all trade subscriptions
        let mut keys_to_remove = Vec::new();

        for (key, _) in self.subscriptions.read().iter() {
            if key.starts_with("trade-") {
                keys_to_remove.push(key.clone());
            }
        }

        // Send stop signal for each subscription
        for key in keys_to_remove {
            if let Some(tx) = self.subscriptions.write().remove(&key) {
                let _ = tx.send(true);
            }

            // Abort task
            if let Some(handle) = self.ws_handles.write().remove(&key) {
                handle.abort();
            }
        }

        Ok(())
    }

    async fn subscribe_orderbook(
        &self,
        symbols: SmallVec<[String; 8]>,
    ) -> Result<mpsc::Receiver<Self::DepthMessage>> {
        // Ensure connection is established
        self.connect().await?;

        // Create channel for orderbook messages
        let (tx, rx) = mpsc::channel(1024);

        // Create subscription message for level2 updates
        let subscription = SubscriptionMessage::orderbook(symbols.to_vec());
        let connection_id = format!("orderbook-{}", symbols.join(","));

        // Create WebSocket connection and handle
        let handle = self
            .create_orderbook_connection(symbols, tx, subscription)
            .await?;

        // Save task handle
        self.ws_handles.write().insert(connection_id.into(), handle);

        Ok(rx)
    }

    async fn unsubscribe_orderbook(&self) -> Result<()> {
        // Find all orderbook subscriptions
        let mut keys_to_remove = Vec::new();

        for (key, _) in self.subscriptions.read().iter() {
            if key.starts_with("orderbook-") {
                keys_to_remove.push(key.clone());
            }
        }

        // Send stop signal for each subscription
        for key in keys_to_remove {
            if let Some(tx) = self.subscriptions.write().remove(&key) {
                let _ = tx.send(true);
            }

            // Abort task
            if let Some(handle) = self.ws_handles.write().remove(&key) {
                handle.abort();
            }
        }

        Ok(())
    }

    #[inline]
    async fn get_instruments(&self) -> Result<Vec<Box<dyn Instrument>>> {
        // Fetch instruments from Bybit API
        let url = format!(
            "{}/v5/market/instruments-info?category=linear",
            self.config.rest_config.base_url
        );

        let response = self
            .http_client
            .get(&url)
            .send()
            .await
            .context("Failed to fetch Bybit instruments info")?;

        // Check if the request was successful
        if !response.status().is_success() {
            return Err(anyhow!(
                "Failed to fetch Bybit instruments info: HTTP {}",
                response.status()
            ));
        }

        // Parse the response using simd_json
        let bytes = response
            .bytes()
            .await
            .context("Failed to get response bytes")?;
        let mut bytes_vec = bytes.to_vec();
        let instruments_info: simd_json::OwnedValue = simd_json::from_slice(&mut bytes_vec)
            .context("Failed to parse Bybit instruments info response")?;

        // Extract the instruments from the response
        let result = instruments_info
            .get("result")
            .ok_or_else(|| anyhow!("Invalid response format: missing result"))?;

        let list = result
            .get("list")
            .and_then(|l| l.as_array())
            .ok_or_else(|| anyhow!("Invalid response format: expected array of instruments"))?;

        // Convert to instruments
        let mut instruments = Vec::with_capacity(list.len());

        for instrument_data in list {
            // Skip instruments that are not trading
            let status = instrument_data
                .get("status")
                .and_then(|s| s.as_str())
                .unwrap_or("");
            if status != "Trading" {
                continue;
            }

            let symbol = instrument_data
                .get("symbol")
                .and_then(|s| s.as_str())
                .ok_or_else(|| {
                    anyhow!("Invalid instrument data: 'symbol' field not found or not a String")
                })?;

            // Create a basic instrument implementation
            let instrument_id = InstrumentId::new(symbol, Venue::Bybit);
            let instrument =
                Box::new(BybitFuturesInstrument { id: instrument_id }) as Box<dyn Instrument>;
            instruments.push(instrument);
        }

        Ok(instruments)
    }

    #[inline]
    async fn get_historical_trades(
        &self,
        _symbol: &str,
        _limit: Option<u32>,
    ) -> Result<Vec<MarketTrade>> {
        // Fetch historical trades from Bybit API
        // Not fully implemented yet
        Err(anyhow!("Historical trades not implemented yet"))
    }

    async fn get_orderbook_snapshot(
        &self,
        symbol: &str,
        depth: Option<u32>,
    ) -> Result<OrderBookSnapshot> {
        self.fetch_orderbook_snapshot(symbol, depth).await
    }

    #[inline]
    async fn get_realtime_orderbook(&self, _symbol: &str) -> Result<SharedSimdOrderBook> {
        // Not yet implemented
        Err(anyhow!("Real-time orderbook not implemented yet"))
    }

    #[inline]
    async fn is_connected(&self) -> bool {
        *self.connection_status.read() == ConnectionState::Connected
    }

    #[inline]
    async fn connection_status(&self) -> ConnectionState {
        *self.connection_status.read()
    }

    #[inline]
    async fn get_stats(&self) -> ConnectionStats {
        self.stats.read().clone()
    }

    #[inline]
    async fn ping(&self) -> Result<u64> {
        // For WebSocketClient pattern, ping/pong is handled automatically
        // Return a simple network latency test via HTTP
        let start = self.clock.raw();

        let response = self
            .http_client
            .get(format!(
                "{}/v5/market/time",
                self.config.rest_config.base_url
            ))
            .send()
            .await?;

        if response.status().is_success() {
            Ok(self.clock.raw().saturating_sub(start))
        } else {
            Err(anyhow!("Ping failed with status: {}", response.status()))
        }
    }

    #[inline]
    async fn reset_connection(&self) -> Result<()> {
        // First, unsubscribe all active subscriptions
        let mut keys_to_remove = Vec::new();

        for (key, _) in self.subscriptions.read().iter() {
            keys_to_remove.push(key.clone());
        }

        // Send stop signal for each subscription
        for key in &keys_to_remove {
            if let Some(tx) = self.subscriptions.write().remove(key) {
                let _ = tx.send(true);
            }

            // Abort task
            if let Some(handle) = self.ws_handles.write().remove(key) {
                handle.abort();
            }
        }

        // Set status to disconnected
        *self.connection_status.write() = ConnectionState::Disconnected;

        // Reset stats
        *self.stats.write() = ConnectionStats::default();

        // Connect again
        self.connect().await
    }

    fn get_rate_limits(&self) -> Vec<RateLimit> {
        BYBIT_FUTURES_RATE_LIMITS.to_vec()
    }

    async fn shutdown(&mut self) -> Result<()> {
        // Unsubscribe from all active subscriptions
        let mut keys_to_remove = Vec::new();

        for (key, _) in self.subscriptions.read().iter() {
            keys_to_remove.push(key.clone());
        }

        // Send stop signal for each subscription
        for key in keys_to_remove {
            if let Some(tx) = self.subscriptions.write().remove(&key) {
                let _ = tx.send(true);
            }

            // Abort task
            if let Some(handle) = self.ws_handles.write().remove(&key) {
                handle.abort();
            }
        }

        // Set connection status to disconnected
        *self.connection_status.write() = ConnectionState::Disconnected;

        Ok(())
    }
}

/// Simple message handler for Bybit Futures WebSocket messages
#[derive(Debug)]
struct BybitFuturesMessageHandler {
    handler_type: HandlerType,
    clock: Clock,
    stats: Arc<RwLock<ConnectionStats>>,
    trade_tx: Option<mpsc::Sender<MarketTradeResponse>>,
    orderbook_tx: Option<mpsc::Sender<OrderbookResponse>>,
    subscription: SubscriptionMessage,
}

#[derive(Debug)]
enum HandlerType {
    Trade,
    Orderbook,
}

impl BybitFuturesMessageHandler {
    const fn new_trade_handler(
        clock: Clock,
        stats: Arc<RwLock<ConnectionStats>>,
        tx: mpsc::Sender<MarketTradeResponse>,
        subscription: SubscriptionMessage,
    ) -> Self {
        Self {
            handler_type: HandlerType::Trade,
            clock,
            stats,
            trade_tx: Some(tx),
            orderbook_tx: None,
            subscription,
        }
    }

    const fn new_orderbook_handler(
        clock: Clock,
        stats: Arc<RwLock<ConnectionStats>>,
        tx: mpsc::Sender<OrderbookResponse>,
        subscription: SubscriptionMessage,
    ) -> Self {
        Self {
            handler_type: HandlerType::Orderbook,
            clock,
            stats,
            trade_tx: None,
            orderbook_tx: Some(tx),
            subscription,
        }
    }
}

#[async_trait]
impl rusty_common::websocket::MessageHandler for BybitFuturesMessageHandler {
    async fn on_message(&mut self, message: Message) -> std::result::Result<(), WebSocketError> {
        let local_time = self.clock.raw();

        // Only process text messages
        if let Some(text) = message.as_text() {
            BybitFuturesProvider::update_receive_stats(self.stats.clone(), text.len(), local_time);

            // Parse the message as JSON first
            let mut message_bytes = text.as_bytes().to_vec();
            let json_value: simd_json::OwnedValue = simd_json::from_slice(&mut message_bytes)
                .map_err(|e| {
                    WebSocketError::MessageProcessingError(format!("Failed to parse JSON: {e}"))
                })?;

            // Check if this is a data message or a response
            if let Some(topic) = json_value["topic"].as_str() {
                match self.handler_type {
                    HandlerType::Trade if topic.starts_with("publicTrade.") => {
                        if let Ok(trade_response) =
                            simd_json::from_slice::<MarketTradeResponse>(&mut message_bytes)
                            && let Some(ref tx) = self.trade_tx
                            && let Err(e) = tx.send(trade_response).await
                        {
                            log::warn!("Failed to send trade message: {e}");
                        }
                    }
                    HandlerType::Orderbook if topic.starts_with("orderbook.") => {
                        if let Ok(orderbook_response) =
                            simd_json::from_slice::<OrderbookResponse>(&mut message_bytes)
                            && let Some(ref tx) = self.orderbook_tx
                            && let Err(e) = tx.send(orderbook_response).await
                        {
                            log::warn!("Failed to send orderbook message: {e}");
                        }
                    }
                    _ => {
                        // Handle other message types (subscriptions, heartbeat, error)
                        if let Some(op) = json_value["op"].as_str() {
                            match op {
                                "subscribe" => {
                                    log::info!("Bybit Futures subscription confirmed: {text}");
                                }
                                "pong" => {
                                    log::trace!("Bybit Futures pong received");
                                }
                                _ => {
                                    log::debug!("Unhandled Bybit Futures message op: {op}");
                                }
                            }
                        }
                    }
                }
            }
        }

        Ok(())
    }

    async fn on_connected(&mut self) -> std::result::Result<(), WebSocketError> {
        // Send subscription message after connection
        let subscription_json = simd_json::to_string(&self.subscription).map_err(|e| {
            WebSocketError::MessageProcessingError(format!("Failed to serialize subscription: {e}"))
        })?;

        log::info!("Sending Bybit Futures subscription: {subscription_json}");

        // Note: The actual sending would be handled by the WebSocketClient
        // This is just for logging purposes
        Ok(())
    }

    async fn on_disconnected(&mut self) -> std::result::Result<(), WebSocketError> {
        log::info!("Disconnected from Bybit Futures WebSocket");
        Ok(())
    }

    async fn on_error(&mut self, error: WebSocketError) -> std::result::Result<(), WebSocketError> {
        log::error!("Bybit Futures WebSocket error: {error}");
        Ok(())
    }
}