feat(api): add movie CRUD API (#56)

crates/arr-api/Cargo.toml

@@ -7,16 +7,20 @@ repository.workspace = true
 publish = false
 
 [dependencies]
+arr-db = { workspace = true }
 axum = { workspace = true }
 reqwest = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
+sqlx = { workspace = true }
 tokio = { workspace = true }
+tracing = { workspace = true }
 utoipa = { workspace = true }
 utoipa-axum = { workspace = true }
 utoipa-scalar = { workspace = true }
 
 [dev-dependencies]
+tempfile = { workspace = true }
 wiremock = { workspace = true }
 
 [lints]

crates/arr-api/src/lib.rs

@@ -7,6 +7,7 @@
 //! to compile, and the gate in DESIGN.md §12 fails with it.
 
 mod health;
+mod movies;
 mod state;
 
 use axum::routing::get;
@@ -17,7 +18,8 @@ use utoipa_axum::routes;
 use utoipa_scalar::{Scalar, Servable};
 
 pub use health::{Check, Health, HealthReport, Status};
-pub use state::{AppState, Upstreams, DEFAULT_TMDB_URL};
+pub use movies::{Accepted, AttentionQueues, CreateMovie, ErrorBody, Movie, Release, UpdateMovie};
+pub use state::{AppState, 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.
@@ -35,13 +37,23 @@ pub const DOCS_PATH: &str = "/api/docs";
         description = "One service in place of Radarr and Sonarr. No authentication: \
                        the perimeter is the VPN (DESIGN.md §2).",
     ),
-    tags((name = "system", description = "Service health and metadata")),
+    tags(
+        (name = "system", description = "Service health and metadata"),
+        (name = "movies", description = "Movie library and actions")
+    ),
 )]
 struct ApiDoc;
 
 /// Every annotated route, still needing state.
 fn api_router() -> OpenApiRouter<AppState> {
-    OpenApiRouter::with_openapi(ApiDoc::openapi()).routes(routes!(health::health))
+    OpenApiRouter::with_openapi(ApiDoc::openapi())
+        .routes(routes!(health::health))
+        .routes(routes!(movies::list, movies::create))
+        .routes(routes!(movies::get, movies::update, movies::delete))
+        .routes(routes!(movies::search))
+        .routes(routes!(movies::releases))
+        .routes(routes!(movies::grab))
+        .routes(routes!(movies::attention))
 }
 
 /// The generated `OpenAPI` document.
@@ -242,6 +254,35 @@ mod tests {
             json["components"]["schemas"]["HealthReport"].is_object(),
             "the response body schema came along with it: {json}"
         );
+
+        for (path, method) in [
+            ("/api/movies", "get"),
+            ("/api/movies", "post"),
+            ("/api/movies/{movie_id}", "get"),
+            ("/api/movies/{movie_id}", "patch"),
+            ("/api/movies/{movie_id}", "delete"),
+            ("/api/movies/{movie_id}/search", "post"),
+            ("/api/movies/{movie_id}/releases", "get"),
+            ("/api/movies/{movie_id}/releases/{release_id}/grab", "post"),
+            ("/api/queues/attention", "get"),
+        ] {
+            assert!(
+                json["paths"][path][method].is_object(),
+                "missing {method} {path}"
+            );
+        }
+        for schema in [
+            "Movie",
+            "CreateMovie",
+            "UpdateMovie",
+            "Release",
+            "AttentionQueues",
+        ] {
+            assert!(
+                json["components"]["schemas"][schema].is_object(),
+                "{schema}"
+            );
+        }
     }
 
     #[tokio::test]

crates/arr-api/src/movies.rs

