From 7c7c467f64e47dcd1defa9ebff2ef6c301b93c02 Mon Sep 17 00:00:00 2001 From: euzu Date: Sat, 25 Oct 2025 20:40:58 +0200 Subject: [PATCH 1/5] Stream table new columns --- Cargo.lock | 1 + backend/src/api/api_utils.rs | 45 +++++---- backend/src/api/model/active_user_manager.rs | 4 +- .../api/model/streams/active_client_stream.rs | 19 ++-- backend/src/repository/bplustree.rs | 98 +++++++++++++++++++ backend/src/repository/xtream_repository.rs | 2 + backend/src/utils/geoip.rs | 84 ++++++++++++++++ backend/src/utils/mod.rs | 1 + frontend/public/assets/i18n/en.json | 6 +- .../components/dashboard/_streams_view.scss | 4 + .../app/components/dashboard/streams_table.rs | 66 +++++++++++-- shared/Cargo.toml | 1 + shared/src/model/config/web_ui.rs | 1 + shared/src/model/stream_info.rs | 11 ++- shared/src/utils/string_utils.rs | 10 ++ shared/src/utils/time_utils.rs | 15 +-- 16 files changed, 319 insertions(+), 49 deletions(-) create mode 100644 backend/src/utils/geoip.rs diff --git a/Cargo.lock b/Cargo.lock index 054477ff3..8a7890b16 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3868,6 +3868,7 @@ dependencies = [ "enum-iterator", "fastrand", "indexmap", + "js-sys", "log", "path-clean", "pest", diff --git a/backend/src/api/api_utils.rs b/backend/src/api/api_utils.rs index c9633a2e2..4ef2cf2da 100644 --- a/backend/src/api/api_utils.rs +++ b/backend/src/api/api_utils.rs @@ -329,7 +329,7 @@ enum ProviderStreamState { pub struct StreamDetails { pub stream: Option, stream_info: ProviderStreamInfo, - pub input_name: Option, + pub provider_name: Option, pub grace_period_millis: u64, pub reconnect_flag: Option>, pub provider_connection_guard: Option>, @@ -340,7 +340,7 @@ impl StreamDetails { Self { stream: Some(stream), stream_info: None, - input_name: None, + provider_name: None, grace_period_millis: default_grace_period_millis(), reconnect_flag: None, provider_connection_guard: None, @@ -484,6 +484,11 @@ async fn create_stream_response_details( &streaming_strategy.provider_stream_state, config_grace_period_millis, ); + let provider_name = streaming_strategy + .provider_connection_guard + .as_ref() + .and_then(|guard| guard.get_provider_name()); + match streaming_strategy.provider_stream_state { // custom stream means we display our own stream like connection exhausted, channel-unavailable... ProviderStreamState::Custom(provider_stream) => { @@ -491,7 +496,7 @@ async fn create_stream_response_details( StreamDetails { stream, stream_info, - input_name: None, + provider_name: provider_name.clone(), grace_period_millis, reconnect_flag: None, provider_connection_guard: streaming_strategy.provider_connection_guard.clone(), @@ -525,14 +530,6 @@ async fn create_stream_response_details( ((None, None), None) }; - // if we have no stream, we should release the provider - if stream.is_none() { - if let Some(guard) = streaming_strategy.provider_connection_guard.take() { - drop(guard); - } - error!("Cant open stream {}", sanitize_sensitive_info(&request_url)); - } - if log_enabled!(log::Level::Debug) { if let Some((headers, status_code, response_url)) = stream_info.as_ref() { debug!( @@ -546,10 +543,18 @@ async fn create_stream_response_details( } } + // if we have no stream, we should release the provider + if stream.is_none() { + if let Some(guard) = streaming_strategy.provider_connection_guard.take() { + drop(guard); + } + error!("Cant open stream {}", sanitize_sensitive_info(&request_url)); + } + StreamDetails { stream, stream_info, - input_name: provider_name, + provider_name, grace_period_millis, reconnect_flag, provider_connection_guard: streaming_strategy.provider_connection_guard.take(), @@ -770,7 +775,7 @@ pub async fn force_provider_stream_response( .await; stream_channel.shared = share_stream; let stream = - ActiveClientStream::new(stream_details, app_state, user, connection_permission, addr, stream_channel) + ActiveClientStream::new(stream_details, app_state, user, connection_permission, addr, stream_channel, req_headers) .await; let (status_code, header_map) = @@ -834,7 +839,7 @@ pub async fn stream_response( let share_stream = is_stream_share_enabled(item_type, target); if share_stream { if let Some(value) = - shared_stream_response(app_state, stream_url, addr, user, connection_permission, stream_channel.clone()).await + shared_stream_response(app_state, stream_url, addr, user, connection_permission, stream_channel.clone(), req_headers).await { return value.into_response(); } @@ -860,10 +865,7 @@ pub async fn stream_response( .stream_info .as_ref() .map(|(h, sc, response_url)| (h.clone(), *sc, response_url.clone())); - let provider_name = stream_details - .provider_connection_guard - .as_ref() - .and_then(|guard| guard.get_provider_name()); + let provider_name = stream_details.provider_name.clone(); let provider_guard = if share_stream { stream_details.provider_connection_guard.take() @@ -872,7 +874,7 @@ pub async fn stream_response( }; stream_channel.shared = share_stream; let stream = - ActiveClientStream::new(stream_details, app_state, user, connection_permission, addr, stream_channel) + ActiveClientStream::new(stream_details, app_state, user, connection_permission, addr, stream_channel, req_headers) .await; let stream_resp = if share_stream { debug_if_enabled!( @@ -986,7 +988,8 @@ async fn shared_stream_response( addr: &str, user: &ProxyUserCredentials, connect_permission: UserConnectionPermission, - mut stream_channel: StreamChannel + mut stream_channel: StreamChannel, + req_headers: &HeaderMap, ) -> Option { if let Some(stream) = SharedStreamManager::subscribe_shared_stream(app_state, stream_url, Some(addr)).await @@ -1007,7 +1010,7 @@ async fn shared_stream_response( let stream_details = StreamDetails::from_stream(stream); stream_channel.shared = true; let stream = - ActiveClientStream::new(stream_details, app_state, user, connect_permission, addr, stream_channel) + ActiveClientStream::new(stream_details, app_state, user, connect_permission, addr, stream_channel, req_headers) .await .boxed(); let mut response = axum::response::Response::builder().status(status_code); diff --git a/backend/src/api/model/active_user_manager.rs b/backend/src/api/model/active_user_manager.rs index 45d26d478..27177e5ff 100644 --- a/backend/src/api/model/active_user_manager.rs +++ b/backend/src/api/model/active_user_manager.rs @@ -1,3 +1,4 @@ +use std::borrow::Cow; use crate::api::model::ActiveProviderManager; use crate::api::model::SharedStreamManager; use crate::model::Config; @@ -280,12 +281,13 @@ impl ActiveUserManager { Self::get_active_connections(&self.user).await } - pub async fn add_connection(&self, username: &str, max_connections: u32, addr: &str, provider: &str, stream_channel: StreamChannel) -> UserConnectionGuard { + pub async fn add_connection(&self, username: &str, max_connections: u32, addr: &str, provider: &str, stream_channel: StreamChannel, user_agent: Cow<'_, str>) -> UserConnectionGuard { let stream_info = StreamInfo::new( username, addr, provider, stream_channel, + user_agent.to_string(), ); { let mut user_map = self.user.write().await; diff --git a/backend/src/api/model/streams/active_client_stream.rs b/backend/src/api/model/streams/active_client_stream.rs index cb2613fd6..437c83952 100644 --- a/backend/src/api/model/streams/active_client_stream.rs +++ b/backend/src/api/model/streams/active_client_stream.rs @@ -15,6 +15,8 @@ use std::pin::Pin; use std::sync::atomic::AtomicU8; use std::sync::{Arc}; use std::task::{Poll}; +use axum::http::header::USER_AGENT; +use axum::http::HeaderMap; use futures::task::AtomicWaker; const INNER_STREAM: u8 = 0_u8; @@ -39,19 +41,16 @@ impl ActiveClientStream { user: &ProxyUserCredentials, connection_permission: UserConnectionPermission, addr: &str, - stream_channel: StreamChannel) -> Self { + stream_channel: StreamChannel, + req_headers: &HeaderMap) -> Self { if connection_permission == UserConnectionPermission::Exhausted { error!("Something is wrong this should not happen"); } let grant_user_grace_period = connection_permission == UserConnectionPermission::GracePeriod; let username = user.username.as_str(); - let provider_name = stream_details - .provider_connection_guard - .as_ref() - .and_then(|guard| guard.get_provider_name()) - .as_deref() - .map_or_else(String::new, ToString::to_string); - let user_connection_guard = Some(app_state.active_users.add_connection(username, user.max_connections, addr, &provider_name, stream_channel).await); + let provider_name = stream_details.provider_name.as_ref().map_or_else(String::new, ToString::to_string); + let user_agent = req_headers.get(USER_AGENT).map(|h| String::from_utf8_lossy(h.as_bytes())).unwrap_or_default(); + let user_connection_guard = Some(app_state.active_users.add_connection(username, user.max_connections, addr, &provider_name, stream_channel, user_agent).await); let cfg = &app_state.app_config; let waker = Some(Arc::new(AtomicWaker::new())); let waker_clone = waker.clone(); @@ -106,8 +105,8 @@ impl ActiveClientStream { let active_provider = Arc::clone(&app_state.active_provider); let shared_stream_manager = Arc::clone(&app_state.shared_stream_manager); - let provider_grace_check = if stream_details.has_grace_period() && stream_details.input_name.is_some() { - let provider_name = stream_details.input_name.as_deref().unwrap_or_default().to_string(); + let provider_grace_check = if stream_details.has_grace_period() && stream_details.provider_name.is_some() { + let provider_name = stream_details.provider_name.as_ref().map_or_else(String::new, ToString::to_string); Some(provider_name) } else { None diff --git a/backend/src/repository/bplustree.rs b/backend/src/repository/bplustree.rs index 06a845e1e..c9bdeaff7 100644 --- a/backend/src/repository/bplustree.rs +++ b/backend/src/repository/bplustree.rs @@ -58,6 +58,50 @@ where } +fn query_tree_le(file: &mut R, key: &K) -> Option +where + K: Ord + Serialize + for<'de> Deserialize<'de> + Clone, + V: Serialize + for<'de> Deserialize<'de> + Clone, +{ + let mut offset = 0; + let mut buffer = vec![0u8; BLOCK_SIZE]; + loop { + match BPlusTreeNode::::deserialize_from_block(file, &mut buffer, offset, false) { + Ok((node, pointers)) => { + if node.is_leaf { + let idx = get_entry_index_upper_bound::(&node.keys, key); + if idx == 0 { + return None; + } else { + return node.values.get(idx - 1).cloned(); + } + } + let child_idx = get_entry_index_upper_bound::(&node.keys, key); + if let Some(child_offsets) = pointers { + if let Some(child_offset) = child_offsets.get(child_idx) { + offset = *child_offset; + } else { + // defensive: if out of bounds try last pointer + if let Some(last) = child_offsets.last() { + offset = *last; + } else { + return None; + } + } + } else { + return None; + } + } + Err(err) => { + error!("Failed to read id tree from file {err}"); + return None; + } + } + } +} + + + #[derive(Serialize, Deserialize, Debug, Clone)] struct BPlusTreeNode { keys: Vec, @@ -193,6 +237,32 @@ where } } + /// Find the largest key <= `key` in this subtree. + /// Returns a reference to (key, value) if found (only valid for leaf entries). + fn find_le(&self, key: &K) -> Option<(&K, &V)> { + if self.is_leaf { + // find index of first key > key, then step one back + let idx = self.get_entry_index_upper_bound(key); + if idx == 0 { + None + } else { + let i = idx - 1; + // safe: leaf guarantees values.len() == keys.len() + Some((&self.keys[i], &self.values[i])) + } + } else { + // descend into the appropriate child (child index = upper_bound) + let child_idx = self.get_entry_index_upper_bound(key); + // child_idx can be equal to children.len() if key > all keys; children.get handles that + if let Some(child) = self.children.get(child_idx) { + child.find_le(key) + } else { + // fallback: if child_idx is out of bounds, try last child (defensive) + self.children.last().and_then(|c| c.find_le(key)) + } + } + } + pub fn traverse(&self, visit: &mut F) where F: FnMut(&Vec, &Vec), @@ -506,6 +576,15 @@ where Ok(Self::new_with_root(root)) } + /// Find the largest key <= `key` in the in-memory tree and return references to (key, value). + pub fn find_le(&self, key: &K) -> Option<(&K, &V)> { + // empty tree + if self.root.keys.is_empty() && self.root.is_leaf && self.root.values.is_empty() { + return None; + } + self.root.find_le(key) + } + pub fn traverse(&self, mut visit: F) where F: FnMut(&Vec, &Vec), @@ -604,6 +683,19 @@ where query_tree(&mut self.file, key) } + /// On-disk: find largest key <= `key` and return owned V (cloned/deserialized) + pub fn query_le(&mut self, key: &K) -> Option { + // use the same buffer/reader pattern as query() + // we need a mutable reader over the inner BufReader + let file = &mut self.file; + // Seek to start to be safe + if file.seek(SeekFrom::Start(0)).is_err() { + // if seek fails, still try to query β€” but bail out with None + return None; + } + query_tree_le(file, key) + } + // pub fn traverse(&mut self, mut visit: F) // where // F: FnMut(&Vec, &Vec), @@ -681,6 +773,12 @@ where } } } + + /// On-disk update helper: find largest key <= `key`. + pub fn query_le(&mut self, key: &K) -> Option { + let mut reader = utils::file_reader(&mut self.file); + query_tree_le(&mut reader, key) + } } pub struct BPlusTreeIterator<'a, K, V> { diff --git a/backend/src/repository/xtream_repository.rs b/backend/src/repository/xtream_repository.rs index c451445e2..878ae1e7f 100644 --- a/backend/src/repository/xtream_repository.rs +++ b/backend/src/repository/xtream_repository.rs @@ -440,6 +440,8 @@ pub async fn xtream_get_item_for_stream_id( PlaylistItemType::Series => { if let Ok(mut item) = xtream_read_series_item_for_stream_id(app_config, mapping.parent_virtual_id, &storage_path) { item.provider_id = mapping.provider_id; + + Ok(item) } else { xtream_read_item_for_stream_id(app_config, virtual_id, &storage_path, XtreamCluster::Series) diff --git a/backend/src/utils/geoip.rs b/backend/src/utils/geoip.rs new file mode 100644 index 000000000..f3ea4f25e --- /dev/null +++ b/backend/src/utils/geoip.rs @@ -0,0 +1,84 @@ +use std::io; +use std::io::{BufRead}; +use std::net::Ipv4Addr; +use std::path::Path; +use serde::{Serialize, Deserialize}; +use crate::repository::bplustree::BPlusTree; + +fn ipv4_to_u32(ip: &str) -> Option { + ip.parse::().ok().map(|a| u32::from(a)) +} + +#[derive(Serialize, Deserialize)] +pub struct GeoIp { + tree: BPlusTree, +} + +impl GeoIp { + + pub fn load(path: &Path) -> io::Result { + let tree = BPlusTree::load(path)?; + Ok(Self { tree }) + } + + pub fn new() -> Self { + Self { tree: BPlusTree::new() } + } + + pub fn import_ipv4_from_csv(&mut self, mut reader: impl BufRead, db_path: &Path) -> std::io::Result { + let mut buf = String::new(); + + while reader.read_line(&mut buf)? > 0 { + let line = buf.trim(); + if line.is_empty() || line.starts_with('#') { continue; } + + let parts: Vec<&str> = line.split(',').collect(); + if parts.len() != 3 { continue; } + + if let (Some(start), Some(end)) = (ipv4_to_u32(parts[0]), ipv4_to_u32(parts[1])) { + let cc = parts[2].trim().to_string(); + self.tree.insert(start, (end, cc)); + } + buf.clear(); + } + self.tree.store(db_path) + } + + pub fn lookup(&self, ip_str: &str) -> Option { + let ip = ipv4_to_u32(ip_str)?; + if let Some((_, (end, cc))) = self.tree.find_le(&ip) { + if ip <= *end { + return Some(cc.to_string()); + } + } + None + } +} + +#[cfg(test)] +mod test { + //https://github.com/datasets/geoip2-ipv4/blob/main/data/geoip2-ipv4.csv + + use std::fs::File; + use std::io::BufReader; + use std::path::PathBuf; + use crate::utils::geoip::GeoIp; + + #[test] + pub fn test_csv() { + let db_file = PathBuf::from("/projects/m3u-test/asn-country-ipv4.db"); + let source = PathBuf::from("/projects/m3u-test/asn-country-ipv4.csv"); + let file = File::open(source).expect("Could not open csv file"); + let reader = BufReader::new(file); + let mut geo_ip = GeoIp::new(); + let _ = geo_ip.import_ipv4_from_csv(reader, &db_file).expect("Could not import csv"); + + let geo_ip = GeoIp::load(&db_file).expect("Failed to load geoip db"); + if let Some(cc) = geo_ip.lookup("72.13.24.23") { + assert_eq!(cc, "US"); + } else { + assert!(false); + } + + } +} \ No newline at end of file diff --git a/backend/src/utils/mod.rs b/backend/src/utils/mod.rs index a99877bb3..d0477a11d 100644 --- a/backend/src/utils/mod.rs +++ b/backend/src/utils/mod.rs @@ -9,6 +9,7 @@ mod trakt; mod json_utils; mod bincode_utils; mod telegram; +mod geoip; pub use self::bincode_utils::*; pub use self::logging::*; diff --git a/frontend/public/assets/i18n/en.json b/frontend/public/assets/i18n/en.json index 6c639ab36..339dbb62f 100644 --- a/frontend/public/assets/i18n/en.json +++ b/frontend/public/assets/i18n/en.json @@ -290,7 +290,11 @@ "GROUP": "Group", "CLIENT_IP": "Client IP", "STREAM_ID": "Stream Id", - "SHARED": "Shared" + "SHARED": "Shared", + "CLUSTER": "Type", + "USER_AGENT": "Player", + "FORMAT": "Format", + "DURATION": "Duration" }, "TITLE": { "USER_BOUQUET_EDITOR": "User group editor" diff --git a/frontend/scss/app/components/dashboard/_streams_view.scss b/frontend/scss/app/components/dashboard/_streams_view.scss index 8ad4295eb..d8439096c 100644 --- a/frontend/scss/app/components/dashboard/_streams_view.scss +++ b/frontend/scss/app/components/dashboard/_streams_view.scss @@ -18,4 +18,8 @@ gap: var(--gap-larger); overflow: auto; } +} + +.tp__stream-table__duration { + letter-spacing: 2px; } \ No newline at end of file diff --git a/frontend/src/app/components/dashboard/streams_table.rs b/frontend/src/app/components/dashboard/streams_table.rs index 53b87cd65..0f87136e9 100644 --- a/frontend/src/app/components/dashboard/streams_table.rs +++ b/frontend/src/app/components/dashboard/streams_table.rs @@ -1,3 +1,4 @@ +use std::borrow::Cow; use crate::app::components::menu_item::MenuItem; use crate::app::components::popup_menu::PopupMenu; use crate::app::components::{AppIcon, Table, TableDefinition, ToggleSwitch}; @@ -7,20 +8,60 @@ use shared::model::{SortOrder, StreamInfo}; use std::fmt::Display; use std::rc::Rc; use std::str::FromStr; +use gloo_timers::callback::Interval; +use gloo_utils::window; +use log::debug; +use wasm_bindgen::JsCast; +use web_sys::Element; use yew::prelude::*; use yew_i18n::use_translation; +use shared::utils::current_time_secs; -const HEADERS: [&str; 8] = [ +const HEADERS: [&str; 11] = [ "LABEL.EMPTY", "LABEL.USERNAME", "LABEL.STREAM_ID", + "LABEL.CLUSTER", "LABEL.CHANNEL", "LABEL.GROUP", "LABEL.CLIENT_IP", "LABEL.PROVIDER", - "LABEL.SHARED" + "LABEL.SHARED", + "LABEL.USER_AGENT", + "LABEL.DURATION" ]; +fn strip_port<'a>(input: &'a str) -> Cow<'a, str> { + if let Some(pos) = input.find(':') { + Cow::Owned(input[..pos].to_string()) + } else { + Cow::Borrowed(input) + } +} + +pub fn format_duration(seconds: u64) -> String { + let hours = seconds / 3600; + let minutes = (seconds % 3600) / 60; + let seconds = seconds % 60; + format!("{hours:02}:{minutes:02}:{seconds:02}") +} + +fn update_timestamps() { + let window = window(); + let document = window.document().unwrap(); + let spans = document.query_selector_all("span[data-ts]").unwrap(); + for i in 0..spans.length() { + if let Some(node) = spans.item(i) { + let el: Element = node.dyn_into().unwrap(); + if let Some(ts_str) = el.get_attribute("data-ts") { + if let Ok(ts) = ts_str.parse::() { + el.set_inner_html(&format_duration(current_time_secs() - ts)); + } + } + } + } +} + #[derive(Properties, PartialEq, Clone)] pub struct StreamsTableProps { pub streams: Option>>, @@ -34,6 +75,14 @@ pub fn StreamsTable(props: &StreamsTableProps) -> Html { let popup_is_open = use_state(|| false); let selected_dto = use_state(|| None::>); + + use_effect_with((), move |_| { + Interval::new(1000, || { + update_timestamps(); + }).forget(); + }); + + let handle_popup_close = { let set_is_open = popup_is_open.clone(); Callback::from(move |()| { @@ -91,11 +140,14 @@ pub fn StreamsTable(props: &StreamsTableProps) -> Html { { dto.channel.provider_id.to_string() } {")"} }, - 3 => html! {dto.channel.title.as_str()}, - 4 => html! {dto.channel.group.as_str()}, - 5 => html! {dto.addr.as_str()}, - 6 => html! {dto.provider.as_str()}, - 7 => html! { }, + 3 => html! {dto.channel.cluster}, + 4 => html! {dto.channel.title.as_str()}, + 5 => html! {dto.channel.group.as_str()}, + 6 => html! { strip_port(&dto.addr)}, + 7 => html! {dto.provider.as_str()}, + 8 => html! { }, + 9 => html! { dto.user_agent.as_str() }, + 10 => html! { {format_duration(dto.ts)} }, _ => html! {""}, } }) diff --git a/shared/Cargo.toml b/shared/Cargo.toml index bc1e091ec..bb30e6594 100644 --- a/shared/Cargo.toml +++ b/shared/Cargo.toml @@ -22,3 +22,4 @@ zeroize = "1" chrono = "0.4.42" bytes = "1" ciborium = "0.2.2" +js-sys = "0.3.81" diff --git a/shared/src/model/config/web_ui.rs b/shared/src/model/config/web_ui.rs index c0aac7aeb..c16f21b08 100644 --- a/shared/src/model/config/web_ui.rs +++ b/shared/src/model/config/web_ui.rs @@ -3,6 +3,7 @@ use crate::model::WebAuthConfigDto; use crate::utils::{default_as_true, is_blank_optional_string}; const RESERVED_PATHS: &[&str] = &[ + "cvs", "live", "movie", "series", diff --git a/shared/src/model/stream_info.rs b/shared/src/model/stream_info.rs index 0edd1d460..2b0ed889d 100644 --- a/shared/src/model/stream_info.rs +++ b/shared/src/model/stream_info.rs @@ -1,5 +1,6 @@ use serde::{Deserialize, Serialize}; use crate::model::{M3uPlaylistItem, PlaylistEntry, PlaylistItemType, XtreamCluster, XtreamPlaylistItem}; +use crate::utils::{current_time_secs, StringExt}; #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] pub struct StreamChannel { @@ -20,7 +21,7 @@ impl XtreamPlaylistItem { item_type: self.item_type, cluster: self.xtream_cluster, group: self.group.clone(), - title: self.title.clone(), + title: String::longest(self.title.as_str(), self.name.as_str()).to_string(), url: self.url.clone(), shared: false, } @@ -35,7 +36,7 @@ impl M3uPlaylistItem { item_type: self.item_type, cluster: XtreamCluster::try_from(self.item_type).unwrap_or(XtreamCluster::Live), group: self.group.clone(), - title: self.title.clone(), + title: String::longest(self.title.as_str(), self.name.as_str()).to_string(), url: self.url.clone(), shared: false, } @@ -48,15 +49,19 @@ pub struct StreamInfo { pub channel: StreamChannel, pub provider: String, pub addr: String, + pub user_agent: String, + pub ts: u64, } impl StreamInfo { - pub fn new(username: &str, addr: &str, provider: &str, stream_channel: StreamChannel) -> Self { + pub fn new(username: &str, addr: &str, provider: &str, stream_channel: StreamChannel, user_agent: String) -> Self { Self { username: username.to_string(), channel: stream_channel, provider: provider.to_string(), addr: addr.to_string(), + user_agent, + ts: current_time_secs(), } } } \ No newline at end of file diff --git a/shared/src/utils/string_utils.rs b/shared/src/utils/string_utils.rs index 4c6f333d4..67a957a80 100644 --- a/shared/src/utils/string_utils.rs +++ b/shared/src/utils/string_utils.rs @@ -127,6 +127,16 @@ pub fn humanize_snake_case(s: &str) -> String { result } +pub trait StringExt { + fn longest<'a>(a: &'a str, b: &'a str) -> &'a str; +} + +impl StringExt for String { + fn longest<'a>(a: &'a str, b: &'a str) -> &'a str { + if a.len() >= b.len() { a } else { b } + } +} + #[cfg(test)] mod test { use std::collections::HashSet; diff --git a/shared/src/utils/time_utils.rs b/shared/src/utils/time_utils.rs index 0fb480747..87267eb5a 100644 --- a/shared/src/utils/time_utils.rs +++ b/shared/src/utils/time_utils.rs @@ -1,9 +1,12 @@ -use std::time::{SystemTime, UNIX_EPOCH}; -use chrono::{DateTime}; - +#[cfg(target_arch = "wasm32")] pub fn current_time_secs() -> u64 { - SystemTime::now() - .duration_since(UNIX_EPOCH) + (js_sys::Date::now() / 1000.0) as u64 +} + +#[cfg(not(target_arch = "wasm32"))] +pub fn current_time_secs() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() .as_secs() } @@ -15,7 +18,7 @@ pub fn unix_ts_to_str(ts: i64) -> Option { } else { ts }; - DateTime::from_timestamp(normalized_ts, 0).map(|dt| dt.format("%d.%m.%Y").to_string()) + chrono::DateTime::from_timestamp(normalized_ts, 0).map(|dt| dt.format("%d.%m.%Y").to_string()) } else { None } From 93d60a210f514d6155d7ce251c15d14ebebf24a6 Mon Sep 17 00:00:00 2001 From: euzu Date: Mon, 27 Oct 2025 08:27:35 +0100 Subject: [PATCH 2/5] Stream table new colmuns --- README.md | 5 ++ backend/src/api/api_utils.rs | 16 ++-- backend/src/api/endpoints/hls_api.rs | 6 +- backend/src/api/endpoints/m3u_api.rs | 4 +- backend/src/api/endpoints/xtream_api.rs | 53 +++++-------- backend/src/api/main_api.rs | 4 +- .../src/api/model/active_provider_manager.rs | 79 ++++++++++++------- backend/src/api/model/active_user_manager.rs | 16 ++-- backend/src/api/model/provider_config.rs | 4 +- .../model/streams/shared_stream_manager.rs | 29 ++++--- backend/src/auth/fingerprint.rs | 14 ++-- backend/src/repository/bplustree.rs | 3 +- backend/src/utils/geoip.rs | 4 +- .../app/components/dashboard/streams_table.rs | 22 +++++- shared/Cargo.toml | 1 + 15 files changed, 153 insertions(+), 107 deletions(-) diff --git a/README.md b/README.md index b011b76d0..e02307a26 100644 --- a/README.md +++ b/README.md @@ -1481,6 +1481,11 @@ user: status: Active ``` +If yu use a reverse proxy in fron of Tuliprox, dont forget to forward +- `X-Real-IP` +- `X-Forwarded-For` + + Now you can do `nginx` configuration like ```config location /tuliprox { diff --git a/backend/src/api/api_utils.rs b/backend/src/api/api_utils.rs index 4ef2cf2da..1a22a50c2 100644 --- a/backend/src/api/api_utils.rs +++ b/backend/src/api/api_utils.rs @@ -469,8 +469,7 @@ async fn create_stream_response_details( connection_permission: UserConnectionPermission, force_provider: Option<&str>, ) -> StreamDetails { - let mut streaming_strategy = - resolve_streaming_strategy(app_state, stream_url, addr, input, force_provider).await; + let mut streaming_strategy = resolve_streaming_strategy(app_state, stream_url, addr, input, force_provider).await; let config_grace_period_millis = app_state .app_config .config @@ -489,6 +488,8 @@ async fn create_stream_response_details( .as_ref() .and_then(|guard| guard.get_provider_name()); + debug!("πŸ„πŸ„πŸ„ using provider {provider_name:?}"); + match streaming_strategy.provider_stream_state { // custom stream means we display our own stream like connection exhausted, channel-unavailable... ProviderStreamState::Custom(provider_stream) => { @@ -886,7 +887,7 @@ pub async fn stream_response( .as_ref() .map_or_else(Vec::new, |(h, _, _)| h.clone()); - if let Some(broadcast_stream) = SharedStreamManager::register_shared_stream( + if let Some((broadcast_stream, _provider)) = SharedStreamManager::register_shared_stream( app_state, stream_url, stream, @@ -991,7 +992,7 @@ async fn shared_stream_response( mut stream_channel: StreamChannel, req_headers: &HeaderMap, ) -> Option { - if let Some(stream) = + if let Some((stream, provider)) = SharedStreamManager::subscribe_shared_stream(app_state, stream_url, Some(addr)).await { debug_if_enabled!( @@ -1007,7 +1008,8 @@ async fn shared_stream_response( headers.clone(), axum::http::StatusCode::OK, ))); - let stream_details = StreamDetails::from_stream(stream); + let mut stream_details = StreamDetails::from_stream(stream); + stream_details.provider_name = provider; stream_channel.shared = true; let stream = ActiveClientStream::new(stream_details, app_state, user, connect_permission, addr, stream_channel, req_headers) @@ -1322,4 +1324,8 @@ pub fn json_or_bin_response(accept: Option<&String>, data: &T) -> return bin_response(data).into_response(); } json_response(data).into_response() +} + +pub fn create_fingerprint(fingerprint: &str, username: &str, virtual_id: u32) -> String { + format!("{fingerprint}{username}{virtual_id}") } \ No newline at end of file diff --git a/backend/src/api/endpoints/hls_api.rs b/backend/src/api/endpoints/hls_api.rs index 5982e7b25..c670f247c 100644 --- a/backend/src/api/endpoints/hls_api.rs +++ b/backend/src/api/endpoints/hls_api.rs @@ -1,4 +1,4 @@ -use crate::api::api_utils::try_unwrap_body; +use crate::api::api_utils::{create_fingerprint, try_unwrap_body}; use crate::api::api_utils::{ force_provider_stream_response, get_stream_alternative_url, is_seek_request, }; @@ -78,7 +78,7 @@ pub(in crate::api) async fn handle_hls_stream_request( { Some(provider_cfg) => { let stream_url = get_stream_alternative_url(&url, input, &provider_cfg); - let user_session_token = format!("{fingerprint}{virtual_id}"); + let user_session_token = create_fingerprint(fingerprint, &user.username, virtual_id); let session_token = app_state.active_users.create_user_session( user, &user_session_token, @@ -218,7 +218,7 @@ async fn hls_api_stream( ) ); - let user_session_token = format!("{fingerprint}{virtual_id}"); + let user_session_token = create_fingerprint(&fingerprint, &user.username, virtual_id); let mut user_session = app_state .active_users .get_and_update_user_session(&user.username, &user_session_token).await; diff --git a/backend/src/api/endpoints/m3u_api.rs b/backend/src/api/endpoints/m3u_api.rs index f59836d28..1ebe70ffd 100644 --- a/backend/src/api/endpoints/m3u_api.rs +++ b/backend/src/api/endpoints/m3u_api.rs @@ -1,4 +1,4 @@ -use crate::api::api_utils::try_unwrap_body; +use crate::api::api_utils::{create_fingerprint, try_unwrap_body}; use crate::api::api_utils::{ force_provider_stream_response, get_user_target, get_user_target_by_credentials, is_seek_request, redirect, redirect_response, resource_response, separate_number_and_remainder, @@ -123,7 +123,7 @@ async fn m3u_api_stream( ); let cluster = XtreamCluster::try_from(pli.item_type).unwrap_or(XtreamCluster::Live); - let session_key = format!("{fingerprint}{virtual_id}"); + let session_key = create_fingerprint(fingerprint, &user.username, virtual_id); let user_session = app_state .active_users .get_and_update_user_session(&user.username, &session_key).await; diff --git a/backend/src/api/endpoints/xtream_api.rs b/backend/src/api/endpoints/xtream_api.rs index e67ef1d9f..b5cb3ed32 100644 --- a/backend/src/api/endpoints/xtream_api.rs +++ b/backend/src/api/endpoints/xtream_api.rs @@ -1,7 +1,7 @@ // https://github.com/tellytv/go.xtream-codes/blob/master/structs.go use crate::api::api_utils; -use crate::api::api_utils::try_unwrap_body; +use crate::api::api_utils::{create_fingerprint, try_unwrap_body}; use crate::api::api_utils::{ force_provider_stream_response, get_user_target, get_user_target_by_credentials, is_seek_request, redirect_response, resource_response, separate_number_and_remainder, @@ -29,7 +29,7 @@ use axum::response::IntoResponse; use bytes::Bytes; use futures::stream::{self, StreamExt}; use futures::Stream; -use log::{debug, error, warn}; +use log::{debug, error, log_enabled, warn}; use serde::{Deserialize, Serialize}; use serde_json::{Map, Value}; use shared::error::create_tuliprox_error_result; @@ -218,25 +218,22 @@ async fn xtream_player_api_stream( api_req: &UserApiRequest, stream_req: ApiStreamRequest<'_>, ) -> impl IntoResponse + Send { + + debug!("Stream Request {stream_req:?} - {req_headers:?}"); + + if log_enabled!(log::Level::Debug) { + let message = format!("Client Request headers {req_headers:?}"); + debug!("{}", sanitize_sensitive_info(&message)); + } + + let (user, target) = try_option_bad_request!( - get_user_target_by_credentials( - stream_req.username, - stream_req.password, - api_req, - app_state - ), + get_user_target_by_credentials( stream_req.username, stream_req.password, api_req, app_state), false, - format!( - "Could not find any user for xc stream {}", - stream_req.username - ) + format!("Could not find any user for xc stream {}", stream_req.username) ); if user.permission_denied(app_state) { - return create_custom_video_stream_response( - &app_state.app_config, - CustomVideoStreamType::UserAccountExpired, - ) - .into_response(); + return create_custom_video_stream_response(&app_state.app_config, CustomVideoStreamType::UserAccountExpired,).into_response(); } let target_name = &target.name; @@ -248,24 +245,14 @@ 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_not_found!( - xtream_repository::xtream_get_item_for_stream_id( - virtual_id, - app_state, - &target, - None - ).await, + xtream_repository::xtream_get_item_for_stream_id(virtual_id, app_state, &target, None).await, true, - format!("Failed to read xtream item for stream id {}", virtual_id) + 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()), + 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 - ) + format!( "Cant find input for target {target_name}, context {}, stream_id {virtual_id}", stream_req.context) ); let (cluster, item_type) = if stream_req.context == ApiStreamContext::Timeshift { @@ -274,7 +261,7 @@ async fn xtream_player_api_stream( (pli.xtream_cluster, pli.item_type) }; - let session_key = format!("{fingerprint}{virtual_id}"); + let session_key = create_fingerprint(fingerprint, &user.username, virtual_id); let user_session = app_state .active_users .get_and_update_user_session(&user.username, &session_key).await; @@ -443,7 +430,7 @@ async fn xtream_player_api_stream_with_token( ) ); - let session_key = format!("{fingerprint}{virtual_id}"); + let session_key = create_fingerprint(fingerprint, "webui", virtual_id); let is_hls_request = pli.item_type == PlaylistItemType::LiveHls || stream_ext.as_deref() == Some(HLS_EXT); diff --git a/backend/src/api/main_api.rs b/backend/src/api/main_api.rs index 6888fcade..d582b5430 100644 --- a/backend/src/api/main_api.rs +++ b/backend/src/api/main_api.rs @@ -69,9 +69,9 @@ fn create_shared_data( let config = app_config.config.load(); let cache = create_cache(&config); let shared_stream_manager = Arc::new(SharedStreamManager::new()); - let (provider_change_tx, provider_change_rx) = tokio::sync::mpsc::channel(10); + let (provider_change_tx, provider_change_rx) = tokio::sync::mpsc::unbounded_channel(); let active_provider = Arc::new(ActiveProviderManager::new(app_config, provider_change_tx)); - let (active_user_change_tx, active_user_change_rx) = tokio::sync::mpsc::channel(10); + let (active_user_change_tx, active_user_change_rx) = tokio::sync::mpsc::unbounded_channel(); let active_users = Arc::new(ActiveUserManager::new( &config, &shared_stream_manager, diff --git a/backend/src/api/model/active_provider_manager.rs b/backend/src/api/model/active_provider_manager.rs index 9c7efdd0e..4c42fca18 100644 --- a/backend/src/api/model/active_provider_manager.rs +++ b/backend/src/api/model/active_provider_manager.rs @@ -1,7 +1,7 @@ use crate::api::model::{ProviderConfig, ProviderConfigConnection, ProviderConfigWrapper, ProviderConnectionChangeCallback, ProviderConnectionChangeSender}; use crate::model::{AppConfig, ConfigInput}; use arc_swap::ArcSwap; -use log::{debug, log_enabled, trace}; +use log::{debug, error, log_enabled, trace}; use shared::utils::{default_grace_period_millis, default_grace_period_timeout_secs, display_vec, sanitize_sensitive_info}; use std::collections::HashMap; use std::fmt; @@ -10,6 +10,7 @@ use std::sync::atomic::{AtomicU64, AtomicU8, AtomicUsize, Ordering}; use std::sync::Arc; use tokio::sync::RwLock; use crate::utils::debug_if_enabled; +use tokio::sync::mpsc::{unbounded_channel, UnboundedSender}; const CONNECTION_STATE_ACTIVE: u8 = 0; const CONNECTION_STATE_SHARED: u8 = 1; @@ -17,9 +18,14 @@ const CONNECTION_STATE_RELEASED: u8 = 2; pub struct ProviderConnectionGuard { allocation: ProviderAllocation, + release_tx: UnboundedSender>, } impl ProviderConnectionGuard { + pub fn new(allocation: ProviderAllocation, release_tx: UnboundedSender>) -> Self { + Self { allocation, release_tx } + } + // for shared streams, we need to disable release // The connection should be released when all shared streams close! pub(crate) fn disable_release(&self) { @@ -31,6 +37,16 @@ impl ProviderConnectionGuard { } } } + + fn send_release(&self, config: &Arc) { + let provider_config = Arc::clone(config); + if let Err(_err) = &self.release_tx.send(Arc::clone(config)) { + tokio::spawn(async move { + provider_config.release().await; + }); + } + } + pub(crate) fn release(&self) { match &self.allocation { ProviderAllocation::Exhausted => {} @@ -38,11 +54,7 @@ impl ProviderConnectionGuard { ProviderAllocation::GracePeriod(state, config) => { // we can't release shared state if state.compare_exchange(CONNECTION_STATE_ACTIVE, CONNECTION_STATE_RELEASED, Ordering::SeqCst, Ordering::SeqCst).is_ok() { - let provider_config = Arc::clone(config); - trace!("Releasing provider connection {:?}", provider_config.name); - tokio::spawn(async move { - provider_config.release().await; - }); + self.send_release(config); } } } @@ -56,23 +68,12 @@ impl ProviderConnectionGuard { | ProviderAllocation::GracePeriod(state, config) => { if state.load(Ordering::SeqCst) < CONNECTION_STATE_RELEASED { state.store(CONNECTION_STATE_RELEASED, Ordering::SeqCst); - let provider_config = Arc::clone(config); - trace!("Forced releasing provider connection {:?}", provider_config.name); - tokio::spawn(async move { - provider_config.release().await; - }); + trace!("Forced releasing provider connection {:?}", config.name); + self.send_release(config); } } } } -} - -impl ProviderConnectionGuard { - pub fn new(allocation: ProviderAllocation) -> Self { - Self { - allocation, - } - } pub fn get_provider_name(&self) -> Option { match self.allocation { @@ -533,10 +534,10 @@ impl ProviderLineupManager { let cfg_name = cfg_input.name.clone(); let on_connection_change: ProviderConnectionChangeCallback = Arc::new(move |_name: &str, connections: usize| { let connection_change_sender = connection_change_sender.clone(); - let cfg_name = cfg_name.clone(); - tokio::spawn(async move { - let _ = connection_change_sender.send((cfg_name, connections)).await; - }); + let provider_cfg_name = cfg_name.clone(); + if let Err(err) = connection_change_sender.send((provider_cfg_name, connections)) { + error!("Failed to send connection change: {cfg_name}: {connections}, {err}"); + } }); let on_connection_change = Arc::new(on_connection_change); @@ -682,18 +683,18 @@ impl ProviderLineupManager { None } - async fn force_exact_acquire_connection(&self, provider_name: &str) -> Arc { + async fn force_exact_acquire_connection(&self, provider_name: &str, release_tx: UnboundedSender>) -> Arc { let providers = self.providers.load(); let allocation = match Self::get_provider_config(provider_name, &providers) { None => ProviderAllocation::Exhausted, // No Name matched, we don't have this provider Some((_lineup, config)) => config.force_allocate().await, }; - Arc::new(ProviderConnectionGuard::new(allocation)) + Arc::new(ProviderConnectionGuard::new(allocation, release_tx)) } // Returns the next available provider connection - async fn acquire_connection(&self, input_name: &str) -> Arc { + async fn acquire_connection(&self, input_name: &str, release_tx: UnboundedSender>) -> Arc { let providers = self.providers.load(); let allocation = match Self::get_provider_config(input_name, &providers) { None => ProviderAllocation::Exhausted, // No Name matched, we don't have this provider @@ -711,7 +712,7 @@ impl ProviderLineupManager { } } - Arc::new(ProviderConnectionGuard::new(allocation)) + Arc::new(ProviderConnectionGuard::new(allocation, release_tx)) } // This method is used for redirects to cycle through provider @@ -780,19 +781,36 @@ impl ProviderLineupManager { pub struct ActiveProviderManager { providers: ProviderLineupManager, connections: RwLock>>, + release_tx: UnboundedSender>, } impl ActiveProviderManager { 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); + // Create the cleanup channel + let (cleanup_tx, mut cleanup_rx) = unbounded_channel::>(); + + // Spawn the async cleanup worker + tokio::spawn(async move { + while let Some(cfg) = cleanup_rx.recv().await { + debug!("🧹 Provider connection releasing {:?}", cfg.name); + cfg.release().await; + } + debug!("Provider cleanup worker terminated"); + }); Self { providers: ProviderLineupManager::new(inputs, grace_period_millis, grace_period_timeout_secs, connection_change_sender), connections: RwLock::new(HashMap::new()), + release_tx: cleanup_tx, } } + fn release_sender(&self) -> UnboundedSender> { + self.release_tx.clone() + } + fn get_config_inputs(cfg: &AppConfig) -> Vec> { cfg.sources.load().sources .iter().flat_map(|s| s.inputs.iter().map(Arc::clone)).collect() @@ -812,14 +830,14 @@ impl ActiveProviderManager { } pub async fn force_exact_acquire_connection(&self, provider_name: &str, addr: &str) -> Arc { - let guard = self.providers.force_exact_acquire_connection(provider_name).await; + let guard = self.providers.force_exact_acquire_connection(provider_name, self.release_sender()).await; self.register_connection(addr, &guard).await; guard } // Returns the next available provider connection pub async fn acquire_connection(&self, input_name: &str, addr: &str) -> Arc { - let guard = self.providers.acquire_connection(input_name).await; + let guard = self.providers.acquire_connection(input_name, self.release_sender()).await; self.register_connection(addr, &guard).await; guard } @@ -844,7 +862,7 @@ impl ActiveProviderManager { async fn register_connection(&self, addr: &str, guard: &Arc) { if !matches!(guard.allocation, ProviderAllocation::Exhausted) { - trace!("Added provider connection {:?}", guard.get_provider_name().unwrap_or_default()); + debug!("🌻🌻🌻 Added provider connection {:?} for {addr}", guard.get_provider_name().unwrap_or_default()); self.connections.write().await.insert(addr.to_string(), Arc::clone(guard)); } } @@ -852,6 +870,7 @@ impl ActiveProviderManager { pub async fn release_connection(&self, addr: &str) { let guard = self.connections.write().await.remove(addr); if let Some(guard) = guard { + debug!("🌸🌸🌸 Released provider connection {:?} for {addr}", guard.get_provider_name().unwrap_or_default()); guard.release(); } } diff --git a/backend/src/api/model/active_user_manager.rs b/backend/src/api/model/active_user_manager.rs index 27177e5ff..8d79c5bb1 100644 --- a/backend/src/api/model/active_user_manager.rs +++ b/backend/src/api/model/active_user_manager.rs @@ -17,8 +17,8 @@ const USER_GC_TTL: u64 = 900; // 15 Min const USER_CON_TTL: u64 = 10_800; // 3 hours const USER_SESSION_LIMIT: usize = 50; -type ActiveUserConnectionChangeSender = tokio::sync::mpsc::Sender; -pub type ActiveUserConnectionChangeReceiver = tokio::sync::mpsc::Receiver; +type ActiveUserConnectionChangeSender = tokio::sync::mpsc::UnboundedSender; +pub type ActiveUserConnectionChangeReceiver = tokio::sync::mpsc::UnboundedReceiver; macro_rules! active_user_manager_shared_impl { () => { @@ -39,7 +39,9 @@ macro_rules! active_user_manager_shared_impl { let is_log_user_enabled = self.is_log_user_enabled(); let user_connection_count = Self::get_active_connections(&user).await; let user_count = user.read().await.iter().filter(|(_, c)| c.connections > 0).count(); - let _= self.connection_change_tx.try_send(ActiveUserConnectionChange::Connections(user_count, user_connection_count)); + if let Err(err) = self.connection_change_tx.send(ActiveUserConnectionChange::Connections(user_count, user_connection_count)) { + error!("Failed to send active user connection change: user-count: {user_count}, user-connection-count: {user_connection_count] {err:?}"); + } if is_log_user_enabled { info!("Active Users: {user_count}, Active User Connections: {user_connection_count}"); } @@ -67,7 +69,9 @@ macro_rules! active_user_manager_shared_impl { self.drop_connection(&addr); self.shared_stream_manager.release_connection(addr, true).await; self.provider_manager.release_connection(addr).await; - let _= self.connection_change_tx.try_send(ActiveUserConnectionChange::Disconnected(addr.to_string())); + let Err(err) = self.connection_change_tx.send(ActiveUserConnectionChange::Disconnected(addr.to_string())) { + error!("Failed to send active user connection change: {err:?}"); + } self.log_active_user().await; } }; @@ -307,7 +311,9 @@ impl ActiveUserManager { user_by_addr.insert(addr.to_owned(), username.to_owned()); } - let _= self.connection_change_tx.try_send(ActiveUserConnectionChange::Connected(stream_info)); + if let Err(err) = self.connection_change_tx.send(ActiveUserConnectionChange::Connected(stream_info)) { + error!("Failed to send connection change: {err}"); + } self.log_active_user().await; UserConnectionGuard { diff --git a/backend/src/api/model/provider_config.rs b/backend/src/api/model/provider_config.rs index 1b4d7c23c..da900e4d4 100644 --- a/backend/src/api/model/provider_config.rs +++ b/backend/src/api/model/provider_config.rs @@ -9,8 +9,8 @@ use tokio::sync::RwLock; use shared::model::InputType; use shared::write_if_some; -pub type ProviderConnectionChangeSender = tokio::sync::mpsc::Sender<(String, usize)>; -pub type ProviderConnectionChangeReceiver = tokio::sync::mpsc::Receiver<(String, usize)>; +pub type ProviderConnectionChangeSender = tokio::sync::mpsc::UnboundedSender<(String, usize)>; +pub type ProviderConnectionChangeReceiver = tokio::sync::mpsc::UnboundedReceiver<(String, usize)>; pub type ProviderConnectionChangeCallback = Arc; diff --git a/backend/src/api/model/streams/shared_stream_manager.rs b/backend/src/api/model/streams/shared_stream_manager.rs index 9abc27b8b..a31b092ff 100644 --- a/backend/src/api/model/streams/shared_stream_manager.rs +++ b/backend/src/api/model/streams/shared_stream_manager.rs @@ -103,7 +103,10 @@ pub struct SharedStreamState { impl Drop for SharedStreamState { fn drop(&mut self) { if let Some(guard) = self.provider_guard.as_ref() { - guard.force_release(); + let guard = guard.clone(); + tokio::spawn(async move { + guard.force_release(); + }); } } } @@ -128,7 +131,7 @@ impl SharedStreamState { } } - async fn subscribe(&self, addr: &str, manager: Arc) -> BoxedProviderStream { + async fn subscribe(&self, addr: &str, manager: Arc) -> (BoxedProviderStream, Option) { let (client_tx, client_rx) = mpsc::channel(self.buf_size); let mut broadcast_rx = self.broadcaster.subscribe(); let cancel_token = CancellationToken::new(); @@ -147,8 +150,8 @@ impl SharedStreamState { let mut loop_cnt = 0; loop { - loop_cnt += 1; - tokio::select! { + loop_cnt += 1; + tokio::select! { biased; () = cancel_token.cancelled() => { @@ -181,7 +184,13 @@ impl SharedStreamState { } manager.release_connection(&address, false).await; }); - convert_stream(ReceiverStream::new(client_rx).boxed()) + + let provider = match &self.provider_guard { + None => None, + Some(connection_guard) => connection_guard.get_provider_name() + }; + + (convert_stream(ReceiverStream::new(client_rx).boxed()), provider) } fn broadcast( @@ -300,6 +309,7 @@ impl SharedStreamManager { } pub async fn release_connection(&self, addr: &str, send_stop_signal: bool) { + let stream_url = { self.shared_streams_by_addr.write().await.remove(addr) }; @@ -336,7 +346,7 @@ impl SharedStreamManager { } - async fn subscribe_stream(&self, stream_url: &str, addr: Option<&str>, manager: Arc) -> Option { + async fn subscribe_stream(&self, stream_url: &str, addr: Option<&str>, manager: Arc) -> Option<(BoxedProviderStream, Option)> { let shared_stream_state = self.shared_streams.read().await.get(stream_url).map(Arc::clone); match shared_stream_state { None => None, @@ -344,8 +354,7 @@ impl SharedStreamManager { if let Some(address) = addr { debug_if_enabled!("Responding to existing shared client stream {}", sanitize_sensitive_info(stream_url)); self.shared_streams_by_addr.write().await.insert(address.to_string(), stream_url.to_owned()); - let stream = stream_state.subscribe(address, manager).await; - Some(stream) + Some(stream_state.subscribe(address, manager).await) } else { None } @@ -364,7 +373,7 @@ impl SharedStreamManager { addr: Option<&str>, headers: Vec<(String, String)>, buffer_size: usize, - provider_guard: Option>) -> Option + provider_guard: Option>) -> Option<(BoxedProviderStream, Option)> where S: Stream> + Unpin + 'static + Send, E: std::fmt::Debug + Send, @@ -383,7 +392,7 @@ impl SharedStreamManager { app_state: &AppState, stream_url: &str, addr: Option<&str>, - ) -> Option { + ) -> Option<(BoxedProviderStream, Option)> { let manager = Arc::clone(&app_state.shared_stream_manager); app_state.shared_stream_manager.subscribe_stream(stream_url, addr, manager).await } diff --git a/backend/src/auth/fingerprint.rs b/backend/src/auth/fingerprint.rs index 791f09459..75343d718 100644 --- a/backend/src/auth/fingerprint.rs +++ b/backend/src/auth/fingerprint.rs @@ -59,18 +59,16 @@ impl Fingerprint { } } - let client_ip = format!("{}:{}", real_ip.as_ref() + let client_ip = real_ip.as_ref() .map(ToString::to_string) .or(forwarded_for.as_ref().map(ToString::to_string)) - .unwrap_or_else(|| addr.ip().to_string()), - addr.port()); + .unwrap_or_else(|| addr.ip().to_string()); + + let client_ip_port =format!("{client_ip}:{}", addr.port()); let ua = user_agent.unwrap_or_else(String::new); - let key = match real_ip.or(forwarded_for) { - Some(xff) => format!("{}{xff}{ua}", addr.ip()), - None => format!("{}{ua}", addr.ip()), - }; + let key = format!("{client_ip }{ua}"); - Ok(Fingerprint(key, client_ip)) + Ok(Fingerprint(key, client_ip_port)) } } \ No newline at end of file diff --git a/backend/src/repository/bplustree.rs b/backend/src/repository/bplustree.rs index c9bdeaff7..c5576d1bf 100644 --- a/backend/src/repository/bplustree.rs +++ b/backend/src/repository/bplustree.rs @@ -72,9 +72,8 @@ where let idx = get_entry_index_upper_bound::(&node.keys, key); if idx == 0 { return None; - } else { - return node.values.get(idx - 1).cloned(); } + return node.values.get(idx - 1).cloned(); } let child_idx = get_entry_index_upper_bound::(&node.keys, key); if let Some(child_offsets) = pointers { diff --git a/backend/src/utils/geoip.rs b/backend/src/utils/geoip.rs index f3ea4f25e..5b9a09515 100644 --- a/backend/src/utils/geoip.rs +++ b/backend/src/utils/geoip.rs @@ -6,7 +6,7 @@ use serde::{Serialize, Deserialize}; use crate::repository::bplustree::BPlusTree; fn ipv4_to_u32(ip: &str) -> Option { - ip.parse::().ok().map(|a| u32::from(a)) + ip.parse::().ok().map(u32::from) } #[derive(Serialize, Deserialize)] @@ -57,7 +57,7 @@ impl GeoIp { #[cfg(test)] mod test { - //https://github.com/datasets/geoip2-ipv4/blob/main/data/geoip2-ipv4.csv + // https://raw.githubusercontent.com/sapics/ip-location-db/refs/heads/main/asn-country/asn-country-ipv4.csv use std::fs::File; use std::io::BufReader; diff --git a/frontend/src/app/components/dashboard/streams_table.rs b/frontend/src/app/components/dashboard/streams_table.rs index 0f87136e9..c813c93fb 100644 --- a/frontend/src/app/components/dashboard/streams_table.rs +++ b/frontend/src/app/components/dashboard/streams_table.rs @@ -31,9 +31,25 @@ const HEADERS: [&str; 11] = [ "LABEL.DURATION" ]; -fn strip_port<'a>(input: &'a str) -> Cow<'a, str> { - if let Some(pos) = input.find(':') { - Cow::Owned(input[..pos].to_string()) +pub fn strip_port<'a>(input: &'a str) -> Cow<'a, str> { + // IPv6 with port: [2001:db8::1]:8080 + if let Some(stripped) = input.strip_prefix('[') { + if let Some(end) = stripped.find(']') { + return Cow::Owned(stripped[..end].to_string()); + } + // Invalid IPv6 + return Cow::Borrowed(input); + } + + // IPv4 or IPv6 without bracket + if let Some((left, right)) = input.rsplit_once(':') { + // If `left` has a colon then its IPv6 without port. + if left.contains(':') { + Cow::Borrowed(input) + } else { + // IPv4:Port + Cow::Owned(left.to_string()) + } } else { Cow::Borrowed(input) } diff --git a/shared/Cargo.toml b/shared/Cargo.toml index bb30e6594..c1abca4ce 100644 --- a/shared/Cargo.toml +++ b/shared/Cargo.toml @@ -22,4 +22,5 @@ zeroize = "1" chrono = "0.4.42" bytes = "1" ciborium = "0.2.2" +#[cfg(target_arch = "wasm32")] js-sys = "0.3.81" From 364e3d69df6dfe4f767eff6ffd0af63212f2a72e Mon Sep 17 00:00:00 2001 From: euzu Date: Mon, 27 Oct 2025 13:58:55 +0100 Subject: [PATCH 3/5] task optimizations for connection release --- Cargo.lock | 85 +++-------------- backend/Cargo.toml | 4 +- backend/src/api/api_utils.rs | 2 + backend/src/api/endpoints/v1_api_playlist.rs | 2 +- backend/src/api/main_api.rs | 4 +- .../src/api/model/active_provider_manager.rs | 1 + backend/src/api/model/active_user_manager.rs | 65 +++++++++---- backend/src/api/model/event_manager.rs | 3 +- .../model/streams/shared_stream_manager.rs | 5 +- backend/src/api/serve.rs | 16 ++-- backend/src/messaging.rs | 92 +++++++++---------- backend/src/processing/playlist_watch.rs | 8 +- backend/src/processing/processor/playlist.rs | 10 +- backend/src/utils/network/xtream.rs | 6 +- backend/src/utils/telegram.rs | 40 ++++---- frontend/Cargo.toml | 2 +- 16 files changed, 154 insertions(+), 191 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 8a7890b16..2fb40753b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,15 +2,6 @@ # It is not intended for manual editing. version = 4 -[[package]] -name = "addr2line" -version = "0.25.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1b5d307320b3181d6d7954e663bd7c774a838b8220fe0593c86d9fb09f498b4b" -dependencies = [ - "gimli", -] - [[package]] name = "adler2" version = "2.0.1" @@ -239,21 +230,6 @@ dependencies = [ "syn 2.0.106", ] -[[package]] -name = "backtrace" -version = "0.3.76" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bb531853791a215d7c62a30daf0dde835f381ab5de4589cfe7c649d2cbe92bd6" -dependencies = [ - "addr2line", - "cfg-if", - "libc", - "miniz_oxide", - "object", - "rustc-demangle", - "windows-link 0.2.1", -] - [[package]] name = "base16ct" version = "0.2.0" @@ -1022,7 +998,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -1337,12 +1313,6 @@ dependencies = [ "wasm-bindgen", ] -[[package]] -name = "gimli" -version = "0.32.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e629b9b98ef3dd8afe6ca2bd0f89306cec16d43d907889945bc5d6687f2f13c7" - [[package]] name = "gloo" version = "0.8.1" @@ -2248,17 +2218,6 @@ dependencies = [ "libc", ] -[[package]] -name = "io-uring" -version = "0.7.10" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "046fa2d4d00aea763528b4950358d0ead425372445dc8ff86312b3c69ff7727b" -dependencies = [ - "bitflags 2.9.4", - "cfg-if", - "libc", -] - [[package]] name = "ipnet" version = "2.11.0" @@ -2342,9 +2301,9 @@ dependencies = [ [[package]] name = "jsonwebtoken" -version = "10.0.0" +version = "10.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f1417155a38e99d7704ddb3ea7445fe57fdbd5d756d727740a9ed8b9ebaed6e1" +checksum = "3d119c6924272d16f0ab9ce41f7aa0bfef9340c00b0bb7ca3dd3b263d4a9150b" dependencies = [ "base64", "ed25519-dalek", @@ -2682,15 +2641,6 @@ dependencies = [ "libc", ] -[[package]] -name = "object" -version = "0.37.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ff76201f031d8863c38aa7f905eca4f53abbfa15f609db4277d44cd8938f33fe" -dependencies = [ - "memchr", -] - [[package]] name = "once_cell" version = "1.21.3" @@ -3365,9 +3315,9 @@ dependencies = [ [[package]] name = "regex" -version = "1.12.1" +version = "1.12.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4a52d8d02cacdb176ef4678de6c052efb4b3da14b78e4db683a4252762be5433" +checksum = "843bc0191f75f3e22651ae5f1e72939ab2f72a4bc30fa80a066bd66edefc24d4" dependencies = [ "aho-corasick", "memchr", @@ -3547,12 +3497,6 @@ dependencies = [ "crossbeam-utils", ] -[[package]] -name = "rustc-demangle" -version = "0.1.26" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "56f7d92ca342cea22a06f2121d944b4fd82af56988c270852495420f961d4ace" - [[package]] name = "rustc-hash" version = "2.1.1" @@ -3578,7 +3522,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -4057,7 +4001,7 @@ dependencies = [ "getrandom 0.3.3", "once_cell", "rustix", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -4160,28 +4104,25 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" [[package]] name = "tokio" -version = "1.47.1" +version = "1.48.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "89e49afdadebb872d3145a5638b59eb0691ea23e46ca484037cfab3b76b95038" +checksum = "ff360e02eab121e0bc37a2d3b4d4dc622e6eda3a8e5253d5435ecf5bd4c68408" dependencies = [ - "backtrace", "bytes", - "io-uring", "libc", "mio", "parking_lot", "pin-project-lite", - "slab", "socket2", "tokio-macros", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] name = "tokio-macros" -version = "2.5.0" +version = "2.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6e06d43f1345a3bcd39f6a56dbb7dcab2ba47e68e8ac134855e7e2bdbaf8cab8" +checksum = "af407857209536a95c8e56f8231ef2c2e2aff839b22e07a1ffcbc617e9db9fa5" dependencies = [ "proc-macro2", "quote", @@ -4840,7 +4781,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 2d7f7ad1d..813eca38d 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -22,7 +22,7 @@ axum = { version = "0.8" , features = ["macros", "default", "ws"]} tower = "0.5" tower-http = { version = "0.6", features = ["cors", "auth", "fs", "compression-full", "trace"] } tower_governor = { version = "0.8", features = ["axum"] } -jsonwebtoken = { version = "10.0", features = ["rust_crypto"] } +jsonwebtoken = { version = "10.1", features = ["rust_crypto"] } rust-argon2 = "3" futures = "0.3" path-clean = "1.0" @@ -41,7 +41,7 @@ flate2 = "1.1" blake3 = "1.8" bytes = "1.10" tokio-stream = { version = "0.1", features = ["sync"] } -tokio = { version = "1.47", features = ["rt-multi-thread", "parking_lot", "fs"] } +tokio = { version = "1.48", features = ["rt-multi-thread", "parking_lot", "fs"] } #tokio = { version = "1.46", features = ["rt-multi-thread", "parking_lot", "fs", "tracing"] } #console-subscriber = "0" #tracing = "0.1" diff --git a/backend/src/api/api_utils.rs b/backend/src/api/api_utils.rs index 1a22a50c2..1c9cc921b 100644 --- a/backend/src/api/api_utils.rs +++ b/backend/src/api/api_utils.rs @@ -1056,6 +1056,8 @@ fn get_add_cache_content( let cache = Arc::clone(cache); let add_cache_content: Arc = Arc::new(move |size| { let res_url = resource_url.clone(); + + // todo spawn, replace with unboundchannel let cache = Arc::clone(&cache); tokio::spawn(async move { if let Some(cache) = cache.load().as_ref() { diff --git a/backend/src/api/endpoints/v1_api_playlist.rs b/backend/src/api/endpoints/v1_api_playlist.rs index e040f5263..f73642ce0 100644 --- a/backend/src/api/endpoints/v1_api_playlist.rs +++ b/backend/src/api/endpoints/v1_api_playlist.rs @@ -63,7 +63,7 @@ async fn playlist_update( let app_config = Arc::clone(&app_state.app_config); let event_manager = Arc::clone(&app_state.event_manager); let playlist_state = Arc::clone(&app_state.playlists); - tokio::spawn(playlist::exec_processing(Arc::clone(&app_state.http_client.load()), app_config, Arc::new(valid_targets), Some(event_manager), Some(playlist_state))); + playlist::exec_processing(Arc::clone(&app_state.http_client.load()), app_config, Arc::new(valid_targets), Some(event_manager), Some(playlist_state)).await; axum::http::StatusCode::ACCEPTED.into_response() } Err(err) => { diff --git a/backend/src/api/main_api.rs b/backend/src/api/main_api.rs index d582b5430..9f3a53ca1 100644 --- a/backend/src/api/main_api.rs +++ b/backend/src/api/main_api.rs @@ -72,12 +72,12 @@ fn create_shared_data( let (provider_change_tx, provider_change_rx) = tokio::sync::mpsc::unbounded_channel(); let active_provider = Arc::new(ActiveProviderManager::new(app_config, provider_change_tx)); let (active_user_change_tx, active_user_change_rx) = tokio::sync::mpsc::unbounded_channel(); - let active_users = Arc::new(ActiveUserManager::new( + let active_users = ActiveUserManager::new( &config, &shared_stream_manager, &active_provider, active_user_change_tx, - )); + ); let event_manager = Arc::new(EventManager::new(active_user_change_rx, provider_change_rx, )); let client = create_http_client(app_config); diff --git a/backend/src/api/model/active_provider_manager.rs b/backend/src/api/model/active_provider_manager.rs index 4c42fca18..82ea42353 100644 --- a/backend/src/api/model/active_provider_manager.rs +++ b/backend/src/api/model/active_provider_manager.rs @@ -41,6 +41,7 @@ impl ProviderConnectionGuard { fn send_release(&self, config: &Arc) { let provider_config = Arc::clone(config); if let Err(_err) = &self.release_tx.send(Arc::clone(config)) { + // Fallback tokio::spawn(async move { provider_config.release().await; }); diff --git a/backend/src/api/model/active_user_manager.rs b/backend/src/api/model/active_user_manager.rs index 8d79c5bb1..af67c060f 100644 --- a/backend/src/api/model/active_user_manager.rs +++ b/backend/src/api/model/active_user_manager.rs @@ -1,5 +1,5 @@ use std::borrow::Cow; -use crate::api::model::ActiveProviderManager; +use crate::api::model::{ActiveProviderManager}; use crate::api::model::SharedStreamManager; use crate::model::Config; use crate::model::ProxyUserCredentials; @@ -10,6 +10,7 @@ use shared::utils::{current_time_secs, default_grace_period_millis, default_grac use std::collections::HashMap; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::Arc; +use tokio::sync::mpsc::{unbounded_channel, UnboundedSender}; use tokio::sync::RwLock; @@ -40,7 +41,7 @@ macro_rules! active_user_manager_shared_impl { let user_connection_count = Self::get_active_connections(&user).await; let user_count = user.read().await.iter().filter(|(_, c)| c.connections > 0).count(); if let Err(err) = self.connection_change_tx.send(ActiveUserConnectionChange::Connections(user_count, user_connection_count)) { - error!("Failed to send active user connection change: user-count: {user_count}, user-connection-count: {user_connection_count] {err:?}"); + error!("Failed to send active user connection change: user-count: {user_count}, user-connection-count: {user_connection_count} {err:?}"); } if is_log_user_enabled { info!("Active Users: {user_count}, Active User Connections: {user_connection_count}"); @@ -69,7 +70,7 @@ macro_rules! active_user_manager_shared_impl { self.drop_connection(&addr); self.shared_stream_manager.release_connection(addr, true).await; self.provider_manager.release_connection(addr).await; - let Err(err) = self.connection_change_tx.send(ActiveUserConnectionChange::Disconnected(addr.to_string())) { + if let Err(err) = self.connection_change_tx.send(ActiveUserConnectionChange::Disconnected(addr.to_string())) { error!("Failed to send active user connection change: {err:?}"); } self.log_active_user().await; @@ -84,7 +85,7 @@ fn get_grace_options(config: &Config) -> (u64, u64) { (grace_period_millis, grace_period_timeout_secs) } -struct ConnectionGuardUserManager { +pub struct ConnectionGuardUserManager { log_active_user: bool, user: Arc>>, user_by_addr: Arc>>, @@ -105,18 +106,28 @@ pub struct UserConnectionGuard { manager: Arc, // username: String, addr: String, + release_tx: UnboundedSender, } + +impl UserConnectionGuard { + pub fn new(manager: Arc, addr: &str, release_tx: UnboundedSender) -> Self { + Self { + manager, + addr: addr.to_string(), + release_tx, + } + } +} + impl Drop for UserConnectionGuard { fn drop(&mut self) { let manager = self.manager.clone(); let addr = self.addr.clone(); - if let Ok(rt) = tokio::runtime::Handle::try_current() { - rt.spawn(async move { + if let Err(_err) = self.release_tx.send(addr.clone()) { + // fallback + tokio::spawn(async move { manager.remove_connection(&addr).await; }); - } else { - // Fallback: no runtime - error!("Runtime not available, cannot cleanly remove connection for {addr}"); } } } @@ -178,14 +189,19 @@ pub struct ActiveUserManager { shared_stream_manager: Arc, provider_manager: Arc, connection_change_tx: ActiveUserConnectionChangeSender, + release_tx: UnboundedSender, } impl ActiveUserManager { - 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); + pub fn new(config: &Config, shared_stream_manager: &Arc, provider_manager: &Arc, connection_change_tx: ActiveUserConnectionChangeSender) -> Arc { + let log_active_user: bool = 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); - Self { + + // Create the cleanup channel + let (cleanup_tx, mut cleanup_rx) = unbounded_channel::(); + + let active_user_manager = Arc::new(Self { 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), @@ -196,11 +212,29 @@ impl ActiveUserManager { shared_stream_manager: Arc::clone(shared_stream_manager), provider_manager: Arc::clone(provider_manager), connection_change_tx, - } + release_tx: cleanup_tx, + }); + + let active_user_manager_clone = Arc::clone(&active_user_manager); + // Spawn the async cleanup worker + tokio::spawn(async move { + while let Some(addr) = cleanup_rx.recv().await { + debug!("🧹 User manager - connection releasing {:?}", addr); + active_user_manager_clone.remove_connection(&addr).await + } + debug!("User manager - cleanup worker terminated"); + }); + + + active_user_manager } active_user_manager_shared_impl!(); + pub fn release_sender(&self) -> UnboundedSender { + self.release_tx.clone() + } + 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); @@ -316,10 +350,7 @@ impl ActiveUserManager { } self.log_active_user().await; - UserConnectionGuard { - manager: Arc::new(self.clone_inner()), - addr: addr.to_owned(), - } + UserConnectionGuard::new(Arc::new(self.clone_inner()), addr, self.release_sender()) } fn is_log_user_enabled(&self) -> bool { diff --git a/backend/src/api/model/event_manager.rs b/backend/src/api/model/event_manager.rs index 46a738d07..0275b3b72 100644 --- a/backend/src/api/model/event_manager.rs +++ b/backend/src/api/model/event_manager.rs @@ -1,5 +1,4 @@ use log::{info, trace}; -use tokio::task; use shared::model::{ActiveUserConnectionChange, ConfigType, PlaylistUpdateState}; use crate::api::model::{ActiveUserConnectionChangeReceiver}; use crate::api::model::{ProviderConnectionChangeReceiver}; @@ -27,7 +26,7 @@ impl EventManager { let (channel_tx, _channel_rx) = tokio::sync::broadcast::channel(10); let channel_tx_clone = channel_tx.clone(); - task::spawn(async move { + tokio::spawn(async move { loop { tokio::select! { Some(event) = active_user_change_rx.recv() => { diff --git a/backend/src/api/model/streams/shared_stream_manager.rs b/backend/src/api/model/streams/shared_stream_manager.rs index a31b092ff..1d02246ac 100644 --- a/backend/src/api/model/streams/shared_stream_manager.rs +++ b/backend/src/api/model/streams/shared_stream_manager.rs @@ -103,10 +103,7 @@ pub struct SharedStreamState { impl Drop for SharedStreamState { fn drop(&mut self) { if let Some(guard) = self.provider_guard.as_ref() { - let guard = guard.clone(); - tokio::spawn(async move { - guard.force_release(); - }); + guard.force_release(); } } } diff --git a/backend/src/api/serve.rs b/backend/src/api/serve.rs index caad8de58..f7b03398b 100644 --- a/backend/src/api/serve.rs +++ b/backend/src/api/serve.rs @@ -134,11 +134,7 @@ where let user_manager_clone = Arc::clone(&user_manager); let mut addr_close_rx = user_manager_clone.get_close_connection_channel(); - let connection_closed = async move || { - debug!("Connection closed: {remote_addr}"); - let addr = remote_addr.to_string(); - user_manager_clone.remove_connection(&addr).await; - }; + let connection_release = user_manager.release_sender(); debug!("Connection opened: {addr_str}"); @@ -148,11 +144,17 @@ where if let Err(err) = result { trace!("failed to serve connection: {err:#}"); } - connection_closed().await; + if let Err(_err) = connection_release.send(remote_addr.to_string()) { + let addr = remote_addr.to_string(); + user_manager_clone.remove_connection(&addr).await; + } break; } () = &mut signal_closed => { - connection_closed().await; + if let Err(_err) = connection_release.send(remote_addr.to_string()) { + let addr = remote_addr.to_string(); + user_manager_clone.remove_connection(&addr).await; + } debug!("Connection gracefully closed: {remote_addr}"); conn.as_mut().graceful_shutdown(); } diff --git a/backend/src/messaging.rs b/backend/src/messaging.rs index 5411cac44..29eed38d2 100644 --- a/backend/src/messaging.rs +++ b/backend/src/messaging.rs @@ -11,27 +11,23 @@ fn is_enabled(kind: MsgKind, cfg: &MessagingConfig) -> bool { cfg.notify_on.contains(&kind) } -fn send_http_post_request(client: &Arc, msg: &str, messaging: &MessagingConfig) { +async fn send_http_post_request(client: &Arc, msg: &str, messaging: &MessagingConfig) { if let Some(rest) = &messaging.rest { - let url = rest.url.clone(); - let data = msg.to_owned(); - let the_client = Arc::clone(client); - tokio::spawn(async move { - match the_client - .post(&url) - .header(header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string()) - .body(data) - .send() - .await - { - Ok(_) => debug!("Text message sent successfully to rest api"), - Err(e) => error!("Text message wasn't sent to rest api because of: {e}"), - } - }); + let data = msg.to_owned(); + match client + .post(&rest.url) + .header(header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string()) + .body(data) + .send() + .await + { + Ok(_) => debug!("Text message sent successfully to rest api"), + Err(e) => error!("Text message wasn't sent to rest api because of: {e}"), + } } } -fn send_telegram_message(client: &Arc, msg: &str, messaging: &MessagingConfig, json: bool) { +async fn send_telegram_message(client: &Arc, msg: &str, messaging: &MessagingConfig, json: bool) { // TODO use proxy settings if let Some(telegram) = &messaging.telegram { let (message, options) = { @@ -48,55 +44,51 @@ fn send_telegram_message(client: &Arc, msg: &str, messaging: &M for chat_id in &telegram.chat_ids { let bot = telegram_create_instance(&telegram.bot_token, chat_id); - telegram_send_message(client, &bot, &message, options.as_ref()); + telegram_send_message(client, &bot, &message, options.as_ref()).await; } } } -fn send_pushover_message(client: &Arc, msg: &str, messaging: &MessagingConfig) { +async fn send_pushover_message(client: &Arc, msg: &str, messaging: &MessagingConfig) { if let Some(pushover) = &messaging.pushover { let encoded_message: String = url::form_urlencoded::Serializer::new(String::new()) .append_pair("token", pushover.token.as_str()) .append_pair("user", pushover.user.as_str()) .append_pair("message", msg) .finish(); - let the_client = Arc::clone(client); - let pushover_url = pushover.url.clone(); - tokio::spawn(async move { - match the_client - .post(pushover_url) - .header(header::CONTENT_TYPE, mime::APPLICATION_WWW_FORM_URLENCODED.to_string()) - .body(encoded_message) - .send() - .await - { - Ok(response) => { - if response.status().is_success() { - debug!("Text message sent successfully to PUSHOVER, status code {}", response.status()); - } else { - error!("Failed to send text message to PUSHOVER, status code {}", response.status()); - } + match client + .post(&pushover.url) + .header(header::CONTENT_TYPE, mime::APPLICATION_WWW_FORM_URLENCODED.to_string()) + .body(encoded_message) + .send() + .await + { + Ok(response) => { + if response.status().is_success() { + debug!("Text message sent successfully to PUSHOVER, status code {}", response.status()); + } else { + error!("Failed to send text message to PUSHOVER, status code {}", response.status()); } - Err(e) => error!("Text message wasn't sent to PUSHOVER api because of: {e}"), } - }); - } -} - -fn dispatch_send_message(client: &Arc, kind: MsgKind, cfg: Option<&MessagingConfig>, msg: &str, json: bool) { - if let Some(messaging) = cfg { - if is_enabled(kind, messaging) { - send_telegram_message(client, msg, messaging, json); - send_http_post_request(client, msg, messaging); - send_pushover_message(client, msg, messaging); + Err(e) => error!("Text message wasn't sent to PUSHOVER api because of: {e}"), } } } -pub fn send_message_json(client: &Arc, kind: MsgKind, cfg: Option<&MessagingConfig>, msg: &str) { - dispatch_send_message(client, kind, cfg, msg, true); +async fn dispatch_send_message(client: &Arc, kind: MsgKind, cfg: Option<&MessagingConfig>, msg: &str, json: bool) { + if let Some(messaging) = cfg { + if is_enabled(kind, messaging) { + send_telegram_message(client, msg, messaging, json).await; + send_http_post_request(client, msg, messaging).await; + send_pushover_message(client, msg, messaging).await; + } + } } -pub fn send_message(client: &Arc, kind: MsgKind, cfg: Option<&MessagingConfig>, msg: &str) { - dispatch_send_message(client, kind, cfg, msg, false); +pub async fn send_message_json(client: &Arc, kind: MsgKind, cfg: Option<&MessagingConfig>, msg: &str) { + dispatch_send_message(client, kind, cfg, msg, true).await; +} + +pub async fn send_message(client: &Arc, kind: MsgKind, cfg: Option<&MessagingConfig>, msg: &str) { + dispatch_send_message(client, kind, cfg, msg, false).await; } diff --git a/backend/src/processing/playlist_watch.rs b/backend/src/processing/playlist_watch.rs index 650a207d5..3c77bbe22 100644 --- a/backend/src/processing/playlist_watch.rs +++ b/backend/src/processing/playlist_watch.rs @@ -8,7 +8,7 @@ use crate::model::Config; use crate::utils; use crate::utils::{bincode_deserialize, bincode_serialize}; -pub fn process_group_watch(client: &Arc, cfg: &Config, target_name: &str, pl: &PlaylistGroup) { +pub async fn process_group_watch(client: &Arc, cfg: &Config, target_name: &str, pl: &PlaylistGroup) { let mut new_tree = BTreeSet::new(); pl.channels.iter().for_each(|chan| { let header = &chan.header; @@ -28,7 +28,7 @@ pub fn process_group_watch(client: &Arc, cfg: &Config, target_n let removed_difference: BTreeSet = loaded_tree.difference(&new_tree).cloned().collect(); if !added_difference.is_empty() || !removed_difference.is_empty() { changed = true; - handle_watch_notification(client, cfg, &added_difference, &removed_difference, target_name, &pl.title); + handle_watch_notification(client, cfg, &added_difference, &removed_difference, target_name, &pl.title).await; } } else { error!("failed to load watch_file {}", &path.to_str().unwrap_or_default()); @@ -60,7 +60,7 @@ struct WatchChanges { pub removed: Vec, } -fn handle_watch_notification(client: &Arc, cfg: &Config, added: &BTreeSet, removed: &BTreeSet, target_name: &str, group_name: &str) { +async fn handle_watch_notification(client: &Arc, cfg: &Config, added: &BTreeSet, removed: &BTreeSet, target_name: &str, group_name: &str) { let added = added.iter().map(std::string::ToString::to_string).collect::>(); let removed = removed.iter().map(std::string::ToString::to_string).collect::>(); if !added.is_empty() || !removed.is_empty() { @@ -73,7 +73,7 @@ fn handle_watch_notification(client: &Arc, cfg: &Config, added: let msg = serde_json::to_string_pretty(&changes).unwrap_or_else(|_| "Error: Failed to serialize watch changes".to_string()); info!("{}", &msg); - send_message(client, MsgKind::Watch, cfg.messaging.as_ref(), &msg); + send_message(client, MsgKind::Watch, cfg.messaging.as_ref(), &msg).await; } } diff --git a/backend/src/processing/processor/playlist.rs b/backend/src/processing/processor/playlist.rs index 8e36a87f7..d2a573de7 100644 --- a/backend/src/processing/processor/playlist.rs +++ b/backend/src/processing/processor/playlist.rs @@ -579,7 +579,7 @@ async fn process_playlist_for_target(app_config: &AppConfig, step.tick("assigning channel counter"); let config = app_config.config.load(); - if process_watch(&config, &client, target, &flat_new_playlist) { + if process_watch(&config, &client, target, &flat_new_playlist).await { step.tick("group watches"); } let result = persist_playlist(app_config, &mut flat_new_playlist, flatten_tvguide(&new_epg).as_ref(), target, playlist_state).await; @@ -620,14 +620,14 @@ async fn process_epg(processed_fetched_playlists: &mut Vec>) (new_epg, new_playlist) } -fn process_watch(cfg: &Config, client: &Arc, target: &ConfigTarget, new_playlist: &Vec) -> bool { +async fn process_watch(cfg: &Config, client: &Arc, target: &ConfigTarget, new_playlist: &Vec) -> bool { if let Some(watches) = &target.watch { if default_as_default().eq_ignore_ascii_case(&target.name) { error!("cant watch a target with no unique name"); } else { for pl in new_playlist { if watches.iter().any(|r| r.is_match(&pl.title)) { - process_group_watch(client, cfg, &target.name, pl); + process_group_watch(client, cfg, &target.name, pl).await; } } } @@ -657,7 +657,7 @@ pub async fn exec_processing(client: Arc, app_config: Arc error!("Failed to serialize playlist stats {err}"), } @@ -672,7 +672,7 @@ pub async fn exec_processing(client: Arc, app_config: Arc, input: &Input if let Ok(cur_status) = ProxyUserStatus::from_str(&status) { if !matches!(cur_status, ProxyUserStatus::Active | ProxyUserStatus::Trial) { warn!("User status for user {username} is {cur_status:?}"); - send_message(client, MsgKind::Info, cfg.messaging.as_ref(), &format!("User status for user {username} is {cur_status:?}")); + send_message(client, MsgKind::Info, cfg.messaging.as_ref(), &format!("User status for user {username} is {cur_status:?}")).await; } } } @@ -175,11 +175,11 @@ async fn xtream_login(cfg: &Config, client: &Arc, input: &Input let datetime = DateTime::from_timestamp(expiration_timestamp, 0).unwrap(); let formatted = datetime.format("%Y-%m-%d %H:%M:%S").to_string(); warn!("User account for user {username} expires {formatted}"); - send_message(client, MsgKind::Info, cfg.messaging.as_ref(), &format!("User account for user {username} expires {formatted}")); + send_message(client, MsgKind::Info, cfg.messaging.as_ref(), &format!("User account for user {username} expires {formatted}")).await; } } else { warn!("User account for user {username} is expired"); - send_message(client, MsgKind::Info, cfg.messaging.as_ref(), &format!("User account for user {username} is expired")); + send_message(client, MsgKind::Info, cfg.messaging.as_ref(), &format!("User account for user {username} is expired")).await; } } } diff --git a/backend/src/utils/telegram.rs b/backend/src/utils/telegram.rs index 75948037e..77bdceb92 100644 --- a/backend/src/utils/telegram.rs +++ b/backend/src/utils/telegram.rs @@ -62,7 +62,7 @@ pub fn telegram_create_instance(bot_token: &str, chat_id: &str) -> BotInstance { } } -pub fn telegram_send_message( +pub async fn telegram_send_message( client: &Arc, instance: &BotInstance, msg: &str, @@ -87,27 +87,25 @@ pub fn telegram_send_message( .map(ToString::to_string), }; - let the_client = Arc::clone(client); - tokio::spawn(async move { - let result = the_client - .post(url) - .json(&request_json_obj) - .send() - .await; + let result = client + .post(url) + .json(&request_json_obj) + .send() + .await; - match result { - Ok(response) => { - if response.status().is_success() { - debug!("Message sent successfully to {chat_id} telegram api"); - } else { - match response.json::().await { - Ok(json) => error!("Message wasn't sent to {chat_id} telegram api because of: {}", json.description), - Err(_) => error!("Message wasn't sent to {chat_id} telegram api. Telegram response could not be parsed!"), - } + match result { + Ok(response) => { + if response.status().is_success() { + debug!("Message sent successfully to {chat_id} telegram api"); + } else { + match response.json::().await { + Ok(json) => error!("Message wasn't sent to {chat_id} telegram api because of: {}", json.description), + Err(_) => error!("Message wasn't sent to {chat_id} telegram api. Telegram response could not be parsed!"), } - }, - Err(e) => error!("Message wasn't sent to {chat_id} telegram api because of: {e}"), - } - }); + } + }, + Err(e) => error!("Message wasn't sent to {chat_id} telegram api because of: {e}"), + } } + diff --git a/frontend/Cargo.toml b/frontend/Cargo.toml index c144e8b76..3e4ed7db5 100644 --- a/frontend/Cargo.toml +++ b/frontend/Cargo.toml @@ -28,7 +28,7 @@ futures = "0.3" prost = "0" wasm-bindgen-futures = "0" bytes = "1" -regex = "1.12.1" +regex = "1.12.2" base64 = "0.22.1" cron = "0.15" fastrand = "2.3.0" From 3ba4a5aaf6f19d410a4685463539974ac8ef397d Mon Sep 17 00:00:00 2001 From: euzu Date: Tue, 28 Oct 2025 13:55:37 +0100 Subject: [PATCH 4/5] hdhomerun fixes --- CHANGELOG.md | 1 + Cargo.lock | 1 + backend/Cargo.toml | 1 + backend/src/api/hdhomerun_proprietary.rs | 18 ++++---- backend/src/api/hdhomerun_ssdp.rs | 46 +++++++++++++------ backend/src/api/model/active_user_manager.rs | 4 +- .../config/hdhomerun_device_view.rs | 2 +- shared/src/utils/hdhomerun_utils.rs | 34 +++++++------- 8 files changed, 67 insertions(+), 40 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6141bb1ef..bd5cc72d6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,7 @@ Output filters are applied after all transformations have been performed, theref - Telegram supports markdown generation for structured json messages. simply set `markdown: true` in telegram config. - Added User-Stream-Connections Table to WebUI - Enhanced STRM output filenames to include detailed media quality info (e.g., 4K, HDR, x265, 5.1) for easy version distinction. +- Added standardized SSDP (Simple Service Discovery Protocol) and the Proprietary HDHomeRun UDP Discovery Protocol (Port 65001) # 3.1.7 (2025-10-10) - Added Dark/Bright theme switch diff --git a/Cargo.lock b/Cargo.lock index d30407665..f7fb0b0b5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4368,6 +4368,7 @@ dependencies = [ "deunicode", "enum-iterator", "env_logger", + "fastrand", "filetime", "flate2", "futures", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 32e4ddcf5..2c70c2906 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -36,6 +36,7 @@ log = "0.4" env_logger = "0.11" bincode = { version = "2.0", features = ["std", "serde"] } rand = "0.9" +fastrand = "2" rpassword = "7.4" flate2 = "1.1" blake3 = "1.8" diff --git a/backend/src/api/hdhomerun_proprietary.rs b/backend/src/api/hdhomerun_proprietary.rs index 4374d43ab..0ad2ad00f 100644 --- a/backend/src/api/hdhomerun_proprietary.rs +++ b/backend/src/api/hdhomerun_proprietary.rs @@ -52,8 +52,11 @@ fn write_tlv_str(buf: &mut BytesMut, tag: u8, value: &str) { if bytes.len() < 0x80 { buf.put_u8(u8::try_from(bytes.len()).unwrap_or(0)); } else { - buf.put_u8(0x82); - buf.put_u16(u16::try_from(bytes.len()).unwrap_or(0)); + let len = u16::try_from(bytes.len()).unwrap_or(0); + let byte_first = 0x80 | ((len & 0x7F) as u8); + let byte_second = ((len >> 7) & 0xFF) as u8; + buf.put_u8(byte_first); + buf.put_u8(byte_second); } buf.put_slice(bytes); } @@ -118,12 +121,11 @@ fn parse_tlv(cursor: &mut Cursor<&[u8]>) -> HashMap> { let len = if (len_buf[0] & 0x80) == 0 { len_buf[0] as usize } else { - let ext_len_bytes = (len_buf[0] & 0x7F) as usize; - let mut ext = vec![0u8; ext_len_bytes]; - if Read::read_exact(cursor, &mut ext).is_err() { + let mut second_byte = [0u8; 1]; + if Read::read_exact(cursor, &mut second_byte).is_err() { break; } - ext.iter().fold(0usize, |acc, b| (acc << 8) | (*b as usize)) + ((second_byte[0] as usize) << 7) + ((len_buf[0] & 0x7F) as usize) }; @@ -212,7 +214,7 @@ async fn handle_tcp_connection( Ok(0) => return, // Connection closed Ok(n) => { let request_data = &buf[..n]; - if request_data.len() < 4 { + if request_data.len() < 8 { continue; } @@ -221,7 +223,7 @@ async fn handle_tcp_connection( if msg_type == packet::HDHOMERUN_TYPE_GETSET_REQ { let response = process_getset_request(request_data, &app_state).await; if response.is_empty() { - error!("Protocol error or oinvalid request"); + error!("Protocol error or invalid request"); return; } if let Err(e) = stream.write_all(&response).await { diff --git a/backend/src/api/hdhomerun_ssdp.rs b/backend/src/api/hdhomerun_ssdp.rs index 3fd30ff76..9ac24e85c 100644 --- a/backend/src/api/hdhomerun_ssdp.rs +++ b/backend/src/api/hdhomerun_ssdp.rs @@ -36,19 +36,39 @@ async fn ssdp_task_loop(socket: UdpSocket, app_config: Arc, server_ho }; let request = String::from_utf8_lossy(&buf[..len]); - if request.starts_with("M-SEARCH") && (request.contains("urn:schemas-upnp-org:device:MediaServer:1") || request.contains("ssdp:all")) { - trace!("Received HDHomeRun M-SEARCH from {remote_addr}"); - let hdhomerun_guard = app_config.hdhomerun.load(); - if let Some(hd_config) = &*hdhomerun_guard { - if hd_config.enabled { - for device in &hd_config.devices { - if device.t_enabled { - let response = create_ssdp_response(device, &server_host); - if let Err(e) = socket.send_to(response.as_bytes(), remote_addr).await { - error!("Failed to send SSDP response to {remote_addr}: {e}"); - } else { - trace!("Sent SSDP response for device '{}' to {remote_addr}", device.name); - } + if !request.starts_with("M-SEARCH") { continue; } + let req = request.to_ascii_lowercase(); + if !req.contains(r#"man: "ssdp:discover""#) { continue; } + // Extract ST and MX (defaults) + let st = req.lines() + .find_map(|l| l.strip_prefix("st:").map(|v| v.trim().to_string())) + .unwrap_or_else(|| "ssdp:all".to_string()); + let mx: u64 = req.lines() + .find_map(|l| l.strip_prefix("mx:").and_then(|v| v.trim().parse().ok())) + .unwrap_or(1); + // Normalize to the set we support + let supported = [ + "urn:schemas-upnp-org:device:mediaserver:1", + "upnp:rootdevice", + "ssdp:all", + ]; + if !supported.contains(&st.as_str()) && st != "ssdp:all" { continue; } + // Randomized delay per MX + let delay_ms = (fastrand::u64(0..=mx*1000)).min(2000); + if delay_ms > 0 { tokio::time::sleep(Duration::from_millis(delay_ms)).await; } + + + trace!("Received HDHomeRun M-SEARCH from {remote_addr}"); + let hdhomerun_guard = app_config.hdhomerun.load(); + if let Some(hd_config) = &*hdhomerun_guard { + if hd_config.enabled { + for device in &hd_config.devices { + if device.t_enabled { + let response = create_ssdp_response(device, &server_host); + if let Err(e) = socket.send_to(response.as_bytes(), remote_addr).await { + error!("Failed to send SSDP response to {remote_addr}: {e}"); + } else { + trace!("Sent SSDP response for device '{}' to {remote_addr}", device.name); } } } diff --git a/backend/src/api/model/active_user_manager.rs b/backend/src/api/model/active_user_manager.rs index af67c060f..28310c324 100644 --- a/backend/src/api/model/active_user_manager.rs +++ b/backend/src/api/model/active_user_manager.rs @@ -219,8 +219,8 @@ impl ActiveUserManager { // Spawn the async cleanup worker tokio::spawn(async move { while let Some(addr) = cleanup_rx.recv().await { - debug!("🧹 User manager - connection releasing {:?}", addr); - active_user_manager_clone.remove_connection(&addr).await + debug!("🧹 User manager - connection releasing {addr:?}"); + active_user_manager_clone.remove_connection(&addr).await; } debug!("User manager - cleanup worker terminated"); }); diff --git a/frontend/src/app/components/config/hdhomerun_device_view.rs b/frontend/src/app/components/config/hdhomerun_device_view.rs index d11483493..c00baafb0 100644 --- a/frontend/src/app/components/config/hdhomerun_device_view.rs +++ b/frontend/src/app/components/config/hdhomerun_device_view.rs @@ -46,7 +46,7 @@ pub fn HdHomerunDeviceView(props: &HdHomerunDeviceViewProps) -> Html { let device_id = props.device_id; let deps = (form_state.clone(), form_state.modified); use_effect_with(deps, move |(state, modified)| { - on_form_change.emit((device_id, *modified, *state.form.clone())); + on_form_change.emit((device_id, *modified, (*state.form).clone())); }); } diff --git a/shared/src/utils/hdhomerun_utils.rs b/shared/src/utils/hdhomerun_utils.rs index 882f2add0..bfbbaa360 100644 --- a/shared/src/utils/hdhomerun_utils.rs +++ b/shared/src/utils/hdhomerun_utils.rs @@ -24,16 +24,20 @@ pub fn validate_hdhr_device_id(device_id: &str) -> bool { } pub fn generate_hdhr_device_id_from_base(base_id: &str) -> String { - let base_sanitized = base_id.chars().filter(|c| c.is_ascii_hexdigit()).collect::(); - let base_padded = if base_sanitized.is_empty() { - return generate_hdhr_device_id(); - } else { - format!("{:0<7}", &base_sanitized[..base_sanitized.len().min(7)]) - }; - - if let Ok(device_id_int_base) = u32::from_str_radix(&base_padded, 16) { - let checksum = calculate_checksum(device_id_int_base); - let final_id = (device_id_int_base & 0xFFFFFFF0) | u32::from(checksum); + let base_sanitized: String = base_id + .chars() + .filter(|c| c.is_ascii_hexdigit()) + .collect::() + .to_uppercase(); + if base_sanitized.is_empty() { + return generate_hdhr_device_id(); + } + // Keep at most 7 hex digits, pad-left with zeros to 7 + let base7 = format!("{:0>7}", &base_sanitized[..base_sanitized.len().min(7)]); + if let Ok(base7_int) = u32::from_str_radix(&base7, 16) { + let base_shifted = base7_int << 4; // bits 4-31 for base, bits 0-3 for checksum + let checksum = calculate_checksum(base_shifted); + let final_id = base_shifted | u32::from(checksum); format!("{:08X}", final_id) } else { generate_hdhr_device_id() @@ -41,10 +45,8 @@ pub fn generate_hdhr_device_id_from_base(base_id: &str) -> String { } pub fn generate_hdhr_device_id() -> String { - let random_part: String = (0..4) - .map(|_| format!("{:X}", fastrand::u8(0..16))) - .collect(); - - let base_id = format!("105{}0", random_part); - generate_hdhr_device_id_from_base(&base_id) + // 3 fixed + 4 random = 7 hex digits base + let rnd = (0..4).map(|_| format!("{:X}", fastrand::u8(0..16))).collect::(); + let base7 = format!("105{rnd}"); + generate_hdhr_device_id_from_base(&base7) } \ No newline at end of file From 3c3c8df763ff34f65a51c75929e21b533ac61f70 Mon Sep 17 00:00:00 2001 From: euzu Date: Tue, 28 Oct 2025 18:42:09 +0100 Subject: [PATCH 5/5] Added GeoIp config and update button to WebUI Integrated GeoIp to stream Table Fixed some HdHomerun issues --- README.md | 23 ++++++- backend/src/api/config_watch.rs | 2 +- backend/src/api/endpoints/v1_api.rs | 63 ++++++++++++++++++- backend/src/api/main_api.rs | 18 ++++++ backend/src/api/model/active_user_manager.rs | 57 +++++++++++------ backend/src/api/model/app_state.rs | 29 ++++++++- backend/src/api/model/provider_config.rs | 9 ++- backend/src/model/config/base.rs | 4 ++ backend/src/model/config/geoip.rs | 28 +++++++++ backend/src/model/config/mod.rs | 2 + backend/src/model/config/reverse_proxy.rs | 5 +- backend/src/repository/storage.rs | 4 ++ backend/src/utils/file/config_reader.rs | 8 +-- backend/src/utils/geoip.rs | 7 +++ backend/src/utils/mod.rs | 1 + frontend/public/assets/i18n/en.json | 9 ++- .../src/app/components/config/config_view.rs | 29 ++++++++- .../config/hdhomerun_config_view.rs | 2 +- .../config/reverse_proxy_config_view.rs | 51 ++++++++++++--- .../app/components/dashboard/streams_table.rs | 40 +++--------- frontend/src/services/config_service.rs | 6 ++ shared/src/model/config/base.rs | 13 ++-- shared/src/model/config/geoip.rs | 11 ++++ shared/src/model/config/hdhomerun.rs | 36 ++++++----- shared/src/model/config/mod.rs | 2 + shared/src/model/config/reverse_proxy.rs | 5 +- shared/src/model/stream_info.rs | 4 +- shared/src/utils/mod.rs | 2 + shared/src/utils/net_utils.rs | 25 ++++++++ 29 files changed, 396 insertions(+), 99 deletions(-) create mode 100644 backend/src/model/config/geoip.rs create mode 100644 shared/src/model/config/geoip.rs create mode 100644 shared/src/utils/net_utils.rs diff --git a/README.md b/README.md index cb89b6a7e..1a8cfc2d3 100644 --- a/README.md +++ b/README.md @@ -217,7 +217,8 @@ Attributes: - `buffer` - `throttle` Allowed units are `KB/s`,`MB/s`,`KiB/s`,`MiB/s`,`kbps`,`mbps`,`Mibps`. Default unit is `kbps` - `grace_period_millis` default set to 300 milliseconds. -- `grace_period_timeout_secs` efault set to 2 seconds. +- `grace_period_timeout_secs` default set to 2 seconds. +- `geopip` is for resolving ip addresses to country names. ##### 1.6.1.1 `retry` If set to `true` on connection loss to provider, the stream will be reconnected. @@ -268,6 +269,26 @@ If the connection is not throttled, the player will play its buffered content lo ##### 1.6.1.4 `grace_period_timeout_secs` How long the grace grant will last, until another grace grant can made. +##### 1.6.1.5 `geoip` +Disabled by default. +Is used to resolve ip addresses to location. +It has 2 attributes: +```yaml + geoip: + enabled: true + url: +``` + +The `url` is optional and default vaue is: `https://raw.githubusercontent.com/sapics/ip-location-db/refs/heads/main/asn-country/asn-country-ipv4.csv` +The format is csv with 3 columns `range_start,range_end,country_code` + +Example: +```csv +1.0.0.0,1.0.0.255,AU +1.0.1.0,1.0.3.255,CN +1.0.4.0,1.0.7.255,AU +``` + #### 1.6.2 `cache` LRU-Cache is for resources. If it is `enabled`, the resources/images are persisted in the given `dir`. If the cache size exceeds `size`, In an LRU cache, the least recently used items are evicted to make room for new items if the cache `size`is exceeded. diff --git a/backend/src/api/config_watch.rs b/backend/src/api/config_watch.rs index 31d708ed3..4d88d2759 100644 --- a/backend/src/api/config_watch.rs +++ b/backend/src/api/config_watch.rs @@ -65,7 +65,7 @@ impl ConfigFile { async fn load_config(app_state: &Arc) -> Result<(), TuliproxError> { let paths = > as Access>::load(&app_state.app_config.paths); let config_file = paths.config_file_path.as_str(); - let config_dto = read_config_file(config_file, true)?; + let config_dto = read_config_file(config_file, true, true)?; let mapping_changed = paths.mapping_file_path.as_ref() != config_dto.mapping_path.as_ref(); let mut config: Config = Config::from(config_dto); config.prepare(paths.config_path.as_str())?; diff --git a/backend/src/api/endpoints/v1_api.rs b/backend/src/api/endpoints/v1_api.rs index 8eddc7125..35743c1d9 100644 --- a/backend/src/api/endpoints/v1_api.rs +++ b/backend/src/api/endpoints/v1_api.rs @@ -8,12 +8,18 @@ use crate::auth::validator_admin; use crate::utils::ip_checker::get_ips; use crate::{VERSION}; use axum::response::IntoResponse; -use shared::model::{IpCheckDto,StatusCheck}; +use shared::model::{InputFetchMethod, IpCheckDto, StatusCheck}; use shared::utils::{concat_path_leading_slash}; -use std::collections::BTreeMap; +use std::collections::{BTreeMap, HashMap}; +use std::io::{Cursor}; use std::sync::Arc; +use log::error; use crate::api::endpoints::extract_accept_header::ExtractAcceptHeader; use crate::api::endpoints::v1_api_config::v1_api_config_register; +use crate::model::InputSource; +use crate::repository::storage::get_geoip_path; +use crate::utils::GeoIp; +use crate::utils::request::download_text_content; async fn create_ipinfo_check(app_state: &Arc) -> Option<(Option, Option)> { let config = app_state.app_config.config.load(); @@ -67,6 +73,58 @@ async fn streams(ExtractAcceptHeader(accept): ExtractAcceptHeader, json_or_bin_response(accept.as_ref(), &streams).into_response() } +async fn geoip_update(axum::extract::State(app_state): axum::extract::State>) -> axum::response::Response { + let config = app_state.app_config.config.load(); + if let Some(geoip) = config.reverse_proxy.as_ref().and_then(|r| r.geoip.as_ref()) { + if geoip.enabled { + let geoip_db_path = &*get_geoip_path(&config.working_dir); + let _file_lock = app_state.app_config.file_locks.write_lock(geoip_db_path); + + let input_source = InputSource { + url: geoip.url.clone(), + username: None, + password: None, + method: InputFetchMethod::GET, + headers: HashMap::default(), + }; + return match download_text_content(Arc::clone(&app_state.http_client.load()), &input_source, None, None).await { + Ok((content, _)) => { + let reader = Cursor::new(content); + let mut geoip = GeoIp::new(); + let result = { + match geoip.import_ipv4_from_csv(reader, geoip_db_path) { + Ok(size) => { + (Some(size), None) + } + Err(err) => (None, Some(err)) + } + }; + + return match result { + (Some(_), None) => { + app_state.geoip.store(Some(Arc::new(geoip))); + axum::http::StatusCode::OK.into_response() + }, + (None, Some(err)) => { + error!("Failed to process geoip db: {err}"); + axum::http::StatusCode::NOT_FOUND.into_response() + }, + _ => { + axum::http::StatusCode::INTERNAL_SERVER_ERROR.into_response() + } + } + } + Err(err) => { + error!("Failed to download geoip db: {err}"); + axum::http::StatusCode::BAD_REQUEST.into_response() + } + } + } + } + axum::http::StatusCode::BAD_REQUEST.into_response() +} + + async fn ipinfo(axum::extract::State(app_state): axum::extract::State>) -> axum::response::Response { if let Some((ipv4, ipv6)) = create_ipinfo_check(&app_state).await { let ipcheck = IpCheckDto { @@ -87,6 +145,7 @@ pub fn v1_api_register(web_auth_enabled: bool, app_state: Arc, web_ui_ router = router .route("/status", axum::routing::get(status)) .route("/streams", axum::routing::get(streams)) + .route("/geoip/update", axum::routing::get(geoip_update)) .route("/file/download", axum::routing::post(download_api::queue_download_file)) .route("/file/download/info", axum::routing::get(download_api::download_file_info)) .route("/ipinfo", axum::routing::get(ipinfo)); diff --git a/backend/src/api/main_api.rs b/backend/src/api/main_api.rs index e285a19e7..90b396cf0 100644 --- a/backend/src/api/main_api.rs +++ b/backend/src/api/main_api.rs @@ -32,6 +32,8 @@ use std::sync::atomic::AtomicI8; use std::sync::Arc; use tokio_util::sync::CancellationToken; use tower_governor::key_extractor::SmartIpKeyExtractor; +use crate::repository::storage::get_geoip_path; +use crate::utils::GeoIp; fn get_web_dir_path(web_ui_enabled: bool, web_root: &str) -> Result { let web_dir = web_root.to_string(); @@ -66,6 +68,20 @@ fn create_shared_data( forced_targets: &Arc, ) -> AppState { let config = app_config.config.load(); + + let use_geoip = config.is_geoip_enabled(); + let geoip = if use_geoip { + let path = get_geoip_path(&config.working_dir); + let _file_lock = app_config.file_locks.read_lock(&path); + let geoip = GeoIp::load(&path).ok(); + if geoip.is_some() { + info!("GeoIp db loaded"); + } + Arc::new(ArcSwapOption::from_pointee(geoip)) + } else { + Arc::new(ArcSwapOption::from(None)) + }; + let cache = create_cache(&config); let shared_stream_manager = Arc::new(SharedStreamManager::new()); let (provider_change_tx, provider_change_rx) = tokio::sync::mpsc::unbounded_channel(); @@ -75,6 +91,7 @@ fn create_shared_data( &config, &shared_stream_manager, &active_provider, + &geoip, active_user_change_tx, ); let event_manager = Arc::new(EventManager::new(active_user_change_rx, provider_change_rx, )); @@ -92,6 +109,7 @@ fn create_shared_data( event_manager, cancel_tokens: Arc::new(ArcSwap::from_pointee(CancelTokens::default())), playlists: Arc::new(PlaylistStorageState::new()), + geoip, } } diff --git a/backend/src/api/model/active_user_manager.rs b/backend/src/api/model/active_user_manager.rs index 28310c324..16dbde059 100644 --- a/backend/src/api/model/active_user_manager.rs +++ b/backend/src/api/model/active_user_manager.rs @@ -6,13 +6,14 @@ use crate::model::ProxyUserCredentials; use jsonwebtoken::get_current_timestamp; use log::{debug, error, info}; use shared::model::{ActiveUserConnectionChange, StreamChannel, StreamInfo, UserConnectionPermission}; -use shared::utils::{current_time_secs, default_grace_period_millis, default_grace_period_timeout_secs, sanitize_sensitive_info}; +use shared::utils::{current_time_secs, default_grace_period_millis, default_grace_period_timeout_secs, sanitize_sensitive_info, strip_port}; use std::collections::HashMap; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::Arc; +use arc_swap::ArcSwapOption; use tokio::sync::mpsc::{unbounded_channel, UnboundedSender}; use tokio::sync::RwLock; - +use crate::utils::GeoIp; const USER_GC_TTL: u64 = 900; // 15 Min const USER_CON_TTL: u64 = 10_800; // 3 hours @@ -54,26 +55,29 @@ macro_rules! active_user_manager_shared_impl { }; if let Some(username) = username_opt { - let mut user = self.user.write().await; - if let Some(connection_data) = user.get_mut(&username) { - if connection_data.connections > 0 { - connection_data.connections -= 1; - } + { + let mut user = self.user.write().await; + if let Some(connection_data) = user.get_mut(&username) { + if connection_data.connections > 0 { + connection_data.connections -= 1; + } - if connection_data.connections < connection_data.max_connections { - connection_data.granted_grace = false; - connection_data.grace_ts = 0; + if connection_data.connections < connection_data.max_connections { + connection_data.granted_grace = false; + connection_data.grace_ts = 0; + } + connection_data.streams.retain(|c| c.addr != addr); } - connection_data.streams.retain(|c| c.addr != addr); } - } - self.drop_connection(&addr); - self.shared_stream_manager.release_connection(addr, true).await; - self.provider_manager.release_connection(addr).await; - if let Err(err) = self.connection_change_tx.send(ActiveUserConnectionChange::Disconnected(addr.to_string())) { - error!("Failed to send active user connection change: {err:?}"); + + self.drop_connection(&addr); + self.shared_stream_manager.release_connection(addr, true).await; + self.provider_manager.release_connection(addr).await; + if let Err(err) = self.connection_change_tx.send(ActiveUserConnectionChange::Disconnected(addr.to_string())) { + error!("Failed to send active user connection change: {err:?}"); + } + self.log_active_user().await; } - self.log_active_user().await; } }; } @@ -188,12 +192,16 @@ pub struct ActiveUserManager { close_signal_tx: tokio::sync::broadcast::Sender, shared_stream_manager: Arc, provider_manager: Arc, + geo_ip: Arc>, connection_change_tx: ActiveUserConnectionChangeSender, release_tx: UnboundedSender, } impl ActiveUserManager { - pub fn new(config: &Config, shared_stream_manager: &Arc, provider_manager: &Arc, connection_change_tx: ActiveUserConnectionChangeSender) -> Arc { + pub fn new(config: &Config, shared_stream_manager: &Arc, + provider_manager: &Arc, + geoip: &Arc>, + connection_change_tx: ActiveUserConnectionChangeSender) -> Arc { let log_active_user: bool = 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); @@ -211,6 +219,7 @@ impl ActiveUserManager { close_signal_tx, shared_stream_manager: Arc::clone(shared_stream_manager), provider_manager: Arc::clone(provider_manager), + geo_ip: Arc::clone(geoip), connection_change_tx, release_tx: cleanup_tx, }); @@ -320,12 +329,22 @@ impl ActiveUserManager { } pub async fn add_connection(&self, username: &str, max_connections: u32, addr: &str, provider: &str, stream_channel: StreamChannel, user_agent: Cow<'_, str>) -> UserConnectionGuard { + let country = { + let geoip = self.geo_ip.load(); + if let Some(geoip_db) = (*geoip).as_ref() { + geoip_db.lookup(&strip_port(addr)) + } else { + None + } + }; + let stream_info = StreamInfo::new( username, addr, provider, stream_channel, user_agent.to_string(), + country, ); { let mut user_map = self.user.write().await; diff --git a/backend/src/api/model/app_state.rs b/backend/src/api/model/app_state.rs index 5dba2c87a..fcdc49fa0 100644 --- a/backend/src/api/model/app_state.rs +++ b/backend/src/api/model/app_state.rs @@ -24,6 +24,8 @@ use std::sync::Arc; use std::time::Duration; use tokio::sync::Mutex; use tokio_util::sync::CancellationToken; +use crate::repository::storage::get_geoip_path; +use crate::utils::GeoIp; macro_rules! cancel_service { ($field: ident, $changes:expr, $cancel_tokens:expr) => { @@ -58,16 +60,18 @@ struct TargetChanges { target: Arc, } +#[allow(clippy::struct_excessive_bools)] pub(in crate::api) struct UpdateChanges { scheduler: bool, hdhomerun: bool, file_watch: bool, + geoip: bool, targets: Option>, } impl UpdateChanges { pub(in crate::api) fn modified(&self) -> bool { - self.scheduler || self.hdhomerun || self.file_watch + self.scheduler || self.hdhomerun || self.file_watch || self.geoip } } @@ -258,6 +262,7 @@ pub struct AppState { pub event_manager: Arc, pub cancel_tokens: Arc>, pub playlists: Arc, + pub geoip: Arc> } impl AppState { @@ -267,12 +272,29 @@ impl AppState { ) -> Result { let changes = self.detect_changes_for_config(&config); config.update_runtime(); + + let use_geoip = config.is_geoip_enabled(); + let working_dir = config.working_dir.clone(); + self.active_users.update_config(&config); self.app_config.set_config(config)?; self.active_provider .update_config(&self.app_config) .await; self.update_config().await; + + if changes.geoip { + let new_geoip = if use_geoip { + let path = get_geoip_path(&working_dir); + let _file_lock = self.app_config.file_locks.read_lock(&path); + GeoIp::load(&path).ok().map(Arc::new) + } else { + None + }; + + self.geoip.store(new_geoip); + } + Ok(changes) } @@ -340,11 +362,15 @@ impl AppState { config.mapping_path.as_ref() ); + let geoip_enabled = config.is_geoip_enabled(); + let geoip_enabled_old = old_config.is_geoip_enabled(); + UpdateChanges { scheduler: changed_schedules, hdhomerun: changed_hdhomerun, file_watch: changed_file_watch, targets: None, + geoip: geoip_enabled != geoip_enabled_old, } } @@ -412,6 +438,7 @@ impl AppState { scheduler: false, hdhomerun: false, file_watch: file_watch_changed, + geoip: false, targets: Some(target_changes), } } diff --git a/backend/src/api/model/provider_config.rs b/backend/src/api/model/provider_config.rs index da900e4d4..97a812fac 100644 --- a/backend/src/api/model/provider_config.rs +++ b/backend/src/api/model/provider_config.rs @@ -261,13 +261,12 @@ impl ProviderConfig { pub async fn release(&self) { let mut guard = self.connection.write().await; if guard.current_connections > 0 { + if guard.current_connections == 1 && self.max_connections > 1 { + guard.granted_grace = false; + guard.grace_ts = 0; + } modify_connections!(self, guard, -1); } - - if guard.current_connections == 0 || guard.current_connections < self.max_connections { - guard.granted_grace = false; - guard.grace_ts = 0; - } } #[inline] diff --git a/backend/src/model/config/base.rs b/backend/src/model/config/base.rs index b9f60aeec..aad2bfae4 100644 --- a/backend/src/model/config/base.rs +++ b/backend/src/model/config/base.rs @@ -117,6 +117,10 @@ impl Config { devices: hdhr.devices.iter().map(|d| d.name.clone()).collect::>(), }) } + + pub fn is_geoip_enabled(&self) -> bool { + self.reverse_proxy.as_ref().is_some_and(|r| r.geoip.as_ref().is_some_and(|g| g.enabled)) + } } macros::from_impl!(Config); diff --git a/backend/src/model/config/geoip.rs b/backend/src/model/config/geoip.rs new file mode 100644 index 000000000..88d44c611 --- /dev/null +++ b/backend/src/model/config/geoip.rs @@ -0,0 +1,28 @@ +use shared::model::GeoIpConfigDto; +use crate::model::macros; + +#[derive(Debug, Clone)] +pub struct GeoIpConfig { + pub(crate) enabled: bool, + pub(crate) url: String, +} + +macros::from_impl!(GeoIpConfig); + +impl From<&GeoIpConfigDto> for GeoIpConfig { + fn from(dto: &GeoIpConfigDto) -> Self { + Self { + enabled: dto.enabled, + url: dto.url.clone(), + } + } +} + +impl From<&GeoIpConfig> for GeoIpConfigDto { + fn from(instance: &GeoIpConfig) -> Self { + Self { + enabled: instance.enabled, + url: instance.url.clone(), + } + } +} diff --git a/backend/src/model/config/mod.rs b/backend/src/model/config/mod.rs index 1cf786415..953414ea5 100644 --- a/backend/src/model/config/mod.rs +++ b/backend/src/model/config/mod.rs @@ -28,6 +28,7 @@ mod api_user; mod trakt_api; pub mod macros; mod favourites; +mod geoip; pub use api::*; pub use api_proxy::*; @@ -46,6 +47,7 @@ pub use proxy::*; pub use rate_limit::*; pub use rename::*; pub use reverse_proxy::*; +pub use geoip::*; pub use schedule::*; pub use sort::*; pub use source::*; diff --git a/backend/src/model/config/reverse_proxy.rs b/backend/src/model/config/reverse_proxy.rs index c7df054fd..18184f6dc 100644 --- a/backend/src/model/config/reverse_proxy.rs +++ b/backend/src/model/config/reverse_proxy.rs @@ -1,6 +1,6 @@ use shared::model::ReverseProxyConfigDto; use crate::model::config::cache::CacheConfig; -use crate::model::{macros, RateLimitConfig, StreamConfig}; +use crate::model::{macros, GeoIpConfig, RateLimitConfig, StreamConfig}; #[derive(Debug, Clone)] pub struct ReverseProxyConfig { @@ -9,6 +9,7 @@ pub struct ReverseProxyConfig { pub stream: Option, pub cache: Option, pub rate_limit: Option, + pub geoip: Option, } macros::from_impl!(ReverseProxyConfig); @@ -21,6 +22,7 @@ impl From<&ReverseProxyConfigDto> for ReverseProxyConfig { stream: dto.stream.as_ref().map(Into::into), cache: dto.cache.as_ref().map(Into::into), rate_limit: dto.rate_limit.as_ref().map(Into::into), + geoip: dto.geoip.as_ref().map(Into::into), } } } @@ -33,6 +35,7 @@ impl From<&ReverseProxyConfig> for ReverseProxyConfigDto { stream: instance.stream.as_ref().map(Into::into), cache: instance.cache.as_ref().map(Into::into), rate_limit: instance.rate_limit.as_ref().map(Into::into), + geoip: instance.geoip.as_ref().map(Into::into), } } } diff --git a/backend/src/repository/storage.rs b/backend/src/repository/storage.rs index 200e58758..b964b706c 100644 --- a/backend/src/repository/storage.rs +++ b/backend/src/repository/storage.rs @@ -32,3 +32,7 @@ pub fn get_input_storage_path(input_name: &str, working_dir: &str) -> std::io::R // Create the directory and return the path or propagate the error std::fs::create_dir_all(&path).map(|()| path) } + +pub fn get_geoip_path(working_dir: &str) -> PathBuf { + Path::new(working_dir).join("geoip.db") +} \ No newline at end of file diff --git a/backend/src/utils/file/config_reader.rs b/backend/src/utils/file/config_reader.rs index 95397a958..d78595eaa 100644 --- a/backend/src/utils/file/config_reader.rs +++ b/backend/src/utils/file/config_reader.rs @@ -82,14 +82,14 @@ pub fn read_sources_file(sources_file: &str, resolve_env: bool, include_computed } } -pub fn read_config_file(config_file: &str, resolve_env: bool) -> Result { +pub fn read_config_file(config_file: &str, resolve_env: bool, include_computed: 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) => { if resolve_env { - config.prepare()?; + config.prepare(include_computed)?; } Ok(config) } @@ -107,7 +107,7 @@ pub fn read_app_config_dto(paths: &ConfigPaths, let sources_file = paths.sources_file_path.as_str(); let api_proxy_file = paths.api_proxy_file_path.as_str(); - let config = read_config_file(config_file, resolve_env)?; + let config = read_config_file(config_file, resolve_env, include_computed)?; let sources = read_sources_file(sources_file, resolve_env, include_computed, config.get_hdhr_device_overview().as_ref())?; let mappings = if let Some(mappings_file) = paths.mapping_file_path.as_ref() { read_mappings_file(mappings_file, resolve_env).unwrap_or(None) @@ -211,7 +211,7 @@ pub fn read_initial_app_config(paths: &mut ConfigPaths, let config_file = paths.config_file_path.as_str(); let sources_file = paths.sources_file_path.as_str(); - let config_dto = read_config_file(config_file, resolve_env)?; + let config_dto = read_config_file(config_file, resolve_env, include_computed)?; let mut sources_dto = read_sources_file(sources_file, resolve_env, include_computed, config_dto.get_hdhr_device_overview().as_ref())?; prepare_sources_batch(&mut sources_dto, include_computed)?; let sources: SourcesConfig = SourcesConfig::try_from(sources_dto)?; diff --git a/backend/src/utils/geoip.rs b/backend/src/utils/geoip.rs index 5b9a09515..1eb7bab6c 100644 --- a/backend/src/utils/geoip.rs +++ b/backend/src/utils/geoip.rs @@ -14,6 +14,7 @@ pub struct GeoIp { tree: BPlusTree, } + impl GeoIp { pub fn load(path: &Path) -> io::Result { @@ -55,6 +56,12 @@ impl GeoIp { } } +impl Default for GeoIp { + fn default() -> Self { + Self::new() + } +} + #[cfg(test)] mod test { // https://raw.githubusercontent.com/sapics/ip-location-db/refs/heads/main/asn-country/asn-country-ipv4.csv diff --git a/backend/src/utils/mod.rs b/backend/src/utils/mod.rs index d0477a11d..066dee31f 100644 --- a/backend/src/utils/mod.rs +++ b/backend/src/utils/mod.rs @@ -15,6 +15,7 @@ pub use self::bincode_utils::*; pub use self::logging::*; pub use self::trakt::*; pub use self::telegram::*; +pub use self::geoip::*; #[macro_export] macro_rules! debug_if_enabled { diff --git a/frontend/public/assets/i18n/en.json b/frontend/public/assets/i18n/en.json index 387d298f7..28ba00f54 100644 --- a/frontend/public/assets/i18n/en.json +++ b/frontend/public/assets/i18n/en.json @@ -297,7 +297,10 @@ "DURATION": "Duration", "ADD_QUALITY_TO_FILENAME": "Quality", "SSDP_DISCOVERY": "SSDP Discovery", - "PROPRIETARY_DISCOVERY": "Proprietary Discovery" + "PROPRIETARY_DISCOVERY": "Proprietary Discovery", + "COUNTRY": "Country", + "GEOIP": "Geo-IP", + "UPDATE_GEOIP_DB": "Update Geo-IP db" }, "TITLE": { "USER_BOUQUET_EDITOR": "User group editor" @@ -321,6 +324,10 @@ }, "SERVER_CONFIG": { "FAIL": "Failed to download server config!" + }, + "GEOIP": { + "SUCCESS": "Successfully downloaded Geo-IP db", + "FAIL": "Failed to download Geo-IP db!" } }, "LOGIN": { diff --git a/frontend/src/app/components/config/config_view.rs b/frontend/src/app/components/config/config_view.rs index 50d582376..cf82fdabd 100644 --- a/frontend/src/app/components/config/config_view.rs +++ b/frontend/src/app/components/config/config_view.rs @@ -28,6 +28,8 @@ const LABEL_CONFIG: &str = "LABEL.CONFIG"; const LABEL_EDIT: &str = "LABEL.EDIT"; const LABEL_VIEW: &str = "LABEL.VIEW"; const LABEL_SAVE: &str = "LABEL.SAVE"; +const LABEL_UPDATE_GEOIP: &str = "LABEL.UPDATE_GEOIP_DB"; + // ========================== macro_rules! collect_modified { @@ -162,7 +164,7 @@ pub fn ConfigView() -> Html { let mut config_dto = config_ctx.config.as_ref().map_or_else(ConfigDto::default, |app_cfg| app_cfg.config.clone()); update_config(&mut config_dto, modified_forms); - match config_dto.prepare() { + match config_dto.prepare(false) { Ok(_) => { let services = services.clone(); let translate = translate.clone(); @@ -214,6 +216,22 @@ pub fn ConfigView() -> Html { }; + let handle_update_geoip = { + let services = services_ctx.clone(); + let translate = translate.clone(); + Callback::from(move |_| { + let services = services.clone(); + let translate = translate.clone(); + wasm_bindgen_futures::spawn_local(async move { + match services.config.update_geoip().await { + Ok(_) => services.toastr.success(translate.t("MESSAGES.DOWNLOAD.GEOIP.SUCCESS")), + Err(_err) => services.toastr.error(translate.t("MESSAGES.DOWNLOAD.GEOIP.FAIL")), + } + }); + }) + }; + + let context = ConfigViewContext { edit_mode: edit_mode.clone(), on_form_change: on_form_change.clone(), @@ -224,11 +242,18 @@ pub fn ConfigView() -> Html {

{ translate.t(LABEL_CONFIG) }

- + })} + +
diff --git a/frontend/src/app/components/config/hdhomerun_config_view.rs b/frontend/src/app/components/config/hdhomerun_config_view.rs index f1fb6d666..ecd0d7b2f 100644 --- a/frontend/src/app/components/config/hdhomerun_config_view.rs +++ b/frontend/src/app/components/config/hdhomerun_config_view.rs @@ -86,7 +86,7 @@ pub fn HdHomerunConfigView() -> Html { .unwrap_or(8901) + 1; new_device.port = next_port; new_device.name = format!("hdhr_{next_port}"); - if let Err(err) = new_device.prepare(devices.len() as u8) { + if let Err(err) = new_device.prepare(devices.len() as u8, false) { error!("Failed to prepare hdhr device: {err}"); } devices.push(new_device); diff --git a/frontend/src/app/components/config/reverse_proxy_config_view.rs b/frontend/src/app/components/config/reverse_proxy_config_view.rs index a1149c4b3..76ff75caf 100644 --- a/frontend/src/app/components/config/reverse_proxy_config_view.rs +++ b/frontend/src/app/components/config/reverse_proxy_config_view.rs @@ -1,15 +1,12 @@ use yew::prelude::*; use yew_i18n::use_translation; -use shared::model::{ - CacheConfigDto, RateLimitConfigDto, StreamConfigDto, ReverseProxyConfigDto, -}; +use shared::model::{CacheConfigDto, RateLimitConfigDto, StreamConfigDto, ReverseProxyConfigDto, GeoIpConfigDto}; use crate::app::context::ConfigContext; use crate::app::components::config::config_view_context::ConfigViewContext; use crate::app::components::config::config_page::ConfigForm; use crate::app::components::config::macros::HasFormData; -use crate::app::components::Card; -use crate::{config_field, config_field_bool, config_field_optional, - edit_field_bool, edit_field_number, edit_field_number_u64, edit_field_text_option, generate_form_reducer}; +use crate::app::components::{Card}; +use crate::{config_field, config_field_bool, config_field_optional, edit_field_bool, edit_field_number, edit_field_number_u64, edit_field_text, edit_field_text_option, generate_form_reducer}; const LABEL_CACHE: &str = "LABEL.CACHE"; const LABEL_ENABLED: &str = "LABEL.ENABLED"; @@ -30,6 +27,8 @@ const LABEL_BURST_SIZE: &str = "LABEL.BURST_SIZE"; const LABEL_RESOURCE_REWRITE_DISABLED: &str = "LABEL.RESOURCE_REWRITE_DISABLED"; const LABEL_DISABLE_REFERER_HEADER: &str = "LABEL.DISABLE_REFERER_HEADER"; +const LABEL_GEOIP: &str = "LABEL.GEOIP"; +const LABEL_URL: &str = "LABEL.URL"; generate_form_reducer!( state: CacheConfigFormState { form: CacheConfigDto }, @@ -64,6 +63,15 @@ generate_form_reducer!( } ); +generate_form_reducer!( + state: GeoIpConfigFormState { form: GeoIpConfigDto }, + action_name: GeoIpConfigFormAction, + fields { + Enabled => enabled: bool, + Url => url: String, + } +); + generate_form_reducer!( state: ReverseProxyConfigFormState { form: ReverseProxyConfigDto }, action_name: ReverseProxyConfigFormAction, @@ -92,22 +100,28 @@ pub fn ReverseProxyConfigView() -> Html { StreamConfigFormState { form: StreamConfigDto::default(), modified: false } }); + let geoip_state: UseReducerHandle = use_reducer(|| { + GeoIpConfigFormState { form: GeoIpConfigDto::default(), modified: false } + }); + { let on_form_change = config_view_ctx.on_form_change.clone(); let reverse_proxy_state = reverse_proxy_state.clone(); let cache_state = cache_state.clone(); let rate_limit_state = rate_limit_state.clone(); let stream_state = stream_state.clone(); + let geoip_state = geoip_state.clone(); use_effect_with( - (reverse_proxy_state, cache_state, rate_limit_state, stream_state), - move |(rp, cache, rl, stream)| { + (reverse_proxy_state, cache_state, rate_limit_state, stream_state, geoip_state), + move |(rp, cache, rl, stream, geoip)| { let mut form = rp.form.clone(); form.cache = Some(cache.form.clone()); form.rate_limit = Some(rl.form.clone()); form.stream = Some(stream.form.clone()); + form.geoip = Some(geoip.form.clone()); - let modified = rp.modified || cache.modified || rl.modified || stream.modified; + let modified = rp.modified || cache.modified || rl.modified || stream.modified || geoip.modified; on_form_change.emit(ConfigForm::ReverseProxy(modified, form)); }, ); @@ -118,6 +132,7 @@ pub fn ReverseProxyConfigView() -> Html { let cache_state = cache_state.clone(); let rate_limit_state = rate_limit_state.clone(); let stream_state = stream_state.clone(); + let geoip_state = geoip_state.clone(); let reverse_proxy_cfg = config_ctx.config.as_ref().and_then(|c| c.config.reverse_proxy.clone()); use_effect_with((reverse_proxy_cfg, config_view_ctx.edit_mode.clone()), move |(cfg, _mode)| { @@ -126,11 +141,13 @@ pub fn ReverseProxyConfigView() -> Html { cache_state.dispatch(CacheConfigFormAction::SetAll(rp.cache.as_ref().map_or_else(CacheConfigDto::default, |c| c.clone()))); rate_limit_state.dispatch(RateLimitConfigFormAction::SetAll(rp.rate_limit.as_ref().map_or_else(RateLimitConfigDto::default, |rl| rl.clone()))); stream_state.dispatch(StreamConfigFormAction::SetAll(rp.stream.as_ref().map_or_else(StreamConfigDto::default, |s| s.clone()))); + geoip_state.dispatch(GeoIpConfigFormAction::SetAll(rp.geoip.as_ref().map_or_else(GeoIpConfigDto::default, |s| s.clone()))); } else { reverse_proxy_state.dispatch(ReverseProxyConfigFormAction::SetAll(ReverseProxyConfigDto::default())); cache_state.dispatch(CacheConfigFormAction::SetAll(CacheConfigDto::default())); rate_limit_state.dispatch(RateLimitConfigFormAction::SetAll(RateLimitConfigDto::default())); stream_state.dispatch(StreamConfigFormAction::SetAll(StreamConfigDto::default())); + geoip_state.dispatch(GeoIpConfigFormAction::SetAll(GeoIpConfigDto::default())); } || () }); @@ -171,6 +188,16 @@ pub fn ReverseProxyConfigView() -> Html { } }; + let render_geoip = || { + html! { + +

{translate.t(LABEL_GEOIP)}

+ { config_field_bool!(geoip_state.form, translate.t(LABEL_ENABLED), enabled) } + { config_field!(geoip_state.form, translate.t(LABEL_URL), url) } + + } + }; + let render_view_mode = || { html! { <> @@ -182,6 +209,7 @@ pub fn ReverseProxyConfigView() -> Html { { render_cache() } { render_rate_limit() } { render_stream() } + { render_geoip() }
} @@ -215,6 +243,11 @@ pub fn ReverseProxyConfigView() -> Html { { edit_field_number!(stream_state, translate.t(LABEL_FORCED_RETRY_INTERVAL_SECS), forced_retry_interval_secs, StreamConfigFormAction::ForcedRetryIntervalSecs) } { edit_field_number_u64!(stream_state, translate.t(LABEL_THROTTLE_KBPS), throttle_kbps, StreamConfigFormAction::ThrottleKbps) } + +

{translate.t(LABEL_GEOIP)}

+ { edit_field_bool!(geoip_state, translate.t(LABEL_ENABLED), enabled, GeoIpConfigFormAction::Enabled) } + { edit_field_text!(geoip_state, translate.t(LABEL_URL), url, GeoIpConfigFormAction::Url) } +
}; diff --git a/frontend/src/app/components/dashboard/streams_table.rs b/frontend/src/app/components/dashboard/streams_table.rs index c813c93fb..1159f7fc5 100644 --- a/frontend/src/app/components/dashboard/streams_table.rs +++ b/frontend/src/app/components/dashboard/streams_table.rs @@ -1,4 +1,3 @@ -use std::borrow::Cow; use crate::app::components::menu_item::MenuItem; use crate::app::components::popup_menu::PopupMenu; use crate::app::components::{AppIcon, Table, TableDefinition, ToggleSwitch}; @@ -10,14 +9,13 @@ use std::rc::Rc; use std::str::FromStr; use gloo_timers::callback::Interval; use gloo_utils::window; -use log::debug; use wasm_bindgen::JsCast; use web_sys::Element; use yew::prelude::*; use yew_i18n::use_translation; -use shared::utils::current_time_secs; +use shared::utils::{current_time_secs, strip_port}; -const HEADERS: [&str; 11] = [ +const HEADERS: [&str; 12] = [ "LABEL.EMPTY", "LABEL.USERNAME", "LABEL.STREAM_ID", @@ -25,36 +23,13 @@ const HEADERS: [&str; 11] = [ "LABEL.CHANNEL", "LABEL.GROUP", "LABEL.CLIENT_IP", + "LABEL.COUNTRY", "LABEL.PROVIDER", "LABEL.SHARED", "LABEL.USER_AGENT", "LABEL.DURATION" ]; -pub fn strip_port<'a>(input: &'a str) -> Cow<'a, str> { - // IPv6 with port: [2001:db8::1]:8080 - if let Some(stripped) = input.strip_prefix('[') { - if let Some(end) = stripped.find(']') { - return Cow::Owned(stripped[..end].to_string()); - } - // Invalid IPv6 - return Cow::Borrowed(input); - } - - // IPv4 or IPv6 without bracket - if let Some((left, right)) = input.rsplit_once(':') { - // If `left` has a colon then its IPv6 without port. - if left.contains(':') { - Cow::Borrowed(input) - } else { - // IPv4:Port - Cow::Owned(left.to_string()) - } - } else { - Cow::Borrowed(input) - } -} - pub fn format_duration(seconds: u64) -> String { let hours = seconds / 3600; let minutes = (seconds % 3600) / 60; @@ -160,10 +135,11 @@ pub fn StreamsTable(props: &StreamsTableProps) -> Html { 4 => html! {dto.channel.title.as_str()}, 5 => html! {dto.channel.group.as_str()}, 6 => html! { strip_port(&dto.addr)}, - 7 => html! {dto.provider.as_str()}, - 8 => html! { }, - 9 => html! { dto.user_agent.as_str() }, - 10 => html! { {format_duration(dto.ts)} }, + 7 => html! { dto.country.as_ref().map_or_else(String::new, |c| c.clone()) }, + 8 => html! {dto.provider.as_str()}, + 9 => html! { }, + 10 => html! { dto.user_agent.as_str() }, + 11 => html! { {format_duration(dto.ts)} }, _ => html! {""}, } }) diff --git a/frontend/src/services/config_service.rs b/frontend/src/services/config_service.rs index c52f19ab0..305dbc425 100644 --- a/frontend/src/services/config_service.rs +++ b/frontend/src/services/config_service.rs @@ -21,6 +21,7 @@ pub struct ConfigService { config_path: String, ip_check_path: String, batch_input_content_path: String, + geoip_path: String, event_service: Rc } @@ -35,6 +36,7 @@ impl ConfigService { config_path: concat_path_leading_slash(&base_href, "api/v1/config"), ip_check_path: concat_path_leading_slash(&base_href, "api/v1/ipinfo"), batch_input_content_path: concat_path_leading_slash(&base_href, "api/v1/config/batchContent"), + geoip_path: concat_path_leading_slash(&base_href, "api/v1/geoip/update"), event_service } } @@ -142,4 +144,8 @@ impl ConfigService { } } + pub async fn update_geoip(&self) -> Result, Error> { + request_get::<()>(&self.geoip_path, None, None).await + } + } diff --git a/shared/src/model/config/base.rs b/shared/src/model/config/base.rs index 07b790220..621b52523 100644 --- a/shared/src/model/config/base.rs +++ b/shared/src/model/config/base.rs @@ -148,7 +148,7 @@ pub struct HdHomeRunDeviceOverview { } impl ConfigDto { - pub fn prepare(&mut self) -> Result<(), TuliproxError> { + pub fn prepare(&mut self, include_computed: bool) -> Result<(), TuliproxError> { if let Some(mins) = self.sleep_timer_mins { if mins == 0 { return Err(TuliproxError::new(TuliproxErrorKind::Info, "`sleep_timer_mins` must be > 0 when specified".to_string())); @@ -157,7 +157,7 @@ impl ConfigDto { self.api.prepare(); self.prepare_web()?; - self.prepare_hdhomerun()?; + self.prepare_hdhomerun(include_computed)?; self.prepare_video_config()?; if let Some(reverse_proxy) = self.reverse_proxy.as_mut() { @@ -180,10 +180,10 @@ impl ConfigDto { Ok(()) } - fn prepare_hdhomerun(&mut self) -> Result<(), TuliproxError> { + fn prepare_hdhomerun(&mut self, include_computed: bool) -> Result<(), TuliproxError> { if let Some(hdhomerun) = &mut self.hdhomerun { if hdhomerun.enabled { - hdhomerun.prepare(self.api.port)?; + hdhomerun.prepare(self.api.port, include_computed)?; } } Ok(()) @@ -251,4 +251,9 @@ impl ConfigDto { self.accept_insecure_ssl_certificates = main_config.accept_insecure_ssl_certificates; } + + pub fn is_geoip_enabled(&self) -> bool { + self.reverse_proxy.as_ref().is_some_and(|r| r.geoip.as_ref().is_some_and(|g| g.enabled)) + } + } \ No newline at end of file diff --git a/shared/src/model/config/geoip.rs b/shared/src/model/config/geoip.rs new file mode 100644 index 000000000..1acf37b55 --- /dev/null +++ b/shared/src/model/config/geoip.rs @@ -0,0 +1,11 @@ + +fn default_geoip_url() -> String { String::from("https://raw.githubusercontent.com/sapics/ip-location-db/refs/heads/main/asn-country/asn-country-ipv4.csv") } + +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, Default, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct GeoIpConfigDto { + #[serde(default)] + pub enabled: bool, + #[serde(default = "default_geoip_url")] + pub url: String, +} diff --git a/shared/src/model/config/hdhomerun.rs b/shared/src/model/config/hdhomerun.rs index 706b48ebd..e0cddcbbf 100644 --- a/shared/src/model/config/hdhomerun.rs +++ b/shared/src/model/config/hdhomerun.rs @@ -1,8 +1,8 @@ -use std::collections::HashSet; -use log::warn; use crate::create_tuliprox_error_result; use crate::error::{TuliproxError, TuliproxErrorKind}; -use crate::utils::{default_as_true, generate_hdhr_device_id, generate_hdhr_device_id_from_base, validate_hdhr_device_id, hash_string, hex_encode}; +use crate::utils::{default_as_true, generate_hdhr_device_id, generate_hdhr_device_id_from_base, hash_string, hex_encode, validate_hdhr_device_id}; +use log::warn; +use std::collections::HashSet; fn default_friendly_name() -> String { String::from("TuliproxTV") } fn default_manufacturer() -> String { String::from("Silicondust") } @@ -60,11 +60,13 @@ impl Default for HdHomeRunDeviceConfigDto { } impl HdHomeRunDeviceConfigDto { - pub fn prepare(&mut self, device_num: u8) -> Result<(), TuliproxError> { + pub fn prepare(&mut self, device_num: u8, include_computed: bool) -> Result<(), TuliproxError> { self.name = self.name.trim().to_string(); if self.name.is_empty() { self.name = format!("device{device_num}"); - warn!("Device name empty, assigned new name: {}", self.name); + if include_computed { + warn!("Device name empty, assigned new name: {}", self.name); + } } if self.tuner_count == 0 { @@ -86,7 +88,9 @@ impl HdHomeRunDeviceConfigDto { let p4 = hex_encode(&hash[8..10]); let p5 = hex_encode(&hash[10..16]); self.device_udn = format!("{p1}-{p2}-{p3}-{p4}-{p5}"); - warn!("HDHomeRun device '{}' is missing a unique device_udn. A new one has been generated: {}", self.name, self.device_udn); + if include_computed { + warn!("HDHomeRun device '{}' is missing a unique device_udn. A new one has been generated: {}", self.name, self.device_udn); + } } else { // Ensure only the UUID part is stored. if let Some(uuid_part) = self.device_udn.strip_prefix("uuid:") { @@ -97,11 +101,15 @@ impl HdHomeRunDeviceConfigDto { // --- Device ID Logic --- if self.device_id.is_empty() { self.device_id = generate_hdhr_device_id(); - warn!("HDHomeRun device '{}' is missing a device_id. A new one has been generated: {}", self.name, self.device_id); + if include_computed { + warn!("HDHomeRun device '{}' is missing a device_id. A new one has been generated: {}", self.name, self.device_id); + } } else if !validate_hdhr_device_id(&self.device_id) { let old_id = self.device_id.clone(); self.device_id = generate_hdhr_device_id_from_base(&self.device_id); - warn!("HDHomeRun device '{}' has an invalid device_id '{}'. A valid one has been generated: {}", self.name, old_id, self.device_id); + if include_computed { + warn!("HDHomeRun device '{}' has an invalid device_id '{}'. A valid one has been generated: {}", self.name, old_id, self.device_id); + } } Ok(()) } @@ -132,13 +140,13 @@ impl HdHomeRunConfigDto { // It's only included to satisfy the frontend compiler. } - pub fn prepare(&mut self, api_port: u16) -> Result<(), TuliproxError> { + pub fn prepare(&mut self, api_port: u16, include_computed: bool) -> Result<(), TuliproxError> { let mut names = HashSet::new(); let mut ports = HashSet::new(); let mut device_ids = HashSet::new(); ports.insert(api_port); for (device_num, device) in (0_u8..).zip(self.devices.iter_mut()) { - device.prepare(device_num)?; + device.prepare(device_num, include_computed)?; if !names.insert(device.name.clone()) { return create_tuliprox_error_result!(TuliproxErrorKind::Info, "HdHomeRun duplicate device name {}", device.name); } @@ -153,10 +161,10 @@ impl HdHomeRunConfigDto { for device in &mut self.devices { if device.port == 0 { while ports.contains(¤t_port) || current_port == 0 { - current_port = current_port.wrapping_add(1); - if current_port == api_port { // full cycle guard - return create_tuliprox_error_result!(TuliproxErrorKind::Info, "No free port available for HdHomeRun devices"); - } + current_port = current_port.wrapping_add(1); + if current_port == api_port { // full cycle guard + return create_tuliprox_error_result!(TuliproxErrorKind::Info, "No free port available for HdHomeRun devices"); + } } device.port = current_port; diff --git a/shared/src/model/config/mod.rs b/shared/src/model/config/mod.rs index 6025f6f5d..0b10eda52 100644 --- a/shared/src/model/config/mod.rs +++ b/shared/src/model/config/mod.rs @@ -33,6 +33,7 @@ mod proxy_type; mod proxy_user_status; mod favourites; +mod geoip; pub use proxy_type::*; pub use proxy_user_status::*; @@ -58,6 +59,7 @@ pub use epg::*; pub use cache::*; pub use rate_limit::*; pub use reverse_proxy::*; +pub use geoip::*; pub use proxy::*; pub use trakt::*; pub use rename::*; diff --git a/shared/src/model/config/reverse_proxy.rs b/shared/src/model/config/reverse_proxy.rs index b3bb4b262..f188762d2 100644 --- a/shared/src/model/config/reverse_proxy.rs +++ b/shared/src/model/config/reverse_proxy.rs @@ -1,5 +1,5 @@ use crate::error::TuliproxError; -use crate::model::{CacheConfigDto, RateLimitConfigDto, StreamConfigDto}; +use crate::model::{CacheConfigDto, GeoIpConfigDto, RateLimitConfigDto, StreamConfigDto}; use log::warn; #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, Default, PartialEq)] @@ -15,6 +15,9 @@ pub struct ReverseProxyConfigDto { pub cache: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub rate_limit: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub geoip: Option, + } impl ReverseProxyConfigDto { diff --git a/shared/src/model/stream_info.rs b/shared/src/model/stream_info.rs index 2b0ed889d..80ad077a1 100644 --- a/shared/src/model/stream_info.rs +++ b/shared/src/model/stream_info.rs @@ -51,10 +51,11 @@ pub struct StreamInfo { pub addr: String, pub user_agent: String, pub ts: u64, + pub country: Option, } impl StreamInfo { - pub fn new(username: &str, addr: &str, provider: &str, stream_channel: StreamChannel, user_agent: String) -> Self { + pub fn new(username: &str, addr: &str, provider: &str, stream_channel: StreamChannel, user_agent: String, country: Option) -> Self { Self { username: username.to_string(), channel: stream_channel, @@ -62,6 +63,7 @@ impl StreamInfo { addr: addr.to_string(), user_agent, ts: current_time_secs(), + country, } } } \ No newline at end of file diff --git a/shared/src/utils/mod.rs b/shared/src/utils/mod.rs index a126cbd92..1d18e8550 100644 --- a/shared/src/utils/mod.rs +++ b/shared/src/utils/mod.rs @@ -9,6 +9,7 @@ mod hash_utils; mod json_utils; mod serde_utils; mod hdhomerun_utils; +mod net_utils; use std::fmt::Display; pub use self::default_utils::*; @@ -22,6 +23,7 @@ pub use self::hash_utils::*; pub use self::json_utils::*; pub use self::serde_utils::*; pub use self::hdhomerun_utils::*; +pub use self::net_utils::*; #[macro_export] macro_rules! write_if_some { diff --git a/shared/src/utils/net_utils.rs b/shared/src/utils/net_utils.rs new file mode 100644 index 000000000..5af64fce9 --- /dev/null +++ b/shared/src/utils/net_utils.rs @@ -0,0 +1,25 @@ +use std::borrow::Cow; + +pub fn strip_port<'a>(input: &'a str) -> Cow<'a, str> { + // IPv6 with port: [2001:db8::1]:8080 + if let Some(stripped) = input.strip_prefix('[') { + if let Some(end) = stripped.find(']') { + return Cow::Owned(stripped[..end].to_string()); + } + // Invalid IPv6 + return Cow::Borrowed(input); + } + + // IPv4 or IPv6 without bracket + if let Some((left, _right)) = input.rsplit_once(':') { + // If `left` has a colon then its IPv6 without port. + if left.contains(':') { + Cow::Borrowed(input) + } else { + // IPv4:Port + Cow::Owned(left.to_string()) + } + } else { + Cow::Borrowed(input) + } +} \ No newline at end of file