//! RSS sync: one empty-query Torznab call per indexer, matched against the
//! whole wanted list locally. See DESIGN.md §6.2.
//!
//! The property this lane is built around is that its cost is one call per
//! indexer, whatever the library holds. Nothing here is per-title, so every
//! wanted item is compared against every result on every pass, forever:
//! there is no backoff, no attempt counter, and no release-date gate. Those
//! belong to targeted search, which pays per title.
//!
//! `blocked` is honoured the other way round from targeted search (§6.3): it
//! stops a title being searched for, and leaves it matching RSS.
//!
//! One pass serves films and episodes alike: each feed item goes through the
//! movie matcher first, then the episode one (`arr_core::matching`). A
//! single-episode release grabs its wanted, missing episode directly; a
//! season pack only grabs when [`season_grab_mode`] allows packs for that
//! season — the same guard §14 gives targeted search, so a season behaves
//! the same however a release is found (§6.2). Like everything else here,
//! none of it backs off or counts attempts.

use std::collections::hash_map::Entry;
use std::collections::{HashMap, HashSet};
use std::path::PathBuf;

use arr_core::grabbing::{season_grab_mode, SeasonGrabFacts, SeasonGrabMode};
use arr_core::matching::{
    match_episode, match_movie, MatchKind, MatchShape, ReleaseIds, WantedEpisode, WantedMovie,
};
use arr_core::{EpisodeId, Language, MovieId};
use arr_db::{Blacklist, Db, MoviePolicy, TitlePolicy};
use arr_dl::TransmissionClient;
use arr_indexer::{ProwlarrClient, SearchRelease, SearchRequest};

use crate::grab::{
    backoff_elapsed, store_episode_release, store_release, Eligible, GrabError, GrabScope,
    GrabTarget, Grabber, SeedingRules,
};
use crate::indexers::IndexerDirectory;
use crate::reconcile::{Action, ActionFuture, Outcome};

/// Reads every indexer's feed and grabs what it can tie to a wanted title.
#[derive(Debug)]
pub struct RssAction {
    prowlarr: ProwlarrClient,
    grabber: Grabber,
    indexers: IndexerDirectory,
}

impl RssAction {
    #[must_use]
    pub fn new(
        prowlarr: ProwlarrClient,
        transmission: TransmissionClient,
        download_dir: PathBuf,
        seeding: SeedingRules,
    ) -> Self {
        Self {
            indexers: IndexerDirectory::new(prowlarr.clone()),
            grabber: Grabber::new(prowlarr.clone(), transmission, download_dir, seeding),
            prowlarr,
        }
    }

    async fn tick(&self, database: &Db) -> Result<Vec<Outcome>, GrabError> {
        let wanted = wanted_movies(database).await?;
        let episodes = wanted_episodes(database).await?;
        if wanted.is_empty() && episodes.is_empty() {
            return Ok(Vec::new());
        }
        let indexers = self.indexers.searchable().await?;
        if indexers.is_empty() {
            tracing::warn!("no indexer advertises a text search; no RSS feed to read");
            return Ok(Vec::new());
        }

        let releases = self.feeds(&indexers).await;
        let blacklist = Blacklist::load(database.pool()).await?;
        let (movies, tv) = self
            .match_feeds(database, &wanted, &episodes, releases, &blacklist)
            .await?;

        let mut outcomes = Vec::new();
        for (movie, winner) in movies {
            let outcome = self
                .grabber
                .send_winner(
                    database,
                    &GrabTarget {
                        scope: GrabScope::Movie { movie_id: movie.id },
                        title: &movie.title,
                        // §6.2: RSS never backs off, so a failed grab here
                        // must not spend one of targeted search's attempts.
                        counts_as_attempt: false,
                    },
                    &movie.policy,
                    &blacklist,
                    winner,
                )
                .await?;
            outcomes.extend(outcome);
        }
        for TvWinner {
            grabbable,
            scope,
            candidate: winner,
        } in tv
        {
            let outcome = self
                .grabber
                .send_winner(
                    database,
                    &GrabTarget {
                        scope,
                        title: &grabbable.series_title,
                        // Same property as the movie lane above.
                        counts_as_attempt: false,
                    },
                    &grabbable.policy,
                    &blacklist,
                    winner,
                )
                .await?;
            outcomes.extend(outcome);
        }
        Ok(outcomes)
    }

    /// One empty-query call per indexer (§6.2). A tracker that fails is
    /// logged and skipped: the next pass is ten minutes away.
    async fn feeds(&self, indexers: &[i64]) -> Vec<SearchRelease> {
        let mut releases = Vec::new();
        for search in self
            .prowlarr
            .search_indexers(indexers, &SearchRequest::Rss)
            .await
        {
            if let Some(error) = search.error {
                tracing::warn!(indexer_id = search.indexer_id, %error, "RSS feed failed");
            }
            releases.extend(search.releases);
        }
        releases
    }

