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 chrono::{DateTime, Utc};
21use futures::future::join_all;
22use nautilus_common::{cache::database::CacheMap, enums::SerializationEncoding};
23use nautilus_model::{
24    accounts::AccountAny,
25    data::{CustomData, DataType, HasTsInit},
26    events::{AccountState, OrderEventAny, OrderFilled},
27    identifiers::{AccountId, ClientId, ClientOrderId, InstrumentId, PositionId},
28    instruments::{InstrumentAny, SyntheticInstrument},
29    orders::OrderAny,
30    position::Position,
31    types::Currency,
32};
33use redis::{AsyncCommands, aio::ConnectionManager};
34use serde::{Serialize, de::DeserializeOwned};
35use serde_json::Value;
36use ustr::Ustr;
37
38use super::get_index_key;
39
40// Collection keys
41const INDEX: &str = "index";
42const GENERAL: &str = "general";
43const CURRENCIES: &str = "currencies";
44const INSTRUMENTS: &str = "instruments";
45const SYNTHETICS: &str = "synthetics";
46const ACCOUNTS: &str = "accounts";
47const ORDERS: &str = "orders";
48const POSITIONS: &str = "positions";
49const ACTORS: &str = "actors";
50const STRATEGIES: &str = "strategies";
51const CUSTOM: &str = "custom";
52const REDIS_DELIMITER: char = ':';
53
54// Index keys
55const INDEX_ORDER_IDS: &str = "index:order_ids";
56const INDEX_ORDER_POSITION: &str = "index:order_position";
57const INDEX_ORDER_CLIENT: &str = "index:order_client";
58const INDEX_ORDERS: &str = "index:orders";
59const INDEX_ORDERS_OPEN: &str = "index:orders_open";
60const INDEX_ORDERS_CLOSED: &str = "index:orders_closed";
61const INDEX_ORDERS_EMULATED: &str = "index:orders_emulated";
62const INDEX_ORDERS_INFLIGHT: &str = "index:orders_inflight";
63const INDEX_POSITIONS: &str = "index:positions";
64const INDEX_POSITIONS_OPEN: &str = "index:positions_open";
65const INDEX_POSITIONS_CLOSED: &str = "index:positions_closed";
66
67#[derive(Debug)]
68pub struct DatabaseQueries;
69
70impl DatabaseQueries {
71    /// Serializes the given `payload` using the specified `encoding` to a byte vector.
72    ///
73    /// # Errors
74    ///
75    /// Returns an error if serialization to the chosen encoding fails.
76    pub fn serialize_payload<T: Serialize>(
77        encoding: SerializationEncoding,
78        payload: &T,
79    ) -> anyhow::Result<Vec<u8>> {
80        match encoding {
81            SerializationEncoding::MsgPack => {
82                let mut value = serde_json::to_value(payload)?;
83                convert_timestamps(&mut value);
84                rmp_serde::to_vec(&value)
85                    .map_err(|e| anyhow::anyhow!("Failed to serialize msgpack `payload`: {e}"))
86            }
87            SerializationEncoding::Json => {
88                let mut value = serde_json::to_value(payload)?;
89                convert_timestamps(&mut value);
90                serde_json::to_vec(&value)
91                    .map_err(|e| anyhow::anyhow!("Failed to serialize json `payload`: {e}"))
92            }
93            SerializationEncoding::Sbe => {
94                anyhow::bail!("SBE encoding is not supported for Redis cache payloads")
95            }
96            SerializationEncoding::Capnp => {
97                anyhow::bail!("Cap'n Proto encoding is not supported for Redis cache payloads")
98            }
99        }
100    }
101
102    /// Deserializes the given byte slice `payload` into type `T` using the specified `encoding`.
103    ///
104    /// # Errors
105    ///
106    /// Returns an error if deserialization from the chosen encoding fails or converting to the target type fails.
107    pub fn deserialize_payload<T: DeserializeOwned>(
108        encoding: SerializationEncoding,
109        payload: &[u8],
110    ) -> anyhow::Result<T> {
111        let mut value = match encoding {
112            SerializationEncoding::MsgPack => rmp_serde::from_slice(payload)
113                .map_err(|e| anyhow::anyhow!("Failed to deserialize msgpack `payload`: {e}"))?,
114            SerializationEncoding::Json => serde_json::from_slice(payload)
115                .map_err(|e| anyhow::anyhow!("Failed to deserialize json `payload`: {e}"))?,
116            SerializationEncoding::Sbe => {
117                anyhow::bail!("SBE encoding is not supported for Redis cache payloads")
118            }
119            SerializationEncoding::Capnp => {
120                anyhow::bail!("Cap'n Proto encoding is not supported for Redis cache payloads")
121            }
122        };
123
124        convert_timestamp_strings(&mut value);
125
126        serde_json::from_value(value)
127            .map_err(|e| anyhow::anyhow!("Failed to convert value to target type: {e}"))
128    }
129
130    /// Scans Redis for keys matching the given `pattern`.
131    ///
132    /// # Errors
133    ///
134    /// Returns an error if the Redis scan operation fails.
135    pub async fn scan_keys(
136        con: &mut ConnectionManager,
137        pattern: String,
138    ) -> anyhow::Result<Vec<String>> {
139        let mut result = Vec::new();
140        let mut cursor = 0u64;
141
142        loop {
143            let scan_result: (u64, Vec<String>) = redis::cmd("SCAN")
144                .arg(cursor)
145                .arg("MATCH")
146                .arg(&pattern)
147                .arg("COUNT")
148                .arg(5000)
149                .query_async(con)
150                .await?;
151
152            let (new_cursor, keys) = scan_result;
153            result.extend(keys);
154
155            // If cursor is 0, we've completed the full scan
156            if new_cursor == 0 {
157                break;
158            }
159
160            cursor = new_cursor;
161        }
162
163        Ok(result)
164    }
165
166    /// Bulk reads multiple keys from Redis using MGET for efficiency.
167    ///
168    /// # Errors
169    ///
170    /// Returns an error if the underlying Redis MGET operation fails.
171    pub async fn read_bulk(
172        con: &ConnectionManager,
173        keys: &[String],
174    ) -> anyhow::Result<Vec<Option<Bytes>>> {
175        if keys.is_empty() {
176            return Ok(vec![]);
177        }
178
179        let mut con = con.clone();
180
181        // Use MGET to fetch all keys in a single network operation
182        let results: Vec<Option<Vec<u8>>> =
183            redis::cmd("MGET").arg(keys).query_async(&mut con).await?;
184
185        // Convert Vec<u8> to Bytes
186        let bytes_results: Vec<Option<Bytes>> = results
187            .into_iter()
188            .map(|opt| opt.map(Bytes::from))
189            .collect();
190
191        Ok(bytes_results)
192    }
193
194    /// Bulk reads multiple keys from Redis using MGET, batched into chunks.
195    ///
196    /// Keys are batched into chunks of `batch_size` to avoid exceeding Redis
197    /// request size limits on some providers.
198    ///
199    /// # Errors
200    ///
201    /// Returns an error if `batch_size` is zero or if the underlying Redis MGET operation fails.
202    pub async fn read_bulk_batched(
203        con: &ConnectionManager,
204        keys: &[String],
205        batch_size: usize,
206    ) -> anyhow::Result<Vec<Option<Bytes>>> {
207        if batch_size == 0 {
208            anyhow::bail!("`batch_size` must be greater than zero");
209        }
210
211        if keys.is_empty() {
212            return Ok(vec![]);
213        }
214
215        let mut all_results: Vec<Option<Bytes>> = Vec::with_capacity(keys.len());
216
217        for chunk in keys.chunks(batch_size) {
218            let mut con = con.clone();
219
220            let results: Vec<Option<Vec<u8>>> =
221                redis::cmd("MGET").arg(chunk).query_async(&mut con).await?;
222
223            all_results.extend(results.into_iter().map(|opt| opt.map(Bytes::from)));
224        }
225
226        Ok(all_results)
227    }
228
229    /// Reads raw byte payloads for `key` under `trader_key` from Redis.
230    ///
231    /// # Errors
232    ///
233    /// Returns an error if the underlying Redis read operation fails or if the collection is unsupported.
234    pub async fn read(
235        con: &ConnectionManager,
236        trader_key: &str,
237        key: &str,
238    ) -> anyhow::Result<Vec<Bytes>> {
239        let collection = Self::get_collection_key(key)?;
240        let full_key = format!("{trader_key}{REDIS_DELIMITER}{key}");
241
242        let mut con = con.clone();
243
244        match collection {
245            INDEX => Self::read_index(&mut con, &full_key).await,
246            GENERAL | CURRENCIES | INSTRUMENTS | SYNTHETICS | ACTORS | STRATEGIES => {
247                Self::read_string(&mut con, &full_key).await
248            }
249            ACCOUNTS | ORDERS | POSITIONS => Self::read_list(&mut con, &full_key).await,
250            _ => anyhow::bail!("Unsupported operation: `read` for collection '{collection}'"),
251        }
252    }
253
254    /// Loads all cache data (currencies, instruments, synthetics, accounts, orders, positions) for `trader_key`.
255    ///
256    /// # Errors
257    ///
258    /// Returns an error if loading any of the individual caches fails or combining data fails.
259    pub async fn load_all(
260        con: &ConnectionManager,
261        encoding: SerializationEncoding,
262        trader_key: &str,
263    ) -> anyhow::Result<CacheMap> {
264        let (currencies, instruments, synthetics, accounts, orders, positions) = tokio::try_join!(
265            Self::load_currencies(con, trader_key, encoding),
266            Self::load_instruments(con, trader_key, encoding),
267            Self::load_synthetics(con, trader_key, encoding),
268            Self::load_accounts(con, trader_key, encoding),
269            Self::load_orders(con, trader_key, encoding),
270            Self::load_positions(con, trader_key, encoding)
271        )
272        .map_err(|e| anyhow::anyhow!("Error loading cache data: {e}"))?;
273
274        // For now, we don't load greeks and yield curves from the database
275        // This will be implemented in the future
276        let greeks = AHashMap::new();
277        let yield_curves = AHashMap::new();
278
279        Ok(CacheMap {
280            currencies,
281            instruments,
282            synthetics,
283            accounts,
284            orders,
285            positions,
286            greeks,
287            yield_curves,
288        })
289    }
290
291    /// Loads all currencies for `trader_key` using the specified `encoding`.
292    ///
293    /// # Errors
294    ///
295    /// Returns an error if scanning keys or reading currency data fails.
296    pub async fn load_currencies(
297        con: &ConnectionManager,
298        trader_key: &str,
299        encoding: SerializationEncoding,
300    ) -> anyhow::Result<AHashMap<Ustr, Currency>> {
301        let mut currencies = AHashMap::new();
302        let pattern = format!("{trader_key}{REDIS_DELIMITER}{CURRENCIES}*");
303        log::debug!("Loading {pattern}");
304
305        let mut con = con.clone();
306        let keys = Self::scan_keys(&mut con, pattern).await?;
307
308        if keys.is_empty() {
309            return Ok(currencies);
310        }
311
312        // Use bulk loading with MGET for efficiency
313        let bulk_values = Self::read_bulk(&con, &keys).await?;
314
315        // Process the bulk results
316        for (key, value_opt) in keys.iter().zip(bulk_values.iter()) {
317            let currency_code = if let Some(code) = key.as_str().rsplit(':').next() {
318                Ustr::from(code)
319            } else {
320                log::error!("Invalid key format: {key}");
321                continue;
322            };
323
324            if let Some(value_bytes) = value_opt {
325                match Self::deserialize_payload(encoding, value_bytes) {
326                    Ok(currency) => {
327                        currencies.insert(currency_code, currency);
328                    }
329                    Err(e) => {
330                        log::error!("Failed to deserialize currency {currency_code}: {e}");
331                    }
332                }
333            } else {
334                log::error!("Currency not found in Redis: {currency_code}");
335            }
336        }
337
338        log::debug!("Loaded {} currencies(s)", currencies.len());
339
340        Ok(currencies)
341    }
342
343    /// Loads all instruments for `trader_key` using the specified `encoding`.
344    ///
345    /// # Errors
346    ///
347    /// Returns an error if scanning keys or reading instrument data fails.
348    /// Loads all instruments for `trader_key` using the specified `encoding`.
349    ///
350    /// # Errors
351    ///
352    /// Returns an error if scanning keys or reading instrument data fails.
353    pub async fn load_instruments(
354        con: &ConnectionManager,
355        trader_key: &str,
356        encoding: SerializationEncoding,
357    ) -> anyhow::Result<AHashMap<InstrumentId, InstrumentAny>> {
358        let mut instruments = AHashMap::new();
359        let pattern = format!("{trader_key}{REDIS_DELIMITER}{INSTRUMENTS}*");
360        log::debug!("Loading {pattern}");
361
362        let mut con = con.clone();
363        let keys = Self::scan_keys(&mut con, pattern).await?;
364
365        let futures: Vec<_> = keys
366            .iter()
367            .map(|key| {
368                let con = con.clone();
369                async move {
370                    let instrument_id = key
371                        .as_str()
372                        .rsplit(':')
373                        .next()
374                        .ok_or_else(|| {
375                            log::error!("Invalid key format: {key}");
376                            "Invalid key format"
377                        })
378                        .and_then(|code| {
379                            InstrumentId::from_str(code).map_err(|e| {
380                                log::error!("Failed to convert to InstrumentId for {key}: {e}");
381                                "Invalid instrument ID"
382                            })
383                        });
384
385                    let Ok(instrument_id) = instrument_id else {
386                        return None;
387                    };
388
389                    match Self::load_instrument(&con, trader_key, &instrument_id, encoding).await {
390                        Ok(Some(instrument)) => Some((instrument_id, instrument)),
391                        Ok(None) => {
392                            log::error!("Instrument not found: {instrument_id}");
393                            None
394                        }
395                        Err(e) => {
396                            log::error!("Failed to load instrument {instrument_id}: {e}");
397                            None
398                        }
399                    }
400                }
401            })
402            .collect();
403
404        // Insert all Instrument_id (key) and Instrument (value) into the HashMap, filtering out None values.
405        instruments.extend(join_all(futures).await.into_iter().flatten());
406        log::debug!("Loaded {} instruments(s)", instruments.len());
407
408        Ok(instruments)
409    }
410
411    /// Loads all synthetic instruments for `trader_key` using the specified `encoding`.
412    ///
413    /// # Errors
414    ///
415    /// Returns an error if scanning keys or reading synthetic instrument data fails.
416    /// Loads all synthetic instruments for `trader_key` using the specified `encoding`.
417    ///
418    /// # Errors
419    ///
420    /// Returns an error if scanning keys or reading synthetic instrument data fails.
421    pub async fn load_synthetics(
422        con: &ConnectionManager,
423        trader_key: &str,
424        encoding: SerializationEncoding,
425    ) -> anyhow::Result<AHashMap<InstrumentId, SyntheticInstrument>> {
426        let mut synthetics = AHashMap::new();
427        let pattern = format!("{trader_key}{REDIS_DELIMITER}{SYNTHETICS}*");
428        log::debug!("Loading {pattern}");
429
430        let mut con = con.clone();
431        let keys = Self::scan_keys(&mut con, pattern).await?;
432
433        let futures: Vec<_> = keys
434            .iter()
435            .map(|key| {
436                let con = con.clone();
437                async move {
438                    let instrument_id = key
439                        .as_str()
440                        .rsplit(':')
441                        .next()
442                        .ok_or_else(|| {
443                            log::error!("Invalid key format: {key}");
444                            "Invalid key format"
445                        })
446                        .and_then(|code| {
447                            InstrumentId::from_str(code).map_err(|e| {
448                                log::error!("Failed to parse InstrumentId for {key}: {e}");
449                                "Invalid instrument ID"
450                            })
451                        });
452
453                    let Ok(instrument_id) = instrument_id else {
454                        return None;
455                    };
456
457                    match Self::load_synthetic(&con, trader_key, &instrument_id, encoding).await {
458                        Ok(Some(synthetic)) => Some((instrument_id, synthetic)),
459                        Ok(None) => {
460                            log::error!("Synthetic not found: {instrument_id}");
461                            None
462                        }
463                        Err(e) => {
464                            log::error!("Failed to load synthetic {instrument_id}: {e}");
465                            None
466                        }
467                    }
468                }
469            })
470            .collect();
471
472        // Insert all Instrument_id (key) and Synthetic (value) into the HashMap, filtering out None values.
473        synthetics.extend(join_all(futures).await.into_iter().flatten());
474        log::debug!("Loaded {} synthetics(s)", synthetics.len());
475
476        Ok(synthetics)
477    }
478
479    /// Loads all accounts for `trader_key` using the specified `encoding`.
480    ///
481    /// # Errors
482    ///
483    /// Returns an error if scanning keys or reading account data fails.
484    /// Loads all accounts for `trader_key` using the specified `encoding`.
485    ///
486    /// # Errors
487    ///
488    /// Returns an error if scanning keys or reading account data fails.
489    pub async fn load_accounts(
490        con: &ConnectionManager,
491        trader_key: &str,
492        encoding: SerializationEncoding,
493    ) -> anyhow::Result<AHashMap<AccountId, AccountAny>> {
494        let mut accounts = AHashMap::new();
495        let pattern = format!("{trader_key}{REDIS_DELIMITER}{ACCOUNTS}*");
496        log::debug!("Loading {pattern}");
497
498        let mut con = con.clone();
499        let keys = Self::scan_keys(&mut con, pattern).await?;
500
501        let futures: Vec<_> = keys
502            .iter()
503            .map(|key| {
504                let con = con.clone();
505                async move {
506                    let account_id = if let Some(code) = key.as_str().rsplit(':').next() {
507                        AccountId::from(code)
508                    } else {
509                        log::error!("Invalid key format: {key}");
510                        return None;
511                    };
512
513                    match Self::load_account(&con, trader_key, &account_id, encoding).await {
514                        Ok(Some(account)) => Some((account_id, account)),
515                        Ok(None) => {
516                            log::error!("Account not found: {account_id}");
517                            None
518                        }
519                        Err(e) => {
520                            log::error!("Failed to load account {account_id}: {e}");
521                            None
522                        }
523                    }
524                }
525            })
526            .collect();
527
528        // Insert all Account_id (key) and Account (value) into the HashMap, filtering out None values.
529        accounts.extend(join_all(futures).await.into_iter().flatten());
530        log::debug!("Loaded {} accounts(s)", accounts.len());
531
532        Ok(accounts)
533    }
534
535    /// Loads all orders for `trader_key` using the specified `encoding`.
536    ///
537    /// # Errors
538    ///
539    /// Returns an error if scanning keys or reading order data fails.
540    /// Loads all orders for `trader_key` using the specified `encoding`.
541    ///
542    /// # Errors
543    ///
544    /// Returns an error if scanning keys or reading order data fails.
545    pub async fn load_orders(
546        con: &ConnectionManager,
547        trader_key: &str,
548        encoding: SerializationEncoding,
549    ) -> anyhow::Result<AHashMap<ClientOrderId, OrderAny>> {
550        let mut orders = AHashMap::new();
551        let pattern = format!("{trader_key}{REDIS_DELIMITER}{ORDERS}*");
552        log::debug!("Loading {pattern}");
553
554        let mut con = con.clone();
555        let keys = Self::scan_keys(&mut con, pattern).await?;
556
557        let futures: Vec<_> = keys
558            .iter()
559            .map(|key| {
560                let con = con.clone();
561                async move {
562                    let client_order_id = if let Some(code) = key.as_str().rsplit(':').next() {
563                        ClientOrderId::from(code)
564                    } else {
565                        log::error!("Invalid key format: {key}");
566                        return None;
567                    };
568
569                    match Self::load_order(&con, trader_key, &client_order_id, encoding).await {
570                        Ok(Some(order)) => Some((client_order_id, order)),
571                        Ok(None) => {
572                            log::error!("Order not found: {client_order_id}");
573                            None
574                        }
575                        Err(e) => {
576                            log::error!("Failed to load order {client_order_id}: {e}");
577                            None
578                        }
579                    }
580                }
581            })
582            .collect();
583
584        // Insert all Client-Order-Id (key) and Order (value) into the HashMap, filtering out None values.
585        orders.extend(join_all(futures).await.into_iter().flatten());
586        log::debug!("Loaded {} order(s)", orders.len());
587
588        Ok(orders)
589    }
590
591    /// Loads all positions for `trader_key` using the specified `encoding`.
592    ///
593    /// # Errors
594    ///
595    /// Returns an error if scanning keys or reading position data fails.
596    /// Loads all positions for `trader_key` using the specified `encoding`.
597    ///
598    /// # Errors
599    ///
600    /// Returns an error if scanning keys or reading position data fails.
601    pub async fn load_positions(
602        con: &ConnectionManager,
603        trader_key: &str,
604        encoding: SerializationEncoding,
605    ) -> anyhow::Result<AHashMap<PositionId, Position>> {
606        let mut positions = AHashMap::new();
607        let pattern = format!("{trader_key}{REDIS_DELIMITER}{POSITIONS}*");
608        log::debug!("Loading {pattern}");
609
610        let mut con = con.clone();
611        let keys = Self::scan_keys(&mut con, pattern).await?;
612
613        let futures: Vec<_> = keys
614            .iter()
615            .map(|key| {
616                let con = con.clone();
617                async move {
618                    let position_id = if let Some(code) = key.as_str().rsplit(':').next() {
619                        PositionId::from(code)
620                    } else {
621                        log::error!("Invalid key format: {key}");
622                        return None;
623                    };
624
625                    match Self::load_position(&con, trader_key, &position_id, encoding).await {
626                        Ok(Some(position)) => Some((position_id, position)),
627                        Ok(None) => {
628                            log::error!("Position not found: {position_id}");
629                            None
630                        }
631                        Err(e) => {
632                            log::error!("Failed to load position {position_id}: {e}");
633                            None
634                        }
635                    }
636                }
637            })
638            .collect();
639
640        // Insert all Position_id (key) and Position (value) into the HashMap, filtering out None values.
641        positions.extend(join_all(futures).await.into_iter().flatten());
642        log::debug!("Loaded {} position(s)", positions.len());
643
644        Ok(positions)
645    }
646
647    /// Loads the order ID to position ID index for `trader_key`.
648    ///
649    /// # Errors
650    ///
651    /// Returns an error if reading or parsing the index fails.
652    pub async fn load_index_order_position(
653        con: &ConnectionManager,
654        trader_key: &str,
655    ) -> anyhow::Result<AHashMap<ClientOrderId, PositionId>> {
656        let index = Self::read_index_hash(con, trader_key, INDEX_ORDER_POSITION).await?;
657        Ok(index
658            .into_iter()
659            .map(|(k, v)| {
660                (
661                    ClientOrderId::from(k.as_str()),
662                    PositionId::from(v.as_str()),
663                )
664            })
665            .collect())
666    }
667
668    /// Loads the order ID to execution client ID index for `trader_key`.
669    ///
670    /// # Errors
671    ///
672    /// Returns an error if reading or parsing the index fails.
673    pub async fn load_index_order_client(
674        con: &ConnectionManager,
675        trader_key: &str,
676    ) -> anyhow::Result<AHashMap<ClientOrderId, ClientId>> {
677        let index = Self::read_index_hash(con, trader_key, INDEX_ORDER_CLIENT).await?;
678        Ok(index
679            .into_iter()
680            .map(|(k, v)| (ClientOrderId::from(k.as_str()), ClientId::from(v.as_str())))
681            .collect())
682    }
683
684    async fn read_index_hash(
685        con: &ConnectionManager,
686        trader_key: &str,
687        key: &str,
688    ) -> anyhow::Result<HashMap<String, String>> {
689        let result = Self::read(con, trader_key, key).await?;
690        if result.is_empty() {
691            return Ok(HashMap::new());
692        }
693
694        serde_json::from_slice(&result[0])
695            .map_err(|e| anyhow::anyhow!("Failed to parse index hash '{key}': {e}"))
696    }
697
698    /// Loads all custom data for `trader_key` matching the given `data_type`.
699    ///
700    /// Keys are stored as `custom:<ts_init_020>:<uuid>`; value is full `CustomData` JSON.
701    /// Scans all custom keys, deserializes, filters by `type_name` (full or short), metadata,
702    /// and identifier to match SQL semantics, then sorts by `ts_init` ascending.
703    ///
704    /// # Errors
705    ///
706    /// Returns an error if scanning, bulk read, or deserialization fails.
707    pub async fn load_custom_data(
708        con: &ConnectionManager,
709        trader_key: &str,
710        data_type: &DataType,
711    ) -> anyhow::Result<Vec<CustomData>> {
712        let pattern = format!("{trader_key}{REDIS_DELIMITER}{CUSTOM}*");
713        log::debug!("Loading custom data {pattern}");
714
715        let mut con = con.clone();
716        let keys = Self::scan_keys(&mut con, pattern).await?;
717
718        if keys.is_empty() {
719            return Ok(Vec::new());
720        }
721
722        let values = Self::read_bulk(&con, &keys).await?;
723        let request_type_name = data_type.type_name();
724        let request_short = request_type_name
725            .rsplit([':', '.'])
726            .next()
727            .unwrap_or(request_type_name);
728        let request_identifier = data_type.identifier().unwrap_or("");
729
730        let mut results = Vec::new();
731
732        for value_opt in values {
733            let Some(value_bytes) = value_opt else {
734                continue;
735            };
736            let custom = match CustomData::from_json_bytes(value_bytes.as_ref()) {
737                Ok(c) => c,
738                Err(e) => {
739                    log::warn!("Failed to deserialize custom data from Redis: {e}");
740                    continue;
741                }
742            };
743            let stored_type_name = custom.data_type.type_name();
744            let type_match =
745                stored_type_name == request_type_name || stored_type_name == request_short;
746            let identifier_match =
747                custom.data_type.identifier().unwrap_or("") == request_identifier;
748            let metadata_match = match (data_type.metadata(), custom.data_type.metadata()) {
749                (None, None) => true,
750                (Some(a), Some(b)) => serde_json::to_value(a).ok() == serde_json::to_value(b).ok(),
751                _ => false,
752            };
753
754            if type_match && identifier_match && metadata_match {
755                results.push(custom);
756            }
757        }
758
759        results.sort_by_key(HasTsInit::ts_init);
760        log::debug!("Loaded {} custom data item(s)", results.len());
761        Ok(results)
762    }
763
764    /// Loads a single currency for `trader_key` and `code` using the specified `encoding`.
765    ///
766    /// # Errors
767    ///
768    /// Returns an error if the underlying read or deserialization fails.
769    pub async fn load_currency(
770        con: &ConnectionManager,
771        trader_key: &str,
772        code: &Ustr,
773        encoding: SerializationEncoding,
774    ) -> anyhow::Result<Option<Currency>> {
775        let key = format!("{CURRENCIES}{REDIS_DELIMITER}{code}");
776        let result = Self::read(con, trader_key, &key).await?;
777
778        if result.is_empty() {
779            return Ok(None);
780        }
781
782        let currency = Self::deserialize_payload(encoding, &result[0])?;
783        Ok(currency)
784    }
785
786    /// Loads a single instrument for `trader_key` and `instrument_id` using the specified `encoding`.
787    ///
788    /// # Errors
789    ///
790    /// Returns an error if the underlying read or deserialization fails.
791    pub async fn load_instrument(
792        con: &ConnectionManager,
793        trader_key: &str,
794        instrument_id: &InstrumentId,
795        encoding: SerializationEncoding,
796    ) -> anyhow::Result<Option<InstrumentAny>> {
797        let key = format!("{INSTRUMENTS}{REDIS_DELIMITER}{instrument_id}");
798        let result = Self::read(con, trader_key, &key).await?;
799        if result.is_empty() {
800            return Ok(None);
801        }
802
803        let instrument: InstrumentAny = Self::deserialize_payload(encoding, &result[0])?;
804        Ok(Some(instrument))
805    }
806
807    /// Loads a single synthetic instrument for `trader_key` and `instrument_id` using the specified `encoding`.
808    ///
809    /// # Errors
810    ///
811    /// Returns an error if the underlying read or deserialization fails.
812    pub async fn load_synthetic(
813        con: &ConnectionManager,
814        trader_key: &str,
815        instrument_id: &InstrumentId,
816        encoding: SerializationEncoding,
817    ) -> anyhow::Result<Option<SyntheticInstrument>> {
818        let key = format!("{SYNTHETICS}{REDIS_DELIMITER}{instrument_id}");
819        let result = Self::read(con, trader_key, &key).await?;
820        if result.is_empty() {
821            return Ok(None);
822        }
823
824        let synthetic: SyntheticInstrument = Self::deserialize_payload(encoding, &result[0])?;
825        Ok(Some(synthetic))
826    }
827
828    /// Loads a single account for `trader_key` and `account_id` using the specified `encoding`.
829    ///
830    /// # Errors
831    ///
832    /// Returns an error if the underlying read or deserialization fails.
833    pub async fn load_account(
834        con: &ConnectionManager,
835        trader_key: &str,
836        account_id: &AccountId,
837        encoding: SerializationEncoding,
838    ) -> anyhow::Result<Option<AccountAny>> {
839        let key = format!("{ACCOUNTS}{REDIS_DELIMITER}{account_id}");
840        let result = Self::read(con, trader_key, &key).await?;
841        if result.is_empty() {
842            return Ok(None);
843        }
844
845        let events: Vec<AccountState> = result
846            .iter()
847            .map(|payload| Self::deserialize_payload(encoding, payload))
848            .collect::<anyhow::Result<_>>()?;
849        let account = AccountAny::from_events(&events)?;
850        Ok(Some(account))
851    }
852
853    /// Loads a single order for `trader_key` and `client_order_id` using the specified `encoding`.
854    ///
855    /// # Errors
856    ///
857    /// Returns an error if the underlying read or deserialization fails.
858    pub async fn load_order(
859        con: &ConnectionManager,
860        trader_key: &str,
861        client_order_id: &ClientOrderId,
862        encoding: SerializationEncoding,
863    ) -> anyhow::Result<Option<OrderAny>> {
864        let key = format!("{ORDERS}{REDIS_DELIMITER}{client_order_id}");
865        let result = Self::read(con, trader_key, &key).await?;
866        if result.is_empty() {
867            return Ok(None);
868        }
869
870        let events: Vec<OrderEventAny> = result
871            .iter()
872            .map(|payload| Self::deserialize_payload(encoding, payload))
873            .collect::<anyhow::Result<_>>()?;
874        let order = OrderAny::from_events(events)?;
875        Ok(Some(order))
876    }
877
878    /// Loads a single position for `trader_key` and `position_id` using the specified `encoding`.
879    ///
880    /// # Errors
881    ///
882    /// Returns an error if the underlying read or deserialization fails.
883    pub async fn load_position(
884        con: &ConnectionManager,
885        trader_key: &str,
886        position_id: &PositionId,
887        encoding: SerializationEncoding,
888    ) -> anyhow::Result<Option<Position>> {
889        let key = format!("{POSITIONS}{REDIS_DELIMITER}{position_id}");
890        let result = Self::read(con, trader_key, &key).await?;
891        if result.is_empty() {
892            return Ok(None);
893        }
894
895        let fills: Vec<OrderFilled> = result
896            .iter()
897            .map(|payload| Self::deserialize_payload(encoding, payload))
898            .collect::<anyhow::Result<_>>()?;
899        let Some((first_fill, remaining_fills)) = fills.split_first() else {
900            return Ok(None);
901        };
902        let Some(instrument) =
903            Self::load_instrument(con, trader_key, &first_fill.instrument_id, encoding).await?
904        else {
905            log::error!(
906                "Instrument not found for position {position_id}: {}",
907                first_fill.instrument_id
908            );
909            return Ok(None);
910        };
911
912        let mut position = Position::new(&instrument, *first_fill);
913        for fill in remaining_fills {
914            if position.trade_ids().contains(&fill.trade_id) {
915                anyhow::bail!(
916                    "Duplicate fill event for position {position_id}: {}",
917                    fill.trade_id
918                );
919            }
920            position.apply(fill);
921        }
922
923        Ok(Some(position))
924    }
925
926    fn get_collection_key(key: &str) -> anyhow::Result<&str> {
927        key.split_once(REDIS_DELIMITER)
928            .map(|(collection, _)| collection)
929            .ok_or_else(|| {
930                anyhow::anyhow!("Invalid `key`, missing a '{REDIS_DELIMITER}' delimiter, was {key}")
931            })
932    }
933
934    async fn read_index(conn: &mut ConnectionManager, key: &str) -> anyhow::Result<Vec<Bytes>> {
935        let index_key = get_index_key(key)?;
936        match index_key {
937            INDEX_ORDER_IDS
938            | INDEX_ORDERS
939            | INDEX_ORDERS_OPEN
940            | INDEX_ORDERS_CLOSED
941            | INDEX_ORDERS_EMULATED
942            | INDEX_ORDERS_INFLIGHT
943            | INDEX_POSITIONS
944            | INDEX_POSITIONS_OPEN
945            | INDEX_POSITIONS_CLOSED => Self::read_set(conn, key).await,
946            INDEX_ORDER_POSITION | INDEX_ORDER_CLIENT => Self::read_hset(conn, key).await,
947            _ => anyhow::bail!("Index unknown '{index_key}' on read"),
948        }
949    }
950
951    async fn read_string(conn: &mut ConnectionManager, key: &str) -> anyhow::Result<Vec<Bytes>> {
952        let result: Vec<u8> = conn.get(key).await?;
953
954        if result.is_empty() {
955            Ok(vec![])
956        } else {
957            Ok(vec![Bytes::from(result)])
958        }
959    }
960
961    async fn read_set(conn: &mut ConnectionManager, key: &str) -> anyhow::Result<Vec<Bytes>> {
962        let result: Vec<Bytes> = conn.smembers(key).await?;
963        Ok(result)
964    }
965
966    async fn read_hset(conn: &mut ConnectionManager, key: &str) -> anyhow::Result<Vec<Bytes>> {
967        let result: HashMap<String, String> = conn.hgetall(key).await?;
968        let json = serde_json::to_string(&result)?;
969        Ok(vec![Bytes::from(json.into_bytes())])
970    }
971
972    async fn read_list(conn: &mut ConnectionManager, key: &str) -> anyhow::Result<Vec<Bytes>> {
973        let result: Vec<Bytes> = conn.lrange(key, 0, -1).await?;
974        Ok(result)
975    }
976}
977
978fn is_timestamp_field(key: &str) -> bool {
979    let expire_match = key == "expire_time_ns";
980    let ts_match = key.starts_with("ts_");
981    expire_match || ts_match
982}
983
984fn convert_timestamps(value: &mut Value) {
985    match value {
986        Value::Object(map) => {
987            for (key, v) in map {
988                if is_timestamp_field(key)
989                    && let Value::Number(n) = v
990                    && let Some(n) = n.as_u64()
991                {
992                    let dt = DateTime::<Utc>::from_timestamp_nanos(n.cast_signed());
993                    *v = Value::String(dt.to_rfc3339_opts(chrono::SecondsFormat::Nanos, true));
994                }
995                convert_timestamps(v);
996            }
997        }
998        Value::Array(arr) => {
999            for item in arr {
1000                convert_timestamps(item);
1001            }
1002        }
1003        _ => {}
1004    }
1005}
1006
1007fn convert_timestamp_strings(value: &mut Value) {
1008    match value {
1009        Value::Object(map) => {
1010            for (key, v) in map {
1011                if is_timestamp_field(key)
1012                    && let Value::String(s) = v
1013                    && let Ok(dt) = DateTime::parse_from_rfc3339(s)
1014                {
1015                    *v = Value::Number(
1016                        (dt.with_timezone(&Utc)
1017                            .timestamp_nanos_opt()
1018                            .expect("Invalid DateTime")
1019                            .cast_unsigned())
1020                        .into(),
1021                    );
1022                }
1023                convert_timestamp_strings(v);
1024            }
1025        }
1026        Value::Array(arr) => {
1027            for item in arr {
1028                convert_timestamp_strings(item);
1029            }
1030        }
1031        _ => {}
1032    }
1033}
nautilus_infrastructure/redis/queries.rs

nautilus_infrastructure/redis/
queries.rs
