From ffb8485939f6d21beb7afa495b2b0ccbfee2e978 Mon Sep 17 00:00:00 2001 From: "naps62-yolo (agent)" Date: Sat, 22 Aug 2026 23:21:30 +0100 Subject: [PATCH] Import pipeline: probe, hardlink, rename, layout (#82) --- ...35bc7d0f24dc115e2515c101f8478464b75e0.json | 12 + ...22c2bce92c6483564c159731e64fdd5f7f8bf.json | 20 + ...a68123fb59610ba8ab8e8f6c1a1189610e377.json | 12 + ...e4759784a92ad87cb151150689fc355473328.json | 12 + ...749cf7a3e1a53a8324e5c816c7745418ae34d.json | 12 + ...bd08215204fadfb92b7f676e857e0b3400408.json | 62 ++ ...fa379d232b045498310debab5ddd97f039294.json | 12 + ...a4f31ebed4321365caceb557c7ea23ba69f7d.json | 12 + Cargo.lock | 1 + crates/arr-core/src/layout.rs | 198 ++++ crates/arr-core/src/lib.rs | 83 +- crates/arr-daemon/Cargo.toml | 1 + crates/arr-daemon/src/grab.rs | 66 +- crates/arr-daemon/src/import.rs | 991 ++++++++++++++++++ crates/arr-daemon/src/main.rs | 12 +- crates/arr-dl/src/lib.rs | 134 +++ crates/arr-parse/src/lib.rs | 21 + crates/arr-parse/tests/parse.rs | 12 + 18 files changed, 1653 insertions(+), 20 deletions(-) create mode 100644 .sqlx/query-0f53855c99d83e4d291e24643f035bc7d0f24dc115e2515c101f8478464b75e0.json create mode 100644 .sqlx/query-48c46aa1908add3d6abff142dbb22c2bce92c6483564c159731e64fdd5f7f8bf.json create mode 100644 .sqlx/query-93d05b3370c722a59239b486f5ca68123fb59610ba8ab8e8f6c1a1189610e377.json create mode 100644 .sqlx/query-b4b63449fdfb0bf5fb7095577d2e4759784a92ad87cb151150689fc355473328.json create mode 100644 .sqlx/query-c249ca062f44f32ee4a554f36a0749cf7a3e1a53a8324e5c816c7745418ae34d.json create mode 100644 .sqlx/query-c602f6e08e0a8d828dfff1d821cbd08215204fadfb92b7f676e857e0b3400408.json create mode 100644 .sqlx/query-c66f29b6edc4bca3a751e85e133fa379d232b045498310debab5ddd97f039294.json create mode 100644 .sqlx/query-ff1f65308ec991007ba9e0d7c22a4f31ebed4321365caceb557c7ea23ba69f7d.json create mode 100644 crates/arr-core/src/layout.rs create mode 100644 crates/arr-daemon/src/import.rs diff --git a/.sqlx/query-0f53855c99d83e4d291e24643f035bc7d0f24dc115e2515c101f8478464b75e0.json b/.sqlx/query-0f53855c99d83e4d291e24643f035bc7d0f24dc115e2515c101f8478464b75e0.json new file mode 100644 index 0000000..6590e3f --- /dev/null +++ b/.sqlx/query-0f53855c99d83e4d291e24643f035bc7d0f24dc115e2515c101f8478464b75e0.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "UPDATE movies\n SET state = 'missing',\n updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')\n WHERE id = ?", + "describe": { + "columns": [], + "parameters": { + "Right": 1 + }, + "nullable": [] + }, + "hash": "0f53855c99d83e4d291e24643f035bc7d0f24dc115e2515c101f8478464b75e0" +} diff --git a/.sqlx/query-48c46aa1908add3d6abff142dbb22c2bce92c6483564c159731e64fdd5f7f8bf.json b/.sqlx/query-48c46aa1908add3d6abff142dbb22c2bce92c6483564c159731e64fdd5f7f8bf.json new file mode 100644 index 0000000..0683baa --- /dev/null +++ b/.sqlx/query-48c46aa1908add3d6abff142dbb22c2bce92c6483564c159731e64fdd5f7f8bf.json @@ -0,0 +1,20 @@ +{ + "db_name": "SQLite", + "query": "SELECT normalised_name AS \"normalised_name!: String\" FROM blacklist", + "describe": { + "columns": [ + { + "name": "normalised_name!: String", + "ordinal": 0, + "type_info": "Text" + } + ], + "parameters": { + "Right": 0 + }, + "nullable": [ + false + ] + }, + "hash": "48c46aa1908add3d6abff142dbb22c2bce92c6483564c159731e64fdd5f7f8bf" +} diff --git a/.sqlx/query-93d05b3370c722a59239b486f5ca68123fb59610ba8ab8e8f6c1a1189610e377.json b/.sqlx/query-93d05b3370c722a59239b486f5ca68123fb59610ba8ab8e8f6c1a1189610e377.json new file mode 100644 index 0000000..b7443d2 --- /dev/null +++ b/.sqlx/query-93d05b3370c722a59239b486f5ca68123fb59610ba8ab8e8f6c1a1189610e377.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "UPDATE movies\n SET state = 'imported',\n updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')\n WHERE id = ?", + "describe": { + "columns": [], + "parameters": { + "Right": 1 + }, + "nullable": [] + }, + "hash": "93d05b3370c722a59239b486f5ca68123fb59610ba8ab8e8f6c1a1189610e377" +} diff --git a/.sqlx/query-b4b63449fdfb0bf5fb7095577d2e4759784a92ad87cb151150689fc355473328.json b/.sqlx/query-b4b63449fdfb0bf5fb7095577d2e4759784a92ad87cb151150689fc355473328.json new file mode 100644 index 0000000..72d9092 --- /dev/null +++ b/.sqlx/query-b4b63449fdfb0bf5fb7095577d2e4759784a92ad87cb151150689fc355473328.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "INSERT INTO blacklist (infohash, normalised_name, reason)\n VALUES (?, ?, ?)\n ON CONFLICT (infohash) DO NOTHING", + "describe": { + "columns": [], + "parameters": { + "Right": 3 + }, + "nullable": [] + }, + "hash": "b4b63449fdfb0bf5fb7095577d2e4759784a92ad87cb151150689fc355473328" +} diff --git a/.sqlx/query-c249ca062f44f32ee4a554f36a0749cf7a3e1a53a8324e5c816c7745418ae34d.json b/.sqlx/query-c249ca062f44f32ee4a554f36a0749cf7a3e1a53a8324e5c816c7745418ae34d.json new file mode 100644 index 0000000..e6be1a8 --- /dev/null +++ b/.sqlx/query-c249ca062f44f32ee4a554f36a0749cf7a3e1a53a8324e5c816c7745418ae34d.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "UPDATE grabs\n SET state = 'imported',\n imported_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')\n WHERE id = ?", + "describe": { + "columns": [], + "parameters": { + "Right": 1 + }, + "nullable": [] + }, + "hash": "c249ca062f44f32ee4a554f36a0749cf7a3e1a53a8324e5c816c7745418ae34d" +} diff --git a/.sqlx/query-c602f6e08e0a8d828dfff1d821cbd08215204fadfb92b7f676e857e0b3400408.json b/.sqlx/query-c602f6e08e0a8d828dfff1d821cbd08215204fadfb92b7f676e857e0b3400408.json new file mode 100644 index 0000000..75b7fee --- /dev/null +++ b/.sqlx/query-c602f6e08e0a8d828dfff1d821cbd08215204fadfb92b7f676e857e0b3400408.json @@ -0,0 +1,62 @@ +{ + "db_name": "SQLite", + "query": "\n SELECT g.id AS \"grab_id!: i64\",\n g.infohash AS \"infohash!: String\",\n m.id AS \"movie_id!: i64\",\n m.tmdb_id AS \"tmdb_id!: i64\",\n m.title AS \"title!: String\",\n m.year,\n m.original_language,\n r.name AS \"release_name!: String\"\n FROM grabs g\n JOIN movies m ON m.id = g.target_id\n JOIN releases r ON r.id = g.release_id\n WHERE g.state = 'downloaded' AND g.target_kind = 'movie'\n ORDER BY g.id\n ", + "describe": { + "columns": [ + { + "name": "grab_id!: i64", + "ordinal": 0, + "type_info": "Integer" + }, + { + "name": "infohash!: String", + "ordinal": 1, + "type_info": "Text" + }, + { + "name": "movie_id!: i64", + "ordinal": 2, + "type_info": "Integer" + }, + { + "name": "tmdb_id!: i64", + "ordinal": 3, + "type_info": "Integer" + }, + { + "name": "title!: String", + "ordinal": 4, + "type_info": "Text" + }, + { + "name": "year", + "ordinal": 5, + "type_info": "Integer" + }, + { + "name": "original_language", + "ordinal": 6, + "type_info": "Text" + }, + { + "name": "release_name!: String", + "ordinal": 7, + "type_info": "Text" + } + ], + "parameters": { + "Right": 0 + }, + "nullable": [ + true, + false, + false, + false, + false, + true, + true, + false + ] + }, + "hash": "c602f6e08e0a8d828dfff1d821cbd08215204fadfb92b7f676e857e0b3400408" +} diff --git a/.sqlx/query-c66f29b6edc4bca3a751e85e133fa379d232b045498310debab5ddd97f039294.json b/.sqlx/query-c66f29b6edc4bca3a751e85e133fa379d232b045498310debab5ddd97f039294.json new file mode 100644 index 0000000..e3636bf --- /dev/null +++ b/.sqlx/query-c66f29b6edc4bca3a751e85e133fa379d232b045498310debab5ddd97f039294.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "UPDATE grabs SET state = 'failed' WHERE id = ?", + "describe": { + "columns": [], + "parameters": { + "Right": 1 + }, + "nullable": [] + }, + "hash": "c66f29b6edc4bca3a751e85e133fa379d232b045498310debab5ddd97f039294" +} diff --git a/.sqlx/query-ff1f65308ec991007ba9e0d7c22a4f31ebed4321365caceb557c7ea23ba69f7d.json b/.sqlx/query-ff1f65308ec991007ba9e0d7c22a4f31ebed4321365caceb557c7ea23ba69f7d.json new file mode 100644 index 0000000..07abf3a --- /dev/null +++ b/.sqlx/query-ff1f65308ec991007ba9e0d7c22a4f31ebed4321365caceb557c7ea23ba69f7d.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "INSERT INTO media_files (owner_kind, owner_id, path, size, probed, waiver)\n VALUES ('movie', ?, ?, ?, ?, ?)\n ON CONFLICT (path) DO UPDATE SET\n size = excluded.size,\n probed = excluded.probed,\n waiver = excluded.waiver,\n updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')", + "describe": { + "columns": [], + "parameters": { + "Right": 5 + }, + "nullable": [] + }, + "hash": "ff1f65308ec991007ba9e0d7c22a4f31ebed4321365caceb557c7ea23ba69f7d" +} diff --git a/Cargo.lock b/Cargo.lock index 619249c..c6fb5df 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -86,6 +86,7 @@ dependencies = [ "arr-indexer", "arr-meta", "arr-parse", + "arr-probe", "axum", "chrono", "include_dir", diff --git a/crates/arr-core/src/layout.rs b/crates/arr-core/src/layout.rs new file mode 100644 index 0000000..b006fda --- /dev/null +++ b/crates/arr-core/src/layout.rs @@ -0,0 +1,198 @@ +//! Library layout and naming (`DESIGN.md` §7.4). Pure string building; the +//! import pipeline joins these onto a root path and does the IO. +//! +//! ```text +//! Dune Part Two (2024) [tmdbid-693134]/ +//! Dune Part Two (2024) [tmdbid-693134] - [2160p][WEB-DL][HDR10].mkv +//! ``` +//! +//! The provider ID in the folder name turns Jellyfin matching into exact +//! lookup, and the attribute tags come from `ffprobe` so they are true — with +//! one exception: no file knows whether it came off a disc or a streaming +//! service (§5.6), so the source tag stays the release name's claim. Release +//! group is deliberately absent. + +use crate::{HdrFormat, Language, ProbedMedia, Source}; + +/// The per-title folder: `Dune Part Two (2024) [tmdbid-693134]`. +/// +/// One folder per title even for a single file, so sidecars stay contained +/// and deletes are atomic. +#[must_use] +pub fn movie_folder(title: &str, year: Option, tmdb_id: i64) -> String { + let title = sanitise(title); + match year { + Some(year) => format!("{title} ({year}) [tmdbid-{tmdb_id}]"), + None => format!("{title} [tmdbid-{tmdb_id}]"), + } +} + +/// The feature's filename: the folder name, the attribute tags, and the +/// source file's extension. +#[must_use] +pub fn movie_file_name( + title: &str, + year: Option, + tmdb_id: i64, + tags: &[String], + extension: Option<&str>, +) -> String { + let stem = movie_folder(title, year, tmdb_id); + let tags = tags.iter().fold(String::new(), |mut out, tag| { + out.push('['); + out.push_str(tag); + out.push(']'); + out + }); + match extension { + Some(extension) => format!("{stem} - {tags}.{extension}"), + None => format!("{stem} - {tags}"), + } +} + +/// The §7.4 attribute tags, in a fixed order: resolution, source, HDR, +/// Portuguese audio. +/// +/// Everything but the source comes from the probe. HDR is tagged only when +/// present, and audio only for Portuguese variants — that is the audit `ls` +/// answers: which files are DV, which of the kids' files are still +/// English-only. +#[must_use] +pub fn attribute_tags(media: &ProbedMedia, claimed_source: Option) -> Vec { + let mut tags = vec![media.resolution.to_string()]; + + if let Some(source) = claimed_source { + if source != Source::Other { + tags.push(source.to_string()); + } + } + + if media.hdr != HdrFormat::Sdr { + tags.push(media.hdr.to_string()); + } + + let mut portuguese: Vec<&Language> = Vec::new(); + for track in &media.audio_tracks { + if matches!( + track.language, + Language::PortuguesePortugal + | Language::PortugueseBrazil + | Language::PortugueseUnverified + ) && !portuguese.contains(&&track.language) + { + portuguese.push(&track.language); + } + } + tags.extend(portuguese.into_iter().map(ToString::to_string)); + + tags +} + +/// A title as a filesystem name: filesystem-hostile characters dropped, +/// whitespace collapsed. TMDB's `Dune: Part Two` becomes `Dune Part Two`. +fn sanitise(title: &str) -> String { + let cleaned: String = title + .chars() + .filter(|c| { + !matches!(c, '<' | '>' | ':' | '"' | '/' | '\\' | '|' | '?' | '*') && !c.is_control() + }) + .collect(); + cleaned + .split_whitespace() + .collect::>() + .join(" ") + .trim_matches(|c| c == '.' || c == ' ') + .to_owned() +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{AudioTrack, DolbyVisionProfile, Resolution, SubtitleTrack}; + + fn probed(resolution: Resolution, hdr: HdrFormat, audio: Vec) -> ProbedMedia { + ProbedMedia { + resolution, + source: None, + hdr, + audio_tracks: audio + .into_iter() + .map(|language| AudioTrack { + language, + title: None, + handler_name: None, + }) + .collect(), + subtitle_tracks: Vec::::new(), + } + } + + /// The exact §7.4 movie example. + #[test] + fn the_design_document_movie_example() { + let media = probed( + Resolution::R2160p, + HdrFormat::Hdr10, + vec![Language::Other("en".into())], + ); + let tags = attribute_tags(&media, Some(Source::WebDl)); + + assert_eq!( + movie_folder("Dune: Part Two", Some(2024), 693_134), + "Dune Part Two (2024) [tmdbid-693134]" + ); + assert_eq!( + movie_file_name("Dune: Part Two", Some(2024), 693_134, &tags, Some("mkv")), + "Dune Part Two (2024) [tmdbid-693134] - [2160p][WEB-DL][HDR10].mkv" + ); + } + + /// The kids audit surface: a pt-PT track is tagged, SDR is not. + #[test] + fn portuguese_audio_is_tagged_and_sdr_is_not() { + let media = probed( + Resolution::R1080p, + HdrFormat::Sdr, + vec![ + Language::Other("en".into()), + Language::PortuguesePortugal, + Language::PortuguesePortugal, + ], + ); + assert_eq!( + attribute_tags(&media, Some(Source::WebDl)), + vec!["1080p", "WEB-DL", "pt-PT"] + ); + } + + /// `ls` shows which files are DV, profile included (§5.3, §7.4). + #[test] + fn dolby_vision_tags_carry_the_profile() { + let media = probed( + Resolution::R2160p, + HdrFormat::DolbyVision(DolbyVisionProfile { + profile: 8, + compatibility_id: Some(1), + }), + vec![], + ); + assert_eq!(attribute_tags(&media, None), vec!["2160p", "DV8.1"]); + } + + /// An unknown source claim is no tag rather than a lie. + #[test] + fn unknown_sources_are_not_tagged() { + let media = probed(Resolution::R1080p, HdrFormat::Sdr, vec![]); + assert_eq!(attribute_tags(&media, Some(Source::Other)), vec!["1080p"]); + assert_eq!(attribute_tags(&media, None), vec!["1080p"]); + } + + #[test] + fn titles_are_sanitised_for_the_filesystem() { + assert_eq!( + movie_folder("What / If: A* Story?", Some(2020), 1), + "What If A Story (2020) [tmdbid-1]" + ); + assert_eq!(movie_folder("Untitled", None, 2), "Untitled [tmdbid-2]"); + } +} diff --git a/crates/arr-core/src/lib.rs b/crates/arr-core/src/lib.rs index 09de6ef..32775a0 100644 --- a/crates/arr-core/src/lib.rs +++ b/crates/arr-core/src/lib.rs @@ -2,9 +2,10 @@ //! //! This crate must never depend on `axum`, `sqlx` or `reqwest`. -use std::{collections::BTreeMap, path::PathBuf, time::SystemTime}; +use std::{collections::BTreeMap, fmt, path::PathBuf, time::SystemTime}; pub mod lang; +pub mod layout; pub mod policy; pub mod score; pub mod status; @@ -54,6 +55,19 @@ pub enum Language { Other(String), } +impl fmt::Display for Language { + /// The tag the policy columns and §7.4 filenames spell it as: + /// `pt-PT`, `pt-BR`, `por-unverified`, anything else verbatim. + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::PortuguesePortugal => formatter.write_str("pt-PT"), + Self::PortugueseBrazil => formatter.write_str("pt-BR"), + Self::PortugueseUnverified => formatter.write_str("por-unverified"), + Self::Other(tag) => formatter.write_str(tag), + } + } +} + #[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)] pub enum Resolution { R2160p, @@ -62,6 +76,18 @@ pub enum Resolution { Other(u16), } +impl fmt::Display for Resolution { + /// The spelling `resolution_pref` and §7.4 filename tags use. + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::R2160p => formatter.write_str("2160p"), + Self::R1080p => formatter.write_str("1080p"), + Self::R720p => formatter.write_str("720p"), + Self::Other(height) => write!(formatter, "{height}p"), + } + } +} + #[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)] pub enum Source { Remux, @@ -77,6 +103,25 @@ pub enum Source { Other, } +impl fmt::Display for Source { + /// The spelling `source_weights` and §7.4 filename tags use. + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str(match self { + Self::Remux => "Remux", + Self::BluRay => "BluRay", + Self::WebDl => "WEB-DL", + Self::WebRip => "WEBRip", + Self::Hdtv => "HDTV", + Self::Dvd => "DVD", + Self::Telecine => "Telecine", + Self::Telesync => "Telesync", + Self::Cam => "CAM", + Self::Screener => "Screener", + Self::Other => "Unknown", + }) + } +} + impl From for Resolution { fn from(value: arr_parse::Resolution) -> Self { match value { @@ -121,6 +166,24 @@ pub enum HdrFormat { DolbyVision(DolbyVisionProfile), } +impl fmt::Display for HdrFormat { + /// The §7.4 filename tag: `HDR10`, `DV8.1`, and so on. Dolby Vision + /// carries its profile because that is the whole point of probing it + /// (§5.3) — `ls` shows which files are DV and which profile. + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Sdr => formatter.write_str("SDR"), + Self::Hdr10 => formatter.write_str("HDR10"), + Self::Hdr10Plus => formatter.write_str("HDR10+"), + Self::Hlg => formatter.write_str("HLG"), + Self::DolbyVision(dv) => match dv.compatibility_id { + Some(compatibility_id) => write!(formatter, "DV{}.{compatibility_id}", dv.profile), + None => write!(formatter, "DV{}", dv.profile), + }, + } + } +} + #[derive(Clone, Debug, Eq, PartialEq)] pub enum RequiredAudio { OriginalLanguage, @@ -258,6 +321,24 @@ pub enum Rule { Other(String), } +impl Rule { + /// The stable name shared by `releases.rejected_rule`, blacklist reasons + /// and `media_files.waiver`, so one rule reads the same everywhere. + #[must_use] + pub fn name(&self) -> String { + match self { + Self::RequiredAudio => "required_audio".into(), + Self::DubBlacklist(_) => "dub_blacklist".into(), + Self::PortugueseUnverified => "portuguese_unverified".into(), + Self::DolbyVisionProfile(_) => "dolby_vision_profile".into(), + Self::Resolution(_) => "resolution".into(), + Self::Source(_) => "source".into(), + Self::Size => "size".into(), + Self::Other(name) => name.clone(), + } + } +} + #[derive(Clone, Debug, Eq, PartialEq)] pub enum Verdict { Eligible, diff --git a/crates/arr-daemon/Cargo.toml b/crates/arr-daemon/Cargo.toml index 564aaf7..242635e 100644 --- a/crates/arr-daemon/Cargo.toml +++ b/crates/arr-daemon/Cargo.toml @@ -19,6 +19,7 @@ arr-dl = { workspace = true } arr-indexer = { workspace = true } arr-meta = { workspace = true } arr-parse = { workspace = true } +arr-probe = { workspace = true } axum = { workspace = true } chrono = { workspace = true } include_dir = { workspace = true } diff --git a/crates/arr-daemon/src/grab.rs b/crates/arr-daemon/src/grab.rs index c654f13..9c0de1b 100644 --- a/crates/arr-daemon/src/grab.rs +++ b/crates/arr-daemon/src/grab.rs @@ -13,12 +13,12 @@ //! - the `grabs` row is written from that response, so a crash between the add //! and the insert heals on the next tick instead of leaving an orphan. -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::path::PathBuf; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use arr_core::policy::{evaluate, Candidate}; -use arr_core::{score::score, Language, Policy, Rule, TitleOverrides, Verdict}; +use arr_core::{score::score, Language, Policy, TitleOverrides, Verdict}; use arr_db::{Db, MoviePolicy}; use arr_dl::{AddTorrent, TorrentSource, TransmissionClient}; use arr_indexer::{ProwlarrClient, SearchRelease, SearchRequest}; @@ -317,7 +317,14 @@ impl GrabAction { let candidates = self .search(database, movie, indexers, &loaded, &original_language) .await?; - let Some(winner) = candidates.into_iter().next() else { + // §6.3: anything that hard-failed post-ffprobe is never grabbed + // again. The same release reappears under new infohashes, so the key + // is the normalised name. + let blacklisted = blacklisted_names(database).await?; + let Some(winner) = candidates + .into_iter() + .find(|candidate| !blacklisted.contains(&arr_parse::normalise(&candidate.name))) + else { tracing::info!( movie_id = movie.id, title = movie.title, @@ -558,6 +565,17 @@ async fn store_release( })) } +/// Every blacklist key (§6.3), for filtering candidates. Household scale: a +/// handful of rows, refetched per title rather than cached. +async fn blacklisted_names(database: &Db) -> Result, GrabError> { + let names = sqlx::query_scalar!( + r#"SELECT normalised_name AS "normalised_name!: String" FROM blacklist"# + ) + .fetch_all(database.pool()) + .await?; + Ok(names.into_iter().collect()) +} + /// Record that the title was searched, so the next tick takes a different one. async fn record_search(database: &Db, movie_id: i64) -> Result<(), GrabError> { sqlx::query!( @@ -605,20 +623,7 @@ fn verdict_columns(verdict: &Verdict) -> (&'static str, Option) { match verdict { Verdict::Eligible => ("eligible", None), Verdict::Waived(_) => ("waived", None), - Verdict::Rejected(rule) => ("rejected", Some(rule_name(rule))), - } -} - -fn rule_name(rule: &Rule) -> String { - match rule { - Rule::RequiredAudio => "required_audio".into(), - Rule::DubBlacklist(_) => "dub_blacklist".into(), - Rule::PortugueseUnverified => "portuguese_unverified".into(), - Rule::DolbyVisionProfile(_) => "dolby_vision_profile".into(), - Rule::Resolution(_) => "resolution".into(), - Rule::Source(_) => "source".into(), - Rule::Size => "size".into(), - Rule::Other(name) => name.clone(), + Verdict::Rejected(rule) => ("rejected", Some(rule.name())), } } @@ -990,6 +995,33 @@ mod tests { assert!(fake.torrents().is_empty()); } + /// §6.3: a hard-failed release is never grabbed again, even when it is + /// still the best-scoring candidate — the next one down wins instead. + #[tokio::test] + async fn a_blacklisted_release_is_passed_over() { + let (_dir, database) = wanted_movie().await; + sqlx::query("INSERT INTO blacklist (infohash, normalised_name, reason) VALUES (?, ?, ?)") + .bind("ffff") + .bind(arr_parse::normalise( + "Dune.Part.Two.2024.2160p.WEB-DL.DDP5.1.Atmos", + )) + .bind("dolby_vision_profile") + .execute(database.pool()) + .await + .unwrap(); + let indexer = prowlarr().await; + let (downloader, fake) = transmission().await; + + action(&indexer, &downloader).tick(&database).await.unwrap(); + + assert_eq!(fake.torrents().len(), 1); + assert!( + fake.torrents()[0].source.ends_with("huge.torrent"), + "the remux wins once the WEB-DL is blacklisted: {}", + fake.torrents()[0].source + ); + } + /// §6.3: `blocked` stops targeted search for a title. #[tokio::test] async fn a_blocked_title_is_not_searched() { diff --git a/crates/arr-daemon/src/import.rs b/crates/arr-daemon/src/import.rs new file mode 100644 index 0000000..bd716ac --- /dev/null +++ b/crates/arr-daemon/src/import.rs @@ -0,0 +1,991 @@ +//! The import pipeline: probe a completed download, judge it against the +//! policy a second time with real evidence, and link it into the library. See +//! DESIGN.md §5.7, §7.2, §7.3 and §7.4. +//! +//! The torrent's own files are never moved, renamed or deleted — the torrent +//! and the library entry are separate lifecycles (§7.3). A hard-failed +//! release is blacklisted and its grab marked failed, but the torrent keeps +//! seeding until Transmission's own limits clear it. +//! +//! Everything here is idempotent from domain rows (§8): a grab in +//! `downloaded` with no import recorded is the gap, and re-running any prefix +//! of the pipeline after a crash converges — the hardlink call tolerates the +//! destination already existing, and the `media_files` insert upserts on +//! path. + +use std::collections::HashMap; +use std::ffi::OsString; +use std::io; +use std::path::{Component, Path, PathBuf}; + +use arr_core::layout; +use arr_core::policy::{evaluate, Candidate}; +use arr_core::{ProbedMedia, Rule, Source, Verdict}; +use arr_db::Db; +use arr_dl::TransmissionClient; +use arr_probe::Prober; + +use crate::reconcile::{Action, ActionFuture, Outcome}; + +/// A failure during one import tick. +#[derive(Debug, thiserror::Error)] +pub enum ImportError { + #[error("database: {0}")] + Database(#[from] sqlx::Error), + #[error("policy: {0}")] + Policy(#[from] arr_db::PolicyError), + #[error("transmission: {0}")] + Transmission(#[from] arr_dl::Error), + #[error("probe: {0}")] + Probe(#[from] arr_probe::Error), + #[error("blocking task: {0}")] + Join(#[from] tokio::task::JoinError), + #[error("{action} {path}: {source}")] + Io { + action: &'static str, + path: PathBuf, + source: io::Error, + }, +} + +/// How the feature reached the library. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Placement { + Linked, + /// `link()` raised `EXDEV`; the file was copied instead (§7.2). + Copied, + /// The destination already existed — an earlier attempt placed it before + /// the process died. Never a partial file: copies land via rename. + AlreadyPlaced, +} + +/// What one probe attempt settled about one path. +#[derive(Debug, Clone)] +enum ProbeOutcome { + Media(Box), + /// A fact about the file — `.nfo`, artwork, corrupt — not the prober. + NotMedia, +} + +/// Imports every downloaded grab: probe, second policy pass, hardlink into +/// the §7.4 layout. +#[derive(Debug)] +pub struct ImportAction { + transmission: TransmissionClient, + prober: Prober, + /// Probe results by path, kept across ticks. The reconcile lane cancels + /// the whole action after its 25 s budget while one probe alone may take + /// up to 60 s, so without this a large multi-file torrent would restart + /// from the first file every tick and never finish. Shared with the + /// detached probe tasks, which is what lets a probe outlive a cancelled + /// tick and still deposit its result. Transient state, rebuilt by + /// re-probing after a restart (§8); entries are dropped once their grab + /// settles. + probed: std::sync::Arc>>, +} + +/// A movie grab Transmission finished downloading, not yet imported. +#[derive(Debug, Clone)] +struct PendingImport { + grab_id: i64, + infohash: String, + movie_id: i64, + tmdb_id: i64, + title: String, + year: Option, + original_language: Option, + release_name: String, +} + +impl ImportAction { + #[must_use] + pub fn new(transmission: TransmissionClient, prober: Prober) -> Self { + Self { + transmission, + prober, + probed: std::sync::Arc::new(tokio::sync::Mutex::new(HashMap::new())), + } + } + + /// Probe every path, reusing results settled on earlier ticks, and + /// return the readable video files. + /// + /// Each probe runs as a detached task that writes the cache itself, so a + /// tick cancelled by the lane's 25 s budget mid-probe loses nothing: the + /// probe finishes in the background (its own 60 s limit still applies) + /// and the next tick reads the deposited result. A tick that re-requests + /// a path an orphaned probe is still on starts a second probe of the same + /// file — bounded overlap, identical result, chosen over tracking + /// in-flight probes. + /// + /// Only facts about a file are cached; a prober failure — missing + /// binary, timeout, unparseable output — is returned so the tick retries. + async fn probe_all( + &self, + paths: &[PathBuf], + ) -> Result, ImportError> { + let mut files = Vec::new(); + for path in paths { + let cached = self.probed.lock().await.get(path).cloned(); + let outcome = if let Some(outcome) = cached { + outcome + } else { + let prober = self.prober.clone(); + let cache = std::sync::Arc::clone(&self.probed); + let target = path.clone(); + tokio::spawn(async move { + let outcome = match prober.probe(target.clone()).await { + Ok(file) => ProbeOutcome::Media(Box::new(file)), + Err(error) if error.is_about_the_file() => { + tracing::debug!(path = %target.display(), %error, "not a video file, skipping"); + ProbeOutcome::NotMedia + } + Err(error) => return Err(error), + }; + cache.lock().await.insert(target, outcome.clone()); + Ok(outcome) + }) + .await?? + }; + if let ProbeOutcome::Media(file) = outcome { + files.push(*file); + } + } + Ok(files) + } + + /// Drop a settled grab's probe results — imported or blacklisted, they + /// will not be needed again. + async fn forget_probes(&self, paths: &[PathBuf]) { + let mut probed = self.probed.lock().await; + for path in paths { + probed.remove(path); + } + } + + async fn tick(&self, database: &Db) -> Result, ImportError> { + let mut outcomes = Vec::new(); + for pending in pending_imports(database).await? { + match self.import_one(database, &pending).await { + Ok(Some(outcome)) => outcomes.push(outcome), + Ok(None) => {} + // One grab's failure must not cost the rest of the tick. + Err(error) => tracing::error!( + grab_id = pending.grab_id, + movie_id = pending.movie_id, + title = pending.title, + %error, + "import failed" + ), + } + } + Ok(outcomes) + } + + async fn import_one( + &self, + database: &Db, + pending: &PendingImport, + ) -> Result, ImportError> { + let Some(loaded) = database.movie_policy(pending.movie_id).await? else { + return Ok(None); + }; + // §5.2: no original language, nothing to judge audio against. + let Some(original_language) = pending.original_language.as_deref() else { + tracing::warn!( + movie_id = pending.movie_id, + title = pending.title, + "no original language yet; not importing" + ); + return Ok(None); + }; + let original_language = arr_db::policy::language(original_language); + + let Some(content) = self.transmission.torrent_content(&pending.infohash).await? else { + // Gone from Transmission. Whether that is a failure or a manual + // removal is issue #24's call; leave the grab alone. + tracing::warn!( + grab_id = pending.grab_id, + infohash = pending.infohash, + "downloaded grab has no torrent in Transmission; not importing" + ); + return Ok(None); + }; + // Torrent-declared names are untrusted input: an absolute or + // `..`-carrying entry would escape the download root and get probed — + // and possibly hardlinked — from anywhere on disk. + let paths: Vec = content + .files + .iter() + .filter_map(|file| { + let path = safe_join(&content.download_dir, &file.path); + if path.is_none() { + tracing::warn!( + grab_id = pending.grab_id, + path = %file.path.display(), + "torrent file path escapes the download root; skipping" + ); + } + path + }) + .collect(); + + // No expected runtime yet: the movies table carries no TMDB runtime, + // so feature selection is by size alone (largest readable video). + let mut candidates = self.probe_all(&paths).await?; + candidates.sort_by_key(|candidate| std::cmp::Reverse(candidate.size)); + let Some(feature) = candidates.into_iter().next() else { + // §5.7 "corrupt, wrong content": nothing in the torrent is a + // readable video file. + self.forget_probes(&paths).await; + return self + .hard_fail(database, pending, "no readable video file") + .await + .map(Some); + }; + + // §5.6 second phase of truth: same policy, real evidence. + let evaluation = evaluate( + &loaded.policy, + &loaded.overrides, + &original_language, + Candidate::PostDownload(&feature.media), + Some(feature.size), + ); + let waiver: Option = match evaluation.verdict { + Verdict::Rejected(rule) => { + self.forget_probes(&paths).await; + return self + .hard_fail(database, pending, &rule.name()) + .await + .map(Some); + } + Verdict::Waived(rule) => Some(rule), + Verdict::Eligible => None, + }; + + // The source tag is the one claim a file cannot verify (§5.6); every + // other tag comes from the probe. + let claimed_source = arr_parse::parse(&pending.release_name) + .source + .map(Source::from); + let tags = layout::attribute_tags(&feature.media, claimed_source); + let extension = feature.path.extension().and_then(|ext| ext.to_str()); + let folder = layout::movie_folder(&pending.title, pending.year, pending.tmdb_id); + let file_name = layout::movie_file_name( + &pending.title, + pending.year, + pending.tmdb_id, + &tags, + extension, + ); + let destination = Path::new(&loaded.root_path).join(folder).join(file_name); + + let source_path = feature.path.clone(); + let link_target = destination.clone(); + let placement = + tokio::task::spawn_blocking(move || place(&source_path, &link_target)).await??; + + record_import(database, pending, &feature, waiver.as_ref(), &destination).await?; + self.forget_probes(&paths).await; + let path_text = destination.to_string_lossy().into_owned(); + tracing::info!( + grab_id = pending.grab_id, + movie_id = pending.movie_id, + title = pending.title, + path = path_text, + placement = ?placement, + waived = waiver.is_some(), + "imported" + ); + Ok(Some(Outcome::new( + format!("grab {} downloaded, not imported", pending.grab_id), + format!("imported {path_text}"), + ))) + } + + /// §5.7 hard fail: blacklist the release, fail the grab, reopen the gap. + /// The torrent is deliberately untouched (§7.3). + async fn hard_fail( + &self, + database: &Db, + pending: &PendingImport, + reason: &str, + ) -> Result { + let normalised = arr_parse::normalise(&pending.release_name); + sqlx::query!( + "INSERT INTO blacklist (infohash, normalised_name, reason) + VALUES (?, ?, ?) + ON CONFLICT (infohash) DO NOTHING", + pending.infohash, + normalised, + reason + ) + .execute(database.pool()) + .await?; + sqlx::query!( + "UPDATE grabs SET state = 'failed' WHERE id = ?", + pending.grab_id + ) + .execute(database.pool()) + .await?; + sqlx::query!( + "UPDATE movies + SET state = 'missing', + updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') + WHERE id = ?", + pending.movie_id + ) + .execute(database.pool()) + .await?; + + tracing::warn!( + grab_id = pending.grab_id, + movie_id = pending.movie_id, + title = pending.title, + release = pending.release_name, + reason, + "hard fail post-probe; blacklisted, torrent left seeding" + ); + Ok(Outcome::new( + format!("grab {} hard-failed post-probe: {reason}", pending.grab_id), + format!("blacklisted {}", pending.release_name), + )) + } +} + +impl Action for ImportAction { + fn name(&self) -> &'static str { + "import" + } + + fn run<'a>(&'a self, database: &'a Db) -> ActionFuture<'a> { + Box::pin(async move { self.tick(database).await.map_err(Into::into) }) + } +} + +/// Settle a placed file into the rows: the `media_files` record (§4), the +/// grab and the movie. The upsert on path is the crash seam — a re-run after +/// a death between the link and here converges instead of erroring. +async fn record_import( + database: &Db, + pending: &PendingImport, + feature: &arr_probe::ProbedFile, + waiver: Option<&Rule>, + destination: &Path, +) -> Result<(), ImportError> { + let probed = probed_json(&feature.media).to_string(); + let waiver_json = waiver.map(|rule| serde_json::json!({ "rule": rule.name() }).to_string()); + let size = i64::try_from(feature.size).unwrap_or(i64::MAX); + let path_text = destination.to_string_lossy().into_owned(); + sqlx::query!( + "INSERT INTO media_files (owner_kind, owner_id, path, size, probed, waiver) + VALUES ('movie', ?, ?, ?, ?, ?) + ON CONFLICT (path) DO UPDATE SET + size = excluded.size, + probed = excluded.probed, + waiver = excluded.waiver, + updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')", + pending.movie_id, + path_text, + size, + probed, + waiver_json + ) + .execute(database.pool()) + .await?; + sqlx::query!( + "UPDATE grabs + SET state = 'imported', + imported_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') + WHERE id = ?", + pending.grab_id + ) + .execute(database.pool()) + .await?; + sqlx::query!( + "UPDATE movies + SET state = 'imported', + updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') + WHERE id = ?", + pending.movie_id + ) + .execute(database.pool()) + .await?; + Ok(()) +} + +/// The gap, straight out of the domain rows (§8): a movie grab Transmission +/// finished that no import has settled. +async fn pending_imports(database: &Db) -> Result, ImportError> { + let rows = sqlx::query!( + r#" + SELECT g.id AS "grab_id!: i64", + g.infohash AS "infohash!: String", + m.id AS "movie_id!: i64", + m.tmdb_id AS "tmdb_id!: i64", + m.title AS "title!: String", + m.year, + m.original_language, + r.name AS "release_name!: String" + FROM grabs g + JOIN movies m ON m.id = g.target_id + JOIN releases r ON r.id = g.release_id + WHERE g.state = 'downloaded' AND g.target_kind = 'movie' + ORDER BY g.id + "# + ) + .fetch_all(database.pool()) + .await?; + + Ok(rows + .into_iter() + .map(|row| PendingImport { + grab_id: row.grab_id, + infohash: row.infohash, + movie_id: row.movie_id, + tmdb_id: row.tmdb_id, + title: row.title, + year: row.year, + original_language: row.original_language, + release_name: row.release_name, + }) + .collect()) +} + +/// The `probed` column (§4, §5.6): what `ffprobe` found, in the spellings the +/// policy columns use. +fn probed_json(media: &ProbedMedia) -> serde_json::Value { + serde_json::json!({ + "resolution": media.resolution.to_string(), + "source": media.source.map(|source| source.to_string()), + "hdr": media.hdr.to_string(), + "audio_tracks": media + .audio_tracks + .iter() + .map(|track| serde_json::json!({ + "language": track.language.to_string(), + "title": track.title, + "handler_name": track.handler_name, + })) + .collect::>(), + "sub_tracks": media + .subtitle_tracks + .iter() + .map(|track| serde_json::json!({ "language": track.language.to_string() })) + .collect::>(), + }) +} + +/// Join a torrent-declared file path onto the download root, refusing +/// anything that could land outside it: absolute paths, drive prefixes and +/// `..` components. `None` means the entry is hostile or malformed. +fn safe_join(root: &Path, declared: &Path) -> Option { + let mut clean = PathBuf::new(); + for component in declared.components() { + match component { + Component::Normal(part) => clean.push(part), + Component::CurDir => {} + Component::RootDir | Component::Prefix(_) | Component::ParentDir => return None, + } + } + if clean.as_os_str().is_empty() { + return None; + } + Some(root.join(clean)) +} + +/// Hardlink `source` to `destination`, falling back to copy on `EXDEV` only +/// (§7.2). No configuration flag. +fn place(source: &Path, destination: &Path) -> Result { + if let Some(parent) = destination.parent() { + std::fs::create_dir_all(parent).map_err(|error| ImportError::Io { + action: "create library folder", + path: parent.to_path_buf(), + source: error, + })?; + } + + match std::fs::hard_link(source, destination) { + Ok(()) => Ok(Placement::Linked), + // A completed earlier attempt: links and copies both land whole + // (copies via rename), so an existing destination is a finished one. + Err(error) if error.kind() == io::ErrorKind::AlreadyExists => Ok(Placement::AlreadyPlaced), + Err(error) if error.kind() == io::ErrorKind::CrossesDevices => { + copy_into_place(source, destination) + } + Err(error) => Err(ImportError::Io { + action: "hardlink into", + path: destination.to_path_buf(), + source: error, + }), + } +} + +/// Copy through a dot-name in the destination folder, then rename, so the +/// library never shows a partial file. +fn copy_into_place(source: &Path, destination: &Path) -> Result { + let mut temp_name = OsString::from("."); + temp_name.push(destination.file_name().unwrap_or_default()); + temp_name.push(".partial"); + let temp = destination.with_file_name(temp_name); + + let copied = std::fs::copy(source, &temp).and_then(|_| std::fs::rename(&temp, destination)); + if let Err(error) = copied { + let _ = std::fs::remove_file(&temp); + return Err(ImportError::Io { + action: "copy into", + path: destination.to_path_buf(), + source: error, + }); + } + Ok(Placement::Copied) +} + +#[cfg(test)] +#[allow(clippy::unwrap_used)] +mod tests { + use std::os::unix::fs::{MetadataExt, PermissionsExt}; + use std::path::PathBuf; + + use serde_json::json; + use wiremock::matchers::method; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + use super::*; + + const INFOHASH: &str = "0123456789abcdef0123456789abcdef01234567"; + const RELEASE_NAME: &str = "Dune.Part.Two.2024.2160p.WEB-DL.DDP5.1.Atmos"; + + /// A 2160p HDR10 file with an English track — what the seeded main-movies + /// policy accepts. The size is the container's claim, matching §5.5's + /// band; the bytes on disk are tiny. + const HDR10_PROBE: &str = r#"{ + "format": {"format_name": "matroska,webm", "duration": "9060.0", "size": "23622320128"}, + "streams": [ + {"codec_type": "video", "codec_name": "hevc", "width": 3840, "height": 1600, + "color_transfer": "smpte2084"}, + {"codec_type": "audio", "codec_name": "eac3", "tags": {"language": "eng"}} + ] + }"#; + + /// The same file as a Dolby Vision Profile 5 stream — §5.3's hard reject. + const DV5_PROBE: &str = r#"{ + "format": {"format_name": "matroska,webm", "duration": "9060.0", "size": "23622320128"}, + "streams": [ + {"codec_type": "video", "codec_name": "hevc", "width": 3840, "height": 1600, + "color_transfer": "smpte2084", + "side_data_list": [{"side_data_type": "DOVI configuration record", "dv_profile": 5}]}, + {"codec_type": "audio", "codec_name": "eac3", "tags": {"language": "eng"}} + ] + }"#; + + struct Harness { + _dir: tempfile::TempDir, + database: Db, + downloads: PathBuf, + library: PathBuf, + action: ImportAction, + _server: MockServer, + } + + /// An `ffprobe` stand-in: canned JSON for media, a `tty` document for the + /// `.nfo`, so feature selection sees what the real binary would report. + fn fake_ffprobe(directory: &Path, media_json: &str) -> PathBuf { + let path = directory.join("ffprobe"); + let script = format!( + "#!/bin/sh\nfor arg; do last=\"$arg\"; done\ncase \"$last\" in\n *.nfo) printf '%s' '{{\"format\":{{\"format_name\":\"tty\"}}}}' ;;\n *) cat <<'PROBE_EOF'\n{media_json}\nPROBE_EOF\n;;\nesac\n" + ); + std::fs::write(&path, script).unwrap(); + let mut permissions = std::fs::metadata(&path).unwrap().permissions(); + permissions.set_mode(0o755); + std::fs::set_permissions(&path, permissions).unwrap(); + path + } + + async fn harness(media_json: &str) -> Harness { + harness_with( + media_json, + json!([ + {"name": "Dune/Dune.mkv", "length": 13, "bytesCompleted": 13}, + {"name": "Dune/Dune.nfo", "length": 10, "bytesCompleted": 10} + ]), + ) + .await + } + + async fn harness_with(media_json: &str, files: serde_json::Value) -> Harness { + let dir = tempfile::tempdir().unwrap(); + let downloads = dir.path().join("downloads"); + let library = dir.path().join("library"); + std::fs::create_dir_all(downloads.join("Dune")).unwrap(); + std::fs::create_dir_all(&library).unwrap(); + std::fs::write(downloads.join("Dune/Dune.mkv"), b"feature bytes").unwrap(); + std::fs::write(downloads.join("Dune/Dune.nfo"), b"not a film").unwrap(); + + let database = Db::connect(dir.path().join("arr.db")).await.unwrap(); + database.migrate().await.unwrap(); + let library_text = library.to_string_lossy().into_owned(); + sqlx::query("UPDATE roots SET path = ? WHERE kind = 'movie' AND audience = 'main'") + .bind(&library_text) + .execute(database.pool()) + .await + .unwrap(); + sqlx::query( + "INSERT INTO movies (tmdb_id, title, year, original_language, root_id, state) + SELECT 693134, 'Dune: Part Two', 2024, 'en', id, 'grabbed' + FROM roots WHERE kind = 'movie' AND audience = 'main'", + ) + .execute(database.pool()) + .await + .unwrap(); + let release_id = sqlx::query( + "INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed, verdict) + VALUES (7, 'good', ?, 23622320128, 'magnet:x', '{}', 'eligible')", + ) + .bind(RELEASE_NAME) + .execute(database.pool()) + .await + .unwrap() + .last_insert_rowid(); + sqlx::query( + "INSERT INTO grabs (release_id, target_kind, target_id, infohash, state) + VALUES (?, 'movie', 1, ?, 'downloaded')", + ) + .bind(release_id) + .bind(INFOHASH) + .execute(database.pool()) + .await + .unwrap(); + + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "result": "success", + "arguments": {"torrents": [{ + "hashString": INFOHASH, + "downloadDir": downloads.to_string_lossy(), + "files": files + }]} + }))) + .mount(&server) + .await; + + let prober = Prober::new().with_binary(fake_ffprobe(dir.path(), media_json)); + let action = ImportAction::new(TransmissionClient::new(&server.uri()).unwrap(), prober); + + Harness { + _dir: dir, + database, + downloads, + library, + action, + _server: server, + } + } + + fn expected_library_file(library: &Path) -> PathBuf { + library + .join("Dune Part Two (2024) [tmdbid-693134]") + .join("Dune Part Two (2024) [tmdbid-693134] - [2160p][WEB-DL][HDR10].mkv") + } + + /// The issue's acceptance case: the exact §7.4 path exists, and the + /// torrent's own file still exists with a link count of two. + #[tokio::test] + async fn the_feature_lands_on_the_design_layout_and_keeps_seeding() { + let h = harness(HDR10_PROBE).await; + + let outcomes = h.action.tick(&h.database).await.unwrap(); + + assert_eq!(outcomes.len(), 1); + let library_file = expected_library_file(&h.library); + assert!(library_file.is_file(), "missing {}", library_file.display()); + + let seeding_file = h.downloads.join("Dune/Dune.mkv"); + let metadata = std::fs::metadata(&seeding_file).unwrap(); + assert_eq!(metadata.nlink(), 2, "§7.2: hardlinked, not moved or copied"); + + let (path, probed, waiver): (String, String, Option) = + sqlx::query_as("SELECT path, probed, waiver FROM media_files WHERE owner_kind = 'movie' AND owner_id = 1") + .fetch_one(h.database.pool()) + .await + .unwrap(); + assert_eq!(path, library_file.to_string_lossy()); + assert!(probed.contains("\"2160p\""), "{probed}"); + assert!(probed.contains("HDR10"), "{probed}"); + assert_eq!(waiver, None); + + let (grab_state, imported_at): (String, Option) = + sqlx::query_as("SELECT state, imported_at FROM grabs WHERE infohash = ?") + .bind(INFOHASH) + .fetch_one(h.database.pool()) + .await + .unwrap(); + assert_eq!(grab_state, "imported"); + assert!(imported_at.is_some()); + let movie_state: String = sqlx::query_scalar("SELECT state FROM movies WHERE id = 1") + .fetch_one(h.database.pool()) + .await + .unwrap(); + assert_eq!(movie_state, "imported"); + } + + /// §5.3 through §5.7: Profile 5 is a hard fail — blacklisted, grab + /// failed, gap reopened, and the torrent's files untouched. + #[tokio::test] + async fn a_dolby_vision_profile_5_file_hard_fails() { + let h = harness(DV5_PROBE).await; + + let outcomes = h.action.tick(&h.database).await.unwrap(); + + assert_eq!(outcomes.len(), 1); + assert!( + std::fs::read_dir(&h.library).unwrap().next().is_none(), + "nothing may reach the library" + ); + assert!(h.downloads.join("Dune/Dune.mkv").is_file(), "§7.3"); + + let (infohash, normalised, reason): (String, String, String) = + sqlx::query_as("SELECT infohash, normalised_name, reason FROM blacklist") + .fetch_one(h.database.pool()) + .await + .unwrap(); + assert_eq!(infohash, INFOHASH); + assert_eq!(normalised, arr_parse::normalise(RELEASE_NAME)); + assert_eq!(reason, "dolby_vision_profile"); + + let grab_state: String = sqlx::query_scalar("SELECT state FROM grabs") + .fetch_one(h.database.pool()) + .await + .unwrap(); + assert_eq!(grab_state, "failed"); + let movie_state: String = sqlx::query_scalar("SELECT state FROM movies WHERE id = 1") + .fetch_one(h.database.pool()) + .await + .unwrap(); + assert_eq!( + movie_state, "missing", + "the gap reopens for the next candidate" + ); + } + + /// §8: killed between the hardlink and the bookkeeping, a restart + /// converges instead of failing on the existing destination. + #[tokio::test] + async fn a_restart_after_the_link_converges() { + let h = harness(HDR10_PROBE).await; + h.action.tick(&h.database).await.unwrap(); + + // The crash: the file is placed, the database never heard. + sqlx::query("DELETE FROM media_files") + .execute(h.database.pool()) + .await + .unwrap(); + sqlx::query("UPDATE grabs SET state = 'downloaded', imported_at = NULL") + .execute(h.database.pool()) + .await + .unwrap(); + sqlx::query("UPDATE movies SET state = 'grabbed'") + .execute(h.database.pool()) + .await + .unwrap(); + + let outcomes = h.action.tick(&h.database).await.unwrap(); + + assert_eq!(outcomes.len(), 1); + let files: i64 = sqlx::query_scalar("SELECT count(*) FROM media_files") + .fetch_one(h.database.pool()) + .await + .unwrap(); + assert_eq!(files, 1); + let metadata = std::fs::metadata(h.downloads.join("Dune/Dune.mkv")).unwrap(); + assert_eq!(metadata.nlink(), 2, "no second link, no copy"); + } + + /// A settled tick is idle — an imported grab is not a gap. + #[tokio::test] + async fn a_second_tick_imports_nothing_new() { + let h = harness(HDR10_PROBE).await; + h.action.tick(&h.database).await.unwrap(); + + let outcomes = h.action.tick(&h.database).await.unwrap(); + + assert!(outcomes.is_empty()); + } + + /// Torrent-declared names are untrusted: absolute and `..`-carrying + /// entries are skipped, and the import proceeds from what remains. + #[tokio::test] + async fn hostile_torrent_paths_never_leave_the_download_root() { + let h = harness_with( + HDR10_PROBE, + json!([ + {"name": "../outside.mkv", "length": 13, "bytesCompleted": 13}, + {"name": "/tmp/absolute.mkv", "length": 13, "bytesCompleted": 13}, + {"name": "Dune/Dune.mkv", "length": 13, "bytesCompleted": 13} + ]), + ) + .await; + + let outcomes = h.action.tick(&h.database).await.unwrap(); + + assert_eq!(outcomes.len(), 1); + assert!(expected_library_file(&h.library).is_file()); + let files: i64 = sqlx::query_scalar("SELECT count(*) FROM media_files") + .fetch_one(h.database.pool()) + .await + .unwrap(); + assert_eq!(files, 1, "only the safe path is imported"); + } + + /// A torrent whose every entry escapes the root has nothing importable: + /// hard fail, not an escape. + #[tokio::test] + async fn a_torrent_of_only_hostile_paths_hard_fails() { + let h = harness_with( + HDR10_PROBE, + json!([{"name": "../../etc/passwd", "length": 13, "bytesCompleted": 13}]), + ) + .await; + + let outcomes = h.action.tick(&h.database).await.unwrap(); + + assert_eq!(outcomes.len(), 1); + let reason: String = sqlx::query_scalar("SELECT reason FROM blacklist") + .fetch_one(h.database.pool()) + .await + .unwrap(); + assert_eq!(reason, "no readable video file"); + assert!(std::fs::read_dir(&h.library).unwrap().next().is_none()); + } + + #[test] + fn safe_join_refuses_escapes_and_keeps_normal_paths() { + let root = Path::new("/downloads"); + assert_eq!( + safe_join(root, Path::new("Dune/./Dune.mkv")), + Some(PathBuf::from("/downloads/Dune/Dune.mkv")) + ); + assert_eq!(safe_join(root, Path::new("../outside.mkv")), None); + assert_eq!(safe_join(root, Path::new("Dune/../../outside.mkv")), None); + assert_eq!(safe_join(root, Path::new("/etc/passwd")), None); + assert_eq!(safe_join(root, Path::new("")), None); + } + + /// The reconcile lane cancels the action after 25 s while one probe may + /// take 60 s, so results settled on one tick must survive to the next — + /// otherwise a large torrent restarts from file one forever. + #[tokio::test] + async fn probe_results_are_reused_across_calls() { + let dir = tempfile::tempdir().unwrap(); + let media = dir.path().join("film.mkv"); + std::fs::write(&media, b"bytes").unwrap(); + let counter = dir.path().join("count"); + let script = dir.path().join("ffprobe"); + std::fs::write( + &script, + format!( + "#!/bin/sh\necho x >> {}\ncat <<'PROBE_EOF'\n{HDR10_PROBE}\nPROBE_EOF\n", + counter.display() + ), + ) + .unwrap(); + let mut permissions = std::fs::metadata(&script).unwrap().permissions(); + permissions.set_mode(0o755); + std::fs::set_permissions(&script, permissions).unwrap(); + + let action = ImportAction::new( + TransmissionClient::new("http://127.0.0.1:1").unwrap(), + Prober::new().with_binary(&script), + ); + let paths = vec![media]; + + action.probe_all(&paths).await.unwrap(); + action.probe_all(&paths).await.unwrap(); + assert_eq!( + std::fs::read_to_string(&counter).unwrap().lines().count(), + 1, + "the second call reuses the first call's result" + ); + + action.forget_probes(&paths).await; + action.probe_all(&paths).await.unwrap(); + assert_eq!( + std::fs::read_to_string(&counter).unwrap().lines().count(), + 2, + "a settled grab's entries are dropped" + ); + } + + /// The probe outlives a cancelled tick: the detached task deposits its + /// result after the caller's future is dropped, and the next tick reads + /// it instead of restarting the same probe forever. + #[tokio::test] + async fn a_cancelled_probe_still_deposits_its_result() { + let dir = tempfile::tempdir().unwrap(); + let media = dir.path().join("film.mkv"); + std::fs::write(&media, b"bytes").unwrap(); + let counter = dir.path().join("count"); + let script = dir.path().join("ffprobe"); + std::fs::write( + &script, + format!( + "#!/bin/sh\nsleep 0.3\necho x >> {}\ncat <<'PROBE_EOF'\n{HDR10_PROBE}\nPROBE_EOF\n", + counter.display() + ), + ) + .unwrap(); + let mut permissions = std::fs::metadata(&script).unwrap().permissions(); + permissions.set_mode(0o755); + std::fs::set_permissions(&script, permissions).unwrap(); + let action = ImportAction::new( + TransmissionClient::new("http://127.0.0.1:1").unwrap(), + Prober::new().with_binary(&script), + ); + let paths = vec![media]; + + // The lane's budget, in miniature: the tick is cancelled mid-probe. + let cancelled = tokio::time::timeout( + std::time::Duration::from_millis(50), + action.probe_all(&paths), + ) + .await; + assert!(cancelled.is_err()); + + // The detached probe finishes on its own and deposits the result. + tokio::time::sleep(std::time::Duration::from_millis(600)).await; + let files = action.probe_all(&paths).await.unwrap(); + assert_eq!(files.len(), 1); + assert_eq!( + std::fs::read_to_string(&counter).unwrap().lines().count(), + 1, + "the next tick reused the deposited result instead of re-probing" + ); + } + + /// The `EXDEV` fallback path lands whole files via rename (§7.2). + #[test] + fn the_copy_fallback_lands_a_whole_file_and_cleans_up() { + let dir = tempfile::tempdir().unwrap(); + let source = dir.path().join("source.mkv"); + std::fs::write(&source, b"feature bytes").unwrap(); + let destination = dir.path().join("library").join("feature.mkv"); + std::fs::create_dir_all(destination.parent().unwrap()).unwrap(); + + let placement = copy_into_place(&source, &destination).unwrap(); + + assert_eq!(placement, Placement::Copied); + assert_eq!(std::fs::read(&destination).unwrap(), b"feature bytes"); + assert!( + std::fs::read_dir(destination.parent().unwrap()) + .unwrap() + .all(|entry| !entry + .unwrap() + .file_name() + .to_string_lossy() + .contains("partial")), + "no partial file left behind" + ); + } +} diff --git a/crates/arr-daemon/src/main.rs b/crates/arr-daemon/src/main.rs index 4a407a8..0ae001c 100644 --- a/crates/arr-daemon/src/main.rs +++ b/crates/arr-daemon/src/main.rs @@ -2,6 +2,7 @@ mod config; mod grab; +mod import; pub mod reconcile; mod web; @@ -14,6 +15,7 @@ use arr_db::Db; use arr_meta::TmdbClient; use config::Config; use grab::{GrabAction, SeedingLimits}; +use import::ImportAction; use reconcile::{ReconcileLoop, Tick}; use tower_http::trace::TraceLayer; @@ -89,17 +91,17 @@ async fn run() -> Result<(), Error> { let database = Db::connect(&config.database_path).await?; database.migrate().await?; + let transmission = arr_dl::TransmissionClient::new(&config.transmission_url)?; let mut reconcile = ReconcileLoop::new(database.clone()); // Without a Prowlarr key nothing can be searched, so the grab lane stays // unregistered rather than failing a tick every 30 seconds. if let Some(key) = config.prowlarr_api_key.clone() { let prowlarr = arr_indexer::ProwlarrClient::new(config.prowlarr_url.clone(), key)?; - let transmission = arr_dl::TransmissionClient::new(&config.transmission_url)?; reconcile = reconcile.register( Tick::Reconcile, GrabAction::new( prowlarr, - transmission, + transmission.clone(), config.download_dir.clone(), SeedingLimits { ratio: config.seed_ratio_limit, @@ -110,6 +112,12 @@ async fn run() -> Result<(), Error> { } else { tracing::warn!("no Prowlarr API key: nothing will be grabbed"); } + // Grab before import, so a download that completes on this tick is + // imported on this tick. + reconcile = reconcile.register( + Tick::Reconcile, + ImportAction::new(transmission, arr_probe::Prober::new()), + ); // Jellyseerr's Radarr shim (DESIGN.md §9.4) reads the same database and // needs its own TMDB client for `movie/lookup`. diff --git a/crates/arr-dl/src/lib.rs b/crates/arr-dl/src/lib.rs index a2a59fd..644b4db 100644 --- a/crates/arr-dl/src/lib.rs +++ b/crates/arr-dl/src/lib.rs @@ -70,6 +70,24 @@ pub struct Torrent { pub labels: Vec, } +/// One file inside a torrent, as Transmission reports it. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct TorrentFile { + /// Path relative to the torrent's download directory, torrent folder + /// included. + pub path: PathBuf, + /// Size in bytes. + pub size: u64, +} + +/// Where one torrent's data lives on disk. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct TorrentContent { + pub hash: String, + pub download_dir: PathBuf, + pub files: Vec, +} + /// A Transmission RPC failure. #[derive(Debug, thiserror::Error)] pub enum Error { @@ -218,6 +236,46 @@ impl TransmissionClient { Ok(response.torrents.into_iter().map(Torrent::from).collect()) } + /// The file list and download directory of one torrent, by infohash. + /// + /// `None` when Transmission no longer knows the hash — the operator may + /// have removed the torrent by hand. + /// + /// # Errors + /// + /// Returns an error for transport failures or rejected/malformed RPC + /// responses. + pub async fn torrent_content(&self, hash: &str) -> Result, Error> { + let arguments = self + .call( + "torrent-get", + json!({ + "ids": [hash], + "fields": ["hashString", "downloadDir", "files"] + }), + ) + .await?; + let response: RpcContentList = serde_json::from_value(arguments) + .map_err(|error| Error::InvalidResponse(error.to_string()))?; + + Ok(response + .torrents + .into_iter() + .next() + .map(|torrent| TorrentContent { + hash: torrent.hash, + download_dir: torrent.download_dir, + files: torrent + .files + .into_iter() + .map(|file| TorrentFile { + path: file.name, + size: file.length, + }) + .collect(), + })) + } + /// Remove one torrent, optionally deleting its downloaded data. /// /// # Errors @@ -291,6 +349,27 @@ struct RpcTorrentList { torrents: Vec, } +#[derive(Debug, Deserialize)] +struct RpcContentList { + torrents: Vec, +} + +#[derive(Debug, Deserialize)] +struct RpcContent { + #[serde(rename = "hashString")] + hash: String, + #[serde(rename = "downloadDir")] + download_dir: PathBuf, + #[serde(default)] + files: Vec, +} + +#[derive(Debug, Deserialize)] +struct RpcFile { + name: PathBuf, + length: u64, +} + #[derive(Debug, Deserialize)] struct RpcTorrent { id: i64, @@ -459,6 +538,61 @@ mod tests { assert_eq!(torrents[0].state, TorrentState::Unknown(99)); } + #[tokio::test] + async fn torrent_content_lists_files_and_download_dir() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(body_partial_json(json!({ + "method": "torrent-get", + "arguments": {"ids": ["abc"]} + }))) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "result": "success", + "arguments": {"torrents": [{ + "hashString": "abc", + "downloadDir": "/downloads", + "files": [ + {"name": "Movie/Movie.mkv", "length": 100, "bytesCompleted": 100}, + {"name": "Movie/Movie.nfo", "length": 5, "bytesCompleted": 5} + ] + }]} + }))) + .expect(1) + .mount(&server) + .await; + + let client = TransmissionClient::new(&server.uri()).expect("client"); + let content = client + .torrent_content("abc") + .await + .expect("content") + .expect("torrent known"); + + assert_eq!(content.download_dir, PathBuf::from("/downloads")); + assert_eq!(content.files.len(), 2); + assert_eq!(content.files[0].path, PathBuf::from("Movie/Movie.mkv")); + assert_eq!(content.files[0].size, 100); + } + + #[tokio::test] + async fn a_torrent_transmission_forgot_is_none_not_an_error() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "result": "success", + "arguments": {"torrents": []} + }))) + .mount(&server) + .await; + + let client = TransmissionClient::new(&server.uri()).expect("client"); + assert!(client + .torrent_content("gone") + .await + .expect("call succeeds") + .is_none()); + } + fn success() -> ResponseTemplate { ResponseTemplate::new(200).set_body_json(json!({"result": "success", "arguments": {}})) } diff --git a/crates/arr-parse/src/lib.rs b/crates/arr-parse/src/lib.rs index 53983aa..7485072 100644 --- a/crates/arr-parse/src/lib.rs +++ b/crates/arr-parse/src/lib.rs @@ -332,6 +332,27 @@ pub fn parse(name: &str) -> NameClaims { claims } +/// Collapse a release name to the blacklist key (`DESIGN.md` §6.3): lowercase, +/// every run of non-alphanumerics as one dot. The same release reappearing +/// with different separators or a different infohash still matches. +#[must_use] +pub fn normalise(name: &str) -> String { + let mut out = String::with_capacity(name.len()); + let mut pending_gap = false; + for character in name.chars() { + if character.is_alphanumeric() { + if pending_gap && !out.is_empty() { + out.push('.'); + } + pending_gap = false; + out.extend(character.to_lowercase()); + } else { + pending_gap = true; + } + } + out +} + fn apply_marker(claims: &mut NameClaims, sources: &mut Vec, m: Marker) { match m { Marker::Resolution(r) => { diff --git a/crates/arr-parse/tests/parse.rs b/crates/arr-parse/tests/parse.rs index 6b150ac..1078ca6 100644 --- a/crates/arr-parse/tests/parse.rs +++ b/crates/arr-parse/tests/parse.rs @@ -423,3 +423,15 @@ fn non_tv_names_claim_no_episode() { assert_eq!(parse(name).episode, None, "{name}"); } } + +#[test] +fn normalisation_is_separator_and_case_blind() { + let key = arr_parse::normalise("Dune.Part.Two.2024.2160p.WEB-DL.DDP5.1.Atmos"); + assert_eq!(key, "dune.part.two.2024.2160p.web.dl.ddp5.1.atmos"); + assert_eq!( + arr_parse::normalise("dune part two [2024] 2160p WEB DL DDP5 1 atmos "), + key, + "the same release under different separators is the same blacklist key" + ); + assert_eq!(arr_parse::normalise("!!!"), ""); +}