nautilus_binance/spot/websocket/trading/

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.
// -------------------------------------------------------------------------------------------------

//! Binance Spot WebSocket API message handler.
//!
//! The handler runs in a dedicated Tokio task as the I/O boundary between the client
//! orchestrator and the network layer. It exclusively owns the `WebSocketClient` and
//! processes commands from the client via an unbounded channel.
//!
//! ## Responsibilities
//!
//! - Command processing: Receives `BinanceSpotWsTradingCommand` from client, serializes to JSON requests.
//! - Response decoding: Parses SBE binary responses using schema 3 decoders.
//! - Request correlation: Matches responses to pending requests by ID.
//! - Message transformation: Emits `BinanceSpotWsTradingMessage` events to client via channel.

use std::{
    fmt::Debug,
    sync::{
        Arc,
        atomic::{AtomicBool, AtomicU64, Ordering},
    },
};

use ahash::AHashMap;
use nautilus_network::{RECONNECTED, websocket::WebSocketClient};
use tokio_tungstenite::tungstenite::Message;

use super::{
    client::BINANCE_WS_RATE_LIMIT_KEY_ORDER,
    error::{BinanceWsApiError, BinanceWsApiResult},
    messages::{
        BinanceSpotWsTradingCommand, BinanceSpotWsTradingMessage, BinanceSpotWsTradingRequest,
        BinanceSpotWsTradingRequestMeta, method,
    },
};
use crate::{
    common::credential::{SigningCredential, canonical_ws_query_string},
    spot::{
        enums::BinanceSpotUserDataEventType,
        http::parse,
        sbe::spot::{
            ReadBuf,
            error_response_codec::ErrorResponseDecoder,
            message_header_codec,
            web_socket_response_codec::{SBE_TEMPLATE_ID, WebSocketResponseDecoder},
        },
    },
};

/// Binance Spot WebSocket API handler.
///
/// Runs in a dedicated Tokio task, processing commands from the client
/// and transforming raw WebSocket messages into Nautilus domain events.
/// Messages are sent to the client via the output channel.
pub struct BinanceSpotWsTradingHandler {
    signal: Arc<AtomicBool>,
    inner: Option<WebSocketClient>,
    cmd_rx: tokio::sync::mpsc::UnboundedReceiver<BinanceSpotWsTradingCommand>,
    raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
    out_tx: tokio::sync::mpsc::UnboundedSender<BinanceSpotWsTradingMessage>,
    credential: Arc<SigningCredential>,
    pending_requests: AHashMap<String, BinanceSpotWsTradingRequestMeta>,
    request_id_counter: AtomicU64,
    recv_window_ms: Option<u64>,
}

impl Debug for BinanceSpotWsTradingHandler {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct(stringify!(BinanceSpotWsTradingHandler))
            .field("inner", &self.inner.as_ref().map(|_| "<client>"))
            .field(
                "pending_requests",
                &format!("{} pending", self.pending_requests.len()),
            )
            .finish_non_exhaustive()
    }
}

impl BinanceSpotWsTradingHandler {
    /// Creates a new handler instance.
    #[must_use]
    pub fn new(
        signal: Arc<AtomicBool>,
        cmd_rx: tokio::sync::mpsc::UnboundedReceiver<BinanceSpotWsTradingCommand>,
        raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
        out_tx: tokio::sync::mpsc::UnboundedSender<BinanceSpotWsTradingMessage>,
        credential: Arc<SigningCredential>,
    ) -> Self {
        Self {
            signal,
            inner: None,
            cmd_rx,
            raw_rx,
            out_tx,
            credential,
            pending_requests: AHashMap::new(),
            request_id_counter: AtomicU64::new(1000),
            recv_window_ms: None,
        }
    }

    /// Configures the receive window added before request signing.
    #[must_use]
    pub const fn with_recv_window(mut self, recv_window_ms: Option<u64>) -> Self {
        self.recv_window_ms = recv_window_ms;
        self
    }

    /// Runs the main event loop for commands and raw messages.
    ///
    /// Sends output messages via `out_tx` channel. Returns `false` when disconnected
    /// or the signal is set, indicating the handler should exit.
    pub async fn run(&mut self) -> bool {
        loop {
            if self.signal.load(Ordering::Relaxed) {
                return false;
            }

            tokio::select! {
                Some(cmd) = self.cmd_rx.recv() => {
                    match cmd {
                        BinanceSpotWsTradingCommand::SetClient(client) => {
                            log::debug!("Handler received WebSocket client");
                            self.inner = Some(client);
                            self.emit(BinanceSpotWsTradingMessage::Connected);
                        }
                        BinanceSpotWsTradingCommand::Disconnect => {
                            log::debug!("Handler disconnecting WebSocket client");
                            self.inner = None;
                            return false;
                        }
                        BinanceSpotWsTradingCommand::PlaceOrder { id, params } => {
                            if let Err(e) = self.handle_place_order(id.clone(), params).await {
                                log::debug!("Failed to handle place order command: {e}");
                                self.pending_requests.remove(&id);
                                self.emit(BinanceSpotWsTradingMessage::RequestFailed {
                                    request_id: id,
                                    msg: e.to_string(),
                                });
                            }
                        }
                        BinanceSpotWsTradingCommand::CancelOrder { id, params } => {
                            if let Err(e) = self.handle_cancel_order(id.clone(), params).await {
                                log::debug!("Failed to handle cancel order command: {e}");
                                self.pending_requests.remove(&id);
                                self.emit(BinanceSpotWsTradingMessage::RequestFailed {
                                    request_id: id,
                                    msg: e.to_string(),
                                });
                            }
                        }
                        BinanceSpotWsTradingCommand::CancelReplaceOrder { id, params } => {
                            if let Err(e) = self.handle_cancel_replace_order(id.clone(), params).await {
                                log::debug!("Failed to handle cancel replace command: {e}");
                                self.pending_requests.remove(&id);
                                self.emit(BinanceSpotWsTradingMessage::RequestFailed {
                                    request_id: id,
                                    msg: e.to_string(),
                                });
                            }
                        }
                        BinanceSpotWsTradingCommand::CancelAllOrders { id, symbol } => {
                            if let Err(e) = self.handle_cancel_all_orders(id.clone(), symbol).await {
                                log::debug!("Failed to handle cancel all command: {e}");
                                self.pending_requests.remove(&id);
                                self.emit(BinanceSpotWsTradingMessage::RequestFailed {
                                    request_id: id,
                                    msg: e.to_string(),
                                });
                            }
                        }
                        BinanceSpotWsTradingCommand::SessionLogon => {
                            if let Err(e) = self.handle_session_logon().await {
                                log::error!("Session logon failed: {e}");
                                self.emit(BinanceSpotWsTradingMessage::AuthenticationRejected(
                                    format!("Session logon failed: {e}"),
                                ));
                            }
                        }
                        BinanceSpotWsTradingCommand::SubscribeUserData => {
                            if let Err(e) = self.handle_subscribe_user_data().await {
                                log::error!("User data subscribe failed: {e}");
                                self.emit(
                                    BinanceSpotWsTradingMessage::UserDataSubscriptionRejected(
                                        format!("User data subscribe failed: {e}"),
                                    ),
                                );
                            }
                        }
                    }
                }
                Some(msg) = self.raw_rx.recv() => {
                    if let Message::Text(ref text) = msg
                        && text.as_str() == RECONNECTED
                    {
                        log::debug!("Handler received reconnection signal");

                        // Fail any pending requests - they won't get responses on new connection
                        self.fail_pending_requests();

                        self.emit(BinanceSpotWsTradingMessage::Reconnected);
                        continue;
                    }

                    self.handle_message(msg);
                }
                else => {
                    // Both channels closed
                    return false;
                }
            }
        }
    }

