rusty_ems/exchanges/coinbase/

websocket_trading.rs

//! Coinbase WebSocket Trading Implementation
//!
//! Provides WebSocket trading functionality for Coinbase exchange with support for:
//! - Market data feeds (ticker, level2, matches)
//! - Order management via user channel
//! - Authentication for private channels
//! - Heartbeat monitoring
//! - Sequence number tracking
//! - Automatic reconnection

use anyhow::{Result, bail};
use async_trait::async_trait;
use flume::Sender;
use futures::{SinkExt, StreamExt};
use log::{debug, error, info, warn};
use parking_lot::RwLock;
use quanta::Clock;
use rust_decimal::Decimal;
use rusty_common::SmartString;
use rusty_common::auth::exchanges::coinbase::CoinbaseAuth;
use rusty_common::collections::FxHashMap;
use rusty_common::time;
use rusty_common::utils::id_generation;
use rusty_common::websocket::connector::{WebSocketSink, WebSocketStream};
use rusty_common::websocket::{Message, WebSocketConfig};
use rusty_model::{
    data::orderbook::{OrderBook, PriceLevel},
    enums::{OrderSide, OrderStatus, OrderType},
    instruments::InstrumentId,
    trading_order::Order,
    venues::Venue,
};
use simd_json::prelude::{ValueAsArray, ValueAsScalar, ValueObjectAccess};
use simd_json::{json, value::owned::Value as JsonValue};
use smallvec::SmallVec;
use std::collections::VecDeque;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::time::Duration;
use tokio::sync::RwLock as AsyncRwLock;
use tokio::task::JoinHandle;
use tokio::time::interval;
use yawc;

// Removed unused imports
use crate::execution_engine::{Exchange, ExecutionReport};

/// Type alias for level2 update result containing product ID and price level changes
type Level2UpdateResult = Result<(SmartString, Vec<(SmartString, Decimal, Decimal)>)>;

/// WebSocket API URLs
const COINBASE_WS_URL: &str = "wss://ws-feed.exchange.coinbase.com";
const COINBASE_WS_DIRECT_URL: &str = "wss://ws-direct.exchange.coinbase.com";
const COINBASE_WS_SANDBOX_URL: &str = "wss://ws-feed-public.sandbox.exchange.coinbase.com";
const COINBASE_WS_DIRECT_SANDBOX_URL: &str = "wss://ws-direct.sandbox.exchange.coinbase.com";

/// Connection management constants
const HEARTBEAT_INTERVAL_SECONDS: u64 = 1;
const RECONNECT_DELAY_SECONDS: u64 = 5;
const MAX_RECONNECT_ATTEMPTS: u32 = 10;
const SUBSCRIBE_TIMEOUT_SECONDS: u64 = 5;
const MAX_MESSAGE_SIZE: usize = 10 * 1024 * 1024; // 10MB

/// Order tracking structure
#[derive(Debug, Clone)]
struct OrderInfo {
    order_id: SmartString,
    client_order_id: SmartString,
    symbol: SmartString,
    side: OrderSide,
    order_type: OrderType,
    price: Decimal,
    quantity: Decimal,
    remaining_quantity: Decimal,
    executed_quantity: Decimal,
    status: OrderStatus,
    timestamp: u64,
}

/// Sequence tracking for reliable message ordering
#[derive(Debug)]
struct SequenceTracker {
    /// Product ID -> last sequence number
    sequences: FxHashMap<SmartString, u64>,
    /// Missing sequences to be fetched via REST
    missing_sequences: VecDeque<(SmartString, u64, u64)>,
}

impl SequenceTracker {
    fn new() -> Self {
        Self {
            sequences: FxHashMap::default(),
            missing_sequences: VecDeque::new(),
        }
    }

    /// Update sequence number and detect gaps
    fn update_sequence(&mut self, product_id: &str, new_sequence: u64) -> bool {
        let product_id = SmartString::from(product_id);

        if let Some(&last_sequence) = self.sequences.get(&product_id) {
            if new_sequence <= last_sequence {
                // Out of order or duplicate
                return false;
            }

            if new_sequence > last_sequence + 1 {
                // Gap detected
                self.missing_sequences.push_back((
                    product_id.clone(),
                    last_sequence + 1,
                    new_sequence - 1,
                ));
                warn!("Sequence gap detected for {product_id}: {last_sequence} -> {new_sequence}");
            }
        }

        self.sequences.insert(product_id, new_sequence);
        true
    }
}

