From 225be33b55b25510a9b8d557351383dd881acef9 Mon Sep 17 00:00:00 2001 From: "naps62-yolo (agent)" Date: Sun, 23 Aug 2026 01:04:15 +0100 Subject: [PATCH] feat: RSS sync worker (#91) --- ...ed1912b950a84d1ce7eaa406339b024aac82d.json | 20 + ...312f26fd0e57018ba6a4667d53b590080d4a7.json | 38 ++ crates/arr-api/src/search.rs | 1 + crates/arr-core/src/lib.rs | 2 + crates/arr-core/src/matching.rs | 278 +++++++++ crates/arr-daemon/src/grab.rs | 198 +++--- crates/arr-daemon/src/indexers.rs | 98 +++ crates/arr-daemon/src/main.rs | 118 ++-- crates/arr-daemon/src/rss.rs | 574 ++++++++++++++++++ crates/arr-daemon/tests/fixtures/rss.xml | 61 ++ crates/arr-indexer/src/search.rs | 43 ++ 11 files changed, 1285 insertions(+), 146 deletions(-) create mode 100644 .sqlx/query-7fa29678b941e5eb680db2dcb11ed1912b950a84d1ce7eaa406339b024aac82d.json create mode 100644 .sqlx/query-d318af0005db414990095d0d86e312f26fd0e57018ba6a4667d53b590080d4a7.json create mode 100644 crates/arr-core/src/matching.rs create mode 100644 crates/arr-daemon/src/indexers.rs create mode 100644 crates/arr-daemon/src/rss.rs create mode 100644 crates/arr-daemon/tests/fixtures/rss.xml diff --git a/.sqlx/query-7fa29678b941e5eb680db2dcb11ed1912b950a84d1ce7eaa406339b024aac82d.json b/.sqlx/query-7fa29678b941e5eb680db2dcb11ed1912b950a84d1ce7eaa406339b024aac82d.json new file mode 100644 index 0000000..f601041 --- /dev/null +++ b/.sqlx/query-7fa29678b941e5eb680db2dcb11ed1912b950a84d1ce7eaa406339b024aac82d.json @@ -0,0 +1,20 @@ +{ + "db_name": "SQLite", + "query": "SELECT original_language FROM movies WHERE id = ?", + "describe": { + "columns": [ + { + "name": "original_language", + "ordinal": 0, + "type_info": "Text" + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + true + ] + }, + "hash": "7fa29678b941e5eb680db2dcb11ed1912b950a84d1ce7eaa406339b024aac82d" +} diff --git a/.sqlx/query-d318af0005db414990095d0d86e312f26fd0e57018ba6a4667d53b590080d4a7.json b/.sqlx/query-d318af0005db414990095d0d86e312f26fd0e57018ba6a4667d53b590080d4a7.json new file mode 100644 index 0000000..e3dcf1c --- /dev/null +++ b/.sqlx/query-d318af0005db414990095d0d86e312f26fd0e57018ba6a4667d53b590080d4a7.json @@ -0,0 +1,38 @@ +{ + "db_name": "SQLite", + "query": "\n SELECT m.id AS \"id!: i64\",\n m.tmdb_id AS \"tmdb_id!: i64\",\n m.title AS \"title!: String\",\n m.year\n FROM movies m\n WHERE m.wanted = 1\n AND NOT EXISTS (\n SELECT 1 FROM media_files f\n WHERE f.owner_kind = 'movie' AND f.owner_id = m.id\n )\n AND NOT EXISTS (\n SELECT 1 FROM grabs g\n WHERE g.target_kind = 'movie' AND g.target_id = m.id\n AND g.state IN ('sent', 'downloaded', 'imported')\n )\n ORDER BY m.id\n ", + "describe": { + "columns": [ + { + "name": "id!: i64", + "ordinal": 0, + "type_info": "Integer" + }, + { + "name": "tmdb_id!: i64", + "ordinal": 1, + "type_info": "Integer" + }, + { + "name": "title!: String", + "ordinal": 2, + "type_info": "Text" + }, + { + "name": "year", + "ordinal": 3, + "type_info": "Integer" + } + ], + "parameters": { + "Right": 0 + }, + "nullable": [ + false, + false, + false, + true + ] + }, + "hash": "d318af0005db414990095d0d86e312f26fd0e57018ba6a4667d53b590080d4a7" +} diff --git a/crates/arr-api/src/search.rs b/crates/arr-api/src/search.rs index a9ecc38..5149e92 100644 --- a/crates/arr-api/src/search.rs +++ b/crates/arr-api/src/search.rs @@ -831,6 +831,7 @@ mod tests { seeders: Some(8), publish_date: None, download_url: "https://tracker/release".into(), + tmdb_id: None, }; let parsed = arr_parse::parse(&release.name); let core_score = score(&policy, Candidate::PreGrab(&parsed), 0, 8); diff --git a/crates/arr-core/src/lib.rs b/crates/arr-core/src/lib.rs index 32775a0..5bfe270 100644 --- a/crates/arr-core/src/lib.rs +++ b/crates/arr-core/src/lib.rs @@ -6,12 +6,14 @@ use std::{collections::BTreeMap, fmt, path::PathBuf, time::SystemTime}; pub mod lang; pub mod layout; +pub mod matching; pub mod policy; pub mod score; pub mod status; pub mod tracking; pub use arr_parse::NameClaims as ParsedRelease; +pub use matching::{match_movie, MatchKind, MovieMatch, WantedMovie}; pub use score::{Score, ScoreWeights}; pub use status::{derive_series_status, SeriesStatus}; diff --git a/crates/arr-core/src/matching.rs b/crates/arr-core/src/matching.rs new file mode 100644 index 0000000..b71cb79 --- /dev/null +++ b/crates/arr-core/src/matching.rs @@ -0,0 +1,278 @@ +//! Matching an RSS result against the wanted list (`DESIGN.md` §6.2). +//! +//! RSS is an empty-query feed: every item is offered to the whole wanted list +//! and nothing about the query says which title an item is for. That makes a +//! false positive expensive — it grabs the wrong film — and a false negative +//! nearly free, because the feed is read again ten minutes later and the +//! targeted search still runs. So every rule here is deliberately strict: +//! +//! - an ID the tracker itself supplied beats anything read off the name, and +//! an ID that matches nothing wanted ends the comparison rather than +//! falling back to the title; +//! - a title match needs the years to agree, and a wanted title with a known +//! year is never matched by a release that does not state one; +//! - anything that matches two wanted titles matches neither. + +use crate::MovieId; +use arr_parse::NameClaims; + +/// A wanted title, in the shape matching needs. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct WantedMovie { + pub id: MovieId, + /// TMDB id, compared against the one the tracker attached to the item. + pub tmdb_id: Option, + pub title: String, + pub year: Option, +} + +/// What tied a release to a wanted title. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum MatchKind { + /// The tracker supplied a TMDB id and it is one of ours. + TmdbId, + /// The parsed release name and the wanted title agree, years included. + TitleAndYear, +} + +/// One wanted title an RSS item was matched to. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct MovieMatch { + pub movie: MovieId, + pub kind: MatchKind, +} + +/// The wanted title an RSS item belongs to, or `None` when nothing matches +/// conservatively. +/// +/// `tmdb_id` is what the indexer attached to the item, not anything inferred +/// from the name; `claims` is the parsed release name. +#[must_use] +pub fn match_movie( + wanted: &[WantedMovie], + tmdb_id: Option, + claims: &NameClaims, +) -> Option { + if let Some(tmdb_id) = tmdb_id { + return unique( + wanted + .iter() + .filter(|movie| movie.tmdb_id == Some(tmdb_id)) + .map(|movie| MovieMatch { + movie: movie.id, + kind: MatchKind::TmdbId, + }), + ); + } + + // A season or episode tag makes this a TV release, whatever the title + // says. Movies are the only thing RSS grabs today (§13, phase 4). + if claims.episode.is_some() { + return None; + } + let title = match_key(claims.title.as_deref()?); + if title.is_empty() { + return None; + } + + unique( + wanted + .iter() + .filter(|movie| { + match_key(&movie.title) == title && years_agree(movie.year, claims.year) + }) + .map(|movie| MovieMatch { + movie: movie.id, + kind: MatchKind::TitleAndYear, + }), + ) +} + +/// A release with no year matches only a title that has none either: films +/// share titles across decades, and the remake is not the one that is wanted. +fn years_agree(wanted: Option, release: Option) -> bool { + match (wanted, release) { + (Some(wanted), Some(release)) => wanted == release, + (None, None) => true, + _ => false, + } +} + +/// Two matches are an ambiguity, and an ambiguity is not a match. +fn unique(mut matches: impl Iterator) -> Option { + let first = matches.next()?; + matches.next().is_none().then_some(first) +} + +/// The form two titles are compared in: lower case, accents folded to ASCII, +/// `&` spelled out, and every run of anything else one space. +/// +/// This absorbs the separators and punctuation a release name loses ("Dune: +/// Part Two" against `Dune.Part.Two`) without absorbing a word, so a longer +/// or shorter title stays a different title. +fn match_key(title: &str) -> String { + let mut key = String::with_capacity(title.len()); + let mut gap = false; + for character in title.chars() { + if let Some(folded) = fold(character) { + separate(&mut key, &mut gap); + key.push_str(folded); + } else if character.is_alphanumeric() { + separate(&mut key, &mut gap); + key.extend(character.to_lowercase()); + } else { + gap = !key.is_empty(); + } + } + key +} + +fn separate(key: &mut String, gap: &mut bool) { + if *gap { + key.push(' '); + } + *gap = false; +} + +/// The ASCII a letter is written as when a tracker cannot spell the accent, +/// and the word an `&` is spelled out as. `None` leaves the character to +/// [`match_key`], which keeps any other letter or digit as it is. +fn fold(character: char) -> Option<&'static str> { + Some(match character { + '&' => "and", + 'à' | 'á' | 'â' | 'ã' | 'ä' | 'å' | 'À' | 'Á' | 'Â' | 'Ã' | 'Ä' | 'Å' => "a", + 'ç' | 'Ç' => "c", + 'è' | 'é' | 'ê' | 'ë' | 'È' | 'É' | 'Ê' | 'Ë' => "e", + 'ì' | 'í' | 'î' | 'ï' | 'Ì' | 'Í' | 'Î' | 'Ï' => "i", + 'ñ' | 'Ñ' => "n", + 'ò' | 'ó' | 'ô' | 'õ' | 'ö' | 'ø' | 'Ò' | 'Ó' | 'Ô' | 'Õ' | 'Ö' | 'Ø' => "o", + 'ù' | 'ú' | 'û' | 'ü' | 'Ù' | 'Ú' | 'Û' | 'Ü' => "u", + 'ý' | 'ÿ' | 'Ý' => "y", + 'ß' => "ss", + 'æ' | 'Æ' => "ae", + _ => return None, + }) +} + +#[cfg(test)] +mod tests { + use super::{match_key, match_movie, MatchKind, MovieMatch, WantedMovie}; + use crate::MovieId; + + fn wanted() -> Vec { + vec![ + WantedMovie { + id: MovieId(1), + tmdb_id: Some(693_134), + title: "Dune: Part Two".to_owned(), + year: Some(2024), + }, + WantedMovie { + id: MovieId(2), + tmdb_id: Some(438_631), + title: "Dune".to_owned(), + year: Some(2021), + }, + WantedMovie { + id: MovieId(3), + tmdb_id: Some(194), + title: "Amélie".to_owned(), + year: Some(2001), + }, + ] + } + + fn matched(name: &str, tmdb_id: Option) -> Option { + match_movie(&wanted(), tmdb_id, &arr_parse::parse(name)) + } + + #[test] + fn a_supplied_id_matches_whatever_the_name_says() { + assert_eq!( + matched("Some.Scene.Name.2024.2160p.WEB-DL", Some(693_134)), + Some(MovieMatch { + movie: MovieId(1), + kind: MatchKind::TmdbId, + }) + ); + } + + /// The tracker said which film this is and it is not one of ours. Falling + /// back to the title here is how a wrong film gets grabbed. + #[test] + fn a_supplied_id_that_is_not_wanted_ends_the_comparison() { + assert_eq!(matched("Dune.Part.Two.2024.2160p.WEB-DL", Some(11)), None); + } + + #[test] + fn separators_punctuation_and_accents_do_not_block_a_title_match() { + assert_eq!( + matched("Dune.Part.Two.2024.2160p.WEB-DL.DDP5.1.Atmos-GROUP", None), + Some(MovieMatch { + movie: MovieId(1), + kind: MatchKind::TitleAndYear, + }) + ); + assert_eq!( + matched("Amelie.2001.1080p.BluRay.x264-GROUP", None), + Some(MovieMatch { + movie: MovieId(3), + kind: MatchKind::TitleAndYear, + }) + ); + } + + /// The near miss: one title is a prefix of the other, same year band, + /// same franchise. Nothing here may match. + #[test] + fn a_near_miss_does_not_match() { + assert_eq!(matched("Dune.2024.2160p.WEB-DL-GROUP", None), None); + assert_eq!(matched("Dune.Part.One.2021.1080p.WEB-DL", None), None); + assert_eq!(matched("Dune.Part.Two.2023.1080p.WEB-DL", None), None); + assert_eq!(matched("Dune.Prophecy.2024.1080p.WEB-DL", None), None); + } + + #[test] + fn a_release_with_no_year_is_not_matched_to_a_title_that_has_one() { + assert_eq!(matched("Dune.Part.Two.2160p.WEB-DL", None), None); + } + + #[test] + fn an_episode_tag_is_never_a_movie() { + let wanted = vec![WantedMovie { + id: MovieId(4), + tmdb_id: Some(1), + title: "Fallout".to_owned(), + year: Some(2024), + }]; + let claims = arr_parse::parse("Fallout.2024.S01E03.1080p.WEB-DL"); + assert_eq!(match_movie(&wanted, None, &claims), None); + } + + #[test] + fn two_wanted_titles_that_both_match_are_an_ambiguity() { + let wanted = vec![ + WantedMovie { + id: MovieId(5), + tmdb_id: Some(1), + title: "The Thing".to_owned(), + year: Some(1982), + }, + WantedMovie { + id: MovieId(6), + tmdb_id: Some(2), + title: "The Thing".to_owned(), + year: Some(1982), + }, + ]; + let claims = arr_parse::parse("The.Thing.1982.1080p.BluRay"); + assert_eq!(match_movie(&wanted, None, &claims), None); + } + + #[test] + fn ampersands_and_spelling_agree() { + assert_eq!(match_key("Fast & Furious"), "fast and furious"); + assert_eq!(match_key("Fast and Furious"), "fast and furious"); + assert_eq!(match_key(" Dune: Part Two "), "dune part two"); + } +} diff --git a/crates/arr-daemon/src/grab.rs b/crates/arr-daemon/src/grab.rs index e2c9744..c398895 100644 --- a/crates/arr-daemon/src/grab.rs +++ b/crates/arr-daemon/src/grab.rs @@ -16,7 +16,7 @@ use std::collections::HashMap; use std::path::PathBuf; use std::sync::Arc; -use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; +use std::time::{SystemTime, UNIX_EPOCH}; use arr_core::policy::{evaluate, Candidate}; use arr_core::{score::score, Language, Policy, TitleOverrides, Verdict}; @@ -25,6 +25,7 @@ use arr_dl::{AddTorrent, TorrentSource, TransmissionClient}; use arr_indexer::{ProwlarrClient, SearchRelease, SearchRequest}; use arr_meta::TmdbClient; +use crate::indexers::{DiscoveryError, IndexerDirectory}; use crate::reconcile::{Action, ActionFuture, Outcome}; /// How many titles one tick may search. The reconcile lane has a 25 s budget @@ -33,16 +34,6 @@ use crate::reconcile::{Action, ActionFuture, Outcome}; /// and release-date gating are issue #26. const MOVIES_PER_TICK: i64 = 5; -/// How long a discovered indexer list is reused. Prowlarr enumerates one -/// indexer per call plus a `t=caps` probe each, all sequential, so paying for -/// it every 30 s would leave the tick no room to search. -const INDEXER_CACHE_TTL: Duration = Duration::from_mins(15); - -/// The share of the reconcile lane's 25 s budget discovery may spend. One -/// unresponsive tracker must not cancel the whole action before a single -/// targeted search has run; past this the last known list is used instead. -const INDEXER_DISCOVERY_TIMEOUT: Duration = Duration::from_secs(8); - /// Seeding obligations, per tracker in principle (§7.3) and per install in /// practice until issue #25 gives them a home. Both are set on the torrent at /// add time and enforced by Transmission. @@ -79,8 +70,8 @@ pub enum GrabError { Database(#[from] sqlx::Error), #[error("policy: {0}")] Policy(#[from] arr_db::PolicyError), - #[error("prowlarr: {0}")] - Prowlarr(#[from] arr_indexer::Error), + #[error("{0}")] + Discovery(#[from] DiscoveryError), #[error("TMDB: {0}")] Metadata(#[from] arr_meta::Error), #[error("movie {0} has an invalid TMDB id")] @@ -92,15 +83,6 @@ pub enum GrabError { name: String, source: serde_json::Error, }, - #[error("indexer discovery timed out after {0:?} and none is known yet")] - IndexerDiscoveryTimeout(Duration), -} - -/// The searchable indexer ids, and when they were last discovered. -#[derive(Debug, Default)] -struct IndexerCache { - ids: Vec, - refreshed_at: Option, } /// Sends the best eligible release for every wanted movie that has neither a @@ -108,14 +90,9 @@ struct IndexerCache { #[derive(Debug)] pub struct GrabAction { prowlarr: ProwlarrClient, - transmission: TransmissionClient, - download_dir: PathBuf, - seeding: SeedingRules, + grabber: Grabber, tmdb: Option>, - indexers: tokio::sync::RwLock, - /// [`INDEXER_DISCOVERY_TIMEOUT`], overridden by tests that cannot wait - /// out the real one. Mirrors `ReconcileLoop`'s action timeout override. - discovery_timeout: Duration, + indexers: IndexerDirectory, } impl GrabAction { @@ -127,13 +104,10 @@ impl GrabAction { seeding: SeedingRules, ) -> Self { Self { + indexers: IndexerDirectory::new(prowlarr.clone()), prowlarr, - transmission, - download_dir, - seeding, + grabber: Grabber::new(transmission, download_dir, seeding), tmdb: None, - indexers: tokio::sync::RwLock::new(IndexerCache::default()), - discovery_timeout: INDEXER_DISCOVERY_TIMEOUT, } } @@ -165,7 +139,7 @@ impl GrabAction { if !search_due(&movie) { continue; } - let searchable = self.searchable_indexers().await?; + let searchable = self.indexers.searchable().await?; if searchable.is_empty() { tracing::warn!("no indexer advertises a text search; nothing can be grabbed"); return Ok(outcomes); @@ -290,6 +264,7 @@ impl GrabAction { } let torrents: HashMap = self + .grabber .transmission .list_torrents() .await? @@ -327,51 +302,6 @@ impl GrabAction { Ok(outcomes) } - /// The indexers that accept a text search, cached across ticks. - /// - /// Discovery is bounded and its result is reused, because Prowlarr probes - /// capabilities one indexer at a time: a single slow tracker would - /// otherwise burn the whole reconcile budget before any search runs. A - /// refresh that fails or times out keeps the previous list rather than - /// stopping the tick, and only an empty cache turns that into an error. - async fn searchable_indexers(&self) -> Result, GrabError> { - { - let cache = self.indexers.read().await; - if cache - .refreshed_at - .is_some_and(|at| at.elapsed() < INDEXER_CACHE_TTL) - { - return Ok(cache.ids.clone()); - } - } - - let discovered = - tokio::time::timeout(self.discovery_timeout, self.prowlarr.indexers()).await; - let mut cache = self.indexers.write().await; - match discovered { - Ok(Ok(indexers)) => { - cache.ids = indexers - .iter() - .filter(|indexer| indexer.capabilities.search.available) - .map(|indexer| indexer.id) - .collect(); - cache.refreshed_at = Some(Instant::now()); - } - Ok(Err(error)) if cache.ids.is_empty() => return Err(error.into()), - Err(_) if cache.ids.is_empty() => { - return Err(GrabError::IndexerDiscoveryTimeout(self.discovery_timeout)) - } - Ok(Err(error)) => { - tracing::warn!(%error, "indexer discovery failed; using the last known list"); - } - Err(_) => tracing::warn!( - timeout_seconds = self.discovery_timeout.as_secs(), - "indexer discovery timed out; using the last known list" - ), - } - Ok(cache.ids.clone()) - } - /// Search every indexer for one title, cache each candidate with its /// verdict and score (§9.3), and return the eligible ones best first. /// @@ -475,9 +405,67 @@ impl GrabAction { return Ok(None); }; - self.send_winner(database, movie, &loaded, &blacklist, winner) + self.grabber + .send_winner( + database, + &GrabTarget { + movie_id: movie.id, + title: &movie.title, + counts_as_attempt: true, + }, + &loaded, + &blacklist, + winner, + ) .await } +} + +/// Sending a chosen release to Transmission and recording the grab. +/// +/// Targeted search and RSS (§6.2) differ in how a title is chosen and in +/// whether a failed attempt counts toward a backoff; from the winning +/// release onward they are the same writes, so they share this. +#[derive(Debug)] +pub(crate) struct Grabber { + transmission: TransmissionClient, + download_dir: PathBuf, + seeding: SeedingRules, +} + +/// The title a winning release is being grabbed for. +#[derive(Debug)] +pub(crate) struct GrabTarget<'a> { + pub(crate) movie_id: i64, + pub(crate) title: &'a str, + /// Whether a grab that does not complete counts toward the targeted + /// search backoff (§6.2). RSS never backs off, so it passes `false`. + pub(crate) counts_as_attempt: bool, +} + +impl Grabber { + pub(crate) fn new( + transmission: TransmissionClient, + download_dir: PathBuf, + seeding: SeedingRules, + ) -> Self { + Self { + transmission, + download_dir, + seeding, + } + } + + async fn record_attempt( + &self, + database: &Db, + target: &GrabTarget<'_>, + ) -> Result<(), GrabError> { + if target.counts_as_attempt { + record_search(database, target.movie_id).await?; + } + Ok(()) + } /// Add the winning release to Transmission and record the grab. /// @@ -485,10 +473,10 @@ impl GrabAction { /// not a completed grab — a Transmission error or a blacklisted-infohash /// drop must not leave the same release to repeat next tick with no /// backoff. - async fn send_winner( + pub(crate) async fn send_winner( &self, database: &Db, - movie: &PendingMovie, + target: &GrabTarget<'_>, loaded: &MoviePolicy, blacklist: &Blacklist, winner: Eligible, @@ -507,7 +495,7 @@ impl GrabAction { { Ok(added) => added, Err(error) => { - record_search(database, movie.id).await?; + self.record_attempt(database, target).await?; return Err(error.into()); } }; @@ -517,8 +505,8 @@ impl GrabAction { // Transmission has fetched it, so the same blacklisted torrent can // reach here under a new name. if blacklist.blocks_infohash(&infohash) { - record_search(database, movie.id).await?; - self.drop_blacklisted_torrent(database, movie, &winner, &added) + self.record_attempt(database, target).await?; + self.drop_blacklisted_torrent(database, target, &winner, &added) .await?; return Ok(None); } @@ -531,7 +519,7 @@ impl GrabAction { ON CONFLICT (infohash) DO NOTHING RETURNING id AS "id!: i64""#, winner.id, - movie.id, + target.movie_id, infohash ) .fetch_optional(database.pool()) @@ -540,14 +528,14 @@ impl GrabAction { "UPDATE movies SET state = 'downloading', updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') WHERE id = ?", - movie.id + target.movie_id ) .execute(database.pool()) .await?; let Some(inserted) = inserted else { tracing::info!( - movie_id = movie.id, + movie_id = target.movie_id, infohash, "grab already recorded for this torrent" ); @@ -555,8 +543,8 @@ impl GrabAction { }; tracing::info!( - movie_id = movie.id, - title = movie.title, + movie_id = target.movie_id, + title = target.title, release = winner.name, score = winner.score, infohash, @@ -564,7 +552,7 @@ impl GrabAction { "grabbed" ); Ok(Some(Outcome::new( - format!("movie {} wanted with no file", movie.id), + format!("movie {} wanted with no file", target.movie_id), format!("grabbed {} as grab {}", winner.name, inserted.id), ))) } @@ -577,7 +565,7 @@ impl GrabAction { async fn drop_blacklisted_torrent( &self, database: &Db, - movie: &PendingMovie, + target: &GrabTarget<'_>, winner: &Eligible, added: &arr_dl::AddedTorrent, ) -> Result<(), GrabError> { @@ -604,7 +592,7 @@ impl GrabAction { // The earlier grab's torrent, still working off its seeding // obligation (§7.3). Nothing here deletes a torrent. tracing::warn!( - movie_id = movie.id, + movie_id = target.movie_id, release = release_name, infohash = added.hash, "blacklisted torrent re-listed under a new name; left seeding" @@ -614,7 +602,7 @@ impl GrabAction { // obligation and has nothing on disk worth keeping. self.transmission.remove_torrent(added.id, true).await?; tracing::warn!( - movie_id = movie.id, + movie_id = target.movie_id, release = release_name, infohash = added.hash, "blacklisted torrent re-listed under a new name; removed" @@ -652,13 +640,13 @@ struct PendingMovie { /// The eligible view of a stored release, ranked for selection. #[derive(Debug, Clone)] -struct Eligible { +pub(crate) struct Eligible { id: i64, - indexer_id: i64, - guid: String, - name: String, + pub(crate) indexer_id: i64, + pub(crate) guid: String, + pub(crate) name: String, download_url: String, - score: i64, + pub(crate) score: i64, } /// The gap, straight out of the domain rows (§8). @@ -759,7 +747,7 @@ fn is_digitally_released(digital_release: Option<&str>) -> bool { /// trigger (§6.3: "including by RSS") and so the stored row says why — a /// blacklisted release is rejected under the `blacklisted` rule, which is /// what stops §9.3's manual view from offering it as a clean match. -async fn store_release( +pub(crate) async fn store_release( database: &Db, movie_id: i64, release: &SearchRelease, @@ -919,6 +907,7 @@ fn rfc3339(time: SystemTime) -> Option { #[allow(clippy::unwrap_used)] mod tests { use std::sync::{Arc, Mutex}; + use std::time::Duration; use serde_json::{json, Value}; use wiremock::matchers::{method, path, query_param}; @@ -1697,14 +1686,17 @@ mod tests { let (downloader, fake) = transmission().await; let mut action = action(&indexer, &downloader); - action.discovery_timeout = Duration::from_millis(50); + action.indexers.discovery_timeout = Duration::from_millis(50); let error = action .tick(&database) .await .expect_err("discovery cannot complete"); - assert!(matches!(error, GrabError::IndexerDiscoveryTimeout(_))); + assert!(matches!( + error, + GrabError::Discovery(DiscoveryError::Timeout(_)) + )); assert!(fake.torrents().is_empty()); } diff --git a/crates/arr-daemon/src/indexers.rs b/crates/arr-daemon/src/indexers.rs new file mode 100644 index 0000000..cb6b2c0 --- /dev/null +++ b/crates/arr-daemon/src/indexers.rs @@ -0,0 +1,98 @@ +//! The searchable indexer list, discovered once and reused. See DESIGN.md +//! §6.1. +//! +//! Prowlarr enumerates one indexer per call and probes `t=caps` for each, +//! sequentially, so both the 30 s grab lane and the 10 min RSS lane would +//! spend most of their budget rediscovering the same list. Each lane keeps +//! its own directory: they run concurrently and neither should wait on the +//! other's HTTP calls. + +use std::time::{Duration, Instant}; + +use arr_indexer::ProwlarrClient; + +/// How long a discovered indexer list is reused. +const CACHE_TTL: Duration = Duration::from_mins(15); + +/// The share of a tick's budget discovery may spend. One unresponsive +/// tracker must not cancel the whole action before a single search has run; +/// past this the last known list is used instead. +const DISCOVERY_TIMEOUT: Duration = Duration::from_secs(8); + +/// A discovery that produced nothing usable at all. +#[derive(Debug, thiserror::Error)] +pub enum DiscoveryError { + #[error("prowlarr: {0}")] + Prowlarr(#[from] arr_indexer::Error), + #[error("indexer discovery timed out after {0:?} and none is known yet")] + Timeout(Duration), +} + +#[derive(Debug, Default)] +struct Cache { + ids: Vec, + refreshed_at: Option, +} + +/// The indexers that accept a text search, cached across ticks. +#[derive(Debug)] +pub struct IndexerDirectory { + prowlarr: ProwlarrClient, + cache: tokio::sync::RwLock, + /// [`DISCOVERY_TIMEOUT`], overridden by tests that cannot wait out the + /// real one. Mirrors `ReconcileLoop`'s action timeout override. + pub(crate) discovery_timeout: Duration, +} + +impl IndexerDirectory { + pub fn new(prowlarr: ProwlarrClient) -> Self { + Self { + prowlarr, + cache: tokio::sync::RwLock::new(Cache::default()), + discovery_timeout: DISCOVERY_TIMEOUT, + } + } + + /// The ids of every indexer advertising a text search — which is also + /// the RSS feed, an empty-query text search (§6.2). + /// + /// A refresh that fails or times out keeps the previous list rather than + /// stopping the tick, and only an empty cache turns that into an error. + pub async fn searchable(&self) -> Result, DiscoveryError> { + { + let cache = self.cache.read().await; + if cache + .refreshed_at + .is_some_and(|at| at.elapsed() < CACHE_TTL) + { + return Ok(cache.ids.clone()); + } + } + + let discovered = + tokio::time::timeout(self.discovery_timeout, self.prowlarr.indexers()).await; + let mut cache = self.cache.write().await; + match discovered { + Ok(Ok(indexers)) => { + cache.ids = indexers + .iter() + .filter(|indexer| indexer.capabilities.search.available) + .map(|indexer| indexer.id) + .collect(); + cache.refreshed_at = Some(Instant::now()); + } + Ok(Err(error)) if cache.ids.is_empty() => return Err(error.into()), + Err(_) if cache.ids.is_empty() => { + return Err(DiscoveryError::Timeout(self.discovery_timeout)) + } + Ok(Err(error)) => { + tracing::warn!(%error, "indexer discovery failed; using the last known list"); + } + Err(_) => tracing::warn!( + timeout_seconds = self.discovery_timeout.as_secs(), + "indexer discovery timed out; using the last known list" + ), + } + Ok(cache.ids.clone()) + } +} diff --git a/crates/arr-daemon/src/main.rs b/crates/arr-daemon/src/main.rs index 4ddd7ba..7cd64f0 100644 --- a/crates/arr-daemon/src/main.rs +++ b/crates/arr-daemon/src/main.rs @@ -3,8 +3,10 @@ mod config; mod grab; mod import; +mod indexers; mod reaper; pub mod reconcile; +mod rss; mod web; use std::process::ExitCode; @@ -19,6 +21,7 @@ use grab::{GrabAction, SeedingLimits, SeedingRules}; use import::ImportAction; use reaper::ReaperAction; use reconcile::{ReconcileLoop, Tick}; +use rss::RssAction; use tower_http::trace::TraceLayer; /// Dump the `OpenAPI` document and exit, instead of serving. `just gen-client` @@ -103,49 +106,7 @@ async fn run() -> Result<(), Error> { } else { None }; - 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), Some(tmdb)) = (config.prowlarr_api_key.clone(), tmdb.clone()) { - let prowlarr = arr_indexer::ProwlarrClient::new(config.prowlarr_url.clone(), key)?; - reconcile = reconcile.register( - Tick::Reconcile, - GrabAction::new( - prowlarr, - transmission.clone(), - config.download_dir.clone(), - SeedingRules::new( - SeedingLimits { - ratio: config.seed_ratio_limit, - idle_minutes: config.seed_idle_limit_minutes, - }, - config - .tracker_seeding - .iter() - .map(|(&id, rule)| { - ( - id, - SeedingLimits { - ratio: rule.ratio, - idle_minutes: rule.min_seed_time, - }, - ) - }) - .collect(), - ), - ) - .with_tmdb(tmdb), - ); - } else { - tracing::warn!("Prowlarr or TMDB is not configured: 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.clone(), arr_probe::Prober::new()), - ); - reconcile = reconcile.register(Tick::Reaper, ReaperAction::new(transmission)); + let reconcile = reconcile_loop(&database, &config, &transmission, tmdb.as_ref())?; // Jellyseerr's Radarr shim (DESIGN.md §9.4) reads the same database and // needs its own TMDB client for `movie/lookup`. @@ -202,6 +163,77 @@ async fn run() -> Result<(), Error> { } } +/// Wire the reconcile lanes (DESIGN.md §8). Grab and RSS both need a +/// Prowlarr key and grab needs TMDB as well; a lane whose upstream is not +/// configured stays unregistered rather than failing every tick. +fn reconcile_loop( + database: &Db, + config: &Config, + transmission: &arr_dl::TransmissionClient, + tmdb: Option<&Arc>, +) -> Result { + let mut reconcile = ReconcileLoop::new(database.clone()); + let seeding = SeedingRules::new( + SeedingLimits { + ratio: config.seed_ratio_limit, + idle_minutes: config.seed_idle_limit_minutes, + }, + config + .tracker_seeding + .iter() + .map(|(&id, rule)| { + ( + id, + SeedingLimits { + ratio: rule.ratio, + idle_minutes: rule.min_seed_time, + }, + ) + }) + .collect(), + ); + let prowlarr = config + .prowlarr_api_key + .clone() + .map(|key| arr_indexer::ProwlarrClient::new(config.prowlarr_url.clone(), key)) + .transpose()?; + + if let (Some(prowlarr), Some(tmdb)) = (prowlarr.as_ref(), tmdb) { + reconcile = reconcile.register( + Tick::Reconcile, + GrabAction::new( + prowlarr.clone(), + transmission.clone(), + config.download_dir.clone(), + seeding.clone(), + ) + .with_tmdb(Arc::clone(tmdb)), + ); + } else { + tracing::warn!("Prowlarr or TMDB is not configured: nothing will be grabbed"); + } + // RSS needs no TMDB: it matches what the feeds already carry against the + // wanted list (§6.2). + if let Some(prowlarr) = prowlarr { + reconcile = reconcile.register( + Tick::Rss, + RssAction::new( + prowlarr, + transmission.clone(), + config.download_dir.clone(), + seeding, + ), + ); + } + // Grab before import, so a download that completes on this tick is + // imported on this tick. + reconcile = reconcile.register( + Tick::Reconcile, + ImportAction::new(transmission.clone(), arr_probe::Prober::new()), + ); + Ok(reconcile.register(Tick::Reaper, ReaperAction::new(transmission.clone()))) +} + /// Stop accepting on Ctrl-C, or on the SIGTERM a service manager sends. async fn shutdown() { let interrupt = async { diff --git a/crates/arr-daemon/src/rss.rs b/crates/arr-daemon/src/rss.rs new file mode 100644 index 0000000..4007b7f --- /dev/null +++ b/crates/arr-daemon/src/rss.rs @@ -0,0 +1,574 @@ +//! RSS sync: one empty-query Torznab call per indexer, matched against the +//! whole wanted list locally. See DESIGN.md §6.2. +//! +//! The property this lane is built around is that its cost is one call per +//! indexer, whatever the library holds. Nothing here is per-title, so every +//! wanted item is compared against every result on every pass, forever: +//! there is no backoff, no attempt counter, and no release-date gate. Those +//! belong to targeted search, which pays per title. +//! +//! `blocked` is honoured the other way round from targeted search (§6.3): it +//! stops a title being searched for, and leaves it matching RSS. + +use std::collections::hash_map::Entry; +use std::collections::HashMap; +use std::path::PathBuf; + +use arr_core::matching::{match_movie, MatchKind, WantedMovie}; +use arr_core::{Language, MovieId}; +use arr_db::{Blacklist, Db, MoviePolicy}; +use arr_dl::TransmissionClient; +use arr_indexer::{ProwlarrClient, SearchRelease, SearchRequest}; + +use crate::grab::{store_release, Eligible, GrabError, GrabTarget, Grabber, SeedingRules}; +use crate::indexers::IndexerDirectory; +use crate::reconcile::{Action, ActionFuture, Outcome}; + +/// Reads every indexer's feed and grabs what it can tie to a wanted title. +#[derive(Debug)] +pub struct RssAction { + prowlarr: ProwlarrClient, + grabber: Grabber, + indexers: IndexerDirectory, +} + +impl RssAction { + #[must_use] + pub fn new( + prowlarr: ProwlarrClient, + transmission: TransmissionClient, + download_dir: PathBuf, + seeding: SeedingRules, + ) -> Self { + Self { + indexers: IndexerDirectory::new(prowlarr.clone()), + prowlarr, + grabber: Grabber::new(transmission, download_dir, seeding), + } + } + + async fn tick(&self, database: &Db) -> Result, GrabError> { + let wanted = wanted_movies(database).await?; + if wanted.is_empty() { + return Ok(Vec::new()); + } + let indexers = self.indexers.searchable().await?; + if indexers.is_empty() { + tracing::warn!("no indexer advertises a text search; no RSS feed to read"); + return Ok(Vec::new()); + } + + let releases = self.feeds(&indexers).await; + let blacklist = Blacklist::load(database.pool()).await?; + let winners = self + .match_feeds(database, &wanted, releases, &blacklist) + .await?; + + let mut outcomes = Vec::new(); + for (movie, winner) in winners { + let outcome = self + .grabber + .send_winner( + database, + &GrabTarget { + movie_id: movie.id, + title: &movie.title, + // §6.2: RSS never backs off, so a failed grab here + // must not spend one of targeted search's attempts. + counts_as_attempt: false, + }, + &movie.policy, + &blacklist, + winner, + ) + .await?; + outcomes.extend(outcome); + } + Ok(outcomes) + } + + /// One empty-query call per indexer (§6.2). A tracker that fails is + /// logged and skipped: the next pass is ten minutes away. + async fn feeds(&self, indexers: &[i64]) -> Vec { + let mut releases = Vec::new(); + for search in self + .prowlarr + .search_indexers(indexers, &SearchRequest::Rss) + .await + { + if let Some(error) = search.error { + tracing::warn!(indexer_id = search.indexer_id, %error, "RSS feed failed"); + } + releases.extend(search.releases); + } + releases + } + + /// Tie each feed item to a wanted title, cache it with its verdict, and + /// keep the best eligible candidate per title. + /// + /// One title takes at most one torrent per pass, and the ordering is the + /// same total order targeted search uses, so the two triggers agree on + /// which release wins. + async fn match_feeds( + &self, + database: &Db, + wanted: &[WantedMovie], + releases: Vec, + blacklist: &Blacklist, + ) -> Result, GrabError> { + let mut policies: HashMap> = HashMap::new(); + let mut best: HashMap = HashMap::new(); + + for release in releases { + let claims = arr_parse::parse(&release.name); + let Some(matched) = match_movie(wanted, release.tmdb_id, &claims) else { + continue; + }; + let movie_id = matched.movie.0; + let grabbable = match policies.entry(movie_id) { + Entry::Occupied(entry) => entry.into_mut(), + Entry::Vacant(entry) => { + entry.insert(load_grabbable(database, wanted, movie_id).await?) + } + }; + let Some(grabbable) = grabbable.as_ref() else { + continue; + }; + tracing::info!( + movie_id, + release = release.name, + by = match matched.kind { + MatchKind::TmdbId => "tmdb id", + MatchKind::TitleAndYear => "title and year", + }, + "RSS result matched a wanted title" + ); + let stored = store_release( + database, + movie_id, + &release, + &grabbable.policy.policy, + &grabbable.policy.overrides, + &grabbable.language, + blacklist, + ) + .await?; + let Some(candidate) = stored else { + continue; + }; + let incumbent = best.get(&movie_id); + if incumbent.is_none_or(|incumbent| beats(&candidate, incumbent)) { + best.insert(movie_id, candidate); + } + } + + let mut winners: Vec<(Grabbable, Eligible)> = best + .into_iter() + .filter_map(|(movie_id, candidate)| { + let grabbable = policies.get(&movie_id).and_then(Clone::clone)?; + Some((grabbable, candidate)) + }) + .collect(); + winners.sort_by_key(|(movie, _)| movie.id); + Ok(winners) + } +} + +impl Action for RssAction { + fn name(&self) -> &'static str { + "rss" + } + + fn run<'a>(&'a self, database: &'a Db) -> ActionFuture<'a> { + Box::pin(async move { self.tick(database).await.map_err(Into::into) }) + } +} + +/// A matched title with everything a grab needs loaded. +#[derive(Clone, Debug)] +struct Grabbable { + id: i64, + title: String, + policy: MoviePolicy, + language: Language, +} + +/// Better score first, then the tiebreaks targeted search sorts by, so both +/// triggers pick the same release out of the same candidates. +fn beats(candidate: &Eligible, incumbent: &Eligible) -> bool { + candidate + .score + .cmp(&incumbent.score) + .then_with(|| incumbent.indexer_id.cmp(&candidate.indexer_id)) + .then_with(|| incumbent.guid.cmp(&candidate.guid)) + == std::cmp::Ordering::Greater +} + +/// Every wanted title with neither a file nor a grab in flight. +/// +/// Unlike targeted search's work list this is not limited or ordered by when +/// a title was last searched: matching is local, so the cost of carrying the +/// whole list is a string comparison per title per feed item (§6.2). Blocked +/// titles are here on purpose (§6.3). +async fn wanted_movies(database: &Db) -> Result, GrabError> { + let rows = sqlx::query!( + r#" + SELECT m.id AS "id!: i64", + m.tmdb_id AS "tmdb_id!: i64", + m.title AS "title!: String", + m.year + FROM movies m + WHERE m.wanted = 1 + AND NOT EXISTS ( + SELECT 1 FROM media_files f + WHERE f.owner_kind = 'movie' AND f.owner_id = m.id + ) + AND NOT EXISTS ( + SELECT 1 FROM grabs g + WHERE g.target_kind = 'movie' AND g.target_id = m.id + AND g.state IN ('sent', 'downloaded', 'imported') + ) + ORDER BY m.id + "# + ) + .fetch_all(database.pool()) + .await?; + + Ok(rows + .into_iter() + .map(|row| WantedMovie { + id: MovieId(row.id), + tmdb_id: u32::try_from(row.tmdb_id).ok(), + title: row.title, + year: row.year.and_then(|year| u16::try_from(year).ok()), + }) + .collect()) +} + +/// The policy and original language a matched title needs, or `None` when +/// the title cannot be evaluated yet. +async fn load_grabbable( + database: &Db, + wanted: &[WantedMovie], + movie_id: i64, +) -> Result, GrabError> { + let Some(title) = wanted + .iter() + .find(|movie| movie.id == MovieId(movie_id)) + .map(|movie| movie.title.clone()) + else { + return Ok(None); + }; + let Some(policy) = database.movie_policy(movie_id).await? else { + return Ok(None); + }; + // §5.2: every language rule is expressed against the title's own + // original language, and guessing it is worse than not grabbing. + let language: Option = sqlx::query_scalar!( + "SELECT original_language FROM movies WHERE id = ?", + movie_id + ) + .fetch_optional(database.pool()) + .await? + .flatten(); + let Some(language) = language else { + tracing::warn!(movie_id, "no original language yet; not grabbing from RSS"); + return Ok(None); + }; + + Ok(Some(Grabbable { + id: movie_id, + title, + policy, + language: arr_db::policy::language(&language), + })) +} + +#[cfg(test)] +#[allow(clippy::unwrap_used)] +mod tests { + use std::sync::{Arc, Mutex}; + + use serde_json::{json, Value}; + use wiremock::matchers::{method, path, query_param, query_param_is_missing}; + use wiremock::{Mock, MockServer, Request, Respond, ResponseTemplate}; + + use super::*; + use crate::grab::SeedingLimits; + use arr_dl::TransmissionClient; + + /// The recorded feed: two releases of one wanted title, one that only a + /// TMDB id identifies, one near miss, one TV item and one film nobody + /// asked for. + const FEED: &str = include_str!("../tests/fixtures/rss.xml"); + + const INDEXERS: [i64; 2] = [7, 9]; + + /// Enough of Transmission to add a torrent and list nothing back. + #[derive(Clone, Debug, Default)] + struct FakeTransmission { + added: Arc>>, + } + + impl Respond for FakeTransmission { + fn respond(&self, request: &Request) -> ResponseTemplate { + let body: Value = serde_json::from_slice(&request.body).unwrap(); + match body["method"].as_str().unwrap_or_default() { + "torrent-add" => { + let source = body["arguments"]["filename"] + .as_str() + .unwrap_or_default() + .to_owned(); + let mut added = self.added.lock().unwrap(); + added.push(source.clone()); + let id = i64::try_from(added.len()).unwrap(); + success(&json!({"torrent-added": { + "id": id, "name": source, "hashString": format!("{id:040x}") + }})) + } + "torrent-get" => success(&json!({"torrents": []})), + _ => success(&json!({})), + } + } + } + + fn success(arguments: &Value) -> ResponseTemplate { + ResponseTemplate::new(200).set_body_json(json!({ + "result": "success", "arguments": arguments + })) + } + + /// Two indexers, both advertising a text search, both serving the same + /// feed to an empty query. + async fn prowlarr() -> MockServer { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/api/v1/indexer")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!([ + {"id": 7, "name": "alpha", "enable": true}, + {"id": 9, "name": "beta", "enable": true}, + ]))) + .mount(&server) + .await; + for id in INDEXERS { + Mock::given(method("GET")) + .and(path(format!("/{id}/api"))) + .and(query_param("t", "caps")) + .respond_with(ResponseTemplate::new(200).set_body_string( + r#""#, + )) + .mount(&server) + .await; + // The RSS call is `t=search` with no `q` at all (§6.2). + Mock::given(method("GET")) + .and(path(format!("/{id}/api"))) + .and(query_param("t", "search")) + .and(query_param_is_missing("q")) + .respond_with(ResponseTemplate::new(200).set_body_string(FEED)) + .mount(&server) + .await; + } + server + } + + async fn transmission() -> (MockServer, FakeTransmission) { + let server = MockServer::start().await; + let fake = FakeTransmission::default(); + Mock::given(method("POST")) + .respond_with(fake.clone()) + .mount(&server) + .await; + (server, fake) + } + + /// `(tmdb_id, title, year, blocked)`. + async fn wanted(movies: &[(i64, &str, i64, bool)]) -> (tempfile::TempDir, Db) { + let dir = tempfile::tempdir().unwrap(); + let database = Db::connect(dir.path().join("arr.db")).await.unwrap(); + database.migrate().await.unwrap(); + for &(tmdb_id, title, year, blocked) in movies { + sqlx::query( + "INSERT INTO movies (tmdb_id, title, year, original_language, blocked, root_id) + SELECT ?, ?, ?, 'en', ?, id + FROM roots WHERE kind = 'movie' AND audience = 'main'", + ) + .bind(tmdb_id) + .bind(title) + .bind(year) + .bind(i64::from(blocked)) + .execute(database.pool()) + .await + .unwrap(); + } + (dir, database) + } + + fn action(prowlarr: &MockServer, transmission: &MockServer) -> RssAction { + RssAction::new( + ProwlarrClient::new(prowlarr.uri(), "key").unwrap(), + TransmissionClient::new(&transmission.uri()).unwrap(), + PathBuf::from("/mnt/media/transmission/complete"), + SeedingRules::new( + SeedingLimits { + ratio: 1.0, + idle_minutes: 60, + }, + HashMap::new(), + ), + ) + } + + /// `(movie title, grabbed release name)` for every grab that was made. + async fn grabbed(database: &Db) -> Vec<(String, String)> { + sqlx::query_as::<_, (String, String)>( + "SELECT m.title, r.name + FROM grabs g + JOIN movies m ON m.id = g.target_id + JOIN releases r ON r.id = g.release_id + ORDER BY m.title", + ) + .fetch_all(database.pool()) + .await + .unwrap() + } + + async fn stored_releases(database: &Db) -> Vec { + sqlx::query_scalar::<_, String>("SELECT DISTINCT guid FROM releases ORDER BY guid") + .fetch_all(database.pool()) + .await + .unwrap() + } + + /// The acceptance case: a recorded feed against a seeded wanted list + /// grabs exactly the intended titles, by TMDB id where the tracker + /// supplied one. + #[tokio::test] + async fn exactly_the_intended_titles_match() { + let (_dir, database) = wanted(&[ + (693_134, "Dune: Part Two", 2024, false), + (933_260, "The Substance", 2024, false), + (438_631, "Dune", 2021, false), + ]) + .await; + let indexer = prowlarr().await; + let (downloader, fake) = transmission().await; + + let outcomes = action(&indexer, &downloader).tick(&database).await.unwrap(); + + assert_eq!(outcomes.len(), 3); + assert_eq!( + grabbed(&database).await, + vec![ + ( + "Dune".to_owned(), + "Dune.2021.2160p.WEB-DL.DDP5.1.Atmos.H.265-GROUP".to_owned() + ), + ( + "Dune: Part Two".to_owned(), + "Dune.Part.Two.2024.2160p.WEB-DL.DDP5.1.Atmos.H.265-GROUP".to_owned() + ), + ( + "The Substance".to_owned(), + "La.Sustancia.2024.2160p.WEB-DL.DDP5.1.H.265-GRUPO".to_owned() + ), + ], + "one torrent per title, and the 1080p Dune: Part Two loses to the 2160p one" + ); + assert_eq!(fake.added.lock().unwrap().len(), 3); + // The TV item and the film nobody asked for are not cached against + // any title: an unmatched release belongs to nothing. Both indexers + // carry the same four, under their own indexer id. + assert_eq!( + stored_releases(&database).await, + vec!["alpha-1", "alpha-2", "alpha-3", "alpha-4"] + ); + let rows: i64 = sqlx::query_scalar("SELECT count(*) FROM releases") + .fetch_one(database.pool()) + .await + .unwrap(); + assert_eq!(rows, 8); + } + + /// The near miss. "Dune: Part Three" is one word from a release that is + /// in the feed, from the same franchise, in the same year. + #[tokio::test] + async fn a_near_miss_is_not_grabbed() { + let (_dir, database) = wanted(&[(9_999_999, "Dune: Part Three", 2024, false)]).await; + let indexer = prowlarr().await; + let (downloader, fake) = transmission().await; + + let outcomes = action(&indexer, &downloader).tick(&database).await.unwrap(); + + assert!(outcomes.is_empty()); + assert!(fake.added.lock().unwrap().is_empty()); + assert!(stored_releases(&database).await.is_empty()); + } + + /// §6.2: RSS never backs off, so nothing it does may spend one of + /// targeted search's attempts. + #[tokio::test] + async fn matching_never_touches_the_search_backoff() { + let (_dir, database) = wanted(&[ + (693_134, "Dune: Part Two", 2024, false), + (9_999_999, "Dune: Part Three", 2024, false), + ]) + .await; + let indexer = prowlarr().await; + let (downloader, _fake) = transmission().await; + + action(&indexer, &downloader).tick(&database).await.unwrap(); + + let attempts: Vec<(i64, Option)> = + sqlx::query_as("SELECT search_attempts, last_searched_at FROM movies ORDER BY id") + .fetch_all(database.pool()) + .await + .unwrap(); + assert_eq!(attempts, vec![(0, None), (0, None)]); + } + + /// §6.3: `blocked` stops targeted search and leaves RSS matching on. + #[tokio::test] + async fn a_blocked_title_still_matches_rss() { + let (_dir, database) = wanted(&[(693_134, "Dune: Part Two", 2024, true)]).await; + let indexer = prowlarr().await; + let (downloader, _fake) = transmission().await; + + let outcomes = action(&indexer, &downloader).tick(&database).await.unwrap(); + + assert_eq!(outcomes.len(), 1); + } + + /// The property the whole lane exists for: cost is one call per indexer, + /// whatever the wanted list holds. + #[tokio::test] + async fn one_empty_query_call_per_indexer() { + let (_dir, database) = wanted(&[ + (693_134, "Dune: Part Two", 2024, false), + (933_260, "The Substance", 2024, false), + (438_631, "Dune", 2021, false), + ]) + .await; + let indexer = prowlarr().await; + let (downloader, _fake) = transmission().await; + + action(&indexer, &downloader).tick(&database).await.unwrap(); + + for id in INDEXERS { + let feeds = indexer + .received_requests() + .await + .unwrap() + .into_iter() + .filter(|request| { + request.url.path() == format!("/{id}/api") + && request + .url + .query_pairs() + .any(|(name, value)| name == "t" && value == "search") + }) + .count(); + assert_eq!(feeds, 1); + } + } +} diff --git a/crates/arr-daemon/tests/fixtures/rss.xml b/crates/arr-daemon/tests/fixtures/rss.xml new file mode 100644 index 0000000..3653e72 --- /dev/null +++ b/crates/arr-daemon/tests/fixtures/rss.xml @@ -0,0 +1,61 @@ + + + + Alpha Tracker + + Dune.Part.Two.2024.2160p.WEB-DL.DDP5.1.Atmos.H.265-GROUP + alpha-1 + https://indexer.invalid/download/alpha-1 + Sat, 22 Aug 2026 09:14:00 +0000 + + + + + Dune.Part.Two.2024.1080p.WEB-DL.DD5.1.H.264-OTHER + alpha-2 + https://indexer.invalid/download/alpha-2 + Sat, 22 Aug 2026 08:02:00 +0000 + + + + + La.Sustancia.2024.2160p.WEB-DL.DDP5.1.H.265-GRUPO + alpha-3 + https://indexer.invalid/download/alpha-3 + Sat, 22 Aug 2026 07:40:00 +0000 + + + + + + Dune.2021.2160p.WEB-DL.DDP5.1.Atmos.H.265-GROUP + alpha-4 + https://indexer.invalid/download/alpha-4 + Sat, 22 Aug 2026 06:55:00 +0000 + + + + + Dune.Prophecy.S01E03.2024.2160p.WEB-DL.DDP5.1.H.265-GROUP + alpha-5 + https://indexer.invalid/download/alpha-5 + Sat, 22 Aug 2026 06:31:00 +0000 + + + + + Nosferatu.2024.2160p.WEB-DL.DDP5.1.Atmos.H.265-GROUP + alpha-6 + https://indexer.invalid/download/alpha-6 + Sat, 22 Aug 2026 05:12:00 +0000 + + + + + diff --git a/crates/arr-indexer/src/search.rs b/crates/arr-indexer/src/search.rs index 3ba9ae9..339ff3a 100644 --- a/crates/arr-indexer/src/search.rs +++ b/crates/arr-indexer/src/search.rs @@ -69,6 +69,9 @@ pub struct SearchRelease { pub seeders: Option, pub publish_date: Option, pub download_url: String, + /// TMDB id the tracker attached to the item, when it attached one. An ID + /// the indexer supplies beats anything read off the release name. + pub tmdb_id: Option, } /// Results from one indexer in a multi-indexer search. @@ -279,6 +282,7 @@ struct ReleaseBuilder { publish_date: Option, link: Option, enclosure_url: Option, + tmdb_id: Option, } impl ReleaseBuilder { @@ -295,6 +299,7 @@ impl ReleaseBuilder { seeders: self.seeders, publish_date: self.publish_date, download_url, + tmdb_id: self.tmdb_id, }) } @@ -417,6 +422,12 @@ fn read_element_attributes( .and_then(|value| value.parse().ok()) .or(builder.seeders); } + Some("tmdbid") => { + builder.tmdb_id = value + .and_then(|value| value.trim().parse().ok()) + .filter(|&id| id != 0) + .or(builder.tmdb_id); + } _ => {} } } @@ -717,6 +728,38 @@ mod tests { ); } + /// An id the tracker attached to the item is what RSS matching prefers + /// over anything read off the release name (§6.2). + #[test] + fn a_torznab_tmdb_id_is_kept() { + let body = br#" + + + + La.Sustancia.2024.2160p.WEB-DL + identified + https://indexer.invalid/download/identified + + + + Nosferatu.2024.2160p.WEB-DL + anonymous + https://indexer.invalid/download/anonymous + + + + + "#; + + let releases = parse_releases(3, body).expect("feed structure is valid"); + + assert_eq!(releases[0].tmdb_id, Some(933_260)); + assert_eq!( + releases[1].tmdb_id, None, + "a zero is the tracker saying it does not know" + ); + } + /// A pack and one episode of that pack carry the same series title, so /// the tag on the release name is the only thing separating them. #[test]