nautilus_kraken/websocket/spot_v2/
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//! WebSocket message handler for Kraken Spot v2.
17
18use std::{
19    collections::VecDeque,
20    sync::{
21        Arc,
22        atomic::{AtomicBool, Ordering},
23    },
24};
25
26#[cfg(test)]
27use nautilus_core::string::secret::REDACTED;
28use nautilus_core::string::secret::SecretString;
29use nautilus_network::{
30    RECONNECTED,
31    websocket::{SubscriptionState, WebSocketClient},
32};
33use serde::Deserialize;
34use serde_json::{Value, value::RawValue};
35use tokio_tungstenite::tungstenite::Message;
36
37use super::{
38    enums::{KrakenWsChannel, KrakenWsMessageType},
39    messages::{
40        KrakenSpotWsMessage, KrakenWsBookData, KrakenWsExecutionData, KrakenWsOhlcData,
41        KrakenWsRawMessage, KrakenWsResponse, KrakenWsTickerData, KrakenWsTradeData,
42    },
43    parse::parse_order_response,
44};
45use crate::{
46    common::consts::{KRAKEN_RATE_LIMIT_KEY_ORDER, KRAKEN_RATE_LIMIT_KEY_SUBSCRIPTION},
47    websocket::spot_v2::level_3::messages::{KrakenL3Snapshot, KrakenL3UpdateData},
48};
49
50/// Commands sent from the outer client to the inner message handler.
51#[derive(Debug)]
52pub enum SpotHandlerCommand {
53    SetClient(WebSocketClient),
54    Disconnect,
55    Subscribe { payload: SecretString },
56    Unsubscribe { payload: SecretString },
57    Ping { payload: SecretString },
58    SendOrderRequest { req_id: u64, payload: SecretString },
59}
60
61/// WebSocket message handler for Kraken Spot v2.
62pub(super) struct SpotFeedHandler {
63    signal: Arc<AtomicBool>,
64    inner: Option<WebSocketClient>,
65    cmd_rx: tokio::sync::mpsc::UnboundedReceiver<SpotHandlerCommand>,
66    raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
67    subscriptions: SubscriptionState,
68    pending_messages: VecDeque<KrakenSpotWsMessage>,
69}
70
71impl SpotFeedHandler {
72    /// Creates a new [`SpotFeedHandler`] instance.
73    pub(super) fn new(
74        signal: Arc<AtomicBool>,
75        cmd_rx: tokio::sync::mpsc::UnboundedReceiver<SpotHandlerCommand>,
76        raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
77        subscriptions: SubscriptionState,
78    ) -> Self {
79        Self {
80            signal,
81            inner: None,
82            cmd_rx,
83            raw_rx,
84            subscriptions,
85            pending_messages: VecDeque::new(),
86        }
87    }
88
89    pub(super) fn is_stopped(&self) -> bool {
90        self.signal.load(Ordering::Relaxed)
91    }
92
93    fn is_subscribed(&self, topic: &str) -> bool {
94        self.subscriptions.all_topics().iter().any(|t| t == topic)
95    }
96
97    /// Processes messages and commands, returning when stopped or stream ends.
98    pub(super) async fn next(&mut self) -> Option<KrakenSpotWsMessage> {
99        if let Some(msg) = self.pending_messages.pop_front() {
100            return Some(msg);
101        }
102
103        loop {
104            tokio::select! {
105                Some(cmd) = self.cmd_rx.recv() => {
106                    match cmd {
107                        SpotHandlerCommand::SetClient(client) => {
108                            log::debug!("WebSocketClient received by handler");
109                            self.inner = Some(client);
110                        }
111                        SpotHandlerCommand::Disconnect => {
112                            log::debug!("Disconnect command received");
113
114                            if let Some(client) = self.inner.take() {
115                                client.disconnect().await;
116                            }
117                        }
118                        SpotHandlerCommand::Subscribe { payload }
119                        | SpotHandlerCommand::Unsubscribe { payload } => {
120                            if let Some(client) = &self.inner
121                                && let Err(e) = client.send_text(payload.expose_secret().to_owned(), Some(KRAKEN_RATE_LIMIT_KEY_SUBSCRIPTION.as_slice())).await
122                            {
123                                log::error!("Failed to send text: {e}");
124                            }
125                        }
126                        SpotHandlerCommand::Ping { payload } => {
127                            if let Some(client) = &self.inner
128                                && let Err(e) = client.send_text(payload.expose_secret().to_owned(), None).await
129                            {
130                                log::error!("Failed to send text: {e}");
131                            }
132                        }
133                        SpotHandlerCommand::SendOrderRequest { req_id, payload } => {
134                            if let Some(client) = &self.inner {
135                                if let Err(e) = client.send_text(payload.expose_secret().to_owned(), Some(KRAKEN_RATE_LIMIT_KEY_ORDER.as_slice())).await {
136                                    log::error!(
137                                        "Kraken WS send_order_request failed req_id={req_id}: {e}"
138                                    );
139                                } else {
140                                    log::debug!(
141                                        "Kraken WS send_order_request enqueued req_id={req_id}"
142                                    );
143                                }
144                            } else {
145                                log::error!(
146                                    "Kraken WS send_order_request without active client req_id={req_id}"
147                                );
148                            }
149                        }
150                    }
151                }
152
153                msg = self.raw_rx.recv() => {
154                    let msg = match msg {
155                        Some(msg) => msg,
156                        None => {
157                            log::debug!("WebSocket stream closed");
158                            return None;
159                        }
160                    };
161
162                    if let Message::Ping(data) = &msg {
163                        log::trace!("Received ping frame with {} bytes", data.len());
164
165                        if let Some(client) = &self.inner
166                            && let Err(e) = client.send_pong(data.to_vec()).await
167                        {
168                            log::warn!("Failed to send pong frame: {e}");
169                        }
170                        continue;
171                    }
172
173                    if self.signal.load(Ordering::Relaxed) {
174                        log::debug!("Stop signal received");
175                        return None;
176                    }
177
178                    let text = match msg {
179                        Message::Text(text) => text.to_string(),
180                        Message::Binary(data) => {
181                            match String::from_utf8(data.to_vec()) {
182                                Ok(text) => text,
183                                Err(e) => {
184                                    log::warn!("Failed to decode binary message: {e}");
185                                    continue;
186                                }
187                            }
188                        }
189                        Message::Pong(_) => {
190                            log::trace!("Received pong");
191                            continue;
192                        }
193                        Message::Close(_) => {
194                            log::debug!("WebSocket connection closed");
195                            return None;
196                        }
197                        Message::Frame(_) => {
198                            log::trace!("Received raw frame");
199                            continue;
200                        }
201                        _ => continue,
202                    };
203
204                    if text == RECONNECTED {
205                        log::debug!("Received WebSocket reconnected signal");
206                        return Some(KrakenSpotWsMessage::Reconnected);
207                    }
208
209                    if let Some(msg) = self.parse_message(&text) {
210                        return Some(msg);
211                    }
212                }
213            }
214        }
215    }
216
217    fn parse_message(&self, text: &str) -> Option<KrakenSpotWsMessage> {
218        // Fast pre-filter for high-frequency control messages (no JSON parsing)
219        if text.len() < 50 && text.starts_with("{\"channel\":\"") {
220            if text.contains("heartbeat") {
221                log::trace!("Received heartbeat");
222                return None;
223            }
224
225            if text.contains("status") {
226                log::debug!("Received status message");
227                return None;
228            }
229        }
230
231        if text.contains("\"level3\"")
232            && let Some(msg) = parse_level3_text(text)
233        {
234            return self.handle_l3_message(msg);
235        }
236
237        let value: Value = match serde_json::from_str(text) {
238            Ok(v) => v,
239            Err(e) => {
240                log::warn!("Failed to parse message: {e}");
241                return None;
242            }
243        };
244
245        if value.get("method").is_some() {
246            match parse_order_response(text) {
247                Ok(Some(msg)) => return Some(msg),
248                Ok(None) => {}
249                Err(e) => log::warn!("Failed to parse order response: {e}"),
250            }
251            return self.handle_control_message(value);
252        }
253
254        if value.get("channel").is_some() && value.get("data").is_some() {
255            match serde_json::from_str::<KrakenWsRawMessage>(text) {
256                Ok(msg) => return self.handle_data_message(msg),
257                Err(e) => {
258                    log::debug!("Failed to parse data message: {e}");
259                    return None;
260                }
261            }
262        }
263
264        log::debug!("Unhandled message structure: {text}");
265        None
266    }
267
268    /// Logs a control response and yields every `subscribe` answer to the stream, so a consumer
269    /// holding a subscription can see the venue reject it rather than infer it from silence.
270    fn handle_control_message(&self, value: Value) -> Option<KrakenSpotWsMessage> {
271        match serde_json::from_value::<KrakenWsResponse>(value) {
272            Ok(response) => match response {
273                KrakenWsResponse::Subscribe(sub) => {
274                    if sub.success {
275                        if let Some(result) = &sub.result {
276                            log::debug!(
277                                "Subscription confirmed: channel={:?}, req_id={:?}",
278                                result.channel,
279                                sub.req_id
280                            );
281                        } else {
282                            log::debug!("Subscription confirmed: req_id={:?}", sub.req_id);
283                        }
284                    } else {
285                        log::warn!(
286                            "Subscription failed: error={:?}, req_id={:?}, symbol={:?}",
287                            sub.error,
288                            sub.req_id,
289                            sub.symbol
290                        );
291                    }
292
293                    Some(KrakenSpotWsMessage::SubscriptionAck {
294                        req_id: sub.req_id,
295                        symbol: sub.symbol,
296                        success: sub.success,
297                        error: sub.error,
298                    })
299                }
300                KrakenWsResponse::Unsubscribe(unsub) => {
301                    if unsub.success {
302                        log::debug!("Unsubscription confirmed: req_id={:?}", unsub.req_id);
303                    } else {
304                        log::warn!(
305                            "Unsubscription failed: error={:?}, req_id={:?}, symbol={:?}",
306                            unsub.error,
307                            unsub.req_id,
308                            unsub.symbol
309                        );
310                    }
311                    None
312                }
313                KrakenWsResponse::Pong(pong) => {
314                    log::trace!("Received pong: req_id={:?}", pong.req_id);
315                    None
316                }
317                KrakenWsResponse::Other => {
318                    log::debug!("Received unknown control response");
319                    None
320                }
321            },
322            Err(_) => {
323                log::debug!("Received control message (failed to parse details)");
324                None
325            }
326        }
327    }
328
329    fn handle_data_message(&self, msg: KrakenWsRawMessage) -> Option<KrakenSpotWsMessage> {
330        match msg.channel {
331            KrakenWsChannel::Book => self.handle_book_message(msg),
332            KrakenWsChannel::Ticker => self.handle_ticker_message(msg),
333            KrakenWsChannel::Trade => self.handle_trade_message(msg),
334            KrakenWsChannel::Ohlc => self.handle_ohlc_message(msg),
335            KrakenWsChannel::Executions => self.handle_executions_message(msg),
336            KrakenWsChannel::Level3 => {
337                // Well-formed level3 messages take the fast path in `parse_message`, so
338                // reaching here means the venue sent one that failed to parse
339                log::warn!("Dropping malformed level3 message");
340                None
341            }
342            _ => {
343                log::warn!("Unhandled channel: {:?}", msg.channel);
344                None
345            }
346        }
347    }
348
349    fn handle_book_message(&self, msg: KrakenWsRawMessage) -> Option<KrakenSpotWsMessage> {
350        let is_snapshot = msg.event_type == KrakenWsMessageType::Snapshot;
351        let mut book_data = Vec::new();
352
353        for data in msg.data {
354            match serde_json::from_str::<KrakenWsBookData>(data.get()) {
355                Ok(bd) => {
356                    if !self.is_subscribed(&format!("book:{}", bd.symbol)) {
357                        continue;
358                    }
359                    book_data.push(bd);
360                }
361                Err(e) => log::error!("Failed to deserialize book data: {e}"),
362            }
363        }
364
365        if book_data.is_empty() {
366            None
367        } else {
368            Some(KrakenSpotWsMessage::Book {
369                data: book_data,
370                is_snapshot,
371            })
372        }
373    }
374
375    fn handle_ticker_message(&self, msg: KrakenWsRawMessage) -> Option<KrakenSpotWsMessage> {
376        let mut tickers = Vec::new();
377
378        for data in msg.data {
379            match serde_json::from_str::<KrakenWsTickerData>(data.get()) {
380                Ok(td) => {
381                    let symbol = &td.symbol;
382                    let quotes_key = format!("quotes:{symbol}");
383                    let ticker_key = format!("ticker:{symbol}");
384                    if !self.is_subscribed(&quotes_key) && !self.is_subscribed(&ticker_key) {
385                        continue;
386                    }
387                    tickers.push(td);
388                }
389                Err(e) => log::error!("Failed to deserialize ticker data: {e}"),
390            }
391        }
392
393        if tickers.is_empty() {
394            None
395        } else {
396            Some(KrakenSpotWsMessage::Ticker(tickers))
397        }
398    }
399
400    fn handle_trade_message(&self, msg: KrakenWsRawMessage) -> Option<KrakenSpotWsMessage> {
401        let mut trades = Vec::new();
402
403        for data in msg.data {
404            match serde_json::from_str::<KrakenWsTradeData>(data.get()) {
405                Ok(td) => trades.push(td),
406                Err(e) => log::error!("Failed to deserialize trade data: {e}"),
407            }
408        }
409
410        if trades.is_empty() {
411            None
412        } else {
413            Some(KrakenSpotWsMessage::Trade(trades))
414        }
415    }
416
417    fn handle_ohlc_message(&self, msg: KrakenWsRawMessage) -> Option<KrakenSpotWsMessage> {
418        let mut ohlc_data = Vec::new();
419
420        for data in msg.data {
421            match serde_json::from_str::<KrakenWsOhlcData>(data.get()) {
422                Ok(od) => ohlc_data.push(od),
423                Err(e) => log::error!("Failed to deserialize OHLC data: {e}"),
424            }
425        }
426
427        if ohlc_data.is_empty() {
428            None
429        } else {
430            Some(KrakenSpotWsMessage::Ohlc(ohlc_data))
431        }
432    }
433
434    fn handle_executions_message(&self, msg: KrakenWsRawMessage) -> Option<KrakenSpotWsMessage> {
435        let mut executions = Vec::new();
436
437        for data in msg.data {
438            match serde_json::from_str::<KrakenWsExecutionData>(data.get()) {
439                Ok(ed) => executions.push(ed),
440                Err(e) => log::error!("Failed to deserialize execution data: {e}"),
441            }
442        }
443
444        if executions.is_empty() {
445            None
446        } else {
447            Some(KrakenSpotWsMessage::Execution(executions))
448        }
449    }
450}
451
452#[derive(Deserialize)]
453struct Level3RawMessage<'a> {
454    channel: &'a str,
455    #[serde(rename = "type")]
456    msg_type: &'a str,
457    #[serde(borrow)]
458    data: Vec<&'a RawValue>,
459}
460
461fn parse_level3_text(text: &str) -> Option<KrakenSpotWsMessage> {
462    let msg: Level3RawMessage<'_> = serde_json::from_str(text).ok()?;
463    if msg.channel != "level3" {
464        return None;
465    }
466    let first = msg.data.first()?.get();
467
468    match msg.msg_type {
469        "snapshot" => match serde_json::from_str::<KrakenL3Snapshot>(first) {
470            Ok(snap) => Some(KrakenSpotWsMessage::L3Snapshot(snap)),
471            Err(e) => {
472                log::warn!("Failed to deserialize L3 snapshot: {e}");
473                None
474            }
475        },
476        "update" => match serde_json::from_str::<KrakenL3UpdateData>(first) {
477            Ok(update) => Some(KrakenSpotWsMessage::L3Update(update)),
478            Err(e) => {
479                log::warn!("Failed to deserialize L3 update: {e}");
480                None
481            }
482        },
483        _ => None,
484    }
485}
486
487impl SpotFeedHandler {
488    fn handle_l3_message(&self, msg: KrakenSpotWsMessage) -> Option<KrakenSpotWsMessage> {
489        let symbol = match &msg {
490            KrakenSpotWsMessage::L3Snapshot(s) => &s.symbol,
491            KrakenSpotWsMessage::L3Update(u) => &u.symbol,
492            _ => return None,
493        };
494
495        if !self.is_subscribed(&format!("level3:{symbol}")) {
496            return None;
497        }
498        Some(msg)
499    }
500}
501
502#[cfg(test)]
503mod tests {
504    use rstest::rstest;
505    use rust_decimal_macros::dec;
506
507    use super::*;
508
509    fn create_test_handler() -> SpotFeedHandler {
510        let signal = Arc::new(AtomicBool::new(false));
511        let (_cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
512        let (_raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
513        let subscriptions = SubscriptionState::new(':');
514
515        SpotFeedHandler::new(signal, cmd_rx, raw_rx, subscriptions)
516    }
517
518    #[rstest]
519    fn test_ticker_message_filtered_without_quotes_subscription() {
520        let handler = create_test_handler();
521
522        let json = r#"{
523            "channel": "ticker",
524            "type": "snapshot",
525            "data": [{
526                "symbol": "BTC/USD",
527                "bid": 105944.20,
528                "bid_qty": 2.5,
529                "ask": 105944.30,
530                "ask_qty": 3.2,
531                "last": 105899.40,
532                "volume": 163.28908096,
533                "vwap": 105904.39279,
534                "low": 104711.00,
535                "high": 106613.10,
536                "change": 250.00,
537                "change_pct": 0.24,
538                "timestamp": "2022-12-25T09:30:59.123456Z"
539            }]
540        }"#;
541
542        let result = handler.parse_message(json);
543        assert!(
544            result.is_none(),
545            "Ticker message should be filtered when no quotes subscription exists"
546        );
547    }
548
549    #[rstest]
550    fn test_ticker_message_passes_with_quotes_subscription() {
551        let handler = create_test_handler();
552        handler.subscriptions.mark_subscribe("quotes:BTC/USD");
553        handler.subscriptions.confirm_subscribe("quotes:BTC/USD");
554
555        let json = r#"{
556            "channel": "ticker",
557            "type": "snapshot",
558            "data": [{
559                "symbol": "BTC/USD",
560                "bid": 105944.20,
561                "bid_qty": 2.5,
562                "ask": 105944.30,
563                "ask_qty": 3.2,
564                "last": 105899.40,
565                "volume": 163.28908096,
566                "vwap": 105904.39279,
567                "low": 104711.00,
568                "high": 106613.10,
569                "change": 250.00,
570                "change_pct": 0.24,
571                "timestamp": "2022-12-25T09:30:59.123456Z"
572            }]
573        }"#;
574
575        let result = handler.parse_message(json);
576        assert!(
577            result.is_some(),
578            "Ticker message should pass with quotes subscription"
579        );
580
581        match result.unwrap() {
582            KrakenSpotWsMessage::Ticker(data) => {
583                assert!(!data.is_empty(), "Should have ticker data");
584            }
585            _ => panic!("Expected Ticker message"),
586        }
587    }
588
589    #[rstest]
590    fn test_ticker_message_passes_with_ticker_subscription() {
591        let handler = create_test_handler();
592        handler.subscriptions.mark_subscribe("ticker:BTC/USD");
593        handler.subscriptions.confirm_subscribe("ticker:BTC/USD");
594
595        let json = r#"{
596            "channel": "ticker",
597            "type": "snapshot",
598            "data": [{
599                "symbol": "BTC/USD",
600                "bid": 105944.20,
601                "bid_qty": 2.5,
602                "ask": 105944.30,
603                "ask_qty": 3.2,
604                "last": 105899.40,
605                "volume": 163.28908096,
606                "vwap": 105904.39279,
607                "low": 104711.00,
608                "high": 106613.10,
609                "change": 250.00,
610                "change_pct": 0.24,
611                "timestamp": "2022-12-25T09:30:59.123456Z"
612            }]
613        }"#;
614
615        let result = handler.parse_message(json);
616        assert!(
617            result.is_some(),
618            "Ticker message should pass with ticker: subscription"
619        );
620
621        match result.unwrap() {
622            KrakenSpotWsMessage::Ticker(data) => {
623                assert!(!data.is_empty(), "Should have ticker data");
624            }
625            _ => panic!("Expected Ticker message"),
626        }
627    }
628
629    #[rstest]
630    fn test_ticker_message_preserves_decimal_precision() {
631        let handler = create_test_handler();
632        handler.subscriptions.mark_subscribe("ticker:BTC/USD");
633        handler.subscriptions.confirm_subscribe("ticker:BTC/USD");
634        let json = include_str!("../../../test_data/ws_ticker_precision.json");
635
636        let message = handler.parse_message(json).unwrap();
637
638        let KrakenSpotWsMessage::Ticker(data) = message else {
639            panic!("Expected Ticker message, was {message:?}");
640        };
641        let ticker = &data[0];
642        assert_eq!(ticker.symbol, "BTC/USD");
643        assert_eq!(ticker.bid, dec!(123456789.123456789));
644        assert_eq!(ticker.bid_qty, dec!(0.1234567890123456789012345678));
645        assert_eq!(ticker.ask, dec!(123456789.223456789));
646        assert_eq!(ticker.ask_qty, dec!(0.2234567890123456789012345678));
647        assert_eq!(ticker.last, dec!(123456789.323456789));
648        assert_eq!(ticker.volume, dec!(123456789.423456789));
649        assert_eq!(ticker.vwap, dec!(123456789.523456789));
650        assert_eq!(ticker.low, dec!(123456789.623456789));
651        assert_eq!(ticker.high, dec!(123456789.723456789));
652        assert_eq!(ticker.change, dec!(123456789.823456789));
653        assert_eq!(ticker.change_pct, dec!(0.9234567890123456789012345678));
654        assert_eq!(
655            ticker.timestamp,
656            "2022-12-25T09:30:59.123456Z"
657                .parse::<jiff::Timestamp>()
658                .unwrap()
659        );
660    }
661
662    #[rstest]
663    fn test_book_message_filtered_without_book_subscription() {
664        let handler = create_test_handler();
665
666        let json = r#"{
667            "channel": "book",
668            "type": "snapshot",
669            "data": [{
670                "symbol": "BTC/USD",
671                "bids": [{"price": 105944.20, "qty": 2.5}],
672                "asks": [{"price": 105944.30, "qty": 3.2}],
673                "checksum": 12345,
674                "timestamp": "2023-10-06T17:35:55.440295Z"
675            }]
676        }"#;
677
678        let result = handler.parse_message(json);
679        assert!(
680            result.is_none(),
681            "Book message should be filtered when no book subscription exists"
682        );
683    }
684
685    #[rstest]
686    fn test_book_message_passes_with_book_subscription() {
687        let handler = create_test_handler();
688        handler.subscriptions.mark_subscribe("book:BTC/USD");
689        handler.subscriptions.confirm_subscribe("book:BTC/USD");
690
691        let json = r#"{
692            "channel": "book",
693            "type": "snapshot",
694            "data": [{
695                "symbol": "BTC/USD",
696                "bids": [{"price": 105944.20, "qty": 2.5}],
697                "asks": [{"price": 105944.30, "qty": 3.2}],
698                "checksum": 12345,
699                "timestamp": "2023-10-06T17:35:55.440295Z"
700            }]
701        }"#;
702
703        let result = handler.parse_message(json);
704        assert!(
705            result.is_some(),
706            "Book message should pass with book subscription"
707        );
708
709        match result.unwrap() {
710            KrakenSpotWsMessage::Book { data, is_snapshot } => {
711                assert!(!data.is_empty());
712                assert!(is_snapshot);
713            }
714            _ => panic!("Expected Book message"),
715        }
716    }
717
718    #[rstest]
719    fn test_send_order_request_variant_construction() {
720        let cmd = SpotHandlerCommand::SendOrderRequest {
721            req_id: 7,
722            payload: SecretString::from(r#"{"method":"add_order","req_id":7}"#.to_string()),
723        };
724
725        match cmd {
726            SpotHandlerCommand::SendOrderRequest { req_id, payload } => {
727                assert_eq!(req_id, 7);
728                assert!(payload.expose_secret().contains("add_order"));
729            }
730            _ => panic!("Expected SendOrderRequest, was a different variant"),
731        }
732    }
733
734    #[rstest]
735    fn test_command_debug_redacts_payload() {
736        let cmd = SpotHandlerCommand::SendOrderRequest {
737            req_id: 7,
738            payload: SecretString::from("auth-token".to_string()),
739        };
740
741        let debug = format!("{cmd:?}");
742
743        assert!(debug.contains(REDACTED));
744        assert!(!debug.contains("auth-token"));
745    }
746
747    #[rstest]
748    #[tokio::test]
749    async fn test_send_order_request_without_active_client_does_not_panic() {
750        let signal = Arc::new(AtomicBool::new(false));
751        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
752        let (raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel::<Message>();
753        let subscriptions = SubscriptionState::new(':');
754
755        let mut handler = SpotFeedHandler::new(signal.clone(), cmd_rx, raw_rx, subscriptions);
756
757        cmd_tx
758            .send(SpotHandlerCommand::SendOrderRequest {
759                req_id: 42,
760                payload: SecretString::from(r#"{"method":"add_order","req_id":42}"#.to_string()),
761            })
762            .unwrap();
763
764        drop(cmd_tx);
765        drop(raw_tx);
766
767        let result = handler.next().await;
768        assert!(
769            result.is_none(),
770            "Handler should return None when streams close"
771        );
772    }
773
774    #[rstest]
775    fn test_quotes_and_book_subscriptions_independent() {
776        let handler = create_test_handler();
777        handler.subscriptions.mark_subscribe("quotes:BTC/USD");
778        handler.subscriptions.confirm_subscribe("quotes:BTC/USD");
779
780        let book_json = r#"{
781            "channel": "book",
782            "type": "snapshot",
783            "data": [{
784                "symbol": "BTC/USD",
785                "bids": [{"price": 105944.20, "qty": 2.5}],
786                "asks": [{"price": 105944.30, "qty": 3.2}],
787                "checksum": 12345,
788                "timestamp": "2023-10-06T17:35:55.440295Z"
789            }]
790        }"#;
791
792        let book_result = handler.parse_message(book_json);
793        assert!(
794            book_result.is_none(),
795            "Book message should be filtered without book: subscription"
796        );
797
798        let ticker_json = r#"{
799            "channel": "ticker",
800            "type": "snapshot",
801            "data": [{
802                "symbol": "BTC/USD",
803                "bid": 105944.20,
804                "bid_qty": 2.5,
805                "ask": 105944.30,
806                "ask_qty": 3.2,
807                "last": 105899.40,
808                "volume": 163.28908096,
809                "vwap": 105904.39279,
810                "low": 104711.00,
811                "high": 106613.10,
812                "change": 250.00,
813                "change_pct": 0.24,
814                "timestamp": "2022-12-25T09:30:59.123456Z"
815            }]
816        }"#;
817
818        let ticker_result = handler.parse_message(ticker_json);
819        assert!(
820            ticker_result.is_some(),
821            "Ticker should pass with quotes subscription"
822        );
823    }
824
825    #[rstest]
826    fn test_parse_message_routes_add_order_response_to_order_response_variant() {
827        use super::super::enums::KrakenWsMethod;
828
829        let handler = create_test_handler();
830        let json = r#"{"method":"add_order","req_id":42,"success":true,"time_in":"2026-05-05T10:00:00.123Z","time_out":"2026-05-05T10:00:00.125Z","result":{"order_id":"OABCDE-12345-FGHIJ","cl_ord_id":"O-20260505-000001","order_userref":0}}"#;
831
832        let result = handler.parse_message(json);
833        match result {
834            Some(KrakenSpotWsMessage::OrderResponse(resp)) => {
835                assert_eq!(resp.method, KrakenWsMethod::AddOrder);
836                assert_eq!(resp.req_id, Some(42));
837                assert!(resp.success);
838            }
839            other => panic!("expected OrderResponse, was {other:?}"),
840        }
841    }
842
843    /// A rejected `subscribe` reaches the stream with the request id, the pair and the venue's
844    /// reason, so the holder of the subscription can act on it.
845    #[rstest]
846    fn test_parse_message_yields_a_subscribe_rejection() {
847        let handler = create_test_handler();
848        let json = r#"{"error":"Currency pair not supported BOGUS/NOPE","method":"subscribe","req_id":7,"success":false,"symbol":"BOGUS/NOPE","time_in":"2026-05-05T10:00:00.123Z","time_out":"2026-05-05T10:00:00.125Z"}"#;
849
850        let result = handler.parse_message(json);
851
852        let Some(KrakenSpotWsMessage::SubscriptionAck {
853            req_id,
854            symbol,
855            success,
856            error,
857        }) = result
858        else {
859            panic!("expected a SubscriptionAck, was {result:?}");
860        };
861        assert_eq!(req_id, Some(7));
862        assert_eq!(symbol, Some(ustr::Ustr::from("BOGUS/NOPE")));
863        assert!(!success);
864        assert_eq!(
865            error.as_deref(),
866            Some("Currency pair not supported BOGUS/NOPE")
867        );
868    }
869
870    /// A confirmed `subscribe` reaches the stream too, so the consumer can retire the request.
871    #[rstest]
872    fn test_parse_message_yields_a_subscribe_confirmation() {
873        let handler = create_test_handler();
874        let json = r#"{"method":"subscribe","result":{"channel":"book","symbol":"BTC/USD","depth":10,"snapshot":true},"success":true,"time_in":"2026-05-05T10:00:00.123Z","time_out":"2026-05-05T10:00:00.125Z","req_id":3}"#;
875
876        let result = handler.parse_message(json);
877
878        let Some(KrakenSpotWsMessage::SubscriptionAck {
879            req_id,
880            symbol,
881            success,
882            error,
883        }) = result
884        else {
885            panic!("expected a SubscriptionAck, was {result:?}");
886        };
887        assert_eq!(req_id, Some(3));
888        assert_eq!(symbol, None, "a confirmation names the pair in result only");
889        assert!(success);
890        assert!(error.is_none());
891
892        let unsubscribe = r#"{"method":"unsubscribe","result":{"channel":"book","symbol":"BTC/USD"},"success":true,"time_in":"2026-05-05T10:00:00.123Z","time_out":"2026-05-05T10:00:00.125Z","req_id":4}"#;
893        assert!(
894            handler.parse_message(unsubscribe).is_none(),
895            "only subscribe answers reach the stream"
896        );
897    }
898
899    /// The handler yields a subscribe answer and the book frames after it in wire order, and a
900    /// failed unsubscribe, which the venue reports under the subscribe method with the
901    /// unsubscribe's request id, reaches the stream as a subscribe answer.
902    #[rstest]
903    #[tokio::test]
904    async fn test_next_yields_answers_and_book_frames_in_wire_order() {
905        let signal = Arc::new(AtomicBool::new(false));
906        let (_cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
907        let (raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
908        let subscriptions = SubscriptionState::new(':');
909        subscriptions.mark_subscribe("book:BTC/USD");
910        subscriptions.confirm_subscribe("book:BTC/USD");
911        let mut handler = SpotFeedHandler::new(signal, cmd_rx, raw_rx, subscriptions);
912
913        for text in [
914            r#"{"method":"subscribe","result":{"channel":"book","symbol":"BTC/USD","depth":10,"snapshot":true},"success":true,"time_in":"2026-05-05T10:00:00.123Z","time_out":"2026-05-05T10:00:00.125Z","req_id":3}"#,
915            r#"{"channel":"book","type":"snapshot","data":[{"symbol":"BTC/USD","bids":[{"price":105944.20,"qty":2.5}],"asks":[{"price":105944.30,"qty":3.2}],"checksum":12345,"timestamp":"2023-10-06T17:35:55.440295Z"}]}"#,
916            r#"{"channel":"book","type":"update","data":[{"symbol":"BTC/USD","bids":[{"price":105944.20,"qty":2.0}],"asks":[],"checksum":12346,"timestamp":"2023-10-06T17:35:55.540295Z"}]}"#,
917            r#"{"error":"Subscription with depth 10 not Found BTC/USD","method":"subscribe","req_id":502,"success":false,"symbol":"BTC/USD","time_in":"2026-05-05T10:00:01.123Z","time_out":"2026-05-05T10:00:01.125Z"}"#,
918        ] {
919            raw_tx.send(Message::Text(text.into())).unwrap();
920        }
921
922        let Some(KrakenSpotWsMessage::SubscriptionAck {
923            req_id: Some(3),
924            success: true,
925            ..
926        }) = handler.next().await
927        else {
928            panic!("expected the confirmation first");
929        };
930        let Some(KrakenSpotWsMessage::Book {
931            is_snapshot: true, ..
932        }) = handler.next().await
933        else {
934            panic!("expected the snapshot second");
935        };
936        let Some(KrakenSpotWsMessage::Book {
937            is_snapshot: false, ..
938        }) = handler.next().await
939        else {
940            panic!("expected the update third");
941        };
942        let failed_unsubscribe = handler.next().await;
943        let Some(KrakenSpotWsMessage::SubscriptionAck {
944            req_id: Some(502),
945            symbol: Some(symbol),
946            success: false,
947            ..
948        }) = failed_unsubscribe
949        else {
950            panic!(
951                "expected the failed unsubscribe as a subscribe answer, was {failed_unsubscribe:?}"
952            );
953        };
954        assert_eq!(symbol, ustr::Ustr::from("BTC/USD"));
955    }
956
957    #[rstest]
958    fn test_parse_level3_snapshot_with_subscription_passes() {
959        let handler = create_test_handler();
960        handler.subscriptions.mark_subscribe("level3:BTC/USD");
961        handler.subscriptions.confirm_subscribe("level3:BTC/USD");
962
963        let json = r#"{
964            "channel": "level3",
965            "type": "snapshot",
966            "data": [{
967                "symbol": "BTC/USD",
968                "bids": [],
969                "asks": [],
970                "checksum": 0,
971                "timestamp": "2024-01-01T00:00:00Z"
972            }]
973        }"#;
974
975        let result = handler.parse_message(json);
976        assert!(matches!(result, Some(KrakenSpotWsMessage::L3Snapshot(_))));
977    }
978
979    #[rstest]
980    fn test_parse_level3_update_without_subscription_filtered() {
981        let handler = create_test_handler();
982        let json = r#"{
983            "channel": "level3",
984            "type": "update",
985            "data": [{
986                "symbol": "BTC/USD",
987                "bids": [],
988                "asks": [],
989                "checksum": 0,
990                "timestamp": "2024-01-01T00:00:00Z"
991            }]
992        }"#;
993
994        let result = handler.parse_message(json);
995        assert!(result.is_none());
996    }
997
998    #[rstest]
999    #[case::empty_data(r#"{"channel":"level3","type":"update","data":[]}"#)]
1000    #[case::snapshot_missing_bids(
1001        r#"{"channel":"level3","type":"snapshot","data":[{"symbol":"BTC/USD","asks":[],"checksum":0,"timestamp":"2024-01-01T00:00:00Z"}]}"#
1002    )]
1003    #[case::unexpected_type(r#"{"channel":"level3","type":"error","data":[{}]}"#)]
1004    fn test_parse_level3_malformed_message_is_dropped(#[case] json: &str) {
1005        let handler = create_test_handler();
1006        handler.subscriptions.mark_subscribe("level3:BTC/USD");
1007        handler.subscriptions.confirm_subscribe("level3:BTC/USD");
1008
1009        let result = handler.parse_message(json);
1010        assert!(result.is_none());
1011    }
1012
1013    #[rstest]
1014    fn test_parse_level3_snapshot_compact_json() {
1015        let handler = create_test_handler();
1016        handler.subscriptions.mark_subscribe("level3:BTC/USD");
1017        handler.subscriptions.confirm_subscribe("level3:BTC/USD");
1018        let json = r#"{"channel":"level3","type":"snapshot","data":[{"symbol":"BTC/USD","bids":[],"asks":[],"checksum":0,"timestamp":"2024-01-01T00:00:00Z"}]}"#;
1019        assert!(matches!(
1020            handler.parse_message(json),
1021            Some(KrakenSpotWsMessage::L3Snapshot(_))
1022        ));
1023    }
1024
1025    #[rstest]
1026    fn test_parse_level3_snapshot_preserves_raw_decimal() {
1027        let handler = create_test_handler();
1028        handler.subscriptions.mark_subscribe("level3:BTC/USD");
1029        handler.subscriptions.confirm_subscribe("level3:BTC/USD");
1030
1031        let json = r#"{
1032            "channel": "level3",
1033            "type": "snapshot",
1034            "data": [{
1035                "symbol": "BTC/USD",
1036                "bids": [{
1037                    "order_id": "order-bid-1",
1038                    "limit_price": 42000.50000,
1039                    "order_qty": 0.01000000,
1040                    "timestamp": "2024-01-01T00:00:00Z"
1041                }],
1042                "asks": [],
1043                "checksum": 0,
1044                "timestamp": "2024-01-01T00:00:00Z"
1045            }]
1046        }"#;
1047
1048        let Some(KrakenSpotWsMessage::L3Snapshot(snap)) = handler.parse_message(json) else {
1049            panic!("expected L3 snapshot");
1050        };
1051        assert_eq!(snap.bids[0].limit_price.raw, "42000.50000");
1052        assert_eq!(snap.bids[0].order_qty.raw, "0.01000000");
1053    }
1054}
nautilus_kraken/websocket/spot_v2/handler.rs

nautilus_kraken/websocket/spot_v2/
handler.rs
