diff --git a/backend/src/api/endpoints/websocket_api.rs b/backend/src/api/endpoints/websocket_api.rs index 37061469a..8a319eef9 100644 --- a/backend/src/api/endpoints/websocket_api.rs +++ b/backend/src/api/endpoints/websocket_api.rs @@ -9,6 +9,7 @@ use axum::{ use log::{error, info}; use shared::model::{ProtocolHandler, ProtocolMessage, WsCloseCode, PROTOCOL_VERSION}; use std::sync::Arc; +use crate::api::model::event_manager::EventMessage; // WebSocket upgrade handler async fn websocket_handler( @@ -42,108 +43,130 @@ fn verify_auth_admin_token(auth_token: &str, secret_key: Option<&Vec>) -> bo } } +fn get_secret_key(app_state: &AppState, auth: bool) -> Option> { + if !auth { + return None; + } + + app_state.app_config.config.load() + .web_ui.as_ref() + .and_then(|c| c.auth.as_ref()) + .map(|c| { + let secret_key: &[u8] = c.secret.as_ref(); + secret_key.to_vec() + }) +} + +async fn handle_handshake( + msg: Message, + socket: &mut WebSocket, + version: u8, +) -> Result<(), String> { + if let Message::Binary(bytes) = msg { + if bytes.len() == 1 { + let client_version = bytes[0]; + if client_version == version { + socket.send(Message::binary(bytes)).await.map_err(|e| e.to_string())?; + return Ok(()); + } + error!("Protokol Version mismatch: server={version}, client={client_version}"); + } + } + + let _ = socket.send(Message::Close(Some(CloseFrame { + code: WsCloseCode::Protocol.code(), + reason: "Unsupported protocol".into(), + }))).await; + + Err("Protocol version mismatch".into()) +} + +async fn handle_protocol_message( + msg: Message, + socket: &mut WebSocket, + app_state: &Arc, + auth: bool, + secret_key: Option<&Vec>, +) -> Result<(), String> { + if let Message::Binary(bytes) = msg { + match ProtocolMessage::from_bytes(bytes) { + Ok(ProtocolMessage::StatusRequest(auth_token)) => { + if !auth || verify_auth_admin_token(&auth_token, secret_key) { + let status = create_status_check(app_state).await; + let response = ProtocolMessage::StatusResponse(status).to_bytes().map_err(|e| e.to_string())?; + socket.send(Message::Binary(response)).await.map_err(|e| e.to_string())?; + } + } + Ok(_) => { + error!("Unexpected protocol message after handshake"); + } + Err(e) => { + error!("Invalid websocket message: {e}"); + } + } + } + Ok(()) +} + +async fn handle_incoming_message( + result: Result, + socket: &mut WebSocket, + handler: &mut ProtocolHandler, + app_state: &Arc, + auth: bool, + secret_key: Option<&Vec>, +) -> Result<(), String> { + let msg = result.map_err(|e| e.to_string())?; + + match handler { + ProtocolHandler::Version(version) => { + handle_handshake(msg, socket, *version).await?; + *handler = ProtocolHandler::Default; + Ok(()) + }, + ProtocolHandler::Default => handle_protocol_message(msg, socket, app_state, auth, secret_key).await, + } +} + +async fn handle_event_message(socket: &mut WebSocket, event: EventMessage) -> Result<(), String> { + match event { + EventMessage::ActiveUserChange(users, connections) => { + let msg = ProtocolMessage::ActiveUserResponse(users, connections).to_bytes().map_err(|e| e.to_string())?; + socket.send(Message::Binary(msg)).await.map_err(|e| e.to_string()) + } + EventMessage::ActiveProviderChange(provider, connections) => { + let msg = ProtocolMessage::ActiveProviderResponse(provider, connections).to_bytes().map_err(|e| e.to_string())?; + socket.send(Message::Binary(msg)).await.map_err(|e| e.to_string()) + + } + } +} // WebSocket communication logic -#[allow(clippy::too_many_lines)] async fn handle_socket(mut socket: WebSocket, app_state: Arc, auth: bool) { - let secret_key = if auth { - 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: &[u8] = web_auth_config.secret.as_ref(); - Some(secret_key.to_vec()) - } else { - None - } - } else { - None - }; + let secret_key = get_secret_key(&app_state, auth); - let mut active_user_change_rx = app_state.active_users.get_active_user_change_channel(); - let mut active_provider_change_rx = app_state.active_provider.get_active_provider_change_channel(); + let mut event_rx = app_state.event_manager.get_event_channel(); let mut handler = ProtocolHandler::Version(PROTOCOL_VERSION); loop { tokio::select! { maybe_msg = socket.recv() => { - match maybe_msg { - Some(Ok(msg)) => { - match handler { - ProtocolHandler::Version(version) => { - let mut version_error = true; - if let Message::Binary(bytes) = msg { - if bytes.len() == 1 { - let client_version = bytes[0]; - if version == client_version { - if socket.send(Message::binary(bytes)).await.is_err() { - error!("Error sending websocket message"); - } else { - version_error = false; - handler = ProtocolHandler::Default; - } - } else { - error!("Version mismatch: server={version}, client={client_version}"); - } - } - } - if version_error { - let _ = socket.send(Message::Close(Some(CloseFrame { - code: WsCloseCode::Protocol.code(), - reason: "Unsupported protocol".into(), - }))).await; - break; - } - } - - ProtocolHandler::Default => { - if let Message::Binary(bytes) = msg { - match ProtocolMessage::from_bytes(bytes) { - Ok(ProtocolMessage::StatusRequest(auth_token)) => { - if !auth || verify_auth_admin_token(&auth_token, secret_key.as_ref()) { - let status = create_status_check(&app_state).await; - if let Ok(response) = ProtocolMessage::StatusResponse(status).to_bytes() { - if socket.send(Message::Binary(response)).await.is_err() { - error!("Failed to send websocket status response"); - } - } - } - } - Ok(_) => { - error!("Unexpected protocol message after handshake"); - } - Err(err) => { - error!("Invalid websocket message: {err}"); - } - } - } - } - } - } - Some(Err(err)) => { - error!("WebSocket error: {err}"); - break; - } - None => { - // WebSocket closed + if let Some(msg) = maybe_msg { + if let Err(e) = handle_incoming_message(msg, &mut socket, &mut handler, &app_state, auth, secret_key.as_ref()).await { + error!("WebSocket message handling error: {e}"); break; } + } else { + break; } } - Ok((user_count, connection_count)) = active_user_change_rx.recv() => { - if let Ok(payload) = ProtocolMessage::ActiveUserResponse(user_count, connection_count).to_bytes() { - if let Err(e) = socket.send(Message::Binary(payload)).await { - error!("Failed to send active user change: {e}"); - break; - } - } - } - - Ok((provider, connection_count)) = active_provider_change_rx.recv() => { - if let Ok(payload) = ProtocolMessage::ActiveProviderResponse(provider, connection_count).to_bytes() { - if let Err(e) = socket.send(Message::Binary(payload)).await { - error!("Failed to send active user change: {e}"); - break; - } + Ok(event) = event_rx.recv() => { + if let Err(e) = handle_event_message(&mut socket, event).await { + error!("Failed to send active user change: {e}"); + break; } } } diff --git a/backend/src/api/main_api.rs b/backend/src/api/main_api.rs index e770dd38f..72500b10d 100644 --- a/backend/src/api/main_api.rs +++ b/backend/src/api/main_api.rs @@ -24,9 +24,10 @@ use axum::Router; use log::{error, info}; use std::io::ErrorKind; use std::path::PathBuf; -use std::sync::Arc; +use std::sync::{Arc}; use tokio_util::sync::CancellationToken; use tower_governor::key_extractor::SmartIpKeyExtractor; +use crate::api::model::event_manager::EventManager; fn get_web_dir_path(web_ui_enabled: bool, web_root: &str) -> Result { let web_dir = web_root.to_string(); @@ -56,8 +57,11 @@ fn create_shared_data(app_config: &Arc, forced_targets: &Arc, forced_targets: &Arc(cfg: &ConfigInput, get_connection: Option, connection_change_sender: ConnectionChangeSender) -> Self + fn new<'a, F>(cfg: &ConfigInput, get_connection: Option, connection_change_sender: ProviderConnectionChangeSender) -> Self where F: Fn(&str) -> Option<&'a ProviderConfigConnection>, { @@ -236,7 +235,7 @@ struct MultiProviderLineup { } impl MultiProviderLineup { - pub fn new<'a, F>(input: &ConfigInput, get_connection: Option, connection_change_sender: &ConnectionChangeSender) -> Self + pub fn new<'a, F>(input: &ConfigInput, get_connection: Option, connection_change_sender: &ProviderConnectionChangeSender) -> Self where F: Fn(&str) -> Option<&'a ProviderConfigConnection> + Copy, { @@ -458,11 +457,11 @@ struct ProviderLineupManager { grace_period_timeout_secs: AtomicU64, inputs: Arc>>>, providers: Arc>>, - connection_change_tx: ConnectionChangeSender, + connection_change_tx: ProviderConnectionChangeSender, } impl ProviderLineupManager { - pub fn new(inputs: Vec>, grace_period_millis: u64, grace_period_timeout_secs: u64, connection_change_tx: ConnectionChangeSender) -> Self { + pub fn new(inputs: Vec>, grace_period_millis: u64, grace_period_timeout_secs: u64, connection_change_tx: ProviderConnectionChangeSender) -> Self { let lineups = inputs.iter().map(|i| Self::create_lineup(i, None, connection_change_tx.clone())).collect(); Self { grace_period_millis: AtomicU64::new(grace_period_millis), @@ -473,11 +472,7 @@ impl ProviderLineupManager { } } - pub fn get_active_provider_change_channel(&self) -> Receiver<(String, usize)> { - self.connection_change_tx.subscribe() - } - - fn create_lineup(input: &ConfigInput, provider_connections: Option<&HashMap<&str, ProviderConfigConnection>>, connection_change_sender: ConnectionChangeSender) -> ProviderLineup { + fn create_lineup(input: &ConfigInput, provider_connections: Option<&HashMap<&str, ProviderConfigConnection>>, connection_change_sender: ProviderConnectionChangeSender) -> ProviderLineup { let get_connections = provider_connections.map(|c| |name: &str| c.get(name)); if input.aliases.as_ref().is_some_and(|a| !a.is_empty()) { @@ -732,13 +727,12 @@ pub struct ActiveProviderManager { } impl ActiveProviderManager { - pub fn new(cfg: &AppConfig) -> Self { + pub fn new(cfg: &AppConfig, connection_change_sender: ProviderConnectionChangeSender) -> Self { let (grace_period_millis, grace_period_timeout_secs) = Self::get_grace_options(cfg); let inputs = Self::get_config_inputs(cfg); - let (connection_change_tx, _) = tokio::sync::broadcast::channel(10); Self { - providers: ProviderLineupManager::new(inputs, grace_period_millis, grace_period_timeout_secs, connection_change_tx), + providers: ProviderLineupManager::new(inputs, grace_period_millis, grace_period_timeout_secs, connection_change_sender), connections: DashMap::new(), } } @@ -804,10 +798,6 @@ impl ActiveProviderManager { guard.release(); } } - - pub fn get_active_provider_change_channel(&self) -> tokio::sync::broadcast::Receiver<(String, usize)> { - self.providers.get_active_provider_change_channel() - } } #[cfg(test)] diff --git a/backend/src/api/model/active_user_manager.rs b/backend/src/api/model/active_user_manager.rs index a9aabacb0..43bf90191 100644 --- a/backend/src/api/model/active_user_manager.rs +++ b/backend/src/api/model/active_user_manager.rs @@ -10,6 +10,9 @@ use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::Arc; use crate::api::model::active_provider_manager::ActiveProviderManager; +type ActiveUserConnectionChangeSender = tokio::sync::mpsc::Sender<(usize, usize)>; +pub type ActiveUserConnectionChangeReceiver = tokio::sync::mpsc::Receiver<(usize, usize)>; + macro_rules! active_user_manager_shared_impl { () => { #[inline] @@ -21,7 +24,7 @@ macro_rules! active_user_manager_shared_impl { let user = Arc::clone(&self.user); let user_connection_count = Self::get_active_connections(&user); let user_count = user.len(); - let _= self.active_user_change_tx.send((user_count, user_connection_count)); + let _= self.connection_change_tx.try_send((user_count, user_connection_count)); if self.is_log_user_enabled() { info!("Active Users: {user_count}, Active User Connections: {user_connection_count}"); @@ -73,7 +76,7 @@ struct ConnectionGuardUserManager { user_by_addr: Arc>, shared_stream_manager: Arc, provider_manager: Arc, - active_user_change_tx: tokio::sync::broadcast::Sender<(usize, usize)>, + connection_change_tx: ActiveUserConnectionChangeSender, } impl ConnectionGuardUserManager { @@ -148,15 +151,14 @@ pub struct ActiveUserManager { close_signal_tx: tokio::sync::broadcast::Sender, shared_stream_manager: Arc, provider_manager: Arc, - active_user_change_tx: tokio::sync::broadcast::Sender<(usize, usize)>, + connection_change_tx: ActiveUserConnectionChangeSender, } impl ActiveUserManager { - pub fn new(config: &Config, shared_stream_manager: &Arc, provider_manager: &Arc) -> Self { + pub fn new(config: &Config, shared_stream_manager: &Arc, provider_manager: &Arc, connection_change_tx: ActiveUserConnectionChangeSender) -> 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) = get_grace_options(config); let (close_signal_tx, _) = tokio::sync::broadcast::channel(10); - let (active_user_change_tx, _) = tokio::sync::broadcast::channel(10); Self { grace_period_millis: AtomicU64::new(grace_period_millis), grace_period_timeout_secs: AtomicU64::new(grace_period_timeout_secs), @@ -167,7 +169,7 @@ impl ActiveUserManager { close_signal_tx, shared_stream_manager: Arc::clone(shared_stream_manager), provider_manager: Arc::clone(provider_manager), - active_user_change_tx, + connection_change_tx, } } @@ -188,7 +190,7 @@ impl ActiveUserManager { user_by_addr: Arc::clone(&self.user_by_addr), shared_stream_manager: Arc::clone(&self.shared_stream_manager), provider_manager: Arc::clone(&self.provider_manager), - active_user_change_tx: self.active_user_change_tx.clone(), + connection_change_tx: self.connection_change_tx.clone(), } } @@ -362,10 +364,6 @@ impl ActiveUserManager { self.close_signal_tx.subscribe() } - pub fn get_active_user_change_channel(&self) -> tokio::sync::broadcast::Receiver<(usize, usize)> { - self.active_user_change_tx.subscribe() - } - pub fn get_user_session(&self, username: &str, token: &str) -> Option { self.update_user_session(username, token) } diff --git a/backend/src/api/model/app_state.rs b/backend/src/api/model/app_state.rs index 77318aaeb..62a5907d1 100644 --- a/backend/src/api/model/app_state.rs +++ b/backend/src/api/model/app_state.rs @@ -17,6 +17,7 @@ use std::time::Duration; use tokio::sync::Mutex; use tokio_util::sync::CancellationToken; use crate::api::config_watch::exec_config_watch; +use crate::api::model::event_manager::EventManager; macro_rules! cancel_service { ($field: ident, $changes:expr, $cancel_tokens:expr) => { @@ -168,6 +169,7 @@ pub struct AppState { pub shared_stream_manager: Arc, pub active_users: Arc, pub active_provider: Arc, + pub event_manager: Arc, pub cancel_tokens: Arc>, } diff --git a/backend/src/api/model/event_manager.rs b/backend/src/api/model/event_manager.rs new file mode 100644 index 000000000..0df0d5f64 --- /dev/null +++ b/backend/src/api/model/event_manager.rs @@ -0,0 +1,62 @@ +use log::error; +use tokio::task; +use crate::api::model::active_user_manager::{ActiveUserConnectionChangeReceiver}; +use crate::api::model::provider_config::{ProviderConnectionChangeReceiver}; + +#[derive(Clone, PartialEq)] +pub enum EventMessage { + ActiveUserChange(usize, usize), // user_count, connection count + ActiveProviderChange(String, usize) // provider name, connections +} + +pub struct EventManager { + channel_tx: tokio::sync::broadcast::Sender, + #[allow(dead_code)] + channel_rx: tokio::sync::broadcast::Receiver, +} + +impl EventManager { + pub fn new(mut active_user_change_rx: ActiveUserConnectionChangeReceiver, + mut provider_change_rx: ProviderConnectionChangeReceiver, + ) -> Self { + let (channel_tx, channel_rx) = tokio::sync::broadcast::channel(10); + + let channel_tx_clone = channel_tx.clone(); + task::spawn(async move { + loop { + tokio::select! { + Some((user_count, connection_count)) = active_user_change_rx.recv() => { + if let Err(e) = channel_tx_clone.send(EventMessage::ActiveUserChange(user_count, connection_count)) { + error!("Failed to send active user change event: {e}"); + break; + } + } + + Some((provider, connection_count)) = provider_change_rx.recv() => { + if let Err(e) = channel_tx_clone.send(EventMessage::ActiveProviderChange(provider, connection_count)) { + error!("Failed to send active provider change event: {e}"); + break; + } + } + } + } + }); + + Self { + channel_tx, + channel_rx, + } + } + + pub fn get_event_channel(&self) -> tokio::sync::broadcast::Receiver { + self.channel_tx.subscribe() + } + + pub fn send_event(&self, event: EventMessage) { + if let Err(err) = self.channel_tx.send(event) { + error!("Failed to send event: {err}"); + } + } + +} + diff --git a/backend/src/api/model/mod.rs b/backend/src/api/model/mod.rs index 14f4fc5d8..1aebde331 100644 --- a/backend/src/api/model/mod.rs +++ b/backend/src/api/model/mod.rs @@ -9,4 +9,5 @@ pub(crate) mod streams; pub(in crate::api) mod active_user_manager; pub(in crate::api) mod active_provider_manager; pub(in crate::api) mod stream; -pub(in crate::api) mod provider_config; \ No newline at end of file +pub(in crate::api) mod provider_config; +pub(in crate::api) mod event_manager; \ No newline at end of file diff --git a/backend/src/api/model/provider_config.rs b/backend/src/api/model/provider_config.rs index 4871b6277..3791cf868 100644 --- a/backend/src/api/model/provider_config.rs +++ b/backend/src/api/model/provider_config.rs @@ -3,11 +3,12 @@ use crate::model::{ConfigInput, ConfigInputAlias, InputUserInfo}; use jsonwebtoken::get_current_timestamp; use log::{debug}; use std::ops::Deref; -use std::sync::Arc; +use std::sync::{Arc}; use tokio::sync::RwLock; use shared::model::InputType; -pub type ConnectionChangeSender = tokio::sync::broadcast::Sender<(String, usize)>; +pub type ProviderConnectionChangeSender = tokio::sync::mpsc::Sender<(String, usize)>; +pub type ProviderConnectionChangeReceiver = tokio::sync::mpsc::Receiver<(String, usize)>; #[derive(Debug, Clone, Copy)] pub enum ProviderConfigAllocation { @@ -41,7 +42,7 @@ pub struct ProviderConfig { max_connections: usize, priority: i16, connection: RwLock, - connection_change_tx: tokio::sync::broadcast::Sender<(String, usize)>, + connection_change_tx: ProviderConnectionChangeSender, } impl PartialEq for ProviderConfig { @@ -61,16 +62,16 @@ impl PartialEq for ProviderConfig { macro_rules! modify_connections { ($self:ident, $guard:ident, +1) => {{ $guard.current_connections += 1; - $self.notify_connection_change($guard.current_connections); + $self.notify_connection_change($guard.current_connections).await; }}; ($self:ident, $guard:ident, -1) => {{ $guard.current_connections -= 1; - $self.notify_connection_change($guard.current_connections); + $self.notify_connection_change($guard.current_connections).await; }}; } impl ProviderConfig { - pub fn new<'a, F>(cfg: &ConfigInput, get_connection: Option, connection_change_tx: tokio::sync::broadcast::Sender<(String, usize)>) -> Self + pub fn new<'a, F>(cfg: &ConfigInput, get_connection: Option, connection_change_tx: ProviderConnectionChangeSender) -> Self where F: Fn(&str) -> Option<&'a ProviderConfigConnection>, { @@ -88,7 +89,7 @@ impl ProviderConfig { } } - pub fn new_alias<'a, F>(cfg: &ConfigInput, alias: &ConfigInputAlias, get_connection: Option, connection_change_tx: ConnectionChangeSender) -> Self + pub fn new_alias<'a, F>(cfg: &ConfigInput, alias: &ConfigInputAlias, get_connection: Option, connection_change_tx: ProviderConnectionChangeSender) -> Self where F: Fn(&str) -> Option<&'a ProviderConfigConnection>, { @@ -110,8 +111,8 @@ impl ProviderConfig { InputUserInfo::new(self.input_type, self.username.as_deref(), self.password.as_deref(), &self.url) } - fn notify_connection_change(&self, new_connections: usize) { - let _ = self.connection_change_tx.send((self.name.clone(), new_connections)); + async fn notify_connection_change(&self, new_connections: usize) { + let _ = self.connection_change_tx.send((self.name.clone(), new_connections)).await; } #[inline] diff --git a/backend/src/processing/parser/m3u.rs b/backend/src/processing/parser/m3u.rs index e05c05349..d66790ba1 100644 --- a/backend/src/processing/parser/m3u.rs +++ b/backend/src/processing/parser/m3u.rs @@ -307,4 +307,18 @@ mod test { assert_eq!(pli.chno, "7"); assert_eq!(pli.group, "Sydney"); } + + #[test] + fn test_process_header_xui_id() { + let input: &str = "hello"; + let video_suffixes = Vec::new(); + let url = "http://hello.de/hello.ts"; + let line = r#"#EXTINF:-1 tvg-id="abc-seven" xui-id="provider-123" group-title="Sydney", Seven"#; + + let pli = process_header(input, &video_suffixes, line, url); + assert_eq!(pli.title, "Seven"); + assert_eq!(pli.id, "provider-123"); // Should use xui-id + assert_eq!(pli.epg_channel_id, Some("abc-seven".to_string())); // Should preserve original tvg-id + assert_eq!(pli.group, "Sydney"); + } } \ No newline at end of file diff --git a/backend/src/processing/processor/playlist.rs b/backend/src/processing/processor/playlist.rs index 3c30760a7..0ba79152c 100644 --- a/backend/src/processing/processor/playlist.rs +++ b/backend/src/processing/processor/playlist.rs @@ -286,7 +286,7 @@ async fn process_source(client: Arc, cfg: Arc, sourc if source_downloaded { if source_playlists.is_empty() { debug!("Source at index {source_idx} is empty"); - errors.push(notify_err!(format!("Source at {source_idx} is empty"))); + errors.push(notify_err!(format!("Source at index {source_idx} is empty: {}", source.inputs.iter().map(|i| i.name.as_str()).collect::>().join(", ")))); } else { debug_if_enabled!("Source has {} groups", source_playlists.iter().map(|fpl| fpl.playlistgroups.len()).sum::()); for target in &source.targets { diff --git a/webui/src/app/components/dashboard/stats_view.rs b/webui/src/app/components/dashboard/stats_view.rs index 4f1464b45..4f12c6bb9 100644 --- a/webui/src/app/components/dashboard/stats_view.rs +++ b/webui/src/app/components/dashboard/stats_view.rs @@ -12,7 +12,7 @@ pub fn StatsView() -> Html { };