From d8797e1996106bc33ecab06e8e4cb0f0992c2825 Mon Sep 17 00:00:00 2001 From: Miguel Palhas Date: Mon, 24 Aug 2026 18:12:10 +0100 Subject: [PATCH] feat(daemon): refresh metadata on add MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The metadata lane runs daily, so a series added a moment ago showed no seasons for up to 24 hours and a movie had no digital release date — the field §6.2 gates targeted search on. AppState now carries a MetadataCommand channel alongside the movie, episode and season ones. Both create handlers send on it after the row is committed, and a new daemon lane drains it. Its own task rather than an arm of manual::run: a refresh against TMDB can take a while and must not sit in front of an operator's manual search. The add never waits on TMDB and never fails because of it. A refresh that fails leaves metadata_refreshed_at NULL, which is what the daily sweep already treats as due, so the title is retried rather than lost. A command naming a title deleted in between finds no row and does nothing. METADATA_INTERVAL is unchanged. Co-Authored-By: Claude Opus 5 --- ...35d45e9d1b2c639b71102e241f50982502eb.json} | 4 +- ...eadce81719e044c8fb65ba404b45b70bf8d8c.json | 12 + ...7c74f232ec6543787f64efeb53bffa8170113.json | 12 - ...337f18909a58d0f863c794f93f8285b9133c8.json | 26 ++ ...55064d5a83eafe5bc95ddda9a9056d5724f10.json | 92 +++++ crates/arr-api/src/lib.rs | 3 +- crates/arr-api/src/movies.rs | 59 ++- crates/arr-api/src/series.rs | 69 +++- crates/arr-api/src/state.rs | 41 ++ crates/arr-daemon/src/grab.rs | 157 ++++---- crates/arr-daemon/src/main.rs | 27 +- crates/arr-daemon/src/metadata.rs | 367 ++++++++++++++++++ crates/arr-daemon/src/series_refresh.rs | 34 +- 13 files changed, 803 insertions(+), 100 deletions(-) rename .sqlx/{query-de2153eefd6ec03ceeaf640d599271834cb06c94eda7a9956c5bd7399593c0ae.json => query-0516e88e07d3708ba4ccc52d7ba635d45e9d1b2c639b71102e241f50982502eb.json} (58%) create mode 100644 .sqlx/query-0f0966881a69a9f4009df7b27ebeadce81719e044c8fb65ba404b45b70bf8d8c.json delete mode 100644 .sqlx/query-462039959030ee881bcce2daf9d7c74f232ec6543787f64efeb53bffa8170113.json create mode 100644 .sqlx/query-a1be0fa3691226439f768d11600337f18909a58d0f863c794f93f8285b9133c8.json create mode 100644 .sqlx/query-e57dd914010e4346ee3b5cc64fe55064d5a83eafe5bc95ddda9a9056d5724f10.json create mode 100644 crates/arr-daemon/src/metadata.rs diff --git a/.sqlx/query-de2153eefd6ec03ceeaf640d599271834cb06c94eda7a9956c5bd7399593c0ae.json b/.sqlx/query-0516e88e07d3708ba4ccc52d7ba635d45e9d1b2c639b71102e241f50982502eb.json similarity index 58% rename from .sqlx/query-de2153eefd6ec03ceeaf640d599271834cb06c94eda7a9956c5bd7399593c0ae.json rename to .sqlx/query-0516e88e07d3708ba4ccc52d7ba635d45e9d1b2c639b71102e241f50982502eb.json index b1c5124..3ef8d12 100644 --- a/.sqlx/query-de2153eefd6ec03ceeaf640d599271834cb06c94eda7a9956c5bd7399593c0ae.json +++ b/.sqlx/query-0516e88e07d3708ba4ccc52d7ba635d45e9d1b2c639b71102e241f50982502eb.json @@ -1,6 +1,6 @@ { "db_name": "SQLite", - "query": "UPDATE movies SET metadata_refreshed_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')\n WHERE id = ?", + "query": "UPDATE movies SET metadata_refreshed_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')\n WHERE id = ?", "describe": { "columns": [], "parameters": { @@ -8,5 +8,5 @@ }, "nullable": [] }, - "hash": "de2153eefd6ec03ceeaf640d599271834cb06c94eda7a9956c5bd7399593c0ae" + "hash": "0516e88e07d3708ba4ccc52d7ba635d45e9d1b2c639b71102e241f50982502eb" } diff --git a/.sqlx/query-0f0966881a69a9f4009df7b27ebeadce81719e044c8fb65ba404b45b70bf8d8c.json b/.sqlx/query-0f0966881a69a9f4009df7b27ebeadce81719e044c8fb65ba404b45b70bf8d8c.json new file mode 100644 index 0000000..ef6824d --- /dev/null +++ b/.sqlx/query-0f0966881a69a9f4009df7b27ebeadce81719e044c8fb65ba404b45b70bf8d8c.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "UPDATE movies\n SET title = ?, year = ?, original_language = ?, digital_release = ?,\n imdb_id = ?, poster_path = ?, backdrop_path = ?, vote_average = ?,\n metadata_refreshed_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now'),\n search_attempts = 0, last_searched_at = NULL,\n updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')\n WHERE id = ? AND (\n title IS NOT ? OR year IS NOT ? OR original_language IS NOT ?\n OR digital_release IS NOT ? OR imdb_id IS NOT ?\n OR poster_path IS NOT ? OR backdrop_path IS NOT ?\n OR vote_average IS NOT ?\n )", + "describe": { + "columns": [], + "parameters": { + "Right": 17 + }, + "nullable": [] + }, + "hash": "0f0966881a69a9f4009df7b27ebeadce81719e044c8fb65ba404b45b70bf8d8c" +} diff --git a/.sqlx/query-462039959030ee881bcce2daf9d7c74f232ec6543787f64efeb53bffa8170113.json b/.sqlx/query-462039959030ee881bcce2daf9d7c74f232ec6543787f64efeb53bffa8170113.json deleted file mode 100644 index c9c1698..0000000 --- a/.sqlx/query-462039959030ee881bcce2daf9d7c74f232ec6543787f64efeb53bffa8170113.json +++ /dev/null @@ -1,12 +0,0 @@ -{ - "db_name": "SQLite", - "query": "UPDATE movies\n SET title = ?, year = ?, original_language = ?, digital_release = ?,\n imdb_id = ?, poster_path = ?, backdrop_path = ?, vote_average = ?,\n metadata_refreshed_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now'),\n search_attempts = 0, last_searched_at = NULL,\n updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')\n WHERE id = ? AND (\n title IS NOT ? OR year IS NOT ? OR original_language IS NOT ?\n OR digital_release IS NOT ? OR imdb_id IS NOT ?\n OR poster_path IS NOT ? OR backdrop_path IS NOT ?\n OR vote_average IS NOT ?\n )", - "describe": { - "columns": [], - "parameters": { - "Right": 17 - }, - "nullable": [] - }, - "hash": "462039959030ee881bcce2daf9d7c74f232ec6543787f64efeb53bffa8170113" -} diff --git a/.sqlx/query-a1be0fa3691226439f768d11600337f18909a58d0f863c794f93f8285b9133c8.json b/.sqlx/query-a1be0fa3691226439f768d11600337f18909a58d0f863c794f93f8285b9133c8.json new file mode 100644 index 0000000..4720e1d --- /dev/null +++ b/.sqlx/query-a1be0fa3691226439f768d11600337f18909a58d0f863c794f93f8285b9133c8.json @@ -0,0 +1,26 @@ +{ + "db_name": "SQLite", + "query": "SELECT tmdb_id AS \"tmdb_id!: i64\", metadata_refreshed_at\n FROM movies WHERE id = ?", + "describe": { + "columns": [ + { + "name": "tmdb_id!: i64", + "ordinal": 0, + "type_info": "Integer" + }, + { + "name": "metadata_refreshed_at", + "ordinal": 1, + "type_info": "Text" + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + false, + true + ] + }, + "hash": "a1be0fa3691226439f768d11600337f18909a58d0f863c794f93f8285b9133c8" +} diff --git a/.sqlx/query-e57dd914010e4346ee3b5cc64fe55064d5a83eafe5bc95ddda9a9056d5724f10.json b/.sqlx/query-e57dd914010e4346ee3b5cc64fe55064d5a83eafe5bc95ddda9a9056d5724f10.json new file mode 100644 index 0000000..42f6e34 --- /dev/null +++ b/.sqlx/query-e57dd914010e4346ee3b5cc64fe55064d5a83eafe5bc95ddda9a9056d5724f10.json @@ -0,0 +1,92 @@ +{ + "db_name": "SQLite", + "query": "SELECT id AS \"id!: i64\", tmdb_id AS \"tmdb_id!: i64\", tvdb_id,\n title AS \"title!: String\", year, original_language,\n root_id AS \"root_id!: i64\", auto_track AS \"auto_track!: bool\",\n upstream_ended AS \"upstream_ended!: bool\", metadata_refreshed_at,\n poster_path, backdrop_path, vote_average\n FROM series WHERE id = ?", + "describe": { + "columns": [ + { + "name": "id!: i64", + "ordinal": 0, + "type_info": "Integer" + }, + { + "name": "tmdb_id!: i64", + "ordinal": 1, + "type_info": "Integer" + }, + { + "name": "tvdb_id", + "ordinal": 2, + "type_info": "Integer" + }, + { + "name": "title!: String", + "ordinal": 3, + "type_info": "Text" + }, + { + "name": "year", + "ordinal": 4, + "type_info": "Integer" + }, + { + "name": "original_language", + "ordinal": 5, + "type_info": "Text" + }, + { + "name": "root_id!: i64", + "ordinal": 6, + "type_info": "Integer" + }, + { + "name": "auto_track!: bool", + "ordinal": 7, + "type_info": "Integer" + }, + { + "name": "upstream_ended!: bool", + "ordinal": 8, + "type_info": "Integer" + }, + { + "name": "metadata_refreshed_at", + "ordinal": 9, + "type_info": "Text" + }, + { + "name": "poster_path", + "ordinal": 10, + "type_info": "Text" + }, + { + "name": "backdrop_path", + "ordinal": 11, + "type_info": "Text" + }, + { + "name": "vote_average", + "ordinal": 12, + "type_info": "Float" + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + false, + false, + true, + false, + true, + true, + false, + false, + false, + true, + true, + true, + true + ] + }, + "hash": "e57dd914010e4346ee3b5cc64fe55064d5a83eafe5bc95ddda9a9056d5724f10" +} diff --git a/crates/arr-api/src/lib.rs b/crates/arr-api/src/lib.rs index 2f887c1..1770cbf 100644 --- a/crates/arr-api/src/lib.rs +++ b/crates/arr-api/src/lib.rs @@ -40,7 +40,8 @@ pub use series::{ UpdateSeason, UpdateSeries, }; pub use state::{ - AppState, EpisodeCommand, MovieCommand, SeasonCommand, Upstreams, DEFAULT_TMDB_URL, + AppState, EpisodeCommand, MetadataCommand, MovieCommand, SeasonCommand, Upstreams, + DEFAULT_TMDB_URL, }; pub use trailer::{Trailer, TrailerKind}; diff --git a/crates/arr-api/src/movies.rs b/crates/arr-api/src/movies.rs index b315b78..ebb7361 100644 --- a/crates/arr-api/src/movies.rs +++ b/crates/arr-api/src/movies.rs @@ -9,7 +9,7 @@ use serde::{Deserialize, Serialize}; use utoipa::{IntoParams, ToSchema}; use crate::owners::Owner; -use crate::state::{AppState, MovieCommand}; +use crate::state::{AppState, MetadataCommand, MovieCommand}; #[derive(Debug, Clone, Serialize, ToSchema)] pub struct Movie { @@ -330,9 +330,18 @@ pub async fn create( ) .execute(pool(&state)?) .await?; + let movie_id = result.last_insert_rowid(); + // Issue #176: the digital release date §6.2 gates targeted search on + // comes from a refresh, and the metadata lane is daily. Ask it to run + // now. Asynchronous by design — the add is already committed and must + // not wait on, or fail because of, TMDB. If nothing is draining, the + // daily sweep still picks the movie up: `metadata_refreshed_at` is NULL. + if let Err(error) = state.send_metadata_command(MetadataCommand::Movie { movie_id }) { + tracing::warn!(movie_id, %error, "metadata refresh not queued for the new movie"); + } Ok(( StatusCode::CREATED, - Json(load_movie(&state, result.last_insert_rowid()).await?), + Json(load_movie(&state, movie_id).await?), )) } @@ -1278,6 +1287,52 @@ mod tests { assert_eq!(title_target(root, root), None); } + /// Issue #176: §6.2 gates targeted search on the digital release date, + /// and that only ever arrives from a metadata refresh. Adding a movie + /// queues one rather than waiting for the daily lane. TMDB here is a + /// closed port: the add still returns 201 and the command is still + /// queued. + #[tokio::test] + async fn adding_a_movie_queues_a_refresh_even_with_tmdb_down() { + let dir = tempfile::tempdir().expect("tempdir"); + let database = arr_db::Db::connect(dir.path().join("arr.db")) + .await + .expect("connect database"); + database.migrate().await.expect("migrate database"); + let state = AppState::new( + Upstreams::new("http://127.0.0.1:1".into(), "http://127.0.0.1:1".into()) + .with_tmdb_url("http://127.0.0.1:1".into()) + .with_tmdb_api_key(Some("key".into())), + ) + .expect("state") + .with_database(database); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let address = listener.local_addr().expect("address"); + let served = state.clone(); + tokio::spawn(async move { axum::serve(listener, router(served)).await.expect("serve") }); + let base = format!("http://{address}"); + + let movie = add_movie(&base, 693_134, 1).await; + let movie_id = movie["id"].as_i64().expect("id"); + + assert_eq!( + state.next_metadata_command().await.expect("command"), + MetadataCommand::Movie { movie_id } + ); + let refreshed_at: Option = + sqlx::query_scalar("SELECT metadata_refreshed_at FROM movies WHERE id = ?") + .bind(movie_id) + .fetch_one(state.database().expect("database").pool()) + .await + .expect("movie row"); + assert!( + refreshed_at.is_none(), + "an unrefreshed movie stays due for the scheduled sweep" + ); + } + #[tokio::test] async fn release_actions_are_scoped_to_the_movie() { let (_dir, state, base) = application().await; diff --git a/crates/arr-api/src/series.rs b/crates/arr-api/src/series.rs index 2994d7f..a483891 100644 --- a/crates/arr-api/src/series.rs +++ b/crates/arr-api/src/series.rs @@ -32,7 +32,7 @@ use utoipa::{IntoParams, ToSchema}; use crate::movies::{pool, rescore, Accepted, ApiError, ErrorBody, Release}; use crate::owners::Owner; use crate::search::tmdb_client; -use crate::state::{AppState, EpisodeCommand, SeasonCommand}; +use crate::state::{AppState, EpisodeCommand, MetadataCommand, SeasonCommand}; /// A series with the status derived from its episodes (§4.2). #[derive(Debug, Clone, Serialize, ToSchema)] @@ -471,9 +471,18 @@ pub async fn create( ) .execute(pool(&state)?) .await?; + let series_id = result.last_insert_rowid(); + // Issue #176: a new series has no seasons until a refresh reveals them, + // and the metadata lane is daily. Ask it to run now. Asynchronous by + // design — the add is already committed and must not wait on, or fail + // because of, TMDB. If nothing is draining, the daily sweep still picks + // the series up: `metadata_refreshed_at` is NULL. + if let Err(error) = state.send_metadata_command(MetadataCommand::Series { series_id }) { + tracing::warn!(series_id, %error, "metadata refresh not queued for the new series"); + } Ok(( StatusCode::CREATED, - Json(load_series(&state, result.last_insert_rowid()).await?), + Json(load_series(&state, series_id).await?), )) } @@ -1740,6 +1749,62 @@ mod tests { assert_eq!(vote, Some(8.417)); } + /// Issue #176: the daily metadata lane is what reveals a new series' + /// seasons, so adding one queues a refresh instead of leaving the page + /// empty for up to a day. TMDB here is a closed port: the add still + /// returns 201 and the command is still queued, because the refresh is + /// the lane's problem and not the request's. + #[tokio::test] + async fn adding_a_series_queues_a_refresh_even_with_tmdb_down() { + let dir = tempfile::tempdir().expect("tempdir"); + let database = arr_db::Db::connect(dir.path().join("arr.db")) + .await + .expect("connect database"); + database.migrate().await.expect("migrate database"); + let state = AppState::new( + Upstreams::new("http://127.0.0.1:1".into(), "http://127.0.0.1:1".into()) + .with_tmdb_url("http://127.0.0.1:1".into()) + .with_tmdb_api_key(Some("key".into())), + ) + .expect("state") + .with_database(database); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let address = listener.local_addr().expect("address"); + let served = state.clone(); + tokio::spawn(async move { axum::serve(listener, router(served)).await.expect("serve") }); + let base = format!("http://{address}"); + + let response = reqwest::Client::new() + .post(format!("{base}/api/series")) + .json(&serde_json::json!({ + "tmdb_id": 82_728, "title": "Bluey", "root_id": 3 + })) + .send() + .await + .expect("create series"); + assert_eq!(response.status(), StatusCode::CREATED); + let series_id = response.json::().await.expect("json")["id"] + .as_i64() + .expect("series id"); + + assert_eq!( + state.next_metadata_command().await.expect("command"), + MetadataCommand::Series { series_id } + ); + let refreshed_at: Option = + sqlx::query_scalar("SELECT metadata_refreshed_at FROM series WHERE id = ?") + .bind(series_id) + .fetch_one(state.database().expect("database").pool()) + .await + .expect("series row"); + assert!( + refreshed_at.is_none(), + "an unrefreshed series stays due for the scheduled sweep" + ); + } + #[tokio::test] async fn series_must_sit_on_a_tv_root() { let (_dir, state, base) = application().await; diff --git a/crates/arr-api/src/state.rs b/crates/arr-api/src/state.rs index 61e3a7b..de8c4c0 100644 --- a/crates/arr-api/src/state.rs +++ b/crates/arr-api/src/state.rs @@ -74,6 +74,8 @@ pub struct AppState { pending_episode_commands: Arc>>, season_commands: mpsc::Sender, pending_season_commands: Arc>>, + metadata_commands: mpsc::Sender, + pending_metadata_commands: Arc>>, } /// Work explicitly requested through the movie API. @@ -104,6 +106,22 @@ pub enum SeasonCommand { Grab { season_id: i64, release_id: i64 }, } +/// A title whose TMDB metadata should be refreshed now rather than on the +/// daily lane's next tick (issue #176). +/// +/// Sent when a title is added: a new series has no seasons at all until a +/// refresh reveals them, and a new movie has no digital release date, so +/// waiting up to a day is the difference between a usable page and an empty +/// one. Unlike the other three commands this is not an operator action, so a +/// full channel is dropped rather than reported — the scheduled sweep still +/// owns the title, because a title that was never refreshed keeps +/// `metadata_refreshed_at` NULL. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MetadataCommand { + Series { series_id: i64 }, + Movie { movie_id: i64 }, +} + impl AppState { /// Build the state, including the shared HTTP client. /// @@ -115,6 +133,7 @@ impl AppState { let (movie_commands, pending_movie_commands) = mpsc::channel(64); let (episode_commands, pending_episode_commands) = mpsc::channel(64); let (season_commands, pending_season_commands) = mpsc::channel(64); + let (metadata_commands, pending_metadata_commands) = mpsc::channel(64); Ok(Self { http, upstreams: Arc::new(upstreams), @@ -125,6 +144,8 @@ impl AppState { pending_episode_commands: Arc::new(tokio::sync::Mutex::new(pending_episode_commands)), season_commands, pending_season_commands: Arc::new(tokio::sync::Mutex::new(pending_season_commands)), + metadata_commands, + pending_metadata_commands: Arc::new(tokio::sync::Mutex::new(pending_metadata_commands)), }) } @@ -162,6 +183,16 @@ impl AppState { self.pending_season_commands.lock().await.recv().await } + /// Wait for the next on-demand metadata refresh in the daemon's metadata + /// lane. + /// + /// # Errors + /// + /// If every sender has been dropped. + pub async fn next_metadata_command(&self) -> Option { + self.pending_metadata_commands.lock().await.recv().await + } + pub(crate) fn http(&self) -> &reqwest::Client { &self.http } @@ -194,4 +225,14 @@ impl AppState { ) -> Result<(), mpsc::error::TrySendError> { self.season_commands.try_send(command) } + + /// Ask the metadata lane to refresh one title now. Best effort by + /// design: an add must not fail because the channel is full or because + /// nothing is draining it, so the caller logs and carries on. + pub(crate) fn send_metadata_command( + &self, + command: MetadataCommand, + ) -> Result<(), mpsc::error::TrySendError> { + self.metadata_commands.try_send(command) + } } diff --git a/crates/arr-daemon/src/grab.rs b/crates/arr-daemon/src/grab.rs index 7add623..66b0eb9 100644 --- a/crates/arr-daemon/src/grab.rs +++ b/crates/arr-daemon/src/grab.rs @@ -183,7 +183,7 @@ impl GrabAction { let tmdb_id = u32::try_from(movie.tmdb_id).map_err(|_| GrabError::InvalidTmdbId(movie.id))?; let metadata = tmdb.movie(tmdb_id).await?; - let changed = self.store_metadata(database, &movie, &metadata).await?; + let changed = store_movie_metadata(database, movie.id, &metadata).await?; if changed { tracing::info!( movie_id = movie.id, @@ -219,82 +219,6 @@ impl GrabAction { )) } - /// Write one refresh's fields to the row, guarded so unchanged data moves - /// nothing. Returns whether anything did. - async fn store_metadata( - &self, - database: &Db, - movie: &PendingMovie, - metadata: &arr_meta::Movie, - ) -> Result { - let title = metadata.title.clone(); - let year = metadata.year().map(i64::from); - let original_language = - (!metadata.original_language.is_empty()).then_some(metadata.original_language.clone()); - let digital_release = metadata.digital_release.map(|date| date.to_string()); - // §6.2: the id RSS matching prefers, and the one Torznab movie - // searches take. TMDB does not know one for every title. - let imdb_id = metadata.imdb_id.clone(); - // §9.6: these three are the exception to "rich detail is not - // persisted" — pure-SQL views render artwork without a TMDB call. - let poster_path = metadata.poster_path.clone(); - let backdrop_path = metadata.backdrop_path.clone(); - let vote_average = metadata.vote_average; - let title_ref = title.as_str(); - let original_language_ref = original_language.as_deref(); - let digital_release_ref = digital_release.as_deref(); - let imdb_id_ref = imdb_id.as_deref(); - let poster_path_ref = poster_path.as_deref(); - let backdrop_path_ref = backdrop_path.as_deref(); - let changed = sqlx::query!( - r#"UPDATE movies - SET title = ?, year = ?, original_language = ?, digital_release = ?, - imdb_id = ?, poster_path = ?, backdrop_path = ?, vote_average = ?, - metadata_refreshed_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now'), - search_attempts = 0, last_searched_at = NULL, - updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') - WHERE id = ? AND ( - title IS NOT ? OR year IS NOT ? OR original_language IS NOT ? - OR digital_release IS NOT ? OR imdb_id IS NOT ? - OR poster_path IS NOT ? OR backdrop_path IS NOT ? - OR vote_average IS NOT ? - )"#, - title_ref, - year, - original_language_ref, - digital_release_ref, - imdb_id_ref, - poster_path_ref, - backdrop_path_ref, - vote_average, - movie.id, - title_ref, - year, - original_language_ref, - digital_release_ref, - imdb_id_ref, - poster_path_ref, - backdrop_path_ref, - vote_average, - ) - .execute(database.pool()) - .await? - .rows_affected() - != 0; - if !changed { - // Still stamp the refresh even when nothing changed, or the TTL - // gate above never engages and every tick pays for TMDB again. - sqlx::query!( - "UPDATE movies SET metadata_refreshed_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') - WHERE id = ?", - movie.id - ) - .execute(database.pool()) - .await?; - } - Ok(changed) - } - /// Search every indexer for one title, cache each candidate with its /// verdict and score (§9.3), and return the eligible ones best first. /// @@ -999,6 +923,85 @@ async fn pending_movies(database: &Db) -> Result, GrabError> { .collect()) } +/// Write one refresh's fields to the movie row, guarded so unchanged data +/// moves nothing. Returns whether anything did. +/// +/// A free function rather than a [`GrabAction`] method because the metadata +/// lane refreshes a title on demand (issue #176) with nothing but a TMDB +/// client — grabbing needs Prowlarr, refreshing does not. +pub(crate) async fn store_movie_metadata( + database: &Db, + movie_id: i64, + metadata: &arr_meta::Movie, +) -> Result { + let title = metadata.title.clone(); + let year = metadata.year().map(i64::from); + let original_language = + (!metadata.original_language.is_empty()).then_some(metadata.original_language.clone()); + let digital_release = metadata.digital_release.map(|date| date.to_string()); + // §6.2: the id RSS matching prefers, and the one Torznab movie + // searches take. TMDB does not know one for every title. + let imdb_id = metadata.imdb_id.clone(); + // §9.6: these three are the exception to "rich detail is not + // persisted" — pure-SQL views render artwork without a TMDB call. + let poster_path = metadata.poster_path.clone(); + let backdrop_path = metadata.backdrop_path.clone(); + let vote_average = metadata.vote_average; + let title_ref = title.as_str(); + let original_language_ref = original_language.as_deref(); + let digital_release_ref = digital_release.as_deref(); + let imdb_id_ref = imdb_id.as_deref(); + let poster_path_ref = poster_path.as_deref(); + let backdrop_path_ref = backdrop_path.as_deref(); + let changed = sqlx::query!( + r#"UPDATE movies + SET title = ?, year = ?, original_language = ?, digital_release = ?, + imdb_id = ?, poster_path = ?, backdrop_path = ?, vote_average = ?, + metadata_refreshed_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now'), + search_attempts = 0, last_searched_at = NULL, + updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') + WHERE id = ? AND ( + title IS NOT ? OR year IS NOT ? OR original_language IS NOT ? + OR digital_release IS NOT ? OR imdb_id IS NOT ? + OR poster_path IS NOT ? OR backdrop_path IS NOT ? + OR vote_average IS NOT ? + )"#, + title_ref, + year, + original_language_ref, + digital_release_ref, + imdb_id_ref, + poster_path_ref, + backdrop_path_ref, + vote_average, + movie_id, + title_ref, + year, + original_language_ref, + digital_release_ref, + imdb_id_ref, + poster_path_ref, + backdrop_path_ref, + vote_average, + ) + .execute(database.pool()) + .await? + .rows_affected() + != 0; + if !changed { + // Still stamp the refresh even when nothing changed, or the TTL + // gate above never engages and every tick pays for TMDB again. + sqlx::query!( + "UPDATE movies SET metadata_refreshed_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') + WHERE id = ?", + movie_id + ) + .execute(database.pool()) + .await?; + } + Ok(changed) +} + fn search_due(movie: &PendingMovie) -> bool { backoff_elapsed(movie.search_attempts, movie.last_searched_at.as_deref()) } diff --git a/crates/arr-daemon/src/main.rs b/crates/arr-daemon/src/main.rs index d243b0e..34c45c9 100644 --- a/crates/arr-daemon/src/main.rs +++ b/crates/arr-daemon/src/main.rs @@ -8,6 +8,7 @@ mod import; mod indexers; mod jellyfin; mod manual; +mod metadata; mod notify; mod reaper; pub mod reconcile; @@ -125,6 +126,10 @@ async fn run() -> Result<(), Error> { let notifier = Notifier::new(config.ntfy_url.clone())?; let (reconcile, manual_grab, manual_tv) = reconcile_loop(&database, &config, &transmission, tmdb.as_ref(), ¬ifier)?; + // Issue #176: the on-demand half of the metadata lane needs its own + // handle — the sweep's `SeriesRefreshAction` is owned by `ReconcileLoop`, + // and the compat shim takes the other clone below. + let metadata_tmdb = tmdb.clone(); // Jellyseerr's Radarr shim (DESIGN.md §9.4) reads the same database and // needs its own TMDB client for `movie/lookup`. @@ -162,12 +167,18 @@ async fn run() -> Result<(), Error> { // in the 64-slot buffer forever. Issue #132: the episode and season // channels are drained by the same lane. let mut manual_task = tokio::spawn(manual::run( - state, - database, + state.clone(), + database.clone(), manual_grab, manual_tv, - shutdown_rx, + shutdown_rx.clone(), )); + // Issue #176: adding a title queues a refresh here rather than waiting + // for the daily sweep. Its own task, not an arm of `manual::run`: a + // refresh can take a while against TMDB and must not sit in front of an + // operator's manual search. + let mut metadata_task = + tokio::spawn(metadata::run(state, database, metadata_tmdb, shutdown_rx)); let signal_tx = shutdown_tx.clone(); let server = async move { axum::serve(listener, app) @@ -184,18 +195,28 @@ async fn run() -> Result<(), Error> { let _ = shutdown_tx.send(true); reconcile_task.await?; manual_task.await?; + metadata_task.await?; result.map_err(Error::Serve) } result = &mut reconcile_task => { result?; let _ = shutdown_tx.send(true); manual_task.await?; + metadata_task.await?; server.await.map_err(Error::Serve) } result = &mut manual_task => { result?; let _ = shutdown_tx.send(true); reconcile_task.await?; + metadata_task.await?; + server.await.map_err(Error::Serve) + } + result = &mut metadata_task => { + result?; + let _ = shutdown_tx.send(true); + reconcile_task.await?; + manual_task.await?; server.await.map_err(Error::Serve) } } diff --git a/crates/arr-daemon/src/metadata.rs b/crates/arr-daemon/src/metadata.rs new file mode 100644 index 0000000..9156580 --- /dev/null +++ b/crates/arr-daemon/src/metadata.rs @@ -0,0 +1,367 @@ +//! The on-demand half of the metadata lane. See DESIGN.md §8 and issue #176. +//! +//! The scheduled sweep is daily (`Tick::Metadata`), which is right for +//! keeping a library current and wrong for a title added a moment ago: a new +//! series has no seasons at all until a refresh reveals them, and a new movie +//! has no digital release date, which is what §6.2 gates targeted search on. +//! Shortening the tick would not fix either — it would still leave a visible +//! wait and would spend a TMDB call per title per tick. +//! +//! So `POST /api/series` and `POST /api/movies` send a [`MetadataCommand`] +//! after the row is committed, exactly as the manual search and grab +//! endpoints send theirs (issues #107 and #132), and this lane drains it. +//! The add itself never waits on TMDB and never fails because of it; the +//! worst a broken refresh can do is leave `metadata_refreshed_at` NULL, which +//! is what the sweep already treats as due. + +use std::sync::Arc; + +use arr_api::{AppState, MetadataCommand}; +use arr_db::Db; +use arr_meta::TmdbClient; +use tokio::sync::watch; + +use crate::grab::{metadata_refresh_due, store_movie_metadata, GrabError}; +use crate::series_refresh::SeriesRefreshAction; + +/// Run until every sender is dropped or `shutdown` fires. +/// +/// `tmdb` is `None` when TMDB is not configured, which already disables the +/// scheduled sweep (`main.rs` warns about it). Commands are still drained so +/// the channel never fills. +pub async fn run( + state: AppState, + database: Db, + tmdb: Option>, + mut shutdown: watch::Receiver, +) { + let series = tmdb + .as_ref() + .map(|tmdb| SeriesRefreshAction::new(Arc::clone(tmdb))); + loop { + let command = tokio::select! { + biased; + changed = shutdown.changed() => { + if changed.is_err() || *shutdown.borrow() { + return; + } + continue; + } + command = state.next_metadata_command() => command, + }; + let Some(command) = command else { + return; + }; + let (Some(tmdb), Some(series)) = (tmdb.as_ref(), series.as_ref()) else { + tracing::warn!("metadata refresh requested but TMDB is not configured"); + continue; + }; + match command { + MetadataCommand::Series { series_id } => { + // A title deleted between the send and the drain is not an + // error: `refresh_now` finds no row and does nothing. + match series.refresh_now(&database, series_id).await { + Ok(_) => {} + // A failed refresh leaves `metadata_refreshed_at` NULL, + // so the daily sweep retries the series. The add stands. + Err(error) => { + tracing::error!(series_id, %error, "on-demand series refresh failed"); + } + } + } + MetadataCommand::Movie { movie_id } => { + if let Err(error) = refresh_movie(tmdb, &database, movie_id).await { + tracing::error!(movie_id, %error, "on-demand movie refresh failed"); + } + } + } + } +} + +/// Refresh one movie now, on the same terms as the series path: a row that +/// is gone is not an error, and the sweep's TTL gate applies so a title +/// something else already refreshed costs no second TMDB call. +async fn refresh_movie(tmdb: &TmdbClient, database: &Db, movie_id: i64) -> Result<(), GrabError> { + let movie = sqlx::query!( + r#"SELECT tmdb_id AS "tmdb_id!: i64", metadata_refreshed_at + FROM movies WHERE id = ?"#, + movie_id + ) + .fetch_optional(database.pool()) + .await?; + let Some(movie) = movie else { + return Ok(()); + }; + if !metadata_refresh_due(movie.metadata_refreshed_at.as_deref()) { + return Ok(()); + } + let tmdb_id = u32::try_from(movie.tmdb_id).map_err(|_| GrabError::InvalidTmdbId(movie_id))?; + let metadata = tmdb.movie(tmdb_id).await?; + store_movie_metadata(database, movie_id, &metadata).await?; + Ok(()) +} + +#[cfg(test)] +#[allow(clippy::unwrap_used)] +mod tests { + use arr_meta::TmdbClient; + use tempfile::TempDir; + use wiremock::matchers::{method, path, query_param}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + use super::*; + + const MOVIE_METADATA: &str = r#"{ + "id": 693134, + "title": "Dune Part Two", + "original_title": "Dune: Part Two", + "original_language": "en", + "origin_country": ["US"], + "release_date": "2024-02-27", + "poster_path": "/dune-two.jpg", + "release_dates": {"results": [{"release_dates": [{ + "type": 4, "release_date": "2024-04-16T00:00:00.000Z" + }]}]} + }"#; + + fn series_detail() -> serde_json::Value { + serde_json::json!({ + "id": 82_728, + "name": "Bluey", + "original_language": "en", + "first_air_date": "2018-10-01", + "status": "Returning Series", + "seasons": [{"season_number": 1, "episode_count": 2}], + "external_ids": {"tvdb_id": 361_391} + }) + } + + fn season_detail() -> serde_json::Value { + serde_json::json!({ + "season_number": 1, + "episodes": [ + {"episode_number": 1, "name": "Magic Xylophone", "air_date": "2018-10-01"}, + {"episode_number": 2, "name": "Hospital", "air_date": "2018-10-02"} + ] + }) + } + + /// A TMDB that answers for the series, and one that answers for the + /// movie. `status` lets a test serve an error instead. + async fn tmdb_series(status: u16) -> MockServer { + let server = MockServer::start().await; + let response = if status == 200 { + ResponseTemplate::new(200).set_body_json(series_detail()) + } else { + ResponseTemplate::new(status) + }; + Mock::given(method("GET")) + .and(path("/tv/82728")) + .respond_with(response) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path("/tv/82728/season/1")) + .respond_with(ResponseTemplate::new(200).set_body_json(season_detail())) + .mount(&server) + .await; + server + } + + async fn tmdb_movie(status: u16) -> MockServer { + let server = MockServer::start().await; + let response = if status == 200 { + ResponseTemplate::new(200).set_body_string(MOVIE_METADATA) + } else { + ResponseTemplate::new(status) + }; + Mock::given(method("GET")) + .and(path("/3/movie/693134")) + .and(query_param("append_to_response", "release_dates")) + .respond_with(response) + .mount(&server) + .await; + server + } + + fn series_action(server: &MockServer) -> SeriesRefreshAction { + SeriesRefreshAction::new(Arc::new( + TmdbClient::builder("key".to_owned()) + .base_url(server.uri()) + .build() + .unwrap(), + )) + } + + fn movie_client(server: &MockServer) -> TmdbClient { + TmdbClient::builder("key".to_owned()) + .base_url(format!("{}/3/", server.uri())) + .build() + .unwrap() + } + + /// A series and a movie as `POST /api/series` and `POST /api/movies` + /// leave them: added, never refreshed. + async fn added_titles() -> (TempDir, Db) { + let directory = tempfile::tempdir().unwrap(); + let database = Db::connect(directory.path().join("arr.db")).await.unwrap(); + database.migrate().await.unwrap(); + sqlx::query( + "INSERT INTO series (tmdb_id, title, root_id, auto_track) + SELECT 82728, 'Bluey', id, 1 FROM roots WHERE kind = 'tv' AND audience = 'main'", + ) + .execute(database.pool()) + .await + .unwrap(); + sqlx::query( + "INSERT INTO movies (tmdb_id, title, year, original_language, root_id) + SELECT 693134, 'Dune Part Two', 2024, 'en', id + FROM roots WHERE kind = 'movie' AND audience = 'main'", + ) + .execute(database.pool()) + .await + .unwrap(); + (directory, database) + } + + async fn series_stamp(database: &Db) -> Option { + sqlx::query_scalar("SELECT metadata_refreshed_at FROM series WHERE id = 1") + .fetch_one(database.pool()) + .await + .unwrap() + } + + async fn movie_stamp(database: &Db) -> Option { + sqlx::query_scalar("SELECT metadata_refreshed_at FROM movies WHERE id = 1") + .fetch_one(database.pool()) + .await + .unwrap() + } + + /// Issue #176: the whole point. A series added a moment ago has its + /// seasons revealed by the command, not by a tick up to a day away. + #[tokio::test] + async fn an_added_series_is_refreshed_on_demand() { + let (_dir, database) = added_titles().await; + let server = tmdb_series(200).await; + + series_action(&server) + .refresh_now(&database, 1) + .await + .unwrap(); + + let episodes: i64 = sqlx::query_scalar("SELECT count(*) FROM episodes") + .fetch_one(database.pool()) + .await + .unwrap(); + assert_eq!(episodes, 2, "the seasons TMDB knows must be revealed"); + assert!(series_stamp(&database).await.is_some()); + } + + /// A movie's digital release date is what §6.2 gates targeted search on, + /// and it only ever arrives from a refresh. + #[tokio::test] + async fn an_added_movie_is_refreshed_on_demand() { + let (_dir, database) = added_titles().await; + let server = tmdb_movie(200).await; + + refresh_movie(&movie_client(&server), &database, 1) + .await + .unwrap(); + + let (digital_release, poster): (Option, Option) = + sqlx::query_as("SELECT digital_release, poster_path FROM movies WHERE id = 1") + .fetch_one(database.pool()) + .await + .unwrap(); + assert_eq!(digital_release.as_deref(), Some("2024-04-16")); + assert_eq!(poster.as_deref(), Some("/dune-two.jpg")); + assert!(movie_stamp(&database).await.is_some()); + } + + /// A failed refresh must not roll the add back and must not stamp the + /// row, or the scheduled sweep would treat the title as done. + #[tokio::test] + async fn a_failed_series_refresh_leaves_the_stamp_null_for_the_sweep() { + let (_dir, database) = added_titles().await; + let server = tmdb_series(500).await; + + let result = series_action(&server).refresh_now(&database, 1).await; + + assert!(result.is_err()); + assert!(series_stamp(&database).await.is_none()); + let series: i64 = sqlx::query_scalar("SELECT count(*) FROM series") + .fetch_one(database.pool()) + .await + .unwrap(); + assert_eq!(series, 1, "the add is not rolled back"); + } + + #[tokio::test] + async fn a_failed_movie_refresh_leaves_the_stamp_null_for_the_sweep() { + let (_dir, database) = added_titles().await; + let server = tmdb_movie(500).await; + + let result = refresh_movie(&movie_client(&server), &database, 1).await; + + assert!(result.is_err()); + assert!(movie_stamp(&database).await.is_none()); + let movies: i64 = sqlx::query_scalar("SELECT count(*) FROM movies") + .fetch_one(database.pool()) + .await + .unwrap(); + assert_eq!(movies, 1, "the add is not rolled back"); + } + + /// Add then delete before the lane drains: nothing to refresh, and that + /// is not a failure. + #[tokio::test] + async fn a_command_naming_a_deleted_title_is_not_an_error() { + let (_dir, database) = added_titles().await; + let series = tmdb_series(200).await; + let movie = tmdb_movie(200).await; + + series_action(&series) + .refresh_now(&database, 404) + .await + .unwrap(); + refresh_movie(&movie_client(&movie), &database, 404) + .await + .unwrap(); + + assert!(series.received_requests().await.unwrap().is_empty()); + assert!(movie.received_requests().await.unwrap().is_empty()); + } + + /// The TTL gate the scheduled sweep uses applies here too: a title the + /// sweep just refreshed does not pay for a second TMDB call. + #[tokio::test] + async fn a_title_the_sweep_just_refreshed_costs_no_second_call() { + let (_dir, database) = added_titles().await; + let series = tmdb_series(200).await; + let movie = tmdb_movie(200).await; + sqlx::query( + "UPDATE series SET metadata_refreshed_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')", + ) + .execute(database.pool()) + .await + .unwrap(); + sqlx::query( + "UPDATE movies SET metadata_refreshed_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')", + ) + .execute(database.pool()) + .await + .unwrap(); + + series_action(&series) + .refresh_now(&database, 1) + .await + .unwrap(); + refresh_movie(&movie_client(&movie), &database, 1) + .await + .unwrap(); + + assert!(series.received_requests().await.unwrap().is_empty()); + assert!(movie.received_requests().await.unwrap().is_empty()); + } +} diff --git a/crates/arr-daemon/src/series_refresh.rs b/crates/arr-daemon/src/series_refresh.rs index 8a7880a..46d7471 100644 --- a/crates/arr-daemon/src/series_refresh.rs +++ b/crates/arr-daemon/src/series_refresh.rs @@ -46,7 +46,7 @@ fn is_upstream_ended(status: &str) -> bool { } #[derive(Debug, thiserror::Error)] -enum RefreshError { +pub(crate) enum RefreshError { #[error("database: {0}")] Database(#[from] sqlx::Error), #[error("tmdb: {0}")] @@ -101,6 +101,38 @@ impl SeriesRefreshAction { Ok(outcomes) } + /// Refresh one series now, for the on-demand metadata lane (issue #176). + /// + /// A series deleted between the command being sent and drained is not an + /// error — there is simply nothing to refresh. The sweep's TTL gate + /// applies here too, so a title the sweep already refreshed costs no + /// second TMDB call. + pub(crate) async fn refresh_now( + &self, + database: &Db, + series_id: i64, + ) -> Result, RefreshError> { + let stale = sqlx::query_as!( + DueSeries, + r#"SELECT id AS "id!: i64", tmdb_id AS "tmdb_id!: i64", tvdb_id, + title AS "title!: String", year, original_language, + root_id AS "root_id!: i64", auto_track AS "auto_track!: bool", + upstream_ended AS "upstream_ended!: bool", metadata_refreshed_at, + poster_path, backdrop_path, vote_average + FROM series WHERE id = ?"#, + series_id + ) + .fetch_optional(database.pool()) + .await?; + let Some(stale) = stale else { + return Ok(None); + }; + if !metadata_refresh_due(stale.metadata_refreshed_at.as_deref()) { + return Ok(None); + } + self.refresh_series(database, &stale).await + } + async fn refresh_series( &self, database: &Db,