feat(daemon): reconcile episodes that vanish upstream

.sqlx/query-3974fc9a150b9d1feb9401f7c8075a774d0570104d0b089fd2e5ed2e85e66f0c.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "DELETE FROM episodes WHERE id = ?",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": []
+  },
+  "hash": "3974fc9a150b9d1feb9401f7c8075a774d0570104d0b089fd2e5ed2e85e66f0c"
+}

.sqlx/query-40f68614d733827b00499cb65a4c9b0219c18d2321720c060e1d22bb398d7d3d.json

@@ -0,0 +1,20 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT EXISTS(\n                       SELECT 1 FROM media_files\n                       WHERE owner_kind = 'episode' AND owner_id = ?\n                   ) AS \"exists!: bool\"",
+  "describe": {
+    "columns": [
+      {
+        "name": "exists!: bool",
+        "ordinal": 0,
+        "type_info": "Integer"
+      }
+    ],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "40f68614d733827b00499cb65a4c9b0219c18d2321720c060e1d22bb398d7d3d"
+}

.sqlx/query-625b669e6a408eec81bbb235fb2eb3922336dfebee6f3fec7987fd538e472c34.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "UPDATE episodes SET vanished = 1, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') WHERE id = ? AND vanished = 0",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": []
+  },
+  "hash": "625b669e6a408eec81bbb235fb2eb3922336dfebee6f3fec7987fd538e472c34"
+}

.sqlx/query-a0653d4df0c32346bede52dbc3c6e501b3ef1e0d742e6eb06e1dc2fb4c524fb7.json

@@ -1,6 +1,6 @@
 {
   "db_name": "SQLite",
-  "query": "SELECT id AS \"id!: i64\", number AS \"number!: i64\",\n                      title AS \"title!: String\", air_date\n               FROM episodes WHERE season_id = ?",
+  "query": "SELECT id AS \"id!: i64\", number AS \"number!: i64\",\n                      title AS \"title!: String\", air_date,\n                      vanished AS \"vanished!: bool\"\n               FROM episodes WHERE season_id = ?",
   "describe": {
     "columns": [
       {
@@ -22,6 +22,11 @@
         "name": "air_date",
         "ordinal": 3,
         "type_info": "Text"
+      },
+      {
+        "name": "vanished!: bool",
+        "ordinal": 4,
+        "type_info": "Integer"
       }
     ],
     "parameters": {
@@ -31,8 +36,9 @@
       true,
       false,
       false,
-      true
+      true,
+      false
     ]
   },
-  "hash": "dde89ed2cd3d783c5946a593e4ef31e9aef2846b9061c8dccd6c6f6be8c04644"
+  "hash": "a0653d4df0c32346bede52dbc3c6e501b3ef1e0d742e6eb06e1dc2fb4c524fb7"
 }

.sqlx/query-caa918385c5caf36e1ebec08b9d4f1f21dacc8cdae322af3d4af82224d5b96c8.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "UPDATE episodes SET vanished = 0, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') WHERE id = ? AND vanished != 0",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": []
+  },
+  "hash": "caa918385c5caf36e1ebec08b9d4f1f21dacc8cdae322af3d4af82224d5b96c8"
+}

crates/arr-daemon/src/series_refresh.rs

@@ -13,8 +13,14 @@
 //! - `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
+//! Episodes that vanish upstream follow two rules (#122): without a file of
-//! is idempotent: over unchanged TMDB data only the stamp moves.
+//! their own they are deleted outright, cascading through `episode_releases`;
+//! with one they are flagged `vanished` instead, because deleting the row
+//! would orphan a real file (`media_files` is polymorphic on its owner). A
+//! vanished number that reappears clears its flag again. Renumbering needs no
+//! matching of its own — it is just these two rules seen from both ends.
+//!
+//! Refresh is idempotent: over unchanged TMDB data only the stamp moves.
 
 use std::collections::HashMap;
 use std::sync::Arc;
