feat: pluggable cache store adapter for custom backends

src/cache_config.rs

@@ -1,6 +1,9 @@
 use moka::future::Cache;
+use std::sync::Arc;
 use std::time::Duration;
 
+pub use wacore::store::cache::CacheStore;
+
 /// Configuration for a single cache instance.
 ///
 /// Controls the expiry timeout and maximum capacity of a moka cache.
@@ -49,12 +52,64 @@ impl CacheEntryConfig {
     }
 }
 
+/// Per-cache custom store overrides.
+///
+/// Each field is an optional [`CacheStore`] for that specific cache. When
+/// `None`, the default in-process moka cache is used.
+///
+/// # Example — only group and device on Redis
+///
+/// ```rust,ignore
+/// let redis = Arc::new(MyRedisCacheStore::new("redis://localhost:6379"));
+/// let config = CacheConfig {
+///     cache_stores: CacheStores {
+///         group_cache: Some(redis.clone()),
+///         device_cache: Some(redis.clone()),
+///         ..Default::default()
+///     },
+///     ..Default::default()
+/// };
+/// ```
+#[derive(Default, Clone)]
+pub struct CacheStores {
+    /// Custom store for group metadata cache.
+    pub group_cache: Option<Arc<dyn CacheStore>>,
+    /// Custom store for device list cache.
+    pub device_cache: Option<Arc<dyn CacheStore>>,
+    /// Custom store for device registry cache.
+    pub device_registry_cache: Option<Arc<dyn CacheStore>>,
+    /// Custom store for LID-PN bidirectional mapping cache.
+    pub lid_pn_cache: Option<Arc<dyn CacheStore>>,
+}
+
+impl CacheStores {
+    /// Set the same [`CacheStore`] for all pluggable caches at once.
+    ///
+    /// Coordination caches (`session_locks`, `message_queues`, etc.) and the
+    /// signal write-behind cache always remain in-process regardless of this
+    /// setting.
+    ///
+    /// # Example
+    ///
+    /// ```rust,ignore
+    /// let stores = CacheStores::all(Arc::new(MyRedisCacheStore::new("redis://localhost:6379")));
+    /// ```
+    pub fn all(store: Arc<dyn CacheStore>) -> Self {
+        Self {
+            group_cache: Some(store.clone()),
+            device_cache: Some(store.clone()),
+            device_registry_cache: Some(store.clone()),
+            lid_pn_cache: Some(store),
+        }
+    }
+}
+
 /// Configuration for all client caches and resource pools.
 ///
 /// All fields default to WhatsApp Web behavior. Use `..Default::default()` to
 /// override only specific settings.
 ///
-/// # Example
+/// # Example — tune TTL/capacity
 ///
 /// ```rust,ignore
 /// use whatsapp_rust::{CacheConfig, CacheEntryConfig};
