feat(db): SQLite schema and movie migrations (#48)

crates/arr-db/src/lib.rs

@@ -1 +1,221 @@
-//! arr-db — see DESIGN.md.
+//! arr-db — SQLite persistence and migrations. See DESIGN.md §10.
+//!
+//! A few thousand rows and a single writer, so the pool exists for readers
+//! and the schema does the enforcing: foreign keys on, WAL, `STRICT` tables.
+
+use std::path::Path;
+
+use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions, SqliteSynchronous};
+use sqlx::{migrate::MigrateError, SqlitePool};
+
+/// The migrations embedded in the binary, so a deploy is one file.
+pub static MIGRATOR: sqlx::migrate::Migrator = sqlx::migrate!("./migrations");
+
+/// How long a writer waits for the write lock before giving up.
+const BUSY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
+
+/// A connection pool with the pragmas this schema assumes already applied.
+#[derive(Debug, Clone)]
+pub struct Db {
+    pool: SqlitePool,
+}
+
+impl Db {
+    /// Open (creating if absent) the database at `path`.
+    ///
+    /// Every connection gets `foreign_keys` on — SQLite defaults it *off* per
+    /// connection, so setting it anywhere but here is a way to forget it.
+    ///
+    /// # Errors
+    ///
+    /// If the file cannot be created or opened.
+    pub async fn connect(path: impl AsRef<Path>) -> Result<Self, sqlx::Error> {
+        let options = SqliteConnectOptions::new()
+            .filename(path)
+            .create_if_missing(true)
+            .foreign_keys(true)
+            .journal_mode(SqliteJournalMode::Wal)
+            // WAL makes NORMAL durable against process crashes, which is the
+            // failure this actually has. FULL costs an fsync per commit.
+            .synchronous(SqliteSynchronous::Normal)
+            .busy_timeout(BUSY_TIMEOUT);
+
+        let pool = SqlitePoolOptions::new().connect_with(options).await?;
+
+        Ok(Self { pool })
+    }
+
+    /// Apply every migration that has not run yet.
+    ///
+    /// # Errors
+    ///
+    /// If a migration fails, or one already applied no longer matches its
+    /// recorded checksum.
+    pub async fn migrate(&self) -> Result<(), MigrateError> {
+        MIGRATOR.run(&self.pool).await
+    }
+
+    /// The underlying pool, for crates that own their own queries.
+    #[must_use]
+    pub fn pool(&self) -> &SqlitePool {
+        &self.pool
+    }
+
+    /// Every configured root with the policy attached to it (§5.1).
+    ///
+    /// # Errors
+    ///
+    /// If the query fails.
+    pub async fn list_roots(&self) -> Result<Vec<Root>, sqlx::Error> {
+        sqlx::query_as!(
+            Root,
+            r#"
+            SELECT   r.id       AS "id!: i64",
+                     r.kind     AS "kind!: String",
+                     r.audience AS "audience!: String",
+                     r.path     AS "path!: String",
+                     r.policy_id,
+                     p.name     AS "policy_name!: String"
+            FROM     roots r
+            JOIN     policies p ON p.id = r.policy_id
+            ORDER BY r.kind, r.audience
+            "#
+        )
+        .fetch_all(&self.pool)
+        .await
+    }
+}
+
+/// A library root and the policy it carries (§4, §5.1).
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Root {
+    pub id: i64,
+    /// `movie` or `tv`.
+    pub kind: String,
+    /// `main` or `kids`.
+    pub audience: String,
+    pub path: String,
+    pub policy_id: i64,
+    pub policy_name: String,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{Db, MIGRATOR};
+    use sqlx::Row;
+
+    /// A migrated database in a directory that lives as long as the guard.
+    async fn fresh() -> (tempfile::TempDir, Db) {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let db = Db::connect(dir.path().join("arr.db"))
+            .await
+            .expect("connect");
+        db.migrate().await.expect("migrate");
+        (dir, db)
+    }
+
+    #[tokio::test]
+    async fn migrations_apply_from_empty() {
+        let (_dir, db) = fresh().await;
+
+        let applied = MIGRATOR.iter().count();
+        let recorded: i64 = sqlx::query_scalar("SELECT count(*) FROM _sqlx_migrations")
+            .fetch_one(db.pool())
+            .await
+            .expect("count migrations");
+
+        assert_eq!(recorded, i64::try_from(applied).expect("fits"));
+    }
+
+    #[tokio::test]
+    async fn migrations_are_idempotent() {
+        let (_dir, db) = fresh().await;
+        db.migrate().await.expect("second run is a no-op");
+    }
+
+    #[tokio::test]
+    async fn wal_and_foreign_keys_are_on() {
+        let (_dir, db) = fresh().await;
+
+        let journal: String = sqlx::query_scalar("PRAGMA journal_mode")
+            .fetch_one(db.pool())
+            .await
+            .expect("journal_mode");
+        assert_eq!(journal.to_lowercase(), "wal");
+
+        let foreign_keys: i64 = sqlx::query_scalar("PRAGMA foreign_keys")
+            .fetch_one(db.pool())
+            .await
+            .expect("foreign_keys");
+        assert_eq!(foreign_keys, 1);
+    }
+
+    #[tokio::test]
+    async fn seeds_two_movie_roots_with_distinct_policies() {
+        let (_dir, db) = fresh().await;
+
+        let roots = db.list_roots().await.expect("list roots");
+
+        assert_eq!(roots.len(), 2, "§5.1: two movie roots, TV comes with #34");
+        assert!(roots.iter().all(|r| r.kind == "movie"));
+        assert_eq!(roots[0].audience, "kids");
+        assert_eq!(roots[0].path, "/mnt/media/movies/kids");
+        assert_eq!(roots[1].audience, "main");
+        assert_eq!(roots[1].path, "/mnt/media/movies/main");
+        assert_ne!(roots[0].policy_id, roots[1].policy_id);
+    }
+
+    #[tokio::test]
+    async fn kids_policy_blacklists_brazilian_dubs_and_prefers_web_dl() {
+        let (_dir, db) = fresh().await;
+
+        let row = sqlx::query(
+            "SELECT p.dub_blacklist, p.source_weights
+             FROM policies p JOIN roots r ON r.policy_id = p.id
+             WHERE r.audience = 'kids'",
+        )
+        .fetch_one(db.pool())
+        .await
+        .expect("kids policy");
+
+        let dub_blacklist: String = row.get(0);
+        let source_weights: String = row.get(1);
+
+        assert!(dub_blacklist.contains("pt-BR"), "§5.2");
+        // §5.2: European Portuguese dubs are a streaming artefact.
+        let webdl = source_weights.find("WEB-DL").expect("WEB-DL weight");
+        let remux = source_weights.find("Remux").expect("Remux weight");
+        assert!(
+            webdl < remux,
+            "WEB-DL is listed first for kids: {source_weights}"
+        );
+    }
+
+    #[tokio::test]
+    async fn foreign_keys_are_enforced() {
+        let (_dir, db) = fresh().await;
+
+        let err = sqlx::query("INSERT INTO movies (tmdb_id, title, root_id) VALUES (1, 'x', 999)")
+            .execute(db.pool())
+            .await
+            .expect_err("root 999 does not exist");
+
+        assert!(
+            err.to_string().to_lowercase().contains("foreign key"),
+            "{err}"
+        );
+    }
+
+    #[tokio::test]
+    async fn movie_state_is_constrained() {
+        let (_dir, db) = fresh().await;
+
+        sqlx::query(
+            "INSERT INTO movies (tmdb_id, title, root_id, state)
+             SELECT 693134, 'Dune Part Two', id, 'nonsense' FROM roots WHERE audience = 'main'",
+        )
+        .execute(db.pool())
+        .await
+        .expect_err("state is a closed set");
+    }
+}