nautilus_coinbase/websocket/

handler.rs

// -------------------------------------------------------------------------------------------------
//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
//  https://nautechsystems.io
//
//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
//  You may not use this file except in compliance with the License.
//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
// -------------------------------------------------------------------------------------------------

//! Feed handler for parsing Coinbase WebSocket messages into Nautilus types.

use std::{fmt::Debug, sync::Arc};

use ahash::AHashMap;
use nautilus_core::{
    AtomicMap, UnixNanos,
    string::secret::SecretString,
    time::{AtomicTime, get_atomic_clock_realtime},
};
use nautilus_model::{
    data::{Bar, BarType, InstrumentStatus, OrderBookDeltas, QuoteTick, TradeTick},
    identifiers::{AccountId, InstrumentId, Symbol},
    instruments::{Instrument, InstrumentAny},
    reports::OrderStatusReport,
};
use nautilus_network::{RECONNECTED, websocket::WebSocketClient};
use tokio_tungstenite::tungstenite::Message;
use ustr::Ustr;

use crate::{
    common::consts::COINBASE_VENUE,
    websocket::{
        client::COINBASE_WS_SUBSCRIPTION_KEYS,
        messages::{CoinbaseWsMessage, WsEventType, WsOrderUpdate},
        parse::{
            parse_ws_candle, parse_ws_l2_snapshot, parse_ws_l2_update, parse_ws_status_product,
            parse_ws_ticker, parse_ws_trade, parse_ws_user_event_to_order_status_report,
        },
    },
};

fn instrument_id_from_product(product_id: &Ustr) -> InstrumentId {
    InstrumentId::new(Symbol::new(*product_id), *COINBASE_VENUE)
}

fn resolve_instrument_id_from_aliases(
    aliases: &AHashMap<Ustr, Ustr>,
    product_id: &Ustr,
) -> InstrumentId {
    let resolved = aliases.get(product_id).copied().unwrap_or(*product_id);
    instrument_id_from_product(&resolved)
}

/// Commands sent from [`super::client::CoinbaseWebSocketClient`] to the feed handler.
pub enum HandlerCommand {
    /// Provides the network-level WebSocket client.
    SetClient(WebSocketClient),
    /// Subscribes with a serialized payload that is zeroized on drop.
    Subscribe {
        channel: crate::common::enums::CoinbaseWsChannel,
        product_ids: Vec<Ustr>,
        payload: SecretString,
    },
    /// Unsubscribes with a serialized payload that is zeroized on drop.
    Unsubscribe {
        channel: crate::common::enums::CoinbaseWsChannel,
        product_ids: Vec<Ustr>,
        payload: SecretString,
    },
    /// Disconnects the WebSocket.
    Disconnect,
    /// Caches instruments for precision lookups during parsing.
    InitializeInstruments(Vec<InstrumentAny>),
    /// Updates a single instrument in the cache.
    UpdateInstrument(Box<InstrumentAny>),
    /// Registers a bar type for candle parsing.
    AddBarType { key: String, bar_type: BarType },
    /// Removes a bar type registration.
    RemoveBarType { key: String },
    /// Sets the account ID used when emitting user-channel execution reports.
    SetAccountId(AccountId),
}

impl Debug for HandlerCommand {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::SetClient(_) => f.write_str("SetClient"),
            Self::Subscribe { channel, .. } => write!(f, "Subscribe({channel:?})"),
            Self::Unsubscribe { channel, .. } => write!(f, "Unsubscribe({channel:?})"),
            Self::Disconnect => f.write_str("Disconnect"),
            Self::InitializeInstruments(v) => write!(f, "InitializeInstruments({})", v.len()),
            Self::UpdateInstrument(i) => write!(f, "UpdateInstrument({})", i.id()),
            Self::AddBarType { key, .. } => write!(f, "AddBarType({key})"),
            Self::RemoveBarType { key } => write!(f, "RemoveBarType({key})"),
            Self::SetAccountId(id) => write!(f, "SetAccountId({id})"),
        }
    }
}

/// Carrier for a single user-channel order update.
///
/// Pairs the parsed [`OrderStatusReport`] with the resolved instrument and
/// the raw venue payload so downstream consumers (e.g. the execution client)
/// can diff cumulative quantity and fees against their own tracked state.
///
/// `is_snapshot` is true when the wrapping `WsUserEvent` was a `snapshot`
/// type. Snapshots restate the current cumulative state of every open order
/// and must NOT be interpreted as fresh fills, otherwise a cold start (or
/// any state-clearing reconnect) would synthesize phantom fills covering the
/// entire pre-existing cumulative quantity.
#[derive(Debug, Clone)]
pub struct UserOrderUpdate {
    pub report: Box<OrderStatusReport>,
    pub update: Box<WsOrderUpdate>,
    pub instrument: InstrumentAny,
    pub is_snapshot: bool,
    pub ts_event: UnixNanos,
    pub ts_init: UnixNanos,
}