@@ -65,7 +120,24 @@ impl CacheEntryConfig {
 ///     ..Default::default()
 /// };
 /// ```
-#[derive(Debug, Clone)]
+///
+/// # Example — Redis for group and device caches only
+///
+/// ```rust,ignore
+/// use std::sync::Arc;
+/// use whatsapp_rust::{CacheConfig, CacheStores};
+///
+/// let redis = Arc::new(MyRedisCacheStore::new("redis://localhost:6379"));
+/// let config = CacheConfig {
+///     cache_stores: CacheStores {
+///         group_cache: Some(redis.clone()),
+///         device_cache: Some(redis.clone()),
+///         ..Default::default()
+///     },
+///     ..Default::default()
+/// };
+/// ```
+#[derive(Clone)]
 pub struct CacheConfig {
     /// Group metadata cache (time_to_live). Default: 1h TTL, 250 entries.
     pub group_cache: CacheEntryConfig,
@@ -104,6 +176,63 @@ pub struct CacheConfig {
     /// TTL in seconds for sent messages in DB before periodic cleanup.
     /// 0 = no automatic cleanup. Default: 300 (5 minutes).
     pub sent_message_ttl_secs: u64,
+
+    // --- Custom store overrides ---
+    /// Per-cache custom store overrides.
+    ///
+    /// For each field set to `Some(store)`, the corresponding cache uses that
+    /// backend instead of the default in-process moka cache. Fields left as
+    /// `None` keep the default moka behaviour.
+    ///
+    /// Coordination caches (`session_locks`, `message_queues`,
+    /// `message_enqueue_locks`), the signal write-behind cache, and
+    /// `pdo_pending_requests` always stay in-process — they hold live Rust
+    /// objects (mutexes, channel senders, oneshot senders) that cannot be
+    /// serialised to an external store.
+    pub cache_stores: CacheStores,
+}
+
+impl std::fmt::Debug for CacheConfig {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("CacheConfig")
+            .field("group_cache", &self.group_cache)
+            .field("device_cache", &self.device_cache)
+            .field("device_registry_cache", &self.device_registry_cache)
+            .field("lid_pn_cache", &self.lid_pn_cache)
+            .field("retried_group_messages", &self.retried_group_messages)
+            .field("recent_messages", &self.recent_messages)
+            .field("message_retry_counts", &self.message_retry_counts)
+            .field("pdo_pending_requests", &self.pdo_pending_requests)
+            .field("session_locks_capacity", &self.session_locks_capacity)
+            .field("message_queues_capacity", &self.message_queues_capacity)
+            .field(
+                "message_enqueue_locks_capacity",
+                &self.message_enqueue_locks_capacity,
+            )
+            .field("max_pooled_buffers", &self.max_pooled_buffers)
+            .field(
+                "max_pooled_buffer_capacity",
+                &self.max_pooled_buffer_capacity,
+            )
+            .field("sent_message_ttl_secs", &self.sent_message_ttl_secs)
+            .field(
+                "cache_stores.group_cache",
+                &self.cache_stores.group_cache.is_some(),
+            )
+            .field(
+                "cache_stores.device_cache",
+                &self.cache_stores.device_cache.is_some(),
+            )
+            .field(
+                "cache_stores.device_registry_cache",
+                &self.cache_stores.device_registry_cache.is_some(),
+            )
+            .field(
+                "cache_stores.lid_pn_cache",
+                &self.cache_stores.lid_pn_cache.is_some(),
+            )
+            .finish()
+    }
 }
 
 impl Default for CacheConfig {
@@ -126,6 +255,7 @@ impl Default for CacheConfig {
             max_pooled_buffers: 8,
             max_pooled_buffer_capacity: 256 * 1024,
             sent_message_ttl_secs: 300,
+            cache_stores: CacheStores::default(),
         }
     }
 }

src/cache_store.rs

