diff --git a/backend/src/api/api_utils.rs b/backend/src/api/api_utils.rs index 0a4f4c6b7..79ec157df 100644 --- a/backend/src/api/api_utils.rs +++ b/backend/src/api/api_utils.rs @@ -325,11 +325,17 @@ async fn resolve_streaming_strategy( fingerprint: &Fingerprint, input: &ConfigInput, force_provider: Option<&Arc>, + allow_provider_grace: bool, ) -> StreamingStrategy { // allocate a provider connection let provider_connection_handle = match force_provider { Some(provider) => app_state.active_provider.force_exact_acquire_connection(provider, &fingerprint.addr).await, - None => app_state.active_provider.acquire_connection(&input.name, &fingerprint.addr).await, + None => { + app_state + .active_provider + .acquire_connection_with_grace(&input.name, &fingerprint.addr, allow_provider_grace) + .await + } }; // panel_api provisioning/loading is handled later in the stream creation flow @@ -412,10 +418,11 @@ async fn create_stream_response_details( share_stream: bool, connection_permission: UserConnectionPermission, force_provider: Option<&Arc>, + allow_provider_grace: bool, virtual_id: VirtualId, ) -> Result { let mut streaming_strategy = - resolve_streaming_strategy(app_state, stream_url, fingerprint, input, force_provider).await; + resolve_streaming_strategy(app_state, stream_url, fingerprint, input, force_provider, allow_provider_grace).await; let mut grace_period_options = app_state.get_grace_options(); grace_period_options.period_millis = get_grace_period_millis( connection_permission, @@ -704,6 +711,18 @@ pub async fn force_provider_stream_response( // This is critical for users with a connection limit of 1 to avoid "Provider exhausted" or provider-side 502/509 errors during seeking. app_state.connection_manager.release_provider_connection(&user_session.addr).await; + let preferred_provider = if app_state.active_provider.is_exhausted(&user_session.provider).await { + debug_if_enabled!( + "Forced provider {} is exhausted for {}; acquiring next available provider instead", + sanitize_sensitive_info(&user_session.provider), + sanitize_sensitive_info(&user_session.token) + ); + None + } else { + Some(&user_session.provider) + }; + let allow_provider_grace = preferred_provider.is_some(); + let stream_details = match create_stream_response_details( app_state, &stream_options, @@ -714,7 +733,8 @@ pub async fn force_provider_stream_response( item_type, share_stream, connection_permission, - Some(&user_session.provider), + preferred_provider, + allow_provider_grace, stream_channel.virtual_id, ) .await @@ -859,6 +879,7 @@ pub async fn stream_response( share_stream, connection_permission, None, + true, stream_channel.virtual_id, ) .await diff --git a/backend/src/api/config_file.rs b/backend/src/api/config_file.rs index d983f4f8a..91f3f04fe 100644 --- a/backend/src/api/config_file.rs +++ b/backend/src/api/config_file.rs @@ -1,12 +1,10 @@ use crate::api::model::{update_app_state_config, update_app_state_sources, AppState, EventMessage}; -use crate::model::{Config, SourcesConfig}; +use crate::model::{Config, Mappings, SourcesConfig}; use crate::utils; use crate::utils::{ read_templates, prepare_sources_batch, read_config_file, read_mappings_file_unprepared, read_mappings_file_with_templates, read_sources_file, read_sources_file_from_path_with_templates, }; -use arc_swap::access::Access; -use arc_swap::ArcSwap; use log::{debug, error, info}; use shared::error::TuliproxError; use shared::model::{ConfigPaths, ConfigType, PatternTemplate}; @@ -23,11 +21,45 @@ pub enum ConfigFile { SourceFile, } -impl ConfigFile { - fn load_prepared_global_templates(app_state: &Arc) -> Result>, TuliproxError> { - let paths = > as Access>::load(&app_state.app_config.paths); - let config = > as Access>::load(&app_state.app_config.config); +#[derive(Debug)] +struct PreparedMappingsReload { + mapping_file_path: String, + mapping_files: Vec, + mappings: Mappings, +} +/// Fully prepared sources + mappings data, ready to apply without any I/O or fallible work. +struct PreparedSourcesReload { + sources: SourcesConfig, + mapping: Option, + sources_file: String, +} + +/// What dependent reload (if any) was prepared alongside a config change. +enum PreparedFollowUp { + Unchanged, + Mapping(Option), + Sources(PreparedSourcesReload), +} + +impl ConfigFile { + // ----------------------------------------------------------------- + // Template helpers + // ----------------------------------------------------------------- + + /// Load and merge global templates using the current app state. + fn load_prepared_global_templates(app_state: &Arc) -> Result>, TuliproxError> { + let paths = app_state.app_config.paths.load(); + let config = app_state.app_config.config.load(); + Self::load_prepared_global_templates_with_config(&paths, &config) + } + + /// Load and merge global templates using explicitly-provided config/paths. + /// Used in the prepare phase when the new config has not yet been applied to `app_state`. + fn load_prepared_global_templates_with_config( + paths: &ConfigPaths, + config: &Config, + ) -> Result>, TuliproxError> { let sources_inline_templates = read_sources_file(paths.sources_file_path.as_str(), false, false, None)?.templates; let mapping_inline_templates = if let Some(mapping_file_path) = paths.mapping_file_path.as_ref() { @@ -46,29 +78,57 @@ impl ConfigFile { Ok(template_bundle.prepared) } + // ----------------------------------------------------------------- + // Mapping prepare / apply + // ----------------------------------------------------------------- + + fn prepare_mapping_reload( + mapping_file_path: Option<&str>, + prepared_templates: Option<&[PatternTemplate]>, + ) -> Result, TuliproxError> { + let Some(mapping_file_path) = mapping_file_path else { + return Ok(None); + }; + + match read_mappings_file_with_templates(mapping_file_path, true, prepared_templates) { + Ok(Some((mapping_files, mappings_cfg))) => Ok(Some(PreparedMappingsReload { + mapping_file_path: mapping_file_path.to_string(), + mapping_files, + mappings: Mappings::from(&mappings_cfg), + })), + Ok(None) => { + info!("No mapping file loaded {mapping_file_path}"); + Ok(None) + } + Err(err) => { + error!("Failed to load mapping file {err}"); + Err(err) + } + } + } + + fn apply_mapping_reload( + app_state: &Arc, + prepared_mapping: Option, + ) { + if let Some(prepared_mapping) = prepared_mapping { + app_state + .app_config + .set_mappings(prepared_mapping.mapping_file_path.as_str(), &prepared_mapping.mappings); + for mapping_file in prepared_mapping.mapping_files { + info!("Loaded mapping file {}", mapping_file.display()); + } + } + } + fn load_mapping_with_templates( app_state: &Arc, prepared_templates: Option<&[PatternTemplate]>, ) -> Result<(), TuliproxError> { - let paths = > as Access>::load(&app_state.app_config.paths); - if let Some(mapping_file_path) = paths.mapping_file_path.as_ref() { - match read_mappings_file_with_templates(mapping_file_path, true, prepared_templates) { - Ok(Some((mapping_files, mappings_cfg))) => { - let mappings = crate::model::Mappings::from(&mappings_cfg); - app_state.app_config.set_mappings(mapping_file_path, &mappings); - for mapping_file in mapping_files { - info!("Loaded mapping file {}", mapping_file.display()); - } - } - Ok(None) => { - info!("No mapping file loaded {mapping_file_path}"); - } - Err(err) => { - error!("Failed to load mapping file {err}"); - return Err(err); - } - } - } + let paths = app_state.app_config.paths.load(); + let prepared_mapping = + Self::prepare_mapping_reload(paths.mapping_file_path.as_deref(), prepared_templates)?; + Self::apply_mapping_reload(app_state, prepared_mapping); Ok(()) } @@ -77,15 +137,66 @@ impl ConfigFile { Self::load_mapping_with_templates(app_state, prepared_templates.as_deref()) } + // ----------------------------------------------------------------- + // Sources prepare / apply + // ----------------------------------------------------------------- + + /// Prepare sources + mappings using explicitly-provided config/paths (prepare phase only — no state changes). + async fn prepare_sources_reload_with_config( + config: &Config, + paths: &ConfigPaths, + ) -> Result { + let sources_file = paths.sources_file_path.clone(); + let prepared_templates = Self::load_prepared_global_templates_with_config(paths, config)?; + let mut sources_dto = read_sources_file_from_path_with_templates( + &PathBuf::from(sources_file.as_str()), + true, + true, + config.get_hdhr_device_overview().as_ref(), + prepared_templates.as_deref(), + )?; + prepare_sources_batch(&mut sources_dto, true).await?; + let sources: SourcesConfig = SourcesConfig::try_from(sources_dto)?; + let prepared_mapping = + Self::prepare_mapping_reload(paths.mapping_file_path.as_deref(), prepared_templates.as_deref())?; + Ok(PreparedSourcesReload { sources, mapping: prepared_mapping, sources_file }) + } + + /// Apply a fully-prepared sources reload to app state (infallible under normal conditions). + async fn apply_sources_reload( + app_state: &Arc, + prepared: PreparedSourcesReload, + ) -> Result<(), TuliproxError> { + update_app_state_sources(app_state, prepared.sources).await?; + Self::apply_mapping_reload(app_state, prepared.mapping); + info!("Loaded sources file {}", prepared.sources_file); + Ok(()) + } + + /// Public entry-point for Sources / Template file-watch events. + /// Reads config from current app state (config itself has not changed). + pub async fn load_sources(app_state: &Arc) -> Result<(), TuliproxError> { + let paths = app_state.app_config.paths.load(); + let config = app_state.app_config.config.load(); + // PREPARE — no state mutations + let prepared = Self::prepare_sources_reload_with_config(&config, &paths).await?; + // APPLY — only reached when preparation fully succeeded + Self::apply_sources_reload(app_state, prepared).await + } + + // ----------------------------------------------------------------- + // API proxy + // ----------------------------------------------------------------- + async fn load_api_proxy(app_state: &Arc) -> Result<(), TuliproxError> { match utils::read_api_proxy_config(&app_state.app_config, true).await { Ok(Some(api_proxy)) => { app_state.app_config.set_api_proxy(api_proxy)?; - let paths = > as Access>::load(&app_state.app_config.paths); + let paths = app_state.app_config.paths.load(); info!("Loaded Api Proxy File: {:?}", &paths.api_proxy_file_path); } Ok(None) => { - let paths = > as Access>::load(&app_state.app_config.paths); + let paths = app_state.app_config.paths.load(); info!("Could not load Api Proxy File: {:?}", &paths.api_proxy_file_path); } Err(err) => { @@ -96,10 +207,16 @@ impl ConfigFile { Ok(()) } + // ----------------------------------------------------------------- + // Main config (config.yml) — full prepare-then-apply + // ----------------------------------------------------------------- + async fn load_config(app_state: &Arc) -> Result<(), TuliproxError> { - let paths = > as Access>::load(&app_state.app_config.paths); - let config_file = paths.config_file_path.as_str(); - let config_dto = read_config_file(config_file, true, true)?; + let paths = app_state.app_config.paths.load(); + let config_file = paths.config_file_path.clone(); + + // ── Parse ──────────────────────────────────────────────────── + let config_dto = read_config_file(config_file.as_str(), true, true)?; let default_mapping_path = utils::get_default_mappings_path(paths.config_path.as_str()); let current_mapping_path = paths @@ -127,37 +244,88 @@ impl ConfigFile { let mut config: Config = Config::from(config_dto); config.prepare(paths.config_path.as_str()).await?; - update_app_state_config(app_state, config).await?; - info!("Loaded config file {config_file}"); - if template_changed { - Self::load_sources(app_state).await?; + + // Compute effective runtime paths for the NEW config before apply. + // This ensures prepare-phase reads/validates against the same files that will be active after apply. + let mut effective_paths = paths.as_ref().clone(); + effective_paths.mapping_file_path = Some(next_mapping_path.clone()); + effective_paths.template_file_path = Some(next_template_path.clone()); + + // ── PREPARE PHASE ──────────────────────────────────────────── + // All dependent data is loaded/validated here, using the NEW config values + // but without touching any live state. If anything fails, we return an error + // and the currently-running state remains completely unchanged. + let follow_up: PreparedFollowUp = if template_changed { + // Template path changed → sources depend on new templates, reload everything. + let prepared = Self::prepare_sources_reload_with_config(&config, &effective_paths).await?; + PreparedFollowUp::Sources(prepared) } else if mapping_changed { - Self::load_mapping(app_state)?; + // Only mapping path changed; templates are the same → load templates once. + let prepared_templates = + Self::load_prepared_global_templates_with_config(&effective_paths, &config)?; + let prepared = Self::prepare_mapping_reload( + effective_paths.mapping_file_path.as_deref(), + prepared_templates.as_deref(), + )?; + PreparedFollowUp::Mapping(prepared) + } else { + PreparedFollowUp::Unchanged + }; + + let previous_config: Config = (*app_state.app_config.config.load_full()).clone(); + let previous_sources: SourcesConfig = (*app_state.app_config.sources.load_full()).clone(); + let previous_forced_targets = app_state.forced_targets.load_full(); + + // ── APPLY PHASE ────────────────────────────────────────────── + // All preparation succeeded — safe to update live state now. + if let Err(err) = update_app_state_config(app_state, config).await { + error!("Failed to apply config reload: {err}. Attempting config rollback."); + if let Err(rollback_err) = update_app_state_config(app_state, previous_config).await { + error!("Failed to rollback config after reload failure: {rollback_err}"); + error!( + "Config reload and rollback both failed; runtime state may be inconsistent. Please restart the service." + ); + } + return Err(err); } + + let follow_up_result: Result<(), TuliproxError> = match follow_up { + PreparedFollowUp::Unchanged => Ok(()), + PreparedFollowUp::Mapping(prepared) => { + Self::apply_mapping_reload(app_state, prepared); + Ok(()) + } + PreparedFollowUp::Sources(prepared) => Self::apply_sources_reload(app_state, prepared).await, + }; + + if let Err(err) = follow_up_result { + error!("Failed to apply dependent reload after config update: {err}. Attempting rollback."); + + if let Err(rollback_err) = update_app_state_config(app_state, previous_config).await { + error!("Failed to rollback config after dependent reload failure: {rollback_err}"); + error!( + "Dependent reload and config rollback both failed; runtime state may be inconsistent. Please restart the service." + ); + } + + app_state.forced_targets.store(previous_forced_targets); + if let Err(rollback_err) = update_app_state_sources(app_state, previous_sources).await { + error!("Failed to rollback sources after dependent reload failure: {rollback_err}"); + error!( + "Source rollback failed after dependent reload error; runtime state may be inconsistent. Please restart the service." + ); + } + + return Err(err); + } + + info!("Loaded config file {config_file}"); Ok(()) } - pub async fn load_sources(app_state: &Arc) -> Result<(), TuliproxError> { - let paths = > as Access>::load(&app_state.app_config.paths); - let sources_file = paths.sources_file_path.as_str(); - let prepared_templates = Self::load_prepared_global_templates(app_state)?; - let mut sources_dto = { - let config = > as Access>::load(&app_state.app_config.config); - read_sources_file_from_path_with_templates( - &PathBuf::from(sources_file), - true, - true, - config.get_hdhr_device_overview().as_ref(), - prepared_templates.as_deref(), - )? - }; - prepare_sources_batch(&mut sources_dto, true).await?; - let sources: SourcesConfig = SourcesConfig::try_from(sources_dto)?; - update_app_state_sources(app_state, sources).await?; - info!("Loaded sources file {sources_file}"); - // mappings are not stored, so we need to reload and apply them if sources change. - Self::load_mapping_with_templates(app_state, prepared_templates.as_deref()) - } + // ----------------------------------------------------------------- + // Reload dispatcher + // ----------------------------------------------------------------- async fn reload_source_file(app_state: &Arc) -> Result<(), TuliproxError> { // TODO selective update and not complete sources update ? diff --git a/backend/src/api/model/active_provider_manager.rs b/backend/src/api/model/active_provider_manager.rs index f0860a1d9..99f58ddf1 100644 --- a/backend/src/api/model/active_provider_manager.rs +++ b/backend/src/api/model/active_provider_manager.rs @@ -145,16 +145,28 @@ impl ActiveProviderManager { priority: i8, ) -> Option { // 1. Try to acquire directly - let allocation = if force { - self.providers.force_exact_acquire_connection(provider_or_input_name).await + let (allow_grace, allocation) = if force { + ( + true, + self.providers + .force_exact_acquire_connection(provider_or_input_name) + .await, + ) } else { match allow_grace_override { - Some(allow_grace) => { + Some(allow_grace) => ( + allow_grace, self.providers - .acquire_connection_with_grace_override(provider_or_input_name, allow_grace) - .await - } - None => self.providers.acquire_connection(provider_or_input_name).await, + .acquire_connection_with_grace_override( + provider_or_input_name, + allow_grace, + ) + .await, + ), + None => ( + true, + self.providers.acquire_connection(provider_or_input_name).await, + ), } }; @@ -164,7 +176,9 @@ impl ActiveProviderManager { // 2. If exhausted, try preemption (kick lower priority connection) if !force { - if let Some(preempted_alloc) = self.try_preempt_connection(provider_or_input_name, priority).await { + if let Some(preempted_alloc) = + self.try_preempt_connection(provider_or_input_name, priority, allow_grace).await + { return self.register_allocation(preempted_alloc, addr, priority).await; } } @@ -195,7 +209,12 @@ impl ActiveProviderManager { Some(ProviderHandle::new(*addr, allocation_id, allocation, Some(cancel_token))) } - async fn try_preempt_connection(&self, input_name: &Arc, new_priority: i8) -> Option { + async fn try_preempt_connection( + &self, + input_name: &Arc, + new_priority: i8, + allow_grace: bool, + ) -> Option { // Victim: (addr, alloc_id, priority, created_at, is_shared) let mut victim: Option<(ClientConnectionId, AllocationId, i8, Instant, bool)> = None; @@ -272,8 +291,11 @@ impl ActiveProviderManager { }; self.release_handle(&handle).await; - // Now try acquire again - let allocation = self.providers.acquire_connection(input_name).await; + // Now try acquire again preserving the original grace policy. + let allocation = self + .providers + .acquire_connection_with_grace_override(input_name, allow_grace) + .await; if !matches!(allocation, ProviderAllocation::Exhausted) { return Some(allocation); } @@ -293,6 +315,16 @@ impl ActiveProviderManager { self.acquire_connection_inner(input_name, addr, false, None, DEFAULT_USER_PRIORITY).await } + /// Acquire a provider connection while explicitly controlling provider-side grace allocations. + pub async fn acquire_connection_with_grace( + &self, + input_name: &Arc, + addr: &SocketAddr, + allow_grace: bool, + ) -> Option { + self.acquire_connection_inner(input_name, addr, false, Some(allow_grace), DEFAULT_USER_PRIORITY).await + } + /// Acquire a provider connection while optionally disabling provider grace allocations. pub async fn acquire_connection_for_probe( &self, @@ -315,6 +347,10 @@ impl ActiveProviderManager { self.providers.is_over_limit(provider_name).await } + pub async fn is_exhausted(&self, provider_name: &Arc) -> bool { + self.providers.is_exhausted(provider_name).await + } + pub async fn release_connection(&self, addr: &SocketAddr) { // Single connection - all index updates in one lock scope let single_allocations = { @@ -532,3 +568,169 @@ impl ActiveProviderManager { self.providers.active_connection_count().await } } + +#[cfg(test)] +mod tests { + use super::ActiveProviderManager; + use crate::api::model::EventManager; + use crate::model::{AppConfig, Config, ConfigInput, ConfigInputAlias, SourcesConfig}; + use crate::utils::FileLockManager; + use arc_swap::{ArcSwap, ArcSwapOption}; + use shared::model::{ConfigPaths, InputFetchMethod, InputType}; + use shared::utils::Internable; + use std::collections::HashMap; + use std::net::SocketAddr; + use std::sync::Arc; + + fn create_test_app_config_with_dual_provider_pool() -> AppConfig { + let input = Arc::new(ConfigInput { + id: 1, + name: "provider_1".intern(), + input_type: InputType::Xtream, + headers: HashMap::default(), + url: "http://provider-1.example".to_string(), + username: Some("user1".to_string()), + password: Some("pass1".to_string()), + enabled: true, + priority: 0, + max_connections: 1, + method: InputFetchMethod::default(), + aliases: Some(vec![ConfigInputAlias { + id: 2, + name: "provider_2".intern(), + url: "http://provider-2.example".to_string(), + username: Some("user2".to_string()), + password: Some("pass2".to_string()), + priority: 1, + max_connections: 1, + exp_date: None, + enabled: true, + }]), + ..ConfigInput::default() + }); + + let sources = SourcesConfig { + inputs: vec![input], + ..SourcesConfig::default() + }; + + AppConfig { + config: Arc::new(ArcSwap::from_pointee(Config::default())), + sources: Arc::new(ArcSwap::from_pointee(sources)), + hdhomerun: Arc::new(ArcSwapOption::default()), + api_proxy: Arc::new(ArcSwapOption::default()), + file_locks: Arc::new(FileLockManager::default()), + paths: Arc::new(ArcSwap::from_pointee(ConfigPaths { + config_path: String::new(), + config_file_path: String::new(), + sources_file_path: String::new(), + mapping_file_path: None, + mapping_files_used: None, + template_file_path: None, + template_files_used: None, + api_proxy_file_path: String::new(), + custom_stream_response_path: None, + })), + custom_stream_response: Arc::new(ArcSwapOption::default()), + access_token_secret: [0; 32], + encrypt_secret: [0; 16], + ffprobe_available: Arc::default(), + } + } + + #[tokio::test] + async fn test_force_session_fallback_uses_different_provider_when_current_is_busy() { + let app_cfg = create_test_app_config_with_dual_provider_pool(); + let event_manager = Arc::new(EventManager::new()); + let manager = ActiveProviderManager::new(&app_cfg, &event_manager); + + let input_name = "provider_1".intern(); + let client_1_addr: SocketAddr = "127.0.0.1:41001".parse().unwrap(); + let client_2_addr: SocketAddr = "127.0.0.1:41002".parse().unwrap(); + + // Step 1: Client1 starts movie -> provider_1 + let first_alloc = manager.acquire_connection(&input_name, &client_1_addr).await.expect("client1 initial allocation"); + assert_eq!( + first_alloc.allocation.get_provider_name().as_deref(), + Some(input_name.as_ref()) + ); + + // Step 2: Client1 stops -> release provider_1 + manager.release_connection(&client_1_addr).await; + + // Step 3: Client2 starts live -> provider_1 + let live_alloc = manager.acquire_connection(&input_name, &client_2_addr).await.expect("client2 live allocation"); + let busy_provider = live_alloc.allocation.get_provider_name().expect("provider name expected"); + assert_eq!(busy_provider.as_ref(), input_name.as_ref()); + assert!(manager.is_exhausted(&busy_provider).await); + + // Step 4: Client1 restarts same movie. + // This emulates force-session fallback path by acquiring without provider grace. + let fallback_alloc = manager + .acquire_connection_with_grace(&input_name, &client_1_addr, false) + .await + .expect("client1 fallback allocation without grace"); + let fallback_provider = fallback_alloc + .allocation + .get_provider_name() + .expect("fallback provider expected"); + + assert_ne!(fallback_provider.as_ref(), busy_provider.as_ref()); + assert_eq!(fallback_provider.as_ref(), "provider_2"); + + manager.release_connection(&client_1_addr).await; + manager.release_connection(&client_2_addr).await; + } + + #[tokio::test] + async fn test_seek_reacquire_stays_on_same_provider_account_until_stop() { + let app_cfg = create_test_app_config_with_dual_provider_pool(); + let event_manager = Arc::new(EventManager::new()); + let manager = ActiveProviderManager::new(&app_cfg, &event_manager); + + let input_name = "provider_1".intern(); + let client_1_addr: SocketAddr = "127.0.0.1:42001".parse().unwrap(); + let client_2_addr: SocketAddr = "127.0.0.1:42002".parse().unwrap(); + + // Initial playback for client1. + let first_alloc = manager + .acquire_connection(&input_name, &client_1_addr) + .await + .expect("client1 initial allocation"); + let pinned_provider = first_alloc + .allocation + .get_provider_name() + .expect("provider name expected"); + assert_eq!(pinned_provider.as_ref(), "provider_1"); + + // Another client occupies the alternate account while client1 keeps seeking. + let second_alloc = manager + .acquire_connection(&input_name, &client_2_addr) + .await + .expect("client2 allocation"); + let second_provider = second_alloc + .allocation + .get_provider_name() + .expect("provider name expected"); + assert_eq!(second_provider.as_ref(), "provider_2"); + + // Simulate repeated seek/range reconnects for client1: + // release old connection for the same client, then force exact pinned provider. + for _ in 0..3 { + manager.release_connection(&client_1_addr).await; + let seek_alloc = manager + .force_exact_acquire_connection(&pinned_provider, &client_1_addr) + .await + .expect("seek reacquire should stay on pinned provider"); + let seek_provider = seek_alloc + .allocation + .get_provider_name() + .expect("provider name expected"); + assert_eq!(seek_provider.as_ref(), pinned_provider.as_ref()); + } + + // Stream stop / cleanup. + manager.release_connection(&client_1_addr).await; + manager.release_connection(&client_2_addr).await; + } +} diff --git a/backend/src/api/model/provider_lineup_manager.rs b/backend/src/api/model/provider_lineup_manager.rs index 368a4e43d..bb438d995 100644 --- a/backend/src/api/model/provider_lineup_manager.rs +++ b/backend/src/api/model/provider_lineup_manager.rs @@ -882,6 +882,15 @@ impl ProviderLineupManager { } } + pub async fn is_exhausted(&self, provider_name: &Arc) -> bool { + let providers = self.providers.load(); + if let Some((_, config)) = Self::get_provider_config_by_name(provider_name, &providers) { + config.is_exhausted().await + } else { + false + } + } + pub fn is_provider_for_input(&self, provider_name: &str, input_name: &str) -> bool { let lineups = self.providers.load(); if let Some((lineup, _)) = Self::get_provider_config_by_name(&provider_name.into(), &lineups) { @@ -1211,4 +1220,4 @@ mod tests { assert_eq!(available_count.load(Ordering::Acquire), 0); assert_eq!(grace_period_count.load(Ordering::Acquire), 0); } -} \ No newline at end of file +} diff --git a/backend/src/messaging.rs b/backend/src/messaging.rs index 9378143d6..16f3184c5 100644 --- a/backend/src/messaging.rs +++ b/backend/src/messaging.rs @@ -6,7 +6,7 @@ use log::{debug, error}; use reqwest::{header, Method}; use serde_json::json; use shared::model::{InputFetchMethod, MsgKind}; -use shared::utils::{json_str_to_markdown, Internable}; +use shared::utils::{escape_markdown_v2, json_str_to_markdown, Internable}; use std::borrow::Cow; use std::collections::HashMap; use std::str::FromStr; @@ -164,6 +164,7 @@ async fn send_telegram_message(app_config: &Arc, client: &reqwest::Cl if let Some(telegram) = &messaging.telegram { let kind = content.kind(); let template = telegram.templates.get(&kind).map(String::as_str); + let has_template = template.is_some(); let msg = if let Some(templ) = template { render_template(app_config, client, Some(templ), content).await @@ -187,8 +188,12 @@ async fn send_telegram_message(app_config: &Arc, client: &reqwest::Cl if let Ok(md) = json_str_to_markdown(&msg) { (Cow::Owned(md), Some(SendMessageOption { parse_mode: SendMessageParseMode::MarkdownV2 })) } else { - // If it's already rendered markdown (from template), just use it - (Cow::Borrowed(&msg), Some(SendMessageOption { parse_mode: SendMessageParseMode::MarkdownV2 })) + // Keep template markdown as-is, but escape plain text to avoid MarkdownV2 parse errors. + if has_template { + (Cow::Borrowed(&msg), Some(SendMessageOption { parse_mode: SendMessageParseMode::MarkdownV2 })) + } else { + (Cow::Owned(escape_markdown_v2(&msg)), Some(SendMessageOption { parse_mode: SendMessageParseMode::MarkdownV2 })) + } } } else { (Cow::Borrowed(&msg), None) diff --git a/backend/src/processing/parser/xtream.rs b/backend/src/processing/parser/xtream.rs index e57e74dcd..6dcebf1a5 100644 --- a/backend/src/processing/parser/xtream.rs +++ b/backend/src/processing/parser/xtream.rs @@ -65,6 +65,7 @@ pub fn create_xtream_series_episode_url(url: &str, username: &str, password: &st pub fn parse_xtream_series_info(parent_uuid: &UUIDType, series_info: &SeriesStreamProperties, group_title: &str, series_name: &Arc, input: &ConfigInput, // Add series_release_date parameter series_release_date: Option<&Arc>, + parent_source_ordinal: u32, ) -> Option> { let url = input.url.as_str(); let (username, password) = ( @@ -73,7 +74,8 @@ pub fn parse_xtream_series_info(parent_uuid: &UUIDType, series_info: &SeriesStre ); if let Some(episodes) = series_info.details.as_ref().and_then(|d| d.episodes.as_ref()) { - let result: Vec = episodes.iter().map(|episode| { + let base_source_ordinal = if parent_source_ordinal == 0 { 1 } else { parent_source_ordinal }; + let result: Vec = episodes.iter().enumerate().map(|(index, episode)| { let episode_id = episode.id.to_string(); let episode_url = create_xtream_series_episode_url(url, username, password, episode); @@ -100,6 +102,7 @@ pub fn parse_xtream_series_info(parent_uuid: &UUIDType, series_info: &SeriesStre additional_properties: Some(StreamProperties::Episode(Box::new(episode_info))), category_id: 0, input_name: input.name.intern(), + source_ordinal: base_source_ordinal.saturating_add(u32::try_from(index).unwrap_or(u32::MAX)), ..Default::default() } } @@ -390,9 +393,15 @@ where #[cfg(test)] mod tests { + use super::parse_xtream_series_info; use crate::processing::parser::xtream::map_to_xtream_streams; + use crate::model::ConfigInput; use crate::utils::async_file_reader; - use shared::model::{XtreamCluster, XtreamSeriesInfo}; + use shared::model::{ + UUIDType, + SeriesStreamDetailEpisodeProperties, SeriesStreamDetailProperties, SeriesStreamProperties, + XtreamCluster, XtreamSeriesInfo, + }; use shared::utils::Internable; use std::fs; @@ -444,4 +453,48 @@ mod tests { serde::Deserializer::deserialize_any(&mut deserializer, visitor).unwrap(); assert_eq!(count, 1); } + + #[test] + fn test_parse_xtream_series_info_sets_episode_source_ordinals_from_parent() { + let input = ConfigInput { + name: "input".intern(), + url: "http://provider.example".to_string(), + username: Some("user".to_string()), + password: Some("pass".to_string()), + ..ConfigInput::default() + }; + + let episode_1: SeriesStreamDetailEpisodeProperties = serde_json::from_str( + r#"{"id":101,"episode_num":1,"season":1,"title":"S01E01","container_extension":"mp4"}"#, + ) + .unwrap(); + let episode_2: SeriesStreamDetailEpisodeProperties = serde_json::from_str( + r#"{"id":102,"episode_num":2,"season":1,"title":"S01E02","container_extension":"mp4"}"#, + ) + .unwrap(); + + let series_props = SeriesStreamProperties { + details: Some(SeriesStreamDetailProperties { + year: None, + seasons: None, + episodes: Some(vec![episode_1, episode_2]), + }), + ..SeriesStreamProperties::default() + }; + let parent_uuid = UUIDType::from_valid_uuid("parent_uuid"); + + let episodes = parse_xtream_series_info( + &parent_uuid, + &series_props, + "Series Group", + &"Series Name".intern(), + &input, + None, + 42, + ) + .expect("episodes should be parsed"); + + assert_eq!(episodes[0].header.source_ordinal, 42); + assert_eq!(episodes[1].header.source_ordinal, 43); + } } diff --git a/backend/src/processing/processor/sort.rs b/backend/src/processing/processor/sort.rs index d8ae90698..af81db91a 100644 --- a/backend/src/processing/processor/sort.rs +++ b/backend/src/processing/processor/sort.rs @@ -3,6 +3,7 @@ use shared::foundation::ValueProvider; use shared::model::{PlaylistGroup, SortOrder, SortTarget}; use std::cmp::Ordering; use std::sync::Arc; +use crate::utils::normalized_source_ordinal; fn apply_sort_order(order: SortOrder, ordering: Ordering) -> Ordering { match (order, ordering) { @@ -247,8 +248,16 @@ fn playlist_comparator( macro_rules! sort_groups_by_source_order { ($groups: ident) => { $groups.sort_by(|a, b| { - let order1 = a.channels.first().as_ref().map_or(u32::MAX, |c| c.header.source_ordinal); - let order2 = b.channels.first().as_ref().map_or(u32::MAX, |c| c.header.source_ordinal); + let order1 = a + .channels + .first() + .as_ref() + .map_or(u32::MAX, |c| normalized_source_ordinal(c.header.source_ordinal)); + let order2 = b + .channels + .first() + .as_ref() + .map_or(u32::MAX, |c| normalized_source_ordinal(c.header.source_ordinal)); order1.cmp(&order2) }); }; @@ -288,7 +297,9 @@ pub(in crate::processing::processor) fn sort_playlist( ) -> bool { let Some(sort) = &target.sort else { for group in &mut *playlist { - group.channels.sort_by_key(|a| a.header.source_ordinal); + group + .channels + .sort_by_key(|a| normalized_source_ordinal(a.header.source_ordinal)); } sort_groups_by_source_order!(playlist); return true; @@ -325,8 +336,14 @@ fn sort_groups(groups: &mut Vec, rules: &[ConfigSortRule], match_ return ord; } - let order1 = groups[*left_idx].channels.first().map_or(u32::MAX, |c| c.header.source_ordinal); - let order2 = groups[*right_idx].channels.first().map_or(u32::MAX, |c| c.header.source_ordinal); + let order1 = groups[*left_idx] + .channels + .first() + .map_or(u32::MAX, |c| normalized_source_ordinal(c.header.source_ordinal)); + let order2 = groups[*right_idx] + .channels + .first() + .map_or(u32::MAX, |c| normalized_source_ordinal(c.header.source_ordinal)); order1.cmp(&order2) }); @@ -343,7 +360,9 @@ fn sort_channels_in_groups(groups: &mut [PlaylistGroup], rules: &[ConfigSortRule if channel_rules.is_empty() { for group in groups { - group.channels.sort_by_key(|a| a.header.source_ordinal); + group + .channels + .sort_by_key(|a| normalized_source_ordinal(a.header.source_ordinal)); } return; } @@ -359,7 +378,8 @@ fn sort_channels_in_groups(groups: &mut [PlaylistGroup], rules: &[ConfigSortRule return ord; } - group.channels[*left_idx].header.source_ordinal.cmp(&group.channels[*right_idx].header.source_ordinal) + normalized_source_ordinal(group.channels[*left_idx].header.source_ordinal) + .cmp(&normalized_source_ordinal(group.channels[*right_idx].header.source_ordinal)) }); reorder_by_indices(&mut group.channels, &channel_indices); @@ -684,4 +704,47 @@ mod tests { let expected = vec!["A-1-y", "A-2-x"].into_iter().map(Into::into).collect::>>(); assert_eq!(expected, sorted); } + + #[test] + fn test_sort_groups_normalized_source_ordinal_tiebreaker_pushes_zero_last() { + let group_zero = make_group( + 1, + "zero", + vec![PlaylistItem { + header: PlaylistItemHeader { + title: Arc::from("Same Caption"), + source_ordinal: 0, + ..Default::default() + }, + }], + ); + let group_non_zero = make_group( + 2, + "non-zero", + vec![PlaylistItem { + header: PlaylistItemHeader { + title: Arc::from("Same Caption"), + source_ordinal: 7, + ..Default::default() + }, + }], + ); + + let group_rule = ConfigSortRule { + target: SortTarget::Group, + field: ItemField::Caption, + order: SortOrder::Asc, + // Both groups have the same sequence/rule priority and same caption value. + sequence: Some(vec![shared::model::REGEX_CACHE.get_or_compile(r"^Same Caption$").unwrap()]), + filter: Filter::default(), + }; + + let mut groups = vec![group_zero, group_non_zero]; + sort_groups(&mut groups, &[group_rule], false); + + assert_eq!(groups[0].title.as_ref(), "non-zero"); + assert_eq!(groups[0].channels[0].header.source_ordinal, 7); + assert_eq!(groups[1].title.as_ref(), "zero"); + assert_eq!(groups[1].channels[0].header.source_ordinal, 0); + } } diff --git a/backend/src/processing/processor/xtream_series.rs b/backend/src/processing/processor/xtream_series.rs index adfe3d88f..cdb2c7023 100644 --- a/backend/src/processing/processor/xtream_series.rs +++ b/backend/src/processing/processor/xtream_series.rs @@ -395,6 +395,7 @@ fn expand_series_item(pli: &PlaylistItem, input: &ConfigInput) -> Option = groups_map.into_values().collect(); - groups.sort_by_key(|group| { - group.channels.first().map_or(u32::MAX, |c| c.header.source_ordinal) - }); + groups.sort_by_key(|group| group.channels.first().map_or(u32::MAX, |c| normalized_source_ordinal(c.header.source_ordinal))); groups } @@ -382,14 +382,14 @@ macro_rules! impl_single_file_disk_source { } // Sort channels within each group for group in groups_map.values_mut() { - group.channels.sort_by_key(|item| item.header.source_ordinal); + group + .channels + .sort_by_key(|item| normalized_source_ordinal(item.header.source_ordinal)); } // Sort groups based on the source_ordinal of their first channel let mut groups: Vec = groups_map.into_values().collect(); - groups.sort_by_key(|group| { - group.channels.first().map_or(u32::MAX, |c| c.header.source_ordinal) - }); + groups.sort_by_key(|group| group.channels.first().map_or(u32::MAX, |c| normalized_source_ordinal(c.header.source_ordinal))); groups } else { vec![] @@ -527,11 +527,11 @@ impl PlaylistSource for MemoryPlaylistSource { fn sort_by_provider_ordinal(&mut self) { let playlist = Arc::make_mut(&mut self.playlist); for group in &mut *playlist { - group.channels.sort_by_key(|item| item.header.source_ordinal); + group + .channels + .sort_by_key(|item| normalized_source_ordinal(item.header.source_ordinal)); } - playlist.sort_by_key(|group| - group.channels.first().map_or(u32::MAX, |c| c.header.source_ordinal) - ); + playlist.sort_by_key(|group| group.channels.first().map_or(u32::MAX, |c| normalized_source_ordinal(c.header.source_ordinal))); } } diff --git a/backend/src/utils/file/config_reader.rs b/backend/src/utils/file/config_reader.rs index dbef5c89a..196b13bfa 100644 --- a/backend/src/utils/file/config_reader.rs +++ b/backend/src/utils/file/config_reader.rs @@ -762,11 +762,9 @@ fn build_templates_to_persist( let config = app_state.app_config.config.load(); let paths = app_state.app_config.paths.load(); - let existing_source_inline_templates = match read_sources_file( - paths.sources_file_path.as_str(), + let existing_source_inline_templates = match parse_sources_file_from_path( + Path::new(paths.sources_file_path.as_str()), true, - false, - None, ) { Ok(existing_sources) => existing_sources.templates, Err(err) => { diff --git a/backend/src/utils/mod.rs b/backend/src/utils/mod.rs index 85efabb87..33eb54245 100644 --- a/backend/src/utils/mod.rs +++ b/backend/src/utils/mod.rs @@ -12,6 +12,7 @@ mod telegram; mod geoip; mod db_viewer; mod epg_parser; +mod ordinal; pub mod ffmpeg; pub use self::binary_utils::*; @@ -22,6 +23,7 @@ pub use self::geoip::*; pub use self::db_viewer::*; pub use shared::utils::*; pub use self::epg_parser::*; +pub use self::ordinal::*; #[macro_export] macro_rules! debug_if_enabled { @@ -75,4 +77,4 @@ pub use self::compression::*; pub use self::file::*; pub use self::network::*; pub use self::crypto_utils::*; -pub use self::step_measure::*; \ No newline at end of file +pub use self::step_measure::*; diff --git a/backend/src/utils/network/xtream.rs b/backend/src/utils/network/xtream.rs index 00a62e36c..56a776c96 100644 --- a/backend/src/utils/network/xtream.rs +++ b/backend/src/utils/network/xtream.rs @@ -145,7 +145,17 @@ pub async fn get_xtream_stream_info(client: &reqwest::Client, // Capture release date for children let series_release_date = series_stream_props.release_date.clone(); - if let Some(mut episodes) = parse_xtream_series_info(&pli.get_uuid(), &series_stream_props, &group, &series_name, input, series_release_date.as_ref()) { + if let Some(mut episodes) = parse_xtream_series_info( + &pli.get_uuid(), + &series_stream_props, + &group, + &series_name, + input, + series_release_date.as_ref(), + // `pli` is `XtreamPlaylistItem`, which stores header fields flattened. + // `source_ordinal` is copied from `PlaylistItemHeader.source_ordinal` on conversion. + pli.source_ordinal, + ) { let config = &app_state.app_config.config.load(); match get_target_storage_path(config, target.name.as_str()) { None => { diff --git a/backend/src/utils/ordinal.rs b/backend/src/utils/ordinal.rs new file mode 100644 index 000000000..f24f9d396 --- /dev/null +++ b/backend/src/utils/ordinal.rs @@ -0,0 +1,8 @@ +#[inline] +pub fn normalized_source_ordinal(source_ordinal: u32) -> u32 { + if source_ordinal == 0 { + u32::MAX + } else { + source_ordinal + } +} diff --git a/shared/src/foundation/filter.rs b/shared/src/foundation/filter.rs index 83a0bc5a3..b5295578f 100644 --- a/shared/src/foundation/filter.rs +++ b/shared/src/foundation/filter.rs @@ -381,8 +381,25 @@ fn get_parser_binary_op(expr: &Pair) -> Result Vec { + let mut placeholders = IndexSet::new(); + for captures in CONSTANTS.re_template_var.captures_iter(input) { + if let Some(inner) = captures.get(1) { + let value = inner.as_str(); + if !value.is_empty() { + placeholders.insert(format!("!{value}!")); + } + } + } + placeholders.into_iter().collect() +} + pub fn get_filter(filter_text: &str, templates: Option<&[PatternTemplate]>) -> Result { let source = apply_templates_to_pattern_single(filter_text, templates)?; + let unresolved_placeholders = unresolved_template_placeholders(&source); + if !unresolved_placeholders.is_empty() { + return info_err_res!("Unknown template placeholder(s) in filter: {}", unresolved_placeholders.join(", ")); + } match FilterParser::parse(Rule::main, &source) { Ok(pairs) => { @@ -860,4 +877,12 @@ mod tests { Err(e) => panic!("{e}"), } } + + #[test] + fn test_filter_unknown_template_placeholder_reports_error() { + let err = get_filter("!UNKNOWN_FILTER!", None).expect_err("unknown placeholder should fail"); + let msg = err.to_string(); + assert!(msg.contains("Unknown template placeholder(s) in filter")); + assert!(msg.contains("!UNKNOWN_FILTER!")); + } } diff --git a/shared/src/utils/json_utils.rs b/shared/src/utils/json_utils.rs index 754bacbff..3e96188f3 100644 --- a/shared/src/utils/json_utils.rs +++ b/shared/src/utils/json_utils.rs @@ -37,7 +37,7 @@ pub fn get_string_from_serde_value(value: &Value) -> Option { const MARKDOWN_SPECIAL_CHARS: &str = r#"_*[]()~`>#+-=|{}.!\"#; -fn escape_markdown_v2(text: &str) -> String { +pub fn escape_markdown_v2(text: &str) -> String { let mut escaped = String::new(); for c in text.chars() { if MARKDOWN_SPECIAL_CHARS.contains(c) {