diff --git a/crates/remux-server/src/addons/mod.rs b/crates/remux-server/src/addons/mod.rs index 0efb4ea9e..da87c1469 100644 --- a/crates/remux-server/src/addons/mod.rs +++ b/crates/remux-server/src/addons/mod.rs @@ -2753,47 +2753,29 @@ fn match_probe_version<'a>( }) } +const STREAMS_TTL_SECS: i64 = 60; + +fn streams_are_fresh( + refreshed_at: Option, + now: chrono::NaiveDateTime, +) -> bool { + refreshed_at.is_some_and(|refreshed_at| { + (now - refreshed_at).num_seconds() < STREAMS_TTL_SECS + }) +} + +fn can_refresh_cached_streams_in_background(total: i64, p2p: i64) -> bool { + total > 0 && total == p2p +} + impl AddonService { #[tracing::instrument(skip_all, fields(title = %media.title, kind = %media.kind))] - pub async fn refresh_streams( + async fn refresh_streams_inner( &self, media: &mut db::Media, ctx: &AppContext, user_id: Option, ) -> Result<()> { - const STREAMS_TTL_SECS: i64 = 60; - static STREAM_LOCKS: KeyedLock = KeyedLock::new(); - - // Fast path: TTL not expired — skip the lock entirely. - let is_fresh = |refreshed: Option| { - refreshed.is_some_and(|r| { - (chrono::Utc::now().naive_utc() - r).num_seconds() < STREAMS_TTL_SECS - }) - }; - if is_fresh(media.streams_refreshed_at) { - return Ok(()); - } - - // Acquire per-media lock to prevent concurrent refreshes. - let _guard = STREAM_LOCKS - .lock(media.id) - .await; - - // Re-check after acquiring lock — another task may have just refreshed. - let refreshed_at = sqlx::query_scalar::<_, Option>( - "SELECT streams_refreshed_at FROM media WHERE id = ?", - ) - .bind(media.id) - .fetch_optional(&ctx.db) - .await - .ok() - .flatten() - .flatten(); - if is_fresh(refreshed_at) { - media.streams_refreshed_at = refreshed_at; - return Ok(()); - } - media .grandparent(&ctx.db) .await @@ -2963,6 +2945,87 @@ impl AddonService { Ok(()) } + pub async fn refresh_streams( + &self, + media: &mut db::Media, + ctx: &AppContext, + user_id: Option, + ) -> Result<()> { + static STREAM_LOCKS: KeyedLock = KeyedLock::new(); + + let now = chrono::Utc::now().naive_utc(); + if streams_are_fresh(media.streams_refreshed_at, now) { + return Ok(()); + } + + // Torrent descriptors are durable, so stale cached torrent results can + // be served immediately while their provider data refreshes. HTTP + // streams (including debrid links) may expire and must be refreshed + // synchronously before playback can select them. + let cached_counts = sqlx::query_as::<_, (i64, i64)>( + "SELECT COUNT(*), COALESCE(SUM(CASE \ + WHEN json_type(stream_info, '$.descriptor.Torrent') IS NOT NULL \ + THEN 1 ELSE 0 END), 0) \ + FROM media WHERE parent_id = ? AND kind = 'stream'", + ) + .bind(media.id) + .fetch_one(&ctx.db) + .await + .unwrap_or((0, 0)); + if can_refresh_cached_streams_in_background(cached_counts.0, cached_counts.1) { + let service = self.clone(); + let mut media = media.clone(); + let ctx = ctx.clone(); + tokio::spawn(async move { + let _guard = STREAM_LOCKS + .lock(media.id) + .await; + let refreshed_at = + sqlx::query_scalar::<_, Option>( + "SELECT streams_refreshed_at FROM media WHERE id = ?", + ) + .bind(media.id) + .fetch_optional(&ctx.db) + .await + .ok() + .flatten() + .flatten(); + if streams_are_fresh(refreshed_at, chrono::Utc::now().naive_utc()) { + return; + } + if let Err(error) = service + .refresh_streams_inner(&mut media, &ctx, user_id) + .await + { + warn!(media_id = %media.id, %error, "background stream refresh failed"); + } + }); + return Ok(()); + } + + // Serialize cold loads and re-check after waiting because another + // request may have populated this item while the lock was held. + let _guard = STREAM_LOCKS + .lock(media.id) + .await; + let refreshed_at = sqlx::query_scalar::<_, Option>( + "SELECT streams_refreshed_at FROM media WHERE id = ?", + ) + .bind(media.id) + .fetch_optional(&ctx.db) + .await + .ok() + .flatten() + .flatten(); + if streams_are_fresh(refreshed_at, chrono::Utc::now().naive_utc()) { + media.streams_refreshed_at = refreshed_at; + return Ok(()); + } + + self.refresh_streams_inner(media, ctx, user_id) + .await + } + pub async fn fetch_segments( &self, media: &db::Media, @@ -3178,6 +3241,28 @@ pub fn make_media_id(addon_id: Uuid, local_id: &str) -> String { mod tests { use super::*; + #[test] + fn stream_freshness_uses_the_refresh_ttl() { + let now = chrono::Utc::now().naive_utc(); + assert!(streams_are_fresh( + Some(now - chrono::Duration::seconds(STREAMS_TTL_SECS - 1)), + now + )); + assert!(!streams_are_fresh( + Some(now - chrono::Duration::seconds(STREAMS_TTL_SECS)), + now + )); + assert!(!streams_are_fresh(None, now)); + } + + #[test] + fn only_torrent_only_caches_refresh_in_the_background() { + assert!(can_refresh_cached_streams_in_background(3, 3)); + assert!(!can_refresh_cached_streams_in_background(0, 0)); + assert!(!can_refresh_cached_streams_in_background(3, 2)); + assert!(!can_refresh_cached_streams_in_background(3, 0)); + } + fn torrent_stream(hash: &str, file_hint: &str, file_idx: usize) -> db::Media { db::Media { stream_info: Some(crate::stream::StreamInfo {