yolo/arr
3
0
Fork 0
Code Issues 12 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(arr): add subtitle queries

Browse Source 
Recording a subtitle is idempotent on both keys the schema carries: the
sidecar path, and the language an embedded track satisfies, so a second
probe or a re-import converges instead of duplicating. Attempts are
upserted per (media file, language); the work list is every language that
is not satisfied, newest import first, which is the order §15 wants the
daily allowance spent in.
This commit is contained in:
Miguel Palhas
2026-08-24 21:46:13 +01:00
parent 151135b545
commit c3bd009442
13 changed files with 1291 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.sqlx/query-08c356bfcd7f7cd7a4bac0bf0ce44ce3276577cb18f32b0c01839150e7827bca.json
Generated
+12
View File
@@ -0,0 +1,12 @@
{
"db_name": "SQLite",
"query": "INSERT INTO subtitle_attempts\n (media_file_id, language, state, attempts, last_attempt_at, last_failure)\n VALUES (?, ?, ?, 1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'), ?)\n ON CONFLICT (media_file_id, language) DO UPDATE\n SET state = excluded.state,\n attempts = subtitle_attempts.attempts + 1,\n last_attempt_at = excluded.last_attempt_at,\n last_failure = excluded.last_failure,\n updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')",
"describe": {
"columns": [],
"parameters": {
"Right": 4
},
"nullable": []
},
"hash": "08c356bfcd7f7cd7a4bac0bf0ce44ce3276577cb18f32b0c01839150e7827bca"
}
.sqlx/query-47ad9162ba373dfbdf061898f5f148e65de261b494453804db912a919a78c2c5.json
Generated
+98
View File
@@ -0,0 +1,98 @@
{
"db_name": "SQLite",
"query": "SELECT a.media_file_id AS \"media_file_id!: i64\",\n a.language AS \"language!: String\",\n a.state AS \"state!: SubtitleState\",\n a.attempts AS \"attempts!: i64\",\n a.last_attempt_at,\n a.last_failure,\n f.path AS \"media_file_path!: String\"\n FROM subtitle_attempts a\n JOIN media_files f ON f.id = a.media_file_id\n WHERE a.state IN ('wanted', 'failed')\n ORDER BY f.created_at DESC, a.last_attempt_at, a.language\n LIMIT ?",
"describe": {
"columns": [
{
"name": "media_file_id!: i64",
"ordinal": 0,
"type_info": "Integer",
"origin": {
"Table": {
"table": "subtitle_attempts",
"name": "media_file_id"
}
}
},
{
"name": "language!: String",
"ordinal": 1,
"type_info": "Text",
"origin": {
"Table": {
"table": "subtitle_attempts",
"name": "language"
}
}
},
{
"name": "state!: SubtitleState",
"ordinal": 2,
"type_info": "Text",
"origin": {
"Table": {
"table": "subtitle_attempts",
"name": "state"
}
}
},
{
"name": "attempts!: i64",
"ordinal": 3,
"type_info": "Integer",
"origin": {
"Table": {
"table": "subtitle_attempts",
"name": "attempts"
}
}
},
{
"name": "last_attempt_at",
"ordinal": 4,
"type_info": "Text",
"origin": {
"Table": {
"table": "subtitle_attempts",
"name": "last_attempt_at"
}
}
},
{
"name": "last_failure",
"ordinal": 5,
"type_info": "Text",
"origin": {
"Table": {
"table": "subtitle_attempts",
"name": "last_failure"
}
}
},
{
"name": "media_file_path!: String",
"ordinal": 6,
"type_info": "Text",
"origin": {
"Table": {
"table": "media_files",
"name": "path"
}
}
}
],
"parameters": {
"Right": 1
},
"nullable": [
false,
false,
false,
false,
true,
true,
false
]
},
"hash": "47ad9162ba373dfbdf061898f5f148e65de261b494453804db912a919a78c2c5"
}
.sqlx/query-6236e1be0e268c678271dd72be52454347da6637f0bf3ecb52dda82cddfe459a.json
Generated
+12
View File
@@ -0,0 +1,12 @@
{
"db_name": "SQLite",
"query": "INSERT INTO subtitle_attempts (media_file_id, language)\n VALUES (?, ?)\n ON CONFLICT (media_file_id, language) DO NOTHING",
"describe": {
"columns": [],
"parameters": {
"Right": 2
},
"nullable": []
},
"hash": "6236e1be0e268c678271dd72be52454347da6637f0bf3ecb52dda82cddfe459a"
}
.sqlx/query-62d625c322c8c64b48317cdbf466d74043708d8e12997e0efec23a18eaaaaace.json
Generated
+158
View File
@@ -0,0 +1,158 @@
{
"db_name": "SQLite",
"query": "SELECT id AS \"id!: i64\",\n media_file_id AS \"media_file_id!: i64\",\n language AS \"language!: String\",\n origin AS \"origin!: SubtitleOrigin\",\n provider,\n candidate_id,\n engine,\n forced AS \"forced!: bool\",\n sdh AS \"sdh!: bool\",\n synced AS \"synced!: bool\",\n sync_rejected AS \"sync_rejected!: bool\",\n path\n FROM subtitle_files\n WHERE media_file_id = ?\n ORDER BY language, id",
"describe": {
"columns": [
{
"name": "id!: i64",
"ordinal": 0,
"type_info": "Integer",
"origin": {
"Table": {
"table": "subtitle_files",
"name": "id"
}
}
},
{
"name": "media_file_id!: i64",
"ordinal": 1,
"type_info": "Integer",
"origin": {
"Table": {
"table": "subtitle_files",
"name": "media_file_id"
}
}
},
{
"name": "language!: String",
"ordinal": 2,
"type_info": "Text",
"origin": {
"Table": {
"table": "subtitle_files",
"name": "language"
}
}
},
{
"name": "origin!: SubtitleOrigin",
"ordinal": 3,
"type_info": "Text",
"origin": {
"Table": {
"table": "subtitle_files",
"name": "origin"
}
}
},
{
"name": "provider",
"ordinal": 4,
"type_info": "Text",
"origin": {
"Table": {
"table": "subtitle_files",
"name": "provider"
}
}
},
{
"name": "candidate_id",
"ordinal": 5,
"type_info": "Text",
"origin": {
"Table": {
"table": "subtitle_files",
"name": "candidate_id"
}
}
},
{
"name": "engine",
"ordinal": 6,
"type_info": "Text",
"origin": {
"Table": {
"table": "subtitle_files",
"name": "engine"
}
}
},
{
"name": "forced!: bool",
"ordinal": 7,
"type_info": "Integer",
"origin": {
"Table": {
"table": "subtitle_files",
"name": "forced"
}
}
},
{
"name": "sdh!: bool",
"ordinal": 8,
"type_info": "Integer",
"origin": {
"Table": {
"table": "subtitle_files",
"name": "sdh"
}
}
},
{
"name": "synced!: bool",
"ordinal": 9,
"type_info": "Integer",
"origin": {
"Table": {
"table": "subtitle_files",
"name": "synced"
}
}
},
{
"name": "sync_rejected!: bool",
"ordinal": 10,
"type_info": "Integer",
"origin": {
"Table": {
"table": "subtitle_files",
"name": "sync_rejected"
}
}
},
{
"name": "path",
"ordinal": 11,
"type_info": "Text",
"origin": {
"Table": {
"table": "subtitle_files",
"name": "path"
}
}
}
],
"parameters": {
"Right": 1
},
"nullable": [
false,
false,
false,
false,
true,
true,
true,
false,
false,
false,
false,
true
]
},
"hash": "62d625c322c8c64b48317cdbf466d74043708d8e12997e0efec23a18eaaaaace"
}
.sqlx/query-6eaeee584dc98b28b9605faabab0eecf7839b7755c956731617b475921481701.json
Generated
+26
View File
@@ -0,0 +1,26 @@
{
"db_name": "SQLite",
"query": "INSERT INTO subtitle_files\n (media_file_id, language, origin, provider, candidate_id, engine,\n forced, sdh, synced, sync_rejected, path)\n VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)\n ON CONFLICT DO NOTHING\n RETURNING id AS \"id!: i64\"",
"describe": {
"columns": [
{
"name": "id!: i64",
"ordinal": 0,
"type_info": "Integer",
"origin": {
"Table": {
"table": "subtitle_files",
"name": "id"
}
}
}
],
"parameters": {
"Right": 11
},
"nullable": [
null
]
},
"hash": "6eaeee584dc98b28b9605faabab0eecf7839b7755c956731617b475921481701"
}
.sqlx/query-874079e2ed732468a1f249e6aa026753c78042e164d11eb7cc014c9500211b64.json
Generated
+12
View File
@@ -0,0 +1,12 @@
{
"db_name": "SQLite",
"query": "UPDATE subtitle_files\n SET synced = ?, sync_rejected = ?,\n updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')\n WHERE id = ?",
"describe": {
"columns": [],
"parameters": {
"Right": 3
},
"nullable": []
},
"hash": "874079e2ed732468a1f249e6aa026753c78042e164d11eb7cc014c9500211b64"
}
.sqlx/query-90480a6b6e724d7e4d05239799ad2f586c57deea100320fe433da007396f1e48.json
Generated
+12
View File
@@ -0,0 +1,12 @@
{
"db_name": "SQLite",
"query": "DELETE FROM subtitle_files WHERE id = ?",
"describe": {
"columns": [],
"parameters": {
"Right": 1
},
"nullable": []
},
"hash": "90480a6b6e724d7e4d05239799ad2f586c57deea100320fe433da007396f1e48"
}
.sqlx/query-adf45101bc3ba20d35b59794053029e89d22ada5c7ffecd47187ec34261bc223.json
Generated
+86
View File
@@ -0,0 +1,86 @@
{
"db_name": "SQLite",
"query": "SELECT media_file_id AS \"media_file_id!: i64\",\n language AS \"language!: String\",\n state AS \"state!: SubtitleState\",\n attempts AS \"attempts!: i64\",\n last_attempt_at,\n last_failure\n FROM subtitle_attempts\n WHERE media_file_id = ?\n ORDER BY language",
"describe": {
"columns": [
{
"name": "media_file_id!: i64",
"ordinal": 0,
"type_info": "Integer",
"origin": {
"Table": {
"table": "subtitle_attempts",
"name": "media_file_id"
}
}
},
{
"name": "language!: String",
"ordinal": 1,
"type_info": "Text",
"origin": {
"Table": {
"table": "subtitle_attempts",
"name": "language"
}
}
},
{
"name": "state!: SubtitleState",
"ordinal": 2,
"type_info": "Text",
"origin": {
"Table": {
"table": "subtitle_attempts",
"name": "state"
}
}
},
{
"name": "attempts!: i64",
"ordinal": 3,
"type_info": "Integer",
"origin": {
"Table": {
"table": "subtitle_attempts",
"name": "attempts"
}
}
},
{
"name": "last_attempt_at",
"ordinal": 4,
"type_info": "Text",
"origin": {
"Table": {
"table": "subtitle_attempts",
"name": "last_attempt_at"
}
}
},
{
"name": "last_failure",
"ordinal": 5,
"type_info": "Text",
"origin": {
"Table": {
"table": "subtitle_attempts",
"name": "last_failure"
}
}
}
],
"parameters": {
"Right": 1
},
"nullable": [
false,
false,
false,
false,
true,
true
]
},
"hash": "adf45101bc3ba20d35b59794053029e89d22ada5c7ffecd47187ec34261bc223"
}
.sqlx/query-c5a9e43cfeb315e70bbb603f22acc519d7df3cd0b4d606d0d9056414ac6c503b.json
Generated
+12
View File
@@ -0,0 +1,12 @@
{
"db_name": "SQLite",
"query": "INSERT INTO subtitle_attempts (media_file_id, language, state)\n VALUES (?, ?, 'satisfied')\n ON CONFLICT (media_file_id, language) DO UPDATE\n SET state = 'satisfied',\n last_failure = NULL,\n updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')",
"describe": {
"columns": [],
"parameters": {
"Right": 2
},
"nullable": []
},
"hash": "c5a9e43cfeb315e70bbb603f22acc519d7df3cd0b4d606d0d9056414ac6c503b"
}
.sqlx/query-e5b075d0b7fe5410d3e55d5daf7837d0a9398b1954e713e80a1d82c0a7e172a6.json
Generated
+26
View File
@@ -0,0 +1,26 @@
{
"db_name": "SQLite",
"query": "SELECT id AS \"id!: i64\" FROM subtitle_files WHERE path = ?",
"describe": {
"columns": [
{
"name": "id!: i64",
"ordinal": 0,
"type_info": "Integer",
"origin": {
"Table": {
"table": "subtitle_files",
"name": "id"
}
}
}
],
"parameters": {
"Right": 1
},
"nullable": [
false
]
},
"hash": "e5b075d0b7fe5410d3e55d5daf7837d0a9398b1954e713e80a1d82c0a7e172a6"
}
.sqlx/query-fe15c4a3774e663b03455300d0baec1dadc6cf995c20df803bc4bf49d1cc7c77.json
Generated
+26
View File
@@ -0,0 +1,26 @@
{
"db_name": "SQLite",
"query": "SELECT id AS \"id!: i64\" FROM subtitle_files\n WHERE media_file_id = ? AND language = ? AND origin = 'embedded'\n AND forced = ? AND sdh = ?",
"describe": {
"columns": [
{
"name": "id!: i64",
"ordinal": 0,
"type_info": "Integer",
"origin": {
"Table": {
"table": "subtitle_files",
"name": "id"
}
}
}
],
"parameters": {
"Right": 4
},
"nullable": [
false
]
},
"hash": "fe15c4a3774e663b03455300d0baec1dadc6cf995c20df803bc4bf49d1cc7c77"
}
crates/arr-db/src/lib.rs
+5
View File
@@ -7,9 +7,14 @@ use std::path::Path;
pub mod blacklist;
pub mod policy;
pub mod subtitles;
pub use blacklist::Blacklist;
pub use policy::{MoviePolicy, PolicyColumns, PolicyError, TitlePolicy};
pub use subtitles::{
NewSubtitleFile, PendingSubtitle, SubtitleAttempt, SubtitleFile, SubtitleOrigin, SubtitleState,
SubtitleSync,
};
use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions, SqliteSynchronous};
use sqlx::{migrate::MigrateError, SqlitePool};
crates/arr-db/src/subtitles.rs
+806
View File
@@ -0,0 +1,806 @@
//! Subtitle files and per-language attempts (`DESIGN.md` §15).
//!
//! Two shapes, and the split matters. A **subtitle file** is a fact about
//! what exists: an embedded track, a sidecar extracted from one, a provider
//! fetch, or a machine translation. An **attempt** is a fact about what arr
//! has tried for one wanted language on one file.
//!
//! Satisfaction is read off the files (§15: a language is satisfied when a
//! subtitle in it exists, embedded or sidecar), so the attempt row never
//! becomes a second source of truth for it — it carries only the state the
//! reconcile loop needs to decide whether to spend anything: how many times
//! it has tried, when last, and why that failed.
use sqlx::SqlitePool;
/// Where a subtitle came from (§15).
#[derive(Debug, Clone, Copy, PartialEq, Eq, sqlx::Type)]
#[sqlx(rename_all = "lowercase")]
pub enum SubtitleOrigin {
/// A track inside the video, never extracted. It satisfies a language
/// and has no file on disk — an image-format track can be nothing else.
Embedded,
/// A sidecar written out from a text-format embedded track, which is a
/// legal translation source.
Extracted,
/// Fetched from a named provider.
Provider,
/// Machine translated from an existing subtitle.
Translated,
}
/// What `alass` did to a subtitle (§15).
///
/// It reports no confidence value, so its output is accepted unless it is
/// implausible. The schema stores that as two flags and forbids both at once;
/// the three states they can spell are this enum.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum SubtitleSync {
/// Not run. Embedded tracks are never synced.
#[default]
NotRun,
/// Ran, and its shift was accepted.
Synced,
/// Ran, and its output was implausible, so the unsynced original was
/// kept and this file is flagged.
Rejected,
}
impl SubtitleSync {
/// The `(synced, sync_rejected)` column pair.
const fn columns(self) -> (bool, bool) {
match self {
Self::NotRun => (false, false),
Self::Synced => (true, false),
Self::Rejected => (false, true),
}
}
/// Read the column pair back. Both set is impossible — the schema's
/// `CHECK` rejects it — and reads as rejected, the safer of the two.
const fn from_columns(synced: bool, rejected: bool) -> Self {
match (synced, rejected) {
(_, true) => Self::Rejected,
(true, false) => Self::Synced,
(false, false) => Self::NotRun,
}
}
}
/// Where one wanted language stands for one media file (§15).
#[derive(Debug, Clone, Copy, PartialEq, Eq, sqlx::Type)]
#[sqlx(rename_all = "lowercase")]
pub enum SubtitleState {
/// A gap. Nothing satisfies it yet and the loop may spend on it.
Wanted,
/// A subtitle in this language exists. §15's no-upgrade rule applies:
/// arr stops working on it, including when it was machine made.
Satisfied,
/// The last attempt failed. `last_failure` says why and `attempts` backs
/// the next one off.
Failed,
/// At the daily allowance. A queue state, not an error (§15). The
/// allowance itself is #197.
Capped,
/// No provider has it and there is no text source to translate from, so
/// retrying spends budget for nothing.
Unavailable,
}
/// A subtitle arr knows about.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SubtitleFile {
pub id: i64,
pub media_file_id: i64,
/// As `arr_core::Language` spells it: `pt-PT`, `pt-BR`, `en`.
pub language: String,
pub origin: SubtitleOrigin,
pub provider: Option<String>,
pub candidate_id: Option<String>,
pub engine: Option<String>,
pub forced: bool,
pub sdh: bool,
pub sync: SubtitleSync,
/// `None` only for [`SubtitleOrigin::Embedded`].
pub path: Option<String>,
}
/// A subtitle about to be recorded.
///
/// Built through the four constructors rather than field by field: which of
/// `provider`, `candidate_id`, `engine` and `path` may be set is decided by
/// the origin, and the schema enforces exactly that. Getting it wrong should
/// not compile rather than fail at the insert.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct NewSubtitleFile {
media_file_id: i64,
language: String,
origin: SubtitleOrigin,
provider: Option<String>,
candidate_id: Option<String>,
engine: Option<String>,
forced: bool,
sdh: bool,
sync: SubtitleSync,
path: Option<String>,
}
impl NewSubtitleFile {
fn new(media_file_id: i64, language: &str, origin: SubtitleOrigin) -> Self {
Self {
media_file_id,
language: language.to_owned(),
origin,
provider: None,
candidate_id: None,
engine: None,
forced: false,
sdh: false,
sync: SubtitleSync::NotRun,
path: None,
}
}
/// A track inside the video that was not extracted.
#[must_use]
pub fn embedded(media_file_id: i64, language: &str) -> Self {
Self::new(media_file_id, language, SubtitleOrigin::Embedded)
}
/// A sidecar written out from a text-format embedded track.
#[must_use]
pub fn extracted(media_file_id: i64, language: &str, path: &str) -> Self {
Self {
path: Some(path.to_owned()),
..Self::new(media_file_id, language, SubtitleOrigin::Extracted)
}
}
/// A sidecar fetched from a provider. `candidate_id` is whatever that
/// provider calls the row that was picked, so the same fetch is
/// recognisable later.
#[must_use]
pub fn fetched(
media_file_id: i64,
language: &str,
provider: &str,
candidate_id: &str,
path: &str,
) -> Self {
Self {
provider: Some(provider.to_owned()),
candidate_id: Some(candidate_id.to_owned()),
path: Some(path.to_owned()),
..Self::new(media_file_id, language, SubtitleOrigin::Provider)
}
}
/// A machine translation, carrying the engine that made it.
#[must_use]
pub fn translated(media_file_id: i64, language: &str, engine: &str, path: &str) -> Self {
Self {
engine: Some(engine.to_owned()),
path: Some(path.to_owned()),
..Self::new(media_file_id, language, SubtitleOrigin::Translated)
}
}
/// Foreign lines and signs only. Never satisfies a want (§15).
#[must_use]
pub fn forced(mut self) -> Self {
self.forced = true;
self
}
/// Complete, with sound descriptions. Satisfies, ranked below plain.
#[must_use]
pub fn sdh(mut self) -> Self {
self.sdh = true;
self
}
/// What `alass` made of it. §15 runs it on every fetched and every
/// translated subtitle, so this is set at the same moment the file is.
#[must_use]
pub fn sync(mut self, sync: SubtitleSync) -> Self {
self.sync = sync;
self
}
}
/// Record a subtitle, returning its row id.
///
/// Idempotent on both keys the schema carries: the sidecar path, and the
/// language an embedded track satisfies. A second probe of the same video, or
/// a re-import that finds the same sidecar, converges instead of duplicating.
/// The first row wins — §15 has no upgrade loop, and replacing a subtitle is
/// a manual action that deletes first.
///
/// # Errors
///
/// If the insert fails.
pub async fn record_file(
pool: &SqlitePool,
subtitle: &NewSubtitleFile,
) -> Result<i64, sqlx::Error> {
let origin = subtitle.origin;
let (synced, sync_rejected) = subtitle.sync.columns();
let inserted = sqlx::query_scalar!(
r#"INSERT INTO subtitle_files
(media_file_id, language, origin, provider, candidate_id, engine,
forced, sdh, synced, sync_rejected, path)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT DO NOTHING
RETURNING id AS "id!: i64""#,
subtitle.media_file_id,
subtitle.language,
origin,
subtitle.provider,
subtitle.candidate_id,
subtitle.engine,
subtitle.forced,
subtitle.sdh,
synced,
sync_rejected,
subtitle.path,
)
.fetch_optional(pool)
.await?;
if let Some(id) = inserted {
return Ok(id);
}
// The insert hit one of the two unique keys. Which one is decided by the
// origin, because only an embedded row has no path.
if let Some(path) = subtitle.path.as_deref() {
sqlx::query_scalar!(
r#"SELECT id AS "id!: i64" FROM subtitle_files WHERE path = ?"#,
path
)
.fetch_one(pool)
.await
} else {
sqlx::query_scalar!(
r#"SELECT id AS "id!: i64" FROM subtitle_files
WHERE media_file_id = ? AND language = ? AND origin = 'embedded'
AND forced = ? AND sdh = ?"#,
subtitle.media_file_id,
subtitle.language,
subtitle.forced,
subtitle.sdh,
)
.fetch_one(pool)
.await
}
}
/// Every subtitle known for one media file, ordered by language.
///
/// # Errors
///
/// If the query fails.
pub async fn files_for(
pool: &SqlitePool,
media_file_id: i64,
) -> Result<Vec<SubtitleFile>, sqlx::Error> {
let rows = sqlx::query!(
r#"SELECT id AS "id!: i64",
media_file_id AS "media_file_id!: i64",
language AS "language!: String",
origin AS "origin!: SubtitleOrigin",
provider,
candidate_id,
engine,
forced AS "forced!: bool",
sdh AS "sdh!: bool",
synced AS "synced!: bool",
sync_rejected AS "sync_rejected!: bool",
path
FROM subtitle_files
WHERE media_file_id = ?
ORDER BY language, id"#,
media_file_id
)
.fetch_all(pool)
.await?;
Ok(rows
.into_iter()
.map(|row| SubtitleFile {
id: row.id,
media_file_id: row.media_file_id,
language: row.language,
origin: row.origin,
provider: row.provider,
candidate_id: row.candidate_id,
engine: row.engine,
forced: row.forced,
sdh: row.sdh,
sync: SubtitleSync::from_columns(row.synced, row.sync_rejected),
path: row.path,
})
.collect())
}
/// Record what `alass` did to a subtitle already on disk.
///
/// A rejected sync flags the row (§15); it never deletes it, because the
/// unsynced original is still a subtitle in that language.
///
/// # Errors
///
/// If the update fails.
pub async fn mark_synced(
pool: &SqlitePool,
id: i64,
sync: SubtitleSync,
) -> Result<(), sqlx::Error> {
let (synced, rejected) = sync.columns();
sqlx::query!(
"UPDATE subtitle_files
SET synced = ?, sync_rejected = ?,
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
WHERE id = ?",
synced,
rejected,
id
)
.execute(pool)
.await?;
Ok(())
}
/// Forget a subtitle, for the manual replacement §15 allows. Deleting the row
/// does not touch the file on disk.
///
/// # Errors
///
/// If the delete fails.
pub async fn delete_file(pool: &SqlitePool, id: i64) -> Result<(), sqlx::Error> {
sqlx::query!("DELETE FROM subtitle_files WHERE id = ?", id)
.execute(pool)
.await?;
Ok(())
}
/// One wanted language on one media file.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SubtitleAttempt {
pub media_file_id: i64,
pub language: String,
pub state: SubtitleState,
pub attempts: i64,
pub last_attempt_at: Option<String>,
/// Why the last attempt failed. Cleared by a success.
pub last_failure: Option<String>,
}
/// A gap the reconcile loop may work on, with the video it belongs to.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PendingSubtitle {
pub attempt: SubtitleAttempt,
/// The video the subtitle sits next to (§7.4).
pub media_file_path: String,
}
/// Start tracking a wanted language, if it is not tracked already.
///
/// The wanted set is global (§15), so this is called for every media file the
/// loop sees rather than driven by anything per root. Existing rows are left
/// alone: this must not reset a backoff or undo a `satisfied`.
///
/// # Errors
///
/// If the insert fails.
pub async fn want(
pool: &SqlitePool,
media_file_id: i64,
language: &str,
) -> Result<(), sqlx::Error> {
sqlx::query!(
"INSERT INTO subtitle_attempts (media_file_id, language)
VALUES (?, ?)
ON CONFLICT (media_file_id, language) DO NOTHING",
media_file_id,
language
)
.execute(pool)
.await?;
Ok(())
}
/// Record that an attempt just ran, and where it left the language.
///
/// Bumps the count and the timestamp — that pair is the backoff — and stores
/// `failure` as the reason, which a non-failing state clears.
///
/// # Errors
///
/// If the upsert fails.
pub async fn record_attempt(
pool: &SqlitePool,
media_file_id: i64,
language: &str,
state: SubtitleState,
failure: Option<&str>,
) -> Result<(), sqlx::Error> {
sqlx::query!(
"INSERT INTO subtitle_attempts
(media_file_id, language, state, attempts, last_attempt_at, last_failure)
VALUES (?, ?, ?, 1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'), ?)
ON CONFLICT (media_file_id, language) DO UPDATE
SET state = excluded.state,
attempts = subtitle_attempts.attempts + 1,
last_attempt_at = excluded.last_attempt_at,
last_failure = excluded.last_failure,
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')",
media_file_id,
language,
state,
failure
)
.execute(pool)
.await?;
Ok(())
}
/// Mark a language satisfied without counting an attempt.
///
/// An embedded track satisfies a language with nothing spent, so satisfaction
/// is not always the end of an attempt.
///
/// # Errors
///
/// If the upsert fails.
pub async fn mark_satisfied(
pool: &SqlitePool,
media_file_id: i64,
language: &str,
) -> Result<(), sqlx::Error> {
sqlx::query!(
"INSERT INTO subtitle_attempts (media_file_id, language, state)
VALUES (?, ?, 'satisfied')
ON CONFLICT (media_file_id, language) DO UPDATE
SET state = 'satisfied',
last_failure = NULL,
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')",
media_file_id,
language
)
.execute(pool)
.await?;
Ok(())
}
/// Every tracked language for one media file, ordered by language.
///
/// # Errors
///
/// If the query fails.
pub async fn attempts_for(
pool: &SqlitePool,
media_file_id: i64,
) -> Result<Vec<SubtitleAttempt>, sqlx::Error> {
sqlx::query_as!(
SubtitleAttempt,
r#"SELECT media_file_id AS "media_file_id!: i64",
language AS "language!: String",
state AS "state!: SubtitleState",
attempts AS "attempts!: i64",
last_attempt_at,
last_failure
FROM subtitle_attempts
WHERE media_file_id = ?
ORDER BY language"#,
media_file_id
)
.fetch_all(pool)
.await
}
/// The loop's work list: languages that are not satisfied, newest import
/// first (§15), least recently attempted first within one import.
///
/// Newest first is what lets enabling this on an existing library drain the
/// backlog over days instead of hammering every provider at once. `capped`
/// and `unavailable` rows are excluded — neither is work the loop can do now.
///
/// # Errors
///
/// If the query fails.
pub async fn pending(pool: &SqlitePool, limit: i64) -> Result<Vec<PendingSubtitle>, sqlx::Error> {
let rows = sqlx::query!(
r#"SELECT a.media_file_id AS "media_file_id!: i64",
a.language AS "language!: String",
a.state AS "state!: SubtitleState",
a.attempts AS "attempts!: i64",
a.last_attempt_at,
a.last_failure,
f.path AS "media_file_path!: String"
FROM subtitle_attempts a
JOIN media_files f ON f.id = a.media_file_id
WHERE a.state IN ('wanted', 'failed')
ORDER BY f.created_at DESC, a.last_attempt_at, a.language
LIMIT ?"#,
limit
)
.fetch_all(pool)
.await?;
Ok(rows
.into_iter()
.map(|row| PendingSubtitle {
attempt: SubtitleAttempt {
media_file_id: row.media_file_id,
language: row.language,
state: row.state,
attempts: row.attempts,
last_attempt_at: row.last_attempt_at,
last_failure: row.last_failure,
},
media_file_path: row.media_file_path,
})
.collect())
}
#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
use super::{
attempts_for, delete_file, files_for, mark_satisfied, mark_synced, pending, record_attempt,
record_file, want, NewSubtitleFile, SubtitleOrigin, SubtitleState, SubtitleSync,
};
use crate::Db;
use sqlx::SqlitePool;
async fn database() -> (Db, tempfile::TempDir) {
let dir = tempfile::tempdir().unwrap();
let database = Db::connect(dir.path().join("test.db")).await.unwrap();
database.migrate().await.unwrap();
(database, dir)
}
/// A media file owned by a movie that does not exist: `owner_id` is
/// polymorphic and carries no foreign key, and none of this cares.
async fn media_file(pool: &SqlitePool, path: &str, imported_at: &str) -> i64 {
sqlx::query_scalar(
"INSERT INTO media_files (owner_kind, owner_id, path, size, created_at)
VALUES ('movie', 1, ?, 1, ?) RETURNING id",
)
.bind(path)
.bind(imported_at)
.fetch_one(pool)
.await
.unwrap()
}
#[tokio::test]
async fn a_second_probe_re_records_the_same_embedded_track() {
let (db, _dir) = database().await;
let file = media_file(db.pool(), "/m/a.mkv", "2026-01-01T00:00:00.000Z").await;
let first = record_file(db.pool(), &NewSubtitleFile::embedded(file, "en"))
.await
.unwrap();
let again = record_file(db.pool(), &NewSubtitleFile::embedded(file, "en"))
.await
.unwrap();
assert_eq!(
first, again,
"one row per language an embedded track covers"
);
// A forced track is a different fact about the same language, and
// §15 says it never satisfies one, so it is its own row.
let forced = record_file(db.pool(), &NewSubtitleFile::embedded(file, "en").forced())
.await
.unwrap();
assert_ne!(first, forced);
let rows = files_for(db.pool(), file).await.unwrap();
assert_eq!(rows.len(), 2);
assert!(rows.iter().all(|row| row.path.is_none()));
assert!(rows
.iter()
.all(|row| row.origin == SubtitleOrigin::Embedded));
}
#[tokio::test]
async fn a_sidecar_is_keyed_on_its_path() {
let (db, _dir) = database().await;
let file = media_file(db.pool(), "/m/b.mkv", "2026-01-01T00:00:00.000Z").await;
let fetched =
NewSubtitleFile::fetched(file, "pt-PT", "opensubtitles", "42", "/m/b.pt-PT.srt");
let first = record_file(db.pool(), &fetched).await.unwrap();
let again = record_file(db.pool(), &fetched).await.unwrap();
assert_eq!(first, again);
let translated = NewSubtitleFile::translated(file, "pt-PT", "deepl", "/m/b.pt-PT.mt.srt")
.sync(SubtitleSync::Synced);
record_file(db.pool(), &translated).await.unwrap();
let rows = files_for(db.pool(), file).await.unwrap();
assert_eq!(rows.len(), 2);
let provider = rows
.iter()
.find(|row| row.origin == SubtitleOrigin::Provider)
.unwrap();
assert_eq!(provider.provider.as_deref(), Some("opensubtitles"));
assert_eq!(provider.candidate_id.as_deref(), Some("42"));
assert!(provider.engine.is_none());
let machine = rows
.iter()
.find(|row| row.origin == SubtitleOrigin::Translated)
.unwrap();
assert_eq!(machine.engine.as_deref(), Some("deepl"));
assert_eq!(machine.sync, SubtitleSync::Synced);
}
#[tokio::test]
async fn the_schema_ties_every_optional_column_to_the_origin() {
let (db, _dir) = database().await;
let file = media_file(db.pool(), "/m/c.mkv", "2026-01-01T00:00:00.000Z").await;
// An embedded track has no file; everything else is a sidecar.
let with_path = sqlx::query(
"INSERT INTO subtitle_files (media_file_id, language, origin, path)
VALUES (?, 'en', 'embedded', '/m/c.en.srt')",
)
.bind(file)
.execute(db.pool())
.await;
assert!(with_path.is_err());
// A provider fetch names its provider; a translation names its engine.
let nameless = sqlx::query(
"INSERT INTO subtitle_files (media_file_id, language, origin, path)
VALUES (?, 'en', 'provider', '/m/c.en.srt')",
)
.bind(file)
.execute(db.pool())
.await;
assert!(nameless.is_err());
// `alass` either shifted it or was rejected, never both.
let both = sqlx::query(
"INSERT INTO subtitle_files
(media_file_id, language, origin, engine, path, synced, sync_rejected)
VALUES (?, 'en', 'translated', 'deepl', '/m/c.en.mt.srt', 1, 1)",
)
.bind(file)
.execute(db.pool())
.await;
assert!(both.is_err());
}
#[tokio::test]
async fn a_rejected_sync_flags_the_row_rather_than_removing_it() {
let (db, _dir) = database().await;
let file = media_file(db.pool(), "/m/d.mkv", "2026-01-01T00:00:00.000Z").await;
let id = record_file(
db.pool(),
&NewSubtitleFile::fetched(file, "en", "podnapisi", "7", "/m/d.en.srt"),
)
.await
.unwrap();
mark_synced(db.pool(), id, SubtitleSync::Rejected)
.await
.unwrap();
let row = files_for(db.pool(), file).await.unwrap().remove(0);
assert_eq!(row.sync, SubtitleSync::Rejected);
delete_file(db.pool(), id).await.unwrap();
assert!(files_for(db.pool(), file).await.unwrap().is_empty());
}
#[tokio::test]
async fn an_attempt_counts_and_a_success_clears_the_reason() {
let (db, _dir) = database().await;
let file = media_file(db.pool(), "/m/e.mkv", "2026-01-01T00:00:00.000Z").await;
want(db.pool(), file, "pt-PT").await.unwrap();
let fresh = attempts_for(db.pool(), file).await.unwrap().remove(0);
assert_eq!(fresh.state, SubtitleState::Wanted);
assert_eq!(fresh.attempts, 0);
assert!(fresh.last_attempt_at.is_none());
for _ in 0..2 {
record_attempt(
db.pool(),
file,
"pt-PT",
SubtitleState::Failed,
Some("provider_unreachable"),
)
.await
.unwrap();
}
let failed = attempts_for(db.pool(), file).await.unwrap().remove(0);
assert_eq!(
failed.attempts, 2,
"the count is what backs the provider off"
);
assert_eq!(failed.state, SubtitleState::Failed);
assert_eq!(failed.last_failure.as_deref(), Some("provider_unreachable"));
assert!(failed.last_attempt_at.is_some());
mark_satisfied(db.pool(), file, "pt-PT").await.unwrap();
let done = attempts_for(db.pool(), file).await.unwrap().remove(0);
assert_eq!(done.state, SubtitleState::Satisfied);
assert!(done.last_failure.is_none());
assert_eq!(done.attempts, 2, "an embedded track spends nothing");
}
#[tokio::test]
async fn wanting_a_tracked_language_again_resets_nothing() {
let (db, _dir) = database().await;
let file = media_file(db.pool(), "/m/f.mkv", "2026-01-01T00:00:00.000Z").await;
want(db.pool(), file, "en").await.unwrap();
record_attempt(
db.pool(),
file,
"en",
SubtitleState::Failed,
Some("rate_limited"),
)
.await
.unwrap();
want(db.pool(), file, "en").await.unwrap();
let row = attempts_for(db.pool(), file).await.unwrap().remove(0);
assert_eq!(row.attempts, 1);
assert_eq!(row.state, SubtitleState::Failed);
}
#[tokio::test]
async fn the_work_list_is_newest_import_first_and_skips_settled_languages() {
let (db, _dir) = database().await;
let old = media_file(db.pool(), "/m/old.mkv", "2026-01-01T00:00:00.000Z").await;
let new = media_file(db.pool(), "/m/new.mkv", "2026-06-01T00:00:00.000Z").await;
for file in [old, new] {
want(db.pool(), file, "pt-PT").await.unwrap();
want(db.pool(), file, "en").await.unwrap();
}
mark_satisfied(db.pool(), new, "en").await.unwrap();
record_attempt(db.pool(), old, "en", SubtitleState::Capped, None)
.await
.unwrap();
let work = pending(db.pool(), 10).await.unwrap();
let seen: Vec<_> = work
.iter()
.map(|row| (row.media_file_path.as_str(), row.attempt.language.as_str()))
.collect();
assert_eq!(
seen,
vec![("/m/new.mkv", "pt-PT"), ("/m/old.mkv", "pt-PT")],
"satisfied and capped languages are not work"
);
assert_eq!(pending(db.pool(), 1).await.unwrap().len(), 1);
}
#[tokio::test]
async fn deleting_a_media_file_takes_its_subtitles_with_it() {
let (db, _dir) = database().await;
let file = media_file(db.pool(), "/m/g.mkv", "2026-01-01T00:00:00.000Z").await;
record_file(db.pool(), &NewSubtitleFile::embedded(file, "en"))
.await
.unwrap();
want(db.pool(), file, "en").await.unwrap();
sqlx::query("DELETE FROM media_files WHERE id = ?")
.bind(file)
.execute(db.pool())
.await
.unwrap();
assert!(files_for(db.pool(), file).await.unwrap().is_empty());
assert!(attempts_for(db.pool(), file).await.unwrap().is_empty());
}
}