    /// Sends a message to the output channel.
    fn emit(&self, msg: BinanceSpotWsTradingMessage) {
        if let Err(e) = self.out_tx.send(msg) {
            log::error!("Failed to send message to output channel: {e}");
        }
    }

    /// Fails all pending requests after a reconnection.
    fn fail_pending_requests(&mut self) {
        if self.pending_requests.is_empty() {
            return;
        }

        let count = self.pending_requests.len();
        log::warn!("Failing {count} pending requests after reconnection");

        let pending = std::mem::take(&mut self.pending_requests);
        for (request_id, _meta) in pending {
            self.emit(BinanceSpotWsTradingMessage::RequestFailed {
                request_id,
                msg: "Connection lost before response received".to_string(),
            });
        }
    }

    async fn handle_place_order(
        &mut self,
        id: String,
        params: crate::spot::http::query::NewOrderParams,
    ) -> BinanceWsApiResult<()> {
        let params_json = serde_json::to_value(&params)
            .map_err(|e| BinanceWsApiError::ClientError(e.to_string()))?;
        let signed_params = self.sign_params(params_json)?;

        let request = BinanceSpotWsTradingRequest::new(&id, method::ORDER_PLACE, signed_params);
        self.pending_requests
            .insert(id.clone(), BinanceSpotWsTradingRequestMeta::PlaceOrder);
        self.send_request(request).await
    }

    async fn handle_cancel_order(
        &mut self,
        id: String,
        params: crate::spot::http::query::CancelOrderParams,
    ) -> BinanceWsApiResult<()> {
        let params_json = serde_json::to_value(&params)
            .map_err(|e| BinanceWsApiError::ClientError(e.to_string()))?;
        let signed_params = self.sign_params(params_json)?;

        let request = BinanceSpotWsTradingRequest::new(&id, method::ORDER_CANCEL, signed_params);
        self.pending_requests
            .insert(id.clone(), BinanceSpotWsTradingRequestMeta::CancelOrder);
        self.send_request(request).await
    }

    async fn handle_cancel_replace_order(
        &mut self,
        id: String,
        params: crate::spot::http::query::CancelReplaceOrderParams,
    ) -> BinanceWsApiResult<()> {
        let params_json = serde_json::to_value(&params)
            .map_err(|e| BinanceWsApiError::ClientError(e.to_string()))?;
        let signed_params = self.sign_params(params_json)?;

        let request =
            BinanceSpotWsTradingRequest::new(&id, method::ORDER_CANCEL_REPLACE, signed_params);
        self.pending_requests.insert(
            id.clone(),
            BinanceSpotWsTradingRequestMeta::CancelReplaceOrder,
        );
        self.send_request(request).await
    }

    async fn handle_cancel_all_orders(
        &mut self,
        id: String,
        symbol: String,
    ) -> BinanceWsApiResult<()> {
        let params_json = serde_json::json!({ "symbol": symbol });
        let signed_params = self.sign_params(params_json)?;

        let request =
            BinanceSpotWsTradingRequest::new(&id, method::OPEN_ORDERS_CANCEL_ALL, signed_params);
        self.pending_requests
            .insert(id.clone(), BinanceSpotWsTradingRequestMeta::CancelAllOrders);
        self.send_request(request).await
    }

    async fn handle_session_logon(&mut self) -> BinanceWsApiResult<()> {
        let id = self.next_request_id();
        let params_json = serde_json::json!({});
        let signed_params = self.sign_params(params_json)?;

        let request = BinanceSpotWsTradingRequest::new(&id, method::SESSION_LOGON, signed_params);
        self.pending_requests
            .insert(id, BinanceSpotWsTradingRequestMeta::SessionLogon);
        self.send_request(request).await
    }

    async fn handle_subscribe_user_data(&mut self) -> BinanceWsApiResult<()> {
        let id = self.next_request_id();
        let request = BinanceSpotWsTradingRequest::new(
            &id,
            "userDataStream.subscribe",
            serde_json::json!({}),
        );
        self.pending_requests
            .insert(id, BinanceSpotWsTradingRequestMeta::SubscribeUserData);
        self.send_request(request).await
    }