@@ -0,0 +1,201 @@
+//! Typed cache wrapper that dispatches to either moka (in-process) or a custom
+//! [`CacheStore`] backend (e.g., Redis).
+//!
+//! [`TypedCache`] presents the same interface regardless of the backing store.
+//! Keys are serialised via [`Display`]; values are serialised with `serde_json`
+//! only on the custom-store path — the moka path has zero extra overhead.
+
+use std::borrow::Borrow;
+use std::fmt::Display;
+use std::marker::PhantomData;
+use std::sync::Arc;
+use std::time::Duration;
+
+use moka::future::Cache;
+use serde::{Serialize, de::DeserializeOwned};
+
+pub use wacore::store::cache::CacheStore;
+
+// ── Internal storage variant ──────────────────────────────────────────────────
+
+enum Inner<K, V> {
+    Moka(Cache<K, V>),
+    Custom {
+        store: Arc<dyn CacheStore>,
+        namespace: &'static str,
+        ttl: Option<Duration>,
+        _marker: PhantomData<fn(K, V)>,
+    },
+}
+
+// ── TypedCache ─────────────────────────────────────────────────────────────────
+
+/// A cache over `K → V` backed by either moka or any [`CacheStore`].
+///
+/// The moka path has **zero extra overhead** — values are stored in-process
+/// without any serialisation.  The custom-store path serialises values with
+/// `serde_json` and keys via [`Display`].
+///
+/// Methods mirror moka's API so call sites need no changes.
+pub struct TypedCache<K, V> {
+    inner: Inner<K, V>,
+}
+
+impl<K, V> TypedCache<K, V>
+where
+    K: std::hash::Hash + Eq + Send + Sync + 'static,
+    V: Clone + Send + Sync + 'static,
+{
+    /// Wrap an existing moka cache (zero overhead vs. using moka directly).
+    pub fn from_moka(cache: Cache<K, V>) -> Self {
+        Self {
+            inner: Inner::Moka(cache),
+        }
+    }
+}
+
+impl<K, V> TypedCache<K, V>
+where
+    K: std::hash::Hash + Eq + Display + Send + Sync + 'static,
+    V: Clone + Serialize + DeserializeOwned + Send + Sync + 'static,
+{
+    /// Create a cache backed by a custom store.
+    ///
+    /// - `namespace` — unique string for this cache (e.g., `"group"`)
+    /// - `ttl` — forwarded to [`CacheStore::set`]; `None` means no expiry
+    pub fn from_store(
+        store: Arc<dyn CacheStore>,
+        namespace: &'static str,
+        ttl: Option<Duration>,
+    ) -> Self {
+        Self {
+            inner: Inner::Custom {
+                store,
+                namespace,
+                ttl,
+                _marker: PhantomData,
+            },
+        }
+    }
+
+    /// Look up a value.
+    ///
+    /// Accepts borrowed keys (`&str` for `String`, `&Jid` for `Jid`, etc.)
+    /// following the same pattern as [`std::collections::HashMap::get`].
+    ///
+    /// Cache misses and deserialisation failures both return `None`; the
+    /// caller re-fetches from the authoritative source.
+    pub async fn get<Q>(&self, key: &Q) -> Option<V>
+    where
+        K: Borrow<Q>,
+        Q: std::hash::Hash + Eq + Display + ?Sized,
+    {
+        match &self.inner {
+            Inner::Moka(cache) => cache.get(key).await,
+            Inner::Custom {
+                store, namespace, ..
+            } => {
+                let key_str = key.to_string();
+                match store.get(namespace, &key_str).await {
+                    Ok(Some(bytes)) => serde_json::from_slice(&bytes)
+                        .inspect_err(|e| {
+                            log::warn!(
+                                "TypedCache[{namespace}]: deserialise failed for {key_str}: {e}"
+                            );
+                        })
+                        .ok(),
+                    Ok(None) => None,
+                    Err(e) => {
+                        log::warn!("TypedCache[{namespace}]: get({key_str}) error: {e}");
+                        None
+                    }
+                }
+            }
+        }
+    }
+
+    /// Insert or update a value (takes ownership of key and value).
+    pub async fn insert(&self, key: K, value: V) {
+        match &self.inner {
+            Inner::Moka(cache) => cache.insert(key, value).await,
+            Inner::Custom {
+                store,
+                namespace,
+                ttl,
+                ..
+            } => {
+                let key_str = key.to_string();
+                match serde_json::to_vec(&value) {
+                    Ok(bytes) => {
+                        if let Err(e) = store.set(namespace, &key_str, &bytes, *ttl).await {
+                            log::warn!("TypedCache[{namespace}]: set({key_str}) error: {e}");
+                        }
+                    }
+                    Err(e) => {
+                        log::warn!("TypedCache[{namespace}]: serialise failed for {key_str}: {e}");
+                    }
+                }
+            }
+        }
+    }
+
+    /// Remove a single key.
+    ///
+    /// Accepts borrowed keys following the same pattern as `get`.
+    pub async fn invalidate<Q>(&self, key: &Q)
+    where
+        K: Borrow<Q>,
+        Q: std::hash::Hash + Eq + Display + ?Sized,
+    {
+        match &self.inner {
+            Inner::Moka(cache) => cache.invalidate(key).await,
+            Inner::Custom {
+                store, namespace, ..
+            } => {
+                let key_str = key.to_string();
+                if let Err(e) = store.delete(namespace, &key_str).await {
+                    log::warn!("TypedCache[{namespace}]: delete({key_str}) error: {e}");
+                }
+            }
+        }
+    }
+
+    /// Remove all entries.  For the custom backend this spawns a fire-and-forget
+    /// task (mirrors moka's non-`async` `invalidate_all`).
+    pub fn invalidate_all(&self) {
+        match &self.inner {
+            Inner::Moka(cache) => cache.invalidate_all(),
+            Inner::Custom {
+                store, namespace, ..
+            } => {
+                let store = store.clone();
+                let ns = *namespace;
+                tokio::spawn(async move {
+                    if let Err(e) = store.clear(ns).await {
+                        log::warn!("TypedCache[{ns}]: clear() error: {e}");
+                    }
+                });
+            }
+        }
+    }
+
+    /// Run any pending internal housekeeping tasks (moka only).
+    ///
+    /// For the moka backend this ensures all writes have been applied before
+    /// calling [`entry_count`](Self::entry_count), which can otherwise lag.
+    /// For custom backends this is a no-op.
+    pub async fn run_pending_tasks(&self) {
+        if let Inner::Moka(cache) = &self.inner {
+            cache.run_pending_tasks().await;
+        }
+    }
+
+    /// Approximate number of cached entries (for diagnostics; always `0` for
+    /// custom backends that don't support cheap counts).
+    pub fn entry_count(&self) -> u64 {
+        match &self.inner {
+            Inner::Moka(cache) => cache.entry_count(),
+            Inner::Custom { .. } => 0,
+        }
+    }
+}