diff --git a/CHANGELOG.md b/CHANGELOG.md index 5a5903c0d..23bc1b032 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -70,7 +70,7 @@ connections (e.g., background tasks or standard users) when provider slots are full. - **Background Metadata Queue**: Metadata resolution (VOD/Series) and stream analysis are now queued per input and processed in the background when provider connections are idle. This prevents "No Connections" errors for active users during playlist updates. -- **Live TV Probing**: Added support for probing Live TV streams (`probe_live`) to determine codecs and resolution. This runs as a low-priority +- **Stream Probing**: Added support for probing streams (`probe_live|vod|series`) to determine codecs and resolution. This runs as a low-priority background task. - **Discord Notifications**: Support for Discord notifications via webhooks with optional Handlebars templates. - **Enhanced REST Messaging**: Support for custom HTTP methods, headers, and Handlebars templating. @@ -104,6 +104,7 @@ active URL of the specified provider. - **source.yml (input options)**: - Added `resolve_tmdb`: Triggers TMDB lookup if ID is missing. - Added `probe_stream`: Triggers ffprobe if technical info is missing. + - Added `probe_delay`: Delay between probe tasks (default `50` seconds). - **source.yml (target output)**: - Added `probe_live`: Enables background probing for Live TV streams (default disabled). - Added `probe_live_interval_hours`: Sets the frequency for re-probing Live TV streams. diff --git a/README.md b/README.md index 7a1933676..6cd7c9cd3 100644 --- a/README.md +++ b/README.md @@ -1111,7 +1111,8 @@ Each input has the following attributes: - `probe_vod`: `true`|`false` (default `false`) for VOD probing (requires `probe_stream` + ffprobe). - `probe_live`: `true`|`false` (default `false`) for background live probing. - `probe_live_interval_hours`: number (default `120`) re-probe interval for live channels. - - `resolve_delay`: seconds (default `2`) shared delay for resolve/probe metadata requests. + - `resolve_delay`: seconds (default `2`) shared delay for resolve metadata requests. + - `probe_delay`: seconds (default `2`) delay between probe tasks. - `aliases` for alias definitions for the same provider with different credentials - `staged` for side loading processed playlists. If you already have a provider configured but want to load the playlist from a different source — for example, @@ -1705,6 +1706,7 @@ Legacy target-level resolve/probe fields are no longer supported. - `resolve_series`: If `true` and you have xtream input with m3u output, series metadata is resolved. - `resolve_vod`: If `true` and you have xtream input, VOD metadata is resolved. - `resolve_delay`: Delay in seconds between metadata requests (default `2`) to reduce provider ban risk. +- `probe_delay`: Delay in seconds between probe requests (default `2`). - `resolve_background`: If `true` (default), metadata jobs are queued in background. If `false`, they run blocking. For `resolve_(vod|series)` data is cached per input; only new/changed entries are updated. diff --git a/backend/src/api/api_utils.rs b/backend/src/api/api_utils.rs index 0d05e03a7..feebde8c9 100644 --- a/backend/src/api/api_utils.rs +++ b/backend/src/api/api_utils.rs @@ -23,7 +23,7 @@ use axum::http::{header, HeaderMap, HeaderValue, Response, StatusCode}; use axum::response::IntoResponse; use bytes::Bytes; use chrono::{DateTime, Utc}; -use futures::{stream, StreamExt, TryStreamExt}; +use futures::{stream, Stream, StreamExt, TryStreamExt}; use jsonwebtoken::{decode, Algorithm, DecodingKey, Validation}; use log::{debug, error, info, log_enabled, trace, warn}; use serde::Serialize; @@ -32,7 +32,7 @@ use shared::model::{ Claims, InputFetchMethod, PlaylistEntry, PlaylistItemType, ProxyType, StreamChannel, TargetType, UserConnectionPermission, VirtualId, XtreamCluster, }; -use shared::utils::{bin_serialize, human_readable_kbps, trim_slash, Internable, CONTENT_TYPE_CBOR}; +use shared::utils::{bin_serialize, human_readable_kbps, trim_slash, Internable, CONTENT_TYPE_CBOR, CONTENT_TYPE_JSON}; use shared::utils::{extract_extension_from_url, replace_url_extension, sanitize_sensitive_info, DASH_EXT, HLS_EXT}; use std::borrow::Cow; use std::collections::HashMap; @@ -1489,6 +1489,17 @@ where stream_json_array(data) } +pub fn stream_json_or_bin_response_stream(accept: Option<&str>, data: S) -> axum::response::Response +where + P: serde::Serialize + Send + 'static, + S: Stream + Send + Unpin + 'static, +{ + if accept.is_some_and(|a| a.contains(CONTENT_TYPE_CBOR)) { + return stream_bin_array_stream(data); + } + stream_json_array_stream(data) +} + pub fn create_session_fingerprint(fingerprint: &Fingerprint, username: &str, virtual_id: u32) -> String { concat_string!(&fingerprint.key, "|", username, "|", &virtual_id.to_string()) } @@ -1518,7 +1529,7 @@ where .chain(stream::once(async { Ok::<_, Infallible>(Bytes::from_static(b"]")) })), ); - try_unwrap_body!(Response::builder().header(header::CONTENT_TYPE, "application/json").body(body)) + try_unwrap_body!(Response::builder().header(header::CONTENT_TYPE, CONTENT_TYPE_JSON).body(body)) } pub fn stream_bin_array

