feat(api): series, season and episode endpoints (#81)

crates/arr-api/src/lib.rs

@@ -11,6 +11,7 @@ mod movies;
 mod owners;
 mod roots;
 mod search;
+mod series;
 mod state;
 
 use axum::routing::get;
@@ -25,7 +26,11 @@ pub use movies::{Accepted, AttentionQueues, CreateMovie, ErrorBody, Movie, Relea
 pub use owners::{CreateOwner, Owner, UpdateOwner};
 pub use roots::Root;
 pub use search::{ClassifiedRelease, SearchResponse};
-pub use state::{AppState, MovieCommand, Upstreams, DEFAULT_TMDB_URL};
+pub use series::{
+    CreateEpisode, CreateSeason, CreateSeries, Episode, Season, Series, UpdateEpisode,
+    UpdateSeason, UpdateSeries,
+};
+pub use state::{AppState, EpisodeCommand, MovieCommand, Upstreams, DEFAULT_TMDB_URL};
 
 /// Where the generated document is served, and where `just gen-client` reads
 /// it back from when it is fetched rather than dumped from the binary.
@@ -46,6 +51,7 @@ pub const DOCS_PATH: &str = "/api/docs";
     tags(
         (name = "system", description = "Service health and metadata"),
         (name = "movies", description = "Movie library and actions"),
+        (name = "series", description = "Series, seasons and episodes (DESIGN.md §4.1, §4.2)"),
         (name = "owners", description = "Owner tags and filtered views (DESIGN.md §4.3)"),
         (name = "search", description = "Unified title and release search"),
         (name = "roots", description = "Root folders and their policies")
@@ -65,6 +71,16 @@ fn api_router() -> OpenApiRouter<AppState> {
         .routes(routes!(movies::attention))
         .routes(routes!(movies::list_owners))
         .routes(routes!(movies::tag_owner, movies::untag_owner))
+        .routes(routes!(series::list, series::create))
+        .routes(routes!(series::get, series::update, series::delete))
+        .routes(routes!(series::seasons, series::create_season))
+        .routes(routes!(series::update_season))
+        .routes(routes!(series::get_episode, series::update_episode))
+        .routes(routes!(series::search_episode))
+        .routes(routes!(series::episode_releases))
+        .routes(routes!(series::grab_episode))
+        .routes(routes!(series::list_owners))
+        .routes(routes!(series::tag_owner, series::untag_owner))
         .routes(routes!(owners::list, owners::create))
         .routes(routes!(owners::get, owners::update, owners::delete))
         .routes(routes!(search::search))

crates/arr-api/src/movies.rs

@@ -85,6 +85,9 @@ pub struct ErrorBody {
 #[derive(Debug)]
 pub enum ApiError {
     NotFound,
+    SeriesNotFound,
+    SeasonNotFound,
+    EpisodeNotFound,
     OwnerNotFound,
     Conflict(String),
     Invalid(String),
@@ -96,6 +99,9 @@ impl IntoResponse for ApiError {
     fn into_response(self) -> Response {
         let (status, error) = match self {
             Self::NotFound => (StatusCode::NOT_FOUND, "movie not found".to_string()),
+            Self::SeriesNotFound => (StatusCode::NOT_FOUND, "series not found".to_string()),
+            Self::SeasonNotFound => (StatusCode::NOT_FOUND, "season not found".to_string()),
+            Self::EpisodeNotFound => (StatusCode::NOT_FOUND, "episode not found".to_string()),
             Self::OwnerNotFound => (StatusCode::NOT_FOUND, "owner not found".to_string()),
             Self::Conflict(error) => (StatusCode::CONFLICT, error),
             Self::Invalid(error) => (StatusCode::UNPROCESSABLE_ENTITY, error),
@@ -104,7 +110,7 @@ impl IntoResponse for ApiError {
                 "database unavailable".into(),
             ),
             Self::Database(error) => {
-                tracing::error!(%error, "movie API database error");
+                tracing::error!(%error, "API database error");
                 (StatusCode::INTERNAL_SERVER_ERROR, "database error".into())
             }
         };

crates/arr-api/src/search.rs

@@ -4,7 +4,7 @@ use arr_core::policy::{evaluate, Candidate};
 use arr_core::score::score;
 use arr_core::{Language, Policy, Rule, TitleOverrides, Verdict};
 use arr_db::policy::language;
-use arr_indexer::{ProwlarrClient, SearchRelease, SearchRequest};
+use arr_indexer::{ProwlarrClient, SearchRelease, SearchRequest, TvSelector, TvTarget};
 use axum::extract::{Query, State};
 use axum::Json;
 use chrono::{DateTime, Utc};
@@ -19,9 +19,14 @@ pub struct SearchQuery {
     q: String,
 }
 
+/// Which title a manual release search is for. Exactly one of the two.
+///
+/// Intent lives at the leaf (`DESIGN.md` §4.1), so the TV side of this is an
+/// episode even when the release that satisfies it is a season pack.
 #[derive(Debug, Deserialize, IntoParams)]
 pub struct ReleasesQuery {
-    movie_id: i64,
+    movie_id: Option<i64>,
+    episode_id: Option<i64>,
 }
 
 #[derive(Debug, Clone, Serialize, ToSchema)]
@@ -203,6 +208,7 @@ async fn search_tmdb(
     responses(
         (status = 200, body = [ClassifiedRelease]),
         (status = 404, body = ErrorBody),
+        (status = 422, body = ErrorBody),
         (status = 500, body = ErrorBody),
         (status = 503, body = ErrorBody)
     )
@@ -211,16 +217,33 @@ pub async fn releases(
     State(state): State<AppState>,
     Query(query): Query<ReleasesQuery>,
 ) -> Result<Json<Vec<ClassifiedRelease>>, ApiError> {
+    let mut classified = match (query.movie_id, query.episode_id) {
+        (Some(movie_id), None) => movie_releases(&state, movie_id).await?,
+        (None, Some(episode_id)) => episode_releases(&state, episode_id).await?,
+        _ => {
+            return Err(ApiError::Invalid(
+                "pass exactly one of movie_id and episode_id".into(),
+            ))
+        }
+    };
+    classified.sort_by_key(|release| (bucket(&release.verdict), -release.score));
+    Ok(Json(classified))
+}
+
+async fn movie_releases(
+    state: &AppState,
+    movie_id: i64,
+) -> Result<Vec<ClassifiedRelease>, ApiError> {
     let database = state.database().ok_or(ApiError::Unavailable)?;
-    let movie = sqlx::query!(r#"SELECT title AS "title!: String", tmdb_id AS "tmdb_id!: i64", original_language FROM movies WHERE id = ?"#, query.movie_id)
+    let movie = sqlx::query!(r#"SELECT title AS "title!: String", tmdb_id AS "tmdb_id!: i64", original_language FROM movies WHERE id = ?"#, movie_id)
         .fetch_optional(database.pool()).await?.ok_or(ApiError::NotFound)?;
     let loaded = database
-        .movie_policy(query.movie_id)
+        .movie_policy(movie_id)
         .await
         .map_err(|error| ApiError::Database(error.to_string()))?
         .ok_or(ApiError::NotFound)?;
 
-    let tmdb = tmdb_client(&state)?
+    let tmdb = tmdb_client(state)?
         .movie(
             u32::try_from(movie.tmdb_id)
                 .map_err(|_| ApiError::Invalid("movie has invalid TMDB id".into()))?,
@@ -233,13 +256,7 @@ pub async fn releases(
         },
         |imdb_id| SearchRequest::Movie { imdb_id },
     );
-    let upstreams = state.upstreams();
-    let api_key = upstreams
-        .prowlarr_api_key
-        .clone()
-        .ok_or(ApiError::Unavailable)?;
-    let prowlarr = ProwlarrClient::new(upstreams.prowlarr_url.clone(), api_key)
-        .map_err(|_| ApiError::Unavailable)?;
+    let prowlarr = prowlarr_client(state)?;
     let indexers = prowlarr
         .indexers()
         .await
@@ -282,8 +299,80 @@ pub async fn releases(
             }
         }
     }
-    classified.sort_by_key(|release| (bucket(&release.verdict), -release.score));
-    Ok(Json(classified))
+    Ok(classified)
+}
+
+/// Classified releases for one episode (`DESIGN.md` §6.1, §9.3).
+///
+/// The series carries no TVDB ID yet, so `tv_request` falls back to a text
+/// search built from the title and the `SxxEyy` tag. Both widen the result
+/// set rather than narrowing it, which the buckets already handle.
+async fn episode_releases(
+    state: &AppState,
+    episode_id: i64,
+) -> Result<Vec<ClassifiedRelease>, ApiError> {
+    let database = state.database().ok_or(ApiError::Unavailable)?;
+    let episode = sqlx::query!(
+        r#"SELECT s.title AS "series_title!: String", s.original_language,
+                  se.number AS "season_number!: i64", e.number AS "episode_number!: i64"
+           FROM episodes e
+           JOIN seasons se ON se.id = e.season_id
+           JOIN series s   ON s.id = se.series_id
+           WHERE e.id = ?"#,
+        episode_id
+    )
+    .fetch_optional(database.pool())
+    .await?
+    .ok_or(ApiError::EpisodeNotFound)?;
+    let loaded = database
+        .episode_policy(episode_id)
+        .await
+        .map_err(|error| ApiError::Database(error.to_string()))?
+        .ok_or(ApiError::EpisodeNotFound)?;
+
+    // §5.2. The language rule is written against the title's original
+    // language, so guessing one would silently change every verdict.
+    let original_language = episode.original_language.as_deref().ok_or_else(|| {
+        ApiError::Invalid("series has no original_language; refresh its metadata first".into())
+    })?;
+    let original_language = title_language(original_language, &[]);
+
+    let target = TvTarget {
+        tvdb_id: None,
+        title: episode.series_title,
+        selector: TvSelector::Episode {
+            season: u32::try_from(episode.season_number).unwrap_or_default(),
+            episode: u32::try_from(episode.episode_number).unwrap_or_default(),
+        },
+    };
+    let prowlarr = prowlarr_client(state)?;
+    let indexers = prowlarr
+        .indexers()
+        .await
+        .map_err(|_| ApiError::Unavailable)?;
+
+    let mut classified = Vec::new();
+    for indexer in indexers {
+        let Some(request) = indexer.capabilities.tv_request(&target) else {
+            continue;
+        };
+        match prowlarr.search_indexer(indexer.id, &request).await {
+            Ok(releases) => {
+                for release in releases {
+                    classified.push(classify(
+                        release,
+                        &loaded.policy,
+                        &loaded.overrides,
+                        &original_language,
+                    )?);
+                }
+            }
+            Err(error) => {
+                tracing::warn!(indexer_id = indexer.id, %error, "manual episode search failed");
+            }
+        }
+    }
+    Ok(classified)
 }
 
 impl From<arr_meta::MovieSearchResult> for TmdbMovie {
@@ -329,6 +418,15 @@ fn escape_like(input: &str) -> String {
         .replace('_', "\\_")
 }
 
+fn prowlarr_client(state: &AppState) -> Result<ProwlarrClient, ApiError> {
+    let upstreams = state.upstreams();
+    let api_key = upstreams
+        .prowlarr_api_key
+        .clone()
+        .ok_or(ApiError::Unavailable)?;
+    ProwlarrClient::new(upstreams.prowlarr_url.clone(), api_key).map_err(|_| ApiError::Unavailable)
+}
+
 fn tmdb_client(state: &AppState) -> Result<arr_meta::TmdbClient, ApiError> {
     let upstreams = state.upstreams();
     let key = upstreams
@@ -590,6 +688,75 @@ mod tests {
         assert_eq!(eligible["score"], 0);
     }
 
+    #[tokio::test]
+    async fn episode_releases_search_by_season_and_episode_tag() {
+        let tmdb = MockServer::start().await;
+        let prowlarr = MockServer::start().await;
+        Mock::given(method("GET"))
+            .and(path("/api/v1/indexer"))
+            .respond_with(
+                ResponseTemplate::new(200)
+                    .set_body_json(serde_json::json!([{"id":9,"name":"tracker","enable":true}])),
+            )
+            .mount(&prowlarr)
+            .await;
+        // No TVDB ID on the series, so the indexer's tvsearch cannot be
+        // addressed by ID and the text fallback carries the SxxEyy tag.
+        Mock::given(method("GET")).and(path("/9/api")).and(query_param("t", "caps"))
+            .respond_with(ResponseTemplate::new(200).set_body_string("<caps><searching><search available=\"yes\" supportedParams=\"q\"/><tv-search available=\"yes\" supportedParams=\"q,tvdbid,season,ep\"/></searching></caps>"))
+            .mount(&prowlarr).await;
+        Mock::given(method("GET"))
+            .and(path("/9/api"))
+            .and(query_param("t", "search"))
+            .and(query_param("q", "Bluey S01E02"))
+            .respond_with(ResponseTemplate::new(200).set_body_string(r"<rss><channel><item><title>Bluey.S01E02.1080p.WEB-DL</title><guid>ep</guid><link>https://tracker/ep</link><size>1500000000</size></item></channel></rss>"))
+            .mount(&prowlarr).await;
+
+        let (_dir, state, base) = application(&tmdb, &prowlarr).await;
+        let pool = state.database().expect("database").pool();
+        let root_id: i64 =
+            sqlx::query_scalar("SELECT id FROM roots WHERE kind = 'tv' AND audience = 'main'")
+                .fetch_one(pool)
+                .await
+                .expect("TV root");
+        let series_id: i64 = sqlx::query_scalar("INSERT INTO series (tmdb_id, title, year, original_language, root_id) VALUES (82728, 'Bluey', 2018, 'en', ?) RETURNING id")
+            .bind(root_id).fetch_one(pool).await.expect("series");
+        let season_id: i64 = sqlx::query_scalar(
+            "INSERT INTO seasons (series_id, number) VALUES (?, 1) RETURNING id",
+        )
+        .bind(series_id)
+        .fetch_one(pool)
+        .await
+        .expect("season");
+        let episode_id: i64 = sqlx::query_scalar("INSERT INTO episodes (season_id, number, title, wanted) VALUES (?, 2, 'Hospital', 1) RETURNING id")
+            .bind(season_id).fetch_one(pool).await.expect("episode");
+
+        let response = reqwest::get(format!("{base}/api/releases?episode_id={episode_id}"))
+            .await
+            .expect("releases");
+        assert_eq!(response.status(), 200);
+        let releases: Vec<serde_json::Value> = response.json().await.expect("json");
+        assert_eq!(releases.len(), 1);
+        assert_eq!(releases[0]["guid"], "ep");
+        assert_eq!(releases[0]["parsed"]["episode"]["kind"], "episodes");
+        assert_eq!(releases[0]["parsed"]["episode"]["season"], 1);
+        assert_eq!(releases[0]["parsed"]["episode"]["episodes"][0], 2);
+    }
+
+    #[tokio::test]
+    async fn a_manual_search_names_exactly_one_title() {
+        let tmdb = MockServer::start().await;
+        let prowlarr = MockServer::start().await;
+        let (_dir, _state, base) = application(&tmdb, &prowlarr).await;
+
+        for query in ["", "movie_id=1&episode_id=1"] {
+            let response = reqwest::get(format!("{base}/api/releases?{query}"))
+                .await
+                .expect("releases");
+            assert_eq!(response.status(), 422, "query: {query}");
+        }
+    }
+
     #[test]
     fn releases_without_sizes_skip_the_size_score() {
         let policy = Policy {

crates/arr-api/src/state.rs

@@ -70,6 +70,8 @@ pub struct AppState {
     database: Option<Db>,
     movie_commands: mpsc::Sender<MovieCommand>,
     pending_movie_commands: Arc<tokio::sync::Mutex<mpsc::Receiver<MovieCommand>>>,
+    episode_commands: mpsc::Sender<EpisodeCommand>,
+    pending_episode_commands: Arc<tokio::sync::Mutex<mpsc::Receiver<EpisodeCommand>>>,
 }
 
 /// Work explicitly requested through the movie API.
@@ -79,6 +81,16 @@ pub enum MovieCommand {
     Grab { movie_id: i64, release_id: i64 },
 }
 
+/// Work explicitly requested through the series API.
+///
+/// Intent lives at the leaf (`DESIGN.md` §4.1), so a manual action names an
+/// episode even when the release that satisfies it is a season pack.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum EpisodeCommand {
+    Search { episode_id: i64 },
+    Grab { episode_id: i64, release_id: i64 },
+}
+
 impl AppState {
     /// Build the state, including the shared HTTP client.
     ///
@@ -88,12 +100,15 @@ impl AppState {
     pub fn new(upstreams: Upstreams) -> Result<Self, reqwest::Error> {
         let http = reqwest::Client::builder().timeout(PROBE_TIMEOUT).build()?;
         let (movie_commands, pending_movie_commands) = mpsc::channel(64);
+        let (episode_commands, pending_episode_commands) = mpsc::channel(64);
         Ok(Self {
             http,
             upstreams: Arc::new(upstreams),
             database: None,
             movie_commands,
             pending_movie_commands: Arc::new(tokio::sync::Mutex::new(pending_movie_commands)),
+            episode_commands,
+            pending_episode_commands: Arc::new(tokio::sync::Mutex::new(pending_episode_commands)),
         })
     }
 
@@ -113,6 +128,15 @@ impl AppState {
         self.pending_movie_commands.lock().await.recv().await
     }
 
+    /// Wait for the next manual episode action in the daemon's reconcile loop.
+    ///
+    /// # Errors
+    ///
+    /// If every sender has been dropped.
+    pub async fn next_episode_command(&self) -> Option<EpisodeCommand> {
+        self.pending_episode_commands.lock().await.recv().await
+    }
+
     pub(crate) fn http(&self) -> &reqwest::Client {
         &self.http
     }
@@ -131,4 +155,11 @@ impl AppState {
     ) -> Result<(), mpsc::error::TrySendError<MovieCommand>> {
         self.movie_commands.try_send(command)
     }
+
+    pub(crate) fn send_episode_command(
+        &self,
+        command: EpisodeCommand,
+    ) -> Result<(), mpsc::error::TrySendError<EpisodeCommand>> {
+        self.episode_commands.try_send(command)
+    }
 }

crates/arr-db/migrations/0006_episode_releases.sql

@@ -0,0 +1,11 @@
+-- The TV counterpart of movie_releases (0003). One release comes back for
+-- more than one episode — a season pack matches every episode in the season —
+-- so the release row stays unique by indexer and guid, and each match is a
+-- row here.
+CREATE TABLE episode_releases (
+    episode_id INTEGER NOT NULL REFERENCES episodes (id) ON DELETE CASCADE,
+    release_id INTEGER NOT NULL REFERENCES releases (id) ON DELETE CASCADE,
+    PRIMARY KEY (episode_id, release_id)
+) STRICT;
+
+CREATE INDEX episode_releases_release ON episode_releases (release_id);

crates/arr-db/src/lib.rs

@@ -7,7 +7,7 @@ use std::path::Path;
 
 pub mod policy;
 
-pub use policy::{MoviePolicy, PolicyColumns, PolicyError};
+pub use policy::{MoviePolicy, PolicyColumns, PolicyError, TitlePolicy};
 
 use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions, SqliteSynchronous};
 use sqlx::{migrate::MigrateError, SqlitePool};

crates/arr-db/src/policy.rs

@@ -27,13 +27,13 @@ pub enum PolicyError {
     },
 }
 
-/// The effective policy for one movie, plus the root it is attached to.
+/// The effective policy for one title, plus the root it is attached to.
 ///
 /// The root's `kind` and `audience` are the Transmission label and the
 /// on-disk layout (§7.1, §7.4), and they only exist together with the policy,
 /// so they are returned together.
 #[derive(Debug, Clone, PartialEq, Eq)]
-pub struct MoviePolicy {
+pub struct TitlePolicy {
     pub policy: Policy,
     /// Per-title relaxations and tightenings of the root policy (§5.1).
     pub overrides: TitleOverrides,
@@ -45,6 +45,9 @@ pub struct MoviePolicy {
     pub root_path: String,
 }
 
+/// What [`Db::movie_policy`] returned before episodes needed the same shape.
+pub type MoviePolicy = TitlePolicy;
+
 /// The raw policy columns, as the `policies` table stores them (§5.5, §10).
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct PolicyColumns {
@@ -139,7 +142,7 @@ impl Db {
     ///
     /// If the query fails, or a policy column does not hold the JSON its
     /// migration promises.
-    pub async fn movie_policy(&self, movie_id: i64) -> Result<Option<MoviePolicy>, PolicyError> {
+    pub async fn movie_policy(&self, movie_id: i64) -> Result<Option<TitlePolicy>, PolicyError> {
         let row = sqlx::query!(
             r#"
             SELECT   m.overrides       AS "overrides!: String",
@@ -184,7 +187,82 @@ impl Db {
         }
         .to_policy()?;
 
-        Ok(Some(MoviePolicy {
+        Ok(Some(TitlePolicy {
+            policy,
+            overrides: TitleOverrides {
+                only_4k: overrides.only_4k,
+                allow_english_audio: overrides.allow_english_audio,
+            },
+            root_id: row.root_id,
+            root_kind: row.root_kind,
+            root_audience: row.root_audience,
+            root_path: row.root_path,
+        }))
+    }
+
+    /// The policy attached to an episode's series root, with that series'
+    /// overrides.
+    ///
+    /// Overrides sit on the series (§5.1): an episode is a leaf carrying
+    /// intent, never its own policy.
+    ///
+    /// `None` when the episode does not exist.
+    ///
+    /// # Errors
+    ///
+    /// If the query fails, or a policy column does not hold the JSON its
+    /// migration promises.
+    pub async fn episode_policy(
+        &self,
+        episode_id: i64,
+    ) -> Result<Option<TitlePolicy>, PolicyError> {
+        let row = sqlx::query!(
+            r#"
+            SELECT   s.overrides       AS "overrides!: String",
+                     r.id              AS "root_id!: i64",
+                     r.kind            AS "root_kind!: String",
+                     r.audience        AS "root_audience!: String",
+                     r.path            AS "root_path!: String",
+                     p.id              AS "policy_id!: i64",
+                     p.name            AS "policy_name!: String",
+                     p.required_audio  AS "required_audio!: String",
+                     p.dub_blacklist   AS "dub_blacklist!: String",
+                     p.hdr_rules       AS "hdr_rules!: String",
+                     p.size_bands      AS "size_bands!: String",
+                     p.resolution_pref AS "resolution_pref!: String",
+                     p.source_weights  AS "source_weights!: String",
+                     p.score_weights   AS "score_weights!: String"
+            FROM     episodes e
+            JOIN     seasons se  ON se.id = e.season_id
+            JOIN     series s    ON s.id = se.series_id
+            JOIN     roots r     ON r.id = s.root_id
+            JOIN     policies p  ON p.id = r.policy_id
+            WHERE    e.id = ?
+            "#,
+            episode_id
+        )
+        .fetch_optional(self.pool())
+        .await?;
+
+        let Some(row) = row else {
+            return Ok(None);
+        };
+
+        let overrides: OverridesJson = json("overrides", &row.overrides)?;
+        let policy = PolicyColumns {
+            id: row.policy_id,
+            name: row.policy_name,
+            required_audio: row.required_audio,
+            dub_blacklist: row.dub_blacklist,
+            hdr_rules: row.hdr_rules,
+            size_bands: row.size_bands,
+            resolution_pref: row.resolution_pref,
+            source_weights: row.source_weights,
+            score_weights: row.score_weights,
+        }
+        .to_policy()?;
+
+        Ok(Some(TitlePolicy {
             policy,
             overrides: TitleOverrides {
                 only_4k: overrides.only_4k,