//! §9.5 *needs a decision* → the operator alone: a movie or series entered
//! the no-PT-source queue, or hard-failed twice on different releases (the
//! same queues `GET /api/queues/attention` reports, §9.3).
//!
//! §5.7 sets the bar for the hard-fail side: two failures on *different*
//! releases, both inside `arr_db::ATTENTION_WINDOW`, against a target still
//! waiting for a file. One bad torrent is not a decision, a failure already
//! dealt with ages out (#226), and a target that has since been acquired
//! leaves at once (#238). The season lane reads that last condition off its
//! episodes, which is where intent lives (§4.1). `GET /api/queues/attention`
//! filters identically, or the two channels tell the operator different
//! stories.
//!
//! Edge-triggered per title: it notifies once when the title enters either
//! queue, and is forgotten once it leaves both, so a future re-entry notifies
//! again. A series notifies as its series, never per episode — a broken
//! twenty-episode season is one message, which is the restraint §9.5 exists
//! for. The notified set is in-memory and transient — a restart re-notifies
//! whatever is currently queued, same tradeoff `ImportAction`'s probe cache
//! makes.

use std::collections::{HashMap, HashSet};
use std::sync::Arc;

use arr_db::Db;
use tokio::sync::Mutex;

use crate::notify::Notifier;
use crate::reconcile::{Action, ActionFuture, Outcome};

/// What is queued. Movies and series have independent id sequences, so the
/// kind travels with the id.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
enum Subject {
    Movie(i64),
    Series(i64),
}

/// One series' standing in the TV queues, rolled up from its episodes.
#[derive(Debug, Default)]
struct TvEntry {
    /// Wanted episodes whose every candidate was rejected for language.
    no_pt_source: Vec<i64>,
    /// Episodes two different releases hard-failed post-probe (§5.7).
    hard_failed_episodes: Vec<i64>,
    /// Seasons two different pack releases hard-failed on (§5.7), sending the
    /// season back to per-episode grabbing.
    failed_season_packs: Vec<i64>,
}

impl TvEntry {
    fn reason(&self) -> String {
        let mut parts = Vec::new();
        if !self.no_pt_source.is_empty() {
            parts.push(plural(
                self.no_pt_source.len(),
                "episode found no Portuguese-audio source",
                "episodes found no Portuguese-audio source",
            ));
        }
        if !self.hard_failed_episodes.is_empty() {
            parts.push(plural(
                self.hard_failed_episodes.len(),
                "episode hard-failed twice on different releases",
                "episodes hard-failed twice on different releases",
            ));
        }
        if !self.failed_season_packs.is_empty() {
            parts.push(plural(
                self.failed_season_packs.len(),
                "season hard-failed twice on different packs",
                "seasons hard-failed twice on different packs",
            ));
        }
        parts.join("; ")
    }
}

fn plural(count: usize, one: &str, many: &str) -> String {
    if count == 1 {
        format!("1 {one}")
    } else {
        format!("{count} {many}")
    }
}

#[derive(Debug)]
pub struct AttentionAction {
    notifier: Notifier,
    operator_topic: String,
    notified: Arc<Mutex<HashSet<Subject>>>,
}

impl AttentionAction {
    #[must_use]
    pub fn new(notifier: Notifier, operator_topic: String) -> Self {
        Self {
            notifier,
            operator_topic,
            notified: Arc::new(Mutex::new(HashSet::new())),
        }
    }

