rusty_feeder/exchange/coinbase/

provider.rs

//! Coinbase exchange provider implementation
//!
//! High-performance provider for Coinbase Advanced Trading API 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::{
    orderbook::Level2Update, subscription::SubscriptionMessage, trade::TradeMessage,
};
use super::types::{COINBASE_API_URL, COINBASE_RATE_LIMITS, COINBASE_WS_ADVANCED_URL};
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};
use smallvec::SmallVec;
use tokio::sync::{mpsc, watch};
use tokio::task::JoinHandle;

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

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

impl Instrument for CoinbaseInstrument {
    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())
    }
}

/// Coinbase Advanced Trading provider implementation
///
/// High-performance provider for Coinbase Advanced Trading with WebSocket streaming
/// and REST API support. Optimized for HFT applications with nanosecond precision.
#[derive(Debug)]
pub struct CoinbaseProvider {
    /// 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 CoinbaseProvider {
    fn default() -> Self {
        Self::new()
    }
}

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

    /// Create a new Coinbase 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 = COINBASE_WS_ADVANCED_URL.into();
        default_config.websocket_config.ping_interval_milliseconds = 30000; // 30 seconds
        default_config.websocket_config.use_compression = false;

        // Set default REST API configuration
        default_config.rest_config.base_url = COINBASE_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::Coinbase, 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(())
    }
}

#[async_trait]
impl Provider for CoinbaseProvider {
    type TradeMessage = TradeMessage;
    type DepthMessage = Level2Update;
    type InstrumentMessage = Value;

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

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

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

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

    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(())
    }

    #[inline]
    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)
    }

    #[inline]
    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(())
    }

    #[inline]
    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)
    }

    #[inline]
    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_realtime_orderbook(&self, _symbol: &str) -> Result<SharedSimdOrderBook> {
        // Not yet implemented
        Err(anyhow!("Real-time orderbook not implemented yet"))
    }

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

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

        // Check if the request was successful
        if !response.status().is_success() {
            return Err(anyhow!(
                "Failed to fetch Coinbase products 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 products_info: simd_json::OwnedValue = simd_json::from_slice(&mut bytes_vec)
            .context("Failed to parse Coinbase products info response")?;

        // Extract the products from the response
        let products = products_info
            .as_array()
            .ok_or_else(|| anyhow!("Invalid response format: expected array of products"))?;

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

        for product_data in products {
            // Skip products that are not trading
            let status = product_data["status"].as_str().unwrap_or("");
            if status != "online" {
                continue;
            }

            let product_id = product_data["product_id"].as_str().ok_or_else(|| {
                anyhow!("Invalid product data: 'product_id' field not found or not a String")
            })?;

            // Create a basic instrument implementation
            let instrument_id = InstrumentId::new(product_id, Venue::Coinbase);
            let instrument =
                Box::new(CoinbaseInstrument { 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 Coinbase API
        // Not fully implemented yet
        Err(anyhow!("Historical trades not implemented yet"))
    }

    #[inline]
    async fn get_orderbook_snapshot(
        &self,
        symbol: &str,
        depth: Option<u32>,
    ) -> Result<OrderBookSnapshot> {
        let _limit = depth.unwrap_or(50);
        let url = format!(
            "{}/products/{}/book?level=2",
            self.config.rest_config.base_url, symbol
        );

        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
            .map_err(|e| anyhow!("Failed to get response bytes: {}", e))?;
        let mut bytes_vec = bytes.to_vec();
        let snapshot_data: simd_json::OwnedValue = simd_json::from_slice(&mut bytes_vec)
            .map_err(|e| anyhow!("Failed to parse JSON: {}", e))?;

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

        let sequence = snapshot_data["sequence"].as_u64().unwrap_or(0);
        let mut order_book_snapshot =
            OrderBookSnapshot::new_empty(instrument_id, timestamp, sequence);

        // Parse bids
        if let Some(bids) = snapshot_data["bids"].as_array() {
            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
        if let Some(asks) = snapshot_data["asks"].as_array() {
            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)
    }

    #[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!("{}/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
    }

    #[inline]
    fn get_rate_limits(&self) -> Vec<RateLimit> {
        COINBASE_RATE_LIMITS.to_vec()
    }
}

impl CoinbaseProvider {
    /// Create a WebSocket connection for trade data
    #[inline]
    async fn create_trade_connection(
        &self,
        symbols: SmallVec<[String; 8]>,
        tx: mpsc::Sender<TradeMessage>,
        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 = CoinbaseMessageHandler::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<Level2Update>,
        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 = CoinbaseMessageHandler::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)
    }
}

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

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

impl CoinbaseMessageHandler {
    const fn new_trade_handler(
        clock: Clock,
        stats: Arc<RwLock<ConnectionStats>>,
        tx: mpsc::Sender<TradeMessage>,
        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<Level2Update>,
        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 CoinbaseMessageHandler {
    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() {
            CoinbaseProvider::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(message_type) = json_value["type"].as_str() {
                match self.handler_type {
                    HandlerType::Trade
                        if message_type == "match" || message_type == "last_match" =>
                    {
                        if let Ok(trade_response) =
                            simd_json::from_slice::<TradeMessage>(&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 message_type == "l2update" => {
                        if let Ok(orderbook_response) =
                            simd_json::from_slice::<Level2Update>(&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)
                        match message_type {
                            "subscriptions" => {
                                log::info!("Coinbase subscription confirmed: {text}");
                            }
                            "heartbeat" => {
                                log::trace!("Coinbase heartbeat received");
                            }
                            "error" => {
                                log::error!("Coinbase WebSocket error: {text}");
                            }
                            _ => {
                                log::debug!("Unhandled Coinbase message type: {message_type}");
                            }
                        }
                    }
                }
            }
        }

        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 Coinbase 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 Coinbase WebSocket");
        Ok(())
    }

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