hashiverse_lib/client/client_storage/
sqlite_client_storage.rs

1//! # SQLite-backed [`ClientStorage`] implementation
2//!
3//! File-backed persistence using `rusqlite`. One table per
4//! [`crate::client::client_storage::client_storage::BUCKETS`] entry, each with
5//! `(key TEXT PRIMARY KEY, value BLOB, last_accessed INTEGER)`. Opened with WAL mode
6//! (concurrent readers while a writer commits) and `PRAGMA synchronous = NORMAL` — safe
7//! for local cache use, not for power-loss durability.
8//!
9//! A `parking_lot::Mutex` serialises all SQLite access so we never have two goroutines
10//! hitting the same connection at once. This is the store the native server binary and
11//! the desktop reference client use.
12
13use crate::client::client_storage::client_storage::{ClientStorage, BUCKETS};
14use crate::tools::time::TimeMillis;
15use anyhow::Context;
16use rusqlite::Connection;
17use std::path::PathBuf;
18use std::sync::Arc;
19use parking_lot::Mutex;
20
21pub struct SqliteClientStorage {
22    connection: Arc<Mutex<Connection>>,
23}
24
25impl SqliteClientStorage {
26    /// Creates a new SqliteClientStorage at `{data_dir}/client_storage.db`.
27    /// The `data_dir` directory must already exist.
28    pub async fn new(data_dir: PathBuf) -> anyhow::Result<Arc<Self>> {
29        let database_path = data_dir.join("client_storage.db");
30        let connection = Connection::open(&database_path)
31            .with_context(|| format!("Failed to open SQLite database at {}", database_path.display()))?;
32
33        // WAL mode allows concurrent readers and writers without blocking.
34        // NORMAL sync reduces fsync calls for better write performance — safe against
35        // process crashes but not OS/power crashes (acceptable for a local cache).
36        connection.execute_batch("PRAGMA journal_mode=WAL; PRAGMA synchronous=NORMAL;")?;
37
38        for bucket in BUCKETS {
39            let table_name = sanitize_bucket_name(bucket);
40            connection.execute_batch(&format!(
41                "CREATE TABLE IF NOT EXISTS [{table_name}] (
42                    key TEXT PRIMARY KEY,
43                    value BLOB NOT NULL,
44                    last_accessed INTEGER NOT NULL
45                )"
46            ))?;
47        }
48
49        Ok(Arc::new(Self {
50            connection: Arc::new(Mutex::new(connection)),
51        }))
52    }
53}
54
55fn sanitize_bucket_name(bucket: &str) -> String {
56    bucket.replace(|c: char| !c.is_alphanumeric() && c != '_', "_")
57}
58
59#[async_trait::async_trait]
60impl ClientStorage for SqliteClientStorage {
61    async fn count(&self, bucket: &str) -> anyhow::Result<usize> {
62        let table_name = sanitize_bucket_name(bucket);
63        let connection = self.connection.lock();
64        let count: i64 = connection.query_row(
65            &format!("SELECT COUNT(*) FROM [{table_name}]"),
66            [],
67            |row| row.get(0),
68        )?;
69        let count = usize::try_from(count)?;
70        Ok(count)
71    }
72
73    async fn keys(&self, bucket: &str) -> anyhow::Result<Vec<String>> {
74        let table_name = sanitize_bucket_name(bucket);
75        let connection = self.connection.lock();
76        let mut statement = connection.prepare(&format!("SELECT key FROM [{table_name}]"))?;
77        let keys = statement
78            .query_map([], |row| row.get::<_, String>(0))?
79            .collect::<Result<Vec<_>, _>>()?;
80        Ok(keys)
81    }
82
83    async fn get(&self, bucket: &str, key: &str, time_millis: TimeMillis) -> anyhow::Result<Option<Vec<u8>>> {
84        let table_name = sanitize_bucket_name(bucket);
85        let connection = self.connection.lock();
86
87        let result: Option<Vec<u8>> = connection
88            .query_row(
89                &format!("SELECT value FROM [{table_name}] WHERE key = ?1"),
90                [key],
91                |row| row.get(0),
92            )
93            .optional()?;
94
95        if result.is_some() && time_millis > TimeMillis::zero() {
96            connection.execute(
97                &format!("UPDATE [{table_name}] SET last_accessed = ?1 WHERE key = ?2"),
98                rusqlite::params![time_millis.0, key],
99            )?;
100        }
101
102        Ok(result)
103    }
104
105    async fn put(&self, bucket: &str, key: &str, value: Vec<u8>, time_millis: TimeMillis) -> anyhow::Result<()> {
106        let table_name = sanitize_bucket_name(bucket);
107        let connection = self.connection.lock();
108        connection.execute(
109            &format!("INSERT OR REPLACE INTO [{table_name}] (key, value, last_accessed) VALUES (?1, ?2, ?3)"),
110            rusqlite::params![key, value, time_millis.0],
111        )?;
112        Ok(())
113    }
114
115    async fn remove(&self, bucket: &str, key: &str) -> anyhow::Result<()> {
116        let table_name = sanitize_bucket_name(bucket);
117        let connection = self.connection.lock();
118        connection.execute(
119            &format!("DELETE FROM [{table_name}] WHERE key = ?1"),
120            [key],
121        )?;
122        Ok(())
123    }
124
125    async fn trim(&self, bucket: &str, max_count: usize) -> anyhow::Result<()> {
126        let table_name = sanitize_bucket_name(bucket);
127        let connection = self.connection.lock();
128
129        let count: i64 = connection.query_row(
130            &format!("SELECT COUNT(*) FROM [{table_name}]"),
131            [],
132            |row| row.get(0),
133        )?;
134        let count = usize::try_from(count)?;
135
136        if count > max_count {
137            let num_to_delete = count - max_count;
138            connection.execute(
139                &format!("DELETE FROM [{table_name}] WHERE key IN (SELECT key FROM [{table_name}] ORDER BY last_accessed ASC LIMIT ?1)"),
140                [i64::try_from(num_to_delete)?],
141            )?;
142        }
143
144        Ok(())
145    }
146
147    async fn reset(&self) -> anyhow::Result<()> {
148        let connection = self.connection.lock();
149        for bucket in BUCKETS {
150            let table_name = sanitize_bucket_name(bucket);
151            connection.execute_batch(&format!("DELETE FROM [{table_name}]"))?;
152        }
153        Ok(())
154    }
155}
156
157use rusqlite::OptionalExtension;
158
159#[cfg(test)]
160mod tests {
161    use crate::client::client_storage::client_storage;
162    use crate::client::client_storage::sqlite_client_storage::SqliteClientStorage;
163    use crate::tools::tools::get_temp_dir;
164
165    #[tokio::test]
166    async fn add_test() {
167        let (_temp_dir, temp_dir_path) = get_temp_dir().unwrap();
168        let storage = SqliteClientStorage::new(temp_dir_path.into()).await.unwrap();
169        client_storage::tests::add_test(storage).await;
170    }
171
172    #[tokio::test]
173    async fn trim_test() {
174        let (_temp_dir, temp_dir_path) = get_temp_dir().unwrap();
175        let storage = SqliteClientStorage::new(temp_dir_path.into()).await.unwrap();
176        client_storage::tests::trim_test(storage).await;
177    }
178}
hashiverse_lib/client/client_storage/sqlite_client_storage.rs

hashiverse_lib/client/client_storage/
sqlite_client_storage.rs