    /// Tie each feed item to a wanted title, cache it with its verdict, and
    /// keep the best eligible candidate per title.
    ///
    /// One title takes at most one torrent per pass, and the ordering is the
    /// same total order targeted search uses, so the two triggers agree on
    /// which release wins.
    async fn match_feeds(
        &self,
        database: &Db,
        wanted: &[WantedMovie],
        episodes: &[WantedEpisodeRow],
        releases: Vec<SearchRelease>,
        blacklist: &Blacklist,
    ) -> Result<(Vec<(Grabbable, Eligible)>, Vec<TvWinner>), GrabError> {
        let mut policies: HashMap<i64, Option<Grabbable>> = HashMap::new();
        let mut best: HashMap<i64, Eligible> = HashMap::new();
        let mut tv = TvCandidates::default();

        for release in releases {
            let claims = arr_parse::parse(&release.name);
            let ids = ReleaseIds {
                tmdb_id: release.tmdb_id,
                imdb_id: release.imdb_id.clone(),
            };
            if let Some(matched) = match_movie(wanted, &ids, &claims) {
                let movie_id = matched.movie.0;
                let grabbable = match policies.entry(movie_id) {
                    Entry::Occupied(entry) => entry.into_mut(),
                    Entry::Vacant(entry) => {
                        entry.insert(load_grabbable(database, wanted, movie_id).await?)
                    }
                };
                let Some(grabbable) = grabbable.as_ref() else {
                    continue;
                };
                tracing::info!(
                    movie_id,
                    release = release.name,
                    by = match matched.kind {
                        MatchKind::TmdbId => "tmdb id",
                        MatchKind::ImdbId => "imdb id",
                        MatchKind::TitleAndYear => "title and year",
                    },
                    "RSS result matched a wanted title"
                );
                if let Some(candidate) = store_release(
                    database,
                    movie_id,
                    &release,
                    &grabbable.policy.policy,
                    &grabbable.policy.overrides,
                    &grabbable.language,
                    blacklist,
                )
                .await?
                {
                    let incumbent = best.get(&movie_id);
                    if incumbent.is_none_or(|incumbent| beats(&candidate, incumbent)) {
                        best.insert(movie_id, candidate);
                    }
                }
                continue;
            }
            tv.offer(database, episodes, &release, &ids, &claims, blacklist)
                .await?;
        }

        let tv_winners = tv.winners(database, episodes).await?;
        let mut movies: Vec<(Grabbable, Eligible)> = best
            .into_iter()
            .filter_map(|(movie_id, candidate)| {
                let grabbable = policies.get(&movie_id).and_then(Clone::clone)?;
                Some((grabbable, candidate))
            })
            .collect();
        movies.sort_by_key(|(movie, _)| movie.id);
        Ok((movies, tv_winners))
    }
}

/// The episode lane's state over one feed pass: the grab context per season
/// and the best eligible candidate per wanted episode and per season's pack.
#[derive(Default)]
struct TvCandidates {
    policies: HashMap<i64, Option<TvGrabbable>>,
    best_single: HashMap<i64, Eligible>,
    /// Per season: the best pack alongside the open episodes it covers.
    best_pack: HashMap<i64, (Eligible, Vec<i64>)>,
}

impl TvCandidates {
    /// Offer one feed item to every wanted episode. A release holds either
    /// exactly one episode or whole seasons, so each row it matches lands in
    /// exactly one bucket.
    async fn offer(
        &mut self,
        database: &Db,
        episodes: &[WantedEpisodeRow],
        release: &SearchRelease,
        ids: &ReleaseIds,
        claims: &arr_parse::NameClaims,
        blacklist: &Blacklist,
    ) -> Result<(), GrabError> {
        let mut packs: HashMap<i64, Vec<i64>> = HashMap::new();
        let mut single: Option<(i64, MatchKind)> = None;
        for row in episodes {
            let Some(matched) = match_episode(&row.wanted, ids, claims) else {
                continue;
            };
            match matched.shape {
                MatchShape::SeasonPack => {
                    packs
                        .entry(row.season_id)
                        .or_default()
                        .push(row.wanted.id.0);
                }
                MatchShape::SingleEpisode => single = Some((row.wanted.id.0, matched.kind)),
            }
        }
        for (season_id, covered) in packs {
            let Some(candidate) = self
                .store_candidate(database, episodes, season_id, &covered, release, blacklist)
                .await?
            else {
                continue;
            };
            let incumbent = self.best_pack.get(&season_id);
            if incumbent.is_none_or(|(incumbent, _)| beats(&candidate, incumbent)) {
                self.best_pack.insert(season_id, (candidate, covered));
            }
        }
        if let Some((episode_id, kind)) = single {
            tracing::info!(
                episode_id,
                release = release.name,
                by = tv_match_kind(kind),
                "RSS result matched a wanted episode"
            );
            let Some(row) = episodes.iter().find(|row| row.wanted.id.0 == episode_id) else {
                return Ok(());
            };
            let Some(candidate) = self
                .store_candidate(
                    database,
                    episodes,
                    row.season_id,
                    &[episode_id],
                    release,
                    blacklist,
                )
                .await?
            else {
                return Ok(());
            };
            let incumbent = self.best_single.get(&episode_id);
            if incumbent.is_none_or(|incumbent| beats(&candidate, incumbent)) {
                self.best_single.insert(episode_id, candidate);
            }
        }
        Ok(())
    }

    /// Classify and store a feed item against the open episodes it covers,
    /// loading the season's grab context on first use.
    async fn store_candidate(
        &mut self,
        database: &Db,
        episodes: &[WantedEpisodeRow],
        season_id: i64,
        covered: &[i64],
        release: &SearchRelease,
        blacklist: &Blacklist,
    ) -> Result<Option<Eligible>, GrabError> {
        let grabbable = match self.policies.entry(season_id) {
            Entry::Occupied(entry) => entry.into_mut(),
            Entry::Vacant(entry) => {
                entry.insert(load_tv_grabbable(database, episodes, season_id).await?)
            }
        };
        let Some(grabbable) = grabbable.as_ref() else {
            return Ok(None);
        };
        let (_, stored) = store_episode_release(
            database,
            covered,
            release,
            &grabbable.policy.policy,
            &grabbable.policy.overrides,
            &grabbable.language,
            blacklist,
        )
        .await?;
        Ok(stored)
    }

