diff --git a/CHANGELOG.md b/CHANGELOG.md index 92815a026..4124d97b6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -320,6 +320,10 @@ ## ๐Ÿ› Fixes +- **Playlist Cache Load Failures No Longer Silent**: Xtream and M3U storage loads that fail due to corruption, + version mismatch, or task panics now log an error before falling back to an empty playlist, instead of silently + serving empty data. A genuinely missing storage file (first run) is logged at debug only, so normal startup stays + quiet. - **EPG Output Selection In Mixed Targets**: Fixed ambiguous EPG file selection when a target exposes both Xtream and M3U outputs. - Web UI playlist EPG and stream EPG APIs now explicitly prefer M3U EPG data and fall back to Xtream when M3U EPG is unavailable. - Xtream short-EPG now explicitly resolves Xtream EPG data. @@ -438,6 +442,84 @@ - `allowed_countries`: ISO-style country codes resolved through GeoIP. - The rules use OR semantics: any matching CIDR or country allows the request. +## ๐Ÿš€ Performance + +- **EPG Programme Lookup**: `EpgChannel::get_programme_with_limit` now binary-searches the sorted programme list + for the current window boundary instead of scanning linearly, reducing per-channel EPG lookup from `O(p)` to + `O(log p)` on large guides. The selected programmes are unchanged. +- **Item Type Field Access**: Reading the `Type` field during sorting and filtering no longer re-interns the + type label on every access. The five fixed type labels are now interned once and cached, so `Type`-keyed + rules return a cheap `Arc` clone instead of performing an interner hash-map lookup per item. +- **EPG XML Parsing**: The Web UI XMLTV parser no longer allocates an owned `String` for every XML element. It now + borrows the element name (via `from_utf8_lossy` without `to_string`) and tracks the active text element with a + small enum, removing one to two heap allocations per element on large EPG guides. +- **VOD/Recording Download Loop**: The per-chunk download-control check is now throttled to at most every 200ms + instead of running on every chunk. Pause/cancel/restart still take effect immediately via the existing + `control_notify` wakeups in the `select!` loop; the throttled poll only covers the rare race where a control + change fires mid-write, removing an unthrottled per-chunk lock read from multi-GB downloads. +- **Playlist Filtering**: `apply_filter_to_playlist` now pre-sizes each surviving-channel buffer to the source + group length instead of growing from a fixed small capacity, removing repeated reallocations while collecting + survivors for permissive filters (the common case where most channels pass). +- **Remote File Download**: `get_remote_content_as_file` no longer drains the response stream twice. It previously + consumed the whole body in an initial loop with no idle protection and then re-entered an idle-timeout `select!` + loop on an already-exhausted stream (dead code). The redundant first loop is removed, so the download runs through + a single loop and the idle timeout now actually guards the transfer. +- **Client Stream Keep-Alive Handling**: After a run of empty keep-alive chunks, `ClientStream` now backs off with a + short timer instead of immediately re-waking the task. A provider that streams an endless sequence of empty chunks + can no longer keep the task hot with back-to-back `Poll::Pending` re-wakes. +- **Library Orphan Cleanup**: `cleanup_orphaned` no longer reloads the entire metadata library for every orphaned + entry (and again for each thumbnail reference check). It now reads the library once, partitions entries into + orphaned and surviving, and deletes unreferenced thumbnails against a precomputed surviving-thumbnail set, turning + an O(orphans ร— library) scan into a single linear pass. +- **Series Expansion Merge**: Applying resolved series groups back into the in-memory playlist no longer re-scans + (and re-normalizes the title of) every existing group for each incoming group. A new batched `merge_groups`/ + `extend_playlist` path builds `(cluster, normalized_title)` and `(cluster, id)` indexes once and merges the whole + batch in a single linear pass, replacing the previous O(groupsยฒ) behavior on series-heavy inputs while moving + channels instead of cloning them. +- **Group Flatten Pre-Sizing**: `flatten_groups` now pre-allocates its merge buffer and category index to the + incoming group count instead of growing them from empty, reducing reallocation churn when assembling large target + playlists. +- **Custom Provider Playlist Streaming**: The custom provider playlist endpoint now streams its UI items lazily via + `stream_json_or_bin_response_stream` (like the target and input endpoints) instead of collecting every channel into + a second `Vec` and serializing the whole body at once, lowering peak memory and latency for large + provider imports. + +## ๐Ÿงน Maintainability + +- **Output Capability Descriptor**: Introduced `TargetType::capabilities()` โ€” a single exhaustive table describing + whether each output format supports playlist filtering, EPG, and the in-memory cache. The EPG write/path sites now + derive "this format has no EPG" from the table instead of silent empty match arms, and a new `TargetOutput::filter()` + accessor co-locates the per-format filter lookup so the playlist writer no longer re-matches every variant. Adding an + output format now centers on one descriptor entry plus the format-specific writer. +- **Input Capability Descriptor**: Introduced `InputType::capabilities()` โ€” a single exhaustive table describing each + input type's persistence family (`InputPersistence`), whether generic probing needs a provider connection, and + whether the custom-provider endpoint can serve it. Backend persist/load routing, the stream-probe requirement check, + and the custom-provider endpoint now derive from this table instead of re-listing `InputType` variants in parallel + `match` arms, so most new input variants only need a single descriptor entry and the dependent sites stay in sync. +- **ItemField Single-Source Binding**: The directly-bound `ItemField` variants (`Group`/`Name`/`Title`/`Url`/`Input`) + are now listed once and both `get_field_value` and `set_field_value` are generated from that single list via a + callback macro, so a new simple field can no longer be wired into the read half but forgotten in the write half (or + vice versa). The asymmetric `Genre`/`Type`/`Caption` cases stay explicit because their read/write behavior differs. +- **Resolver Stage Summaries**: The background VOD and series resolve-queueing loops now emit a single concise + debug summary (total / queued / expanded / skipped-by-filter / skipped-by-resolve-filter / elapsed) per stage in + addition to the existing detailed per-item logs, making operational analysis of the hot resolver paths easier + without raising the default log level. +- **InputType Family Helpers**: Added `InputType::is_batch()` and routed the scattered + `matches!(input_type, M3uBatch | XtreamBatch)` / `Xtream | XtreamBatch` / `M3u | M3uBatch` checks across the backend + and frontend through the centralized `is_batch()` / `is_xtream()` / `is_m3u()` methods, so the batch/family rules + live on the enum instead of being re-encoded (and risking silent drift) at every call site. +- **PlaylistItemType Classification Helpers**: Added `PlaylistItemType::is_video()` / `is_series()` and routed the + repeated `matches!(item_type, Video | LocalVideo)` and `matches!(item_type, Series | LocalSeries)` boolean checks + (filter evaluation, stream probing, trakt matching, playlist persistence, custom-provider filtering, metadata probe + routing) through them, so the family classification is defined once on the enum instead of re-encoded per call site. +- **XtreamCluster Behavior Centralization**: Added `XtreamCluster::info_action_and_id_field()` (and a backend + `xtream_cluster_category_collection()` helper) so the per-cluster `player_api` action / id-field and the + `cat_live`/`cat_vod`/`cat_series` collection names are derived once instead of being re-matched at each storage and + network call site, reducing the chance of a forgotten or inconsistent cluster arm. +- **Derived GeoIP Error Display**: `GeoIpUpdateError` now derives its `Display`/`Error` implementation via `thiserror` + instead of a hand-written `match self` formatter, so adding a variant only requires an `#[error(...)]` attribute next + to it rather than remembering to extend a separate formatting block (a classic source of stale or missing messages). + ## 3.3.0 (2026-04-02) ## โš ๏ธ Breaking Changes 3.3.0 diff --git a/Cargo.lock b/Cargo.lock index 714fc4f12..b3d38cd1c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4671,6 +4671,7 @@ dependencies = [ "strum_macros", "sysinfo", "tempfile", + "thiserror 2.0.18", "tokio", "tokio-rustls", "tokio-stream", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index d29f93f06..1a6307ad9 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -32,6 +32,7 @@ futures.workspace = true path-clean.workspace = true pest.workspace = true pest_derive.workspace = true +thiserror.workspace = true aes = "0.8.4" ctr = "0.9.2" deunicode.workspace = true diff --git a/backend/src/api/endpoints/api_playlist_utils.rs b/backend/src/api/endpoints/api_playlist_utils.rs index 4f47917d4..5d1e2fe46 100644 --- a/backend/src/api/endpoints/api_playlist_utils.rs +++ b/backend/src/api/endpoints/api_playlist_utils.rs @@ -1,6 +1,6 @@ use crate::api::model::AppState; use crate::{ - api::api_utils::{empty_json_list_response, json_or_bin_response, stream_json_or_bin_response_stream}, + api::api_utils::{empty_json_list_response, stream_json_or_bin_response_stream}, model::{ConfigInput, ConfigTarget}, repository::{ iter_raw_m3u_input_playlist, iter_raw_m3u_target_playlist, iter_raw_xtream_input_playlist, @@ -12,11 +12,8 @@ use axum::response::IntoResponse; use log::warn; use serde_json::json; use shared::utils::{concat_path, concat_path_leading_slash, obfuscate_text, Internable}; -use shared::{ - model::{ - InputType, M3uPlaylistItem, PlaylistItemType, TargetType, UiPlaylistItem, XtreamCluster, XtreamPlaylistItem, - }, - utils::interner_gc, +use shared::model::{ + InputPersistence, M3uPlaylistItem, TargetType, UiPlaylistItem, XtreamCluster, XtreamPlaylistItem, }; use std::sync::Arc; use tokio_stream::StreamExt; @@ -42,7 +39,7 @@ pub(in crate::api::endpoints) async fn get_playlist_for_target( }; let item_filter = if cluster == XtreamCluster::Series { |pli: &XtreamPlaylistItem| { - !matches!(pli.item_type, PlaylistItemType::Series | PlaylistItemType::LocalSeries) + !pli.item_type.is_series() } } else { |_pli: &XtreamPlaylistItem| true @@ -55,7 +52,7 @@ pub(in crate::api::endpoints) async fn get_playlist_for_target( }; let item_filter = if cluster == XtreamCluster::Series { |pli: &M3uPlaylistItem| { - !matches!(pli.item_type, PlaylistItemType::Series | PlaylistItemType::LocalSeries) + !pli.item_type.is_series() } } else { |_pli: &M3uPlaylistItem| true @@ -99,13 +96,13 @@ pub(in crate::api::endpoints) async fn get_playlist_for_input( accept: Option<&str>, ) -> impl IntoResponse + Send { if let Some(input) = cfg_input { - if matches!(input.input_type, InputType::Xtream | InputType::XtreamBatch) { + if input.input_type.is_xtream() { let Some(channel_iterator) = iter_raw_xtream_input_playlist(&app_state.app_config, input, cluster).await else { return empty_json_list_response(); }; let converted_stream = channel_iterator.map(UiPlaylistItem::from); return stream_json_or_bin_response_stream(accept, converted_stream).into_response(); - } else if matches!(input.input_type, InputType::M3u | InputType::M3uBatch) { + } else if input.input_type.is_m3u() { let Some(channels) = iter_raw_m3u_input_playlist(&app_state.app_config, input, Some(cluster)).await else { return empty_json_list_response(); }; @@ -132,23 +129,23 @@ pub(in crate::api::endpoints) async fn get_playlist_for_custom_provider( let cfg = app_state.app_config.config.load(); match cfg_input { Some(input) => { - let (result, errors) = match input.get_download_input_type() { - InputType::M3u | InputType::M3uBatch => { + let (result, errors) = match input.get_download_input_type().persistence() { + InputPersistence::M3u => { m3u::download_m3u_playlist(&app_state.app_config, client, &cfg, input).await } - InputType::Xtream | InputType::XtreamBatch => { + InputPersistence::Xtream => { let (pl, err, _) = xtream::download_xtream_playlist(&app_state.app_config, client, input, Some(&[cluster])).await; (pl, err) } - InputType::Library => { + InputPersistence::Library => { return ( axum::http::StatusCode::BAD_REQUEST, axum::Json(json!({ "error": "Library inputs are not supported on this endpoint"})), ) .into_response(); } - InputType::Emby | InputType::Jellyfin | InputType::Plex => { + InputPersistence::MediaServer => { return ( axum::http::StatusCode::BAD_REQUEST, axum::Json(json!({ "error": "Media-server inputs are not supported on this endpoint yet"})), @@ -161,10 +158,12 @@ pub(in crate::api::endpoints) async fn get_playlist_for_custom_provider( (axum::http::StatusCode::BAD_REQUEST, axum::Json(json!({"error": error_strings.join(", ")}))) .into_response() } else { - let channels: Vec = - result.iter().flat_map(|g| g.channels.iter()).map(UiPlaylistItem::from).collect(); - interner_gc(); - json_or_bin_response(accept, &channels).into_response() + // Stream the UI conversion lazily (like the target/input endpoints) instead of + // collecting the whole playlist into a second Vec and serializing it all at once. + let converted_stream = tokio_stream::iter( + result.into_iter().flat_map(|g| g.channels).map(|pli| UiPlaylistItem::from(&pli)), + ); + stream_json_or_bin_response_stream(accept, converted_stream).into_response() } } None => { diff --git a/backend/src/api/endpoints/download_api.rs b/backend/src/api/endpoints/download_api.rs index 3ebfb6cd7..414efb2f9 100644 --- a/backend/src/api/endpoints/download_api.rs +++ b/backend/src/api/endpoints/download_api.rs @@ -27,6 +27,11 @@ const DOWNLOAD_PROGRESS_LOG_INTERVAL: Duration = Duration::from_secs(5); const DOWNLOAD_PROGRESS_LOG_BYTES: u64 = 16 * 1024 * 1024; const DOWNLOAD_SNAPSHOT_UPDATE_INTERVAL: Duration = Duration::from_secs(2); const DOWNLOAD_SNAPSHOT_UPDATE_BYTES: u64 = 4 * 1024 * 1024; +// Pause/cancel/restart are delivered immediately via `control_notify` while the +// worker is parked in the `select!`. This poll is only a fallback for the rare +// race where a control change fires while a chunk is being written (notify is not +// persisted), so it does not need to run on every chunk. +const DOWNLOAD_CONTROL_POLL_INTERVAL: Duration = Duration::from_millis(200); const RECORDING_PROGRESS_UPDATE_INTERVAL: Duration = Duration::from_secs(5); type ProviderCapacities = Vec<(Arc, usize, usize)>; @@ -386,18 +391,15 @@ async fn download_file( let mut last_progress_logged_bytes = downloaded; let mut last_snapshot_update_at = Instant::now(); let mut last_snapshot_update_bytes = downloaded; + let mut control_poll = time::interval(DOWNLOAD_CONTROL_POLL_INTERVAL); + control_poll.set_missed_tick_behavior(time::MissedTickBehavior::Skip); + control_poll.tick().await; let deadline_at = recording_deadline_instant(&file_download); let mut deadline_sleep = deadline_at.map(|deadline| { Box::pin(time::sleep_until(deadline)) as Pin> }); loop { - if let Some(result) = - handle_download_control(&active, current_download_control(&control_signal), &mut buf_writer).await - { - return result; - } - if deadline_at.is_some_and(|deadline| Instant::now() >= deadline) { if let Some(lock) = active.write().await.as_mut() { lock.paused = false; @@ -430,6 +432,16 @@ async fn download_file( } continue; } + _ = control_poll.tick() => { + if let Some(result) = handle_download_control( + &active, + current_download_control(&control_signal), + &mut buf_writer, + ).await { + return result; + } + continue; + } () = deadline_sleep.as_mut() => return DownloadExecutionResult::Completed, next_item = stream.try_next() => next_item.map_err(to_io_error), } @@ -447,6 +459,16 @@ async fn download_file( } continue; } + _ = control_poll.tick() => { + if let Some(result) = handle_download_control( + &active, + current_download_control(&control_signal), + &mut buf_writer, + ).await { + return result; + } + continue; + } next_item = stream.try_next() => next_item.map_err(to_io_error), } } else if let Some(deadline_sleep) = deadline_sleep.as_mut() { @@ -462,6 +484,16 @@ async fn download_file( } continue; } + _ = control_poll.tick() => { + if let Some(result) = handle_download_control( + &active, + current_download_control(&control_signal), + &mut buf_writer, + ).await { + return result; + } + continue; + } () = deadline_sleep.as_mut() => return DownloadExecutionResult::Completed, next_item = stream.try_next() => next_item.map_err(to_io_error), } @@ -478,6 +510,16 @@ async fn download_file( } continue; } + _ = control_poll.tick() => { + if let Some(result) = handle_download_control( + &active, + current_download_control(&control_signal), + &mut buf_writer, + ).await { + return result; + } + continue; + } next_item = stream.try_next() => next_item.map_err(to_io_error), } }; diff --git a/backend/src/api/endpoints/extract_accept_header.rs b/backend/src/api/endpoints/extract_accept_header.rs index dcee01b21..ef0c99fce 100644 --- a/backend/src/api/endpoints/extract_accept_header.rs +++ b/backend/src/api/endpoints/extract_accept_header.rs @@ -12,12 +12,15 @@ where { type Rejection = (StatusCode, &'static str); - async fn from_request_parts(parts: &mut Parts, _state: &B) -> Result { + fn from_request_parts( + parts: &mut Parts, + _state: &B, + ) -> impl std::future::Future> + Send { if let Some(accept_type) = parts.headers.get(axum::http::header::ACCEPT) { if let Ok(val) = accept_type.to_str() { - return Ok(ExtractAcceptHeader(Some(val.to_string()))); + return std::future::ready(Ok(ExtractAcceptHeader(Some(val.to_string())))); } } - Ok(ExtractAcceptHeader(None)) + std::future::ready(Ok(ExtractAcceptHeader(None))) } } diff --git a/backend/src/api/endpoints/v1_api_playlist.rs b/backend/src/api/endpoints/v1_api_playlist.rs index 7d1682f48..7069a7f9a 100644 --- a/backend/src/api/endpoints/v1_api_playlist.rs +++ b/backend/src/api/endpoints/v1_api_playlist.rs @@ -341,7 +341,7 @@ async fn playlist_series_info( PlaylistRequest::Input(input_name) => { if let Some(input) = app_state.app_config.get_input_by_name(&input_name.intern()) { - if matches!(input.input_type, InputType::Xtream | InputType::XtreamBatch) { + if input.input_type.is_xtream() { // We cannot call `xtream_get_stream_info_response` directly here because that path // depends on target-local virtual-id mapping (`xtream_get_item_for_stream_id`). // Input/custom requests only provide provider_id, so we resolve series info from diff --git a/backend/src/api/endpoints/xmltv_api.rs b/backend/src/api/endpoints/xmltv_api.rs index 29524e1bb..04f58848b 100644 --- a/backend/src/api/endpoints/xmltv_api.rs +++ b/backend/src/api/endpoints/xmltv_api.rs @@ -76,6 +76,9 @@ pub(in crate::api) fn get_epg_path_for_target_by_type( target: &ConfigTarget, output_type: TargetType, ) -> Option { + if !output_type.supports_epg() { + return None; + } match output_type { TargetType::Xtream => { if let Some(storage_path) = xtream_get_storage_path(config, &target.name) { diff --git a/backend/src/api/model/metadata_update_manager.rs b/backend/src/api/model/metadata_update_manager.rs index b97673cd7..6793e55bb 100644 --- a/backend/src/api/model/metadata_update_manager.rs +++ b/backend/src/api/model/metadata_update_manager.rs @@ -3209,7 +3209,7 @@ impl InputWorker { fn batchable_generic_probe_xtream_cluster(item_type: PlaylistItemType) -> Option { if item_type.is_live() { Some(XtreamCluster::Live) - } else if matches!(item_type, PlaylistItemType::Video | PlaylistItemType::LocalVideo) { + } else if item_type.is_video() { Some(XtreamCluster::Video) } else { None @@ -3402,7 +3402,7 @@ impl InputWorker { let task_key = TaskKey::from_task(task); let probe_identifier = if unique_id.trim().is_empty() { url.as_str() } else { unique_id.as_str() }; - if matches!(input.input_type, InputType::Xtream | InputType::XtreamBatch) { + if input.input_type.is_xtream() { if let Some(cluster) = Self::batchable_generic_probe_xtream_cluster(*item_type) { let Ok(provider_id) = unique_id.parse::() else { warn!("Skipping xtream generic probe update with non-numeric id: {unique_id}"); diff --git a/backend/src/api/model/streams/client_stream.rs b/backend/src/api/model/streams/client_stream.rs index a90c8b3dc..98d6b4ed6 100644 --- a/backend/src/api/model/streams/client_stream.rs +++ b/backend/src/api/model/streams/client_stream.rs @@ -17,6 +17,13 @@ use std::{ }; use tokio_util::sync::{CancellationToken, WaitForCancellationFutureOwned}; +/// Number of consecutive empty keep-alive chunks tolerated in a single poll +/// before backing off, to avoid spinning on a misbehaving provider. +const EMPTY_CHUNK_SKIP_LIMIT: u32 = 10; +/// Cool-down applied after exceeding [`EMPTY_CHUNK_SKIP_LIMIT`] so an endless +/// run of empty chunks cannot keep the task hot via immediate re-wakes. +const EMPTY_CHUNK_BACKOFF: std::time::Duration = std::time::Duration::from_millis(5); + /// This stream counts the send bytes for reconnecting to the actual position and /// sets the `close_signal` if the client drops the connection. pub(in crate::api::model) struct ClientStream { @@ -24,6 +31,7 @@ pub(in crate::api::model) struct ClientStream { close_signal: CancellationToken, close_cancelled: Pin>, total_bytes: Option>, + empty_backoff: Option>>, url: String, } @@ -35,7 +43,7 @@ impl ClientStream { url: &str, ) -> Self { let close_cancelled = Box::pin(close_signal.clone().cancelled_owned()); - Self { inner, close_signal, close_cancelled, total_bytes, url: url.to_string() } + Self { inner, close_signal, close_cancelled, total_bytes, empty_backoff: None, url: url.to_string() } } } @@ -47,10 +55,18 @@ impl Stream for ClientStream { if this.close_cancelled.as_mut().poll(cx).is_ready() { return Poll::Ready(None); } + // If we are cooling down after a run of empty keep-alive chunks, wait for + // the timer to elapse before polling the provider again. + if let Some(backoff) = this.empty_backoff.as_mut() { + match backoff.as_mut().poll(cx) { + Poll::Ready(()) => this.empty_backoff = None, + Poll::Pending => return Poll::Pending, + } + } // Bound empty-chunk skips per poll invocation. A misbehaving provider // that sends an endless run of empty keep-alive chunks must not spin - // the executor indefinitely: after 10 consecutive empty chunks we - // yield back to the runtime via wake_by_ref + Poll::Pending. + // the executor indefinitely: after EMPTY_CHUNK_SKIP_LIMIT consecutive + // empty chunks we back off for a short cool-down before retrying. let mut empty_chunk_count = 0u32; loop { match Pin::new(&mut this.inner).poll_next(cx) { @@ -60,8 +76,11 @@ impl Stream for ClientStream { // Some providers send empty chunks as heartbeats; closing on them // would prematurely terminate valid streams. empty_chunk_count += 1; - if empty_chunk_count > 10 { - cx.waker().wake_by_ref(); + if empty_chunk_count > EMPTY_CHUNK_SKIP_LIMIT { + let mut backoff = Box::pin(tokio::time::sleep(EMPTY_CHUNK_BACKOFF)); + // Poll once so the timer registers our waker. + let _ = backoff.as_mut().poll(cx); + this.empty_backoff = Some(backoff); return Poll::Pending; } trace!("client stream: skipping empty keep-alive chunk"); diff --git a/backend/src/api/serve.rs b/backend/src/api/serve.rs index 7cf0b2ca6..173f6fe52 100644 --- a/backend/src/api/serve.rs +++ b/backend/src/api/serve.rs @@ -64,7 +64,7 @@ async fn handle_connection( remote_addr: SocketAddr, connection_manager: Arc, ) where - M: for<'a> Service + Send + 'static, + M: Service + Send + 'static, for<'a> >::Future: Send, S: Service + Clone + Send + 'static, S::Future: Send, diff --git a/backend/src/auth/auth_basic.rs b/backend/src/auth/auth_basic.rs index b7f23135f..79b06c7cd 100644 --- a/backend/src/auth/auth_basic.rs +++ b/backend/src/auth/auth_basic.rs @@ -14,8 +14,11 @@ where { type Rejection = Rejection; - async fn from_request_parts(req: &mut Parts, _: &B) -> Result { - Self::decode_request_parts(req) + fn from_request_parts( + req: &mut Parts, + _: &B, + ) -> impl std::future::Future> + Send { + std::future::ready(Self::decode_request_parts(req)) } } @@ -56,4 +59,4 @@ fn decode(input: &str) -> Result<(String, String), Rejection> { } else { Err((StatusCode::FORBIDDEN, "Authorization header contains no password")) } -} \ No newline at end of file +} diff --git a/backend/src/auth/auth_bearer.rs b/backend/src/auth/auth_bearer.rs index 4564bf10c..e6f16ec96 100644 --- a/backend/src/auth/auth_bearer.rs +++ b/backend/src/auth/auth_bearer.rs @@ -13,8 +13,11 @@ where { type Rejection = Rejection; - async fn from_request_parts(req: &mut Parts, _: &B) -> Result { - Self::decode_request_parts(req) + fn from_request_parts( + req: &mut Parts, + _: &B, + ) -> impl std::future::Future> + Send { + std::future::ready(Self::decode_request_parts(req)) } } diff --git a/backend/src/library/metadata_storage.rs b/backend/src/library/metadata_storage.rs index fc400a884..beed09235 100644 --- a/backend/src/library/metadata_storage.rs +++ b/backend/src/library/metadata_storage.rs @@ -1,7 +1,7 @@ use crate::library::metadata::{MediaMetadata, MetadataCacheEntry}; use log::{debug, error, info}; use path_clean::PathClean; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::fmt::Write; use std::path::PathBuf; use std::sync::Arc; @@ -165,18 +165,62 @@ impl MetadataStorage { // Cleans up metadata for files that no longer exist pub async fn cleanup_orphaned(&self) -> std::io::Result { + let _guard = self.mutation_guard.lock().await; let entries = self.load_all().await; - let mut deleted_count = 0; + // Partition into orphaned (source file missing) and surviving entries in a + // single pass so we read the library exactly once for the whole cleanup. + let mut orphaned = Vec::new(); + let mut surviving = Vec::new(); for entry in entries { - if !fs::try_exists(&entry.file_path).await.unwrap_or(false) { - info!("Removing orphaned metadata for missing file: {}", entry.file_path); - if let Err(e) = self.delete_by_uuid(&entry.uuid).await { + if fs::try_exists(&entry.file_path).await.unwrap_or(false) { + surviving.push(entry); + } else { + orphaned.push(entry); + } + } + + if orphaned.is_empty() { + return Ok(0); + } + + // Thumbnails referenced by any surviving entry must be kept. Compute the + // set once instead of reloading the library for every orphaned entry. + let mut surviving_thumbnails: HashSet = surviving + .iter() + .flat_map(referenced_thumbnail_ids_for_entry) + .collect(); + + let mut deleted_count = 0; + let mut candidate_thumbnails: HashSet = HashSet::new(); + for entry in &orphaned { + info!("Removing orphaned metadata for missing file: {}", entry.file_path); + let file_path = self.get_library_metadata_file_path(&entry.uuid); + match fs::remove_file(&file_path).await { + Ok(()) => deleted_count += 1, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} + Err(e) => { error!("Failed to delete orphaned metadata: {e}"); - } else { - deleted_count += 1; + surviving_thumbnails.extend(referenced_thumbnail_ids_for_entry(entry)); + continue; } } + for thumbnail_id in referenced_thumbnail_ids_for_entry(entry) { + candidate_thumbnails.insert(thumbnail_id); + } + } + + // Delete thumbnails that the removed entries referenced and that no + // surviving entry still references. + for thumbnail_id in candidate_thumbnails { + if surviving_thumbnails.contains(&thumbnail_id) { + continue; + } + let path = self.get_thumbnail_path(&thumbnail_id); + if fs::try_exists(&path).await.unwrap_or(false) { + debug!("Deleting thumbnail file: {}", path.display()); + let _ = fs::remove_file(path).await; + } } if deleted_count > 0 { diff --git a/backend/src/media_server/catalog.rs b/backend/src/media_server/catalog.rs index ec26d645a..76f7ffec5 100644 --- a/backend/src/media_server/catalog.rs +++ b/backend/src/media_server/catalog.rs @@ -215,6 +215,7 @@ mod tests { } } + #[allow(clippy::unused_async_trait_impl)] impl MediaServerCatalogClient for MockMediaServerCatalogClient { async fn discover(&self) -> Result { Ok(MediaServerStatus { diff --git a/backend/src/media_server/playback.rs b/backend/src/media_server/playback.rs index 0b59d2e81..b1bcb25ed 100644 --- a/backend/src/media_server/playback.rs +++ b/backend/src/media_server/playback.rs @@ -258,6 +258,7 @@ mod tests { stream_error: Option, } + #[allow(clippy::unused_async_trait_impl)] impl MediaServerCatalogClient for MockPlaybackClient { async fn discover(&self) -> Result { unreachable!() } async fn list_libraries(&self) -> Result, MediaServerError> { unreachable!() } diff --git a/backend/src/model/config/input.rs b/backend/src/model/config/input.rs index 5823f102f..eccf835e8 100644 --- a/backend/src/model/config/input.rs +++ b/backend/src/model/config/input.rs @@ -602,7 +602,7 @@ impl ConfigInput { } // For batch definitions, validate root URL/credentials before alias promotion in prepare_batch(). - if self.enabled && matches!(self.input_type, InputType::M3uBatch | InputType::XtreamBatch) { + if self.enabled && self.input_type.is_batch() { check_input_credentials!(self, self.input_type, false, false); check_input_connections!(self, self.input_type, false); } @@ -661,7 +661,7 @@ impl ConfigInput { } fn prepare_batch(&mut self) -> Option { - if matches!(self.input_type, InputType::M3uBatch | InputType::XtreamBatch) { + if self.input_type.is_batch() { let input_type = if self.input_type == InputType::M3uBatch { InputType::M3u } else { diff --git a/backend/src/model/config/target.rs b/backend/src/model/config/target.rs index 7d6ff42c1..9d9eeb0df 100644 --- a/backend/src/model/config/target.rs +++ b/backend/src/model/config/target.rs @@ -192,6 +192,25 @@ impl From<&TargetOutput> for TargetType { } } +impl TargetOutput { + /// Output format of this configured target output. + #[must_use] + pub fn target_type(&self) -> TargetType { TargetType::from(self) } + + /// The optional playlist filter configured for this output, co-located so + /// the per-format filter access lives in one place rather than being + /// re-matched at every call site. + #[must_use] + pub fn filter(&self) -> Option<&Filter> { + match self { + TargetOutput::Xtream(out) => out.filter.as_ref(), + TargetOutput::M3u(out) => out.filter.as_ref(), + TargetOutput::Strm(out) => out.filter.as_ref(), + TargetOutput::HdHomeRun(_) => None, + } + } +} + macros::from_impl!(TargetOutput); impl From<&TargetOutputDto> for TargetOutput { fn from(dto: &TargetOutputDto) -> Self { diff --git a/backend/src/model/playlist.rs b/backend/src/model/playlist.rs index 8ad60e927..e8aeebd24 100644 --- a/backend/src/model/playlist.rs +++ b/backend/src/model/playlist.rs @@ -23,6 +23,15 @@ impl FetchedPlaylist<'_> { } } + /// Merge a batch of groups in one indexed pass instead of calling + /// [`Self::update_playlist`] once per group (which re-scans the whole + /// in-memory playlist each time). Only affects in-memory sources. + pub fn extend_playlist(&mut self, groups: Vec) { + if self.source.is_memory() { + self.source.extend_playlist(groups); + } + } + pub fn is_memory(&self) -> bool { self.source.is_memory() } diff --git a/backend/src/model/xmltv.rs b/backend/src/model/xmltv.rs index 53e16515a..cf7d4fbbf 100644 --- a/backend/src/model/xmltv.rs +++ b/backend/src/model/xmltv.rs @@ -23,6 +23,12 @@ pub const EPG_ATTRIB_ID: &str = "id"; pub const EPG_ATTRIB_CHANNEL: &str = "channel"; pub const EPG_TAG_DISPLAY_NAME: &str = "display-name"; pub const EPG_TAG_ICON: &str = "icon"; +pub const EPG_TAG_TITLE: &str = "title"; +pub const EPG_TAG_DESC: &str = "desc"; +pub const EPG_ATTRIB_START: &str = "start"; +pub const EPG_ATTRIB_STOP: &str = "stop"; +pub const EPG_ATTRIB_CATCHUP_ID: &str = "catchup-id"; +pub const EPG_ATTRIB_SRC: &str = "src"; // https://github.com/XMLTV/xmltv/blob/master/xmltv.dtd @@ -171,6 +177,16 @@ pub fn get_attr_value(attr: &quick_xml::events::attributes::Attribute) -> Option // This function filters a timeslot starting from yesterday. #[allow(clippy::too_many_lines)] async fn parse_xmltv_for_web_ui(reader: R) -> Result, TuliproxError> { + // Tracks which text-bearing element we are currently inside, without allocating + // a String per XML event just to compare against a few known tag names. + #[derive(Clone, Copy, PartialEq)] + enum TextTag { + DisplayName, + Title, + Desc, + Other, + } + let mut reader = quick_xml::reader::Reader::from_reader(async_file_reader(reader)); let mut buf = Vec::new(); @@ -180,7 +196,7 @@ async fn parse_xmltv_for_web_ui(reader: R) -> Resul let mut current_channel: Option = None; let mut current_programme: Option = None; - let mut current_tag = String::new(); + let mut current_tag = TextTag::Other; // only 1 day old epg let now = Utc::now(); @@ -192,14 +208,22 @@ async fn parse_xmltv_for_web_ui(reader: R) -> Resul loop { match reader.read_event_into_async(&mut buf).await { Ok(Event::Empty(e) | Event::Start(e)) => { - let tag = String::from_utf8_lossy(e.name().as_ref()).to_string(); - current_tag.clone_from(&tag); + let name = e.name(); + // `from_utf8_lossy` borrows for valid UTF-8 (the common case), so no + // allocation happens per element here. + let tag = String::from_utf8_lossy(name.as_ref()); + current_tag = match tag.as_ref() { + EPG_TAG_DISPLAY_NAME => TextTag::DisplayName, + EPG_TAG_TITLE => TextTag::Title, + EPG_TAG_DESC => TextTag::Desc, + _ => TextTag::Other, + }; - match tag.as_str() { + match tag.as_ref() { EPG_TAG_CHANNEL => { let mut id = None; for attr in e.attributes().flatten() { - if attr.key.as_ref() == b"id" { + if attr.key.as_ref() == EPG_ATTRIB_ID.as_bytes() { if let Some(value) = get_attr_value(&attr) { id = Some(value); break; @@ -219,12 +243,15 @@ async fn parse_xmltv_for_web_ui(reader: R) -> Resul let mut catchup_id = None; current_programme = None; for attr in e.attributes().flatten() { - match attr.key.as_ref() { - b"start" => start = get_attr_value(&attr), - b"stop" => stop = get_attr_value(&attr), - b"channel" => channel = get_attr_value(&attr), - b"catchup-id" => catchup_id = get_attr_value(&attr), - _ => {} + let key = attr.key.as_ref(); + if key == EPG_ATTRIB_START.as_bytes() { + start = get_attr_value(&attr); + } else if key == EPG_ATTRIB_STOP.as_bytes() { + stop = get_attr_value(&attr); + } else if key == EPG_ATTRIB_CHANNEL.as_bytes() { + channel = get_attr_value(&attr); + } else if key == EPG_ATTRIB_CATCHUP_ID.as_bytes() { + catchup_id = get_attr_value(&attr); } } if let (Some(pstart), Some(pstop), Some(pchannel)) = (start, stop, channel) { @@ -240,7 +267,7 @@ async fn parse_xmltv_for_web_ui(reader: R) -> Resul EPG_TAG_ICON => { if let Some(channel) = &mut current_channel { for attr in e.attributes().flatten() { - if attr.key.as_ref() == b"src" { + if attr.key.as_ref() == EPG_ATTRIB_SRC.as_bytes() { if let Some(icon) = get_attr_value(&attr) { if !icon.is_empty() { channel.icon = Some(icon); @@ -258,15 +285,15 @@ async fn parse_xmltv_for_web_ui(reader: R) -> Resul let text = decoded.trim(); if !text.is_empty() { if let Some(channel) = &mut current_channel { - if current_tag == EPG_TAG_DISPLAY_NAME { + if current_tag == TextTag::DisplayName { channel.title = Some(concat_text(channel.title.as_ref(), text)); } } if let Some(program) = &mut current_programme { - if current_tag == "title" { + if current_tag == TextTag::Title { program.title = Some(concat_text(program.title.as_ref(), text)); - } else if current_tag == "desc" { + } else if current_tag == TextTag::Desc { program.desc = Some(concat_text(program.desc.as_ref(), text)); } } @@ -274,8 +301,8 @@ async fn parse_xmltv_for_web_ui(reader: R) -> Resul } } Ok(Event::End(e)) => { - let tag = String::from_utf8_lossy(e.name().as_ref()).to_string(); - match tag.as_str() { + let name = e.name(); + match String::from_utf8_lossy(name.as_ref()).as_ref() { EPG_TAG_CHANNEL => { if let Some(channel) = current_channel.take() { channels.push(channel); @@ -288,7 +315,7 @@ async fn parse_xmltv_for_web_ui(reader: R) -> Resul } _ => {} } - current_tag.clear(); + current_tag = TextTag::Other; } Ok(Event::Eof) => break, Err(err) => return Err(TuliproxError::Parse(err.to_string())), diff --git a/backend/src/processing/geoip.rs b/backend/src/processing/geoip.rs index 4911791ce..101fbca10 100644 --- a/backend/src/processing/geoip.rs +++ b/backend/src/processing/geoip.rs @@ -11,27 +11,18 @@ use shared::{ }; use std::{collections::HashMap, io::Cursor, sync::Arc}; -#[derive(Debug)] +#[derive(Debug, thiserror::Error)] pub(crate) enum GeoIpUpdateError { + #[error("GeoIp update is disabled")] Disabled, + #[error("Failed to download geoip db: {0}")] DownloadFailed(String), + #[error("Failed to process geoip db: {0}")] ProcessFailed(String), + #[error("Unknown GeoIp processing error")] UnknownProcessing, } -impl std::fmt::Display for GeoIpUpdateError { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - Self::Disabled => write!(f, "GeoIp update is disabled"), - Self::DownloadFailed(err) => write!(f, "Failed to download geoip db: {err}"), - Self::ProcessFailed(err) => write!(f, "Failed to process geoip db: {err}"), - Self::UnknownProcessing => write!(f, "Unknown GeoIp processing error"), - } - } -} - -impl std::error::Error for GeoIpUpdateError {} - pub(crate) async fn update_geoip_db(app_state: &Arc) -> Result<(), GeoIpUpdateError> { let config = app_state.app_config.config.load(); if let Some(geoip) = config.reverse_proxy.as_ref().and_then(|r| r.geoip.as_ref()) { diff --git a/backend/src/processing/processor/playlist.rs b/backend/src/processing/processor/playlist.rs index 7f27dbf55..37fc1d478 100644 --- a/backend/src/processing/processor/playlist.rs +++ b/backend/src/processing/processor/playlist.rs @@ -115,10 +115,15 @@ fn filter_playlist(source: &mut PlaylistSource, target: &ConfigTarget) -> Option } pub fn apply_filter_to_playlist(playlist: &mut [PlaylistGroup], filter: &Filter) -> Option> { - let mut new_playlist = Vec::with_capacity(128); - for pg in playlist.iter_mut() { - let channels = - pg.channels.iter().filter(|&pli| is_valid(pli, filter, false)).cloned().collect::>(); + // NOTE: the source `playlist` is intentionally cloned (not drained) here because + // the caller reuses the same slice for every target output and for the no-filter + // fallback path, so the survivors cannot be moved out of it. Cap the initial + // allocation so selective filters do not retain capacity for every source item. + const INITIAL_FILTERED_GROUP_CAPACITY: usize = 256; + let mut new_playlist = Vec::with_capacity(playlist.len()); + for pg in playlist.iter() { + let mut channels = Vec::with_capacity(pg.channels.len().min(INITIAL_FILTERED_GROUP_CAPACITY)); + channels.extend(pg.channels.iter().filter(|&pli| is_valid(pli, filter, false)).cloned()); if !channels.is_empty() { new_playlist.push(PlaylistGroup { id: pg.id, @@ -1074,9 +1079,10 @@ fn execute_pipe<'a>( // This method is needed, because of duplicate group names in different inputs. // We merge the same group names considering cluster together. fn flatten_groups(playlistgroups: Vec) -> Vec { - let mut sort_order: Vec = vec![]; + let upper_bound = playlistgroups.len(); + let mut sort_order: Vec = Vec::with_capacity(upper_bound); let mut idx: usize = 0; - let mut group_map: HashMap = HashMap::new(); + let mut group_map: HashMap = HashMap::with_capacity(upper_bound); for group in playlistgroups { let normalized_title: Arc = shared::utils::deunicode_string(&group.title).to_lowercase().intern(); let key = (group.xtream_cluster, normalized_title); diff --git a/backend/src/processing/processor/stream_probe.rs b/backend/src/processing/processor/stream_probe.rs index 84c62bb91..36d960a0b 100644 --- a/backend/src/processing/processor/stream_probe.rs +++ b/backend/src/processing/processor/stream_probe.rs @@ -58,7 +58,7 @@ pub enum GenericProbeMetadataOutcome { } fn requires_provider_connection_for_generic_probe(input_type: InputType) -> bool { - !(matches!(input_type, InputType::Library) || input_type.is_media_server()) + input_type.capabilities().requires_provider_connection_for_probe } fn uses_seekable_remote_probe(item_type: PlaylistItemType, is_remote_probe: bool) -> bool { @@ -177,9 +177,9 @@ async fn prepare_generic_stream_metadata( InputType::Xtream | InputType::XtreamBatch => { let cluster = if item_type.is_live() { XtreamCluster::Live - } else if matches!(item_type, PlaylistItemType::Video | PlaylistItemType::LocalVideo) { + } else if item_type.is_video() { XtreamCluster::Video - } else if matches!(item_type, PlaylistItemType::Series | PlaylistItemType::LocalSeries) { + } else if item_type.is_series() { XtreamCluster::Series } else { // Generic probing currently supports live/video/series payload shapes. @@ -223,9 +223,7 @@ async fn prepare_generic_stream_metadata( debug!("Skipping unsupported generic stream probe for {unique_id}: {safe_probe_url}"); return Ok(PreparedGenericProbeOutcome::Noop); } - let is_remote_probe = reqwest::Url::parse(&probe_url) - .ok() - .is_some_and(|url| matches!(url.scheme(), "http" | "https")); + let is_remote_probe = reqwest::Url::parse(&probe_url).is_ok_and(|url| matches!(url.scheme(), "http" | "https")); let config = app_config.config.load(); let metadata_update = config.metadata_update.clone().unwrap_or_default(); let ffprobe_timeout = metadata_update.ffprobe.timeout.unwrap_or(60); @@ -422,7 +420,7 @@ pub fn update_properties( raw_audio: Option>, stats: ProbeStreamStats, ) { - if matches!(item_type, PlaylistItemType::Video | PlaylistItemType::LocalVideo) { + if item_type.is_video() { let mut props = if let Some(StreamProperties::Video(p)) = props_opt { *p.clone() } else { @@ -453,9 +451,8 @@ pub fn update_properties( } *props_opt = Some(StreamProperties::Video(Box::new(props))); } - else if matches!(item_type, PlaylistItemType::Series | PlaylistItemType::LocalSeries) { - let mut props = if let Some(StreamProperties::Episode(p)) = props_opt { - *p.clone() + else if item_type.is_series() { + let mut props = if let Some(StreamProperties::Episode(p)) = props_opt { *p.clone() } else { EpisodeStreamProperties { episode_id: virtual_id, diff --git a/backend/src/processing/processor/trakt.rs b/backend/src/processing/processor/trakt.rs index aaf8201b4..0fb5c5816 100644 --- a/backend/src/processing/processor/trakt.rs +++ b/backend/src/processing/processor/trakt.rs @@ -42,7 +42,7 @@ fn is_compatible_content_type(cluster: XtreamCluster, content_type: TraktContent fn is_matchable_playlist_item(item_type: PlaylistItemType, content_type: TraktContentType) -> bool { match content_type { - TraktContentType::Vod => matches!(item_type, PlaylistItemType::Video | PlaylistItemType::LocalVideo), + TraktContentType::Vod => item_type.is_video(), TraktContentType::Series => matches!(item_type, PlaylistItemType::SeriesInfo | PlaylistItemType::LocalSeriesInfo), TraktContentType::Both => matches!( item_type, @@ -246,7 +246,7 @@ fn series_children_by_parent_code(playlist: &[PlaylistGroup]) -> HashMap, Vec<&PlaylistItem>>::new(); for playlist_group in playlist { for channel in &playlist_group.channels { - if matches!(channel.header.item_type, PlaylistItemType::Series | PlaylistItemType::LocalSeries) + if channel.header.item_type.is_series() && !channel.header.parent_code.is_empty() { children.entry(channel.header.parent_code.clone()).or_default().push(channel); diff --git a/backend/src/processing/processor/xtream.rs b/backend/src/processing/processor/xtream.rs index 33967fd4c..a0cc2b3f5 100644 --- a/backend/src/processing/processor/xtream.rs +++ b/backend/src/processing/processor/xtream.rs @@ -143,9 +143,8 @@ pub async fn update_live_stream_metadata( let mut success = false; let mut not_found = false; - let is_remote_probe = reqwest::Url::parse(probe_url_cow.as_ref()) - .ok() - .is_some_and(|u| matches!(u.scheme(), "http" | "https")); + let is_remote_probe = + reqwest::Url::parse(probe_url_cow.as_ref()).is_ok_and(|u| matches!(u.scheme(), "http" | "https")); let probe_params = crate::utils::ffmpeg::ProbeParams { url: probe_url_cow.as_ref(), user_agent: user_agent.as_deref(), diff --git a/backend/src/processing/processor/xtream_series.rs b/backend/src/processing/processor/xtream_series.rs index ebfd19f60..645189b34 100644 --- a/backend/src/processing/processor/xtream_series.rs +++ b/backend/src/processing/processor/xtream_series.rs @@ -131,9 +131,7 @@ async fn playlist_resolve_series_info( } // Apply resolved episodes to playlist - for group in new_playlist { - fpl.update_playlist(&group).await; - } + fpl.extend_playlist(new_playlist); } fn sync_resolved_series_properties(provider_fpl: &mut FetchedPlaylist<'_>, processed_fpl: &mut FetchedPlaylist<'_>) { @@ -198,8 +196,16 @@ fn queue_background_series_info( // Extract filter before iterating to avoid borrow conflict let resolve_filter = fpl.input.options.as_ref().and_then(|o| o.resolve_filter.as_ref()); + let stage_started = Instant::now(); + let mut total = 0usize; + let mut skipped_filter = 0usize; + let mut skipped_resolve_filter = 0usize; + let mut queued = 0usize; + let mut expanded = 0usize; for pli in fpl.items_mut() { + total += 1; if !filter(pli) { + skipped_filter += 1; continue; } @@ -207,6 +213,7 @@ fn queue_background_series_info( if let Some(r_filter) = resolve_filter { let provider = ValueProvider { pli, match_as_ascii: false }; if !r_filter.filter(&provider) { + skipped_resolve_filter += 1; continue; } } @@ -245,13 +252,22 @@ fn queue_background_series_info( ); } mgr.queue_task_background(input_name_arc.clone(), task); + queued += 1; } } if let Some(group_obj) = expand_series_item(pli, input) { groups_to_add.push(group_obj); + expanded += 1; } } + + if log_enabled!(Level::Debug) { + debug!( + "[Task] Series resolve queueing summary for input {input_name_arc}: total={total}, queued={queued}, expanded={expanded}, skipped_filter={skipped_filter}, skipped_resolve_filter={skipped_resolve_filter}, elapsed_ms={}", + stage_started.elapsed().as_millis() + ); + } groups_to_add } @@ -997,8 +1013,7 @@ pub async fn update_series_metadata( active_handle, ); let is_remote_probe = reqwest::Url::parse(probe_url.as_ref()) - .ok() - .is_some_and(|u| matches!(u.scheme(), "http" | "https")); + .is_ok_and(|u| matches!(u.scheme(), "http" | "https")); let probe_params = crate::utils::ffmpeg::ProbeParams { url: probe_url.as_ref(), user_agent: user_agent.as_deref(), diff --git a/backend/src/processing/processor/xtream_vod.rs b/backend/src/processing/processor/xtream_vod.rs index f12c9628e..05e215606 100644 --- a/backend/src/processing/processor/xtream_vod.rs +++ b/backend/src/processing/processor/xtream_vod.rs @@ -447,8 +447,15 @@ fn queue_background_vod_info( let input = fpl.input; // Extract filter before iterating to avoid borrow conflict let resolve_filter = fpl.input.options.as_ref().and_then(|o| o.resolve_filter.as_ref()); + let stage_started = std::time::Instant::now(); + let mut total = 0usize; + let mut skipped_filter = 0usize; + let mut skipped_resolve_filter = 0usize; + let mut queued = 0usize; for pli in fpl.items_mut() { + total += 1; if !filter(pli) { + skipped_filter += 1; continue; } @@ -456,6 +463,7 @@ fn queue_background_vod_info( if let Some(r_filter) = resolve_filter { let provider = ValueProvider { pli, match_as_ascii: false }; if !r_filter.filter(&provider) { + skipped_resolve_filter += 1; continue; } } @@ -500,8 +508,17 @@ fn queue_background_vod_info( ); } mgr.queue_task_background(input.name.clone(), task); + queued += 1; } } + + if log_enabled!(Level::Debug) { + debug!( + "[Task] VOD resolve queueing summary for input {}: total={total}, queued={queued}, skipped_filter={skipped_filter}, skipped_resolve_filter={skipped_resolve_filter}, elapsed_ms={}", + input.name, + stage_started.elapsed().as_millis() + ); + } } async fn update_vod_info_immediate( @@ -859,9 +876,8 @@ pub async fn update_vod_metadata( temp_handle.as_ref().and_then(ProbeHandleGuard::handle), active_handle, ); - let is_remote_probe = reqwest::Url::parse(probe_url.as_ref()) - .ok() - .is_some_and(|u| matches!(u.scheme(), "http" | "https")); + let is_remote_probe = + reqwest::Url::parse(probe_url.as_ref()).is_ok_and(|u| matches!(u.scheme(), "http" | "https")); let probe_params = crate::utils::ffmpeg::ProbeParams { url: probe_url.as_ref(), user_agent: user_agent.as_deref(), diff --git a/backend/src/repository/epg_repository.rs b/backend/src/repository/epg_repository.rs index 3e8d3e745..252ed0003 100644 --- a/backend/src/repository/epg_repository.rs +++ b/backend/src/repository/epg_repository.rs @@ -71,6 +71,11 @@ fn build_epg_rename_map(playlist: Option<&[PlaylistGroup]>) -> HashMap, pub async fn epg_write_for_target(cfg: &Config, target: &ConfigTarget, target_path: &Path, epg: Option<&Epg>, output: &TargetOutput, playlist: Option<&[PlaylistGroup]>) -> Result<(), TuliproxError> { + if !output.target_type().supports_epg() { + // Formats without EPG support are skipped here via the shared capability + // table rather than a silent empty match arm. + return Ok(()); + } if let Some(epg_data) = epg { let rename_map = Arc::new(build_epg_rename_map(playlist)); let epg_data = Arc::new(epg_data.clone()); diff --git a/backend/src/repository/playlist_repository.rs b/backend/src/repository/playlist_repository.rs index d761de392..0effe4ac2 100644 --- a/backend/src/repository/playlist_repository.rs +++ b/backend/src/repository/playlist_repository.rs @@ -16,10 +16,10 @@ use crate::repository::{ }; use crate::repository::{TargetIdMapping, VirtualIdRecord}; use crate::utils; -use log::{info, warn}; +use log::{debug, info, warn}; use shared::error::{ TuliproxError}; use shared::model::xtream_const::XTREAM_CLUSTER; -use shared::model::{InputType, M3uPlaylistItem, PlaylistEntry, PlaylistGroup, PlaylistItem, PlaylistItemHeader, PlaylistItemType, SeriesStreamDetailEpisodeProperties, SeriesStreamDetailProperties, StreamProperties, UUIDType, VirtualId, XtreamCluster, XtreamPlaylistItem}; +use shared::model::{InputPersistence, M3uPlaylistItem, PlaylistEntry, PlaylistGroup, PlaylistItem, PlaylistItemHeader, PlaylistItemType, SeriesStreamDetailEpisodeProperties, SeriesStreamDetailProperties, StreamProperties, UUIDType, VirtualId, XtreamCluster, XtreamPlaylistItem}; use shared::utils::{is_dash_url, is_hls_url, Internable}; use std::collections::{HashMap, HashSet}; use std::path::{Path, PathBuf}; @@ -78,7 +78,7 @@ pub async fn persist_playlist(app_config: &Arc, playlist: &mut [Playl let uuid = header.get_uuid(); let item_type = header.item_type; - let parent_virtual_id = if matches!(item_type, PlaylistItemType::Series | PlaylistItemType::LocalSeries) { + let parent_virtual_id = if item_type.is_series() { target_id_mapping.get_parent_virtual_id_by_uuid(uuid).unwrap_or_default() } else { 0 @@ -110,12 +110,8 @@ pub async fn persist_playlist(app_config: &Arc, playlist: &mut [Playl drop(media_server_series); for output in &target.output { - let mut filtered: Option> = match output { - TargetOutput::Xtream(out) => out.filter.as_ref().and_then(|flt| apply_filter_to_playlist(playlist, flt)), - TargetOutput::M3u(out) => out.filter.as_ref().and_then(|flt| apply_filter_to_playlist(playlist, flt)), - TargetOutput::Strm(out) => out.filter.as_ref().and_then(|flt| apply_filter_to_playlist(playlist, flt)), - TargetOutput::HdHomeRun(_) => None, - }; + let mut filtered: Option> = + output.filter().and_then(|flt| apply_filter_to_playlist(playlist, flt)); let pl: &mut [PlaylistGroup] = if let Some(filtered_playlist) = filtered.as_mut() { filtered_playlist.as_mut_slice() @@ -450,7 +446,7 @@ fn rewrite_series_episode_parent_virtual_ids(playlist: &mut [PlaylistGroup], tar for group in playlist.iter_mut() { for channel in &mut group.channels { let header = &mut channel.header; - if matches!(header.item_type, PlaylistItemType::Series | PlaylistItemType::LocalSeries) { + if header.item_type.is_series() { if let Some(parent_virtual_id) = series_parent_virtual_ids.get(&header.parent_code) { let provider_id = header.get_provider_id().unwrap_or_default(); let item_type = header.item_type; @@ -492,7 +488,11 @@ async fn load_target_id_mapping_as_tree(app_config: &AppConfig, target_path: &Pa .map_err(|err| TuliproxError::Config(format!("Could not find path for target {target_name} err:{err}"))) } -async fn load_xtream_playlist_as_tree(app_config: &AppConfig, storage_path: &Path, cluster: XtreamCluster) -> BPlusTree { +async fn load_xtream_playlist_as_tree( + app_config: &AppConfig, + storage_path: &Path, + cluster: XtreamCluster, +) -> Result, TuliproxError> { let xtream_path = xtream_get_file_path(storage_path, cluster); let file_lock = app_config.file_locks.read_lock(&xtream_path).await; // Move B+Tree query and iteration to spawn_blocking to avoid blocking tokio runtime @@ -501,8 +501,19 @@ async fn load_xtream_playlist_as_tree(app_config: &AppConfig, storage_path: &Pat let _guard = file_lock; BPlusTree::::load(&path_clone) }).await { - Ok(Ok(tree)) => tree, - _ => BPlusTree::new(), + Ok(Ok(tree)) => Ok(tree), + Ok(Err(err)) if err.kind() == std::io::ErrorKind::NotFound => { + debug!("No xtream {cluster} storage at {}, serving empty playlist", xtream_path.display()); + Ok(BPlusTree::new()) + } + Ok(Err(err)) => Err(TuliproxError::RepositoryXtream(format!( + "Failed to load xtream {cluster} storage {}: {err}", + xtream_path.display() + ))), + Err(join_err) => Err(TuliproxError::RepositoryXtream(format!( + "Failed to join xtream {cluster} storage load task {}: {join_err}", + xtream_path.display() + ))), } } @@ -520,9 +531,9 @@ async fn load_xtream_target_storage(app_config: &AppConfig, target: &ConfigTarge let storage_path = xtream_get_storage_path(&config, target.name.as_str()).ok_or_else(|| TuliproxError::Config(format!("Could not find path for target {} xtream output", target.name)))?; - let live_storage = load_xtream_playlist_as_tree(app_config, &storage_path, XtreamCluster::Live).await; - let vod_storage = load_xtream_playlist_as_tree(app_config, &storage_path, XtreamCluster::Video).await; - let series_storage = load_xtream_playlist_as_tree(app_config, &storage_path, XtreamCluster::Series).await; + let live_storage = load_xtream_playlist_as_tree(app_config, &storage_path, XtreamCluster::Live).await?; + let vod_storage = load_xtream_playlist_as_tree(app_config, &storage_path, XtreamCluster::Video).await?; + let series_storage = load_xtream_playlist_as_tree(app_config, &storage_path, XtreamCluster::Series).await?; Ok(PlaylistXtreamStorage { live: live_storage, @@ -545,7 +556,18 @@ async fn load_m3u_target_storage(app_config: &AppConfig, target: &ConfigTarget) BPlusTree::::load(&path_clone) }).await { Ok(Ok(tree)) => Ok(tree), - _ => Ok(BPlusTree::new()), + Ok(Err(err)) if err.kind() == std::io::ErrorKind::NotFound => { + debug!("No m3u storage at {}, serving empty playlist", m3u_path.display()); + Ok(BPlusTree::new()) + } + Ok(Err(err)) => Err(TuliproxError::RepositoryM3u(format!( + "Failed to load m3u storage {}: {err}", + m3u_path.display() + ))), + Err(join_err) => Err(TuliproxError::RepositoryM3u(format!( + "Failed to join m3u storage load task {}: {join_err}", + m3u_path.display() + ))), } } @@ -593,12 +615,12 @@ pub async fn persist_input_playlist(app_config: &Arc, input: &ConfigI } }; - match input.get_download_input_type() { - InputType::Xtream | InputType::XtreamBatch => { + match input.get_download_input_type().persistence() { + InputPersistence::Xtream => { persist_input_xtream_playlist(app_config, &storage_path, playlist).await } - InputType::M3u | InputType::M3uBatch => { + InputPersistence::M3u => { // Persist M3U let file_path = get_input_m3u_playlist_file_path(&storage_path, &input.name); if let Err(err) = persist_input_m3u_playlist(app_config, &file_path, &playlist).await { @@ -606,7 +628,7 @@ pub async fn persist_input_playlist(app_config: &Arc, input: &ConfigI } (playlist, None) } - InputType::Library => { + InputPersistence::Library => { // Persist local library playlist let file_path = get_input_local_library_playlist_file_path(&storage_path, &input.name); let (playlist, result) = persist_input_library_playlist(app_config, &file_path, playlist).await; @@ -615,7 +637,7 @@ pub async fn persist_input_playlist(app_config: &Arc, input: &ConfigI } (playlist, None) } - InputType::Emby | InputType::Jellyfin | InputType::Plex => { + InputPersistence::MediaServer => { let file_path = get_input_media_server_playlist_file_path(&storage_path, &input.name); let (playlist, result) = persist_input_media_server_playlist(app_config, &file_path, playlist).await; if let Err(err) = result { @@ -633,8 +655,8 @@ pub async fn load_input_playlist(ctx: &PlaylistProcessingContext, input: &Config .map_err(|e| TuliproxError::Config(format!("Error getting input path: {e}")))?; let disk_based_processing = cfg.disk_based_processing; - match input.get_download_input_type() { - InputType::Xtream | InputType::XtreamBatch => { + match input.get_download_input_type().persistence() { + InputPersistence::Xtream => { let clusters_to_load = clusters.unwrap_or(&XTREAM_CLUSTER); if disk_based_processing { let source = PlaylistSource::xtream_disk( @@ -646,7 +668,7 @@ pub async fn load_input_playlist(ctx: &PlaylistProcessingContext, input: &Config Ok(MemoryPlaylistSource::new(groups).into_source()) } } - InputType::M3u | InputType::M3uBatch => { + InputPersistence::M3u => { // Load M3U let file_path = get_input_m3u_playlist_file_path(&storage_path, &input.name); if disk_based_processing && file_path.exists() { @@ -658,7 +680,7 @@ pub async fn load_input_playlist(ctx: &PlaylistProcessingContext, input: &Config Ok(MemoryPlaylistSource::new(groups).into_source()) } } - InputType::Library => { + InputPersistence::Library => { let file_path = get_input_local_library_playlist_file_path(&storage_path, &input.name); if disk_based_processing && file_path.exists() { Ok(PlaylistSource::local_library_disk( @@ -669,7 +691,7 @@ pub async fn load_input_playlist(ctx: &PlaylistProcessingContext, input: &Config Ok(MemoryPlaylistSource::new(groups).into_source()) } } - InputType::Emby | InputType::Jellyfin | InputType::Plex => { + InputPersistence::MediaServer => { let file_path = get_input_media_server_playlist_file_path(&storage_path, &input.name); if disk_based_processing && file_path.exists() { Ok(PlaylistSource::media_server_disk( diff --git a/backend/src/repository/playlist_source.rs b/backend/src/repository/playlist_source.rs index 7a44e8c0f..65af774a4 100644 --- a/backend/src/repository/playlist_source.rs +++ b/backend/src/repository/playlist_source.rs @@ -11,7 +11,7 @@ use shared::model::{M3uPlaylistItem, PlaylistEntry, PlaylistGroup, PlaylistItem, use shared::model::UUIDType; use shared::utils::Internable; use std::borrow::Cow; -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use std::path::{Path, PathBuf}; use std::sync::Arc; @@ -240,6 +240,21 @@ impl PlaylistSource { } } + /// Batched, indexed equivalent of repeatedly calling [`Self::update_playlist`] + /// for in-memory playlists. Groups whose cluster is in the active skip set + /// are dropped, matching `update_playlist`. Only the in-memory source merges + /// groups, mirroring `FetchedPlaylist::update_playlist`. + pub fn extend_playlist(&mut self, groups: Vec) { + let filtered: Vec = if let Some(skip_set) = self.skip_set.as_ref() { + groups.into_iter().filter(|plg| !skip_set.contains(&plg.xtream_cluster)).collect() + } else { + groups + }; + if let PlaylistSourceKind::Memory(source) = &mut self.kind { + source.merge_groups(filtered); + } + } + pub fn get_missing_vod_info_count(&mut self) -> usize { if let Some(skip_set) = self.skip_set.as_ref() { return match &mut self.kind { @@ -885,6 +900,46 @@ impl MemoryPlaylistSource { } pub fn into_source(self) -> PlaylistSource { PlaylistSource::memory(self) } + + /// Merge a batch of groups into the in-memory playlist in a single pass. + /// + /// Equivalent to calling [`PlaylistSourceOps::update_playlist`] for every + /// incoming group, but builds `(cluster, normalized_title)` and + /// `(cluster, id)` indexes over the existing groups once instead of + /// re-scanning (and re-normalizing every existing title) for each incoming + /// group. This turns the series-expansion merge from O(groups^2) into a + /// single linear pass. Title matches take precedence over id matches, and + /// newly pushed groups are indexed so later groups in the same batch can + /// merge into them, preserving the sequential semantics. + pub(crate) fn merge_groups(&mut self, incoming: Vec) { + if incoming.is_empty() { + return; + } + let playlist = Arc::make_mut(&mut self.playlist); + let mut by_title: HashMap<(XtreamCluster, String), usize> = HashMap::with_capacity(playlist.len()); + let mut by_id: HashMap<(XtreamCluster, u32), usize> = HashMap::with_capacity(playlist.len()); + for (idx, grp) in playlist.iter().enumerate() { + by_title + .entry((grp.xtream_cluster, shared::utils::deunicode_string(&grp.title).to_lowercase())) + .or_insert(idx); + by_id.entry((grp.xtream_cluster, grp.id)).or_insert(idx); + } + for plg in incoming { + let title_key = (plg.xtream_cluster, shared::utils::deunicode_string(&plg.title).to_lowercase()); + if let Some(&idx) = by_title.get(&title_key) { + playlist[idx].channels.extend(plg.channels); + continue; + } + if let Some(&idx) = by_id.get(&(plg.xtream_cluster, plg.id)) { + playlist[idx].channels.extend(plg.channels); + continue; + } + let new_idx = playlist.len(); + by_title.entry(title_key).or_insert(new_idx); + by_id.entry((plg.xtream_cluster, plg.id)).or_insert(new_idx); + playlist.push(plg); + } + } } impl Default for MemoryPlaylistSource { diff --git a/backend/src/repository/xtream_repository.rs b/backend/src/repository/xtream_repository.rs index 9eab660ae..7bf1ceace 100644 --- a/backend/src/repository/xtream_repository.rs +++ b/backend/src/repository/xtream_repository.rs @@ -51,6 +51,19 @@ pub fn get_collection_path(path: &Path, collection: &str) -> PathBuf { path.join(format!("{collection}.json")) } +/// Returns the category-collection base name for an [`XtreamCluster`]. +/// +/// Centralizes the per-cluster `cat_live` / `cat_vod` / `cat_series` mapping so the +/// path-deriving call sites read a single property instead of re-matching the cluster. +#[inline] +pub(crate) const fn xtream_cluster_category_collection(cluster: XtreamCluster) -> &'static str { + match cluster { + XtreamCluster::Live => storage_const::COL_CAT_LIVE, + XtreamCluster::Video => storage_const::COL_CAT_VOD, + XtreamCluster::Series => storage_const::COL_CAT_SERIES, + } +} + #[inline] pub fn get_live_cat_collection_path(path: &Path) -> PathBuf { get_collection_path(path, storage_const::COL_CAT_LIVE) @@ -748,11 +761,7 @@ where } pub(crate) async fn xtream_get_playlist_categories(config: &Config, target_name: &str, cluster: XtreamCluster) -> Option> { - let path = xtream_get_collection_path(config, target_name, match cluster { - XtreamCluster::Live => storage_const::COL_CAT_LIVE, - XtreamCluster::Video => storage_const::COL_CAT_VOD, - XtreamCluster::Series => storage_const::COL_CAT_SERIES, - }); + let path = xtream_get_collection_path(config, target_name, xtream_cluster_category_collection(cluster)); if let Ok(file_path) = path { if let Ok(content) = tokio::fs::read_to_string(&file_path).await { return serde_json::from_str::>(&content).ok(); @@ -1134,11 +1143,7 @@ pub async fn persist_input_xtream_playlist(app_config: &Arc, storage_ let root_path = storage_path.to_path_buf(); let app_cfg = app_config.clone(); for cluster in XTREAM_CLUSTER { - let col_path = match cluster { - XtreamCluster::Live => get_collection_path(&root_path, storage_const::COL_CAT_LIVE), - XtreamCluster::Video => get_collection_path(&root_path, storage_const::COL_CAT_VOD), - XtreamCluster::Series => get_collection_path(&root_path, storage_const::COL_CAT_SERIES), - }; + let col_path = get_collection_path(&root_path, xtream_cluster_category_collection(cluster)); let data = fetched_categories.get_mut(cluster); // if there is no data save only if no file exists! Prevent data loss from failed download attempt if !data.is_empty() || !file_exists_async(&col_path).await { @@ -1446,11 +1451,7 @@ pub async fn load_input_xtream_playlist(app_config: &Arc, storage_pat for &cluster in clusters { let xtream_path = xtream_get_file_path(storage_path, cluster); if xtream_path.exists() { - let cat_col_name = match cluster { - XtreamCluster::Live => storage_const::COL_CAT_LIVE, - XtreamCluster::Video => storage_const::COL_CAT_VOD, - XtreamCluster::Series => storage_const::COL_CAT_SERIES, - }; + let cat_col_name = xtream_cluster_category_collection(cluster); let cat_path = get_collection_path(storage_path, cat_col_name); if cat_path.exists() { diff --git a/backend/src/utils/network/request.rs b/backend/src/utils/network/request.rs index 710145e01..9a5129edf 100644 --- a/backend/src/utils/network/request.rs +++ b/backend/src/utils/network/request.rs @@ -1160,10 +1160,6 @@ pub async fn get_remote_content_as_file( let mut writer = async_file_writer(File::create(file_path).await?); let mut stream = response.bytes_stream(); - while let Some(chunk) = stream.next().await { - let bytes = chunk.map_err(|e| string_to_io_error(format!("Failed to read chunk: {e}")))?; - writer.write_all(&bytes).await?; - } let idle_timeout = tokio::time::Duration::from_secs(STREAM_IDLE_TIMEOUT); let idle = sleep(idle_timeout); diff --git a/backend/src/utils/network/xtream.rs b/backend/src/utils/network/xtream.rs index 0251090e6..160cf29ae 100644 --- a/backend/src/utils/network/xtream.rs +++ b/backend/src/utils/network/xtream.rs @@ -52,11 +52,7 @@ pub fn get_xtream_player_api_action_url(input: &ConfigInput, action: &str) -> Op } pub fn get_xtream_player_api_info_url(input: &ConfigInput, cluster: XtreamCluster, stream_id: u32) -> Option { - let (action, stream_id_field) = match cluster { - XtreamCluster::Live => (crate::model::XC_ACTION_GET_LIVE_INFO, crate::model::XC_LIVE_ID), - XtreamCluster::Video => (crate::model::XC_ACTION_GET_VOD_INFO, crate::model::XC_VOO_ID), - XtreamCluster::Series => (crate::model::XC_ACTION_GET_SERIES_INFO, crate::model::XC_SERIES_ID), - }; + let (action, stream_id_field) = cluster.info_action_and_id_field(); get_xtream_player_api_action_url(input, action).map(|action_url| format!("{action_url}&{stream_id_field}={stream_id}")) } @@ -409,7 +405,7 @@ async fn download_xtream_from_source( } else { match input.resolve_url(&input_source.url) { Ok(url) => url.into_owned(), - Err(err) => return (Vec::with_capacity(0), vec![err], false), + Err(err) => return (Vec::new(), vec![err], false), } }; @@ -418,15 +414,14 @@ async fn download_xtream_from_source( if let Err(err) = xtream_login(app_config, client, &input_source_login, username).await { error!("Could not log in with xtream user {username} for provider {}. {err}", input.name); - return (Vec::with_capacity(0), vec![err], false); + return (Vec::new(), vec![err], false); } let mut playlist_groups: Vec = Vec::with_capacity(128); let cfg = app_config.config.load(); let storage_dir = &cfg.storage_dir; - let use_disk_based_processing = - cfg.disk_based_processing && matches!(source_input_type, InputType::Xtream | InputType::XtreamBatch); + let use_disk_based_processing = cfg.disk_based_processing && source_input_type.is_xtream(); let mut errors = vec![]; for (xtream_cluster, category, stream) in &ACTIONS { diff --git a/frontend/src/app/components/playlist/input_table.rs b/frontend/src/app/components/playlist/input_table.rs index 02dcff992..80b793f64 100644 --- a/frontend/src/app/components/playlist/input_table.rs +++ b/frontend/src/app/components/playlist/input_table.rs @@ -8,7 +8,7 @@ use crate::{ i18n::use_translation, }; use shared::{ - model::{ConfigInputAliasDto, ConfigInputDto, InputType, SortOrder}, + model::{ConfigInputAliasDto, ConfigInputDto, SortOrder}, utils::unix_ts_to_str, }; use std::rc::Rc; @@ -59,7 +59,7 @@ pub fn InputTable(props: &InputTableProps) -> Html { /> }, 1 => html! { dto.name.as_ref() }, 2 => html! { }, - 3 => html! { if matches!(dto.input_type, InputType::XtreamBatch | InputType::M3uBatch) { + 3 => html! { if dto.input_type.is_batch() { } else { {dto.url.as_str()} diff --git a/frontend/src/app/components/playlist/playlist_rows.rs b/frontend/src/app/components/playlist/playlist_rows.rs index dd88983d6..c366bfc2b 100644 --- a/frontend/src/app/components/playlist/playlist_rows.rs +++ b/frontend/src/app/components/playlist/playlist_rows.rs @@ -1,5 +1,5 @@ use crate::app::components::InputRow; -use shared::model::{ConfigInputDto, ConfigTargetDto, InputType, SourcesConfigDto}; +use shared::model::{ConfigInputDto, ConfigTargetDto, SourcesConfigDto}; use std::{collections::HashMap, rc::Rc, sync::Arc}; pub type PlaylistRows = Vec<(Vec>, Vec>)>; @@ -55,7 +55,7 @@ pub fn collect_provider_buttons(rows: &PlaylistRows) -> Vec { let InputRow::Input(input) = &**row else { continue; }; - if matches!(input.input_type, InputType::M3uBatch | InputType::XtreamBatch) { + if input.input_type.is_batch() { continue; } if seen.insert(input.id) { diff --git a/shared/src/foundation/filter.rs b/shared/src/foundation/filter.rs index e5b306c22..6b03d82e7 100644 --- a/shared/src/foundation/filter.rs +++ b/shared/src/foundation/filter.rs @@ -139,12 +139,8 @@ impl Filter { if let Some(value) = provider.get(field.as_ref()) { get_filter_item_type(&value).is_some_and(|pli_type| { let is_match = match item_type { - PlaylistItemType::Video => { - matches!(pli_type, PlaylistItemType::Video | PlaylistItemType::LocalVideo) - } - PlaylistItemType::Series => { - matches!(pli_type, PlaylistItemType::Series | PlaylistItemType::LocalSeries) - } + PlaylistItemType::Video => pli_type.is_video(), + PlaylistItemType::Series => pli_type.is_series(), _ => pli_type.eq(item_type), }; if log_enabled!(Level::Trace) { diff --git a/shared/src/foundation/value_provider.rs b/shared/src/foundation/value_provider.rs index 415bc2a21..033e1b00d 100644 --- a/shared/src/foundation/value_provider.rs +++ b/shared/src/foundation/value_provider.rs @@ -102,43 +102,68 @@ macro_rules! get_genre { pub use get_genre; pub use set_genre; +/// Canonical list of `ItemField` variants that map 1:1 to a simple `Arc` slot on +/// `PlaylistItemHeader`. +/// +/// Both `get_field_value` and `set_field_value` are generated from this single list (via a +/// callback macro), so a directly-bound field cannot be added to the read half but forgotten in +/// the write half (or vice versa). The asymmetric fields (`Genre`, `Type`, `Caption`) stay +/// explicit because their read/write behavior differs. +macro_rules! for_each_direct_field { + ($cb:ident) => { + $cb! { + Group => group, + Name => name, + Title => title, + Url => url, + Input => input_name, + } + }; +} + pub fn get_field_value(pli: &PlaylistItem, field: ItemField) -> Arc { let header = &pli.header; - match field { - ItemField::Group => Arc::clone(&header.group), - ItemField::Name => Arc::clone(&header.name), - ItemField::Title => Arc::clone(&header.title), - ItemField::Genre => get_genre!(header).unwrap_or_else(|| "".intern()), - ItemField::Url => Arc::clone(&header.url), - ItemField::Input => Arc::clone(&header.input_name), - ItemField::Type => header.item_type.intern(), - ItemField::Caption => { - if header.title.is_empty() { - Arc::clone(&header.name) - } else { - Arc::clone(&header.title) + + macro_rules! get_arms { + ($($variant:ident => $prop:ident),+ $(,)?) => { + match field { + $(ItemField::$variant => Arc::clone(&header.$prop),)+ + ItemField::Genre => get_genre!(header).unwrap_or_else(|| "".intern()), + ItemField::Type => header.item_type.interned_label(), + ItemField::Caption => { + if header.title.is_empty() { + Arc::clone(&header.name) + } else { + Arc::clone(&header.title) + } + } } - } + }; } + + for_each_direct_field!(get_arms) } pub fn set_field_value(pli: &mut PlaylistItem, field: ItemField, value: &str) -> bool { let header = &mut pli.header; - match field { - ItemField::Group => header.group = value.intern(), - ItemField::Name => header.name = value.intern(), - ItemField::Title => header.title = value.intern(), - ItemField::Genre => { - return set_genre!(header, value); - } - ItemField::Url => header.url = value.intern(), - ItemField::Input => header.input_name = value.intern(), - ItemField::Caption => { - header.title = value.intern(); - header.name = header.title.clone(); - } - ItemField::Type => {} + + macro_rules! set_arms { + ($($variant:ident => $prop:ident),+ $(,)?) => { + match field { + $(ItemField::$variant => header.$prop = value.intern(),)+ + ItemField::Genre => { + return set_genre!(header, value); + } + ItemField::Caption => { + header.title = value.intern(); + header.name = header.title.clone(); + } + ItemField::Type => {} + } + }; } + + for_each_direct_field!(set_arms); true } diff --git a/shared/src/model/config/input.rs b/shared/src/model/config/input.rs index 167d47aa8..4326800f9 100644 --- a/shared/src/model/config/input.rs +++ b/shared/src/model/config/input.rs @@ -113,12 +113,75 @@ pub enum InputType { impl InputType { pub fn is_xtream(&self) -> bool { matches!(self, Self::Xtream | Self::XtreamBatch) } pub fn is_m3u(&self) -> bool { matches!(self, Self::M3u | Self::M3uBatch) } + pub fn is_batch(&self) -> bool { matches!(self, Self::M3uBatch | Self::XtreamBatch) } pub fn uses_standard_input_url(&self) -> bool { matches!(self, Self::M3u | Self::Xtream | Self::M3uBatch | Self::XtreamBatch) } pub fn is_library(&self) -> bool { matches!(self, Self::Library) } pub fn is_media_server(&self) -> bool { matches!(self, Self::Emby | Self::Jellyfin | Self::Plex) } + + /// Single source of truth for the categorical behavior of an input type. + /// + /// Adding a new [`InputType`] variant forces an arm here (the match is + /// exhaustive), and every site that consumes [`InputCapabilities`] โ€” + /// persistence/load routing, probe requirements, the custom-provider + /// endpoint gate โ€” stays in sync automatically instead of relying on a + /// parallel `match` somewhere else that is easy to forget. + #[must_use] + pub const fn capabilities(self) -> InputCapabilities { + match self { + Self::M3u | Self::M3uBatch => InputCapabilities { + persistence: InputPersistence::M3u, + requires_provider_connection_for_probe: true, + served_on_custom_provider_endpoint: true, + }, + Self::Xtream | Self::XtreamBatch => InputCapabilities { + persistence: InputPersistence::Xtream, + requires_provider_connection_for_probe: true, + served_on_custom_provider_endpoint: true, + }, + Self::Library => InputCapabilities { + persistence: InputPersistence::Library, + requires_provider_connection_for_probe: false, + served_on_custom_provider_endpoint: false, + }, + Self::Emby | Self::Jellyfin | Self::Plex => InputCapabilities { + persistence: InputPersistence::MediaServer, + requires_provider_connection_for_probe: false, + served_on_custom_provider_endpoint: false, + }, + } + } + + /// Persistence/load backend family for this input type. + #[must_use] + pub const fn persistence(self) -> InputPersistence { self.capabilities().persistence } +} + +/// Storage/loading backend family an [`InputType`] maps onto. +/// +/// Multiple input variants collapse onto the same persistence family (for +/// example every media-server variant shares the same on-disk format), so +/// persist/load routing can match on this instead of re-listing variants. +#[derive(Debug, Copy, Clone, PartialEq, Eq)] +pub enum InputPersistence { + M3u, + Xtream, + Library, + MediaServer, +} + +/// Categorical capabilities of an [`InputType`], declared once in +/// [`InputType::capabilities`]. +#[derive(Debug, Copy, Clone, PartialEq, Eq)] +pub struct InputCapabilities { + /// Storage/loading backend family. + pub persistence: InputPersistence, + /// Whether generic stream probing must open a provider connection. + pub requires_provider_connection_for_probe: bool, + /// Whether the custom-provider HTTP endpoint can serve this input. + pub served_on_custom_provider_endpoint: bool, } #[derive( @@ -664,9 +727,7 @@ impl ConfigInputDto { if self.enabled { self.prepare_media_server_input()?; } - if self.url.starts_with(PROVIDER_SCHEME_PREFIX) - && matches!(self.input_type, InputType::M3uBatch | InputType::XtreamBatch) - { + if self.url.starts_with(PROVIDER_SCHEME_PREFIX) && self.input_type.is_batch() { return Err(TuliproxError::ConfigInput(format!( "input type {} does not support provider:// URLs for batch definitions; use batch:// URL (input: {})", self.input_type, self.name @@ -855,9 +916,7 @@ impl ConfigInputDto { pub fn prepare_type(&mut self) -> Result<(), TuliproxError> { self.url = self.url.trim().to_string(); self.normalize_input_type_from_batch_url(); - if self.url.starts_with(PROVIDER_SCHEME_PREFIX) - && matches!(self.input_type, InputType::M3uBatch | InputType::XtreamBatch) - { + if self.url.starts_with(PROVIDER_SCHEME_PREFIX) && self.input_type.is_batch() { return Err(TuliproxError::ConfigInput(format!( "input type {} does not support provider:// URLs for batch definitions; use batch:// URL", self.input_type diff --git a/shared/src/model/epg.rs b/shared/src/model/epg.rs index 28aca260f..ba95180aa 100644 --- a/shared/src/model/epg.rs +++ b/shared/src/model/epg.rs @@ -58,16 +58,60 @@ impl EpgChannel { pub fn get_programme_with_limit(&self, limit: u32) -> Vec<&EpgProgramme> { let now = Utc::now().timestamp(); + self.get_programme_with_limit_at(limit, now) + } - // find index for first relevant entry - let start_idx = self - .programmes - .iter() - .position(|p| (p.start <= now && now <= p.stop) || (p.start >= now)) - .unwrap_or(self.programmes.len()); // nothing found, empty response + fn get_programme_with_limit_at(&self, limit: u32, now: i64) -> Vec<&EpgProgramme> { + // Programmes are sorted by start, not by stop. Use binary search only + // for the first future entry, then scan the earlier prefix for an + // overlapping programme whose stop time has not passed. + let first_future = self.programmes.partition_point(|programme| programme.start < now); + let start_idx = + self.programmes[..first_future].iter().position(|programme| programme.stop >= now).unwrap_or(first_future); + self.programmes.iter().skip(start_idx).filter(|programme| programme.stop >= now).take(limit as usize).collect() + } +} - // slice from start_idx, max. limit - self.programmes.iter().skip(start_idx).take(limit as usize).collect() +#[cfg(test)] +mod tests { + use super::{EpgChannel, EpgProgramme}; + use crate::utils::Internable; + + #[test] + fn programme_limit_handles_non_monotonic_stop_times() { + let channel = EpgChannel { + id: "channel".intern(), + title: None, + icon: None, + programmes: vec![ + EpgProgramme::new(0, 100, "channel".intern()), + EpgProgramme::new(10, 20, "channel".intern()), + EpgProgramme::new(30, 40, "channel".intern()), + ], + }; + + let programmes = channel.get_programme_with_limit_at(1, 50); + assert_eq!(programmes.len(), 1); + assert_eq!(programmes[0].start, 0); + } + + #[test] + fn programme_limit_excludes_expired_entries_after_current_programme() { + let channel = EpgChannel { + id: "channel".intern(), + title: None, + icon: None, + programmes: vec![ + EpgProgramme::new(0, 100, "channel".intern()), + EpgProgramme::new(10, 20, "channel".intern()), + EpgProgramme::new(60, 80, "channel".intern()), + ], + }; + + let programmes = channel.get_programme_with_limit_at(2, 50); + assert_eq!(programmes.len(), 2); + assert_eq!(programmes[0].start, 0); + assert_eq!(programmes[1].start, 60); } } diff --git a/shared/src/model/playlist.rs b/shared/src/model/playlist.rs index 5369c28d7..ce4cb0d20 100644 --- a/shared/src/model/playlist.rs +++ b/shared/src/model/playlist.rs @@ -47,6 +47,18 @@ impl XtreamCluster { Self::Series => "series", } } + + /// Returns the xtream `player_api` info action and the stream-id query field for this cluster. + /// + /// Keeps the per-cluster `(action, id_field)` mapping attached to the enum so call sites read a + /// single property instead of re-deriving it with a local `match`. + pub fn info_action_and_id_field(&self) -> (&'static str, &'static str) { + match self { + Self::Live => (xtream_const::XC_ACTION_GET_LIVE_INFO, xtream_const::XC_LIVE_ID), + Self::Video => (xtream_const::XC_ACTION_GET_VOD_INFO, xtream_const::XC_VOD_ID), + Self::Series => (xtream_const::XC_ACTION_GET_SERIES_INFO, xtream_const::XC_SERIES_ID), + } + } } impl TryFrom for XtreamCluster { @@ -148,6 +160,12 @@ impl PlaylistItemType { pub fn is_live_adaptive(&self) -> bool { matches!(self, PlaylistItemType::LiveHls | PlaylistItemType::LiveDash) } + /// True for VOD item types (`Video` or `LocalVideo`). + pub fn is_video(&self) -> bool { matches!(self, PlaylistItemType::Video | PlaylistItemType::LocalVideo) } + + /// True for concrete series item types (`Series` or `LocalSeries`); excludes the `SeriesInfo` containers. + pub fn is_series(&self) -> bool { matches!(self, PlaylistItemType::Series | PlaylistItemType::LocalSeries) } + /// Controls address tracking only. /// Do not use this to decide whether a playback request should use session-based admission /// or whether a logical playback must stay on the same provider account. @@ -185,6 +203,24 @@ impl PlaylistItemType { } } + /// Returns a cached interned `Arc` of this type's label. + /// + /// `intern()` performs an interner hash-map lookup on every call. Because the + /// label is one of only five fixed values, this caches the interned `Arc` per + /// label in a `OnceLock` and returns a cheap `Arc::clone`, avoiding repeated + /// interner lookups on hot sort/filter paths. + pub fn interned_label(&self) -> Arc { + static CACHE: [std::sync::OnceLock>; 5] = [const { std::sync::OnceLock::new() }; 5]; + let idx = match self { + Self::Live | Self::LiveHls | Self::LiveDash | Self::LiveUnknown => 0, + Self::Video | Self::LocalVideo => 1, + Self::Series | Self::LocalSeries => 2, + Self::SeriesInfo | Self::LocalSeriesInfo => 3, + Self::Catchup => 4, + }; + Arc::clone(CACHE[idx].get_or_init(|| self.as_str().intern())) + } + pub fn is_cluster(&self, cluster: XtreamCluster) -> bool { match self { Self::Live | Self::LiveHls | Self::LiveDash | Self::LiveUnknown => cluster == XtreamCluster::Live, diff --git a/shared/src/model/target_type.rs b/shared/src/model/target_type.rs index 57e3f38ce..e48f5ac91 100644 --- a/shared/src/model/target_type.rs +++ b/shared/src/model/target_type.rs @@ -11,3 +11,52 @@ pub enum TargetType { Strm, HdHomeRun, } + +impl TargetType { + /// Single source of truth for the categorical capabilities of an output + /// format. + /// + /// Adding a new [`TargetType`] variant forces an arm here (the match is + /// exhaustive), and the filtering / EPG / memory-cache sites derive from + /// this table instead of each re-deciding per variant โ€” so a new format can + /// no longer silently inherit a blank "no-op" arm in one of those places. + #[must_use] + pub const fn capabilities(self) -> TargetCapabilities { + match self { + Self::Xtream => { + TargetCapabilities { supports_filter: true, supports_epg: true, supports_memory_cache: true } + } + Self::M3u => TargetCapabilities { supports_filter: true, supports_epg: true, supports_memory_cache: true }, + Self::Strm => { + TargetCapabilities { supports_filter: true, supports_epg: false, supports_memory_cache: false } + } + Self::HdHomeRun => { + TargetCapabilities { supports_filter: false, supports_epg: false, supports_memory_cache: false } + } + } + } + + /// Whether this output format produces an EPG. + #[must_use] + pub const fn supports_epg(self) -> bool { self.capabilities().supports_epg } + + /// Whether this output format honors a playlist filter. + #[must_use] + pub const fn supports_filter(self) -> bool { self.capabilities().supports_filter } + + /// Whether this output format can be served from the in-memory playlist cache. + #[must_use] + pub const fn supports_memory_cache(self) -> bool { self.capabilities().supports_memory_cache } +} + +/// Categorical capabilities of a [`TargetType`], declared once in +/// [`TargetType::capabilities`]. +#[derive(Debug, Copy, Clone, PartialEq, Eq)] +pub struct TargetCapabilities { + /// Whether the format honors a playlist filter. + pub supports_filter: bool, + /// Whether the format produces an EPG. + pub supports_epg: bool, + /// Whether the format can be served from the in-memory playlist cache. + pub supports_memory_cache: bool, +} diff --git a/shared/src/model/xtream_const.rs b/shared/src/model/xtream_const.rs index 7dcda0826..5758a7403 100644 --- a/shared/src/model/xtream_const.rs +++ b/shared/src/model/xtream_const.rs @@ -1,7 +1,7 @@ use crate::model::XtreamCluster; pub const XC_LIVE_ID: &str = "live_id"; -pub const XC_VOO_ID: &str = "vod_id"; +pub const XC_VOD_ID: &str = "vod_id"; pub const XC_SERIES_ID: &str = "series_id"; pub const XC_ACTION_GET_SERIES_INFO: &str = "get_series_info";