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

nautilus_kraken/websocket/futures/
client.rs
