nautilus_binance/spot/websocket/trading/
handler.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Binance Spot WebSocket API message handler.
17//!
18//! The handler runs in a dedicated Tokio task as the I/O boundary between the client
19//! orchestrator and the network layer. It exclusively owns the `WebSocketClient` and
20//! processes commands from the client via an unbounded channel.
21//!
22//! ## Responsibilities
23//!
24//! - Command processing: Receives `BinanceSpotWsTradingCommand` from client, serializes to JSON requests.
25//! - Response decoding: Parses SBE binary responses using schema 3 decoders.
26//! - Request correlation: Matches responses to pending requests by ID.
27//! - Message transformation: Emits `BinanceSpotWsTradingMessage` events to client via channel.
28
29use std::{
30    fmt::Debug,
31    sync::{
32        Arc,
33        atomic::{AtomicBool, AtomicU64, Ordering},
34    },
35};
36
37use ahash::AHashMap;
38use nautilus_network::{RECONNECTED, websocket::WebSocketClient};
39use tokio_tungstenite::tungstenite::Message;
40
41use super::{
42    client::BINANCE_WS_RATE_LIMIT_KEY_ORDER,
43    error::{BinanceWsApiError, BinanceWsApiResult},
44    messages::{
45        BinanceSpotWsTradingCommand, BinanceSpotWsTradingMessage, BinanceSpotWsTradingRequest,
46        BinanceSpotWsTradingRequestMeta, method,
47    },
48};
49use crate::{
50    common::credential::SigningCredential,
51    spot::{
52        enums::BinanceSpotUserDataEventType,
53        http::{models::BinanceCancelOrderResponse, parse},
54        sbe::spot::{
55            ReadBuf,
56            error_response_codec::ErrorResponseDecoder,
57            message_header_codec,
58            web_socket_response_codec::{SBE_TEMPLATE_ID, WebSocketResponseDecoder},
59        },
60    },
61};
62
63/// Binance Spot WebSocket API handler.
64///
65/// Runs in a dedicated Tokio task, processing commands from the client
66/// and transforming raw WebSocket messages into Nautilus domain events.
67/// Messages are sent to the client via the output channel.
68pub struct BinanceSpotWsTradingHandler {
69    signal: Arc<AtomicBool>,
70    inner: Option<WebSocketClient>,
71    cmd_rx: tokio::sync::mpsc::UnboundedReceiver<BinanceSpotWsTradingCommand>,
72    raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
73    out_tx: tokio::sync::mpsc::UnboundedSender<BinanceSpotWsTradingMessage>,
74    credential: Arc<SigningCredential>,
75    pending_requests: AHashMap<String, BinanceSpotWsTradingRequestMeta>,
76    request_id_counter: AtomicU64,
77}
78
79impl Debug for BinanceSpotWsTradingHandler {
80    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
81        f.debug_struct(stringify!(BinanceSpotWsTradingHandler))
82            .field("inner", &self.inner.as_ref().map(|_| "<client>"))
83            .field(
84                "pending_requests",
85                &format!("{} pending", self.pending_requests.len()),
86            )
87            .finish_non_exhaustive()
88    }
89}
90
91impl BinanceSpotWsTradingHandler {
92    /// Creates a new handler instance.
93    #[must_use]
94    pub fn new(
95        signal: Arc<AtomicBool>,
96        cmd_rx: tokio::sync::mpsc::UnboundedReceiver<BinanceSpotWsTradingCommand>,
97        raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
98        out_tx: tokio::sync::mpsc::UnboundedSender<BinanceSpotWsTradingMessage>,
99        credential: Arc<SigningCredential>,
100    ) -> Self {
101        Self {
102            signal,
103            inner: None,
104            cmd_rx,
105            raw_rx,
106            out_tx,
107            credential,
108            pending_requests: AHashMap::new(),
109            request_id_counter: AtomicU64::new(1000),
110        }
111    }
112
113    /// Runs the main event loop for commands and raw messages.
114    ///
115    /// Sends output messages via `out_tx` channel. Returns `false` when disconnected
116    /// or the signal is set, indicating the handler should exit.
117    pub async fn run(&mut self) -> bool {
118        loop {
119            if self.signal.load(Ordering::Relaxed) {
120                return false;
121            }
122
123            tokio::select! {
124                Some(cmd) = self.cmd_rx.recv() => {
125                    match cmd {
126                        BinanceSpotWsTradingCommand::SetClient(client) => {
127                            log::debug!("Handler received WebSocket client");
128                            self.inner = Some(client);
129                            self.emit(BinanceSpotWsTradingMessage::Connected);
130                        }
131                        BinanceSpotWsTradingCommand::Disconnect => {
132                            log::debug!("Handler disconnecting WebSocket client");
133                            self.inner = None;
134                            return false;
135                        }
136                        BinanceSpotWsTradingCommand::PlaceOrder { id, params } => {
137                            if let Err(e) = self.handle_place_order(id.clone(), params).await {
138                                log::error!("Failed to handle place order command: {e}");
139                                self.pending_requests.remove(&id);
140                                self.emit(BinanceSpotWsTradingMessage::RequestFailed {
141                                    request_id: id,
142                                    msg: e.to_string(),
143                                });
144                            }
145                        }
146                        BinanceSpotWsTradingCommand::CancelOrder { id, params } => {
147                            if let Err(e) = self.handle_cancel_order(id.clone(), params).await {
148                                log::error!("Failed to handle cancel order command: {e}");
149                                self.pending_requests.remove(&id);
150                                self.emit(BinanceSpotWsTradingMessage::RequestFailed {
151                                    request_id: id,
152                                    msg: e.to_string(),
153                                });
154                            }
155                        }
156                        BinanceSpotWsTradingCommand::CancelReplaceOrder { id, params } => {
157                            if let Err(e) = self.handle_cancel_replace_order(id.clone(), params).await {
158                                log::error!("Failed to handle cancel replace command: {e}");
159                                self.pending_requests.remove(&id);
160                                self.emit(BinanceSpotWsTradingMessage::RequestFailed {
161                                    request_id: id,
162                                    msg: e.to_string(),
163                                });
164                            }
165                        }
166                        BinanceSpotWsTradingCommand::CancelAllOrders { id, symbol } => {
167                            if let Err(e) = self.handle_cancel_all_orders(id.clone(), symbol).await {
168                                log::error!("Failed to handle cancel all command: {e}");
169                                self.pending_requests.remove(&id);
170                                self.emit(BinanceSpotWsTradingMessage::RequestFailed {
171                                    request_id: id,
172                                    msg: e.to_string(),
173                                });
174                            }
175                        }
176                        BinanceSpotWsTradingCommand::SessionLogon => {
177                            if let Err(e) = self.handle_session_logon().await {
178                                log::error!("Session logon failed: {e}");
179                                self.emit(BinanceSpotWsTradingMessage::Error(
180                                    format!("Session logon failed: {e}"),
181                                ));
182                            }
183                        }
184                        BinanceSpotWsTradingCommand::SubscribeUserData => {
185                            if let Err(e) = self.handle_subscribe_user_data().await {
186                                log::error!("User data subscribe failed: {e}");
187                                self.emit(BinanceSpotWsTradingMessage::Error(
188                                    format!("User data subscribe failed: {e}"),
189                                ));
190                            }
191                        }
192                    }
193                }
194                Some(msg) = self.raw_rx.recv() => {
195                    if let Message::Text(ref text) = msg
196                        && text.as_str() == RECONNECTED
197                    {
198                        log::debug!("Handler received reconnection signal");
199
200                        // Fail any pending requests - they won't get responses on new connection
201                        self.fail_pending_requests();
202
203                        self.emit(BinanceSpotWsTradingMessage::Reconnected);
204                        continue;
205                    }
206
207                    self.handle_message(msg);
208                }
209                else => {
210                    // Both channels closed
211                    return false;
212                }
213            }
214        }
215    }
216
217    /// Sends a message to the output channel.
218    fn emit(&self, msg: BinanceSpotWsTradingMessage) {
219        if let Err(e) = self.out_tx.send(msg) {
220            log::error!("Failed to send message to output channel: {e}");
221        }
222    }
223
224    /// Fails all pending requests after a reconnection.
225    fn fail_pending_requests(&mut self) {
226        if self.pending_requests.is_empty() {
227            return;
228        }
229
230        let count = self.pending_requests.len();
231        log::warn!("Failing {count} pending requests after reconnection");
232
233        let pending = std::mem::take(&mut self.pending_requests);
234        for (request_id, _meta) in pending {
235            self.emit(BinanceSpotWsTradingMessage::RequestFailed {
236                request_id,
237                msg: "Connection lost before response received".to_string(),
238            });
239        }
240    }
241
242    async fn handle_place_order(
243        &mut self,
244        id: String,
245        params: crate::spot::http::query::NewOrderParams,
246    ) -> BinanceWsApiResult<()> {
247        let params_json = serde_json::to_value(&params)
248            .map_err(|e| BinanceWsApiError::ClientError(e.to_string()))?;
249        let signed_params = self.sign_params(params_json)?;
250
251        let request = BinanceSpotWsTradingRequest::new(&id, method::ORDER_PLACE, signed_params);
252        self.pending_requests
253            .insert(id.clone(), BinanceSpotWsTradingRequestMeta::PlaceOrder);
254        self.send_request(request).await
255    }
256
257    async fn handle_cancel_order(
258        &mut self,
259        id: String,
260        params: crate::spot::http::query::CancelOrderParams,
261    ) -> BinanceWsApiResult<()> {
262        let params_json = serde_json::to_value(&params)
263            .map_err(|e| BinanceWsApiError::ClientError(e.to_string()))?;
264        let signed_params = self.sign_params(params_json)?;
265
266        let request = BinanceSpotWsTradingRequest::new(&id, method::ORDER_CANCEL, signed_params);
267        self.pending_requests
268            .insert(id.clone(), BinanceSpotWsTradingRequestMeta::CancelOrder);
269        self.send_request(request).await
270    }
271
272    async fn handle_cancel_replace_order(
273        &mut self,
274        id: String,
275        params: crate::spot::http::query::CancelReplaceOrderParams,
276    ) -> BinanceWsApiResult<()> {
277        let params_json = serde_json::to_value(&params)
278            .map_err(|e| BinanceWsApiError::ClientError(e.to_string()))?;
279        let signed_params = self.sign_params(params_json)?;
280
281        let request =
282            BinanceSpotWsTradingRequest::new(&id, method::ORDER_CANCEL_REPLACE, signed_params);
283        self.pending_requests.insert(
284            id.clone(),
285            BinanceSpotWsTradingRequestMeta::CancelReplaceOrder,
286        );
287        self.send_request(request).await
288    }
289
290    async fn handle_cancel_all_orders(
291        &mut self,
292        id: String,
293        symbol: String,
294    ) -> BinanceWsApiResult<()> {
295        let params_json = serde_json::json!({ "symbol": symbol });
296        let signed_params = self.sign_params(params_json)?;
297
298        let request =
299            BinanceSpotWsTradingRequest::new(&id, method::OPEN_ORDERS_CANCEL_ALL, signed_params);
300        self.pending_requests
301            .insert(id.clone(), BinanceSpotWsTradingRequestMeta::CancelAllOrders);
302        self.send_request(request).await
303    }
304
305    async fn handle_session_logon(&mut self) -> BinanceWsApiResult<()> {
306        let id = self.next_request_id();
307        let params_json = serde_json::json!({});
308        let signed_params = self.sign_params(params_json)?;
309
310        let request = BinanceSpotWsTradingRequest::new(&id, "session.logon", signed_params);
311        self.pending_requests
312            .insert(id, BinanceSpotWsTradingRequestMeta::SessionLogon);
313        self.send_request(request).await
314    }
315
316    async fn handle_subscribe_user_data(&mut self) -> BinanceWsApiResult<()> {
317        let id = self.next_request_id();
318        let request = BinanceSpotWsTradingRequest::new(
319            &id,
320            "userDataStream.subscribe",
321            serde_json::json!({}),
322        );
323        self.pending_requests
324            .insert(id, BinanceSpotWsTradingRequestMeta::SubscribeUserData);
325        self.send_request(request).await
326    }
327
328    fn next_request_id(&self) -> String {
329        let id = self.request_id_counter.fetch_add(1, Ordering::Relaxed);
330        format!("ws-{id}")
331    }
332
333    fn sign_params(&self, mut params: serde_json::Value) -> BinanceWsApiResult<serde_json::Value> {
334        let timestamp = std::time::SystemTime::now()
335            .duration_since(std::time::UNIX_EPOCH)
336            .map_err(|e| BinanceWsApiError::ClientError(e.to_string()))?
337            .as_millis() as i64;
338
339        if let Some(obj) = params.as_object_mut() {
340            obj.insert("timestamp".to_string(), serde_json::json!(timestamp));
341            obj.insert(
342                "apiKey".to_string(),
343                serde_json::json!(self.credential.api_key()),
344            );
345        }
346
347        let query_string = serde_urlencoded::to_string(&params)
348            .map_err(|e| BinanceWsApiError::ClientError(e.to_string()))?;
349        let signature = self.credential.sign(&query_string);
350
351        if let Some(obj) = params.as_object_mut() {
352            obj.insert("signature".to_string(), serde_json::json!(signature));
353        }
354
355        Ok(params)
356    }
357
358    async fn send_request(
359        &mut self,
360        request: BinanceSpotWsTradingRequest,
361    ) -> BinanceWsApiResult<()> {
362        let client = self.inner.as_mut().ok_or_else(|| {
363            BinanceWsApiError::ConnectionError("WebSocket not connected".to_string())
364        })?;
365
366        let json = serde_json::to_string(&request)
367            .map_err(|e| BinanceWsApiError::ClientError(e.to_string()))?;
368
369        log::debug!(
370            "Sending WebSocket API request id={} method={}",
371            request.id,
372            request.method
373        );
374
375        // Apply rate limiting for order operations
376        client
377            .send_text(json, Some(BINANCE_WS_RATE_LIMIT_KEY_ORDER.as_slice()))
378            .await
379            .map_err(|e| {
380                BinanceWsApiError::ConnectionError(format!("Failed to send request: {e}"))
381            })?;
382
383        Ok(())
384    }
385
386    fn handle_message(&mut self, msg: Message) {
387        match msg {
388            Message::Binary(data) => self.handle_binary_response(&data),
389            Message::Text(text) => self.handle_text_response(&text),
390            Message::Ping(_) | Message::Pong(_) => {}
391            Message::Close(frame) => {
392                log::debug!("WebSocket closed: {frame:?}");
393            }
394            Message::Frame(_) => {}
395        }
396    }
397
398    fn handle_binary_response(&mut self, data: &[u8]) {
399        match self.decode_ws_api_response(data) {
400            Ok(response) => self.emit(response),
401            Err(e) => {
402                log::error!("Failed to decode WebSocket API response: {e}");
403                self.emit(BinanceSpotWsTradingMessage::Error(e.to_string()));
404            }
405        }
406    }
407
408    fn handle_text_response(&mut self, text: &str) {
409        let json: serde_json::Value = match serde_json::from_str(text) {
410            Ok(j) => j,
411            Err(e) => {
412                log::warn!("Failed to parse text response as JSON: {e}");
413                return;
414            }
415        };
416
417        // User data events arrive wrapped: {"subscriptionId": N, "event": {...}}
418        if let Some(event) = json.get("event") {
419            self.handle_user_data_event(event);
420            return;
421        }
422
423        // WS API responses have an "id" field for request correlation
424        if let Some(id) = json.get("id") {
425            let id_str = match id {
426                serde_json::Value::String(s) => s.clone(),
427                serde_json::Value::Number(n) => n.to_string(),
428                _ => return,
429            };
430
431            if let Some(meta) = self.pending_requests.remove(&id_str) {
432                // Check for error: nested {"error": {"code": N, "msg": "..."}}
433                // or top-level {"code": N, "msg": "..."}
434                let error_info = json
435                    .get("error")
436                    .map(|e| {
437                        (
438                            e.get("code").and_then(|v| v.as_i64()).unwrap_or(-1),
439                            e.get("msg")
440                                .and_then(|v| v.as_str())
441                                .unwrap_or("Unknown error")
442                                .to_string(),
443                        )
444                    })
445                    .or_else(|| {
446                        json.get("code").and_then(|c| c.as_i64()).map(|code| {
447                            let msg = json
448                                .get("msg")
449                                .and_then(|v| v.as_str())
450                                .unwrap_or("Unknown error")
451                                .to_string();
452                            (code, msg)
453                        })
454                    });
455
456                if let Some((code, msg)) = error_info {
457                    let rejection = self.create_rejection(id_str, code as i32, msg, meta);
458                    self.emit(rejection);
459                    return;
460                }
461
462                // Success response
463                match meta {
464                    BinanceSpotWsTradingRequestMeta::SessionLogon => {
465                        log::debug!("Session authenticated");
466                        self.emit(BinanceSpotWsTradingMessage::Authenticated);
467                    }
468                    BinanceSpotWsTradingRequestMeta::SubscribeUserData => {
469                        let subscription_id = json
470                            .get("result")
471                            .and_then(|r| r.get("subscriptionId"))
472                            .map(|v| v.to_string())
473                            .unwrap_or_default();
474                        log::debug!("User data stream subscribed: id={subscription_id}");
475                        self.emit(BinanceSpotWsTradingMessage::UserDataSubscribed {
476                            subscription_id,
477                        });
478                    }
479                    _ => {
480                        // Order operation responses come as SBE binary, not JSON text.
481                        // If we get a JSON success for an order operation, log it.
482                        log::debug!("Unexpected JSON success for request {id_str}: {json}");
483                    }
484                }
485                return;
486            }
487
488            // Error response without matching pending request
489            if let Some(code) = json.get("code").and_then(|v| v.as_i64()) {
490                let msg = json
491                    .get("msg")
492                    .and_then(|v| v.as_str())
493                    .unwrap_or("Unknown error");
494                log::warn!(
495                    "Received error response without matching request ID: code={code} msg={msg}"
496                );
497            }
498            return;
499        }
500
501        // Stream termination event
502        if json.get("eventStreamTerminated").is_some() {
503            log::warn!("User data stream terminated, resubscribe needed");
504            return;
505        }
506
507        log::debug!("Unhandled text message: {text}");
508    }
509
510    fn handle_user_data_event(&self, event: &serde_json::Value) {
511        if let Some(msg) = classify_user_data_event(event) {
512            self.emit(msg);
513        }
514    }
515
516    fn decode_ws_api_response(
517        &mut self,
518        data: &[u8],
519    ) -> Result<BinanceSpotWsTradingMessage, BinanceWsApiError> {
520        // Check template ID before parsing
521        if data.len() >= message_header_codec::ENCODED_LENGTH {
522            let buf = ReadBuf::new(data);
523            let template_id = buf.get_u16_at(2);
524
525            // User data stream events arrive as SBE with their own template IDs
526            // (not wrapped in WebSocketResponse template 50).
527            match template_id {
528                601 => {
529                    log::debug!("Received SBE BalanceUpdateEvent ({} bytes)", data.len());
530                    match super::decode_sbe::decode_balance_update(data) {
531                        Ok(msg) => {
532                            log::debug!(
533                                "SBE balance update: asset={}, delta={}",
534                                msg.asset,
535                                msg.delta
536                            );
537                            return Ok(BinanceSpotWsTradingMessage::BalanceUpdate(msg));
538                        }
539                        Err(e) => {
540                            log::error!("Failed to decode SBE BalanceUpdateEvent: {e}");
541                            return Ok(BinanceSpotWsTradingMessage::Error(format!(
542                                "SBE BalanceUpdateEvent decode failed: {e}"
543                            )));
544                        }
545                    }
546                }
547                603 => {
548                    log::debug!("Received SBE ExecutionReportEvent ({} bytes)", data.len());
549                    match super::decode_sbe::decode_execution_report(data) {
550                        Ok(report) => {
551                            log::debug!(
552                                "SBE execution report: symbol={}, order_id={}, exec={:?}, status={:?}",
553                                report.symbol,
554                                report.order_id,
555                                report.execution_type,
556                                report.order_status
557                            );
558                            return Ok(BinanceSpotWsTradingMessage::ExecutionReport(Box::new(
559                                report,
560                            )));
561                        }
562                        Err(e) => {
563                            log::error!("Failed to decode SBE ExecutionReportEvent: {e}");
564                            return Ok(BinanceSpotWsTradingMessage::Error(format!(
565                                "SBE ExecutionReportEvent decode failed: {e}"
566                            )));
567                        }
568                    }
569                }
570                606 => {
571                    log::debug!(
572                        "Received SBE ListStatusEvent ({} bytes), not yet decoded",
573                        data.len()
574                    );
575                    return Ok(BinanceSpotWsTradingMessage::Error(
576                        "SBE ListStatusEvent decoding not yet implemented".to_string(),
577                    ));
578                }
579                607 => {
580                    log::debug!(
581                        "Received SBE OutboundAccountPositionEvent ({} bytes)",
582                        data.len()
583                    );
584
585                    match super::decode_sbe::decode_account_position(data) {
586                        Ok(msg) => {
587                            log::debug!("SBE account position: {} balance(s)", msg.balances.len());
588                            return Ok(BinanceSpotWsTradingMessage::AccountPosition(msg));
589                        }
590                        Err(e) => {
591                            log::error!("Failed to decode SBE OutboundAccountPositionEvent: {e}");
592                            return Ok(BinanceSpotWsTradingMessage::Error(format!(
593                                "SBE OutboundAccountPositionEvent decode failed: {e}"
594                            )));
595                        }
596                    }
597                }
598                610 => {
599                    let event_time = parse_server_shutdown_event_time_ms(data);
600                    log::warn!(
601                        "Binance server shutdown notice (SBE, event_time={event_time}); disconnect expected within ~10 minutes",
602                    );
603                    return Ok(BinanceSpotWsTradingMessage::ServerShutdown { event_time });
604                }
605                _ => {} // Fall through to WebSocketResponse parsing
606            }
607        }
608
609        // Standard WebSocketResponse envelope (template 50)
610        let (request_id, status, result_data) = self.parse_envelope(data)?;
611
612        // Look up the pending request by ID
613        let meta = self.pending_requests.remove(&request_id).ok_or_else(|| {
614            BinanceWsApiError::UnknownRequestId(format!("No pending request for ID: {request_id}"))
615        })?;
616
617        // Check for error status (non-200)
618        if status != 200 {
619            let (code, msg) = Self::try_decode_sbe_error(&result_data).unwrap_or((
620                status as i32,
621                format!("Request failed with status {status}"),
622            ));
623            return Ok(self.create_rejection(request_id, code, msg, meta));
624        }
625
626        // Decode the inner payload based on request type
627        match meta {
628            BinanceSpotWsTradingRequestMeta::PlaceOrder => {
629                let response = parse::decode_new_order_full(&result_data)?;
630                Ok(BinanceSpotWsTradingMessage::OrderAccepted {
631                    request_id,
632                    response,
633                })
634            }
635            BinanceSpotWsTradingRequestMeta::CancelOrder => {
636                let response = parse::decode_cancel_order(&result_data)?;
637                Ok(BinanceSpotWsTradingMessage::OrderCanceled {
638                    request_id,
639                    response,
640                })
641            }
642            BinanceSpotWsTradingRequestMeta::CancelReplaceOrder => {
643                // Cancel-replace returns both cancel and new order info
644                let new_order_response = parse::decode_new_order_full(&result_data)?;
645                let cancel_response = BinanceCancelOrderResponse {
646                    price_exponent: new_order_response.price_exponent,
647                    qty_exponent: new_order_response.qty_exponent,
648                    order_id: 0,
649                    order_list_id: None,
650                    transact_time: new_order_response.transact_time,
651                    price_mantissa: 0,
652                    orig_qty_mantissa: 0,
653                    executed_qty_mantissa: 0,
654                    cummulative_quote_qty_mantissa: 0,
655                    status: crate::spot::sbe::spot::order_status::OrderStatus::Canceled,
656                    time_in_force: new_order_response.time_in_force,
657                    order_type: new_order_response.order_type,
658                    side: new_order_response.side,
659                    self_trade_prevention_mode: new_order_response.self_trade_prevention_mode,
660                    client_order_id: String::new(),
661                    orig_client_order_id: String::new(),
662                    symbol: new_order_response.symbol.clone(),
663                };
664                Ok(BinanceSpotWsTradingMessage::CancelReplaceAccepted {
665                    request_id,
666                    cancel_response,
667                    new_order_response,
668                })
669            }
670            BinanceSpotWsTradingRequestMeta::CancelAllOrders => {
671                let responses = parse::decode_cancel_open_orders(&result_data)?;
672                Ok(BinanceSpotWsTradingMessage::AllOrdersCanceled {
673                    request_id,
674                    responses,
675                })
676            }
677            BinanceSpotWsTradingRequestMeta::SessionLogon => {
678                log::debug!("Session authenticated (SBE response)");
679                Ok(BinanceSpotWsTradingMessage::Authenticated)
680            }
681            BinanceSpotWsTradingRequestMeta::SubscribeUserData => {
682                log::debug!("User data stream subscribed (SBE response)");
683                Ok(BinanceSpotWsTradingMessage::UserDataSubscribed {
684                    subscription_id: request_id,
685                })
686            }
687        }
688    }
689
690    /// Parses the WebSocketResponse SBE envelope.
691    ///
692    /// Returns (request_id, status, result_payload).
693    fn parse_envelope(&self, data: &[u8]) -> Result<(String, u16, Vec<u8>), BinanceWsApiError> {
694        if data.len() < message_header_codec::ENCODED_LENGTH {
695            return Err(BinanceWsApiError::DecodeError(
696                crate::spot::sbe::error::SbeDecodeError::BufferTooShort {
697                    expected: message_header_codec::ENCODED_LENGTH,
698                    actual: data.len(),
699                },
700            ));
701        }
702
703        let buf = ReadBuf::new(data);
704
705        // Parse message header
706        let block_length = buf.get_u16_at(0);
707        let template_id = buf.get_u16_at(2);
708
709        if template_id != SBE_TEMPLATE_ID {
710            return Err(BinanceWsApiError::DecodeError(
711                crate::spot::sbe::error::SbeDecodeError::UnknownTemplateId(template_id),
712            ));
713        }
714
715        let version = buf.get_u16_at(6);
716
717        // Create decoder at offset after message header
718        let decoder = WebSocketResponseDecoder::default().wrap(
719            buf,
720            message_header_codec::ENCODED_LENGTH,
721            block_length,
722            version,
723        );
724
725        // Read status from fixed block (offset 1 within block)
726        let status = decoder.status();
727
728        // Skip rate_limits group
729        let mut rate_limits = decoder.rate_limits_decoder();
730        while rate_limits.advance().unwrap_or(None).is_some() {}
731        let mut decoder = rate_limits.parent().map_err(|e| {
732            BinanceWsApiError::ClientError(format!("Failed to get parent from rate_limits: {e}"))
733        })?;
734
735        // Extract request ID
736        let id_coords = decoder.id_decoder();
737        let id_bytes = decoder.id_slice(id_coords);
738        let request_id = String::from_utf8_lossy(id_bytes).to_string();
739
740        // Extract result payload - copy to owned Vec to avoid lifetime issues
741        let result_coords = decoder.result_decoder();
742        let result_data = decoder.result_slice(result_coords).to_vec();
743
744        Ok((request_id, status, result_data))
745    }
746
747    fn create_rejection(
748        &self,
749        request_id: String,
750        code: i32,
751        msg: String,
752        meta: BinanceSpotWsTradingRequestMeta,
753    ) -> BinanceSpotWsTradingMessage {
754        match meta {
755            BinanceSpotWsTradingRequestMeta::PlaceOrder => {
756                BinanceSpotWsTradingMessage::OrderRejected {
757                    request_id,
758                    code,
759                    msg,
760                }
761            }
762            BinanceSpotWsTradingRequestMeta::CancelOrder => {
763                BinanceSpotWsTradingMessage::CancelRejected {
764                    request_id,
765                    code,
766                    msg,
767                }
768            }
769            BinanceSpotWsTradingRequestMeta::CancelReplaceOrder => {
770                BinanceSpotWsTradingMessage::CancelReplaceRejected {
771                    request_id,
772                    code,
773                    msg,
774                }
775            }
776            BinanceSpotWsTradingRequestMeta::CancelAllOrders => {
777                BinanceSpotWsTradingMessage::CancelRejected {
778                    request_id,
779                    code,
780                    msg,
781                }
782            }
783            BinanceSpotWsTradingRequestMeta::SessionLogon
784            | BinanceSpotWsTradingRequestMeta::SubscribeUserData => {
785                BinanceSpotWsTradingMessage::Error(format!("code={code}: {msg}"))
786            }
787        }
788    }
789
790    // Decodes the SBE error response to extract the Binance error code and message
791    fn try_decode_sbe_error(data: &[u8]) -> Option<(i32, String)> {
792        const HEADER_LEN: usize = 8;
793
794        if data.len()
795            < HEADER_LEN + crate::spot::sbe::spot::error_response_codec::SBE_BLOCK_LENGTH as usize
796        {
797            return None;
798        }
799
800        let buf = ReadBuf::new(data);
801        let header = message_header_codec::MessageHeaderDecoder::default().wrap(buf, 0);
802        if header.template_id() != crate::spot::sbe::spot::error_response_codec::SBE_TEMPLATE_ID {
803            return None;
804        }
805
806        let mut decoder = ErrorResponseDecoder::default().header(header, 0);
807        let code = i32::from(decoder.code());
808        let msg_coords = decoder.msg_decoder();
809        let msg_bytes = decoder.msg_slice(msg_coords);
810        let msg = String::from_utf8_lossy(msg_bytes).into_owned();
811
812        Some((code, msg))
813    }
814}
815
816/// Classifies a JSON user-data event into a trading message, if any.
817///
818/// Returns `None` when the event type is unknown or the payload fails to
819/// deserialize; in that case the caller logs and drops the event.
820pub(crate) fn classify_user_data_event(
821    event: &serde_json::Value,
822) -> Option<BinanceSpotWsTradingMessage> {
823    let event_type = event
824        .get("e")
825        .and_then(|v| serde_json::from_value::<BinanceSpotUserDataEventType>(v.clone()).ok())
826        .unwrap_or(BinanceSpotUserDataEventType::Unknown);
827
828    match event_type {
829        BinanceSpotUserDataEventType::ExecutionReport => {
830            match serde_json::from_value::<super::user_data::BinanceSpotExecutionReport>(
831                event.clone(),
832            ) {
833                Ok(report) => {
834                    log::debug!(
835                        "Execution report: symbol={}, order_id={}, exec={:?}, status={:?}",
836                        report.symbol,
837                        report.order_id,
838                        report.execution_type,
839                        report.order_status
840                    );
841                    Some(BinanceSpotWsTradingMessage::ExecutionReport(Box::new(
842                        report,
843                    )))
844                }
845                Err(e) => {
846                    log::warn!("Failed to parse execution report: {e}");
847                    None
848                }
849            }
850        }
851        BinanceSpotUserDataEventType::OutboundAccountPosition => {
852            match serde_json::from_value::<super::user_data::BinanceSpotAccountPositionMsg>(
853                event.clone(),
854            ) {
855                Ok(msg) => {
856                    log::debug!("Account position update: {} balance(s)", msg.balances.len());
857                    Some(BinanceSpotWsTradingMessage::AccountPosition(msg))
858                }
859                Err(e) => {
860                    log::warn!("Failed to parse account position: {e}");
861                    None
862                }
863            }
864        }
865        BinanceSpotUserDataEventType::BalanceUpdate => {
866            match serde_json::from_value::<super::user_data::BinanceSpotBalanceUpdateMsg>(
867                event.clone(),
868            ) {
869                Ok(msg) => {
870                    log::debug!("Balance update: asset={}, delta={}", msg.asset, msg.delta);
871                    Some(BinanceSpotWsTradingMessage::BalanceUpdate(msg))
872                }
873                Err(e) => {
874                    log::warn!("Failed to parse balance update: {e}");
875                    None
876                }
877            }
878        }
879        BinanceSpotUserDataEventType::ServerShutdown => {
880            let event_time = event.get("E").and_then(|v| v.as_i64()).unwrap_or_default();
881            log::warn!(
882                "Binance server shutdown notice (event_time={event_time}); disconnect expected within ~10 minutes",
883            );
884            Some(BinanceSpotWsTradingMessage::ServerShutdown { event_time })
885        }
886        BinanceSpotUserDataEventType::ListenKeyExpired
887        | BinanceSpotUserDataEventType::ExternalLockUpdate
888        | BinanceSpotUserDataEventType::EventStreamTerminated
889        | BinanceSpotUserDataEventType::Unknown => {
890            log::debug!("Unhandled user data event type: {event_type:?}");
891            None
892        }
893    }
894}
895
896/// Parses the `event_time` from an SBE `ServerShutdownEvent` (template 610) frame.
897///
898/// The SBE field is microseconds; the trading message variant documents
899/// milliseconds (matching the JSON dispatch), so this divides by 1_000.
900/// Returns `0` when the buffer is too short to contain the field.
901pub(crate) fn parse_server_shutdown_event_time_ms(data: &[u8]) -> i64 {
902    if data.len() < message_header_codec::ENCODED_LENGTH + 8 {
903        return 0;
904    }
905    let buf = ReadBuf::new(data);
906    buf.get_i64_at(message_header_codec::ENCODED_LENGTH) / 1_000
907}
908
909#[cfg(test)]
910mod tests {
911    use rstest::rstest;
912
913    use super::*;
914
915    #[rstest]
916    #[case::microseconds_converted_to_ms(1_700_000_000_000_000_i64, 1_700_000_000_000_i64)]
917    #[case::zero(0_i64, 0_i64)]
918    #[case::negative(-1_000_i64, -1_i64)]
919    fn test_parse_server_shutdown_event_time_ms(
920        #[case] event_time_us: i64,
921        #[case] expected_ms: i64,
922    ) {
923        let mut buf = vec![0u8; message_header_codec::ENCODED_LENGTH];
924        buf.extend_from_slice(&event_time_us.to_le_bytes());
925        assert_eq!(parse_server_shutdown_event_time_ms(&buf), expected_ms);
926    }
927
928    #[rstest]
929    fn test_parse_server_shutdown_event_time_ms_short_buffer_returns_zero() {
930        let buf = vec![0u8; message_header_codec::ENCODED_LENGTH + 4];
931        assert_eq!(parse_server_shutdown_event_time_ms(&buf), 0);
932    }
933
934    #[rstest]
935    fn test_classify_user_data_event_server_shutdown_emits_variant() {
936        let event = serde_json::json!({"e": "serverShutdown", "E": 1_700_000_000_000_i64});
937        let msg = classify_user_data_event(&event).expect("expected ServerShutdown");
938        match msg {
939            BinanceSpotWsTradingMessage::ServerShutdown { event_time } => {
940                assert_eq!(event_time, 1_700_000_000_000);
941            }
942            other => panic!("expected ServerShutdown variant, was {other:?}"),
943        }
944    }
945
946    #[rstest]
947    fn test_classify_user_data_event_server_shutdown_missing_event_time_defaults_to_zero() {
948        let event = serde_json::json!({"e": "serverShutdown"});
949        let msg = classify_user_data_event(&event).expect("expected ServerShutdown");
950        match msg {
951            BinanceSpotWsTradingMessage::ServerShutdown { event_time } => {
952                assert_eq!(event_time, 0);
953            }
954            other => panic!("expected ServerShutdown variant, was {other:?}"),
955        }
956    }
957
958    #[rstest]
959    fn test_classify_user_data_event_unknown_returns_none() {
960        let event = serde_json::json!({"e": "somethingElse"});
961        assert!(classify_user_data_event(&event).is_none());
962    }
963}
nautilus_binance/spot/websocket/trading/handler.rs

nautilus_binance/spot/websocket/trading/
handler.rs
