diff --git a/backend/src/api/api_utils.rs b/backend/src/api/api_utils.rs index 9880a7dd5..5916d0ebd 100644 --- a/backend/src/api/api_utils.rs +++ b/backend/src/api/api_utils.rs @@ -120,7 +120,7 @@ pub async fn serve_file(file_path: &Path, mime_type: mime::Mime) -> impl axum::r pub fn get_user_target_by_username(username: &str, app_state: &AppState) -> Option<(ProxyUserCredentials, Arc)> { if !username.is_empty() { - return app_state.config.get_target_for_username(username); + return app_state.app_config.get_target_for_username(username); } None } @@ -128,13 +128,13 @@ pub fn get_user_target_by_username(username: &str, app_state: &AppState) -> Opti pub fn get_user_target_by_credentials<'a>(username: &str, password: &str, api_req: &'a UserApiRequest, app_state: &'a AppState) -> Option<(ProxyUserCredentials, Arc)> { if !username.is_empty() && !password.is_empty() { - app_state.config.get_target_for_user(username, password) + app_state.app_config.get_target_for_user(username, password) } else { let token = api_req.token.as_str().trim(); if token.is_empty() { None } else { - app_state.config.get_target_for_user_by_token(token) + app_state.app_config.get_target_for_user_by_token(token) } } } @@ -174,7 +174,7 @@ pub struct StreamOptions { /// Returns a `StreamOptions` instance with the resolved configuration. fn get_stream_options(app_state: &AppState) -> StreamOptions { let (stream_retry, stream_force_retry_secs, buffer_enabled, buffer_size) = app_state - .config.config.load() + .app_config.config.load() .reverse_proxy .as_ref() .and_then(|reverse_proxy| reverse_proxy.stream.as_ref()) @@ -298,7 +298,7 @@ async fn resolve_streaming_strategy(app_state: &AppState, stream_url: &str, inpu let stream_response_params = match &*provider_connection_guard { ProviderAllocation::Exhausted => { debug!("Input {} is exhausted. No connections allowed.", input.name); - let stream = create_provider_connections_exhausted_stream(&app_state.config, &[]); + let stream = create_provider_connections_exhausted_stream(&app_state.app_config, &[]); ProviderStreamState::Custom(stream) } ProviderAllocation::Available(ref provider) @@ -345,7 +345,7 @@ async fn create_stream_response_details(app_state: &AppState, force_provider: Option<&str>) -> StreamDetails { let mut streaming_strategy = resolve_streaming_strategy(app_state, stream_url, input, force_provider).await; - let config_grace_period_millis = app_state.config.config.load().reverse_proxy.as_ref() + let config_grace_period_millis = app_state.app_config.config.load().reverse_proxy.as_ref() .and_then(|r| r.stream.as_ref()).map_or_else(default_grace_period_millis, |s| s.grace_period_millis); let grace_period_millis = get_grace_period_millis(connection_permission, &streaming_strategy.provider_stream_state, config_grace_period_millis); match streaming_strategy.provider_stream_state { @@ -367,7 +367,7 @@ async fn create_stream_response_details(app_state: &AppState, let ((stream, stream_info), reconnect_flag) = if let Ok(url) = parsed_url { let provider_stream_factory_options = ProviderStreamFactoryOptions::new(item_type, share_stream, stream_options, &url, req_headers, streaming_strategy.input_headers.as_ref()); let reconnect_flag = provider_stream_factory_options.get_reconnect_flag_clone(); - let provider_stream = match create_provider_stream(Arc::clone(&app_state.config), Arc::clone(&app_state.http_client), provider_stream_factory_options).await { + let provider_stream = match create_provider_stream(Arc::clone(&app_state.app_config), Arc::clone(&app_state.http_client), provider_stream_factory_options).await { None => (None, None), Some((stream, info)) => { (Some(stream), info) @@ -555,7 +555,7 @@ pub async fn force_provider_stream_response(app_state: &AppState, } drop(stream_details.provider_connection_guard.take()); if let (Some(stream), _stream_info) = - create_channel_unavailable_stream(&app_state.config, &[], StatusCode::BAD_GATEWAY) + create_channel_unavailable_stream(&app_state.app_config, &[], StatusCode::BAD_GATEWAY) { debug!("Streaming custom stream"); axum::response::Response::builder().status(StatusCode::OK).body(Body::from_stream(stream)).unwrap().into_response() @@ -579,7 +579,7 @@ pub async fn stream_response(app_state: &AppState, if log_enabled!(log::Level::Trace) { trace!("Try to open stream {}", sanitize_sensitive_info(stream_url)); } if connection_permission == UserConnectionPermission::Exhausted { - return create_custom_video_stream_response(&app_state.config, CustomVideoStreamType::UserConnectionsExhausted).into_response(); + return create_custom_video_stream_response(&app_state.app_config, CustomVideoStreamType::UserConnectionsExhausted).into_response(); } let share_stream = is_stream_share_enabled(item_type, target); @@ -646,7 +646,7 @@ pub async fn stream_response(app_state: &AppState, } fn get_stream_throttle(app_state: &AppState) -> u64 { - app_state.config.config.load() + app_state.app_config.config.load() .reverse_proxy .as_ref() .and_then(|reverse_proxy| reverse_proxy.stream.as_ref()) @@ -775,7 +775,7 @@ pub fn get_username_from_auth_header( token: &str, app_state: &Arc, ) -> Option { - if let Some(web_auth_config) = &app_state.config.config.load().web_ui.as_ref().and_then(|c| c.auth.as_ref()) { + if let Some(web_auth_config) = &app_state.app_config.config.load().web_ui.as_ref().and_then(|c| c.auth.as_ref()) { let secret_key: &str = web_auth_config.secret.as_ref(); if let Ok(token_data) = decode::( token, diff --git a/backend/src/api/config_watch.rs b/backend/src/api/config_watch.rs index fce37decd..56d7c799f 100644 --- a/backend/src/api/config_watch.rs +++ b/backend/src/api/config_watch.rs @@ -1,13 +1,14 @@ use crate::api::model::app_state::AppState; -use shared::error::{TuliproxError, TuliproxErrorKind}; use crate::utils; -use crate::utils::is_directory; +use crate::utils::{is_directory, read_config_file, read_sources_file}; use log::{debug, error, info}; use notify::event::{AccessKind, AccessMode}; use notify::{recommended_watcher, Event, EventKind, RecursiveMode, Watcher}; +use shared::error::{TuliproxError, TuliproxErrorKind}; use std::collections::HashMap; use std::path::{Path, PathBuf}; use std::sync::{mpsc, Arc}; +use crate::model::{Config, SourcesConfig}; enum ConfigFile { Config, @@ -18,17 +19,20 @@ enum ConfigFile { impl ConfigFile { fn load_mappping(app_state: &Arc) -> Result<(), TuliproxError> { - match utils::read_mappings(app_state.config.t_mapping_file_path.as_str(), true) { - Ok(Some(mappings_cfg)) => { - app_state.config.set_mappings(&mappings_cfg); - info!("Loaded mapping file {}", app_state.config.t_mapping_file_path.as_str()); - } - Ok(None) => { - info!("No mapping file loaded {}", app_state.config.t_mapping_file_path.as_str()); - } - Err(err) => { - error!("Failed to load mapping file {err}"); - return Err(err); + let paths = app_state.app_config.paths.load(); + if let Some(mapping_file_path) = paths.mapping_file_path.as_ref() { + match utils::read_mappings(mapping_file_path, true) { + Ok(Some(mappings_cfg)) => { + app_state.app_config.set_mappings(&mappings_cfg); + info!("Loaded mapping file {mapping_file_path}"); + } + Ok(None) => { + info!("No mapping file loaded {mapping_file_path}"); + } + Err(err) => { + error!("Failed to load mapping file {err}"); + return Err(err); + } } } @@ -36,9 +40,15 @@ impl ConfigFile { } fn load_api_proxy(app_state: &Arc) -> Result<(), TuliproxError> { - match utils::read_api_proxy_config(&app_state.config) { - Ok(()) => { - info!("Api Proxy File: {:?}", &app_state.config.t_api_proxy_file_path); + match utils::read_api_proxy_config(&app_state.app_config, true) { + Ok(Some(api_proxy)) => { + app_state.app_config.set_api_proxy(api_proxy)?; + let paths = app_state.app_config.paths.load(); + info!("Loaded Api Proxy File: {:?}", &paths.api_proxy_file_path); + } + Ok(None) => { + let paths = app_state.app_config.paths.load(); + info!("Coul dnot load Api Proxy File: {:?}", &paths.api_proxy_file_path); } Err(err) => { error!("Failed to load api-proxy file {err}"); @@ -47,12 +57,35 @@ impl ConfigFile { } Ok(()) } + + fn load_config(app_state: &Arc) -> Result<(), TuliproxError> { + let paths = app_state.app_config.paths.load(); + let config_file = paths.config_file_path.as_str(); + let config_dto = read_config_file(config_file, true)?; + let mut config: Config = Config::from(config_dto); + config.prepare(paths.config_path.as_str())?; + info!("Loaded config file {config_file}"); + app_state.set_config(config)?; + Ok(()) + } + + fn load_sources(app_state: &Arc) -> Result<(), TuliproxError> { + let paths = app_state.app_config.paths.load(); + let sources_file = paths.sources_file_path.as_str(); + let sources_dto = read_sources_file(sources_file, true, true)?; + let sources: SourcesConfig = sources_dto.into(); + info!("Loaded sources file {sources_file}"); + app_state.app_config.set_sources(sources)?; + Ok(()) + } + pub(crate) fn reload(&self, file_path: &Path, app_state: &Arc) -> Result<(), TuliproxError> { debug!("File change detected {}", file_path.display()); match self { ConfigFile::ApiProxy => ConfigFile::load_api_proxy(app_state), ConfigFile::Mapping => ConfigFile::load_mappping(app_state), - ConfigFile::Config | ConfigFile::Sources => { Ok(()) } + ConfigFile::Config => ConfigFile::load_config(app_state), + ConfigFile::Sources => ConfigFile::load_sources(app_state), } } } @@ -60,11 +93,13 @@ impl ConfigFile { pub async fn exec_config_watch(app_state: &Arc) -> Result<(), TuliproxError> { let (tx, rx) = mpsc::channel::>(); + let paths = &app_state.app_config.paths.load(); + let mapping_file_path = paths.mapping_file_path.as_ref().map_or_else(String::new, ToString::to_string); let mut files = HashMap::new(); - [(&app_state.config.t_config_file_path, ConfigFile::Config), - (&app_state.config.t_api_proxy_file_path, ConfigFile::ApiProxy), - (&app_state.config.t_mapping_file_path, ConfigFile::Mapping), - (&app_state.config.t_sources_file_path, ConfigFile::Sources) + [(&paths.config_file_path, ConfigFile::Config), + (&paths.api_proxy_file_path, ConfigFile::ApiProxy), + (&mapping_file_path, ConfigFile::Mapping), + (&paths.sources_file_path, ConfigFile::Sources) ].into_iter() .filter(|(path, _)| !path.is_empty()) .for_each(|(path, config_file)| { files.insert(PathBuf::from(path), (config_file, is_directory(path))); }); @@ -77,8 +112,8 @@ pub async fn exec_config_watch(app_state: &Arc) -> Result<(), Tuliprox // Add a path to be watched. All files and directories at that path and // below will be monitored for changes. - let path = Path::new(app_state.config.t_config_path.as_str()); - let recursive_mode = if utils::is_directory(&app_state.config.t_mapping_file_path) { RecursiveMode::Recursive } else { RecursiveMode::NonRecursive }; + let path = Path::new(paths.config_path.as_str()); + let recursive_mode = if !mapping_file_path.is_empty() && utils::is_directory(&mapping_file_path) { RecursiveMode::Recursive } else { RecursiveMode::NonRecursive }; watcher.watch(path, recursive_mode).map_err(|err| TuliproxError::new(TuliproxErrorKind::Info, format!("Failed to start config file watcher {err}")))?; info!("Watching config file changes {}", path.display()); diff --git a/backend/src/api/endpoints/download_api.rs b/backend/src/api/endpoints/download_api.rs index 5b4ae5372..8f452109a 100644 --- a/backend/src/api/endpoints/download_api.rs +++ b/backend/src/api/endpoints/download_api.rs @@ -118,7 +118,7 @@ pub async fn queue_download_file( axum::extract::State(app_state): axum::extract::State>, axum::extract::Json(req): axum::extract::Json, ) -> impl axum::response::IntoResponse + Send { - let app_config = &*app_state.config; + let app_config = &*app_state.app_config; let config = > as Access>::load(&app_config.config); if let Some(download_cfg) = config.video.as_ref().unwrap().download.as_ref() { @@ -129,7 +129,7 @@ pub async fn queue_download_file( Some(file_download) => { app_state.downloads.queue.lock().await.push_back(file_download.clone()); if app_state.downloads.active.read().await.is_none() { - match run_download_queue(&app_state.config, download_cfg, &app_state.downloads).await { + match run_download_queue(&app_state.app_config, download_cfg, &app_state.downloads).await { Ok(()) => {} Err(err) => return (axum::http::StatusCode::INTERNAL_SERVER_ERROR, axum::Json(json!({"error": err}))).into_response(), } diff --git a/backend/src/api/endpoints/hdhomerun_api.rs b/backend/src/api/endpoints/hdhomerun_api.rs index b90a0be97..defdaab79 100644 --- a/backend/src/api/endpoints/hdhomerun_api.rs +++ b/backend/src/api/endpoints/hdhomerun_api.rs @@ -157,8 +157,8 @@ where } fn create_device(app_state: &Arc) -> Option { - if let Some(credentials) = app_state.app_state.config.get_user_credentials(&app_state.device.t_username) { - let server_info = app_state.app_state.config.get_user_server_info(&credentials); + if let Some(credentials) = app_state.app_state.app_config.get_user_credentials(&app_state.device.t_username) { + let server_info = app_state.app_state.app_config.get_user_server_info(&credentials); let device = &app_state.device; let device_url = format!("{}://{}:{}", server_info.protocol, server_info.host, device.port); Some(Device { @@ -237,7 +237,7 @@ async fn lineup(app_state: &Arc, cfg: &Arc, creden .body(axum::body::Body::from_stream(body_stream)) .unwrap().into_response(); } else if (use_all || use_xtream) && target.has_output(&TargetType::Xtream) { - let server_info = app_state.app_state.config.get_user_server_info(credentials); + let server_info = app_state.app_state.app_config.get_user_server_info(credentials); let base_url = server_info.get_base_url(); let base_url_live = if credentials.proxy.is_redirect(PlaylistItemType::Live) || target.is_force_redirect(PlaylistItemType::Live) { None } else { Some(base_url.clone()) }; @@ -273,7 +273,7 @@ async fn lineup(app_state: &Arc, cfg: &Arc, creden } async fn auth_lineup_json(AuthBasic((username, password)): AuthBasic, axum::extract::State(app_state): axum::extract::State>) -> impl IntoResponse { - let cfg = Arc::clone(&app_state.app_state.config); + let cfg = Arc::clone(&app_state.app_state.app_config); if let Some((credentials, target)) = cfg.get_target_for_username(&app_state.device.t_username) { if !username.eq(&credentials.username) || !password.eq(&credentials.password) { return axum::http::StatusCode::UNAUTHORIZED.into_response(); @@ -285,7 +285,7 @@ async fn auth_lineup_json(AuthBasic((username, password)): AuthBasic, axum::extr } async fn lineup_json(axum::extract::State(app_state): axum::extract::State>) -> impl IntoResponse { - let cfg = Arc::clone(&app_state.app_state.config); + let cfg = Arc::clone(&app_state.app_state.app_config); if let Some((credentials, target)) = cfg.get_target_for_username(&app_state.device.t_username) { let user_credentials = Arc::new(credentials); return lineup(&app_state, &cfg, &user_credentials, &target).await.into_response(); diff --git a/backend/src/api/endpoints/hls_api.rs b/backend/src/api/endpoints/hls_api.rs index c3028c6c1..8a6ddba4f 100644 --- a/backend/src/api/endpoints/hls_api.rs +++ b/backend/src/api/endpoints/hls_api.rs @@ -44,7 +44,7 @@ pub(in crate::api) async fn handle_hls_stream_request( input: &ConfigInput, connection_permission: UserConnectionPermission) -> impl IntoResponse + Send { let url = replace_url_extension(hls_url, HLS_EXT); - let server_info = app_state.config.get_user_server_info(user); + let server_info = app_state.app_config.get_user_server_info(user); let (request_url, session_token) = match user_session { Some(session) => { @@ -72,7 +72,7 @@ pub(in crate::api) async fn handle_hls_stream_request( match request::download_text_content(Arc::clone(&app_state.http_client), input, &request_url, None).await { Ok((content, response_url)) => { let rewrite_hls_props = RewriteHlsProps { - secret: &app_state.config.t_encrypt_secret, + secret: &app_state.app_config.encrypt_secret, base_url: &server_info.get_base_url(), content: &content, hls_url: response_url, @@ -85,7 +85,7 @@ pub(in crate::api) async fn handle_hls_stream_request( } Err(err) => { error!("Failed to download m3u8 {}", sanitize_sensitive_info(err.to_string().as_str())); - create_custom_video_stream_response(&app_state.config, CustomVideoStreamType::ChannelUnavailable).into_response() + create_custom_video_stream_response(&app_state.app_config, CustomVideoStreamType::ChannelUnavailable).into_response() } } } @@ -97,29 +97,29 @@ async fn hls_api_stream( axum::extract::State(app_state): axum::extract::State>, ) -> impl axum::response::IntoResponse + Send { let (user, target) = try_option_bad_request!( - app_state.config.get_target_for_user(¶ms.username, ¶ms.password), false, + app_state.app_config.get_target_for_user(¶ms.username, ¶ms.password), false, format!("Could not find any user {}", params.username)); if user.permission_denied(&app_state) { - return create_custom_video_stream_response(&app_state.config, CustomVideoStreamType::UserAccountExpired).into_response(); + return create_custom_video_stream_response(&app_state.app_config, CustomVideoStreamType::UserAccountExpired).into_response(); } let target_name = &target.name; let virtual_id = params.stream_id; - let input = try_option_bad_request!(app_state.config.get_input_by_id(params.input_id), true, format!("Cant find input for target {target_name}, context {}, stream_id {virtual_id}", XtreamCluster::Live)); + let input = try_option_bad_request!(app_state.app_config.get_input_by_id(params.input_id), true, format!("Cant find input for target {target_name}, context {}, stream_id {virtual_id}", XtreamCluster::Live)); let user_session_token = format!("{fingerprint}{virtual_id}"); let mut user_session = app_state.active_users.get_user_session(&user.username, &user_session_token).await; if let Some(session) = &mut user_session { if session.permission == UserConnectionPermission::Exhausted { - return create_custom_video_stream_response(&app_state.config, CustomVideoStreamType::UserConnectionsExhausted).into_response(); + return create_custom_video_stream_response(&app_state.app_config, CustomVideoStreamType::UserConnectionsExhausted).into_response(); } if app_state.active_provider.is_over_limit(&session.provider).await { - return create_custom_video_stream_response(&app_state.config, CustomVideoStreamType::ProviderConnectionsExhausted).into_response(); + return create_custom_video_stream_response(&app_state.app_config, CustomVideoStreamType::ProviderConnectionsExhausted).into_response(); } - let hls_url = match get_hls_session_token_and_url_from_token(&app_state.config.t_encrypt_secret, ¶ms.token) { + let hls_url = match get_hls_session_token_and_url_from_token(&app_state.app_config.encrypt_secret, ¶ms.token) { Some((Some(session_token), hls_url)) if session.token.eq(&session_token) => hls_url, _ => return axum::http::StatusCode::BAD_REQUEST.into_response(), }; @@ -136,7 +136,7 @@ async fn hls_api_stream( let connection_permission = user.connection_permission(&app_state).await; if connection_permission == UserConnectionPermission::Exhausted { - return create_custom_video_stream_response(&app_state.config, CustomVideoStreamType::UserConnectionsExhausted).into_response(); + return create_custom_video_stream_response(&app_state.app_config, CustomVideoStreamType::UserConnectionsExhausted).into_response(); } if is_hls_url(&session.stream_url) { diff --git a/backend/src/api/endpoints/m3u_api.rs b/backend/src/api/endpoints/m3u_api.rs index c40a2920a..0a6bb6bbc 100644 --- a/backend/src/api/endpoints/m3u_api.rs +++ b/backend/src/api/endpoints/m3u_api.rs @@ -22,7 +22,7 @@ async fn m3u_api( ) -> impl axum::response::IntoResponse + Send { match get_user_target(api_req, app_state) { Some((user, target)) => { - match m3u_load_rewrite_playlist(&app_state.config, &target, &user).await { + match m3u_load_rewrite_playlist(&app_state.app_config, &target, &user).await { Ok(m3u_iter) => { // Convert the iterator into a stream of `Bytes` let content_stream = stream::iter(m3u_iter.map(|line| Ok::(Bytes::from([line.to_string().as_bytes(), b"\n"].concat())))); @@ -69,7 +69,7 @@ async fn m3u_api_stream( ) -> impl axum::response::IntoResponse + Send { let (user, target) = try_option_bad_request!(get_user_target_by_credentials(stream_req.username, stream_req.password, api_req, app_state), false, format!("Could not find any user {}", stream_req.username)); if user.permission_denied(app_state) { - return create_custom_video_stream_response(&app_state.config, CustomVideoStreamType::UserAccountExpired).into_response(); + return create_custom_video_stream_response(&app_state.app_config, CustomVideoStreamType::UserAccountExpired).into_response(); } let target_name = &target.name; @@ -80,8 +80,8 @@ async fn m3u_api_stream( let (action_stream_id, stream_ext) = separate_number_and_remainder(stream_req.stream_id); let virtual_id: u32 = try_result_bad_request!(action_stream_id.trim().parse()); - let pli = try_result_bad_request!(m3u_get_item_for_stream_id(virtual_id, &app_state.config, &target).await, true, format!("Failed to read m3u item for stream id {}", virtual_id)); - let input = try_option_bad_request!(app_state.config.get_input_by_name(pli.input_name.as_str()), true, format!("Cant find input for target {target_name}, stream_id {virtual_id}")); + let pli = try_result_bad_request!(m3u_get_item_for_stream_id(virtual_id, &app_state.app_config, &target).await, true, format!("Failed to read m3u item for stream id {}", virtual_id)); + let input = try_option_bad_request!(app_state.app_config.get_input_by_name(pli.input_name.as_str()), true, format!("Cant find input for target {target_name}, stream_id {virtual_id}")); let cluster = XtreamCluster::try_from(pli.item_type).unwrap_or(XtreamCluster::Live); @@ -90,11 +90,11 @@ async fn m3u_api_stream( let session_url = if let Some(session) = &user_session { if session.permission == UserConnectionPermission::Exhausted { - return create_custom_video_stream_response(&app_state.config, CustomVideoStreamType::UserConnectionsExhausted).into_response(); + return create_custom_video_stream_response(&app_state.app_config, CustomVideoStreamType::UserConnectionsExhausted).into_response(); } if app_state.active_provider.is_over_limit(&session.provider).await { - return create_custom_video_stream_response(&app_state.config, CustomVideoStreamType::ProviderConnectionsExhausted).into_response(); + return create_custom_video_stream_response(&app_state.app_config, CustomVideoStreamType::ProviderConnectionsExhausted).into_response(); } if session.virtual_id == virtual_id && is_seek_request(cluster, req_headers).await { // partial request means we are in reverse proxy mode, seek happened @@ -107,7 +107,7 @@ async fn m3u_api_stream( let connection_permission = user.connection_permission(app_state).await; if connection_permission == UserConnectionPermission::Exhausted { - return create_custom_video_stream_response(&app_state.config, CustomVideoStreamType::UserConnectionsExhausted).into_response(); + return create_custom_video_stream_response(&app_state.app_config, CustomVideoStreamType::UserConnectionsExhausted).into_response(); } let context = ApiStreamContext::try_from(cluster).unwrap_or(ApiStreamContext::Live); @@ -159,7 +159,7 @@ async fn m3u_api_resource( debug!("Target has no m3u playlist {target_name}"); return StatusCode::BAD_REQUEST.into_response(); } - let m3u_item = match m3u_get_item_for_stream_id(m3u_stream_id, &app_state.config, &target).await { + let m3u_item = match m3u_get_item_for_stream_id(m3u_stream_id, &app_state.app_config, &target).await { Ok(item) => item, Err(err) => { error!("Failed to get m3u url: {}", sanitize_sensitive_info(err.to_string().as_str())); diff --git a/backend/src/api/endpoints/user_api.rs b/backend/src/api/endpoints/user_api.rs index e561b84b6..405851b8a 100644 --- a/backend/src/api/endpoints/user_api.rs +++ b/backend/src/api/endpoints/user_api.rs @@ -52,7 +52,7 @@ async fn playlist_categories( } let target_name = &target.name; let xtream_stream = if target.has_output(&TargetType::Xtream) { - let config = &app_state.config.config.load(); + let config = &app_state.app_config.config.load(); let live_categories = get_categories_from_xtream(xtream_get_playlist_categories(config, target_name, XtreamCluster::Live).await); let vod_categories = get_categories_from_xtream(xtream_get_playlist_categories(config, target_name, XtreamCluster::Video).await); let series_categories = get_categories_from_xtream(xtream_get_playlist_categories(config, target_name, XtreamCluster::Series).await); @@ -70,7 +70,7 @@ async fn playlist_categories( }; let m3u_stream = if target.has_output(&TargetType::M3u) { - let live_categories = get_categories_from_m3u_playlist(&target, &app_state.config).await; + let live_categories = get_categories_from_m3u_playlist(&target, &app_state.app_config).await; stream::iter(vec![ Ok::(Bytes::from(r#"{"live": "#)), Ok::(Bytes::from(serde_json::to_string(&live_categories).unwrap_or("[]".to_string()))), @@ -108,7 +108,7 @@ async fn save_playlist_bouquet( if user.permission_denied(&app_state) { return axum::http::StatusCode::FORBIDDEN.into_response(); } - let config = &app_state.config.config.load(); + let config = &app_state.app_config.config.load(); match save_user_bouquet(config, &target.name, &username, &bouquet).await { Ok(()) => { return axum::http::StatusCode::OK.into_response(); @@ -131,7 +131,7 @@ async fn playlist_bouquet( if user.permission_denied(&app_state) { return axum::http::StatusCode::FORBIDDEN.into_response(); } - let config = &app_state.config.config.load(); + let config = &app_state.app_config.config.load(); let xtream = load_user_bouquet_as_json(config, &username, TargetType::Xtream).await; let m3u = load_user_bouquet_as_json(config, &username, TargetType::M3u).await; return axum::response::Response::builder() diff --git a/backend/src/api/endpoints/v1_api.rs b/backend/src/api/endpoints/v1_api.rs index c92e6147a..5dfc37102 100644 --- a/backend/src/api/endpoints/v1_api.rs +++ b/backend/src/api/endpoints/v1_api.rs @@ -5,7 +5,7 @@ use crate::api::model::app_state::AppState; use crate::api::model::request::{PlaylistRequest, PlaylistRequestType}; use crate::auth::create_access_token; use crate::auth::validator_admin; -use crate::model::{AppConfig, ConfigSource, ConfigTarget, StatusCheck, TargetUser}; +use crate::model::{StatusCheck, TargetUser}; use crate::model::{ConfigInput, ConfigInputOptions}; use crate::processing::processor::playlist; use crate::repository::user_repository::store_api_user; @@ -15,11 +15,10 @@ use axum::response::IntoResponse; use log::error; use serde_json::json; use shared::error::TuliproxError; -use shared::model::{ApiProxyConfigDto, ApiProxyServerInfoDto, ConfigApiDto, ConfigDto, ConfigRenameDto, ConfigSortDto, InputType, TargetOutputDto, TargetUserDto, XtreamPlaylistItem}; +use shared::model::{ApiProxyConfigDto, ApiProxyServerInfoDto, ConfigDto, InputType, TargetUserDto, XtreamPlaylistItem}; use shared::utils::sanitize_sensitive_info; use std::collections::{BTreeMap, HashSet}; use std::sync::Arc; -use crate::api::model::config::{ServerConfig, ServerInputConfig, ServerSourceConfig, ServerTargetConfig}; fn intern_save_config_api_proxy(backup_dir: &str, api_proxy: &ApiProxyConfigDto, file_path: &str) -> Option { match utils::save_api_proxy(file_path, backup_dir, api_proxy) { @@ -68,20 +67,21 @@ async fn save_config_api_proxy_user( } } - if let Some(old_api_proxy) = app_state.config.t_api_proxy.load().clone() { + if let Some(old_api_proxy) = app_state.app_config.api_proxy.load().clone() { let mut api_proxy = (*old_api_proxy).clone(); api_proxy.user = users.iter().map(TargetUser::from).collect(); let new_api_proxy = Arc::new(api_proxy); - app_state.config.t_api_proxy.store(Some(Arc::clone(&new_api_proxy))); + app_state.app_config.api_proxy.store(Some(Arc::clone(&new_api_proxy))); if new_api_proxy.use_user_db { - if let Err(err) = store_api_user(&app_state.config, &new_api_proxy.user) { + if let Err(err) = store_api_user(&app_state.app_config, &new_api_proxy.user) { return (axum::http::StatusCode::INTERNAL_SERVER_ERROR, axum::Json(json!({"error": err.to_string()}))).into_response(); } } else { - let config = app_state.config.config.load(); + let config = app_state.app_config.config.load(); let backup_dir = config.backup_dir.as_ref().unwrap().as_str(); - if let Some(err) = intern_save_config_api_proxy(backup_dir, &ApiProxyConfigDto::from(&*new_api_proxy), app_state.config.t_api_proxy_file_path.as_str()) { + let paths = app_state.app_config.paths.load(); + if let Some(err) = intern_save_config_api_proxy(backup_dir, &ApiProxyConfigDto::from(&*new_api_proxy), paths.api_proxy_file_path.as_str()) { return (axum::http::StatusCode::INTERNAL_SERVER_ERROR, axum::Json(json!({"error": err.to_string()}))).into_response(); } } @@ -94,8 +94,9 @@ async fn save_config_main( axum::extract::Json(cfg): axum::extract::Json, ) -> impl axum::response::IntoResponse + Send { if cfg.is_valid() { - let file_path = app_state.config.t_config_file_path.as_str(); - let config = app_state.config.config.load(); + let paths = app_state.app_config.paths.load(); + let file_path = paths.config_file_path.as_str(); + let config = app_state.app_config.config.load(); let backup_dir = config.backup_dir.as_ref().unwrap().as_str(); if let Some(err) = intern_save_config_main(file_path, backup_dir, &cfg) { return (axum::http::StatusCode::INTERNAL_SERVER_ERROR, axum::Json(json!({"error": err.to_string()}))).into_response(); @@ -117,14 +118,15 @@ async fn save_config_api_proxy_config( } // TODO wenn hot reload an ist wird doppelt geladen - if let Some(old_api_proxy) = app_state.config.t_api_proxy.load().clone() { + if let Some(old_api_proxy) = app_state.app_config.api_proxy.load().clone() { let mut api_proxy = (*old_api_proxy).clone(); api_proxy.server = req_api_proxy.iter().map(Into::into).collect(); let new_api_proxy = Arc::new(api_proxy); - app_state.config.t_api_proxy.store(Some(Arc::clone(&new_api_proxy))); - let config = app_state.config.config.load(); + app_state.app_config.api_proxy.store(Some(Arc::clone(&new_api_proxy))); + let config = app_state.app_config.config.load(); let backup_dir = config.backup_dir.as_ref().unwrap().as_str(); - if let Some(err) = intern_save_config_api_proxy(backup_dir, &ApiProxyConfigDto::from(new_api_proxy.as_ref()), app_state.config.t_api_proxy_file_path.as_str()) { + let paths = app_state.app_config.paths.load(); + if let Some(err) = intern_save_config_api_proxy(backup_dir, &ApiProxyConfigDto::from(new_api_proxy.as_ref()), paths.api_proxy_file_path.as_str()) { return (axum::http::StatusCode::INTERNAL_SERVER_ERROR, axum::Json(json!({"error": err.to_string()}))).into_response(); } } @@ -136,10 +138,10 @@ async fn playlist_update( axum::extract::Json(targets): axum::extract::Json>, ) -> impl axum::response::IntoResponse + Send { let user_targets = if targets.is_empty() { None } else { Some(targets) }; - let process_targets = app_state.config.sources.load().validate_targets(user_targets.as_ref()); + let process_targets = app_state.app_config.sources.load().validate_targets(user_targets.as_ref()); match process_targets { Ok(valid_targets) => { - tokio::spawn(playlist::exec_processing(Arc::clone(&app_state.http_client), Arc::clone(&app_state.config), Arc::new(valid_targets))); + tokio::spawn(playlist::exec_processing(Arc::clone(&app_state.http_client), Arc::clone(&app_state.app_config), Arc::new(valid_targets))); axum::http::StatusCode::OK.into_response() } Err(err) => { @@ -192,18 +194,18 @@ async fn playlist_content( axum::extract::State(app_state): axum::extract::State>, axum::extract::Json(playlist_req): axum::extract::Json, ) -> impl IntoResponse + Send { - let config = app_state.config.config.load(); + let config = app_state.app_config.config.load(); match playlist_req.rtype { PlaylistRequestType::Input => { if let Some(source_id) = playlist_req.source_id { - get_playlist(Arc::clone(&app_state.http_client), app_state.config.get_input_by_id(source_id).as_deref(), &config).await.into_response() + get_playlist(Arc::clone(&app_state.http_client), app_state.app_config.get_input_by_id(source_id).as_deref(), &config).await.into_response() } else { (axum::http::StatusCode::BAD_REQUEST, axum::Json(json!({"error": "Invalid input"}))).into_response() } } PlaylistRequestType::Target => { if let Some(source_id) = playlist_req.source_id { - get_playlist_for_target(app_state.config.get_target_by_id(source_id).as_deref(), &app_state.config).await.into_response() + get_playlist_for_target(app_state.app_config.get_target_by_id(source_id).as_deref(), &app_state.app_config).await.into_response() } else { (axum::http::StatusCode::BAD_REQUEST, axum::Json(json!({"error": "Invalid target"}))).into_response() } @@ -232,10 +234,10 @@ async fn playlist_webplayer( axum::extract::State(app_state): axum::extract::State>, axum::extract::Json(playlist_item): axum::extract::Json, ) -> impl axum::response::IntoResponse + Send { - let access_token = create_access_token(&app_state.config.t_access_token_secret, 5); - let config = app_state.config.config.load(); + let access_token = create_access_token(&app_state.app_config.access_token_secret, 5); + let config = app_state.app_config.config.load(); let server_name = config.web_ui.as_ref().and_then(|web_ui| web_ui.player_server.as_ref()).map_or("default", |server_name| server_name.as_str()); - let server_info = app_state.config.get_server_info(server_name); + let server_info = app_state.app_config.get_server_info(server_name); let base_url = server_info.get_base_url(); format!("{base_url}/token/{access_token}/{target_id}/{}/{}", playlist_item.xtream_cluster.as_stream_type(), playlist_item.virtual_id).into_response() } @@ -243,82 +245,15 @@ async fn playlist_webplayer( async fn config( axum::extract::State(app_state): axum::extract::State>, ) -> impl axum::response::IntoResponse + Send { - let map_input = |i: &Arc| ServerInputConfig { - id: i.id, - name: i.name.clone(), - input_type: i.input_type, - url: i.url.clone(), - username: i.username.clone(), - password: i.password.clone(), - persist: i.persist.clone(), - enabled: i.enabled, - }; - - let map_target = |t: &Arc| { - let mapping = t.mapping.load(); - ServerTargetConfig { - id: t.id, - enabled: t.enabled, - name: t.name.clone(), - options: t.options.clone(), - sort: t.sort.as_ref().map(ConfigSortDto::from), - filter: t.filter.to_string(), - output: t.output.iter().map(TargetOutputDto::from).collect(), - rename: t.rename.as_ref().map(|l| l.iter().map(ConfigRenameDto::from).collect()), - mapping: mapping.as_ref().map(|m| m.iter().map(|m| m.id.to_string()).collect()), - processing_order: t.processing_order, - watch: t.watch.as_ref().map(|w| w.iter().map(std::string::ToString::to_string).collect()), + let paths = app_state.app_config.paths.load(); + match utils::read_app_config_dto(&paths, + true, false) { + Ok(app_config) => axum::response::Json(app_config).into_response(), + Err(err) => { + error!("Failed to read config files: {err}"); + axum::http::StatusCode::INTERNAL_SERVER_ERROR.into_response() } - }; - - let map_source = |s: &ConfigSource| ServerSourceConfig { - inputs: s.inputs.iter().map(map_input).collect(), - targets: s.targets.iter().map(map_target).collect(), - }; - - let map_config = |app_config: &AppConfig| { - let sources = app_config.sources.load(); - let config = app_config.config.load(); - ServerConfig { - api: ConfigApiDto::from(&config.api), - threads: config.threads, - working_dir: config.working_dir.clone(), - backup_dir: config.backup_dir.clone(), - user_config_dir: config.user_config_dir.clone(), - log: config.log.as_ref().map(Into::into), - update_on_boot: config.update_on_boot, - web_ui: config.web_ui.as_ref().map(Into::into), - schedules: config.schedules.as_ref().map(|l| l.iter().map(Into::into).collect()), - reverse_proxy: config.reverse_proxy.as_ref().map(Into::into), - messaging: config.messaging.as_ref().map(Into::into), - video: config.video.as_ref().map(Into::into), - sources: sources.sources.iter().map(map_source).collect(), - proxy: config.proxy.as_ref().map(Into::into), - ipcheck: config.ipcheck.as_ref().map(Into::into), - api_proxy: utils::read_api_proxy_dto(&app_state.config, false), - } - }; - - let mut result = match utils::read_config(app_state.config.t_config_path.as_str(), - app_state.config.t_config_file_path.as_str(), - app_state.config.t_sources_file_path.as_str(), - app_state.config.t_api_proxy_file_path.as_str(), - Some(app_state.config.t_mapping_file_path.to_string()), - false) { - Ok(mut cfg) => { - let _ = cfg.prepare(false); - map_config(&cfg) - } - Err(_) => map_config(&app_state.config) - }; - - // if we didn't read it from file then we should use it from app_state - if result.api_proxy.is_none() { - let value = app_state.config.t_api_proxy.load(); - result.api_proxy = value.as_ref().map(|a| ApiProxyConfigDto::from(a.as_ref())); } - - axum::response::Json(result).into_response() } @@ -331,7 +266,7 @@ struct IpCheck { } async fn create_ipinfo_check(app_state: &Arc) -> Option<(Option, Option)> { - let config = app_state.config.config.load(); + let config = app_state.app_config.config.load(); if let Some(ipcheck) = config.ipcheck.as_ref() { if let Ok(check) = get_ips(&app_state.http_client, ipcheck).await { return Some(check); @@ -404,7 +339,7 @@ pub fn v1_api_register(web_auth_enabled: bool, app_state: Arc, web_ui_ .route("/playlist", axum::routing::post(playlist_content)) .route("/file/download", axum::routing::post(download_api::queue_download_file)) .route("/file/download/info", axum::routing::get(download_api::download_file_info)); - let config = app_state.config.config.load(); + let config = app_state.app_config.config.load(); if config.ipcheck.is_some() { router = router.route("/ipinfo", axum::routing::get(ipinfo)); } diff --git a/backend/src/api/endpoints/web_index.rs b/backend/src/api/endpoints/web_index.rs index d5c065ad4..0c60009bc 100644 --- a/backend/src/api/endpoints/web_index.rs +++ b/backend/src/api/endpoints/web_index.rs @@ -18,7 +18,7 @@ async fn token( axum::extract::State(app_state): axum::extract::State>, axum::extract::Json(mut req): axum::extract::Json, ) -> impl axum::response::IntoResponse + Send { - let config = &app_state.config.config.load(); + let config = &app_state.app_config.config.load(); match config.web_ui.as_ref().and_then(|c| c.auth.as_ref()) { None => no_web_auth_token().into_response(), Some(web_auth) => { @@ -37,7 +37,7 @@ async fn token( } } } - if let Some(credentials) = app_state.config.get_user_credentials(username) { + if let Some(credentials) = app_state.app_config.get_user_credentials(username) { if credentials.password == password { if let Ok(token) = create_jwt_user(web_auth, username) { req.zeroize(); @@ -57,7 +57,7 @@ async fn token_refresh( AuthBearer(token): AuthBearer, axum::extract::State(app_state): axum::extract::State>, ) -> impl axum::response::IntoResponse + Send { - let config = &app_state.config.config.load(); + let config = &app_state.app_config.config.load(); match &config.web_ui.as_ref().and_then(|c| c.auth.as_ref()) { None => no_web_auth_token().into_response(), Some(web_auth) => { @@ -85,7 +85,7 @@ async fn token_refresh( async fn index( axum::extract::State(app_state): axum::extract::State>, ) -> impl axum::response::IntoResponse + Send { - let config = &app_state.config.config.load(); + let config = &app_state.app_config.config.load(); let path: PathBuf = [&config.api.web_root, "index.html"].iter().collect(); if let Some(web_ui_path) = &config.web_ui.as_ref().and_then(|c| c.path.as_ref()) { match tokio::fs::read_to_string(&path).await { @@ -115,7 +115,7 @@ async fn index( async fn index_config( axum::extract::State(app_state): axum::extract::State>, ) -> impl axum::response::IntoResponse + Send { - let config = &app_state.config.config.load(); + let config = &app_state.app_config.config.load(); let path: PathBuf = [&config.api.web_root, "config.json"].iter().collect(); if let Some(web_ui_path) = &config.web_ui.as_ref().and_then(|c| c.path.as_ref()) { match tokio::fs::read_to_string(&path).await { diff --git a/backend/src/api/endpoints/xmltv_api.rs b/backend/src/api/endpoints/xmltv_api.rs index 1e0702921..da55c662c 100644 --- a/backend/src/api/endpoints/xmltv_api.rs +++ b/backend/src/api/endpoints/xmltv_api.rs @@ -192,7 +192,7 @@ async fn xmltv_api( return axum::http::StatusCode::FORBIDDEN.into_response(); } - let config = &app_state.config.config.load(); + let config = &app_state.app_config.config.load(); let Some(epg_path) = get_epg_path_for_target(config, &target) else { // No epg configured, No processing or timeshift, epg can't be mapped to the channels. // we do not deliver epg diff --git a/backend/src/api/endpoints/xtream_api.rs b/backend/src/api/endpoints/xtream_api.rs index 625527b73..771a34f57 100644 --- a/backend/src/api/endpoints/xtream_api.rs +++ b/backend/src/api/endpoints/xtream_api.rs @@ -166,9 +166,9 @@ pub(in crate::api) fn get_xtream_player_api_stream_url( } async fn get_user_info(user: &ProxyUserCredentials, app_state: &AppState) -> XtreamAuthorizationResponse { - let server_info = app_state.config.get_user_server_info(user); + let server_info = app_state.app_config.get_user_server_info(user); let active_connections = app_state.get_active_connections_for_user(&user.username).await; - XtreamAuthorizationResponse::new(&server_info, user, active_connections, app_state.config.config.load().user_access_control) + XtreamAuthorizationResponse::new(&server_info, user, active_connections, app_state.app_config.config.load().user_access_control) } async fn xtream_player_api_stream( @@ -180,7 +180,7 @@ async fn xtream_player_api_stream( ) -> impl IntoResponse + Send { let (user, target) = try_option_bad_request!(get_user_target_by_credentials(stream_req.username, stream_req.password, api_req, app_state), false, format!("Could not find any user {}", stream_req.username)); if user.permission_denied(app_state) { - return create_custom_video_stream_response(&app_state.config, CustomVideoStreamType::UserAccountExpired).into_response(); + return create_custom_video_stream_response(&app_state.app_config, CustomVideoStreamType::UserAccountExpired).into_response(); } let target_name = &target.name; @@ -191,8 +191,8 @@ async fn xtream_player_api_stream( let (action_stream_id, stream_ext) = separate_number_and_remainder(stream_req.stream_id); let virtual_id: u32 = try_result_bad_request!(action_stream_id.trim().parse()); - let (pli, mapping) = try_result_bad_request!(xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.config, &target, None), true, format!("Failed to read xtream item for stream id {}", virtual_id)); - let input = try_option_bad_request!(app_state.config.get_input_by_name(pli.input_name.as_str()), true, format!("Cant find input for target {target_name}, context {}, stream_id {virtual_id}", stream_req.context)); + let (pli, mapping) = try_result_bad_request!(xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.app_config, &target, None), true, format!("Failed to read xtream item for stream id {}", virtual_id)); + let input = try_option_bad_request!(app_state.app_config.get_input_by_name(pli.input_name.as_str()), true, format!("Cant find input for target {target_name}, context {}, stream_id {virtual_id}", stream_req.context)); let cluster = pli.xtream_cluster; let item_type = if stream_req.context == ApiStreamContext::Timeshift { PlaylistItemType::Catchup } else { pli.item_type }; @@ -202,11 +202,11 @@ async fn xtream_player_api_stream( let session_url = if let Some(session) = &user_session { if session.permission == UserConnectionPermission::Exhausted { - return create_custom_video_stream_response(&app_state.config, CustomVideoStreamType::UserConnectionsExhausted).into_response(); + return create_custom_video_stream_response(&app_state.app_config, CustomVideoStreamType::UserConnectionsExhausted).into_response(); } if app_state.active_provider.is_over_limit(&session.provider).await { - return create_custom_video_stream_response(&app_state.config, CustomVideoStreamType::ProviderConnectionsExhausted).into_response(); + return create_custom_video_stream_response(&app_state.app_config, CustomVideoStreamType::ProviderConnectionsExhausted).into_response(); } if session.virtual_id == virtual_id && is_seek_request(cluster, req_headers).await { @@ -221,7 +221,7 @@ async fn xtream_player_api_stream( let connection_permission = user.connection_permission(app_state).await; if connection_permission == UserConnectionPermission::Exhausted { - return create_custom_video_stream_response(&app_state.config, CustomVideoStreamType::UserConnectionsExhausted).into_response(); + return create_custom_video_stream_response(&app_state.app_config, CustomVideoStreamType::UserConnectionsExhausted).into_response(); } let context = stream_req.context; @@ -271,7 +271,7 @@ async fn xtream_player_api_stream_with_token( target_id: u16, stream_req: ApiStreamRequest<'_>, ) -> impl IntoResponse + Send { - if let Some(target) = app_state.config.get_target_by_id(target_id) { + if let Some(target) = app_state.app_config.get_target_by_id(target_id) { let target_name = &target.name; if !target.has_output(&TargetType::Xtream) { debug!("Target has no xtream output {target_name}"); @@ -279,14 +279,14 @@ async fn xtream_player_api_stream_with_token( } let (action_stream_id, stream_ext) = separate_number_and_remainder(stream_req.stream_id); let virtual_id: u32 = try_result_bad_request!(action_stream_id.trim().parse()); - let (pli, _mapping) = try_result_bad_request!(xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.config, &target, None), true, format!("Failed to read xtream item for stream id {}", virtual_id)); - let input = try_option_bad_request!(app_state.config.get_input_by_name(pli.input_name.as_str()), true, format!("Cant find input for target {target_name}, context {}, stream_id {virtual_id}", stream_req.context)); + let (pli, _mapping) = try_result_bad_request!(xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.app_config, &target, None), true, format!("Failed to read xtream item for stream id {}", virtual_id)); + let input = try_option_bad_request!(app_state.app_config.get_input_by_name(pli.input_name.as_str()), true, format!("Cant find input for target {target_name}, context {}, stream_id {virtual_id}", stream_req.context)); let session_key = format!("{fingerprint}{virtual_id}"); let is_hls_request = pli.item_type == PlaylistItemType::LiveHls || stream_ext.as_deref() == Some(HLS_EXT); - let config = app_state.config.config.load(); + let config = app_state.app_config.config.load(); let server = config.web_ui.as_ref().and_then(|web_ui| web_ui.player_server.as_ref()).map_or("default", |server_name| server_name.as_str()); let user = ProxyUserCredentials { @@ -475,11 +475,11 @@ async fn xtream_player_api_resource( } let virtual_id: u32 = try_result_bad_request!(resource_req.stream_id.trim().parse()); let resource = resource_req.action_path.trim(); - let (pli, _) = try_result_bad_request!(xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.config, &target, None), true, format!("Failed to read xtream item for stream id {}", virtual_id)); + let (pli, _) = try_result_bad_request!(xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.app_config, &target, None), true, format!("Failed to read xtream item for stream id {}", virtual_id)); let stream_url = if resource.starts_with(crate::model::XC_INFO_RESOURCE_PREFIX) { - try_result_bad_request!(xtream_get_info_resource_url(&app_state.config, &pli, &target, resource)) + try_result_bad_request!(xtream_get_info_resource_url(&app_state.app_config, &pli, &target, resource)) } else if resource.starts_with(crate::model::XC_SEASON_RESOURCE_PREFIX) { - try_result_bad_request!(xtream_get_season_resource_url(&app_state.config, &pli, &target, resource)) + try_result_bad_request!(xtream_get_season_resource_url(&app_state.app_config, &pli, &target, resource)) } else { pli.get_field(resource) }; @@ -612,15 +612,15 @@ async fn xtream_get_stream_info_response(app_state: &AppState, user: &ProxyUserC Err(_) => return axum::http::StatusCode::BAD_REQUEST.into_response() }; - if let Ok((pli, virtual_record)) = xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.config, target, Some(cluster)) { + if let Ok((pli, virtual_record)) = xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.app_config, target, Some(cluster)) { if pli.provider_id > 0 { let input_name = &pli.input_name; - if let Some(input) = app_state.config.get_input_by_name(input_name.as_str()) { + if let Some(input) = app_state.app_config.get_input_by_name(input_name.as_str()) { if let Some(info_url) = xtream::get_xtream_player_api_info_url(&input, cluster, pli.provider_id) { // Redirect is only possible for live streams, vod and series info needs to be modified if user.proxy == ProxyType::Redirect && cluster == XtreamCluster::Live { return redirect(&info_url).into_response(); - } else if let Ok(content) = xtream::get_xtream_stream_info(Arc::clone(&app_state.http_client), &app_state.config, user, &input, target, &pli, info_url.as_str(), cluster).await { + } else if let Ok(content) = xtream::get_xtream_stream_info(Arc::clone(&app_state.http_client), &app_state.app_config, user, &input, target, &pli, info_url.as_str(), cluster).await { return axum::response::Response::builder() .status(StatusCode::OK) .header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string()) @@ -666,10 +666,10 @@ async fn xtream_get_short_epg(app_state: &AppState, user: &ProxyUserCredentials, Err(_) => return axum::http::StatusCode::BAD_REQUEST.into_response() }; - if let Ok((pli, _)) = xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.config, target, None) { + if let Ok((pli, _)) = xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.app_config, target, None) { if pli.provider_id > 0 { let input_name = &pli.input_name; - if let Some(input) = app_state.config.get_input_by_name(input_name.as_str()) { + if let Some(input) = app_state.app_config.get_input_by_name(input_name.as_str()) { if let Some(action_url) = xtream::get_xtream_player_api_action_url(&input, crate::model::XC_ACTION_GET_SHORT_EPG) { let mut info_url = format!("{action_url}&{}={}", crate::model::XC_TAG_STREAM_ID, pli.provider_id); if !(limit.is_empty() || limit.eq("0")) { @@ -727,16 +727,16 @@ async fn xtream_player_api_handle_content_action(config: &Config, target_name: & async fn xtream_get_catchup_response(app_state: &AppState, target: &ConfigTarget, stream_id: &str, start: &str, end: &str) -> impl IntoResponse + Send { let virtual_id: u32 = try_result_bad_request!(FromStr::from_str(stream_id)); - let (pli, _) = try_result_bad_request!(xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.config, target, Some(XtreamCluster::Live))); - let input = try_option_bad_request!(app_state.config.get_input_by_name(pli.input_name.as_str())); + let (pli, _) = try_result_bad_request!(xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.app_config, target, Some(XtreamCluster::Live))); + let input = try_option_bad_request!(app_state.app_config.get_input_by_name(pli.input_name.as_str())); let info_url = try_option_bad_request!(xtream::get_xtream_player_api_action_url(&input, crate::model::XC_ACTION_GET_CATCHUP_TABLE) .map(|action_url| format!("{action_url}&{}={}&start={start}&end={end}", crate::model::XC_TAG_STREAM_ID, pli.provider_id))); let content = try_result_bad_request!(xtream::get_xtream_stream_info_content(Arc::clone(&app_state.http_client), info_url.as_str(), &input).await); let mut doc: Map = try_result_bad_request!(serde_json::from_str(&content)); let epg_listings = try_option_bad_request!(doc.get_mut(crate::model::XC_TAG_EPG_LISTINGS).and_then(Value::as_array_mut)); - let config = &app_state.config.config.load(); + let config = &app_state.app_config.config.load(); let target_path = try_option_bad_request!(get_target_storage_path(config, target.name.as_str())); - let (mut target_id_mapping, file_lock) = get_target_id_mapping(&app_state.config, &target_path).await; + let (mut target_id_mapping, file_lock) = get_target_id_mapping(&app_state.app_config, &target_path).await; for epg_list_item in epg_listings.iter_mut().filter_map(Value::as_object_mut) { // TODO epg_id if let Some(catchup_provider_id) = epg_list_item.get(crate::model::XC_TAG_ID).and_then(Value::as_str).and_then(|id| id.parse::().ok()) { @@ -798,7 +798,7 @@ async fn xtream_player_api( } // Process specific playlist actions - let (skip_live, skip_vod, skip_series) = if let Some(inputs) = app_state.config.get_inputs_for_target(&target.name) { + let (skip_live, skip_vod, skip_series) = if let Some(inputs) = app_state.app_config.get_inputs_for_target(&target.name) { inputs.iter().fold((true, true, true), |acc, i| { let (l, v, s) = acc; i.options.as_ref().map_or((false, false, false), |o| (l && o.xtream_skip_live, v && o.xtream_skip_vod, s && o.xtream_skip_series)) @@ -831,18 +831,18 @@ async fn xtream_player_api( let category_id = api_req.category_id.trim().parse::().ok(); // Handle general content actions if let Some(response) = xtream_player_api_handle_content_action( - &app_state.config.config.load(), &target.name, action, category_id, &user, + &app_state.app_config.config.load(), &target.name, action, category_id, &user, ).await { return response.into_response(); } let result = match action { crate::model::XC_ACTION_GET_LIVE_STREAMS => - skip_flag_optional!(skip_live, xtream_repository::xtream_load_rewrite_playlist(XtreamCluster::Live, &app_state.config, &target, category_id, &user).await), + skip_flag_optional!(skip_live, xtream_repository::xtream_load_rewrite_playlist(XtreamCluster::Live, &app_state.app_config, &target, category_id, &user).await), crate::model::XC_ACTION_GET_VOD_STREAMS => - skip_flag_optional!(skip_vod, xtream_repository::xtream_load_rewrite_playlist(XtreamCluster::Video, &app_state.config, &target, category_id, &user).await), + skip_flag_optional!(skip_vod, xtream_repository::xtream_load_rewrite_playlist(XtreamCluster::Video, &app_state.app_config, &target, category_id, &user).await), crate::model::XC_ACTION_GET_SERIES => - skip_flag_optional!(skip_series, xtream_repository::xtream_load_rewrite_playlist(XtreamCluster::Series, &app_state.config, &target, category_id, &user).await), + skip_flag_optional!(skip_series, xtream_repository::xtream_load_rewrite_playlist(XtreamCluster::Series, &app_state.app_config, &target, category_id, &user).await), _ => Some(Err(info_err!(format!("Cant find action: {action} for target: {}", &target.name)) )), }; diff --git a/backend/src/api/main_api.rs b/backend/src/api/main_api.rs index 356c07c2b..0bc2ebeb7 100644 --- a/backend/src/api/main_api.rs +++ b/backend/src/api/main_api.rs @@ -81,7 +81,7 @@ async fn create_shared_data(app_config: &Arc) -> AppState { let client = builder.build().unwrap_or_else(|_| Client::new()); AppState { - config: Arc::clone(app_config), + app_config: Arc::clone(app_config), http_client: Arc::new(client), downloads: Arc::new(DownloadQueue::new()), cache, @@ -175,7 +175,7 @@ fn create_compression_layer() -> tower_http::compression::CompressionLayer { fn start_hdhomerun(app_config: &Arc, app_state: &Arc, infos: &mut Vec) { let config = app_config.config.load(); let host = config.api.host.to_string(); - let guard = app_config.t_hdhomerun.load(); + let guard = app_config.hdhomerun.load(); if let Some(hdhomerun) = &*guard { if hdhomerun.enabled { for device in &hdhomerun.devices { @@ -247,7 +247,7 @@ pub async fn start_server(app_config: Arc, targets: Arc, allocation: ProviderAllocation, } impl ProviderConnectionGuard { - pub fn new(_manager: Arc, allocation: ProviderAllocation) -> Self { + pub fn new(allocation: ProviderAllocation) -> Self { Self { - // manager, allocation, } } @@ -395,20 +393,18 @@ impl MultiProviderLineup { } pub struct ActiveProviderManager { - grace_period_millis: u64, - grace_period_timeout_secs: u64, + grace_period_millis: AtomicU64, + grace_period_timeout_secs: AtomicU64, providers: Arc>>, } impl ActiveProviderManager { pub async fn new(cfg: &AppConfig) -> Self { - let (grace_period_millis, grace_period_timeout_secs) = cfg.config.load().reverse_proxy.as_ref() - .and_then(|r| r.stream.as_ref()) - .map_or_else(|| (default_grace_period_millis(), default_grace_period_timeout_secs()), |s| (s.grace_period_millis, s.grace_period_timeout_secs)); + let (grace_period_millis, grace_period_timeout_secs) = Self::get_grace_options(cfg); let mut this = Self { - grace_period_millis, - grace_period_timeout_secs, + grace_period_millis: AtomicU64::new(grace_period_millis), + grace_period_timeout_secs: AtomicU64::new(grace_period_timeout_secs), providers: Arc::new(RwLock::new(Vec::new())), }; for source in &cfg.sources.load().sources { @@ -419,12 +415,17 @@ impl ActiveProviderManager { this } - fn clone_inner(&self) -> Self { - Self { - grace_period_millis: self.grace_period_millis, - grace_period_timeout_secs: self.grace_period_timeout_secs, - providers: Arc::clone(&self.providers), - } + fn get_grace_options(cfg: &AppConfig) -> (u64, u64) { + let (grace_period_millis, grace_period_timeout_secs) = cfg.config.load().reverse_proxy.as_ref() + .and_then(|r| r.stream.as_ref()) + .map_or_else(|| (default_grace_period_millis(), default_grace_period_timeout_secs()), |s| (s.grace_period_millis, s.grace_period_timeout_secs)); + (grace_period_millis, grace_period_timeout_secs) + } + + pub fn update_config(&self, cfg: &AppConfig) { + let (grace_period_millis, grace_period_timeout_secs) = Self::get_grace_options(cfg); + self.grace_period_millis.store(grace_period_millis, Ordering::Relaxed); + self.grace_period_timeout_secs.store(grace_period_timeout_secs, Ordering::Relaxed); } pub async fn add_provider(&mut self, input: &ConfigInput) { @@ -474,7 +475,7 @@ impl ActiveProviderManager { Some((_lineup, config)) => config.force_allocate().await, }; - ProviderConnectionGuard::new(Arc::new(self.clone_inner()), allocation) + ProviderConnectionGuard::new(allocation) } // Returns the next available provider connection @@ -482,7 +483,8 @@ impl ActiveProviderManager { let providers = self.providers.read().await; let allocation = match Self::get_provider_config(input_name, &providers) { None => ProviderAllocation::Exhausted, // No Name matched, we don't have this provider - Some((lineup, _config)) => lineup.acquire(self.grace_period_millis > 0, self.grace_period_timeout_secs).await + Some((lineup, _config)) => lineup.acquire(self.grace_period_millis.load(Ordering::Relaxed) > 0, + self.grace_period_timeout_secs.load(Ordering::Relaxed)).await }; if log_enabled!(log::Level::Debug) { @@ -495,7 +497,7 @@ impl ActiveProviderManager { } } - ProviderConnectionGuard::new(Arc::new(self.clone_inner()), allocation) + ProviderConnectionGuard::new(allocation) } // This method is used for redirects to cycle through provider @@ -505,7 +507,7 @@ impl ActiveProviderManager { match Self::get_provider_config(input_name, &providers) { None => None, Some((lineup, _config)) => { - let cfg = lineup.get_next(self.grace_period_timeout_secs).await; + let cfg = lineup.get_next(self.grace_period_timeout_secs.load(Ordering::Relaxed)).await; if log_enabled!(log::Level::Debug) { if let Some(ref c) = cfg { debug!("Using provider {}", c.name); @@ -564,7 +566,7 @@ impl ActiveProviderManager { pub async fn is_over_limit(&self, provider_name: &str) -> bool { let providers = self.providers.read().await; if let Some((_, config)) = Self::get_provider_config(provider_name, &providers) { - config.is_over_limit(self.grace_period_timeout_secs).await + config.is_over_limit(self.grace_period_timeout_secs.load(Ordering::Relaxed)).await } else { false } diff --git a/backend/src/api/model/active_user_manager.rs b/backend/src/api/model/active_user_manager.rs index 583be66c2..ec6b43b2e 100644 --- a/backend/src/api/model/active_user_manager.rs +++ b/backend/src/api/model/active_user_manager.rs @@ -4,16 +4,76 @@ use shared::utils::{current_time_secs, default_grace_period_millis, default_grac use jsonwebtoken::get_current_timestamp; use log::{debug, info}; use std::collections::HashMap; -use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::Arc; use tokio::sync::RwLock; use shared::model::UserConnectionPermission; +macro_rules! active_user_manager_shared_impl { + () => { + #[inline] + async fn get_active_connections(user: &Arc>>) -> usize { + user.read().await.values().map(|c| c.connections as usize).sum() + } + + fn log_active_user(&self) { + if self.is_log_user_enabled() { + let user = Arc::clone(&self.user); + tokio::spawn(async move { + let user_count = user.read().await.len(); + let user_connection_count = Self::get_active_connections(&user).await; + info!("Active Users: {user_count}, Active User Connections: {user_connection_count}"); + }); + } + } + }; +} + const USER_CON_TTL: u64 = 10_800; // 3 hours const USER_SESSION_LIMIT: usize = 50; + + +fn get_grace_options(config: &Config) -> (u64, u64) { + let (grace_period_millis, grace_period_timeout_secs) = config.reverse_proxy.as_ref() + .and_then(|r| r.stream.as_ref()) + .map_or_else(|| (default_grace_period_millis(), default_grace_period_timeout_secs()), |s| (s.grace_period_millis, s.grace_period_timeout_secs)); + (grace_period_millis, grace_period_timeout_secs) +} + +struct ConnectionGuardUserManager { + log_active_user: bool, + user: Arc>>, +} + +impl ConnectionGuardUserManager { + active_user_manager_shared_impl!(); + fn is_log_user_enabled(&self)-> bool { + self.log_active_user + } + async fn remove_connection(&self, username: &str) { + let mut lock = self.user.write().await; + if let Some(connection_data) = lock.get_mut(username) { + if connection_data.connections > 0 { + connection_data.connections -= 1; + } + + if connection_data.connections == 0 { + lock.remove(username); + } else if connection_data.connections < connection_data.max_connections { + // Grace timeout expired, reset grace counters + connection_data.granted_grace = false; + connection_data.grace_ts = 0; + } + } + drop(lock); + + self.log_active_user(); + } +} + pub struct UserConnectionGuard { - manager: Arc, + manager: Arc, username: String, } impl Drop for UserConnectionGuard { @@ -68,9 +128,9 @@ impl UserConnectionData { } pub struct ActiveUserManager { - grace_period_millis: u64, - grace_period_timeout_secs: u64, - log_active_user: bool, + grace_period_millis: AtomicU64, + grace_period_timeout_secs: AtomicU64, + log_active_user: AtomicBool, user: Arc>>, gc_ts: Option, } @@ -78,26 +138,29 @@ pub struct ActiveUserManager { impl ActiveUserManager { pub fn new(config: &Config) -> Self { let log_active_user = config.log.as_ref().is_some_and(|l| l.log_active_user); - let (grace_period_millis, grace_period_timeout_secs) = config.reverse_proxy.as_ref() - .and_then(|r| r.stream.as_ref()) - .map_or_else(|| (default_grace_period_millis(), default_grace_period_timeout_secs()), |s| (s.grace_period_millis, s.grace_period_timeout_secs)); + let (grace_period_millis, grace_period_timeout_secs) = get_grace_options(config); Self { - grace_period_millis, - grace_period_timeout_secs, - log_active_user, + grace_period_millis : AtomicU64::new(grace_period_millis), + grace_period_timeout_secs: AtomicU64::new(grace_period_timeout_secs), + log_active_user: AtomicBool::new(log_active_user), user: Arc::new(RwLock::new(HashMap::new())), gc_ts: Some(AtomicU64::new(current_time_secs())), } } - fn clone_inner(&self) -> Self { - Self { - grace_period_millis: self.grace_period_millis, - grace_period_timeout_secs: self.grace_period_timeout_secs, - log_active_user: self.log_active_user, + pub fn update_config(&self, config: &Config) { + let log_active_user = config.log.as_ref().is_some_and(|l| l.log_active_user); + let (grace_period_millis, grace_period_timeout_secs) = get_grace_options(config); + self.grace_period_millis.store(grace_period_millis, Ordering::Relaxed); + self.grace_period_timeout_secs.store(grace_period_timeout_secs, Ordering::Relaxed); + self.log_active_user.store(log_active_user, Ordering::Relaxed); + } + + fn clone_inner(&self) -> ConnectionGuardUserManager { + ConnectionGuardUserManager { + log_active_user: self.log_active_user.load(Ordering::Relaxed), user: Arc::clone(&self.user), - gc_ts: None, } } @@ -121,7 +184,7 @@ impl ActiveUserManager { let now = get_current_timestamp(); // Check if user already used grace period if connection_data.granted_grace { - if current_connections > connection_data.max_connections && now - connection_data.grace_ts <= self.grace_period_timeout_secs { + if current_connections > connection_data.max_connections && now - connection_data.grace_ts <= self.grace_period_timeout_secs.load(Ordering::Relaxed) { // Grace timeout still active, deny connection debug!("User access denied, grace exhausted, too many connections: {username}"); return UserConnectionPermission::Exhausted; @@ -131,7 +194,7 @@ impl ActiveUserManager { connection_data.grace_ts = 0; } - if self.grace_period_millis > 0 && current_connections == connection_data.max_connections { + if self.grace_period_millis.load(Ordering::Relaxed) > 0 && current_connections == connection_data.max_connections { // Allow grace period once connection_data.granted_grace = true; connection_data.grace_ts = now; @@ -166,11 +229,6 @@ impl ActiveUserManager { Self::get_active_connections(&self.user).await } - #[inline] - async fn get_active_connections(user: &Arc>>) -> usize { - user.read().await.values().map(|c| c.connections as usize).sum() - } - pub async fn add_connection(&self, username: &str, max_connections: u32) -> UserConnectionGuard { let mut lock = self.user.write().await; if let Some(connection_data) = lock.get_mut(username) { @@ -189,24 +247,9 @@ impl ActiveUserManager { } } - async fn remove_connection(&self, username: &str) { - let mut lock = self.user.write().await; - if let Some(connection_data) = lock.get_mut(username) { - if connection_data.connections > 0 { - connection_data.connections -= 1; - } - - if connection_data.connections == 0 { - lock.remove(username); - } else if connection_data.connections < connection_data.max_connections { - // Grace timeout expired, reset grace counters - connection_data.granted_grace = false; - connection_data.grace_ts = 0; - } - } - drop(lock); - - self.log_active_user(); + active_user_manager_shared_impl!(); + fn is_log_user_enabled(&self)-> bool { + self.log_active_user.load(Ordering::Relaxed) } fn find_user_session<'a>(token: &'a str, sessions: &'a [UserSession]) -> Option<&'a UserSession> { @@ -294,17 +337,6 @@ impl ActiveUserManager { None } - fn log_active_user(&self) { - if self.log_active_user { - let user = Arc::clone(&self.user); - tokio::spawn(async move { - let user_count = user.read().await.len(); - let user_connection_count = Self::get_active_connections(&user).await; - info!("Active Users: {user_count}, Active User Connections: {user_connection_count}"); - }); - } - } - async fn gc(&self) { if let Some(gc_ts) = &self.gc_ts { let ts = gc_ts.load(Ordering::Acquire); diff --git a/backend/src/api/model/app_state.rs b/backend/src/api/model/app_state.rs index 5cad952c1..aacefdf03 100644 --- a/backend/src/api/model/app_state.rs +++ b/backend/src/api/model/app_state.rs @@ -1,16 +1,17 @@ use tokio::sync::{Mutex}; use std::sync::Arc; +use shared::error::TuliproxError; use shared::model::UserConnectionPermission; use crate::api::model::active_provider_manager::ActiveProviderManager; use crate::api::model::active_user_manager::ActiveUserManager; use crate::api::model::download::DownloadQueue; use crate::api::model::streams::shared_stream_manager::SharedStreamManager; -use crate::model::{AppConfig, HdHomeRunDeviceConfig}; +use crate::model::{AppConfig, Config, HdHomeRunDeviceConfig}; use crate::tools::lru_cache::LRUResourceCache; #[derive(Clone)] pub struct AppState { - pub config: Arc, + pub app_config: Arc, pub http_client: Arc, pub downloads: Arc, pub cache: Arc>>, @@ -20,6 +21,13 @@ pub struct AppState { } impl AppState { + + pub fn set_config(&self, config: Config) -> Result<(), TuliproxError> { + self.active_users.update_config(&config); + self.active_provider.update_config(&self.app_config); + self.app_config.set_config(config) + } + pub async fn get_active_connections_for_user(&self, username: &str) -> u32 { self.active_users.user_connections(username).await } diff --git a/backend/src/api/model/streams/active_client_stream.rs b/backend/src/api/model/streams/active_client_stream.rs index 9cba0b09e..edf8ad2ec 100644 --- a/backend/src/api/model/streams/active_client_stream.rs +++ b/backend/src/api/model/streams/active_client_stream.rs @@ -47,11 +47,12 @@ impl ActiveClientStream { let grant_user_grace_period = connection_permission == UserConnectionPermission::GracePeriod; let username = user.username.as_str(); let user_connection_guard = Some(active_user.add_connection(username, user.max_connections).await); - let cfg = &app_state.config; + let cfg = &app_state.app_config; let waker = Arc::new(Mutex::new(None)); let waker_clone = Arc::clone(&waker); let grace_stop_flag = Self::stream_grace_period(&stream_details, grant_user_grace_period, user, &active_user, &active_provider, &waker_clone); - let custom_video = cfg.t_custom_stream_response.as_ref() + let custom_response = cfg.custom_stream_response.load(); + let custom_video = custom_response.as_ref() .map_or((None, None), |c| ( c.user_connections_exhausted.clone(), @@ -59,7 +60,7 @@ impl ActiveClientStream { )); let stream = stream_details.stream.take().unwrap(); - let stream = match app_state.config.config.load().sleep_timer_mins { + let stream = match app_state.app_config.config.load().sleep_timer_mins { None => stream, Some(mins) => { let secs = u32::try_from((u64::from(mins) * 60).min(u64::from(u32::MAX))).unwrap_or(0); diff --git a/backend/src/api/model/streams/provider_stream.rs b/backend/src/api/model/streams/provider_stream.rs index fc7d999e4..0895b15a5 100644 --- a/backend/src/api/model/streams/provider_stream.rs +++ b/backend/src/api/model/streams/provider_stream.rs @@ -30,22 +30,26 @@ fn create_video_stream(video_buffer: Option<&TransportStreamBuffer>, headers: &[ } pub fn create_channel_unavailable_stream(cfg: &AppConfig, headers: &[(String, String)], status: StatusCode) -> ProviderStreamResponse { - let video = cfg.t_custom_stream_response.as_ref().and_then(|c| c.channel_unavailable.as_ref()); + let custom_stream_response = cfg.custom_stream_response.load(); + let video = custom_stream_response.as_ref().and_then(|c| c.channel_unavailable.as_ref()); create_video_stream(video, headers, &format!("Streaming response channel unavailable for status {status}")) } pub fn create_user_connections_exhausted_stream(cfg: &AppConfig, headers: &[(String, String)]) -> ProviderStreamResponse { - let video = cfg.t_custom_stream_response.as_ref().and_then(|c| c.user_connections_exhausted.as_ref()); + let custom_stream_response = cfg.custom_stream_response.load(); + let video = custom_stream_response.as_ref().and_then(|c| c.user_connections_exhausted.as_ref()); create_video_stream(video, headers, "Streaming response user connections exhausted") } pub fn create_provider_connections_exhausted_stream(cfg: &AppConfig, headers: &[(String, String)]) -> ProviderStreamResponse { - let video = cfg.t_custom_stream_response.as_ref().and_then(|c| c.provider_connections_exhausted.as_ref()); + let custom_stream_response = cfg.custom_stream_response.load(); + let video = custom_stream_response.as_ref().and_then(|c| c.provider_connections_exhausted.as_ref()); create_video_stream(video, headers, "Streaming response provider connections exhausted") } pub fn create_user_account_expired_stream(cfg: &AppConfig, headers: &[(String, String)]) -> ProviderStreamResponse { - let video = cfg.t_custom_stream_response.as_ref().and_then(|c| c.user_account_expired.as_ref()); + let custom_stream_response = cfg.custom_stream_response.load(); + let video = custom_stream_response.as_ref().and_then(|c| c.user_account_expired.as_ref()); create_video_stream(video, headers, "Streaming response user account expired") } diff --git a/backend/src/auth/authenticator.rs b/backend/src/auth/authenticator.rs index 30d2874dd..3ba48575d 100644 --- a/backend/src/auth/authenticator.rs +++ b/backend/src/auth/authenticator.rs @@ -82,7 +82,7 @@ fn validate_request( token: &str, verify_fn: fn(&str, &[u8]) -> bool, ) -> Result<(), ()> { - if let Some(web_auth_config) = &app_state.config.config.load().web_ui.as_ref().and_then(|c| c.auth.as_ref()) { + if let Some(web_auth_config) = &app_state.app_config.config.load().web_ui.as_ref().and_then(|c| c.auth.as_ref()) { let secret_key = web_auth_config.secret.as_ref(); if verify_fn(token, secret_key) { return Ok(()); @@ -111,7 +111,7 @@ pub async fn validator_user( next: axum::middleware::Next, ) -> Result { if let Some(username) = get_username_from_auth_header(&token, &app_state) { - if let Some(user) = app_state.config.get_user_credentials(&username) { + if let Some(user) = app_state.app_config.get_user_credentials(&username) { if !user.ui_enabled { return Err(axum::http::StatusCode::FORBIDDEN); } diff --git a/backend/src/main.rs b/backend/src/main.rs index 38d7f4820..185525925 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -20,6 +20,7 @@ use log::{error, info}; use std::fs::File; use std::path::{Path, PathBuf}; use std::sync::Arc; +use shared::model::ConfigPaths; use shared::utils::set_sanitize_sensitive_info; use crate::utils::init_logger; @@ -92,12 +93,9 @@ fn main() { return; } - let config_path: String = utils::resolve_directory_path(&resolve_env_var(&args.config_path.unwrap_or_else(utils::get_default_config_path))); - let config_file: String = resolve_env_var(&args.config_file.unwrap_or_else(|| utils::get_default_config_file_path(&config_path))); - let api_proxy_file = resolve_env_var(&args.api_proxy.unwrap_or_else(|| utils::get_default_api_proxy_config_path(config_path.as_str()))); - let mappings_file = args.mapping_file.as_ref(); + let mut config_paths = get_file_paths(&args); - init_logger(args.log_level.as_ref(), config_file.as_str()); + init_logger(args.log_level.as_ref(), config_paths.config_file_path.as_str()); info!("Version: {VERSION}"); if let Some(bts) = BUILD_TIMESTAMP.to_string().parse::>().ok().map(|datetime| datetime.format("%Y-%m-%d %H:%M:%S %Z").to_string()) { @@ -105,13 +103,10 @@ fn main() { } if args.healthcheck { - healthcheck(config_file.as_str()); + healthcheck(config_paths.config_file_path.as_str()); } - let sources_file: String = args.source_file.unwrap_or_else(|| utils::get_default_sources_file_path(&config_path)); - let app_config = utils::read_config(config_path.as_str(), config_file.as_str(), - sources_file.as_str(), api_proxy_file.as_str(), - mappings_file.cloned(), true).unwrap_or_else(|err| exit!("{}", err)); + let app_config = utils::read_app_config(&mut config_paths, true, true, args.server).unwrap_or_else(|err| exit!("{}", err)); let config = app_config.config.load(); set_sanitize_sensitive_info(config.log.as_ref().is_none_or(|l| l.sanitize_sensitive_info)); @@ -123,39 +118,28 @@ fn main() { let sources = app_config.sources.load(); let targets = sources.validate_targets(args.target.as_ref()).unwrap_or_else(|err| exit!("{}", err)); + let paths = app_config.paths.load(); info!("Current time: {}", chrono::offset::Local::now().format("%Y-%m-%d %H:%M:%S")); info!("Temp dir: {}", temp_path.display()); info!("Working dir: {:?}", &config.working_dir); - info!("Config dir: {:?}", &app_config.t_config_path); - info!("Config file: {:?}", &app_config.t_config_file_path); - info!("Source file: {:?}", &app_config.t_sources_file_path); - info!("Api Proxy File: {:?}", &app_config.t_api_proxy_file_path); - match utils::read_mappings(&app_config.t_mapping_file_path, true) { - Ok(Some(mappings)) => { - info!("Mapping file: {:?}", &app_config.t_mapping_file_path); - app_config.set_mappings(&mappings); - } - Ok(None) => { - info!("Mapping file: not used"); - }, - Err(err) => exit!("{err}"), - } + info!("Config dir: {:?}", &paths.config_path); + info!("Config file: {:?}", &paths.config_file_path); + info!("Source file: {:?}", &paths.sources_file_path); + info!("Api Proxy File: {:?}", &paths.api_proxy_file_path); + info!("Mapping file: {:?}", &paths.mapping_file_path.as_ref().map_or_else(|| "not used", |v| v.as_str())); + if let Some(cache) = config.reverse_proxy.as_ref().and_then(|r| r.cache.as_ref()) { if cache.enabled { info!("Cache dir: {}", cache.dir); } } - if let Some(resource_path) = app_config.t_custom_stream_response_path.as_ref() { + if let Some(resource_path) = paths.custom_stream_response_path.as_ref() { info!("Resource path: {resource_path}"); } let rt = tokio::runtime::Runtime::new().unwrap(); let () = rt.block_on(async { if args.server { - match utils::read_api_proxy_config(&app_config) { - Ok(()) => {} - Err(err) => exit!("{err}"), - } start_in_server_mode(Arc::new(app_config), Arc::new(targets)).await; } else { start_in_cli_mode(Arc::new(app_config), Arc::new(targets)).await; @@ -163,6 +147,23 @@ fn main() { }); } +fn get_file_paths(args: &Args) -> ConfigPaths { + let config_path: String = utils::resolve_directory_path(&resolve_env_var(&args.config_path.as_ref().map_or_else(utils::get_default_config_path, ToString::to_string))); + let config_file: String = resolve_env_var(&args.config_file.as_ref().map_or_else(|| utils::get_default_config_file_path(&config_path), ToString::to_string)); + let api_proxy_file = resolve_env_var(&args.api_proxy.as_ref().map_or_else(|| utils::get_default_api_proxy_config_path(config_path.as_str()), ToString::to_string)); + let sources_file: String = resolve_env_var(&args.source_file.as_ref().map_or_else(|| utils::get_default_sources_file_path(&config_path), ToString::to_string)); + let mappings_file = args.mapping_file.as_ref().map(|p| resolve_env_var(p)); + + ConfigPaths { + config_path, + config_file_path: config_file, + sources_file_path: sources_file, + mapping_file_path: mappings_file, // need to be set after config read + api_proxy_file_path: api_proxy_file, + custom_stream_response_path: None, + } +} + fn create_directories(cfg: &Config, temp_path: &Path) { // Collect the paths into a vector. let paths_strings = [ diff --git a/backend/src/model/config/api_proxy.rs b/backend/src/model/config/api_proxy.rs index 3f7ba7f0e..ae6f986f5 100644 --- a/backend/src/model/config/api_proxy.rs +++ b/backend/src/model/config/api_proxy.rs @@ -7,7 +7,7 @@ use std::fs; use std::sync::Arc; use arc_swap::access::Access; use arc_swap::ArcSwap; -use shared::model::{ApiProxyConfigDto, ApiProxyServerInfoDto, TargetUserDto}; +use shared::model::{ApiProxyConfigDto, ApiProxyServerInfoDto, ConfigPaths, TargetUserDto}; use crate::{utils}; #[derive(Debug, Clone)] @@ -98,6 +98,8 @@ impl ApiProxyConfig { // When we switch from one to other we need to migrate the existing data. /// # Panics pub fn migrate_api_user(&mut self, cfg: &AppConfig, errors: &mut Vec) { + let paths = > as Access>::load(&cfg.paths); + let api_proxy_file = paths.api_proxy_file_path.as_str(); if self.use_user_db { // we have user defined in config file. // we migrate them to the db and delete them from the config file @@ -105,7 +107,6 @@ impl ApiProxyConfig { if let Err(err) = merge_api_user(cfg, &self.user) { errors.push(err.to_string()); } else { - let api_proxy_file = cfg.t_api_proxy_file_path.as_str(); let config = > as Access>::load(&cfg.config); let backup_dir = config.backup_dir.as_ref().unwrap().as_str(); self.user = vec![]; @@ -141,7 +142,6 @@ impl ApiProxyConfig { } } } - let api_proxy_file = cfg.t_api_proxy_file_path.as_str(); let config = > as Access>::load(&cfg.config); let backup_dir = config.backup_dir.as_ref().unwrap().as_str(); diff --git a/backend/src/model/config/api_user.rs b/backend/src/model/config/api_user.rs index e41c79b44..d6ae68983 100644 --- a/backend/src/model/config/api_user.rs +++ b/backend/src/model/config/api_user.rs @@ -77,7 +77,7 @@ impl ProxyUserCredentials { } pub fn has_permissions(&self, app_state: &AppState) -> bool { - let config = > as Access>::load(&app_state.config.config); + let config = > as Access>::load(&app_state.app_config.config); if config.user_access_control { if let Some(exp_date) = self.exp_date.as_ref() { let now = Local::now(); @@ -103,7 +103,7 @@ impl ProxyUserCredentials { } pub async fn connection_permission(&self, app_state: &AppState) -> UserConnectionPermission { - let config = > as Access>::load(&app_state.config.config); + let config = > as Access>::load(&app_state.app_config.config); if self.max_connections > 0 && config.user_access_control { // we allow requests with max connection reached, but we should block streaming after grace period return app_state.get_connection_permission(&self.username, self.max_connections).await; diff --git a/backend/src/model/config/app.rs b/backend/src/model/config/app.rs index 0474df228..b3c3b85c0 100644 --- a/backend/src/model/config/app.rs +++ b/backend/src/model/config/app.rs @@ -10,6 +10,7 @@ use log::{debug, error}; use rand::Rng; use shared::create_tuliprox_error_result; use shared::error::{TuliproxError, TuliproxErrorKind}; +use shared::model::ConfigPaths; use crate::api::model::streams::transport_stream_buffer::TransportStreamBuffer; use crate::model::{ApiProxyConfig, ApiProxyServerInfo, Config, ConfigInput, ConfigInputOptions, ConfigTarget, CustomStreamResponse, HdHomeRunConfig, Mappings, ProxyUserCredentials, SourcesConfig, TargetOutput}; use crate::utils; @@ -31,26 +32,52 @@ fn generate_secret() -> [u8; 32] { pub struct AppConfig { pub config: Arc>, pub sources: Arc>, - pub t_hdhomerun: Arc>, - pub t_api_proxy: Arc>, - pub t_config_path: String, - pub t_config_file_path: String, - pub t_sources_file_path: String, - pub t_mapping_file_path: String, - pub t_api_proxy_file_path: String, - pub t_custom_stream_response_path: Option, + pub hdhomerun: Arc>, + pub api_proxy: Arc>, pub file_locks: Arc, - pub t_custom_stream_response: Option, - pub t_access_token_secret: [u8; 32], - pub t_encrypt_secret: [u8; 16], + pub paths: Arc>, + pub custom_stream_response: Arc>, + pub access_token_secret: [u8; 32], + pub encrypt_secret: [u8; 16], } impl AppConfig { - pub fn set_api_proxy(&self, api_proxy: Option>) -> Result<(), TuliproxError> { - self.t_api_proxy.store(api_proxy); + + pub fn set_config(&self, config: Config) -> Result<(), TuliproxError> { + self.config.store(Arc::new(config)); + self.prepare_custom_stream_response(); + Ok(()) + } + + pub fn set_sources(&self, sources: SourcesConfig) -> Result<(), TuliproxError> { + self.sources.store(Arc::new(sources)); + self.prepare_sources()?; + Ok(()) + } + + pub fn set_api_proxy(&self, api_proxy: ApiProxyConfig) -> Result<(), TuliproxError> { + self.api_proxy.store(Some(Arc::new(api_proxy))); self.check_target_user() } + pub fn set_mappings(&self, mappings_cfg: &Mappings) { + let sources = > as Access>::load(&self.sources); + for source in &sources.sources { + for target in &source.targets { + if let Some(mapping_ids) = &target.mapping_ids { + let mut target_mappings = Vec::with_capacity(128); + for mapping_id in mapping_ids { + let mapping = mappings_cfg.get_mapping(mapping_id); + if let Some(mappings) = mapping { + target_mappings.push(mappings); + } + } + target.mapping.store(if target_mappings.is_empty() { None } else { Some(Arc::new(target_mappings)) }); + } + } + } + } + fn check_username(&self, output_username: Option<&str>, target_name: &str) -> Result<(), TuliproxError> { if let Some(username) = output_username { if let Some((_, config_target)) = self.get_target_for_username(username) { @@ -66,7 +93,7 @@ impl AppConfig { } } fn check_target_user(&self) -> Result<(), TuliproxError> { - let check_homerun = self.t_hdhomerun.load().as_ref().is_some_and(|h| h.enabled); + let check_homerun = self.hdhomerun.load().as_ref().is_some_and(|h| h.enabled); let sources = > as Access>::load(&self.sources); for source in &sources.sources { for target in &source.targets { @@ -80,7 +107,7 @@ impl AppConfig { if check_homerun { let hdhr_name = &hdhomerun_output.device; self.check_username(Some(&hdhomerun_output.username), &target.name)?; - if let Some(old_hdhomerun) = self.t_hdhomerun.load().clone() { + if let Some(old_hdhomerun) = self.hdhomerun.load().clone() { let mut hdhomerun = (*old_hdhomerun).clone(); for device in &mut hdhomerun.devices { if &device.name == hdhr_name { @@ -88,7 +115,7 @@ impl AppConfig { device.t_enabled = true; } } - self.t_hdhomerun.store(Some(Arc::new(hdhomerun))); + self.hdhomerun.store(Some(Arc::new(hdhomerun))); } } } @@ -97,7 +124,7 @@ impl AppConfig { } } - let guard = self.t_hdhomerun.load(); + let guard = self.hdhomerun.load(); if let Some(hdhomerun) = &*guard { for device in &hdhomerun.devices { if !device.t_enabled { @@ -142,22 +169,22 @@ impl AppConfig { pub fn get_target_for_username(&self, username: &str) -> Option<(ProxyUserCredentials, Arc)> { if let Some(credentials) = self.get_user_credentials(username) { - return self.t_api_proxy.load().as_ref() + return self.api_proxy.load().as_ref() .and_then(|api_proxy| self.intern_get_target_for_user(api_proxy.get_target_name(&credentials.username, &credentials.password))); } None } pub fn get_target_for_user(&self, username: &str, password: &str) -> Option<(ProxyUserCredentials, Arc)> { - self.t_api_proxy.load().as_ref().and_then(|api_proxy| self.intern_get_target_for_user(api_proxy.get_target_name(username, password))) + self.api_proxy.load().as_ref().and_then(|api_proxy| self.intern_get_target_for_user(api_proxy.get_target_name(username, password))) } pub fn get_target_for_user_by_token(&self, token: &str) -> Option<(ProxyUserCredentials, Arc)> { - self.t_api_proxy.load().as_ref().as_ref().and_then(|api_proxy| self.intern_get_target_for_user(api_proxy.get_target_name_by_token(token))) + self.api_proxy.load().as_ref().as_ref().and_then(|api_proxy| self.intern_get_target_for_user(api_proxy.get_target_name_by_token(token))) } pub fn get_user_credentials(&self, username: &str) -> Option { - self.t_api_proxy.load().as_ref().as_ref().and_then(|api_proxy| api_proxy.get_user_credentials(username)) + self.api_proxy.load().as_ref().as_ref().and_then(|api_proxy| api_proxy.get_user_credentials(username)) } pub fn get_input_by_name(&self, input_name: &str) -> Option> { @@ -201,24 +228,6 @@ impl AppConfig { sources.get_target_by_id(target_id) } - pub fn set_mappings(&self, mappings_cfg: &Mappings) { - let sources = > as Access>::load(&self.sources); - for source in &sources.sources { - for target in &source.targets { - if let Some(mapping_ids) = &target.mapping_ids { - let mut target_mappings = Vec::with_capacity(128); - for mapping_id in mapping_ids { - let mapping = mappings_cfg.get_mapping(mapping_id); - if let Some(mappings) = mapping { - target_mappings.push(mappings); - } - } - target.mapping.store(if target_mappings.is_empty() { None } else { Some(Arc::new(target_mappings)) }); - } - } - } - } - fn check_unique_input_names(&self) -> Result<(), TuliproxError> { let mut seen_names = HashSet::new(); let sources = > as Access>::load(&self.sources); @@ -250,7 +259,7 @@ impl AppConfig { } - fn check_scheduled_targets(&mut self, target_names: &HashSet>) -> Result<(), TuliproxError> { + fn check_scheduled_targets(&self, target_names: &HashSet>) -> Result<(), TuliproxError> { let config = > as Access>::load(&self.config); if let Some(schedules) = &config.schedules { for schedule in schedules { @@ -272,21 +281,25 @@ impl AppConfig { pub fn prepare(&mut self, include_computed: bool) -> Result<(), TuliproxError> { if include_computed { - self.t_access_token_secret = generate_secret(); - self.t_encrypt_secret = <&[u8] as TryInto<[u8; 16]>>::try_into(&generate_secret()[0..16]).map_err(|err| TuliproxError::new(TuliproxErrorKind::Info, err.to_string()))?; + self.access_token_secret = generate_secret(); + self.encrypt_secret = <&[u8] as TryInto<[u8; 16]>>::try_into(&generate_secret()[0..16]).map_err(|err| TuliproxError::new(TuliproxErrorKind::Info, err.to_string()))?; self.prepare_custom_stream_response(); } - let sources = > as Access>::load(&self.sources); - let target_names = sources.get_unique_target_names(); - self.check_scheduled_targets(&target_names)?; - self.check_unique_input_names()?; + self.prepare_sources()?; Ok(()) } + fn prepare_sources(&self) -> Result<(), TuliproxError> { + let sources = > as Access>::load(&self.sources); + let target_names = sources.get_unique_target_names(); + self.check_scheduled_targets(&target_names)?; + self.check_unique_input_names()?; + Ok(()) + } - fn prepare_custom_stream_response(&mut self) { + fn prepare_custom_stream_response(&self) { let config = > as Access>::load(&self.config); if let Some(custom_stream_response_path) = config.custom_stream_response_path.as_ref() { fn load_and_set_file(file_path: &Path) -> Option { @@ -323,17 +336,22 @@ impl AppConfig { let path = PathBuf::from(custom_stream_response_path); let path = utils::make_path_absolute(&path, &config.working_dir); - self.t_custom_stream_response_path = Some(path.to_string_lossy().to_string()); + + let paths = self.paths.load_full(); + let mut new_paths = paths.as_ref().clone(); + new_paths.custom_stream_response_path = Some(path.to_string_lossy().to_string()); + self.paths.store(Arc::new(new_paths)); + let channel_unavailable = load_and_set_file(&path.join(CHANNEL_UNAVAILABLE)); let user_connections_exhausted = load_and_set_file(&path.join(USER_CONNECTIONS_EXHAUSTED)); let provider_connections_exhausted = load_and_set_file(&path.join(PROVIDER_CONNECTIONS_EXHAUSTED)); let user_account_expired = load_and_set_file(&path.join(USER_ACCOUNT_EXPIRED)); - self.t_custom_stream_response = Some(CustomStreamResponse { + self.custom_stream_response.store(Some(Arc::new(CustomStreamResponse { channel_unavailable, user_connections_exhausted, provider_connections_exhausted, user_account_expired, - }); + }))); } } @@ -342,7 +360,7 @@ impl AppConfig { /// /// Will panic if default server invalid pub fn get_server_info(&self, server_info_name: &str) -> ApiProxyServerInfo { - let guard = self.t_api_proxy.load(); + let guard = self.api_proxy.load(); if let Ok(api_proxy) = guard.as_ref().ok_or_else(|| { TuliproxError::new(TuliproxErrorKind::Info, "API proxy config not loaded".to_string()) }) { diff --git a/backend/src/repository/user_repository.rs b/backend/src/repository/user_repository.rs index c086a1571..d92167c44 100644 --- a/backend/src/repository/user_repository.rs +++ b/backend/src/repository/user_repository.rs @@ -109,7 +109,8 @@ impl StoredProxyUserCredentials { pub fn get_api_user_db_path(cfg: &AppConfig) -> PathBuf { - PathBuf::from(&cfg.t_config_path).join(storage_const::API_USER_DB_FILE) + let paths = cfg.paths.load(); + PathBuf::from(&paths.config_path).join(storage_const::API_USER_DB_FILE) } @@ -397,10 +398,10 @@ pub async fn user_get_bouquet_filter(config: &Config, username: &str, category_i #[cfg(test)] mod tests { use super::*; - use shared::model::{ProxyType, ProxyUserStatus}; + use shared::model::{ConfigPaths, ProxyType, ProxyUserStatus}; use std::env::temp_dir; use std::sync::Arc; - use arc_swap::ArcSwapAny; + use arc_swap::{ArcSwap, ArcSwapAny, ArcSwapOption}; use crate::utils::FileLockManager; #[test] @@ -471,21 +472,22 @@ mod tests { let mut cfg = AppConfig { config: Arc::new(ArcSwapAny::default()), sources: Arc::new(ArcSwapAny::default()), - t_hdhomerun: Arc::new(ArcSwapAny::default()), - t_api_proxy: Arc::new(ArcSwapAny::default()), - t_config_path: String::new(), - t_config_file_path: String::new(), - t_sources_file_path: String::new(), - t_mapping_file_path: String::new(), - t_api_proxy_file_path: String::new(), - t_custom_stream_response_path: None, + hdhomerun: Arc::new(ArcSwapAny::default()), + api_proxy: Arc::new(ArcSwapAny::default()), + paths: Arc::new(ArcSwap::from(Arc::new(ConfigPaths { + config_path: temp_dir().to_string_lossy().to_string(), + config_file_path: "".to_string(), + sources_file_path: "".to_string(), + mapping_file_path: None, + api_proxy_file_path: "".to_string(), + custom_stream_response_path: None, + }))), file_locks: Arc::new(FileLockManager::default()), - t_custom_stream_response: None, - t_access_token_secret: Default::default(), - t_encrypt_secret: Default::default(), + custom_stream_response: Arc::new(ArcSwapAny::default()), + access_token_secret: Default::default(), + encrypt_secret: Default::default(), }; let target_user = vec![user]; - cfg.t_config_path = temp_dir().to_string_lossy().to_string(); let _ = store_api_user(&cfg, &target_user); let user_list = load_api_user(&cfg); diff --git a/backend/src/utils/file/config_reader.rs b/backend/src/utils/file/config_reader.rs index 24c1835be..ddcc82781 100644 --- a/backend/src/utils/file/config_reader.rs +++ b/backend/src/utils/file/config_reader.rs @@ -1,21 +1,22 @@ +use crate::model::Config; use crate::model::{ApiProxyConfig, AppConfig, SourcesConfig}; -use crate::model::{Config}; -use shared::error::{create_tuliprox_error, info_err, to_io_error, TuliproxError, TuliproxErrorKind}; -use crate::utils::{open_file, EnvResolvingReader, FileLockManager}; -use crate::utils::{file_reader}; +use crate::utils; +use crate::utils::file_reader; use crate::utils::sys_utils::exit; -use shared::utils::CONSTANTS; +use crate::utils::{open_file, read_mappings_file, EnvResolvingReader, FileLockManager}; +use arc_swap::{ArcSwap, ArcSwapAny}; use chrono::Local; use log::{error, info, warn}; use serde::Serialize; +use shared::error::{create_tuliprox_error, info_err, to_io_error, TuliproxError, TuliproxErrorKind}; +use shared::model::{ApiProxyConfigDto, AppConfigDto, ConfigDto, ConfigPaths, SourcesConfigDto}; +use shared::utils::CONSTANTS; use std::env; use std::fs::File; use std::io::{self, BufReader, Read}; -use std::path::{PathBuf}; +use std::path::PathBuf; use std::sync::Arc; -use arc_swap::{ArcSwap, ArcSwapAny}; -use shared::model::{ApiProxyConfigDto, ConfigDto, SourcesConfigDto}; -use crate::utils; +use arc_swap::access::{Access}; enum EitherReader { Left(L), @@ -41,32 +42,37 @@ pub fn config_file_reader(file: File, resolve_env: bool) -> impl Read } } -pub fn read_api_proxy_config(cfg: &AppConfig) -> Result<(), TuliproxError> { - let api_proxy_config = read_api_proxy(cfg, true); - match api_proxy_config { - None => { - warn!("cant read api_proxy_config file: {}", cfg.t_api_proxy_file_path.as_str()); - Ok(()) - } - Some(config) => { - cfg.set_api_proxy(Some(Arc::new(config)))?; - Ok(()) +pub fn read_api_proxy_config(config: &AppConfig, resolve_env: bool) -> Result, TuliproxError> { + let paths = > as Access>::load(&config.paths); + let api_proxy_file_path = paths.api_proxy_file_path.as_str(); + if let Some(api_proxy_dto) = read_api_proxy_file(api_proxy_file_path, resolve_env)? { + let mut errors = vec![]; + let mut api_proxy: ApiProxyConfig = ApiProxyConfig::from(&api_proxy_dto); + api_proxy.migrate_api_user(config, &mut errors); + if !errors.is_empty() { + for error in errors { + error!("{error}"); + } } + Ok(Some(api_proxy)) + } else { + warn!("cant read api_proxy_config file: {api_proxy_file_path}"); + Ok(None) } } -pub fn read_sources(sources_file: &str, resolve_env: bool, include_computed: bool) -> Result { - +pub fn read_sources_file(sources_file: &str, resolve_env: bool, include_computed: bool) -> Result { match open_file(&std::path::PathBuf::from(sources_file)) { Ok(file) => { let maybe_sources: Result = serde_yaml::from_reader(config_file_reader(file, resolve_env)); match maybe_sources { Ok(mut sources) => { - if let Err(err) = sources.prepare(include_computed) { - Err(info_err!(format!("Can't read the sources-config file: {sources_file}: {err}"))) - } else { - Ok(sources) + if resolve_env { + if let Err(err) = sources.prepare(include_computed) { + return Err(info_err!(format!("Can't read the sources-config file: {sources_file}: {err}"))); + } } + Ok(sources) } Err(err) => Err(info_err!(format!("Can't read the sources-config file: {sources_file}: {err}"))) } @@ -75,90 +81,142 @@ pub fn read_sources(sources_file: &str, resolve_env: bool, include_computed: boo } } -pub fn read_config(config_path: &str, config_file: &str, sources_file: &str, api_proxy_file: &str, mappings_file: Option, include_computed: bool) -> Result { - - let resolve_env = true; - let sources_dto = read_sources(sources_file, resolve_env, include_computed)?; - - let config_dto = match open_file(&std::path::PathBuf::from(config_file)) { +pub fn read_config_file(config_file: &str, resolve_env: bool) -> Result { + match open_file(&std::path::PathBuf::from(config_file)) { Ok(file) => { let maybe_config: Result = serde_yaml::from_reader(config_file_reader(file, resolve_env)); match maybe_config { - Ok(mut config) => { - config.prepare()?; - config - }, - Err(err) => return Err(info_err!(format!("Can't read the config file: {config_file}: {err}"))) + Ok(mut config) => { + if resolve_env { + config.prepare()?; + } + Ok(config) + } + Err(err) => Err(info_err!(format!("Can't read the config file: {config_file}: {err}"))) } } - Err(err) => return Err(info_err!(format!("Can't read the config file: {config_file}: {err}"))) + Err(err) => Err(info_err!(format!("Can't read the config file: {config_file}: {err}"))) + } +} + +pub fn read_app_config_dto(paths: &ConfigPaths, + resolve_env: bool, + include_computed: bool) -> Result { + let config_file = paths.config_file_path.as_str(); + let sources_file = paths.sources_file_path.as_str(); + let api_proxy_file = paths.api_proxy_file_path.as_str(); + + let sources = read_sources_file(sources_file, resolve_env, include_computed)?; + let config = read_config_file(config_file, resolve_env)?; + let mappings = if let Some(mappings_file) = paths.mapping_file_path.as_ref() { + read_mappings_file(mappings_file, resolve_env).unwrap_or(None) + } else { + None }; + let api_proxy = read_api_proxy_file(api_proxy_file, resolve_env).unwrap_or(None); + + Ok(AppConfigDto { + config, + sources, + mappings, + api_proxy, + }) +} + + +pub fn read_app_config(paths: &mut ConfigPaths, + resolve_env: bool, + include_computed: bool, + server_mode: bool) -> Result { + let config_path = paths.config_path.as_str(); + let config_file = paths.config_file_path.as_str(); + let sources_file = paths.sources_file_path.as_str(); + + let sources_dto = read_sources_file(sources_file, resolve_env, include_computed)?; let sources: SourcesConfig = sources_dto.into(); + let config_dto = read_config_file(config_file, resolve_env)?; let mut config: Config = Config::from(config_dto); config.prepare(config_path)?; - let mappings_file_path = config.mapping_path.as_ref().map_or_else(String::new, |mp| mp.trim().to_string()); - let mappings_file_path = if mappings_file_path.is_empty() { - resolve_env_var(&mappings_file.unwrap_or_else(|| utils::get_default_mappings_path(config_path))) - } else { - mappings_file_path - }; + if paths.mapping_file_path.is_none() { + let mut path = config.mapping_path.as_ref().map_or_else(|| utils::get_default_mappings_path(config_path), ToString::to_string); + if resolve_env { + path = resolve_env_var(&path); + } + paths.mapping_file_path.replace(path); + } let mut app_config = AppConfig { config: Arc::new(ArcSwap::from_pointee(config)), sources: Arc::new(ArcSwap::from_pointee(sources)), - t_hdhomerun: Arc::new(ArcSwapAny::default()), - t_api_proxy: Arc::new(ArcSwapAny::default()), - t_config_path: config_path.to_string(), - t_config_file_path: config_file.to_string(), - t_sources_file_path: sources_file.to_string(), - t_mapping_file_path: mappings_file_path, - t_api_proxy_file_path: api_proxy_file.to_string(), - t_custom_stream_response_path: None, + hdhomerun: Arc::new(ArcSwapAny::default()), + api_proxy: Arc::new(ArcSwapAny::default()), + paths: Arc::new(ArcSwap::from_pointee(paths.clone())), file_locks: Arc::new(FileLockManager::default()), - t_custom_stream_response: None, - t_access_token_secret: Default::default(), - t_encrypt_secret: Default::default(), + custom_stream_response: Arc::new(ArcSwapAny::default()), + access_token_secret: Default::default(), + encrypt_secret: Default::default(), }; app_config.prepare(include_computed)?; + + if let Some(mappings_file) = &paths.mapping_file_path { + match utils::read_mappings(mappings_file.as_str(), resolve_env) { + Ok(Some(mappings)) => app_config.set_mappings(&mappings), + Ok(None) => info!("Mapping file: not used"), + Err(err) => exit!("{err}"), + } + } + + if server_mode { + match read_api_proxy_config(&app_config, resolve_env) { + Ok(Some(api_proxy)) => app_config.set_api_proxy(api_proxy)?, + Ok(None) => info!("Api-Proxy file: not used"), + Err(err) => exit!("{err}"), + } + } + Ok(app_config) } -pub fn read_api_proxy_dto(config: &AppConfig, resolve_env: bool) -> Option { - let api_proxy_file = config.t_api_proxy_file_path.as_str(); - open_file(&std::path::PathBuf::from(api_proxy_file)).map_or(None, |file| { +pub fn read_api_proxy_file(api_proxy_file: &str, resolve_env: bool) -> Result, TuliproxError> { + open_file(&std::path::PathBuf::from(api_proxy_file)).map_or(Ok(None), |file| { let maybe_api_proxy: Result = serde_yaml::from_reader(config_file_reader(file, resolve_env)); match maybe_api_proxy { Ok(mut api_proxy_dto) => { - if let Err(err) = api_proxy_dto.prepare() { - exit!("cant read api-proxy-config file: {err}"); - } else { - Some(api_proxy_dto) + if resolve_env { + if let Err(err) = api_proxy_dto.prepare() { + exit!("cant read api-proxy-config file: {err}"); + } } + Ok(Some(api_proxy_dto)) } Err(err) => { - error!("cant read api-proxy-config file: {err}"); - None + Err(info_err!(format!("cant read api-proxy-config file: {err}"))) } } }) } - pub fn read_api_proxy(config: &AppConfig, resolve_env: bool) -> Option { - if let Some(api_proxy_dto) = read_api_proxy_dto(config, resolve_env) { - let mut errors = vec![]; - let mut api_proxy: ApiProxyConfig = api_proxy_dto.into(); - api_proxy.migrate_api_user(config, &mut errors); - if !errors.is_empty() { - for error in errors { - error!("{error}"); + let paths = > as Access>::load(&config.paths); + match read_api_proxy_file(paths.api_proxy_file_path.as_str(), resolve_env) { + Ok(Some(api_proxy_dto)) => { + let mut errors = vec![]; + let mut api_proxy: ApiProxyConfig = api_proxy_dto.into(); + api_proxy.migrate_api_user(config, &mut errors); + if !errors.is_empty() { + for error in errors { + error!("{error}"); + } } + Some(api_proxy) + } + Ok(None) => None, + Err(err) => { + error!("Failed to read Api-Proxy file {err}"); + None } - Some(api_proxy) - } else { - None } } diff --git a/backend/src/utils/file/mapping_reader.rs b/backend/src/utils/file/mapping_reader.rs index bfba68951..40c839ccf 100644 --- a/backend/src/utils/file/mapping_reader.rs +++ b/backend/src/utils/file/mapping_reader.rs @@ -113,14 +113,14 @@ fn read_mappings_from_directory(path: &Path, resolve_env: bool) -> Result