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

nautilus_kraken/websocket/futures/
client.rs
