From 555bc2cce1eb33f4e83a190d1dc3bc3b8f8eeb51 Mon Sep 17 00:00:00 2001 From: euzu <33094714+euzu@users.noreply.github.com> Date: Tue, 10 Feb 2026 11:53:42 +0100 Subject: [PATCH] Feature/provider config (#570) New Features - Automatic provider failover & rotation across multiple provider URLs with a provider:// scheme and capped retries - Configurable failover redirect patterns to influence retry/failover behavior UI - New interface to view and edit failover redirect patterns in reverse proxy configuration - Expanded docs covering Provider Failover & Rotation and enriched messaging templating Reliability - Improved URL resolution, error reporting and logging for more robust provider handling --- CHANGELOG.md | 4 + README.md | 57 +++- backend/src/api/api_utils.rs | 63 ++-- backend/src/api/endpoints/v1_api.rs | 1 + .../src/api/model/provider_lineup_manager.rs | 1 + .../api/model/streams/active_client_stream.rs | 6 +- .../model/streams/provider_stream_factory.rs | 64 +++- backend/src/api/panel_api.rs | 52 ++-- backend/src/messaging.rs | 1 + backend/src/model/config/input.rs | 251 +++++++++++++++- backend/src/model/config/reverse_proxy.rs | 31 +- backend/src/model/config/source.rs | 84 +++++- backend/src/model/input_source.rs | 62 ++-- backend/src/model/xmltv.rs | 1 + backend/src/repository/alias_repository.rs | 10 +- backend/src/utils/network/request.rs | 284 ++++++++++++------ backend/src/utils/network/xtream.rs | 25 +- frontend/public/assets/i18n/en.json | 5 +- .../config/reverse_proxy_config_view.rs | 62 +++- shared/src/error/tuliprox_error.rs | 11 +- shared/src/model/config/input.rs | 48 ++- shared/src/model/config/reverse_proxy.rs | 23 +- shared/src/model/config/source.rs | 25 +- shared/src/utils/request.rs | 23 +- 24 files changed, 1006 insertions(+), 188 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9ec8fc30f..085a8395f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -50,6 +50,10 @@ - **Added `disk_based_processing`**: (boolean, default `false`) to `config.yml`. When enabled, input playlists are processed from disk instead of memory. - **User-Agent `default_user_agent`**: Ensures that outgoing requests always pass a default user agent. - **Streaming**: Added `grace_period_hold_stream` configuration option to delay stream output until grace period connection checks are completed. +- **Provider Failover & Rotation**: Tuliprox supports robust failover mechanisms for streaming providers. + You can use the special `provider:///...` URL scheme in your configurations. Tuliprox will automatically resolve this to the current active URL of the specified provider. + If the current URL fails (e.g., 5xx error, timeout), Tuliprox automatically rotates to the next available URL for that provider. + It tracks failures and prevents infinite loops by limiting attempts to the number of available URLs. - Added `epg_request_timeshift: [-+]hh:mm or TimeZone`, example `Europe/Paris`, `America/New_York`, `-2:30`(-2h30m), `+0:15` (15m), `2` (2h), `:30` (30m), `:3` (3m) ## ⚙️ Engine & Storage Optimizations diff --git a/README.md b/README.md index cad448ccf..cfc20e4a5 100644 --- a/README.md +++ b/README.md @@ -121,7 +121,60 @@ With this configuration, you should create a `data` directory where you execute Be aware that different configurations (e.g. user bouquets) along the playlists are stored in this directory. -### 1.4 `messaging` +## 1.4 Provider Failover & Rotation +Tuliprox supports robust failover mechanisms for streaming providers. If a provider has multiple URLs defined (or aliases), Tuliprox can automatically rotate between them in case of failures. + +### 1.4.1 `provider://` Scheme +You can use the special `provider:///...` URL scheme in your configurations. Tuliprox will automatically resolve this to the current active URL of the specified provider. +- If the current URL fails (e.g., 5xx error, timeout), Tuliprox automatically rotates to the next available URL for that provider. +- It tracks failures and prevents infinite loops by limiting attempts to the number of available URLs. + +### 1.4.2 Automatic Failover triggers +Failover is triggered automatically on: +- Network Timeouts +- Request Timeout (408) +- Server Errors (500, 502, 503, 504) +- Specific Client Errors (404 Not Found, 410 Gone, 429 Too Many Requests) + +It does **not** trigger on Authentication errors (401, 403), as those usually indicate invalid credentials rather than a server issue. + +### 1.4.3 Provider Failover Configuration Example +Define a provider with multiple URLs and reference it from your inputs/sources. Tuliprox will resolve the active URL and rotate to the next entry on failover conditions. + +```yaml +templates: + - name: ALL_CHANNELS + value: Group ~ ".*" +provider: + - name: my_provider + urls: + - http://hello.provider.me + - http://stable.golden-bridge.con + - http://sleep.time.now.net +inputs: + - name: my_input + type: xtream_batch + headers: + User-Agent: TiviMate/5.1.6 (Android 12) + url: provider://my_provider # the name is the same as defined in provider: section + cache_duration: 1d + priority: 0 + max_connections: 0 + method: GET +sources: + - inputs: + - my_input + targets: + - name: my_channels + filter: "!ALL_CHANNELS!" + output: + - type: xtream + - type: m3u +``` + +--- + +### 1.5 `messaging` `messaging` is an optional configuration for receiving messages. Currently `telegram`, `discord`, `rest` and `pushover.net` is supported. @@ -167,7 +220,7 @@ messaging: url: `optional`, default is `https://api.pushover.net/1/messages.json` ``` -### 1.4.1 Messaging Templating +### 1.5.1 Messaging Templating For `discord`, `telegram` and `rest` messaging, you can use [Handlebars](https://handlebarsjs.com/) templates to format the message body. **Loading Templates:** diff --git a/backend/src/api/api_utils.rs b/backend/src/api/api_utils.rs index a75776528..a2ce14c90 100644 --- a/backend/src/api/api_utils.rs +++ b/backend/src/api/api_utils.rs @@ -10,7 +10,7 @@ use crate::auth::Fingerprint; use crate::model::{ConfigInput}; use crate::model::{ConfigTarget, ProxyUserCredentials}; use crate::tools::lru_cache::LRUResourceCache; -use crate::utils::request::{content_type_from_ext, parse_range, send_with_retry}; +use crate::utils::request::{content_type_from_ext, parse_range, send_with_retry_and_provider}; use crate::utils::{async_file_reader, async_file_writer, create_new_file_for_write, get_file_extension}; use crate::utils::{debug_if_enabled, trace_if_enabled}; use crate::utils::request; @@ -26,7 +26,7 @@ use futures::{stream, StreamExt, TryStreamExt}; use jsonwebtoken::{decode, Algorithm, DecodingKey, Validation}; use log::{debug, error, info, log_enabled, trace, warn}; use serde::Serialize; -use shared::concat_string; +use shared::{concat_string}; use shared::model::{Claims, InputFetchMethod, PlaylistEntry, PlaylistItemType, ProxyType, StreamChannel, TargetType, UserConnectionPermission, VirtualId, XtreamCluster}; use shared::utils::{bin_serialize, human_readable_kbps, trim_slash, Internable, CONTENT_TYPE_CBOR}; use shared::utils::{ @@ -132,6 +132,7 @@ pub use try_result_bad_request; pub use try_result_not_found; pub use try_unwrap_body; pub use internal_server_error; +use shared::error::TuliproxError; use crate::api::panel_api::{can_provision_on_exhausted, create_panel_api_provisioning_stream_details}; pub fn get_server_time() -> String { @@ -438,13 +439,14 @@ async fn create_stream_response_details( stream_url: &str, fingerprint: &Fingerprint, req_headers: &HeaderMap, - input: &ConfigInput, + input: &Arc, item_type: PlaylistItemType, share_stream: bool, connection_permission: UserConnectionPermission, force_provider: Option<&Arc>, virtual_id: VirtualId, -) -> StreamDetails { +) -> Result { + let mut streaming_strategy = resolve_streaming_strategy(app_state, stream_url, fingerprint, input, force_provider).await; let mut grace_period_options = app_state.get_grace_options(); grace_period_options.period_millis = get_grace_period_millis( @@ -471,21 +473,21 @@ async fn create_stream_response_details( "panel_api: provider connections exhausted; sending provisioning stream for input {}", sanitize_sensitive_info(&input.name) ); - return create_panel_api_provisioning_stream_details( + return Ok(create_panel_api_provisioning_stream_details( app_state, input, guard_provider_name.clone(), &grace_period_options, fingerprint.addr, virtual_id, - ); + )); } match streaming_strategy.provider_stream_state { // custom stream means we display our own stream like connection exhausted, channel-unavailable... ProviderStreamState::Custom(provider_stream) => { let (stream, stream_info) = provider_stream; - StreamDetails { + Ok(StreamDetails { stream, stream_info, provider_name: guard_provider_name.clone(), @@ -493,7 +495,7 @@ async fn create_stream_response_details( disable_provider_grace: false, reconnect_flag: None, provider_handle: streaming_strategy.provider_handle.clone(), - } + }) } ProviderStreamState::Available(_provider_name, request_url) | ProviderStreamState::GracePeriod(_provider_name, request_url) => { @@ -501,7 +503,7 @@ async fn create_stream_response_details( let ((stream, stream_info), reconnect_flag) = if let Ok(url) = parsed_url { let default_user_agent = app_state.app_config.config.load().default_user_agent.clone(); let disabled_headers = app_state.get_disabled_headers(); - let provider_stream_factory_options = ProviderStreamFactoryOptions::new( + let mut provider_stream_factory_options = ProviderStreamFactoryOptions::new( fingerprint.addr, item_type, share_stream, @@ -512,6 +514,10 @@ async fn create_stream_response_details( disabled_headers.as_ref(), default_user_agent.as_deref(), ); + + let provider_config = input.get_resolve_provider(url.as_ref()); + provider_stream_factory_options.set_provider(provider_config); + let reconnect_flag = provider_stream_factory_options.get_reconnect_flag_clone(); let provider_stream = match create_provider_stream( app_state, @@ -551,7 +557,7 @@ async fn create_stream_response_details( streaming_strategy.provider_handle.take() }; - StreamDetails { + Ok(StreamDetails { stream, stream_info, provider_name: guard_provider_name.clone(), @@ -559,7 +565,7 @@ async fn create_stream_response_details( disable_provider_grace: false, reconnect_flag, provider_handle, - } + }) } } } @@ -749,7 +755,7 @@ pub async fn force_provider_stream_response( user_session: &UserSession, mut stream_channel: StreamChannel, req_headers: &HeaderMap, - input: &ConfigInput, + input: &Arc, user: &ProxyUserCredentials, ) -> impl IntoResponse + Send { let stream_options = get_stream_options(app_state); @@ -761,7 +767,7 @@ 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 stream_details = create_stream_response_details( + let stream_details = match create_stream_response_details( app_state, &stream_options, &user_session.stream_url, @@ -774,7 +780,13 @@ pub async fn force_provider_stream_response( Some(&user_session.provider), stream_channel.virtual_id, ) - .await; + .await { + Ok(stream_details) => stream_details, + Err(err) => { + error!("Failed to stream: {err}"); + return StatusCode::INTERNAL_SERVER_ERROR.into_response(); + } + }; if stream_details.has_stream() { let provider_response = stream_details @@ -830,8 +842,8 @@ pub async fn stream_response( mut stream_channel: StreamChannel, stream_url: &str, req_headers: &HeaderMap, - input: &ConfigInput, - target: &ConfigTarget, + input: &Arc, + target: &Arc, user: &ProxyUserCredentials, connection_permission: UserConnectionPermission, ) -> impl IntoResponse + Send { @@ -866,7 +878,7 @@ pub async fn stream_response( }; let stream_options = get_stream_options(app_state); - let mut stream_details = create_stream_response_details( + let mut stream_details = match create_stream_response_details( app_state, &stream_options, stream_url, @@ -878,7 +890,13 @@ pub async fn stream_response( connection_permission, None, stream_channel.virtual_id, - ).await; + ).await{ + Ok(stream_details) => stream_details, + Err(err) => { + error!("Failed to stream: {err}"); + return StatusCode::INTERNAL_SERVER_ERROR.into_response(); + } + }; if stream_details.has_stream() { // let content_length = get_stream_content_length(provider_response.as_ref()); @@ -1358,15 +1376,18 @@ async fn fetch_resource_with_retry( let disabled_headers = app_state.get_disabled_headers(); - let Ok(response) = send_with_retry( + let provider_config = input.and_then(|i| i.get_resolve_provider(url.as_str())); + let Ok(response) = send_with_retry_and_provider( &app_state.app_config, url, - || { + provider_config.as_ref(), + false, + |resolved_url| { request::get_client_request( &app_state.http_client.load(), input.map_or(InputFetchMethod::GET, |i| i.method), input.map(|i| &i.headers), - url, + resolved_url, Some(req_headers), disabled_headers.as_ref(), default_user_agent.as_deref(), diff --git a/backend/src/api/endpoints/v1_api.rs b/backend/src/api/endpoints/v1_api.rs index 7f10af99e..40affdce4 100644 --- a/backend/src/api/endpoints/v1_api.rs +++ b/backend/src/api/endpoints/v1_api.rs @@ -85,6 +85,7 @@ async fn geoip_update(axum::extract::State(app_state): axum::extract::State>, range_requested: bool, reconnect_flag: Arc, + provider: Option>, } impl ProviderStreamFactoryOptions { @@ -103,9 +104,18 @@ impl ProviderStreamFactoryOptions { default_user_agent, range_bytes, range_requested: requested_range.is_some(), + provider: None, } } + pub fn set_provider(&mut self, provider: Option>) { + self.provider = provider; + } + + pub fn get_provider(&self) -> Option<&Arc> { + self.provider.as_ref() + } + #[inline] fn is_piped(&self) -> bool { self.pipe_stream @@ -358,17 +368,36 @@ fn should_use_manual_redirects(app_state: &Arc) -> bool { async fn send_with_manual_redirects( request_client: &reqwest::Client, stream_options: &ProviderStreamFactoryOptions, -) -> Result { + app_state: &Arc, +) -> Result { let mut current_url = stream_options.get_url().clone(); - let mut remaining = 10u8; + let mut remaining_redirects = 10u8; + let provider = stream_options.get_provider().cloned(); loop { - let (client, _partial_content) = prepare_client(request_client, stream_options, Some(¤t_url)); - let response = client.send().await?; + let result = send_with_retry_and_provider( + &app_state.app_config, + ¤t_url, + provider.as_ref(), + true, + |resolved_url| prepare_client(request_client, stream_options, Some(resolved_url)).0 + ).await; + + let response = match result { + Ok(resp) => resp, + Err(e) => { + // send_with_retry_and_provider already applies provider failover policy. + // Do not rotate again here, otherwise non-failover errors (e.g. auth) may + // incorrectly switch provider URLs. + debug!("Manual redirect failed: {}", sanitize_sensitive_info(e.to_string().as_str())); + return Err(e); + } + }; + let status = response.status(); if status.is_redirection() { - if remaining == 0 { + if remaining_redirects == 0 { return Ok(response); } let location = response.headers().get(reqwest::header::LOCATION); @@ -385,13 +414,14 @@ async fn send_with_manual_redirects( return Ok(response); }; current_url = next_url; - remaining = remaining.saturating_sub(1); + remaining_redirects = remaining_redirects.saturating_sub(1); continue; } return Ok(response); } } +#[allow(clippy::too_many_lines)] async fn provider_stream_request( app_state: &Arc, request_client: &reqwest::Client, @@ -399,10 +429,22 @@ async fn provider_stream_request( ) -> Result, StatusCode> { let response_result = if should_use_manual_redirects(app_state) { let client_no_redirect = app_state.http_client_no_redirect.load(); - send_with_manual_redirects(&client_no_redirect, stream_options).await + send_with_manual_redirects(&client_no_redirect, stream_options, app_state).await } else { - let (client, _partial_content) = prepare_client(request_client, stream_options, None); - client.send().await + // Use send_with_retry_and_provider for automatic failover support + let url = stream_options.get_url(); + let provider = stream_options.get_provider().cloned(); + + send_with_retry_and_provider( + &app_state.app_config, + url, + provider.as_ref(), + false, + |resolved_url| { + let (client, _partial_content) = prepare_client(request_client, stream_options, Some(resolved_url)); + client + } + ).await }; match response_result { Ok(mut response) => { diff --git a/backend/src/api/panel_api.rs b/backend/src/api/panel_api.rs index 83da6e887..792fda3d6 100644 --- a/backend/src/api/panel_api.rs +++ b/backend/src/api/panel_api.rs @@ -736,12 +736,21 @@ async fn panel_client_info( } async fn fetch_root_user_api_info( - app_state: &AppState, + app_state: &Arc, input: &ConfigInput, -) -> Option { - let (username, password) = extract_account_creds_from_input(input)?; - let base_url = get_base_url_from_str(input.url.as_str()).unwrap_or_else(|| input.url.clone()); - let mut url = Url::parse(base_url.as_str()).ok()?; +) -> Result, TuliproxError> { + + + let Some((username, password)) = extract_account_creds_from_input(input) else { + return Ok(None); + }; + + let resolved_url = input.resolve()?; + let base_url = get_base_url_from_str(&resolved_url).unwrap_or_else(|| resolved_url.to_string()); + + let Ok(mut url) = Url::parse(base_url.as_str()) else { + return info_err_res!("panel_api: invalid base_url: {}", sanitize_sensitive_info(&base_url)); + }; url.set_path("/player_api.php"); { let mut pairs = url.query_pairs_mut(); @@ -758,7 +767,7 @@ async fn fetch_root_user_api_info( sanitize_sensitive_info(&input.name), sanitize_sensitive_info(err.to_string().as_str()) ); - return None; + return Ok(None); } }; @@ -776,11 +785,11 @@ async fn fetch_root_user_api_info( .and_then(get_string_from_serde_value) .and_then(|tz| tz.parse::().ok()); - Some(UserApiAccountInfo { + Ok(Some(UserApiAccountInfo { exp_date, server_now_ts, server_tz, - }) + })) } fn resolve_panel_expire_mode( @@ -2455,8 +2464,8 @@ async fn sync_panel_api_for_input_on_boot( } if let Some((root_username, root_password)) = extract_account_creds_from_input(input.as_ref()) { - let user_info = fetch_root_user_api_info(app_state.as_ref(), input.as_ref()).await; - let (root_exp_date, server_tz, skew_secs) = if let Some(info) = user_info { + let user_info = fetch_root_user_api_info(app_state, input.as_ref()).await; + let (root_exp_date, server_tz, skew_secs) = if let Ok(Some(info)) = user_info { let local_now = i64::try_from(get_current_timestamp()).unwrap_or(0); let skew_secs = info.server_now_ts.unwrap_or(local_now) - local_now; (info.exp_date, info.server_tz, Some(skew_secs)) @@ -3938,7 +3947,7 @@ pub(crate) async fn run_panel_api_provisioning_probe( stop_signal: Arc, addr: SocketAddr, virtual_id: VirtualId, -) { +) -> Result<(), TuliproxError> { let Some(panel_cfg) = input.panel_api.as_ref() else { debug_if_enabled!( "panel_api provisioning probe skipped (missing config) for input {}", @@ -3949,7 +3958,7 @@ pub(crate) async fn run_panel_api_provisioning_probe( .connection_manager .kick_connection(&addr, virtual_id, 0) .await; - return; + return Ok(()); }; if !panel_cfg.enabled { debug_if_enabled!( @@ -3961,7 +3970,7 @@ pub(crate) async fn run_panel_api_provisioning_probe( .connection_manager .kick_connection(&addr, virtual_id, 0) .await; - return; + return Ok(()); } if panel_cfg.url.trim().is_empty() { debug_if_enabled!( @@ -3973,7 +3982,7 @@ pub(crate) async fn run_panel_api_provisioning_probe( .connection_manager .kick_connection(&addr, virtual_id, 0) .await; - return; + return Ok(()); } let max_wait_secs = panel_cfg.provisioning.timeout_sec; @@ -4023,10 +4032,12 @@ pub(crate) async fn run_panel_api_provisioning_probe( .connection_manager .kick_connection(&addr, virtual_id, 0) .await; - return; + return Ok(()); }; - let Some(test_url) = build_panel_api_test_url(input.url.as_str(), username, password) else { + let resolved_url = input.resolve()?; + + let Some(test_url) = build_panel_api_test_url(&resolved_url, username, password) else { if max_wait_secs > 0 { tokio::time::sleep(Duration::from_secs(max_wait_secs)).await; } @@ -4039,7 +4050,7 @@ pub(crate) async fn run_panel_api_provisioning_probe( .connection_manager .kick_connection(&addr, virtual_id, 0) .await; - return; + return Ok(()); }; let probe_delay = Duration::from_secs(probe_interval_secs); @@ -4111,6 +4122,7 @@ pub(crate) async fn run_panel_api_provisioning_probe( .connection_manager .kick_connection(&addr, virtual_id, 0) .await; + Ok(()) } pub fn create_panel_api_provisioning_stream_details( @@ -4151,14 +4163,16 @@ pub fn create_panel_api_provisioning_stream_details( let input_clone = input.clone(); let stop_clone = Arc::clone(&stop_signal); tokio::spawn(async move { - run_panel_api_provisioning_probe( + if let Err(err) = run_panel_api_provisioning_probe( app_state_clone, input_clone, stop_clone, addr, virtual_id, ) - .await; + .await { + error!("Error running Probe: {err:?}"); + } }); StreamDetails { diff --git a/backend/src/messaging.rs b/backend/src/messaging.rs index 096c83c69..1c956c7c4 100644 --- a/backend/src/messaging.rs +++ b/backend/src/messaging.rs @@ -260,6 +260,7 @@ async fn resolve_template<'a>(app_config: &'a Arc, http_client: &'a r let input_source = InputSource { name: "Template".intern(), url, + provider: None, username: None, password: None, method: InputFetchMethod::GET, diff --git a/backend/src/model/config/input.rs b/backend/src/model/config/input.rs index 8656474fe..4613d19fb 100644 --- a/backend/src/model/config/input.rs +++ b/backend/src/model/config/input.rs @@ -1,12 +1,13 @@ -use crate::model::{macros, EpgConfig, PanelApiConfig}; +use crate::model::{macros, ConfigProvider, EpgConfig, PanelApiConfig}; use crate::repository::get_csv_file_path; use chrono::Utc; use log::warn; -use shared::check_input_credentials; use shared::error::TuliproxError; use shared::model::{ConfigInputAliasDto, ConfigInputDto, ConfigInputOptionsDto, InputFetchMethod, InputType, StagedInputDto}; -use shared::utils::{get_credentials_from_url, Internable}; +use shared::utils::{get_credentials_from_url, parse_provider_scheme_url_parts, sanitize_sensitive_info, Internable, PROVIDER_SCHEME_PREFIX}; use shared::{check_input_connections, info_err_res, write_if_some}; +use shared::{check_input_credentials, concat_string, info_err}; +use std::borrow::Cow; use std::collections::HashMap; use std::fmt; use std::path::PathBuf; @@ -78,6 +79,8 @@ pub struct StagedInput { pub method: InputFetchMethod, pub input_type: InputType, pub headers: HashMap, + /// Provider configuration for failover support when using `provider://` scheme. + pub provider_config: Option>, } macros::from_impl!(StagedInput); @@ -91,6 +94,7 @@ impl From<&StagedInputDto> for StagedInput { password: dto.password.clone(), method: dto.method, headers: dto.headers.clone(), + provider_config: None, // Resolved later in ConfigInput::prepare() } } } @@ -147,16 +151,46 @@ pub struct ConfigInput { pub t_batch_url: Option, pub panel_api: Option, pub cache_duration_seconds: u64, + pub provider_configs: Option>>, } impl ConfigInput { - pub fn prepare(&mut self) -> Result, TuliproxError> { + pub fn prepare(&mut self, provider_configs: &[Arc]) -> Result, TuliproxError> { + let mut used_provider_configs: Vec> = vec![]; let batch_file_path = self.prepare_batch(); self.name = self.name.trim().intern(); + + let resolve_provider_config = |url: &str| -> Result, TuliproxError> { + let (host, _path) = parse_provider_scheme_url_parts(url).map_err(|err| { + info_err!( + "Malformed provider URL {}: {}", + sanitize_sensitive_info(url), + sanitize_sensitive_info(err.to_string().as_str()) + ) + })?; + + provider_configs + .iter() + .find(|p| p.name.as_ref() == host) + .cloned() + .ok_or_else(|| info_err!("Failed to resolve provider config for {}", sanitize_sensitive_info(url))) + }; + + if self.url.starts_with(PROVIDER_SCHEME_PREFIX) { + let provider_cfg = resolve_provider_config(&self.url)?; + used_provider_configs.push(provider_cfg); + } + if self.enabled { check_input_credentials!(self, self.input_type, false, false); check_input_connections!(self, self.input_type, false); if let Some(staged_input) = &mut self.staged { + if staged_input.url.starts_with(PROVIDER_SCHEME_PREFIX) { + let provider_cfg = resolve_provider_config(&staged_input.url)?; + staged_input.provider_config = Some(provider_cfg.clone()); + used_provider_configs.push(provider_cfg); + } + check_input_credentials!(staged_input, staged_input.input_type, false, true); if !matches!(staged_input.input_type, InputType::M3u | InputType::Xtream) { return info_err_res!("Staged input can only be from type m3u or xtream"); @@ -174,9 +208,20 @@ impl ConfigInput { warn!("Account {} expired for provider: {}", alias.username.as_ref().map_or("?", |s| s.as_str()), alias.name); alias.enabled = false; } + + if alias.url.starts_with(PROVIDER_SCHEME_PREFIX) { + let provider_cfg = resolve_provider_config(&alias.url)?; + if !used_provider_configs.iter().any(|p| p.name == provider_cfg.name) { + used_provider_configs.push(provider_cfg); + } + } } } + if !used_provider_configs.is_empty() { + self.provider_configs = Some(used_provider_configs); + } + if let Some(panel_api) = &mut self.panel_api { panel_api.prepare()?; } @@ -214,7 +259,6 @@ impl ConfigInput { self.t_batch_url = Some(self.url.clone()); let file_path = get_csv_file_path(self.url.as_str()).ok(); if self.enabled { - if let Some(aliases) = self.aliases.as_mut() { if !aliases.is_empty() { for alias in aliases.iter_mut() { @@ -273,6 +317,7 @@ impl ConfigInput { t_batch_url: None, panel_api: self.panel_api.clone(), cache_duration_seconds: self.cache_duration_seconds, + provider_configs: self.provider_configs.clone(), } } @@ -292,6 +337,42 @@ impl ConfigInput { } }) } + + pub fn resolve_url<'a>(&self, url: &'a str) -> Result, TuliproxError> { + if !url.starts_with(PROVIDER_SCHEME_PREFIX) { + return Ok(Cow::Borrowed(url)); + } + + let (host, _path) = parse_provider_scheme_url_parts(url)?; + + let provider_config = self.provider_configs + .as_ref() + .and_then(|configs| configs.iter().find(|p| p.name.as_ref() == host)) + .cloned(); + + if let Some(provider) = provider_config { + let (_, resolved) = resolve_provider_scheme_url_with_provider(url, Some(provider))?; + Ok(resolved) + } else { + info_err_res!("Provider config for '{}' not found in input '{}'", host, self.name) + } + } + + pub fn resolve(&self) -> Result, TuliproxError> { + self.resolve_url(&self.url) + } + + pub fn get_resolve_provider(&self, url: &str) -> Option> { + if !url.starts_with(PROVIDER_SCHEME_PREFIX) { + return None; + } + if let Some(provider) = self.provider_configs.as_ref() { + if let Ok((host, _path)) = parse_provider_scheme_url_parts(url) { + return provider.iter().find(|pc| pc.name.as_ref() == host).cloned(); + } + } + None + } } macros::from_impl!(ConfigInput); @@ -318,6 +399,7 @@ impl From<&ConfigInputDto> for ConfigInput { t_batch_url: None, panel_api: dto.panel_api.as_ref().map(PanelApiConfig::from), cache_duration_seconds: dto.cache_duration_seconds, + provider_configs: None, } } } @@ -347,6 +429,7 @@ impl fmt::Display for ConfigInput { } } + pub fn is_input_expired(exp_date: Option) -> bool { match exp_date { Some(ts) => { @@ -356,3 +439,161 @@ pub fn is_input_expired(exp_date: Option) -> bool { None => false, } } + +/// Resolves a custom "provider://" URL using a pre-provided provider configuration. +/// If the URL does not use the custom scheme, it returns the original URL. +pub fn resolve_provider_scheme_url_with_provider( + stream_url: &str, + provider_config: Option>, +) -> Result<(Option>, Cow<'_, str>), TuliproxError> { + if !stream_url.starts_with(PROVIDER_SCHEME_PREFIX) { + return Ok((None, Cow::Borrowed(stream_url))); + } + + let (_host, path_and_query) = parse_provider_scheme_url_parts(stream_url)?; + + let provider = provider_config.ok_or_else(|| { + info_err!("Provider config missing for resolution of: '{}'", sanitize_sensitive_info(stream_url)) + })?; + + let final_url = assemble_provider_url(&provider, path_and_query)?; + Ok((Some(provider), Cow::Owned(final_url))) +} + +/// Internal helper to build the final URL string +fn assemble_provider_url(provider: &ConfigProvider, path_and_query: &str) -> Result { + let base = provider.get_current_url() + .ok_or_else(|| info_err!("Provider '{}' has no URLs available", provider.name))?; + + // Add http:// scheme if no scheme is present + let base_with_scheme = if base.contains("://") { + base.to_string() + } else { + concat_string!("http://", base) + }; + + let mut final_url = base_with_scheme.trim_end_matches('/').to_string(); + if !path_and_query.is_empty() { + final_url.push_str(path_and_query); + } + Ok(final_url) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::model::ConfigProvider; + use std::borrow::Cow; + use std::sync::atomic::AtomicUsize; + use std::sync::Arc; + + #[test] + fn test_resolve_url_normal() { + let input = ConfigInput { + url: "http://example.com/stream".to_string(), + ..Default::default() + }; + let resolved = input.resolve_url("http://example.com/stream").unwrap(); + assert_eq!(resolved, "http://example.com/stream"); + assert!(matches!(resolved, Cow::Borrowed(_))); + } + + #[test] + fn test_resolve_url_provider() { + let provider = ConfigProvider { + name: "myprovider".into(), + urls: vec!["http://provider.com".into()], + current_url_index: AtomicUsize::new(0), + }; + let input = ConfigInput { + name: "test_input".into(), + provider_configs: Some(vec![Arc::new(provider)]), + ..Default::default() + }; + + let resolved = input.resolve_url("provider://myprovider/stream").unwrap(); + assert_eq!(resolved, "http://provider.com/stream"); + assert!(matches!(resolved, Cow::Owned(_))); + } + + #[test] + fn test_resolve_url_provider_missing() { + let input = ConfigInput { + name: "test_input".into(), + provider_configs: Some(vec![]), + ..Default::default() + }; + + let err = input.resolve_url("provider://myprovider/stream").unwrap_err(); + assert!(err.to_string().contains("Provider config for 'myprovider' not found")); + } + + #[test] + fn test_resolve_default() { + let input = ConfigInput { + url: "http://example.com/stream".to_string(), + ..Default::default() + }; + let resolved = input.resolve().unwrap(); + assert_eq!(resolved, "http://example.com/stream"); + } + + #[test] + fn test_prepare_fails_on_malformed_provider_url_in_main_input() { + let mut input = ConfigInput { + name: "test_input".into(), + input_type: InputType::M3u, + url: "provider:///bad".to_string(), + enabled: false, + ..Default::default() + }; + + let err = input.prepare(&[]).unwrap_err(); + assert!(err.to_string().contains("Malformed provider URL")); + } + + #[test] + fn test_prepare_fails_on_malformed_provider_url_in_staged_input() { + let mut input = ConfigInput { + name: "test_input".into(), + input_type: InputType::M3u, + url: "http://example.com/playlist.m3u".to_string(), + enabled: true, + staged: Some(StagedInput { + name: "staged".into(), + input_type: InputType::M3u, + url: "provider:///bad".to_string(), + ..Default::default() + }), + ..Default::default() + }; + + let err = input.prepare(&[]).unwrap_err(); + assert!(err.to_string().contains("Malformed provider URL")); + } + + #[test] + fn test_prepare_fails_on_malformed_provider_url_in_alias() { + let mut input = ConfigInput { + name: "test_input".into(), + input_type: InputType::M3u, + url: "http://example.com/playlist.m3u".to_string(), + enabled: true, + aliases: Some(vec![ConfigInputAlias { + id: 1, + name: "alias".into(), + url: "provider:///bad".to_string(), + username: None, + password: None, + priority: 0, + max_connections: 0, + exp_date: None, + enabled: true, + }]), + ..Default::default() + }; + + let err = input.prepare(&[]).unwrap_err(); + assert!(err.to_string().contains("Malformed provider URL")); + } +} diff --git a/backend/src/model/config/reverse_proxy.rs b/backend/src/model/config/reverse_proxy.rs index 3e6a1994e..8e2770a60 100644 --- a/backend/src/model/config/reverse_proxy.rs +++ b/backend/src/model/config/reverse_proxy.rs @@ -1,8 +1,10 @@ use crate::model::config::cache::CacheConfig; use crate::model::{macros, GeoIpConfig, RateLimitConfig, StreamConfig}; -use shared::model::{ResourceRetryConfigDto, ReverseProxyConfigDto, ReverseProxyDisabledHeaderConfigDto}; +use regex::Regex; +use shared::model::{ResourceRetryConfigDto, ReverseProxyConfigDto, ReverseProxyDisabledHeaderConfigDto, REGEX_CACHE}; use shared::utils::{default_resource_retry_attempts, default_resource_retry_backoff_ms, default_resource_retry_backoff_multiplier, hex_to_u8_16, u8_16_to_hex}; use std::cmp::max; +use std::sync::Arc; #[derive(Debug, Clone)] pub struct ReverseProxyDisabledHeaderConfig { @@ -35,6 +37,7 @@ pub struct ResourceRetryConfig { pub max_attempts: u32, pub backoff_millis: u64, pub backoff_multiplier: f64, + pub failover_redirect_patterns: Vec>, } impl Default for ResourceRetryConfig { @@ -43,10 +46,16 @@ impl Default for ResourceRetryConfig { max_attempts: default_resource_retry_attempts(), backoff_millis: default_resource_retry_backoff_ms(), backoff_multiplier: default_resource_retry_backoff_multiplier(), + failover_redirect_patterns: default_failover_redirect_patterns(), } } } +/// Default failover redirect pattern when none is configured +fn default_failover_redirect_patterns() -> Vec> { + vec![REGEX_CACHE.get_or_compile("service-abuse").expect("default redirect failover regex must compile")] +} + impl ResourceRetryConfig { pub fn get_retry_values(&self) -> (u32, u64, f64) { ( @@ -78,20 +87,40 @@ impl From<&ResourceRetryConfigDto> for ResourceRetryConfig { } else { 1.0 }; + + // Compile patterns, default to service-abuse if none or empty + let patterns = dto.failover_redirect_patterns + .as_ref() + .filter(|v| !v.is_empty()) + .map_or_else(default_failover_redirect_patterns, |patterns| { + patterns.iter() + .filter_map(|p| REGEX_CACHE.get_or_compile(p).map_err(|e| { + log::warn!("Failed to compile failover redirect pattern '{p}': {e}"); + e + }).ok()) + .collect() + }); + Self { max_attempts: dto.max_attempts, backoff_millis: dto.backoff_millis, backoff_multiplier: multiplier, + failover_redirect_patterns: patterns, } } } impl From<&ResourceRetryConfig> for ResourceRetryConfigDto { fn from(cfg: &ResourceRetryConfig) -> Self { + let patterns: Vec = cfg.failover_redirect_patterns + .iter() + .map(|re| re.as_str().to_string()) + .collect(); Self { max_attempts: cfg.max_attempts, backoff_millis: cfg.backoff_millis, backoff_multiplier: cfg.backoff_multiplier, + failover_redirect_patterns: if patterns.is_empty() { None } else { Some(patterns) }, } } } diff --git a/backend/src/model/config/source.rs b/backend/src/model/config/source.rs index 8182895b4..ba9ed45a2 100644 --- a/backend/src/model/config/source.rs +++ b/backend/src/model/config/source.rs @@ -1,11 +1,82 @@ use crate::model::{macros, ConfigInput, ConfigTarget, ProcessTargets}; use shared::error::{info_err_res, TuliproxError}; -use shared::model::{ConfigSourceDto, PatternTemplate, SourcesConfigDto}; +use shared::model::{ConfigProviderDto, ConfigSourceDto, PatternTemplate, SourcesConfigDto}; use std::borrow::Cow; use std::collections::{HashMap, HashSet}; use std::path::PathBuf; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; +#[derive(Debug)] +pub struct ConfigProvider { + pub name: Arc, + pub urls: Vec>, + pub current_url_index: AtomicUsize, +} + +impl Clone for ConfigProvider { + fn clone(&self) -> Self { + Self { + name: self.name.clone(), + urls: self.urls.clone(), + current_url_index: AtomicUsize::new(self.current_url_index.load(Ordering::Relaxed)), + } + } +} + + +macros::from_impl!(ConfigProvider); +impl From<&ConfigProviderDto> for ConfigProvider { + fn from(dto: &ConfigProviderDto) -> Self { + Self { + name: dto.name.clone(), + urls: dto.urls.clone(), + current_url_index: AtomicUsize::new(0), + } + } +} + +impl ConfigProvider { + /// Gets the current URL from the provider + pub fn get_current_url(&self) -> Option<&Arc> { + let index = self.current_url_index.load(Ordering::Relaxed); + self.urls.get(index) + } + + /// Resets the current URL index to 0 + pub fn reset_index(&self) { + self.current_url_index.store(0, Ordering::Relaxed); + } + + /// Gets the current URL index + #[inline] + pub fn get_current_index(&self) -> usize { + self.current_url_index.load(Ordering::Relaxed) + } + + /// Rotates to next URL, checking if a full cycle has been completed. + /// Returns None if we've cycled back to the `start_index`, indicating all URLs were tried. + /// + /// Use this method when you need to try all URLs exactly once before failing. + /// Call `get_current_index()` at the start of a failover session to get the `start_index`. + pub fn rotate_to_next_url_with_cycle_check(&self, start_index: usize) -> Option<&Arc> { + let len = self.urls.len(); + if len == 0 { + return None; + } + + let previous = self.current_url_index + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| { + let next = (current + 1) % len; + // If we've cycled back to start, we've tried all URLs. + (next != start_index).then_some(next) + }) + .ok()?; + let next = (previous + 1) % len; + self.urls.get(next) + } +} + #[derive(Debug, Clone)] pub struct ConfigSource { pub inputs: Vec>, @@ -40,6 +111,7 @@ impl ConfigSource { pub struct SourcesConfig { pub batch_files: Vec, pub templates: Option>, + pub provider: Vec>, pub inputs: Vec>, pub sources: Vec, } @@ -51,11 +123,14 @@ impl TryFrom<&SourcesConfigDto> for SourcesConfig { let mut inputs = Vec::>::new(); let mut batch_files = Vec::::new(); let mut input_names = HashSet::new(); + let provider: Vec<_> = dto.provider.as_ref() + .map(|list| list.iter().map(ConfigProvider::from).map(Arc::new).collect()) + .unwrap_or_default(); for input_dto in &dto.inputs { let mut input = ConfigInput::from(input_dto); // Prepare input - if let Some(path) = input.prepare()? { + if let Some(path) = input.prepare(&provider)? { batch_files.push(path); } input_names.insert(input.name.clone()); @@ -76,6 +151,7 @@ impl TryFrom<&SourcesConfigDto> for SourcesConfig { Ok(Self { batch_files, templates: dto.templates.clone(), + provider, inputs, sources, }) @@ -179,4 +255,8 @@ impl SourcesConfig { pub fn get_input_by_name(&self, name: &Arc) -> Option<&Arc> { self.inputs.iter().find(|i| &i.name == name) } + + pub fn get_provider_by_name(&self, name: &str) -> Option<&Arc> { + self.provider.iter().find(|p| p.name.as_ref() == name) + } } diff --git a/backend/src/model/input_source.rs b/backend/src/model/input_source.rs index fc5daab63..99d2f259c 100644 --- a/backend/src/model/input_source.rs +++ b/backend/src/model/input_source.rs @@ -1,12 +1,19 @@ use std::collections::HashMap; use std::sync::Arc; use shared::model::InputFetchMethod; -use crate::model::{ConfigInput, StagedInput}; +use crate::model::{ConfigInput, ConfigProvider, StagedInput}; +/// Represents an input source for fetching content. +/// +/// When created from a `ConfigInput` that uses the `provider://` scheme, +/// the provider context is preserved to enable URL failover on errors. #[derive(Clone, Debug)] pub struct InputSource { pub name: Arc, pub url: String, + /// The provider associated with this input, if the URL uses `provider://` scheme. + /// This enables failover to alternative URLs when the current URL fails. + pub provider: Option>, pub username: Option, pub password: Option, pub method: InputFetchMethod, @@ -14,34 +21,55 @@ pub struct InputSource { } impl InputSource { + /// Creates a new `InputSource` with a different URL while preserving provider context. + /// + /// The provider is preserved so that failover can occur even when the URL + /// is derived from the original (e.g., adding query parameters, changing paths). pub fn with_url(&self, url: String) -> Self { Self { name: self.name.clone(), url, + provider: self.provider.clone(), username: self.username.clone(), password: self.password.clone(), method: self.method, headers: self.headers.clone(), } } + + /// Returns the provider for this input source, if available. + /// + /// This can be used to pass to `send_with_retry_and_provider` for failover support. + #[inline] + pub fn get_provider(&self) -> Option<&Arc> { + self.provider.as_ref() + } } -macro_rules! impl_input_source_from { - ($input_type:ty) => { - impl From<&$input_type> for InputSource { - fn from(input: &$input_type) -> Self { - Self { - name: input.name.clone(), - url: input.url.clone(), - username: input.username.clone(), - password: input.password.clone(), - method: input.method, - headers: input.headers.clone(), - } - } +impl From<&ConfigInput> for InputSource { + fn from(input: &ConfigInput) -> Self { + Self { + name: input.name.clone(), + url: input.url.clone(), + provider: input.get_resolve_provider(&input.url), + username: input.username.clone(), + password: input.password.clone(), + method: input.method, + headers: input.headers.clone(), } - }; + } } -impl_input_source_from!(ConfigInput); -impl_input_source_from!(StagedInput); \ No newline at end of file +impl From<&StagedInput> for InputSource { + fn from(input: &StagedInput) -> Self { + Self { + name: input.name.clone(), + url: input.url.clone(), + provider: input.provider_config.clone(), + username: input.username.clone(), + password: input.password.clone(), + method: input.method, + headers: input.headers.clone(), + } + } +} \ No newline at end of file diff --git a/backend/src/model/xmltv.rs b/backend/src/model/xmltv.rs index 0176d6495..c4870cb41 100644 --- a/backend/src/model/xmltv.rs +++ b/backend/src/model/xmltv.rs @@ -128,6 +128,7 @@ pub async fn parse_xmltv_for_web_ui_from_url(app_state: &Arc, url: &st let input_source: InputSource = InputSource { name: "xmltv".intern(), url: request_url.to_string(), + provider: None, username: None, password: None, method: InputFetchMethod::GET, diff --git a/backend/src/repository/alias_repository.rs b/backend/src/repository/alias_repository.rs index 06bac3fbf..fdb4a423a 100644 --- a/backend/src/repository/alias_repository.rs +++ b/backend/src/repository/alias_repository.rs @@ -7,7 +7,7 @@ use log::{error, warn}; use shared::error::{string_to_io_error, to_io_error, TuliproxError}; use shared::info_err; use shared::model::{ConfigInputAliasDto, InputType}; -use shared::utils::{get_credentials_from_url, get_credentials_from_url_str, parse_timestamp, sanitize_sensitive_info, Internable}; +use shared::utils::{get_credentials_from_url, get_credentials_from_url_str, parse_timestamp, sanitize_sensitive_info, Internable, PROVIDER_SCHEME_PREFIX}; use std::io; use std::io::{BufRead, Cursor, Error}; use std::path::{Path, PathBuf}; @@ -61,6 +61,11 @@ fn build_m3u_url( fn csv_assign_mandatory_fields(alias: &mut ConfigInputAliasDto, input_type: InputType) { if !alias.url.is_empty() { + let mut provider_scheme = false; + if alias.url.starts_with(PROVIDER_SCHEME_PREFIX) { + provider_scheme = true; + alias.url = alias.url.replacen(PROVIDER_SCHEME_PREFIX, "http://", 1); + } match Url::parse(alias.url.as_str()) { Ok(url) => { let (username, password) = get_credentials_from_url(&url); @@ -107,6 +112,9 @@ fn csv_assign_mandatory_fields(alias: &mut ConfigInputAliasDto, input_type: Inpu warn!("Could not parse URL '{}' for alias: {err}", sanitize_sensitive_info(&alias.url)); } } + if provider_scheme { + alias.url = alias.url.replacen("http://", PROVIDER_SCHEME_PREFIX, 1); + } } } diff --git a/backend/src/utils/network/request.rs b/backend/src/utils/network/request.rs index 72189d3e0..fe087bff1 100644 --- a/backend/src/utils/network/request.rs +++ b/backend/src/utils/network/request.rs @@ -1,29 +1,27 @@ use crate::api::model::persist_pipe_stream::tee_dyn_reader; use crate::api::model::{AppState, STREAM_IDLE_TIMEOUT}; -use crate::model::{format_elapsed_time, AppConfig, Config, InputSource, ReverseProxyDisabledHeaderConfig}; -use crate::model::{ConfigInput, ResourceRetryConfig}; +use crate::model::{format_elapsed_time, AppConfig, Config, ConfigProvider, InputSource, ReverseProxyDisabledHeaderConfig}; +use crate::model::{resolve_provider_scheme_url_with_provider, ConfigInput, ResourceRetryConfig}; use crate::utils::compression::compression_utils::{is_deflate, is_gzip}; use crate::utils::{async_file_reader, async_file_writer, debug_if_enabled}; use crate::utils::{get_file_path, persist_file}; -use axum::http::header::RETRY_AFTER; use futures::{StreamExt, TryStreamExt}; use log::{debug, error, log_enabled, trace, warn, Level}; use reqwest::header::CONTENT_ENCODING; -use reqwest::redirect::Policy; use reqwest::header::{HeaderMap, HeaderName, HeaderValue}; -use reqwest::{StatusCode}; +use reqwest::redirect::Policy; +use reqwest::StatusCode; use shared::error::{notify_err_res, string_to_io_error, TuliproxError}; use shared::model::{InputFetchMethod, DEFAULT_USER_AGENT}; -use shared::utils::{ - filter_request_header, human_readable_byte_size, sanitize_sensitive_info, ENCODING_DEFLATE, - ENCODING_GZIP, -}; +use shared::utils::{filter_request_header, human_readable_byte_size, + sanitize_sensitive_info, ENCODING_DEFLATE, ENCODING_GZIP}; use std::collections::HashMap; use std::io::{Error, ErrorKind}; use std::path::{Path, PathBuf}; use std::pin::Pin; use std::sync::{Arc, Once}; use std::time::Duration; +use regex::Regex; use tokio::fs::File; use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncReadExt, AsyncWriteExt}; use tokio::time::sleep; @@ -40,11 +38,11 @@ fn log_proxy_diagnostics(config: &Config) { .url .contains('@') || proxy_cfg.url.contains("://") - && proxy_cfg - .url - .split("://") - .nth(1) - .is_some_and(|part| part.contains('@')); + && proxy_cfg + .url + .split("://") + .nth(1) + .is_some_and(|part| part.contains('@')); let has_explicit_credentials = proxy_cfg.username.as_ref().is_some() || proxy_cfg.password.as_ref().is_some(); debug!( @@ -128,6 +126,28 @@ pub fn content_type_from_ext(ext: &str) -> &'static str { } } +fn resolve_provider_url_for_attempt( + url: &Url, + provider: Option<&Arc>, +) -> Url { + let Some(provider) = provider else { + return url.clone(); + }; + + match resolve_provider_scheme_url_with_provider(url.as_str(), Some(provider.clone())) { + Ok((_provider, resolved)) => { + if resolved.as_ref() == url.as_str() { + return url.clone(); + } + Url::parse(resolved.as_ref()).unwrap_or_else(|_| url.clone()) + } + Err(err) => { + debug!("Failed to resolve provider URL: {err}"); + url.clone() + } + } +} + #[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss, clippy::cast_precision_loss)] pub fn calculate_retry_backoff(base_delay_ms: u64, multiplier: f64, attempt: u32) -> u64 { @@ -145,18 +165,28 @@ pub fn calculate_retry_backoff(base_delay_ms: u64, multiplier: f64, attempt: u32 } } -pub async fn send_with_retry( +/// Sends a request with retry logic and optional provider failover support. +#[allow(clippy::too_many_lines)] +pub async fn send_with_retry_and_provider( app_config: &Arc, - url: &Url, - mut send: impl FnMut() -> reqwest::RequestBuilder, + url: &Url, // Used primarily for logging/context + provider: Option<&Arc>, + allow_redirects: bool, + mut send: impl FnMut(&Url) -> reqwest::RequestBuilder, ) -> Result { let config = app_config.config.load(); - let (max_attempts, backoff_ms, backoff_multiplier) = config + let (max_attempts, backoff_ms, backoff_multiplier, failover_patterns) = config .reverse_proxy .as_ref() .map_or_else( - ResourceRetryConfig::get_default_retry_values, - |rp| rp.resource_retry.get_retry_values(), + || { + let (a, b, c) = ResourceRetryConfig::get_default_retry_values(); + (a, b, c, ResourceRetryConfig::default().failover_redirect_patterns) + }, + |rp| { + let (a, b, c) = rp.resource_retry.get_retry_values(); + (a, b, c, rp.resource_retry.failover_redirect_patterns.clone()) + }, ); drop(config); @@ -164,95 +194,146 @@ pub async fn send_with_retry( let idle = sleep(idle_timeout); tokio::pin!(idle); - for attempt in 0..max_attempts { - loop { + // Record the starting URL index for full-cycle detection. + // This allows us to try all URLs even when starting from a non-zero index. + let start_index = provider.as_ref().map_or(0, |p| p.get_current_index()); + let max_provider_attempts = provider.as_ref().map_or(0, |p| p.urls.len()); + let mut provider_attempts = usize::from(max_provider_attempts > 0); + + 'provider_loop: loop { + + // 2. Retry loop for the current URL + for attempt in 0..max_attempts { + let resolved_url = resolve_provider_url_for_attempt(url, provider); + // Reset the idle timer for a new attempt + idle.as_mut().reset(tokio::time::Instant::now() + idle_timeout); + tokio::select! { () = &mut idle => { warn!("Request idle for too long: {}", sanitize_sensitive_info(url.as_str())); - break; + // 1. Try Provider Failover first + if max_provider_attempts > 1 && provider_attempts < max_provider_attempts { + if let Some(p) = provider { + if p.rotate_to_next_url_with_cycle_check(start_index).is_some() { + provider_attempts += 1; + let current_index = p.get_current_index(); + warn!("Provider '{}' idle timeout -> switching to index {}", p.name, current_index); + continue 'provider_loop; + } + } + } + + // 2. If no provider or rotation failed, check if we can retry the same URL + if attempt < max_attempts - 1 { + let delay = calculate_retry_backoff(backoff_ms, backoff_multiplier, attempt); + warn!("Idle timeout, retrying same URL in {}ms (attempt {})", delay, attempt + 1); + tokio::time::sleep(Duration::from_millis(delay)).await; + continue; // This will restart the 'for attempt' loop + } + + return Err(string_to_io_error(format!("Request timed out and no retries left: {}", sanitize_sensitive_info(url.as_str())))); } - result = send().send() => { - idle.as_mut().reset(tokio::time::Instant::now() + idle_timeout); - + result = send(&resolved_url).send() => { match result { Ok(response) => { let status = response.status(); - - if status.is_success() { + if allow_redirects && status.is_redirection() { + return Ok(response); + } + let is_failover = is_failover_redirect(response.url(), &failover_patterns); + if !is_failover && status.is_success() { return Ok(response); } - let should_retry = status.is_server_error() - || matches!( - status, - StatusCode::REQUEST_TIMEOUT - | StatusCode::TOO_EARLY - | StatusCode::TOO_MANY_REQUESTS - ); + // Failover check: Should we switch to the next provider URL? + if (is_failover || should_trigger_failover(status)) + && max_provider_attempts > 1 + && provider_attempts < max_provider_attempts + { + if let Some(p) = provider { + if p.rotate_to_next_url_with_cycle_check(start_index).is_some() { + provider_attempts += 1; + let current_index = p.get_current_index(); + warn!("Provider '{}' failover: status {} -> switching to URL index {current_index}", + p.name, format_http_status(status)); + continue 'provider_loop; + } + } + } - if attempt < max_attempts - 1 && should_retry { - let wait_dur = response - .headers() - .get(RETRY_AFTER) - .and_then(|h| h.to_str().ok()) - .and_then(|s| s.parse::().ok()) - .map_or_else( - || { - let delay = calculate_retry_backoff( - backoff_ms, - backoff_multiplier, - attempt, - ); - tokio::time::Duration::from_millis(delay) - }, - tokio::time::Duration::from_secs, - ); + // Standard retry check for the same URL + let is_retryable = status.is_server_error() + || matches!(status, StatusCode::TOO_MANY_REQUESTS | StatusCode::REQUEST_TIMEOUT); - tokio::time::sleep(wait_dur).await; + if attempt < max_attempts - 1 && is_retryable { + perform_backoff(attempt, backoff_ms, backoff_multiplier, &response).await; continue; } - return Err(string_to_io_error(format!( - "Request failed with status {} {}", - format_http_status(status), - sanitize_sensitive_info(url.as_str()) - ))); + return Err(string_to_io_error(format!("Request failed ({}): {}", + format_http_status(status), sanitize_sensitive_info(url.as_str())))); } Err(err) => { + // Connection errors (Timeout/Connect) trigger failover if provider exists + if (err.is_timeout() || err.is_connect()) + && max_provider_attempts > 1 + && provider_attempts < max_provider_attempts + { + if let Some(p) = provider { + if p.rotate_to_next_url_with_cycle_check(start_index).is_some() { + provider_attempts += 1; + let current_index = p.get_current_index(); + warn!("Provider '{}' failover: connection error -> switching to index {}", p.name, current_index); + continue 'provider_loop; + } + } + } + + // If not a provider or rotation failed, try standard retry if (err.is_timeout() || err.is_connect()) && attempt < max_attempts - 1 { - let delay = calculate_retry_backoff( - backoff_ms, - backoff_multiplier, - attempt, - ); - tokio::time::sleep(tokio::time::Duration::from_millis(delay)).await; + let delay = calculate_retry_backoff(backoff_ms, backoff_multiplier, attempt); + tokio::time::sleep(Duration::from_millis(delay)).await; continue; } - error!( - "Received failure from server {}: {}", - sanitize_sensitive_info(url.as_str()), - sanitize_sensitive_info(err.to_string().as_str()) - ); - - return Err(string_to_io_error(format!( - "Request failed: {} {}", - sanitize_sensitive_info(url.as_str()), - sanitize_sensitive_info(err.to_string().as_str()) - ))); + return Err(string_to_io_error(format!("Request error: {}", sanitize_sensitive_info(err.to_string().as_str())))); } } } } } + + // 2. If per-URL retries are exhausted, try next provider URL as a last resort + if max_provider_attempts > 1 && provider_attempts < max_provider_attempts { + if let Some(p) = provider { + if p.rotate_to_next_url_with_cycle_check(start_index).is_some() { + provider_attempts += 1; + continue 'provider_loop; + } + } + } + + break; } - Err(string_to_io_error(format!( - "Failed to download file from {} after all retry attempts", - sanitize_sensitive_info(url.as_str()) - ))) + Err(string_to_io_error("All attempts and providers exhausted".to_string())) +} + +fn is_failover_redirect(url: &Url, patterns: &[Arc]) -> bool { + let redirect_url = url.as_str(); + patterns.iter().any(|pattern| pattern.is_match(redirect_url)) +} + +/// Helper to handle sleep duration for retries, respecting Retry-After headers +async fn perform_backoff(attempt: u32, ms: u64, mult: f64, response: &reqwest::Response) { + let wait_dur = response.headers() + .get(reqwest::header::RETRY_AFTER) + .and_then(|h| h.to_str().ok()) + .and_then(|s| s.parse::().ok()).map_or_else(|| Duration::from_millis(calculate_retry_backoff(ms, mult, attempt)), Duration::from_secs); + + tokio::time::sleep(wait_dur).await; } pub async fn get_input_epg_content_as_file( @@ -655,15 +736,19 @@ pub async fn get_remote_content_as_file( let default_user_agent = config.default_user_agent.clone(); drop(config); - let response = send_with_retry( + let provider_config = input.get_resolve_provider(url.as_str()); + + let response = send_with_retry_and_provider( app_config, url, - || { + provider_config.as_ref(), + false, + |resolved_url| { get_client_request( client, input.method, Some(&input.headers), - url, + resolved_url, custom_headers.as_ref(), None, default_user_agent.as_deref(), @@ -807,15 +892,17 @@ pub async fn get_remote_content_as_stream( }) .collect(); - let response = send_with_retry( + let response = send_with_retry_and_provider( app_config, url, - || { + input.get_provider(), + false, + |resolved_url| { get_client_request( client, input.method, Some(&headers), - url, + resolved_url, None, None, default_user_agent.as_deref(), @@ -1206,6 +1293,33 @@ pub fn is_uri(url: &str) -> bool { .is_ok_and(|u| u.scheme().eq_ignore_ascii_case("file") || u.scheme().eq_ignore_ascii_case("http") || u.scheme().eq_ignore_ascii_case("https")) } +/// Checks if a status code or error indicates a need for failover +/// +/// Returns true for server-side errors that might be resolved by trying another URL. +/// Returns false for client-side errors (401, 403, etc.) where the problem is with +/// credentials or permissions, not the server availability. +pub fn should_trigger_failover(status: StatusCode) -> bool { + matches!( + status, + StatusCode::NOT_FOUND + | StatusCode::GONE + | StatusCode::SERVICE_UNAVAILABLE + | StatusCode::BAD_GATEWAY + | StatusCode::GATEWAY_TIMEOUT + | StatusCode::INTERNAL_SERVER_ERROR + | StatusCode::TOO_MANY_REQUESTS + | StatusCode::REQUEST_TIMEOUT + ) + // Explicitly NOT triggering failover for: + // - 401 Unauthorized (wrong credentials) + // - 403 Forbidden (permission issue) + // - 402 Payment Required (subscription issue) + // - 407 Proxy Authentication Required (proxy credentials issue) + // - 451 Unavailable For Legal Reasons (geo-blocking) + // - 429 To many requests block + // - 408 Request takes too long +} + #[cfg(test)] mod tests { use shared::utils::{get_base_url_from_str, replace_url_extension, sanitize_sensitive_info}; diff --git a/backend/src/utils/network/xtream.rs b/backend/src/utils/network/xtream.rs index f5ffe3a5b..11540f927 100644 --- a/backend/src/utils/network/xtream.rs +++ b/backend/src/utils/network/xtream.rs @@ -11,14 +11,14 @@ use crate::repository::{ensure_input_storage_path, get_input_storage_path, get_t use crate::repository::{get_live_cat_collection_path, get_series_cat_collection_path, get_vod_cat_collection_path, xtream_get_file_path, CategoryEntry}; use crate::repository::{get_target_id_mapping, rewrite_provider_series_info_episode_virtual_id, ProviderEpisodeKey}; use crate::repository::{persist_input_vod_info, persists_input_series_info, write_playlist_batch_item_upsert, write_playlist_item_update}; -use crate::utils::{request}; +use crate::utils::request; use chrono::{DateTime, Utc}; use log::{debug, error, info, warn}; use shared::error::TuliproxError; use shared::model::{PlaylistEntry, PlaylistGroup, ProxyUserStatus, SeriesStreamProperties, StreamProperties, VideoStreamProperties, XtreamCluster, XtreamPlaylistItem, XtreamSeriesInfo, XtreamVideoInfo, XtreamVideoInfoDoc}; -use shared::utils::{extract_extension_from_url, get_i64_from_serde_value, get_string_from_serde_value, sanitize_sensitive_info, Internable}; +use shared::utils::{extract_extension_from_url, get_i64_from_serde_value, get_string_from_serde_value, sanitize_sensitive_info, Internable, PROVIDER_SCHEME_PREFIX}; use std::collections::HashMap; use std::io::Error; use std::path::Path; @@ -89,7 +89,8 @@ pub async fn get_xtream_stream_info(client: &reqwest::Client, return serde_json::to_string(&content).map_err(|err| info_err!("{err}")); } - let input_source = InputSource::from(input).with_url(info_url.to_owned()); + let resolved_url = input.resolve_url(info_url)?; + let input_source = InputSource::from(input).with_url(resolved_url.to_string()); if let Ok(content) = get_xtream_stream_info_content(app_config, client, &input_source, false).await { if content.is_empty() { return Err(info_err!("Provider returned no response for stream with id: {}/{}/{}", @@ -351,7 +352,23 @@ pub async fn download_xtream_playlist(app_config: &Arc, client: &reqw let username = input_source.username.as_ref().map_or("", |v| v); let password = input_source.password.as_ref().map_or("", |v| v); - let base_url = get_xtream_stream_url_base(&input_source.url, username, password); + let is_provider_url = input_source.url.starts_with(PROVIDER_SCHEME_PREFIX); + let base_input_url = if is_provider_url { + // Keep provider:// unresolved; send_with_retry_and_provider resolves per attempt + // so failover can switch provider hosts. + input_source.url.clone() + } else { + match input.resolve_url(&input_source.url) { + Ok(url) => url.into_owned(), + Err(err) => return (Vec::with_capacity(0), vec![err], false), + } + }; + + let base_url = if is_provider_url { + base_input_url.clone() + } else { + get_xtream_stream_url_base(&base_input_url, username, password) + }; let input_source_login = input_source.with_url(base_url.clone()); check_alias_user_state(app_config, client, input).await; diff --git a/frontend/public/assets/i18n/en.json b/frontend/public/assets/i18n/en.json index 7cbc23f73..47a1daf31 100644 --- a/frontend/public/assets/i18n/en.json +++ b/frontend/public/assets/i18n/en.json @@ -418,7 +418,10 @@ "MSG_KIND_WATCH": "Watch", "API_PROXY_CONFIG": "Api Proxy Config", "USE_USER_DB": "Use User DB", - "ADD_SERVER": "Add Server" + "ADD_SERVER": "Add Server", + "FAILOVER_REDIRECT_PATTERNS": "Failover Redirect Patterns", + "ADD_PATTERN": "Add Pattern" + }, "TITLE": { "USER_BOUQUET_EDITOR": "Playlist Category Selection" diff --git a/frontend/src/app/components/config/reverse_proxy_config_view.rs b/frontend/src/app/components/config/reverse_proxy_config_view.rs index 321df2e73..70df96f1a 100644 --- a/frontend/src/app/components/config/reverse_proxy_config_view.rs +++ b/frontend/src/app/components/config/reverse_proxy_config_view.rs @@ -8,9 +8,10 @@ use crate::app::context::ConfigContext; use crate::app::components::config::config_view_context::ConfigViewContext; use crate::app::components::config::config_page::{ConfigForm, LABEL_REVERSE_PROXY_CONFIG}; use crate::app::components::{Card}; -use crate::{config_field, config_field_bool, config_field_custom, config_field_hide, config_field_optional, +use crate::{config_field, config_field_bool, config_field_child, config_field_custom, config_field_hide, config_field_optional, edit_field_bool, edit_field_list, edit_field_number, edit_field_number_f64, edit_field_number_u64, edit_field_number_usize, edit_field_text, edit_field_text_option, generate_form_reducer}; +use crate::app::components::Chip; const LABEL_CACHE: &str = "LABEL.CACHE"; const LABEL_ENABLED: &str = "LABEL.ENABLED"; @@ -40,6 +41,8 @@ const LABEL_RESOURCE_RETRY: &str = "LABEL.RESOURCE_RETRY"; const LABEL_MAX_ATTEMPTS: &str = "LABEL.MAX_ATTEMPTS"; const LABEL_BACKOFF_MILLIS: &str = "LABEL.BACKOFF_MILLIS"; const LABEL_BACKOFF_MULTIPLIER: &str = "LABEL.BACKOFF_MULTIPLIER"; +const LABEL_FAILOVER_REDIRECT_PATTERNS: &str = "LABEL.FAILOVER_REDIRECT_PATTERNS"; +const LABEL_ADD_PATTERN: &str = "LABEL.ADD_PATTERN"; const LABEL_DISABLED_HEADER: &str = "LABEL.DISABLED_HEADER"; const LABEL_REFERER_HEADER: &str = "LABEL.REFERER_HEADER"; const LABEL_X_HEADER: &str = "LABEL.X_HEADER"; @@ -79,6 +82,26 @@ generate_form_reducer!( } ); +/// Simple wrapper for failover patterns to use Vec directly with edit_field_list +#[derive(Debug, Clone, Default, PartialEq)] +pub struct FailoverPatternsDto { + pub patterns: Vec, +} + +impl FailoverPatternsDto { + pub fn is_empty(&self) -> bool { + self.patterns.is_empty() + } +} + +generate_form_reducer!( + state: FailoverPatternsFormState { form: FailoverPatternsDto }, + action_name: FailoverPatternsFormAction, + fields { + Patterns => patterns: Vec, + } +); + generate_form_reducer!( state: StreamConfigFormState { form: StreamConfigDto }, action_name: StreamConfigFormAction, @@ -164,6 +187,10 @@ pub fn ReverseProxyConfigView() -> Html { StreamBufferConfigFormState { form: StreamBufferConfigDto::default(), modified: false } }); + let failover_patterns_state: UseReducerHandle = use_reducer(|| { + FailoverPatternsFormState { form: FailoverPatternsDto::default(), modified: false } + }); + { let on_form_change = config_view_ctx.on_form_change.clone(); let reverse_proxy_state = reverse_proxy_state.clone(); @@ -174,6 +201,7 @@ pub fn ReverseProxyConfigView() -> Html { let stream_state = stream_state.clone(); let geoip_state = geoip_state.clone(); let stream_buffer_state = stream_buffer_state.clone(); + let failover_patterns_state = failover_patterns_state.clone(); use_effect_with( ( @@ -185,8 +213,9 @@ pub fn ReverseProxyConfigView() -> Html { stream_state, geoip_state, stream_buffer_state, + failover_patterns_state, ), - move |(rp, disabled_header, cache, rl, resource_retry, stream, geoip, stream_buffer)| { + move |(rp, disabled_header, cache, rl, resource_retry, stream, geoip, stream_buffer, failover_patterns)| { let mut form = rp.form.clone(); let mut stream_form = stream.form.clone(); stream_form.buffer = if stream_buffer.form.is_empty() { @@ -197,7 +226,13 @@ pub fn ReverseProxyConfigView() -> Html { form.cache = Some(cache.form.clone()); form.rate_limit = Some(rl.form.clone()); - form.resource_retry = Some(resource_retry.form.clone()); + let mut resource_retry_form = resource_retry.form.clone(); + resource_retry_form.failover_redirect_patterns = if failover_patterns.form.is_empty() { + None + } else { + Some(failover_patterns.form.patterns.clone()) + }; + form.resource_retry = Some(resource_retry_form); form.stream = Some(stream_form); form.geoip = Some(geoip.form.clone()); form.disabled_header = if disabled_header.form.is_empty() { @@ -213,7 +248,8 @@ pub fn ReverseProxyConfigView() -> Html { || resource_retry.modified || stream.modified || geoip.modified - || stream_buffer.modified; + || stream_buffer.modified + || failover_patterns.modified; on_form_change.emit(ConfigForm::ReverseProxy(modified, form)); }, ); @@ -228,6 +264,7 @@ pub fn ReverseProxyConfigView() -> Html { let stream_state = stream_state.clone(); let geoip_state = geoip_state.clone(); let stream_buffer_state = stream_buffer_state.clone(); + let failover_patterns_state = failover_patterns_state.clone(); let reverse_proxy_cfg = config_ctx.config.as_ref().and_then(|c| c.config.reverse_proxy.clone()); use_effect_with((reverse_proxy_cfg, config_view_ctx.edit_mode.clone()), move |(cfg, _mode)| { @@ -240,6 +277,9 @@ pub fn ReverseProxyConfigView() -> Html { stream_state.dispatch(StreamConfigFormAction::SetAll(rp.stream.as_ref().map_or_else(StreamConfigDto::default, |s| s.clone()))); geoip_state.dispatch(GeoIpConfigFormAction::SetAll(rp.geoip.as_ref().map_or_else(GeoIpConfigDto::default, |s| s.clone()))); stream_buffer_state.dispatch(StreamBufferConfigFormAction::SetAll(rp.stream.as_ref().and_then(|s| s.buffer.clone()).unwrap_or_default())); + failover_patterns_state.dispatch(FailoverPatternsFormAction::SetAll(FailoverPatternsDto { + patterns: rp.resource_retry.as_ref().and_then(|rr| rr.failover_redirect_patterns.clone()).unwrap_or_default() + })); } else { reverse_proxy_state.dispatch(ReverseProxyConfigFormAction::SetAll(ReverseProxyConfigDto::default())); disabled_header_state.dispatch(ReverseProxyDisabledHeaderConfigFormAction::SetAll(ReverseProxyDisabledHeaderConfigDto::default())); @@ -249,6 +289,7 @@ pub fn ReverseProxyConfigView() -> Html { stream_state.dispatch(StreamConfigFormAction::SetAll(StreamConfigDto::default())); geoip_state.dispatch(GeoIpConfigFormAction::SetAll(GeoIpConfigDto::default())); stream_buffer_state.dispatch(StreamBufferConfigFormAction::SetAll(StreamBufferConfigDto::default())); + failover_patterns_state.dispatch(FailoverPatternsFormAction::SetAll(FailoverPatternsDto::default())); } || () }); @@ -330,6 +371,7 @@ pub fn ReverseProxyConfigView() -> Html { }; let render_resource_retry_view = || { + let patterns = &failover_patterns_state.form.patterns; html! {

{translate.t(LABEL_RESOURCE_RETRY)}

@@ -341,6 +383,17 @@ pub fn ReverseProxyConfigView() -> Html { format_float_localized(resource_retry_state.form.backoff_multiplier, 4, true) ) } + { config_field_child!(translate.t(LABEL_FAILOVER_REDIRECT_PATTERNS), { + html! { +
+ if patterns.is_empty() { + + } else { + { for patterns.iter().map(|p| html! { }) } + } +
+ } + })} } }; @@ -352,6 +405,7 @@ pub fn ReverseProxyConfigView() -> Html { { edit_field_number!(resource_retry_state, translate.t(LABEL_MAX_ATTEMPTS), max_attempts, ResourceRetryConfigFormAction::MaxAttempts) } { edit_field_number_u64!(resource_retry_state, translate.t(LABEL_BACKOFF_MILLIS), backoff_millis, ResourceRetryConfigFormAction::BackoffMillis) } { edit_field_number_f64!(resource_retry_state, translate.t(LABEL_BACKOFF_MULTIPLIER), backoff_multiplier, ResourceRetryConfigFormAction::BackoffMultiplier) } + { edit_field_list!(failover_patterns_state, translate.t(LABEL_FAILOVER_REDIRECT_PATTERNS), patterns, FailoverPatternsFormAction::Patterns, translate.t(LABEL_ADD_PATTERN)) } } }; diff --git a/shared/src/error/tuliprox_error.rs b/shared/src/error/tuliprox_error.rs index 3ab7af9b1..69f0e2d5e 100644 --- a/shared/src/error/tuliprox_error.rs +++ b/shared/src/error/tuliprox_error.rs @@ -14,8 +14,15 @@ macro_rules! get_errors_notify_message { .map(|err| err.message.as_str()) .collect::>() .join("\r\n"); - if $size > 0 && text.len() > std::cmp::max($size - 3, 3) { - Some(format!("{}...", text.get(0..$size).unwrap())) + + let max_size = $size; + // Check if we actually need to truncate + if max_size > 0 && text.chars().count() > max_size { + // Safely take 'max_size' characters and add ellipsis + let truncated: String = text.chars().take(max_size).collect(); + Some(format!("{}...", truncated)) + } else if text.is_empty() { + None } else { Some(text) } diff --git a/shared/src/model/config/input.rs b/shared/src/model/config/input.rs index 0ce00b53b..179c34e21 100644 --- a/shared/src/model/config/input.rs +++ b/shared/src/model/config/input.rs @@ -1,10 +1,10 @@ use super::PanelApiConfigDto; use crate::error::{TuliproxError, TuliproxErrorKind}; use crate::model::EpgConfigDto; -use crate::utils::{arc_str_serde, default_as_true, deserialize_timestamp, get_credentials_from_url_str, get_trimmed_string, - is_false, is_true, is_zero_u16, sanitize_sensitive_info, - serialize_option_vec_flow_map_items, trim_last_slash}; -use crate::utils::{is_blank_optional_string, Internable}; +use crate::utils::{arc_str_serde, default_as_true, deserialize_timestamp, get_credentials_from_url_str, + get_trimmed_string, is_false, is_true, is_zero_u16, parse_provider_scheme_url_parts, + sanitize_sensitive_info, serialize_option_vec_flow_map_items, trim_last_slash, PROVIDER_SCHEME_PREFIX}; +use crate::utils::{is_blank_optional_string, Internable, arc_str_vec_serde}; use crate::{check_input_connections, check_input_credentials, info_err_res}; use enum_iterator::Sequence; @@ -361,7 +361,7 @@ impl Default for ConfigInputDto { impl ConfigInputDto { #[allow(clippy::cast_possible_truncation)] - pub fn prepare(&mut self, index: u16, _include_computed: bool) -> Result { + pub fn prepare(&mut self, index: u16, _include_computed: bool, provider_names: &HashSet) -> Result { self.name = self.name.trim().intern(); if self.name.is_empty() { return info_err_res!("name for input is mandatory"); @@ -384,12 +384,27 @@ impl ConfigInputDto { self.persist = get_trimmed_string(self.persist.as_deref()); + if self.url.starts_with(PROVIDER_SCHEME_PREFIX) { + if let Ok((host, _path)) = parse_provider_scheme_url_parts(&self.url) { + if !provider_names.contains(host) { + return info_err_res!("Provider name {host} is not defined"); + } + } + } + let mut current_index = index + 1; self.id = current_index; if let Some(aliases) = self.aliases.as_mut() { let input_type = &self.input_type; for alias in aliases { current_index = alias.prepare(current_index, input_type)?; + if alias.url.starts_with(PROVIDER_SCHEME_PREFIX) { + if let Ok((host, _path)) = parse_provider_scheme_url_parts(&alias.url) { + if !provider_names.contains(host) { + return info_err_res!("Provider name {host} is not defined"); + } + } + } } } @@ -530,3 +545,26 @@ impl ConfigInputDto { Err(TuliproxError::new(TuliproxErrorKind::Info, format!("No matching input or alias found for input '{input_name}' with username '{username}'"))) } } + +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct ConfigProviderDto { + #[serde(with = "arc_str_serde")] + pub name: Arc, + #[serde(with = "arc_str_vec_serde")] + pub urls: Vec>, +} + +impl ConfigProviderDto { + pub fn prepare(&mut self) -> Result<(), TuliproxError> { + self.name = self.name.trim().intern(); + if self.name.is_empty() { + return info_err_res!("Name for provider is mandatory"); + } + self.urls = self.urls.drain(..).filter(|url| !url.trim().is_empty()).map(|u| u.trim().intern()).collect(); + if self.urls.is_empty() { + return info_err_res!("Urls for provider is mandatory"); + } + Ok(()) + } +} \ No newline at end of file diff --git a/shared/src/model/config/reverse_proxy.rs b/shared/src/model/config/reverse_proxy.rs index 71910d7b4..47687700b 100644 --- a/shared/src/model/config/reverse_proxy.rs +++ b/shared/src/model/config/reverse_proxy.rs @@ -1,6 +1,6 @@ use crate::error::{TuliproxError, TuliproxErrorKind}; use crate::model::{CacheConfigDto, GeoIpConfigDto, RateLimitConfigDto, StreamConfigDto}; -use crate::utils::{is_false, default_resource_retry_attempts, +use crate::utils::{is_false, is_empty_optional_vec, default_resource_retry_attempts, default_resource_retry_backoff_ms, default_resource_retry_backoff_multiplier, is_default_resource_retry_attempts, @@ -8,6 +8,7 @@ use crate::utils::{is_false, default_resource_retry_attempts, is_default_resource_retry_backoff_multiplier, hex_to_u8_16}; use log::warn; +use crate::info_err_res; #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, Default, PartialEq)] #[serde(deny_unknown_fields)] @@ -117,6 +118,11 @@ impl ReverseProxyConfigDto { rate_limit.prepare()?; } } + + if let Some(resource_retry) = self.resource_retry.as_mut() { + resource_retry.prepare()?; + } + Ok(()) } } @@ -130,6 +136,8 @@ pub struct ResourceRetryConfigDto { pub backoff_millis: u64, #[serde(default = "default_resource_retry_backoff_multiplier", skip_serializing_if = "is_default_resource_retry_backoff_multiplier")] pub backoff_multiplier: f64, + #[serde(default, skip_serializing_if = "is_empty_optional_vec")] + pub failover_redirect_patterns: Option>, } impl Default for ResourceRetryConfigDto { @@ -138,6 +146,7 @@ impl Default for ResourceRetryConfigDto { max_attempts: default_resource_retry_attempts(), backoff_millis: default_resource_retry_backoff_ms(), backoff_multiplier: default_resource_retry_backoff_multiplier(), + failover_redirect_patterns: None, } } } @@ -147,5 +156,17 @@ impl ResourceRetryConfigDto { self.max_attempts == default_resource_retry_attempts() && self.backoff_millis == default_resource_retry_backoff_ms() && (self.backoff_multiplier - default_resource_retry_backoff_multiplier()).abs() < f64::EPSILON + && is_empty_optional_vec(&self.failover_redirect_patterns) + } + + pub fn prepare(&mut self) -> Result<(), TuliproxError> { + if let Some(failover_redirect_patterns) = self.failover_redirect_patterns.as_mut() { + for pattern in failover_redirect_patterns { + if let Err(err) = crate::model::REGEX_CACHE.get_or_compile(pattern) { + return info_err_res!("Can't parse regex: {pattern} {err}"); + } + } + } + Ok(()) } } diff --git a/shared/src/model/config/source.rs b/shared/src/model/config/source.rs index 9e8decd05..8d59c1170 100644 --- a/shared/src/model/config/source.rs +++ b/shared/src/model/config/source.rs @@ -3,7 +3,7 @@ use std::sync::Arc; use crate::info_err_res; use crate::error::{TuliproxError}; use crate::foundation::prepare_templates; -use crate::model::{ConfigInputDto, HdHomeRunDeviceOverview, PatternTemplate}; +use crate::model::{ConfigInputDto, ConfigProviderDto, HdHomeRunDeviceOverview, PatternTemplate}; use crate::model::config::target::ConfigTargetDto; use crate::utils::{arc_str_vec_serde, default_as_default, Internable}; @@ -35,6 +35,8 @@ impl ConfigSourceDto { pub struct SourcesConfigDto { #[serde(default, skip_serializing_if = "Option::is_none")] pub templates: Option>, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub provider: Option>, pub inputs: Vec, pub sources: Vec, } @@ -42,19 +44,34 @@ pub struct SourcesConfigDto { impl SourcesConfigDto { pub fn prepare(&mut self, include_computed: bool, hdhr_config: Option<&HdHomeRunDeviceOverview>) -> Result<(), TuliproxError> { self.prepare_templates()?; - self.prepare_sources(include_computed, hdhr_config)?; + let provider_names =self.prepare_providers()?; + self.prepare_sources(include_computed, hdhr_config, &provider_names)?; self.check_unique_target_names()?; Ok(()) } - fn prepare_sources(&mut self, include_computed: bool, hdhr_config: Option<&HdHomeRunDeviceOverview>) -> Result<(), TuliproxError> { + fn prepare_providers(&mut self) -> Result, TuliproxError> { + let mut names = HashSet::new(); + if let Some(providers) = &mut self.provider { + for provider in providers { + provider.prepare()?; + if names.contains(provider.name.as_ref()) { + return info_err_res!("Provider names should be unique: {}", provider.name); + } + names.insert(provider.name.to_string()); + } + } + Ok(names) + } + + fn prepare_sources(&mut self, include_computed: bool, hdhr_config: Option<&HdHomeRunDeviceOverview>, provider_names: &HashSet) -> Result<(), TuliproxError> { // prepare sources and set id's let mut source_index: u16 = 0; let mut input_index: u16 = 0; let mut target_index: u16 = 1; // Prepare global inputs for input in &mut self.inputs { - input_index = input.prepare(input_index, include_computed)?; + input_index = input.prepare(input_index, include_computed, provider_names)?; } for source in &mut self.sources { diff --git a/shared/src/utils/request.rs b/shared/src/utils/request.rs index fdd7f80d7..ab853a68d 100644 --- a/shared/src/utils/request.rs +++ b/shared/src/utils/request.rs @@ -1,9 +1,11 @@ use std::borrow::Cow; use std::sync::atomic::Ordering; use url::Url; -use crate::concat_string; +use crate::{concat_string, info_err, info_err_res}; +use crate::error::TuliproxError; use crate::utils::{CONSTANTS, DASH_EXT, DASH_EXT_FRAGMENT, DASH_EXT_QUERY, HLS_EXT, HLS_EXT_FRAGMENT, HLS_EXT_QUERY}; +pub const PROVIDER_SCHEME_PREFIX: &str = "provider://"; pub const CONTENT_TYPE_JSON: &str = "application/json"; pub const CONTENT_TYPE_CBOR: &str = "application/cbor"; @@ -148,4 +150,23 @@ pub fn concat_path_leading_slash(first: &str, second: &str) -> String { } let path = path.trim_start_matches('/'); format!("/{path}") +} + +/// Internal helper to parse the provider URL into (host, path_and_query) +pub fn parse_provider_scheme_url_parts(stream_url: &str) -> Result<(&str, &str), TuliproxError> { + + let rest = stream_url.strip_prefix(PROVIDER_SCHEME_PREFIX).ok_or_else(|| { + info_err!("Not a provider URL: '{}'", sanitize_sensitive_info(stream_url)) + })?; + + let (host, path) = match rest.find('/') { + Some(idx) => (&rest[..idx], &rest[idx..]), + None => (rest, ""), + }; + + if host.is_empty() { + return info_err_res!("Provider host is empty in URL: '{}'", sanitize_sensitive_info(stream_url)); + } + + Ok((host, path)) } \ No newline at end of file