//! 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()); } }