    async fn tick(&self, database: &Db) -> Result<Vec<Outcome>, sqlx::Error> {
        let mut queued: HashMap<Subject, (String, String)> = HashMap::new();

        let no_pt_source = sqlx::query!(
            r#"
            SELECT id AS "id!: i64", title AS "title!: String", year
            FROM   movies
            WHERE  id IN (
                SELECT m.id
                FROM   movies m
                JOIN   roots root ON root.id = m.root_id
                WHERE  root.audience = 'kids'
                  AND  m.wanted = 1 AND m.blocked = 0 AND m.state = 'missing'
                  AND  m.search_attempts > 0
                  AND  NOT EXISTS (
                          SELECT 1 FROM movie_releases mr
                          JOIN   releases r ON r.id = mr.release_id
                          WHERE  mr.movie_id = m.id AND r.verdict IN ('eligible', 'waived')
                      )
            )
            "#
        )
        .fetch_all(database.pool())
        .await?;
        for row in no_pt_source {
            queued.insert(
                Subject::Movie(row.id),
                (
                    title_with_year(&row.title, row.year),
                    "no Portuguese-audio source found".to_string(),
                ),
            );
        }

        let needs_decision = sqlx::query!(
            r#"
            SELECT id AS "id!: i64", title AS "title!: String", year
            FROM   movies
            WHERE  movies.wanted = 1 AND movies.state != 'available'
              AND  (SELECT count(DISTINCT g.release_id)
                    FROM   grabs g
                    WHERE  g.target_kind = 'movie' AND g.target_id = movies.id
                      AND  g.state = 'failed'
                      AND  g.failed_at >= strftime('%Y-%m-%dT%H:%M:%fZ', 'now', ?)) >= 2
            "#,
            arr_db::ATTENTION_WINDOW
        )
        .fetch_all(database.pool())
        .await?;
        for row in needs_decision {
            queued.entry(Subject::Movie(row.id)).or_insert_with(|| {
                (
                    title_with_year(&row.title, row.year),
                    "hard-failed twice on different releases".to_string(),
                )
            });
        }

        for (series_id, title, year, entry) in queue_tv(database).await? {
            queued.insert(
                Subject::Series(series_id),
                (title_with_year(&title, year), entry.reason()),
            );
        }

        let mut notified = self.notified.lock().await;
        let mut outcomes = Vec::new();
        for (subject, (title, reason)) in &queued {
            if !notified.insert(subject.clone()) {
                continue;
            }
            match self
                .notifier
                .send(&self.operator_topic, title, reason)
                .await
            {
                Ok(()) => {
                    let (kind, id) = match subject {
                        Subject::Movie(id) => ("movie", *id),
                        Subject::Series(id) => ("series", *id),
                    };
                    outcomes.push(Outcome::new(
                        format!("{kind} {id} needs a decision"),
                        format!("notified operator: {reason}"),
                    ));
                }
                Err(error) => {
                    tracing::warn!(%error, ?subject, "needs-decision notification failed");
                }
            }
        }
        notified.retain(|subject| queued.contains_key(subject));

        Ok(outcomes)
    }
}

fn title_with_year(title: &str, year: Option<i64>) -> String {
    match year {
        Some(year) => format!("{title} ({year})"),
        None => title.to_string(),
    }
}

/// The series' roll-up entry, created empty on first sight.
fn tv_entry(
    tv: &mut HashMap<i64, (String, Option<i64>, TvEntry)>,
    series_id: i64,
    title: String,
    year: Option<i64>,
) -> &mut TvEntry {
    &mut tv
        .entry(series_id)
        .or_insert_with(|| (title, year, TvEntry::default()))
        .2
}