/// Nautilus-typed messages produced by the feed handler.
#[derive(Debug, Clone)]
pub enum NautilusWsMessage {
    /// Trade tick from market_trades channel.
    Trade(TradeTick),
    /// Quote tick from ticker channel.
    Quote(QuoteTick),
    /// Order book deltas from l2_data channel.
    Deltas(OrderBookDeltas),
    /// Bar from candles channel.
    Bar(Bar),
    /// Order status update from the user channel.
    UserOrder(Box<UserOrderUpdate>),
    /// Futures balance summary snapshot from the
    /// `futures_balance_summary` channel.
    FuturesBalanceSummary(Box<crate::websocket::messages::WsFcmBalanceSummary>),
    /// Instrument status update from the `status` channel. Emitted for every
    /// product the venue reports; the data client filters by subscription.
    InstrumentStatus(Box<InstrumentStatus>),
    /// The connection was re-established after a drop.
    Reconnected,
    /// An error occurred during message processing.
    Error(String),
}

/// Processes raw WebSocket messages into Nautilus domain types.
#[derive(Debug)]
pub struct FeedHandler {
    clock: &'static AtomicTime,
    signal: Arc<std::sync::atomic::AtomicBool>,
    client: Option<WebSocketClient>,
    cmd_rx: tokio::sync::mpsc::UnboundedReceiver<HandlerCommand>,
    raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
    instruments: AHashMap<InstrumentId, InstrumentAny>,
    /// Shared with [`super::client::CoinbaseWebSocketClient`]; consulted in
    /// `resolve_instrument_id_from_aliases` to re-key inbound messages whose wire
    /// `product_id` is the canonical alias of a subscribed/submitted product.
    subscription_aliases: Arc<AtomicMap<Ustr, Ustr>>,
    bar_types: AHashMap<String, BarType>,
    account_id: Option<AccountId>,
    buffer: Vec<NautilusWsMessage>,
    last_heartbeat_counter: Option<u64>,
}

impl FeedHandler {
    /// Creates a new [`FeedHandler`] instance.
    pub fn new(
        signal: Arc<std::sync::atomic::AtomicBool>,
        cmd_rx: tokio::sync::mpsc::UnboundedReceiver<HandlerCommand>,
        raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
        subscription_aliases: Arc<AtomicMap<Ustr, Ustr>>,
    ) -> Self {
        Self {
            clock: get_atomic_clock_realtime(),
            signal,
            client: None,
            cmd_rx,
            raw_rx,
            instruments: AHashMap::new(),
            subscription_aliases,
            bar_types: AHashMap::new(),
            account_id: None,
            buffer: Vec::new(),
            last_heartbeat_counter: None,
        }
    }

    /// Sets the account ID used to stamp user-channel execution reports.
    pub fn set_account_id(&mut self, account_id: AccountId) {
        self.account_id = Some(account_id);
    }

    /// Polls for the next output message, processing commands and raw messages.
    ///
    /// Returns `None` when the handler should shut down.
    pub async fn next(&mut self) -> Option<NautilusWsMessage> {
        // Check signal before draining buffer so disconnect takes
        // priority over pending buffered messages
        if self.signal.load(std::sync::atomic::Ordering::Acquire) {
            self.buffer.clear();
            return None;
        }

        if let Some(msg) = self.buffer.pop() {
            return Some(msg);
        }

        loop {
            if self.signal.load(std::sync::atomic::Ordering::Acquire) {
                return None;
            }

            tokio::select! {
                Some(cmd) = self.cmd_rx.recv() => {
                    match cmd {
                        HandlerCommand::SetClient(client) => {
                            self.client = Some(client);
                        }
                        HandlerCommand::Subscribe { payload, .. }
                        | HandlerCommand::Unsubscribe { payload, .. } => {
                            self.send_subscription(&payload).await;
                        }
                        HandlerCommand::Disconnect => {
                            if let Some(client) = self.client.take() {
                                // Transition to CLOSED immediately without waiting
                                // for ACTIVE (avoids blocking during reconnect)
                                client.notify_closed();
                            }
                            return None;
                        }
                        HandlerCommand::InitializeInstruments(instruments) => {
                            for inst in instruments {
                                self.instruments.insert(inst.id(), inst);
                            }
                        }
                        HandlerCommand::UpdateInstrument(inst) => {
                            self.instruments.insert(inst.id(), *inst);
                        }
                        HandlerCommand::AddBarType { key, bar_type } => {
                            self.bar_types.insert(key, bar_type);
                        }
                        HandlerCommand::RemoveBarType { key } => {
                            self.bar_types.remove(&key);
                        }
                        HandlerCommand::SetAccountId(account_id) => {
                            self.account_id = Some(account_id);
                        }
                    }
                }
                Some(raw) = self.raw_rx.recv() => {
                    match raw {
                        Message::Text(text) => {
                            if let Some(msg) = self.handle_text(&text) {
                                return Some(msg);
                            }
                        }
                        Message::Ping(data) => {
                            if let Some(client) = &self.client
                                && let Err(e) = client.send_pong(data.to_vec()).await
                            {
                                log::warn!("Failed to send pong: {e}");
                            }
                        }
                        Message::Close(_) => return None,
                        _ => {}
                    }
                }
                else => return None,
            }
        }
    }