@@ -237,7 +243,8 @@ impl SeriesRefreshAction {
     ///
     /// 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.
+    /// the season is tracked (§4.1). Numbers TMDB no longer lists are
+    /// reconciled by `reconcile_vanished` (#122).
     async fn sync_season(
         &self,
         executor: &mut sqlx::SqliteConnection,
@@ -249,7 +256,8 @@ impl SeriesRefreshAction {
         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
+                      title AS "title!: String", air_date,
+                      vanished AS "vanished!: bool"
                FROM episodes WHERE season_id = ?"#,
             season_id
         )
@@ -259,7 +267,12 @@ impl SeriesRefreshAction {
         .map(|episode| {
             (
                 episode.number,
-                (episode.id, episode.title, episode.air_date),
+                (
+                    episode.id,
+                    episode.title,
+                    episode.air_date,
+                    episode.vanished,
+                ),
             )
         })
         .collect::<HashMap<_, _>>();
@@ -268,7 +281,7 @@ impl SeriesRefreshAction {
         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) {
+            if let Some(&(id, ref title, ref air_date, vanished)) = 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!(
@@ -283,6 +296,17 @@ impl SeriesRefreshAction {
                     .await?;
                     changed = true;
                 }
+                // The number is back upstream — a TMDB reversal, or a
+                // renumber seen from the other end. The conflict is over.
+                if vanished {
+                    sqlx::query!(
+                        "UPDATE episodes SET vanished = 0, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') WHERE id = ? AND vanished != 0",
+                        id
+                    )
+                    .execute(&mut *executor)
+                    .await?;
+                    changed = true;
+                }
             } else {
                 fresh.push(CoreEpisode {
                     id: EpisodeId(0),
@@ -313,6 +337,56 @@ impl SeriesRefreshAction {
             .await?;
             changed = true;
         }
+        changed |= self.reconcile_vanished(executor, &known, &detail).await?;
+        Ok(changed)
+    }
+
+    /// #122. Numbers the library knows that TMDB no longer lists have
+    /// vanished upstream. One without a `media_files` row is deleted — the
+    /// foreign key cascades through `episode_releases` — and one with a file
+    /// is flagged instead: `media_files.path` is UNIQUE and its owner is
+    /// polymorphic, so dropping the row would orphan a real file and block
+    /// re-importing that path, the same trap `movies.rs` documents on the
+    /// movie side. Nothing here touches the disk.
+    async fn reconcile_vanished(
+        &self,
+        executor: &mut sqlx::SqliteConnection,
+        known: &HashMap<i64, (i64, String, Option<String>, bool)>,
+        detail: &TmdbSeasonDetail,
+    ) -> Result<bool, RefreshError> {
+        let upstream: HashMap<i64, ()> = detail
+            .episodes
+            .iter()
+            .map(|source| (i64::from(source.number), ()))
+            .collect();
+        let mut changed = false;
+        for (number, &(id, ..)) in known {
+            if upstream.contains_key(number) {
+                continue;
+            }
+            let has_file = sqlx::query_scalar!(
+                r#"SELECT EXISTS(
+                       SELECT 1 FROM media_files
+                       WHERE owner_kind = 'episode' AND owner_id = ?
+                   ) AS "exists!: bool""#,
+                id
+            )
+            .fetch_one(&mut *executor)
+            .await?;
+            if has_file {
+                sqlx::query!(
+                    "UPDATE episodes SET vanished = 1, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') WHERE id = ? AND vanished = 0",
+                    id
+                )
+                .execute(&mut *executor)
+                .await?;
+            } else {
+                sqlx::query!("DELETE FROM episodes WHERE id = ?", id)
+                    .execute(&mut *executor)
+                    .await?;
+            }
+            changed = true;
+        }
         Ok(changed)
     }
 }
@@ -728,4 +802,175 @@ mod tests {
                 .unwrap();
         assert!(ended);
     }
+
+    async fn season_one_episode(database: &Db, number: i64) -> i64 {
+        sqlx::query_scalar(
+            "SELECT e.id FROM episodes e JOIN seasons s ON s.id = e.season_id
+             WHERE s.number = 1 AND e.number = ?",
+        )
+        .bind(number)
+        .fetch_one(database.pool())
+        .await
+        .unwrap()
+    }
+
+    /// A stored release candidate for `episode_id`, standing in for the rows
+    /// a search leaves behind.
+    async fn attach_release(database: &Db, episode_id: i64) {
+        let release_id: i64 = sqlx::query_scalar(
+            "INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed, verdict)
+             VALUES (7, ?, 'release', 10737418240, 'https://tracker/x.torrent', '{}', 'eligible')
+             RETURNING id",
+        )
+        .bind(format!("guid-{episode_id}"))
+        .fetch_one(database.pool())
+        .await
+        .unwrap();
+        sqlx::query("INSERT INTO episode_releases (episode_id, release_id) VALUES (?, ?)")
+            .bind(episode_id)
+            .bind(release_id)
+            .execute(database.pool())
+            .await
+            .unwrap();
+    }
+
+    /// An imported file for `episode_id`, as the import tick records it.
+    async fn attach_file(database: &Db, episode_id: i64) {
+        sqlx::query(
+            "INSERT INTO media_files (owner_kind, owner_id, path, size) VALUES ('episode', ?, ?, 1)",
+        )
+        .bind(episode_id)
+        .bind(format!("/mnt/media/tv/main/episode-{episode_id}.mkv"))
+        .execute(database.pool())
+        .await
+        .unwrap();
+    }
+
+    /// #122. TMDB dropped episode 2: with no file of its own it is deleted,
+    /// and its stored releases cascade with it. The wanted flag that would
+    /// otherwise pin the series at `incomplete` goes with the row.
+    #[tokio::test]
+    async fn a_vanished_episode_without_a_file_is_deleted() {
+        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 episode_two = season_one_episode(&database, 2).await;
+        attach_release(&database, episode_two).await;
+        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-01"}
+            ])),
+        )
+        .await;
+
+        let outcomes = action(&server).tick(&database).await.unwrap();
+
+        assert_eq!(outcomes.len(), 1, "a removal is work worth reporting");
+        let remaining: Vec<i64> = sqlx::query_scalar(
+            "SELECT e.number 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!(remaining, vec![1]);
+        let cascaded: i64 =
+            sqlx::query_scalar("SELECT count(*) FROM episode_releases WHERE episode_id = ?")
+                .bind(episode_two)
+                .fetch_one(database.pool())
+                .await
+                .unwrap();
+        assert_eq!(cascaded, 0);
+    }
+
+    /// #122. With a file on record the vanished episode is never deleted:
+    /// `media_files.path` is UNIQUE and its owner polymorphic, so dropping
+    /// the row would orphan a real file. It is flagged instead — the conflict
+    /// the operator resolves — and the file row stays put.
+    #[tokio::test]
+    async fn a_vanished_episode_with_a_file_is_flagged_not_deleted() {
+        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 episode_two = season_one_episode(&database, 2).await;
+        attach_file(&database, episode_two).await;
+        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-01"}
+            ])),
+        )
+        .await;
+
+        let outcomes = action(&server).tick(&database).await.unwrap();
+
+        assert_eq!(outcomes.len(), 1);
+        let (vanished, files): (i64, i64) = sqlx::query_as(
+            "SELECT e.vanished,
+                    (SELECT count(*) FROM media_files f
+                     WHERE f.owner_kind = 'episode' AND f.owner_id = e.id)
+             FROM episodes e WHERE e.id = ?",
+        )
+        .bind(episode_two)
+        .fetch_one(database.pool())
+        .await
+        .unwrap();
+        assert_eq!(
+            (vanished, files),
+            (1, 1),
+            "flagged as a conflict, file intact"
+        );
+    }
+
+    /// Idempotence over the new writes too: a second refresh over the same
+    /// TMDB data neither re-reports nor rewrites, and a number TMDB restores
+    /// clears the flag again.
+    #[tokio::test]
+    async fn a_restored_number_clears_the_vanished_flag() {
+        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 episode_two = season_one_episode(&database, 2).await;
+        attach_file(&database, episode_two).await;
+        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-01"}
+            ])),
+        )
+        .await;
+        action(&server).tick(&database).await.unwrap();
+        expire_refresh(&database).await;
+        // TMDB puts the episode back where it was.
+        server.reset().await;
+        mount(
+            &server,
+            "Returning Series",
+            season_one_body(&json!([
+                {"episode_number": 1, "name": "Magic Xylophone", "air_date": "2018-10-01"},
+                {"episode_number": 2, "name": "Hospital", "air_date": "2018-10-02"}
+            ])),
+        )
+        .await;
+
+        let outcomes = action(&server).tick(&database).await.unwrap();
+
+        assert_eq!(outcomes.len(), 1);
+        let vanished: bool = sqlx::query_scalar("SELECT vanished FROM episodes WHERE id = ?")
+            .bind(episode_two)
+            .fetch_one(database.pool())
+            .await
+            .unwrap();
+        assert!(!vanished, "the conflict is over once TMDB lists it again");
+    }
 }