arr/crates/arr-daemon/src/metadata.rs

//! 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<Arc<TmdbClient>>,
    mut shutdown: watch::Receiver<bool>,
) {
    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<String> {
        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<String> {
        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<String>, Option<String>) =
            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());
    }
}