capp_cache/
file.rs

1use std::{
2    marker::PhantomData,
3    path::{Path, PathBuf},
4};
5
6use async_trait::async_trait;
7use chrono::{Datelike, Utc};
8use fjall::{Database, Keyspace, KeyspaceCreateOptions, PersistMode};
9use serde::{Deserialize, Serialize};
10use tokio::sync::Mutex;
11use tracing::debug;
12
13use crate::{CacheEntry, CacheEntryState, CacheError, HttpCache};
14
15#[derive(Debug, Clone, Serialize, Deserialize)]
16struct CacheIndexEntry {
17    file_path: String,
18    state: CacheEntryState,
19    status_code: u16,
20    created_at: chrono::DateTime<Utc>,
21    last_accessed: chrono::DateTime<Utc>,
22    error_count: i32,
23    last_error: Option<String>,
24}
25
26/// File-backed HTTP cache with a Fjall metadata index.
27///
28/// Layout under the provided base directory:
29/// - `index/`: Fjall keyspace storing cache metadata keyed by cache key
30/// - `files/YYYY/MM/DD/*.json`: full serialized `CacheEntry<T>` payloads
31pub struct FileHttpCache<T>
32where
33    T: Send + Sync + Serialize + for<'de> Deserialize<'de>,
34{
35    base_dir: PathBuf,
36    files_dir: PathBuf,
37    db: Database,
38    entries: Keyspace,
39    lock: Mutex<()>,
40    _marker: PhantomData<T>,
41}
42
43impl<T> FileHttpCache<T>
44where
45    T: Send + Sync + Serialize + for<'de> Deserialize<'de>,
46{
47    pub fn open(path: impl AsRef<Path>) -> Result<Self, CacheError> {
48        let base_dir = path.as_ref().to_path_buf();
49        let files_dir = base_dir.join("files");
50        let index_dir = base_dir.join("index");
51
52        std::fs::create_dir_all(&files_dir)?;
53        std::fs::create_dir_all(&index_dir)?;
54
55        let db = Database::builder(&index_dir).open()?;
56        let entries = db.keyspace("entries", KeyspaceCreateOptions::default)?;
57
58        Ok(Self {
59            base_dir,
60            files_dir,
61            db,
62            entries,
63            lock: Mutex::new(()),
64            _marker: PhantomData,
65        })
66    }
67
68    pub async fn clear_all(&self) -> Result<u64, CacheError> {
69        let _guard = self.lock.lock().await;
70
71        let keys = self.collect_keys()?;
72        let deleted = keys.len() as u64;
73
74        for key in keys {
75            self.remove_internal(&key).await?;
76        }
77
78        Ok(deleted)
79    }
80
81    fn collect_keys(&self) -> Result<Vec<String>, CacheError> {
82        let mut keys = Vec::new();
83        for item in self.entries.iter() {
84            let (key, _) = item.into_inner()?;
85            let key = String::from_utf8(key.to_vec())
86                .map_err(|err| CacheError::Deserialization(err.to_string()))?;
87            keys.push(key);
88        }
89        Ok(keys)
90    }
91
92    fn file_path_from_metadata(
93        &self,
94        metadata: &CacheIndexEntry,
95    ) -> Result<PathBuf, CacheError> {
96        let rel = PathBuf::from(&metadata.file_path);
97        if rel.is_absolute() {
98            return Err(CacheError::Path(format!(
99                "cache metadata contains absolute path: {}",
100                metadata.file_path
101            )));
102        }
103        Ok(self.base_dir.join(rel))
104    }
105
106    fn new_relative_file_path(&self, key: &str) -> PathBuf {
107        let now = Utc::now();
108        let hash = format!("{:x}", md5::compute(key.as_bytes()));
109        PathBuf::from("files")
110            .join(format!("{:04}", now.year()))
111            .join(format!("{:02}", now.month()))
112            .join(format!("{:02}", now.day()))
113            .join(format!("{}-{}.json", now.timestamp_millis(), hash))
114    }
115
116    fn read_metadata(
117        &self,
118        key: &str,
119    ) -> Result<Option<CacheIndexEntry>, CacheError> {
120        let Some(bytes) = self.entries.get(key.as_bytes())? else {
121            return Ok(None);
122        };
123
124        let metadata = serde_json::from_slice(bytes.as_ref())
125            .map_err(|err| CacheError::Deserialization(err.to_string()))?;
126        Ok(Some(metadata))
127    }
128
129    fn write_metadata(
130        &self,
131        key: &str,
132        metadata: &CacheIndexEntry,
133    ) -> Result<(), CacheError> {
134        let bytes = serde_json::to_vec(metadata)
135            .map_err(|err| CacheError::Serialization(err.to_string()))?;
136        self.entries.insert(key.as_bytes(), bytes)?;
137        self.db.persist(PersistMode::SyncAll)?;
138        Ok(())
139    }
140
141    async fn write_entry(
142        &self,
143        path: &Path,
144        entry: &CacheEntry<T>,
145    ) -> Result<(), CacheError> {
146        let bytes = serde_json::to_vec(entry)
147            .map_err(|err| CacheError::Serialization(err.to_string()))?;
148
149        if let Some(parent) = path.parent() {
150            tokio::fs::create_dir_all(parent).await?;
151        }
152
153        tokio::fs::write(path, bytes).await?;
154        Ok(())
155    }
156
157    async fn read_entry(
158        &self,
159        path: &Path,
160    ) -> Result<Option<CacheEntry<T>>, CacheError> {
161        match tokio::fs::read(path).await {
162            Ok(bytes) => {
163                let entry = serde_json::from_slice(&bytes)
164                    .map_err(|err| CacheError::Deserialization(err.to_string()))?;
165                Ok(Some(entry))
166            }
167            Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(None),
168            Err(err) => Err(CacheError::Io(err)),
169        }
170    }
171
172    async fn remove_internal(&self, key: &str) -> Result<(), CacheError> {
173        let Some(metadata) = self.read_metadata(key)? else {
174            return Ok(());
175        };
176
177        let path = self.file_path_from_metadata(&metadata)?;
178        match tokio::fs::remove_file(&path).await {
179            Ok(()) => {}
180            Err(err) if err.kind() == std::io::ErrorKind::NotFound => {}
181            Err(err) => return Err(CacheError::Io(err)),
182        }
183
184        self.entries.remove(key.as_bytes())?;
185        self.db.persist(PersistMode::SyncAll)?;
186        Ok(())
187    }
188}
189
190#[async_trait]
191impl<T> HttpCache<T> for FileHttpCache<T>
192where
193    T: Send + Sync + Serialize + for<'de> Deserialize<'de>,
194{
195    async fn get(&self, key: &str) -> Result<Option<CacheEntry<T>>, CacheError> {
196        let _guard = self.lock.lock().await;
197
198        let Some(mut metadata) = self.read_metadata(key)? else {
199            return Ok(None);
200        };
201
202        let path = self.file_path_from_metadata(&metadata)?;
203        let Some(mut entry) = self.read_entry(&path).await? else {
204            self.entries.remove(key.as_bytes())?;
205            self.db.persist(PersistMode::SyncAll)?;
206            return Ok(None);
207        };
208
209        let now = Utc::now();
210        entry.last_accessed = now;
211        metadata.last_accessed = now;
212        self.write_entry(&path, &entry).await?;
213        self.write_metadata(key, &metadata)?;
214        Ok(Some(entry))
215    }
216
217    async fn set(
218        &self,
219        key: &str,
220        value: T,
221        status_code: u16,
222        state: CacheEntryState,
223    ) -> Result<(), CacheError> {
224        let _guard = self.lock.lock().await;
225
226        let existing = self.read_metadata(key)?;
227        let now = Utc::now();
228        let rel_path = existing
229            .as_ref()
230            .map(|metadata| PathBuf::from(&metadata.file_path))
231            .unwrap_or_else(|| self.new_relative_file_path(key));
232        let path = self.base_dir.join(&rel_path);
233        let created_at = existing
234            .as_ref()
235            .map(|metadata| metadata.created_at)
236            .unwrap_or(now);
237        let error_count = if state == CacheEntryState::Error {
238            1
239        } else {
240            0
241        };
242        let last_error = if state == CacheEntryState::Error {
243            Some("cache entry marked as error".to_string())
244        } else {
245            None
246        };
247
248        let entry = CacheEntry {
249            key: key.to_string(),
250            value,
251            status_code,
252            state: state.clone(),
253            created_at,
254            last_accessed: now,
255            error_count,
256            last_error: last_error.clone(),
257        };
258
259        self.write_entry(&path, &entry).await?;
260
261        let metadata = CacheIndexEntry {
262            file_path: rel_path.to_string_lossy().into_owned(),
263            state,
264            status_code,
265            created_at,
266            last_accessed: now,
267            error_count,
268            last_error,
269        };
270        self.write_metadata(key, &metadata)?;
271
272        debug!("cached HTTP entry for key: {}", key);
273        Ok(())
274    }
275
276    async fn update_state(
277        &self,
278        key: &str,
279        state: CacheEntryState,
280    ) -> Result<(), CacheError> {
281        let _guard = self.lock.lock().await;
282
283        let Some(mut metadata) = self.read_metadata(key)? else {
284            return Err(CacheError::NotFound(key.to_string()));
285        };
286
287        let path = self.file_path_from_metadata(&metadata)?;
288        let Some(mut entry) = self.read_entry(&path).await? else {
289            self.entries.remove(key.as_bytes())?;
290            self.db.persist(PersistMode::SyncAll)?;
291            return Err(CacheError::NotFound(key.to_string()));
292        };
293
294        let now = Utc::now();
295        entry.state = state.clone();
296        entry.last_accessed = now;
297        metadata.state = state.clone();
298        metadata.last_accessed = now;
299
300        if state == CacheEntryState::Error {
301            entry.error_count += 1;
302            metadata.error_count += 1;
303            let message = "cache entry marked as error".to_string();
304            entry.last_error = Some(message.clone());
305            metadata.last_error = Some(message);
306        }
307
308        self.write_entry(&path, &entry).await?;
309        self.write_metadata(key, &metadata)?;
310        Ok(())
311    }
312
313    async fn remove(&self, key: &str) -> Result<(), CacheError> {
314        let _guard = self.lock.lock().await;
315
316        if self.read_metadata(key)?.is_none() {
317            return Err(CacheError::NotFound(key.to_string()));
318        }
319
320        self.remove_internal(key).await
321    }
322
323    async fn cleanup(&self) -> Result<u64, CacheError> {
324        let _guard = self.lock.lock().await;
325
326        let mut keys_to_delete = Vec::new();
327        for item in self.entries.iter() {
328            let (key, value) = item.into_inner()?;
329            let key = String::from_utf8(key.to_vec())
330                .map_err(|err| CacheError::Deserialization(err.to_string()))?;
331            let metadata: CacheIndexEntry = serde_json::from_slice(value.as_ref())
332                .map_err(|err| CacheError::Deserialization(err.to_string()))?;
333
334            if metadata.state == CacheEntryState::Invalid
335                || (metadata.state == CacheEntryState::Error
336                    && metadata.error_count > 3)
337            {
338                keys_to_delete.push(key);
339            }
340        }
341
342        let deleted = keys_to_delete.len() as u64;
343        for key in keys_to_delete {
344            self.remove_internal(&key).await?;
345        }
346
347        Ok(deleted)
348    }
349
350    async fn contains(&self, key: &str) -> Result<bool, CacheError> {
351        let _guard = self.lock.lock().await;
352        Ok(self.read_metadata(key)?.is_some())
353    }
354}
355
356impl<T> std::fmt::Debug for FileHttpCache<T>
357where
358    T: Send + Sync + Serialize + for<'de> Deserialize<'de>,
359{
360    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
361        f.debug_struct("FileHttpCache")
362            .field("base_dir", &self.base_dir)
363            .field("files_dir", &self.files_dir)
364            .finish()
365    }
366}
capp_cache/file.rs

capp_cache/
file.rs
