nautilus_infrastructure/redis/
queries.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
16use std::{collections::HashMap, str::FromStr};
17
18use ahash::AHashMap;
19use bytes::Bytes;
20use futures::future::join_all;
21use jiff::Timestamp;
22use nautilus_common::{
23    cache::database::{CacheMap, register_loaded_currencies},
24    enums::SerializationEncoding,
25};
26use nautilus_model::{
27    accounts::AccountAny,
28    data::{CustomData, DataType, HasTsInit, InstrumentClose},
29    enums::CurrencyType,
30    events::{AccountState, OrderEventAny, OrderFilled, PositionSnapshot},
31    identifiers::{AccountId, ClientId, ClientOrderId, InstrumentId, PositionId},
32    instruments::{InstrumentAny, SyntheticInstrument},
33    orders::OrderAny,
34    position::Position,
35    types::Currency,
36};
37use redis::{AsyncCommands, aio::ConnectionManager};
38use serde::{Deserialize, Serialize, de::DeserializeOwned};
39use serde_json::Value;
40use ustr::Ustr;
41
42use super::get_index_key;
43
44/// Persisted representation of a [`Currency`].
45///
46/// A `Currency` serializes as its bare code, which only resolves in a process where that code is
47/// already registered, so a currency minted at runtime could not be restored. This record carries
48/// the same fields the Postgres `currency` table stores, keeping the two backends equivalent.
49#[derive(Debug, Clone, Serialize, Deserialize)]
50pub(crate) struct CurrencyRecord {
51    pub code: String,
52    pub precision: u8,
53    pub iso4217: u16,
54    pub name: String,
55    pub currency_type: CurrencyType,
56}
57
58impl From<&Currency> for CurrencyRecord {
59    fn from(currency: &Currency) -> Self {
60        Self {
61            code: currency.code.to_string(),
62            precision: currency.precision,
63            iso4217: currency.iso4217,
64            name: currency.name.to_string(),
65            currency_type: currency.currency_type,
66        }
67    }
68}
69
70impl TryFrom<CurrencyRecord> for Currency {
71    type Error = anyhow::Error;
72
73    fn try_from(record: CurrencyRecord) -> Result<Self, Self::Error> {
74        Self::new_checked(
75            record.code.as_str(),
76            record.precision,
77            record.iso4217,
78            record.name.as_str(),
79            record.currency_type,
80        )
81        .map_err(|e| anyhow::anyhow!("Invalid currency record: {e}"))
82    }
83}
84
85/// Outcome of decoding one persisted currency entry.
86enum CurrencyDecode {
87    Currency(Currency),
88    /// A record, or a payload that is not a bare code, that does not yield a valid currency.
89    InvalidRecord(anyhow::Error),
90    /// A bare code that is not registered, so the entry holds nothing recoverable.
91    Undecodable(anyhow::Error),
92}
93
94// Collection keys
95const INDEX: &str = "index";
96const GENERAL: &str = "general";
97const CURRENCIES: &str = "currencies";
98const INSTRUMENTS: &str = "instruments";
99const INSTRUMENT_CLOSES: &str = "instrument_closes";
100const SYNTHETICS: &str = "synthetics";
101const ACCOUNTS: &str = "accounts";
102const ORDERS: &str = "orders";
103const POSITIONS: &str = "positions";
104const SNAPSHOTS: &str = "snapshots";
105const ACTORS: &str = "actors";
106const STRATEGIES: &str = "strategies";
107const CUSTOM: &str = "custom";
108const REDIS_DELIMITER: char = ':';
109
110// Index keys
111const INDEX_ORDER_IDS: &str = "index:order_ids";
112const INDEX_ORDER_POSITION: &str = "index:order_position";
113const INDEX_ORDER_CLIENT: &str = "index:order_client";
114const INDEX_ORDERS: &str = "index:orders";
115const INDEX_ORDERS_OPEN: &str = "index:orders_open";
116const INDEX_ORDERS_CLOSED: &str = "index:orders_closed";
117const INDEX_ORDERS_EMULATED: &str = "index:orders_emulated";
118const INDEX_ORDERS_INFLIGHT: &str = "index:orders_inflight";
119const INDEX_POSITIONS: &str = "index:positions";
120const INDEX_POSITIONS_OPEN: &str = "index:positions_open";
121const INDEX_POSITIONS_CLOSED: &str = "index:positions_closed";
122
123#[derive(Debug)]
124pub struct DatabaseQueries;
125
126impl DatabaseQueries {
127    /// Serializes the given `payload` using the specified `encoding` to a byte vector.
128    ///
129    /// # Errors
130    ///
131    /// Returns an error if serialization to the chosen encoding fails.
132    pub fn serialize_payload<T: Serialize>(
133        encoding: SerializationEncoding,
134        payload: &T,
135    ) -> anyhow::Result<Vec<u8>> {
136        match encoding {
137            SerializationEncoding::MsgPack => {
138                let mut value = serde_json::to_value(payload)?;
139                convert_timestamps(&mut value);
140                rmp_serde::to_vec(&value)
141                    .map_err(|e| anyhow::anyhow!("Failed to serialize msgpack `payload`: {e}"))
142            }
143            SerializationEncoding::Json => {
144                let mut value = serde_json::to_value(payload)?;
145                convert_timestamps(&mut value);
146                serde_json::to_vec(&value)
147                    .map_err(|e| anyhow::anyhow!("Failed to serialize json `payload`: {e}"))
148            }
149            SerializationEncoding::Sbe => {
150                anyhow::bail!("SBE encoding is not supported for Redis cache payloads")
151            }
152            SerializationEncoding::Capnp => {
153                anyhow::bail!("Cap'n Proto encoding is not supported for Redis cache payloads")
154            }
155        }
156    }
157
158    /// Deserializes the given byte slice `payload` into type `T` using the specified `encoding`.
159    ///
160    /// # Errors
161    ///
162    /// Returns an error if deserialization from the chosen encoding fails or converting to the target type fails.
163    pub fn deserialize_payload<T: DeserializeOwned>(
164        encoding: SerializationEncoding,
165        payload: &[u8],
166    ) -> anyhow::Result<T> {
167        let mut value = match encoding {
168            SerializationEncoding::MsgPack => rmp_serde::from_slice(payload)
169                .map_err(|e| anyhow::anyhow!("Failed to deserialize msgpack `payload`: {e}"))?,
170            SerializationEncoding::Json => serde_json::from_slice(payload)
171                .map_err(|e| anyhow::anyhow!("Failed to deserialize json `payload`: {e}"))?,
172            SerializationEncoding::Sbe => {
173                anyhow::bail!("SBE encoding is not supported for Redis cache payloads")
174            }
175            SerializationEncoding::Capnp => {
176                anyhow::bail!("Cap'n Proto encoding is not supported for Redis cache payloads")
177            }
178        };
179
180        convert_timestamp_strings(&mut value);
181
182        serde_json::from_value(value)
183            .map_err(|e| anyhow::anyhow!("Failed to convert value to target type: {e}"))
184    }
185
186    /// Scans Redis for keys matching the given `pattern`.
187    ///
188    /// # Errors
189    ///
190    /// Returns an error if the Redis scan operation fails.
191    pub async fn scan_keys(
192        con: &mut ConnectionManager,
193        pattern: String,
194    ) -> anyhow::Result<Vec<String>> {
195        let mut result = Vec::new();
196        let mut cursor = 0u64;
197
198        loop {
199            let scan_result: (u64, Vec<String>) = redis::cmd("SCAN")
200                .arg(cursor)
201                .arg("MATCH")
202                .arg(&pattern)
203                .arg("COUNT")
204                .arg(5000)
205                .query_async(con)
206                .await?;
207
208            let (new_cursor, keys) = scan_result;
209            result.extend(keys);
210
211            // If cursor is 0, we've completed the full scan
212            if new_cursor == 0 {
213                break;
214            }
215
216            cursor = new_cursor;
217        }
218
219        Ok(result)
220    }
221
222    /// Bulk reads multiple keys from Redis using MGET for efficiency.
223    ///
224    /// # Errors
225    ///
226    /// Returns an error if the underlying Redis MGET operation fails.
227    pub async fn read_bulk(
228        con: &ConnectionManager,
229        keys: &[String],
230    ) -> anyhow::Result<Vec<Option<Bytes>>> {
231        if keys.is_empty() {
232            return Ok(vec![]);
233        }
234
235        let mut con = con.clone();
236
237        // Use MGET to fetch all keys in a single network operation
238        let results: Vec<Option<Vec<u8>>> =
239            redis::cmd("MGET").arg(keys).query_async(&mut con).await?;
240
241        // Convert Vec<u8> to Bytes
242        let bytes_results: Vec<Option<Bytes>> = results
243            .into_iter()
244            .map(|opt| opt.map(Bytes::from))
245            .collect();
246
247        Ok(bytes_results)
248    }
249
250    /// Bulk reads multiple keys from Redis using MGET, batched into chunks.
251    ///
252    /// Keys are batched into chunks of `batch_size` to avoid exceeding Redis
253    /// request size limits on some providers.
254    ///
255    /// # Errors
256    ///
257    /// Returns an error if `batch_size` is zero or if the underlying Redis MGET operation fails.
258    pub async fn read_bulk_batched(
259        con: &ConnectionManager,
260        keys: &[String],
261        batch_size: usize,
262    ) -> anyhow::Result<Vec<Option<Bytes>>> {
263        if batch_size == 0 {
264            anyhow::bail!("`batch_size` must be greater than zero");
265        }
266
267        if keys.is_empty() {
268            return Ok(vec![]);
269        }
270
271        let mut all_results: Vec<Option<Bytes>> = Vec::with_capacity(keys.len());
272
273        for chunk in keys.chunks(batch_size) {
274            let mut con = con.clone();
275
276            let results: Vec<Option<Vec<u8>>> =
277                redis::cmd("MGET").arg(chunk).query_async(&mut con).await?;
278
279            all_results.extend(results.into_iter().map(|opt| opt.map(Bytes::from)));
280        }
281
282        Ok(all_results)
283    }
284
285    /// Reads raw byte payloads for `key` under `trader_key` from Redis.
286    ///
287    /// # Errors
288    ///
289    /// Returns an error if the underlying Redis read operation fails or if the collection is unsupported.
290    pub async fn read(
291        con: &ConnectionManager,
292        trader_key: &str,
293        key: &str,
294    ) -> anyhow::Result<Vec<Bytes>> {
295        let collection = Self::get_collection_key(key)?;
296        let full_key = format!("{trader_key}{REDIS_DELIMITER}{key}");
297
298        let mut con = con.clone();
299
300        match collection {
301            INDEX => Self::read_index(&mut con, &full_key).await,
302            GENERAL | CURRENCIES | INSTRUMENTS | INSTRUMENT_CLOSES | SYNTHETICS | ACTORS
303            | STRATEGIES => Self::read_string(&mut con, &full_key).await,
304            ACCOUNTS | ORDERS | POSITIONS | SNAPSHOTS => Self::read_list(&mut con, &full_key).await,
305            _ => anyhow::bail!("Unsupported operation: `read` for collection '{collection}'"),
306        }
307    }
308
309    /// Loads all cache data (currencies, instruments, synthetics, accounts, orders, positions) for `trader_key`.
310    ///
311    /// # Errors
312    ///
313    /// Returns an error if loading any of the individual caches fails or combining data fails.
314    pub async fn load_all(
315        con: &ConnectionManager,
316        encoding: SerializationEncoding,
317        trader_key: &str,
318    ) -> anyhow::Result<CacheMap> {
319        // Currencies must be registered before the dependent payloads decode, because a `Money`
320        // or a `Currency` in them resolves its code through the global registry.
321        let mut currencies = Self::load_currencies(con, trader_key, encoding).await?;
322        register_loaded_currencies(&mut currencies)?;
323
324        let (instruments, instrument_closes, synthetics, accounts, orders, positions) =
325            tokio::try_join!(
326                Self::load_instruments(con, trader_key, encoding),
327                Self::load_instrument_closes(con, trader_key, encoding),
328                Self::load_synthetics(con, trader_key, encoding),
329                Self::load_accounts(con, trader_key, encoding),
330                Self::load_orders(con, trader_key, encoding),
331                Self::load_positions(con, trader_key, encoding)
332            )
333            .map_err(|e| anyhow::anyhow!("Error loading cache data: {e}"))?;
334
335        // For now, we don't load greeks and yield curves from the database
336        // This will be implemented in the future
337        let greeks = AHashMap::new();
338        let yield_curves = AHashMap::new();
339
340        Ok(CacheMap {
341            currencies,
342            instruments,
343            instrument_closes,
344            synthetics,
345            accounts,
346            orders,
347            positions,
348            greeks,
349            yield_curves,
350        })
351    }
352
353    /// Decodes a persisted currency, accepting both the record and a bare code.
354    ///
355    /// A bare-code entry holds only the code, which resolves if that code is registered; an
356    /// unregistered one is skipped by bulk loading, since the entry holds nothing recoverable. Anything
357    /// that is not a bare code is a record, and a record that does not decode or describes an
358    /// invalid currency is an error, since the intent is recoverable and dropping it would omit a
359    /// currency the dependent payloads need.
360    fn decode_currency(encoding: SerializationEncoding, value_bytes: &Bytes) -> CurrencyDecode {
361        match Self::deserialize_payload::<CurrencyRecord>(encoding, value_bytes) {
362            Ok(record) => match Currency::try_from(record) {
363                Ok(currency) => CurrencyDecode::Currency(currency),
364                Err(e) => CurrencyDecode::InvalidRecord(e),
365            },
366            Err(record_err) => match Self::deserialize_payload::<String>(encoding, value_bytes) {
367                Ok(code) => match Currency::try_from_str(&code) {
368                    Some(currency) => CurrencyDecode::Currency(currency),
369                    None => CurrencyDecode::Undecodable(anyhow::anyhow!(
370                        "bare currency code {code} is not registered"
371                    )),
372                },
373                Err(_) => CurrencyDecode::InvalidRecord(anyhow::anyhow!(
374                    "payload is not a bare code and does not decode as a currency record: {record_err}"
375                )),
376            },
377        }
378    }
379
380    /// Loads all currencies for `trader_key` using the specified `encoding`.
381    ///
382    /// # Errors
383    ///
384    /// Returns an error if scanning keys or reading currency data fails, or if a stored record
385    /// describes an invalid currency.
386    pub async fn load_currencies(
387        con: &ConnectionManager,
388        trader_key: &str,
389        encoding: SerializationEncoding,
390    ) -> anyhow::Result<AHashMap<Ustr, Currency>> {
391        let mut currencies = AHashMap::new();
392        let prefix = format!("{trader_key}{REDIS_DELIMITER}{CURRENCIES}{REDIS_DELIMITER}");
393        let pattern = format!("{prefix}*");
394        log::debug!("Loading {pattern}");
395
396        let mut con = con.clone();
397        let keys = Self::scan_keys(&mut con, pattern).await?;
398
399        if keys.is_empty() {
400            return Ok(currencies);
401        }
402
403        // Use bulk loading with MGET for efficiency
404        let bulk_values = Self::read_bulk(&con, &keys).await?;
405
406        // Process the bulk results
407        for (key, value_opt) in keys.iter().zip(bulk_values.iter()) {
408            // A code can contain the delimiter (Hyperliquid HIP-3 assets such as `xyz:TSLA`), so
409            // the code is everything after the collection prefix, not the last segment.
410            let currency_code = match key.as_str().strip_prefix(prefix.as_str()) {
411                Some(code) if !code.is_empty() => Ustr::from(code),
412                _ => {
413                    log::error!("Invalid key format: {key}");
414                    continue;
415                }
416            };
417
418            if let Some(value_bytes) = value_opt {
419                match Self::decode_currency(encoding, value_bytes) {
420                    CurrencyDecode::Currency(currency) => {
421                        // The key names the code the dependents resolve, so a record filed under
422                        // another code would register one currency and index another.
423                        if currency.code != currency_code {
424                            anyhow::bail!(
425                                "Currency record under key {currency_code} holds code {}",
426                                currency.code
427                            );
428                        }
429                        currencies.insert(currency_code, currency);
430                    }
431                    CurrencyDecode::InvalidRecord(e) => {
432                        anyhow::bail!("Invalid currency record for {currency_code}: {e}");
433                    }
434                    CurrencyDecode::Undecodable(e) => {
435                        log::error!(
436                            "Skipping currency {currency_code}: {e}; payloads denominated in it will not decode"
437                        );
438                    }
439                }
440            } else {
441                log::error!("Currency not found in Redis: {currency_code}");
442            }
443        }
444
445        log::debug!("Loaded {} currencies(s)", currencies.len());
446
447        Ok(currencies)
448    }
449
450    /// Loads all instruments for `trader_key` using the specified `encoding`.
451    ///
452    /// # Errors
453    ///
454    /// Returns an error if scanning keys or reading instrument data fails.
455    /// Loads all instruments for `trader_key` using the specified `encoding`.
456    ///
457    /// # Errors
458    ///
459    /// Returns an error if scanning keys or reading instrument data fails.
460    pub async fn load_instruments(
461        con: &ConnectionManager,
462        trader_key: &str,
463        encoding: SerializationEncoding,
464    ) -> anyhow::Result<AHashMap<InstrumentId, InstrumentAny>> {
465        let mut instruments = AHashMap::new();
466        let prefix = format!("{trader_key}{REDIS_DELIMITER}{INSTRUMENTS}{REDIS_DELIMITER}");
467        let pattern = format!("{prefix}*");
468        log::debug!("Loading {pattern}");
469
470        let mut con = con.clone();
471        let keys = Self::scan_keys(&mut con, pattern).await?;
472
473        let futures: Vec<_> = keys
474            .iter()
475            .map(|key| {
476                let con = con.clone();
477                let prefix = &prefix;
478                async move {
479                    let instrument_id = parse_instrument_key(key, prefix);
480
481                    let Ok(instrument_id) = instrument_id else {
482                        log::error!("Failed to parse InstrumentId from Redis key: {key}");
483                        return None;
484                    };
485
486                    match Self::load_instrument(&con, trader_key, &instrument_id, encoding).await {
487                        Ok(Some(instrument)) => Some((instrument_id, instrument)),
488                        Ok(None) => {
489                            log::error!("Instrument not found: {instrument_id}");
490                            None
491                        }
492                        Err(e) => {
493                            log::error!("Failed to load instrument {instrument_id}: {e}");
494                            None
495                        }
496                    }
497                }
498            })
499            .collect();
500
501        // Insert all Instrument_id (key) and Instrument (value) into the HashMap, filtering out None values.
502        instruments.extend(join_all(futures).await.into_iter().flatten());
503        log::debug!("Loaded {} instruments(s)", instruments.len());
504
505        Ok(instruments)
506    }
507
508    /// Loads all instrument closes for `trader_key`.
509    ///
510    /// Missing or invalid close data fails the load so recovery cannot silently omit a close.
511    ///
512    /// # Errors
513    ///
514    /// Returns an error if scanning, reading, parsing, or deserializing instrument closes fails.
515    pub async fn load_instrument_closes(
516        con: &ConnectionManager,
517        trader_key: &str,
518        encoding: SerializationEncoding,
519    ) -> anyhow::Result<AHashMap<InstrumentId, InstrumentClose>> {
520        let prefix = format!("{trader_key}{REDIS_DELIMITER}{INSTRUMENT_CLOSES}{REDIS_DELIMITER}");
521        let pattern = format!("{prefix}*");
522        log::debug!("Loading {pattern}");
523
524        let mut con = con.clone();
525        let keys = Self::scan_keys(&mut con, pattern).await?;
526        let values = Self::read_bulk(&con, &keys).await?;
527        let mut closes = AHashMap::with_capacity(keys.len());
528
529        for (key, value) in keys.into_iter().zip(values) {
530            let instrument_id = parse_instrument_key(&key, &prefix)?;
531            let value = value
532                .ok_or_else(|| anyhow::anyhow!("Instrument close not found in Redis: {key}"))?;
533            let close: InstrumentClose = Self::deserialize_payload(encoding, &value)?;
534            anyhow::ensure!(
535                close.instrument_id == instrument_id,
536                "Instrument close key ID {instrument_id} did not match payload ID {}",
537                close.instrument_id,
538            );
539            closes.insert(instrument_id, close);
540        }
541
542        log::debug!("Loaded {} instrument close(s)", closes.len());
543        Ok(closes)
544    }
545
546    /// Loads all synthetic instruments for `trader_key` using the specified `encoding`.
547    ///
548    /// # Errors
549    ///
550    /// Returns an error if scanning keys or reading synthetic instrument data fails.
551    pub async fn load_synthetics(
552        con: &ConnectionManager,
553        trader_key: &str,
554        encoding: SerializationEncoding,
555    ) -> anyhow::Result<AHashMap<InstrumentId, SyntheticInstrument>> {
556        let mut synthetics = AHashMap::new();
557        let pattern = format!("{trader_key}{REDIS_DELIMITER}{SYNTHETICS}*");
558        log::debug!("Loading {pattern}");
559
560        let mut con = con.clone();
561        let keys = Self::scan_keys(&mut con, pattern).await?;
562
563        let futures: Vec<_> = keys
564            .iter()
565            .map(|key| {
566                let con = con.clone();
567                async move {
568                    let instrument_id = key
569                        .as_str()
570                        .rsplit(':')
571                        .next()
572                        .ok_or_else(|| {
573                            log::error!("Invalid key format: {key}");
574                            "Invalid key format"
575                        })
576                        .and_then(|code| {
577                            InstrumentId::from_str(code).map_err(|e| {
578                                log::error!("Failed to parse InstrumentId for {key}: {e}");
579                                "Invalid instrument ID"
580                            })
581                        });
582
583                    let Ok(instrument_id) = instrument_id else {
584                        return None;
585                    };
586
587                    match Self::load_synthetic(&con, trader_key, &instrument_id, encoding).await {
588                        Ok(Some(synthetic)) => Some((instrument_id, synthetic)),
589                        Ok(None) => {
590                            log::error!("Synthetic not found: {instrument_id}");
591                            None
592                        }
593                        Err(e) => {
594                            log::error!("Failed to load synthetic {instrument_id}: {e}");
595                            None
596                        }
597                    }
598                }
599            })
600            .collect();
601
602        // Insert all Instrument_id (key) and Synthetic (value) into the HashMap, filtering out None values.
603        synthetics.extend(join_all(futures).await.into_iter().flatten());
604        log::debug!("Loaded {} synthetics(s)", synthetics.len());
605
606        Ok(synthetics)
607    }
608
609    /// Loads all accounts for `trader_key` using the specified `encoding`.
610    ///
611    /// # Errors
612    ///
613    /// Returns an error if scanning keys or reading account data fails.
614    /// Loads all accounts for `trader_key` using the specified `encoding`.
615    ///
616    /// # Errors
617    ///
618    /// Returns an error if scanning keys or reading account data fails.
619    pub async fn load_accounts(
620        con: &ConnectionManager,
621        trader_key: &str,
622        encoding: SerializationEncoding,
623    ) -> anyhow::Result<AHashMap<AccountId, AccountAny>> {
624        let mut accounts = AHashMap::new();
625        let pattern = format!("{trader_key}{REDIS_DELIMITER}{ACCOUNTS}*");
626        log::debug!("Loading {pattern}");
627
628        let mut con = con.clone();
629        let keys = Self::scan_keys(&mut con, pattern).await?;
630
631        let futures: Vec<_> = keys
632            .iter()
633            .map(|key| {
634                let con = con.clone();
635                async move {
636                    let account_id = if let Some(code) = key.as_str().rsplit(':').next() {
637                        AccountId::from(code)
638                    } else {
639                        log::error!("Invalid key format: {key}");
640                        return None;
641                    };
642
643                    match Self::load_account(&con, trader_key, &account_id, encoding).await {
644                        Ok(Some(account)) => Some((account_id, account)),
645                        Ok(None) => {
646                            log::error!("Account not found: {account_id}");
647                            None
648                        }
649                        Err(e) => {
650                            log::error!("Failed to load account {account_id}: {e}");
651                            None
652                        }
653                    }
654                }
655            })
656            .collect();
657
658        // Insert all Account_id (key) and Account (value) into the HashMap, filtering out None values.
659        accounts.extend(join_all(futures).await.into_iter().flatten());
660        log::debug!("Loaded {} accounts(s)", accounts.len());
661
662        Ok(accounts)
663    }
664
665    /// Loads all orders for `trader_key` using the specified `encoding`.
666    ///
667    /// # Errors
668    ///
669    /// Returns an error if scanning keys or reading order data fails.
670    /// Loads all orders for `trader_key` using the specified `encoding`.
671    ///
672    /// # Errors
673    ///
674    /// Returns an error if scanning keys or reading order data fails.
675    pub async fn load_orders(
676        con: &ConnectionManager,
677        trader_key: &str,
678        encoding: SerializationEncoding,
679    ) -> anyhow::Result<AHashMap<ClientOrderId, OrderAny>> {
680        let mut orders = AHashMap::new();
681        let pattern = format!("{trader_key}{REDIS_DELIMITER}{ORDERS}*");
682        log::debug!("Loading {pattern}");
683
684        let mut con = con.clone();
685        let keys = Self::scan_keys(&mut con, pattern).await?;
686
687        let futures: Vec<_> = keys
688            .iter()
689            .map(|key| {
690                let con = con.clone();
691                async move {
692                    let client_order_id = if let Some(code) = key.as_str().rsplit(':').next() {
693                        ClientOrderId::from(code)
694                    } else {
695                        log::error!("Invalid key format: {key}");
696                        return None;
697                    };
698
699                    match Self::load_order(&con, trader_key, &client_order_id, encoding).await {
700                        Ok(Some(order)) => Some((client_order_id, order)),
701                        Ok(None) => {
702                            log::error!("Order not found: {client_order_id}");
703                            None
704                        }
705                        Err(e) => {
706                            log::error!("Failed to load order {client_order_id}: {e}");
707                            None
708                        }
709                    }
710                }
711            })
712            .collect();
713
714        // Insert all Client-Order-Id (key) and Order (value) into the HashMap, filtering out None values.
715        orders.extend(join_all(futures).await.into_iter().flatten());
716        log::debug!("Loaded {} order(s)", orders.len());
717
718        Ok(orders)
719    }
720
721    /// Loads all positions for `trader_key` using the specified `encoding`.
722    ///
723    /// # Errors
724    ///
725    /// Returns an error if scanning keys or reading position data fails.
726    /// Loads all positions for `trader_key` using the specified `encoding`.
727    ///
728    /// # Errors
729    ///
730    /// Returns an error if scanning keys or reading position data fails.
731    pub async fn load_positions(
732        con: &ConnectionManager,
733        trader_key: &str,
734        encoding: SerializationEncoding,
735    ) -> anyhow::Result<AHashMap<PositionId, Position>> {
736        let mut positions = AHashMap::new();
737        let pattern = format!("{trader_key}{REDIS_DELIMITER}{POSITIONS}*");
738        log::debug!("Loading {pattern}");
739
740        let mut con = con.clone();
741        let keys = Self::scan_keys(&mut con, pattern).await?;
742
743        let futures: Vec<_> = keys
744            .iter()
745            .map(|key| {
746                let con = con.clone();
747                async move {
748                    let position_id = if let Some(code) = key.as_str().rsplit(':').next() {
749                        PositionId::from(code)
750                    } else {
751                        log::error!("Invalid key format: {key}");
752                        return None;
753                    };
754
755                    match Self::load_position(&con, trader_key, &position_id, encoding).await {
756                        Ok(Some(position)) => Some((position_id, position)),
757                        Ok(None) => {
758                            log::error!("Position not found: {position_id}");
759                            None
760                        }
761                        Err(e) => {
762                            log::error!("Failed to load position {position_id}: {e}");
763                            None
764                        }
765                    }
766                }
767            })
768            .collect();
769
770        // Insert all Position_id (key) and Position (value) into the HashMap, filtering out None values.
771        positions.extend(join_all(futures).await.into_iter().flatten());
772        log::debug!("Loaded {} position(s)", positions.len());
773
774        Ok(positions)
775    }
776
777    /// Loads the order ID to position ID index for `trader_key`.
778    ///
779    /// # Errors
780    ///
781    /// Returns an error if reading or parsing the index fails.
782    pub async fn load_index_order_position(
783        con: &ConnectionManager,
784        trader_key: &str,
785    ) -> anyhow::Result<AHashMap<ClientOrderId, PositionId>> {
786        let index = Self::read_index_hash(con, trader_key, INDEX_ORDER_POSITION).await?;
787        Ok(index
788            .into_iter()
789            .map(|(k, v)| {
790                (
791                    ClientOrderId::from(k.as_str()),
792                    PositionId::from(v.as_str()),
793                )
794            })
795            .collect())
796    }
797
798    /// Loads the order ID to execution client ID index for `trader_key`.
799    ///
800    /// # Errors
801    ///
802    /// Returns an error if reading or parsing the index fails.
803    pub async fn load_index_order_client(
804        con: &ConnectionManager,
805        trader_key: &str,
806    ) -> anyhow::Result<AHashMap<ClientOrderId, ClientId>> {
807        let index = Self::read_index_hash(con, trader_key, INDEX_ORDER_CLIENT).await?;
808        Ok(index
809            .into_iter()
810            .map(|(k, v)| (ClientOrderId::from(k.as_str()), ClientId::from(v.as_str())))
811            .collect())
812    }
813
814    async fn read_index_hash(
815        con: &ConnectionManager,
816        trader_key: &str,
817        key: &str,
818    ) -> anyhow::Result<HashMap<String, String>> {
819        let result = Self::read(con, trader_key, key).await?;
820        if result.is_empty() {
821            return Ok(HashMap::new());
822        }
823
824        serde_json::from_slice(&result[0])
825            .map_err(|e| anyhow::anyhow!("Failed to parse index hash '{key}': {e}"))
826    }
827
828    /// Loads all custom data for `trader_key` matching the given `data_type`.
829    ///
830    /// Keys are stored as `custom:<ts_init_020>:<uuid>`; value is full `CustomData` JSON.
831    /// Scans all custom keys, deserializes, filters by `type_name` (full or short), metadata,
832    /// and identifier to match SQL semantics, then sorts by `ts_init` ascending.
833    ///
834    /// # Errors
835    ///
836    /// Returns an error if scanning, bulk read, or deserialization fails.
837    pub async fn load_custom_data(
838        con: &ConnectionManager,
839        trader_key: &str,
840        data_type: &DataType,
841    ) -> anyhow::Result<Vec<CustomData>> {
842        let pattern = format!("{trader_key}{REDIS_DELIMITER}{CUSTOM}*");
843        log::debug!("Loading custom data {pattern}");
844
845        let mut con = con.clone();
846        let keys = Self::scan_keys(&mut con, pattern).await?;
847
848        if keys.is_empty() {
849            return Ok(Vec::new());
850        }
851
852        let values = Self::read_bulk(&con, &keys).await?;
853        let request_type_name = data_type.type_name();
854        let request_short = request_type_name
855            .rsplit([':', '.'])
856            .next()
857            .unwrap_or(request_type_name);
858        let request_identifier = data_type.identifier().unwrap_or("");
859
860        let mut results = Vec::new();
861
862        for value_opt in values {
863            let Some(value_bytes) = value_opt else {
864                continue;
865            };
866            let custom = match CustomData::from_json_bytes(value_bytes.as_ref()) {
867                Ok(c) => c,
868                Err(e) => {
869                    log::warn!("Failed to deserialize custom data from Redis: {e}");
870                    continue;
871                }
872            };
873            let stored_type_name = custom.data_type.type_name();
874            let type_match =
875                stored_type_name == request_type_name || stored_type_name == request_short;
876            let identifier_match =
877                custom.data_type.identifier().unwrap_or("") == request_identifier;
878            let metadata_match = match (data_type.metadata(), custom.data_type.metadata()) {
879                (None, None) => true,
880                (Some(a), Some(b)) => serde_json::to_value(a).ok() == serde_json::to_value(b).ok(),
881                _ => false,
882            };
883
884            if type_match && identifier_match && metadata_match {
885                results.push(custom);
886            }
887        }
888
889        results.sort_by_key(HasTsInit::ts_init);
890        log::debug!("Loaded {} custom data item(s)", results.len());
891        Ok(results)
892    }
893
894    /// Loads a single currency for `trader_key` and `code` using the specified `encoding`.
895    ///
896    /// # Errors
897    ///
898    /// Returns an error if the underlying read or deserialization fails.
899    pub async fn load_currency(
900        con: &ConnectionManager,
901        trader_key: &str,
902        code: &Ustr,
903        encoding: SerializationEncoding,
904    ) -> anyhow::Result<Option<Currency>> {
905        let key = format!("{CURRENCIES}{REDIS_DELIMITER}{code}");
906        let result = Self::read(con, trader_key, &key).await?;
907
908        if result.is_empty() {
909            return Ok(None);
910        }
911
912        match Self::decode_currency(encoding, &result[0]) {
913            CurrencyDecode::Currency(currency) => Ok(Some(currency)),
914            CurrencyDecode::InvalidRecord(e) | CurrencyDecode::Undecodable(e) => {
915                Err(e.context(format!("Failed to load currency {code}")))
916            }
917        }
918    }
919
920    /// Loads a single instrument for `trader_key` and `instrument_id` using the specified `encoding`.
921    ///
922    /// # Errors
923    ///
924    /// Returns an error if the underlying read or deserialization fails.
925    pub async fn load_instrument(
926        con: &ConnectionManager,
927        trader_key: &str,
928        instrument_id: &InstrumentId,
929        encoding: SerializationEncoding,
930    ) -> anyhow::Result<Option<InstrumentAny>> {
931        let key = format!("{INSTRUMENTS}{REDIS_DELIMITER}{instrument_id}");
932        let result = Self::read(con, trader_key, &key).await?;
933        if result.is_empty() {
934            return Ok(None);
935        }
936
937        let instrument: InstrumentAny = Self::deserialize_payload(encoding, &result[0])?;
938        Ok(Some(instrument))
939    }
940
941    /// Loads a single synthetic instrument for `trader_key` and `instrument_id` using the specified `encoding`.
942    ///
943    /// # Errors
944    ///
945    /// Returns an error if the underlying read or deserialization fails.
946    pub async fn load_synthetic(
947        con: &ConnectionManager,
948        trader_key: &str,
949        instrument_id: &InstrumentId,
950        encoding: SerializationEncoding,
951    ) -> anyhow::Result<Option<SyntheticInstrument>> {
952        let key = format!("{SYNTHETICS}{REDIS_DELIMITER}{instrument_id}");
953        let result = Self::read(con, trader_key, &key).await?;
954        if result.is_empty() {
955            return Ok(None);
956        }
957
958        let synthetic: SyntheticInstrument = Self::deserialize_payload(encoding, &result[0])?;
959        Ok(Some(synthetic))
960    }
961
962    /// Loads a single account for `trader_key` and `account_id` using the specified `encoding`.
963    ///
964    /// # Errors
965    ///
966    /// Returns an error if the underlying read or deserialization fails.
967    pub async fn load_account(
968        con: &ConnectionManager,
969        trader_key: &str,
970        account_id: &AccountId,
971        encoding: SerializationEncoding,
972    ) -> anyhow::Result<Option<AccountAny>> {
973        let key = format!("{ACCOUNTS}{REDIS_DELIMITER}{account_id}");
974        let result = Self::read(con, trader_key, &key).await?;
975        if result.is_empty() {
976            return Ok(None);
977        }
978
979        let events: Vec<AccountState> = result
980            .iter()
981            .map(|payload| Self::deserialize_payload(encoding, payload))
982            .collect::<anyhow::Result<_>>()?;
983        let account = AccountAny::from_events(&events)?;
984        Ok(Some(account))
985    }
986
987    /// Loads a single order for `trader_key` and `client_order_id` using the specified `encoding`.
988    ///
989    /// # Errors
990    ///
991    /// Returns an error if the underlying read or deserialization fails.
992    pub async fn load_order(
993        con: &ConnectionManager,
994        trader_key: &str,
995        client_order_id: &ClientOrderId,
996        encoding: SerializationEncoding,
997    ) -> anyhow::Result<Option<OrderAny>> {
998        let key = format!("{ORDERS}{REDIS_DELIMITER}{client_order_id}");
999        let result = Self::read(con, trader_key, &key).await?;
1000        if result.is_empty() {
1001            return Ok(None);
1002        }
1003
1004        let events: Vec<OrderEventAny> = result
1005            .iter()
1006            .map(|payload| Self::deserialize_payload(encoding, payload))
1007            .collect::<anyhow::Result<_>>()?;
1008        let order = OrderAny::from_events(events)?;
1009        Ok(Some(order))
1010    }
1011
1012    /// Loads a single position for `trader_key` and `position_id` using the specified `encoding`.
1013    ///
1014    /// # Errors
1015    ///
1016    /// Returns an error if the underlying read or deserialization fails.
1017    pub async fn load_position(
1018        con: &ConnectionManager,
1019        trader_key: &str,
1020        position_id: &PositionId,
1021        encoding: SerializationEncoding,
1022    ) -> anyhow::Result<Option<Position>> {
1023        let snapshot_key =
1024            format!("{SNAPSHOTS}{REDIS_DELIMITER}{POSITIONS}{REDIS_DELIMITER}{position_id}");
1025        let snapshots = Self::read(con, trader_key, &snapshot_key).await?;
1026        for payload in snapshots.iter().rev() {
1027            let snapshot: PositionSnapshot = Self::deserialize_payload(encoding, payload)?;
1028            if let Some(replay_state) = snapshot.replay_state {
1029                return serde_json::from_value(replay_state)
1030                    .map(Some)
1031                    .map_err(|e| anyhow::anyhow!("Failed to decode position replay state: {e}"));
1032            }
1033        }
1034
1035        let key = format!("{POSITIONS}{REDIS_DELIMITER}{position_id}");
1036        let result = Self::read(con, trader_key, &key).await?;
1037        if result.is_empty() {
1038            return Ok(None);
1039        }
1040
1041        let fills: Vec<OrderFilled> = result
1042            .iter()
1043            .map(|payload| Self::deserialize_payload(encoding, payload))
1044            .collect::<anyhow::Result<_>>()?;
1045        let Some((first_fill, remaining_fills)) = fills.split_first() else {
1046            return Ok(None);
1047        };
1048        let Some(instrument) =
1049            Self::load_instrument(con, trader_key, &first_fill.instrument_id, encoding).await?
1050        else {
1051            log::error!(
1052                "Instrument not found for position {position_id}: {}",
1053                first_fill.instrument_id
1054            );
1055            return Ok(None);
1056        };
1057
1058        let mut position = Position::new(&instrument, first_fill.clone());
1059        for fill in remaining_fills {
1060            if position.trade_ids.contains(&fill.trade_id) {
1061                anyhow::bail!(
1062                    "Duplicate fill event for position {position_id}: {}",
1063                    fill.trade_id
1064                );
1065            }
1066            position.apply(fill);
1067        }
1068
1069        Ok(Some(position))
1070    }
1071
1072    fn get_collection_key(key: &str) -> anyhow::Result<&str> {
1073        key.split_once(REDIS_DELIMITER)
1074            .map(|(collection, _)| collection)
1075            .ok_or_else(|| {
1076                anyhow::anyhow!("Invalid `key`, missing a '{REDIS_DELIMITER}' delimiter, was {key}")
1077            })
1078    }
1079
1080    async fn read_index(conn: &mut ConnectionManager, key: &str) -> anyhow::Result<Vec<Bytes>> {
1081        let index_key = get_index_key(key)?;
1082        match index_key {
1083            INDEX_ORDER_IDS
1084            | INDEX_ORDERS
1085            | INDEX_ORDERS_OPEN
1086            | INDEX_ORDERS_CLOSED
1087            | INDEX_ORDERS_EMULATED
1088            | INDEX_ORDERS_INFLIGHT
1089            | INDEX_POSITIONS
1090            | INDEX_POSITIONS_OPEN
1091            | INDEX_POSITIONS_CLOSED => Self::read_set(conn, key).await,
1092            INDEX_ORDER_POSITION | INDEX_ORDER_CLIENT => Self::read_hset(conn, key).await,
1093            _ => anyhow::bail!("Index unknown '{index_key}' on read"),
1094        }
1095    }
1096
1097    async fn read_string(conn: &mut ConnectionManager, key: &str) -> anyhow::Result<Vec<Bytes>> {
1098        let result: Vec<u8> = conn.get(key).await?;
1099
1100        if result.is_empty() {
1101            Ok(vec![])
1102        } else {
1103            Ok(vec![Bytes::from(result)])
1104        }
1105    }
1106
1107    async fn read_set(conn: &mut ConnectionManager, key: &str) -> anyhow::Result<Vec<Bytes>> {
1108        let result: Vec<Bytes> = conn.smembers(key).await?;
1109        Ok(result)
1110    }
1111
1112    async fn read_hset(conn: &mut ConnectionManager, key: &str) -> anyhow::Result<Vec<Bytes>> {
1113        let result: HashMap<String, String> = conn.hgetall(key).await?;
1114        let json = serde_json::to_string(&result)?;
1115        Ok(vec![Bytes::from(json.into_bytes())])
1116    }
1117
1118    async fn read_list(conn: &mut ConnectionManager, key: &str) -> anyhow::Result<Vec<Bytes>> {
1119        let result: Vec<Bytes> = conn.lrange(key, 0, -1).await?;
1120        Ok(result)
1121    }
1122}
1123
1124fn parse_instrument_key(key: &str, prefix: &str) -> anyhow::Result<InstrumentId> {
1125    let value = key
1126        .strip_prefix(prefix)
1127        .filter(|value| !value.is_empty())
1128        .ok_or_else(|| anyhow::anyhow!("Invalid instrument key '{key}'"))?;
1129    InstrumentId::from_str(value)
1130        .map_err(|e| anyhow::anyhow!("Failed to parse instrument ID from key '{key}': {e}"))
1131}
1132
1133fn is_timestamp_field(key: &str) -> bool {
1134    let expire_match = key == "expire_time_ns";
1135    let ts_match = key.starts_with("ts_");
1136    expire_match || ts_match
1137}
1138
1139fn convert_timestamps(value: &mut Value) {
1140    match value {
1141        Value::Object(map) => {
1142            for (key, v) in map {
1143                if is_timestamp_field(key)
1144                    && let Value::Number(n) = v
1145                    && let Some(n) = n.as_u64()
1146                {
1147                    let dt = Timestamp::from_nanosecond(i128::from(n))
1148                        .expect("UnixNanos is within Jiff's timestamp range");
1149                    *v = Value::String(format!("{dt:.9}"));
1150                }
1151                convert_timestamps(v);
1152            }
1153        }
1154        Value::Array(arr) => {
1155            for item in arr {
1156                convert_timestamps(item);
1157            }
1158        }
1159        _ => {}
1160    }
1161}
1162
1163fn convert_timestamp_strings(value: &mut Value) {
1164    match value {
1165        Value::Object(map) => {
1166            for (key, v) in map {
1167                if is_timestamp_field(key)
1168                    && let Value::String(s) = v
1169                    && let Ok(dt) = s.parse::<Timestamp>()
1170                {
1171                    let nanos = u64::try_from(dt.as_nanosecond()).expect("Invalid timestamp");
1172                    *v = Value::Number(nanos.into());
1173                }
1174                convert_timestamp_strings(v);
1175            }
1176        }
1177        Value::Array(arr) => {
1178            for item in arr {
1179                convert_timestamp_strings(item);
1180            }
1181        }
1182        _ => {}
1183    }
1184}
1185
1186#[cfg(test)]
1187mod tests {
1188    use std::str::FromStr;
1189
1190    use bytes::Bytes;
1191    use nautilus_common::enums::SerializationEncoding;
1192    use nautilus_core::UnixNanos;
1193    use nautilus_model::{
1194        enums::{AccountType, CurrencyType},
1195        events::AccountState,
1196        identifiers::{AccountId, InstrumentId},
1197        types::{AccountBalance, Currency, Money},
1198    };
1199    use rstest::rstest;
1200    use serde::Deserialize;
1201
1202    use super::{CurrencyDecode, CurrencyRecord, DatabaseQueries, parse_instrument_key};
1203
1204    /// A currency the process never registered must survive the round trip.
1205    ///
1206    /// This is the whole point of the record: `Currency`'s own encoding is the bare code, which
1207    /// only resolves against the global registry.
1208    #[rstest]
1209    #[case(SerializationEncoding::Json)]
1210    #[case(SerializationEncoding::MsgPack)]
1211    fn test_currency_record_round_trips_an_unregistered_code(
1212        #[case] encoding: SerializationEncoding,
1213    ) {
1214        // A distinct code per encoding: `CURRENCY_MAP` is process-global and tests run in parallel.
1215        let code = match encoding {
1216            SerializationEncoding::MsgPack => "ZZREDISM",
1217            _ => "ZZREDISJ",
1218        };
1219        assert!(
1220            Currency::try_from_str(code).is_none(),
1221            "the fixture code must be unregistered for this test to mean anything"
1222        );
1223
1224        let currency = Currency::new(code, 3, 0, "Redis Fixture", CurrencyType::Crypto);
1225        let payload =
1226            DatabaseQueries::serialize_payload(encoding, &CurrencyRecord::from(&currency)).unwrap();
1227
1228        let CurrencyDecode::Currency(restored) =
1229            DatabaseQueries::decode_currency(encoding, &Bytes::from(payload))
1230        else {
1231            panic!("a valid record decodes to a currency");
1232        };
1233
1234        assert_eq!(restored.code.as_str(), code);
1235        assert_eq!(restored.precision, 3);
1236        assert_eq!(restored.iso4217, 0);
1237        assert_eq!(restored.name.as_str(), "Redis Fixture");
1238        assert_eq!(restored.currency_type, CurrencyType::Crypto);
1239    }
1240
1241    /// A record written before the full record existed must keep loading.
1242    #[rstest]
1243    #[case(SerializationEncoding::Json)]
1244    #[case(SerializationEncoding::MsgPack)]
1245    fn test_legacy_bare_code_still_loads(#[case] encoding: SerializationEncoding) {
1246        let payload = DatabaseQueries::serialize_payload(encoding, &Currency::USD()).unwrap();
1247
1248        let CurrencyDecode::Currency(restored) =
1249            DatabaseQueries::decode_currency(encoding, &Bytes::from(payload))
1250        else {
1251            panic!("a registered bare code decodes to a currency");
1252        };
1253
1254        assert_eq!(restored, Currency::USD());
1255    }
1256
1257    /// A record that decodes but fails validation is an error, not a skip.
1258    #[rstest]
1259    #[case(SerializationEncoding::Json)]
1260    #[case(SerializationEncoding::MsgPack)]
1261    fn test_invalid_record_is_classified_as_invalid(#[case] encoding: SerializationEncoding) {
1262        let record = CurrencyRecord {
1263            code: "ZZQ1".to_string(),
1264            precision: 20,
1265            iso4217: 0,
1266            name: "ZZQ1".to_string(),
1267            currency_type: CurrencyType::Crypto,
1268        };
1269        let bytes = Bytes::from(DatabaseQueries::serialize_payload(encoding, &record).unwrap());
1270
1271        assert!(matches!(
1272            DatabaseQueries::decode_currency(encoding, &bytes),
1273            CurrencyDecode::InvalidRecord(_)
1274        ));
1275    }
1276
1277    /// A record-shaped payload that does not decode as a record is an error, not a bare code.
1278    ///
1279    /// A record written by a build with a currency type this one lacks must fail the load rather
1280    /// than vanish, or the payloads denominated in it vanish with it.
1281    #[rstest]
1282    #[case(SerializationEncoding::Json)]
1283    #[case(SerializationEncoding::MsgPack)]
1284    fn test_record_with_an_unknown_currency_type_is_invalid(
1285        #[case] encoding: SerializationEncoding,
1286    ) {
1287        let record = serde_json::json!({
1288            "code": "ZZQ3",
1289            "precision": 6,
1290            "iso4217": 0,
1291            "name": "ZZQ3",
1292            "currency_type": "STABLECOIN",
1293        });
1294        let bytes = Bytes::from(DatabaseQueries::serialize_payload(encoding, &record).unwrap());
1295
1296        assert!(matches!(
1297            DatabaseQueries::decode_currency(encoding, &bytes),
1298            CurrencyDecode::InvalidRecord(_)
1299        ));
1300    }
1301
1302    /// An unregistered bare code is skipped by bulk loading: the entry holds nothing recoverable.
1303    #[rstest]
1304    #[case(SerializationEncoding::Json)]
1305    #[case(SerializationEncoding::MsgPack)]
1306    fn test_unknown_bare_code_is_classified_as_undecodable(
1307        #[case] encoding: SerializationEncoding,
1308    ) {
1309        let bytes = Bytes::from(DatabaseQueries::serialize_payload(encoding, &"ZZQ2").unwrap());
1310
1311        let CurrencyDecode::Undecodable(error) = DatabaseQueries::decode_currency(encoding, &bytes)
1312        else {
1313            panic!("an unregistered bare code is undecodable");
1314        };
1315        assert!(
1316            error.to_string().contains("ZZQ2"),
1317            "the error names the code so the operator can register it: {error}"
1318        );
1319    }
1320
1321    #[derive(Debug, Deserialize, PartialEq, Eq)]
1322    struct TimestampPayload {
1323        ts_event: UnixNanos,
1324        ts_init: UnixNanos,
1325    }
1326
1327    #[rstest]
1328    #[case(SerializationEncoding::Json)]
1329    #[case(SerializationEncoding::MsgPack)]
1330    fn test_deserialize_chrono_timestamp_payload(#[case] encoding: SerializationEncoding) {
1331        let json = include_bytes!("../../test_data/redis_cache_timestamp_chrono.json");
1332        let payload = match encoding {
1333            SerializationEncoding::Json => json.to_vec(),
1334            SerializationEncoding::MsgPack => {
1335                let value = serde_json::from_slice::<serde_json::Value>(json).unwrap();
1336                rmp_serde::to_vec(&value).unwrap()
1337            }
1338            _ => unreachable!(),
1339        };
1340
1341        let result =
1342            DatabaseQueries::deserialize_payload::<TimestampPayload>(encoding, &payload).unwrap();
1343
1344        assert_eq!(
1345            result,
1346            TimestampPayload {
1347                ts_event: UnixNanos::from(1_123_456_789),
1348                ts_init: UnixNanos::from(2_987_654_321),
1349            }
1350        );
1351    }
1352
1353    #[rstest]
1354    #[case(SerializationEncoding::Json)]
1355    #[case(SerializationEncoding::MsgPack)]
1356    fn test_wallet_account_state_round_trips_unregistered_currency(
1357        #[case] encoding: SerializationEncoding,
1358    ) {
1359        let currency = Currency::new(
1360            "ENG729C",
1361            6,
1362            0,
1363            "Cache round-trip token",
1364            CurrencyType::Crypto,
1365        );
1366        let total = Money::from_mantissa_exponent(123_456_789, -6, currency);
1367        let state = AccountState::new(
1368            AccountId::new("WALLET-CACHE-001"),
1369            AccountType::Wallet,
1370            vec![AccountBalance::new(total, Money::zero(currency), total)],
1371            vec![],
1372            true,
1373            nautilus_core::UUID4::new(),
1374            UnixNanos::from(1),
1375            UnixNanos::from(2),
1376            None,
1377        );
1378        assert!(Currency::try_from_str("ENG729C").is_none());
1379
1380        let payload = DatabaseQueries::serialize_payload(encoding, &state).unwrap();
1381        let restored: AccountState =
1382            DatabaseQueries::deserialize_payload(encoding, &payload).unwrap();
1383        let restored = restored.balances[0];
1384
1385        assert_eq!(restored.total.raw(), total.raw());
1386        assert_eq!(restored.locked.raw(), 0);
1387        assert_eq!(restored.free.raw(), total.raw());
1388        assert_eq!(restored.currency.code, currency.code);
1389        assert_eq!(restored.currency.precision, currency.precision);
1390        assert_eq!(restored.currency.iso4217, currency.iso4217);
1391        assert_eq!(restored.currency.name, currency.name);
1392        assert_eq!(restored.currency.currency_type, currency.currency_type);
1393        assert!(Currency::try_from_str("ENG729C").is_none());
1394    }
1395
1396    #[rstest]
1397    #[case("0xC31E54c7a869B9FcBEcc14363CF510d1c41fa443.Arbitrum:UniswapV3")]
1398    #[case(concat!(
1399        "0xc9bc8043294146424a4e4607d8ad837d",
1400        "6a659142822bbaaabc83bb57e7447461.Arbitrum:UniswapV4",
1401    ))]
1402    fn test_parse_instrument_key_preserves_colons_in_venue(#[case] value: &str) {
1403        let prefix = "TRADER-001:instruments:";
1404        let key = format!("{prefix}{value}");
1405
1406        let result = parse_instrument_key(&key, prefix).unwrap();
1407
1408        assert_eq!(result, InstrumentId::from_str(value).unwrap());
1409    }
1410}
nautilus_infrastructure/redis/queries.rs

nautilus_infrastructure/redis/
queries.rs
