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

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<ConfigInput>,
     item_type: PlaylistItemType,
     share_stream: bool,
     connection_permission: UserConnectionPermission,
     force_provider: Option<&Arc<str>>,
     virtual_id: VirtualId,
-) -> StreamDetails {
+) -> Result<StreamDetails, TuliproxError> {
+
     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<ConfigInput>,
     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<ConfigInput>,
+    target: &Arc<ConfigTarget>,
     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(),

backend/src/api/endpoints/v1_api.rs

@@ -85,6 +85,7 @@ async fn geoip_update(axum::extract::State(app_state): axum::extract::State<Arc<
             let input_source =  InputSource {
                 name: "GeoIP".intern(),
                 url,
+                provider: None,
                 username: None,
                 password: None,
                 method: InputFetchMethod::GET,

backend/src/api/model/provider_lineup_manager.rs

@@ -949,6 +949,7 @@ mod tests {
             exp_date: None,
             t_batch_url: None,
             panel_api: None,
+            provider_configs: None,
             cache_duration_seconds: 0,
         }
     }

backend/src/api/model/streams/active_client_stream.rs

@@ -481,14 +481,16 @@ fn stream_grace_period(
                             let stop_signal = provisioning_info.stop_signal;
                             let addr = fingerprint.addr;
                             tokio::spawn(async move {
-                                run_panel_api_provisioning_probe(
+                                if let Err(err) = run_panel_api_provisioning_probe(
                                     app_state,
                                     input,
                                     stop_signal,
                                     addr,
                                     virtual_id,
                                 )
-                                    .await;
+                                    .await {
+                                    error!("Error running Probe: {err:?}");
+                                }
                             });
                         } else {
                             stream_strategy_flag_copy

backend/src/api/model/streams/provider_stream_factory.rs

@@ -3,10 +3,10 @@ use crate::api::model::{get_response_headers, AppState, CustomVideoStreamType};
 use crate::api::model::StreamError;
 use crate::api::model::{create_channel_unavailable_stream, get_header_filter_for_item_type};
 use crate::api::model::{BoxedProviderStream, ProviderStreamFactoryResponse};
-use crate::model::{ReverseProxyDisabledHeaderConfig};
+use crate::model::{ConfigProvider, ReverseProxyDisabledHeaderConfig};
 use crate::tools::atomic_once_flag::AtomicOnceFlag;
 use crate::utils::debug_if_enabled;
-use crate::utils::request::{classify_content_type, get_request_headers, MimeCategory};
+use crate::utils::request::{classify_content_type, get_request_headers, MimeCategory, send_with_retry_and_provider};
 use futures::stream::{self};
 use futures::{StreamExt, TryStreamExt};
 use log::{debug, log_enabled, warn};
@@ -42,6 +42,7 @@ pub struct ProviderStreamFactoryOptions {
     range_bytes: Arc<Option<AtomicUsize>>,
     range_requested: bool,
     reconnect_flag: Arc<AtomicOnceFlag>,
+    provider: Option<Arc<ConfigProvider>>,
 }
 
 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<Arc<ConfigProvider>>) {
+        self.provider = provider;
+    }
+
+    pub fn get_provider(&self) -> Option<&Arc<ConfigProvider>> {
+        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<AppState>) -> bool {
 async fn send_with_manual_redirects(
     request_client: &reqwest::Client,
     stream_options: &ProviderStreamFactoryOptions,
-) -> Result<reqwest::Response, reqwest::Error> {
+    app_state: &Arc<AppState>,
+) -> Result<reqwest::Response, std::io::Error> {
     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(&current_url));
-        let response = client.send().await?;
+        let result = send_with_retry_and_provider(
+            &app_state.app_config,
+            &current_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<AppState>,
     request_client: &reqwest::Client,
@@ -399,10 +429,22 @@ async fn provider_stream_request(
 ) -> Result<Option<ProviderStreamFactoryResponse>, 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) => {

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<AppState>,
     input: &ConfigInput,
-) -> Option<UserApiAccountInfo> {
-    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<Option<UserApiAccountInfo>, 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::<Tz>().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<AtomicOnceFlag>,
     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 {