diff --git a/CHANGELOG.md b/CHANGELOG.md index d809b8131..519576d38 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,75 @@ ## 🌟 New Features +- **UI Micro-Interactions**: + - Cards now gently lift with a soft shadow on hover, buttons give a subtle press/ripple feedback when clicked, + and collapse/accordion chevrons smoothly rotate between open and closed states. All effects honor + `prefers-reduced-motion: reduce`. + +- **Animated Theme Transitions**: + - Switching themes now cross-dissolves colors, backgrounds, borders, and shadows instead of snapping instantly. + The transition is applied only during the switch (never during normal hover/interaction) and is fully disabled + for users who set `prefers-reduced-motion: reduce`. + +- **Resilient Sidebar Initialization**: + - The sidebar no longer panics if the global `window` object is unavailable; resize handling and responsive + collapse now degrade gracefully instead of crashing the app. + +- **Improved Screen-Reader Support**: + - Toast notifications are now announced by screen readers via an `aria-live` region (errors assertively, others + politely). + - The sidebar toggle button exposes its `aria-expanded` state and an accessible label, and the active navigation + item is marked with `aria-current="page"`. + +- **Table Empty-State Message**: + - Paged/data tables now show a localized "No content" message in their empty state instead of just an icon, + making it clearer when a query or filter returns no rows. (Table headers already stick to the top while scrolling.) + +- **Debounced Filter Editor Input**: + - Typing in the filter editor no longer re-parses and previews the filter on every keystroke; parsing and change + notifications are now debounced, keeping the textarea responsive while editing large filters. + +- **Persisted UI Preferences**: + - The sidebar collapsed/expanded state and the stream-history table page size are now remembered across sessions + (the active theme was already persisted), so the UI restores your last layout on reload. + +- **Explicit Button Type On Shared Controls**: + - The shared `IconButton` and `TextButton` primitives now render with `type="button"`, preventing accidental + form submission when used inside forms. + +- **Descriptive Image Alt Text**: + - Logo images on the login screen and sidebar now use the configured app title for their alt text, and playlist + channel logos use the channel title, improving screen-reader accessibility. + +- **Popup Menu Keyboard Dismissal**: + - Popup menus now close when pressing `Escape`, in addition to clicking outside, improving keyboard accessibility. + +- **Recoverable Error Boundary**: + - Added an `ErrorBoundary` component that wraps each main view (dashboard, stats, streams, downloads, users, + sources, playlists, EPG, RBAC, config) and the API-user playlist, so a recoverable failure shows a fallback with + a retry button instead of leaving the section blank — and a failure in one view no longer affects the others. + - Descendants can report a recoverable error through the boundary's context handle; retrying re-mounts the + protected subtree so the user can recover without reloading the whole app. + +- **Runtime-Discovered UI Languages With RTL Support**: + - The frontend now reads the available UI languages at runtime from an `assets/i18n/index.json` manifest, so adding a + language only requires shipping a `.json` locale file and adding an entry to the manifest — no code change. + - A language picker appears in the toolbar (next to the theme picker) whenever more than one language is available, + and the chosen language is remembered across sessions. + - Each language declares its text direction (`ltr`/`rtl`); the document direction is updated accordingly, providing + baseline support for right-to-left languages such as Arabic. + +- **Confirmation For Destructive Download/Recording Actions**: + - Cancelling a transfer/recording and removing a download entry now prompt a confirmation dialog before proceeding. + - The confirmation dialog focuses the safe (Cancel) action by default, consistent with other destructive actions + (delete user, delete target, delete RBAC user/group). + +- **Toast Notification UX Upgrade**: + - Auto-dismiss toasts now show a countdown progress bar that reflects the remaining time before they disappear. + - Hovering a toast pauses both the dismiss timer and its progress bar, and resumes from where it left off on mouse leave. + - Error toasts gain a "copy details" action that copies the full message to the clipboard for easier bug reports. + - The progress bar and entrance animation respect the `prefers-reduced-motion` accessibility setting. + - **M3U Catchup / Archive Preservation And Proxying**: - Tuliprox now preserves standard M3U catchup/archive attributes during M3U import and export. - Supported preserved attributes include: @@ -193,6 +262,8 @@ ## 🐛 Fixes +- **Async Local File Serving**: The local-file stream handler now canonicalizes paths with `tokio::fs::canonicalize` + instead of the blocking `std` call, so the async runtime is no longer blocked while resolving the file path. - **Template Expansion Efficiency**: Optimized `template.yml` / `template.d` multi-template expansion so sequence-style templates no longer duplicate unrelated entries during dependency resolution. - Sequence templates still resolve correctly and preserve order. diff --git a/backend/src/api/api_utils.rs b/backend/src/api/api_utils.rs index 277e04eb8..bf0e2b390 100644 --- a/backend/src/api/api_utils.rs +++ b/backend/src/api/api_utils.rs @@ -296,18 +296,18 @@ pub(crate) fn record_connect_failed_attempt(attempt: ConnectFailedAttempt<'_>) { .and_then(|value| value.to_str().ok()) .unwrap_or_default() .to_string(); - let info = StreamInfo::new( - 0, - 0, - &attempt.user.username, - &attempt.fingerprint.addr, - &attempt.fingerprint.client_ip, - attempt.provider_name, - attempt.stream_channel, + let info = StreamInfo::new(shared::model::StreamInfoParams { + uid: 0, + meter_uid: 0, + username: &attempt.user.username, + addr: &attempt.fingerprint.addr, + client_ip: &attempt.fingerprint.client_ip, + provider: attempt.provider_name, + stream_channel: attempt.stream_channel, user_agent, - None, - None, - ); + country_code: None, + session_token: None, + }); // Resolve target_name from target_id using the stable target config name. let target_name = attempt.app_state.app_config.get_target_by_id(info.channel.target_id).as_deref().map(|t| (&t.name).intern()); @@ -2952,7 +2952,7 @@ pub(crate) async fn local_stream_response( let path = PathBuf::from(pli.url.strip_prefix("file://").unwrap_or(&pli.url)); // Canonicalize and validate the path - let path = match path.canonicalize() { + let path = match tokio::fs::canonicalize(&path).await { Ok(canonical) => canonical, Err(err) => { error!("Local file path is corrupt {}: {err}", path.display()); diff --git a/backend/src/api/endpoints/hdhomerun_api.rs b/backend/src/api/endpoints/hdhomerun_api.rs index 776bd55c4..5de818cc9 100644 --- a/backend/src/api/endpoints/hdhomerun_api.rs +++ b/backend/src/api/endpoints/hdhomerun_api.rs @@ -93,6 +93,20 @@ impl Device { } } +#[inline] +#[allow(clippy::unnecessary_wraps)] +fn serialize_lineup(lineup: &Lineup, has_next: bool) -> Result { + match serde_json::to_string(lineup) { + Ok(mut content) => { + if has_next { + content.push(','); + } + Ok(Bytes::from(content)) + } + Err(_) => Ok(Bytes::from("")), + } +} + fn xtream_item_to_lineup_stream( cfg: Arc, cluster: XtreamCluster, @@ -135,15 +149,7 @@ where guide_name: item.title.clone(), url: stream_url.clone(), }; - match serde_json::to_string(&lineup) { - Ok(mut content) => { - if has_next { - content.push(','); - } - Ok(Bytes::from(content)) - } - Err(_) => Ok(Bytes::from("")), - } + serialize_lineup(&lineup, has_next) }); mapped.left_stream() } @@ -163,15 +169,7 @@ where guide_name: item.title.clone(), url: if item.t_stream_url.is_empty() { item.url.clone() } else { item.t_stream_url.clone() }, }; - match serde_json::to_string(&lineup) { - Ok(mut content) => { - if has_next { - content.push(','); - } - Ok(Bytes::from(content)) - } - Err(_) => Ok(Bytes::from("")), - } + serialize_lineup(&lineup, has_next) }); mapped.left_stream() } diff --git a/backend/src/api/endpoints/rbac_api.rs b/backend/src/api/endpoints/rbac_api.rs index 9f2eaa093..3deea30d7 100644 --- a/backend/src/api/endpoints/rbac_api.rs +++ b/backend/src/api/endpoints/rbac_api.rs @@ -267,6 +267,36 @@ fn store_reprepared_web_auth(app_state: &AppState) -> Result<(), String> { Ok(()) } +async fn save_and_reprepare_auth_file( + app_state: &Arc, + file_path: &FsPath, + content: &str, +) -> Result<(), (StatusCode, Json)> { + let _lock = app_state.app_config.file_locks.write_lock(file_path).await; + let previous_content = match tokio::fs::read_to_string(file_path).await { + Ok(content) => Some(content), + Err(err) if err.kind() == std::io::ErrorKind::NotFound => None, + Err(err) => return Err((StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": err.to_string()})))), + }; + if let Err(err) = write_text_file_atomic(file_path, content).await { + return Err((StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": err.to_string()})))); + } + if let Err(err) = store_reprepared_web_auth(app_state) { + let rollback_result = match previous_content { + Some(previous_content) => write_text_file_atomic(file_path, &previous_content).await, + None => tokio::fs::remove_file(file_path).await, + }; + if let Err(rollback_err) = rollback_result { + return Err(( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"error": format!("{err}; failed to restore previous auth file: {rollback_err}")})), + )); + } + return Err((StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": err})))); + } + Ok(()) +} + fn token_from_extensions_or_headers(request: &mut axum::extract::Request) -> Result { if let Some(token) = request.extensions().get::().cloned() { return Ok(token); @@ -429,12 +459,8 @@ async fn create_user( }); let (userfile_path, _) = resolve_auth_paths(&web_auth, &config_path); - let _lock = app_state.app_config.file_locks.write_lock(&userfile_path).await; - if let Err(err) = write_text_file_atomic(&userfile_path, &serialize_users_file(&users)).await { - return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": err.to_string()}))).into_response(); - } - if let Err(err) = store_reprepared_web_auth(&app_state) { - return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": err}))).into_response(); + if let Err(response) = save_and_reprepare_auth_file(&app_state, &userfile_path, &serialize_users_file(&users)).await { + return response.into_response(); } info!("RBAC API: created web UI user '{username}'"); @@ -507,12 +533,8 @@ async fn update_user( } let (userfile_path, _) = resolve_auth_paths(&web_auth, &config_path); - let _lock = app_state.app_config.file_locks.write_lock(&userfile_path).await; - if let Err(err) = write_text_file_atomic(&userfile_path, &serialize_users_file(&users)).await { - return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": err.to_string()}))).into_response(); - } - if let Err(err) = store_reprepared_web_auth(&app_state) { - return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": err}))).into_response(); + if let Err(response) = save_and_reprepare_auth_file(&app_state, &userfile_path, &serialize_users_file(&users)).await { + return response.into_response(); } info!("RBAC API: updated web UI user '{username}'"); @@ -550,12 +572,8 @@ async fn delete_user( users.remove(user_index); let (userfile_path, _) = resolve_auth_paths(&web_auth, &config_path); - let _lock = app_state.app_config.file_locks.write_lock(&userfile_path).await; - if let Err(err) = write_text_file_atomic(&userfile_path, &serialize_users_file(&users)).await { - return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": err.to_string()}))).into_response(); - } - if let Err(err) = store_reprepared_web_auth(&app_state) { - return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": err}))).into_response(); + if let Err(response) = save_and_reprepare_auth_file(&app_state, &userfile_path, &serialize_users_file(&users)).await { + return response.into_response(); } info!("RBAC API: deleted web UI user '{username}'"); @@ -597,12 +615,8 @@ async fn create_group( }); let (_, groupfile_path) = resolve_auth_paths(&web_auth, &config_path); - let _lock = app_state.app_config.file_locks.write_lock(&groupfile_path).await; - if let Err(err) = write_text_file_atomic(&groupfile_path, &serialize_groups_file(&groups)).await { - return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": err.to_string()}))).into_response(); - } - if let Err(err) = store_reprepared_web_auth(&app_state) { - return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": err}))).into_response(); + if let Err(response) = save_and_reprepare_auth_file(&app_state, &groupfile_path, &serialize_groups_file(&groups)).await { + return response.into_response(); } info!("RBAC API: created group '{name}'"); @@ -642,12 +656,8 @@ async fn update_group( groups[group_index].permissions = normalize_permissions(&request.permissions); let (_, groupfile_path) = resolve_auth_paths(&web_auth, &config_path); - let _lock = app_state.app_config.file_locks.write_lock(&groupfile_path).await; - if let Err(err) = write_text_file_atomic(&groupfile_path, &serialize_groups_file(&groups)).await { - return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": err.to_string()}))).into_response(); - } - if let Err(err) = store_reprepared_web_auth(&app_state) { - return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": err}))).into_response(); + if let Err(response) = save_and_reprepare_auth_file(&app_state, &groupfile_path, &serialize_groups_file(&groups)).await { + return response.into_response(); } info!("RBAC API: updated group '{name}'"); @@ -696,12 +706,8 @@ async fn delete_group( groups.remove(group_index); let (_, groupfile_path) = resolve_auth_paths(&web_auth, &config_path); - let _lock = app_state.app_config.file_locks.write_lock(&groupfile_path).await; - if let Err(err) = write_text_file_atomic(&groupfile_path, &serialize_groups_file(&groups)).await { - return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": err.to_string()}))).into_response(); - } - if let Err(err) = store_reprepared_web_auth(&app_state) { - return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": err}))).into_response(); + if let Err(response) = save_and_reprepare_auth_file(&app_state, &groupfile_path, &serialize_groups_file(&groups)).await { + return response.into_response(); } info!("RBAC API: deleted group '{name}'"); diff --git a/backend/src/api/endpoints/v1_api_config.rs b/backend/src/api/endpoints/v1_api_config.rs index 544d0aaea..b29ac1702 100644 --- a/backend/src/api/endpoints/v1_api_config.rs +++ b/backend/src/api/endpoints/v1_api_config.rs @@ -96,6 +96,8 @@ fn decode_permissions(app_state: &AppState, token: &str) -> Option { - api_proxy_dto.user = vec![]; + filter_api_proxy_by_permissions(&mut api_proxy_dto, PermissionSet::new()); let response = axum::response::Json(api_proxy_dto).into_response(); return response_with_revision_header(response, HEADER_CONFIG_API_PROXY_REVISION, &revision); } @@ -367,78 +371,7 @@ async fn save_config_api_proxy_config( response_with_revision_header(StatusCode::OK.into_response(), HEADER_CONFIG_API_PROXY_REVISION, &updated_revision) } -async fn config_public(axum::extract::State(app_state): axum::extract::State>) -> impl IntoResponse + Send { - let (config_file_path, sources_file_path, api_proxy_file_path) = { - let paths = app_state.app_config.paths.load(); - ( - paths.config_file_path.clone(), - paths.sources_file_path.clone(), - paths.api_proxy_file_path.clone(), - ) - }; - - let main_revision = match read_file_revision(&config_file_path).await { - Ok(revision) => revision, - Err(err) => { - error!("Failed to read revision for config.yml '{config_file_path}': {err}"); - return internal_server_error!(); - } - }; - let sources_revision = match read_file_revision(&sources_file_path).await { - Ok(revision) => revision, - Err(err) => { - error!("Failed to read revision for source.yml '{sources_file_path}': {err}"); - return internal_server_error!(); - } - }; - let api_proxy_revision = match read_file_revision(&api_proxy_file_path).await { - Ok(revision) => revision, - Err(err) => { - error!("Failed to read revision for api-proxy.yml '{api_proxy_file_path}': {err}"); - return internal_server_error!(); - } - }; - - let read_result = { - let paths = app_state.app_config.paths.load(); - utils::read_app_config_dto(&paths, true, false).await - }; - match read_result { - Ok(mut app_config) => { - if let Err(err) = prepare_sources_batch(&mut app_config.sources, false).await { - error!("Failed to prepare sources batch: {err}"); - internal_server_error!() - } else if let Err(err) = prepare_users(&mut app_config, &app_state.app_config).await { - error!("Failed to prepare users: {err}"); - internal_server_error!() - } else { - let response = axum::response::Json(app_config).into_response(); - let response = response_with_revision_header(response, HEADER_CONFIG_MAIN_REVISION, &main_revision); - let response = response_with_revision_header(response, HEADER_CONFIG_SOURCES_REVISION, &sources_revision); - response_with_revision_header(response, HEADER_CONFIG_API_PROXY_REVISION, &api_proxy_revision) - } - } - Err(err) => { - error!("Failed to read config files: {err}"); - internal_server_error!() - } - } -} - -async fn config( - AuthBearer(token): AuthBearer, - axum::extract::State(app_state): axum::extract::State>, -) -> impl IntoResponse + Send { - let Some(permissions) = decode_permissions(&app_state, &token) else { - return axum::http::StatusCode::UNAUTHORIZED.into_response(); - }; - if !has_any_permission( - permissions, - &[Permission::ConfigRead, Permission::SourceRead, Permission::UserRead], - ) { - return axum::http::StatusCode::FORBIDDEN.into_response(); - } - +async fn get_config_common(app_state: &Arc, permissions: PermissionSet) -> axum::response::Response { let (config_file_path, sources_file_path, api_proxy_file_path) = { let paths = app_state.app_config.paths.load(); ( @@ -497,6 +430,27 @@ async fn config( } } +async fn config_public(axum::extract::State(app_state): axum::extract::State>) -> impl IntoResponse + Send { + get_config_common(&app_state, PermissionSet::new()).await +} + +async fn config( + AuthBearer(token): AuthBearer, + axum::extract::State(app_state): axum::extract::State>, +) -> impl IntoResponse + Send { + let Some(permissions) = decode_permissions(&app_state, &token) else { + return axum::http::StatusCode::UNAUTHORIZED.into_response(); + }; + if !has_any_permission( + permissions, + &[Permission::ConfigRead, Permission::SourceRead, Permission::UserRead], + ) { + return axum::http::StatusCode::FORBIDDEN.into_response(); + } + + get_config_common(&app_state, permissions).await +} + async fn get_config_api_proxy_config( AuthBearer(token): AuthBearer, axum::extract::State(app_state): axum::extract::State>, @@ -731,6 +685,26 @@ mod tests { assert!(result.is_none(), "exact revision match should be accepted"); } + fn make_test_api_proxy() -> ApiProxyConfigDto { + ApiProxyConfigDto { + server: vec![ApiProxyServerInfoDto { + name: String::from("main"), + protocol: String::from("http"), + host: String::from("localhost"), + port: None, + timezone: String::from("UTC"), + message: String::from("hello"), + path: None, + }], + user: vec![TargetUserDto { + target: String::from("target-a"), + credentials: vec![], + }], + use_user_db: true, + auth_error_status: 401, + } + } + #[test] fn filter_app_config_clears_unauthorized_sections() { let permissions: PermissionSet = Permission::ConfigRead.into(); @@ -747,23 +721,7 @@ mod tests { }, mappings: None, templates: Some(TemplateDefinitionDto::default()), - api_proxy: Some(ApiProxyConfigDto { - server: vec![ApiProxyServerInfoDto { - name: String::from("main"), - protocol: String::from("http"), - host: String::from("localhost"), - port: None, - timezone: String::from("UTC"), - message: String::from("hello"), - path: None, - }], - user: vec![TargetUserDto { - target: String::from("target-a"), - credentials: vec![], - }], - use_user_db: true, - auth_error_status: 401, - }), + api_proxy: Some(make_test_api_proxy()), }; filter_app_config_by_permissions(&mut app_config, permissions); @@ -781,29 +739,40 @@ mod tests { #[test] fn filter_api_proxy_keeps_user_section_only_with_user_read() { let permissions: PermissionSet = Permission::UserRead.into(); - let mut api_proxy = ApiProxyConfigDto { - server: vec![ApiProxyServerInfoDto { - name: String::from("main"), - protocol: String::from("http"), - host: String::from("localhost"), - port: None, - timezone: String::from("UTC"), - message: String::from("hello"), - path: None, - }], - user: vec![TargetUserDto { - target: String::from("target-a"), - credentials: vec![], - }], - use_user_db: true, - auth_error_status: 401, - }; + let mut api_proxy = make_test_api_proxy(); filter_api_proxy_by_permissions(&mut api_proxy, permissions); assert!(api_proxy.server.is_empty()); assert_eq!(api_proxy.user.len(), 1); - assert!(api_proxy.use_user_db); + assert!(!api_proxy.use_user_db); + assert_eq!(api_proxy.auth_error_status, 403); + } + + #[test] + fn public_permissions_scrub_all_config_sections() { + let mut app_config = AppConfigDto { + config: ConfigDto { + storage_dir: Some(String::from("storage")), + ..ConfigDto::default() + }, + sources: SourcesConfigDto { + inputs: vec![], + sources: vec![], + provider: Some(vec![]), + templates: Some(vec![]), + }, + mappings: None, + templates: Some(TemplateDefinitionDto::default()), + api_proxy: Some(make_test_api_proxy()), + }; + + filter_app_config_by_permissions(&mut app_config, PermissionSet::new()); + + assert_eq!(app_config.config, ConfigDto::default()); + assert_eq!(app_config.sources, SourcesConfigDto::default()); + assert!(app_config.templates.is_none()); + assert_eq!(app_config.api_proxy, Some(ApiProxyConfigDto::default())); } #[test] diff --git a/backend/src/api/model/active_provider_manager.rs b/backend/src/api/model/active_provider_manager.rs index 28c685c74..7d23611c4 100644 --- a/backend/src/api/model/active_provider_manager.rs +++ b/backend/src/api/model/active_provider_manager.rs @@ -62,6 +62,13 @@ enum PriorityOwner { Shared(SharedConnectionId), } +struct AcquireProviderParams<'a> { + addr: &'a SocketAddr, + priority: i8, + kind: ConnectionKind, + session_owner: Option<&'a str>, +} + #[derive(Debug, Clone)] pub struct ProviderHandle { pub client_id: ClientConnectionId, @@ -333,44 +340,37 @@ impl ActiveProviderManager { async fn acquire_exact_connection_inner( &self, provider_name: &Arc, - addr: &SocketAddr, allow_grace: bool, - priority: i8, - kind: ConnectionKind, - session_owner: Option<&str>, + params: &AcquireProviderParams<'_>, ) -> Option { - if self.is_reserved_for_other(provider_name, session_owner).await { + if self.is_reserved_for_other(provider_name, params.session_owner).await { return None; } let allocation = self.providers.acquire_exact_connection_with_grace_override(provider_name, allow_grace).await; if matches!(allocation, ProviderAllocation::Exhausted) { return None; } - self.register_allocation(allocation, addr, priority, kind).await + self.register_allocation(allocation, params.addr, params.priority, params.kind).await } - #[allow(clippy::too_many_arguments)] async fn finalize_lineup_allocation( &self, input_name: &Arc, - addr: &SocketAddr, allow_grace: bool, - priority: i8, - kind: ConnectionKind, allocation: ProviderAllocation, - session_owner: Option<&str>, + params: &AcquireProviderParams<'_>, ) -> Option { if matches!(&allocation, ProviderAllocation::GracePeriod(_)) - && self.evict_lower_priority_on_input(input_name, priority, kind, session_owner).await + && self.evict_lower_priority_on_input(input_name, params.priority, params.kind, params.session_owner).await { let evicted_on_same = !self.providers.is_over_limit(&allocation.get_provider_name().unwrap_or_default()).await; if !evicted_on_same { let new_alloc = self.providers.acquire_connection_with_grace_override(input_name, allow_grace).await; if !matches!(new_alloc, ProviderAllocation::Exhausted) { if let Some(provider_name) = new_alloc.get_provider_name() { - if !self.is_reserved_for_other(&provider_name, session_owner).await { + if !self.is_reserved_for_other(&provider_name, params.session_owner).await { allocation.release().await; - return self.register_allocation(new_alloc, addr, priority, kind).await; + return self.register_allocation(new_alloc, params.addr, params.priority, params.kind).await; } } new_alloc.release().await; @@ -378,7 +378,7 @@ impl ActiveProviderManager { } } - self.register_allocation(allocation, addr, priority, kind).await + self.register_allocation(allocation, params.addr, params.priority, params.kind).await } fn select_victim_from_index( @@ -494,16 +494,12 @@ impl ActiveProviderManager { } } - #[allow(clippy::too_many_arguments)] async fn acquire_connection_inner( &self, provider_or_input_name: &Arc, - addr: &SocketAddr, force: bool, allow_grace_override: Option, - priority: i8, - kind: ConnectionKind, - session_owner: Option<&str>, + params: &AcquireProviderParams<'_>, ) -> Option { let allow_grace = if force { true } else { allow_grace_override.unwrap_or(true) }; @@ -512,13 +508,13 @@ impl ActiveProviderManager { if matches!(allocation, ProviderAllocation::Exhausted) { return None; } - return self.register_allocation(allocation, addr, priority, kind).await; + return self.register_allocation(allocation, params.addr, params.priority, params.kind).await; } - if let Some(owner) = session_owner { + if let Some(owner) = params.session_owner { if let Some(reserved_provider) = self.get_reserved_provider_for_owner(provider_or_input_name, owner).await { return self - .acquire_exact_connection_inner(&reserved_provider, addr, allow_grace, priority, kind, Some(owner)) + .acquire_exact_connection_inner(&reserved_provider, allow_grace, params) .await; } } @@ -532,11 +528,11 @@ impl ActiveProviderManager { break; } if let Some(provider_name) = allocation.get_provider_name() { - if self.is_reserved_for_other(&provider_name, session_owner).await { + if self.is_reserved_for_other(&provider_name, params.session_owner).await { debug_if_enabled!( "Skipping reserved provider {} for {}", sanitize_sensitive_info(&provider_name), - sanitize_sensitive_info(&addr.to_string()) + sanitize_sensitive_info(¶ms.addr.to_string()) ); skipped_reserved.insert(provider_name); allocation.release().await; @@ -549,21 +545,18 @@ impl ActiveProviderManager { return self .finalize_lineup_allocation( provider_or_input_name, - addr, allow_grace, - priority, - kind, allocation, - session_owner, + params, ) .await; } if let Some(preempted_alloc) = self - .try_preempt_connection(provider_or_input_name, priority, allow_grace, kind, session_owner) + .try_preempt_connection(provider_or_input_name, params.priority, allow_grace, params.kind, params.session_owner) .await { - return self.register_allocation(preempted_alloc, addr, priority, kind).await; + return self.register_allocation(preempted_alloc, params.addr, params.priority, params.kind).await; } None @@ -911,7 +904,9 @@ impl ActiveProviderManager { kind: ConnectionKind, session_owner: Option<&str>, ) -> Option { - self.acquire_exact_connection_inner(provider_name, addr, allow_grace, priority, kind, session_owner).await + self.acquire_exact_connection_inner(provider_name, allow_grace, &AcquireProviderParams { + addr, priority, kind, session_owner, + }).await } pub async fn force_exact_acquire_connection( @@ -933,7 +928,9 @@ impl ActiveProviderManager { priority: i8, kind: ConnectionKind, ) -> Option { - self.acquire_connection_inner(input_name, addr, false, None, priority, kind, None).await + self.acquire_connection_inner(input_name, false, None, &AcquireProviderParams { + addr, priority, kind, session_owner: None, + }).await } /// Acquire a provider connection while explicitly controlling provider-side grace allocations. @@ -957,20 +954,26 @@ impl ActiveProviderManager { kind: ConnectionKind, session_owner: Option<&str>, ) -> Option { - self.acquire_connection_inner(input_name, addr, false, Some(allow_grace), priority, kind, session_owner).await + self.acquire_connection_inner(input_name, false, Some(allow_grace), &AcquireProviderParams { + addr, priority, kind, session_owner, + }).await } /// Acquire a provider connection for probe tasks with configurable priority. /// Probes never consume grace capacity. pub async fn acquire_connection_for_probe(&self, input_name: &Arc, priority: i8) -> Option { - self.acquire_connection_inner(input_name, &DUMMY_ADDR, false, Some(false), priority, ConnectionKind::Normal, None).await + self.acquire_connection_inner(input_name, false, Some(false), &AcquireProviderParams { + addr: &DUMMY_ADDR, priority, kind: ConnectionKind::Normal, session_owner: None, + }).await } /// Acquire a provider connection for background transfers (downloads/recordings). /// Transfers participate in the same provider priority/preemption model as normal /// streams, but they never consume grace capacity and wait externally on notifications. pub async fn acquire_connection_for_download(&self, input_name: &Arc, priority: i8) -> Option { - self.acquire_connection_inner(input_name, &DUMMY_ADDR, false, Some(false), priority, ConnectionKind::Normal, None).await + self.acquire_connection_inner(input_name, false, Some(false), &AcquireProviderParams { + addr: &DUMMY_ADDR, priority, kind: ConnectionKind::Normal, session_owner: None, + }).await } // This method is used for redirects to cycle through the provider diff --git a/backend/src/api/model/active_user_manager.rs b/backend/src/api/model/active_user_manager.rs index 7a11f681a..848af4a0f 100644 --- a/backend/src/api/model/active_user_manager.rs +++ b/backend/src/api/model/active_user_manager.rs @@ -522,6 +522,17 @@ impl SocketRegistration { } } +struct UserSessionParams<'a> { + session_token: &'a str, + virtual_id: u32, + provider: &'a str, + stream_url: &'a str, + addr: &'a SocketAddr, + connection_permission: UserConnectionPermission, + connection_kind: Option, + socket_bound: bool, +} + pub struct ActiveUserManager { grace_period_millis: AtomicU64, grace_period_timeout_secs: AtomicU64, @@ -1483,18 +1494,18 @@ impl ActiveUserManager { let effective_connection_kind = reserved_session_kind.unwrap_or(connection_kind); let country_code = self.lookup_country(&fingerprint.client_ip); - let mut stream_info = StreamInfo::new( + let mut stream_info = StreamInfo::new(shared::model::StreamInfoParams { uid, meter_uid, username, - &fingerprint.addr, - &fingerprint.client_ip, + addr: &fingerprint.addr, + client_ip: &fingerprint.client_ip, provider, - stream_channel.clone(), - user_agent_string, + stream_channel: stream_channel.clone(), + user_agent: user_agent_string, country_code, session_token, - ); + }); if let Some(token) = session_token { if let Some(session) = connection_data.sessions.iter().find(|s| s.token == token) { @@ -1567,31 +1578,21 @@ impl ActiveUserManager { } - #[allow(clippy::too_many_arguments)] - fn new_user_session( - session_token: &str, - virtual_id: u32, - provider: &str, - stream_url: &str, - addr: &SocketAddr, - connection_permission: UserConnectionPermission, - connection_kind: Option, - socket_bound: bool, - ) -> UserSession { + fn new_user_session(params: &UserSessionParams<'_>) -> UserSession { let now = current_time_secs(); UserSession { - token: session_token.to_string(), + token: params.session_token.to_string(), transition_version: 1, - virtual_id, - provider: provider.intern(), - stream_url: stream_url.intern(), - addr: *addr, - socket_bound, - active_addrs: vec![*addr], + virtual_id: params.virtual_id, + provider: params.provider.intern(), + stream_url: params.stream_url.intern(), + addr: *params.addr, + socket_bound: params.socket_bound, + active_addrs: vec![*params.addr], ts: now, started_at: now, - permission: connection_permission, - connection_kind, + permission: params.connection_permission, + connection_kind: params.connection_kind, lifecycle: PlaybackLifecycle::Prepared, } } @@ -1821,7 +1822,7 @@ impl ActiveUserManager { return version; } - let session = Self::new_user_session( + let session = Self::new_user_session(&UserSessionParams { session_token, virtual_id, provider, @@ -1830,7 +1831,7 @@ impl ActiveUserManager { connection_permission, connection_kind, socket_bound, - ); + }); let version = session.transition_version; connection_data.add_session(session); let divergence_snapshot = Self::collect_divergence_snapshot(connection_data, &username); @@ -1939,7 +1940,7 @@ impl ActiveUserManager { debug_if_enabled!("Creating first session for user {username} {}", sanitize_sensitive_info(stream_url)); let mut data = UserConnectionData::new(0, user.max_connections, user.soft_connections); let session = - Self::new_user_session( + Self::new_user_session(&UserSessionParams { session_token, virtual_id, provider, @@ -1948,7 +1949,7 @@ impl ActiveUserManager { connection_permission, connection_kind, socket_bound, - ); + }); data.add_session(session); data }); @@ -2005,7 +2006,7 @@ impl ActiveUserManager { user.username, sanitize_sensitive_info(stream_url) ); - let session = Self::new_user_session( + let session = Self::new_user_session(&UserSessionParams { session_token, virtual_id, provider, @@ -2014,7 +2015,7 @@ impl ActiveUserManager { connection_permission, connection_kind, socket_bound, - ); + }); let token = session.token.clone(); connection_data.add_session(session); let divergence_snapshot = Self::collect_divergence_snapshot(connection_data, &username); @@ -7845,9 +7846,14 @@ mod tests { data.increment_kind(ConnectionKind::Normal); // Add a stream whose session_token doesn't match any counted session - let orphan_stream = StreamInfo::new( - 903, 0, &user.username, &addr, "127.0.0.1", "provider-a".intern(), - StreamChannel { + let orphan_stream = StreamInfo::new(shared::model::StreamInfoParams { + uid: 903, + meter_uid: 0, + username: &user.username, + addr: &addr, + client_ip: "127.0.0.1", + provider: "provider-a".intern(), + stream_channel: StreamChannel { target_id: 1, virtual_id: 9003, provider_id: 1, input_name: "provider-a".intern(), item_type: PlaylistItemType::Live, cluster: XtreamCluster::Live, group: "g".intern(), title: "t".intern(), @@ -7855,8 +7861,10 @@ mod tests { shared: false, shared_joined_existing: None, shared_stream_id: None, technical: None, epg_channel_id: None, }, - "ua".to_string(), None, Some("tok-orphan"), - ); + user_agent: "ua".to_string(), + country_code: None, + session_token: Some("tok-orphan"), + }); data.streams.push(orphan_stream); data.stream_kinds.insert(903, ConnectionKind::Normal); } diff --git a/backend/src/api/model/connection_manager.rs b/backend/src/api/model/connection_manager.rs index 5628d7ae4..a749474e8 100644 --- a/backend/src/api/model/connection_manager.rs +++ b/backend/src/api/model/connection_manager.rs @@ -273,7 +273,7 @@ pub(crate) enum CleanupEvent { } async fn handle_release_connection(deps: &CleanupWorkerDeps, addr: SocketAddr) { - release_connection_with_reason_from_deps(deps, &addr, DisconnectReason::Cleanup, true).await; + release_connection_parts(deps, &addr, DisconnectReason::Cleanup, true).await; } async fn release_connection_with_reason( @@ -282,74 +282,45 @@ async fn release_connection_with_reason( reason: DisconnectReason, send_shared_stop_signal: bool, ) { - release_connection_parts( - &connection_manager.user_manager, - &connection_manager.provider_manager, - &connection_manager.shared_stream_manager, - &connection_manager.event_manager, - &connection_manager.capacity_notify, - &connection_manager.history_writer, - addr, - reason, - send_shared_stop_signal, - ) - .await; + let deps = CleanupWorkerDeps { + user_manager: Arc::clone(&connection_manager.user_manager), + provider_manager: Arc::clone(&connection_manager.provider_manager), + shared_stream_manager: Arc::clone(&connection_manager.shared_stream_manager), + event_manager: Arc::clone(&connection_manager.event_manager), + capacity_notify: Arc::clone(&connection_manager.capacity_notify), + history_writer: Arc::clone(&connection_manager.history_writer), + }; + release_connection_parts(&deps, addr, reason, send_shared_stop_signal).await; } -async fn release_connection_with_reason_from_deps( +async fn release_connection_parts( deps: &CleanupWorkerDeps, addr: &SocketAddr, reason: DisconnectReason, send_shared_stop_signal: bool, -) { - release_connection_parts( - &deps.user_manager, - &deps.provider_manager, - &deps.shared_stream_manager, - &deps.event_manager, - &deps.capacity_notify, - &deps.history_writer, - addr, - reason, - send_shared_stop_signal, - ) - .await; -} - -#[allow(clippy::too_many_arguments)] -async fn release_connection_parts( - user_manager: &Arc, - provider_manager: &Arc, - shared_stream_manager: &Arc, - event_manager: &Arc, - capacity_notify: &Arc, - history_writer: &Arc>, - addr: &SocketAddr, - reason: DisconnectReason, - send_shared_stop_signal: bool, ) { let removed = if matches!(reason, DisconnectReason::ClientKicked) { - user_manager.release_connection_as_kicked(addr).await + deps.user_manager.release_connection_as_kicked(addr).await } else { - user_manager.release_connection(addr).await + deps.user_manager.release_connection(addr).await }; if matches!(reason, DisconnectReason::ClientKicked) { for stream_info in &removed.removed_streams { if let Some(session_token) = stream_info.session_token.as_deref() { - provider_manager.clear_provider_reservation(session_token).await; + deps.provider_manager.clear_provider_reservation(session_token).await; } } // Explicitly terminate all sessions for the kicked addr. This expires them // immediately rather than leaving them for TTL-based GC cleanup. if let Some(ref username) = removed.disconnected_user { - user_manager.terminate_sessions_for_addr(username, addr).await; + deps.user_manager.terminate_sessions_for_addr(username, addr).await; } } for stream_info in &removed.removed_streams { - let (bytes_sent, first_byte_latency_ms) = event_manager.read_meter_qos(stream_info.meter_uid).await; - event_manager.unregister_meter_client(stream_info.uid).await; + let (bytes_sent, first_byte_latency_ms) = deps.event_manager.read_meter_qos(stream_info.meter_uid).await; + deps.event_manager.unregister_meter_client(stream_info.uid).await; emit_disconnect_record( - history_writer, + &deps.history_writer, stream_info, reason, &DisconnectQos { bytes_sent, first_byte_latency_ms, ..Default::default() }, @@ -357,12 +328,12 @@ async fn release_connection_parts( None, ); } - provider_manager.release_connection(addr).await; - shared_stream_manager.release_connection(addr, send_shared_stop_signal).await; + deps.provider_manager.release_connection(addr).await; + deps.shared_stream_manager.release_connection(addr, send_shared_stop_signal).await; if removed.addr_removed && !removed.removed_streams.is_empty() { - event_manager.send_event(EventMessage::ActiveUser(ActiveUserConnectionChange::Disconnected(*addr))); + deps.event_manager.send_event(EventMessage::ActiveUser(ActiveUserConnectionChange::Disconnected(*addr))); } - notify_capacity(capacity_notify.as_ref()); + notify_capacity(deps.capacity_notify.as_ref()); } async fn handle_release_stream( @@ -1246,7 +1217,18 @@ mod tests { technical: None, epg_channel_id: None, }; - StreamInfo::new(0, 0, "test", &addr, "127.0.0.1", provider.intern(), channel, String::new(), None, None) + StreamInfo::new(shared::model::StreamInfoParams { + uid: 0, + meter_uid: 0, + username: "test", + addr: &addr, + client_ip: "127.0.0.1", + provider: provider.intern(), + stream_channel: channel, + user_agent: String::new(), + country_code: None, + session_token: None, + }) } fn create_test_app_config() -> AppConfig { diff --git a/backend/src/api/model/streams/provider_stream_factory.rs b/backend/src/api/model/streams/provider_stream_factory.rs index f0142dbcd..34671e288 100644 --- a/backend/src/api/model/streams/provider_stream_factory.rs +++ b/backend/src/api/model/streams/provider_stream_factory.rs @@ -226,18 +226,18 @@ impl ProviderStreamFactoryOptions { let username = self.username.as_deref()?; let client_ip = self.client_ip.as_deref()?; let stream_channel = self.stream_channel.clone()?; - Some(StreamInfo::new( - 0, - 0, + Some(StreamInfo::new(shared::model::StreamInfoParams { + uid: 0, + meter_uid: 0, username, - &self.addr, + addr: &self.addr, client_ip, - provider_name, + provider: provider_name, stream_channel, - self.user_agent.clone().unwrap_or_default(), - None, - None, - )) + user_agent: self.user_agent.clone().unwrap_or_default(), + country_code: None, + session_token: None, + })) } fn get_connect_failure_stage(&self) -> Option { self.connect_failure_stage } diff --git a/backend/src/library/nfo_reader.rs b/backend/src/library/nfo_reader.rs index ee9f788a1..be50623d4 100644 --- a/backend/src/library/nfo_reader.rs +++ b/backend/src/library/nfo_reader.rs @@ -64,116 +64,118 @@ impl NfoReader { /// Parses movie NFO content fn parse_movie_nfo(content: &str) -> Option { - let mut reader = Reader::from_str(content); - reader.config_mut().trim_text(true); - let mut movie = MovieMetadata { source: MetadataSource::KodiNfo, last_updated: chrono::Utc::now().timestamp(), ..MovieMetadata::default() }; - let mut buf = Vec::new(); - let mut current_text = String::new(); - let mut in_actor = false; - let mut current_actor = Actor { - name: String::new(), - role: None, - thumb: None, - }; - - loop { - match reader.read_event_into(&mut buf) { - Ok(Event::Start(e)) => { - let tag_name = String::from_utf8_lossy(e.name().as_ref()).to_string(); - if tag_name == "actor" { - in_actor = true; - current_actor = Actor { - name: String::new(), - role: None, - thumb: None, - }; - } - current_text.clear(); + let parsed = Self::read_nfo_events(content, |tag, text, in_actor, current_actor| { + match tag { + "title" if !in_actor => movie.title.clone_from(text), + "originaltitle" => movie.original_title = Some(text.clone()), + "year" => movie.year = text.parse().ok(), + "plot" => movie.plot = Some(text.clone()), + "tagline" => movie.tagline = Some(text.clone()), + "runtime" => { + let runtime_str = text.split_whitespace().next().unwrap_or(""); + movie.runtime = runtime_str.parse().ok(); } - Ok(Event::Text(e)) => { - if let Ok(decoded) = e.decode() { - current_text.push_str(decoded.trim()); + "mpaa" => movie.mpaa = Some(text.clone()), + "id" | "imdb" | "imdbid" => movie.imdb_id = Some(text.clone()), + "tmdbid" => movie.tmdb_id = text.parse().ok(), + "rating" => movie.rating = text.parse().ok(), + "genre" => push_to_field_list!(movie.genres, text), + "director" => push_to_field_list!(movie.directors, text), + "credits" | "writer" => push_to_field_list!(movie.writers, text), + "studio" => push_to_field_list!(movie.studios, text), + "thumb" | "poster" => { + if in_actor { + current_actor.thumb = Some(text.clone()); } else { - current_text.clear(); + movie.poster = Some(text.clone()); } } - Ok(Event::End(e)) => { - let tag_name = String::from_utf8_lossy(e.name().as_ref()).to_string(); - match tag_name.as_str() { - "title" if !in_actor => movie.title.clone_from(¤t_text), - "originaltitle" => movie.original_title = Some(current_text.clone()), - "year" => movie.year = current_text.parse().ok(), - "plot" => movie.plot = Some(current_text.clone()), - "tagline" => movie.tagline = Some(current_text.clone()), - "runtime" => { - // Runtime might be in format "136" or "136 min" - let runtime_str = current_text.split_whitespace().next().unwrap_or(""); - movie.runtime = runtime_str.parse().ok(); - } - "mpaa" => movie.mpaa = Some(current_text.clone()), - "id" | "imdb" | "imdbid" => movie.imdb_id = Some(current_text.clone()), - "tmdbid" => movie.tmdb_id = current_text.parse().ok(), - "rating" => movie.rating = current_text.parse().ok(), - "genre" => push_to_field_list!(movie.genres, current_text), - "director" => push_to_field_list!(movie.directors, current_text), - "credits" | "writer" => push_to_field_list!(movie.writers, current_text), - "studio" => push_to_field_list!(movie.studios, current_text), - "thumb" | "poster" => movie.poster = Some(current_text.clone()), - "fanart" => movie.fanart = Some(current_text.clone()), - "name" if in_actor => current_actor.name.clone_from(¤t_text), - "role" if in_actor => current_actor.role = Some(current_text.clone()), - "actor" => { - if !current_actor.name.is_empty() { - push_to_field_list!(movie.actors, current_actor); - } - in_actor = false; - } - _ => {} + "fanart" => movie.fanart = Some(text.clone()), + "name" if in_actor => current_actor.name.clone_from(text), + "role" if in_actor => current_actor.role = Some(text.clone()), + "actor" if !current_actor.name.is_empty() => { + push_to_field_list!(movie.actors, current_actor); } - current_text.clear(); - } - Ok(Event::Eof) => break, - Err(e) => { - error!("Error parsing movie NFO: {e}"); - return None; - } _ => {} } - buf.clear(); - } + }); - if movie.title.is_empty() { - None - } else { + if parsed && !movie.title.is_empty() { Some(MediaMetadata::Movie(movie)) + } else { + None } } /// Parses TV series NFO content fn parse_series_nfo(content: &str) -> Option { - let mut reader = Reader::from_str(content); - reader.config_mut().trim_text(true); - let mut series = SeriesMetadata { source: MetadataSource::KodiNfo, last_updated: chrono::Utc::now().timestamp(), ..SeriesMetadata::default() }; + let parsed = Self::read_nfo_events(content, |tag, text, in_actor, current_actor| { + match tag { + "title" if !in_actor => series.title.clone_from(text), + "originaltitle" => series.original_title = Some(text.clone()), + "year" | "premiered" => { + if let Some(year_str) = text.split('-').next() { + series.year = year_str.parse().ok(); + } + } + "plot" => series.plot = Some(text.clone()), + "mpaa" => series.mpaa = Some(text.clone()), + "id" | "imdb" | "imdbid" => series.imdb_id = Some(text.clone()), + "tmdbid" => series.tmdb_id = text.parse().ok(), + "tvdbid" => series.tvdb_id = text.parse().ok(), + "rating" => series.rating = text.parse().ok(), + "genre" => push_to_field_list!(series.genres, text), + "studio" => push_to_field_list!(series.studios, text), + "thumb" | "poster" => { + if in_actor { + current_actor.thumb = Some(text.clone()); + } else { + series.poster = Some(text.clone()); + } + } + "fanart" => series.fanart = Some(text.clone()), + "status" => series.status = Some(text.clone()), + "name" if in_actor => current_actor.name.clone_from(text), + "role" if in_actor => current_actor.role = Some(text.clone()), + "actor" if !current_actor.name.is_empty() => { + push_to_field_list!(series.actors, current_actor); + } + _ => {} + } + }); + + if parsed && !series.title.is_empty() { + Some(MediaMetadata::Series(series)) + } else { + None + } + } + + /// Generic XML event loop for NFO files. + /// Calls `on_end_tag(tag_name, current_text, in_actor, current_actor)` for each closing tag. + /// Returns `true` if parsing completed without fatal errors. + fn read_nfo_events(content: &str, mut on_end_tag: F) -> bool + where + F: FnMut(&str, &String, bool, &mut Actor), + { + let mut reader = Reader::from_str(content); + reader.config_mut().trim_text(true); let mut buf = Vec::new(); let mut current_text = String::new(); let mut in_actor = false; - let mut current_actor = Actor { - name: String::new(), - role: None, - thumb: None, - }; + let mut current_actor = Actor { name: String::new(), role: None, thumb: None }; loop { match reader.read_event_into(&mut buf) { @@ -181,11 +183,7 @@ impl NfoReader { let tag_name = String::from_utf8_lossy(e.name().as_ref()).to_string(); if tag_name == "actor" { in_actor = true; - current_actor = Actor { - name: String::new(), - role: None, - thumb: None, - }; + current_actor = Actor { name: String::new(), role: None, thumb: None }; } current_text.clear(); } @@ -196,55 +194,33 @@ impl NfoReader { current_text.clear(); } } + Ok(Event::CData(e)) => { + if let Ok(decoded) = e.decode() { + current_text.push_str(decoded.trim()); + } else { + current_text.clear(); + } + } Ok(Event::End(e)) => { let tag_name = String::from_utf8_lossy(e.name().as_ref()).to_string(); - match tag_name.as_str() { - "title" if !in_actor => series.title.clone_from(¤t_text), - "originaltitle" => series.original_title = Some(current_text.clone()), - "year" | "premiered" => { - // Extract year from date like "2008-01-20" - if let Some(year_str) = current_text.split('-').next() { - series.year = year_str.parse().ok(); - } - } - "plot" => series.plot = Some(current_text.clone()), - "mpaa" => series.mpaa = Some(current_text.clone()), - "id" | "imdb" | "imdbid" => series.imdb_id = Some(current_text.clone()), - "tmdbid" => series.tmdb_id = current_text.parse().ok(), - "tvdbid" => series.tvdb_id = current_text.parse().ok(), - "rating" => series.rating = current_text.parse().ok(), - "genre" => push_to_field_list!(series.genres, current_text), - "studio" => push_to_field_list!(series.studios, current_text), - "thumb" | "poster" => series.poster = Some(current_text.clone()), - "fanart" => series.fanart = Some(current_text.clone()), - "status" => series.status = Some(current_text.clone()), - "name" if in_actor => current_actor.name.clone_from(¤t_text), - "role" if in_actor => current_actor.role = Some(current_text.clone()), - "actor" => { - if !current_actor.name.is_empty() { - push_to_field_list!(series.actors, current_actor); - } - in_actor = false; - } - _ => {} + let tag = tag_name.as_str(); + let was_actor = tag == "actor"; + on_end_tag(tag, ¤t_text, in_actor, &mut current_actor); + if was_actor { + in_actor = false; } current_text.clear(); } Ok(Event::Eof) => break, Err(e) => { - error!("Error parsing series NFO: {e}"); - return None; + error!("Error parsing NFO: {e}"); + return false; } _ => {} } buf.clear(); } - - if series.title.is_empty() { - None - } else { - Some(MediaMetadata::Series(series)) - } + true } } @@ -256,7 +232,7 @@ mod tests { async fn test_parse_movie_nfo() { let nfo_content = r#" - The Matrix + <![CDATA[The Matrix]]> The Matrix 1999 A computer hacker learns about the true nature of reality. @@ -271,6 +247,12 @@ mod tests { Lana Wachowski Lilly Wachowski Warner Bros. + movie-poster.jpg + + Keanu Reeves + Neo + neo.jpg + "#; let metadata = NfoReader::parse_movie_nfo(nfo_content); @@ -283,6 +265,8 @@ mod tests { assert_eq!(movie.tmdb_id, Some(603)); assert_eq!(movie.genres.as_ref().map(Vec::len).unwrap_or_default(), 2); assert_eq!(movie.directors.as_ref().map(Vec::len).unwrap_or_default(), 2); + assert_eq!(movie.poster.as_deref(), Some("movie-poster.jpg")); + assert_eq!(movie.actors.as_ref().and_then(|actors| actors.first()).and_then(|actor| actor.thumb.as_deref()), Some("neo.jpg")); } else { panic!("Expected movie metadata"); } @@ -305,6 +289,12 @@ mod tests { Thriller AMC Ended + series-poster.jpg + + Bryan Cranston + Walter White + walter.jpg + "#; let metadata = NfoReader::parse_series_nfo(nfo_content); @@ -318,6 +308,11 @@ mod tests { assert_eq!(series.tvdb_id, Some(81189)); assert_eq!(series.genres.as_ref().map(Vec::len).unwrap_or_default(), 3); assert_eq!(series.status, Some("Ended".to_string())); + assert_eq!(series.poster.as_deref(), Some("series-poster.jpg")); + assert_eq!( + series.actors.as_ref().and_then(|actors| actors.first()).and_then(|actor| actor.thumb.as_deref()), + Some("walter.jpg") + ); } else { panic!("Expected series metadata"); } diff --git a/backend/src/library/processor.rs b/backend/src/library/processor.rs index 7b36dd6e2..da300ce18 100644 --- a/backend/src/library/processor.rs +++ b/backend/src/library/processor.rs @@ -747,11 +747,13 @@ impl LibraryProcessor { match FfmpegExecutor::new() .probe_url( - file_path, - None, - ffprobe.analyze_duration_micros, - ffprobe.probe_size_bytes, - ffprobe.timeout.unwrap_or(60), + &crate::utils::ffmpeg::ProbeParams { + url: file_path, + user_agent: None, + analyze_duration: ffprobe.analyze_duration_micros, + probe_size: ffprobe.probe_size_bytes, + timeout_secs: ffprobe.timeout.unwrap_or(60), + }, // Local file probing does not traverse the network, so no proxy config is applied. None, ) diff --git a/backend/src/processing/processor/sort.rs b/backend/src/processing/processor/sort.rs index fb075d956..fbdb2efba 100644 --- a/backend/src/processing/processor/sort.rs +++ b/backend/src/processing/processor/sort.rs @@ -245,22 +245,11 @@ fn playlist_comparator( } } -macro_rules! sort_groups_by_source_order { - ($groups: ident) => { - $groups.sort_by(|a, b| { - let order1 = a - .channels - .first() - .as_ref() - .map_or(u32::MAX, |c| normalized_source_ordinal(c.header.source_ordinal)); - let order2 = b - .channels - .first() - .as_ref() - .map_or(u32::MAX, |c| normalized_source_ordinal(c.header.source_ordinal)); - order1.cmp(&order2) - }); - }; +fn get_group_source_ordinal(group: &PlaylistGroup) -> u32 { + group + .channels + .first() + .map_or(u32::MAX, |c| normalized_source_ordinal(c.header.source_ordinal)) } fn compare_cached_rule_entries( @@ -305,7 +294,7 @@ pub(in crate::processing::processor) fn sort_playlist( .channels .sort_by_key(|a| normalized_source_ordinal(a.header.source_ordinal)); } - sort_groups_by_source_order!(playlist); + playlist.sort_by_key(get_group_source_ordinal); return true; }; @@ -326,7 +315,7 @@ fn sort_groups(groups: &mut Vec, rules: &[ConfigSortRule], match_ .collect(); if group_rules.is_empty() { - sort_groups_by_source_order!(groups); + groups.sort_by_key(get_group_source_ordinal); return; } @@ -340,15 +329,7 @@ fn sort_groups(groups: &mut Vec, rules: &[ConfigSortRule], match_ return ord; } - let order1 = groups[*left_idx] - .channels - .first() - .map_or(u32::MAX, |c| normalized_source_ordinal(c.header.source_ordinal)); - let order2 = groups[*right_idx] - .channels - .first() - .map_or(u32::MAX, |c| normalized_source_ordinal(c.header.source_ordinal)); - order1.cmp(&order2) + get_group_source_ordinal(&groups[*left_idx]).cmp(&get_group_source_ordinal(&groups[*right_idx])) }); reorder_by_indices(groups, &group_indices); diff --git a/backend/src/processing/processor/stream_probe.rs b/backend/src/processing/processor/stream_probe.rs index c7108f209..84c62bb91 100644 --- a/backend/src/processing/processor/stream_probe.rs +++ b/backend/src/processing/processor/stream_probe.rs @@ -248,41 +248,24 @@ async fn prepare_generic_stream_metadata( acquired_handle.as_ref().and_then(ProbeHandleGuard::handle), active_handle, ); + let params = crate::utils::ffmpeg::ProbeParams { + url: &probe_url, + user_agent: user_agent.as_deref(), + analyze_duration, + probe_size, + timeout_secs: ffprobe_timeout, + }; let probe_data = if uses_seekable_remote_probe(item_type, is_remote_probe) { FfmpegExecutor::new() - .probe_remote_seekable_url_with_cancel( - client, - &probe_url, - user_agent.as_deref(), - analyze_duration, - probe_size, - ffprobe_timeout, - cancel_token, - ) + .probe_remote_seekable_url_with_cancel(client, ¶ms, cancel_token) .await } else if is_remote_probe { FfmpegExecutor::new() - .probe_remote_url_with_cancel( - client, - &probe_url, - user_agent.as_deref(), - analyze_duration, - probe_size, - ffprobe_timeout, - cancel_token, - ) + .probe_remote_url_with_cancel(client, ¶ms, cancel_token) .await } else { FfmpegExecutor::new() - .probe_url_with_cancel( - &probe_url, - user_agent.as_deref(), - analyze_duration, - probe_size, - ffprobe_timeout, - config.proxy.as_ref(), - cancel_token, - ) + .probe_url_with_cancel(¶ms, config.proxy.as_ref(), cancel_token) .await }; diff --git a/backend/src/processing/processor/xtream.rs b/backend/src/processing/processor/xtream.rs index 377805026..33967fd4c 100644 --- a/backend/src/processing/processor/xtream.rs +++ b/backend/src/processing/processor/xtream.rs @@ -146,25 +146,18 @@ pub async fn update_live_stream_metadata( let is_remote_probe = reqwest::Url::parse(probe_url_cow.as_ref()) .ok() .is_some_and(|u| matches!(u.scheme(), "http" | "https")); + let probe_params = crate::utils::ffmpeg::ProbeParams { + url: probe_url_cow.as_ref(), + user_agent: user_agent.as_deref(), + analyze_duration, + probe_size, + timeout_secs: ffprobe_timeout, + }; let probe_result = if is_remote_probe { - FfmpegExecutor::new().probe_remote_url( - client, - probe_url_cow.as_ref(), - user_agent.as_deref(), - analyze_duration, - probe_size, - ffprobe_timeout, - ) + FfmpegExecutor::new().probe_remote_url(client, &probe_params) .await } else { - FfmpegExecutor::new().probe_url( - probe_url_cow.as_ref(), - user_agent.as_deref(), - analyze_duration, - probe_size, - ffprobe_timeout, - config.proxy.as_ref(), - ) + FfmpegExecutor::new().probe_url(&probe_params, config.proxy.as_ref()) .await }; match probe_result { diff --git a/backend/src/processing/processor/xtream_series.rs b/backend/src/processing/processor/xtream_series.rs index f02e65b22..ebfd19f60 100644 --- a/backend/src/processing/processor/xtream_series.rs +++ b/backend/src/processing/processor/xtream_series.rs @@ -999,24 +999,23 @@ pub async fn update_series_metadata( let is_remote_probe = reqwest::Url::parse(probe_url.as_ref()) .ok() .is_some_and(|u| matches!(u.scheme(), "http" | "https")); + let probe_params = crate::utils::ffmpeg::ProbeParams { + url: probe_url.as_ref(), + user_agent: user_agent.as_deref(), + analyze_duration: probe_settings.analyze_duration_micros, + probe_size: probe_settings.probe_size_bytes, + timeout_secs: probe_settings.timeout_secs, + }; let probe_result = if is_remote_probe { FfmpegExecutor::new().probe_remote_seekable_url_with_cancel( client, - probe_url.as_ref(), - user_agent.as_deref(), - probe_settings.analyze_duration_micros, - probe_settings.probe_size_bytes, - probe_settings.timeout_secs, + &probe_params, cancel_token, ) .await } else { FfmpegExecutor::new().probe_url_with_cancel( - probe_url.as_ref(), - user_agent.as_deref(), - probe_settings.analyze_duration_micros, - probe_settings.probe_size_bytes, - probe_settings.timeout_secs, + &probe_params, config.proxy.as_ref(), cancel_token, ) diff --git a/backend/src/processing/processor/xtream_vod.rs b/backend/src/processing/processor/xtream_vod.rs index d5645d98c..f12c9628e 100644 --- a/backend/src/processing/processor/xtream_vod.rs +++ b/backend/src/processing/processor/xtream_vod.rs @@ -862,24 +862,23 @@ pub async fn update_vod_metadata( let is_remote_probe = reqwest::Url::parse(probe_url.as_ref()) .ok() .is_some_and(|u| matches!(u.scheme(), "http" | "https")); + let probe_params = crate::utils::ffmpeg::ProbeParams { + url: probe_url.as_ref(), + user_agent: user_agent.as_deref(), + analyze_duration, + probe_size, + timeout_secs: ffprobe_timeout, + }; let probe_result = if is_remote_probe { FfmpegExecutor::new().probe_remote_seekable_url_with_cancel( client, - probe_url.as_ref(), - user_agent.as_deref(), - analyze_duration, - probe_size, - ffprobe_timeout, + &probe_params, cancel_token, ) .await } else { FfmpegExecutor::new().probe_url_with_cancel( - probe_url.as_ref(), - user_agent.as_deref(), - analyze_duration, - probe_size, - ffprobe_timeout, + &probe_params, config.proxy.as_ref(), cancel_token, ) diff --git a/backend/src/repository/m3u_playlist_iterator.rs b/backend/src/repository/m3u_playlist_iterator.rs index 321a9bdcc..fde3781c7 100644 --- a/backend/src/repository/m3u_playlist_iterator.rs +++ b/backend/src/repository/m3u_playlist_iterator.rs @@ -26,11 +26,14 @@ pub struct M3uPlaylistIterator { inner: LockedReceiverStream<(M3uPlaylistItem, bool)>, } -#[allow(clippy::too_many_arguments)] +struct UrlRewriteContext<'a> { + base_url: &'a str, + username: &'a str, + password: &'a str, +} + fn build_rewritten_url( - base_url: &str, - username: &str, - password: &str, + ctx: &UrlRewriteContext<'_>, source_url: &str, m3u_pli: &M3uPlaylistItem, typed: bool, @@ -55,7 +58,7 @@ fn build_rewritten_url( "" }; - let mut cap = base_url.len() + prefix_path.len() + username.len() + password.len() + 32; // separators and id + let mut cap = ctx.base_url.len() + prefix_path.len() + ctx.username.len() + ctx.password.len() + 32; // separators and id if typed { cap += stream_type.len() + 1; } @@ -63,14 +66,14 @@ fn build_rewritten_url( let rewritten_url = if typed { shared::concat_string!( cap = cap; - base_url, "/", prefix_path, "/", stream_type, "/", - username, "/", password, "/", &m3u_pli.virtual_id.to_string() + ctx.base_url, "/", prefix_path, "/", stream_type, "/", + ctx.username, "/", ctx.password, "/", &m3u_pli.virtual_id.to_string() ) } else { shared::concat_string!( cap = cap; - base_url, "/", prefix_path, "/", - username, "/", password, "/", &m3u_pli.virtual_id.to_string() + ctx.base_url, "/", prefix_path, "/", + ctx.username, "/", ctx.password, "/", &m3u_pli.virtual_id.to_string() ) }; @@ -121,9 +124,7 @@ fn resolve_effective_source_url<'a>( #[allow(clippy::too_many_arguments)] fn apply_rewrite( mut m3u_pli: M3uPlaylistItem, - base_url: &str, - username: &str, - password: &str, + ctx: &UrlRewriteContext<'_>, target_id: u16, encrypt_secret: &[u8; 16], input_by_name: &HashMap, Arc>, @@ -142,8 +143,8 @@ fn apply_rewrite( if let Some(catchup) = live.catchup.as_ref() { build_m3u_catchup_rewrite( encrypt_secret, - base_url, - username, + ctx.base_url, + ctx.username, target_id, m3u_pli.virtual_id, effective_source_url.as_ref(), @@ -163,9 +164,7 @@ fn apply_rewrite( if should_rewrite_urls { let stream_url = build_rewritten_url( - base_url, - username, - password, + ctx, effective_source_url.as_ref(), &m3u_pli, flags.contains(M3uPlaylistIteratorFlags::IncludeTypeInUrl), @@ -175,9 +174,7 @@ fn apply_rewrite( let resource_url = if flags.contains(M3uPlaylistIteratorFlags::RewriteResource) { let source_url = if m3u_pli.logo.is_empty() { m3u_pli.logo_small.as_ref() } else { m3u_pli.logo.as_ref() }; Some(build_rewritten_url( - base_url, - username, - password, + ctx, source_url, &m3u_pli, false, @@ -293,11 +290,15 @@ impl M3uPlaylistIterator { } } + let rewrite_ctx = UrlRewriteContext { + base_url: &base_url, + username: &username, + password: &password, + }; + let item = apply_rewrite( item, - &base_url, - &username, - &password, + &rewrite_ctx, target_id, &encrypt_secret, &input_by_name, @@ -383,7 +384,7 @@ impl Stream for M3uPlaylistM3uTextIterator { #[cfg(test)] mod tests { - use super::{apply_rewrite, M3uPlaylistIteratorFlags, M3uPlaylistIteratorFlagsSet}; + use super::{apply_rewrite, M3uPlaylistIteratorFlags, M3uPlaylistIteratorFlagsSet, UrlRewriteContext}; use crate::model::{ConfigInput, ConfigProvider, ProviderDnsCache}; use shared::{ model::{M3uPlaylistItem, PlaylistItemType, ProviderUrlSelectionPolicy, ProxyType}, @@ -440,11 +441,14 @@ mod tests { fn redirect_without_mask_resolves_provider_scheme_url() { let input = provider_input(); let input_by_name = HashMap::from([(Arc::clone(&input.name), input)]); + let ctx = UrlRewriteContext { + base_url: "https://example.com", + username: "user", + password: "pass", + }; let rewritten = apply_rewrite( m3u_item("provider://example/live/user/pass/813294.ts"), - "https://example.com", - "user", - "pass", + &ctx, 1, &[7u8; 16], &input_by_name, @@ -459,11 +463,14 @@ mod tests { #[test] fn redirect_without_mask_keeps_regular_url() { let input_by_name = HashMap::new(); + let ctx = UrlRewriteContext { + base_url: "https://example.com", + username: "user", + password: "pass", + }; let rewritten = apply_rewrite( m3u_item("http://example.com/live/user/pass/813294.ts"), - "https://example.com", - "user", - "pass", + &ctx, 1, &[7u8; 16], &input_by_name, @@ -478,11 +485,14 @@ mod tests { #[test] fn redirect_without_mask_preserves_provider_scheme_url_when_input_is_missing() { let input_by_name = HashMap::new(); + let ctx = UrlRewriteContext { + base_url: "https://example.com", + username: "user", + password: "pass", + }; let rewritten = apply_rewrite( m3u_item("provider://example/live/user/pass/813294.ts"), - "https://example.com", - "user", - "pass", + &ctx, 1, &[7u8; 16], &input_by_name, @@ -500,12 +510,14 @@ mod tests { let input_by_name = HashMap::from([(Arc::clone(&input.name), input)]); let mut flags = M3uPlaylistIteratorFlagsSet::new(); flags.set(M3uPlaylistIteratorFlags::MaskRedirectUrl); - + let ctx = UrlRewriteContext { + base_url: "https://example.com", + username: "user", + password: "pass", + }; let rewritten = apply_rewrite( m3u_item("provider://example/live/user/pass/813294.ts"), - "https://example.com", - "user", - "pass", + &ctx, 1, &[7u8; 16], &input_by_name, diff --git a/backend/src/repository/qos_snapshot_repository.rs b/backend/src/repository/qos_snapshot_repository.rs index 65d41081d..79f83e46e 100644 --- a/backend/src/repository/qos_snapshot_repository.rs +++ b/backend/src/repository/qos_snapshot_repository.rs @@ -365,13 +365,9 @@ mod tests { use shared::utils::Internable; use super::{QosAggregationCheckpoint, QosSnapshotRecord, QosSnapshotRepository, QosSnapshotWindow}; - #[test] - fn qos_snapshot_repository_roundtrips_snapshot_and_checkpoint() { - let temp = tempdir().expect("tempdir should succeed"); - let repo = QosSnapshotRepository::open(temp.path()).expect("repo should open"); - - let snapshot = QosSnapshotRecord { - stream_identity_key: "stream-a".to_string(), + fn make_test_snapshot(identity_key: &str, score: u8) -> QosSnapshotRecord { + QosSnapshotRecord { + stream_identity_key: identity_key.to_string(), input_name: "input-a".intern(), target_name: "target-a".intern(), provider_name: "provider-a".intern(), @@ -382,14 +378,22 @@ mod tests { last_event_at: 1_700_000_123, window_24h: QosSnapshotWindow { connect_count: 3, - score: 81, + score, confidence: 60, ..QosSnapshotWindow::default() }, window_7d: QosSnapshotWindow::default(), window_30d: QosSnapshotWindow::default(), daily_buckets: BTreeMap::default(), - }; + } + } + + #[test] + fn qos_snapshot_repository_roundtrips_snapshot_and_checkpoint() { + let temp = tempdir().expect("tempdir should succeed"); + let repo = QosSnapshotRepository::open(temp.path()).expect("repo should open"); + + let snapshot = make_test_snapshot("stream-a", 81); repo.put_snapshot(&snapshot).expect("put snapshot should succeed"); let loaded = repo @@ -416,21 +420,7 @@ mod tests { let temp = tempdir().expect("tempdir should succeed"); let repo = QosSnapshotRepository::open(temp.path()).expect("repo should open"); - let snapshot = QosSnapshotRecord { - stream_identity_key: "stream-a".to_string(), - input_name: "input-a".intern(), - target_name: "target-a".intern(), - provider_name: "provider-a".intern(), - provider_id: 22, - virtual_id: 33, - item_type: PlaylistItemType::Live, - updated_at: 1_700_000_000, - last_event_at: 1_700_000_123, - window_24h: QosSnapshotWindow::default(), - window_7d: QosSnapshotWindow::default(), - window_30d: QosSnapshotWindow::default(), - daily_buckets: BTreeMap::default(), - }; + let snapshot = make_test_snapshot("stream-a", 0); repo.put_snapshot(&snapshot).expect("put snapshot should succeed"); assert!(repo.delete_snapshot("stream-a").expect("delete should succeed")); @@ -453,26 +443,7 @@ mod tests { let temp = tempdir().expect("tempdir should succeed"); let repo = QosSnapshotRepository::open(temp.path()).expect("repo should open"); - let snapshot = QosSnapshotRecord { - stream_identity_key: "stream-a".to_string(), - input_name: "input-a".intern(), - target_name: "target-a".intern(), - provider_name: "provider-a".intern(), - provider_id: 22, - virtual_id: 33, - item_type: PlaylistItemType::Live, - updated_at: 1_700_000_000, - last_event_at: 1_700_000_123, - window_24h: QosSnapshotWindow { - connect_count: 3, - score: 81, - confidence: 60, - ..QosSnapshotWindow::default() - }, - window_7d: QosSnapshotWindow::default(), - window_30d: QosSnapshotWindow::default(), - daily_buckets: BTreeMap::default(), - }; + let snapshot = make_test_snapshot("stream-a", 81); repo.put_snapshot(&snapshot).expect("put snapshot should succeed"); let loaded = QosSnapshotRepository::get_snapshot_read_only(temp.path(), "stream-a") diff --git a/backend/src/repository/stream_history/archive.rs b/backend/src/repository/stream_history/archive.rs index 8340c8bae..dd0f0a4f7 100644 --- a/backend/src/repository/stream_history/archive.rs +++ b/backend/src/repository/stream_history/archive.rs @@ -222,7 +222,7 @@ mod tests { use crate::model::{StreamHistoryConfig, StreamHistoryRecord}; use lz4_flex::frame::FrameDecoder; use tempfile::TempDir; - use shared::model::{PlaylistItemType, StreamHistoryEventType}; + use shared::model::PlaylistItemType; use shared::utils::Internable; fn test_config(dir: &str, batch_size: usize) -> StreamHistoryConfig { @@ -235,50 +235,17 @@ mod tests { } fn make_record(session_id: u64) -> StreamHistoryRecord { - StreamHistoryRecord { - schema_version: RECORD_SCHEMA_VERSION, - event_type: StreamHistoryEventType::Connect, - event_ts_utc: now_utc_secs(), - partition_day_utc: current_utc_day(), - session_id, - source_addr: None, - api_username: Some("alice".to_string()), - provider_name: Some("acme".intern()), - provider_username: None, - input_name: Some("input".intern()), - virtual_id: Some(1), - item_type: Some(PlaylistItemType::Live), - title: Some("Test".to_string()), - group: None, - country: None, - user_agent: None, - shared: None, - shared_joined_existing: None, - shared_stream_id: None, - provider_id: None, - cluster: None, - container: None, - stream_url_hash: None, - stream_identity_key: None, - video_codec: None, - audio_codec: None, - audio_channels: None, - resolution: None, - fps: None, - connect_ts_utc: None, - disconnect_ts_utc: None, - session_duration: None, - bytes_sent: None, - first_byte_latency_ms: None, - provider_reconnect_count: None, - failure_stage: None, - provider_http_status: None, - provider_error_class: None, - connect_failure_reason: None, - disconnect_reason: None, - previous_session_id: None, - target_name: None, - } + let mut r = crate::repository::stream_history::tests::make_base_test_record(); + r.event_ts_utc = now_utc_secs(); + r.partition_day_utc = current_utc_day(); + r.session_id = session_id; + r.api_username = Some("alice".to_string()); + r.provider_name = Some("acme".intern()); + r.input_name = Some("input".intern()); + r.virtual_id = Some(1); + r.item_type = Some(PlaylistItemType::Live); + r.title = Some("Test".to_string()); + r } /// Write a minimal `.pending` file and return its path. diff --git a/backend/src/repository/stream_history/mod.rs b/backend/src/repository/stream_history/mod.rs index 8523bd007..895b055d4 100644 --- a/backend/src/repository/stream_history/mod.rs +++ b/backend/src/repository/stream_history/mod.rs @@ -10,3 +10,140 @@ pub use storage::*; pub use reader::*; pub use async_iterator::*; +#[cfg(test)] +pub(crate) mod tests { + use crate::model::{StreamHistoryRecord, RECORD_SCHEMA_VERSION}; + use shared::model::StreamHistoryEventType; + + pub fn make_base_test_record() -> StreamHistoryRecord { + StreamHistoryRecord { + schema_version: RECORD_SCHEMA_VERSION, + event_type: StreamHistoryEventType::Connect, + event_ts_utc: 1_742_600_001, + partition_day_utc: "2025-03-21".to_string(), + session_id: 1, + source_addr: None, + api_username: None, + provider_name: None, + provider_username: None, + input_name: None, + virtual_id: None, + item_type: None, + title: None, + group: None, + country: None, + user_agent: None, + shared: None, + shared_joined_existing: None, + shared_stream_id: None, + provider_id: None, + cluster: None, + container: None, + stream_url_hash: None, + stream_identity_key: None, + video_codec: None, + audio_codec: None, + audio_channels: None, + resolution: None, + fps: None, + connect_ts_utc: None, + disconnect_ts_utc: None, + session_duration: None, + bytes_sent: None, + first_byte_latency_ms: None, + provider_reconnect_count: None, + failure_stage: None, + provider_http_status: None, + provider_error_class: None, + connect_failure_reason: None, + disconnect_reason: None, + previous_session_id: None, + target_name: None, + } + } + + pub fn sample_connect_record() -> StreamHistoryRecord { + let mut r = make_base_test_record(); + r.event_ts_utc = 1_742_600_001; + r.session_id = 999; + r.source_addr = Some("192.0.2.1:12345".to_string()); + r.api_username = Some("alice".to_string()); + r.provider_name = Some(shared::utils::Internable::intern("acme-tv")); + r.provider_username = Some("acme_user".to_string()); + r.input_name = Some(shared::utils::Internable::intern("provider-input")); + r.virtual_id = Some(1234); + r.item_type = Some(shared::model::PlaylistItemType::Live); + r.title = Some("News Channel".to_string()); + r.group = Some("News".to_string()); + r.country = Some("DE".to_string()); + r.user_agent = Some("VLC/3.0".to_string()); + r.shared = Some(false); + r.provider_id = Some(1); + r.cluster = Some("live".to_string()); + r.container = Some("mpegts".to_string()); + r.stream_url_hash = Some("abc123".to_string()); + r.stream_identity_key = Some("identity123".to_string()); + r.video_codec = Some("H.264".to_string()); + r.audio_codec = Some("AAC".to_string()); + r.audio_channels = Some("STEREO".to_string()); + r.resolution = Some("1920x1080".to_string()); + r.fps = Some("50".to_string()); + r.connect_ts_utc = Some(1_742_600_001); + r + } + + pub fn sample_disconnect_record(session_id: u64) -> StreamHistoryRecord { + let mut r = sample_connect_record(); + r.event_type = StreamHistoryEventType::Disconnect; + r.event_ts_utc = 1_742_603_601; + r.partition_day_utc = "2025-03-22".to_string(); + r.session_id = session_id; + r.disconnect_ts_utc = Some(1_742_603_601); + r.session_duration = Some(3600); + r.bytes_sent = Some(1_234_567_890); + r.first_byte_latency_ms = Some(150); + r.provider_reconnect_count = Some(0); + r.disconnect_reason = Some(shared::model::DisconnectReason::ClientClosed); + r + } + + pub fn sample_stream_info() -> shared::model::StreamInfo { + let addr: std::net::SocketAddr = "192.0.2.1:12345".parse().unwrap(); + let mut info = shared::model::StreamInfo::new(shared::model::StreamInfoParams { + uid: 999, + meter_uid: 1001, + username: "alice", + addr: &addr, + client_ip: "192.0.2.1", + provider: shared::utils::Internable::intern("acme-tv"), + stream_channel: shared::model::StreamChannel { + target_id: 1, + virtual_id: 1234, + provider_id: 1, + input_name: shared::utils::Internable::intern("provider-input"), + item_type: shared::model::PlaylistItemType::Live, + cluster: shared::model::XtreamCluster::Live, + group: shared::utils::Internable::intern("News"), + title: shared::utils::Internable::intern("News Channel"), + url: shared::utils::Internable::intern("http://localhost/stream.ts"), + shared: false, + shared_joined_existing: None, + shared_stream_id: None, + technical: Some(shared::model::StreamTechnicalInfo { + container: "mpegts".to_string(), + resolution: "1920x1080".to_string(), + fps: "50".to_string(), + video_codec: "H.264".to_string(), + audio_codec: "AAC".to_string(), + audio_channels: "STEREO".to_string(), + }), + epg_channel_id: None, + }, + user_agent: String::from("VLC/3.0"), + country_code: Some(String::from("DE")), + session_token: None, + }); + info.ts = 1_742_600_001; + info + } +} diff --git a/backend/src/repository/stream_history/reader.rs b/backend/src/repository/stream_history/reader.rs index 5625fdbea..42251c7b7 100644 --- a/backend/src/repository/stream_history/reader.rs +++ b/backend/src/repository/stream_history/reader.rs @@ -23,10 +23,8 @@ pub struct StreamHistoryFileReader { file_path: String, } -impl StreamHistoryFileReader> { - pub fn from_pending(path: &Path, time_range: Option<(u64, u64)>) -> io::Result<(Self, FileHeaderBody)> { - let file = File::open(path)?; - let mut reader = BufReader::new(file); +impl StreamHistoryFileReader { + fn new_with_header(mut reader: R, time_range: Option<(u64, u64)>, file_path: String) -> io::Result<(Self, FileHeaderBody)> { read_and_verify_file_magic(&mut reader)?; let header: FileHeaderBody = read_framed(&mut reader)?; Ok((Self { @@ -36,27 +34,23 @@ impl StreamHistoryFileReader> { payload_buf: Vec::new(), payload_offset: 0, done: false, - file_path: path.display().to_string(), + file_path, }, header)) } } +impl StreamHistoryFileReader> { + pub fn from_pending(path: &Path, time_range: Option<(u64, u64)>) -> io::Result<(Self, FileHeaderBody)> { + let file = File::open(path)?; + Self::new_with_header(BufReader::new(file), time_range, path.display().to_string()) + } +} + impl StreamHistoryFileReader>> { pub fn from_archive(path: &Path, time_range: Option<(u64, u64)>) -> io::Result<(Self, FileHeaderBody)> { let file = File::open(path)?; let decoder = FrameDecoder::new(file); - let mut reader = BufReader::new(decoder); - read_and_verify_file_magic(&mut reader)?; - let header: FileHeaderBody = read_framed(&mut reader)?; - Ok((Self { - reader, - time_range, - current_block_remaining: 0, - payload_buf: Vec::new(), - payload_offset: 0, - done: false, - file_path: path.display().to_string(), - }, header)) + Self::new_with_header(BufReader::new(decoder), time_range, path.display().to_string()) } } @@ -440,66 +434,21 @@ mod tests { use crate::repository::stream_history::storage::*; use std::io::Write; use tempfile::NamedTempFile; - use shared::model::StreamHistoryEventType; + use shared::utils::Internable; use crate::model::{RECORD_SCHEMA_VERSION}; fn make_test_record(ts: u64, username: &str) -> StreamHistoryRecord { - StreamHistoryRecord { - schema_version: RECORD_SCHEMA_VERSION, - event_type: StreamHistoryEventType::Connect, - event_ts_utc: ts, - partition_day_utc: "2026-03-22".to_string(), - session_id: 1, - source_addr: None, - api_username: Some(username.to_string()), - provider_name: None, - provider_username: None, - input_name: Some("input".intern()), - virtual_id: None, - item_type: None, - title: None, - group: None, - country: None, - user_agent: None, - shared: None, - shared_joined_existing: None, - shared_stream_id: None, - provider_id: None, - cluster: None, - container: None, - stream_url_hash: None, - stream_identity_key: None, - video_codec: None, - audio_codec: None, - audio_channels: None, - resolution: None, - fps: None, - connect_ts_utc: None, - disconnect_ts_utc: None, - session_duration: None, - bytes_sent: None, - first_byte_latency_ms: None, - provider_reconnect_count: None, - failure_stage: None, - provider_http_status: None, - provider_error_class: None, - connect_failure_reason: None, - disconnect_reason: None, - previous_session_id: None, - target_name: None, - } + let mut r = crate::repository::stream_history::tests::make_base_test_record(); + r.event_ts_utc = ts; + r.api_username = Some(username.to_string()); + r.input_name = Some("input".intern()); + r } /// 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(); - - // Write file header - write_file_magic(&mut f).unwrap(); - let header = FileHeaderBody { + fn make_test_file_header() -> FileHeaderBody { + FileHeaderBody { container_format_version: CONTAINER_FORMAT_VERSION, record_schema_version: RECORD_SCHEMA_VERSION, source_kind: "stream_history".to_string(), @@ -515,7 +464,17 @@ mod tests { total_record_count: None, min_event_ts_utc: None, max_event_ts_utc: None, - }; + } + } + + #[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(); + + // Write file header + write_file_magic(&mut f).unwrap(); + let header = make_test_file_header(); write_framed(&mut f, &header).unwrap(); if !records.is_empty() { @@ -599,23 +558,7 @@ mod tests { let tmp = NamedTempFile::new().unwrap(); let mut f = tmp.as_file().try_clone().unwrap(); write_file_magic(&mut f).unwrap(); - let header = FileHeaderBody { - container_format_version: CONTAINER_FORMAT_VERSION, - record_schema_version: RECORD_SCHEMA_VERSION, - source_kind: "stream_history".to_string(), - created_at_ts_utc: 0, - partition_day_ts_utc: "2026-03-22".to_string(), - writer_instance_id: 1, - host_id: None, - compression_kind: CompressionKind::None, - finalized: false, - record_encoding_kind: RecordEncodingKind::MessagePackNamed, - finalized_at_ts_utc: None, - total_block_count: None, - total_record_count: None, - min_event_ts_utc: None, - max_event_ts_utc: None, - }; + let header = make_test_file_header(); write_framed(&mut f, &header).unwrap(); // Write block magic but truncate before block header write_block_magic(&mut f).unwrap(); @@ -639,23 +582,7 @@ mod tests { // Write valid header write_file_magic(&mut f).unwrap(); - let header = FileHeaderBody { - container_format_version: CONTAINER_FORMAT_VERSION, - record_schema_version: RECORD_SCHEMA_VERSION, - source_kind: "stream_history".to_string(), - created_at_ts_utc: 0, - partition_day_ts_utc: "2026-03-22".to_string(), - writer_instance_id: 1, - host_id: None, - compression_kind: CompressionKind::None, - finalized: false, - record_encoding_kind: RecordEncodingKind::MessagePackNamed, - finalized_at_ts_utc: None, - total_block_count: None, - total_record_count: None, - min_event_ts_utc: None, - max_event_ts_utc: None, - }; + let header = make_test_file_header(); write_framed(&mut f, &header).unwrap(); // Write a bad block with huge payload_len but only write a small amount diff --git a/backend/src/repository/stream_history/storage.rs b/backend/src/repository/stream_history/storage.rs index f4effb492..bd1aad628 100644 --- a/backend/src/repository/stream_history/storage.rs +++ b/backend/src/repository/stream_history/storage.rs @@ -118,15 +118,20 @@ pub fn read_framed Deserialize<'de>>(reader: &mut R) -> io: reader.read_exact(&mut payload)?; let mut crc_buf = [0u8; 4]; reader.read_exact(&mut crc_buf)?; + verify_crc(crc_buf, &payload)?; + deserialize_named(&payload) +} + +fn verify_crc(crc_buf: [u8; 4], payload: &[u8]) -> io::Result<()> { let expected_crc = u32::from_be_bytes(crc_buf); - let actual_crc = crc32fast::hash(&payload); + let actual_crc = crc32fast::hash(payload); if actual_crc != expected_crc { return Err(io::Error::new( io::ErrorKind::InvalidData, format!("CRC mismatch: expected {expected_crc:#010x}, got {actual_crc:#010x}"), )); } - deserialize_named(&payload) + Ok(()) } /// Write the file magic bytes at the beginning of a `.pending` file. @@ -214,28 +219,16 @@ pub async fn async_read_framed Deserialize<'de reader.read_exact(&mut payload).await?; let mut crc_buf = [0u8; 4]; reader.read_exact(&mut crc_buf).await?; - let expected_crc = u32::from_be_bytes(crc_buf); - let actual_crc = crc32fast::hash(&payload); - if actual_crc != expected_crc { - return Err(io::Error::new( - io::ErrorKind::InvalidData, - format!("CRC mismatch: expected {expected_crc:#010x}, got {actual_crc:#010x}"), - )); - } + verify_crc(crc_buf, &payload)?; deserialize_named(&payload) } #[cfg(test)] mod tests { use super::*; - use shared::{ - model::{PlaylistItemType, StreamChannel, StreamInfo, StreamTechnicalInfo, XtreamCluster}, - utils::Internable, - }; - use std::net::SocketAddr; + use shared::model::{StreamHistoryEventType, DisconnectReason, FailureStage, ConnectFailureReason}; + use crate::model::{StreamHistoryRecord, RECORD_SCHEMA_VERSION, DisconnectQos}; use std::io::Cursor; - use shared::model::{ConnectFailureReason, FailureStage, DisconnectReason, StreamHistoryEventType}; - use crate::model::{DisconnectQos, StreamHistoryRecord, RECORD_SCHEMA_VERSION}; use crate::utils::encode_base64_hash; fn sample_file_header() -> FileHeaderBody { @@ -270,139 +263,7 @@ mod tests { } } - fn sample_connect_record() -> StreamHistoryRecord { - StreamHistoryRecord { - schema_version: RECORD_SCHEMA_VERSION, - event_type: StreamHistoryEventType::Connect, - event_ts_utc: 1_742_600_001, - partition_day_utc: "2026-03-22".to_string(), - session_id: 999, - source_addr: Some("192.0.2.1:12345".to_string()), - api_username: Some("alice".to_string()), - provider_name: Some("acme-tv".intern()), - provider_username: Some("acme_user".to_string()), - input_name: Some("provider-input".intern()), - virtual_id: Some(1234), - item_type: Some(PlaylistItemType::Live), - title: Some("News Channel".to_string()), - group: Some("News".to_string()), - country: Some("DE".to_string()), - user_agent: Some("VLC/3.0".to_string()), - shared: Some(false), - shared_joined_existing: None, - shared_stream_id: None, - provider_id: Some(1), - cluster: Some("live".to_string()), - container: Some("mpegts".to_string()), - stream_url_hash: Some("abc123".to_string()), - stream_identity_key: Some("identity123".to_string()), - video_codec: Some("H.264".to_string()), - audio_codec: Some("AAC".to_string()), - audio_channels: Some("STEREO".to_string()), - resolution: Some("1920x1080".to_string()), - fps: Some("50".to_string()), - connect_ts_utc: Some(1_742_600_001), - disconnect_ts_utc: None, - session_duration: None, - bytes_sent: None, - first_byte_latency_ms: None, - provider_reconnect_count: None, - failure_stage: None, - provider_http_status: None, - provider_error_class: None, - connect_failure_reason: None, - disconnect_reason: None, - previous_session_id: None, - target_name: None, - } - } - - fn sample_disconnect_record(session_id: u64) -> StreamHistoryRecord { - StreamHistoryRecord { - schema_version: RECORD_SCHEMA_VERSION, - event_type: StreamHistoryEventType::Disconnect, - event_ts_utc: 1_742_603_601, - partition_day_utc: "2026-03-22".to_string(), - session_id, - source_addr: Some("192.0.2.1:12345".to_string()), - api_username: Some("alice".to_string()), - provider_name: Some("acme-tv".intern()), - provider_username: Some("acme_user".to_string()), - input_name: Some("provider-input".intern()), - virtual_id: Some(1234), - item_type: Some(PlaylistItemType::Live), - title: Some("News Channel".to_string()), - group: Some("News".to_string()), - country: Some("DE".to_string()), - user_agent: Some("VLC/3.0".to_string()), - shared: Some(false), - shared_joined_existing: None, - shared_stream_id: None, - provider_id: Some(1), - cluster: Some("live".to_string()), - container: Some("mpegts".to_string()), - stream_url_hash: Some("abc123".to_string()), - stream_identity_key: Some("identity123".to_string()), - video_codec: Some("H.264".to_string()), - audio_codec: Some("AAC".to_string()), - audio_channels: Some("STEREO".to_string()), - resolution: Some("1920x1080".to_string()), - fps: Some("50".to_string()), - connect_ts_utc: Some(1_742_600_001), - disconnect_ts_utc: Some(1_742_603_601), - session_duration: Some(3600), - bytes_sent: Some(1_234_567_890), - first_byte_latency_ms: Some(150), - provider_reconnect_count: Some(0), - failure_stage: None, - provider_http_status: None, - provider_error_class: None, - connect_failure_reason: None, - disconnect_reason: Some(DisconnectReason::ClientClosed), - previous_session_id: None, - target_name: None, - } - } - - fn sample_stream_info() -> StreamInfo { - let addr: SocketAddr = "192.0.2.1:12345".parse().unwrap(); - let mut info = StreamInfo::new( - 999, - 1001, - "alice", - &addr, - "192.0.2.1", - "acme-tv".intern(), - StreamChannel { - target_id: 1, - virtual_id: 1234, - provider_id: 1, - input_name: "provider-input".intern(), - item_type: PlaylistItemType::Live, - cluster: XtreamCluster::Live, - group: "News".intern(), - title: "News Channel".intern(), - url: "http://localhost/stream.ts".intern(), - shared: false, - shared_joined_existing: None, - shared_stream_id: None, - technical: Some(StreamTechnicalInfo { - container: "mpegts".to_string(), - resolution: "1920x1080".to_string(), - fps: "50".to_string(), - video_codec: "H.264".to_string(), - audio_codec: "AAC".to_string(), - audio_channels: "STEREO".to_string(), - }), - epg_channel_id: None, - }, - String::from("VLC/3.0"), - Some(String::from("DE")), - None, - ); - info.ts = 1_742_600_001; - info - } + use crate::repository::stream_history::tests::{sample_connect_record, sample_disconnect_record, sample_stream_info}; #[test] fn stream_history_header_round_trip() { diff --git a/backend/src/repository/stream_history/writer.rs b/backend/src/repository/stream_history/writer.rs index 935e48bf0..66bdf36b3 100644 --- a/backend/src/repository/stream_history/writer.rs +++ b/backend/src/repository/stream_history/writer.rs @@ -494,50 +494,18 @@ mod tests { } fn make_record(session_id: u64, event_type: StreamHistoryEventType) -> StreamHistoryRecord { - StreamHistoryRecord { - schema_version: RECORD_SCHEMA_VERSION, - event_type, - event_ts_utc: now_utc_secs(), - partition_day_utc: current_utc_day(), - session_id, - source_addr: None, - api_username: Some("user1".to_string()), - provider_name: Some("acme".intern()), - provider_username: None, - input_name: Some("input".intern()), - virtual_id: Some(1), - item_type: Some(PlaylistItemType::Live), - title: Some("Test Channel".to_string()), - group: None, - country: None, - user_agent: None, - shared: None, - shared_joined_existing: None, - shared_stream_id: None, - provider_id: None, - cluster: None, - container: None, - stream_url_hash: None, - stream_identity_key: None, - video_codec: None, - audio_codec: None, - audio_channels: None, - resolution: None, - fps: None, - connect_ts_utc: None, - disconnect_ts_utc: None, - session_duration: None, - bytes_sent: None, - first_byte_latency_ms: None, - provider_reconnect_count: None, - failure_stage: None, - provider_http_status: None, - provider_error_class: None, - connect_failure_reason: None, - disconnect_reason: None, - previous_session_id: None, - target_name: None, - } + let mut r = crate::repository::stream_history::tests::make_base_test_record(); + r.event_type = event_type; + r.event_ts_utc = now_utc_secs(); + r.partition_day_utc = current_utc_day(); + r.session_id = session_id; + r.api_username = Some("user1".to_string()); + r.provider_name = Some("acme".intern()); + r.input_name = Some("input".intern()); + r.virtual_id = Some(1); + r.item_type = Some(PlaylistItemType::Live); + r.title = Some("Test Channel".to_string()); + r } #[tokio::test] diff --git a/backend/src/repository/user_repository.rs b/backend/src/repository/user_repository.rs index e25564ddd..6a64347f1 100644 --- a/backend/src/repository/user_repository.rs +++ b/backend/src/repository/user_repository.rs @@ -503,98 +503,50 @@ mod tests { use std::sync::Arc; use tempfile::tempdir; + fn make_test_credential(username: &str, status: ProxyUserStatus) -> ProxyUserCredentials { + ProxyUserCredentials { + username: username.to_string(), + password: "Test".to_string(), + token: Some("Test".to_string()), + proxy: ProxyType::Reverse(None), + server: Some("default".to_string()), + epg_timeshift: None, + epg_request_timeshift: None, + created_at: None, + exp_date: Some(1_672_705_545), + max_connections: 1, + status: Some(status), + output_clusters: ClusterFlags::all(), + ui_enabled: true, + comment: None, + priority: 0, + soft_connections: 0, + soft_priority: 0, + t_is_api_user: false, + network_access: None, + } + } + #[tokio::test] #[allow(clippy::too_many_lines)] pub async fn save_target_user() { let user = TargetUser { target: "test".to_string(), credentials: vec![ - Arc::new(ProxyUserCredentials { - username: "Test".to_string(), - password: "Test".to_string(), - token: Some("Test".to_string()), - proxy: ProxyType::Reverse(None), - server: Some("default".to_string()), - epg_timeshift: None, - epg_request_timeshift: None, - created_at: None, - exp_date: Some(1_672_705_545), - max_connections: 1, - status: Some(ProxyUserStatus::Active), - output_clusters: ClusterFlags::all(), - ui_enabled: true, - comment: None, - priority: 0, - soft_connections: 0, - soft_priority: 0, - t_is_api_user: false, - network_access: None, - }), - Arc::new(ProxyUserCredentials { - username: "Test2".to_string(), - password: "Test".to_string(), - token: Some("Test".to_string()), - proxy: ProxyType::Reverse(None), - server: Some("default".to_string()), - epg_timeshift: None, - epg_request_timeshift: None, - created_at: None, - exp_date: Some(1_672_705_545), - max_connections: 1, - status: Some(ProxyUserStatus::Expired), - output_clusters: ClusterFlags::all(), - ui_enabled: true, - comment: None, - priority: 0, - soft_connections: 0, - soft_priority: 0, - t_is_api_user: false, - network_access: None, - }), - Arc::new(ProxyUserCredentials { - username: "Test3".to_string(), - password: "Test".to_string(), - token: Some("Test".to_string()), - proxy: ProxyType::Reverse(None), - server: Some("default".to_string()), - epg_timeshift: None, - epg_request_timeshift: None, - created_at: None, - exp_date: Some(1_672_705_545), - max_connections: 1, - status: Some(ProxyUserStatus::Expired), - output_clusters: ClusterFlags::all(), - ui_enabled: true, - comment: None, - priority: 0, - soft_connections: 0, - soft_priority: 0, - t_is_api_user: false, - network_access: None, - }), - Arc::new(ProxyUserCredentials { - username: "Test4".to_string(), - password: "Test".to_string(), - token: Some("Test".to_string()), - proxy: ProxyType::Reverse(None), - server: Some("default".to_string()), - epg_timeshift: None, - epg_request_timeshift: None, - created_at: None, - exp_date: Some(1_672_705_545), - max_connections: 1, - status: Some(ProxyUserStatus::Expired), - output_clusters: ClusterFlags::Live | ClusterFlags::Vod, - ui_enabled: true, - comment: None, - priority: -10, // non-zero priority to verify round-trip serialization - soft_connections: 2, - soft_priority: -3, - t_is_api_user: false, - network_access: Some(crate::model::NetworkAccess { + Arc::new(make_test_credential("Test", ProxyUserStatus::Active)), + Arc::new(make_test_credential("Test2", ProxyUserStatus::Expired)), + Arc::new(make_test_credential("Test3", ProxyUserStatus::Expired)), + Arc::new({ + let mut c = make_test_credential("Test4", ProxyUserStatus::Expired); + c.output_clusters = ClusterFlags::Live | ClusterFlags::Vod; + c.priority = -10; + c.soft_connections = 2; + c.soft_priority = -3; + c.network_access = Some(crate::model::NetworkAccess { allowed_countries: vec!["DE".to_string(), "AT".to_string()], allowed_networks: vec!["10.0.0.0/8".parse().unwrap(), "192.168.1.0/24".parse().unwrap()], - }), + }); + c }), ], }; diff --git a/backend/src/tools/flags_builder.rs b/backend/src/tools/flags_builder.rs index 2b96d1bcd..55573ba41 100644 --- a/backend/src/tools/flags_builder.rs +++ b/backend/src/tools/flags_builder.rs @@ -193,37 +193,7 @@ pub fn main() -> Result<(), Box> { #[cfg(test)] mod tests { use super::*; - use shared::utils::{country_code_to_index, flags::FlagFileHeader, index_to_country_code, FlagsLoader}; - - #[test] - fn test_country_code_encoding() { - assert_eq!(country_code_to_index("AA"), Some(0)); - assert_eq!(country_code_to_index("AB"), Some(1)); - assert_eq!(country_code_to_index("BA"), Some(26)); - assert_eq!(country_code_to_index("ZZ"), Some(675)); - assert_eq!(country_code_to_index("A"), None); - assert_eq!(country_code_to_index("AAA"), None); - assert_eq!(country_code_to_index("A1"), None); - assert_eq!(country_code_to_index("1A"), None); - } - - #[test] - fn test_index_to_country_code() { - assert_eq!(index_to_country_code(0), Some("AA".to_string())); - assert_eq!(index_to_country_code(1), Some("AB".to_string())); - assert_eq!(index_to_country_code(26), Some("BA".to_string())); - assert_eq!(index_to_country_code(675), Some("ZZ".to_string())); - assert_eq!(index_to_country_code(676), None); - } - - #[test] - fn test_round_trip() { - for code in ["AA", "DE", "US", "ZZ", "AB", "ZY"] { - let index = country_code_to_index(code).unwrap(); - let decoded = index_to_country_code(index).unwrap(); - assert_eq!(decoded, code); - } - } + use shared::utils::{flags::FlagFileHeader, FlagsLoader}; #[test] fn test_build_flags() { diff --git a/backend/src/utils/ffmpeg.rs b/backend/src/utils/ffmpeg.rs index 800844908..722f5a5ed 100644 --- a/backend/src/utils/ffmpeg.rs +++ b/backend/src/utils/ffmpeg.rs @@ -40,6 +40,16 @@ pub struct ProbeStreamStats { pub bitrate: Option, } +/// Common parameters shared across all probe variants. +#[derive(Debug, Clone, Copy)] +pub struct ProbeParams<'a> { + pub url: &'a str, + pub user_agent: Option<&'a str>, + pub analyze_duration: u64, + pub probe_size: u64, + pub timeout_secs: u64, +} + #[derive(Debug, Clone, Copy, Default)] pub struct FfmpegExecutor; @@ -110,13 +120,10 @@ impl FfmpegExecutor { pub async fn probe_url( &self, - url: &str, - user_agent: Option<&str>, - analyze_duration: u64, - probe_size: u64, - timeout_secs: u64, + params: &ProbeParams<'_>, proxy_cfg: Option<&ProxyConfig>, ) -> ProbeUrlOutcome { + let ProbeParams { url, user_agent, analyze_duration, probe_size, timeout_secs } = *params; // Determine timeout: Ensure it's at least as long as the analyze duration + buffer, // but respect the user setting if it's longer. let analyze_overhead = Duration::from_micros(analyze_duration) + Duration::from_secs(5); @@ -159,16 +166,12 @@ impl FfmpegExecutor { ProbeUrlOutcome::Failed(ProbeFailureKind::Other) } - #[allow(clippy::too_many_arguments)] pub async fn probe_remote_url( &self, client: &Client, - url: &str, - user_agent: Option<&str>, - analyze_duration: u64, - probe_size: u64, - timeout_secs: u64, + params: &ProbeParams<'_>, ) -> ProbeUrlOutcome { + let ProbeParams { url, user_agent, analyze_duration, probe_size, timeout_secs } = *params; let analyze_overhead = Duration::from_micros(analyze_duration) + Duration::from_secs(5); let config_timeout = Duration::from_secs(timeout_secs); let timeout_val = std::cmp::max(analyze_overhead, config_timeout); @@ -191,16 +194,12 @@ impl FfmpegExecutor { } } - #[allow(clippy::too_many_arguments)] pub async fn probe_remote_seekable_url( &self, client: &Client, - url: &str, - user_agent: Option<&str>, - analyze_duration: u64, - probe_size: u64, - timeout_secs: u64, + params: &ProbeParams<'_>, ) -> ProbeUrlOutcome { + let ProbeParams { url, user_agent, analyze_duration, probe_size, timeout_secs } = *params; let analyze_overhead = Duration::from_micros(analyze_duration) + Duration::from_secs(5); let config_timeout = Duration::from_secs(timeout_secs); let timeout_val = std::cmp::max(analyze_overhead, config_timeout); @@ -224,14 +223,9 @@ impl FfmpegExecutor { } /// Wrapper around [`Self::probe_url`] that races the probe against an optional cancellation token. - #[allow(clippy::too_many_arguments)] pub async fn probe_url_with_cancel( &self, - url: &str, - user_agent: Option<&str>, - analyze_duration: u64, - probe_size: u64, - timeout_secs: u64, + params: &ProbeParams<'_>, proxy_cfg: Option<&ProxyConfig>, cancel_token: Option<&tokio_util::sync::CancellationToken>, ) -> ProbeUrlOutcome { @@ -239,63 +233,53 @@ impl FfmpegExecutor { tokio::select! { biased; () = token.cancelled() => { - warn!("Probe preempted for {}", shared::utils::sanitize_sensitive_info(url)); + warn!("Probe preempted for {}", shared::utils::sanitize_sensitive_info(params.url)); ProbeUrlOutcome::Failed(ProbeFailureKind::Cancelled) } - result = self.probe_url(url, user_agent, analyze_duration, probe_size, timeout_secs, proxy_cfg) => result, + result = self.probe_url(params, proxy_cfg) => result, } } else { - self.probe_url(url, user_agent, analyze_duration, probe_size, timeout_secs, proxy_cfg).await + self.probe_url(params, proxy_cfg).await } } - #[allow(clippy::too_many_arguments)] pub async fn probe_remote_url_with_cancel( &self, client: &Client, - url: &str, - user_agent: Option<&str>, - analyze_duration: u64, - probe_size: u64, - timeout_secs: u64, + params: &ProbeParams<'_>, cancel_token: Option<&tokio_util::sync::CancellationToken>, ) -> ProbeUrlOutcome { if let Some(token) = cancel_token { tokio::select! { biased; () = token.cancelled() => { - warn!("Probe preempted for {}", shared::utils::sanitize_sensitive_info(url)); + warn!("Probe preempted for {}", shared::utils::sanitize_sensitive_info(params.url)); ProbeUrlOutcome::Failed(ProbeFailureKind::Cancelled) } - result = self.probe_remote_url(client, url, user_agent, analyze_duration, probe_size, timeout_secs) => result, + result = self.probe_remote_url(client, params) => result, } } else { - self.probe_remote_url(client, url, user_agent, analyze_duration, probe_size, timeout_secs).await + self.probe_remote_url(client, params).await } } - #[allow(clippy::too_many_arguments)] pub async fn probe_remote_seekable_url_with_cancel( &self, client: &Client, - url: &str, - user_agent: Option<&str>, - analyze_duration: u64, - probe_size: u64, - timeout_secs: u64, + params: &ProbeParams<'_>, cancel_token: Option<&tokio_util::sync::CancellationToken>, ) -> ProbeUrlOutcome { if let Some(token) = cancel_token { tokio::select! { biased; () = token.cancelled() => { - warn!("Probe preempted for {}", shared::utils::sanitize_sensitive_info(url)); + warn!("Probe preempted for {}", shared::utils::sanitize_sensitive_info(params.url)); ProbeUrlOutcome::Failed(ProbeFailureKind::Cancelled) } - result = self.probe_remote_seekable_url(client, url, user_agent, analyze_duration, probe_size, timeout_secs) => result, + result = self.probe_remote_seekable_url(client, params) => result, } } else { - self.probe_remote_seekable_url(client, url, user_agent, analyze_duration, probe_size, timeout_secs).await + self.probe_remote_seekable_url(client, params).await } } @@ -562,6 +546,17 @@ async fn ensure_probe_file_length(path: &Path, content_length: Option) -> R }) } +async fn flush_probe_file(file: &mut fs::File, path: &Path) -> Result<(), ProbeFailureKind> { + file.flush().await.map_err(|err| { + warn!( + "Failed to flush seekable ffprobe temp file {}: {}", + path.display(), + err + ); + ProbeFailureKind::Other + }) +} + async fn probe_local_path_with_ffprobe(path: &Path, analyze_duration: u64, probe_size: u64) -> ProbeUrlOutcome { let display_path = path.to_string_lossy().into_owned(); let mut command = Command::new("ffprobe"); @@ -601,16 +596,16 @@ async fn stage_seekable_probe_file( ) -> Result { let head_window = seekable_probe_window_bytes(probe_size); let tail_window = seekable_probe_window_bytes(probe_size); - let head_result = fetch_remote_span_to_file( + let head_result = fetch_remote_span_to_file(&RemoteSpanRequest { client, url, user_agent, - Some((0, head_window.saturating_sub(1))), - temp_path, - 0, - probe_bytes_limit(head_window), - true, - ) + range: Some((0, head_window.saturating_sub(1))), + path: temp_path, + offset: 0, + max_bytes: probe_bytes_limit(head_window), + allow_full_body_write_on_200: true, + }) .await?; ensure_probe_file_length(temp_path, head_result.total_length).await?; @@ -657,16 +652,16 @@ async fn stage_seekable_probe_file( tail_start ); - let tail_result = fetch_remote_span_to_file( + let tail_result = fetch_remote_span_to_file(&RemoteSpanRequest { client, url, user_agent, - Some((tail_start, content_length - 1)), - temp_path, - tail_start, - probe_bytes_limit(content_length - tail_start), - false, - ) + range: Some((tail_start, content_length - 1)), + path: temp_path, + offset: tail_start, + max_bytes: probe_bytes_limit(content_length - tail_start), + allow_full_body_write_on_200: false, + }) .await?; if tail_result.status == StatusCode::PARTIAL_CONTENT && tail_result.bytes_written > 0 { @@ -685,17 +680,19 @@ async fn stage_seekable_probe_file( Ok(stage) } -#[allow(clippy::too_many_arguments)] -async fn fetch_remote_span_to_file( - client: &Client, - url: &str, - user_agent: Option<&str>, +struct RemoteSpanRequest<'a> { + client: &'a Client, + url: &'a str, + user_agent: Option<&'a str>, range: Option<(u64, u64)>, - path: &Path, + path: &'a Path, offset: u64, max_bytes: usize, allow_full_body_write_on_200: bool, -) -> Result { +} + +async fn fetch_remote_span_to_file(req: &RemoteSpanRequest<'_>) -> Result { + let RemoteSpanRequest { client, url, user_agent, range, path, offset, max_bytes, allow_full_body_write_on_200 } = *req; let mut request = client.get(url); if let Some(ua) = user_agent { request = request.header(USER_AGENT, ua); @@ -792,6 +789,7 @@ async fn fetch_remote_span_to_file( break; } } + flush_probe_file(&mut file, path).await?; Ok(SpanWriteResult { status, diff --git a/backend/src/utils/file/file_utils.rs b/backend/src/utils/file/file_utils.rs index 132fd1e88..f867ac5b2 100644 --- a/backend/src/utils/file/file_utils.rs +++ b/backend/src/utils/file/file_utils.rs @@ -383,6 +383,20 @@ where Ok(()) } +pub fn collect_yaml_files(path: &Path) -> std::io::Result> { + let mut files = vec![]; + let mut visit = |entry: &std::fs::DirEntry, metadata: &std::fs::Metadata| { + if metadata.is_file() { + let file_path = entry.path(); + if file_path.extension().is_some_and(|ext| ext == "yml" || ext == "yaml") { + files.push(file_path); + } + } + }; + traverse_dir(path, &mut visit)?; + Ok(files) +} + pub fn prepare_file_path(persist: Option<&str>, storage_dir: &str, action: &str) -> Option { let persist_file: Option = persist.map(|persist_path| prepare_persist_path(persist_path, action)); @@ -529,7 +543,7 @@ pub fn get_file_extension(path: &str) -> Option { #[cfg(test)] mod tests { use super::{ - normalize_string_path, resolve_mapping_file_path, resolve_template_file_path, + collect_yaml_files, normalize_string_path, resolve_mapping_file_path, resolve_template_file_path, resolve_template_persist_file_path, }; use std::path::PathBuf; @@ -604,4 +618,15 @@ mod tests { let expected = config_path.join("template.d").join("template.yml"); assert_eq!(PathBuf::from(resolved), expected); } + + #[test] + fn test_collect_yaml_files_accepts_both_extensions() { + let temp_dir = tempfile::tempdir().unwrap(); + std::fs::write(temp_dir.path().join("one.yml"), "").unwrap(); + std::fs::write(temp_dir.path().join("two.yaml"), "").unwrap(); + std::fs::write(temp_dir.path().join("ignored.txt"), "").unwrap(); + + let files = collect_yaml_files(temp_dir.path()).unwrap(); + assert_eq!(files.len(), 2); + } } diff --git a/backend/src/utils/file/mapping_reader.rs b/backend/src/utils/file/mapping_reader.rs index f2889958e..c92cd58a9 100644 --- a/backend/src/utils/file/mapping_reader.rs +++ b/backend/src/utils/file/mapping_reader.rs @@ -1,7 +1,6 @@ use std::collections::HashMap; use crate::model::{Mappings}; use shared::error::TuliproxError; -use crate::utils::traverse_dir; use crate::utils::{config_file_reader, open_file}; use log::{warn}; use std::path::{Path, PathBuf}; @@ -75,16 +74,8 @@ fn merge_mapping_definitions(mappings: Vec) -> MappingsDto { } fn read_mappings_from_directory(path: &Path, resolve_env: bool) -> Result, MappingsDto)>, TuliproxError> { - let mut files = vec![]; - let mut visit = |entry: &std::fs::DirEntry, metadata: &std::fs::Metadata| { - if metadata.is_file() { - let file_path = entry.path(); - if file_path.extension().is_some_and(|ext| ext == "yml") { - files.push(file_path); - } - } - }; - traverse_dir(path, &mut visit).map_err(|err| TuliproxError::Io(format!("Failed to read mappings {err}")))?; + let mut files = crate::utils::collect_yaml_files(path) + .map_err(|err| TuliproxError::Io(format!("Failed to read mappings {err}")))?; files.sort_by(|a, b| a.file_name().cmp(&b.file_name())); diff --git a/backend/src/utils/file/template_reader.rs b/backend/src/utils/file/template_reader.rs index 6634394e2..552d52520 100644 --- a/backend/src/utils/file/template_reader.rs +++ b/backend/src/utils/file/template_reader.rs @@ -1,4 +1,3 @@ -use crate::utils::traverse_dir; use crate::utils::{config_file_reader, open_file}; use log::{debug, warn}; use shared::error::TuliproxError; @@ -69,16 +68,8 @@ fn read_templates_from_directory( path: &Path, resolve_env: bool, ) -> Result, TemplateDefinitionDto)>, TuliproxError> { - let mut files = vec![]; - let mut visit = |entry: &std::fs::DirEntry, metadata: &std::fs::Metadata| { - if metadata.is_file() { - let file_path = entry.path(); - if file_path.extension().is_some_and(|ext| ext == "yml") { - files.push(file_path); - } - } - }; - traverse_dir(path, &mut visit).map_err(|err| TuliproxError::Config(format!("Failed to read templates {err}")))?; + let mut files = crate::utils::collect_yaml_files(path) + .map_err(|err| TuliproxError::Config(format!("Failed to read templates {err}")))?; files.sort(); diff --git a/frontend/Cargo.toml b/frontend/Cargo.toml index 518eb6b2e..3837db458 100644 --- a/frontend/Cargo.toml +++ b/frontend/Cargo.toml @@ -40,11 +40,7 @@ strum_macros.workspace = true [dependencies.web-sys] version = "0.3.91" -features = ["Document", "Element", "Node", "Window", "HtmlCollection", "HtmlImageElement", "HtmlInputElement", - "DomRect", "HtmlDialogElement", "CssStyleDeclaration", - "Event", "MutationObserver", "MutationObserverInit", - "WebGlRenderingContext", "WebGlProgram", "WebGlShader", "WebGlBuffer", "WebGlUniformLocation", - "HtmlCanvasElement", "HtmlSelectElement"] +features = ["Document", "Element", "Node", "Window", "HtmlCollection", "HtmlImageElement", "HtmlInputElement", "DomRect", "HtmlDialogElement", "CssStyleDeclaration", "Event", "MutationObserver", "MutationObserverInit", "WebGlRenderingContext", "WebGlProgram", "WebGlShader", "WebGlBuffer", "WebGlUniformLocation", "HtmlCanvasElement", "HtmlSelectElement", "DomTokenList"] [dev-dependencies] wasm-bindgen-test = "0.3.64" diff --git a/frontend/public/assets/i18n/en.json b/frontend/public/assets/i18n/en.json index 2208b49ea..0af72e7c5 100644 --- a/frontend/public/assets/i18n/en.json +++ b/frontend/public/assets/i18n/en.json @@ -884,6 +884,7 @@ "CONTENT_SECURITY_POLICY": "Content Security Policy", "CONTENT_TYPE": "Content Type", "COPY_CREDENTIALS": "Copy Credentials", + "COPY_DETAILS": "Copy details", "COPY_LINK_PROVIDER_URL": "Copy Provider Url", "COPY_LINK_TULIPROX_VIRTUAL_ID": "Copy Virtual Id", "COPY_LINK_TULIPROX_WEBPLAYER_URL": "Copy Webplayer Url", @@ -1262,6 +1263,7 @@ "SKIP_VOD": "Skip VOD", "SLEEP_TIMER_MINS": "Sleep Timer (mins)", "SMART_EPG": "Smart Epg", + "SOMETHING_WENT_WRONG": "Something went wrong", "SORT": "Sort", "SORT_SETTINGS": "Sort Settings", "SOURCE": "Source", @@ -1400,6 +1402,7 @@ "ACTIVE_STREAMS": "Active Streams", "PLAYER": "Player", "TITLE": "Title", + "TOGGLE_SIDEBAR": "Toggle sidebar", "SELECTED": "Selected", "FILTERED": "Filtered", "TOTAL": "Total", @@ -1441,7 +1444,8 @@ "NEXT_PAGE": "Next Page", "PREVIOUS_PAGE": "Previous Page", "PAGE": "Page", - "PAGES": "Pages" + "PAGES": "Pages", + "NO_CONTENT": "No content" }, "MESSAGES": { "CLIPBOARD_NOT_SUPPORTED": "Clipboard not supported.\nYour browser or current context does not allow clipboard access.\nPlease use HTTPS or localhost.", @@ -1456,6 +1460,8 @@ "DOWNLOAD_CANCELLED": "Download cancelled.", "DOWNLOAD_REMOVED": "Download removed.", "DOWNLOAD_RETRIED": "Download retried.", + "CONFIRM_CANCEL": "Do you really want to cancel this transfer?", + "CONFIRM_REMOVE": "Do you really want to remove this entry?", "FAIL": "Failed to download!", "INVALID_RECORD_DURATION": "Recording duration must be a positive number of minutes.", "INVALID_RECORD_START": "Recording start time is invalid.", diff --git a/frontend/public/assets/i18n/index.json b/frontend/public/assets/i18n/index.json new file mode 100644 index 000000000..708e12d53 --- /dev/null +++ b/frontend/public/assets/i18n/index.json @@ -0,0 +1,5 @@ +{ + "languages": [ + { "code": "en", "label": "English", "dir": "ltr" } + ] +} diff --git a/frontend/public/assets/icons.json b/frontend/public/assets/icons.json index aa84ed50d..3ee32d07b 100644 --- a/frontend/public/assets/icons.json +++ b/frontend/public/assets/icons.json @@ -289,6 +289,10 @@ "keys": ["Editor"], "path": "M8 5v14l11-7z" }, + { + "keys": ["Language", "Globe", "Translate"], + "path": "M11.99 2C6.47 2 2 6.48 2 12s4.47 10 9.99 10C17.52 22 22 17.52 22 12S17.52 2 11.99 2zm6.93 6h-2.95c-.32-1.25-.78-2.45-1.38-3.56 1.84.63 3.37 1.91 4.33 3.56zM12 4.04c.83 1.2 1.48 2.53 1.91 3.96h-3.82c.43-1.43 1.08-2.76 1.91-3.96zM4.26 14C4.1 13.36 4 12.69 4 12s.1-1.36.26-2h3.38c-.08.66-.14 1.32-.14 2 0 .68.06 1.34.14 2H4.26zm.82 2h2.95c.32 1.25.78 2.45 1.38 3.56-1.84-.63-3.37-1.9-4.33-3.56zm2.95-8H5.08c.96-1.66 2.49-2.93 4.33-3.56C8.81 5.55 8.35 6.75 8.03 8zM12 19.96c-.83-1.2-1.48-2.53-1.91-3.96h3.82c-.43 1.43-1.08 2.76-1.91 3.96zM14.34 14H9.66c-.09-.66-.16-1.32-.16-2 0-.68.07-1.35.16-2h4.68c.09.65.16 1.32.16 2 0 .68-.07 1.34-.16 2zm.25 5.56c.6-1.11 1.06-2.31 1.38-3.56h2.95c-.96 1.65-2.49 2.93-4.33 3.56zM16.36 14c.08-.66.14-1.32.14-2 0-.68-.06-1.34-.14-2h3.38c.16.64.26 1.31.26 2s-.1 1.36-.26 2h-3.38z" + }, { "keys": ["ScheduleAdd"], "path": "m 15.427906,2 0.0026,4.3103469 -4.402514,-0.00424 0.0016,2.4026963 4.402514,0.00475 0.0026,4.2565168 2.402815,0.0026 -0.0021,-4.2565172 4.164457,0.00517 -0.0016,-2.403213 -4.164974,-0.00465 -0.0026,-4.3103469 z M 9.2342365,7.5143019 c -3.9841557,5e-7 -7.2343587,3.2535111 -7.2343586,7.2428491 0,3.989335 3.2502029,7.242849 7.2343586,7.242849 3.9841545,0 7.2322915,-3.253514 7.2322915,-7.242849 0,-0.06325 -0.0014,-0.126255 -0.0031,-0.18912 H 14.68171 c 0.0022,0.06278 0.0041,0.125793 0.0041,0.18912 0,3.025716 -2.429816,5.460169 -5.4516058,5.460169 -3.0217902,0 -5.4531561,-2.434453 -5.4531561,-5.460169 0,-3.025718 2.4313659,-5.4586186 5.4531561,-5.4586186 0.1480969,0 0.2937091,0.00764 0.4387121,0.019119 C 9.6657863,8.738416 9.6404593,8.1019918 9.6781163,7.5292876 9.531058,7.5202214 9.3835267,7.5143032 9.2342349,7.5143032 Z M 8.078806,10.651303 v 5.534575 h 5.462975 V 14.71943 H 9.5432468 v -4.068127 z" diff --git a/frontend/scss/app/_component.scss b/frontend/scss/app/_component.scss index c0de78ad8..809924f50 100644 --- a/frontend/scss/app/_component.scss +++ b/frontend/scss/app/_component.scss @@ -64,6 +64,8 @@ @forward "components/search"; @forward "components/loading_indicator"; @forward "components/theme_picker"; +@forward "components/language_picker"; +@forward "components/error_boundary"; @forward "components/toastr"; @forward "components/drop_down_icon_button"; @forward "components/downloads"; diff --git a/frontend/scss/app/components/_card.scss b/frontend/scss/app/components/_card.scss index 856c3e23b..d3520cb46 100644 --- a/frontend/scss/app/components/_card.scss +++ b/frontend/scss/app/components/_card.scss @@ -8,4 +8,18 @@ border-radius: var(--border-radius); position: relative; gap: var(--gap-default); -} \ No newline at end of file + + @media (prefers-reduced-motion: no-preference) { + transition: + transform 200ms ease, + box-shadow 200ms ease, + border-color 200ms ease; + } + + @media (hover: hover) { + &:hover { + box-shadow: 0 8px 24px var(--theme-palette-neutral-surface-660, rgba(0, 0, 0, 0.25)); + background-color: var(--card-hover-background-color); + } + } +} diff --git a/frontend/scss/app/components/_collapse_panel.scss b/frontend/scss/app/components/_collapse_panel.scss index 8638ccce1..8633b55a3 100644 --- a/frontend/scss/app/components/_collapse_panel.scss +++ b/frontend/scss/app/components/_collapse_panel.scss @@ -34,6 +34,20 @@ } } + &__chevron { + display: inline-flex; + align-items: center; + transform-origin: center; + + @media (prefers-reduced-motion: no-preference) { + transition: transform 220ms ease; + } + } + + &:not(.tp__collapsed) .tp__collapse-panel__chevron { + transform: rotate(180deg); + } + &__body { display: flex; flex-flow: column; diff --git a/frontend/scss/app/components/_error_boundary.scss b/frontend/scss/app/components/_error_boundary.scss new file mode 100644 index 000000000..d87958412 --- /dev/null +++ b/frontend/scss/app/components/_error_boundary.scss @@ -0,0 +1,40 @@ +.tp__error-boundary { + display: flex; + flex-flow: column; + justify-content: center; + align-items: center; + gap: var(--padding-mini); + padding: var(--padding-default); + width: 100%; + box-sizing: border-box; + text-align: center; + + &__icon { + display: flex; + justify-content: center; + align-items: center; + color: var(--color-error, #d9534f); + fill: currentColor; + + .svg-icon { + width: 2.5rem; + height: 2.5rem; + } + } + + &__title { + font-weight: 600; + font-size: 1.1rem; + } + + &__message { + max-width: 40rem; + opacity: 0.85; + white-space: pre-wrap; + overflow-wrap: anywhere; + } + + &__retry { + margin-top: var(--padding-mini); + } +} diff --git a/frontend/scss/app/components/_icon_button.scss b/frontend/scss/app/components/_icon_button.scss index 23d1ca25a..1466dc63d 100644 --- a/frontend/scss/app/components/_icon_button.scss +++ b/frontend/scss/app/components/_icon_button.scss @@ -12,6 +12,8 @@ box-sizing: border-box; height: min-content; width: min-content; + position: relative; + overflow: hidden; &:focus, &:hover { @@ -24,6 +26,36 @@ pointer-events: none; } + @media (prefers-reduced-motion: no-preference) { + transition: transform 120ms ease; + + &::after { + content: ""; + position: absolute; + inset: 0; + pointer-events: none; + border-radius: inherit; + background-image: radial-gradient(circle, currentColor 12%, transparent 12.5%); + background-position: center; + background-repeat: no-repeat; + opacity: 0; + transform: scale(12); + transition: + transform 0.5s ease, + opacity 0.8s ease; + } + + &:active { + transform: scale(0.92); + + &::after { + opacity: 0.2; + transform: scale(0); + transition: 0s; + } + } + } + &__remove { fill: var(--menu-button-warn-color); &:hover { @@ -67,4 +99,4 @@ fill: var(--text-button-active-hover-color); outline: none; } -} \ No newline at end of file +} diff --git a/frontend/scss/app/components/_language_picker.scss b/frontend/scss/app/components/_language_picker.scss new file mode 100644 index 000000000..5a21c4a2e --- /dev/null +++ b/frontend/scss/app/components/_language_picker.scss @@ -0,0 +1,4 @@ +.tp__language-picker__item.active { + background-color: var(--menu-item-hover-background-color); + color: var(--menu-item-hover-color); +} diff --git a/frontend/scss/app/components/_menu_item.scss b/frontend/scss/app/components/_menu_item.scss index 1d0fd01c0..e8dbcf661 100644 --- a/frontend/scss/app/components/_menu_item.scss +++ b/frontend/scss/app/components/_menu_item.scss @@ -5,9 +5,15 @@ gap: var(--gap-default); color: var(--menu-text-color); fill: currentColor; + appearance: none; + background-color: transparent; + border: none; + box-sizing: content-box; padding: 12px; border-radius: var(--border-radius); cursor: pointer; + font: inherit; + text-align: left; height: 1.3rem; .svg-icon { @@ -23,4 +29,4 @@ background-color: var(--menu-item-hover-background-color); color: var(--menu-item-hover-color); } -} \ No newline at end of file +} diff --git a/frontend/scss/app/components/_text_button.scss b/frontend/scss/app/components/_text_button.scss index 3cad61cdd..5883a7bf0 100644 --- a/frontend/scss/app/components/_text_button.scss +++ b/frontend/scss/app/components/_text_button.scss @@ -17,6 +17,8 @@ padding: 0 var(--padding-default); font-weight: bold; font-size: var(--font-size-norm); + position: relative; + overflow: hidden; svg { width: 1.5rem; } @@ -31,6 +33,39 @@ color: var(--text-button-hover-color); outline: none; } + + @media (prefers-reduced-motion: no-preference) { + transition: + background-color 160ms ease, + color 160ms ease, + transform 120ms ease; + + &::after { + content: ""; + position: absolute; + inset: 0; + pointer-events: none; + border-radius: inherit; + background-image: radial-gradient(circle, currentColor 10%, transparent 10.5%); + background-position: center; + background-repeat: no-repeat; + opacity: 0; + transform: scale(12); + transition: + transform 0.5s ease, + opacity 0.8s ease; + } + + &:active:not(.disabled) { + transform: scale(0.98); + + &::after { + opacity: 0.18; + transform: scale(0); + transition: 0s; + } + } + } } .tp__text-button.primary:not(.active) { diff --git a/frontend/scss/app/components/_toastr.scss b/frontend/scss/app/components/_toastr.scss index bf7574631..d1f0540da 100644 --- a/frontend/scss/app/components/_toastr.scss +++ b/frontend/scss/app/components/_toastr.scss @@ -18,6 +18,8 @@ font-size: var(--font-size-md); box-shadow: var(--toastr-shadow); animation: fadeIn 0.3s ease-out; + position: relative; + overflow: hidden; &__message { display: flex; @@ -28,6 +30,33 @@ width: 1.5rem; height: 1.5rem; } + + &__progress { + position: absolute; + left: 0; + bottom: 0; + height: 3px; + width: 100%; + transform-origin: left center; + background-color: currentColor; + opacity: 0.45; + animation-name: tp-toast-progress; + animation-timing-function: linear; + animation-fill-mode: forwards; + } + + &:hover &__progress { + animation-play-state: paused; + } +} + +@keyframes tp-toast-progress { + from { + transform: scaleX(1); + } + to { + transform: scaleX(0); + } } .tp__toast.success { @@ -73,3 +102,14 @@ transform: translateY(0); } } + +@media (prefers-reduced-motion: reduce) { + .tp__toast { + animation: none; + } + + .tp__toast__progress { + animation: none; + display: none; + } +} diff --git a/frontend/scss/main.scss b/frontend/scss/main.scss index 075a9e807..561d15101 100644 --- a/frontend/scss/main.scss +++ b/frontend/scss/main.scss @@ -49,3 +49,17 @@ body { max-width: 100%; background-color: inherit; } + +@media (prefers-reduced-motion: no-preference) { + body.tp-theme-transition, + body.tp-theme-transition * { + transition: + background-color 280ms ease, + color 280ms ease, + border-color 280ms ease, + outline-color 280ms ease, + fill 280ms ease, + stroke 280ms ease, + box-shadow 280ms ease !important; + } +} diff --git a/frontend/src/app/components/accordion_panel.rs b/frontend/src/app/components/accordion_panel.rs index 3ab608808..095fddb57 100644 --- a/frontend/src/app/components/accordion_panel.rs +++ b/frontend/src/app/components/accordion_panel.rs @@ -38,7 +38,9 @@ pub fn AccordionPanel(props: &AccordionPanelProps) -> Html { { props.title_content.clone().unwrap_or_else(|| html! { &props.title }) } - + + +
{ for props.children.iter() } diff --git a/frontend/src/app/components/api_user/api_user_view.rs b/frontend/src/app/components/api_user/api_user_view.rs index 102cb2995..28c46e7a2 100644 --- a/frontend/src/app/components/api_user/api_user_view.rs +++ b/frontend/src/app/components/api_user/api_user_view.rs @@ -1,9 +1,10 @@ use crate::{ app::components::{ - api_user::playlist::ApiUserPlaylist, loading_indicator::BusyIndicator, theme::Theme, AppIcon, IconButton, - ThemePicker, ToastrView, WebsocketStatus, + api_user::playlist::ApiUserPlaylist, loading_indicator::BusyIndicator, theme::Theme, AppIcon, ErrorBoundary, + IconButton, LanguagePicker, ThemePicker, ToastrView, WebsocketStatus, }, hooks::use_service_context, + i18n::use_translation, provider::DialogProvider, }; use yew::{component, html, use_state, Callback, Html}; @@ -11,6 +12,7 @@ use yew::{component, html, use_state, Callback, Html}; #[component] pub fn ApiUserView() -> Html { let services = use_service_context(); + let translate = use_translation(); let theme = use_state(Theme::get_current_theme); let handle_theme_select = { @@ -45,12 +47,15 @@ pub fn ApiUserView() -> Html {
+
- + + +
diff --git a/frontend/src/app/components/collapse_panel.rs b/frontend/src/app/components/collapse_panel.rs index f314f14fb..3594a1601 100644 --- a/frontend/src/app/components/collapse_panel.rs +++ b/frontend/src/app/components/collapse_panel.rs @@ -37,7 +37,9 @@ pub fn CollapsePanel(props: &CollapsePanelProps) -> Html { { props.title_content.clone().unwrap_or_else(|| html! { &props.title }) } - + + +
{ for props.children.iter() } diff --git a/frontend/src/app/components/dashboard/stream_history_view.rs b/frontend/src/app/components/dashboard/stream_history_view.rs index 9d53330aa..72d941dab 100644 --- a/frontend/src/app/components/dashboard/stream_history_view.rs +++ b/frontend/src/app/components/dashboard/stream_history_view.rs @@ -1,11 +1,11 @@ use crate::{ app::components::{ Country, DateInput, DropDownOption, PagedTable, RevealContent, Search, TabItem, TabSet, Table, TableDefinition, - TextButton, + TextButton, PAGE_SIZES, TP_PAGE_SIZE_KEY, }, hooks::use_service_context, i18n::use_translation, - utils::{format_bytes, format_duration, format_ts}, + utils::{format_bytes, format_duration, format_ts, get_local_storage_item, set_local_storage_item}, }; use futures::join; use shared::{ @@ -201,7 +201,12 @@ pub fn StreamHistoryView() -> Html { let to_date = use_state(|| Some(today_start_ts())); let paged_response = use_state(|| None::>); let page = use_state(|| 1u32); - let page_size = use_state(default_page_size); + let page_size = use_state(|| { + get_local_storage_item(TP_PAGE_SIZE_KEY) + .and_then(|v| v.parse::().ok()) + .filter(|size| PAGE_SIZES.contains(size)) + .unwrap_or_else(default_page_size) + }); let summaries = use_state(Vec::::new); let qos_snapshots = use_state(Vec::::new); let selected_qos_snapshot = use_state(|| None::); @@ -391,6 +396,7 @@ pub fn StreamHistoryView() -> Html { Callback::from(move |new_size: u16| { page.set(1); page_size.set(new_size); + set_local_storage_item(TP_PAGE_SIZE_KEY, &new_size.to_string()); fetch_history_page(1, new_size, (*search_filter).clone()); }) }; diff --git a/frontend/src/app/components/downloads.rs b/frontend/src/app/components/downloads.rs index 37fef9f96..20a6ec330 100644 --- a/frontend/src/app/components/downloads.rs +++ b/frontend/src/app/components/downloads.rs @@ -2,7 +2,8 @@ use crate::{ app::components::{IconButton, Table, TableDefinition, TextButton}, hooks::use_service_context, i18n::use_translation, - model::EventMessage, + model::{DialogResult, EventMessage}, + services::DialogService, utils::format_bytes, }; use shared::{ @@ -284,6 +285,7 @@ fn apply_download_delta( pub fn downloads_view() -> Html { let translate = use_translation(); let services = use_service_context(); + let dialog = use_context::().expect("Dialog service not found"); let has_download_write = services.auth.has_permission(Permission::DownloadWrite); let active_tab = use_state(|| DownloadTab::Queue); let queue_state = use_state(|| Rc::new(Vec::::new())); @@ -407,12 +409,17 @@ pub fn downloads_view() -> Html { let services = services.clone(); let translate = translate.clone(); let active_download = active_download.clone(); + let dialog = dialog.clone(); Callback::from(move |uuid: String| { let request_downloads = request_downloads.clone(); let services = services.clone(); let translate = translate.clone(); let active_download = active_download.clone(); + let dialog = dialog.clone(); spawn_local(async move { + if dialog.confirm(&translate.t("MESSAGES.DOWNLOAD.CONFIRM_CANCEL")).await != DialogResult::Ok { + return; + } let optimistic_delta = optimistic_active_delta(&active_download, &uuid, TransferStatusDto::Cancelled); let result = services.downloads.cancel_download(uuid).await; handle_download_action_result( @@ -431,11 +438,16 @@ pub fn downloads_view() -> Html { let request_downloads = request_downloads.clone(); let services = services.clone(); let translate = translate.clone(); + let dialog = dialog.clone(); Callback::from(move |uuid: String| { let request_downloads = request_downloads.clone(); let services = services.clone(); let translate = translate.clone(); + let dialog = dialog.clone(); spawn_local(async move { + if dialog.confirm(&translate.t("MESSAGES.DOWNLOAD.CONFIRM_REMOVE")).await != DialogResult::Ok { + return; + } let result = services.downloads.remove_download(uuid).await; handle_download_action_result( result, diff --git a/frontend/src/app/components/error_boundary.rs b/frontend/src/app/components/error_boundary.rs new file mode 100644 index 000000000..b2249e232 --- /dev/null +++ b/frontend/src/app/components/error_boundary.rs @@ -0,0 +1,93 @@ +use crate::{ + app::components::{AppIcon, TextButton}, + i18n::use_translation, +}; +use yew::prelude::*; + +#[derive(Clone, PartialEq)] +pub struct ErrorBoundaryHandle { + report: Callback, + reset: Callback<()>, +} + +impl ErrorBoundaryHandle { + #[allow(dead_code)] + pub fn report(&self, message: impl Into) { self.report.emit(message.into()); } + + #[allow(dead_code)] + pub fn reset(&self) { self.reset.emit(()); } +} + +#[derive(Properties, Clone, PartialEq)] +pub struct ErrorBoundaryProps { + #[prop_or_default] + pub name: AttrValue, + #[prop_or_default] + pub class: Classes, + #[prop_or_default] + pub children: Children, +} + +#[component] +pub fn ErrorBoundary(props: &ErrorBoundaryProps) -> Html { + let translate = use_translation(); + let error = use_state(|| None::); + let generation = use_state(|| 0usize); + + let report = { + let error = error.clone(); + Callback::from(move |message: String| { + log::error!("ErrorBoundary captured: {message}"); + error.set(Some(message)); + }) + }; + + let reset = { + let error = error.clone(); + let generation = generation.clone(); + Callback::from(move |()| { + error.set(None); + generation.set(*generation + 1); + }) + }; + + let handle = ErrorBoundaryHandle { report, reset: reset.clone() }; + + if let Some(message) = error.as_ref() { + let on_retry = { + let reset = reset.clone(); + Callback::from(move |_: String| reset.emit(())) + }; + let title = if props.name.is_empty() { + translate.t("LABEL.SOMETHING_WENT_WRONG") + } else { + format!("{} — {}", props.name, translate.t("LABEL.SOMETHING_WENT_WRONG")) + }; + + return html! { + + }; + } + + html! { + context={handle} key={*generation}> + { for props.children.iter() } + > + } +} + +#[hook] +pub fn use_error_boundary() -> Option { use_context::() } diff --git a/frontend/src/app/components/filter/filter_editor.rs b/frontend/src/app/components/filter/filter_editor.rs index 6cfb6f9ca..110b77819 100644 --- a/frontend/src/app/components/filter/filter_editor.rs +++ b/frontend/src/app/components/filter/filter_editor.rs @@ -10,7 +10,11 @@ use shared::{ model::PatternTemplate, }; use web_sys::InputEvent; -use yew::{classes, component, html, use_context, use_effect_with, use_state, Callback, Html, Properties, TargetCast}; +use yew::{ + classes, component, html, use_context, use_effect_with, use_mut_ref, use_state, Callback, Html, Properties, + TargetCast, +}; +use yew_hooks::use_debounce; #[derive(Properties, Clone, PartialEq, Debug)] pub struct FilterEditorProps { @@ -23,6 +27,8 @@ pub struct FilterEditorProps { pub on_templates_change: Callback>>, } +const FILTER_INPUT_DEBOUNCE_MS: u32 = 300; + pub(crate) fn parse_filter_preview( filter: Option<&str>, templates: Option<&[PatternTemplate]>, @@ -46,6 +52,10 @@ pub fn FilterEditor(props: &FilterEditorProps) -> Html { let templates_state = use_state(|| cfg_templates.clone()); let filter_state = use_state(|| props.filter.clone()); + // Raw textarea value, updated immediately on every keystroke so typing stays responsive. + let input_value = use_state(|| props.filter.clone().unwrap_or_default()); + // Latest pending value awaiting the debounced parse/emit. + let pending_value = use_mut_ref(|| props.filter.clone()); { let templates = templates_state.clone(); @@ -58,8 +68,12 @@ pub fn FilterEditor(props: &FilterEditorProps) -> Html { { let filter = filter_state.clone(); + let input_value = input_value.clone(); + let pending_value = pending_value.clone(); use_effect_with(props.filter.clone(), move |flt| { filter.set(flt.clone()); + input_value.set(flt.clone().unwrap_or_default()); + *pending_value.borrow_mut() = flt.clone(); }); } @@ -73,19 +87,30 @@ pub fn FilterEditor(props: &FilterEditorProps) -> Html { }); } - let handle_filter_input = { + let debounce = { let filter = filter_state.clone(); let on_filter_change = props.on_filter_change.clone(); + let pending_value = pending_value.clone(); + use_debounce( + move || { + let next = pending_value.borrow().clone(); + filter.set(next.clone()); + on_filter_change.emit(next); + }, + FILTER_INPUT_DEBOUNCE_MS, + ) + }; + + let handle_filter_input = { + let input_value = input_value.clone(); + let pending_value = pending_value.clone(); + let debounce = debounce.clone(); Callback::from(move |event: InputEvent| { if let Some(input) = event.target_dyn_into::() { let value = input.value(); - if value.is_empty() { - filter.set(None); - on_filter_change.emit(None); - } else { - filter.set(Some(value.clone())); - on_filter_change.emit(Some(value)); - } + input_value.set(value.clone()); + *pending_value.borrow_mut() = if value.is_empty() { None } else { Some(value) }; + debounce.run(); } }) }; @@ -116,7 +141,7 @@ pub fn FilterEditor(props: &FilterEditorProps) -> Html {
-