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        let connection_mode = ws_client.connection_mode_atomic();
368        let reconnect_handle = ws_client.reconnect_handle();
369
370        let (out_tx, out_rx) = tokio::sync::mpsc::unbounded_channel::<KrakenFuturesWsMessage>();
371        self.out_rx = Some(Arc::new(out_rx));
372
373        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel::<FuturesHandlerCommand>();
374
375        if let Err(e) = cmd_tx.send(FuturesHandlerCommand::SetClient(ws_client)) {
376            return Err(KrakenWsError::ConnectionError(format!(
377                "Failed to send WebSocketClient to handler: {e}"
378            )));
379        }
380
381        {
382            let mut command_sender = self.cmd_tx.write().await;
383            *command_sender = cmd_tx.clone();
384            self.connection_mode.store(connection_mode);
385        }
386
387        if let Some(control) = &self.socket_control {
388            control.register(move || reconnect_handle.request_reconnect());
389        }
390
391        let signal = self.signal.clone();
392        let subscriptions = self.subscriptions.clone();
393        let subscription_payloads = self.subscription_payloads.clone();
394        let cmd_tx_for_reconnect = cmd_tx.clone();
395        let credential_for_reconnect = self.credential.clone();
396        let original_challenge_for_reconnect = self.original_challenge.clone();
397        let signed_challenge_for_reconnect = self.signed_challenge.clone();
398        let auth_tracker_for_reconnect = self.auth_tracker.clone();
399
400        let handler_task = async move {
401            let mut handler =
402                FuturesFeedHandler::new(signal.clone(), cmd_rx, raw_rx, subscriptions.clone());
403            let mut pending_resubscribe = false;
404
405            loop {
406                match handler.next().await {
407                    Some(KrakenFuturesWsMessage::Reconnected) => {
408                        if signal.load(Ordering::Relaxed) {
409                            continue;
410                        }
411                        log::info!("WebSocket reconnected");
412
413                        subscriptions.reset_after_reconnect();
414
415                        auth_tracker_for_reconnect.invalidate();
416                        *original_challenge_for_reconnect.write().await = None;
417                        *signed_challenge_for_reconnect.write().await = None;
418
419                        let payloads = subscription_payloads.read().await.clone();
420
421                        // Resubscribe public topics straight away; they don't depend on auth,
422                        // so don't tie their restoration to the challenge outcome.
423                        resubscribe_public(&cmd_tx_for_reconnect, &subscriptions, &payloads);
424
425                        let has_private =
426                            payloads.keys().any(|k| k == "open_orders" || k == "fills");
427
428                        pending_resubscribe = false;
429
430                        if has_private {
431                            if let Some(ref cred) = credential_for_reconnect {
432                                match build_challenge_payload(cred) {
433                                    Ok(payload) => {
434                                        let _rx = auth_tracker_for_reconnect.begin();
435
436                                        if let Err(e) = cmd_tx_for_reconnect.send(
437                                            FuturesHandlerCommand::RequestChallenge { payload },
438                                        ) {
439                                            log::error!("Failed to queue reconnect challenge: {e}");
440                                        } else {
441                                            pending_resubscribe = true;
442                                        }
443                                    }
444                                    Err(e) => {
445                                        log::error!("Failed to serialize reconnect challenge: {e}");
446                                    }
447                                }
448                            } else {
449                                log::warn!(
450                                    "Private subscriptions exist but no credentials available"
451                                );
452                            }
453                        }
454
455                        if let Err(e) = out_tx.send(KrakenFuturesWsMessage::Reconnected) {
456                            log::debug!("Output channel closed: {e}");
457                            break;
458                        }
459                    }
460                    Some(KrakenFuturesWsMessage::Challenge(challenge)) => {
461                        let Some(ref cred) = credential_for_reconnect else {
462                            log::warn!("Challenge received but no credentials configured");
463                            auth_tracker_for_reconnect.fail("no credentials");
464                            continue;
465                        };
466
467                        let challenge = SecretString::from(challenge);
468                        match cred
469                            .sign_ws_challenge(challenge.expose_secret())
470                            .map(SecretString::from)
471                        {
472                            Ok(signed) => {
473                                *original_challenge_for_reconnect.write().await =
474                                    Some(challenge.clone());
475                                *signed_challenge_for_reconnect.write().await =
476                                    Some(signed.clone());
477                                auth_tracker_for_reconnect.succeed();
478                                log::debug!("Signed WebSocket challenge");
479
480                                if pending_resubscribe {
481                                    let payloads = subscription_payloads.read().await;
482                                    resubscribe_private(
483                                        &cmd_tx_for_reconnect,
484                                        &subscriptions,
485                                        &payloads,
486                                        cred,
487                                        challenge.expose_secret(),
488                                        signed.expose_secret(),
489                                    );
490                                    pending_resubscribe = false;
491                                }
492                            }
493                            Err(e) => {
494                                log::error!("Failed to sign challenge: {e}");
495                                auth_tracker_for_reconnect.fail(e.to_string());
496                                pending_resubscribe = false;
497                            }
498                        }
499                    }
500                    Some(msg) => {
501                        if let Err(e) = out_tx.send(msg) {
502                            log::debug!("Output channel closed: {e}");
503                            break;
504                        }
505                    }
506                    None => {
507                        log::debug!("Handler stream ended");
508                        break;
509                    }
510                }
511            }
512
513            log::debug!("Futures handler task exiting");
514        };
515
516        if let Err(e) = handler_spawner.spawn(handler_task) {
517            if let Some(control) = &self.socket_control {
518                control.deregister();
519            }
520            self.out_rx = None;
521            return Err(KrakenWsError::ConnectionError(format!(
522                "Failed to register WebSocket handler task: {e}"
523            )));
524        }
525
526        log::debug!("Futures WebSocket connected successfully");
527        Ok(())
528    }
529
530    /// Disconnects from the WebSocket server.
531    pub async fn disconnect(&mut self) -> Result<(), KrakenWsError> {
532        let connect_lock = Arc::clone(&self.connect_lock);
533        let _connect_guard = connect_lock.lock().await;
534        self.disconnect_locked().await
535    }
536
537    async fn disconnect_locked(&self) -> Result<(), KrakenWsError> {
538        log::debug!("Disconnecting Futures WebSocket");
539
540        self.handler_tasks.begin_shutdown();
541        self.signal.store(true, Ordering::Relaxed);
542
543        if let Err(e) = self
544            .cmd_tx
545            .read()
546            .await
547            .send(FuturesHandlerCommand::Disconnect)
548        {
549            log::debug!(
550                "Failed to send disconnect command (handler may already be shut down): {e}"
551            );
552        }
553
554        let task_result = self
555            .handler_tasks
556            .finish_shutdown(
557                tokio::time::Duration::from_secs(2),
558                tokio::time::Duration::from_secs(2),
559            )
560            .await;
561
562        self.subscriptions.clear();
563        self.subscription_payloads.write().await.clear();
564        self.auth_tracker.fail("Disconnected");
565
566        if let Some(control) = &self.socket_control {
567            control.deregister();
568        }
569
570        match task_result {
571            Ok(()) => Ok(()),
572            Err(error @ TaskShutdownError::Timeout { .. }) => Err(KrakenWsError::Timeout(format!(
573                "Futures WebSocket handler shutdown timed out: {error}"
574            ))),
575            Err(e) => Err(KrakenWsError::Disconnected(format!(
576                "Futures WebSocket handler shutdown failed: {e}"
577            ))),
578        }
579    }
580
581    /// Closes the WebSocket connection.
582    pub async fn close(&mut self) -> Result<(), KrakenWsError> {
583        self.disconnect().await
584    }
585
586    /// Subscribes to mark price updates for the given instrument.
587    pub async fn subscribe_mark_price(
588        &self,
589        instrument_id: InstrumentId,
590    ) -> Result<(), KrakenWsError> {
591        let symbol = instrument_id.symbol;
592        let key = format!("mark:{symbol}");
593
594        if !self.subscriptions.add_reference(&key) {
595            return Ok(());
596        }
597
598        self.subscriptions.mark_subscribe(&key);
599        self.subscriptions.confirm_subscribe(&key);
600        self.ensure_ticker_subscribed(symbol).await
601    }
602
603    /// Unsubscribes from mark price updates for the given instrument.
604    pub async fn unsubscribe_mark_price(
605        &self,
606        instrument_id: InstrumentId,
607    ) -> Result<(), KrakenWsError> {
608        let symbol = instrument_id.symbol;
609        let key = format!("mark:{symbol}");
610
611        if !self.subscriptions.remove_reference(&key) {
612            return Ok(());
613        }
614
615        self.subscriptions.mark_unsubscribe(&key);
616        self.subscriptions.confirm_unsubscribe(&key);
617        self.maybe_unsubscribe_ticker(symbol).await
618    }
619
620    /// Subscribes to index price updates for the given instrument.
621    pub async fn subscribe_index_price(
622        &self,
623        instrument_id: InstrumentId,
624    ) -> Result<(), KrakenWsError> {
625        let symbol = instrument_id.symbol;
626        let key = format!("index:{symbol}");
627
628        if !self.subscriptions.add_reference(&key) {
629            return Ok(());
630        }
631
632        self.subscriptions.mark_subscribe(&key);
633        self.subscriptions.confirm_subscribe(&key);
634        self.ensure_ticker_subscribed(symbol).await
635    }
636
637    /// Unsubscribes from index price updates for the given instrument.
638    pub async fn unsubscribe_index_price(
639        &self,
640        instrument_id: InstrumentId,
641    ) -> Result<(), KrakenWsError> {
642        let symbol = instrument_id.symbol;
643        let key = format!("index:{symbol}");
644
645        if !self.subscriptions.remove_reference(&key) {
646            return Ok(());
647        }
648
649        self.subscriptions.mark_unsubscribe(&key);
650        self.subscriptions.confirm_unsubscribe(&key);
651        self.maybe_unsubscribe_ticker(symbol).await
652    }
653
654    /// Subscribes to funding rate updates for the given instrument.
655    pub async fn subscribe_funding_rate(
656        &self,
657        instrument_id: InstrumentId,
658    ) -> Result<(), KrakenWsError> {
659        let symbol = instrument_id.symbol;
660        let key = format!("funding:{symbol}");
661
662        if !self.subscriptions.add_reference(&key) {
663            return Ok(());
664        }
665
666        self.subscriptions.mark_subscribe(&key);
667        self.subscriptions.confirm_subscribe(&key);
668        self.ensure_ticker_subscribed(symbol).await
669    }
670
671    /// Unsubscribes from funding rate updates for the given instrument.
672    pub async fn unsubscribe_funding_rate(
673        &self,
674        instrument_id: InstrumentId,
675    ) -> Result<(), KrakenWsError> {
676        let symbol = instrument_id.symbol;
677        let key = format!("funding:{symbol}");
678
679        if !self.subscriptions.remove_reference(&key) {
680            return Ok(());
681        }
682
683        self.subscriptions.mark_unsubscribe(&key);
684        self.subscriptions.confirm_unsubscribe(&key);
685        self.maybe_unsubscribe_ticker(symbol).await
686    }
687
688    /// Subscribes to quote updates for the given instrument.
689    ///
690    /// Uses the order book channel for low-latency top-of-book quotes.
691    pub async fn subscribe_quotes(&self, instrument_id: InstrumentId) -> Result<(), KrakenWsError> {
692        let symbol = instrument_id.symbol;
693        let key = format!("quotes:{symbol}");
694
695        if !self.subscriptions.add_reference(&key) {
696            return Ok(());
697        }
698
699        self.subscriptions.mark_subscribe(&key);
700        self.subscriptions.confirm_subscribe(&key);
701
702        // Use book feed for low-latency quotes (not throttled ticker)
703        self.ensure_book_subscribed(symbol).await
704    }
705
706    /// Unsubscribes from quote updates for the given instrument.
707    pub async fn unsubscribe_quotes(
708        &self,
709        instrument_id: InstrumentId,
710    ) -> Result<(), KrakenWsError> {
711        let symbol = instrument_id.symbol;
712        let key = format!("quotes:{symbol}");
713
714        if !self.subscriptions.remove_reference(&key) {
715            return Ok(());
716        }
717
718        self.subscriptions.mark_unsubscribe(&key);
719        self.subscriptions.confirm_unsubscribe(&key);
720        self.maybe_unsubscribe_book(symbol).await
721    }
722
723    /// Subscribes to trade updates for the given instrument.
724    pub async fn subscribe_trades(&self, instrument_id: InstrumentId) -> Result<(), KrakenWsError> {
725        let symbol = instrument_id.symbol;
726        let key = format!("trades:{symbol}");
727
728        if !self.subscriptions.add_reference(&key) {
729            return Ok(());
730        }
731
732        self.subscriptions.mark_subscribe(&key);
733        let payload = self
734            .send_subscribe_feed(KrakenFuturesFeed::Trade, vec![symbol.to_string()])
735            .await?;
736        self.subscriptions.confirm_subscribe(&key);
737        self.subscription_payloads
738            .write()
739            .await
740            .insert(key, payload);
741        Ok(())
742    }
743
744    /// Unsubscribes from trade updates for the given instrument.
745    pub async fn unsubscribe_trades(
746        &self,
747        instrument_id: InstrumentId,
748    ) -> Result<(), KrakenWsError> {
749        let symbol = instrument_id.symbol;
750        let key = format!("trades:{symbol}");
751
752        if !self.subscriptions.remove_reference(&key) {
753            return Ok(());
754        }
755
756        self.subscriptions.mark_unsubscribe(&key);
757        self.send_unsubscribe_feed(KrakenFuturesFeed::Trade, vec![symbol.to_string()])
758            .await?;
759        self.subscriptions.confirm_unsubscribe(&key);
760        self.subscription_payloads.write().await.remove(&key);
761        Ok(())
762    }
763
764    /// Subscribes to order book updates for the given instrument.
765    ///
766    /// Note: The `depth` parameter is accepted for API compatibility with spot client but is
767    /// not used by Kraken Futures (full book is always returned).
768    pub async fn subscribe_book(
769        &self,
770        instrument_id: InstrumentId,
771        _depth: Option<u32>,
772    ) -> Result<(), KrakenWsError> {
773        let symbol = instrument_id.symbol;
774
775        let deltas_key = format!("deltas:{symbol}");
776        self.subscriptions.add_reference(&deltas_key);
777        self.subscriptions.mark_subscribe(&deltas_key);
778        self.subscriptions.confirm_subscribe(&deltas_key);
779
780        self.ensure_book_subscribed(symbol).await
781    }
782
783    /// Unsubscribes from order book updates for the given instrument.
784    pub async fn unsubscribe_book(&self, instrument_id: InstrumentId) -> Result<(), KrakenWsError> {
785        let symbol = instrument_id.symbol;
786
787        let deltas_key = format!("deltas:{symbol}");
788        self.subscriptions.remove_reference(&deltas_key);
789        self.subscriptions.mark_unsubscribe(&deltas_key);
790        self.subscriptions.confirm_unsubscribe(&deltas_key);
791
792        self.maybe_unsubscribe_book(symbol).await
793    }
794
795    async fn ensure_ticker_subscribed(&self, symbol: Symbol) -> Result<(), KrakenWsError> {
796        let ticker_key = format!("ticker:{symbol}");
797
798        if !self.subscriptions.add_reference(&ticker_key) {
799            return Ok(());
800        }
801
802        self.subscriptions.mark_subscribe(&ticker_key);
803        let payload = self
804            .send_subscribe_feed(KrakenFuturesFeed::Ticker, vec![symbol.to_string()])
805            .await?;
806        self.subscriptions.confirm_subscribe(&ticker_key);
807        self.subscription_payloads
808            .write()
809            .await
810            .insert(ticker_key, payload);
811        Ok(())
812    }
813
814    async fn maybe_unsubscribe_ticker(&self, symbol: Symbol) -> Result<(), KrakenWsError> {
815        let ticker_key = format!("ticker:{symbol}");
816
817        if !self.subscriptions.remove_reference(&ticker_key) {
818            return Ok(());
819        }
820
821        self.subscriptions.mark_unsubscribe(&ticker_key);
822        self.send_unsubscribe_feed(KrakenFuturesFeed::Ticker, vec![symbol.to_string()])
823            .await?;
824        self.subscriptions.confirm_unsubscribe(&ticker_key);
825        self.subscription_payloads.write().await.remove(&ticker_key);
826        Ok(())
827    }
828
829    async fn ensure_book_subscribed(&self, symbol: Symbol) -> Result<(), KrakenWsError> {
830        let book_key = format!("book:{symbol}");
831
832        if !self.subscriptions.add_reference(&book_key) {
833            return Ok(());
834        }
835
836        self.subscriptions.mark_subscribe(&book_key);
837        let payload = self
838            .send_subscribe_feed(KrakenFuturesFeed::Book, vec![symbol.to_string()])
839            .await?;
840        self.subscriptions.confirm_subscribe(&book_key);
841        self.subscription_payloads
842            .write()
843            .await
844            .insert(book_key, payload);
845        Ok(())
846    }
847
848    async fn maybe_unsubscribe_book(&self, symbol: Symbol) -> Result<(), KrakenWsError> {
849        let book_key = format!("book:{symbol}");
850
851        if !self.subscriptions.remove_reference(&book_key) {
852            return Ok(());
853        }
854
855        self.subscriptions.mark_unsubscribe(&book_key);
856        self.send_unsubscribe_feed(KrakenFuturesFeed::Book, vec![symbol.to_string()])
857            .await?;
858        self.subscriptions.confirm_unsubscribe(&book_key);
859        self.subscription_payloads.write().await.remove(&book_key);
860        Ok(())
861    }
862
863    /// Gets the output receiver for processed messages.
864    pub fn take_output_rx(
865        &mut self,
866    ) -> Option<tokio::sync::mpsc::UnboundedReceiver<KrakenFuturesWsMessage>> {
867        self.out_rx.take().and_then(|arc| Arc::try_unwrap(arc).ok())
868    }
869
870    /// Set authentication credentials directly (for when challenge is obtained externally).
871    pub async fn set_auth_credentials(
872        &self,
873        original_challenge: String,
874        signed_challenge: String,
875    ) -> Result<(), KrakenWsError> {
876        let _credential = self.credential.as_ref().ok_or_else(|| {
877            KrakenWsError::AuthenticationError("API credentials required".to_string())
878        })?;
879
880        let original_challenge = SecretString::from(original_challenge);
881        let signed_challenge = SecretString::from(signed_challenge);
882        *self.original_challenge.write().await = Some(original_challenge);
883        *self.signed_challenge.write().await = Some(signed_challenge);
884        self.auth_tracker.succeed();
885
886        Ok(())
887    }
888
889    /// Sign a challenge with the API credentials.
890    ///
891    /// Returns the signed challenge on success.
892    pub fn sign_challenge(&self, challenge: &str) -> Result<String, KrakenWsError> {
893        let credential = self.credential.as_ref().ok_or_else(|| {
894            KrakenWsError::AuthenticationError("API credentials required".to_string())
895        })?;
896
897        credential.sign_ws_challenge(challenge).map_err(|e| {
898            KrakenWsError::AuthenticationError(format!("Failed to sign challenge: {e}"))
899        })
900    }
901
902    /// Complete authentication with a received challenge.
903    pub async fn authenticate_with_challenge(&self, challenge: &str) -> Result<(), KrakenWsError> {
904        let credential = self.credential.as_ref().ok_or_else(|| {
905            KrakenWsError::AuthenticationError("API credentials required".to_string())
906        })?;
907
908        let signed_challenge = credential.sign_ws_challenge(challenge).map_err(|e| {
909            KrakenWsError::AuthenticationError(format!("Failed to sign challenge: {e}"))
910        })?;
911
912        self.set_auth_credentials(challenge.to_string(), signed_challenge)
913            .await
914    }
915
916    /// Sets the account ID for execution report parsing.
917    pub fn set_account_id(&self, account_id: AccountId) {
918        *self.account_id.write() = Some(account_id);
919    }
920
921    /// Returns the account ID if set.
922    #[must_use]
923    pub fn account_id(&self) -> Option<AccountId> {
924        *self.account_id.read()
925    }
926
927    /// Returns a reference to the shared account ID.
928    #[must_use]
929    pub fn account_id_shared(&self) -> &Arc<RwLock<Option<AccountId>>> {
930        &self.account_id
931    }
932
933    /// Returns a reference to the truncated ID map.
934    #[must_use]
935    pub fn truncated_id_map(&self) -> &Arc<AtomicMap<String, ClientOrderId>> {
936        &self.truncated_id_map
937    }
938
939    /// Returns a reference to the order-to-instrument map.
940    #[must_use]
941    pub fn order_instrument_map(&self) -> &Arc<AtomicMap<String, InstrumentId>> {
942        &self.order_instrument_map
943    }
944
945    /// Returns a reference to the shared instruments map.
946    #[must_use]
947    pub fn instruments_shared(&self) -> &Arc<AtomicMap<InstrumentId, InstrumentAny>> {
948        &self.instruments
949    }
950
951    /// Returns a reference to the subscription state.
952    #[must_use]
953    pub fn subscriptions(&self) -> &SubscriptionState {
954        &self.subscriptions
955    }
956
957    /// Caches an instrument for execution report parsing.
958    pub fn cache_instrument(&self, instrument: InstrumentAny) {
959        self.instruments.insert(instrument.id(), instrument);
960    }
961
962    /// Caches multiple instruments for execution report parsing.
963    pub fn cache_instruments(&self, instruments: &[InstrumentAny]) {
964        self.instruments.rcu(|m| {
965            for instrument in instruments {
966                m.insert(instrument.id(), instrument.clone());
967            }
968        });
969    }
970
971    /// Caches a client order for shortened ID resolution and instrument lookup.
972    ///
973    /// An order sent with its client order ID shortened by `truncate_cl_ord_id` is echoed in
974    /// that form. This method stores the mapping from the shortened to the full ID, and from
975    /// venue order ID to instrument ID for cancel messages.
976    pub fn cache_client_order(
977        &self,
978        client_order_id: ClientOrderId,
979        venue_order_id: Option<VenueOrderId>,
980        instrument_id: InstrumentId,
981        _trader_id: TraderId,
982        _strategy_id: StrategyId,
983    ) {
984        let truncated = truncate_cl_ord_id(&client_order_id);
985
986        if truncated != client_order_id.as_str() {
987            self.truncated_id_map.insert(truncated, client_order_id);
988        }
989
990        if let Some(venue_id) = venue_order_id {
991            self.order_instrument_map
992                .insert(venue_id.to_string(), instrument_id);
993        }
994    }
995
996    /// Subscribes to open orders feed (private, requires authentication).
997    pub async fn subscribe_open_orders(&self) -> Result<(), KrakenWsError> {
998        let key = "open_orders";
999        if !self.subscriptions.add_reference(key) {
1000            return Ok(());
1001        }
1002
1003        self.subscriptions.mark_subscribe(key);
1004        let payload = self
1005            .send_private_subscribe_feed(KrakenFuturesFeed::OpenOrders)
1006            .await?;
1007        self.subscriptions.confirm_subscribe(key);
1008        self.subscription_payloads
1009            .write()
1010            .await
1011            .insert(key.to_string(), payload);
1012        Ok(())
1013    }
1014
1015    /// Subscribes to fills feed (private, requires authentication).
1016    pub async fn subscribe_fills(&self) -> Result<(), KrakenWsError> {
1017        let key = "fills";
1018        if !self.subscriptions.add_reference(key) {
1019            return Ok(());
1020        }
1021
1022        self.subscriptions.mark_subscribe(key);
1023        let payload = self
1024            .send_private_subscribe_feed(KrakenFuturesFeed::Fills)
1025            .await?;
1026        self.subscriptions.confirm_subscribe(key);
1027        self.subscription_payloads
1028            .write()
1029            .await
1030            .insert(key.to_string(), payload);
1031        Ok(())
1032    }
1033
1034    /// Subscribes to both open orders and fills (convenience method).
1035    pub async fn subscribe_executions(&self) -> Result<(), KrakenWsError> {
1036        self.subscribe_open_orders().await?;
1037        self.subscribe_fills().await?;
1038        Ok(())
1039    }
1040
1041    async fn send_subscribe_feed(
1042        &self,
1043        feed: KrakenFuturesFeed,
1044        product_ids: Vec<String>,
1045    ) -> Result<SecretString, KrakenWsError> {
1046        let request = KrakenFuturesRequest {
1047            event: KrakenFuturesEvent::Subscribe,
1048            feed,
1049            product_ids,
1050        };
1051        let payload = SecretString::from(
1052            serde_json::to_string(&request).map_err(|e| KrakenWsError::JsonError(e.to_string()))?,
1053        );
1054        self.cmd_tx
1055            .read()
1056            .await
1057            .send(FuturesHandlerCommand::Subscribe {
1058                payload: payload.clone(),
1059            })
1060            .map_err(|e| KrakenWsError::ChannelError(e.to_string()))?;
1061        Ok(payload)
1062    }
1063
1064    async fn send_unsubscribe_feed(
1065        &self,
1066        feed: KrakenFuturesFeed,
1067        product_ids: Vec<String>,
1068    ) -> Result<(), KrakenWsError> {
1069        let request = KrakenFuturesRequest {
1070            event: KrakenFuturesEvent::Unsubscribe,
1071            feed,
1072            product_ids,
1073        };
1074        let payload = SecretString::from(
1075            serde_json::to_string(&request).map_err(|e| KrakenWsError::JsonError(e.to_string()))?,
1076        );
1077        self.cmd_tx
1078            .read()
1079            .await
1080            .send(FuturesHandlerCommand::Unsubscribe { payload })
1081            .map_err(|e| KrakenWsError::ChannelError(e.to_string()))?;
1082        Ok(())
1083    }
1084
1085    async fn send_private_subscribe_feed(
1086        &self,
1087        feed: KrakenFuturesFeed,
1088    ) -> Result<SecretString, KrakenWsError> {
1089        let credential = self.credential.as_ref().ok_or_else(|| {
1090            KrakenWsError::AuthenticationError("API credentials required".to_string())
1091        })?;
1092        let original_challenge = self
1093            .original_challenge
1094            .read()
1095            .await
1096            .clone()
1097            .ok_or_else(|| {
1098                KrakenWsError::AuthenticationError(
1099                    "Must authenticate before subscribing to private feeds".to_string(),
1100                )
1101            })?;
1102        let signed_challenge = self.signed_challenge.read().await.clone().ok_or_else(|| {
1103            KrakenWsError::AuthenticationError(
1104                "Must authenticate before subscribing to private feeds".to_string(),
1105            )
1106        })?;
1107
1108        let request = Zeroizing::new(KrakenFuturesPrivateSubscribeRequest {
1109            event: KrakenFuturesEvent::Subscribe,
1110            feed,
1111            api_key: credential.api_key().into(),
1112            original_challenge,
1113            signed_challenge,
1114        });
1115        let payload = SecretString::from(
1116            serde_json::to_string(&*request)
1117                .map_err(|e| KrakenWsError::JsonError(e.to_string()))?,
1118        );
1119        drop(request);
1120        self.cmd_tx
1121            .read()
1122            .await
1123            .send(FuturesHandlerCommand::Subscribe {
1124                payload: payload.clone(),
1125            })
1126            .map_err(|e| KrakenWsError::ChannelError(e.to_string()))?;
1127        Ok(payload)
1128    }
1129}
1130
1131fn update_private_payload_credentials(
1132    payload: &str,
1133    api_key: &str,
1134    original_challenge: &str,
1135    signed_challenge: &str,
1136) -> Option<SecretString> {
1137    let mut value: serde_json::Value = serde_json::from_str(payload).ok()?;
1138    let obj = value.as_object_mut()?;
1139    obj.insert(
1140        "api_key".to_string(),
1141        serde_json::Value::String(api_key.to_string()),
1142    );
1143    obj.insert(
1144        "original_challenge".to_string(),
1145        serde_json::Value::String(original_challenge.to_string()),
1146    );
1147    obj.insert(
1148        "signed_challenge".to_string(),
1149        serde_json::Value::String(signed_challenge.to_string()),
1150    );
1151    let payload = serde_json::to_string(&value).ok().map(SecretString::from);
1152    zeroize_json_value(&mut value);
1153    payload
1154}
1155
1156fn build_challenge_payload(credential: &KrakenCredential) -> serde_json::Result<SecretString> {
1157    let request = Zeroizing::new(KrakenFuturesChallengeRequest {
1158        event: KrakenFuturesEvent::Challenge,
1159        api_key: credential.api_key().into(),
1160    });
1161    serde_json::to_string(&*request).map(SecretString::from)
1162}
1163
1164fn is_private_feed_key(key: &str) -> bool {
1165    key == "open_orders" || key == "fills"
1166}
1167
1168fn resubscribe_public(
1169    cmd_tx: &tokio::sync::mpsc::UnboundedSender<FuturesHandlerCommand>,
1170    subscriptions: &SubscriptionState,
1171    payloads: &HashMap<String, SecretString>,
1172) {
1173    for (key, payload) in payloads {
1174        if is_private_feed_key(key) {
1175            continue;
1176        }
1177
1178        if let Err(e) = cmd_tx.send(FuturesHandlerCommand::Subscribe {
1179            payload: payload.clone(),
1180        }) {
1181            log::error!("Failed to send resubscribe: error={e}, topic={key}");
1182            continue;
1183        }
1184
1185        subscriptions.mark_subscribe(key);
1186    }
1187}
1188
1189fn resubscribe_private(
1190    cmd_tx: &tokio::sync::mpsc::UnboundedSender<FuturesHandlerCommand>,
1191    subscriptions: &SubscriptionState,
1192    payloads: &HashMap<String, SecretString>,
1193    credential: &KrakenCredential,
1194    original_challenge: &str,
1195    signed_challenge: &str,
1196) {
1197    for (key, payload) in payloads {
1198        if !is_private_feed_key(key) {
1199            continue;
1200        }
1201
1202        let Some(updated) = update_private_payload_credentials(
1203            payload.expose_secret(),
1204            credential.api_key(),
1205            original_challenge,
1206            signed_challenge,
1207        ) else {
1208            log::error!("Failed to update private payload for {key}");
1209            continue;
1210        };
1211
1212        if let Err(e) = cmd_tx.send(FuturesHandlerCommand::Subscribe { payload: updated }) {
1213            log::error!("Failed to send resubscribe: error={e}, topic={key}");
1214            continue;
1215        }
1216
1217        subscriptions.mark_subscribe(key);
1218    }
1219}
1220
1221#[cfg(test)]
1222mod tests {
1223    use std::time::Duration;
1224
1225    use base64::{Engine, engine::general_purpose::STANDARD};
1226    use futures_util::StreamExt;
1227    use nautilus_network::websocket::AuthTracker;
1228    use nautilus_testkit::connection::race_command_with_publication;
1229    use rstest::rstest;
1230
1231    use super::*;
1232
1233    struct DropSignal(Arc<AtomicBool>);
1234
1235    impl Drop for DropSignal {
1236        fn drop(&mut self) {
1237            self.0.store(true, Ordering::Release);
1238        }
1239    }
1240
1241    fn test_credential() -> KrakenCredential {
1242        let secret = STANDARD.encode(b"test_secret_key_24bytes!");
1243        KrakenCredential::new("test_key", secret)
1244    }
1245
1246    #[tokio::test]
1247    async fn command_during_sender_publication_reaches_initialized_handler() {
1248        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
1249        let addr = listener.local_addr().unwrap();
1250        let (received_tx, mut received_rx) = tokio::sync::mpsc::unbounded_channel();
1251
1252        let server = tokio::spawn(async move {
1253            let (socket, _) = listener.accept().await.unwrap();
1254            let mut socket = tokio_tungstenite::accept_async(socket).await.unwrap();
1255            while let Some(Ok(message)) = socket.next().await {
1256                if let tokio_tungstenite::tungstenite::Message::Text(text) = message {
1257                    let payload: serde_json::Value = serde_json::from_str(text.as_str()).unwrap();
1258                    let _ = received_tx.send(payload);
1259                }
1260            }
1261        });
1262
1263        let url = format!("ws://{addr}");
1264        let mut ws_client = KrakenFuturesWebSocketClient::new(url, 30, None);
1265        let observer = ws_client.clone();
1266        let expected = serde_json::json!({
1267            "event": "subscribe",
1268            "feed": "trade",
1269            "product_ids": [
1270                "PI_XBTUSD"
1271            ]
1272        });
1273
1274        let command = FuturesHandlerCommand::Subscribe {
1275            payload: SecretString::from(expected.to_string()),
1276        };
1277
1278        race_command_with_publication(
1279            Arc::clone(&ws_client.cmd_tx),
1280            ws_client.connect(),
1281            move || command,
1282            || observer.is_active(),
1283        )
1284        .await
1285        .expect("connect failed");
1286        let received = tokio::time::timeout(Duration::from_secs(2), received_rx.recv()).await;
1287        ws_client.close().await.expect("disconnect failed");
1288        server.abort();
1289
1290        let payload = received
1291            .expect("startup command must reach server")
1292            .unwrap();
1293        assert_eq!(payload, expected);
1294    }
1295
1296    #[rstest]
1297    #[case("wss://stream.example.com/feed", "wss://stream.example.com")]
1298    #[case(
1299        "wss://SYN-user:SYN-password@stream.example.com:9443/SYN-path?token=SYN-query#SYN-fragment",
1300        "wss://stream.example.com:9443"
1301    )]
1302    fn test_debug_preserves_url_origin(#[case] url: &str, #[case] origin: &str) {
1303        let client = KrakenFuturesWebSocketClient::new(url.to_string(), 30, None);
1304        let debug = format!("{client:?}");
1305        assert!(debug.contains(&format!("url: {origin:?}")));
1306        assert!(!debug.contains("SYN-"));
1307        assert_eq!(client.url.expose(), url);
1308    }
1309
1310    #[rstest]
1311    #[tokio::test]
1312    async fn test_debug_redacts_auth_state_and_proxy_url() {
1313        let client = KrakenFuturesWebSocketClient::with_credentials(
1314            "wss://test".to_string(),
1315            30,
1316            Some(test_credential()),
1317            None,
1318            TransportBackend::default(),
1319            Some("http://user:proxy-secret@localhost".to_string()),
1320        );
1321        client
1322            .set_auth_credentials(
1323                "original-challenge".to_string(),
1324                "signed-challenge".to_string(),
1325            )
1326            .await
1327            .unwrap();
1328
1329        let debug = format!("{client:?}");
1330
1331        assert!(debug.contains(REDACTED));
1332        assert!(!debug.contains("proxy-secret"));
1333        assert!(!debug.contains("original-challenge"));
1334        assert!(!debug.contains("signed-challenge"));
1335    }
1336
1337    #[tokio::test]
1338    async fn test_last_client_owner_drop_aborts_handler_task() {
1339        let client = KrakenFuturesWebSocketClient::new("wss://test".to_string(), 30, None);
1340        let (started_tx, started_rx) = tokio::sync::oneshot::channel();
1341        let dropped = Arc::new(AtomicBool::new(false));
1342        let drop_signal = DropSignal(Arc::clone(&dropped));
1343        client
1344            .handler_tasks
1345            .spawn(async move {
1346                let _drop_signal = drop_signal;
1347                started_tx.send(()).expect("started receiver");
1348                std::future::pending::<()>().await;
1349            })
1350            .expect("handler task should register");
1351        started_rx.await.expect("handler task started");
1352        let clone = client.clone();
1353
1354        drop(client);
1355        assert!(!dropped.load(Ordering::Acquire));
1356        drop(clone);
1357
1358        tokio::time::timeout(tokio::time::Duration::from_secs(1), async {
1359            while !dropped.load(Ordering::Acquire) {
1360                tokio::task::yield_now().await;
1361            }
1362        })
1363        .await
1364        .expect("handler task aborted");
1365    }
1366
1367    #[rstest]
1368    fn test_build_challenge_payload_emits_expected_event() {
1369        let credential = test_credential();
1370        let payload = build_challenge_payload(&credential).expect("serializes");
1371        assert!(payload.expose_secret().contains(r#""event":"challenge""#));
1372        assert!(payload.expose_secret().contains(r#""api_key":"test_key""#));
1373    }
1374
1375    #[rstest]
1376    #[tokio::test]
1377    async fn test_resubscribe_public_skips_private_feeds() {
1378        let (cmd_tx, mut cmd_rx) = tokio::sync::mpsc::unbounded_channel::<FuturesHandlerCommand>();
1379        let subscriptions = SubscriptionState::new(KRAKEN_FUTURES_WS_TOPIC_DELIMITER);
1380
1381        let mut payloads = HashMap::new();
1382        payloads.insert(
1383            "trades:PI_XBTUSD".to_string(),
1384            SecretString::from(
1385                r#"{"event":"subscribe","feed":"trade","product_ids":["PI_XBTUSD"]}"#.to_string(),
1386            ),
1387        );
1388        payloads.insert(
1389            "open_orders".to_string(),
1390            SecretString::from(r#"{"event":"subscribe","feed":"open_orders"}"#.to_string()),
1391        );
1392
1393        resubscribe_public(&cmd_tx, &subscriptions, &payloads);
1394
1395        let mut subscribed = Vec::new();
1396        while let Ok(FuturesHandlerCommand::Subscribe { payload }) = cmd_rx.try_recv() {
1397            subscribed.push(payload);
1398        }
1399
1400        assert_eq!(
1401            subscribed.len(),
1402            1,
1403            "only the public feed should resubscribe"
1404        );
1405        assert!(subscribed[0].expose_secret().contains("PI_XBTUSD"));
1406    }
1407
1408    #[rstest]
1409    #[tokio::test]
1410    async fn test_resubscribe_public_restores_publics_even_with_credentialed_client() {
1411        // The reconnect path runs resubscribe_public() unconditionally, so a
1412        // credentialed client's public feeds keep flowing even if re-auth fails.
1413        let (cmd_tx, mut cmd_rx) = tokio::sync::mpsc::unbounded_channel::<FuturesHandlerCommand>();
1414        let subscriptions = SubscriptionState::new(KRAKEN_FUTURES_WS_TOPIC_DELIMITER);
1415
1416        let mut payloads = HashMap::new();
1417        payloads.insert(
1418            "trades:PI_XBTUSD".to_string(),
1419            SecretString::from(
1420                r#"{"event":"subscribe","feed":"trade","product_ids":["PI_XBTUSD"]}"#.to_string(),
1421            ),
1422        );
1423
1424        resubscribe_public(&cmd_tx, &subscriptions, &payloads);
1425
1426        match cmd_rx.try_recv().expect("public subscribe expected") {
1427            FuturesHandlerCommand::Subscribe { payload } => {
1428                assert!(payload.expose_secret().contains("PI_XBTUSD"));
1429            }
1430            other => panic!("expected Subscribe, was {other:?}"),
1431        }
1432    }
1433
1434    #[rstest]
1435    #[tokio::test]
1436    async fn test_resubscribe_private_patches_credentials() {
1437        let (cmd_tx, mut cmd_rx) = tokio::sync::mpsc::unbounded_channel::<FuturesHandlerCommand>();
1438        let subscriptions = SubscriptionState::new(KRAKEN_FUTURES_WS_TOPIC_DELIMITER);
1439        let credential = test_credential();
1440
1441        let mut payloads = HashMap::new();
1442        payloads.insert(
1443            "open_orders".to_string(),
1444            SecretString::from(
1445                r#"{"event":"subscribe","feed":"open_orders","api_key":"","original_challenge":"","signed_challenge":""}"#
1446                    .to_string(),
1447            ),
1448        );
1449        payloads.insert(
1450            "trades:PI_XBTUSD".to_string(),
1451            SecretString::from(
1452                r#"{"event":"subscribe","feed":"trade","product_ids":["PI_XBTUSD"]}"#.to_string(),
1453            ),
1454        );
1455
1456        resubscribe_private(
1457            &cmd_tx,
1458            &subscriptions,
1459            &payloads,
1460            &credential,
1461            "server-challenge",
1462            "signed-value",
1463        );
1464
1465        let mut subscribed = Vec::new();
1466        while let Ok(FuturesHandlerCommand::Subscribe { payload }) = cmd_rx.try_recv() {
1467            subscribed.push(payload);
1468        }
1469
1470        assert_eq!(
1471            subscribed.len(),
1472            1,
1473            "only the private feed should resubscribe"
1474        );
1475        let value: serde_json::Value =
1476            serde_json::from_str(subscribed[0].expose_secret()).expect("payload is valid JSON");
1477        assert_eq!(value["event"], "subscribe");
1478        assert_eq!(value["feed"], "open_orders");
1479        assert_eq!(value["api_key"], "test_key");
1480        assert_eq!(value["original_challenge"], "server-challenge");
1481        assert_eq!(value["signed_challenge"], "signed-value");
1482    }
1483
1484    #[rstest]
1485    #[tokio::test]
1486    async fn test_auth_tracker_succeed_completes_wait_for_result() {
1487        let tracker = AuthTracker::new();
1488        let receiver = tracker.begin();
1489
1490        let tracker_for_responder = tracker.clone();
1491
1492        tokio::spawn(async move {
1493            tokio::time::sleep(tokio::time::Duration::from_millis(20)).await;
1494            tracker_for_responder.succeed();
1495        });
1496
1497        tracker
1498            .wait_for_result::<KrakenWsError>(tokio::time::Duration::from_secs(1), receiver)
1499            .await
1500            .expect("auth should succeed");
1501
1502        assert!(tracker.is_authenticated());
1503    }
1504
1505    #[rstest]
1506    #[tokio::test]
1507    async fn test_auth_tracker_wait_for_result_times_out() {
1508        let tracker = AuthTracker::new();
1509        let receiver = tracker.begin();
1510
1511        let err = tracker
1512            .wait_for_result::<KrakenWsError>(tokio::time::Duration::from_millis(20), receiver)
1513            .await
1514            .expect_err("should time out");
1515
1516        assert!(matches!(err, KrakenWsError::AuthenticationError(_)));
1517        assert!(!tracker.is_authenticated());
1518    }
1519
1520    #[rstest]
1521    #[tokio::test]
1522    async fn test_authenticate_without_credentials_errors() {
1523        let client = KrakenFuturesWebSocketClient::new(
1524            "wss://futures.kraken.com/ws/v1".to_string(),
1525            60,
1526            None,
1527        );
1528
1529        let err = client.authenticate().await.expect_err("should fail");
1530        assert!(
1531            matches!(err, KrakenWsError::AuthenticationError(ref msg) if msg.contains("API credentials required")),
1532            "unexpected error: {err:?}"
1533        );
1534    }
1535
1536    #[rstest]
1537    #[tokio::test]
1538    async fn test_set_auth_credentials_marks_tracker_authenticated() {
1539        let client = KrakenFuturesWebSocketClient::with_credentials(
1540            "wss://futures.kraken.com/ws/v1".to_string(),
1541            60,
1542            Some(test_credential()),
1543            None,
1544            TransportBackend::default(),
1545            None,
1546        );
1547
1548        assert!(!client.is_authenticated());
1549
1550        client
1551            .set_auth_credentials("orig-challenge".to_string(), "signed-challenge".to_string())
1552            .await
1553            .expect("should succeed");
1554
1555        assert!(client.is_authenticated());
1556        client
1557            .wait_until_authenticated(0.05)
1558            .await
1559            .expect("should return immediately");
1560    }
1561
1562    #[rstest]
1563    #[tokio::test]
1564    async fn test_set_auth_credentials_without_credentials_errors() {
1565        let client = KrakenFuturesWebSocketClient::new(
1566            "wss://futures.kraken.com/ws/v1".to_string(),
1567            60,
1568            None,
1569        );
1570
1571        let err = client
1572            .set_auth_credentials("orig".to_string(), "signed".to_string())
1573            .await
1574            .expect_err("should fail");
1575        assert!(matches!(err, KrakenWsError::AuthenticationError(_)));
1576        assert!(!client.is_authenticated());
1577    }
1578
1579    #[rstest]
1580    #[tokio::test]
1581    async fn test_authenticate_with_challenge_updates_state() {
1582        let client = KrakenFuturesWebSocketClient::with_credentials(
1583            "wss://futures.kraken.com/ws/v1".to_string(),
1584            60,
1585            Some(test_credential()),
1586            None,
1587            TransportBackend::default(),
1588            None,
1589        );
1590
1591        client
1592            .authenticate_with_challenge("server-challenge")
1593            .await
1594            .expect("should succeed");
1595
1596        assert!(client.is_authenticated());
1597    }
1598
1599    #[rstest]
1600    #[tokio::test]
1601    async fn test_wait_until_authenticated_resolves_after_success() {
1602        let client = KrakenFuturesWebSocketClient::with_credentials(
1603            "wss://futures.kraken.com/ws/v1".to_string(),
1604            60,
1605            Some(test_credential()),
1606            None,
1607            TransportBackend::default(),
1608            None,
1609        );
1610
1611        let client_for_responder = client.clone();
1612
1613        tokio::spawn(async move {
1614            tokio::time::sleep(tokio::time::Duration::from_millis(20)).await;
1615            client_for_responder
1616                .set_auth_credentials("orig".to_string(), "signed".to_string())
1617                .await
1618                .expect("succeeds");
1619        });
1620
1621        client
1622            .wait_until_authenticated(1.0)
1623            .await
1624            .expect("should resolve once credentials are set");
1625    }
1626
1627    #[rstest]
1628    #[tokio::test]
1629    async fn test_wait_until_authenticated_times_out() {
1630        let client = KrakenFuturesWebSocketClient::with_credentials(
1631            "wss://futures.kraken.com/ws/v1".to_string(),
1632            60,
1633            Some(test_credential()),
1634            None,
1635            TransportBackend::default(),
1636            None,
1637        );
1638
1639        let err = client
1640            .wait_until_authenticated(0.05)
1641            .await
1642            .expect_err("should time out");
1643        assert!(matches!(err, KrakenWsError::AuthenticationError(_)));
1644    }
1645}
nautilus_kraken/websocket/futures/client.rs

nautilus_kraken/websocket/futures/
client.rs
