fix(api): preserve movie action requests

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

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<AppState>,
     Path(id): Path<i64>,
 ) -> Result<(StatusCode, Json<Accepted>), 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<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.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<serde_json::Value> =
             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)

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<Upstreams>,
     database: Option<Db>,
-    movie_commands: broadcast::Sender<MovieCommand>,
-    pending_movie_commands: Arc<tokio::sync::Mutex<broadcast::Receiver<MovieCommand>>>,
+    movie_commands: mpsc::Sender<MovieCommand>,
+    pending_movie_commands: Arc<tokio::sync::Mutex<mpsc::Receiver<MovieCommand>>>,
 }
 
 /// 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<Self, reqwest::Error> {
         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<MovieCommand> {
-        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<MovieCommand, broadcast::error::RecvError> {
+    /// If every sender has been dropped.
+    pub async fn next_movie_command(&self) -> Option<MovieCommand> {
         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<MovieCommand>> {
+        self.movie_commands.send(command).await
     }
 }