    async fn send_subscription(&self, payload: &SecretString) {
        let Some(client) = &self.client else {
            log::warn!("Cannot send subscription, no WebSocket client set");
            return;
        };

        if let Err(e) = client
            .send_text(
                payload.expose_secret().to_owned(),
                Some(COINBASE_WS_SUBSCRIPTION_KEYS.as_slice()),
            )
            .await
        {
            log::error!("Failed to send subscription: {e}");
        }
    }

    fn handle_text(&mut self, text: &str) -> Option<NautilusWsMessage> {
        if text == RECONNECTED {
            self.last_heartbeat_counter = None;
            return Some(NautilusWsMessage::Reconnected);
        }

        let ts_init = self.clock.get_time_ns();

        let msg: CoinbaseWsMessage = match serde_json::from_str(text) {
            Ok(m) => m,
            Err(e) => {
                log::warn!("Failed to parse WS message: {e}");
                return None;
            }
        };

        match msg {
            CoinbaseWsMessage::L2Data {
                timestamp, events, ..
            } => self.handle_l2_events(&events, &timestamp, ts_init),
            CoinbaseWsMessage::MarketTrades { events, .. } => {
                self.handle_market_trades(&events, ts_init)
            }
            CoinbaseWsMessage::Ticker {
                timestamp, events, ..
            }
            | CoinbaseWsMessage::TickerBatch {
                timestamp, events, ..
            } => self.handle_ticker(&events, &timestamp, ts_init),
            CoinbaseWsMessage::Candles { events, .. } => self.handle_candles(&events, ts_init),
            CoinbaseWsMessage::Heartbeats { events, .. } => {
                for event in events {
                    if let Some((expected, actual)) =
                        self.note_heartbeat_counter(event.heartbeat_counter)
                    {
                        log::warn!("Heartbeat counter gap: expected {expected}, was {actual}");
                    }
                }
                None
            }
            CoinbaseWsMessage::Subscriptions { events, .. } => {
                // Coinbase emits this after every subscribe and unsubscribe
                // with the full current subscription set, so it's noisy at
                // INFO and not strictly a "confirmation" of the latest action.
                log::debug!("Subscription state: {events:?}");
                None
            }
            CoinbaseWsMessage::User {
                timestamp, events, ..
            } => self.handle_user_events(&events, &timestamp, ts_init),
            CoinbaseWsMessage::FuturesBalanceSummary { events, .. } => {
                self.handle_futures_balance_summary(events)
            }
            CoinbaseWsMessage::Status {
                timestamp, events, ..
            } => self.handle_status_events(&events, &timestamp, ts_init),
        }
    }

    fn note_heartbeat_counter(&mut self, counter: u64) -> Option<(u64, u64)> {
        let gap = self.last_heartbeat_counter.and_then(|last| {
            let expected = last.saturating_add(1);
            (counter != expected).then_some((expected, counter))
        });
        self.last_heartbeat_counter = Some(counter);
        gap
    }