(iter: Box + Send>) -> axum::response::Response @@ -1555,6 +1566,62 @@ where try_unwrap_body!(Response::builder().header(header::CONTENT_TYPE, CONTENT_TYPE_CBOR).body(body)) } +pub fn stream_json_array_stream(stream: S) -> axum::response::Response +where + P: serde::Serialize + Send + 'static, + S: Stream + Send + Unpin + 'static, +{ + let stream = stream::unfold((stream, true), |(mut stream, first)| async move { + match stream.next().await { + Some(item) => { + let mut json = String::new(); + if !first { + json.push(','); + } + let element = serde_json::to_string(&item).ok()?; + json.push_str(&element); + Some((Ok::(Bytes::from(json)), (stream, false))) + } + None => None, + } + }); + + let body = Body::from_stream( + stream::once(async { Ok::<_, Infallible>(Bytes::from_static(b"[")) }) + .chain(stream) + .chain(stream::once(async { Ok::<_, Infallible>(Bytes::from_static(b"]")) })), + ); + + try_unwrap_body!(Response::builder().header(header::CONTENT_TYPE, CONTENT_TYPE_JSON).body(body)) +} + +pub fn stream_bin_array_stream(stream: S) -> axum::response::Response +where + P: serde::Serialize + Send + 'static, + S: Stream + Send + Unpin + 'static, +{ + let stream = stream::unfold(stream, |mut stream| async move { + match stream.next().await { + Some(item) => match bin_serialize(&item) { + Ok(buf) => Some((Ok::(Bytes::from(buf)), stream)), + Err(err) => { + warn!("CBOR serialization error in stream: {err}"); + Some((Ok::(Bytes::new()), stream)) + } + }, + None => None, + } + }); + + let body = Body::from_stream( + stream::once(async { Ok::<_, Infallible>(Bytes::from_static(&[0x9f])) }) + .chain(stream) + .chain(stream::once(async { Ok::<_, Infallible>(Bytes::from_static(&[0xff])) })), + ); + + try_unwrap_body!(Response::builder().header(header::CONTENT_TYPE, CONTENT_TYPE_CBOR).body(body)) +} + pub fn create_api_proxy_user(app_state: &Arc) -> ProxyUserCredentials { let config = app_state.app_config.config.load(); diff --git a/backend/src/api/endpoints/api_playlist_utils.rs b/backend/src/api/endpoints/api_playlist_utils.rs index 77937eba1..392af30f3 100644 --- a/backend/src/api/endpoints/api_playlist_utils.rs +++ b/backend/src/api/endpoints/api_playlist_utils.rs @@ -4,14 +4,15 @@ use axum::response::IntoResponse; use serde_json::{json}; use shared::model::{InputType, M3uPlaylistItem, PlaylistItemType, TargetType, UiPlaylistItem, XtreamCluster, XtreamPlaylistItem}; use std::sync::Arc; -use crate::api::api_utils::{empty_json_list_response, json_or_bin_response, stream_json_or_bin_response}; +use crate::api::api_utils::{empty_json_list_response, json_or_bin_response, stream_json_or_bin_response_stream}; use shared::utils::interner_gc; use crate::repository::{iter_raw_m3u_input_playlist, iter_raw_m3u_target_playlist, iter_raw_xtream_input_playlist, iter_raw_xtream_target_playlist}; +use tokio_stream::StreamExt; pub(in crate::api::endpoints) async fn get_playlist_for_target(cfg_target: Option<&ConfigTarget>, cfg: &AppConfig, cluster: XtreamCluster, accept: Option<&str>) -> impl IntoResponse + Send { if let Some(target) = cfg_target { if target.has_output(TargetType::Xtream) { - let Some((_guard, channel_iterator)) = iter_raw_xtream_target_playlist(cfg, target, cluster).await else { + let Some(channel_iterator) = iter_raw_xtream_target_playlist(cfg, target, cluster).await else { return empty_json_list_response(); }; let item_filter = if cluster == XtreamCluster::Series { @@ -19,10 +20,10 @@ pub(in crate::api::endpoints) async fn get_playlist_for_target(cfg_target: Optio } else { |_pli: &XtreamPlaylistItem| true }; - let converted_iterator: Box + Send> = Box::new(channel_iterator.filter(item_filter).map(UiPlaylistItem::from)); - return stream_json_or_bin_response(accept, converted_iterator).into_response(); + let converted_stream = channel_iterator.filter(item_filter).map(UiPlaylistItem::from); + return stream_json_or_bin_response_stream(accept, converted_stream).into_response(); } else if target.has_output(TargetType::M3u) { - let Some((_guard, channel_iterator)) = iter_raw_m3u_target_playlist(cfg, target, Some(cluster)).await else { + let Some(channel_iterator) = iter_raw_m3u_target_playlist(cfg, target, Some(cluster)).await else { return empty_json_list_response(); }; let item_filter = if cluster == XtreamCluster::Series { @@ -31,8 +32,19 @@ pub(in crate::api::endpoints) async fn get_playlist_for_target(cfg_target: Optio |_pli: &M3uPlaylistItem| true }; - let converted_iterator: Box + Send> = Box::new(channel_iterator.filter(item_filter).map(UiPlaylistItem::from)); - return stream_json_or_bin_response(accept, converted_iterator).into_response(); + let converted_stream = channel_iterator.filter_map(move |res| { + match res { + Ok(pli) => { + if item_filter(&pli) { + Some(Ok(UiPlaylistItem::from(pli))) + } else { + None + } + } + Err(err) => Some(Err(err.to_string())), + } + }); + return stream_json_or_bin_response_stream(accept, converted_stream).into_response(); } } (axum::http::StatusCode::BAD_REQUEST, axum::Json(json!({"error": "Invalid Arguments"}))).into_response() @@ -42,17 +54,17 @@ pub(in crate::api::endpoints) async fn get_playlist_for_target(cfg_target: Optio pub(in crate::api::endpoints) async fn get_playlist_for_input(cfg_input: Option<&Arc>, cfg: &AppConfig, cluster: XtreamCluster, accept: Option<&str>) -> impl IntoResponse + Send { if let Some(input) = cfg_input { if matches!(input.input_type, InputType::Xtream | InputType::XtreamBatch) { - let Some((_guard, channel_iterator)) = iter_raw_xtream_input_playlist(cfg, input, cluster).await else { - return empty_json_list_response(); - }; - let converted_iterator: Box + Send> = Box::new(channel_iterator.map(UiPlaylistItem::from)); - return stream_json_or_bin_response(accept, converted_iterator).into_response(); - } else if matches!(input.input_type, InputType::M3u | InputType::M3uBatch) { - let Some((_guard, channels)) = iter_raw_m3u_input_playlist(cfg, input, Some(cluster)).await else { + let Some(channel_iterator) = iter_raw_xtream_input_playlist(cfg, input, cluster).await else { return empty_json_list_response(); }; - let converted_iterator: Box + Send> = Box::new(channels.map(UiPlaylistItem::from)); - return stream_json_or_bin_response(accept, converted_iterator).into_response(); + let converted_stream = channel_iterator.map(UiPlaylistItem::from); + return stream_json_or_bin_response_stream(accept, converted_stream).into_response(); + } else if matches!(input.input_type, InputType::M3u | InputType::M3uBatch) { + let Some(channels) = iter_raw_m3u_input_playlist(cfg, input, Some(cluster)).await else { + return empty_json_list_response(); + }; + let converted_stream = channels.map(|res| res.map(UiPlaylistItem::from).map_err(|err| err.to_string())); + return stream_json_or_bin_response_stream(accept, converted_stream).into_response(); } } (axum::http::StatusCode::BAD_REQUEST, axum::Json(json!({"error": "Invalid Arguments"}))).into_response() diff --git a/backend/src/api/endpoints/hdhomerun_api.rs b/backend/src/api/endpoints/hdhomerun_api.rs index 6768c21db..6561b56aa 100644 --- a/backend/src/api/endpoints/hdhomerun_api.rs +++ b/backend/src/api/endpoints/hdhomerun_api.rs @@ -98,7 +98,7 @@ fn xtream_item_to_lineup_stream( channels: Option, ) -> impl Stream> where - I: Iterator + 'static, + I: Stream + Send + Unpin + 'static, { match channels { Some(chans) => { @@ -142,7 +142,7 @@ where Err(_) => Ok(Bytes::from("")), } }); - stream::iter(mapped).left_stream() + mapped.left_stream() } None => stream::once(async { Ok(Bytes::from("")) }).right_stream(), } @@ -150,7 +150,7 @@ where fn m3u_item_to_lineup_stream(channels: Option) -> impl Stream> where - I: Iterator + 'static, + I: Stream + Send + Unpin + 'static, { match channels { Some(chans) => { @@ -174,7 +174,7 @@ where Err(_) => Ok(Bytes::from("")), } }); - stream::iter(mapped).left_stream() + mapped.left_stream() } None => stream::once(async { Ok(Bytes::from("")) }).right_stream(), } @@ -276,16 +276,22 @@ async fn lineup_status( cfg.get_target_for_username(&app_state.device.t_username) { if target.has_output(TargetType::M3u) { - if let Some((_guard, iter)) = iter_raw_m3u_target_playlist(&cfg, &target, None).await + if let Some(iter) = iter_raw_m3u_target_playlist(&cfg, &target, None).await { - iter.count() + iter.filter_map(|res| async move { res.ok() }).count().await } else { 0 } } else if target.has_output(TargetType::Xtream) { let credentials = Arc::new(user); - let live = XtreamPlaylistIterator::new(XtreamCluster::Live, &cfg, &target, None, &credentials).await.map_or(0, std::iter::Iterator::count); - let vod = XtreamPlaylistIterator::new(XtreamCluster::Video, &cfg, &target, None, &credentials).await.map_or(0, std::iter::Iterator::count); + let live = match XtreamPlaylistIterator::new(XtreamCluster::Live, &cfg, &target, None, &credentials).await { + Ok(stream) => stream.count().await, + Err(_) => 0, + }; + let vod = match XtreamPlaylistIterator::new(XtreamCluster::Video, &cfg, &target, None, &credentials).await { + Ok(stream) => stream.count().await, + Err(_) => 0, + }; live + vod } else { 0 diff --git a/backend/src/api/endpoints/m3u_api.rs b/backend/src/api/endpoints/m3u_api.rs index 1c577f4e0..139dbf788 100644 --- a/backend/src/api/endpoints/m3u_api.rs +++ b/backend/src/api/endpoints/m3u_api.rs @@ -15,7 +15,7 @@ use crate::repository::storage_const; use crate::utils::debug_if_enabled; use axum::response::IntoResponse; use bytes::Bytes; -use futures::stream; +use futures::StreamExt; use log::{debug, error}; use shared::model::{FieldGetAccessor, PlaylistEntry, PlaylistItemType, TargetType, UserConnectionPermission, XtreamCluster}; use shared::utils::{concat_path, extract_extension_from_url, sanitize_sensitive_info, HLS_EXT}; @@ -26,12 +26,11 @@ async fn m3u_api(api_req: &UserApiRequest, app_state: &AppState) -> impl IntoRes Some((user, target)) => { match m3u_load_rewrite_playlist(&app_state.app_config, &target, &user).await { Ok(m3u_iter) => { - // Convert the iterator into a stream of `Bytes` - let content_stream = stream::iter(m3u_iter.map(|line| { - Ok::(Bytes::from( - [line.clone().as_bytes(), b"\n"].concat(), - )) - })); + // Convert the stream into a stream of `Bytes` + let content_stream = m3u_iter.map(|mut line| { + line.push('\n'); + Ok::(Bytes::from(line)) + }); let mut builder = axum::response::Response::builder() .status(axum::http::StatusCode::OK) diff --git a/backend/src/api/endpoints/user_api.rs b/backend/src/api/endpoints/user_api.rs index d7b7678a9..2699c5e62 100644 --- a/backend/src/api/endpoints/user_api.rs +++ b/backend/src/api/endpoints/user_api.rs @@ -29,12 +29,19 @@ fn get_categories_from_xtream(categories: Option>) - async fn get_categories_from_m3u_playlist(target: &ConfigTarget, config: &AppConfig) -> Vec> { let mut groups = Vec::new(); - if let Some((_guard, iter)) = iter_raw_m3u_target_playlist(config, target, None).await { + if let Some(mut iter) = iter_raw_m3u_target_playlist(config, target, None).await { let mut unique_groups = HashSet::new(); - for item in iter { - if !unique_groups.contains(&item.group) { - unique_groups.insert(item.group.clone()); - groups.push(item.group.clone()); + while let Some(item) = iter.next().await { + match item { + Ok(item) => { + if unique_groups.insert(item.group.clone()) { + groups.push(item.group.clone()); + } + } + Err(err) => { + error!("Failed to read M3U playlist for categories: {err}"); + break; + } } } } diff --git a/backend/src/api/endpoints/v1_api_playlist.rs b/backend/src/api/endpoints/v1_api_playlist.rs index f664de91d..7fc7faeee 100644 --- a/backend/src/api/endpoints/v1_api_playlist.rs +++ b/backend/src/api/endpoints/v1_api_playlist.rs @@ -29,7 +29,8 @@ fn create_config_input_for_m3u(url: &str) -> ConfigInput { ConfigInputFlags::XtreamLiveStreamUsePrefix, ConfigInputFlags::ResolveBackground, ]), - resolve_delay: 2, + resolve_delay: shared::utils::default_resolve_delay_secs(), + probe_delay: shared::utils::default_probe_delay_secs(), probe_live_interval_hours: 120, }), ..Default::default() @@ -50,7 +51,8 @@ fn create_config_input_for_xtream(username: &str, password: &str, host: &str) -> ConfigInputFlags::XtreamLiveStreamUsePrefix, ConfigInputFlags::ResolveBackground, ]), - resolve_delay: 2, + resolve_delay: shared::utils::default_resolve_delay_secs(), + probe_delay: shared::utils::default_probe_delay_secs(), probe_live_interval_hours: 120, }), ..Default::default() diff --git a/backend/src/api/endpoints/xmltv_api.rs b/backend/src/api/endpoints/xmltv_api.rs index 4bb817aaf..6e132d27e 100644 --- a/backend/src/api/endpoints/xmltv_api.rs +++ b/backend/src/api/endpoints/xmltv_api.rs @@ -1,4 +1,4 @@ -use crate::api::api_utils::{empty_json_response_as_array, get_user_target, get_user_target_by_credentials, internal_server_error, resource_response, stream_json_or_bin_response, try_unwrap_body}; +use crate::api::api_utils::{empty_json_response_as_array, get_user_target, get_user_target_by_credentials, internal_server_error, resource_response, stream_json_or_bin_response_stream, try_unwrap_body}; use crate::api::model::AppState; use crate::api::model::UserApiRequest; use crate::model::{Config, EPG_ATTRIB_ID, EPG_TAG_CHANNEL}; @@ -6,7 +6,7 @@ use crate::model::{ConfigTarget, ProxyUserCredentials, TargetOutput}; use crate::repository::m3u_get_epg_file_path_for_target; use crate::repository::storage_const; use crate::repository::XML_PREAMBLE; -use crate::repository::{get_target_storage_path, BPlusTreeQuery}; +use crate::repository::{get_target_storage_path, BPlusTreeQuery, LockedReceiverStream}; use crate::repository::{xtream_get_epg_file_path_for_target, xtream_get_storage_path}; use crate::utils; use crate::utils::{deobscure_text, file_exists_async, format_xmltv_time_utc, get_epg_processing_options, obscure_text, EpgProcessingOptions, EpgTimeShift}; @@ -19,6 +19,8 @@ use shared::model::{EpgChannel, EpgProgramme, ShortEpgDto, ShortEpgResultDto}; use std::path::{Path, PathBuf}; use std::sync::Arc; use tokio::io::AsyncWriteExt; +use tokio::sync::mpsc; +use tokio::task; use tokio_util::io::ReaderStream; pub fn get_empty_epg_response() -> axum::response::Response { @@ -90,16 +92,27 @@ pub async fn serve_epg_web_ui( target: &Arc, ) -> axum::response::Response { if file_exists_async(epg_path).await { - let _file_lock = app_state.app_config.file_locks.read_lock(epg_path).await; - match BPlusTreeQuery::, EpgChannel>::try_new(epg_path) { - Ok(query) => { - let iterator: Box + Send> = Box::new(query.disk_iter().map(|(_, v)| v)); - return stream_json_or_bin_response(accept, iterator); + let iter_lock = app_state.app_config.file_locks.read_lock(epg_path).await; + let bg_lock = app_state.app_config.file_locks.read_lock(epg_path).await; + let epg_path = epg_path.to_path_buf(); + let target_name = target.name.clone(); + let (tx, rx) = mpsc::channel::(64); + + task::spawn_blocking(move || { + let _guard = bg_lock; + let Ok(query) = BPlusTreeQuery::, EpgChannel>::try_new(&epg_path) else { + error!("Failed to open epg db for target {} {}", target_name, epg_path.display()); + return; + }; + for (_, channel) in query.disk_iter() { + if tx.blocking_send(channel).is_err() { + break; + } } - Err(err) => { - error!("Failed to open epg db for target {} {} - {err}", target.name, epg_path.display()); - } - } + }); + + let stream = LockedReceiverStream::new(rx, iter_lock); + return stream_json_or_bin_response_stream(accept, stream); } try_unwrap_body!(empty_json_response_as_array()) } @@ -124,15 +137,6 @@ async fn serve_epg_with_rewrites( return get_empty_epg_response(); } - let _file_lock = app_state.app_config.file_locks.read_lock(epg_path).await; - let mut query = match BPlusTreeQuery::, EpgChannel>::try_new(epg_path) { - Ok(query) => query, - Err(err) => { - error!("Failed to open BPlusTreeQuery {} - {err}", epg_path.display()); - return get_empty_epg_response(); - } - }; - let epg_processing_options = get_epg_processing_options(app_state, user, target); let base_url = if !matches!(epg_processing_options.time_shift, EpgTimeShift::None) || epg_processing_options.rewrite_urls { @@ -144,18 +148,38 @@ async fn serve_epg_with_rewrites( let limit = limit.unwrap_or_default(); + let bg_lock = app_state.app_config.file_locks.read_lock(epg_path).await; + let epg_path = epg_path.to_path_buf(); + let (channel_tx, mut channel_rx) = mpsc::channel::(256); + + task::spawn_blocking(move || { + let _guard = bg_lock; + let Ok(mut query) = BPlusTreeQuery::, EpgChannel>::try_new(&epg_path) else { + error!("Failed to open BPlusTreeQuery {}", epg_path.display()); + return; + }; + + for (_, channel) in query.iter() { + if channel_tx.blocking_send(channel).is_err() { + break; + } + } + }); + let (mut tx, rx) = tokio::io::duplex(8192); tokio::spawn(async move { // Work-Around BytesText DocType escape, see below if let Err(err) = tx.write_all(XML_PREAMBLE.as_ref()).await { error!("EPG: Failed to write xml header {err}"); + return; } if let Err(err) = tx.write_all(r#""#.as_bytes()).await { error!("EPG: Failed to write xml tv header {err}"); + return; } let mut writer = quick_xml::writer::Writer::new(tx); - for (_, channel) in query.iter() { + while let Some(channel) = channel_rx.recv().await { let programmes = if limit > 0 { channel.get_programme_with_limit(limit) } else { @@ -237,20 +261,33 @@ async fn serve_epg_with_rewrites( } async fn get_epg_channel(app_state: &Arc, channel_id: &Arc, epg_path: &Path) -> Option { - let _file_lock = app_state.app_config.file_locks.read_lock(epg_path).await; - match BPlusTreeQuery::, EpgChannel>::try_new(epg_path) { - Ok(mut query) => { - match query.query(channel_id) { - Ok(Some(item)) => return Some(item), - Ok(None) => {} + let file_lock = app_state.app_config.file_locks.read_lock(epg_path).await; + let epg_path = epg_path.to_path_buf(); + let channel_id = Arc::clone(channel_id); + + match task::spawn_blocking(move || -> Option { + let _guard = file_lock; + match BPlusTreeQuery::, EpgChannel>::try_new(&epg_path) { + Ok(mut query) => match query.query(&channel_id) { + Ok(Some(item)) => Some(item), + Ok(None) => None, Err(err) => { error!("Failed to query db file {}: {err}", epg_path.display()); + None } + }, + Err(err) => { + error!("Failed to read db file {}: {err}", epg_path.display()); + None } } - Err(err) => { error!("Failed to read db file {}: {err}", epg_path.display()); } + }).await { + Ok(result) => result, + Err(err) => { + error!("Failed to run epg query task: {err}"); + None + } } - None } fn format_xmltv_time(ts: i64) -> String { @@ -422,4 +459,3 @@ pub fn xmltv_api_register() -> axum::Router> { axum::routing::get(epg_api_resource), ) } - diff --git a/backend/src/api/endpoints/xtream_api.rs b/backend/src/api/endpoints/xtream_api.rs index 174b93377..8e088b3db 100644 --- a/backend/src/api/endpoints/xtream_api.rs +++ b/backend/src/api/endpoints/xtream_api.rs @@ -1339,19 +1339,21 @@ async fn xtream_player_api( } } -fn xtream_create_content_stream( - xtream_iter: impl Iterator, -) -> impl Stream> { - stream::once(async { Ok::(Bytes::from("[")) }).chain( - stream::iter(xtream_iter.map(move |(mut line, has_next)| { - if has_next { - line.push(','); - } - Ok::(Bytes::from(line)) - })).chain(stream::once(async { - Ok::(Bytes::from("]")) - })), - ) +fn xtream_create_content_stream( + xtream_iter: S, +) -> impl Stream> +where + S: Stream + Send + Unpin + 'static, +{ + let mapped = xtream_iter.map(move |(mut line, has_next)| { + if has_next { + line.push(','); + } + Ok::(Bytes::from(line)) + }); + stream::once(async { Ok::(Bytes::from("[")) }) + .chain(mapped) + .chain(stream::once(async { Ok::(Bytes::from("]")) })) } async fn xtream_player_api_get( diff --git a/backend/src/api/model/metadata_update_manager.rs b/backend/src/api/model/metadata_update_manager.rs index 0ed55c2cd..ccb35529f 100644 --- a/backend/src/api/model/metadata_update_manager.rs +++ b/backend/src/api/model/metadata_update_manager.rs @@ -10,6 +10,7 @@ use shared::error::TuliproxError; use shared::model::{InputType, LiveStreamProperties, PlaylistItemType, SeriesStreamProperties, UUIDType, VideoStreamProperties, XtreamCluster, XtreamPlaylistItem}; use shared::utils::generate_playlist_uuid; use std::sync::atomic::{AtomicU64, Ordering}; +use parking_lot::Mutex as ParkingMutex; use std::sync::{Arc, Weak}; use std::time::{Duration, Instant}; use tokio::sync::{mpsc, Mutex}; @@ -321,7 +322,7 @@ impl MetadataUpdateManager { struct DbHandle { _guard: FileReadGuard, - query: BPlusTreeQuery, + query: Arc>>, } struct InputWorker { @@ -754,15 +755,23 @@ impl InputWorker { let Some(storage_path) = crate::repository::xtream_get_storage_path(config, target_name) else { continue; }; let mapping_file = get_target_id_mapping_file(&target_path); - // Read mapping under lock and release lock immediately after opening. - let mapping = { - let _lock = app_state.app_config.file_locks.read_lock(&mapping_file).await; - match TargetIdMapping::new(&mapping_file, false) { - Ok(m) => m, - Err(e) => { - error!("Failed to open ID mapping for target {target_name}: {e}"); - continue; - } + // Read mapping under lock inside spawn_blocking to avoid blocking async executor. + let file_lock = app_state.app_config.file_locks.read_lock(&mapping_file).await; + let mapping_file_clone = mapping_file.clone(); + let mapping = match tokio::task::spawn_blocking(move || { + let _guard = file_lock; + TargetIdMapping::new(&mapping_file_clone, false) + }) + .await + { + Ok(Ok(mapping)) => mapping, + Ok(Err(e)) => { + error!("Failed to open ID mapping for target {target_name}: {e}"); + continue; + } + Err(err) => { + error!("Failed to open ID mapping for target {target_name}: {err}"); + continue; } }; @@ -961,19 +970,33 @@ impl InputWorker { let target_name = target.name.as_str(); let xtream_path = xtream_get_file_path(storage_path, XtreamCluster::Video); - let updates = { - let _lock_read = app_state.app_config.file_locks.read_lock(&xtream_path).await; - let mut updates = Vec::with_capacity(virtual_updates.len()); - if let Ok(mut query) = BPlusTreeQuery::::try_new(&xtream_path) { - for (virtual_id, props) in &virtual_updates { - if let Ok(Some(mut item)) = query.query_zero_copy(virtual_id) { - item.additional_properties = - Some(shared::model::StreamProperties::Video(Box::new((*props).clone()))); - updates.push(item); - } + let updates_input: Vec<(u32, VideoStreamProperties)> = virtual_updates + .into_iter() + .map(|(vid, props)| (vid, props.clone())) + .collect(); + + let file_lock = app_state.app_config.file_locks.read_lock(&xtream_path).await; + let xtream_path = xtream_path.clone(); + let updates = match tokio::task::spawn_blocking(move || -> Vec { + let _guard = file_lock; + let mut updates = Vec::with_capacity(updates_input.len()); + let Ok(mut query) = BPlusTreeQuery::::try_new(&xtream_path) else { + return updates; + }; + for (virtual_id, props) in updates_input { + if let Ok(Some(mut item)) = query.query_zero_copy(&virtual_id) { + item.additional_properties = + Some(shared::model::StreamProperties::Video(Box::new(props))); + updates.push(item); } } updates + }).await { + Ok(updates) => updates, + Err(err) => { + error!("Failed to read VOD updates from disk for {target_name}: {err}"); + Vec::new() + } }; if updates.is_empty() { @@ -1007,19 +1030,33 @@ impl InputWorker { let target_name = target.name.as_str(); let xtream_path = xtream_get_file_path(storage_path, XtreamCluster::Series); - let updates = { - let _lock_read = app_state.app_config.file_locks.read_lock(&xtream_path).await; - let mut updates = Vec::with_capacity(virtual_updates.len()); - if let Ok(mut query) = BPlusTreeQuery::::try_new(&xtream_path) { - for (virtual_id, props) in &virtual_updates { - if let Ok(Some(mut item)) = query.query_zero_copy(virtual_id) { - item.additional_properties = - Some(shared::model::StreamProperties::Series(Box::new((*props).clone()))); - updates.push(item); - } + let updates_input: Vec<(u32, SeriesStreamProperties)> = virtual_updates + .into_iter() + .map(|(vid, props)| (vid, props.clone())) + .collect(); + + let file_lock = app_state.app_config.file_locks.read_lock(&xtream_path).await; + let xtream_path = xtream_path.clone(); + let updates = match tokio::task::spawn_blocking(move || -> Vec { + let _guard = file_lock; + let mut updates = Vec::with_capacity(updates_input.len()); + let Ok(mut query) = BPlusTreeQuery::::try_new(&xtream_path) else { + return updates; + }; + for (virtual_id, props) in updates_input { + if let Ok(Some(mut item)) = query.query_zero_copy(&virtual_id) { + item.additional_properties = + Some(shared::model::StreamProperties::Series(Box::new(props))); + updates.push(item); } } updates + }).await { + Ok(updates) => updates, + Err(err) => { + error!("Failed to read Series updates from disk for {target_name}: {err}"); + Vec::new() + } }; if updates.is_empty() { @@ -1053,19 +1090,33 @@ impl InputWorker { let target_name = target.name.as_str(); let xtream_path = xtream_get_file_path(storage_path, XtreamCluster::Live); - let updates = { - let _lock_read = app_state.app_config.file_locks.read_lock(&xtream_path).await; - let mut updates = Vec::with_capacity(virtual_updates.len()); - if let Ok(mut query) = BPlusTreeQuery::::try_new(&xtream_path) { - for (virtual_id, props) in &virtual_updates { - if let Ok(Some(mut item)) = query.query_zero_copy(virtual_id) { - item.additional_properties = - Some(shared::model::StreamProperties::Live(Box::new((*props).clone()))); - updates.push(item); - } + let updates_input: Vec<(u32, LiveStreamProperties)> = virtual_updates + .into_iter() + .map(|(vid, props)| (vid, props.clone())) + .collect(); + + let file_lock = app_state.app_config.file_locks.read_lock(&xtream_path).await; + let xtream_path = xtream_path.clone(); + let updates = match tokio::task::spawn_blocking(move || -> Vec { + let _guard = file_lock; + let mut updates = Vec::with_capacity(updates_input.len()); + let Ok(mut query) = BPlusTreeQuery::::try_new(&xtream_path) else { + return updates; + }; + for (virtual_id, props) in updates_input { + if let Ok(Some(mut item)) = query.query_zero_copy(&virtual_id) { + item.additional_properties = + Some(shared::model::StreamProperties::Live(Box::new(props))); + updates.push(item); } } updates + }).await { + Ok(updates) => updates, + Err(err) => { + error!("Failed to read Live updates from disk for {target_name}: {err}"); + Vec::new() + } }; if updates.is_empty() { @@ -1107,38 +1158,53 @@ impl InputWorker { } } } - async fn get_or_open_query<'a>( + async fn get_or_open_query( input_name: &str, app_state: &Arc, cluster: XtreamCluster, - db_handles: &'a mut HashMap, + db_handles: &mut HashMap, failed_clusters: &mut HashSet, - ) -> Option<&'a mut BPlusTreeQuery> { + ) -> Option>>> { if failed_clusters.contains(&cluster) { return None; } - if let std::collections::hash_map::Entry::Vacant(e) = db_handles.entry(cluster) { + if let std::collections::hash_map::Entry::Vacant(entry) = db_handles.entry(cluster) { let working_dir = &app_state.app_config.config.load().working_dir; if let Ok(storage_path) = get_input_storage_path(input_name, working_dir).await { let file_path = xtream_get_file_path(&storage_path, cluster); if file_path.exists() { let lock = app_state.app_config.file_locks.read_lock(&file_path).await; - if let Ok(query) = BPlusTreeQuery::try_new(&file_path) { - e.insert(DbHandle { _guard: lock, query }); + let file_path = file_path.clone(); + let query = match tokio::task::spawn_blocking(move || { + BPlusTreeQuery::::try_new(&file_path) + }).await { + Ok(Ok(query)) => Some(query), + Ok(Err(err)) => { + error!("Failed to open BPlusTreeQuery for {cluster}: {err}"); + None + } + Err(err) => { + error!("Failed to open BPlusTreeQuery for {cluster}: {err}"); + None + } + }; + + if let Some(query) = query { + entry.insert(DbHandle { + _guard: lock, + query: Arc::new(ParkingMutex::new(query)), + }); } else { failed_clusters.insert(cluster); } } else { - // File doesn't exist, technically a failure to open but acceptable. - // We don't mark as failure to allow creation if it appears, but for read logic - // we could cache the non-existence if we wanted. - // For now, let's strictly follow the instruction about "failure cache" for *errors* or corruption. + // File doesn't exist; do not mark as failure to allow future creation. } } } - db_handles.get_mut(&cluster).map(|h| &mut h.query) + db_handles.get(&cluster).map(|h| Arc::clone(&h.query)) } // Helper for get_item_name with caching @@ -1157,8 +1223,21 @@ impl InputWorker { }; if let ProviderIdType::Id(vid) = id { + let stream_id = *vid; if let Some(query) = Self::get_or_open_query(input_name, app_state, cluster, db_handles, failed_clusters).await { - if let Ok(Some(item)) = query.query_zero_copy(vid) { + let query = Arc::clone(&query); + let item = match tokio::task::spawn_blocking(move || { + let mut guard = query.lock(); + guard.query_zero_copy(&stream_id).ok().flatten() + }).await { + Ok(item) => item, + Err(err) => { + error!("Failed to query item name for {stream_id}: {err}"); + None + } + }; + + if let Some(item) = item { return Some(if item.title.is_empty() { item.name.to_string() } else { item.title.to_string() }); } } diff --git a/backend/src/model/config/input.rs b/backend/src/model/config/input.rs index e383d0f3d..339d3f48d 100644 --- a/backend/src/model/config/input.rs +++ b/backend/src/model/config/input.rs @@ -67,6 +67,7 @@ impl std::ops::BitOr for ConfigInputFlagsSet { pub struct ConfigInputOptions { pub flags: ConfigInputFlagsSet, pub resolve_delay: u16, + pub probe_delay: u16, pub probe_live_interval_hours: u32, } @@ -109,6 +110,7 @@ impl From<&ConfigInputOptionsDto> for ConfigInputOptions { Self { flags, resolve_delay: dto.resolve_delay, + probe_delay: dto.probe_delay, probe_live_interval_hours: dto.probe_live_interval_hours, } } diff --git a/backend/src/processing/processor/playlist.rs b/backend/src/processing/processor/playlist.rs index 24806874c..aad225b36 100644 --- a/backend/src/processing/processor/playlist.rs +++ b/backend/src/processing/processor/playlist.rs @@ -48,13 +48,11 @@ use shared::model::{ }; use shared::model::{InputStats, PlaylistStats, SourceStats, TargetStats, UUIDType}; use shared::utils::{ - create_alias_uuid, default_as_default, default_probe_live_interval, default_resolve_delay_secs, interner_gc, + create_alias_uuid, default_as_default, default_probe_delay_secs, default_probe_live_interval, interner_gc, Internable, }; use std::time::Instant; -const DEFAULT_PROBE_STREAM_DELAY: u16 = 50; - fn is_valid(pli: &PlaylistItem, filter: &Filter, match_as_ascii: bool) -> bool { let provider = ValueProvider { pli, match_as_ascii }; filter.filter(&provider) @@ -955,11 +953,11 @@ fn get_live_probe_interval_settings( return None; } target.get_xtream_output().map(|_| { - let (resolve_delay, input_probe_live_interval_hours) = input_options.map_or( - (default_resolve_delay_secs(), default_probe_live_interval()), - |options| (options.resolve_delay, options.probe_live_interval_hours), + let (probe_delay, input_probe_live_interval_hours) = input_options.map_or( + (default_probe_delay_secs(), default_probe_live_interval()), + |options| (options.probe_delay, options.probe_live_interval_hours), ); - (resolve_delay, u64::from(input_probe_live_interval_hours) * 3600) + (probe_delay, u64::from(input_probe_live_interval_hours) * 3600) }) } @@ -992,6 +990,7 @@ fn provider_id_from_item(item: &PlaylistItem) -> Option { } } +#[allow(clippy::too_many_lines)] async fn playlist_probe(ctx: &PlaylistProcessingContext, target: &ConfigTarget, fpl: &mut FetchedPlaylist<'_>) { let Some(mgr) = ctx.metadata_manager.as_ref() else { return; @@ -1012,6 +1011,7 @@ async fn playlist_probe(ctx: &PlaylistProcessingContext, target: &ConfigTarget, let input_name = fpl.input.name.clone(); let input_type = fpl.input.input_type; + let xtream_probe_handled = input_type.is_xtream() && target.get_xtream_output().is_some(); let live_probe_settings = if probe_live_enabled { get_live_probe_interval_settings(target, input_type, Some(opts)).map(|(delay, interval_secs)| { let interval_signed = i64::try_from(interval_secs).unwrap_or(i64::MAX); @@ -1038,14 +1038,14 @@ async fn playlist_probe(ctx: &PlaylistProcessingContext, target: &ConfigTarget, continue; } - if let Some((resolve_delay, interval_secs, cutoff_ts)) = live_probe_settings { + if let Some((probe_delay, interval_secs, cutoff_ts)) = live_probe_settings { if needs_live_probe(&item, cutoff_ts) { if let Some(provider_id) = provider_id_from_item(&item) { if queued_live_keys.insert(provider_id.clone()) { let task = UpdateTask::ProbeLive { id: provider_id, reason: ResolveReasonSet::from_variants(&[ResolveReason::Probe]), - delay: resolve_delay, + delay: probe_delay, interval: interval_secs, }; mgr.queue_task_background(input_name.clone(), task); @@ -1062,11 +1062,19 @@ async fn playlist_probe(ctx: &PlaylistProcessingContext, target: &ConfigTarget, if !probe_vod_enabled { continue; } + // Xtream outputs handle VOD probe as part of the resolve pipeline (after resolve). + if xtream_probe_handled { + continue; + } } PlaylistItemType::Series | PlaylistItemType::LocalSeries => { if !probe_series_enabled { continue; } + // Xtream outputs handle Series probe as part of the resolve pipeline (after resolve). + if xtream_probe_handled { + continue; + } } _ => continue, } @@ -1094,7 +1102,7 @@ async fn playlist_probe(ctx: &PlaylistProcessingContext, target: &ConfigTarget, url: item.header.url.to_string(), item_type: item.header.item_type, reason: ResolveReasonSet::from_variants(&[ResolveReason::MissingDetails]), - delay: DEFAULT_PROBE_STREAM_DELAY, + delay: opts.probe_delay, }; mgr.queue_task_background(input_name.clone(), task); queued_stream_count += 1; diff --git a/backend/src/processing/processor/xtream.rs b/backend/src/processing/processor/xtream.rs index 704d52d34..19b433b3b 100644 --- a/backend/src/processing/processor/xtream.rs +++ b/backend/src/processing/processor/xtream.rs @@ -1,3 +1,4 @@ +use parking_lot::Mutex; use std::sync::Arc; use shared::error::{TuliproxError}; use crate::model::{AppConfig, ConfigInput, ConfigInputFlags}; @@ -9,12 +10,13 @@ use crate::processing::parser::xtream::create_xtream_url; use crate::api::model::ProviderIdType; /// Updates metadata for a single Live stream (primarily probing) +#[allow(clippy::too_many_lines)] pub async fn update_live_stream_metadata( app_config: &Arc, input: &ConfigInput, id: ProviderIdType, save: bool, - db_query: Option<&mut BPlusTreeQuery>, + db_query: Option>>>, ) -> Result, TuliproxError> { let working_dir = &app_config.config.load().working_dir; let storage_path = get_input_storage_path(&input.name, working_dir).await @@ -27,24 +29,54 @@ pub async fn update_live_stream_metadata( let stream_id_opt = if let ProviderIdType::Id(vid) = id { Some(vid) } else { None }; if let Some(stream_id) = stream_id_opt { - // Use provided query or open new one if let Some(query) = db_query { - if let Ok(Some(item)) = query.query_zero_copy(&stream_id) { + let query = Arc::clone(&query); + let item = match tokio::task::spawn_blocking(move || { + let mut guard = query.lock(); + guard.query_zero_copy(&stream_id).ok().flatten() + }) + .await + { + Ok(item) => item, + Err(err) => { + warn!("Failed to query Live metadata from disk for {stream_id}: {err}"); + None + } + }; + + if let Some(item) = item { existing_item = Some(item.clone()); if let Some(StreamProperties::Live(p)) = item.additional_properties.as_ref() { - props = Some(*p.clone()); + props = Some(*p.clone()); } } } else { let xtream_path = xtream_get_file_path(&storage_path, XtreamCluster::Live); if xtream_path.exists() { - let _file_lock = app_config.file_locks.read_lock(&xtream_path).await; - if let Ok(mut query) = BPlusTreeQuery::::try_new(&xtream_path) { - if let Ok(Some(item)) = query.query_zero_copy(&stream_id) { - existing_item = Some(item.clone()); - if let Some(StreamProperties::Live(p)) = item.additional_properties.as_ref() { - props = Some(*p.clone()); - } + let file_lock = app_config.file_locks.read_lock(&xtream_path).await; + let xtream_path = xtream_path.clone(); + let item = match tokio::task::spawn_blocking(move || { + let _guard = file_lock; + let mut query = BPlusTreeQuery::::try_new(&xtream_path)?; + query.query_zero_copy(&stream_id) + }) + .await + { + Ok(Ok(item)) => item, + Ok(Err(err)) => { + warn!("Failed to query Live metadata from disk for {stream_id}: {err}"); + None + } + Err(err) => { + warn!("Failed to query Live metadata from disk for {stream_id}: {err}"); + None + } + }; + + if let Some(item) = item { + existing_item = Some(item.clone()); + if let Some(StreamProperties::Live(p)) = item.additional_properties.as_ref() { + props = Some(*p.clone()); } } } diff --git a/backend/src/processing/processor/xtream_series.rs b/backend/src/processing/processor/xtream_series.rs index 9e261e26a..c829900db 100644 --- a/backend/src/processing/processor/xtream_series.rs +++ b/backend/src/processing/processor/xtream_series.rs @@ -21,6 +21,7 @@ use shared::error::TuliproxError; use shared::model::{MediaQuality, PlaylistEntry, PlaylistItem, SeriesStreamProperties, StreamProperties, XtreamPlaylistItem, XtreamSeriesInfo}; use shared::model::{PlaylistGroup, PlaylistItemType, XtreamCluster}; use std::collections::HashMap; +use parking_lot::Mutex; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -247,7 +248,7 @@ async fn process_immediate_series_info( // Keep an optional read query open and reopen lazily only when needed. let xtream_path = xtream_get_file_path(&storage_path, XtreamCluster::Series); - let mut db_query_holder = None; + let mut db_query_holder: Option>>> = None; let mut _db_lock_holder = None; let mut groups_to_add = Vec::new(); @@ -272,14 +273,31 @@ async fn process_immediate_series_info( if !reasons.is_empty() { if let Some(active_provider) = ctx.provider_manager.as_ref() { if db_query_holder.is_none() && xtream_path.exists() { - let lock = ctx.config.file_locks.read_lock(&xtream_path).await; - if let Ok(query) = BPlusTreeQuery::::try_new(&xtream_path) { - db_query_holder = Some(query); - _db_lock_holder = Some(lock); + let file_lock = ctx.config.file_locks.read_lock(&xtream_path).await; + let xtream_path = xtream_path.clone(); + let query = match tokio::task::spawn_blocking(move || { + let guard = file_lock; + let query = BPlusTreeQuery::::try_new(&xtream_path)?; + Ok::<_, std::io::Error>((query, guard)) + }).await { + Ok(Ok((query, guard))) => Some((query, guard)), + Ok(Err(err)) => { + error!("Failed to open BPlusTreeQuery for Series: {err}"); + None + } + Err(err) => { + error!("Failed to open BPlusTreeQuery for Series: {err}"); + None + } + }; + + if let Some((query, guard)) = query { + db_query_holder = Some(Arc::new(Mutex::new(query))); + _db_lock_holder = Some(guard); } } // Pass the optional reference to the query - let db_query_ref = db_query_holder.as_mut(); + let db_query_ref = db_query_holder.as_ref().map(Arc::clone); match update_series_info_immediate(ctx, active_provider, input, pli, provider_id.clone(), &reasons, db_query_ref).await { Ok(Some(updated_props)) => { @@ -401,7 +419,7 @@ async fn update_series_info_immediate( pli: &PlaylistItem, id: ProviderIdType, reasons: &ResolveReasonSet, - db_query: Option<&mut BPlusTreeQuery>, + db_query: Option>>>, ) -> Result, TuliproxError> { let fetch_info = reasons.contains(ResolveReason::Info); @@ -431,7 +449,7 @@ fn check_resolve_reasons( check_resolve_tmdb(resolve_options, resolve_tmdb_enabled, pli, needs_info, &mut reasons); if resolve_options.has_flag(ResolveOptionsFlags::Probe) { - check_needs_probe(pli, needs_info, &mut reasons); + check_needs_probe(pli, &mut reasons); } reasons } @@ -467,23 +485,24 @@ fn check_resolve_tmdb( } } -fn check_needs_probe(pli: &mut PlaylistItem, needs_info: bool, reasons: &mut ResolveReasonSet) { - if !needs_info { - return; - } - if let Some(StreamProperties::Series(series_stream_props)) = pli.header.additional_properties.as_ref() { - if let Some(details) = series_stream_props.details.as_ref() { - if let Some(episodes) = details.episodes.as_ref() { - for ep in episodes { +fn check_needs_probe(pli: &PlaylistItem, reasons: &mut ResolveReasonSet) { + let needs_probe = match pli.header.additional_properties.as_ref() { + Some(StreamProperties::Series(series_stream_props)) => series_stream_props + .details + .as_ref() + .and_then(|d| d.episodes.as_ref()) + .is_some_and(|episodes| { + episodes.iter().any(|ep| { let missing_video = !MediaQuality::is_valid_media_info(ep.video.as_deref()); let missing_audio = !MediaQuality::is_valid_media_info(ep.audio.as_deref()); - if missing_video || missing_audio { - reasons.set(ResolveReason::Probe); - break; - } - } - } - } + missing_video || missing_audio + }) + }), + _ => false, + }; + + if needs_probe { + reasons.set(ResolveReason::Probe); } } @@ -506,7 +525,7 @@ pub async fn update_series_metadata( save: bool, fetch_info: bool, do_probe: bool, - db_query: Option<&mut BPlusTreeQuery>, + db_query: Option>>>, ) -> Result, TuliproxError> { let working_dir = &app_config.config.load().working_dir; let storage_path = get_input_storage_path(&input.name, working_dir).await @@ -523,9 +542,26 @@ pub async fn update_series_metadata( let series_id_opt = if let ProviderIdType::Id(vid) = id { Some(vid) } else { None }; if let Some(series_id) = series_id_opt { - // Use provided query or open new one if let Some(query) = db_query { - if let Ok(Some(item)) = query.query_zero_copy(&series_id) { + let query = Arc::clone(&query); + let item = match tokio::task::spawn_blocking(move || { + let mut guard = query.lock(); + guard.query_zero_copy(&series_id) + }) + .await + { + Ok(Ok(item)) => item, + Ok(Err(err)) => { + error!("Failed to query Series metadata from disk for {series_id}: {err}"); + None + } + Err(err) => { + error!("Failed to query Series metadata from disk for {series_id}: {err}"); + None + } + }; + + if let Some(item) = item { existing_item = Some(item.clone()); if let Some(StreamProperties::Series(p)) = item.additional_properties.as_ref() { props = Some(*p.clone()); @@ -534,13 +570,30 @@ pub async fn update_series_metadata( } else { let xtream_path = xtream_get_file_path(&storage_path, XtreamCluster::Series); if xtream_path.exists() { - let _file_lock = app_config.file_locks.read_lock(&xtream_path).await; - if let Ok(mut query) = BPlusTreeQuery::::try_new(&xtream_path) { - if let Ok(Some(item)) = query.query_zero_copy(&series_id) { - existing_item = Some(item.clone()); - if let Some(StreamProperties::Series(p)) = item.additional_properties.as_ref() { - props = Some(*p.clone()); - } + let file_lock = app_config.file_locks.read_lock(&xtream_path).await; + let xtream_path = xtream_path.clone(); + let item = match tokio::task::spawn_blocking(move || { + let _guard = file_lock; + let mut query = BPlusTreeQuery::::try_new(&xtream_path)?; + query.query_zero_copy(&series_id) + }) + .await + { + Ok(Ok(item)) => item, + Ok(Err(err)) => { + error!("Failed to query Series metadata from disk for {series_id}: {err}"); + None + } + Err(err) => { + error!("Failed to query Series metadata from disk for {series_id}: {err}"); + None + } + }; + + if let Some(item) = item { + existing_item = Some(item.clone()); + if let Some(StreamProperties::Series(p)) = item.additional_properties.as_ref() { + props = Some(*p.clone()); } } } @@ -676,12 +729,14 @@ pub async fn update_series_metadata( let input_password = input.password.as_deref().unwrap_or(""); let mut probed_count = 0; + let mut missing_any = false; for ep in episodes { let missing_video = !MediaQuality::is_valid_media_info(ep.video.as_deref()); let missing_audio = !MediaQuality::is_valid_media_info(ep.audio.as_deref()); if missing_video || missing_audio { + missing_any = true; // Acquire Connection logic let temp_handle = if active_handle.is_some() { @@ -725,8 +780,14 @@ pub async fn update_series_metadata( } if probed_count > 0 { info!("Probed {probed_count} episodes for Series ID {display_id}"); + } else if !missing_any { + debug!("Series probe skipped for ID {display_id} (all episodes already have A/V details)"); } + } else { + debug!("Series probe skipped for ID {display_id} (no episode details available)"); } + } else { + debug!("Series probe skipped for ID {display_id} (no series details available)"); } } diff --git a/backend/src/processing/processor/xtream_vod.rs b/backend/src/processing/processor/xtream_vod.rs index bf14c2545..3d1191a5e 100644 --- a/backend/src/processing/processor/xtream_vod.rs +++ b/backend/src/processing/processor/xtream_vod.rs @@ -18,6 +18,7 @@ use serde_json::Value; use shared::error::TuliproxError; use shared::model::{MediaQuality, PlaylistEntry, PlaylistItem, PlaylistItemType, StreamProperties, VideoStreamDetailProperties, VideoStreamProperties, XtreamCluster, XtreamPlaylistItem, XtreamVideoInfo}; use std::collections::HashMap; +use parking_lot::Mutex; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -180,7 +181,7 @@ async fn process_immediate_vod_info(ctx: &PlaylistProcessingContext, fpl: &mut F // Keep an optional read query open and reopen lazily only when needed. let xtream_path = xtream_get_file_path(&storage_path, XtreamCluster::Video); - let mut db_query_holder = None; + let mut db_query_holder: Option>>> = None; let mut _db_lock_holder = None; let mut batch: Vec<(ProviderIdType, VideoStreamProperties)> = Vec::with_capacity(BATCH_SIZE); @@ -205,13 +206,28 @@ async fn process_immediate_vod_info(ctx: &PlaylistProcessingContext, fpl: &mut F if let Some(prov_mgr) = ctx.provider_manager.as_ref() { if db_query_holder.is_none() && xtream_path.exists() { let lock = ctx.config.file_locks.read_lock(&xtream_path).await; - if let Ok(query) = BPlusTreeQuery::::try_new(&xtream_path) { - db_query_holder = Some(query); + let xtream_path = xtream_path.clone(); + let query = match tokio::task::spawn_blocking(move || { + BPlusTreeQuery::::try_new(&xtream_path) + }).await { + Ok(Ok(query)) => Some(query), + Ok(Err(err)) => { + error!("Failed to open BPlusTreeQuery for VOD: {err}"); + None + } + Err(err) => { + error!("Failed to open BPlusTreeQuery for VOD: {err}"); + None + } + }; + + if let Some(query) = query { + db_query_holder = Some(Arc::new(Mutex::new(query))); _db_lock_holder = Some(lock); } } // Pass the optional reference to the query - let db_query_ref = db_query_holder.as_mut(); + let db_query_ref = db_query_holder.as_ref().map(Arc::clone); match update_vod_info_immediate(ctx, prov_mgr, input, pli, provider_id.clone(), &reasons, db_query_ref).await { Ok(Some(updated_props)) => { @@ -311,17 +327,14 @@ fn check_resolve_reasons( // TMDB check check_resolve_tmdb(resolve_options, resolve_tmdb_enabled, pli, needs_info, &mut reasons); - // Probe check + // Probe check (independent of resolve info fetch; we still probe when A/V details are missing) if do_probe { - check_needs_probe(pli, needs_info, &mut reasons); + check_needs_probe(pli, &mut reasons); } reasons } -fn check_needs_probe(pli: &PlaylistItem, needs_info: bool, reasons: &mut ResolveReasonSet) { - if !needs_info { - return; - } +fn check_needs_probe(pli: &PlaylistItem, reasons: &mut ResolveReasonSet) { let needs_probe = match pli.header.additional_properties.as_ref() { Some(StreamProperties::Video(props)) => { let details = props.details.as_ref(); @@ -410,7 +423,7 @@ async fn update_vod_info_immediate( pli: &PlaylistItem, id: ProviderIdType, reasons: &ResolveReasonSet, - db_query: Option<&mut BPlusTreeQuery>, + db_query: Option>>>, ) -> Result, TuliproxError> { let fetch_info = reasons.contains(ResolveReason::Info); @@ -449,7 +462,7 @@ pub async fn update_vod_metadata( save: bool, fetch_info: bool, do_probe: bool, - db_query: Option<&mut BPlusTreeQuery>, + db_query: Option>>>, ) -> Result, TuliproxError> { let working_dir = &app_config.config.load().working_dir; let storage_path = get_input_storage_path(&input.name, working_dir).await @@ -467,9 +480,22 @@ pub async fn update_vod_metadata( let stream_id_opt = if let ProviderIdType::Id(vid) = id { Some(vid) } else { None }; if let Some(stream_id) = stream_id_opt { - // Use provided query or open new one if let Some(query) = db_query { - if let Ok(Some(item)) = query.query_zero_copy(&stream_id) { + let query = Arc::clone(&query); + let item = match tokio::task::spawn_blocking(move || { + let mut guard = query.lock(); + guard.query_zero_copy(&stream_id).ok().flatten() + }) + .await + { + Ok(item) => item, + Err(err) => { + error!("Failed to query VOD metadata from disk for {stream_id}: {err}"); + None + } + }; + + if let Some(item) = item { existing_item = Some(item.clone()); if let Some(StreamProperties::Video(p)) = item.additional_properties.as_ref() { props = Some(*p.clone()); @@ -478,13 +504,30 @@ pub async fn update_vod_metadata( } else { let xtream_path = xtream_get_file_path(&storage_path, XtreamCluster::Video); if xtream_path.exists() { - let _file_lock = app_config.file_locks.read_lock(&xtream_path).await; - if let Ok(mut query) = BPlusTreeQuery::::try_new(&xtream_path) { - if let Ok(Some(item)) = query.query_zero_copy(&stream_id) { - existing_item = Some(item.clone()); - if let Some(StreamProperties::Video(p)) = item.additional_properties.as_ref() { - props = Some(*p.clone()); - } + let file_lock = app_config.file_locks.read_lock(&xtream_path).await; + let xtream_path = xtream_path.clone(); + let item = match tokio::task::spawn_blocking(move || { + let _guard = file_lock; + let mut query = BPlusTreeQuery::::try_new(&xtream_path)?; + query.query_zero_copy(&stream_id) + }) + .await + { + Ok(Ok(item)) => item, + Ok(Err(err)) => { + error!("Failed to query VOD metadata from disk for {stream_id}: {err}"); + None + } + Err(err) => { + error!("Failed to query VOD metadata from disk for {stream_id}: {err}"); + None + } + }; + + if let Some(item) = item { + existing_item = Some(item.clone()); + if let Some(StreamProperties::Video(p)) = item.additional_properties.as_ref() { + props = Some(*p.clone()); } } } diff --git a/backend/src/repository/bplustree.rs b/backend/src/repository/bplustree.rs index c32e8d653..9d96d32f4 100644 --- a/backend/src/repository/bplustree.rs +++ b/backend/src/repository/bplustree.rs @@ -2827,6 +2827,44 @@ where Ok(query) } + /// Clone an existing query without re-reading the file header. + /// This avoids synchronous disk initialization while still providing + /// an independent reader. + pub fn try_clone(&self) -> io::Result { + let (file, mmap) = if self.mmap.is_some() { + if self.filepath.as_os_str().is_empty() { + return Err(io::Error::new(io::ErrorKind::InvalidInput, "Missing filepath for mmap clone")); + } + let file = File::open(&self.filepath)?; + if let Some(mapped) = unsafe { Mmap::map(&file).ok() } { + (None, Some(mapped)) + } else { + (Some(utils::file_reader(file)), None) + } + } else if let Some(file) = &self.file { + let cloned = file.get_ref().try_clone()?; + (Some(utils::file_reader(cloned)), None) + } else { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "No data source available to clone", + )); + }; + + Ok(Self { + file, + mmap, + filepath: self.filepath.clone(), + file_identity: self.file_identity, + buffer: vec![0u8; PAGE_SIZE_USIZE], + cache: IndexMap::with_capacity(CACHE_CAPACITY), + node_cache: IndexMap::with_capacity(CACHE_CAPACITY), + root_offset: self.root_offset, + _marker_k: PhantomData, + _marker_v: PhantomData, + }) + } + /// Returns the filepath this query was opened from. pub fn filepath(&self) -> &Path { &self.filepath @@ -2993,6 +3031,10 @@ where BPlusTreeDiskIterator::new(self) } + pub(crate) fn into_sorted_parts(self) -> (PathBuf, Option>, Option) { + (self.filepath, self.file, self.mmap) + } + /// Owned iterator that traverses the tree in order defined by a secondary sorted index. /// /// The index path is automatically derived from the tree filepath by changing @@ -3180,16 +3222,18 @@ where /// Generic reader that can be either a sorted index iterator or a regular disk iterator. /// Used for fallback logic (Sorted -> Unsorted). -pub enum PlaylistIteratorReader { - Sorted(super::sorted_index::BPlusTreeSortedIteratorOwned), - Unsorted(BPlusTreeDiskIteratorOwned), +pub enum PlaylistIteratorReader { + Sorted(super::sorted_index::BPlusTreeSortedIteratorOwned), + Unsorted(BPlusTreeDiskIteratorOwned), } -impl Iterator for PlaylistIteratorReader +impl Iterator for PlaylistIteratorReader where + K: Ord + Serialize + for<'de> Deserialize<'de> + Clone, V: Serialize + for<'de> Deserialize<'de> + Clone, + SortKey: for<'de> Deserialize<'de>, { - type Item = io::Result<(u32, V)>; + type Item = io::Result<(K, V)>; fn next(&mut self) -> Option { match self { diff --git a/backend/src/repository/epg_repository.rs b/backend/src/repository/epg_repository.rs index cc7d25f13..370468f9d 100644 --- a/backend/src/repository/epg_repository.rs +++ b/backend/src/repository/epg_repository.rs @@ -23,56 +23,72 @@ pub const XML_PREAMBLE: &str = r#" // // DOCTYPE via events (DO NOT USE): // writer.write_event_async(quick_xml::events::Event::DocType(quick_xml::events::BytesText::new(r#"tv SYSTEM "xmltv.dtd""#))) // .await.map_err(|e| notify_err!("failed to write doctype: {}", e))?; -pub fn epg_write_file(target: &ConfigTarget, epg: &Epg, path: &Path, playlist: Option<&[PlaylistGroup]>) -> Result<(), TuliproxError> { +pub fn epg_write_file( + target_name: &str, + epg: &Epg, + path: &Path, + rename_map: &HashMap, Arc, S>, +) -> Result<(), TuliproxError> { if epg.children.is_empty() { return Ok(()); } - // If the epg titles differ from playlist, then we should use the ones from playlist - // Build a temporary rename map with zero allocations (uses references) - let mut rename_map: HashMap<&Arc, &Arc> = HashMap::new(); - if let Some(pl) = playlist { - for group in pl { - for channel in &group.channels { - if let Some(epg_id) = &channel.header.epg_channel_id { - if !epg_id.is_empty() { - rename_map.insert(epg_id, &channel.header.name); - } - } - } - } - } - let mut tree = BPlusTree::, EpgChannel>::new(); for channel in &epg.children { if !channel.programmes.is_empty() { let mut chan = (**channel).clone(); - if let Some(&title) = rename_map.get(&chan.id) { + if let Some(title) = rename_map.get(&chan.id) { chan.title = Some(Arc::clone(title)); } chan.programmes.sort_by_key(|p| p.start); tree.insert(Arc::clone(&channel.id), chan); } } - drop(rename_map); - tree.store(path).map_err(|err| notify_err!("Failed to write epg for target {}: {} - {err}", target.name, path.display()))?; + tree.store(path).map_err(|err| notify_err!("Failed to write epg for target {}: {} - {err}", target_name, path.display()))?; - debug_if_enabled!("Epg for target {} written to {}", target.name, path.display()); + debug_if_enabled!("Epg for target {} written to {}", target_name, path.display()); Ok(()) } +fn build_epg_rename_map(playlist: Option<&[PlaylistGroup]>) -> HashMap, Arc> { + let mut rename_map = HashMap::new(); + if let Some(pl) = playlist { + for group in pl { + for channel in &group.channels { + if let Some(epg_id) = &channel.header.epg_channel_id { + if !epg_id.is_empty() { + rename_map.insert(Arc::clone(epg_id), Arc::clone(&channel.header.name)); + } + } + } + } + } + rename_map +} + pub async fn epg_write_for_target(cfg: &Config, target: &ConfigTarget, target_path: &Path, epg: Option<&Epg>, output: &TargetOutput, playlist: Option<&[PlaylistGroup]>) -> Result<(), TuliproxError> { if let Some(epg_data) = epg { + let rename_map = Arc::new(build_epg_rename_map(playlist)); + let epg_data = Arc::new(epg_data.clone()); match output { TargetOutput::Xtream(_) => { match xtream_get_storage_path(cfg, &target.name) { Some(path) => { let epg_path = xtream_get_epg_file_path_for_target(&path); debug_if_enabled!("writing xtream epg to {}", epg_path.display()); - epg_write_file(target, epg_data, &epg_path, playlist)?; + let target_name = target.name.clone(); + let target_name_err = target_name.clone(); + let rename_map = Arc::clone(&rename_map); + let epg_data = Arc::clone(&epg_data); + let epg_path = epg_path.clone(); + tokio::task::spawn_blocking(move || { + epg_write_file(&target_name, &epg_data, &epg_path, &rename_map) + }) + .await + .map_err(|err| notify_err!("Failed to write epg for target {}: {err}", target_name_err))??; } None => return Err(notify_err!("failed to write epg for target: {}, storage path not found", target.name)), } @@ -80,7 +96,16 @@ pub async fn epg_write_for_target(cfg: &Config, target: &ConfigTarget, target_pa TargetOutput::M3u(_) => { let path = m3u_get_epg_file_path_for_target(target_path); debug_if_enabled!("writing m3u epg to {}", path.display()); - epg_write_file(target, epg_data, &path, playlist)?; + let target_name = target.name.clone(); + let target_name_err = target_name.clone(); + let rename_map = Arc::clone(&rename_map); + let epg_data = Arc::clone(&epg_data); + let path = path.clone(); + tokio::task::spawn_blocking(move || { + epg_write_file(&target_name, &epg_data, &path, &rename_map) + }) + .await + .map_err(|err| notify_err!("Failed to write epg for target {}: {err}", target_name_err))??; } TargetOutput::Strm(_) | TargetOutput::HdHomeRun(_) => {} } diff --git a/backend/src/repository/library_repository.rs b/backend/src/repository/library_repository.rs index 3219d71f6..26e328d73 100644 --- a/backend/src/repository/library_repository.rs +++ b/backend/src/repository/library_repository.rs @@ -1,6 +1,6 @@ use crate::model::AppConfig; use crate::repository::bplustree::{BPlusTree, BPlusTreeQuery}; -use shared::error::{notify_err_res, TuliproxError}; +use shared::error::{notify_err, TuliproxError}; use shared::model::{PlaylistGroup, PlaylistItem, StreamProperties, XtreamCluster, XtreamPlaylistItem}; use std::path::Path; use std::sync::Arc; @@ -9,36 +9,63 @@ use shared::model::UUIDType; use crate::repository::xtream_repository::CategoryKey; use crate::utils::file_exists_async; use std::collections::HashMap; +use tokio::task; -pub async fn persist_input_library_playlist(app_config: &Arc, library_path: &Path, playlist: &[PlaylistGroup]) -> Result<(), TuliproxError> { +pub async fn persist_input_library_playlist( + app_config: &Arc, + library_path: &Path, + playlist: Vec, +) -> (Vec, Result<(), TuliproxError>) { if playlist.is_empty() { - return Ok(()); - } - let _file_lock = app_config.file_locks.write_lock(library_path).await; - - // Keep previously probed technical metadata for unchanged local files. - let mut existing_by_uuid: HashMap = HashMap::new(); - if library_path.exists() { - if let Ok(mut query) = BPlusTreeQuery::::try_new(library_path) { - for (uuid, item) in query.iter() { - existing_by_uuid.insert(uuid, item); - } - } + return (playlist, Ok(())); } - let mut tree = BPlusTree::new(); - for pg in playlist { - for item in &pg.channels { - let mut xtream = XtreamPlaylistItem::from(item); - if let Some(existing) = existing_by_uuid.get(&item.header.uuid) { - preserve_local_probe_state_if_unchanged(&mut xtream, existing); + let file_lock = app_config.file_locks.write_lock(library_path).await; + let library_path = library_path.to_path_buf(); + let library_path_err = library_path.clone(); + let playlist = Arc::new(playlist); + let playlist_for_task = Arc::clone(&playlist); + + let result = task::spawn_blocking(move || -> Result<(), TuliproxError> { + let _guard = file_lock; + + // Keep previously probed technical metadata for unchanged local files. + let mut existing_by_uuid: HashMap = HashMap::new(); + if library_path.exists() { + if let Ok(mut query) = BPlusTreeQuery::::try_new(&library_path) { + for (uuid, item) in query.iter() { + existing_by_uuid.insert(uuid, item); + } } - tree.insert(item.header.uuid, xtream); } - } - match tree.store(library_path) { - Ok(_) => Ok(()), - Err(err) => notify_err_res!("failed to write local library playlist: {} - {err}", library_path.display()) + + let mut tree = BPlusTree::new(); + for pg in playlist_for_task.iter() { + for item in &pg.channels { + let mut xtream = XtreamPlaylistItem::from(item); + if let Some(existing) = existing_by_uuid.get(&item.header.uuid) { + preserve_local_probe_state_if_unchanged(&mut xtream, existing); + } + tree.insert(item.header.uuid, xtream); + } + } + + tree + .store(&library_path) + .map(|_| ()) + .map_err(|err| notify_err!("failed to write local library playlist: {} - {err}", library_path.display())) + }) + .await + .map_err(|err| notify_err!("failed to write local library playlist: {} - {err}", library_path_err.display())); + + let playlist = match Arc::try_unwrap(playlist) { + Ok(playlist) => playlist, + Err(playlist) => (*playlist).clone(), + }; + + match result { + Ok(res) => (playlist, res), + Err(err) => (playlist, Err(err)), } } @@ -80,32 +107,41 @@ fn preserve_local_probe_state_if_unchanged(new_item: &mut XtreamPlaylistItem, ol pub async fn load_input_local_library_playlist(app_config: &Arc, lib_path: &Path) -> Result, TuliproxError> { - let mut groups: IndexMap = IndexMap::new(); - if file_exists_async(lib_path).await { - // Load Items - let _file_lock = app_config.file_locks.read_lock(lib_path).await; - if let Ok(mut query) = BPlusTreeQuery::::try_new(lib_path) { - let mut group_cnt = 0; - for (_, ref item) in query.iter() { - let cluster = XtreamCluster::try_from(item.item_type).unwrap_or(XtreamCluster::Live); - let key = (cluster, item.group.clone()); - groups.entry(key) - .or_insert_with(|| { - group_cnt += 1; - PlaylistGroup { - id: group_cnt, - title: item.group.clone(), - channels: Vec::new(), - xtream_cluster: cluster, - } - }) - .channels.push(PlaylistItem::from(item)); + let file_lock = app_config.file_locks.read_lock(lib_path).await; + let lib_path = lib_path.to_path_buf(); + let lib_path_err = lib_path.clone(); + + let groups = task::spawn_blocking(move || -> Result, TuliproxError> { + let _guard = file_lock; + let mut groups: IndexMap = IndexMap::new(); + if let Ok(mut query) = BPlusTreeQuery::::try_new(&lib_path) { + let mut group_cnt = 0; + for (_, ref item) in query.iter() { + let cluster = XtreamCluster::try_from(item.item_type).unwrap_or(XtreamCluster::Live); + let key = (cluster, item.group.clone()); + groups.entry(key) + .or_insert_with(|| { + group_cnt += 1; + PlaylistGroup { + id: group_cnt, + title: item.group.clone(), + channels: Vec::new(), + xtream_cluster: cluster, + } + }) + .channels.push(PlaylistItem::from(item)); + } } - } + Ok(groups.into_values().collect()) + }) + .await + .map_err(|err| notify_err!("failed to read local library playlist: {} - {err}", lib_path_err.display()))??; + + return Ok(groups); } - Ok(groups.into_values().collect()) + Ok(Vec::new()) } #[cfg(test)] diff --git a/backend/src/repository/m3u_playlist_iterator.rs b/backend/src/repository/m3u_playlist_iterator.rs index 5fede5db0..5c051d4b6 100644 --- a/backend/src/repository/m3u_playlist_iterator.rs +++ b/backend/src/repository/m3u_playlist_iterator.rs @@ -1,19 +1,21 @@ use shared::error::info_err; -use shared::error::{TuliproxError}; +use shared::error::TuliproxError; use crate::model::{AppConfig, ProxyUserCredentials}; -use crate::model::{ConfigTarget}; +use crate::model::ConfigTarget; use shared::create_bitset; use shared::model::{ConfigTargetOptions, M3uPlaylistItem, PlaylistItemType, ProxyType, TargetType, XtreamCluster}; -use crate::repository::{BPlusTreeQuery, PlaylistIteratorReader}; +use crate::repository::{LockedReceiverStream, open_playlist_reader}; use crate::repository::m3u_get_file_path_for_db; use crate::repository::{ensure_target_storage_path, get_file_path_for_db_index}; use crate::repository::storage_const; use crate::repository::user_get_bouquet_filter; -use crate::utils::FileReadGuard; -use std::collections::HashSet; -use std::iter::Peekable; +use futures::Stream; +use std::pin::Pin; +use std::task::{Context, Poll}; use log::error; use shared::utils::{extract_extension_from_url, Internable}; +use tokio::sync::mpsc; +use tokio::task; create_bitset!( u8, @@ -24,21 +26,116 @@ create_bitset!( ); -struct M3uPlaylistCursor { - reader: Peekable>, - lookup_item: Option, +pub struct M3uPlaylistIterator { + inner: LockedReceiverStream<(M3uPlaylistItem, bool)>, } -pub struct M3uPlaylistIterator { - cursor: Box, - base_url: String, - username: String, - password: String, - target_options: Option, +fn build_rewritten_url( + base_url: &str, + username: &str, + password: &str, + source_url: &str, + m3u_pli: &M3uPlaylistItem, + typed: bool, + prefix_path: &str, +) -> String { + // Build URL efficiently with a single allocation using concat_string! macro + let stream_type: &str = if typed { + match m3u_pli.item_type { + PlaylistItemType::Live + | PlaylistItemType::Catchup + | PlaylistItemType::LiveUnknown + | PlaylistItemType::LiveHls + | PlaylistItemType::LiveDash => "live", + PlaylistItemType::Video | PlaylistItemType::LocalVideo => "movie", + PlaylistItemType::Series | PlaylistItemType::SeriesInfo | PlaylistItemType::LocalSeries | PlaylistItemType::LocalSeriesInfo => "series", + } + } else { + "" + }; + + let mut cap = base_url.len() + + prefix_path.len() + + username.len() + + password.len() + + 32; // separators and id + if typed { cap += stream_type.len() + 1; } + + let rewritten_url = if typed { + shared::concat_string!( + cap = cap; + base_url, "/", prefix_path, "/", stream_type, "/", + username, "/", password, "/", &m3u_pli.virtual_id.to_string() + ) + } else { + shared::concat_string!( + cap = cap; + base_url, "/", prefix_path, "/", + username, "/", password, "/", &m3u_pli.virtual_id.to_string() + ) + }; + + extract_extension_from_url(source_url) + .map(|ext| shared::concat_string!(&rewritten_url, &ext)) + .unwrap_or(rewritten_url) +} + +fn apply_rewrite( + mut m3u_pli: M3uPlaylistItem, + base_url: &str, + username: &str, + password: &str, + target_options: Option<&ConfigTargetOptions>, flags: M3uPlaylistIteratorFlagsSet, proxy_type: ProxyType, - filter: Option>, - _file_lock: FileReadGuard, +) -> M3uPlaylistItem { + let is_redirect = proxy_type.is_redirect(m3u_pli.item_type) + || target_options + .and_then(|o| o.force_redirect.as_ref()) + .is_some_and(|f| f.has_cluster(m3u_pli.item_type)); + let should_rewrite_urls = if is_redirect { + flags.contains(M3uPlaylistIteratorFlags::MaskRedirectUrl) + } else { + true + }; + + if should_rewrite_urls { + let stream_url = build_rewritten_url( + base_url, + username, + password, + m3u_pli.url.as_ref(), + &m3u_pli, + flags.contains(M3uPlaylistIteratorFlags::IncludeTypeInUrl), + storage_const::M3U_STREAM_PATH, + ); + let resource_url = if flags.contains(M3uPlaylistIteratorFlags::RewriteResource) { + let source_url = if m3u_pli.logo.is_empty() { + m3u_pli.logo_small.as_ref() + } else { + m3u_pli.logo.as_ref() + }; + Some(build_rewritten_url( + base_url, + username, + password, + source_url, + &m3u_pli, + false, + storage_const::M3U_RESOURCE_PATH, + )) + } else { + None + }; + m3u_pli.t_stream_url = stream_url.intern(); + m3u_pli.t_resource_url = resource_url; + } else { + // Keep original URL (clone required because target field is distinct) + m3u_pli.t_stream_url = m3u_pli.url.clone(); + m3u_pli.t_resource_url = None; + } + + m3u_pli } @@ -56,26 +153,8 @@ impl M3uPlaylistIterator { let target_path = ensure_target_storage_path(&config, target.name.as_str()).await?; let m3u_path = m3u_get_file_path_for_db(&target_path); - let file_lock = cfg.file_locks.read_lock(&m3u_path).await; - - let index_path = get_file_path_for_db_index(&m3u_path); - let reader = if index_path.exists() { - let query = BPlusTreeQuery::::try_new(&m3u_path) - .map_err(|err| info_err!("Could not open BPlusTreeQuery {m3u_path:?} - {err}"))?; - match query.disk_iter_sorted() { - Ok(reader) => PlaylistIteratorReader::Sorted(reader), - Err(err) => { - error!("Sorted index error for m3u, fallback: {err}"); - let query = BPlusTreeQuery::::try_new(&m3u_path) - .map_err(|err| info_err!("Could not open BPlusTreeQuery {m3u_path:?} - {err}"))?; - PlaylistIteratorReader::Unsorted(query.disk_iter()) - } - } - } else { - let query = BPlusTreeQuery::::try_new(&m3u_path) - .map_err(|err| info_err!("Could not open BPlusTreeQuery {m3u_path:?} - {err}"))?; - PlaylistIteratorReader::Unsorted(query.disk_iter()) - }.peekable(); + let iter_lock = cfg.file_locks.read_lock(&m3u_path).await; + let bg_lock = cfg.file_locks.read_lock(&m3u_path).await; let filter = user_get_bouquet_filter(&config, &user.username, None, TargetType::M3u, XtreamCluster::Live).await; let mut flags = M3uPlaylistIteratorFlagsSet::new(); @@ -90,167 +169,97 @@ impl M3uPlaylistIterator { } let server_info = cfg.get_user_server_info(user); - Ok(Self { - cursor: Box::new(M3uPlaylistCursor { - reader, - lookup_item: None, - }), - base_url: server_info.get_base_url(), - username: user.username.clone(), - password: user.password.clone(), - target_options: target.options.clone(), - flags, - filter, - proxy_type: user.proxy, - _file_lock: file_lock, // Save lock inside struct - }) - } + let base_url = server_info.get_base_url(); + let username = user.username.clone(); + let password = user.password.clone(); + let proxy_type = user.proxy; + let target_options = target.options.clone(); - fn get_rewritten_url(&self, m3u_pli: &M3uPlaylistItem, typed: bool, prefix_path: &str) -> String { - // Build URL efficiently with a single allocation using concat_string! macro - let stream_type: &str = if typed { - match m3u_pli.item_type { - PlaylistItemType::Live - | PlaylistItemType::Catchup - | PlaylistItemType::LiveUnknown - | PlaylistItemType::LiveHls - | PlaylistItemType::LiveDash => "live", - PlaylistItemType::Video | PlaylistItemType::LocalVideo => "movie", - PlaylistItemType::Series | PlaylistItemType::SeriesInfo | PlaylistItemType::LocalSeries | PlaylistItemType::LocalSeriesInfo => "series", - } - } else { - "" - }; + let m3u_path = m3u_path.clone(); + let index_path = get_file_path_for_db_index(&m3u_path); + let (tx, rx) = mpsc::channel::<(M3uPlaylistItem, bool)>(256); - let mut cap = self.base_url.len() - + prefix_path.len() - + self.username.len() - + self.password.len() - + 32; // separators and id - if typed { cap += stream_type.len() + 1; } - - let rewritten_url = if typed { - shared::concat_string!( - cap = cap; - &self.base_url, "/", prefix_path, "/", stream_type, "/", - &self.username, "/", &self.password, "/", &m3u_pli.virtual_id.to_string() - ) - } else { - shared::concat_string!( - cap = cap; - &self.base_url, "/", prefix_path, "/", - &self.username, "/", &self.password, "/", &m3u_pli.virtual_id.to_string() - ) - }; - - extract_extension_from_url(&m3u_pli.url).map(|ext| shared::concat_string!(&rewritten_url, &ext)).unwrap_or(rewritten_url) - } - - fn get_stream_url(&self, m3u_pli: &M3uPlaylistItem, typed: bool) -> String { - self.get_rewritten_url(m3u_pli, typed, storage_const::M3U_STREAM_PATH) - } - fn get_resource_url(&self, m3u_pli: &M3uPlaylistItem) -> String { - self.get_rewritten_url(m3u_pli, false, storage_const::M3U_RESOURCE_PATH) - } - - fn find_next_matching(reader: &mut Peekable>, set: &HashSet) -> Option { - loop { - match reader.next() { - Some(Ok((_, item))) => { - if set.contains(&*item.group) { - return Some(item); - } + let m3u_path_for_log = m3u_path.clone(); + let index_path_for_log = index_path.clone(); + let handle = task::spawn_blocking(move || { + let _guard = bg_lock; + let reader = match open_playlist_reader::( + &m3u_path, + &index_path, + Some("Sorted index error for m3u, fallback"), + ) { + Ok(reader) => reader, + Err(err) => { + error!("Failed to open M3U playlist DB {}: {err}", m3u_path.display()); + return; } - Some(Err(e)) => { - error!("Iterator error: {e}"); - return None; - } - None => return None, - } - } - } - - fn get_next(&mut self) -> Option<(M3uPlaylistItem, bool)> { - let entry = if let Some(set) = &self.filter { - if let Some(current_item) = self.cursor.lookup_item.take() { - // Avoid cloning strings while filtering - let next_valid = Self::find_next_matching(&mut self.cursor.reader, set); - self.cursor.lookup_item = next_valid; - let has_next = self.cursor.lookup_item.is_some(); - Some((current_item, has_next)) - } else { - let current_item = Self::find_next_matching(&mut self.cursor.reader, set); - if let Some(item) = current_item { - self.cursor.lookup_item = Self::find_next_matching(&mut self.cursor.reader, set); - let has_next = self.cursor.lookup_item.is_some(); - Some((item, has_next)) - } else { - None - } - } - } else { - match self.cursor.reader.next() { - Some(Ok((_, v))) => { - let has_next = matches!(self.cursor.reader.peek(), Some(Ok(_))); - Some((v, has_next)) - }, - Some(Err(e)) => { - error!("Iterator error: {e}"); - None - } - None => None, - } - }; - - // TODO hls and unknown reverse proxy - entry.map(|(mut m3u_pli, has_next)| { - let is_redirect = self.proxy_type.is_redirect(m3u_pli.item_type) - || self - .target_options - .as_ref() - .and_then(|o| o.force_redirect.as_ref()) - .is_some_and(|f| f.has_cluster(m3u_pli.item_type)); - let should_rewrite_urls = if is_redirect { - self.flags.contains(M3uPlaylistIteratorFlags::MaskRedirectUrl) - } else { - true }; - if should_rewrite_urls { - let stream_url = self.get_stream_url( - &m3u_pli, - self.flags - .contains(M3uPlaylistIteratorFlags::IncludeTypeInUrl), - ); - let resource_url = if self.flags.contains(M3uPlaylistIteratorFlags::RewriteResource) { - Some(self.get_resource_url(&m3u_pli)) - } else { - None + let mut pending: Option = None; + for entry in reader { + let item = match entry { + Ok((_, item)) => item, + Err(err) => { + error!("Iterator error: {err}"); + continue; + } }; - m3u_pli.t_stream_url = stream_url.intern(); - m3u_pli.t_resource_url = resource_url; - } else { - // Keep original URL (clone required because target field is distinct) - m3u_pli.t_stream_url = m3u_pli.url.clone(); - m3u_pli.t_resource_url = None; + + if let Some(set) = &filter { + if !set.contains(item.group.as_ref()) { + continue; + } + } + + let item = apply_rewrite( + item, + &base_url, + &username, + &password, + target_options.as_ref(), + flags, + proxy_type, + ); + + if let Some(prev) = pending.replace(item) { + if tx.blocking_send((prev, true)).is_err() { + return; + } + } } - (m3u_pli, has_next) + if let Some(last) = pending { + let _ = tx.blocking_send((last, false)); + } + }); + tokio::spawn(async move { + if let Err(err) = handle.await { + error!( + "M3U playlist iterator task failed for {} (index {}): {err}", + m3u_path_for_log.display(), + index_path_for_log.display() + ); + } + }); + + Ok(Self { + inner: LockedReceiverStream::new(rx, iter_lock), // Save lock inside struct }) } } -impl Iterator for M3uPlaylistIterator { +impl Stream for M3uPlaylistIterator { type Item = (M3uPlaylistItem, bool); - fn next(&mut self) -> Option { - self.get_next() + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.inner).poll_next(cx) } } pub struct M3uPlaylistM3uTextIterator { inner: M3uPlaylistIterator, started: bool, + target_options: Option, } impl M3uPlaylistM3uTextIterator { @@ -262,23 +271,27 @@ impl M3uPlaylistM3uTextIterator { Ok(Self { inner: M3uPlaylistIterator::new(cfg, target, user).await?, started: false, + target_options: target.options.clone(), }) } } -impl Iterator for M3uPlaylistM3uTextIterator { +impl Stream for M3uPlaylistM3uTextIterator { type Item = String; - fn next(&mut self) -> Option { + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { if !self.started { self.started = true; - return Some("#EXTM3U".to_string()); + return Poll::Ready(Some("#EXTM3U".to_string())); } - // TODO hls and unknown reverse proxy - self.inner.get_next().map(|(m3u_pli, _has_next)| { - let target_options = self.inner.target_options.as_ref(); - m3u_pli.to_m3u(target_options, true) - }) + match Pin::new(&mut self.inner).poll_next(cx) { + Poll::Ready(Some((m3u_pli, _has_next))) => { + let target_options = self.target_options.as_ref(); + Poll::Ready(Some(m3u_pli.to_m3u(target_options, true))) + } + Poll::Ready(None) => Poll::Ready(None), + Poll::Pending => Poll::Pending, + } } } diff --git a/backend/src/repository/m3u_repository.rs b/backend/src/repository/m3u_repository.rs index c59aabf0f..4f3f0d7a8 100644 --- a/backend/src/repository/m3u_repository.rs +++ b/backend/src/repository/m3u_repository.rs @@ -1,32 +1,43 @@ -use crate::api::model::AppState; -use crate::model::{AppConfig, ConfigInput, ProxyUserCredentials}; -use crate::model::{Config, ConfigTarget, M3uTargetOutput}; -use crate::repository::bplustree::{BPlusTree, BPlusTreeQuery}; -use crate::repository::m3u_playlist_iterator::M3uPlaylistM3uTextIterator; -use crate::repository::playlist_repository::get_input_m3u_playlist_file_path; -use crate::repository::storage::{get_input_storage_path, get_target_storage_path}; -use crate::repository::{storage_const}; -use crate::repository::xtream_repository::CategoryKey; -use crate::utils; -use crate::utils::{async_file_writer, file_exists_async, FileReadGuard, IO_BUFFER_SIZE}; +use crate::{ + api::model::AppState, + model::{AppConfig, Config, ConfigInput, ConfigTarget, M3uTargetOutput, ProxyUserCredentials}, + repository::{ + bplustree::{BPlusTree, BPlusTreeQuery}, + open_playlist_reader, + LockedReceiverStream, + m3u_playlist_iterator::M3uPlaylistM3uTextIterator, + playlist_repository::get_input_m3u_playlist_file_path, + storage::{get_file_path_for_db_index, get_input_storage_path, get_target_storage_path}, + storage_const, + xtream_repository::CategoryKey, + }, + utils, + utils::{async_file_writer, file_exists_async}, +}; use indexmap::IndexMap; use log::error; -use shared::{concat_string, notify_err_res}; -use shared::error::{notify_err, str_to_io_error, string_to_io_error, TuliproxError}; -use shared::model::{M3uPlaylistItem, PlaylistGroup}; -use shared::model::{PlaylistItem, PlaylistItemType, XtreamCluster}; -use std::io::Error; -use std::path::{Path, PathBuf}; -use std::sync::Arc; use serde::{Deserialize, Serialize}; -use tokio::fs; -use tokio::io::AsyncWriteExt; -use tokio::task; +use shared::{ + concat_string, + error::{notify_err, str_to_io_error, string_to_io_error, TuliproxError}, + model::{M3uPlaylistItem, PlaylistGroup, PlaylistItem, PlaylistItemType, XtreamCluster}, + notify_err_res, + utils::PROVIDER_SCHEME_PREFIX, +}; +use std::{ + collections::HashMap, + io::Error, + path::{Path, PathBuf}, + sync::Arc, +}; +use tokio::sync::mpsc; +use tokio_stream::Stream; +use tokio::{fs, io::AsyncWriteExt, task}; macro_rules! cant_write_result { ($path:expr, $err:expr) => { - notify_err!("failed to write m3u playlist: {} - {}", $path.display() ,$err) - } + notify_err!("failed to write m3u playlist: {} - {}", $path.display(), $err) + }; } macro_rules! await_playlist_write { @@ -38,25 +49,31 @@ macro_rules! await_playlist_write { } pub fn m3u_get_file_path_for_db(target_path: &Path) -> PathBuf { - target_path.join(storage_const::PATH_M3U).join(concat_string!(storage_const::FILE_M3U, ".", storage_const::FILE_SUFFIX_DB)) + target_path.join(storage_const::PATH_M3U).join(concat_string!( + storage_const::FILE_M3U, + ".", + storage_const::FILE_SUFFIX_DB + )) } pub fn m3u_get_epg_file_path_for_target(target_path: &Path) -> PathBuf { - let path = target_path.join(storage_const::PATH_M3U).join(concat_string!(storage_const::FILE_M3U, ".", storage_const::FILE_SUFFIX_DB)); + let path = target_path.join(storage_const::PATH_M3U).join(concat_string!( + storage_const::FILE_M3U, + ".", + storage_const::FILE_SUFFIX_DB + )); utils::add_prefix_to_filename(&path, "epg_", Some(storage_const::FILE_SUFFIX_DB)) } pub fn m3u_get_storage_path(cfg: &Config, target_name: &str) -> Option { - get_target_storage_path(cfg, target_name).map(|target_path| target_path.join(PathBuf::from(storage_const::PATH_M3U))) + get_target_storage_path(cfg, target_name) + .map(|target_path| target_path.join(PathBuf::from(storage_const::PATH_M3U))) } pub async fn ensure_m3u_storage_path(cfg: &Config, target_name: &str) -> Result { if let Some(path) = m3u_get_storage_path(cfg, target_name) { if tokio::fs::create_dir_all(&path).await.is_err() { - let msg = format!( - "Failed to save m3u data, can't create directory {}", - &path.display() - ); + let msg = format!("Failed to save m3u data, can't create directory {}", &path.display()); return notify_err_res!("{msg}"); } Ok(path) @@ -66,32 +83,38 @@ pub async fn ensure_m3u_storage_path(cfg: &Config, target_name: &str) -> Result< } } -async fn persist_m3u_playlist_as_text( - cfg: &Config, - target: &ConfigTarget, - target_output: &M3uTargetOutput, - m3u_playlist: &[M3uPlaylistItem], -) -> Result<(), TuliproxError> { - let Some(filename) = target_output.filename.as_ref() else { return Ok(()); }; - let Some(m3u_filename) = utils::get_file_path(&cfg.working_dir, Some(PathBuf::from(filename))) else { return Ok(()); }; +fn provider_m3u_filename(path: &Path) -> PathBuf { + let stem = path.file_stem().map(|s| s.to_string_lossy()).unwrap_or_default(); + let extension = path.extension().map(|ext| ext.to_string_lossy()); + let new_name = match extension { + Some(ext) if !ext.is_empty() => concat_string!(&stem, "_provider.", &ext), + _ => concat_string!(&stem, "_provider"), + }; + path.with_file_name(new_name) +} - let file = await_playlist_write!(fs::File::create(&m3u_filename), "Can't write m3u plain playlist {} - {}", m3u_filename.display()); +async fn write_m3u_text_file( + m3u_filename: &Path, + m3u_playlist: &[M3uPlaylistItem], + mut build_line: F, +) -> Result<(), TuliproxError> +where + F: FnMut(&M3uPlaylistItem) -> Result, +{ + let file = await_playlist_write!( + fs::File::create(m3u_filename), + "Can't write m3u plain playlist {} - {}", + m3u_filename.display() + ); // Larger buffer for sequential writes to reduce syscalls let mut writer = async_file_writer(file); await_playlist_write!(writer.write_all(b"#EXTM3U\n"), "Failed to write header to {} - {}", m3u_filename.display()); - let mut write_counter = 0usize; - for m3u in m3u_playlist { - let line = m3u.to_m3u(target.options.as_ref(), false); + let line = build_line(m3u)?; let bytes = line.as_bytes(); await_playlist_write!(writer.write_all(bytes), "Failed to write entry to {} - {}", m3u_filename.display()); await_playlist_write!(writer.write_all(b"\n"), "Failed to write newline to {} - {}", m3u_filename.display()); - write_counter += bytes.len() + 1; - if write_counter >= IO_BUFFER_SIZE { - await_playlist_write!(writer.flush(), "Failed to flush {} - {}", m3u_filename.display()); - write_counter = 0; - } } await_playlist_write!(writer.flush(), "Failed to flush {} - {}", m3u_filename.display()); @@ -99,6 +122,123 @@ async fn persist_m3u_playlist_as_text( Ok(()) } +fn temp_m3u_filename(path: &Path) -> PathBuf { + let file_name = path.file_name().map(|n| n.to_string_lossy()).unwrap_or_default(); + path.with_file_name(format!("{file_name}.tmp")) +} + +async fn write_m3u_text_file_atomic( + m3u_filename: &Path, + m3u_playlist: &[M3uPlaylistItem], + build_line: F, +) -> Result<(), TuliproxError> +where + F: FnMut(&M3uPlaylistItem) -> Result, +{ + let tmp_path = temp_m3u_filename(m3u_filename); + write_m3u_text_file(&tmp_path, m3u_playlist, build_line).await?; + await_playlist_write!( + fs::rename(&tmp_path, m3u_filename), + "Failed to replace {} - {}", + m3u_filename.display() + ); + Ok(()) +} + +fn replace_m3u_url_line(line: &mut String, url: &str) { + let trimmed = line.trim_end_matches(|c: char| c == '\n' || c.is_whitespace()); + let mut rebuilt = if let Some((meta, _old_url)) = trimmed.rsplit_once('\n') { + // Preserve metadata line and a single newline separator. + let mut out = String::with_capacity(meta.len() + 1 + url.len()); + out.push_str(meta); + out.push('\n'); + out + } else { + let mut out = String::with_capacity(trimmed.len() + 1 + url.len()); + out.push_str(trimmed); + out.push('\n'); + out + }; + + rebuilt.push_str(url); + *line = rebuilt; +} + +async fn persist_m3u_playlist_as_text( + app_config: &AppConfig, + target: &ConfigTarget, + target_output: &M3uTargetOutput, + m3u_playlist: &[M3uPlaylistItem], +) -> Result<(), TuliproxError> { + let Some(filename) = target_output.filename.as_ref() else { + return Ok(()); + }; + let working_dir = &app_config.config.load().working_dir; + let Some(m3u_filename) = utils::get_file_path(working_dir, Some(PathBuf::from(filename))) else { + return Ok(()); + }; + + let target_options = target.options.as_ref(); + let sources = app_config.sources.load(); + let provider_input_by_name: HashMap, Arc> = if let Some(source) = + sources.sources.iter().find(|source| source.targets.iter().any(|t| t.name == target.name)) + { + sources + .inputs + .iter() + .filter(|input| input.url.starts_with(PROVIDER_SCHEME_PREFIX)) + .filter(|input| source.inputs.contains(&input.name)) + .map(|input| (Arc::clone(&input.name), Arc::clone(input))) + .collect() + } else { + HashMap::new() + }; + + if provider_input_by_name.is_empty() { + write_m3u_text_file_atomic(&m3u_filename, m3u_playlist, |m3u| Ok(m3u.to_m3u(target_options, false))).await?; + } else { + let provider_filename = provider_m3u_filename(&m3u_filename); + let provider_tmp = temp_m3u_filename(&provider_filename); + let m3u_tmp = temp_m3u_filename(&m3u_filename); + + write_m3u_text_file(&provider_tmp, m3u_playlist, |m3u| Ok(m3u.to_m3u(target_options, false))).await?; + + write_m3u_text_file(&m3u_tmp, m3u_playlist, |m3u| { + let effective_url = if m3u.t_stream_url.is_empty() { &m3u.url } else { &m3u.t_stream_url }; + if !effective_url.starts_with(PROVIDER_SCHEME_PREFIX) { + return Ok(m3u.to_m3u(target_options, false)); + } + + let input = provider_input_by_name + .get(&m3u.input_name) + .ok_or_else(|| notify_err!("Input '{}' not found for provider URL resolution", m3u.input_name))?; + + let resolved = input.resolve_url(effective_url)?; + if resolved.as_ref() == effective_url.as_ref() { + return Ok(m3u.to_m3u(target_options, false)); + } + + let mut line = m3u.to_m3u(target_options, false); + replace_m3u_url_line(&mut line, resolved.as_ref()); + Ok(line) + }) + .await?; + + await_playlist_write!( + fs::rename(&m3u_tmp, &m3u_filename), + "Failed to replace {} - {}", + m3u_filename.display() + ); + await_playlist_write!( + fs::rename(&provider_tmp, &provider_filename), + "Failed to replace {} - {}", + provider_filename.display() + ); + } + + Ok(()) +} + pub async fn m3u_write_playlist( cfg: &AppConfig, target: &ConfigTarget, @@ -117,15 +257,15 @@ pub async fn m3u_write_playlist( let m3u_playlist = new_playlist .iter() .flat_map(|pg| &pg.channels) - .filter(|&pli| !matches!(pli.header.item_type, PlaylistItemType::SeriesInfo | PlaylistItemType::LocalSeriesInfo)) + .filter(|&pli| { + !matches!(pli.header.item_type, PlaylistItemType::SeriesInfo | PlaylistItemType::LocalSeriesInfo) + }) .map(M3uPlaylistItem::from) .collect::>(); let file_lock = cfg.file_locks.write_lock(&m3u_path).await; - if let Err(err) = persist_m3u_playlist_as_text(&cfg.config.load(), target, target_output, &m3u_playlist).await { - error!("Persisting m3u playlist failed: {err}"); - } + persist_m3u_playlist_as_text(cfg, target, target_output, &m3u_playlist).await?; let m3u_path_clone = m3u_path.clone(); @@ -136,11 +276,12 @@ pub async fn m3u_write_playlist( for m3u in m3u_playlist { tree.insert(m3u.virtual_id, m3u); } - tree.store_with_index(&m3u_path_clone, |pli| pli.source_ordinal).map_err(|err| cant_write_result!(&m3u_path_clone, err))?; + tree.store_with_index(&m3u_path_clone, |pli| pli.source_ordinal) + .map_err(|err| cant_write_result!(&m3u_path_clone, err))?; Ok(()) }) - .await - .map_err(|err| notify_err!("failed to write m3u playlist: {} - {err}", m3u_path.display()))??; + .await + .map_err(|err| notify_err!("failed to write m3u playlist: {} - {err}", m3u_path.display()))??; Ok(()) } @@ -153,8 +294,12 @@ pub async fn m3u_load_rewrite_playlist( M3uPlaylistM3uTextIterator::new(cfg, target, user).await } -pub async fn m3u_get_item_for_stream_id(stream_id: u32, app_state: &AppState, target: &ConfigTarget) -> Result { - if stream_id < 1 { +pub async fn m3u_get_item_for_stream_id( + stream_id: u32, + app_state: &AppState, + target: &ConfigTarget, +) -> Result { + if stream_id == 0 { return Err(str_to_io_error("id should start with 1")); } { @@ -168,27 +313,42 @@ pub async fn m3u_get_item_for_stream_id(stream_id: u32, app_state: &AppState, ta } let cfg: &AppConfig = &app_state.app_config; - let target_path = get_target_storage_path(&cfg.config.load(), target.name.as_str()).ok_or_else(|| string_to_io_error(format!("Could not find path for target {}", &target.name)))?; + let target_path = get_target_storage_path(&cfg.config.load(), target.name.as_str()) + .ok_or_else(|| string_to_io_error(format!("Could not find path for target {}", &target.name)))?; let m3u_path = m3u_get_file_path_for_db(&target_path); - let _file_lock = cfg.file_locks.read_lock(&m3u_path).await; + let file_lock = cfg.file_locks.read_lock(&m3u_path).await; + let m3u_path_clone = m3u_path.clone(); - let mut query = BPlusTreeQuery::::try_new(&m3u_path)?; - match query.query_zero_copy(&stream_id) { - Ok(Some(item)) => Ok(item), - Ok(None) => Err(string_to_io_error(format!("Item not found: {stream_id}"))), - Err(err) => Err(string_to_io_error(format!("Query failed for {stream_id}: {err}"))), - } + task::spawn_blocking(move || -> Result { + let _guard = file_lock; + let mut query = BPlusTreeQuery::::try_new(&m3u_path_clone)?; + match query.query_zero_copy(&stream_id) { + Ok(Some(item)) => Ok(item), + Ok(None) => Err(string_to_io_error(format!("Item not found: {stream_id}"))), + Err(err) => Err(string_to_io_error(format!("Query failed for {stream_id}: {err}"))), + } + }) + .await + .map_err(|err| string_to_io_error(format!("Query task failed for {stream_id}: {err}")))? } } -pub async fn iter_raw_m3u_target_playlist(config: &AppConfig, target: &ConfigTarget, cluster: Option) -> Option<(FileReadGuard, Box + Send>)> { +pub async fn iter_raw_m3u_target_playlist( + config: &AppConfig, + target: &ConfigTarget, + cluster: Option, +) -> Option> + Send + Unpin>> { let target_path = get_target_storage_path(&config.config.load(), target.name.as_str())?; let m3u_path = m3u_get_file_path_for_db(&target_path); iter_raw_m3u_playlist::(config, &m3u_path, cluster).await } -pub async fn iter_raw_m3u_input_playlist(app_config: &AppConfig, input: &ConfigInput, cluster: Option) -> Option<(FileReadGuard, Box + Send>)> { +pub async fn iter_raw_m3u_input_playlist( + app_config: &AppConfig, + input: &ConfigInput, + cluster: Option, +) -> Option> + Send + Unpin>> { let working_dir = &app_config.config.load().working_dir; let storage_path = get_input_storage_path(&input.name, working_dir).await.ok()?; let m3u_path = get_input_m3u_playlist_file_path(&storage_path, &input.name); @@ -196,45 +356,85 @@ pub async fn iter_raw_m3u_input_playlist(app_config: &AppConfig, input: &ConfigI iter_raw_m3u_playlist::>(app_config, &m3u_path, cluster).await } -async fn iter_raw_m3u_playlist(app_config: &AppConfig, m3u_path: &Path, cluster: Option) -> Option<(FileReadGuard, Box + Send>)> +async fn iter_raw_m3u_playlist( + app_config: &AppConfig, + m3u_path: &Path, + cluster: Option, +) -> Option> + Send + Unpin>> where ItemKey: Ord + Serialize + for<'de> Deserialize<'de> + Clone + Send + Sync + 'static, SortKey: for<'de> Deserialize<'de> + Send + 'static, { - let file_lock = app_config.file_locks.read_lock(m3u_path).await; + // Two read locks: iter_lock is held by LockedReceiverStream for the consumer lifetime, + // while bg_lock is moved into spawn_blocking to guard the on-disk reader. + let iter_lock = app_config.file_locks.read_lock(m3u_path).await; if !file_exists_async(m3u_path).await { return None; } + let bg_lock = app_config.file_locks.read_lock(m3u_path).await; - let iter: Box + Send> = { - match BPlusTreeQuery::::try_new(m3u_path) { - Ok(tree) => match tree.disk_iter_sorted::() { - Ok(sorted_iter) => Box::new(sorted_iter.filter_map(Result::ok).map(|(_, v)| v).filter(move |v| cluster.is_none_or(|c| v.item_type.is_cluster(c)))), - Err(_) => { - match BPlusTreeQuery::::try_new(m3u_path) { - Ok(tree) => Box::new(tree.disk_iter().map(|(_, v)| v).filter(move |v| cluster.is_none_or(|c| v.item_type.is_cluster(c)))), - Err(_) => return None, - } - } + let m3u_path = m3u_path.to_path_buf(); + let index_path = get_file_path_for_db_index(&m3u_path); + let (tx, rx) = mpsc::channel::>(256); + + let m3u_path_for_log = m3u_path.clone(); + let err_tx = tx.clone(); + let join_err_tx = tx.clone(); + let handle = task::spawn_blocking(move || { + let _guard = bg_lock; + let reader = match open_playlist_reader::( + &m3u_path, + &index_path, + None, + ) { + Ok(reader) => reader, + Err(err) => { + let _ = err_tx.blocking_send(Err(err)); + return; } - Err(_) => { - return None + }; + + for entry in reader { + let item = match entry { + Ok((_, item)) => item, + Err(err) => { + error!("M3U playlist reader error: {err}"); + continue; + } + }; + if cluster.is_none_or(|c| item.item_type.is_cluster(c)) + && tx.blocking_send(Ok(item)).is_err() + { + break; } } - }; + }); + tokio::spawn(async move { + if let Err(err) = handle.await { + error!("M3U playlist reader task failed for {}: {err}", m3u_path_for_log.display()); + let _ = join_err_tx.send(Err(notify_err!( + "M3U playlist reader task failed for {}: {err}", + m3u_path_for_log.display() + ))) + .await; + } + }); - Some((file_lock, iter)) + let stream: Box> + Send + Unpin> = + Box::new(LockedReceiverStream::new(rx, iter_lock)); + Some(stream) } -pub async fn persist_input_m3u_playlist(app_config: &Arc, m3u_path: &Path, playlist: &[PlaylistGroup]) -> Result<(), TuliproxError> { +pub async fn persist_input_m3u_playlist( + app_config: &Arc, + m3u_path: &Path, + playlist: &[PlaylistGroup], +) -> Result<(), TuliproxError> { let file_lock = app_config.file_locks.write_lock(m3u_path).await; let m3u_path_clone = m3u_path.to_path_buf(); - let playlist_items: Vec = playlist - .iter() - .flat_map(|pg| &pg.channels) - .map(M3uPlaylistItem::from) - .collect(); + let playlist_items: Vec = + playlist.iter().flat_map(|pg| &pg.channels).map(M3uPlaylistItem::from).collect(); task::spawn_blocking(move || -> Result<(), TuliproxError> { let _guard = file_lock; @@ -245,24 +445,34 @@ pub async fn persist_input_m3u_playlist(app_config: &Arc, m3u_path: & tree.store(&m3u_path_clone).map_err(|err| cant_write_result!(&m3u_path_clone, err))?; Ok(()) }) - .await - .map_err(|err| notify_err!("failed to write m3u playlist: {} - {err}", m3u_path.display()))??; + .await + .map_err(|err| notify_err!("failed to write m3u playlist: {} - {err}", m3u_path.display()))??; Ok(()) } -pub async fn load_input_m3u_playlist(app_config: &Arc, m3u_path: &Path) -> Result, TuliproxError> { - let mut groups: IndexMap = IndexMap::new(); +pub async fn load_input_m3u_playlist( + app_config: &Arc, + m3u_path: &Path, +) -> Result, TuliproxError> { + if !file_exists_async(m3u_path).await { + return Ok(Vec::new()); + } - if file_exists_async(m3u_path).await { - // Load Items - let _file_lock = app_config.file_locks.read_lock(m3u_path).await; - if let Ok(mut query) = BPlusTreeQuery::, M3uPlaylistItem>::try_new(m3u_path) { + let file_lock = app_config.file_locks.read_lock(m3u_path).await; + let m3u_path = m3u_path.to_path_buf(); + let m3u_path_err = m3u_path.clone(); + + let groups = task::spawn_blocking(move || -> Result, TuliproxError> { + let _guard = file_lock; + let mut groups: IndexMap = IndexMap::new(); + if let Ok(mut query) = BPlusTreeQuery::, M3uPlaylistItem>::try_new(&m3u_path) { let mut group_cnt = 0; for (_, ref item) in query.iter() { let cluster = XtreamCluster::try_from(item.item_type).unwrap_or(XtreamCluster::Live); let key = (cluster, item.group.clone()); - groups.entry(key) + groups + .entry(key) .or_insert_with(|| { group_cnt += 1; PlaylistGroup { @@ -272,10 +482,43 @@ pub async fn load_input_m3u_playlist(app_config: &Arc, m3u_path: &Pat xtream_cluster: cluster, } }) - .channels.push(PlaylistItem::from(item)); + .channels + .push(PlaylistItem::from(item)); } } + Ok(groups.into_values().collect()) + }) + .await + .map_err(|err| notify_err!("failed to read m3u playlist: {} - {err}", m3u_path_err.display()))??; + + Ok(groups) +} + +#[cfg(test)] +mod tests { + use super::replace_m3u_url_line; + + #[test] + fn replace_m3u_url_line_trims_trailing_whitespace() { + let mut line = "#EXTINF:-1 tvg-id=\"1\" tvg-name=\"Test\" group-title=\"G\",Title\nhttp://old\n\n \t".to_string(); + replace_m3u_url_line(&mut line, "http://new"); + assert_eq!( + line, + "#EXTINF:-1 tvg-id=\"1\" tvg-name=\"Test\" group-title=\"G\",Title\nhttp://new" + ); } - Ok(groups.into_values().collect()) + #[test] + fn replace_m3u_url_line_handles_crlf() { + let mut line = "#EXTINF:-1,Title\r\nhttp://old\r\n".to_string(); + replace_m3u_url_line(&mut line, "http://new"); + assert_eq!(line, "#EXTINF:-1,Title\r\nhttp://new"); + } + + #[test] + fn replace_m3u_url_line_handles_no_metadata() { + let mut line = "http://old".to_string(); + replace_m3u_url_line(&mut line, "http://new"); + assert_eq!(line, "http://old\nhttp://new"); + } } diff --git a/backend/src/repository/mod.rs b/backend/src/repository/mod.rs index 32977a650..346c2fd20 100644 --- a/backend/src/repository/mod.rs +++ b/backend/src/repository/mod.rs @@ -15,6 +15,7 @@ mod playlist_source; mod library_repository; mod sorted_index; mod alias_repository; +mod playlist_stream; pub use storage::*; pub use target_id_mapping::*; @@ -32,6 +33,7 @@ pub use sorted_index::*; pub use alias_repository::*; pub use playlist_source::*; pub use library_repository::*; +pub use playlist_stream::*; #[cfg(test)] mod bplustree_stress; diff --git a/backend/src/repository/playlist_repository.rs b/backend/src/repository/playlist_repository.rs index ec122b7c4..90b769d91 100644 --- a/backend/src/repository/playlist_repository.rs +++ b/backend/src/repository/playlist_repository.rs @@ -409,7 +409,8 @@ pub async fn persist_input_playlist(app_config: &Arc, input: &ConfigI } }; let file_path = get_input_local_library_playlist_file_path(&storage_path, &input.name); - if let Err(err) = persist_input_library_playlist(app_config, &file_path, &playlist).await { + let (playlist, result) = persist_input_library_playlist(app_config, &file_path, playlist).await; + if let Err(err) = result { return (playlist, Some(err)); } (playlist, None) diff --git a/backend/src/repository/playlist_source.rs b/backend/src/repository/playlist_source.rs index 075d8efae..6a35cae05 100644 --- a/backend/src/repository/playlist_source.rs +++ b/backend/src/repository/playlist_source.rs @@ -229,18 +229,32 @@ impl PlaylistSource for XtreamDiskPlaylistSource { } fn clone_box(&self) -> Box { - let live_path = xtream_get_file_path(&self.storage_path, XtreamCluster::Live); - let vod_path = xtream_get_file_path(&self.storage_path, XtreamCluster::Video); - let series_path = xtream_get_file_path(&self.storage_path, XtreamCluster::Series); - - let live = self.live.as_ref().and_then(|(_, guard)| { - BPlusTreeQuery::try_new(&live_path).ok().map(|q| (q, Arc::clone(guard))) + let live = self.live.as_ref().and_then(|(query, guard)| { + match query.try_clone() { + Ok(q) => Some((q, Arc::clone(guard))), + Err(err) => { + warn!("PlaylistSource::clone_box failed to clone live query: {err}"); + None + } + } }); - let vod = self.vod.as_ref().and_then(|(_, guard)| { - BPlusTreeQuery::try_new(&vod_path).ok().map(|q| (q, Arc::clone(guard))) + let vod = self.vod.as_ref().and_then(|(query, guard)| { + match query.try_clone() { + Ok(q) => Some((q, Arc::clone(guard))), + Err(err) => { + warn!("PlaylistSource::clone_box failed to clone vod query: {err}"); + None + } + } }); - let series = self.series.as_ref().and_then(|(_, guard)| { - BPlusTreeQuery::try_new(&series_path).ok().map(|q| (q, Arc::clone(guard))) + let series = self.series.as_ref().and_then(|(query, guard)| { + match query.try_clone() { + Ok(q) => Some((q, Arc::clone(guard))), + Err(err) => { + warn!("PlaylistSource::clone_box failed to clone series query: {err}"); + None + } + } }); Box::new(Self { @@ -271,7 +285,7 @@ impl PlaylistSource for XtreamDiskPlaylistSource { } macro_rules! impl_single_file_disk_source { - ($name:ident, $key_type:tt, $entry_type:tt) => { + ($name:ident, $key_type:ty, $entry_type:ty) => { paste::paste! { pub struct [<$name DiskPlaylistSource>] { app_config: Arc, @@ -324,7 +338,6 @@ macro_rules! impl_single_file_disk_source { } } - fn items<'a>(&'a mut self) -> Box> + Send + 'a> { if let Some(pl) = self.playlist.as_mut() { let iter = pl.iter().map(|(_, item)| Cow::Owned(PlaylistItem::from(&item))); @@ -383,9 +396,7 @@ macro_rules! impl_single_file_disk_source { } } fn clone_box(&self) -> Box { - let playlist = if self.playlist.is_some() && self.guard.is_some() { - BPlusTreeQuery::try_new(&self.file_path).ok() - } else { None }; + let playlist = self.playlist.as_ref().and_then(|query| query.try_clone().ok()); Box::new(Self { app_config: Arc::clone(&self.app_config), @@ -414,7 +425,7 @@ macro_rules! impl_single_file_disk_source { }; } -impl_single_file_disk_source!(M3u, String, M3uPlaylistItem); +impl_single_file_disk_source!(M3u, Arc, M3uPlaylistItem); impl_single_file_disk_source!(LocalLibrary, UUIDType, XtreamPlaylistItem); @@ -514,17 +525,27 @@ impl PlaylistSource for MemoryPlaylistSource { async fn load_bplustree_query(app_config: &Arc, file_path: &Path) -> Option<(BPlusTreeQuery, FileReadGuard)> where - K: Ord + Serialize + for<'de> Deserialize<'de> + Clone, + K: Ord + Serialize + for<'de> Deserialize<'de> + Clone + Send + 'static, P: Serialize + for<'de> Deserialize<'de> + Clone + Send + 'static, { if file_path.exists() { let guard = app_config.file_locks.read_lock(file_path).await; - match BPlusTreeQuery::::try_new(file_path) { - Ok(query) => Some((query, guard)), + let file_path = file_path.to_path_buf(); + let file_path_err = file_path.clone(); + match tokio::task::spawn_blocking(move || { + BPlusTreeQuery::::try_new(&file_path).map(|query| (query, guard)) + }) + .await + { + Ok(Ok((query, guard))) => Some((query, guard)), + Ok(Err(err)) => { + error!("Error loading disk playlist {}: {err}", file_path_err.display()); + None + } Err(err) => { - error!("Error loading disk playlist {}: {err}", file_path.display()); + error!("Error loading disk playlist {}: {err}", file_path_err.display()); None } } } else { None } -} \ No newline at end of file +} diff --git a/backend/src/repository/playlist_stream.rs b/backend/src/repository/playlist_stream.rs new file mode 100644 index 000000000..0283e3528 --- /dev/null +++ b/backend/src/repository/playlist_stream.rs @@ -0,0 +1,69 @@ +use crate::utils::FileReadGuard; +use futures::Stream; +use log::error; +use serde::{Deserialize, Serialize}; +use shared::error::{info_err, TuliproxError}; +use std::path::Path; +use std::pin::Pin; +use std::task::{Context, Poll}; +use tokio::sync::mpsc; +use tokio_stream::wrappers::ReceiverStream; + +use super::{BPlusTreeQuery, BPlusTreeSortedIteratorOwned, PlaylistIteratorReader, SortedIndexReader}; + +/// Stream wrapper that holds a file read lock for the lifetime of the stream. +pub struct LockedReceiverStream { + rx: ReceiverStream, + _guard: FileReadGuard, +} + +impl LockedReceiverStream { + pub fn new(rx: mpsc::Receiver, guard: FileReadGuard) -> Self { + Self { + rx: ReceiverStream::new(rx), + _guard: guard, + } + } +} + +impl Stream for LockedReceiverStream { + type Item = T; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.rx).poll_next(cx) + } +} + +/// Open a playlist reader with sorted-index fallback. +/// +/// NOTE: This performs disk I/O and should be used inside `spawn_blocking`. +pub fn open_playlist_reader( + path: &Path, + index_path: &Path, + sorted_err_prefix: Option<&str>, +) -> Result, TuliproxError> +where + K: Ord + Serialize + for<'de> Deserialize<'de> + Clone, + V: Serialize + for<'de> Deserialize<'de> + Clone, + SortKey: for<'de> Deserialize<'de>, +{ + let query = BPlusTreeQuery::::try_new(path) + .map_err(|err| info_err!("Could not open BPlusTreeQuery {path:?} - {err}"))?; + + if index_path.exists() { + match SortedIndexReader::::open(index_path) { + Ok(index_reader) => { + let (filepath, file, mmap) = query.into_sorted_parts(); + let reader = BPlusTreeSortedIteratorOwned::from_index_reader(index_reader, filepath, file, mmap); + return Ok(PlaylistIteratorReader::Sorted(reader)); + } + Err(err) => { + if let Some(prefix) = sorted_err_prefix { + error!("{prefix}: {err}"); + } + } + } + } + + Ok(PlaylistIteratorReader::Unsorted(query.disk_iter())) +} diff --git a/backend/src/repository/sorted_index.rs b/backend/src/repository/sorted_index.rs index 1cb397ad3..0f53a65e9 100644 --- a/backend/src/repository/sorted_index.rs +++ b/backend/src/repository/sorted_index.rs @@ -335,6 +335,22 @@ where V: for<'de> Deserialize<'de>, SortKey: for<'de> Deserialize<'de>, { + pub(crate) fn from_index_reader( + index_reader: SortedIndexReader, + filepath: PathBuf, + tree_file: Option>, + mmap: Option, + ) -> Self { + Self { + index_reader, + tree_file, + mmap, + filepath, + block_cache: IndexMap::with_capacity(CACHE_CAPACITY), + _marker: PhantomData, + } + } + /// Create a new owned sorted iterator. /// /// The index path is automatically derived from the tree filepath diff --git a/backend/src/repository/xtream_playlist_iterator.rs b/backend/src/repository/xtream_playlist_iterator.rs index 1228a4220..536084052 100644 --- a/backend/src/repository/xtream_playlist_iterator.rs +++ b/backend/src/repository/xtream_playlist_iterator.rs @@ -1,24 +1,21 @@ use crate::model::ConfigTarget; use crate::model::{xtream_mapping_option_from_target_options, AppConfig, ProxyUserCredentials}; -use crate::repository::{BPlusTreeQuery, PlaylistIteratorReader}; - +use crate::repository::{LockedReceiverStream, open_playlist_reader}; use crate::repository::user_get_bouquet_filter; use crate::repository::{xtream_get_file_path, xtream_get_storage_path}; -use crate::utils::FileReadGuard; +use futures::Stream; use log::error; use shared::error::{TuliproxError, info_err, info_err_res}; use shared::model::{PlaylistItemType, TargetType, XtreamCluster, XtreamMappingOptions, XtreamPlaylistItem}; use std::collections::HashSet; use crate::repository::get_file_path_for_db_index; +use std::pin::Pin; +use std::task::{Context, Poll}; +use tokio::sync::mpsc; +use tokio::task; pub struct XtreamPlaylistIterator { - reader: PlaylistIteratorReader, - options: XtreamMappingOptions, - cluster: XtreamCluster, - // Use parsed numeric filter to avoid per-item String allocations (no to_string per check) - filter_ids: Option>, - lookup_item: Option<(XtreamPlaylistItem, bool)>, // this is for filtered iteration - _file_lock: FileReadGuard, + inner: LockedReceiverStream<(XtreamPlaylistItem, bool)>, } impl XtreamPlaylistIterator { @@ -32,37 +29,16 @@ impl XtreamPlaylistIterator { // TODO use playlist memory cache and keep sorted - let xtream_output = target.get_xtream_output().ok_or_else(|| info_err!("Unexpected: xtream output required for target {}", target.name))?; + debug_assert!(target.get_xtream_output().is_some()); let config = app_config.config.load(); if let Some(storage_path) = xtream_get_storage_path(&config, target.name.as_str()) { let xtream_path = xtream_get_file_path(&storage_path, cluster); if !xtream_path.exists() { return info_err_res!("No {cluster} entries found for target {}", &target.name); } - let file_lock = app_config.file_locks.read_lock(&xtream_path).await; - - let index_path = get_file_path_for_db_index(&xtream_path); - let reader = if index_path.exists() { - let query = BPlusTreeQuery::::try_new(&xtream_path) - .map_err(|err| info_err!("Could not open BPlusTreeQuery {xtream_path:?} - {err}"))?; - match query.disk_iter_sorted() { - Ok(reader) => PlaylistIteratorReader::Sorted(reader), - Err(err) => { - error!("Sorted index error, falling back to unsorted: {err}"); - // Query was consumed, re-open for fallback - let query = BPlusTreeQuery::::try_new(&xtream_path) - .map_err(|err| info_err!("Could not open BPlusTreeQuery {xtream_path:?} - {err}"))?; - PlaylistIteratorReader::Unsorted(query.disk_iter()) - } - } - } else { - let query = BPlusTreeQuery::::try_new(&xtream_path) - .map_err(|err| info_err!("Could not open BPlusTreeQuery {xtream_path:?} - {err}"))?; - PlaylistIteratorReader::Unsorted(query.disk_iter()) - }; - - let server_info = app_config.get_user_server_info(user); - let options = xtream_mapping_option_from_target_options(target, xtream_output, app_config, user, Some(server_info.get_base_url().as_str())); + // Hold iter_lock for the stream lifetime (LockedReceiverStream), and bg_lock for the background reader. + let iter_lock = app_config.file_locks.read_lock(&xtream_path).await; + let bg_lock = app_config.file_locks.read_lock(&xtream_path).await; let filter = user_get_bouquet_filter(&config, &user.username, category_id, TargetType::Xtream, cluster).await; // Parse bouquet filter (strings) once into u32 set to minimize per-item allocations @@ -75,13 +51,55 @@ impl XtreamPlaylistIterator { }).collect() }); + let xtream_path = xtream_path.clone(); + let index_path = get_file_path_for_db_index(&xtream_path); + let (tx, rx) = mpsc::channel::<(XtreamPlaylistItem, bool)>(256); + + task::spawn_blocking(move || { + let _guard = bg_lock; + let reader = match open_playlist_reader::( + &xtream_path, + &index_path, + Some("Sorted index error, falling back to unsorted"), + ) { + Ok(reader) => reader, + Err(err) => { + error!( + "Failed to open Xtream playlist DB {} (cluster {cluster}): {err}", + xtream_path.display() + ); + return; + } + }; + + let mut pending: Option = None; + for entry in reader { + let item = match entry { + Ok((_, item)) => item, + Err(err) => { + error!("Error reading sorted index: {err}"); + continue; + } + }; + + if !Self::matches_filters(cluster, filter_ids.as_ref(), &item) { + continue; + } + + if let Some(prev) = pending.replace(item) { + if tx.blocking_send((prev, true)).is_err() { + return; + } + } + } + + if let Some(last) = pending { + let _ = tx.blocking_send((last, false)); + } + }); + Ok(Self { - reader, - options, - cluster, - filter_ids, - _file_lock: file_lock, - lookup_item: None, + inner: LockedReceiverStream::new(rx, iter_lock), }) } else { info_err_res!("Failed to find xtream storage for target {}", &target.name) @@ -105,53 +123,19 @@ impl XtreamPlaylistIterator { true } - /// Helper to find the next matching item from the iterator, handling `io::Result`. - fn find_next_matching(&mut self) -> Option<(XtreamPlaylistItem, bool)> { - let filter_ids = self.filter_ids.as_ref(); - let cluster = self.cluster; - - loop { - match self.reader.next() { - Some(Ok((_, item))) => { - if Self::matches_filters(cluster, filter_ids, &item) { - return Some((item, false)); - } - // Continue to next item if filter doesn't match - } - Some(Err(err)) => { - error!("Error reading sorted index: {err}"); - return None; - } - None => return None, - } - } - } - - fn get_next(&mut self) -> Option<(XtreamPlaylistItem, bool)> { - if self.lookup_item.is_none() { - self.lookup_item = self.find_next_matching(); - } - - let (current_item, _) = self.lookup_item.take()?; - - // prefetch next - self.lookup_item = self.find_next_matching(); - let has_next = self.lookup_item.is_some(); - - Some((current_item, has_next)) - } } -impl Iterator for XtreamPlaylistIterator { +impl Stream for XtreamPlaylistIterator { type Item = (XtreamPlaylistItem, bool); - fn next(&mut self) -> Option { - self.get_next() + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.inner).poll_next(cx) } } pub struct XtreamPlaylistJsonIterator { inner: XtreamPlaylistIterator, + options: XtreamMappingOptions, } impl XtreamPlaylistJsonIterator { @@ -162,23 +146,36 @@ impl XtreamPlaylistJsonIterator { category_id: Option, user: &ProxyUserCredentials, ) -> Result { + let xtream_output = target.get_xtream_output().ok_or_else(|| info_err!("Unexpected: xtream output required for target {}", target.name))?; + let server_info = config.get_user_server_info(user); + let options = xtream_mapping_option_from_target_options( + target, + xtream_output, + config, + user, + Some(server_info.get_base_url().as_str()), + ); Ok(Self { - inner: XtreamPlaylistIterator::new(cluster, config, target, category_id, user).await? + inner: XtreamPlaylistIterator::new(cluster, config, target, category_id, user).await?, + options, }) } } -impl Iterator for XtreamPlaylistJsonIterator { +impl Stream for XtreamPlaylistJsonIterator { type Item = (String, bool); - fn next(&mut self) -> Option { - self.inner.get_next().map(|(pli, has_next)| { - let json = serde_json::to_string(&pli.to_document(&self.inner.options)) - .unwrap_or_else(|err| { - error!("Failed to serialize playlist item {}: {err}", pli.virtual_id); - "{}".to_string() - }); - (json, has_next) - }) + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + match Pin::new(&mut self.inner).poll_next(cx) { + Poll::Ready(Some((pli, has_next))) => { + let json = serde_json::to_string(&pli.to_document(&self.options)) + .unwrap_or_else(|err| { + error!("Failed to serialize playlist item {}: {err}", pli.virtual_id); + "{}".to_string() + }); + Poll::Ready(Some((json, has_next))) + } + Poll::Ready(None) => Poll::Ready(None), + Poll::Pending => Poll::Pending, + } } } - diff --git a/backend/src/repository/xtream_repository.rs b/backend/src/repository/xtream_repository.rs index cf87e6c6e..5ea05b463 100644 --- a/backend/src/repository/xtream_repository.rs +++ b/backend/src/repository/xtream_repository.rs @@ -6,10 +6,11 @@ use crate::model::{ConfigInput, PlaylistXtreamCategory}; use crate::processing::parser::xtream; use crate::repository::bplustree::{BPlusTree, BPlusTreeQuery, BPlusTreeUpdate, FlushPolicy}; use crate::repository::playlist_scratch::PlaylistScratch; -use crate::repository::storage::{ensure_input_storage_path, get_input_storage_path, get_target_id_mapping_file, get_target_storage_path}; +use crate::repository::storage::{ensure_input_storage_path, get_file_path_for_db_index, get_input_storage_path, get_target_id_mapping_file, get_target_storage_path}; use crate::repository::storage_const; use crate::repository::target_id_mapping::VirtualIdRecord; use crate::repository::xtream_playlist_iterator::XtreamPlaylistJsonIterator; +use crate::repository::open_playlist_reader; use crate::utils::json_write_documents_to_file; use crate::utils::request::DynReader; use crate::utils::FileReadGuard; @@ -31,6 +32,8 @@ use std::fs::File; use std::io::{Error, ErrorKind}; use std::path::{Path, PathBuf}; use std::sync::Arc; +use tokio::sync::mpsc; +use tokio_stream::wrappers::ReceiverStream; macro_rules! cant_write_result { ($path:expr, $err:expr) => { @@ -441,15 +444,19 @@ async fn xtream_read_item_for_stream_id( cluster: XtreamCluster, ) -> Result { let xtream_path = xtream_get_file_path(storage_path, cluster); - { - let _file_lock = cfg.file_locks.read_lock(&xtream_path).await; - let mut query = BPlusTreeQuery::::try_new(&xtream_path)?; + let file_lock = cfg.file_locks.read_lock(&xtream_path).await; + let xtream_path_clone = xtream_path.clone(); + tokio::task::spawn_blocking(move || -> Result { + let _guard = file_lock; + let mut query = BPlusTreeQuery::::try_new(&xtream_path_clone)?; match query.query_zero_copy(&stream_id) { Ok(Some(item)) => Ok(item), Ok(None) => Err(Error::new(ErrorKind::NotFound, format!("Item {stream_id} not found in {cluster}"))), Err(err) => Err(Error::other(format!("Query failed for {stream_id} in {cluster}: {err}"))), } - } + }) + .await + .map_err(|err| Error::other(format!("Query task failed for {stream_id} in {cluster}: {err}")))? } async fn xtream_read_series_item_for_stream_id( @@ -458,15 +465,19 @@ async fn xtream_read_series_item_for_stream_id( storage_path: &Path, ) -> Result { let xtream_path = xtream_get_file_path(storage_path, XtreamCluster::Series); - { - let _file_lock = cfg.file_locks.read_lock(&xtream_path).await; - let mut query = BPlusTreeQuery::::try_new(&xtream_path)?; + let file_lock = cfg.file_locks.read_lock(&xtream_path).await; + let xtream_path_clone = xtream_path.clone(); + tokio::task::spawn_blocking(move || -> Result { + let _guard = file_lock; + let mut query = BPlusTreeQuery::::try_new(&xtream_path_clone)?; match query.query_zero_copy(&stream_id) { Ok(Some(item)) => Ok(item), Ok(None) => Err(Error::new(ErrorKind::NotFound, format!("Item {stream_id} not found in series"))), Err(err) => Err(Error::other(format!("Query failed for {stream_id} in series: {err}"))), } - } + }) + .await + .map_err(|err| Error::other(format!("Query task failed for {stream_id} in series: {err}")))? } @@ -572,14 +583,21 @@ pub async fn xtream_get_item_for_stream_id( xtream_read_item_for_stream_id(app_config, virtual_id, &storage_path, cluster).await } else { let target_id_mapping_file = get_target_id_mapping_file(&target_path); - let _file_lock = app_config.file_locks.read_lock(&target_id_mapping_file).await; - - let mut target_id_mapping = BPlusTreeQuery::::try_new(&target_id_mapping_file).map_err(|err| string_to_io_error(format!("Could not load id mapping for target {} err:{err}", target.name)))?; - let mapping = match target_id_mapping.query_zero_copy(&virtual_id) { - Ok(Some(record)) => Ok(record), - Ok(None) => Err(string_to_io_error(format!("Could not find mapping for target {} and id {}", target.name, virtual_id))), - Err(err) => Err(string_to_io_error(format!("Query failed for id {virtual_id}: {err}"))), - }?; + let target_name = target.name.clone(); + let file_lock = app_config.file_locks.read_lock(&target_id_mapping_file).await; + let target_id_mapping_file_clone = target_id_mapping_file.clone(); + let mapping = tokio::task::spawn_blocking(move || -> Result { + let _guard = file_lock; + let mut target_id_mapping = BPlusTreeQuery::::try_new(&target_id_mapping_file_clone) + .map_err(|err| string_to_io_error(format!("Could not load id mapping for target {target_name} err:{err}")))?; + match target_id_mapping.query_zero_copy(&virtual_id) { + Ok(Some(record)) => Ok(record), + Ok(None) => Err(string_to_io_error(format!("Could not find mapping for target {target_name} and id {virtual_id}"))), + Err(err) => Err(string_to_io_error(format!("Query failed for id {virtual_id}: {err}"))), + } + }) + .await + .map_err(|err| string_to_io_error(format!("Mapping query task failed for id {virtual_id}: {err}")))??; match mapping.item_type { PlaylistItemType::SeriesInfo | PlaylistItemType::LocalSeriesInfo => { @@ -627,50 +645,59 @@ pub async fn xtream_load_rewrite_playlist( XtreamPlaylistJsonIterator::new(cluster, config, target, category_id, user).await } -pub async fn iter_raw_xtream_target_playlist(app_config: &AppConfig, target: &ConfigTarget, cluster: XtreamCluster) -> Option<(FileReadGuard, Box + Send>)> { +pub async fn iter_raw_xtream_target_playlist(app_config: &AppConfig, target: &ConfigTarget, cluster: XtreamCluster) -> Option + Send + Unpin>> { let config = app_config.config.load(); let storage_path = xtream_get_storage_path(&config, target.name.as_str())?; let xtream_path = xtream_get_file_path(&storage_path, cluster); - iter_raw_xtream_playlist::(app_config, &xtream_path).await + iter_raw_xtream_playlist(app_config, &xtream_path).await } -pub async fn iter_raw_xtream_input_playlist(app_config: &AppConfig, input: &ConfigInput, cluster: XtreamCluster) -> Option<(FileReadGuard, Box + Send>)> { +pub async fn iter_raw_xtream_input_playlist(app_config: &AppConfig, input: &ConfigInput, cluster: XtreamCluster) -> Option + Send + Unpin>> { let config = app_config.config.load(); let working_dir = &config.working_dir; let storage_path = get_input_storage_path(&input.name, working_dir).await.ok()?; let xtream_path = xtream_get_file_path(&storage_path, cluster); - iter_raw_xtream_playlist::(app_config, &xtream_path).await + iter_raw_xtream_playlist(app_config, &xtream_path).await } -async fn iter_raw_xtream_playlist(app_config: &AppConfig, xtream_path: &Path) -> Option<(FileReadGuard, Box + Send>)> -where - ItemKey: Ord + Serialize + for<'de> Deserialize<'de> + Clone + Send + Sync + 'static, - SortKey: for<'de> Deserialize<'de> + Send + 'static, -{ - let file_lock = app_config.file_locks.read_lock(xtream_path).await; +async fn iter_raw_xtream_playlist(app_config: &AppConfig, xtream_path: &Path) -> Option + Send + Unpin>> { if !file_exists_async(xtream_path).await { return None; } + let bg_lock = app_config.file_locks.read_lock(xtream_path).await; - let iter: Box + Send> = { - match BPlusTreeQuery::::try_new(xtream_path) { - Ok(tree) => match tree.disk_iter_sorted::() { - Ok(sorted_iter) => Box::new(sorted_iter.filter_map(Result::ok).map(|(_, v)| v)), - Err(_) => { - match BPlusTreeQuery::::try_new(xtream_path) { - Ok(tree) => Box::new(tree.disk_iter().map(|(_, v)| v)), - Err(_) => return None, - } + let xtream_path = xtream_path.to_path_buf(); + let index_path = get_file_path_for_db_index(&xtream_path); + let (tx, rx) = mpsc::channel::(256); + + tokio::task::spawn_blocking(move || { + let _guard = bg_lock; + let Ok(reader) = open_playlist_reader::( + &xtream_path, + &index_path, + None, + ) else { + return; + }; + + for entry in reader { + let item = match entry { + Ok((_, item)) => item, + Err(err) => { + error!("Xtream playlist reader error: {err}"); + continue; } - } - Err(_) => { - return None + }; + if tx.blocking_send(item).is_err() { + break; } } - }; + }); - Some((file_lock, iter)) + let stream: Box + Send + Unpin> = + Box::new(ReceiverStream::new(rx)); + Some(stream) } pub fn playlist_iter_to_stream(channels: Option<(FileReadGuard, I)>) -> impl Stream> @@ -960,7 +987,7 @@ pub async fn persist_input_xtream_playlist_cluster_to_disk( // Explicitly release the lock (though RAII would handle it too) drop(swap_lock); - log::debug!("Cluster {cluster} updated and swapped successfully."); + log::debug!("Cluster {cluster} updated successfully."); Ok(()) } @@ -1014,13 +1041,31 @@ pub async fn persist_input_xtream_playlist(app_config: &Arc, storage_ let xtream_path = xtream_get_file_path(storage_path, cluster); if file_exists_async(&xtream_path).await { let file_lock = app_config.file_locks.read_lock(&xtream_path).await; - let stored_entries = stored_scratch.get_mut(cluster); - if let Ok(mut query) = BPlusTreeQuery::::try_new(&xtream_path) { - for (_, doc) in query.iter() { - stored_entries.insert(doc.provider_id, doc); + let xtream_path = xtream_path.clone(); + let stored_entries = match tokio::task::spawn_blocking(move || { + let _guard = file_lock; + let mut entries = IndexMap::new(); + if let Ok(mut query) = BPlusTreeQuery::::try_new(&xtream_path) { + for (_, doc) in query.iter() { + entries.insert(doc.provider_id, doc); + } } + entries + }) + .await + { + Ok(entries) => Some(entries), + Err(err) => { + errors.push(format!( + "Failed to load stored xtream playlist entries for {cluster}: {err}" + )); + None + } + }; + + if let Some(entries) = stored_entries { + *stored_scratch.get_mut(cluster) = entries; } - drop(file_lock); } } @@ -1380,19 +1425,34 @@ pub async fn load_input_xtream_playlist(app_config: &Arc, storage_pat } // Load Items - let _file_lock = app_config.file_locks.read_lock(&xtream_path).await; - if let Ok(mut query) = BPlusTreeQuery::::try_new(&xtream_path) { - for (_, ref item) in query.iter() { - let cat_id = item.category_id; - groups.entry((cluster, cat_id)) - .or_insert_with(|| PlaylistGroup { - id: cat_id, - title: "Unknown".intern(), - channels: Vec::new(), - xtream_cluster: cluster, - }) - .channels.push(PlaylistItem::from(item)); + let file_lock = app_config.file_locks.read_lock(&xtream_path).await; + let xtream_display = xtream_path.display().to_string(); + let xtream_path = xtream_path.clone(); + let items = tokio::task::spawn_blocking(move || -> Result, TuliproxError> { + let _guard = file_lock; + let mut items = Vec::new(); + if let Ok(mut query) = BPlusTreeQuery::::try_new(&xtream_path) { + for (_, item) in query.iter() { + items.push(item); + } } + Ok(items) + }) + .await + .map_err(|err| notify_err!("failed to read xtream playlist: {} - {err}", xtream_display))??; + + for item in items { + let cat_id = item.category_id; + groups + .entry((cluster, cat_id)) + .or_insert_with(|| PlaylistGroup { + id: cat_id, + title: "Unknown".intern(), + channels: Vec::new(), + xtream_cluster: cluster, + }) + .channels + .push(PlaylistItem::from(&item)); } } } diff --git a/backend/src/utils/db_viewer.rs b/backend/src/utils/db_viewer.rs index 9a45e27f6..656c861ca 100644 --- a/backend/src/utils/db_viewer.rs +++ b/backend/src/utils/db_viewer.rs @@ -68,10 +68,41 @@ fn dump_db(filename: &str, db_type: DbType) -> bool { } } DbType::M3u => { - if let Ok(mut query) = BPlusTreeQuery::::try_new(&path) { - let iterator = query.iter(); - return print_json_from_iter(iterator); - } + // M3U DB keys can be u32 (target playlists) or Arc (input playlists). + let err_u32 = match BPlusTreeQuery::::try_new(&path) { + Ok(mut query) => match query.len() { + Ok(_) => { + let iterator = query.iter(); + return print_json_from_iter(iterator); + } + Err(err) => { + error!("Failed to read M3U DB as u32 keys: {err}"); + Some(err.to_string()) + } + }, + Err(err) => Some(err.to_string()), + }; + + let err_str = match BPlusTreeQuery::, M3uPlaylistItem>::try_new(&path) { + Ok(mut query) => match query.len() { + Ok(_) => { + let iterator = query.iter(); + return print_json_from_iter(iterator); + } + Err(err) => { + error!("Failed to read M3U DB as string keys: {err}"); + Some(err.to_string()) + } + }, + Err(err) => Some(err.to_string()), + }; + + error!( + "Failed to open M3U DB with any known key type at {}: u32_err={:?}, string_err={:?}", + path.display(), + err_u32, + err_str + ); } DbType::Epg => { if let Ok(mut query) = BPlusTreeQuery::, EpgChannel>::try_new(&path) { diff --git a/backend/src/utils/network/request.rs b/backend/src/utils/network/request.rs index e8894e1c3..faf0c6e57 100644 --- a/backend/src/utils/network/request.rs +++ b/backend/src/utils/network/request.rs @@ -15,10 +15,7 @@ use reqwest::redirect::Policy; use reqwest::StatusCode; use shared::error::{notify_err_res, string_to_io_error, TuliproxError}; use shared::model::{format_elapsed_time, InputFetchMethod, DEFAULT_USER_AGENT}; -use shared::utils::{ - filter_request_header, human_readable_byte_size, sanitize_sensitive_info, ENCODING_DEFLATE, - ENCODING_GZIP, -}; +use shared::utils::{filter_request_header, human_readable_byte_size, sanitize_sensitive_info, CONTENT_TYPE_JSON, ENCODING_DEFLATE, ENCODING_GZIP}; use std::collections::HashMap; use std::io::{Error, ErrorKind}; use std::path::{Path, PathBuf}; @@ -103,7 +100,7 @@ pub fn classify_content_type(headers: &[(String, String)]) -> MimeCategory { v if v.starts_with("video/") || v == "application/octet-stream" => MimeCategory::Video, v if v.contains("mpegurl") => MimeCategory::M3U8, v if v.starts_with("image/") => MimeCategory::Image, - v if v.starts_with("application/json") || v.ends_with("+json") => MimeCategory::Json, + v if v.starts_with(CONTENT_TYPE_JSON) || v.ends_with("+json") => MimeCategory::Json, v if v.starts_with("application/xml") || v.ends_with("+xml") || v == "text/xml" => MimeCategory::Xml, v if v.starts_with("text/") => MimeCategory::Text, _ => MimeCategory::Unclassified, diff --git a/frontend/public/assets/i18n/en.json b/frontend/public/assets/i18n/en.json index 845e11897..effcfa57e 100644 --- a/frontend/public/assets/i18n/en.json +++ b/frontend/public/assets/i18n/en.json @@ -368,6 +368,7 @@ "ADD_WATCH": "Add Watch", "ADD_MAPPING": "Add Mapping", "RESOLVE_DELAY_SEC": "Resolve Delay (sec)", + "PROBE_DELAY_SEC": "Probe Delay (sec)", "RESOLVE_BACKGROUND": "Resolve in Background", "PROBE": "Probe", "PROBE_LIVE_INTERVAL_HOURS": "Live Interval (hour)", diff --git a/frontend/src/app/components/playlist/input/input_options.rs b/frontend/src/app/components/playlist/input/input_options.rs index bff02b003..89964d2ce 100644 --- a/frontend/src/app/components/playlist/input/input_options.rs +++ b/frontend/src/app/components/playlist/input/input_options.rs @@ -1,7 +1,7 @@ use crate::app::components::{chip::convert_bool_to_chip_style, make_tags, CollapsePanel, Tag, TagList}; use shared::{ model::ConfigInputDto, - utils::{default_probe_live_interval, default_resolve_delay_secs}, + utils::{default_probe_delay_secs, default_probe_live_interval, default_resolve_delay_secs}, }; use std::rc::Rc; use yew::prelude::*; @@ -52,6 +52,10 @@ pub fn InputOptions(props: &InputOptionsProps) -> Html { class: convert_bool_to_chip_style(false), label: translate.t("LABEL.SERIES").to_string(), }), + Rc::new(Tag { + class: convert_bool_to_chip_style(false), + label: format!("{} / {}s", translate.t("LABEL.PROBE_DELAY_SEC"), default_probe_delay_secs()), + }), ], ), Some(options) => { @@ -68,6 +72,7 @@ pub fn InputOptions(props: &InputOptionsProps) -> Html { || options.probe_vod || options.probe_live || options.resolve_delay != default_resolve_delay_secs() + || options.probe_delay != default_probe_delay_secs() || options.probe_live_interval_hours != default_probe_live_interval(); ( @@ -109,6 +114,10 @@ pub fn InputOptions(props: &InputOptionsProps) -> Html { class: convert_bool_to_chip_style(options.probe_series), label: translate.t("LABEL.SERIES").to_string(), }), + Rc::new(Tag { + class: convert_bool_to_chip_style(options.probe_delay != default_probe_delay_secs()), + label: format!("{} / {}s", translate.t("LABEL.PROBE_DELAY_SEC"), options.probe_delay), + }), ], ) } diff --git a/frontend/src/app/components/source_editor/input_form.rs b/frontend/src/app/components/source_editor/input_form.rs index b83b12282..60c008872 100644 --- a/frontend/src/app/components/source_editor/input_form.rs +++ b/frontend/src/app/components/source_editor/input_form.rs @@ -49,6 +49,7 @@ const LABEL_RESOLVE_TMDB: &str = "LABEL.RESOLVE_TMDB"; const LABEL_RESOLVE: &str = "LABEL.RESOLVE"; const LABEL_PROBE: &str = "LABEL.PROBE"; const LABEL_RESOLVE_DELAY_SEC: &str = "LABEL.RESOLVE_DELAY_SEC"; +const LABEL_PROBE_DELAY_SEC: &str = "LABEL.PROBE_DELAY_SEC"; const LABEL_RESOLVE_BACKGROUND: &str = "LABEL.RESOLVE_BACKGROUND"; const LABEL_PROBE_LIVE_INTERVAL_HOURS: &str = "LABEL.PROBE_LIVE_INTERVAL_HOURS"; const LABEL_METADATA: &str = "LABEL.METADATA"; @@ -122,6 +123,7 @@ generate_form_reducer!( ResolveSeries => resolve_series: bool, ResolveVod => resolve_vod: bool, ResolveDelay => resolve_delay: u16, + ProbeDelay => probe_delay: u16, ProbeLive => probe_live: bool, ProbeVod => probe_vod: bool, ProbeSeries => probe_series: bool, @@ -452,6 +454,7 @@ pub fn ConfigInputView(props: &ConfigInputViewProps) -> Html { { edit_field_bool!(input_options_state, translate.t(LABEL_XTREAM_SKIP_SERIES), probe_series, ConfigInputOptionsFormAction::ProbeSeries) }

+ { edit_field_number_u16!(input_options_state, translate.t(LABEL_PROBE_DELAY_SEC), probe_delay, ConfigInputOptionsFormAction::ProbeDelay) } { edit_field_number_u32!(input_options_state, translate.t(LABEL_PROBE_LIVE_INTERVAL_HOURS), probe_live_interval_hours, ConfigInputOptionsFormAction::ProbeLiveIntervalHours) }
diff --git a/shared/src/model/config/input.rs b/shared/src/model/config/input.rs index 4bea1be9f..d2901df03 100644 --- a/shared/src/model/config/input.rs +++ b/shared/src/model/config/input.rs @@ -5,12 +5,12 @@ use crate::{ info_err_res, model::EpgConfigDto, utils::{ - arc_str_serde, arc_str_vec_serde, default_as_true, default_probe_live_interval, default_resolve_background, - default_resolve_delay_secs, default_xtream_live_stream_use_prefix, deserialize_timestamp, - get_credentials_from_url_str, get_trimmed_string, is_blank_optional_string, is_default_probe_live_interval, - is_default_resolve_delay_secs, is_false, is_true, is_zero_u16, parse_provider_scheme_url_parts, - sanitize_sensitive_info, serialize_option_vec_flow_map_items, trim_last_slash, Internable, - PROVIDER_SCHEME_PREFIX, + arc_str_serde, arc_str_vec_serde, default_as_true, default_probe_delay_secs, default_probe_live_interval, + default_resolve_background, default_resolve_delay_secs, default_xtream_live_stream_use_prefix, + deserialize_timestamp, get_credentials_from_url_str, get_trimmed_string, is_blank_optional_string, + is_default_probe_delay_secs, is_default_probe_live_interval, is_default_resolve_delay_secs, is_false, is_true, + is_zero_u16, parse_provider_scheme_url_parts, sanitize_sensitive_info, serialize_option_vec_flow_map_items, + trim_last_slash, Internable, PROVIDER_SCHEME_PREFIX, }, }; use enum_iterator::Sequence; @@ -189,6 +189,8 @@ pub struct ConfigInputOptionsDto { pub probe_vod: bool, #[serde(default = "default_resolve_delay_secs", skip_serializing_if = "is_default_resolve_delay_secs")] pub resolve_delay: u16, + #[serde(default = "default_probe_delay_secs", skip_serializing_if = "is_default_probe_delay_secs")] + pub probe_delay: u16, #[serde(default, alias = "resolve_live", skip_serializing_if = "is_false")] pub probe_live: bool, #[serde( @@ -214,6 +216,7 @@ impl Default for ConfigInputOptionsDto { probe_series: false, probe_vod: false, resolve_delay: default_resolve_delay_secs(), + probe_delay: default_probe_delay_secs(), probe_live: false, probe_live_interval_hours: default_probe_live_interval(), } @@ -234,6 +237,7 @@ impl ConfigInputOptionsDto { && !self.probe_series && !self.probe_vod && is_default_resolve_delay_secs(&self.resolve_delay) + && is_default_probe_delay_secs(&self.probe_delay) && !self.probe_live && is_default_probe_live_interval(&self.probe_live_interval_hours) } @@ -251,6 +255,7 @@ impl ConfigInputOptionsDto { self.probe_series = false; self.probe_vod = false; self.resolve_delay = default_resolve_delay_secs(); + self.probe_delay = default_probe_delay_secs(); self.probe_live = false; self.probe_live_interval_hours = default_probe_live_interval(); } diff --git a/shared/src/utils/default_utils.rs b/shared/src/utils/default_utils.rs index 339cca137..a896482a2 100644 --- a/shared/src/utils/default_utils.rs +++ b/shared/src/utils/default_utils.rs @@ -21,6 +21,10 @@ pub fn default_as_default() -> String { "default".into() } // used to prevent frequent requests that could trigger a provider ban. pub const fn default_resolve_delay_secs() -> u16 { 2 } pub const fn is_default_resolve_delay_secs(v: &u16) -> bool { *v == default_resolve_delay_secs() } +// Default delay values for probing streams (ffprobe), +// used to avoid excessive probing under rapid playlist changes. +pub const fn default_probe_delay_secs() -> u16 { 2 } +pub const fn is_default_probe_delay_secs(v: &u16) -> bool { *v == default_probe_delay_secs() } // Default grace values to accommodate rapid channel changes and seek requests, // helping avoid triggering hard max_connection enforcement. pub const fn default_grace_period_millis() -> u64 { 2000 }