diff --git a/CHANGELOG.md b/CHANGELOG.md index f061458d0..dcd7b50b6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -171,6 +171,8 @@ active URL of the specified provider. - Added `metadata_update.ffprobe.enabled` (default: false), `metadata_update.ffprobe.timeout`, and ffprobe probe/analyze size settings. - `metadata_update.ffprobe.analyze_duration` and `metadata_update.ffprobe.live_analyze_duration` require explicit unit suffixes (`s|m|h|d`). - FFprobe settings are configured under `metadata_update.ffprobe` (not under `video`). + - Added `metadata_update.probe_fairness_resolve_burst` (default `200`) to control fairness between resolve and probe tasks. + After N consecutive resolve-domain tasks, one pending probe-domain task is prioritized to avoid probe starvation. - Added `template_path` (optional): path to a template file (`template.yml`) or directory (`template.d` style). - **source.yml (input options)**: - Added `resolve_tmdb`: Triggers TMDB lookup if ID is missing. diff --git a/README.md b/README.md index c3d870557..c72a838d9 100644 --- a/README.md +++ b/README.md @@ -488,6 +488,7 @@ metadata_update: worker_idle_timeout: 1m max_queue_size: 100000 no_change_cache_ttl_secs: 3600 + probe_fairness_resolve_burst: 200 ``` **Field Reference (`metadata_update`):** @@ -523,6 +524,8 @@ metadata_update: - `max_queue_size` (default `100000`): Maximum pending metadata tasks per input before new tasks are rejected. - `no_change_cache_ttl_secs` (default `3600`): TTL in seconds for the per-item no-change deduplication cache. While valid, identical reason sets are skipped. +- `probe_fairness_resolve_burst` (default `200`): After this many consecutive resolve-domain tasks, the worker prefers one + pending probe-domain task to prevent probe starvation. - `ffprobe.enabled` (default `false`): Globally enables/disables FFprobe-based stream analysis. - `ffprobe.timeout` (default `60`): FFprobe process timeout in seconds. - `ffprobe.analyze_duration` (default `10s`): FFprobe `-analyzeduration` @@ -540,6 +543,7 @@ metadata_update: - `worker_idle_timeout`: How long a metadata worker stays alive with no new work before it can stop and free resources. - `max_queue_size`: Safety limit for how many metadata tasks are kept in memory at once. - `no_change_cache_ttl_secs`: How long (seconds) a no-change resolve result is remembered before the same reason set is allowed to execute again. +- `probe_fairness_resolve_burst`: How many resolve tasks may run in a row before one pending probe task is prioritized. - `log.queue_interval`: How often Tuliprox writes queue status logs. - `log.progress_interval`: How often Tuliprox writes progress logs while tasks are running. - `resolve.max_retry_backoff`: Longest wait time between repeated resolve retries. diff --git a/backend/src/api/model/metadata_update_manager.rs b/backend/src/api/model/metadata_update_manager.rs index 45ebf6749..de0f3bbae 100644 --- a/backend/src/api/model/metadata_update_manager.rs +++ b/backend/src/api/model/metadata_update_manager.rs @@ -88,6 +88,7 @@ struct MetadataUpdateRuntimeSettings { worker_idle_timeout_secs: u64, max_queue_size: usize, no_change_cache_ttl_secs: u64, + probe_fairness_resolve_burst: usize, probe_retry_backoff_step_1_secs: u64, probe_retry_backoff_step_2_secs: u64, probe_retry_backoff_step_3_secs: u64, @@ -133,6 +134,7 @@ impl MetadataUpdateRuntimeSettings { worker_idle_timeout_secs: cfg.worker_idle_timeout_secs.max(1), max_queue_size: cfg.max_queue_size.max(1), no_change_cache_ttl_secs: cfg.no_change_cache_ttl_secs.max(1), + probe_fairness_resolve_burst: cfg.probe_fairness_resolve_burst.max(1), probe_retry_backoff_step_1_secs: cfg.probe.retry_backoff_step_1_secs.max(1), probe_retry_backoff_step_2_secs: cfg.probe.retry_backoff_step_2_secs.max(1), probe_retry_backoff_step_3_secs: cfg.probe.retry_backoff_step_3_secs.max(1), @@ -879,6 +881,27 @@ impl MetadataUpdateManager { /// Get the number of active workers (for monitoring/debugging) pub fn active_worker_count(&self) -> usize { self.workers.len() } + + /// Returns `true` when an equivalent task is already pending for this input and + /// submitting `task` would not change the queued payload after merge semantics. + pub fn is_redundant_with_pending_task(&self, input_name: &str, task: &UpdateTask) -> bool { + let Some(ctx) = self.workers.get(input_name) else { + return false; + }; + + // If the worker channel is already closed, let normal enqueue recovery run. + if ctx.sender.is_closed() { + return false; + } + + let key = TaskKey::from_task(task); + let Some(entry) = ctx.pending_tasks.get(&key) else { + return false; + }; + + let mut merged = entry.task.lock().clone(); + !Self::merge_task_payload(&mut merged, task.clone()) + } } struct DbHandle { @@ -931,6 +954,7 @@ impl InputWorker { let mut last_progress_log_at = Instant::now(); let mut queue_cycle_active = false; let mut cycle_had_changes = false; + let mut consecutive_resolve_tasks = 0_usize; let input_name = self.input_name.clone(); let app_state_weak = self.app_state_weak.clone(); @@ -948,8 +972,24 @@ impl InputWorker { runtime_settings = self.runtime_settings(); last_runtime_settings_refresh_at = Instant::now(); } - let task_data = if let Some(t) = next_task.take() { - Some(t) + let prefer_probe = consecutive_resolve_tasks >= runtime_settings.probe_fairness_resolve_burst; + let task_data = if let Some(prefetched) = next_task.take() { + if prefer_probe && Self::retry_domain_for_task(&prefetched.1) == RetryDomain::Resolve { + if let Some(probe_task) = self.take_pending_probe_task_snapshot() { + next_task = Some(prefetched); + Some(probe_task) + } else { + Some(prefetched) + } + } else { + Some(prefetched) + } + } else if prefer_probe { + if let Some(probe_task) = self.take_pending_probe_task_snapshot() { + Some(probe_task) + } else { + self.recv_task_fast_or_wait(&runtime_settings).await + } } else { self.recv_task_fast_or_wait(&runtime_settings).await }; @@ -983,6 +1023,7 @@ impl InputWorker { debug!("Background metadata update queue has entries for input {input_name}; starting processing"); } + let current_retry_domain = Self::retry_domain_for_task(¤t_task); let delay_secs = current_task.delay(); let mut schedule_requeue_at_ts: Option = None; let mut remove_current_task = false; @@ -1474,6 +1515,12 @@ impl InputWorker { } } + if current_retry_domain == RetryDomain::Resolve { + consecutive_resolve_tasks = consecutive_resolve_tasks.saturating_add(1); + } else { + consecutive_resolve_tasks = 0; + } + // Try to get the next task immediately to keep locks open. // Ignore phantom signals (channel key without pending map entry). if next_task.is_none() { @@ -1512,6 +1559,7 @@ impl InputWorker { queue_cycle_active = false; processed_vod_count = 0; processed_series_count = 0; + consecutive_resolve_tasks = 0; if cycle_had_changes { info!("All pending metadata resolves completed for input {input_name} (with changes)"); if let Some(app_state) = app_state_weak.as_ref().and_then(Weak::upgrade) { @@ -1752,6 +1800,22 @@ impl InputWorker { Some((key, task, generation)) } + fn take_pending_probe_task_snapshot(&self) -> Option<(TaskKey, UpdateTask, u64)> { + for entry in self.pending_tasks.iter() { + if self.scheduled_requeues.contains_key(entry.key()) { + continue; + } + + let generation = entry.generation.load(Ordering::Relaxed); + let task = entry.task.lock().clone(); + if Self::retry_domain_for_task(&task) == RetryDomain::Probe { + return Some((entry.key().clone(), task, generation)); + } + } + + None + } + fn runtime_settings(&self) -> MetadataUpdateRuntimeSettings { MetadataUpdateRuntimeSettings::from_app_state(self.app_state_weak.as_ref()) } @@ -3545,6 +3609,46 @@ mod tests { assert_eq!(InputWorker::retry_domain_for_task(&task), RetryDomain::Resolve); } + #[test] + fn take_pending_probe_task_snapshot_skips_scheduled_requeues() { + let (tx, rx) = mpsc::channel::(8); + let pending_tasks = Arc::new(DashMap::new()); + let pending_task_count = Arc::new(AtomicUsize::new(1)); + let worker = create_test_worker("input_probe_skip", tx, rx, pending_tasks.clone(), pending_task_count); + + let key = TaskKey::Live(100); + let task = UpdateTask::ProbeLive { + id: ProviderIdType::Id(100), + reason: ResolveReasonSet::from_variants(&[ResolveReason::Probe]), + delay: 0, + interval: 60, + }; + pending_tasks.insert(key.clone(), PendingTask::new(task)); + worker.scheduled_requeues.insert(key, chrono::Utc::now().timestamp().saturating_add(30)); + + assert!(worker.take_pending_probe_task_snapshot().is_none()); + } + + #[test] + fn take_pending_probe_task_snapshot_accepts_probe_only_resolve() { + let (tx, rx) = mpsc::channel::(8); + let pending_tasks = Arc::new(DashMap::new()); + let pending_task_count = Arc::new(AtomicUsize::new(1)); + let worker = create_test_worker("input_probe_only_resolve", tx, rx, pending_tasks.clone(), pending_task_count); + + let key = TaskKey::Vod(77); + let task = UpdateTask::ResolveVod { + id: ProviderIdType::Id(77), + reason: ResolveReasonSet::from_variants(&[ResolveReason::Probe]), + delay: 0, + }; + pending_tasks.insert(key.clone(), PendingTask::new(task)); + + let snapshot = worker.take_pending_probe_task_snapshot().expect("expected probe-domain snapshot"); + assert_eq!(snapshot.0, key); + assert_eq!(InputWorker::retry_domain_for_task(&snapshot.1), RetryDomain::Probe); + } + #[test] fn playlist_trigger_ignores_probe_only_changes() { let task = UpdateTask::ProbeLive { diff --git a/backend/src/api/setup_api.rs b/backend/src/api/setup_api.rs index 5902c90a1..f20eb2fb7 100644 --- a/backend/src/api/setup_api.rs +++ b/backend/src/api/setup_api.rs @@ -496,20 +496,50 @@ async fn setup_root_file( async fn api_not_found() -> impl IntoResponse + Send { StatusCode::NOT_FOUND.into_response() } -async fn persist_yaml_file(file_path: &FsPath, payload: &T) -> Result<(), String> { - let mut content = String::new(); - let options = serde_saphyr::SerializerOptions { prefer_block_scalars: false, ..Default::default() }; - serde_saphyr::to_fmt_writer_with_options(&mut content, payload, options).map_err(|err| err.to_string())?; - tokio::fs::write(file_path, content).await.map_err(|err| err.to_string()) +enum SetupPersistWriteError { + Serialize(String), + Io(std::io::Error), } -async fn ensure_parent_dir(file_path: &FsPath) -> Result<(), String> { +async fn persist_yaml_file( + file_path: &FsPath, + payload: &T, +) -> Result<(), SetupPersistWriteError> { + let mut content = String::new(); + let options = serde_saphyr::SerializerOptions { prefer_block_scalars: false, ..Default::default() }; + serde_saphyr::to_fmt_writer_with_options(&mut content, payload, options) + .map_err(|err| SetupPersistWriteError::Serialize(err.to_string()))?; + tokio::fs::write(file_path, content).await.map_err(SetupPersistWriteError::Io) +} + +async fn ensure_parent_dir(file_path: &FsPath) -> std::io::Result<()> { if let Some(parent_dir) = file_path.parent() { - tokio::fs::create_dir_all(parent_dir).await.map_err(|err| err.to_string())?; + tokio::fs::create_dir_all(parent_dir).await?; } Ok(()) } +fn setup_io_error_response(file_path: &FsPath, operation: &str, err: &std::io::Error) -> axum::response::Response { + let (status, message) = if err.kind() == ErrorKind::PermissionDenied { + error!( + "Setup mode: permission denied while trying to {operation} '{}': {err}", + file_path.display() + ); + ( + StatusCode::FORBIDDEN, + "Permission denied while accessing requested resource".to_string(), + ) + } else { + error!( + "Setup mode: failed to {operation} '{}': {err}", + file_path.display() + ); + (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error".to_string()) + }; + + (status, axum::Json(json!({ "error": message }))).into_response() +} + fn validate_web_users(users: &[SetupWebUserCredentialDto]) -> Result, String> { if users.is_empty() { return Err("At least one WebUI user is required".to_string()); @@ -598,13 +628,13 @@ enum SetupPersistAction<'a> { } impl SetupPersistAction<'_> { - async fn write(&self, temp_path: &FsPath) -> Result<(), String> { + async fn write(&self, temp_path: &FsPath) -> Result<(), SetupPersistWriteError> { match self { Self::Config(payload) => persist_yaml_file(temp_path, *payload).await, Self::Sources(payload) => persist_yaml_file(temp_path, *payload).await, Self::Templates(payload) => persist_yaml_file(temp_path, *payload).await, Self::ApiProxy(payload) => persist_yaml_file(temp_path, *payload).await, - Self::UserFile(content) => tokio::fs::write(temp_path, content).await.map_err(|err| err.to_string()), + Self::UserFile(content) => tokio::fs::write(temp_path, content).await.map_err(SetupPersistWriteError::Io), } } } @@ -688,13 +718,7 @@ async fn setup_complete_inner( } for file_path in persist_paths { if let Err(err) = ensure_parent_dir(file_path).await { - return ( - StatusCode::INTERNAL_SERVER_ERROR, - axum::Json( - json!({ "error": format!("Failed to create parent directory for {}: {err}", file_path.display()) }), - ), - ) - .into_response(); + return setup_io_error_response(file_path, "create parent directory", &err); } } @@ -767,11 +791,20 @@ async fn setup_complete_inner( for (_, cleanup_path, _) in pending_writes.iter().take(index + 1) { cleanup_temp_file(cleanup_path).await; } - return ( - StatusCode::INTERNAL_SERVER_ERROR, - axum::Json(json!({ "error": format!("Failed to write {}: {err}", target_path.display()) })), - ) - .into_response(); + return match err { + SetupPersistWriteError::Io(io_err) => setup_io_error_response(target_path, "write file", &io_err), + SetupPersistWriteError::Serialize(serialize_err) => { + error!( + "Setup mode: failed to serialize setup content for '{}': {serialize_err}", + target_path.display() + ); + ( + StatusCode::INTERNAL_SERVER_ERROR, + axum::Json(json!({ "error": "Internal server error" })), + ) + .into_response() + } + }; } } @@ -787,11 +820,7 @@ async fn setup_complete_inner( for (_, temp_path) in &file_pairs { cleanup_temp_file(temp_path).await; } - return ( - StatusCode::INTERNAL_SERVER_ERROR, - axum::Json(json!({ "error": format!("Failed to read current {}: {err}", target.display()) })), - ) - .into_response(); + return setup_io_error_response(target, "read current file", &err); } }; snapshots.push(snapshot); @@ -814,11 +843,7 @@ async fn setup_complete_inner( for (_, pending_temp) in file_pairs.iter().skip(index) { cleanup_temp_file(pending_temp).await; } - return ( - StatusCode::INTERNAL_SERVER_ERROR, - axum::Json(json!({ "error": format!("Failed to write {}: {err}", target.display()) })), - ) - .into_response(); + return setup_io_error_response(target, "replace file", &err); } committed_indices.push(index); } @@ -921,8 +946,14 @@ pub async fn start_setup_server(paths: &ConfigPaths, missing_files: &[String]) - #[cfg(test)] mod tests { - use super::api_proxy_or_default; + use super::{api_proxy_or_default, setup_complete_inner, SetupCompleteRequestDto, SetupModeState, SetupWebUserCredentialDto}; + use axum::{body::to_bytes, http::StatusCode}; use shared::model::{ApiProxyConfigDto, AppConfigDto, TargetUserDto}; + use std::{ + path::PathBuf, + sync::{atomic::AtomicBool, Arc}, + }; + use tokio::sync::{oneshot, Mutex, RwLock}; #[test] fn setup_api_proxy_or_default_preserves_existing_credentials() { @@ -953,4 +984,70 @@ mod tests { assert_eq!(api_proxy.server.len(), 1); assert!(!api_proxy.server[0].host.trim().is_empty()); } + + #[cfg(unix)] + #[tokio::test] + async fn setup_complete_reports_permission_denied_for_unwritable_target_dir() + -> Result<(), Box> { + use std::os::unix::fs::PermissionsExt; + struct PermissionResetGuard(PathBuf); + impl Drop for PermissionResetGuard { + fn drop(&mut self) { + if let Ok(metadata) = std::fs::metadata(&self.0) { + let mut permissions = metadata.permissions(); + permissions.set_mode(0o755); + let _ = std::fs::set_permissions(&self.0, permissions); + } + } + } + // Safety: `geteuid` is a pure libc query with no preconditions. + if unsafe { libc::geteuid() } == 0 { + return Ok(()); + } + + let temp_dir = tempfile::tempdir()?; + let readonly_dir = temp_dir.path().join("readonly"); + tokio::fs::create_dir_all(&readonly_dir).await?; + + let mut readonly_permissions = tokio::fs::metadata(&readonly_dir).await?.permissions(); + readonly_permissions.set_mode(0o555); + tokio::fs::set_permissions(&readonly_dir, readonly_permissions).await?; + let _permission_reset_guard = PermissionResetGuard(readonly_dir.clone()); + + let (shutdown_tx, _shutdown_rx) = oneshot::channel::<()>(); + let state = Arc::new(SetupModeState { + draft: Arc::new(RwLock::new(AppConfigDto::default())), + home_path: temp_dir.path().to_string_lossy().to_string(), + output_dir: readonly_dir.clone(), + config_file_path: readonly_dir.join("config.yml"), + source_file_path: readonly_dir.join("source.yml"), + api_proxy_file_path: readonly_dir.join("api-proxy.yml"), + user_file_path: readonly_dir.join("user.txt"), + web_dir: PathBuf::from(temp_dir.path()), + missing_files: Arc::new(vec![]), + completed: AtomicBool::new(false), + shutdown_tx: Mutex::new(Some(shutdown_tx)), + }); + + let req = SetupCompleteRequestDto { + app_config: AppConfigDto::default(), + web_users: vec![SetupWebUserCredentialDto { + username: "admin".to_string(), + password: "password123".to_string(), + }], + }; + let response = setup_complete_inner(state, req).await; + assert_eq!(response.status(), StatusCode::FORBIDDEN); + + let body = to_bytes(response.into_body(), usize::MAX).await?; + let payload: serde_json::Value = serde_json::from_slice(&body)?; + let error_message = payload + .get("error") + .and_then(serde_json::Value::as_str) + .ok_or_else(|| std::io::Error::other("setup response is missing error field"))?; + + assert_eq!(error_message, "Permission denied while accessing requested resource"); + + Ok(()) + } } diff --git a/backend/src/model/config/metadata_update.rs b/backend/src/model/config/metadata_update.rs index cc8de48aa..a7e053c22 100644 --- a/backend/src/model/config/metadata_update.rs +++ b/backend/src/model/config/metadata_update.rs @@ -18,6 +18,7 @@ pub struct MetadataUpdateConfig { pub worker_idle_timeout_secs: u64, pub max_queue_size: usize, pub no_change_cache_ttl_secs: u64, + pub probe_fairness_resolve_burst: usize, } #[derive(Debug, Clone)] @@ -149,6 +150,7 @@ impl From<&MetadataUpdateConfigDto> for MetadataUpdateConfig { worker_idle_timeout: normalized.worker_idle_timeout, max_queue_size: normalized.max_queue_size.max(1), no_change_cache_ttl_secs: normalized.no_change_cache_ttl_secs.max(1), + probe_fairness_resolve_burst: normalized.probe_fairness_resolve_burst.max(1), } } } @@ -166,6 +168,7 @@ impl From<&MetadataUpdateConfig> for MetadataUpdateConfigDto { worker_idle_timeout: instance.worker_idle_timeout.clone(), max_queue_size: instance.max_queue_size, no_change_cache_ttl_secs: instance.no_change_cache_ttl_secs, + probe_fairness_resolve_burst: instance.probe_fairness_resolve_burst, } } } diff --git a/backend/src/processing/processor/playlist.rs b/backend/src/processing/processor/playlist.rs index 800da6d4b..1503483c4 100644 --- a/backend/src/processing/processor/playlist.rs +++ b/backend/src/processing/processor/playlist.rs @@ -1174,6 +1174,15 @@ async fn playlist_probe(ctx: &PlaylistProcessingContext, target: &ConfigTarget, if needs_live_probe(&item, cutoff_ts) { if let Some(provider_id) = provider_id_from_item(&item) { if queued_live_keys.insert(provider_id.clone()) { + let task = UpdateTask::ProbeLive { + id: provider_id.clone(), + reason: ResolveReasonSet::from_variants(&[ResolveReason::Probe]), + delay: probe_delay, + interval: interval_secs, + }; + if mgr.is_redundant_with_pending_task(input_name.as_ref(), &task) { + continue; + } if log_enabled!(Level::Debug) { let last_probed = match item.header.additional_properties.as_ref() { Some(StreamProperties::Live(props)) => props.last_probed_timestamp, @@ -1184,12 +1193,6 @@ async fn playlist_probe(ctx: &PlaylistProcessingContext, target: &ConfigTarget, input_name, provider_id, last_probed, cutoff_ts, interval_secs, item.header.title ); } - let task = UpdateTask::ProbeLive { - id: provider_id, - reason: ResolveReasonSet::from_variants(&[ResolveReason::Probe]), - delay: probe_delay, - interval: interval_secs, - }; mgr.queue_task_background(input_name.clone(), task); queued_live_count += 1; } @@ -1238,18 +1241,21 @@ async fn playlist_probe(ctx: &PlaylistProcessingContext, target: &ConfigTarget, continue; } - debug!( - "[Task] Creating ProbeStream task for input {}: scope={}, unique_id={}, item_type={:?}, title=\"{}\"", - input_name, probe_scope, unique_id, item.header.item_type, item.header.title - ); let task = UpdateTask::ProbeStream { - probe_scope, - unique_id, + probe_scope: probe_scope.clone(), + unique_id: unique_id.clone(), url: item.header.url.to_string(), item_type: item.header.item_type, reason: ResolveReasonSet::from_variants(&[ResolveReason::MissingDetails]), delay: opts.probe_delay, }; + if mgr.is_redundant_with_pending_task(input_name.as_ref(), &task) { + continue; + } + debug!( + "[Task] Creating ProbeStream task for input {}: scope={}, unique_id={}, item_type={:?}, title=\"{}\"", + input_name, probe_scope, unique_id, item.header.item_type, item.header.title + ); mgr.queue_task_background(input_name.clone(), task); queued_stream_count += 1; } diff --git a/backend/src/processing/processor/xtream_series.rs b/backend/src/processing/processor/xtream_series.rs index 512b1b74e..7aa13aecd 100644 --- a/backend/src/processing/processor/xtream_series.rs +++ b/backend/src/processing/processor/xtream_series.rs @@ -207,25 +207,27 @@ fn queue_background_series_info( let reasons = check_resolve_reasons(resolve_options, resolve_tmdb_enabled, pli); if !reasons.is_empty() { - if log_enabled!(Level::Debug) { - let has_details = pli.has_details(); - let (has_tmdb, has_date) = match pli.header.additional_properties.as_ref() { - Some(StreamProperties::Series(props)) => { - (props.tmdb.is_some(), props.release_date.is_some()) - } - _ => (false, false), - }; - debug!( - "[Task] Creating ResolveSeries task for input {}: id={}, reasons={}, has_details={}, has_tmdb={}, has_date={}, title=\"{}\"", - input_name_arc, provider_id, reasons, has_details, has_tmdb, has_date, pli.header.title - ); - } let task = UpdateTask::ResolveSeries { - id: provider_id, + id: provider_id.clone(), reason: reasons, delay: resolve_options.resolve_delay, }; - mgr.queue_task_background(input_name_arc.clone(), task); + if !mgr.is_redundant_with_pending_task(input_name_arc.as_ref(), &task) { + if log_enabled!(Level::Debug) { + let has_details = pli.has_details(); + let (has_tmdb, has_date) = match pli.header.additional_properties.as_ref() { + Some(StreamProperties::Series(props)) => { + (props.tmdb.is_some(), props.release_date.is_some()) + } + _ => (false, false), + }; + debug!( + "[Task] Creating ResolveSeries task for input {}: id={}, reasons={}, has_details={}, has_tmdb={}, has_date={}, title=\"{}\"", + input_name_arc, provider_id, reasons, has_details, has_tmdb, has_date, pli.header.title + ); + } + mgr.queue_task_background(input_name_arc.clone(), task); + } } } diff --git a/backend/src/processing/processor/xtream_vod.rs b/backend/src/processing/processor/xtream_vod.rs index e1bbdf13c..526c6e966 100644 --- a/backend/src/processing/processor/xtream_vod.rs +++ b/backend/src/processing/processor/xtream_vod.rs @@ -441,6 +441,12 @@ fn queue_background_vod_info( let reasons = check_resolve_reasons(resolve_options, do_probe, resolve_tmdb_enabled, pli); if !reasons.is_empty() { + let task = + UpdateTask::ResolveVod { id: provider_id.clone(), reason: reasons, delay: resolve_options.resolve_delay }; + if mgr.is_redundant_with_pending_task(input.name.as_ref(), &task) { + continue; + } + if log_enabled!(Level::Debug) { let has_details = pli.has_details(); let (has_tmdb, has_date, has_video, has_audio) = @@ -461,8 +467,6 @@ fn queue_background_vod_info( input.name, provider_id, reasons, has_details, has_tmdb, has_date, has_video, has_audio, pli.header.title ); } - let task = - UpdateTask::ResolveVod { id: provider_id, reason: reasons, delay: resolve_options.resolve_delay }; mgr.queue_task_background(input.name.clone(), task); } } diff --git a/backend/src/repository/playlist_source.rs b/backend/src/repository/playlist_source.rs index 1ec4be6da..34521dd53 100644 --- a/backend/src/repository/playlist_source.rs +++ b/backend/src/repository/playlist_source.rs @@ -485,8 +485,16 @@ impl PlaylistSource for MemoryPlaylistSource { fn update_playlist<'a>(&'a mut self, plg: &'a PlaylistGroup) -> BoxFuture<'a, ()> { Box::pin(async move { let playlist = Arc::make_mut(&mut self.playlist); + let incoming_title = shared::utils::deunicode_string(&plg.title).to_lowercase(); for grp in playlist.iter_mut() { - if grp.id == plg.id { + let existing_title = shared::utils::deunicode_string(&grp.title).to_lowercase(); + if grp.xtream_cluster == plg.xtream_cluster && existing_title == incoming_title { + grp.channels.extend(plg.channels.iter().cloned()); + return; + } + } + for grp in playlist.iter_mut() { + if grp.xtream_cluster == plg.xtream_cluster && grp.id == plg.id { grp.channels.extend(plg.channels.iter().cloned()); return; } @@ -569,3 +577,55 @@ where } } else { None } } + +#[cfg(test)] +mod tests { + use super::{MemoryPlaylistSource, PlaylistGroup, PlaylistItem, PlaylistSource, XtreamCluster}; + use shared::model::PlaylistItemHeader; + use shared::utils::Internable; + + fn make_item(title: &str, group: &str, category_id: u32) -> PlaylistItem { + PlaylistItem { + header: PlaylistItemHeader { + title: title.intern(), + group: group.intern(), + category_id, + xtream_cluster: XtreamCluster::Series, + ..Default::default() + }, + } + } + + #[tokio::test] + async fn update_playlist_prefers_group_title_over_colliding_group_id() { + let first_group = PlaylistGroup { + id: 1, + title: "A-First".intern(), + channels: vec![make_item("first-item", "A-First", 11)], + xtream_cluster: XtreamCluster::Series, + }; + let target_group = PlaylistGroup { + id: 2, + title: "B-Series".intern(), + channels: vec![make_item("target-item", "B-Series", 22)], + xtream_cluster: XtreamCluster::Series, + }; + let mut source = MemoryPlaylistSource::new(vec![first_group, target_group]); + + // Simulates a mapped delta group whose local pipeline id restarted at 1. + let incoming = PlaylistGroup { + id: 1, + title: "B-Series".intern(), + channels: vec![make_item("new-episode", "B-Series", 22)], + xtream_cluster: XtreamCluster::Series, + }; + source.update_playlist(&incoming).await; + + let groups = source.take_groups(); + assert_eq!(groups.len(), 2); + assert_eq!(groups[0].title.as_ref(), "A-First"); + assert_eq!(groups[0].channels.len(), 1); + assert_eq!(groups[1].title.as_ref(), "B-Series"); + assert_eq!(groups[1].channels.len(), 2); + } +} diff --git a/bin/build_local.sh b/bin/build_local.sh old mode 100644 new mode 100755 diff --git a/frontend/public/assets/i18n/en.json b/frontend/public/assets/i18n/en.json index de1c8d54e..d8cfce852 100644 --- a/frontend/public/assets/i18n/en.json +++ b/frontend/public/assets/i18n/en.json @@ -442,9 +442,12 @@ "SANITIZE_SENSITIVE_INFO": "Attempts to mask passwords and tokens in the logging output for security." }, "METADATA_UPDATE_CONFIG": { + "METADATA_PATH": "Directory for metadata cache and retry-state files.\n\nRelative paths are resolved against `storage_dir`.\n\nExamples: `metadata`, `/app/data/metadata`.", "MAX_QUEUE_SIZE": "Maximum number of metadata tasks kept in memory for one input worker.\n\nWhen this limit is reached, additional tasks are rejected to protect system stability and memory usage.", "RETRY_DELAY": "Minimum delay for transient requeue scheduling.\n\nUse this to prevent immediate hot-loop retries when tasks are temporarily deferred.\n\nUse duration units like `2s`, `5s`, `10s`.", - "WORKER_IDLE_TIMEOUT": "How long a metadata worker waits for new tasks while idle before it can stop itself.\n\nA lower value releases resources sooner; a higher value keeps workers warm for bursty workloads.\n\nUse duration units like `30s`, `1m`, `5m`." + "WORKER_IDLE_TIMEOUT": "How long a metadata worker waits for new tasks while idle before it can stop itself.\n\nA lower value releases resources sooner; a higher value keeps workers warm for bursty workloads.\n\nUse duration units like `30s`, `1m`, `5m`.", + "NO_CHANGE_CACHE_TTL_SECS": "How long no-change resolve results are remembered (in seconds).\n\nWhile valid, identical resolve reason sets for the same item are skipped to reduce redundant work.\n\nExample: `3600` (1 hour).", + "PROBE_FAIRNESS_RESOLVE_BURST": "Fairness threshold between resolve and probe work.\n\nAfter this many consecutive resolve-domain tasks, one pending probe-domain task is prioritized to avoid probe starvation.\n\nLower values favor probe responsiveness; higher values favor resolve throughput." }, "OUTPUT_HDHOMERUN_FORM": { "USE_OUTPUT": "Enables or disables output creation specifically for HDHomeRun format." @@ -901,6 +904,7 @@ "METADATA_MAX_RESOLVE_RETRY_BACKOFF": "Max resolve retry backoff", "METADATA_NO_CHANGE_CACHE_TTL_SECS": "No-change cache TTL (seconds)", "METADATA_PATH": "Metadata Path", + "METADATA_PROBE_FAIRNESS_RESOLVE_BURST": "Resolve burst before probe fairness", "METADATA_PROBE_COOLDOWN": "Probe cooldown", "METADATA_PROBE_RETRY_BACKOFF_STEP_1": "Probe backoff step 1", "METADATA_PROBE_RETRY_BACKOFF_STEP_2": "Probe backoff step 2", @@ -1248,6 +1252,7 @@ "LIBRARY": "Configure library settings.", "LOG": "Configure logging settings.", "MAIN": "Configure main application settings.", + "METADATA_UPDATE": "Configure metadata update and queue behavior.", "MESSAGING": "Configure messaging settings.", "PROXY": "Configure outgoing proxy settings.", "REVERSE_PROXY": "Configure reverse proxy settings.", @@ -1269,6 +1274,7 @@ "LIBRARY": "Library", "LOG": "Log", "MAIN": "Main", + "METADATA_UPDATE": "Metadata Update", "MESSAGING": "Messaging", "NEXT": "Next", "PROXY": "Proxy", diff --git a/frontend/src/app/components/config/metadata_update_config_view.rs b/frontend/src/app/components/config/metadata_update_config_view.rs index 377e0827c..742ee979a 100644 --- a/frontend/src/app/components/config/metadata_update_config_view.rs +++ b/frontend/src/app/components/config/metadata_update_config_view.rs @@ -32,6 +32,7 @@ const LABEL_RETRY_DELAY: &str = "LABEL.METADATA_RETRY_DELAY"; const LABEL_PROBE_RETRY_LOAD_RETRY_DELAY: &str = "LABEL.METADATA_PROBE_RETRY_LOAD_RETRY_DELAY"; const LABEL_WORKER_IDLE_TIMEOUT: &str = "LABEL.METADATA_WORKER_IDLE_TIMEOUT"; const LABEL_NO_CHANGE_CACHE_TTL_SECS: &str = "LABEL.METADATA_NO_CHANGE_CACHE_TTL_SECS"; +const LABEL_PROBE_FAIRNESS_RESOLVE_BURST: &str = "LABEL.METADATA_PROBE_FAIRNESS_RESOLVE_BURST"; const LABEL_PROBE_RETRY_BACKOFF_STEP_1: &str = "LABEL.METADATA_PROBE_RETRY_BACKOFF_STEP_1"; const LABEL_PROBE_RETRY_BACKOFF_STEP_2: &str = "LABEL.METADATA_PROBE_RETRY_BACKOFF_STEP_2"; const LABEL_PROBE_RETRY_BACKOFF_STEP_3: &str = "LABEL.METADATA_PROBE_RETRY_BACKOFF_STEP_3"; @@ -69,6 +70,7 @@ generate_form_reducer!( MaxQueueSize => max_queue_size: usize, WorkerIdleTimeout => worker_idle_timeout: String, NoChangeCacheTtlSecs => no_change_cache_ttl_secs: u64, + ProbeFairnessResolveBurst => probe_fairness_resolve_burst: usize, } ); @@ -268,6 +270,7 @@ pub fn MetadataUpdateConfigView() -> Html { { config_field!(form_state.form, translate.t(LABEL_MAX_QUEUE_SIZE), max_queue_size) } { config_field!(form_state.form, translate.t(LABEL_WORKER_IDLE_TIMEOUT), worker_idle_timeout) } { config_field!(form_state.form, translate.t(LABEL_NO_CHANGE_CACHE_TTL_SECS), no_change_cache_ttl_secs) } + { config_field!(form_state.form, translate.t(LABEL_PROBE_FAIRNESS_RESOLVE_BURST), probe_fairness_resolve_burst) } @@ -369,6 +372,7 @@ pub fn MetadataUpdateConfigView() -> Html { { edit_field_number_usize!(form_state, translate.t(LABEL_MAX_QUEUE_SIZE), max_queue_size, MetadataUpdateConfigFormAction::MaxQueueSize) } { edit_field_text!(form_state, translate.t(LABEL_WORKER_IDLE_TIMEOUT), worker_idle_timeout, MetadataUpdateConfigFormAction::WorkerIdleTimeout) } { edit_field_number_u64!(form_state, translate.t(LABEL_NO_CHANGE_CACHE_TTL_SECS), no_change_cache_ttl_secs, MetadataUpdateConfigFormAction::NoChangeCacheTtlSecs) } + { edit_field_number_usize!(form_state, translate.t(LABEL_PROBE_FAIRNESS_RESOLVE_BURST), probe_fairness_resolve_burst, MetadataUpdateConfigFormAction::ProbeFairnessResolveBurst) } diff --git a/frontend/src/app/components/setup/setup_context.rs b/frontend/src/app/components/setup/setup_context.rs index 0c6c42ccf..a4732ad68 100644 --- a/frontend/src/app/components/setup/setup_context.rs +++ b/frontend/src/app/components/setup/setup_context.rs @@ -15,6 +15,7 @@ pub enum SetupStep { Proxy, IpCheck, Video, + MetadataUpdate, HdHomerun, Library, Sources, @@ -25,7 +26,7 @@ pub enum SetupStep { impl SetupStep { #[cfg(test)] - pub const ALL_VARIANTS: [Self; 16] = [ + pub const ALL_VARIANTS: [Self; 17] = [ Self::Welcome, Self::Api, Self::WebUi, @@ -36,6 +37,7 @@ impl SetupStep { Self::Proxy, Self::IpCheck, Self::Video, + Self::MetadataUpdate, Self::HdHomerun, Self::Library, Self::Sources, @@ -44,7 +46,7 @@ impl SetupStep { Self::Finish, ]; - pub const ORDER: [Self; 16] = [ + pub const ORDER: [Self; 17] = [ Self::Welcome, Self::Api, Self::WebUi, @@ -55,6 +57,7 @@ impl SetupStep { Self::Proxy, Self::IpCheck, Self::Video, + Self::MetadataUpdate, Self::HdHomerun, Self::Library, Self::Sources, @@ -103,6 +106,7 @@ impl SetupStep { Self::Proxy => "SETUP.LABEL.PROXY", Self::IpCheck => "SETUP.LABEL.IP_CHECK", Self::Video => "SETUP.LABEL.VIDEO", + Self::MetadataUpdate => "SETUP.LABEL.METADATA_UPDATE", Self::HdHomerun => "SETUP.LABEL.HDHOMERUN", Self::Library => "SETUP.LABEL.LIBRARY", Self::Sources => "SETUP.LABEL.SOURCES", @@ -124,6 +128,7 @@ impl SetupStep { Self::Proxy => "SETUP.DESC.PROXY", Self::IpCheck => "SETUP.DESC.IP_CHECK", Self::Video => "SETUP.DESC.VIDEO", + Self::MetadataUpdate => "SETUP.DESC.METADATA_UPDATE", Self::HdHomerun => "SETUP.DESC.HDHOMERUN", Self::Library => "SETUP.DESC.LIBRARY", Self::Sources => "SETUP.DESC.SOURCES", @@ -144,6 +149,7 @@ impl SetupStep { Self::Proxy => Some(ConfigPage::Proxy), Self::IpCheck => Some(ConfigPage::IpCheck), Self::Video => Some(ConfigPage::Video), + Self::MetadataUpdate => Some(ConfigPage::MetadataUpdate), Self::HdHomerun => Some(ConfigPage::HdHomerun), Self::Library => Some(ConfigPage::Library), Self::Schedules => Some(ConfigPage::Schedules), @@ -165,6 +171,7 @@ impl fmt::Display for SetupStep { SetupStep::Proxy => "proxy", SetupStep::IpCheck => "ipcheck", SetupStep::Video => "video", + SetupStep::MetadataUpdate => "metadata_update", SetupStep::HdHomerun => "hdhomerun", SetupStep::Library => "library", SetupStep::Sources => "sources", diff --git a/frontend/src/app/components/setup/step_config.rs b/frontend/src/app/components/setup/step_config.rs index 67aabfa24..025f21390 100644 --- a/frontend/src/app/components/setup/step_config.rs +++ b/frontend/src/app/components/setup/step_config.rs @@ -7,8 +7,8 @@ use crate::{ components::{ config::{ ApiConfigView, ConfigForm, ConfigViewContext, HdHomerunConfigView, IpCheckConfigView, - LibraryConfigView, LogConfigView, MainConfigView, MessagingConfigView, ProxyConfigView, - ReverseProxyConfigView, SchedulesConfigView, VideoConfigView, WebUiConfigView, + LibraryConfigView, LogConfigView, MainConfigView, MessagingConfigView, MetadataUpdateConfigView, + ProxyConfigView, ReverseProxyConfigView, SchedulesConfigView, VideoConfigView, WebUiConfigView, }, Card, SetupContext, SetupStep, TextButton, }, @@ -36,6 +36,7 @@ fn render_config_page(step: SetupStep) -> Html { SetupStep::Proxy => html! { }, SetupStep::IpCheck => html! { }, SetupStep::Video => html! { }, + SetupStep::MetadataUpdate => html! { }, SetupStep::HdHomerun => html! { }, SetupStep::Library => html! { }, SetupStep::Schedules => html! { }, diff --git a/frontend/src/error.rs b/frontend/src/error.rs index fef5cb1a5..d885a7780 100644 --- a/frontend/src/error.rs +++ b/frontend/src/error.rs @@ -27,8 +27,8 @@ pub enum Error { Unauthorized, /// 403 - #[error("Forbidden")] - Forbidden, + #[error("{0}")] + Forbidden(String), /// 404 #[error("Not Found")] @@ -47,8 +47,8 @@ pub enum Error { UnprocessableEntity(ErrorSetInfo), /// 500 - #[error("Internal Server Error")] - InternalServerError, + #[error("{0}")] + InternalServerError(String), /// serde deserialize error #[error("Deserialize Error")] diff --git a/frontend/src/services/config_service.rs b/frontend/src/services/config_service.rs index 1b6626fe9..d28676a54 100644 --- a/frontend/src/services/config_service.rs +++ b/frontend/src/services/config_service.rs @@ -2,7 +2,8 @@ use crate::{ error::Error, model::WebConfig, services::{ - get_base_href, request_get, request_get_meta, request_post, request_post_meta, request_put_meta, EventService, + get_base_href, request_get, request_get_meta, request_post, request_post_meta, request_put_meta, Encoding, + EventService, }, }; use futures_signals::signal::{Mutable, SignalExt}; @@ -277,7 +278,7 @@ impl ConfigService { pub async fn get_batch_input_content(&self, input: &ConfigInputDto) -> Option { let id = input.id.to_string(); let path = concat_path(&self.batch_input_content_path, &id); - request_get::(&path, None, Some("text/plain".to_owned())).await.unwrap_or_else(|err| { + request_get::(&path, None, Some(Encoding::Text)).await.unwrap_or_else(|err| { error!("{err}"); None }) diff --git a/frontend/src/services/playlist_service.rs b/frontend/src/services/playlist_service.rs index facf13511..5171bb3b3 100644 --- a/frontend/src/services/playlist_service.rs +++ b/frontend/src/services/playlist_service.rs @@ -1,4 +1,4 @@ -use crate::services::{get_base_href, request_post}; +use crate::services::{get_base_href, request_post, Encoding}; use futures::join; use indexmap::IndexMap; use log::error; @@ -7,7 +7,7 @@ use shared::{ EpgChannel, EpgTv, PlaylistEpgRequest, PlaylistRequest, SeriesStreamProperties, UiPlaylistCategories, UiPlaylistGroup, UiPlaylistItem, WebplayerUrlRequest, XtreamCluster, XtreamSeriesInfoDoc, }, - utils::{concat_path_leading_slash, ACCEPT_PREFER_CBOR}, + utils::concat_path_leading_slash, }; use std::rc::Rc; @@ -54,19 +54,19 @@ impl PlaylistService { &self.playlist_api_live_path, playlist_request, None, - Some(ACCEPT_PREFER_CBOR.to_string()) + Some(Encoding::Cbor), ), request_post::<&PlaylistRequest, Vec>( &self.playlist_api_vod_path, playlist_request, None, - Some(ACCEPT_PREFER_CBOR.to_string()) + Some(Encoding::Cbor), ), request_post::<&PlaylistRequest, Vec>( &self.playlist_api_series_path, playlist_request, None, - Some(ACCEPT_PREFER_CBOR.to_string()) + Some(Encoding::Cbor), ), ); @@ -109,7 +109,7 @@ impl PlaylistService { &self.playlist_api_webplayer_url_path, &request, None, - Some("text/plain".to_string()), + Some(Encoding::Text), ) .await .unwrap_or_else(|err| { @@ -123,7 +123,7 @@ impl PlaylistService { &self.playlist_api_epg_path, &request, None, - Some(ACCEPT_PREFER_CBOR.to_string()), + Some(Encoding::Cbor), ) .await { @@ -141,20 +141,15 @@ impl PlaylistService { playlist_request: &PlaylistRequest, ) -> Option { let path = format!("{}/{}/{}", self.playlist_api_series_info_path, pli.virtual_id, pli.provider_id); - request_post::<&PlaylistRequest, XtreamSeriesInfoDoc>( - &path, - playlist_request, - None, - Some(ACCEPT_PREFER_CBOR.to_string()), - ) - .await - .map_or_else( - |err| { - error!("{err}"); - None - }, - |response| response.as_ref().map(|doc| SeriesStreamProperties::from_info_doc(doc, pli.virtual_id)), - ) + request_post::<&PlaylistRequest, XtreamSeriesInfoDoc>(&path, playlist_request, None, Some(Encoding::Cbor)) + .await + .map_or_else( + |err| { + error!("{err}"); + None + }, + |response| response.as_ref().map(|doc| SeriesStreamProperties::from_info_doc(doc, pli.virtual_id)), + ) } pub async fn get_episode(&self, virtual_id: u32, playlist_request: &PlaylistRequest) -> Option { diff --git a/frontend/src/services/requests.rs b/frontend/src/services/requests.rs index ff06884cc..464edf3be 100644 --- a/frontend/src/services/requests.rs +++ b/frontend/src/services/requests.rs @@ -1,11 +1,11 @@ use crate::error::{Error, ErrorInfo, ErrorSetInfo}; use gloo_storage::{LocalStorage, Storage}; -use log::error; -use reqwasm::http::Request; +use log::{error, warn}; +use reqwasm::http::{Request, Response}; use serde::{de::DeserializeOwned, Serialize}; use serde_json::Value; -use shared::utils::{bin_deserialize, CONTENT_TYPE_CBOR, CONTENT_TYPE_JSON}; -use std::collections::HashMap; +use shared::utils::{bin_deserialize, bin_serialize, CONTENT_TYPE_CBOR, CONTENT_TYPE_JSON}; +use std::{collections::HashMap, sync::OnceLock}; use web_sys::window; enum RequestMethod { @@ -22,12 +22,118 @@ pub struct ResponseMeta { pub headers: HashMap, } +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum Encoding { + Json, + Cbor, + Text, +} + +enum EncodedBody { + Text(String), + Binary(Vec), +} + +impl Encoding { + const fn as_content_type(self) -> &'static str { + match self { + Self::Json => CONTENT_TYPE_JSON, + Self::Cbor => CONTENT_TYPE_CBOR, + Self::Text => "text/plain", + } + } + + fn encode(self, body: &B) -> Result { + match self { + Self::Cbor => { + let bytes = bin_serialize(body).map_err(|_| Error::RequestError)?; + Ok(EncodedBody::Binary(bytes)) + } + Self::Json => { + let text = serde_json::to_string(body).map_err(|_| Error::RequestError)?; + Ok(EncodedBody::Text(text)) + } + Self::Text => { + let value = serde_json::to_value(body).map_err(|_| Error::RequestError)?; + match value { + Value::String(text) => Ok(EncodedBody::Text(text)), + _ => Err(Error::RequestError), + } + } + } + } +} + +fn apply_encoded_body(request: Request, body: EncodedBody, content_type: &'static str) -> Request { + match body { + EncodedBody::Text(payload) => request.body(payload).header("Content-Type", content_type), + EncodedBody::Binary(payload) => request.body(payload).header("Content-Type", content_type), + } +} + +fn encoding_from_content_type(content_type: &str) -> Option { + if content_type.contains(CONTENT_TYPE_CBOR) { + Some(Encoding::Cbor) + } else if content_type.contains(CONTENT_TYPE_JSON) { + Some(Encoding::Json) + } else if content_type.contains("text/") { + Some(Encoding::Text) + } else { + None + } +} + +async fn decode_response_body(response: Response, encoding: Encoding) -> Result +where + T: DeserializeOwned + 'static + std::fmt::Debug, +{ + match encoding { + Encoding::Cbor => { + let bytes = response.binary().await.map_err(|err| { + error!("Failed to read response body for CBOR decode: {err}"); + Error::DeserializeError + })?; + bin_deserialize::(&bytes).map_err(|err| { + error!("Failed to deserialize {err}"); + Error::DeserializeError + }) + } + Encoding::Json => response.json::().await.map_err(|_| Error::DeserializeError), + Encoding::Text => match response.text().await { + Ok(content) => serde_json::from_value::(Value::String(content)).map_err(|_| Error::DeserializeError), + Err(_) => Err(Error::RequestError), + }, + } +} + +async fn extract_error_message_checked(response: Response) -> Result { + let content_type = response.headers().get("content-type").unwrap_or_default().to_ascii_lowercase(); + match encoding_from_content_type(&content_type) { + Some(Encoding::Cbor) => { + let bytes = response.binary().await.map_err(|_| Error::DeserializeError)?; + let data = bin_deserialize::(&bytes).map_err(|_| Error::DeserializeError)?; + Ok(data.error) + } + Some(Encoding::Json) => { + let data = response.json::().await.map_err(|_| Error::DeserializeError)?; + Ok(data.error) + } + _ => Ok(response.text().await.unwrap_or_default()), + } +} + +async fn extract_error_message(response: Response) -> String { + extract_error_message_checked(response).await.unwrap_or_default() +} + const TOKEN_KEY: &str = "tuliprox.token"; pub fn get_token() -> Option { LocalStorage::get(TOKEN_KEY).ok() } pub fn set_token(token: Option<&str>) { if let Some(t) = token { - LocalStorage::set(TOKEN_KEY, String::from(t)).expect("failed to set"); + if let Err(err) = LocalStorage::set(TOKEN_KEY, String::from(t)) { + warn!("failed to set token in localStorage: {err:?}"); + } } else { LocalStorage::delete(TOKEN_KEY); } @@ -48,8 +154,8 @@ async fn request( method: RequestMethod, url: &str, body: B, - content_type: Option, - response_type: Option, + content_type: Option, + response_type: Option, request_headers: Option<&[(String, String)]>, response_header_keys: Option<&[&str]>, ) -> Result, Error> @@ -57,17 +163,17 @@ where T: DeserializeOwned + 'static + std::fmt::Debug, B: Serialize + std::fmt::Debug, { - let c_type = content_type.as_ref().map_or(CONTENT_TYPE_JSON, |c| c.as_str()); - let r_type = response_type.as_ref().map_or(CONTENT_TYPE_JSON, |c| c.as_str()); + let c_type = content_type.unwrap_or(Encoding::Json); + let r_type = response_type.unwrap_or(Encoding::Json); let mut request = match method { RequestMethod::Get => Request::get(url), RequestMethod::Post => { - let json = serde_json::to_string(&body).map_err(|_| Error::RequestError)?; - Request::post(url).body(json).header("Content-Type", c_type) + let encoded = c_type.encode(&body)?; + apply_encoded_body(Request::post(url), encoded, c_type.as_content_type()) } RequestMethod::Put => { - let json = serde_json::to_string(&body).map_err(|_| Error::RequestError)?; - Request::put(url).body(json).header("Content-Type", c_type) + let encoded = c_type.encode(&body)?; + apply_encoded_body(Request::put(url), encoded, c_type.as_content_type()) } // RequestMethod::PATCH => Request::patch(&url).body(serde_json::to_string(&body).unwrap()), RequestMethod::Delete => Request::delete(url), @@ -81,9 +187,7 @@ where } } - if r_type.contains(CONTENT_TYPE_CBOR) { - request = request.header("Accept", CONTENT_TYPE_CBOR); - } + request = request.header("Accept", r_type.as_content_type()); match request.send().await { Ok(response) => { @@ -98,106 +202,59 @@ where } match status { 200 | 205 | 206 => { - let content_type = response.headers().get("content-type").unwrap_or(r_type.to_string()); - let is_json = content_type.contains(CONTENT_TYPE_JSON); - let is_bin = !is_json && content_type.contains(CONTENT_TYPE_CBOR); - if (is_json || is_bin) && std::any::TypeId::of::() == std::any::TypeId::of::<()>() { + let content_type = response.headers().get("content-type").unwrap_or_default().to_ascii_lowercase(); + if std::any::TypeId::of::() == std::any::TypeId::of::<()>() { // `T = ()` valid let _ = response.binary().await; return Ok(ResponseMeta { body: None, headers: response_headers }); } - - if is_bin { - match response.binary().await { - Ok(bytes) => match bin_deserialize::(&bytes) { - Ok(data) => Ok(ResponseMeta { body: Some(data), headers: response_headers }), - Err(err) => { - error!("Failed to deserialize {err}"); - Err(Error::DeserializeError) - } - }, - Err(err) => { - error!("Failed to deserialize {err}"); - Err(Error::DeserializeError) - } - } - } else if is_json { - let data: Result = response.json::().await; - if let Ok(data) = data { - Ok(ResponseMeta { body: Some(data), headers: response_headers }) - } else { - Err(Error::DeserializeError) - } - } else { - match response.text().await { - Ok(content) => match serde_json::from_value::(Value::String(content)) { - Ok(parsed) => Ok(ResponseMeta { body: Some(parsed), headers: response_headers }), - Err(_err) => Err(Error::DeserializeError), - }, - Err(_err) => Err(Error::RequestError), - } - } + let decode_encoding = encoding_from_content_type(&content_type).unwrap_or(r_type); + let decoded = decode_response_body::(response, decode_encoding).await?; + Ok(ResponseMeta { body: Some(decoded), headers: response_headers }) } 201 | 202 | 204 => Ok(ResponseMeta { body: None, headers: response_headers }), 400 => { - let ct = response.headers().get("content-type").unwrap_or_default(); - let is_json = ct.contains(CONTENT_TYPE_JSON); - let is_bin = !is_json && ct.contains(CONTENT_TYPE_CBOR); - let data: Result = if is_bin { - match response.binary().await { - Ok(bytes) => bin_deserialize::(&bytes).map_err(|_| Error::DeserializeError), - Err(_) => Err(Error::DeserializeError), - } - } else { - response.json::().await.map_err(|_| Error::DeserializeError) - }; - - if let Ok(data) = data { - Err(Error::BadRequest(data.error)) - } else { + let message = extract_error_message(response).await; + if message.trim().is_empty() { Err(Error::BadRequest("400".to_string())) + } else { + Err(Error::BadRequest(message)) } } 401 => Err(Error::Unauthorized), - 403 => Err(Error::Forbidden), + 403 => { + let message = extract_error_message(response).await; + if message.trim().is_empty() { + Err(Error::Forbidden("Forbidden".to_string())) + } else { + Err(Error::Forbidden(message)) + } + } 404 => Err(Error::NotFound), 409 => { - let ct = response.headers().get("content-type").unwrap_or_default(); - let is_json = ct.contains(CONTENT_TYPE_JSON); - let is_bin = !is_json && ct.contains(CONTENT_TYPE_CBOR); - let data: Result = if is_bin { - match response.binary().await { - Ok(bytes) => bin_deserialize::(&bytes).map_err(|_| Error::DeserializeError), - Err(_) => Err(Error::DeserializeError), - } - } else { - response.json::().await.map_err(|_| Error::DeserializeError) - }; - if let Ok(data) = data { - Err(Error::Conflict(data.error)) - } else { + let message = extract_error_message(response).await; + if message.trim().is_empty() { Err(Error::Conflict("Configuration conflict (409)".to_string())) + } else { + Err(Error::Conflict(message)) } } 428 => { - let ct = response.headers().get("content-type").unwrap_or_default(); - let is_json = ct.contains(CONTENT_TYPE_JSON); - let is_bin = !is_json && ct.contains(CONTENT_TYPE_CBOR); - let data: Result = if is_bin { - match response.binary().await { - Ok(bytes) => bin_deserialize::(&bytes).map_err(|_| Error::DeserializeError), - Err(_) => Err(Error::DeserializeError), - } - } else { - response.json::().await.map_err(|_| Error::DeserializeError) - }; - if let Ok(data) = data { - Err(Error::PreconditionRequired(data.error)) - } else { + let message = extract_error_message(response).await; + if message.trim().is_empty() { Err(Error::PreconditionRequired("Missing precondition header (428)".to_string())) + } else { + Err(Error::PreconditionRequired(message)) + } + } + 500 => { + let message = extract_error_message(response).await; + if message.trim().is_empty() { + Err(Error::InternalServerError("Internal Server Error".to_string())) + } else { + Err(Error::InternalServerError(message)) } } - 500 => Err(Error::InternalServerError), 422 => { let ct = response.headers().get("content-type").unwrap_or_default(); let is_json = ct.contains(CONTENT_TYPE_JSON); @@ -230,8 +287,8 @@ where /// Delete request pub async fn request_delete( url: &str, - content_type: Option, - response_type: Option, + content_type: Option, + response_type: Option, ) -> Result, Error> where T: DeserializeOwned + 'static + std::fmt::Debug, @@ -242,8 +299,8 @@ where /// Get request pub async fn request_get( url: &str, - content_type: Option, - response_type: Option, + content_type: Option, + response_type: Option, ) -> Result, Error> where T: DeserializeOwned + 'static + std::fmt::Debug, @@ -253,8 +310,8 @@ where pub async fn request_get_meta( url: &str, - content_type: Option, - response_type: Option, + content_type: Option, + response_type: Option, response_header_keys: &[&str], ) -> Result, Error> where @@ -274,8 +331,8 @@ where pub async fn request_post( url: &str, body: B, - content_type: Option, - response_type: Option, + content_type: Option, + response_type: Option, ) -> Result, Error> where T: DeserializeOwned + 'static + std::fmt::Debug, @@ -287,8 +344,8 @@ where pub async fn request_post_meta( url: &str, body: B, - content_type: Option, - response_type: Option, + content_type: Option, + response_type: Option, request_headers: Option<&[(String, String)]>, response_header_keys: &[&str], ) -> Result, Error> @@ -304,8 +361,8 @@ where pub async fn request_put( url: &str, body: B, - content_type: Option, - response_type: Option, + content_type: Option, + response_type: Option, ) -> Result, Error> where T: DeserializeOwned + 'static + std::fmt::Debug, @@ -317,8 +374,8 @@ where pub async fn request_put_meta( url: &str, body: B, - content_type: Option, - response_type: Option, + content_type: Option, + response_type: Option, request_headers: Option<&[(String, String)]>, response_header_keys: &[&str], ) -> Result, Error> @@ -336,16 +393,22 @@ pub fn limit(count: u32, p: u32) -> String { format!("limit={count}&offset={offset}") } +static BASE_HREF: OnceLock = OnceLock::new(); + pub fn get_base_href() -> String { - let mut href = window() - .and_then(|w| w.document()) - .and_then(|doc| doc.query_selector("base").ok().flatten()) - .and_then(|base| base.get_attribute("href")) - .map_or_else(|| "/".to_owned(), |s| s.trim().to_owned()); + BASE_HREF + .get_or_init(|| { + let mut href = window() + .and_then(|w| w.document()) + .and_then(|doc| doc.query_selector("base").ok().flatten()) + .and_then(|base| base.get_attribute("href")) + .map_or_else(|| "/".to_owned(), |s| s.trim().to_owned()); - if !href.ends_with('/') { - href.push('/'); - } + if !href.ends_with('/') { + href.push('/'); + } - href + href + }) + .clone() } diff --git a/shared/src/model/config/metadata_update.rs b/shared/src/model/config/metadata_update.rs index 1f57980e9..747d1c3a6 100644 --- a/shared/src/model/config/metadata_update.rs +++ b/shared/src/model/config/metadata_update.rs @@ -7,19 +7,20 @@ use crate::{ default_metadata_ffprobe_probe_size, default_metadata_max_attempts_probe, default_metadata_max_attempts_resolve, default_metadata_max_queue_size, default_metadata_max_resolve_retry_backoff, default_metadata_no_change_cache_ttl_secs, default_metadata_path, - default_metadata_probe_cooldown, default_metadata_probe_retry_backoff_step_1, - default_metadata_probe_retry_backoff_step_2, default_metadata_probe_retry_backoff_step_3, - default_metadata_probe_retry_load_retry_delay, default_metadata_progress_log_interval, - default_metadata_queue_log_interval, default_metadata_resolve_exhaustion_reset_gap, - default_metadata_resolve_min_retry_base, default_metadata_retry_delay, default_metadata_tmdb_cooldown, - default_metadata_worker_idle_timeout, default_tmdb_api_key, default_tmdb_cache_duration_days, - default_tmdb_language, default_tmdb_match_threshold, default_tmdb_rate_limit_ms, deserialize_as_string, - is_default_metadata_backoff_jitter_percent, is_default_metadata_ffprobe_analyze_duration, - is_default_metadata_ffprobe_live_analyze_duration, is_default_metadata_ffprobe_live_probe_size, - is_default_metadata_ffprobe_probe_size, is_default_metadata_max_attempts_probe, - is_default_metadata_max_attempts_resolve, is_default_metadata_max_queue_size, - is_default_metadata_max_resolve_retry_backoff, is_default_metadata_no_change_cache_ttl_secs, - is_default_metadata_path, is_default_metadata_probe_cooldown, is_default_metadata_probe_retry_backoff_step_1, + default_metadata_probe_cooldown, default_metadata_probe_fairness_resolve_burst, + default_metadata_probe_retry_backoff_step_1, default_metadata_probe_retry_backoff_step_2, + default_metadata_probe_retry_backoff_step_3, default_metadata_probe_retry_load_retry_delay, + default_metadata_progress_log_interval, default_metadata_queue_log_interval, + default_metadata_resolve_exhaustion_reset_gap, default_metadata_resolve_min_retry_base, + default_metadata_retry_delay, default_metadata_tmdb_cooldown, default_metadata_worker_idle_timeout, + default_tmdb_api_key, default_tmdb_cache_duration_days, default_tmdb_language, default_tmdb_match_threshold, + default_tmdb_rate_limit_ms, deserialize_as_string, is_default_metadata_backoff_jitter_percent, + is_default_metadata_ffprobe_analyze_duration, is_default_metadata_ffprobe_live_analyze_duration, + is_default_metadata_ffprobe_live_probe_size, is_default_metadata_ffprobe_probe_size, + is_default_metadata_max_attempts_probe, is_default_metadata_max_attempts_resolve, + is_default_metadata_max_queue_size, is_default_metadata_max_resolve_retry_backoff, + is_default_metadata_no_change_cache_ttl_secs, is_default_metadata_path, is_default_metadata_probe_cooldown, + is_default_metadata_probe_fairness_resolve_burst, is_default_metadata_probe_retry_backoff_step_1, is_default_metadata_probe_retry_backoff_step_2, is_default_metadata_probe_retry_backoff_step_3, is_default_metadata_probe_retry_load_retry_delay, is_default_metadata_progress_log_interval, is_default_metadata_queue_log_interval, is_default_metadata_resolve_exhaustion_reset_gap, @@ -70,6 +71,11 @@ pub struct MetadataUpdateConfigDto { skip_serializing_if = "is_default_metadata_no_change_cache_ttl_secs" )] pub no_change_cache_ttl_secs: u64, + #[serde( + default = "default_metadata_probe_fairness_resolve_burst", + skip_serializing_if = "is_default_metadata_probe_fairness_resolve_burst" + )] + pub probe_fairness_resolve_burst: usize, } impl Default for MetadataUpdateConfigDto { @@ -85,6 +91,7 @@ impl Default for MetadataUpdateConfigDto { worker_idle_timeout: default_metadata_worker_idle_timeout(), max_queue_size: default_metadata_max_queue_size(), no_change_cache_ttl_secs: default_metadata_no_change_cache_ttl_secs(), + probe_fairness_resolve_burst: default_metadata_probe_fairness_resolve_burst(), } } } @@ -493,6 +500,7 @@ impl MetadataUpdateConfigDto { && self.worker_idle_timeout == default_metadata_worker_idle_timeout() && self.max_queue_size == default_metadata_max_queue_size() && self.no_change_cache_ttl_secs == default_metadata_no_change_cache_ttl_secs() + && self.probe_fairness_resolve_burst == default_metadata_probe_fairness_resolve_burst() } pub fn clean(&mut self) { @@ -522,6 +530,7 @@ impl MetadataUpdateConfigDto { self.max_queue_size = self.max_queue_size.max(MIN_QUEUE_SIZE); self.no_change_cache_ttl_secs = self.no_change_cache_ttl_secs.max(MIN_DURATION_SECS); + self.probe_fairness_resolve_burst = self.probe_fairness_resolve_burst.max(MIN_QUEUE_SIZE); self.clean(); @@ -624,6 +633,7 @@ mod tests { cfg.probe.max_attempts = 0; cfg.max_queue_size = 0; cfg.no_change_cache_ttl_secs = 0; + cfg.probe_fairness_resolve_burst = 0; cfg.ffprobe.timeout = Some(0); cfg.ffprobe.analyze_duration = "0s".to_string(); cfg.ffprobe.probe_size = "0".to_string(); @@ -635,6 +645,7 @@ mod tests { assert_eq!(cfg.probe.max_attempts, 1); assert_eq!(cfg.max_queue_size, 1); assert_eq!(cfg.no_change_cache_ttl_secs, 1); + assert_eq!(cfg.probe_fairness_resolve_burst, 1); assert_eq!(cfg.ffprobe.timeout, Some(1)); assert_eq!(cfg.ffprobe.analyze_duration, "1s"); assert_eq!(cfg.ffprobe.probe_size, "1B"); diff --git a/shared/src/utils/default_utils.rs b/shared/src/utils/default_utils.rs index a2bb0b122..75debc98c 100644 --- a/shared/src/utils/default_utils.rs +++ b/shared/src/utils/default_utils.rs @@ -210,6 +210,10 @@ pub const fn default_metadata_no_change_cache_ttl_secs() -> u64 { 3600 } pub const fn is_default_metadata_no_change_cache_ttl_secs(v: &u64) -> bool { *v == default_metadata_no_change_cache_ttl_secs() } +pub const fn default_metadata_probe_fairness_resolve_burst() -> usize { 200 } +pub const fn is_default_metadata_probe_fairness_resolve_burst(v: &usize) -> bool { + *v == default_metadata_probe_fairness_resolve_burst() +} pub fn default_metadata_ffprobe_analyze_duration() -> String { "10s".to_string() } pub fn is_default_metadata_ffprobe_analyze_duration(v: &String) -> bool { *v == default_metadata_ffprobe_analyze_duration()