/// Coinbase WebSocket trading implementation
pub struct CoinbaseWebsocketTrading {
    /// Authentication handler
    auth: Arc<CoinbaseAuth>,

    /// WebSocket connection
    ws_sink: Arc<AsyncRwLock<Option<WebSocketSink>>>,
    ws_stream: Arc<AsyncRwLock<Option<WebSocketStream>>>,

    /// Configuration
    sandbox: bool,
    use_direct: bool,

    /// Connection state
    is_connected: Arc<AtomicBool>,
    is_authenticated: Arc<AtomicBool>,
    last_heartbeat: Arc<AtomicU64>,
    reconnect_attempts: Arc<AtomicU64>,

    /// Subscriptions
    subscribed_products: Arc<RwLock<Vec<SmartString>>>,
    subscribed_channels: Arc<RwLock<Vec<SmartString>>>,

    /// Order tracking
    orders: Arc<RwLock<FxHashMap<SmartString, OrderInfo>>>,
    client_order_map: Arc<RwLock<FxHashMap<SmartString, SmartString>>>,

    /// Sequence tracking
    sequence_tracker: Arc<RwLock<SequenceTracker>>,

    /// Order books for subscribed products
    order_books: Arc<RwLock<FxHashMap<SmartString, OrderBook>>>,

    /// Monitoring
    clock: Clock,

    /// Task handles
    heartbeat_handle: Arc<RwLock<Option<JoinHandle<()>>>>,
    message_handle: Arc<RwLock<Option<JoinHandle<()>>>>,

    /// Execution report channel
    report_sender: Arc<RwLock<Option<Sender<ExecutionReport>>>>,
}

impl CoinbaseWebsocketTrading {
    /// Create a new Coinbase WebSocket trading instance
    pub fn new(auth: Arc<CoinbaseAuth>, sandbox: bool, use_direct: bool) -> Self {
        let clock = Clock::new();

        Self {
            auth,
            ws_sink: Arc::new(AsyncRwLock::new(None)),
            ws_stream: Arc::new(AsyncRwLock::new(None)),
            sandbox,
            use_direct,
            is_connected: Arc::new(AtomicBool::new(false)),
            is_authenticated: Arc::new(AtomicBool::new(false)),
            last_heartbeat: Arc::new(AtomicU64::new(0)),
            reconnect_attempts: Arc::new(AtomicU64::new(0)),
            subscribed_products: Arc::new(RwLock::new(Vec::new())),
            subscribed_channels: Arc::new(RwLock::new(Vec::new())),
            orders: Arc::new(RwLock::new(FxHashMap::default())),
            client_order_map: Arc::new(RwLock::new(FxHashMap::default())),
            sequence_tracker: Arc::new(RwLock::new(SequenceTracker::new())),
            order_books: Arc::new(RwLock::new(FxHashMap::default())),
            clock,
            heartbeat_handle: Arc::new(RwLock::new(None)),
            message_handle: Arc::new(RwLock::new(None)),
            report_sender: Arc::new(RwLock::new(None)),
        }
    }

