feat(daemon): daily series metadata refresh lane
This commit is contained in:
Miguel Palhas
@@ -1121,7 +1121,7 @@ pub(crate) fn backoff_elapsed(search_attempts: i64, last_searched_at: Option<&st
|
||||
last_searched_at.with_timezone(&chrono::Utc) + backoff <= chrono::Utc::now()
|
||||
}
|
||||
|
||||
fn metadata_refresh_due(metadata_refreshed_at: Option<&str>) -> bool {
|
||||
pub(crate) fn metadata_refresh_due(metadata_refreshed_at: Option<&str>) -> bool {
|
||||
let Some(refreshed_at) = metadata_refreshed_at else {
|
||||
return true;
|
||||
};
|
||||
|
||||
@@ -12,6 +12,7 @@ mod notify;
|
||||
mod reaper;
|
||||
pub mod reconcile;
|
||||
mod rss;
|
||||
mod series_refresh;
|
||||
mod tv_grab;
|
||||
mod web;
|
||||
|
||||
@@ -31,6 +32,7 @@ use notify::Notifier;
|
||||
use reaper::ReaperAction;
|
||||
use reconcile::{ReconcileLoop, Tick};
|
||||
use rss::RssAction;
|
||||
use series_refresh::SeriesRefreshAction;
|
||||
use tower_http::trace::TraceLayer;
|
||||
use tv_grab::TvGrabAction;
|
||||
|
||||
@@ -268,6 +270,13 @@ fn reconcile_loop(
|
||||
),
|
||||
);
|
||||
}
|
||||
// §8: metadata refresh is its own daily lane, staggered against the
|
||||
// other ticks. Without TMDB there is nothing to refresh from.
|
||||
if let Some(tmdb) = tmdb {
|
||||
reconcile = reconcile.register(Tick::Metadata, SeriesRefreshAction::new(Arc::clone(tmdb)));
|
||||
} else {
|
||||
tracing::warn!("TMDB is not configured: series metadata refresh is disabled");
|
||||
}
|
||||
let jellyfin = jellyfin::JellyfinClient::new(
|
||||
config.jellyfin_url.clone(),
|
||||
config.jellyfin_api_key.clone(),
|
||||
|
||||
@@ -0,0 +1,731 @@
|
||||
//! The daily series metadata refresh. See DESIGN.md §8 and issue #121.
|
||||
//!
|
||||
//! The database rows are the work list (§8): a series whose
|
||||
//! `metadata_refreshed_at` has passed its time-to-live is due. One refresh
|
||||
//! pulls the series detail from TMDB and closes every gap it reveals:
|
||||
//!
|
||||
//! - seasons TMDB knows that the library does not are inserted, with
|
||||
//! `apply_auto_track` deciding their tracking rule;
|
||||
//! - episodes revealed into an existing season arrive wanted when that season
|
||||
//! is `tracked` (`apply_tracked`, §4.1);
|
||||
//! - air dates and titles of known episodes follow TMDB, because both §4.2's
|
||||
//! `airing` window and §6.2's do-not-search-before-it-exists gate read them;
|
||||
//! - `upstream_ended` follows TMDB's status, which `ended` is derived from;
|
||||
//! - a missing `tvdb_id` is backfilled (#120).
|
||||
//!
|
||||
//! Episodes that vanish or renumber upstream are left alone (#122). Refresh
|
||||
//! is idempotent: over unchanged TMDB data only the stamp moves.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
|
||||
use arr_core::tracking::{apply_auto_track, apply_tracked, RefreshedSeason};
|
||||
use arr_core::{
|
||||
Episode as CoreEpisode, EpisodeId, Language, MediaState, RootId, SeasonId, SeriesId,
|
||||
TitleOverrides,
|
||||
};
|
||||
use arr_db::policy::language;
|
||||
use arr_db::Db;
|
||||
use arr_meta::{Season as TmdbSeasonDetail, Series as TmdbSeries, TmdbClient};
|
||||
|
||||
use crate::grab::metadata_refresh_due;
|
||||
use crate::reconcile::{Action, ActionFuture, Outcome};
|
||||
|
||||
/// TMDB statuses that mean no further episodes will ever appear.
|
||||
fn is_upstream_ended(status: &str) -> bool {
|
||||
matches!(status, "Ended" | "Canceled")
|
||||
}
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
enum RefreshError {
|
||||
#[error("database: {0}")]
|
||||
Database(#[from] sqlx::Error),
|
||||
#[error("tmdb: {0}")]
|
||||
Tmdb(#[from] arr_meta::Error),
|
||||
#[error("series {0}: tmdb_id does not fit")]
|
||||
InvalidTmdbId(i64),
|
||||
}
|
||||
|
||||
/// Refreshes one series' metadata from TMDB per day (§8), on the same TTL
|
||||
/// gate as the movie lane.
|
||||
#[derive(Debug)]
|
||||
pub struct SeriesRefreshAction {
|
||||
tmdb: Arc<TmdbClient>,
|
||||
}
|
||||
|
||||
impl SeriesRefreshAction {
|
||||
#[must_use]
|
||||
pub fn new(tmdb: Arc<TmdbClient>) -> Self {
|
||||
Self { tmdb }
|
||||
}
|
||||
|
||||
async fn tick(&self, database: &Db) -> Result<Vec<Outcome>, RefreshError> {
|
||||
let due = 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
|
||||
FROM series
|
||||
ORDER BY metadata_refreshed_at IS NOT NULL, metadata_refreshed_at, id"#
|
||||
)
|
||||
.fetch_all(database.pool())
|
||||
.await?;
|
||||
let mut outcomes = Vec::new();
|
||||
for stale in due {
|
||||
if !metadata_refresh_due(stale.metadata_refreshed_at.as_deref()) {
|
||||
continue;
|
||||
}
|
||||
match self.refresh_series(database, &stale).await {
|
||||
Ok(Some(outcome)) => outcomes.push(outcome),
|
||||
Ok(None) => {}
|
||||
// One series' failure must not cost the rest of the tick.
|
||||
Err(error) => tracing::error!(
|
||||
series_id = stale.id,
|
||||
title = stale.title,
|
||||
%error,
|
||||
"series metadata refresh failed"
|
||||
),
|
||||
}
|
||||
}
|
||||
Ok(outcomes)
|
||||
}
|
||||
|
||||
async fn refresh_series(
|
||||
&self,
|
||||
database: &Db,
|
||||
stale: &DueSeries,
|
||||
) -> Result<Option<Outcome>, RefreshError> {
|
||||
let tmdb_id =
|
||||
u32::try_from(stale.tmdb_id).map_err(|_| RefreshError::InvalidTmdbId(stale.id))?;
|
||||
let metadata = self.tmdb.series(tmdb_id).await?;
|
||||
|
||||
let mut transaction = database.pool().begin().await?;
|
||||
let mut changed = self
|
||||
.sync_series_fields(&mut transaction, stale, &metadata)
|
||||
.await?;
|
||||
|
||||
// Existing seasons keep their tracking rule; their episode lists are
|
||||
// brought up to date in place. Seasons TMDB knows and the library
|
||||
// does not go through the one function that owns the reveal rule.
|
||||
let existing = sqlx::query!(
|
||||
r#"SELECT id AS "id!: i64", number AS "number!: i64", tracked AS "tracked!: bool"
|
||||
FROM seasons WHERE series_id = ?"#,
|
||||
stale.id
|
||||
)
|
||||
.fetch_all(&mut *transaction)
|
||||
.await?
|
||||
.into_iter()
|
||||
.map(|season| (season.number, (season.id, season.tracked)))
|
||||
.collect::<HashMap<_, _>>();
|
||||
let mut revealed = Vec::new();
|
||||
for summary in &metadata.seasons {
|
||||
if let Some(&(season_id, tracked)) = existing.get(&i64::from(summary.number)) {
|
||||
changed |= self
|
||||
.sync_season(
|
||||
&mut transaction,
|
||||
season_id,
|
||||
tracked,
|
||||
tmdb_id,
|
||||
summary.number,
|
||||
)
|
||||
.await?;
|
||||
} else {
|
||||
let detail = self.tmdb.season(tmdb_id, summary.number).await?;
|
||||
revealed.push(RefreshedSeason {
|
||||
season: arr_core::Season {
|
||||
id: SeasonId(0),
|
||||
series_id: SeriesId(stale.id),
|
||||
number: season_number(summary.number),
|
||||
tracked: false,
|
||||
},
|
||||
episodes: core_episodes(stale.id, &detail),
|
||||
is_new: true,
|
||||
});
|
||||
}
|
||||
}
|
||||
apply_auto_track(&core_series(stale), &mut revealed);
|
||||
for refreshed in revealed {
|
||||
self.insert_revealed_season(&mut transaction, stale.id, refreshed)
|
||||
.await?;
|
||||
changed = true;
|
||||
}
|
||||
|
||||
transaction.commit().await?;
|
||||
stamp_refreshed(database, stale.id).await?;
|
||||
Ok(changed.then(|| {
|
||||
Outcome::new(
|
||||
format!("series {} ({})", stale.id, stale.title),
|
||||
"refreshed metadata from TMDB",
|
||||
)
|
||||
}))
|
||||
}
|
||||
|
||||
/// The series-level fields a refresh owns. Both writes are guarded so
|
||||
/// unchanged data moves nothing.
|
||||
async fn sync_series_fields(
|
||||
&self,
|
||||
executor: &mut sqlx::SqliteConnection,
|
||||
stale: &DueSeries,
|
||||
metadata: &TmdbSeries,
|
||||
) -> Result<bool, RefreshError> {
|
||||
let mut changed = false;
|
||||
// §6.1: `t=tvsearch` prefers the TVDB id. Series added before the id
|
||||
// existed on add (#120) pick theirs up here.
|
||||
if stale.tvdb_id.is_none() {
|
||||
if let Some(tvdb_id) = metadata.tvdb_id.map(i64::from) {
|
||||
sqlx::query!(
|
||||
"UPDATE series SET tvdb_id = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') WHERE id = ?",
|
||||
tvdb_id,
|
||||
stale.id
|
||||
)
|
||||
.execute(&mut *executor)
|
||||
.await?;
|
||||
changed = true;
|
||||
}
|
||||
}
|
||||
let ended = is_upstream_ended(&metadata.status);
|
||||
if ended != stale.upstream_ended {
|
||||
sqlx::query!(
|
||||
"UPDATE series SET upstream_ended = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') WHERE id = ?",
|
||||
ended,
|
||||
stale.id
|
||||
)
|
||||
.execute(&mut *executor)
|
||||
.await?;
|
||||
changed = true;
|
||||
}
|
||||
Ok(changed)
|
||||
}
|
||||
|
||||
/// Persist one season the library has not seen, with `apply_auto_track`'s
|
||||
/// verdict on its tracking rule already applied.
|
||||
async fn insert_revealed_season(
|
||||
&self,
|
||||
executor: &mut sqlx::SqliteConnection,
|
||||
series_id: i64,
|
||||
refreshed: RefreshedSeason,
|
||||
) -> Result<(), RefreshError> {
|
||||
let season_number = i64::from(refreshed.season.number);
|
||||
let season_id = sqlx::query_scalar!(
|
||||
"INSERT INTO seasons (series_id, number, tracked) VALUES (?, ?, ?) RETURNING id",
|
||||
series_id,
|
||||
season_number,
|
||||
refreshed.season.tracked
|
||||
)
|
||||
.fetch_one(&mut *executor)
|
||||
.await?;
|
||||
for episode in &refreshed.episodes {
|
||||
let episode_number = i64::from(episode.number);
|
||||
let episode_title = episode.title.clone();
|
||||
let air_date = air_date_string(episode.air_date);
|
||||
sqlx::query!(
|
||||
"INSERT INTO episodes (season_id, number, title, air_date, wanted) VALUES (?, ?, ?, ?, ?)",
|
||||
season_id,
|
||||
episode_number,
|
||||
episode_title,
|
||||
air_date,
|
||||
episode.wanted
|
||||
)
|
||||
.execute(&mut *executor)
|
||||
.await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Bring one persisted season's episodes up to date with TMDB's.
|
||||
///
|
||||
/// Known episodes get guarded updates — nothing writes unless a value
|
||||
/// actually moved — and unknown numbers are revealed wanted exactly when
|
||||
/// the season is tracked (§4.1). Numbers absent from TMDB stay untouched.
|
||||
async fn sync_season(
|
||||
&self,
|
||||
executor: &mut sqlx::SqliteConnection,
|
||||
season_id: i64,
|
||||
tracked: bool,
|
||||
tmdb_id: u32,
|
||||
number: u32,
|
||||
) -> Result<bool, RefreshError> {
|
||||
let detail = self.tmdb.season(tmdb_id, number).await?;
|
||||
let known = sqlx::query!(
|
||||
r#"SELECT id AS "id!: i64", number AS "number!: i64",
|
||||
title AS "title!: String", air_date
|
||||
FROM episodes WHERE season_id = ?"#,
|
||||
season_id
|
||||
)
|
||||
.fetch_all(&mut *executor)
|
||||
.await?
|
||||
.into_iter()
|
||||
.map(|episode| {
|
||||
(
|
||||
episode.number,
|
||||
(episode.id, episode.title, episode.air_date),
|
||||
)
|
||||
})
|
||||
.collect::<HashMap<_, _>>();
|
||||
|
||||
let mut fresh = Vec::new();
|
||||
let mut changed = false;
|
||||
for source in &detail.episodes {
|
||||
let number = i64::from(source.number);
|
||||
if let Some(&(id, ref title, ref air_date)) = known.get(&number) {
|
||||
let new_air_date = source.air_date.map(|date| date.to_string());
|
||||
if *title != source.title || *air_date != new_air_date {
|
||||
sqlx::query!(
|
||||
"UPDATE episodes SET title = ?, air_date = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') WHERE id = ? AND (title IS NOT ? OR air_date IS NOT ?)",
|
||||
source.title,
|
||||
new_air_date,
|
||||
id,
|
||||
source.title,
|
||||
new_air_date
|
||||
)
|
||||
.execute(&mut *executor)
|
||||
.await?;
|
||||
changed = true;
|
||||
}
|
||||
} else {
|
||||
fresh.push(CoreEpisode {
|
||||
id: EpisodeId(0),
|
||||
season_id: SeasonId(season_id),
|
||||
number: season_number(source.number),
|
||||
title: source.title.clone(),
|
||||
air_date: source.air_date.map(system_time),
|
||||
wanted: false,
|
||||
state: MediaState::Missing,
|
||||
search_attempts: 0,
|
||||
last_searched_at: None,
|
||||
});
|
||||
}
|
||||
}
|
||||
apply_tracked(tracked, &mut fresh);
|
||||
for episode in fresh {
|
||||
let episode_number = i64::from(episode.number);
|
||||
let air_date = air_date_string(episode.air_date);
|
||||
sqlx::query!(
|
||||
"INSERT INTO episodes (season_id, number, title, air_date, wanted) VALUES (?, ?, ?, ?, ?)",
|
||||
season_id,
|
||||
episode_number,
|
||||
episode.title,
|
||||
air_date,
|
||||
episode.wanted
|
||||
)
|
||||
.execute(&mut *executor)
|
||||
.await?;
|
||||
changed = true;
|
||||
}
|
||||
Ok(changed)
|
||||
}
|
||||
}
|
||||
|
||||
impl Action for SeriesRefreshAction {
|
||||
fn name(&self) -> &'static str {
|
||||
"series-metadata-refresh"
|
||||
}
|
||||
|
||||
fn run<'a>(&'a self, database: &'a Db) -> ActionFuture<'a> {
|
||||
Box::pin(async move { self.tick(database).await.map_err(Into::into) })
|
||||
}
|
||||
}
|
||||
|
||||
struct DueSeries {
|
||||
id: i64,
|
||||
tmdb_id: i64,
|
||||
tvdb_id: Option<i64>,
|
||||
title: String,
|
||||
year: Option<i64>,
|
||||
original_language: Option<String>,
|
||||
root_id: i64,
|
||||
auto_track: bool,
|
||||
upstream_ended: bool,
|
||||
metadata_refreshed_at: Option<String>,
|
||||
}
|
||||
|
||||
/// TMDB numbers are unbounded; ours are `u16` (`CHECK (number >= 0)`,
|
||||
/// STRICT). A number past `u16::MAX` cannot match anything real and would
|
||||
/// never be grabbed anyway.
|
||||
fn season_number(number: u32) -> u16 {
|
||||
u16::try_from(number.min(u32::from(u16::MAX))).unwrap_or(u16::MAX)
|
||||
}
|
||||
|
||||
fn core_series(series: &DueSeries) -> arr_core::Series {
|
||||
arr_core::Series {
|
||||
id: SeriesId(series.id),
|
||||
tmdb_id: u64::try_from(series.tmdb_id).unwrap_or_default(),
|
||||
title: series.title.clone(),
|
||||
year: series
|
||||
.year
|
||||
.and_then(|year| u16::try_from(year).ok())
|
||||
.unwrap_or_default(),
|
||||
original_language: series
|
||||
.original_language
|
||||
.as_deref()
|
||||
.map_or(Language::Other(String::new()), language),
|
||||
root_id: RootId(series.root_id),
|
||||
auto_track: series.auto_track,
|
||||
overrides: TitleOverrides::default(),
|
||||
upstream_ended: series.upstream_ended,
|
||||
blocked: false,
|
||||
}
|
||||
}
|
||||
|
||||
fn core_episodes(series_id: i64, detail: &TmdbSeasonDetail) -> Vec<CoreEpisode> {
|
||||
detail
|
||||
.episodes
|
||||
.iter()
|
||||
.map(|source| CoreEpisode {
|
||||
id: EpisodeId(0),
|
||||
season_id: SeasonId(series_id),
|
||||
number: season_number(source.number),
|
||||
title: source.title.clone(),
|
||||
air_date: source.air_date.map(system_time),
|
||||
wanted: false,
|
||||
state: MediaState::Missing,
|
||||
search_attempts: 0,
|
||||
last_searched_at: None,
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// An `air_date` as the API writes it, so a refresh and a hand-created
|
||||
/// season agree byte for byte and the guarded update sees no change.
|
||||
fn air_date_string(air_date: Option<std::time::SystemTime>) -> Option<String> {
|
||||
air_date.map(|when| {
|
||||
chrono::DateTime::<chrono::Utc>::from(when)
|
||||
.date_naive()
|
||||
.to_string()
|
||||
})
|
||||
}
|
||||
|
||||
/// Dates arrive as `%Y-%m-%d`; midnight UTC carries them losslessly.
|
||||
fn system_time(date: chrono::NaiveDate) -> std::time::SystemTime {
|
||||
date.and_time(chrono::NaiveTime::MIN).and_utc().into()
|
||||
}
|
||||
|
||||
async fn stamp_refreshed(database: &Db, series_id: i64) -> Result<(), RefreshError> {
|
||||
// Stamped even when nothing changed, or the TTL gate above never engages
|
||||
// and every tick pays for TMDB again — same reason as the movie lane.
|
||||
sqlx::query!(
|
||||
"UPDATE series SET metadata_refreshed_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') WHERE id = ?",
|
||||
series_id
|
||||
)
|
||||
.execute(database.pool())
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[allow(clippy::unwrap_used)]
|
||||
mod tests {
|
||||
use std::sync::Arc;
|
||||
|
||||
use arr_db::Db;
|
||||
use serde_json::json;
|
||||
use tempfile::TempDir;
|
||||
use wiremock::matchers::{method, path};
|
||||
use wiremock::{Mock, MockServer, ResponseTemplate};
|
||||
|
||||
use super::*;
|
||||
|
||||
fn series_body(status: &str) -> serde_json::Value {
|
||||
json!({
|
||||
"id": 82_728,
|
||||
"name": "Bluey",
|
||||
"original_language": "en",
|
||||
"first_air_date": "2018-10-01",
|
||||
"status": status,
|
||||
"seasons": [
|
||||
{"season_number": 1, "episode_count": 2},
|
||||
{"season_number": 2, "episode_count": 1}
|
||||
],
|
||||
"external_ids": {"tvdb_id": 361_391}
|
||||
})
|
||||
}
|
||||
|
||||
fn season_one_body(episodes: &serde_json::Value) -> serde_json::Value {
|
||||
json!({"season_number": 1, "episodes": episodes})
|
||||
}
|
||||
|
||||
fn two_episodes() -> serde_json::Value {
|
||||
json!([
|
||||
{"episode_number": 1, "name": "Magic Xylophone", "air_date": "2018-10-01"},
|
||||
{"episode_number": 2, "name": "Hospital", "air_date": "2018-10-02"}
|
||||
])
|
||||
}
|
||||
|
||||
fn season_two_body() -> serde_json::Value {
|
||||
json!({
|
||||
"season_number": 2,
|
||||
"episodes": [
|
||||
{"episode_number": 1, "name": "Dance Mode", "air_date": "2019-04-01"}
|
||||
]
|
||||
})
|
||||
}
|
||||
|
||||
/// A TMDB serving one series with two seasons. `reset` between phases of
|
||||
/// a test and remount, counting requests by delta around each phase.
|
||||
async fn tmdb(status: &str, season_one: serde_json::Value) -> MockServer {
|
||||
let server = MockServer::start().await;
|
||||
mount(&server, status, season_one).await;
|
||||
server
|
||||
}
|
||||
|
||||
async fn mount(server: &MockServer, status: &str, season_one: serde_json::Value) {
|
||||
Mock::given(method("GET"))
|
||||
.and(path("/tv/82728"))
|
||||
.respond_with(ResponseTemplate::new(200).set_body_json(series_body(status)))
|
||||
.mount(server)
|
||||
.await;
|
||||
Mock::given(method("GET"))
|
||||
.and(path("/tv/82728/season/1"))
|
||||
.respond_with(ResponseTemplate::new(200).set_body_json(season_one))
|
||||
.mount(server)
|
||||
.await;
|
||||
Mock::given(method("GET"))
|
||||
.and(path("/tv/82728/season/2"))
|
||||
.respond_with(ResponseTemplate::new(200).set_body_json(season_two_body()))
|
||||
.mount(server)
|
||||
.await;
|
||||
}
|
||||
|
||||
/// Fresh client per tick, as a restarted process would have — so the only
|
||||
/// thing that can suppress a request is the persisted TTL gate, never the
|
||||
/// in-process response cache.
|
||||
fn action(server: &MockServer) -> SeriesRefreshAction {
|
||||
SeriesRefreshAction::new(Arc::new(
|
||||
TmdbClient::builder("test-key".to_owned())
|
||||
.base_url(server.uri())
|
||||
.build()
|
||||
.unwrap(),
|
||||
))
|
||||
}
|
||||
|
||||
async fn seeded_series(auto_track: bool) -> (TempDir, Db) {
|
||||
let directory = tempfile::tempdir().unwrap();
|
||||
let database = Db::connect(directory.path().join("arr.db")).await.unwrap();
|
||||
database.migrate().await.unwrap();
|
||||
let root_id: i64 =
|
||||
sqlx::query_scalar("SELECT id FROM roots WHERE kind = 'tv' AND audience = 'main'")
|
||||
.fetch_one(database.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
sqlx::query(
|
||||
"INSERT INTO series (tmdb_id, title, root_id, auto_track) VALUES (82728, 'Bluey', ?, ?)",
|
||||
)
|
||||
.bind(root_id)
|
||||
.bind(auto_track)
|
||||
.execute(database.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
(directory, database)
|
||||
}
|
||||
|
||||
/// Force the next tick past the refresh TTL.
|
||||
async fn expire_refresh(database: &Db) {
|
||||
sqlx::query(
|
||||
"UPDATE series SET metadata_refreshed_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-7 hours')",
|
||||
)
|
||||
.execute(database.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn first_refresh_reveals_seasons_and_applies_auto_track() {
|
||||
let (_dir, database) = seeded_series(true).await;
|
||||
let server = tmdb("Returning Series", season_one_body(&two_episodes())).await;
|
||||
|
||||
let outcomes = action(&server).tick(&database).await.unwrap();
|
||||
|
||||
assert_eq!(outcomes.len(), 1);
|
||||
let rows: Vec<(i64, bool)> =
|
||||
sqlx::query_as("SELECT number, tracked FROM seasons ORDER BY number")
|
||||
.fetch_all(database.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(rows, vec![(1, true), (2, true)]);
|
||||
let wanted: i64 = sqlx::query_scalar(
|
||||
"SELECT count(*) FROM episodes WHERE wanted = 1 AND state = 'missing'",
|
||||
)
|
||||
.fetch_one(database.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(wanted, 3);
|
||||
// #120 backfill: the series was added before it had an id on file.
|
||||
let tvdb_id: Option<i64> =
|
||||
sqlx::query_scalar("SELECT tvdb_id FROM series WHERE tmdb_id = 82728")
|
||||
.fetch_one(database.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(tvdb_id, Some(361_391));
|
||||
let stamped: i64 = sqlx::query_scalar(
|
||||
"SELECT count(*) FROM series WHERE metadata_refreshed_at IS NOT NULL",
|
||||
)
|
||||
.fetch_one(database.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(stamped, 1);
|
||||
}
|
||||
|
||||
/// `auto_track` is a rule about reveals, not intent: without it nothing
|
||||
/// arrives wanted, even though everything is revealed.
|
||||
#[tokio::test]
|
||||
async fn untracked_series_reveals_without_intent() {
|
||||
let (_dir, database) = seeded_series(false).await;
|
||||
let server = tmdb("Returning Series", season_one_body(&two_episodes())).await;
|
||||
|
||||
action(&server).tick(&database).await.unwrap();
|
||||
|
||||
let rows: Vec<(i64, bool)> =
|
||||
sqlx::query_as("SELECT number, tracked FROM seasons ORDER BY number")
|
||||
.fetch_all(database.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(rows, vec![(1, false), (2, false)]);
|
||||
let wanted: i64 = sqlx::query_scalar("SELECT count(*) FROM episodes WHERE wanted = 1")
|
||||
.fetch_one(database.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(wanted, 0);
|
||||
}
|
||||
|
||||
/// The refresh has its own TTL: a stamped series costs no TMDB calls,
|
||||
/// whatever else is due.
|
||||
#[tokio::test]
|
||||
async fn refresh_is_throttled_by_its_own_ttl() {
|
||||
let (_dir, database) = seeded_series(true).await;
|
||||
let server = tmdb("Returning Series", season_one_body(&two_episodes())).await;
|
||||
|
||||
action(&server).tick(&database).await.unwrap();
|
||||
let after_first = server.received_requests().await.unwrap().len();
|
||||
|
||||
let second = action(&server).tick(&database).await.unwrap();
|
||||
assert!(second.is_empty());
|
||||
assert_eq!(
|
||||
server.received_requests().await.unwrap().len(),
|
||||
after_first,
|
||||
"the TTL gate must keep the tick off TMDB entirely"
|
||||
);
|
||||
}
|
||||
|
||||
/// A second run over unchanged TMDB data writes nothing and marks nothing
|
||||
/// wanted twice — refresh is idempotent.
|
||||
#[tokio::test]
|
||||
async fn a_second_run_over_unchanged_data_writes_nothing() {
|
||||
let (_dir, database) = seeded_series(true).await;
|
||||
let server = tmdb("Returning Series", season_one_body(&two_episodes())).await;
|
||||
action(&server).tick(&database).await.unwrap();
|
||||
expire_refresh(&database).await;
|
||||
|
||||
let outcomes = action(&server).tick(&database).await.unwrap();
|
||||
|
||||
assert!(
|
||||
outcomes.is_empty(),
|
||||
"unchanged data must not report work done"
|
||||
);
|
||||
let rows: Vec<(i64, bool)> =
|
||||
sqlx::query_as("SELECT number, tracked FROM seasons ORDER BY number")
|
||||
.fetch_all(database.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(rows, vec![(1, true), (2, true)]);
|
||||
let counts: (i64, i64) =
|
||||
sqlx::query_as("SELECT count(*), coalesce(sum(wanted), 0) FROM episodes")
|
||||
.fetch_one(database.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(counts, (3, 3));
|
||||
}
|
||||
|
||||
/// Episodes revealed into an existing season follow that season's rule:
|
||||
/// wanted when tracked, not otherwise. Known episodes' air dates and
|
||||
/// titles follow TMDB; §4.2's `airing` window reads them.
|
||||
#[tokio::test]
|
||||
async fn revealed_episodes_follow_the_season_rule_and_details_follow_tmdb() {
|
||||
let (_dir, database) = seeded_series(false).await;
|
||||
let server = tmdb("Returning Series", season_one_body(&two_episodes())).await;
|
||||
action(&server).tick(&database).await.unwrap();
|
||||
// One tracked season, one not — both already revealed above.
|
||||
sqlx::query("UPDATE seasons SET tracked = 1 WHERE number = 1")
|
||||
.execute(database.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
expire_refresh(&database).await;
|
||||
server.reset().await;
|
||||
mount(
|
||||
&server,
|
||||
"Returning Series",
|
||||
season_one_body(&json!([
|
||||
{"episode_number": 1, "name": "Magic Xylophone!", "air_date": "2018-10-08"},
|
||||
{"episode_number": 2, "name": "Hospital", "air_date": "2018-10-02"},
|
||||
{"episode_number": 3, "name": "Bike", "air_date": null}
|
||||
])),
|
||||
)
|
||||
.await;
|
||||
|
||||
let outcomes = action(&server).tick(&database).await.unwrap();
|
||||
|
||||
assert_eq!(outcomes.len(), 1);
|
||||
let season_one: Vec<(i64, String, Option<String>, bool)> = sqlx::query_as(
|
||||
"SELECT e.number, e.title, e.air_date, e.wanted
|
||||
FROM episodes e JOIN seasons s ON s.id = e.season_id
|
||||
WHERE s.number = 1 ORDER BY e.number",
|
||||
)
|
||||
.fetch_all(database.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
season_one,
|
||||
vec![
|
||||
// Episodes 1 and 2 were revealed before tracking was turned
|
||||
// on, and refresh never grants intent implicitly; only the
|
||||
// newly revealed 3 follows the rule.
|
||||
(
|
||||
1,
|
||||
"Magic Xylophone!".to_owned(),
|
||||
Some("2018-10-08".to_owned()),
|
||||
false
|
||||
),
|
||||
(
|
||||
2,
|
||||
"Hospital".to_owned(),
|
||||
Some("2018-10-02".to_owned()),
|
||||
false
|
||||
),
|
||||
(3, "Bike".to_owned(), None, true),
|
||||
],
|
||||
"tracked season: new episode arrives wanted, changed details move"
|
||||
);
|
||||
let season_two: Vec<bool> = sqlx::query_as(
|
||||
"SELECT e.wanted FROM episodes e JOIN seasons s ON s.id = e.season_id
|
||||
WHERE s.number = 2",
|
||||
)
|
||||
.fetch_all(database.pool())
|
||||
.await
|
||||
.unwrap()
|
||||
.into_iter()
|
||||
.map(|(wanted,): (bool,)| wanted)
|
||||
.collect();
|
||||
assert_eq!(season_two, vec![false], "untracked season stays unwanted");
|
||||
}
|
||||
|
||||
/// §4.2 derives `ended` from `upstream_ended`, so TMDB's status must
|
||||
/// reach the column.
|
||||
#[tokio::test]
|
||||
async fn upstream_ended_follows_tmdb_status() {
|
||||
let (_dir, database) = seeded_series(true).await;
|
||||
let server = tmdb("Returning Series", season_one_body(&two_episodes())).await;
|
||||
action(&server).tick(&database).await.unwrap();
|
||||
expire_refresh(&database).await;
|
||||
server.reset().await;
|
||||
mount(&server, "Canceled", season_one_body(&two_episodes())).await;
|
||||
|
||||
action(&server).tick(&database).await.unwrap();
|
||||
|
||||
let ended: bool =
|
||||
sqlx::query_scalar("SELECT upstream_ended FROM series WHERE tmdb_id = 82728")
|
||||
.fetch_one(database.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(ended);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user