From fae862b78a2671afb0e3bb45fca8e648f541176c Mon Sep 17 00:00:00 2001 From: euzu <33094714+euzu@users.noreply.github.com> Date: Fri, 16 Oct 2026 10:42:04 +0200 Subject: [PATCH] Feature/hdhomerun auto channel (#920) * hdhomnerun auto_channel --- .../app/src/api/endpoints/hdhomerun_api.rs | 1248 +++++++++++++++-- backend/repository/src/lib.rs | 2 + backend/repository/src/lineup_guide_index.rs | 623 ++++++++ .../repository/src/m3u_playlist_iterator.rs | 189 ++- .../src/playlist_repository/persist.rs | 176 ++- .../src/xtream_playlist_iterator.rs | 173 ++- 6 files changed, 2199 insertions(+), 212 deletions(-) create mode 100644 backend/repository/src/lineup_guide_index.rs diff --git a/backend/app/src/api/endpoints/hdhomerun_api.rs b/backend/app/src/api/endpoints/hdhomerun_api.rs index 5a6352211..c729ada37 100644 --- a/backend/app/src/api/endpoints/hdhomerun_api.rs +++ b/backend/app/src/api/endpoints/hdhomerun_api.rs @@ -1,12 +1,17 @@ use crate::{ api::{ api_utils::{coalesce_byte_stream, internal_server_error, try_unwrap_body}, - model::HdHomerunAppState, + model::{AppState, HdHomerunAppState}, }, auth::{try_check_network_access_only, AuthBasic, Fingerprint}, model::{AppConfig, ConfigInputFlags, ConfigTarget, ProxyUserCredentials}, processing::parser::xtream::get_xtream_url, - repository::{iter_raw_m3u_target_playlist, M3uPlaylistIterator, XtreamPlaylistIterator}, + repository::{ + get_target_storage_path, hdhomerun_lineup_view, iter_raw_m3u_target_playlist, lineup_guide_candidates, + lineup_guide_index_view, m3u_lineup_guide_index_path, xtream_lineup_guide_index_path, LineupGuideCandidate, + LineupGuideLookup, LineupGuideView, M3uPlaylistIterator, M3uUserItemLookup, XtreamPlaylistIterator, + XtreamUserItemLookup, + }, utils::arc_str_serde, }; use axum::response::IntoResponse; @@ -20,7 +25,7 @@ use shared::{ model::{M3uPlaylistItem, PlaylistItemType, TargetType, XtreamCluster, XtreamPlaylistItem}, utils::concat_path, }; -use std::sync::Arc; +use std::{path::PathBuf, sync::Arc}; #[derive(Serialize, Deserialize, Clone)] struct Lineup { @@ -108,6 +113,58 @@ fn serialize_lineup(lineup: &Lineup, has_next: bool) -> Result { } } +fn get_xtream_item_stream_url( + cfg: &Arc, + cluster: XtreamCluster, + credentials: &ProxyUserCredentials, + base_url: Option<&str>, + item: &XtreamPlaylistItem, +) -> Arc { + let input = cfg.get_input_by_name(&item.input_name); + let (live_stream_use_prefix, live_stream_without_extension) = input.as_ref().map_or((true, false), |i| { + ( + i.has_flag(ConfigInputFlags::XtreamLiveStreamUsePrefix), + i.has_flag(ConfigInputFlags::XtreamLiveStreamWithoutExtension), + ) + }); + let container_extension = item.get_container_extension(); + match base_url { + None => item.url.clone(), + Some(url) => get_xtream_url( + cluster, + url, + &credentials.username, + &credentials.password, + item.virtual_id.get(), + container_extension.as_deref(), + live_stream_use_prefix, + live_stream_without_extension, + ) + .into(), + } +} + +#[inline] +fn get_m3u_item_stream_url(item: &M3uPlaylistItem) -> Arc { + if item.t_stream_url.is_empty() { + item.url.clone() + } else { + item.t_stream_url.clone() + } +} + +/// Base urls for xtream live and vod stream urls; `None` means the provider url is used (redirect). +fn xtream_base_urls( + credentials: &ProxyUserCredentials, + target: &ConfigTarget, + base_url: String, +) -> (Option, Option) { + let use_base_url = |item_type| !(credentials.proxy.is_redirect(item_type) || target.is_force_redirect(item_type)); + let live = use_base_url(PlaylistItemType::Live).then(|| base_url.clone()); + let vod = use_base_url(PlaylistItemType::Video).then_some(base_url); + (live, vod) +} + fn xtream_item_to_lineup_stream( cfg: Arc, cluster: XtreamCluster, @@ -122,40 +179,21 @@ where Some(chans) => { let mapped = chans.map(move |entry| { let (item, has_next) = entry.map_err(|error| error.to_string())?; - let input = cfg.get_input_by_name(&item.input_name); - let (live_stream_use_prefix, live_stream_without_extension) = - input.as_ref().map_or((true, false), |i| { - ( - i.has_flag(ConfigInputFlags::XtreamLiveStreamUsePrefix), - i.has_flag(ConfigInputFlags::XtreamLiveStreamWithoutExtension), - ) - }); - let container_extension = item.get_container_extension(); - let stream_url = match &base_url { - None => item.url.clone(), - Some(url) => get_xtream_url( - cluster, - url, - &credentials.username, - &credentials.password, - item.virtual_id.get(), - container_extension.as_deref(), - live_stream_use_prefix, - live_stream_without_extension, - ) - .into(), - }; + let stream_url = get_xtream_item_stream_url(&cfg, cluster, &credentials, base_url.as_deref(), &item); let lineup = Lineup { guide_number: item.epg_channel_id.unwrap_or(item.name.clone()), guide_name: item.title.clone(), - url: stream_url.clone(), + url: stream_url, }; serialize_lineup(&lineup, has_next) }); mapped.left_stream() } - None => stream::once(async { Ok(Bytes::from("")) }).right_stream(), + // A missing cluster contributes nothing. A placeholder chunk would still be visible to the + // caller that peeks at both clusters to decide whether a separator is needed, which ends the + // lineup on a trailing comma. + None => stream::empty().right_stream(), } } @@ -170,13 +208,13 @@ where let lineup = Lineup { guide_number: item.epg_channel_id.clone().unwrap_or(item.name.clone()), guide_name: item.title.clone(), - url: if item.t_stream_url.is_empty() { item.url.clone() } else { item.t_stream_url.clone() }, + url: get_m3u_item_stream_url(&item), }; serialize_lineup(&lineup, has_next) }); mapped.left_stream() } - None => stream::once(async { Ok(Bytes::from("")) }).right_stream(), + None => stream::empty().right_stream(), } } @@ -372,11 +410,8 @@ async fn lineup( credentials: &Arc, target: &ConfigTarget, ) -> impl IntoResponse { - let use_output = target.get_hdhomerun_output().as_ref().and_then(|o| o.use_output); - let use_all = use_output.is_none(); - let use_m3u = use_output.as_ref() == Some(&TargetType::M3u); - let use_xtream = use_output.as_ref() == Some(&TargetType::Xtream); - if (use_all || use_m3u) && target.has_output(TargetType::M3u) { + let view = hdhomerun_lineup_view(target); + if view == Some(LineupGuideView::M3u) { let iterator = M3uPlaylistIterator::new(cfg, target, credentials).await.ok(); let stream = m3u_item_to_lineup_stream(iterator); let body_stream = stream::once(async { Ok(Bytes::from("[")) }) @@ -386,26 +421,11 @@ async fn lineup( .status(axum::http::StatusCode::OK) .header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string()) .body(axum::body::Body::from_stream(coalesce_byte_stream(body_stream)))); - } else if (use_all || use_xtream) && target.has_output(TargetType::Xtream) { + } else if view == Some(LineupGuideView::Xtream) { let Some(server_info) = app_state.app_state.app_config.get_user_server_info(credentials) else { return axum::http::StatusCode::SERVICE_UNAVAILABLE.into_response(); }; - let base_url = server_info.get_base_url(); - - let base_url_live = if credentials.proxy.is_redirect(PlaylistItemType::Live) - || target.is_force_redirect(PlaylistItemType::Live) - { - None - } else { - Some(base_url.clone()) - }; - let base_url_vod = if credentials.proxy.is_redirect(PlaylistItemType::Video) - || target.is_force_redirect(PlaylistItemType::Video) - { - None - } else { - Some(base_url) - }; + let (base_url_live, base_url_vod) = xtream_base_urls(credentials, target, server_info.get_base_url()); let live_channels = if credentials.allows_cluster(XtreamCluster::Live) { XtreamPlaylistIterator::new(XtreamCluster::Live, cfg, target, None, credentials).await.ok() @@ -456,56 +476,407 @@ async fn lineup( axum::http::StatusCode::NOT_FOUND.into_response() } +/// Resolves the device user and checks basic auth (when given) and network access. +fn authorize_device_user( + app_state: &HdHomerunAppState, + fingerprint: &Fingerprint, + basic_auth: Option<&(String, String)>, +) -> Result<(Arc, Arc), Box> { + let cfg = &app_state.app_state.app_config; + let Some((credentials, target)) = cfg.get_target_for_username(&app_state.device.t_username) else { + return Err(Box::new(axum::http::StatusCode::NOT_FOUND.into_response())); + }; + if let Some((username, password)) = basic_auth { + if *username != credentials.username || *password != credentials.password { + return Err(Box::new(axum::http::StatusCode::UNAUTHORIZED.into_response())); + } + } + try_check_network_access_only(&credentials, fingerprint, cfg, &app_state.app_state.geoip) + .map_err(|err| Box::new(err.into_player_response(cfg.get_auth_error_status())))?; + Ok((credentials, target)) +} + +async fn authorized_lineup( + app_state: &Arc, + fingerprint: &Fingerprint, + basic_auth: Option<&(String, String)>, +) -> axum::response::Response { + match authorize_device_user(app_state, fingerprint, basic_auth) { + Ok((credentials, target)) => { + lineup(app_state, &app_state.app_state.app_config, &credentials, &target).await.into_response() + } + Err(response) => *response, + } +} + async fn auth_lineup_json( fingerprint: Fingerprint, - AuthBasic((username, password)): AuthBasic, + AuthBasic(basic_auth): AuthBasic, axum::extract::State(app_state): axum::extract::State>, ) -> impl IntoResponse { - let cfg = Arc::clone(&app_state.app_state.app_config); - if let Some((credentials, target)) = cfg.get_target_for_username(&app_state.device.t_username) { - if !username.eq(&credentials.username) || !password.eq(&credentials.password) { - return axum::http::StatusCode::UNAUTHORIZED.into_response(); - } - if let Err(e) = try_check_network_access_only( - &credentials, - &fingerprint, - &app_state.app_state.app_config, - &app_state.app_state.geoip, - ) { - return e.into_player_response(app_state.app_state.app_config.get_auth_error_status()); - } - let user_credentials = Arc::clone(&credentials); - return lineup(&app_state, &cfg, &user_credentials, &target).await.into_response(); - } - axum::http::StatusCode::NOT_FOUND.into_response() + authorized_lineup(&app_state, &fingerprint, Some(&basic_auth)).await } async fn lineup_json( fingerprint: Fingerprint, axum::extract::State(app_state): axum::extract::State>, ) -> impl IntoResponse { - let cfg = Arc::clone(&app_state.app_state.app_config); - if let Some((credentials, target)) = cfg.get_target_for_username(&app_state.device.t_username) { - if let Err(e) = try_check_network_access_only( - &credentials, - &fingerprint, - &app_state.app_state.app_config, - &app_state.app_state.geoip, - ) { - return e.into_player_response(app_state.app_state.app_config.get_auth_error_status()); + authorized_lineup(&app_state, &fingerprint, None).await +} + +/// Channel identifier requested via `/auto/{channel}` or `/tuner{n}/{channel}`. +/// +/// Clients request the `GuideNumber` from `lineup.json`, optionally as `v` and/or with a +/// `.ts` suffix. The `GuideNumber` is the epg channel id or the name, so a leading `v` is only +/// treated as virtual channel prefix when the remainder is numeric (`v101`, `v2.1`); `vox.de` +/// stays untouched. +#[derive(Debug, PartialEq)] +struct ChannelQuery<'a> { + raw: &'a str, + numeric: Option<&'a str>, + virtual_id: Option, +} + +impl<'a> ChannelQuery<'a> { + fn parse(channel: &'a str) -> Option { + let trimmed = channel.trim(); + let raw = match trimmed.rsplit_once('.') { + Some((stem, ext)) if ext.eq_ignore_ascii_case("ts") => stem.trim(), + _ => trimmed, + }; + if raw.is_empty() { + return None; } - let user_credentials = Arc::clone(&credentials); - return lineup(&app_state, &cfg, &user_credentials, &target).await.into_response(); + let numeric = raw + .strip_prefix(['v', 'V']) + .filter(|rest| !rest.is_empty() && rest.bytes().all(|b| b.is_ascii_digit() || b == b'.')); + let virtual_id = numeric.unwrap_or(raw).parse::().ok(); + Some(Self { raw, numeric, virtual_id }) } - axum::http::StatusCode::NOT_FOUND.into_response() + + /// Match quality, lower is better: exact `GuideNumber`, `GuideNumber` without `v` prefix, + /// virtual id. The `GuideNumber` wins so a numeric guide number never resolves to another + /// channel that happens to carry the same virtual id. + fn rank(&self, channel: &C) -> Option { + let guide_number = channel.guide_number(); + if guide_number == self.raw { + Some(0) + } else if self.numeric == Some(guide_number) { + Some(1) + } else if self.virtual_id == Some(channel.virtual_id()) { + Some(2) + } else { + None + } + } +} + +trait LineupChannel { + fn guide_number(&self) -> &str; + fn virtual_id(&self) -> u32; +} + +impl LineupChannel for M3uPlaylistItem { + fn guide_number(&self) -> &str { self.epg_channel_id.as_deref().unwrap_or(&self.name) } + fn virtual_id(&self) -> u32 { self.virtual_id.get() } +} + +impl LineupChannel for XtreamPlaylistItem { + fn guide_number(&self) -> &str { self.epg_channel_id.as_deref().unwrap_or(&self.name) } + fn virtual_id(&self) -> u32 { self.virtual_id.get() } +} + +/// Best matching channel of a lineup iterator together with its rank. +/// +/// Uses the same user scoped iterators as `lineup.json`, so bouquet, user filter and url rewrite +/// apply exactly as for the published lineup. +async fn find_channel(mut channels: S, query: &ChannelQuery<'_>) -> Option<(u8, C)> +where + C: LineupChannel, + S: Stream> + Unpin, +{ + let mut best: Option<(u8, C)> = None; + while let Some(entry) = channels.next().await { + let Ok((channel, _has_next)) = entry else { continue }; + let Some(rank) = query.rank(&channel) else { continue }; + if rank == 0 { + return Some((rank, channel)); + } + if best.as_ref().is_none_or(|(best_rank, _)| rank < *best_rank) { + best = Some((rank, channel)); + } + } + best +} + +/// Keys the guide index is asked for, in the order the lineup scan ranks them: the exact guide +/// number, then the guide number without a leading `v`. +/// +/// Both are the strings `lineup.json` publishes. They are plain request data, so they are not +/// interned into the shared pool. +fn guide_lookup_keys(query: &ChannelQuery<'_>) -> Vec> { + query.numeric.map_or_else(|| vec![Arc::from(query.raw)], |numeric| vec![Arc::from(query.raw), Arc::from(numeric)]) +} + +/// Guide index file of `view`, when the target publishes an `HDHomeRun` lineup from it. +/// +/// Read path only: the target directory is not created here. +fn guide_index_path(app_state: &AppState, target: &ConfigTarget, view: LineupGuideView) -> Option { + if lineup_guide_index_view(target) != Some(view) { + return None; + } + let config = app_state.app_config.config.load(); + match view { + LineupGuideView::M3u => get_target_storage_path(&config, target.name.as_str()) + .map(|target_path| m3u_lineup_guide_index_path(&target_path)), + LineupGuideView::Xtream => xtream_lineup_guide_index_path(&config, &target.name), + } +} + +/// Candidates the guide index of `view` holds for `query`, or `None` without a usable index. +async fn guide_index_candidates( + app_state: &AppState, + target: &ConfigTarget, + view: LineupGuideView, + query: &ChannelQuery<'_>, +) -> Option> { + let index_path = guide_index_path(app_state, target, view)?; + match lineup_guide_candidates(&app_state.app_config.file_locks, index_path, guide_lookup_keys(query)).await { + LineupGuideLookup::Candidates(candidates) => Some(candidates), + LineupGuideLookup::NoEntry => Some(Vec::new()), + LineupGuideLookup::Missing => None, + } +} + +/// Clusters the lineup of the Xtream view publishes, in the order the scan searches them. +const LINEUP_CLUSTERS: [XtreamCluster; 2] = [XtreamCluster::Live, XtreamCluster::Video]; + +/// Answer of a target guide index. +enum GuideIndexAnswer { + /// The index named a channel this user's output publishes. + Url(Arc), + /// The index is usable and nothing this request can address is published to this user, so the + /// target does not serve it. Answering that without scanning is what keeps an unknown channel + /// cheap: the index follows every rewrite of the playlist db it describes, and the writers + /// outside that step only touch stream metadata and episodes, never a guide number. + NotServed, + /// No usable index. The caller falls back to its playlist scan, which stays the reference. + Unavailable, +} + +/// Answers a tune request from the guide index of the M3U view. +/// +/// Walks the candidates in the order the scan ranks them: every feed of the exact guide number, +/// then of the guide number without `v`, both in lineup order, and finally the virtual id itself. +/// The first one this user's output publishes wins. Channel failover will pick among the same +/// candidates by quality of service instead. +async fn resolve_m3u_channel_via_guide_index( + app_state: &AppState, + credentials: &Arc, + target: &ConfigTarget, + query: &ChannelQuery<'_>, +) -> GuideIndexAnswer { + let Some(candidates) = guide_index_candidates(app_state, target, LineupGuideView::M3u, query).await else { + return GuideIndexAnswer::Unavailable; + }; + // Nothing to look up: answer before preparing the per-user pipeline. + if candidates.is_empty() && query.virtual_id.is_none() { + return GuideIndexAnswer::NotServed; + } + let lookup = match M3uUserItemLookup::new(&app_state.app_config, &app_state.playlists, target, credentials).await { + Ok(lookup) => lookup, + Err(err) => { + warn!("Failed to prepare m3u output of target {} for HDHomeRun: {err}", target.name); + return GuideIndexAnswer::Unavailable; + } + }; + + for LineupGuideCandidate { key, entries } in candidates { + for entry in entries { + // A feed that is gone or hidden for this user does not end the walk: the next feed of + // the same guide number may be the one the lineup publishes. An item that no longer + // carries the guide number it was found under means the index is behind. + if let Some(item) = lookup.get(entry.virtual_id).await { + if item.guide_number() == key.as_ref() { + return GuideIndexAnswer::Url(get_m3u_item_stream_url(&item)); + } + } + } + } + if let Some(stream_id) = query.virtual_id { + if let Some(item) = lookup.get(stream_id).await { + return GuideIndexAnswer::Url(get_m3u_item_stream_url(&item)); + } + } + GuideIndexAnswer::NotServed +} + +/// Answers a tune request from the guide index of the Xtream view. +/// +/// Same walk as the M3U counterpart. The lineup publishes live and video, so an entry of a cluster +/// this user may not watch is skipped while the remaining feeds are still tried, and the virtual id +/// is looked up in the live db before the video db, like the scan does. +async fn resolve_xtream_channel_via_guide_index( + app_state: &AppState, + credentials: &Arc, + target: &ConfigTarget, + query: &ChannelQuery<'_>, +) -> GuideIndexAnswer { + let cfg = &app_state.app_config; + let Some(server_info) = cfg.get_user_server_info(credentials) else { + return GuideIndexAnswer::Unavailable; + }; + let Some(candidates) = guide_index_candidates(app_state, target, LineupGuideView::Xtream, query).await else { + return GuideIndexAnswer::Unavailable; + }; + if candidates.is_empty() && query.virtual_id.is_none() { + return GuideIndexAnswer::NotServed; + } + let lookup = XtreamUserItemLookup::new(cfg, &app_state.playlists, target, credentials, &LINEUP_CLUSTERS).await; + let (base_url_live, base_url_vod) = xtream_base_urls(credentials, target, server_info.get_base_url()); + let stream_url = |cluster: XtreamCluster, item: &XtreamPlaylistItem| { + let base_url = if cluster == XtreamCluster::Live { &base_url_live } else { &base_url_vod }; + GuideIndexAnswer::Url(get_xtream_item_stream_url(cfg, cluster, credentials, base_url.as_deref(), item)) + }; + + for LineupGuideCandidate { key, entries } in candidates { + for entry in entries { + if let Some(item) = lookup.get(entry.virtual_id, entry.cluster).await { + if item.guide_number() == key.as_ref() { + return stream_url(entry.cluster, &item); + } + } + } + } + if let Some(stream_id) = query.virtual_id { + for cluster in LINEUP_CLUSTERS { + if let Some(item) = lookup.get(stream_id, cluster).await { + return stream_url(cluster, &item); + } + } + } + GuideIndexAnswer::NotServed +} + +/// Rewritten stream url of the channel a tune request addresses. +/// +/// The target guide index answers first, so a request costs a few keyed lookups instead of a full +/// playlist read, and a request nothing published matches ends as "not served" right there. The +/// scan only runs when the target has no usable index. +async fn resolve_channel_stream_url( + app_state: &AppState, + credentials: &Arc, + target: &ConfigTarget, + query: &ChannelQuery<'_>, +) -> Option> { + let cfg = &app_state.app_config; + match hdhomerun_lineup_view(target)? { + LineupGuideView::M3u => { + match resolve_m3u_channel_via_guide_index(app_state, credentials, target, query).await { + GuideIndexAnswer::Url(url) => return Some(url), + GuideIndexAnswer::NotServed => return None, + GuideIndexAnswer::Unavailable => {} + } + let channels = M3uPlaylistIterator::new(cfg, target, credentials).await.ok()?; + find_channel(channels, query).await.map(|(_, item)| get_m3u_item_stream_url(&item)) + } + LineupGuideView::Xtream => { + match resolve_xtream_channel_via_guide_index(app_state, credentials, target, query).await { + GuideIndexAnswer::Url(url) => return Some(url), + GuideIndexAnswer::NotServed => return None, + GuideIndexAnswer::Unavailable => {} + } + let server_info = cfg.get_user_server_info(credentials)?; + let (base_url_live, base_url_vod) = xtream_base_urls(credentials, target, server_info.get_base_url()); + let mut best: Option<(u8, Arc)> = None; + for (cluster, base_url) in [(XtreamCluster::Live, base_url_live), (XtreamCluster::Video, base_url_vod)] { + if !credentials.allows_cluster(cluster) { + continue; + } + let Ok(channels) = XtreamPlaylistIterator::new(cluster, cfg, target, None, credentials).await else { + continue; + }; + let Some((rank, item)) = find_channel(channels, query).await else { continue }; + if best.as_ref().is_none_or(|(best_rank, _)| rank < *best_rank) { + let url = get_xtream_item_stream_url(cfg, cluster, credentials, base_url.as_deref(), &item); + if rank == 0 { + return Some(url); + } + best = Some((rank, url)); + } + } + best.map(|(_, url)| url) + } + } +} + +/// Redirects to the stream url published in `lineup.json`. +/// +/// Connection limits are enforced by the stream endpoint (`max_connections`, grace period). A +/// pre-check here would reject channel switches while the previous stream is still counted. +async fn tune_channel( + app_state: &Arc, + fingerprint: &Fingerprint, + basic_auth: Option<&(String, String)>, + tuner_num: Option, + channel: &str, +) -> axum::response::Response { + let (credentials, target) = match authorize_device_user(app_state, fingerprint, basic_auth) { + Ok(user) => user, + Err(response) => return *response, + }; + + if let Some(tuner) = tuner_num.filter(|tuner| *tuner >= app_state.device.tuner_count) { + warn!( + "Requested tuner {tuner} exceeds tuner count {} for device '{}'", + app_state.device.tuner_count, app_state.device.name + ); + return axum::http::StatusCode::NOT_FOUND.into_response(); + } + + let Some(query) = ChannelQuery::parse(channel) else { + return axum::http::StatusCode::NOT_FOUND.into_response(); + }; + + let Some(stream_url) = resolve_channel_stream_url(&app_state.app_state, &credentials, &target, &query).await else { + warn!("Channel '{channel}' not found for HDHomeRun device '{}'", app_state.device.name); + return axum::http::StatusCode::NOT_FOUND.into_response(); + }; + axum::response::Redirect::temporary(&stream_url).into_response() } async fn auto_channel( - axum::extract::State(_app_state): axum::extract::State>, + fingerprint: Fingerprint, + axum::extract::State(app_state): axum::extract::State>, axum::extract::Path(channel): axum::extract::Path, ) -> impl IntoResponse { - warn!("HdHomerun api not implemented for auto_channel {channel}"); - axum::http::StatusCode::NOT_FOUND.into_response() + tune_channel(&app_state, &fingerprint, None, None, &channel).await +} + +async fn tuner_channel( + fingerprint: Fingerprint, + axum::extract::State(app_state): axum::extract::State>, + axum::extract::Path((tuner_num, channel)): axum::extract::Path<(u8, String)>, +) -> impl IntoResponse { + tune_channel(&app_state, &fingerprint, None, Some(tuner_num), &channel).await +} + +async fn auth_auto_channel( + fingerprint: Fingerprint, + AuthBasic(basic_auth): AuthBasic, + axum::extract::State(app_state): axum::extract::State>, + axum::extract::Path(channel): axum::extract::Path, +) -> impl IntoResponse { + tune_channel(&app_state, &fingerprint, Some(&basic_auth), None, &channel).await +} + +async fn auth_tuner_channel( + fingerprint: Fingerprint, + AuthBasic(basic_auth): AuthBasic, + axum::extract::State(app_state): axum::extract::State>, + axum::extract::Path((tuner_num, channel)): axum::extract::Path<(u8, String)>, +) -> impl IntoResponse { + tune_channel(&app_state, &fingerprint, Some(&basic_auth), Some(tuner_num), &channel).await } pub fn hdhr_api_register(basic_auth: bool) -> axum::Router> { @@ -519,6 +890,709 @@ pub fn hdhr_api_register(basic_auth: bool) -> axum::Router Arc { + let app_state = create_test_app_state(Config::default()); + let dto = HdHomeRunDeviceConfigDto { name: "test_hdhr".to_string(), tuner_count, ..Default::default() }; + let mut device: HdHomeRunDeviceConfig = (&dto).into(); + device.t_username = username.to_string(); + device.t_enabled = true; + Arc::new(HdHomerunAppState { app_state, device: Arc::new(device), hd_scan_state: Arc::new(AtomicI8::new(-1)) }) + } + + fn make_request(uri: &str) -> Request { + let mut req = Request::builder().method("GET").uri(uri).body(Body::empty()).unwrap(); + req.extensions_mut().insert(axum::extract::ConnectInfo(std::net::SocketAddr::from(([127, 0, 0, 1], 1234)))); + req + } + + fn m3u_item(virtual_id: u32, name: &str, epg_channel_id: Option<&str>) -> PlaylistItem { + m3u_item_in_group(virtual_id, name, epg_channel_id, "News") + } + + fn m3u_item_in_group(virtual_id: u32, name: &str, epg_channel_id: Option<&str>, group: &str) -> PlaylistItem { + PlaylistItem { + header: PlaylistItemHeader { + id: virtual_id.to_string().intern(), + // Persistence derives the virtual id of an item from its uuid, so the fixtures need + // one per channel to keep them apart. + uuid: UUIDType::from_valid_uuid(&format!("hdhr-test-{virtual_id}")), + virtual_id: VirtualId::new(virtual_id), + name: name.intern(), + title: name.intern(), + group: group.intern(), + epg_channel_id: epg_channel_id.map(Internable::intern), + url: format!("{PROVIDER_HOST}/live/puser/ppass/{virtual_id}.ts").intern(), + input_name: "input".intern(), + input_stream_id: virtual_id.to_string().intern(), + item_type: PlaylistItemType::Live, + xtream_cluster: XtreamCluster::Live, + ..PlaylistItemHeader::default() + }, + } + } + + /// Device whose target is persisted on disk and served to a reverse proxy user. + /// + /// With a `lineup` source the target also publishes an `HDHomeRun` output and the playlist goes + /// through the persistence entry point that writes the guide index of the lineup view. Without + /// it the M3U playlist is written directly, the way a target that never publishes a lineup is + /// persisted and leaves no index behind. + async fn hdhomerun_device( + items: Vec, + lineup: Option, + ) -> (tempfile::TempDir, Arc, Arc) { + let temp = tempfile::tempdir().unwrap(); + let state = test_hdhr_app_state("hdhr", 2); + let app_config = &state.app_state.app_config; + + let mut config = (*app_config.config.load_full()).clone(); + config.storage_dir = temp.path().to_string_lossy().into_owned(); + config.user_config_dir = Some(temp.path().to_string_lossy().into_owned()); + app_config.config.store(Arc::new(config)); + + let mut output = vec![match lineup { + Some(LineupSource::Xtream) => TargetOutputDto::Xtream(XtreamTargetOutputDto::default()), + _ => TargetOutputDto::M3u(M3uTargetOutputDto::default()), + }]; + if let Some(source) = lineup { + let use_output = match source { + LineupSource::M3u => TargetType::M3u, + LineupSource::Xtream => TargetType::Xtream, + }; + output.push(TargetOutputDto::HdHomeRun(HdHomeRunTargetOutputDto { + use_output: Some(use_output), + ..Default::default() + })); + } + let target = + ConfigTarget::from(&ConfigTargetDto { id: 1, name: "default".to_string(), output, ..Default::default() }); + let input = Arc::new(ConfigInput { id: 1, name: "input".intern(), enabled: true, ..ConfigInput::default() }); + app_config.sources.store(Arc::new(SourcesConfig { + inputs: vec![Arc::clone(&input)], + sources: vec![ConfigSource { + inputs: vec![Arc::clone(&input.name)], + targets: vec![Arc::new(target.clone())], + }], + ..SourcesConfig::default() + })); + + let mut user = ProxyUserCredentials::default(); + user.username = "hdhr".to_string(); + user.password = "secret".to_string(); + user.proxy = ProxyType::Reverse(None); + user.output_clusters = ClusterFlags::all(); + app_config.api_proxy.store(Some(Arc::new(ApiProxyConfig { + server: vec![ApiProxyServerInfo { + name: "default".to_string(), + protocol: "http".to_string(), + host: "tuliprox.local".to_string(), + port: Some("8901".to_string()), + timezone: "UTC".to_string(), + message: String::new(), + path: None, + }], + user: vec![TargetUser { target: target.name.clone(), credentials: vec![Arc::new(user)] }], + ..Default::default() + }))); + + let mut groups = + vec![PlaylistGroup { id: 1, title: "News".intern(), channels: items, xtream_cluster: XtreamCluster::Live }]; + if lineup.is_some() { + persist_playlist(app_config, &mut groups, None, &target, None, TargetPlaylistPersistOptions::default()) + .await + .expect("target playlist persisted"); + } else { + let target_path = ensure_target_storage_path(&app_config.config.load(), &target.name).await.unwrap(); + m3u_write_playlist(app_config, &target, target.get_m3u_output().unwrap(), &target_path, &groups, false) + .await + .unwrap(); + } + (temp, state, Arc::new(target)) + } + + /// Device with a target that publishes no `HDHomeRun` output. + async fn m3u_device(items: Vec) -> (tempfile::TempDir, Arc) { + let (temp, state, _target) = hdhomerun_device(items, None).await; + (temp, state) + } + + /// Guide index file of the device target. + async fn guide_index_path(state: &Arc, target: &ConfigTarget) -> std::path::PathBuf { + let config = state.app_state.app_config.config.load(); + let target_path = ensure_target_storage_path(&config, target.name.as_str()).await.unwrap(); + m3u_lineup_guide_index_path(&target_path) + } + + /// Guide index file of the device target of the Xtream view. + fn xtream_guide_index_path(state: &Arc, target: &ConfigTarget) -> std::path::PathBuf { + xtream_lineup_guide_index_path(&state.app_state.app_config.config.load(), &target.name) + .expect("xtream storage path") + } + + /// The lineup the device publishes, parsed the way a client reads it. + async fn lineup_entries(state: &Arc) -> Vec { + let response = + hdhr_api_register(false).with_state(Arc::clone(state)).oneshot(make_request("/lineup.json")).await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let body = axum::body::to_bytes(response.into_body(), usize::MAX).await.unwrap(); + serde_json::from_slice(&body).expect("lineup parses") + } + + async fn location(state: &Arc, uri: &str) -> Result { + let resp = hdhr_api_register(false).with_state(Arc::clone(state)).oneshot(make_request(uri)).await.unwrap(); + if resp.status() != StatusCode::TEMPORARY_REDIRECT { + return Err(resp.status()); + } + Ok(resp.headers()[header::LOCATION].to_str().unwrap().to_string()) + } + + #[test] + fn channel_query_strips_numeric_virtual_channel_prefix_only() { + let q = |channel| ChannelQuery::parse(channel).map(|q| (q.raw, q.numeric, q.virtual_id)); + assert_eq!(q("v101"), Some(("v101", Some("101"), Some(101)))); + assert_eq!(q("V101.ts"), Some(("V101", Some("101"), Some(101)))); + assert_eq!(q("101.TS"), Some(("101", None, Some(101)))); + assert_eq!(q(" v2.1 "), Some(("v2.1", Some("2.1"), None))); + assert_eq!(q("vox.de"), Some(("vox.de", None, None))); + assert_eq!(q("VIVA"), Some(("VIVA", None, None))); + assert_eq!(q("v"), Some(("v", None, None))); + assert_eq!(q(".ts"), None); + assert_eq!(q(" "), None); + assert_eq!(q("ä.ts"), Some(("ä", None, None))); + } + + #[tokio::test] + async fn find_channel_prefers_guide_number_over_virtual_id() { + let items: Vec = [m3u_item(101, "Other", None), m3u_item(7, "Numeric", Some("101"))] + .iter() + .map(M3uPlaylistItem::from) + .collect(); + let find = |channel: &'static str| { + let items = items.clone(); + async move { + let query = ChannelQuery::parse(channel).unwrap(); + let channels = stream::iter(items.into_iter().map(|item| Ok((item, false)))); + find_channel(channels, &query).await.map(|(rank, item)| (rank, item.virtual_id.get())) + } + }; + assert_eq!(find("101").await, Some((0, 7))); + assert_eq!(find("v101").await, Some((1, 7))); + assert_eq!(find("Other").await, Some((0, 101))); + assert_eq!(find("v7").await, Some((2, 7))); + assert_eq!(find("v999").await, None); + } + + #[tokio::test] + async fn auto_channel_redirects_to_proxied_stream_url() { + let (_temp, state) = m3u_device(vec![ + m3u_item(1, "VOX", Some("vox.de")), + m3u_item(2, "Das Erste", Some("101")), + m3u_item(101, "Other", None), + ]) + .await; + + // virtual id, guide number starting with `v`, explicit tuner with `.ts` suffix + for uri in ["/auto/v1", "/auto/vox.de", "/tuner1/vox.de.ts"] { + assert_eq!( + location(&state, uri).await.as_deref(), + Ok("http://tuliprox.local:8901/m3u-stream/hdhr/secret/1.ts"), + "{uri}" + ); + } + // numeric guide number wins over the virtual id 101 lookup + assert_eq!( + location(&state, "/auto/v101").await.as_deref(), + Ok("http://tuliprox.local:8901/m3u-stream/hdhr/secret/2.ts") + ); + } + + #[tokio::test] + async fn tune_rejects_unknown_channel_and_tuner() { + let (_temp, state) = m3u_device(vec![m3u_item(1, "VOX", Some("vox.de"))]).await; + assert_eq!(location(&state, "/auto/v404").await, Err(StatusCode::NOT_FOUND)); + assert_eq!(location(&state, "/auto/ox.de").await, Err(StatusCode::NOT_FOUND)); + assert_eq!(location(&state, "/tuner2/vox.de").await, Err(StatusCode::NOT_FOUND)); + assert_eq!(location(&state, "/tunerX/vox.de").await, Err(StatusCode::BAD_REQUEST)); + } + + #[tokio::test] + async fn auto_channel_unknown_user_returns_not_found() { + let state = test_hdhr_app_state("unknown_user", 2); + assert_eq!(location(&state, "/auto/v1").await, Err(StatusCode::NOT_FOUND)); + assert_eq!(location(&state, "/tuner0/v1").await, Err(StatusCode::NOT_FOUND)); + } + + #[tokio::test] + async fn auth_channel_routes_require_basic_auth() { + let state = test_hdhr_app_state("test_user", 2); + for uri in ["/auto/v1", "/tuner0/v1"] { + let resp = hdhr_api_register(true).with_state(Arc::clone(&state)).oneshot(make_request(uri)).await.unwrap(); + assert_eq!(resp.status(), StatusCode::UNAUTHORIZED, "{uri}"); + } + } + + #[tokio::test] + async fn guide_index_follows_the_lineup_output() { + let (_temp, state, target) = + hdhomerun_device(vec![m3u_item(1, "VOX", Some("vox.de"))], Some(LineupSource::M3u)).await; + let index_path = guide_index_path(&state, &target).await; + assert!(index_path.exists(), "a target with an HDHomeRun output gets a guide index"); + assert_eq!( + location(&state, "/auto/vox.de").await.as_deref(), + Ok("http://tuliprox.local:8901/m3u-stream/hdhr/secret/1.ts") + ); + + // The same playlist persisted without the lineup output leaves no index behind. + let without_lineup = ConfigTarget::from(&ConfigTargetDto { + id: 1, + name: "default".to_string(), + output: vec![TargetOutputDto::M3u(M3uTargetOutputDto::default())], + ..Default::default() + }); + let mut groups = vec![PlaylistGroup { + id: 1, + title: "News".intern(), + channels: vec![m3u_item(1, "VOX", Some("vox.de"))], + xtream_cluster: XtreamCluster::Live, + }]; + persist_playlist( + &state.app_state.app_config, + &mut groups, + None, + &without_lineup, + None, + TargetPlaylistPersistOptions::default(), + ) + .await + .expect("playlist persisted"); + + assert!(!index_path.exists(), "no index for a target without an HDHomeRun output"); + } + + #[tokio::test] + async fn guide_index_of_a_view_that_is_gone_is_cleaned_up() { + // The lineup output stays, only the view behind it changes from Xtream to M3U. The index of + // the vanished view would describe a playlist this target no longer serves. + let (_temp, state, target) = + hdhomerun_device(vec![m3u_item(1, "VOX", Some("vox.de"))], Some(LineupSource::Xtream)).await; + let xtream_index = xtream_guide_index_path(&state, &target); + assert!(xtream_index.exists(), "the Xtream view gets its index"); + + let m3u_target = ConfigTarget::from(&ConfigTargetDto { + id: 1, + name: "default".to_string(), + output: vec![ + TargetOutputDto::M3u(M3uTargetOutputDto::default()), + TargetOutputDto::HdHomeRun(HdHomeRunTargetOutputDto { + use_output: Some(TargetType::M3u), + ..Default::default() + }), + ], + ..Default::default() + }); + let mut groups = vec![PlaylistGroup { + id: 1, + title: "News".intern(), + channels: vec![m3u_item(1, "VOX", Some("vox.de"))], + xtream_cluster: XtreamCluster::Live, + }]; + persist_playlist( + &state.app_state.app_config, + &mut groups, + None, + &m3u_target, + None, + TargetPlaylistPersistOptions::default(), + ) + .await + .expect("playlist persisted"); + + assert!(!xtream_index.exists(), "the index of the vanished view is removed"); + assert!(guide_index_path(&state, &m3u_target).await.exists(), "the M3U view gets its index"); + } + + #[tokio::test] + async fn channel_resolution_uses_the_guide_index_before_the_scan() { + // Two channels carry the same guide number. The index of the persisted playlist points at + // the first one, so replacing it with an index of the second one answers only if the + // request is served from the index: the scan would still return the first channel. + let (_temp, state, target) = hdhomerun_device( + vec![m3u_item(1, "VOX", Some("vox.de")), m3u_item(2, "VOX HD", Some("vox.de"))], + Some(LineupSource::M3u), + ) + .await; + let index_path = guide_index_path(&state, &target).await; + assert_eq!( + location(&state, "/auto/vox.de").await.as_deref(), + Ok("http://tuliprox.local:8901/m3u-stream/hdhr/secret/1.ts") + ); + + let second_channel = vec![PlaylistGroup { + id: 1, + title: "News".intern(), + channels: vec![m3u_item(2, "VOX HD", Some("vox.de"))], + xtream_cluster: XtreamCluster::Live, + }]; + lineup_guide_index_write( + &state.app_state.app_config.file_locks, + &second_channel, + LineupGuideView::M3u, + &index_path, + None, + ) + .await + .expect("index rewritten"); + + assert_eq!( + location(&state, "/auto/vox.de").await.as_deref(), + Ok("http://tuliprox.local:8901/m3u-stream/hdhr/secret/2.ts") + ); + } + + #[tokio::test] + async fn unknown_guide_number_is_answered_without_scanning_the_playlist() { + // The index is authoritative for the guide numbers it was written with: what it does not + // know is not served. Replacing it with an index of another playlist makes that observable, + // because the playlist still holds a channel carrying the requested guide number, which the + // scan would have found. + let (_temp, state, target) = + hdhomerun_device(vec![m3u_item(1, "VOX", Some("vox.de"))], Some(LineupSource::M3u)).await; + let elsewhere = vec![PlaylistGroup { + id: 1, + title: "News".intern(), + channels: vec![m3u_item(9, "Other", Some("other.de"))], + xtream_cluster: XtreamCluster::Live, + }]; + lineup_guide_index_write( + &state.app_state.app_config.file_locks, + &elsewhere, + LineupGuideView::M3u, + &guide_index_path(&state, &target).await, + None, + ) + .await + .expect("index rewritten"); + + assert_eq!(location(&state, "/auto/vox.de").await, Err(StatusCode::NOT_FOUND)); + // A virtual id is still resolved through the index, which knows no guide number for it. + assert_eq!( + location(&state, "/auto/v1").await.as_deref(), + Ok("http://tuliprox.local:8901/m3u-stream/hdhr/secret/1.ts") + ); + } + + #[tokio::test] + async fn channel_resolution_falls_back_to_the_scan_without_an_index() { + let (_temp, state, target) = hdhomerun_device( + vec![m3u_item(1, "VOX", Some("vox.de")), m3u_item(2, "VOX HD", Some("vox.de"))], + Some(LineupSource::M3u), + ) + .await; + std::fs::remove_file(guide_index_path(&state, &target).await).expect("index removed"); + + // Guide number, its numeric form and the bare virtual id all still resolve. + for uri in ["/auto/vox.de", "/auto/v2", "/tuner1/vox.de.ts"] { + let expected = if uri == "/auto/v2" { + "http://tuliprox.local:8901/m3u-stream/hdhr/secret/2.ts" + } else { + "http://tuliprox.local:8901/m3u-stream/hdhr/secret/1.ts" + }; + assert_eq!(location(&state, uri).await.as_deref(), Ok(expected), "{uri}"); + } + } + + #[tokio::test] + async fn index_serves_the_next_feed_when_the_first_is_outside_the_bouquet() { + // Both feeds of the guide number are in the index, in lineup order. The first one is outside + // this user's bouquet, so the request must not leak it and has to end up on the second one. + let (_temp, state, _target) = hdhomerun_device( + vec![ + m3u_item_in_group(1, "Hidden HD", Some("vox.de"), "Hidden"), + m3u_item_in_group(2, "VOX", Some("vox.de"), "News"), + ], + Some(LineupSource::M3u), + ) + .await; + let bouquet = PlaylistBouquetDto { + xtream: None, + m3u: Some(PlaylistClusterBouquetDto { live: Some(vec!["News".to_string()]), ..Default::default() }), + }; + save_user_bouquet(&state.app_state.app_config, "default", "hdhr", &bouquet).await.expect("bouquet saved"); + + assert_eq!( + location(&state, "/auto/vox.de").await.as_deref(), + Ok("http://tuliprox.local:8901/m3u-stream/hdhr/secret/2.ts") + ); + } + + #[tokio::test] + async fn guide_number_with_no_visible_feed_is_answered_without_scanning_the_playlist() { + // Only the feed the bouquet excludes is in the index, so every candidate is hidden for this + // user. The request ends as 404 instead of walking the playlist, which still holds a second + // feed of that guide number and would have answered it. + let (_temp, state, target) = hdhomerun_device( + vec![ + m3u_item_in_group(1, "Hidden HD", Some("vox.de"), "Hidden"), + m3u_item_in_group(2, "VOX", Some("vox.de"), "News"), + ], + Some(LineupSource::M3u), + ) + .await; + let hidden_only = vec![PlaylistGroup { + id: 1, + title: "News".intern(), + channels: vec![m3u_item_in_group(1, "Hidden HD", Some("vox.de"), "Hidden")], + xtream_cluster: XtreamCluster::Live, + }]; + lineup_guide_index_write( + &state.app_state.app_config.file_locks, + &hidden_only, + LineupGuideView::M3u, + &guide_index_path(&state, &target).await, + None, + ) + .await + .expect("index rewritten"); + let bouquet = PlaylistBouquetDto { + xtream: None, + m3u: Some(PlaylistClusterBouquetDto { live: Some(vec!["News".to_string()]), ..Default::default() }), + }; + save_user_bouquet(&state.app_state.app_config, "default", "hdhr", &bouquet).await.expect("bouquet saved"); + + assert_eq!(location(&state, "/auto/vox.de").await, Err(StatusCode::NOT_FOUND)); + } + + #[tokio::test] + async fn xtream_tuner_redirects_to_the_published_lineup_url() { + let (_temp, state, target) = hdhomerun_device( + vec![m3u_item(1, "VOX", Some("vox.de")), m3u_item(2, "Das Erste", Some("101"))], + Some(LineupSource::Xtream), + ) + .await; + assert!(xtream_guide_index_path(&state, &target).exists(), "the Xtream view gets its own index"); + + let published = lineup_entries(&state).await; + let expected = published + .iter() + .find(|entry| entry.guide_number.as_ref() == "vox.de") + .expect("the lineup publishes the channel"); + + // The tuner has to redirect to exactly the url the lineup published, for the guide number + // as well as for the virtual id a client may build itself. + for uri in ["/auto/vox.de", "/auto/v1"] { + assert_eq!(location(&state, uri).await.as_deref(), Ok(expected.url.as_ref()), "{uri}"); + } + } + + fn vod_item(virtual_id: u32, name: &str, epg_channel_id: Option<&str>) -> PlaylistItem { + let mut item = m3u_item(virtual_id, name, epg_channel_id); + item.header.item_type = PlaylistItemType::Video; + item.header.xtream_cluster = XtreamCluster::Video; + item + } + + fn live_and_vod_groups(live: Vec, vod: Vec) -> Vec { + let mut groups = + vec![PlaylistGroup { id: 1, title: "News".intern(), channels: live, xtream_cluster: XtreamCluster::Live }]; + if !vod.is_empty() { + groups.push(PlaylistGroup { + id: 2, + title: "Movies".intern(), + channels: vod, + xtream_cluster: XtreamCluster::Video, + }); + } + groups + } + + #[tokio::test] + async fn update_that_keeps_a_cluster_db_carries_its_entries_into_the_index() { + // A provider answering without movies leaves the published video db in place. The index + // rebuilt from that update learns the live channels from it and carries the movies over + // from the previous index, which was written together with the kept video db. + let (_temp, state, target) = + hdhomerun_device(vec![m3u_item(1, "VOX", Some("vox.de"))], Some(LineupSource::Xtream)).await; + let app_config = &state.app_state.app_config; + let index_path = xtream_guide_index_path(&state, &target); + let persist = |groups: Vec| { + let target = Arc::clone(&target); + async move { + let mut groups = groups; + persist_playlist(app_config, &mut groups, None, &target, None, TargetPlaylistPersistOptions::default()) + .await + .expect("playlist persisted"); + } + }; + let indexed_clusters = |guide: &'static str| { + let index_path = index_path.clone(); + async move { + match lineup_guide_candidates(&app_config.file_locks, index_path, vec![guide.intern()]).await { + LineupGuideLookup::Candidates(candidates) => candidates + .into_iter() + .flat_map(|candidate| candidate.entries) + .map(|entry| entry.cluster) + .collect::>(), + LineupGuideLookup::NoEntry | LineupGuideLookup::Missing => Vec::new(), + } + } + }; + let live_only = || { + live_and_vod_groups( + vec![m3u_item(1, "VOX", Some("vox.de")), m3u_item(3, "Sport", Some("sport.de"))], + Vec::new(), + ) + }; + + persist(live_and_vod_groups( + vec![m3u_item(1, "VOX", Some("vox.de"))], + vec![vod_item(2, "Film", Some("film.de"))], + )) + .await; + let film_url = lineup_entries(&state) + .await + .into_iter() + .find(|entry| entry.guide_number.as_ref() == "film.de") + .expect("the lineup publishes the movie") + .url; + + persist(live_only()).await; + assert_eq!(indexed_clusters("film.de").await, vec![XtreamCluster::Video], "the movie is carried over"); + assert_eq!(indexed_clusters("sport.de").await, vec![XtreamCluster::Live], "the new live channel is indexed"); + assert_eq!(location(&state, "/auto/film.de").await.as_deref(), Ok(film_url.as_ref())); + assert!(location(&state, "/auto/sport.de").await.is_ok()); + + // Without a previous index nothing describes the kept video db, so no index is written and + // the scan answers instead. + std::fs::remove_file(&index_path).expect("index removed"); + persist(live_only()).await; + assert!(!index_path.exists(), "an index that cannot describe the kept video db is not written"); + assert_eq!(location(&state, "/auto/film.de").await.as_deref(), Ok(film_url.as_ref())); + } + + #[tokio::test] + async fn virtual_id_is_tried_when_every_guide_number_feed_is_hidden() { + // `v2` names the guide number `2` of a feed outside the bouquet and the virtual id of a + // published channel. The scan ranks the virtual id after the guide number, so the index + // has to fall through to it instead of answering 404. + let (_temp, state, _target) = hdhomerun_device( + vec![ + m3u_item_in_group(1, "Hidden", Some("2"), "Hidden"), + m3u_item_in_group(2, "VOX", Some("vox.de"), "News"), + ], + Some(LineupSource::M3u), + ) + .await; + let bouquet = PlaylistBouquetDto { + xtream: None, + m3u: Some(PlaylistClusterBouquetDto { live: Some(vec!["News".to_string()]), ..Default::default() }), + }; + save_user_bouquet(&state.app_state.app_config, "default", "hdhr", &bouquet).await.expect("bouquet saved"); + + assert_eq!( + location(&state, "/auto/v2").await.as_deref(), + Ok("http://tuliprox.local:8901/m3u-stream/hdhr/secret/2.ts") + ); + } + + #[tokio::test] + async fn guide_index_failure_does_not_fail_the_playlist_publication() { + // The guide index is a sidecar. A directory where the index file belongs makes removing and + // writing it fail; the playlist must still be published and the tuners scan it instead. + let (_temp, state, target) = + hdhomerun_device(vec![m3u_item(1, "VOX", Some("vox.de"))], Some(LineupSource::M3u)).await; + let index_path = guide_index_path(&state, &target).await; + std::fs::remove_file(&index_path).expect("index removed"); + std::fs::create_dir(&index_path).expect("blocking directory created"); + + let mut groups = vec![PlaylistGroup { + id: 1, + title: "News".intern(), + channels: vec![m3u_item(1, "VOX", Some("vox.de")), m3u_item(2, "Sport", Some("sport.de"))], + xtream_cluster: XtreamCluster::Live, + }]; + persist_playlist( + &state.app_state.app_config, + &mut groups, + None, + &target, + None, + TargetPlaylistPersistOptions::default(), + ) + .await + .expect("playlist published despite the index failure"); + + assert!(index_path.is_dir(), "the index could not be written"); + assert_eq!( + location(&state, "/auto/sport.de").await.as_deref(), + Ok("http://tuliprox.local:8901/m3u-stream/hdhr/secret/2.ts") + ); + } + + #[tokio::test] + async fn percent_encoded_guide_number_is_decoded_before_matching() { + // A channel without EPG id publishes its name as guide number, spaces and all. Clients send + // it percent-encoded; the index and the scan both have to see the decoded name. + let items = || vec![m3u_item(1, "Das Erste HD", None), m3u_item(2, "AC/DC TV", None)]; + let (_temp, indexed, _target) = hdhomerun_device(items(), Some(LineupSource::M3u)).await; + let (_temp_scan, scanned) = m3u_device(items()).await; + + for state in [&indexed, &scanned] { + assert_eq!( + location(state, "/auto/Das%20Erste%20HD").await.as_deref(), + Ok("http://tuliprox.local:8901/m3u-stream/hdhr/secret/1.ts") + ); + assert_eq!( + location(state, "/auto/AC%2FDC%20TV").await.as_deref(), + Ok("http://tuliprox.local:8901/m3u-stream/hdhr/secret/2.ts") + ); + } + } } diff --git a/backend/repository/src/lib.rs b/backend/repository/src/lib.rs index 9a330be1e..42d161eae 100644 --- a/backend/repository/src/lib.rs +++ b/backend/repository/src/lib.rs @@ -11,6 +11,7 @@ mod epg_group_index; mod epg_repository; mod geoip; mod library_repository; +mod lineup_guide_index; mod live_stream_metadata_repository; mod m3u_playlist_iterator; mod m3u_repository; @@ -47,6 +48,7 @@ pub use epg_group_index::*; pub use epg_repository::*; pub use geoip::*; pub use library_repository::*; +pub use lineup_guide_index::*; pub use live_stream_metadata_repository::{ load_input_live_bitrate_bps, persist_input_live_bitrate_bps, LiveBitratePersistenceOutcome, }; diff --git a/backend/repository/src/lineup_guide_index.rs b/backend/repository/src/lineup_guide_index.rs new file mode 100644 index 000000000..539f1634d --- /dev/null +++ b/backend/repository/src/lineup_guide_index.rs @@ -0,0 +1,623 @@ +//! `GuideNumber` index of the `HDHomeRun` lineup of a target. +//! +//! `lineup.json` publishes one guide number per channel — the EPG channel id, else the channel +//! name — and the tuner endpoints receive that string back as `/auto/{channel}`. Answering it by +//! walking the playlist iterator reads the complete target playlist per tune (measured: 0.4 to +//! 0.8 s for 50 000 channels), while the keyed lookup by virtual id this index feeds takes about +//! a millisecond. +//! +//! The index is a sidecar of one target playlist view, written by the same persistence step that +//! writes the playlist it describes, and only for targets that publish an `HDHomeRun` output. A +//! missing index is therefore not an error: readers fall back to the scan, which stays the +//! reference implementation. +//! +//! Every feed of a guide number is kept, not only the one the lineup publishes first. The tuner +//! serves the first feed that this user's output publishes, and channel failover will pick among +//! the same list by quality of service instead. + +use crate::{m3u_get_file_path_for_db, storage_const, xtream_get_file_path, xtream_get_storage_path}; +use serde::{Deserialize, Serialize}; +use shared::{ + concat_string, + error::TuliproxError, + model::{PlaylistGroup, PlaylistItemType, TargetType, XtreamCluster}, +}; +use std::{ + io, + path::{Path, PathBuf}, + sync::Arc, +}; +use tuliprox_btree::{BPlusTree, BPlusTreeQuery}; +use tuliprox_core::{ + model::{Config, ConfigTarget}, + utils::FileLockManager, +}; + +/// Target playlist view a lineup is published from. +/// +/// `lineup.json` serves either the M3U target playlist or the Xtream target playlists, selected +/// by the `use_output` of the `HDHomeRun` output, so each view needs its own index. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum LineupGuideView { + /// The flat M3U target playlist: one key space holding every item type. + M3u, + /// The Xtream target playlists: live and video, each in its own db. + Xtream, +} + +/// The item a guide number resolves to. +/// +/// A guide number usually names one channel, but providers publish one EPG id for several feeds of +/// the same channel (SD, HD, FHD, backup), which is also what channel failover groups. The index +/// therefore keeps every feed of a guide number, ordered the way the lineup scan meets them. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct LineupGuideEntry { + /// Cluster holding the virtual id, needed to fetch from the right Xtream db. + pub cluster: XtreamCluster, + pub virtual_id: u32, +} + +/// Guide number to every [`LineupGuideEntry`] carrying it. +pub type LineupGuideTree = BPlusTree, Vec>; + +/// The candidates one requested key has, in the order the lineup meets them. +#[derive(Debug, Clone, PartialEq)] +pub struct LineupGuideCandidate { + /// The lookup key that matched, so the caller can verify the item still carries it. + pub key: Arc, + pub entries: Vec, +} + +/// Outcome of a guide number lookup. +#[derive(Debug, Clone, PartialEq)] +pub enum LineupGuideLookup { + /// No usable index: the target never published an `HDHomeRun` lineup, or a persistence step is + /// between removing the old index and publishing the new one. + Missing, + /// The index exists but knows none of the requested keys. + NoEntry, + /// Candidates of every requested key the index knows, in the order the keys were asked for, so + /// the caller can tell a guide number a channel carries from the stripped form of another. + Candidates(Vec), +} + +/// View `lineup.json` is published from, selected by the `use_output` of the `HDHomeRun` output. +/// +/// Without a `use_output` the M3U playlist wins over the Xtream playlists. `None` means the target +/// has no output a lineup can be published from. +/// +/// This deliberately does not require the target to carry the `HDHomeRun` output itself: the device +/// serves the target of its user, and `lineup.json` has always published that target's playlist +/// with the default selection when the output is not on it. Whether a guide index exists is the +/// separate question [`lineup_guide_index_view`] answers. +pub fn hdhomerun_lineup_view(target: &ConfigTarget) -> Option { + let use_output = target.get_hdhomerun_output().and_then(|output| output.use_output); + let has_m3u = target.has_output(TargetType::M3u); + let has_xtream = target.has_output(TargetType::Xtream); + match use_output { + None | Some(TargetType::M3u) if has_m3u => Some(LineupGuideView::M3u), + None | Some(TargetType::Xtream) if has_xtream => Some(LineupGuideView::Xtream), + _ => None, + } +} + +/// View whose guide index a target keeps: only a target publishing an `HDHomeRun` output has one. +pub fn lineup_guide_index_view(target: &ConfigTarget) -> Option { + if target.has_output(TargetType::HdHomeRun) { + hdhomerun_lineup_view(target) + } else { + None + } +} + +fn lineup_guide_file_name() -> String { concat_string!("lineup_guide.", storage_const::FILE_SUFFIX_DB) } + +/// Index file of the M3U view, next to the M3U playlist db of the target. +pub fn m3u_lineup_guide_index_path(target_path: &Path) -> PathBuf { + m3u_get_file_path_for_db(target_path).with_file_name(lineup_guide_file_name()) +} + +/// Index file of the Xtream view, next to the live playlist db of the target. +pub fn xtream_lineup_guide_index_path(config: &Config, target_name: &str) -> Option { + xtream_get_storage_path(config, target_name) + .map(|path| xtream_get_file_path(&path, XtreamCluster::Live).with_file_name(lineup_guide_file_name())) +} + +/// Rank of a cluster when the lineup of the Xtream view is published: the live db is searched +/// before the video db, so live wins a guide number two channels share. +fn cluster_rank(cluster: XtreamCluster) -> u8 { + match cluster { + XtreamCluster::Live => 0, + XtreamCluster::Video => 1, + XtreamCluster::Series => 2, + } +} + +/// Cluster rank, source ordinal and virtual id: the order the lineup meets its items in. +type LineupPosition = (u8, u32, u32); + +/// Position of an item in the lineup of `view`. +/// +/// The playlist dbs are read through their sorted index, which orders by `source_ordinal` (see +/// `m3u_write_playlist` and the Xtream publication). The M3U db is one key space, the Xtream view is +/// live first and then video. The virtual id breaks ties the way the db key does. +fn lineup_position( + view: LineupGuideView, + cluster: XtreamCluster, + source_ordinal: u32, + virtual_id: u32, +) -> LineupPosition { + let cluster_rank = match view { + LineupGuideView::M3u => 0, + LineupGuideView::Xtream => cluster_rank(cluster), + }; + (cluster_rank, source_ordinal, virtual_id) +} + +/// One index row: guide number, lineup position and the entry. +type LineupGuideRow = (Arc, LineupPosition, LineupGuideEntry); + +/// Entries of clusters a persistence step keeps, read from the index that already describes them. +/// +/// An Xtream update without items for one cluster keeps that cluster's db, so the new index cannot +/// learn its channels from the update. They are carried over from the previous index instead: it +/// was written together with the kept db, which has not changed since. +#[derive(Debug, Default)] +pub struct LineupGuideCarryOver { + rows: Vec, +} + +/// Rows of the playlist channels the index of `view` holds. +fn playlist_rows(playlist: &[PlaylistGroup], view: LineupGuideView) -> Vec { + playlist + .iter() + .flat_map(|group| &group.channels) + .filter_map(|item| { + let header = &item.header; + let cluster = header.xtream_cluster; + let indexed = match view { + LineupGuideView::M3u => { + !matches!(header.item_type, PlaylistItemType::SeriesInfo | PlaylistItemType::LocalSeriesInfo) + } + LineupGuideView::Xtream => matches!(cluster, XtreamCluster::Live | XtreamCluster::Video), + }; + if !indexed { + return None; + } + let guide = header.epg_channel_id.as_ref().unwrap_or(&header.name); + if guide.is_empty() { + return None; + } + let virtual_id = header.virtual_id.get(); + Some(( + Arc::clone(guide), + lineup_position(view, cluster, header.source_ordinal, virtual_id), + LineupGuideEntry { cluster, virtual_id }, + )) + }) + .collect() +} + +/// Groups rows by guide number into the tree, each group in lineup order. +/// +/// The rows are one flat list sorted by guide number and lineup position, so grouping needs no map +/// with one vector per guide number next to the tree being built. +fn tree_from_rows(mut rows: Vec) -> LineupGuideTree { + rows.sort_unstable_by(|left, right| (&left.0, left.1).cmp(&(&right.0, right.1))); + + let mut tree = LineupGuideTree::new(); + let mut rows = rows.into_iter().peekable(); + while let Some((guide, _, entry)) = rows.next() { + let mut entries = vec![entry]; + while let Some((_, _, entry)) = rows.next_if(|(next, _, _)| *next == guide) { + // The same item listed in several groups carries different source ordinals, so its + // rows need not be adjacent. A guide number has only a handful of feeds, which keeps + // this check cheap; the first occurrence keeps its lineup position. + if !entries.contains(&entry) { + entries.push(entry); + } + } + tree.insert(guide, entries); + } + tree +} + +/// Builds the index of one target playlist view. +/// +/// The Xtream view indexes live and video only, matching the channels `lineup.json` publishes for +/// it. The M3U view indexes what its db holds: every item except series info, which +/// `m3u_write_playlist` leaves out. +/// +/// The guide number is the one `lineup.json` publishes: the EPG channel id when it is set, +/// otherwise the name. An empty EPG channel id publishes the empty guide number, which no request +/// can produce, so such items are left out. +pub fn build_lineup_guide_index(playlist: &[PlaylistGroup], view: LineupGuideView) -> LineupGuideTree { + tree_from_rows(playlist_rows(playlist, view)) +} + +/// Reads the entries of `clusters` from the index at `path`, to carry them into the next index. +/// +/// Only the rows of the kept clusters are held; the rest of the index is streamed past. Each keeps +/// its position within its guide number, which is lineup order: the clusters rank live before +/// video, and the previous index already lists the feeds of one cluster in lineup order. `None` +/// when the index is missing or unreadable, so the caller cannot describe the kept clusters. +pub async fn lineup_guide_index_carry_over( + file_locks: &FileLockManager, + path: &Path, + clusters: &[XtreamCluster], +) -> Option { + let file_lock = file_locks.read_lock(path).await; + let path = path.to_path_buf(); + let clusters = clusters.to_vec(); + let task = tokio::task::spawn_blocking(move || -> io::Result { + let _guard = file_lock; + let mut query = BPlusTreeQuery::, Vec>::try_new(&path)?; + let mut rows = Vec::new(); + for entry in query.iter() { + let (guide, entries) = entry?; + for (ordinal, entry) in (0u32..).zip(entries) { + if clusters.contains(&entry.cluster) { + let position = (cluster_rank(entry.cluster), ordinal, entry.virtual_id); + rows.push((Arc::clone(&guide), position, entry)); + } + } + } + Ok(LineupGuideCarryOver { rows }) + }); + match task.await { + Ok(Ok(carry_over)) => Some(carry_over), + Ok(Err(err)) => { + log::debug!("No lineup guide index to carry over: {err}"); + None + } + Err(err) => { + log::warn!("Lineup guide index carry over task failed: {err}"); + None + } + } +} + +/// Builds the index of `view` and stores it at `path`, together with the entries `carry_over` +/// brings from the previous index. +/// +/// Only collecting the rows from the borrowed playlist runs on the calling task. Sorting, building +/// the tree, creating the directory and the store run on a blocking thread under the write lock of +/// the index file, so a reader never opens a file the store is still copying (the store falls back +/// to a copy when it cannot rename). +pub async fn lineup_guide_index_write( + file_locks: &FileLockManager, + playlist: &[PlaylistGroup], + view: LineupGuideView, + path: &Path, + carry_over: Option, +) -> Result<(), TuliproxError> { + let mut rows = playlist_rows(playlist, view); + if let Some(carry_over) = carry_over { + rows.extend(carry_over.rows); + } + let file_lock = file_locks.write_lock(path).await; + let path = path.to_path_buf(); + tokio::task::spawn_blocking(move || { + let _guard = file_lock; + let mut tree = tree_from_rows(rows); + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent).map_err(|err| { + TuliproxError::RepositoryPlaylist(format!( + "Failed to create lineup guide index directory {}: {err}", + parent.display() + )) + })?; + } + tree.store(&path).map(|_| ()).map_err(|err| { + TuliproxError::RepositoryPlaylist(format!("Failed to write lineup guide index {}: {err}", path.display())) + }) + }) + .await + .map_err(|err| TuliproxError::RepositoryPlaylist(format!("Lineup guide index write task failed: {err}")))? +} + +/// Removes the indexes at `paths`, so they cannot describe a playlist they were not built for. +/// +/// A missing file is fine: it is what a reader expects before the first persistence step, and +/// what a failed write leaves behind. Every target persistence step asks for this, so the common +/// case of no index at all costs one blocking hop for the existence checks and takes no lock. +pub async fn lineup_guide_index_remove(file_locks: &FileLockManager, paths: &[PathBuf]) -> Result<(), TuliproxError> { + let candidates = paths.to_vec(); + let existing = tokio::task::spawn_blocking(move || { + // An unanswerable existence check is treated as "exists", so the removal is still tried. + candidates.into_iter().filter(|path| path.try_exists().unwrap_or(true)).collect::>() + }) + .await + .map_err(|err| TuliproxError::RepositoryPlaylist(format!("Lineup guide index removal task failed: {err}")))?; + + for path in existing { + let _guard = file_locks.write_lock(&path).await; + match tokio::fs::remove_file(&path).await { + Err(err) if err.kind() != io::ErrorKind::NotFound => { + return Err(TuliproxError::RepositoryPlaylist(format!( + "Failed to remove lineup guide index {}: {err}", + path.display() + ))); + } + _ => {} + } + } + Ok(()) +} + +/// Every candidate the index holds for `keys`, grouped per key in the order the keys were given. +/// +/// Runs the file access on a blocking thread, so all keys are answered from one open index. An +/// index that cannot be opened is reported as [`LineupGuideLookup::Missing`] so the caller uses its +/// scan instead of failing the request. +pub async fn lineup_guide_candidates( + file_locks: &FileLockManager, + path: PathBuf, + keys: Vec>, +) -> LineupGuideLookup { + let file_lock = file_locks.read_lock(&path).await; + let task = tokio::task::spawn_blocking(move || { + let _guard = file_lock; + let Ok(mut query) = BPlusTreeQuery::, Vec>::try_new(&path) else { + return LineupGuideLookup::Missing; + }; + let mut candidates = Vec::with_capacity(keys.len()); + for key in keys { + match query.query(&key) { + Ok(Some(entries)) => candidates.push(LineupGuideCandidate { key, entries }), + Ok(None) => {} + Err(err) => { + log::warn!("Failed to query lineup guide index {}: {err}", path.display()); + return LineupGuideLookup::Missing; + } + } + } + if candidates.is_empty() { + LineupGuideLookup::NoEntry + } else { + LineupGuideLookup::Candidates(candidates) + } + }); + + match task.await { + Ok(lookup) => lookup, + // A lookup that could not run is "no index", so the caller falls back to its scan. + Err(err) => { + log::warn!("Lineup guide index lookup task failed: {err}"); + LineupGuideLookup::Missing + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use shared::{ + model::{PlaylistItem, PlaylistItemHeader, VirtualId}, + utils::Internable, + }; + use tempfile::TempDir; + use tuliprox_core::utils::FileLockManager; + + fn channel(virtual_id: u32, name: &str, epg: Option<&str>, cluster: XtreamCluster) -> PlaylistItem { + PlaylistItem { + header: PlaylistItemHeader { + virtual_id: VirtualId::new(virtual_id), + name: name.intern(), + epg_channel_id: epg.map(Internable::intern), + xtream_cluster: cluster, + ..PlaylistItemHeader::default() + }, + } + } + + fn group(id: u32, cluster: XtreamCluster, channels: Vec) -> PlaylistGroup { + PlaylistGroup { id, title: "Group".intern(), channels, xtream_cluster: cluster } + } + + /// Index content as `(guide number, candidates)`, ordered for comparison. + fn entries(tree: &LineupGuideTree) -> Vec<(String, Vec<(String, u32)>)> { + let mut entries: Vec<_> = tree + .iter() + .map(|(guide, candidates)| { + let candidates = candidates.iter().map(|entry| (entry.cluster.to_string(), entry.virtual_id)).collect(); + (guide.to_string(), candidates) + }) + .collect(); + entries.sort(); + entries + } + + #[test] + fn m3u_view_keeps_every_feed_of_a_guide_number_in_lineup_order() { + let playlist = vec![ + group(1, XtreamCluster::Live, vec![channel(9, "live feed", Some("shared.de"), XtreamCluster::Live)]), + group( + 2, + XtreamCluster::Video, + vec![ + channel(4, "vod feed", Some("shared.de"), XtreamCluster::Video), + channel(5, "without epg", None, XtreamCluster::Video), + ], + ), + ]; + + let tree = build_lineup_guide_index(&playlist, LineupGuideView::M3u); + + // The M3U playlist is one key space; with equal source ordinals the virtual id decides, + // and a channel without an EPG id is published under its name. + assert_eq!( + entries(&tree), + vec![ + ("shared.de".to_string(), vec![("video".to_string(), 4), ("live".to_string(), 9)]), + ("without epg".to_string(), vec![("video".to_string(), 5)]), + ] + ); + let candidates = tree.query(&"shared.de".intern()).expect("guide number is indexed"); + assert_eq!(candidates.first().map(|entry| entry.virtual_id), Some(4)); + } + + #[test] + fn feeds_follow_the_source_order_of_the_playlist_db_not_the_virtual_id() { + let mut hd = channel(3, "HD", Some("shared.de"), XtreamCluster::Live); + hd.header.source_ordinal = 1; + let mut sd = channel(1, "SD", Some("shared.de"), XtreamCluster::Live); + sd.header.source_ordinal = 2; + let mut vod = channel(2, "VOD", Some("shared.de"), XtreamCluster::Video); + vod.header.source_ordinal = 0; + let playlist = vec![group(1, XtreamCluster::Live, vec![sd, hd]), group(2, XtreamCluster::Video, vec![vod])]; + + // The dbs are read through their sorted index by source ordinal: the M3U db as one key + // space, the Xtream view live before video. + let ids = |view| { + build_lineup_guide_index(&playlist, view) + .query(&"shared.de".intern()) + .map(|entries| entries.iter().map(|entry| entry.virtual_id).collect::>()) + }; + assert_eq!(ids(LineupGuideView::M3u), Some(vec![2, 3, 1])); + assert_eq!(ids(LineupGuideView::Xtream), Some(vec![3, 1, 2])); + } + + #[test] + fn m3u_view_leaves_series_info_out() { + let mut series = channel(1, "series", Some("series.de"), XtreamCluster::Series); + series.header.item_type = PlaylistItemType::SeriesInfo; + let playlist = vec![group(1, XtreamCluster::Series, vec![series])]; + + // `m3u_write_playlist` does not store series info, so the index must not point at it. + assert!(entries(&build_lineup_guide_index(&playlist, LineupGuideView::M3u)).is_empty()); + } + + #[test] + fn xtream_view_lists_live_before_video_and_leaves_series_out() { + let playlist = vec![ + group(1, XtreamCluster::Live, vec![channel(9, "live feed", Some("shared.de"), XtreamCluster::Live)]), + group( + 2, + XtreamCluster::Video, + vec![ + channel(4, "vod feed", Some("shared.de"), XtreamCluster::Video), + channel(5, "smaller id", Some("only.vod"), XtreamCluster::Video), + channel(6, "smaller id again", Some("only.vod"), XtreamCluster::Video), + ], + ), + group(3, XtreamCluster::Series, vec![channel(1, "series", Some("series.de"), XtreamCluster::Series)]), + ]; + + let tree = build_lineup_guide_index(&playlist, LineupGuideView::Xtream); + + // The lineup searches the live db before the video db, and the whole series db is not part + // of the lineup at all. Within one cluster the smaller id comes first. + assert_eq!( + entries(&tree), + vec![ + ("only.vod".to_string(), vec![("video".to_string(), 5), ("video".to_string(), 6)]), + ("shared.de".to_string(), vec![("live".to_string(), 9), ("video".to_string(), 4)]), + ] + ); + } + + #[test] + fn empty_epg_channel_id_is_not_indexed_under_the_name() { + let playlist = vec![group( + 1, + XtreamCluster::Live, + vec![channel(1, "named", Some(""), XtreamCluster::Live), channel(2, "", None, XtreamCluster::Live)], + )]; + + let tree = build_lineup_guide_index(&playlist, LineupGuideView::M3u); + + // `lineup.json` publishes the empty EPG id as the guide number of the first channel, which + // no request can ask for, so neither the empty id nor the name may answer for it. + assert!(entries(&tree).is_empty()); + } + + #[tokio::test] + async fn index_round_trips_and_remove_tolerates_missing_files() { + let temp = TempDir::new().expect("temp dir created"); + let path = temp.path().join("lineup_guide.db"); + let locks = FileLockManager::default(); + let playlist = + vec![group(1, XtreamCluster::Live, vec![channel(7, "VOX", Some("vox.de"), XtreamCluster::Live)])]; + + assert_eq!( + lineup_guide_candidates(&locks, path.clone(), vec!["vox.de".intern()]).await, + LineupGuideLookup::Missing + ); + lineup_guide_index_write(&locks, &playlist, LineupGuideView::M3u, &path, None).await.expect("index written"); + assert_eq!( + lineup_guide_candidates(&locks, path.clone(), vec!["missing.de".intern(), "vox.de".intern()]).await, + LineupGuideLookup::Candidates(vec![LineupGuideCandidate { + key: "vox.de".intern(), + entries: vec![LineupGuideEntry { cluster: XtreamCluster::Live, virtual_id: 7 }], + }]) + ); + assert_eq!( + lineup_guide_candidates(&locks, path.clone(), vec!["missing.de".intern()]).await, + LineupGuideLookup::NoEntry + ); + + lineup_guide_index_remove(&locks, std::slice::from_ref(&path)).await.expect("index removed"); + lineup_guide_index_remove(&locks, std::slice::from_ref(&path)).await.expect("removing a missing index is fine"); + assert_eq!(lineup_guide_candidates(&locks, path, vec!["vox.de".intern()]).await, LineupGuideLookup::Missing); + } + + #[tokio::test] + async fn carry_over_keeps_the_feed_order_of_the_kept_cluster_after_the_rewritten_one() { + let temp = TempDir::new().expect("temp dir created"); + let path = temp.path().join("lineup_guide.db"); + let locks = FileLockManager::default(); + let mut first = channel(5, "VOD HD", Some("shared.de"), XtreamCluster::Video); + first.header.source_ordinal = 1; + let mut second = channel(4, "VOD SD", Some("shared.de"), XtreamCluster::Video); + second.header.source_ordinal = 2; + let previous = vec![ + group(1, XtreamCluster::Live, vec![channel(9, "old live", Some("shared.de"), XtreamCluster::Live)]), + group( + 2, + XtreamCluster::Video, + vec![first, second, channel(6, "film", Some("film.de"), XtreamCluster::Video)], + ), + ]; + lineup_guide_index_write(&locks, &previous, LineupGuideView::Xtream, &path, None).await.expect("index written"); + + // The update rewrites live only: the old live entry goes, the video entries stay in order. + let carry_over = lineup_guide_index_carry_over(&locks, &path, &[XtreamCluster::Video]) + .await + .expect("previous index readable"); + let update = + vec![group(1, XtreamCluster::Live, vec![channel(7, "new live", Some("shared.de"), XtreamCluster::Live)])]; + lineup_guide_index_write(&locks, &update, LineupGuideView::Xtream, &path, Some(carry_over)) + .await + .expect("index rewritten"); + + let tree = BPlusTreeQuery::, Vec>::try_new(&path) + .expect("index opens") + .query(&"shared.de".intern()); + let ids = + tree.expect("query runs").map(|entries| entries.iter().map(|entry| entry.virtual_id).collect::>()); + assert_eq!(ids, Some(vec![7, 5, 4])); + assert!(lineup_guide_index_carry_over(&locks, &temp.path().join("missing.db"), &[XtreamCluster::Video]) + .await + .is_none()); + } + + #[test] + fn an_item_listed_in_several_groups_is_indexed_once() { + let mut first = channel(5, "VOX", Some("vox.de"), XtreamCluster::Live); + first.header.source_ordinal = 1; + let mut other = channel(7, "VOX HD", Some("vox.de"), XtreamCluster::Live); + other.header.source_ordinal = 2; + let mut again = channel(5, "VOX", Some("vox.de"), XtreamCluster::Live); + again.header.source_ordinal = 3; + let playlist = + vec![group(1, XtreamCluster::Live, vec![first, other]), group(2, XtreamCluster::Live, vec![again])]; + + // The second listing of virtual id 5 is not adjacent to the first after sorting, the other + // feed lies between them. + let ids = build_lineup_guide_index(&playlist, LineupGuideView::M3u) + .query(&"vox.de".intern()) + .map(|entries| entries.iter().map(|entry| entry.virtual_id).collect::>()); + assert_eq!(ids, Some(vec![5, 7])); + } +} diff --git a/backend/repository/src/m3u_playlist_iterator.rs b/backend/repository/src/m3u_playlist_iterator.rs index a1fda3412..812adcf48 100644 --- a/backend/repository/src/m3u_playlist_iterator.rs +++ b/backend/repository/src/m3u_playlist_iterator.rs @@ -1,6 +1,6 @@ use crate::{ - ensure_target_storage_path, get_file_path_for_db_index, m3u_get_file_path_for_db, open_playlist_reader, - storage_const, user_get_bouquet_filter, LockedReceiverStream, + ensure_target_storage_path, get_file_path_for_db_index, m3u_get_file_path_for_db, m3u_get_item_for_stream_id, + open_playlist_reader, storage_const, user_get_bouquet_filter, LockedReceiverStream, PlaylistStorageState, }; use futures::{Stream, StreamExt}; use log::error; @@ -8,14 +8,14 @@ use shared::{ create_bitset, error::TuliproxError, model::{ - ConfigTargetOptions, M3uPlaylistItem, PlaylistItemType, ProxyType, ResourceOutputPolicy, StreamProperties, - TargetType, XtreamCluster, + ClusterFlags, ConfigTargetOptions, M3uPlaylistItem, PlaylistItemType, ProxyType, ResourceOutputPolicy, + StreamProperties, TargetType, XtreamCluster, }, utils::{concat_path, extract_extension_from_url, sanitize_sensitive_info, Internable, PROVIDER_SCHEME_PREFIX}, }; use std::{ borrow::Cow, - collections::HashMap, + collections::{HashMap, HashSet}, pin::Pin, sync::Arc, task::{Context, Poll}, @@ -283,25 +283,31 @@ async fn rewrite_player_resources_for_output(item: &mut M3uPlaylistItem, ctx: &U } } -#[allow(clippy::too_many_lines)] -impl M3uPlaylistIterator { - pub async fn new( - cfg: &AppConfig, - target: &ConfigTarget, - user: &ProxyUserCredentials, - ) -> Result { - // TODO use playlist memory cache, but be aware of sorting ! +/// Everything the M3U output pipeline needs for one item of one user. +/// +/// The playlist iterators and the single-item lookup below have to answer with the same item for +/// the same user, so the per-item checks and the URL rewrite live here and both use them. +struct M3uItemPipeline { + filter: Option>, + user_filter: Option>, + output_clusters: ClusterFlags, + base_url: String, + username: String, + password: String, + target_id: u16, + proxy_type: ProxyType, + target_options: Option, + encrypt_secret: [u8; 16], + input_by_name: HashMap, Arc>, + flags: M3uPlaylistIteratorFlagsSet, +} +impl M3uItemPipeline { + async fn new(cfg: &AppConfig, target: &ConfigTarget, user: &ProxyUserCredentials) -> Result { let m3u_output = target.get_m3u_output().ok_or_else(|| { TuliproxError::Config(format!("Unexpected failure, missing m3u target output for target {}", target.name)) })?; let config = cfg.config.load(); - 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 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(); if m3u_output.include_type_in_url { @@ -313,19 +319,116 @@ impl M3uPlaylistIterator { if cfg.is_reverse_proxy_resource_rewrite_enabled() { flags.set(M3uPlaylistIteratorFlags::RewriteResource); } - let base_url = cfg.get_user_server_info(user).map(|si| si.get_base_url()).unwrap_or_default(); let encrypt_secret = cfg.get_reverse_proxy_rewrite_secret().unwrap_or(cfg.encrypt_secret); - let username = user.username.clone(); - let password = user.password.clone(); - let target_id = target.id; - let proxy_type = user.proxy; - let output_clusters = user.output_clusters; - let user_filter = user.t_filter.clone(); - let target_options = target.options.clone(); let input_by_name: HashMap, Arc> = cfg.sources.load().inputs.iter().map(|input| (Arc::clone(&input.name), Arc::clone(input))).collect(); + Ok(Self { + filter, + user_filter: user.t_filter.clone(), + output_clusters: user.output_clusters, + base_url, + username: user.username.clone(), + password: user.password.clone(), + target_id: target.id, + proxy_type: user.proxy, + target_options: target.options.clone(), + encrypt_secret, + input_by_name, + flags, + }) + } + + /// The item the M3U output publishes for this user, or `None` when it is not part of it. + #[inline] + fn apply(&self, item: M3uPlaylistItem) -> Option { + if !self.output_clusters.has_cluster(item.item_type) { + return None; + } + + if let Some(set) = &self.filter { + if !set.contains(item.group.as_ref()) { + return None; + } + } + + // per-user content filter (plan AND user), applied post-cache + if let Some(user_filter) = &self.user_filter { + let pli = shared::model::PlaylistItem::from(&item); + let provider = shared::foundation::ValueProvider { pli: &pli, match_as_ascii: false }; + if !user_filter.filter(&provider) { + return None; + } + } + + let rewrite_ctx = + UrlRewriteContext { base_url: &self.base_url, username: &self.username, password: &self.password }; + Some(apply_rewrite( + item, + &rewrite_ctx, + self.target_id, + &self.encrypt_secret, + &self.input_by_name, + self.target_options.as_ref(), + self.flags, + self.proxy_type, + )) + } +} + +/// Single item lookup of one user's M3U output. +/// +/// The `HDHomeRun` tuner endpoints resolve the guide numbers they find in the target guide index +/// with this instead of walking the playlist. It answers with the same item the iterators publish, +/// and the per-user pipeline (bouquet, content filter, url rewrite) is prepared once for all the +/// candidates of one request. +pub struct M3uUserItemLookup<'a> { + pipeline: M3uItemPipeline, + cfg: &'a AppConfig, + playlists: &'a PlaylistStorageState, + target: &'a ConfigTarget, +} + +impl<'a> M3uUserItemLookup<'a> { + pub async fn new( + cfg: &'a AppConfig, + playlists: &'a PlaylistStorageState, + target: &'a ConfigTarget, + user: &ProxyUserCredentials, + ) -> Result { + Ok(Self { pipeline: M3uItemPipeline::new(cfg, target, user).await?, cfg, playlists, target }) + } + + /// The item the output publishes for `stream_id`, or `None` when the playlist does not hold it + /// or this user's filters exclude it. + pub async fn get(&self, stream_id: u32) -> Option { + match m3u_get_item_for_stream_id(stream_id, self.cfg, self.playlists, self.target).await { + Ok(item) => self.pipeline.apply(item), + Err(err) => { + log::debug!("No m3u playlist item {stream_id} for target {}: {err}", self.target.name); + None + } + } + } +} + +impl M3uPlaylistIterator { + pub async fn new( + cfg: &AppConfig, + target: &ConfigTarget, + user: &ProxyUserCredentials, + ) -> Result { + // TODO use playlist memory cache, but be aware of sorting ! + + let pipeline = M3uItemPipeline::new(cfg, target, user).await?; + let config = cfg.config.load(); + 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 iter_lock = cfg.file_locks.read_lock(&m3u_path).await; + let bg_lock = cfg.file_locks.read_lock(&m3u_path).await; + let m3u_path = m3u_path.clone(); let index_path = get_file_path_for_db_index(&m3u_path); let (tx, rx) = mpsc::channel::>(256); @@ -358,37 +461,9 @@ impl M3uPlaylistIterator { } }; - if !output_clusters.has_cluster(item.item_type) { + let Some(item) = pipeline.apply(item) else { continue; - } - - if let Some(set) = &filter { - if !set.contains(item.group.as_ref()) { - continue; - } - } - - // per-user content filter (plan AND user), applied post-cache - if let Some(user_filter) = &user_filter { - let pli = shared::model::PlaylistItem::from(&item); - let provider = shared::foundation::ValueProvider { pli: &pli, match_as_ascii: false }; - if !user_filter.filter(&provider) { - continue; - } - } - - let rewrite_ctx = UrlRewriteContext { base_url: &base_url, username: &username, password: &password }; - - let item = apply_rewrite( - item, - &rewrite_ctx, - target_id, - &encrypt_secret, - &input_by_name, - target_options.as_ref(), - flags, - proxy_type, - ); + }; if let Some(prev) = pending.replace(item) { if tx.blocking_send(Ok((prev, true))).is_err() { diff --git a/backend/repository/src/playlist_repository/persist.rs b/backend/repository/src/playlist_repository/persist.rs index 4c4ac8543..9aa0cd0f6 100644 --- a/backend/repository/src/playlist_repository/persist.rs +++ b/backend/repository/src/playlist_repository/persist.rs @@ -9,12 +9,21 @@ use super::{ PlaylistStorageState, TargetPlaylistPersistOptions, }; use crate::{ - ensure_target_storage_path, epg_write_for_target, load_input_local_library_playlist, m3u_write_playlist, - persist_input_library_playlist, write_strm_playlist, + ensure_target_storage_path, epg_write_for_target, lineup_guide_index_carry_over, lineup_guide_index_remove, + lineup_guide_index_view, lineup_guide_index_write, load_input_local_library_playlist, m3u_lineup_guide_index_path, + m3u_write_playlist, persist_input_library_playlist, write_strm_playlist, xtream_get_file_path, + xtream_get_storage_path, xtream_lineup_guide_index_path, LineupGuideCarryOver, LineupGuideView, }; -use shared::{error::TuliproxError, model::PlaylistGroup}; -use std::{path::Path, sync::Arc}; -use tuliprox_core::model::{AppConfig, ConfigTarget, Epg, TargetOutput}; +use log::warn; +use shared::{ + error::TuliproxError, + model::{ClusterFlags, PlaylistGroup, XtreamCluster}, +}; +use std::{ + path::{Path, PathBuf}, + sync::Arc, +}; +use tuliprox_core::model::{AppConfig, Config, ConfigTarget, Epg, TargetOutput}; pub async fn persist_playlist( app_config: &Arc, @@ -113,6 +122,22 @@ async fn persist_playlist_views_with_mode( return Err(vec![error]); } + // Only the view `lineup.json` is published from keeps a guide index. Any other index would + // describe a playlist the reader no longer asks, and a later re-configuration could mistake it + // for a current one. + // + // The guide index is a sidecar: none of its failures may fail the playlist publication or skip + // the memory cache reload below. A missing index only makes the tuners scan the playlist. + let lineup_view = lineup_guide_index_view(target); + let outdated_indexes: Vec = [LineupGuideView::M3u, LineupGuideView::Xtream] + .into_iter() + .filter(|view| lineup_view != Some(*view)) + .filter_map(|view| lineup_guide_index_path(config, target, view, &target_path)) + .collect(); + if let Err(err) = lineup_guide_index_remove(&app_config.file_locks, &outdated_indexes).await { + warn!("Outdated HDHomeRun guide index of target {} not removed: {err}", target.name); + } + for (output, prepared_output) in target.output.iter().zip(&mut prepared_outputs) { let output_publication_plan = options.publication_plan.with_output_filter(output.filter().is_some()); let source_playlist = match output { @@ -129,16 +154,51 @@ async fn persist_playlist_views_with_mode( let allows_empty_base = library_empty.replaces_empty_target() || output_publication_plan.allows_empty_base_output(); + let replace_empty_clusters = + options.accepted_empty_clusters | library_empty.replacement_clusters() | curation_empty_clusters; + + // Guide index of the HDHomeRun lineup. It describes the playlist db of this output, so it + // follows what the writer below does with that db: + // - rewritten: the old index goes before the db and the new one after it. A reader that + // loses the index falls back to its playlist scan, while one that kept an index of another + // playlist would resolve guide numbers to channels this target no longer serves. + // - kept (an empty update that may not replace it): the old index still describes it. + // - partly rewritten (one Xtream cluster kept): the entries of the kept cluster are carried + // over from the old index, read before it goes. Without an old index to read them from, + // the index goes and readers scan until a full update brings it back. + let mut guide_index = match (lineup_view, output) { + (Some(view @ LineupGuideView::M3u), TargetOutput::M3u(_)) => { + // `m3u_write_playlist` keeps the db of an empty playlist it may not publish. + let action = if pl.is_empty() && !allows_empty_base { + GuideIndexAction::Keep + } else { + GuideIndexAction::Rebuild(None) + }; + lineup_guide_index_path(config, target, view, &target_path).map(|path| (view, path, action)) + } + (Some(view @ LineupGuideView::Xtream), TargetOutput::Xtream(_)) => { + match lineup_guide_index_path(config, target, view, &target_path) { + Some(path) => { + let action = + xtream_guide_index_action(app_config, target, pl, replace_empty_clusters, &path).await; + Some((view, path, action)) + } + None => None, + } + } + _ => None, + }; + if let Some((_, path, GuideIndexAction::Rebuild(_) | GuideIndexAction::Remove)) = &guide_index { + if let Err(err) = lineup_guide_index_remove(&app_config.file_locks, std::slice::from_ref(path)).await { + // The rebuild below still replaces it. Only if that fails too does the old index + // stay, which is why both failures are logged. + warn!("HDHomeRun guide index of target {} not removed: {err}", target.name); + } + } + let result = match output { TargetOutput::Xtream(_xtream_output) => { - persist_xtream_target_playlist( - app_config, - target, - pl, - options.accepted_empty_clusters | library_empty.replacement_clusters() | curation_empty_clusters, - persistence_mode, - ) - .await + persist_xtream_target_playlist(app_config, target, pl, replace_empty_clusters, persistence_mode).await } TargetOutput::M3u(m3u_output) => { m3u_write_playlist(app_config, target, m3u_output, &target_path, pl, allows_empty_base).await @@ -151,6 +211,16 @@ async fn persist_playlist_views_with_mode( match result { Ok(()) => { + if let Some((view, path, GuideIndexAction::Rebuild(carry_over))) = &mut guide_index { + if let Err(err) = + lineup_guide_index_write(&app_config.file_locks, pl, *view, path, carry_over.take()).await + { + warn!( + "HDHomeRun guide index of target {} not written, tuners scan the playlist: {err}", + target.name + ); + } + } let allows_empty_output = match output { TargetOutput::Xtream(_) => { !curation_empty_clusters.is_empty() @@ -217,3 +287,83 @@ pub async fn load_input_media_server_playlist( ) -> Result, TuliproxError> { load_input_local_library_playlist(app_config, file_path).await } + +/// What a persistence step does with the guide index of the output it writes. +#[derive(Debug)] +enum GuideIndexAction { + /// The playlist db is rewritten from `pl`, so is the index, plus the entries of the clusters the + /// db keeps. + Rebuild(Option), + /// The playlist db stays as it is, and so does the index describing it. + Keep, + /// Only part of the db is rewritten and no old index describes the rest. + Remove, +} + +/// Guide index action of the Xtream view, from what `xtream_write_playlist` does per cluster: a +/// cluster without items keeps its db unless it may be replaced by an empty one. A cluster that has +/// no db yet keeps nothing, so `pl` still describes it. The entries of a kept cluster are read from +/// the index at `index_path` now, before that index is removed for the rewrite. +async fn xtream_guide_index_action( + app_config: &AppConfig, + target: &ConfigTarget, + pl: &[PlaylistGroup], + replace_empty_clusters: ClusterFlags, + index_path: &Path, +) -> GuideIndexAction { + // The writer sorts items into clusters by their own cluster, not by their group's, so the items + // are checked. One pass answers both clusters and stops once both are seen. + let (mut has_live, mut has_video) = (false, false); + for item in pl.iter().flat_map(|group| &group.channels) { + match item.header.xtream_cluster { + XtreamCluster::Live => has_live = true, + XtreamCluster::Video => has_video = true, + XtreamCluster::Series => {} + } + if has_live && has_video { + break; + } + } + + let storage_path = xtream_get_storage_path(&app_config.config.load(), &target.name); + let mut rewritten = false; + let mut kept = Vec::new(); + for (cluster, flag, has_items) in + [(XtreamCluster::Live, ClusterFlags::Live, has_live), (XtreamCluster::Video, ClusterFlags::Vod, has_video)] + { + if has_items || replace_empty_clusters.contains(flag) { + rewritten = true; + continue; + } + let has_db = match &storage_path { + Some(path) => tokio::fs::try_exists(xtream_get_file_path(path, cluster)).await.unwrap_or(true), + None => false, + }; + if has_db { + kept.push(cluster); + } + } + if kept.is_empty() { + GuideIndexAction::Rebuild(None) + } else if !rewritten { + GuideIndexAction::Keep + } else { + match lineup_guide_index_carry_over(&app_config.file_locks, index_path, &kept).await { + Some(carry_over) => GuideIndexAction::Rebuild(Some(carry_over)), + None => GuideIndexAction::Remove, + } + } +} + +/// Guide index file of `view` for `target`. +fn lineup_guide_index_path( + config: &Config, + target: &ConfigTarget, + view: LineupGuideView, + target_path: &Path, +) -> Option { + match view { + LineupGuideView::M3u => Some(m3u_lineup_guide_index_path(target_path)), + LineupGuideView::Xtream => xtream_lineup_guide_index_path(config, &target.name), + } +} diff --git a/backend/repository/src/xtream_playlist_iterator.rs b/backend/repository/src/xtream_playlist_iterator.rs index 4e3841f91..099fe9ef2 100644 --- a/backend/repository/src/xtream_playlist_iterator.rs +++ b/backend/repository/src/xtream_playlist_iterator.rs @@ -1,6 +1,6 @@ use crate::{ get_file_path_for_db_index, open_playlist_reader, user_get_bouquet_filter, xtream_get_file_path, - xtream_get_storage_path, LockedReceiverStream, + xtream_get_item_for_stream_id, xtream_get_storage_path, LockedReceiverStream, PlaylistStorageState, }; use futures::{Stream, StreamExt}; use log::error; @@ -16,7 +16,7 @@ use std::{ }; use tokio::{sync::mpsc, task}; use tuliprox_core::{ - model::{xtream_mapping_option_from_target_options, AppConfig, ConfigTarget, ProxyUserCredentials}, + model::{xtream_mapping_option_from_target_options, AppConfig, Config, ConfigTarget, ProxyUserCredentials}, utils::prepare_xtream_resource_hosts, }; @@ -28,6 +28,69 @@ fn is_cluster_allowed_for_user(user: &ProxyUserCredentials, cluster: XtreamClust user.allows_cluster(cluster) } +/// Bouquet and content filter of one user for one Xtream cluster. +/// +/// The playlist iterator and the single-item lookup below have to answer with the same item for +/// the same user, so the per-item checks live here and both use them. +struct XtreamItemFilter { + cluster: XtreamCluster, + filter_ids: Option>, + user_filter: Option>, +} + +impl XtreamItemFilter { + async fn new( + config: &Config, + cluster: XtreamCluster, + category_id: Option, + user: &ProxyUserCredentials, + ) -> Self { + 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 + let filter_ids: Option> = filter.as_ref().map(|set| { + set.iter() + .filter_map(|s| { + s.parse::() + .map_err(|e| { + error!("Failed to parse bouquet filter id '{s}': {e}"); + e + }) + .ok() + }) + .collect() + }); + Self { cluster, filter_ids, user_filter: user.t_filter.clone() } + } + + #[inline] + fn matches(&self, item: &XtreamPlaylistItem) -> bool { + // We can't serve episodes within series + if self.cluster == XtreamCluster::Series + && !matches!(item.item_type, PlaylistItemType::SeriesInfo | PlaylistItemType::LocalSeriesInfo) + { + return false; + } + + // category_id-Filter + if let Some(set) = &self.filter_ids { + if !set.contains(&item.category_id) { + return false; + } + } + + // per-user content filter (plan AND user), applied post-cache + if let Some(filter) = &self.user_filter { + let pli = shared::model::PlaylistItem::from(item); + let provider = shared::foundation::ValueProvider { pli: &pli, match_as_ascii: false }; + if !filter.filter(&provider) { + return false; + } + } + + true + } +} + impl XtreamPlaylistIterator { fn empty() -> Self { // Ghost Channel pattern @@ -59,26 +122,11 @@ impl XtreamPlaylistIterator { 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 - let filter_ids: Option> = filter.as_ref().map(|set| { - set.iter() - .filter_map(|s| { - s.parse::() - .map_err(|e| { - error!("Failed to parse bouquet filter id '{s}': {e}"); - e - }) - .ok() - }) - .collect() - }); + let item_filter = XtreamItemFilter::new(&config, cluster, category_id, user).await; let xtream_path = xtream_path.clone(); let index_path = get_file_path_for_db_index(&xtream_path); let (tx, rx) = mpsc::channel::>(256); - let user_filter = user.t_filter.clone(); let xtream_path_for_log = xtream_path.clone(); let join_error_tx = tx.clone(); @@ -110,7 +158,7 @@ impl XtreamPlaylistIterator { } }; - if !Self::matches_filters(cluster, filter_ids.as_ref(), user_filter.as_ref(), &item) { + if !item_filter.matches(&item) { continue; } @@ -145,37 +193,53 @@ impl XtreamPlaylistIterator { Err(TuliproxError::Config(format!("Failed to find xtream storage for target {}", target.name))) } } +} - fn matches_filters( - cluster: XtreamCluster, - filter_ids: Option<&HashSet>, - user_filter: Option<&Arc>, - item: &XtreamPlaylistItem, - ) -> bool { - // We can't serve episodes within series - if cluster == XtreamCluster::Series - && !matches!(item.item_type, PlaylistItemType::SeriesInfo | PlaylistItemType::LocalSeriesInfo) +/// Single item lookup of one user's Xtream output. +/// +/// The `HDHomeRun` tuner endpoints resolve the guide numbers they find in the target guide index +/// with this instead of walking the playlists. It answers with the same item the iterators publish, +/// and the bouquet and content filter of every cluster are prepared once for all the candidates of +/// one request. +pub struct XtreamUserItemLookup<'a> { + app_config: &'a Arc, + playlists: &'a PlaylistStorageState, + target: &'a ConfigTarget, + /// Filters of the requested clusters this user may watch. + filters: Vec, +} + +impl<'a> XtreamUserItemLookup<'a> { + pub async fn new( + app_config: &'a Arc, + playlists: &'a PlaylistStorageState, + target: &'a ConfigTarget, + user: &ProxyUserCredentials, + clusters: &[XtreamCluster], + ) -> Self { + let config = app_config.config.load(); + let mut filters = Vec::with_capacity(clusters.len()); + for &cluster in clusters { + if is_cluster_allowed_for_user(user, cluster) { + filters.push(XtreamItemFilter::new(&config, cluster, None, user).await); + } + } + Self { app_config, playlists, target, filters } + } + + /// The item the output publishes for `stream_id` in `cluster`, or `None` when the cluster is + /// not served to this user, the playlist does not hold the id, or this user's filters exclude it. + pub async fn get(&self, stream_id: u32, cluster: XtreamCluster) -> Option { + let filter = self.filters.iter().find(|filter| filter.cluster == cluster)?; + match xtream_get_item_for_stream_id(stream_id, self.app_config, self.playlists, self.target, Some(cluster)) + .await { - return false; - } - - // category_id-Filter - if let Some(set) = filter_ids { - if !set.contains(&item.category_id) { - return false; + Ok(item) => filter.matches(&item).then_some(item), + Err(err) => { + log::debug!("No xtream playlist item {stream_id} in {cluster} for target {}: {err}", self.target.name); + None } } - - // per-user content filter (plan AND user), applied post-cache - if let Some(filter) = user_filter { - let pli = shared::model::PlaylistItem::from(item); - let provider = shared::foundation::ValueProvider { pli: &pli, match_as_ascii: false }; - if !filter.filter(&provider) { - return false; - } - } - - true } } @@ -238,7 +302,7 @@ impl Stream for XtreamPlaylistJsonIterator { #[cfg(test)] mod tests { - use super::{is_cluster_allowed_for_user, XtreamPlaylistIterator}; + use super::{is_cluster_allowed_for_user, XtreamItemFilter, XtreamPlaylistIterator}; use crate::LockedReceiverStream; use futures::StreamExt; use shared::{ @@ -286,16 +350,15 @@ mod tests { shared::foundation::get_filter(r#"NOT Group ~ "^VIP.*""#, None).expect("filter parses"), ); + let without_filter = XtreamItemFilter { cluster: XtreamCluster::Live, filter_ids: None, user_filter: None }; + let with_filter = + XtreamItemFilter { cluster: XtreamCluster::Live, filter_ids: None, user_filter: Some(filter) }; + // no filter: everything passes - assert!(XtreamPlaylistIterator::matches_filters(XtreamCluster::Live, None, None, &make_item("VIP Sports"))); + assert!(without_filter.matches(&make_item("VIP Sports"))); // filter hides matching groups, keeps the rest - assert!(!XtreamPlaylistIterator::matches_filters( - XtreamCluster::Live, - None, - Some(&filter), - &make_item("VIP Sports") - )); - assert!(XtreamPlaylistIterator::matches_filters(XtreamCluster::Live, None, Some(&filter), &make_item("News"))); + assert!(!with_filter.matches(&make_item("VIP Sports"))); + assert!(with_filter.matches(&make_item("News"))); } #[tokio::test]