From fab619483530d89a6b109efe6ebb97dc31cad1ba Mon Sep 17 00:00:00 2001 From: Miguel Palhas Date: Sat, 22 Aug 2026 20:21:59 +0100 Subject: [PATCH 1/4] feat(api): add movie CRUD endpoints --- ...80154a43f96a79844c6c3a399463ad557e0eb.json | 86 ++++ ...2d5d455e8e652531986724b9109c6d839594f.json | 86 ++++ ...590f840b75f579aea4fa5ba0f58ffa00d8b3e.json | 12 + ...0b41857754a136548836fdba8fbd63a8412f7.json | 86 ++++ ...02fe0324515951be0bddb19b5acf405c8be4f.json | 86 ++++ ...570f6f8f3dfbf8ae93bc515b087872044dffe.json | 20 + ...b6d54cbad525b1d1d63c2a1495c24af44db7f.json | 86 ++++ ...e3f4c19cb52dcfff66772145cd80c53c16883.json | 12 + ...94a8cd270a4dedacdfd07bff37c8f455934d9.json | 12 + ...1cbe1aea538f04e31b74f1bca0e7bd00c7248.json | 20 + Cargo.lock | 6 + crates/arr-api/Cargo.toml | 4 + crates/arr-api/src/lib.rs | 47 +- crates/arr-api/src/movies.rs | 484 ++++++++++++++++++ crates/arr-api/src/state.rs | 47 ++ crates/arr-daemon/Cargo.toml | 2 + crates/arr-daemon/src/main.rs | 10 +- .../arr-db/migrations/0003_movie_releases.sql | 9 + 18 files changed, 1111 insertions(+), 4 deletions(-) create mode 100644 .sqlx/query-1f82542cde76554edd497ce056c80154a43f96a79844c6c3a399463ad557e0eb.json create mode 100644 .sqlx/query-546b4c3047094fe0874a1d28d802d5d455e8e652531986724b9109c6d839594f.json create mode 100644 .sqlx/query-5b9a8489e7223e9aa2929f68257590f840b75f579aea4fa5ba0f58ffa00d8b3e.json create mode 100644 .sqlx/query-67b52bc6c6b748a09dbf5e903990b41857754a136548836fdba8fbd63a8412f7.json create mode 100644 .sqlx/query-8eaf4d28651be8e464cb68037c202fe0324515951be0bddb19b5acf405c8be4f.json create mode 100644 .sqlx/query-9de2cc5614917f11ada32f9ff0b570f6f8f3dfbf8ae93bc515b087872044dffe.json create mode 100644 .sqlx/query-aebbabd41e2086ac37dbd9d2151b6d54cbad525b1d1d63c2a1495c24af44db7f.json create mode 100644 .sqlx/query-e431381ad41c1c2f7b9c89509d5e3f4c19cb52dcfff66772145cd80c53c16883.json create mode 100644 .sqlx/query-e476863873e5574440db796925a94a8cd270a4dedacdfd07bff37c8f455934d9.json create mode 100644 .sqlx/query-ee86c3a4d78ef8ec0ca71343c5f1cbe1aea538f04e31b74f1bca0e7bd00c7248.json create mode 100644 crates/arr-api/src/movies.rs create mode 100644 crates/arr-db/migrations/0003_movie_releases.sql diff --git a/.sqlx/query-1f82542cde76554edd497ce056c80154a43f96a79844c6c3a399463ad557e0eb.json b/.sqlx/query-1f82542cde76554edd497ce056c80154a43f96a79844c6c3a399463ad557e0eb.json new file mode 100644 index 0000000..07c918f --- /dev/null +++ b/.sqlx/query-1f82542cde76554edd497ce056c80154a43f96a79844c6c3a399463ad557e0eb.json @@ -0,0 +1,86 @@ +{ + "db_name": "SQLite", + "query": "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.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", + "describe": { + "columns": [ + { + "name": "id!: i64", + "ordinal": 0, + "type_info": "Integer" + }, + { + "name": "tmdb_id!: i64", + "ordinal": 1, + "type_info": "Integer" + }, + { + "name": "title!: String", + "ordinal": 2, + "type_info": "Text" + }, + { + "name": "year", + "ordinal": 3, + "type_info": "Integer" + }, + { + "name": "original_language", + "ordinal": 4, + "type_info": "Text" + }, + { + "name": "root_id!: i64", + "ordinal": 5, + "type_info": "Integer" + }, + { + "name": "wanted!: bool", + "ordinal": 6, + "type_info": "Integer" + }, + { + "name": "overrides!: serde_json::Value", + "ordinal": 7, + "type_info": "Text" + }, + { + "name": "state!: String", + "ordinal": 8, + "type_info": "Text" + }, + { + "name": "blocked!: bool", + "ordinal": 9, + "type_info": "Integer" + }, + { + "name": "search_attempts!: i64", + "ordinal": 10, + "type_info": "Integer" + }, + { + "name": "last_searched_at", + "ordinal": 11, + "type_info": "Text" + } + ], + "parameters": { + "Right": 0 + }, + "nullable": [ + true, + false, + false, + true, + true, + false, + false, + false, + false, + false, + false, + true + ] + }, + "hash": "1f82542cde76554edd497ce056c80154a43f96a79844c6c3a399463ad557e0eb" +} diff --git a/.sqlx/query-546b4c3047094fe0874a1d28d802d5d455e8e652531986724b9109c6d839594f.json b/.sqlx/query-546b4c3047094fe0874a1d28d802d5d455e8e652531986724b9109c6d839594f.json new file mode 100644 index 0000000..d94373f --- /dev/null +++ b/.sqlx/query-546b4c3047094fe0874a1d28d802d5d455e8e652531986724b9109c6d839594f.json @@ -0,0 +1,86 @@ +{ + "db_name": "SQLite", + "query": "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 = ?", + "describe": { + "columns": [ + { + "name": "id!: i64", + "ordinal": 0, + "type_info": "Integer" + }, + { + "name": "tmdb_id!: i64", + "ordinal": 1, + "type_info": "Integer" + }, + { + "name": "title!: String", + "ordinal": 2, + "type_info": "Text" + }, + { + "name": "year", + "ordinal": 3, + "type_info": "Integer" + }, + { + "name": "original_language", + "ordinal": 4, + "type_info": "Text" + }, + { + "name": "root_id!: i64", + "ordinal": 5, + "type_info": "Integer" + }, + { + "name": "wanted!: bool", + "ordinal": 6, + "type_info": "Integer" + }, + { + "name": "overrides!: serde_json::Value", + "ordinal": 7, + "type_info": "Text" + }, + { + "name": "state!: String", + "ordinal": 8, + "type_info": "Text" + }, + { + "name": "blocked!: bool", + "ordinal": 9, + "type_info": "Integer" + }, + { + "name": "search_attempts!: i64", + "ordinal": 10, + "type_info": "Integer" + }, + { + "name": "last_searched_at", + "ordinal": 11, + "type_info": "Text" + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + false, + false, + false, + true, + true, + false, + false, + false, + false, + false, + false, + true + ] + }, + "hash": "546b4c3047094fe0874a1d28d802d5d455e8e652531986724b9109c6d839594f" +} diff --git a/.sqlx/query-5b9a8489e7223e9aa2929f68257590f840b75f579aea4fa5ba0f58ffa00d8b3e.json b/.sqlx/query-5b9a8489e7223e9aa2929f68257590f840b75f579aea4fa5ba0f58ffa00d8b3e.json new file mode 100644 index 0000000..ee88abf --- /dev/null +++ b/.sqlx/query-5b9a8489e7223e9aa2929f68257590f840b75f579aea4fa5ba0f58ffa00d8b3e.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "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 = ?", + "describe": { + "columns": [], + "parameters": { + "Right": 8 + }, + "nullable": [] + }, + "hash": "5b9a8489e7223e9aa2929f68257590f840b75f579aea4fa5ba0f58ffa00d8b3e" +} diff --git a/.sqlx/query-67b52bc6c6b748a09dbf5e903990b41857754a136548836fdba8fbd63a8412f7.json b/.sqlx/query-67b52bc6c6b748a09dbf5e903990b41857754a136548836fdba8fbd63a8412f7.json new file mode 100644 index 0000000..f98073c --- /dev/null +++ b/.sqlx/query-67b52bc6c6b748a09dbf5e903990b41857754a136548836fdba8fbd63a8412f7.json @@ -0,0 +1,86 @@ +{ + "db_name": "SQLite", + "query": "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", + "describe": { + "columns": [ + { + "name": "id!: i64", + "ordinal": 0, + "type_info": "Integer" + }, + { + "name": "tmdb_id!: i64", + "ordinal": 1, + "type_info": "Integer" + }, + { + "name": "title!: String", + "ordinal": 2, + "type_info": "Text" + }, + { + "name": "year", + "ordinal": 3, + "type_info": "Integer" + }, + { + "name": "original_language", + "ordinal": 4, + "type_info": "Text" + }, + { + "name": "root_id!: i64", + "ordinal": 5, + "type_info": "Integer" + }, + { + "name": "wanted!: bool", + "ordinal": 6, + "type_info": "Integer" + }, + { + "name": "overrides!: serde_json::Value", + "ordinal": 7, + "type_info": "Text" + }, + { + "name": "state!: String", + "ordinal": 8, + "type_info": "Text" + }, + { + "name": "blocked!: bool", + "ordinal": 9, + "type_info": "Integer" + }, + { + "name": "search_attempts!: i64", + "ordinal": 10, + "type_info": "Integer" + }, + { + "name": "last_searched_at", + "ordinal": 11, + "type_info": "Text" + } + ], + "parameters": { + "Right": 0 + }, + "nullable": [ + false, + false, + false, + true, + true, + false, + false, + false, + false, + false, + false, + true + ] + }, + "hash": "67b52bc6c6b748a09dbf5e903990b41857754a136548836fdba8fbd63a8412f7" +} diff --git a/.sqlx/query-8eaf4d28651be8e464cb68037c202fe0324515951be0bddb19b5acf405c8be4f.json b/.sqlx/query-8eaf4d28651be8e464cb68037c202fe0324515951be0bddb19b5acf405c8be4f.json new file mode 100644 index 0000000..9137cec --- /dev/null +++ b/.sqlx/query-8eaf4d28651be8e464cb68037c202fe0324515951be0bddb19b5acf405c8be4f.json @@ -0,0 +1,86 @@ +{ + "db_name": "SQLite", + "query": "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", + "describe": { + "columns": [ + { + "name": "id!: i64", + "ordinal": 0, + "type_info": "Integer" + }, + { + "name": "tmdb_id!: i64", + "ordinal": 1, + "type_info": "Integer" + }, + { + "name": "title!: String", + "ordinal": 2, + "type_info": "Text" + }, + { + "name": "year", + "ordinal": 3, + "type_info": "Integer" + }, + { + "name": "original_language", + "ordinal": 4, + "type_info": "Text" + }, + { + "name": "root_id!: i64", + "ordinal": 5, + "type_info": "Integer" + }, + { + "name": "wanted!: bool", + "ordinal": 6, + "type_info": "Integer" + }, + { + "name": "overrides!: serde_json::Value", + "ordinal": 7, + "type_info": "Text" + }, + { + "name": "state!: String", + "ordinal": 8, + "type_info": "Text" + }, + { + "name": "blocked!: bool", + "ordinal": 9, + "type_info": "Integer" + }, + { + "name": "search_attempts!: i64", + "ordinal": 10, + "type_info": "Integer" + }, + { + "name": "last_searched_at", + "ordinal": 11, + "type_info": "Text" + } + ], + "parameters": { + "Right": 0 + }, + "nullable": [ + false, + false, + false, + true, + true, + false, + false, + false, + false, + false, + false, + true + ] + }, + "hash": "8eaf4d28651be8e464cb68037c202fe0324515951be0bddb19b5acf405c8be4f" +} diff --git a/.sqlx/query-9de2cc5614917f11ada32f9ff0b570f6f8f3dfbf8ae93bc515b087872044dffe.json b/.sqlx/query-9de2cc5614917f11ada32f9ff0b570f6f8f3dfbf8ae93bc515b087872044dffe.json new file mode 100644 index 0000000..e676ff2 --- /dev/null +++ b/.sqlx/query-9de2cc5614917f11ada32f9ff0b570f6f8f3dfbf8ae93bc515b087872044dffe.json @@ -0,0 +1,20 @@ +{ + "db_name": "SQLite", + "query": "SELECT EXISTS(SELECT 1 FROM movie_releases WHERE movie_id = ? AND release_id = ?) AS 'exists!: bool'", + "describe": { + "columns": [ + { + "name": "exists!: bool", + "ordinal": 0, + "type_info": "Integer" + } + ], + "parameters": { + "Right": 2 + }, + "nullable": [ + false + ] + }, + "hash": "9de2cc5614917f11ada32f9ff0b570f6f8f3dfbf8ae93bc515b087872044dffe" +} diff --git a/.sqlx/query-aebbabd41e2086ac37dbd9d2151b6d54cbad525b1d1d63c2a1495c24af44db7f.json b/.sqlx/query-aebbabd41e2086ac37dbd9d2151b6d54cbad525b1d1d63c2a1495c24af44db7f.json new file mode 100644 index 0000000..5e79f1e --- /dev/null +++ b/.sqlx/query-aebbabd41e2086ac37dbd9d2151b6d54cbad525b1d1d63c2a1495c24af44db7f.json @@ -0,0 +1,86 @@ +{ + "db_name": "SQLite", + "query": "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", + "describe": { + "columns": [ + { + "name": "id!: i64", + "ordinal": 0, + "type_info": "Integer" + }, + { + "name": "indexer_id!: i64", + "ordinal": 1, + "type_info": "Integer" + }, + { + "name": "guid!: String", + "ordinal": 2, + "type_info": "Text" + }, + { + "name": "name!: String", + "ordinal": 3, + "type_info": "Text" + }, + { + "name": "size!: i64", + "ordinal": 4, + "type_info": "Integer" + }, + { + "name": "seeders", + "ordinal": 5, + "type_info": "Integer" + }, + { + "name": "publish_date", + "ordinal": 6, + "type_info": "Text" + }, + { + "name": "download_url!: String", + "ordinal": 7, + "type_info": "Text" + }, + { + "name": "parsed!: serde_json::Value", + "ordinal": 8, + "type_info": "Text" + }, + { + "name": "score", + "ordinal": 9, + "type_info": "Float" + }, + { + "name": "verdict", + "ordinal": 10, + "type_info": "Text" + }, + { + "name": "rejected_rule", + "ordinal": 11, + "type_info": "Text" + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + true, + false, + false, + false, + false, + true, + true, + false, + false, + true, + true, + true + ] + }, + "hash": "aebbabd41e2086ac37dbd9d2151b6d54cbad525b1d1d63c2a1495c24af44db7f" +} diff --git a/.sqlx/query-e431381ad41c1c2f7b9c89509d5e3f4c19cb52dcfff66772145cd80c53c16883.json b/.sqlx/query-e431381ad41c1c2f7b9c89509d5e3f4c19cb52dcfff66772145cd80c53c16883.json new file mode 100644 index 0000000..c2562bf --- /dev/null +++ b/.sqlx/query-e431381ad41c1c2f7b9c89509d5e3f4c19cb52dcfff66772145cd80c53c16883.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "DELETE FROM movies WHERE id = ?", + "describe": { + "columns": [], + "parameters": { + "Right": 1 + }, + "nullable": [] + }, + "hash": "e431381ad41c1c2f7b9c89509d5e3f4c19cb52dcfff66772145cd80c53c16883" +} diff --git a/.sqlx/query-e476863873e5574440db796925a94a8cd270a4dedacdfd07bff37c8f455934d9.json b/.sqlx/query-e476863873e5574440db796925a94a8cd270a4dedacdfd07bff37c8f455934d9.json new file mode 100644 index 0000000..547cfaf --- /dev/null +++ b/.sqlx/query-e476863873e5574440db796925a94a8cd270a4dedacdfd07bff37c8f455934d9.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "INSERT INTO movies (tmdb_id, title, year, original_language, root_id, wanted, blocked, overrides) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + "describe": { + "columns": [], + "parameters": { + "Right": 8 + }, + "nullable": [] + }, + "hash": "e476863873e5574440db796925a94a8cd270a4dedacdfd07bff37c8f455934d9" +} diff --git a/.sqlx/query-ee86c3a4d78ef8ec0ca71343c5f1cbe1aea538f04e31b74f1bca0e7bd00c7248.json b/.sqlx/query-ee86c3a4d78ef8ec0ca71343c5f1cbe1aea538f04e31b74f1bca0e7bd00c7248.json new file mode 100644 index 0000000..bfc5699 --- /dev/null +++ b/.sqlx/query-ee86c3a4d78ef8ec0ca71343c5f1cbe1aea538f04e31b74f1bca0e7bd00c7248.json @@ -0,0 +1,20 @@ +{ + "db_name": "SQLite", + "query": "SELECT EXISTS(SELECT 1 FROM roots WHERE id = ? AND kind = 'movie') AS 'exists!: bool'", + "describe": { + "columns": [ + { + "name": "exists!: bool", + "ordinal": 0, + "type_info": "Integer" + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + false + ] + }, + "hash": "ee86c3a4d78ef8ec0ca71343c5f1cbe1aea538f04e31b74f1bca0e7bd00c7248" +} diff --git a/Cargo.lock b/Cargo.lock index 8904f19..5507102 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -30,11 +30,15 @@ dependencies = [ name = "arr-api" version = "0.1.0" dependencies = [ + "arr-db", "axum", "reqwest", "serde", "serde_json", + "sqlx", + "tempfile", "tokio", + "tracing", "utoipa", "utoipa-axum", "utoipa-scalar", @@ -54,9 +58,11 @@ name = "arr-daemon" version = "0.1.0" dependencies = [ "arr-api", + "arr-db", "axum", "reqwest", "serde", + "sqlx", "tempfile", "thiserror", "tokio", diff --git a/crates/arr-api/Cargo.toml b/crates/arr-api/Cargo.toml index eb05a31..da35873 100644 --- a/crates/arr-api/Cargo.toml +++ b/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] diff --git a/crates/arr-api/src/lib.rs b/crates/arr-api/src/lib.rs index bf1999d..f8dcbdf 100644 --- a/crates/arr-api/src/lib.rs +++ b/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 { - 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] diff --git a/crates/arr-api/src/movies.rs b/crates/arr-api/src/movies.rs new file mode 100644 index 0000000..1d33ff7 --- /dev/null +++ b/crates/arr-api/src/movies.rs @@ -0,0 +1,484 @@ +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, + pub original_language: Option, + 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, +} + +#[derive(Debug, Deserialize, ToSchema)] +pub struct CreateMovie { + pub tmdb_id: i64, + pub title: String, + pub year: Option, + pub original_language: Option, + 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, + pub year: Option>, + pub original_language: Option>, + pub root_id: Option, + pub wanted: Option, + pub blocked: Option, + pub overrides: Option, +} + +#[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, + pub publish_date: Option, + pub download_url: String, + pub parsed: serde_json::Value, + pub score: Option, + pub verdict: Option, + pub rejected_rule: Option, +} + +#[derive(Debug, Clone, Serialize, ToSchema)] +pub struct AttentionQueues { + pub no_pt_source: Vec, + pub needs_decision: Vec, +} + +#[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 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 { + 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 = 503, body = ErrorBody)) +)] +pub async fn list(State(state): State) -> Result>, 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)) +)] +pub async fn create( + State(state): State, + Json(input): Json, +) -> Result<(StatusCode, Json), 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)) +)] +pub async fn get( + State(state): State, + Path(id): Path, +) -> Result, 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 = 422, body = ErrorBody)) +)] +pub async fn update( + State(state): State, + Path(id): Path, + Json(input): Json, +) -> Result, 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)) +)] +pub async fn delete( + State(state): State, + Path(id): Path, +) -> Result { + 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)) +)] +pub async fn search( + State(state): State, + Path(id): Path, +) -> Result<(StatusCode, Json), ApiError> { + load_movie(&state, id).await?; + state.send_movie_command(MovieCommand::Search { movie_id: id }); + 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)) +)] +pub async fn releases( + State(state): State, + Path(id): Path, +) -> Result>, 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)) +)] +pub async fn grab( + State(state): State, + Path((movie_id, release_id)): Path<(i64, i64)>, +) -> Result<(StatusCode, Json), 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, + }); + Ok((StatusCode::ACCEPTED, Json(Accepted { accepted: true }))) +} + +#[utoipa::path( + get, path = "/api/queues/attention", tag = "movies", + responses((status = 200, body = AttentionQueues), (status = 503, body = ErrorBody)) +)] +pub async fn attention(State(state): State) -> Result, 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.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 = 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 mut commands = state.subscribe_movie_commands(); + + let releases: Vec = + 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!( + commands.recv().await.expect("command"), + MovieCommand::Grab { + movie_id, + release_id + } + ); + } +} diff --git a/crates/arr-api/src/state.rs b/crates/arr-api/src/state.rs index ec53a87..e577fd5 100644 --- a/crates/arr-api/src/state.rs +++ b/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::broadcast; + /// 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, + database: Option, + movie_commands: broadcast::Sender, + pending_movie_commands: Arc>>, +} + +/// 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,38 @@ impl AppState { /// If the TLS backend cannot be initialised. pub fn new(upstreams: Upstreams) -> Result { let http = reqwest::Client::builder().timeout(PROBE_TIMEOUT).build()?; + let (movie_commands, pending_movie_commands) = broadcast::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 + } + + /// Subscribe to manual search and grab requests. + #[must_use] + pub fn subscribe_movie_commands(&self) -> broadcast::Receiver { + self.movie_commands.subscribe() + } + + /// Wait for the next manual movie action in the daemon's reconcile loop. + /// + /// # Errors + /// + /// If this receiver lagged behind the bounded channel. + pub async fn next_movie_command(&self) -> Result { + self.pending_movie_commands.lock().await.recv().await + } + pub(crate) fn http(&self) -> &reqwest::Client { &self.http } @@ -86,4 +125,12 @@ 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) { + let _ = self.movie_commands.send(command); + } } diff --git a/crates/arr-daemon/Cargo.toml b/crates/arr-daemon/Cargo.toml index 8ecd7b1..83cd807 100644 --- a/crates/arr-daemon/Cargo.toml +++ b/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 } diff --git a/crates/arr-daemon/src/main.rs b/crates/arr-daemon/src/main.rs index 4774b65..93f42d6 100644 --- a/crates/arr-daemon/src/main.rs +++ b/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()); diff --git a/crates/arr-db/migrations/0003_movie_releases.sql b/crates/arr-db/migrations/0003_movie_releases.sql new file mode 100644 index 0000000..c717dad --- /dev/null +++ b/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); -- 2.47.3 From 4357fcf893237991bf93abb6286207eb5fe13c43 Mon Sep 17 00:00:00 2001 From: Miguel Palhas Date: Sat, 22 Aug 2026 20:28:55 +0100 Subject: [PATCH 2/4] fix(api): document and test error paths --- crates/arr-api/src/movies.rs | 131 ++++++++++++++++++++++++++++++++--- 1 file changed, 122 insertions(+), 9 deletions(-) diff --git a/crates/arr-api/src/movies.rs b/crates/arr-api/src/movies.rs index 1d33ff7..370b62e 100644 --- a/crates/arr-api/src/movies.rs +++ b/crates/arr-api/src/movies.rs @@ -165,7 +165,11 @@ async fn load_movie(state: &AppState, id: i64) -> Result { #[utoipa::path( get, path = "/api/movies", tag = "movies", - responses((status = 200, body = [Movie]), (status = 503, body = ErrorBody)) + responses( + (status = 200, body = [Movie]), + (status = 500, body = ErrorBody), + (status = 503, body = ErrorBody) + ) )] pub async fn list(State(state): State) -> Result>, 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"#) @@ -176,7 +180,13 @@ pub async fn list(State(state): State) -> Result>, Api #[utoipa::path( post, path = "/api/movies", tag = "movies", request_body = CreateMovie, - responses((status = 201, body = Movie), (status = 409, body = ErrorBody), (status = 422, body = ErrorBody)) + 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, @@ -214,7 +224,12 @@ pub async fn create( #[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)) + responses( + (status = 200, body = Movie), + (status = 404, body = ErrorBody), + (status = 500, body = ErrorBody), + (status = 503, body = ErrorBody) + ) )] pub async fn get( State(state): State, @@ -226,7 +241,14 @@ pub async fn get( #[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 = 422, body = ErrorBody)) + 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, @@ -268,7 +290,13 @@ pub async fn update( #[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)) + 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, @@ -286,7 +314,12 @@ pub async fn delete( #[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)) + responses( + (status = 202, body = Accepted), + (status = 404, body = ErrorBody), + (status = 500, body = ErrorBody), + (status = 503, body = ErrorBody) + ) )] pub async fn search( State(state): State, @@ -300,7 +333,12 @@ pub async fn search( #[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)) + responses( + (status = 200, body = [Release]), + (status = 404, body = ErrorBody), + (status = 500, body = ErrorBody), + (status = 503, body = ErrorBody) + ) )] pub async fn releases( State(state): State, @@ -316,7 +354,12 @@ pub async fn 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)) + responses( + (status = 202, body = Accepted), + (status = 404, body = ErrorBody), + (status = 500, body = ErrorBody), + (status = 503, body = ErrorBody) + ) )] pub async fn grab( State(state): State, @@ -337,7 +380,11 @@ pub async fn grab( #[utoipa::path( get, path = "/api/queues/attention", tag = "movies", - responses((status = 200, body = AttentionQueues), (status = 503, body = ErrorBody)) + responses( + (status = 200, body = AttentionQueues), + (status = 500, body = ErrorBody), + (status = 503, body = ErrorBody) + ) )] pub async fn attention(State(state): State) -> Result, 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.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"#) @@ -481,4 +528,70 @@ mod tests { } ); } + + #[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 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); + } } -- 2.47.3 From f40b7588d59a7fd29a242423901448c2bcdebba5 Mon Sep 17 00:00:00 2001 From: Miguel Palhas Date: Sat, 22 Aug 2026 20:31:42 +0100 Subject: [PATCH 3/4] fix(api): preserve movie action requests --- ...2c6548e7d9a16d992bfb2a53140445aaf4ce.json} | 4 +- crates/arr-api/src/movies.rs | 53 +++++++++++++++---- crates/arr-api/src/state.rs | 25 ++++----- 3 files changed, 56 insertions(+), 26 deletions(-) rename .sqlx/{query-1f82542cde76554edd497ce056c80154a43f96a79844c6c3a399463ad557e0eb.json => query-511f13a7ae12a66289d6c7d099cf2c6548e7d9a16d992bfb2a53140445aaf4ce.json} (84%) diff --git a/.sqlx/query-1f82542cde76554edd497ce056c80154a43f96a79844c6c3a399463ad557e0eb.json b/.sqlx/query-511f13a7ae12a66289d6c7d099cf2c6548e7d9a16d992bfb2a53140445aaf4ce.json similarity index 84% rename from .sqlx/query-1f82542cde76554edd497ce056c80154a43f96a79844c6c3a399463ad557e0eb.json rename to .sqlx/query-511f13a7ae12a66289d6c7d099cf2c6548e7d9a16d992bfb2a53140445aaf4ce.json index 07c918f..df6a3be 100644 --- a/.sqlx/query-1f82542cde76554edd497ce056c80154a43f96a79844c6c3a399463ad557e0eb.json +++ b/.sqlx/query-511f13a7ae12a66289d6c7d099cf2c6548e7d9a16d992bfb2a53140445aaf4ce.json @@ -1,6 +1,6 @@ { "db_name": "SQLite", - "query": "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.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", + "query": "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", "describe": { "columns": [ { @@ -82,5 +82,5 @@ true ] }, - "hash": "1f82542cde76554edd497ce056c80154a43f96a79844c6c3a399463ad557e0eb" + "hash": "511f13a7ae12a66289d6c7d099cf2c6548e7d9a16d992bfb2a53140445aaf4ce" } diff --git a/crates/arr-api/src/movies.rs b/crates/arr-api/src/movies.rs index 370b62e..f89a585 100644 --- a/crates/arr-api/src/movies.rs +++ b/crates/arr-api/src/movies.rs @@ -317,6 +317,7 @@ pub async fn delete( responses( (status = 202, body = Accepted), (status = 404, body = ErrorBody), + (status = 409, body = ErrorBody), (status = 500, body = ErrorBody), (status = 503, body = ErrorBody) ) @@ -325,8 +326,14 @@ pub async fn search( State(state): State, Path(id): Path, ) -> Result<(StatusCode, Json), ApiError> { - load_movie(&state, id).await?; - state.send_movie_command(MovieCommand::Search { movie_id: id }); + 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 }) + .await + .map_err(|_| ApiError::Unavailable)?; Ok((StatusCode::ACCEPTED, Json(Accepted { accepted: true }))) } @@ -371,10 +378,13 @@ pub async fn grab( if !exists { return Err(ApiError::NotFound); } - state.send_movie_command(MovieCommand::Grab { - movie_id, - release_id, - }); + state + .send_movie_command(MovieCommand::Grab { + movie_id, + release_id, + }) + .await + .map_err(|_| ApiError::Unavailable)?; Ok((StatusCode::ACCEPTED, Json(Accepted { accepted: true }))) } @@ -387,7 +397,7 @@ pub async fn grab( ) )] pub async fn attention(State(state): State) -> Result, 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.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"#) + 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"#) @@ -501,8 +511,6 @@ mod tests { .execute(pool) .await .expect("association"); - let mut commands = state.subscribe_movie_commands(); - let releases: Vec = reqwest::get(format!("{base}/api/movies/{movie_id}/releases")) .await @@ -521,7 +529,7 @@ mod tests { .expect("grab"); assert_eq!(response.status(), StatusCode::ACCEPTED); assert_eq!( - commands.recv().await.expect("command"), + state.next_movie_command().await.expect("command"), MovieCommand::Grab { movie_id, release_id @@ -573,6 +581,31 @@ mod tests { 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) diff --git a/crates/arr-api/src/state.rs b/crates/arr-api/src/state.rs index e577fd5..0567a16 100644 --- a/crates/arr-api/src/state.rs +++ b/crates/arr-api/src/state.rs @@ -5,7 +5,7 @@ use std::sync::Arc; use std::time::Duration; use arr_db::Db; -use tokio::sync::broadcast; +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. @@ -67,8 +67,8 @@ pub struct AppState { http: reqwest::Client, upstreams: Arc, database: Option, - movie_commands: broadcast::Sender, - pending_movie_commands: Arc>>, + movie_commands: mpsc::Sender, + pending_movie_commands: Arc>>, } /// Work explicitly requested through the movie API. @@ -86,7 +86,7 @@ impl AppState { /// If the TLS backend cannot be initialised. pub fn new(upstreams: Upstreams) -> Result { let http = reqwest::Client::builder().timeout(PROBE_TIMEOUT).build()?; - let (movie_commands, pending_movie_commands) = broadcast::channel(64); + let (movie_commands, pending_movie_commands) = mpsc::channel(64); Ok(Self { http, upstreams: Arc::new(upstreams), @@ -103,18 +103,12 @@ impl AppState { self } - /// Subscribe to manual search and grab requests. - #[must_use] - pub fn subscribe_movie_commands(&self) -> broadcast::Receiver { - self.movie_commands.subscribe() - } - /// Wait for the next manual movie action in the daemon's reconcile loop. /// /// # Errors /// - /// If this receiver lagged behind the bounded channel. - pub async fn next_movie_command(&self) -> Result { + /// If every sender has been dropped. + pub async fn next_movie_command(&self) -> Option { self.pending_movie_commands.lock().await.recv().await } @@ -130,7 +124,10 @@ impl AppState { self.database.as_ref() } - pub(crate) fn send_movie_command(&self, command: MovieCommand) { - let _ = self.movie_commands.send(command); + pub(crate) async fn send_movie_command( + &self, + command: MovieCommand, + ) -> Result<(), mpsc::error::SendError> { + self.movie_commands.send(command).await } } -- 2.47.3 From b2e4e157c3021e88e621573bafdeb223de26af82 Mon Sep 17 00:00:00 2001 From: Miguel Palhas Date: Sat, 22 Aug 2026 20:33:39 +0100 Subject: [PATCH 4/4] fix(api): reject full action queues --- crates/arr-api/src/movies.rs | 21 +++++++++++++++++++-- crates/arr-api/src/state.rs | 6 +++--- 2 files changed, 22 insertions(+), 5 deletions(-) diff --git a/crates/arr-api/src/movies.rs b/crates/arr-api/src/movies.rs index f89a585..2c352dc 100644 --- a/crates/arr-api/src/movies.rs +++ b/crates/arr-api/src/movies.rs @@ -332,7 +332,6 @@ pub async fn search( } state .send_movie_command(MovieCommand::Search { movie_id: id }) - .await .map_err(|_| ApiError::Unavailable)?; Ok((StatusCode::ACCEPTED, Json(Accepted { accepted: true }))) } @@ -383,7 +382,6 @@ pub async fn grab( movie_id, release_id, }) - .await .map_err(|_| ApiError::Unavailable)?; Ok((StatusCode::ACCEPTED, Json(Accepted { accepted: true }))) } @@ -537,6 +535,25 @@ mod tests { ); } + #[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; diff --git a/crates/arr-api/src/state.rs b/crates/arr-api/src/state.rs index 0567a16..0030cd5 100644 --- a/crates/arr-api/src/state.rs +++ b/crates/arr-api/src/state.rs @@ -124,10 +124,10 @@ impl AppState { self.database.as_ref() } - pub(crate) async fn send_movie_command( + pub(crate) fn send_movie_command( &self, command: MovieCommand, - ) -> Result<(), mpsc::error::SendError> { - self.movie_commands.send(command).await + ) -> Result<(), mpsc::error::TrySendError> { + self.movie_commands.try_send(command) } } -- 2.47.3