    /// Get WebSocket URL based on configuration
    const fn get_ws_url(&self) -> &'static str {
        match (self.sandbox, self.use_direct) {
            (true, true) => COINBASE_WS_DIRECT_SANDBOX_URL,
            (true, false) => COINBASE_WS_SANDBOX_URL,
            (false, true) => COINBASE_WS_DIRECT_URL,
            (false, false) => COINBASE_WS_URL,
        }
    }

    /// Subscribe to channels
    async fn subscribe(&self, channels: Vec<&str>, product_ids: Vec<&str>) -> Result<()> {
        // Use the proper WebSocket subscription generation
        let ws_subscription = self.auth.generate_ws_subscription(&channels)?;

        let mut subscribe_msg = json!({
            "type": ws_subscription.message_type,
            "product_ids": product_ids,
            "channels": ws_subscription.channels
        });

        // Add authentication fields based on the key type
        if let Some(jwt) = ws_subscription.jwt {
            subscribe_msg["jwt"] = json!(jwt);
        }
        if let Some(api_key) = ws_subscription.api_key {
            subscribe_msg["key"] = json!(api_key);
        }
        if let Some(timestamp) = ws_subscription.timestamp {
            subscribe_msg["timestamp"] = json!(timestamp);
        }
        if let Some(signature) = ws_subscription.signature {
            subscribe_msg["signature"] = json!(signature);
        }

        let subscribe_msg_str = subscribe_msg.to_string();

        if let Some(sink) = &mut *self.ws_sink.write().await {
            sink.send(Message::text(subscribe_msg_str).to_frame_view())
                .await?;

            // Update subscriptions
            self.subscribed_products
                .write()
                .extend(product_ids.iter().map(|&p| SmartString::from(p)));
            self.subscribed_channels
                .write()
                .extend(channels.iter().map(|&c| SmartString::from(c)));

            debug!("Subscribed to channels: {channels:?} for products: {product_ids:?}");
        } else {
            bail!("WebSocket connection not established");
        }

        Ok(())
    }

    /// Connect and start message processing
    async fn connect_internal(&self, report_sender: Sender<ExecutionReport>) -> Result<()> {
        // Store the report sender
        *self.report_sender.write() = Some(report_sender.clone());

        let config = WebSocketConfig::new(
            rusty_common::types::Exchange::Coinbase,
            self.get_ws_url().to_string(),
        );

        // Connect to WebSocket
        let ws_url = config.url.parse()?;
        let ws = yawc::WebSocket::connect(ws_url).await?;
        let (ws_sink, ws_stream) = ws.split();
        *self.ws_sink.write().await = Some(ws_sink);
        *self.ws_stream.write().await = Some(ws_stream);

        self.is_connected.store(true, Ordering::Relaxed);

        // Subscribe to required channels
        let product_ids: Vec<String> = {
            self.subscribed_products
                .read()
                .iter()
                .map(std::string::ToString::to_string)
                .collect()
        };

        let product_id_refs: Vec<&str> = product_ids
            .iter()
            .map(std::string::String::as_str)
            .collect();

        self.subscribe(
            vec!["heartbeat", "ticker", "matches", "user", "level2"],
            product_id_refs,
        )
        .await?;

        // Start heartbeat monitoring
        self.start_heartbeat_monitor();

        // Start message processing
        self.start_message_processing();

        info!("Coinbase WebSocket connected to {}", self.get_ws_url());
        Ok(())
    }

    /// Start heartbeat monitoring task
    fn start_heartbeat_monitor(&self) {
        let last_heartbeat = self.last_heartbeat.clone();
        let clock = self.clock.clone();
        let is_connected = self.is_connected.clone();

        let handle = tokio::spawn(async move {
            let mut interval = interval(Duration::from_secs(HEARTBEAT_INTERVAL_SECONDS * 2));

            loop {
                interval.tick().await;

                if !is_connected.load(Ordering::Relaxed) {
                    break;
                }

                let last_hb = last_heartbeat.load(Ordering::Relaxed);
                let now = clock.raw() / 1_000_000_000; // Convert to seconds

                if now - last_hb > HEARTBEAT_INTERVAL_SECONDS * 3 {
                    warn!("Heartbeat timeout detected, connection may be dead");
                    is_connected.store(false, Ordering::Relaxed);
                    break;
                }
            }

            debug!("Heartbeat monitor task terminated");
        });

        *self.heartbeat_handle.write() = Some(handle);
    }

    /// Start processing incoming messages
    fn start_message_processing(&self) {
        let ws_stream = self.ws_stream.clone();
        let last_heartbeat = self.last_heartbeat.clone();
        let clock = self.clock.clone();
        let is_connected = self.is_connected.clone();
        let is_authenticated = self.is_authenticated.clone();
        let orders = self.orders.clone();
        let client_order_map = self.client_order_map.clone();
        let sequence_tracker = self.sequence_tracker.clone();
        let report_sender = self.report_sender.clone();
        let order_books = self.order_books.clone();

        let handle = tokio::spawn(async move {
            loop {
                if !is_connected.load(Ordering::Relaxed) {
                    break;
                }

                if let Some(stream) = &mut *ws_stream.write().await {
                    if let Some(frame) = stream.next().await {
                        let msg = Message::from_frame_view(frame);
                        match msg {
                            Message::Text(text) => {
                                if let Err(e) = Self::process_message(
                                    &text,
                                    &last_heartbeat,
                                    &clock,
                                    &is_authenticated,
                                    &orders,
                                    &client_order_map,
                                    &sequence_tracker,
                                    &report_sender,
                                    &order_books,
                                )
                                .await
                                {
                                    error!("Failed to process message: {e}");
                                }
                            }
                            Message::Binary(_) => {
                                warn!("Received unexpected binary message");
                            }
                            _ => {}
                        }
                    } else {
                        warn!("WebSocket stream closed");
                        is_connected.store(false, Ordering::Relaxed);
                        break;
                    }
                }
            }

            debug!("Message processing task terminated");
        });

        *self.message_handle.write() = Some(handle);
    }

    /// Process incoming WebSocket message
    #[allow(clippy::too_many_arguments)]
    async fn process_message(
        text: &str,
        last_heartbeat: &Arc<AtomicU64>,
        clock: &Clock,
        is_authenticated: &Arc<AtomicBool>,
        orders: &Arc<RwLock<FxHashMap<SmartString, OrderInfo>>>,
        client_order_map: &Arc<RwLock<FxHashMap<SmartString, SmartString>>>,
        sequence_tracker: &Arc<RwLock<SequenceTracker>>,
        report_sender: &Arc<RwLock<Option<Sender<ExecutionReport>>>>,
        order_books: &Arc<RwLock<FxHashMap<SmartString, OrderBook>>>,
    ) -> Result<()> {
        let mut text_bytes = text.as_bytes().to_vec();
        let msg = simd_json::from_slice::<JsonValue>(&mut text_bytes)?;

        let msg_type = msg
            .get("type")
            .and_then(|v| v.as_str())
            .map(std::string::ToString::to_string);

        if let Some(msg_type) = msg_type {
            match msg_type.as_str() {
                "subscriptions" => {
                    info!("Subscription confirmed: {text}");
                    is_authenticated.store(true, Ordering::Relaxed);
                }
                "heartbeat" => {
                    last_heartbeat.store(clock.raw() / 1_000_000_000, Ordering::Relaxed);

                    if let Some(sequence) = msg
                        .get("sequence")
                        .and_then(simd_json::prelude::ValueAsScalar::as_u64)
                        && let Some(product_id) = msg.get("product_id").and_then(|v| v.as_str())
                    {
                        sequence_tracker
                            .write()
                            .update_sequence(product_id, sequence);
                    }
                }
                "ticker" => {
                    // Process ticker updates
                    Self::process_ticker(msg).await?;
                }
                "snapshot" => {
                    // Process level2 snapshot
                    Self::process_snapshot_static(msg, order_books).await?;
                }
                "l2update" => {
                    // Process level2 update
                    if let Some(sequence) = msg
                        .get("sequence")
                        .and_then(simd_json::prelude::ValueAsScalar::as_u64)
                        && let Some(product_id) = msg.get("product_id").and_then(|v| v.as_str())
                        && sequence_tracker
                            .write()
                            .update_sequence(product_id, sequence)
                    {
                        Self::process_l2_update_static(msg, order_books).await?;
                    }
                }
                "match" | "last_match" => {
                    // Process trade matches
                    if let Some(sequence) = msg
                        .get("sequence")
                        .and_then(simd_json::prelude::ValueAsScalar::as_u64)
                        && let Some(product_id) = msg.get("product_id").and_then(|v| v.as_str())
                    {
                        sequence_tracker
                            .write()
                            .update_sequence(product_id, sequence);
                    }
                    Self::process_match(msg).await?;
                }
                "received" | "open" | "done" | "change" => {
                    // Process order updates from user channel
                    Self::process_order_update(
                        msg,
                        &msg_type,
                        orders,
                        client_order_map,
                        report_sender,
                    )
                    .await?;
                }
                "error" => {
                    let message = msg
                        .get("message")
                        .and_then(|v| v.as_str())
                        .unwrap_or("Unknown error");
                    error!("WebSocket error: {message}");
                }
                _ => {
                    debug!("Unhandled message type: {msg_type}");
                }
            }
        }

        Ok(())
    }

    /// Process ticker message
    async fn process_ticker(msg: JsonValue) -> Result<()> {
        // TODO: Implement ticker processing if needed
        debug!("Ticker update: {msg:?}");
        Ok(())
    }

    /// Process level2 snapshot
    async fn process_snapshot(&self, msg: JsonValue) -> Result<()> {
        let order_book = Self::parse_level2_snapshot(&msg)?;
        let product_id = order_book.symbol.clone();

        // Store the order book
        self.order_books
            .write()
            .insert(product_id.clone(), order_book);

        debug!("Processed L2 snapshot for {product_id}");
        Ok(())
    }

    /// Static version for `process_snapshot`
    async fn process_snapshot_static(
        msg: JsonValue,
        order_books: &Arc<RwLock<FxHashMap<SmartString, OrderBook>>>,
    ) -> Result<()> {
        let order_book = Self::parse_level2_snapshot(&msg)?;
        let product_id = order_book.symbol.clone();

        // Store the order book
        order_books.write().insert(product_id.clone(), order_book);

        debug!("Processed L2 snapshot for {product_id}");
        Ok(())
    }

    /// Process level2 update
    async fn process_l2_update(&self, msg: JsonValue) -> Result<()> {
        let (product_id, changes) = Self::parse_level2_update(&msg)?;

        // Apply updates to the order book
        if let Some(order_book) = self.order_books.write().get_mut(&product_id) {
            Self::apply_level2_update(order_book, &changes);
            debug!("Applied {} L2 updates to {}", changes.len(), product_id);
        } else {
            warn!("Received L2 update for unknown product: {product_id}");
        }

        Ok(())
    }

    /// Static version for `process_l2_update`
    async fn process_l2_update_static(
        msg: JsonValue,
        order_books: &Arc<RwLock<FxHashMap<SmartString, OrderBook>>>,
    ) -> Result<()> {
        let (product_id, changes) = Self::parse_level2_update(&msg)?;

        // Apply updates to the order book
        if let Some(order_book) = order_books.write().get_mut(&product_id) {
            Self::apply_level2_update(order_book, &changes);
            debug!("Applied {} L2 updates to {}", changes.len(), product_id);
        } else {
            warn!("Received L2 update for unknown product: {product_id}");
        }

        Ok(())
    }

    /// Process match message
    async fn process_match(msg: JsonValue) -> Result<()> {
        // TODO: Implement match processing if needed
        debug!("Match: {msg:?}");
        Ok(())
    }

    /// Process order update from user channel
    async fn process_order_update(
        msg: JsonValue,
        msg_type: &str,
        orders: &Arc<RwLock<FxHashMap<SmartString, OrderInfo>>>,
        client_order_map: &Arc<RwLock<FxHashMap<SmartString, SmartString>>>,
        report_sender: &Arc<RwLock<Option<Sender<ExecutionReport>>>>,
    ) -> Result<()> {
        let order_id = msg.get("order_id").and_then(|v| v.as_str()).unwrap_or("");
        let client_oid = msg.get("client_oid").and_then(|v| v.as_str()).unwrap_or("");
        let product_id = msg.get("product_id").and_then(|v| v.as_str()).unwrap_or("");

        // Map client order ID to order ID
        if !client_oid.is_empty() && !order_id.is_empty() {
            client_order_map
                .write()
                .insert(SmartString::from(client_oid), SmartString::from(order_id));
        }

        let (exec_type, order_status) = match msg_type {
            "received" => ("New", OrderStatus::New),
            "open" => ("New", OrderStatus::New),
            "done" => {
                let reason = msg.get("reason").and_then(|v| v.as_str()).unwrap_or("");
                match reason {
                    "filled" => ("Trade", OrderStatus::Filled),
                    "canceled" => ("Canceled", OrderStatus::Cancelled),
                    _ => ("Rejected", OrderStatus::Rejected),
                }
            }
            "match" => ("Trade", OrderStatus::PartiallyFilled),
            "change" => ("Replaced", OrderStatus::New),
            _ => {
                log::warn!("Unknown Coinbase WebSocket message type: {msg_type}");
                ("Unknown", OrderStatus::Unknown)
            }
        };

        // Parse quantities and prices for ExecutionReport
        let executed_quantity = if msg_type == "match" {
            msg.get("size")
                .and_then(|v| v.as_str())
                .and_then(|s| Decimal::from_str_exact(s).ok())
                .unwrap_or(Decimal::ZERO)
        } else {
            Decimal::ZERO
        };

        let remaining_quantity = msg
            .get("remaining_size")
            .and_then(|v| v.as_str())
            .and_then(|s| Decimal::from_str_exact(s).ok())
            .unwrap_or(Decimal::ZERO);

        let total_quantity = msg
            .get("size")
            .and_then(|v| v.as_str())
            .and_then(|s| Decimal::from_str_exact(s).ok())
            .unwrap_or(Decimal::ZERO);

        let execution_price = if msg_type == "match" {
            msg.get("price")
                .and_then(|v| v.as_str())
                .and_then(|s| Decimal::from_str_exact(s).ok())
        } else {
            None
        };

        let reject_reason = if msg_type == "done" {
            let reason = msg.get("reason").and_then(|v| v.as_str()).unwrap_or("");
            if reason == "canceled" || reason == "rejected" {
                Some(reason.into())
            } else {
                None
            }
        } else {
            None
        };

        // Create execution report with valid fields only
        let report = ExecutionReport {
            id: id_generation::generate_multi_part_id(
                &["coinbase", client_oid, &uuid::Uuid::new_v4().to_string()],
                '_',
            ),
            order_id: order_id.into(),
            exchange_timestamp: time::get_epoch_timestamp_ns(),
            system_timestamp: time::get_epoch_timestamp_ns(),
            instrument_id: InstrumentId::new(product_id, Venue::Coinbase),
            status: order_status,
            filled_quantity: executed_quantity,
            remaining_quantity,
            execution_price,
            reject_reason,
            exchange_execution_id: Some(order_id.into()),
            is_final: matches!(
                order_status,
                OrderStatus::Filled | OrderStatus::Cancelled | OrderStatus::Rejected
            ),
        };

        // Send execution report
        if let Some(sender) = &*report_sender.read()
            && let Err(e) = sender.send(report)
        {
            error!("Failed to send execution report: {e}");
        }

        Ok(())
    }

    /// Place order via REST API (WebSocket doesn't support order placement)
    async fn place_order_rest(&self, order: &Order) -> Result<SmartString> {
        // Coinbase WebSocket doesn't support order placement
        // Orders must be placed via REST API
        // This is a placeholder - implement REST API call
        bail!("Order placement via WebSocket not supported - use REST API")
    }

    /// Cancel order via REST API
    async fn cancel_order_rest(&self, order_id: &str) -> Result<()> {
        // Coinbase WebSocket doesn't support order cancellation
        // Orders must be cancelled via REST API
        bail!("Order cancellation via WebSocket not supported - use REST API")
    }
}

