diff --git a/.sqlx/query-35ace0417f613340f82522549d70d0dffeb680afb57511cb17fa15b3a571b974.json b/.sqlx/query-35ace0417f613340f82522549d70d0dffeb680afb57511cb17fa15b3a571b974.json new file mode 100644 index 0000000..25fe714 --- /dev/null +++ b/.sqlx/query-35ace0417f613340f82522549d70d0dffeb680afb57511cb17fa15b3a571b974.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "INSERT INTO subtitle_attempts (media_file_id, language)\n SELECT id, ? FROM media_files WHERE probed IS NOT NULL\n ON CONFLICT (media_file_id, language) DO NOTHING", + "describe": { + "columns": [], + "parameters": { + "Right": 1 + }, + "nullable": [] + }, + "hash": "35ace0417f613340f82522549d70d0dffeb680afb57511cb17fa15b3a571b974" +} diff --git a/.sqlx/query-af10ccd655f39235f525a531e38850b2e60ff713a038ee1e7c04c616b4613b52.json b/.sqlx/query-af10ccd655f39235f525a531e38850b2e60ff713a038ee1e7c04c616b4613b52.json new file mode 100644 index 0000000..52b1bab --- /dev/null +++ b/.sqlx/query-af10ccd655f39235f525a531e38850b2e60ff713a038ee1e7c04c616b4613b52.json @@ -0,0 +1,26 @@ +{ + "db_name": "SQLite", + "query": "SELECT probed FROM media_files WHERE id = ?", + "describe": { + "columns": [ + { + "name": "probed", + "ordinal": 0, + "type_info": "Text", + "origin": { + "Table": { + "table": "media_files", + "name": "probed" + } + } + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + true + ] + }, + "hash": "af10ccd655f39235f525a531e38850b2e60ff713a038ee1e7c04c616b4613b52" +} diff --git a/.sqlx/query-caa03a8daf9064de5c4724fcfecf70cc13f9a188468395bf5558aa3068c8a4e4.json b/.sqlx/query-caa03a8daf9064de5c4724fcfecf70cc13f9a188468395bf5558aa3068c8a4e4.json new file mode 100644 index 0000000..fe59ef1 --- /dev/null +++ b/.sqlx/query-caa03a8daf9064de5c4724fcfecf70cc13f9a188468395bf5558aa3068c8a4e4.json @@ -0,0 +1,50 @@ +{ + "db_name": "SQLite", + "query": "SELECT wanted_languages AS \"wanted_languages!: String\",\n providers_enabled AS \"providers_enabled!: String\",\n translation_engine AS \"translation_engine: String\"\n FROM subtitle_settings WHERE id = 1", + "describe": { + "columns": [ + { + "name": "wanted_languages!: String", + "ordinal": 0, + "type_info": "Text", + "origin": { + "Table": { + "table": "subtitle_settings", + "name": "wanted_languages" + } + } + }, + { + "name": "providers_enabled!: String", + "ordinal": 1, + "type_info": "Text", + "origin": { + "Table": { + "table": "subtitle_settings", + "name": "providers_enabled" + } + } + }, + { + "name": "translation_engine: String", + "ordinal": 2, + "type_info": "Text", + "origin": { + "Table": { + "table": "subtitle_settings", + "name": "translation_engine" + } + } + } + ], + "parameters": { + "Right": 0 + }, + "nullable": [ + false, + false, + true + ] + }, + "hash": "caa03a8daf9064de5c4724fcfecf70cc13f9a188468395bf5558aa3068c8a4e4" +} diff --git a/crates/arr-daemon/src/main.rs b/crates/arr-daemon/src/main.rs index 82edeaa..f8c8b9d 100644 --- a/crates/arr-daemon/src/main.rs +++ b/crates/arr-daemon/src/main.rs @@ -13,6 +13,7 @@ mod reaper; pub mod reconcile; mod rss; mod series_refresh; +mod subtitles; mod tv_grab; mod web; @@ -33,6 +34,7 @@ use reaper::ReaperAction; use reconcile::{ReconcileLoop, Tick}; use rss::RssAction; use series_refresh::SeriesRefreshAction; +use subtitles::SubtitleAction; use tower_http::trace::TraceLayer; use tv_grab::TvGrabAction; @@ -331,6 +333,9 @@ fn reconcile_loop( ), ); + // §15: subtitle gaps are reconciled from the same rows the API writes. + reconcile = reconcile.register(Tick::Reconcile, subtitle_action(config, notifier)?); + // §9.5 *needs a decision* and *broken* both go to the operator alone; // without a topic configured there is nowhere to send them. if let Some(operator_topic) = &config.ntfy_operator_topic { @@ -474,6 +479,30 @@ fn jellyfin_client(config: &Config) -> Result Result { + let action = SubtitleAction::new( + subtitle_providers( + config.opensubtitles_api_key.clone(), + config.opensubtitles_username.clone(), + config.opensubtitles_password.clone(), + ), + Vec::new(), + arr_subs::Syncer::new().with_binary(config.alass_path.clone()), + arr_probe::Extractor::new().with_binary(config.ffmpeg_path.clone()), + jellyfin_client(config)?, + ); + Ok(match &config.ntfy_operator_topic { + Some(topic) => action.with_notifier(notifier.clone(), topic.clone()), + None => action, + }) +} + /// The subtitle providers this deployment can reach (DESIGN.md §15). /// /// Credentials are bootstrap config and never reach the database (§10), so diff --git a/crates/arr-daemon/src/subtitles.rs b/crates/arr-daemon/src/subtitles.rs new file mode 100644 index 0000000..72d88be --- /dev/null +++ b/crates/arr-daemon/src/subtitles.rs @@ -0,0 +1,1621 @@ +//! Closing subtitle gaps on the reconcile tick. DESIGN.md §8, §15. +//! +//! The rows are the work list: `subtitle_attempts` tracks one row per +//! (media file, wanted language), and every row that is not settled is a +//! gap. Per gap, in order: an embedded track satisfies it for free; then the +//! enabled providers are searched and the ranked winner fetched, synced and +//! written; then — with no waiting window — a machine translation from an +//! existing subtitle, extracting a text-format embedded track first when +//! that is the only source; and when none of that is possible the reason is +//! recorded and the row left for the missing-subtitles queue (#202). +//! +//! Fetch and translate outlive the reconcile lane's action timeout — +//! `alass` alone may take minutes — so a gap is closed by a detached task, +//! the same pattern import uses for probes. The task records every outcome +//! in domain rows before it exits, so a crash mid-close converges on the +//! next tick (§8). An in-memory in-flight set stops one gap being worked +//! twice; it dies with the process, and so do the tasks it guarded. +//! +//! §5.4's rule applies unchanged: once a language is satisfied — by a +//! machine translation too — the loop stops. There is no re-search and no +//! replacement; that is a manual action through the API. + +use std::collections::{BTreeSet, HashSet}; +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use arr_core::subs::{rank, SubtitleTarget, SubtitleVerdict}; +use arr_core::{layout, Language}; +use arr_db::subtitles::{self as db, PendingSubtitle, SubtitleOrigin, SubtitleState}; +use arr_db::Db; +use arr_probe::Extractor; +use arr_subs::{Backend, Candidate, MediaFile, MediaRef, Provider, SearchRequest, Syncer}; +use tokio::sync::Mutex; + +use crate::grab::backoff_elapsed; +use crate::notify::Notifier; +use crate::reconcile::{Action, ActionFuture, Outcome}; +use arr_api::jellyfin::JellyfinClient; + +/// How many pending rows one tick considers. Larger than the in-flight cap +/// so a run of backed-off newest imports cannot starve older gaps. +const PENDING_LIMIT: i64 = 100; + +/// How many gaps may be closing at once. Each holds a provider download, an +/// `alass` run or a translation in flight; more buys nothing but rate limits. +const MAX_IN_FLIGHT: usize = 4; + +/// A failure during one subtitle tick. +#[derive(Debug, thiserror::Error)] +pub enum SubtitleError { + #[error("database: {0}")] + Database(#[from] sqlx::Error), + #[error("subtitle settings: {0}")] + Settings(String), +} + +/// The runtime-editable half of §15's configuration, read once per tick. +#[derive(Debug, Clone)] +struct Settings { + wanted: Vec, + providers_enabled: BTreeSet, + translation_engine: Option, +} + +async fn load_settings(database: &Db) -> Result { + let row = sqlx::query!( + r#"SELECT wanted_languages AS "wanted_languages!: String", + providers_enabled AS "providers_enabled!: String", + translation_engine AS "translation_engine: String" + FROM subtitle_settings WHERE id = 1"# + ) + .fetch_one(database.pool()) + .await?; + let wanted: Vec = serde_json::from_str(&row.wanted_languages) + .map_err(|error| SubtitleError::Settings(error.to_string()))?; + let providers_enabled: BTreeSet = serde_json::from_str(&row.providers_enabled) + .map_err(|error| SubtitleError::Settings(error.to_string()))?; + Ok(Settings { + wanted, + providers_enabled, + translation_engine: row.translation_engine, + }) +} + +/// Whether a subtitle in `have` settles a want for `wanted` (§15: pt-PT +/// preferred, pt-BR always fine). An unverified Portuguese embedded track +/// counts too: it satisfies viewing even when no signal resolved the +/// flavour. +fn satisfies(wanted: &str, have: &str) -> bool { + have == wanted || (wanted == "pt-PT" && matches!(have, "pt-BR" | "por-unverified")) +} + +/// The languages worth fetching when `wanted` is the gap, preferred first. +fn acceptable(wanted: &Language) -> Vec { + match wanted { + Language::PortuguesePortugal => { + vec![Language::PortuguesePortugal, Language::PortugueseBrazil] + } + other => vec![other.clone()], + } +} + +/// One embedded subtitle track as `media_files.probed` records it. +/// +/// Parsed leniently: rows probed before #189 carry no codec or dispositions, +/// and an old shape must read as "a track in this language, nothing else +/// known" rather than fail the whole file. +#[derive(Debug, Clone)] +struct EmbeddedTrack { + /// Position among the file's subtitle streams — the index `ffmpeg` + /// extraction maps. + index: usize, + language: String, + codec: Option, + forced: bool, + sdh: bool, +} + +impl EmbeddedTrack { + /// §15: only text formats extract and may feed a translator. + fn is_text(&self) -> bool { + matches!(self.codec.as_deref(), Some("subrip" | "ass" | "mov_text")) + } +} + +fn embedded_tracks(probed: &str) -> Vec { + let Ok(value) = serde_json::from_str::(probed) else { + return Vec::new(); + }; + let Some(tracks) = value.get("sub_tracks").and_then(|tracks| tracks.as_array()) else { + return Vec::new(); + }; + tracks + .iter() + .enumerate() + .filter_map(|(index, track)| { + let language = match track { + serde_json::Value::String(language) => language.clone(), + serde_json::Value::Object(fields) => fields.get("language")?.as_str()?.to_owned(), + _ => return None, + }; + Some(EmbeddedTrack { + index, + language, + codec: track + .get("codec") + .and_then(|codec| codec.as_str()) + .map(str::to_owned), + forced: track + .get("forced") + .and_then(serde_json::Value::as_bool) + .unwrap_or(false), + sdh: track + .get("sdh") + .and_then(serde_json::Value::as_bool) + .unwrap_or(false), + }) + }) + .collect() +} + +/// Everything a close needs to know about the file it targets. The same +/// facts the manual API loads: the ranking inputs of §15. +#[derive(Debug, Clone)] +struct Target { + media_file_id: i64, + path: PathBuf, + size: u64, + media: MediaRef, + release_name: Option, + release_group: Option, + source: Option, +} + +impl Target { + fn search_request(&self, languages: Vec) -> SearchRequest { + SearchRequest { + file: MediaFile { + path: self.path.clone(), + size: self.size, + release_name: self.release_name.clone(), + media: self.media, + }, + languages, + } + } + + /// Where a sidecar for `language` belongs: next to the video, named by + /// §15's rule. `None` when the media path has no name or folder to sit + /// in, which no §7.4 layout produces. + fn sidecar(&self, language: &Language, machine_translated: bool) -> Option { + let name = self.path.file_name()?.to_str()?; + let parent = self.path.parent()?; + Some(parent.join(layout::subtitle_name(name, language, machine_translated))) + } +} + +async fn load_target(database: &Db, media_file_id: i64) -> Result, SubtitleError> { + let Some(file) = sqlx::query!( + r#"SELECT id AS "id!: i64", + path AS "path!: String", + size AS "size!: i64", + owner_kind AS "owner_kind!: String", + owner_id AS "owner_id!: i64" + FROM media_files WHERE id = ?"#, + media_file_id + ) + .fetch_optional(database.pool()) + .await? + else { + return Ok(None); + }; + + let Some(media) = media_ref(database, &file.owner_kind, file.owner_id).await? else { + return Ok(None); + }; + let release = imported_release(database, &file.owner_kind, file.owner_id).await?; + let (release_name, claims) = match release { + Some((name, parsed)) => { + let claims: Option = serde_json::from_value(parsed).ok(); + (Some(name), claims) + } + None => (None, None), + }; + + Ok(Some(Target { + media_file_id: file.id, + path: PathBuf::from(file.path), + size: u64::try_from(file.size).unwrap_or(0), + media, + release_name, + release_group: claims.as_ref().and_then(|claims| claims.group.clone()), + source: claims + .as_ref() + .and_then(|claims| claims.source) + .map(Into::into), + })) +} + +/// The TMDB coordinates providers search by (§15). +async fn media_ref( + database: &Db, + owner_kind: &str, + owner_id: i64, +) -> Result, SubtitleError> { + if owner_kind == "movie" { + let tmdb_id = sqlx::query_scalar!( + r#"SELECT tmdb_id AS "tmdb_id!: i64" FROM movies WHERE id = ?"#, + owner_id + ) + .fetch_optional(database.pool()) + .await?; + return Ok(tmdb_id.map(|tmdb_id| MediaRef::Movie { + tmdb_id: u64::try_from(tmdb_id).unwrap_or(0), + })); + } + + let row = sqlx::query!( + r#"SELECT sr.tmdb_id AS "tmdb_id!: i64", + s.number AS "season!: i64", + e.number AS "episode!: i64" + FROM episodes e + JOIN seasons s ON s.id = e.season_id + JOIN series sr ON sr.id = s.series_id + WHERE e.id = ?"#, + owner_id + ) + .fetch_optional(database.pool()) + .await?; + + Ok(row.map(|row| MediaRef::Episode { + tmdb_id: u64::try_from(row.tmdb_id).unwrap_or(0), + season: u16::try_from(row.season).unwrap_or(0), + episode: u16::try_from(row.episode).unwrap_or(0), + })) +} + +/// The release a file was imported under, best effort — a missing name costs +/// the exact-name tier in ranking and nothing else. An episode that arrived +/// inside a season pack has no grab of its own, so the season's grab is the +/// fallback. +async fn imported_release( + database: &Db, + owner_kind: &str, + owner_id: i64, +) -> Result, SubtitleError> { + let own = sqlx::query!( + r#"SELECT r.name AS "name!: String", + r.parsed AS "parsed!: serde_json::Value" + FROM grabs g JOIN releases r ON r.id = g.release_id + WHERE g.target_kind = ? AND g.target_id = ? + ORDER BY g.imported_at DESC, g.id DESC LIMIT 1"#, + owner_kind, + owner_id + ) + .fetch_optional(database.pool()) + .await?; + if let Some(row) = own { + return Ok(Some((row.name, row.parsed))); + } + if owner_kind != "episode" { + return Ok(None); + } + + let pack = sqlx::query!( + r#"SELECT r.name AS "name!: String", + r.parsed AS "parsed!: serde_json::Value" + FROM episodes e + JOIN grabs g ON g.target_kind = 'season' AND g.target_id = e.season_id + JOIN releases r ON r.id = g.release_id + WHERE e.id = ? + ORDER BY g.imported_at DESC, g.id DESC LIMIT 1"#, + owner_id + ) + .fetch_optional(database.pool()) + .await?; + Ok(pack.map(|row| (row.name, row.parsed))) +} + +/// The file's own `moviehash`, when it can be computed. Blocking IO, so off +/// the async worker; a search that cannot hash still ranks. +async fn moviehash(path: &Path, size: u64) -> Option { + let owned = path.to_path_buf(); + let computed = tokio::task::spawn_blocking(move || arr_subs::moviehash(&owned, size)) + .await + .ok()?; + match computed { + Ok(hash) => hash, + Err(error) => { + tracing::debug!(path = %path.display(), %error, "moviehash not computed"); + None + } + } +} + +/// Write a sidecar whole or not at all, so Jellyfin never reads a half file. +async fn write_sidecar(destination: &Path, text: &str) -> Result<(), String> { + let temp = destination.with_extension("srt.partial"); + tokio::fs::write(&temp, text) + .await + .map_err(|error| format!("{}: {error}", temp.display()))?; + if let Err(error) = tokio::fs::rename(&temp, destination).await { + let _ = tokio::fs::remove_file(&temp).await; + return Err(format!("{}: {error}", destination.display())); + } + Ok(()) +} + +const fn db_sync_state(state: arr_subs::SyncState) -> arr_db::SubtitleSync { + match state { + arr_subs::SyncState::NotRun => arr_db::SubtitleSync::NotRun, + arr_subs::SyncState::Synced => arr_db::SubtitleSync::Synced, + arr_subs::SyncState::Rejected => arr_db::SubtitleSync::Rejected, + } +} + +/// The shared clients one detached close carries. Cheap to clone: every +/// field pools internally or is behind an [`Arc`]. +#[derive(Debug, Clone)] +struct Worker { + providers: Arc>>, + backends: Arc>>, + syncer: Syncer, + extractor: Extractor, + jellyfin: JellyfinClient, + notifier: Option<(Notifier, String)>, + /// Which providers and translators are currently notified as broken + /// (§9.5, edge-triggered like `BrokenAction`). Transient: a restart + /// re-notifies whatever still fails. + broken: Arc>>, + /// Gaps a detached task is closing right now, keyed by + /// (media file, wanted language). + in_flight: Arc>>, +} + +/// What one close settled to, for the log. +enum Closed { + Fetched { + provider: String, + language: String, + }, + Translated { + engine: String, + }, + Recorded { + state: SubtitleState, + reason: String, + }, +} + +impl Worker { + fn provider(&self, id: &str) -> Option> { + self.providers + .iter() + .find(|provider| provider.id().as_str() == id) + .map(Arc::clone) + } + + /// §9.5: a provider or translator being unreachable folds into the one + /// existing *broken* notification to the operator. Edge-triggered — once + /// per outage, re-armed by the first success. + async fn notify_broken(&self, name: &str) { + let mut broken = self.broken.lock().await; + if !broken.insert(name.to_owned()) { + return; + } + drop(broken); + let Some((notifier, topic)) = &self.notifier else { + return; + }; + if let Err(error) = notifier + .send(topic, "arr: broken", &format!("{name} is unreachable")) + .await + { + tracing::warn!(%error, name, "broken notification failed"); + } + } + + async fn clear_broken(&self, name: &str) { + self.broken.lock().await.remove(name); + } + + /// Close one gap end to end and record where it landed. Every exit + /// writes a domain row, so the next tick sees the truth whatever + /// happened here. + async fn close_and_log(&self, database: Db, gap: PendingSubtitle, settings: Settings) { + let key = (gap.attempt.media_file_id, gap.attempt.language.clone()); + match self.close(&database, &gap, &settings).await { + Ok(Closed::Fetched { provider, language }) => tracing::info!( + media_file_id = gap.attempt.media_file_id, + wanted = %gap.attempt.language, + %provider, + %language, + "subtitle fetched" + ), + Ok(Closed::Translated { engine }) => tracing::info!( + media_file_id = gap.attempt.media_file_id, + wanted = %gap.attempt.language, + %engine, + "subtitle machine translated" + ), + Ok(Closed::Recorded { state, reason }) => tracing::info!( + media_file_id = gap.attempt.media_file_id, + wanted = %gap.attempt.language, + state = ?state, + %reason, + "subtitle gap left open" + ), + Err(error) => tracing::error!( + media_file_id = gap.attempt.media_file_id, + wanted = %gap.attempt.language, + %error, + "subtitle close failed" + ), + } + self.in_flight.lock().await.remove(&key); + } + + async fn close( + &self, + database: &Db, + gap: &PendingSubtitle, + settings: &Settings, + ) -> Result { + let media_file_id = gap.attempt.media_file_id; + let wanted_tag = gap.attempt.language.as_str(); + let Some(target) = load_target(database, media_file_id).await? else { + // The file vanished between the work list and now; the cascade + // will take the attempt row with it. + return Ok(Closed::Recorded { + state: SubtitleState::Failed, + reason: "media file no longer exists".to_owned(), + }); + }; + let wanted = arr_db::policy::language(wanted_tag); + let languages = acceptable(&wanted); + + let providers: Vec> = self + .providers + .iter() + .filter(|provider| settings.providers_enabled.contains(provider.id().as_str())) + .map(Arc::clone) + .collect(); + + let request = target.search_request(languages.clone()); + let mut offered: Vec = Vec::new(); + let mut errors: Vec = Vec::new(); + for provider in &providers { + let name = format!("subtitle provider {}", provider.id()); + match provider.search(&request).await { + Ok(candidates) => { + self.clear_broken(&name).await; + offered.extend( + candidates + .into_iter() + .filter(|candidate| languages.contains(&candidate.language)), + ); + } + Err(error) => { + if matches!(error, arr_subs::Error::Transport { .. }) { + self.notify_broken(&name).await; + } + errors.push(error); + } + } + } + + let hash = moviehash(&target.path, target.size).await; + if let Some(winner) = choose(&target, hash.as_deref(), &offered, &languages) { + return self.fetch(database, &target, wanted_tag, winner).await; + } + + // A provider error is not an empty answer (§15): the candidate may + // exist where nothing could be asked, so translating now would + // satisfy the language forever with a worse subtitle. Back off and + // retry instead. + if !errors.is_empty() { + let state = if errors + .iter() + .all(|error| matches!(error, arr_subs::Error::RateLimited { .. })) + { + SubtitleState::Capped + } else { + SubtitleState::Failed + }; + let reason = errors + .iter() + .map(ToString::to_string) + .collect::>() + .join("; "); + db::record_attempt( + database.pool(), + media_file_id, + wanted_tag, + state, + Some(&reason), + ) + .await?; + return Ok(Closed::Recorded { state, reason }); + } + + self.translate(database, &target, &wanted, wanted_tag, settings) + .await + } + + /// Download the ranked winner, sync it, write it, record it (§15). + async fn fetch( + &self, + database: &Db, + target: &Target, + wanted_tag: &str, + winner: Candidate, + ) -> Result { + let media_file_id = target.media_file_id; + let record_failure = |reason: String| async move { + db::record_attempt( + database.pool(), + media_file_id, + wanted_tag, + SubtitleState::Failed, + Some(&reason), + ) + .await?; + Ok(Closed::Recorded { + state: SubtitleState::Failed, + reason, + }) + }; + + let provider_name = winner.provider.to_string(); + let Some(provider) = self.provider(&provider_name) else { + return record_failure(format!("provider {provider_name} vanished mid-close")).await; + }; + let broken_name = format!("subtitle provider {provider_name}"); + let fetched = match provider.download(&winner.id).await { + Ok(fetched) => fetched, + Err(error) => { + if matches!(error, arr_subs::Error::Transport { .. }) { + self.notify_broken(&broken_name).await; + } + let state = if matches!(error, arr_subs::Error::RateLimited { .. }) { + SubtitleState::Capped + } else { + SubtitleState::Failed + }; + let reason = error.to_string(); + db::record_attempt( + database.pool(), + media_file_id, + wanted_tag, + state, + Some(&reason), + ) + .await?; + return Ok(Closed::Recorded { state, reason }); + } + }; + self.clear_broken(&broken_name).await; + + let text = match fetched.to_srt() { + Ok(text) => text, + Err(error) => return record_failure(error.to_string()).await, + }; + let Some(destination) = target.sidecar(&winner.language, false) else { + return record_failure(format!( + "media file path {} has no folder", + target.path.display() + )) + .await; + }; + if let Err(error) = write_sidecar(&destination, &text).await { + return record_failure(error).await; + } + + let sync = self.syncer.settle(&target.path, &destination).await; + if let Some(synced) = &sync.content { + if let Err(error) = write_sidecar(&destination, synced).await { + return record_failure(error).await; + } + } + + let language = winner.language.to_string(); + let mut record = arr_db::NewSubtitleFile::fetched( + media_file_id, + &language, + &provider_name, + winner.id.as_str(), + &destination.to_string_lossy(), + ) + .sync(db_sync_state(sync.state)); + if winner.sdh { + record = record.sdh(); + } + db::record_file(database.pool(), &record).await?; + db::mark_satisfied(database.pool(), media_file_id, wanted_tag).await?; + self.refresh_jellyfin().await; + Ok(Closed::Fetched { + provider: provider_name, + language, + }) + } + + /// §15's immediate translation: no provider has the language, so make + /// one from an existing subtitle — a fetched sidecar, or a text-format + /// embedded track extracted first. + async fn translate( + &self, + database: &Db, + target: &Target, + wanted: &Language, + wanted_tag: &str, + settings: &Settings, + ) -> Result { + let media_file_id = target.media_file_id; + let record = |state: SubtitleState, reason: String| async move { + db::record_attempt( + database.pool(), + media_file_id, + wanted_tag, + state, + Some(&reason), + ) + .await?; + Ok(Closed::Recorded { state, reason }) + }; + + let Some((source_path, source_language)) = + self.translation_source(database, target).await? + else { + return record( + SubtitleState::Unavailable, + format!( + "no provider has {wanted_tag} and there is no text source to translate from" + ), + ) + .await; + }; + + let Some(engine) = settings.translation_engine.clone() else { + return record( + SubtitleState::Failed, + "no translation engine is configured".to_owned(), + ) + .await; + }; + let Some(backend) = self + .backends + .iter() + .find(|backend| backend.id().as_str() == engine) + .map(Arc::clone) + else { + return record( + SubtitleState::Failed, + format!("translation engine '{engine}' is not compiled into this binary"), + ) + .await; + }; + + let raw = match tokio::fs::read_to_string(&source_path).await { + Ok(raw) => raw, + Err(error) => { + return record(SubtitleState::Failed, format!("{source_path}: {error}")).await + } + }; + let cues = match arr_subs::srt::parse(&raw) { + Ok(cues) => cues, + Err(error) => { + return record( + SubtitleState::Failed, + format!("{source_path}: not SRT: {error}"), + ) + .await + } + }; + + let broken_name = format!("translator {engine}"); + let translated = + match arr_subs::translate::translate(backend.as_ref(), &cues, &source_language, wanted) + .await + { + Ok(translated) => translated, + Err(error) => { + if error.is_transient() { + self.notify_broken(&broken_name).await; + } + return record(SubtitleState::Failed, error.to_string()).await; + } + }; + self.clear_broken(&broken_name).await; + + self.write_translation(database, target, wanted, wanted_tag, engine, &translated) + .await + } + + /// Sync and write a finished translation, record it, settle the want. + async fn write_translation( + &self, + database: &Db, + target: &Target, + wanted: &Language, + wanted_tag: &str, + engine: String, + translated: &[arr_subs::Cue], + ) -> Result { + let media_file_id = target.media_file_id; + let record = |state: SubtitleState, reason: String| async move { + db::record_attempt( + database.pool(), + media_file_id, + wanted_tag, + state, + Some(&reason), + ) + .await?; + Ok(Closed::Recorded { state, reason }) + }; + + let Some(destination) = target.sidecar(wanted, true) else { + return record( + SubtitleState::Failed, + format!("media file path {} has no folder", target.path.display()), + ) + .await; + }; + if let Err(error) = write_sidecar(&destination, &arr_subs::srt::render(translated)).await { + return record(SubtitleState::Failed, error).await; + } + + let sync = self.syncer.settle(&target.path, &destination).await; + if let Some(synced) = &sync.content { + if let Err(error) = write_sidecar(&destination, synced).await { + return record(SubtitleState::Failed, error).await; + } + } + + db::record_file( + database.pool(), + &arr_db::NewSubtitleFile::translated( + media_file_id, + wanted_tag, + &engine, + &destination.to_string_lossy(), + ) + .sync(db_sync_state(sync.state)), + ) + .await?; + db::mark_satisfied(database.pool(), media_file_id, wanted_tag).await?; + self.refresh_jellyfin().await; + Ok(Closed::Translated { engine }) + } + + /// The subtitle to translate from: any non-forced sidecar arr wrote — a + /// real one before a machine translation — or, failing that, a + /// text-format embedded track extracted now (§15). + async fn translation_source( + &self, + database: &Db, + target: &Target, + ) -> Result, SubtitleError> { + let files = db::files_for(database.pool(), target.media_file_id).await?; + let sidecar = files + .iter() + .filter(|file| !file.forced && file.path.is_some()) + .min_by_key(|file| matches!(file.origin, SubtitleOrigin::Translated)); + if let Some(file) = sidecar { + if let Some(path) = file.path.clone() { + return Ok(Some((path, arr_db::policy::language(&file.language)))); + } + } + + let probed = sqlx::query_scalar!( + r#"SELECT probed FROM media_files WHERE id = ?"#, + target.media_file_id + ) + .fetch_optional(database.pool()) + .await? + .flatten(); + let Some(probed) = probed else { + return Ok(None); + }; + let Some(track) = embedded_tracks(&probed) + .into_iter() + .find(|track| !track.forced && track.is_text()) + else { + return Ok(None); + }; + + let language = arr_db::policy::language(&track.language); + let Some(destination) = target.sidecar(&language, false) else { + return Ok(None); + }; + if let Err(error) = self + .extractor + .extract_srt(&target.path, track.index, &destination) + .await + { + tracing::warn!( + path = %target.path.display(), + stream = track.index, + %error, + "embedded track extraction failed" + ); + return Ok(None); + } + let mut record = arr_db::NewSubtitleFile::extracted( + target.media_file_id, + &track.language, + &destination.to_string_lossy(), + ); + if track.sdh { + record = record.sdh(); + } + db::record_file(database.pool(), &record).await?; + Ok(Some((destination.to_string_lossy().into_owned(), language))) + } + + /// The same single rescan import makes (§7.5): Jellyfin's watcher misses + /// a sidecar dropped next to a file it already knows, and a failure here + /// must not fail the write that already landed. + async fn refresh_jellyfin(&self) { + if let Err(error) = self.jellyfin.refresh().await { + tracing::warn!(%error, "jellyfin refresh failed"); + } + } +} + +/// The ranked winner across every provider's answer, preferred language +/// first (§15: pt-PT before pt-BR), plain before SDH within one language. +fn choose( + target: &Target, + hash: Option<&str>, + offered: &[Candidate], + languages: &[Language], +) -> Option { + let ranking_target = SubtitleTarget { + moviehash: hash, + release_name: target.release_name.as_deref(), + release_group: target.release_group.as_deref(), + source: target.source, + }; + for language in languages { + let bucket: Vec<&Candidate> = offered + .iter() + .filter(|candidate| &candidate.language == language) + .collect(); + let cores: Vec<_> = bucket + .iter() + .map(|candidate| candidate.to_core(hash)) + .collect(); + let winner = rank(&ranking_target, &cores) + .into_iter() + .find(|ranked| ranked.verdict == SubtitleVerdict::Eligible); + if let Some(winner) = winner { + return Some(bucket[winner.index].clone()); + } + } + None +} + +/// Closes §15 subtitle gaps: seeds the wanted set over every probed media +/// file, settles what embedded tracks already satisfy, and dispatches one +/// detached close per remaining gap. +#[derive(Debug)] +pub struct SubtitleAction { + worker: Worker, + /// Spawn closes as detached tasks. Tests run them inline so a tick's + /// effects are observable when it returns. + detach: bool, +} + +impl SubtitleAction { + #[must_use] + pub fn new( + providers: Vec>, + backends: Vec>, + syncer: Syncer, + extractor: Extractor, + jellyfin: JellyfinClient, + ) -> Self { + Self { + worker: Worker { + providers: Arc::new(providers), + backends: Arc::new(backends), + syncer, + extractor, + jellyfin, + notifier: None, + broken: Arc::new(Mutex::new(BTreeSet::new())), + in_flight: Arc::new(Mutex::new(BTreeSet::new())), + }, + detach: true, + } + } + + /// Send §9.5 *broken* to the operator when a provider or translator + /// stops answering. Without a topic the outage is only logged. + #[must_use] + pub fn with_notifier(mut self, notifier: Notifier, operator_topic: String) -> Self { + self.worker.notifier = Some((notifier, operator_topic)); + self + } + + #[cfg(test)] + fn inline(mut self) -> Self { + self.detach = false; + self + } + + async fn tick(&self, database: &Db) -> Result, SubtitleError> { + let settings = load_settings(database).await?; + for language in &settings.wanted { + seed_wants(database, language).await?; + } + + let pending = db::pending(database.pool(), PENDING_LIMIT).await?; + let mut outcomes = Vec::new(); + let mut embedded_recorded: HashSet = HashSet::new(); + for gap in pending { + let media_file_id = gap.attempt.media_file_id; + let language = gap.attempt.language.clone(); + // A row seeded under an earlier wanted set is not a gap the + // operator still wants closed; it stays inert until wanted again. + if !settings.wanted.contains(&language) { + continue; + } + if !backoff_elapsed(gap.attempt.attempts, gap.attempt.last_attempt_at.as_deref()) { + continue; + } + if self + .worker + .in_flight + .lock() + .await + .contains(&(media_file_id, language.clone())) + { + continue; + } + + if embedded_recorded.insert(media_file_id) { + record_embedded(database, media_file_id).await?; + } + if satisfied_by_file(database, media_file_id, &language).await? { + db::mark_satisfied(database.pool(), media_file_id, &language).await?; + outcomes.push(Outcome::new( + format!("{language} missing for {}", gap.media_file_path), + "an existing subtitle satisfies it", + )); + continue; + } + + { + let mut in_flight = self.worker.in_flight.lock().await; + if in_flight.len() >= MAX_IN_FLIGHT { + break; + } + in_flight.insert((media_file_id, language.clone())); + } + outcomes.push(Outcome::new( + format!("{language} missing for {}", gap.media_file_path), + "closing the gap", + )); + let worker = self.worker.clone(); + let database = database.clone(); + let settings = settings.clone(); + let close = async move { worker.close_and_log(database, gap, settings).await }; + if self.detach { + tokio::spawn(close); + } else { + close.await; + } + } + Ok(outcomes) + } +} + +/// Start tracking `language` for every probed media file (§15: the wanted +/// set is global). Existing rows are left alone — this never resets a +/// backoff or reopens a settled language. +async fn seed_wants(database: &Db, language: &str) -> Result<(), SubtitleError> { + sqlx::query!( + "INSERT INTO subtitle_attempts (media_file_id, language) + SELECT id, ? FROM media_files WHERE probed IS NOT NULL + ON CONFLICT (media_file_id, language) DO NOTHING", + language + ) + .execute(database.pool()) + .await?; + Ok(()) +} + +/// Record every embedded track the probe found as a subtitle row (#189's +/// leftover half), idempotently — the unique index converges a re-probe. +async fn record_embedded(database: &Db, media_file_id: i64) -> Result<(), SubtitleError> { + let probed = sqlx::query_scalar!( + r#"SELECT probed FROM media_files WHERE id = ?"#, + media_file_id + ) + .fetch_optional(database.pool()) + .await? + .flatten(); + let Some(probed) = probed else { + return Ok(()); + }; + for track in embedded_tracks(&probed) { + let mut record = arr_db::NewSubtitleFile::embedded(media_file_id, &track.language); + if track.forced { + record = record.forced(); + } + if track.sdh { + record = record.sdh(); + } + db::record_file(database.pool(), &record).await?; + } + Ok(()) +} + +/// Whether any non-forced subtitle already answers `wanted` (§15: embedded +/// or sidecar, machine translation included). +async fn satisfied_by_file( + database: &Db, + media_file_id: i64, + wanted: &str, +) -> Result { + let files = db::files_for(database.pool(), media_file_id).await?; + Ok(files + .iter() + .any(|file| !file.forced && satisfies(wanted, &file.language))) +} + +impl Action for SubtitleAction { + fn name(&self) -> &'static str { + "subtitles" + } + + fn run<'a>(&'a self, database: &'a Db) -> ActionFuture<'a> { + Box::pin(async move { self.tick(database).await.map_err(Into::into) }) + } +} + +#[cfg(test)] +#[allow(clippy::unwrap_used)] +mod tests { + use std::sync::atomic::{AtomicUsize, Ordering}; + + use arr_subs::{ + BackendId, Batch, CandidateId, DownloadFuture, Fetched, ProviderId, SearchFuture, + SubtitleFormat, TranslateFuture, TranslatedCue, + }; + use wiremock::matchers::{method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + use super::*; + + const SRT: &str = "1\n00:00:01,000 --> 00:00:02,000\nhello there\n"; + + #[derive(Debug, Clone, Copy)] + enum Fail { + Transport, + RateLimited, + } + + #[derive(Debug)] + struct StubProvider { + name: &'static str, + candidates: Vec, + searches: Arc, + fail: Option, + } + + impl StubProvider { + fn new(name: &'static str, candidates: Vec) -> Self { + Self { + name, + candidates, + searches: Arc::new(AtomicUsize::new(0)), + fail: None, + } + } + + fn failing(name: &'static str, fail: Fail) -> Self { + Self { + fail: Some(fail), + ..Self::new(name, Vec::new()) + } + } + } + + impl Provider for StubProvider { + fn id(&self) -> ProviderId { + ProviderId::new(self.name) + } + + fn search<'a>(&'a self, _request: &'a SearchRequest) -> SearchFuture<'a> { + Box::pin(async move { + self.searches.fetch_add(1, Ordering::SeqCst); + match self.fail { + Some(Fail::Transport) => Err(arr_subs::Error::Transport { + provider: self.id(), + source: Box::new(std::io::Error::other("connection refused")), + }), + Some(Fail::RateLimited) => Err(arr_subs::Error::RateLimited { + provider: self.id(), + retry_after: None, + }), + None => Ok(self.candidates.clone()), + } + }) + } + + fn download<'a>(&'a self, id: &'a CandidateId) -> DownloadFuture<'a> { + Box::pin(async move { + Ok(Fetched { + id: id.clone(), + language: Language::PortuguesePortugal, + format: SubtitleFormat::Srt, + content: SRT.as_bytes().to_vec(), + }) + }) + } + } + + #[derive(Debug)] + struct StubBackend; + + impl Backend for StubBackend { + fn id(&self) -> BackendId { + BackendId::new("openai") + } + + fn supports(&self, _target: &Language) -> bool { + true + } + + fn translate<'a>(&'a self, batch: &'a Batch) -> TranslateFuture<'a> { + Box::pin(async move { + Ok(batch + .cues + .iter() + .map(|cue| TranslatedCue { + number: cue.number, + text: cue.text.to_uppercase(), + }) + .collect()) + }) + } + } + + fn candidate(provider: &str, id: &str, language: Language) -> Candidate { + Candidate { + provider: ProviderId::new(provider), + id: CandidateId::new(id), + language, + hash_match: false, + release_name: None, + group: None, + source: None, + rating: None, + download_count: None, + forced: false, + sdh: false, + } + } + + struct Fixture { + database: Db, + directory: tempfile::TempDir, + media_file_id: i64, + video: PathBuf, + } + + impl Fixture { + async fn new(probed: &serde_json::Value) -> Self { + let directory = tempfile::tempdir().unwrap(); + let database = Db::connect(directory.path().join("arr.db")).await.unwrap(); + database.migrate().await.unwrap(); + + let root_id: i64 = sqlx::query_scalar( + "SELECT id FROM roots WHERE kind = 'movie' AND audience = 'main'", + ) + .fetch_one(database.pool()) + .await + .unwrap(); + let movie_id: i64 = sqlx::query_scalar( + "INSERT INTO movies (tmdb_id, title, root_id) VALUES (1, 'Movie', ?) RETURNING id", + ) + .bind(root_id) + .fetch_one(database.pool()) + .await + .unwrap(); + + let video = directory.path().join("Movie (2024) - [1080p].mkv"); + tokio::fs::write(&video, b"not really a video") + .await + .unwrap(); + let media_file_id: i64 = sqlx::query_scalar( + "INSERT INTO media_files (owner_kind, owner_id, path, size, probed) + VALUES ('movie', ?, ?, 18, ?) RETURNING id", + ) + .bind(movie_id) + .bind(video.to_string_lossy().into_owned()) + .bind(probed.to_string()) + .fetch_one(database.pool()) + .await + .unwrap(); + + Self { + database, + directory, + media_file_id, + video, + } + } + + async fn configure(&self, wanted: &str, engine: Option<&str>) { + sqlx::query( + "UPDATE subtitle_settings SET wanted_languages = ?, translation_engine = ?", + ) + .bind(wanted) + .bind(engine) + .execute(self.database.pool()) + .await + .unwrap(); + } + + async fn attempt(&self, language: &str) -> Option { + db::attempts_for(self.database.pool(), self.media_file_id) + .await + .unwrap() + .into_iter() + .find(|attempt| attempt.language == language) + } + + async fn files(&self) -> Vec { + db::files_for(self.database.pool(), self.media_file_id) + .await + .unwrap() + } + } + + fn action( + providers: Vec>, + backends: Vec>, + ) -> SubtitleAction { + SubtitleAction::new( + providers, + backends, + Syncer::new().with_binary("alass-not-installed"), + Extractor::new().with_binary("ffmpeg-not-installed"), + JellyfinClient::new("http://127.0.0.1:1", None).unwrap(), + ) + .inline() + } + + fn no_tracks() -> serde_json::Value { + serde_json::json!({ "sub_tracks": [] }) + } + + #[tokio::test] + async fn an_embedded_track_satisfies_without_asking_any_provider() { + let fixture = Fixture::new(&serde_json::json!({ "sub_tracks": [ + { "language": "en", "codec": "subrip", "forced": false, "sdh": false }, + { "language": "pt-PT", "codec": "hdmv_pgs_subtitle", "forced": false, "sdh": false }, + ] })) + .await; + let provider = StubProvider::new("opensubtitles", Vec::new()); + let searches = Arc::clone(&provider.searches); + let action = action(vec![Arc::new(provider)], Vec::new()); + + let outcomes = action.run(&fixture.database).await.unwrap(); + + assert_eq!(outcomes.len(), 2); + for language in ["en", "pt-PT"] { + let attempt = fixture.attempt(language).await.unwrap(); + assert_eq!(attempt.state, SubtitleState::Satisfied); + assert_eq!(attempt.attempts, 0, "an embedded track spends nothing"); + } + assert_eq!(searches.load(Ordering::SeqCst), 0); + let files = fixture.files().await; + assert_eq!(files.len(), 2); + assert!(files + .iter() + .all(|file| file.origin == SubtitleOrigin::Embedded)); + } + + #[tokio::test] + async fn a_forced_track_satisfies_nothing() { + let fixture = Fixture::new(&serde_json::json!({ "sub_tracks": [ + { "language": "en", "codec": "subrip", "forced": true, "sdh": false }, + ] })) + .await; + fixture.configure(r#"["en"]"#, None).await; + let action = action( + vec![Arc::new(StubProvider::new("opensubtitles", Vec::new()))], + Vec::new(), + ); + + action.run(&fixture.database).await.unwrap(); + + // No candidate anywhere and a forced track is not a translation + // source, so the language is recorded as unavailable (§15). + let attempt = fixture.attempt("en").await.unwrap(); + assert_eq!(attempt.state, SubtitleState::Unavailable); + assert!(attempt + .last_failure + .unwrap() + .contains("no text source to translate from")); + } + + #[tokio::test] + async fn the_ranked_winner_is_fetched_synced_and_recorded() { + let fixture = Fixture::new(&no_tracks()).await; + fixture.configure(r#"["pt-PT"]"#, None).await; + let provider = StubProvider::new( + "opensubtitles", + vec![candidate( + "opensubtitles", + "42", + Language::PortuguesePortugal, + )], + ); + let action = action(vec![Arc::new(provider)], Vec::new()); + + action.run(&fixture.database).await.unwrap(); + + let sidecar = fixture + .directory + .path() + .join("Movie (2024) - [1080p].pt-PT.srt"); + let written = tokio::fs::read_to_string(&sidecar).await.unwrap(); + assert_eq!(written.trim(), SRT.trim()); + + let files = fixture.files().await; + assert_eq!(files.len(), 1); + assert_eq!(files[0].origin, SubtitleOrigin::Provider); + assert_eq!(files[0].provider.as_deref(), Some("opensubtitles")); + assert_eq!(files[0].candidate_id.as_deref(), Some("42")); + // `alass` is deliberately not installed here, so the sync did not run + // — the §15 flags record exactly that. + assert_eq!(files[0].sync, arr_db::SubtitleSync::NotRun); + let attempt = fixture.attempt("pt-PT").await.unwrap(); + assert_eq!(attempt.state, SubtitleState::Satisfied); + } + + #[tokio::test] + async fn pt_br_is_accepted_when_no_pt_pt_exists() { + let fixture = Fixture::new(&no_tracks()).await; + fixture.configure(r#"["pt-PT"]"#, None).await; + let provider = StubProvider::new( + "opensubtitles", + vec![candidate("opensubtitles", "7", Language::PortugueseBrazil)], + ); + let action = action(vec![Arc::new(provider)], Vec::new()); + + action.run(&fixture.database).await.unwrap(); + + let files = fixture.files().await; + assert_eq!(files[0].language, "pt-BR"); + assert!(files[0].path.as_deref().unwrap().ends_with(".pt-BR.srt")); + let attempt = fixture.attempt("pt-PT").await.unwrap(); + assert_eq!(attempt.state, SubtitleState::Satisfied); + } + + #[tokio::test] + async fn pt_pt_is_preferred_over_a_better_ranked_pt_br() { + let fixture = Fixture::new(&no_tracks()).await; + fixture.configure(r#"["pt-PT"]"#, None).await; + let stronger_br = Candidate { + hash_match: true, + download_count: Some(1_000_000), + ..candidate("opensubtitles", "br", Language::PortugueseBrazil) + }; + let provider = StubProvider::new( + "opensubtitles", + vec![ + stronger_br, + candidate("opensubtitles", "pt", Language::PortuguesePortugal), + ], + ); + let action = action(vec![Arc::new(provider)], Vec::new()); + + action.run(&fixture.database).await.unwrap(); + + let files = fixture.files().await; + assert_eq!(files[0].language, "pt-PT"); + assert_eq!(files[0].candidate_id.as_deref(), Some("pt")); + } + + #[tokio::test] + async fn an_unreachable_provider_backs_off_and_notifies_once() { + let fixture = Fixture::new(&no_tracks()).await; + fixture.configure(r#"["pt-PT"]"#, None).await; + let ntfy = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/ops")) + .respond_with(ResponseTemplate::new(200)) + .mount(&ntfy) + .await; + let provider = StubProvider::failing("opensubtitles", Fail::Transport); + let searches = Arc::clone(&provider.searches); + let action = action(vec![Arc::new(provider)], Vec::new()) + .with_notifier(Notifier::new(ntfy.uri()).unwrap(), "ops".to_owned()); + + action.run(&fixture.database).await.unwrap(); + let attempt = fixture.attempt("pt-PT").await.unwrap(); + assert_eq!(attempt.state, SubtitleState::Failed); + assert_eq!(attempt.attempts, 1); + assert!(attempt.last_failure.unwrap().contains("unreachable")); + + // Within the backoff window nothing is retried and nothing is + // re-notified. + action.run(&fixture.database).await.unwrap(); + assert_eq!(searches.load(Ordering::SeqCst), 1); + assert_eq!(ntfy.received_requests().await.unwrap().len(), 1); + } + + #[tokio::test] + async fn a_rate_limited_provider_is_a_queue_state_not_a_failure() { + let fixture = Fixture::new(&no_tracks()).await; + fixture.configure(r#"["pt-PT"]"#, None).await; + let provider = StubProvider::failing("opensubtitles", Fail::RateLimited); + let searches = Arc::clone(&provider.searches); + let action = action(vec![Arc::new(provider)], Vec::new()); + + action.run(&fixture.database).await.unwrap(); + let attempt = fixture.attempt("pt-PT").await.unwrap(); + assert_eq!(attempt.state, SubtitleState::Capped); + + // `capped` rows are not in the work list, so the provider is left + // alone until something else moves the row. + action.run(&fixture.database).await.unwrap(); + assert_eq!(searches.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn no_candidate_translates_immediately_from_a_fetched_subtitle() { + let fixture = Fixture::new(&no_tracks()).await; + fixture.configure(r#"["pt-PT"]"#, Some("openai")).await; + let source = fixture + .directory + .path() + .join("Movie (2024) - [1080p].en.srt"); + tokio::fs::write(&source, SRT).await.unwrap(); + db::record_file( + fixture.database.pool(), + &arr_db::NewSubtitleFile::fetched( + fixture.media_file_id, + "en", + "opensubtitles", + "1", + &source.to_string_lossy(), + ), + ) + .await + .unwrap(); + let action = action( + vec![Arc::new(StubProvider::new("opensubtitles", Vec::new()))], + vec![Arc::new(StubBackend)], + ); + + action.run(&fixture.database).await.unwrap(); + + let translated = fixture + .directory + .path() + .join("Movie (2024) - [1080p].pt-PT.mt.srt"); + let written = tokio::fs::read_to_string(&translated).await.unwrap(); + assert!(written.contains("HELLO THERE")); + let files = fixture.files().await; + let machine = files + .iter() + .find(|file| file.origin == SubtitleOrigin::Translated) + .unwrap(); + assert_eq!(machine.engine.as_deref(), Some("openai")); + assert_eq!(machine.language, "pt-PT"); + let attempt = fixture.attempt("pt-PT").await.unwrap(); + assert_eq!(attempt.state, SubtitleState::Satisfied); + } + + #[tokio::test] + async fn no_engine_configured_backs_off_instead_of_translating() { + let fixture = Fixture::new(&no_tracks()).await; + fixture.configure(r#"["pt-PT"]"#, None).await; + let source = fixture + .directory + .path() + .join("Movie (2024) - [1080p].en.srt"); + tokio::fs::write(&source, SRT).await.unwrap(); + db::record_file( + fixture.database.pool(), + &arr_db::NewSubtitleFile::fetched( + fixture.media_file_id, + "en", + "opensubtitles", + "1", + &source.to_string_lossy(), + ), + ) + .await + .unwrap(); + let action = action( + vec![Arc::new(StubProvider::new("opensubtitles", Vec::new()))], + Vec::new(), + ); + + action.run(&fixture.database).await.unwrap(); + + let attempt = fixture.attempt("pt-PT").await.unwrap(); + assert_eq!(attempt.state, SubtitleState::Failed); + assert_eq!( + attempt.last_failure.as_deref(), + Some("no translation engine is configured") + ); + } + + #[tokio::test] + async fn a_text_embedded_track_is_extracted_as_the_translation_source() { + let fixture = Fixture::new(&serde_json::json!({ "sub_tracks": [ + { "language": "en", "codec": "subrip", "forced": false, "sdh": false }, + ] })) + .await; + fixture.configure(r#"["pt-PT"]"#, Some("openai")).await; + // A real container with a real text track, extracted by real ffmpeg — + // the same fixture arr-probe's extraction tests use. + let clip = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .join("../arr-probe/tests/fixtures/movie-1080p.mkv"); + tokio::fs::copy(&clip, &fixture.video).await.unwrap(); + let action = SubtitleAction::new( + vec![Arc::new(StubProvider::new("opensubtitles", Vec::new()))], + vec![Arc::new(StubBackend)], + Syncer::new().with_binary("alass-not-installed"), + Extractor::new(), + JellyfinClient::new("http://127.0.0.1:1", None).unwrap(), + ) + .inline(); + + action.run(&fixture.database).await.unwrap(); + + let files = fixture.files().await; + let extracted = files + .iter() + .find(|file| file.origin == SubtitleOrigin::Extracted) + .unwrap(); + assert!(extracted.path.as_deref().unwrap().ends_with(".en.srt")); + let machine = files + .iter() + .find(|file| file.origin == SubtitleOrigin::Translated) + .unwrap(); + assert_eq!(machine.language, "pt-PT"); + assert_eq!( + fixture.attempt("pt-PT").await.unwrap().state, + SubtitleState::Satisfied + ); + } + + #[tokio::test] + async fn a_language_no_longer_wanted_is_left_alone() { + let fixture = Fixture::new(&no_tracks()).await; + fixture.configure(r#"["pt-PT"]"#, None).await; + db::want(fixture.database.pool(), fixture.media_file_id, "es") + .await + .unwrap(); + let provider = StubProvider::new( + "opensubtitles", + vec![candidate( + "opensubtitles", + "42", + Language::PortuguesePortugal, + )], + ); + let searches = Arc::clone(&provider.searches); + let action = action(vec![Arc::new(provider)], Vec::new()); + + action.run(&fixture.database).await.unwrap(); + + assert_eq!(searches.load(Ordering::SeqCst), 1, "only the pt-PT gap"); + assert_eq!( + fixture.attempt("es").await.unwrap().state, + SubtitleState::Wanted + ); + } +}