nautilus_kraken/websocket/futures/
client.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 client for the Kraken Futures v1 streaming API.
17
18use std::{
19    collections::HashMap,
20    sync::{
21        Arc, RwLock,
22        atomic::{AtomicBool, AtomicU8, Ordering},
23    },
24};
25
26use arc_swap::ArcSwap;
27use nautilus_common::live::get_runtime;
28use nautilus_core::AtomicMap;
29use nautilus_model::{
30    identifiers::{
31        AccountId, ClientOrderId, InstrumentId, StrategyId, Symbol, TraderId, VenueOrderId,
32    },
33    instruments::{Instrument, InstrumentAny},
34};
35use nautilus_network::{
36    mode::ConnectionMode,
37    websocket::{
38        AuthTracker, SubscriptionState, TransportBackend, WebSocketClient, WebSocketConfig,
39        channel_message_handler,
40    },
41};
42use tokio_util::sync::CancellationToken;
43
44use super::{
45    handler::{FuturesFeedHandler, FuturesHandlerCommand},
46    messages::{
47        KrakenFuturesChallengeRequest, KrakenFuturesEvent, KrakenFuturesFeed,
48        KrakenFuturesPrivateSubscribeRequest, KrakenFuturesRequest, KrakenFuturesWsMessage,
49    },
50};
51use crate::{
52    common::{
53        consts::{KRAKEN_FUTURES_WS_SUBSCRIPTION_QUOTA, KRAKEN_RATE_LIMIT_KEY_SUBSCRIPTION},
54        credential::KrakenCredential,
55        parse::truncate_cl_ord_id,
56    },
57    websocket::error::KrakenWsError,
58};
59
60/// Topic delimiter for Kraken Futures WebSocket subscriptions.
61///
62/// Topics use colon format: `feed:symbol` (e.g., `trades:PF_ETHUSD`).
63pub const KRAKEN_FUTURES_WS_TOPIC_DELIMITER: char = ':';
64
65/// WebSocket client for the Kraken Futures v1 streaming API.
66#[derive(Debug)]
67#[cfg_attr(
68    feature = "python",
69    pyo3::pyclass(module = "nautilus_trader.core.nautilus_pyo3.kraken", from_py_object)
70)]
71#[cfg_attr(
72    feature = "python",
73    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.adapters.kraken")
74)]
75pub struct KrakenFuturesWebSocketClient {
76    url: String,
77    heartbeat_secs: u64,
78    signal: Arc<AtomicBool>,
79    connection_mode: Arc<ArcSwap<AtomicU8>>,
80    cmd_tx: Arc<tokio::sync::RwLock<tokio::sync::mpsc::UnboundedSender<FuturesHandlerCommand>>>,
81    out_rx: Option<Arc<tokio::sync::mpsc::UnboundedReceiver<KrakenFuturesWsMessage>>>,
82    task_handle: Option<Arc<tokio::task::JoinHandle<()>>>,
83    subscriptions: SubscriptionState,
84    subscription_payloads: Arc<tokio::sync::RwLock<HashMap<String, String>>>,
85    auth_tracker: AuthTracker,
86    cancellation_token: CancellationToken,
87    credential: Option<KrakenCredential>,
88    original_challenge: Arc<tokio::sync::RwLock<Option<String>>>,
89    signed_challenge: Arc<tokio::sync::RwLock<Option<String>>>,
90    account_id: Arc<RwLock<Option<AccountId>>>,
91    truncated_id_map: Arc<AtomicMap<String, ClientOrderId>>,
92    order_instrument_map: Arc<AtomicMap<String, InstrumentId>>,
93    instruments: Arc<AtomicMap<InstrumentId, InstrumentAny>>,
94    transport_backend: TransportBackend,
95    proxy_url: Option<String>,
96}
97
98impl Clone for KrakenFuturesWebSocketClient {
99    fn clone(&self) -> Self {
100        Self {
101            url: self.url.clone(),
102            heartbeat_secs: self.heartbeat_secs,
103            signal: Arc::clone(&self.signal),
104            connection_mode: Arc::clone(&self.connection_mode),
105            cmd_tx: Arc::clone(&self.cmd_tx),
106            out_rx: self.out_rx.clone(),
107            task_handle: self.task_handle.clone(),
108            subscriptions: self.subscriptions.clone(),
109            subscription_payloads: Arc::clone(&self.subscription_payloads),
110            auth_tracker: self.auth_tracker.clone(),
111            cancellation_token: self.cancellation_token.clone(),
112            credential: self.credential.clone(),
113            original_challenge: Arc::clone(&self.original_challenge),
114            signed_challenge: Arc::clone(&self.signed_challenge),
115            account_id: Arc::clone(&self.account_id),
116            truncated_id_map: Arc::clone(&self.truncated_id_map),
117            order_instrument_map: Arc::clone(&self.order_instrument_map),
118            instruments: Arc::clone(&self.instruments),
119            transport_backend: self.transport_backend,
120            proxy_url: self.proxy_url.clone(),
121        }
122    }
123}
124
125impl KrakenFuturesWebSocketClient {
126    /// Creates a new client with the given URL.
127    #[must_use]
128    pub fn new(url: String, heartbeat_secs: u64, proxy_url: Option<String>) -> Self {
129        Self::with_credentials(
130            url,
131            heartbeat_secs,
132            None,
133            TransportBackend::default(),
134            proxy_url,
135        )
136    }
137
138    /// Creates a new client with API credentials for authenticated feeds.
139    #[must_use]
140    pub fn with_credentials(
141        url: String,
142        heartbeat_secs: u64,
143        credential: Option<KrakenCredential>,
144        transport_backend: TransportBackend,
145        proxy_url: Option<String>,
146    ) -> Self {
147        let (cmd_tx, _cmd_rx) = tokio::sync::mpsc::unbounded_channel::<FuturesHandlerCommand>();
148        let initial_mode = AtomicU8::new(ConnectionMode::Closed.as_u8());
149        let connection_mode = Arc::new(ArcSwap::from_pointee(initial_mode));
150
151        Self {
152            url,
153            heartbeat_secs,
154            signal: Arc::new(AtomicBool::new(false)),
155            connection_mode,
156            cmd_tx: Arc::new(tokio::sync::RwLock::new(cmd_tx)),
157            out_rx: None,
158            task_handle: None,
159            subscriptions: SubscriptionState::new(KRAKEN_FUTURES_WS_TOPIC_DELIMITER),
160            subscription_payloads: Arc::new(tokio::sync::RwLock::new(HashMap::new())),
161            auth_tracker: AuthTracker::new(),
162            cancellation_token: CancellationToken::new(),
163            credential,
164            original_challenge: Arc::new(tokio::sync::RwLock::new(None)),
165            signed_challenge: Arc::new(tokio::sync::RwLock::new(None)),
166            account_id: Arc::new(RwLock::new(None)),
167            truncated_id_map: Arc::new(AtomicMap::new()),
168            order_instrument_map: Arc::new(AtomicMap::new()),
169            instruments: Arc::new(AtomicMap::new()),
170            transport_backend,
171            proxy_url,
172        }
173    }
174
175    /// Returns true if the client has API credentials set.
176    #[must_use]
177    pub fn has_credentials(&self) -> bool {
178        self.credential.is_some()
179    }
180
181    /// Returns the WebSocket URL.
182    #[must_use]
183    pub fn url(&self) -> &str {
184        &self.url
185    }
186
187    /// Returns true if the connection is closed.
188    #[must_use]
189    pub fn is_closed(&self) -> bool {
190        ConnectionMode::from_u8(self.connection_mode.load().load(Ordering::Relaxed))
191            == ConnectionMode::Closed
192    }
193
194    /// Returns true if the connection is active.
195    #[must_use]
196    pub fn is_active(&self) -> bool {
197        ConnectionMode::from_u8(self.connection_mode.load().load(Ordering::Relaxed))
198            == ConnectionMode::Active
199    }
200
201    /// Waits until the WebSocket connection is active or timeout.
202    pub async fn wait_until_active(&self, timeout_secs: f64) -> Result<(), KrakenWsError> {
203        let timeout = tokio::time::Duration::from_secs_f64(timeout_secs);
204
205        tokio::time::timeout(timeout, async {
206            while !self.is_active() {
207                tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
208            }
209        })
210        .await
211        .map_err(|_| {
212            KrakenWsError::ConnectionError(format!(
213                "WebSocket connection timeout after {timeout_secs} seconds"
214            ))
215        })?;
216
217        Ok(())
218    }
219
220    /// Returns true if the WebSocket is authenticated for private feeds.
221    #[must_use]
222    pub fn is_authenticated(&self) -> bool {
223        self.auth_tracker.is_authenticated()
224    }
225
226    /// Waits until the WebSocket is authenticated or the timeout elapses.
227    ///
228    /// Returns an error on timeout or explicit auth failure.
229    pub async fn wait_until_authenticated(&self, timeout_secs: f64) -> Result<(), KrakenWsError> {
230        let timeout = tokio::time::Duration::from_secs_f64(timeout_secs);
231        if self.auth_tracker.wait_for_authenticated(timeout).await {
232            Ok(())
233        } else {
234            Err(KrakenWsError::AuthenticationError(format!(
235                "Authentication not completed within {timeout_secs} seconds"
236            )))
237        }
238    }
239
240    /// Authenticates the WebSocket connection for private feeds.
241    ///
242    /// Sends a challenge request and waits for the handler to parse the response,
243    /// sign it, and mark the `AuthTracker` successful. Private subscriptions gate
244    /// on the stored challenge / signed-challenge pair.
245    pub async fn authenticate(&self) -> Result<(), KrakenWsError> {
246        let credential = self.credential.as_ref().ok_or_else(|| {
247            KrakenWsError::AuthenticationError("API credentials required".to_string())
248        })?;
249
250        let payload = build_challenge_payload(credential)
251            .map_err(|e| KrakenWsError::JsonError(e.to_string()))?;
252
253        let receiver = self.auth_tracker.begin();
254
255        self.cmd_tx
256            .read()
257            .await
258            .send(FuturesHandlerCommand::RequestChallenge { payload })
259            .map_err(|e| KrakenWsError::ChannelError(e.to_string()))?;
260
261        self.auth_tracker
262            .wait_for_result::<KrakenWsError>(tokio::time::Duration::from_secs(10), receiver)
263            .await?;
264
265        log::debug!("Futures WebSocket authentication successful");
266        Ok(())
267    }
268
269    /// Connects to the WebSocket server.
270    pub async fn connect(&mut self) -> Result<(), KrakenWsError> {
271        log::debug!("Connecting to Futures WebSocket: {}", self.url);
272
273        self.signal.store(false, Ordering::Relaxed);
274
275        let (raw_handler, raw_rx) = channel_message_handler();
276
277        let ws_config = WebSocketConfig {
278            url: self.url.clone(),
279            headers: vec![],
280            heartbeat: Some(self.heartbeat_secs),
281            heartbeat_msg: None, // Use WebSocket ping frames, not text messages
282            reconnect_timeout_ms: Some(5_000),
283            reconnect_delay_initial_ms: Some(500),
284            reconnect_delay_max_ms: Some(5_000),
285            reconnect_backoff_factor: Some(1.5),
286            reconnect_jitter_ms: Some(250),
287            reconnect_max_attempts: None,
288            idle_timeout_ms: None,
289            backend: self.transport_backend,
290            proxy_url: self.proxy_url.clone(),
291        };
292
293        let keyed_quotas = vec![(
294            KRAKEN_RATE_LIMIT_KEY_SUBSCRIPTION[0].to_string(),
295            *KRAKEN_FUTURES_WS_SUBSCRIPTION_QUOTA,
296        )];
297
298        let ws_client =
299            WebSocketClient::connect(ws_config, Some(raw_handler), None, None, keyed_quotas, None)
300                .await
301                .map_err(|e| KrakenWsError::ConnectionError(e.to_string()))?;
302
303        self.connection_mode
304            .store(ws_client.connection_mode_atomic());
305
306        let (out_tx, out_rx) = tokio::sync::mpsc::unbounded_channel::<KrakenFuturesWsMessage>();
307        self.out_rx = Some(Arc::new(out_rx));
308
309        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel::<FuturesHandlerCommand>();
310        *self.cmd_tx.write().await = cmd_tx.clone();
311
312        if let Err(e) = cmd_tx.send(FuturesHandlerCommand::SetClient(ws_client)) {
313            return Err(KrakenWsError::ConnectionError(format!(
314                "Failed to send WebSocketClient to handler: {e}"
315            )));
316        }
317
318        let signal = self.signal.clone();
319        let subscriptions = self.subscriptions.clone();
320        let subscription_payloads = self.subscription_payloads.clone();
321        let cmd_tx_for_reconnect = cmd_tx.clone();
322        let credential_for_reconnect = self.credential.clone();
323        let original_challenge_for_reconnect = self.original_challenge.clone();
324        let signed_challenge_for_reconnect = self.signed_challenge.clone();
325        let auth_tracker_for_reconnect = self.auth_tracker.clone();
326
327        let stream_handle = get_runtime().spawn(async move {
328            let mut handler =
329                FuturesFeedHandler::new(signal.clone(), cmd_rx, raw_rx, subscriptions.clone());
330            let mut pending_resubscribe = false;
331
332            loop {
333                match handler.next().await {
334                    Some(KrakenFuturesWsMessage::Reconnected) => {
335                        if signal.load(Ordering::Relaxed) {
336                            continue;
337                        }
338                        log::info!("WebSocket reconnected");
339
340                        let confirmed_topics = subscriptions.all_topics();
341                        for topic in &confirmed_topics {
342                            subscriptions.mark_failure(topic);
343                        }
344
345                        auth_tracker_for_reconnect.invalidate();
346                        *original_challenge_for_reconnect.write().await = None;
347                        *signed_challenge_for_reconnect.write().await = None;
348
349                        let payloads = subscription_payloads.read().await.clone();
350
351                        // Resubscribe public topics straight away; they don't depend on auth,
352                        // so don't tie their restoration to the challenge outcome.
353                        resubscribe_public(&cmd_tx_for_reconnect, &subscriptions, &payloads);
354
355                        let has_private =
356                            payloads.keys().any(|k| k == "open_orders" || k == "fills");
357
358                        pending_resubscribe = false;
359
360                        if has_private {
361                            if let Some(ref cred) = credential_for_reconnect {
362                                match build_challenge_payload(cred) {
363                                    Ok(payload) => {
364                                        let _rx = auth_tracker_for_reconnect.begin();
365
366                                        if let Err(e) = cmd_tx_for_reconnect.send(
367                                            FuturesHandlerCommand::RequestChallenge { payload },
368                                        ) {
369                                            log::error!("Failed to queue reconnect challenge: {e}");
370                                        } else {
371                                            pending_resubscribe = true;
372                                        }
373                                    }
374                                    Err(e) => {
375                                        log::error!("Failed to serialize reconnect challenge: {e}");
376                                    }
377                                }
378                            } else {
379                                log::warn!(
380                                    "Private subscriptions exist but no credentials available"
381                                );
382                            }
383                        }
384
385                        if let Err(e) = out_tx.send(KrakenFuturesWsMessage::Reconnected) {
386                            log::debug!("Output channel closed: {e}");
387                            break;
388                        }
389                    }
390                    Some(KrakenFuturesWsMessage::Challenge(challenge)) => {
391                        let Some(ref cred) = credential_for_reconnect else {
392                            log::warn!("Challenge received but no credentials configured");
393                            auth_tracker_for_reconnect.fail("no credentials");
394                            continue;
395                        };
396
397                        match cred.sign_ws_challenge(&challenge) {
398                            Ok(signed) => {
399                                *original_challenge_for_reconnect.write().await =
400                                    Some(challenge.clone());
401                                *signed_challenge_for_reconnect.write().await =
402                                    Some(signed.clone());
403                                auth_tracker_for_reconnect.succeed();
404                                log::debug!("Signed WebSocket challenge");
405
406                                if pending_resubscribe {
407                                    let payloads = subscription_payloads.read().await;
408                                    resubscribe_private(
409                                        &cmd_tx_for_reconnect,
410                                        &subscriptions,
411                                        &payloads,
412                                        cred,
413                                        challenge.as_str(),
414                                        signed.as_str(),
415                                    );
416                                    pending_resubscribe = false;
417                                }
418                            }
419                            Err(e) => {
420                                log::error!("Failed to sign challenge: {e}");
421                                auth_tracker_for_reconnect.fail(e.to_string());
422                                pending_resubscribe = false;
423                            }
424                        }
425                    }
426                    Some(msg) => {
427                        if let Err(e) = out_tx.send(msg) {
428                            log::debug!("Output channel closed: {e}");
429                            break;
430                        }
431                    }
432                    None => {
433                        log::debug!("Handler stream ended");
434                        break;
435                    }
436                }
437            }
438
439            log::debug!("Futures handler task exiting");
440        });
441
442        self.task_handle = Some(Arc::new(stream_handle));
443
444        log::debug!("Futures WebSocket connected successfully");
445        Ok(())
446    }
447
448    /// Disconnects from the WebSocket server.
449    pub async fn disconnect(&mut self) -> Result<(), KrakenWsError> {
450        log::debug!("Disconnecting Futures WebSocket");
451
452        self.signal.store(true, Ordering::Relaxed);
453
454        if let Err(e) = self
455            .cmd_tx
456            .read()
457            .await
458            .send(FuturesHandlerCommand::Disconnect)
459        {
460            log::debug!(
461                "Failed to send disconnect command (handler may already be shut down): {e}"
462            );
463        }
464
465        if let Some(task_handle) = self.task_handle.take() {
466            match Arc::try_unwrap(task_handle) {
467                Ok(handle) => {
468                    match tokio::time::timeout(tokio::time::Duration::from_secs(2), handle).await {
469                        Ok(Ok(())) => log::debug!("Task handle completed successfully"),
470                        Ok(Err(e)) => log::error!("Task handle encountered an error: {e:?}"),
471                        Err(_) => {
472                            log::warn!("Timeout waiting for task handle");
473                        }
474                    }
475                }
476                Err(arc_handle) => {
477                    log::debug!("Cannot take ownership of task handle, aborting");
478                    arc_handle.abort();
479                }
480            }
481        }
482
483        self.subscriptions.clear();
484        self.subscription_payloads.write().await.clear();
485        self.auth_tracker.fail("Disconnected");
486        Ok(())
487    }
488
489    /// Closes the WebSocket connection.
490    pub async fn close(&mut self) -> Result<(), KrakenWsError> {
491        self.disconnect().await
492    }
493
494    /// Subscribes to mark price updates for the given instrument.
495    pub async fn subscribe_mark_price(
496        &self,
497        instrument_id: InstrumentId,
498    ) -> Result<(), KrakenWsError> {
499        let symbol = instrument_id.symbol;
500        let key = format!("mark:{symbol}");
501
502        if !self.subscriptions.add_reference(&key) {
503            return Ok(());
504        }
505
506        self.subscriptions.mark_subscribe(&key);
507        self.subscriptions.confirm_subscribe(&key);
508        self.ensure_ticker_subscribed(symbol).await
509    }
510
511    /// Unsubscribes from mark price updates for the given instrument.
512    pub async fn unsubscribe_mark_price(
513        &self,
514        instrument_id: InstrumentId,
515    ) -> Result<(), KrakenWsError> {
516        let symbol = instrument_id.symbol;
517        let key = format!("mark:{symbol}");
518
519        if !self.subscriptions.remove_reference(&key) {
520            return Ok(());
521        }
522
523        self.subscriptions.mark_unsubscribe(&key);
524        self.subscriptions.confirm_unsubscribe(&key);
525        self.maybe_unsubscribe_ticker(symbol).await
526    }
527
528    /// Subscribes to index price updates for the given instrument.
529    pub async fn subscribe_index_price(
530        &self,
531        instrument_id: InstrumentId,
532    ) -> Result<(), KrakenWsError> {
533        let symbol = instrument_id.symbol;
534        let key = format!("index:{symbol}");
535
536        if !self.subscriptions.add_reference(&key) {
537            return Ok(());
538        }
539
540        self.subscriptions.mark_subscribe(&key);
541        self.subscriptions.confirm_subscribe(&key);
542        self.ensure_ticker_subscribed(symbol).await
543    }
544
545    /// Unsubscribes from index price updates for the given instrument.
546    pub async fn unsubscribe_index_price(
547        &self,
548        instrument_id: InstrumentId,
549    ) -> Result<(), KrakenWsError> {
550        let symbol = instrument_id.symbol;
551        let key = format!("index:{symbol}");
552
553        if !self.subscriptions.remove_reference(&key) {
554            return Ok(());
555        }
556
557        self.subscriptions.mark_unsubscribe(&key);
558        self.subscriptions.confirm_unsubscribe(&key);
559        self.maybe_unsubscribe_ticker(symbol).await
560    }
561
562    /// Subscribes to funding rate updates for the given instrument.
563    pub async fn subscribe_funding_rate(
564        &self,
565        instrument_id: InstrumentId,
566    ) -> Result<(), KrakenWsError> {
567        let symbol = instrument_id.symbol;
568        let key = format!("funding:{symbol}");
569
570        if !self.subscriptions.add_reference(&key) {
571            return Ok(());
572        }
573
574        self.subscriptions.mark_subscribe(&key);
575        self.subscriptions.confirm_subscribe(&key);
576        self.ensure_ticker_subscribed(symbol).await
577    }
578
579    /// Unsubscribes from funding rate updates for the given instrument.
580    pub async fn unsubscribe_funding_rate(
581        &self,
582        instrument_id: InstrumentId,
583    ) -> Result<(), KrakenWsError> {
584        let symbol = instrument_id.symbol;
585        let key = format!("funding:{symbol}");
586
587        if !self.subscriptions.remove_reference(&key) {
588            return Ok(());
589        }
590
591        self.subscriptions.mark_unsubscribe(&key);
592        self.subscriptions.confirm_unsubscribe(&key);
593        self.maybe_unsubscribe_ticker(symbol).await
594    }
595
596    /// Subscribes to quote updates for the given instrument.
597    ///
598    /// Uses the order book channel for low-latency top-of-book quotes.
599    pub async fn subscribe_quotes(&self, instrument_id: InstrumentId) -> Result<(), KrakenWsError> {
600        let symbol = instrument_id.symbol;
601        let key = format!("quotes:{symbol}");
602
603        if !self.subscriptions.add_reference(&key) {
604            return Ok(());
605        }
606
607        self.subscriptions.mark_subscribe(&key);
608        self.subscriptions.confirm_subscribe(&key);
609
610        // Use book feed for low-latency quotes (not throttled ticker)
611        self.ensure_book_subscribed(symbol).await
612    }
613
614    /// Unsubscribes from quote updates for the given instrument.
615    pub async fn unsubscribe_quotes(
616        &self,
617        instrument_id: InstrumentId,
618    ) -> Result<(), KrakenWsError> {
619        let symbol = instrument_id.symbol;
620        let key = format!("quotes:{symbol}");
621
622        if !self.subscriptions.remove_reference(&key) {
623            return Ok(());
624        }
625
626        self.subscriptions.mark_unsubscribe(&key);
627        self.subscriptions.confirm_unsubscribe(&key);
628        self.maybe_unsubscribe_book(symbol).await
629    }
630
631    /// Subscribes to trade updates for the given instrument.
632    pub async fn subscribe_trades(&self, instrument_id: InstrumentId) -> Result<(), KrakenWsError> {
633        let symbol = instrument_id.symbol;
634        let key = format!("trades:{symbol}");
635
636        if !self.subscriptions.add_reference(&key) {
637            return Ok(());
638        }
639
640        self.subscriptions.mark_subscribe(&key);
641        let payload = self
642            .send_subscribe_feed(KrakenFuturesFeed::Trade, vec![symbol.to_string()])
643            .await?;
644        self.subscriptions.confirm_subscribe(&key);
645        self.subscription_payloads
646            .write()
647            .await
648            .insert(key, payload);
649        Ok(())
650    }
651
652    /// Unsubscribes from trade updates for the given instrument.
653    pub async fn unsubscribe_trades(
654        &self,
655        instrument_id: InstrumentId,
656    ) -> Result<(), KrakenWsError> {
657        let symbol = instrument_id.symbol;
658        let key = format!("trades:{symbol}");
659
660        if !self.subscriptions.remove_reference(&key) {
661            return Ok(());
662        }
663
664        self.subscriptions.mark_unsubscribe(&key);
665        self.send_unsubscribe_feed(KrakenFuturesFeed::Trade, vec![symbol.to_string()])
666            .await?;
667        self.subscriptions.confirm_unsubscribe(&key);
668        self.subscription_payloads.write().await.remove(&key);
669        Ok(())
670    }
671
672    /// Subscribes to order book updates for the given instrument.
673    ///
674    /// Note: The `depth` parameter is accepted for API compatibility with spot client but is
675    /// not used by Kraken Futures (full book is always returned).
676    pub async fn subscribe_book(
677        &self,
678        instrument_id: InstrumentId,
679        _depth: Option<u32>,
680    ) -> Result<(), KrakenWsError> {
681        let symbol = instrument_id.symbol;
682
683        let deltas_key = format!("deltas:{symbol}");
684        self.subscriptions.add_reference(&deltas_key);
685        self.subscriptions.mark_subscribe(&deltas_key);
686        self.subscriptions.confirm_subscribe(&deltas_key);
687
688        self.ensure_book_subscribed(symbol).await
689    }
690
691    /// Unsubscribes from order book updates for the given instrument.
692    pub async fn unsubscribe_book(&self, instrument_id: InstrumentId) -> Result<(), KrakenWsError> {
693        let symbol = instrument_id.symbol;
694
695        let deltas_key = format!("deltas:{symbol}");
696        self.subscriptions.remove_reference(&deltas_key);
697        self.subscriptions.mark_unsubscribe(&deltas_key);
698        self.subscriptions.confirm_unsubscribe(&deltas_key);
699
700        self.maybe_unsubscribe_book(symbol).await
701    }
702
703    async fn ensure_ticker_subscribed(&self, symbol: Symbol) -> Result<(), KrakenWsError> {
704        let ticker_key = format!("ticker:{symbol}");
705
706        if !self.subscriptions.add_reference(&ticker_key) {
707            return Ok(());
708        }
709
710        self.subscriptions.mark_subscribe(&ticker_key);
711        let payload = self
712            .send_subscribe_feed(KrakenFuturesFeed::Ticker, vec![symbol.to_string()])
713            .await?;
714        self.subscriptions.confirm_subscribe(&ticker_key);
715        self.subscription_payloads
716            .write()
717            .await
718            .insert(ticker_key, payload);
719        Ok(())
720    }
721
722    async fn maybe_unsubscribe_ticker(&self, symbol: Symbol) -> Result<(), KrakenWsError> {
723        let ticker_key = format!("ticker:{symbol}");
724
725        if !self.subscriptions.remove_reference(&ticker_key) {
726            return Ok(());
727        }
728
729        self.subscriptions.mark_unsubscribe(&ticker_key);
730        self.send_unsubscribe_feed(KrakenFuturesFeed::Ticker, vec![symbol.to_string()])
731            .await?;
732        self.subscriptions.confirm_unsubscribe(&ticker_key);
733        self.subscription_payloads.write().await.remove(&ticker_key);
734        Ok(())
735    }
736
737    async fn ensure_book_subscribed(&self, symbol: Symbol) -> Result<(), KrakenWsError> {
738        let book_key = format!("book:{symbol}");
739
740        if !self.subscriptions.add_reference(&book_key) {
741            return Ok(());
742        }
743
744        self.subscriptions.mark_subscribe(&book_key);
745        let payload = self
746            .send_subscribe_feed(KrakenFuturesFeed::Book, vec![symbol.to_string()])
747            .await?;
748        self.subscriptions.confirm_subscribe(&book_key);
749        self.subscription_payloads
750            .write()
751            .await
752            .insert(book_key, payload);
753        Ok(())
754    }
755
756    async fn maybe_unsubscribe_book(&self, symbol: Symbol) -> Result<(), KrakenWsError> {
757        let book_key = format!("book:{symbol}");
758
759        if !self.subscriptions.remove_reference(&book_key) {
760            return Ok(());
761        }
762
763        self.subscriptions.mark_unsubscribe(&book_key);
764        self.send_unsubscribe_feed(KrakenFuturesFeed::Book, vec![symbol.to_string()])
765            .await?;
766        self.subscriptions.confirm_unsubscribe(&book_key);
767        self.subscription_payloads.write().await.remove(&book_key);
768        Ok(())
769    }
770
771    /// Gets the output receiver for processed messages.
772    pub fn take_output_rx(
773        &mut self,
774    ) -> Option<tokio::sync::mpsc::UnboundedReceiver<KrakenFuturesWsMessage>> {
775        self.out_rx.take().and_then(|arc| Arc::try_unwrap(arc).ok())
776    }
777
778    /// Set authentication credentials directly (for when challenge is obtained externally).
779    pub async fn set_auth_credentials(
780        &self,
781        original_challenge: String,
782        signed_challenge: String,
783    ) -> Result<(), KrakenWsError> {
784        let _credential = self.credential.as_ref().ok_or_else(|| {
785            KrakenWsError::AuthenticationError("API credentials required".to_string())
786        })?;
787
788        *self.original_challenge.write().await = Some(original_challenge);
789        *self.signed_challenge.write().await = Some(signed_challenge);
790        self.auth_tracker.succeed();
791
792        Ok(())
793    }
794
795    /// Sign a challenge with the API credentials.
796    ///
797    /// Returns the signed challenge on success.
798    pub fn sign_challenge(&self, challenge: &str) -> Result<String, KrakenWsError> {
799        let credential = self.credential.as_ref().ok_or_else(|| {
800            KrakenWsError::AuthenticationError("API credentials required".to_string())
801        })?;
802
803        credential.sign_ws_challenge(challenge).map_err(|e| {
804            KrakenWsError::AuthenticationError(format!("Failed to sign challenge: {e}"))
805        })
806    }
807
808    /// Complete authentication with a received challenge.
809    pub async fn authenticate_with_challenge(&self, challenge: &str) -> Result<(), KrakenWsError> {
810        let credential = self.credential.as_ref().ok_or_else(|| {
811            KrakenWsError::AuthenticationError("API credentials required".to_string())
812        })?;
813
814        let signed_challenge = credential.sign_ws_challenge(challenge).map_err(|e| {
815            KrakenWsError::AuthenticationError(format!("Failed to sign challenge: {e}"))
816        })?;
817
818        self.set_auth_credentials(challenge.to_string(), signed_challenge)
819            .await
820    }
821
822    /// Sets the account ID for execution report parsing.
823    pub fn set_account_id(&self, account_id: AccountId) {
824        if let Ok(mut guard) = self.account_id.write() {
825            *guard = Some(account_id);
826        }
827    }
828
829    /// Returns the account ID if set.
830    #[must_use]
831    pub fn account_id(&self) -> Option<AccountId> {
832        self.account_id.read().ok().and_then(|g| *g)
833    }
834
835    /// Returns a reference to the shared account ID.
836    #[must_use]
837    pub fn account_id_shared(&self) -> &Arc<RwLock<Option<AccountId>>> {
838        &self.account_id
839    }
840
841    /// Returns a reference to the truncated ID map.
842    #[must_use]
843    pub fn truncated_id_map(&self) -> &Arc<AtomicMap<String, ClientOrderId>> {
844        &self.truncated_id_map
845    }
846
847    /// Returns a reference to the order-to-instrument map.
848    #[must_use]
849    pub fn order_instrument_map(&self) -> &Arc<AtomicMap<String, InstrumentId>> {
850        &self.order_instrument_map
851    }
852
853    /// Returns a reference to the shared instruments map.
854    #[must_use]
855    pub fn instruments_shared(&self) -> &Arc<AtomicMap<InstrumentId, InstrumentAny>> {
856        &self.instruments
857    }
858
859    /// Returns a reference to the subscription state.
860    #[must_use]
861    pub fn subscriptions(&self) -> &SubscriptionState {
862        &self.subscriptions
863    }
864
865    /// Caches an instrument for execution report parsing.
866    pub fn cache_instrument(&self, instrument: InstrumentAny) {
867        self.instruments.insert(instrument.id(), instrument);
868    }
869
870    /// Caches multiple instruments for execution report parsing.
871    pub fn cache_instruments(&self, instruments: &[InstrumentAny]) {
872        self.instruments.rcu(|m| {
873            for instrument in instruments {
874                m.insert(instrument.id(), instrument.clone());
875            }
876        });
877    }
878
879    /// Caches a client order for truncated ID resolution and instrument lookup.
880    ///
881    /// Kraken Futures limits client order IDs to 18 characters, so orders with
882    /// longer IDs are truncated. This method stores the mapping from truncated
883    /// to full ID, and from venue order ID to instrument ID for cancel messages.
884    pub fn cache_client_order(
885        &self,
886        client_order_id: ClientOrderId,
887        venue_order_id: Option<VenueOrderId>,
888        instrument_id: InstrumentId,
889        _trader_id: TraderId,
890        _strategy_id: StrategyId,
891    ) {
892        let truncated = truncate_cl_ord_id(&client_order_id);
893
894        if truncated != client_order_id.as_str() {
895            self.truncated_id_map.insert(truncated, client_order_id);
896        }
897
898        if let Some(venue_id) = venue_order_id {
899            self.order_instrument_map
900                .insert(venue_id.to_string(), instrument_id);
901        }
902    }
903
904    /// Subscribes to open orders feed (private, requires authentication).
905    pub async fn subscribe_open_orders(&self) -> Result<(), KrakenWsError> {
906        let key = "open_orders";
907        if !self.subscriptions.add_reference(key) {
908            return Ok(());
909        }
910
911        self.subscriptions.mark_subscribe(key);
912        let payload = self
913            .send_private_subscribe_feed(KrakenFuturesFeed::OpenOrders)
914            .await?;
915        self.subscriptions.confirm_subscribe(key);
916        self.subscription_payloads
917            .write()
918            .await
919            .insert(key.to_string(), payload);
920        Ok(())
921    }
922
923    /// Subscribes to fills feed (private, requires authentication).
924    pub async fn subscribe_fills(&self) -> Result<(), KrakenWsError> {
925        let key = "fills";
926        if !self.subscriptions.add_reference(key) {
927            return Ok(());
928        }
929
930        self.subscriptions.mark_subscribe(key);
931        let payload = self
932            .send_private_subscribe_feed(KrakenFuturesFeed::Fills)
933            .await?;
934        self.subscriptions.confirm_subscribe(key);
935        self.subscription_payloads
936            .write()
937            .await
938            .insert(key.to_string(), payload);
939        Ok(())
940    }
941
942    /// Subscribes to both open orders and fills (convenience method).
943    pub async fn subscribe_executions(&self) -> Result<(), KrakenWsError> {
944        self.subscribe_open_orders().await?;
945        self.subscribe_fills().await?;
946        Ok(())
947    }
948
949    async fn send_subscribe_feed(
950        &self,
951        feed: KrakenFuturesFeed,
952        product_ids: Vec<String>,
953    ) -> Result<String, KrakenWsError> {
954        let request = KrakenFuturesRequest {
955            event: KrakenFuturesEvent::Subscribe,
956            feed,
957            product_ids,
958        };
959        let payload =
960            serde_json::to_string(&request).map_err(|e| KrakenWsError::JsonError(e.to_string()))?;
961        self.cmd_tx
962            .read()
963            .await
964            .send(FuturesHandlerCommand::Subscribe {
965                payload: payload.clone(),
966            })
967            .map_err(|e| KrakenWsError::ChannelError(e.to_string()))?;
968        Ok(payload)
969    }
970
971    async fn send_unsubscribe_feed(
972        &self,
973        feed: KrakenFuturesFeed,
974        product_ids: Vec<String>,
975    ) -> Result<(), KrakenWsError> {
976        let request = KrakenFuturesRequest {
977            event: KrakenFuturesEvent::Unsubscribe,
978            feed,
979            product_ids,
980        };
981        let payload =
982            serde_json::to_string(&request).map_err(|e| KrakenWsError::JsonError(e.to_string()))?;
983        self.cmd_tx
984            .read()
985            .await
986            .send(FuturesHandlerCommand::Unsubscribe { payload })
987            .map_err(|e| KrakenWsError::ChannelError(e.to_string()))?;
988        Ok(())
989    }
990
991    async fn send_private_subscribe_feed(
992        &self,
993        feed: KrakenFuturesFeed,
994    ) -> Result<String, KrakenWsError> {
995        let credential = self.credential.as_ref().ok_or_else(|| {
996            KrakenWsError::AuthenticationError("API credentials required".to_string())
997        })?;
998        let original_challenge = self
999            .original_challenge
1000            .read()
1001            .await
1002            .clone()
1003            .ok_or_else(|| {
1004                KrakenWsError::AuthenticationError(
1005                    "Must authenticate before subscribing to private feeds".to_string(),
1006                )
1007            })?;
1008        let signed_challenge = self.signed_challenge.read().await.clone().ok_or_else(|| {
1009            KrakenWsError::AuthenticationError(
1010                "Must authenticate before subscribing to private feeds".to_string(),
1011            )
1012        })?;
1013
1014        let request = KrakenFuturesPrivateSubscribeRequest {
1015            event: KrakenFuturesEvent::Subscribe,
1016            feed,
1017            api_key: credential.api_key().to_string(),
1018            original_challenge,
1019            signed_challenge,
1020        };
1021        let payload =
1022            serde_json::to_string(&request).map_err(|e| KrakenWsError::JsonError(e.to_string()))?;
1023        self.cmd_tx
1024            .read()
1025            .await
1026            .send(FuturesHandlerCommand::Subscribe {
1027                payload: payload.clone(),
1028            })
1029            .map_err(|e| KrakenWsError::ChannelError(e.to_string()))?;
1030        Ok(payload)
1031    }
1032}
1033
1034fn update_private_payload_credentials(
1035    payload: &str,
1036    api_key: &str,
1037    original_challenge: &str,
1038    signed_challenge: &str,
1039) -> Option<String> {
1040    let mut value: serde_json::Value = serde_json::from_str(payload).ok()?;
1041    let obj = value.as_object_mut()?;
1042    obj.insert(
1043        "api_key".to_string(),
1044        serde_json::Value::String(api_key.to_string()),
1045    );
1046    obj.insert(
1047        "original_challenge".to_string(),
1048        serde_json::Value::String(original_challenge.to_string()),
1049    );
1050    obj.insert(
1051        "signed_challenge".to_string(),
1052        serde_json::Value::String(signed_challenge.to_string()),
1053    );
1054    serde_json::to_string(&value).ok()
1055}
1056
1057fn build_challenge_payload(credential: &KrakenCredential) -> serde_json::Result<String> {
1058    let request = KrakenFuturesChallengeRequest {
1059        event: KrakenFuturesEvent::Challenge,
1060        api_key: credential.api_key().to_string(),
1061    };
1062    serde_json::to_string(&request)
1063}
1064
1065fn is_private_feed_key(key: &str) -> bool {
1066    key == "open_orders" || key == "fills"
1067}
1068
1069fn resubscribe_public(
1070    cmd_tx: &tokio::sync::mpsc::UnboundedSender<FuturesHandlerCommand>,
1071    subscriptions: &SubscriptionState,
1072    payloads: &HashMap<String, String>,
1073) {
1074    for (key, payload) in payloads {
1075        if is_private_feed_key(key) {
1076            continue;
1077        }
1078
1079        if let Err(e) = cmd_tx.send(FuturesHandlerCommand::Subscribe {
1080            payload: payload.clone(),
1081        }) {
1082            log::error!("Failed to send resubscribe: error={e}, topic={key}");
1083            continue;
1084        }
1085
1086        subscriptions.mark_subscribe(key);
1087    }
1088}
1089
1090fn resubscribe_private(
1091    cmd_tx: &tokio::sync::mpsc::UnboundedSender<FuturesHandlerCommand>,
1092    subscriptions: &SubscriptionState,
1093    payloads: &HashMap<String, String>,
1094    credential: &KrakenCredential,
1095    original_challenge: &str,
1096    signed_challenge: &str,
1097) {
1098    for (key, payload) in payloads {
1099        if !is_private_feed_key(key) {
1100            continue;
1101        }
1102
1103        let Some(updated) = update_private_payload_credentials(
1104            payload,
1105            credential.api_key(),
1106            original_challenge,
1107            signed_challenge,
1108        ) else {
1109            log::error!("Failed to update private payload for {key}");
1110            continue;
1111        };
1112
1113        if let Err(e) = cmd_tx.send(FuturesHandlerCommand::Subscribe { payload: updated }) {
1114            log::error!("Failed to send resubscribe: error={e}, topic={key}");
1115            continue;
1116        }
1117
1118        subscriptions.mark_subscribe(key);
1119    }
1120}
1121
1122#[cfg(test)]
1123mod tests {
1124    use base64::{Engine, engine::general_purpose::STANDARD};
1125    use nautilus_network::websocket::AuthTracker;
1126    use rstest::rstest;
1127
1128    use super::*;
1129
1130    fn test_credential() -> KrakenCredential {
1131        let secret = STANDARD.encode(b"test_secret_key_24bytes!");
1132        KrakenCredential::new("test_key", secret)
1133    }
1134
1135    #[rstest]
1136    fn test_build_challenge_payload_emits_expected_event() {
1137        let credential = test_credential();
1138        let payload = build_challenge_payload(&credential).expect("serializes");
1139        assert!(payload.contains(r#""event":"challenge""#));
1140        assert!(payload.contains(r#""api_key":"test_key""#));
1141    }
1142
1143    #[rstest]
1144    #[tokio::test]
1145    async fn test_resubscribe_public_skips_private_feeds() {
1146        let (cmd_tx, mut cmd_rx) = tokio::sync::mpsc::unbounded_channel::<FuturesHandlerCommand>();
1147        let subscriptions = SubscriptionState::new(KRAKEN_FUTURES_WS_TOPIC_DELIMITER);
1148
1149        let mut payloads = HashMap::new();
1150        payloads.insert(
1151            "trades:PI_XBTUSD".to_string(),
1152            r#"{"event":"subscribe","feed":"trade","product_ids":["PI_XBTUSD"]}"#.to_string(),
1153        );
1154        payloads.insert(
1155            "open_orders".to_string(),
1156            r#"{"event":"subscribe","feed":"open_orders"}"#.to_string(),
1157        );
1158
1159        resubscribe_public(&cmd_tx, &subscriptions, &payloads);
1160
1161        let mut subscribed = Vec::new();
1162        while let Ok(FuturesHandlerCommand::Subscribe { payload }) = cmd_rx.try_recv() {
1163            subscribed.push(payload);
1164        }
1165
1166        assert_eq!(
1167            subscribed.len(),
1168            1,
1169            "only the public feed should resubscribe"
1170        );
1171        assert!(subscribed[0].contains("PI_XBTUSD"));
1172    }
1173
1174    #[rstest]
1175    #[tokio::test]
1176    async fn test_resubscribe_public_restores_publics_even_with_credentialed_client() {
1177        // The reconnect path runs resubscribe_public() unconditionally, so a
1178        // credentialed client's public feeds keep flowing even if re-auth fails.
1179        let (cmd_tx, mut cmd_rx) = tokio::sync::mpsc::unbounded_channel::<FuturesHandlerCommand>();
1180        let subscriptions = SubscriptionState::new(KRAKEN_FUTURES_WS_TOPIC_DELIMITER);
1181
1182        let mut payloads = HashMap::new();
1183        payloads.insert(
1184            "trades:PI_XBTUSD".to_string(),
1185            r#"{"event":"subscribe","feed":"trade","product_ids":["PI_XBTUSD"]}"#.to_string(),
1186        );
1187
1188        resubscribe_public(&cmd_tx, &subscriptions, &payloads);
1189
1190        match cmd_rx.try_recv().expect("public subscribe expected") {
1191            FuturesHandlerCommand::Subscribe { payload } => {
1192                assert!(payload.contains("PI_XBTUSD"));
1193            }
1194            other => panic!("expected Subscribe, was {other:?}"),
1195        }
1196    }
1197
1198    #[rstest]
1199    #[tokio::test]
1200    async fn test_resubscribe_private_patches_credentials() {
1201        let (cmd_tx, mut cmd_rx) = tokio::sync::mpsc::unbounded_channel::<FuturesHandlerCommand>();
1202        let subscriptions = SubscriptionState::new(KRAKEN_FUTURES_WS_TOPIC_DELIMITER);
1203        let credential = test_credential();
1204
1205        let mut payloads = HashMap::new();
1206        payloads.insert(
1207            "open_orders".to_string(),
1208            r#"{"event":"subscribe","feed":"open_orders","api_key":"","original_challenge":"","signed_challenge":""}"#.to_string(),
1209        );
1210        payloads.insert(
1211            "trades:PI_XBTUSD".to_string(),
1212            r#"{"event":"subscribe","feed":"trade","product_ids":["PI_XBTUSD"]}"#.to_string(),
1213        );
1214
1215        resubscribe_private(
1216            &cmd_tx,
1217            &subscriptions,
1218            &payloads,
1219            &credential,
1220            "server-challenge",
1221            "signed-value",
1222        );
1223
1224        let mut subscribed = Vec::new();
1225        while let Ok(FuturesHandlerCommand::Subscribe { payload }) = cmd_rx.try_recv() {
1226            subscribed.push(payload);
1227        }
1228
1229        assert_eq!(
1230            subscribed.len(),
1231            1,
1232            "only the private feed should resubscribe"
1233        );
1234        let value: serde_json::Value =
1235            serde_json::from_str(&subscribed[0]).expect("payload is valid JSON");
1236        assert_eq!(value["event"], "subscribe");
1237        assert_eq!(value["feed"], "open_orders");
1238        assert_eq!(value["api_key"], "test_key");
1239        assert_eq!(value["original_challenge"], "server-challenge");
1240        assert_eq!(value["signed_challenge"], "signed-value");
1241    }
1242
1243    #[rstest]
1244    #[tokio::test]
1245    async fn test_auth_tracker_succeed_completes_wait_for_result() {
1246        let tracker = AuthTracker::new();
1247        let receiver = tracker.begin();
1248
1249        let tracker_for_responder = tracker.clone();
1250
1251        tokio::spawn(async move {
1252            tokio::time::sleep(tokio::time::Duration::from_millis(20)).await;
1253            tracker_for_responder.succeed();
1254        });
1255
1256        tracker
1257            .wait_for_result::<KrakenWsError>(tokio::time::Duration::from_secs(1), receiver)
1258            .await
1259            .expect("auth should succeed");
1260
1261        assert!(tracker.is_authenticated());
1262    }
1263
1264    #[rstest]
1265    #[tokio::test]
1266    async fn test_auth_tracker_wait_for_result_times_out() {
1267        let tracker = AuthTracker::new();
1268        let receiver = tracker.begin();
1269
1270        let err = tracker
1271            .wait_for_result::<KrakenWsError>(tokio::time::Duration::from_millis(20), receiver)
1272            .await
1273            .expect_err("should time out");
1274
1275        assert!(matches!(err, KrakenWsError::AuthenticationError(_)));
1276        assert!(!tracker.is_authenticated());
1277    }
1278
1279    #[rstest]
1280    #[tokio::test]
1281    async fn test_authenticate_without_credentials_errors() {
1282        let client = KrakenFuturesWebSocketClient::new(
1283            "wss://futures.kraken.com/ws/v1".to_string(),
1284            60,
1285            None,
1286        );
1287
1288        let err = client.authenticate().await.expect_err("should fail");
1289        assert!(
1290            matches!(err, KrakenWsError::AuthenticationError(ref msg) if msg.contains("API credentials required")),
1291            "unexpected error: {err:?}"
1292        );
1293    }
1294
1295    #[rstest]
1296    #[tokio::test]
1297    async fn test_set_auth_credentials_marks_tracker_authenticated() {
1298        let client = KrakenFuturesWebSocketClient::with_credentials(
1299            "wss://futures.kraken.com/ws/v1".to_string(),
1300            60,
1301            Some(test_credential()),
1302            TransportBackend::default(),
1303            None,
1304        );
1305
1306        assert!(!client.is_authenticated());
1307
1308        client
1309            .set_auth_credentials("orig-challenge".to_string(), "signed-challenge".to_string())
1310            .await
1311            .expect("should succeed");
1312
1313        assert!(client.is_authenticated());
1314        client
1315            .wait_until_authenticated(0.05)
1316            .await
1317            .expect("should return immediately");
1318    }
1319
1320    #[rstest]
1321    #[tokio::test]
1322    async fn test_set_auth_credentials_without_credentials_errors() {
1323        let client = KrakenFuturesWebSocketClient::new(
1324            "wss://futures.kraken.com/ws/v1".to_string(),
1325            60,
1326            None,
1327        );
1328
1329        let err = client
1330            .set_auth_credentials("orig".to_string(), "signed".to_string())
1331            .await
1332            .expect_err("should fail");
1333        assert!(matches!(err, KrakenWsError::AuthenticationError(_)));
1334        assert!(!client.is_authenticated());
1335    }
1336
1337    #[rstest]
1338    #[tokio::test]
1339    async fn test_authenticate_with_challenge_updates_state() {
1340        let client = KrakenFuturesWebSocketClient::with_credentials(
1341            "wss://futures.kraken.com/ws/v1".to_string(),
1342            60,
1343            Some(test_credential()),
1344            TransportBackend::default(),
1345            None,
1346        );
1347
1348        client
1349            .authenticate_with_challenge("server-challenge")
1350            .await
1351            .expect("should succeed");
1352
1353        assert!(client.is_authenticated());
1354    }
1355
1356    #[rstest]
1357    #[tokio::test]
1358    async fn test_wait_until_authenticated_resolves_after_success() {
1359        let client = KrakenFuturesWebSocketClient::with_credentials(
1360            "wss://futures.kraken.com/ws/v1".to_string(),
1361            60,
1362            Some(test_credential()),
1363            TransportBackend::default(),
1364            None,
1365        );
1366
1367        let client_for_responder = client.clone();
1368
1369        tokio::spawn(async move {
1370            tokio::time::sleep(tokio::time::Duration::from_millis(20)).await;
1371            client_for_responder
1372                .set_auth_credentials("orig".to_string(), "signed".to_string())
1373                .await
1374                .expect("succeeds");
1375        });
1376
1377        client
1378            .wait_until_authenticated(1.0)
1379            .await
1380            .expect("should resolve once credentials are set");
1381    }
1382
1383    #[rstest]
1384    #[tokio::test]
1385    async fn test_wait_until_authenticated_times_out() {
1386        let client = KrakenFuturesWebSocketClient::with_credentials(
1387            "wss://futures.kraken.com/ws/v1".to_string(),
1388            60,
1389            Some(test_credential()),
1390            TransportBackend::default(),
1391            None,
1392        );
1393
1394        let err = client
1395            .wait_until_authenticated(0.05)
1396            .await
1397            .expect_err("should time out");
1398        assert!(matches!(err, KrakenWsError::AuthenticationError(_)));
1399    }
1400}
nautilus_kraken/websocket/futures/client.rs

nautilus_kraken/websocket/futures/
client.rs
