From 9eef3b24e069f534b476e2516e48f89fdd165d24 Mon Sep 17 00:00:00 2001 From: euzu <33094714+euzu@users.noreply.github.com> Date: Fri, 17 Apr 2026 11:57:43 +0200 Subject: [PATCH] Feature/filter resolve probe (#720) * Filter for resolve * Filter for probe --- CHANGELOG.md | 2 + backend/src/api/api_utils.rs | 232 +++++++++--------- .../src/api/endpoints/api_playlist_utils.rs | 110 +++++---- .../src/api/endpoints/stream_history_api.rs | 50 ++-- backend/src/api/endpoints/v1_api_config.rs | 7 +- backend/src/api/endpoints/v1_api_playlist.rs | 56 +++-- backend/src/api/endpoints/web_index.rs | 30 +-- backend/src/api/endpoints/websocket_api.rs | 8 +- backend/src/api/main_api.rs | 134 +++++----- .../src/api/model/active_provider_manager.rs | 2 +- backend/src/api/model/active_user_manager.rs | 2 + backend/src/api/model/admission_strategy.rs | 2 +- backend/src/api/model/app_state.rs | 36 +-- backend/src/api/model/connection_manager.rs | 7 +- .../src/api/model/metadata_update_manager.rs | 59 ++--- .../src/api/model/qos_aggregation_manager.rs | 105 ++++---- backend/src/api/model/request.rs | 8 +- .../model/streams/shared_stream_manager.rs | 5 +- backend/src/api/sys_usage.rs | 2 +- backend/src/library/classifier.rs | 8 +- backend/src/library/processor.rs | 46 ++-- backend/src/model/config/input.rs | 7 +- backend/src/processing/processor/library.rs | 20 +- backend/src/processing/processor/playlist.rs | 94 +++++-- backend/src/processing/processor/sort.rs | 6 +- .../src/processing/processor/stream_probe.rs | 2 +- .../src/processing/processor/xtream_series.rs | 22 ++ .../src/processing/processor/xtream_vod.rs | 22 ++ backend/src/repository/playlist_repository.rs | 2 +- .../src/repository/qos_snapshot_repository.rs | 7 +- .../src/repository/stream_history/archive.rs | 1 + .../src/repository/stream_history/reader.rs | 6 +- .../src/repository/stream_history/storage.rs | 1 + backend/src/repository/user_repository.rs | 1 + backend/src/repository/xtream_repository.rs | 4 +- backend/src/runtime_config_report.rs | 4 +- backend/src/utils/stream_history_viewer.rs | 6 +- docs/src/configuration/source.md | 30 +-- frontend/public/assets/i18n/en.json | 4 + .../app/components/playlist/input_table.rs | 198 +++------------ .../playlist/playlist_settings_view.rs | 10 +- frontend/src/app/components/sidebar.rs | 16 +- .../components/source_editor/input_form.rs | 26 +- frontend/src/hooks/use_server_status.rs | 18 +- shared/src/model/config/input.rs | 117 +++++++-- shared/src/model/config/source.rs | 2 +- shared/src/model/config/stream.rs | 99 +++++--- shared/src/model/xtream.rs | 54 ++-- shared/src/utils/flags.rs | 1 + 49 files changed, 912 insertions(+), 779 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index fdf136ad5..e07754ab7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,8 @@ ## 🌟 New Features +- **Input Resolve Filter**: Added `resolve_filter` option to input configuration to selectively resolve only entries matching a filter expression. +- **Input Probe Filter**: Added `probe_filter` option to input configuration to selectively probe only entries matching a filter expression. - **Soft Connections And Soft Priority**: API users can now be configured with `soft_connections` and `soft_priority`. - Soft connections allow a user to consume additional preemptible provider slots above `max_connections`. - `soft_priority` is only applied while a connection is using a soft slot; once a regular slot becomes available again, the running connection diff --git a/backend/src/api/api_utils.rs b/backend/src/api/api_utils.rs index 272cfb83b..43808f8c2 100644 --- a/backend/src/api/api_utils.rs +++ b/backend/src/api/api_utils.rs @@ -561,7 +561,7 @@ pub(in crate::api) async fn resolve_admission_with_strategies( activate_unbound_session: bool, eviction_reentry_guard: EvictionReentryGuard<'_>, ) -> (crate::api::model::ConnectionAdmission, Option) { - use crate::api::model::{AdmissionDecision, ConnectionAdmission, ConnectionKind, StrategyContext, evaluate_strategy}; + use crate::api::model::{evaluate_strategy, AdmissionDecision, ConnectionAdmission, ConnectionKind, StrategyContext}; use shared::model::UserConnectionPermission; let admission = get_admission_for_request( @@ -573,7 +573,7 @@ pub(in crate::api) async fn resolve_admission_with_strategies( session_token, activate_unbound_session, ) - .await; + .await; if admission.permission != UserConnectionPermission::Exhausted { return (admission, None); @@ -610,7 +610,7 @@ pub(in crate::api) async fn resolve_admission_with_strategies( eviction_reentry_guard, &target.addr, ) - .await + .await { debug!( "Skipping eviction strategy {strategy:?} for recently evicted request of user {username} targeting {}", @@ -633,7 +633,7 @@ pub(in crate::api) async fn resolve_admission_with_strategies( session_token, activate_unbound_session, ) - .await; + .await; if retry_admission.permission == UserConnectionPermission::Allowed { return (retry_admission, None); } @@ -722,7 +722,7 @@ async fn activate_session_before_stream_open( EvictionReentryGuard::Session(session_token) }, ) - .await; + .await; (admission, grace_mode, created_placeholder) } @@ -905,10 +905,10 @@ fn get_grace_period_millis( ) -> u64 { if config_grace_period_millis > 0 && ( - matches!(stream_response_params, ProviderStreamState::GracePeriod(_, _)) // provider grace period + matches!(stream_response_params, ProviderStreamState::GracePeriod(_, _)) // provider grace period || connection_permission == UserConnectionPermission::GracePeriod - // user grace period - ) + // user grace period + ) { config_grace_period_millis } else { @@ -952,7 +952,7 @@ async fn create_stream_response_details( session_owner, }, ) - .await; + .await; let mut grace_period_options = app_state.get_grace_options(); grace_period_options.period_millis = get_grace_period_millis( connection_permission, @@ -1063,7 +1063,7 @@ async fn create_stream_response_details( &app_state.http_client.load(), provider_stream_factory_options, ) - .await + .await { None => (None, None), Some((stream, info)) => (Some(stream), info), @@ -1241,7 +1241,7 @@ fn is_throttled_stream(item_type: PlaylistItemType, throttle_kbps: usize) -> boo fn prepare_body_stream(app_state: &Arc, item_type: PlaylistItemType, stream: S) -> axum::body::Body where - S: futures::Stream> + Send + 'static, + S: futures::Stream> + Send + 'static, { let throttle_kbps = usize::try_from(get_stream_throttle(app_state)).unwrap_or_default(); let body_stream = if is_throttled_stream(item_type, throttle_kbps) { @@ -1304,7 +1304,7 @@ pub async fn force_provider_stream_response( Some(user_session.token.as_str()), grace_mode.map(|mode| matches!(mode, crate::api::model::GraceMode::Hold)), ) - .await + .await { Ok(stream_details) => stream_details, Err(err) => { @@ -1327,7 +1327,7 @@ pub async fn force_provider_stream_response( stream_details.stream.is_some(), stream_details.has_deferred_provider_open(), ) - .await; + .await; let provider_response = stream_details.stream_info.as_ref().map(|(h, sc, url, cvt)| (h.clone(), *sc, url.clone(), *cvt)); if ctx.session_reservation_ttl_secs > 0 { @@ -1358,7 +1358,7 @@ pub async fn force_provider_stream_response( meter_uid: metering.meter_uid, meter_stream: metering.meter_stream, }) - .await; + .await; let (status_code, header_map) = get_stream_response_with_headers(provider_response.map(|(h, s, _, _)| (h, s))); let mut response = axum::response::Response::builder().status(status_code); @@ -1439,7 +1439,7 @@ pub async fn stream_response( socket_bound: item_type.uses_socket_bound_session(), }, ) - .await; + .await; let grace_mode = resolved_grace_mode.or(grace_mode); connection_permission = final_admission.permission; connection_kind = final_admission.kind.unwrap_or(connection_kind); @@ -1462,7 +1462,7 @@ pub async fn stream_response( session_token, req_headers, ) - .await + .await { return value.into_response(); } @@ -1483,7 +1483,7 @@ pub async fn stream_response( session_token, req_headers, ) - .await + .await { debug_if_enabled!( "Opportunistic shared stream reuse for {}", @@ -1515,7 +1515,7 @@ pub async fn stream_response( &fingerprint.addr, CustomVideoStreamType::UserConnectionsExhausted, ) - .into_response(); + .into_response(); } let stream_options = get_stream_options(app_state); @@ -1540,7 +1540,7 @@ pub async fn stream_response( Some(session_token), grace_mode.map(|m| matches!(m, crate::api::model::GraceMode::Hold)), ) - .await + .await { Ok(stream_details) => stream_details, Err(err) => { @@ -1589,7 +1589,7 @@ pub async fn stream_response( stream_details.stream.is_some(), stream_details.has_deferred_provider_open(), ) - .await; + .await; let mut is_stream_shared = share_stream && !stream_details.has_deferred_provider_open(); if let Some((_header, _status_code, _url, Some(_custom_video))) = stream_details.stream_info.as_ref() { @@ -1624,7 +1624,7 @@ pub async fn stream_response( meter_uid: metering.meter_uid, meter_stream: metering.meter_stream, }) - .await; + .await; let stream_resp = if is_stream_shared { debug_if_enabled!( "Streaming shared stream request from {}", @@ -1643,7 +1643,7 @@ pub async fn stream_response( connection_priority_for_kind(user, connection_kind), connection_kind, ) - .await + .await { let (status_code, header_map) = get_stream_response_with_headers(provider_response.map(|(h, s, _, _)| (h, s))); @@ -1862,7 +1862,7 @@ async fn try_shared_stream_response_if_any( connection_priority_for_kind(user, connection_kind), connection_kind, ) - .await + .await { debug_if_enabled!("Using shared stream {}", sanitize_sensitive_info(stream_url)); if let Some(headers) = app_state.shared_stream_manager.get_shared_state_headers(stream_url).await { @@ -1910,8 +1910,8 @@ async fn try_shared_stream_response_if_any( meter_uid: metering.meter_uid, meter_stream: metering.meter_stream, }) - .await - .boxed(); + .await + .boxed(); let mut response = axum::response::Response::builder().status(status_code); for (key, value) in &header_map { response = response.header(key, value); @@ -1948,15 +1948,15 @@ pub async fn local_stream_response( let allow_session_reopen = if let Some(session_token) = playback_session_token { user.max_connections > 0 && app_state - .active_users - .connection_permission_for_session( - &user.username, - user.max_connections, - user.soft_connections, - session_token, - ) - .await - != UserConnectionPermission::Exhausted + .active_users + .connection_permission_for_session( + &user.username, + user.max_connections, + user.soft_connections, + session_token, + ) + .await + != UserConnectionPermission::Exhausted } else { false }; @@ -1976,7 +1976,7 @@ pub async fn local_stream_response( &fingerprint.addr, CustomVideoStreamType::UserConnectionsExhausted, ) - .into_response(); + .into_response(); } connection_permission = UserConnectionPermission::Allowed; } @@ -2080,7 +2080,7 @@ pub async fn local_stream_response( socket_bound: pli.item_type.uses_socket_bound_session(), }, ) - .await; + .await; grace_mode = resolved_grace_mode; connection_permission = final_admission.permission; connection_kind = final_admission.kind.unwrap_or(connection_kind); @@ -2095,7 +2095,7 @@ pub async fn local_stream_response( &fingerprint.addr, CustomVideoStreamType::UserConnectionsExhausted, ) - .into_response(); + .into_response(); } } let mut grace_period_options = app_state.get_grace_options(); @@ -2144,7 +2144,7 @@ pub async fn local_stream_response( meter_uid: 0, meter_stream: false, }) - .await; + .await; let mut response = Response::new(axum::body::Body::from_stream(stream)); @@ -2322,7 +2322,7 @@ async fn fetch_resource_with_retry( default_user_agent.as_deref(), ) }) - .await + .await else { return None; }; @@ -2367,8 +2367,8 @@ pub async fn resource_response( mime_type.unwrap_or_else(|| mime::APPLICATION_OCTET_STREAM.to_string()), Some("public, max-age=14400"), ) - .await - .into_response(); + .await + .into_response(); } } trace_if_enabled!("Try to fetch resource {}", sanitize_sensitive_info(resource_url)); @@ -2458,7 +2458,7 @@ pub fn json_or_bin_response(accept: Option<&str>, data: &T) -> imp pub fn stream_json_or_bin_response

( accept: Option<&str>, - data: Box + Send>, + data: Box + Send>, ) -> axum::response::Response where P: serde::Serialize + Send + 'static, @@ -2472,7 +2472,7 @@ where pub fn stream_json_or_bin_response_stream(accept: Option<&str>, data: S) -> axum::response::Response where P: serde::Serialize + Send + 'static, - S: Stream + Send + Unpin + 'static, + S: Stream + Send + Unpin + 'static, { if accept.is_some_and(|a| a.contains(CONTENT_TYPE_CBOR)) { return stream_bin_array_stream(data); @@ -2504,13 +2504,13 @@ pub(crate) fn should_allow_exhausted_shared_reconnect( ) -> bool { share_stream && user_session.is_some_and(|session| { - session.permission != UserConnectionPermission::Exhausted - && session.virtual_id == requested_virtual_id - && session.stream_url.as_ref() == requested_stream_url - }) + session.permission != UserConnectionPermission::Exhausted + && session.virtual_id == requested_virtual_id + && session.stream_url.as_ref() == requested_stream_url + }) } -pub fn stream_json_array

(iter: Box + Send>) -> axum::response::Response +pub fn stream_json_array

(iter: Box + Send>) -> axum::response::Response where P: serde::Serialize + Send + 'static, { @@ -2538,7 +2538,7 @@ where try_unwrap_body!(Response::builder().header(header::CONTENT_TYPE, CONTENT_TYPE_JSON).body(body)) } -pub fn stream_bin_array

(iter: Box + Send>) -> axum::response::Response +pub fn stream_bin_array

(iter: Box + Send>) -> axum::response::Response where P: serde::Serialize + Send + 'static, { @@ -2562,11 +2562,11 @@ where // CBOR: start indefinite-length array Ok::<_, Infallible>(Bytes::from_static(&[0x9f])) }) - .chain(stream) - .chain(stream::once(async { - // CBOR: end indefinite-length array - Ok::<_, Infallible>(Bytes::from_static(&[0xff])) - })), + .chain(stream) + .chain(stream::once(async { + // CBOR: end indefinite-length array + Ok::<_, Infallible>(Bytes::from_static(&[0xff])) + })), ); try_unwrap_body!(Response::builder().header(header::CONTENT_TYPE, CONTENT_TYPE_CBOR).body(body)) @@ -2575,7 +2575,7 @@ where pub fn stream_json_array_stream(stream: S) -> axum::response::Response where P: serde::Serialize + Send + 'static, - S: Stream + Send + Unpin + 'static, + S: Stream + Send + Unpin + 'static, { let stream = stream::unfold((stream, true), |(mut stream, first)| async move { match stream.next().await { @@ -2604,7 +2604,7 @@ where pub fn stream_bin_array_stream(stream: S) -> axum::response::Response where P: serde::Serialize + Send + 'static, - S: Stream + Send + Unpin + 'static, + S: Stream + Send + Unpin + 'static, { let stream = stream::unfold(stream, |mut stream| async move { match stream.next().await { @@ -2675,10 +2675,10 @@ pub fn empty_json_response_as_array() -> axum::http::Result provider_name, - _ => panic!("fallback-enabled request should allocate a provider"), - }; + .await; + let (ProviderStreamState::Available(Some(fallback_provider), _) | ProviderStreamState::GracePeriod(Some(fallback_provider), _)) = fallback.provider_stream_state else + { panic!("fallback-enabled request should allocate a provider") }; assert_eq!(fallback_provider.as_ref(), "provider_2"); app_state.active_provider.release_connection(&busy_addr).await; @@ -2868,7 +2865,7 @@ mod tests { socket_bound: false, active_addrs: vec!["127.0.0.1:1234".parse().unwrap_or_else(|_| unreachable!())], ts: 1, - started_at:1, + started_at: 1, permission: UserConnectionPermission::Allowed, counted: false, }; @@ -3103,19 +3100,21 @@ mod tests { } fn create_test_app_state_with_stream_config(stream: crate::model::StreamConfig) -> Arc { - let mut config = Config::default(); - config.reverse_proxy = Some(crate::model::ReverseProxyConfig { - resource_rewrite_disabled: false, - rewrite_secret: [0; 16], - resource_retry: crate::model::ResourceRetryConfig::default(), - disabled_header: None, - stream: Some(stream), - cache: None, - rate_limit: None, - geoip: None, - stream_history: None, - qos_aggregation: None, - }); + let config = Config { + reverse_proxy: Some(crate::model::ReverseProxyConfig { + resource_rewrite_disabled: false, + rewrite_secret: [0; 16], + resource_retry: crate::model::ResourceRetryConfig::default(), + disabled_header: None, + stream: Some(stream), + cache: None, + rate_limit: None, + geoip: None, + stream_history: None, + qos_aggregation: None, + }), + ..Config::default() + }; let mut app_cfg = create_test_app_config(); app_cfg.config = Arc::new(ArcSwap::from_pointee(config)); @@ -3318,7 +3317,7 @@ mod tests { false, EvictionReentryGuard::Session("tok-third"), ) - .await; + .await; assert_eq!(admission.permission, UserConnectionPermission::Allowed); assert_eq!(admission.kind, Some(crate::api::model::ConnectionKind::Soft)); @@ -3379,7 +3378,7 @@ mod tests { false, EvictionReentryGuard::Session("vod-session"), ) - .await; + .await; assert_eq!(session_based.0.permission, UserConnectionPermission::Allowed); let connection_based = resolve_admission_with_strategies( @@ -3394,7 +3393,7 @@ mod tests { false, EvictionReentryGuard::Session("vod-session"), ) - .await; + .await; assert_eq!(connection_based.0.permission, UserConnectionPermission::Exhausted); } @@ -3482,7 +3481,7 @@ mod tests { false, EvictionReentryGuard::SocketPlayback { virtual_id: 9001 }, ) - .await; + .await; assert_eq!(admission.permission, UserConnectionPermission::Exhausted); assert_eq!(grace_mode, None); @@ -3575,7 +3574,7 @@ mod tests { false, EvictionReentryGuard::SocketPlayback { virtual_id: 9103 }, ) - .await; + .await; assert_eq!(admission.permission, UserConnectionPermission::Allowed); assert!(app_state.active_users.active_streams().await.is_empty()); @@ -3664,7 +3663,7 @@ mod tests { false, EvictionReentryGuard::Session("session-other"), ) - .await; + .await; assert_eq!(admission.permission, UserConnectionPermission::Allowed); assert_eq!(grace_mode, None); @@ -3755,7 +3754,7 @@ mod tests { false, EvictionReentryGuard::SocketPlayback { virtual_id: 9201 }, ) - .await; + .await; assert_eq!(admission.permission, UserConnectionPermission::Allowed); assert_eq!(admission.kind, Some(crate::api::model::ConnectionKind::Soft)); @@ -3808,8 +3807,8 @@ mod tests { None, false, ) - .await - .into_response(); + .await + .into_response(); let active_streams = app_state.active_users.active_streams().await; assert_eq!(active_streams.len(), 1, "local file streaming should register an active stream"); @@ -3819,8 +3818,7 @@ mod tests { #[tokio::test] async fn local_stream_response_rechecks_limits_before_registering_socket_bound_streams() { let mut app_cfg = create_test_app_config(); - let mut config = Config::default(); - config.user_access_control = true; + let config = Config { user_access_control: true, ..Config::default() }; app_cfg.config = Arc::new(ArcSwap::from_pointee(config)); let app_state = create_test_app_state_for_config(Arc::new(app_cfg)); let temp_dir = tempfile::tempdir().expect("tempdir"); @@ -3868,8 +3866,8 @@ mod tests { Some(&first_token), false, ) - .await - .into_response(); + .await + .into_response(); let _second_response = local_stream_response( &second_fingerprint, @@ -3884,8 +3882,8 @@ mod tests { Some(&second_token), false, ) - .await - .into_response(); + .await + .into_response(); assert_eq!(app_state.active_users.user_connections(&user.username).await, 1); assert_eq!(app_state.active_users.active_streams().await.len(), 1); @@ -3929,7 +3927,7 @@ mod tests { 0, crate::api::model::ConnectionKind::Normal, ) - .await; + .await; assert!(registered.is_some(), "shared stream should register"); let mut user = ProxyUserCredentials::default(); @@ -3986,8 +3984,8 @@ mod tests { false, None, ) - .await - .into_response(); + .await + .into_response(); assert_eq!(response.status(), StatusCode::OK); let session_admission = app_state @@ -4039,8 +4037,8 @@ mod tests { None, false, ) - .await - .into_response(); + .await + .into_response(); assert!(!should_compress_response(&response)); } @@ -4089,8 +4087,8 @@ mod tests { Some(playback_session_token), false, ) - .await - .into_response(); + .await + .into_response(); let _second_response = local_stream_response( &second_fingerprint, @@ -4105,8 +4103,8 @@ mod tests { Some(playback_session_token), false, ) - .await - .into_response(); + .await + .into_response(); let active_streams = app_state.active_users.active_streams().await; assert_eq!(active_streams.len(), 1, "stable playback token should reuse the tracked local connection"); @@ -4160,8 +4158,8 @@ mod tests { Some(playback_session_token), false, ) - .await - .into_response(); + .await + .into_response(); let second_response = local_stream_response( &second_fingerprint, @@ -4176,8 +4174,8 @@ mod tests { Some(playback_session_token), false, ) - .await - .into_response(); + .await + .into_response(); assert_eq!(second_response.status(), StatusCode::OK); @@ -4234,8 +4232,8 @@ mod tests { Some(playback_session_token), false, ) - .await - .into_response(); + .await + .into_response(); let second_response = local_stream_response( &second_fingerprint, @@ -4250,8 +4248,8 @@ mod tests { Some(playback_session_token), false, ) - .await - .into_response(); + .await + .into_response(); assert_eq!(second_response.status(), StatusCode::OK); @@ -4315,8 +4313,8 @@ mod tests { true, EvictionReentryGuard::Session("tok-hls-first"), ) - .await - .0; + .await + .0; let second_admission = resolve_admission_with_strategies( &app_state, &user.username, @@ -4329,8 +4327,8 @@ mod tests { true, EvictionReentryGuard::Session("tok-hls-second"), ) - .await - .0; + .await + .0; assert_eq!(first_admission.permission, UserConnectionPermission::Allowed); assert_eq!(second_admission.permission, UserConnectionPermission::Exhausted); diff --git a/backend/src/api/endpoints/api_playlist_utils.rs b/backend/src/api/endpoints/api_playlist_utils.rs index 07054fa77..5110a9f95 100644 --- a/backend/src/api/endpoints/api_playlist_utils.rs +++ b/backend/src/api/endpoints/api_playlist_utils.rs @@ -92,56 +92,6 @@ fn rewrite_resource_url(encrypt_secret: &[u8; 16], resource_url: &str, item: UiP item } -#[cfg(test)] -mod tests { - use super::rewrite_resource_url; - use shared::{ - model::{PlaylistItemType, UiPlaylistItem, XtreamCluster}, - utils::{obfuscate_text, Internable}, - }; - - fn sample_item(logo: &str) -> UiPlaylistItem { - UiPlaylistItem { - virtual_id: 1, - provider_id: "provider".intern(), - name: "name".intern(), - title: "title".intern(), - group: "group".intern(), - logo: logo.intern(), - url: "file:///tmp/video.mkv".intern(), - item_type: PlaylistItemType::Live, - xtream_cluster: XtreamCluster::Live, - category_id: 0, - rating: 0.0, - input_name: "test".intern(), - } - } - - #[test] - fn rewrite_resource_url_keeps_internal_api_paths() { - let secret = [7u8; 16]; - let item = sample_item("/api/v1/library/thumbnail/test-uuid"); - - let rewritten = rewrite_resource_url(&secret, "/api/v1/playlist/resource", item); - - assert_eq!(rewritten.logo.as_ref(), "/api/v1/library/thumbnail/test-uuid"); - } - - #[test] - fn rewrite_resource_url_wraps_external_urls() { - let secret = [7u8; 16]; - let item = sample_item("https://example.com/poster.jpg"); - - let rewritten = rewrite_resource_url(&secret, "/api/v1/playlist/resource", item); - let expected_suffix = obfuscate_text(&secret, "https://example.com/poster.jpg"); - - assert_eq!( - rewritten.logo.as_ref(), - format!("/api/v1/playlist/resource/{expected_suffix}") - ); - } -} - pub(in crate::api::endpoints) async fn get_playlist_for_input( cfg_input: Option<&Arc>, app_state: &Arc, @@ -215,3 +165,63 @@ pub(in crate::api::endpoints) async fn get_playlist_for_custom_provider( } } } + +#[cfg(test)] +mod tests { + use super::rewrite_resource_url; + use shared::{ + model::{PlaylistItemType, UiPlaylistItem, XtreamCluster}, + utils::{obfuscate_text, Internable}, + }; + + fn sample_item(logo: &str) -> UiPlaylistItem { + UiPlaylistItem { + virtual_id: 1, + provider_id: "provider".intern(), + name: "name".intern(), + title: "title".intern(), + group: "group".intern(), + logo: logo.intern(), + url: "file:///tmp/video.mkv".intern(), + item_type: PlaylistItemType::Live, + xtream_cluster: XtreamCluster::Live, + category_id: 0, + rating: 0.0, + input_name: "test".intern(), + } + } + + #[test] + fn rewrite_resource_url_keeps_internal_api_paths() { + let secret = [7u8; 16]; + let item = sample_item("/api/v1/library/thumbnail/test-uuid"); + + let rewritten = rewrite_resource_url(&secret, "/api/v1/playlist/resource", item); + + assert_eq!(rewritten.logo.as_ref(), "/api/v1/library/thumbnail/test-uuid"); + } + + #[test] + fn rewrite_resource_url_returns_unchanged_for_empty_logo() { + let secret = [7u8; 16]; + let item = sample_item(""); + + let rewritten = rewrite_resource_url(&secret, "/api/v1/playlist/resource", item); + + assert_eq!(rewritten.logo.as_ref(), ""); + } + + #[test] + fn rewrite_resource_url_wraps_external_urls() { + let secret = [7u8; 16]; + let item = sample_item("https://example.com/poster.jpg"); + + let rewritten = rewrite_resource_url(&secret, "/api/v1/playlist/resource", item); + let expected_suffix = obfuscate_text(&secret, "https://example.com/poster.jpg"); + + assert_eq!( + rewritten.logo.as_ref(), + format!("/api/v1/playlist/resource/{expected_suffix}") + ); + } +} diff --git a/backend/src/api/endpoints/stream_history_api.rs b/backend/src/api/endpoints/stream_history_api.rs index 7b3adc0dd..db7e2b565 100644 --- a/backend/src/api/endpoints/stream_history_api.rs +++ b/backend/src/api/endpoints/stream_history_api.rs @@ -3,20 +3,20 @@ use std::io; use std::path::Path; use std::sync::Arc; -use axum::extract::{Path as AxumPath, Query, State}; -use axum::http::StatusCode; -use axum::response::{IntoResponse, Response}; -use serde::{Deserialize, Serialize}; -use shared::model::StreamHistoryEventType; use crate::api::api_utils::json_or_bin_response; use crate::api::endpoints::extract_accept_header::ExtractAcceptHeader; use crate::api::model::AppState; -use crate::model::{StreamHistoryRecord}; +use crate::model::StreamHistoryRecord; use crate::repository::{QosSnapshotRecord, QosSnapshotRepository, StreamHistoryFileReader}; use crate::utils::stream_history_viewer::{ discover_files, resolve_time_range, CompiledFilter, StreamHistoryQuery, TimeRange, }; +use axum::extract::{Path as AxumPath, Query, State}; +use axum::http::StatusCode; +use axum::response::{IntoResponse, Response}; +use serde::{Deserialize, Serialize}; +use shared::model::StreamHistoryEventType; #[derive(Deserialize)] pub(crate) struct HistoryQueryParams { @@ -276,7 +276,7 @@ impl Iterator for RecordFileIter { loop { if self.current_iter.is_none() { let file = self.files.next()?; - let iter: Box>> = if file.is_archive { + let iter: Box>> = if file.is_archive { match StreamHistoryFileReader::from_archive(&file.path, Some(self.time_range)) { Ok((r, _)) => Box::new(r), Err(e) => { @@ -528,12 +528,12 @@ fn load_qos_snapshot(storage_dir: &str, stream_identity_key: &str) -> io::Result #[cfg(test)] mod tests { - use shared::model::{DisconnectReason, PlaylistItemType}; use super::*; use crate::model::{QosAggregationConfig, ResourceRetryConfig, ReverseProxyConfig}; use crate::repository::{ QosSnapshotDailyBucket, QosSnapshotRecord, QosSnapshotWindow, }; + use shared::model::{DisconnectReason, PlaylistItemType}; use shared::utils::Internable; fn make_record( @@ -657,23 +657,25 @@ mod tests { #[test] fn get_qos_storage_directory_requires_enabled_qos_aggregation() { - let mut cfg = crate::model::Config::default(); - cfg.storage_dir = "/var/lib/tuliprox".to_string(); - cfg.reverse_proxy = Some(ReverseProxyConfig { - resource_rewrite_disabled: false, - rewrite_secret: [0; 16], - resource_retry: ResourceRetryConfig::default(), - disabled_header: None, - stream: None, - cache: None, - rate_limit: None, - geoip: None, - stream_history: None, - qos_aggregation: Some(QosAggregationConfig { - enabled: false, - interval_secs: 300, + let mut cfg = crate::model::Config { + storage_dir: "/var/lib/tuliprox".to_string(), + reverse_proxy: Some(ReverseProxyConfig { + resource_rewrite_disabled: false, + rewrite_secret: [0; 16], + resource_retry: ResourceRetryConfig::default(), + disabled_header: None, + stream: None, + cache: None, + rate_limit: None, + geoip: None, + stream_history: None, + qos_aggregation: Some(QosAggregationConfig { + enabled: false, + interval_secs: 300, + }), }), - }); + ..crate::model::Config::default() + }; assert!(get_qos_storage_directory_from_config(&cfg).is_none()); diff --git a/backend/src/api/endpoints/v1_api_config.rs b/backend/src/api/endpoints/v1_api_config.rs index e97b3cb47..544d0aaea 100644 --- a/backend/src/api/endpoints/v1_api_config.rs +++ b/backend/src/api/endpoints/v1_api_config.rs @@ -700,6 +700,7 @@ mod tests { utils::{HEADER_CONFIG_SOURCES_REVISION, HEADER_IF_MATCH}, }; use std::sync::Arc; + use shared::model::TemplateDefinitionDto; #[test] fn require_matching_revision_rejects_missing_if_match_header() { @@ -745,7 +746,7 @@ mod tests { templates: Some(vec![]), }, mappings: None, - templates: Some(Default::default()), + templates: Some(TemplateDefinitionDto::default()), api_proxy: Some(ApiProxyConfigDto { server: vec![ApiProxyServerInfoDto { name: String::from("main"), @@ -820,7 +821,7 @@ mod tests { providers: None, }; - let input_source = build_xtream_login_input_source(&request, &[provider.clone()]) + let input_source = build_xtream_login_input_source(&request, std::slice::from_ref(&provider)) .expect("provider url should resolve against runtime providers"); assert_eq!(input_source.url, "provider://b1g/player_api.php?username=demo&password=secret"); @@ -874,7 +875,7 @@ mod tests { assert_eq!(input_source.url, "provider://strong/player_api.php?username=bubble&password=gum"); assert_eq!(input_source.provider.as_ref().map(|provider| provider.name.as_ref()), Some("strong")); assert_eq!( - input_source.provider.as_ref().and_then(|provider| provider.urls.first()).map(|url| url.as_ref()), + input_source.provider.as_ref().and_then(|provider| provider.urls.first()).map(std::convert::AsRef::as_ref), Some("http://request.example") ); } diff --git a/backend/src/api/endpoints/v1_api_playlist.rs b/backend/src/api/endpoints/v1_api_playlist.rs index 449869bfb..0a6ccc695 100644 --- a/backend/src/api/endpoints/v1_api_playlist.rs +++ b/backend/src/api/endpoints/v1_api_playlist.rs @@ -36,6 +36,8 @@ fn create_config_input_for_m3u(url: &str) -> ConfigInput { resolve_delay: shared::utils::default_resolve_delay_secs(), probe_delay: shared::utils::default_probe_delay_secs(), probe_live_interval_hours: 120, + resolve_filter: None, + probe_filter: None, }), ..Default::default() } @@ -55,6 +57,8 @@ fn create_config_input_for_xtream(username: &str, password: &str, host: &str) -> resolve_delay: shared::utils::default_resolve_delay_secs(), probe_delay: shared::utils::default_probe_delay_secs(), probe_live_interval_hours: 120, + resolve_filter: None, + probe_filter: None, }), ..Default::default() } @@ -583,6 +587,27 @@ pub fn v1_api_playlist_register_with_permissions( ) } +async fn playlist_episode_item( + axum::extract::Path(virtual_id): axum::extract::Path, + axum::extract::State(app_state): axum::extract::State>, + axum::extract::Json(playlist_req): axum::extract::Json, +) -> impl IntoResponse + Send { + if let PlaylistRequest::Target(target_id) = playlist_req { + if let Some(target) = app_state.app_config.get_target_by_id(target_id) { + if target.has_output(TargetType::Xtream) { + if let Ok(vid) = virtual_id.parse::() { + if let Ok(pli) = + xtream_get_item_for_stream_id(vid, &app_state, &target, Some(XtreamCluster::Series)).await + { + return axum::Json(json!(UiPlaylistItem::from(pli))).into_response(); + } + } + } + } + } + axum::http::StatusCode::NO_CONTENT.into_response() +} + #[cfg(test)] mod tests { use super::resolve_provider_url_for_request; @@ -612,6 +637,7 @@ mod tests { use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; use tower::ServiceExt; + use shared::model::ProcessingOrder; /// Generate an XMLTV datetime string in the format `YYYYMMDDHHmmss +0000` /// offset by `hours_from_now` hours from the current time. @@ -762,7 +788,7 @@ mod tests { mapping_ids: None, mapping: Arc::default(), favourites: None, - processing_order: Default::default(), + processing_order: ProcessingOrder::default(), watch: None, use_memory_cache: false, }); @@ -825,7 +851,7 @@ mod tests { mapping_ids: None, mapping: Arc::default(), favourites: None, - processing_order: Default::default(), + processing_order: ProcessingOrder::default(), watch: None, use_memory_cache: false, }); @@ -875,7 +901,7 @@ mod tests { mapping_ids: None, mapping: Arc::default(), favourites: None, - processing_order: Default::default(), + processing_order: ProcessingOrder::default(), watch: None, use_memory_cache: false, }); @@ -1062,7 +1088,7 @@ mod tests { assert_eq!(channels[0].id.as_ref(), "demo.channel"); assert_eq!(channels[0].programmes.len(), 1); assert_eq!( - channels[0].programmes[0].title.as_ref().map(|title| title.as_ref()), + channels[0].programmes[0].title.as_ref().map(std::convert::AsRef::as_ref), Some("Morning Show") ); } @@ -1184,6 +1210,7 @@ mod tests { } #[tokio::test] + #[allow(clippy::too_many_lines)] async fn playlist_epg_input_route_merges_multiple_cached_sources_by_priority() { let temp_dir = tempdir().expect("temp dir"); let provider = ConfigProvider::from(&ConfigProviderDto { @@ -1356,24 +1383,3 @@ mod tests { assert!(!body_text.contains("Secondary Show"), "{body_text}"); } } - -async fn playlist_episode_item( - axum::extract::Path(virtual_id): axum::extract::Path, - axum::extract::State(app_state): axum::extract::State>, - axum::extract::Json(playlist_req): axum::extract::Json, -) -> impl IntoResponse + Send { - if let PlaylistRequest::Target(target_id) = playlist_req { - if let Some(target) = app_state.app_config.get_target_by_id(target_id) { - if target.has_output(TargetType::Xtream) { - if let Ok(vid) = virtual_id.parse::() { - if let Ok(pli) = - xtream_get_item_for_stream_id(vid, &app_state, &target, Some(XtreamCluster::Series)).await - { - return axum::Json(json!(UiPlaylistItem::from(pli))).into_response(); - } - } - } - } - } - axum::http::StatusCode::NO_CONTENT.into_response() -} diff --git a/backend/src/api/endpoints/web_index.rs b/backend/src/api/endpoints/web_index.rs index 51d4121cb..36c0f214f 100644 --- a/backend/src/api/endpoints/web_index.rs +++ b/backend/src/api/endpoints/web_index.rs @@ -192,21 +192,6 @@ fn inject_nonce_with_parser(html: String, nonce_b64: &str) -> String { lol_html::rewrite_str(&html, settings).unwrap_or(html) } -#[cfg(test)] -mod tests { - use super::api_user_can_access_web_ui; - - #[test] - fn rejects_api_user_when_ui_is_disabled() { - assert!(!api_user_can_access_web_ui(false)); - } - - #[test] - fn allows_api_user_when_ui_is_enabled() { - assert!(api_user_can_access_web_ui(true)); - } -} - async fn index( axum::extract::State(app_state): axum::extract::State>, ) -> impl axum::response::IntoResponse + Send { @@ -432,3 +417,18 @@ pub fn index_register_with_path(web_dir_path: &Path, web_ui_path: &str) -> axum: ) .nest(&format!("/{web_ui_path}/"), web_ui_router) } + +#[cfg(test)] +mod tests { + use super::api_user_can_access_web_ui; + + #[test] + fn rejects_api_user_when_ui_is_disabled() { + assert!(!api_user_can_access_web_ui(false)); + } + + #[test] + fn allows_api_user_when_ui_is_enabled() { + assert!(api_user_can_access_web_ui(true)); + } +} diff --git a/backend/src/api/endpoints/websocket_api.rs b/backend/src/api/endpoints/websocket_api.rs index 999adecf0..c88a922bc 100644 --- a/backend/src/api/endpoints/websocket_api.rs +++ b/backend/src/api/endpoints/websocket_api.rs @@ -13,7 +13,7 @@ use log::{error, trace}; use shared::{ model::{ Permission, ProtocolHandler, ProtocolHandlerMemory, ProtocolMessage, UserCommand, UserRole, WsCloseCode, - PERM_ALL, ROLE_ADMIN, PROTOCOL_VERSION, + PERM_ALL, PROTOCOL_VERSION, ROLE_ADMIN, }, utils::{concat_path_leading_slash, default_kick_secs}, }; @@ -505,8 +505,10 @@ mod tests { #[test] fn test_websocket_runtime_events_denied_for_default_permissions() { - let mut mem = ProtocolHandlerMemory::default(); - mem.role = UserRole::User; + let mem = ProtocolHandlerMemory { + role: UserRole::User, + ..ProtocolHandlerMemory::default() + }; assert!(!websocket_can_receive_runtime_events( &mem, diff --git a/backend/src/api/main_api.rs b/backend/src/api/main_api.rs index f85d729b0..1358ee9c5 100644 --- a/backend/src/api/main_api.rs +++ b/backend/src/api/main_api.rs @@ -389,73 +389,6 @@ async fn wait_for_shutdown_signal() -> Result<&'static str, std::io::Error> { } } -#[cfg(not(unix))] -async fn wait_for_shutdown_signal() -> Result<&'static str, std::io::Error> { - tokio::signal::ctrl_c().await?; - Ok("Ctrl+C") -} - -#[cfg(test)] -mod tests { - use super::{corrupt_downloads_state_path, recover_persisted_downloads_state}; - use crate::api::model::DownloadQueue; - use std::{path::PathBuf, time::{SystemTime, UNIX_EPOCH}}; - - fn temp_state_file(name: &str) -> PathBuf { - let nanos = SystemTime::now() - .duration_since(UNIX_EPOCH) - .expect("time") - .as_nanos(); - std::env::temp_dir().join(format!("tuliprox_{name}_{nanos}.json")) - } - - #[test] - fn corrupt_downloads_state_path_is_unique_for_rapid_calls() { - let state_file = temp_state_file("corrupt_downloads_unique"); - - let first = corrupt_downloads_state_path(&state_file); - let second = corrupt_downloads_state_path(&state_file); - - assert_ne!(first, second); - } - - #[tokio::test] - async fn recover_persisted_downloads_state_renames_corrupt_file_and_continues() { - let state_file = temp_state_file("corrupt_downloads_state"); - std::fs::write(&state_file, "{ not valid json").expect("write corrupt state"); - let corrupt_dir = state_file.parent().expect("state dir").to_path_buf(); - let expected_prefix = format!( - "{}_corrupt.", - state_file.file_stem().and_then(|stem| stem.to_str()).expect("state stem") - ); - let expected_extension = state_file.extension().and_then(|ext| ext.to_str()).expect("state ext"); - let downloads = DownloadQueue::new_with_state_file(Some(state_file.clone())); - - recover_persisted_downloads_state(&downloads).await.expect("recovery should continue"); - - assert!(!state_file.exists()); - let matching_corrupt_paths = std::fs::read_dir(&corrupt_dir) - .expect("read corrupt dir") - .filter_map(Result::ok) - .map(|entry| entry.path()) - .filter(|path| { - path.file_name() - .and_then(|name| name.to_str()) - .is_some_and(|name| name.starts_with(&expected_prefix) && name.ends_with(expected_extension)) - }) - .collect::>(); - assert_eq!(matching_corrupt_paths.len(), 1); - assert!(downloads.queue.lock().await.is_empty()); - assert!(downloads.scheduled.read().await.is_empty()); - assert!(downloads.active.read().await.is_none()); - assert!(downloads.finished.read().await.is_empty()); - - for path in matching_corrupt_paths { - let _ = std::fs::remove_file(path); - } - } -} - fn exec_update_on_boot(client: &reqwest::Client, app_state: &Arc, targets: &Arc) -> bool { let cfg = &app_state.app_config; let update_on_boot = { @@ -898,3 +831,70 @@ fn exec_input_update_listener(app_state: &Arc, targets: &Arc Result<&'static str, std::io::Error> { + tokio::signal::ctrl_c().await?; + Ok("Ctrl+C") +} + +#[cfg(test)] +mod tests { + use super::{corrupt_downloads_state_path, recover_persisted_downloads_state}; + use crate::api::model::DownloadQueue; + use std::{path::PathBuf, time::{SystemTime, UNIX_EPOCH}}; + + fn temp_state_file(name: &str) -> PathBuf { + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("time") + .as_nanos(); + std::env::temp_dir().join(format!("tuliprox_{name}_{nanos}.json")) + } + + #[test] + fn corrupt_downloads_state_path_is_unique_for_rapid_calls() { + let state_file = temp_state_file("corrupt_downloads_unique"); + + let first = corrupt_downloads_state_path(&state_file); + let second = corrupt_downloads_state_path(&state_file); + + assert_ne!(first, second); + } + + #[tokio::test] + async fn recover_persisted_downloads_state_renames_corrupt_file_and_continues() { + let state_file = temp_state_file("corrupt_downloads_state"); + std::fs::write(&state_file, "{ not valid json").expect("write corrupt state"); + let corrupt_dir = state_file.parent().expect("state dir").to_path_buf(); + let expected_prefix = format!( + "{}_corrupt.", + state_file.file_stem().and_then(|stem| stem.to_str()).expect("state stem") + ); + let expected_extension = state_file.extension().and_then(|ext| ext.to_str()).expect("state ext"); + let downloads = DownloadQueue::new_with_state_file(Some(state_file.clone())); + + recover_persisted_downloads_state(&downloads).await.expect("recovery should continue"); + + assert!(!state_file.exists()); + let matching_corrupt_paths = std::fs::read_dir(&corrupt_dir) + .expect("read corrupt dir") + .filter_map(Result::ok) + .map(|entry| entry.path()) + .filter(|path| { + path.file_name() + .and_then(|name| name.to_str()) + .is_some_and(|name| name.starts_with(&expected_prefix) && name.ends_with(expected_extension)) + }) + .collect::>(); + assert_eq!(matching_corrupt_paths.len(), 1); + assert!(downloads.queue.lock().await.is_empty()); + assert!(downloads.scheduled.read().await.is_empty()); + assert!(downloads.active.read().await.is_none()); + assert!(downloads.finished.read().await.is_empty()); + + for path in matching_corrupt_paths { + let _ = std::fs::remove_file(path); + } + } +} diff --git a/backend/src/api/model/active_provider_manager.rs b/backend/src/api/model/active_provider_manager.rs index 1c8d6ccb8..4b4ca0abb 100644 --- a/backend/src/api/model/active_provider_manager.rs +++ b/backend/src/api/model/active_provider_manager.rs @@ -2152,7 +2152,7 @@ mod tests { { let connections = manager.connections.read().await; let tree = connections.priority_index.get(&input_name); - let is_empty = tree.is_none_or(|t| t.is_empty()); + let is_empty = tree.is_none_or(std::collections::BTreeMap::is_empty); assert!(is_empty, "index should be empty after releasing all connections"); } diff --git a/backend/src/api/model/active_user_manager.rs b/backend/src/api/model/active_user_manager.rs index f5c224c17..82165e9d5 100644 --- a/backend/src/api/model/active_user_manager.rs +++ b/backend/src/api/model/active_user_manager.rs @@ -3863,6 +3863,7 @@ mod tests { } #[tokio::test] + #[allow(clippy::too_many_lines)] async fn vod_session_survives_overlapping_and_seek_sockets() { let config = Config::default(); let geoip = Arc::new(ArcSwapOption::::default()); @@ -4100,6 +4101,7 @@ mod tests { } #[tokio::test] + #[allow(clippy::too_many_lines)] async fn binding_reserved_sessions_keeps_hard_and_soft_counts_stable() { let config = Config::default(); let geoip = Arc::new(ArcSwapOption::::default()); diff --git a/backend/src/api/model/admission_strategy.rs b/backend/src/api/model/admission_strategy.rs index cddf778db..c61e2a10f 100644 --- a/backend/src/api/model/admission_strategy.rs +++ b/backend/src/api/model/admission_strategy.rs @@ -177,7 +177,7 @@ mod tests { use std::net::{IpAddr, Ipv4Addr}; fn addr(port: u16) -> SocketAddr { - SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), port) + SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), port) } fn candidate(port: u16, ip: &str, ts: u64) -> EvictionCandidate { diff --git a/backend/src/api/model/app_state.rs b/backend/src/api/model/app_state.rs index 1ededf467..2b5004785 100644 --- a/backend/src/api/model/app_state.rs +++ b/backend/src/api/model/app_state.rs @@ -1,16 +1,17 @@ +use crate::utils::LRUResourceCache; use crate::{ api::{ config_watch::exec_config_watch, endpoints::download_api::{resume_download_worker_if_needed, spawn_download_services}, + model::active_user_manager::ConnectionAdmission, model::provider_dns_manager::exec_provider_dns, model::{ - qos_aggregation_manager::exec_qos_aggregation, - metadata_update_manager::MetadataUpdateManager, ActiveProviderManager, ActiveUserManager, + metadata_update_manager::MetadataUpdateManager, + qos_aggregation_manager::exec_qos_aggregation, ActiveProviderManager, ActiveUserManager, ConnectionManager, DownloadQueue, EventManager, PlaylistStorage, PlaylistStorageState, SharedStreamManager, UpdateGuard, }, scheduler::exec_scheduler, - model::active_user_manager::ConnectionAdmission, }, model::{ AppConfig, Config, ConfigProvider, ConfigTarget, GracePeriodOptions, HdHomeRunConfig, HdHomeRunDeviceConfig, @@ -40,7 +41,6 @@ use tokio::sync::{mpsc, Mutex}; use tokio::task; use tokio_util::sync::CancellationToken; use url::Url; -use crate::utils::LRUResourceCache; macro_rules! cancel_service { ($field: ident, $flag:expr, $changes:expr, $cancel_tokens:expr) => { @@ -663,7 +663,7 @@ impl AppState { || proxy_env_present() } - pub fn get_encrypt_secret(&self) -> [u8;16] { + pub fn get_encrypt_secret(&self) -> [u8; 16] { self.app_config .get_reverse_proxy_rewrite_secret() .unwrap_or(self.app_config.encrypt_secret) @@ -705,7 +705,7 @@ fn should_use_manual_redirect_for_proxy(proxy_url: &str) -> bool { fn should_use_manual_redirects_for_env_vars(vars: I) -> bool where - I: IntoIterator, + I: IntoIterator, K: AsRef, V: AsRef, { @@ -956,17 +956,19 @@ mod tests { #[test] fn qos_aggregation_changed_detects_stream_history_batch_size_changes() { - let mut old_config = Config::default(); - old_config.reverse_proxy = Some(crate::model::ReverseProxyConfig::from(&ReverseProxyConfigDto { - stream_history: Some(StreamHistoryConfigDto { - stream_history_enabled: true, - stream_history_batch_size: 64, - stream_history_retention_days: 7, - stream_history_directory: "/tmp/history".to_string(), - }), - qos_aggregation: Some(QosAggregationConfigDto { enabled: true, interval_secs: 60 }), - ..Default::default() - })); + let old_config = Config { + reverse_proxy: Some(crate::model::ReverseProxyConfig::from(&ReverseProxyConfigDto { + stream_history: Some(StreamHistoryConfigDto { + stream_history_enabled: true, + stream_history_batch_size: 64, + stream_history_retention_days: 7, + stream_history_directory: "/tmp/history".to_string(), + }), + qos_aggregation: Some(QosAggregationConfigDto { enabled: true, interval_secs: 60 }), + ..Default::default() + })), + ..Config::default() + }; let mut new_config = old_config.clone(); if let Some(reverse_proxy) = new_config.reverse_proxy.as_mut() { diff --git a/backend/src/api/model/connection_manager.rs b/backend/src/api/model/connection_manager.rs index b21c28880..1cae8d477 100644 --- a/backend/src/api/model/connection_manager.rs +++ b/backend/src/api/model/connection_manager.rs @@ -1248,9 +1248,10 @@ mod tests { sender.enqueue(2_u8); for _ in 0..50 { - let state = super::lock_backpressure_state(sender.state.as_ref()); - let ready = state.overflow.is_empty() && state.draining; - drop(state); + let ready = { + let state = super::lock_backpressure_state(sender.state.as_ref()); + state.overflow.is_empty() && state.draining + }; if ready { break; } diff --git a/backend/src/api/model/metadata_update_manager.rs b/backend/src/api/model/metadata_update_manager.rs index 8353fd01b..27e415bca 100644 --- a/backend/src/api/model/metadata_update_manager.rs +++ b/backend/src/api/model/metadata_update_manager.rs @@ -17,6 +17,7 @@ use dashmap::{mapref::entry::Entry, DashMap}; use log::{debug, error, info, warn}; use parking_lot::Mutex as ParkingMutex; use serde::{Deserialize, Serialize}; +use shared::utils::default_probe_user_priority; use shared::{ create_bitset, error::TuliproxError, @@ -39,7 +40,6 @@ use std::{ }; use tokio::sync::{mpsc, RwLock, Semaphore}; use tokio_util::sync::CancellationToken; -use shared::utils::default_probe_user_priority; const METADATA_RETRY_STATE_FILE: &str = "metadata_retry_state.db"; const TASK_ERR_NO_CONNECTION: &str = "No connection available"; @@ -603,7 +603,6 @@ enum SubmitTaskResult { } impl MetadataUpdateManager { - pub fn new(cancel_token: CancellationToken) -> Self { Self { workers: DashMap::new(), @@ -945,7 +944,7 @@ impl MetadataUpdateManager { max_queue_size, task_to_queue.clone(), ) - .await + .await { SubmitTaskResult::QueuedOrMerged => return, SubmitTaskResult::QueueFull => { @@ -1490,7 +1489,7 @@ impl InputWorker { &mut self.db_handles, &mut self.failed_clusters, ) - .await + .await }; match task_result { @@ -2616,7 +2615,7 @@ impl InputWorker { input_name, updates, ) - .await + .await { error!("Failed to flush VOD batch for input {input_name}: {e}"); } @@ -2639,7 +2638,7 @@ impl InputWorker { input_name, updates, ) - .await + .await { error!("Failed to flush Series batch for input {input_name}: {e}"); } @@ -2662,7 +2661,7 @@ impl InputWorker { input_name, updates, ) - .await + .await { error!("Failed to flush Live batch for input {input_name}: {e}"); } @@ -2911,7 +2910,7 @@ impl InputWorker { } updates }) - .await + .await { Ok(updates) => updates, Err(err) => { @@ -2969,7 +2968,7 @@ impl InputWorker { } updates }) - .await + .await { Ok(updates) => updates, Err(err) => { @@ -3027,7 +3026,7 @@ impl InputWorker { } updates }) - .await + .await { Ok(updates) => updates, Err(err) => { @@ -3094,7 +3093,7 @@ impl InputWorker { let query = match spawn_blocking_limited(move || { BPlusTreeQuery::::try_new(&file_path) }) - .await + .await { Ok(Ok(query)) => Some(query), Ok(Err(err)) => { @@ -3146,7 +3145,7 @@ impl InputWorker { let mut guard = query.lock(); guard.query_zero_copy(&stream_id).ok().flatten() }) - .await + .await { Ok(item) => item, Err(err) => { @@ -3276,7 +3275,7 @@ impl InputWorker { db_handles, failed_clusters, ) - .await + .await } } else { Self::execute_task_inner_static( @@ -3291,7 +3290,7 @@ impl InputWorker { db_handles, failed_clusters, ) - .await + .await } }; let needs_probe_timeout = Self::is_probe_task(task) @@ -3407,7 +3406,7 @@ impl InputWorker { Some(&tmdb_and_date_present), Some(&probe_pending), ) - .await + .await { Ok(Some(props)) => { collector.add_vod(id.clone(), props); @@ -3463,7 +3462,7 @@ impl InputWorker { Some(&tmdb_and_date_present), Some(&probe_pending), ) - .await + .await { Ok(Some(props)) => { collector.add_series(id.clone(), props); @@ -3492,7 +3491,7 @@ impl InputWorker { active_handle, &app_state.active_provider, ) - .await + .await { Ok(Some(props)) => { collector.add_live(id.clone(), props); @@ -3521,13 +3520,13 @@ impl InputWorker { active_handle, probe_priority, ) - .await?; + .await?; match outcome { GenericProbeOutcome::Updated | GenericProbeOutcome::Noop => Ok((false, false)), - GenericProbeOutcome::ProbeFailed => Err(shared::error::TuliproxError::Config(format!( + GenericProbeOutcome::ProbeFailed => Err(shared::error::TuliproxError::Config(format!( "Probe stream task failed for key {task_key:?} ({probe_identifier})" - ))), + ))), } } } @@ -3796,7 +3795,7 @@ mod tests { queue_size, task_initial, ) - .await; + .await; let task_merge = UpdateTask::ResolveVod { id: ProviderIdType::Id(42), @@ -3812,7 +3811,7 @@ mod tests { queue_size, task_merge, ) - .await; + .await; let first_signal = rx.try_recv().expect("first signal should be queued"); assert_eq!(first_signal, TaskKey::Vod(42)); @@ -3856,7 +3855,7 @@ mod tests { queue_size, initial, ) - .await; + .await; let identical_merge = UpdateTask::ResolveVod { id: ProviderIdType::Id(42), @@ -3872,7 +3871,7 @@ mod tests { queue_size, identical_merge, ) - .await; + .await; let first_signal = rx.try_recv().expect("first signal should be queued"); assert_eq!(first_signal, TaskKey::Vod(42)); @@ -3908,7 +3907,7 @@ mod tests { queue_size, initial, ) - .await; + .await; let merged_in = UpdateTask::ProbeStream { probe_scope: Arc::from("scope_a"), @@ -3926,7 +3925,7 @@ mod tests { queue_size, merged_in, ) - .await; + .await; let first_signal = rx.try_recv().expect("first signal should be queued"); assert_eq!(first_signal, TaskKey::Stream { scope: Arc::from("scope_a"), id: Arc::from("uid_1") }); @@ -3983,7 +3982,7 @@ mod tests { MetadataUpdateRuntimeSettings::default().max_queue_size, incoming_task, ) - .await; + .await; assert_eq!(result, SubmitTaskResult::ChannelClosed); assert!(!pending_tasks.contains_key(&key)); @@ -4103,8 +4102,10 @@ mod tests { } let mut worker = create_test_worker("input_d", tx, rx, pending_tasks.clone(), pending_task_count); - let mut runtime_settings = MetadataUpdateRuntimeSettings::default(); - runtime_settings.no_change_cache_ttl_secs = 1; + let runtime_settings = MetadataUpdateRuntimeSettings { + no_change_cache_ttl_secs: 1, + ..MetadataUpdateRuntimeSettings::default() + }; let stale_instant = Instant::now().checked_sub(Duration::from_secs(2)).unwrap_or_else(Instant::now); worker.recently_completed_no_change.insert( diff --git a/backend/src/api/model/qos_aggregation_manager.rs b/backend/src/api/model/qos_aggregation_manager.rs index e8078328f..161c15dea 100644 --- a/backend/src/api/model/qos_aggregation_manager.rs +++ b/backend/src/api/model/qos_aggregation_manager.rs @@ -12,13 +12,13 @@ use std::time::Duration; use crate::api::model::AppState; use crate::model::{Config, StreamHistoryRecord}; use crate::repository::{ - extract_day_from_filename, - QosSnapshotDailyBucket, QosSnapshotRecord, QosSnapshotRepository, QosSnapshotWindow, + extract_day_from_filename, + QosSnapshotDailyBucket, QosSnapshotRecord, QosSnapshotRepository, QosSnapshotWindow, StreamHistoryFileReader, }; -use tokio_util::sync::CancellationToken; -use shared::model::{ConnectFailureReason, DisconnectReason, FailureStage, PlaylistItemType, StreamHistoryEventType}; use crate::utils::{current_utc_day, now_utc_secs}; +use shared::model::{ConnectFailureReason, DisconnectReason, FailureStage, PlaylistItemType, StreamHistoryEventType}; +use tokio_util::sync::CancellationToken; const MAX_COMPLETED_DAY_PARTITIONS_PER_RUN: usize = 3; @@ -41,10 +41,10 @@ pub(in crate::api) fn qos_aggregation_is_enabled(config: &Config) -> bool { .is_some_and(|qos| { qos.enabled && config - .reverse_proxy - .as_ref() - .and_then(|rp| rp.stream_history.as_ref()) - .is_some_and(|history| history.stream_history_enabled) + .reverse_proxy + .as_ref() + .and_then(|rp| rp.stream_history.as_ref()) + .is_some_and(|history| history.stream_history_enabled) }) } @@ -197,7 +197,7 @@ pub(crate) fn fold_record_into_bucket(bucket: &mut QosSnapshotDailyBucket, recor bucket.last_failure_ts = Some(bucket.last_failure_ts.map_or(record.event_ts_utc, |ts| ts.max(record.event_ts_utc))); } - }, + } StreamHistoryEventType::Failure => { // Failure events represent intermediate disconnects within an HLS session. // Count them as runtime aborts since they reflect streaming failures. @@ -544,22 +544,24 @@ struct HistoryDayFile { mod tests { use std::collections::BTreeMap; - use tempfile::tempdir; - use shared::model::{FailureStage, ConnectFailureReason, DisconnectReason, PlaylistItemType, StreamHistoryEventType}; - use shared::utils::Internable; - use crate::model::{Config, QosAggregationConfig, ReverseProxyConfig, StreamHistoryConfig, StreamHistoryRecord, RECORD_SCHEMA_VERSION}; + use super::{fold_record_into_bucket, history_day_revision, qos_aggregation_is_enabled, rebuild_windows, run_aggregation_once}; + use crate::model::{Config, QosAggregationConfig, ResourceRetryConfig, ReverseProxyConfig, StreamHistoryConfig, StreamHistoryRecord, RECORD_SCHEMA_VERSION}; use crate::repository::{ serialize_named, write_block_magic, write_file_magic, write_framed, BlockHeaderBody, - CompressionKind, CONTAINER_FORMAT_VERSION, - FileHeaderBody, QosSnapshotDailyBucket, QosSnapshotRecord, QosSnapshotRepository, QosSnapshotWindow, - RecordEncodingKind, SOURCE_KIND_STREAM_HISTORY, + CompressionKind, FileHeaderBody, + QosSnapshotDailyBucket, QosSnapshotRecord, QosSnapshotRepository, QosSnapshotWindow, RecordEncodingKind, + CONTAINER_FORMAT_VERSION, SOURCE_KIND_STREAM_HISTORY, }; use crate::utils::current_utc_day; - use super::{fold_record_into_bucket, history_day_revision, qos_aggregation_is_enabled, rebuild_windows, run_aggregation_once}; + use shared::model::{ConnectFailureReason, DisconnectReason, FailureStage, PlaylistItemType, StreamHistoryEventType}; + use shared::utils::Internable; + use tempfile::tempdir; + use std::io::Write as _; + fn write_pending_history_records( history_dir: &std::path::Path, - records: Vec, + records: &[StreamHistoryRecord], ) { if records.is_empty() { return; @@ -593,7 +595,7 @@ mod tests { let mut payload = Vec::new(); let mut first_ts = u64::MAX; let mut last_ts = 0_u64; - for record in &records { + for record in records { first_ts = first_ts.min(record.event_ts_utc); last_ts = last_ts.max(record.event_ts_utc); let encoded = serialize_named(record).expect("record should serialize"); @@ -613,7 +615,6 @@ mod tests { }; write_block_magic(&mut file).expect("block magic should write"); write_framed(&mut file, &block_header).expect("block header should write"); - use std::io::Write as _; file.write_all(&payload).expect("payload should write"); file.flush().expect("pending file should flush"); } @@ -723,7 +724,7 @@ mod tests { let temp = tempdir().expect("tempdir should succeed"); let history_dir = temp.path(); let record = base_record(StreamHistoryEventType::Connect); - write_pending_history_records(history_dir, vec![record.clone(), record]); + write_pending_history_records(history_dir, &[record.clone(), record]); let history_dir_copy = history_dir.to_path_buf(); let revision = tokio::task::spawn_blocking(move || history_day_revision(&history_dir_copy, "2026-04-02")) @@ -791,25 +792,27 @@ mod tests { #[test] fn qos_aggregation_is_enabled_requires_stream_history() { - let mut config = Config::default(); - config.reverse_proxy = Some(ReverseProxyConfig { - resource_rewrite_disabled: false, - rewrite_secret: [0; 16], - resource_retry: Default::default(), - disabled_header: None, - stream: None, - cache: None, - rate_limit: None, - geoip: None, - stream_history: Some(StreamHistoryConfig { - stream_history_enabled: false, - ..Default::default() + let mut config = Config { + reverse_proxy: Some(ReverseProxyConfig { + resource_rewrite_disabled: false, + rewrite_secret: [0; 16], + resource_retry: ResourceRetryConfig::default(), + disabled_header: None, + stream: None, + cache: None, + rate_limit: None, + geoip: None, + stream_history: Some(StreamHistoryConfig { + stream_history_enabled: false, + ..Default::default() + }), + qos_aggregation: Some(QosAggregationConfig { + enabled: true, + interval_secs: 300, + }), }), - qos_aggregation: Some(QosAggregationConfig { - enabled: true, - interval_secs: 300, - }), - }); + ..Config::default() + }; assert!(!qos_aggregation_is_enabled(&config)); @@ -836,14 +839,14 @@ mod tests { connect.target_name = Some("target-a".intern()); connect.stream_identity_key = Some("stream-a".to_string()); connect.shared_joined_existing = Some(false); - write_pending_history_records(&history_dir, vec![connect]); + write_pending_history_records(&history_dir, &[connect]); let repo = std::sync::Arc::new(QosSnapshotRepository::open(temp.path()).expect("repo should open")); - + let repo_c = repo.clone(); let h_dir = history_dir.clone(); - let t_day = today.clone(); - tokio::task::spawn_blocking(move || run_aggregation_once(&*repo_c, &h_dir, &t_day)) + let now = today.clone(); + tokio::task::spawn_blocking(move || run_aggregation_once(&repo_c, &h_dir, &now)) .await.unwrap().expect("aggregation should succeed"); let snapshot = repo @@ -870,20 +873,20 @@ mod tests { connect.target_name = Some("target-a".intern()); connect.stream_identity_key = Some("stream-a".to_string()); connect.shared_joined_existing = Some(false); - write_pending_history_records(&history_dir, vec![connect]); + write_pending_history_records(&history_dir, &[connect]); let repo = std::sync::Arc::new(QosSnapshotRepository::open(temp.path()).expect("repo should open")); let repo_c1 = repo.clone(); let h_dir1 = history_dir.clone(); let t_day1 = today.clone(); - tokio::task::spawn_blocking(move || run_aggregation_once(&*repo_c1, &h_dir1, &t_day1)) + tokio::task::spawn_blocking(move || run_aggregation_once(&repo_c1, &h_dir1, &t_day1)) .await.unwrap().expect("first aggregation should succeed"); let repo_c2 = repo.clone(); let h_dir2 = history_dir.clone(); let t_day2 = today.clone(); - tokio::task::spawn_blocking(move || run_aggregation_once(&*repo_c2, &h_dir2, &t_day2)) + tokio::task::spawn_blocking(move || run_aggregation_once(&repo_c2, &h_dir2, &t_day2)) .await.unwrap().expect("second aggregation should succeed"); let snapshot = repo @@ -917,16 +920,16 @@ mod tests { connect.target_name = Some("target-a".intern()); connect.stream_identity_key = Some("stream-a".to_string()); connect.shared_joined_existing = Some(false); - write_pending_history_records(&history_dir, vec![connect]); + write_pending_history_records(&history_dir, &[connect]); } let repo = std::sync::Arc::new(QosSnapshotRepository::open(temp.path()).expect("repo should open")); let repo_c1 = repo.clone(); let h_dir1 = history_dir.clone(); - tokio::task::spawn_blocking(move || run_aggregation_once(&*repo_c1, &h_dir1, "2036-04-05")) + tokio::task::spawn_blocking(move || run_aggregation_once(&repo_c1, &h_dir1, "2036-04-05")) .await.unwrap().expect("first aggregation should succeed"); - + let checkpoint = repo.load_checkpoint().expect("checkpoint should load"); assert_eq!(checkpoint.last_completed_day_utc.as_deref(), Some("2036-04-03")); @@ -939,9 +942,9 @@ mod tests { let repo_c2 = repo.clone(); let h_dir2 = history_dir.clone(); - tokio::task::spawn_blocking(move || run_aggregation_once(&*repo_c2, &h_dir2, "2036-04-05")) + tokio::task::spawn_blocking(move || run_aggregation_once(&repo_c2, &h_dir2, "2036-04-05")) .await.unwrap().expect("second aggregation should succeed"); - + let checkpoint = repo.load_checkpoint().expect("checkpoint should load"); assert_eq!(checkpoint.last_completed_day_utc.as_deref(), Some("2036-04-04")); diff --git a/backend/src/api/model/request.rs b/backend/src/api/model/request.rs index 7c3c0fdbf..4721fda1e 100644 --- a/backend/src/api/model/request.rs +++ b/backend/src/api/model/request.rs @@ -403,7 +403,7 @@ mod tests { super::ParseBodyError::PayloadTooLarge(msg) => { assert!(msg.contains("body too large"), "unexpected error: {msg}"); } - other => panic!("unexpected error: {other:?}"), + other@ super::ParseBodyError::BadRequest(_) => panic!("unexpected error: {other:?}"), } } @@ -416,11 +416,7 @@ mod tests { .body(Body::from(oversized)) .expect("request should build"); - let response = match super::UserApiRequestQueryOrBody::from_request(request, &()).await { - Ok(_) => panic!("oversized body should reject extractor"), - Err(response) => response, - }; - + let Err(response) = super::UserApiRequestQueryOrBody::from_request(request, &()).await else { panic!("oversized body should reject extractor") }; assert_eq!(response.status(), StatusCode::PAYLOAD_TOO_LARGE); } diff --git a/backend/src/api/model/streams/shared_stream_manager.rs b/backend/src/api/model/streams/shared_stream_manager.rs index 4f703f0f7..d7c77cb0b 100644 --- a/backend/src/api/model/streams/shared_stream_manager.rs +++ b/backend/src/api/model/streams/shared_stream_manager.rs @@ -945,10 +945,7 @@ mod tests { drop(state); let first = timeout(Duration::from_secs(2), stream.next()).await; - let maybe_chunk = match first { - Ok(v) => v, - Err(_) => panic!("timed out waiting for fallback chunk"), - }; + let Ok(maybe_chunk) = first else { panic!("timed out waiting for fallback chunk") }; let chunk = match maybe_chunk { Some(Ok(bytes)) => bytes, Some(Err(err)) => panic!("fallback stream returned error: {err}"), diff --git a/backend/src/api/sys_usage.rs b/backend/src/api/sys_usage.rs index e96cde882..4b90d84fa 100644 --- a/backend/src/api/sys_usage.rs +++ b/backend/src/api/sys_usage.rs @@ -658,7 +658,7 @@ mod tests { fn test_net_tracker_reports_bytes_per_second() { let mut tracker = super::NetTracker::new(); let _ = tracker.sample(1000, 500); - tracker.last_sample_at = tracker.last_sample_at.map(|instant| instant - Duration::from_secs(2)); + tracker.last_sample_at = tracker.last_sample_at.map(|instant| instant.checked_sub(Duration::from_secs(2)).unwrap()); let (rx_rate, tx_rate) = tracker.sample(3000, 1500); assert!((999.0..=1001.0).contains(&rx_rate)); assert!((499.0..=501.0).contains(&tx_rate)); diff --git a/backend/src/library/classifier.rs b/backend/src/library/classifier.rs index aecdce369..ae88b5c07 100644 --- a/backend/src/library/classifier.rs +++ b/backend/src/library/classifier.rs @@ -210,7 +210,7 @@ mod tests { assert_eq!(season, 2); assert_eq!(episode, 5); } - _ => panic!("Expected Series classification"), + MediaClassification::Movie { .. } => panic!("Expected Series classification"), } // Counter should record next episode after the parsed one (key is lowercased) assert_eq!(counters[&("breaking bad".to_string(), 2)], 6); @@ -259,15 +259,15 @@ mod tests { // ShowA and ShowB each start at episode 1 independently match ca1 { MediaClassification::Series { episode, .. } => assert_eq!(episode, 1), - _ => panic!("Expected Series"), + MediaClassification::Movie { .. } => panic!("Expected Series"), } match cb1 { MediaClassification::Series { episode, .. } => assert_eq!(episode, 1), - _ => panic!("Expected Series"), + MediaClassification::Movie { .. } => panic!("Expected Series"), } match ca2 { MediaClassification::Series { episode, .. } => assert_eq!(episode, 2), - _ => panic!("Expected Series"), + MediaClassification::Movie { .. } => panic!("Expected Series"), } } diff --git a/backend/src/library/processor.rs b/backend/src/library/processor.rs index 0fc9f3710..e89ef6173 100644 --- a/backend/src/library/processor.rs +++ b/backend/src/library/processor.rs @@ -3,7 +3,7 @@ use crate::library::metadata::{EpisodeMetadata, MediaMetadata, MetadataCacheEntr use crate::library::metadata_resolver::MetadataResolver; use crate::library::metadata_storage::MetadataStorage; use crate::library::scanner::LibraryScanner; -use crate::library::{MediaGroup, MediaGrouper, thumbnail::{self, ThumbnailExtractor}}; +use crate::library::{thumbnail::{self, ThumbnailExtractor}, MediaGroup, MediaGrouper}; use crate::model::{AppConfig, LibraryConfig, MetadataUpdateConfig}; use crate::utils::ffmpeg::{FfmpegExecutor, ProbeUrlOutcome}; use log::{debug, error, info, warn}; @@ -15,7 +15,7 @@ use std::path::PathBuf; use std::pin::Pin; use std::sync::Arc; -type FfprobeAvailabilityFuture = Pin + Send>>; +type FfprobeAvailabilityFuture = Pin + Send>>; type FfprobeAvailabilityChecker = Arc FfprobeAvailabilityFuture + Send + Sync>; // Action taken when processing a file @@ -238,22 +238,22 @@ impl LibraryProcessor { // TODO: Implement enrich_metadata_with_ffprobe to add technical info (resolution, codecs) from local files //fn enrich_metadata_with_ffprobe(&self, _metadata: &mut MediaMetadata, _file_path: &str, _can_probe: bool) { - //if !can_probe { return; } + //if !can_probe { return; } - // let _url = format!("file://{file_path}"); // Simple file URL for ffmpeg - // - // // TODO: Logic for series episodes iteration - // match metadata { - // MediaMetadata::Movie(_movie) => { - // // Currently we don't have fields in MovieMetadata to store tech info - // // But we could add them. For now, let's just log. - // // In the future this should update the metadata struct. - // debug!("Probe logic for local movie {} not yet fully integrated into Metadata struct", file_path); - // } - // MediaMetadata::Series(_) => { - // // Series handle episodes separately - // } - // } + // let _url = format!("file://{file_path}"); // Simple file URL for ffmpeg + // + // // TODO: Logic for series episodes iteration + // match metadata { + // MediaMetadata::Movie(_movie) => { + // // Currently we don't have fields in MovieMetadata to store tech info + // // But we could add them. For now, let's just log. + // // In the future this should update the metadata struct. + // debug!("Probe logic for local movie {} not yet fully integrated into Metadata struct", file_path); + // } + // MediaMetadata::Series(_) => { + // // Series handle episodes separately + // } + // } //} // Processes a single video file @@ -783,11 +783,13 @@ mod tests { ); processor.app_config = None; - let mut metadata_update = MetadataUpdateConfig::default(); - metadata_update.ffprobe = crate::model::FfprobeConfig::from(&FfprobeConfigDto { - enabled: true, - ..FfprobeConfigDto::default() - }); + let metadata_update = MetadataUpdateConfig { + ffprobe: crate::model::FfprobeConfig::from(&FfprobeConfigDto { + enabled: true, + ..FfprobeConfigDto::default() + }), + ..MetadataUpdateConfig::default() + }; processor.metadata_update_config = Some(metadata_update); assert!(!processor.is_local_ffprobe_enabled().await); diff --git a/backend/src/model/config/input.rs b/backend/src/model/config/input.rs index 9abff6bda..4be586d53 100644 --- a/backend/src/model/config/input.rs +++ b/backend/src/model/config/input.rs @@ -2,6 +2,7 @@ use crate::model::{macros, ConfigProvider, EpgConfig, PanelApiConfig}; use crate::repository::get_csv_file_path; use chrono::Utc; use log::warn; +use shared::foundation::Filter; use shared::{apply_flags, create_bitset}; use shared::error::TuliproxError; use shared::model::{ClusterSource, ConfigInputAliasDto, ConfigInputDto, ConfigInputOptionsDto, InputFetchMethod, InputType, StagedInputDto, XtreamCluster}; @@ -41,6 +42,8 @@ pub struct ConfigInputOptions { pub resolve_delay: u16, pub probe_delay: u16, pub probe_live_interval_hours: u32, + pub resolve_filter: Option, + pub probe_filter: Option, } macros::from_impl!(ConfigInputOptions); @@ -87,6 +90,8 @@ impl From<&ConfigInputOptionsDto> for ConfigInputOptions { resolve_delay: dto.resolve_delay, probe_delay: dto.probe_delay, probe_live_interval_hours: dto.probe_live_interval_hours, + resolve_filter: dto.t_resolve_filter.clone(), + probe_filter: dto.t_probe_filter.clone(), } } } @@ -1113,7 +1118,7 @@ mod tests { let mut input = ConfigInput { name: "xtream_batch_missing_root_url".into(), input_type: InputType::XtreamBatch, - url: "".to_string(), + url: String::new(), enabled: true, aliases: Some(vec![ConfigInputAlias { id: 1, diff --git a/backend/src/processing/processor/library.rs b/backend/src/processing/processor/library.rs index 6a36c2c1b..1565cbf01 100644 --- a/backend/src/processing/processor/library.rs +++ b/backend/src/processing/processor/library.rs @@ -552,9 +552,9 @@ mod tests { let seasons = series.details.as_ref().and_then(|details| details.seasons.as_ref()).expect("seasons missing"); let season = seasons.first().expect("season missing"); - assert_eq!(season.cover.as_deref().map(|v| v.as_ref()), Some("/api/v1/library/thumbnail/series-thumb-uuid")); - assert_eq!(season.cover_tmdb.as_deref().map(|v| v.as_ref()), Some("/api/v1/library/thumbnail/series-thumb-uuid")); - assert_eq!(season.cover_big.as_deref().map(|v| v.as_ref()), Some("/api/v1/library/thumbnail/series-thumb-uuid")); + assert_eq!(season.cover.as_deref(), Some("/api/v1/library/thumbnail/series-thumb-uuid")); + assert_eq!(season.cover_tmdb.as_deref(), Some("/api/v1/library/thumbnail/series-thumb-uuid")); + assert_eq!(season.cover_big.as_deref(), Some("/api/v1/library/thumbnail/series-thumb-uuid")); assert!(season.overview.is_none()); } @@ -612,11 +612,11 @@ mod tests { }; let details = video.details.as_ref().expect("details missing"); - assert_eq!(details.video.as_deref().map(|value| value.as_ref()), Some("{\"codec_name\":\"hevc\"}")); - assert_eq!(details.audio.as_deref().map(|value| value.as_ref()), Some("{\"codec_name\":\"aac\"}")); - assert_eq!(details.duration_secs.as_deref().map(|value| value.as_ref()), Some("5430")); - assert_eq!(details.duration.as_deref().map(|value| value.as_ref()), Some("01:30:30")); - assert_eq!(details.runtime.as_deref().map(|value| value.as_ref()), Some("5430")); + assert_eq!(details.video.as_deref(), Some("{\"codec_name\":\"hevc\"}")); + assert_eq!(details.audio.as_deref(), Some("{\"codec_name\":\"aac\"}")); + assert_eq!(details.duration_secs.as_deref(), Some("5430")); + assert_eq!(details.duration.as_deref(), Some("01:30:30")); + assert_eq!(details.runtime.as_deref(), Some("5430")); assert_eq!(details.bitrate, 8_200_000); } @@ -651,8 +651,8 @@ mod tests { let episodes = series.details.as_ref().and_then(|details| details.episodes.as_ref()).expect("episodes missing"); let episode = episodes.first().expect("episode missing"); - assert_eq!(episode.video.as_deref().map(|value| value.as_ref()), Some("{\"codec_name\":\"h264\"}")); - assert_eq!(episode.audio.as_deref().map(|value| value.as_ref()), Some("{\"codec_name\":\"ac3\"}")); + assert_eq!(episode.video.as_deref(), Some("{\"codec_name\":\"h264\"}")); + assert_eq!(episode.audio.as_deref(), Some("{\"codec_name\":\"ac3\"}")); assert_eq!(episode.duration_secs, 1_541); assert_eq!(episode.duration.as_ref(), "00:25:41"); assert_eq!(episode.bitrate, 3_100_000); diff --git a/backend/src/processing/processor/playlist.rs b/backend/src/processing/processor/playlist.rs index 27b5d75be..f3fe1a0ff 100644 --- a/backend/src/processing/processor/playlist.rs +++ b/backend/src/processing/processor/playlist.rs @@ -1,3 +1,4 @@ +use super::filtered_playlist_source::FilteredPlaylistSource; use crate::{ api::{ model::{ @@ -29,7 +30,6 @@ use crate::{ debug_if_enabled, epg, log_memory_snapshot, m3u, trace_if_enabled, xtream, StepMeasure, StepMeasureCallback, }, }; -use super::filtered_playlist_source::FilteredPlaylistSource; use futures::{FutureExt, StreamExt}; use indexmap::IndexMap; use log::{debug, error, info, log_enabled, warn, Level}; @@ -231,10 +231,10 @@ fn map_playlist(source: &mut dyn PlaylistSource, target: &ConfigTarget) -> Optio let mapping_binding = target.mapping.load(); let mappings = mapping_binding.as_ref()?; let valid_mappings = mappings.iter().filter(|m| m.mapper.as_ref().is_some_and(|v| !v.is_empty())); - let iter: Box> = Box::new(source.into_items()); + let iter: Box> = Box::new(source.into_items()); let mapped_iter = valid_mappings.fold(iter, |iter, mapping| { Box::new(iter.flat_map(move |chan| map_channel_and_flatten(chan, mapping))) - as Box> + as Box> }); let mut next_groups: IndexMap = IndexMap::new(); let mut grp_id: u32 = 0; @@ -464,7 +464,7 @@ async fn playlist_download_from_input( input, Some(xtream_clusters_to_download.as_slice()), ) - .await; + .await; xtream_error_count = xtream_errors.len(); playlist.extend(xtream_playlist); all_errors.extend(xtream_errors); @@ -490,7 +490,7 @@ async fn playlist_download_from_input( input, Some(xtream_clusters_to_download.as_slice()), ) - .await; + .await; let xtream_error_count = e.len(); (p, e, persisted, 0, xtream_error_count) } @@ -658,7 +658,7 @@ async fn process_source( &mut errors, consume_input_source, ) - .await + .await { Ok(()) => { target_stats.push(TargetStats::success(&target.name)); @@ -1133,7 +1133,7 @@ async fn process_playlist_for_target( target, ctx.playlist_state.as_ref(), ) - .await; + .await; step.stop("Persisting playlists"); log_memory_snapshot(format!("target '{}' after_persist", target.name).as_str()); result @@ -1256,17 +1256,42 @@ async fn playlist_probe(ctx: &PlaylistProcessingContext, target: &ConfigTarget, let mut queued_live_count = 0usize; let mut queued_stream_count = 0usize; + let probe_filter = fpl.input.options.as_ref().and_then(|o| o.probe_filter.as_ref()); + for item in fpl.items() { + if !is_probe_supported_item_type(item.header.item_type) { continue; } - match item.header.item_type { PlaylistItemType::Live => { if !probe_live_enabled { continue; } + } + PlaylistItemType::Video | PlaylistItemType::LocalVideo => { + if !probe_vod_enabled { + continue; + } + } + PlaylistItemType::Series | PlaylistItemType::LocalSeries => { + if !probe_series_enabled { + continue; + } + } + _ => continue, + } + // If input has a probe filter and this item doesn't match, skip probing + if let Some(p_filter) = probe_filter { + let provider = ValueProvider { pli: &item, match_as_ascii: false }; + if !p_filter.filter(&provider) { + continue; + } + } + + match item.header.item_type { + PlaylistItemType::Live => { if let Some((probe_delay, interval_secs, cutoff_ts)) = live_probe_settings { if needs_live_probe(&item, cutoff_ts) { if let Some(provider_id) = provider_id_from_item(&item) { @@ -1301,18 +1326,12 @@ async fn playlist_probe(ctx: &PlaylistProcessingContext, target: &ConfigTarget, // generic probe path to keep behaviour consistent with non-xtream outputs. } PlaylistItemType::Video | PlaylistItemType::LocalVideo => { - if !probe_vod_enabled { - continue; - } // Xtream outputs handle VOD probe as part of the resolve pipeline (after resolve). if xtream_probe_handled { continue; } } PlaylistItemType::Series | PlaylistItemType::LocalSeries => { - if !probe_series_enabled { - continue; - } // Xtream outputs handle Series probe as part of the resolve pipeline (after resolve). if xtream_probe_handled { continue; @@ -1436,8 +1455,8 @@ async fn process_watch( .filter(|pl| watches.iter().any(|r| r.is_match(&pl.title))) .map(|pl| process_group_watch(app_config, client, &target.name, pl)), ) - .for_each_concurrent(16, |f| f) - .await; + .for_each_concurrent(16, |f| f) + .await; true } else { @@ -1522,7 +1541,7 @@ pub async fn exec_processing( max_update_duration, std::panic::AssertUnwindSafe(process_sources(&ctx)).catch_unwind(), ) - .await; + .await; let (stats, errors) = match process_result { Ok(Ok((stats, errors))) => (stats, errors), Ok(Err(_)) => { @@ -1593,6 +1612,8 @@ pub async fn exec_processing( mod tests { use super::*; use crate::model::StagedInput; + use shared::foundation::{get_filter, ValueProvider}; + use shared::model::{PlaylistItem, PlaylistItemHeader, PlaylistItemType}; use shared::utils::Internable; fn item_with_props(props: StreamProperties) -> PlaylistItem { @@ -1770,4 +1791,43 @@ mod tests { assert_eq!(filtered_groups.len(), 1); assert_eq!(filtered_groups[0].xtream_cluster, XtreamCluster::Video); } + + + fn make_test_item(name: &str, item_type: PlaylistItemType) -> PlaylistItem { + let header = PlaylistItemHeader { + name: name.into(), + group: "Test Group".intern(), + item_type, + ..Default::default() + }; + PlaylistItem { header } + } + + #[test] + fn test_filter_evalutes_correctly() { + let filter = get_filter(r#"name ~ "Allowed""#, None).unwrap(); + + let allowed_item = make_test_item("Allowed Channel", PlaylistItemType::Live); + let denied_item = make_test_item("Denied Channel", PlaylistItemType::Live); + + let allowed_provider = ValueProvider { pli: &allowed_item, match_as_ascii: false }; + let denied_provider = ValueProvider { pli: &denied_item, match_as_ascii: false }; + + assert!(filter.filter(&allowed_provider)); + assert!(!filter.filter(&denied_provider)); + } + + #[test] + fn test_filter_with_type_comparison() { + let filter = get_filter("type = vod", None).unwrap(); + + let vod_item = make_test_item("Test Movie", PlaylistItemType::Video); + let live_item = make_test_item("Test Channel", PlaylistItemType::Live); + + let vod_provider = ValueProvider { pli: &vod_item, match_as_ascii: false }; + let live_provider = ValueProvider { pli: &live_item, match_as_ascii: false }; + + assert!(filter.filter(&vod_provider)); + assert!(!filter.filter(&live_provider)); + } } diff --git a/backend/src/processing/processor/sort.rs b/backend/src/processing/processor/sort.rs index 240af30e3..70f70506e 100644 --- a/backend/src/processing/processor/sort.rs +++ b/backend/src/processing/processor/sort.rs @@ -795,14 +795,14 @@ mod tests { header: PlaylistItemHeader { title: Arc::from("HD"), source_ordinal: 2, ..Default::default() }, }], ); - let group_uhd = make_group( + let group_ultra_hd = make_group( 2, "uhd", vec![PlaylistItem { header: PlaylistItemHeader { title: Arc::from("UHD"), source_ordinal: 3, ..Default::default() }, }], ); - let group_fhd = make_group( + let group_full_hd = make_group( 3, "fhd", vec![PlaylistItem { @@ -822,7 +822,7 @@ mod tests { filter: Filter::default(), }; - let mut groups = vec![group_hd, group_uhd, group_fhd]; + let mut groups = vec![group_hd, group_ultra_hd, group_full_hd]; sort_groups(&mut groups, &[group_sort], false); let sorted = groups.iter().map(|group| group.channels[0].header.title.clone()).collect::>(); diff --git a/backend/src/processing/processor/stream_probe.rs b/backend/src/processing/processor/stream_probe.rs index 3bed794c2..b00bfc93f 100644 --- a/backend/src/processing/processor/stream_probe.rs +++ b/backend/src/processing/processor/stream_probe.rs @@ -403,7 +403,7 @@ mod tests { panic!("expected video properties"); }; let details = video.details.as_ref().unwrap_or_else(|| unreachable!()); - assert_eq!(details.duration_secs.as_deref().map(|value| value.as_ref()), Some("1541")); + assert_eq!(details.duration_secs.as_deref().map(std::convert::AsRef::as_ref), Some("1541")); assert_eq!(details.bitrate, 3_100_000); } } diff --git a/backend/src/processing/processor/xtream_series.rs b/backend/src/processing/processor/xtream_series.rs index c8eef0817..c4e3c2645 100644 --- a/backend/src/processing/processor/xtream_series.rs +++ b/backend/src/processing/processor/xtream_series.rs @@ -34,6 +34,7 @@ use std::collections::{HashMap, HashSet}; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::time::{Duration, Instant}; +use shared::foundation::ValueProvider; use shared::utils::default_probe_user_priority; create_resolve_options_function_for_xtream_target!(series); @@ -193,12 +194,22 @@ fn queue_background_series_info( let input = fpl.input; let input_name_arc = input.name.clone(); + // Extract filter before iterating to avoid borrow conflict + let resolve_filter = fpl.input.options.as_ref().and_then(|o| o.resolve_filter.as_ref()); for pli in fpl.items_mut() { if !filter(pli) { continue; } + // If input has a resolve filter and this item doesn't match, skip processing + if let Some(r_filter) = resolve_filter { + let provider = ValueProvider { pli, match_as_ascii: false }; + if !r_filter.filter(&provider) { + continue; + } + } + let provider_id = if let Ok(uid) = pli.header.id.parse::() { ProviderIdType::Id(uid) } else { @@ -277,11 +288,22 @@ async fn process_immediate_series_info( SeriesProbeSettings::from_metadata_update(config.metadata_update.as_ref()) }; + // Extract filter before iterating to avoid borrow conflict + let resolve_filter = fpl.input.options.as_ref().and_then(|o| o.resolve_filter.as_ref()); + for pli in fpl.items_mut() { if !filter(pli) { continue; } + // If input has a resolve filter and this item doesn't match, skip processing + if let Some(r_filter) = resolve_filter { + let provider = ValueProvider { pli, match_as_ascii: false }; + if !r_filter.filter(&provider) { + continue; + } + } + let provider_id = if let Ok(uid) = pli.header.id.parse::() { ProviderIdType::Id(uid) } else { diff --git a/backend/src/processing/processor/xtream_vod.rs b/backend/src/processing/processor/xtream_vod.rs index 4f67c655a..f2e1dd51c 100644 --- a/backend/src/processing/processor/xtream_vod.rs +++ b/backend/src/processing/processor/xtream_vod.rs @@ -31,6 +31,7 @@ use std::collections::{HashMap, HashSet}; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::time::{Duration, Instant}; +use shared::foundation::ValueProvider; use shared::utils::default_probe_user_priority; create_resolve_options_function_for_xtream_target!(vod); @@ -164,11 +165,22 @@ async fn process_immediate_vod_info( let mut processed_count = 0; let mut last_log_time = Instant::now(); + // Extract filter before iterating to avoid borrow conflict + let resolve_filter = fpl.input.options.as_ref().and_then(|o| o.resolve_filter.as_ref()); + for pli in fpl.items_mut() { if !filter(pli) { continue; } + // If input has a filter and this item doesn't match, skip processing + if let Some(r_filter) = resolve_filter { + let provider = ValueProvider { pli, match_as_ascii: false }; + if !r_filter.filter(&provider) { + continue; + } + } + let provider_id = if let Ok(uid) = pli.header.id.parse::() { ProviderIdType::Id(uid) } else { @@ -429,11 +441,21 @@ 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()); for pli in fpl.items_mut() { if !filter(pli) { continue; } + // If input has a resolve filter and this item doesn't match, skip processing + if let Some(r_filter) = resolve_filter { + let provider = ValueProvider { pli, match_as_ascii: false }; + if !r_filter.filter(&provider) { + continue; + } + } + let provider_id = if let Ok(uid) = pli.header.id.parse::() { ProviderIdType::Id(uid) } else { diff --git a/backend/src/repository/playlist_repository.rs b/backend/src/repository/playlist_repository.rs index c6f24cc91..3b6448902 100644 --- a/backend/src/repository/playlist_repository.rs +++ b/backend/src/repository/playlist_repository.rs @@ -630,7 +630,7 @@ mod tests { }]; for (idx, channel) in playlist[0].channels.iter_mut().enumerate() { - let uuid = channel.header.uuid.clone(); + let uuid = channel.header.uuid; let provider_id = channel.header.get_provider_id().unwrap_or_default(); let item_type = channel.header.item_type; channel.header.virtual_id = target_id_mapping.get_and_update_virtual_id(&uuid, provider_id, item_type, 0); diff --git a/backend/src/repository/qos_snapshot_repository.rs b/backend/src/repository/qos_snapshot_repository.rs index e0cf09a76..78a5be070 100644 --- a/backend/src/repository/qos_snapshot_repository.rs +++ b/backend/src/repository/qos_snapshot_repository.rs @@ -208,6 +208,7 @@ fn load_snapshot_tree(path: &Path) -> io::Result PathBuf { use crate::repository::stream_history::{ BlockHeaderBody, CONTAINER_FORMAT_VERSION, CompressionKind, diff --git a/backend/src/repository/stream_history/reader.rs b/backend/src/repository/stream_history/reader.rs index 2715220d2..5625fdbea 100644 --- a/backend/src/repository/stream_history/reader.rs +++ b/backend/src/repository/stream_history/reader.rs @@ -492,6 +492,7 @@ mod tests { } /// Helper: write a valid pending file with given records into a temp file + #[allow(clippy::cast_possible_truncation)] fn write_test_pending_file(records: &[StreamHistoryRecord]) -> NamedTempFile { let tmp = NamedTempFile::new().unwrap(); let mut f = tmp.as_file().try_clone().unwrap(); @@ -625,10 +626,11 @@ mod tests { // Should not panic, just yield no records or an error let results: Vec<_> = reader.collect(); // Either empty or contains an error — but no panic - assert!(results.is_empty() || results.iter().any(|r| r.is_err())); + assert!(results.is_empty() || results.iter().any(std::result::Result::is_err)); } #[test] + #[allow(clippy::cast_possible_truncation)] fn test_magic_recovery_oom_protection() { // Test that we don't OOM when block header has oversized payload_len // and that we can continue reading after skipping it @@ -683,7 +685,7 @@ mod tests { // The iterator may be empty after graceful recovery, but it must never yield // a successfully decoded record from this corrupt truncated payload. - assert!(results.iter().all(|r| r.is_err())); + assert!(results.iter().all(std::result::Result::is_err)); } // Additional test for lz4 archive round-trip: diff --git a/backend/src/repository/stream_history/storage.rs b/backend/src/repository/stream_history/storage.rs index 256bbc9ab..2025e0723 100644 --- a/backend/src/repository/stream_history/storage.rs +++ b/backend/src/repository/stream_history/storage.rs @@ -417,6 +417,7 @@ mod tests { } #[test] + #[allow(clippy::cast_possible_truncation)] fn stream_history_block_header_round_trip() { let payload = b"test_payload"; let payload_crc = crc32fast::hash(payload); diff --git a/backend/src/repository/user_repository.rs b/backend/src/repository/user_repository.rs index a8ea21ea5..19f2cf5d8 100644 --- a/backend/src/repository/user_repository.rs +++ b/backend/src/repository/user_repository.rs @@ -393,6 +393,7 @@ mod tests { use std::sync::Arc; #[tokio::test] + #[allow(clippy::too_many_lines)] pub async fn save_target_user() { let user = TargetUser { diff --git a/backend/src/repository/xtream_repository.rs b/backend/src/repository/xtream_repository.rs index 3c95b08a7..8cdfb1828 100644 --- a/backend/src/repository/xtream_repository.rs +++ b/backend/src/repository/xtream_repository.rs @@ -1618,14 +1618,14 @@ mod tests { ..VideoStreamProperties::default() })); let old_props = StreamProperties::Video(Box::new(VideoStreamProperties { - tmdb: Some(317981), + tmdb: Some(317_981), ..VideoStreamProperties::default() })); let changed = merge_preserved_stream_properties(&mut new_props, &old_props); assert!(changed); match new_props { - StreamProperties::Video(video) => assert_eq!(video.tmdb, Some(317981)), + StreamProperties::Video(video) => assert_eq!(video.tmdb, Some(317_981)), _ => panic!("expected video properties"), } } diff --git a/backend/src/runtime_config_report.rs b/backend/src/runtime_config_report.rs index 8dc40379d..df4a9993e 100644 --- a/backend/src/runtime_config_report.rs +++ b/backend/src/runtime_config_report.rs @@ -296,7 +296,7 @@ mod tests { let home_dir = config_dir.join("home"); fs::create_dir_all(&home_dir).await.expect("home dir"); - let config_yml = r#" + let config_yml = r" storage_dir: data api: host: 0.0.0.0 @@ -309,7 +309,7 @@ proxy: url: http://proxy.example username: proxy-user password: proxy-pass -"#; +"; let source_yml = r#" inputs: - name: demo diff --git a/backend/src/utils/stream_history_viewer.rs b/backend/src/utils/stream_history_viewer.rs index 2c81aeb02..eb7eec7f3 100644 --- a/backend/src/utils/stream_history_viewer.rs +++ b/backend/src/utils/stream_history_viewer.rs @@ -483,14 +483,14 @@ mod tests { let ts = parse_date_or_datetime("2026-03-22").unwrap(); // 2026-03-22 00:00:00 UTC assert_eq!(ts, chrono::NaiveDate::from_ymd_opt(2026, 3, 22).unwrap() - .and_hms_opt(0, 0, 0).unwrap().and_utc().timestamp() as u64); + .and_hms_opt(0, 0, 0).unwrap().and_utc().timestamp().cast_unsigned()); } #[test] fn test_parse_datetime_no_seconds() { let ts = parse_date_or_datetime("2026-03-22 14:30").unwrap(); let expected = chrono::NaiveDate::from_ymd_opt(2026, 3, 22).unwrap() - .and_hms_opt(14, 30, 0).unwrap().and_utc().timestamp() as u64; + .and_hms_opt(14, 30, 0).unwrap().and_utc().timestamp().cast_unsigned(); assert_eq!(ts, expected); } @@ -498,7 +498,7 @@ mod tests { fn test_parse_datetime_with_seconds() { let ts = parse_date_or_datetime("2026-03-22 14:30:45").unwrap(); let expected = chrono::NaiveDate::from_ymd_opt(2026, 3, 22).unwrap() - .and_hms_opt(14, 30, 45).unwrap().and_utc().timestamp() as u64; + .and_hms_opt(14, 30, 45).unwrap().and_utc().timestamp().cast_unsigned(); assert_eq!(ts, expected); } diff --git a/docs/src/configuration/source.md b/docs/src/configuration/source.md index 9228ae342..51b5f693b 100644 --- a/docs/src/configuration/source.md +++ b/docs/src/configuration/source.md @@ -204,20 +204,22 @@ headers: Controls the behavior during download and asynchronous metadata resolution (see the *Metadata Update* chapter) for this specific provider. -| Parameter | Type | Default | Technical Impact & Background | -|:---------------------------------------|:-----|:--------|:--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| -| `xtream_skip_live` / `vod` / `series` | Bool | `false` | Immediately ignores entire categories during the Xtream API download. Saves massive amounts of RAM and runtime if you only want Live-TV from a specific provider, for instance. | -| `xtream_live_stream_without_extension` | Bool | `false` | Strips `.ts` from generated stream URLs. | -| `xtream_live_stream_use_prefix` | Bool | `true` | Injects the `/live/` prefix into URLs. | -| `disable_hls_streaming` | Bool | `false` | Forces Tuliprox to play Live-TV as a raw MPEG-TS (`.ts`) stream, skipping HLS (`.m3u8`) reverse-proxy handling, and forcing direct TS endpoints. | -| `resolve_tmdb` | Bool | `false` | Enables TMDB queries for this specific input based on parsed titles to fill missing posters and release years. | -| `probe_stream` | Bool | `false` | Uses FFprobe to read A/V details (HDR, 4K). Respects `max_connections`. | -| `resolve_background` | Bool | `true` | Metadata scans run asynchronously in the background so the general playlist update (which blocks clients) finishes instantly. | -| `resolve_series` / `resolve_vod` | Bool | `false` | Fetches missing details like Plot or Cast via the Provider's API (`get_vod_info` / `get_series_info`). | -| `probe_series` / `probe_vod` | Bool | `false` | Allows explicit FFprobe analysis of movies or entire TV show seasons. | -| `probe_live` | Bool | `false` | Allows FFprobe to periodically tap into Live-TV streams in the background. | -| `probe_live_interval_hours` | Int | `120` | Interval after which a Live stream is re-analyzed (Important as backup streams often change resolutions). | -| `resolve_delay` / `probe_delay` | Int | `2` | **Ban Protection:** Hard wait time (in seconds) between API or Probe requests to the *same* provider! Prevents API spamming. | +| Parameter | Type | Default | Technical Impact & Background | +|:---------------------------------------|:---------|:--------|:--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| `xtream_skip_live` / `vod` / `series` | Bool | `false` | Immediately ignores entire categories during the Xtream API download. Saves massive amounts of RAM and runtime if you only want Live-TV from a specific provider, for instance. | +| `xtream_live_stream_without_extension` | Bool | `false` | Strips `.ts` from generated stream URLs. | +| `xtream_live_stream_use_prefix` | Bool | `true` | Injects the `/live/` prefix into URLs. | +| `disable_hls_streaming` | Bool | `false` | Forces Tuliprox to play Live-TV as a raw MPEG-TS (`.ts`) stream, skipping HLS (`.m3u8`) reverse-proxy handling, and forcing direct TS endpoints. | +| `resolve_tmdb` | Bool | `false` | Enables TMDB queries for this specific input based on parsed titles to fill missing posters and release years. | +| `probe_stream` | Bool | `false` | Uses FFprobe to read A/V details (HDR, 4K). Respects `max_connections`. | +| `resolve_background` | Bool | `true` | Metadata scans run asynchronously in the background so the general playlist update (which blocks clients) finishes instantly. | +| `resolve_series` / `resolve_vod` | Bool | `false` | Fetches missing details like Plot or Cast via the Provider's API (`get_vod_info` / `get_series_info`). | +| `probe_series` / `probe_vod` | Bool | `false` | Allows explicit FFprobe analysis of movies or entire TV show seasons. | +| `probe_live` | Bool | `false` | Allows FFprobe to periodically tap into Live-TV streams in the background. | +| `probe_live_interval_hours` | Int | `120` | Interval after which a Live stream is re-analyzed (Important as backup streams often change resolutions). | +| `resolve_delay` / `probe_delay` | Int | `2` | **Ban Protection:** Hard wait time (in seconds) between API or Probe requests to the *same* provider! Prevents API spamming. | +| `resolve_filter` | String | - | Filter expression to selectively resolve only entries matching the condition. Uses the same Filter syntax. | +| `probe_filter` | String | - | Filter expression to selectively probe only entries matching the condition. Uses the same Filter syntax. | > **Note:** For `resolve_vod` and `resolve_series`, data is cached per input and only new or changed entries are > updated. diff --git a/frontend/public/assets/i18n/en.json b/frontend/public/assets/i18n/en.json index cbdfd65ba..c7b622c82 100644 --- a/frontend/public/assets/i18n/en.json +++ b/frontend/public/assets/i18n/en.json @@ -311,8 +311,10 @@ "PROBE_LIVE_INTERVAL_HOURS": "Interval in hours between re-probing live channels (default: 120).", "PROBE_SERIES": "Enables technical analysis for series entries (requires ffprobe and probe_stream).", "PROBE_VOD": "Enables technical analysis for VOD entries (requires ffprobe and probe_stream).", + "PROBE_FILTER": "Filter expression to selectively probe only matching entries (e.g. 'name ~ \"S01\"').", "RESOLVE_BACKGROUND": "If true, metadata resolve tasks are processed in the background (default true).", "RESOLVE_DELAY": "Delay in seconds between metadata resolve requests (default 2).", + "RESOLVE_FILTER": "Filter expression to selectively resolve only matching entries.", "RESOLVE_SERIES": "Enables metadata resolution (TMDB/NFO) for series entries.", "RESOLVE_TMDB": "Attempts to resolve missing TMDB IDs based on title if missing from provider.", "RESOLVE_VOD": "Enables metadata resolution (TMDB/NFO) for VOD entries.", @@ -1141,6 +1143,7 @@ "PROBE": "Probe", "PROBE_DELAY_SEC": "Probe Delay (sec)", "PROBE_LIVE_INTERVAL_HOURS": "Live Interval (hour)", + "PROBE_FILTER": "Probe Filter", "PROBE_STREAM": "Probe Stream", "PROCESSING_ORDER": "Processing order", "PROCESS_PARALLEL": "Process Parallel", @@ -1193,6 +1196,7 @@ "RESOLVE": "Resolve", "RESOLVE_BACKGROUND": "Resolve in Background", "RESOLVE_DELAY_SEC": "Resolve Delay (sec)", + "RESOLVE_FILTER": "Resolve Filter", "RESOLVE_TMDB": "Resolve TMDB", "RESOURCE_RETRY": "Resource Retry", "RESOURCE_REWRITE_DISABLE": "Resource Rewrite disable", diff --git a/frontend/src/app/components/playlist/input_table.rs b/frontend/src/app/components/playlist/input_table.rs index d5da44620..6ed78c8e9 100644 --- a/frontend/src/app/components/playlist/input_table.rs +++ b/frontend/src/app/components/playlist/input_table.rs @@ -1,24 +1,19 @@ use crate::{ app::components::{ - convert_bool_to_chip_style, menu_item::MenuItem, popup_menu::PopupMenu, AppIcon, BatchInputContentView, Chip, - EpgConfigView, HideContent, InputHeaders, InputOptions, InputTypeView, RevealContent, StagedInputView, Table, - TableDefinition, + convert_bool_to_chip_style, AppIcon, BatchInputContentView, Chip, EpgConfigView, HideContent, InputHeaders, + InputOptions, InputTypeView, RevealContent, StagedInputView, Table, TableDefinition, }, html_if, i18n::use_translation, - model::DialogResult, - services::DialogService, }; use shared::{ - error::TuliproxError, model::{ConfigInputAliasDto, ConfigInputDto, InputType, SortOrder}, utils::unix_ts_to_str, }; -use std::{fmt::Display, rc::Rc, str::FromStr}; -use yew::{platform::spawn_local, prelude::*}; +use std::rc::Rc; +use yew::prelude::*; -const HEADERS: [&str; 16] = [ - "LABEL.EMPTY", +const HEADERS: [&str; 15] = [ "LABEL.ENABLED", "LABEL.NAME", "LABEL.INPUT_TYPE", @@ -45,39 +40,11 @@ pub enum InputRow { #[derive(Properties, PartialEq, Clone)] pub struct InputTableProps { pub inputs: Option>>, - #[prop_or_default] - pub on_edit: Option>>, - #[prop_or_default] - pub on_delete: Option>, } #[component] pub fn InputTable(props: &InputTableProps) -> Html { let translate = use_translation(); - let dialog = use_context::().expect("Dialog service not found"); - let popup_anchor_ref = use_state(|| None::); - let popup_is_open = use_state(|| false); - let selected_dto = use_state(|| None::>); - - let handle_popup_close = { - let set_is_open = popup_is_open.clone(); - Callback::from(move |()| { - set_is_open.set(false); - }) - }; - - let handle_popup_onclick = { - let set_selected_dto = selected_dto.clone(); - let set_anchor_ref = popup_anchor_ref.clone(); - let set_is_open = popup_is_open.clone(); - Callback::from(move |(dto, event): (Rc, MouseEvent)| { - if let Some(target) = event.target_dyn_into::() { - set_selected_dto.set(Some(dto.clone())); - set_anchor_ref.set(Some(target)); - set_is_open.set(true); - } - }) - }; let render_header_cell = { let translator = translate.clone(); @@ -96,44 +63,33 @@ pub fn InputTable(props: &InputTableProps) -> Html { let render_data_cell = { let translator = translate.clone(); - let popup_onclick = handle_popup_onclick.clone(); - Callback::<(usize, usize, Rc), Html>::from(move |(row, col, input): (usize, usize, Rc)| { + Callback::<(usize, usize, Rc), Html>::from(move |(_row, col, input): (usize, usize, Rc)| { match &*input { InputRow::Input(dto) => match col { - 0 => { - let popup_onclick = popup_onclick.clone(); - html! { - - } - } - 1 => html! { html! { }, - 2 => html! { dto.name.as_ref() }, - 3 => html! { }, - 4 => html! { if matches!(dto.input_type, InputType::XtreamBatch | InputType::M3uBatch) { + 1 => html! { dto.name.as_ref() }, + 2 => html! { }, + 3 => html! { if matches!(dto.input_type, InputType::XtreamBatch | InputType::M3uBatch) { } else { {dto.url.as_str()} } }, - 5 => dto.username.as_ref().map_or_else(|| html! {}, |u| html! {u}), - 6 => dto + 4 => dto.username.as_ref().map_or_else(|| html! {}, |u| html! {u}), + 5 => dto .password .as_ref() .map_or_else(|| html! {}, |pwd| html! { }), - 7 => dto.persist.as_ref().map_or_else(|| html! {}, |p| html! {p}), - 8 => { + 6 => dto.persist.as_ref().map_or_else(|| html! {}, |p| html! {p}), + 7 => { html! { } } - 9 => html! { dto.priority.to_string() }, - 10 => html! { dto.max_connections.to_string() }, - 11 => html! { dto.method.to_string() }, - 12 => html_if!(dto.epg.is_some(), + 8 => html! { dto.priority.to_string() }, + 9 => html! { dto.max_connections.to_string() }, + 10 => html! { dto.method.to_string() }, + 11 => html_if!(dto.epg.is_some(), { Html { })}}> }), - 13 => { + 12 => { html! { } } - 14 => html_if!(dto.staged.is_some(), + 13 => html_if!(dto.staged.is_some(), { }), - 15 => dto + 14 => dto .exp_date .as_ref() .and_then(|ts| unix_ts_to_str(*ts)) @@ -159,31 +115,21 @@ pub fn InputTable(props: &InputTableProps) -> Html { _ => html! {""}, }, InputRow::Alias(alias, _dto) => match col { - 0 => { - let popup_onclick = popup_onclick.clone(); - html! { - - } - } - 1 => html! { + 0 => html! { }, - 2 => html! { alias.name.as_ref() }, - 4 => html! { alias.url.as_str() }, - 5 => alias.username.as_ref().map_or_else(|| html! {}, |u| html! {u}), - 6 => alias + 1 => html! { alias.name.as_ref() }, + 3 => html! { alias.url.as_str() }, + 4 => alias.username.as_ref().map_or_else(|| html! {}, |u| html! {u}), + 5 => alias .password .as_ref() .map_or_else(|| html! {}, |pwd| html! { }), - 9 => html! { alias.priority.to_string() }, - 10 => html! { alias.max_connections.to_string() }, - 15 => alias + 8 => html! { alias.priority.to_string() }, + 9 => html! { alias.max_connections.to_string() }, + 14 => alias .exp_date .as_ref() .and_then(|ts| unix_ts_to_str(*ts)) @@ -195,47 +141,6 @@ pub fn InputTable(props: &InputTableProps) -> Html { }) }; - let handle_menu_click = { - let popup_is_open_state = popup_is_open.clone(); - let confirm = dialog.clone(); - let translate = translate.clone(); - let on_edit = props.on_edit.clone(); - let on_delete = props.on_delete.clone(); - let selected_dto = selected_dto.clone(); - - Callback::from(move |(name, _): (String, _)| { - if let Ok(action) = TableAction::from_str(&name) { - match action { - TableAction::Edit => { - if let (Some(on_edit), Some(selected)) = (&on_edit, &*selected_dto) { - if let InputRow::Input(dto) = &**selected { - on_edit.emit(dto.clone()); - } - } - } - TableAction::Refresh => {} - TableAction::Delete => { - let confirm = confirm.clone(); - let translator = translate.clone(); - let on_delete = on_delete.clone(); - let selected_dto = selected_dto.clone(); - spawn_local(async move { - let result = confirm.confirm(&translator.t("MESSAGES.CONFIRM_DELETE")).await; - if result == DialogResult::Ok { - if let (Some(on_delete), Some(selected)) = (on_delete, &*selected_dto) { - if let InputRow::Input(dto) = &**selected { - on_delete.emit(dto.name.to_string()); - } - } - } - }); - } - } - } - popup_is_open_state.set(false); - }) - }; - let is_sortable = Callback::::from(move |_col| false); let on_sort = Callback::, ()>::from(move |_args| {}); @@ -265,13 +170,7 @@ pub fn InputTable(props: &InputTableProps) -> Html { { if let Some(definition) = table_definition.as_ref() { html! { - <> - definition={definition.clone()} /> - - - - - + definition={definition.clone()} /> } } else { html! {} @@ -280,40 +179,3 @@ pub fn InputTable(props: &InputTableProps) -> Html { } } - -#[derive(Debug, Clone, Eq, PartialEq)] -enum TableAction { - Edit, - Refresh, - Delete, -} - -impl Display for TableAction { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!( - f, - "{}", - match self { - Self::Edit => "edit", - Self::Refresh => "refresh", - Self::Delete => "delete", - } - ) - } -} - -impl FromStr for TableAction { - type Err = TuliproxError; - - fn from_str(s: &str) -> Result { - if s.eq("edit") { - Ok(Self::Edit) - } else if s.eq("refresh") { - Ok(Self::Refresh) - } else if s.eq("delete") { - Ok(Self::Delete) - } else { - Err(TuliproxError::Config(format!("Unknown TableAction: {}", s))) - } - } -} diff --git a/frontend/src/app/components/playlist/playlist_settings_view.rs b/frontend/src/app/components/playlist/playlist_settings_view.rs index 3de07a57d..0be64168e 100644 --- a/frontend/src/app/components/playlist/playlist_settings_view.rs +++ b/frontend/src/app/components/playlist/playlist_settings_view.rs @@ -1,18 +1,10 @@ -use crate::{ - app::components::{Breadcrumbs, PlaylistList}, - i18n::use_translation, -}; -use std::rc::Rc; +use crate::app::components::PlaylistList; use yew::prelude::*; #[component] pub fn PlaylistSettingsView() -> Html { - let translate = use_translation(); - let breadcrumbs = use_state(|| Rc::new(vec![translate.t("LABEL.PLAYLISTS"), translate.t("LABEL.LIST")])); - html! {

-
diff --git a/frontend/src/app/components/sidebar.rs b/frontend/src/app/components/sidebar.rs index 18110c04f..d22a943f5 100644 --- a/frontend/src/app/components/sidebar.rs +++ b/frontend/src/app/components/sidebar.rs @@ -164,7 +164,10 @@ pub fn Sidebar(props: &SidebarProps) -> Html { })} {html_if!(auth.has_permission(Permission::SourceRead), { + <> + + })} } @@ -177,10 +180,7 @@ pub fn Sidebar(props: &SidebarProps) -> Html { })} {html_if!(auth.has_permission(Permission::PlaylistRead), { - <> - - - + })} {html_if!(auth.has_permission(Permission::EpgRead), { @@ -225,7 +225,10 @@ pub fn Sidebar(props: &SidebarProps) -> Html { })} {html_if!(auth.has_permission(Permission::SourceRead), { + <> + + })} {html_if!( auth.has_any_permissions(Permission::PlaylistRead | Permission::PlaylistWrite | Permission::EpgRead), @@ -237,10 +240,7 @@ pub fn Sidebar(props: &SidebarProps) -> Html { })} {html_if!(auth.has_permission(Permission::PlaylistRead), { - <> - - - + })} {html_if!(auth.has_permission(Permission::EpgRead), { diff --git a/frontend/src/app/components/source_editor/input_form.rs b/frontend/src/app/components/source_editor/input_form.rs index 60a56b1ea..b4191bb18 100644 --- a/frontend/src/app/components/source_editor/input_form.rs +++ b/frontend/src/app/components/source_editor/input_form.rs @@ -3,8 +3,8 @@ use crate::{ components::{ config::HasFormData, key_value_editor::KeyValueEditor, select::Select, AliasItemForm, BlockId, BlockInstance, Card, DropDownOption, DropDownSelection, EditMode, EpgSmartMatchForm, EpgSourceItemForm, - IconButton, Panel, ProviderItemForm, RadioButtonGroup, SourceEditorContext, TextButton, TitledCard, - ToolAction, + FilterInput, IconButton, Panel, ProviderItemForm, RadioButtonGroup, SourceEditorContext, TextButton, + TitledCard, ToolAction, }, ConfigContext, }, @@ -77,6 +77,8 @@ const LABEL_PROBE: &str = "LABEL.PROBE"; const LABEL_RESOLVE_DELAY_SEC: &str = "LABEL.RESOLVE_DELAY_SEC"; const LABEL_PROBE_DELAY_SEC: &str = "LABEL.PROBE_DELAY_SEC"; const LABEL_RESOLVE_BACKGROUND: &str = "LABEL.RESOLVE_BACKGROUND"; +const LABEL_RESOLVE_FILTER: &str = "LABEL.RESOLVE_FILTER"; +const LABEL_PROBE_FILTER: &str = "LABEL.PROBE_FILTER"; const LABEL_PROBE_LIVE_INTERVAL_HOURS: &str = "LABEL.PROBE_LIVE_INTERVAL_HOURS"; const LABEL_METADATA: &str = "LABEL.METADATA"; const LABEL_CACHE_DURATION: &str = "LABEL.CACHE_DURATION"; @@ -204,6 +206,8 @@ generate_form_reducer!( ProbeVod => probe_vod: bool, ProbeSeries => probe_series: bool, ProbeLiveIntervalHours => probe_live_interval_hours: u32, + ResolveFilter => resolve_filter: Option, + ProbeFilter => probe_filter: Option, } ); @@ -780,6 +784,7 @@ pub fn ConfigInputView(props: &ConfigInputViewProps) -> Html { { config_field_custom!(translate.t(LABEL_RESOLVE_DELAY_SEC), input_options_state.form.resolve_delay.to_string()) }
{ config_field_bool!(input_options_state.form, translate.t(LABEL_RESOLVE_BACKGROUND), resolve_background) } + { config_field_optional!(input_options_state.form, translate.t(LABEL_RESOLVE_FILTER), resolve_filter) }
@@ -791,6 +796,7 @@ pub fn ConfigInputView(props: &ConfigInputViewProps) -> Html { { config_field_custom!(translate.t(LABEL_PROBE_DELAY_SEC), input_options_state.form.probe_delay.to_string()) } { config_field_custom!(translate.t(LABEL_PROBE_LIVE_INTERVAL_HOURS), input_options_state.form.probe_live_interval_hours.to_string()) }
+ { config_field_optional!(input_options_state.form, translate.t(LABEL_PROBE_FILTER), probe_filter) } })} @@ -840,6 +846,14 @@ pub fn ConfigInputView(props: &ConfigInputViewProps) -> Html { { edit_field_number_u16!(input_options_state, translate.t(LABEL_RESOLVE_DELAY_SEC), resolve_delay, ConfigInputOptionsFormAction::ResolveDelay) } { edit_field_bool!(input_options_state, translate.t(LABEL_RESOLVE_BACKGROUND), resolve_background, ConfigInputOptionsFormAction::ResolveBackground) } + { config_field_child!(translate.t(LABEL_RESOLVE_FILTER), "INPUT_FORM.RESOLVE_FILTER", { + let input_options_state_filter = input_options_state.clone(); + html! { + | { + input_options_state_filter.dispatch(ConfigInputOptionsFormAction::ResolveFilter(new_filter)); + })} /> + } + })}
@@ -851,6 +865,14 @@ pub fn ConfigInputView(props: &ConfigInputViewProps) -> Html { { edit_field_number_u16!(input_options_state, translate.t(LABEL_PROBE_DELAY_SEC), probe_delay, ConfigInputOptionsFormAction::ProbeDelay) } { edit_field_number_u32!(input_options_state, translate.t(LABEL_PROBE_LIVE_INTERVAL_HOURS), probe_live_interval_hours, ConfigInputOptionsFormAction::ProbeLiveIntervalHours) }
+ { config_field_child!(translate.t(LABEL_PROBE_FILTER), "INPUT_FORM.PROBE_FILTER", { + let input_options_state_filter = input_options_state.clone(); + html! { + | { + input_options_state_filter.dispatch(ConfigInputOptionsFormAction::ProbeFilter(new_filter)); + })} /> + } + })} })} diff --git a/frontend/src/hooks/use_server_status.rs b/frontend/src/hooks/use_server_status.rs index 084869cff..73ae57d63 100644 --- a/frontend/src/hooks/use_server_status.rs +++ b/frontend/src/hooks/use_server_status.rs @@ -321,7 +321,7 @@ mod tests { use shared::{ model::{ ActiveUserConnectionChange, DownloadsDelta, DownloadsResponse, FileDownloadDto, PlaylistItemType, - StatusCheck, StreamChannel, StreamInfo, TaskKindDto, TaskPriorityDto, TransferStatusDto, XtreamCluster, + StreamChannel, StreamInfo, TaskKindDto, TaskPriorityDto, TransferStatusDto, XtreamCluster, }, utils::Internable, }; @@ -393,13 +393,15 @@ mod tests { #[test] fn test_connections_zero_clears_stale_stream_rows() { - let mut status = StatusCheck::default(); - status.active_users = 1; - status.active_user_connections = 1; - status.active_user_streams = vec![ - test_stream(1, "127.0.0.1:1234", Some("tok-a"), PlaylistItemType::Video), - test_stream(2, "127.0.0.1:5678", Some("tok-b"), PlaylistItemType::Series), - ]; + let mut status = shared::model::StatusCheck { + active_users: 1, + active_user_connections: 1, + active_user_streams: vec![ + test_stream(1, "127.0.0.1:1234", Some("tok-a"), PlaylistItemType::Video), + test_stream(2, "127.0.0.1:5678", Some("tok-b"), PlaylistItemType::Series), + ], + ..Default::default() + }; apply_active_user_change(&mut status, ActiveUserConnectionChange::Connections(0, 0)); diff --git a/shared/src/model/config/input.rs b/shared/src/model/config/input.rs index 2cc61ff16..6b129d53a 100644 --- a/shared/src/model/config/input.rs +++ b/shared/src/model/config/input.rs @@ -2,7 +2,8 @@ use super::PanelApiConfigDto; use crate::{ check_input_connections, check_input_credentials, error::TuliproxError, - model::EpgConfigDto, + foundation::{get_filter, Filter}, + model::{EpgConfigDto, PatternTemplate}, utils::{ arc_str_serde, arc_str_vec_serde, default_as_true, default_probe_delay_secs, default_probe_live_interval, default_resolve_background, default_resolve_delay_secs, default_xtream_live_stream_use_prefix, @@ -222,6 +223,14 @@ pub struct ConfigInputOptionsDto { skip_serializing_if = "is_default_probe_live_interval" )] pub probe_live_interval_hours: u32, + #[serde(default, skip_serializing_if = "is_blank_optional_string")] + pub resolve_filter: Option, + #[serde(skip)] + pub t_resolve_filter: Option, + #[serde(default, skip_serializing_if = "is_blank_optional_string")] + pub probe_filter: Option, + #[serde(skip)] + pub t_probe_filter: Option, } impl Default for ConfigInputOptionsDto { @@ -242,6 +251,10 @@ impl Default for ConfigInputOptionsDto { probe_delay: default_probe_delay_secs(), probe_live: false, probe_live_interval_hours: default_probe_live_interval(), + resolve_filter: None, + t_resolve_filter: None, + probe_filter: None, + t_probe_filter: None, } } } @@ -263,6 +276,8 @@ impl ConfigInputOptionsDto { && is_default_probe_delay_secs(&self.probe_delay) && !self.probe_live && is_default_probe_live_interval(&self.probe_live_interval_hours) + && self.resolve_filter.as_ref().is_none_or(|s| s.trim().is_empty()) + && self.probe_filter.as_ref().is_none_or(|s| s.trim().is_empty()) } pub fn clean(&mut self) { @@ -281,6 +296,20 @@ impl ConfigInputOptionsDto { self.probe_delay = default_probe_delay_secs(); self.probe_live = false; self.probe_live_interval_hours = default_probe_live_interval(); + self.resolve_filter = None; + self.t_resolve_filter = None; + self.probe_filter = None; + self.t_probe_filter = None; + } + + pub fn prepare(&mut self, templates: Option<&[PatternTemplate]>) -> Result<(), TuliproxError> { + if let Some(raw_filter) = &self.resolve_filter { + self.t_resolve_filter = Some(get_filter(raw_filter, templates)?); + } + if let Some(raw_filter) = &self.probe_filter { + self.t_probe_filter = Some(get_filter(raw_filter, templates)?); + } + Ok(()) } } @@ -546,6 +575,7 @@ impl ConfigInputDto { index: u16, _include_computed: bool, provider_names: &HashSet, + templates: Option<&[PatternTemplate]>, ) -> Result { self.name = self.name.trim().intern(); if self.name.is_empty() { @@ -647,6 +677,11 @@ impl ConfigInputDto { } } + // Prepare filter options + if let Some(options) = self.options.as_mut() { + options.prepare(templates)?; + } + Ok(current_index) } @@ -1198,7 +1233,8 @@ mod tests { }; dto.prepare_type().expect("prepare type should succeed"); - dto.prepare(0, true, &HashSet::new()).expect("prepare should succeed and infer batch type from batch:// URL"); + dto.prepare(0, true, &HashSet::new(), None) + .expect("prepare should succeed and infer batch type from batch:// URL"); assert_eq!(dto.input_type, InputType::XtreamBatch); } @@ -1224,7 +1260,7 @@ mod tests { dto.prepare_type().expect("prepare type should normalize non-batch URL to xtream"); assert_eq!(dto.input_type, InputType::Xtream); - dto.prepare(0, true, &HashSet::new()).expect("prepare should succeed for regular URL with aliases"); + dto.prepare(0, true, &HashSet::new(), None).expect("prepare should succeed for regular URL with aliases"); assert_eq!(dto.input_type, InputType::Xtream); } @@ -1239,7 +1275,8 @@ mod tests { ..ConfigInputDto::default() }; - dto.prepare(0, true, &HashSet::new()).expect("batch:// input must be normalized before credential validation"); + dto.prepare(0, true, &HashSet::new(), None) + .expect("batch:// input must be normalized before credential validation"); assert_eq!(dto.input_type, InputType::XtreamBatch); } @@ -1264,7 +1301,7 @@ mod tests { }; let err = dto - .prepare(0, true, &HashSet::new()) + .prepare(0, true, &HashSet::new(), None) .expect_err("prepare should treat provider:// URL as regular input (non-batch) and validate provider"); assert!(err.to_string().contains("Provider name myprovider is not defined"), "Error: {err}"); } @@ -1290,7 +1327,7 @@ mod tests { }; let err = dto - .prepare(0, true, &HashSet::new()) + .prepare(0, true, &HashSet::new(), None) .expect_err("prepare must require root input url even when aliases are present"); assert!(err.to_string().contains("url for input is mandatory"), "Error: {err}"); assert!(err.to_string().contains("xtream_missing_root_url"), "Error: {err}"); @@ -1314,8 +1351,9 @@ mod tests { ..ConfigInputDto::default() }; - let err = - dto.prepare(0, true, &HashSet::new()).expect_err("prepare must require root credentials for non-batch URL"); + let err = dto + .prepare(0, true, &HashSet::new(), None) + .expect_err("prepare must require root credentials for non-batch URL"); assert!(err.to_string().contains("for input type xtream: username and password are mandatory"), "Error: {err}"); assert!(err.to_string().contains("xtream_batch_missing_root_creds"), "Error: {err}"); } @@ -1341,7 +1379,7 @@ mod tests { }; let err = dto - .prepare(0, true, &HashSet::new()) + .prepare(0, true, &HashSet::new(), None) .expect_err("prepare must reject root credentials when using batch:// for xtream-batch"); assert!(err.to_string().contains("with batch:// URL should not define username or password"), "Error: {err}"); assert!(err.to_string().contains("xtream_batch_with_root_creds"), "Error: {err}"); @@ -1376,7 +1414,8 @@ mod tests { ..StagedInputDto::default() }); - let err = dto.prepare(0, true, &HashSet::new()).expect_err("should reject vod_source=staged for M3U staged"); + let err = + dto.prepare(0, true, &HashSet::new(), None).expect_err("should reject vod_source=staged for M3U staged"); assert!(err.to_string().contains("Staged M3U input cannot provide VOD or Series"), "Error: {err}"); } @@ -1395,7 +1434,8 @@ mod tests { ..StagedInputDto::default() }); - let err = dto.prepare(0, true, &HashSet::new()).expect_err("should reject series_source=staged for M3U staged"); + let err = + dto.prepare(0, true, &HashSet::new(), None).expect_err("should reject series_source=staged for M3U staged"); assert!(err.to_string().contains("Staged M3U input cannot provide VOD or Series"), "Error: {err}"); } @@ -1418,7 +1458,7 @@ mod tests { ..StagedInputDto::default() }); - dto.prepare(0, true, &HashSet::new()).expect("xtream staged with all cluster sources should succeed"); + dto.prepare(0, true, &HashSet::new(), None).expect("xtream staged with all cluster sources should succeed"); } #[test] @@ -1441,8 +1481,9 @@ mod tests { ..StagedInputDto::default() }); - let err = - dto.prepare(0, true, &HashSet::new()).expect_err("expected validation error for missing staged source"); + let err = dto + .prepare(0, true, &HashSet::new(), None) + .expect_err("expected validation error for missing staged source"); assert!(err.to_string().contains("no cluster source uses 'staged'"), "Error: {err}"); } @@ -1468,7 +1509,7 @@ mod tests { }); let err = dto - .prepare(0, true, &HashSet::new()) + .prepare(0, true, &HashSet::new(), None) .expect_err("skipped staged cluster must not satisfy staged-source requirement"); assert!(err.to_string().contains("no cluster source uses 'staged'"), "Error: {err}"); } @@ -1489,7 +1530,7 @@ mod tests { ..StagedInputDto::default() }); - dto.prepare(0, true, &HashSet::new()) + dto.prepare(0, true, &HashSet::new(), None) .expect("staged M3U vod_source=staged is valid when VOD cluster is skipped"); } @@ -1511,7 +1552,7 @@ mod tests { ..StagedInputDto::default() }); - dto.prepare(0, true, &HashSet::new()) + dto.prepare(0, true, &HashSet::new(), None) .expect("disabled staged input should not enforce cluster source validation"); } @@ -1532,7 +1573,7 @@ mod tests { ..StagedInputDto::default() }); - dto.prepare(0, true, &HashSet::new()) + dto.prepare(0, true, &HashSet::new(), None) .expect("disabled staged input should not enforce staged M3U cluster validation"); } @@ -1551,7 +1592,7 @@ mod tests { ..StagedInputDto::default() }); - dto.prepare(0, true, &HashSet::new()) + dto.prepare(0, true, &HashSet::new(), None) .expect("disabled staged input should not enforce provider URL validation"); } @@ -1585,4 +1626,42 @@ mod tests { assert!(staged.vod_source.is_none()); assert!(staged.series_source.is_none()); } + + #[test] + fn test_config_input_options_dto_filter_prepare_parses_valid_filter() { + let mut dto = ConfigInputOptionsDto { + resolve_filter: Some(r#"name ~ "test""#.to_string()), + ..ConfigInputOptionsDto::default() + }; + dto.prepare(None).expect("valid filter should parse"); + assert!(dto.t_resolve_filter.is_some()); + } + + #[test] + fn test_config_input_options_dto_filter_prepare_rejects_invalid_filter() { + let mut dto = ConfigInputOptionsDto { + resolve_filter: Some(r#"name ~ "["#.to_string()), // invalid regex + ..ConfigInputOptionsDto::default() + }; + let result = dto.prepare(None); + assert!(result.is_err()); + } + + #[test] + fn test_config_input_options_dto_filter_prepare_with_unknown_template_placeholder() { + let mut dto = ConfigInputOptionsDto { + resolve_filter: Some(r#"name ~ "!UNKNOWN!""#.to_string()), + ..ConfigInputOptionsDto::default() + }; + let result = dto.prepare(None); + assert!(result.is_err()); + assert!(result.unwrap_err().to_string().contains("Unknown template placeholder")); + } + + #[test] + fn test_config_input_options_dto_filter_none_prepares_successfully() { + let mut dto = ConfigInputOptionsDto { resolve_filter: None, ..ConfigInputOptionsDto::default() }; + dto.prepare(None).expect("None filter should prepare successfully"); + assert!(dto.t_resolve_filter.is_none()); + } } diff --git a/shared/src/model/config/source.rs b/shared/src/model/config/source.rs index de32cf642..9a9e43c38 100644 --- a/shared/src/model/config/source.rs +++ b/shared/src/model/config/source.rs @@ -92,7 +92,7 @@ impl SourcesConfigDto { let mut input_credentials = HashSet::new(); // Prepare global inputs for input in &mut self.inputs { - input_index = input.prepare(input_index, include_computed, provider_names)?; + input_index = input.prepare(input_index, include_computed, provider_names, prepared_templates)?; if let (Some(username), Some(password)) = (input.username.as_ref(), input.password.as_ref()) { let key = (username, password); if !input_credentials.insert(key) { diff --git a/shared/src/model/config/stream.rs b/shared/src/model/config/stream.rs index 310e7f5a1..2d14b409b 100644 --- a/shared/src/model/config/stream.rs +++ b/shared/src/model/config/stream.rs @@ -70,7 +70,7 @@ pub struct StreamConfigDto { skip_serializing_if = "is_default_grace_period_timeout_secs" )] pub grace_period_timeout_secs: u64, - /// If true (default), wait for grace period check before streaming. + /// If true (default), wait for a grace period check before streaming. #[serde(default = "default_as_true", skip_serializing_if = "is_true")] pub grace_period_hold_stream: bool, #[serde(default = "default_hls_session_ttl_secs", skip_serializing_if = "is_default_hls_session_ttl_secs")] @@ -225,8 +225,7 @@ mod tests { #[test] fn test_empty_list_accepted() { - let mut dto = StreamConfigDto::default(); - dto.admission_strategies = Some(vec![]); + let mut dto = StreamConfigDto { admission_strategies: Some(vec![]), ..StreamConfigDto::default() }; assert!(dto.prepare().is_ok()); } @@ -238,9 +237,13 @@ mod tests { #[test] fn test_duplicate_rejected() { - let mut dto = StreamConfigDto::default(); - dto.admission_strategies = - Some(vec![AdmissionStrategy::EvictUserSameIpOldest, AdmissionStrategy::EvictUserSameIpOldest]); + let mut dto = StreamConfigDto { + admission_strategies: Some(vec![ + AdmissionStrategy::EvictUserSameIpOldest, + AdmissionStrategy::EvictUserSameIpOldest, + ]), + ..StreamConfigDto::default() + }; let err = dto.prepare().unwrap_err(); let msg = format!("{err}"); assert!(msg.contains("Duplicate admission strategy"), "msg: {msg}"); @@ -248,9 +251,10 @@ mod tests { #[test] fn test_mutually_exclusive_grace_rejected() { - let mut dto = StreamConfigDto::default(); - dto.admission_strategies = - Some(vec![AdmissionStrategy::GraceInstantStream, AdmissionStrategy::GraceHoldStream]); + let mut dto = StreamConfigDto { + admission_strategies: Some(vec![AdmissionStrategy::GraceInstantStream, AdmissionStrategy::GraceHoldStream]), + ..StreamConfigDto::default() + }; let err = dto.prepare().unwrap_err(); let msg = format!("{err}"); assert!(msg.contains("mutually exclusive"), "msg: {msg}"); @@ -258,32 +262,44 @@ mod tests { #[test] fn test_valid_ordered_subset_accepted() { - let mut dto = StreamConfigDto::default(); - dto.admission_strategies = Some(vec![AdmissionStrategy::EvictUserOldest, AdmissionStrategy::GraceHoldStream]); + let mut dto = StreamConfigDto { + admission_strategies: Some(vec![AdmissionStrategy::EvictUserOldest, AdmissionStrategy::GraceHoldStream]), + ..StreamConfigDto::default() + }; assert!(dto.prepare().is_ok()); } #[test] fn test_safe_ordering_same_ip_oldest_before_oldest_accepted() { - let mut dto = StreamConfigDto::default(); - dto.admission_strategies = - Some(vec![AdmissionStrategy::EvictUserSameIpOldest, AdmissionStrategy::EvictUserOldest]); + let mut dto = StreamConfigDto { + admission_strategies: Some(vec![ + AdmissionStrategy::EvictUserSameIpOldest, + AdmissionStrategy::EvictUserOldest, + ]), + ..StreamConfigDto::default() + }; assert!(dto.prepare().is_ok()); } #[test] fn test_safe_ordering_same_ip_latest_before_latest_accepted() { - let mut dto = StreamConfigDto::default(); - dto.admission_strategies = - Some(vec![AdmissionStrategy::EvictUserSameIpLatest, AdmissionStrategy::EvictUserLatest]); + let mut dto = StreamConfigDto { + admission_strategies: Some(vec![ + AdmissionStrategy::EvictUserSameIpLatest, + AdmissionStrategy::EvictUserLatest, + ]), + ..StreamConfigDto::default() + }; assert!(dto.prepare().is_ok()); } #[test] fn test_grace_strategy_requires_positive_grace_period() { - let mut dto = StreamConfigDto::default(); - dto.grace_period_millis = 0; - dto.admission_strategies = Some(vec![AdmissionStrategy::GraceHoldStream]); + let mut dto = StreamConfigDto { + grace_period_millis: 0, + admission_strategies: Some(vec![AdmissionStrategy::GraceHoldStream]), + ..StreamConfigDto::default() + }; let err = dto.prepare().unwrap_err(); let msg = format!("{err}"); assert!(msg.contains("grace_period_millis"), "msg: {msg}"); @@ -291,9 +307,13 @@ mod tests { #[test] fn test_shadowed_same_ip_oldest_order_rejected() { - let mut dto = StreamConfigDto::default(); - dto.admission_strategies = - Some(vec![AdmissionStrategy::EvictUserOldest, AdmissionStrategy::EvictUserSameIpOldest]); + let mut dto = StreamConfigDto { + admission_strategies: Some(vec![ + AdmissionStrategy::EvictUserOldest, + AdmissionStrategy::EvictUserSameIpOldest, + ]), + ..StreamConfigDto::default() + }; let err = dto.prepare().unwrap_err(); let msg = format!("{err}"); assert!(msg.contains("shadowed"), "msg: {msg}"); @@ -301,9 +321,13 @@ mod tests { #[test] fn test_shadowed_same_ip_latest_order_rejected() { - let mut dto = StreamConfigDto::default(); - dto.admission_strategies = - Some(vec![AdmissionStrategy::EvictUserLatest, AdmissionStrategy::EvictUserSameIpLatest]); + let mut dto = StreamConfigDto { + admission_strategies: Some(vec![ + AdmissionStrategy::EvictUserLatest, + AdmissionStrategy::EvictUserSameIpLatest, + ]), + ..StreamConfigDto::default() + }; let err = dto.prepare().unwrap_err(); let msg = format!("{err}"); assert!(msg.contains("shadowed"), "msg: {msg}"); @@ -311,9 +335,13 @@ mod tests { #[test] fn test_cross_order_pair_with_different_selection_policy_is_rejected() { - let mut dto = StreamConfigDto::default(); - dto.admission_strategies = - Some(vec![AdmissionStrategy::EvictUserOldest, AdmissionStrategy::EvictUserSameIpLatest]); + let mut dto = StreamConfigDto { + admission_strategies: Some(vec![ + AdmissionStrategy::EvictUserOldest, + AdmissionStrategy::EvictUserSameIpLatest, + ]), + ..StreamConfigDto::default() + }; let err = dto.prepare().unwrap_err(); let msg = format!("{err}"); assert!(msg.contains("shadowed"), "msg: {msg}"); @@ -329,23 +357,24 @@ mod tests { AdmissionStrategy::GraceInstantStream, AdmissionStrategy::GraceHoldStream, ] { - let mut dto = StreamConfigDto::default(); - dto.admission_strategies = Some(vec![s]); + let mut dto = StreamConfigDto { admission_strategies: Some(vec![s]), ..StreamConfigDto::default() }; assert!(dto.prepare().is_ok(), "Failed for {s:?}"); } } #[test] fn test_is_empty_with_strategies() { - let mut dto = StreamConfigDto::default(); - dto.admission_strategies = Some(vec![AdmissionStrategy::GraceInstantStream]); + let dto = StreamConfigDto { + admission_strategies: Some(vec![AdmissionStrategy::GraceInstantStream]), + ..StreamConfigDto::default() + }; + assert!(!dto.is_empty()); } #[test] fn test_is_empty_with_explicit_empty_strategy_list() { - let mut dto = StreamConfigDto::default(); - dto.admission_strategies = Some(vec![]); + let dto = StreamConfigDto { admission_strategies: Some(vec![]), ..StreamConfigDto::default() }; assert!(!dto.is_empty()); } diff --git a/shared/src/model/xtream.rs b/shared/src/model/xtream.rs index a349310e9..8f6bf757a 100644 --- a/shared/src/model/xtream.rs +++ b/shared/src/model/xtream.rs @@ -256,33 +256,6 @@ pub struct XtreamSeriesInfo { pub episodes: Option>, } -#[cfg(test)] -mod tests { - use super::XtreamVideoInfo; - use serde_json::json; - - #[test] - fn xtream_video_info_accepts_numeric_tmdb_id() { - let parsed: XtreamVideoInfo = serde_json::from_value(json!({ - "info": { - "tmdb_id": 1285728, - "name": "A Normal Woman" - }, - "movie_data": { - "stream_id": 982447, - "name": "A Normal Woman - 2025", - "added": "1771006438", - "category_id": "727", - "container_extension": "mkv", - "direct_source": "" - } - })) - .expect("numeric tmdb_id should deserialize"); - - assert_eq!(parsed.info.tmdb_id.as_ref(), "1285728"); - } -} - // sometimes episodes are a map with season as key, sometimes an array fn deserialize_episodes<'de, D>(deserializer: D) -> Result>, D::Error> where @@ -360,3 +333,30 @@ where } } } + +#[cfg(test)] +mod tests { + use super::XtreamVideoInfo; + use serde_json::json; + + #[test] + fn xtream_video_info_accepts_numeric_tmdb_id() { + let parsed: XtreamVideoInfo = serde_json::from_value(json!({ + "info": { + "tmdb_id": 1285728, + "name": "A Normal Woman" + }, + "movie_data": { + "stream_id": 982447, + "name": "A Normal Woman - 2025", + "added": "1771006438", + "category_id": "727", + "container_extension": "mkv", + "direct_source": "" + } + })) + .expect("numeric tmdb_id should deserialize"); + + assert_eq!(parsed.info.tmdb_id.as_ref(), "1285728"); + } +} diff --git a/shared/src/utils/flags.rs b/shared/src/utils/flags.rs index 98d647d42..4d79149b0 100644 --- a/shared/src/utils/flags.rs +++ b/shared/src/utils/flags.rs @@ -346,6 +346,7 @@ mod tests { let mut block_table = vec![FlagBlockEntry { offset: 0, compressed_len: 0, raw_len: 0 }; BLOCK_COUNT]; let mut compressed_blocks = Vec::new(); + #[allow(clippy::needless_range_loop)] for block_id in 0..BLOCK_COUNT { let mut raw_block = Vec::new(); for (code, svg) in flags {