    fn next_request_id(&self) -> String {
        let id = self.request_id_counter.fetch_add(1, Ordering::Relaxed);
        format!("ws-{id}")
    }

    fn sign_params(&self, mut params: serde_json::Value) -> BinanceWsApiResult<serde_json::Value> {
        let timestamp = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map_err(|e| BinanceWsApiError::ClientError(e.to_string()))?
            .as_millis() as i64;

        if let Some(obj) = params.as_object_mut() {
            obj.insert("timestamp".to_string(), serde_json::json!(timestamp));
            obj.insert(
                "apiKey".to_string(),
                serde_json::json!(self.credential.api_key()),
            );

            if let Some(recv_window_ms) = self.recv_window_ms {
                obj.insert("recvWindow".to_string(), serde_json::json!(recv_window_ms));
            }
        }

        // Sign over a key-sorted query string: Binance's WS API verifies the
        // signature against the parameters sorted by key, which must not depend
        // on serde_json's Map iteration order (issue #4410).
        let query_string = canonical_ws_query_string(
            params
                .as_object()
                .into_iter()
                .flatten()
                .map(|(k, v)| (k.as_str(), v)),
        )
        .map_err(|e| BinanceWsApiError::ClientError(e.to_string()))?;
        let signature = self.credential.sign(&query_string);

        if let Some(obj) = params.as_object_mut() {
            obj.insert("signature".to_string(), serde_json::json!(signature));
        }

        Ok(params)
    }

    async fn send_request(
        &mut self,
        request: BinanceSpotWsTradingRequest,
    ) -> BinanceWsApiResult<()> {
        let client = self.inner.as_mut().ok_or_else(|| {
            BinanceWsApiError::ConnectionError("WebSocket not connected".to_string())
        })?;

        let json = serde_json::to_string(&request)
            .map_err(|e| BinanceWsApiError::ClientError(e.to_string()))?;

        log::debug!(
            "Sending WebSocket API request id={} method={}",
            request.id,
            request.method
        );

        let keys = Some(BINANCE_WS_RATE_LIMIT_KEY_ORDER.as_slice());

        let is_session_setup = matches!(
            self.pending_requests.get(&request.id),
            Some(
                BinanceSpotWsTradingRequestMeta::SessionLogon
                    | BinanceSpotWsTradingRequestMeta::SubscribeUserData
            )
        );

        let result = if is_session_setup {
            client
                .send_text_on_connection(json, keys, client.connection_epoch())
                .await
        } else {
            client.send_text(json, keys).await
        };

        result.map_err(|e| {
            BinanceWsApiError::ConnectionError(format!("Failed to send request: {e}"))
        })?;

        Ok(())
    }

    fn handle_message(&mut self, msg: Message) {
        match msg {
            Message::Binary(data) => self.handle_binary_response(&data),
            Message::Text(text) => self.handle_text_response(&text),
            Message::Ping(_) | Message::Pong(_) => {}
            Message::Close(frame) => {
                log::debug!("WebSocket closed: {frame:?}");
            }
            Message::Frame(_) => {}
        }
    }

    fn handle_binary_response(&mut self, data: &[u8]) {
        match self.decode_ws_api_response(data) {
            Ok(response) => self.emit(response),
            Err(e) => {
                log::error!("Failed to decode WebSocket API response: {e}");
                self.emit(BinanceSpotWsTradingMessage::Error(e.to_string()));
            }
        }
    }

    fn handle_text_response(&mut self, text: &str) {
        let json: serde_json::Value = match serde_json::from_str(text) {
            Ok(j) => j,
            Err(e) => {
                log::warn!("Failed to parse text response as JSON: {e}");
                return;
            }
        };

        // User data events arrive wrapped: {"subscriptionId": N, "event": {...}}
        if let Some(event) = json.get("event") {
            self.handle_user_data_event(event);
            return;
        }

        // Legacy listen-key streams, including Binance US, send the event directly.
        if json.get("e").is_some() {
            self.handle_user_data_event(&json);
            return;
        }

        // WS API responses have an "id" field for request correlation
        if let Some(id) = json.get("id") {
            let id_str = match id {
                serde_json::Value::String(s) => s.clone(),
                serde_json::Value::Number(n) => n.to_string(),
                _ => return,
            };

            if let Some(meta) = self.pending_requests.remove(&id_str) {
                // Check for error: nested {"error": {"code": N, "msg": "..."}}
                // or top-level {"code": N, "msg": "..."}
                let error_info = json
                    .get("error")
                    .map(|e| {
                        (
                            e.get("code")
                                .and_then(|v| v.as_i64())
                                .and_then(|code| i32::try_from(code).ok()),
                            e.get("msg")
                                .and_then(|v| v.as_str())
                                .unwrap_or("Unknown error")
                                .to_string(),
                        )
                    })
                    .or_else(|| {
                        json.get("code").and_then(|c| c.as_i64()).map(|code| {
                            let msg = json
                                .get("msg")
                                .and_then(|v| v.as_str())
                                .unwrap_or("Unknown error")
                                .to_string();
                            (i32::try_from(code).ok(), msg)
                        })
                    });

                if let Some((code, msg)) = error_info {
                    let status = json
                        .get("status")
                        .and_then(|v| v.as_u64())
                        .map(|s| s as u16);
                    let rejection = match code {
                        Some(code) => {
                            self.create_rejection(id_str, status.unwrap_or(0), code, msg, meta)
                        }
                        None => BinanceSpotWsTradingMessage::RequestFailed {
                            request_id: id_str,
                            msg: format!("Missing or invalid venue error code: {msg}"),
                        },
                    };
                    self.emit(rejection);
                    return;
                }

                // Success response
                match meta {
                    BinanceSpotWsTradingRequestMeta::SessionLogon => {
                        log::debug!("Session authenticated");
                        self.emit(BinanceSpotWsTradingMessage::Authenticated);
                    }
                    BinanceSpotWsTradingRequestMeta::SubscribeUserData => {
                        let subscription_id = json
                            .get("result")
                            .and_then(|r| r.get("subscriptionId"))
                            .map(|v| v.to_string())
                            .unwrap_or_default();
                        log::debug!("User data stream subscribed: id={subscription_id}");
                        self.emit(BinanceSpotWsTradingMessage::UserDataSubscribed {
                            subscription_id,
                        });
                    }
                    _ => {
                        // Order operation responses come as SBE binary, not JSON text.
                        // If we get a JSON success for an order operation, log it.
                        log::debug!("Unexpected JSON success for request {id_str}");
                    }
                }
                return;
            }

            // Error response without matching pending request
            if let Some(code) = json.get("code").and_then(|v| v.as_i64()) {
                let msg = json
                    .get("msg")
                    .and_then(|v| v.as_str())
                    .unwrap_or("Unknown error");
                log::warn!(
                    "Received error response without matching request ID: code={code} msg={msg}"
                );
            }
            return;
        }

        // Stream termination event
        if json.get("eventStreamTerminated").is_some() {
            log::warn!("User data stream terminated, resubscribe needed");
            return;
        }

        log::debug!("Unhandled text message: {} bytes", text.len());
    }