/// TV roll-up (§9.5): every queued series with what put it there — wanted
/// episodes whose every candidate was rejected for language, episodes two
/// different releases hard-failed post-probe, and seasons two different packs
/// hard-failed on. One entry per series, so the notification can be one per
/// series however long the broken season is.
///
/// Both hard-fail lanes carry §5.7's liveness condition: an episode is queued
/// only while `wanted` and not `available`, and a season only while at least
/// one of its episodes is. A season pack that failed twice and then fell back
/// to per-episode grabbing (§6.2) drops out as those episodes land.
async fn queue_tv(database: &Db) -> Result<Vec<(i64, String, Option<i64>, TvEntry)>, sqlx::Error> {
    let mut tv = HashMap::new();

    let tv_no_pt_source = sqlx::query!(
        r#"
        SELECT s.id AS "series_id!: i64", s.title AS "title!: String", s.year,
               e.id AS "episode_id!: i64"
        FROM   episodes e
        JOIN   seasons se ON se.id = e.season_id
        JOIN   series s ON s.id = se.series_id
        JOIN   roots root ON root.id = s.root_id
        WHERE  root.audience = 'kids'
          AND  s.blocked = 0
          AND  e.wanted = 1 AND e.state = 'missing' AND e.search_attempts > 0
          AND  NOT EXISTS (
                  SELECT 1 FROM episode_releases er
                  JOIN   releases r ON r.id = er.release_id
                  WHERE  er.episode_id = e.id AND r.verdict IN ('eligible', 'waived')
              )
        "#
    )
    .fetch_all(database.pool())
    .await?;
    for row in tv_no_pt_source {
        tv_entry(&mut tv, row.series_id, row.title, row.year)
            .no_pt_source
            .push(row.episode_id);
    }

    let episode_hard_fails = sqlx::query!(
        r#"
        SELECT s.id AS "series_id!: i64", s.title AS "title!: String", s.year,
               g.target_id AS "episode_id!: i64"
        FROM   grabs g
        JOIN   episodes e ON g.target_kind = 'episode' AND e.id = g.target_id
        JOIN   seasons se ON se.id = e.season_id
        JOIN   series s ON s.id = se.series_id
        WHERE  g.state = 'failed'
          AND  g.failed_at >= strftime('%Y-%m-%dT%H:%M:%fZ', 'now', ?)
          AND  e.wanted = 1 AND e.state != 'available'
        GROUP BY s.id, s.title, s.year, e.id
        HAVING count(DISTINCT g.release_id) >= 2
        "#,
        arr_db::ATTENTION_WINDOW
    )
    .fetch_all(database.pool())
    .await?;
    for row in episode_hard_fails {
        tv_entry(&mut tv, row.series_id, row.title, row.year)
            .hard_failed_episodes
            .push(row.episode_id);
    }

    let season_pack_fails = sqlx::query!(
        r#"
        SELECT s.id AS "series_id!: i64", s.title AS "title!: String", s.year,
               g.target_id AS "season_id!: i64"
        FROM   grabs g
        JOIN   seasons se ON g.target_kind = 'season' AND se.id = g.target_id
        JOIN   series s ON s.id = se.series_id
        WHERE  g.state = 'failed'
          AND  g.failed_at >= strftime('%Y-%m-%dT%H:%M:%fZ', 'now', ?)
          AND  EXISTS (
                  SELECT 1 FROM episodes e
                  WHERE  e.season_id = se.id
                    AND  e.wanted = 1 AND e.state != 'available'
               )
        GROUP BY s.id, s.title, s.year, se.id
        HAVING count(DISTINCT g.release_id) >= 2
        "#,
        arr_db::ATTENTION_WINDOW
    )
    .fetch_all(database.pool())
    .await?;
    for row in season_pack_fails {
        tv_entry(&mut tv, row.series_id, row.title, row.year)
            .failed_season_packs
            .push(row.season_id);
    }

    Ok(tv
        .into_iter()
        .map(|(id, (t, y, e))| (id, t, y, e))
        .collect())
}

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

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

