pywatt_sdk/data/cache/
mod.rs

1use async_trait::async_trait;
2use serde::{Deserialize, Serialize};
3use std::fmt::Debug;
4use std::time::Duration;
5use thiserror::Error;
6use uuid;
7
8// Modules
9pub mod file;
10pub mod in_memory;
11pub mod memcached;
12pub mod patterns;
13pub mod proxy_service;
14pub mod redis;
15pub mod tests;
16
17// Re-exports
18pub use file::FileCache;
19pub use in_memory::InMemoryCache;
20#[cfg(feature = "memcached")]
21pub use memcached::MemcachedCache;
22pub use redis::RedisCache;
23
24/// Cache policy enum for different caching strategies
25#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
26pub enum CachePolicy {
27    /// Least Recently Used policy
28    LRU,
29    /// First In First Out policy
30    FIFO,
31    /// Most Recently Used policy
32    MRU,
33    /// Least Frequently Used policy
34    LFU,
35    /// No eviction policy
36    None,
37}
38
39impl Default for CachePolicy {
40    fn default() -> Self {
41        Self::LRU
42    }
43}
44
45/// Cache type enum for different cache service implementations
46#[derive(Default, Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
47pub enum CacheType {
48    /// In-memory cache (preferred variant)
49    #[default]
50    InMemory,
51    /// Redis cache
52    Redis,
53    /// Memcached cache
54    Memcached,
55    /// File-based cache
56    File,
57}
58
59/// Cache configuration – aligned with test expectations.
60#[derive(Debug, Clone, Serialize, Deserialize)]
61pub struct CacheConfig {
62    /// Cache implementation type
63    pub cache_type: CacheType,
64    /// Cache eviction policy
65    pub policy: CachePolicy,
66    /// Maximum size in bytes (optional)
67    pub max_size_bytes: Option<usize>,
68
69    // ------------------------------------------------------------------
70    // TTL handling
71    // ------------------------------------------------------------------
72    /// Default TTL for cache entries, expressed as seconds – **primary field used by tests**
73    pub default_ttl_seconds: u64,
74    /// Optional `Duration` based representation kept for legacy code-paths.
75    pub default_ttl: Option<Duration>,
76
77    // ------------------------------------------------------------------
78    // Connection parameters
79    // ------------------------------------------------------------------
80    pub hosts: Vec<String>,
81    pub port: Option<u16>,
82    /// Network dial timeout
83    pub connection_timeout_seconds: u64,
84    /// Per-operation timeout
85    pub operation_timeout_seconds: u64,
86
87    // ------------------------------------------------------------------
88    // Authentication & security
89    // ------------------------------------------------------------------
90    pub username: Option<String>,
91    pub password: Option<String>,
92    /// Enable TLS/SSL if supported by backend
93    pub tls_enabled: bool,
94
95    // ------------------------------------------------------------------
96    // Backend-specific options
97    // ------------------------------------------------------------------
98    pub database: Option<u8>,
99    pub file_path: Option<String>,
100    /// Optional key namespace/prefix
101    pub namespace: Option<String>,
102
103    // ------------------------------------------------------------------
104    // Pool configuration (re-used from the database module)
105    // ------------------------------------------------------------------
106    #[cfg(feature = "database")]
107    pub pool: crate::data::database::PoolConfig,
108    #[cfg(not(feature = "database"))]
109    pub pool: DefaultPoolConfig,
110
111    /// Additional opaque parameters
112    pub extra_params: std::collections::HashMap<String, String>,
113}
114
115/// Default pool configuration when database feature is not enabled
116#[cfg(not(feature = "database"))]
117#[derive(Debug, Clone, Serialize, Deserialize)]
118pub struct DefaultPoolConfig {
119    pub max_connections: u32,
120    pub min_connections: u32,
121    pub idle_timeout_seconds: u64,
122    pub max_lifetime_seconds: u64,
123    pub acquire_timeout_seconds: u64,
124}
125
126#[cfg(not(feature = "database"))]
127impl Default for DefaultPoolConfig {
128    fn default() -> Self {
129        Self {
130            max_connections: 10,
131            min_connections: 1,
132            idle_timeout_seconds: 300,
133            max_lifetime_seconds: 1800,
134            acquire_timeout_seconds: 30,
135        }
136    }
137}
138
139impl CacheConfig {
140    /// Helper accessor to obtain an actual `Duration` for the default TTL.
141    pub fn get_default_ttl(&self) -> Duration {
142        self.default_ttl
143            .unwrap_or_else(|| Duration::from_secs(self.default_ttl_seconds))
144    }
145}
146
147impl Default for CacheConfig {
148    fn default() -> Self {
149        Self {
150            cache_type: CacheType::InMemory,
151            policy: CachePolicy::default(),
152            max_size_bytes: Some(1024 * 1024 * 10), // 10 MB
153            default_ttl_seconds: 300,
154            default_ttl: None,
155            hosts: vec!["localhost".to_string()],
156            port: None,
157            connection_timeout_seconds: 5,
158            operation_timeout_seconds: 2,
159            username: None,
160            password: None,
161            tls_enabled: false,
162            database: None,
163            file_path: None,
164            namespace: None,
165            #[cfg(feature = "database")]
166            pool: crate::data::database::PoolConfig::default(),
167            #[cfg(not(feature = "database"))]
168            pool: DefaultPoolConfig::default(),
169            extra_params: std::collections::HashMap::new(),
170        }
171    }
172}
173
174/// Cache error type
175#[derive(Debug, Error)]
176pub enum CacheError {
177    /// Connection error
178    #[error("connection error: {0}")]
179    Connection(String),
180
181    /// Set operation error
182    #[error("set error: {0}")]
183    Set(String),
184
185    /// Get operation error
186    #[error("get error: {0}")]
187    Get(String),
188
189    /// Delete operation error
190    #[error("delete error: {0}")]
191    Delete(String),
192
193    /// Flush operation error
194    #[error("flush error: {0}")]
195    Flush(String),
196
197    /// Cache configuration error
198    #[error("configuration error: {0}")]
199    Configuration(String),
200
201    /// Cache serialization/deserialization error
202    #[error("serialization error: {0}")]
203    Serialization(String),
204
205    /// Cache IPC error
206    #[error("IPC error: {0}")]
207    Ipc(String),
208
209    /// Operation error
210    #[error("operation error: {0}")]
211    Operation(String),
212
213    /// Internal SDK implementation error
214    #[error("internal error: {0}")]
215    Internal(String),
216}
217
218/// Cache result type
219pub type CacheResult<T> = std::result::Result<T, CacheError>;
220
221/// Cache statistics – matches test expectations
222#[derive(Debug, Clone, Default, Serialize, Deserialize)]
223pub struct CacheStats {
224    /// Number of cache hits
225    pub hits: Option<u64>,
226    /// Number of cache misses
227    pub misses: Option<u64>,
228    /// Number of set operations
229    pub sets: Option<u64>,
230    /// Number of delete operations
231    pub deletes: Option<u64>,
232    /// Total number of cached items
233    pub item_count: Option<u64>,
234    /// Memory usage in bytes (if provided by backend)
235    pub memory_used_bytes: Option<u64>,
236    /// Additional backend-specific metrics
237    pub additional_metrics: std::collections::HashMap<String, String>,
238}
239
240/// Cache service interface
241#[async_trait]
242pub trait CacheService: Send + Sync {
243    /// Get a value from the cache
244    async fn get(&self, key: &str) -> CacheResult<Option<Vec<u8>>>;
245
246    /// Set a value in the cache
247    async fn set(&self, key: &str, value: &[u8], ttl: Option<Duration>) -> CacheResult<()>;
248
249    /// Delete a value from the cache
250    async fn delete(&self, key: &str) -> CacheResult<bool>;
251
252    /// Delete all values from the cache
253    async fn flush(&self) -> CacheResult<()>;
254
255    /// Get cache statistics
256    async fn stats(&self) -> CacheResult<CacheStats>;
257
258    /// Ping the cache service
259    async fn ping(&self) -> CacheResult<()>;
260
261    /// Close the connection (if applicable)
262    async fn close(&self) -> CacheResult<()>;
263
264    // ------------------------------------------------------------------
265    // Extended operations (required by tests) – default implementations
266    // ------------------------------------------------------------------
267
268    /// Check if a key exists without fetching its value
269    async fn exists(&self, key: &str) -> CacheResult<bool> {
270        // Fallback implementation – use `get` and map to boolean
271        Ok(self.get(key).await?.is_some())
272    }
273
274    /// Set a value only if the key does not already exist (NX)
275    async fn set_nx(&self, key: &str, value: &[u8], ttl: Option<Duration>) -> CacheResult<bool> {
276        // Default implementation using get and set
277        if self.exists(key).await? {
278            Ok(false)
279        } else {
280            self.set(key, value, ttl).await?;
281            Ok(true)
282        }
283    }
284
285    /// Atomically fetch the current value and replace it
286    async fn get_set(&self, key: &str, value: &[u8]) -> CacheResult<Option<Vec<u8>>> {
287        // Default implementation using get and set
288        let old_value = self.get(key).await?;
289        self.set(key, value, None).await?;
290        Ok(old_value)
291    }
292
293    /// Increment a numeric value (signed)
294    async fn increment(&self, key: &str, delta: i64) -> CacheResult<i64> {
295        // Default implementation using get, parse, increment, and set
296        let current_value = match self.get(key).await? {
297            Some(bytes) => {
298                let value_str = String::from_utf8(bytes)
299                    .map_err(|e| CacheError::Operation(format!("Invalid UTF-8: {}", e)))?;
300                value_str.parse::<i64>()
301                    .map_err(|e| CacheError::Operation(format!("Invalid integer: {}", e)))?
302            }
303            None => 0,
304        };
305        
306        let new_value = current_value + delta;
307        self.set(key, new_value.to_string().as_bytes(), None).await?;
308        Ok(new_value)
309    }
310
311    /// Decrement convenience wrapper – default delegates to `increment`.
312    async fn decrement(&self, key: &str, delta: i64) -> CacheResult<i64> {
313        self.increment(key, -delta).await
314    }
315
316    /// Set multiple key/value pairs in a single operation
317    async fn set_many(
318        &self,
319        items: &std::collections::HashMap<String, Vec<u8>>,
320        ttl: Option<Duration>,
321    ) -> CacheResult<()> {
322        // Default implementation: set each item individually
323        for (key, value) in items {
324            self.set(key, value, ttl).await?;
325        }
326        Ok(())
327    }
328
329    /// Fetch many keys at once – implementation should skip missing keys
330    async fn get_many(
331        &self,
332        keys: &[String],
333    ) -> CacheResult<std::collections::HashMap<String, Vec<u8>>> {
334        // Default implementation: get each key individually
335        let mut result = std::collections::HashMap::new();
336        for key in keys {
337            if let Some(value) = self.get(key).await? {
338                result.insert(key.clone(), value);
339            }
340        }
341        Ok(result)
342    }
343
344    /// Delete many keys at once – return number of keys removed
345    async fn delete_many(&self, keys: &[String]) -> CacheResult<u64> {
346        // Default implementation: delete each key individually
347        let mut count = 0;
348        for key in keys {
349            if self.delete(key).await? {
350                count += 1;
351            }
352        }
353        Ok(count)
354    }
355
356    /// Clear the cache or a namespace
357    async fn clear(&self, namespace: Option<&str>) -> CacheResult<()> {
358        // Default implementation: This operation is complex and backend-specific
359        // For safety, we'll provide a basic flush if no namespace is specified
360        if namespace.is_none() {
361            self.flush().await
362        } else {
363            Err(CacheError::Operation(
364                "Namespace-specific clear not implemented for this backend. Use flush() to clear all.".to_string(),
365            ))
366        }
367    }
368
369    /// Acquire a simple lock – returns token if lock acquired
370    async fn lock(&self, key: &str, ttl: Duration) -> CacheResult<Option<String>> {
371        // Default implementation using set_nx
372        let lock_key = format!("lock:{}", key);
373        let token = uuid::Uuid::new_v4().to_string();
374        
375        if self.set_nx(&lock_key, token.as_bytes(), Some(ttl)).await? {
376            Ok(Some(token))
377        } else {
378            Ok(None)
379        }
380    }
381
382    /// Release a lock
383    async fn unlock(&self, key: &str, token: &str) -> CacheResult<bool> {
384        // Default implementation: check token and delete if matches
385        let lock_key = format!("lock:{}", key);
386        
387        // Check if the lock exists and has the correct token
388        if let Some(stored_token_bytes) = self.get(&lock_key).await? {
389            let stored_token = String::from_utf8(stored_token_bytes)
390                .map_err(|e| CacheError::Operation(format!("Invalid UTF-8 in stored token: {}", e)))?;
391            
392            if stored_token == token {
393                self.delete(&lock_key).await?;
394                Ok(true)
395            } else {
396                Ok(false) // Token mismatch
397            }
398        } else {
399            Ok(false) // Lock doesn't exist
400        }
401    }
402
403    /// Expose backend type – useful for down-casting & diagnostics
404    fn get_cache_type(&self) -> CacheType {
405        CacheType::InMemory
406    }
407
408    /// Helper accessor for the backend default TTL
409    fn get_default_ttl(&self) -> Duration {
410        Duration::from_secs(0)
411    }
412
413    /// Convenience helper: fetch value as UTF-8 string
414    async fn get_string(&self, key: &str) -> CacheResult<Option<String>> {
415        match self.get(key).await? {
416            Some(bytes) => match String::from_utf8(bytes) {
417                Ok(s) => Ok(Some(s)),
418                Err(e) => Err(CacheError::Get(format!("Invalid UTF-8: {}", e))),
419            },
420            None => Ok(None),
421        }
422    }
423
424    /// Convenience helper: set value as UTF-8 string
425    async fn set_string(&self, key: &str, value: &str, ttl: Option<Duration>) -> CacheResult<()> {
426        self.set(key, value.as_bytes(), ttl).await
427    }
428}
429
430/// Create a cache service from a configuration
431pub async fn create_cache_service(config: &CacheConfig) -> CacheResult<Box<dyn CacheService>> {
432    match config.cache_type {
433        CacheType::InMemory => {
434            let cache = InMemoryCache::new(config);
435            Ok(Box::new(cache) as Box<dyn CacheService>)
436        }
437        CacheType::Redis => {
438            #[cfg(feature = "redis_cache")]
439            {
440                let cache = RedisCache::connect(config).await?;
441                Ok(Box::new(cache) as Box<dyn CacheService>)
442            }
443            #[cfg(not(feature = "redis_cache"))]
444            {
445                Err(CacheError::Configuration(
446                    "Redis cache support is not enabled. Enable the 'redis_cache' feature."
447                        .to_string(),
448                ))
449            }
450        }
451        CacheType::Memcached => {
452            #[cfg(feature = "memcached")]
453            {
454                let cache = MemcachedCache::connect(config).await?;
455                Ok(Box::new(cache) as Box<dyn CacheService>)
456            }
457            #[cfg(not(feature = "memcached"))]
458            {
459                Err(CacheError::Configuration(
460                    "Memcached support is not enabled. Enable the 'memcached' feature.".to_string(),
461                ))
462            }
463        }
464        CacheType::File => {
465            #[cfg(feature = "file_cache")]
466            {
467                let cache = FileCache::new(config).await?;
468                Ok(Box::new(cache) as Box<dyn CacheService>)
469            }
470            #[cfg(not(feature = "file_cache"))]
471            {
472                Err(CacheError::Configuration(
473                    "File cache support is not enabled. Enable the 'file_cache' feature."
474                        .to_string(),
475                ))
476            }
477        }
478    }
479}
pywatt_sdk/data/cache/mod.rs

pywatt_sdk/data/cache/
mod.rs