// Public parsing functions for order book processing
impl CoinbaseWebsocketTrading {
    /// Parse level2 snapshot message into `OrderBook`
    pub fn parse_level2_snapshot(msg: &JsonValue) -> Result<OrderBook> {
        let product_id = msg
            .get("product_id")
            .and_then(|v| v.as_str())
            .ok_or_else(|| anyhow::anyhow!("Missing product_id in snapshot"))?;

        let bids = msg
            .get("bids")
            .and_then(|v| v.as_array())
            .ok_or_else(|| anyhow::anyhow!("Missing bids in snapshot"))?;

        let asks = msg
            .get("asks")
            .and_then(|v| v.as_array())
            .ok_or_else(|| anyhow::anyhow!("Missing asks in snapshot"))?;

        // Parse bids
        let mut bid_levels = SmallVec::with_capacity(bids.len());
        for bid in bids {
            if let Some(arr) = bid.as_array()
                && arr.len() >= 2
            {
                let price = arr[0]
                    .as_str()
                    .and_then(|s| s.parse::<Decimal>().ok())
                    .ok_or_else(|| anyhow::anyhow!("Invalid bid price"))?;
                let quantity = arr[1]
                    .as_str()
                    .and_then(|s| s.parse::<Decimal>().ok())
                    .ok_or_else(|| anyhow::anyhow!("Invalid bid quantity"))?;
                bid_levels.push(PriceLevel::new(price, quantity));
            }
        }

        // Parse asks
        let mut ask_levels = SmallVec::with_capacity(asks.len());
        for ask in asks {
            if let Some(arr) = ask.as_array()
                && arr.len() >= 2
            {
                let price = arr[0]
                    .as_str()
                    .and_then(|s| s.parse::<Decimal>().ok())
                    .ok_or_else(|| anyhow::anyhow!("Invalid ask price"))?;
                let quantity = arr[1]
                    .as_str()
                    .and_then(|s| s.parse::<Decimal>().ok())
                    .ok_or_else(|| anyhow::anyhow!("Invalid ask quantity"))?;
                ask_levels.push(PriceLevel::new(price, quantity));
            }
        }

        // Sort bids descending and asks ascending
        bid_levels.sort_by(|a: &PriceLevel, b: &PriceLevel| b.price.cmp(&a.price));
        ask_levels.sort_by(|a: &PriceLevel, b: &PriceLevel| a.price.cmp(&b.price));

        Ok(OrderBook::new(
            product_id,
            time::get_epoch_timestamp_ns(),
            time::get_epoch_timestamp_ns(),
            bid_levels,
            ask_levels,
        ))
    }

