feat(daemon): refresh metadata on add

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 <noreply@anthropic.com>
This commit is contained in:
Miguel Palhas
2026-08-24 18:12:10 +01:00
co-authored by Claude Opus 5
parent e9806d7a84
commit d8797e1996
13 changed files with 803 additions and 100 deletions
@@ -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"
}
@@ -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"
}
@@ -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"
}
@@ -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"
}
@@ -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"
}
+2 -1
View File
@@ -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};
+57 -2
View File
@@ -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<String> =
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;
+67 -2
View File
@@ -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::<serde_json::Value>().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<String> =
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;
+41
View File
@@ -74,6 +74,8 @@ pub struct AppState {
pending_episode_commands: Arc<tokio::sync::Mutex<mpsc::Receiver<EpisodeCommand>>>,
season_commands: mpsc::Sender<SeasonCommand>,
pending_season_commands: Arc<tokio::sync::Mutex<mpsc::Receiver<SeasonCommand>>>,
metadata_commands: mpsc::Sender<MetadataCommand>,
pending_metadata_commands: Arc<tokio::sync::Mutex<mpsc::Receiver<MetadataCommand>>>,
}
/// 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<MetadataCommand> {
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<SeasonCommand>> {
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<MetadataCommand>> {
self.metadata_commands.try_send(command)
}
}
+80 -77
View File
@@ -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<bool, GrabError> {
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<Vec<PendingMovie>, 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<bool, GrabError> {
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())
}
+24 -3
View File
@@ -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(), &notifier)?;
// 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)
}
}
+367
View File
@@ -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<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());
}
}
+33 -1
View File
@@ -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<Option<Outcome>, 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,