#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
    use wiremock::matchers::method;
    use wiremock::{Mock, MockServer, ResponseTemplate};

    use super::*;

    async fn seeded_database() -> (tempfile::TempDir, Db) {
        let dir = tempfile::tempdir().unwrap();
        let database = Db::connect(dir.path().join("arr.db")).await.unwrap();
        database.migrate().await.unwrap();
        (dir, database)
    }

    async fn insert_no_pt_source_movie(database: &Db) -> i64 {
        let root_id: i64 =
            sqlx::query_scalar("SELECT id FROM roots WHERE kind = 'movie' AND audience = 'kids'")
                .fetch_one(database.pool())
                .await
                .unwrap();
        sqlx::query(
            "INSERT INTO movies
                (tmdb_id, title, year, root_id, wanted, state, search_attempts)
             VALUES (1, 'Encanto', 2021, ?, 1, 'missing', 1)
             RETURNING id",
        )
        .bind(root_id)
        .fetch_one(database.pool())
        .await
        .unwrap();
        sqlx::query_scalar("SELECT id FROM movies WHERE tmdb_id = 1")
            .fetch_one(database.pool())
            .await
            .unwrap()
    }

    /// A series on the kids TV root with one season, and `count` wanted,
    /// missing episodes that have been searched. No stored releases: every
    /// candidate was rejected for language (§5.2's no-PT-source case).
    async fn insert_no_pt_source_series(database: &Db, tmdb_id: i64, episode_count: usize) -> i64 {
        let root_id: i64 =
            sqlx::query_scalar("SELECT id FROM roots WHERE kind = 'tv' AND audience = 'kids'")
                .fetch_one(database.pool())
                .await
                .unwrap();
        sqlx::query(
            "INSERT INTO series (tmdb_id, title, year, root_id) VALUES (?, 'Bluey', 2018, ?)",
        )
        .bind(tmdb_id)
        .bind(root_id)
        .execute(database.pool())
        .await
        .unwrap();
        let series_id: i64 = sqlx::query_scalar("SELECT id FROM series WHERE tmdb_id = ?")
            .bind(tmdb_id)
            .fetch_one(database.pool())
            .await
            .unwrap();
        sqlx::query("INSERT INTO seasons (series_id, number) VALUES (?, 1)")
            .bind(series_id)
            .execute(database.pool())
            .await
            .unwrap();
        let season_id: i64 = sqlx::query_scalar("SELECT id FROM seasons WHERE series_id = ?")
            .bind(series_id)
            .fetch_one(database.pool())
            .await
            .unwrap();
        for number in 0..episode_count {
            sqlx::query(
                "INSERT INTO episodes (season_id, number, title, wanted, state, search_attempts)
                 VALUES (?, ?, ?, 1, 'missing', 1)",
            )
            .bind(season_id)
            .bind(i64::try_from(number).unwrap_or(0) + 1)
            .bind(format!("Episode {number}"))
            .execute(database.pool())
            .await
            .unwrap();
        }
        series_id
    }

    /// A wanted, missing episode: the least that keeps its season live for
    /// §5.7's liveness condition.
    async fn insert_wanted_episode(database: &Db, season_id: i64, number: i64) -> i64 {
        sqlx::query_scalar(
            "INSERT INTO episodes (season_id, number, title, wanted, state)
             VALUES (?, ?, ?, 1, 'missing') RETURNING id",
        )
        .bind(season_id)
        .bind(number)
        .bind(format!("Episode {number}"))
        .fetch_one(database.pool())
        .await
        .unwrap()
    }

    /// A failed grab by `release_guid` against `target_kind`/`target_id`,
    /// grabbed `grab_age_days` ago and failed `fail_age_days` ago, so §5.7's
    /// window — which runs from the failure — can be exercised without
    /// waiting a month.
    async fn insert_dated_failed_grab(
        database: &Db,
        target_kind: &str,
        target_id: i64,
        release_guid: &str,
        grab_age_days: i64,
        fail_age_days: i64,
    ) {
        let release_id: i64 = sqlx::query_scalar(
            "INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed, verdict)
             VALUES (7, ?, 'release', 10737418240, 'https://tracker/x.torrent', '{}', 'eligible')
             RETURNING id",
        )
        .bind(release_guid)
        .fetch_one(database.pool())
        .await
        .unwrap();
        sqlx::query(
            "INSERT INTO grabs (release_id, target_kind, target_id, infohash, state, grabbed_at, failed_at)
             VALUES (?, ?, ?, ?, 'failed',
                     strftime('%Y-%m-%dT%H:%M:%fZ', 'now', ?),
                     strftime('%Y-%m-%dT%H:%M:%fZ', 'now', ?))",
        )
        .bind(release_id)
        .bind(target_kind)
        .bind(target_id)
        .bind(format!("hash-{release_guid}"))
        .bind(format!("-{grab_age_days} days"))
        .bind(format!("-{fail_age_days} days"))
        .execute(database.pool())
        .await
        .unwrap();
    }

    /// A failed grab by `release_guid` against `target_kind`/`target_id`,
    /// failed `age_days` in the past.
    async fn insert_aged_failed_grab(
        database: &Db,
        target_kind: &str,
        target_id: i64,
        release_guid: &str,
        age_days: i64,
    ) {
        insert_dated_failed_grab(
            database,
            target_kind,
            target_id,
            release_guid,
            age_days,
            age_days,
        )
        .await;
    }

    /// A failed grab by `release_guid` against `target_kind`/`target_id`,
    /// standing in for what the import tick leaves behind post-probe.
    async fn insert_failed_grab(
        database: &Db,
        target_kind: &str,
        target_id: i64,
        release_guid: &str,
    ) {
        insert_dated_failed_grab(database, target_kind, target_id, release_guid, 0, 0).await;
    }

    async fn action(server: &MockServer) -> AttentionAction {
        Mock::given(method("POST"))
            .respond_with(ResponseTemplate::new(200))
            .mount(server)
            .await;
        AttentionAction::new(
            Notifier::new(server.uri()).unwrap(),
            "operator-topic".to_string(),
        )
    }

    #[tokio::test]
    async fn a_movie_entering_the_queue_notifies_the_operator_once() {
        let (_dir, database) = seeded_database().await;
        insert_no_pt_source_movie(&database).await;

        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .respond_with(ResponseTemplate::new(200))
            .mount(&server)
            .await;
        let action = AttentionAction::new(
            Notifier::new(server.uri()).unwrap(),
            "operator-topic".to_string(),
        );

        let first = action.tick(&database).await.unwrap();
        let second = action.tick(&database).await.unwrap();

        assert_eq!(first.len(), 1, "notifies on the tick it enters the queue");
        assert_eq!(second.len(), 0, "does not repeat while still queued");
        assert_eq!(server.received_requests().await.unwrap().len(), 1);
    }

    #[tokio::test]
    async fn leaving_and_re_entering_the_queue_notifies_again() {
        let (_dir, database) = seeded_database().await;
        let movie_id = insert_no_pt_source_movie(&database).await;

        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .respond_with(ResponseTemplate::new(200))
            .mount(&server)
            .await;
        let action = AttentionAction::new(
            Notifier::new(server.uri()).unwrap(),
            "operator-topic".to_string(),
        );
        action.tick(&database).await.unwrap();

        sqlx::query("UPDATE movies SET blocked = 1 WHERE id = ?")
            .bind(movie_id)
            .execute(database.pool())
            .await
            .unwrap();
        action.tick(&database).await.unwrap();

        sqlx::query("UPDATE movies SET blocked = 0 WHERE id = ?")
            .bind(movie_id)
            .execute(database.pool())
            .await
            .unwrap();
        let third = action.tick(&database).await.unwrap();

        assert_eq!(third.len(), 1, "re-enters and notifies again");
        assert_eq!(server.received_requests().await.unwrap().len(), 2);
    }

    #[tokio::test]
    async fn a_series_entering_the_queue_notifies_once_per_series() {
        let (_dir, database) = seeded_database().await;
        // Twenty broken episodes are one notification, not twenty (§9.5).
        insert_no_pt_source_series(&database, 1, 20).await;

        let server = MockServer::start().await;
        let action = action(&server).await;

        let first = action.tick(&database).await.unwrap();
        let second = action.tick(&database).await.unwrap();

        assert_eq!(first.len(), 1, "one notification for the whole series");
        assert_eq!(second.len(), 0, "does not repeat while still queued");
        assert_eq!(server.received_requests().await.unwrap().len(), 1);
    }

    #[tokio::test]
    async fn a_series_leaving_and_re_entering_the_queue_notifies_again() {
        let (_dir, database) = seeded_database().await;
        let series_id = insert_no_pt_source_series(&database, 1, 1).await;

        let server = MockServer::start().await;
        let action = action(&server).await;
        action.tick(&database).await.unwrap();

        sqlx::query("UPDATE series SET blocked = 1 WHERE id = ?")
            .bind(series_id)
            .execute(database.pool())
            .await
            .unwrap();
        action.tick(&database).await.unwrap();

        sqlx::query("UPDATE series SET blocked = 0 WHERE id = ?")
            .bind(series_id)
            .execute(database.pool())
            .await
            .unwrap();
        let third = action.tick(&database).await.unwrap();

        assert_eq!(third.len(), 1, "re-enters and notifies again");
        assert_eq!(server.received_requests().await.unwrap().len(), 2);
    }

    #[tokio::test]
    async fn an_episode_hard_failed_twice_notifies_its_series() {
        let (_dir, database) = seeded_database().await;
        insert_no_pt_source_series(&database, 1, 0).await;
        let season_id: i64 = sqlx::query_scalar("SELECT id FROM seasons WHERE number = 1")
            .fetch_one(database.pool())
            .await
            .unwrap();
        sqlx::query(
            "INSERT INTO episodes (season_id, number, title, wanted, state)
             VALUES (?, 1, 'Episode 0', 1, 'missing')",
        )
        .bind(season_id)
        .execute(database.pool())
        .await
        .unwrap();
        let episode_id: i64 =
            sqlx::query_scalar("SELECT id FROM episodes WHERE season_id = ? AND number = 1")
                .bind(season_id)
                .fetch_one(database.pool())
                .await
                .unwrap();
        insert_failed_grab(&database, "episode", episode_id, "first").await;
        insert_failed_grab(&database, "episode", episode_id, "second").await;
        // The packs' failures sent this season back to per-episode grabbing;
        // they queue the same series, so it must not double the message.
        insert_failed_grab(&database, "season", season_id, "pack").await;
        insert_failed_grab(&database, "season", season_id, "pack-two").await;

        let server = MockServer::start().await;
        let action = action(&server).await;

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

        assert_eq!(
            outcomes.len(),
            1,
            "both hard-fail conditions roll up to one series notification"
        );
    }

    #[tokio::test]
    async fn a_movie_imported_after_two_hard_fails_leaves_the_queue() {
        let (_dir, database) = seeded_database().await;
        let movie_id = insert_no_pt_source_movie(&database).await;
        insert_failed_grab(&database, "movie", movie_id, "first").await;
        insert_failed_grab(&database, "movie", movie_id, "second").await;

        let server = MockServer::start().await;
        let action = action(&server).await;

        let first = action.tick(&database).await.unwrap();
        assert_eq!(first.len(), 1, "queued while unsatisfied");

        // Imported from a third release: satisfied, no decision needed.
        sqlx::query("UPDATE movies SET state = 'available' WHERE id = ?")
            .bind(movie_id)
            .execute(database.pool())
            .await
            .unwrap();
        let second = action.tick(&database).await.unwrap();

        assert_eq!(second.len(), 0, "leaves the queue once imported");
        assert_eq!(server.received_requests().await.unwrap().len(), 1);
    }

    #[tokio::test]
    async fn an_episode_imported_after_two_hard_fails_leaves_the_queue() {
        let (_dir, database) = seeded_database().await;
        insert_no_pt_source_series(&database, 1, 0).await;
        let season_id: i64 = sqlx::query_scalar("SELECT id FROM seasons WHERE number = 1")
            .fetch_one(database.pool())
            .await
            .unwrap();
        sqlx::query(
            "INSERT INTO episodes (season_id, number, title, wanted, state)
             VALUES (?, 1, 'Episode 0', 1, 'missing')",
        )
        .bind(season_id)
        .execute(database.pool())
        .await
        .unwrap();
        let episode_id: i64 =
            sqlx::query_scalar("SELECT id FROM episodes WHERE season_id = ? AND number = 1")
                .bind(season_id)
                .fetch_one(database.pool())
                .await
                .unwrap();
        insert_failed_grab(&database, "episode", episode_id, "first").await;
        insert_failed_grab(&database, "episode", episode_id, "second").await;

        let server = MockServer::start().await;
        let action = action(&server).await;

        let first = action.tick(&database).await.unwrap();
        assert_eq!(first.len(), 1, "queued while unsatisfied");

        // Imported from a third release: satisfied, no decision needed.
        sqlx::query("UPDATE episodes SET state = 'available' WHERE id = ?")
            .bind(episode_id)
            .execute(database.pool())
            .await
            .unwrap();
        let second = action.tick(&database).await.unwrap();

        assert_eq!(second.len(), 0, "leaves the queue once imported");
        assert_eq!(server.received_requests().await.unwrap().len(), 1);
    }

    /// §5.7: the season lane holds to the same two-distinct-releases bar the
    /// episode lane does, so one bad pack does not notify (#226).
    #[tokio::test]
    async fn one_failed_season_pack_does_not_notify() {
        let (_dir, database) = seeded_database().await;
        insert_no_pt_source_series(&database, 1, 0).await;
        let season_id: i64 = sqlx::query_scalar("SELECT id FROM seasons WHERE number = 1")
            .fetch_one(database.pool())
            .await
            .unwrap();
        insert_wanted_episode(&database, season_id, 1).await;
        insert_failed_grab(&database, "season", season_id, "pack").await;

        let server = MockServer::start().await;
        let action = action(&server).await;

        assert_eq!(
            action.tick(&database).await.unwrap().len(),
            0,
            "one failed pack is the blacklist working, not a decision"
        );

        insert_failed_grab(&database, "season", season_id, "pack-two").await;

        assert_eq!(
            action.tick(&database).await.unwrap().len(),
            1,
            "two distinct packs hard-failed: the operator decides"
        );
    }

    /// §5.7: the queue only holds targets still waiting for a file. A season
    /// whose packs both hard-failed falls back to per-episode grabbing (§6.2);
    /// once every episode has landed the system worked, so the season leaves
    /// the queue at once rather than notifying for 30 days (#238).
    #[tokio::test]
    async fn a_fully_acquired_season_leaves_the_queue() {
        let (_dir, database) = seeded_database().await;
        insert_no_pt_source_series(&database, 1, 0).await;
        let season_id: i64 = sqlx::query_scalar("SELECT id FROM seasons WHERE number = 1")
            .fetch_one(database.pool())
            .await
            .unwrap();
        insert_wanted_episode(&database, season_id, 1).await;
        insert_wanted_episode(&database, season_id, 2).await;
        insert_failed_grab(&database, "season", season_id, "pack").await;
        insert_failed_grab(&database, "season", season_id, "pack-two").await;

        let server = MockServer::start().await;
        let action = action(&server).await;

        assert_eq!(
            action.tick(&database).await.unwrap().len(),
            1,
            "two packs failed and the season still has episodes missing"
        );

        // Per-episode grabbing got the first one. Still a gap, still queued.
        sqlx::query("UPDATE episodes SET state = 'available' WHERE season_id = ? AND number = 1")
            .bind(season_id)
            .execute(database.pool())
            .await
            .unwrap();
        assert_eq!(
            action.tick(&database).await.unwrap().len(),
            0,
            "already notified, and still queued"
        );

        sqlx::query("UPDATE episodes SET state = 'available' WHERE season_id = ?")
            .bind(season_id)
            .execute(database.pool())
            .await
            .unwrap();
        assert_eq!(
            action.tick(&database).await.unwrap().len(),
            0,
            "every episode acquired: nothing left to decide"
        );

        // Proof it actually left rather than merely staying quiet: a season
        // still queued would not notify again on re-entry.
        sqlx::query("UPDATE episodes SET state = 'missing' WHERE season_id = ? AND number = 2")
            .bind(season_id)
            .execute(database.pool())
            .await
            .unwrap();
        assert_eq!(
            action.tick(&database).await.unwrap().len(),
            1,
            "broken again: re-enters the queue and notifies"
        );
        assert_eq!(server.received_requests().await.unwrap().len(), 2);
    }

    /// §5.7: a failure counts for 30 days, so a season dealt with leaves the
    /// queue instead of sitting in it forever (#226).
    #[tokio::test]
    async fn season_failures_older_than_the_window_do_not_notify() {
        let (_dir, database) = seeded_database().await;
        insert_no_pt_source_series(&database, 1, 0).await;
        let season_id: i64 = sqlx::query_scalar("SELECT id FROM seasons WHERE number = 1")
            .fetch_one(database.pool())
            .await
            .unwrap();
        insert_wanted_episode(&database, season_id, 1).await;
        insert_aged_failed_grab(&database, "season", season_id, "old-one", 40).await;
        insert_aged_failed_grab(&database, "season", season_id, "old-two", 35).await;

        let server = MockServer::start().await;
        let action = action(&server).await;

        assert_eq!(
            action.tick(&database).await.unwrap().len(),
            0,
            "failures older than the window are history, not attention"
        );

        insert_aged_failed_grab(&database, "season", season_id, "new-one", 0).await;
        insert_aged_failed_grab(&database, "season", season_id, "new-two", 0).await;

        assert_eq!(
            action.tick(&database).await.unwrap().len(),
            1,
            "still breaking: back in the queue"
        );
    }

    /// §5.7/#239: the window runs from the failure, not the grab. A torrent
    /// that stalls on a slow swarm for five weeks and then hard-fails at
    /// import is fresh evidence the target is broken, however old the grab.
    #[tokio::test]
    async fn a_grab_stalled_past_the_window_before_failing_still_counts() {
        let (_dir, database) = seeded_database().await;
        insert_no_pt_source_series(&database, 1, 0).await;
        let season_id: i64 = sqlx::query_scalar("SELECT id FROM seasons WHERE number = 1")
            .fetch_one(database.pool())
            .await
            .unwrap();
        insert_wanted_episode(&database, season_id, 1).await;
        // Both grabbed 40 days ago — outside the window — but failed today.
        insert_dated_failed_grab(&database, "season", season_id, "stalled-one", 40, 0).await;
        insert_dated_failed_grab(&database, "season", season_id, "stalled-two", 40, 0).await;

        let server = MockServer::start().await;
        let action = action(&server).await;

        assert_eq!(
            action.tick(&database).await.unwrap().len(),
            1,
            "grab age is irrelevant: two fresh failures queue the target"
        );
    }
}