    fn handle_user_data_event(&self, event: &serde_json::Value) {
        if let Some(msg) = classify_user_data_event(event) {
            self.emit(msg);
        }
    }

    fn decode_ws_api_response(
        &mut self,
        data: &[u8],
    ) -> Result<BinanceSpotWsTradingMessage, BinanceWsApiError> {
        // Check template ID before parsing
        if data.len() >= message_header_codec::ENCODED_LENGTH {
            let buf = ReadBuf::new(data);
            let template_id = buf.get_u16_at(2);

            // User data stream events arrive as SBE with their own template IDs
            // (not wrapped in WebSocketResponse template 50).
            match template_id {
                601 => {
                    log::debug!("Received SBE BalanceUpdateEvent ({} bytes)", data.len());
                    match super::decode_sbe::decode_balance_update(data) {
                        Ok(msg) => {
                            log::debug!(
                                "SBE balance update: asset={}, delta={}",
                                msg.asset,
                                msg.delta
                            );
                            return Ok(BinanceSpotWsTradingMessage::BalanceUpdate(msg));
                        }
                        Err(e) => {
                            log::error!("Failed to decode SBE BalanceUpdateEvent: {e}");
                            return Ok(BinanceSpotWsTradingMessage::Error(format!(
                                "SBE BalanceUpdateEvent decode failed: {e}"
                            )));
                        }
                    }
                }
                603 => {
                    log::debug!("Received SBE ExecutionReportEvent ({} bytes)", data.len());
                    match super::decode_sbe::decode_execution_report(data) {
                        Ok(report) => {
                            log::debug!(
                                "SBE execution report: symbol={}, order_id={}, exec={:?}, status={:?}",
                                report.symbol,
                                report.order_id,
                                report.execution_type,
                                report.order_status
                            );
                            return Ok(BinanceSpotWsTradingMessage::ExecutionReport(Box::new(
                                report,
                            )));
                        }
                        Err(e) => {
                            log::error!("Failed to decode SBE ExecutionReportEvent: {e}");
                            return Ok(BinanceSpotWsTradingMessage::Error(format!(
                                "SBE ExecutionReportEvent decode failed: {e}"
                            )));
                        }
                    }
                }
                606 => {
                    return Ok(match super::decode_sbe::decode_list_status(data) {
                        Ok(status) => BinanceSpotWsTradingMessage::ListStatus(status),
                        Err(e) => BinanceSpotWsTradingMessage::Error(format!(
                            "SBE ListStatusEvent decode failed: {e}"
                        )),
                    });
                }
                607 => {
                    log::debug!(
                        "Received SBE OutboundAccountPositionEvent ({} bytes)",
                        data.len()
                    );

                    match super::decode_sbe::decode_account_position(data) {
                        Ok(msg) => {
                            log::debug!("SBE account position: {} balance(s)", msg.balances.len());
                            return Ok(BinanceSpotWsTradingMessage::AccountPosition(msg));
                        }
                        Err(e) => {
                            log::error!("Failed to decode SBE OutboundAccountPositionEvent: {e}");
                            return Ok(BinanceSpotWsTradingMessage::Error(format!(
                                "SBE OutboundAccountPositionEvent decode failed: {e}"
                            )));
                        }
                    }
                }
                610 => {
                    let event_time = parse_server_shutdown_event_time_ms(data);
                    return Ok(BinanceSpotWsTradingMessage::ServerShutdown { event_time });
                }
                _ => {} // Fall through to WebSocketResponse parsing
            }
        }

        // Standard WebSocketResponse envelope (template 50)
        let (request_id, status, result_data) = self.parse_envelope(data)?;

        // Look up the pending request by ID
        let meta = self.pending_requests.remove(&request_id).ok_or_else(|| {
            BinanceWsApiError::UnknownRequestId(format!("No pending request for ID: {request_id}"))
        })?;

        // Check for error status (non-200)
        if status != 200 {
            return Ok(match Self::try_decode_sbe_error(&result_data) {
                Some((code, msg)) => self.create_rejection(request_id, status, code, msg, meta),
                // An undecodable error payload carries no definitive command evidence
                None => BinanceSpotWsTradingMessage::RequestFailed {
                    request_id,
                    msg: format!("Request failed with status {status}; error payload undecodable"),
                },
            });
        }

        // Decode the inner payload based on request type
        match meta {
            BinanceSpotWsTradingRequestMeta::PlaceOrder => {
                let response = parse::decode_new_order_full(&result_data)?;
                Ok(BinanceSpotWsTradingMessage::OrderAccepted {
                    request_id,
                    response,
                })
            }
            BinanceSpotWsTradingRequestMeta::CancelOrder => {
                let response = parse::decode_cancel_order(&result_data)?;
                Ok(BinanceSpotWsTradingMessage::OrderCanceled {
                    request_id,
                    response,
                })
            }
            BinanceSpotWsTradingRequestMeta::CancelReplaceOrder => {
                let (cancel_response, new_order_response) =
                    parse::decode_cancel_replace_orders(&result_data)?;
                Ok(BinanceSpotWsTradingMessage::CancelReplaceAccepted {
                    request_id,
                    cancel_response,
                    new_order_response,
                })
            }
            BinanceSpotWsTradingRequestMeta::CancelAllOrders => {
                let responses = parse::decode_cancel_open_orders(&result_data)?;
                Ok(BinanceSpotWsTradingMessage::AllOrdersCanceled {
                    request_id,
                    responses,
                })
            }
            BinanceSpotWsTradingRequestMeta::SessionLogon => {
                log::debug!("Session authenticated (SBE response)");
                Ok(BinanceSpotWsTradingMessage::Authenticated)
            }
            BinanceSpotWsTradingRequestMeta::SubscribeUserData => {
                log::debug!("User data stream subscribed (SBE response)");
                Ok(BinanceSpotWsTradingMessage::UserDataSubscribed {
                    subscription_id: request_id,
                })
            }
        }
    }

