Files
arr/crates/arr-db/src/lib.rs
T
Miguel Palhas 34e5663920 Merge main into blitz/subtitles
Feedback pass 2 and the size-band work landed on main while this branch
was finishing. Brings them in ahead of the merge back.

# Conflicts:
#	crates/arr-api/src/movies.rs
#	crates/arr-api/src/state.rs
#	crates/arr-daemon/src/main.rs
#	web/src/main.ts
2026-08-25 17:53:18 +01:00

818 lines
30 KiB
Rust

//! 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;
pub mod blacklist;
pub mod policy;
pub mod subtitle_budget;
pub mod subtitles;
pub use blacklist::Blacklist;
pub use policy::{MoviePolicy, PolicyColumns, PolicyError, TitlePolicy};
pub use subtitle_budget::BudgetKind;
pub use subtitles::{
NewSubtitleFile, PendingSubtitle, SubtitleAttempt, SubtitleFile, SubtitleOrigin, SubtitleState,
SubtitleSync,
};
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");
/// §5.7: how long a failed grab keeps counting toward the needs-a-decision
/// queue, as a SQLite time modifier. Measured from `failed_at` — the moment
/// the grab hard-failed, not the moment it was grabbed — so a torrent that
/// stalls past the window and then fails still counts (#239).
///
/// Nothing ever clears a `grabs` row, so without a bound the queue only grows
/// and the one season that wants attention sits behind the ones that do not.
/// Callers pair it with the `failed_at` format:
/// `strftime('%Y-%m-%dT%H:%M:%fZ', 'now', ATTENTION_WINDOW)`.
pub const ATTENTION_WINDOW: &str = "-30 days";
/// 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");
let movie_roots: Vec<_> = roots.iter().filter(|root| root.kind == "movie").collect();
assert_eq!(movie_roots.len(), 2);
assert_eq!(movie_roots[0].audience, "kids");
assert_eq!(movie_roots[0].path, "/mnt/media/movies/kids");
assert_eq!(movie_roots[1].audience, "main");
assert_eq!(movie_roots[1].path, "/mnt/media/movies/main");
assert_ne!(movie_roots[0].policy_id, movie_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 every_policy_carries_its_scoring_numbers() {
let (_dir, db) = fresh().await;
let rows = sqlx::query("SELECT size_bands, score_weights FROM policies")
.fetch_all(db.pool())
.await
.expect("policies");
assert_eq!(rows.len(), 4);
for row in rows {
let size_bands: String = row.get(0);
let score_weights: String = row.get(1);
// §5.5: floor, target and the penalty above target, per resolution.
assert!(size_bands.contains("floor_gib"), "{size_bands}");
assert!(size_bands.contains("target_gib"), "{size_bands}");
assert!(
size_bands.contains("penalty_points_per_gib_over"),
"{size_bands}"
);
// §5.5: the weights that keep source tier and seeders small
// against the size term are tuned by hand too.
assert!(score_weights.contains("size_at_target"), "{score_weights}");
assert!(score_weights.contains("source_tier"), "{score_weights}");
assert!(score_weights.contains("seeder_doubling"), "{score_weights}");
// §5.5: and the step that ranks resolutions against each other.
assert!(score_weights.contains("resolution_step"), "{score_weights}");
}
}
/// §9.6: the stored artwork columns are nullable, and `REAL` is a valid
/// STRICT type — a rating must not be squeezed into an integer.
#[tokio::test]
async fn title_artwork_columns_are_nullable_and_take_reals() {
let (_dir, db) = fresh().await;
sqlx::query(
"INSERT INTO movies (tmdb_id, title, root_id)
SELECT 1, 'No Artwork Yet', id FROM roots WHERE kind = 'movie' LIMIT 1",
)
.execute(db.pool())
.await
.expect("row without artwork");
sqlx::query(
"UPDATE movies SET poster_path = '/dune.jpg', backdrop_path = '/dune-wide.jpg',
vote_average = 8.152
WHERE tmdb_id = 1",
)
.execute(db.pool())
.await
.expect("artwork write");
let (poster, backdrop, vote): (Option<String>, Option<String>, Option<f64>) =
sqlx::query_as("SELECT poster_path, backdrop_path, vote_average FROM movies")
.fetch_one(db.pool())
.await
.expect("movie row");
assert_eq!(poster.as_deref(), Some("/dune.jpg"));
assert_eq!(backdrop.as_deref(), Some("/dune-wide.jpg"));
assert_eq!(vote, Some(8.152));
}
#[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");
}
/// #73: migration 0007 renames the pre-canonical `grabbed`/`imported`
/// values in place rather than dropping the rows that held them.
#[tokio::test]
async fn movie_state_migration_renames_existing_rows_in_place() {
let dir = tempfile::tempdir().expect("tempdir");
let db = Db::connect(dir.path().join("arr.db"))
.await
.expect("connect");
MIGRATOR
.run_to(6, db.pool())
.await
.expect("migrations before #73");
sqlx::query(
"INSERT INTO movies (tmdb_id, title, root_id, state)
SELECT 1, 'Old Vocabulary', id, 'grabbed' FROM roots WHERE kind = 'movie' LIMIT 1",
)
.execute(db.pool())
.await
.expect("legacy 'grabbed' row, valid under the pre-#73 constraint");
sqlx::query(
"INSERT INTO movies (tmdb_id, title, root_id, state)
SELECT 2, 'Also Old', id, 'imported' FROM roots WHERE kind = 'movie' LIMIT 1",
)
.execute(db.pool())
.await
.expect("legacy 'imported' row, valid under the pre-#73 constraint");
db.migrate()
.await
.expect("remaining migrations, including #73");
let renamed_from_grabbed: String =
sqlx::query_scalar("SELECT state FROM movies WHERE tmdb_id = 1")
.fetch_one(db.pool())
.await
.expect("row survives the migration");
let renamed_from_imported: String =
sqlx::query_scalar("SELECT state FROM movies WHERE tmdb_id = 2")
.fetch_one(db.pool())
.await
.expect("row survives the migration");
assert_eq!(renamed_from_grabbed, "downloading");
assert_eq!(renamed_from_imported, "available");
}
/// The four seeded roots, up to but not including migration 0031, with
/// the given legacy paths written straight into the rows.
async fn roots_before_normalisation(paths: &[(i64, &str)]) -> (tempfile::TempDir, Db) {
let dir = tempfile::tempdir().expect("tempdir");
let db = Db::connect(dir.path().join("arr.db"))
.await
.expect("connect");
MIGRATOR
.run_to(30, db.pool())
.await
.expect("migrations before #244");
for (id, path) in paths {
sqlx::query("UPDATE roots SET path = ? WHERE id = ?")
.bind(path)
.bind(id)
.execute(db.pool())
.await
.expect("legacy path, valid under the unique index");
}
(dir, db)
}
async fn root_path(db: &Db, id: i64) -> String {
sqlx::query_scalar("SELECT path FROM roots WHERE id = ?")
.bind(id)
.fetch_one(db.pool())
.await
.expect("root path")
}
/// #244: a root's path could be stored with a trailing separator until
/// #243 normalised the incoming value, and `roots::update` compares a
/// normalised payload against the stored value — so such a row could
/// never be edited again. Migration 0031 strips the separator, and skips
/// a row whose stripped form another row already holds rather than
/// tripping the unique index and refusing to apply.
#[tokio::test]
async fn root_paths_are_normalised_but_never_onto_a_path_in_use() {
let (_dir, db) = roots_before_normalisation(&[
(1, "/mnt/media/movies/main/"),
(2, "/mnt/collide"),
(3, "/mnt/collide/"),
(4, "/mnt/media/tv/kids///"),
])
.await;
db.migrate()
.await
.expect("0031 applies with a collision present");
assert_eq!(root_path(&db, 1).await, "/mnt/media/movies/main");
assert_eq!(
root_path(&db, 4).await,
"/mnt/media/tv/kids",
"every trailing separator goes in one pass"
);
assert_eq!(root_path(&db, 2).await, "/mnt/collide");
assert_eq!(
root_path(&db, 3).await,
"/mnt/collide/",
"left exactly as it is: normalising it would collide with root 2"
);
}
/// #244: the harder half of the same guard. Two rows that strip to the
/// same path, *neither* of which already holds the stripped value, still
/// have to be left alone — normalising them would collide with each
/// other, and a migration that cannot apply stops the daemon booting.
#[tokio::test]
async fn two_rows_that_would_collide_with_each_other_stop_nothing() {
let (_dir, db) =
roots_before_normalisation(&[(1, "/mnt/one/"), (2, "/mnt/dup/"), (3, "/mnt/dup//")])
.await;
db.migrate()
.await
.expect("0031 applies with a mutually colliding pair present");
assert_eq!(root_path(&db, 2).await, "/mnt/dup/");
assert_eq!(root_path(&db, 3).await, "/mnt/dup//");
assert_eq!(
root_path(&db, 1).await,
"/mnt/one",
"the rest of the table is still normalised"
);
}
/// #155: the table-rebuild migrations (0007, 0014, 0021) run outside a
/// transaction so `PRAGMA foreign_keys = OFF` holds and dropping the old
/// tables does not cascade-delete `movie_releases`/`episode_releases`.
#[tokio::test]
async fn table_rebuild_preserves_child_rows() {
let dir = tempfile::tempdir().expect("tempdir");
let db = Db::connect(dir.path().join("arr.db"))
.await
.expect("connect");
MIGRATOR
.run_to(6, db.pool())
.await
.expect("migrations before the first rebuild");
sqlx::query(
"INSERT INTO movies (tmdb_id, title, root_id)
SELECT 693134, 'Dune Part Two', id FROM roots WHERE kind = 'movie' LIMIT 1",
)
.execute(db.pool())
.await
.expect("movie");
let series_id = sqlx::query(
"INSERT INTO series (tmdb_id, title, root_id)
SELECT 82728, 'Bluey', id FROM roots WHERE kind = 'tv' LIMIT 1",
)
.execute(db.pool())
.await
.expect("series")
.last_insert_rowid();
let season_id = sqlx::query("INSERT INTO seasons (series_id, number) VALUES (?, 1)")
.bind(series_id)
.execute(db.pool())
.await
.expect("season")
.last_insert_rowid();
let episode_id =
sqlx::query("INSERT INTO episodes (season_id, number, title) VALUES (?, 1, 'x')")
.bind(season_id)
.execute(db.pool())
.await
.expect("episode")
.last_insert_rowid();
let release_id = sqlx::query(
"INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed)
VALUES (1, 'guid-1', 'Some.Release', 1024, 'http://x', '{}')",
)
.execute(db.pool())
.await
.expect("release")
.last_insert_rowid();
sqlx::query(
"INSERT INTO movie_releases (movie_id, release_id)
SELECT id, ? FROM movies",
)
.bind(release_id)
.execute(db.pool())
.await
.expect("movie link");
sqlx::query("INSERT INTO episode_releases (episode_id, release_id) VALUES (?, ?)")
.bind(episode_id)
.bind(release_id)
.execute(db.pool())
.await
.expect("episode link");
db.migrate().await.expect("remaining migrations");
let movie_links: i64 = sqlx::query_scalar("SELECT count(*) FROM movie_releases")
.fetch_one(db.pool())
.await
.expect("movie_releases");
let episode_links: i64 = sqlx::query_scalar("SELECT count(*) FROM episode_releases")
.fetch_one(db.pool())
.await
.expect("episode_releases");
assert_eq!(movie_links, 1, "movie_releases survive the movies rebuild");
assert_eq!(
episode_links, 1,
"episode_releases survive the episodes rebuild"
);
}
/// #211: 0032 rebuilds `releases`, which is a parent — `grabs`,
/// `movie_releases`, `episode_releases` and `season_releases` all point
/// at it, three of them ON DELETE CASCADE. The rebuild runs with foreign
/// keys off so dropping the old table neither cascades those children
/// away nor is refused by `grabs`.
#[tokio::test]
async fn the_releases_rebuild_preserves_children_and_rows() {
let (_dir, db) = a_deck_with_every_child_row().await;
db.migrate().await.expect("remaining migrations");
for (label, query, expected) in [
("releases", "SELECT count(*) FROM releases", 3),
("movie_releases", "SELECT count(*) FROM movie_releases", 1),
(
"episode_releases",
"SELECT count(*) FROM episode_releases",
1,
),
("season_releases", "SELECT count(*) FROM season_releases", 1),
("grabs", "SELECT count(*) FROM grabs", 1),
] {
let rows: i64 = sqlx::query_scalar(query)
.fetch_one(db.pool())
.await
.expect("count");
assert_eq!(rows, expected, "{label} survives the releases rebuild");
}
// An existing waived row keeps its NULL and reads as it did before.
let waived: Option<String> =
sqlx::query_scalar("SELECT rejected_rule FROM releases WHERE guid = 'waived-guid'")
.fetch_one(db.pool())
.await
.expect("waived row");
assert_eq!(waived, None);
let rejected: Option<String> =
sqlx::query_scalar("SELECT rejected_rule FROM releases WHERE guid = 'rejected-guid'")
.fetch_one(db.pool())
.await
.expect("rejected row");
assert_eq!(rejected.as_deref(), Some("size"));
// Foreign keys are back on for the connection the migration used.
let foreign_keys: i64 = sqlx::query_scalar("PRAGMA foreign_keys")
.fetch_one(db.pool())
.await
.expect("foreign_keys");
assert_eq!(foreign_keys, 1);
}
/// A database migrated to just before 0032, holding one release per
/// verdict and one row in every table that points at `releases`.
async fn a_deck_with_every_child_row() -> (tempfile::TempDir, Db) {
let dir = tempfile::tempdir().expect("tempdir");
let db = Db::connect(dir.path().join("arr.db"))
.await
.expect("connect");
MIGRATOR
.run_to(31, db.pool())
.await
.expect("migrations before the releases rebuild");
let movie_id = sqlx::query(
"INSERT INTO movies (tmdb_id, title, root_id)
SELECT 693134, 'Dune Part Two', id FROM roots WHERE kind = 'movie' LIMIT 1",
)
.execute(db.pool())
.await
.expect("movie")
.last_insert_rowid();
let series_id = sqlx::query(
"INSERT INTO series (tmdb_id, title, root_id)
SELECT 82728, 'Bluey', id FROM roots WHERE kind = 'tv' LIMIT 1",
)
.execute(db.pool())
.await
.expect("series")
.last_insert_rowid();
let season_id = sqlx::query("INSERT INTO seasons (series_id, number) VALUES (?, 1)")
.bind(series_id)
.execute(db.pool())
.await
.expect("season")
.last_insert_rowid();
let episode_id =
sqlx::query("INSERT INTO episodes (season_id, number, title) VALUES (?, 1, 'x')")
.bind(season_id)
.execute(db.pool())
.await
.expect("episode")
.last_insert_rowid();
// One row per verdict, including the waived row this issue is about,
// which under the old constraint could only hold NULL.
for (guid, verdict, rule) in [
("eligible-guid", "eligible", None),
("waived-guid", "waived", None),
("rejected-guid", "rejected", Some("size")),
] {
sqlx::query(
"INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed, verdict, rejected_rule)
VALUES (1, ?, ?, 1024, 'http://x', '{}', ?, ?)",
)
.bind(guid)
.bind(guid)
.bind(verdict)
.bind(rule)
.execute(db.pool())
.await
.expect("release");
}
let release_id: i64 =
sqlx::query_scalar("SELECT id FROM releases WHERE guid = 'eligible-guid'")
.fetch_one(db.pool())
.await
.expect("release id");
sqlx::query("INSERT INTO movie_releases (movie_id, release_id) VALUES (?, ?)")
.bind(movie_id)
.bind(release_id)
.execute(db.pool())
.await
.expect("movie link");
sqlx::query("INSERT INTO episode_releases (episode_id, release_id) VALUES (?, ?)")
.bind(episode_id)
.bind(release_id)
.execute(db.pool())
.await
.expect("episode link");
sqlx::query("INSERT INTO season_releases (season_id, release_id) VALUES (?, ?)")
.bind(season_id)
.bind(release_id)
.execute(db.pool())
.await
.expect("season link");
sqlx::query(
"INSERT INTO grabs (release_id, target_kind, target_id, infohash)
VALUES (?, 'movie', ?, 'infohash-1')",
)
.bind(release_id)
.bind(movie_id)
.execute(db.pool())
.await
.expect("grab");
(dir, db)
}
/// The relaxed constraint (#211) admits a rule on a waiver and still
/// refuses a rejection without one.
#[tokio::test]
async fn a_waiver_may_name_the_rule_it_relaxed() {
let (_dir, db) = fresh().await;
sqlx::query(
"INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed, verdict, rejected_rule)
VALUES (1, 'waived', 'Some.Release', 1024, 'http://x', '{}', 'waived', 'size')",
)
.execute(db.pool())
.await
.expect("a waiver names its rule");
sqlx::query(
"INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed, verdict)
VALUES (1, 'rejected', 'Some.Release', 1024, 'http://x', '{}', 'rejected')",
)
.execute(db.pool())
.await
.expect_err("a rejection still has to name its rule");
}
#[tokio::test]
async fn seeds_two_tv_roots_with_distinct_policies() {
let (_dir, db) = fresh().await;
let roots = db.list_roots().await.expect("list roots");
let tv_roots: Vec<_> = roots.iter().filter(|root| root.kind == "tv").collect();
assert_eq!(tv_roots.len(), 2);
assert_eq!(tv_roots[0].audience, "kids");
assert_eq!(tv_roots[0].path, "/mnt/media/tv/kids");
assert_eq!(tv_roots[1].audience, "main");
assert_eq!(tv_roots[1].path, "/mnt/media/tv/main");
assert_ne!(tv_roots[0].policy_id, tv_roots[1].policy_id);
}
#[tokio::test]
async fn episode_wanted_is_independent_of_tracking_rules() {
let (_dir, db) = fresh().await;
let root_id: i64 =
sqlx::query_scalar("SELECT id FROM roots WHERE kind = 'tv' AND audience = 'main'")
.fetch_one(db.pool())
.await
.expect("TV root");
let series_id = sqlx::query("INSERT INTO series (tmdb_id, title, root_id, auto_track) VALUES (82728, 'Bluey', ?, 1)")
.bind(root_id)
.execute(db.pool())
.await
.expect("series")
.last_insert_rowid();
let season_id =
sqlx::query("INSERT INTO seasons (series_id, number, tracked) VALUES (?, 1, 1)")
.bind(series_id)
.execute(db.pool())
.await
.expect("season")
.last_insert_rowid();
sqlx::query("INSERT INTO episodes (season_id, number, title) VALUES (?, 2, 'Hospital')")
.bind(season_id)
.execute(db.pool())
.await
.expect("episode");
let wanted: bool = sqlx::query_scalar("SELECT wanted FROM episodes")
.fetch_one(db.pool())
.await
.expect("wanted");
assert!(
!wanted,
"auto_track and tracked are rules, not episode intent"
);
sqlx::query("INSERT INTO movies (tmdb_id, title, root_id) SELECT 693134, 'Dune Part Two', id FROM roots WHERE kind = 'movie' AND audience = 'main'")
.execute(db.pool())
.await
.expect("existing movie schema remains writable");
}
#[tokio::test]
async fn aggregates_require_roots_of_their_own_kind() {
let (_dir, db) = fresh().await;
let movie_root: i64 =
sqlx::query_scalar("SELECT id FROM roots WHERE kind = 'movie' LIMIT 1")
.fetch_one(db.pool())
.await
.expect("movie root");
let tv_root: i64 = sqlx::query_scalar("SELECT id FROM roots WHERE kind = 'tv' LIMIT 1")
.fetch_one(db.pool())
.await
.expect("TV root");
sqlx::query("INSERT INTO movies (tmdb_id, title, root_id) VALUES (1, 'x', ?)")
.bind(tv_root)
.execute(db.pool())
.await
.expect_err("movie cannot use TV root");
sqlx::query("INSERT INTO series (tmdb_id, title, root_id) VALUES (2, 'x', ?)")
.bind(movie_root)
.execute(db.pool())
.await
.expect_err("series cannot use movie root");
}
#[tokio::test]
async fn episode_state_and_upstream_completion_are_constrained() {
let (_dir, db) = fresh().await;
let root_id: i64 =
sqlx::query_scalar("SELECT id FROM roots WHERE kind = 'tv' AND audience = 'main'")
.fetch_one(db.pool())
.await
.expect("TV root");
let series_id = sqlx::query(
"INSERT INTO series (tmdb_id, title, root_id, upstream_ended)
VALUES (82728, 'Bluey', ?, 1)",
)
.bind(root_id)
.execute(db.pool())
.await
.expect("ended series")
.last_insert_rowid();
let season_id = sqlx::query("INSERT INTO seasons (series_id, number) VALUES (?, 1)")
.bind(series_id)
.execute(db.pool())
.await
.expect("season")
.last_insert_rowid();
sqlx::query(
"INSERT INTO episodes (season_id, number, title, state)
VALUES (?, 1, 'Magic Xylophone', 'grabbed')",
)
.bind(season_id)
.execute(db.pool())
.await
.expect_err("database mirrors MediaState");
sqlx::query("UPDATE series SET upstream_ended = 2 WHERE id = ?")
.bind(series_id)
.execute(db.pool())
.await
.expect_err("upstream completion is boolean");
}
}