diff --git a/CHANGELOG.md b/CHANGELOG.md index 98c828482..14f2e6f7d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -28,6 +28,8 @@ - WebUI API-User category selection implemented. - Stream table "Copy-To-Clipboard" functions added. - Refactored provider connection handling to avoid potential race conditions. +- Added optional `panel_api` integration to renew expired accounts or provision new aliases when provider connections are exhausted. +- WebUI: Added Panel API config view and new `/api/v1/config/sources` endpoint to persist + hot-reload `sources.yml`. - Added exp_date field to inputs, aliases, and CSV batch files; accepts dates in "YYYY-MM-DD HH:MM:SS" format or Unix timestamps (seconds since epoch). - Added cloudflare_header to reverse proxy disable_header settings. - Added CPU usage to the WebUI view. diff --git a/README.md b/README.md index 7baba05e4..23ec4cde9 100644 --- a/README.md +++ b/README.md @@ -922,22 +922,51 @@ Input alias definition for same provider with same content but different credent ``` #### `panel_api` + If provider connections are exhausted, tuliprox can optionally call a provider panel API to: - renew expired accounts first (based on `exp_date`) - otherwise create a new alias account and persist it The API is configured generically via predefined query parameters; only `type: m3u` is supported. + Use the literal value `auto` to fill sensitive values at runtime: - `api_key: auto` is replaced by `panel_api.api_key` - in `client_renew`, `username: auto` / `password: auto` are replaced by the account being renewed +- in `client_info`, `username: auto` / `password: auto` are replaced by the account being queried `client_info` is used to fetch the exact `exp_date` (via the `expire` field) and is also executed on boot to sync `exp_date` for existing inputs/aliases. For `client_new`, the Panel API call would look like this in the example shown: - ```https://panel.example.tld/api.php?action=new&type=m3u&sub=1&api_key=1234567890``` +```text +https://panel.example.tld/api.php?action=new&type=m3u&sub=1&api_key=1234567890 +``` -Example: +Response evaluation logic +Tuliprox evaluates Panel API responses as JSON with the following logic, depending on the operation: + +`Common rule (all operations)` + - The response must contain `status: true`. If status is missing or not true, the operation is treated as failed. + +`client_new (create alias)` + + - Require `status: true`. + - Attempt to extract credentials directly from the JSON response: + - username + - password + - If one or both fields are missing, tuliprox attempts a fallback extraction from a URL contained in the JSON: + - If the JSON contains a url field, tuliprox parses it and tries to extract username/password from it (e.g., query string or embedded credentials depending on the provider’s URL format). + - If credentials cannot be derived from either the direct fields or the url fallback, the operation is treated as failed and no alias is persisted. + +`client_renew (renew existing account)` + + - Only `status: true` is evaluated. No credentials are extracted or updated as part of renew. + +`client_info (sync expiration)` + + - Require `status: true`. + - Extract the expiration timestamp/date from the JSON field: + - `expire` → used to populate/update exp_date for the corresponding input/alias. ```yaml - sources: - inputs: diff --git a/backend/src/api/api_utils.rs b/backend/src/api/api_utils.rs index b5c3df228..7404f0a83 100644 --- a/backend/src/api/api_utils.rs +++ b/backend/src/api/api_utils.rs @@ -32,9 +32,9 @@ use shared::utils::{ extract_extension_from_url, replace_url_extension, sanitize_sensitive_info, DASH_EXT, HLS_EXT, }; use std::borrow::Cow; -use std::collections::HashMap; use std::io::SeekFrom; use std::path::{Path, PathBuf}; +use std::collections::{BTreeMap, HashMap}; use std::sync::Arc; use std::time::Duration; use tokio::io::{AsyncReadExt, AsyncSeekExt}; @@ -352,38 +352,15 @@ async fn resolve_streaming_strategy( .force_exact_acquire_connection(provider, &fingerprint.addr) .await } else { - // For panel-managed pools we must not allocate providers in grace period, otherwise - // the pool may never reach `Exhausted` and panel provisioning won't be triggered. - let allow_provider_grace = input.panel_api.is_none(); - if !allow_provider_grace { - debug_if_enabled!( - "panel_api: disabling provider grace allocations for input {}", - sanitize_sensitive_info(&input.name) - ); - } app_state .active_provider - .acquire_connection_with_grace_override(&input.name, &fingerprint.addr, allow_provider_grace) + .acquire_connection(&input.name, &fingerprint.addr) .await }; - if provider_connection_handle.is_none() - && force_provider.is_none() - && try_provision_account_on_exhausted(app_state, input).await - { - debug_if_enabled!( - "panel_api: provider pool exhausted for input {}, provision succeeded; re-acquiring connection", - sanitize_sensitive_info(&input.name) - ); - provider_connection_handle = app_state - .active_provider - .acquire_connection_with_grace_override(&input.name, &fingerprint.addr, input.panel_api.is_none()) - .await; - } else if provider_connection_handle.is_none() && force_provider.is_none() && input.panel_api.is_some() { - debug_if_enabled!( - "panel_api: provider pool exhausted for input {}, provision skipped/failed", - sanitize_sensitive_info(&input.name) - ); + if provider_connection_handle.is_none() && force_provider.is_none() && input.panel_api.is_some() { + provider_connection_handle = + try_provision_and_reacquire_on_provider_pool_exhausted(app_state, fingerprint, input).await; } let stream_response_params = @@ -456,6 +433,40 @@ async fn resolve_streaming_strategy( } } +async fn try_provision_and_reacquire_on_provider_pool_exhausted( + app_state: &AppState, + fingerprint: &Fingerprint, + input: &ConfigInput, +) -> Option { + let active_provider_connections = app_state + .active_provider + .active_connections() + .await + .map(|c| c.into_iter().collect::>()); + debug_if_enabled!( + "panel_api: provider pool exhausted for input {} (active_provider_connections={:?})", + sanitize_sensitive_info(&input.name), + active_provider_connections + ); + + if try_provision_account_on_exhausted(app_state, input).await { + debug_if_enabled!( + "panel_api: provider pool exhausted for input {}, provision succeeded; re-acquiring connection", + sanitize_sensitive_info(&input.name) + ); + app_state + .active_provider + .acquire_connection(&input.name, &fingerprint.addr) + .await + } else { + debug_if_enabled!( + "panel_api: provider pool exhausted for input {}, provision skipped/failed", + sanitize_sensitive_info(&input.name) + ); + None + } +} + fn get_grace_period_millis( connection_permission: UserConnectionPermission, stream_response_params: &ProviderStreamState, @@ -494,94 +505,7 @@ async fn create_stream_response_details( .as_ref() .and_then(|r| r.stream.as_ref()) .map_or_else(default_grace_period_millis, |s| s.grace_period_millis); - let mut grace_period_millis = get_grace_period_millis( - connection_permission, - &streaming_strategy.provider_stream_state, - config_grace_period_millis, - ); - - if let ProviderStreamState::GracePeriod(ref provider_grace_check, ref request_url) = - streaming_strategy.provider_stream_state - { - tokio::time::sleep(tokio::time::Duration::from_millis(grace_period_millis)).await; - if let Some(provider_name) = provider_grace_check.as_ref() { - if app_state.active_provider.is_over_limit(provider_name).await { - debug!("Provider connections exhausted after grace period for provider: {provider_name}"); - - let provider_handle = streaming_strategy.provider_handle.take(); - app_state.connection_manager.release_provider_handle(provider_handle).await; - - // If panel_api is configured, try to recover by provisioning a new account (if needed) - // and then re-acquire without provider grace allocations. - let mut recovered = false; - if force_provider.is_none() && input.panel_api.is_some() { - debug_if_enabled!( - "panel_api: provider {} over limit after grace; attempting re-acquire/provision for input {}", - sanitize_sensitive_info(provider_name), - sanitize_sensitive_info(&input.name) - ); - - let mut new_handle = app_state - .active_provider - .acquire_connection_with_grace_override(&input.name, &fingerprint.addr, false) - .await; - - if new_handle.is_none() - && try_provision_account_on_exhausted(app_state, input).await - { - debug_if_enabled!( - "panel_api: provision succeeded after grace exhaustion for input {}; re-acquiring (no grace)", - sanitize_sensitive_info(&input.name) - ); - new_handle = app_state - .active_provider - .acquire_connection_with_grace_override(&input.name, &fingerprint.addr, false) - .await; - } - - if let Some(handle) = new_handle { - if let Some(provider_cfg) = handle.allocation.get_provider_config() { - let (selected_provider_name, url) = if provider_cfg.id == input.id { - (input.name.clone(), stream_url.to_string()) - } else { - ( - provider_cfg.name.clone(), - get_stream_alternative_url(stream_url, input, &provider_cfg), - ) - }; - streaming_strategy.provider_stream_state = - ProviderStreamState::Available(Some(selected_provider_name), url); - streaming_strategy.provider_handle = Some(handle); - recovered = true; - } else { - app_state - .connection_manager - .release_provider_handle(Some(handle)) - .await; - } - } - } - - if !recovered { - app_state - .connection_manager - .update_stream_detail( - &fingerprint.addr, - CustomVideoStreamType::ProviderConnectionsExhausted, - ) - .await; - let stream = create_provider_connections_exhausted_stream(&app_state.app_config, &[]); - streaming_strategy.provider_stream_state = ProviderStreamState::Custom(stream); - } - } - } else { - streaming_strategy.provider_stream_state = - ProviderStreamState::Available(provider_grace_check.clone(), request_url.clone()); - } - } - - // Recompute grace period after potential recovery/strategy changes. - grace_period_millis = get_grace_period_millis( + let grace_period_millis = get_grace_period_millis( connection_permission, &streaming_strategy.provider_stream_state, config_grace_period_millis, diff --git a/backend/src/api/endpoints/v1_api_config.rs b/backend/src/api/endpoints/v1_api_config.rs index b8bc9b075..0e17476b0 100644 --- a/backend/src/api/endpoints/v1_api_config.rs +++ b/backend/src/api/endpoints/v1_api_config.rs @@ -77,7 +77,7 @@ async fn save_config_sources( if let Err(err) = app_state.app_config.set_sources(src) { return (axum::http::StatusCode::INTERNAL_SERVER_ERROR, axum::Json(json!({"error": err.to_string()}))).into_response(); } - app_state.active_provider.update_config(&app_state.app_config).await; + app_state.active_provider.update_config(&app_state.app_config); axum::http::StatusCode::OK.into_response() } Err(err) => (axum::http::StatusCode::BAD_REQUEST, axum::Json(json!({"error": err.to_string()}))).into_response(), diff --git a/backend/src/api/endpoints/xmltv_api.rs b/backend/src/api/endpoints/xmltv_api.rs index 8369651d2..dc3f86124 100644 --- a/backend/src/api/endpoints/xmltv_api.rs +++ b/backend/src/api/endpoints/xmltv_api.rs @@ -521,10 +521,10 @@ mod tests { fn test_parse_timezone() { // This will depend on current DST; we just check it’s within a valid range let berlin = parse_timeshift(Some(&"Europe/Berlin".to_string())).unwrap(); - assert!(berlin == 60 || berlin == 120, "Berlin offset should be 60 or 120, got {}", berlin); + assert!(berlin == 60 || berlin == 120, "Berlin offset should be 60 or 120, got {berlin}"); let new_york = parse_timeshift(Some(&"America/New_York".to_string())).unwrap(); - assert!(new_york == -300 || new_york == -240, "New York offset should be -300 or -240, got {}", new_york); + assert!(new_york == -300 || new_york == -240, "New York offset should be -300 or -240, got {new_york}"); let tokyo = parse_timeshift(Some(&"Asia/Tokyo".to_string())).unwrap(); assert_eq!(tokyo, 540); // always UTC+9 diff --git a/backend/src/api/model/active_provider_manager.rs b/backend/src/api/model/active_provider_manager.rs index e3460d4e6..33fd6ae69 100644 --- a/backend/src/api/model/active_provider_manager.rs +++ b/backend/src/api/model/active_provider_manager.rs @@ -4,28 +4,40 @@ use crate::model::{AppConfig, ConfigInput}; use crate::utils::trace_if_enabled; use log::{error}; use shared::utils::{default_grace_period_millis, default_grace_period_timeout_secs, sanitize_sensitive_info}; +use log::{debug, error}; +use crate::utils::{debug_if_enabled, trace_if_enabled}; +use shared::utils::{default_grace_period_millis, default_grace_period_timeout_secs}; +use shared::utils::sanitize_sensitive_info; use std::collections::{HashMap, HashSet}; use std::net::SocketAddr; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; use tokio::sync::RwLock; use crate::utils::debug_if_enabled; pub type ClientConnectionId = SocketAddr; +type AllocationId = u64; #[derive(Debug, Clone)] pub struct ProviderHandle { pub client_id: ClientConnectionId, + pub allocation_id: AllocationId, pub allocation: ProviderAllocation, } impl ProviderHandle { - pub fn new(client_id: ClientConnectionId, allocation: ProviderAllocation) -> Self { - Self { client_id, allocation } + pub fn new(client_id: ClientConnectionId, allocation_id: AllocationId, allocation: ProviderAllocation) -> Self { + Self { + client_id, + allocation_id, + allocation, + } } } #[derive(Debug, Clone)] struct SharedAllocation { + allocation_id: AllocationId, allocation: ProviderAllocation, connections: HashSet, } @@ -38,13 +50,14 @@ struct SharedConnections { #[derive(Debug, Clone, Default)] struct Connections { - single: HashMap, + single: HashMap>, shared: SharedConnections, } pub struct ActiveProviderManager { providers: ProviderLineupManager, connections: RwLock, + next_allocation_id: AtomicU64, } impl ActiveProviderManager { @@ -54,6 +67,7 @@ impl ActiveProviderManager { Self { providers: ProviderLineupManager::new(inputs, grace_period_millis, grace_period_timeout_secs, event_manager), connections: RwLock::new(Connections::default()), + next_allocation_id: AtomicU64::new(1), } } @@ -69,10 +83,10 @@ impl ActiveProviderManager { (grace_period_millis, grace_period_timeout_secs) } - pub async fn update_config(&self, cfg: &AppConfig) { + pub fn update_config(&self, cfg: &AppConfig) { let (grace_period_millis, grace_period_timeout_secs) = Self::get_grace_options(cfg); let inputs = Self::get_config_inputs(cfg); - self.providers.update_config(inputs, grace_period_millis, grace_period_timeout_secs).await; + self.providers.update_config(inputs, grace_period_millis, grace_period_timeout_secs); } async fn acquire_connection_inner( @@ -100,19 +114,19 @@ impl ActiveProviderManager { ProviderAllocation::Exhausted => {} ProviderAllocation::Available(_) | ProviderAllocation::GracePeriod(_) => { let provider_name = allocation.get_provider_name().unwrap_or_default(); + let allocation_id = self.next_allocation_id.fetch_add(1, Ordering::Relaxed); let mut connections = self.connections.write().await; - if let Some(old) = connections.single.insert(*addr, allocation.clone()) { + let per_addr = connections.single.entry(*addr).or_default(); + if !per_addr.is_empty() { trace_if_enabled!( - "register_connection: address {} already had a allocation for provider {:?} — forcing release on the old allocation", - sanitize_sensitive_info(&addr.to_string()), - old.get_provider_name().unwrap_or_default()); - - drop(connections); - old.release().await; + "register_connection: address {addr} already has {} provider allocations", + per_addr.len() + ); + per_addr.release().await(); } - - debug_if_enabled!("Added provider connection {provider_name:?} for {}", sanitize_sensitive_info(&addr.to_string())); - return Some(ProviderHandle::new(*addr, allocation)); + per_addr.insert(allocation_id, allocation.clone()); + debug_if_enabled!("Added provider connection {provider_name:?} for {}", sanitize_sensitive_info(&addr.to_string())); + return Some(ProviderHandle::new(*addr, allocation_id, allocation)); } } @@ -153,18 +167,20 @@ impl ActiveProviderManager { pub async fn release_connection(&self, addr: &SocketAddr) { // Single connection - let single_allocation = { + let single_allocations = { let mut connections = self.connections.write().await; connections.single.remove(addr) }; - if let Some(allocation) = single_allocation { - debug_if_enabled!( - "Released provider connection {:?} for {}", - allocation.get_provider_name().unwrap_or_default(), - sanitize_sensitive_info(&addr.to_string()) - ); - allocation.release().await; + if let Some(allocations) = single_allocations { + for (_id, allocation) in allocations { + debug_if_enabled!( + "Released provider connection {:?} for {}", + allocation.get_provider_name().unwrap_or_default(), + sanitize_sensitive_info(&addr.to_string()) + ); + allocation.release().await; + } return; } @@ -211,27 +227,104 @@ impl ActiveProviderManager { } pub async fn release_handle(&self, handle: &ProviderHandle) { - self.release_connection(&handle.client_id).await; + let mut released = None; + { + let mut connections = self.connections.write().await; + if let Some(per_addr) = connections.single.get_mut(&handle.client_id) { + released = per_addr.remove(&handle.allocation_id); + if per_addr.is_empty() { + connections.single.remove(&handle.client_id); + } + } + + if released.is_none() { + let mut remove_key: Option = None; + for (key, shared) in &connections.shared.by_key { + if shared.allocation_id == handle.allocation_id { + remove_key = Some(key.clone()); + break; + } + } + + if let Some(key) = remove_key { + if let Some(shared) = connections.shared.by_key.remove(&key) { + released = Some(shared.allocation); + for addr in shared.connections { + connections.shared.key_by_addr.remove(&addr); + } + } + } + } + } + + if let Some(allocation) = released { + allocation.release().await; + } } pub async fn make_shared_connection(&self, addr: &SocketAddr, key: &str) { - let mut connections = self.connections.write().await; - let handle = connections.single.remove(addr); - if let Some(allocation) = handle { - debug_if_enabled!("Shared connection: Promoted connection {} to shared with key {key:?}", sanitize_sensitive_info(&addr.to_string())); - connections.shared.by_key.insert(key.to_string(), SharedAllocation { allocation, connections: HashSet::from([*addr]) }); - connections.shared.key_by_addr.insert(*addr, key.to_string()); + let extras = { + let mut connections = self.connections.write().await; + let mut extras = Vec::new(); + let handle = connections.single.remove(addr).and_then(|m| { + if m.is_empty() { + return None; + } + let mut iter = m.into_iter(); + let (id, allocation) = iter.next().expect("non-empty map"); + for (_extra_id, extra_alloc) in iter { + extras.push(extra_alloc); + } + if !extras.is_empty() { + trace_if_enabled!( + "Shared connection promotion expects a single allocation for {addr}, found {}", + extras.len() + 1 + ); + } + Some(ProviderHandle::new(*addr, id, allocation)) + }); + + if let Some(handle) = &handle { + let provider_name = handle.allocation.get_provider_name().unwrap_or_default(); + debug_if_enabled!( + "Shared connection: promoted addr {addr} provider={} key={}", + sanitize_sensitive_info(&provider_name), + sanitize_sensitive_info(key) + ); + connections.shared.by_key.insert( + key.to_string(), + SharedAllocation { + allocation_id: handle.allocation_id, + allocation: handle.allocation.clone(), + connections: HashSet::from([*addr]), + }, + ); + connections.shared.key_by_addr.insert(*addr, key.to_string()); + } + extras + }; + + for allocation in extras { + allocation.release().await; } } pub async fn add_shared_connection(&self, addr: &SocketAddr, key: &str) { let mut connections = self.connections.write().await; if let Some(shared_allocation) = connections.shared.by_key.get_mut(key) { - debug_if_enabled!("Shared connection: Added connection {} to shared with key {key:?}", sanitize_sensitive_info(&addr.to_string())); + let provider_name = shared_allocation.allocation.get_provider_name().unwrap_or_default(); + debug_if_enabled!( + "Shared connection: added addr {addr} provider={} key={}", + sanitize_sensitive_info(&provider_name), + sanitize_sensitive_info(key) + ); shared_allocation.connections.insert(*addr); connections.shared.key_by_addr.insert(*addr, key.to_string()); } else { - error!("Failed to add shared connection for {}: url: {key:?} not found", sanitize_sensitive_info(&addr.to_string())); + error!( + "Failed to add shared connection for {addr}: url {} not found", + sanitize_sensitive_info(key) + ); } } diff --git a/backend/src/api/model/app_state.rs b/backend/src/api/model/app_state.rs index 4994656ab..07ed25904 100644 --- a/backend/src/api/model/app_state.rs +++ b/backend/src/api/model/app_state.rs @@ -120,7 +120,7 @@ pub async fn update_app_state_sources( ) -> Result<(), TuliproxError> { let targets = sources.validate_targets(Some(&app_state.forced_targets.load().target_names))?; app_state.forced_targets.store(Arc::new(targets)); - let updates = app_state.set_sources(sources).await?; + let updates = app_state.set_sources(sources)?; update_target_caches(app_state, updates.targets.as_ref()).await; restart_services(app_state, &updates); Ok(()) @@ -278,9 +278,7 @@ impl AppState { self.active_users.update_config(&config); self.app_config.set_config(config)?; - self.active_provider - .update_config(&self.app_config) - .await; + self.active_provider.update_config(&self.app_config); self.update_config().await; if changes.geoip { @@ -323,12 +321,10 @@ impl AppState { } } - pub(in crate::api::model) async fn set_sources(&self,sources: SourcesConfig) -> Result { + pub(in crate::api::model) fn set_sources(&self,sources: SourcesConfig) -> Result { let changes = self.detect_changes_for_sources(&sources); self.app_config.set_sources(sources)?; - self.active_provider - .update_config(&self.app_config) - .await; + self.active_provider.update_config(&self.app_config); Ok(changes) } diff --git a/backend/src/api/model/connection_manager.rs b/backend/src/api/model/connection_manager.rs index 54aeca8f8..a6303435b 100644 --- a/backend/src/api/model/connection_manager.rs +++ b/backend/src/api/model/connection_manager.rs @@ -65,7 +65,7 @@ impl ConnectionManager { pub async fn release_provider_handle(&self, provider_handle: Option) { if let Some(handle) = provider_handle { - self.release_provider_connection(&handle.client_id).await; + self.provider_manager.release_handle(&handle).await; } } diff --git a/backend/src/api/model/provider_config.rs b/backend/src/api/model/provider_config.rs index 2153e73aa..f2961139f 100644 --- a/backend/src/api/model/provider_config.rs +++ b/backend/src/api/model/provider_config.rs @@ -45,7 +45,7 @@ pub struct ProviderConfig { max_connections: usize, priority: i16, exp_date: Option, - connection: RwLock, + connection: Arc>, on_connection_change: ProviderConnectionChangeCallback, } @@ -101,10 +101,7 @@ macro_rules! modify_connections { } impl ProviderConfig { - pub fn new<'a, F>(cfg: &ConfigInput, get_connection: Option, on_connection_change: ProviderConnectionChangeCallback) -> Self - where - F: Fn(&str) -> Option<&'a ProviderConfigConnection>, - { + pub fn new(cfg: &ConfigInput, connection: Arc>, on_connection_change: ProviderConnectionChangeCallback) -> Self { let panel_api_enabled = cfg.panel_api.is_some(); let effective_max_connections = if panel_api_enabled && cfg.max_connections == 0 { debug_if_enabled!( @@ -125,15 +122,17 @@ impl ProviderConfig { max_connections: effective_max_connections, priority: cfg.priority, exp_date: cfg.exp_date, - connection: RwLock::new(get_connection.and_then(|f| f(cfg.name.as_str())).map_or_else(Default::default, Clone::clone)), + connection, on_connection_change } } - pub fn new_alias<'a, F>(cfg: &ConfigInput, alias: &ConfigInputAlias, get_connection: Option, on_connection_change: ProviderConnectionChangeCallback) -> Self - where - F: Fn(&str) -> Option<&'a ProviderConfigConnection>, - { + pub fn new_alias( + cfg: &ConfigInput, + alias: &ConfigInputAlias, + connection: Arc>, + on_connection_change: ProviderConnectionChangeCallback, + ) -> Self { let panel_api_enabled = cfg.panel_api.is_some(); let effective_max_connections = if panel_api_enabled && alias.max_connections == 0 { debug_if_enabled!( @@ -154,7 +153,7 @@ impl ProviderConfig { max_connections: effective_max_connections, priority: alias.priority, exp_date: alias.exp_date, - connection: RwLock::new(get_connection.and_then(|f| f(alias.name.as_str())).map_or_else(Default::default, Clone::clone)), + connection, on_connection_change, } } @@ -164,6 +163,11 @@ impl ProviderConfig { self.max_connections } + #[inline] + pub(crate) fn exp_date(&self) -> Option { + self.exp_date + } + pub fn get_user_info(&self) -> Option { InputUserInfo::new(self.input_type, self.username.as_deref(), self.password.as_deref(), &self.url) } @@ -359,15 +363,6 @@ impl ProviderConfigWrapper { } None } - - pub async fn get_connection_info(&self) -> ProviderConfigConnection { - let guard = self.inner.connection.read().await; - ProviderConfigConnection { - current_connections: guard.current_connections, - granted_grace: guard.granted_grace, - grace_ts: guard.grace_ts, - } - } } impl Deref for ProviderConfigWrapper { type Target = ProviderConfig; diff --git a/backend/src/api/model/provider_lineup_manager.rs b/backend/src/api/model/provider_lineup_manager.rs index db597571e..6a7ddff59 100644 --- a/backend/src/api/model/provider_lineup_manager.rs +++ b/backend/src/api/model/provider_lineup_manager.rs @@ -1,14 +1,16 @@ use crate::api::model::provider_config::ProviderConfigWrapper; use crate::api::model::{EventManager, ProviderConfig, ProviderConfigConnection, ProviderConnectionChangeCallback}; use crate::utils::debug_if_enabled; -use crate::model::ConfigInput; +use crate::model::{is_input_expired, ConfigInput}; use arc_swap::ArcSwap; +use dashmap::DashMap; use log::{debug, log_enabled}; use shared::utils::{display_vec, sanitize_sensitive_info}; use std::collections::HashMap; use std::fmt; use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; use std::sync::Arc; +use tokio::sync::RwLock; macro_rules! gen_provider_search { ($fn_name:ident, $field: ident, $crit_type:ty) => { @@ -45,6 +47,16 @@ macro_rules! gen_provider_search { } } +fn get_or_create_provider_connection( + provider_connections: &DashMap>>, + provider_name: &str, +) -> Arc> { + provider_connections + .entry(provider_name.to_string()) + .or_insert_with(|| Arc::new(RwLock::new(ProviderConfigConnection::default()))) + .clone() +} + #[derive(Debug, Clone)] pub enum ProviderAllocation { Exhausted, @@ -165,12 +177,9 @@ struct SingleProviderLineup { } impl SingleProviderLineup { - fn new<'a, F>(cfg: &ConfigInput, get_connection: Option, connection_change: &ProviderConnectionChangeCallback) -> Self - where - F: Fn(&str) -> Option<&'a ProviderConfigConnection>, - { + fn new(cfg: &ConfigInput, connection: Arc>, connection_change: &ProviderConnectionChangeCallback) -> Self { Self { - provider: ProviderConfigWrapper::new(ProviderConfig::new(cfg, get_connection, Arc::clone(connection_change))), + provider: ProviderConfigWrapper::new(ProviderConfig::new(cfg, connection, Arc::clone(connection_change))), } } @@ -214,41 +223,31 @@ impl fmt::Display for ProviderPriorityGroup { } } -impl ProviderPriorityGroup { - async fn is_exhausted(&self) -> bool { - match self { - ProviderPriorityGroup::SingleProviderGroup(g) => g.is_exhausted().await, - ProviderPriorityGroup::MultiProviderGroup(_, groups) => { - for g in groups { - if !g.is_exhausted().await { - return false; - } - } - true - } - } - } -} - - /// Manages multiple providers, ensuring that connections are allocated in a round-robin manner based on priority. #[repr(align(64))] #[derive(Debug)] struct MultiProviderLineup { name: String, providers: Vec, - index: AtomicUsize, } impl MultiProviderLineup { - pub fn new<'a, F>(cfg_input: &ConfigInput, get_connection: Option, connection_change: &ProviderConnectionChangeCallback) -> Self - where - F: Fn(&str) -> Option<&'a ProviderConfigConnection> + Copy, - { - let mut inputs = vec![ProviderConfigWrapper::new(ProviderConfig::new(cfg_input, get_connection, Arc::clone(connection_change)))]; + pub fn new( + cfg_input: &ConfigInput, + provider_connections: &DashMap>>, + connection_change: &ProviderConnectionChangeCallback, + ) -> Self { + let input_connection = get_or_create_provider_connection(provider_connections, cfg_input.name.as_str()); + let mut inputs = vec![ProviderConfigWrapper::new(ProviderConfig::new(cfg_input, input_connection, Arc::clone(connection_change)))]; if let Some(aliases) = &cfg_input.aliases { for alias in aliases { - inputs.push(ProviderConfigWrapper::new(ProviderConfig::new_alias(cfg_input, alias, get_connection, Arc::clone(connection_change)))); + let alias_connection = get_or_create_provider_connection(provider_connections, alias.name.as_str()); + inputs.push(ProviderConfigWrapper::new(ProviderConfig::new_alias( + cfg_input, + alias, + alias_connection, + Arc::clone(connection_change), + ))); } } let mut providers = HashMap::new(); @@ -271,7 +270,6 @@ impl MultiProviderLineup { Self { name: cfg_input.name.clone(), providers, - index: AtomicUsize::new(0), } } @@ -312,6 +310,16 @@ impl MultiProviderLineup { } ProviderPriorityGroup::MultiProviderGroup(index, pg) => { let provider_count = pg.len(); + if grace { + for p in pg { + let result = p.try_allocate(true, grace_period_timeout_secs).await; + if !matches!(result, ProviderAllocation::Exhausted) { + return result; + } + } + return ProviderAllocation::Exhausted; + } + let start = index.fetch_add(1, Ordering::AcqRel) % provider_count; let mut idx = start; @@ -396,72 +404,50 @@ impl MultiProviderLineup { /// } /// ``` async fn acquire(&self, with_grace: bool, grace_period_timeout_secs: u64) -> ProviderAllocation { - let provider_count = self.providers.len(); - let start = self.index.fetch_add(1, Ordering::AcqRel) % provider_count; - let mut idx = start; - - loop { - let priority_group = &self.providers[idx]; - let allocation = { - let without_grace_allocation = Self::acquire_next_provider_from_group(priority_group, false, grace_period_timeout_secs).await; - - if with_grace && matches!(without_grace_allocation, ProviderAllocation::Exhausted) { - Self::acquire_next_provider_from_group(priority_group, true, grace_period_timeout_secs).await - } else { - without_grace_allocation - } - }; - + // Phase 1: prefer providers with available capacity (no grace allocations), + // scanning priority groups from highest -> lowest. + for priority_group in &self.providers { + let allocation = + Self::acquire_next_provider_from_group(priority_group, false, grace_period_timeout_secs).await; if !matches!(allocation, ProviderAllocation::Exhausted) { - if priority_group.is_exhausted().await { - self.index.store((idx + 1) % provider_count, Ordering::Release); - } return allocation; } - - idx = (idx + 1) % provider_count; - - // loop end - if idx == start { - break; - } } + if !with_grace { + return ProviderAllocation::Exhausted; + } + + // Phase 2: all providers are at capacity, allow grace allocations (still respecting priority order). + for priority_group in &self.providers { + let allocation = + Self::acquire_next_provider_from_group(priority_group, true, grace_period_timeout_secs).await; + if !matches!(allocation, ProviderAllocation::Exhausted) { + return allocation; + } + } ProviderAllocation::Exhausted } // it intended to use with redirects to cycle through provider async fn get_next(&self, grace_period_timeout_secs: u64) -> Option> { - let provider_count = self.providers.len(); - - let start = self.index.fetch_add(1, Ordering::AcqRel) % provider_count; - let mut idx = start; - - loop { - let priority_group = &self.providers[idx]; - - let allocation = { - let config = Self::get_next_provider_from_group(priority_group, false, grace_period_timeout_secs).await; - if config.is_none() { - Self::get_next_provider_from_group(priority_group, true, grace_period_timeout_secs).await - } else { - config - } - }; - - if let Some(config) = allocation { - if priority_group.is_exhausted().await { - self.index.store((idx + 1) % provider_count, Ordering::Release); - } + // Phase 1: prefer providers with available capacity (no grace allocations), + // scanning priority groups from highest -> lowest. + for priority_group in &self.providers { + if let Some(config) = + Self::get_next_provider_from_group(priority_group, false, grace_period_timeout_secs).await + { return Some(config); } + } - idx = (idx + 1) % provider_count; - - // loop end - if idx == start { - break; + // Phase 2: no provider is available, allow grace. + for priority_group in &self.providers { + if let Some(config) = + Self::get_next_provider_from_group(priority_group, true, grace_period_timeout_secs).await + { + return Some(config); } } @@ -516,34 +502,42 @@ pub(in crate::api::model) struct ProviderLineupManager { grace_period_timeout_secs: AtomicU64, inputs: Arc>>>, providers: Arc>>, + provider_connections: DashMap>>, event_manager: Arc, } impl ProviderLineupManager { pub fn new(inputs: Vec>, grace_period_millis: u64, grace_period_timeout_secs: u64, event_manager: &Arc) -> Self { - let lineups = inputs.iter().map(|i| Self::create_lineup(i, None, event_manager)).collect(); + let provider_connections: DashMap>> = DashMap::new(); + let lineups = inputs + .iter() + .map(|i| Self::create_lineup(i, &provider_connections, event_manager)) + .collect(); Self { grace_period_millis: AtomicU64::new(grace_period_millis), grace_period_timeout_secs: AtomicU64::new(grace_period_timeout_secs), inputs: Arc::new(ArcSwap::from_pointee(inputs)), providers: Arc::new(ArcSwap::from_pointee(lineups)), + provider_connections, event_manager: Arc::clone(event_manager), } } - fn create_lineup(cfg_input: &ConfigInput, provider_connections: Option<&HashMap<&str, ProviderConfigConnection>>, event_manager: &Arc) -> ProviderLineup { - let get_connections = provider_connections.map(|c| |name: &str| c.get(name)); - - //let cfg_name = cfg_input.name.clone(); + fn create_lineup( + cfg_input: &ConfigInput, + provider_connections: &DashMap>>, + event_manager: &Arc, + ) -> ProviderLineup { let event_manager = Arc::clone(event_manager); let on_connection_change: ProviderConnectionChangeCallback = Arc::new(move |name: &str, connections: usize| { event_manager.send_provider_event(name, connections); }); if cfg_input.aliases.as_ref().is_some_and(|a| !a.is_empty()) { - ProviderLineup::Multi(MultiProviderLineup::new(cfg_input, get_connections, &on_connection_change)) + ProviderLineup::Multi(MultiProviderLineup::new(cfg_input, provider_connections, &on_connection_change)) } else { - ProviderLineup::Single(SingleProviderLineup::new(cfg_input, get_connections, &on_connection_change)) + let connection = get_or_create_provider_connection(provider_connections, cfg_input.name.as_str()); + ProviderLineup::Single(SingleProviderLineup::new(cfg_input, connection, &on_connection_change)) } } @@ -607,7 +601,7 @@ impl ProviderLineupManager { false } - pub async fn update_config(&self, new_inputs: Vec>, grace_period_millis: u64, grace_period_timeout_secs: u64) { + pub fn update_config(&self, new_inputs: Vec>, grace_period_millis: u64, grace_period_timeout_secs: u64) { self.grace_period_millis.store(grace_period_millis, Ordering::Relaxed); self.grace_period_timeout_secs.store(grace_period_timeout_secs, Ordering::Relaxed); @@ -615,34 +609,9 @@ impl ProviderLineupManager { return; } - let old_lineups = self.providers.load(); - let mut provider_connections = HashMap::new(); - for lineup in old_lineups.iter() { - match lineup { - ProviderLineup::Single(single) => { - provider_connections.insert(single.provider.name.as_str(), single.provider.get_connection_info().await); - } - ProviderLineup::Multi(multi) => { - for group in &multi.providers { - match group { - ProviderPriorityGroup::SingleProviderGroup(cfg) => { - provider_connections.insert(cfg.name.as_str(), cfg.get_connection_info().await); - } - ProviderPriorityGroup::MultiProviderGroup(_, cfgs) => { - for cfg in cfgs { - provider_connections.insert(cfg.name.as_str(), cfg.get_connection_info().await); - } - } - } - } - } - } - } - let mut new_lineups: Vec = Vec::with_capacity(new_inputs.len()); - let connections = Some(provider_connections); for input in &new_inputs { - new_lineups.push(Self::create_lineup(input, connections.as_ref(), &self.event_manager)); + new_lineups.push(Self::create_lineup(input, &self.provider_connections, &self.event_manager)); } debug_if_enabled!("inputs {}", sanitize_sensitive_info(&display_vec(&new_inputs))); @@ -696,8 +665,9 @@ impl ProviderLineupManager { allow_grace: bool, ) -> ProviderAllocation { let providers = self.providers.load(); + let lineup_opt = Self::get_provider_config_by_name(input_name, &providers); let with_grace = allow_grace && self.grace_period_millis.load(Ordering::Acquire) > 0; - let allocation = match Self::get_provider_config_by_name(input_name, &providers) { + let allocation = match lineup_opt { None => ProviderAllocation::Exhausted, // No Name matched, we don't have this provider Some((lineup, _config)) => { lineup @@ -705,10 +675,55 @@ impl ProviderLineupManager { .await } }; + if matches!(allocation, ProviderAllocation::Exhausted) { + if let Some((lineup, _cfg)) = lineup_opt { + Self::log_exhausted_pool_snapshot(input_name, lineup).await; + } + } Self::log_allocation(&allocation); allocation } + async fn log_exhausted_pool_snapshot(input_name: &str, lineup: &ProviderLineup) { + if !log_enabled!(log::Level::Debug) { + return; + } + + let mut entries: Vec = Vec::new(); + match lineup { + ProviderLineup::Single(single) => { + entries.push(Self::format_provider_snapshot_entry(&single.provider).await); + } + ProviderLineup::Multi(multi) => { + for group in &multi.providers { + match group { + ProviderPriorityGroup::SingleProviderGroup(cfg) => { + entries.push(Self::format_provider_snapshot_entry(cfg).await); + } + ProviderPriorityGroup::MultiProviderGroup(_, cfgs) => { + for cfg in cfgs { + entries.push(Self::format_provider_snapshot_entry(cfg).await); + } + } + } + } + } + } + + debug_if_enabled!( + "Provider pool exhausted for input {} (pool_snapshot=[{}])", + sanitize_sensitive_info(input_name), + sanitize_sensitive_info(&entries.join(", ")) + ); + } + + async fn format_provider_snapshot_entry(cfg: &ProviderConfigWrapper) -> String { + let current = cfg.get_current_connections().await; + let max = cfg.max_connections(); + let expired = is_input_expired(cfg.exp_date()); + format!("{}:{}/{} expired={expired}", cfg.name, current, max) + } + // This method is used for redirects to cycle through provider // pub async fn get_next_provider(&self, input_name: &str) -> Option> { @@ -874,9 +889,9 @@ mod tests { input.aliases = Some(vec![alias]); let change_callback: ProviderConnectionChangeCallback = Arc::new(dummy_callback); - let dummy_get_connection = |_s: &str| -> Option<&ProviderConfigConnection> { None }; + let provider_connections: DashMap>> = DashMap::new(); // Create MultiProviderLineup with the provider and alias - let lineup = MultiProviderLineup::new(&input, Some(dummy_get_connection), &change_callback); + let lineup = MultiProviderLineup::new(&input, &provider_connections, &change_callback); let rt = tokio::runtime::Runtime::new().unwrap(); rt.block_on(async move { // Test that the alias provider is available @@ -899,8 +914,8 @@ mod tests { // Adding alias with different priority input.aliases = Some(vec![alias]); let change_callback: ProviderConnectionChangeCallback = Arc::new(dummy_callback); - let dummy_get_connection = |_s: &str| -> Option<&ProviderConfigConnection> { None }; - let lineup = MultiProviderLineup::new(&input, Some(dummy_get_connection), &change_callback); + let provider_connections: DashMap>> = DashMap::new(); + let lineup = MultiProviderLineup::new(&input, &provider_connections, &change_callback); // The alias has a higher priority, so the alias should be acquired first let rt = tokio::runtime::Runtime::new().unwrap(); rt.block_on(async move { @@ -921,8 +936,8 @@ mod tests { // Adding multiple aliases input.aliases = Some(vec![alias1, alias2]); let change_callback: ProviderConnectionChangeCallback = Arc::new(dummy_callback); - let dummy_get_connection = |_s: &str| -> Option<&ProviderConfigConnection> { None }; - let lineup = MultiProviderLineup::new(&input, Some(dummy_get_connection), &change_callback); + let provider_connections: DashMap>> = DashMap::new(); + let lineup = MultiProviderLineup::new(&input, &provider_connections, &change_callback); let rt = tokio::runtime::Runtime::new().unwrap(); rt.block_on(async move { // The alias with priority 0 should be acquired first (higher priority) @@ -952,8 +967,8 @@ mod tests { // Adding alias input.aliases = Some(vec![alias1, alias2]); let change_callback: ProviderConnectionChangeCallback = Arc::new(dummy_callback); - let dummy_get_connection = |_s: &str| -> Option<&ProviderConfigConnection> { None }; - let lineup = MultiProviderLineup::new(&input, Some(dummy_get_connection), &change_callback); + let provider_connections: DashMap>> = DashMap::new(); + let lineup = MultiProviderLineup::new(&input, &provider_connections, &change_callback); let rt = tokio::runtime::Runtime::new().unwrap(); rt.block_on(async move { // Acquire connection from alias2 @@ -980,8 +995,9 @@ mod tests { fn test_acquire_when_capacity_available() { let cfg = create_config_input(1, "provider5_1", 1, 2); let change_callback: ProviderConnectionChangeCallback = Arc::new(dummy_callback); - let dummy_get_connection = |_s: &str| -> Option<&ProviderConfigConnection> { None }; - let lineup = SingleProviderLineup::new(&cfg, Some(dummy_get_connection), &change_callback); + let provider_connections: DashMap>> = DashMap::new(); + let connection = get_or_create_provider_connection(&provider_connections, cfg.name.as_str()); + let lineup = SingleProviderLineup::new(&cfg, connection, &change_callback); let rt = tokio::runtime::Runtime::new().unwrap(); rt.block_on(async move { // First acquire attempt should succeed @@ -1001,9 +1017,9 @@ mod tests { fn test_release_connection() { let cfg = create_config_input(1, "provider7_1", 1, 2); let change_callback: ProviderConnectionChangeCallback = Arc::new(dummy_callback); - let dummy_get_connection = |_s: &str| -> Option<&ProviderConfigConnection> { None }; - - let lineup = SingleProviderLineup::new(&cfg, Some(dummy_get_connection), &change_callback); + let provider_connections: DashMap>> = DashMap::new(); + let connection = get_or_create_provider_connection(&provider_connections, cfg.name.as_str()); + let lineup = SingleProviderLineup::new(&cfg, connection, &change_callback); let rt = tokio::runtime::Runtime::new().unwrap(); rt.block_on(async move { // Acquire two connections @@ -1032,8 +1048,8 @@ mod tests { // Create MultiProviderLineup with the provider and alias let change_callback: ProviderConnectionChangeCallback = Arc::new(dummy_callback); - let dummy_get_connection = |_s: &str| -> Option<&ProviderConfigConnection> { None }; - let lineup = MultiProviderLineup::new(&cfg1, Some(dummy_get_connection), &change_callback); + let provider_connections: DashMap>> = DashMap::new(); + let lineup = MultiProviderLineup::new(&cfg1, &provider_connections, &change_callback); let rt = tokio::runtime::Runtime::new().unwrap(); rt.block_on(async move { // Test acquiring the first provider @@ -1065,8 +1081,9 @@ mod tests { fn test_concurrent_acquire() { let cfg = create_config_input(1, "provider9_1", 1, 2); let change_callback: ProviderConnectionChangeCallback = Arc::new(dummy_callback); - let dummy_get_connection = |_s: &str| -> Option<&ProviderConfigConnection> { None }; - let lineup = Arc::new(SingleProviderLineup::new(&cfg, Some(dummy_get_connection), &change_callback)); + let provider_connections: DashMap>> = DashMap::new(); + let connection = get_or_create_provider_connection(&provider_connections, cfg.name.as_str()); + let lineup = Arc::new(SingleProviderLineup::new(&cfg, connection, &change_callback)); let available_count = Arc::new(AtomicU16::new(2)); let grace_period_count = Arc::new(AtomicU16::new(1)); diff --git a/backend/src/api/model/streams/active_client_stream.rs b/backend/src/api/model/streams/active_client_stream.rs index fd6ed6b0c..679536e26 100644 --- a/backend/src/api/model/streams/active_client_stream.rs +++ b/backend/src/api/model/streams/active_client_stream.rs @@ -19,9 +19,8 @@ use futures::task::AtomicWaker; use crate::auth::Fingerprint; const INNER_STREAM: u8 = 0_u8; -const GRACE_BLOCK_STREAM: u8 = 1_u8; -const USER_EXHAUSTED_STREAM: u8 = 2_u8; -const PROVIDER_EXHAUSTED_STREAM: u8 = 3_u8; +const USER_EXHAUSTED_STREAM: u8 = 1_u8; +const PROVIDER_EXHAUSTED_STREAM: u8 = 2_u8; pub(in crate::api) struct ActiveClientStream { inner: BoxedProviderStream, @@ -59,9 +58,10 @@ impl ActiveClientStream { app_state.connection_manager.update_stream_detail(&fingerprint.addr, *cvt).await; } let cfg = &app_state.app_config; - let waker = Some(Arc::new(AtomicWaker::new())); - let waker_clone = waker.clone(); - let grace_stop_flag = Self::stream_grace_period(app_state, &stream_details, grant_user_grace_period, user, fingerprint, waker_clone.clone()); + let waker = Arc::new(AtomicWaker::new()); + let grace_stop_flag = + Self::stream_grace_period(app_state, &stream_details, grant_user_grace_period, user, fingerprint, Some(Arc::clone(&waker))); + let waker = if grace_stop_flag.is_some() { Some(waker) } else { None }; let custom_response = cfg.custom_stream_response.load(); let custom_video = custom_response.as_ref() .map_or((None, None), |c| @@ -128,11 +128,10 @@ impl ActiveClientStream { }; if provider_grace_check.is_some() || user_grace_check.is_some() { - let stream_strategy_flag = Arc::new(AtomicU8::new(GRACE_BLOCK_STREAM)); + let stream_strategy_flag = Arc::new(AtomicU8::new(INNER_STREAM)); let stream_strategy_flag_copy = Arc::clone(&stream_strategy_flag); let grace_period_millis = stream_details.grace_period_millis; - let address = fingerprint.addr; let user_manager = Arc::clone(&active_users); let provider_manager = Arc::clone(&active_provider); let connection_manager = Arc::clone(&connection_manager); @@ -168,7 +167,6 @@ impl ActiveClientStream { } if updated { - connection_manager.release_provider_connection(&address).await; if let Some(flag) = reconnect_flag { flag.notify(); } @@ -182,11 +180,25 @@ impl ActiveClientStream { } None } + + fn stop_provider_stream(&mut self) { + if self.provider_handle.is_some() { + let mgr = Arc::clone(&self.connection_manager); + let handle = self.provider_handle.take(); + self.inner = futures::stream::empty::>().boxed(); + tokio::spawn(async move { + mgr.release_provider_handle(handle).await; + }); + } + } } impl Stream for ActiveClientStream { type Item = Result; fn poll_next(mut self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll> { + if let Some(waker) = &self.waker { + waker.register(cx.waker()); + } let flag = { match &self.send_custom_stream_flag { Some(flag) => flag.load(std::sync::atomic::Ordering::Acquire), @@ -194,17 +206,12 @@ impl Stream for ActiveClientStream { } }; - if flag == GRACE_BLOCK_STREAM { - if let Some(waker) = &self.waker { - waker.register(cx.waker()); - } - return Poll::Pending; - } - if flag == INNER_STREAM { return Pin::new(&mut self.inner).poll_next(cx); } + self.stop_provider_stream(); + let buffer_opt = match flag { USER_EXHAUSTED_STREAM => { self.custom_video.0.as_mut() @@ -231,4 +238,4 @@ impl Drop for ActiveClientStream { mgr.release_provider_handle(hndl).await; }); } -} \ No newline at end of file +} diff --git a/backend/src/api/model/streams/shared_stream_manager.rs b/backend/src/api/model/streams/shared_stream_manager.rs index e3e0f2fc1..6ed204b04 100644 --- a/backend/src/api/model/streams/shared_stream_manager.rs +++ b/backend/src/api/model/streams/shared_stream_manager.rs @@ -430,19 +430,18 @@ impl SharedStreamManager { manager: Arc, ) -> Option<(BoxedProviderStream, Option)> { let shared_state_opt = { - let shared_streams = self.shared_streams.read().await; - shared_streams.by_key.get(stream_url).cloned() + let mut shared_streams = self.shared_streams.write().await; + if let Some(shared_state) = shared_streams.by_key.get(stream_url).cloned() { + shared_streams.key_by_addr.insert(*addr, stream_url.to_owned()); + Some(shared_state) + } else { + None + } }; if let Some(shared_state) = shared_state_opt { - { - let mut shared_streams = self.shared_streams.write().await; - shared_streams.key_by_addr.insert(*addr, stream_url.to_owned()); - } - debug_if_enabled!("Responding to existing shared client stream {} {}", - sanitize_sensitive_info(&addr.to_string()), sanitize_sensitive_info(stream_url) - ); + sanitize_sensitive_info(&addr.to_string()), sanitize_sensitive_info(stream_url)); Some(shared_state.subscribe(addr, manager).await) } else { None diff --git a/backend/src/api/panel_api.rs b/backend/src/api/panel_api.rs index 4259899fc..7b70944f0 100644 --- a/backend/src/api/panel_api.rs +++ b/backend/src/api/panel_api.rs @@ -212,6 +212,34 @@ fn build_panel_url(base_url: &str, query_params: &[(String, String)]) -> Result< Ok(url) } +fn sanitize_panel_api_json_for_log(value: &Value) -> Value { + match value { + Value::Array(arr) => Value::Array(arr.iter().map(sanitize_panel_api_json_for_log).collect()), + Value::Object(obj) => { + let mut out = serde_json::Map::with_capacity(obj.len()); + for (k, v) in obj { + if k.eq_ignore_ascii_case("api_key") || k.eq_ignore_ascii_case("apikey") || k.eq_ignore_ascii_case("token") { + out.insert(k.clone(), Value::String("***".to_string())); + continue; + } + if k.eq_ignore_ascii_case("username") || k.eq_ignore_ascii_case("password") { + out.insert(k.clone(), Value::String("***".to_string())); + continue; + } + if k.eq_ignore_ascii_case("url") { + if let Some(s) = v.as_str() { + out.insert(k.clone(), Value::String(sanitize_sensitive_info(s).into_owned())); + continue; + } + } + out.insert(k.clone(), sanitize_panel_api_json_for_log(v)); + } + Value::Object(out) + } + other => other.clone(), + } +} + async fn panel_get_json(app_state: &AppState, url: Url) -> Result { let client = app_state.http_client.load(); let sanitized = sanitize_sensitive_info(url.as_str()); @@ -228,6 +256,10 @@ async fn panel_get_json(app_state: &AppState, url: Url) -> Result {} + serde_yaml::Value::Null => { + *aliases_value = serde_yaml::Value::Sequence(vec![]); + } + serde_yaml::Value::Mapping(_) => { + // Some users might have a single alias stored as a mapping instead of a list. + // Normalize to a list so we can append safely. + let old = std::mem::replace(aliases_value, serde_yaml::Value::Sequence(vec![])); + if let serde_yaml::Value::Mapping(map) = old { + if let serde_yaml::Value::Sequence(seq) = aliases_value { + seq.push(serde_yaml::Value::Mapping(map)); + } + } + } + _ => { + return create_tuliprox_error_result!(TuliproxErrorKind::Info, "panel_api: input.aliases is not a list in source.yml"); + } } - let Some(alias_seq) = inp_map.get_mut(&aliases_key).and_then(|v| v.as_sequence_mut()) else { + let Some(alias_seq) = aliases_value.as_sequence_mut() else { return create_tuliprox_error_result!(TuliproxErrorKind::Info, "panel_api: input.aliases is not a list in source.yml"); }; @@ -448,12 +499,24 @@ async fn patch_source_yml_update_exp_date( if account_name == input_name { inp_map.insert(serde_yaml::Value::String("exp_date".to_string()), serde_yaml::Value::Number(exp_date.into())); inp_map.insert(serde_yaml::Value::String("enabled".to_string()), serde_yaml::Value::Bool(true)); - } else if let Some(aliases) = inp_map.get_mut(serde_yaml::Value::String("aliases".to_string())).and_then(|v| v.as_sequence_mut()) { - for a in aliases.iter_mut() { - let Some(a_map) = a.as_mapping_mut() else { continue; }; - let a_name = a_map.get(serde_yaml::Value::String("name".to_string())).and_then(|v| v.as_str()); - if a_name == Some(account_name) { - a_map.insert(serde_yaml::Value::String("exp_date".to_string()), serde_yaml::Value::Number(exp_date.into())); + } else if let Some(aliases_value) = inp_map.get_mut(serde_yaml::Value::String("aliases".to_string())) { + // Normalize aliases to a sequence if the file contains a single mapping. + if matches!(aliases_value, serde_yaml::Value::Mapping(_)) { + let old = std::mem::replace(aliases_value, serde_yaml::Value::Sequence(vec![])); + if let serde_yaml::Value::Mapping(map) = old { + if let serde_yaml::Value::Sequence(seq) = aliases_value { + seq.push(serde_yaml::Value::Mapping(map)); + } + } + } + + if let Some(aliases) = aliases_value.as_sequence_mut() { + for a in aliases.iter_mut() { + let Some(a_map) = a.as_mapping_mut() else { continue; }; + let a_name = a_map.get(serde_yaml::Value::String("name".to_string())).and_then(|v| v.as_str()); + if a_name == Some(account_name) { + a_map.insert(serde_yaml::Value::String("exp_date".to_string()), serde_yaml::Value::Number(exp_date.into())); + } } } } @@ -645,7 +708,7 @@ async fn try_renew_expired_account( } } - if let Err(err) = reload_sources(app_state).await { + if let Err(err) = reload_sources(app_state) { debug_if_enabled!("panel_api reload sources failed: {}", err); } return true; @@ -721,7 +784,7 @@ async fn try_create_new_account( } } - if let Err(err) = reload_sources(app_state).await { + if let Err(err) = reload_sources(app_state) { error!("panel_api reload sources failed: {err}"); return false; } @@ -861,7 +924,7 @@ pub(crate) async fn sync_panel_api_exp_dates_on_boot(app_state: &Arc) } if any_change { - if let Err(err) = reload_sources(app_state).await { + if let Err(err) = reload_sources(app_state) { debug_if_enabled!("panel_api boot sync reload sources failed: {}", err); } } @@ -876,4 +939,4 @@ async fn reload_sources(app_state: &Arc) -> Result<(), TuliproxError> // app_state.app_config.set_sources(sources)?; // app_state.active_provider.update_config(&app_state.app_config).await; // Ok(()) -} +} \ No newline at end of file diff --git a/backend/src/processing/processor/playlist.rs b/backend/src/processing/processor/playlist.rs index 63463bc42..01ca8ac10 100644 --- a/backend/src/processing/processor/playlist.rs +++ b/backend/src/processing/processor/playlist.rs @@ -62,8 +62,8 @@ pub fn apply_filter_to_playlist(playlist: &mut [PlaylistGroup], filter: &Filter) } pub fn apply_favourites_to_playlist( - _playlist: &mut Vec, - _favourites_cfg: Option<&Vec>, + _playlist: &mut [PlaylistGroup], + _favourites_cfg: Option<&[ConfigFavourites]>, ) { // if let Some(favourites) = favourites_cfg { // let mut fav_groups: HashMap> = HashMap::new(); @@ -101,7 +101,7 @@ pub fn apply_favourites_to_playlist( fn filter_playlist(playlist: &mut [PlaylistGroup], target: &ConfigTarget) -> Option> { if let Some(mut filtered_playlist) = apply_filter_to_playlist(playlist, &target.filter) { - apply_favourites_to_playlist(&mut filtered_playlist, target.favourites.as_ref()); + apply_favourites_to_playlist(&mut filtered_playlist, target.favourites.as_deref()); Some(filtered_playlist) } else { None diff --git a/backend/src/utils/file/config_reader.rs b/backend/src/utils/file/config_reader.rs index c592434fe..5331b76fd 100644 --- a/backend/src/utils/file/config_reader.rs +++ b/backend/src/utils/file/config_reader.rs @@ -328,7 +328,7 @@ where .map_err(|err| create_tuliprox_error!(TuliproxErrorKind::Info, "Could not write file {}: {}", &path.to_str().unwrap_or("?"), err)) } -pub fn format_sources_yaml_panel_api_query_params_flow_style(yaml: &str) -> String { +fn format_sources_yaml_panel_api_query_params_flow_style_impl(yaml: &str) -> String { let has_trailing_newline = yaml.ends_with('\n'); let lines: Vec<&str> = yaml.split_terminator('\n').collect(); let mut out: Vec = Vec::with_capacity(lines.len()); @@ -420,6 +420,97 @@ pub fn format_sources_yaml_panel_api_query_params_flow_style(yaml: &str) -> Stri } } +fn parse_yaml_key_value_line(s: &str) -> Option<(String, String)> { + let (key, value) = s.split_once(':')?; + let key = key.trim(); + if key.is_empty() { + return None; + } + let value = value.trim(); + let value = if value.is_empty() { "null" } else { value }; + Some((key.to_string(), value.to_string())) +} + +fn format_sources_yaml_aliases_flow_style_impl(yaml: &str) -> String { + let has_trailing_newline = yaml.ends_with('\n'); + let lines: Vec<&str> = yaml.split_terminator('\n').collect(); + let mut out: Vec = Vec::with_capacity(lines.len()); + + let mut aliases_indent: Option = None; + + let mut i = 0usize; + while i < lines.len() { + let line = lines[i]; + let trimmed = line.trim_start(); + let indent = line.len().saturating_sub(trimmed.len()); + + if !trimmed.is_empty() && !trimmed.starts_with('#') { + if let Some(ai) = aliases_indent { + if indent < ai || (indent == ai && trimmed != "aliases:" && !trimmed.starts_with("- ")) { + aliases_indent = None; + } + } + if trimmed == "aliases:" { + aliases_indent = Some(indent); + } + } + + // Convert: + // aliases: + // - name: foo + // url: http://... + // Into: + // - { name: foo, url: http://... } + if aliases_indent.is_some() && trimmed.starts_with("- ") && !trimmed.starts_with("- {") { + let item_indent = indent; + let first = trimmed.strip_prefix("- ").unwrap_or(trimmed); + if let Some((k, v)) = parse_yaml_key_value_line(first) { + let mut parts: Vec<(String, String)> = vec![(k, v)]; + let mut j = i + 1; + while j < lines.len() { + let l2 = lines[j]; + let t2 = l2.trim_start(); + let ind2 = l2.len().saturating_sub(t2.len()); + if t2.is_empty() || t2.starts_with('#') { + break; + } + if ind2 == item_indent.saturating_add(2) && !t2.starts_with("- ") { + if let Some((k2, v2)) = parse_yaml_key_value_line(t2) { + parts.push((k2, v2)); + j += 1; + continue; + } + } + break; + } + + let formatted = parts + .iter() + .map(|(kk, vv)| format!("{kk}: {vv}")) + .collect::>() + .join(", "); + out.push(format!("{}- {{ {} }}", " ".repeat(item_indent), formatted)); + i = j; + continue; + } + } + + out.push(line.to_string()); + i += 1; + } + + if has_trailing_newline { + out.join("\n") + "\n" + } else { + out.join("\n") + } +} + +pub fn format_sources_yaml_panel_api_query_params_flow_style(yaml: &str) -> String { + let yaml = format_sources_yaml_panel_api_query_params_flow_style_impl(yaml); + format_sources_yaml_aliases_flow_style_impl(&yaml) +} + pub async fn save_api_proxy(file_path: &str, backup_dir: &str, config: &ApiProxyConfigDto) -> Result<(), TuliproxError> { write_config_file(file_path, backup_dir, config, "api-proxy.yml", None).await } @@ -479,4 +570,25 @@ mod tests { assert!(out.contains("- { key: type, value: m3u }")); assert!(!out.contains("- key: api_key\n")); } + + #[test] + fn test_sources_yaml_aliases_flow_style() { + let input = r"sources: +- inputs: + - name: demo + aliases: + - name: demo-u1 + url: http://line.example.invalid + username: u1 + password: p1 + priority: 0 + max_connections: 1 + exp_date: 123 +"; + let out = format_sources_yaml_panel_api_query_params_flow_style(input); + assert!( + out.contains("- { name: demo-u1, url: http://line.example.invalid, username: u1, password: p1, priority: 0, max_connections: 1, exp_date: 123 }"), + "out:\n{out}" + ); + } } diff --git a/backend/src/utils/geoip.rs b/backend/src/utils/geoip.rs index 07af9ddf8..5e692a978 100644 --- a/backend/src/utils/geoip.rs +++ b/backend/src/utils/geoip.rs @@ -114,7 +114,7 @@ mod test { if let Some(cc) = geo_ip.lookup("72.13.24.23") { assert_eq!(cc, "US"); } else { - assert!(false); + panic!("GeoIP lookup returned no result"); } } -} \ No newline at end of file +} diff --git a/frontend/src/app/components/api_user/playlist.rs b/frontend/src/app/components/api_user/playlist.rs index 118550665..b877a2366 100644 --- a/frontend/src/app/components/api_user/playlist.rs +++ b/frontend/src/app/components/api_user/playlist.rs @@ -68,7 +68,6 @@ fn to_playlist_cluster(count: (usize, usize, usize), bouquet: Option<&Rc Html { let translate = use_translation(); diff --git a/frontend/src/app/components/userlist/user_table.rs b/frontend/src/app/components/userlist/user_table.rs index ea821d92f..0099f68f9 100644 --- a/frontend/src/app/components/userlist/user_table.rs +++ b/frontend/src/app/components/userlist/user_table.rs @@ -197,9 +197,8 @@ pub fn UserTable(props: &UserTableProps) -> Html { }) }; - let is_sortable = Callback::::from(move |col| { - is_col_sortable(col) - }); + + let is_sortable = Callback::::from(is_col_sortable); let on_sort = { let users = props.users.clone();