multi_tier_cache/
invalidation.rs

1//! Cache invalidation and synchronization module
2//!
3//! This module provides cross-instance cache invalidation using Redis Pub/Sub.
4//! It supports both cache removal (invalidation) and cache updates (refresh).
5
6use crate::error::CacheResult;
7use crate::traits::StreamingBackend;
8use bytes::Bytes;
9use futures_util::StreamExt;
10use redis::AsyncCommands;
11use serde::{Deserialize, Serialize};
12use std::time::{Duration, SystemTime, UNIX_EPOCH};
13use tokio::sync::broadcast;
14use tracing::{error, info, warn};
15use uuid::Uuid;
16
17/// Invalidation message types sent across cache instances via Redis Pub/Sub
18#[derive(Debug, Clone, Serialize, Deserialize)]
19#[serde(tag = "type")]
20pub enum InvalidationMessage {
21    /// Remove a single key from all cache instances
22    Remove { key: String },
23
24    /// Update a key with new value across all cache instances
25    /// This is more efficient than Remove for hot keys as it avoids cache miss
26    Update {
27        key: String,
28        #[serde(with = "serde_bytes_wrapper")]
29        value: Bytes,
30        #[serde(skip_serializing_if = "Option::is_none")]
31        ttl_secs: Option<u64>,
32    },
33
34    /// Remove all keys matching a pattern from all cache instances
35    /// Uses glob-style patterns (e.g., "user:*", "product:123:*")
36    RemovePattern { pattern: String },
37
38    /// Bulk remove multiple keys at once
39    RemoveBulk { keys: Vec<String> },
40}
41
42impl InvalidationMessage {
43    /// Create a Remove message
44    pub fn remove(key: impl Into<String>) -> Self {
45        Self::Remove { key: key.into() }
46    }
47
48    /// Create an Update message
49    pub fn update(key: impl Into<String>, value: Bytes, ttl: Option<Duration>) -> Self {
50        Self::Update {
51            key: key.into(),
52            value,
53            ttl_secs: ttl.map(|d| d.as_secs()),
54        }
55    }
56
57    /// Create a `RemovePattern` message
58    pub fn remove_pattern(pattern: impl Into<String>) -> Self {
59        Self::RemovePattern {
60            pattern: pattern.into(),
61        }
62    }
63
64    /// Create a `RemoveBulk` message
65    #[must_use]
66    pub fn remove_bulk(keys: Vec<String>) -> Self {
67        Self::RemoveBulk { keys }
68    }
69
70    /// Serialize to JSON for transmission
71    ///
72    /// # Errors
73    ///
74    /// Returns an error if serialization fails.
75    pub fn to_json(&self) -> CacheResult<String> {
76        serde_json::to_string(self).map_err(|e| {
77            crate::error::CacheError::SerializationError(format!(
78                "Failed to serialize invalidation message: {e}"
79            ))
80        })
81    }
82
83    /// Deserialize from JSON
84    ///
85    /// # Errors
86    ///
87    /// Returns an error if deserialization fails.
88    pub fn from_json(json: &str) -> CacheResult<Self> {
89        serde_json::from_str(json).map_err(|e| {
90            crate::error::CacheError::SerializationError(format!(
91                "Failed to deserialize invalidation message: {e}"
92            ))
93        })
94    }
95
96    /// Get TTL as Duration if present
97    pub fn ttl(&self) -> Option<Duration> {
98        match self {
99            Self::Update { ttl_secs, .. } => ttl_secs.map(Duration::from_secs),
100            _ => None,
101        }
102    }
103}
104
105/// Helper module for Bytes serialization in JSON
106mod serde_bytes_wrapper {
107    use bytes::Bytes;
108    use serde::{Deserialize, Deserializer, Serializer};
109
110    fn parse_hex_digit(b: u8) -> Option<u8> {
111        match b {
112            b'0'..=b'9' => Some(b - b'0'),
113            b'a'..=b'f' => Some(b - b'a' + 10),
114            b'A'..=b'F' => Some(b - b'A' + 10),
115            _ => None,
116        }
117    }
118
119    pub fn serialize<S>(bytes: &Bytes, serializer: S) -> Result<S::Ok, S::Error>
120    where
121        S: Serializer,
122    {
123        let mut hex_str = String::with_capacity(bytes.len() * 2);
124        for &b in bytes.as_ref() {
125            let high = match b >> 4 {
126                val @ 0..=9 => (b'0' + val) as char,
127                val @ 10..=15 => (b'a' + val - 10) as char,
128                _ => '0',
129            };
130            let low = match b & 0xf {
131                val @ 0..=9 => (b'0' + val) as char,
132                val @ 10..=15 => (b'a' + val - 10) as char,
133                _ => '0',
134            };
135            hex_str.push(high);
136            hex_str.push(low);
137        }
138        serializer.serialize_str(&hex_str)
139    }
140
141    pub fn deserialize<'de, D>(deserializer: D) -> Result<Bytes, D::Error>
142    where
143        D: Deserializer<'de>,
144    {
145        let hex_str = String::deserialize(deserializer)?;
146        let bytes_str = hex_str.as_bytes();
147        if bytes_str.len() % 2 != 0 {
148            return Err(serde::de::Error::custom("Odd-length hex string"));
149        }
150        let mut v = Vec::with_capacity(bytes_str.len() / 2);
151        for chunk in bytes_str.chunks_exact(2) {
152            if let [h, l] = chunk {
153                let high = parse_hex_digit(*h)
154                    .ok_or_else(|| serde::de::Error::custom("Invalid hex character"))?;
155                let low = parse_hex_digit(*l)
156                    .ok_or_else(|| serde::de::Error::custom("Invalid hex character"))?;
157                v.push((high << 4) | low);
158            }
159        }
160        Ok(Bytes::from(v))
161    }
162}
163
164/// Configuration for cache invalidation
165#[derive(Debug, Clone)]
166pub struct InvalidationConfig {
167    /// Redis Pub/Sub channel name for invalidation messages
168    pub channel: String,
169
170    /// Whether to automatically broadcast invalidation on writes
171    pub auto_broadcast_on_write: bool,
172
173    /// Whether to also publish invalidation events to Redis Streams for audit
174    pub enable_audit_stream: bool,
175
176    /// Redis Stream name for invalidation audit trail
177    pub audit_stream: String,
178
179    /// Maximum length of audit stream (older entries are trimmed)
180    pub audit_stream_maxlen: Option<usize>,
181}
182
183impl Default for InvalidationConfig {
184    fn default() -> Self {
185        Self {
186            channel: "cache:invalidate".to_string(),
187            auto_broadcast_on_write: false, // Conservative default
188            enable_audit_stream: false,
189            audit_stream: "cache:invalidations".to_string(),
190            audit_stream_maxlen: Some(10000),
191        }
192    }
193}
194
195/// Handle for sending invalidation messages
196pub struct InvalidationPublisher {
197    connection: redis::aio::ConnectionManager,
198    config: InvalidationConfig,
199}
200
201impl InvalidationPublisher {
202    /// Create a new publisher
203    #[must_use]
204    pub fn new(connection: redis::aio::ConnectionManager, config: InvalidationConfig) -> Self {
205        Self { connection, config }
206    }
207
208    /// Publish an invalidation message to all subscribers
209    ///
210    /// # Errors
211    ///
212    /// Returns an error if serialization or publishing fails.
213    pub async fn publish(&mut self, message: &InvalidationMessage) -> CacheResult<()> {
214        let json = message.to_json()?;
215
216        // Publish to Pub/Sub channel
217        let _: () = self
218            .connection
219            .publish(&self.config.channel, &json)
220            .await
221            .map_err(|e| {
222                crate::error::CacheError::InvalidationError(format!(
223                    "Failed to publish invalidation message: {e}"
224                ))
225            })?;
226
227        // Optionally publish to audit stream
228        if self.config.enable_audit_stream
229            && let Err(e) = self.publish_to_audit_stream(message).await
230        {
231            // Don't fail the invalidation if audit logging fails
232            warn!("Failed to publish to audit stream: {}", e);
233        }
234
235        Ok(())
236    }
237
238    /// Publish to audit stream for observability
239    async fn publish_to_audit_stream(&mut self, message: &InvalidationMessage) -> CacheResult<()> {
240        let timestamp = SystemTime::now()
241            .duration_since(UNIX_EPOCH)
242            .unwrap_or(Duration::ZERO)
243            .as_secs()
244            .to_string();
245
246        // Use &str to avoid unnecessary allocations
247        let (type_str, key_str): (&str, &str);
248        let extra_str: String;
249
250        match message {
251            InvalidationMessage::Remove { key } => {
252                type_str = "remove";
253                key_str = key.as_str();
254                extra_str = String::new();
255            }
256            InvalidationMessage::Update { key, .. } => {
257                type_str = "update";
258                key_str = key.as_str();
259                extra_str = String::new();
260            }
261            InvalidationMessage::RemovePattern { pattern } => {
262                type_str = "remove_pattern";
263                key_str = pattern.as_str();
264                extra_str = String::new();
265            }
266            InvalidationMessage::RemoveBulk { keys } => {
267                type_str = "remove_bulk";
268                key_str = "";
269                extra_str = keys.len().to_string();
270            }
271        }
272
273        let mut fields = vec![("type", type_str), ("timestamp", timestamp.as_str())];
274
275        if !key_str.is_empty() {
276            fields.push(("key", key_str));
277        }
278        if !extra_str.is_empty() {
279            fields.push(("count", extra_str.as_str()));
280        }
281
282        let mut cmd = redis::cmd("XADD");
283        cmd.arg(&self.config.audit_stream);
284
285        if let Some(maxlen) = self.config.audit_stream_maxlen {
286            cmd.arg("MAXLEN").arg("~").arg(maxlen);
287        }
288
289        cmd.arg("*"); // Auto-generate ID
290
291        for (key, value) in fields {
292            cmd.arg(key).arg(value);
293        }
294
295        let _: String = cmd.query_async(&mut self.connection).await.map_err(|e| {
296            crate::error::CacheError::BackendError(format!("Failed to add to audit stream: {e}"))
297        })?;
298
299        Ok(())
300    }
301}
302
303/// Statistics for invalidation operations
304#[derive(Debug, Default, Clone)]
305pub struct InvalidationStats {
306    /// Number of invalidation messages published
307    pub messages_sent: u64,
308
309    /// Number of invalidation messages received
310    pub messages_received: u64,
311
312    /// Number of Remove operations performed
313    pub removes_received: u64,
314
315    /// Number of Update operations performed
316    pub updates_received: u64,
317
318    /// Number of `RemovePattern` operations performed
319    pub patterns_received: u64,
320
321    /// Number of `RemoveBulk` operations performed
322    pub bulk_removes_received: u64,
323
324    /// Number of failed message processing attempts
325    pub processing_errors: u64,
326}
327
328use std::sync::atomic::{AtomicU64, Ordering};
329
330/// Thread-safe statistics for invalidation operations
331#[derive(Debug, Default)]
332pub struct AtomicInvalidationStats {
333    pub messages_sent: AtomicU64,
334    pub messages_received: AtomicU64,
335    pub removes_received: AtomicU64,
336    pub updates_received: AtomicU64,
337    pub patterns_received: AtomicU64,
338    pub bulk_removes_received: AtomicU64,
339    pub processing_errors: AtomicU64,
340}
341
342impl AtomicInvalidationStats {
343    pub fn snapshot(&self) -> InvalidationStats {
344        InvalidationStats {
345            messages_sent: self.messages_sent.load(Ordering::Relaxed),
346            messages_received: self.messages_received.load(Ordering::Relaxed),
347            removes_received: self.removes_received.load(Ordering::Relaxed),
348            updates_received: self.updates_received.load(Ordering::Relaxed),
349            patterns_received: self.patterns_received.load(Ordering::Relaxed),
350            bulk_removes_received: self.bulk_removes_received.load(Ordering::Relaxed),
351            processing_errors: self.processing_errors.load(Ordering::Relaxed),
352        }
353    }
354}
355
356use std::sync::Arc;
357
358/// Handle for subscribing to invalidation messages
359///
360/// This spawns a background task that listens to Redis Pub/Sub and processes
361/// invalidation messages by calling the provided handler callback.
362pub struct InvalidationSubscriber {
363    /// Redis client for creating Pub/Sub connections
364    client: redis::Client,
365    /// Configuration
366    config: InvalidationConfig,
367    /// Statistics
368    stats: Arc<AtomicInvalidationStats>,
369    /// Shutdown signal sender
370    shutdown_tx: broadcast::Sender<()>,
371}
372
373impl InvalidationSubscriber {
374    /// Create a new subscriber
375    ///
376    /// # Arguments
377    /// * `redis_url` - Redis connection URL
378    /// * `config` - Invalidation configuration
379    /// # Errors
380    ///
381    /// Returns an error if Redis client creation fails.
382    pub fn new(redis_url: &str, config: InvalidationConfig) -> CacheResult<Self> {
383        let client = redis::Client::open(redis_url).map_err(|e| {
384            crate::error::CacheError::ConfigError(format!(
385                "Failed to create Redis client for subscriber: {e}"
386            ))
387        })?;
388
389        let (shutdown_tx, _) = broadcast::channel(1);
390
391        Ok(Self {
392            client,
393            config,
394            stats: Arc::new(AtomicInvalidationStats::default()),
395            shutdown_tx,
396        })
397    }
398
399    /// Get a snapshot of current statistics
400    #[must_use]
401    pub fn stats(&self) -> InvalidationStats {
402        self.stats.snapshot()
403    }
404
405    /// Start the subscriber background task
406    ///
407    /// # Arguments
408    /// * `handler` - Async function to handle each invalidation message
409    ///
410    /// # Returns
411    /// Join handle for the background task
412    pub fn start<F, Fut>(&self, handler: F) -> tokio::task::JoinHandle<()>
413    where
414        F: Fn(InvalidationMessage) -> Fut + Send + Sync + 'static,
415        Fut: std::future::Future<Output = CacheResult<()>> + Send + 'static,
416    {
417        let client = self.client.clone();
418        let channel = self.config.channel.clone();
419        let stats = Arc::clone(&self.stats);
420        let mut shutdown_rx = self.shutdown_tx.subscribe();
421
422        tokio::spawn(async move {
423            let handler = Arc::new(handler);
424
425            loop {
426                // Check for shutdown signal
427                if shutdown_rx.try_recv().is_ok() {
428                    info!("Invalidation subscriber shutting down...");
429                    break;
430                }
431
432                // Attempt to connect and subscribe
433                match Self::run_subscriber_loop(
434                    &client,
435                    &channel,
436                    Arc::clone(&handler),
437                    Arc::clone(&stats),
438                    &mut shutdown_rx,
439                )
440                .await
441                {
442                    Ok(()) => {
443                        info!("Invalidation subscriber loop completed normally");
444                        break;
445                    }
446                    Err(e) => {
447                        error!(
448                            "Invalidation subscriber error: {}. Reconnecting in 5s...",
449                            e
450                        );
451                        stats.processing_errors.fetch_add(1, Ordering::Relaxed);
452
453                        // Wait before reconnecting
454                        tokio::select! {
455                            () = tokio::time::sleep(Duration::from_secs(5)) => {},
456                            _ = shutdown_rx.recv() => {
457                                info!("Invalidation subscriber shutting down...");
458                                break;
459                            }
460                        }
461                    }
462                }
463            }
464        })
465    }
466
467    /// Internal subscriber loop
468    async fn run_subscriber_loop<F, Fut>(
469        client: &redis::Client,
470        channel: &str,
471        handler: Arc<F>,
472        stats: Arc<AtomicInvalidationStats>,
473        shutdown_rx: &mut broadcast::Receiver<()>,
474    ) -> CacheResult<()>
475    where
476        F: Fn(InvalidationMessage) -> Fut + Send + Sync + 'static,
477        Fut: std::future::Future<Output = CacheResult<()>> + Send + 'static,
478    {
479        let mut pubsub = client.get_async_pubsub().await.map_err(|e| {
480            crate::error::CacheError::BackendError(format!("Failed to get pubsub connection: {e}"))
481        })?;
482
483        // Subscribe to channel
484        pubsub.subscribe(channel).await.map_err(|e| {
485            crate::error::CacheError::InvalidationError(format!(
486                "Failed to subscribe to channel: {e}"
487            ))
488        })?;
489
490        info!("Subscribed to invalidation channel: {}", channel);
491
492        // Get message stream
493        let mut stream = pubsub.on_message();
494
495        loop {
496            // Wait for message or shutdown signal
497            tokio::select! {
498                msg_result = stream.next() => {
499                    match msg_result {
500                        Some(msg) => {
501                            // Get payload
502                            let payload: String = match msg.get_payload() {
503                                Ok(p) => p,
504                                Err(e) => {
505                                    warn!("Failed to get message payload: {}", e);
506                                    stats.processing_errors.fetch_add(1, Ordering::Relaxed);
507                                    continue;
508                                }
509                            };
510
511                            // Deserialize message
512                            let invalidation_msg = match InvalidationMessage::from_json(&payload) {
513                                Ok(m) => m,
514                                Err(e) => {
515                                    warn!("Failed to deserialize invalidation message: {}", e);
516                                    stats.processing_errors.fetch_add(1, Ordering::Relaxed);
517                                    continue;
518                                }
519                            };
520
521                            // Update stats
522                            stats.messages_received.fetch_add(1, Ordering::Relaxed);
523                            match &invalidation_msg {
524                                InvalidationMessage::Remove { .. } => {
525                                    stats.removes_received.fetch_add(1, Ordering::Relaxed);
526                                }
527                                InvalidationMessage::Update { .. } => {
528                                    stats.updates_received.fetch_add(1, Ordering::Relaxed);
529                                }
530                                InvalidationMessage::RemovePattern { .. } => {
531                                    stats.patterns_received.fetch_add(1, Ordering::Relaxed);
532                                }
533                                InvalidationMessage::RemoveBulk { .. } => {
534                                    stats.bulk_removes_received.fetch_add(1, Ordering::Relaxed);
535                                }
536                            }
537
538                            // Call handler
539                            if let Err(e) = handler(invalidation_msg).await {
540                                error!("Invalidation handler error: {}", e);
541                                stats.processing_errors.fetch_add(1, Ordering::Relaxed);
542                            }
543                        }
544                        None => {
545                            // Stream ended
546                            return Err(crate::error::CacheError::InvalidationError("Pub/Sub message stream ended".to_string()));
547                        }
548                    }
549                }
550                _ = shutdown_rx.recv() => {
551                    return Ok(());
552                }
553            }
554        }
555    }
556
557    /// Signal the subscriber to shutdown
558    pub fn shutdown(&self) {
559        let _ = self.shutdown_tx.send(());
560    }
561}
562
563/// Reliable subscriber using Redis Streams and Consumer Groups
564pub struct ReliableStreamSubscriber {
565    redis_url: String,
566    config: InvalidationConfig,
567    stats: Arc<AtomicInvalidationStats>,
568    shutdown_tx: broadcast::Sender<()>,
569    group_name: String,
570    consumer_name: String,
571}
572
573impl ReliableStreamSubscriber {
574    /// Create a new `ReliableStreamSubscriber`
575    ///
576    /// # Errors
577    ///
578    /// Returns an error if the Redis client fails to open.
579    pub fn new(redis_url: &str, config: InvalidationConfig, group_name: &str) -> CacheResult<Self> {
580        let _client = redis::Client::open(redis_url).map_err(|e| {
581            crate::error::CacheError::ConfigError(format!(
582                "Failed to create Redis client for reliable subscriber: {e}"
583            ))
584        })?;
585
586        let (shutdown_tx, _) = broadcast::channel(1);
587        let consumer_name = format!("consumer-{}", Uuid::new_v4());
588
589        Ok(Self {
590            redis_url: redis_url.to_string(),
591            config,
592            stats: Arc::new(AtomicInvalidationStats::default()),
593            shutdown_tx,
594            group_name: group_name.to_string(),
595            consumer_name,
596        })
597    }
598
599    pub fn start<F, Fut>(&self, handler: F) -> tokio::task::JoinHandle<()>
600    where
601        F: Fn(InvalidationMessage) -> Fut + Send + Sync + 'static,
602        Fut: std::future::Future<Output = CacheResult<()>> + Send + 'static,
603    {
604        let stream_key = self.config.channel.clone();
605        let group_name = self.group_name.clone();
606        let consumer_name = self.consumer_name.clone();
607        let handler = Arc::new(handler);
608        let stats = self.stats.clone();
609        let mut shutdown_rx = self.shutdown_tx.subscribe();
610        let redis_url = self.redis_url.clone();
611
612        tokio::spawn(async move {
613            info!(
614                stream = %stream_key,
615                group = %group_name,
616                consumer = %consumer_name,
617                "Starting reliable stream subscriber"
618            );
619
620            // 1. Ensure stream and group exist
621            let redis_backend = crate::redis_streams::RedisStreams::new(&redis_url).await;
622            match redis_backend {
623                Ok(backend) => {
624                    let _ = backend
625                        .stream_create_group(&stream_key, &group_name, "0")
626                        .await;
627
628                    loop {
629                        // Check shutdown before starting loop
630                        if shutdown_rx.try_recv().is_ok() {
631                            break;
632                        }
633
634                        if let Err(e) = Self::run_reliable_loop(
635                            &backend,
636                            &stream_key,
637                            &group_name,
638                            &consumer_name,
639                            handler.clone(),
640                            stats.clone(),
641                            &mut shutdown_rx,
642                        )
643                        .await
644                        {
645                            error!("Reliable subscriber loop error: {}", e);
646
647                            tokio::select! {
648                                () = tokio::time::sleep(Duration::from_secs(5)) => {},
649                                _ = shutdown_rx.recv() => break,
650                            }
651                        } else {
652                            break; // Normal shutdown
653                        }
654                    }
655                }
656                Err(e) => {
657                    error!(
658                        "Failed to initialize Redis Streams backend for reliable subscriber: {}",
659                        e
660                    );
661                }
662            }
663        })
664    }
665
666    async fn run_reliable_loop<F, Fut>(
667        backend: &dyn crate::traits::StreamingBackend,
668        stream_key: &str,
669        group_name: &str,
670        consumer_name: &str,
671        handler: Arc<F>,
672        stats: Arc<AtomicInvalidationStats>,
673        shutdown_rx: &mut broadcast::Receiver<()>,
674    ) -> CacheResult<()>
675    where
676        F: Fn(InvalidationMessage) -> Fut + Send + Sync + 'static,
677        Fut: std::future::Future<Output = CacheResult<()>> + Send + 'static,
678    {
679        loop {
680            tokio::select! {
681                entries_result = backend.stream_read_group(stream_key, group_name, consumer_name, 10, Some(5000)) => {
682                    let entries = entries_result?;
683                    if entries.is_empty() { continue; }
684
685                    let mut processed_ids = Vec::new();
686                    for (id, fields) in entries {
687                        // Find "payload" field or use first field if it looks like JSON
688                        let payload = fields.iter().find(|(k, _)| k == "payload")
689                            .map(|(_, v)| v.as_str())
690                            .or_else(|| fields.first().map(|(_, v)| v.as_str()));
691
692                        if let Some(msg) = payload.and_then(|json| InvalidationMessage::from_json(json).ok()) {
693                            stats.messages_received.fetch_add(1, Ordering::Relaxed);
694                            if let Err(e) = handler(msg).await {
695                                error!("Reliable handler error: {}", e);
696                                stats.processing_errors.fetch_add(1, Ordering::Relaxed);
697                            } else {
698                                processed_ids.push(id);
699                            }
700                        }
701                    }
702
703                    if !processed_ids.is_empty() {
704                        backend.stream_ack(stream_key, group_name, &processed_ids).await?;
705                    }
706                }
707                _ = shutdown_rx.recv() => return Ok(()),
708            }
709        }
710    }
711
712    /// Signal the subscriber to shutdown
713    pub fn shutdown(&self) {
714        let _ = self.shutdown_tx.send(()).unwrap_or(0);
715    }
716}
717
718#[cfg(test)]
719mod tests {
720    use super::*;
721
722    #[test]
723    fn test_invalidation_message_serialization() -> CacheResult<()> {
724        // Test Remove
725        let msg = InvalidationMessage::remove("test_key");
726        let json = msg.to_json()?;
727        let parsed = InvalidationMessage::from_json(&json)?;
728        match parsed {
729            InvalidationMessage::Remove { key } => assert_eq!(key, "test_key"),
730            _ => panic!("Wrong message type"),
731        }
732
733        // Test Update
734        let msg = InvalidationMessage::update(
735            "test_key",
736            Bytes::from("{\"value\": 123}"),
737            Some(Duration::from_secs(3600)),
738        );
739
740        if let InvalidationMessage::Update {
741            key,
742            value,
743            ttl_secs,
744        } = msg
745        {
746            assert_eq!(key, "test_key");
747            assert_eq!(value, Bytes::from("{\"value\": 123}"));
748            assert_eq!(ttl_secs, Some(3600));
749        } else {
750            panic!("Expected Update message");
751        }
752
753        // Test RemovePattern
754        let msg = InvalidationMessage::remove_pattern("user:*");
755        let json = msg.to_json()?;
756        let parsed = InvalidationMessage::from_json(&json)?;
757        match parsed {
758            InvalidationMessage::RemovePattern { pattern } => assert_eq!(pattern, "user:*"),
759            _ => panic!("Wrong message type"),
760        }
761
762        // Test RemoveBulk
763        let msg = InvalidationMessage::remove_bulk(vec!["key1".to_string(), "key2".to_string()]);
764        let json = msg.to_json()?;
765        let parsed = InvalidationMessage::from_json(&json)?;
766        match parsed {
767            InvalidationMessage::RemoveBulk { keys } => assert_eq!(keys, vec!["key1", "key2"]),
768            _ => panic!("Wrong message type"),
769        }
770        Ok(())
771    }
772
773    #[test]
774    fn test_invalidation_config_default() {
775        let config = InvalidationConfig::default();
776        assert_eq!(config.channel, "cache:invalidate");
777        assert!(!config.auto_broadcast_on_write);
778        assert!(!config.enable_audit_stream);
779    }
780}
multi_tier_cache/invalidation.rs

multi_tier_cache/
invalidation.rs