@@ -0,0 +1,647 @@
+use axum::extract::{Path, State};
+use axum::http::StatusCode;
+use axum::response::{IntoResponse, Response};
+use axum::Json;
+use serde::{Deserialize, Serialize};
+use utoipa::ToSchema;
+
+use crate::state::{AppState, MovieCommand};
+
+#[derive(Debug, Clone, Serialize, ToSchema)]
+pub struct Movie {
+    pub id: i64,
+    pub tmdb_id: i64,
+    pub title: String,
+    pub year: Option<i64>,
+    pub original_language: Option<String>,
+    pub root_id: i64,
+    pub wanted: bool,
+    pub overrides: serde_json::Value,
+    pub state: String,
+    pub blocked: bool,
+    pub search_attempts: i64,
+    pub last_searched_at: Option<String>,
+}
+
+#[derive(Debug, Deserialize, ToSchema)]
+pub struct CreateMovie {
+    pub tmdb_id: i64,
+    pub title: String,
+    pub year: Option<i64>,
+    pub original_language: Option<String>,
+    pub root_id: i64,
+    #[serde(default = "default_true")]
+    pub wanted: bool,
+    #[serde(default)]
+    pub blocked: bool,
+    #[serde(default = "empty_overrides")]
+    pub overrides: serde_json::Value,
+}
+
+#[derive(Debug, Deserialize, ToSchema)]
+pub struct UpdateMovie {
+    pub title: Option<String>,
+    pub year: Option<Option<i64>>,
+    pub original_language: Option<Option<String>>,
+    pub root_id: Option<i64>,
+    pub wanted: Option<bool>,
+    pub blocked: Option<bool>,
+    pub overrides: Option<serde_json::Value>,
+}
+
+#[derive(Debug, Clone, Serialize, ToSchema)]
+pub struct Release {
+    pub id: i64,
+    pub indexer_id: i64,
+    pub guid: String,
+    pub name: String,
+    pub size: i64,
+    pub seeders: Option<i64>,
+    pub publish_date: Option<String>,
+    pub download_url: String,
+    pub parsed: serde_json::Value,
+    pub score: Option<f64>,
+    pub verdict: Option<String>,
+    pub rejected_rule: Option<String>,
+}
+
+#[derive(Debug, Clone, Serialize, ToSchema)]
+pub struct AttentionQueues {
+    pub no_pt_source: Vec<Movie>,
+    pub needs_decision: Vec<Movie>,
+}
+
+#[derive(Debug, Clone, Serialize, ToSchema)]
+pub struct Accepted {
+    pub accepted: bool,
+}
+
+#[derive(Debug, Clone, Serialize, ToSchema)]
+pub struct ErrorBody {
+    pub error: String,
+}
+
+#[derive(Debug)]
+pub enum ApiError {
+    NotFound,
+    Conflict(String),
+    Invalid(String),
+    Unavailable,
+    Database(String),
+}
+
+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::Conflict(error) => (StatusCode::CONFLICT, error),
+            Self::Invalid(error) => (StatusCode::UNPROCESSABLE_ENTITY, error),
+            Self::Unavailable => (
+                StatusCode::SERVICE_UNAVAILABLE,
+                "database unavailable".into(),
+            ),
+            Self::Database(error) => {
+                tracing::error!(%error, "movie API database error");
+                (StatusCode::INTERNAL_SERVER_ERROR, "database error".into())
+            }
+        };
+        (status, Json(ErrorBody { error })).into_response()
+    }
+}
+
+impl From<sqlx::Error> for ApiError {
+    fn from(error: sqlx::Error) -> Self {
+        if matches!(error, sqlx::Error::RowNotFound) {
+            Self::NotFound
+        } else if let sqlx::Error::Database(database) = &error {
+            if database.is_unique_violation() || database.is_foreign_key_violation() {
+                return Self::Conflict(database.message().to_string());
+            }
+            Self::Database(error.to_string())
+        } else {
+            Self::Database(error.to_string())
+        }
+    }
+}
+
+fn default_true() -> bool {
+    true
+}
+
+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(())
+}
+
+fn pool(state: &AppState) -> Result<&sqlx::SqlitePool, ApiError> {
+    state
+        .database()
+        .map(arr_db::Db::pool)
+        .ok_or(ApiError::Unavailable)
+}
+
+async fn load_movie(state: &AppState, id: i64) -> Result<Movie, ApiError> {
+    Ok(sqlx::query_as!(Movie, r#"SELECT id AS "id!: i64", tmdb_id AS "tmdb_id!: i64", title AS "title!: String", year, original_language, root_id AS "root_id!: i64", wanted AS "wanted!: bool", overrides AS "overrides!: serde_json::Value", state AS "state!: String", blocked AS "blocked!: bool", search_attempts AS "search_attempts!: i64", last_searched_at FROM movies WHERE id = ?"#, id)
+        .fetch_one(pool(state)?)
+        .await?)
+}
+
+#[utoipa::path(
+    get, path = "/api/movies", tag = "movies",
+    responses(
+        (status = 200, body = [Movie]),
+        (status = 500, body = ErrorBody),
+        (status = 503, body = ErrorBody)
+    )
+)]
+pub async fn list(State(state): State<AppState>) -> Result<Json<Vec<Movie>>, ApiError> {
+    let movies = sqlx::query_as!(Movie, r#"SELECT id AS "id!: i64", tmdb_id AS "tmdb_id!: i64", title AS "title!: String", year, original_language, root_id AS "root_id!: i64", wanted AS "wanted!: bool", overrides AS "overrides!: serde_json::Value", state AS "state!: String", blocked AS "blocked!: bool", search_attempts AS "search_attempts!: i64", last_searched_at FROM movies ORDER BY title, year, id"#)
+        .fetch_all(pool(&state)?)
+        .await?;
+    Ok(Json(movies))
+}
+
+#[utoipa::path(
+    post, path = "/api/movies", tag = "movies", request_body = CreateMovie,
+    responses(
+        (status = 201, body = Movie),
+        (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<CreateMovie>,
+) -> Result<(StatusCode, Json<Movie>), 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)?;
+    let movie_root = sqlx::query_scalar!(
+        "SELECT EXISTS(SELECT 1 FROM roots WHERE id = ? AND kind = 'movie') AS 'exists!: bool'",
+        input.root_id
+    )
+    .fetch_one(pool(&state)?)
+    .await?;
+    if !movie_root {
+        return Err(ApiError::Invalid("root_id must name a movie root".into()));
+    }
+    let overrides = serde_json::to_string(&input.overrides)
+        .map_err(|error| ApiError::Invalid(error.to_string()))?;
+    let title = input.title.trim();
+    let result = sqlx::query!(
+        "INSERT INTO movies (tmdb_id, title, year, original_language, root_id, wanted, blocked, overrides) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+        input.tmdb_id, title, input.year, input.original_language, input.root_id,
+        input.wanted, input.blocked, overrides
+    )
+    .execute(pool(&state)?)
+    .await?;
+    Ok((
+        StatusCode::CREATED,
+        Json(load_movie(&state, result.last_insert_rowid()).await?),
+    ))
+}
+
+#[utoipa::path(
+    get, path = "/api/movies/{movie_id}", tag = "movies",
+    params(("movie_id" = i64, Path, description = "Movie row id")),
+    responses(
+        (status = 200, body = Movie),
+        (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<Movie>, ApiError> {
+    Ok(Json(load_movie(&state, id).await?))
+}
+
+#[utoipa::path(
+    patch, path = "/api/movies/{movie_id}", tag = "movies", request_body = UpdateMovie,
+    params(("movie_id" = i64, Path, description = "Movie row id")),
+    responses(
+        (status = 200, body = Movie),
+        (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<UpdateMovie>,
+) -> Result<Json<Movie>, ApiError> {
+    let current = load_movie(&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() {
+        let movie_root = sqlx::query_scalar!(
+            "SELECT EXISTS(SELECT 1 FROM roots WHERE id = ? AND kind = 'movie') AS 'exists!: bool'",
+            root_id
+        )
+        .fetch_one(pool(&state)?)
+        .await?;
+        if !movie_root {
+            return Err(ApiError::Invalid("root_id must name a movie root".into()));
+        }
+    }
+    let wanted = input.wanted.unwrap_or(current.wanted);
+    let blocked = input.blocked.unwrap_or(current.blocked);
+    sqlx::query!("UPDATE movies SET title = ?, year = ?, original_language = ?, root_id = ?, wanted = ?, blocked = ?, overrides = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') WHERE id = ?", title, year, original_language, root_id, wanted, blocked, overrides, id)
+        .execute(pool(&state)?)
+        .await?;
+    Ok(Json(load_movie(&state, id).await?))
+}
+
+#[utoipa::path(
+    delete, path = "/api/movies/{movie_id}", tag = "movies",
+    params(("movie_id" = i64, Path, description = "Movie 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> {
+    let result = sqlx::query!("DELETE FROM movies WHERE id = ?", id)
+        .execute(pool(&state)?)
+        .await?;
+    if result.rows_affected() == 0 {
+        return Err(ApiError::NotFound);
+    }
+    Ok(StatusCode::NO_CONTENT)
+}
+
+#[utoipa::path(
+    post, path = "/api/movies/{movie_id}/search", tag = "movies",
+    params(("movie_id" = i64, Path, description = "Movie 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(
+    State(state): State<AppState>,
+    Path(id): Path<i64>,
+) -> Result<(StatusCode, Json<Accepted>), ApiError> {
+    let movie = load_movie(&state, id).await?;
+    if movie.blocked {
+        return Err(ApiError::Conflict("movie is blocked".into()));
+    }
+    state
+        .send_movie_command(MovieCommand::Search { movie_id: id })
+        .map_err(|_| ApiError::Unavailable)?;
+    Ok((StatusCode::ACCEPTED, Json(Accepted { accepted: true })))
+}
+
+#[utoipa::path(
+    get, path = "/api/movies/{movie_id}/releases", tag = "movies",
+    params(("movie_id" = i64, Path, description = "Movie row id")),
+    responses(
+        (status = 200, body = [Release]),
+        (status = 404, body = ErrorBody),
+        (status = 500, body = ErrorBody),
+        (status = 503, body = ErrorBody)
+    )
+)]
+pub async fn releases(
+    State(state): State<AppState>,
+    Path(id): Path<i64>,
+) -> Result<Json<Vec<Release>>, ApiError> {
+    load_movie(&state, id).await?;
+    let 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 movie_releases mr ON mr.release_id = r.id WHERE mr.movie_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?;
+    Ok(Json(releases))
+}
+
+#[utoipa::path(
+    post, path = "/api/movies/{movie_id}/releases/{release_id}/grab", tag = "movies",
+    params(("movie_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(
+    State(state): State<AppState>,
+    Path((movie_id, release_id)): Path<(i64, i64)>,
+) -> Result<(StatusCode, Json<Accepted>), ApiError> {
+    let exists = sqlx::query_scalar!("SELECT EXISTS(SELECT 1 FROM movie_releases WHERE movie_id = ? AND release_id = ?) AS 'exists!: bool'", movie_id, release_id)
+    .fetch_one(pool(&state)?)
+    .await?;
+    if !exists {
+        return Err(ApiError::NotFound);
+    }
+    state
+        .send_movie_command(MovieCommand::Grab {
+            movie_id,
+            release_id,
+        })
+        .map_err(|_| ApiError::Unavailable)?;
+    Ok((StatusCode::ACCEPTED, Json(Accepted { accepted: true })))
+}
+
+#[utoipa::path(
+    get, path = "/api/queues/attention", tag = "movies",
+    responses(
+        (status = 200, body = AttentionQueues),
+        (status = 500, body = ErrorBody),
+        (status = 503, body = ErrorBody)
+    )
+)]
+pub async fn attention(State(state): State<AppState>) -> Result<Json<AttentionQueues>, ApiError> {
+    let no_pt_source = sqlx::query_as!(Movie, r#"SELECT id AS "id!: i64", tmdb_id AS "tmdb_id!: i64", title AS "title!: String", year, original_language, root_id AS "root_id!: i64", wanted AS "wanted!: bool", overrides AS "overrides!: serde_json::Value", state AS "state!: String", blocked AS "blocked!: bool", search_attempts AS "search_attempts!: i64", last_searched_at FROM movies WHERE id IN (SELECT m.id FROM movies m JOIN roots root ON root.id = m.root_id WHERE root.audience = 'kids' AND m.wanted = 1 AND m.blocked = 0 AND m.state = 'missing' AND m.search_attempts > 0 AND NOT EXISTS (SELECT 1 FROM movie_releases mr JOIN releases r ON r.id = mr.release_id WHERE mr.movie_id = m.id AND r.verdict IN ('eligible', 'waived'))) ORDER BY title"#)
+        .fetch_all(pool(&state)?)
+        .await?;
+    let needs_decision = sqlx::query_as!(Movie, r#"SELECT id AS "id!: i64", tmdb_id AS "tmdb_id!: i64", title AS "title!: String", year, original_language, root_id AS "root_id!: i64", wanted AS "wanted!: bool", overrides AS "overrides!: serde_json::Value", state AS "state!: String", blocked AS "blocked!: bool", search_attempts AS "search_attempts!: i64", last_searched_at FROM movies WHERE (SELECT count(DISTINCT g.release_id) FROM grabs g WHERE g.target_kind = 'movie' AND g.target_id = movies.id AND g.state = 'failed') >= 2 ORDER BY title"#)
+        .fetch_all(pool(&state)?)
+        .await?;
+    Ok(Json(AttentionQueues {
+        no_pt_source,
+        needs_decision,
+    }))
+}
+
+#[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 add_movie(base: &str, tmdb_id: i64, root_id: i64) -> serde_json::Value {
+        let response = reqwest::Client::new()
+            .post(format!("{base}/api/movies"))
+            .json(&serde_json::json!({
+                "tmdb_id": tmdb_id, "title": "Dune Part Two", "year": 2024,
+                "original_language": "en", "root_id": root_id,
+                "overrides": {"only_4k": true}
+            }))
+            .send()
+            .await
+            .expect("create movie");
+        assert_eq!(response.status(), StatusCode::CREATED);
+        response.json().await.expect("movie json")
+    }
+
+    #[tokio::test]
+    async fn crud_preserves_intent_and_overrides() {
+        let (_dir, _state, base) = application().await;
+        let movie = add_movie(&base, 693_134, 2).await;
+        let id = movie["id"].as_i64().expect("id");
+        assert_eq!(movie["wanted"], true);
+        assert_eq!(movie["overrides"]["only_4k"], true);
+
+        let updated: serde_json::Value = reqwest::Client::new()
+            .patch(format!("{base}/api/movies/{id}"))
+            .json(&serde_json::json!({
+                "wanted": false, "blocked": true,
+                "overrides": {"allow_english_audio": true}
+            }))
+            .send()
+            .await
+            .expect("update")
+            .json()
+            .await
+            .expect("updated json");
+        assert_eq!(updated["wanted"], false);
+        assert_eq!(updated["blocked"], true);
+        assert_eq!(updated["overrides"]["allow_english_audio"], true);
+
+        let listed: Vec<serde_json::Value> = reqwest::get(format!("{base}/api/movies"))
+            .await
+            .expect("list")
+            .json()
+            .await
+            .expect("list json");
+        assert_eq!(listed.len(), 1);
+
+        let deleted = reqwest::Client::new()
+            .delete(format!("{base}/api/movies/{id}"))
+            .send()
+            .await
+            .expect("delete");
+        assert_eq!(deleted.status(), StatusCode::NO_CONTENT);
+        assert_eq!(
+            reqwest::get(format!("{base}/api/movies/{id}"))
+                .await
+                .expect("get deleted")
+                .status(),
+            StatusCode::NOT_FOUND
+        );
+    }
+
+    #[tokio::test]
+    async fn release_actions_are_scoped_to_the_movie() {
+        let (_dir, state, base) = application().await;
+        let movie = add_movie(&base, 693_134, 2).await;
+        let movie_id = movie["id"].as_i64().expect("id");
+        let pool = state.database().expect("database").pool();
+        let release_id = sqlx::query("INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed, score, verdict) VALUES (7, 'guid', 'release', 1000, 'url', '{}', 42, 'eligible')")
+            .execute(pool).await.expect("release").last_insert_rowid();
+        sqlx::query("INSERT INTO movie_releases (movie_id, release_id) VALUES (?, ?)")
+            .bind(movie_id)
+            .bind(release_id)
+            .execute(pool)
+            .await
+            .expect("association");
+        let releases: Vec<serde_json::Value> =
+            reqwest::get(format!("{base}/api/movies/{movie_id}/releases"))
+                .await
+                .expect("releases")
+                .json()
+                .await
+                .expect("release json");
+        assert_eq!(releases[0]["verdict"], "eligible");
+
+        let response = reqwest::Client::new()
+            .post(format!(
+                "{base}/api/movies/{movie_id}/releases/{release_id}/grab"
+            ))
+            .send()
+            .await
+            .expect("grab");
+        assert_eq!(response.status(), StatusCode::ACCEPTED);
+        assert_eq!(
+            state.next_movie_command().await.expect("command"),
+            MovieCommand::Grab {
+                movie_id,
+                release_id
+            }
+        );
+    }
+
+    #[tokio::test]
+    async fn a_full_action_queue_returns_service_unavailable() {
+        let (_dir, state, base) = application().await;
+        let movie = add_movie(&base, 693_134, 2).await;
+        let movie_id = movie["id"].as_i64().expect("id");
+        for _ in 0..64 {
+            state
+                .send_movie_command(MovieCommand::Search { movie_id })
+                .expect("queue has capacity");
+        }
+
+        let response = reqwest::Client::new()
+            .post(format!("{base}/api/movies/{movie_id}/search"))
+            .send()
+            .await
+            .expect("search");
+        assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
+    }
+
+    #[tokio::test]
+    async fn attention_queues_follow_search_and_failure_state() {
+        let (_dir, state, base) = application().await;
+        let pool = state.database().expect("database").pool();
+        let root_id: i64 = sqlx::query_scalar("SELECT id FROM roots WHERE audience = 'kids'")
+            .fetch_one(pool)
+            .await
+            .expect("kids root");
+        let movie = add_movie(&base, 82728, root_id).await;
+        let movie_id = movie["id"].as_i64().expect("id");
+        sqlx::query("UPDATE movies SET search_attempts = 1 WHERE id = ?")
+            .bind(movie_id)
+            .execute(pool)
+            .await
+            .expect("record search");
+
+        let mut release_ids = Vec::new();
+        for suffix in ["a", "b"] {
+            let release_id = sqlx::query("INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed, verdict, rejected_rule) VALUES (1, ?, ?, 1, 'url', '{}', 'rejected', 'required_audio')")
+                .bind(suffix)
+                .bind(suffix)
+                .execute(pool)
+                .await
+                .expect("release")
+                .last_insert_rowid();
+            release_ids.push(release_id);
+            sqlx::query("INSERT INTO grabs (release_id, target_kind, target_id, infohash, state) VALUES (?, 'movie', ?, ?, 'failed')")
+                .bind(release_id)
+                .bind(movie_id)
+                .bind(format!("hash-{suffix}"))
+                .execute(pool)
+                .await
+                .expect("failed grab");
+        }
+
+        let queues: serde_json::Value = reqwest::get(format!("{base}/api/queues/attention"))
+            .await
+            .expect("queues")
+            .json()
+            .await
+            .expect("queues json");
+        assert_eq!(queues["no_pt_source"][0]["id"], movie_id);
+        assert_eq!(queues["needs_decision"][0]["id"], movie_id);
+
+        sqlx::query("UPDATE movies SET blocked = 1 WHERE id = ?")
+            .bind(movie_id)
+            .execute(pool)
+            .await
+            .expect("block movie");
+        let queues: serde_json::Value = reqwest::get(format!("{base}/api/queues/attention"))
+            .await
+            .expect("blocked queues")
+            .json()
+            .await
+            .expect("blocked queues json");
+        assert_eq!(queues["no_pt_source"].as_array().map(Vec::len), Some(0));
+        let search = reqwest::Client::new()
+            .post(format!("{base}/api/movies/{movie_id}/search"))
+            .send()
+            .await
+            .expect("blocked search");
+        assert_eq!(search.status(), StatusCode::CONFLICT);
+
+        sqlx::query("UPDATE movies SET blocked = 0 WHERE id = ?")
+            .bind(movie_id)
+            .execute(pool)
+            .await
+            .expect("unblock movie");
+
+        sqlx::query("UPDATE releases SET verdict = 'eligible', rejected_rule = NULL WHERE id = ?")
+            .bind(release_ids[0])
+            .execute(pool)
+            .await
+            .expect("eligible release");
+        sqlx::query("INSERT INTO movie_releases (movie_id, release_id) VALUES (?, ?)")
+            .bind(movie_id)
+            .bind(release_ids[0])
+            .execute(pool)
+            .await
+            .expect("associate release");
+
+        let queues: serde_json::Value = reqwest::get(format!("{base}/api/queues/attention"))
+            .await
+            .expect("queues")
+            .json()
+            .await
+            .expect("queues json");
+        assert_eq!(queues["no_pt_source"].as_array().map(Vec::len), Some(0));
+        assert_eq!(queues["needs_decision"][0]["id"], movie_id);
+    }
+}

crates/arr-api/src/state.rs

@@ -4,6 +4,9 @@
 use std::sync::Arc;
 use std::time::Duration;
 
+use arr_db::Db;
+use tokio::sync::mpsc;
+
 /// The TMDB API root. Not a bootstrap setting (DESIGN.md §10) — only the key
 /// is configurable, so this is a constant that tests point elsewhere.
 pub const DEFAULT_TMDB_URL: &str = "https://api.themoviedb.org/3";
@@ -63,6 +66,16 @@ impl Upstreams {
 pub struct AppState {
     http: reqwest::Client,
     upstreams: Arc<Upstreams>,
+    database: Option<Db>,
+    movie_commands: mpsc::Sender<MovieCommand>,
+    pending_movie_commands: Arc<tokio::sync::Mutex<mpsc::Receiver<MovieCommand>>>,
+}
+
+/// Work explicitly requested through the movie API.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum MovieCommand {
+    Search { movie_id: i64 },
+    Grab { movie_id: i64, release_id: i64 },
 }
 
 impl AppState {
@@ -73,12 +86,32 @@ impl AppState {
     /// If the TLS backend cannot be initialised.
     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);
         Ok(Self {
             http,
             upstreams: Arc::new(upstreams),
+            database: None,
+            movie_commands,
+            pending_movie_commands: Arc::new(tokio::sync::Mutex::new(pending_movie_commands)),
         })
     }
 
+    /// Attach the migrated application database.
+    #[must_use]
+    pub fn with_database(mut self, database: Db) -> Self {
+        self.database = Some(database);
+        self
+    }
+
+    /// Wait for the next manual movie action in the daemon's reconcile loop.
+    ///
+    /// # Errors
+    ///
+    /// If every sender has been dropped.
+    pub async fn next_movie_command(&self) -> Option<MovieCommand> {
+        self.pending_movie_commands.lock().await.recv().await
+    }
+
     pub(crate) fn http(&self) -> &reqwest::Client {
         &self.http
     }
@@ -86,4 +119,15 @@ impl AppState {
     pub(crate) fn upstreams(&self) -> &Upstreams {
         &self.upstreams
     }
+
+    pub(crate) fn database(&self) -> Option<&Db> {
+        self.database.as_ref()
+    }
+
+    pub(crate) fn send_movie_command(
+        &self,
+        command: MovieCommand,
+    ) -> Result<(), mpsc::error::TrySendError<MovieCommand>> {
+        self.movie_commands.try_send(command)
+    }
 }

crates/arr-daemon/Cargo.toml

@@ -12,9 +12,11 @@ path = "src/main.rs"
 
 [dependencies]
 arr-api = { workspace = true }
+arr-db = { workspace = true }
 axum = { workspace = true }
 reqwest = { workspace = true }
 serde = { workspace = true }
+sqlx = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
 toml = { workspace = true }

crates/arr-daemon/src/main.rs

@@ -5,6 +5,7 @@ mod config;
 use std::process::ExitCode;
 
 use arr_api::{AppState, Upstreams};
+use arr_db::Db;
 use config::Config;
 use tower_http::trace::TraceLayer;
 
@@ -54,6 +55,10 @@ enum Error {
     Config(#[from] config::ConfigError),
     #[error("http client: {0}")]
     HttpClient(#[from] reqwest::Error),
+    #[error("database: {0}")]
+    Database(#[from] sqlx::Error),
+    #[error("database migration: {0}")]
+    Migration(#[from] sqlx::migrate::MigrateError),
     #[error("bind {addr}: {source}")]
     Bind {
         addr: std::net::SocketAddr,
@@ -65,12 +70,15 @@ enum Error {
 
 async fn run() -> Result<(), Error> {
     let config = Config::load()?;
+    let database = Db::connect(&config.database_path).await?;
+    database.migrate().await?;
 
     let state = AppState::new(
         Upstreams::new(config.prowlarr_url, config.transmission_url)
             .with_prowlarr_api_key(config.prowlarr_api_key)
             .with_tmdb_api_key(config.tmdb_api_key),
-    )?;
+    )?
+    .with_database(database);
 
     let app = arr_api::router(state).layer(TraceLayer::new_for_http());
 

crates/arr-db/migrations/0003_movie_releases.sql

@@ -0,0 +1,9 @@
+-- A release can be returned for more than one title search. Keep the cached
+-- release itself unique by indexer/guid and record each movie association.
+CREATE TABLE movie_releases (
+    movie_id   INTEGER NOT NULL REFERENCES movies (id) ON DELETE CASCADE,
+    release_id INTEGER NOT NULL REFERENCES releases (id) ON DELETE CASCADE,
+    PRIMARY KEY (movie_id, release_id)
+) STRICT;
+
+CREATE INDEX movie_releases_release ON movie_releases (release_id);