    /// Parse level2 update message
    pub fn parse_level2_update(msg: &JsonValue) -> Level2UpdateResult {
        let product_id = msg
            .get("product_id")
            .and_then(|v| v.as_str())
            .ok_or_else(|| anyhow::anyhow!("Missing product_id in update"))?;

        let changes = msg
            .get("changes")
            .and_then(|v| v.as_array())
            .ok_or_else(|| anyhow::anyhow!("Missing changes in update"))?;

        let mut parsed_changes = Vec::with_capacity(changes.len());

        for change in changes {
            if let Some(arr) = change.as_array()
                && arr.len() >= 3
            {
                let side = arr[0]
                    .as_str()
                    .ok_or_else(|| anyhow::anyhow!("Invalid side in change"))?;
                let price = arr[1]
                    .as_str()
                    .and_then(|s| s.parse::<Decimal>().ok())
                    .ok_or_else(|| anyhow::anyhow!("Invalid price in change"))?;
                let quantity = arr[2]
                    .as_str()
                    .and_then(|s| s.parse::<Decimal>().ok())
                    .ok_or_else(|| anyhow::anyhow!("Invalid quantity in change"))?;

                parsed_changes.push((side.into(), price, quantity));
            }
        }

        Ok((product_id.into(), parsed_changes))
    }

