yolo/arr
3
0
Fork 0
Code Issues 12 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
917aa4fa7646aad1071e2236851aee05a8cb8281
arr/crates/arr-daemon/src/grab.rs
T
Miguel Palhas 917aa4fa76 feat: scale size bands by episode runtime 
Implements #209 per §5.5 as amended by #208: a band's floor and target
are rates against a 45-minute reference runtime, scaled by the series'
minutes per episode. A missing or zero runtime applies the bands
unscaled, and movies are never scaled. The runtime is stored on the
series row (new migration), filled on add and by the metadata refresh,
which never blanks a known value against TMDB's frequently-empty
episode_run_time. Composes with #210: allow_below_floor waives against
the scaled floor.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-08-24 22:45:55 +01:00

2808 lines
102 KiB
Rust
Raw Blame History

//! The grab pipeline: close the "wanted movie, no file" gap by searching,
//! scoring, picking a winner and sending it to Transmission. See DESIGN.md
//! §5.4, §6.2, §7.1, §7.3 and §8.
//!
//! There is no grab delay and there is no job queue. The gap is recomputed
//! from domain rows on every tick, so killing the process mid-grab and
//! restarting converges instead of double-grabbing:
//!
//! - a title with a live `grabs` row is not a gap, so it is never re-searched;
//! - Transmission's `torrent-add` is keyed on the infohash, so re-sending the
//! same release returns the torrent that is already there rather than a
//! second one;
//! - 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::{BTreeMap, HashMap};
use std::path::PathBuf;
use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};
use arr_core::policy::{evaluate, Candidate};
use arr_core::score::{claimed_episode_count, score};
use arr_core::{Language, Policy, TitleOverrides, Verdict};
use arr_db::{blacklist, Blacklist, Db, MoviePolicy};
use arr_dl::{AddTorrent, TorrentSource, TransmissionClient};
use arr_indexer::{Download, 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
/// and a search costs one call per indexer per title (§6.2), so the work is
/// bounded here and the least recently searched titles come first. Backoff
/// and release-date gating are issue #26.
const MOVIES_PER_TICK: i64 = 5;
/// 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.
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct SeedingLimits {
pub ratio: f64,
pub idle_minutes: u64,
}
#[derive(Debug, Clone)]
pub struct SeedingRules {
default: SeedingLimits,
trackers: HashMap<i64, SeedingLimits>,
}
impl SeedingRules {
#[must_use]
pub fn new(default: SeedingLimits, trackers: HashMap<i64, SeedingLimits>) -> Self {
Self { default, trackers }
}
fn for_indexer(&self, indexer_id: i64) -> SeedingLimits {
self.trackers
.get(&indexer_id)
.copied()
.unwrap_or(self.default)
}
}
/// A failure during one grab tick.
#[derive(Debug, thiserror::Error)]
pub enum GrabError {
#[error("database: {0}")]
Database(#[from] sqlx::Error),
#[error("policy: {0}")]
Policy(#[from] arr_db::PolicyError),
#[error("{0}")]
Discovery(#[from] DiscoveryError),
#[error("TMDB: {0}")]
Metadata(#[from] arr_meta::Error),
#[error("movie {0} has an invalid TMDB id")]
InvalidTmdbId(i64),
#[error("transmission: {0}")]
Transmission(#[from] arr_dl::Error),
#[error("download link: {0}")]
Download(#[from] arr_indexer::DownloadError),
#[error("release {name}: {source}")]
Parsed {
name: String,
source: serde_json::Error,
},
}
/// Sends the best eligible release for every wanted movie that has neither a
/// file nor a grab in flight.
#[derive(Debug)]
pub struct GrabAction {
prowlarr: ProwlarrClient,
grabber: Grabber,
tmdb: Option<Arc<TmdbClient>>,
indexers: IndexerDirectory,
}
impl GrabAction {
#[must_use]
pub fn new(
prowlarr: ProwlarrClient,
transmission: TransmissionClient,
download_dir: PathBuf,
seeding: SeedingRules,
) -> Self {
Self {
indexers: IndexerDirectory::new(prowlarr.clone()),
grabber: Grabber::new(prowlarr.clone(), transmission, download_dir, seeding),
prowlarr,
tmdb: None,
}
}
#[must_use]
pub fn with_tmdb(mut self, tmdb: Arc<TmdbClient>) -> Self {
self.tmdb = Some(tmdb);
self
}
async fn tick(&self, database: &Db) -> Result<Vec<Outcome>, GrabError> {
let mut outcomes = self.grabber.track_sent_grabs(database).await?;
let gaps = pending_movies(database).await?;
if gaps.is_empty() {
return Ok(outcomes);
}
for movie in gaps {
let (movie_id, title) = (movie.id, movie.title.clone());
let movie = match self.refresh_metadata(database, movie).await {
Ok((movie, true)) => movie,
// Not released yet: no targeted search (§6.2).
Ok((_, false)) => continue,
// A title's metadata is not the rest of the tick's problem,
// same as a grab failure below.
Err(error) => {
tracing::error!(movie_id, title, %error, "metadata refresh failed");
continue;
}
};
if !search_due(&movie) {
continue;
}
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);
}
match self.grab_one(database, &movie, &searchable).await {
Ok(Some(outcome)) => outcomes.push(outcome),
Ok(None) => {}
// One title's failure must not cost the rest of the tick.
Err(error) => tracing::error!(
movie_id = movie.id,
title = movie.title,
%error,
"grab failed"
),
}
}
Ok(outcomes)
}
/// Refresh one title from TMDB, returning it alongside whether it is
/// digitally released (§6.2). Targeted search is gated on that flag; the
/// manual deck refresh of an already-available movie is not, since the
/// file on disk is proof enough and TMDB does not know a digital date for
/// every title (issue #115).
async fn refresh_metadata(
&self,
database: &Db,
movie: PendingMovie,
) -> Result<(PendingMovie, bool), GrabError> {
let Some(tmdb) = &self.tmdb else {
return Ok((movie, true));
};
if !metadata_refresh_due(movie.metadata_refreshed_at.as_deref()) {
let released = is_digitally_released(movie.digital_release.as_deref());
return Ok((movie, released));
}
let tmdb_id =
u32::try_from(movie.tmdb_id).map_err(|_| GrabError::InvalidTmdbId(movie.id))?;
let metadata = tmdb.movie(tmdb_id).await?;
let changed = store_movie_metadata(database, movie.id, &metadata).await?;
if changed {
tracing::info!(
movie_id = movie.id,
"metadata changed; reset targeted search backoff"
);
}
let released = metadata.is_digitally_released(chrono::Utc::now().date_naive());
if !released {
tracing::debug!(
movie_id = movie.id,
"digital release has not happened; skipping targeted search"
);
}
let title = metadata.title.clone();
Ok((
PendingMovie {
id: movie.id,
tmdb_id: movie.tmdb_id,
title,
year: metadata.year().map(i64::from),
original_language: (!metadata.original_language.is_empty())
.then_some(metadata.original_language.clone()),
search_attempts: if changed { 0 } else { movie.search_attempts },
last_searched_at: if changed {
None
} else {
movie.last_searched_at
},
digital_release: metadata.digital_release.map(|date| date.to_string()),
metadata_refreshed_at: None,
},
released,
))
}
/// Search every indexer for one title, cache each candidate with its
/// verdict and score (§9.3), and return the eligible ones best first.
///
/// The order is total, so the same candidate set picks the same winner
/// after a restart and a re-sent grab is a duplicate rather than a second
/// torrent. Waived releases are a manual, one-click decision (§9.3) and
/// never appear here.
async fn search(
&self,
database: &Db,
movie: &PendingMovie,
indexers: &[i64],
loaded: &MoviePolicy,
original_language: &Language,
blacklist: &Blacklist,
) -> Result<Vec<Eligible>, GrabError> {
let request = SearchRequest::Text {
query: search_query(movie),
};
let mut releases = Vec::new();
for search in self.prowlarr.search_indexers(indexers, &request).await {
if let Some(error) = search.error {
tracing::warn!(indexer_id = search.indexer_id, %error, "indexer search failed");
}
releases.extend(search.releases);
}
let mut candidates = Vec::new();
for release in releases {
let stored = store_release(
database,
movie.id,
&release,
&loaded.policy,
&loaded.overrides,
original_language,
blacklist,
)
.await?;
if let Some(candidate) = stored {
candidates.push(candidate);
}
}
candidates.sort_by(|left, right| {
right
.score
.cmp(&left.score)
.then_with(|| left.indexer_id.cmp(&right.indexer_id))
.then_with(|| left.guid.cmp(&right.guid))
});
Ok(candidates)
}
async fn grab_one(
&self,
database: &Db,
movie: &PendingMovie,
indexers: &[i64],
) -> Result<Option<Outcome>, GrabError> {
let loaded = database.movie_policy(movie.id).await?;
let Some(loaded) = loaded else {
return Ok(None);
};
// §5.2: the whole language rule is expressed against the title's own
// original language. Without it there is nothing to evaluate against,
// and guessing gives a child a Brazilian dub or throws away a
// Brazilian film's own soundtrack.
let Some(original_language) = movie.original_language.as_deref() else {
tracing::warn!(
movie_id = movie.id,
title = movie.title,
"no original language yet; not searching"
);
return Ok(None);
};
let original_language = arr_db::policy::language(original_language);
// §6.3: anything that hard-failed post-ffprobe is never grabbed
// again. `store_release` already rejects a blacklisted candidate as
// it classifies it, so this second pass only catches a row that was
// classified before the blacklist entry existed.
let blacklist = Blacklist::load(database.pool()).await?;
let candidates = self
.search(
database,
movie,
indexers,
&loaded,
&original_language,
&blacklist,
)
.await?;
let Some(winner) = candidates.into_iter().find(|candidate| {
!blacklist.blocks_candidate(&candidate.name, &candidate.download_url)
}) else {
record_search(database, movie.id).await?;
tracing::info!(
movie_id = movie.id,
title = movie.title,
"no eligible release"
);
return Ok(None);
};
self.grabber
.send_winner(
database,
&GrabTarget {
scope: GrabScope::Movie { movie_id: movie.id },
title: &movie.title,
counts_as_attempt: true,
},
&loaded,
&blacklist,
winner,
)
.await
}
/// Sweep every searchable indexer for one title and upsert the results
/// into its release deck (§9.3) without choosing a winner.
///
/// Issue #115: a manual search on a movie that is already satisfied is an
/// upgrade view being refreshed, not a gap being closed. It must re-score
/// and re-cache candidates — the same writes as the grab lane — and stop
/// there. Only a wanted movie with nothing on disk auto-grabs.
async fn refresh_deck(
&self,
database: &Db,
movie: &PendingMovie,
indexers: &[i64],
) -> Result<(), GrabError> {
let Some(loaded) = database.movie_policy(movie.id).await? else {
return Ok(());
};
// §5.2: no original language, nothing to evaluate the language rule
// against — same refusal as the grab lane.
let Some(original_language) = movie.original_language.as_deref() else {
tracing::warn!(
movie_id = movie.id,
title = movie.title,
"no original language yet; not searching"
);
return Ok(());
};
let original_language = arr_db::policy::language(original_language);
let blacklist = Blacklist::load(database.pool()).await?;
let eligible = self
.search(
database,
movie,
indexers,
&loaded,
&original_language,
&blacklist,
)
.await?;
record_search(database, movie.id).await?;
tracing::info!(
movie_id = movie.id,
title = movie.title,
eligible = eligible.len(),
"release deck refreshed; not grabbing an already-satisfied movie"
);
Ok(())
}
/// The manual trigger (§6.2, §9.3, issues #107 and #115): reset the
/// backoff so `search_due` cannot skip the title, then sweep the
/// indexers, scoped to this one movie instead of the tick's
/// `MOVIES_PER_TICK` batch.
///
/// What happens to the results depends on the movie, not on the command:
/// a gap — wanted, unblocked, nothing on disk, no grab in flight — gets
/// search and grab; anything else already satisfied gets its deck
/// refreshed and no grab; a blocked movie is refused outright (§6.3).
pub(crate) async fn search_now(
&self,
database: &Db,
movie_id: i64,
) -> Result<Option<Outcome>, GrabError> {
let Some((movie, lane)) = manual_movie(database, movie_id).await? else {
tracing::info!(movie_id, "manual search refused: unknown or blocked movie");
return Ok(None);
};
reset_search_backoff(database, movie_id).await?;
let (movie, released) = self.refresh_metadata(database, movie).await?;
// The release-date gate belongs to targeted search. A movie that is
// already on disk is released whatever TMDB says.
if !released && lane == ManualSearch::SearchAndGrab {
return Ok(None);
}
let searchable = self.indexers.searchable().await?;
if searchable.is_empty() {
tracing::warn!("no indexer advertises a text search; nothing can be grabbed");
return Ok(None);
}
match lane {
ManualSearch::SearchAndGrab => self.grab_one(database, &movie, &searchable).await,
ManualSearch::RefreshDeck => {
self.refresh_deck(database, &movie, &searchable).await?;
Ok(None)
}
}
}
/// The manual one-click grab (§9.3, issue #107): the release is already
/// chosen, so this skips search and scoring and sends it straight to
/// Transmission.
pub(crate) async fn grab_release_now(
&self,
database: &Db,
movie_id: i64,
release_id: i64,
) -> Result<Option<Outcome>, GrabError> {
let Some(loaded) = database.movie_policy(movie_id).await? else {
return Ok(None);
};
let Some(title) = movie_title(database, movie_id).await? else {
return Ok(None);
};
let Some(release) = load_release(database, movie_id, release_id).await? else {
return Ok(None);
};
let blacklist = Blacklist::load(database.pool()).await?;
self.grabber
.send_winner(
database,
&GrabTarget {
scope: GrabScope::Movie { movie_id },
title: &title,
counts_as_attempt: false,
},
&loaded,
&blacklist,
release,
)
.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 {
/// Resolves the winner's indexer link before it is sent on: Transmission
/// cannot reach Prowlarr and arr can (issue #100).
prowlarr: ProwlarrClient,
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) scope: GrabScope,
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,
}
/// What a grab targets: the `grabs` row's kind and id, plus the episodes the
/// torrent covers — attempts and state changes land on those leaves (§4.1).
#[derive(Debug)]
pub(crate) enum GrabScope {
Movie {
movie_id: i64,
},
Episode {
episode_id: i64,
},
/// A season pack: one torrent, one `grabs` row on the season, every
/// missing wanted episode it covers flipped to downloading.
Season {
season_id: i64,
episode_ids: Vec<i64>,
},
}
impl GrabScope {
fn target_kind(&self) -> &'static str {
match self {
Self::Movie { .. } => "movie",
Self::Episode { .. } => "episode",
Self::Season { .. } => "season",
}
}
fn target_id(&self) -> i64 {
match self {
Self::Movie { movie_id } => *movie_id,
Self::Episode { episode_id } => *episode_id,
Self::Season { season_id, .. } => *season_id,
}
}
fn episode_ids(&self) -> &[i64] {
match self {
Self::Movie { .. } => &[],
Self::Episode { episode_id } => std::slice::from_ref(episode_id),
Self::Season { episode_ids, .. } => episode_ids,
}
}
}
impl Grabber {
pub(crate) fn new(
prowlarr: ProwlarrClient,
transmission: TransmissionClient,
download_dir: PathBuf,
seeding: SeedingRules,
) -> Self {
Self {
prowlarr,
transmission,
download_dir,
seeding,
}
}
/// Move grabs Transmission reports as complete out of `sent`, whatever
/// they target.
///
/// Transmission is authoritative and its view is rebuilt on every tick
/// rather than cached (§8), so this is also what reconstructs in-flight
/// state after a restart. Both grab actions call it; whichever runs first
/// does the work and the other finds nothing.
pub(crate) async fn track_sent_grabs(&self, database: &Db) -> Result<Vec<Outcome>, GrabError> {
let sent = sqlx::query!(
r#"SELECT id AS "id!: i64", infohash AS "infohash!: String",
target_kind AS "target_kind!: String", target_id AS "target_id!: i64"
FROM grabs WHERE state = 'sent'"#
)
.fetch_all(database.pool())
.await?;
if sent.is_empty() {
return Ok(Vec::new());
}
let torrents: HashMap<String, f64> = self
.transmission
.list_torrents()
.await?
.into_iter()
.map(|torrent| (torrent.hash.to_ascii_lowercase(), torrent.progress))
.collect();
let mut outcomes = Vec::new();
for grab in sent {
let Some(progress) = torrents.get(&grab.infohash.to_ascii_lowercase()) else {
outcomes.push(
self.vanish(database, grab.id, &grab.target_kind, grab.target_id)
.await?,
);
continue;
};
if *progress < 1.0 {
continue;
}
sqlx::query!(
"UPDATE grabs SET state = 'downloaded' WHERE id = ?",
grab.id
)
.execute(database.pool())
.await?;
outcomes.push(Outcome::new(
format!("grab {} downloaded, still marked sent", grab.id),
format!("marked grab {} downloaded", grab.id),
));
tracing::info!(
grab_id = grab.id,
target_kind = grab.target_kind,
target_id = grab.target_id,
"download complete"
);
}
Ok(outcomes)
}
/// §86/#108: a `sent` grab whose infohash Transmission no longer reports
/// — removed by hand, not a policy failure. Marked `vanished` rather than
/// `failed` so it does not feed the `needs_decision` queue (attention.rs).
/// Nothing is blacklisted, since the release itself never failed policy,
/// and the target is parked rather than reopened, since removing a
/// torrent by hand is human intent, not a gap to refill.
async fn vanish(
&self,
database: &Db,
grab_id: i64,
target_kind: &str,
target_id: i64,
) -> Result<Outcome, GrabError> {
sqlx::query!("UPDATE grabs SET state = 'vanished' WHERE id = ?", grab_id)
.execute(database.pool())
.await?;
park_target(database, target_kind, target_id).await?;
tracing::warn!(
grab_id,
target_kind,
target_id,
"torrent vanished from Transmission; target parked"
);
Ok(Outcome::new(
format!("grab {grab_id} sent, torrent vanished from Transmission"),
format!("parked {target_kind} {target_id}"),
))
}
async fn record_attempt(
&self,
database: &Db,
target: &GrabTarget<'_>,
) -> Result<(), GrabError> {
if !target.counts_as_attempt {
return Ok(());
}
match &target.scope {
GrabScope::Movie { movie_id } => record_search(database, *movie_id).await,
scope => record_episode_search(database, scope.episode_ids()).await,
}
}
/// Resolve the winner's indexer link and add it to Transmission.
///
/// The link is resolved here rather than passed on, because Transmission
/// has no route to Prowlarr and cannot follow its redirect to a magnet
/// (issue #100).
async fn send_to_transmission(
&self,
winner: &Eligible,
loaded: &MoviePolicy,
) -> Result<arr_dl::AddedTorrent, GrabError> {
let seeding = self.seeding.for_indexer(winner.indexer_id);
let source = torrent_source(self.prowlarr.download(&winner.download_url).await?);
Ok(self
.transmission
.add_torrent(AddTorrent {
source,
label: label(loaded),
download_dir: self.download_dir.clone(),
seed_ratio_limit: seeding.ratio,
seed_idle_limit_minutes: seeding.idle_minutes,
})
.await?)
}
/// Add the winning release to Transmission and record the grab.
///
/// The search still counts as an attempt (§6.2) on every exit that is
/// 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.
pub(crate) async fn send_winner(
&self,
database: &Db,
target: &GrabTarget<'_>,
loaded: &MoviePolicy,
blacklist: &Blacklist,
winner: Eligible,
) -> Result<Option<Outcome>, GrabError> {
let added = match self.send_to_transmission(&winner, loaded).await {
Ok(added) => added,
Err(error) => {
self.record_attempt(database, target).await?;
return Err(error);
}
};
let infohash = added.hash.to_ascii_lowercase();
// §6.3's second key. A `.torrent` link hides its infohash until
// Transmission has fetched it, so the same blacklisted torrent can
// reach here under a new name.
if blacklist.blocks_infohash(&infohash) {
self.record_attempt(database, target).await?;
self.drop_blacklisted_torrent(database, target, &winner, &added)
.await?;
return Ok(None);
}
// `infohash` is unique, so re-grabbing the same release conflicts
// with its own earlier row. A `sent`/`downloaded` row is the restart
// case and is left alone; a `vanished` one (§86) is reclaimed, since
// nothing blacklisted the release.
let target_kind = target.scope.target_kind();
let target_id = target.scope.target_id();
let inserted = sqlx::query!(
r#"INSERT INTO grabs (release_id, target_kind, target_id, infohash, state)
VALUES (?, ?, ?, ?, 'sent')
ON CONFLICT (infohash) DO UPDATE SET
release_id = excluded.release_id,
target_kind = excluded.target_kind,
target_id = excluded.target_id,
state = 'sent',
grabbed_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now'),
imported_at = NULL
WHERE grabs.state = 'vanished'
RETURNING id AS "id!: i64""#,
winner.id,
target_kind,
target_id,
infohash
)
.fetch_optional(database.pool())
.await?;
match &target.scope {
GrabScope::Movie { movie_id } => {
sqlx::query!(
"UPDATE movies SET state = 'downloading',
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
WHERE id = ?",
movie_id
)
.execute(database.pool())
.await?;
}
scope => {
for episode_id in scope.episode_ids() {
sqlx::query!(
"UPDATE episodes SET state = 'downloading',
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
WHERE id = ?",
episode_id
)
.execute(database.pool())
.await?;
}
}
}
let Some(inserted) = inserted else {
tracing::info!(
target_kind,
target_id,
infohash,
"grab already recorded for this torrent"
);
return Ok(None);
};
tracing::info!(
target_kind,
target_id,
title = target.title,
release = winner.name,
score = winner.score,
infohash,
was_duplicate = added.was_duplicate,
"grabbed"
);
Ok(Some(Outcome::new(
format!("{target_kind} {target_id} wanted with no file"),
format!("grabbed {} as grab {}", winner.name, inserted.id),
)))
}
/// Undo a grab whose infohash turned out to be blacklisted (§6.3).
///
/// The name is added to the blacklist so the next tick stops at the cheap
/// check instead of paying Transmission again, and no `grabs` row is
/// written, which leaves the title a gap for the next candidate.
async fn drop_blacklisted_torrent(
&self,
database: &Db,
target: &GrabTarget<'_>,
winner: &Eligible,
added: &arr_dl::AddedTorrent,
) -> Result<(), GrabError> {
let release_name = &winner.name;
blacklist::add(
database.pool(),
None,
release_name,
"blacklisted infohash under a new name",
)
.await?;
// `store_release` classified this row before the infohash was known,
// so it still reads eligible. Correct it here rather than waiting for
// the next search to overwrite it: until then §9.3's manual view
// would keep offering a release this tick just refused.
sqlx::query!(
"UPDATE releases SET verdict = 'rejected', rejected_rule = ? WHERE id = ?",
blacklist::RULE,
winner.id
)
.execute(database.pool())
.await?;
if added.was_duplicate {
// The earlier grab's torrent, still working off its seeding
// obligation (§7.3). Nothing here deletes a torrent.
tracing::warn!(
title = target.title,
release = release_name,
infohash = added.hash,
"blacklisted torrent re-listed under a new name; left seeding"
);
} else {
// This tick added it seconds ago, so it carries no seeding
// obligation and has nothing on disk worth keeping.
self.transmission.remove_torrent(added.id, true).await?;
tracing::warn!(
title = target.title,
release = release_name,
infohash = added.hash,
"blacklisted torrent re-listed under a new name; removed"
);
}
Ok(())
}
}
impl Action for GrabAction {
fn name(&self) -> &'static str {
"grab"
}
fn run<'a>(&'a self, database: &'a Db) -> ActionFuture<'a> {
Box::pin(async move { self.tick(database).await.map_err(Into::into) })
}
}
/// A wanted movie with neither a file nor a grab in flight.
#[derive(Debug, Clone)]
struct PendingMovie {
id: i64,
tmdb_id: i64,
title: String,
year: Option<i64>,
/// §5.2's language rules are expressed against this, and guessing it is
/// worse than not grabbing.
original_language: Option<String>,
search_attempts: i64,
last_searched_at: Option<String>,
digital_release: Option<String>,
metadata_refreshed_at: Option<String>,
}
/// The eligible view of a stored release, ranked for selection.
#[derive(Debug, Clone)]
pub(crate) struct Eligible {
pub(crate) id: i64,
pub(crate) indexer_id: i64,
pub(crate) guid: String,
pub(crate) name: String,
pub(crate) download_url: String,
pub(crate) score: i64,
}
/// The gap, straight out of the domain rows (§8).
///
/// A title with an unfinished grab is not a gap — that is what stops a
/// restart from grabbing twice. `blocked` stops targeted search only (§6.3),
/// so a blocked title still matches RSS results, which is issue #27.
async fn pending_movies(database: &Db) -> Result<Vec<PendingMovie>, 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,
m.original_language,
m.search_attempts AS "search_attempts!: i64",
m.last_searched_at,
m.digital_release,
m.metadata_refreshed_at
FROM movies m
WHERE m.wanted = 1
AND m.blocked = 0
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.last_searched_at IS NOT NULL, m.last_searched_at, m.id
LIMIT ?
"#,
MOVIES_PER_TICK
)
.fetch_all(database.pool())
.await?;
Ok(rows
.into_iter()
.map(|row| PendingMovie {
id: row.id,
tmdb_id: row.tmdb_id,
title: row.title,
year: row.year,
original_language: row.original_language,
search_attempts: row.search_attempts,
last_searched_at: row.last_searched_at,
digital_release: row.digital_release,
metadata_refreshed_at: row.metadata_refreshed_at,
})
.collect())
}
/// Write one refresh's fields to the movie row, guarded so unchanged data
/// moves nothing. Returns whether anything did.
///
/// A free function rather than a [`GrabAction`] method because the metadata
/// lane refreshes a title on demand (issue #176) with nothing but a TMDB
/// client — grabbing needs Prowlarr, refreshing does not.
pub(crate) async fn store_movie_metadata(
database: &Db,
movie_id: i64,
metadata: &arr_meta::Movie,
) -> Result<bool, GrabError> {
let title = metadata.title.clone();
let year = metadata.year().map(i64::from);
let original_language =
(!metadata.original_language.is_empty()).then_some(metadata.original_language.clone());
let digital_release = metadata.digital_release.map(|date| date.to_string());
// §6.2: the id RSS matching prefers, and the one Torznab movie
// searches take. TMDB does not know one for every title.
let imdb_id = metadata.imdb_id.clone();
// §9.6: these three are the exception to "rich detail is not
// persisted" — pure-SQL views render artwork without a TMDB call.
let poster_path = metadata.poster_path.clone();
let backdrop_path = metadata.backdrop_path.clone();
let vote_average = metadata.vote_average;
let title_ref = title.as_str();
let original_language_ref = original_language.as_deref();
let digital_release_ref = digital_release.as_deref();
let imdb_id_ref = imdb_id.as_deref();
let poster_path_ref = poster_path.as_deref();
let backdrop_path_ref = backdrop_path.as_deref();
let changed = sqlx::query!(
r#"UPDATE movies
SET title = ?, year = ?, original_language = ?, digital_release = ?,
imdb_id = ?, poster_path = ?, backdrop_path = ?, vote_average = ?,
metadata_refreshed_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now'),
search_attempts = 0, last_searched_at = NULL,
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
WHERE id = ? AND (
title IS NOT ? OR year IS NOT ? OR original_language IS NOT ?
OR digital_release IS NOT ? OR imdb_id IS NOT ?
OR poster_path IS NOT ? OR backdrop_path IS NOT ?
OR vote_average IS NOT ?
)"#,
title_ref,
year,
original_language_ref,
digital_release_ref,
imdb_id_ref,
poster_path_ref,
backdrop_path_ref,
vote_average,
movie_id,
title_ref,
year,
original_language_ref,
digital_release_ref,
imdb_id_ref,
poster_path_ref,
backdrop_path_ref,
vote_average,
)
.execute(database.pool())
.await?
.rows_affected()
!= 0;
if !changed {
// Still stamp the refresh even when nothing changed, or the TTL
// gate above never engages and every tick pays for TMDB again.
sqlx::query!(
"UPDATE movies SET metadata_refreshed_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
WHERE id = ?",
movie_id
)
.execute(database.pool())
.await?;
}
Ok(changed)
}
fn search_due(movie: &PendingMovie) -> bool {
backoff_elapsed(movie.search_attempts, movie.last_searched_at.as_deref())
}
/// What a manual search on one movie is allowed to do (issue #115).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ManualSearch {
/// The movie is a gap: sweep the indexers and grab the winner.
SearchAndGrab,
/// The movie is already satisfied: sweep the indexers, refresh the deck
/// (§9.3), grab nothing.
RefreshDeck,
}
/// Same columns as [`pending_movies`], scoped to one id, without the tick's
/// batch limit — the manual trigger already named which title to search — and
/// without the gap filter, which decides the lane here instead of hiding the
/// movie. `blocked` still refuses (§6.3).
async fn manual_movie(
database: &Db,
movie_id: i64,
) -> Result<Option<(PendingMovie, ManualSearch)>, GrabError> {
let row = sqlx::query!(
r#"
SELECT m.id AS "id!: i64",
m.tmdb_id AS "tmdb_id!: i64",
m.title AS "title!: String",
m.year,
m.original_language,
m.search_attempts AS "search_attempts!: i64",
m.last_searched_at,
m.digital_release,
m.metadata_refreshed_at,
(
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')
)
) AS "is_gap!: i64"
FROM movies m
WHERE m.id = ?
AND m.blocked = 0
"#,
movie_id
)
.fetch_optional(database.pool())
.await?;
Ok(row.map(|row| {
let lane = if row.is_gap == 1 {
ManualSearch::SearchAndGrab
} else {
ManualSearch::RefreshDeck
};
(
PendingMovie {
id: row.id,
tmdb_id: row.tmdb_id,
title: row.title,
year: row.year,
original_language: row.original_language,
search_attempts: row.search_attempts,
last_searched_at: row.last_searched_at,
digital_release: row.digital_release,
metadata_refreshed_at: row.metadata_refreshed_at,
},
lane,
)
}))
}
/// Unconditional, unlike [`record_search`]: the manual trigger's whole point
/// is to ignore the exponential backoff (§6.2).
async fn reset_search_backoff(database: &Db, movie_id: i64) -> Result<(), GrabError> {
sqlx::query!(
"UPDATE movies SET search_attempts = 0, last_searched_at = NULL,
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
WHERE id = ?",
movie_id
)
.execute(database.pool())
.await?;
Ok(())
}
async fn movie_title(database: &Db, movie_id: i64) -> Result<Option<String>, GrabError> {
Ok(
sqlx::query_scalar!("SELECT title FROM movies WHERE id = ?", movie_id)
.fetch_optional(database.pool())
.await?,
)
}
/// The chosen release, still associated with the movie and not hard-failed —
/// a manual grab may take a `waived` release (§9.3), never a `rejected` one.
async fn load_release(
database: &Db,
movie_id: i64,
release_id: i64,
) -> Result<Option<Eligible>, GrabError> {
let row = sqlx::query!(
r#"
SELECT r.id AS "id!: i64",
r.indexer_id AS "indexer_id!: i64",
r.guid AS "guid!: String",
r.name AS "name!: String",
r.download_url AS "download_url!: String",
CAST(COALESCE(r.score, 0) AS INTEGER) AS "score!: i64"
FROM releases r
JOIN movie_releases mr ON mr.release_id = r.id
WHERE mr.movie_id = ? AND r.id = ? AND r.verdict IN ('eligible', 'waived')
"#,
movie_id,
release_id
)
.fetch_optional(database.pool())
.await?;
Ok(row.map(|row| Eligible {
id: row.id,
indexer_id: row.indexer_id,
guid: row.guid,
name: row.name,
download_url: row.download_url,
score: row.score,
}))
}
/// The §6.2 targeted-search backoff, shared by movie and episode search.
pub(crate) fn backoff_elapsed(search_attempts: i64, last_searched_at: Option<&str>) -> bool {
let Some(last_searched_at) = last_searched_at else {
return true;
};
let Ok(last_searched_at) = chrono::DateTime::parse_from_rfc3339(last_searched_at) else {
return true;
};
let backoff = match search_attempts {
1 => chrono::TimeDelta::hours(1),
2 => chrono::TimeDelta::hours(6),
3 => chrono::TimeDelta::days(1),
4 => chrono::TimeDelta::days(3),
_ => chrono::TimeDelta::days(7),
};
last_searched_at.with_timezone(&chrono::Utc) + backoff <= chrono::Utc::now()
}
pub(crate) fn metadata_refresh_due(metadata_refreshed_at: Option<&str>) -> bool {
let Some(refreshed_at) = metadata_refreshed_at else {
return true;
};
let Ok(refreshed_at) = chrono::DateTime::parse_from_rfc3339(refreshed_at) else {
return true;
};
// Independent of the search backoff (§6.2): a title stuck on a day-long
// backoff, or one with no digital release date yet, must not cost a
// TMDB call every tick.
refreshed_at.with_timezone(&chrono::Utc) + chrono::TimeDelta::hours(6) <= chrono::Utc::now()
}
fn is_digitally_released(digital_release: Option<&str>) -> bool {
digital_release
.and_then(|date| date.parse::<chrono::NaiveDate>().ok())
.is_some_and(|date| date <= chrono::Utc::now().date_naive())
}
/// Cache the classified release and associate it with the title.
///
/// Returns the candidate only when the release is eligible: automatic
/// selection never takes a waiver (§9.3).
///
/// The blacklist is applied here rather than at selection so it reaches every
/// 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.
pub(crate) async fn store_release(
database: &Db,
movie_id: i64,
release: &SearchRelease,
policy: &Policy,
overrides: &TitleOverrides,
original_language: &Language,
blacklist: &Blacklist,
) -> Result<Option<Eligible>, GrabError> {
// A movie is one episode's worth and is never runtime-scaled (§5.5).
let (release_id, eligible) = classify_and_store(
database,
release,
policy,
overrides,
original_language,
blacklist,
1,
0,
)
.await?;
sqlx::query!(
"INSERT INTO movie_releases (movie_id, release_id) VALUES (?, ?)
ON CONFLICT DO NOTHING",
movie_id,
release_id
)
.execute(database.pool())
.await?;
Ok(eligible)
}
/// The TV counterpart: one release row, associated with every episode the
/// claim covers — a season pack matches the whole season. Returns the
/// release id alongside the eligible view; the id is what links a season
/// pack into `season_releases`, which lists rejected candidates too.
pub(crate) async fn store_episode_release(
database: &Db,
episode_ids: &[i64],
release: &SearchRelease,
policy: &Policy,
overrides: &TitleOverrides,
original_language: &Language,
blacklist: &Blacklist,
) -> Result<(i64, Option<Eligible>), GrabError> {
// A size band describes one episode (`DESIGN.md` §5.5): the release is
// measured per episode, and a season pack's length comes from the series
// the covered episodes belong to. An unrevealed season divides by one.
let claim = arr_parse::parse(&release.name).episode;
let season_lengths = match episode_ids.first() {
Some(&episode_id)
if claim
.as_ref()
.is_some_and(arr_parse::EpisodeClaim::is_season_pack) =>
{
season_lengths_of(database, episode_id).await?
}
_ => BTreeMap::new(),
};
let episode_count = claimed_episode_count(claim.as_ref(), &season_lengths);
// §5.5: the size bands scale by the series' minutes per episode.
let runtime_minutes = match episode_ids.first() {
Some(&episode_id) => series_runtime_of(database, episode_id).await?,
None => 0,
};
let (release_id, eligible) = classify_and_store(
database,
release,
policy,
overrides,
original_language,
blacklist,
episode_count,
runtime_minutes,
)
.await?;
for episode_id in episode_ids {
sqlx::query!(
"INSERT INTO episode_releases (episode_id, release_id) VALUES (?, ?)
ON CONFLICT DO NOTHING",
episode_id,
release_id
)
.execute(database.pool())
.await?;
}
Ok((release_id, eligible))
}
/// Per-season episode counts for the series one covered episode belongs to
/// (`DESIGN.md` §5.5): the divisor data for season-pack normalisation.
async fn season_lengths_of(
database: &Db,
episode_id: i64,
) -> Result<BTreeMap<u32, u32>, GrabError> {
let rows = sqlx::query!(
r#"SELECT se.number AS "number!: i64", COUNT(e.id) AS "episodes!: i64"
FROM seasons se
LEFT JOIN episodes e ON e.season_id = se.id
WHERE se.series_id = (SELECT s2.series_id FROM episodes e2
JOIN seasons s2 ON s2.id = e2.season_id
WHERE e2.id = ?)
GROUP BY se.number"#,
episode_id
)
.fetch_all(database.pool())
.await?;
Ok(rows
.into_iter()
.map(|row| {
(
u32::try_from(row.number).unwrap_or_default(),
u32::try_from(row.episodes).unwrap_or_default(),
)
})
.collect())
}
/// The minutes-per-episode of the series one covered episode belongs to
/// (`DESIGN.md` §5.5): the scale factor for its size bands. Zero when the
/// series has no known runtime, which applies the bands unscaled.
async fn series_runtime_of(database: &Db, episode_id: i64) -> Result<u32, GrabError> {
let minutes = sqlx::query_scalar!(
r#"SELECT s.runtime_minutes FROM series s
WHERE s.id = (SELECT s2.series_id FROM episodes e
JOIN seasons s2 ON s2.id = e.season_id
WHERE e.id = ?)"#,
episode_id
)
.fetch_optional(database.pool())
.await?
.flatten();
Ok(minutes
.and_then(|minutes| u32::try_from(minutes).ok())
.unwrap_or(0))
}
#[allow(clippy::too_many_arguments)]
async fn classify_and_store(
database: &Db,
release: &SearchRelease,
policy: &Policy,
overrides: &TitleOverrides,
original_language: &Language,
blacklist: &Blacklist,
episode_count: u32,
runtime_minutes: u32,
) -> Result<(i64, Option<Eligible>), GrabError> {
let parsed = arr_parse::parse(&release.name);
let evaluation = evaluate(
policy,
overrides,
original_language,
Candidate::PreGrab(&parsed),
release.size,
episode_count,
runtime_minutes,
);
let scored = score(
policy,
Candidate::PreGrab(&parsed),
release.size.unwrap_or_default(),
release.seeders.unwrap_or_default(),
episode_count,
runtime_minutes,
);
// A release that did not say its size is not a tiny one: scoring it
// against the band's floor would bury it. Same treatment as the manual
// search view, so the ranking the operator sees is the one that picks.
let score = if release.size.is_some() {
scored.total
} else {
scored.source.saturating_add(scored.seeders)
};
let (verdict, rule) = if blacklist.blocks_candidate(&release.name, &release.download_url) {
("rejected", Some(blacklist::RULE.to_owned()))
} else {
verdict_columns(&evaluation.verdict)
};
let parsed_json = serde_json::to_string(&parsed).map_err(|source| GrabError::Parsed {
name: release.name.clone(),
source,
})?;
let size = i64::try_from(release.size.unwrap_or(0)).unwrap_or(i64::MAX);
let seeders = release.seeders.map(i64::from);
let publish_date = release.publish_date.and_then(rfc3339);
let id = sqlx::query_scalar!(
r#"
INSERT INTO releases
(indexer_id, guid, name, size, seeders, publish_date, download_url,
parsed, score, verdict, rejected_rule)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT (indexer_id, guid) DO UPDATE SET
name = excluded.name,
size = excluded.size,
seeders = excluded.seeders,
publish_date = excluded.publish_date,
download_url = excluded.download_url,
parsed = excluded.parsed,
score = excluded.score,
verdict = excluded.verdict,
rejected_rule = excluded.rejected_rule
RETURNING id AS "id!: i64"
"#,
release.indexer_id,
release.guid,
release.name,
size,
seeders,
publish_date,
release.download_url,
parsed_json,
score,
verdict,
rule
)
.fetch_one(database.pool())
.await?;
if verdict != "eligible" {
return Ok((id, None));
}
Ok((
id,
Some(Eligible {
id,
indexer_id: release.indexer_id,
guid: release.guid.clone(),
name: release.name.clone(),
download_url: release.download_url.clone(),
score,
}),
))
}
/// 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!(
"UPDATE movies
SET search_attempts = search_attempts + 1,
last_searched_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now'),
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
WHERE id = ?",
movie_id
)
.execute(database.pool())
.await?;
Ok(())
}
/// The episode-level backoff counter (§6.2), one bump per searched episode. A
/// season-pack search touches every episode it was trying to satisfy, so the
/// whole season backs off together.
pub(crate) async fn record_episode_search(
database: &Db,
episode_ids: &[i64],
) -> Result<(), GrabError> {
for episode_id in episode_ids {
sqlx::query!(
"UPDATE episodes
SET search_attempts = search_attempts + 1,
last_searched_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now'),
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
WHERE id = ?",
episode_id
)
.execute(database.pool())
.await?;
}
Ok(())
}
/// #108, overriding #86: a torrent removed by hand is human intent, not a
/// gap to refill. Clear `wanted` (the leaf intent, DESIGN.md §4.1) and mark
/// the target `parked` instead of reopening it, so neither targeted search
/// nor RSS matching (§6.2) ever pick it back up. A season pack parks only
/// the episodes it was still covering (mirrors `import::hard_fail_tv`'s
/// season case), leaving ones already imported from a partial pack alone.
pub(crate) async fn park_target(
database: &Db,
target_kind: &str,
target_id: i64,
) -> Result<(), sqlx::Error> {
match target_kind {
"movie" => {
sqlx::query!(
"UPDATE movies SET wanted = 0, state = 'parked',
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
WHERE id = ?",
target_id
)
.execute(database.pool())
.await?;
}
"episode" => {
sqlx::query!(
"UPDATE episodes SET wanted = 0, state = 'parked',
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
WHERE id = ?",
target_id
)
.execute(database.pool())
.await?;
}
"season" => {
sqlx::query!(
"UPDATE episodes
SET wanted = 0, state = 'parked',
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
WHERE season_id = ? AND state = 'downloading'
AND NOT EXISTS (
SELECT 1 FROM media_files f
WHERE f.owner_kind = 'episode' AND f.owner_id = episodes.id
)",
target_id
)
.execute(database.pool())
.await?;
}
other => unreachable!("grabs.target_kind CHECK constraint excludes {other:?}"),
}
Ok(())
}
/// `movies-main`, `tv-kids` (§7.1). Distinct from Radarr's own labels, so
/// both stacks can run against one Transmission.
fn label(loaded: &MoviePolicy) -> String {
label_for_root(&loaded.root_kind, &loaded.root_audience)
}
pub(crate) fn label_for_root(kind: &str, audience: &str) -> String {
let kind = if kind == "movie" { "movies" } else { kind };
format!("{kind}-{audience}")
}
/// A resolved download, in the shape Transmission takes it.
fn torrent_source(download: Download) -> TorrentSource {
match download {
Download::Magnet(uri) => TorrentSource::Magnet(uri),
Download::Torrent(bytes) => TorrentSource::Metainfo(bytes),
}
}
fn search_query(movie: &PendingMovie) -> String {
movie.year.map_or_else(
|| movie.title.clone(),
|year| format!("{} {year}", movie.title),
)
}
fn verdict_columns(verdict: &Verdict) -> (&'static str, Option<String>) {
match verdict {
Verdict::Eligible => ("eligible", None),
Verdict::Waived(_) => ("waived", None),
Verdict::Rejected(rule) => ("rejected", Some(rule.name())),
}
}
fn rfc3339(time: SystemTime) -> Option<String> {
let seconds = time.duration_since(UNIX_EPOCH).ok()?.as_secs();
let seconds = i64::try_from(seconds).ok()?;
chrono::DateTime::from_timestamp(seconds, 0).map(|date| date.to_rfc3339())
}
/// Fixture support for the download-link resolve step (issue #100). Grab
/// fixtures carry indexer links, and every lane now fetches them, so the mock
/// indexer has to answer the way Prowlarr does.
#[cfg(test)]
#[allow(clippy::unwrap_used)]
pub(crate) mod test_downloads {
use wiremock::matchers::{method, path_regex};
use wiremock::{Mock, MockServer, Request, Respond, ResponseTemplate};
/// Where rewritten fixture links point on the mock indexer.
const PREFIX: &str = "/dl/";
/// Rewrites the download links in a recorded feed onto `server`.
pub(crate) fn rewrite(feed: &str, links: &str, server: &MockServer) -> String {
feed.replace(links, &format!("{}{PREFIX}", server.uri()))
}
/// Prowlarr's live behaviour: the download endpoint 302s to a magnet.
pub(crate) async fn mount(server: &MockServer) {
Mock::given(method("GET"))
.and(path_regex(format!("^{PREFIX}")))
.respond_with(MagnetRedirect)
.mount(server)
.await;
}
struct MagnetRedirect;
impl Respond for MagnetRedirect {
fn respond(&self, request: &Request) -> ResponseTemplate {
let name = request
.url
.path()
.rsplit('/')
.next()
.unwrap_or_default()
.to_owned();
// Deterministic stand-in for the infohash the tracker would give,
// and distinct per release so duplicates still collapse.
let mut hash: u64 = 5381;
for byte in name.as_bytes() {
hash = hash.wrapping_mul(33) ^ u64::from(*byte);
}
ResponseTemplate::new(302).insert_header(
"location",
format!("magnet:?xt=urn:btih:{hash:040x}&dn={name}"),
)
}
}
}
#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
use base64::Engine as _;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use serde_json::{json, Value};
use wiremock::matchers::{method, path, query_param};
use wiremock::{Mock, MockServer, Request, Respond, ResponseTemplate};
use super::*;
/// A Transmission that dedupes on the infohash, like the real one: the
/// same source added twice is one torrent and a `torrent-duplicate`
/// response. That is the property the restart case leans on.
#[derive(Clone, Debug, Default)]
struct FakeTransmission {
torrents: Arc<Mutex<Vec<FakeTorrent>>>,
}
#[derive(Clone, Debug)]
struct FakeTorrent {
id: i64,
hash: String,
source: String,
labels: Vec<String>,
progress: f64,
}
impl FakeTransmission {
fn torrents(&self) -> Vec<FakeTorrent> {
self.torrents.lock().unwrap().clone()
}
fn complete_all(&self) {
for torrent in self.torrents.lock().unwrap().iter_mut() {
torrent.progress = 1.0;
}
}
fn add(&self, arguments: &Value) -> ResponseTemplate {
// A magnet arrives as `filename`, a torrent body as base64
// `metainfo` — either identifies the torrent for the fake.
let source = arguments["filename"]
.as_str()
.or_else(|| arguments["metainfo"].as_str())
.unwrap_or_default()
.to_owned();
let labels: Vec<String> = arguments["labels"]
.as_array()
.map(|values| {
values
.iter()
.filter_map(|value| value.as_str().map(ToOwned::to_owned))
.collect()
})
.unwrap_or_default();
let mut torrents = self.torrents.lock().unwrap();
if let Some(existing) = torrents.iter().find(|torrent| torrent.source == source) {
return success(&json!({"torrent-duplicate": {
"id": existing.id, "name": existing.source, "hashString": existing.hash
}}));
}
let id = i64::try_from(torrents.len()).unwrap() + 1;
// Deterministic stand-in for the real infohash, which likewise
// comes out the same for the same torrent.
let hash = format!("{:040x}", id * 7);
torrents.push(FakeTorrent {
id,
hash: hash.clone(),
source: source.clone(),
labels,
progress: 0.0,
});
success(&json!({"torrent-added": {"id": id, "name": source, "hashString": hash}}))
}
}
impl Respond for FakeTransmission {
fn respond(&self, request: &Request) -> ResponseTemplate {
let body: Value = serde_json::from_slice(&request.body).unwrap();
let arguments = &body["arguments"];
match body["method"].as_str().unwrap_or_default() {
"torrent-add" => self.add(arguments),
"torrent-get" => {
let torrents: Vec<Value> = self
.torrents()
.into_iter()
.map(|torrent| {
json!({
"id": torrent.id, "name": torrent.source,
"hashString": torrent.hash, "status": 4,
"percentDone": torrent.progress,
"downloadDir": "/downloads", "labels": torrent.labels
})
})
.collect();
success(&json!({"torrents": torrents}))
}
"torrent-remove" => {
let removed: Vec<i64> = arguments["ids"]
.as_array()
.map(|ids| ids.iter().filter_map(serde_json::Value::as_i64).collect())
.unwrap_or_default();
self.torrents
.lock()
.unwrap()
.retain(|torrent| !removed.contains(&torrent.id));
success(&json!({}))
}
_ => success(&json!({})),
}
}
}
fn success(arguments: &Value) -> ResponseTemplate {
ResponseTemplate::new(200).set_body_json(json!({
"result": "success", "arguments": arguments
}))
}
const RSS: &str = r#"<rss><channel>
<item>
<title>Dune.Part.Two.2024.2160p.WEB-DL.DDP5.1.Atmos</title>
<guid>good</guid><link>https://tracker/good.torrent</link>
<size>23622320128</size>
<torznab:attr xmlns:torznab="http://torznab.com/schemas/2015/feed" name="seeders" value="40"/>
</item>
<item>
<title>Dune.Part.Two.2024.2160p.REMUX</title>
<guid>huge</guid><link>https://tracker/huge.torrent</link>
<size>64424509440</size>
<torznab:attr xmlns:torznab="http://torznab.com/schemas/2015/feed" name="seeders" value="12"/>
</item>
<item>
<title>Dune.Part.Two.2024.1080p.CAM</title>
<guid>cam</guid><link>https://tracker/cam.torrent</link>
<size>4000000000</size>
<torznab:attr xmlns:torznab="http://torznab.com/schemas/2015/feed" name="seeders" value="900"/>
</item>
</channel></rss>"#;
const EMPTY_RSS: &str = "<rss><channel></channel></rss>";
const RELEASED_METADATA: &str = r#"{
"id": 693134,
"title": "Dune Part Two",
"original_title": "Dune: Part Two",
"original_language": "en",
"origin_country": ["US"],
"release_date": "2024-02-27",
"release_dates": {"results": [{"release_dates": [{
"type": 4, "release_date": "2024-04-16T00:00:00.000Z"
}]}]}
}"#;
const UNRELEASED_METADATA: &str = r#"{
"id": 693134,
"title": "Dune Part Two",
"original_title": "Dune: Part Two",
"original_language": "en",
"origin_country": ["US"],
"release_date": "2024-02-27",
"release_dates": {"results": []}
}"#;
async fn prowlarr() -> MockServer {
let server = MockServer::start().await;
let feed = test_downloads::rewrite(RSS, "https://tracker/", &server);
test_downloads::mount(&server).await;
mount_indexer(&server, &feed).await;
server
}
/// One indexer advertising a text search, serving `feed` to every query.
async fn mount_indexer(server: &MockServer, feed: &str) {
Mock::given(method("GET"))
.and(path("/api/v1/indexer"))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(json!([{"id": 7, "name": "tracker", "enable": true}])),
)
.mount(server)
.await;
Mock::given(method("GET"))
.and(path("/7/api"))
.and(query_param("t", "caps"))
.respond_with(ResponseTemplate::new(200).set_body_string(
r#"<caps><searching><search available="yes" supportedParams="q"/></searching></caps>"#,
))
.mount(server)
.await;
Mock::given(method("GET"))
.and(path("/7/api"))
.and(query_param("t", "search"))
.respond_with(ResponseTemplate::new(200).set_body_string(feed))
.mount(server)
.await;
}
async fn empty_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": "tracker", "enable": true}])),
)
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/7/api"))
.and(query_param("t", "caps"))
.respond_with(ResponseTemplate::new(200).set_body_string(
r#"<caps><searching><search available="yes" supportedParams="q"/></searching></caps>"#,
))
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/7/api"))
.and(query_param("t", "search"))
.respond_with(ResponseTemplate::new(200).set_body_string(EMPTY_RSS))
.mount(&server)
.await;
server
}
async fn tmdb(metadata: &str) -> MockServer {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/3/movie/693134"))
.and(query_param("append_to_response", "release_dates"))
.respond_with(ResponseTemplate::new(200).set_body_string(metadata))
.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)
}
async fn wanted_movie() -> (tempfile::TempDir, Db) {
let dir = tempfile::tempdir().unwrap();
let database = Db::connect(dir.path().join("arr.db")).await.unwrap();
database.migrate().await.unwrap();
sqlx::query(
"INSERT INTO movies (tmdb_id, title, year, original_language, root_id)
SELECT 693134, 'Dune Part Two', 2024, 'en', id
FROM roots WHERE kind = 'movie' AND audience = 'main'",
)
.execute(database.pool())
.await
.unwrap();
(dir, database)
}
fn action(prowlarr: &MockServer, transmission: &MockServer) -> GrabAction {
GrabAction::new(
ProwlarrClient::new(prowlarr.uri(), "key").unwrap(),
TransmissionClient::new(&transmission.uri()).unwrap(),
PathBuf::from("/mnt/media/transmission/complete"),
SeedingRules::new(
SeedingLimits {
ratio: 1.5,
idle_minutes: 60,
},
HashMap::new(),
),
)
}
fn action_with_tmdb(
prowlarr: &MockServer,
transmission: &MockServer,
metadata: &MockServer,
) -> GrabAction {
action(prowlarr, transmission).with_tmdb(Arc::new(
TmdbClient::builder("key")
.base_url(format!("{}/3/", metadata.uri()))
.build()
.unwrap(),
))
}
async fn targeted_searches(indexer: &MockServer) -> usize {
indexer
.received_requests()
.await
.unwrap()
.into_iter()
.filter(|request| {
request.url.path() == "/7/api"
&& request
.url
.query_pairs()
.any(|(name, value)| name == "t" && value == "search")
})
.count()
}
#[test]
fn seeding_rules_select_by_prowlarr_indexer_id() {
let default = SeedingLimits {
ratio: 1.0,
idle_minutes: 60,
};
let tracker = SeedingLimits {
ratio: 2.5,
idle_minutes: 120,
};
let rules = SeedingRules::new(default, HashMap::from([(7, tracker)]));
assert_eq!(rules.for_indexer(7), tracker);
assert_eq!(rules.for_indexer(8), default);
}
async fn grabs(database: &Db) -> Vec<(i64, String, String)> {
sqlx::query_as::<_, (i64, String, String)>(
"SELECT target_id, infohash, state FROM grabs ORDER BY id",
)
.fetch_all(database.pool())
.await
.unwrap()
}
/// The acceptance case: one wanted movie, one torrent, one grab row.
#[tokio::test]
async fn a_wanted_movie_ends_with_one_torrent_and_one_grab() {
let (_dir, database) = wanted_movie().await;
let indexer = prowlarr().await;
let (downloader, fake) = transmission().await;
let outcomes = action(&indexer, &downloader).tick(&database).await.unwrap();
assert_eq!(outcomes.len(), 1);
assert_eq!(fake.torrents().len(), 1);
let grabs = grabs(&database).await;
assert_eq!(grabs.len(), 1);
assert_eq!(grabs[0].0, 1);
assert_eq!(grabs[0].2, "sent");
// §5.5: the 22 GB WEB-DL at target beats the 60 GB remux, and the CAM
// is a hard filter rather than a low score however many seeders it has.
assert!(fake.torrents()[0].source.ends_with("good.torrent"));
let state: String = sqlx::query_scalar("SELECT state FROM movies WHERE id = 1")
.fetch_one(database.pool())
.await
.unwrap();
assert_eq!(state, "downloading");
}
/// The failure mode the issue names: killed after the torrent is sent and
/// before the row is written, a restart must not send a second one.
#[tokio::test]
async fn a_restart_mid_flight_does_not_grab_twice() {
let (_dir, database) = wanted_movie().await;
let indexer = prowlarr().await;
let (downloader, fake) = transmission().await;
action(&indexer, &downloader).tick(&database).await.unwrap();
// The crash: Transmission has the torrent, the database does not know.
sqlx::query("DELETE FROM grabs")
.execute(database.pool())
.await
.unwrap();
sqlx::query("UPDATE movies SET state = 'missing'")
.execute(database.pool())
.await
.unwrap();
// A fresh action, as a restarted process would build.
action(&indexer, &downloader).tick(&database).await.unwrap();
assert_eq!(fake.torrents().len(), 1);
assert_eq!(grabs(&database).await.len(), 1);
}
/// A title with a grab in flight is not a gap, so a settled tick is idle.
#[tokio::test]
async fn a_second_tick_grabs_nothing_new() {
let (_dir, database) = wanted_movie().await;
let indexer = prowlarr().await;
let (downloader, fake) = transmission().await;
let action = action(&indexer, &downloader);
action.tick(&database).await.unwrap();
let outcomes = action.tick(&database).await.unwrap();
assert!(outcomes.is_empty());
assert_eq!(fake.torrents().len(), 1);
assert_eq!(grabs(&database).await.len(), 1);
}
/// §7.1 and §7.3: the label and both seeding limits are set at add time,
/// not patched afterwards.
#[tokio::test]
async fn the_label_and_both_seed_limits_are_set_at_add_time() {
let (_dir, database) = wanted_movie().await;
let indexer = prowlarr().await;
let (downloader, fake) = transmission().await;
action(&indexer, &downloader).tick(&database).await.unwrap();
assert_eq!(fake.torrents()[0].labels, vec!["movies-main".to_owned()]);
let add = downloader
.received_requests()
.await
.unwrap()
.into_iter()
.filter_map(|request| serde_json::from_slice::<Value>(&request.body).ok())
.find(|body| body["method"] == "torrent-add")
.expect("torrent-add");
assert_eq!(add["arguments"]["seedRatioLimit"], json!(1.5));
assert_eq!(add["arguments"]["seedIdleLimit"], json!(60));
assert_eq!(add["arguments"]["seedRatioMode"], json!(1));
assert_eq!(add["arguments"]["seedIdleMode"], json!(1));
assert_eq!(
add["arguments"]["download-dir"],
json!("/mnt/media/transmission/complete")
);
}
/// Every candidate is cached with its verdict, which is what the manual
/// search view and the attention queues read (§9.3).
#[tokio::test]
async fn every_candidate_is_recorded_with_its_verdict() {
let (_dir, database) = wanted_movie().await;
let indexer = prowlarr().await;
let (downloader, _fake) = transmission().await;
action(&indexer, &downloader).tick(&database).await.unwrap();
let rows = sqlx::query_as::<_, (String, String, Option<String>)>(
"SELECT r.name, r.verdict, r.rejected_rule
FROM releases r JOIN movie_releases mr ON mr.release_id = r.id
WHERE mr.movie_id = 1 ORDER BY r.guid",
)
.fetch_all(database.pool())
.await
.unwrap();
assert_eq!(rows.len(), 3);
let cam = rows.iter().find(|row| row.0.contains("CAM")).unwrap();
assert_eq!(cam.1, "rejected");
assert_eq!(cam.2.as_deref(), Some("source"));
let attempts: i64 = sqlx::query_scalar("SELECT search_attempts FROM movies WHERE id = 1")
.fetch_one(database.pool())
.await
.unwrap();
assert_eq!(attempts, 0);
}
/// Transmission is authoritative (§8): a completed torrent moves its grab
/// out of `sent` without the process having watched it happen.
#[tokio::test]
async fn a_completed_torrent_moves_its_grab_to_downloaded() {
let (_dir, database) = wanted_movie().await;
let indexer = prowlarr().await;
let (downloader, fake) = transmission().await;
let action = action(&indexer, &downloader);
action.tick(&database).await.unwrap();
fake.complete_all();
let outcomes = action.tick(&database).await.unwrap();
assert_eq!(outcomes.len(), 1);
assert_eq!(grabs(&database).await[0].2, "downloaded");
}
/// #108, overriding §86: a torrent removed by hand — gone from
/// Transmission before it finished — parks the movie instead of
/// reopening the gap, and is marked `vanished` rather than `failed` so
/// it never counts toward the `needs_decision` queue (attention.rs).
#[tokio::test]
async fn a_torrent_removed_by_hand_parks_the_movie() {
let (_dir, database) = wanted_movie().await;
let indexer = prowlarr().await;
let (downloader, fake) = transmission().await;
action(&indexer, &downloader).tick(&database).await.unwrap();
fake.torrents.lock().unwrap().clear();
let outcomes = action(&indexer, &downloader).tick(&database).await.unwrap();
assert_eq!(outcomes.len(), 1);
assert_eq!(grabs(&database).await[0].2, "vanished");
let (state, wanted): (String, bool) =
sqlx::query_as("SELECT state, wanted FROM movies WHERE id = 1")
.fetch_one(database.pool())
.await
.unwrap();
assert_eq!(state, "parked");
assert!(!wanted, "the leaf intent is cleared, DESIGN.md §4.1");
}
/// #108: a parked movie is not re-grabbed by the next tick, even with a
/// matching release still available — `wanted = 0` removes it from the
/// work list (§8), which is the whole point of parking rather than
/// reopening.
#[tokio::test]
async fn a_parked_movie_is_not_regrabbed() {
let (_dir, database) = wanted_movie().await;
let indexer = prowlarr().await;
let (downloader, fake) = transmission().await;
let action = action(&indexer, &downloader);
action.tick(&database).await.unwrap();
fake.torrents.lock().unwrap().clear();
action.tick(&database).await.unwrap();
let outcomes = action.tick(&database).await.unwrap();
assert!(outcomes.is_empty(), "parked, nothing left to do");
assert_eq!(fake.torrents().len(), 0, "no fresh torrent sent");
let grabs = grabs(&database).await;
assert_eq!(grabs.len(), 1);
assert_eq!(grabs[0].2, "vanished");
}
/// #108: nothing blacklists the vanished release — a one-click re-want
/// (the manual trigger) can still grab it again, reclaiming the dead
/// row rather than losing the grab to the `infohash` uniqueness
/// constraint.
#[tokio::test]
async fn a_rewanted_movie_can_regrab_the_same_infohash() {
let (_dir, database) = wanted_movie().await;
let indexer = prowlarr().await;
let (downloader, fake) = transmission().await;
let action = action(&indexer, &downloader);
action.tick(&database).await.unwrap();
fake.torrents.lock().unwrap().clear();
action.tick(&database).await.unwrap();
sqlx::query("UPDATE movies SET wanted = 1 WHERE id = 1")
.execute(database.pool())
.await
.unwrap();
let outcomes = action.tick(&database).await.unwrap();
assert_eq!(outcomes.len(), 1, "a fresh grab");
assert_eq!(fake.torrents().len(), 1, "same magnet, same infohash");
let grabs = grabs(&database).await;
assert_eq!(grabs.len(), 1, "the dead row is reclaimed, not duplicated");
assert_eq!(grabs[0].2, "sent");
let state: String = sqlx::query_scalar("SELECT state FROM movies WHERE id = 1")
.fetch_one(database.pool())
.await
.unwrap();
assert_eq!(state, "downloading");
}
/// §5.2: the language rules are expressed against the title's original
/// language, and guessing it is worse than waiting for it.
#[tokio::test]
async fn a_title_with_no_original_language_is_not_searched() {
let (_dir, database) = wanted_movie().await;
sqlx::query("UPDATE movies SET original_language = NULL")
.execute(database.pool())
.await
.unwrap();
let indexer = prowlarr().await;
let (downloader, fake) = transmission().await;
let outcomes = action(&indexer, &downloader).tick(&database).await.unwrap();
assert!(outcomes.is_empty());
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 reaches every trigger, so the exclusion is applied where a
/// release is classified rather than where one is picked: the cached row
/// says `blacklisted`, which is also what stops §9.3's manual view from
/// offering it as a clean match.
#[tokio::test]
async fn a_blacklisted_release_is_cached_as_rejected() {
let (_dir, database) = wanted_movie().await;
blacklist::add(
database.pool(),
Some("ffff"),
"Dune.Part.Two.2024.2160p.WEB-DL.DDP5.1.Atmos",
"dolby_vision_profile",
)
.await
.unwrap();
let indexer = prowlarr().await;
let (downloader, _fake) = transmission().await;
action(&indexer, &downloader).tick(&database).await.unwrap();
let (verdict, rule): (String, Option<String>) =
sqlx::query_as("SELECT verdict, rejected_rule FROM releases WHERE guid = 'good'")
.fetch_one(database.pool())
.await
.unwrap();
assert_eq!(verdict, "rejected");
assert_eq!(rule.as_deref(), Some("blacklisted"));
}
/// §6.3's second key. A `.torrent` link hides its infohash until
/// Transmission has fetched it, so the blacklisted torrent is only
/// recognised after the add — and must not leave a grab behind.
#[tokio::test]
async fn a_blacklisted_infohash_never_becomes_a_grab() {
let (_dir, database) = wanted_movie().await;
// What the fake hands back for the first torrent it accepts.
blacklist::add(
database.pool(),
Some(&format!("{:040x}", 7)),
"Some.Older.Name.Of.The.Same.Torrent",
"required_audio",
)
.await
.unwrap();
let indexer = prowlarr().await;
let (downloader, fake) = transmission().await;
action(&indexer, &downloader).tick(&database).await.unwrap();
assert!(grabs(&database).await.is_empty());
assert!(
fake.torrents().is_empty(),
"a torrent added this tick and then found blacklisted is removed"
);
// Recorded under its new name, so the next tick stops before paying
// Transmission again.
let blacklist = Blacklist::load(database.pool()).await.unwrap();
assert!(blacklist.blocks_name("Dune.Part.Two.2024.2160p.WEB-DL.DDP5.1.Atmos"));
// And the cached row stops reading eligible straight away, so §9.3's
// manual view never offers what this tick just refused.
let (verdict, rule): (String, Option<String>) =
sqlx::query_as("SELECT verdict, rejected_rule FROM releases WHERE guid = 'good'")
.fetch_one(database.pool())
.await
.unwrap();
assert_eq!(verdict, "rejected");
assert_eq!(rule.as_deref(), Some("blacklisted"));
}
/// A movie that is already on disk. `wanted` stays set: §9.3's upgrade
/// flow is exactly a satisfied movie still looking for something better.
async fn available_movie() -> (tempfile::TempDir, Db) {
let (dir, database) = wanted_movie().await;
sqlx::query(
"INSERT INTO media_files (owner_kind, owner_id, path, size)
VALUES ('movie', 1, '/mnt/media/movies/Dune Part Two (2024).mkv', 1)",
)
.execute(database.pool())
.await
.unwrap();
sqlx::query("UPDATE movies SET state = 'available' WHERE id = 1")
.execute(database.pool())
.await
.unwrap();
(dir, database)
}
async fn deck(database: &Db) -> Vec<(String, String)> {
sqlx::query_as::<_, (String, String)>(
"SELECT r.name, r.verdict
FROM releases r JOIN movie_releases mr ON mr.release_id = r.id
WHERE mr.movie_id = 1 ORDER BY r.guid",
)
.fetch_all(database.pool())
.await
.unwrap()
}
/// Issue #115: a manual search on an available movie sweeps the indexers
/// and refreshes the deck §9.3 reads, and grabs nothing. The old code
/// loaded the movie through the gap filter, so this was a silent no-op.
#[tokio::test]
async fn a_manual_search_on_an_available_movie_refreshes_the_deck() {
let (_dir, database) = available_movie().await;
let indexer = prowlarr().await;
let (downloader, fake) = transmission().await;
let outcome = action(&indexer, &downloader)
.search_now(&database, 1)
.await
.unwrap();
assert!(outcome.is_none(), "an available movie must not auto-grab");
assert_eq!(targeted_searches(&indexer).await, 1);
assert_eq!(deck(&database).await.len(), 3, "every candidate is cached");
assert!(fake.torrents().is_empty());
assert!(grabs(&database).await.is_empty());
}
/// The digital-release gate belongs to targeted search (§6.2). A movie
/// with a file on disk is released whatever TMDB knows, so it must not
/// silence the manual deck refresh.
#[tokio::test]
async fn an_available_movie_refreshes_its_deck_before_the_digital_release() {
let (_dir, database) = available_movie().await;
let indexer = prowlarr().await;
let metadata = tmdb(UNRELEASED_METADATA).await;
let (downloader, _fake) = transmission().await;
action_with_tmdb(&indexer, &downloader, &metadata)
.search_now(&database, 1)
.await
.unwrap();
assert_eq!(deck(&database).await.len(), 3);
}
/// A grab already in flight is not a gap either: refresh the deck, do not
/// send a second torrent.
#[tokio::test]
async fn a_manual_search_with_a_grab_in_flight_does_not_grab_again() {
let (_dir, database) = wanted_movie().await;
let indexer = prowlarr().await;
let (downloader, fake) = transmission().await;
let action = action(&indexer, &downloader);
action.tick(&database).await.unwrap();
assert_eq!(grabs(&database).await.len(), 1);
action.search_now(&database, 1).await.unwrap();
assert_eq!(fake.torrents().len(), 1);
assert_eq!(grabs(&database).await.len(), 1);
}
/// The gap lane is unchanged: a wanted movie with nothing on disk still
/// searches and grabs from the manual trigger.
#[tokio::test]
async fn a_manual_search_on_a_wanted_movie_still_grabs() {
let (_dir, database) = wanted_movie().await;
let indexer = prowlarr().await;
let (downloader, fake) = transmission().await;
let outcome = action(&indexer, &downloader)
.search_now(&database, 1)
.await
.unwrap();
assert!(outcome.is_some());
assert_eq!(fake.torrents().len(), 1);
assert_eq!(grabs(&database).await.len(), 1);
}
/// §6.3: blocked refuses the manual trigger too, deck refresh included.
#[tokio::test]
async fn a_manual_search_on_a_blocked_movie_calls_no_indexer() {
let (_dir, database) = available_movie().await;
sqlx::query("UPDATE movies SET blocked = 1")
.execute(database.pool())
.await
.unwrap();
let indexer = prowlarr().await;
let (downloader, fake) = transmission().await;
let outcome = action(&indexer, &downloader)
.search_now(&database, 1)
.await
.unwrap();
assert!(outcome.is_none());
assert_eq!(targeted_searches(&indexer).await, 0);
assert!(deck(&database).await.is_empty());
assert!(fake.torrents().is_empty());
}
/// §6.3: `blocked` stops targeted search for a title.
#[tokio::test]
async fn a_blocked_title_is_not_searched() {
let (_dir, database) = wanted_movie().await;
sqlx::query("UPDATE movies SET blocked = 1")
.execute(database.pool())
.await
.unwrap();
let indexer = prowlarr().await;
let (downloader, fake) = transmission().await;
let outcomes = action(&indexer, &downloader).tick(&database).await.unwrap();
assert!(outcomes.is_empty());
assert!(fake.torrents().is_empty());
assert!(grabs(&database).await.is_empty());
}
#[tokio::test]
async fn unreleased_movies_never_call_an_indexer() {
let (_dir, database) = wanted_movie().await;
let indexer = prowlarr().await;
let metadata = tmdb(UNRELEASED_METADATA).await;
let (downloader, _fake) = transmission().await;
let action = action_with_tmdb(&indexer, &downloader, &metadata);
for _ in 0..3 {
action.tick(&database).await.unwrap();
}
assert!(indexer.received_requests().await.unwrap().is_empty());
let attempts: i64 = sqlx::query_scalar("SELECT search_attempts FROM movies")
.fetch_one(database.pool())
.await
.unwrap();
assert_eq!(attempts, 0);
}
#[tokio::test]
async fn unsuccessful_searches_follow_the_backoff_schedule() {
let (_dir, database) = wanted_movie().await;
let indexer = empty_prowlarr().await;
let metadata = tmdb(RELEASED_METADATA).await;
let (downloader, _fake) = transmission().await;
let action = action_with_tmdb(&indexer, &downloader, &metadata);
action.tick(&database).await.unwrap();
assert_eq!(targeted_searches(&indexer).await, 1);
action.tick(&database).await.unwrap();
assert_eq!(targeted_searches(&indexer).await, 1);
for (delay, expected_searches) in [("-1 hour", 2), ("-6 hours", 3), ("-1 day", 4)] {
sqlx::query(
"UPDATE movies SET last_searched_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', ?)",
)
.bind(delay)
.execute(database.pool())
.await
.unwrap();
action.tick(&database).await.unwrap();
assert_eq!(targeted_searches(&indexer).await, expected_searches);
}
sqlx::query(
"UPDATE movies SET search_attempts = 5, last_searched_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-6 days')",
)
.execute(database.pool())
.await
.unwrap();
action.tick(&database).await.unwrap();
assert_eq!(targeted_searches(&indexer).await, 4);
sqlx::query(
"UPDATE movies SET last_searched_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-7 days')",
)
.execute(database.pool())
.await
.unwrap();
action.tick(&database).await.unwrap();
assert_eq!(targeted_searches(&indexer).await, 5);
}
/// A title stuck on backoff must not pay for TMDB on every tick: the
/// metadata refresh has its own TTL, independent of the search backoff.
///
/// A fresh `GrabAction` (and so a fresh `TmdbClient`) is built for every
/// tick, as a restarted process would, so the only thing that can be
/// suppressing a real TMDB request is the persisted
/// `metadata_refreshed_at` gate rather than the client's own in-process
/// response cache.
#[tokio::test]
async fn metadata_refresh_is_throttled_by_its_own_ttl() {
let (_dir, database) = wanted_movie().await;
let indexer = empty_prowlarr().await;
let metadata = tmdb(RELEASED_METADATA).await;
let (downloader, _fake) = transmission().await;
action_with_tmdb(&indexer, &downloader, &metadata)
.tick(&database)
.await
.unwrap();
assert_eq!(metadata.received_requests().await.unwrap().len(), 1);
assert_eq!(targeted_searches(&indexer).await, 1);
// Due for another search attempt, but the metadata refresh is not
// due yet: TMDB is not called again, and the stored digital release
// still gates the search correctly.
sqlx::query(
"UPDATE movies SET last_searched_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-2 hours')",
)
.execute(database.pool())
.await
.unwrap();
action_with_tmdb(&indexer, &downloader, &metadata)
.tick(&database)
.await
.unwrap();
assert_eq!(metadata.received_requests().await.unwrap().len(), 1);
assert_eq!(targeted_searches(&indexer).await, 2);
// Past the refresh TTL: the next due attempt pays for TMDB again.
sqlx::query(
"UPDATE movies SET last_searched_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-7 hours'),
metadata_refreshed_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-7 hours')",
)
.execute(database.pool())
.await
.unwrap();
action_with_tmdb(&indexer, &downloader, &metadata)
.tick(&database)
.await
.unwrap();
assert_eq!(metadata.received_requests().await.unwrap().len(), 2);
assert_eq!(targeted_searches(&indexer).await, 3);
}
/// §9.6: the daily refresh also writes the three stored artwork fields,
/// so pure-SQL views render a poster without a TMDB call per row.
#[tokio::test]
async fn metadata_refresh_writes_the_stored_artwork_fields() {
let (_dir, database) = wanted_movie().await;
let indexer = empty_prowlarr().await;
let artwork_metadata = RELEASED_METADATA.replace(
r#""original_language": "en","#,
r#""original_language": "en",
"poster_path": "/dune-two.jpg",
"backdrop_path": "/dune-two-wide.jpg",
"vote_average": 8.152,"#,
);
let metadata = tmdb(&artwork_metadata).await;
let (downloader, _fake) = transmission().await;
action_with_tmdb(&indexer, &downloader, &metadata)
.tick(&database)
.await
.unwrap();
let (poster, backdrop, vote): (Option<String>, Option<String>, Option<f64>) =
sqlx::query_as("SELECT poster_path, backdrop_path, vote_average FROM movies")
.fetch_one(database.pool())
.await
.unwrap();
assert_eq!(poster.as_deref(), Some("/dune-two.jpg"));
assert_eq!(backdrop.as_deref(), Some("/dune-two-wide.jpg"));
assert_eq!(vote, Some(8.152));
// Idempotent: an unchanged refresh neither rewrites nor re-reports,
// same as the fields above it.
sqlx::query(
"UPDATE movies SET metadata_refreshed_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-7 hours')",
)
.execute(database.pool())
.await
.unwrap();
action_with_tmdb(&indexer, &downloader, &metadata)
.tick(&database)
.await
.unwrap();
let unchanged: (Option<String>, Option<String>, Option<f64>) =
sqlx::query_as("SELECT poster_path, backdrop_path, vote_average FROM movies")
.fetch_one(database.pool())
.await
.unwrap();
assert_eq!(unchanged, (poster, backdrop, vote));
}
/// TMDB reports `vote_average: 0` for a title nobody has rated — absence,
/// not zero (#156). It must store as NULL, be overwritten when votes
/// arrive, and go back to NULL if they are withdrawn.
#[tokio::test]
async fn metadata_refresh_stores_an_unrated_title_as_null() {
let (_dir, database) = wanted_movie().await;
let indexer = empty_prowlarr().await;
let unrated = || {
RELEASED_METADATA.replace(
r#""original_language": "en","#,
r#""original_language": "en",
"poster_path": "/dune-two.jpg",
"backdrop_path": "/dune-two-wide.jpg",
"vote_average": 0.0,"#,
)
};
let rated = || {
RELEASED_METADATA.replace(
r#""original_language": "en","#,
r#""original_language": "en",
"poster_path": "/dune-two.jpg",
"backdrop_path": "/dune-two-wide.jpg",
"vote_average": 8.152,"#,
)
};
let (downloader, _fake) = transmission().await;
action_with_tmdb(&indexer, &downloader, &tmdb(&unrated()).await)
.tick(&database)
.await
.unwrap();
let vote: Option<f64> =
sqlx::query_scalar("SELECT vote_average FROM movies WHERE tmdb_id IS NOT NULL")
.fetch_one(database.pool())
.await
.unwrap();
assert_eq!(vote, None);
// Votes arrive: the NULL is overwritten.
sqlx::query(
"UPDATE movies SET metadata_refreshed_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-7 hours')",
)
.execute(database.pool())
.await
.unwrap();
action_with_tmdb(&indexer, &downloader, &tmdb(&rated()).await)
.tick(&database)
.await
.unwrap();
let vote: Option<f64> =
sqlx::query_scalar("SELECT vote_average FROM movies WHERE tmdb_id IS NOT NULL")
.fetch_one(database.pool())
.await
.unwrap();
assert_eq!(vote, Some(8.152));
// And withdrawn again.
sqlx::query(
"UPDATE movies SET metadata_refreshed_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-7 hours')",
)
.execute(database.pool())
.await
.unwrap();
action_with_tmdb(&indexer, &downloader, &tmdb(&unrated()).await)
.tick(&database)
.await
.unwrap();
let vote: Option<f64> =
sqlx::query_scalar("SELECT vote_average FROM movies WHERE tmdb_id IS NOT NULL")
.fetch_one(database.pool())
.await
.unwrap();
assert_eq!(vote, None);
}
#[tokio::test]
async fn metadata_changes_reset_a_title_backoff() {
let (_dir, database) = wanted_movie().await;
sqlx::query(
"UPDATE movies SET search_attempts = 4, last_searched_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')",
)
.execute(database.pool())
.await
.unwrap();
let indexer = empty_prowlarr().await;
let metadata = tmdb(&RELEASED_METADATA.replace("Dune Part Two", "Dune: Part Two")).await;
let (downloader, _fake) = transmission().await;
action_with_tmdb(&indexer, &downloader, &metadata)
.tick(&database)
.await
.unwrap();
assert_eq!(targeted_searches(&indexer).await, 1);
let (title, attempts): (String, i64) =
sqlx::query_as("SELECT title, search_attempts FROM movies")
.fetch_one(database.pool())
.await
.unwrap();
assert_eq!(title, "Dune: Part Two");
assert_eq!(attempts, 1);
}
/// Prowlarr probes capabilities one indexer at a time, so paying for
/// discovery every 30 s would leave the tick no room to search.
#[tokio::test]
async fn indexer_discovery_is_cached_across_ticks() {
let (_dir, database) = wanted_movie().await;
let indexer = prowlarr().await;
let (downloader, _fake) = transmission().await;
let action = action(&indexer, &downloader);
action.tick(&database).await.unwrap();
action.tick(&database).await.unwrap();
let enumerations = indexer
.received_requests()
.await
.unwrap()
.into_iter()
.filter(|request| request.url.path() == "/api/v1/indexer")
.count();
assert_eq!(enumerations, 1);
}
/// One unresponsive tracker must not cancel the action before a single
/// search has run.
#[tokio::test]
async fn a_stalled_discovery_gives_up_instead_of_hanging() {
let (_dir, database) = wanted_movie().await;
let indexer = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/api/v1/indexer"))
.respond_with(
ResponseTemplate::new(200)
.set_delay(Duration::from_secs(30))
.set_body_json(json!([])),
)
.mount(&indexer)
.await;
let (downloader, fake) = transmission().await;
let mut action = action(&indexer, &downloader);
action.indexers.discovery_timeout = Duration::from_millis(50);
let error = action
.tick(&database)
.await
.expect_err("discovery cannot complete");
assert!(matches!(
error,
GrabError::Discovery(DiscoveryError::Timeout(_))
));
assert!(fake.torrents().is_empty());
}
/// §100: Transmission has no route to the indexer, so a download link
/// that answers with the torrent itself is forwarded as inline metainfo
/// and the link never leaves arr.
#[tokio::test]
async fn a_torrent_body_is_forwarded_inline_and_the_link_never_leaves_arr() {
const TORRENT: &[u8] = b"d4:infod6:lengthi1e4:name8:good.mkvee";
let (_dir, database) = wanted_movie().await;
let indexer = MockServer::start().await;
let feed = RSS.replace("https://tracker/", &format!("{}/dl/", indexer.uri()));
mount_indexer(&indexer, &feed).await;
Mock::given(method("GET"))
.and(path("/dl/good.torrent"))
.respond_with(
ResponseTemplate::new(200).set_body_raw(TORRENT, "application/x-bittorrent"),
)
.mount(&indexer)
.await;
let (downloader, fake) = transmission().await;
action(&indexer, &downloader).tick(&database).await.unwrap();
let sent = base64::engine::general_purpose::STANDARD
.decode(&fake.torrents()[0].source)
.expect("the torrent is sent as base64 metainfo");
assert_eq!(sent, TORRENT);
let fetched_with_key = indexer
.received_requests()
.await
.unwrap()
.into_iter()
.filter(|request| request.url.path() == "/dl/good.torrent")
.all(|request| request.headers.contains_key("x-api-key"));
assert!(fetched_with_key, "arr resolves the link with its own key");
}
/// A link that redirects to a magnet — Prowlarr's usual answer — reaches
/// Transmission as the magnet, which it can act on without the indexer.
#[tokio::test]
async fn a_link_that_redirects_to_a_magnet_is_sent_as_the_magnet() {
let (_dir, database) = wanted_movie().await;
let indexer = prowlarr().await;
let (downloader, fake) = transmission().await;
action(&indexer, &downloader).tick(&database).await.unwrap();
assert!(
fake.torrents()[0]
.source
.starts_with("magnet:?xt=urn:btih:"),
"{}",
fake.torrents()[0].source
);
}
#[test]
fn a_resolved_download_keeps_its_shape() {
assert_eq!(
torrent_source(Download::Magnet("magnet:?xt=urn:btih:abc".into())),
TorrentSource::Magnet("magnet:?xt=urn:btih:abc".into())
);
assert_eq!(
torrent_source(Download::Torrent(b"d4:infodee".to_vec())),
TorrentSource::Metainfo(b"d4:infodee".to_vec())
);
}
}
Reference in New Issue View Git Blame Copy Permalink