rmqtt_storage/
storage.rs

1//! Abstract storage layer with support for multiple backends (sled, Redis, Redis Cluster, Redb)
2//!
3//! Defines core storage traits and unified interfaces for:
4//! - Key-value storage (StorageDB)
5//! - Map structures (Map)
6//! - List structures (List)
7//!
8//! Provides backend-agnostic enums (DefaultStorageDB, StorageMap, StorageList)
9//! that dispatch operations to concrete implementations based on enabled features.
10
11use core::fmt;
12
13use async_trait::async_trait;
14use serde::de::DeserializeOwned;
15use serde::Serialize;
16
17#[cfg(feature = "redb")]
18use crate::storage_redb::{RedbStorageDB, RedbStorageList, RedbStorageMap};
19#[cfg(feature = "redis")]
20use crate::storage_redis::{RedisStorageDB, RedisStorageList, RedisStorageMap};
21#[cfg(feature = "redis-cluster")]
22use crate::storage_redis_cluster::{
23    RedisStorageDB as RedisClusterStorageDB, RedisStorageList as RedisClusterStorageList,
24    RedisStorageMap as RedisClusterStorageMap,
25};
26#[cfg(feature = "sled")]
27use crate::storage_sled::{SledStorageDB, SledStorageList, SledStorageMap};
28use crate::Result;
29
30#[allow(unused_imports)]
31use crate::TimestampMillis;
32
33#[allow(unused)]
34pub(crate) const SEPARATOR: &[u8] = b"@";
35#[allow(unused)]
36pub(crate) const KEY_PREFIX: &[u8] = b"__rmqtt@";
37#[allow(unused)]
38pub(crate) const KEY_PREFIX_LEN: &[u8] = b"__rmqtt_len@";
39#[allow(unused)]
40pub(crate) const MAP_NAME_PREFIX: &[u8] = b"__rmqtt_map@";
41#[allow(unused)]
42pub(crate) const LIST_NAME_PREFIX: &[u8] = b"__rmqtt_list@";
43
44/// Type alias for storage keys
45pub type Key = Vec<u8>;
46
47/// Result type for iteration items (key-value pair)
48pub type IterItem<V> = Result<(Key, V)>;
49
50/// Asynchronous iterator trait for storage operations
51#[async_trait]
52pub trait AsyncIterator {
53    type Item;
54
55    /// Fetches the next item from the iterator
56    async fn next(&mut self) -> Option<Self::Item>;
57}
58
59/// Trait for splitting byte slices (used in sled backend)
60#[cfg(feature = "sled")]
61pub trait SplitSubslice {
62    /// Splits slice at the first occurrence of given subslice
63    fn split_subslice(&self, subslice: &[u8]) -> Option<(&[u8], &[u8])>;
64}
65
66#[cfg(feature = "sled")]
67impl SplitSubslice for [u8] {
68    fn split_subslice(&self, subslice: &[u8]) -> Option<(&[u8], &[u8])> {
69        self.windows(subslice.len())
70            .position(|window| window == subslice)
71            .map(|index| self.split_at(index + subslice.len()))
72    }
73}
74
75/// Core storage database operations
76#[async_trait]
77#[allow(clippy::len_without_is_empty)]
78pub trait StorageDB: Send + Sync {
79    /// Concrete Map type for this storage
80    type MapType: Map;
81
82    /// Concrete List type for this storage
83    type ListType: List;
84
85    /// Creates or accesses a named map.
86    ///
87    /// This method never fails — if the backend is unavailable a stub is returned
88    /// and actual errors are deferred to operations on the returned map.
89    async fn map<N: AsRef<[u8]> + Sync + Send>(
90        &self,
91        name: N,
92        expire: Option<TimestampMillis>,
93    ) -> Self::MapType;
94
95    /// Removes an entire map
96    async fn map_remove<K>(&self, name: K) -> Result<()>
97    where
98        K: AsRef<[u8]> + Sync + Send;
99
100    /// Checks if a map exists
101    async fn map_contains_key<K: AsRef<[u8]> + Sync + Send>(&self, key: K) -> Result<bool>;
102
103    /// Creates or accesses a named list.
104    ///
105    /// This method never fails — if the backend is unavailable a stub is returned
106    /// and actual errors are deferred to operations on the returned list.
107    async fn list<V: AsRef<[u8]> + Sync + Send>(
108        &self,
109        name: V,
110        expire: Option<TimestampMillis>,
111    ) -> Self::ListType;
112
113    /// Removes an entire list
114    async fn list_remove<K>(&self, name: K) -> Result<()>
115    where
116        K: AsRef<[u8]> + Sync + Send;
117
118    /// Checks if a list exists
119    async fn list_contains_key<K: AsRef<[u8]> + Sync + Send>(&self, key: K) -> Result<bool>;
120
121    /// Inserts a key-value pair
122    async fn insert<K, V>(&self, key: K, val: &V) -> Result<()>
123    where
124        K: AsRef<[u8]> + Sync + Send,
125        V: serde::ser::Serialize + Sync + Send;
126
127    /// Retrieves a value by key
128    async fn get<K, V>(&self, key: K) -> Result<Option<V>>
129    where
130        K: AsRef<[u8]> + Sync + Send,
131        V: DeserializeOwned + Sync + Send;
132
133    /// Removes a key-value pair
134    async fn remove<K>(&self, key: K) -> Result<()>
135    where
136        K: AsRef<[u8]> + Sync + Send;
137
138    /// Batch insert of multiple key-value pairs
139    async fn batch_insert<V>(&self, key_vals: Vec<(Key, V)>) -> Result<()>
140    where
141        V: serde::ser::Serialize + Sync + Send;
142
143    /// Batch removal of keys
144    async fn batch_remove(&self, keys: Vec<Key>) -> Result<()>;
145
146    /// Increments a counter value
147    async fn counter_incr<K>(&self, key: K, increment: isize) -> Result<()>
148    where
149        K: AsRef<[u8]> + Sync + Send;
150
151    /// Decrements a counter value
152    async fn counter_decr<K>(&self, key: K, increment: isize) -> Result<()>
153    where
154        K: AsRef<[u8]> + Sync + Send;
155
156    /// Gets current counter value
157    async fn counter_get<K>(&self, key: K) -> Result<Option<isize>>
158    where
159        K: AsRef<[u8]> + Sync + Send;
160
161    /// Sets counter to specific value
162    async fn counter_set<K>(&self, key: K, val: isize) -> Result<()>
163    where
164        K: AsRef<[u8]> + Sync + Send;
165
166    /// Checks if key exists
167    async fn contains_key<K: AsRef<[u8]> + Sync + Send>(&self, key: K) -> Result<bool>;
168
169    /// Gets number of items in storage (requires "len" feature)
170    #[cfg(feature = "len")]
171    async fn len(&self) -> Result<usize>;
172
173    /// Gets total storage size in bytes
174    async fn db_size(&self) -> Result<usize>;
175
176    /// Sets expiration timestamp for a key (requires "ttl" feature)
177    #[cfg(feature = "ttl")]
178    async fn expire_at<K>(&self, key: K, at: TimestampMillis) -> Result<bool>
179    where
180        K: AsRef<[u8]> + Sync + Send;
181
182    /// Sets expiration duration for a key (requires "ttl" feature)
183    #[cfg(feature = "ttl")]
184    async fn expire<K>(&self, key: K, dur: TimestampMillis) -> Result<bool>
185    where
186        K: AsRef<[u8]> + Sync + Send;
187
188    /// Gets remaining time-to-live for a key (requires "ttl" feature)
189    #[cfg(feature = "ttl")]
190    async fn ttl<K>(&self, key: K) -> Result<Option<TimestampMillis>>
191    where
192        K: AsRef<[u8]> + Sync + Send;
193
194    /// Iterates over all maps in storage
195    async fn map_iter<'a>(
196        &'a mut self,
197    ) -> Result<Box<dyn AsyncIterator<Item = Result<StorageMap>> + Send + 'a>>;
198
199    /// Iterates over all lists in storage
200    async fn list_iter<'a>(
201        &'a mut self,
202    ) -> Result<Box<dyn AsyncIterator<Item = Result<StorageList>> + Send + 'a>>;
203
204    /// Scans keys matching pattern (supports * and ? wildcards)
205    async fn scan<'a, P>(
206        &'a mut self,
207        pattern: P,
208    ) -> Result<Box<dyn AsyncIterator<Item = Result<Key>> + Send + 'a>>
209    where
210        P: AsRef<[u8]> + Send + Sync;
211
212    /// Gets storage backend information
213    async fn info(&self) -> Result<serde_json::Value>;
214}
215
216/// Map (dictionary) storage operations
217#[async_trait]
218pub trait Map: Sync + Send {
219    /// Gets the name of this map
220    fn name(&self) -> &[u8];
221
222    /// Inserts a key-value pair into the map
223    async fn insert<K, V>(&self, key: K, val: &V) -> Result<()>
224    where
225        K: AsRef<[u8]> + Sync + Send,
226        V: serde::ser::Serialize + Sync + Send + ?Sized;
227
228    /// Retrieves a value from the map
229    async fn get<K, V>(&self, key: K) -> Result<Option<V>>
230    where
231        K: AsRef<[u8]> + Sync + Send,
232        V: DeserializeOwned + Sync + Send;
233
234    /// Removes a key from the map
235    async fn remove<K>(&self, key: K) -> Result<()>
236    where
237        K: AsRef<[u8]> + Sync + Send;
238
239    /// Checks if key exists in the map
240    async fn contains_key<K: AsRef<[u8]> + Sync + Send>(&self, key: K) -> Result<bool>;
241
242    /// Gets number of items in map (requires "map_len" feature)
243    #[cfg(feature = "map_len")]
244    async fn len(&self) -> Result<usize>;
245
246    /// Checks if map is empty
247    async fn is_empty(&self) -> Result<bool>;
248
249    /// Clears all entries in the map
250    async fn clear(&self) -> Result<()>;
251
252    /// Removes a key and returns its value
253    async fn remove_and_fetch<K, V>(&self, key: K) -> Result<Option<V>>
254    where
255        K: AsRef<[u8]> + Sync + Send,
256        V: DeserializeOwned + Sync + Send;
257
258    /// Removes all keys with given prefix
259    async fn remove_with_prefix<K>(&self, prefix: K) -> Result<()>
260    where
261        K: AsRef<[u8]> + Sync + Send;
262
263    /// Batch insert of key-value pairs
264    async fn batch_insert<V>(&self, key_vals: Vec<(Key, V)>) -> Result<()>
265    where
266        V: serde::ser::Serialize + Sync + Send;
267
268    /// Batch removal of keys
269    async fn batch_remove(&self, keys: Vec<Key>) -> Result<()>;
270
271    /// Iterates over all key-value pairs
272    async fn iter<'a, V>(
273        &'a mut self,
274    ) -> Result<Box<dyn AsyncIterator<Item = IterItem<V>> + Send + 'a>>
275    where
276        V: DeserializeOwned + Sync + Send + 'a + 'static;
277
278    /// Iterates over all keys
279    async fn key_iter<'a>(
280        &'a mut self,
281    ) -> Result<Box<dyn AsyncIterator<Item = Result<Key>> + Send + 'a>>;
282
283    /// Iterates over key-value pairs with given prefix
284    async fn prefix_iter<'a, P, V>(
285        &'a mut self,
286        prefix: P,
287    ) -> Result<Box<dyn AsyncIterator<Item = IterItem<V>> + Send + 'a>>
288    where
289        P: AsRef<[u8]> + Send + Sync,
290        V: DeserializeOwned + Sync + Send + 'a + 'static;
291
292    /// Sets expiration timestamp for the entire map (requires "ttl" feature)
293    #[cfg(feature = "ttl")]
294    async fn expire_at(&self, at: TimestampMillis) -> Result<bool>;
295
296    /// Sets expiration duration for the entire map (requires "ttl" feature)
297    #[cfg(feature = "ttl")]
298    async fn expire(&self, dur: TimestampMillis) -> Result<bool>;
299
300    /// Gets remaining time-to-live for the map (requires "ttl" feature)
301    #[cfg(feature = "ttl")]
302    async fn ttl(&self) -> Result<Option<TimestampMillis>>;
303}
304
305/// List storage operations
306#[async_trait]
307pub trait List: Sync + Send {
308    /// Gets the name of this list
309    fn name(&self) -> &[u8];
310
311    /// Appends a value to the end of the list
312    async fn push<V>(&self, val: &V) -> Result<()>
313    where
314        V: serde::ser::Serialize + Sync + Send;
315
316    /// Appends multiple values to the list
317    async fn pushs<V>(&self, vals: Vec<V>) -> Result<()>
318    where
319        V: serde::ser::Serialize + Sync + Send;
320
321    /// Pushes with size limit and optional pop-front behavior
322    async fn push_limit<V>(
323        &self,
324        val: &V,
325        limit: usize,
326        pop_front_if_limited: bool,
327    ) -> Result<Option<V>>
328    where
329        V: serde::ser::Serialize + Sync + Send,
330        V: DeserializeOwned;
331
332    /// Removes and returns the first value in the list
333    async fn pop<V>(&self) -> Result<Option<V>>
334    where
335        V: DeserializeOwned + Sync + Send;
336
337    /// Retrieves all values in the list
338    async fn all<V>(&self) -> Result<Vec<V>>
339    where
340        V: DeserializeOwned + Sync + Send;
341
342    /// Gets value by index
343    async fn get_index<V>(&self, idx: usize) -> Result<Option<V>>
344    where
345        V: DeserializeOwned + Sync + Send;
346
347    /// Gets number of items in the list
348    async fn len(&self) -> Result<usize>;
349
350    /// Checks if list is empty
351    async fn is_empty(&self) -> Result<bool>;
352
353    /// Clears all items from the list
354    async fn clear(&self) -> Result<()>;
355
356    /// Iterates over all values
357    async fn iter<'a, V>(
358        &'a mut self,
359    ) -> Result<Box<dyn AsyncIterator<Item = Result<V>> + Send + 'a>>
360    where
361        V: DeserializeOwned + Sync + Send + 'a + 'static;
362
363    /// Sets expiration timestamp for the entire list (requires "ttl" feature)
364    #[cfg(feature = "ttl")]
365    async fn expire_at(&self, at: TimestampMillis) -> Result<bool>;
366
367    /// Sets expiration duration for the entire list (requires "ttl" feature)
368    #[cfg(feature = "ttl")]
369    async fn expire(&self, dur: TimestampMillis) -> Result<bool>;
370
371    /// Gets remaining time-to-live for the list (requires "ttl" feature)
372    #[cfg(feature = "ttl")]
373    async fn ttl(&self) -> Result<Option<TimestampMillis>>;
374}
375
376/// Unified storage backend enum (dispatches to concrete implementations)
377#[derive(Clone)]
378pub enum DefaultStorageDB {
379    #[cfg(feature = "sled")]
380    /// Sled database backend
381    Sled(SledStorageDB),
382    #[cfg(feature = "redis")]
383    /// Redis backend
384    Redis(RedisStorageDB),
385    #[cfg(feature = "redis-cluster")]
386    /// Redis Cluster backend
387    RedisCluster(RedisClusterStorageDB),
388    #[cfg(feature = "redb")]
389    /// Redb backend
390    Redb(RedbStorageDB),
391}
392
393impl DefaultStorageDB {
394    /// Accesses a named map
395    #[inline]
396    pub async fn map<V: AsRef<[u8]> + Sync + Send>(
397        &self,
398        name: V,
399        expire: Option<TimestampMillis>,
400    ) -> StorageMap {
401        match self {
402            #[cfg(feature = "sled")]
403            DefaultStorageDB::Sled(db) => StorageMap::Sled(db.map(name, expire).await),
404            #[cfg(feature = "redis")]
405            DefaultStorageDB::Redis(db) => StorageMap::Redis(db.map(name, expire).await),
406            #[cfg(feature = "redis-cluster")]
407            DefaultStorageDB::RedisCluster(db) => {
408                StorageMap::RedisCluster(db.map(name, expire).await)
409            }
410            #[cfg(feature = "redb")]
411            DefaultStorageDB::Redb(db) => StorageMap::Redb(db.map(name, expire).await),
412        }
413    }
414
415    /// Removes a named map
416    #[inline]
417    pub async fn map_remove<K>(&self, name: K) -> Result<()>
418    where
419        K: AsRef<[u8]> + Sync + Send,
420    {
421        match self {
422            #[cfg(feature = "sled")]
423            DefaultStorageDB::Sled(db) => db.map_remove(name).await,
424            #[cfg(feature = "redis")]
425            DefaultStorageDB::Redis(db) => db.map_remove(name).await,
426            #[cfg(feature = "redis-cluster")]
427            DefaultStorageDB::RedisCluster(db) => db.map_remove(name).await,
428            #[cfg(feature = "redb")]
429            DefaultStorageDB::Redb(db) => db.map_remove(name).await,
430        }
431    }
432
433    /// Checks if map exists
434    #[inline]
435    pub async fn map_contains_key<K: AsRef<[u8]> + Sync + Send>(&self, key: K) -> Result<bool> {
436        match self {
437            #[cfg(feature = "sled")]
438            DefaultStorageDB::Sled(db) => db.map_contains_key(key).await,
439            #[cfg(feature = "redis")]
440            DefaultStorageDB::Redis(db) => db.map_contains_key(key).await,
441            #[cfg(feature = "redis-cluster")]
442            DefaultStorageDB::RedisCluster(db) => db.map_contains_key(key).await,
443            #[cfg(feature = "redb")]
444            DefaultStorageDB::Redb(db) => db.map_contains_key(key).await,
445        }
446    }
447
448    /// Accesses a named list
449    #[inline]
450    pub async fn list<V: AsRef<[u8]> + Sync + Send>(
451        &self,
452        name: V,
453        expire: Option<TimestampMillis>,
454    ) -> StorageList {
455        match self {
456            #[cfg(feature = "sled")]
457            DefaultStorageDB::Sled(db) => StorageList::Sled(db.list(name, expire).await),
458            #[cfg(feature = "redis")]
459            DefaultStorageDB::Redis(db) => StorageList::Redis(db.list(name, expire).await),
460            #[cfg(feature = "redis-cluster")]
461            DefaultStorageDB::RedisCluster(db) => {
462                StorageList::RedisCluster(db.list(name, expire).await)
463            }
464            #[cfg(feature = "redb")]
465            DefaultStorageDB::Redb(db) => StorageList::Redb(db.list(name, expire).await),
466        }
467    }
468
469    /// Removes a named list
470    #[inline]
471    pub async fn list_remove<K>(&self, name: K) -> Result<()>
472    where
473        K: AsRef<[u8]> + Sync + Send,
474    {
475        match self {
476            #[cfg(feature = "sled")]
477            DefaultStorageDB::Sled(db) => db.list_remove(name).await,
478            #[cfg(feature = "redis")]
479            DefaultStorageDB::Redis(db) => db.list_remove(name).await,
480            #[cfg(feature = "redis-cluster")]
481            DefaultStorageDB::RedisCluster(db) => db.list_remove(name).await,
482            #[cfg(feature = "redb")]
483            DefaultStorageDB::Redb(db) => db.list_remove(name).await,
484        }
485    }
486
487    /// Checks if list exists
488    #[inline]
489    pub async fn list_contains_key<K: AsRef<[u8]> + Sync + Send>(&self, key: K) -> Result<bool> {
490        match self {
491            #[cfg(feature = "sled")]
492            DefaultStorageDB::Sled(db) => db.list_contains_key(key).await,
493            #[cfg(feature = "redis")]
494            DefaultStorageDB::Redis(db) => db.list_contains_key(key).await,
495            #[cfg(feature = "redis-cluster")]
496            DefaultStorageDB::RedisCluster(db) => db.list_contains_key(key).await,
497            #[cfg(feature = "redb")]
498            DefaultStorageDB::Redb(db) => db.list_contains_key(key).await,
499        }
500    }
501
502    /// Inserts a key-value pair
503    #[inline]
504    pub async fn insert<K, V>(&self, key: K, val: &V) -> Result<()>
505    where
506        K: AsRef<[u8]> + Sync + Send,
507        V: Serialize + Sync + Send,
508    {
509        match self {
510            #[cfg(feature = "sled")]
511            DefaultStorageDB::Sled(db) => db.insert(key, val).await,
512            #[cfg(feature = "redis")]
513            DefaultStorageDB::Redis(db) => db.insert(key, val).await,
514            #[cfg(feature = "redis-cluster")]
515            DefaultStorageDB::RedisCluster(db) => db.insert(key, val).await,
516            #[cfg(feature = "redb")]
517            DefaultStorageDB::Redb(db) => db.insert(key, val).await,
518        }
519    }
520
521    /// Retrieves a value by key
522    #[inline]
523    pub async fn get<K, V>(&self, key: K) -> Result<Option<V>>
524    where
525        K: AsRef<[u8]> + Sync + Send,
526        V: DeserializeOwned + Sync + Send,
527    {
528        match self {
529            #[cfg(feature = "sled")]
530            DefaultStorageDB::Sled(db) => db.get(key).await,
531            #[cfg(feature = "redis")]
532            DefaultStorageDB::Redis(db) => db.get(key).await,
533            #[cfg(feature = "redis-cluster")]
534            DefaultStorageDB::RedisCluster(db) => db.get(key).await,
535            #[cfg(feature = "redb")]
536            DefaultStorageDB::Redb(db) => db.get(key).await,
537        }
538    }
539
540    /// Removes a key-value pair
541    #[inline]
542    pub async fn remove<K>(&self, key: K) -> Result<()>
543    where
544        K: AsRef<[u8]> + Sync + Send,
545    {
546        match self {
547            #[cfg(feature = "sled")]
548            DefaultStorageDB::Sled(db) => db.remove(key).await,
549            #[cfg(feature = "redis")]
550            DefaultStorageDB::Redis(db) => db.remove(key).await,
551            #[cfg(feature = "redis-cluster")]
552            DefaultStorageDB::RedisCluster(db) => db.remove(key).await,
553            #[cfg(feature = "redb")]
554            DefaultStorageDB::Redb(db) => db.remove(key).await,
555        }
556    }
557
558    /// Batch insert of key-value pairs
559    #[inline]
560    pub async fn batch_insert<V>(&self, key_vals: Vec<(Key, V)>) -> Result<()>
561    where
562        V: serde::ser::Serialize + Sync + Send,
563    {
564        match self {
565            #[cfg(feature = "sled")]
566            DefaultStorageDB::Sled(db) => db.batch_insert(key_vals).await,
567            #[cfg(feature = "redis")]
568            DefaultStorageDB::Redis(db) => db.batch_insert(key_vals).await,
569            #[cfg(feature = "redis-cluster")]
570            DefaultStorageDB::RedisCluster(db) => db.batch_insert(key_vals).await,
571            #[cfg(feature = "redb")]
572            DefaultStorageDB::Redb(db) => db.batch_insert(key_vals).await,
573        }
574    }
575
576    /// Batch removal of keys
577    #[inline]
578    pub async fn batch_remove(&self, keys: Vec<Key>) -> Result<()> {
579        match self {
580            #[cfg(feature = "sled")]
581            DefaultStorageDB::Sled(db) => db.batch_remove(keys).await,
582            #[cfg(feature = "redis")]
583            DefaultStorageDB::Redis(db) => db.batch_remove(keys).await,
584            #[cfg(feature = "redis-cluster")]
585            DefaultStorageDB::RedisCluster(db) => db.batch_remove(keys).await,
586            #[cfg(feature = "redb")]
587            DefaultStorageDB::Redb(db) => db.batch_remove(keys).await,
588        }
589    }
590
591    /// Increments a counter
592    #[inline]
593    pub async fn counter_incr<K>(&self, key: K, increment: isize) -> Result<()>
594    where
595        K: AsRef<[u8]> + Sync + Send,
596    {
597        match self {
598            #[cfg(feature = "sled")]
599            DefaultStorageDB::Sled(db) => db.counter_incr(key, increment).await,
600            #[cfg(feature = "redis")]
601            DefaultStorageDB::Redis(db) => db.counter_incr(key, increment).await,
602            #[cfg(feature = "redis-cluster")]
603            DefaultStorageDB::RedisCluster(db) => db.counter_incr(key, increment).await,
604            #[cfg(feature = "redb")]
605            DefaultStorageDB::Redb(db) => db.counter_incr(key, increment).await,
606        }
607    }
608
609    /// Decrements a counter
610    #[inline]
611    pub async fn counter_decr<K>(&self, key: K, decrement: isize) -> Result<()>
612    where
613        K: AsRef<[u8]> + Sync + Send,
614    {
615        match self {
616            #[cfg(feature = "sled")]
617            DefaultStorageDB::Sled(db) => db.counter_decr(key, decrement).await,
618            #[cfg(feature = "redis")]
619            DefaultStorageDB::Redis(db) => db.counter_decr(key, decrement).await,
620            #[cfg(feature = "redis-cluster")]
621            DefaultStorageDB::RedisCluster(db) => db.counter_decr(key, decrement).await,
622            #[cfg(feature = "redb")]
623            DefaultStorageDB::Redb(db) => db.counter_decr(key, decrement).await,
624        }
625    }
626
627    /// Gets counter value
628    #[inline]
629    pub async fn counter_get<K>(&self, key: K) -> Result<Option<isize>>
630    where
631        K: AsRef<[u8]> + Sync + Send,
632    {
633        match self {
634            #[cfg(feature = "sled")]
635            DefaultStorageDB::Sled(db) => db.counter_get(key).await,
636            #[cfg(feature = "redis")]
637            DefaultStorageDB::Redis(db) => db.counter_get(key).await,
638            #[cfg(feature = "redis-cluster")]
639            DefaultStorageDB::RedisCluster(db) => db.counter_get(key).await,
640            #[cfg(feature = "redb")]
641            DefaultStorageDB::Redb(db) => db.counter_get(key).await,
642        }
643    }
644
645    /// Sets counter value
646    #[inline]
647    pub async fn counter_set<K>(&self, key: K, val: isize) -> Result<()>
648    where
649        K: AsRef<[u8]> + Sync + Send,
650    {
651        match self {
652            #[cfg(feature = "sled")]
653            DefaultStorageDB::Sled(db) => db.counter_set(key, val).await,
654            #[cfg(feature = "redis")]
655            DefaultStorageDB::Redis(db) => db.counter_set(key, val).await,
656            #[cfg(feature = "redis-cluster")]
657            DefaultStorageDB::RedisCluster(db) => db.counter_set(key, val).await,
658            #[cfg(feature = "redb")]
659            DefaultStorageDB::Redb(db) => db.counter_set(key, val).await,
660        }
661    }
662
663    /// Gets number of items (requires "len" feature)
664    #[inline]
665    #[cfg(feature = "len")]
666    pub async fn len(&self) -> Result<usize> {
667        match self {
668            #[cfg(feature = "sled")]
669            DefaultStorageDB::Sled(db) => db.len().await,
670            #[cfg(feature = "redis")]
671            DefaultStorageDB::Redis(db) => db.len().await,
672            #[cfg(feature = "redis-cluster")]
673            DefaultStorageDB::RedisCluster(db) => db.len().await,
674            #[cfg(feature = "redb")]
675            DefaultStorageDB::Redb(db) => db.len().await,
676        }
677    }
678
679    /// Gets total storage size in bytes
680    #[inline]
681    pub async fn db_size(&self) -> Result<usize> {
682        match self {
683            #[cfg(feature = "sled")]
684            DefaultStorageDB::Sled(db) => db.db_size().await,
685            #[cfg(feature = "redis")]
686            DefaultStorageDB::Redis(db) => db.db_size().await,
687            #[cfg(feature = "redis-cluster")]
688            DefaultStorageDB::RedisCluster(db) => db.db_size().await,
689            #[cfg(feature = "redb")]
690            DefaultStorageDB::Redb(db) => db.db_size().await,
691        }
692    }
693
694    /// Checks if key exists
695    #[inline]
696    pub async fn contains_key<K: AsRef<[u8]> + Sync + Send>(&self, key: K) -> Result<bool> {
697        match self {
698            #[cfg(feature = "sled")]
699            DefaultStorageDB::Sled(db) => db.contains_key(key).await,
700            #[cfg(feature = "redis")]
701            DefaultStorageDB::Redis(db) => db.contains_key(key).await,
702            #[cfg(feature = "redis-cluster")]
703            DefaultStorageDB::RedisCluster(db) => db.contains_key(key).await,
704            #[cfg(feature = "redb")]
705            DefaultStorageDB::Redb(db) => db.contains_key(key).await,
706        }
707    }
708
709    /// Sets expiration timestamp (requires "ttl" feature)
710    #[inline]
711    #[cfg(feature = "ttl")]
712    pub async fn expire_at<K>(&self, key: K, at: TimestampMillis) -> Result<bool>
713    where
714        K: AsRef<[u8]> + Sync + Send,
715    {
716        match self {
717            #[cfg(feature = "sled")]
718            DefaultStorageDB::Sled(db) => db.expire_at(key, at).await,
719            #[cfg(feature = "redis")]
720            DefaultStorageDB::Redis(db) => db.expire_at(key, at).await,
721            #[cfg(feature = "redis-cluster")]
722            DefaultStorageDB::RedisCluster(db) => db.expire_at(key, at).await,
723            #[cfg(feature = "redb")]
724            DefaultStorageDB::Redb(db) => db.expire_at(key, at).await,
725        }
726    }
727
728    /// Sets expiration duration (requires "ttl" feature)
729    #[inline]
730    #[cfg(feature = "ttl")]
731    pub async fn expire<K>(&self, key: K, dur: TimestampMillis) -> Result<bool>
732    where
733        K: AsRef<[u8]> + Sync + Send,
734    {
735        match self {
736            #[cfg(feature = "sled")]
737            DefaultStorageDB::Sled(db) => db.expire(key, dur).await,
738            #[cfg(feature = "redis")]
739            DefaultStorageDB::Redis(db) => db.expire(key, dur).await,
740            #[cfg(feature = "redis-cluster")]
741            DefaultStorageDB::RedisCluster(db) => db.expire(key, dur).await,
742            #[cfg(feature = "redb")]
743            DefaultStorageDB::Redb(db) => db.expire(key, dur).await,
744        }
745    }
746
747    /// Gets time-to-live (requires "ttl" feature)
748    #[inline]
749    #[cfg(feature = "ttl")]
750    pub async fn ttl<K>(&self, key: K) -> Result<Option<TimestampMillis>>
751    where
752        K: AsRef<[u8]> + Sync + Send,
753    {
754        match self {
755            #[cfg(feature = "sled")]
756            DefaultStorageDB::Sled(db) => db.ttl(key).await,
757            #[cfg(feature = "redis")]
758            DefaultStorageDB::Redis(db) => db.ttl(key).await,
759            #[cfg(feature = "redis-cluster")]
760            DefaultStorageDB::RedisCluster(db) => db.ttl(key).await,
761            #[cfg(feature = "redb")]
762            DefaultStorageDB::Redb(db) => db.ttl(key).await,
763        }
764    }
765
766    /// Iterates over maps
767    #[inline]
768    pub async fn map_iter<'a>(
769        &'a mut self,
770    ) -> Result<Box<dyn AsyncIterator<Item = Result<StorageMap>> + Send + 'a>> {
771        match self {
772            #[cfg(feature = "sled")]
773            DefaultStorageDB::Sled(db) => db.map_iter().await,
774            #[cfg(feature = "redis")]
775            DefaultStorageDB::Redis(db) => db.map_iter().await,
776            #[cfg(feature = "redis-cluster")]
777            DefaultStorageDB::RedisCluster(db) => db.map_iter().await,
778            #[cfg(feature = "redb")]
779            DefaultStorageDB::Redb(db) => db.map_iter().await,
780        }
781    }
782
783    /// Iterates over lists
784    #[inline]
785    pub async fn list_iter<'a>(
786        &'a mut self,
787    ) -> Result<Box<dyn AsyncIterator<Item = Result<StorageList>> + Send + 'a>> {
788        match self {
789            #[cfg(feature = "sled")]
790            DefaultStorageDB::Sled(db) => db.list_iter().await,
791            #[cfg(feature = "redis")]
792            DefaultStorageDB::Redis(db) => db.list_iter().await,
793            #[cfg(feature = "redis-cluster")]
794            DefaultStorageDB::RedisCluster(db) => db.list_iter().await,
795            #[cfg(feature = "redb")]
796            DefaultStorageDB::Redb(db) => db.list_iter().await,
797        }
798    }
799
800    /// Scans keys matching pattern
801    #[inline]
802    pub async fn scan<'a, P>(
803        &'a mut self,
804        pattern: P,
805    ) -> Result<Box<dyn AsyncIterator<Item = Result<Key>> + Send + 'a>>
806    where
807        P: AsRef<[u8]> + Send + Sync,
808    {
809        match self {
810            #[cfg(feature = "sled")]
811            DefaultStorageDB::Sled(db) => db.scan(pattern).await,
812            #[cfg(feature = "redis")]
813            DefaultStorageDB::Redis(db) => db.scan(pattern).await,
814            #[cfg(feature = "redis-cluster")]
815            DefaultStorageDB::RedisCluster(db) => db.scan(pattern).await,
816            #[cfg(feature = "redb")]
817            DefaultStorageDB::Redb(db) => db.scan(pattern).await,
818        }
819    }
820
821    /// Gets storage information
822    #[inline]
823    pub async fn info(&self) -> Result<serde_json::Value> {
824        match self {
825            #[cfg(feature = "sled")]
826            DefaultStorageDB::Sled(db) => db.info().await,
827            #[cfg(feature = "redis")]
828            DefaultStorageDB::Redis(db) => db.info().await,
829            #[cfg(feature = "redis-cluster")]
830            DefaultStorageDB::RedisCluster(db) => db.info().await,
831            #[cfg(feature = "redb")]
832            DefaultStorageDB::Redb(db) => db.info().await,
833        }
834    }
835}
836
837// ---------------------------------------------------------------------------
838// Inherent raw methods on DefaultStorageDB — NOT part of the StorageDB trait.
839// ---------------------------------------------------------------------------
840#[cfg(feature = "circuit-breaker")]
841impl DefaultStorageDB {
842    pub(crate) async fn insert_raw(&self, key: &[u8], val: &[u8]) -> Result<()> {
843        match self {
844            #[cfg(feature = "sled")]
845            DefaultStorageDB::Sled(db) => db.insert_raw(key, val).await,
846            #[cfg(feature = "redis")]
847            DefaultStorageDB::Redis(db) => db.insert_raw(key, val).await,
848            #[cfg(feature = "redis-cluster")]
849            DefaultStorageDB::RedisCluster(db) => db.insert_raw(key, val).await,
850            #[cfg(feature = "redb")]
851            DefaultStorageDB::Redb(db) => db.insert_raw(key, val).await,
852        }
853    }
854
855    pub(crate) async fn get_raw(&self, key: &[u8]) -> Result<Option<Vec<u8>>> {
856        match self {
857            #[cfg(feature = "sled")]
858            DefaultStorageDB::Sled(db) => db.get_raw(key).await,
859            #[cfg(feature = "redis")]
860            DefaultStorageDB::Redis(db) => db.get_raw(key).await,
861            #[cfg(feature = "redis-cluster")]
862            DefaultStorageDB::RedisCluster(db) => db.get_raw(key).await,
863            #[cfg(feature = "redb")]
864            DefaultStorageDB::Redb(db) => db.get_raw(key).await,
865        }
866    }
867
868    pub(crate) async fn batch_insert_raw(&self, key_vals: Vec<(Vec<u8>, Vec<u8>)>) -> Result<()> {
869        match self {
870            #[cfg(feature = "sled")]
871            DefaultStorageDB::Sled(db) => db.batch_insert_raw(key_vals).await,
872            #[cfg(feature = "redis")]
873            DefaultStorageDB::Redis(db) => db.batch_insert_raw(key_vals).await,
874            #[cfg(feature = "redis-cluster")]
875            DefaultStorageDB::RedisCluster(db) => db.batch_insert_raw(key_vals).await,
876            #[cfg(feature = "redb")]
877            DefaultStorageDB::Redb(db) => db.batch_insert_raw(key_vals).await,
878        }
879    }
880}
881
882/// Unified map implementation enum
883#[derive(Clone)]
884pub enum StorageMap {
885    #[cfg(feature = "sled")]
886    /// Sled map implementation
887    Sled(SledStorageMap),
888    #[cfg(feature = "redis")]
889    /// Redis map implementation
890    Redis(RedisStorageMap),
891    #[cfg(feature = "redis-cluster")]
892    /// Redis Cluster map implementation
893    RedisCluster(RedisClusterStorageMap),
894    #[cfg(feature = "redb")]
895    /// Redb map implementation
896    Redb(RedbStorageMap),
897}
898
899#[async_trait]
900impl Map for StorageMap {
901    fn name(&self) -> &[u8] {
902        match self {
903            #[cfg(feature = "sled")]
904            StorageMap::Sled(m) => m.name(),
905            #[cfg(feature = "redis")]
906            StorageMap::Redis(m) => m.name(),
907            #[cfg(feature = "redis-cluster")]
908            StorageMap::RedisCluster(m) => m.name(),
909            #[cfg(feature = "redb")]
910            StorageMap::Redb(m) => m.name(),
911        }
912    }
913
914    async fn insert<K, V>(&self, key: K, val: &V) -> Result<()>
915    where
916        K: AsRef<[u8]> + Sync + Send,
917        V: Serialize + Sync + Send + ?Sized,
918    {
919        match self {
920            #[cfg(feature = "sled")]
921            StorageMap::Sled(m) => m.insert(key, val).await,
922            #[cfg(feature = "redis")]
923            StorageMap::Redis(m) => m.insert(key, val).await,
924            #[cfg(feature = "redis-cluster")]
925            StorageMap::RedisCluster(m) => m.insert(key, val).await,
926            #[cfg(feature = "redb")]
927            StorageMap::Redb(m) => m.insert(key, val).await,
928        }
929    }
930
931    async fn get<K, V>(&self, key: K) -> Result<Option<V>>
932    where
933        K: AsRef<[u8]> + Sync + Send,
934        V: DeserializeOwned + Sync + Send,
935    {
936        match self {
937            #[cfg(feature = "sled")]
938            StorageMap::Sled(m) => m.get(key).await,
939            #[cfg(feature = "redis")]
940            StorageMap::Redis(m) => m.get(key).await,
941            #[cfg(feature = "redis-cluster")]
942            StorageMap::RedisCluster(m) => m.get(key).await,
943            #[cfg(feature = "redb")]
944            StorageMap::Redb(m) => m.get(key).await,
945        }
946    }
947
948    async fn remove<K>(&self, key: K) -> Result<()>
949    where
950        K: AsRef<[u8]> + Sync + Send,
951    {
952        match self {
953            #[cfg(feature = "sled")]
954            StorageMap::Sled(m) => m.remove(key).await,
955            #[cfg(feature = "redis")]
956            StorageMap::Redis(m) => m.remove(key).await,
957            #[cfg(feature = "redis-cluster")]
958            StorageMap::RedisCluster(m) => m.remove(key).await,
959            #[cfg(feature = "redb")]
960            StorageMap::Redb(m) => m.remove(key).await,
961        }
962    }
963
964    async fn contains_key<K: AsRef<[u8]> + Sync + Send>(&self, key: K) -> Result<bool> {
965        match self {
966            #[cfg(feature = "sled")]
967            StorageMap::Sled(m) => m.contains_key(key).await,
968            #[cfg(feature = "redis")]
969            StorageMap::Redis(m) => m.contains_key(key).await,
970            #[cfg(feature = "redis-cluster")]
971            StorageMap::RedisCluster(m) => m.contains_key(key).await,
972            #[cfg(feature = "redb")]
973            StorageMap::Redb(m) => m.contains_key(key).await,
974        }
975    }
976
977    #[cfg(feature = "map_len")]
978    async fn len(&self) -> Result<usize> {
979        match self {
980            #[cfg(feature = "sled")]
981            StorageMap::Sled(m) => m.len().await,
982            #[cfg(feature = "redis")]
983            StorageMap::Redis(m) => m.len().await,
984            #[cfg(feature = "redis-cluster")]
985            StorageMap::RedisCluster(m) => m.len().await,
986            #[cfg(feature = "redb")]
987            StorageMap::Redb(m) => m.len().await,
988        }
989    }
990
991    async fn is_empty(&self) -> Result<bool> {
992        match self {
993            #[cfg(feature = "sled")]
994            StorageMap::Sled(m) => m.is_empty().await,
995            #[cfg(feature = "redis")]
996            StorageMap::Redis(m) => m.is_empty().await,
997            #[cfg(feature = "redis-cluster")]
998            StorageMap::RedisCluster(m) => m.is_empty().await,
999            #[cfg(feature = "redb")]
1000            StorageMap::Redb(m) => m.is_empty().await,
1001        }
1002    }
1003
1004    async fn clear(&self) -> Result<()> {
1005        match self {
1006            #[cfg(feature = "sled")]
1007            StorageMap::Sled(m) => m.clear().await,
1008            #[cfg(feature = "redis")]
1009            StorageMap::Redis(m) => m.clear().await,
1010            #[cfg(feature = "redis-cluster")]
1011            StorageMap::RedisCluster(m) => m.clear().await,
1012            #[cfg(feature = "redb")]
1013            StorageMap::Redb(m) => m.clear().await,
1014        }
1015    }
1016
1017    async fn remove_and_fetch<K, V>(&self, key: K) -> Result<Option<V>>
1018    where
1019        K: AsRef<[u8]> + Sync + Send,
1020        V: DeserializeOwned + Sync + Send,
1021    {
1022        match self {
1023            #[cfg(feature = "sled")]
1024            StorageMap::Sled(m) => m.remove_and_fetch(key).await,
1025            #[cfg(feature = "redis")]
1026            StorageMap::Redis(m) => m.remove_and_fetch(key).await,
1027            #[cfg(feature = "redis-cluster")]
1028            StorageMap::RedisCluster(m) => m.remove_and_fetch(key).await,
1029            #[cfg(feature = "redb")]
1030            StorageMap::Redb(m) => m.remove_and_fetch(key).await,
1031        }
1032    }
1033
1034    async fn remove_with_prefix<K>(&self, prefix: K) -> Result<()>
1035    where
1036        K: AsRef<[u8]> + Sync + Send,
1037    {
1038        match self {
1039            #[cfg(feature = "sled")]
1040            StorageMap::Sled(m) => m.remove_with_prefix(prefix).await,
1041            #[cfg(feature = "redis")]
1042            StorageMap::Redis(m) => m.remove_with_prefix(prefix).await,
1043            #[cfg(feature = "redis-cluster")]
1044            StorageMap::RedisCluster(m) => m.remove_with_prefix(prefix).await,
1045            #[cfg(feature = "redb")]
1046            StorageMap::Redb(m) => m.remove_with_prefix(prefix).await,
1047        }
1048    }
1049
1050    async fn batch_insert<V>(&self, key_vals: Vec<(Key, V)>) -> Result<()>
1051    where
1052        V: Serialize + Sync + Send,
1053    {
1054        match self {
1055            #[cfg(feature = "sled")]
1056            StorageMap::Sled(m) => m.batch_insert(key_vals).await,
1057            #[cfg(feature = "redis")]
1058            StorageMap::Redis(m) => m.batch_insert(key_vals).await,
1059            #[cfg(feature = "redis-cluster")]
1060            StorageMap::RedisCluster(m) => m.batch_insert(key_vals).await,
1061            #[cfg(feature = "redb")]
1062            StorageMap::Redb(m) => m.batch_insert(key_vals).await,
1063        }
1064    }
1065
1066    async fn batch_remove(&self, keys: Vec<Key>) -> Result<()> {
1067        match self {
1068            #[cfg(feature = "sled")]
1069            StorageMap::Sled(m) => m.batch_remove(keys).await,
1070            #[cfg(feature = "redis")]
1071            StorageMap::Redis(m) => m.batch_remove(keys).await,
1072            #[cfg(feature = "redis-cluster")]
1073            StorageMap::RedisCluster(m) => m.batch_remove(keys).await,
1074            #[cfg(feature = "redb")]
1075            StorageMap::Redb(m) => m.batch_remove(keys).await,
1076        }
1077    }
1078
1079    async fn iter<'a, V>(
1080        &'a mut self,
1081    ) -> Result<Box<dyn AsyncIterator<Item = IterItem<V>> + Send + 'a>>
1082    where
1083        V: DeserializeOwned + Sync + Send + 'a + 'static,
1084    {
1085        match self {
1086            #[cfg(feature = "sled")]
1087            StorageMap::Sled(m) => m.iter().await,
1088            #[cfg(feature = "redis")]
1089            StorageMap::Redis(m) => m.iter().await,
1090            #[cfg(feature = "redis-cluster")]
1091            StorageMap::RedisCluster(m) => m.iter().await,
1092            #[cfg(feature = "redb")]
1093            StorageMap::Redb(m) => m.iter().await,
1094        }
1095    }
1096
1097    async fn key_iter<'a>(
1098        &'a mut self,
1099    ) -> Result<Box<dyn AsyncIterator<Item = Result<Key>> + Send + 'a>> {
1100        match self {
1101            #[cfg(feature = "sled")]
1102            StorageMap::Sled(m) => m.key_iter().await,
1103            #[cfg(feature = "redis")]
1104            StorageMap::Redis(m) => m.key_iter().await,
1105            #[cfg(feature = "redis-cluster")]
1106            StorageMap::RedisCluster(m) => m.key_iter().await,
1107            #[cfg(feature = "redb")]
1108            StorageMap::Redb(m) => m.key_iter().await,
1109        }
1110    }
1111
1112    async fn prefix_iter<'a, P, V>(
1113        &'a mut self,
1114        prefix: P,
1115    ) -> Result<Box<dyn AsyncIterator<Item = IterItem<V>> + Send + 'a>>
1116    where
1117        P: AsRef<[u8]> + Send + Sync,
1118        V: DeserializeOwned + Sync + Send + 'a + 'static,
1119    {
1120        match self {
1121            #[cfg(feature = "sled")]
1122            StorageMap::Sled(m) => m.prefix_iter(prefix).await,
1123            #[cfg(feature = "redis")]
1124            StorageMap::Redis(m) => m.prefix_iter(prefix).await,
1125            #[cfg(feature = "redis-cluster")]
1126            StorageMap::RedisCluster(m) => m.prefix_iter(prefix).await,
1127            #[cfg(feature = "redb")]
1128            StorageMap::Redb(m) => m.prefix_iter(prefix).await,
1129        }
1130    }
1131
1132    #[cfg(feature = "ttl")]
1133    async fn expire_at(&self, at: TimestampMillis) -> Result<bool> {
1134        match self {
1135            #[cfg(feature = "sled")]
1136            StorageMap::Sled(m) => m.expire_at(at).await,
1137            #[cfg(feature = "redis")]
1138            StorageMap::Redis(m) => m.expire_at(at).await,
1139            #[cfg(feature = "redis-cluster")]
1140            StorageMap::RedisCluster(m) => m.expire_at(at).await,
1141            #[cfg(feature = "redb")]
1142            StorageMap::Redb(m) => m.expire_at(at).await,
1143        }
1144    }
1145
1146    #[cfg(feature = "ttl")]
1147    async fn expire(&self, dur: TimestampMillis) -> Result<bool> {
1148        match self {
1149            #[cfg(feature = "sled")]
1150            StorageMap::Sled(m) => m.expire(dur).await,
1151            #[cfg(feature = "redis")]
1152            StorageMap::Redis(m) => m.expire(dur).await,
1153            #[cfg(feature = "redis-cluster")]
1154            StorageMap::RedisCluster(m) => m.expire(dur).await,
1155            #[cfg(feature = "redb")]
1156            StorageMap::Redb(m) => m.expire(dur).await,
1157        }
1158    }
1159
1160    #[cfg(feature = "ttl")]
1161    async fn ttl(&self) -> Result<Option<TimestampMillis>> {
1162        match self {
1163            #[cfg(feature = "sled")]
1164            StorageMap::Sled(m) => m.ttl().await,
1165            #[cfg(feature = "redis")]
1166            StorageMap::Redis(m) => m.ttl().await,
1167            #[cfg(feature = "redis-cluster")]
1168            StorageMap::RedisCluster(m) => m.ttl().await,
1169            #[cfg(feature = "redb")]
1170            StorageMap::Redb(m) => m.ttl().await,
1171        }
1172    }
1173}
1174
1175// ---------------------------------------------------------------------------
1176// Inherent methods on StorageMap — NOT part of the Map trait.
1177// Used internally (e.g., by circuit-breaker) to bypass serde overhead.
1178// ---------------------------------------------------------------------------
1179#[cfg(feature = "circuit-breaker")]
1180impl StorageMap {
1181    /// Inserts raw bytes directly, skipping serialization.
1182    pub(crate) async fn insert_raw(&self, key: &[u8], val: &[u8]) -> Result<()> {
1183        match self {
1184            #[cfg(feature = "sled")]
1185            StorageMap::Sled(m) => m.insert_raw(key, val).await,
1186            #[cfg(feature = "redis")]
1187            StorageMap::Redis(m) => m.insert_raw(key, val).await,
1188            #[cfg(feature = "redis-cluster")]
1189            StorageMap::RedisCluster(m) => m.insert_raw(key, val).await,
1190            #[cfg(feature = "redb")]
1191            StorageMap::Redb(m) => m.insert_raw(key, val).await,
1192        }
1193    }
1194
1195    /// Retrieves raw bytes directly, skipping deserialization.
1196    pub(crate) async fn get_raw(&self, key: &[u8]) -> Result<Option<Vec<u8>>> {
1197        match self {
1198            #[cfg(feature = "sled")]
1199            StorageMap::Sled(m) => m.get_raw(key).await,
1200            #[cfg(feature = "redis")]
1201            StorageMap::Redis(m) => m.get_raw(key).await,
1202            #[cfg(feature = "redis-cluster")]
1203            StorageMap::RedisCluster(m) => m.get_raw(key).await,
1204            #[cfg(feature = "redb")]
1205            StorageMap::Redb(m) => m.get_raw(key).await,
1206        }
1207    }
1208
1209    /// Removes and returns a raw value, skipping postcard deserialization.
1210    pub(crate) async fn remove_and_fetch_raw(&self, key: &[u8]) -> Result<Option<Vec<u8>>> {
1211        match self {
1212            #[cfg(feature = "sled")]
1213            StorageMap::Sled(m) => m.remove_and_fetch_raw(key).await,
1214            #[cfg(feature = "redis")]
1215            StorageMap::Redis(m) => m.remove_and_fetch_raw(key).await,
1216            #[cfg(feature = "redis-cluster")]
1217            StorageMap::RedisCluster(m) => m.remove_and_fetch_raw(key).await,
1218            #[cfg(feature = "redb")]
1219            StorageMap::Redb(m) => m.remove_and_fetch_raw(key).await,
1220        }
1221    }
1222
1223    /// Batch insert of raw key-value pairs, skipping serialization.
1224    pub(crate) async fn batch_insert_raw(&self, key_vals: Vec<(Vec<u8>, Vec<u8>)>) -> Result<()> {
1225        match self {
1226            #[cfg(feature = "sled")]
1227            StorageMap::Sled(m) => m.batch_insert_raw(key_vals).await,
1228            #[cfg(feature = "redis")]
1229            StorageMap::Redis(m) => m.batch_insert_raw(key_vals).await,
1230            #[cfg(feature = "redis-cluster")]
1231            StorageMap::RedisCluster(m) => m.batch_insert_raw(key_vals).await,
1232            #[cfg(feature = "redb")]
1233            StorageMap::Redb(m) => m.batch_insert_raw(key_vals).await,
1234        }
1235    }
1236}
1237
1238/// Unified list implementation enum
1239#[derive(Clone)]
1240pub enum StorageList {
1241    #[cfg(feature = "sled")]
1242    /// Sled list implementation
1243    Sled(SledStorageList),
1244    #[cfg(feature = "redis")]
1245    /// Redis list implementation
1246    Redis(RedisStorageList),
1247    #[cfg(feature = "redis-cluster")]
1248    /// Redis Cluster list implementation
1249    RedisCluster(RedisClusterStorageList),
1250    #[cfg(feature = "redb")]
1251    /// Redb list implementation
1252    Redb(RedbStorageList),
1253}
1254
1255impl fmt::Debug for StorageList {
1256    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1257        let name = match self {
1258            #[cfg(feature = "sled")]
1259            StorageList::Sled(list) => list.name(),
1260            #[cfg(feature = "redis")]
1261            StorageList::Redis(list) => list.name(),
1262            #[cfg(feature = "redis-cluster")]
1263            StorageList::RedisCluster(list) => list.name(),
1264            #[cfg(feature = "redb")]
1265            StorageList::Redb(list) => list.name(),
1266        };
1267
1268        f.debug_tuple(&format!("StorageList({:?})", String::from_utf8_lossy(name)))
1269            .finish()
1270    }
1271}
1272
1273#[async_trait]
1274impl List for StorageList {
1275    fn name(&self) -> &[u8] {
1276        match self {
1277            #[cfg(feature = "sled")]
1278            StorageList::Sled(m) => m.name(),
1279            #[cfg(feature = "redis")]
1280            StorageList::Redis(m) => m.name(),
1281            #[cfg(feature = "redis-cluster")]
1282            StorageList::RedisCluster(m) => m.name(),
1283            #[cfg(feature = "redb")]
1284            StorageList::Redb(m) => m.name(),
1285        }
1286    }
1287
1288    async fn push<V>(&self, val: &V) -> Result<()>
1289    where
1290        V: Serialize + Sync + Send,
1291    {
1292        match self {
1293            #[cfg(feature = "sled")]
1294            StorageList::Sled(list) => list.push(val).await,
1295            #[cfg(feature = "redis")]
1296            StorageList::Redis(list) => list.push(val).await,
1297            #[cfg(feature = "redis-cluster")]
1298            StorageList::RedisCluster(list) => list.push(val).await,
1299            #[cfg(feature = "redb")]
1300            StorageList::Redb(list) => list.push(val).await,
1301        }
1302    }
1303
1304    async fn pushs<V>(&self, vals: Vec<V>) -> Result<()>
1305    where
1306        V: serde::ser::Serialize + Sync + Send,
1307    {
1308        match self {
1309            #[cfg(feature = "sled")]
1310            StorageList::Sled(list) => list.pushs(vals).await,
1311            #[cfg(feature = "redis")]
1312            StorageList::Redis(list) => list.pushs(vals).await,
1313            #[cfg(feature = "redis-cluster")]
1314            StorageList::RedisCluster(list) => list.pushs(vals).await,
1315            #[cfg(feature = "redb")]
1316            StorageList::Redb(list) => list.pushs(vals).await,
1317        }
1318    }
1319
1320    async fn push_limit<V>(
1321        &self,
1322        val: &V,
1323        limit: usize,
1324        pop_front_if_limited: bool,
1325    ) -> Result<Option<V>>
1326    where
1327        V: Serialize + Sync + Send,
1328        V: DeserializeOwned,
1329    {
1330        match self {
1331            #[cfg(feature = "sled")]
1332            StorageList::Sled(list) => list.push_limit(val, limit, pop_front_if_limited).await,
1333            #[cfg(feature = "redis")]
1334            StorageList::Redis(list) => list.push_limit(val, limit, pop_front_if_limited).await,
1335            #[cfg(feature = "redis-cluster")]
1336            StorageList::RedisCluster(list) => {
1337                list.push_limit(val, limit, pop_front_if_limited).await
1338            }
1339            #[cfg(feature = "redb")]
1340            StorageList::Redb(list) => list.push_limit(val, limit, pop_front_if_limited).await,
1341        }
1342    }
1343
1344    async fn pop<V>(&self) -> Result<Option<V>>
1345    where
1346        V: DeserializeOwned + Sync + Send,
1347    {
1348        match self {
1349            #[cfg(feature = "sled")]
1350            StorageList::Sled(list) => list.pop().await,
1351            #[cfg(feature = "redis")]
1352            StorageList::Redis(list) => list.pop().await,
1353            #[cfg(feature = "redis-cluster")]
1354            StorageList::RedisCluster(list) => list.pop().await,
1355            #[cfg(feature = "redb")]
1356            StorageList::Redb(list) => list.pop().await,
1357        }
1358    }
1359
1360    async fn all<V>(&self) -> Result<Vec<V>>
1361    where
1362        V: DeserializeOwned + Sync + Send,
1363    {
1364        match self {
1365            #[cfg(feature = "sled")]
1366            StorageList::Sled(list) => list.all().await,
1367            #[cfg(feature = "redis")]
1368            StorageList::Redis(list) => list.all().await,
1369            #[cfg(feature = "redis-cluster")]
1370            StorageList::RedisCluster(list) => list.all().await,
1371            #[cfg(feature = "redb")]
1372            StorageList::Redb(list) => list.all().await,
1373        }
1374    }
1375
1376    async fn get_index<V>(&self, idx: usize) -> Result<Option<V>>
1377    where
1378        V: DeserializeOwned + Sync + Send,
1379    {
1380        match self {
1381            #[cfg(feature = "sled")]
1382            StorageList::Sled(list) => list.get_index(idx).await,
1383            #[cfg(feature = "redis")]
1384            StorageList::Redis(list) => list.get_index(idx).await,
1385            #[cfg(feature = "redis-cluster")]
1386            StorageList::RedisCluster(list) => list.get_index(idx).await,
1387            #[cfg(feature = "redb")]
1388            StorageList::Redb(list) => list.get_index(idx).await,
1389        }
1390    }
1391
1392    async fn len(&self) -> Result<usize> {
1393        match self {
1394            #[cfg(feature = "sled")]
1395            StorageList::Sled(list) => list.len().await,
1396            #[cfg(feature = "redis")]
1397            StorageList::Redis(list) => list.len().await,
1398            #[cfg(feature = "redis-cluster")]
1399            StorageList::RedisCluster(list) => list.len().await,
1400            #[cfg(feature = "redb")]
1401            StorageList::Redb(list) => list.len().await,
1402        }
1403    }
1404
1405    async fn is_empty(&self) -> Result<bool> {
1406        match self {
1407            #[cfg(feature = "sled")]
1408            StorageList::Sled(list) => list.is_empty().await,
1409            #[cfg(feature = "redis")]
1410            StorageList::Redis(list) => list.is_empty().await,
1411            #[cfg(feature = "redis-cluster")]
1412            StorageList::RedisCluster(list) => list.is_empty().await,
1413            #[cfg(feature = "redb")]
1414            StorageList::Redb(list) => list.is_empty().await,
1415        }
1416    }
1417
1418    async fn clear(&self) -> Result<()> {
1419        match self {
1420            #[cfg(feature = "sled")]
1421            StorageList::Sled(list) => list.clear().await,
1422            #[cfg(feature = "redis")]
1423            StorageList::Redis(list) => list.clear().await,
1424            #[cfg(feature = "redis-cluster")]
1425            StorageList::RedisCluster(list) => list.clear().await,
1426            #[cfg(feature = "redb")]
1427            StorageList::Redb(list) => list.clear().await,
1428        }
1429    }
1430
1431    async fn iter<'a, V>(
1432        &'a mut self,
1433    ) -> Result<Box<dyn AsyncIterator<Item = Result<V>> + Send + 'a>>
1434    where
1435        V: DeserializeOwned + Sync + Send + 'a + 'static,
1436    {
1437        match self {
1438            #[cfg(feature = "sled")]
1439            StorageList::Sled(list) => list.iter().await,
1440            #[cfg(feature = "redis")]
1441            StorageList::Redis(list) => list.iter().await,
1442            #[cfg(feature = "redis-cluster")]
1443            StorageList::RedisCluster(list) => list.iter().await,
1444            #[cfg(feature = "redb")]
1445            StorageList::Redb(list) => list.iter().await,
1446        }
1447    }
1448
1449    #[cfg(feature = "ttl")]
1450    async fn expire_at(&self, at: TimestampMillis) -> Result<bool> {
1451        match self {
1452            #[cfg(feature = "sled")]
1453            StorageList::Sled(l) => l.expire_at(at).await,
1454            #[cfg(feature = "redis")]
1455            StorageList::Redis(l) => l.expire_at(at).await,
1456            #[cfg(feature = "redis-cluster")]
1457            StorageList::RedisCluster(l) => l.expire_at(at).await,
1458            #[cfg(feature = "redb")]
1459            StorageList::Redb(l) => l.expire_at(at).await,
1460        }
1461    }
1462
1463    #[cfg(feature = "ttl")]
1464    async fn expire(&self, dur: TimestampMillis) -> Result<bool> {
1465        match self {
1466            #[cfg(feature = "sled")]
1467            StorageList::Sled(l) => l.expire(dur).await,
1468            #[cfg(feature = "redis")]
1469            StorageList::Redis(l) => l.expire(dur).await,
1470            #[cfg(feature = "redis-cluster")]
1471            StorageList::RedisCluster(l) => l.expire(dur).await,
1472            #[cfg(feature = "redb")]
1473            StorageList::Redb(l) => l.expire(dur).await,
1474        }
1475    }
1476
1477    #[cfg(feature = "ttl")]
1478    async fn ttl(&self) -> Result<Option<TimestampMillis>> {
1479        match self {
1480            #[cfg(feature = "sled")]
1481            StorageList::Sled(l) => l.ttl().await,
1482            #[cfg(feature = "redis")]
1483            StorageList::Redis(l) => l.ttl().await,
1484            #[cfg(feature = "redis-cluster")]
1485            StorageList::RedisCluster(l) => l.ttl().await,
1486            #[cfg(feature = "redb")]
1487            StorageList::Redb(l) => l.ttl().await,
1488        }
1489    }
1490}
1491
1492// ---------------------------------------------------------------------------
1493// Inherent raw methods on StorageList — NOT part of the List trait.
1494// ---------------------------------------------------------------------------
1495#[cfg(feature = "circuit-breaker")]
1496impl StorageList {
1497    pub(crate) async fn push_raw(&self, val: &[u8]) -> Result<()> {
1498        match self {
1499            #[cfg(feature = "sled")]
1500            StorageList::Sled(l) => l.push_raw(val).await,
1501            #[cfg(feature = "redis")]
1502            StorageList::Redis(l) => l.push_raw(val).await,
1503            #[cfg(feature = "redis-cluster")]
1504            StorageList::RedisCluster(l) => l.push_raw(val).await,
1505            #[cfg(feature = "redb")]
1506            StorageList::Redb(l) => l.push_raw(val).await,
1507        }
1508    }
1509
1510    pub(crate) async fn pushs_raw(&self, vals: Vec<Vec<u8>>) -> Result<()> {
1511        match self {
1512            #[cfg(feature = "sled")]
1513            StorageList::Sled(l) => l.pushs_raw(vals).await,
1514            #[cfg(feature = "redis")]
1515            StorageList::Redis(l) => l.pushs_raw(vals).await,
1516            #[cfg(feature = "redis-cluster")]
1517            StorageList::RedisCluster(l) => l.pushs_raw(vals).await,
1518            #[cfg(feature = "redb")]
1519            StorageList::Redb(l) => l.pushs_raw(vals).await,
1520        }
1521    }
1522
1523    pub(crate) async fn push_limit_raw(
1524        &self,
1525        val: &[u8],
1526        limit: usize,
1527        pop_front_if_limited: bool,
1528    ) -> Result<Option<Vec<u8>>> {
1529        match self {
1530            #[cfg(feature = "sled")]
1531            StorageList::Sled(l) => l.push_limit_raw(val, limit, pop_front_if_limited).await,
1532            #[cfg(feature = "redis")]
1533            StorageList::Redis(l) => l.push_limit_raw(val, limit, pop_front_if_limited).await,
1534            #[cfg(feature = "redis-cluster")]
1535            StorageList::RedisCluster(l) => {
1536                l.push_limit_raw(val, limit, pop_front_if_limited).await
1537            }
1538            #[cfg(feature = "redb")]
1539            StorageList::Redb(l) => l.push_limit_raw(val, limit, pop_front_if_limited).await,
1540        }
1541    }
1542
1543    pub(crate) async fn pop_raw(&self) -> Result<Option<Vec<u8>>> {
1544        match self {
1545            #[cfg(feature = "sled")]
1546            StorageList::Sled(l) => l.pop_raw().await,
1547            #[cfg(feature = "redis")]
1548            StorageList::Redis(l) => l.pop_raw().await,
1549            #[cfg(feature = "redis-cluster")]
1550            StorageList::RedisCluster(l) => l.pop_raw().await,
1551            #[cfg(feature = "redb")]
1552            StorageList::Redb(l) => l.pop_raw().await,
1553        }
1554    }
1555
1556    pub(crate) async fn all_raw(&self) -> Result<Vec<Vec<u8>>> {
1557        match self {
1558            #[cfg(feature = "sled")]
1559            StorageList::Sled(l) => l.all_raw().await,
1560            #[cfg(feature = "redis")]
1561            StorageList::Redis(l) => l.all_raw().await,
1562            #[cfg(feature = "redis-cluster")]
1563            StorageList::RedisCluster(l) => l.all_raw().await,
1564            #[cfg(feature = "redb")]
1565            StorageList::Redb(l) => l.all_raw().await,
1566        }
1567    }
1568
1569    pub(crate) async fn get_index_raw(&self, idx: usize) -> Result<Option<Vec<u8>>> {
1570        match self {
1571            #[cfg(feature = "sled")]
1572            StorageList::Sled(l) => l.get_index_raw(idx).await,
1573            #[cfg(feature = "redis")]
1574            StorageList::Redis(l) => l.get_index_raw(idx).await,
1575            #[cfg(feature = "redis-cluster")]
1576            StorageList::RedisCluster(l) => l.get_index_raw(idx).await,
1577            #[cfg(feature = "redb")]
1578            StorageList::Redb(l) => l.get_index_raw(idx).await,
1579        }
1580    }
1581}
1582
1583// ---------------------------------------------------------------------------
1584// DefaultStorageDB / StorageDB trait
1585// ---------------------------------------------------------------------------
rmqtt_storage/storage.rs

rmqtt_storage/
storage.rs
