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

nautilus_kraken/websocket/futures/
client.rs