    /// Apply level2 updates to an order book
    pub fn apply_level2_update(
        order_book: &mut OrderBook,
        changes: &[(SmartString, Decimal, Decimal)],
    ) {
        for (side, price, quantity) in changes {
            match side.as_str() {
                "buy" => {
                    // Update bid side
                    if quantity.is_zero() {
                        // Remove level
                        order_book.bids.retain(|level| level.price != *price);
                    } else {
                        // Add or update level
                        if let Some(level) = order_book.bids.iter_mut().find(|l| l.price == *price)
                        {
                            level.quantity = *quantity;
                        } else {
                            // Insert new level and re-sort
                            order_book.bids.push(PriceLevel::new(*price, *quantity));
                            order_book.bids.sort_by(|a, b| b.price.cmp(&a.price));
                        }
                    }
                }
                "sell" => {
                    // Update ask side
                    if quantity.is_zero() {
                        // Remove level
                        order_book.asks.retain(|level| level.price != *price);
                    } else {
                        // Add or update level
                        if let Some(level) = order_book.asks.iter_mut().find(|l| l.price == *price)
                        {
                            level.quantity = *quantity;
                        } else {
                            // Insert new level and re-sort
                            order_book.asks.push(PriceLevel::new(*price, *quantity));
                            order_book.asks.sort_by(|a, b| a.price.cmp(&b.price));
                        }
                    }
                }
                _ => {
                    warn!("Unknown side in L2 update: {side}");
                }
            }
        }
    }
}