    fn handle_l2_events(
        &mut self,
        events: &[crate::websocket::messages::WsL2DataEvent],
        timestamp: &str,
        ts_init: UnixNanos,
    ) -> Option<NautilusWsMessage> {
        let ts_event = match crate::http::parse::parse_rfc3339_timestamp(timestamp) {
            Ok(ts) => ts,
            Err(e) => {
                log::warn!("Failed to parse L2 message timestamp {timestamp}: {e}");
                ts_init
            }
        };

        let mut first: Option<NautilusWsMessage> = None;
        let aliases = self.subscription_aliases.load();

        for event in events {
            let instrument_id = resolve_instrument_id_from_aliases(&aliases, &event.product_id);

            let instrument = match self.instruments.get(&instrument_id) {
                Some(inst) => inst,
                None => {
                    log::warn!("No instrument cached for {instrument_id}");
                    continue;
                }
            };

            let result = match event.event_type {
                WsEventType::Snapshot => parse_ws_l2_snapshot(event, instrument, ts_event, ts_init),
                WsEventType::Update => parse_ws_l2_update(event, instrument, ts_event, ts_init),
            };

            match result {
                Ok(deltas) => {
                    let msg = NautilusWsMessage::Deltas(deltas);

                    if first.is_none() {
                        first = Some(msg);
                    } else {
                        self.buffer.push(msg);
                    }
                }
                Err(e) => log::warn!("Failed to parse L2 event: {e}"),
            }
        }

        if first.is_some() {
            self.buffer.reverse();
        }
        first
    }

    fn handle_market_trades(
        &mut self,
        events: &[crate::websocket::messages::WsMarketTradesEvent],
        ts_init: UnixNanos,
    ) -> Option<NautilusWsMessage> {
        let aliases = self.subscription_aliases.load();

        for event in events {
            for trade in &event.trades {
                let instrument_id = resolve_instrument_id_from_aliases(&aliases, &trade.product_id);

                let instrument = match self.instruments.get(&instrument_id) {
                    Some(inst) => inst,
                    None => {
                        log::warn!("No instrument cached for {instrument_id}");
                        continue;
                    }
                };

                match parse_ws_trade(trade, instrument, ts_init) {
                    Ok(tick) => {
                        self.buffer_remaining_trades(events, event, trade, ts_init);
                        // Reverse so pop() drains in exchange order
                        self.buffer.reverse();
                        return Some(NautilusWsMessage::Trade(tick));
                    }
                    Err(e) => log::warn!("Failed to parse trade: {e}"),
                }
            }
        }
        None
    }

    fn buffer_remaining_trades(
        &mut self,
        events: &[crate::websocket::messages::WsMarketTradesEvent],
        current_event: &crate::websocket::messages::WsMarketTradesEvent,
        current_trade: &crate::websocket::messages::WsTrade,
        ts_init: UnixNanos,
    ) {
        let mut found_current = false;
        let aliases = self.subscription_aliases.load();

        for event in events {
            let is_current_event = std::ptr::eq(event, current_event);

            for trade in &event.trades {
                if !found_current {
                    if is_current_event && std::ptr::eq(trade, current_trade) {
                        found_current = true;
                    }
                    continue;
                }

                let instrument_id = resolve_instrument_id_from_aliases(&aliases, &trade.product_id);

                if let Some(instrument) = self.instruments.get(&instrument_id)
                    && let Ok(tick) = parse_ws_trade(trade, instrument, ts_init)
                {
                    self.buffer.push(NautilusWsMessage::Trade(tick));
                }
            }
        }
    }

    fn handle_ticker(
        &mut self,
        events: &[crate::websocket::messages::WsTickerEvent],
        timestamp: &str,
        ts_init: UnixNanos,
    ) -> Option<NautilusWsMessage> {
        let ts_event = crate::http::parse::parse_rfc3339_timestamp(timestamp).unwrap_or(ts_init);

        let mut first: Option<NautilusWsMessage> = None;
        let aliases = self.subscription_aliases.load();

        for event in events {
            for ticker in &event.tickers {
                let instrument_id =
                    resolve_instrument_id_from_aliases(&aliases, &ticker.product_id);

                let instrument = match self.instruments.get(&instrument_id) {
                    Some(inst) => inst,
                    None => {
                        log::warn!("No instrument cached for {instrument_id}");
                        continue;
                    }
                };

                match parse_ws_ticker(ticker, instrument, ts_event, ts_init) {
                    Ok(quote) => {
                        let msg = NautilusWsMessage::Quote(quote);

                        if first.is_none() {
                            first = Some(msg);
                        } else {
                            self.buffer.push(msg);
                        }
                    }
                    Err(e) => log::warn!("Failed to parse ticker: {e}"),
                }
            }
        }

        if first.is_some() {
            self.buffer.reverse();
        }
        first
    }

