grabcli_sqlite/
lib.rs

1//! SQLite cache backend for grabcli: [`SqliteCache`] implements
2//! [`grabcli_core::cache::Cache`] with one database at
3//! `<root>/cache.sqlite` (WAL, mode 0600, inside a 0700 directory):
4//!
5//! - **Responses** (`entries`): raw upstream JSON by key hash.
6//! - **Catalog** ([`catalog`]): restaurants, where they were seen from, menu
7//!   categories and items, with FTS5 word + trigram indexes.
8//! - **Images** ([`images`]): content-addressed files under `<root>/images/`,
9//!   indexed in SQLite and evicted LRU by total size.
10//!
11//! The database is opened lazily on first use; if it can't be opened every
12//! call errors, which [`CacheHandle`](grabcli_core::cache::CacheHandle)
13//! treats as a miss. Housekeeping (`stats`, `clear`, `prune`) is only on
14//! this type, for `grab cache`. Schema migrations live in `db.rs`: append to
15//! `MIGRATIONS`, never edit a shipped one.
16//!
17//! # Example: configure a cache
18//!
19//! ```no_run
20//! use grabcli_core::cache::CacheConfig;
21//! use grabcli_sqlite::SqliteCache;
22//!
23//! let cache = SqliteCache::new("/path/to/private/cache", CacheConfig::default());
24//! let stats = cache.stats()?;
25//! println!("cached responses: {}", stats.entries);
26//! # Ok::<(), grabcli_core::Error>(())
27//! ```
28//!
29//! The first operation opens or creates the database. Give each application its
30//! own private cache directory. Pass an `Arc<SqliteCache>` as the core [`Cache`]
31//! implementation when creating a context.
32
33pub mod catalog;
34mod db;
35mod err;
36pub mod images;
37
38use std::collections::BTreeMap;
39use std::path::{Path, PathBuf};
40use std::sync::{Arc, Mutex, MutexGuard, OnceLock};
41
42use async_trait::async_trait;
43use grabcli_core::cache::{
44    Cache, CacheConfig, CacheEntry, CacheKey, CacheKind, CachedImage, CategoryRow, ItemHit,
45    ItemQuery, ItemRow, RestaurantHit, RestaurantQuery, RestaurantRow, is_fresh,
46};
47use grabcli_core::error::{Error, ErrorKind};
48use grabcli_core::jwt;
49use rusqlite::{Connection, OptionalExtension, params};
50use serde::Serialize;
51
52use crate::err::{DbError, Result};
53
54pub use db::DB_FILE;
55
56type CoreResult<T> = grabcli_core::error::Result<T>;
57
58/// Per-kind numbers in [`CacheStats`].
59#[derive(Debug, Clone, Default, Serialize, PartialEq, Eq)]
60pub struct KindStats {
61    pub entries: u64,
62    pub bytes: u64,
63    pub expired: u64,
64}
65
66/// Image store numbers in [`CacheStats`].
67#[derive(Debug, Clone, Default, Serialize, PartialEq, Eq)]
68pub struct ImageStats {
69    /// Distinct files (several URLs may share one).
70    pub files: u64,
71    pub urls: u64,
72    pub bytes: u64,
73    pub max_bytes: u64,
74}
75
76/// `grab cache stats`.
77#[derive(Debug, Clone, Default, Serialize)]
78pub struct CacheStats {
79    /// The cache directory.
80    pub path: String,
81    pub db_path: String,
82    /// Database + WAL size on disk.
83    pub db_bytes: u64,
84    pub schema_version: u64,
85    pub enabled: bool,
86    pub entries: u64,
87    /// Total size of stored response bodies.
88    pub total_bytes: u64,
89    pub expired: u64,
90    pub kinds: BTreeMap<String, KindStats>,
91    /// Row counts of the catalog tables.
92    pub tables: BTreeMap<String, u64>,
93    pub images: ImageStats,
94}
95
96/// What `grab cache prune` removed.
97#[derive(Debug, Clone, Default, Serialize, PartialEq, Eq)]
98pub struct PruneReport {
99    pub entries: u64,
100    pub restaurants: u64,
101    pub restaurant_locations: u64,
102    pub menu_categories: u64,
103    pub menu_items: u64,
104    pub images_evicted: u64,
105    pub image_bytes_freed: u64,
106}
107
108/// What `grab cache clear --all` removed.
109#[derive(Debug, Clone, Default, Serialize, PartialEq, Eq)]
110pub struct ClearReport {
111    pub entries: u64,
112    pub catalog_rows: u64,
113    pub images: u64,
114}
115
116/// The lazily opened database, shared by clones of a [`Cache`].
117#[derive(Default)]
118struct DbCell {
119    conn: OnceLock<std::result::Result<Mutex<Connection>, String>>,
120}
121
122impl std::fmt::Debug for DbCell {
123    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
124        let state = match self.conn.get() {
125            None => "unopened",
126            Some(Ok(_)) => "open",
127            Some(Err(_)) => "failed",
128        };
129        f.debug_struct("DbCell").field("state", &state).finish()
130    }
131}
132
133/// The SQLite cache. Cheap to clone (clones share one connection).
134#[derive(Debug, Clone)]
135pub struct SqliteCache {
136    root: PathBuf,
137    config: CacheConfig,
138    db: Arc<DbCell>,
139}
140
141impl SqliteCache {
142    /// A cache rooted at `root` (the database is `root/cache.sqlite`).
143    /// `config` supplies the image cap, the catalog max age and the TTLs used
144    /// by [`stats`](Self::stats) and [`prune`](Self::prune).
145    pub fn new(root: impl Into<PathBuf>, config: CacheConfig) -> Self {
146        Self {
147            root: root.into(),
148            config,
149            db: Arc::new(DbCell::default()),
150        }
151    }
152
153    /// The cache directory.
154    pub fn root(&self) -> &Path {
155        &self.root
156    }
157
158    pub fn db_path(&self) -> PathBuf {
159        self.root.join(db::DB_FILE)
160    }
161
162    pub fn config(&self) -> &CacheConfig {
163        &self.config
164    }
165
166    pub fn ttl(&self, kind: CacheKind) -> u64 {
167        self.config.ttl(kind)
168    }
169
170    /// The open connection, opening it on first use.
171    fn conn(&self) -> Result<MutexGuard<'_, Connection>> {
172        if self.root.as_os_str().is_empty() {
173            return Err(DbError(Error::invalid("no cache directory")));
174        }
175        let cell = self.db.conn.get_or_init(|| {
176            db::open(&self.db_path(), jwt::now())
177                .map(Mutex::new)
178                .map_err(|e| e.0.message)
179        });
180        match cell {
181            Ok(m) => Ok(m.lock().unwrap_or_else(std::sync::PoisonError::into_inner)),
182            Err(msg) => Err(DbError(Error::new(ErrorKind::Io, msg.clone()))),
183        }
184    }
185
186    fn get_entry_sync(&self, key: &CacheKey) -> Result<Option<CacheEntry>> {
187        let conn = self.conn()?;
188        Ok(conn
189            .query_row(
190                "SELECT identity, body, fetched_at, expires_at FROM entries WHERE key = ?1",
191                [key.hash()],
192                |r| {
193                    Ok(CacheEntry {
194                        identity: r.get(0)?,
195                        body: r.get(1)?,
196                        fetched_at: r.get(2)?,
197                        expires_at: r.get(3)?,
198                    })
199                },
200            )
201            .optional()?)
202    }
203
204    fn put_entry_sync(&self, key: &CacheKey, e: &CacheEntry) -> Result<()> {
205        let conn = self.conn()?;
206        conn.execute(
207            "INSERT OR REPLACE INTO entries \
208             (key, namespace, kind, identity, body, fetched_at, expires_at, etag) \
209             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, NULL)",
210            params![
211                key.hash(),
212                key.namespace(),
213                key.kind().as_str(),
214                e.identity,
215                e.body,
216                e.fetched_at,
217                e.expires_at
218            ],
219        )?;
220        Ok(())
221    }
222
223    pub fn stats(&self) -> CoreResult<CacheStats> {
224        Ok(self.stats_at(jwt::now())?)
225    }
226
227    #[doc(hidden)]
228    pub fn stats_at(&self, now: i64) -> Result<CacheStats> {
229        let mut s = CacheStats {
230            path: self.root.display().to_string(),
231            db_path: self.db_path().display().to_string(),
232            enabled: self.config.enabled(),
233            ..CacheStats::default()
234        };
235        for kind in CacheKind::ALL {
236            s.kinds.insert(kind.as_str().into(), KindStats::default());
237        }
238        {
239            let conn = self.conn()?;
240            s.schema_version = u64::try_from(db::schema_version(&conn)?).unwrap_or(0);
241            let mut stmt = conn.prepare(
242                "SELECT kind, fetched_at, expires_at, length(CAST(body AS BLOB)) FROM entries",
243            )?;
244            let rows = stmt.query_map([], |r| {
245                Ok((
246                    r.get::<_, String>(0)?,
247                    r.get::<_, i64>(1)?,
248                    r.get::<_, i64>(2)?,
249                    r.get::<_, i64>(3)?,
250                ))
251            })?;
252            for row in rows {
253                let (kind, fetched_at, expires_at, bytes) = row?;
254                let bytes = u64::try_from(bytes).unwrap_or(0);
255                let ttl = kind.parse::<CacheKind>().map(|k| self.ttl(k)).unwrap_or(0);
256                let k = s.kinds.entry(kind).or_default();
257                k.entries += 1;
258                k.bytes += bytes;
259                s.entries += 1;
260                s.total_bytes += bytes;
261                if !is_fresh(fetched_at, expires_at, ttl, now) {
262                    k.expired += 1;
263                    s.expired += 1;
264                }
265            }
266            drop(stmt);
267            for table in catalog::TABLES {
268                let n: i64 =
269                    conn.query_row(&format!("SELECT COUNT(*) FROM {table}"), [], |r| r.get(0))?;
270                s.tables
271                    .insert(table.to_string(), u64::try_from(n).unwrap_or(0));
272            }
273            s.images = images::stats(&conn, self.config.image_max_bytes())?;
274        }
275        s.db_bytes = db::db_size(&self.db_path());
276        Ok(s)
277    }
278
279    /// Remove response entries, optionally only of one kind and/or only
280    /// expired ones. Returns how many were removed. Catalog rows and images are
281    /// kept (see [`clear_all`](Self::clear_all) / [`prune`](Self::prune)).
282    pub fn clear(&self, kind: Option<CacheKind>, expired_only: bool) -> CoreResult<u64> {
283        Ok(self.clear_at(kind, expired_only, jwt::now())?)
284    }
285
286    #[doc(hidden)]
287    pub fn clear_at(&self, kind: Option<CacheKind>, expired_only: bool, now: i64) -> Result<u64> {
288        let conn = self.conn()?;
289        let mut removed = 0;
290        let kinds: Vec<CacheKind> = kind.map_or_else(|| CacheKind::ALL.to_vec(), |k| vec![k]);
291        for k in kinds {
292            let n = if expired_only {
293                let ttl = i64::try_from(self.ttl(k)).unwrap_or(i64::MAX);
294                conn.execute(
295                    "DELETE FROM entries WHERE kind = ?1 AND \
296                     (expires_at <= ?2 OR fetched_at > ?2 OR ?2 - fetched_at >= ?3)",
297                    params![k.as_str(), now, ttl],
298                )?
299            } else {
300                conn.execute("DELETE FROM entries WHERE kind = ?1", [k.as_str()])?
301            };
302            removed += n as u64;
303        }
304        if kind.is_none() && !expired_only {
305            // Rows of kinds this binary doesn't know (written by another
306            // version) go too.
307            removed += conn.execute("DELETE FROM entries", [])? as u64;
308        }
309        Ok(removed)
310    }
311
312    /// Remove everything: responses, catalog rows and image files.
313    pub fn clear_all(&self) -> CoreResult<ClearReport> {
314        Ok(self.clear_all_inner()?)
315    }
316
317    fn clear_all_inner(&self) -> Result<ClearReport> {
318        let mut report = ClearReport::default();
319        {
320            let conn = self.conn()?;
321            report.entries = conn.execute("DELETE FROM entries", [])? as u64;
322            for table in catalog::TABLES {
323                report.catalog_rows += conn.execute(&format!("DELETE FROM {table}"), [])? as u64;
324            }
325            report.images = images::clear(&conn, &self.root)?;
326            conn.execute_batch("PRAGMA wal_checkpoint(TRUNCATE); VACUUM;")?;
327        }
328        Ok(report)
329    }
330
331    /// Housekeeping: drop expired responses, catalog rows older than
332    /// `index_max_age_days`, and evict images over `image_max_mb` (LRU), then
333    /// compact the database.
334    pub fn prune(&self) -> CoreResult<PruneReport> {
335        Ok(self.prune_at(jwt::now())?)
336    }
337
338    #[doc(hidden)]
339    pub fn prune_at(&self, now: i64) -> Result<PruneReport> {
340        let mut report = PruneReport {
341            entries: self.clear_at(None, true, now)?,
342            ..PruneReport::default()
343        };
344        let conn = self.conn()?;
345        // Entries of unknown kinds past their stored expiry.
346        report.entries += conn.execute("DELETE FROM entries WHERE expires_at <= ?1", [now])? as u64;
347        let cutoff =
348            now.saturating_sub(i64::try_from(self.config.index_max_age_secs()).unwrap_or(i64::MAX));
349        let c = catalog::prune_older_than(&conn, cutoff)?;
350        report.restaurants = c[0];
351        report.restaurant_locations = c[1];
352        report.menu_categories = c[2];
353        report.menu_items = c[3];
354        let (n, bytes) = images::evict(&conn, &self.root, self.config.image_max_bytes())?;
355        report.images_evicted = n;
356        report.image_bytes_freed = bytes;
357        conn.execute_batch("PRAGMA wal_checkpoint(TRUNCATE); VACUUM; PRAGMA optimize;")?;
358        Ok(report)
359    }
360}
361
362fn core<T>(r: Result<T>) -> CoreResult<T> {
363    r.map_err(Into::into)
364}
365
366#[async_trait]
367impl Cache for SqliteCache {
368    async fn get_entry(&self, key: &CacheKey) -> CoreResult<Option<CacheEntry>> {
369        core(self.get_entry_sync(key))
370    }
371
372    async fn put_entry(&self, key: &CacheKey, entry: &CacheEntry) -> CoreResult<()> {
373        core(self.put_entry_sync(key, entry))
374    }
375
376    async fn index_restaurants(
377        &self,
378        rows: &[RestaurantRow],
379        seen_from: Option<(f64, f64)>,
380        now: i64,
381    ) -> CoreResult<()> {
382        core(self.index_restaurants_at(rows, seen_from, now))
383    }
384
385    async fn index_items(&self, items: &[ItemRow], now: i64) -> CoreResult<()> {
386        core(self.index_items_at(items, now))
387    }
388
389    async fn index_menu(
390        &self,
391        restaurant: &RestaurantRow,
392        categories: &[CategoryRow],
393        items: &[ItemRow],
394        seen_from: Option<(f64, f64)>,
395        now: i64,
396    ) -> CoreResult<()> {
397        core(self.index_menu_at(restaurant, categories, items, seen_from, now))
398    }
399
400    async fn search_restaurants(&self, q: &RestaurantQuery) -> CoreResult<Vec<RestaurantHit>> {
401        core(SqliteCache::search_restaurants(self, q))
402    }
403
404    async fn search_items(&self, q: &ItemQuery) -> CoreResult<Vec<ItemHit>> {
405        core(SqliteCache::search_items(self, q))
406    }
407
408    async fn restaurant(&self, id: &str) -> CoreResult<Option<RestaurantHit>> {
409        core(SqliteCache::restaurant(self, id))
410    }
411
412    async fn image(&self, url: &str, now: i64) -> CoreResult<Option<CachedImage>> {
413        Ok(self.cached_image_at(url, now))
414    }
415
416    async fn put_image(
417        &self,
418        url: &str,
419        bytes: &[u8],
420        content_type: &str,
421        now: i64,
422    ) -> CoreResult<CachedImage> {
423        core(self.store_image_at(url, bytes, content_type, now))
424    }
425}
426
427#[cfg(test)]
428mod tests {
429    use super::*;
430    use grabcli_core::cache::{CacheHandle, CacheMode};
431    use serde_json::{Value, json};
432
433    fn cache(dir: &Path) -> SqliteCache {
434        SqliteCache::new(dir, CacheConfig::default())
435    }
436
437    fn handle(c: &SqliteCache) -> CacheHandle {
438        CacheHandle::new(Arc::new(c.clone()), c.config.clone(), CacheMode::Normal)
439    }
440
441    fn key() -> CacheKey {
442        CacheKey::new("food", CacheKind::Menu, "merchant")
443            .param("id", "2-EXAMPLE")
444            .location(1.23, 103.81)
445    }
446
447    async fn get_at(c: &SqliteCache, k: &CacheKey, now: i64) -> Option<Value> {
448        handle(c).lookup_at::<Value>(k, now).await.map(|f| f.value)
449    }
450
451    #[tokio::test]
452    async fn roundtrip_through_handle() -> CoreResult<()> {
453        let dir = tempfile::tempdir()?;
454        let c = cache(dir.path());
455        let h = handle(&c);
456        h.put_at(&key(), &json!({"a": 1}), 1_000).await?;
457        assert_eq!(get_at(&c, &key(), 1_000).await, Some(json!({"a": 1})));
458        // A tampered identity is a miss.
459        c.conn()?
460            .execute("UPDATE entries SET identity = 'x'", [])
461            .map_err(DbError::from)?;
462        assert_eq!(get_at(&c, &key(), 1_000).await, None);
463
464        // An unusable cache dir: reads miss, get_or_fetch still works.
465        let file = dir.path().join("not-a-dir");
466        std::fs::write(&file, b"x")?;
467        let broken = handle(&cache(&file));
468        assert_eq!(broken.get::<Value>(&key()).await, None);
469        assert!(broken.put(&key(), &json!(1)).await.is_err());
470        let v: Value = broken
471            .get_or_fetch(&key(), || async { Ok(json!(2)) })
472            .await?;
473        assert_eq!(v, json!(2));
474        Ok(())
475    }
476
477    #[tokio::test]
478    async fn stats_clear_and_prune() -> CoreResult<()> {
479        let dir = tempfile::tempdir()?;
480        let c = cache(dir.path());
481        let h = handle(&c);
482        let menu = key();
483        let list = CacheKey::new("food", CacheKind::List, "search").param("keyword", "x");
484        let mart = CacheKey::new("mart", CacheKind::Categories, "categories");
485        h.put_at(&menu, &json!({"a": 1}), 0).await?; // expired by `now` below
486        h.put_at(&list, &json!([1, 2]), 10_000).await?;
487        h.put_at(&mart, &json!([]), 10_000).await?;
488        let now = 10_001;
489
490        let s = c.stats_at(now)?;
491        assert_eq!(s.entries, 3);
492        assert_eq!(s.expired, 1);
493        assert!(s.total_bytes > 0);
494        assert!(s.db_bytes > 0);
495        assert_eq!(s.schema_version, db::MIGRATIONS.len() as u64);
496        assert_eq!(s.kinds["menu"].expired, 1);
497        assert_eq!(s.kinds["list"].entries, 1);
498        assert_eq!(s.kinds["categories"].entries, 1);
499        assert_eq!(s.kinds["details"].entries, 0);
500        assert_eq!(s.tables["restaurants"], 0);
501
502        assert_eq!(c.clear_at(None, true, now)?, 1);
503        assert_eq!(c.stats_at(now)?.entries, 2);
504        assert_eq!(c.clear_at(Some(CacheKind::List), false, now)?, 1);
505        assert_eq!(get_at(&c, &list, now).await, None);
506        assert_eq!(get_at(&c, &mart, now).await, Some(json!([])));
507
508        // Prune removes what has expired by then.
509        let later = 10_000 + CacheKind::Categories.default_ttl() as i64;
510        let r = c.prune_at(later)?;
511        assert_eq!(r.entries, 1);
512        assert_eq!(c.stats_at(later)?.entries, 0);
513
514        // Expired rows are also pruned (boundedly) when the db is opened.
515        h.put_at(&menu, &json!(1), 0).await?;
516        let reopened = cache(dir.path());
517        assert_eq!(reopened.stats()?.entries, 0);
518        Ok(())
519    }
520
521    #[tokio::test]
522    #[cfg(unix)]
523    async fn private_permissions() -> CoreResult<()> {
524        use std::os::unix::fs::PermissionsExt;
525        let dir = tempfile::tempdir()?;
526        let root = dir.path().join("grabcli");
527        let c = cache(&root);
528        handle(&c).put(&key(), &json!(1)).await?;
529        assert_eq!(
530            std::fs::metadata(c.db_path())?.permissions().mode() & 0o777,
531            0o600
532        );
533        assert_eq!(
534            std::fs::metadata(&root)?.permissions().mode() & 0o777,
535            0o700
536        );
537        Ok(())
538    }
539}
540
541/// Package README examples, compiled as doctests.
542#[cfg(doctest)]
543#[doc = include_str!("../README.md")]
544struct ReadmeDoctests;
grabcli_sqlite/lib.rs

grabcli_sqlite/
lib.rs
