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 } }