arr/crates/arr-api/src/series.rs

//! The TV side of the library API. See DESIGN.md §4, §4.1 and §4.2.
//!
//! Nothing here is generic over media kind (§11): movies are concrete in
//! `movies.rs` and series are concrete here. What the two do share is the
//! policy loader, the release table and the owner tags, which already exist.
//!
//! Two rules shape the endpoints:
//!
//! - **Intent lives at the leaf** (§4.1). `wanted` is set on an episode.
//!   `auto_track` on the series and `tracked` on the season are rules that
//!   decide what happens to episodes a metadata refresh reveals; neither is
//!   intent, and neither is read to answer "is this wanted".
//! - **Status is derived, never stored** (§4.2). Every series the API returns
//!   carries a status computed from its episodes at request time.

use std::collections::{BTreeMap, HashMap};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use arr_core::tracking::{apply_auto_track, apply_tracked, RefreshedSeason};
use arr_core::{
    derive_series_status, search_backoff, season_grab_reason, EpisodeId, Language, MediaState,
    PerEpisodeReason, RootId, SeasonGrabFacts, SeasonId, SeriesId, SeriesStatus, TitleOverrides,
};
use arr_db::policy::language;
use axum::extract::{Path, Query, State};
use axum::http::StatusCode;
use axum::Json;
use chrono::{DateTime, NaiveDate, NaiveTime, Utc};
use serde::{Deserialize, Serialize};
use utoipa::{IntoParams, ToSchema};

use crate::movies::{pool, rescore, Accepted, ApiError, ErrorBody, Release};
use crate::owners::Owner;
use crate::search::tmdb_client;
use crate::state::{AppState, EpisodeCommand, MetadataCommand, SeasonCommand};

/// A series with the status derived from its episodes (§4.2).
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct Series {
    pub id: i64,
    pub tmdb_id: i64,
    /// The Torznab `tvdbid` (§6.1), when TMDB knows one. Null means indexer
    /// searches fall back to the title text query.
    pub tvdb_id: Option<i64>,
    pub title: String,
    pub year: Option<i64>,
    pub original_language: Option<String>,
    pub root_id: i64,
    /// §4.1. A rule about seasons metadata reveals, not intent.
    pub auto_track: bool,
    pub overrides: serde_json::Value,
    /// Whether the show finished upstream, which `ended` is derived from.
    pub upstream_ended: bool,
    pub blocked: bool,
    /// Path fragment, not a URL; stored at add time and refreshed with
    /// metadata (#145), so pure-SQL views never need TMDB (§9.6).
    pub poster_path: Option<String>,
    /// TMDB's rating, out of 10; `null` when TMDB has no votes for it.
    pub vote_average: Option<f64>,
    /// NULL until a metadata refresh has stamped it (#160). Issue #177: this
    /// is what tells the SPA a series with no seasons yet is still seeding
    /// its first refresh, rather than a title upstream genuinely lists none.
    pub metadata_refreshed_at: Option<String>,
    /// `airing`, `incomplete`, `waiting`, `complete` or `ended` (§4.2).
    pub status: String,
    /// Episodes currently marked wanted (§4.1 — the only intent).
    pub wanted_episodes: i64,
    /// Wanted episodes already on disk.
    pub available_episodes: i64,
}

#[derive(Debug, Deserialize, ToSchema)]
pub struct CreateSeries {
    pub tmdb_id: i64,
    pub title: String,
    pub year: Option<i64>,
    pub original_language: Option<String>,
    pub root_id: i64,
    #[serde(default)]
    pub auto_track: bool,
    #[serde(default)]
    pub blocked: bool,
    #[serde(default)]
    pub upstream_ended: bool,
    #[serde(default = "empty_overrides")]
    pub overrides: serde_json::Value,
}

#[derive(Debug, Deserialize, ToSchema)]
pub struct UpdateSeries {
    pub title: Option<String>,
    pub year: Option<Option<i64>>,
    pub original_language: Option<Option<String>>,
    pub root_id: Option<i64>,
    pub auto_track: Option<bool>,
    pub blocked: Option<bool>,
    pub upstream_ended: Option<bool>,
    pub overrides: Option<serde_json::Value>,
}

/// A season and every episode in it, which is how the UI reads it.
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct Season {
    pub id: i64,
    pub series_id: i64,
    pub number: i64,
    /// §4.1. Whether new episodes of this season arrive wanted.
    pub tracked: bool,
    /// #141. The season vanished upstream while files remained under it,
    /// which is why the row still exists. A conflict for the operator to
    /// resolve; nothing was deleted from disk.
    pub vanished: bool,
    pub episodes: Vec<Episode>,
}

#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct Episode {
    pub id: i64,
    /// The owning series — the episode deck route needs it without walking
    /// seasons first.
    pub series_id: i64,
    pub season_id: i64,
    pub season_number: i64,
    pub number: i64,
    pub title: String,
    pub air_date: Option<String>,
    /// §4.1. The only intent in the TV aggregate.
    pub wanted: bool,
    pub state: String,
    /// #122. The episode vanished upstream while a file of its own remained,
    /// which is why the row still exists. A conflict for the operator to
    /// resolve; nothing was deleted from disk.
    pub vanished: bool,
    pub search_attempts: i64,
    pub last_searched_at: Option<String>,
}

/// A season revealed by metadata, with the episodes it holds.
///
/// The series' `auto_track` decides whether those episodes arrive wanted
/// (§4.1); the request never says so directly.
#[derive(Debug, Deserialize, ToSchema)]
pub struct CreateSeason {
    pub number: i64,
    #[serde(default)]
    pub episodes: Vec<CreateEpisode>,
}

#[derive(Debug, Deserialize, ToSchema)]
pub struct CreateEpisode {
    pub number: i64,
    pub title: String,
    /// ISO-8601 date, as TMDB gives it.
    pub air_date: Option<String>,
}

/// Both fields of a season a person can set by hand.
///
/// Turning `tracked` on marks every already-revealed episode wanted, and
/// episodes revealed later follow while it stays on; turning it off clears
/// `wanted` on all of them (§4.1, #171) — idempotently, so re-sending
/// `tracked: false` to an already-untracked season still clears its
/// stranded wanted rows. `wanted` is the one-click "grab this season", and
/// writes intent onto every episode already in it — the two are deliberately
/// separate (§4.1).
#[derive(Debug, Deserialize, ToSchema)]
pub struct UpdateSeason {
    pub tracked: Option<bool>,
    pub wanted: Option<bool>,
}

#[derive(Debug, Deserialize, ToSchema)]
pub struct UpdateEpisode {
    pub wanted: Option<bool>,
}

fn empty_overrides() -> serde_json::Value {
    serde_json::json!({})
}

fn validate_overrides(value: &serde_json::Value) -> Result<(), ApiError> {
    let Some(object) = value.as_object() else {
        return Err(ApiError::Invalid("overrides must be an object".into()));
    };
    if object
        .keys()
        .any(|key| key != "only_4k" && key != "allow_english_audio")
    {
        return Err(ApiError::Invalid(
            "overrides supports only only_4k and allow_english_audio".into(),
        ));
    }
    if object.values().any(|value| !value.is_boolean()) {
        return Err(ApiError::Invalid("override values must be booleans".into()));
    }
    Ok(())
}

/// The stored columns, before the derived status is attached.
struct SeriesRow {
    id: i64,
    tmdb_id: i64,
    tvdb_id: Option<i64>,
    title: String,
    year: Option<i64>,
    original_language: Option<String>,
    root_id: i64,
    auto_track: bool,
    overrides: serde_json::Value,
    upstream_ended: bool,
    blocked: bool,
    /// Stored at add time and refreshed with metadata (#145), so pure-SQL
    /// views never need TMDB (§9.6).
    poster_path: Option<String>,
    /// TMDB's rating, out of 10; `null` when TMDB has no votes for it.
    vote_average: Option<f64>,
    /// NULL until a metadata refresh has stamped it (#160): the fact the
    /// auto-track rule needs to tell a seeding refresh from a later one.
    metadata_refreshed_at: Option<String>,
}

struct EpisodeRow {
    series_id: i64,
    id: i64,
    season_id: i64,
    season_number: i64,
    number: i64,
    title: String,
    air_date: Option<String>,
    wanted: bool,
    state: String,
    vanished: bool,
    search_attempts: i64,
    last_searched_at: Option<String>,
}

fn media_state(value: &str) -> MediaState {
    match value {
        "downloading" => MediaState::Downloading,
        "available" => MediaState::Available,
        _ => MediaState::Missing,
    }
}

/// An `air_date` as TMDB writes it, or as a full timestamp if one ever
/// arrives that way. Anything else is treated as unknown, which §4.2 already
/// has a meaning for: it cannot pull a series into `airing`.
fn air_date(value: Option<&str>) -> Option<SystemTime> {
    let value = value?;
    let timestamp = if let Ok(date) = value.parse::<NaiveDate>() {
        date.and_time(NaiveTime::MIN).and_utc().timestamp()
    } else {
        value.parse::<DateTime<Utc>>().ok()?.timestamp()
    };
    let seconds = u64::try_from(timestamp.abs()).ok()?;
    if timestamp < 0 {
        UNIX_EPOCH.checked_sub(Duration::from_secs(seconds))
    } else {
        UNIX_EPOCH.checked_add(Duration::from_secs(seconds))
    }
}

fn core_series(row: &SeriesRow) -> arr_core::Series {
    arr_core::Series {
        id: SeriesId(row.id),
        tmdb_id: u64::try_from(row.tmdb_id).unwrap_or_default(),
        title: row.title.clone(),
        year: row
            .year
            .and_then(|year| u16::try_from(year).ok())
            .unwrap_or_default(),
        original_language: row
            .original_language
            .as_deref()
            .map_or(Language::Other(String::new()), language),
        root_id: RootId(row.root_id),
        auto_track: row.auto_track,
        overrides: TitleOverrides::default(),
        upstream_ended: row.upstream_ended,
        blocked: row.blocked,
    }
}

fn core_episode(row: &EpisodeRow) -> arr_core::Episode {
    arr_core::Episode {
        id: EpisodeId(row.id),
        season_id: SeasonId(row.season_id),
        season_number: u16::try_from(row.season_number).unwrap_or_default(),
        number: u16::try_from(row.number).unwrap_or_default(),
        title: row.title.clone(),
        air_date: air_date(row.air_date.as_deref()),
        wanted: row.wanted,
        state: media_state(&row.state),
        search_attempts: u32::try_from(row.search_attempts).unwrap_or_default(),
        last_searched_at: None,
    }
}

fn status_name(status: SeriesStatus) -> &'static str {
    match status {
        SeriesStatus::Airing => "airing",
        SeriesStatus::Incomplete => "incomplete",
        SeriesStatus::Waiting => "waiting",
        SeriesStatus::Complete => "complete",
        SeriesStatus::Ended => "ended",
    }
}

fn with_status(row: &SeriesRow, episodes: &[arr_core::Episode], now: SystemTime) -> Series {
    // §4.2: season 0 is invisible to status, so it stays out of the counters
    // too. A series reading `complete` next to `42/52 eps` is the confusion
    // this avoids. The season number rides on each episode (#131), so no
    // parallel slice can be forgotten.
    let wanted = episodes
        .iter()
        .filter(|episode| episode.wanted && episode.season_number != 0);
    let available = wanted
        .clone()
        .filter(|episode| episode.state == MediaState::Available);
    Series {
        id: row.id,
        tmdb_id: row.tmdb_id,
        tvdb_id: row.tvdb_id,
        title: row.title.clone(),
        year: row.year,
        original_language: row.original_language.clone(),
        root_id: row.root_id,
        auto_track: row.auto_track,
        overrides: row.overrides.clone(),
        upstream_ended: row.upstream_ended,
        blocked: row.blocked,
        poster_path: row.poster_path.clone(),
        vote_average: row.vote_average,
        metadata_refreshed_at: row.metadata_refreshed_at.clone(),
        status: status_name(derive_series_status(&core_series(row), episodes, now)).to_owned(),
        wanted_episodes: i64::try_from(wanted.count()).unwrap_or(i64::MAX),
        available_episodes: i64::try_from(available.count()).unwrap_or(i64::MAX),
    }
}

/// Every episode in the library, keyed by the series it belongs to.
///
/// One query rather than one per series: the whole table is a few thousand
/// rows for a single household (§10), and the status of every listed series
/// needs all of them anyway.
async fn tv_by_series(state: &AppState) -> Result<HashMap<i64, Vec<arr_core::Episode>>, ApiError> {
    let rows = sqlx::query_as!(
        EpisodeRow,
        r#"SELECT se.series_id AS "series_id!: i64", e.id AS "id!: i64", e.season_id AS "season_id!: i64", se.number AS "season_number!: i64", e.number AS "number!: i64", e.title AS "title!: String", e.air_date, e.wanted AS "wanted!: bool", e.state AS "state!: String", e.vanished AS "vanished!: bool", e.search_attempts AS "search_attempts!: i64", e.last_searched_at
           FROM episodes e JOIN seasons se ON se.id = e.season_id"#
    )
    .fetch_all(pool(state)?)
    .await?;

    let mut episodes: HashMap<i64, Vec<arr_core::Episode>> = HashMap::new();
    for row in &rows {
        episodes
            .entry(row.series_id)
            .or_default()
            .push(core_episode(row));
    }
    Ok(episodes)
}

