Miguel Palhas
+38
@@ -0,0 +1,38 @@
|
||||
{
|
||||
"db_name": "SQLite",
|
||||
"query": "\n SELECT s.id AS \"series_id!: i64\", s.title AS \"title!: String\", s.year,\n g.target_id AS \"season_id!: i64\"\n FROM grabs g\n JOIN seasons se ON g.target_kind = 'season' AND se.id = g.target_id\n JOIN series s ON s.id = se.series_id\n WHERE g.state = 'failed'\n GROUP BY s.id, s.title, s.year, se.id\n ",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"name": "series_id!: i64",
|
||||
"ordinal": 0,
|
||||
"type_info": "Integer"
|
||||
},
|
||||
{
|
||||
"name": "title!: String",
|
||||
"ordinal": 1,
|
||||
"type_info": "Text"
|
||||
},
|
||||
{
|
||||
"name": "year",
|
||||
"ordinal": 2,
|
||||
"type_info": "Integer"
|
||||
},
|
||||
{
|
||||
"name": "season_id!: i64",
|
||||
"ordinal": 3,
|
||||
"type_info": "Integer"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Right": 0
|
||||
},
|
||||
"nullable": [
|
||||
false,
|
||||
false,
|
||||
true,
|
||||
false
|
||||
]
|
||||
},
|
||||
"hash": "4ddb143ab51ca61ac782f22cff84f01f7d58d8a424310ae0743fb0c91577665e"
|
||||
}
|
||||
+38
@@ -0,0 +1,38 @@
|
||||
{
|
||||
"db_name": "SQLite",
|
||||
"query": "\n SELECT s.id AS \"series_id!: i64\", s.title AS \"title!: String\", s.year,\n e.id AS \"episode_id!: i64\"\n FROM episodes e\n JOIN seasons se ON se.id = e.season_id\n JOIN series s ON s.id = se.series_id\n JOIN roots root ON root.id = s.root_id\n WHERE root.audience = 'kids'\n AND s.blocked = 0\n AND e.wanted = 1 AND e.state = 'missing' AND e.search_attempts > 0\n AND NOT EXISTS (\n SELECT 1 FROM episode_releases er\n JOIN releases r ON r.id = er.release_id\n WHERE er.episode_id = e.id AND r.verdict IN ('eligible', 'waived')\n )\n ",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"name": "series_id!: i64",
|
||||
"ordinal": 0,
|
||||
"type_info": "Integer"
|
||||
},
|
||||
{
|
||||
"name": "title!: String",
|
||||
"ordinal": 1,
|
||||
"type_info": "Text"
|
||||
},
|
||||
{
|
||||
"name": "year",
|
||||
"ordinal": 2,
|
||||
"type_info": "Integer"
|
||||
},
|
||||
{
|
||||
"name": "episode_id!: i64",
|
||||
"ordinal": 3,
|
||||
"type_info": "Integer"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Right": 0
|
||||
},
|
||||
"nullable": [
|
||||
false,
|
||||
false,
|
||||
true,
|
||||
true
|
||||
]
|
||||
},
|
||||
"hash": "a071c1bedfdd1347713cac523cb1aa9522dc9a07fb5d681655622971398a0ff4"
|
||||
}
|
||||
+44
@@ -0,0 +1,44 @@
|
||||
{
|
||||
"db_name": "SQLite",
|
||||
"query": "\n SELECT s.id AS \"series_id!: i64\", s.tmdb_id AS \"tmdb_id!: i64\",\n s.title AS \"title!: String\", s.year, e.id AS \"episode_id!: i64\"\n FROM episodes e\n JOIN seasons se ON se.id = e.season_id\n JOIN series s ON s.id = se.series_id\n JOIN roots root ON root.id = s.root_id\n WHERE root.audience = 'kids'\n AND s.blocked = 0\n AND e.wanted = 1 AND e.state = 'missing' AND e.search_attempts > 0\n AND NOT EXISTS (\n SELECT 1 FROM episode_releases er\n JOIN releases r ON r.id = er.release_id\n WHERE er.episode_id = e.id AND r.verdict IN ('eligible', 'waived')\n )\n ORDER BY se.number, e.number\n ",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"name": "series_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": "episode_id!: i64",
|
||||
"ordinal": 4,
|
||||
"type_info": "Integer"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Right": 0
|
||||
},
|
||||
"nullable": [
|
||||
false,
|
||||
false,
|
||||
false,
|
||||
true,
|
||||
true
|
||||
]
|
||||
},
|
||||
"hash": "a36a611d7c587f1c1880113e334861867891d55dc8bb7fe79b1dc42e36054379"
|
||||
}
|
||||
+44
@@ -0,0 +1,44 @@
|
||||
{
|
||||
"db_name": "SQLite",
|
||||
"query": "\n SELECT s.id AS \"series_id!: i64\", s.tmdb_id AS \"tmdb_id!: i64\",\n s.title AS \"title!: String\", s.year, g.target_id AS \"episode_id!: i64\"\n FROM grabs g\n JOIN episodes e ON g.target_kind = 'episode' AND e.id = g.target_id\n JOIN seasons se ON se.id = e.season_id\n JOIN series s ON s.id = se.series_id\n WHERE g.state = 'failed'\n GROUP BY s.id, s.tmdb_id, s.title, s.year, e.id\n HAVING count(DISTINCT g.release_id) >= 2\n ",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"name": "series_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": "episode_id!: i64",
|
||||
"ordinal": 4,
|
||||
"type_info": "Integer"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Right": 0
|
||||
},
|
||||
"nullable": [
|
||||
false,
|
||||
false,
|
||||
false,
|
||||
true,
|
||||
false
|
||||
]
|
||||
},
|
||||
"hash": "b4b58a625ce7b57ba423d39296cb310c6b90c3feb283885b3573ee7666b1725b"
|
||||
}
|
||||
+44
@@ -0,0 +1,44 @@
|
||||
{
|
||||
"db_name": "SQLite",
|
||||
"query": "\n SELECT s.id AS \"series_id!: i64\", s.tmdb_id AS \"tmdb_id!: i64\",\n s.title AS \"title!: String\", s.year, g.target_id AS \"season_id!: i64\"\n FROM grabs g\n JOIN seasons se ON g.target_kind = 'season' AND se.id = g.target_id\n JOIN series s ON s.id = se.series_id\n WHERE g.state = 'failed'\n GROUP BY s.id, s.tmdb_id, s.title, s.year, se.id\n ",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"name": "series_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": "season_id!: i64",
|
||||
"ordinal": 4,
|
||||
"type_info": "Integer"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Right": 0
|
||||
},
|
||||
"nullable": [
|
||||
false,
|
||||
false,
|
||||
false,
|
||||
true,
|
||||
false
|
||||
]
|
||||
},
|
||||
"hash": "cd09d4c34f0596319fa2ade088ea9009fa24e453f669bcc2357fdcd999b06193"
|
||||
}
|
||||
+38
@@ -0,0 +1,38 @@
|
||||
{
|
||||
"db_name": "SQLite",
|
||||
"query": "\n SELECT s.id AS \"series_id!: i64\", s.title AS \"title!: String\", s.year,\n g.target_id AS \"episode_id!: i64\"\n FROM grabs g\n JOIN episodes e ON g.target_kind = 'episode' AND e.id = g.target_id\n JOIN seasons se ON se.id = e.season_id\n JOIN series s ON s.id = se.series_id\n WHERE g.state = 'failed'\n GROUP BY s.id, s.title, s.year, e.id\n HAVING count(DISTINCT g.release_id) >= 2\n ",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"name": "series_id!: i64",
|
||||
"ordinal": 0,
|
||||
"type_info": "Integer"
|
||||
},
|
||||
{
|
||||
"name": "title!: String",
|
||||
"ordinal": 1,
|
||||
"type_info": "Text"
|
||||
},
|
||||
{
|
||||
"name": "year",
|
||||
"ordinal": 2,
|
||||
"type_info": "Integer"
|
||||
},
|
||||
{
|
||||
"name": "episode_id!: i64",
|
||||
"ordinal": 3,
|
||||
"type_info": "Integer"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Right": 0
|
||||
},
|
||||
"nullable": [
|
||||
false,
|
||||
false,
|
||||
true,
|
||||
false
|
||||
]
|
||||
},
|
||||
"hash": "faa9e5cf0cf869161dd4878f6fb8707e378b7d9c6516d602afb49b1e30792207"
|
||||
}
|
||||
@@ -343,6 +343,7 @@ mod tests {
|
||||
"UpdateMovie",
|
||||
"Release",
|
||||
"AttentionQueues",
|
||||
"SeriesAttention",
|
||||
"Series",
|
||||
] {
|
||||
assert!(
|
||||
|
||||
@@ -93,6 +93,25 @@ pub struct MovieFile {
|
||||
pub struct AttentionQueues {
|
||||
pub no_pt_source: Vec<Movie>,
|
||||
pub needs_decision: Vec<Movie>,
|
||||
/// TV side (§9.5): series whose wanted episodes found no Portuguese-audio
|
||||
/// source.
|
||||
pub tv_no_pt_source: Vec<SeriesAttention>,
|
||||
/// TV side: series put here by hard failures on episodes or season packs.
|
||||
pub tv_needs_decision: Vec<SeriesAttention>,
|
||||
}
|
||||
|
||||
/// A series in a TV attention queue, with what put it there. One row per
|
||||
/// series — the notification rolls up the same way (§9.5).
|
||||
#[derive(Debug, Clone, Serialize, ToSchema)]
|
||||
pub struct SeriesAttention {
|
||||
pub series_id: i64,
|
||||
pub tmdb_id: i64,
|
||||
pub title: String,
|
||||
pub year: Option<i64>,
|
||||
/// Episodes that put the series in this queue.
|
||||
pub episodes: Vec<i64>,
|
||||
/// Seasons whose pack grab failed and fell back to per-episode grabbing.
|
||||
pub seasons: Vec<i64>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, ToSchema)]
|
||||
@@ -627,12 +646,144 @@ 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, 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"
|
||||
FROM episodes e
|
||||
JOIN seasons se ON se.id = e.season_id
|
||||
JOIN series s ON s.id = se.series_id
|
||||
JOIN roots root ON root.id = s.root_id
|
||||
WHERE root.audience = 'kids'
|
||||
AND s.blocked = 0
|
||||
AND e.wanted = 1 AND e.state = 'missing' AND e.search_attempts > 0
|
||||
AND NOT EXISTS (
|
||||
SELECT 1 FROM episode_releases er
|
||||
JOIN releases r ON r.id = er.release_id
|
||||
WHERE er.episode_id = e.id AND r.verdict IN ('eligible', 'waived')
|
||||
)
|
||||
ORDER BY se.number, e.number
|
||||
"#
|
||||
)
|
||||
.fetch_all(database)
|
||||
.await?;
|
||||
let episode_hard_fails = sqlx::query!(
|
||||
r#"
|
||||
SELECT s.id AS "series_id!: i64", s.tmdb_id AS "tmdb_id!: i64",
|
||||
s.title AS "title!: String", s.year, g.target_id AS "episode_id!: i64"
|
||||
FROM grabs g
|
||||
JOIN episodes e ON g.target_kind = 'episode' AND e.id = g.target_id
|
||||
JOIN seasons se ON se.id = e.season_id
|
||||
JOIN series s ON s.id = se.series_id
|
||||
WHERE g.state = 'failed'
|
||||
GROUP BY s.id, s.tmdb_id, s.title, s.year, e.id
|
||||
HAVING count(DISTINCT g.release_id) >= 2
|
||||
"#
|
||||
)
|
||||
.fetch_all(database)
|
||||
.await?;
|
||||
let season_pack_fails = sqlx::query!(
|
||||
r#"
|
||||
SELECT s.id AS "series_id!: i64", s.tmdb_id AS "tmdb_id!: i64",
|
||||
s.title AS "title!: String", s.year, g.target_id AS "season_id!: i64"
|
||||
FROM grabs g
|
||||
JOIN seasons se ON g.target_kind = 'season' AND se.id = g.target_id
|
||||
JOIN series s ON s.id = se.series_id
|
||||
WHERE g.state = 'failed'
|
||||
GROUP BY s.id, s.tmdb_id, s.title, s.year, se.id
|
||||
"#
|
||||
)
|
||||
.fetch_all(database)
|
||||
.await?;
|
||||
|
||||
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,
|
||||
);
|
||||
}
|
||||
|
||||
let mut tv_needs_decision: Vec<SeriesAttention> = Vec::new();
|
||||
for row in episode_hard_fails {
|
||||
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_decision
|
||||
.iter_mut()
|
||||
.find(|entry| entry.series_id == row.series_id)
|
||||
{
|
||||
Some(entry) => entry.seasons.push(row.season_id),
|
||||
None => tv_needs_decision.push(SeriesAttention {
|
||||
series_id: row.series_id,
|
||||
tmdb_id: row.tmdb_id,
|
||||
title: row.title,
|
||||
year: row.year,
|
||||
episodes: Vec::new(),
|
||||
seasons: vec![row.season_id],
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
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(
|
||||
get, path = "/api/movies/{movie_id}/owners", tag = "movies",
|
||||
params(("movie_id" = i64, Path, description = "Movie row id")),
|
||||
@@ -724,6 +875,7 @@ pub async fn untag_owner(
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::{router, Upstreams};
|
||||
use sqlx::Row;
|
||||
|
||||
async fn application() -> (tempfile::TempDir, AppState, String) {
|
||||
let dir = tempfile::tempdir().expect("tempdir");
|
||||
@@ -1183,4 +1335,142 @@ mod tests {
|
||||
assert_eq!(queues["no_pt_source"].as_array().map(Vec::len), Some(0));
|
||||
assert_eq!(queues["needs_decision"][0]["id"], movie_id);
|
||||
}
|
||||
|
||||
/// §9.5 TV: the attention queues report series, with the episodes and
|
||||
/// seasons that put each one there.
|
||||
#[tokio::test]
|
||||
async fn attention_queues_report_tv_series() {
|
||||
let (_dir, state, base) = application().await;
|
||||
let pool = state.database().expect("database").pool();
|
||||
let (series_id, season_id, episode_id) = seed_queued_series(pool).await;
|
||||
|
||||
let queues: serde_json::Value = reqwest::get(format!("{base}/api/queues/attention"))
|
||||
.await
|
||||
.expect("queues")
|
||||
.json()
|
||||
.await
|
||||
.expect("queues json");
|
||||
assert_eq!(queues["tv_no_pt_source"][0]["series_id"], series_id);
|
||||
assert_eq!(
|
||||
queues["tv_no_pt_source"][0]["episodes"]
|
||||
.as_array()
|
||||
.map(Vec::len),
|
||||
Some(2)
|
||||
);
|
||||
assert_eq!(queues["tv_needs_decision"][0]["series_id"], series_id);
|
||||
assert_eq!(
|
||||
queues["tv_needs_decision"][0]["episodes"],
|
||||
serde_json::json!([episode_id])
|
||||
);
|
||||
assert_eq!(
|
||||
queues["tv_needs_decision"][0]["seasons"],
|
||||
serde_json::json!([season_id])
|
||||
);
|
||||
|
||||
// Blocking the series empties its no-PT-source lane.
|
||||
sqlx::query("UPDATE series SET blocked = 1 WHERE id = ?")
|
||||
.bind(series_id)
|
||||
.execute(pool)
|
||||
.await
|
||||
.expect("block series");
|
||||
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["tv_no_pt_source"].as_array().map(Vec::len), Some(0));
|
||||
assert_eq!(
|
||||
queues["tv_needs_decision"].as_array().map(Vec::len),
|
||||
Some(1),
|
||||
"hard failures queue regardless of blocking"
|
||||
);
|
||||
}
|
||||
|
||||
/// One series hitting all three §9.5 TV entry conditions: two wanted,
|
||||
/// searched episodes whose every candidate was rejected for language; a
|
||||
/// season pack that hard-failed; and an episode two different releases
|
||||
/// hard-failed on.
|
||||
async fn seed_queued_series(pool: &sqlx::SqlitePool) -> (i64, i64, i64) {
|
||||
let root_id: i64 =
|
||||
sqlx::query_scalar("SELECT id FROM roots WHERE kind = 'tv' AND audience = 'kids'")
|
||||
.fetch_one(pool)
|
||||
.await
|
||||
.expect("kids tv root");
|
||||
let series_id: i64 = sqlx::query(
|
||||
"INSERT INTO series (tmdb_id, title, year, root_id) VALUES (1, 'Bluey', 2018, ?)
|
||||
RETURNING id",
|
||||
)
|
||||
.bind(root_id)
|
||||
.fetch_one(pool)
|
||||
.await
|
||||
.expect("series")
|
||||
.get(0);
|
||||
let season_id: i64 = sqlx::query_scalar(
|
||||
"INSERT INTO seasons (series_id, number) VALUES (?, 1) RETURNING id",
|
||||
)
|
||||
.bind(series_id)
|
||||
.fetch_one(pool)
|
||||
.await
|
||||
.expect("season");
|
||||
|
||||
for number in [1, 2] {
|
||||
sqlx::query(
|
||||
"INSERT INTO episodes (season_id, number, title, wanted, state, search_attempts)
|
||||
VALUES (?, ?, ?, 1, 'missing', 1)",
|
||||
)
|
||||
.bind(season_id)
|
||||
.bind(number)
|
||||
.bind(format!("Episode {number}"))
|
||||
.execute(pool)
|
||||
.await
|
||||
.expect("episode");
|
||||
}
|
||||
sqlx::query(
|
||||
"INSERT INTO episodes (season_id, number, title, wanted, state)
|
||||
VALUES (?, 3, 'Episode 3', 1, 'missing')",
|
||||
)
|
||||
.bind(season_id)
|
||||
.execute(pool)
|
||||
.await
|
||||
.expect("episode");
|
||||
let episode_id: i64 =
|
||||
sqlx::query_scalar("SELECT id FROM episodes WHERE season_id = ? AND number = 3")
|
||||
.bind(season_id)
|
||||
.fetch_one(pool)
|
||||
.await
|
||||
.expect("episode id");
|
||||
for (kind, guid, suffix) in [
|
||||
("season", "pack", "pack"),
|
||||
("episode", "first", "a"),
|
||||
("episode", "second", "b"),
|
||||
] {
|
||||
let release_id: i64 = sqlx::query(
|
||||
"INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed, verdict)
|
||||
VALUES (1, ?, 'release', 1, 'url', '{}', 'eligible') RETURNING id",
|
||||
)
|
||||
.bind(guid)
|
||||
.fetch_one(pool)
|
||||
.await
|
||||
.expect("release")
|
||||
.get(0);
|
||||
sqlx::query(
|
||||
"INSERT INTO grabs (release_id, target_kind, target_id, infohash, state)
|
||||
VALUES (?, ?, ?, ?, 'failed')",
|
||||
)
|
||||
.bind(release_id)
|
||||
.bind(kind)
|
||||
.bind(if kind == "season" {
|
||||
season_id
|
||||
} else {
|
||||
episode_id
|
||||
})
|
||||
.bind(format!("hash-{suffix}"))
|
||||
.execute(pool)
|
||||
.await
|
||||
.expect("failed grab");
|
||||
}
|
||||
|
||||
(series_id, season_id, episode_id)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,11 +1,14 @@
|
||||
//! §9.5 *needs a decision* → the operator alone: a movie entered the
|
||||
//! no-PT-source queue, or hard-failed twice on different releases (the same
|
||||
//! two queues `GET /api/queues/attention` reports, §9.3).
|
||||
//! §9.5 *needs a decision* → the operator alone: a movie or series entered
|
||||
//! the no-PT-source queue, or hard-failed twice on different releases (the
|
||||
//! same queues `GET /api/queues/attention` reports, §9.3).
|
||||
//!
|
||||
//! Edge-triggered: a movie notifies once when it enters either queue, and is
|
||||
//! forgotten once it leaves both, so a future re-entry notifies again. The
|
||||
//! notified set is in-memory and transient — a restart re-notifies whatever
|
||||
//! is currently queued, same tradeoff `ImportAction`'s probe cache makes.
|
||||
//! Edge-triggered per title: it notifies once when the title enters either
|
||||
//! queue, and is forgotten once it leaves both, so a future re-entry notifies
|
||||
//! again. A series notifies as its series, never per episode — a broken
|
||||
//! twenty-episode season is one message, which is the restraint §9.5 exists
|
||||
//! for. The notified set is in-memory and transient — a restart re-notifies
|
||||
//! whatever is currently queued, same tradeoff `ImportAction`'s probe cache
|
||||
//! makes.
|
||||
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use std::sync::Arc;
|
||||
@@ -16,11 +19,67 @@ use tokio::sync::Mutex;
|
||||
use crate::notify::Notifier;
|
||||
use crate::reconcile::{Action, ActionFuture, Outcome};
|
||||
|
||||
/// What is queued. Movies and series have independent id sequences, so the
|
||||
/// kind travels with the id.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
|
||||
enum Subject {
|
||||
Movie(i64),
|
||||
Series(i64),
|
||||
}
|
||||
|
||||
/// One series' standing in the TV queues, rolled up from its episodes.
|
||||
#[derive(Debug, Default)]
|
||||
struct TvEntry {
|
||||
/// Wanted episodes whose every candidate was rejected for language.
|
||||
no_pt_source: Vec<i64>,
|
||||
/// Episodes two different releases hard-failed post-probe (§5.7).
|
||||
hard_failed_episodes: Vec<i64>,
|
||||
/// Seasons whose pack grab hard-failed, sending the season back to
|
||||
/// per-episode grabbing.
|
||||
failed_season_packs: Vec<i64>,
|
||||
}
|
||||
|
||||
impl TvEntry {
|
||||
fn reason(&self) -> String {
|
||||
let mut parts = Vec::new();
|
||||
if !self.no_pt_source.is_empty() {
|
||||
parts.push(plural(
|
||||
self.no_pt_source.len(),
|
||||
"episode found no Portuguese-audio source",
|
||||
"episodes found no Portuguese-audio source",
|
||||
));
|
||||
}
|
||||
if !self.hard_failed_episodes.is_empty() {
|
||||
parts.push(plural(
|
||||
self.hard_failed_episodes.len(),
|
||||
"episode hard-failed twice on different releases",
|
||||
"episodes hard-failed twice on different releases",
|
||||
));
|
||||
}
|
||||
if !self.failed_season_packs.is_empty() {
|
||||
parts.push(plural(
|
||||
self.failed_season_packs.len(),
|
||||
"season pack hard-failed",
|
||||
"season packs hard-failed",
|
||||
));
|
||||
}
|
||||
parts.join("; ")
|
||||
}
|
||||
}
|
||||
|
||||
fn plural(count: usize, one: &str, many: &str) -> String {
|
||||
if count == 1 {
|
||||
format!("1 {one}")
|
||||
} else {
|
||||
format!("{count} {many}")
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub struct AttentionAction {
|
||||
notifier: Notifier,
|
||||
operator_topic: String,
|
||||
notified: Arc<Mutex<HashSet<i64>>>,
|
||||
notified: Arc<Mutex<HashSet<Subject>>>,
|
||||
}
|
||||
|
||||
impl AttentionAction {
|
||||
@@ -34,7 +93,7 @@ impl AttentionAction {
|
||||
}
|
||||
|
||||
async fn tick(&self, database: &Db) -> Result<Vec<Outcome>, sqlx::Error> {
|
||||
let mut queued: HashMap<i64, (String, &'static str)> = HashMap::new();
|
||||
let mut queued: HashMap<Subject, (String, String)> = HashMap::new();
|
||||
|
||||
let no_pt_source = sqlx::query!(
|
||||
r#"
|
||||
@@ -59,10 +118,10 @@ impl AttentionAction {
|
||||
.await?;
|
||||
for row in no_pt_source {
|
||||
queued.insert(
|
||||
row.id,
|
||||
Subject::Movie(row.id),
|
||||
(
|
||||
title_with_year(&row.title, row.year),
|
||||
"no Portuguese-audio source found",
|
||||
"no Portuguese-audio source found".to_string(),
|
||||
),
|
||||
);
|
||||
}
|
||||
@@ -80,18 +139,25 @@ impl AttentionAction {
|
||||
.fetch_all(database.pool())
|
||||
.await?;
|
||||
for row in needs_decision {
|
||||
queued.entry(row.id).or_insert_with(|| {
|
||||
queued.entry(Subject::Movie(row.id)).or_insert_with(|| {
|
||||
(
|
||||
title_with_year(&row.title, row.year),
|
||||
"hard-failed twice on different releases",
|
||||
"hard-failed twice on different releases".to_string(),
|
||||
)
|
||||
});
|
||||
}
|
||||
|
||||
for (series_id, title, year, entry) in queue_tv(database).await? {
|
||||
queued.insert(
|
||||
Subject::Series(series_id),
|
||||
(title_with_year(&title, year), entry.reason()),
|
||||
);
|
||||
}
|
||||
|
||||
let mut notified = self.notified.lock().await;
|
||||
let mut outcomes = Vec::new();
|
||||
for (&movie_id, (title, reason)) in &queued {
|
||||
if !notified.insert(movie_id) {
|
||||
for (subject, (title, reason)) in &queued {
|
||||
if !notified.insert(subject.clone()) {
|
||||
continue;
|
||||
}
|
||||
match self
|
||||
@@ -99,16 +165,22 @@ impl AttentionAction {
|
||||
.send(&self.operator_topic, title, reason)
|
||||
.await
|
||||
{
|
||||
Ok(()) => outcomes.push(Outcome::new(
|
||||
format!("movie {movie_id} needs a decision"),
|
||||
format!("notified operator: {reason}"),
|
||||
)),
|
||||
Ok(()) => {
|
||||
let (kind, id) = match subject {
|
||||
Subject::Movie(id) => ("movie", *id),
|
||||
Subject::Series(id) => ("series", *id),
|
||||
};
|
||||
outcomes.push(Outcome::new(
|
||||
format!("{kind} {id} needs a decision"),
|
||||
format!("notified operator: {reason}"),
|
||||
));
|
||||
}
|
||||
Err(error) => {
|
||||
tracing::warn!(%error, movie_id, "needs-decision notification failed");
|
||||
tracing::warn!(%error, ?subject, "needs-decision notification failed");
|
||||
}
|
||||
}
|
||||
}
|
||||
notified.retain(|movie_id| queued.contains_key(movie_id));
|
||||
notified.retain(|subject| queued.contains_key(subject));
|
||||
|
||||
Ok(outcomes)
|
||||
}
|
||||
@@ -121,6 +193,99 @@ fn title_with_year(title: &str, year: Option<i64>) -> String {
|
||||
}
|
||||
}
|
||||
|
||||
/// The series' roll-up entry, created empty on first sight.
|
||||
fn tv_entry(
|
||||
tv: &mut HashMap<i64, (String, Option<i64>, TvEntry)>,
|
||||
series_id: i64,
|
||||
title: String,
|
||||
year: Option<i64>,
|
||||
) -> &mut TvEntry {
|
||||
&mut tv
|
||||
.entry(series_id)
|
||||
.or_insert_with(|| (title, year, TvEntry::default()))
|
||||
.2
|
||||
}
|
||||
|
||||
/// TV roll-up (§9.5): every queued series with what put it there — wanted
|
||||
/// episodes whose every candidate was rejected for language, episodes two
|
||||
/// different releases hard-failed post-probe, and seasons whose pack grab
|
||||
/// hard-failed. One entry per series, so the notification can be one per
|
||||
/// series however long the broken season is.
|
||||
async fn queue_tv(database: &Db) -> Result<Vec<(i64, String, Option<i64>, TvEntry)>, sqlx::Error> {
|
||||
let mut tv = HashMap::new();
|
||||
|
||||
let tv_no_pt_source = sqlx::query!(
|
||||
r#"
|
||||
SELECT s.id AS "series_id!: i64", s.title AS "title!: String", s.year,
|
||||
e.id AS "episode_id!: i64"
|
||||
FROM episodes e
|
||||
JOIN seasons se ON se.id = e.season_id
|
||||
JOIN series s ON s.id = se.series_id
|
||||
JOIN roots root ON root.id = s.root_id
|
||||
WHERE root.audience = 'kids'
|
||||
AND s.blocked = 0
|
||||
AND e.wanted = 1 AND e.state = 'missing' AND e.search_attempts > 0
|
||||
AND NOT EXISTS (
|
||||
SELECT 1 FROM episode_releases er
|
||||
JOIN releases r ON r.id = er.release_id
|
||||
WHERE er.episode_id = e.id AND r.verdict IN ('eligible', 'waived')
|
||||
)
|
||||
"#
|
||||
)
|
||||
.fetch_all(database.pool())
|
||||
.await?;
|
||||
for row in tv_no_pt_source {
|
||||
tv_entry(&mut tv, row.series_id, row.title, row.year)
|
||||
.no_pt_source
|
||||
.push(row.episode_id);
|
||||
}
|
||||
|
||||
let episode_hard_fails = sqlx::query!(
|
||||
r#"
|
||||
SELECT s.id AS "series_id!: i64", s.title AS "title!: String", s.year,
|
||||
g.target_id AS "episode_id!: i64"
|
||||
FROM grabs g
|
||||
JOIN episodes e ON g.target_kind = 'episode' AND e.id = g.target_id
|
||||
JOIN seasons se ON se.id = e.season_id
|
||||
JOIN series s ON s.id = se.series_id
|
||||
WHERE g.state = 'failed'
|
||||
GROUP BY s.id, s.title, s.year, e.id
|
||||
HAVING count(DISTINCT g.release_id) >= 2
|
||||
"#
|
||||
)
|
||||
.fetch_all(database.pool())
|
||||
.await?;
|
||||
for row in episode_hard_fails {
|
||||
tv_entry(&mut tv, row.series_id, row.title, row.year)
|
||||
.hard_failed_episodes
|
||||
.push(row.episode_id);
|
||||
}
|
||||
|
||||
let season_pack_fails = sqlx::query!(
|
||||
r#"
|
||||
SELECT s.id AS "series_id!: i64", s.title AS "title!: String", s.year,
|
||||
g.target_id AS "season_id!: i64"
|
||||
FROM grabs g
|
||||
JOIN seasons se ON g.target_kind = 'season' AND se.id = g.target_id
|
||||
JOIN series s ON s.id = se.series_id
|
||||
WHERE g.state = 'failed'
|
||||
GROUP BY s.id, s.title, s.year, se.id
|
||||
"#
|
||||
)
|
||||
.fetch_all(database.pool())
|
||||
.await?;
|
||||
for row in season_pack_fails {
|
||||
tv_entry(&mut tv, row.series_id, row.title, row.year)
|
||||
.failed_season_packs
|
||||
.push(row.season_id);
|
||||
}
|
||||
|
||||
Ok(tv
|
||||
.into_iter()
|
||||
.map(|(id, (t, y, e))| (id, t, y, e))
|
||||
.collect())
|
||||
}
|
||||
|
||||
impl Action for AttentionAction {
|
||||
fn name(&self) -> &'static str {
|
||||
"attention"
|
||||
@@ -168,6 +333,94 @@ mod tests {
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
/// A series on the kids TV root with one season, and `count` wanted,
|
||||
/// missing episodes that have been searched. No stored releases: every
|
||||
/// candidate was rejected for language (§5.2's no-PT-source case).
|
||||
async fn insert_no_pt_source_series(database: &Db, tmdb_id: i64, episode_count: usize) -> i64 {
|
||||
let root_id: i64 =
|
||||
sqlx::query_scalar("SELECT id FROM roots WHERE kind = 'tv' AND audience = 'kids'")
|
||||
.fetch_one(database.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
sqlx::query(
|
||||
"INSERT INTO series (tmdb_id, title, year, root_id) VALUES (?, 'Bluey', 2018, ?)",
|
||||
)
|
||||
.bind(tmdb_id)
|
||||
.bind(root_id)
|
||||
.execute(database.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
let series_id: i64 = sqlx::query_scalar("SELECT id FROM series WHERE tmdb_id = ?")
|
||||
.bind(tmdb_id)
|
||||
.fetch_one(database.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
sqlx::query("INSERT INTO seasons (series_id, number) VALUES (?, 1)")
|
||||
.bind(series_id)
|
||||
.execute(database.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
let season_id: i64 = sqlx::query_scalar("SELECT id FROM seasons WHERE series_id = ?")
|
||||
.bind(series_id)
|
||||
.fetch_one(database.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
for number in 0..episode_count {
|
||||
sqlx::query(
|
||||
"INSERT INTO episodes (season_id, number, title, wanted, state, search_attempts)
|
||||
VALUES (?, ?, ?, 1, 'missing', 1)",
|
||||
)
|
||||
.bind(season_id)
|
||||
.bind(i64::try_from(number).unwrap_or(0) + 1)
|
||||
.bind(format!("Episode {number}"))
|
||||
.execute(database.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
series_id
|
||||
}
|
||||
|
||||
/// A failed grab by `release_guid` against `target_kind`/`target_id`,
|
||||
/// standing in for what the import tick leaves behind post-probe.
|
||||
async fn insert_failed_grab(
|
||||
database: &Db,
|
||||
target_kind: &str,
|
||||
target_id: i64,
|
||||
release_guid: &str,
|
||||
) {
|
||||
let release_id: i64 = sqlx::query_scalar(
|
||||
"INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed, verdict)
|
||||
VALUES (7, ?, 'release', 10737418240, 'https://tracker/x.torrent', '{}', 'eligible')
|
||||
RETURNING id",
|
||||
)
|
||||
.bind(release_guid)
|
||||
.fetch_one(database.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
sqlx::query(
|
||||
"INSERT INTO grabs (release_id, target_kind, target_id, infohash, state)
|
||||
VALUES (?, ?, ?, ?, 'failed')",
|
||||
)
|
||||
.bind(release_id)
|
||||
.bind(target_kind)
|
||||
.bind(target_id)
|
||||
.bind(format!("hash-{release_guid}"))
|
||||
.execute(database.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
async fn action(server: &MockServer) -> AttentionAction {
|
||||
Mock::given(method("POST"))
|
||||
.respond_with(ResponseTemplate::new(200))
|
||||
.mount(server)
|
||||
.await;
|
||||
AttentionAction::new(
|
||||
Notifier::new(server.uri()).unwrap(),
|
||||
"operator-topic".to_string(),
|
||||
)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_movie_entering_the_queue_notifies_the_operator_once() {
|
||||
let (_dir, database) = seeded_database().await;
|
||||
@@ -224,4 +477,88 @@ mod tests {
|
||||
assert_eq!(third.len(), 1, "re-enters and notifies again");
|
||||
assert_eq!(server.received_requests().await.unwrap().len(), 2);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_series_entering_the_queue_notifies_once_per_series() {
|
||||
let (_dir, database) = seeded_database().await;
|
||||
// Twenty broken episodes are one notification, not twenty (§9.5).
|
||||
insert_no_pt_source_series(&database, 1, 20).await;
|
||||
|
||||
let server = MockServer::start().await;
|
||||
let action = action(&server).await;
|
||||
|
||||
let first = action.tick(&database).await.unwrap();
|
||||
let second = action.tick(&database).await.unwrap();
|
||||
|
||||
assert_eq!(first.len(), 1, "one notification for the whole series");
|
||||
assert_eq!(second.len(), 0, "does not repeat while still queued");
|
||||
assert_eq!(server.received_requests().await.unwrap().len(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_series_leaving_and_re_entering_the_queue_notifies_again() {
|
||||
let (_dir, database) = seeded_database().await;
|
||||
let series_id = insert_no_pt_source_series(&database, 1, 1).await;
|
||||
|
||||
let server = MockServer::start().await;
|
||||
let action = action(&server).await;
|
||||
action.tick(&database).await.unwrap();
|
||||
|
||||
sqlx::query("UPDATE series SET blocked = 1 WHERE id = ?")
|
||||
.bind(series_id)
|
||||
.execute(database.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
action.tick(&database).await.unwrap();
|
||||
|
||||
sqlx::query("UPDATE series SET blocked = 0 WHERE id = ?")
|
||||
.bind(series_id)
|
||||
.execute(database.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
let third = action.tick(&database).await.unwrap();
|
||||
|
||||
assert_eq!(third.len(), 1, "re-enters and notifies again");
|
||||
assert_eq!(server.received_requests().await.unwrap().len(), 2);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn an_episode_hard_failed_twice_notifies_its_series() {
|
||||
let (_dir, database) = seeded_database().await;
|
||||
insert_no_pt_source_series(&database, 1, 0).await;
|
||||
let season_id: i64 = sqlx::query_scalar("SELECT id FROM seasons WHERE number = 1")
|
||||
.fetch_one(database.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
sqlx::query(
|
||||
"INSERT INTO episodes (season_id, number, title, wanted, state)
|
||||
VALUES (?, 1, 'Episode 0', 1, 'missing')",
|
||||
)
|
||||
.bind(season_id)
|
||||
.execute(database.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
let episode_id: i64 =
|
||||
sqlx::query_scalar("SELECT id FROM episodes WHERE season_id = ? AND number = 1")
|
||||
.bind(season_id)
|
||||
.fetch_one(database.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
insert_failed_grab(&database, "episode", episode_id, "first").await;
|
||||
insert_failed_grab(&database, "episode", episode_id, "second").await;
|
||||
// The pack's failure sent this season back to per-episode grabbing;
|
||||
// it queues the same series, so it must not double the message.
|
||||
insert_failed_grab(&database, "season", season_id, "pack").await;
|
||||
|
||||
let server = MockServer::start().await;
|
||||
let action = action(&server).await;
|
||||
|
||||
let outcomes = action.tick(&database).await.unwrap();
|
||||
|
||||
assert_eq!(
|
||||
outcomes.len(),
|
||||
1,
|
||||
"both hard-fail conditions roll up to one series notification"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
+17
-1
@@ -11,9 +11,21 @@ export interface AttentionMovie extends LibraryMovie {
|
||||
last_searched_at: string | null;
|
||||
}
|
||||
|
||||
/** A series in a TV attention queue, with what put it there (§9.5). */
|
||||
export interface SeriesAttention {
|
||||
series_id: number;
|
||||
tmdb_id: number;
|
||||
title: string;
|
||||
year: number | null;
|
||||
episodes: number[];
|
||||
seasons: number[];
|
||||
}
|
||||
|
||||
export interface AttentionQueues {
|
||||
no_pt_source: AttentionMovie[];
|
||||
needs_decision: AttentionMovie[];
|
||||
tv_no_pt_source: SeriesAttention[];
|
||||
tv_needs_decision: SeriesAttention[];
|
||||
}
|
||||
|
||||
export type AttentionOutcome =
|
||||
@@ -66,7 +78,11 @@ export async function allowEnglishAudio(movieId: number): Promise<AllowEnglishOu
|
||||
return { kind: "done", searchQueued: true };
|
||||
}
|
||||
|
||||
/** Entries across both queues — the rail badge's number. */
|
||||
/** Entries across both queues — the rail badge's number.
|
||||
*
|
||||
* Deliberately movies only for now: the TV lanes are new and the QUEUES
|
||||
* deck does not render them yet (the backend landed in issue 126).
|
||||
*/
|
||||
export function attentionTotal(queues: AttentionQueues): number {
|
||||
return queues.no_pt_source.length + queues.needs_decision.length;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user