    fn handle_user_events(
        &mut self,
        events: &[crate::websocket::messages::WsUserEvent],
        timestamp: &str,
        ts_init: UnixNanos,
    ) -> Option<NautilusWsMessage> {
        let Some(account_id) = self.account_id else {
            log::debug!(
                "Dropping user event: account_id not set (call SetAccountId after connect)"
            );
            return None;
        };

        let ts_event = match crate::http::parse::parse_rfc3339_timestamp(timestamp) {
            Ok(ts) => ts,
            Err(e) => {
                log::warn!("Failed to parse user message timestamp {timestamp}: {e}");
                ts_init
            }
        };

        let mut first: Option<NautilusWsMessage> = None;
        let aliases = self.subscription_aliases.load();

        for event in events {
            let is_snapshot = matches!(event.event_type, WsEventType::Snapshot);

            for order in &event.orders {
                let instrument_id = resolve_instrument_id_from_aliases(&aliases, &order.product_id);
                let instrument = match self.instruments.get(&instrument_id).cloned() {
                    Some(inst) => inst,
                    None => {
                        log::warn!("No instrument cached for {instrument_id}");
                        continue;
                    }
                };

                self.emit_user_event_messages(
                    order,
                    &instrument,
                    account_id,
                    is_snapshot,
                    ts_event,
                    ts_init,
                    &mut first,
                );
            }
        }

        if first.is_some() {
            self.buffer.reverse();
        }
        first
    }

    #[allow(clippy::too_many_arguments)]
    fn emit_user_event_messages(
        &mut self,
        order: &WsOrderUpdate,
        instrument: &InstrumentAny,
        account_id: AccountId,
        is_snapshot: bool,
        ts_event: UnixNanos,
        ts_init: UnixNanos,
        first: &mut Option<NautilusWsMessage>,
    ) {
        let report = match parse_ws_user_event_to_order_status_report(
            order, instrument, account_id, ts_event, ts_init,
        ) {
            Ok(r) => r,
            Err(e) => {
                log::warn!("Failed to parse user order update: {e}");
                return;
            }
        };

        let msg = NautilusWsMessage::UserOrder(Box::new(UserOrderUpdate {
            report: Box::new(report),
            update: Box::new(order.clone()),
            instrument: instrument.clone(),
            is_snapshot,
            ts_event,
            ts_init,
        }));

        if first.is_none() {
            *first = Some(msg);
        } else {
            self.buffer.push(msg);
        }
    }

    fn handle_status_events(
        &mut self,
        events: &[crate::websocket::messages::WsStatusEvent],
        timestamp: &str,
        ts_init: UnixNanos,
    ) -> Option<NautilusWsMessage> {
        let ts_event = crate::http::parse::parse_rfc3339_timestamp(timestamp).unwrap_or(ts_init);

        let mut first: Option<NautilusWsMessage> = None;
        let aliases = self.subscription_aliases.load();

        for event in events {
            for product in &event.products {
                let canonical = product.id;
                let resolved = resolve_instrument_id_from_aliases(&aliases, &canonical);
                let Some(status) = parse_ws_status_product(product, resolved, ts_event, ts_init)
                else {
                    continue;
                };
                let msg = NautilusWsMessage::InstrumentStatus(Box::new(status));

                if first.is_none() {
                    first = Some(msg);
                } else {
                    self.buffer.push(msg);
                }
            }
        }

        if first.is_some() {
            self.buffer.reverse();
        }
        first
    }

    fn handle_futures_balance_summary(
        &mut self,
        events: Vec<crate::websocket::messages::WsFuturesBalanceSummaryEvent>,
    ) -> Option<NautilusWsMessage> {
        let mut first: Option<NautilusWsMessage> = None;

        for event in events {
            let msg = NautilusWsMessage::FuturesBalanceSummary(Box::new(event.fcm_balance_summary));

            if first.is_none() {
                first = Some(msg);
            } else {
                self.buffer.push(msg);
            }
        }

        if first.is_some() {
            self.buffer.reverse();
        }
        first
    }

    fn handle_candles(
        &mut self,
        events: &[crate::websocket::messages::WsCandlesEvent],
        ts_init: UnixNanos,
    ) -> Option<NautilusWsMessage> {
        let mut first: Option<NautilusWsMessage> = None;
        let aliases = self.subscription_aliases.load();

        for event in events {
            for candle in &event.candles {
                let key = candle.product_id.as_str();

                let bar_type = match self.bar_types.get(key) {
                    Some(bt) => *bt,
                    None => {
                        log::debug!("No bar type registered for {key}");
                        continue;
                    }
                };

                let instrument_id =
                    resolve_instrument_id_from_aliases(&aliases, &candle.product_id);

                let instrument = match self.instruments.get(&instrument_id) {
                    Some(inst) => inst,
                    None => {
                        log::warn!("No instrument cached for {instrument_id}");
                        continue;
                    }
                };

                match parse_ws_candle(candle, bar_type, instrument, ts_init) {
                    Ok(bar) => {
                        let msg = NautilusWsMessage::Bar(bar);

                        if first.is_none() {
                            first = Some(msg);
                        } else {
                            self.buffer.push(msg);
                        }
                    }
                    Err(e) => log::warn!("Failed to parse candle: {e}"),
                }
            }
        }

        if first.is_some() {
            self.buffer.reverse();
        }
        first
    }
}

