refactor(api): extract tv attention lanes

crates/arr-api/src/movies.rs

@@ -646,7 +646,25 @@ pub async fn attention(State(state): State<AppState>) -> Result<Json<AttentionQu
     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, (SELECT f.waiver FROM media_files f WHERE f.owner_kind = 'movie' AND f.owner_id = movies.id AND f.waiver IS NOT NULL ORDER BY f.id LIMIT 1) AS "waiver?: serde_json::Value" 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?;
-    let tv_no_pt_source = sqlx::query!(
+    let (tv_no_pt_source, tv_needs_decision) = tv_attention(&state).await?;
+
+    Ok(Json(AttentionQueues {
+        no_pt_source,
+        needs_decision,
+        tv_no_pt_source,
+        tv_needs_decision,
+    }))
+}
+
+/// The TV lanes of the attention queues (§9.5): one entry per series with
+/// the episodes and seasons that put it there. The two hard-fail conditions
+/// share a lane; a series arriving through both is merged into one entry.
+async fn tv_attention(
+    state: &AppState,
+) -> Result<(Vec<SeriesAttention>, Vec<SeriesAttention>), ApiError> {
+    let database = pool(state)?;
+
+    let no_pt_rows = sqlx::query!(
         r#"
             SELECT s.id AS "series_id!: i64", s.tmdb_id AS "tmdb_id!: i64",
                    s.title AS "title!: String", s.year, e.id AS "episode_id!: i64"
@@ -665,7 +683,7 @@ pub async fn attention(State(state): State<AppState>) -> Result<Json<AttentionQu
             ORDER BY se.number, e.number
             "#
     )
-    .fetch_all(pool(&state)?)
+    .fetch_all(database)
     .await?;
     let episode_hard_fails = sqlx::query!(
         r#"
@@ -680,7 +698,7 @@ pub async fn attention(State(state): State<AppState>) -> Result<Json<AttentionQu
         HAVING count(DISTINCT g.release_id) >= 2
         "#
     )
-    .fetch_all(pool(&state)?)
+    .fetch_all(database)
     .await?;
     let season_pack_fails = sqlx::query!(
         r#"
@@ -693,53 +711,39 @@ pub async fn attention(State(state): State<AppState>) -> Result<Json<AttentionQu
         GROUP BY s.id, s.tmdb_id, s.title, s.year, se.id
         "#
     )
-    .fetch_all(pool(&state)?)
+    .fetch_all(database)
     .await?;
 
-    let mut tv_no_pt: Vec<SeriesAttention> = Vec::new();
-    for row in tv_no_pt_source {
-        match tv_no_pt
-            .iter_mut()
-            .find(|entry| entry.series_id == row.series_id)
-        {
-            Some(entry) => entry.episodes.push(row.episode_id),
-            None => tv_no_pt.push(SeriesAttention {
-                series_id: row.series_id,
-                tmdb_id: row.tmdb_id,
-                title: row.title,
-                year: row.year,
-                episodes: vec![row.episode_id],
-                seasons: Vec::new(),
-            }),
-        }
+    let mut tv_no_pt_source: Vec<SeriesAttention> = Vec::new();
+    for row in no_pt_rows {
+        merge_episode(
+            &mut tv_no_pt_source,
+            row.series_id,
+            row.tmdb_id,
+            row.title,
+            row.year,
+            row.episode_id,
+        );
     }
 
-    // The two hard-fail lanes share one queue; a series can arrive through
-    // both and is merged into one entry.
-    let mut tv_needs: Vec<SeriesAttention> = Vec::new();
+    let mut tv_needs_decision: Vec<SeriesAttention> = Vec::new();
     for row in episode_hard_fails {
-        match tv_needs
-            .iter_mut()
-            .find(|entry| entry.series_id == row.series_id)
-        {
-            Some(entry) => entry.episodes.push(row.episode_id),
-            None => tv_needs.push(SeriesAttention {
-                series_id: row.series_id,
-                tmdb_id: row.tmdb_id,
-                title: row.title,
-                year: row.year,
-                episodes: vec![row.episode_id],
-                seasons: Vec::new(),
-            }),
-        }
+        merge_episode(
+            &mut tv_needs_decision,
+            row.series_id,
+            row.tmdb_id,
+            row.title,
+            row.year,
+            row.episode_id,
+        );
     }
     for row in season_pack_fails {
-        match tv_needs
+        match tv_needs_decision
             .iter_mut()
             .find(|entry| entry.series_id == row.series_id)
         {
             Some(entry) => entry.seasons.push(row.season_id),
-            None => tv_needs.push(SeriesAttention {
+            None => tv_needs_decision.push(SeriesAttention {
                 series_id: row.series_id,
                 tmdb_id: row.tmdb_id,
                 title: row.title,
@@ -750,12 +754,34 @@ pub async fn attention(State(state): State<AppState>) -> Result<Json<AttentionQu
         }
     }
 
-    Ok(Json(AttentionQueues {
-        no_pt_source,
-        needs_decision,
-        tv_no_pt_source: tv_no_pt,
-        tv_needs_decision: tv_needs,
-    }))
+    Ok((tv_no_pt_source, tv_needs_decision))
+}
+
+/// One more qualifying episode for its series, creating the series' entry on
+/// first sight.
+fn merge_episode(
+    entries: &mut Vec<SeriesAttention>,
+    series_id: i64,
+    tmdb_id: i64,
+    title: String,
+    year: Option<i64>,
+    episode_id: i64,
+) {
+    if let Some(entry) = entries
+        .iter_mut()
+        .find(|entry| entry.series_id == series_id)
+    {
+        entry.episodes.push(episode_id);
+        return;
+    }
+    entries.push(SeriesAttention {
+        series_id,
+        tmdb_id,
+        title,
+        year,
+        episodes: vec![episode_id],
+        seasons: Vec::new(),
+    });
 }
 
 #[utoipa::path(