#[async_trait]
impl Exchange for CoinbaseWebsocketTrading {
    fn venue(&self) -> Venue {
        Venue::Coinbase
    }

    async fn place_order(
        &self,
        order: Order,
        report_sender: Sender<ExecutionReport>,
    ) -> Result<()> {
        // Coinbase WebSocket doesn't support order placement
        // Must use REST API
        bail!("Order placement not supported via WebSocket - use REST API")
    }

    async fn cancel_order(
        &self,
        order_id: SmartString,
        report_sender: Sender<ExecutionReport>,
    ) -> Result<()> {
        // Coinbase WebSocket doesn't support order cancellation
        // Must use REST API
        bail!("Order cancellation not supported via WebSocket - use REST API")
    }

    async fn modify_order(
        &self,
        order_id: SmartString,
        new_price: Option<Decimal>,
        new_quantity: Option<Decimal>,
        report_sender: Sender<ExecutionReport>,
    ) -> Result<()> {
        // Coinbase WebSocket doesn't support order modification
        bail!("Order modification not supported via WebSocket - use REST API")
    }

    async fn cancel_all_orders(
        &self,
        instrument_id: Option<InstrumentId>,
        report_sender: Sender<ExecutionReport>,
    ) -> Result<()> {
        // Coinbase WebSocket doesn't support mass cancellation
        bail!("Mass order cancellation not supported via WebSocket - use REST API")
    }