#[cfg(test)]
mod tests {
    use std::sync::{Arc, atomic::AtomicBool};

    use nautilus_model::{
        identifiers::Symbol,
        instruments::CurrencyPair,
        types::{Currency, Price, Quantity},
    };
    use rstest::rstest;

    use super::*;
    use crate::common::{consts::COINBASE_VENUE, testing::load_test_fixture};

    fn test_handler() -> FeedHandler {
        let (_cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
        let (_raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
        FeedHandler::new(
            Arc::new(AtomicBool::new(false)),
            cmd_rx,
            raw_rx,
            Arc::new(AtomicMap::new()),
        )
    }

    fn btc_usd_instrument() -> InstrumentAny {
        let instrument_id = InstrumentId::new(Symbol::new("BTC-USD"), *COINBASE_VENUE);
        InstrumentAny::CurrencyPair(
            CurrencyPair::builder()
                .instrument_id(instrument_id)
                .raw_symbol(Symbol::new("BTC-USD"))
                .base_currency(Currency::get_or_create_crypto("BTC"))
                .quote_currency(Currency::get_or_create_crypto("USD"))
                .price_precision(2)
                .size_precision(8)
                .price_increment(Price::from("0.01"))
                .size_increment(Quantity::from("0.00000001"))
                .min_quantity(Quantity::from("0.00000001"))
                .ts_event(UnixNanos::default())
                .ts_init(UnixNanos::default())
                .build()
                .unwrap(),
        )
    }

    #[rstest]
    fn test_handle_text_drops_user_channel_when_account_id_unset() {
        let json = load_test_fixture("ws_user.json");
        let mut handler = test_handler();

        // account_id is intentionally left unset; events should be dropped
        assert!(handler.handle_text(&json).is_none());
        assert!(handler.buffer.is_empty());
    }

    #[rstest]
    fn test_handle_user_event_emits_user_order_update() {
        use nautilus_model::{
            enums::{OrderSide, OrderStatus},
            identifiers::AccountId,
            types::Quantity,
        };

        use crate::common::enums::CoinbaseProductType;

        let json = load_test_fixture("ws_user.json");
        let mut handler = test_handler();
        handler.set_account_id(AccountId::new("COINBASE-001"));
        handler
            .instruments
            .insert(btc_usd_instrument().id(), btc_usd_instrument());

        let msg = handler
            .handle_text(&json)
            .expect("handler should emit a user-channel update");

        match msg {
            NautilusWsMessage::UserOrder(carrier) => {
                // Status report fields.
                assert_eq!(carrier.report.account_id.as_str(), "COINBASE-001");
                assert_eq!(carrier.report.instrument_id, btc_usd_instrument().id());
                assert_eq!(
                    carrier.report.venue_order_id.as_str(),
                    "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
                );
                assert_eq!(
                    carrier.report.client_order_id.unwrap().as_str(),
                    "11111-000000-000001"
                );
                assert_eq!(carrier.report.order_side, OrderSide::Buy.into());
                assert_eq!(carrier.report.order_status, OrderStatus::Accepted);
                assert_eq!(carrier.report.filled_qty, Quantity::from("0.00000000"));
                assert_eq!(carrier.report.quantity, Quantity::from("0.00100000"));

                // Raw venue update fields.
                assert_eq!(carrier.update.product_id, "BTC-USD");
                assert_eq!(carrier.update.product_type, CoinbaseProductType::Spot);
                assert_eq!(carrier.update.cumulative_quantity, "0");
                assert_eq!(carrier.update.leaves_quantity, "0.001");

                // Carrier metadata.
                assert_eq!(carrier.instrument.id(), btc_usd_instrument().id());
                assert!(carrier.ts_event.as_u64() > 0);
            }
            other => panic!("expected UserOrder, was {other:?}"),
        }
    }

    #[rstest]
    fn test_handle_text_emits_instrument_status_from_status_channel() {
        use nautilus_model::enums::MarketStatusAction;

        let json = r#"{
          "channel": "status",
          "client_id": "",
          "timestamp": "2023-02-09T20:29:49.753424311Z",
          "sequence_num": 0,
          "events": [
            {
              "type": "snapshot",
              "products": [
                {
                  "product_type": "SPOT",
                  "id": "BTC-USD",
                  "base_currency": "BTC",
                  "quote_currency": "USD",
                  "base_increment": "0.00000001",
                  "quote_increment": "0.01",
                  "display_name": "BTC/USD",
                  "status": "online",
                  "status_message": "",
                  "min_market_funds": "1"
                },
                {
                  "product_type": "SPOT",
                  "id": "ETH-USD",
                  "base_currency": "ETH",
                  "quote_currency": "USD",
                  "base_increment": "0.00000001",
                  "quote_increment": "0.01",
                  "display_name": "ETH/USD",
                  "status": "offline",
                  "status_message": "maintenance",
                  "min_market_funds": "1"
                }
              ]
            }
          ]
        }"#;
        let mut handler = test_handler();

        let first = handler
            .handle_text(json)
            .expect("status channel must emit InstrumentStatus");
        let NautilusWsMessage::InstrumentStatus(status) = first else {
            panic!("expected InstrumentStatus, was {first:?}");
        };
        assert_eq!(status.instrument_id, btc_usd_instrument().id());
        assert_eq!(status.action, MarketStatusAction::Trading);
        assert_eq!(status.is_trading, Some(true));
        assert!(status.reason.is_none());

        // The second product (ETH-USD offline) is buffered behind the BTC one.
        assert_eq!(handler.buffer.len(), 1);
        let NautilusWsMessage::InstrumentStatus(status) = handler.buffer.pop().unwrap() else {
            panic!("expected buffered InstrumentStatus");
        };
        assert_eq!(status.action, MarketStatusAction::Halt);
        assert_eq!(status.is_trading, Some(false));
        assert_eq!(
            status.reason.map(|s| s.to_string()),
            Some("maintenance".to_string())
        );
    }

    #[rstest]
    fn test_handle_l2_update_uses_batch_timestamp_for_all_deltas() {
        let json = load_test_fixture("ws_l2_data_update.json");
        let mut handler = test_handler();
        handler
            .instruments
            .insert(btc_usd_instrument().id(), btc_usd_instrument());

        let msg = handler
            .handle_text(&json)
            .expect("handler should emit deltas for a valid L2 update");

        let deltas = match msg {
            NautilusWsMessage::Deltas(d) => d,
            other => panic!("expected Deltas, was {other:?}"),
        };

        assert!(!deltas.deltas.is_empty());
        let expected_ts = deltas.deltas[0].ts_event;
        for delta in &deltas.deltas {
            assert_eq!(
                delta.ts_event, expected_ts,
                "all deltas in a batch must share ts_event"
            );
        }
    }

    #[rstest]
    fn test_handle_l2_update_malformed_timestamp_falls_back_to_ts_init() {
        let json = load_test_fixture("ws_l2_data_update.json")
            .replace("2026-04-07T14:30:01.456789Z", "not-a-valid-timestamp");
        let mut handler = test_handler();
        handler
            .instruments
            .insert(btc_usd_instrument().id(), btc_usd_instrument());

        let msg = handler
            .handle_text(&json)
            .expect("handler should still emit deltas when timestamp is malformed");

        let deltas = match msg {
            NautilusWsMessage::Deltas(d) => d,
            other => panic!("expected Deltas, was {other:?}"),
        };

        assert!(!deltas.deltas.is_empty());
        for delta in &deltas.deltas {
            assert_eq!(
                delta.ts_event, delta.ts_init,
                "malformed timestamp must fall back to ts_init"
            );
        }
    }

    #[rstest]
    fn test_handle_text_emits_futures_balance_summary_snapshot() {
        use rust_decimal::Decimal;

        let json = r#"{
          "channel": "futures_balance_summary",
          "client_id": "",
          "timestamp": "2023-02-09T20:33:57.609931463Z",
          "sequence_num": 0,
          "events": [
            {
              "type": "snapshot",
              "fcm_balance_summary": {
                "futures_buying_power": "100.00",
                "total_usd_balance": "200.00",
                "cbi_usd_balance": "300.00",
                "cfm_usd_balance": "400.00",
                "total_open_orders_hold_amount": "500.00",
                "unrealized_pnl": "600.00",
                "daily_realized_pnl": "0",
                "initial_margin": "700.00",
                "available_margin": "800.00",
                "liquidation_threshold": "900.00",
                "liquidation_buffer_amount": "1000.00",
                "liquidation_buffer_percentage": "1000",
                "intraday_margin_window_measure": {
                  "margin_window_type": "FCM_MARGIN_WINDOW_TYPE_INTRADAY",
                  "margin_level": "MARGIN_LEVEL_TYPE_BASE",
                  "initial_margin": "100.00",
                  "maintenance_margin": "200.00",
                  "liquidation_buffer_percentage": "1000",
                  "total_hold": "100.00",
                  "futures_buying_power": "400.00"
                },
                "overnight_margin_window_measure": {
                  "margin_window_type": "FCM_MARGIN_WINDOW_TYPE_OVERNIGHT",
                  "margin_level": "MARGIN_LEVEL_TYPE_BASE",
                  "initial_margin": "300.00",
                  "maintenance_margin": "200.00",
                  "liquidation_buffer_percentage": "1000",
                  "total_hold": "-30.00",
                  "futures_buying_power": "2000.00"
                }
              }
            }
          ]
        }"#;
        let mut handler = test_handler();

        let msg = handler
            .handle_text(json)
            .expect("handler should emit a futures balance summary");
        match msg {
            NautilusWsMessage::FuturesBalanceSummary(summary) => {
                assert_eq!(summary.futures_buying_power, Decimal::from(100));
                assert_eq!(summary.total_usd_balance, Decimal::from(200));
                assert_eq!(summary.total_open_orders_hold_amount, Decimal::from(500));
                assert_eq!(summary.available_margin, Decimal::from(800));
                let intraday = &summary.intraday_margin_window_measure;
                assert_eq!(intraday.initial_margin, Decimal::from(100));
                assert_eq!(intraday.maintenance_margin, Decimal::from(200));
                let overnight = &summary.overnight_margin_window_measure;
                assert_eq!(overnight.initial_margin, Decimal::from(300));
                assert_eq!(overnight.maintenance_margin, Decimal::from(200));
                // `total_hold` carries negative values on the wire; ensure
                // the signed decimal survives the round trip.
                assert_eq!(overnight.total_hold, "-30".parse::<Decimal>().unwrap());
            }
            other => panic!("expected FuturesBalanceSummary, was {other:?}"),
        }
    }