async fn load_series_row(state: &AppState, id: i64) -> Result<SeriesRow, ApiError> {
    sqlx::query_as!(SeriesRow, r#"SELECT id AS "id!: i64", tmdb_id AS "tmdb_id!: i64", tvdb_id, title AS "title!: String", year, original_language, root_id AS "root_id!: i64", auto_track AS "auto_track!: bool", overrides AS "overrides!: serde_json::Value", upstream_ended AS "upstream_ended!: bool", blocked AS "blocked!: bool", poster_path, vote_average, metadata_refreshed_at FROM series WHERE id = ?"#, id)
        .fetch_optional(pool(state)?)
        .await?
        .ok_or(ApiError::SeriesNotFound)
}

async fn load_series(state: &AppState, id: i64) -> Result<Series, ApiError> {
    let row = load_series_row(state, id).await?;
    let episodes = sqlx::query_as!(
        EpisodeRow,
        r#"SELECT se.series_id AS "series_id!: i64", e.id AS "id!: i64", e.season_id AS "season_id!: i64", se.number AS "season_number!: i64", e.number AS "number!: i64", e.title AS "title!: String", e.air_date, e.wanted AS "wanted!: bool", e.state AS "state!: String", e.vanished AS "vanished!: bool", e.search_attempts AS "search_attempts!: i64", e.last_searched_at
           FROM episodes e JOIN seasons se ON se.id = e.season_id WHERE se.series_id = ?"#,
        id
    )
    .fetch_all(pool(state)?)
    .await?;
    let episodes: Vec<_> = episodes.iter().map(core_episode).collect();
    Ok(with_status(&row, &episodes, SystemTime::now()))
}

async fn require_tv_root(state: &AppState, root_id: i64) -> Result<(), ApiError> {
    let tv_root = sqlx::query_scalar!(
        "SELECT EXISTS(SELECT 1 FROM roots WHERE id = ? AND kind = 'tv') AS 'exists!: bool'",
        root_id
    )
    .fetch_one(pool(state)?)
    .await?;
    if tv_root {
        Ok(())
    } else {
        Err(ApiError::Invalid("root_id must name a TV root".into()))
    }
}

#[derive(Debug, Deserialize, IntoParams)]
pub struct ListSeriesQuery {
    /// Restrict to series tagged with this owner (DESIGN.md §4.3).
    pub owner_id: Option<i64>,
}

#[utoipa::path(
    get, path = "/api/series", tag = "series",
    params(ListSeriesQuery),
    responses(
        (status = 200, body = [Series]),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn list(
    State(state): State<AppState>,
    Query(query): Query<ListSeriesQuery>,
) -> Result<Json<Vec<Series>>, ApiError> {
    let rows = if let Some(owner_id) = query.owner_id {
        sqlx::query_as!(SeriesRow, r#"SELECT s.id AS "id!: i64", s.tmdb_id AS "tmdb_id!: i64", s.tvdb_id, s.title AS "title!: String", s.year, s.original_language, s.root_id AS "root_id!: i64", s.auto_track AS "auto_track!: bool", s.overrides AS "overrides!: serde_json::Value", s.upstream_ended AS "upstream_ended!: bool", s.blocked AS "blocked!: bool", poster_path, vote_average, s.metadata_refreshed_at FROM series s JOIN title_owners t ON t.title_kind = 'series' AND t.title_id = s.id WHERE t.owner_id = ? ORDER BY s.title, s.year, s.id"#, owner_id)
            .fetch_all(pool(&state)?)
            .await?
    } else {
        sqlx::query_as!(SeriesRow, r#"SELECT id AS "id!: i64", tmdb_id AS "tmdb_id!: i64", tvdb_id, title AS "title!: String", year, original_language, root_id AS "root_id!: i64", auto_track AS "auto_track!: bool", overrides AS "overrides!: serde_json::Value", upstream_ended AS "upstream_ended!: bool", blocked AS "blocked!: bool", poster_path, vote_average, metadata_refreshed_at FROM series ORDER BY title, year, id"#)
            .fetch_all(pool(&state)?)
            .await?
    };

    let episodes = tv_by_series(&state).await?;
    let now = SystemTime::now();
    let no_episodes: Vec<arr_core::Episode> = Vec::new();
    Ok(Json(
        rows.iter()
            .map(|row| with_status(row, episodes.get(&row.id).unwrap_or(&no_episodes), now))
            .collect(),
    ))
}

#[utoipa::path(
    post, path = "/api/series", tag = "series", request_body = CreateSeries,
    responses(
        (status = 201, body = Series),
        (status = 409, body = ErrorBody),
        (status = 422, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn create(
    State(state): State<AppState>,
    Json(input): Json<CreateSeries>,
) -> Result<(StatusCode, Json<Series>), ApiError> {
    if input.title.trim().is_empty() || input.tmdb_id <= 0 {
        return Err(ApiError::Invalid("tmdb_id and title are required".into()));
    }
    validate_overrides(&input.overrides)?;
    require_tv_root(&state, input.root_id).await?;
    let overrides = serde_json::to_string(&input.overrides)
        .map_err(|error| ApiError::Invalid(error.to_string()))?;
    let title = input.title.trim();
    // §6.1: the TVDB id is what `t=tvsearch` is addressed by, but TMDB not
    // knowing one must not block adding the series — the search falls back
    // to the title text query until a refresh fills it in (#121). The same
    // response carries §9.6's stored artwork fields, so a series added today
    // has a poster before tomorrow's refresh. Best effort either way.
    let tmdb_series = lookup_tmdb_series(&state, input.tmdb_id).await;
    let tvdb_id = tmdb_series.as_ref().and_then(|s| s.tvdb_id).map(i64::from);
    let poster_path = tmdb_series.as_ref().and_then(|s| s.poster_path.clone());
    let backdrop_path = tmdb_series.as_ref().and_then(|s| s.backdrop_path.clone());
    let vote_average = tmdb_series.as_ref().and_then(|s| s.vote_average);
    let result = sqlx::query!(
        "INSERT INTO series (tmdb_id, tvdb_id, title, year, original_language, root_id, auto_track, upstream_ended, blocked, overrides, poster_path, backdrop_path, vote_average) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
        input.tmdb_id, tvdb_id, title, input.year, input.original_language, input.root_id,
        input.auto_track, input.upstream_ended, input.blocked, overrides,
        poster_path, backdrop_path, vote_average,
    )
    .execute(pool(&state)?)
    .await?;
    let series_id = result.last_insert_rowid();
    // Issue #176: a new series has no seasons until a refresh reveals them,
    // and the metadata lane is daily. Ask it to run now. Asynchronous by
    // design — the add is already committed and must not wait on, or fail
    // because of, TMDB. If nothing is draining, the daily sweep still picks
    // the series up: `metadata_refreshed_at` is NULL.
    if let Err(error) = state.send_metadata_command(MetadataCommand::Series { series_id }) {
        tracing::warn!(series_id, %error, "metadata refresh not queued for the new series");
    }
    Ok((
        StatusCode::CREATED,
        Json(load_series(&state, series_id).await?),
    ))
}

/// Best effort: `None` when TMDB has no such id or cannot be reached.
async fn lookup_tmdb_series(state: &AppState, tmdb_id: i64) -> Option<arr_meta::Series> {
    let client = tmdb_client(state).ok()?;
    client.series(u32::try_from(tmdb_id).ok()?).await.ok()
}

#[utoipa::path(
    get, path = "/api/series/{series_id}", tag = "series",
    params(("series_id" = i64, Path, description = "Series row id")),
    responses(
        (status = 200, body = Series),
        (status = 404, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn get(
    State(state): State<AppState>,
    Path(id): Path<i64>,
) -> Result<Json<Series>, ApiError> {
    Ok(Json(load_series(&state, id).await?))
}

#[utoipa::path(
    post, path = "/api/series/{series_id}/refresh-metadata", tag = "series",
    params(("series_id" = i64, Path, description = "Series row id")),
    responses(
        (status = 202, body = Accepted),
        (status = 404, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
/// Issue #177: the same on-demand command `create` sends on add, resent by
/// hand when a page gave up polling for it. Same terms — asynchronous, best
/// effort, and a failure leaves `metadata_refreshed_at` NULL for the daily
/// sweep to pick up.
pub async fn refresh_metadata(
    State(state): State<AppState>,
    Path(series_id): Path<i64>,
) -> Result<(StatusCode, Json<Accepted>), ApiError> {
    load_series_row(&state, series_id).await?;
    state
        .send_metadata_command(MetadataCommand::Series { series_id })
        .map_err(|_| ApiError::Unavailable)?;
    Ok((StatusCode::ACCEPTED, Json(Accepted { accepted: true })))
}

#[utoipa::path(
    patch, path = "/api/series/{series_id}", tag = "series", request_body = UpdateSeries,
    params(("series_id" = i64, Path, description = "Series row id")),
    responses(
        (status = 200, body = Series),
        (status = 404, body = ErrorBody),
        (status = 409, body = ErrorBody),
        (status = 422, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn update(
    State(state): State<AppState>,
    Path(id): Path<i64>,
    Json(input): Json<UpdateSeries>,
) -> Result<Json<Series>, ApiError> {
    let current = load_series_row(&state, id).await?;
    let title = input.title.unwrap_or(current.title);
    if title.trim().is_empty() {
        return Err(ApiError::Invalid("title cannot be empty".into()));
    }
    let overrides = input.overrides.unwrap_or(current.overrides);
    validate_overrides(&overrides)?;
    let overrides =
        serde_json::to_string(&overrides).map_err(|error| ApiError::Invalid(error.to_string()))?;
    let title = title.trim();
    let year = input.year.unwrap_or(current.year);
    let original_language = input.original_language.unwrap_or(current.original_language);
    let root_id = input.root_id.unwrap_or(current.root_id);
    if input.root_id.is_some() {
        require_tv_root(&state, root_id).await?;
    }
    let auto_track = input.auto_track.unwrap_or(current.auto_track);
    let upstream_ended = input.upstream_ended.unwrap_or(current.upstream_ended);
    let blocked = input.blocked.unwrap_or(current.blocked);
    sqlx::query!("UPDATE series SET title = ?, year = ?, original_language = ?, root_id = ?, auto_track = ?, upstream_ended = ?, blocked = ?, overrides = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') WHERE id = ?", title, year, original_language, root_id, auto_track, upstream_ended, blocked, overrides, id)
        .execute(pool(&state)?)
        .await?;
    Ok(Json(load_series(&state, id).await?))
}

#[utoipa::path(
    delete, path = "/api/series/{series_id}", tag = "series",
    params(("series_id" = i64, Path, description = "Series row id")),
    responses(
        (status = 204),
        (status = 404, body = ErrorBody),
        (status = 409, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn delete(
    State(state): State<AppState>,
    Path(id): Path<i64>,
) -> Result<StatusCode, ApiError> {
    // The row is loaded first so a missing series is 404 before anything
    // touches the disk.
    load_series_row(&state, id).await?;
    // The whole-title scope of the same removal the season and episode
    // endpoints use (#174). The intent clear it performs is redundant here —
    // the episode rows go with the series row below — but sharing one path
    // is what keeps the three scopes from drifting apart.
    remove_scope_files(&state, FileScope::Series(id)).await?;
    // Owner tags go with the title they tagged.
    sqlx::query!(
        "DELETE FROM title_owners WHERE title_kind = 'series' AND title_id = ?",
        id
    )
    .execute(pool(&state)?)
    .await?;
    let result = sqlx::query!("DELETE FROM series WHERE id = ?", id)
        .execute(pool(&state)?)
        .await?;
    if result.rows_affected() == 0 {
        return Err(ApiError::SeriesNotFound);
    }
    Ok(StatusCode::NO_CONTENT)
}

/// What one removal call covers. The series variant is the whole title; the
/// other two are the sub-series scopes #174 adds, and they must never widen
/// past themselves — removing one episode leaves its siblings and the rest
/// of the season on disk.
#[derive(Debug, Clone, Copy)]
enum FileScope {
    /// Series row id.
    Series(i64),
    /// Season row id, not its number.
    Season(i64),
    /// Episode row id.
    Episode(i64),
}

/// Unlink what this scope put under its series' root. Mirrors the movie
/// handler in `movies.rs`, and is the only unlink path below a series (#174).
///
/// The service knows only what it wrote (§2), so the targets come from
/// `media_files`, never from a scan and never from re-deriving the §7.4 name
/// — a series renamed after import would derive a folder that does not exist
/// while the real one stayed.
///
/// What a target *is* depends on the scope. A whole series resolves each file
/// to its title folder, which makes that delete atomic (§7.4): season
/// subfolders, sidecar subtitles and artwork go with it. A season or a single
/// episode resolves to the recorded file and nothing else — the title folder
/// holds the siblings this call must not touch, and a season subfolder would
/// have to be re-derived to be named, which §2 forbids. Sidecars beside a
/// removed episode therefore stay; they are not rows this service wrote.
///
/// The torrent is untouched (§7.3). It keeps seeding under its own rule and
/// the reaper deletes it; a hardlinked file loses only its library name.
///
/// Failure leaves the database alone, so the operator sees the files still
/// recorded and can retry rather than losing the record of what is on disk.
async fn remove_library_files(state: &AppState, scope: FileScope) -> Result<(), ApiError> {
    let root = scope_root(state, scope).await?;
    let paths = scope_paths(state, scope).await?;

    let mut targets: Vec<std::path::PathBuf> = Vec::new();
    for path in &paths {
        let resolved = match scope {
            FileScope::Series(_) => crate::movies::title_target(&root, path),
            FileScope::Season(_) | FileScope::Episode(_) => contained_file(&root, path),
        };
        let Some(target) = resolved else {
            // Outside its own root: not ours to delete. The row still goes,
            // so the operator sees the file leave the library and stay on
            // disk.
            tracing::warn!(%path, %root, "media file is outside its root, not deleted");
            continue;
        };
        if !targets.contains(&target) {
            targets.push(target);
        }
    }

    for target in targets {
        let metadata = match tokio::fs::symlink_metadata(&target).await {
            Ok(metadata) => metadata,
            // Already gone is the state we wanted.
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
            Err(error) => return Err(ApiError::Filesystem(error.to_string())),
        };
        let removed = if metadata.is_dir() {
            tokio::fs::remove_dir_all(&target).await
        } else {
            tokio::fs::remove_file(&target).await
        };
        match removed {
            Ok(()) => tracing::info!(target = %target.display(), "removed library files"),
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
            Err(error) => return Err(ApiError::Filesystem(error.to_string())),
        }
    }
    Ok(())
}

/// The library root the scope's series sits on. Every unlink is measured
/// against it, so it is looked up rather than assumed.
async fn scope_root(state: &AppState, scope: FileScope) -> Result<String, ApiError> {
    Ok(match scope {
        FileScope::Series(id) => {
            sqlx::query_scalar!(
                r#"SELECT r.path AS "path!: String" FROM roots r JOIN series s ON s.root_id = r.id WHERE s.id = ?"#,
                id
            )
            .fetch_one(pool(state)?)
            .await?
        }
        FileScope::Season(id) => {
            sqlx::query_scalar!(
                r#"SELECT r.path AS "path!: String"
                   FROM roots r
                   JOIN series s ON s.root_id = r.id
                   JOIN seasons se ON se.series_id = s.id
                   WHERE se.id = ?"#,
                id
            )
            .fetch_one(pool(state)?)
            .await?
        }
        FileScope::Episode(id) => {
            sqlx::query_scalar!(
                r#"SELECT r.path AS "path!: String"
                   FROM roots r
                   JOIN series s ON s.root_id = r.id
                   JOIN seasons se ON se.series_id = s.id
                   JOIN episodes e ON e.season_id = se.id
                   WHERE e.id = ?"#,
                id
            )
            .fetch_one(pool(state)?)
            .await?
        }
    })
}

/// Every file this service recorded for the scope, and nothing else. The
/// `WHERE` clause is the whole guard against a narrow call widening.
async fn scope_paths(state: &AppState, scope: FileScope) -> Result<Vec<String>, ApiError> {
    Ok(match scope {
        FileScope::Series(id) => {
            sqlx::query_scalar!(
                r#"SELECT mf.path AS "path!: String"
                   FROM media_files mf
                   JOIN episodes e ON mf.owner_kind = 'episode' AND e.id = mf.owner_id
                   JOIN seasons se ON se.id = e.season_id
                   WHERE se.series_id = ?"#,
                id
            )
            .fetch_all(pool(state)?)
            .await?
        }
        FileScope::Season(id) => {
            sqlx::query_scalar!(
                r#"SELECT mf.path AS "path!: String"
                   FROM media_files mf
                   JOIN episodes e ON mf.owner_kind = 'episode' AND e.id = mf.owner_id
                   WHERE e.season_id = ?"#,
                id
            )
            .fetch_all(pool(state)?)
            .await?
        }
        FileScope::Episode(id) => {
            sqlx::query_scalar!(
                r#"SELECT path AS "path!: String"
                   FROM media_files WHERE owner_kind = 'episode' AND owner_id = ?"#,
                id
            )
            .fetch_all(pool(state)?)
            .await?
        }
    })
}

/// The one recorded file, when it really sits inside the root. `None` when it
/// does not, which is the guard that keeps a sub-series delete inside the
/// library it belongs to.
///
/// Unlike [`crate::movies::title_target`] this keeps the whole relative path
/// rather than its first component, so it can only ever name the file the row
/// records. Every component must be a plain name: one `..` anywhere would
/// climb back out of the root it just proved it was under.
fn contained_file(root: &str, file: &str) -> Option<std::path::PathBuf> {
    let root = std::path::Path::new(root);
    let relative = std::path::Path::new(file).strip_prefix(root).ok()?;
    if relative.as_os_str().is_empty() {
        // The root itself is never a file of ours.
        return None;
    }
    if !relative
        .components()
        .all(|component| matches!(component, std::path::Component::Normal(_)))
    {
        return None;
    }
    Some(root.join(relative))
}

/// Removes the files a scope below a series covers, drops their `media_files`
/// rows and clears the intent behind them (#174).
///
/// One action, both halves: the operator asked for removal and un-wanting
/// together, not two controls to remember to use in order.
///
/// The season and episode rows themselves stay. TMDB owns that metadata and
/// the next refresh would recreate them, which is what separates this from
/// `DELETE /api/series/{id}`.
///
/// Disk first, database second, so a filesystem failure leaves the rows
/// describing what is still there.
async fn remove_scope_files(state: &AppState, scope: FileScope) -> Result<(), ApiError> {
    remove_library_files(state, scope).await?;

    // One transaction: the file rows and the intent they carried land
    // together or not at all.
    let mut transaction = pool(state)?.begin().await?;
    // `media_files.path` is UNIQUE and the owner is polymorphic, so nothing
    // cascades and nothing here deletes an owning row: leaving these behind
    // would block re-importing the same path after a re-grab.
    match scope {
        FileScope::Series(id) => {
            sqlx::query!(
                "DELETE FROM media_files WHERE owner_kind = 'episode' AND owner_id IN (
                     SELECT e.id FROM episodes e JOIN seasons se ON se.id = e.season_id
                     WHERE se.series_id = ?)",
                id
            )
            .execute(&mut *transaction)
            .await?;
        }
        FileScope::Season(id) => {
            sqlx::query!(
                "DELETE FROM media_files WHERE owner_kind = 'episode' AND owner_id IN (
                     SELECT e.id FROM episodes e WHERE e.season_id = ?)",
                id
            )
            .execute(&mut *transaction)
            .await?;
        }
        FileScope::Episode(id) => {
            sqlx::query!(
                "DELETE FROM media_files WHERE owner_kind = 'episode' AND owner_id = ?",
                id
            )
            .execute(&mut *transaction)
            .await?;
        }
    }

    // The intent goes through `arr_core::tracking::apply_tracked` with
    // `false`, the same §4.1 rule an untracked season runs (#171), so both
    // paths agree on what clearing intent means. An episode-scoped call
    // hands it a slice of one: the rule cannot reach a sibling it was not
    // given.
    let rows = match scope {
        FileScope::Series(id) => {
            sqlx::query_as!(
                EpisodeRow,
                r#"SELECT se.series_id AS "series_id!: i64", e.id AS "id!: i64", e.season_id AS "season_id!: i64", se.number AS "season_number!: i64", e.number AS "number!: i64", e.title AS "title!: String", e.air_date, e.wanted AS "wanted!: bool", e.state AS "state!: String", e.vanished AS "vanished!: bool", e.search_attempts AS "search_attempts!: i64", e.last_searched_at
                   FROM episodes e JOIN seasons se ON se.id = e.season_id WHERE se.series_id = ?"#,
                id
            )
            .fetch_all(&mut *transaction)
            .await?
        }
        FileScope::Season(id) => {
            sqlx::query_as!(
                EpisodeRow,
                r#"SELECT se.series_id AS "series_id!: i64", e.id AS "id!: i64", e.season_id AS "season_id!: i64", se.number AS "season_number!: i64", e.number AS "number!: i64", e.title AS "title!: String", e.air_date, e.wanted AS "wanted!: bool", e.state AS "state!: String", e.vanished AS "vanished!: bool", e.search_attempts AS "search_attempts!: i64", e.last_searched_at
                   FROM episodes e JOIN seasons se ON se.id = e.season_id WHERE e.season_id = ?"#,
                id
            )
            .fetch_all(&mut *transaction)
            .await?
        }
        FileScope::Episode(id) => {
            sqlx::query_as!(
                EpisodeRow,
                r#"SELECT se.series_id AS "series_id!: i64", e.id AS "id!: i64", e.season_id AS "season_id!: i64", se.number AS "season_number!: i64", e.number AS "number!: i64", e.title AS "title!: String", e.air_date, e.wanted AS "wanted!: bool", e.state AS "state!: String", e.vanished AS "vanished!: bool", e.search_attempts AS "search_attempts!: i64", e.last_searched_at
                   FROM episodes e JOIN seasons se ON se.id = e.season_id WHERE e.id = ?"#,
                id
            )
            .fetch_all(&mut *transaction)
            .await?
        }
    };
    let mut episodes: Vec<_> = rows.iter().map(core_episode).collect();
    apply_tracked(false, &mut episodes);
    for episode in &episodes {
        // `available` was true of an episode with a file. It no longer has
        // one, so it goes back to `missing` the way a failed import already
        // puts it — `wanted` is now 0, so this opens no gap. `downloading`
        // is left alone: that grab is still in flight.
        sqlx::query!(
            "UPDATE episodes
             SET    wanted = ?,
                    state = CASE WHEN state = 'available' THEN 'missing' ELSE state END,
                    updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
             WHERE  id = ?",
            episode.wanted,
            episode.id.0
        )
        .execute(&mut *transaction)
        .await?;
    }
    transaction.commit().await?;
    Ok(())
}

#[utoipa::path(
    delete, path = "/api/series/{series_id}/seasons/{season_number}/files", tag = "series",
    params(
        ("series_id" = i64, Path, description = "Series row id"),
        ("season_number" = i64, Path, description = "Season number, not its row id")
    ),
    responses(
        (status = 204),
        (status = 404, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn delete_season_files(
    State(state): State<AppState>,
    Path((series_id, number)): Path<(i64, i64)>,
) -> Result<StatusCode, ApiError> {
    // Resolved first so an unknown series or season is 404 before anything
    // touches the disk. A season with nothing on disk is not an error — the
    // call still clears intent.
    load_series_row(&state, series_id).await?;
    let season_id = sqlx::query_scalar!(
        r#"SELECT id AS "id!: i64" FROM seasons WHERE series_id = ? AND number = ?"#,
        series_id,
        number
    )
    .fetch_optional(pool(&state)?)
    .await?
    .ok_or(ApiError::SeasonNotFound)?;

    remove_scope_files(&state, FileScope::Season(season_id)).await?;
    Ok(StatusCode::NO_CONTENT)
}

#[utoipa::path(
    delete, path = "/api/episodes/{episode_id}/files", tag = "series",
    params(("episode_id" = i64, Path, description = "Episode row id")),
    responses(
        (status = 204),
        (status = 404, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn delete_episode_files(
    State(state): State<AppState>,
    Path(id): Path<i64>,
) -> Result<StatusCode, ApiError> {
    // 404 before the disk, and an episode with no file still clears intent.
    load_episode(&state, id).await?;
    remove_scope_files(&state, FileScope::Episode(id)).await?;
    Ok(StatusCode::NO_CONTENT)
}

async fn load_seasons(state: &AppState, series_id: i64) -> Result<Vec<Season>, ApiError> {
    let seasons = sqlx::query!(
        // §9.6: newest-first, so season 0 lands last under plain numeric
        // descending order — exactly where the design puts it.
        r#"SELECT id AS "id!: i64", series_id AS "series_id!: i64", number AS "number!: i64", tracked AS "tracked!: bool", vanished AS "vanished!: bool" FROM seasons WHERE series_id = ? ORDER BY number DESC"#,
        series_id
    )
    .fetch_all(pool(state)?)
    .await?;
    let episodes = sqlx::query_as!(
        EpisodeRow,
        r#"SELECT se.series_id AS "series_id!: i64", e.id AS "id!: i64", e.season_id AS "season_id!: i64", se.number AS "season_number!: i64", e.number AS "number!: i64", e.title AS "title!: String", e.air_date, e.wanted AS "wanted!: bool", e.state AS "state!: String", e.vanished AS "vanished!: bool", e.search_attempts AS "search_attempts!: i64", e.last_searched_at
           FROM episodes e JOIN seasons se ON se.id = e.season_id WHERE se.series_id = ? ORDER BY se.number DESC, e.number DESC"#,
        series_id
    )
    .fetch_all(pool(state)?)
    .await?;

    Ok(seasons
        .into_iter()
        .map(|season| Season {
            id: season.id,
            series_id: season.series_id,
            number: season.number,
            tracked: season.tracked,
            vanished: season.vanished,
            episodes: episodes
                .iter()
                .filter(|episode| episode.season_id == season.id)
                .map(|episode| Episode {
                    id: episode.id,
                    series_id,
                    season_id: episode.season_id,
                    season_number: episode.season_number,
                    number: episode.number,
                    title: episode.title.clone(),
                    air_date: episode.air_date.clone(),
                    wanted: episode.wanted,
                    state: episode.state.clone(),
                    vanished: episode.vanished,
                    search_attempts: episode.search_attempts,
                    last_searched_at: episode.last_searched_at.clone(),
                })
                .collect(),
        })
        .collect())
}

#[utoipa::path(
    get, path = "/api/series/{series_id}/seasons", tag = "series",
    params(("series_id" = i64, Path, description = "Series row id")),
    responses(
        (status = 200, body = [Season]),
        (status = 404, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn seasons(
    State(state): State<AppState>,
    Path(series_id): Path<i64>,
) -> Result<Json<Vec<Season>>, ApiError> {
    load_series_row(&state, series_id).await?;
    Ok(Json(load_seasons(&state, series_id).await?))
}

/// Records a season metadata has revealed, applying the series' tracking rule.
#[utoipa::path(
    post, path = "/api/series/{series_id}/seasons", tag = "series",
    request_body = CreateSeason,
    params(("series_id" = i64, Path, description = "Series row id")),
    responses(
        (status = 201, body = Season),
        (status = 404, body = ErrorBody),
        (status = 409, body = ErrorBody),
        (status = 422, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn create_season(
    State(state): State<AppState>,
    Path(series_id): Path<i64>,
    Json(input): Json<CreateSeason>,
) -> Result<(StatusCode, Json<Season>), ApiError> {
    let series = load_series_row(&state, series_id).await?;
    if input.number < 0 {
        return Err(ApiError::Invalid("season number cannot be negative".into()));
    }
    if input.episodes.iter().any(|episode| episode.number < 0) {
        return Err(ApiError::Invalid(
            "episode number cannot be negative".into(),
        ));
    }
    // The column rejects the empty string (#153); say so before the database
    // has to.
    if input
        .episodes
        .iter()
        .any(|episode| episode.title.trim().is_empty())
    {
        return Err(ApiError::Invalid("episode title cannot be empty".into()));
    }
    let mut numbers: Vec<i64> = input
        .episodes
        .iter()
        .map(|episode| episode.number)
        .collect();
    numbers.sort_unstable();
    if numbers.windows(2).any(|pair| pair[0] == pair[1]) {
        return Err(ApiError::Invalid(
            "episode numbers must be unique within the season".into(),
        ));
    }

    // §4.1. The request does not say whether the episodes are wanted; the
    // series' auto_track rule does, through the one function that owns it.
    // #160: whether a refresh has happened yet rides in from the row — a
    // hand-added season on a never-refreshed series is still seeding, and
    // §4.1 does not track what was there at add time.
    let mut revealed = [RefreshedSeason {
        season: arr_core::Season {
            id: SeasonId(0),
            series_id: SeriesId(series_id),
            number: u16::try_from(input.number).unwrap_or_default(),
            tracked: false,
        },
        episodes: input
            .episodes
            .iter()
            .map(|episode| arr_core::Episode {
                id: EpisodeId(0),
                season_id: SeasonId(0),
                season_number: u16::try_from(input.number).unwrap_or_default(),
                number: u16::try_from(episode.number).unwrap_or_default(),
                title: episode.title.clone(),
                air_date: air_date(episode.air_date.as_deref()),
                wanted: false,
                state: MediaState::Missing,
                search_attempts: 0,
                last_searched_at: None,
            })
            .collect(),
        is_new: true,
    }];
    apply_auto_track(
        &core_series(&series),
        series.metadata_refreshed_at.is_some(),
        &mut revealed,
    );
    let [revealed] = revealed;

    // One transaction: a rejected episode must not leave the season behind,
    // or the retry that fixes the request collides with it instead.
    let mut transaction = pool(&state)?.begin().await?;
    let season_id = sqlx::query!(
        "INSERT INTO seasons (series_id, number, tracked) VALUES (?, ?, ?)",
        series_id,
        input.number,
        revealed.season.tracked
    )
    .execute(&mut *transaction)
    .await?
    .last_insert_rowid();

    for (episode, source) in revealed.episodes.iter().zip(&input.episodes) {
        sqlx::query!(
            "INSERT INTO episodes (season_id, number, title, air_date, wanted) VALUES (?, ?, ?, ?, ?)",
            season_id,
            source.number,
            source.title,
            source.air_date,
            episode.wanted
        )
        .execute(&mut *transaction)
        .await?;
    }
    transaction.commit().await?;

    let seasons = load_seasons(&state, series_id).await?;
    let season = seasons
        .into_iter()
        .find(|season| season.id == season_id)
        .ok_or(ApiError::SeasonNotFound)?;
    Ok((StatusCode::CREATED, Json(season)))
}

#[utoipa::path(
    patch, path = "/api/series/{series_id}/seasons/{season_number}", tag = "series",
    request_body = UpdateSeason,
    params(
        ("series_id" = i64, Path, description = "Series row id"),
        ("season_number" = i64, Path, description = "Season number, not its row id")
    ),
    responses(
        (status = 200, body = Season),
        (status = 404, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn update_season(
    State(state): State<AppState>,
    Path((series_id, number)): Path<(i64, i64)>,
    Json(input): Json<UpdateSeason>,
) -> Result<Json<Season>, ApiError> {
    load_series_row(&state, series_id).await?;
    let season = sqlx::query!(
        r#"SELECT id AS "id!: i64", tracked AS "tracked!: bool" FROM seasons WHERE series_id = ? AND number = ?"#,
        series_id,
        number
    )
    .fetch_optional(pool(&state)?)
    .await?
    .ok_or(ApiError::SeasonNotFound)?;
    let season_id = season.id;

    // One transaction: the flag write and the episode intent it implies
    // land together or not at all (#171).
    let mut transaction = pool(&state)?.begin().await?;

    if let Some(tracked) = input.tracked {
        sqlx::query!(
            "UPDATE seasons SET tracked = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') WHERE id = ?",
            tracked,
            season_id
        )
        .execute(&mut *transaction)
        .await?;
        // §4.1, as `arr_core::tracking::apply_tracked` decides it: turning
        // tracking on marks every already-revealed episode wanted; turning
        // it off clears them all, season 0 included. The rule runs even when
        // the flag already holds this value (#171): re-sending `false` to an
        // untracked season is how stranded wanted rows get cleared.
        let rows = sqlx::query_as!(
            EpisodeRow,
            r#"SELECT se.series_id AS "series_id!: i64", e.id AS "id!: i64", e.season_id AS "season_id!: i64", se.number AS "season_number!: i64", e.number AS "number!: i64", e.title AS "title!: String", e.air_date, e.wanted AS "wanted!: bool", e.state AS "state!: String", e.vanished AS "vanished!: bool", e.search_attempts AS "search_attempts!: i64", e.last_searched_at
               FROM episodes e JOIN seasons se ON se.id = e.season_id WHERE e.season_id = ?"#,
            season_id
        )
        .fetch_all(&mut *transaction)
        .await?;
        let mut episodes: Vec<_> = rows.iter().map(core_episode).collect();
        apply_tracked(tracked, &mut episodes);
        for episode in &episodes {
            sqlx::query!(
                "UPDATE episodes SET wanted = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') WHERE id = ?",
                episode.wanted,
                episode.id.0
            )
            .execute(&mut *transaction)
            .await?;
        }
    }
    // §4.1. Marking a season wanted is intent written onto its episodes, so
    // an untracked series with one wanted season needs no special case.
    if let Some(wanted) = input.wanted {
        sqlx::query!(
            "UPDATE episodes SET wanted = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') WHERE season_id = ?",
            wanted,
            season_id
        )
        .execute(&mut *transaction)
        .await?;
    }
    transaction.commit().await?;

    let seasons = load_seasons(&state, series_id).await?;
    seasons
        .into_iter()
        .find(|season| season.id == season_id)
        .map(Json)
        .ok_or(ApiError::SeasonNotFound)
}

async fn load_episode(state: &AppState, id: i64) -> Result<Episode, ApiError> {
    let row = sqlx::query_as!(
        EpisodeRow,
        r#"SELECT se.series_id AS "series_id!: i64", e.id AS "id!: i64", e.season_id AS "season_id!: i64", se.number AS "season_number!: i64", e.number AS "number!: i64", e.title AS "title!: String", e.air_date, e.wanted AS "wanted!: bool", e.state AS "state!: String", e.vanished AS "vanished!: bool", e.search_attempts AS "search_attempts!: i64", e.last_searched_at
           FROM episodes e JOIN seasons se ON se.id = e.season_id WHERE e.id = ?"#,
        id
    )
    .fetch_optional(pool(state)?)
    .await?
    .ok_or(ApiError::EpisodeNotFound)?;
    Ok(Episode {
        id: row.id,
        series_id: row.series_id,
        season_id: row.season_id,
        season_number: row.season_number,
        number: row.number,
        title: row.title,
        air_date: row.air_date,
        wanted: row.wanted,
        state: row.state,
        vanished: row.vanished,
        search_attempts: row.search_attempts,
        last_searched_at: row.last_searched_at,
    })
}

#[utoipa::path(
    get, path = "/api/episodes/{episode_id}", tag = "series",
    params(("episode_id" = i64, Path, description = "Episode row id")),
    responses(
        (status = 200, body = Episode),
        (status = 404, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn get_episode(
    State(state): State<AppState>,
    Path(id): Path<i64>,
) -> Result<Json<Episode>, ApiError> {
    Ok(Json(load_episode(&state, id).await?))
}

/// Sets the only intent the TV aggregate carries (§4.1).
#[utoipa::path(
    patch, path = "/api/episodes/{episode_id}", tag = "series",
    request_body = UpdateEpisode,
    params(("episode_id" = i64, Path, description = "Episode row id")),
    responses(
        (status = 200, body = Episode),
        (status = 404, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn update_episode(
    State(state): State<AppState>,
    Path(id): Path<i64>,
    Json(input): Json<UpdateEpisode>,
) -> Result<Json<Episode>, ApiError> {
    load_episode(&state, id).await?;
    if let Some(wanted) = input.wanted {
        sqlx::query!(
            "UPDATE episodes SET wanted = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') WHERE id = ?",
            wanted,
            id
        )
        .execute(pool(&state)?)
        .await?;
    }
    Ok(Json(load_episode(&state, id).await?))
}

#[utoipa::path(
    post, path = "/api/episodes/{episode_id}/search", tag = "series",
    params(("episode_id" = i64, Path, description = "Episode row id")),
    responses(
        (status = 202, body = Accepted),
        (status = 404, body = ErrorBody),
        (status = 409, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn search_episode(
    State(state): State<AppState>,
    Path(id): Path<i64>,
) -> Result<(StatusCode, Json<Accepted>), ApiError> {
    load_episode(&state, id).await?;
    // §6.3. `blocked` stops targeted search for the whole series.
    let blocked = sqlx::query_scalar!(
        r#"SELECT s.blocked AS "blocked!: bool" FROM episodes e JOIN seasons se ON se.id = e.season_id JOIN series s ON s.id = se.series_id WHERE e.id = ?"#,
        id
    )
    .fetch_one(pool(&state)?)
    .await?;
    if blocked {
        return Err(ApiError::Conflict("series is blocked".into()));
    }
    state
        .send_episode_command(EpisodeCommand::Search { episode_id: id })
        .map_err(|_| ApiError::Unavailable)?;
    Ok((StatusCode::ACCEPTED, Json(Accepted { accepted: true })))
}

#[utoipa::path(
    get, path = "/api/episodes/{episode_id}/releases", tag = "series",
    params(("episode_id" = i64, Path, description = "Episode row id")),
    responses(
        (status = 200, body = [Release]),
        (status = 404, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn episode_releases(
    State(state): State<AppState>,
    Path(id): Path<i64>,
) -> Result<Json<Vec<Release>>, ApiError> {
    let episode = load_episode(&state, id).await?;
    let mut releases = sqlx::query_as!(Release, r#"SELECT r.id AS "id!: i64", r.indexer_id AS "indexer_id!: i64", r.guid AS "guid!: String", r.name AS "name!: String", r.size AS "size!: i64", r.seeders, r.publish_date, r.download_url AS "download_url!: String", r.parsed AS "parsed!: serde_json::Value", r.score, r.verdict, r.rejected_rule FROM releases r JOIN episode_releases er ON er.release_id = r.id WHERE er.episode_id = ? ORDER BY CASE r.verdict WHEN 'eligible' THEN 0 WHEN 'waived' THEN 1 ELSE 2 END, r.score DESC, r.id"#, id)
        .fetch_all(pool(&state)?)
        .await?;
    let policy = state
        .database()
        .ok_or(ApiError::Unavailable)?
        .episode_policy(id)
        .await
        .map_err(|error| ApiError::Database(error.to_string()))?
        .ok_or(ApiError::EpisodeNotFound)?
        .policy;
    let lengths = season_lengths(&state, episode.series_id).await?;
    rescore(&mut releases, &policy, Some(&lengths))?;
    Ok(Json(releases))
}

#[utoipa::path(
    post, path = "/api/episodes/{episode_id}/releases/{release_id}/grab", tag = "series",
    params(("episode_id" = i64, Path), ("release_id" = i64, Path)),
    responses(
        (status = 202, body = Accepted),
        (status = 404, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn grab_episode(
    State(state): State<AppState>,
    Path((episode_id, release_id)): Path<(i64, i64)>,
) -> Result<(StatusCode, Json<Accepted>), ApiError> {
    let exists = sqlx::query_scalar!("SELECT EXISTS(SELECT 1 FROM episode_releases WHERE episode_id = ? AND release_id = ?) AS 'exists!: bool'", episode_id, release_id)
        .fetch_one(pool(&state)?)
        .await?;
    if !exists {
        return Err(ApiError::EpisodeNotFound);
    }
    state
        .send_episode_command(EpisodeCommand::Grab {
            episode_id,
            release_id,
        })
        .map_err(|_| ApiError::Unavailable)?;
    Ok((StatusCode::ACCEPTED, Json(Accepted { accepted: true })))
}

/// Resolves a season by its number within one series, so the deck is
/// addressed the way the UI shows seasons (`/series/{id}/seasons/{n}`).
/// Per-season episode counts for one series: the divisor data behind
/// per-episode size normalisation (`DESIGN.md` §5.5). A season with no
/// revealed episodes counts zero, which `claimed_episode_count` treats as
/// unknown.
pub(crate) async fn season_lengths(
    state: &AppState,
    series_id: i64,
) -> Result<BTreeMap<u32, u32>, ApiError> {
    let rows = sqlx::query!(
        r#"SELECT se.number AS "number!: i64", COUNT(e.id) AS "episodes!: i64"
           FROM seasons se LEFT JOIN episodes e ON e.season_id = se.id
           WHERE se.series_id = ?
           GROUP BY se.number"#,
        series_id
    )
    .fetch_all(pool(state)?)
    .await?;
    Ok(rows
        .into_iter()
        .map(|row| {
            (
                u32::try_from(row.number).unwrap_or_default(),
                u32::try_from(row.episodes).unwrap_or_default(),
            )
        })
        .collect())
}

async fn load_season_id(state: &AppState, series_id: i64, number: i64) -> Result<i64, ApiError> {
    sqlx::query_scalar!(
        r#"SELECT id AS "id!: i64" FROM seasons WHERE series_id = ? AND number = ?"#,
        series_id,
        number
    )
    .fetch_optional(pool(state)?)
    .await?
    .ok_or(ApiError::SeasonNotFound)
}

#[utoipa::path(
    post, path = "/api/series/{series_id}/seasons/{season_number}/search", tag = "series",
    params(
        ("series_id" = i64, Path, description = "Series row id"),
        ("season_number" = i64, Path, description = "Season number, not its row id")
    ),
    responses(
        (status = 202, body = Accepted),
        (status = 404, body = ErrorBody),
        (status = 409, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn search_season(
    State(state): State<AppState>,
    Path((series_id, number)): Path<(i64, i64)>,
) -> Result<(StatusCode, Json<Accepted>), ApiError> {
    load_series_row(&state, series_id).await?;
    let season_id = load_season_id(&state, series_id, number).await?;
    // §6.3. `blocked` stops targeted search for the whole series.
    let blocked = sqlx::query_scalar!(
        r#"SELECT blocked AS "blocked!: bool" FROM series WHERE id = ?"#,
        series_id
    )
    .fetch_one(pool(&state)?)
    .await?;
    if blocked {
        return Err(ApiError::Conflict("series is blocked".into()));
    }
    state
        .send_season_command(SeasonCommand::Search { season_id })
        .map_err(|_| ApiError::Unavailable)?;
    Ok((StatusCode::ACCEPTED, Json(Accepted { accepted: true })))
}

#[utoipa::path(
    get, path = "/api/series/{series_id}/seasons/{season_number}/releases", tag = "series",
    params(
        ("series_id" = i64, Path, description = "Series row id"),
        ("season_number" = i64, Path, description = "Season number, not its row id")
    ),
    responses(
        (status = 200, body = [Release]),
        (status = 404, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn season_releases(
    State(state): State<AppState>,
    Path((series_id, number)): Path<(i64, i64)>,
) -> Result<Json<Vec<Release>>, ApiError> {
    load_series_row(&state, series_id).await?;
    let season_id = load_season_id(&state, series_id, number).await?;
    let mut releases = sqlx::query_as!(Release, r#"SELECT r.id AS "id!: i64", r.indexer_id AS "indexer_id!: i64", r.guid AS "guid!: String", r.name AS "name!: String", r.size AS "size!: i64", r.seeders, r.publish_date, r.download_url AS "download_url!: String", r.parsed AS "parsed!: serde_json::Value", r.score, r.verdict, r.rejected_rule FROM releases r JOIN season_releases sr ON sr.release_id = r.id WHERE sr.season_id = ? ORDER BY CASE r.verdict WHEN 'eligible' THEN 0 WHEN 'waived' THEN 1 ELSE 2 END, r.score DESC, r.id"#, season_id)
        .fetch_all(pool(&state)?)
        .await?;
    let policy = state
        .database()
        .ok_or(ApiError::Unavailable)?
        .season_policy(season_id)
        .await
        .map_err(|error| ApiError::Database(error.to_string()))?
        .ok_or(ApiError::SeasonNotFound)?
        .policy;
    let lengths = season_lengths(&state, series_id).await?;
    rescore(&mut releases, &policy, Some(&lengths))?;
    Ok(Json(releases))
}

/// Which lane a season's missing episodes take (#182, §6.2).
#[derive(Debug, Clone, Copy, Serialize, ToSchema)]
#[serde(rename_all = "snake_case")]
pub enum SeasonLane {
    SeasonPack,
    PerEpisode,
}

/// What holds a season off the pack lane, in the order that answers the
/// operator's question best — the most fundamental reason first (#182).
#[derive(Debug, Clone, Copy, Serialize, ToSchema)]
#[serde(rename_all = "snake_case")]
pub enum PackLaneReason {
    NoEpisodes,
    StillAiring,
    EpisodesOnDisk,
    PackBackoff,
}

/// Why the season release deck holds what it holds (#182).
///
/// An empty deck is three different truths, and until this endpoint existed
/// the deck could not tell them apart, so it blamed backoff for all three:
/// a pack sweep that has not landed yet, a pack sweep that landed and found
/// nothing, and a season on the per-episode lane, for which no pack sweep is
/// coming at all. The lane says which; `last_pack_search_at` separates the
/// first two, exactly as a movie's `last_searched_at` does (#177).
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct SeasonPackState {
    pub lane: SeasonLane,
    /// `None` when the season is on the pack lane.
    pub reason: Option<PackLaneReason>,
    /// Failed pack grabs so far. §6.2's backoff step is indexed by this, so
    /// it is also how long the lane stays quiet (#181).
    pub pack_failures: i64,
    /// When the pack lane reopens on its own, RFC3339. `None` unless a
    /// failure's window is currently open.
    pub pack_retry_at: Option<String>,
    /// When a pack sweep for this season last completed. `None` means no
    /// pack search has ever run, so an empty deck is pending, not settled.
    pub last_pack_search_at: Option<String>,
}

#[utoipa::path(
    get, path = "/api/series/{series_id}/seasons/{season_number}/pack-state", tag = "series",
    params(
        ("series_id" = i64, Path, description = "Series row id"),
        ("season_number" = i64, Path, description = "Season number, not its row id")
    ),
    responses(
        (status = 200, body = SeasonPackState),
        (status = 404, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn season_pack_state(
    State(state): State<AppState>,
    Path((series_id, number)): Path<(i64, i64)>,
) -> Result<Json<SeasonPackState>, ApiError> {
    load_series_row(&state, series_id).await?;
    let season_id = load_season_id(&state, series_id, number).await?;
    let pool = pool(&state)?;

    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 "has_file!: bool"
           FROM     episodes e
           WHERE    e.season_id = ?"#,
        season_id
    )
    .fetch_all(pool)
    .await?;
    let last_pack_search_at = sqlx::query_scalar!(
        "SELECT last_pack_search_at FROM seasons WHERE id = ?",
        season_id
    )
    .fetch_one(pool)
    .await?;
    // The same failed-pack tally the grab lane backs off on (#181), read
    // here so the deck can name the window instead of guessing at one.
    let failed = 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(pool)
    .await?;

    let window = chrono::TimeDelta::from_std(search_backoff(failed.failures)).unwrap_or_default();
    let reopens_at = if failed.failures > 0 {
        failed
            .last_failed_at
            .as_deref()
            .and_then(|at| DateTime::parse_from_rfc3339(at).ok())
            .map(|at| at.with_timezone(&Utc) + window)
    } else {
        None
    };
    let pack_retry_at = reopens_at.filter(|at| *at > Utc::now());

    let air_dates: Vec<Option<SystemTime>> = episodes
        .iter()
        .map(|episode| air_date(episode.air_date.as_deref()))
        .collect();
    let reason = season_grab_reason(&SeasonGrabFacts {
        air_dates: &air_dates,
        now: SystemTime::now(),
        any_episode_on_disk: episodes.iter().any(|episode| episode.has_file),
        pack_backoff_active: pack_retry_at.is_some(),
    });

    Ok(Json(SeasonPackState {
        lane: if reason.is_some() {
            SeasonLane::PerEpisode
        } else {
            SeasonLane::SeasonPack
        },
        reason: reason.map(|reason| match reason {
            PerEpisodeReason::NoEpisodes => PackLaneReason::NoEpisodes,
            PerEpisodeReason::StillAiring => PackLaneReason::StillAiring,
            PerEpisodeReason::EpisodesOnDisk => PackLaneReason::EpisodesOnDisk,
            PerEpisodeReason::PackBackoff => PackLaneReason::PackBackoff,
        }),
        pack_failures: failed.failures,
        pack_retry_at: pack_retry_at.map(|at| at.to_rfc3339()),
        last_pack_search_at,
    }))
}

#[utoipa::path(
    post, path = "/api/series/{series_id}/seasons/{season_number}/releases/{release_id}/grab", tag = "series",
    params(("series_id" = i64, Path), ("season_number" = i64, Path), ("release_id" = i64, Path)),
    responses(
        (status = 202, body = Accepted),
        (status = 404, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn grab_season_release(
    State(state): State<AppState>,
    Path((series_id, number, release_id)): Path<(i64, i64, i64)>,
) -> Result<(StatusCode, Json<Accepted>), ApiError> {
    load_series_row(&state, series_id).await?;
    let season_id = load_season_id(&state, series_id, number).await?;
    let exists = sqlx::query_scalar!("SELECT EXISTS(SELECT 1 FROM season_releases WHERE season_id = ? AND release_id = ?) AS 'exists!: bool'", season_id, release_id)
        .fetch_one(pool(&state)?)
        .await?;
    if !exists {
        return Err(ApiError::SeasonNotFound);
    }
    state
        .send_season_command(SeasonCommand::Grab {
            season_id,
            release_id,
        })
        .map_err(|_| ApiError::Unavailable)?;
    Ok((StatusCode::ACCEPTED, Json(Accepted { accepted: true })))
}

/// One imported episode file, keyed to its episode so the detail view can
/// attach the file's probed §7.4 attributes to the episode row.
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct EpisodeFile {
    pub episode_id: i64,
    pub path: String,
    pub size: i64,
    pub probed: Option<serde_json::Value>,
    /// The §5.7 rule relaxed to allow this import, when one was.
    pub waiver: Option<String>,
}

#[utoipa::path(
    get, path = "/api/series/{series_id}/files", tag = "series",
    params(("series_id" = i64, Path, description = "Series row id")),
    responses(
        (status = 200, body = [EpisodeFile]),
        (status = 404, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn files(
    State(state): State<AppState>,
    Path(id): Path<i64>,
) -> Result<Json<Vec<EpisodeFile>>, ApiError> {
    load_series_row(&state, id).await?;
    let files = sqlx::query_as!(EpisodeFile, r#"SELECT e.id AS "episode_id!: i64", mf.path AS "path!: String", mf.size AS "size!: i64", mf.probed AS "probed?: serde_json::Value", json_extract(mf.waiver, '$.rule') AS "waiver?: String" FROM media_files mf JOIN episodes e ON mf.owner_kind = 'episode' AND e.id = mf.owner_id JOIN seasons se ON se.id = e.season_id WHERE se.series_id = ? ORDER BY mf.path"#, id)
        .fetch_all(pool(&state)?)
        .await?;
    Ok(Json(files))
}

#[utoipa::path(
    get, path = "/api/series/{series_id}/owners", tag = "series",
    params(("series_id" = i64, Path, description = "Series row id")),
    responses(
        (status = 200, body = [Owner]),
        (status = 404, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn list_owners(
    State(state): State<AppState>,
    Path(id): Path<i64>,
) -> Result<Json<Vec<Owner>>, ApiError> {
    load_series_row(&state, id).await?;
    let owners = sqlx::query_as!(
        Owner,
        r#"SELECT o.id AS "id!: i64", o.name AS "name!: String", o.ntfy_topic AS "ntfy_topic!: String"
           FROM owners o
           JOIN title_owners t ON t.owner_id = o.id
           WHERE t.title_kind = 'series' AND t.title_id = ?
           ORDER BY o.name"#,
        id
    )
    .fetch_all(pool(&state)?)
    .await?;
    Ok(Json(owners))
}

#[utoipa::path(
    put, path = "/api/series/{series_id}/owners/{owner_id}", tag = "series",
    params(("series_id" = i64, Path), ("owner_id" = i64, Path)),
    responses(
        (status = 204),
        (status = 404, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn tag_owner(
    State(state): State<AppState>,
    Path((series_id, owner_id)): Path<(i64, i64)>,
) -> Result<StatusCode, ApiError> {
    load_series_row(&state, series_id).await?;
    let owner_exists = sqlx::query_scalar!(
        "SELECT EXISTS(SELECT 1 FROM owners WHERE id = ?) AS 'exists!: bool'",
        owner_id
    )
    .fetch_one(pool(&state)?)
    .await?;
    if !owner_exists {
        return Err(ApiError::OwnerNotFound);
    }
    sqlx::query!(
        "INSERT INTO title_owners (title_kind, title_id, owner_id) VALUES ('series', ?, ?) ON CONFLICT DO NOTHING",
        series_id, owner_id
    )
    .execute(pool(&state)?)
    .await?;
    Ok(StatusCode::NO_CONTENT)
}

#[utoipa::path(
    delete, path = "/api/series/{series_id}/owners/{owner_id}", tag = "series",
    params(("series_id" = i64, Path), ("owner_id" = i64, Path)),
    responses(
        (status = 204),
        (status = 404, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn untag_owner(
    State(state): State<AppState>,
    Path((series_id, owner_id)): Path<(i64, i64)>,
) -> Result<StatusCode, ApiError> {
    load_series_row(&state, series_id).await?;
    sqlx::query!(
        "DELETE FROM title_owners WHERE title_kind = 'series' AND title_id = ? AND owner_id = ?",
        series_id,
        owner_id
    )
    .execute(pool(&state)?)
    .await?;
    Ok(StatusCode::NO_CONTENT)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{router, Upstreams};

    async fn application() -> (tempfile::TempDir, AppState, String) {
        let dir = tempfile::tempdir().expect("tempdir");
        let database = arr_db::Db::connect(dir.path().join("arr.db"))
            .await
            .expect("connect database");
        database.migrate().await.expect("migrate database");
        let state = AppState::new(Upstreams::new(
            "http://127.0.0.1:1".into(),
            "http://127.0.0.1:1".into(),
        ))
        .expect("state")
        .with_database(database);
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind");
        let address = listener.local_addr().expect("address");
        let app = router(state.clone());
        tokio::spawn(async move { axum::serve(listener, app).await.expect("serve") });
        (dir, state, format!("http://{address}"))
    }

    async fn tv_root(state: &AppState, audience: &str) -> i64 {
        sqlx::query_scalar("SELECT id FROM roots WHERE kind = 'tv' AND audience = ?")
            .bind(audience)
            .fetch_one(state.database().expect("database").pool())
            .await
            .expect("TV root")
    }

    async fn add_series(base: &str, root_id: i64, auto_track: bool) -> serde_json::Value {
        let response = reqwest::Client::new()
            .post(format!("{base}/api/series"))
            .json(&serde_json::json!({
                "tmdb_id": 82_728, "title": "Bluey", "year": 2018,
                "original_language": "en", "root_id": root_id,
                "auto_track": auto_track
            }))
            .send()
            .await
            .expect("create series");
        assert_eq!(response.status(), StatusCode::CREATED);
        response.json().await.expect("series json")
    }

    async fn add_season(
        base: &str,
        series_id: i64,
        number: i64,
        episodes: serde_json::Value,
    ) -> serde_json::Value {
        let response = reqwest::Client::new()
            .post(format!("{base}/api/series/{series_id}/seasons"))
            .json(&serde_json::json!({ "number": number, "episodes": episodes }))
            .send()
            .await
            .expect("create season");
        assert_eq!(response.status(), StatusCode::CREATED);
        response.json().await.expect("season json")
    }

    /// §9.6: the three stored artwork fields come off the series detail
    /// response at add time — the same call that used to fetch only the
    /// TVDB id — so a series added today has a poster before tomorrow's
    /// refresh.
    #[tokio::test]
    async fn creating_a_series_stores_artwork_from_tmdb() {
        let tmdb = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/tv/82728"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": 82_728,
                    "name": "Bluey",
                    "status": "Returning Series",
                    "poster_path": "/bluey.jpg",
                    "backdrop_path": "/bluey-wide.jpg",
                    "vote_average": 8.417,
                    "external_ids": {"tvdb_id": 361_391}
                })),
            )
            .mount(&tmdb)
            .await;

        let dir = tempfile::tempdir().expect("tempdir");
        let database = arr_db::Db::connect(dir.path().join("arr.db"))
            .await
            .expect("connect database");
        database.migrate().await.expect("migrate database");
        let state = AppState::new(
            Upstreams::new("http://127.0.0.1:1".into(), "http://127.0.0.1:1".into())
                .with_tmdb_url(tmdb.uri())
                .with_tmdb_api_key(Some("key".into())),
        )
        .expect("state")
        .with_database(database);
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind");
        let address = listener.local_addr().expect("address");
        let served = state.clone();
        tokio::spawn(async move { axum::serve(listener, router(served)).await.expect("serve") });
        let base = format!("http://{address}");

        let response = reqwest::Client::new()
            .post(format!("{base}/api/series"))
            .json(&serde_json::json!({
                "tmdb_id": 82_728, "title": "Bluey",
                "original_language": "en", "root_id": 3
            }))
            .send()
            .await
            .expect("create series");
        assert_eq!(response.status(), StatusCode::CREATED);
        let series_id = response.json::<serde_json::Value>().await.expect("json")["id"]
            .as_i64()
            .expect("series id");

        let (tvdb_id, poster, backdrop, vote): (
            Option<i64>,
            Option<String>,
            Option<String>,
            Option<f64>,
        ) = sqlx::query_as(
            "SELECT tvdb_id, poster_path, backdrop_path, vote_average FROM series WHERE id = ?",
        )
        .bind(series_id)
        .fetch_one(state.database().expect("database").pool())
        .await
        .expect("series row");
        assert_eq!(tvdb_id, Some(361_391));
        assert_eq!(poster.as_deref(), Some("/bluey.jpg"));
        assert_eq!(backdrop.as_deref(), Some("/bluey-wide.jpg"));
        assert_eq!(vote, Some(8.417));
    }

    /// Issue #176: the daily metadata lane is what reveals a new series'
    /// seasons, so adding one queues a refresh instead of leaving the page
    /// empty for up to a day. TMDB here is a closed port: the add still
    /// returns 201 and the command is still queued, because the refresh is
    /// the lane's problem and not the request's.
    #[tokio::test]
    async fn adding_a_series_queues_a_refresh_even_with_tmdb_down() {
        let dir = tempfile::tempdir().expect("tempdir");
        let database = arr_db::Db::connect(dir.path().join("arr.db"))
            .await
            .expect("connect database");
        database.migrate().await.expect("migrate database");
        let state = AppState::new(
            Upstreams::new("http://127.0.0.1:1".into(), "http://127.0.0.1:1".into())
                .with_tmdb_url("http://127.0.0.1:1".into())
                .with_tmdb_api_key(Some("key".into())),
        )
        .expect("state")
        .with_database(database);
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind");
        let address = listener.local_addr().expect("address");
        let served = state.clone();
        tokio::spawn(async move { axum::serve(listener, router(served)).await.expect("serve") });
        let base = format!("http://{address}");

        let response = reqwest::Client::new()
            .post(format!("{base}/api/series"))
            .json(&serde_json::json!({
                "tmdb_id": 82_728, "title": "Bluey", "root_id": 3
            }))
            .send()
            .await
            .expect("create series");
        assert_eq!(response.status(), StatusCode::CREATED);
        let series_id = response.json::<serde_json::Value>().await.expect("json")["id"]
            .as_i64()
            .expect("series id");

        assert_eq!(
            state.next_metadata_command().await.expect("command"),
            MetadataCommand::Series { series_id }
        );
        let refreshed_at: Option<String> =
            sqlx::query_scalar("SELECT metadata_refreshed_at FROM series WHERE id = ?")
                .bind(series_id)
                .fetch_one(state.database().expect("database").pool())
                .await
                .expect("series row");
        assert!(
            refreshed_at.is_none(),
            "an unrefreshed series stays due for the scheduled sweep"
        );
    }

    /// Issue #177: the series response carries `metadata_refreshed_at` so
    /// the SPA can tell a never-refreshed series from a settled empty one.
    #[tokio::test]
    async fn series_response_exposes_metadata_refreshed_at() {
        let (_dir, state, base) = application().await;
        let root_id = tv_root(&state, "main").await;
        let series = add_series(&base, root_id, true).await;
        assert!(series["metadata_refreshed_at"].is_null());
    }

    /// Issue #177: the retry control a gave-up poll offers resends the same
    /// on-demand command `create` sends, so a second refresh attempt does
    /// not need the scheduled sweep to come around.
    #[tokio::test]
    async fn refreshing_metadata_by_hand_queues_the_same_command() {
        let (_dir, state, base) = application().await;
        let root_id = tv_root(&state, "main").await;
        let series = add_series(&base, root_id, true).await;
        let series_id = series["id"].as_i64().expect("series id");
        assert_eq!(
            state
                .next_metadata_command()
                .await
                .expect("create's command"),
            MetadataCommand::Series { series_id }
        );

        let response = reqwest::Client::new()
            .post(format!("{base}/api/series/{series_id}/refresh-metadata"))
            .send()
            .await
            .expect("refresh metadata");

        assert_eq!(response.status(), StatusCode::ACCEPTED);
        assert_eq!(
            state
                .next_metadata_command()
                .await
                .expect("retry's command"),
            MetadataCommand::Series { series_id }
        );
    }

    #[tokio::test]
    async fn refreshing_metadata_for_an_unknown_series_is_a_404() {
        let (_dir, _state, base) = application().await;

        let response = reqwest::Client::new()
            .post(format!("{base}/api/series/404/refresh-metadata"))
            .send()
            .await
            .expect("refresh metadata");

        assert_eq!(response.status(), StatusCode::NOT_FOUND);
    }

    #[tokio::test]
    async fn series_must_sit_on_a_tv_root() {
        let (_dir, state, base) = application().await;
        let movie_root: i64 =
            sqlx::query_scalar("SELECT id FROM roots WHERE kind = 'movie' LIMIT 1")
                .fetch_one(state.database().expect("database").pool())
                .await
                .expect("movie root");

        let response = reqwest::Client::new()
            .post(format!("{base}/api/series"))
            .json(&serde_json::json!({
                "tmdb_id": 82_728, "title": "Bluey", "root_id": movie_root
            }))
            .send()
            .await
            .expect("create series");
        assert_eq!(response.status(), StatusCode::UNPROCESSABLE_ENTITY);
    }

    /// #160. A season added before the series' first metadata refresh is part
    /// of the back catalogue at add time, so the rule does not track it. Once
    /// a refresh has happened, a hand-revealed season is genuinely new and
    /// arrives tracked with its episodes wanted.
    #[tokio::test]
    async fn auto_track_decides_whether_a_new_season_arrives_wanted() {
        let (_dir, state, base) = application().await;
        let root_id = tv_root(&state, "kids").await;

        let tracked = add_series(&base, root_id, true).await;
        let tracked_id = tracked["id"].as_i64().expect("id");
        // Seeding: no refresh has ever stamped this series.
        let season = add_season(
            &base,
            tracked_id,
            1,
            serde_json::json!([
                {"number": 1, "title": "The Magic Xylophone", "air_date": "2018-10-01"},
                {"number": 2, "title": "Hospital", "air_date": "2018-10-02"}
            ]),
        )
        .await;
        assert_eq!(season["tracked"], false);
        assert!(
            season["episodes"]
                .as_array()
                .expect("episodes")
                .iter()
                .all(|episode| episode["wanted"] == false),
            "§4.1: the back catalogue at add time is never auto-tracked"
        );

        sqlx::query(
            "UPDATE series SET metadata_refreshed_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
             WHERE id = ?",
        )
        .bind(tracked_id)
        .execute(state.database().expect("database").pool())
        .await
        .expect("stamp refreshed");
        let season = add_season(
            &base,
            tracked_id,
            2,
            serde_json::json!([
                {"number": 1, "title": "Dance Mode", "air_date": "2019-04-01"}
            ]),
        )
        .await;
        assert_eq!(season["tracked"], true);
        assert!(
            season["episodes"]
                .as_array()
                .expect("episodes")
                .iter()
                .all(|episode| episode["wanted"] == true),
            "§4.1: from the second refresh onward the rule tracks what is new"
        );

        let untracked_id = sqlx::query_scalar::<_, i64>(
            "INSERT INTO series (tmdb_id, title, root_id, auto_track) VALUES (1668, 'Friends', ?, 0) RETURNING id",
        )
        .bind(root_id)
        .fetch_one(state.database().expect("database").pool())
        .await
        .expect("untracked series");
        let season = add_season(
            &base,
            untracked_id,
            2,
            serde_json::json!([{"number": 1, "title": "The One", "air_date": "1995-09-21"}]),
        )
        .await;
        assert_eq!(season["tracked"], false);
        assert_eq!(season["episodes"][0]["wanted"], false);
    }

    /// #153. An empty episode title is rejected up front — the column and the
    /// TMDB boundary both refuse it, so the API must too.
    #[tokio::test]
    async fn an_empty_episode_title_is_rejected() {
        let (_dir, state, base) = application().await;
        let root_id = tv_root(&state, "main").await;
        let series = add_series(&base, root_id, false).await;
        let series_id = series["id"].as_i64().expect("id");

        for title in ["", "   "] {
            let response = reqwest::Client::new()
                .post(format!("{base}/api/series/{series_id}/seasons"))
                .json(&serde_json::json!({"number": 1, "episodes": [
                    {"number": 1, "title": title}
                ]}))
                .send()
                .await
                .expect("create season");
            assert_eq!(response.status(), StatusCode::UNPROCESSABLE_ENTITY);
        }
    }

    #[tokio::test]
    async fn a_rejected_season_leaves_nothing_behind_to_retry_over() {
        let (_dir, state, base) = application().await;
        let root_id = tv_root(&state, "main").await;
        let series = add_series(&base, root_id, false).await;
        let series_id = series["id"].as_i64().expect("id");

        let response = reqwest::Client::new()
            .post(format!("{base}/api/series/{series_id}/seasons"))
            .json(&serde_json::json!({"number": 1, "episodes": [
                {"number": 1, "title": "One"},
                {"number": 1, "title": "One again"}
            ]}))
            .send()
            .await
            .expect("create season");
        assert_eq!(response.status(), StatusCode::UNPROCESSABLE_ENTITY);

        let seasons: i64 = sqlx::query_scalar("SELECT count(*) FROM seasons")
            .fetch_one(state.database().expect("database").pool())
            .await
            .expect("count seasons");
        assert_eq!(seasons, 0, "the season number stays free for the retry");

        let season = add_season(
            &base,
            series_id,
            1,
            serde_json::json!([{"number": 1, "title": "One"}]),
        )
        .await;
        assert_eq!(season["number"], 1);
    }

    #[tokio::test]
    async fn a_vanished_season_reaches_the_seasons_endpoint() {
        let (_dir, state, base) = application().await;
        let root_id = tv_root(&state, "main").await;
        let series = add_series(&base, root_id, false).await;
        let series_id = series["id"].as_i64().expect("id");
        let clean = add_season(
            &base,
            series_id,
            1,
            serde_json::json!([{"number": 1, "title": "One"}]),
        )
        .await;
        assert_eq!(clean["vanished"], false);
        add_season(
            &base,
            series_id,
            2,
            serde_json::json!([{"number": 1, "title": "One"}]),
        )
        .await;

        sqlx::query("UPDATE seasons SET vanished = 1 WHERE series_id = ? AND number = 2")
            .bind(series_id)
            .execute(state.database().expect("database").pool())
            .await
            .expect("flag season vanished");

        let seasons: serde_json::Value = reqwest::Client::new()
            .get(format!("{base}/api/series/{series_id}/seasons"))
            .send()
            .await
            .expect("list seasons")
            .json()
            .await
            .expect("seasons json");
        let seasons = seasons.as_array().expect("seasons array");
        let vanished = |number: i64| {
            seasons
                .iter()
                .find(|season| season["number"] == number)
                .expect("season")["vanished"]
                .clone()
        };
        assert_eq!(vanished(1), false);
        assert_eq!(
            vanished(2),
            true,
            "#141: the conflict flag is not stopped at the database"
        );
    }

    /// §9.6: seasons come back newest-first within a series and episodes
    /// newest-first within each season, with season 0 wherever descending
    /// numeric order puts it — last. The SPA renders this order as given,
    /// so it is the API's to get right (#162).
    #[tokio::test]
    async fn seasons_and_episodes_are_returned_newest_first() {
        let (_dir, state, base) = application().await;
        let root_id = tv_root(&state, "main").await;
        let series = add_series(&base, root_id, false).await;
        let series_id = series["id"].as_i64().expect("id");
        for (season, episodes) in [
            (
                0,
                serde_json::json!([
                    {"number": 1, "title": "Special one"},
                    {"number": 2, "title": "Special two"}
                ]),
            ),
            (1, serde_json::json!([{"number": 1, "title": "Pilot"}])),
            (
                2,
                serde_json::json!([
                    {"number": 1, "title": "One"},
                    {"number": 2, "title": "Two"},
                    {"number": 3, "title": "Three"}
                ]),
            ),
        ] {
            add_season(&base, series_id, season, episodes).await;
        }

        let seasons: serde_json::Value =
            reqwest::get(format!("{base}/api/series/{series_id}/seasons"))
                .await
                .expect("list seasons")
                .json()
                .await
                .expect("seasons json");
        let seasons = seasons.as_array().expect("seasons array");
        let numbers: Vec<i64> = seasons
            .iter()
            .map(|season| season["number"].as_i64().expect("number"))
            .collect();
        assert_eq!(numbers, [2, 1, 0], "§9.6: newest-first within the series");

        let episode_numbers = |index: usize| -> Vec<i64> {
            seasons[index]["episodes"]
                .as_array()
                .expect("episodes")
                .iter()
                .map(|episode| episode["number"].as_i64().expect("episode number"))
                .collect()
        };
        assert_eq!(
            episode_numbers(0),
            [3, 2, 1],
            "§9.6: newest-first in a season"
        );
        assert_eq!(episode_numbers(1), [1]);
        assert_eq!(episode_numbers(2), [2, 1]);
    }

    #[tokio::test]
    async fn intent_is_set_on_seasons_and_on_single_episodes() {
        let (_dir, state, base) = application().await;
        let root_id = tv_root(&state, "main").await;
        let series = add_series(&base, root_id, false).await;
        let series_id = series["id"].as_i64().expect("id");
        let season = add_season(
            &base,
            series_id,
            2,
            serde_json::json!([
                {"number": 1, "title": "One", "air_date": "2020-01-01"},
                {"number": 2, "title": "Two", "air_date": "2020-01-08"}
            ]),
        )
        .await;
        assert_eq!(season["episodes"][0]["wanted"], false);

        // The whole season, in one click.
        let updated: serde_json::Value = reqwest::Client::new()
            .patch(format!("{base}/api/series/{series_id}/seasons/2"))
            .json(&serde_json::json!({"wanted": true}))
            .send()
            .await
            .expect("update season")
            .json()
            .await
            .expect("season json");
        assert!(updated["episodes"]
            .as_array()
            .expect("episodes")
            .iter()
            .all(|episode| episode["wanted"] == true));
        assert_eq!(
            updated["tracked"], false,
            "§4.1: wanting a season is not tracking the series"
        );

        // And one episode on its own.
        let episode_id = updated["episodes"][1]["id"].as_i64().expect("episode id");
        let episode: serde_json::Value = reqwest::Client::new()
            .patch(format!("{base}/api/episodes/{episode_id}"))
            .json(&serde_json::json!({"wanted": false}))
            .send()
            .await
            .expect("update episode")
            .json()
            .await
            .expect("episode json");
        assert_eq!(episode["wanted"], false);
    }

    /// #171. The §4.1 rule runs whenever the flag is sent: off clears
    /// wanted — including on a season that was already untracked, which is
    /// how rows stranded by the old ruling get out — and on marks every
    /// revealed episode wanted.
    #[tokio::test]
    async fn untracking_a_season_clears_its_episodes_wanted() {
        async fn patch_season(
            base: &str,
            series_id: i64,
            body: serde_json::Value,
        ) -> serde_json::Value {
            reqwest::Client::new()
                .patch(format!("{base}/api/series/{series_id}/seasons/1"))
                .json(&body)
                .send()
                .await
                .expect("update season")
                .json()
                .await
                .expect("season json")
        }

        let (_dir, state, base) = application().await;
        let root_id = tv_root(&state, "main").await;
        let series = add_series(&base, root_id, false).await;
        let series_id = series["id"].as_i64().expect("id");
        let season = add_season(
            &base,
            series_id,
            1,
            serde_json::json!([
                {"number": 1, "title": "One", "air_date": "2020-01-01"},
                {"number": 2, "title": "Two", "air_date": "2020-01-08"}
            ]),
        )
        .await;
        assert_eq!(season["tracked"], false);
        assert_eq!(season["episodes"][0]["wanted"], false);

        // A season stranded by the old ruling: untracked, yet its episodes
        // are all wanted. Re-sending the flag it already holds must still
        // do the work (#171).
        let updated = patch_season(&base, series_id, serde_json::json!({"wanted": true})).await;
        assert!(updated["episodes"]
            .as_array()
            .expect("episodes")
            .iter()
            .all(|episode| episode["wanted"] == true));
        let updated = patch_season(&base, series_id, serde_json::json!({"tracked": false})).await;
        assert_eq!(updated["tracked"], false);
        assert!(
            updated["episodes"]
                .as_array()
                .expect("episodes")
                .iter()
                .all(|episode| episode["wanted"] == false),
            "#171: clearing an already-untracked season is the way out"
        );

        let updated = patch_season(&base, series_id, serde_json::json!({"tracked": true})).await;
        assert_eq!(updated["tracked"], true);
        assert!(
            updated["episodes"]
                .as_array()
                .expect("episodes")
                .iter()
                .all(|episode| episode["wanted"] == true),
            "§4.1: turning tracked on marks every revealed episode wanted"
        );

        let updated = patch_season(&base, series_id, serde_json::json!({"tracked": false})).await;
        assert_eq!(updated["tracked"], false);
        assert!(
            updated["episodes"]
                .as_array()
                .expect("episodes")
                .iter()
                .all(|episode| episode["wanted"] == false),
            "§4.1: turning tracked off clears every revealed episode"
        );
    }

    /// #171. Untracking a season with no episodes touches nothing and is not
    /// an error.
    #[tokio::test]
    async fn untracking_an_empty_season_is_a_no_op() {
        let (_dir, state, base) = application().await;
        let root_id = tv_root(&state, "main").await;
        let series = add_series(&base, root_id, false).await;
        let series_id = series["id"].as_i64().expect("id");
        let season = add_season(&base, series_id, 1, serde_json::json!([])).await;
        assert_eq!(season["tracked"], false);

        let response = reqwest::Client::new()
            .patch(format!("{base}/api/series/{series_id}/seasons/1"))
            .json(&serde_json::json!({"tracked": false}))
            .send()
            .await
            .expect("update season");
        assert_eq!(response.status(), StatusCode::OK);
    }

    #[tokio::test]
    async fn listed_series_carry_a_derived_status() {
        let (_dir, state, base) = application().await;
        let root_id = tv_root(&state, "main").await;
        let series = add_series(&base, root_id, false).await;
        let series_id = series["id"].as_i64().expect("id");
        assert_eq!(
            series["status"], "complete",
            "nothing wanted is nothing missing"
        );

        let season = add_season(
            &base,
            series_id,
            1,
            serde_json::json!([{"number": 1, "title": "Pilot", "air_date": "2001-01-01"}]),
        )
        .await;
        let episode_id = season["episodes"][0]["id"].as_i64().expect("episode id");
        reqwest::Client::new()
            .patch(format!("{base}/api/episodes/{episode_id}"))
            .json(&serde_json::json!({"wanted": true}))
            .send()
            .await
            .expect("want the episode");

        let listed: Vec<serde_json::Value> = reqwest::get(format!("{base}/api/series"))
            .await
            .expect("list")
            .json()
            .await
            .expect("list json");
        assert_eq!(listed.len(), 1);
        assert_eq!(
            listed[0]["status"], "incomplete",
            "§4.2: an aired wanted episode with no file"
        );
        assert_eq!(listed[0]["wanted_episodes"], 1);
        assert_eq!(listed[0]["available_episodes"], 0);

        sqlx::query("UPDATE episodes SET state = 'available' WHERE id = ?")
            .bind(episode_id)
            .execute(state.database().expect("database").pool())
            .await
            .expect("import the episode");
        let after: serde_json::Value = reqwest::get(format!("{base}/api/series/{series_id}"))
            .await
            .expect("get")
            .json()
            .await
            .expect("series json");
        assert_eq!(after["status"], "complete");
    }

    /// #182: an empty season deck has to say which empty it is. The lane
    /// answers "no pack is coming"; `last_pack_search_at` answers "one ran
    /// and found nothing" — and until both existed the deck blamed backoff
    /// for every case, including seasons that were never on the pack lane.
    #[tokio::test]
    async fn the_pack_state_names_why_a_season_deck_is_empty() {
        let (_dir, state, base) = application().await;
        let root_id = tv_root(&state, "main").await;
        let series = add_series(&base, root_id, false).await;
        let series_id = series["id"].as_i64().expect("id");
        let season = add_season(
            &base,
            series_id,
            1,
            serde_json::json!([
                {"number": 1, "title": "Pilot", "air_date": "2001-01-01"},
                {"number": 2, "title": "Second", "air_date": "2001-01-08"}
            ]),
        )
        .await;
        let season_id = season["id"].as_i64().expect("season id");
        let episode_id = season["episodes"][0]["id"].as_i64().expect("episode id");
        let pool = state.database().expect("database").pool();

        let state_of = |url: String| async move {
            let response = reqwest::get(url).await.expect("pack state");
            assert_eq!(response.status(), StatusCode::OK);
            response
                .json::<serde_json::Value>()
                .await
                .expect("pack state json")
        };
        let url = format!("{base}/api/series/{series_id}/seasons/1/pack-state");

        let fresh = state_of(url.clone()).await;
        assert_eq!(fresh["lane"], "season_pack");
        assert_eq!(fresh["reason"], serde_json::Value::Null);
        assert_eq!(fresh["last_pack_search_at"], serde_json::Value::Null);
        assert_eq!(fresh["pack_failures"], 0);

        // A pack sweep that ran turns an empty deck from pending to settled.
        sqlx::query(
            "UPDATE seasons SET last_pack_search_at = '2026-01-01T00:00:00.000Z' WHERE id = ?",
        )
        .bind(season_id)
        .execute(pool)
        .await
        .expect("stamp the sweep");
        let swept = state_of(url.clone()).await;
        assert_eq!(swept["lane"], "season_pack");
        assert_eq!(swept["last_pack_search_at"], "2026-01-01T00:00:00.000Z");

        // §6.2/#181: a failed pack quiets the lane, and the deck says until
        // when rather than only that it is quiet.
        let release_id = sqlx::query_scalar::<_, i64>(
            "INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed, verdict)
             VALUES (7, 'pack', 'Bluey S01 1080p WEB-DL', 1000, 'url', '{}', 'eligible')
             RETURNING id",
        )
        .fetch_one(pool)
        .await
        .expect("release");
        sqlx::query(
            "INSERT INTO grabs (release_id, target_kind, target_id, infohash, state, grabbed_at)
             VALUES (?, 'season', ?, 'hash', 'failed', strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))",
        )
        .bind(release_id)
        .bind(season_id)
        .execute(pool)
        .await
        .expect("failed pack grab");
        let quiet = state_of(url.clone()).await;
        assert_eq!(quiet["lane"], "per_episode");
        assert_eq!(quiet["reason"], "pack_backoff");
        assert_eq!(quiet["pack_failures"], 1);
        assert!(
            quiet["pack_retry_at"].is_string(),
            "the deck offers a date, not just a closed door"
        );

        // §14 outranks it: a pack would re-import what is on disk, so
        // clearing the failure would not earn a pack anyway.
        sqlx::query("INSERT INTO media_files (owner_kind, owner_id, path, size) VALUES ('episode', ?, '/library/e01.mkv', 1)")
            .bind(episode_id)
            .execute(pool)
            .await
            .expect("file on disk");
        let on_disk = state_of(url.clone()).await;
        assert_eq!(on_disk["reason"], "episodes_on_disk");

        // An unaired episode outranks both.
        sqlx::query(
            "UPDATE episodes SET air_date = '2999-01-01' WHERE season_id = ? AND number = 2",
        )
        .bind(season_id)
        .execute(pool)
        .await
        .expect("unair the finale");
        let airing = state_of(url).await;
        assert_eq!(airing["reason"], "still_airing");
    }

    #[tokio::test]
    async fn manual_episode_actions_are_scoped_and_respect_blocked() {
        let (_dir, state, base) = application().await;
        let root_id = tv_root(&state, "main").await;
        let series = add_series(&base, root_id, false).await;
        let series_id = series["id"].as_i64().expect("id");
        let season = add_season(
            &base,
            series_id,
            1,
            serde_json::json!([{"number": 1, "title": "Pilot", "air_date": "2001-01-01"}]),
        )
        .await;
        let episode_id = season["episodes"][0]["id"].as_i64().expect("episode id");

        let response = reqwest::Client::new()
            .post(format!("{base}/api/episodes/{episode_id}/search"))
            .send()
            .await
            .expect("search");
        assert_eq!(response.status(), StatusCode::ACCEPTED);
        assert_eq!(
            state.next_episode_command().await.expect("command"),
            EpisodeCommand::Search { episode_id }
        );

        let pool = state.database().expect("database").pool();
        let release_id = sqlx::query_scalar::<_, i64>("INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed, score, verdict) VALUES (7, 'guid', 'Bluey S01E01 1080p WEB-DL', 1000, 'url', '{}', 42, 'eligible') RETURNING id")
            .fetch_one(pool)
            .await
            .expect("release");
        sqlx::query("INSERT INTO episode_releases (episode_id, release_id) VALUES (?, ?)")
            .bind(episode_id)
            .bind(release_id)
            .execute(pool)
            .await
            .expect("association");

        let releases: Vec<serde_json::Value> =
            reqwest::get(format!("{base}/api/episodes/{episode_id}/releases"))
                .await
                .expect("releases")
                .json()
                .await
                .expect("releases json");
        assert_eq!(releases[0]["verdict"], "eligible");

        let response = reqwest::Client::new()
            .post(format!(
                "{base}/api/episodes/{episode_id}/releases/{release_id}/grab"
            ))
            .send()
            .await
            .expect("grab");
        assert_eq!(response.status(), StatusCode::ACCEPTED);
        assert_eq!(
            state.next_episode_command().await.expect("command"),
            EpisodeCommand::Grab {
                episode_id,
                release_id
            }
        );

        // A release belonging to another episode is not grabbable through
        // this one.
        let other = sqlx::query_scalar::<_, i64>("INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed, verdict) VALUES (7, 'other', 'other', 1, 'url', '{}', 'eligible') RETURNING id")
            .fetch_one(pool)
            .await
            .expect("other release");
        let response = reqwest::Client::new()
            .post(format!(
                "{base}/api/episodes/{episode_id}/releases/{other}/grab"
            ))
            .send()
            .await
            .expect("grab");
        assert_eq!(response.status(), StatusCode::NOT_FOUND);

        reqwest::Client::new()
            .patch(format!("{base}/api/series/{series_id}"))
            .json(&serde_json::json!({"blocked": true}))
            .send()
            .await
            .expect("block the series");
        let response = reqwest::Client::new()
            .post(format!("{base}/api/episodes/{episode_id}/search"))
            .send()
            .await
            .expect("blocked search");
        assert_eq!(response.status(), StatusCode::CONFLICT);
    }

    /// Issue #125: seasons get the same §9.3 surface as movies and episodes —
    /// a targeted search, a deck scoped to one season and rescored against
    /// the current policy, and a grab by release id.
    #[tokio::test]
    async fn manual_season_actions_are_scoped_and_rescore_the_deck() {
        let (_dir, state, base) = application().await;
        let root_id = tv_root(&state, "main").await;
        let series = add_series(&base, root_id, false).await;
        let series_id = series["id"].as_i64().expect("id");
        let season = add_season(
            &base,
            series_id,
            1,
            serde_json::json!([{"number": 1, "title": "Pilot", "air_date": "2001-01-01"}]),
        )
        .await;

        let response = reqwest::Client::new()
            .post(format!("{base}/api/series/{series_id}/seasons/1/search"))
            .send()
            .await
            .expect("search");
        assert_eq!(response.status(), StatusCode::ACCEPTED);
        let season_id = season["id"].as_i64().expect("season id");
        assert_eq!(
            state.next_season_command().await.expect("command"),
            SeasonCommand::Search { season_id }
        );

        // A stale stored score proves the deck is rescored at read time,
        // like the movie deck (#114).
        let pool = state.database().expect("database").pool();
        let name = "Bluey S01 1080p WEB-DL x264-GROUP";
        let parsed = arr_parse::parse(name);
        let release_id = sqlx::query_scalar::<_, i64>(
            "INSERT INTO releases (indexer_id, guid, name, size, seeders, download_url, parsed, score, verdict)
             VALUES (7, 'pack', ?, 1000, 50, 'url', ?, -12345, 'eligible') RETURNING id",
        )
        .bind(name)
        .bind(serde_json::to_string(&parsed).expect("parsed json"))
        .fetch_one(pool)
        .await
        .expect("release");

        sqlx::query("INSERT INTO season_releases (season_id, release_id) VALUES (?, ?)")
            .bind(season_id)
            .bind(release_id)
            .execute(pool)
            .await
            .expect("association");

        let releases: Vec<serde_json::Value> =
            reqwest::get(format!("{base}/api/series/{series_id}/seasons/1/releases"))
                .await
                .expect("releases")
                .json()
                .await
                .expect("releases json");
        assert_eq!(releases.len(), 1);
        assert_eq!(releases[0]["verdict"], "eligible");
        assert_ne!(
            releases[0]["score"].as_f64(),
            Some(-12345.0),
            "the deck reflects the current policy, not the stored score"
        );

        let response = reqwest::Client::new()
            .post(format!(
                "{base}/api/series/{series_id}/seasons/1/releases/{release_id}/grab"
            ))
            .send()
            .await
            .expect("grab");
        assert_eq!(response.status(), StatusCode::ACCEPTED);
        assert_eq!(
            state.next_season_command().await.expect("command"),
            SeasonCommand::Grab {
                season_id,
                release_id
            }
        );

        // A release belonging to another season is not grabbable through
        // this one, and an unknown season number is not another season.
        let other = sqlx::query_scalar::<_, i64>("INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed, verdict) VALUES (7, 'other', 'other', 1, 'url', '{}', 'eligible') RETURNING id")
            .fetch_one(pool)
            .await
            .expect("other release");
        let response = reqwest::Client::new()
            .post(format!(
                "{base}/api/series/{series_id}/seasons/1/releases/{other}/grab"
            ))
            .send()
            .await
            .expect("grab");
        assert_eq!(response.status(), StatusCode::NOT_FOUND);
        let response = reqwest::get(format!("{base}/api/series/{series_id}/seasons/2/releases"))
            .await
            .expect("unknown season");
        assert_eq!(response.status(), StatusCode::NOT_FOUND);
    }

    #[tokio::test]
    async fn a_blocked_series_refuses_a_season_search() {
        let (_dir, state, base) = application().await;
        let root_id = tv_root(&state, "main").await;
        let series = add_series(&base, root_id, false).await;
        let series_id = series["id"].as_i64().expect("id");
        add_season(
            &base,
            series_id,
            1,
            serde_json::json!([{"number": 1, "title": "Pilot"}]),
        )
        .await;
        reqwest::Client::new()
            .patch(format!("{base}/api/series/{series_id}"))
            .json(&serde_json::json!({"blocked": true}))
            .send()
            .await
            .expect("block the series");

        let response = reqwest::Client::new()
            .post(format!("{base}/api/series/{series_id}/seasons/1/search"))
            .send()
            .await
            .expect("blocked search");
        assert_eq!(response.status(), StatusCode::CONFLICT);
    }

    #[tokio::test]
    async fn owner_tags_filter_the_series_list() {
        let (_dir, state, base) = application().await;
        let root_id = tv_root(&state, "kids").await;
        let series = add_series(&base, root_id, true).await;
        let series_id = series["id"].as_i64().expect("id");
        let owner_id = sqlx::query_scalar::<_, i64>(
            "INSERT INTO owners (name, ntfy_topic) VALUES ('kid', 'arr-kid') RETURNING id",
        )
        .fetch_one(state.database().expect("database").pool())
        .await
        .expect("owner");

        let empty: Vec<serde_json::Value> =
            reqwest::get(format!("{base}/api/series?owner_id={owner_id}"))
                .await
                .expect("filtered list")
                .json()
                .await
                .expect("json");
        assert!(empty.is_empty());

        let tagged = reqwest::Client::new()
            .put(format!("{base}/api/series/{series_id}/owners/{owner_id}"))
            .send()
            .await
            .expect("tag");
        assert_eq!(tagged.status(), StatusCode::NO_CONTENT);

        let filtered: Vec<serde_json::Value> =
            reqwest::get(format!("{base}/api/series?owner_id={owner_id}"))
                .await
                .expect("filtered list")
                .json()
                .await
                .expect("json");
        assert_eq!(filtered.len(), 1);
        assert_eq!(filtered[0]["id"], series_id);

        let owners: Vec<serde_json::Value> =
            reqwest::get(format!("{base}/api/series/{series_id}/owners"))
                .await
                .expect("owners")
                .json()
                .await
                .expect("json");
        assert_eq!(owners.len(), 1);

        reqwest::Client::new()
            .delete(format!("{base}/api/series/{series_id}/owners/{owner_id}"))
            .send()
            .await
            .expect("untag");
        let owners: Vec<serde_json::Value> =
            reqwest::get(format!("{base}/api/series/{series_id}/owners"))
                .await
                .expect("owners")
                .json()
                .await
                .expect("json");
        assert!(owners.is_empty());
    }

    #[tokio::test]
    async fn deleting_a_series_takes_its_seasons_and_episodes() {
        let (_dir, state, base) = application().await;
        let root_id = tv_root(&state, "main").await;
        let series = add_series(&base, root_id, true).await;
        let series_id = series["id"].as_i64().expect("id");
        add_season(
            &base,
            series_id,
            1,
            serde_json::json!([{"number": 1, "title": "Pilot", "air_date": "2001-01-01"}]),
        )
        .await;

        let response = reqwest::Client::new()
            .delete(format!("{base}/api/series/{series_id}"))
            .send()
            .await
            .expect("delete");
        assert_eq!(response.status(), StatusCode::NO_CONTENT);

        let episodes: i64 = sqlx::query_scalar("SELECT count(*) FROM episodes")
            .fetch_one(state.database().expect("database").pool())
            .await
            .expect("count episodes");
        assert_eq!(episodes, 0);
    }

    /// A series on disk for one test: the §7.4 title folder with a season
    /// subfolder holding one episode file and one sidecar subtitle.
    async fn library_on_disk(
        state: &AppState,
        episode_id: i64,
        root: &std::path::Path,
    ) -> std::path::PathBuf {
        let folder = root.join("Bluey (2018) [tmdbid-82728]");
        let season = folder.join("Season 01");
        tokio::fs::create_dir_all(&season)
            .await
            .expect("create title folder");
        let feature = season.join("Bluey (2018) - S01E01 - Pilot [1080p][WEB-DL].mkv");
        tokio::fs::write(&feature, b"episode").await.expect("write");
        tokio::fs::write(season.join("bluey.s01e01.pt.srt"), b"subs")
            .await
            .expect("write sidecar");

        let pool = state.database().expect("database").pool();
        let root_path = root.to_str().expect("utf-8 root");
        sqlx::query("UPDATE roots SET path = ? WHERE kind = 'tv' AND audience = 'main'")
            .bind(root_path)
            .execute(pool)
            .await
            .expect("point the root at the tempdir");
        sqlx::query(
            "INSERT INTO media_files (owner_kind, owner_id, path, size) VALUES ('episode', ?, ?, 7)",
        )
        .bind(episode_id)
        .bind(feature.to_str().expect("utf-8 path"))
        .execute(pool)
        .await
        .expect("media file");
        folder
    }

    /// The §7.4 title folder is the unit of deletion, so the season
    /// subfolder and sidecars go with it — and the root is never touched.
    #[tokio::test]
    async fn deleting_a_series_removes_the_whole_title_folder() {
        let (_dir, state, base) = application().await;
        let root_id = tv_root(&state, "main").await;
        let series = add_series(&base, root_id, true).await;
        let series_id = series["id"].as_i64().expect("id");
        let season = add_season(
            &base,
            series_id,
            1,
            serde_json::json!([{"number": 1, "title": "Pilot", "air_date": "2001-01-01"}]),
        )
        .await;
        let episode_id = season["episodes"][0]["id"].as_i64().expect("episode id");
        let root = tempfile::tempdir().expect("root");
        let folder = library_on_disk(&state, episode_id, root.path()).await;

        let response = reqwest::Client::new()
            .delete(format!("{base}/api/series/{series_id}"))
            .send()
            .await
            .expect("delete");
        assert_eq!(response.status(), StatusCode::NO_CONTENT);
        assert!(
            !folder.exists(),
            "the title folder, its seasons and its sidecars are gone"
        );
        assert!(root.path().exists(), "the root survives its titles");

        let pool = state.database().expect("database").pool();
        let orphans: i64 =
            sqlx::query_scalar("SELECT count(*) FROM media_files WHERE owner_kind = 'episode'")
                .fetch_one(pool)
                .await
                .expect("count files");
        assert_eq!(orphans, 0, "the file rows go with the files");
    }

    /// A missing series is 404 before anything touches the disk.
    #[tokio::test]
    async fn deleting_a_missing_series_is_a_404() {
        let (_dir, _state, base) = application().await;
        let response = reqwest::Client::new()
            .delete(format!("{base}/api/series/999"))
            .send()
            .await
            .expect("delete");
        assert_eq!(response.status(), StatusCode::NOT_FOUND);
    }

    /// The detail view reads its episode files through the series, keyed by
    /// episode id, so one request carries every §7.4 attribute tag it shows.
    #[tokio::test]
    async fn series_files_are_keyed_by_episode() {
        let (_dir, state, base) = application().await;
        let root_id = tv_root(&state, "kids").await;
        let series = add_series(&base, root_id, true).await;
        let series_id = series["id"].as_i64().expect("id");
        let season = add_season(
            &base,
            series_id,
            1,
            serde_json::json!([
                {"number": 1, "title": "The Magic Xylophone", "air_date": "2018-10-01"},
                {"number": 2, "title": "Hospital", "air_date": "2018-10-02"}
            ]),
        )
        .await;
        let episodes = season["episodes"].as_array().expect("episodes");
        let first = episodes[0]["id"].as_i64().expect("episode id");
        let pool = state.database().expect("database").pool();
        sqlx::query(
            r"INSERT INTO media_files (owner_kind, owner_id, path, size, probed, waiver)
               VALUES ('episode', ?, ?, 7, ?, ?)",
        )
        .bind(first)
        .bind("/mnt/media/tv/kids/Bluey (2018) [tmdbid-82728]/Season 01/Bluey S01E01.mkv")
        .bind(r#"{"resolution":"1080p","source":null,"hdr":"SDR","audio_tracks":[{"language":"pt-PT","title":null,"handler_name":null}],"sub_tracks":[]}"#)
        .bind(r#"{"rule":"required_audio"}"#)
        .execute(pool)
        .await
        .expect("media file");

        let response = reqwest::get(format!("{base}/api/series/{series_id}/files"))
            .await
            .expect("fetch files");
        assert_eq!(response.status(), StatusCode::OK);
        let files: serde_json::Value = response.json().await.expect("files json");
        let rows = files.as_array().expect("array");
        assert_eq!(rows.len(), 1, "only imported episodes carry a file");
        assert_eq!(rows[0]["episode_id"], first);
        assert_eq!(rows[0]["probed"]["resolution"], "1080p");
        assert_eq!(rows[0]["probed"]["audio_tracks"][0]["language"], "pt-PT");
        assert_eq!(rows[0]["waiver"], "required_audio");

        let missing = reqwest::get(format!("{base}/api/series/999/files"))
            .await
            .expect("fetch files");
        assert_eq!(missing.status(), StatusCode::NOT_FOUND);
    }

    /// The guard that keeps a delete inside the library: a path that is not
    /// under the series' root is left alone, whatever the row says.
    #[tokio::test]
    async fn a_series_file_outside_its_root_is_never_unlinked() {
        let (_dir, state, base) = application().await;
        let root_id = tv_root(&state, "main").await;
        let series = add_series(&base, root_id, true).await;
        let series_id = series["id"].as_i64().expect("id");
        let season = add_season(
            &base,
            series_id,
            1,
            serde_json::json!([{"number": 1, "title": "Pilot"}]),
        )
        .await;
        let episode_id = season["episodes"][0]["id"].as_i64().expect("episode id");

        let root = tempfile::tempdir().expect("root");
        sqlx::query("UPDATE roots SET path = ? WHERE id = ?")
            .bind(root.path().to_str().expect("utf-8 root"))
            .bind(root_id)
            .execute(state.database().expect("database").pool())
            .await
            .expect("point the root at the tempdir");

        let elsewhere = tempfile::tempdir().expect("elsewhere");
        let stray = elsewhere.path().join("not-ours.mkv");
        tokio::fs::write(&stray, b"stray").await.expect("write");
        sqlx::query(
            "INSERT INTO media_files (owner_kind, owner_id, path, size) VALUES ('episode', ?, ?, 5)",
        )
        .bind(episode_id)
        .bind(stray.to_str().expect("utf-8 path"))
        .execute(state.database().expect("database").pool())
        .await
        .expect("media file");

        let response = reqwest::Client::new()
            .delete(format!("{base}/api/series/{series_id}"))
            .send()
            .await
            .expect("delete");
        assert_eq!(response.status(), StatusCode::NO_CONTENT);
        assert!(
            stray.exists(),
            "a path outside the root is not ours to delete"
        );
    }

    /// Puts one episode file plus a sidecar into a §7.4 season folder and
    /// records the file. Returns the file's path.
    async fn episode_file_on_disk(
        state: &AppState,
        episode_id: i64,
        root: &std::path::Path,
        season: i64,
        name: &str,
    ) -> std::path::PathBuf {
        let folder = root
            .join("Bluey (2018) [tmdbid-82728]")
            .join(format!("Season {season:02}"));
        tokio::fs::create_dir_all(&folder)
            .await
            .expect("create season folder");
        let file = folder.join(name);
        tokio::fs::write(&file, b"episode").await.expect("write");

        sqlx::query(
            "INSERT INTO media_files (owner_kind, owner_id, path, size) VALUES ('episode', ?, ?, 7)",
        )
        .bind(episode_id)
        .bind(file.to_str().expect("utf-8 path"))
        .execute(state.database().expect("database").pool())
        .await
        .expect("media file");
        file
    }

    async fn point_root_at(state: &AppState, root_id: i64, path: &std::path::Path) {
        sqlx::query("UPDATE roots SET path = ? WHERE id = ?")
            .bind(path.to_str().expect("utf-8 root"))
            .bind(root_id)
            .execute(state.database().expect("database").pool())
            .await
            .expect("point the root at the tempdir");
    }

    async fn wanted_of(state: &AppState, episode_id: i64) -> bool {
        sqlx::query_scalar::<_, bool>("SELECT wanted FROM episodes WHERE id = ?")
            .bind(episode_id)
            .fetch_one(state.database().expect("database").pool())
            .await
            .expect("wanted")
    }

    /// #174: one call unlinks the season's files, drops their `media_files`
    /// rows and clears the intent behind them. The season and episode rows
    /// stay — TMDB owns that metadata — and the next season is untouched.
    #[tokio::test]
    async fn removing_a_season_takes_its_files_and_its_wanted() {
        let (_dir, state, base) = application().await;
        let root_id = tv_root(&state, "main").await;
        let series = add_series(&base, root_id, true).await;
        let series_id = series["id"].as_i64().expect("id");
        let root = tempfile::tempdir().expect("root");
        point_root_at(&state, root_id, root.path()).await;

        let first = add_season(
            &base,
            series_id,
            1,
            serde_json::json!([
                {"number": 1, "title": "The Magic Xylophone"},
                {"number": 2, "title": "Hospital"}
            ]),
        )
        .await;
        let second = add_season(
            &base,
            series_id,
            2,
            serde_json::json!([{"number": 1, "title": "Dance Mode"}]),
        )
        .await;
        let s01e01 = first["episodes"]
            .as_array()
            .expect("episodes")
            .iter()
            .find(|episode| episode["number"] == 1)
            .expect("s01e01")["id"]
            .as_i64()
            .expect("id");
        let s01e02 = first["episodes"]
            .as_array()
            .expect("episodes")
            .iter()
            .find(|episode| episode["number"] == 2)
            .expect("s01e02")["id"]
            .as_i64()
            .expect("id");
        let s02e01 = second["episodes"][0]["id"].as_i64().expect("id");

        let one = episode_file_on_disk(&state, s01e01, root.path(), 1, "Bluey - S01E01.mkv").await;
        let two = episode_file_on_disk(&state, s01e02, root.path(), 1, "Bluey - S01E02.mkv").await;
        let other =
            episode_file_on_disk(&state, s02e01, root.path(), 2, "Bluey - S02E01.mkv").await;

        let pool = state.database().expect("database").pool();
        for episode in [s01e01, s01e02, s02e01] {
            sqlx::query("UPDATE episodes SET wanted = 1, state = 'available' WHERE id = ?")
                .bind(episode)
                .execute(pool)
                .await
                .expect("seed wanted");
        }

        let response = reqwest::Client::new()
            .delete(format!("{base}/api/series/{series_id}/seasons/1/files"))
            .send()
            .await
            .expect("delete season files");
        assert_eq!(response.status(), StatusCode::NO_CONTENT);

        assert!(!one.exists(), "the season's files are gone from disk");
        assert!(!two.exists(), "the season's files are gone from disk");
        assert!(other.exists(), "another season's file is not in scope");

        let rows: Vec<String> =
            sqlx::query_scalar("SELECT path FROM media_files WHERE owner_kind = 'episode'")
                .fetch_all(pool)
                .await
                .expect("files");
        assert_eq!(
            rows,
            vec![other.to_str().expect("utf-8").to_string()],
            "only the season's rows go"
        );

        assert!(!wanted_of(&state, s01e01).await, "intent cleared");
        assert!(!wanted_of(&state, s01e02).await, "across the whole season");
        assert!(wanted_of(&state, s02e01).await, "and nowhere else");

        let state_of: String = sqlx::query_scalar("SELECT state FROM episodes WHERE id = ?")
            .bind(s01e01)
            .fetch_one(pool)
            .await
            .expect("state");
        assert_eq!(
            state_of, "missing",
            "an episode with no file is not available"
        );

        let seasons: i64 = sqlx::query_scalar("SELECT count(*) FROM seasons WHERE series_id = ?")
            .bind(series_id)
            .fetch_one(pool)
            .await
            .expect("seasons");
        let episodes: i64 = sqlx::query_scalar(
            "SELECT count(*) FROM episodes e JOIN seasons se ON se.id = e.season_id
             WHERE se.series_id = ?",
        )
        .bind(series_id)
        .fetch_one(pool)
        .await
        .expect("episodes");
        assert_eq!(seasons, 2, "TMDB owns the season rows, so they stay");
        assert_eq!(episodes, 3, "and the episode rows with them");
    }

    /// #174: the narrow scope really is narrow. Removing one episode leaves
    /// its sibling's file, row and intent exactly as they were, and leaves
    /// the season folder standing.
    #[tokio::test]
    async fn removing_one_episode_leaves_its_siblings_alone() {
        let (_dir, state, base) = application().await;
        let root_id = tv_root(&state, "main").await;
        let series = add_series(&base, root_id, true).await;
        let series_id = series["id"].as_i64().expect("id");
        let root = tempfile::tempdir().expect("root");
        point_root_at(&state, root_id, root.path()).await;

        let season = add_season(
            &base,
            series_id,
            1,
            serde_json::json!([
                {"number": 1, "title": "The Magic Xylophone"},
                {"number": 2, "title": "Hospital"}
            ]),
        )
        .await;
        let episodes = season["episodes"].as_array().expect("episodes");
        let first = episodes
            .iter()
            .find(|episode| episode["number"] == 1)
            .expect("s01e01")["id"]
            .as_i64()
            .expect("id");
        let second = episodes
            .iter()
            .find(|episode| episode["number"] == 2)
            .expect("s01e02")["id"]
            .as_i64()
            .expect("id");

        let one = episode_file_on_disk(&state, first, root.path(), 1, "Bluey - S01E01.mkv").await;
        let two = episode_file_on_disk(&state, second, root.path(), 1, "Bluey - S01E02.mkv").await;
        let pool = state.database().expect("database").pool();
        for episode in [first, second] {
            sqlx::query("UPDATE episodes SET wanted = 1 WHERE id = ?")
                .bind(episode)
                .execute(pool)
                .await
                .expect("seed wanted");
        }

        let response = reqwest::Client::new()
            .delete(format!("{base}/api/episodes/{first}/files"))
            .send()
            .await
            .expect("delete episode files");
        assert_eq!(response.status(), StatusCode::NO_CONTENT);

        assert!(!one.exists(), "the episode's file is gone");
        assert!(two.exists(), "its sibling's file is not in scope");
        assert!(
            two.parent().expect("season folder").exists(),
            "and neither is the season folder around them"
        );
        assert!(!wanted_of(&state, first).await, "intent cleared for it");
        assert!(wanted_of(&state, second).await, "and not for its sibling");

        let remaining: Vec<i64> =
            sqlx::query_scalar("SELECT owner_id FROM media_files WHERE owner_kind = 'episode'")
                .fetch_all(pool)
                .await
                .expect("files");
        assert_eq!(remaining, vec![second], "only the episode's row goes");
    }

    /// #174: removal is not conditional on there being anything to remove.
    /// A scope with no files still clears intent, and still answers 204.
    #[tokio::test]
    async fn removing_a_scope_with_no_files_still_clears_intent() {
        let (_dir, state, base) = application().await;
        let root_id = tv_root(&state, "main").await;
        let series = add_series(&base, root_id, true).await;
        let series_id = series["id"].as_i64().expect("id");
        let root = tempfile::tempdir().expect("root");
        point_root_at(&state, root_id, root.path()).await;

        let season = add_season(
            &base,
            series_id,
            1,
            serde_json::json!([{"number": 1, "title": "The Magic Xylophone"}]),
        )
        .await;
        let episode_id = season["episodes"][0]["id"].as_i64().expect("id");
        sqlx::query("UPDATE episodes SET wanted = 1 WHERE id = ?")
            .bind(episode_id)
            .execute(state.database().expect("database").pool())
            .await
            .expect("seed wanted");

        let response = reqwest::Client::new()
            .delete(format!("{base}/api/series/{series_id}/seasons/1/files"))
            .send()
            .await
            .expect("delete season files");
        assert_eq!(response.status(), StatusCode::NO_CONTENT);
        assert!(!wanted_of(&state, episode_id).await, "intent still cleared");

        // And again on the episode, which now has neither file nor intent.
        let response = reqwest::Client::new()
            .delete(format!("{base}/api/episodes/{episode_id}/files"))
            .send()
            .await
            .expect("delete episode files");
        assert_eq!(response.status(), StatusCode::NO_CONTENT);
    }

    /// #174: an unknown scope is 404, and an unknown series is 404 even for
    /// a season number that exists under some other series.
    #[tokio::test]
    async fn removing_files_from_an_unknown_scope_is_a_404() {
        let (_dir, state, base) = application().await;
        let root_id = tv_root(&state, "main").await;
        let series = add_series(&base, root_id, true).await;
        let series_id = series["id"].as_i64().expect("id");
        add_season(
            &base,
            series_id,
            1,
            serde_json::json!([{"number": 1, "title": "The Magic Xylophone"}]),
        )
        .await;

        let client = reqwest::Client::new();
        for path in [
            format!("api/series/{series_id}/seasons/9/files"),
            "api/series/999/seasons/1/files".to_string(),
            "api/episodes/999/files".to_string(),
        ] {
            let response = client
                .delete(format!("{base}/{path}"))
                .send()
                .await
                .expect("delete");
            assert_eq!(response.status(), StatusCode::NOT_FOUND, "{path}");
        }
    }

    /// #174 reuses the §7.4 containment guard: a recorded path that is not
    /// under the series' root is left on disk, whatever the row says. Its
    /// row still goes, so the library stops claiming the file.
    #[tokio::test]
    async fn an_episode_file_outside_its_root_is_never_unlinked() {
        let (_dir, state, base) = application().await;
        let root_id = tv_root(&state, "main").await;
        let series = add_series(&base, root_id, true).await;
        let series_id = series["id"].as_i64().expect("id");
        let root = tempfile::tempdir().expect("root");
        point_root_at(&state, root_id, root.path()).await;

        let season = add_season(
            &base,
            series_id,
            1,
            serde_json::json!([{"number": 1, "title": "The Magic Xylophone"}]),
        )
        .await;
        let episode_id = season["episodes"][0]["id"].as_i64().expect("id");

        let elsewhere = tempfile::tempdir().expect("elsewhere");
        let stray = elsewhere.path().join("not-ours.mkv");
        tokio::fs::write(&stray, b"stray").await.expect("write");
        sqlx::query(
            "INSERT INTO media_files (owner_kind, owner_id, path, size) VALUES ('episode', ?, ?, 5)",
        )
        .bind(episode_id)
        .bind(stray.to_str().expect("utf-8 path"))
        .execute(state.database().expect("database").pool())
        .await
        .expect("media file");

        let response = reqwest::Client::new()
            .delete(format!("{base}/api/episodes/{episode_id}/files"))
            .send()
            .await
            .expect("delete episode files");
        assert_eq!(response.status(), StatusCode::NO_CONTENT);
        assert!(
            stray.exists(),
            "a path outside the root is not ours to delete"
        );
    }

    /// The sub-series guard keeps the whole relative path, so it can only
    /// name the recorded file — and one `..` anywhere is enough to refuse.
    #[test]
    fn a_contained_file_is_the_recorded_path_under_the_root() {
        let root = "/mnt/media/tv/main";
        assert_eq!(
            contained_file(
                root,
                "/mnt/media/tv/main/Bluey (2018) [tmdbid-82728]/Season 01/Bluey S01E01.mkv"
            ),
            Some(std::path::PathBuf::from(
                "/mnt/media/tv/main/Bluey (2018) [tmdbid-82728]/Season 01/Bluey S01E01.mkv"
            ))
        );
        assert_eq!(contained_file(root, "/mnt/media/tv/kids/other.mkv"), None);
        assert_eq!(contained_file(root, root), None);
        assert_eq!(
            contained_file(root, "/mnt/media/tv/main/../kids/other.mkv"),
            None,
            "one `..` climbs back out of the root it was under"
        );
    }

    #[test]
    fn air_dates_parse_as_dates_and_as_timestamps() {
        assert_eq!(
            air_date(Some("1970-01-02")),
            Some(UNIX_EPOCH + Duration::from_hours(24))
        );
        assert_eq!(
            air_date(Some("1970-01-02T00:00:00Z")),
            Some(UNIX_EPOCH + Duration::from_hours(24))
        );
        assert_eq!(air_date(Some("not a date")), None);
        assert_eq!(air_date(None), None);
    }
}