    /// Parses the WebSocketResponse SBE envelope.
    ///
    /// Returns (request_id, status, result_payload).
    fn parse_envelope(&self, data: &[u8]) -> Result<(String, u16, Vec<u8>), BinanceWsApiError> {
        if data.len() < message_header_codec::ENCODED_LENGTH {
            return Err(BinanceWsApiError::DecodeError(
                crate::spot::sbe::error::SbeDecodeError::BufferTooShort {
                    expected: message_header_codec::ENCODED_LENGTH,
                    actual: data.len(),
                },
            ));
        }

        let buf = ReadBuf::new(data);

        // Parse message header
        let block_length = buf.get_u16_at(0);
        let template_id = buf.get_u16_at(2);

        if template_id != SBE_TEMPLATE_ID {
            return Err(BinanceWsApiError::DecodeError(
                crate::spot::sbe::error::SbeDecodeError::UnknownTemplateId(template_id),
            ));
        }

        let version = buf.get_u16_at(6);

        // Create decoder at offset after message header
        let decoder = WebSocketResponseDecoder::default().wrap(
            buf,
            message_header_codec::ENCODED_LENGTH,
            block_length,
            version,
        );

        // Read status from fixed block (offset 1 within block)
        let status = decoder.status();

        // Skip rate_limits group
        let mut rate_limits = decoder.rate_limits_decoder();
        while rate_limits.advance().unwrap_or(None).is_some() {}
        let mut decoder = rate_limits.parent().map_err(|e| {
            BinanceWsApiError::ClientError(format!("Failed to get parent from rate_limits: {e}"))
        })?;

        // Extract request ID
        let id_coords = decoder.id_decoder();
        let id_bytes = decoder.id_slice(id_coords);
        let request_id = String::from_utf8_lossy(id_bytes).to_string();

        // Extract result payload - copy to owned Vec to avoid lifetime issues
        let result_coords = decoder.result_decoder();
        let result_data = decoder.result_slice(result_coords).to_vec();

        Ok((request_id, status, result_data))
    }

    fn create_rejection(
        &self,
        request_id: String,
        status: u16,
        code: i32,
        msg: String,
        meta: BinanceSpotWsTradingRequestMeta,
    ) -> BinanceSpotWsTradingMessage {
        match meta {
            BinanceSpotWsTradingRequestMeta::PlaceOrder => {
                BinanceSpotWsTradingMessage::OrderRejected {
                    request_id,
                    status,
                    code,
                    msg,
                }
            }
            BinanceSpotWsTradingRequestMeta::CancelOrder => {
                BinanceSpotWsTradingMessage::CancelRejected {
                    request_id,
                    status,
                    code,
                    msg,
                }
            }
            BinanceSpotWsTradingRequestMeta::CancelReplaceOrder => {
                BinanceSpotWsTradingMessage::CancelReplaceRejected {
                    request_id,
                    status,
                    code,
                    msg,
                }
            }
            BinanceSpotWsTradingRequestMeta::CancelAllOrders => {
                BinanceSpotWsTradingMessage::CancelRejected {
                    request_id,
                    status,
                    code,
                    msg,
                }
            }
            BinanceSpotWsTradingRequestMeta::SessionLogon => {
                BinanceSpotWsTradingMessage::AuthenticationRejected(format!("code={code}: {msg}"))
            }
            BinanceSpotWsTradingRequestMeta::SubscribeUserData => {
                BinanceSpotWsTradingMessage::UserDataSubscriptionRejected(format!(
                    "code={code}: {msg}"
                ))
            }
        }
    }

    // Decodes the SBE error response to extract the Binance error code and message
    fn try_decode_sbe_error(data: &[u8]) -> Option<(i32, String)> {
        const HEADER_LEN: usize = 8;

        if data.len()
            < HEADER_LEN + crate::spot::sbe::spot::error_response_codec::SBE_BLOCK_LENGTH as usize
        {
            return None;
        }

        let buf = ReadBuf::new(data);
        let header = message_header_codec::MessageHeaderDecoder::default().wrap(buf, 0);
        if header.template_id() != crate::spot::sbe::spot::error_response_codec::SBE_TEMPLATE_ID {
            return None;
        }

        let mut decoder = ErrorResponseDecoder::default().header(header, 0);
        let code = i32::from(decoder.code());
        let msg_coords = decoder.msg_decoder();
        let msg_bytes = decoder.msg_slice(msg_coords);
        let msg = String::from_utf8_lossy(msg_bytes).into_owned();

        Some((code, msg))
    }
}