    /// Turn the per-season and per-episode winners into grabs. A season with
    /// an eligible pack only takes it when [`season_grab_mode`] allows packs
    /// there — the guard targeted search obeys (§6.2, §14) — and when it
    /// does, the season's singles stand down. Otherwise each open episode
    /// takes its own winner.
    async fn winners(
        &self,
        database: &Db,
        episodes: &[WantedEpisodeRow],
    ) -> Result<Vec<TvWinner>, GrabError> {
        let mut winners = Vec::new();
        let mut covered_by_packs: HashSet<i64> = HashSet::new();

        let mut seasons: Vec<i64> = self.best_pack.keys().copied().collect();
        seasons.sort_unstable();
        for season_id in seasons {
            let Some((candidate, episode_ids)) = self.best_pack.get(&season_id) else {
                continue;
            };
            let Some(grabbable) = self.policies.get(&season_id).and_then(Clone::clone) else {
                continue;
            };
            if !pack_allowed(database, season_id).await? {
                tracing::info!(
                    season_id,
                    series = grabbable.series_title,
                    "RSS skips a season pack the season's grab mode refuses"
                );
                continue;
            }
            covered_by_packs.extend(episode_ids.iter().copied());
            winners.push(TvWinner {
                grabbable,
                scope: GrabScope::Season {
                    season_id,
                    episode_ids: episode_ids.clone(),
                },
                candidate: candidate.clone(),
            });
        }

        let mut singles: Vec<i64> = self.best_single.keys().copied().collect();
        singles.sort_unstable();
        for episode_id in singles {
            if covered_by_packs.contains(&episode_id) {
                continue;
            }
            let Some(row) = episodes.iter().find(|row| row.wanted.id.0 == episode_id) else {
                continue;
            };
            let Some(grabbable) = self.policies.get(&row.season_id).and_then(Clone::clone) else {
                continue;
            };
            let Some(candidate) = self.best_single.get(&episode_id) else {
                continue;
            };
            winners.push(TvWinner {
                grabbable,
                scope: GrabScope::Episode { episode_id },
                candidate: candidate.clone(),
            });
        }
        Ok(winners)
    }
}

impl Action for RssAction {
    fn name(&self) -> &'static str {
        "rss"
    }

    fn run<'a>(&'a self, database: &'a Db) -> ActionFuture<'a> {
        Box::pin(async move { self.tick(database).await.map_err(Into::into) })
    }
}

/// A matched title with everything a grab needs loaded.
#[derive(Clone, Debug)]
struct Grabbable {
    id: i64,
    title: String,
    policy: MoviePolicy,
    language: Language,
}

/// Better score first, then the tiebreaks targeted search sorts by, so both
/// triggers pick the same release out of the same candidates.
fn beats(candidate: &Eligible, incumbent: &Eligible) -> bool {
    candidate
        .score
        .cmp(&incumbent.score)
        .then_with(|| incumbent.indexer_id.cmp(&candidate.indexer_id))
        .then_with(|| incumbent.guid.cmp(&candidate.guid))
        == std::cmp::Ordering::Greater
}

/// Every wanted title with neither a file nor a grab in flight.
///
/// Unlike targeted search's work list this is not limited or ordered by when
/// a title was last searched: matching is local, so the cost of carrying the
/// whole list is a string comparison per title per feed item (§6.2). Blocked
/// titles are here on purpose (§6.3).
async fn wanted_movies(database: &Db) -> Result<Vec<WantedMovie>, GrabError> {
    let rows = sqlx::query!(
        r#"
        SELECT   m.id      AS "id!: i64",
                 m.tmdb_id AS "tmdb_id!: i64",
                 m.imdb_id AS "imdb_id: String",
                 m.title   AS "title!: String",
                 m.year
        FROM     movies m
        WHERE    m.wanted = 1
          AND    NOT EXISTS (
                     SELECT 1 FROM media_files f
                     WHERE  f.owner_kind = 'movie' AND f.owner_id = m.id
                 )
          AND    NOT EXISTS (
                     SELECT 1 FROM grabs g
                     WHERE  g.target_kind = 'movie' AND g.target_id = m.id
                       AND  g.state IN ('sent', 'downloaded', 'imported')
                 )
        ORDER BY m.id
        "#
    )
    .fetch_all(database.pool())
    .await?;

    Ok(rows
        .into_iter()
        .map(|row| WantedMovie {
            id: MovieId(row.id),
            tmdb_id: u32::try_from(row.tmdb_id).ok(),
            imdb_id: row.imdb_id,
            title: row.title,
            year: row.year.and_then(|year| u16::try_from(year).ok()),
        })
        .collect())
}

/// A matched series season with everything a grab needs loaded.
#[derive(Clone, Debug)]
struct TvGrabbable {
    series_title: String,
    policy: TitlePolicy,
    language: Language,
}

/// One winner of the episode lane: what to grab, for whom, and which
/// release won.
struct TvWinner {
    grabbable: TvGrabbable,
    scope: GrabScope,
    candidate: Eligible,
}

/// A wanted episode with the season context matching and grabbing need.
struct WantedEpisodeRow {
    wanted: WantedEpisode,
    season_id: i64,
}

