Skip to main content

max / audiofiles

14.4 KB · 335 lines History Blame Raw
1 //! SQLite database wrapper with versioned migrations for samples, VFS, tags, and analysis tables.
2
3 mod migrations;
4 mod stats;
5
6 use std::path::Path;
7
8 use rusqlite::Connection;
9 use synckit_config::{ConfigError, ConfigStore};
10 use thiserror::Error;
11 use tracing::instrument;
12
13 use crate::config_key::{CONFIG, ConfigKey};
14 use migrations::{MIGRATIONS, register_hash_row_id};
15
16 pub use migrations::SCHEMA_VERSION;
17
18 #[derive(Error, Debug)]
19 pub enum DbError {
20 #[error("SQLite error: {0}")]
21 Sqlite(#[from] rusqlite::Error),
22
23 /// The vault's schema is ahead of what this build knows how to read, so
24 /// opening it would mean querying a shape this code has never seen. See
25 /// [`Database::migrate`] for why that is refused rather than tolerated.
26 #[error(
27 "this vault was written by a newer version of audiofiles \
28 (vault schema {found}, this build understands {supported}). \
29 Update audiofiles to open it."
30 )]
31 VaultTooNew {
32 /// `PRAGMA user_version` read off the vault.
33 found: i32,
34 /// The highest version this build can produce, [`SCHEMA_VERSION`].
35 supported: i32,
36 },
37 }
38
39 /// The config store wraps rusqlite; its one failure mode is the database, so it
40 /// folds into [`DbError::Sqlite`] rather than carrying a second SQL error type.
41 impl From<ConfigError> for DbError {
42 fn from(error: ConfigError) -> Self {
43 match error {
44 ConfigError::Db(error) => DbError::Sqlite(error),
45 }
46 }
47 }
48
49 /// Core database wrapper. All access is synchronous, no async runtime needed,
50 /// safe to use from a CLAP plugin host thread.
51 pub struct Database {
52 conn: Connection,
53 /// The shared config store over `user_config`. Attached, not opened: the
54 /// table and its sync triggers are stood up by this crate's migrations, so
55 /// the store drives the existing table rather than creating its own.
56 config: ConfigStore,
57 }
58
59 /// Compile-time proof that a write transaction is open on the connection.
60 ///
61 /// Constructed only by [`Database::transaction`], and required by the row-write
62 /// functions that must run inside a batched transaction rather than as
63 /// standalone autocommits in a loop (e.g. [`crate::analysis::save_analysis`],
64 /// [`crate::rules::apply_tag_sourced`]). Holding a `&Tx` is the only way to call
65 /// those functions, so "bulk loop of per-row autocommits", the chronic
66 /// per-row-commit pattern, does not compile: there is no `Tx` to pass except
67 /// inside a `transaction` closure that already batches the whole loop.
68 ///
69 /// The token is a zero-sized marker; the SQL still runs on `db.conn()`, which
70 /// participates in the ambient `BEGIN IMMEDIATE` opened by `transaction`.
71 pub struct Tx(());
72
73 impl Database {
74 /// Open (or create) the database at the given path and run migrations.
75 #[instrument(skip_all)]
76 pub fn open(path: impl AsRef<Path>) -> Result<Self, DbError> {
77 let conn = Connection::open(path)?;
78 conn.execute_batch(
79 // WAL + synchronous=NORMAL is the standard durable-but-fast pairing:
80 // commits no longer fsync individually (only at checkpoint), which is
81 // what made the import path ~2 fsyncs/file. NORMAL under WAL can lose
82 // only the last few committed transactions on power loss, never
83 // corruption, acceptable for a local sample library. The cache /
84 // mmap / temp_store pragmas cut page churn on large scans and the
85 // import write batch.
86 "PRAGMA journal_mode=WAL;\
87 PRAGMA synchronous=NORMAL;\
88 PRAGMA foreign_keys=ON;\
89 PRAGMA busy_timeout=5000;\
90 PRAGMA cache_size=-16000;\
91 PRAGMA mmap_size=268435456;\
92 PRAGMA temp_store=MEMORY;\
93 PRAGMA wal_checkpoint(TRUNCATE);",
94 )?;
95 register_hash_row_id(&conn)?;
96 let mut db = Self {
97 conn,
98 config: ConfigStore::attached(&CONFIG),
99 };
100 db.migrate()?;
101 db.seed_config_key_policy()?;
102 Ok(db)
103 }
104
105 /// Flush the WAL back into the main database file and remove the -shm file.
106 ///
107 /// Call after large write batches (e.g. import completion) to keep the
108 /// WAL index fresh and avoid stale memory-mapped state on macOS.
109 pub fn wal_checkpoint(&self) -> Result<(), DbError> {
110 self.conn.execute_batch("PRAGMA wal_checkpoint(TRUNCATE)")?;
111 Ok(())
112 }
113
114 /// Open an in-memory database (for tests).
115 #[instrument(skip_all)]
116 pub fn open_in_memory() -> Result<Self, DbError> {
117 let conn = Connection::open_in_memory()?;
118 conn.execute_batch("PRAGMA foreign_keys=ON;")?;
119 register_hash_row_id(&conn)?;
120 let mut db = Self {
121 conn,
122 config: ConfigStore::attached(&CONFIG),
123 };
124 db.migrate()?;
125 db.seed_config_key_policy()?;
126 Ok(db)
127 }
128
129 /// Seed `config_key_policy` from the [`CONFIG`] spec, the single source of
130 /// truth for which `user_config` keys may sync. Run at every open so the SQL
131 /// export triggers always reflect the current registry: adding a key in Rust
132 /// is enough, no migration needed. Idempotent (clear + reinsert the closed
133 /// set); the rows are the spec's [`policy_rows`](synckit_config::ConfigSpec::policy_rows),
134 /// so an undeclared key has no row and the export filter cannot admit it.
135 fn seed_config_key_policy(&self) -> Result<(), DbError> {
136 self.conn.execute("DELETE FROM config_key_policy", [])?;
137 let mut stmt = self
138 .conn
139 .prepare("INSERT INTO config_key_policy (key, replicated) VALUES (?1, ?2)")?;
140 for row in CONFIG.policy_rows() {
141 stmt.execute(rusqlite::params![row.key, i64::from(row.replicated)])?;
142 }
143 Ok(())
144 }
145
146 /// Read a `user_config` value through the shared config store, `None` when
147 /// unset. The store is attached to the `user_config` table this crate's
148 /// migrations own.
149 pub fn get_config(&self, key: ConfigKey) -> Result<Option<String>, DbError> {
150 Ok(self.config.get(&self.conn, key.as_str())?)
151 }
152
153 /// Write a `user_config` value through the shared config store.
154 ///
155 /// An upsert on the key: writing a key that already exists fires the table's
156 /// UPDATE trigger once, not a DELETE followed by an INSERT the way the old
157 /// `INSERT OR REPLACE` did, so a synced key enqueues one changelog row per
158 /// edit rather than a spurious delete-then-insert pair.
159 pub fn set_config(&self, key: ConfigKey, value: &str) -> Result<(), DbError> {
160 self.config.set(&self.conn, key.as_str(), value)?;
161 Ok(())
162 }
163
164 /// Remove a `user_config` key through the shared config store. Absent
165 /// already is not an error.
166 pub fn delete_config(&self, key: ConfigKey) -> Result<(), DbError> {
167 self.config.unset(&self.conn, key.as_str())?;
168 Ok(())
169 }
170
171 /// Apply pending migrations using PRAGMA user_version as the version tracker.
172 ///
173 /// Each migration step runs inside a transaction so the schema change and
174 /// version bump are atomic, a crash between the two can no longer leave the
175 /// database in an inconsistent state.
176 ///
177 /// The runner is bounded on both sides. Forward is the ordinary case.
178 /// Backward is not possible and must not be attempted silently: a vault
179 /// carrying a version this build has never heard of was written by a newer
180 /// audiofiles, and every query past this point assumes a schema this code
181 /// has seen. Applying nothing and returning `Ok` reads as success and then
182 /// queries a shape it does not understand, survivable for an added column
183 /// and silent data loss for a dropped one, a rename, or a NOT NULL the
184 /// older code never populates.
185 ///
186 /// Nothing exotic is needed to reach it: a rollback to an older release
187 /// after a bad update, a restored backup, or one vault opened from two
188 /// machines running different versions, which is the shape the
189 /// vault-per-detachable-drive workflow is made of.
190 #[instrument(skip_all)]
191 fn migrate(&mut self) -> Result<(), DbError> {
192 let version: i32 = self
193 .conn
194 .query_row("PRAGMA user_version", [], |row| row.get(0))?;
195
196 if version > SCHEMA_VERSION {
197 return Err(DbError::VaultTooNew {
198 found: version,
199 supported: SCHEMA_VERSION,
200 });
201 }
202
203 for (i, sql) in MIGRATIONS.iter().enumerate() {
204 let target = (i + 1) as i32;
205 if version < target {
206 let batch = format!("BEGIN;\n{sql}\nPRAGMA user_version = {target};\nCOMMIT;");
207 match self.conn.execute_batch(&batch) {
208 Ok(()) => {}
209 Err(e) if e.to_string().contains("duplicate column") => {
210 // Recovery path: a prior partial migration committed
211 // some ALTERs before crashing. Re-run the migration in
212 // pieces, tolerating "duplicate column" on ALTERs and
213 // "already exists" on CREATEs (both mean: the prior
214 // partial run got there already; the desired final
215 // state is still reachable). Any OTHER error here is
216 // a real failure, we roll back and surface it,
217 // because silently bumping user_version on a partially
218 // applied schema is the worst possible outcome.
219 let _ = self.conn.execute_batch("ROLLBACK");
220 self.conn.execute_batch("BEGIN")?;
221
222 // ALTER TABLEs first, individually, tolerating duplicates.
223 for line in sql.lines() {
224 let trimmed = line.trim();
225 if trimmed.to_uppercase().starts_with("ALTER TABLE")
226 && trimmed.to_uppercase().contains("ADD COLUMN")
227 && let Err(alter_err) = self.conn.execute_batch(trimmed)
228 && !alter_err.to_string().contains("duplicate column")
229 {
230 let _ = self.conn.execute_batch("ROLLBACK");
231 return Err(DbError::Sqlite(alter_err));
232 }
233 }
234
235 // Non-ALTER statements (CREATE TABLE / INDEX /
236 // TRIGGER, DROP IF EXISTS, INSERT OR IGNORE, plain
237 // INSERT / UPDATE / DELETE). After M018, every
238 // migration from M003 onward is replay-safe by
239 // construction (verified by the
240 // migration_replay_from_version_two_against_full_schema
241 // regression test), so this batch should succeed
242 // cleanly even against a populated schema. "already
243 // exists" stays tolerable as a belt-and-braces guard
244 // for pre-idempotent migration bodies. Anything else
245 // is a real failure, fail fast, don't bump.
246 let non_alter: String = sql
247 .lines()
248 .filter(|l| {
249 let t = l.trim().to_uppercase();
250 !(t.starts_with("ALTER TABLE") && t.contains("ADD COLUMN"))
251 })
252 .collect::<Vec<_>>()
253 .join("\n");
254 if !non_alter.trim().is_empty()
255 && let Err(e) = self.conn.execute_batch(&non_alter)
256 && !e.to_string().contains("already exists")
257 {
258 let _ = self.conn.execute_batch("ROLLBACK");
259 return Err(DbError::Sqlite(e));
260 }
261
262 self.conn
263 .execute_batch(&format!("PRAGMA user_version = {target};\nCOMMIT;"))?;
264 }
265 Err(e) => return Err(DbError::Sqlite(e)),
266 }
267 }
268 }
269
270 Ok(())
271 }
272
273 /// Run a closure inside a SQLite transaction.
274 ///
275 /// Uses `BEGIN IMMEDIATE` to acquire a write lock upfront, preventing
276 /// deadlocks when the closure issues writes. The closure receives a [`Tx`]
277 /// token proving a transaction is open; pass it to row-write functions that
278 /// require batching (the token cannot be constructed any other way, so those
279 /// functions cannot be called in an un-batched per-row loop). The closure
280 /// accesses the same `Database` through the shared `Mutex<Database>`, which is
281 /// safe because the caller already holds the lock.
282 #[instrument(skip_all)]
283 pub fn transaction<T, F>(&self, f: F) -> Result<T, DbError>
284 where
285 F: FnOnce(&Tx) -> Result<T, DbError>,
286 {
287 self.conn.execute_batch("BEGIN IMMEDIATE")?;
288 match f(&Tx(())) {
289 Ok(val) => {
290 self.conn.execute_batch("COMMIT")?;
291 Ok(val)
292 }
293 Err(e) => {
294 if let Err(rb_err) = self.conn.execute_batch("ROLLBACK") {
295 tracing::warn!("ROLLBACK failed after transaction error: {rb_err}");
296 }
297 Err(e)
298 }
299 }
300 }
301
302 /// [`transaction`](Self::transaction) for closures that produce a
303 /// [`CoreError`](crate::error::CoreError), i.e. the row-write batchers in
304 /// `analysis`, `rules`, and `harvest`, which call functions returning
305 /// `CoreError`. Same `BEGIN IMMEDIATE` / commit / rollback semantics and the
306 /// same [`Tx`] proof token; only the closure's error type differs.
307 #[instrument(skip_all)]
308 pub fn transaction_core<T, F>(&self, f: F) -> Result<T, crate::error::CoreError>
309 where
310 F: FnOnce(&Tx) -> Result<T, crate::error::CoreError>,
311 {
312 self.conn.execute_batch("BEGIN IMMEDIATE")?;
313 match f(&Tx(())) {
314 Ok(val) => {
315 self.conn.execute_batch("COMMIT")?;
316 Ok(val)
317 }
318 Err(e) => {
319 if let Err(rb_err) = self.conn.execute_batch("ROLLBACK") {
320 tracing::warn!("ROLLBACK failed after transaction error: {rb_err}");
321 }
322 Err(e)
323 }
324 }
325 }
326
327 /// Borrow the underlying connection for queries.
328 pub fn conn(&self) -> &Connection {
329 &self.conn
330 }
331 }
332
333 #[cfg(test)]
334 mod tests;
335