diff --git a/.sqlx/query-0a3680a6cf353c210c1609bef5061f5026f61256c901564548d56f816dd4c68d.json b/.sqlx/query-0a3680a6cf353c210c1609bef5061f5026f61256c901564548d56f816dd4c68d.json new file mode 100644 index 0000000..929e086 --- /dev/null +++ b/.sqlx/query-0a3680a6cf353c210c1609bef5061f5026f61256c901564548d56f816dd4c68d.json @@ -0,0 +1,38 @@ +{ + "db_name": "SQLite", + "query": "\n SELECT se.id AS \"season_id!: i64\",\n se.number AS \"season_number!: i64\",\n s.title AS \"series_title!: String\",\n s.original_language\n FROM seasons se\n JOIN series s ON s.id = se.series_id\n WHERE s.blocked = 0\n AND EXISTS (\n SELECT 1 FROM episodes e\n WHERE e.season_id = se.id\n AND e.wanted = 1\n AND NOT EXISTS (\n SELECT 1 FROM media_files f\n WHERE f.owner_kind = 'episode' AND f.owner_id = e.id\n )\n AND NOT EXISTS (\n SELECT 1 FROM grabs g\n WHERE g.target_kind = 'episode' AND g.target_id = e.id\n AND g.state IN ('sent', 'downloaded', 'imported')\n )\n )\n AND NOT EXISTS (\n SELECT 1 FROM grabs g\n WHERE g.target_kind = 'season' AND g.target_id = se.id\n AND g.state IN ('sent', 'downloaded', 'imported')\n )\n ORDER BY se.id\n ", + "describe": { + "columns": [ + { + "name": "season_id!: i64", + "ordinal": 0, + "type_info": "Integer" + }, + { + "name": "season_number!: i64", + "ordinal": 1, + "type_info": "Integer" + }, + { + "name": "series_title!: String", + "ordinal": 2, + "type_info": "Text" + }, + { + "name": "original_language", + "ordinal": 3, + "type_info": "Text" + } + ], + "parameters": { + "Right": 0 + }, + "nullable": [ + false, + false, + false, + true + ] + }, + "hash": "0a3680a6cf353c210c1609bef5061f5026f61256c901564548d56f816dd4c68d" +} diff --git a/.sqlx/query-08313d5c002d1355542a40eb94b905f0091229e0433b038fdb55a49e557efa06.json b/.sqlx/query-107d5702236ede18bc5a88daedd0fff699d5c80399c0d0139e1e89732b58b169.json similarity index 71% rename from .sqlx/query-08313d5c002d1355542a40eb94b905f0091229e0433b038fdb55a49e557efa06.json rename to .sqlx/query-107d5702236ede18bc5a88daedd0fff699d5c80399c0d0139e1e89732b58b169.json index 4910fcb..0f3b42b 100644 --- a/.sqlx/query-08313d5c002d1355542a40eb94b905f0091229e0433b038fdb55a49e557efa06.json +++ b/.sqlx/query-107d5702236ede18bc5a88daedd0fff699d5c80399c0d0139e1e89732b58b169.json @@ -1,6 +1,6 @@ { "db_name": "SQLite", - "query": "SELECT id AS \"id!: i64\", infohash AS \"infohash!: String\", target_id AS \"target_id!: i64\"\n FROM grabs WHERE state = 'sent'", + "query": "SELECT id AS \"id!: i64\", infohash AS \"infohash!: String\",\n target_kind AS \"target_kind!: String\", target_id AS \"target_id!: i64\"\n FROM grabs WHERE state = 'sent'", "describe": { "columns": [ { @@ -14,8 +14,13 @@ "type_info": "Text" }, { - "name": "target_id!: i64", + "name": "target_kind!: String", "ordinal": 2, + "type_info": "Text" + }, + { + "name": "target_id!: i64", + "ordinal": 3, "type_info": "Integer" } ], @@ -25,8 +30,9 @@ "nullable": [ true, false, + false, false ] }, - "hash": "08313d5c002d1355542a40eb94b905f0091229e0433b038fdb55a49e557efa06" + "hash": "107d5702236ede18bc5a88daedd0fff699d5c80399c0d0139e1e89732b58b169" } diff --git a/.sqlx/query-0e3b37aa1da32a0a5266c3841012fae52ce6394fcdc6c8e42eaeec0d7d6fb663.json b/.sqlx/query-12f31ea2737b9a80b7d23b4c4029bbfebb0291b73f6eb5ec6084b05c38ae753f.json similarity index 53% rename from .sqlx/query-0e3b37aa1da32a0a5266c3841012fae52ce6394fcdc6c8e42eaeec0d7d6fb663.json rename to .sqlx/query-12f31ea2737b9a80b7d23b4c4029bbfebb0291b73f6eb5ec6084b05c38ae753f.json index 450ad73..c58c142 100644 --- a/.sqlx/query-0e3b37aa1da32a0a5266c3841012fae52ce6394fcdc6c8e42eaeec0d7d6fb663.json +++ b/.sqlx/query-12f31ea2737b9a80b7d23b4c4029bbfebb0291b73f6eb5ec6084b05c38ae753f.json @@ -1,6 +1,6 @@ { "db_name": "SQLite", - "query": "INSERT INTO grabs (release_id, target_kind, target_id, infohash, state)\n VALUES (?, 'movie', ?, ?, 'sent')\n ON CONFLICT (infohash) DO NOTHING\n RETURNING id AS \"id!: i64\"", + "query": "INSERT INTO grabs (release_id, target_kind, target_id, infohash, state)\n VALUES (?, ?, ?, ?, 'sent')\n ON CONFLICT (infohash) DO NOTHING\n RETURNING id AS \"id!: i64\"", "describe": { "columns": [ { @@ -10,11 +10,11 @@ } ], "parameters": { - "Right": 3 + "Right": 4 }, "nullable": [ true ] }, - "hash": "0e3b37aa1da32a0a5266c3841012fae52ce6394fcdc6c8e42eaeec0d7d6fb663" + "hash": "12f31ea2737b9a80b7d23b4c4029bbfebb0291b73f6eb5ec6084b05c38ae753f" } diff --git a/.sqlx/query-33de7e17f5defa761de0d6e174dca40e2d883b9990bec497355422f5b5e6dfe5.json b/.sqlx/query-33de7e17f5defa761de0d6e174dca40e2d883b9990bec497355422f5b5e6dfe5.json new file mode 100644 index 0000000..2d5e8f5 --- /dev/null +++ b/.sqlx/query-33de7e17f5defa761de0d6e174dca40e2d883b9990bec497355422f5b5e6dfe5.json @@ -0,0 +1,62 @@ +{ + "db_name": "SQLite", + "query": "\n SELECT e.id AS \"id!: i64\",\n e.number AS \"number!: i64\",\n e.air_date,\n e.wanted AS \"wanted!: bool\",\n e.search_attempts AS \"search_attempts!: i64\",\n e.last_searched_at,\n EXISTS (\n SELECT 1 FROM media_files f\n WHERE f.owner_kind = 'episode' AND f.owner_id = e.id\n ) AS \"has_file!: bool\",\n EXISTS (\n SELECT 1 FROM grabs g\n WHERE g.target_kind = 'episode' AND g.target_id = e.id\n AND g.state IN ('sent', 'downloaded', 'imported')\n ) AS \"in_flight!: bool\"\n FROM episodes e\n WHERE e.season_id = ?\n ORDER BY e.number\n ", + "describe": { + "columns": [ + { + "name": "id!: i64", + "ordinal": 0, + "type_info": "Integer" + }, + { + "name": "number!: i64", + "ordinal": 1, + "type_info": "Integer" + }, + { + "name": "air_date", + "ordinal": 2, + "type_info": "Text" + }, + { + "name": "wanted!: bool", + "ordinal": 3, + "type_info": "Integer" + }, + { + "name": "search_attempts!: i64", + "ordinal": 4, + "type_info": "Integer" + }, + { + "name": "last_searched_at", + "ordinal": 5, + "type_info": "Text" + }, + { + "name": "has_file!: bool", + "ordinal": 6, + "type_info": "Null" + }, + { + "name": "in_flight!: bool", + "ordinal": 7, + "type_info": "Null" + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + true, + false, + true, + false, + false, + true, + null, + null + ] + }, + "hash": "33de7e17f5defa761de0d6e174dca40e2d883b9990bec497355422f5b5e6dfe5" +} diff --git a/.sqlx/query-3d4bd62a88c610eb85c3dc0e39be8d485fa355c2febd8e9030ca8f69ae6a4f38.json b/.sqlx/query-3d4bd62a88c610eb85c3dc0e39be8d485fa355c2febd8e9030ca8f69ae6a4f38.json new file mode 100644 index 0000000..de8d225 --- /dev/null +++ b/.sqlx/query-3d4bd62a88c610eb85c3dc0e39be8d485fa355c2febd8e9030ca8f69ae6a4f38.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "INSERT INTO media_files (owner_kind, owner_id, path, size, probed, waiver)\n VALUES ('episode', ?, ?, ?, ?, ?)\n ON CONFLICT (path) DO UPDATE SET\n size = excluded.size,\n probed = excluded.probed,\n waiver = excluded.waiver,\n updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')", + "describe": { + "columns": [], + "parameters": { + "Right": 5 + }, + "nullable": [] + }, + "hash": "3d4bd62a88c610eb85c3dc0e39be8d485fa355c2febd8e9030ca8f69ae6a4f38" +} diff --git a/.sqlx/query-67719bd71584183785d05f28095debebac93f27189194b680dca887dea725dd9.json b/.sqlx/query-67719bd71584183785d05f28095debebac93f27189194b680dca887dea725dd9.json new file mode 100644 index 0000000..cc0cf1f --- /dev/null +++ b/.sqlx/query-67719bd71584183785d05f28095debebac93f27189194b680dca887dea725dd9.json @@ -0,0 +1,68 @@ +{ + "db_name": "SQLite", + "query": "\n SELECT g.id AS \"grab_id!: i64\",\n g.infohash AS \"infohash!: String\",\n se.id AS \"season_id!: i64\",\n se.number AS \"season_number!: i64\",\n s.tmdb_id AS \"series_tmdb_id!: i64\",\n s.title AS \"series_title!: String\",\n s.year AS \"series_year\",\n s.original_language,\n r.name AS \"release_name!: String\"\n FROM grabs g\n JOIN seasons se ON se.id = g.target_id\n JOIN series s ON s.id = se.series_id\n JOIN releases r ON r.id = g.release_id\n WHERE g.state = 'downloaded' AND g.target_kind = 'season'\n ORDER BY g.id\n ", + "describe": { + "columns": [ + { + "name": "grab_id!: i64", + "ordinal": 0, + "type_info": "Integer" + }, + { + "name": "infohash!: String", + "ordinal": 1, + "type_info": "Text" + }, + { + "name": "season_id!: i64", + "ordinal": 2, + "type_info": "Integer" + }, + { + "name": "season_number!: i64", + "ordinal": 3, + "type_info": "Integer" + }, + { + "name": "series_tmdb_id!: i64", + "ordinal": 4, + "type_info": "Integer" + }, + { + "name": "series_title!: String", + "ordinal": 5, + "type_info": "Text" + }, + { + "name": "series_year", + "ordinal": 6, + "type_info": "Integer" + }, + { + "name": "original_language", + "ordinal": 7, + "type_info": "Text" + }, + { + "name": "release_name!: String", + "ordinal": 8, + "type_info": "Text" + } + ], + "parameters": { + "Right": 0 + }, + "nullable": [ + true, + false, + false, + false, + false, + false, + true, + true, + false + ] + }, + "hash": "67719bd71584183785d05f28095debebac93f27189194b680dca887dea725dd9" +} diff --git a/.sqlx/query-6c8f695a08095b3941a7b6196830aa04462a1e71a2b5ee823e050104cf1b1889.json b/.sqlx/query-6c8f695a08095b3941a7b6196830aa04462a1e71a2b5ee823e050104cf1b1889.json new file mode 100644 index 0000000..5e5f32c --- /dev/null +++ b/.sqlx/query-6c8f695a08095b3941a7b6196830aa04462a1e71a2b5ee823e050104cf1b1889.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "UPDATE episodes\n SET search_attempts = search_attempts + 1,\n last_searched_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now'),\n updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')\n WHERE id = ?", + "describe": { + "columns": [], + "parameters": { + "Right": 1 + }, + "nullable": [] + }, + "hash": "6c8f695a08095b3941a7b6196830aa04462a1e71a2b5ee823e050104cf1b1889" +} diff --git a/.sqlx/query-aa1739e8bc9d2dd48fcfc3238d25187a1f48546af2c05cd618ba1c1b9f7ac791.json b/.sqlx/query-792226f98ff3c19945154c7fda9e36f1fcf4fea39541f19989ee312da55c9f47.json similarity index 51% rename from .sqlx/query-aa1739e8bc9d2dd48fcfc3238d25187a1f48546af2c05cd618ba1c1b9f7ac791.json rename to .sqlx/query-792226f98ff3c19945154c7fda9e36f1fcf4fea39541f19989ee312da55c9f47.json index 5dc3a8e..c0380d3 100644 --- a/.sqlx/query-aa1739e8bc9d2dd48fcfc3238d25187a1f48546af2c05cd618ba1c1b9f7ac791.json +++ b/.sqlx/query-792226f98ff3c19945154c7fda9e36f1fcf4fea39541f19989ee312da55c9f47.json @@ -1,6 +1,6 @@ { "db_name": "SQLite", - "query": "UPDATE movies SET state = 'downloading',\n updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')\n WHERE id = ?", + "query": "UPDATE movies SET state = 'downloading',\n updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')\n WHERE id = ?", "describe": { "columns": [], "parameters": { @@ -8,5 +8,5 @@ }, "nullable": [] }, - "hash": "aa1739e8bc9d2dd48fcfc3238d25187a1f48546af2c05cd618ba1c1b9f7ac791" + "hash": "792226f98ff3c19945154c7fda9e36f1fcf4fea39541f19989ee312da55c9f47" } diff --git a/.sqlx/query-7cf1d44cc741e98ec06438256b593a70882f3274bed650f7691ff4850573a428.json b/.sqlx/query-7cf1d44cc741e98ec06438256b593a70882f3274bed650f7691ff4850573a428.json new file mode 100644 index 0000000..62d653e --- /dev/null +++ b/.sqlx/query-7cf1d44cc741e98ec06438256b593a70882f3274bed650f7691ff4850573a428.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "INSERT INTO episode_releases (episode_id, release_id) VALUES (?, ?)\n ON CONFLICT DO NOTHING", + "describe": { + "columns": [], + "parameters": { + "Right": 2 + }, + "nullable": [] + }, + "hash": "7cf1d44cc741e98ec06438256b593a70882f3274bed650f7691ff4850573a428" +} diff --git a/.sqlx/query-8cdf7339d395e21af86bd71e011e5681e0c00b1f9692a7142043067ff64a00a1.json b/.sqlx/query-8cdf7339d395e21af86bd71e011e5681e0c00b1f9692a7142043067ff64a00a1.json new file mode 100644 index 0000000..548b46d --- /dev/null +++ b/.sqlx/query-8cdf7339d395e21af86bd71e011e5681e0c00b1f9692a7142043067ff64a00a1.json @@ -0,0 +1,20 @@ +{ + "db_name": "SQLite", + "query": "SELECT EXISTS (\n SELECT 1 FROM grabs\n WHERE target_kind = 'season' AND target_id = ? AND state = 'failed'\n ) AS \"failed!: bool\"", + "describe": { + "columns": [ + { + "name": "failed!: bool", + "ordinal": 0, + "type_info": "Integer" + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + false + ] + }, + "hash": "8cdf7339d395e21af86bd71e011e5681e0c00b1f9692a7142043067ff64a00a1" +} diff --git a/.sqlx/query-8fac200bd76490e737c10bbaaa968325ea83c6ee558b2291abc5c1fa943334f0.json b/.sqlx/query-8fac200bd76490e737c10bbaaa968325ea83c6ee558b2291abc5c1fa943334f0.json new file mode 100644 index 0000000..3dc77af --- /dev/null +++ b/.sqlx/query-8fac200bd76490e737c10bbaaa968325ea83c6ee558b2291abc5c1fa943334f0.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "UPDATE grabs\n SET state = 'imported',\n imported_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')\n WHERE id = ?", + "describe": { + "columns": [], + "parameters": { + "Right": 1 + }, + "nullable": [] + }, + "hash": "8fac200bd76490e737c10bbaaa968325ea83c6ee558b2291abc5c1fa943334f0" +} diff --git a/.sqlx/query-9318221b7be54520b2ce3b9df1af806f02bc531356b2939e8179ab3be4d1a998.json b/.sqlx/query-9318221b7be54520b2ce3b9df1af806f02bc531356b2939e8179ab3be4d1a998.json new file mode 100644 index 0000000..38e7a74 --- /dev/null +++ b/.sqlx/query-9318221b7be54520b2ce3b9df1af806f02bc531356b2939e8179ab3be4d1a998.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "UPDATE grabs\n SET state = 'imported',\n imported_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')\n WHERE id = ?", + "describe": { + "columns": [], + "parameters": { + "Right": 1 + }, + "nullable": [] + }, + "hash": "9318221b7be54520b2ce3b9df1af806f02bc531356b2939e8179ab3be4d1a998" +} diff --git a/.sqlx/query-a509854dc66a54218a285c1dbcb4b2500a1f9b529d98bc37a39c7da861a4f3dc.json b/.sqlx/query-a509854dc66a54218a285c1dbcb4b2500a1f9b529d98bc37a39c7da861a4f3dc.json new file mode 100644 index 0000000..030a440 --- /dev/null +++ b/.sqlx/query-a509854dc66a54218a285c1dbcb4b2500a1f9b529d98bc37a39c7da861a4f3dc.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "UPDATE episodes\n SET state = 'missing',\n updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')\n WHERE season_id = ? AND state = 'downloading'\n AND NOT EXISTS (\n SELECT 1 FROM media_files f\n WHERE f.owner_kind = 'episode' AND f.owner_id = episodes.id\n )", + "describe": { + "columns": [], + "parameters": { + "Right": 1 + }, + "nullable": [] + }, + "hash": "a509854dc66a54218a285c1dbcb4b2500a1f9b529d98bc37a39c7da861a4f3dc" +} diff --git a/.sqlx/query-a71c62a2e9cb2ba19f37427347dbf37d194fed7b272371131f07b3ac4eb0bf33.json b/.sqlx/query-a71c62a2e9cb2ba19f37427347dbf37d194fed7b272371131f07b3ac4eb0bf33.json new file mode 100644 index 0000000..565a89a --- /dev/null +++ b/.sqlx/query-a71c62a2e9cb2ba19f37427347dbf37d194fed7b272371131f07b3ac4eb0bf33.json @@ -0,0 +1,74 @@ +{ + "db_name": "SQLite", + "query": "\n SELECT g.id AS \"grab_id!: i64\",\n g.infohash AS \"infohash!: String\",\n e.id AS \"episode_id!: i64\",\n se.id AS \"season_id!: i64\",\n se.number AS \"season_number!: i64\",\n s.tmdb_id AS \"series_tmdb_id!: i64\",\n s.title AS \"series_title!: String\",\n s.year AS \"series_year\",\n s.original_language,\n r.name AS \"release_name!: String\"\n FROM grabs g\n JOIN episodes e ON 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 JOIN releases r ON r.id = g.release_id\n WHERE g.state = 'downloaded' AND g.target_kind = 'episode'\n ORDER BY g.id\n ", + "describe": { + "columns": [ + { + "name": "grab_id!: i64", + "ordinal": 0, + "type_info": "Integer" + }, + { + "name": "infohash!: String", + "ordinal": 1, + "type_info": "Text" + }, + { + "name": "episode_id!: i64", + "ordinal": 2, + "type_info": "Integer" + }, + { + "name": "season_id!: i64", + "ordinal": 3, + "type_info": "Integer" + }, + { + "name": "season_number!: i64", + "ordinal": 4, + "type_info": "Integer" + }, + { + "name": "series_tmdb_id!: i64", + "ordinal": 5, + "type_info": "Integer" + }, + { + "name": "series_title!: String", + "ordinal": 6, + "type_info": "Text" + }, + { + "name": "series_year", + "ordinal": 7, + "type_info": "Integer" + }, + { + "name": "original_language", + "ordinal": 8, + "type_info": "Text" + }, + { + "name": "release_name!: String", + "ordinal": 9, + "type_info": "Text" + } + ], + "parameters": { + "Right": 0 + }, + "nullable": [ + true, + false, + false, + false, + false, + false, + false, + true, + true, + false + ] + }, + "hash": "a71c62a2e9cb2ba19f37427347dbf37d194fed7b272371131f07b3ac4eb0bf33" +} diff --git a/.sqlx/query-b9511e409f9521914406f522c9cf63bf97436d5af605413f8660066d4779f6ca.json b/.sqlx/query-b9511e409f9521914406f522c9cf63bf97436d5af605413f8660066d4779f6ca.json new file mode 100644 index 0000000..5bee8dc --- /dev/null +++ b/.sqlx/query-b9511e409f9521914406f522c9cf63bf97436d5af605413f8660066d4779f6ca.json @@ -0,0 +1,38 @@ +{ + "db_name": "SQLite", + "query": "\n SELECT e.id AS \"id!: i64\",\n e.number AS \"number!: i64\",\n e.title AS \"title!: String\",\n EXISTS (\n SELECT 1 FROM media_files f\n WHERE f.owner_kind = 'episode' AND f.owner_id = e.id\n ) AS \"has_file!: bool\"\n FROM episodes e\n WHERE e.season_id = ?\n ORDER BY e.number\n ", + "describe": { + "columns": [ + { + "name": "id!: i64", + "ordinal": 0, + "type_info": "Integer" + }, + { + "name": "number!: i64", + "ordinal": 1, + "type_info": "Integer" + }, + { + "name": "title!: String", + "ordinal": 2, + "type_info": "Text" + }, + { + "name": "has_file!: bool", + "ordinal": 3, + "type_info": "Null" + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + true, + false, + false, + null + ] + }, + "hash": "b9511e409f9521914406f522c9cf63bf97436d5af605413f8660066d4779f6ca" +} diff --git a/.sqlx/query-c9f250c5b9056ab7e5199f3e3d4553a3996fccc0951b7214562eeb01b554dc58.json b/.sqlx/query-c9f250c5b9056ab7e5199f3e3d4553a3996fccc0951b7214562eeb01b554dc58.json new file mode 100644 index 0000000..2cd46a1 --- /dev/null +++ b/.sqlx/query-c9f250c5b9056ab7e5199f3e3d4553a3996fccc0951b7214562eeb01b554dc58.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "UPDATE episodes SET state = 'downloading',\n updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')\n WHERE id = ?", + "describe": { + "columns": [], + "parameters": { + "Right": 1 + }, + "nullable": [] + }, + "hash": "c9f250c5b9056ab7e5199f3e3d4553a3996fccc0951b7214562eeb01b554dc58" +} diff --git a/.sqlx/query-cc8029578a0bb2d6d95c190922a34c0d8bc2a831f839975f1b5b737e2c04ce8e.json b/.sqlx/query-cc8029578a0bb2d6d95c190922a34c0d8bc2a831f839975f1b5b737e2c04ce8e.json new file mode 100644 index 0000000..093416f --- /dev/null +++ b/.sqlx/query-cc8029578a0bb2d6d95c190922a34c0d8bc2a831f839975f1b5b737e2c04ce8e.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "UPDATE episodes\n SET state = 'available',\n updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')\n WHERE id = ?", + "describe": { + "columns": [], + "parameters": { + "Right": 1 + }, + "nullable": [] + }, + "hash": "cc8029578a0bb2d6d95c190922a34c0d8bc2a831f839975f1b5b737e2c04ce8e" +} diff --git a/.sqlx/query-d97e6b799312c8659f3a78532be26f8183b563e011c4e059747da0fa82fb54d6.json b/.sqlx/query-d97e6b799312c8659f3a78532be26f8183b563e011c4e059747da0fa82fb54d6.json new file mode 100644 index 0000000..f115ee7 --- /dev/null +++ b/.sqlx/query-d97e6b799312c8659f3a78532be26f8183b563e011c4e059747da0fa82fb54d6.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "UPDATE episodes\n SET state = 'missing',\n updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')\n WHERE id = ?", + "describe": { + "columns": [], + "parameters": { + "Right": 1 + }, + "nullable": [] + }, + "hash": "d97e6b799312c8659f3a78532be26f8183b563e011c4e059747da0fa82fb54d6" +} diff --git a/crates/arr-core/src/grabbing.rs b/crates/arr-core/src/grabbing.rs new file mode 100644 index 0000000..7f8ec4e --- /dev/null +++ b/crates/arr-core/src/grabbing.rs @@ -0,0 +1,123 @@ +//! Season-pack versus per-episode grab selection (`DESIGN.md` §13 phase 6). +//! +//! The operator's rule: season packs only when the season is fully released; +//! while a season is airing, grab per episode. A completed season with no +//! episodes on disk prefers the pack — one torrent, better seeded, consistent +//! encode. A pack that hard-failed must not cost the whole season, so the +//! season falls back to per-episode instead of being blacklisted outright. +//! +//! Re-grabbing a pack once an airing season completes is deliberately not +//! done (§14): a season with any episode already on disk grabs per episode. + +use std::time::SystemTime; + +/// How a season's missing wanted episodes should be grabbed next. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum SeasonGrabMode { + /// One season-pack torrent for the whole season. + SeasonPack, + /// One grab per aired wanted episode. + PerEpisode, +} + +/// Everything the season-pack decision depends on. +#[derive(Clone, Copy, Debug)] +pub struct SeasonGrabFacts<'a> { + /// One entry per episode the season is known to hold, aired or not. + /// `None` is an announced episode with no date yet. + pub air_dates: &'a [Option], + pub now: SystemTime, + /// Whether any episode of the season already has a file (§14: nothing + /// re-grabs a pack over episodes on disk, and a pack must not re-import + /// what exists). + pub any_episode_on_disk: bool, + /// Whether a season-pack grab for this season already hard-failed. + pub pack_hard_failed: bool, +} + +/// Picks the grab mode for one season. +/// +/// A season is fully released only when every known episode has an air date +/// in the past. An episode with no date could still be unaired, and grabbing +/// a "complete" pack of a season that is not complete costs a whole torrent +/// of the wrong thing — so an undated episode keeps the season per-episode. +#[must_use] +pub fn season_grab_mode(facts: &SeasonGrabFacts<'_>) -> SeasonGrabMode { + let fully_released = !facts.air_dates.is_empty() + && facts + .air_dates + .iter() + .all(|date| date.is_some_and(|date| date <= facts.now)); + + if fully_released && !facts.any_episode_on_disk && !facts.pack_hard_failed { + SeasonGrabMode::SeasonPack + } else { + SeasonGrabMode::PerEpisode + } +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use super::*; + + const DAY: Duration = Duration::from_hours(24); + + fn facts(air_dates: &[Option]) -> SeasonGrabFacts<'_> { + SeasonGrabFacts { + air_dates, + now: SystemTime::UNIX_EPOCH + 100 * DAY, + any_episode_on_disk: false, + pack_hard_failed: false, + } + } + + #[test] + fn a_fully_released_season_with_nothing_on_disk_takes_the_pack() { + let aired = [ + Some(SystemTime::UNIX_EPOCH + 10 * DAY), + Some(SystemTime::UNIX_EPOCH + 17 * DAY), + ]; + assert_eq!(season_grab_mode(&facts(&aired)), SeasonGrabMode::SeasonPack); + } + + #[test] + fn an_airing_season_grabs_per_episode() { + let airing = [ + Some(SystemTime::UNIX_EPOCH + 10 * DAY), + Some(SystemTime::UNIX_EPOCH + 110 * DAY), + ]; + assert_eq!( + season_grab_mode(&facts(&airing)), + SeasonGrabMode::PerEpisode + ); + } + + #[test] + fn an_undated_episode_keeps_the_season_per_episode() { + let undated = [Some(SystemTime::UNIX_EPOCH + 10 * DAY), None]; + assert_eq!( + season_grab_mode(&facts(&undated)), + SeasonGrabMode::PerEpisode + ); + assert_eq!(season_grab_mode(&facts(&[])), SeasonGrabMode::PerEpisode); + } + + #[test] + fn a_hard_failed_pack_falls_back_to_per_episode() { + let aired = [Some(SystemTime::UNIX_EPOCH + 10 * DAY)]; + let mut facts = facts(&aired); + facts.pack_hard_failed = true; + assert_eq!(season_grab_mode(&facts), SeasonGrabMode::PerEpisode); + } + + /// §14: episodes already on disk are never re-grabbed as part of a pack. + #[test] + fn a_season_with_an_episode_on_disk_grabs_per_episode() { + let aired = [Some(SystemTime::UNIX_EPOCH + 10 * DAY)]; + let mut facts = facts(&aired); + facts.any_episode_on_disk = true; + assert_eq!(season_grab_mode(&facts), SeasonGrabMode::PerEpisode); + } +} diff --git a/crates/arr-core/src/layout.rs b/crates/arr-core/src/layout.rs index b006fda..68a30a1 100644 --- a/crates/arr-core/src/layout.rs +++ b/crates/arr-core/src/layout.rs @@ -50,6 +50,53 @@ pub fn movie_file_name( } } +/// The per-series folder: `Bluey (2018) [tmdbid-82728]`. Same shape as a +/// movie's, and the same reasons for it. +#[must_use] +pub fn series_folder(title: &str, year: Option, tmdb_id: i64) -> String { + movie_folder(title, year, tmdb_id) +} + +/// The per-season folder inside a series folder: `Season 01`. +#[must_use] +pub fn season_folder(season: u16) -> String { + format!("Season {season:02}") +} + +/// An episode's filename: `Bluey (2018) - S01E02 - Hospital [1080p][WEB-DL][pt-PT].mkv`. +/// +/// Unlike the folder there is no provider ID — Jellyfin matches episodes by +/// the `SxxEyy` tag once the folder pinned the series. +#[must_use] +#[allow(clippy::too_many_arguments)] +pub fn episode_file_name( + series_title: &str, + year: Option, + season: u16, + episode: u16, + episode_title: &str, + tags: &[String], + extension: Option<&str>, +) -> String { + let series = sanitise(series_title); + let series = match year { + Some(year) => format!("{series} ({year})"), + None => series, + }; + let episode_title = sanitise(episode_title); + let tags = tags.iter().fold(String::new(), |mut out, tag| { + out.push('['); + out.push_str(tag); + out.push(']'); + out + }); + let stem = format!("{series} - S{season:02}E{episode:02} - {episode_title} {tags}"); + match extension { + Some(extension) => format!("{stem}.{extension}"), + None => stem, + } +} + /// The §7.4 attribute tags, in a fixed order: resolution, source, HDR, /// Portuguese audio. /// @@ -147,6 +194,27 @@ mod tests { ); } + /// The exact §7.4 TV example. + #[test] + fn the_design_document_episode_example() { + let media = probed( + Resolution::R1080p, + HdrFormat::Sdr, + vec![Language::PortuguesePortugal], + ); + let tags = attribute_tags(&media, Some(Source::WebDl)); + + assert_eq!( + series_folder("Bluey", Some(2018), 82_728), + "Bluey (2018) [tmdbid-82728]" + ); + assert_eq!(season_folder(1), "Season 01"); + assert_eq!( + episode_file_name("Bluey", Some(2018), 1, 2, "Hospital", &tags, Some("mkv")), + "Bluey (2018) - S01E02 - Hospital [1080p][WEB-DL][pt-PT].mkv" + ); + } + /// The kids audit surface: a pt-PT track is tagged, SDR is not. #[test] fn portuguese_audio_is_tagged_and_sdr_is_not() { diff --git a/crates/arr-core/src/lib.rs b/crates/arr-core/src/lib.rs index 5bfe270..f34a611 100644 --- a/crates/arr-core/src/lib.rs +++ b/crates/arr-core/src/lib.rs @@ -4,6 +4,7 @@ use std::{collections::BTreeMap, fmt, path::PathBuf, time::SystemTime}; +pub mod grabbing; pub mod lang; pub mod layout; pub mod matching; @@ -13,6 +14,7 @@ pub mod status; pub mod tracking; pub use arr_parse::NameClaims as ParsedRelease; +pub use grabbing::{season_grab_mode, SeasonGrabFacts, SeasonGrabMode}; pub use matching::{match_movie, MatchKind, MovieMatch, WantedMovie}; pub use score::{Score, ScoreWeights}; pub use status::{derive_series_status, SeriesStatus}; diff --git a/crates/arr-daemon/src/grab.rs b/crates/arr-daemon/src/grab.rs index c398895..787f011 100644 --- a/crates/arr-daemon/src/grab.rs +++ b/crates/arr-daemon/src/grab.rs @@ -118,7 +118,7 @@ impl GrabAction { } async fn tick(&self, database: &Db) -> Result, GrabError> { - let mut outcomes = self.track_sent_grabs(database).await?; + let mut outcomes = self.grabber.track_sent_grabs(database).await?; let gaps = pending_movies(database).await?; if gaps.is_empty() { return Ok(outcomes); @@ -247,61 +247,6 @@ impl GrabAction { })) } - /// Move grabs Transmission reports as complete out of `sent`. - /// - /// Transmission is authoritative and its view is rebuilt on every tick - /// rather than cached (§8), so this is also what reconstructs in-flight - /// state after a restart. - async fn track_sent_grabs(&self, database: &Db) -> Result, GrabError> { - let sent = sqlx::query!( - r#"SELECT id AS "id!: i64", infohash AS "infohash!: String", target_id AS "target_id!: i64" - FROM grabs WHERE state = 'sent'"# - ) - .fetch_all(database.pool()) - .await?; - if sent.is_empty() { - return Ok(Vec::new()); - } - - let torrents: HashMap = self - .grabber - .transmission - .list_torrents() - .await? - .into_iter() - .map(|torrent| (torrent.hash.to_ascii_lowercase(), torrent.progress)) - .collect(); - - let mut outcomes = Vec::new(); - for grab in sent { - let Some(progress) = torrents.get(&grab.infohash.to_ascii_lowercase()) else { - // Gone from Transmission. Deciding whether that is a failure - // or a manual removal is issue #86's; leaving the row alone - // keeps this tick from re-grabbing behind the operator. - continue; - }; - if *progress < 1.0 { - continue; - } - sqlx::query!( - "UPDATE grabs SET state = 'downloaded' WHERE id = ?", - grab.id - ) - .execute(database.pool()) - .await?; - outcomes.push(Outcome::new( - format!("grab {} downloaded, still marked sent", grab.id), - format!("marked grab {} downloaded", grab.id), - )); - tracing::info!( - grab_id = grab.id, - movie_id = grab.target_id, - "download complete" - ); - } - Ok(outcomes) - } - /// Search every indexer for one title, cache each candidate with its /// verdict and score (§9.3), and return the eligible ones best first. /// @@ -409,7 +354,7 @@ impl GrabAction { .send_winner( database, &GrabTarget { - movie_id: movie.id, + scope: GrabScope::Movie { movie_id: movie.id }, title: &movie.title, counts_as_attempt: true, }, @@ -436,13 +381,57 @@ pub(crate) struct Grabber { /// The title a winning release is being grabbed for. #[derive(Debug)] pub(crate) struct GrabTarget<'a> { - pub(crate) movie_id: i64, + pub(crate) scope: GrabScope, pub(crate) title: &'a str, /// Whether a grab that does not complete counts toward the targeted /// search backoff (§6.2). RSS never backs off, so it passes `false`. pub(crate) counts_as_attempt: bool, } +/// What a grab targets: the `grabs` row's kind and id, plus the episodes the +/// torrent covers — attempts and state changes land on those leaves (§4.1). +#[derive(Debug)] +pub(crate) enum GrabScope { + Movie { + movie_id: i64, + }, + Episode { + episode_id: i64, + }, + /// A season pack: one torrent, one `grabs` row on the season, every + /// missing wanted episode it covers flipped to downloading. + Season { + season_id: i64, + episode_ids: Vec, + }, +} + +impl GrabScope { + fn target_kind(&self) -> &'static str { + match self { + Self::Movie { .. } => "movie", + Self::Episode { .. } => "episode", + Self::Season { .. } => "season", + } + } + + fn target_id(&self) -> i64 { + match self { + Self::Movie { movie_id } => *movie_id, + Self::Episode { episode_id } => *episode_id, + Self::Season { season_id, .. } => *season_id, + } + } + + fn episode_ids(&self) -> &[i64] { + match self { + Self::Movie { .. } => &[], + Self::Episode { episode_id } => std::slice::from_ref(episode_id), + Self::Season { episode_ids, .. } => episode_ids, + } + } +} + impl Grabber { pub(crate) fn new( transmission: TransmissionClient, @@ -456,15 +445,76 @@ impl Grabber { } } + /// Move grabs Transmission reports as complete out of `sent`, whatever + /// they target. + /// + /// Transmission is authoritative and its view is rebuilt on every tick + /// rather than cached (§8), so this is also what reconstructs in-flight + /// state after a restart. Both grab actions call it; whichever runs first + /// does the work and the other finds nothing. + pub(crate) async fn track_sent_grabs(&self, database: &Db) -> Result, GrabError> { + let sent = sqlx::query!( + r#"SELECT id AS "id!: i64", infohash AS "infohash!: String", + target_kind AS "target_kind!: String", target_id AS "target_id!: i64" + FROM grabs WHERE state = 'sent'"# + ) + .fetch_all(database.pool()) + .await?; + if sent.is_empty() { + return Ok(Vec::new()); + } + + let torrents: HashMap = self + .transmission + .list_torrents() + .await? + .into_iter() + .map(|torrent| (torrent.hash.to_ascii_lowercase(), torrent.progress)) + .collect(); + + let mut outcomes = Vec::new(); + for grab in sent { + let Some(progress) = torrents.get(&grab.infohash.to_ascii_lowercase()) else { + // Gone from Transmission. Deciding whether that is a failure + // or a manual removal is issue #86's; leaving the row alone + // keeps this tick from re-grabbing behind the operator. + continue; + }; + if *progress < 1.0 { + continue; + } + sqlx::query!( + "UPDATE grabs SET state = 'downloaded' WHERE id = ?", + grab.id + ) + .execute(database.pool()) + .await?; + outcomes.push(Outcome::new( + format!("grab {} downloaded, still marked sent", grab.id), + format!("marked grab {} downloaded", grab.id), + )); + tracing::info!( + grab_id = grab.id, + target_kind = grab.target_kind, + target_id = grab.target_id, + "download complete" + ); + } + Ok(outcomes) + } + async fn record_attempt( &self, database: &Db, target: &GrabTarget<'_>, ) -> Result<(), GrabError> { - if target.counts_as_attempt { - record_search(database, target.movie_id).await?; + if !target.counts_as_attempt { + return Ok(()); + } + match &target.scope { + GrabScope::Movie { movie_id } => record_search(database, *movie_id).await, + scope => record_episode_search(database, scope.episode_ids()).await, } - Ok(()) } /// Add the winning release to Transmission and record the grab. @@ -513,29 +563,49 @@ impl Grabber { // A duplicate here is the restart case: the torrent was added before // the process died. `DO NOTHING` keeps the original row. + let target_kind = target.scope.target_kind(); + let target_id = target.scope.target_id(); let inserted = sqlx::query!( r#"INSERT INTO grabs (release_id, target_kind, target_id, infohash, state) - VALUES (?, 'movie', ?, ?, 'sent') + VALUES (?, ?, ?, ?, 'sent') ON CONFLICT (infohash) DO NOTHING RETURNING id AS "id!: i64""#, winner.id, - target.movie_id, + target_kind, + target_id, infohash ) .fetch_optional(database.pool()) .await?; - sqlx::query!( - "UPDATE movies SET state = 'downloading', - updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') - WHERE id = ?", - target.movie_id - ) - .execute(database.pool()) - .await?; + match &target.scope { + GrabScope::Movie { movie_id } => { + sqlx::query!( + "UPDATE movies SET state = 'downloading', + updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') + WHERE id = ?", + movie_id + ) + .execute(database.pool()) + .await?; + } + scope => { + for episode_id in scope.episode_ids() { + sqlx::query!( + "UPDATE episodes SET state = 'downloading', + updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') + WHERE id = ?", + episode_id + ) + .execute(database.pool()) + .await?; + } + } + } let Some(inserted) = inserted else { tracing::info!( - movie_id = target.movie_id, + target_kind, + target_id, infohash, "grab already recorded for this torrent" ); @@ -543,7 +613,8 @@ impl Grabber { }; tracing::info!( - movie_id = target.movie_id, + target_kind, + target_id, title = target.title, release = winner.name, score = winner.score, @@ -552,7 +623,7 @@ impl Grabber { "grabbed" ); Ok(Some(Outcome::new( - format!("movie {} wanted with no file", target.movie_id), + format!("{target_kind} {target_id} wanted with no file"), format!("grabbed {} as grab {}", winner.name, inserted.id), ))) } @@ -592,7 +663,7 @@ impl Grabber { // The earlier grab's torrent, still working off its seeding // obligation (§7.3). Nothing here deletes a torrent. tracing::warn!( - movie_id = target.movie_id, + title = target.title, release = release_name, infohash = added.hash, "blacklisted torrent re-listed under a new name; left seeding" @@ -602,7 +673,7 @@ impl Grabber { // obligation and has nothing on disk worth keeping. self.transmission.remove_torrent(added.id, true).await?; tracing::warn!( - movie_id = target.movie_id, + title = target.title, release = release_name, infohash = added.hash, "blacklisted torrent re-listed under a new name; removed" @@ -645,7 +716,7 @@ pub(crate) struct Eligible { pub(crate) indexer_id: i64, pub(crate) guid: String, pub(crate) name: String, - download_url: String, + pub(crate) download_url: String, pub(crate) score: i64, } @@ -703,13 +774,18 @@ async fn pending_movies(database: &Db) -> Result, GrabError> { } fn search_due(movie: &PendingMovie) -> bool { - let Some(last_searched_at) = &movie.last_searched_at else { + backoff_elapsed(movie.search_attempts, movie.last_searched_at.as_deref()) +} + +/// The §6.2 targeted-search backoff, shared by movie and episode search. +pub(crate) fn backoff_elapsed(search_attempts: i64, last_searched_at: Option<&str>) -> bool { + let Some(last_searched_at) = last_searched_at else { return true; }; let Ok(last_searched_at) = chrono::DateTime::parse_from_rfc3339(last_searched_at) else { return true; }; - let backoff = match movie.search_attempts { + let backoff = match search_attempts { 1 => chrono::TimeDelta::hours(1), 2 => chrono::TimeDelta::hours(6), 3 => chrono::TimeDelta::days(1), @@ -756,6 +832,67 @@ pub(crate) async fn store_release( original_language: &Language, blacklist: &Blacklist, ) -> Result, GrabError> { + let (release_id, eligible) = classify_and_store( + database, + release, + policy, + overrides, + original_language, + blacklist, + ) + .await?; + sqlx::query!( + "INSERT INTO movie_releases (movie_id, release_id) VALUES (?, ?) + ON CONFLICT DO NOTHING", + movie_id, + release_id + ) + .execute(database.pool()) + .await?; + Ok(eligible) +} + +/// The TV counterpart: one release row, associated with every episode the +/// claim covers — a season pack matches the whole season. +pub(crate) async fn store_episode_release( + database: &Db, + episode_ids: &[i64], + release: &SearchRelease, + policy: &Policy, + overrides: &TitleOverrides, + original_language: &Language, + blacklist: &Blacklist, +) -> Result, GrabError> { + let (release_id, eligible) = classify_and_store( + database, + release, + policy, + overrides, + original_language, + blacklist, + ) + .await?; + for episode_id in episode_ids { + sqlx::query!( + "INSERT INTO episode_releases (episode_id, release_id) VALUES (?, ?) + ON CONFLICT DO NOTHING", + episode_id, + release_id + ) + .execute(database.pool()) + .await?; + } + Ok(eligible) +} + +async fn classify_and_store( + database: &Db, + release: &SearchRelease, + policy: &Policy, + overrides: &TitleOverrides, + original_language: &Language, + blacklist: &Blacklist, +) -> Result<(i64, Option), GrabError> { let parsed = arr_parse::parse(&release.name); let evaluation = evaluate( policy, @@ -824,26 +961,20 @@ pub(crate) async fn store_release( .fetch_one(database.pool()) .await?; - sqlx::query!( - "INSERT INTO movie_releases (movie_id, release_id) VALUES (?, ?) - ON CONFLICT DO NOTHING", - movie_id, - id - ) - .execute(database.pool()) - .await?; - if verdict != "eligible" { - return Ok(None); + return Ok((id, None)); } - Ok(Some(Eligible { + Ok(( id, - indexer_id: release.indexer_id, - guid: release.guid.clone(), - name: release.name.clone(), - download_url: release.download_url.clone(), - score, - })) + Some(Eligible { + id, + indexer_id: release.indexer_id, + guid: release.guid.clone(), + name: release.name.clone(), + download_url: release.download_url.clone(), + score, + }), + )) } /// Record that the title was searched, so the next tick takes a different one. @@ -861,6 +992,28 @@ async fn record_search(database: &Db, movie_id: i64) -> Result<(), GrabError> { Ok(()) } +/// The episode-level backoff counter (§6.2), one bump per searched episode. A +/// season-pack search touches every episode it was trying to satisfy, so the +/// whole season backs off together. +pub(crate) async fn record_episode_search( + database: &Db, + episode_ids: &[i64], +) -> Result<(), GrabError> { + for episode_id in episode_ids { + sqlx::query!( + "UPDATE episodes + SET search_attempts = search_attempts + 1, + last_searched_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now'), + updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') + WHERE id = ?", + episode_id + ) + .execute(database.pool()) + .await?; + } + Ok(()) +} + /// `movies-main`, `tv-kids` (§7.1). Distinct from Radarr's own labels, so /// both stacks can run against one Transmission. fn label(loaded: &MoviePolicy) -> String { diff --git a/crates/arr-daemon/src/import.rs b/crates/arr-daemon/src/import.rs index 3dff3a2..3c872d2 100644 --- a/crates/arr-daemon/src/import.rs +++ b/crates/arr-daemon/src/import.rs @@ -182,6 +182,18 @@ impl ImportAction { ), } } + for pending in pending_tv_imports(database).await? { + match self.import_tv_one(database, &pending).await { + Ok(Some(outcome)) => outcomes.push(outcome), + Ok(None) => {} + Err(error) => tracing::error!( + grab_id = pending.grab_id, + series = pending.series_title, + %error, + "tv import failed" + ), + } + } Ok(outcomes) } @@ -204,34 +216,12 @@ impl ImportAction { }; let original_language = arr_db::policy::language(original_language); - let Some(content) = self.transmission.torrent_content(&pending.infohash).await? else { - // Gone from Transmission. Whether that is a failure or a manual - // removal is issue #86's call; leave the grab alone. - tracing::warn!( - grab_id = pending.grab_id, - infohash = pending.infohash, - "downloaded grab has no torrent in Transmission; not importing" - ); + let Some(paths) = self + .torrent_paths(pending.grab_id, &pending.infohash) + .await? + else { return Ok(None); }; - // Torrent-declared names are untrusted input: an absolute or - // `..`-carrying entry would escape the download root and get probed — - // and possibly hardlinked — from anywhere on disk. - let paths: Vec = content - .files - .iter() - .filter_map(|file| { - let path = safe_join(&content.download_dir, &file.path); - if path.is_none() { - tracing::warn!( - grab_id = pending.grab_id, - path = %file.path.display(), - "torrent file path escapes the download root; skipping" - ); - } - path - }) - .collect(); // No expected runtime yet: the movies table carries no TMDB runtime, // so feature selection is by size alone (largest readable video). @@ -308,6 +298,289 @@ impl ImportAction { ))) } + /// The torrent's files as safe local paths, or `None` when Transmission + /// no longer has the torrent. + /// + /// Torrent-declared names are untrusted input: an absolute or + /// `..`-carrying entry would escape the download root and get probed — + /// and possibly hardlinked — from anywhere on disk. + async fn torrent_paths( + &self, + grab_id: i64, + infohash: &str, + ) -> Result>, ImportError> { + let Some(content) = self.transmission.torrent_content(infohash).await? else { + // Gone from Transmission. Whether that is a failure or a manual + // removal is issue #86's call; leave the grab alone. + tracing::warn!( + grab_id, + infohash, + "downloaded grab has no torrent in Transmission; not importing" + ); + return Ok(None); + }; + Ok(Some( + content + .files + .iter() + .filter_map(|file| { + let path = safe_join(&content.download_dir, &file.path); + if path.is_none() { + tracing::warn!( + grab_id, + path = %file.path.display(), + "torrent file path escapes the download root; skipping" + ); + } + path + }) + .collect(), + )) + } + + /// Import one downloaded TV grab: a single episode or a season pack. + /// + /// A pack maps each video file to an episode by the `SxxEyy` tag in its + /// own name, then imports the episodes that are missing. Episodes already + /// on disk are skipped, never re-imported. If any mapped file fails the + /// policy hard, the whole pack hard-fails: that release is blacklisted + /// and the episodes reopen as gaps, which the grab selection then fills + /// per episode rather than writing the season off. + async fn import_tv_one( + &self, + database: &Db, + pending: &PendingTvImport, + ) -> Result, ImportError> { + // §5.2: no original language, nothing to judge audio against. + let Some(original_language) = pending.original_language.as_deref() else { + tracing::warn!( + series = pending.series_title, + "no original language yet; not importing" + ); + return Ok(None); + }; + let original_language = arr_db::policy::language(original_language); + + let episodes = target_episodes(database, pending).await?; + let Some(first) = episodes.first() else { + return Ok(None); + }; + let Some(loaded) = database.episode_policy(first.id).await? else { + return Ok(None); + }; + + let Some(paths) = self + .torrent_paths(pending.grab_id, &pending.infohash) + .await? + else { + return Ok(None); + }; + let candidates = self.probe_all(&paths).await?; + let assignments = assign_files(pending, &episodes, candidates); + if assignments.is_empty() { + self.forget_probes(&paths).await; + return self + .hard_fail_tv(database, pending, "no file matches a wanted episode") + .await + .map(Some); + } + + // §5.6 second phase of truth, over every file that would be + // imported, before anything is placed: one hard failure condemns + // the whole release (§5.7), not the episodes. + let mut imports = Vec::new(); + for assignment in assignments { + if assignment.episode.has_file { + // The partial-overlap case: this episode exists on disk and + // is not re-imported, whatever the pack carries for it. + tracing::info!( + grab_id = pending.grab_id, + episode_id = assignment.episode.id, + "episode already on disk; skipping its file in the pack" + ); + continue; + } + let evaluation = evaluate( + &loaded.policy, + &loaded.overrides, + &original_language, + Candidate::PostDownload(&assignment.file.media), + Some(assignment.file.size), + ); + let waiver = match evaluation.verdict { + Verdict::Rejected(rule) => { + self.forget_probes(&paths).await; + return self + .hard_fail_tv(database, pending, &rule.name()) + .await + .map(Some); + } + Verdict::Waived(rule) => Some(rule), + Verdict::Eligible => None, + }; + imports.push((assignment, waiver)); + } + if imports.is_empty() { + // Everything the pack holds is already on disk. Nothing to + // place; the grab is settled. + sqlx::query!( + "UPDATE grabs + SET state = 'imported', + imported_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') + WHERE id = ?", + pending.grab_id + ) + .execute(database.pool()) + .await?; + self.forget_probes(&paths).await; + return Ok(Some(Outcome::new( + format!("grab {} downloaded, not imported", pending.grab_id), + "every episode in the pack was already on disk".to_owned(), + ))); + } + + let imported = self + .place_episodes(database, pending, &loaded.root_path, imports) + .await?; + + sqlx::query!( + "UPDATE grabs + SET state = 'imported', + imported_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') + WHERE id = ?", + pending.grab_id + ) + .execute(database.pool()) + .await?; + self.forget_probes(&paths).await; + self.refresh_jellyfin().await; + Ok(Some(Outcome::new( + format!("grab {} downloaded, not imported", pending.grab_id), + format!( + "imported {imported} episode file(s) of {}", + pending.series_title + ), + ))) + } + + /// Hardlink each judged file into the §7.4 TV layout and settle its rows. + async fn place_episodes( + &self, + database: &Db, + pending: &PendingTvImport, + root_path: &str, + imports: Vec<(Assignment, Option)>, + ) -> Result { + let claimed_source = arr_parse::parse(&pending.release_name) + .source + .map(Source::from); + let season_number = u16::try_from(pending.season_number).unwrap_or_default(); + let mut imported = 0usize; + for (assignment, waiver) in imports { + let episode = &assignment.episode; + let feature = &assignment.file; + let tags = layout::attribute_tags(&feature.media, claimed_source); + let extension = feature.path.extension().and_then(|ext| ext.to_str()); + let destination = Path::new(root_path) + .join(layout::series_folder( + &pending.series_title, + pending.series_year, + pending.series_tmdb_id, + )) + .join(layout::season_folder(season_number)) + .join(layout::episode_file_name( + &pending.series_title, + pending.series_year, + season_number, + u16::try_from(episode.number).unwrap_or_default(), + &episode.title, + &tags, + extension, + )); + + let source_path = feature.path.clone(); + let link_target = destination.clone(); + tokio::task::spawn_blocking(move || place(&source_path, &link_target)).await??; + + record_episode_import(database, episode.id, feature, waiver.as_ref(), &destination) + .await?; + imported += 1; + tracing::info!( + grab_id = pending.grab_id, + episode_id = episode.id, + series = pending.series_title, + path = %destination.display(), + waived = waiver.is_some(), + "imported" + ); + } + Ok(imported) + } + + /// §5.7 hard fail for a TV grab: blacklist the release, fail the grab and + /// reopen only the episodes it was downloading. The season is never + /// blacklisted — grab selection falls back to per-episode. + async fn hard_fail_tv( + &self, + database: &Db, + pending: &PendingTvImport, + reason: &str, + ) -> Result { + arr_db::blacklist::add( + database.pool(), + Some(&pending.infohash), + &pending.release_name, + reason, + ) + .await?; + sqlx::query!( + "UPDATE grabs SET state = 'failed' WHERE id = ?", + pending.grab_id + ) + .execute(database.pool()) + .await?; + match pending.episode_id { + Some(episode_id) => { + sqlx::query!( + "UPDATE episodes + SET state = 'missing', + updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') + WHERE id = ?", + episode_id + ) + .execute(database.pool()) + .await?; + } + None => { + sqlx::query!( + "UPDATE episodes + SET state = 'missing', + updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') + WHERE season_id = ? AND state = 'downloading' + AND NOT EXISTS ( + SELECT 1 FROM media_files f + WHERE f.owner_kind = 'episode' AND f.owner_id = episodes.id + )", + pending.season_id + ) + .execute(database.pool()) + .await?; + } + } + + tracing::warn!( + grab_id = pending.grab_id, + series = pending.series_title, + release = pending.release_name, + reason, + "hard fail post-probe; release blacklisted, episodes reopened, torrent left seeding" + ); + Ok(Outcome::new( + format!("grab {} hard-failed post-probe: {reason}", pending.grab_id), + format!("blacklisted {}", pending.release_name), + )) + } + /// §7.5: the filesystem watcher misses the just-hardlinked file. A /// failure to reach Jellyfin must not fail the import, which has already /// succeeded. @@ -462,6 +735,258 @@ async fn pending_imports(database: &Db) -> Result, ImportErro .collect()) } +/// A TV grab Transmission finished downloading — one episode or a season +/// pack — not yet imported. +#[derive(Debug, Clone)] +struct PendingTvImport { + grab_id: i64, + infohash: String, + /// `Some` for an episode grab, `None` for a season pack. + episode_id: Option, + season_id: i64, + season_number: i64, + series_tmdb_id: i64, + series_title: String, + series_year: Option, + original_language: Option, + release_name: String, +} + +/// An episode a downloaded TV grab could satisfy. +#[derive(Debug, Clone)] +struct TargetEpisode { + id: i64, + number: i64, + title: String, + has_file: bool, +} + +/// One probed video file tied to the episode it holds. +#[derive(Debug)] +struct Assignment { + episode: TargetEpisode, + file: arr_probe::ProbedFile, +} + +/// The TV side of the gap (§8): downloaded episode and season grabs that no +/// import has settled. +async fn pending_tv_imports(database: &Db) -> Result, ImportError> { + let mut pending = Vec::new(); + + let episode_rows = sqlx::query!( + r#" + SELECT g.id AS "grab_id!: i64", + g.infohash AS "infohash!: String", + e.id AS "episode_id!: i64", + se.id AS "season_id!: i64", + se.number AS "season_number!: i64", + s.tmdb_id AS "series_tmdb_id!: i64", + s.title AS "series_title!: String", + s.year AS "series_year", + s.original_language, + r.name AS "release_name!: String" + FROM grabs g + JOIN episodes e ON e.id = g.target_id + JOIN seasons se ON se.id = e.season_id + JOIN series s ON s.id = se.series_id + JOIN releases r ON r.id = g.release_id + WHERE g.state = 'downloaded' AND g.target_kind = 'episode' + ORDER BY g.id + "# + ) + .fetch_all(database.pool()) + .await?; + pending.extend(episode_rows.into_iter().map(|row| PendingTvImport { + grab_id: row.grab_id, + infohash: row.infohash, + episode_id: Some(row.episode_id), + season_id: row.season_id, + season_number: row.season_number, + series_tmdb_id: row.series_tmdb_id, + series_title: row.series_title, + series_year: row.series_year, + original_language: row.original_language, + release_name: row.release_name, + })); + + let season_rows = sqlx::query!( + r#" + SELECT g.id AS "grab_id!: i64", + g.infohash AS "infohash!: String", + se.id AS "season_id!: i64", + se.number AS "season_number!: i64", + s.tmdb_id AS "series_tmdb_id!: i64", + s.title AS "series_title!: String", + s.year AS "series_year", + s.original_language, + r.name AS "release_name!: String" + FROM grabs g + JOIN seasons se ON se.id = g.target_id + JOIN series s ON s.id = se.series_id + JOIN releases r ON r.id = g.release_id + WHERE g.state = 'downloaded' AND g.target_kind = 'season' + ORDER BY g.id + "# + ) + .fetch_all(database.pool()) + .await?; + pending.extend(season_rows.into_iter().map(|row| PendingTvImport { + grab_id: row.grab_id, + infohash: row.infohash, + episode_id: None, + season_id: row.season_id, + season_number: row.season_number, + series_tmdb_id: row.series_tmdb_id, + series_title: row.series_title, + series_year: row.series_year, + original_language: row.original_language, + release_name: row.release_name, + })); + + pending.sort_by_key(|row| row.grab_id); + Ok(pending) +} + +/// The episodes a grab could satisfy: one for an episode grab, the whole +/// season for a pack. +async fn target_episodes( + database: &Db, + pending: &PendingTvImport, +) -> Result, ImportError> { + let rows = sqlx::query!( + r#" + SELECT e.id AS "id!: i64", + e.number AS "number!: i64", + e.title AS "title!: String", + EXISTS ( + SELECT 1 FROM media_files f + WHERE f.owner_kind = 'episode' AND f.owner_id = e.id + ) AS "has_file!: bool" + FROM episodes e + WHERE e.season_id = ? + ORDER BY e.number + "#, + pending.season_id + ) + .fetch_all(database.pool()) + .await?; + let episodes = rows.into_iter().map(|row| TargetEpisode { + id: row.id, + number: row.number, + title: row.title, + has_file: row.has_file, + }); + Ok(match pending.episode_id { + Some(episode_id) => episodes + .filter(|episode| episode.id == episode_id) + .collect(), + None => episodes.collect(), + }) +} + +/// Tie each readable video file to the episode its own name claims (§5.6: +/// per-file names are the only pre-probe truth a pack carries). +/// +/// A file claiming several episodes lands on the first target it covers, one +/// file per episode, largest file winning a collision. For a single-episode +/// grab whose only video file carries no tag, the file is the episode. +fn assign_files( + pending: &PendingTvImport, + episodes: &[TargetEpisode], + files: Vec, +) -> Vec { + let season = u32::try_from(pending.season_number).unwrap_or_default(); + let mut by_episode: HashMap = HashMap::new(); + let mut untagged: Vec = Vec::new(); + + for file in files { + let name = file.path.file_name().and_then(|name| name.to_str()); + let claim = name.and_then(|name| arr_parse::parse(name).episode); + let Some(claim) = claim else { + untagged.push(file); + continue; + }; + let covered = episodes.iter().find(|episode| { + claim.covers(season, u32::try_from(episode.number).unwrap_or_default()) + }); + let Some(episode) = covered else { + continue; + }; + match by_episode.entry(episode.id) { + std::collections::hash_map::Entry::Vacant(entry) => { + entry.insert(file); + } + std::collections::hash_map::Entry::Occupied(mut entry) => { + if file.size > entry.get().size { + entry.insert(file); + } + } + } + } + + // A single-episode torrent often names its one file after nothing + // useful. One target, one untagged video: that is the episode. + if pending.episode_id.is_some() && by_episode.is_empty() && untagged.len() == 1 { + if let (Some(episode), Some(file)) = (episodes.first(), untagged.pop()) { + by_episode.insert(episode.id, file); + } + } + + let mut assignments: Vec = episodes + .iter() + .filter_map(|episode| { + by_episode.remove(&episode.id).map(|file| Assignment { + episode: episode.clone(), + file, + }) + }) + .collect(); + assignments.sort_by_key(|assignment| assignment.episode.number); + assignments +} + +/// Settle a placed episode file into the rows: the `media_files` record, and +/// the episode itself. The upsert on path is the same crash seam the movie +/// import leans on. +async fn record_episode_import( + database: &Db, + episode_id: i64, + feature: &arr_probe::ProbedFile, + waiver: Option<&Rule>, + destination: &Path, +) -> Result<(), ImportError> { + let probed = probed_json(&feature.media).to_string(); + let waiver_json = waiver.map(|rule| serde_json::json!({ "rule": rule.name() }).to_string()); + let size = i64::try_from(feature.size).unwrap_or(i64::MAX); + let path_text = destination.to_string_lossy().into_owned(); + sqlx::query!( + "INSERT INTO media_files (owner_kind, owner_id, path, size, probed, waiver) + VALUES ('episode', ?, ?, ?, ?, ?) + ON CONFLICT (path) DO UPDATE SET + size = excluded.size, + probed = excluded.probed, + waiver = excluded.waiver, + updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')", + episode_id, + path_text, + size, + probed, + waiver_json + ) + .execute(database.pool()) + .await?; + sqlx::query!( + "UPDATE episodes + SET state = 'available', + updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') + WHERE id = ?", + episode_id + ) + .execute(database.pool()) + .await?; + Ok(()) +} + /// The `probed` column (§4, §5.6): what `ffprobe` found, in the spellings the /// policy columns use. fn probed_json(media: &ProbedMedia) -> serde_json::Value { @@ -1049,6 +1574,276 @@ mod tests { ); } + /// The TV probe: a 2160p HDR10 file with an English track, sized inside + /// the 2160p band. + const TV_HDR10_PROBE: &str = r#"{ + "format": {"format_name": "matroska,webm", "duration": "3300.0", "size": "10737418240"}, + "streams": [ + {"codec_type": "video", "codec_name": "hevc", "width": 3840, "height": 1600, + "color_transfer": "smpte2084"}, + {"codec_type": "audio", "codec_name": "eac3", "tags": {"language": "eng"}} + ] + }"#; + + const TV_DV5_PROBE: &str = r#"{ + "format": {"format_name": "matroska,webm", "duration": "3300.0", "size": "10737418240"}, + "streams": [ + {"codec_type": "video", "codec_name": "hevc", "width": 3840, "height": 1600, + "color_transfer": "smpte2084", + "side_data_list": [{"side_data_type": "DOVI configuration record", "dv_profile": 5}]}, + {"codec_type": "audio", "codec_name": "eac3", "tags": {"language": "eng"}} + ] + }"#; + + const PACK_RELEASE_NAME: &str = "Fallout.S01.2160p.WEB-DL.DDP5.1.Atmos"; + + struct TvHarness { + _dir: tempfile::TempDir, + database: Db, + downloads: PathBuf, + library: PathBuf, + action: ImportAction, + _server: MockServer, + } + + /// A downloaded season-pack grab for Fallout S01E01-E02, its two files + /// sitting in the download root. + async fn tv_harness(media_json: &str) -> TvHarness { + let dir = tempfile::tempdir().unwrap(); + let downloads = dir.path().join("downloads"); + let library = dir.path().join("library"); + std::fs::create_dir_all(downloads.join("Fallout.S01")).unwrap(); + std::fs::create_dir_all(&library).unwrap(); + std::fs::write(downloads.join("Fallout.S01/Fallout.S01E01.mkv"), b"e1").unwrap(); + std::fs::write(downloads.join("Fallout.S01/Fallout.S01E02.mkv"), b"e2").unwrap(); + + let database = Db::connect(dir.path().join("arr.db")).await.unwrap(); + database.migrate().await.unwrap(); + let library_text = library.to_string_lossy().into_owned(); + sqlx::query("UPDATE roots SET path = ? WHERE kind = 'tv' AND audience = 'main'") + .bind(&library_text) + .execute(database.pool()) + .await + .unwrap(); + sqlx::query( + "INSERT INTO series (tmdb_id, title, year, original_language, root_id) + SELECT 106379, 'Fallout', 2024, 'en', id + FROM roots WHERE kind = 'tv' AND audience = 'main'", + ) + .execute(database.pool()) + .await + .unwrap(); + let season_id: i64 = sqlx::query_scalar( + "INSERT INTO seasons (series_id, number) VALUES (1, 1) RETURNING id", + ) + .fetch_one(database.pool()) + .await + .unwrap(); + for number in 1..=2 { + sqlx::query( + "INSERT INTO episodes (season_id, number, title, air_date, wanted, state) + VALUES (?, ?, ?, '2024-04-11', 1, 'downloading')", + ) + .bind(season_id) + .bind(number) + .bind(format!("The Episode {number}")) + .execute(database.pool()) + .await + .unwrap(); + } + let release_id = sqlx::query( + "INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed, verdict) + VALUES (7, 'pack', ?, 85899345920, 'magnet:x', '{}', 'eligible')", + ) + .bind(PACK_RELEASE_NAME) + .execute(database.pool()) + .await + .unwrap() + .last_insert_rowid(); + sqlx::query( + "INSERT INTO grabs (release_id, target_kind, target_id, infohash, state) + VALUES (?, 'season', ?, ?, 'downloaded')", + ) + .bind(release_id) + .bind(season_id) + .bind(INFOHASH) + .execute(database.pool()) + .await + .unwrap(); + + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "result": "success", + "arguments": {"torrents": [{ + "hashString": INFOHASH, + "downloadDir": downloads.to_string_lossy(), + "files": [ + {"name": "Fallout.S01/Fallout.S01E01.mkv", "length": 2, "bytesCompleted": 2}, + {"name": "Fallout.S01/Fallout.S01E02.mkv", "length": 2, "bytesCompleted": 2} + ] + }]} + }))) + .mount(&server) + .await; + let jellyfin_server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/Library/Refresh")) + .respond_with(ResponseTemplate::new(204)) + .mount(&jellyfin_server) + .await; + + let prober = Prober::new().with_binary(fake_ffprobe(dir.path(), media_json)); + let jellyfin = JellyfinClient::new(jellyfin_server.uri(), None).unwrap(); + let action = ImportAction::new( + TransmissionClient::new(&server.uri()).unwrap(), + prober, + jellyfin, + ); + + TvHarness { + _dir: dir, + database, + downloads, + library, + action, + _server: server, + } + } + + fn expected_episode_file(library: &Path, number: u16) -> PathBuf { + library + .join("Fallout (2024) [tmdbid-106379]") + .join("Season 01") + .join(format!( + "Fallout (2024) - S01E{number:02} - The Episode {number} [2160p][WEB-DL][HDR10].mkv" + )) + } + + /// A season pack lands each episode file on the §7.4 TV layout. + #[tokio::test] + async fn a_season_pack_imports_every_episode() { + let h = tv_harness(TV_HDR10_PROBE).await; + + let outcomes = h.action.tick(&h.database).await.unwrap(); + + assert_eq!(outcomes.len(), 1); + for number in 1..=2u16 { + let file = expected_episode_file(&h.library, number); + assert!(file.is_file(), "missing {}", file.display()); + } + let states: Vec = sqlx::query_scalar("SELECT state FROM episodes ORDER BY number") + .fetch_all(h.database.pool()) + .await + .unwrap(); + assert_eq!(states, vec!["available", "available"]); + let grab_state: String = sqlx::query_scalar("SELECT state FROM grabs") + .fetch_one(h.database.pool()) + .await + .unwrap(); + assert_eq!(grab_state, "imported"); + assert!( + h.downloads.join("Fallout.S01/Fallout.S01E01.mkv").is_file(), + "§7.3: the torrent keeps seeding" + ); + } + + /// The fourth acceptance case: a pack containing an episode already on + /// disk must not re-import what exists. + #[tokio::test] + async fn a_season_pack_never_reimports_an_episode_already_on_disk() { + let h = tv_harness(TV_HDR10_PROBE).await; + let existing = h.library.join("existing-e01.mkv"); + std::fs::write(&existing, b"the copy that is already there").unwrap(); + sqlx::query( + "INSERT INTO media_files (owner_kind, owner_id, path, size) + SELECT 'episode', id, ?, 30 FROM episodes WHERE number = 1", + ) + .bind(existing.to_string_lossy().into_owned()) + .execute(h.database.pool()) + .await + .unwrap(); + sqlx::query("UPDATE episodes SET state = 'available' WHERE number = 1") + .execute(h.database.pool()) + .await + .unwrap(); + + let outcomes = h.action.tick(&h.database).await.unwrap(); + + assert_eq!(outcomes.len(), 1); + assert!( + !expected_episode_file(&h.library, 1).exists(), + "episode 1 is on disk already and must not be re-imported" + ); + assert!(expected_episode_file(&h.library, 2).is_file()); + let episode_one_files: Vec<(String, i64)> = sqlx::query_as( + "SELECT f.path, f.size FROM media_files f + JOIN episodes e ON e.id = f.owner_id + WHERE f.owner_kind = 'episode' AND e.number = 1", + ) + .fetch_all(h.database.pool()) + .await + .unwrap(); + assert_eq!( + episode_one_files, + vec![(existing.to_string_lossy().into_owned(), 30)], + "episode 1 keeps exactly its pre-existing file row" + ); + let grab_state: String = sqlx::query_scalar("SELECT state FROM grabs") + .fetch_one(h.database.pool()) + .await + .unwrap(); + assert_eq!(grab_state, "imported"); + } + + /// The third acceptance case, import side: a pack whose file hard-fails + /// blacklists that release and reopens the episodes — it does not + /// blacklist or block the season. + #[tokio::test] + async fn a_hard_failed_pack_reopens_the_season_per_episode() { + let h = tv_harness(TV_DV5_PROBE).await; + + let outcomes = h.action.tick(&h.database).await.unwrap(); + + assert_eq!(outcomes.len(), 1); + assert!( + std::fs::read_dir(&h.library).unwrap().next().is_none(), + "nothing may reach the library" + ); + let blacklist: Vec<(String, String)> = + sqlx::query_as("SELECT normalised_name, reason FROM blacklist") + .fetch_all(h.database.pool()) + .await + .unwrap(); + assert_eq!( + blacklist, + vec![( + arr_parse::normalise(PACK_RELEASE_NAME), + "dolby_vision_profile".to_owned() + )], + "only the release is blacklisted, never the season" + ); + let grab_state: String = sqlx::query_scalar("SELECT state FROM grabs") + .fetch_one(h.database.pool()) + .await + .unwrap(); + assert_eq!(grab_state, "failed"); + let states: Vec<(String, bool)> = + sqlx::query_as("SELECT state, wanted FROM episodes ORDER BY number") + .fetch_all(h.database.pool()) + .await + .unwrap(); + assert_eq!( + states, + vec![("missing".to_owned(), true), ("missing".to_owned(), true)], + "the gap reopens per episode, still wanted" + ); + assert!( + h.downloads.join("Fallout.S01/Fallout.S01E01.mkv").is_file(), + "§7.3: the torrent is untouched" + ); + } + /// The `EXDEV` fallback path lands whole files via rename (§7.2). #[test] fn the_copy_fallback_lands_a_whole_file_and_cleans_up() { diff --git a/crates/arr-daemon/src/indexers.rs b/crates/arr-daemon/src/indexers.rs index cb6b2c0..44659e0 100644 --- a/crates/arr-daemon/src/indexers.rs +++ b/crates/arr-daemon/src/indexers.rs @@ -9,7 +9,7 @@ use std::time::{Duration, Instant}; -use arr_indexer::ProwlarrClient; +use arr_indexer::{Indexer, ProwlarrClient}; /// How long a discovered indexer list is reused. const CACHE_TTL: Duration = Duration::from_mins(15); @@ -30,10 +30,21 @@ pub enum DiscoveryError { #[derive(Debug, Default)] struct Cache { - ids: Vec, + indexers: Vec, refreshed_at: Option, } +impl Cache { + /// Text-searchable ids, the shape the movie lanes consume. + fn ids(&self) -> Vec { + self.indexers + .iter() + .filter(|indexer| indexer.capabilities.search.available) + .map(|indexer| indexer.id) + .collect() + } +} + /// The indexers that accept a text search, cached across ticks. #[derive(Debug)] pub struct IndexerDirectory { @@ -59,13 +70,23 @@ impl IndexerDirectory { /// A refresh that fails or times out keeps the previous list rather than /// stopping the tick, and only an empty cache turns that into an error. pub async fn searchable(&self) -> Result, DiscoveryError> { + Ok(self.refreshed().await?.ids()) + } + + /// Every indexer with its capabilities, for callers that build their own + /// per-indexer request — TV search widens or narrows per `t=caps` (§6.1). + pub async fn indexers(&self) -> Result, DiscoveryError> { + Ok(self.refreshed().await?.indexers.clone()) + } + + async fn refreshed(&self) -> Result, DiscoveryError> { { let cache = self.cache.read().await; if cache .refreshed_at .is_some_and(|at| at.elapsed() < CACHE_TTL) { - return Ok(cache.ids.clone()); + return Ok(cache); } } @@ -74,15 +95,11 @@ impl IndexerDirectory { let mut cache = self.cache.write().await; match discovered { Ok(Ok(indexers)) => { - cache.ids = indexers - .iter() - .filter(|indexer| indexer.capabilities.search.available) - .map(|indexer| indexer.id) - .collect(); + cache.indexers = indexers; cache.refreshed_at = Some(Instant::now()); } - Ok(Err(error)) if cache.ids.is_empty() => return Err(error.into()), - Err(_) if cache.ids.is_empty() => { + Ok(Err(error)) if cache.indexers.is_empty() => return Err(error.into()), + Err(_) if cache.indexers.is_empty() => { return Err(DiscoveryError::Timeout(self.discovery_timeout)) } Ok(Err(error)) => { @@ -93,6 +110,7 @@ impl IndexerDirectory { "indexer discovery timed out; using the last known list" ), } - Ok(cache.ids.clone()) + drop(cache); + Ok(self.cache.read().await) } } diff --git a/crates/arr-daemon/src/main.rs b/crates/arr-daemon/src/main.rs index f261422..95d4dee 100644 --- a/crates/arr-daemon/src/main.rs +++ b/crates/arr-daemon/src/main.rs @@ -8,6 +8,7 @@ mod jellyfin; mod reaper; pub mod reconcile; mod rss; +mod tv_grab; mod web; use std::process::ExitCode; @@ -24,6 +25,7 @@ use reaper::ReaperAction; use reconcile::{ReconcileLoop, Tick}; use rss::RssAction; use tower_http::trace::TraceLayer; +use tv_grab::TvGrabAction; /// Dump the `OpenAPI` document and exit, instead of serving. `just gen-client` /// uses this so the TypeScript client can be regenerated without a port or a @@ -215,6 +217,20 @@ fn reconcile_loop( } else { tracing::warn!("Prowlarr or TMDB is not configured: nothing will be grabbed"); } + // TV grabbing needs no TMDB at grab time: air dates are already on the + // episode rows, which is the same gate the digital release date is for + // movies (§6.2). + if let Some(prowlarr) = prowlarr.as_ref() { + reconcile = reconcile.register( + Tick::Reconcile, + TvGrabAction::new( + prowlarr.clone(), + transmission.clone(), + config.download_dir.clone(), + seeding.clone(), + ), + ); + } // RSS needs no TMDB: it matches what the feeds already carry against the // wanted list (§6.2). if let Some(prowlarr) = prowlarr { diff --git a/crates/arr-daemon/src/rss.rs b/crates/arr-daemon/src/rss.rs index 4007b7f..ad64d86 100644 --- a/crates/arr-daemon/src/rss.rs +++ b/crates/arr-daemon/src/rss.rs @@ -20,7 +20,9 @@ use arr_db::{Blacklist, Db, MoviePolicy}; use arr_dl::TransmissionClient; use arr_indexer::{ProwlarrClient, SearchRelease, SearchRequest}; -use crate::grab::{store_release, Eligible, GrabError, GrabTarget, Grabber, SeedingRules}; +use crate::grab::{ + store_release, Eligible, GrabError, GrabScope, GrabTarget, Grabber, SeedingRules, +}; use crate::indexers::IndexerDirectory; use crate::reconcile::{Action, ActionFuture, Outcome}; @@ -71,7 +73,7 @@ impl RssAction { .send_winner( database, &GrabTarget { - movie_id: movie.id, + scope: GrabScope::Movie { movie_id: movie.id }, title: &movie.title, // §6.2: RSS never backs off, so a failed grab here // must not spend one of targeted search's attempts. diff --git a/crates/arr-daemon/src/tv_grab.rs b/crates/arr-daemon/src/tv_grab.rs new file mode 100644 index 0000000..8afb566 --- /dev/null +++ b/crates/arr-daemon/src/tv_grab.rs @@ -0,0 +1,1098 @@ +//! The TV grab pipeline: season packs versus per-episode grabs. See +//! DESIGN.md §13 phase 6 and `arr_core::grabbing`. +//! +//! The operator's rule: a fully released season with nothing on disk takes +//! one season-pack torrent; an airing season grabs episodes individually as +//! they appear. A pack that hard-failed post-probe blacklists only that +//! release (§6.3) — the season falls back to per-episode instead of being +//! blacklisted outright, and a pack overlapping episodes already on disk is +//! never chosen (§14 defers pack re-grabs entirely). +//! +//! Everything is recomputed from domain rows every tick (§8): an episode +//! with a file or a live grab — its own or its season's — is not a gap. + +use std::path::PathBuf; + +use arr_core::grabbing::{season_grab_mode, SeasonGrabFacts, SeasonGrabMode}; +use arr_core::Language; +use arr_db::{Blacklist, Db, TitlePolicy}; +use arr_dl::TransmissionClient; +use arr_indexer::{ProwlarrClient, SearchRelease, TvSelector, TvTarget}; +use arr_parse::EpisodeClaim; + +use crate::grab::{ + backoff_elapsed, record_episode_search, store_episode_release, Eligible, GrabError, GrabScope, + GrabTarget, Grabber, SeedingRules, +}; +use crate::indexers::IndexerDirectory; +use crate::reconcile::{Action, ActionFuture, Outcome}; + +/// How many seasons one tick may search. Same budget reasoning as the movie +/// lane: one search costs one call per indexer (§6.2). +const SEASONS_PER_TICK: usize = 3; + +/// How many per-episode searches one tick may run across all seasons. +const EPISODE_SEARCHES_PER_TICK: usize = 5; + +/// Grabs season packs or single episodes for every season with wanted, +/// missing episodes. +#[derive(Debug)] +pub struct TvGrabAction { + prowlarr: ProwlarrClient, + grabber: Grabber, + indexers: IndexerDirectory, +} + +impl TvGrabAction { + #[must_use] + pub fn new( + prowlarr: ProwlarrClient, + transmission: TransmissionClient, + download_dir: PathBuf, + seeding: SeedingRules, + ) -> Self { + Self { + indexers: IndexerDirectory::new(prowlarr.clone()), + prowlarr, + grabber: Grabber::new(transmission, download_dir, seeding), + } + } + + async fn tick(&self, database: &Db) -> Result, GrabError> { + let mut outcomes = self.grabber.track_sent_grabs(database).await?; + let seasons = pending_seasons(database).await?; + if seasons.is_empty() { + return Ok(outcomes); + } + let blacklist = Blacklist::load(database.pool()).await?; + + let mut searched_seasons = 0; + let mut episode_searches = 0; + for season in seasons { + if searched_seasons >= SEASONS_PER_TICK || episode_searches >= EPISODE_SEARCHES_PER_TICK + { + break; + } + match self + .grab_season( + database, + &season, + &blacklist, + EPISODE_SEARCHES_PER_TICK - episode_searches, + ) + .await + { + Ok(Some(work)) => { + searched_seasons += 1; + episode_searches += work.episode_searches; + outcomes.extend(work.outcomes); + } + Ok(None) => {} + // One season's failure must not cost the rest of the tick. + Err(error) => tracing::error!( + season_id = season.season_id, + series = season.series_title, + %error, + "tv grab failed" + ), + } + } + Ok(outcomes) + } + + /// Search and grab for one season. `None` means the season was skipped + /// without spending an indexer call — nothing due, or no language yet. + async fn grab_season( + &self, + database: &Db, + season: &PendingSeason, + blacklist: &Blacklist, + episode_budget: usize, + ) -> Result, GrabError> { + // §5.2: without the title's original language there is nothing to + // evaluate a release against, and guessing is worse than waiting. + let Some(original_language) = season.original_language.as_deref() else { + tracing::warn!( + series = season.series_title, + "no original language yet; not searching" + ); + return Ok(None); + }; + let original_language = arr_db::policy::language(original_language); + + let episodes = season_episodes(database, season.season_id).await?; + let gaps: Vec<&SeasonEpisode> = episodes + .iter() + .filter(|episode| episode.wanted && !episode.has_file && !episode.in_flight) + .collect(); + let Some(first_gap) = gaps.first() else { + return Ok(None); + }; + let Some(loaded) = database.episode_policy(first_gap.id).await? else { + return Ok(None); + }; + + let now = std::time::SystemTime::now(); + let air_dates: Vec> = episodes + .iter() + .map(|episode| air_date_time(episode.air_date.as_deref())) + .collect(); + let mode = season_grab_mode(&SeasonGrabFacts { + air_dates: &air_dates, + now, + any_episode_on_disk: episodes.iter().any(|episode| episode.has_file), + pack_hard_failed: pack_hard_failed(database, season.season_id).await?, + }); + + let aired_due_gaps: Vec<&SeasonEpisode> = gaps + .iter() + .copied() + .filter(|episode| { + air_date_time(episode.air_date.as_deref()).is_some_and(|date| date <= now) + && backoff_elapsed(episode.search_attempts, episode.last_searched_at.as_deref()) + }) + .collect(); + if aired_due_gaps.is_empty() { + return Ok(None); + } + + match mode { + SeasonGrabMode::SeasonPack => self + .grab_pack_or_fall_back( + database, + season, + &episodes, + &gaps, + &aired_due_gaps, + &loaded, + &original_language, + blacklist, + ) + .await + .map(Some), + SeasonGrabMode::PerEpisode => self + .grab_episodes( + database, + season, + &episodes, + &aired_due_gaps, + &loaded, + &original_language, + blacklist, + episode_budget, + ) + .await + .map(Some), + } + } + + /// One `t=tvsearch` for the whole season; the pack wins when one is + /// eligible, otherwise the same result set feeds per-episode selection — + /// "prefers the pack" is a preference, not a veto on episodes. + #[allow(clippy::too_many_arguments)] + async fn grab_pack_or_fall_back( + &self, + database: &Db, + season: &PendingSeason, + episodes: &[SeasonEpisode], + gaps: &[&SeasonEpisode], + due: &[&SeasonEpisode], + loaded: &TitlePolicy, + original_language: &Language, + blacklist: &Blacklist, + ) -> Result { + let candidates = self + .search( + database, + season, + episodes, + TvSelector::Season { + season: season.season_number_u32(), + }, + loaded, + original_language, + blacklist, + ) + .await?; + + let mut work = SeasonWork::searched(); + let gap_ids: Vec = gaps.iter().map(|episode| episode.id).collect(); + if let Some(winner) = best_candidate( + candidates.iter().filter(|candidate| { + candidate + .claim + .as_ref() + .is_some_and(|claim| is_this_seasons_pack(claim, season.season_number_u32())) + }), + blacklist, + ) { + let outcome = self + .grabber + .send_winner( + database, + &GrabTarget { + scope: GrabScope::Season { + season_id: season.season_id, + episode_ids: gap_ids, + }, + title: &season.series_title, + counts_as_attempt: true, + }, + loaded, + blacklist, + winner.clone(), + ) + .await?; + work.outcomes.extend(outcome); + return Ok(work); + } + + // No eligible pack: fill episodes from the same results, and record + // the attempt on whatever stays open so the season backs off (§6.2). + let mut unfilled = Vec::new(); + for episode in due { + let covering = candidates.iter().filter(|candidate| { + candidate.claim.as_ref().is_some_and(|claim| { + is_single_episode(claim, season.season_number_u32(), episode.number_u32()) + }) + }); + if let Some(winner) = best_candidate(covering, blacklist) { + let outcome = self + .grabber + .send_winner( + database, + &GrabTarget { + scope: GrabScope::Episode { + episode_id: episode.id, + }, + title: &season.series_title, + counts_as_attempt: true, + }, + loaded, + blacklist, + winner.clone(), + ) + .await?; + work.outcomes.extend(outcome); + } else { + unfilled.push(episode.id); + } + } + if !unfilled.is_empty() { + record_episode_search(database, &unfilled).await?; + tracing::info!( + season_id = season.season_id, + series = season.series_title, + episodes = unfilled.len(), + "no eligible release for season or its episodes" + ); + } + Ok(work) + } + + /// One `t=tvsearch` per due episode (§6.2: one call per wanted item). + #[allow(clippy::too_many_arguments)] + async fn grab_episodes( + &self, + database: &Db, + season: &PendingSeason, + episodes: &[SeasonEpisode], + due: &[&SeasonEpisode], + loaded: &TitlePolicy, + original_language: &Language, + blacklist: &Blacklist, + budget: usize, + ) -> Result { + let mut work = SeasonWork::searched(); + for episode in due.iter().take(budget) { + work.episode_searches += 1; + let candidates = self + .search( + database, + season, + episodes, + TvSelector::Episode { + season: season.season_number_u32(), + episode: episode.number_u32(), + }, + loaded, + original_language, + blacklist, + ) + .await?; + let covering = candidates.iter().filter(|candidate| { + candidate.claim.as_ref().is_some_and(|claim| { + is_single_episode(claim, season.season_number_u32(), episode.number_u32()) + }) + }); + if let Some(winner) = best_candidate(covering, blacklist) { + let outcome = self + .grabber + .send_winner( + database, + &GrabTarget { + scope: GrabScope::Episode { + episode_id: episode.id, + }, + title: &season.series_title, + counts_as_attempt: true, + }, + loaded, + blacklist, + winner.clone(), + ) + .await?; + work.outcomes.extend(outcome); + } else { + record_episode_search(database, &[episode.id]).await?; + tracing::info!( + episode_id = episode.id, + series = season.series_title, + "no eligible release for episode" + ); + } + } + Ok(work) + } + + /// Search every indexer with its own widened or narrowed request (§6.1), + /// cache each result with its verdict, and return the candidates with + /// what they claim to cover. + #[allow(clippy::too_many_arguments)] + async fn search( + &self, + database: &Db, + season: &PendingSeason, + episodes: &[SeasonEpisode], + selector: TvSelector, + loaded: &TitlePolicy, + original_language: &Language, + blacklist: &Blacklist, + ) -> Result, GrabError> { + let target = TvTarget { + // The series table carries no TVDB id yet; `tv_request` falls + // back to a text search built from the title and the tag. + tvdb_id: None, + title: season.series_title.clone(), + selector, + }; + let indexers = self.indexers.indexers().await?; + let mut releases: Vec = Vec::new(); + let mut any_searchable = false; + for indexer in &indexers { + let Some(request) = indexer.capabilities.tv_request(&target) else { + continue; + }; + any_searchable = true; + match self.prowlarr.search_indexer(indexer.id, &request).await { + Ok(found) => releases.extend(found), + Err(error) => { + tracing::warn!(indexer_id = indexer.id, %error, "indexer tv search failed"); + } + } + } + if !any_searchable { + tracing::warn!("no indexer accepts a TV or text search; nothing can be grabbed"); + } + + let mut candidates = Vec::new(); + for release in releases { + let claim = arr_parse::parse(&release.name).episode; + let covered: Vec = match &claim { + Some(claim) => episodes + .iter() + .filter(|episode| match claim { + EpisodeClaim::Episodes { .. } => { + claim.covers(season.season_number_u32(), episode.number_u32()) + } + EpisodeClaim::Season { .. } | EpisodeClaim::Seasons { .. } => { + claim.covers_season(season.season_number_u32()) + } + EpisodeClaim::Daily { .. } => false, + }) + .map(|episode| episode.id) + .collect(), + None => Vec::new(), + }; + if covered.is_empty() { + continue; + } + let stored = store_episode_release( + database, + &covered, + &release, + &loaded.policy, + &loaded.overrides, + original_language, + blacklist, + ) + .await?; + candidates.push(TvCandidate { + claim, + eligible: stored, + }); + } + Ok(candidates) + } +} + +impl Action for TvGrabAction { + fn name(&self) -> &'static str { + "tv-grab" + } + + fn run<'a>(&'a self, database: &'a Db) -> ActionFuture<'a> { + Box::pin(async move { self.tick(database).await.map_err(Into::into) }) + } +} + +/// What one season's turn produced. +#[derive(Debug, Default)] +struct SeasonWork { + outcomes: Vec, + episode_searches: usize, +} + +impl SeasonWork { + fn searched() -> Self { + Self::default() + } +} + +/// A stored search result and what its name claims to hold. +#[derive(Debug)] +struct TvCandidate { + claim: Option, + eligible: Option, +} + +/// A season with at least one wanted, missing episode and no live grab. +#[derive(Debug)] +struct PendingSeason { + season_id: i64, + season_number: i64, + series_title: String, + original_language: Option, +} + +impl PendingSeason { + fn season_number_u32(&self) -> u32 { + u32::try_from(self.season_number).unwrap_or_default() + } +} + +/// One episode row with the derived facts selection needs. +#[derive(Debug)] +struct SeasonEpisode { + id: i64, + number: i64, + air_date: Option, + wanted: bool, + search_attempts: i64, + last_searched_at: Option, + has_file: bool, + in_flight: bool, +} + +impl SeasonEpisode { + fn number_u32(&self) -> u32 { + u32::try_from(self.number).unwrap_or_default() + } +} + +/// The gap, straight out of the domain rows (§8). A season already being +/// downloaded as a pack is not a gap, nor is one whose every open episode is +/// individually in flight. +async fn pending_seasons(database: &Db) -> Result, GrabError> { + let rows = sqlx::query!( + r#" + SELECT se.id AS "season_id!: i64", + se.number AS "season_number!: i64", + s.title AS "series_title!: String", + s.original_language + FROM seasons se + JOIN series s ON s.id = se.series_id + WHERE s.blocked = 0 + AND EXISTS ( + SELECT 1 FROM episodes e + WHERE e.season_id = se.id + AND e.wanted = 1 + AND NOT EXISTS ( + SELECT 1 FROM media_files f + WHERE f.owner_kind = 'episode' AND f.owner_id = e.id + ) + AND NOT EXISTS ( + SELECT 1 FROM grabs g + WHERE g.target_kind = 'episode' AND g.target_id = e.id + AND g.state IN ('sent', 'downloaded', 'imported') + ) + ) + AND NOT EXISTS ( + SELECT 1 FROM grabs g + WHERE g.target_kind = 'season' AND g.target_id = se.id + AND g.state IN ('sent', 'downloaded', 'imported') + ) + ORDER BY se.id + "# + ) + .fetch_all(database.pool()) + .await?; + + Ok(rows + .into_iter() + .map(|row| PendingSeason { + season_id: row.season_id, + season_number: row.season_number, + series_title: row.series_title, + original_language: row.original_language, + }) + .collect()) +} + +async fn season_episodes(database: &Db, season_id: i64) -> Result, GrabError> { + let rows = sqlx::query!( + r#" + SELECT e.id AS "id!: i64", + e.number AS "number!: i64", + e.air_date, + e.wanted AS "wanted!: bool", + e.search_attempts AS "search_attempts!: i64", + e.last_searched_at, + EXISTS ( + SELECT 1 FROM media_files f + WHERE f.owner_kind = 'episode' AND f.owner_id = e.id + ) AS "has_file!: bool", + EXISTS ( + SELECT 1 FROM grabs g + WHERE g.target_kind = 'episode' AND g.target_id = e.id + AND g.state IN ('sent', 'downloaded', 'imported') + ) AS "in_flight!: bool" + FROM episodes e + WHERE e.season_id = ? + ORDER BY e.number + "#, + season_id + ) + .fetch_all(database.pool()) + .await?; + + Ok(rows + .into_iter() + .map(|row| SeasonEpisode { + id: row.id, + number: row.number, + air_date: row.air_date, + wanted: row.wanted, + search_attempts: row.search_attempts, + last_searched_at: row.last_searched_at, + has_file: row.has_file, + in_flight: row.in_flight, + }) + .collect()) +} + +/// Whether a season-pack grab for this season already hard-failed — the +/// fall-back-to-per-episode signal. +async fn pack_hard_failed(database: &Db, season_id: i64) -> Result { + let failed = sqlx::query_scalar!( + r#"SELECT EXISTS ( + SELECT 1 FROM grabs + WHERE target_kind = 'season' AND target_id = ? AND state = 'failed' + ) AS "failed!: bool""#, + season_id + ) + .fetch_one(database.pool()) + .await?; + Ok(failed) +} + +/// A pack for exactly this season. Multi-season packs are never selected +/// automatically: they drag other seasons in with them. +fn is_this_seasons_pack(claim: &EpisodeClaim, season: u32) -> bool { + match claim { + EpisodeClaim::Season { season: claimed } => *claimed == season, + EpisodeClaim::Seasons { first, last } => first == last && *first == season, + EpisodeClaim::Episodes { .. } | EpisodeClaim::Daily { .. } => false, + } +} + +/// A release holding exactly this one episode. Multi-episode files are left +/// to manual grabs: two episodes automatically picking the same torrent +/// would fight over one `grabs` row. +fn is_single_episode(claim: &EpisodeClaim, season: u32, episode: u32) -> bool { + matches!( + claim, + EpisodeClaim::Episodes { episodes, .. } if episodes.len() == 1 + ) && claim.covers(season, episode) +} + +/// The eligible view of a candidate set, in the same total order targeted +/// movie search uses, minus anything the blacklist knows (§6.3). +fn best_candidate<'a>( + candidates: impl Iterator, + blacklist: &Blacklist, +) -> Option<&'a Eligible> { + let mut eligible: Vec<&Eligible> = candidates + .filter_map(|candidate| candidate.eligible.as_ref()) + .collect(); + eligible.sort_by(|left, right| { + right + .score + .cmp(&left.score) + .then_with(|| left.indexer_id.cmp(&right.indexer_id)) + .then_with(|| left.guid.cmp(&right.guid)) + }); + eligible + .into_iter() + .find(|candidate| !blacklist.blocks_candidate(&candidate.name, &candidate.download_url)) +} + +/// An `air_date` as TMDB writes it (`YYYY-MM-DD`), or a full timestamp if +/// one ever arrives that way. Unknown is unaired: it cannot complete a +/// season and cannot be searched for. +fn air_date_time(value: Option<&str>) -> Option { + let value = value?; + let timestamp = if let Ok(date) = value.parse::() { + date.and_time(chrono::NaiveTime::MIN).and_utc().timestamp() + } else { + value + .parse::>() + .ok()? + .timestamp() + }; + let seconds = u64::try_from(timestamp.abs()).ok()?; + if timestamp < 0 { + std::time::UNIX_EPOCH.checked_sub(std::time::Duration::from_secs(seconds)) + } else { + std::time::UNIX_EPOCH.checked_add(std::time::Duration::from_secs(seconds)) + } +} + +#[cfg(test)] +#[allow(clippy::unwrap_used)] +mod tests { + use std::collections::HashMap; + use std::sync::{Arc, Mutex}; + + use serde_json::{json, Value}; + use wiremock::matchers::{method, path, query_param}; + use wiremock::{Mock, MockServer, Request, Respond, ResponseTemplate}; + + use crate::grab::SeedingLimits; + + use super::*; + + /// A Transmission that dedupes on the infohash, like the real one. + #[derive(Clone, Debug, Default)] + struct FakeTransmission { + torrents: Arc>>, + } + + #[derive(Clone, Debug)] + struct FakeTorrent { + id: i64, + hash: String, + source: String, + progress: f64, + } + + impl FakeTransmission { + fn torrents(&self) -> Vec { + self.torrents.lock().unwrap().clone() + } + + fn add(&self, arguments: &Value) -> ResponseTemplate { + let source = arguments["filename"] + .as_str() + .unwrap_or_default() + .to_owned(); + let mut torrents = self.torrents.lock().unwrap(); + if let Some(existing) = torrents.iter().find(|torrent| torrent.source == source) { + return success(&json!({"torrent-duplicate": { + "id": existing.id, "name": existing.source, "hashString": existing.hash + }})); + } + let id = i64::try_from(torrents.len()).unwrap() + 1; + let hash = format!("{:040x}", id * 7); + torrents.push(FakeTorrent { + id, + hash: hash.clone(), + source: source.clone(), + progress: 0.0, + }); + success(&json!({"torrent-added": {"id": id, "name": source, "hashString": hash}})) + } + } + + impl Respond for FakeTransmission { + fn respond(&self, request: &Request) -> ResponseTemplate { + let body: Value = serde_json::from_slice(&request.body).unwrap(); + match body["method"].as_str().unwrap_or_default() { + "torrent-add" => self.add(&body["arguments"]), + "torrent-get" => { + let torrents: Vec = self + .torrents() + .into_iter() + .map(|torrent| { + json!({ + "id": torrent.id, "name": torrent.source, + "hashString": torrent.hash, "status": 4, + "percentDone": torrent.progress, + "downloadDir": "/downloads", "labels": [] + }) + }) + .collect(); + success(&json!({"torrents": torrents})) + } + _ => success(&json!({})), + } + } + } + + fn success(arguments: &Value) -> ResponseTemplate { + ResponseTemplate::new(200).set_body_json(json!({ + "result": "success", "arguments": arguments + })) + } + + /// A season pack and its three episodes, all eligible under the seeded + /// TV main policy (§5.5 bands: 2160p floor 8 GiB). + const TV_RSS: &str = r#" + + Fallout.S01.2160p.WEB-DL.DDP5.1.Atmos + packhttps://tracker/pack.torrent + 85899345920 + + + + Fallout.S01E01.2160p.WEB-DL.DDP5.1 + e01https://tracker/e01.torrent + 10737418240 + + + + Fallout.S01E02.2160p.WEB-DL.DDP5.1 + e02https://tracker/e02.torrent + 10737418240 + + + + Fallout.S01E03.2160p.WEB-DL.DDP5.1 + e03https://tracker/e03.torrent + 10737418240 + + + "#; + + async fn prowlarr() -> MockServer { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/api/v1/indexer")) + .respond_with( + ResponseTemplate::new(200) + .set_body_json(json!([{"id": 7, "name": "tracker", "enable": true}])), + ) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path("/7/api")) + .and(query_param("t", "caps")) + .respond_with(ResponseTemplate::new(200).set_body_string( + r#""#, + )) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path("/7/api")) + .and(query_param("t", "search")) + .respond_with(ResponseTemplate::new(200).set_body_string(TV_RSS)) + .mount(&server) + .await; + server + } + + async fn transmission() -> (MockServer, FakeTransmission) { + let server = MockServer::start().await; + let fake = FakeTransmission::default(); + Mock::given(method("POST")) + .respond_with(fake.clone()) + .mount(&server) + .await; + (server, fake) + } + + /// Fallout S01 with three wanted episodes at the given air dates. + async fn wanted_season(air_dates: &[&str]) -> (tempfile::TempDir, Db, i64) { + let dir = tempfile::tempdir().unwrap(); + let database = Db::connect(dir.path().join("arr.db")).await.unwrap(); + database.migrate().await.unwrap(); + sqlx::query( + "INSERT INTO series (tmdb_id, title, year, original_language, root_id) + SELECT 106379, 'Fallout', 2024, 'en', id + FROM roots WHERE kind = 'tv' AND audience = 'main'", + ) + .execute(database.pool()) + .await + .unwrap(); + let season_id: i64 = sqlx::query_scalar( + "INSERT INTO seasons (series_id, number) VALUES (1, 1) RETURNING id", + ) + .fetch_one(database.pool()) + .await + .unwrap(); + for (index, air_date) in air_dates.iter().enumerate() { + sqlx::query( + "INSERT INTO episodes (season_id, number, title, air_date, wanted) + VALUES (?, ?, ?, ?, 1)", + ) + .bind(season_id) + .bind(i64::try_from(index).unwrap() + 1) + .bind(format!("Episode {}", index + 1)) + .bind(air_date) + .execute(database.pool()) + .await + .unwrap(); + } + (dir, database, season_id) + } + + fn action(prowlarr: &MockServer, transmission: &MockServer) -> TvGrabAction { + TvGrabAction::new( + ProwlarrClient::new(prowlarr.uri(), "key").unwrap(), + TransmissionClient::new(&transmission.uri()).unwrap(), + PathBuf::from("/mnt/media/transmission/complete"), + SeedingRules::new( + SeedingLimits { + ratio: 1.5, + idle_minutes: 60, + }, + HashMap::new(), + ), + ) + } + + async fn grabs(database: &Db) -> Vec<(String, i64, String)> { + sqlx::query_as::<_, (String, i64, String)>( + "SELECT target_kind, target_id, state FROM grabs ORDER BY id", + ) + .fetch_all(database.pool()) + .await + .unwrap() + } + + async fn episode_states(database: &Db) -> Vec { + sqlx::query_scalar::<_, String>("SELECT state FROM episodes ORDER BY number") + .fetch_all(database.pool()) + .await + .unwrap() + } + + /// The first acceptance case: a completed season with nothing on disk + /// takes the pack — one torrent, one season grab, every episode + /// downloading. + #[tokio::test] + async fn a_completed_season_takes_one_season_pack() { + let (_dir, database, season_id) = + wanted_season(&["2024-04-11", "2024-04-11", "2024-04-11"]).await; + let indexer = prowlarr().await; + let (downloader, fake) = transmission().await; + + let outcomes = action(&indexer, &downloader).tick(&database).await.unwrap(); + + assert_eq!(outcomes.len(), 1); + assert_eq!(fake.torrents().len(), 1); + assert!(fake.torrents()[0].source.ends_with("pack.torrent")); + assert_eq!( + grabs(&database).await, + vec![("season".to_owned(), season_id, "sent".to_owned())] + ); + assert_eq!( + episode_states(&database).await, + vec!["downloading", "downloading", "downloading"] + ); + } + + /// The second acceptance case: an airing season grabs episodes + /// individually as they appear — never the pack, never an unaired + /// episode. + #[tokio::test] + async fn an_airing_season_grabs_aired_episodes_individually() { + let (_dir, database, _season_id) = + wanted_season(&["2024-04-11", "2024-04-18", "2999-01-01"]).await; + let indexer = prowlarr().await; + let (downloader, fake) = transmission().await; + + action(&indexer, &downloader).tick(&database).await.unwrap(); + + let sources: Vec = fake + .torrents() + .into_iter() + .map(|torrent| torrent.source) + .collect(); + assert_eq!(sources.len(), 2, "{sources:?}"); + assert!(sources[0].ends_with("e01.torrent")); + assert!(sources[1].ends_with("e02.torrent")); + let grabs = grabs(&database).await; + assert_eq!(grabs.len(), 2); + assert!(grabs + .iter() + .all(|(kind, _, state)| kind == "episode" && state == "sent")); + assert_eq!( + episode_states(&database).await, + vec!["downloading", "downloading", "missing"] + ); + } + + /// The third acceptance case, selection side: a hard-failed pack put the + /// release on the blacklist and the season falls back to per-episode — + /// the pack is not tried again and the episodes are not written off. + #[tokio::test] + async fn a_hard_failed_pack_falls_back_to_per_episode() { + let (_dir, database, season_id) = + wanted_season(&["2024-04-11", "2024-04-11", "2024-04-11"]).await; + // What the import tick left behind: a failed season grab and the + // pack release on the blacklist (§6.3). + let release_id: i64 = sqlx::query_scalar( + "INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed, verdict) + VALUES (7, 'pack', 'Fallout.S01.2160p.WEB-DL.DDP5.1.Atmos', 85899345920, + 'https://tracker/pack.torrent', '{}', 'eligible') + RETURNING id", + ) + .fetch_one(database.pool()) + .await + .unwrap(); + sqlx::query( + "INSERT INTO grabs (release_id, target_kind, target_id, infohash, state) + VALUES (?, 'season', ?, 'dead', 'failed')", + ) + .bind(release_id) + .bind(season_id) + .execute(database.pool()) + .await + .unwrap(); + arr_db::blacklist::add( + database.pool(), + Some("dead"), + "Fallout.S01.2160p.WEB-DL.DDP5.1.Atmos", + "dolby_vision_profile", + ) + .await + .unwrap(); + let indexer = prowlarr().await; + let (downloader, fake) = transmission().await; + + action(&indexer, &downloader).tick(&database).await.unwrap(); + + let sources: Vec = fake + .torrents() + .into_iter() + .map(|torrent| torrent.source) + .collect(); + assert_eq!(sources.len(), 3, "{sources:?}"); + assert!(sources + .iter() + .all(|source| !source.ends_with("pack.torrent"))); + let grabs = grabs(&database).await; + assert!(grabs + .iter() + .filter(|(_, _, state)| state == "sent") + .all(|(kind, _, _)| kind == "episode")); + assert_eq!( + episode_states(&database).await, + vec!["downloading", "downloading", "downloading"] + ); + } + + /// §14: a season with an episode already on disk never grabs the pack, + /// even when fully released. The remaining episodes come individually. + #[tokio::test] + async fn a_partially_present_season_grabs_the_rest_per_episode() { + let (_dir, database, _season_id) = + wanted_season(&["2024-04-11", "2024-04-11", "2024-04-11"]).await; + sqlx::query( + "INSERT INTO media_files (owner_kind, owner_id, path, size) + SELECT 'episode', id, '/library/e01.mkv', 1 FROM episodes WHERE number = 1", + ) + .execute(database.pool()) + .await + .unwrap(); + sqlx::query("UPDATE episodes SET state = 'available' WHERE number = 1") + .execute(database.pool()) + .await + .unwrap(); + let indexer = prowlarr().await; + let (downloader, fake) = transmission().await; + + action(&indexer, &downloader).tick(&database).await.unwrap(); + + let sources: Vec = fake + .torrents() + .into_iter() + .map(|torrent| torrent.source) + .collect(); + assert_eq!(sources.len(), 2, "{sources:?}"); + assert!(sources[0].ends_with("e02.torrent")); + assert!(sources[1].ends_with("e03.torrent")); + } + + /// An empty result backs the whole season off together (§6.2) instead of + /// hammering the tracker every 30 s. + #[tokio::test] + async fn a_fruitless_season_search_records_the_attempt() { + let (_dir, database, _season_id) = wanted_season(&["2024-04-11", "2024-04-11"]).await; + let indexer = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/api/v1/indexer")) + .respond_with( + ResponseTemplate::new(200) + .set_body_json(json!([{"id": 7, "name": "tracker", "enable": true}])), + ) + .mount(&indexer) + .await; + Mock::given(method("GET")) + .and(path("/7/api")) + .and(query_param("t", "caps")) + .respond_with(ResponseTemplate::new(200).set_body_string( + r#""#, + )) + .mount(&indexer) + .await; + Mock::given(method("GET")) + .and(path("/7/api")) + .and(query_param("t", "search")) + .respond_with( + ResponseTemplate::new(200).set_body_string(""), + ) + .mount(&indexer) + .await; + let (downloader, fake) = transmission().await; + let action = action(&indexer, &downloader); + + action.tick(&database).await.unwrap(); + assert!(fake.torrents().is_empty()); + let attempts: Vec = + sqlx::query_scalar("SELECT search_attempts FROM episodes ORDER BY number") + .fetch_all(database.pool()) + .await + .unwrap(); + assert_eq!(attempts, vec![1, 1]); + + // Inside the backoff window nothing is searched again. + let searches_after_first: usize = indexer + .received_requests() + .await + .unwrap() + .iter() + .filter(|request| request.url.path() == "/7/api") + .count(); + action.tick(&database).await.unwrap(); + let searches_after_second: usize = indexer + .received_requests() + .await + .unwrap() + .iter() + .filter(|request| request.url.path() == "/7/api") + .count(); + assert_eq!(searches_after_first, searches_after_second); + } +} diff --git a/crates/arr-db/migrations/0010_season_grabs.sql b/crates/arr-db/migrations/0010_season_grabs.sql new file mode 100644 index 0000000..8807326 --- /dev/null +++ b/crates/arr-db/migrations/0010_season_grabs.sql @@ -0,0 +1,23 @@ +-- A grab may target a whole season: one season-pack torrent satisfies every +-- episode in it (DESIGN.md §13 phase 6). SQLite cannot alter a CHECK, so the +-- table is rebuilt. +CREATE TABLE grabs_new ( + id INTEGER PRIMARY KEY, + release_id INTEGER NOT NULL REFERENCES releases (id), + target_kind TEXT NOT NULL CHECK (target_kind IN ('movie', 'episode', 'season')), + target_id INTEGER NOT NULL, + infohash TEXT NOT NULL UNIQUE, + state TEXT NOT NULL DEFAULT 'sent' + CHECK (state IN ('sent', 'downloaded', 'imported', 'failed')), + grabbed_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), + imported_at TEXT +) STRICT; + +INSERT INTO grabs_new (id, release_id, target_kind, target_id, infohash, state, grabbed_at, imported_at) +SELECT id, release_id, target_kind, target_id, infohash, state, grabbed_at, imported_at FROM grabs; + +DROP TABLE grabs; +ALTER TABLE grabs_new RENAME TO grabs; + +CREATE INDEX grabs_state ON grabs (state); +CREATE INDEX grabs_target ON grabs (target_kind, target_id);