fn tv_match_kind(kind: MatchKind) -> &'static str {
    match kind {
        MatchKind::TmdbId => "tmdb id",
        MatchKind::ImdbId => "imdb id",
        MatchKind::TitleAndYear => "title",
    }
}

/// Every wanted episode still open — wanted, no file on disk, no live grab
/// of its own or of its season. Same shape as the movie list: unlimited,
/// unordered by recency, blocked series included on purpose (§6.3).
async fn wanted_episodes(database: &Db) -> Result<Vec<WantedEpisodeRow>, GrabError> {
    let rows = sqlx::query!(
        r#"
        SELECT   e.id     AS "id!: i64",
                 e.number AS "episode!: i64",
                 se.id     AS "season_id!: i64",
                 se.number AS "season!: i64",
                 s.tmdb_id AS "series_tmdb_id!: i64",
                 s.title   AS "series_title!: String"
        FROM     episodes e
        JOIN     seasons se ON se.id = e.season_id
        JOIN     series s ON s.id = se.series_id
        WHERE    e.wanted = 1
          AND    NOT EXISTS (
                      SELECT 1 FROM media_files f
                      WHERE  f.owner_kind = 'episode' AND f.owner_id = e.id
                  )
          AND    NOT EXISTS (
                      SELECT 1 FROM grabs g
                      WHERE  g.target_kind = 'episode' AND g.target_id = e.id
                        AND  g.state IN ('sent', 'downloaded', 'imported')
                  )
          AND    NOT EXISTS (
                      SELECT 1 FROM grabs g
                      WHERE  g.target_kind = 'season' AND g.target_id = se.id
                        AND  g.state IN ('sent', 'downloaded', 'imported')
                  )
        ORDER BY e.id
        "#
    )
    .fetch_all(database.pool())
    .await?;

    Ok(rows
        .into_iter()
        .map(|row| WantedEpisodeRow {
            wanted: WantedEpisode {
                id: EpisodeId(row.id),
                tmdb_id: u32::try_from(row.series_tmdb_id).ok(),
                // The series table carries no `IMDb` id yet, so this lane
                // matches on TMDB id or title until one is backfilled.
                imdb_id: None,
                title: row.series_title,
                season: u32::try_from(row.season).unwrap_or_default(),
                episode: u32::try_from(row.episode).unwrap_or_default(),
            },
            season_id: row.season_id,
        })
        .collect())
}

/// The policy and original language a matched season needs, or `None` when
/// it cannot be evaluated yet. Same refusals as [`load_grabbable`].
async fn load_tv_grabbable(
    database: &Db,
    episodes: &[WantedEpisodeRow],
    season_id: i64,
) -> Result<Option<TvGrabbable>, GrabError> {
    let Some(series_title) = episodes
        .iter()
        .find(|row| row.season_id == season_id)
        .map(|row| row.wanted.title.clone())
    else {
        return Ok(None);
    };
    let Some(policy) = database.season_policy(season_id).await? else {
        return Ok(None);
    };
    // §5.2, same as films.
    let language: Option<String> = sqlx::query_scalar!(
        r#"
        SELECT   s.original_language
        FROM     seasons se
        JOIN     series s ON s.id = se.series_id
        WHERE    se.id = ?
        "#,
        season_id
    )
    .fetch_optional(database.pool())
    .await?
    .flatten();
    let Some(language) = language else {
        tracing::warn!(
            season_id,
            series = series_title,
            "no original language yet; not grabbing from RSS"
        );
        return Ok(None);
    };

    Ok(Some(TvGrabbable {
        series_title,
        policy,
        language: arr_db::policy::language(&language),
    }))
}

/// Whether §6.2's guard lets a season pack be grabbed here: exactly what
/// [`season_grab_mode`] requires of targeted search, computed over every
/// episode the season is known to hold (§14, amended in #117).
async fn pack_allowed(database: &Db, season_id: i64) -> Result<bool, GrabError> {
    let episodes = sqlx::query!(
        r#"
        SELECT   e.air_date,
                 EXISTS (
                     SELECT 1 FROM media_files f
                     WHERE  f.owner_kind = 'episode' AND f.owner_id = e.id
                 ) AS "on_disk!: bool"
        FROM     episodes e
        WHERE    e.season_id = ?
        ORDER BY e.number
        "#,
        season_id
    )
    .fetch_all(database.pool())
    .await?;
    let failed_packs = sqlx::query!(
        r#"SELECT count(*)        AS "failures!: i64",
                  max(grabbed_at) AS "last_failed_at?: String"
           FROM grabs
           WHERE target_kind = 'season' AND target_id = ? AND state = 'failed'"#,
        season_id
    )
    .fetch_one(database.pool())
    .await?;

    Ok(season_grab_mode(&SeasonGrabFacts {
        air_dates: &episodes
            .iter()
            .map(|episode| air_date_time(episode.air_date.as_deref()))
            .collect::<Vec<_>>(),
        now: std::time::SystemTime::now(),
        any_episode_on_disk: episodes.iter().any(|episode| episode.on_disk),
        pack_backoff_active: failed_packs.failures > 0
            && !backoff_elapsed(
                failed_packs.failures,
                failed_packs.last_failed_at.as_deref(),
            ),
    }) == SeasonGrabMode::SeasonPack)
}

