Merge #121: daily series metadata refresh

Closes #121

crates/arr-daemon/src/grab.rs

@@ -1121,7 +1121,7 @@ pub(crate) fn backoff_elapsed(search_attempts: i64, last_searched_at: Option<&st
     last_searched_at.with_timezone(&chrono::Utc) + backoff <= chrono::Utc::now()
 }
 
-fn metadata_refresh_due(metadata_refreshed_at: Option<&str>) -> bool {
+pub(crate) fn metadata_refresh_due(metadata_refreshed_at: Option<&str>) -> bool {
     let Some(refreshed_at) = metadata_refreshed_at else {
         return true;
     };

crates/arr-daemon/src/main.rs

@@ -12,6 +12,7 @@ mod notify;
 mod reaper;
 pub mod reconcile;
 mod rss;
+mod series_refresh;
 mod tv_grab;
 mod web;
 
@@ -31,6 +32,7 @@ use notify::Notifier;
 use reaper::ReaperAction;
 use reconcile::{ReconcileLoop, Tick};
 use rss::RssAction;
+use series_refresh::SeriesRefreshAction;
 use tower_http::trace::TraceLayer;
 use tv_grab::TvGrabAction;
 
@@ -268,6 +270,13 @@ fn reconcile_loop(
             ),
         );
     }
