diff --git a/backend/src/api/endpoints/hls_api.rs b/backend/src/api/endpoints/hls_api.rs index 0bef4b36b..015d6acb0 100644 --- a/backend/src/api/endpoints/hls_api.rs +++ b/backend/src/api/endpoints/hls_api.rs @@ -407,7 +407,14 @@ async fn hls_api_stream( return axum::http::StatusCode::BAD_REQUEST.into_response(); } - let connection_permission = user.connection_permission(&app_state).await; + let connection_permission = if user.max_connections > 0 && app_state.app_config.config.load().user_access_control { + app_state + .active_users + .connection_permission_for_session(&user.username, user.max_connections, &session.token) + .await + } else { + UserConnectionPermission::Allowed + }; if connection_permission == UserConnectionPermission::Exhausted { return create_custom_video_stream_response( &app_state, diff --git a/backend/src/api/endpoints/v1_api_config.rs b/backend/src/api/endpoints/v1_api_config.rs index 410a57774..4d5f8cd96 100644 --- a/backend/src/api/endpoints/v1_api_config.rs +++ b/backend/src/api/endpoints/v1_api_config.rs @@ -9,6 +9,7 @@ use crate::{ utils::{ persist_messaging_templates, prepare_sources_batch, prepare_users, read_api_proxy_file, request::download_text_content, + xtream::{get_xtream_stream_url_base, xtream_login}, }, }; use axum::{ @@ -20,16 +21,16 @@ use log::error; use serde_json::json; use shared::{ error::TuliproxError, - model::{ApiProxyConfigDto, ConfigDto, SourcesConfigDto}, + model::{ApiProxyConfigDto, ConfigDto, SourcesConfigDto, XtreamLoginRequest}, utils::{ HEADER_CONFIG_API_PROXY_REVISION, HEADER_CONFIG_MAIN_REVISION, HEADER_CONFIG_SOURCES_REVISION, HEADER_IF_MATCH, }, }; +use shared::model::InputFetchMethod; +use std::collections::HashMap; use std::path::Path; use std::sync::Arc; - - fn file_revision_from_bytes(bytes: &[u8]) -> String { blake3::hash(bytes).to_hex().to_string() } async fn read_file_revision(path: &str) -> Result { @@ -422,10 +423,49 @@ async fn config_batch_content( .into_response() } +async fn get_xtream_login_info( + axum::extract::State(app_state): axum::extract::State>, + axum::extract::Json(request): axum::extract::Json, +) -> impl IntoResponse + Send { + if request.url.trim().is_empty() { + return (StatusCode::BAD_REQUEST, axum::Json(json!({"error": "URL is required"}))).into_response(); + } + if request.username.trim().is_empty() { + return (StatusCode::BAD_REQUEST, axum::Json(json!({"error": "Username is required"}))).into_response(); + } + if request.password.trim().is_empty() { + return (StatusCode::BAD_REQUEST, axum::Json(json!({"error": "Password is required"}))).into_response(); + } + + let base_url = get_xtream_stream_url_base(&request.url, &request.username, &request.password); + let input_source = InputSource { + name: "xtream_login".into(), + url: base_url, + provider: None, + username: Some(request.username.clone()), + password: Some(request.password.clone()), + method: InputFetchMethod::GET, + headers: HashMap::new(), + }; + let http_client = app_state.http_client.load(); + match xtream_login(&app_state.app_config, &http_client, &input_source, &request.username).await { + Ok(login_info) => axum::Json(login_info.unwrap_or_default()).into_response(), + Err(err) => { + error!("Failed to get xtream login info: {err}"); + ( + StatusCode::BAD_GATEWAY, + axum::Json(json!({"error": "Failed to get Xtream login info"})), + ) + .into_response() + } + } +} + pub fn v1_api_config_register(router: Router>) -> axum::Router> { router .route("/config", axum::routing::get(config)) .route("/config/batchContent/{input_id}", axum::routing::get(config_batch_content)) + .route("/config/xtream/login-info", axum::routing::post(get_xtream_login_info)) .route("/config/main", axum::routing::post(save_config_main)) .route("/config/sources", axum::routing::post(save_config_sources)) .route("/config/apiproxy", axum::routing::get(get_config_api_proxy_config)) diff --git a/backend/src/api/model/active_user_manager.rs b/backend/src/api/model/active_user_manager.rs index 3fc450407..7405afb53 100644 --- a/backend/src/api/model/active_user_manager.rs +++ b/backend/src/api/model/active_user_manager.rs @@ -322,6 +322,47 @@ impl ActiveUserManager { UserConnectionPermission::Allowed } + pub async fn connection_permission_for_session( + &self, + username: &str, + max_connections: u32, + session_token: &str, + ) -> UserConnectionPermission { + if max_connections == 0 { + return UserConnectionPermission::Allowed; + } + + { + let connections = self.connections.read().await; + let Some(connection_data) = connections.by_key.get(username) else { + return UserConnectionPermission::Allowed; + }; + + if connection_data + .streams + .iter() + .any(|stream| stream.session_token.as_deref() == Some(session_token)) + { + return UserConnectionPermission::Allowed; + } + } + + let mut connections = self.connections.write().await; + let Some(connection_data) = connections.by_key.get_mut(username) else { + return UserConnectionPermission::Allowed; + }; + + if connection_data + .streams + .iter() + .any(|stream| stream.session_token.as_deref() == Some(session_token)) + { + return UserConnectionPermission::Allowed; + } + + self.check_connection_permission(username, connection_data) + } + pub async fn active_users_and_connections(&self) -> (usize, usize) { let user_connections = self.connections.read().await; user_connections @@ -408,13 +449,13 @@ impl ActiveUserManager { .streams .iter_mut() .find(|stream_info| { - stream_info.addr == fingerprint.addr - && match session_token { - Some(token) => stream_info.session_token.as_deref() == Some(token), - None => stream_info.session_token.is_none(), - } + match session_token { + Some(token) => stream_info.session_token.as_deref() == Some(token), + None => stream_info.addr == fingerprint.addr && stream_info.session_token.is_none(), + } }) .map(|stream_info| { + stream_info.addr = fingerprint.addr; stream_info.channel = stream_channel.clone(); stream_info.provider = provider.to_string(); stream_info.user_agent.clone_from(&user_agent_string); @@ -778,6 +819,57 @@ mod tests { assert_eq!(manager.user_connections(username).await, 0); } + #[tokio::test] + async fn test_same_session_token_on_new_addr_reuses_logical_connection() { + let config = Config::default(); + let geoip = Arc::new(ArcSwapOption::::default()); + let event_manager = Arc::new(EventManager::new()); + let manager = ActiveUserManager::new(&config, &geoip, &event_manager); + + let first_addr: SocketAddr = "127.0.0.1:55021".parse().unwrap(); + let second_addr: SocketAddr = "127.0.0.1:55022".parse().unwrap(); + let first = Fingerprint::new("fp-key-1".to_string(), "127.0.0.1".to_string(), first_addr); + let second = Fingerprint::new("fp-key-2".to_string(), "127.0.0.1".to_string(), second_addr); + + manager.add_connection(&first_addr).await; + manager + .update_connection( + "user1", + 1, + &first, + "provider-a", + test_channel(2001), + Cow::Borrowed("ua"), + Some("tok-hls"), + ) + .await; + + assert_eq!( + manager.connection_permission_for_session("user1", 1, "tok-hls").await, + UserConnectionPermission::Allowed + ); + + manager.add_connection(&second_addr).await; + manager + .update_connection( + "user1", + 1, + &second, + "provider-a", + test_channel(2001), + Cow::Borrowed("ua"), + Some("tok-hls"), + ) + .await; + + assert_eq!(manager.user_connections("user1").await, 1); + + let streams = manager.active_streams().await; + assert_eq!(streams.len(), 1); + assert_eq!(streams[0].addr, second_addr); + assert_eq!(streams[0].session_token.as_deref(), Some("tok-hls")); + } + #[tokio::test] async fn stale_anonymous_socket_registration_is_pruned_by_gc() { let config = Config::default(); diff --git a/backend/src/api/model/metadata_update_manager.rs b/backend/src/api/model/metadata_update_manager.rs index d3cb65149..4f370adc8 100644 --- a/backend/src/api/model/metadata_update_manager.rs +++ b/backend/src/api/model/metadata_update_manager.rs @@ -184,11 +184,13 @@ pub enum UpdateTask { id: ProviderIdType, reason: ResolveReasonSet, delay: u16, + source_last_modified: Option, }, ResolveSeries { id: ProviderIdType, reason: ResolveReasonSet, delay: u16, + source_last_modified: Option, }, ProbeLive { id: ProviderIdType, @@ -221,10 +223,10 @@ impl UpdateTask { impl std::fmt::Display for UpdateTask { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { - UpdateTask::ResolveVod { id, reason, delay } => { + UpdateTask::ResolveVod { id, reason, delay, .. } => { write!(f, "Resolve VOD {id} (Reason: {reason}, Delay: {delay}sec)") } - UpdateTask::ResolveSeries { id, reason, delay } => { + UpdateTask::ResolveSeries { id, reason, delay, .. } => { write!(f, "Resolve Series {id} (Reason: {reason}, Delay: {delay}sec)") } UpdateTask::ProbeLive { id, reason, delay, interval } => { @@ -290,10 +292,19 @@ struct RetryState { next_allowed_at_ts: i64, cooldown_until_ts: Option, last_error: Option, + source_last_modified: Option, } impl RetryState { - fn new() -> Self { Self { attempts: 0, next_allowed_at_ts: 0, cooldown_until_ts: None, last_error: None } } + fn new() -> Self { + Self { + attempts: 0, + next_allowed_at_ts: 0, + cooldown_until_ts: None, + last_error: None, + source_last_modified: None, + } + } } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -402,6 +413,7 @@ struct RetryStateDbValue { next_allowed_at_ts: i64, cooldown_until_ts: Option, last_error: Option, + source_last_modified: Option, } impl RetryStateDbValue { @@ -411,11 +423,16 @@ impl RetryStateDbValue { next_allowed_at_ts: state.next_allowed_at_ts, cooldown_until_ts: state.cooldown_until_ts, last_error: state.last_error.clone(), + source_last_modified: state.source_last_modified, } } fn into_retry_state(self) -> Option { - if self.attempts == 0 && self.next_allowed_at_ts <= 0 && self.cooldown_until_ts.is_none() { + if self.attempts == 0 + && self.next_allowed_at_ts <= 0 + && self.cooldown_until_ts.is_none() + && self.source_last_modified.is_none() + { return None; } Some(RetryState { @@ -423,6 +440,7 @@ impl RetryStateDbValue { next_allowed_at_ts: self.next_allowed_at_ts, cooldown_until_ts: self.cooldown_until_ts, last_error: self.last_error, + source_last_modified: self.source_last_modified, }) } } @@ -566,6 +584,8 @@ pub struct MetadataUpdateManager { /// Producer-visible view of active resolve cooldowns so repeated playlist refreshes /// can skip creating tasks that are still suppressed anyway. resolve_enqueue_suppressions: Arc>, + /// Producer-visible TMDB/date suppression keyed by the last seen series `last_modified`. + tmdb_source_markers: Arc>, /// Tracks inputs for which producer-side retry suppression state was already loaded from disk. enqueue_state_loaded_inputs: Arc, ()>>, /// Backoff for failed producer-side retry-state loads to avoid repeated blocking disk reads. @@ -593,6 +613,7 @@ impl MetadataUpdateManager { cancel_token: ArcSwap::from_pointee(cancel_token), next_worker_id: AtomicU64::new(1), resolve_enqueue_suppressions: Arc::new(DashMap::new()), + tmdb_source_markers: Arc::new(DashMap::new()), enqueue_state_loaded_inputs: Arc::new(DashMap::new()), enqueue_state_load_retry_at_ts: Arc::new(DashMap::new()), last_resolve_enqueue_suppression_prune_at_ts: AtomicI64::new(0), @@ -606,6 +627,7 @@ impl MetadataUpdateManager { self.is_shutdown_flag.store(true, Ordering::Release); self.cancel_token.load_full().cancel(); self.resolve_enqueue_suppressions.clear(); + self.tmdb_source_markers.clear(); self.enqueue_state_loaded_inputs.clear(); self.enqueue_state_load_retry_at_ts.clear(); } @@ -627,6 +649,7 @@ impl MetadataUpdateManager { let old_token = self.cancel_token.swap(Arc::new(cancel_token)); self.workers.clear(); self.resolve_enqueue_suppressions.clear(); + self.tmdb_source_markers.clear(); self.enqueue_state_loaded_inputs.clear(); self.enqueue_state_load_retry_at_ts.clear(); old_token @@ -745,10 +768,15 @@ impl MetadataUpdateManager { .cooldown_until_ts .unwrap_or(resolve_state.next_allowed_at_ts); if suppressed_until_ts > now_ts { - let scoped_key = ScopedTaskKey::new(input_name_cloned.clone(), task_key); + let scoped_key = ScopedTaskKey::new(input_name_cloned.clone(), task_key.clone()); self.resolve_enqueue_suppressions.insert(scoped_key, suppressed_until_ts); } } + if let Some(source_last_modified) = state.tmdb.as_ref().and_then(|tmdb_state| tmdb_state.source_last_modified) + { + let scoped_key = ScopedTaskKey::new(input_name_cloned.clone(), task_key.clone()); + self.tmdb_source_markers.insert(scoped_key, source_last_modified); + } } self.enqueue_state_load_retry_at_ts.remove(input_name); @@ -760,6 +788,30 @@ impl MetadataUpdateManager { self.should_skip_enqueue_cached(input_name.as_ref(), task) } + fn strip_tmdb_reasons_for_enqueue(&self, input_name: &str, task: UpdateTask) -> Option { + let scoped_key = Self::scoped_task_key(input_name, &task); + let current_last_modified = InputWorker::task_source_last_modified(&task); + let previous_last_modified = self.tmdb_source_markers.get(&scoped_key).map(|entry| *entry); + if (current_last_modified.is_some() || previous_last_modified.is_some()) + && current_last_modified == previous_last_modified + && InputWorker::task_has_tmdb_reason(&task) + { + InputWorker::strip_tmdb_reasons(&task) + } else { + Some(task) + } + } + + async fn prepare_task_for_enqueue(&self, input_name: Arc, task: UpdateTask) -> Option { + self.ensure_enqueue_state_loaded_for_input(&input_name).await; + let prepared_task = self.strip_tmdb_reasons_for_enqueue(input_name.as_ref(), task)?; + if self.should_skip_enqueue_cached(input_name.as_ref(), &prepared_task) { + None + } else { + Some(prepared_task) + } + } + /// Spawn a background task to queue the update. /// This is a fire-and-forget method that returns immediately. pub fn queue_task_background(self: &Arc, input_name: Arc, task: UpdateTask) { @@ -780,6 +832,10 @@ impl MetadataUpdateManager { #[allow(clippy::too_many_lines)] pub async fn queue_task(&self, input_name: Arc, task: UpdateTask) { debug!("[Task] Queuing task for input {input_name}: {task}"); + let Some(task_to_queue) = self.prepare_task_for_enqueue(input_name.clone(), task).await else { + return; + }; + // Read app state once and reuse for worker creation when needed. let app_state_weak = { let guard = self.app_state.lock().await; @@ -788,7 +844,6 @@ impl MetadataUpdateManager { let runtime_settings = MetadataUpdateRuntimeSettings::from_app_state(app_state_weak.as_ref()); let max_queue_size = runtime_settings.max_queue_size; - let task_to_queue = task; let mut channel_closed_attempt: u32 = 0; loop { // Atomically ensure there is exactly one worker context per input. @@ -843,6 +898,8 @@ impl MetadataUpdateManager { scheduled_requeues: Arc::new(DashMap::new()), recently_completed_no_change: HashMap::new(), resolve_enqueue_suppressions: Arc::clone(&self.resolve_enqueue_suppressions), + tmdb_source_markers: Arc::clone(&self.tmdb_source_markers), + dirty_retry_state_keys: HashSet::new(), }, )); @@ -1031,18 +1088,23 @@ impl MetadataUpdateManager { // Merge logic match (existing, task) { ( - UpdateTask::ResolveVod { reason: r1, delay: d1, .. }, - UpdateTask::ResolveVod { reason: r2, delay: d2, .. }, - ) - | ( - UpdateTask::ResolveSeries { reason: r1, delay: d1, .. }, - UpdateTask::ResolveSeries { reason: r2, delay: d2, .. }, + UpdateTask::ResolveVod { reason: r1, delay: d1, source_last_modified: lm1, .. }, + UpdateTask::ResolveVod { reason: r2, delay: d2, source_last_modified: lm2, .. }, + ) | ( + UpdateTask::ResolveSeries { reason: r1, delay: d1, source_last_modified: lm1, .. }, + UpdateTask::ResolveSeries { reason: r2, delay: d2, source_last_modified: lm2, .. }, ) => { let previous_reason = *r1; let previous_delay = *d1; + let previous_last_modified = *lm1; *r1 |= r2; *d1 = min(*d1, d2); - changed = *r1 != previous_reason || *d1 != previous_delay; + *lm1 = match (*lm1, lm2) { + (Some(left), Some(right)) => Some(left.max(right)), + (None, some @ Some(_)) | (some @ Some(_), None) => some, + (None, None) => None, + }; + changed = *r1 != previous_reason || *d1 != previous_delay || *lm1 != previous_last_modified; } ( UpdateTask::ProbeStream { reason: r1, delay: d1, url: url1, item_type: item_type1, .. }, @@ -1146,6 +1208,8 @@ struct InputWorker { // Stores the reason set so that tasks with new/different reasons are not wrongly skipped. recently_completed_no_change: HashMap, resolve_enqueue_suppressions: Arc>, + tmdb_source_markers: Arc>, + dirty_retry_state_keys: HashSet, } #[derive(Debug, Clone, Copy)] @@ -1275,7 +1339,29 @@ impl InputWorker { let mut clear_tmdb_state = false; if let Some(state_bundle) = self.retry_states.get(¤t_key) { if let Some(tmdb_state) = state_bundle.get(RetryDomain::Tmdb) { - if let Some(cooldown_until_ts) = tmdb_state.cooldown_until_ts { + let source_unchanged = tmdb_state.source_last_modified.is_some() + && tmdb_state.source_last_modified == Self::task_source_last_modified(&task_for_execution); + if source_unchanged && Self::task_has_tmdb_reason(&task_for_execution) { + if let Some(stripped_task) = Self::strip_tmdb_reasons(&task_for_execution) { + debug!( + "[Task] TMDB/date unchanged for input {}: {}, continuing with non-TMDB reasons (last_modified={:?})", + input_name, + task_for_execution, + tmdb_state.source_last_modified + ); + task_for_execution = stripped_task; + } else { + debug!( + "[Metadata-Task] Skipping task (TMDB/date unresolved and source unchanged) for input {}: {} (last_modified={:?})", + input_name, + task_for_execution, + tmdb_state.source_last_modified + ); + self.scheduled_requeues.remove(¤t_key); + remove_current_task = true; + skip_execution = true; + } + } else if let Some(cooldown_until_ts) = tmdb_state.cooldown_until_ts { if now_ts < cooldown_until_ts { if let Some(stripped_task) = Self::strip_tmdb_reasons(&task_for_execution) { debug!( @@ -1375,6 +1461,9 @@ impl InputWorker { if clear_active_retry_state && active_retry_domain == RetryDomain::Resolve { self.clear_resolve_enqueue_suppression(¤t_key); } + if clear_tmdb_state { + self.clear_tmdb_source_marker(¤t_key); + } metadata_persist_state = Some(state_after_clear); } } @@ -1421,7 +1510,9 @@ impl InputWorker { let active_retry_domain = Self::retry_domain_for_task(&task_for_execution); let task_has_tmdb_reason = Self::task_has_tmdb_reason(&task_for_execution); + let task_source_last_modified = Self::task_source_last_modified(&task_for_execution); let mut should_remove_retry_entry = false; + let mut clear_tmdb_marker = false; let mut state_after_success: Option = None; if let Some(state_bundle) = self.retry_states.get_mut(¤t_key) { @@ -1435,6 +1526,7 @@ impl InputWorker { tmdb_state.cooldown_until_ts = Some(cooldown_until_ts); tmdb_state.last_error = Some("TMDB lookup completed without matching result".to_string()); + tmdb_state.source_last_modified = task_source_last_modified; debug!( "[Metadata-Task] TMDB resolve produced no match (existing retry state), entering cooldown for input {}: {} (cooldown_until={}, cooldown_duration={}s)", input_name, @@ -1444,6 +1536,7 @@ impl InputWorker { ); } else { state_bundle.clear_domain(RetryDomain::Tmdb); + clear_tmdb_marker = true; } } @@ -1463,6 +1556,7 @@ impl InputWorker { next_allowed_at_ts: cooldown_until_ts, cooldown_until_ts: Some(cooldown_until_ts), last_error: Some("TMDB lookup completed without matching result".to_string()), + source_last_modified: task_source_last_modified, }), updated_at_ts: now_ts.max(1), }; @@ -1475,11 +1569,23 @@ impl InputWorker { cooldown_until_ts, runtime_settings.tmdb_cooldown_secs ); + } else if task_has_tmdb_reason { + self.clear_tmdb_source_marker(¤t_key); } if should_remove_retry_entry { self.retry_states.remove(¤t_key); } + if clear_tmdb_marker { + self.clear_tmdb_source_marker(¤t_key); + } + if task_outcome.tmdb_pending { + if let Some(source_last_modified) = task_source_last_modified { + self.set_tmdb_source_marker(¤t_key, source_last_modified); + } else { + self.clear_tmdb_source_marker(¤t_key); + } + } if should_remove_retry_entry || state_after_success.is_some() { metadata_persist_state = Some(state_after_success); } @@ -1713,6 +1819,7 @@ impl InputWorker { } if let Some(state) = metadata_persist_state { + self.dirty_retry_state_keys.insert(current_key.clone()); self.persist_metadata_retry_state(¤t_key, state.as_ref()).await; } @@ -1883,19 +1990,26 @@ impl InputWorker { self.metadata_retry_load_retry_at_ts = None; } - async fn persist_metadata_retry_state(&self, key: &TaskKey, state: Option<&TaskRetryState>) { + async fn persist_metadata_retry_state(&mut self, key: &TaskKey, state: Option<&TaskRetryState>) { let Some(path) = self.metadata_retry_state_path.clone() else { return; }; + if !self.dirty_retry_state_keys.contains(key) { + return; + } - let key = key.clone(); + let key_for_persist = key.clone(); + let key_for_dirty = key.clone(); let state = state.cloned(); let input_name = self.input_name.clone(); let persist_result = - spawn_blocking_limited(move || persist_metadata_retry_state_to_disk(&path, &key, state.as_ref())).await; + spawn_blocking_limited(move || persist_metadata_retry_state_to_disk(&path, &key_for_persist, state.as_ref())) + .await; match persist_result { - Ok(Ok(())) => {} + Ok(Ok(())) => { + self.dirty_retry_state_keys.remove(&key_for_dirty); + } Ok(Err(err)) => warn!("Failed to persist metadata retry state for input {input_name}: {err}"), Err(err) => warn!("Failed to persist metadata retry state for input {input_name}: {err}"), } @@ -2145,6 +2259,8 @@ impl InputWorker { for key in stale_retry_keys { if self.retry_states.remove(&key).is_some() { self.clear_resolve_enqueue_suppression(&key); + self.clear_tmdb_source_marker(&key); + self.dirty_retry_state_keys.insert(key.clone()); debug_if_enabled!("Pruned stale metadata retry state for input {}: {:?}", self.input_name, key); self.persist_metadata_retry_state(&key, None).await; } @@ -2317,6 +2433,16 @@ impl InputWorker { self.resolve_enqueue_suppressions.remove(&scoped_key); } + fn set_tmdb_source_marker(&self, current_key: &TaskKey, source_last_modified: u64) { + let scoped_key = ScopedTaskKey::new(self.input_name.clone(), current_key.clone()); + self.tmdb_source_markers.insert(scoped_key, source_last_modified); + } + + fn clear_tmdb_source_marker(&self, current_key: &TaskKey) { + let scoped_key = ScopedTaskKey::new(self.input_name.clone(), current_key.clone()); + self.tmdb_source_markers.remove(&scoped_key); + } + fn sync_resolve_enqueue_suppression_from_retry_state(&self, current_key: &TaskKey, state: &TaskRetryState, now_ts: i64) { let Some(resolve_state) = state.resolve.as_ref() else { self.clear_resolve_enqueue_suppression(current_key); @@ -2348,26 +2474,46 @@ impl InputWorker { } } + #[inline] + fn task_source_last_modified(task: &UpdateTask) -> Option { + match task { + UpdateTask::ResolveVod { source_last_modified, .. } | UpdateTask::ResolveSeries { source_last_modified, .. } => { + *source_last_modified + } + _ => None, + } + } + fn strip_tmdb_reasons(task: &UpdateTask) -> Option { match task { - UpdateTask::ResolveVod { id, reason, delay } => { + UpdateTask::ResolveVod { id, reason, delay, source_last_modified } => { let mut next_reason = *reason; next_reason.unset(ResolveReason::Tmdb); next_reason.unset(ResolveReason::Date); if next_reason.is_empty() { None } else { - Some(UpdateTask::ResolveVod { id: id.clone(), reason: next_reason, delay: *delay }) + Some(UpdateTask::ResolveVod { + id: id.clone(), + reason: next_reason, + delay: *delay, + source_last_modified: *source_last_modified, + }) } } - UpdateTask::ResolveSeries { id, reason, delay } => { + UpdateTask::ResolveSeries { id, reason, delay, source_last_modified } => { let mut next_reason = *reason; next_reason.unset(ResolveReason::Tmdb); next_reason.unset(ResolveReason::Date); if next_reason.is_empty() { None } else { - Some(UpdateTask::ResolveSeries { id: id.clone(), reason: next_reason, delay: *delay }) + Some(UpdateTask::ResolveSeries { + id: id.clone(), + reason: next_reason, + delay: *delay, + source_last_modified: *source_last_modified, + }) } } _ => Some(task.clone()), @@ -3414,6 +3560,8 @@ mod tests { scheduled_requeues: Arc::new(DashMap::new()), recently_completed_no_change: HashMap::new(), resolve_enqueue_suppressions: Arc::new(DashMap::new()), + tmdb_source_markers: Arc::new(DashMap::new()), + dirty_retry_state_keys: HashSet::new(), } } @@ -3435,6 +3583,7 @@ mod tests { id: ProviderIdType::Id(id), reason: ResolveReasonSet::default(), delay: 0, + source_last_modified: None, }, ) .await; @@ -3460,6 +3609,7 @@ mod tests { id: ProviderIdType::Id(42), reason: ResolveReasonSet::from_variants(&[ResolveReason::Info]), delay: 1, + source_last_modified: None, }; let scoped_key = ScopedTaskKey::new(Arc::from(input_name), TaskKey::from_task(&task)); manager @@ -3476,6 +3626,146 @@ mod tests { assert!(!manager.resolve_enqueue_suppressions.contains_key(&scoped_key)); } + #[test] + fn strip_tmdb_reasons_for_enqueue_skips_unchanged_series_tmdb_only_task() { + let manager = MetadataUpdateManager::new(CancellationToken::new()); + let input_name = "input_tmdb_marker"; + let task = UpdateTask::ResolveSeries { + id: ProviderIdType::Id(42), + reason: ResolveReasonSet::from_variants(&[ResolveReason::Tmdb, ResolveReason::Date]), + delay: 1, + source_last_modified: Some(777), + }; + let scoped_key = ScopedTaskKey::new(Arc::from(input_name), TaskKey::from_task(&task)); + manager.tmdb_source_markers.insert(scoped_key, 777); + + assert!(manager.strip_tmdb_reasons_for_enqueue(input_name, task).is_none()); + } + + #[test] + fn strip_tmdb_reasons_for_enqueue_keeps_non_tmdb_reasons_for_unchanged_series_source() { + let manager = MetadataUpdateManager::new(CancellationToken::new()); + let input_name = "input_tmdb_probe"; + let task = UpdateTask::ResolveSeries { + id: ProviderIdType::Id(42), + reason: ResolveReasonSet::from_variants(&[ResolveReason::Tmdb, ResolveReason::Probe]), + delay: 1, + source_last_modified: Some(777), + }; + let scoped_key = ScopedTaskKey::new(Arc::from(input_name), TaskKey::from_task(&task)); + manager.tmdb_source_markers.insert(scoped_key, 777); + + let prepared = manager + .strip_tmdb_reasons_for_enqueue(input_name, task) + .expect("probe reason should remain"); + match prepared { + UpdateTask::ResolveSeries { reason, source_last_modified, .. } => { + assert_eq!(source_last_modified, Some(777)); + assert!(!reason.contains(ResolveReason::Tmdb)); + assert!(!reason.contains(ResolveReason::Date)); + assert!(reason.contains(ResolveReason::Probe)); + } + other => panic!("unexpected task type after enqueue strip: {other:?}"), + } + } + + #[test] + fn strip_tmdb_reasons_for_enqueue_keeps_unknown_series_timestamp_and_tmdb_reason() { + let manager = MetadataUpdateManager::new(CancellationToken::new()); + let input_name = "input_tmdb_unknown_series"; + let task = UpdateTask::ResolveSeries { + id: ProviderIdType::Id(42), + reason: ResolveReasonSet::from_variants(&[ResolveReason::Tmdb, ResolveReason::Date]), + delay: 1, + source_last_modified: None, + }; + let scoped_key = ScopedTaskKey::new(Arc::from(input_name), TaskKey::from_task(&task)); + + assert!(!manager.tmdb_source_markers.contains_key(&scoped_key)); + + let prepared = manager + .strip_tmdb_reasons_for_enqueue(input_name, task) + .expect("unknown timestamps should not be suppressed"); + match prepared { + UpdateTask::ResolveSeries { reason, source_last_modified, .. } => { + assert_eq!(source_last_modified, None); + assert!(reason.contains(ResolveReason::Tmdb)); + assert!(reason.contains(ResolveReason::Date)); + } + other => panic!("unexpected task type after enqueue strip: {other:?}"), + } + } + + #[test] + fn strip_tmdb_reasons_for_enqueue_skips_unchanged_vod_tmdb_only_task() { + let manager = MetadataUpdateManager::new(CancellationToken::new()); + let input_name = "input_vod_tmdb_marker"; + let task = UpdateTask::ResolveVod { + id: ProviderIdType::Id(42), + reason: ResolveReasonSet::from_variants(&[ResolveReason::Tmdb, ResolveReason::Date]), + delay: 1, + source_last_modified: Some(777), + }; + let scoped_key = ScopedTaskKey::new(Arc::from(input_name), TaskKey::from_task(&task)); + manager.tmdb_source_markers.insert(scoped_key, 777); + + assert!(manager.strip_tmdb_reasons_for_enqueue(input_name, task).is_none()); + } + + #[test] + fn strip_tmdb_reasons_for_enqueue_keeps_non_tmdb_reasons_for_unchanged_vod_source() { + let manager = MetadataUpdateManager::new(CancellationToken::new()); + let input_name = "input_vod_tmdb_probe"; + let task = UpdateTask::ResolveVod { + id: ProviderIdType::Id(42), + reason: ResolveReasonSet::from_variants(&[ResolveReason::Tmdb, ResolveReason::Probe]), + delay: 1, + source_last_modified: Some(777), + }; + let scoped_key = ScopedTaskKey::new(Arc::from(input_name), TaskKey::from_task(&task)); + manager.tmdb_source_markers.insert(scoped_key, 777); + + let prepared = manager + .strip_tmdb_reasons_for_enqueue(input_name, task) + .expect("probe reason should remain"); + match prepared { + UpdateTask::ResolveVod { reason, source_last_modified, .. } => { + assert_eq!(source_last_modified, Some(777)); + assert!(!reason.contains(ResolveReason::Tmdb)); + assert!(!reason.contains(ResolveReason::Date)); + assert!(reason.contains(ResolveReason::Probe)); + } + other => panic!("unexpected task type after enqueue strip: {other:?}"), + } + } + + #[test] + fn strip_tmdb_reasons_for_enqueue_keeps_unknown_vod_timestamp_and_tmdb_reason() { + let manager = MetadataUpdateManager::new(CancellationToken::new()); + let input_name = "input_tmdb_unknown_vod"; + let task = UpdateTask::ResolveVod { + id: ProviderIdType::Id(42), + reason: ResolveReasonSet::from_variants(&[ResolveReason::Tmdb, ResolveReason::Date]), + delay: 1, + source_last_modified: None, + }; + let scoped_key = ScopedTaskKey::new(Arc::from(input_name), TaskKey::from_task(&task)); + + assert!(!manager.tmdb_source_markers.contains_key(&scoped_key)); + + let prepared = manager + .strip_tmdb_reasons_for_enqueue(input_name, task) + .expect("unknown timestamps should not be suppressed"); + match prepared { + UpdateTask::ResolveVod { reason, source_last_modified, .. } => { + assert_eq!(source_last_modified, None); + assert!(reason.contains(ResolveReason::Tmdb)); + assert!(reason.contains(ResolveReason::Date)); + } + other => panic!("unexpected task type after enqueue strip: {other:?}"), + } + } + #[tokio::test] async fn submit_task_merges_existing_task_and_increments_generation() { let (tx, mut rx) = mpsc::channel::(8); @@ -3486,6 +3776,7 @@ mod tests { id: ProviderIdType::Id(42), reason: ResolveReasonSet::from_variants(&[ResolveReason::Info]), delay: 10, + source_last_modified: None, }; let queue_size = MetadataUpdateRuntimeSettings::default().max_queue_size; MetadataUpdateManager::submit_task( @@ -3502,6 +3793,7 @@ mod tests { id: ProviderIdType::Id(42), reason: ResolveReasonSet::from_variants(&[ResolveReason::Probe]), delay: 2, + source_last_modified: None, }; MetadataUpdateManager::submit_task( tx, @@ -3545,6 +3837,7 @@ mod tests { id: ProviderIdType::Id(42), reason: ResolveReasonSet::from_variants(&[ResolveReason::Tmdb]), delay: 10, + source_last_modified: None, }; MetadataUpdateManager::submit_task( tx.clone(), @@ -3560,6 +3853,7 @@ mod tests { id: ProviderIdType::Id(42), reason: ResolveReasonSet::from_variants(&[ResolveReason::Tmdb]), delay: 10, + source_last_modified: None, }; MetadataUpdateManager::submit_task( tx, @@ -3661,6 +3955,7 @@ mod tests { id: ProviderIdType::Id(42), reason: ResolveReasonSet::from_variants(&[ResolveReason::Info]), delay: 10, + source_last_modified: None, }; pending_tasks.insert(key.clone(), PendingTask::new(existing_task)); @@ -3668,6 +3963,7 @@ mod tests { id: ProviderIdType::Id(42), reason: ResolveReasonSet::from_variants(&[ResolveReason::Probe]), delay: 2, + source_last_modified: None, }; let result = MetadataUpdateManager::submit_task( @@ -3698,6 +3994,7 @@ mod tests { id: ProviderIdType::Id(7), reason: ResolveReasonSet::from_variants(&[ResolveReason::Info]), delay: 0, + source_last_modified: None, }), ); if let Some(entry) = pending_tasks.get(&key) { @@ -3722,6 +4019,7 @@ mod tests { id: ProviderIdType::Id(9), reason: ResolveReasonSet::from_variants(&[ResolveReason::Info]), delay: 0, + source_last_modified: None, }; pending_tasks.insert(key.clone(), PendingTask::new(task.clone())); @@ -3754,6 +4052,7 @@ mod tests { id: ProviderIdType::Id(10), reason: ResolveReasonSet::from_variants(&[ResolveReason::Info]), delay: 0, + source_last_modified: None, }; let mut worker = create_test_worker("input_c", tx, rx, pending_tasks, pending_task_count); let runtime_settings = MetadataUpdateRuntimeSettings::default(); @@ -3786,6 +4085,7 @@ mod tests { id: ProviderIdType::Id(11), reason: ResolveReasonSet::from_variants(&[ResolveReason::Info]), delay: 0, + source_last_modified: None, }; pending_tasks.insert(key.clone(), PendingTask::new(task.clone())); @@ -3858,6 +4158,7 @@ mod tests { id: ProviderIdType::Id(100), reason: ResolveReasonSet::from_variants(&[ResolveReason::Tmdb, ResolveReason::Date]), delay: 5, + source_last_modified: None, }; assert!(InputWorker::strip_tmdb_reasons(&task).is_none()); @@ -3869,6 +4170,7 @@ mod tests { id: ProviderIdType::Id(5), reason: ResolveReasonSet::from_variants(&[ResolveReason::Tmdb, ResolveReason::Probe, ResolveReason::Info]), delay: 1, + source_last_modified: Some(123), }; let stripped = InputWorker::strip_tmdb_reasons(&task).expect("task should keep non-tmdb reasons"); @@ -3894,18 +4196,21 @@ mod tests { next_allowed_at_ts: 1_700_043_200, cooldown_until_ts: Some(1_700_043_200), last_error: Some("resolve exhausted".to_string()), + source_last_modified: None, }), probe: Some(RetryState { attempts: 3, next_allowed_at_ts: 1_700_000_000, cooldown_until_ts: Some(1_700_086_400), last_error: Some("probe timeout".to_string()), + source_last_modified: None, }), tmdb: Some(RetryState { attempts: 0, next_allowed_at_ts: 1_700_172_800, cooldown_until_ts: Some(1_700_172_800), last_error: Some("tmdb no match".to_string()), + source_last_modified: Some(999), }), updated_at_ts: 1_700_000_000, }; @@ -3925,6 +4230,7 @@ mod tests { let loaded_tmdb = loaded_state.tmdb.as_ref().expect("tmdb retry state should be present"); assert_eq!(loaded_tmdb.cooldown_until_ts, Some(1_700_172_800)); assert_eq!(loaded_tmdb.last_error.as_deref(), Some("tmdb no match")); + assert_eq!(loaded_tmdb.source_last_modified, Some(999)); persist_metadata_retry_state_to_disk(&path, &key, None).expect("state clear should succeed"); let cleared = load_metadata_retry_states_from_disk(&path).expect("state reload should succeed"); @@ -3971,6 +4277,7 @@ mod tests { id: ProviderIdType::Id(7), reason: ResolveReasonSet::from_variants(&[ResolveReason::Probe]), delay: 0, + source_last_modified: None, }; assert_eq!(InputWorker::retry_domain_for_task(&task), RetryDomain::Probe); } @@ -3981,6 +4288,7 @@ mod tests { id: ProviderIdType::Id(11), reason: ResolveReasonSet::from_variants(&[ResolveReason::Probe, ResolveReason::Info]), delay: 0, + source_last_modified: None, }; assert_eq!(InputWorker::retry_domain_for_task(&task), RetryDomain::Resolve); } @@ -4017,6 +4325,7 @@ mod tests { id: ProviderIdType::Id(77), reason: ResolveReasonSet::from_variants(&[ResolveReason::Probe]), delay: 0, + source_last_modified: None, }; pending_tasks.insert(key.clone(), PendingTask::new(task)); @@ -4042,6 +4351,7 @@ mod tests { id: ProviderIdType::Id(33), reason: ResolveReasonSet::from_variants(&[ResolveReason::Info]), delay: 0, + source_last_modified: None, }; assert!(InputWorker::should_trigger_playlist_update_for_task(&task, true)); assert!(!InputWorker::should_trigger_playlist_update_for_task(&task, false)); diff --git a/backend/src/api/model/streams/active_client_stream.rs b/backend/src/api/model/streams/active_client_stream.rs index 1ec4f20ca..825a1f801 100644 --- a/backend/src/api/model/streams/active_client_stream.rs +++ b/backend/src/api/model/streams/active_client_stream.rs @@ -231,8 +231,16 @@ impl ActiveClientStreamState { self.inner = None; let video_type = Self::custom_video_type_for_mode(mode); + let reason = match mode { + StreamMode::ChannelUnavailable => "unavailable provider channel", + StreamMode::UserExhausted => "user grace period exhaustion", + StreamMode::ProviderExhausted => "provider grace period exhaustion", + StreamMode::Provisioning => "provider grace period provisioning", + StreamMode::LowPriorityPreempted => "low-priority preemption", + StreamMode::Inner | StreamMode::GracePending => "stream mode transition", + }; debug_if_enabled!( - "Provider stream stopped due to grace period or unavailable provider channel for {}", + "Provider stream stopped due to {reason} for {}", sanitize_sensitive_info(&addr.to_string()) ); self.connection_manager.send_cleanup(CleanupEvent::UpdateDetailAndReleaseProvider { diff --git a/backend/src/api/model/streams/provider_stream_factory.rs b/backend/src/api/model/streams/provider_stream_factory.rs index cab5a5491..0d934ec8e 100644 --- a/backend/src/api/model/streams/provider_stream_factory.rs +++ b/backend/src/api/model/streams/provider_stream_factory.rs @@ -106,7 +106,7 @@ impl ProviderStreamFactoryOptions { Some(Arc::new(AtomicUsize::new(requested_range.unwrap_or(0)))) }; let mut flags = ProviderStreamFactoryFlagsSet::new(); - if stream_options.stream_retry { + if stream_options.stream_retry && !item_type.is_live_adaptive() { flags.set(ProviderStreamFactoryFlags::ReconnectEnabled); } if stream_options.pipe_provider_stream { @@ -790,4 +790,40 @@ mod tests { assert!(!options.was_range_requested()); // Stripped by filter assert_eq!(options.get_total_bytes_send(), None); } + + #[test] + fn test_provider_stream_factory_options_disables_reconnect_for_live_adaptive_streams() { + let addr = "127.0.0.1:8080".parse().unwrap(); + let stream_url = Url::parse("http://example.com/segment.ts").unwrap(); + let req_headers = HeaderMap::new(); + let stream_options = + StreamOptions { stream_retry: true, buffer_enabled: true, buffer_size: 1024, pipe_provider_stream: false }; + + let hls_options = ProviderStreamFactoryOptions::new( + addr, + PlaylistItemType::LiveHls, + false, + &stream_options, + &stream_url, + &req_headers, + None, + None, + None, + ); + + let dash_options = ProviderStreamFactoryOptions::new( + addr, + PlaylistItemType::LiveDash, + false, + &stream_options, + &stream_url, + &req_headers, + None, + None, + None, + ); + + assert!(!hls_options.should_reconnect()); + assert!(!dash_options.should_reconnect()); + } } diff --git a/backend/src/model/xtream.rs b/backend/src/model/xtream.rs index 93e8389e2..ad64938e2 100644 --- a/backend/src/model/xtream.rs +++ b/backend/src/model/xtream.rs @@ -4,19 +4,13 @@ use crate::model::{ConfigTarget, XtreamTargetOutput}; use serde::{Deserialize, Serialize}; use serde_json::Value; use shared::model::{ - PlaylistItem, PlaylistItemType, PlaylistItemTypeSet, ProxyUserStatus, XtreamMappingFlags, - XtreamMappingFlagsSet, XtreamMappingOptions, + PlaylistItem, PlaylistItemType, PlaylistItemTypeSet, XtreamMappingFlags, XtreamMappingFlagsSet, + XtreamMappingOptions, }; use shared::utils::{arc_str_serde, concat_path_leading_slash, deserialize_number_from_string_or_zero}; use enum_iterator::all; use crate::model::XtreamTargetFlags; -#[derive(Debug, Default)] -pub struct XtreamLoginInfo { - pub status: Option, - pub exp_date: Option, -} - #[derive(Deserialize, Default)] pub struct XtreamCategory { #[serde(deserialize_with = "deserialize_number_from_string_or_zero", serialize_with = "shared::utils::serialize_number_as_string")] diff --git a/backend/src/processing/parser/m3u.rs b/backend/src/processing/parser/m3u.rs index 21d299a10..bfcefa817 100644 --- a/backend/src/processing/parser/m3u.rs +++ b/backend/src/processing/parser/m3u.rs @@ -6,56 +6,76 @@ use std::borrow::BorrowMut; use std::sync::Arc; use tokio::io::AsyncBufReadExt; use indexmap::IndexMap; -use shared::concat_string; use crate::repository::CategoryKey; // other implementations like calculating text_distance on all titles took too much time // we keep it now as simple as possible and less memory intensive. fn get_title_group(text: &Arc) -> Arc { - let alphabetic_only: String = text.chars().map(|c| if c.is_alphanumeric() { c } else { ' ' }).collect(); - let parts = alphabetic_only.split_whitespace(); let mut combination = String::new(); - for p in parts { - combination = concat_string!(&combination, " " , p); - if combination.len() > 2 { - return combination.intern(); + let mut in_word = false; + for c in text.chars() { + if c.is_alphanumeric() { + if !in_word && !combination.is_empty() { + combination.push(' '); + } + in_word = true; + combination.push(c); + } else { + if in_word && combination.len() > 2 { + return combination.intern(); + } + in_word = false; } } + if combination.len() > 2 { + return combination.intern(); + } text.clone() } +/// Reads a quoted value into the stack and returns the start offset. +/// The value is `&stack[offset..]`. Caller must call `stack.truncate(offset)` after use. #[inline] -fn token_value(stack: &mut String, it: &mut std::str::Chars) -> String { - // Use .find() to skip until the first double quote (") character. +fn token_value(stack: &mut String, it: &mut std::str::Chars) -> usize { + let offset = stack.len(); + // `any()` intentionally consumes chars from the iterator, advancing it past the opening '"'. + // Returns true when a quote is found, so read_value can extract the content until the closing '"'. if it.any(|ch| ch == '"') { - // If a quote is found, call get_value to extract the value. - return get_value(stack, it); + read_value(stack, it); } - // If no double quote is found, return an empty string. - String::new() + offset } -fn get_value(stack: &mut String, it: &mut std::str::Chars) -> String { +/// Reads a quoted or comma-delimited value into the stack and returns the start offset. +fn read_value(stack: &mut String, it: &mut std::str::Chars) { for c in it.skip_while(|c| c.is_whitespace()) { if c == '"' { break; } stack.push(c); } - - let result = (*stack).clone(); - stack.clear(); - result } -fn token_till(stack: &mut String, it: &mut std::str::Chars, stop_char: char, start_with_alpha: bool) -> Option { +/// Reads a comma-delimited value (title) into the stack and returns the start offset. +#[inline] +fn title_value(stack: &mut String, it: &mut std::str::Chars) -> usize { + let offset = stack.len(); + read_value(stack, it); + offset +} + +/// Reads characters into the stack until `stop_char` is reached. +/// Returns `Some(offset)` where `&stack[offset..]` is the token, or `None` if empty. +/// Caller must call `stack.truncate(offset)` after use. +fn token_till(stack: &mut String, it: &mut std::str::Chars, stop_char: char, start_with_alpha: bool) -> Option { + let offset = stack.len(); let mut skip_non_alpha = start_with_alpha; for ch in it.by_ref() { if ch == stop_char { break; } - if stack.is_empty() && ch.is_whitespace() { + if stack.len() == offset && ch.is_whitespace() { continue; } @@ -69,12 +89,10 @@ fn token_till(stack: &mut String, it: &mut std::str::Chars, stop_char: char, sta stack.push(ch); } - if stack.is_empty() { + if stack.len() == offset { None } else { - let result = (*stack).clone(); - stack.clear(); - Some(result) + Some(offset) } } @@ -101,19 +119,57 @@ fn create_empty_playlistitem_header(input_name: &Arc, url: String) -> Playl } } -macro_rules! process_header_fields { - ($header:expr, $token:expr, $(($prop:ident, $field:expr)),*; $val:expr) => { - match $token { - $( - $field => $header.$prop = $val, - )* - _ => {} - } - }; +enum M3uToken { + ProviderId, + TvgChno, + GroupTitle, + TvgId, + TvgName, + TvgLogo, + TvgLogoSmall, + ParentCode, + AudioTrack, + TimeShift, + TvgRec, + PossibleId, + Unknown, +} + +fn classify_token(t: &str) -> M3uToken { + if t.eq_ignore_ascii_case("xui-id") || t.eq_ignore_ascii_case("cuid") { + M3uToken::ProviderId + } else if t.eq_ignore_ascii_case("tvg-chno") { + M3uToken::TvgChno + } else if t.eq_ignore_ascii_case("group-title") { + M3uToken::GroupTitle + } else if t.eq_ignore_ascii_case("tvg-id") { + M3uToken::TvgId + } else if t.eq_ignore_ascii_case("tvg-name") { + M3uToken::TvgName + } else if t.eq_ignore_ascii_case("tvg-logo") { + M3uToken::TvgLogo + } else if t.eq_ignore_ascii_case("tvg-logo-small") { + M3uToken::TvgLogoSmall + } else if t.eq_ignore_ascii_case("parent-code") { + M3uToken::ParentCode + } else if t.eq_ignore_ascii_case("audio-track") { + M3uToken::AudioTrack + } else if t.eq_ignore_ascii_case("timeshift") { + M3uToken::TimeShift + } else if t.eq_ignore_ascii_case("tvg-rec") { + M3uToken::TvgRec + } else if t.eq_ignore_ascii_case("id") || + (t.len() > 2 + && !t.as_bytes()[..3].eq_ignore_ascii_case(b"tvg") + && t.as_bytes()[t.len() - 2..].eq_ignore_ascii_case(b"id")) + { + M3uToken::PossibleId + } else { + M3uToken::Unknown + } } fn process_header(input_name: &Arc, video_suffixes: &[String], content: &str, url: String) -> PlaylistItemHeader { - let url_id = extract_id_from_url(&url); let url_types = if video_suffixes.iter().any(|suffix| url.ends_with(suffix)) { // TODO find Series based on group or configured names Some((XtreamCluster::Video, PlaylistItemType::Video)) @@ -124,65 +180,67 @@ fn process_header(input_name: &Arc, video_suffixes: &[String], content: &st let mut plih = create_empty_playlistitem_header(input_name, url); let mut it = content.chars(); let mut stack = String::with_capacity(64); - let line_token = token_till(&mut stack, &mut it, ':', false); - if line_token.as_deref() == Some("#EXTINF") { + let is_extinf = token_till(&mut stack, &mut it, ':', false) + .is_some_and(|off| stack[off..].eq_ignore_ascii_case("#EXTINF")); + stack.clear(); + if is_extinf { let mut provider_id = None::; + let mut fallback_id = None::; let mut c = skip_digit(&mut it); - loop { - match c { - None => break, - Some(chr) => { - if chr.is_whitespace() { - // skip - } else if chr == ',' { - plih.title = get_value(&mut stack, &mut it).intern(); - } else { - stack.push(chr); - let token = token_till(&mut stack, &mut it, '=', true); - if let Some(t) = token { - let value = token_value(&mut stack, &mut it); - let token = t.to_lowercase(); - if token.as_str() == "xui-id" { - if !value.is_empty() { - provider_id = Some(value); - } - } else if token == "tvg-chno" { - plih.chno = value.parse::().unwrap_or(0); - } else if token == "group-title" { - plih.group = value.intern(); - } else if token == "tvg-id" { - plih.id = value.intern(); - } else if token == "tvg-name" { - plih.name = value.intern(); - } else if token == "tvg-logo" { - plih.logo = value.intern(); - } else if token == "tvg-logo-small" { - plih.logo_small = value.intern(); - } else { - process_header_fields!(plih, token.as_str(), - (parent_code, "parent-code"), - (audio_track, "audio-track"), - (time_shift, "timeshift"), - (rec, "tvg-rec"); value.intern()); + while let Some(chr) = c { + match chr { + _ if chr.is_whitespace() => {} + ',' => { + let off = title_value(&mut stack, &mut it); + plih.title = stack[off..].intern(); + stack.truncate(off); + } + _ => { + let tok_start = stack.len(); + stack.push(chr); + if token_till(&mut stack, &mut it, '=', true).is_some() { + let token = classify_token(&stack[tok_start..]); + stack.clear(); + let val_off = token_value(&mut stack, &mut it); + match token { + M3uToken::ProviderId if stack.len() > val_off => { + provider_id = Some(stack[val_off..].to_owned()); } + M3uToken::TvgChno => plih.chno = stack[val_off..].parse::().unwrap_or(0), + M3uToken::GroupTitle => plih.group = stack[val_off..].intern(), + M3uToken::TvgId => plih.epg_channel_id = if stack.len() == val_off { None } else { Some(stack[val_off..].intern()) }, + M3uToken::TvgName => plih.name = stack[val_off..].intern(), + M3uToken::TvgLogo => plih.logo = stack[val_off..].intern(), + M3uToken::TvgLogoSmall => plih.logo_small = stack[val_off..].intern(), + M3uToken::ParentCode => plih.parent_code = stack[val_off..].intern(), + M3uToken::AudioTrack => plih.audio_track = stack[val_off..].intern(), + M3uToken::TimeShift => plih.time_shift = stack[val_off..].intern(), + M3uToken::TvgRec => plih.rec = stack[val_off..].intern(), + // Unknown panel-specific ID fields (e.g. "stream-id", "channel-uid") + M3uToken::PossibleId + if fallback_id.is_none() + && stack.len() > val_off + && stack[val_off..].bytes().all(|b| b.is_ascii_digit()) => + { + fallback_id = Some(stack[val_off..].to_owned()); + } + _ => {} } + stack.clear(); } } } c = it.next(); } - if plih.id.is_empty() { - plih.epg_channel_id = None; - if let Some(pid) = provider_id { - plih.id = pid.intern(); - } else if !url_id.is_empty() { - plih.id = url_id.intern(); - } + if let Some(pid) = provider_id { + plih.id = pid.intern(); + } else if let Some(fid) = fallback_id { + plih.id = fid.intern(); } else { - plih.epg_channel_id = Some(Arc::clone(&plih.id)); - if let Some(pid) = provider_id { - plih.id = pid.intern(); + let url_id = extract_id_from_url(&plih.url); + if !url_id.is_empty() { + plih.id = url_id.intern(); } } } @@ -286,12 +344,14 @@ mod test { fn test_process_header_1() { let input = "hello".intern(); let video_suffixes = Vec::new(); - let url = "http://hello.de/hello.ts"; + let url = "http://hello.de/live/user/pass/70001.ts"; let line = r#"#EXTINF:-1 channel-id="abc-seven" tvg-id="abc-seven" tvg-logo="https://abc.nz/.images/seven.png" tvg-chno="7" group-title="Sydney" , Seven"#; let pli = process_header(&input, &video_suffixes, line, url.to_string()); assert_eq!(pli.title, "Seven".intern()); - assert_eq!(pli.id, "abc-seven".intern()); + // tvg-id is preserved as epg_channel_id, id falls back to numeric url segment + assert_eq!(pli.epg_channel_id, Some("abc-seven".intern())); + assert_eq!(pli.id, "70001".intern()); assert_eq!(pli.logo, "https://abc.nz/.images/seven.png".intern()); assert_eq!(pli.chno, 7); assert_eq!(&*pli.group, "Sydney"); @@ -301,17 +361,84 @@ mod test { fn test_process_header_2() { let input = "hello".intern(); let video_suffixes = Vec::new(); - let url = "http://hello.de/hello.ts"; + let url = "http://hello.de/live/user/pass/70002.ts"; let line = r#"#EXTINF:-1 channel-id="abc-seven" tvg-id="abc-seven" tvg-logo="https://abc.nz/.images/seven.png" tvg-chno="7" group-title="Sydney", Seven"#; let pli = process_header(&input, &video_suffixes, line, url.to_string()); assert_eq!(pli.title, "Seven".intern()); - assert_eq!(pli.id, "abc-seven".intern()); + assert_eq!(pli.epg_channel_id, Some("abc-seven".intern())); + assert_eq!(pli.id, "70002".intern()); assert_eq!(pli.logo, "https://abc.nz/.images/seven.png".intern()); assert_eq!(pli.chno, 7); assert_eq!(&*pli.group, "Sydney"); } + #[test] + fn test_process_header_cuid_format() { + let input = "test".intern(); + let video_suffixes = Vec::new(); + let url = "http://line.trx-ott.com/live/18be61b480/fc19249ec409/1905905.ts"; + let line = r#"#EXTINF:0 CUID="12046" tvg-name="UK-NOWTV| SKY CRIME FHD" tvg-id="skycrime.uk" tvg-logo="https://logo.m3uassets.com/skycrime.png" group-title="🔪Murder Mystery",UK-NOWTV| SKY CRIME FHD"#; + + let pli = process_header(&input, &video_suffixes, line, url.to_string()); + assert_eq!(pli.name, "UK-NOWTV| SKY CRIME FHD".intern()); + assert_eq!(pli.title, "UK-NOWTV| SKY CRIME FHD".intern()); + assert_eq!(pli.id, "12046".intern()); // CUID is recognized as provider_id, overrides tvg-id + assert_eq!(pli.logo, "https://logo.m3uassets.com/skycrime.png".intern()); + assert_eq!(&*pli.group, "🔪Murder Mystery"); + assert_eq!(pli.epg_channel_id, Some("skycrime.uk".intern())); + } + + #[test] + fn test_process_header_tvg_id_uses_url_id_fallback() { + let input = "test".intern(); + let video_suffixes = Vec::new(); + // Numeric last segment in URL -> extract_id_from_url returns "1905905" + let url = "http://line.trx-ott.com/live/user/pass/1905905.ts"; + let line = r#"#EXTINF:-1 tvg-id="skycrime.uk" tvg-name="SKY CRIME" group-title="Crime",SKY CRIME"#; + + let pli = process_header(&input, &video_suffixes, line, url.to_string()); + assert_eq!(pli.id, "1905905".intern()); // url_id used as fallback + assert_eq!(pli.epg_channel_id, Some("skycrime.uk".intern())); // tvg-id preserved for EPG + } + + #[test] + fn test_process_header_no_tvg_id_no_provider_id() { + let input = "test".intern(); + let video_suffixes = Vec::new(); + let url = "http://example.com/live/user/pass/12345.ts"; + let line = r#"#EXTINF:-1 tvg-name="Test Channel" group-title="Group",Test Channel"#; + + let pli = process_header(&input, &video_suffixes, line, url.to_string()); + assert_eq!(pli.id, "12345".intern()); // url_id as sole fallback + assert_eq!(pli.epg_channel_id, None); // no tvg-id -> no epg_channel_id + } + + #[test] + fn test_process_header_expiring_query_params_id_fallback() { + let input = "test".intern(); + let video_suffixes = Vec::new(); + let line = r#"#EXTINF:-1 tvg-name="Test Channel" group-title="Group",Test Channel"#; + + let pli = process_header( + &input, + &video_suffixes, + line, + "http://example.com/live/user/pass/1905905.ts?expires=1712345678&token=alpha".to_string(), + ); + assert_eq!(pli.id, "1905905".intern()); + assert_eq!(pli.epg_channel_id, None); + + let pli_variant = process_header( + &input, + &video_suffixes, + line, + "http://example.com/live/user/pass/1905905.ts?expires=1719999999&token=beta".to_string(), + ); + assert_eq!(pli_variant.id, "1905905".intern()); + assert_eq!(pli_variant.epg_channel_id, None); + } + #[test] fn test_process_header_xui_id() { let input = "hello".intern(); @@ -325,4 +452,41 @@ mod test { assert_eq!(pli.epg_channel_id, Some("abc-seven".intern())); // Should preserve original tvg-id assert_eq!(&*pli.group, "Sydney"); } -} \ No newline at end of file + + #[test] + fn test_process_header_unknown_numeric_id_field() { + let input = "test".intern(); + let video_suffixes = Vec::new(); + let url = "http://example.com/live/user/pass/99999.ts"; + // "stream-id" is an unknown field ending in "id" with a numeric value + let line = r#"#EXTINF:-1 stream-id="55555" tvg-name="Test Channel" group-title="Group",Test Channel"#; + + let pli = process_header(&input, &video_suffixes, line, url.to_string()); + assert_eq!(pli.id, "55555".intern()); // fallback numeric id field detected + assert_eq!(pli.epg_channel_id, None); + } + + #[test] + fn test_process_header_unknown_id_non_numeric_ignored() { + let input = "test".intern(); + let video_suffixes = Vec::new(); + let url = "http://example.com/live/user/pass/99999.ts"; + // "channel-id" has a non-numeric value, should be ignored as fallback + let line = r#"#EXTINF:-1 channel-id="abc-def" tvg-name="Test" group-title="G",Test"#; + + let pli = process_header(&input, &video_suffixes, line, url.to_string()); + assert_eq!(pli.id, "99999".intern()); // falls through to url_id + } + + #[test] + fn test_process_header_explicit_cuid_overrides_fallback_id() { + let input = "test".intern(); + let video_suffixes = Vec::new(); + let url = "http://example.com/live/user/pass/99999.ts"; + // Both CUID (explicit) and stream-id (fallback) present: CUID wins + let line = r#"#EXTINF:-1 stream-id="55555" CUID="77777" tvg-name="Test" group-title="G",Test"#; + + let pli = process_header(&input, &video_suffixes, line, url.to_string()); + assert_eq!(pli.id, "77777".intern()); // CUID takes priority + } +} diff --git a/backend/src/processing/processor/xtream_series.rs b/backend/src/processing/processor/xtream_series.rs index 9c6a42a76..044d71e5d 100644 --- a/backend/src/processing/processor/xtream_series.rs +++ b/backend/src/processing/processor/xtream_series.rs @@ -112,7 +112,7 @@ async fn playlist_resolve_series_info( let resolve_tmdb_enabled = fpl.input.has_flag(ConfigInputFlags::ResolveTmdb); let groups_to_add = if resolve_options.has_flag(ResolveOptionsFlags::Background) && ctx.metadata_manager.is_some() { - queue_background_series_info(ctx, fpl, filter, &resolve_options, skip_resolve, resolve_tmdb_enabled).await + queue_background_series_info(ctx, fpl, filter, &resolve_options, skip_resolve, resolve_tmdb_enabled) } else { process_immediate_series_info(ctx, fpl, filter, &resolve_options, skip_resolve, resolve_tmdb_enabled).await }; @@ -178,7 +178,7 @@ fn sync_resolved_series_properties(provider_fpl: &mut FetchedPlaylist<'_>, proce } } -async fn queue_background_series_info( +fn queue_background_series_info( ctx: &PlaylistProcessingContext, fpl: &mut FetchedPlaylist<'_>, filter: impl Fn(&PlaylistItem) -> bool, @@ -213,23 +213,26 @@ async fn queue_background_series_info( id: provider_id.clone(), reason: reasons, delay: resolve_options.resolve_delay, + source_last_modified: pli + .header + .additional_properties + .as_ref() + .and_then(StreamProperties::get_last_modified), }; - if !mgr.should_skip_enqueue(input_name_arc.clone(), &task).await { - 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); + 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 d7306b4c0..89c6f0321 100644 --- a/backend/src/processing/processor/xtream_vod.rs +++ b/backend/src/processing/processor/xtream_vod.rs @@ -134,7 +134,7 @@ async fn playlist_resolve_vod_info( let resolve_tmdb_enabled = fpl.input.has_flag(ConfigInputFlags::ResolveTmdb); if resolve_options.has_flag(ResolveOptionsFlags::Background) && ctx.metadata_manager.is_some() { - queue_background_vod_info(ctx, fpl, filter, &resolve_options, do_probe, resolve_tmdb_enabled).await; + queue_background_vod_info(ctx, fpl, filter, &resolve_options, do_probe, resolve_tmdb_enabled); } else { process_immediate_vod_info(ctx, fpl, filter, resolve_options, do_probe, resolve_tmdb_enabled).await; } @@ -416,7 +416,7 @@ fn check_resolve_tmdb( } } -async fn queue_background_vod_info( +fn queue_background_vod_info( ctx: &PlaylistProcessingContext, fpl: &mut FetchedPlaylist<'_>, filter: impl Fn(&PlaylistItem) -> bool, @@ -443,12 +443,16 @@ async 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.should_skip_enqueue(input.name.clone(), &task).await { - continue; - } - + let task = UpdateTask::ResolveVod { + id: provider_id.clone(), + reason: reasons, + delay: resolve_options.resolve_delay, + source_last_modified: pli + .header + .additional_properties + .as_ref() + .and_then(StreamProperties::get_last_modified), + }; if log_enabled!(Level::Debug) { let has_details = pli.has_details(); let (has_tmdb, has_date, has_video, has_audio) = diff --git a/backend/src/utils/network/xtream.rs b/backend/src/utils/network/xtream.rs index 9794204b4..8fe186a19 100644 --- a/backend/src/utils/network/xtream.rs +++ b/backend/src/utils/network/xtream.rs @@ -1,7 +1,7 @@ use crate::api::model::AppState; use crate::messaging::send_message; use crate::model::{is_input_expired, xtream_mapping_option_from_target_options, AppConfig, - ConfigInput, ConfigInputFlags, ConfigTarget, MessageContent, XtreamLoginInfo, XtreamTargetOutput}; + ConfigInput, ConfigInputFlags, ConfigTarget, MessageContent, XtreamTargetOutput}; use crate::model::{InputSource, ProxyUserCredentials}; use shared::model::ClusterSource; use crate::processing::parser::xtream; @@ -16,7 +16,7 @@ use log::{error, info, warn}; use shared::error::TuliproxError; use shared::model::{PlaylistEntry, PlaylistGroup, ProxyUserStatus, SeriesStreamProperties, StreamProperties, VideoStreamProperties, InputType, XtreamCluster, XtreamPlaylistItem, - XtreamSeriesInfo, XtreamVideoInfo, XtreamVideoInfoDoc}; + XtreamSeriesInfo, XtreamLoginInfo, XtreamVideoInfo, XtreamVideoInfoDoc}; use shared::utils::{extract_extension_from_url, get_i64_from_serde_value, get_string_from_serde_value, sanitize_sensitive_info, Internable, PROVIDER_SCHEME_PREFIX}; use std::collections::HashMap; use std::io::Error; @@ -30,7 +30,11 @@ const THREE_DAYS_IN_SECS: i64 = 3 * 24 * 60 * 60; #[inline] pub fn get_xtream_stream_url_base(url: &str, username: &str, password: &str) -> String { - format!("{url}/player_api.php?username={username}&password={password}") + let query = url::form_urlencoded::Serializer::new(String::new()) + .append_pair("username", username) + .append_pair("password", password) + .finish(); + format!("{url}/player_api.php?{query}") } pub fn get_xtream_player_api_action_url(input: &ConfigInput, action: &str) -> Option { @@ -276,7 +280,7 @@ const ACTIONS: [(XtreamCluster, &str, &str); 3] = [ (XtreamCluster::Video, crate::model::XC_ACTION_GET_VOD_CATEGORIES, crate::model::XC_ACTION_GET_VOD_STREAMS), (XtreamCluster::Series, crate::model::XC_ACTION_GET_SERIES_CATEGORIES, crate::model::XC_ACTION_GET_SERIES)]; -async fn xtream_login(app_config: &Arc, client: &reqwest::Client, input: &InputSource, username: &str) -> Result, TuliproxError> { +pub async fn xtream_login(app_config: &Arc, client: &reqwest::Client, input: &InputSource, username: &str) -> Result, TuliproxError> { let content = if let Ok(content) = request::get_input_json_content(app_config, client, input, None, false).await { content } else { diff --git a/bin/build_fe.sh b/bin/build_fe.sh index 70add1836..bf59d65e3 100755 --- a/bin/build_fe.sh +++ b/bin/build_fe.sh @@ -10,43 +10,6 @@ die() { exit 1 } -maybe_optimize_wasm() { - local wasm_opt_version - local wasm_file - local tmp_file - local -a wasm_files - - if ! command -v wasm-opt >/dev/null 2>&1; then - die "'wasm-opt' is required for frontend builds. Install the wasm tools with './bin/install_wasm_tools.sh 128' and add the returned bin path to PATH." - fi - - wasm_opt_version="$(wasm-opt --version 2>/dev/null || echo "unknown")" - - shopt -s nullglob - wasm_files=(dist/*_bg.wasm) - shopt -u nullglob - - if [ "${#wasm_files[@]}" -eq 0 ]; then - die "No frontend wasm artifact found for wasm-opt." - fi - - for wasm_file in "${wasm_files[@]}"; do - tmp_file="$(mktemp "${TMPDIR:-/tmp}/tuliprox-wasm-opt.XXXXXX.wasm")" - if wasm-opt -O --all-features --output="${tmp_file}" "${wasm_file}"; then - if mv "${tmp_file}" "${wasm_file}"; then - echo "✅ Optimized $(basename "${wasm_file}") with ${wasm_opt_version}" - continue - fi - - rm -f "${tmp_file}" - die "Failed to replace $(basename "${wasm_file}") with the optimized wasm artifact." - fi - - rm -f "${tmp_file}" - die "wasm-opt failed for $(basename "${wasm_file}") with ${wasm_opt_version}. Install the wasm tools with './bin/install_wasm_tools.sh 128' and add the returned bin path to PATH." - done -} - MODE="${1:-release}" case "${MODE}" in release) @@ -69,7 +32,6 @@ fi cd "${WORKING_DIR}/frontend" rm -rf dist env -u NO_COLOR RUSTFLAGS="--remap-path-prefix $HOME=~" trunk "${TRUNK_ARGS[@]}" -maybe_optimize_wasm if [ ! -d build/docs ]; then die "Documentation output directory 'frontend/build/docs' does not exist." diff --git a/docs/src/deployment.md b/docs/src/deployment.md index 7d5b39393..b19244e85 100644 --- a/docs/src/deployment.md +++ b/docs/src/deployment.md @@ -97,8 +97,8 @@ The repository ships helper scripts under `bin/`: `bin/build_fe.sh` is the shared entry point for docs plus frontend assets. -Wasm optimization is always part of the frontend build. -`bin/build_fe.sh` expects a compatible `wasm-opt` in `PATH`. +Wasm optimization is handled by Trunk during the build (via `data-wasm-opt` in `index.html`). +Trunk requires a compatible `wasm-opt` in `PATH`. Recommended local setup: diff --git a/frontend/index.html b/frontend/index.html index a9de1c07c..d16504af3 100644 --- a/frontend/index.html +++ b/frontend/index.html @@ -9,7 +9,7 @@ - + Tuliprox diff --git a/frontend/public/assets/i18n/en.json b/frontend/public/assets/i18n/en.json index 11a71077b..4eaca7676 100644 --- a/frontend/public/assets/i18n/en.json +++ b/frontend/public/assets/i18n/en.json @@ -1250,6 +1250,9 @@ "SOURCE_SELECTOR": { "MISSING_SELECTION": "No entry selected" }, + "SOURCE_EDITOR": { + "URL_USERNAME_AND_PASSWORD_MANDATORY": "Url, Username and Password are mandatory!" + }, "TARGET_NOT_EXISTS": "Target does not exist", "USER_DELETED": "User successfully deleted" }, diff --git a/frontend/scss/app/components/_input.scss b/frontend/scss/app/components/_input.scss index 0dc1d8c82..abd97c0ca 100644 --- a/frontend/scss/app/components/_input.scss +++ b/frontend/scss/app/components/_input.scss @@ -87,5 +87,8 @@ body[data-theme="bright"] input[type="date"]::-webkit-calendar-picker-indicator filter: none; } -.tp__input-date { -} \ No newline at end of file +.tp__input-date-tools { + input { + padding-right: 0; + } +} diff --git a/frontend/src/app/components/config/macros.rs b/frontend/src/app/components/config/macros.rs index 18c82eae6..a27b8316c 100644 --- a/frontend/src/app/components/config/macros.rs +++ b/frontend/src/app/components/config/macros.rs @@ -612,6 +612,28 @@ macro_rules! edit_field_date { }}; } +#[macro_export] +macro_rules! edit_field_exp_date { + ($instance:expr, $label:expr, $field:ident, $action:path, $tool_action:expr) => {{ + let instance = $instance.clone(); + + html! { +
+ <$crate::app::components::DateInputAction + label={$label} + name={stringify!($field)} + field_id={Some($crate::app::components::dto_field_id(&instance.form, stringify!($field)))} + value={instance.form.$field.clone()} + on_change={Callback::from(move |value: Option| { + instance.dispatch($action(value)); + })} + tool_action={$tool_action} + /> +
+ } + }}; +} + #[macro_export] macro_rules! edit_field_list { ($instance:expr, $label:expr, $field:ident, $action:path, $placeholder:expr) => {{ diff --git a/frontend/src/app/components/date_input.rs b/frontend/src/app/components/date_input.rs index dbc2bf6e1..8b41ba025 100644 --- a/frontend/src/app/components/date_input.rs +++ b/frontend/src/app/components/date_input.rs @@ -1,6 +1,6 @@ use crate::app::components::FieldLabel; use web_sys::HtmlInputElement; -use yew::{component, html, use_effect_with, Callback, Html, NodeRef, Properties, TargetCast}; +use yew::{classes, component, html, use_effect_with, Callback, Html, NodeRef, Properties, TargetCast}; #[derive(Properties, Clone, PartialEq, Debug)] pub struct DateInputProps { @@ -18,8 +18,28 @@ pub struct DateInputProps { pub on_change: Option>>, // None or Some(timestamp) } +#[derive(Properties, Clone, PartialEq, Debug)] +pub(crate) struct DateInputBaseProps { + #[prop_or_default] + pub name: String, + #[prop_or_default] + pub field_id: Option, + #[prop_or_default] + pub label: Option, + #[prop_or_default] + pub input_ref: Option, + #[prop_or_default] + pub value: Option, + #[prop_or_default] + pub on_change: Option>>, + #[prop_or_default] + pub tools: Html, + #[prop_or_default] + pub extra_class: Option<&'static str>, +} + #[component] -pub fn DateInput(props: &DateInputProps) -> Html { +pub(crate) fn DateInputBase(props: &DateInputBaseProps) -> Html { let local_ref = props.input_ref.clone().unwrap_or_default(); { @@ -28,7 +48,6 @@ pub fn DateInput(props: &DateInputProps) -> Html { use_effect_with(value, move |val| { if let Some(input) = local_ref.cast::() { if let Some(ts) = val { - // Timestamp -> yyyy-mm-dd if let Some(date) = chrono::DateTime::from_timestamp(*ts, 0) { input.set_value(&date.format("%Y-%m-%d").to_string()); } @@ -44,7 +63,7 @@ pub fn DateInput(props: &DateInputProps) -> Html { let onchange_cb = props.on_change.clone(); Callback::from(move |event: yew::events::Event| { if let Some(input) = event.target_dyn_into::() { - let value = input.value(); // "2025-08-22" + let value = input.value(); let ts = if value.is_empty() { None } else { @@ -61,7 +80,7 @@ pub fn DateInput(props: &DateInputProps) -> Html { }; html! { -
+
{ if let Some(label) = &props.label { html! { Html { name={props.name.clone()} onchange={handle_change} /> + {props.tools.clone()}
} } + +#[component] +pub fn DateInput(props: &DateInputProps) -> Html { + html! { + + } +} diff --git a/frontend/src/app/components/date_input_action.rs b/frontend/src/app/components/date_input_action.rs new file mode 100644 index 000000000..7d5b95181 --- /dev/null +++ b/frontend/src/app/components/date_input_action.rs @@ -0,0 +1,75 @@ +use crate::app::components::{date_input::DateInputBase, IconButton}; +use web_sys::MouseEvent; +use yew::{component, html, Callback, Html, NodeRef, Properties}; + +#[derive(Clone, PartialEq, Debug)] +pub struct ToolAction { + pub name: Option, + pub icon: String, + pub hint: Option, + pub class: Option, + pub onclick: Callback, +} + +#[derive(Properties, Clone, PartialEq, Debug)] +pub struct DateInputActionProps { + #[prop_or_default] + pub name: String, + #[prop_or_default] + pub field_id: Option, + #[prop_or_default] + pub label: Option, + #[prop_or_default] + pub input_ref: Option, + #[prop_or_default] + pub value: Option, + #[prop_or_default] + pub on_change: Option>>, + #[prop_or_default] + pub tool_action: Option, +} + +#[component] +pub fn DateInputAction(props: &DateInputActionProps) -> Html { + let handle_tool_click = { + let tool_action = props.tool_action.clone(); + Callback::from(move |(_, event): (String, MouseEvent)| { + if let Some(action) = tool_action.as_ref() { + action.onclick.emit(event); + } + }) + }; + + let tools = if let Some(action) = props.tool_action.as_ref() { + html! { +
+ +
+ } + } else { + Html::default() + }; + + html! { + + } +} diff --git a/frontend/src/app/components/mod.rs b/frontend/src/app/components/mod.rs index 2309386b4..982bf1204 100644 --- a/frontend/src/app/components/mod.rs +++ b/frontend/src/app/components/mod.rs @@ -14,6 +14,7 @@ mod csv_table; mod custom_dialog; mod dashboard; mod date_input; +mod date_input_action; mod drop_down_icon_button; mod hide_content; mod home; @@ -64,11 +65,11 @@ mod setup; pub(crate) use self::{ accordion::*, accordion_panel::*, authentication::*, breadcrumbs::*, card::*, cell_value::*, chip::*, - cluster_flags_input::*, collapse_panel::*, csv_table::*, custom_dialog::*, dashboard::*, drop_down_icon_button::*, - field_explanation::*, field_id::*, filter::*, hide_content::*, home::*, icon_button::*, key_value_editor::*, - loading_screen::*, login::*, no_content::*, panel::*, particle_flow_background::*, playlist::*, - radio_button_group::*, reveal_content::*, role_based_content::*, search::*, setup::*, sidebar::*, source_editor::*, - svg_icon::*, table::*, tabset::*, tag_list::*, text_button::*, textarea::*, title_card::*, toastr::*, - toggle_switch::*, userlist::*, websocket_status::*, + cluster_flags_input::*, collapse_panel::*, csv_table::*, custom_dialog::*, dashboard::*, date_input_action::*, + drop_down_icon_button::*, field_explanation::*, field_id::*, filter::*, hide_content::*, home::*, icon_button::*, + key_value_editor::*, loading_screen::*, login::*, no_content::*, panel::*, particle_flow_background::*, + playlist::*, radio_button_group::*, reveal_content::*, role_based_content::*, search::*, setup::*, sidebar::*, + source_editor::*, svg_icon::*, table::*, tabset::*, tag_list::*, text_button::*, textarea::*, title_card::*, + toastr::*, toggle_switch::*, userlist::*, websocket_status::*, }; pub use self::{confirm_dialog::*, content_dialog::*}; diff --git a/frontend/src/app/components/source_editor/alias_item_form.rs b/frontend/src/app/components/source_editor/alias_item_form.rs index 06e0df6e3..4d7eade31 100644 --- a/frontend/src/app/components/source_editor/alias_item_form.rs +++ b/frontend/src/app/components/source_editor/alias_item_form.rs @@ -1,12 +1,16 @@ use crate::{ - app::components::{Card, TextButton}, - edit_field_bool, edit_field_date, edit_field_number_i16, edit_field_number_u16, edit_field_text, + app::components::{Card, TextButton, ToolAction}, + edit_field_bool, edit_field_exp_date, edit_field_number_i16, edit_field_number_u16, edit_field_text, edit_field_text_option, generate_form_reducer, + hooks::use_service_context, i18n::use_translation, }; -use shared::{model::ConfigInputAliasDto, utils::Internable}; +use shared::{ + model::{ConfigInputAliasDto, InputType, XtreamLoginRequest}, + utils::Internable, +}; use std::sync::Arc; -use yew::prelude::*; +use yew::{platform::spawn_local, prelude::*}; const LABEL_ALIAS_NAME: &str = "LABEL.ALIAS_NAME"; const LABEL_URL: &str = "LABEL.URL"; @@ -36,6 +40,7 @@ generate_form_reducer!( pub struct AliasItemFormProps { pub on_submit: Callback, pub on_cancel: Callback<()>, + pub input_type: InputType, #[prop_or_default] pub initial: Option, } @@ -43,6 +48,7 @@ pub struct AliasItemFormProps { #[component] pub fn AliasItemForm(props: &AliasItemFormProps) -> Html { let translate = use_translation(); + let services = use_service_context(); let form_state: UseReducerHandle = use_reducer(|| AliasFormState { form: props.initial.clone().unwrap_or_else(|| ConfigInputAliasDto { @@ -58,6 +64,9 @@ pub fn AliasItemForm(props: &AliasItemFormProps) -> Html { }), modified: false, }); + let exp_date_loading = use_state(|| false); + let exp_date_request_in_flight = use_mut_ref(|| false); + let exp_date_request_token = use_mut_ref(|| 0_u64); let handle_submit = { let form_state = form_state.clone(); @@ -77,6 +86,86 @@ pub fn AliasItemForm(props: &AliasItemFormProps) -> Html { }) }; + let exp_date_tool_action = if props.input_type.is_xtream() { + let services = services.clone(); + let form_state = form_state.clone(); + let exp_date_loading = exp_date_loading.clone(); + let exp_date_request_in_flight = exp_date_request_in_flight.clone(); + let exp_date_request_token = exp_date_request_token.clone(); + let translate = translate.clone(); + + Some(ToolAction { + name: Some("RefreshAliasExpDate".to_string()), + icon: "Refresh".to_string(), + hint: Some(translate.t("LABEL.RESOLVE")), + class: (*exp_date_loading).then(|| "loading".to_string()), + onclick: Callback::from(move |_event: MouseEvent| { + if *exp_date_request_in_flight.borrow() { + return; + } + + let url = form_state.form.url.clone(); + let username = form_state.form.username.clone().unwrap_or_default(); + let password = form_state.form.password.clone().unwrap_or_default(); + + if url.trim().is_empty() || username.trim().is_empty() || password.trim().is_empty() { + services.toastr.error(translate.t("MESSAGES.SOURCE_EDITOR.URL_USERNAME_AND_PASSWORD_MANDATORY")); + return; + } + + *exp_date_request_in_flight.borrow_mut() = true; + let request_token = { + let mut token = exp_date_request_token.borrow_mut(); + *token += 1; + *token + }; + exp_date_loading.set(true); + let services = services.clone(); + let form_state = form_state.clone(); + let exp_date_loading = exp_date_loading.clone(); + let exp_date_request_in_flight = exp_date_request_in_flight.clone(); + let exp_date_request_token = exp_date_request_token.clone(); + let request = XtreamLoginRequest { url, username, password }; + + spawn_local(async move { + let current_snapshot = || { + ( + form_state.form.url.clone(), + form_state.form.username.clone().unwrap_or_default(), + form_state.form.password.clone().unwrap_or_default(), + ) + }; + match services.config.get_xtream_login_info(&request).await { + Ok(login_info) => { + if *exp_date_request_token.borrow() == request_token { + let snapshot_matches = current_snapshot() + == (request.url.clone(), request.username.clone(), request.password.clone()); + if snapshot_matches { + if let Some(exp_date) = login_info.exp_date { + form_state.dispatch(AliasFormAction::ExpDate(Some(exp_date))); + } else { + services.toastr.warning("No expiration date returned by provider"); + } + } + } + } + Err(err) => { + if *exp_date_request_token.borrow() == request_token { + services.toastr.error(err.to_string()); + } + } + } + if *exp_date_request_token.borrow() == request_token { + *exp_date_request_in_flight.borrow_mut() = false; + exp_date_loading.set(false); + } + }); + }), + }) + } else { + None + }; + html! { { edit_field_bool!(form_state, translate.t(LABEL_ENABLED), enabled, AliasFormAction::Enabled) } @@ -88,7 +177,7 @@ pub fn AliasItemForm(props: &AliasItemFormProps) -> Html { { edit_field_number_i16!(form_state, translate.t(LABEL_PRIORITY), priority, AliasFormAction::Priority) } { edit_field_number_u16!(form_state, translate.t(LABEL_MAX_CONNECTIONS), max_connections, AliasFormAction::MaxConnections) } - { edit_field_date!(form_state, translate.t(LABEL_EXP_DATE), exp_date, AliasFormAction::ExpDate) } + { edit_field_exp_date!(form_state, translate.t(LABEL_EXP_DATE), exp_date, AliasFormAction::ExpDate, exp_date_tool_action) }
Html { let translate = use_translation(); + let services = use_service_context(); let source_editor_ctx = use_context::(); let config_ctx = use_context::(); let fetch_methods = use_memo((), |_| { @@ -265,6 +268,9 @@ pub fn ConfigInputView(props: &ConfigInputViewProps) -> Html { let show_provider_form_state = use_state(|| false); let edit_alias = use_state(|| None::); let edit_provider = use_state(|| None::); + let exp_date_loading = use_state(|| false); + let exp_date_request_in_flight = use_mut_ref(|| false); + let exp_date_request_token = use_mut_ref(|| 0_u64); let staged_input_types = use_memo(staged_input_state.form.input_type, |input_type| { let default_it = input_type; @@ -751,6 +757,87 @@ pub fn ConfigInputView(props: &ConfigInputViewProps) -> Html { let render_input = || { let input_method_selection = Rc::new(vec![input_form_state.form.method.to_string()]); let input_form_state_disp = input_form_state.clone(); + let exp_date_tool_action = if input_form_state.form.input_type.is_xtream() { + let services = services.clone(); + let input_form_state = input_form_state.clone(); + let exp_date_loading = exp_date_loading.clone(); + let exp_date_request_in_flight = exp_date_request_in_flight.clone(); + let exp_date_request_token = exp_date_request_token.clone(); + let translate = translate.clone(); + + Some(ToolAction { + name: Some("RefreshExpDate".to_string()), + icon: "Refresh".to_string(), + hint: Some(translate.t(LABEL_RESOLVE)), + class: (*exp_date_loading).then(|| "loading".to_string()), + onclick: Callback::from(move |_event: MouseEvent| { + if *exp_date_request_in_flight.borrow() { + return; + } + + let url = input_form_state.form.url.clone(); + let username = input_form_state.form.username.clone().unwrap_or_default(); + let password = input_form_state.form.password.clone().unwrap_or_default(); + + if url.trim().is_empty() || username.trim().is_empty() || password.trim().is_empty() { + services + .toastr + .error(translate.t("MESSAGES.SOURCE_EDITOR.URL_USERNAME_AND_PASSWORD_MANDATORY")); + return; + } + + *exp_date_request_in_flight.borrow_mut() = true; + let request_token = { + let mut token = exp_date_request_token.borrow_mut(); + *token += 1; + *token + }; + exp_date_loading.set(true); + let services = services.clone(); + let input_form_state = input_form_state.clone(); + let exp_date_loading = exp_date_loading.clone(); + let exp_date_request_in_flight = exp_date_request_in_flight.clone(); + let exp_date_request_token = exp_date_request_token.clone(); + let request = XtreamLoginRequest { url, username, password }; + + spawn_local(async move { + let current_snapshot = || { + ( + input_form_state.form.url.clone(), + input_form_state.form.username.clone().unwrap_or_default(), + input_form_state.form.password.clone().unwrap_or_default(), + ) + }; + match services.config.get_xtream_login_info(&request).await { + Ok(login_info) => { + if *exp_date_request_token.borrow() == request_token { + let snapshot_matches = current_snapshot() + == (request.url.clone(), request.username.clone(), request.password.clone()); + if snapshot_matches { + if let Some(exp_date) = login_info.exp_date { + input_form_state.dispatch(ConfigInputFormAction::ExpDate(Some(exp_date))); + } else { + services.toastr.warning("No expiration date returned by provider"); + } + } + } + } + Err(err) => { + if *exp_date_request_token.borrow() == request_token { + services.toastr.error(err.to_string()); + } + } + } + if *exp_date_request_token.borrow() == request_token { + *exp_date_request_in_flight.borrow_mut() = false; + exp_date_loading.set(false); + } + }); + }), + }) + } else { + None + }; html! { @@ -766,7 +853,7 @@ pub fn ConfigInputView(props: &ConfigInputViewProps) -> Html { { edit_field_text_option!(input_form_state, translate.t(LABEL_PASSWORD), password, ConfigInputFormAction::Password, true) } { edit_field_number_u16!(input_form_state, translate.t(LABEL_MAX_CONNECTIONS), max_connections, ConfigInputFormAction::MaxConnections) } { edit_field_number_i16!(input_form_state, translate.t(LABEL_PRIORITY), priority, ConfigInputFormAction::Priority) } - { edit_field_date!(input_form_state, translate.t(LABEL_EXP_DATE), exp_date, ConfigInputFormAction::ExpDate) } + { edit_field_exp_date!(input_form_state, translate.t(LABEL_EXP_DATE), exp_date, ConfigInputFormAction::ExpDate, exp_date_tool_action) } { edit_field_text_option!(input_form_state, translate.t(LABEL_CACHE_DURATION), cache_duration, ConfigInputFormAction::CacheDuration) } { config_field_child!(translate.t(LABEL_FETCH_METHOD), "INPUT_FORM.FETCH_METHOD", { html! { @@ -798,6 +885,7 @@ pub fn ConfigInputView(props: &ConfigInputViewProps) -> Html { if *show_alias_form { Result { + request_post::<&XtreamLoginRequest, XtreamLoginInfo>(&self.xtream_login_info_path, request, None, None) + .await + .map(|response| response.unwrap_or_default()) + } + pub async fn save_config(&self, dto: ConfigDto) -> Result<(), Error> { let mut revision = self.revisions.borrow().main.clone(); if revision.is_none() { diff --git a/shared/src/model/playlist.rs b/shared/src/model/playlist.rs index b4259965e..53af96f4c 100644 --- a/shared/src/model/playlist.rs +++ b/shared/src/model/playlist.rs @@ -150,6 +150,8 @@ impl PlaylistItemType { ) } + pub fn is_live_adaptive(&self) -> bool { matches!(self, PlaylistItemType::LiveHls | PlaylistItemType::LiveDash) } + pub fn as_u8(self) -> u8 { self as u8 } pub fn as_str(&self) -> &'static str { diff --git a/shared/src/model/xtream.rs b/shared/src/model/xtream.rs index 9eac2ec5a..30eba1167 100644 --- a/shared/src/model/xtream.rs +++ b/shared/src/model/xtream.rs @@ -1,10 +1,36 @@ -use crate::utils::{ - arc_str_option_serde, arc_str_serde, deserialize_as_string_array, deserialize_json_as_opt_string, - deserialize_number_from_string, deserialize_number_from_string_or_zero, serialize_json_as_opt_string, +use crate::{ + model::ProxyUserStatus, + utils::{ + arc_str_option_serde, arc_str_serde, deserialize_as_string_array, deserialize_json_as_opt_string, + deserialize_number_from_string, deserialize_number_from_string_or_zero, serialize_json_as_opt_string, + }, }; use serde::{ser::SerializeMap, Deserialize, Deserializer, Serialize, Serializer}; use serde_json::Value; -use std::{collections::BTreeMap, sync::Arc}; +use std::{collections::BTreeMap, fmt, sync::Arc}; + +#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq)] +pub struct XtreamLoginInfo { + pub status: Option, + pub exp_date: Option, +} + +#[derive(Serialize, Deserialize, Default, Clone, PartialEq)] +pub struct XtreamLoginRequest { + pub url: String, + pub username: String, + pub password: String, +} + +impl fmt::Debug for XtreamLoginRequest { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("XtreamLoginRequest") + .field("url", &self.url) + .field("username", &"") + .field("password", &"") + .finish() + } +} #[derive(Serialize, Deserialize, Debug, Clone, PartialEq)] pub struct XtreamVideoInfoMovieData {