/// An `air_date` as TMDB writes it (`YYYY-MM-DD`), or a full timestamp if
/// one ever arrives that way. Unknown is unaired.
fn air_date_time(value: Option<&str>) -> Option<std::time::SystemTime> {
    let value = value?;
    let timestamp = if let Ok(date) = value.parse::<chrono::NaiveDate>() {
        date.and_time(chrono::NaiveTime::MIN).and_utc().timestamp()
    } else {
        value
            .parse::<chrono::DateTime<chrono::Utc>>()
            .ok()?
            .timestamp()
    };
    let seconds = u64::try_from(timestamp.abs()).ok()?;
    if timestamp < 0 {
        std::time::UNIX_EPOCH.checked_sub(std::time::Duration::from_secs(seconds))
    } else {
        std::time::UNIX_EPOCH.checked_add(std::time::Duration::from_secs(seconds))
    }
}

/// The policy and original language a matched title needs, or `None` when
/// the title cannot be evaluated yet.
async fn load_grabbable(
    database: &Db,
    wanted: &[WantedMovie],
    movie_id: i64,
) -> Result<Option<Grabbable>, GrabError> {
    let Some(title) = wanted
        .iter()
        .find(|movie| movie.id == MovieId(movie_id))
        .map(|movie| movie.title.clone())
    else {
        return Ok(None);
    };
    let Some(policy) = database.movie_policy(movie_id).await? else {
        return Ok(None);
    };
    // §5.2: every language rule is expressed against the title's own
    // original language, and guessing it is worse than not grabbing.
    let language: Option<String> = sqlx::query_scalar!(
        "SELECT original_language FROM movies WHERE id = ?",
        movie_id
    )
    .fetch_optional(database.pool())
    .await?
    .flatten();
    let Some(language) = language else {
        tracing::warn!(movie_id, "no original language yet; not grabbing from RSS");
        return Ok(None);
    };

    Ok(Some(Grabbable {
        id: movie_id,
        title,
        policy,
        language: arr_db::policy::language(&language),
    }))
}