/// Classifies a JSON user-data event into a trading message, if any.
///
/// Returns `None` when the event type is unknown or the payload fails to
/// deserialize; in that case the caller logs and drops the event.
pub(crate) fn classify_user_data_event(
    event: &serde_json::Value,
) -> Option<BinanceSpotWsTradingMessage> {
    let event_type = event
        .get("e")
        .and_then(|v| serde_json::from_value::<BinanceSpotUserDataEventType>(v.clone()).ok())
        .unwrap_or(BinanceSpotUserDataEventType::Unknown);

    match event_type {
        BinanceSpotUserDataEventType::ExecutionReport => {
            match serde_json::from_value::<super::user_data::BinanceSpotExecutionReport>(
                event.clone(),
            ) {
                Ok(report) => {
                    log::debug!(
                        "Execution report: symbol={}, order_id={}, exec={:?}, status={:?}",
                        report.symbol,
                        report.order_id,
                        report.execution_type,
                        report.order_status
                    );
                    Some(BinanceSpotWsTradingMessage::ExecutionReport(Box::new(
                        report,
                    )))
                }
                Err(e) => {
                    log::warn!("Failed to parse execution report: {e}");
                    None
                }
            }
        }
        BinanceSpotUserDataEventType::OutboundAccountPosition => {
            match serde_json::from_value::<super::user_data::BinanceSpotAccountPositionMsg>(
                event.clone(),
            ) {
                Ok(msg) => {
                    log::debug!("Account position update: {} balance(s)", msg.balances.len());
                    Some(BinanceSpotWsTradingMessage::AccountPosition(msg))
                }
                Err(e) => {
                    log::warn!("Failed to parse account position: {e}");
                    None
                }
            }
        }
        BinanceSpotUserDataEventType::BalanceUpdate => {
            match serde_json::from_value::<super::user_data::BinanceSpotBalanceUpdateMsg>(
                event.clone(),
            ) {
                Ok(msg) => {
                    log::debug!("Balance update: asset={}, delta={}", msg.asset, msg.delta);
                    Some(BinanceSpotWsTradingMessage::BalanceUpdate(msg))
                }
                Err(e) => {
                    log::warn!("Failed to parse balance update: {e}");
                    None
                }
            }
        }
        BinanceSpotUserDataEventType::ServerShutdown => {
            let event_time = event.get("E").and_then(|v| v.as_i64()).unwrap_or_default();
            Some(BinanceSpotWsTradingMessage::ServerShutdown { event_time })
        }
        BinanceSpotUserDataEventType::ListenKeyExpired
        | BinanceSpotUserDataEventType::ExternalLockUpdate
        | BinanceSpotUserDataEventType::EventStreamTerminated
        | BinanceSpotUserDataEventType::Unknown => {
            log::debug!("Unhandled user data event type: {event_type:?}");
            None
        }
    }
}

/// Parses the `event_time` from an SBE `ServerShutdownEvent` (template 610) frame.
///
/// The SBE field is microseconds; the trading message variant documents
/// milliseconds (matching the JSON dispatch), so this divides by 1_000.
/// Returns `0` when the buffer is too short to contain the field.
pub(crate) fn parse_server_shutdown_event_time_ms(data: &[u8]) -> i64 {
    if data.len() < message_header_codec::ENCODED_LENGTH + 8 {
        return 0;
    }
    let buf = ReadBuf::new(data);
    buf.get_i64_at(message_header_codec::ENCODED_LENGTH) / 1_000
}

#[cfg(test)]
mod tests {
    use std::time::Duration;

    use futures_util::StreamExt;
    use nautilus_common::testing::wait_until_async;
    use nautilus_network::{
        error::SendError,
        websocket::{AuthTracker, WebSocketConfig},
    };
    use rstest::rstest;

    use super::*;
    use crate::spot::sbe::spot::{
        cancel_order_response_codec::CancelOrderResponseDecoder,
        new_order_full_response_codec::NewOrderFullResponseDecoder,
        self_trade_prevention_mode::SelfTradePreventionMode,
    };

    #[tokio::test]
    async fn test_authentication_with_full_replay_buffer() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let address = listener.local_addr().unwrap();
        let (replayed_tx, replayed_rx) = tokio::sync::oneshot::channel();

        let server = tokio::spawn(async move {
            let (first, _) = listener.accept().await.unwrap();
            let _first = tokio_tungstenite::accept_async(first).await.unwrap();
            let (replacement, _) = listener.accept().await.unwrap();
            let mut replacement = tokio_tungstenite::accept_async(replacement).await.unwrap();
            let auth = replacement.next().await.unwrap().unwrap();
            let auth: serde_json::Value = serde_json::from_str(auth.to_text().unwrap()).unwrap();
            assert_eq!(auth["method"], method::SESSION_LOGON);
            assert_eq!(auth["id"], "ws-1000");
            assert_eq!(auth["params"]["apiKey"], "api-key");
            let subscribe = replacement.next().await.unwrap().unwrap();
            let subscribe: serde_json::Value =
                serde_json::from_str(subscribe.to_text().unwrap()).unwrap();
            assert_eq!(subscribe["method"], "userDataStream.subscribe");
            assert_eq!(subscribe["id"], "ws-1001");
            assert_eq!(subscribe["params"], serde_json::json!({}));
            assert_eq!(
                replacement.next().await.unwrap().unwrap(),
                Message::text("held")
            );
            replayed_tx.send(()).unwrap();
            std::future::pending::<()>().await;
        });