    #[rstest]
    fn test_handle_text_routes_reconnected_sentinel() {
        let mut handler = test_handler();
        let result = handler.handle_text(RECONNECTED);
        assert!(matches!(result, Some(NautilusWsMessage::Reconnected)));
    }

    #[rstest]
    fn test_heartbeat_counter_tracks_sequence_and_detects_gaps() {
        let mut handler = test_handler();

        assert_eq!(handler.note_heartbeat_counter(42), None);
        assert_eq!(handler.last_heartbeat_counter, Some(42));
        assert_eq!(handler.note_heartbeat_counter(43), None);
        assert_eq!(handler.last_heartbeat_counter, Some(43));
        assert_eq!(handler.note_heartbeat_counter(45), Some((44, 45)));
        assert_eq!(handler.last_heartbeat_counter, Some(45));
        assert_eq!(handler.note_heartbeat_counter(40), Some((46, 40)));
        assert_eq!(handler.last_heartbeat_counter, Some(40));
    }

    #[rstest]
    fn test_handle_text_heartbeats_updates_counter_and_reconnect_resets_it() {
        let json = load_test_fixture("ws_heartbeats.json");
        let mut handler = test_handler();

        assert!(handler.handle_text(&json).is_none());
        assert_eq!(handler.last_heartbeat_counter, Some(42));

        let next = json.replace("\"heartbeat_counter\": 42", "\"heartbeat_counter\": 43");
        assert!(handler.handle_text(&next).is_none());
        assert_eq!(handler.last_heartbeat_counter, Some(43));

        let result = handler.handle_text(RECONNECTED);
        assert!(matches!(result, Some(NautilusWsMessage::Reconnected)));
        assert_eq!(handler.last_heartbeat_counter, None);
    }

    #[rstest]
    fn test_signal_release_acquire_exits_handler_loop() {
        use std::sync::atomic::Ordering;

        let signal = Arc::new(AtomicBool::new(false));
        let (_cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
        let (_raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
        let mut handler =
            FeedHandler::new(signal.clone(), cmd_rx, raw_rx, Arc::new(AtomicMap::new()));

        signal.store(true, Ordering::Release);

        let runtime = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let result = runtime.block_on(async { handler.next().await });
        assert!(result.is_none(), "{result:?}");
    }
}