+    // §8: metadata refresh is its own daily lane, staggered against the
+    // other ticks. Without TMDB there is nothing to refresh from.
+    if let Some(tmdb) = tmdb {
+        reconcile = reconcile.register(Tick::Metadata, SeriesRefreshAction::new(Arc::clone(tmdb)));
+    } else {
+        tracing::warn!("TMDB is not configured: series metadata refresh is disabled");
+    }
     let jellyfin = jellyfin::JellyfinClient::new(
         config.jellyfin_url.clone(),
         config.jellyfin_api_key.clone(),

crates/arr-daemon/src/series_refresh.rs

@@ -0,0 +1,731 @@
+//! The daily series metadata refresh. See DESIGN.md §8 and issue #121.
+//!
+//! The database rows are the work list (§8): a series whose
+//! `metadata_refreshed_at` has passed its time-to-live is due. One refresh
+//! pulls the series detail from TMDB and closes every gap it reveals:
+//!
+//! - seasons TMDB knows that the library does not are inserted, with
+//!   `apply_auto_track` deciding their tracking rule;
+//! - episodes revealed into an existing season arrive wanted when that season
+//!   is `tracked` (`apply_tracked`, §4.1);
+//! - air dates and titles of known episodes follow TMDB, because both §4.2's
+//!   `airing` window and §6.2's do-not-search-before-it-exists gate read them;
+//! - `upstream_ended` follows TMDB's status, which `ended` is derived from;
+//! - a missing `tvdb_id` is backfilled (#120).
+//!
+//! Episodes that vanish or renumber upstream are left alone (#122). Refresh
+//! is idempotent: over unchanged TMDB data only the stamp moves.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use arr_core::tracking::{apply_auto_track, apply_tracked, RefreshedSeason};
+use arr_core::{
+    Episode as CoreEpisode, EpisodeId, Language, MediaState, RootId, SeasonId, SeriesId,
+    TitleOverrides,
+};
+use arr_db::policy::language;
+use arr_db::Db;
+use arr_meta::{Season as TmdbSeasonDetail, Series as TmdbSeries, TmdbClient};
+
+use crate::grab::metadata_refresh_due;
+use crate::reconcile::{Action, ActionFuture, Outcome};
+
+/// TMDB statuses that mean no further episodes will ever appear.
+fn is_upstream_ended(status: &str) -> bool {
+    matches!(status, "Ended" | "Canceled")
+}
+
+#[derive(Debug, thiserror::Error)]
+enum RefreshError {
+    #[error("database: {0}")]
+    Database(#[from] sqlx::Error),
+    #[error("tmdb: {0}")]
+    Tmdb(#[from] arr_meta::Error),
+    #[error("series {0}: tmdb_id does not fit")]
+    InvalidTmdbId(i64),
+}
+
+/// Refreshes one series' metadata from TMDB per day (§8), on the same TTL
+/// gate as the movie lane.
+#[derive(Debug)]
+pub struct SeriesRefreshAction {
+    tmdb: Arc<TmdbClient>,
+}
+
+impl SeriesRefreshAction {
+    #[must_use]
+    pub fn new(tmdb: Arc<TmdbClient>) -> Self {
+        Self { tmdb }
+    }
+
+    async fn tick(&self, database: &Db) -> Result<Vec<Outcome>, RefreshError> {
+        let due = sqlx::query_as!(
+            DueSeries,
+            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",
+                      upstream_ended AS "upstream_ended!: bool", metadata_refreshed_at
+               FROM series
+               ORDER BY metadata_refreshed_at IS NOT NULL, metadata_refreshed_at, id"#
+        )
+        .fetch_all(database.pool())
+        .await?;
+        let mut outcomes = Vec::new();
+        for stale in due {
+            if !metadata_refresh_due(stale.metadata_refreshed_at.as_deref()) {
+                continue;
+            }
+            match self.refresh_series(database, &stale).await {
+                Ok(Some(outcome)) => outcomes.push(outcome),
+                Ok(None) => {}
+                // One series' failure must not cost the rest of the tick.
+                Err(error) => tracing::error!(
+                    series_id = stale.id,
+                    title = stale.title,
+                    %error,
+                    "series metadata refresh failed"
+                ),
+            }
+        }
+        Ok(outcomes)
+    }
+
+    async fn refresh_series(
+        &self,
+        database: &Db,
+        stale: &DueSeries,
+    ) -> Result<Option<Outcome>, RefreshError> {
+        let tmdb_id =
+            u32::try_from(stale.tmdb_id).map_err(|_| RefreshError::InvalidTmdbId(stale.id))?;
+        let metadata = self.tmdb.series(tmdb_id).await?;
+
+        let mut transaction = database.pool().begin().await?;
+        let mut changed = self
+            .sync_series_fields(&mut transaction, stale, &metadata)
+            .await?;
+
+        // Existing seasons keep their tracking rule; their episode lists are
+        // brought up to date in place. Seasons TMDB knows and the library
+        // does not go through the one function that owns the reveal rule.
+        let existing = sqlx::query!(
+            r#"SELECT id AS "id!: i64", number AS "number!: i64", tracked AS "tracked!: bool"
+               FROM seasons WHERE series_id = ?"#,
+            stale.id
+        )
+        .fetch_all(&mut *transaction)
+        .await?
+        .into_iter()
+        .map(|season| (season.number, (season.id, season.tracked)))
+        .collect::<HashMap<_, _>>();
+        let mut revealed = Vec::new();
+        for summary in &metadata.seasons {
+            if let Some(&(season_id, tracked)) = existing.get(&i64::from(summary.number)) {
+                changed |= self
+                    .sync_season(
+                        &mut transaction,
+                        season_id,
+                        tracked,
+                        tmdb_id,
+                        summary.number,
+                    )
+                    .await?;
+            } else {
+                let detail = self.tmdb.season(tmdb_id, summary.number).await?;
+                revealed.push(RefreshedSeason {
+                    season: arr_core::Season {
+                        id: SeasonId(0),
+                        series_id: SeriesId(stale.id),
+                        number: season_number(summary.number),
+                        tracked: false,
+                    },
+                    episodes: core_episodes(stale.id, &detail),
+                    is_new: true,
+                });
+            }
+        }
+        apply_auto_track(&core_series(stale), &mut revealed);
+        for refreshed in revealed {
+            self.insert_revealed_season(&mut transaction, stale.id, refreshed)
+                .await?;
+            changed = true;
+        }
+
+        transaction.commit().await?;
+        stamp_refreshed(database, stale.id).await?;
+        Ok(changed.then(|| {
+            Outcome::new(
+                format!("series {} ({})", stale.id, stale.title),
+                "refreshed metadata from TMDB",
+            )
+        }))
+    }
+
+    /// The series-level fields a refresh owns. Both writes are guarded so
+    /// unchanged data moves nothing.
+    async fn sync_series_fields(
+        &self,
+        executor: &mut sqlx::SqliteConnection,
+        stale: &DueSeries,
+        metadata: &TmdbSeries,
+    ) -> Result<bool, RefreshError> {
+        let mut changed = false;
+        // §6.1: `t=tvsearch` prefers the TVDB id. Series added before the id
+        // existed on add (#120) pick theirs up here.
+        if stale.tvdb_id.is_none() {
+            if let Some(tvdb_id) = metadata.tvdb_id.map(i64::from) {
+                sqlx::query!(
+                    "UPDATE series SET tvdb_id = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') WHERE id = ?",
+                    tvdb_id,
+                    stale.id
+                )
+                .execute(&mut *executor)
+                .await?;
+                changed = true;
+            }
+        }
+        let ended = is_upstream_ended(&metadata.status);
+        if ended != stale.upstream_ended {
+            sqlx::query!(
+                "UPDATE series SET upstream_ended = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') WHERE id = ?",
+                ended,
+                stale.id
+            )
+            .execute(&mut *executor)
+            .await?;
+            changed = true;
+        }
+        Ok(changed)
+    }
+
+    /// Persist one season the library has not seen, with `apply_auto_track`'s
+    /// verdict on its tracking rule already applied.
+    async fn insert_revealed_season(
+        &self,
+        executor: &mut sqlx::SqliteConnection,
+        series_id: i64,
+        refreshed: RefreshedSeason,
+    ) -> Result<(), RefreshError> {
+        let season_number = i64::from(refreshed.season.number);
+        let season_id = sqlx::query_scalar!(
+            "INSERT INTO seasons (series_id, number, tracked) VALUES (?, ?, ?) RETURNING id",
+            series_id,
+            season_number,
+            refreshed.season.tracked
+        )
+        .fetch_one(&mut *executor)
+        .await?;
+        for episode in &refreshed.episodes {
+            let episode_number = i64::from(episode.number);
+            let episode_title = episode.title.clone();
+            let air_date = air_date_string(episode.air_date);
+            sqlx::query!(
+                "INSERT INTO episodes (season_id, number, title, air_date, wanted) VALUES (?, ?, ?, ?, ?)",
+                season_id,
+                episode_number,
+                episode_title,
+                air_date,
+                episode.wanted
+            )
+            .execute(&mut *executor)
+            .await?;
+        }
+        Ok(())
+    }
+
+    /// Bring one persisted season's episodes up to date with TMDB's.
+    ///
+    /// Known episodes get guarded updates — nothing writes unless a value
+    /// actually moved — and unknown numbers are revealed wanted exactly when
+    /// the season is tracked (§4.1). Numbers absent from TMDB stay untouched.
+    async fn sync_season(
+        &self,
+        executor: &mut sqlx::SqliteConnection,
+        season_id: i64,
+        tracked: bool,
+        tmdb_id: u32,
+        number: u32,
+    ) -> Result<bool, RefreshError> {
+        let detail = self.tmdb.season(tmdb_id, number).await?;
+        let known = sqlx::query!(
+            r#"SELECT id AS "id!: i64", number AS "number!: i64",
+                      title AS "title!: String", air_date
+               FROM episodes WHERE season_id = ?"#,
+            season_id
+        )
+        .fetch_all(&mut *executor)
+        .await?
+        .into_iter()
+        .map(|episode| {
+            (
+                episode.number,
+                (episode.id, episode.title, episode.air_date),
+            )
+        })
+        .collect::<HashMap<_, _>>();
+
+        let mut fresh = Vec::new();
+        let mut changed = false;
+        for source in &detail.episodes {
+            let number = i64::from(source.number);
+            if let Some(&(id, ref title, ref air_date)) = known.get(&number) {
+                let new_air_date = source.air_date.map(|date| date.to_string());
+                if *title != source.title || *air_date != new_air_date {
+                    sqlx::query!(
+                        "UPDATE episodes SET title = ?, air_date = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') WHERE id = ? AND (title IS NOT ? OR air_date IS NOT ?)",
+                        source.title,
+                        new_air_date,
+                        id,
+                        source.title,
+                        new_air_date
+                    )
+                    .execute(&mut *executor)
+                    .await?;
+                    changed = true;
+                }
+            } else {
+                fresh.push(CoreEpisode {
+                    id: EpisodeId(0),
+                    season_id: SeasonId(season_id),
+                    number: season_number(source.number),
+                    title: source.title.clone(),
+                    air_date: source.air_date.map(system_time),
+                    wanted: false,
+                    state: MediaState::Missing,
+                    search_attempts: 0,
+                    last_searched_at: None,
+                });
+            }
+        }
+        apply_tracked(tracked, &mut fresh);
+        for episode in fresh {
+            let episode_number = i64::from(episode.number);
+            let air_date = air_date_string(episode.air_date);
+            sqlx::query!(
+                "INSERT INTO episodes (season_id, number, title, air_date, wanted) VALUES (?, ?, ?, ?, ?)",
+                season_id,
+                episode_number,
+                episode.title,
+                air_date,
+                episode.wanted
+            )
+            .execute(&mut *executor)
+            .await?;
+            changed = true;
+        }
+        Ok(changed)
+    }
+}
+
+impl Action for SeriesRefreshAction {
+    fn name(&self) -> &'static str {
+        "series-metadata-refresh"
+    }
+
+    fn run<'a>(&'a self, database: &'a Db) -> ActionFuture<'a> {
+        Box::pin(async move { self.tick(database).await.map_err(Into::into) })
+    }
+}
+
+struct DueSeries {
+    id: i64,
+    tmdb_id: i64,
+    tvdb_id: Option<i64>,
+    title: String,
+    year: Option<i64>,
+    original_language: Option<String>,
+    root_id: i64,
+    auto_track: bool,
+    upstream_ended: bool,
+    metadata_refreshed_at: Option<String>,
+}
+
+/// TMDB numbers are unbounded; ours are `u16` (`CHECK (number >= 0)`,
+/// STRICT). A number past `u16::MAX` cannot match anything real and would
+/// never be grabbed anyway.
+fn season_number(number: u32) -> u16 {
+    u16::try_from(number.min(u32::from(u16::MAX))).unwrap_or(u16::MAX)
+}
+
+fn core_series(series: &DueSeries) -> arr_core::Series {
+    arr_core::Series {
+        id: SeriesId(series.id),
+        tmdb_id: u64::try_from(series.tmdb_id).unwrap_or_default(),
+        title: series.title.clone(),
+        year: series
+            .year
+            .and_then(|year| u16::try_from(year).ok())
+            .unwrap_or_default(),
+        original_language: series
+            .original_language
+            .as_deref()
+            .map_or(Language::Other(String::new()), language),
+        root_id: RootId(series.root_id),
+        auto_track: series.auto_track,
+        overrides: TitleOverrides::default(),
+        upstream_ended: series.upstream_ended,
+        blocked: false,
+    }
+}
+
+fn core_episodes(series_id: i64, detail: &TmdbSeasonDetail) -> Vec<CoreEpisode> {
+    detail
+        .episodes
+        .iter()
+        .map(|source| CoreEpisode {
+            id: EpisodeId(0),
+            season_id: SeasonId(series_id),
+            number: season_number(source.number),
+            title: source.title.clone(),
+            air_date: source.air_date.map(system_time),
+            wanted: false,
+            state: MediaState::Missing,
+            search_attempts: 0,
+            last_searched_at: None,
+        })
+        .collect()
+}
+
+/// An `air_date` as the API writes it, so a refresh and a hand-created
+/// season agree byte for byte and the guarded update sees no change.
+fn air_date_string(air_date: Option<std::time::SystemTime>) -> Option<String> {
+    air_date.map(|when| {
+        chrono::DateTime::<chrono::Utc>::from(when)
+            .date_naive()
+            .to_string()
+    })
+}
+
+/// Dates arrive as `%Y-%m-%d`; midnight UTC carries them losslessly.
+fn system_time(date: chrono::NaiveDate) -> std::time::SystemTime {
+    date.and_time(chrono::NaiveTime::MIN).and_utc().into()
+}
+
+async fn stamp_refreshed(database: &Db, series_id: i64) -> Result<(), RefreshError> {
+    // Stamped even when nothing changed, or the TTL gate above never engages
+    // and every tick pays for TMDB again — same reason as the movie lane.
+    sqlx::query!(
+        "UPDATE series SET metadata_refreshed_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') WHERE id = ?",
+        series_id
+    )
+    .execute(database.pool())
+    .await?;
+    Ok(())
+}
+
+#[cfg(test)]
+#[allow(clippy::unwrap_used)]
+mod tests {
+    use std::sync::Arc;
+
+    use arr_db::Db;
+    use serde_json::json;
+    use tempfile::TempDir;
+    use wiremock::matchers::{method, path};
+    use wiremock::{Mock, MockServer, ResponseTemplate};
+
+    use super::*;
+
+    fn series_body(status: &str) -> serde_json::Value {
+        json!({
+            "id": 82_728,
+            "name": "Bluey",
+            "original_language": "en",
+            "first_air_date": "2018-10-01",
+            "status": status,
+            "seasons": [
+                {"season_number": 1, "episode_count": 2},
+                {"season_number": 2, "episode_count": 1}
+            ],
+            "external_ids": {"tvdb_id": 361_391}
+        })
+    }
+
+    fn season_one_body(episodes: &serde_json::Value) -> serde_json::Value {
+        json!({"season_number": 1, "episodes": episodes})
+    }
+
+    fn two_episodes() -> serde_json::Value {
+        json!([
+            {"episode_number": 1, "name": "Magic Xylophone", "air_date": "2018-10-01"},
+            {"episode_number": 2, "name": "Hospital", "air_date": "2018-10-02"}
+        ])
+    }
+
+    fn season_two_body() -> serde_json::Value {
+        json!({
+            "season_number": 2,
+            "episodes": [
+                {"episode_number": 1, "name": "Dance Mode", "air_date": "2019-04-01"}
+            ]
+        })
+    }
+
+    /// A TMDB serving one series with two seasons. `reset` between phases of
+    /// a test and remount, counting requests by delta around each phase.
+    async fn tmdb(status: &str, season_one: serde_json::Value) -> MockServer {
+        let server = MockServer::start().await;
+        mount(&server, status, season_one).await;
+        server
+    }
+
+    async fn mount(server: &MockServer, status: &str, season_one: serde_json::Value) {
+        Mock::given(method("GET"))
+            .and(path("/tv/82728"))
+            .respond_with(ResponseTemplate::new(200).set_body_json(series_body(status)))
+            .mount(server)
+            .await;
+        Mock::given(method("GET"))
+            .and(path("/tv/82728/season/1"))
+            .respond_with(ResponseTemplate::new(200).set_body_json(season_one))
+            .mount(server)
+            .await;
+        Mock::given(method("GET"))
+            .and(path("/tv/82728/season/2"))
+            .respond_with(ResponseTemplate::new(200).set_body_json(season_two_body()))
+            .mount(server)
+            .await;
+    }
+
+    /// Fresh client per tick, as a restarted process would have — so the only
+    /// thing that can suppress a request is the persisted TTL gate, never the
+    /// in-process response cache.
+    fn action(server: &MockServer) -> SeriesRefreshAction {
+        SeriesRefreshAction::new(Arc::new(
+            TmdbClient::builder("test-key".to_owned())
+                .base_url(server.uri())
+                .build()
+                .unwrap(),
+        ))
+    }
+
+    async fn seeded_series(auto_track: bool) -> (TempDir, Db) {
+        let directory = tempfile::tempdir().unwrap();
+        let database = Db::connect(directory.path().join("arr.db")).await.unwrap();
+        database.migrate().await.unwrap();
+        let root_id: i64 =
+            sqlx::query_scalar("SELECT id FROM roots WHERE kind = 'tv' AND audience = 'main'")
+                .fetch_one(database.pool())
+                .await
+                .unwrap();
+        sqlx::query(
+            "INSERT INTO series (tmdb_id, title, root_id, auto_track) VALUES (82728, 'Bluey', ?, ?)",
+        )
+        .bind(root_id)
+        .bind(auto_track)
+        .execute(database.pool())
+        .await
+        .unwrap();
+        (directory, database)
+    }
+
+    /// Force the next tick past the refresh TTL.
+    async fn expire_refresh(database: &Db) {
+        sqlx::query(
+            "UPDATE series SET metadata_refreshed_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-7 hours')",
+        )
+        .execute(database.pool())
+        .await
+        .unwrap();
+    }
+
+    #[tokio::test]
+    async fn first_refresh_reveals_seasons_and_applies_auto_track() {
+        let (_dir, database) = seeded_series(true).await;
+        let server = tmdb("Returning Series", season_one_body(&two_episodes())).await;
+
+        let outcomes = action(&server).tick(&database).await.unwrap();
+
+        assert_eq!(outcomes.len(), 1);
+        let rows: Vec<(i64, bool)> =
+            sqlx::query_as("SELECT number, tracked FROM seasons ORDER BY number")
+                .fetch_all(database.pool())
+                .await
+                .unwrap();
+        assert_eq!(rows, vec![(1, true), (2, true)]);
+        let wanted: i64 = sqlx::query_scalar(
+            "SELECT count(*) FROM episodes WHERE wanted = 1 AND state = 'missing'",
+        )
+        .fetch_one(database.pool())
+        .await
+        .unwrap();
+        assert_eq!(wanted, 3);
+        // #120 backfill: the series was added before it had an id on file.
+        let tvdb_id: Option<i64> =
+            sqlx::query_scalar("SELECT tvdb_id FROM series WHERE tmdb_id = 82728")
+                .fetch_one(database.pool())
+                .await
+                .unwrap();
+        assert_eq!(tvdb_id, Some(361_391));
+        let stamped: i64 = sqlx::query_scalar(
+            "SELECT count(*) FROM series WHERE metadata_refreshed_at IS NOT NULL",
+        )
+        .fetch_one(database.pool())
+        .await
+        .unwrap();
+        assert_eq!(stamped, 1);
+    }
+
+    /// `auto_track` is a rule about reveals, not intent: without it nothing
+    /// arrives wanted, even though everything is revealed.
+    #[tokio::test]
+    async fn untracked_series_reveals_without_intent() {
+        let (_dir, database) = seeded_series(false).await;
+        let server = tmdb("Returning Series", season_one_body(&two_episodes())).await;
+
+        action(&server).tick(&database).await.unwrap();
+
+        let rows: Vec<(i64, bool)> =
+            sqlx::query_as("SELECT number, tracked FROM seasons ORDER BY number")
+                .fetch_all(database.pool())
+                .await
+                .unwrap();
+        assert_eq!(rows, vec![(1, false), (2, false)]);
+        let wanted: i64 = sqlx::query_scalar("SELECT count(*) FROM episodes WHERE wanted = 1")
+            .fetch_one(database.pool())
+            .await
+            .unwrap();
+        assert_eq!(wanted, 0);
+    }
+
+    /// The refresh has its own TTL: a stamped series costs no TMDB calls,
+    /// whatever else is due.
+    #[tokio::test]
+    async fn refresh_is_throttled_by_its_own_ttl() {
+        let (_dir, database) = seeded_series(true).await;
+        let server = tmdb("Returning Series", season_one_body(&two_episodes())).await;
+
+        action(&server).tick(&database).await.unwrap();
+        let after_first = server.received_requests().await.unwrap().len();
+
+        let second = action(&server).tick(&database).await.unwrap();
+        assert!(second.is_empty());
+        assert_eq!(
+            server.received_requests().await.unwrap().len(),
+            after_first,
+            "the TTL gate must keep the tick off TMDB entirely"
+        );
+    }
+
+    /// A second run over unchanged TMDB data writes nothing and marks nothing
+    /// wanted twice — refresh is idempotent.
+    #[tokio::test]
+    async fn a_second_run_over_unchanged_data_writes_nothing() {
+        let (_dir, database) = seeded_series(true).await;
+        let server = tmdb("Returning Series", season_one_body(&two_episodes())).await;
+        action(&server).tick(&database).await.unwrap();
+        expire_refresh(&database).await;
+
+        let outcomes = action(&server).tick(&database).await.unwrap();
+
+        assert!(
+            outcomes.is_empty(),
+            "unchanged data must not report work done"
+        );
+        let rows: Vec<(i64, bool)> =
+            sqlx::query_as("SELECT number, tracked FROM seasons ORDER BY number")
+                .fetch_all(database.pool())
+                .await
+                .unwrap();
+        assert_eq!(rows, vec![(1, true), (2, true)]);
+        let counts: (i64, i64) =
+            sqlx::query_as("SELECT count(*), coalesce(sum(wanted), 0) FROM episodes")
+                .fetch_one(database.pool())
+                .await
+                .unwrap();
+        assert_eq!(counts, (3, 3));
+    }
+
+    /// Episodes revealed into an existing season follow that season's rule:
+    /// wanted when tracked, not otherwise. Known episodes' air dates and
+    /// titles follow TMDB; §4.2's `airing` window reads them.
+    #[tokio::test]
+    async fn revealed_episodes_follow_the_season_rule_and_details_follow_tmdb() {
+        let (_dir, database) = seeded_series(false).await;
+        let server = tmdb("Returning Series", season_one_body(&two_episodes())).await;
+        action(&server).tick(&database).await.unwrap();
+        // One tracked season, one not — both already revealed above.
+        sqlx::query("UPDATE seasons SET tracked = 1 WHERE number = 1")
+            .execute(database.pool())
+            .await
+            .unwrap();
+        expire_refresh(&database).await;
+        server.reset().await;
+        mount(
+            &server,
+            "Returning Series",
+            season_one_body(&json!([
+                {"episode_number": 1, "name": "Magic Xylophone!", "air_date": "2018-10-08"},
+                {"episode_number": 2, "name": "Hospital", "air_date": "2018-10-02"},
+                {"episode_number": 3, "name": "Bike", "air_date": null}
+            ])),
+        )
+        .await;
+
+        let outcomes = action(&server).tick(&database).await.unwrap();
+
+        assert_eq!(outcomes.len(), 1);
+        let season_one: Vec<(i64, String, Option<String>, bool)> = sqlx::query_as(
+            "SELECT e.number, e.title, e.air_date, e.wanted
+             FROM episodes e JOIN seasons s ON s.id = e.season_id
+             WHERE s.number = 1 ORDER BY e.number",
+        )
+        .fetch_all(database.pool())
+        .await
+        .unwrap();
+        assert_eq!(
+            season_one,
+            vec![
+                // Episodes 1 and 2 were revealed before tracking was turned
+                // on, and refresh never grants intent implicitly; only the
+                // newly revealed 3 follows the rule.
+                (
+                    1,
+                    "Magic Xylophone!".to_owned(),
+                    Some("2018-10-08".to_owned()),
+                    false
+                ),
+                (
+                    2,
+                    "Hospital".to_owned(),
+                    Some("2018-10-02".to_owned()),
+                    false
+                ),
+                (3, "Bike".to_owned(), None, true),
+            ],
+            "tracked season: new episode arrives wanted, changed details move"
+        );
+        let season_two: Vec<bool> = sqlx::query_as(
+            "SELECT e.wanted FROM episodes e JOIN seasons s ON s.id = e.season_id
+             WHERE s.number = 2",
+        )
+        .fetch_all(database.pool())
+        .await
+        .unwrap()
+        .into_iter()
+        .map(|(wanted,): (bool,)| wanted)
+        .collect();
+        assert_eq!(season_two, vec![false], "untracked season stays unwanted");
+    }
+
+    /// §4.2 derives `ended` from `upstream_ended`, so TMDB's status must
+    /// reach the column.
+    #[tokio::test]
+    async fn upstream_ended_follows_tmdb_status() {
+        let (_dir, database) = seeded_series(true).await;
+        let server = tmdb("Returning Series", season_one_body(&two_episodes())).await;
+        action(&server).tick(&database).await.unwrap();
+        expire_refresh(&database).await;
+        server.reset().await;
+        mount(&server, "Canceled", season_one_body(&two_episodes())).await;
+
+        action(&server).tick(&database).await.unwrap();
+
+        let ended: bool =
+            sqlx::query_scalar("SELECT upstream_ended FROM series WHERE tmdb_id = 82728")
+                .fetch_one(database.pool())
+                .await
+                .unwrap();
+        assert!(ended);
+    }
+}

crates/arr-db/migrations/0018_series_metadata_refresh.sql

@@ -0,0 +1,4 @@
+-- §8: the daily series metadata refresh is gated on this stamp, exactly as
+-- the movie lane gates on `movies.metadata_refreshed_at` (0009). NULL means
+-- never refreshed, which makes a newly added series immediately due.
+ALTER TABLE series ADD COLUMN metadata_refreshed_at TEXT;