        let tracker = AuthTracker::new();
        let (_cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
        let (_raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
        let (out_tx, _out_rx) = tokio::sync::mpsc::unbounded_channel();

        let mut handler = BinanceSpotWsTradingHandler::new(
            Arc::new(AtomicBool::new(false)),
            cmd_rx,
            raw_rx,
            out_tx,
            Arc::new(SigningCredential::new(
                "api-key".to_string(),
                "secret".to_string(),
            )),
        );
        let config = WebSocketConfig::builder()
            .url(format!("ws://{address}"))
            .writer_capacity(1)
            .reconnect_delay_initial_ms(1)
            .reconnect_delay_max_ms(1)
            .reconnect_jitter_ms(0)
            .build()
            .unwrap();
        let client = WebSocketClient::builder()
            .config(config)
            .message_handler(Arc::new(|_| {}))
            .connect()
            .await
            .unwrap();
        client.set_auth_tracker(tracker.clone(), true);

        // Enqueue and request reconnect without yielding so the writer retains this message
        client.send_text("held".to_string(), None).await.unwrap();
        assert!(client.request_reconnect());
        wait_until_async(
            || async { client.is_active() && client.connection_epoch() == 1 },
            Duration::from_secs(5),
        )
        .await;
        assert!(matches!(
            client.send_text("overflow".to_string(), None).await,
            Err(SendError::BufferFull)
        ));
        handler.inner = Some(client);
        tokio::time::timeout(Duration::from_secs(5), handler.handle_session_logon())
            .await
            .unwrap()
            .unwrap();
        tokio::time::timeout(Duration::from_secs(5), handler.handle_subscribe_user_data())
            .await
            .unwrap()
            .unwrap();
        tracker.succeed();
        tokio::time::timeout(Duration::from_secs(5), replayed_rx)
            .await
            .unwrap()
            .unwrap();
        handler.inner.as_ref().unwrap().disconnect().await;
        server.abort();
    }

    #[rstest]
    #[case::valid(false)]
    #[case::malformed(true)]
    fn test_list_status_routes_typed_message_or_controlled_error(#[case] malformed: bool) {
        let (_cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
        let (_raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
        let (out_tx, _out_rx) = tokio::sync::mpsc::unbounded_channel();

        let mut handler = BinanceSpotWsTradingHandler::new(
            Arc::new(AtomicBool::new(false)),
            cmd_rx,
            raw_rx,
            out_tx,
            Arc::new(SigningCredential::new(
                "api-key".to_owned(),
                "secret".to_owned(),
            )),
        );
        let mut data = crate::common::testing::load_fixture_bytes(
            "spot/user_data_sbe/list_status_executing.sbe",
        );

        if malformed {
            data.pop();
        }

        let message = handler.decode_ws_api_response(&data).unwrap();
        if malformed {
            let BinanceSpotWsTradingMessage::Error(reason) = message else {
                panic!("Expected controlled error");
            };

            assert!(reason.starts_with("SBE ListStatusEvent decode failed:"));
        } else {
            let BinanceSpotWsTradingMessage::ListStatus(status) = message else {
                panic!("Expected typed list status");
            };

            assert_eq!(
                status,
                super::super::decode_sbe::decode_list_status(&data).unwrap()
            );
        }
    }

    #[rstest]
    fn test_cancel_replace_response_decodes_both_orders() {
        let (_cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
        let (_raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
        let (out_tx, _out_rx) = tokio::sync::mpsc::unbounded_channel();
        let mut handler = BinanceSpotWsTradingHandler::new(
            Arc::new(AtomicBool::new(false)),
            cmd_rx,
            raw_rx,
            out_tx,
            Arc::new(SigningCredential::new(
                "api-key".to_string(),
                "secret".to_string(),
            )),
        );
        let placed = include_bytes!(
            "../../../../test_data/spot/user_data_sbe/mainnet/web_socket_response_1.sbe"
        );
        let canceled = include_bytes!(
            "../../../../test_data/spot/user_data_sbe/mainnet/web_socket_response_2.sbe"
        );
        let (request_id, _, replacement) = handler.parse_envelope(placed).unwrap();
        let (_, _, cancellation) = handler.parse_envelope(canceled).unwrap();
        let expected_new = parse::decode_new_order_full(&replacement).unwrap();
        let expected_cancel = parse::decode_cancel_order(&cancellation).unwrap();
        let mut payload = Vec::new();

        for field in [
            2_u16,
            crate::spot::sbe::spot::cancel_replace_order_response_codec::SBE_TEMPLATE_ID,
            crate::spot::sbe::spot::SBE_SCHEMA_ID,
            crate::spot::sbe::spot::SBE_SCHEMA_VERSION,
        ] {
            payload.extend_from_slice(&field.to_le_bytes());
        }
        let success =
            crate::spot::sbe::spot::cancel_replace_status::CancelReplaceStatus::Success as u8;
        payload.extend_from_slice(&[success, success]);
        payload.extend_from_slice(&(cancellation.len() as u16).to_le_bytes());
        payload.extend_from_slice(&cancellation);
        payload.extend_from_slice(&(replacement.len() as u32).to_le_bytes());
        payload.extend_from_slice(&replacement);
        let mut response = placed[..placed.len() - replacement.len() - 4].to_vec();
        response.extend_from_slice(&(payload.len() as u32).to_le_bytes());
        response.extend_from_slice(&payload);
        handler.pending_requests.insert(
            request_id.clone(),
            BinanceSpotWsTradingRequestMeta::CancelReplaceOrder,
        );

        let decoded = handler.decode_ws_api_response(&response).unwrap();

        let BinanceSpotWsTradingMessage::CancelReplaceAccepted {
            request_id: actual_id,
            cancel_response,
            new_order_response,
        } = decoded
        else {
            panic!("Expected cancel-replace acceptance");
        };
        assert_eq!(actual_id, request_id);
        assert_eq!(cancel_response, expected_cancel);
        assert_eq!(new_order_response, expected_new);
        assert!(handler.pending_requests.is_empty());
    }

    #[rstest]
    fn test_mainnet_order_responses_match_generated_decoders() {
        let (_cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
        let (_raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
        let (out_tx, _out_rx) = tokio::sync::mpsc::unbounded_channel();
        let handler = BinanceSpotWsTradingHandler::new(
            Arc::new(AtomicBool::new(false)),
            cmd_rx,
            raw_rx,
            out_tx,
            Arc::new(SigningCredential::new(
                "api-key".to_string(),
                "secret".to_string(),
            )),
        );
        let placed = include_bytes!(
            "../../../../test_data/spot/user_data_sbe/mainnet/web_socket_response_1.sbe"
        );
        let canceled = include_bytes!(
            "../../../../test_data/spot/user_data_sbe/mainnet/web_socket_response_2.sbe"
        );
        let (_, _, replacement) = handler.parse_envelope(placed).unwrap();
        let (_, _, cancellation) = handler.parse_envelope(canceled).unwrap();
        let placed_header = message_header_codec::MessageHeaderDecoder::default()
            .wrap(ReadBuf::new(&replacement), 0);
        let placed_decoder = NewOrderFullResponseDecoder::default().header(placed_header, 0);
        let canceled_header = message_header_codec::MessageHeaderDecoder::default()
            .wrap(ReadBuf::new(&cancellation), 0);
        let canceled_decoder = CancelOrderResponseDecoder::default().header(canceled_header, 0);

        let new_order = parse::decode_new_order_full(&replacement).unwrap();
        let cancel = parse::decode_cancel_order(&cancellation).unwrap();

        assert_eq!(
            new_order.self_trade_prevention_mode,
            placed_decoder.self_trade_prevention_mode()
        );
        assert_eq!(new_order.working_time, placed_decoder.working_time());
        assert_eq!(new_order.stop_price_mantissa, placed_decoder.stop_price());
        assert_eq!(
            cancel.self_trade_prevention_mode,
            canceled_decoder.self_trade_prevention_mode()
        );
        assert_eq!(
            cancel.self_trade_prevention_mode,
            SelfTradePreventionMode::ExpireMaker
        );
    }

    #[rstest]
    #[case::missing(None)]
    #[case::overflow(Some(i64::MAX))]
    fn test_json_error_without_valid_code_is_ambiguous(#[case] code: Option<i64>) {
        let (_cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
        let (_raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
        let (out_tx, mut out_rx) = tokio::sync::mpsc::unbounded_channel();
        let mut handler = BinanceSpotWsTradingHandler::new(
            Arc::new(AtomicBool::new(false)),
            cmd_rx,
            raw_rx,
            out_tx,
            Arc::new(SigningCredential::new(
                "test-key".to_string(),
                "test-secret".to_string(),
            )),
        );
        handler.pending_requests.insert(
            "order-1".to_string(),
            BinanceSpotWsTradingRequestMeta::PlaceOrder,
        );
        let response = serde_json::json!({"id": "order-1", "status": 400, "error": {"code": code, "msg": "invalid request"}});

        handler.handle_text_response(&response.to_string());

        let BinanceSpotWsTradingMessage::RequestFailed { request_id, msg } =
            out_rx.try_recv().unwrap()
        else {
            panic!("expected ambiguous request failure");
        };
        assert_eq!(request_id, "order-1");
        assert_eq!(msg, "Missing or invalid venue error code: invalid request");
        assert!(handler.pending_requests.is_empty());
        assert!(out_rx.try_recv().is_err());
    }

    #[rstest]
    #[case::microseconds_converted_to_ms(1_700_000_000_000_000_i64, 1_700_000_000_000_i64)]
    #[case::zero(0_i64, 0_i64)]
    #[case::negative(-1_000_i64, -1_i64)]
    fn test_parse_server_shutdown_event_time_ms(
        #[case] event_time_us: i64,
        #[case] expected_ms: i64,
    ) {
        let mut buf = vec![0u8; message_header_codec::ENCODED_LENGTH];
        buf.extend_from_slice(&event_time_us.to_le_bytes());
        assert_eq!(parse_server_shutdown_event_time_ms(&buf), expected_ms);
    }

    #[rstest]
    fn test_parse_server_shutdown_event_time_ms_short_buffer_returns_zero() {
        let buf = vec![0u8; message_header_codec::ENCODED_LENGTH + 4];
        assert_eq!(parse_server_shutdown_event_time_ms(&buf), 0);
    }

    #[rstest]
    fn test_classify_user_data_event_server_shutdown_emits_variant() {
        let event = serde_json::json!({"e": "serverShutdown", "E": 1_700_000_000_000_i64});
        let msg = classify_user_data_event(&event).expect("expected ServerShutdown");
        match msg {
            BinanceSpotWsTradingMessage::ServerShutdown { event_time } => {
                assert_eq!(event_time, 1_700_000_000_000);
            }
            other => panic!("expected ServerShutdown variant, was {other:?}"),
        }
    }

    #[rstest]
    fn test_classify_user_data_event_server_shutdown_missing_event_time_defaults_to_zero() {
        let event = serde_json::json!({"e": "serverShutdown"});
        let msg = classify_user_data_event(&event).expect("expected ServerShutdown");
        match msg {
            BinanceSpotWsTradingMessage::ServerShutdown { event_time } => {
                assert_eq!(event_time, 0);
            }
            other => panic!("expected ServerShutdown variant, was {other:?}"),
        }
    }

    #[rstest]
    fn test_classify_user_data_event_unknown_returns_none() {
        let event = serde_json::json!({"e": "somethingElse"});
        assert!(classify_user_data_event(&event).is_none());
    }

    #[rstest]
    fn test_sign_params_includes_recv_window_in_signature() {
        let (_cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
        let (_raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
        let (out_tx, _out_rx) = tokio::sync::mpsc::unbounded_channel();
        let credential = Arc::new(SigningCredential::new(
            "api-key".to_string(),
            "hmac-secret".to_string(),
        ));
        let handler = BinanceSpotWsTradingHandler::new(
            Arc::new(AtomicBool::new(false)),
            cmd_rx,
            raw_rx,
            out_tx,
            credential.clone(),
        )
        .with_recv_window(Some(45_000));

        let signed = handler
            .sign_params(serde_json::json!({"symbol": "BTCUSDT"}))
            .unwrap();
        let mut unsigned = signed.clone();
        let signature = unsigned
            .as_object_mut()
            .unwrap()
            .remove("signature")
            .unwrap();
        let query = canonical_ws_query_string(
            unsigned
                .as_object()
                .unwrap()
                .iter()
                .map(|(key, value)| (key.as_str(), value)),
        )
        .unwrap();

        assert_eq!(signed["recvWindow"], 45_000);
        assert_eq!(signature, credential.sign(&query));
    }
}