#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
    use std::sync::{Arc, Mutex};

    use serde_json::{json, Value};
    use wiremock::matchers::{method, path, query_param, query_param_is_missing};
    use wiremock::{Mock, MockServer, Request, Respond, ResponseTemplate};

    use super::*;
    use crate::grab::{test_downloads, SeedingLimits};
    use arr_dl::TransmissionClient;

    /// The recorded feed: two releases of one wanted title, one that only a
    /// TMDB id identifies, one near miss, one TV item and one film nobody
    /// asked for.
    const FEED: &str = include_str!("../tests/fixtures/rss.xml");

    /// A season pack and its three episodes, all eligible under the seeded
    /// TV main policy (§5.5 bands: 2160p floor 3 GiB, per #95).
    const TV_FEED: &str = r#"<rss version="2.0" xmlns:torznab="http://torznab.com/schemas/2015/feed">
      <channel>
        <item>
          <title>Fallout.S01.2160p.WEB-DL.DDP5.1.Atmos.H.265-GROUP</title>
          <guid>tv-pack</guid><link>https://indexer.invalid/download/tv-pack</link>
          <enclosure url="https://indexer.invalid/download/tv-pack" length="85899345920"
                     type="application/x-bittorrent"/>
          <torznab:attr name="seeders" value="120"/>
        </item>
        <item>
          <title>Fallout.S01E01.2160p.WEB-DL.DDP5.1.H.265-GROUP</title>
          <guid>tv-e01</guid><link>https://indexer.invalid/download/tv-e01</link>
          <enclosure url="https://indexer.invalid/download/tv-e01" length="10737418240"
                     type="application/x-bittorrent"/>
          <torznab:attr name="seeders" value="80"/>
        </item>
        <item>
          <title>Fallout.S01E02.2160p.WEB-DL.DDP5.1.H.265-GROUP</title>
          <guid>tv-e02</guid><link>https://indexer.invalid/download/tv-e02</link>
          <enclosure url="https://indexer.invalid/download/tv-e02" length="10737418240"
                     type="application/x-bittorrent"/>
          <torznab:attr name="seeders" value="80"/>
        </item>
        <item>
          <title>Fallout.S01E03.2160p.WEB-DL.DDP5.1.H.265-GROUP</title>
          <guid>tv-e03</guid><link>https://indexer.invalid/download/tv-e03</link>
          <enclosure url="https://indexer.invalid/download/tv-e03" length="10737418240"
                     type="application/x-bittorrent"/>
          <torznab:attr name="seeders" value="80"/>
        </item>
        <item>
          <title>The.Fallout.2021.S01E03.1080p.WEB-DL-GROUP</title>
          <guid>tv-near-miss</guid><link>https://indexer.invalid/download/tv-near-miss</link>
          <enclosure url="https://indexer.invalid/download/tv-near-miss" length="10737418240"
                     type="application/x-bittorrent"/>
          <torznab:attr name="seeders" value="80"/>
        </item>
      </channel>
    </rss>"#;

    const INDEXERS: [i64; 2] = [7, 9];

    /// Enough of Transmission to add a torrent and list nothing back.
    #[derive(Clone, Debug, Default)]
    struct FakeTransmission {
        added: Arc<Mutex<Vec<String>>>,
    }

    impl Respond for FakeTransmission {
        fn respond(&self, request: &Request) -> ResponseTemplate {
            let body: Value = serde_json::from_slice(&request.body).unwrap();
            match body["method"].as_str().unwrap_or_default() {
                "torrent-add" => {
                    let source = body["arguments"]["filename"]
                        .as_str()
                        .unwrap_or_default()
                        .to_owned();
                    let mut added = self.added.lock().unwrap();
                    added.push(source.clone());
                    let id = i64::try_from(added.len()).unwrap();
                    success(&json!({"torrent-added": {
                        "id": id, "name": source, "hashString": format!("{id:040x}")
                    }}))
                }
                "torrent-get" => success(&json!({"torrents": []})),
                _ => success(&json!({})),
            }
        }
    }

    fn success(arguments: &Value) -> ResponseTemplate {
        ResponseTemplate::new(200).set_body_json(json!({
            "result": "success", "arguments": arguments
        }))
    }

    /// Two indexers, both advertising a text search, both serving the same
    /// feed to an empty query.
    async fn prowlarr(feed: &str) -> MockServer {
        let server = MockServer::start().await;
        let feed = test_downloads::rewrite(feed, "https://indexer.invalid/download/", &server);
        test_downloads::mount(&server).await;
        Mock::given(method("GET"))
            .and(path("/api/v1/indexer"))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!([
                {"id": 7, "name": "alpha", "enable": true},
                {"id": 9, "name": "beta", "enable": true},
            ])))
            .mount(&server)
            .await;
        for id in INDEXERS {
            Mock::given(method("GET"))
                .and(path(format!("/{id}/api")))
                .and(query_param("t", "caps"))
                .respond_with(ResponseTemplate::new(200).set_body_string(
                    r#"<caps><searching><search available="yes" supportedParams="q"/></searching></caps>"#,
                ))
                .mount(&server)
                .await;
            // The RSS call is `t=search` with no `q` at all (§6.2).
            Mock::given(method("GET"))
                .and(path(format!("/{id}/api")))
                .and(query_param("t", "search"))
                .and(query_param_is_missing("q"))
                .respond_with(ResponseTemplate::new(200).set_body_string(&feed))
                .mount(&server)
                .await;
        }
        server
    }

    async fn transmission() -> (MockServer, FakeTransmission) {
        let server = MockServer::start().await;
        let fake = FakeTransmission::default();
        Mock::given(method("POST"))
            .respond_with(fake.clone())
            .mount(&server)
            .await;
        (server, fake)
    }

    /// `(tmdb_id, title, year, blocked)`.
    async fn wanted(movies: &[(i64, &str, i64, bool)]) -> (tempfile::TempDir, Db) {
        let dir = tempfile::tempdir().unwrap();
        let database = Db::connect(dir.path().join("arr.db")).await.unwrap();
        database.migrate().await.unwrap();
        for &(tmdb_id, title, year, blocked) in movies {
            sqlx::query(
                "INSERT INTO movies (tmdb_id, title, year, original_language, blocked, root_id)
                 SELECT ?, ?, ?, 'en', ?, id
                 FROM roots WHERE kind = 'movie' AND audience = 'main'",
            )
            .bind(tmdb_id)
            .bind(title)
            .bind(year)
            .bind(i64::from(blocked))
            .execute(database.pool())
            .await
            .unwrap();
        }
        (dir, database)
    }

    fn action(prowlarr: &MockServer, transmission: &MockServer) -> RssAction {
        RssAction::new(
            ProwlarrClient::new(prowlarr.uri(), "key").unwrap(),
            TransmissionClient::new(&transmission.uri()).unwrap(),
            PathBuf::from("/mnt/media/transmission/complete"),
            SeedingRules::new(
                SeedingLimits {
                    ratio: 1.0,
                    idle_minutes: 60,
                },
                HashMap::new(),
            ),
        )
    }

    /// Fallout S01 with wanted episodes airing at the given dates. Returns
    /// the season id and the episode ids in episode order.
    async fn wanted_series(database: &Db, air_dates: &[&str]) -> (i64, Vec<i64>) {
        sqlx::query(
            "INSERT INTO series (tmdb_id, title, year, original_language, root_id)
             SELECT 106379, 'Fallout', 2024, 'en', id
             FROM roots WHERE kind = 'tv' AND audience = 'main'",
        )
        .execute(database.pool())
        .await
        .unwrap();
        let season_id: i64 = sqlx::query_scalar(
            "INSERT INTO seasons (series_id, number) VALUES (1, 1) RETURNING id",
        )
        .fetch_one(database.pool())
        .await
        .unwrap();
        let mut episodes = Vec::new();
        for (index, air_date) in air_dates.iter().enumerate() {
            let id: i64 = sqlx::query_scalar(
                "INSERT INTO episodes (season_id, number, title, air_date, wanted)
                 VALUES (?, ?, ?, ?, 1) RETURNING id",
            )
            .bind(season_id)
            .bind(i64::try_from(index).unwrap() + 1)
            .bind(format!("Episode {}", index + 1))
            .bind(air_date)
            .fetch_one(database.pool())
            .await
            .unwrap();
            episodes.push(id);
        }
        (season_id, episodes)
    }

    /// `(target_kind, target_id, state)` for every TV grab that was made.
    async fn tv_grabs(database: &Db) -> Vec<(String, i64, String)> {
        sqlx::query_as::<_, (String, i64, String)>(
            "SELECT target_kind, target_id, state FROM grabs
             WHERE target_kind IN ('episode', 'season') ORDER BY target_kind, target_id",
        )
        .fetch_all(database.pool())
        .await
        .unwrap()
    }

    async fn episode_states(database: &Db) -> Vec<String> {
        sqlx::query_scalar::<_, String>("SELECT state FROM episodes ORDER BY number")
            .fetch_all(database.pool())
            .await
            .unwrap()
    }

    /// `(movie title, grabbed release name)` for every grab that was made.
    async fn grabbed(database: &Db) -> Vec<(String, String)> {
        sqlx::query_as::<_, (String, String)>(
            "SELECT m.title, r.name
             FROM   grabs g
             JOIN   movies m ON m.id = g.target_id
             JOIN   releases r ON r.id = g.release_id
             ORDER BY m.title",
        )
        .fetch_all(database.pool())
        .await
        .unwrap()
    }

    async fn stored_releases(database: &Db) -> Vec<String> {
        sqlx::query_scalar::<_, String>("SELECT DISTINCT guid FROM releases ORDER BY guid")
            .fetch_all(database.pool())
            .await
            .unwrap()
    }

    /// The acceptance case: a recorded feed against a seeded wanted list
    /// grabs exactly the intended titles, by TMDB id where the tracker
    /// supplied one.
    #[tokio::test]
    async fn exactly_the_intended_titles_match() {
        let (_dir, database) = wanted(&[
            (693_134, "Dune: Part Two", 2024, false),
            (933_260, "The Substance", 2024, false),
            (438_631, "Dune", 2021, false),
        ])
        .await;
        let indexer = prowlarr(FEED).await;
        let (downloader, fake) = transmission().await;

        let outcomes = action(&indexer, &downloader).tick(&database).await.unwrap();

        assert_eq!(outcomes.len(), 3);
        assert_eq!(
            grabbed(&database).await,
            vec![
                (
                    "Dune".to_owned(),
                    "Dune.2021.2160p.WEB-DL.DDP5.1.Atmos.H.265-GROUP".to_owned()
                ),
                (
                    "Dune: Part Two".to_owned(),
                    "Dune.Part.Two.2024.2160p.WEB-DL.DDP5.1.Atmos.H.265-GROUP".to_owned()
                ),
                (
                    "The Substance".to_owned(),
                    "La.Sustancia.2024.2160p.WEB-DL.DDP5.1.H.265-GRUPO".to_owned()
                ),
            ],
            "one torrent per title, and the 1080p Dune: Part Two loses to the 2160p one"
        );
        assert_eq!(fake.added.lock().unwrap().len(), 3);
        // The TV item and the film nobody asked for are not cached against
        // any title: an unmatched release belongs to nothing. Both indexers
        // carry the same four, under their own indexer id.
        assert_eq!(
            stored_releases(&database).await,
            vec!["alpha-1", "alpha-2", "alpha-3", "alpha-4"]
        );
        let rows: i64 = sqlx::query_scalar("SELECT count(*) FROM releases")
            .fetch_one(database.pool())
            .await
            .unwrap();
        assert_eq!(rows, 8);
    }

    /// The near miss. "Dune: Part Three" is one word from a release that is
    /// in the feed, from the same franchise, in the same year.
    #[tokio::test]
    async fn a_near_miss_is_not_grabbed() {
        let (_dir, database) = wanted(&[(9_999_999, "Dune: Part Three", 2024, false)]).await;
        let indexer = prowlarr(FEED).await;
        let (downloader, fake) = transmission().await;

        let outcomes = action(&indexer, &downloader).tick(&database).await.unwrap();

        assert!(outcomes.is_empty());
        assert!(fake.added.lock().unwrap().is_empty());
        assert!(stored_releases(&database).await.is_empty());
    }

    /// §6.2: RSS never backs off, so nothing it does may spend one of
    /// targeted search's attempts.
    #[tokio::test]
    async fn matching_never_touches_the_search_backoff() {
        let (_dir, database) = wanted(&[
            (693_134, "Dune: Part Two", 2024, false),
            (9_999_999, "Dune: Part Three", 2024, false),
        ])
        .await;
        let indexer = prowlarr(FEED).await;
        let (downloader, _fake) = transmission().await;

        action(&indexer, &downloader).tick(&database).await.unwrap();

        let attempts: Vec<(i64, Option<String>)> =
            sqlx::query_as("SELECT search_attempts, last_searched_at FROM movies ORDER BY id")
                .fetch_all(database.pool())
                .await
                .unwrap();
        assert_eq!(attempts, vec![(0, None), (0, None)]);
    }

    /// §6.3: `blocked` stops targeted search and leaves RSS matching on.
    #[tokio::test]
    async fn a_blocked_title_still_matches_rss() {
        let (_dir, database) = wanted(&[(693_134, "Dune: Part Two", 2024, true)]).await;
        let indexer = prowlarr(FEED).await;
        let (downloader, _fake) = transmission().await;

        let outcomes = action(&indexer, &downloader).tick(&database).await.unwrap();

        assert_eq!(outcomes.len(), 1);
    }

    /// The property the whole lane exists for: cost is one call per indexer,
    /// whatever the wanted list holds.
    #[tokio::test]
    async fn one_empty_query_call_per_indexer() {
        let (_dir, database) = wanted(&[
            (693_134, "Dune: Part Two", 2024, false),
            (933_260, "The Substance", 2024, false),
            (438_631, "Dune", 2021, false),
        ])
        .await;
        let indexer = prowlarr(FEED).await;
        let (downloader, _fake) = transmission().await;

        action(&indexer, &downloader).tick(&database).await.unwrap();

        for id in INDEXERS {
            let feeds = indexer
                .received_requests()
                .await
                .unwrap()
                .into_iter()
                .filter(|request| {
                    request.url.path() == format!("/{id}/api")
                        && request
                            .url
                            .query_pairs()
                            .any(|(name, value)| name == "t" && value == "search")
                })
                .count();
            assert_eq!(feeds, 1);
        }
    }

    /// A single-episode release in the feed closes a wanted, missing
    /// episode's gap, with no targeted-search attempt spent (§6.2). The
    /// season here is airing, so its pack is not eligible ([`season_grab_mode`]).
    #[tokio::test]
    async fn a_missing_episode_is_grabbed_from_rss() {
        let (_dir, database) = wanted(&[]).await;
        let (_season_id, episodes) = wanted_series(&database, &["2024-04-11", "2099-01-01"]).await;
        let indexer = prowlarr(TV_FEED).await;
        let (downloader, fake) = transmission().await;

        let outcomes = action(&indexer, &downloader).tick(&database).await.unwrap();

        assert_eq!(outcomes.len(), 2);
        assert_eq!(
            tv_grabs(&database).await,
            episodes
                .iter()
                .map(|id| ("episode".to_owned(), *id, "sent".to_owned()))
                .collect::<Vec<_>>(),
            "each gap takes its own single; the pack and the near miss stay out"
        );
        assert_eq!(fake.added.lock().unwrap().len(), 2);
        let attempts: Vec<(i64, Option<String>)> =
            sqlx::query_as("SELECT search_attempts, last_searched_at FROM episodes ORDER BY id")
                .fetch_all(database.pool())
                .await
                .unwrap();
        assert!(attempts
            .iter()
            .all(|(count, at)| *count == 0 && at.is_none()));
    }

    /// The acceptance case: one feed pass over a fully aired season with
    /// nothing on disk takes the pack alone, and the singles stand down.
    #[tokio::test]
    async fn a_completed_season_takes_one_pack_from_rss() {
        let (_dir, database) = wanted(&[]).await;
        let (season_id, _episodes) =
            wanted_series(&database, &["2024-04-11", "2024-04-18", "2024-04-25"]).await;
        // Only the pack is wanted here: strip the singles so the pass has to
        // pick between shapes on coverage, not on availability.
        let pack_only = TV_FEED.replace(
            "Fallout.S01E01.2160p.WEB-DL.DDP5.1.H.265-GROUP</title>",
            "Unrelated.S01E01.2160p.WEB-DL-GROUP</title>",
        );
        let pack_only = pack_only.replace(
            "Fallout.S01E02.2160p.WEB-DL.DDP5.1.H.265-GROUP</title>",
            "Unrelated.S01E02.2160p.WEB-DL-GROUP</title>",
        );
        let pack_only = pack_only.replace(
            "Fallout.S01E03.2160p.WEB-DL.DDP5.1.H.265-GROUP</title>",
            "Unrelated.S01E03.2160p.WEB-DL-GROUP</title>",
        );
        let indexer = prowlarr(&pack_only).await;
        let (downloader, fake) = transmission().await;

        let outcomes = action(&indexer, &downloader).tick(&database).await.unwrap();

        assert_eq!(outcomes.len(), 1);
        assert_eq!(fake.added.lock().unwrap().len(), 1);
        assert_eq!(
            tv_grabs(&database).await,
            vec![("season".to_owned(), season_id, "sent".to_owned())]
        );
        assert_eq!(
            episode_states(&database).await,
            vec!["downloading"; 3],
            "the pack flips every episode it covers"
        );
    }

    /// §6.2 with #117's guard: an episode already on disk keeps the season
    /// per-episode here too — the pack is skipped and the open gaps take
    /// their singles.
    #[tokio::test]
    async fn a_pack_for_a_season_with_a_file_on_disk_is_skipped() {
        let (_dir, database) = wanted(&[]).await;
        let (_season_id, episodes) =
            wanted_series(&database, &["2024-04-11", "2024-04-18", "2024-04-25"]).await;
        sqlx::query(
            "INSERT INTO media_files (owner_kind, owner_id, path, size)
             VALUES ('episode', ?, '/tv/fallout/s01e01.mkv', 1)",
        )
        .bind(episodes[0])
        .execute(database.pool())
        .await
        .unwrap();
        let indexer = prowlarr(TV_FEED).await;
        let (downloader, _fake) = transmission().await;

        action(&indexer, &downloader).tick(&database).await.unwrap();

        let grabs = tv_grabs(&database).await;
        assert!(
            !grabs.iter().any(|(kind, _, _)| kind == "season"),
            "no pack over an episode already on disk"
        );
        // Episode 1 is not wanted any more — it has a file — and the near
        // miss never matched, so exactly the two open gaps are filled.
        assert_eq!(
            grabs,
            vec![
                ("episode".to_owned(), episodes[1], "sent".to_owned()),
                ("episode".to_owned(), episodes[2], "sent".to_owned()),
            ]
        );
    }

    /// §6.3: `blocked` stops targeted search for a series and leaves RSS
    /// matching on. The one-episode season here is fully aired, so what
    /// matches is its pack — the same release targeted search would take.
    #[tokio::test]
    async fn a_blocked_series_still_matches_rss() {
        let (_dir, database) = wanted(&[]).await;
        let (season_id, _episodes) = wanted_series(&database, &["2024-04-11"]).await;
        sqlx::query("UPDATE series SET blocked = 1")
            .execute(database.pool())
            .await
            .unwrap();
        let indexer = prowlarr(TV_FEED).await;
        let (downloader, _fake) = transmission().await;

        let outcomes = action(&indexer, &downloader).tick(&database).await.unwrap();

        assert_eq!(outcomes.len(), 1);
        assert_eq!(
            tv_grabs(&database).await,
            vec![("season".to_owned(), season_id, "sent".to_owned())]
        );
    }
}