    async fn get_order_status(&self, order_id: &str) -> Result<OrderStatus> {
        // Check local order tracking
        if let Some(order_info) = self.orders.read().get(order_id) {
            Ok(order_info.status)
        } else {
            bail!("Order not found in local tracking")
        }
    }

    async fn connect(&self, report_sender: Sender<ExecutionReport>) -> Result<()> {
        // Use the proper report_sender channel instead of creating a dummy one
        self.connect_internal(report_sender).await
    }

    async fn disconnect(&self) -> Result<()> {
        self.is_connected.store(false, Ordering::Relaxed);
        self.is_authenticated.store(false, Ordering::Relaxed);

        // Close WebSocket connection
        if let Some(sink) = &mut *self.ws_sink.write().await {
            let _ = sink.close().await;
        }
        *self.ws_sink.write().await = None;
        *self.ws_stream.write().await = None;

        // Cancel background tasks
        if let Some(handle) = self.heartbeat_handle.write().take() {
            handle.abort();
        }

        if let Some(handle) = self.message_handle.write().take() {
            handle.abort();
        }

        // Clear tracking data
        self.orders.write().clear();
        self.client_order_map.write().clear();
        self.sequence_tracker.write().sequences.clear();

        info!("Coinbase WebSocket disconnected");
        Ok(())
    }

    async fn is_connected(&self) -> bool {
        self.is_connected.load(Ordering::Relaxed) && self.is_authenticated.load(Ordering::Relaxed)
    }

    async fn get_instruments(&self) -> Result<SmallVec<[InstrumentId; 32]>> {
        // WebSocket doesn't provide instrument query functionality
        // You would need to use REST API or maintain a cached list
        bail!("Instrument query not supported via WebSocket")
    }

    async fn send_fix_message(&self, message: Vec<u8>) -> Result<()> {
        // Coinbase WebSocket doesn't support FIX protocol
        bail!("FIX protocol not supported on Coinbase WebSocket")
    }

    async fn receive_fix_message(&self) -> Result<Vec<u8>> {
        // Coinbase WebSocket doesn't support FIX protocol
        bail!("FIX protocol not supported on Coinbase WebSocket")
    }
}