arr/crates/arr-daemon/src/grab.rs

//! 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::HashMap;
use std::path::PathBuf;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use arr_core::policy::{evaluate, Candidate};
use arr_core::{score::score, Language, Policy, TitleOverrides, Verdict};
use arr_db::{blacklist, Blacklist, Db, MoviePolicy};
use arr_dl::{AddTorrent, TorrentSource, TransmissionClient};
use arr_indexer::{ProwlarrClient, SearchRelease, SearchRequest};

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;

/// 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.
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct SeedingLimits {
    pub ratio: f64,
    pub idle_minutes: u64,
}

/// 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("prowlarr: {0}")]
    Prowlarr(#[from] arr_indexer::Error),
    #[error("transmission: {0}")]
    Transmission(#[from] arr_dl::Error),
    #[error("release {name}: {source}")]
    Parsed {
        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<i64>,
    refreshed_at: Option<Instant>,
}

/// 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,
    transmission: TransmissionClient,
    download_dir: PathBuf,
    seeding: SeedingLimits,
    indexers: tokio::sync::RwLock<IndexerCache>,
    /// [`INDEXER_DISCOVERY_TIMEOUT`], overridden by tests that cannot wait
    /// out the real one. Mirrors `ReconcileLoop`'s action timeout override.
    discovery_timeout: Duration,
}

impl GrabAction {
    #[must_use]
    pub fn new(
        prowlarr: ProwlarrClient,
        transmission: TransmissionClient,
        download_dir: PathBuf,
        seeding: SeedingLimits,
    ) -> Self {
        Self {
            prowlarr,
            transmission,
            download_dir,
            seeding,
            indexers: tokio::sync::RwLock::new(IndexerCache::default()),
            discovery_timeout: INDEXER_DISCOVERY_TIMEOUT,
        }
    }

    async fn tick(&self, database: &Db) -> Result<Vec<Outcome>, GrabError> {
        let mut outcomes = self.track_sent_grabs(database).await?;
        let gaps = pending_movies(database).await?;
        if gaps.is_empty() {
            return Ok(outcomes);
        }

        let searchable = self.searchable_indexers().await?;
        if searchable.is_empty() {
            tracing::warn!("no indexer advertises a text search; nothing can be grabbed");
            return Ok(outcomes);
        }

        for movie in gaps {
            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)
    }

    /// Move grabs Transmission reports as complete out of `sent`.
    ///
    /// 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.
    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_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 {
                // Gone from Transmission. Deciding whether that is a failure
                // or a manual removal is issue #86's; leaving the row alone
                // keeps this tick from re-grabbing behind the operator.
                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,
                movie_id = grab.target_id,
                "download complete"
            );
        }
        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<Vec<i64>, 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.
    ///
    /// 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);
        }
        record_search(database, movie.id).await?;

        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 {
            tracing::info!(
                movie_id = movie.id,
                title = movie.title,
                "no eligible release"
            );
            return Ok(None);
        };

        let added = self
            .transmission
            .add_torrent(AddTorrent {
                source: torrent_source(&winner.download_url),
                label: label(&loaded),
                download_dir: self.download_dir.clone(),
                seed_ratio_limit: self.seeding.ratio,
                seed_idle_limit_minutes: self.seeding.idle_minutes,
            })
            .await?;
        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.drop_blacklisted_torrent(database, movie, &winner.name, &added)
                .await?;
            return Ok(None);
        }

        // A duplicate here is the restart case: the torrent was added before
        // the process died. `DO NOTHING` keeps the original row.
        let inserted = sqlx::query!(
            r#"INSERT INTO grabs (release_id, target_kind, target_id, infohash, state)
               VALUES (?, 'movie', ?, ?, 'sent')
               ON CONFLICT (infohash) DO NOTHING
               RETURNING id AS "id!: i64""#,
            winner.id,
            movie.id,
            infohash
        )
        .fetch_optional(database.pool())
        .await?;
        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?;

        let Some(inserted) = inserted else {
            tracing::info!(
                movie_id = movie.id,
                infohash,
                "grab already recorded for this torrent"
            );
            return Ok(None);
        };

        tracing::info!(
            movie_id = movie.id,
            title = movie.title,
            release = winner.name,
            score = winner.score,
            infohash,
            was_duplicate = added.was_duplicate,
            "grabbed"
        );
        Ok(Some(Outcome::new(
            format!("movie {} wanted with no file", movie.id),
            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,
        movie: &PendingMovie,
        release_name: &str,
        added: &arr_dl::AddedTorrent,
    ) -> Result<(), GrabError> {
        blacklist::add(
            database.pool(),
            None,
            release_name,
            "blacklisted infohash under a new name",
        )
        .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!(
                movie_id = movie.id,
                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!(
                movie_id = movie.id,
                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,
    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>,
}

/// The eligible view of a stored release, ranked for selection.
#[derive(Debug, Clone)]
struct Eligible {
    id: i64,
    indexer_id: i64,
    guid: String,
    name: String,
    download_url: String,
    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.title AS "title!: String",
                 m.year,
                 m.original_language
        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,
            title: row.title,
            year: row.year,
            original_language: row.original_language,
        })
        .collect())
}

/// 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.
async fn store_release(
    database: &Db,
    movie_id: i64,
    release: &SearchRelease,
    policy: &Policy,
    overrides: &TitleOverrides,
    original_language: &Language,
    blacklist: &Blacklist,
) -> Result<Option<Eligible>, GrabError> {
    let parsed = arr_parse::parse(&release.name);
    let evaluation = evaluate(
        policy,
        overrides,
        original_language,
        Candidate::PreGrab(&parsed),
        release.size,
    );
    let scored = score(
        policy,
        Candidate::PreGrab(&parsed),
        release.size.unwrap_or_default(),
        release.seeders.unwrap_or_default(),
    );
    // 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?;

    sqlx::query!(
        "INSERT INTO movie_releases (movie_id, release_id) VALUES (?, ?)
         ON CONFLICT DO NOTHING",
        movie_id,
        id
    )
    .execute(database.pool())
    .await?;

    if verdict != "eligible" {
        return Ok(None);
    }
    Ok(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(())
}

/// `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 {
    let kind = if loaded.root_kind == "movie" {
        "movies"
    } else {
        &loaded.root_kind
    };
    format!("{kind}-{}", loaded.root_audience)
}

/// Indexers hand out magnets and `.torrent` links interchangeably;
/// Transmission takes either in the same field.
fn torrent_source(download_url: &str) -> TorrentSource {
    if download_url.starts_with("magnet:") {
        TorrentSource::Magnet(download_url.to_owned())
    } else {
        TorrentSource::Url(download_url.to_owned())
    }
}

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())
}

#[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};
    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 {
            let source = arguments["filename"]
                .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>"#;

    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": "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(RSS))
            .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"),
            SeedingLimits {
                ratio: 1.5,
                idle_minutes: 60,
            },
        )
    }

    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, 1);
    }

    /// 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");
    }

    /// §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"));
    }

    /// §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());
    }

    /// 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.discovery_timeout = Duration::from_millis(50);

        let error = action
            .tick(&database)
            .await
            .expect_err("discovery cannot complete");

        assert!(matches!(error, GrabError::IndexerDiscoveryTimeout(_)));
        assert!(fake.torrents().is_empty());
    }

    #[test]
    fn indexer_links_and_magnets_both_reach_transmission() {
        assert_eq!(
            torrent_source("magnet:?xt=urn:btih:abc"),
            TorrentSource::Magnet("magnet:?xt=urn:btih:abc".into())
        );
        assert_eq!(
            torrent_source("https://tracker/good.torrent"),
            TorrentSource::Url("https://tracker/good.torrent".into())
        );
    }
}