From fc51a3acd46b83059cd6ced9ff6583c9dadd4313 Mon Sep 17 00:00:00 2001 From: euzu <33094714+euzu@users.noreply.github.com> Date: Thu, 24 Sep 2026 00:51:12 +0200 Subject: [PATCH] staged m3u to xtream fix (#877) * staged m3u to xtream fix * staged m3u to xtream fix * staged m3u to xtream fix * fixed race some conditions * fixed race some conditions * fixed race some conditions * fixed race some conditions * fixed race some conditions --------- Co-authored-by: euzu --- CHANGELOG.md | 10 +- backend/app/src/api/api_utils/tests.rs | 102 ++- .../src/processor/playlist/ingest.rs | 432 ++++++++- .../processing/src/processor/playlist/mod.rs | 4 +- .../src/processor/playlist/tests.rs | 829 +++++++++++++++++- .../session/src/active_provider_manager.rs | 155 +++- .../session/src/active_user_manager/mod.rs | 27 +- backend/session/src/admission.rs | 100 ++- .../session/src/provider_lineup_manager.rs | 131 +-- docs/src/configuration/source.md | 44 +- .../components/source_editor/block_view.rs | 26 +- .../components/source_editor/editor_view.rs | 596 ++++++------- 12 files changed, 1973 insertions(+), 483 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index cd5acdfb1..673d9dad6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -58,9 +58,13 @@ provider inputs (with `enabled`, `live_source`, `vod_source`, and `series_source`) has been removed. A staged source is now its own input with `type: staged`. It points to one non-staged `m3u` / `xtream` provider through `staged.for_input`, and `staged.clusters` selects which clusters (`live`, - `vod`, `series`) are loaded from the staged playlist. Clusters not selected there are loaded from the - provider input itself. The merged result is stored under the provider input, so playlist delivery and - stream/API routing continue to use the provider. + `vod`, `series`) are overlaid by the staged playlist. Inside such a cluster of an `xtream` provider each staged group + replaces the provider category it belongs to, matched by the staged channels' stream IDs and only then by category ID + or group title; provider categories without a staged counterpart stay as they are. For any other provider type the + selected clusters are replaced entirely by the staged groups, and the clusters not selected are loaded from the + provider input itself. The merged result is stored under the provider input, so playlist delivery and stream/API + routing continue to use the provider. See + [Staged Sources](docs/src/configuration/source.md#25-staged-sources-staged) for the matching rules. Before: diff --git a/backend/app/src/api/api_utils/tests.rs b/backend/app/src/api/api_utils/tests.rs index a20288284..b5a02de12 100644 --- a/backend/app/src/api/api_utils/tests.rs +++ b/backend/app/src/api/api_utils/tests.rs @@ -6052,7 +6052,7 @@ async fn grace_context_is_populated_when_grace_strategy_is_actually_granted() { async fn evaluate_remaining_strategies_evicts_after_used_grace() { // Strategies: [GraceHoldStream, EvictUserOldest] // Grace was used at index 0, so only EvictUserOldest (index 1) is evaluated. - // Eviction frees the slot -> Allowed. + // The new request stays retryable while the evicted stream still owns its provider slot. let strategies = vec![AdmissionStrategy::GraceHoldStream, AdmissionStrategy::EvictUserOldest]; let grace_context = GraceResolutionContext { strategy_index: 0, strategies: strategies.into(), kind: None }; @@ -6073,6 +6073,9 @@ async fn evaluate_remaining_strategies_evicts_after_used_grace() { recent_eviction_reentry_ttl: std::time::Duration::from_millis(1500), admission_strategies: Some(vec![AdmissionStrategy::GraceHoldStream, AdmissionStrategy::EvictUserOldest]), }); + let provider_config = create_test_provider_app_config(); + app_state.app_config.sources.store(provider_config.sources.load_full()); + app_state.active_provider.update_config(&app_state.app_config); let addr1: SocketAddr = "127.0.0.1:55701".parse().unwrap_or_else(|_| unreachable!()); let addr2: SocketAddr = "10.0.0.5:55702".parse().unwrap_or_else(|_| unreachable!()); @@ -6121,23 +6124,71 @@ async fn evaluate_remaining_strategies_evicts_after_used_grace() { .await .expect("stream should be created"); - let result = evaluate_remaining_strategies_after_grace( - &app_state.admission_ctx(), - AdmissionRequest { - username: "remaining-evict", - max_connections: 1, - soft_connections: 0, - client_ip: &fingerprint2.client_ip, - request_addr: &fingerprint2.addr, - use_session_admission: true, - session_token: Some("tok-new"), - activate_unbound_session: true, - eviction_reentry_guard: EvictionReentryGuard::Session("tok-new"), - }, - &grace_context, - Some(crate::api::model::ConnectionKind::Normal), + let provider_handle = app_state + .active_provider + .acquire_connection_with_grace_for_session( + &"provider_1".intern(), + &addr1, + false, + 0, + crate::api::model::ConnectionKind::Normal, + Some("tok-counted"), + ) + .expect("old stream should occupy the only provider slot"); + assert!(app_state.active_provider.register_body_owner(provider_handle.allocation_id)); + let close_rx = app_state.connection_manager.register_close_socket(addr1); + let manager = Arc::clone(&app_state.connection_manager); + let provider = Arc::clone(&app_state.active_provider); + let release_body = Arc::new(tokio::sync::Notify::new()); + let release_body_after_timeout = Arc::clone(&release_body); + let close_task = tokio::spawn(async move { + assert_eq!(close_rx.await.expect("kick close signal"), shared::model::DisconnectReason::ClientKicked); + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + manager.release_provider_deferred(&addr1).await; + assert_eq!(provider.get_provider_connections_count(), 1, "body owner still holds provider capacity"); + release_body_after_timeout.notified().await; + provider.release_handle(&provider_handle); + provider_handle.completion_token.as_ref().expect("body completion token").cancel(); + manager.unregister_close_socket(&addr1); + }); + + let request = || AdmissionRequest { + username: "remaining-evict", + max_connections: 1, + soft_connections: 0, + client_ip: &fingerprint2.client_ip, + request_addr: &fingerprint2.addr, + use_session_admission: true, + session_token: Some("tok-new"), + activate_unbound_session: true, + eviction_reentry_guard: EvictionReentryGuard::Session("tok-new"), + }; + let result = tokio::time::timeout( + std::time::Duration::from_secs(5), + evaluate_remaining_strategies_after_grace( + &app_state.admission_ctx(), + request(), + &grace_context, + Some(crate::api::model::ConnectionKind::Normal), + ), ) - .await; + .await + .expect("admission must remain retryable while the provider slot is held"); + + assert_eq!(result.admission.permission(), UserConnectionPermission::Exhausted); + assert_eq!(app_state.active_provider.get_provider_connections_count(), 1); + + let retry = tokio::time::timeout( + std::time::Duration::from_secs(5), + resolve_admission_with_strategies(&app_state.admission_ctx(), request()), + ) + .await + .expect("retry must remain bounded while the provider slot is held"); + assert_eq!(retry.admission.permission(), UserConnectionPermission::Exhausted); + + release_body.notify_one(); + close_task.await.expect("old transport cleanup"); + let result = resolve_admission_with_strategies(&app_state.admission_ctx(), request()).await; assert_eq!( result.admission.permission(), @@ -6145,6 +6196,23 @@ async fn evaluate_remaining_strategies_evicts_after_used_grace() { "EvictUserOldest should free the slot" ); assert!(result.grace_context.is_none(), "no grace context on eviction success"); + assert_eq!( + app_state.active_provider.get_provider_connections_count(), + 0, + "admission must not return while the evicted stream still owns the provider slot" + ); + let replacement = app_state.active_provider.acquire_connection_with_grace_for_session( + &"provider_1".intern(), + &addr2, + false, + 0, + crate::api::model::ConnectionKind::Normal, + Some("tok-new"), + ); + assert!(replacement.is_some(), "newly admitted stream must acquire the freed provider slot"); + if let Some(replacement) = replacement { + app_state.active_provider.release_handle(&replacement); + } } #[tokio::test] diff --git a/backend/processing/src/processor/playlist/ingest.rs b/backend/processing/src/processor/playlist/ingest.rs index 68fbb891f..2614eeb03 100644 --- a/backend/processing/src/processor/playlist/ingest.rs +++ b/backend/processing/src/processor/playlist/ingest.rs @@ -485,25 +485,432 @@ pub(crate) fn cluster_selected(cluster: XtreamCluster, clusters: ClusterFlags) - } } +fn prepare_staged_xtream_group( + mut group: PlaylistGroup, + provider: &ConfigInput, + provider_urls: &HashMap<(XtreamCluster, u32), &Arc>, + username: &str, + password: &str, + live_stream_use_prefix: bool, + live_stream_without_extension: bool, +) -> PlaylistGroup { + let initial_count = group.channels.len(); + group.channels.retain_mut(|item| { + // The staged item's ID identifies its original Xtream stream. + let Ok(stream_id) = item.header.id.parse::() else { + return false; + }; + // Existing streams keep the provider's playback URL, including a direct source URL. + item.header.url = provider_urls.get(&(group.xtream_cluster, stream_id)).map_or_else( + || { + let metadata_extension = item.header.get_container_extension(); + let container_extension = if group.xtream_cluster == XtreamCluster::Video { + metadata_extension.as_deref().or_else(|| { + shared::utils::extract_extension_from_url(&item.header.url) + .and_then(|extension| extension.strip_prefix('.')) + }) + } else { + None + }; + tuliprox_parser::xtream::get_xtream_url( + group.xtream_cluster, + &provider.url, + username, + password, + stream_id, + container_extension, + live_stream_use_prefix, + live_stream_without_extension, + ) + .into() + }, + |url| Arc::clone(url), + ); + item.header.input_name = Arc::clone(&provider.name); + true + }); + let dropped = initial_count - group.channels.len(); + if dropped > 0 { + warn!("Skipped {dropped} staged channel(s) in group '{}' without a numeric Xtream stream ID", group.title); + } + group +} + +/// Overlays the staged source on the provider playlist of the clusters it covers. +/// +/// The staged playlist is the provider playlist restructured by an external tool: group and channel +/// names may differ, the ids do not. Staged groups are therefore resolved against the provider groups +/// by id, never by name ([`match_staged_groups`]). An overlaid group keeps the provider category id; +/// a staged group that introduces a new category keeps its own id only when that id is still free. A +/// staged group whose channels carry no numeric stream id leaves the provider category untouched, so +/// unusable overlay rows cannot remove provider content. pub(crate) fn apply_staged_overlay_groups( - provider_name: &Arc, + provider: &ConfigInput, + staged_type: StagedInputType, clusters: ClusterFlags, provider_groups: Vec, staged_groups: Vec, ) -> Vec { - let mut groups: Vec = - provider_groups.into_iter().filter(|group| !cluster_selected(group.xtream_cluster, clusters)).collect(); + if provider.input_type != InputType::Xtream { + let mut groups: Vec = + provider_groups.into_iter().filter(|group| !cluster_selected(group.xtream_cluster, clusters)).collect(); + groups.extend(staged_groups.into_iter().filter(|group| cluster_selected(group.xtream_cluster, clusters)).map( + |mut group| { + for item in &mut group.channels { + item.header.input_name = Arc::clone(&provider.name); + } + group + }, + )); + return groups; + } - groups.extend(staged_groups.into_iter().filter(|group| cluster_selected(group.xtream_cluster, clusters)).map( - |mut group| { - for item in &mut group.channels { - item.header.input_name = Arc::clone(provider_name); + let (Some(username), Some(password)) = (provider.username.as_deref(), provider.password.as_deref()) else { + warn!("Skipping staged channels for Xtream input '{}' without credentials", provider.name); + return provider_groups; + }; + let live_stream_use_prefix = provider.has_flag(ConfigInputFlags::XtreamLiveStreamUsePrefix); + let live_stream_without_extension = provider.has_flag(ConfigInputFlags::XtreamLiveStreamWithoutExtension); + + let staged_stream_ids: HashSet<(XtreamCluster, u32)> = staged_groups + .iter() + .filter(|group| cluster_selected(group.xtream_cluster, clusters)) + .flat_map(|group| { + group.channels.iter().filter_map(|item| { + item.header.id.parse::().ok().map(|stream_id| (group.xtream_cluster, stream_id)) + }) + }) + .collect(); + let mut provider_urls = HashMap::with_capacity(staged_stream_ids.len()); + for group in &provider_groups { + if !cluster_selected(group.xtream_cluster, clusters) { + continue; + } + for item in &group.channels { + if let Ok(stream_id) = item.header.id.parse::() { + let key = (group.xtream_cluster, stream_id); + if staged_stream_ids.contains(&key) { + provider_urls.entry(key).or_insert(&item.header.url); + } } - group - }, - )); + } + } - groups + let mut staged_selected: Vec> = staged_groups + .into_iter() + .filter(|group| cluster_selected(group.xtream_cluster, clusters)) + .map(|group| { + Some(prepare_staged_xtream_group( + group, + provider, + &provider_urls, + username, + password, + live_stream_use_prefix, + live_stream_without_extension, + )) + }) + .collect(); + + // Category ids of the provider playlist are authoritative and stay reserved across the overlay: + // persistence keys input groups by `(cluster, id)`, so two groups of one cluster sharing an id + // would be collapsed into a single category. Only staged ids are renumbered; a provider playlist + // that already repeats a category id is passed through unchanged. + let mut used_ids = GroupCategoryIds::from_groups(&provider_groups); + let assignment = match_staged_groups(&provider_groups, &staged_selected, staged_type, clusters); + + let mut result: Vec = Vec::with_capacity(provider_groups.len() + staged_selected.len()); + + for (provider_idx, provider_group) in provider_groups.into_iter().enumerate() { + if !cluster_selected(provider_group.xtream_cluster, clusters) { + result.push(provider_group); + continue; + } + + let matched = assignment.staged_of(provider_idx).and_then(|staged_idx| staged_selected[staged_idx].take()); + let Some(mut staged) = matched.filter(|staged| !staged.channels.is_empty()) else { + // A staged group whose channels carried no numeric stream id keeps the provider category, + // so the overlaid cluster is never emptied by unusable overlay rows. + result.push(provider_group); + continue; + }; + + // The provider category id is the identity of the overlaid category: a staged group matched + // by stream ids or title takes it over instead of keeping its own numbering. + staged.id = + if provider_group.id == 0 { used_ids.allocate(provider_group.xtream_cluster) } else { provider_group.id }; + sync_channel_category_id(&mut staged); + result.push(staged); + } + + // Append newly introduced staged groups that had no corresponding provider group. + for mut staged in staged_selected.into_iter().flatten().filter(|staged| !staged.channels.is_empty()) { + if staged.id == 0 { + // A staged group without a category id introduces a new category, so it only needs a free id. + staged.id = used_ids.allocate(staged.xtream_cluster); + } else if !used_ids.claim(staged.xtream_cluster, staged.id) { + let allocated = used_ids.allocate(staged.xtream_cluster); + warn!( + "Staged group '{}' uses category id {} which is already taken in {}; renumbered to {allocated}", + staged.title, staged.id, staged.xtream_cluster + ); + staged.id = allocated; + } + sync_channel_category_id(&mut staged); + result.push(staged); + } + + result +} + +/// Keeps the channels of a group consistent with its final category id. +/// +/// Persistence re-syncs this through [`PlaylistGroup::on_load`], but the overlay result is consumed as +/// a whole: a reader must not see the category id of the staged source on a group that was just given +/// the provider category id. +fn sync_channel_category_id(group: &mut PlaylistGroup) { + let category_id = group.id; + for item in &mut group.channels { + item.header.category_id = category_id; + } +} + +/// Staged groups resolved against the provider groups they overlay. +/// +/// Both directions are kept in sync, so one staged group can never overlay two provider categories +/// and one provider category can never receive two staged groups. +struct StagedOverlayAssignment { + staged_by_provider: Vec>, + provider_by_staged: Vec>, + has_stream_id_evidence: Vec, +} + +impl StagedOverlayAssignment { + fn new(provider_len: usize, staged_len: usize) -> Self { + Self { + staged_by_provider: vec![None; provider_len], + provider_by_staged: vec![None; staged_len], + has_stream_id_evidence: vec![false; staged_len], + } + } + + /// Assigns `staged_idx` to `provider_idx`; returns `false` when either side is already taken. + fn assign(&mut self, provider_idx: usize, staged_idx: usize) -> bool { + if self.staged_by_provider[provider_idx].is_some() || self.provider_by_staged[staged_idx].is_some() { + return false; + } + self.staged_by_provider[provider_idx] = Some(staged_idx); + self.provider_by_staged[staged_idx] = Some(provider_idx); + true + } + + fn staged_of(&self, provider_idx: usize) -> Option { self.staged_by_provider[provider_idx] } + + fn is_assigned(&self, staged_idx: usize) -> bool { self.provider_by_staged[staged_idx].is_some() } +} + +/// Resolves every staged group against the provider groups it overlays. +/// +/// The staged playlist is the provider playlist restructured by an external tool: group and channel +/// names may be rewritten, the ids do not. The passes therefore rank the id evidence above the name: +/// +/// 1. the channel stream ids decide. A staged channel carries the provider stream id, and the provider +/// playlist itself states which category owns a stream, so this also covers staged sources whose +/// group ids are not provider category ids (a `m3u` staged playlist numbers its groups positionally). +/// 2. the category id decides for Xtream staged groups whose channels the provider playlist does not know. +/// M3U staged group ids are positional and cannot identify provider categories. +/// 3. the title remains the last resort for staged groups without provider stream-id evidence. +fn match_staged_groups( + provider_groups: &[PlaylistGroup], + staged_groups: &[Option], + staged_type: StagedInputType, + clusters: ClusterFlags, +) -> StagedOverlayAssignment { + let mut assignment = StagedOverlayAssignment::new(provider_groups.len(), staged_groups.len()); + match_staged_by_stream_id(provider_groups, staged_groups, clusters, &mut assignment); + if staged_type == StagedInputType::Xtream { + match_staged_by_category_id(provider_groups, staged_groups, clusters, &mut assignment); + } + match_staged_by_title(provider_groups, staged_groups, clusters, &mut assignment); + assignment +} + +/// Pass 1: a staged group overlays the provider category owning its streams, strongest overlap first. +fn match_staged_by_stream_id( + provider_groups: &[PlaylistGroup], + staged_groups: &[Option], + clusters: ClusterFlags, + assignment: &mut StagedOverlayAssignment, +) { + // Only the staged stream ids are indexed, so the memory cost is bounded by the overlay and not + // by the size of the provider playlist. + let staged_stream_ids: HashSet<(XtreamCluster, &str)> = staged_groups + .iter() + .flatten() + .flat_map(|staged| staged.channels.iter().map(|item| (staged.xtream_cluster, item.header.id.as_ref()))) + .filter(|(_, stream_id)| !stream_id.is_empty()) + .collect(); + if staged_stream_ids.is_empty() { + return; + } + + let mut provider_idx_by_stream_id: HashMap<(XtreamCluster, &str), usize> = HashMap::new(); + for (provider_idx, provider_group) in provider_groups.iter().enumerate() { + if !cluster_selected(provider_group.xtream_cluster, clusters) { + continue; + } + for item in &provider_group.channels { + let key = (provider_group.xtream_cluster, item.header.id.as_ref()); + if staged_stream_ids.contains(&key) { + provider_idx_by_stream_id.entry(key).or_insert(provider_idx); + } + } + } + if provider_idx_by_stream_id.is_empty() { + return; + } + + // Every staged group is offered all provider categories owning its streams, ranked by overlap: when + // a stronger overlap takes the best-matching category, the next candidate can still claim the group. + let mut candidates: Vec = Vec::new(); + let mut hits_by_provider: HashMap = HashMap::new(); + for (staged_idx, staged) in staged_groups.iter().enumerate() { + let Some(staged) = staged else { + continue; + }; + hits_by_provider.clear(); + for item in &staged.channels { + if let Some(provider_idx) = provider_idx_by_stream_id.get(&(staged.xtream_cluster, item.header.id.as_ref())) + { + *hits_by_provider.entry(*provider_idx).or_insert(0) += 1; + } + } + assignment.has_stream_id_evidence[staged_idx] = !hits_by_provider.is_empty(); + candidates.extend(hits_by_provider.iter().map(|(provider_idx, hits)| StreamOverlap { + hits: *hits, + staged_idx, + provider_idx: *provider_idx, + })); + } + + // The strongest overlap wins first, so a staged group never loses its category to a weaker candidate. + candidates + .sort_unstable_by_key(|overlap| (std::cmp::Reverse(overlap.hits), overlap.staged_idx, overlap.provider_idx)); + for overlap in candidates { + assignment.assign(overlap.provider_idx, overlap.staged_idx); + } +} + +/// One provider category a staged group shares streams with. +struct StreamOverlap { + hits: usize, + staged_idx: usize, + provider_idx: usize, +} + +/// Pass 2: a staged group overlays the provider category carrying its id. +fn match_staged_by_category_id( + provider_groups: &[PlaylistGroup], + staged_groups: &[Option], + clusters: ClusterFlags, + assignment: &mut StagedOverlayAssignment, +) { + let mut provider_idx_by_id: HashMap<(XtreamCluster, u32), usize> = HashMap::new(); + for (provider_idx, provider_group) in provider_groups.iter().enumerate() { + if provider_group.id != 0 && cluster_selected(provider_group.xtream_cluster, clusters) { + provider_idx_by_id.entry((provider_group.xtream_cluster, provider_group.id)).or_insert(provider_idx); + } + } + + for (staged_idx, staged) in staged_groups.iter().enumerate() { + let Some(staged) = staged else { + continue; + }; + if staged.id == 0 || assignment.is_assigned(staged_idx) || assignment.has_stream_id_evidence[staged_idx] { + continue; + } + if let Some(provider_idx) = provider_idx_by_id.get(&(staged.xtream_cluster, staged.id)).copied() { + assignment.assign(provider_idx, staged_idx); + } + } +} + +/// Pass 3: a staged group overlays the provider category of the same name. +fn match_staged_by_title( + provider_groups: &[PlaylistGroup], + staged_groups: &[Option], + clusters: ClusterFlags, + assignment: &mut StagedOverlayAssignment, +) { + // Staged groups of the same name are interchangeable, so one lookup per provider category is enough. + let mut staged_by_title: HashMap<(XtreamCluster, &str), Vec> = HashMap::new(); + for (staged_idx, staged) in staged_groups.iter().enumerate() { + if assignment.is_assigned(staged_idx) || assignment.has_stream_id_evidence[staged_idx] { + continue; + } + if let Some(staged) = staged { + staged_by_title.entry((staged.xtream_cluster, staged.title.as_ref())).or_default().push(staged_idx); + } + } + + for (provider_idx, provider_group) in provider_groups.iter().enumerate() { + if assignment.staged_of(provider_idx).is_some() || !cluster_selected(provider_group.xtream_cluster, clusters) { + continue; + } + let key = (provider_group.xtream_cluster, provider_group.title.as_ref()); + let Some(free_groups) = staged_by_title.get_mut(&key) else { + continue; + }; + if !free_groups.is_empty() { + // Buckets stay tiny, so taking the first free group keeps the earlier first-match semantics. + let staged_idx = free_groups.remove(0); + assignment.assign(provider_idx, staged_idx); + } + } +} + +/// Cluster-scoped category ids of a merged playlist. +/// +/// Provider category ids are reserved up front because they are authoritative: an overlaid group +/// keeps them, and a staged group that introduces a new category must not take them over. +#[derive(Default)] +struct GroupCategoryIds { + used: HashMap>, + next_free: HashMap, +} + +impl GroupCategoryIds { + fn from_groups(groups: &[PlaylistGroup]) -> Self { + let mut ids = Self::default(); + for group in groups { + if group.id != 0 { + ids.used.entry(group.xtream_cluster).or_default().insert(group.id); + } + } + for (cluster, used) in &ids.used { + ids.next_free.insert(*cluster, used.iter().copied().max().unwrap_or(0).saturating_add(1)); + } + ids + } + + /// Claims `id` for `cluster`; a zero id or an id already in use is not claimable. + fn claim(&mut self, cluster: XtreamCluster, id: u32) -> bool { + id != 0 && self.used.entry(cluster).or_default().insert(id) + } + + /// Reserves and returns the next free id of `cluster`. + fn allocate(&mut self, cluster: XtreamCluster) -> u32 { + let used = self.used.entry(cluster).or_default(); + let mut candidate = self.next_free.get(&cluster).copied().unwrap_or(1); + // Zero is never a valid category id. The loop wraps at the u32 edge and ends at the latest + // when it reaches the first free id, which the finite id set always contains. + while candidate == 0 || used.contains(&candidate) { + candidate = candidate.checked_add(1).unwrap_or(1); + } + used.insert(candidate); + self.next_free.insert(cluster, candidate.saturating_add(1)); + candidate + } } pub(crate) fn should_apply_staged_overlay(download_result: &PlaylistDownloadResult) -> bool { @@ -1275,7 +1682,8 @@ pub(crate) async fn download_input, + staged_groups: Vec, +) -> Vec { + super::apply_staged_overlay_groups(provider, StagedInputType::Xtream, clusters, provider_groups, staged_groups) +} + #[derive(Clone, Default)] struct PlaylistRunCollectSink(Arc>>); @@ -682,22 +691,36 @@ fn test_group(cluster: XtreamCluster, item_name: &str, input_name: &str) -> Play #[test] fn pipeline_transparency_staged_overlay_replaces_groups_and_neutralizes_overlaid_runtime_facts() { - let provider_name = "provider".intern(); + let provider = ConfigInput { + name: "provider".intern(), + input_type: InputType::Xtream, + url: "http://provider.example".to_string(), + username: Some("real-user".to_string()), + password: Some("real-pass".to_string()), + options: Some(ConfigInputOptions::defaults().clone()), + ..Default::default() + }; let provider_groups = vec![ test_group(XtreamCluster::Live, "provider-live", "provider"), test_group(XtreamCluster::Video, "provider-vod", "provider"), ]; - let staged_groups = vec![ - test_group(XtreamCluster::Live, "staged-live", "staged"), - test_group(XtreamCluster::Series, "staged-series", "staged"), - ]; + let mut staged_live = test_group(XtreamCluster::Live, "staged-live", "staged"); + staged_live.channels[0].header.id = "11203".intern(); + staged_live.channels[0].header.url = "http://iptvhost.example/live/fake-user/fake-pass/11203.ts".intern(); + let mut invalid = test_group(XtreamCluster::Live, "invalid", "staged").channels.remove(0); + invalid.header.id = "invalid".intern(); + invalid.header.url = "http://iptvhost.example/live/fake-user/fake-pass/invalid.ts".intern(); + staged_live.channels.push(invalid); + let staged_groups = vec![staged_live, test_group(XtreamCluster::Series, "staged-series", "staged")]; - let groups = apply_staged_overlay_groups(&provider_name, ClusterFlags::Live, provider_groups, staged_groups); + let groups = apply_staged_overlay_groups(&provider, ClusterFlags::Live, provider_groups, staged_groups); assert_eq!(groups.len(), 2); - assert_eq!(groups[0].title.as_ref(), "provider-vod"); + assert_eq!(groups[0].title.as_ref(), "staged-live"); + assert_eq!(groups[0].channels.len(), 1); assert_eq!(groups[0].channels[0].header.input_name.as_ref(), "provider"); - assert_eq!(groups[1].title.as_ref(), "staged-live"); + assert_eq!(groups[0].channels[0].header.url.as_ref(), "http://provider.example/live/real-user/real-pass/11203.ts"); + assert_eq!(groups[1].title.as_ref(), "provider-vod"); assert_eq!(groups[1].channels[0].header.input_name.as_ref(), "provider"); let mut telemetry = PlaylistUpdateInputTelemetry { @@ -754,6 +777,796 @@ fn pipeline_transparency_staged_overlay_replaces_groups_and_neutralizes_overlaid assert_eq!(video.decision, Some(PlaylistUpdateClusterDecision::Accepted)); } +#[test] +fn staged_xtream_vod_url_uses_provider_credentials_and_one_extension() { + let provider = ConfigInput { + name: "provider".intern(), + input_type: InputType::Xtream, + url: "http://provider.example".to_string(), + username: Some("real-user".to_string()), + password: Some("real-pass".to_string()), + ..Default::default() + }; + let mut staged_vod = test_group(XtreamCluster::Video, "staged-vod", "staged"); + staged_vod.channels[0].header.id = "310".intern(); + staged_vod.channels[0].header.url = "http://iptvhost.example/movie/fake-user/fake-pass/310.mkv".intern(); + + let groups = apply_staged_overlay_groups(&provider, ClusterFlags::Vod, Vec::new(), vec![staged_vod]); + + assert_eq!(groups[0].channels[0].header.url.as_ref(), "http://provider.example/movie/real-user/real-pass/310.mkv"); +} + +#[test] +fn staged_xtream_overlay_keeps_known_provider_stream_urls() { + let provider = ConfigInput { + name: "provider".intern(), + input_type: InputType::Xtream, + url: "http://provider.example".to_string(), + username: Some("real-user".to_string()), + password: Some("real-pass".to_string()), + ..Default::default() + }; + let mut provider_vod = test_group(XtreamCluster::Video, "Movies", "provider"); + provider_vod.channels[0].header.id = "310".intern(); + provider_vod.channels[0].header.url = "https://cdn.example/video?id=310".intern(); + let mut second_provider = provider_vod.channels[0].clone(); + second_provider.header.id = "311".intern(); + second_provider.header.url = "http://provider.example/movie/real-user/real-pass/311.mp4".intern(); + provider_vod.channels.push(second_provider); + + let mut staged_vod = test_group(XtreamCluster::Video, "Edited Movies", "staged"); + staged_vod.channels[0].header.id = "310".intern(); + staged_vod.channels[0].header.url = "http://editor.example/movie/310.mkv".intern(); + let mut second_staged = staged_vod.channels[0].clone(); + second_staged.header.id = "311".intern(); + second_staged.header.url = "http://editor.example/movie/311.mkv".intern(); + staged_vod.channels.push(second_staged); + + let groups = apply_staged_overlay_groups(&provider, ClusterFlags::Vod, vec![provider_vod], vec![staged_vod]); + + assert_eq!(groups[0].title.as_ref(), "Edited Movies"); + assert_eq!(groups[0].channels[0].header.url.as_ref(), "https://cdn.example/video?id=310"); + assert_eq!(groups[0].channels[1].header.url.as_ref(), "http://provider.example/movie/real-user/real-pass/311.mp4"); +} + +#[test] +fn staged_xtream_new_vod_stream_uses_metadata_extension() { + let provider = ConfigInput { + name: "provider".intern(), + input_type: InputType::Xtream, + url: "http://provider.example".to_string(), + username: Some("real-user".to_string()), + password: Some("real-pass".to_string()), + ..Default::default() + }; + let mut staged_vod = test_group(XtreamCluster::Video, "New Movies", "staged"); + staged_vod.channels[0].header.id = "310".intern(); + staged_vod.channels[0].header.url = "http://editor.example/movie/310".intern(); + staged_vod.channels[0].header.additional_properties = + Some(StreamProperties::Video(Box::new(shared::model::VideoStreamProperties { + container_extension: "mkv".intern(), + ..Default::default() + }))); + + let groups = apply_staged_overlay_groups(&provider, ClusterFlags::Vod, Vec::new(), vec![staged_vod]); + + assert_eq!(groups[0].channels[0].header.url.as_ref(), "http://provider.example/movie/real-user/real-pass/310.mkv"); +} + +#[test] +fn staged_xtream_live_url_respects_prefix_and_without_extension_flags() { + let provider_prefix = ConfigInput { + name: "provider".intern(), + input_type: InputType::Xtream, + url: "http://provider.example".to_string(), + username: Some("real-user".to_string()), + password: Some("real-pass".to_string()), + options: Some(ConfigInputOptions { + flags: ConfigInputFlags::XtreamLiveStreamUsePrefix.into(), + ..ConfigInputOptions::defaults().clone() + }), + ..Default::default() + }; + let mut staged_live_prefix = test_group(XtreamCluster::Live, "staged-live", "staged"); + staged_live_prefix.channels[0].header.id = "11203".intern(); + staged_live_prefix.channels[0].header.url = "http://iptvhost.example/fake-user/fake-pass/11203.ts".intern(); + + let groups_prefix = + apply_staged_overlay_groups(&provider_prefix, ClusterFlags::Live, Vec::new(), vec![staged_live_prefix]); + + assert_eq!( + groups_prefix[0].channels[0].header.url.as_ref(), + "http://provider.example/live/real-user/real-pass/11203.ts" + ); + + let provider_no_ext = ConfigInput { + name: "provider".intern(), + input_type: InputType::Xtream, + url: "http://provider.example".to_string(), + username: Some("real-user".to_string()), + password: Some("real-pass".to_string()), + options: Some(ConfigInputOptions { + flags: ConfigInputFlags::XtreamLiveStreamWithoutExtension.into(), + ..ConfigInputOptions::defaults().clone() + }), + ..Default::default() + }; + let mut staged_live_no_ext = test_group(XtreamCluster::Live, "staged-live", "staged"); + staged_live_no_ext.channels[0].header.id = "11203".intern(); + staged_live_no_ext.channels[0].header.url = "http://iptvhost.example/fake-user/fake-pass/11203.ts".intern(); + + let groups_no_ext = + apply_staged_overlay_groups(&provider_no_ext, ClusterFlags::Live, Vec::new(), vec![staged_live_no_ext]); + + assert_eq!(groups_no_ext[0].channels[0].header.url.as_ref(), "http://provider.example/real-user/real-pass/11203"); +} + +#[test] +fn staged_xtream_overlay_recomputes_uuid_on_group_load() { + let provider = ConfigInput { + name: "provider".intern(), + input_type: InputType::Xtream, + url: "http://provider.example".to_string(), + username: Some("real-user".to_string()), + password: Some("real-pass".to_string()), + options: Some(ConfigInputOptions { + flags: ConfigInputFlags::XtreamLiveStreamUsePrefix.into(), + ..ConfigInputOptions::defaults().clone() + }), + ..Default::default() + }; + let mut staged_live = test_group(XtreamCluster::Live, "staged-live", "staged"); + staged_live.channels[0].header.id = "11203".intern(); + staged_live.channels[0].header.freeze_input_stream_id(); + staged_live.channels[0].header.url = "http://iptvhost.example/live/fake-user/fake-pass/11203.ts".intern(); + staged_live.channels[0].header.gen_uuid(); + let old_uuid = staged_live.channels[0].header.get_uuid().clone(); + + let mut groups = apply_staged_overlay_groups(&provider, ClusterFlags::Live, Vec::new(), vec![staged_live]); + + assert_eq!(groups[0].channels[0].header.input_name.as_ref(), "provider"); + let reconstructed_url = "http://provider.example/live/real-user/real-pass/11203.ts"; + assert_eq!(groups[0].channels[0].header.url.as_ref(), reconstructed_url); + assert_eq!(groups[0].channels[0].header.get_input_stream_id().as_deref(), Some("11203")); + + groups[0].on_load(); + + let expected_uuid = shared::utils::generate_runtime_playlist_uuid( + "provider", + "11203", + shared::model::PlaylistItemType::Live, + reconstructed_url, + ); + assert_ne!(groups[0].channels[0].header.get_uuid(), &old_uuid); + assert_eq!(groups[0].channels[0].header.get_uuid(), &expected_uuid); + assert_eq!(groups[0].channels[0].header.get_input_stream_id().as_deref(), Some("11203")); +} + +#[test] +fn staged_xtream_overlay_full_cluster_staging_replaces_all_matching_groups() { + let provider = ConfigInput { + name: "provider".intern(), + input_type: InputType::Xtream, + url: "http://provider.example".to_string(), + username: Some("real-user".to_string()), + password: Some("real-pass".to_string()), + ..Default::default() + }; + let mut provider_action = test_group(XtreamCluster::Video, "Action", "provider"); + provider_action.channels[0].header.id = "10".intern(); + let mut provider_comedy = test_group(XtreamCluster::Video, "Comedy", "provider"); + provider_comedy.channels[0].header.id = "20".intern(); + + let mut staged_action = test_group(XtreamCluster::Video, "Action", "staged"); + staged_action.channels[0].header.id = "11".intern(); + let mut staged_comedy = test_group(XtreamCluster::Video, "Comedy", "staged"); + staged_comedy.channels[0].header.id = "21".intern(); + + let groups = apply_staged_overlay_groups( + &provider, + ClusterFlags::Vod, + vec![provider_action, provider_comedy], + vec![staged_action, staged_comedy], + ); + + assert_eq!(groups.len(), 2); + assert_eq!(groups[0].title.as_ref(), "Action"); + assert_eq!(groups[0].channels[0].header.id.as_ref(), "11"); + assert_eq!(groups[0].channels[0].header.input_name.as_ref(), "provider"); + assert_eq!(groups[1].title.as_ref(), "Comedy"); + assert_eq!(groups[1].channels[0].header.id.as_ref(), "21"); + assert_eq!(groups[1].channels[0].header.input_name.as_ref(), "provider"); +} + +#[test] +fn staged_xtream_overlay_partial_cluster_staging_preserves_unstaged_provider_groups() { + let provider = ConfigInput { + name: "provider".intern(), + input_type: InputType::Xtream, + url: "http://provider.example".to_string(), + username: Some("real-user".to_string()), + password: Some("real-pass".to_string()), + ..Default::default() + }; + let mut provider_action = test_group(XtreamCluster::Video, "Action", "provider"); + provider_action.channels[0].header.id = "10".intern(); + let mut provider_comedy = test_group(XtreamCluster::Video, "Comedy", "provider"); + provider_comedy.channels[0].header.id = "20".intern(); + let mut provider_drama = test_group(XtreamCluster::Video, "Drama", "provider"); + provider_drama.channels[0].header.id = "30".intern(); + + let mut staged_action = test_group(XtreamCluster::Video, "Action", "staged"); + staged_action.channels[0].header.id = "11".intern(); + + let groups = apply_staged_overlay_groups( + &provider, + ClusterFlags::Vod, + vec![provider_action, provider_comedy, provider_drama], + vec![staged_action], + ); + + assert_eq!(groups.len(), 3); + assert_eq!(groups[0].title.as_ref(), "Action"); + assert_eq!(groups[0].channels[0].header.id.as_ref(), "11"); + assert_eq!(groups[1].title.as_ref(), "Comedy"); + assert_eq!(groups[1].channels[0].header.id.as_ref(), "20"); + assert_eq!(groups[2].title.as_ref(), "Drama"); + assert_eq!(groups[2].channels[0].header.id.as_ref(), "30"); +} + +#[test] +fn staged_xtream_overlay_no_staged_groups_for_selected_cluster_preserves_all_provider_groups() { + let provider = ConfigInput { + name: "provider".intern(), + input_type: InputType::Xtream, + url: "http://provider.example".to_string(), + username: Some("real-user".to_string()), + password: Some("real-pass".to_string()), + ..Default::default() + }; + let provider_news = test_group(XtreamCluster::Live, "News", "provider"); + let provider_sports = test_group(XtreamCluster::Live, "Sports", "provider"); + let staged_series = test_group(XtreamCluster::Series, "Shows", "staged"); + + let groups = apply_staged_overlay_groups( + &provider, + ClusterFlags::Live, + vec![provider_news, provider_sports], + vec![staged_series], + ); + + assert_eq!(groups.len(), 2); + assert_eq!(groups[0].title.as_ref(), "News"); + assert_eq!(groups[1].title.as_ref(), "Sports"); +} + +#[test] +fn staged_xtream_overlay_all_invalid_ids_falls_back_to_provider_group() { + let provider = ConfigInput { + name: "provider".intern(), + input_type: InputType::Xtream, + url: "http://provider.example".to_string(), + username: Some("real-user".to_string()), + password: Some("real-pass".to_string()), + ..Default::default() + }; + let mut provider_sports = test_group(XtreamCluster::Live, "Sports", "provider"); + provider_sports.channels[0].header.id = "501".intern(); + + let mut staged_sports = test_group(XtreamCluster::Live, "Sports", "staged"); + staged_sports.channels[0].header.id = "non-numeric-a".intern(); + let mut second_invalid = staged_sports.channels[0].clone(); + second_invalid.header.id = "non-numeric-b".intern(); + staged_sports.channels.push(second_invalid); + + let groups = apply_staged_overlay_groups(&provider, ClusterFlags::Live, vec![provider_sports], vec![staged_sports]); + + assert_eq!(groups.len(), 1); + assert_eq!(groups[0].title.as_ref(), "Sports"); + assert_eq!(groups[0].channels.len(), 1); + assert_eq!(groups[0].channels[0].header.id.as_ref(), "501"); +} + +#[test] +fn staged_xtream_overlay_partially_valid_ids_keeps_valid_and_drops_invalid() { + let provider = ConfigInput { + name: "provider".intern(), + input_type: InputType::Xtream, + url: "http://provider.example".to_string(), + username: Some("real-user".to_string()), + password: Some("real-pass".to_string()), + ..Default::default() + }; + let mut provider_sports = test_group(XtreamCluster::Live, "Sports", "provider"); + provider_sports.channels[0].header.id = "501".intern(); + + let mut staged_sports = test_group(XtreamCluster::Live, "Sports", "staged"); + staged_sports.channels[0].header.id = "502".intern(); + let mut invalid = staged_sports.channels[0].clone(); + invalid.header.id = "invalid-stream".intern(); + staged_sports.channels.push(invalid); + + let groups = apply_staged_overlay_groups(&provider, ClusterFlags::Live, vec![provider_sports], vec![staged_sports]); + + assert_eq!(groups.len(), 1); + assert_eq!(groups[0].title.as_ref(), "Sports"); + assert_eq!(groups[0].channels.len(), 1); + assert_eq!(groups[0].channels[0].header.id.as_ref(), "502"); +} + +#[test] +fn staged_xtream_overlay_missing_credentials_preserves_all_provider_groups() { + let provider = ConfigInput { + name: "provider".intern(), + input_type: InputType::Xtream, + url: "http://provider.example".to_string(), + username: None, + password: None, + ..Default::default() + }; + let provider_news = test_group(XtreamCluster::Live, "News", "provider"); + let provider_sports = test_group(XtreamCluster::Live, "Sports", "provider"); + let mut staged_sports = test_group(XtreamCluster::Live, "Sports", "staged"); + staged_sports.channels[0].header.id = "501".intern(); + + let groups = apply_staged_overlay_groups( + &provider, + ClusterFlags::Live, + vec![provider_news, provider_sports], + vec![staged_sports], + ); + + assert_eq!(groups.len(), 2); + assert_eq!(groups[0].title.as_ref(), "News"); + assert_eq!(groups[1].title.as_ref(), "Sports"); +} + +#[test] +fn staged_xtream_overlay_multiple_groups_with_different_clusters() { + let provider = ConfigInput { + name: "provider".intern(), + input_type: InputType::Xtream, + url: "http://provider.example".to_string(), + username: Some("real-user".to_string()), + password: Some("real-pass".to_string()), + ..Default::default() + }; + let mut provider_live = test_group(XtreamCluster::Live, "News", "provider"); + provider_live.channels[0].header.id = "100".intern(); + let mut provider_action = test_group(XtreamCluster::Video, "Action", "provider"); + provider_action.channels[0].header.id = "200".intern(); + let mut provider_comedy = test_group(XtreamCluster::Video, "Comedy", "provider"); + provider_comedy.channels[0].header.id = "300".intern(); + let mut provider_series = test_group(XtreamCluster::Series, "Shows", "provider"); + provider_series.channels[0].header.id = "400".intern(); + + let mut staged_live = test_group(XtreamCluster::Live, "News", "staged"); + staged_live.channels[0].header.id = "101".intern(); + let mut staged_action = test_group(XtreamCluster::Video, "Action", "staged"); + staged_action.channels[0].header.id = "201".intern(); + + let groups = apply_staged_overlay_groups( + &provider, + ClusterFlags::Live | ClusterFlags::Vod, + vec![provider_live, provider_action, provider_comedy, provider_series], + vec![staged_live, staged_action], + ); + + assert_eq!(groups.len(), 4); + assert_eq!(groups[0].title.as_ref(), "News"); + assert_eq!(groups[0].channels[0].header.id.as_ref(), "101"); + assert_eq!(groups[1].title.as_ref(), "Action"); + assert_eq!(groups[1].channels[0].header.id.as_ref(), "201"); + assert_eq!(groups[2].title.as_ref(), "Comedy"); + assert_eq!(groups[2].channels[0].header.id.as_ref(), "300"); + assert_eq!(groups[3].title.as_ref(), "Shows"); + assert_eq!(groups[3].channels[0].header.id.as_ref(), "400"); +} + +#[test] +fn staged_xtream_overlay_preserves_original_group_order() { + let provider = ConfigInput { + name: "provider".intern(), + input_type: InputType::Xtream, + url: "http://provider.example".to_string(), + username: Some("real-user".to_string()), + password: Some("real-pass".to_string()), + ..Default::default() + }; + let mut live_1 = test_group(XtreamCluster::Live, "Live1", "provider"); + live_1.channels[0].header.id = "1".intern(); + let mut vod_1 = test_group(XtreamCluster::Video, "Vod1", "provider"); + vod_1.channels[0].header.id = "2".intern(); + let mut live_2 = test_group(XtreamCluster::Live, "Live2", "provider"); + live_2.channels[0].header.id = "3".intern(); + let mut vod_2 = test_group(XtreamCluster::Video, "Vod2", "provider"); + vod_2.channels[0].header.id = "4".intern(); + + let mut staged_live_1 = test_group(XtreamCluster::Live, "Live1", "staged"); + staged_live_1.channels[0].header.id = "11".intern(); + let mut staged_live_2 = test_group(XtreamCluster::Live, "Live2", "staged"); + staged_live_2.channels[0].header.id = "33".intern(); + + let groups = apply_staged_overlay_groups( + &provider, + ClusterFlags::Live, + vec![live_1, vod_1, live_2, vod_2], + vec![staged_live_1, staged_live_2], + ); + + assert_eq!(groups.len(), 4); + assert_eq!(groups[0].title.as_ref(), "Live1"); + assert_eq!(groups[0].channels[0].header.id.as_ref(), "11"); + assert_eq!(groups[1].title.as_ref(), "Vod1"); + assert_eq!(groups[1].channels[0].header.id.as_ref(), "2"); + assert_eq!(groups[2].title.as_ref(), "Live2"); + assert_eq!(groups[2].channels[0].header.id.as_ref(), "33"); + assert_eq!(groups[3].title.as_ref(), "Vod2"); + assert_eq!(groups[3].channels[0].header.id.as_ref(), "4"); +} + +/// The merged playlist reaches persistence keyed by `(cluster, id)`, which collapses groups that +/// share an id within one cluster. +fn assert_unique_category_ids(groups: &[PlaylistGroup]) { + let mut seen = std::collections::HashSet::new(); + for group in groups { + assert!( + seen.insert((group.xtream_cluster, group.id)), + "category id {} is used by more than one {} group", + group.id, + group.xtream_cluster + ); + } +} + +#[test] +fn staged_overlay_syncs_channel_category_ids_with_the_final_group_id() { + let provider = ConfigInput { + name: "provider".intern(), + input_type: InputType::Xtream, + url: "http://provider.example".to_string(), + username: Some("real-user".to_string()), + password: Some("real-pass".to_string()), + ..Default::default() + }; + let mut provider_news = test_group(XtreamCluster::Live, "News", "provider"); + provider_news.id = 5; + provider_news.channels[0].header.id = "100".intern(); + + // Matched by stream id, so the group takes over the provider category id. + let mut staged_rewritten = test_group(XtreamCluster::Live, "Rewritten", "staged"); + staged_rewritten.id = 7; + staged_rewritten.channels[0].header.id = "100".intern(); + staged_rewritten.channels[0].header.category_id = 7; + // No provider counterpart: the group becomes a new category with an allocated id. + let mut staged_kids = test_group(XtreamCluster::Live, "Kids", "staged"); + staged_kids.id = 0; + staged_kids.channels[0].header.id = "300".intern(); + + let groups = apply_staged_overlay_groups( + &provider, + ClusterFlags::Live, + vec![provider_news], + vec![staged_rewritten, staged_kids], + ); + + assert_eq!(groups.len(), 2); + assert_eq!(groups[0].id, 5); + assert_eq!(groups[0].channels[0].header.category_id, 5); + assert_eq!(groups[1].id, 6); + assert_eq!(groups[1].channels[0].header.category_id, 6); + assert_unique_category_ids(&groups); +} + +#[test] +fn staged_overlay_on_non_xtream_provider_replaces_cluster_and_rewrites_input_name() { + let provider = ConfigInput { + name: "provider".intern(), + input_type: InputType::M3u, + url: "http://provider.example/list.m3u".to_string(), + ..Default::default() + }; + let provider_live = test_group(XtreamCluster::Live, "provider-live", "provider"); + let provider_vod = test_group(XtreamCluster::Video, "provider-vod", "provider"); + let mut staged_live = test_group(XtreamCluster::Live, "staged-live", "staged"); + staged_live.channels[0].header.url = "http://editor.example/live/1.ts".intern(); + + let groups = apply_staged_overlay_groups( + &provider, + ClusterFlags::Live, + vec![provider_live, provider_vod], + vec![staged_live], + ); + + assert_eq!(groups.len(), 2); + assert_eq!(groups[0].title.as_ref(), "provider-vod"); + assert_eq!(groups[0].channels[0].header.input_name.as_ref(), "provider"); + assert_eq!(groups[1].title.as_ref(), "staged-live"); + // A non-Xtream provider cannot rebuild Xtream stream URLs, so the staged URL and its group id stay. + assert_eq!(groups[1].channels[0].header.input_name.as_ref(), "provider"); + assert_eq!(groups[1].channels[0].header.url.as_ref(), "http://editor.example/live/1.ts"); + assert_eq!(groups[1].id, 1); +} + +#[test] +fn staged_xtream_overlay_matches_by_category_id_not_by_title() { + let provider = ConfigInput { + name: "provider".intern(), + input_type: InputType::Xtream, + url: "http://provider.example".to_string(), + username: Some("real-user".to_string()), + password: Some("real-pass".to_string()), + ..Default::default() + }; + // The staged tool renamed the provider category 7 and moved its streams; the title now collides + // with the unrelated provider category 5. + let mut provider_kids = test_group(XtreamCluster::Live, "Kids", "provider"); + provider_kids.id = 5; + provider_kids.channels[0].header.id = "105".intern(); + let mut provider_news = test_group(XtreamCluster::Live, "News", "provider"); + provider_news.id = 7; + provider_news.channels[0].header.id = "107".intern(); + + let mut staged_renamed = test_group(XtreamCluster::Live, "Kids", "staged"); + staged_renamed.id = 7; + staged_renamed.channels[0].header.id = "207".intern(); + + let groups = apply_staged_overlay_groups( + &provider, + ClusterFlags::Live, + vec![provider_kids, provider_news], + vec![staged_renamed], + ); + + assert_eq!(groups.len(), 2); + // Category 5 keeps its content: the staged group belongs to category 7. + assert_eq!(groups[0].title.as_ref(), "Kids"); + assert_eq!(groups[0].id, 5); + assert_eq!(groups[0].channels[0].header.id.as_ref(), "105"); + // Category 7 carries the staged rename and the staged stream, under its own id. + assert_eq!(groups[1].title.as_ref(), "Kids"); + assert_eq!(groups[1].id, 7); + assert_eq!(groups[1].channels[0].header.id.as_ref(), "207"); + assert_unique_category_ids(&groups); +} + +#[test] +fn staged_xtream_overlay_matches_by_stream_id_over_group_id() { + let provider = ConfigInput { + name: "provider".intern(), + input_type: InputType::Xtream, + url: "http://provider.example".to_string(), + username: Some("real-user".to_string()), + password: Some("real-pass".to_string()), + ..Default::default() + }; + let mut provider_news = test_group(XtreamCluster::Live, "News", "provider"); + provider_news.id = 5; + provider_news.channels[0].header.id = "100".intern(); + let mut provider_sports = test_group(XtreamCluster::Live, "Sports", "provider"); + provider_sports.id = 7; + provider_sports.channels[0].header.id = "200".intern(); + + // A m3u staged playlist numbers its groups on its own, so the group id points at another + // provider category. The stream id of the overlaid channel proves which category it replaces. + let mut staged_sports = test_group(XtreamCluster::Live, "Sports Rewritten", "staged"); + staged_sports.id = 5; + staged_sports.channels[0].header.id = "200".intern(); + + let groups = apply_staged_overlay_groups( + &provider, + ClusterFlags::Live, + vec![provider_news, provider_sports], + vec![staged_sports], + ); + + assert_eq!(groups.len(), 2); + // Category 5 keeps its content: no staged channel belongs to it. + assert_eq!(groups[0].title.as_ref(), "News"); + assert_eq!(groups[0].id, 5); + assert_eq!(groups[0].channels[0].header.id.as_ref(), "100"); + // Category 7 carries the rewritten staged group, under its own id. + assert_eq!(groups[1].title.as_ref(), "Sports Rewritten"); + assert_eq!(groups[1].id, 7); + assert_eq!(groups[1].channels[0].header.id.as_ref(), "200"); + assert_unique_category_ids(&groups); +} + +#[test] +fn staged_xtream_split_group_does_not_replace_an_unrelated_category() { + let provider = ConfigInput { + name: "provider".intern(), + input_type: InputType::Xtream, + url: "http://provider.example".to_string(), + username: Some("real-user".to_string()), + password: Some("real-pass".to_string()), + ..Default::default() + }; + let mut provider_sports = test_group(XtreamCluster::Live, "Sports", "provider"); + provider_sports.id = 5; + provider_sports.channels[0].header.id = "100".intern(); + let mut second_sport = provider_sports.channels[0].clone(); + second_sport.header.id = "101".intern(); + provider_sports.channels.push(second_sport); + let mut provider_news = test_group(XtreamCluster::Live, "News", "provider"); + provider_news.id = 7; + provider_news.channels[0].header.id = "200".intern(); + + let mut staged_first = test_group(XtreamCluster::Live, "Sports A", "staged"); + staged_first.id = 5; + staged_first.channels[0].header.id = "100".intern(); + let mut staged_second = test_group(XtreamCluster::Live, "News", "staged"); + staged_second.id = 7; + staged_second.channels[0].header.id = "101".intern(); + + let groups = apply_staged_overlay_groups( + &provider, + ClusterFlags::Live, + vec![provider_sports, provider_news], + vec![staged_first, staged_second], + ); + + assert_eq!(groups.len(), 3); + assert_eq!(groups[0].id, 5); + assert_eq!(groups[0].title.as_ref(), "Sports A"); + assert_eq!(groups[1].id, 7); + assert_eq!(groups[1].channels[0].header.id.as_ref(), "200"); + assert_eq!(groups[2].title.as_ref(), "News"); + assert_eq!(groups[2].channels[0].header.id.as_ref(), "101"); + assert_unique_category_ids(&groups); +} + +#[test] +fn staged_m3u_unknown_streams_do_not_match_positional_category_id() { + let provider = ConfigInput { + name: "provider".intern(), + input_type: InputType::Xtream, + url: "http://provider.example".to_string(), + username: Some("real-user".to_string()), + password: Some("real-pass".to_string()), + ..Default::default() + }; + let mut provider_sports = test_group(XtreamCluster::Live, "Sports", "provider"); + provider_sports.id = 5; + provider_sports.channels[0].header.id = "100".intern(); + let mut staged_new = test_group(XtreamCluster::Live, "New", "staged"); + staged_new.id = 5; + staged_new.channels[0].header.id = "300".intern(); + + let groups = super::apply_staged_overlay_groups( + &provider, + StagedInputType::M3u, + ClusterFlags::Live, + vec![provider_sports], + vec![staged_new], + ); + + assert_eq!(groups.len(), 2); + assert_eq!(groups[0].title.as_ref(), "Sports"); + assert_eq!(groups[0].channels[0].header.id.as_ref(), "100"); + assert_eq!(groups[1].title.as_ref(), "New"); + assert_eq!(groups[1].channels[0].header.id.as_ref(), "300"); + assert_unique_category_ids(&groups); +} + +#[test] +fn staged_xtream_overlay_falls_back_to_the_secondary_stream_overlap() { + let provider = ConfigInput { + name: "provider".intern(), + input_type: InputType::Xtream, + url: "http://provider.example".to_string(), + username: Some("real-user".to_string()), + password: Some("real-pass".to_string()), + ..Default::default() + }; + let mut provider_alpha = test_group(XtreamCluster::Live, "Alpha", "provider"); + provider_alpha.id = 5; + provider_alpha.channels[0].header.id = "100".intern(); + let mut provider_beta = test_group(XtreamCluster::Live, "Beta", "provider"); + provider_beta.id = 7; + provider_beta.channels[0].header.id = "200".intern(); + let mut beta_second = provider_beta.channels[0].clone(); + beta_second.header.id = "201".intern(); + provider_beta.channels.push(beta_second); + + // Both staged groups overlap Beta; the stronger one takes it, so the other has to fall back to the + // provider category it also owns streams of instead of becoming a new category. + let mut staged_strong = test_group(XtreamCluster::Live, "Beta Rewritten", "staged"); + staged_strong.id = 0; + staged_strong.channels[0].header.id = "200".intern(); + let mut strong_second = staged_strong.channels[0].clone(); + strong_second.header.id = "201".intern(); + staged_strong.channels.push(strong_second); + + let mut staged_weak = test_group(XtreamCluster::Live, "Alpha Rewritten", "staged"); + staged_weak.id = 0; + staged_weak.channels[0].header.id = "200".intern(); + let mut weak_second = staged_weak.channels[0].clone(); + weak_second.header.id = "201".intern(); + staged_weak.channels.push(weak_second); + let mut weak_third = staged_weak.channels[0].clone(); + weak_third.header.id = "100".intern(); + staged_weak.channels.push(weak_third); + + let groups = apply_staged_overlay_groups( + &provider, + ClusterFlags::Live, + vec![provider_alpha, provider_beta], + vec![staged_strong, staged_weak], + ); + + assert_eq!(groups.len(), 2); + assert_eq!(groups[0].title.as_ref(), "Alpha Rewritten"); + assert_eq!(groups[0].id, 5); + assert_eq!(groups[0].channels.len(), 3); + assert_eq!(groups[1].title.as_ref(), "Beta Rewritten"); + assert_eq!(groups[1].id, 7); + assert_eq!(groups[1].channels.len(), 2); + assert_unique_category_ids(&groups); +} + +#[test] +fn staged_xtream_overlay_gives_a_new_category_a_free_category_id() { + let provider = ConfigInput { + name: "provider".intern(), + input_type: InputType::Xtream, + url: "http://provider.example".to_string(), + username: Some("real-user".to_string()), + password: Some("real-pass".to_string()), + ..Default::default() + }; + let mut provider_sports = test_group(XtreamCluster::Live, "Sports", "provider"); + provider_sports.id = 2; + provider_sports.channels[0].header.id = "200".intern(); + + // Staged groups without a category id must not reach persistence as a shared zero key. + let mut staged_kids = test_group(XtreamCluster::Live, "Kids", "staged"); + staged_kids.id = 0; + staged_kids.channels[0].header.id = "300".intern(); + let mut staged_movies = test_group(XtreamCluster::Live, "Movies", "staged"); + staged_movies.id = 0; + staged_movies.channels[0].header.id = "400".intern(); + + let groups = apply_staged_overlay_groups( + &provider, + ClusterFlags::Live, + vec![provider_sports], + vec![staged_kids, staged_movies], + ); + + assert_eq!(groups.len(), 3); + assert_eq!(groups[0].title.as_ref(), "Sports"); + assert_eq!(groups[0].id, 2); + assert_eq!(groups[1].title.as_ref(), "Kids"); + assert_eq!(groups[1].id, 3); + assert_eq!(groups[2].title.as_ref(), "Movies"); + assert_eq!(groups[2].id, 4); + assert_unique_category_ids(&groups); +} + +#[test] +fn staged_xtream_overlay_empty_staged_group_without_provider_fallback_is_omitted() { + let provider = ConfigInput { + name: "provider".intern(), + input_type: InputType::Xtream, + url: "http://provider.example".to_string(), + username: Some("real-user".to_string()), + password: Some("real-pass".to_string()), + ..Default::default() + }; + let mut provider_action = test_group(XtreamCluster::Video, "Action", "provider"); + provider_action.channels[0].header.id = "10".intern(); + + let mut staged_horror = test_group(XtreamCluster::Video, "Horror", "staged"); + staged_horror.id = 2; + staged_horror.channels[0].header.id = "non-numeric".intern(); + + let groups = apply_staged_overlay_groups(&provider, ClusterFlags::Vod, vec![provider_action], vec![staged_horror]); + + assert_eq!(groups.len(), 1); + assert_eq!(groups[0].title.as_ref(), "Action"); +} + #[test] fn staged_overlay_is_skipped_when_provider_playlist_is_cached() { let result = PlaylistDownloadResult::new(vec![], vec![], true, false); diff --git a/backend/session/src/active_provider_manager.rs b/backend/session/src/active_provider_manager.rs index d4ed5f9cb..2de48279e 100644 --- a/backend/session/src/active_provider_manager.rs +++ b/backend/session/src/active_provider_manager.rs @@ -38,6 +38,18 @@ type PreemptionCandidate = (PriorityOwner, AllocationId, i8, Instant); type PriorityKey = (i8, Reverse, AllocationId); const PREEMPTION_COMPLETION_TIMEOUT: Duration = Duration::from_millis(1500); +const EVICTED_PROVIDER_RELEASE_POLL_INTERVAL: Duration = Duration::from_millis(10); + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct ProviderReleaseSnapshot { + pub addr: SocketAddr, + single_allocations: Vec, + shared_subscribers: Vec, +} + +impl ProviderReleaseSnapshot { + pub fn is_empty(&self) -> bool { self.single_allocations.is_empty() && self.shared_subscribers.is_empty() } +} #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ConnectionKind { @@ -118,6 +130,7 @@ impl AcquireProviderParams<'_> { #[derive(Debug, Clone)] struct SharedAllocation { allocation_id: AllocationId, + origin_subscriber_id: SharedSubscriberId, allocation: ProviderAllocation, /// Keyed by unique subscriber id, never by socket: two external clients behind one /// reverse proxy must not collapse into a single entry. @@ -444,6 +457,85 @@ impl std::ops::Deref for ActiveProviderManager { } impl ActiveProviderManager { + fn has_connections_for_addr(&self, addr: &SocketAddr) -> bool { + let _transition = self.lock_capacity_transition(); + let connections = self.read_connections(); + connections.single.values().any(|info| info.client_addr == *addr) + || connections + .shared + .by_key + .values() + .any(|shared| shared.connections.values().any(|subscriber| subscriber.addr == *addr)) + } + + /// Waits for every allocation currently using an address to leave the registry. + /// Admission handoffs use `wait_for_snapshot_release` to ignore later allocations at that address. + pub async fn wait_for_addr_release(&self, addr: &SocketAddr, timeout: Duration) -> bool { + let deadline = TokioInstant::now() + timeout; + loop { + if !self.has_connections_for_addr(addr) { + return true; + } + let now = TokioInstant::now(); + if now >= deadline { + return false; + } + tokio::time::sleep_until((now + EVICTED_PROVIDER_RELEASE_POLL_INTERVAL).min(deadline)).await; + } + } + + pub(crate) fn release_snapshot_for_addr(&self, addr: &SocketAddr) -> ProviderReleaseSnapshot { + let _transition = self.lock_capacity_transition(); + let connections = self.read_connections(); + let single_allocations = connections + .single + .values() + .filter_map(|info| (info.client_addr == *addr).then_some(info.allocation_id)) + .collect(); + let shared_subscribers = connections + .shared + .by_key + .values() + .flat_map(|shared| { + shared.connections.iter().filter_map(|(id, subscriber)| (subscriber.addr == *addr).then_some(*id)) + }) + .collect(); + ProviderReleaseSnapshot { addr: *addr, single_allocations, shared_subscribers } + } + + fn has_connections_from_snapshot(&self, snapshot: &ProviderReleaseSnapshot) -> bool { + let _transition = self.lock_capacity_transition(); + let connections = self.read_connections(); + snapshot.single_allocations.iter().any(|id| { + connections.single.contains_key(id) + || connections.shared.shared_by_allocation_id.get(id).is_some_and(|key| { + connections.shared.by_key.get(key).is_some_and(|shared| { + connections.shared.key_by_subscriber.get(&shared.origin_subscriber_id) == Some(key) + }) + }) + }) || snapshot.shared_subscribers.iter().any(|id| connections.shared.key_by_subscriber.contains_key(id)) + } + + /// Waits for the kicked transport's original provider allocations to leave the registry. + /// The socket close may be signalled before its response bodies and provider handles are dropped. + pub(crate) async fn wait_for_snapshot_release( + &self, + snapshot: &ProviderReleaseSnapshot, + timeout: Duration, + ) -> bool { + let deadline = TokioInstant::now() + timeout; + loop { + if !self.has_connections_from_snapshot(snapshot) { + return true; + } + let now = TokioInstant::now(); + if now >= deadline { + return false; + } + tokio::time::sleep_until((now + EVICTED_PROVIDER_RELEASE_POLL_INTERVAL).min(deadline)).await; + } + } + fn upsert_priority_entry( connections: &mut Connections, provider_name: &Arc, @@ -2515,6 +2607,7 @@ impl ActiveProviderManager { Arc::clone(&shared_key), SharedAllocation { allocation_id: handle.allocation_id, + origin_subscriber_id: subscriber_id, allocation: info.allocation, connections: HashMap::from([( subscriber_id, @@ -2632,7 +2725,7 @@ impl ActiveProviderManager { #[cfg(test)] mod tests { use super::{ActiveProviderManager, ConnectionKind, PlaybackLeaseRef}; - use crate::{EventManager, SharedStreamManager}; + use crate::{ActiveUserManager, EventManager, SharedStreamManager}; use arc_swap::{ArcSwap, ArcSwapOption}; use shared::{ defaults::{default_probe_user_priority, default_user_priority}, @@ -2873,6 +2966,66 @@ mod tests { Ok(()) } + #[tokio::test] + async fn release_snapshot_tracks_original_allocation_after_addr_reuse() { + let app_cfg = create_test_app_config_with_pool(2, 3); + let events = Arc::new(EventManager::new()); + let provider = ActiveProviderManager::new(&app_cfg, &events); + let addr = SocketAddr::from(([127, 0, 0, 1], 50_022)); + let input = "provider_1".intern(); + let original = provider + .acquire_connection_with_grace_for_session(&input, &addr, false, 0, ConnectionKind::Normal, Some("old")) + .expect("original allocation"); + let original_snapshot = provider.release_snapshot_for_addr(&addr); + assert!(!provider.wait_for_snapshot_release(&original_snapshot, Duration::ZERO).await); + + provider.release_handle(&original); + let replacement = provider + .acquire_connection_with_grace_for_session(&input, &addr, false, 0, ConnectionKind::Normal, Some("new")) + .expect("replacement allocation"); + let replacement_snapshot = provider.release_snapshot_for_addr(&addr); + assert!(provider.wait_for_snapshot_release(&original_snapshot, Duration::ZERO).await); + assert!(!provider.wait_for_snapshot_release(&replacement_snapshot, Duration::ZERO).await); + + let geoip = Arc::new(ArcSwapOption::default()); + let users = ActiveUserManager::new(&Config::default(), &geoip, &events); + users.set_pending_provider_release("user", original_snapshot.clone()).await; + users.set_pending_provider_release("user", replacement_snapshot.clone()).await; + assert!(!users.clear_pending_provider_release("user", &original_snapshot).await); + assert_eq!(users.pending_provider_release("user").await, Some(replacement_snapshot.clone())); + assert!(users.clear_pending_provider_release("user", &replacement_snapshot).await); + assert_eq!(users.pending_provider_release("user").await, None); + provider.release_handle(&replacement); + } + + #[tokio::test] + async fn release_snapshot_tracks_shared_subscriber_without_waiting_for_other_subscribers() { + let app_cfg = create_test_app_config_single_provider_pool(); + let events = Arc::new(EventManager::new()); + let provider = ActiveProviderManager::new(&app_cfg, &events); + let input = "provider_1".intern(); + let first_addr = SocketAddr::from(([127, 0, 0, 1], 50_023)); + let second_addr = SocketAddr::from(([127, 0, 0, 1], 50_024)); + let first = SharedSubscriberId::from_stream_uid(50_023); + let second = SharedSubscriberId::from_stream_uid(50_024); + let origin = + provider.acquire_connection(&input, &first_addr, 0, ConnectionKind::Normal).expect("shared origin"); + let before_promotion = provider.release_snapshot_for_addr(&first_addr); + assert!(provider.make_shared_connection(&origin, "shared-release", first)); + provider + .add_shared_connection(&second_addr, second, "shared-release", 0, ConnectionKind::Normal) + .expect("second subscriber"); + let snapshot = provider.release_snapshot_for_addr(&first_addr); + assert!(!provider.wait_for_snapshot_release(&before_promotion, Duration::ZERO).await); + assert!(!provider.wait_for_snapshot_release(&snapshot, Duration::ZERO).await); + + provider.release_connection(&first_addr); + assert!(provider.wait_for_snapshot_release(&before_promotion, Duration::ZERO).await); + assert!(provider.wait_for_snapshot_release(&snapshot, Duration::ZERO).await); + assert_eq!(provider.get_provider_connections_count(), 1); + provider.release_connection(&second_addr); + } + #[tokio::test(start_paused = true)] async fn manifest_retries_without_media_do_not_reserve_capacity() { let app_cfg = create_test_app_config_with_pool(2, 3); diff --git a/backend/session/src/active_user_manager/mod.rs b/backend/session/src/active_user_manager/mod.rs index 1f53da87a..7e636a464 100644 --- a/backend/session/src/active_user_manager/mod.rs +++ b/backend/session/src/active_user_manager/mod.rs @@ -1,5 +1,5 @@ use crate::{ - active_provider_manager::ConnectionKind, + active_provider_manager::{ConnectionKind, ProviderReleaseSnapshot}, connection_manager::CleanupEvent, stream::{uses_direct_body_idle_timeout, DIRECT_BODY_IDLE_TIMEOUT_SECS}, ActiveProviderManager, EventManager, @@ -690,6 +690,9 @@ pub struct ActiveUserManager { cleanup_tx: tokio::sync::OnceCell>, provider_manager: tokio::sync::OnceCell>, transition_gates: Mutex>>>, + // An evicted stream can keep its provider slot after its user count is released. + // Retain the handoff across bounded admission attempts until the slot is gone. + pending_provider_releases: Mutex>, pub dropped_cleanup_events: AtomicU64, reentry_suppressed_total: AtomicU64, divergence_cache: Mutex>, @@ -791,6 +794,7 @@ impl ActiveUserManager { cleanup_tx: tokio::sync::OnceCell::new(), provider_manager: tokio::sync::OnceCell::new(), transition_gates: Mutex::new(HashMap::new()), + pending_provider_releases: Mutex::new(HashMap::new()), dropped_cleanup_events: AtomicU64::new(0), reentry_suppressed_total: AtomicU64::new(0), divergence_cache: Mutex::new(LruCache::new(DIVERGENCE_CACHE_CAPACITY)), @@ -845,6 +849,27 @@ impl ActiveUserManager { gate.lock_owned().await } + pub(crate) async fn pending_provider_release(&self, username: &str) -> Option { + self.pending_provider_releases.lock().await.get(username).cloned() + } + + pub(crate) async fn set_pending_provider_release(&self, username: &str, snapshot: ProviderReleaseSnapshot) { + self.pending_provider_releases.lock().await.insert(username.to_owned(), snapshot); + } + + pub(crate) async fn clear_pending_provider_release( + &self, + username: &str, + snapshot: &ProviderReleaseSnapshot, + ) -> bool { + let mut pending = self.pending_provider_releases.lock().await; + if pending.get(username) != Some(snapshot) { + return false; + } + pending.remove(username); + true + } + fn should_reuse_stream_for_session(existing_stream: &StreamInfo, incoming_channel: &StreamChannel) -> bool { existing_stream.channel.item_type.requires_provider_affinity() || incoming_channel.item_type.requires_provider_affinity() diff --git a/backend/session/src/admission.rs b/backend/session/src/admission.rs index 235b69b1d..33539e33a 100644 --- a/backend/session/src/admission.rs +++ b/backend/session/src/admission.rs @@ -18,7 +18,7 @@ use crate::{ }; use log::debug; use shared::model::{AdmissionStrategy, ConnectionDenied, EventMessage, UserConnectionPermission, VirtualId}; -use std::sync::Arc; +use std::{sync::Arc, time::Duration}; use tuliprox_core::model::{AppConfig, Fingerprint, ProxyUserCredentials}; /// The handles admission reads from the running server. @@ -34,6 +34,7 @@ pub struct AdmissionCtx { /// Default duration an eviction is remembered if not configured. pub const DEFAULT_RECENT_EVICTION_REENTRY_TTL_MS: u64 = shared::defaults::DEFAULT_RECENT_EVICTION_REENTRY_TTL_MS; +const EVICTED_PROVIDER_RELEASE_TIMEOUT: Duration = Duration::from_millis(1500); /// Reentry cooldown from a resolved stream config, falling back to the default when /// no `reverse_proxy.stream` block is configured. @@ -287,6 +288,22 @@ enum StrategyLoopResult { Rejected(crate::AdmissionRejectionReason), } +async fn pending_provider_release_ready(adm: &AdmissionCtx, username: &str) -> bool { + let Some(snapshot) = adm.active_users.pending_provider_release(username).await else { + return true; + }; + if !adm + .connection_manager + .provider_manager + .wait_for_snapshot_release(&snapshot, EVICTED_PROVIDER_RELEASE_TIMEOUT) + .await + { + debug!("Provider allocation for evicted connection {} remains active after close timeout", snapshot.addr); + return false; + } + adm.active_users.clear_pending_provider_release(username, &snapshot).await +} + /// Shared strategy-evaluation loop used by both the initial admission path /// (`resolve_admission_with_strategies`) and the remaining-strategies path /// (`evaluate_remaining_strategies_after_grace`). @@ -361,7 +378,36 @@ where let connections_before = adm.active_users.user_connections(username).await; let ttl = get_reentry_ttl(adm); adm.active_users.mark_recent_eviction_guard_for_addr(&target.addr, *request_addr, ttl).await; + let pending_release = if target.addr == *request_addr { + None + } else { + let snapshot = adm.connection_manager.provider_manager.release_snapshot_for_addr(&target.addr); + if snapshot.is_empty() { + None + } else { + adm.active_users.set_pending_provider_release(username, snapshot.clone()).await; + Some(snapshot) + } + }; adm.connection_manager.release_connection_as_kicked(&target.addr).await; + // This request cannot wait for its own transport to close. + if let Some(snapshot) = pending_release { + let released = adm + .connection_manager + .provider_manager + .wait_for_snapshot_release(&snapshot, EVICTED_PROVIDER_RELEASE_TIMEOUT) + .await; + if !released { + debug!( + "Provider allocation for evicted connection {} remains active after close timeout", + target.addr + ); + return StrategyLoopResult::Rejected(crate::AdmissionRejectionReason::UserConnectionsExhausted); + } + if !adm.active_users.clear_pending_provider_release(username, &snapshot).await { + return StrategyLoopResult::Rejected(crate::AdmissionRejectionReason::UserConnectionsExhausted); + } + } performed_legitimate_eviction = true; let retry_admission = get_admission_for_request(adm, request).await; if retry_admission.permission() == UserConnectionPermission::Allowed { @@ -416,7 +462,34 @@ pub async fn resolve_admission_with_strategies( let username = request.username; let admission = get_admission_for_request(adm, &request).await; + if admission.permission() != UserConnectionPermission::Exhausted + && adm.active_users.pending_provider_release(username).await.is_none() + { + return AdmissionStrategyResolution { admission, grace_mode: None, grace_context: None }; + } + + let _admission_guard = adm.active_users.acquire_user_admission(username).await; + + if !pending_provider_release_ready(adm, username).await { + return AdmissionStrategyResolution { + admission: crate::ConnectionAdmission::exhausted( + crate::AdmissionRejectionReason::UserConnectionsExhausted, + admission.kind(), + ), + grace_mode: None, + grace_context: None, + }; + } + + // Re-read admission now that the gate is held. The first read above happened + // before we queued on the gate, so a request ahead of us may have released + // the very slot we are about to evict somebody for. Walking the strategies on + // the stale snapshot kicks a live connection to free a slot that is already + // free. + let admission = get_admission_for_request(adm, &request).await; + if admission.permission() != UserConnectionPermission::Exhausted { + debug!("Admission became available while waiting on the admission gate for user {username}"); return AdmissionStrategyResolution { admission, grace_mode: None, grace_context: None }; } @@ -433,20 +506,6 @@ pub async fn resolve_admission_with_strategies( }; } - let _admission_guard = adm.active_users.acquire_user_admission(username).await; - - // Re-read admission now that the gate is held. The first read above happened - // before we queued on the gate, so a request ahead of us may have released - // the very slot we are about to evict somebody for. Walking the strategies on - // the stale snapshot kicks a live connection to free a slot that is already - // free. - let admission = get_admission_for_request(adm, &request).await; - - if admission.permission() != UserConnectionPermission::Exhausted { - debug!("Admission became available while waiting on the admission gate for user {username}"); - return AdmissionStrategyResolution { admission, grace_mode: None, grace_context: None }; - } - let build_grace_ctx = |global_idx: usize| GraceResolutionContext { strategy_index: global_idx, strategies: strategies.clone(), @@ -513,6 +572,17 @@ pub async fn evaluate_remaining_strategies_after_grace( let _admission_guard = adm.active_users.acquire_user_admission(username).await; + if !pending_provider_release_ready(adm, username).await { + return AdmissionStrategyResolution { + admission: crate::ConnectionAdmission::exhausted( + crate::AdmissionRejectionReason::UserConnectionsExhausted, + original_kind, + ), + grace_mode: None, + grace_context: None, + }; + } + match evaluate_admission_strategy_loop( adm, &request, diff --git a/backend/session/src/provider_lineup_manager.rs b/backend/session/src/provider_lineup_manager.rs index 5d2cb9a5e..81f91a7e6 100644 --- a/backend/session/src/provider_lineup_manager.rs +++ b/backend/session/src/provider_lineup_manager.rs @@ -278,32 +278,7 @@ impl MultiProviderLineup { names } - /// Attempts to acquire the next available provider from a specific priority group. - /// - /// # Parameters - /// - `priority_group`: The provider group to search within. - /// - /// # Returns - /// - `ProviderAllocation`: A reference to the next available provider in the specified group. - /// - /// # Behavior - /// - Iterates through the providers in the given group in a round-robin manner. - /// - Checks if a provider has available capacity before selecting it. - /// - Uses atomic operations to maintain fair provider selection. - /// - /// # Thread Safety - /// - Uses `RwLock` for safe concurrent access. - /// - Ensures fair provider allocation across multiple threads. - /// - /// # Example Usage - /// ```text - /// let lineup = MultiProviderLineup::new(&config); - /// match lineup.acquire_next_provider_from_group(priority_group).await { - /// ProviderAllocation::Exhausted => println!("All providers exhausted"), - /// ProviderAllocation::Available(provider) => println!("Provider available {}", provider.name), - /// ProviderAllocation::GracePeriod(provider) => println!("Provider with grace period {}", provider.name), - /// } - /// ``` + /// Attempts to acquire the next available provider from a priority group in round-robin order. fn acquire_next_provider_from_group( priority_group: &ProviderPriorityGroup, grace: bool, @@ -400,93 +375,61 @@ impl MultiProviderLineup { } /// Attempts to acquire a provider from the lineup based on priority and availability. - /// - /// # Returns - /// - `ProviderAllocation`: A reference to the acquired provider if allocation was successful. - /// - /// # Behavior - /// - The method iterates through provider priority groups in a round-robin fashion. - /// - It attempts to allocate a provider from the highest priority group first. - /// - If a provider has available capacity, it is returned. - /// - If all providers in a group are exhausted, it moves to the next group. - /// - Updates the internal index to ensure fair distribution of requests. - /// - /// # Thread Safety - /// - Uses atomic operations (`AtomicUsize`) for thread-safe indexing. - /// - Uses `RwLock` for thread-safe provider allocation. - /// - /// # Example Usage - /// ```text - /// let lineup = MultiProviderLineup::new(&config); - /// match lineup.acquire().await { - /// ProviderAllocation::Exhausted => println!("All providers exhausted"), - /// ProviderAllocation::Available(provider) => println!("Provider available {}", provider.name), - /// ProviderAllocation::GracePeriod(provider) => println!("Provider with grace period {}", provider.name), - /// } - /// ``` #[cfg(test)] fn acquire(&self, with_grace: bool, grace_period_timeout_secs: u64) -> ProviderAllocation { self.acquire_excluding(with_grace, grace_period_timeout_secs, &std::collections::HashSet::new()) } + fn scan_priority_groups( + &self, + with_grace: bool, + mut select: impl FnMut(&ProviderPriorityGroup, bool) -> Option, + ) -> Option { + // Prefer providers with available capacity (no grace allocations), + // scanning priority groups from highest -> lowest. + for priority_group in &self.providers { + if let Some(item) = select(priority_group, false) { + return Some(item); + } + } + + if !with_grace { + return None; + } + + // If every provider is at capacity, allow grace allocations while respecting priority order. + for priority_group in &self.providers { + if let Some(item) = select(priority_group, true) { + return Some(item); + } + } + + None + } + fn acquire_excluding( &self, with_grace: bool, grace_period_timeout_secs: u64, excluded_providers: &std::collections::HashSet>, ) -> ProviderAllocation { - // Prefer providers with available capacity (no grace allocations), - // scanning priority groups from highest -> lowest. - for priority_group in &self.providers { + self.scan_priority_groups(with_grace, |priority_group, grace| { let allocation = Self::acquire_next_provider_from_group( priority_group, - false, + grace, grace_period_timeout_secs, excluded_providers, ); - if !matches!(allocation, ProviderAllocation::Exhausted) { - return allocation; - } - } - - if !with_grace { - return ProviderAllocation::Exhausted; - } - - // If every provider is at capacity, allow grace allocations while respecting priority order. - for priority_group in &self.providers { - let allocation = Self::acquire_next_provider_from_group( - priority_group, - true, - grace_period_timeout_secs, - excluded_providers, - ); - if !matches!(allocation, ProviderAllocation::Exhausted) { - return allocation; - } - } - - ProviderAllocation::Exhausted + (!matches!(allocation, ProviderAllocation::Exhausted)).then_some(allocation) + }) + .unwrap_or(ProviderAllocation::Exhausted) } - // it intended to use with redirects to cycle through provider + // It is intended to use with redirects to cycle through providers. fn get_next(&self, grace_period_timeout_secs: u64) -> Option> { - // Prefer providers with available capacity (no grace allocations), - // scanning priority groups from highest -> lowest. - for priority_group in &self.providers { - if let Some(config) = Self::get_next_provider_from_group(priority_group, false, grace_period_timeout_secs) { - return Some(config); - } - } - - // If no provider is available, allow grace. - for priority_group in &self.providers { - if let Some(config) = Self::get_next_provider_from_group(priority_group, true, grace_period_timeout_secs) { - return Some(config); - } - } - - None + self.scan_priority_groups(true, |priority_group, grace| { + Self::get_next_provider_from_group(priority_group, grace, grace_period_timeout_secs) + }) } #[cfg(test)] diff --git a/docs/src/configuration/source.md b/docs/src/configuration/source.md index f50596e93..3dcd6aaa3 100644 --- a/docs/src/configuration/source.md +++ b/docs/src/configuration/source.md @@ -987,22 +987,45 @@ and the catalog page limit are configured on the parent input in the Web UI or Y ### 2.5 Staged Sources (`staged`) -The **staged input** is a first-class input type for pre-formatted playlists. Tuliprox reads the selected playlist -clusters from the staged source, then stores the merged result in the linked provider input. Stream delivery and API -requests continue to use that provider input. +The **staged input** is a first-class input type for pre-formatted playlists. Tuliprox overlays the selected clusters +of the linked provider with the staged groups, then stores the merged result in that provider input. Stream delivery +and API requests continue to use the provider input. This is useful when an external playlist editor already has the desired channel order, groups, and original stream IDs. -For example, an IPTV editor can provide the Live playlist layout while the actual streams are still opened against the -Xtream or M3U provider. +A staged playlist is typically a restructured copy of the provider playlist: the streams stay the same, while group +names, channel names, and their order may differ. For example, an IPTV editor can provide the Live playlist layout while +the actual streams are still opened against the Xtream or M3U provider. **Data flow:** -* `staged input -> provider input`: the staged input is an overlay for that provider. Clusters listed in - `staged.clusters` are loaded from the staged input; the remaining clusters are loaded from the provider input itself. +* `staged input -> provider input`: the staged input is an overlay for that provider. Inside the clusters listed in + `staged.clusters`, each staged group of an `xtream` provider replaces the provider category it belongs to; provider + categories that the staged playlist does not represent stay as they are. For any other provider type the listed + clusters are replaced entirely by the staged groups. Clusters not listed are loaded from the provider input itself. The merged playlist is persisted under the provider input, and streaming/API requests still target the provider input. * `staged input -> target` is not supported. Use a normal `m3u` or `xtream` input if the source should be connected directly to a target. +**Overlay matching (`xtream` providers):** + +* Matching is ID-driven. A staged group overlays the provider category that owns its staged stream IDs, strongest overlap + first; when that category is already claimed by a stronger overlap, the group's next-ranked category is used. A group + with known provider stream IDs that loses the assignment is added as a new category. When none of its stream IDs are + known, an `xtream` staged group is resolved by category ID; an `m3u` staged group skips that step because its category + IDs are positional. The group title acts as a last resort. A group title therefore never overrides a stream ID match, + so a renamed or split group cannot take over an unrelated category by accident. +* The overlaid category keeps its own category ID; the staged playlist supplies the group name and the channel order. +* Known streams keep their provider playback URL, including a direct source URL. New streams use the provider address and + credentials; for VOD, the staged stream's container metadata supplies the extension when available. +* Every staged channel needs a numeric provider stream ID in its `header.id`, either from an `xui-id` / `cuid` + attribute or from the numeric last URL segment. Rows without one are skipped, a single warning per group reports how + many rows were dropped, and the provider category stays in place instead of being emptied. +* A staged group that matches no provider category is added as a new category. Its category ID is reused when it is + still free, otherwise Tuliprox assigns an unused ID, so two groups of one cluster never share an ID. + +For every other provider type the staged groups of a listed cluster are taken as they are: their group IDs, URLs, and +channel IDs stay unchanged, and the clusters they cover are replaced completely. + #### Configuration Example (Provider With Staged Live Overlay) In this setup, Live-TV comes from the external staged playlist, while VOD and Series come from the original Xtream @@ -1034,13 +1057,16 @@ inputs: | `method` | Enum | No | `GET` | HTTP request method (`GET` or `POST`). Not inherited from the provider input. | | `headers` | Dict | No | | Custom HTTP headers for the staged download. Not inherited from the provider input. | | `staged.for_input` | String | Yes | | Provider input name. Must reference a non-staged `m3u` or `xtream` input. | -| `staged.clusters` | List | No | all | Clusters loaded from the staged input: `live`, `vod`, `series`. | +| `staged.clusters` | List | No | all | Clusters whose categories are overlaid from the staged input: `live`, `vod`, `series`. | #### Staged Cluster Behavior & Validation -`staged.clusters` is the group of clusters loaded from the staged input. +`staged.clusters` names the clusters whose categories are overlaid from the staged input. * The referenced provider supplies all clusters not listed in `staged.clusters`. +* Inside a listed cluster of an `xtream` provider, only the categories represented by the staged playlist are overlaid; + provider categories without a staged counterpart stay available. For any other provider type the listed clusters are + replaced entirely. * `staged.for_input` must reference an existing non-staged `m3u` or `xtream` input. * Each provider input can have at most one staged overlay. * `staged.clusters` must not be empty. diff --git a/frontend/src/app/components/source_editor/block_view.rs b/frontend/src/app/components/source_editor/block_view.rs index cd8dd39ac..3f0180ae9 100644 --- a/frontend/src/app/components/source_editor/block_view.rs +++ b/frontend/src/app/components/source_editor/block_view.rs @@ -9,6 +9,11 @@ use yew::{ }; const DOUBLE_TAP_THRESHOLD_MS: f64 = 320.0; + +fn is_span_target(e: &E) -> bool { + e.target_dyn_into::().is_some_and(|target| target.tag_name().eq_ignore_ascii_case("span")) +} + #[derive(Properties, PartialEq)] pub struct BlockProps { pub(crate) block: Block, @@ -53,11 +58,8 @@ pub fn BlockView(props: &BlockProps) -> Html { let on_block_mouse_down = props.on_mouse_down.clone(); Callback::from(move |e: MouseEvent| { e.prevent_default(); - if let Some(target) = e.target_dyn_into::() { - let tag = target.tag_name().to_lowercase(); - if &tag == "span" { - return; - } + if is_span_target(&e) { + return; } e.stop_propagation(); on_block_mouse_down.emit((block_id, e)); @@ -71,11 +73,8 @@ pub fn BlockView(props: &BlockProps) -> Html { let handle_touch_start = { let on_block_touch_start = props.on_touch_start.clone(); Callback::from(move |e: TouchEvent| { - if let Some(target) = e.target_dyn_into::() { - let tag = target.tag_name().to_lowercase(); - if &tag == "span" { - return; - } + if is_span_target(&e) { + return; } e.stop_propagation(); on_block_touch_start.emit((block_id, e)); @@ -86,11 +85,8 @@ pub fn BlockView(props: &BlockProps) -> Html { let on_edit = props.on_edit.clone(); let last_touch_end_ts = last_touch_end_ts.clone(); Callback::from(move |e: TouchEvent| { - if let Some(target) = e.target_dyn_into::() { - let tag = target.tag_name().to_lowercase(); - if &tag == "span" { - return; - } + if is_span_target(&e) { + return; } let mut last_touch_end_ts = last_touch_end_ts.borrow_mut(); diff --git a/frontend/src/app/components/source_editor/editor_view.rs b/frontend/src/app/components/source_editor/editor_view.rs index 75d31e49f..6c3e51a2b 100644 --- a/frontend/src/app/components/source_editor/editor_view.rs +++ b/frontend/src/app/components/source_editor/editor_view.rs @@ -233,7 +233,8 @@ fn clear_active_interaction(editor_state: &mut EditorState) -> bool { let had_active_interaction = editor_state.is_panning || editor_state.drag.block_id.is_some() || editor_state.selection.is_selecting - || editor_state.selection.selection_rect.is_some(); + || editor_state.selection.selection_rect.is_some() + || editor_state.pinch_distance.is_some(); editor_state.block_elements.clear(); editor_state.connection_elements.clear(); if editor_state.drag.block_id.is_some() { @@ -245,6 +246,91 @@ fn clear_active_interaction(editor_state: &mut EditorState) -> bool { had_active_interaction } +fn is_canvas_background(target: Option<&web_sys::Element>, canvas: Option<&web_sys::Element>) -> bool { + let (Some(target), Some(canvas)) = (target, canvas) else { + return false; + }; + target.is_same_node(Some(canvas)) || target.tag_name().eq_ignore_ascii_case("svg") +} + +fn start_canvas_pan(editor_state: &mut EditorState, client_x: f32, client_y: f32) { + editor_state.selection.reset_selection(); + editor_state.is_panning = true; + editor_state.pan_start = (client_x, client_y); +} + +fn pan_canvas(editor_state: &mut EditorState, client_x: f32, client_y: f32) -> MoveBlockParams { + let (start_x, start_y) = editor_state.pan_start; + let dx = client_x - start_x; + let dy = client_y - start_y; + let (canvas_ox, canvas_oy) = editor_state.canvas_offset; + editor_state.canvas_offset = (canvas_ox + dx, canvas_oy + dy); + editor_state.pan_start = (client_x, client_y); + + let initial_positions: Vec<(BlockId, Position)> = editor_state.blocks.iter().map(|b| (b.id, b.position)).collect(); + + (0.0, 0.0, (0.0, 0.0), initial_positions) +} + +fn start_block_drag(editor_state: &mut EditorState, block_id: BlockId, canvas_pos: (f32, f32), ctrl_key: bool) { + let Some(block) = editor_state.get_block(block_id).cloned() else { + return; + }; + + editor_state.selection.group_initial_positions.clear(); + editor_state.drag.dragging_group.clear(); + + let (selected_blocks, new_selection) = { + let is_selected = editor_state.selection.selected_blocks.contains(&block_id); + + if is_selected && ctrl_key { + editor_state.selection.selected_blocks.remove(&block_id); + (editor_state.selection.selected_blocks.clone(), None) + } else if !is_selected { + (HashSet::from([block_id]), Some(block_id)) + } else { + (editor_state.selection.selected_blocks.clone(), None) + } + }; + + let initial_pos: Vec<(BlockId, Position)> = + selected_blocks.iter().filter_map(|id| editor_state.get_block(*id).map(|b| (*id, b.position))).collect(); + + editor_state.drag.dragging_group = selected_blocks; + editor_state.selection.group_initial_positions = initial_pos; + + editor_state + .drag + .with_drag_block_offset(block_id, (canvas_pos.0 - block.position.0, canvas_pos.1 - block.position.1)); + + if let Some(block) = new_selection { + if !ctrl_key { + editor_state.selection.selected_blocks.clear(); + } + editor_state.selection.selected_blocks.insert(block); + } + + editor_state.selection.group_anchor_mouse = canvas_pos; +} + +fn compute_drag_move_params(editor_state: &EditorState, canvas_x: f32, canvas_y: f32) -> Option { + let block_id = editor_state.drag.block_id?; + if editor_state.drag.dragging_group.contains(&block_id) + && !editor_state.selection.group_initial_positions.is_empty() + { + Some(( + canvas_x, + canvas_y, + editor_state.selection.group_anchor_mouse, + editor_state.selection.group_initial_positions.clone(), + )) + } else { + editor_state.get_block(block_id).map(|block| { + (canvas_x, canvas_y, editor_state.selection.group_anchor_mouse, vec![(block_id, block.position)]) + }) + } +} + fn screen_from_world(position: Position, canvas_offset: Position, zoom_factor: f32) -> Position { ((position.0 * zoom_factor) + canvas_offset.0, (position.1 * zoom_factor) + canvas_offset.1) } @@ -1387,136 +1473,42 @@ pub fn SourceEditor(props: &SourceEditorProps) -> Html { }; // ----------------- Drag block logic ----------------- - let handle_block_mouse_down = { + let start_block_drag_action = { let editor_state_ref = editor_state_ref.clone(); let canvas_ref = canvas_ref.clone(); let cursor_grabbing = cursor_grabbing.clone(); - Callback::from(move |(block_id, e): (BlockId, MouseEvent)| { - if !can_write_sources { + move |block_id: BlockId, client_x: f32, client_y: f32, ctrl_key: bool| { + if !can_write_sources || editor_state_ref.borrow().pending_line.is_some() { return; } - e.prevent_default(); - e.stop_propagation(); - - if editor_state_ref.borrow().pending_line.is_some() { - return; - } - - let ctrl_key = e.ctrl_key(); if let Some(canvas) = canvas_ref.cast::() { cursor_grabbing.set(true); let rect = canvas.get_bounding_client_rect(); - let mouse_x = e.client_x() as f32 - rect.left() as f32; - let mouse_y = e.client_y() as f32 - rect.top() as f32; + let canvas_x = client_x - rect.left() as f32; + let canvas_y = client_y - rect.top() as f32; - let possible_block = editor_state_ref.borrow().get_block(block_id).cloned(); let mut editor_state = editor_state_ref.borrow_mut(); - - if let Some(block) = possible_block { - // Prepare group - editor_state.selection.group_initial_positions.clear(); - editor_state.drag.dragging_group.clear(); - - // Neue Auswahllogik: - let (selected_blocks, new_selection) = { - let is_selected = editor_state.selection.selected_blocks.contains(&block_id); - - if is_selected && ctrl_key { - // Ctrl + Click on existing block -> remove from selection - editor_state.selection.selected_blocks.remove(&block_id); - (editor_state.selection.selected_blocks.clone(), None) - } else if !is_selected { - // Block not selected, select only this block - (HashSet::from([block_id]), Some(block_id)) - } else { - // The block is selected and Ctrl is not pressed -> the selection remains as is. - (editor_state.selection.selected_blocks.clone(), None) - } - }; - - // initial positions for drag - let mut initial_pos = Vec::new(); - for id in &selected_blocks { - if let Some(b) = editor_state.get_block(*id) { - initial_pos.push((*id, b.position)); - } - } - - editor_state.drag.dragging_group = selected_blocks.clone(); - editor_state.selection.group_initial_positions = initial_pos; - - // Drag-Offset calculation - editor_state - .drag - .with_drag_block_offset(block_id, (mouse_x - block.position.0, mouse_y - block.position.1)); - - // update selection - if let Some(block) = new_selection { - if !ctrl_key { - editor_state.selection.selected_blocks.clear(); - } - editor_state.selection.selected_blocks.insert(block); - } - - editor_state.selection.group_anchor_mouse = (mouse_x, mouse_y); - } + start_block_drag(&mut editor_state, block_id, (canvas_x, canvas_y), ctrl_key); } + } + }; + + let handle_block_mouse_down = { + let start_block_drag_action = start_block_drag_action.clone(); + Callback::from(move |(block_id, e): (BlockId, MouseEvent)| { + e.prevent_default(); + e.stop_propagation(); + start_block_drag_action(block_id, e.client_x() as f32, e.client_y() as f32, e.ctrl_key()); }) }; let handle_block_touch_start = { - let editor_state_ref = editor_state_ref.clone(); - let canvas_ref = canvas_ref.clone(); - let cursor_grabbing = cursor_grabbing.clone(); - + let start_block_drag_action = start_block_drag_action; Callback::from(move |(block_id, e): (BlockId, TouchEvent)| { - if !can_write_sources { - return; - } e.stop_propagation(); - - if editor_state_ref.borrow().pending_line.is_some() { - return; - } - if let Some(touch) = e.touches().item(0) { - if let Some(canvas) = canvas_ref.cast::() { - cursor_grabbing.set(true); - let rect = canvas.get_bounding_client_rect(); - let touch_x = touch.client_x() as f32 - rect.left() as f32; - let touch_y = touch.client_y() as f32 - rect.top() as f32; - - let possible_block = editor_state_ref.borrow().get_block(block_id).cloned(); - let mut editor_state = editor_state_ref.borrow_mut(); - - if let Some(block) = possible_block { - editor_state.selection.group_initial_positions.clear(); - editor_state.drag.dragging_group.clear(); - - let selected_blocks = if editor_state.selection.selected_blocks.contains(&block_id) { - editor_state.selection.selected_blocks.clone() - } else { - HashSet::from([block_id]) - }; - - let mut initial_pos = Vec::new(); - for id in &selected_blocks { - if let Some(b) = editor_state.get_block(*id) { - initial_pos.push((*id, b.position)); - } - } - - editor_state.drag.dragging_group = selected_blocks.clone(); - editor_state.selection.group_initial_positions = initial_pos; - editor_state - .drag - .with_drag_block_offset(block_id, (touch_x - block.position.0, touch_y - block.position.1)); - editor_state.selection.selected_blocks.clear(); - editor_state.selection.selected_blocks.extend(selected_blocks); - editor_state.selection.group_anchor_mouse = (touch_x, touch_y); - } - } + start_block_drag_action(block_id, touch.client_x() as f32, touch.client_y() as f32, false); } }) }; @@ -1532,47 +1524,40 @@ pub fn SourceEditor(props: &SourceEditorProps) -> Html { if mouse_button != 0 && mouse_button != 2 { return; } - if let Some(target) = e.target_dyn_into::() { - if let Some(canvas) = canvas_ref.cast::() { - let tag = target.tag_name().to_lowercase(); - if target.is_same_node(Some(&canvas)) || tag == "svg" { - e.prevent_default(); - e.stop_propagation(); - let mut editor_state = editor_state_ref.borrow_mut(); - if e.button() == 0 { - // left button - if editor_state.selection.is_selecting { - editor_state.selection.reset_selection(); - } else { - // selection area mode - if let Some(rect_el) = canvas_ref.cast::() { - let rect = rect_el.get_bounding_client_rect(); - let mouse_x = e.client_x() as f32 - rect.left() as f32; - let mouse_y = e.client_y() as f32 - rect.top() as f32; - if e.ctrl_key() { - editor_state.selection.with_selecting_start_and_rect( - true, - (mouse_x, mouse_y), - Some((mouse_x, mouse_y, 0.0, 0.0)), - ); - } else { - editor_state.selection.with_selecting_start_rect_and_clear_blocks( - true, - (mouse_x, mouse_y), - Some((mouse_x, mouse_y, 0.0, 0.0)), - ); - } - } - } - } else if e.button() == 2 { - // right button - editor_state.selection.reset_selection(); - // Right button panning - cursor_grabbing.set(true); - editor_state.is_panning = true; - editor_state.pan_start = (e.client_x() as f32, e.client_y() as f32); + if is_canvas_background( + e.target_dyn_into::().as_ref(), + canvas_ref.cast::().as_ref(), + ) { + e.prevent_default(); + e.stop_propagation(); + let mut editor_state = editor_state_ref.borrow_mut(); + if e.button() == 0 { + // left button + if editor_state.selection.is_selecting { + editor_state.selection.reset_selection(); + } else if let Some(rect_el) = canvas_ref.cast::() { + // selection area mode + let rect = rect_el.get_bounding_client_rect(); + let mouse_x = e.client_x() as f32 - rect.left() as f32; + let mouse_y = e.client_y() as f32 - rect.top() as f32; + if e.ctrl_key() { + editor_state.selection.with_selecting_start_and_rect( + true, + (mouse_x, mouse_y), + Some((mouse_x, mouse_y, 0.0, 0.0)), + ); + } else { + editor_state.selection.with_selecting_start_rect_and_clear_blocks( + true, + (mouse_x, mouse_y), + Some((mouse_x, mouse_y, 0.0, 0.0)), + ); } } + } else if e.button() == 2 { + // right button panning + start_canvas_pan(&mut editor_state, e.client_x() as f32, e.client_y() as f32); + cursor_grabbing.set(true); } } }) @@ -1672,21 +1657,9 @@ pub fn SourceEditor(props: &SourceEditorProps) -> Html { let is_panning = { editor_state_ref.borrow().is_panning }; if is_panning { - let initial_positions: Vec<(BlockId, Position)> = - { editor_state_ref.borrow().blocks.iter().map(|b| (b.id, b.position)).collect() }; - - { - let mut editor_state = editor_state_ref.borrow_mut(); - let (start_x, start_y) = editor_state.pan_start; - let dx = client_x as f32 - start_x; - let dy = client_y as f32 - start_y; - let (canvas_ox, canvas_oy) = editor_state.canvas_offset; - editor_state.canvas_offset = (canvas_ox + dx, canvas_oy + dy); - editor_state.pan_start = (client_x as f32, client_y as f32); - }; - + let move_params = pan_canvas(&mut editor_state_ref.borrow_mut(), client_x as f32, client_y as f32); // Keep panning smooth by moving already-rendered nodes directly. - move_blocks.emit((0.0, 0.0, (0.0, 0.0), initial_positions)); + move_blocks.emit(move_params); return; } @@ -1782,34 +1755,7 @@ pub fn SourceEditor(props: &SourceEditorProps) -> Html { } } - let to_move = { - let editor_state = editor_state_ref.borrow(); - // Update dragging block (Single or Group) - if let Some(block_id) = editor_state.drag.block_id { - // If the dragged block is member of a selection -> move group - if editor_state.drag.dragging_group.contains(&block_id) - && !editor_state.selection.group_initial_positions.is_empty() - { - Some(( - mouse_x, - mouse_y, - editor_state.selection.group_anchor_mouse, - editor_state.selection.group_initial_positions.clone(), - )) - } else { - // Single drag block - if let Some(block) = editor_state.get_block(block_id) { - let positions = vec![(block_id, block.position)]; - Some((mouse_x, mouse_y, editor_state.selection.group_anchor_mouse, positions)) - } else { - None - } - } - } else { - None - } - }; - if let Some(move_it) = to_move { + if let Some(move_it) = compute_drag_move_params(&editor_state_ref.borrow(), mouse_x, mouse_y) { move_blocks.emit(move_it); // Drag updates are applied directly to DOM for smoothness. // Avoid full re-render on every mouse move while dragging blocks. @@ -1824,11 +1770,11 @@ pub fn SourceEditor(props: &SourceEditorProps) -> Html { }) }; - let handle_canvas_mouse_up = { + let end_active_interaction = { let editor_state_ref = editor_state_ref.clone(); let cursor_grabbing = cursor_grabbing.clone(); let force_update = force_update.clone(); - Callback::from(move |_e: MouseEvent| { + Rc::new(move || { let mut editor_state = editor_state_ref.borrow_mut(); let had_active_interaction = clear_active_interaction(&mut editor_state); cursor_grabbing.set(false); @@ -1838,6 +1784,11 @@ pub fn SourceEditor(props: &SourceEditorProps) -> Html { }) }; + let handle_canvas_mouse_up = { + let end_active_interaction = end_active_interaction.clone(); + Callback::from(move |_e: MouseEvent| end_active_interaction()) + }; + let handle_canvas_touch_start = { let editor_state_ref = editor_state_ref.clone(); let canvas_ref = canvas_ref.clone(); @@ -1861,19 +1812,15 @@ pub fn SourceEditor(props: &SourceEditorProps) -> Html { } return; } - if let Some(target) = e.target_dyn_into::() { - if let Some(canvas) = canvas_ref.cast::() { - let tag = target.tag_name().to_lowercase(); - if target.is_same_node(Some(&canvas)) || tag == "svg" { - if let Some(touch) = e.touches().item(0) { - e.stop_propagation(); - let mut editor_state = editor_state_ref.borrow_mut(); - editor_state.selection.reset_selection(); - cursor_grabbing.set(true); - editor_state.is_panning = true; - editor_state.pan_start = (touch.client_x() as f32, touch.client_y() as f32); - } - } + if is_canvas_background( + e.target_dyn_into::().as_ref(), + canvas_ref.cast::().as_ref(), + ) { + if let Some(touch) = e.touches().item(0) { + e.stop_propagation(); + let mut editor_state = editor_state_ref.borrow_mut(); + start_canvas_pan(&mut editor_state, touch.client_x() as f32, touch.client_y() as f32); + cursor_grabbing.set(true); } } }) @@ -1920,52 +1867,23 @@ pub fn SourceEditor(props: &SourceEditorProps) -> Html { } if let Some(touch) = e.touches().item(0) { - let (is_panning, drag_block_id, group_anchor_mouse, group_initial_positions, single_block_position) = { - let editor_state = editor_state_ref.borrow(); - let drag_block_id = editor_state.drag.block_id; - let single_block_position = - drag_block_id.and_then(|block_id| editor_state.get_block(block_id).map(|block| block.position)); - ( - editor_state.is_panning, - drag_block_id, - editor_state.selection.group_anchor_mouse, - editor_state.selection.group_initial_positions.clone(), - single_block_position, - ) - }; + let client_x = touch.client_x() as f32; + let client_y = touch.client_y() as f32; + + let is_panning = editor_state_ref.borrow().is_panning; if is_panning { e.stop_propagation(); - - let initial_positions: Vec<(BlockId, Position)> = - { editor_state_ref.borrow().blocks.iter().map(|b| (b.id, b.position)).collect() }; - - { - let mut editor_state = editor_state_ref.borrow_mut(); - let (start_x, start_y) = editor_state.pan_start; - let dx = touch.client_x() as f32 - start_x; - let dy = touch.client_y() as f32 - start_y; - let (canvas_ox, canvas_oy) = editor_state.canvas_offset; - editor_state.canvas_offset = (canvas_ox + dx, canvas_oy + dy); - editor_state.pan_start = (touch.client_x() as f32, touch.client_y() as f32); - } - - move_blocks.emit((0.0, 0.0, (0.0, 0.0), initial_positions)); - } else if let Some(block_id) = drag_block_id { + let move_params = pan_canvas(&mut editor_state_ref.borrow_mut(), client_x, client_y); + move_blocks.emit(move_params); + } else if editor_state_ref.borrow().drag.block_id.is_some() { e.stop_propagation(); if let Some(canvas) = canvas_ref.cast::() { let rect = canvas.get_bounding_client_rect(); - let touch_x = touch.client_x() as f32 - rect.left() as f32; - let touch_y = touch.client_y() as f32 - rect.top() as f32; + let touch_x = client_x - rect.left() as f32; + let touch_y = client_y - rect.top() as f32; - let to_move = if group_initial_positions.is_empty() { - single_block_position - .map(|position| (touch_x, touch_y, group_anchor_mouse, vec![(block_id, position)])) - } else { - Some((touch_x, touch_y, group_anchor_mouse, group_initial_positions)) - }; - - if let Some(move_it) = to_move { + if let Some(move_it) = compute_drag_move_params(&editor_state_ref.borrow(), touch_x, touch_y) { move_blocks.emit(move_it); } else { force_update.set(*force_update + 1); @@ -1977,20 +1895,8 @@ pub fn SourceEditor(props: &SourceEditorProps) -> Html { }; let handle_canvas_touch_end = { - let editor_state_ref = editor_state_ref.clone(); - let cursor_grabbing = cursor_grabbing.clone(); - let force_update = force_update.clone(); - Callback::from(move |_e: TouchEvent| { - let mut editor_state = editor_state_ref.borrow_mut(); - if editor_state.pinch_distance.take().is_some() { - force_update.set(*force_update + 1); - } - let had_active_interaction = clear_active_interaction(&mut editor_state); - cursor_grabbing.set(false); - if had_active_interaction { - force_update.set(*force_update + 1); - } - }) + let end_active_interaction = end_active_interaction.clone(); + Callback::from(move |_e: TouchEvent| end_active_interaction()) }; let handle_canvas_wheel = { @@ -2032,62 +1938,28 @@ pub fn SourceEditor(props: &SourceEditorProps) -> Html { }) }; - // Ensure interaction state is cleaned up even when mouseup happens outside the canvas. + // Ensure interaction state is cleaned up even when mouseup or touchend/touchcancel happens outside the canvas. { - let editor_state_ref = editor_state_ref.clone(); - let cursor_grabbing = cursor_grabbing.clone(); - let force_update = force_update.clone(); + let end_active_interaction = end_active_interaction.clone(); use_effect(move || { - let handler = Closure::wrap(Box::new(move |_event: MouseEvent| { - let mut editor_state = editor_state_ref.borrow_mut(); - let had_active_interaction = clear_active_interaction(&mut editor_state); - if !had_active_interaction { - return; - } - cursor_grabbing.set(false); - force_update.set(*force_update + 1); + let handler = Closure::wrap(Box::new(move |_event: web_sys::Event| { + end_active_interaction(); }) as Box); + const EVENTS: [&str; 3] = ["mouseup", "touchend", "touchcancel"]; if let Some(browser_window) = window() { - let _ = browser_window.add_event_listener_with_callback("mouseup", handler.as_ref().unchecked_ref()); - } - - move || { - if let Some(browser_window) = window() { + for event_name in EVENTS { let _ = - browser_window.remove_event_listener_with_callback("mouseup", handler.as_ref().unchecked_ref()); + browser_window.add_event_listener_with_callback(event_name, handler.as_ref().unchecked_ref()); } } - }); - } - - { - let editor_state_ref = editor_state_ref.clone(); - let cursor_grabbing = cursor_grabbing.clone(); - let force_update = force_update.clone(); - use_effect(move || { - let handler = Closure::wrap(Box::new(move |_event: TouchEvent| { - let mut editor_state = editor_state_ref.borrow_mut(); - let had_active_interaction = clear_active_interaction(&mut editor_state); - if !had_active_interaction { - return; - } - cursor_grabbing.set(false); - force_update.set(*force_update + 1); - }) as Box); - - if let Some(browser_window) = window() { - let _ = browser_window.add_event_listener_with_callback("touchend", handler.as_ref().unchecked_ref()); - let _ = - browser_window.add_event_listener_with_callback("touchcancel", handler.as_ref().unchecked_ref()); - } move || { if let Some(browser_window) = window() { - let _ = browser_window - .remove_event_listener_with_callback("touchend", handler.as_ref().unchecked_ref()); - let _ = browser_window - .remove_event_listener_with_callback("touchcancel", handler.as_ref().unchecked_ref()); + for event_name in EVENTS { + let _ = browser_window + .remove_event_listener_with_callback(event_name, handler.as_ref().unchecked_ref()); + } } } }); @@ -2576,4 +2448,116 @@ mod tests { fn initial_layout_view_matches_manual_layout_origin() { assert_eq!(initial_layout_view_transform(), ((0.0, 0.0), 1.0)); } + + #[test] + fn start_canvas_pan_and_pan_canvas_update_offsets_and_positions() { + let mut state = EditorState { + blocks: vec![ + Block { + id: 1, + block_type: BlockType::InputM3u, + position: (10.0, 20.0), + instance: create_instance(BlockType::InputM3u), + }, + Block { + id: 2, + block_type: BlockType::InputM3u, + position: (30.0, 40.0), + instance: create_instance(BlockType::InputM3u), + }, + ], + ..EditorState::default() + }; + + start_canvas_pan(&mut state, 100.0, 200.0); + assert!(state.is_panning); + assert_eq!(state.pan_start, (100.0, 200.0)); + + let params = pan_canvas(&mut state, 150.0, 260.0); + assert_eq!(state.canvas_offset, (50.0, 60.0)); + assert_eq!(state.pan_start, (150.0, 260.0)); + assert_eq!(params.3, vec![(1, (10.0, 20.0)), (2, (30.0, 40.0))]); + } + + #[test] + fn start_block_drag_handles_single_drag_and_group_drag_params() { + let mut state = EditorState { + blocks: vec![ + Block { + id: 1, + block_type: BlockType::InputM3u, + position: (100.0, 100.0), + instance: create_instance(BlockType::InputM3u), + }, + Block { + id: 2, + block_type: BlockType::InputM3u, + position: (200.0, 200.0), + instance: create_instance(BlockType::InputM3u), + }, + ], + ..EditorState::default() + }; + + // Start dragging block 1 without ctrl (touch or regular click) + start_block_drag(&mut state, 1, (110.0, 115.0), false); + assert_eq!(state.drag.block_id, Some(1)); + assert_eq!(state.drag.drag_offset, (10.0, 15.0)); + assert!(state.selection.selected_blocks.contains(&1)); + assert_eq!(state.selection.group_anchor_mouse, (110.0, 115.0)); + assert_eq!(state.selection.group_initial_positions, vec![(1, (100.0, 100.0))]); + + // Computing drag move parameters + let drag_params = compute_drag_move_params(&state, 150.0, 160.0); + assert_eq!(drag_params, Some((150.0, 160.0, (110.0, 115.0), vec![(1, (100.0, 100.0))]))); + } + + #[test] + fn start_block_drag_handles_ctrl_selection_toggle() { + let mut state = EditorState { + blocks: vec![ + Block { + id: 1, + block_type: BlockType::InputM3u, + position: (100.0, 100.0), + instance: create_instance(BlockType::InputM3u), + }, + Block { + id: 2, + block_type: BlockType::InputM3u, + position: (200.0, 200.0), + instance: create_instance(BlockType::InputM3u), + }, + ], + ..EditorState::default() + }; + + // Select block 1 + start_block_drag(&mut state, 1, (105.0, 105.0), false); + assert_eq!(state.selection.selected_blocks, HashSet::from([1])); + + // Ctrl-click block 2 adds it to selection + start_block_drag(&mut state, 2, (205.0, 205.0), true); + assert_eq!(state.selection.selected_blocks, HashSet::from([1, 2])); + + // Ctrl-click block 1 removes it from selection + start_block_drag(&mut state, 1, (105.0, 105.0), true); + assert_eq!(state.selection.selected_blocks, HashSet::from([2])); + } + + #[test] + fn clear_active_interaction_resets_panning_dragging_selection_and_pinch() { + let mut state = EditorState { is_panning: true, pinch_distance: Some(42.0), ..EditorState::default() }; + state.drag.block_id = Some(1); + state.selection.is_selecting = true; + + assert!(clear_active_interaction(&mut state)); + assert!(!state.is_panning); + assert!(state.drag.block_id.is_none()); + assert!(!state.selection.is_selecting); + assert!(state.pinch_distance.is_none()); + + // Calling again with no active interaction returns false + assert!(!clear_active_interaction(&mut state)); + } }