diff --git a/CHANGELOG.md b/CHANGELOG.md index 9e22820ea..343e03416 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,17 @@ ## ⚠️ Breaking Changes +- **`playlist.watch.changed` carries every watch change of one update.** The payload is now + `{ target, changes: [...] }`, where each change has a `kind` (`groups_added`, `groups_removed`, `channels_added`, + `channels_removed`). Templates using `{{watch.group}}`, `{{watch.added}}` or `{{watch.removed}}` must be updated: + + ```handlebars + {{!-- before --}} + {{watch.group}}: {{#each watch.added}}{{this}} {{/each}} + {{!-- after --}} + {{#each watch.channels_added}}{{group}}: {{#each items}}{{this}} {{/each}}{{/each}} + ``` + - **`clear_invalid_epg_ids` moved to `epg_output.clear_invalid_ids`.** The target option now sits next to the other EPG id output options. The top-level `options.clear_invalid_epg_ids` and its legacy alias `options.required_epg` are rejected while loading. To migrate, move the value: @@ -317,16 +328,12 @@ layer to stop discarding the distinction: an empty `Some` is now load-bearing and means "configured and unusable", which is not the same as "not configured". `playlist.watch.unmatched` covers the fourth silence — a pattern matching no group looks exactly like a group that has not changed, so a typo in `watch` was invisible. - - **`playlist.groups.changed`**: `watch` tracked channels inside named groups and was blind to the group set itself, - in both directions. A group appearing was silent — no baseline file, so one was written, nothing was emitted, and - the group's entire channel list read as "not new" from then on — and a group vanishing was worse, since it is absent - from the refreshed playlist and no code path observed the disappearance at all. The target's group titles are now - diffed against a persisted index before the per-group fan-out, so it sees every group rather than only the ones the - watch patterns name; the question is which groups exist, not what is inside the watched ones. The index sits beside - the per-group directory (`.groups.bin`, not `/__groups.bin`) so it cannot collide with a group whose - sanitized title matches, and first sight writes the baseline silently — announcing every existing group as new on - the first refresh after an upgrade would be noise. Gated on `target.watch` being configured, so it costs nothing for - targets that never asked to be watched. + - **Watch group contexts**: a `watch` entry can be a map with `pattern` and `on` (`channels`, `group_added`, + `group_removed`), so groups appearing in or vanishing from a target are reported, filtered per entry. Plain string + entries keep watching channels only. The target's group titles are diffed against a persisted index + (`.groups.bin`) that always holds every group, so editing a pattern never reads as a change; first sight + writes the baseline silently. Group changes arrive in `playlist.watch.changed`; the short-lived + `playlist.groups.changed` event is gone and its name is accepted as an alias. - **`metadata.update.failed` for an input whose tasks burn through their retries**: the completion event only fires when a cycle drains *with changes*, and an exhausted task only reached a `debug!`, so an input whose resolves fail every time emitted a start and then nothing for as long as it stayed broken — on the bus, indistinguishable from one @@ -1094,6 +1101,10 @@ ## 🐛 Fixes +- **Long Telegram and ntfy notifications are no longer rejected.** Telegram accepts at most 4096 characters (UTF-16 + units) and ntfy 4096 bytes per message; neither channel had a limit, so a large watch or statistics notification + failed and landed in the dead letter queue. Telegram now cuts the final text, after `MarkdownV2` escaping, and drops + an escape that the cut separated from its character; ntfy bodies are truncated like Pushover's. - **Short EPG and player guides stay empty when playlist and XMLTV ids differ only in case.** A live channel whose EPG id matches a guide channel case-insensitively (for example `usa4k.us` against ``) now adopts the guide's spelling, so the playlist `tvg-id` / `epg_channel_id`, the XMLTV `` and the EPG db @@ -1732,6 +1743,8 @@ ## 🛠 Maintenance +- **`ClusterFlags` is built on `create_bitset!`.** It was the last user of the `bitflags` crate, which is removed + from `shared`. The API and the persisted `[live,vod]` form are unchanged. - **Testkit scenarios can pause and tune reconnect windows.** A step with only `pause_millis` (1 to 300000) idles the scenario, and `policy_contract` accepts `hls_session_ttl_secs` and `provider_affinity_ttl_secs`. The new scenario `m3u-hls-provider-affinity-after-lease-expiry` uses both to prove that a returning HLS playback stays on its diff --git a/Cargo.lock b/Cargo.lock index 19e19fd84..3e530811d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3799,7 +3799,6 @@ version = "3.3.147" dependencies = [ "aes", "base64 0.23.1", - "bitflags 2.13.1", "blake3", "brotli", "bytes", diff --git a/README.md b/README.md index 4a5f285d0..ea9a7fa96 100644 --- a/README.md +++ b/README.md @@ -328,6 +328,8 @@ Notification capabilities include: - Preview/test endpoint so templates can be rendered without sending them - Events for provider failures, exhausted pools, priority fallback, user connection denial, auth activity, config reload failures, metadata failures, scheduled-task failures, recording lifecycle, disk alerts, and more +- Playlist change tracking: get told when channels in the groups you care about change, or when whole groups appear + or disappear — for example a new event or PPV group — each filtered by your own group patterns - Event statistics and a recent-event ring for troubleshooting - Disk usage alerts with warning/critical thresholds and configurable repeat intervals diff --git a/backend/app/src/api/endpoints/v1_api_config.rs b/backend/app/src/api/endpoints/v1_api_config.rs index 673cbae54..9bc148e40 100644 --- a/backend/app/src/api/endpoints/v1_api_config.rs +++ b/backend/app/src/api/endpoints/v1_api_config.rs @@ -1254,7 +1254,7 @@ mod tests { processing: Some(String::from("group ~ 'secret'")), ..Default::default() }, - watch: Some(vec![String::from("watch-expr")]), + watch: Some(vec![shared::model::ConfigWatchDto::channels("watch-expr")]), mapping: Some(vec![String::from("mapping-name")]), ..Default::default() }], diff --git a/backend/app/src/api/endpoints/websocket_api.rs b/backend/app/src/api/endpoints/websocket_api.rs index 0e9c15091..569c8d49f 100644 --- a/backend/app/src/api/endpoints/websocket_api.rs +++ b/backend/app/src/api/endpoints/websocket_api.rs @@ -516,7 +516,6 @@ fn to_protocol_message(event: EventMessage) -> Option<(ProtocolMessage, &'static | EventMessage::DiskAlert(_) | EventMessage::ConfigReloadFailed(_) | EventMessage::PlaylistWatchChanged(_) - | EventMessage::PlaylistGroupsChanged(_) | EventMessage::PlaylistWatchDisabled(_) | EventMessage::PlaylistWatchUnmatched(_) | EventMessage::RecordingLifecycle(_) @@ -704,9 +703,9 @@ mod tests { use shared::model::{ AuthAuditEvent, AuthAuditOutcome, Claims, ConfigReloadFailure, DiskAlert, DiskAlertLevel, LibraryScanProgressEvent, LibraryScanSummary, LibraryScanSummaryStatus, MetadataUpdateFailure, MsgKind, - Permission, PlaylistGroupsChanged, PlaylistUpdateProgressEvent, PlaylistUpdateRunId, PlaylistUpdateRunOrder, - PlaylistUpdateState, PlaylistUpdateSummary, ProtocolHandler, ProtocolHandlerMemory, ProtocolMessage, - ProviderAccountEvent, ProviderAccountState, ProviderFailureKind, ProviderFetchFailure, ProviderPoolExhausted, + Permission, PlaylistUpdateProgressEvent, PlaylistUpdateRunId, PlaylistUpdateRunOrder, PlaylistUpdateState, + PlaylistUpdateSummary, ProtocolHandler, ProtocolHandlerMemory, ProtocolMessage, ProviderAccountEvent, + ProviderAccountState, ProviderFailureKind, ProviderFetchFailure, ProviderPoolExhausted, ProviderPriorityFallback, RecordingLifecycleMessage, RoleSet, UserId, UserRole, WatchChanges, WatchDisabled, WatchDisabledReason, WatchUnmatched, CURRENT_PERMISSION_SCHEMA_VERSION, PERM_ALL, PROTOCOL_VERSION, TOKEN_NO_AUTH, @@ -745,13 +744,12 @@ mod tests { // every notification-only kind against `expected = true` and the // test failed on `DiskAlert` - it has been red since the lifecycle // events joined the bus. - const NOT_ON_THE_WIRE: [EventKind; 29] = [ + const NOT_ON_THE_WIRE: [EventKind; 28] = [ EventKind::ConnectionDenied, EventKind::ScheduledTaskFailed, EventKind::ProviderFetchFailed, EventKind::ProviderPoolExhausted, EventKind::ProviderPriorityFallback, - EventKind::PlaylistGroupsChanged, EventKind::PlaylistWatchDisabled, EventKind::PlaylistWatchUnmatched, EventKind::NotificationDeadLettered, @@ -893,13 +891,7 @@ mod tests { paths: "config.yml".to_string(), error: "boom".to_string(), }), - EventMessage::PlaylistWatchChanged(WatchChanges::new( - "t".to_string(), - "g".to_string(), - Vec::new(), - Vec::new(), - )), - EventMessage::PlaylistGroupsChanged(PlaylistGroupsChanged::new("t".to_string(), Vec::new(), Vec::new())), + EventMessage::PlaylistWatchChanged(WatchChanges::new("t".to_string(), Vec::new())), EventMessage::PlaylistWatchDisabled(WatchDisabled::new( "t".to_string(), WatchDisabledReason::InvalidPatterns, diff --git a/backend/app/src/api/tasks/notification_bridge.rs b/backend/app/src/api/tasks/notification_bridge.rs index aa4d3b8dc..36b1e95ce 100644 --- a/backend/app/src/api/tasks/notification_bridge.rs +++ b/backend/app/src/api/tasks/notification_bridge.rs @@ -23,10 +23,10 @@ use log::{debug, warn}; use shared::model::{ notification::{EventId, Severity}, AuthAuditEvent, AuthAuditOutcome, ConnectionDenied, EventKind, EventKindMask, EventMessage, - LibraryScanSummaryStatus, MetadataUpdateFailure, PlaylistGroupsChanged, ProviderFetchFailure, - ProviderPoolExhausted, ProviderPriorityFallback, ScheduledTaskFailure, ServerLifecycleEvent, ServerLifecycleState, - StreamProbeFailure, StreamProbeFailureReason, UserLifecycleEvent, UserLifecycleState, WatchDisabled, - WatchDisabledReason, WatchUnmatched, + LibraryScanSummaryStatus, MetadataUpdateFailure, ProviderFetchFailure, ProviderPoolExhausted, + ProviderPriorityFallback, ScheduledTaskFailure, ServerLifecycleEvent, ServerLifecycleState, StreamProbeFailure, + StreamProbeFailureReason, UserLifecycleEvent, UserLifecycleState, WatchDisabled, WatchDisabledReason, + WatchUnmatched, }; use std::{fmt::Write, sync::Arc}; use tokio::sync::broadcast::error::RecvError; @@ -77,7 +77,6 @@ const NOTIFIABLE_KINDS: EventKindMask = EventKindMask::new() .with(EventKind::DiskAlert) .with(EventKind::ConfigReloadFailed) .with(EventKind::PlaylistWatchChanged) - .with(EventKind::PlaylistGroupsChanged) .with(EventKind::PlaylistWatchDisabled) .with(EventKind::PlaylistWatchUnmatched) .with(EventKind::ProviderAccountStatus) @@ -210,17 +209,16 @@ pub fn to_notification(message: &EventMessage) -> Option { Some(NotificationEvent::new(id, title.clone(), title)) } - // The three that already have a `MessageContent` shape reuse it: - // `from_content` is what built their title, body and template fields - // before, and rebuilding that here by hand would be a second copy - // free to drift from the templates that render it. + // Events that already have a `MessageContent` shape reuse its + // builders: `from_content` is what built their title, body and + // template fields before, and rebuilding that here by hand would be a + // second copy free to drift from the templates that render it. Watch + // changes use the same builders through a borrowing entry point, so + // the title lists are not cloned per notification. EventMessage::DiskAlert(alert) => { Some(NotificationEvent::from_content(&MessageContent::DiskAlert(alert.clone()))) } - EventMessage::PlaylistWatchChanged(changes) => { - Some(NotificationEvent::from_content(&MessageContent::Watch(changes.clone()))) - } - EventMessage::PlaylistGroupsChanged(changes) => Some(groups_changed_notification(id, changes)), + EventMessage::PlaylistWatchChanged(changes) => Some(NotificationEvent::from_watch_changes(changes)), EventMessage::PlaylistWatchDisabled(disabled) => { Some(watch_disabled_notification(id, disabled, message.severity())) } @@ -312,34 +310,6 @@ fn metadata_failure_notification( NotificationEvent::new(id, title, body).with_severity(severity).with_fields(failure) } -fn groups_changed_notification(id: EventId, changes: &PlaylistGroupsChanged) -> NotificationEvent { - let title = - format!("{} group(s) added, {} removed in {}", changes.added_total, changes.removed_total, changes.target); - let mut body = title.clone(); - push_sampled_list(&mut body, "Added", &changes.added, changes.added_total); - push_sampled_list(&mut body, "Removed", &changes.removed, changes.removed_total); - NotificationEvent::new(id, title, body).with_fields(changes) -} - -/// Render one direction of a sampled change set. -/// -/// The lists carry titles only and may be shorter than the total - or empty - -/// so the omission is spelled out rather than smuggled into the list. -fn push_sampled_list(body: &mut String, label: &str, names: &[String], total: usize) { - if total == 0 { - return; - } - let _ = write!(body, "\n\n{label} ({total}):"); - for name in names { - body.push('\n'); - body.push_str(name); - } - let omitted = total.saturating_sub(names.len()); - if omitted > 0 { - let _ = write!(body, "\n... {omitted} more not listed"); - } -} - fn watch_disabled_notification(id: EventId, disabled: &WatchDisabled, severity: Severity) -> NotificationEvent { let what = match disabled.reason { WatchDisabledReason::InvalidPatterns => "no watch pattern compiled".to_string(), @@ -501,12 +471,11 @@ mod tests { notification::{registry, Severity}, ActiveUserConnectionChange, AuthAuditEvent, AuthAuditOutcome, ConfigReloadFailure, ConfigType, ConnectionDenied, DiskAlert, DiskAlertLevel, EventKind, LibraryScanProgressEvent, LibraryScanSummary, - LibraryScanSummaryStatus, MetadataUpdateFailure, MsgKind, NotificationDeadLetter, PlaylistGroupsChanged, - PlaylistUpdateProgressEvent, PlaylistUpdateState, PlaylistUpdateSummary, ProviderAccountEvent, - ProviderAccountState, ProviderFailureKind, ProviderFetchFailure, ProviderPoolExhausted, - ProviderPriorityFallback, RecordingLifecycleMessage, ScheduledTaskFailure, ServerLifecycleEvent, - StreamProbeFailure, StreamProbeFailureReason, SystemInfo, UserLifecycleEvent, UserLifecycleState, WatchChanges, - WatchDisabled, WatchDisabledReason, WatchUnmatched, + LibraryScanSummaryStatus, MetadataUpdateFailure, MsgKind, NotificationDeadLetter, PlaylistUpdateProgressEvent, + PlaylistUpdateState, PlaylistUpdateSummary, ProviderAccountEvent, ProviderAccountState, ProviderFailureKind, + ProviderFetchFailure, ProviderPoolExhausted, ProviderPriorityFallback, RecordingLifecycleMessage, + ScheduledTaskFailure, ServerLifecycleEvent, StreamProbeFailure, StreamProbeFailureReason, SystemInfo, + UserLifecycleEvent, UserLifecycleState, WatchChanges, WatchDisabled, WatchDisabledReason, WatchUnmatched, }; use std::sync::Arc; @@ -606,13 +575,7 @@ mod tests { paths: "config.yml".to_string(), error: "boom".to_string(), }), - EventMessage::PlaylistWatchChanged(WatchChanges::new( - "t".to_string(), - "g".to_string(), - Vec::new(), - Vec::new(), - )), - EventMessage::PlaylistGroupsChanged(PlaylistGroupsChanged::new("t".to_string(), Vec::new(), Vec::new())), + EventMessage::PlaylistWatchChanged(WatchChanges::new("t".to_string(), Vec::new())), EventMessage::PlaylistWatchDisabled(WatchDisabled::new( "t".to_string(), WatchDisabledReason::InvalidPatterns, diff --git a/backend/core/src/model/config/target.rs b/backend/core/src/model/config/target.rs index f900e0af5..bf1b86976 100644 --- a/backend/core/src/model/config/target.rs +++ b/backend/core/src/model/config/target.rs @@ -11,7 +11,7 @@ use shared::{ model::{ ConfigTargetDto, ConfigTargetOptions, DeduplicateConfig, HdHomeRunTargetOutputDto, M3uTargetOutputDto, MappingStage, PlaylistItemType, ProcessingOrder, StrmExportStyle, StrmTargetOutputDto, TargetOutputDto, - TargetType, TraktConfigDto, XtreamTargetOutputDto, + TargetType, TraktConfigDto, WatchContext, WatchContextSet, XtreamTargetOutputDto, }, }; use std::{borrow::Cow, sync::Arc}; @@ -319,6 +319,28 @@ impl From<&TargetOutput> for TargetOutputDto { } } +/// One compiled `watch` entry. +/// +/// `contexts` is the DTO's `u8` bitset, copied as is: "does this entry apply +/// to X" is asked for every group, and a bit test is all it costs. +#[derive(Debug, Clone)] +pub struct ConfigWatch { + pub pattern: Arc, + pub contexts: WatchContextSet, +} + +impl ConfigWatch { + /// Whether matching no current group means something is wrong. + /// + /// An entry waiting for a group to appear matches nothing by design, and a + /// `group_removed`-only entry is matched against the previous titles, not + /// the current playlist. + #[must_use] + pub fn reports_unmatched(&self) -> bool { + self.contexts.contains(WatchContext::Channels) && !self.contexts.contains(WatchContext::GroupAdded) + } +} + #[derive(Debug, Clone)] pub struct ConfigTarget { pub id: u16, @@ -335,7 +357,7 @@ pub struct ConfigTarget { pub favourites: Option>, pub processing_order: ProcessingOrder, pub execution_plan: TargetExecutionPlan, - pub watch: Option>>, + pub watch: Option>, pub use_memory_cache: bool, } @@ -414,10 +436,10 @@ impl From<&ConfigTargetDto> for ConfigTarget { // `playlist.watch.disabled`. watch: dto.watch.as_ref().map(|list| { list.iter() - .filter_map(|s| match shared::model::REGEX_CACHE.get_or_compile(s) { - Ok(re) => Some(re), + .filter_map(|entry| match shared::model::REGEX_CACHE.get_or_compile(&entry.pattern) { + Ok(pattern) => Some(ConfigWatch { pattern, contexts: entry.on }), Err(e) => { - log::warn!("Invalid watch regex pattern '{s}': {e}"); + log::warn!("Invalid watch regex pattern '{}': {e}", entry.pattern); None } }) @@ -432,6 +454,27 @@ impl From<&ConfigTargetDto> for ConfigTarget { mod tests { use super::*; + #[test] + fn watch_contexts_compile_to_flags() { + let dto = ConfigTargetDto { + name: "t".to_string(), + watch: Some(vec![ + shared::model::ConfigWatchDto::channels("^A"), + shared::model::ConfigWatchDto::new( + "^B", + [shared::model::WatchContext::GroupAdded, shared::model::WatchContext::Channels], + ), + ]), + ..ConfigTargetDto::default() + }; + let target = ConfigTarget::from(&dto); + let watch = target.watch.expect("watch"); + assert_eq!(watch[0].contexts, dto.watch.as_ref().unwrap()[0].on); + assert!(watch[0].reports_unmatched()); + assert!(watch[1].contexts.contains(WatchContext::GroupAdded)); + assert!(!watch[1].reports_unmatched()); + } + #[test] fn effective_curation_retains_canonical_owner_and_credentials_without_legacy_output() { let mut dto: ConfigTargetDto = serde_json::from_value(serde_json::json!({"name": "discovery", "output": [{"type": "m3u"}], diff --git a/backend/core/src/model/notification.rs b/backend/core/src/model/notification.rs index 67ce04c96..540bf1bc1 100644 --- a/backend/core/src/model/notification.rs +++ b/backend/core/src/model/notification.rs @@ -15,7 +15,7 @@ use chrono::Utc; use shared::{ model::{ notification::{registry, EventId, Severity}, - DiskAlert, MsgKind, + DiskAlert, MsgKind, WatchChange, WatchChanges, WatchItems, }, utils::human_readable_byte_size, }; @@ -113,6 +113,17 @@ impl NotificationEvent { event.fields = fields_for(content); event } + + /// A watch notification from a borrowed payload. + /// + /// The event bus hands the bridge a `&WatchChanges`; going through + /// [`Self::from_content`] would mean cloning every title list into a + /// `MessageContent` just to read it back. + pub fn from_watch_changes(changes: &WatchChanges) -> Self { + let mut event = Self::new(registry::PLAYLIST_WATCH_CHANGED, watch_title(changes), watch_body(changes)); + event.fields = serde_json::to_value(changes).unwrap_or(serde_json::Value::Null); + event + } } /// Canonical id for a legacy message. @@ -181,11 +192,7 @@ pub fn recording_lifecycle_text(recording: &RecordingLifecycleMessage) -> String fn title_for(content: &MessageContent) -> String { match content { MessageContent::Info(s) | MessageContent::Error(s) => first_line(s), - MessageContent::Watch(w) => { - // The totals, not the list lengths: a large change ships a sample - // and the subject line must still report what actually happened. - format!("{} channel(s) added, {} removed in {}/{}", w.added_total, w.removed_total, w.target, w.group) - } + MessageContent::Watch(w) => watch_title(w), MessageContent::ProcessingStats(stats) => match (&stats.stats, &stats.errors) { (None, Some(_)) => "Playlist update failed".to_string(), (Some(sources), _) => { @@ -208,25 +215,72 @@ fn title_for(content: &MessageContent) -> String { } } -/// One direction of a watch change, rendered for a plain-text channel. +/// Subject line for a watch change. /// -/// `names` may be a sample of `total` - or empty, when the change was too -/// large to list - so the omission is spelled out here rather than smuggled -/// into the list itself. -fn push_watch_section(out: &mut String, label: &str, names: &[String], total: usize) { - if total == 0 { +/// The totals, not the list lengths: a large change ships a sample and the +/// subject line must still report what actually happened. +fn watch_title(changes: &WatchChanges) -> String { + let (mut groups_added, mut groups_removed, mut channel_changes) = (0usize, 0usize, 0usize); + for change in &changes.changes { + match change { + WatchChange::GroupsAdded(items) => groups_added += items.total, + WatchChange::GroupsRemoved(items) => groups_removed += items.total, + WatchChange::ChannelsAdded(group) | WatchChange::ChannelsRemoved(group) => { + channel_changes += group.items.total; + } + } + } + let mut title = format!("Watch changes in {}:", changes.target); + let mut separator = " "; + for (count, what) in + [(groups_added, "group(s) added"), (groups_removed, "group(s) removed"), (channel_changes, "channel change(s)")] + { + if count > 0 { + let _ = write!(title, "{separator}{count} {what}"); + separator = ", "; + } + } + title +} + +/// Plain-text body for a watch change: group sections, then channel sections. +fn watch_body(changes: &WatchChanges) -> String { + let mut out = format!("Watch changes in {}.", changes.target); + for change in &changes.changes { + match change { + WatchChange::GroupsAdded(items) => push_watch_section(&mut out, "New groups", items), + WatchChange::GroupsRemoved(items) => push_watch_section(&mut out, "Removed groups", items), + WatchChange::ChannelsAdded(group) => { + push_watch_section(&mut out, format_args!("{} - added", group.group), &group.items); + } + WatchChange::ChannelsRemoved(group) => { + push_watch_section(&mut out, format_args!("{} - removed", group.group), &group.items); + } + } + } + out +} + +/// One sampled list of a watch change, rendered for a plain-text channel. +/// +/// `list.items` may be a sample of `list.total` - or empty, when the change +/// was too large to list - so the omission is spelled out here rather than +/// smuggled into the list itself. The label is `Display` so per-group labels +/// need no temporary `String`. +fn push_watch_section(out: &mut String, label: impl std::fmt::Display, list: &WatchItems) { + if list.total == 0 { return; } - let _ = write!(out, "\n\n{label} ({total}):"); - if names.is_empty() { + let _ = write!(out, "\n\n{label} ({}):", list.total); + if list.items.is_empty() { out.push_str("\n(list suppressed for a large update)"); return; } - for name in names { + for name in &list.items { out.push('\n'); out.push_str(name); } - let omitted = total.saturating_sub(names.len()); + let omitted = list.total.saturating_sub(list.items.len()); if omitted > 0 { let _ = write!(out, "\n... {omitted} more not listed"); } @@ -239,12 +293,7 @@ fn push_watch_section(out: &mut String, label: &str, names: &[String], total: us fn body_for(content: &MessageContent) -> String { match content { MessageContent::Info(s) | MessageContent::Error(s) => s.clone(), - MessageContent::Watch(w) => { - let mut out = format!("Watched group {}/{} changed.", w.target, w.group); - push_watch_section(&mut out, "Added", &w.added, w.added_total); - push_watch_section(&mut out, "Removed", &w.removed, w.removed_total); - out - } + MessageContent::Watch(w) => watch_body(w), MessageContent::ProcessingStats(stats) => { let mut out = String::new(); if let Some(sources) = &stats.stats { @@ -306,7 +355,10 @@ fn first_line(s: &str) -> String { #[cfg(test)] mod tests { use super::{super::messaging::ProcessingStats, *}; - use shared::model::{DiskAlertLevel, InputStats, InputType, PlaylistStats, SourceStats, TargetStats}; + use shared::model::{ + DiskAlertLevel, InputStats, InputType, PlaylistStats, SourceStats, TargetStats, WatchChange, WatchChanges, + WatchGroupItems, WatchItems, + }; fn stats_content() -> MessageContent { let input = InputStats { @@ -372,20 +424,67 @@ mod tests { } } + fn watch_items(names: &[&str]) -> WatchItems { WatchItems::complete(names.iter().map(|n| Arc::from(*n)).collect()) } + #[test] - fn watch_body_lists_added_and_removed_instead_of_dumping_json() { + fn watch_text_lists_groups_then_channels_instead_of_dumping_json() { // Regression: the legacy default text for a watch change was a raw // `serde_json` dump, which is what Pushover received. - let content = MessageContent::Watch(super::super::messaging::WatchChanges::new( - "T".to_string(), - "G".to_string(), - vec!["Channel A".to_string()], - vec!["Channel B".to_string()], + let content = MessageContent::Watch(WatchChanges::new( + "sports".to_string(), + vec![ + WatchChange::GroupsAdded(watch_items(&["Event Bundesliga"])), + WatchChange::GroupsRemoved(watch_items(&["DE Kids"])), + WatchChange::ChannelsAdded(WatchGroupItems { + group: "DE Sport HD".into(), + items: watch_items(&["Sky 5"]), + }), + WatchChange::ChannelsRemoved(WatchGroupItems { + group: "DE Sport HD".into(), + items: watch_items(&["Sky 6"]), + }), + ], )); let event = NotificationEvent::from_content(&content); - assert!(event.body.contains("Channel A"), "body missing added channel: {}", event.body); - assert!(event.body.contains("Channel B"), "body missing removed channel: {}", event.body); - assert!(!event.body.starts_with('{'), "body is still a json dump: {}", event.body); + assert_eq!(event.title, "Watch changes in sports: 1 group(s) added, 1 group(s) removed, 2 channel change(s)"); + assert_eq!( + event.body, + "Watch changes in sports.\n\nNew groups (1):\nEvent Bundesliga\n\nRemoved groups (1):\nDE Kids\ + \n\nDE Sport HD - added (1):\nSky 5\n\nDE Sport HD - removed (1):\nSky 6" + ); + } + + /// The bridge builds watch notifications from a borrowed payload; it must + /// produce exactly what the owned `MessageContent` path produces. + #[test] + fn from_watch_changes_matches_from_content() { + let changes = WatchChanges::new( + "sports".to_string(), + vec![ + WatchChange::GroupsAdded(watch_items(&["New"])), + WatchChange::ChannelsRemoved(WatchGroupItems { group: "G".into(), items: watch_items(&["C"]) }), + ], + ); + let borrowed = NotificationEvent::from_watch_changes(&changes); + let owned = NotificationEvent::from_content(&MessageContent::Watch(changes)); + assert_eq!(borrowed.id, owned.id); + assert_eq!(borrowed.severity, owned.severity); + assert_eq!(borrowed.title, owned.title); + assert_eq!(borrowed.body, owned.body); + assert_eq!(borrowed.fields, owned.fields); + } + + #[test] + fn suppressed_watch_list_says_so() { + let content = MessageContent::Watch(WatchChanges::new( + "t".to_string(), + vec![WatchChange::ChannelsAdded(WatchGroupItems { + group: "G".into(), + items: WatchItems { items: Vec::new(), total: 900, truncated: true }, + })], + )); + let event = NotificationEvent::from_content(&content); + assert!(event.body.contains("G - added (900):\n(list suppressed for a large update)"), "{}", event.body); } #[test] diff --git a/backend/messaging/src/channels/ntfy.rs b/backend/messaging/src/channels/ntfy.rs index 775469514..cb2c111c8 100644 --- a/backend/messaging/src/channels/ntfy.rs +++ b/backend/messaging/src/channels/ntfy.rs @@ -8,6 +8,10 @@ use reqwest::header; use shared::model::notification::{EventId, Severity}; use tuliprox_core::model::{ChannelRouting, NtfyMessagingConfig}; +/// ntfy's default `message-size-limit`. A larger body is turned into an +/// attachment, or rejected when the server has no attachment cache. +const NTFY_BODY_LIMIT: usize = 4096; + #[derive(Clone)] pub struct NtfyChannel { config: NtfyMessagingConfig, @@ -76,12 +80,30 @@ impl NotificationChannel for NtfyChannel { } } - fn capabilities(&self) -> ChannelCapabilities { ChannelCapabilities::default() } + fn capabilities(&self) -> ChannelCapabilities { + ChannelCapabilities { supports_templates: true, max_body_bytes: Some(NTFY_BODY_LIMIT) } + } } #[cfg(test)] mod tests { - use super::{NtfyChannel, Severity}; + use super::{NtfyChannel, Severity, NTFY_BODY_LIMIT}; + use crate::channel::NotificationChannel; + + #[test] + fn the_body_is_capped_at_the_ntfy_message_size_limit() { + let channel = NtfyChannel::new( + tuliprox_core::model::NtfyMessagingConfig { + url: "http://ntfy.local".to_string(), + topic: "t".to_string(), + token: None, + templates: std::collections::HashMap::new(), + routing: tuliprox_core::model::ChannelRouting::default(), + }, + reqwest::Client::new(), + ); + assert_eq!(channel.capabilities().max_body_bytes, Some(NTFY_BODY_LIMIT)); + } #[test] fn severity_maps_onto_the_ntfy_priority_scale() { diff --git a/backend/messaging/src/channels/telegram.rs b/backend/messaging/src/channels/telegram.rs index e44eebf04..108e20349 100644 --- a/backend/messaging/src/channels/telegram.rs +++ b/backend/messaging/src/channels/telegram.rs @@ -12,6 +12,34 @@ use tuliprox_core::{ utils::{telegram_create_instance, telegram_send_message, SendMessageOption, SendMessageParseMode}, }; +/// Telegram's message limit, in UTF-16 code units. It applies after entity +/// parsing, so measuring the raw `MarkdownV2` text is conservative. +const TELEGRAM_MESSAGE_LIMIT: usize = 4096; + +/// Cut `text` to what Telegram accepts. +/// +/// Done here rather than through `ChannelCapabilities::max_body_bytes`: the +/// limit counts UTF-16 units, not bytes, and `MarkdownV2` escaping grows the +/// text after the dispatcher has truncated it. Always borrows. +fn fit_message(text: &str) -> &str { + // A UTF-16 unit is never fewer than one byte, so short text needs no scan. + if text.len() <= TELEGRAM_MESSAGE_LIMIT { + return text; + } + let mut units = 0usize; + for (index, ch) in text.char_indices() { + units += ch.len_utf16(); + if units > TELEGRAM_MESSAGE_LIMIT { + let cut = &text[..index]; + // An odd run of trailing backslashes is an escape cut off from the + // character it escapes; Telegram rejects that as a parse error. + let backslashes = cut.len() - cut.trim_end_matches('\\').len(); + return if backslashes % 2 == 1 { &cut[..cut.len() - 1] } else { cut }; + } + } + text +} + #[derive(Clone)] pub struct TelegramChannel { config: TelegramMessagingConfig, @@ -59,18 +87,29 @@ impl NotificationChannel for TelegramChannel { let mut failed: Vec = Vec::new(); for chat_id in &self.config.chat_ids { let bot = telegram_create_instance(&self.config.bot_token, chat_id); - let result = - telegram_send_message(&self.app_config, &self.client, &bot, &message, options.as_ref()).await; + let result = telegram_send_message( + &self.app_config, + &self.client, + &bot, + fit_message(&message), + options.as_ref(), + ) + .await; let mut delivered = result.delivered; if self.config.markdown && msg.templated && result.parse_error && !delivered { // Template output can carry dynamic fields that break // MarkdownV2. Retry once, escaped. let escaped = escape_markdown_v2(raw); let escaped_options = SendMessageOption { parse_mode: SendMessageParseMode::MarkdownV2 }; - delivered = - telegram_send_message(&self.app_config, &self.client, &bot, &escaped, Some(&escaped_options)) - .await - .delivered; + delivered = telegram_send_message( + &self.app_config, + &self.client, + &bot, + fit_message(&escaped), + Some(&escaped_options), + ) + .await + .delivered; } if delivered { debug!("Notification delivered to Telegram chat {chat_id}"); @@ -89,3 +128,44 @@ impl NotificationChannel for TelegramChannel { fn capabilities(&self) -> ChannelCapabilities { ChannelCapabilities::default() } } + +#[cfg(test)] +mod tests { + use super::{fit_message, TELEGRAM_MESSAGE_LIMIT}; + + #[test] + fn a_short_message_is_borrowed_unchanged() { + let text = "Watch changes in sports."; + assert!(std::ptr::eq(fit_message(text), text)); + } + + #[test] + fn a_long_message_is_cut_to_the_limit() { + let text = "a".repeat(TELEGRAM_MESSAGE_LIMIT + 900); + assert_eq!(fit_message(&text).len(), TELEGRAM_MESSAGE_LIMIT); + } + + /// Telegram counts UTF-16 code units; an emoji outside the BMP is two. + #[test] + fn the_limit_counts_utf16_units_not_bytes_or_chars() { + let text = "\u{1F534}".repeat(TELEGRAM_MESSAGE_LIMIT); + let fitted = fit_message(&text); + assert_eq!(fitted.chars().count(), TELEGRAM_MESSAGE_LIMIT / 2); + let cyrillic = "ж".repeat(TELEGRAM_MESSAGE_LIMIT); + assert_eq!(fit_message(&cyrillic).chars().count(), TELEGRAM_MESSAGE_LIMIT, "2-byte chars are 1 unit each"); + } + + /// Cutting between a `MarkdownV2` escape and the character it escapes would + /// leave a dangling backslash, which Telegram rejects as a parse error. + #[test] + fn a_cut_escape_drops_its_dangling_backslash() { + let text = format!("{}\\.tail", "a".repeat(TELEGRAM_MESSAGE_LIMIT - 1)); + assert_eq!(fit_message(&text), "a".repeat(TELEGRAM_MESSAGE_LIMIT - 1)); + } + + #[test] + fn an_escaped_backslash_at_the_cut_is_kept() { + let text = format!("{}\\\\tail", "a".repeat(TELEGRAM_MESSAGE_LIMIT - 2)); + assert_eq!(fit_message(&text), format!("{}\\\\", "a".repeat(TELEGRAM_MESSAGE_LIMIT - 2))); + } +} diff --git a/backend/messaging/src/render.rs b/backend/messaging/src/render.rs index 9c5c206d8..5db52c008 100644 --- a/backend/messaging/src/render.rs +++ b/backend/messaging/src/render.rs @@ -12,8 +12,9 @@ use handlebars::{Context, Handlebars, Helper, HelperResult, Output, RenderContext}; use log::{debug, error, warn}; +use serde::Deserialize; use shared::{ - model::{notification::registry, InputFetchMethod}, + model::{notification::registry, InputFetchMethod, WatchChanges}, utils::Internable, }; use std::{ @@ -249,7 +250,14 @@ pub fn context_for(event: &NotificationEvent) -> serde_json::Value { root.insert("message".to_string(), fields.clone()); } id if id == registry::PLAYLIST_WATCH_CHANGED => { - root.insert("watch".to_string(), fields.clone()); + // Templates get the grouped views. The event only carries JSON + // fields (it may come back from the outbox), so decode them in + // place; a payload that does not decode still renders as is. + let watch = WatchChanges::deserialize(fields) + .ok() + .and_then(|changes| serde_json::to_value(changes.template_view()).ok()) + .unwrap_or_else(|| fields.clone()); + root.insert("watch".to_string(), watch); } id if id == registry::PLAYLIST_UPDATE_COMPLETED || id == registry::PLAYLIST_UPDATE_FAILED => { root.insert("processing".to_string(), fields.clone()); @@ -316,8 +324,11 @@ fn legacy_kind_label(event: &NotificationEvent) -> String { #[cfg(test)] mod tests { - use super::{context_for, validate, TemplateSource}; - use shared::model::{DiskAlert, DiskAlertLevel, InputStats, InputType, PlaylistStats, SourceStats, TargetStats}; + use super::{context_for, new_handlebars, validate, TemplateSource}; + use shared::model::{ + DiskAlert, DiskAlertLevel, InputStats, InputType, PlaylistStats, SourceStats, TargetStats, WatchChange, + WatchGroupItems, WatchItems, + }; use tuliprox_core::model::{MessageContent, NotificationEvent, ProcessingStats, WatchChanges}; fn stats_event() -> NotificationEvent { @@ -412,17 +423,49 @@ mod tests { assert_eq!(ctx["disk"]["percent"], 95.0); } + fn watch_event() -> NotificationEvent { + let items = |names: &[&str]| WatchItems::complete(names.iter().map(|n| std::sync::Arc::from(*n)).collect()); + NotificationEvent::from_content(&MessageContent::Watch(WatchChanges::new( + "sports".to_string(), + vec![ + WatchChange::GroupsAdded(items(&["Event Bundesliga"])), + WatchChange::ChannelsAdded(WatchGroupItems { group: "DE Sport HD".into(), items: items(&["Sky 5"]) }), + ], + ))) + } + #[test] - fn legacy_watch_key_still_resolves() { - let event = NotificationEvent::from_content(&MessageContent::Watch(WatchChanges::new( - "T".to_string(), - "G".to_string(), - vec!["A".to_string()], - vec![], - ))); - let ctx = context_for(&event); + fn watch_key_carries_grouped_views() { + let ctx = context_for(&watch_event()); assert_eq!(ctx["kind"], "Watch"); - assert_eq!(ctx["watch"]["target"], "T"); + assert_eq!(ctx["watch"]["target"], "sports"); + assert_eq!(ctx["watch"]["groups_added"]["items"][0], "Event Bundesliga"); + assert_eq!(ctx["watch"]["channels_added"][0]["group"], "DE Sport HD"); + assert!(ctx["watch"].get("groups_removed").is_none()); + assert_eq!(ctx["watch"]["changes"][0]["kind"], "groups_added"); + } + + #[test] + fn the_feature_request_template_renders() { + let template = "{{#if watch.groups_added}}New:{{#each watch.groups_added.items}} {{this}}{{/each}}{{/if}}\ + {{#each watch.channels_added}}|{{group}}:{{#each items}} {{this}}{{/each}}{{/each}}"; + let rendered = new_handlebars().render_template(template, &context_for(&watch_event())).expect("render"); + assert_eq!(rendered, "New: Event Bundesliga|DE Sport HD: Sky 5"); + } + + #[test] + fn the_raw_list_works_with_eq() { + let template = "{{#each watch.changes}}{{#if (eq kind \"groups_added\")}}{{items.[0]}}{{/if}}{{/each}}"; + let rendered = new_handlebars().render_template(template, &context_for(&watch_event())).expect("render"); + assert_eq!(rendered, "Event Bundesliga"); + } + + #[test] + fn undecodable_watch_fields_fall_back_to_raw() { + let mut event = watch_event(); + event.fields = serde_json::json!({ "target": "old", "group": "G" }); + let ctx = context_for(&event); + assert_eq!(ctx["watch"]["group"], "G"); } #[test] diff --git a/backend/processing/src/playlist_watch.rs b/backend/processing/src/playlist_watch.rs index f02441c48..bfe4abe27 100644 --- a/backend/processing/src/playlist_watch.rs +++ b/backend/processing/src/playlist_watch.rs @@ -1,10 +1,11 @@ use log::error; use shared::model::{ - EventMessage, EventSink, PlaylistGroup, PlaylistGroupsChanged, WatchChanges, WatchDisabled, WatchDisabledReason, + EventMessage, EventSink, PlaylistGroup, WatchChange, WatchChanges, WatchContext, WatchDisabled, + WatchDisabledReason, WatchGroupItems, WatchItems, }; use std::{collections::BTreeSet, path::Path, sync::Arc}; use tuliprox_core::{ - model::AppConfig, + model::ConfigWatch, utils, utils::{binary_deserialize, binary_serialize, file_exists_async}, }; @@ -12,45 +13,79 @@ use tuliprox_core::{ const WATCH_NOTIFICATION_LIST_LIMIT: usize = 120; const WATCH_NOTIFICATION_SUMMARY_THRESHOLD: usize = 500; -/// Track which *groups* a target has, as opposed to which channels a group has. +/// Groups that appeared or vanished, kept only where a `group_added` / +/// `group_removed` entry matches. +#[derive(Debug, Default, PartialEq, Eq)] +pub struct GroupChanges { + pub added: Option, + pub removed: Option, +} + +/// One watched group's channel delta. An unchanged group yields no diff at +/// all, so at least one side is always set. +#[derive(Debug, PartialEq, Eq)] +pub struct GroupChannelDiff { + pub group: Arc, + pub added: Option, + pub removed: Option, +} + +/// Sample sorted titles: the list is capped, the total is not. /// -/// `process_group_watch` below is blind to the group set in both directions. -/// A group appearing is silent - it finds no baseline, writes one and emits -/// nothing, so the whole channel list reads as "not new" from then on. A -/// group vanishing is never even looked at: it is absent from the refreshed -/// playlist, so nothing iterates it. +/// Takes an iterator so a set difference is sampled as it is walked, without +/// collecting it first. +fn sample<'a>(titles: impl Iterator>) -> WatchItems { + let mut items = Vec::new(); + let mut total = 0usize; + for title in titles { + if items.len() < WATCH_NOTIFICATION_LIST_LIMIT { + items.push(Arc::clone(title)); + } + total += 1; + } + WatchItems { truncated: items.len() < total, items, total } +} + +fn sample_non_empty<'a>(titles: impl Iterator>) -> Option { + let items = sample(titles); + (items.total > 0).then_some(items) +} + +/// Diff the target's group titles against the persisted index. /// -/// The index lives beside the per-group directory rather than inside it +/// The index holds *every* group, whatever the watch entries say, so editing +/// a pattern never reads as a change and a newly added `group_added` entry +/// already has a baseline. It lives beside the per-group directory /// (`.groups.bin`, not `/__groups.bin`) so it cannot collide -/// with a group whose sanitized title happens to match. -pub async fn process_target_groups_watch( - app_config: &Arc, +/// with a group whose sanitized title matches. +/// +/// First sight writes the baseline and reports nothing: announcing every +/// group as new on the first refresh would be noise, not news. An unreadable +/// index is reported and replaced - a corrupt file never becomes readable +/// again, and keeping it would switch group tracking off for good. +pub async fn diff_target_groups( + storage_dir: &str, events: &E, target_name: &str, playlist: &[PlaylistGroup], -) { - let new_tree: BTreeSet> = playlist.iter().map(|group| group.title.clone()).collect(); + watches: &[ConfigWatch], +) -> GroupChanges { + let new_tree: BTreeSet> = playlist.iter().map(|group| Arc::clone(&group.title)).collect(); let index_filename = format!("{}.groups.bin", utils::sanitize_filename(target_name)); - let cfg = app_config.config.load(); - let Some(path) = utils::get_file_path(&cfg.storage_dir, Some(std::path::PathBuf::from(&index_filename))) else { + let Some(path) = utils::get_file_path(storage_dir, Some(std::path::PathBuf::from(&index_filename))) else { error!("failed to resolve a path for group index {index_filename}"); emit_watch_disabled(events, target_name, "", format!("could not resolve a path for {index_filename}")); - return; + return GroupChanges::default(); }; - // First sight writes the baseline and says nothing. Announcing every - // group as new on the first refresh after an upgrade would be noise, not - // news. + let mut reported = GroupChanges::default(); let mut changed = true; if file_exists_async(&path).await { if let Some(loaded_tree) = load_watch_tree(&path).await { - let added: BTreeSet> = new_tree.difference(&loaded_tree).cloned().collect(); - let removed: BTreeSet> = loaded_tree.difference(&new_tree).cloned().collect(); - if added.is_empty() && removed.is_empty() { - changed = false; - } else { - emit_groups_changed(events, target_name, &added, &removed); + changed = new_tree != loaded_tree; + if changed { + reported = group_changes(&new_tree, &loaded_tree, watches); } } else { error!("failed to load group index {}", path.to_str().unwrap_or_default()); @@ -74,55 +109,97 @@ pub async fn process_target_groups_watch( ); } } + reported } -/// Same sampling discipline as the per-group watch: titles only in the lists, -/// counts that stay true whatever the lists carry. -fn emit_groups_changed( - events: &E, +/// The groups in `current` but not `previous` that a `group_added` entry +/// matches, and the reverse for `group_removed`. +/// +/// The set differences are filtered and sampled while they are walked, so no +/// intermediate set is built, and a direction no entry asks for is never +/// walked at all. +pub fn group_changes( + current: &BTreeSet>, + previous: &BTreeSet>, + watches: &[ConfigWatch], +) -> GroupChanges { + let kept = |from: &BTreeSet>, without: &BTreeSet>, context: WatchContext| { + if !watches.iter().any(|watch| watch.contexts.contains(context)) { + return None; + } + sample_non_empty(from.difference(without).filter(|title| { + watches.iter().any(|watch| watch.contexts.contains(context) && watch.pattern.is_match(title)) + })) + }; + GroupChanges { + added: kept(current, previous, WatchContext::GroupAdded), + removed: kept(previous, current, WatchContext::GroupRemoved), + } +} + +/// Turn one refresh's group and channel changes into a single event payload. +/// +/// Order: group additions, group removals, then channel changes by group +/// title (added before removed), so notifications and tests do not depend on +/// which concurrent group diff finished first. Above the summary threshold, +/// summed over every group, channel lists are dropped and only counts remain: +/// nobody reads thousands of titles, and channels such as Telegram reject a +/// body that large outright. +pub fn assemble_watch_changes( target_name: &str, - added: &BTreeSet>, - removed: &BTreeSet>, -) { - let added_total = added.len(); - let removed_total = removed.len(); - let mut added: Vec = added.iter().map(std::string::ToString::to_string).collect(); - let mut removed: Vec = removed.iter().map(std::string::ToString::to_string).collect(); - let mut truncated = false; - for list in [&mut added, &mut removed] { - if list.len() > WATCH_NOTIFICATION_LIST_LIMIT { - list.truncate(WATCH_NOTIFICATION_LIST_LIMIT); - truncated = true; + groups: GroupChanges, + mut channel_diffs: Vec, +) -> Option { + channel_diffs.sort_unstable_by(|a, b| a.group.cmp(&b.group)); + let channel_total: usize = channel_diffs + .iter() + .flat_map(|diff| diff.added.iter().chain(diff.removed.iter())) + .map(|items| items.total) + .sum(); + let counts_only = channel_total > WATCH_NOTIFICATION_SUMMARY_THRESHOLD; + + let mut changes = Vec::with_capacity(2 + channel_diffs.len() * 2); + changes.extend(groups.added.map(WatchChange::GroupsAdded)); + changes.extend(groups.removed.map(WatchChange::GroupsRemoved)); + let listed = |mut items: WatchItems| { + if counts_only { + items.items = Vec::new(); + items.truncated = true; + } + items + }; + for GroupChannelDiff { group, added, removed } in channel_diffs { + if let Some(items) = added { + changes + .push(WatchChange::ChannelsAdded(WatchGroupItems { group: Arc::clone(&group), items: listed(items) })); + } + if let Some(items) = removed { + changes.push(WatchChange::ChannelsRemoved(WatchGroupItems { group, items: listed(items) })); } } - events.emit(EventMessage::PlaylistGroupsChanged(PlaylistGroupsChanged { - target: target_name.to_string(), - added, - removed, - added_total, - removed_total, - truncated, - })); + (!changes.is_empty()).then(|| WatchChanges::new(target_name.to_string(), changes)) } +/// Diff one watched group's channels against its baseline. pub async fn process_group_watch( - app_config: &Arc, + storage_dir: &str, events: &E, target_name: &str, pl: &PlaylistGroup, -) { - let mut new_tree = BTreeSet::new(); - pl.channels.iter().for_each(|chan| { - let header = &chan.header; - let title = if header.title.is_empty() { header.name.clone() } else { header.title.clone() }; - new_tree.insert(title); - }); +) -> Option { + let new_tree: BTreeSet> = pl + .channels + .iter() + .map(|chan| { + let header = &chan.header; + Arc::clone(if header.title.is_empty() { &header.name } else { &header.title }) + }) + .collect(); let watch_filename = format!("{}/{}.bin", utils::sanitize_filename(target_name), utils::sanitize_filename(&pl.title)); - let cfg = app_config.config.load(); - let Some(path) = utils::get_file_path(&cfg.storage_dir, Some(std::path::PathBuf::from(&watch_filename))) else { + let Some(path) = utils::get_file_path(storage_dir, Some(std::path::PathBuf::from(&watch_filename))) else { error!("failed to write watch_file {watch_filename}"); emit_watch_disabled( events, @@ -130,19 +207,18 @@ pub async fn process_group_watch( &pl.title, format!("could not resolve a storage path for {watch_filename}"), ); - return; + return None; }; - let save_path = path.as_path(); + let mut diff = None; let mut changed = true; if file_exists_async(&path).await { if let Some(loaded_tree) = load_watch_tree(&path).await { - // Find elements in set2 but not in set1 - let added_difference: BTreeSet> = new_tree.difference(&loaded_tree).cloned().collect(); - let removed_difference: BTreeSet> = loaded_tree.difference(&new_tree).cloned().collect(); - changed = !added_difference.is_empty() || !removed_difference.is_empty(); + let added = sample_non_empty(new_tree.difference(&loaded_tree)); + let removed = sample_non_empty(loaded_tree.difference(&new_tree)); + changed = added.is_some() || removed.is_some(); if changed { - handle_watch_notification(events, &added_difference, &removed_difference, target_name, &pl.title); + diff = Some(GroupChannelDiff { group: Arc::clone(&pl.title), added, removed }); } } else { error!("failed to load watch_file {}", path.to_str().unwrap_or_default()); @@ -159,16 +235,17 @@ pub async fn process_group_watch( } if changed { - if let Err(err) = save_watch_tree(save_path, &new_tree).await { - error!("failed to write watch_file {}: {err}", save_path.to_str().unwrap_or_default()); + if let Err(err) = save_watch_tree(path.as_path(), &new_tree).await { + error!("failed to write watch_file {}: {err}", path.to_str().unwrap_or_default()); emit_watch_disabled( events, target_name, &pl.title, - format!("could not write watch state at {}: {err}", save_path.to_str().unwrap_or_default()), + format!("could not write watch state at {}: {err}", path.to_str().unwrap_or_default()), ); } } + diff } /// Report that a group's watch state could not be read or written. @@ -187,56 +264,6 @@ fn emit_watch_disabled(events: &E, target_name: &str, group_name: events.emit(EventMessage::PlaylistWatchDisabled(disabled)); } -/// Turn a group's membership delta into an event. -/// -/// Large changes are sampled rather than listed in full, but the sampling is -/// reported in [`WatchChanges::added_total`] / [`WatchChanges::removed_total`] -/// and [`WatchChanges::truncated`] rather than written into the lists as prose. -fn handle_watch_notification( - events: &E, - added: &BTreeSet>, - removed: &BTreeSet>, - target_name: &str, - group_name: &str, -) { - let added_total = added.len(); - let removed_total = removed.len(); - let total_changed = added_total.saturating_add(removed_total); - if total_changed == 0 { - return; - } - - let mut added: Vec = added.iter().map(std::string::ToString::to_string).collect(); - let mut removed: Vec = removed.iter().map(std::string::ToString::to_string).collect(); - let mut truncated = false; - - if total_changed > WATCH_NOTIFICATION_SUMMARY_THRESHOLD { - // A whole-provider reshuffle. Nobody reads ten thousand titles, and - // every channel routing this pays for the bytes, so send the counts - // alone. - added.clear(); - removed.clear(); - truncated = true; - } else { - for list in [&mut added, &mut removed] { - if list.len() > WATCH_NOTIFICATION_LIST_LIMIT { - list.truncate(WATCH_NOTIFICATION_LIST_LIMIT); - truncated = true; - } - } - } - - events.emit(EventMessage::PlaylistWatchChanged(WatchChanges { - target: target_name.to_string(), - group: group_name.to_string(), - added, - removed, - added_total, - removed_total, - truncated, - })); -} - async fn load_watch_tree(path: &Path) -> Option>> { let encoded = tokio::fs::read(path).await.ok()?; binary_deserialize(&encoded[..]).ok() @@ -253,79 +280,115 @@ async fn save_watch_tree(path: &Path, tree: &BTreeSet>) -> std::io::Res #[cfg(test)] mod tests { - use super::{handle_watch_notification, WATCH_NOTIFICATION_LIST_LIMIT, WATCH_NOTIFICATION_SUMMARY_THRESHOLD}; - use shared::model::{EventMessage, EventSink, WatchChanges}; - use std::{ - collections::BTreeSet, - sync::{Arc, Mutex}, + use super::{ + assemble_watch_changes, group_changes, sample, GroupChanges, GroupChannelDiff, WATCH_NOTIFICATION_LIST_LIMIT, }; + use shared::model::{ConfigWatchDto, WatchChange, WatchContext, WatchItems}; + use std::{collections::BTreeSet, sync::Arc}; + use tuliprox_core::model::ConfigWatch; - #[derive(Default)] - struct CollectSink(Mutex>); + fn set(names: &[&str]) -> BTreeSet> { names.iter().map(|n| Arc::from(*n)).collect() } - impl EventSink for CollectSink { - fn emit(&self, event: EventMessage) { - if let EventMessage::PlaylistWatchChanged(changes) = event { - self.0.lock().unwrap().push(changes); + fn titles(count: usize) -> BTreeSet> { + // Zero-padded so the `BTreeSet` order matches the numeric order. + (0..count).map(|i| Arc::from(format!("Channel {i:05}").as_str())).collect() + } + + fn watch(pattern: &str, contexts: &[WatchContext]) -> ConfigWatch { + ConfigWatch { + pattern: Arc::new(regex::Regex::new(pattern).expect("regex")), + contexts: ConfigWatchDto::new(pattern, contexts.iter().copied()).on, + } + } + + fn diff(group: &str, added: usize, removed: usize) -> GroupChannelDiff { + GroupChannelDiff { + group: Arc::from(group), + added: (added > 0).then(|| sample(titles(added).iter())), + removed: (removed > 0).then(|| sample(titles(removed).iter())), + } + } + + #[test] + fn a_small_list_is_complete() { + let items = sample(titles(3).iter()); + assert_eq!((items.items.len(), items.total, items.truncated), (3, 3, false)); + } + + /// The lists used to carry a synthesised "... N more entries omitted" + /// string beside real channel titles, which a plugin reading the JSON + /// payload could not tell from a channel actually named that. + #[test] + fn a_long_list_keeps_titles_only_and_the_true_total() { + let items = sample(titles(WATCH_NOTIFICATION_LIST_LIMIT + 40).iter()); + assert_eq!(items.items.len(), WATCH_NOTIFICATION_LIST_LIMIT); + assert_eq!(items.total, WATCH_NOTIFICATION_LIST_LIMIT + 40); + assert!(items.truncated); + assert!(items.items.iter().all(|t| t.starts_with("Channel "))); + } + + #[test] + fn group_changes_are_filtered_per_context() { + let previous = set(&["DE Kids", "Event B", "Stays"]); + let current = set(&["Event A", "News 2", "Stays"]); + let watches = [watch("^Event", &[WatchContext::GroupAdded]), watch(".*", &[WatchContext::GroupRemoved])]; + let changes = group_changes(¤t, &previous, &watches); + assert_eq!(changes.added.map(|i| i.items), Some(vec![Arc::from("Event A")])); + assert_eq!(changes.removed.map(|i| i.items), Some(vec![Arc::from("DE Kids"), Arc::from("Event B")])); + } + + #[test] + fn channel_only_watches_report_no_group_changes() { + let changes = group_changes(&set(&["X"]), &set(&["Y"]), &[watch(".*", &[WatchContext::Channels])]); + assert_eq!(changes, GroupChanges::default()); + } + + #[test] + fn nothing_changed_assembles_nothing() { + assert!(assemble_watch_changes("t", GroupChanges::default(), Vec::new()).is_none()); + } + + #[test] + fn changes_are_ordered_groups_first_then_channels_by_group() { + let groups = + GroupChanges { added: Some(sample(set(&["New"]).iter())), removed: Some(sample(set(&["Old"]).iter())) }; + let changes = assemble_watch_changes("t", groups, vec![diff("B", 1, 1), diff("A", 1, 0)]).expect("changes"); + let kinds: Vec = changes + .changes + .iter() + .map(|c| match c { + WatchChange::GroupsAdded(_) => "ga".to_string(), + WatchChange::GroupsRemoved(_) => "gr".to_string(), + WatchChange::ChannelsAdded(g) => format!("ca:{}", g.group), + WatchChange::ChannelsRemoved(g) => format!("cr:{}", g.group), + }) + .collect(); + assert_eq!(kinds, ["ga", "gr", "ca:A", "ca:B", "cr:B"]); + } + + #[test] + fn a_large_aggregate_reports_counts_only_for_channels() { + // 20 groups x 30 changes = 600 > WATCH_NOTIFICATION_SUMMARY_THRESHOLD. + let diffs = (0..20).map(|i| diff(&format!("G{i:02}"), 30, 0)).collect(); + let groups = GroupChanges { added: Some(sample(set(&["New"]).iter())), removed: None }; + let changes = assemble_watch_changes("t", groups, diffs).expect("changes"); + for change in &changes.changes { + match change { + WatchChange::ChannelsAdded(g) => { + assert!(g.items.items.is_empty() && g.items.truncated && g.items.total == 30); + } + WatchChange::GroupsAdded(items) => assert_eq!(items, &WatchItems::complete(vec![Arc::from("New")])), + other => panic!("unexpected {other:?}"), } } } - fn titles(count: usize) -> BTreeSet> { - // Zero-padded so the `BTreeSet` order matches the numeric order, which - // keeps the truncation assertions readable. - (0..count).map(|i| Arc::from(format!("Channel {i:05}").as_str())).collect() - } - - fn emit(added: usize, removed: usize) -> WatchChanges { - let sink = CollectSink::default(); - handle_watch_notification(&sink, &titles(added), &titles(removed), "target", "group"); - let mut captured = sink.0.lock().unwrap(); - assert_eq!(captured.len(), 1, "expected exactly one event"); - captured.pop().unwrap() - } - #[test] - fn an_unchanged_group_emits_nothing() { - let sink = CollectSink::default(); - handle_watch_notification(&sink, &BTreeSet::new(), &BTreeSet::new(), "target", "group"); - assert!(sink.0.lock().unwrap().is_empty()); - } - - #[test] - fn a_small_change_lists_every_title() { - let changes = emit(3, 2); - assert_eq!(changes.added.len(), 3); - assert_eq!(changes.removed.len(), 2); - assert_eq!(changes.added_total, 3); - assert_eq!(changes.removed_total, 2); - assert!(!changes.truncated); - } - - /// The regression this whole field set exists for: the lists used to carry - /// a synthesised "... N more entries omitted" string beside real channel - /// titles, which a plugin reading the JSON payload could not distinguish - /// from a channel actually named that. - #[test] - fn a_truncated_list_carries_only_channel_titles() { - let added = WATCH_NOTIFICATION_LIST_LIMIT + 40; - let changes = emit(added, 0); - - assert_eq!(changes.added.len(), WATCH_NOTIFICATION_LIST_LIMIT); - assert_eq!(changes.added_total, added, "the total must survive the truncation"); - assert!(changes.truncated); - for title in &changes.added { - assert!(title.starts_with("Channel "), "list carries prose, not a channel title: {title}"); - } - } - - #[test] - fn a_very_large_change_reports_counts_only() { - let added = WATCH_NOTIFICATION_SUMMARY_THRESHOLD + 100; - let changes = emit(added, 0); - - assert!(changes.added.is_empty(), "a change this size should not list titles at all"); - assert_eq!(changes.added_total, added); - assert!(changes.truncated); + fn a_small_aggregate_keeps_channel_lists() { + let changes = assemble_watch_changes("t", GroupChanges::default(), vec![diff("G", 3, 2)]).expect("changes"); + assert!(changes.changes.iter().all(|c| match c { + WatchChange::ChannelsAdded(g) | WatchChange::ChannelsRemoved(g) => !g.items.truncated, + _ => false, + })); } } diff --git a/backend/processing/src/processor/playlist/mod.rs b/backend/processing/src/processor/playlist/mod.rs index 4cfb389b2..914517285 100644 --- a/backend/processing/src/processor/playlist/mod.rs +++ b/backend/processing/src/processor/playlist/mod.rs @@ -4,7 +4,7 @@ use crate::{ input_cache, metadata_sink::{MetadataUpdateSink, NoopMetadataSink}, parser::xmltv::{flatten_tvguide, merge_epg_trees, EpgMergeAccumulator}, - playlist_watch::{process_group_watch, process_target_groups_watch}, + playlist_watch::{assemble_watch_changes, diff_target_groups, process_group_watch}, processor::{ epg::{ adopt_guide_live_epg_id_case, clear_invalid_live_epg_ids, process_playlist_epg, @@ -29,8 +29,8 @@ use shared::{ InputRefreshOverride, InputRefreshPolicy, InputStats, InputType, InputUpdateAction, InputUpdateRequest, MappingStage, PipelineStats, PlaylistEntry, PlaylistGroup, PlaylistItem, PlaylistItemType, PlaylistUpdateInputTelemetry, PlaylistUpdateProgressEvent, PlaylistUpdateRunId, PlaylistUpdateRunOrder, - PlaylistUpdateState, PlaylistUpdateSummary, StreamProperties, TargetStats, UUIDType, WatchDisabled, - WatchDisabledReason, WatchUnmatched, XtreamCluster, + PlaylistUpdateState, PlaylistUpdateSummary, StreamProperties, TargetStats, UUIDType, WatchContext, + WatchDisabled, WatchDisabledReason, WatchUnmatched, XtreamCluster, }, utils::{create_alias_uuid, interner_gc, Internable}, }; diff --git a/backend/processing/src/processor/playlist/target.rs b/backend/processing/src/processor/playlist/target.rs index 3d2e5ec6e..37e65e355 100644 --- a/backend/processing/src/processor/playlist/target.rs +++ b/backend/processing/src/processor/playlist/target.rs @@ -1105,34 +1105,48 @@ pub(crate) async fn process_watch( return false; } + // One snapshot for the whole refresh rather than a config load per group. + let config = app_config.config.load_full(); + let storage_dir = config.storage_dir.as_str(); + // Before the per-group fan-out: this is about which groups exist, not // what is inside the ones the patterns name, so it must see every group // rather than only the watched ones. - process_target_groups_watch(app_config, events, &target.name, new_playlist).await; + let group_changes = diff_target_groups(storage_dir, events, &target.name, new_playlist, watches).await; + // Only `channels` entries are matched against the current groups: they + // pick the groups to diff, and they are the only ones the unmatched + // warning reports on (`ConfigWatch::reports_unmatched`). Once a group is + // watched, an entry that already matched elsewhere needs no regex run. let mut matched = vec![false; watches.len()]; - let mut watched_groups = Vec::new(); + let mut group_watches = Vec::new(); for group in new_playlist { - let mut any = false; - for (index, pattern) in watches.iter().enumerate() { - if pattern.is_match(&group.title) { + let mut watch_channels = false; + for (index, watch) in watches.iter().enumerate() { + if !watch.contexts.contains(WatchContext::Channels) || (watch_channels && matched[index]) { + continue; + } + if watch.pattern.is_match(&group.title) { matched[index] = true; - any = true; + watch_channels = true; } } - if any { - watched_groups.push(group); + if watch_channels { + // Futures built up front: a lazy `map` closure over borrowed state + // trips the higher-ranked `Send` check of the spawned pipeline task. + group_watches.push(process_group_watch(storage_dir, events, &target.name, group)); } } - // A pattern that matches nothing looks exactly like a group that has not - // changed. `EventKindMask::from_wire_names` already reports unmatched - // subscription names for the same reason: a typo must surface. + // A channel pattern that matches nothing looks exactly like a group that + // has not changed, so a typo must surface. Entries waiting for a group to + // appear, and removal-only entries, match nothing by design. + // `EventKindMask::from_wire_names` reports unmatched names for the same reason. let unmatched: Vec = watches .iter() - .enumerate() - .filter(|(index, _)| !matched[*index]) - .map(|(_, pattern)| pattern.as_str().to_string()) + .zip(&matched) + .filter(|(watch, matched)| !**matched && watch.reports_unmatched()) + .map(|(watch, _)| watch.pattern.as_str().to_string()) .collect(); if !unmatched.is_empty() { warn!("target '{}' has {} watch pattern(s) matching no group", target.name, unmatched.len()); @@ -1143,11 +1157,12 @@ pub(crate) async fn process_watch( ))); } - futures::stream::iter( - watched_groups.into_iter().map(|pl| process_group_watch(app_config, events, &target.name, pl)), - ) - .for_each_concurrent(16, |f| f) - .await; + let channel_diffs: Vec<_> = + futures::stream::iter(group_watches).buffer_unordered(16).filter_map(futures::future::ready).collect().await; + + if let Some(changes) = assemble_watch_changes(&target.name, group_changes, channel_diffs) { + events.emit(EventMessage::PlaylistWatchChanged(changes)); + } true } diff --git a/backend/processing/src/processor/playlist/tests.rs b/backend/processing/src/processor/playlist/tests.rs index 6976c0b9a..5e97c9828 100644 --- a/backend/processing/src/processor/playlist/tests.rs +++ b/backend/processing/src/processor/playlist/tests.rs @@ -22,6 +22,8 @@ mod mapping_stage; mod curation_effect_gate; +mod watch_contexts; + #[cfg(test)] mod quality_rejection_fallback; diff --git a/backend/processing/src/processor/playlist/tests/curation_effect_gate.rs b/backend/processing/src/processor/playlist/tests/curation_effect_gate.rs index c54ac32d4..edf1f8b3e 100644 --- a/backend/processing/src/processor/playlist/tests/curation_effect_gate.rs +++ b/backend/processing/src/processor/playlist/tests/curation_effect_gate.rs @@ -181,7 +181,7 @@ async fn assert_curation_preserves_empty_mixed_output_guard(canonical: Option PlaylistGroup { + PlaylistGroup { + id: 1, + title: title.intern(), + xtream_cluster: XtreamCluster::Live, + channels: channels + .iter() + .map(|name| PlaylistItem { + header: PlaylistItemHeader { name: name.intern(), title: name.intern(), ..Default::default() }, + }) + .collect(), + } +} + +fn target(watch: Vec) -> ConfigTarget { + ConfigTarget::from(&ConfigTargetDto { name: "sports".to_string(), watch: Some(watch), ..Default::default() }) +} + +fn watch_events(sink: &PlaylistRunCollectSink) -> Vec { + sink.0 + .lock() + .unwrap() + .drain(..) + .filter(|e| { + matches!( + e, + EventMessage::PlaylistWatchChanged(_) + | EventMessage::PlaylistWatchUnmatched(_) + | EventMessage::PlaylistWatchDisabled(_) + ) + }) + .collect() +} + +#[tokio::test] +async fn first_run_is_silent_then_one_event_carries_groups_and_channels() { + let dir = tempdir().expect("tempdir"); + let config = app_config(dir.path()); + let target = target(vec![ + ConfigWatchDto::channels("^DE Sport"), + ConfigWatchDto::new("^Event", [WatchContext::GroupAdded]), + ConfigWatchDto::new(".*", [WatchContext::GroupRemoved]), + ]); + let sink = PlaylistRunCollectSink::default(); + + let first = vec![group("DE Sport HD", &["Sky 1"]), group("DE Kids", &["KiKa"])]; + assert!(process_watch(&config, &sink, &target, &first).await); + assert!(watch_events(&sink).is_empty(), "first run must only write baselines"); + + let second = + vec![group("DE Sport HD", &["Sky 1", "Sky 5"]), group("Event Bundesliga", &["E1"]), group("News 2", &[])]; + assert!(process_watch(&config, &sink, &target, &second).await); + let events = watch_events(&sink); + assert_eq!(events.len(), 1, "{events:?}"); + let EventMessage::PlaylistWatchChanged(changes) = &events[0] else { panic!("{events:?}") }; + assert_eq!(changes.target, "sports"); + assert!( + matches!(&changes.changes[0], WatchChange::GroupsAdded(i) if i.items == vec![Arc::from("Event Bundesliga")]) + ); + assert!(matches!(&changes.changes[1], WatchChange::GroupsRemoved(i) if i.items == vec![Arc::from("DE Kids")])); + assert!(matches!(&changes.changes[2], WatchChange::ChannelsAdded(g) + if g.group.as_ref() == "DE Sport HD" && g.items.items == vec![Arc::from("Sky 5")])); + assert_eq!(changes.changes.len(), 3, "News 2 matches no group_added entry"); +} + +#[tokio::test] +async fn channel_only_watch_reports_no_group_changes() { + let dir = tempdir().expect("tempdir"); + let config = app_config(dir.path()); + let target = target(vec![ConfigWatchDto::channels(".*")]); + let sink = PlaylistRunCollectSink::default(); + assert!(process_watch(&config, &sink, &target, &[group("A", &["a"])]).await); + assert!(process_watch(&config, &sink, &target, &[group("A", &["a"]), group("B", &["b"])]).await); + assert!(watch_events(&sink).is_empty(), "a new group is only a new channel baseline here"); +} + +#[tokio::test] +async fn unmatched_warning_only_for_channel_entries_without_group_added() { + let dir = tempdir().expect("tempdir"); + let config = app_config(dir.path()); + let target = target(vec![ + ConfigWatchDto::channels("^Nope1"), + ConfigWatchDto::new("^Nope2", [WatchContext::Channels, WatchContext::GroupRemoved]), + ConfigWatchDto::new("^Nope3", [WatchContext::Channels, WatchContext::GroupAdded]), + ConfigWatchDto::new("^Nope4", [WatchContext::GroupAdded]), + ConfigWatchDto::new("^Nope5", [WatchContext::GroupRemoved]), + ]); + let sink = PlaylistRunCollectSink::default(); + assert!(process_watch(&config, &sink, &target, &[group("A", &["a"])]).await); + let events = watch_events(&sink); + let EventMessage::PlaylistWatchUnmatched(unmatched) = &events[0] else { panic!("{events:?}") }; + assert_eq!(unmatched.patterns, vec!["^Nope1".to_string(), "^Nope2".to_string()]); +} + +#[tokio::test] +async fn corrupt_group_index_still_reports_channel_changes() { + let dir = tempdir().expect("tempdir"); + let config = app_config(dir.path()); + let target = target(vec![ConfigWatchDto::new(".*", [WatchContext::Channels, WatchContext::GroupAdded])]); + let sink = PlaylistRunCollectSink::default(); + assert!(process_watch(&config, &sink, &target, &[group("A", &["a"])]).await); + std::fs::write(dir.path().join("sports.groups.bin"), [0xff_u8; 3]).expect("corrupt index"); + + assert!(process_watch(&config, &sink, &target, &[group("A", &["a", "b"]), group("B", &["x"])]).await); + let events = watch_events(&sink); + assert!(events.iter().any(|e| matches!(e, EventMessage::PlaylistWatchDisabled(_))), "{events:?}"); + let changes = events + .iter() + .find_map(|e| if let EventMessage::PlaylistWatchChanged(c) = e { Some(c) } else { None }) + .expect("channel change still reported"); + assert!(changes.changes.iter().all(|c| matches!(c, WatchChange::ChannelsAdded(_))), "no group change: {changes:?}"); + + // The index was rewritten, so the next refresh diffs normally. + let third = [group("A", &["a", "b"]), group("B", &["x"]), group("C", &[])]; + assert!(process_watch(&config, &sink, &target, &third).await); + let events = watch_events(&sink); + assert!( + events.iter().any(|e| matches!(e, EventMessage::PlaylistWatchChanged(c) + if matches!(&c.changes[0], WatchChange::GroupsAdded(i) if i.items == vec![Arc::from("C")]))), + "{events:?}" + ); +} diff --git a/docs/src/configuration/config.md b/docs/src/configuration/config.md index 54d749f91..a59dd6821 100644 --- a/docs/src/configuration/config.md +++ b/docs/src/configuration/config.md @@ -415,11 +415,10 @@ schedules: ## 5. Messaging (`messaging`) -Tuliprox can proactively notify you via Push-Notifications when updates fail, finish, or when specific channels are -added/removed -from a watched group. **Why is this useful?** Because it allows you to instantly detect upstream provider issues or -simply let you -know when new movies are added to your playlist. +Tuliprox can proactively notify you via Push-Notifications when updates fail, finish, when channels are added to or +removed from a watched group, or when watched groups appear or disappear (see +[`watch`](source.md#329-watch-watch)). **Why is this useful?** Because it allows you to instantly detect upstream +provider issues, or simply let you know when new movies or a new event group are added to your playlist. ### 5.1 Configuration & Opt-In @@ -458,8 +457,7 @@ negative pattern does, so `["*", "!system.info"]` reads the way it looks. | `system.shutdown` | info | The server is shutting down cleanly. | | `playlist.update.completed` | info | A playlist update finished; carries per-source statistics. | | `playlist.update.failed` | error | A playlist update failed. | -| `playlist.watch.changed` | info | Channels were added to or removed from a watched group. | -| `playlist.groups.changed` | info | Groups were added to or removed from a target. | +| `playlist.watch.changed` | info | Channels or groups were added to or removed from a watched target. | | `playlist.watch.disabled` | warn | A target's watch configuration is set but not working. | | `playlist.watch.unmatched` | warn | Watch patterns matched no group in the refreshed playlist. | | `recording.started` | info | A recording started. | @@ -701,11 +699,31 @@ The legacy keys below continue to work exactly as before, so existing templates task. It includes data on memory allocation peaks, active worker threads, and non-blocking diagnostic warnings. * **Access:** Access properties directly via dot-notation (e.g., `{{processing.memory_peak_mb}}`). Use this to monitor system health and resource consumption during heavy mapping cycles. -* `{{watch}}`: **Change Tracking Data.** Specifically available for the `watch` event kind. It contains a diff-style - object showing exactly which groups or channels were added, removed, or modified compared to the previous state. - * **Access:** Iterate over the change sets using loops. Common keys include `added`, `removed`, and `modified`. - * **Example:** Use `{{#each watch.added}} • {{name}} {{/each}}` to list all new channels detected in the monitored - groups. +* `{{watch}}`: **Change Tracking Data.** Available for `playlist.watch.changed`. One event per target and update. + * `watch.target`: the target name. + * `watch.groups_added` / `watch.groups_removed`: `{ items, total, truncated }`, absent when there is none. + * `watch.channels_added` / `watch.channels_removed`: list of `{ group, items, total, truncated }`, absent when + there is none. + * `watch.changes`: the raw list; each entry has a `kind` (`groups_added`, `groups_removed`, `channels_added`, + `channels_removed`). + * `items` may be shorter than `total` (`truncated: true`), or empty when an update changed more than 500 channels. + * **Example:** + + ```handlebars + {{#if watch.groups_added}} + 🆕 New groups: + {{#each watch.groups_added.items}} + • {{this}} + {{/each}} + {{/if}} + {{#each watch.channels_added}} + ➕ {{group}} ({{total}}): + {{#each items}} + • {{this}} + {{/each}} + {{/each}} + ``` + * `{{recording}}`: **DVR Lifecycle Data.** Available for `recording_started`, `recording_completed`, and `recording_failed`. Common fields are `programme_title`, `channel`, `effective_start`, `effective_end`, `visibility`, `output_filename`, and `failure_reason` for failed recordings. diff --git a/docs/src/configuration/source.md b/docs/src/configuration/source.md index df9126619..466405de2 100644 --- a/docs/src/configuration/source.md +++ b/docs/src/configuration/source.md @@ -1294,7 +1294,7 @@ sources: | `output` | List | Yes | | Mandatory list of output formats. A single target can generate multiple output representations (e.g., `xtream`, `m3u`, `strm`, `hdhomerun`) from the same transformed result set. | | `curation` | Object | No | | Optional target-wide catalog policy with Trakt and/or TMDB Trending sources. Mutually exclusive with legacy `output[].trakt`; see [target-wide curation](#target-wide-curation-curation). | | `favourites` | List | No | | Duplicates final transformed channels into dedicated favorite groups after processing is complete. This adds curated views without changing the original group structure. | -| `watch` | List | No | | Defines watched group patterns. If matching groups change during updates, Tuliprox emits Messaging events so operational changes become observable automatically. | +| `watch` | List | No | | Watch entries for this target: channel changes in matching groups and/or matching groups that appear or disappear. Changes are reported as Messaging events; see [Watch](#329-watch-watch). | | `use_memory_cache` | Bool | No | `false` | If enabled, the final compiled playlist is cached in RAM. This reduces disk access and improves delivery speed, especially for M3U downloads, but increases memory consumption. | --- @@ -2279,25 +2279,44 @@ favourites: ### 3.2.9 Watch (`watch`) -For each target with a *unique name*, you can define watched groups. -It is a list of group patterns Tuliprox monitors for content changes during updates. +For each target with a *unique name*, you can define watch entries. Each entry is a regex matched against final group +names and says which changes to report: -If matching groups gain or lose channels, Tuliprox emits a Messaging event such as: +| `on` value | Reported change | +|:----------------|:---------------------------------------------------------------| +| `channels` | Channels added to or removed from a matching group. | +| `group_added` | A matching group appeared that was not in the previous update. | +| `group_removed` | A group that matched in the previous update is gone. | -* channels added -* channels removed +An entry is either a plain string, which means `on: [channels]`, or a map with `pattern` and `on`: ```yaml watch: - - group: '^Sports' - - group: '^Movies' + # Short form: channel changes in all German sports groups. + - "^DE Sport.*" + # Report new 4K/UHD groups only. + - pattern: "(?i)\\b(4K|UHD)\\b" + on: [group_added] + # Report every group that disappears. + - pattern: ".*" + on: [group_removed] + # Event groups: appearing, disappearing and their channel changes. + - pattern: "^Event.*|^PPV.*" + on: [channels, group_added, group_removed] ``` +The first update of a target only records the current state; changes are reported from the second update on. All +changes of one update arrive as a single `playlist.watch.changed` event. + +A `playlist.watch.unmatched` warning is sent for entries that watch `channels` (without `group_added`) and match no +group. Entries waiting for a group to appear, or reporting only removals, never trigger it. + #### `watch` Parameters -| Parameter | Type | Required | Default | Technical Impact & Background | -|:----------|:---------------|:--------:|:--------|:---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| -| `group` | String (Regex) | Yes | | Regex pattern matched against final group names. This allows Tuliprox to detect meaningful content changes in selected areas of the playlist and notify operators automatically through the configured messaging backends. | +| Parameter | Type | Required | Default | Technical Impact & Background | +|:----------|:---------------|:--------:|:-------------|:----------------------------------------------------------------------| +| `pattern` | String (Regex) | Yes | | Regex matched against final group names. | +| `on` | List | No | `[channels]` | Any of `channels`, `group_added`, `group_removed`. Must not be empty. | > **Note:** `watch` is especially useful for monitoring premium groups, VOD collections, > or unstable provider segments where additions and removals should generate operational alerts. diff --git a/frontend/public/assets/i18n/ar.json b/frontend/public/assets/i18n/ar.json index 5b5aa33da..e49fd3224 100644 --- a/frontend/public/assets/i18n/ar.json +++ b/frontend/public/assets/i18n/ar.json @@ -383,7 +383,7 @@ "MAPPING": "قائمة معرفات التعيين لتطبيقها على هذا الهدف.", "NAME": "اسم قابل للقراءة البشرية لهذا الإدخال. يُستخدم للمعالجة الانتقائية عبر وسيطة CLI -t.", "USE_MEMORY_CACHE": "إذا تم التفعيل، يتم تخزين قائمة التشغيل في ذاكرة الوصول العشوائي لتقليل عبء القرص (يزيد من استخدام الذاكرة).", - "WATCH": "قائمة أنماط المجموعات لمراقبة التغييرات. يتم إرسال الإشعارات عبر المراسلة المكوّنة." + "WATCH": "قائمة أنماط المجموعات للمراقبة. كل نمط هو تعبير نمطي (Regex) يُطابَق مع اسم المجموعة (مثل ^Sport.* أو (?i)event). يبلغ كل إدخال عن تغييرات القنوات و/أو المجموعات الجديدة و/أو المجموعات المحذوفة. يتم إرسال الإشعارات عبر المراسلة المكوّنة." }, "CONFIG_TARGET_OPTIONS": { "SHARE_LIVE_STREAMS": "يُفعّل مشاركة البث المباشر. المفتاح الرئيسي يضبط HLS و MPEG-TS معًا؛ يمكن تغيير كل تنسيق أيضًا بشكل منفصل." @@ -1740,6 +1740,9 @@ "VOD": "VOD", "VOD_SHORT": "ف", "WATCH": "مراقبة", + "WATCH_CHANNELS": "القنوات", + "WATCH_GROUP_ADDED": "مجموعات جديدة", + "WATCH_GROUP_REMOVED": "مجموعات محذوفة", "WEBHOOK_URL": "عنوان URL لـ Webhook", "WEB_AUTH": "مصادقة", "WEB_ROOT": "جذر الويب", diff --git a/frontend/public/assets/i18n/en.json b/frontend/public/assets/i18n/en.json index b7cc55d4d..612a8cce3 100644 --- a/frontend/public/assets/i18n/en.json +++ b/frontend/public/assets/i18n/en.json @@ -383,7 +383,7 @@ "MAPPING": "List of mapping IDs to apply to this target.", "NAME": "Human readable name for this entry. Used for selective processing via -t CLI arg.", "USE_MEMORY_CACHE": "If enabled, the playlist is cached in RAM to reduce disk overhead (increases memory usage).", - "WATCH": "List of group patterns to monitor for changes. Notifications are sent via configured messaging." + "WATCH": "List of group patterns to monitor. Each pattern is a regular expression matched against the group name (e.g. ^Sport.* or (?i)event). Each entry reports channel changes, new groups and/or removed groups. Notifications are sent via configured messaging." }, "CONFIG_TARGET_OPTIONS": { "SHARE_LIVE_STREAMS": "Enables live stream sharing. The master toggle sets HLS and MPEG-TS together; each format can also be changed separately." @@ -1808,6 +1808,9 @@ "VOD": "VOD", "VOD_SHORT": "V", "WATCH": "Watch", + "WATCH_CHANNELS": "Channels", + "WATCH_GROUP_ADDED": "New groups", + "WATCH_GROUP_REMOVED": "Removed groups", "WEBHOOK_URL": "Webhook URL", "WEB_AUTH": "Auth", "WEB_ROOT": "Web-Root", diff --git a/frontend/public/assets/i18n/es.json b/frontend/public/assets/i18n/es.json index 4ac2ce151..6bce949af 100644 --- a/frontend/public/assets/i18n/es.json +++ b/frontend/public/assets/i18n/es.json @@ -383,7 +383,7 @@ "MAPPING": "Lista de ID de mapeo a aplicar a este destino.", "NAME": "Nombre legible para este elemento. Se utiliza para el procesamiento selectivo mediante el argumento CLI -t.", "USE_MEMORY_CACHE": "Si está habilitado, la lista de reproducción se guarda en caché en RAM para reducir la sobrecarga de disco (aumenta el uso de memoria).", - "WATCH": "Lista de patrones de grupo a monitorizar para detectar cambios. Las notificaciones se envían mediante la mensajería configurada." + "WATCH": "Lista de patrones de grupo a monitorizar. Cada patrón es una expresión regular que se compara con el nombre del grupo (p. ej. ^Sport.* o (?i)event). Cada entrada informa de cambios de canales, grupos nuevos y/o grupos eliminados. Las notificaciones se envían mediante la mensajería configurada." }, "CONFIG_TARGET_OPTIONS": { "SHARE_LIVE_STREAMS": "Habilita el uso compartido de streams en directo. El interruptor maestro activa HLS y MPEG-TS juntos; cada formato también se puede cambiar por separado." @@ -1808,6 +1808,9 @@ "VOD": "VOD", "VOD_SHORT": "V", "WATCH": "Vigilancia", + "WATCH_CHANNELS": "Canales", + "WATCH_GROUP_ADDED": "Grupos nuevos", + "WATCH_GROUP_REMOVED": "Grupos eliminados", "WEBHOOK_URL": "URL del webhook", "WEB_AUTH": "Autenticación", "WEB_ROOT": "Web-Root", diff --git a/frontend/public/assets/i18n/ru.json b/frontend/public/assets/i18n/ru.json index b94f14614..4ed56f1f6 100644 --- a/frontend/public/assets/i18n/ru.json +++ b/frontend/public/assets/i18n/ru.json @@ -383,7 +383,7 @@ "MAPPING": "Список идентификаторов сопоставления, применяемых к этой цели.", "NAME": "Понятное человеку имя для этой записи. Используется для выборочной обработки через аргумент CLI -t.", "USE_MEMORY_CACHE": "Если включено, плейлист кэшируется в оперативной памяти для уменьшения нагрузки на диск (увеличивает использование памяти).", - "WATCH": "Список шаблонов групп для мониторинга изменений. Уведомления отправляются через настроенный мессенджер." + "WATCH": "Список шаблонов групп для мониторинга. Каждый шаблон — регулярное выражение, сравниваемое с названием группы (например, ^Sport.* или (?i)event). Каждая запись сообщает об изменениях каналов, новых и/или удалённых группах. Уведомления отправляются через настроенный мессенджер." }, "CONFIG_TARGET_OPTIONS": { "SHARE_LIVE_STREAMS": "Включает совместное использование прямых трансляций. Главный переключатель задаёт HLS и MPEG-TS вместе; каждый формат можно изменить и отдельно." @@ -1726,6 +1726,9 @@ "VOD": "VOD", "VOD_SHORT": "V", "WATCH": "Наблюдение", + "WATCH_CHANNELS": "Каналы", + "WATCH_GROUP_ADDED": "Новые группы", + "WATCH_GROUP_REMOVED": "Удалённые группы", "WEBHOOK_URL": "URL вебхука", "WEB_AUTH": "Авторизация", "WEB_ROOT": "Web-Root", diff --git a/frontend/public/assets/icons.json b/frontend/public/assets/icons.json index 00daeaa2b..ac142f9e1 100644 --- a/frontend/public/assets/icons.json +++ b/frontend/public/assets/icons.json @@ -643,5 +643,6 @@ "path": "M 12 2 C 6.477 2 2 6.477 2 12 C 2 17.523 6.477 22 12 22 C 17.523 22 22 17.523 22 12 C 22 6.477 17.523 2 12 2 Z M 12 20.2 C 7.471 20.2 3.8 16.529 3.8 12 C 3.8 7.471 7.471 3.8 12 3.8 C 16.529 3.8 20.2 7.471 20.2 12 C 20.2 16.529 16.529 20.2 12 20.2 Z M 11.2 5.8 A 0.8 0.8 0 0 1 12.8 5.8 L 12.8 18.2 A 0.8 0.8 0 0 1 11.2 18.2 Z M 15.39 15.39 A 4.8 4.8 0 1 1 15.39 8.61 A 0.8 0.8 0 0 1 14.26 9.74 A 3.2 3.2 0 1 0 14.26 14.26 A 0.8 0.8 0 0 1 15.39 15.39 Z" }, {"keys": ["DragHandle"], "path": "M9 3h2v2H9zm4 0h2v2h-2zM9 7h2v2H9zm4 0h2v2h-2zM9 11h2v2H9zm4 0h2v2h-2zM9 15h2v2H9zm4 0h2v2h-2zM9 19h2v2H9zm4 0h2v2h-2z"}, - {"keys": ["Columns"], "path": "M3 3h5v18H3V3zm6.5 0h5v18h-5V3zM16 3h5v18h-5V3z"} + {"keys": ["Columns"], "path": "M3 3h5v18H3V3zm6.5 0h5v18h-5V3zM16 3h5v18h-5V3z"}, + {"keys": ["Mapping"], "viewport": "0 -960 960 960", "path": "M296-270q-42 35-87.5 32T129-269q-34-28-46.5-73.5T99-436l75-124q-25-22-39.5-53T120-680q0-66 47-113t113-47q66 0 113 47t47 113q0 66-47 113t-113 47q-9 0-18-1t-17-3l-77 130q-11 18-7 35.5t17 28.5q13 11 31 12.5t35-12.5l420-361q42-35 88-31.5t80 31.5q34 28 46 73.5T861-524l-75 124q25 22 39.5 53t14.5 67q0 66-47 113t-113 47q-66 0-113-47t-47-113q0-66 47-113t113-47q9 0 17.5 1t16.5 3l78-130q11-18 7-35.5T782-630q-13-11-31-12.5T716-630L296-270Zm40.5-353.5Q360-647 360-680t-23.5-56.5Q313-760 280-760t-56.5 23.5Q200-713 200-680t23.5 56.5Q247-600 280-600t56.5-23.5Zm400 400Q760-247 760-280t-23.5-56.5Q713-360 680-360t-56.5 23.5Q600-313 600-280t23.5 56.5Q647-200 680-200t56.5-23.5ZM280-680Zm400 400Z"} ] diff --git a/frontend/scss/app/_component.scss b/frontend/scss/app/_component.scss index 86b23364d..25ca8f177 100644 --- a/frontend/scss/app/_component.scss +++ b/frontend/scss/app/_component.scss @@ -88,6 +88,7 @@ @forward "components/form"; @forward "components/api_user/api_user_view"; @forward "components/keyvalue_editor"; +@forward "components/watch_entries_editor"; @forward "components/loading_screen"; @forward "components/websocket_status"; @forward "components/health_banner"; diff --git a/frontend/scss/app/components/_watch_entries_editor.scss b/frontend/scss/app/components/_watch_entries_editor.scss new file mode 100644 index 000000000..7fffd0330 --- /dev/null +++ b/frontend/scss/app/components/_watch_entries_editor.scss @@ -0,0 +1,53 @@ +.tp__watch-entries-editor { + display: flex; + flex-flow: column; + box-sizing: border-box; + gap: var(--gap-small); + + &__form { + position: sticky; + top: 0; + z-index: 1; + display: flex; + flex-flow: column; + box-sizing: border-box; + gap: var(--gap-small); + padding: var(--gap-small); + border: 1px solid var(--border-color); + border-radius: var(--border-radius); + background-color: var(--card-background-color); + } + + &__pattern { + display: flex; + flex-flow: row; + align-items: center; + box-sizing: border-box; + gap: var(--gap-small); + + .tp__input { + flex: 1; + } + + .svg-icon { + height: 2rem; + width: 2rem; + } + } + + &__entries { + display: flex; + flex-flow: row wrap; + box-sizing: border-box; + gap: var(--gap-small); + } + + .tp__input.invalid input { + color: var(--attention-color); + } + + &__error { + color: var(--attention-color); + font-size: var(--font-size-md); + } +} diff --git a/frontend/src/app/components/playlist/target/target_watch.rs b/frontend/src/app/components/playlist/target/target_watch.rs index 3cd825310..28871ae8a 100644 --- a/frontend/src/app/components/playlist/target/target_watch.rs +++ b/frontend/src/app/components/playlist/target/target_watch.rs @@ -1,3 +1,4 @@ +use crate::{app::components::source_editor::watch_entry_label, i18n::use_translation}; use shared::model::ConfigTargetDto; use std::rc::Rc; use yew::prelude::*; @@ -9,13 +10,14 @@ pub struct TargetWatchProps { #[component] pub fn TargetWatch(props: &TargetWatchProps) -> Html { + let translate = use_translation(); match props.target.watch.as_ref() { None => html! {}, Some(watch) => html! {
    - for item in watch.iter() { -
  • { item }
    • + for entry in watch.iter() { +
  • { watch_entry_label(entry, |key| translate.t(key)) }
    • }
diff --git a/frontend/src/app/components/source_editor/mod.rs b/frontend/src/app/components/source_editor/mod.rs index 1f9390087..99fde66e0 100644 --- a/frontend/src/app/components/source_editor/mod.rs +++ b/frontend/src/app/components/source_editor/mod.rs @@ -18,6 +18,7 @@ mod sidebar; mod target_form; mod trakt_chart_item_form; mod trakt_list_item_form; +mod watch_entries_editor; pub(crate) use alias_item_form::*; pub(crate) use block_view::*; @@ -38,3 +39,4 @@ pub(crate) use sidebar::*; pub(crate) use target_form::*; pub(crate) use trakt_chart_item_form::*; pub(crate) use trakt_list_item_form::*; +pub(crate) use watch_entries_editor::*; diff --git a/frontend/src/app/components/source_editor/target_form.rs b/frontend/src/app/components/source_editor/target_form.rs index d970e13c2..7480e304d 100644 --- a/frontend/src/app/components/source_editor/target_form.rs +++ b/frontend/src/app/components/source_editor/target_form.rs @@ -1,8 +1,9 @@ +use super::WatchEntriesEditor; use crate::{ app::components::{ - build_options, config::HasFormData, select::Select, selection_parse_first, AppIcon, BlockId, BlockInstance, - Card, ClusterFlagsInput, ClusterFlagsInputMode, DropDownSelection, EditMode, FilterInput, IconButton, Panel, - SourceEditorContext, TextButton, ToggleSwitch, + build_options, config::HasFormData, dto_field_id, select::Select, selection_parse_first, AppIcon, BlockId, + BlockInstance, Card, ClusterFlagsInput, ClusterFlagsInputMode, DropDownSelection, EditMode, FilterInput, + IconButton, Panel, SourceEditorContext, TextButton, ToggleSwitch, }, config_field, config_field_bool, config_field_child, config_field_custom, edit_field_bool, edit_field_list_option, edit_field_text, generate_form_reducer, @@ -14,7 +15,8 @@ use shared::{ error::TuliproxError, model::{ permission::Permission, ClusterFlags, ConfigTargetDto, ConfigTargetFilterDto, ConfigTargetOptions, - CurationCatalogSelection, CurationConfigDto, ProcessingOrder, TargetBouquetMode, TargetBouquetStatusDto, + ConfigWatchDto, CurationCatalogSelection, CurationConfigDto, ProcessingOrder, TargetBouquetMode, + TargetBouquetStatusDto, }, utils::Internable, }; @@ -31,7 +33,6 @@ const LABEL_PERSIST_FILTER: &str = "LABEL.PERSIST_FILTER"; const LABEL_MAPPING: &str = "LABEL.MAPPING"; const LABEL_WATCH: &str = "LABEL.WATCH"; const LABEL_ADD_MAPPING: &str = "LABEL.ADD_MAPPING"; -const LABEL_ADD_WATCH: &str = "LABEL.ADD_WATCH"; const LABEL_USE_MEMORY_CACHE: &str = "LABEL.USE_MEMORY_CACHE"; const LABEL_PROCESSING_ORDER: &str = "LABEL.PROCESSING_ORDER"; const LABEL_IGNORE_LOGO: &str = "LABEL.IGNORE_LOGO"; @@ -77,11 +78,15 @@ fn curation_summary(curation: &CurationConfigDto, catalog_selection: &str) -> St enum TargetFormPage { Main, Options, + Mapping, + Watch, } impl TargetFormPage { const MAIN: &str = "Main"; const OPTIONS: &str = "Options"; + const MAPPING: &str = "Mapping"; + const WATCH: &str = "Watch"; } impl FromStr for TargetFormPage { @@ -91,6 +96,8 @@ impl FromStr for TargetFormPage { match s { Self::MAIN => Ok(TargetFormPage::Main), Self::OPTIONS => Ok(TargetFormPage::Options), + Self::MAPPING => Ok(TargetFormPage::Mapping), + Self::WATCH => Ok(TargetFormPage::Watch), _ => Err(TuliproxError::Config(format!("Unknown target form page: {s}"))), } } @@ -101,6 +108,8 @@ impl Display for TargetFormPage { match *self { TargetFormPage::Main => write!(f, "{}", TargetFormPage::MAIN), TargetFormPage::Options => write!(f, "{}", TargetFormPage::OPTIONS), + TargetFormPage::Mapping => write!(f, "{}", TargetFormPage::MAPPING), + TargetFormPage::Watch => write!(f, "{}", TargetFormPage::WATCH), } } } @@ -110,6 +119,8 @@ impl Internable for TargetFormPage { match self { Self::Main => TargetFormPage::MAIN, Self::Options => TargetFormPage::OPTIONS, + Self::Mapping => TargetFormPage::MAPPING, + Self::Watch => TargetFormPage::WATCH, } .intern() } @@ -211,7 +222,7 @@ generate_form_reducer!( ProcessingOrder => processing_order: ProcessingOrder, Filter => filter: ConfigTargetFilterDto, Mapping => mapping: Option>, - Watch => watch: Option>, + Watch => watch: Option>, UseMemoryCache => use_memory_cache: bool, } ); @@ -576,8 +587,6 @@ pub fn ConfigTargetView(props: &ConfigTargetViewProps) -> Html { options={processing_orders.clone()} /> }})} - { edit_field_list_option!(target_form_state, translate.t(LABEL_MAPPING), mapping, ConfigTargetFormAction::Mapping, translate.t(LABEL_ADD_MAPPING)) } - { edit_field_list_option!(target_form_state, translate.t(LABEL_WATCH), watch, ConfigTargetFormAction::Watch, translate.t(LABEL_ADD_WATCH)) } } } else { @@ -601,19 +610,51 @@ pub fn ConfigTargetView(props: &ConfigTargetViewProps) -> Html { translate.t(LABEL_PROCESSING_ORDER), target_form_state.form.processing_order.to_string() ) } - { config_field_custom!( - translate.t(LABEL_MAPPING), - target_form_state.form.mapping.as_ref().map_or_else(String::new, |values| values.join(", ")) - ) } - { config_field_custom!( - translate.t(LABEL_WATCH), - target_form_state.form.watch.as_ref().map_or_else(String::new, |values| values.join(", ")) - ) } } } }; + let render_mapping = || { + if props.allow_write { + html! { + + { edit_field_list_option!(target_form_state, translate.t(LABEL_MAPPING), mapping, ConfigTargetFormAction::Mapping, translate.t(LABEL_ADD_MAPPING)) } + + } + } else { + html! { + + { config_field_custom!( + translate.t(LABEL_MAPPING), + target_form_state.form.mapping.as_ref().map_or_else(String::new, |values| values.join(", ")) + ) } + + } + } + }; + + // Watch entries get their own page: a fixed input area on top, the + // configured entries below. + let render_watch = || { + let state = target_form_state.clone(); + html! { + + { config_field_child!(translate.t(LABEL_WATCH), dto_field_id(&target_form_state.form, "watch"), { + html! { + | { + state.dispatch(ConfigTargetFormAction::Watch((!entries.is_empty()).then_some(entries))); + })} + /> + } + })} + + } + }; + let render_edit_mode = || { html! {
@@ -637,6 +678,12 @@ pub fn ConfigTargetView(props: &ConfigTargetViewProps) -> Html { {render_options()} + + {render_mapping()} + + + {render_watch()} +
} @@ -647,6 +694,8 @@ pub fn ConfigTargetView(props: &ConfigTargetViewProps) -> Html {
+ +
} }; diff --git a/frontend/src/app/components/source_editor/watch_entries_editor.rs b/frontend/src/app/components/source_editor/watch_entries_editor.rs new file mode 100644 index 000000000..276eda03f --- /dev/null +++ b/frontend/src/app/components/source_editor/watch_entries_editor.rs @@ -0,0 +1,286 @@ +//! Editor for a target's `watch` entries: a pattern plus the contexts it +//! reports (`channels`, `group_added`, `group_removed`). + +use crate::{ + app::components::{Chip, IconButton, RadioButtonGroup}, + i18n::use_translation, +}; +use shared::model::{ConfigWatchDto, Prepare, WatchContext, WatchContextSet}; +use std::rc::Rc; +use web_sys::HtmlInputElement; +use yew::prelude::*; + +/// Context, its config spelling (the selection value), and its label key. +const CONTEXTS: [(WatchContext, &str, &str); 3] = [ + (WatchContext::Channels, "channels", "LABEL.WATCH_CHANNELS"), + (WatchContext::GroupAdded, "group_added", "LABEL.WATCH_GROUP_ADDED"), + (WatchContext::GroupRemoved, "group_removed", "LABEL.WATCH_GROUP_REMOVED"), +]; + +/// The selection the context picker shows for `on`. +fn context_names(on: WatchContextSet) -> Vec { + CONTEXTS.iter().filter(|(context, _, _)| on.contains(*context)).map(|(_, name, _)| (*name).to_string()).collect() +} + +/// The contexts a picker selection stands for. +fn contexts_from_names(names: &[String]) -> WatchContextSet { + let mut on = WatchContextSet::new(); + for (context, name, _) in CONTEXTS { + if names.iter().any(|selected| selected == name) { + on.set(context); + } + } + on +} + +/// What a fresh input row starts with: the short form. +fn default_contexts() -> WatchContextSet { ConfigWatchDto::channels("").on } + +#[derive(Properties, Clone, PartialEq)] +pub struct WatchEntriesEditorProps { + pub entries: Rc>, + #[prop_or_else(Callback::noop)] + pub on_change: Callback>, + #[prop_or(true)] + pub readonly: bool, +} + +#[derive(Debug, PartialEq)] +enum CandidateDecision { + Accepted(ConfigWatchDto), + /// Blank, duplicate or without a context: nothing to add, nothing to say. + Rejected, + /// The entry would fail config validation; the message says why. + Invalid(String), +} + +/// An entry with no context or a blank or duplicate pattern is never added, +/// so the editor cannot produce a config that `prepare` rejects. +/// +/// The pattern is kept verbatim: whitespace is part of a regex, and config +/// load does not trim either. Trimming only detects blank input. +fn candidate(pattern: &str, on: WatchContextSet, entries: &[ConfigWatchDto]) -> CandidateDecision { + if pattern.trim().is_empty() || on.is_empty() || entries.iter().any(|entry| entry.pattern == pattern) { + return CandidateDecision::Rejected; + } + let mut entry = ConfigWatchDto { pattern: pattern.to_string(), on }; + match entry.prepare(()) { + Ok(()) => CandidateDecision::Accepted(entry), + Err(err) => CandidateDecision::Invalid(err.to_string()), + } +} + +/// Append `pattern` for the short form, `pattern [context, ...]` otherwise. +pub fn push_watch_entry_label(out: &mut String, entry: &ConfigWatchDto, translate: impl Fn(&str) -> String) { + out.push_str(&entry.pattern); + if entry.on == default_contexts() { + return; + } + let mut separator = " ["; + for (context, _, key) in CONTEXTS { + if entry.has(context) { + out.push_str(separator); + out.push_str(&translate(key)); + separator = ", "; + } + } + out.push(']'); +} + +/// [`push_watch_entry_label`] into a fresh `String`. +pub fn watch_entry_label(entry: &ConfigWatchDto, translate: impl Fn(&str) -> String) -> String { + let mut label = String::new(); + push_watch_entry_label(&mut label, entry, translate); + label +} + +#[component] +pub fn WatchEntriesEditor(props: &WatchEntriesEditorProps) -> Html { + // Shared with the callbacks by reference count; a list is only copied + // when an edit actually produces a new one. + let entries = Rc::clone(&props.entries); + let on_change = props.on_change.clone(); + let readonly = props.readonly; + let translate = use_translation(); + let pattern = use_state(String::new); + let on = use_state(default_contexts); + // Why the last add was refused; cleared as soon as the pattern changes. + let error = use_state(|| None::); + + let on_remove = { + let entries = Rc::clone(&entries); + let on_change = on_change.clone(); + Callback::from(move |removed: usize| { + let mut updated = entries.as_ref().clone(); + if removed < updated.len() { + updated.remove(removed); + on_change.emit(updated); + } + }) + }; + + let on_input = { + let pattern = pattern.clone(); + let error = error.clone(); + Callback::from(move |e: InputEvent| { + let input: HtmlInputElement = e.target_unchecked_into(); + pattern.set(input.value()); + if error.is_some() { + error.set(None); + } + }) + }; + + let add = { + let entries = Rc::clone(&entries); + let pattern = pattern.clone(); + let on = on.clone(); + let error = error.clone(); + Callback::from(move |()| match candidate(&pattern, *on, &entries) { + CandidateDecision::Accepted(entry) => { + let mut updated = entries.as_ref().clone(); + updated.push(entry); + on_change.emit(updated); + pattern.set(String::new()); + on.set(default_contexts()); + } + CandidateDecision::Invalid(message) => error.set(Some(message)), + CandidateDecision::Rejected => {} + }) + }; + + let on_keydown = { + let add = add.clone(); + Callback::from(move |e: KeyboardEvent| { + if e.key() == "Enter" { + e.prevent_default(); + e.stop_propagation(); + add.emit(()); + } + }) + }; + + // At least one context stays selected: the group refuses to drop the + // last one, so an entry with an empty `on` cannot be built here. + let context_options = use_memo((), |()| CONTEXTS.iter().map(|(_, name, _)| (*name).to_string()).collect()); + let context_labels = Rc::new(CONTEXTS.iter().map(|(_, _, key)| translate.t(key)).collect::>()); + let on_select_contexts = { + let on = on.clone(); + Callback::from(move |names: Rc>| on.set(contexts_from_names(&names))) + }; + + let on_add_click = Callback::from(move |(_name, e): (String, MouseEvent)| { + e.prevent_default(); + e.stop_propagation(); + add.emit(()); + }); + + html! { +
+ if !readonly { + // Fixed input area: the list below scrolls, this stays put. +
+
+
+
+ +
+
+ +
+ + if let Some(message) = (*error).as_ref() { + { message } + } +
+ } +
+ { for entries.iter().enumerate().map(|(index, entry)| { + html! { + + } + })} +
+
+ } +} + +#[cfg(test)] +mod tests { + use super::{candidate, context_names, contexts_from_names, watch_entry_label, CandidateDecision}; + use shared::model::{ConfigWatchDto, WatchContext, WatchContextSet}; + + fn set(contexts: &[WatchContext]) -> WatchContextSet { ConfigWatchDto::new("", contexts.iter().copied()).on } + + #[test] + fn blank_pattern_is_rejected() { + assert_eq!(candidate(" ", set(&[WatchContext::Channels]), &[]), CandidateDecision::Rejected); + } + + #[test] + fn no_context_is_rejected() { + assert_eq!(candidate("^A", WatchContextSet::new(), &[]), CandidateDecision::Rejected); + } + + #[test] + fn duplicate_pattern_is_rejected() { + let existing = [ConfigWatchDto::channels("^A")]; + assert_eq!(candidate("^A", set(&[WatchContext::GroupAdded]), &existing), CandidateDecision::Rejected); + } + + /// An invalid regex used to be added and only fail on save; the editor + /// now rejects it with the same message config load would give. + #[test] + fn invalid_regex_is_reported() { + match candidate("^(Sport", set(&[WatchContext::Channels]), &[]) { + CandidateDecision::Invalid(message) => { + assert!(message.contains("Invalid watch regular expression"), "{message}"); + } + other => panic!("expected Invalid, got {other:?}"), + } + } + + /// Whitespace is part of a regex: `"^Sport "` must not become `"^Sport"`. + #[test] + fn valid_candidate_keeps_its_pattern_verbatim() { + assert_eq!( + candidate("^Sport ", set(&[WatchContext::GroupAdded]), &[]), + CandidateDecision::Accepted(ConfigWatchDto::new("^Sport ", [WatchContext::GroupAdded])) + ); + } + + #[test] + fn selection_round_trips_through_the_contexts_set() { + let on = set(&[WatchContext::GroupRemoved, WatchContext::Channels]); + let names = context_names(on); + assert_eq!(names.as_slice(), ["channels", "group_removed"]); + assert_eq!(contexts_from_names(&names), on); + } + + #[test] + fn label_shows_contexts_only_for_object_form() { + let t = |key: &str| key.rsplit('.').next().unwrap_or_default().to_lowercase(); + assert_eq!(watch_entry_label(&ConfigWatchDto::channels("^A"), t), "^A"); + assert_eq!( + watch_entry_label(&ConfigWatchDto::new("^B", [WatchContext::GroupAdded, WatchContext::Channels]), t), + "^B [watch_channels, watch_group_added]" + ); + } +} diff --git a/shared/Cargo.toml b/shared/Cargo.toml index 19034737e..a1e37d06a 100644 --- a/shared/Cargo.toml +++ b/shared/Cargo.toml @@ -9,7 +9,6 @@ workspace = true [dependencies] aes = "0.8.4" -bitflags = "2.13.1" ciborium = "0.2.2" hex = "0.4.3" serde_tuple = "1.1.3" diff --git a/shared/src/model/cluster_flags.rs b/shared/src/model/cluster_flags.rs index 56cd03578..b74d8ed5b 100644 --- a/shared/src/model/cluster_flags.rs +++ b/shared/src/model/cluster_flags.rs @@ -1,5 +1,7 @@ -use crate::model::{PlaylistItemType, XtreamCluster}; -use bitflags::bitflags; +use crate::{ + create_bitset, + model::{PlaylistItemType, XtreamCluster}, +}; use serde::{ de, de::{Error, SeqAccess, Visitor}, @@ -7,13 +9,24 @@ use serde::{ }; use std::fmt; -bitflags! { - #[derive(Debug, Copy, Clone, PartialEq, Eq)] - pub struct ClusterFlags: u16 { - const Live = 1; // 0b0000_0001 - const Vod = 1 << 1; // 0b0000_0010 - const Series = 1 << 2; // 0b0000_0100 - } +create_bitset!(u8, ClusterFlag, Live, Vod, Series); + +/// Which xtream clusters (live, vod, series) something applies to. +/// +/// A newtype over the `create_bitset!` set rather than the set itself: the +/// generated set derives a plain `u8` serde form, while this type's wire form +/// is the `[live,vod]` string persisted in configs and user databases. The API +/// mirrors the `bitflags` type it replaced, so call sites read the same. +#[derive(Copy, Clone, PartialEq, Eq, Hash)] +pub struct ClusterFlags(ClusterFlagSet); + +#[allow(non_upper_case_globals)] +impl ClusterFlags { + pub const Live: Self = Self::flag(ClusterFlag::Live); + pub const Vod: Self = Self::flag(ClusterFlag::Vod); + pub const Series: Self = Self::flag(ClusterFlag::Series); + + const fn flag(flag: ClusterFlag) -> Self { Self(ClusterFlagSet(1 << flag as u8)) } } impl Default for ClusterFlags { @@ -21,57 +34,115 @@ impl Default for ClusterFlags { } impl ClusterFlags { - pub fn has_cluster(&self, item_type: PlaylistItemType) -> bool { - match item_type.cluster() { - XtreamCluster::Live => self.contains(ClusterFlags::Live), - XtreamCluster::Video => self.contains(ClusterFlags::Vod), - XtreamCluster::Series => self.contains(ClusterFlags::Series), + #[must_use] + pub const fn all() -> Self { Self(ClusterFlagSet::ALL) } + + #[must_use] + pub const fn empty() -> Self { Self(ClusterFlagSet::new()) } + + /// The raw bits, widened to the `u16` the `bitflags` type exposed. + #[inline] + pub fn bits(&self) -> u16 { u16::from(self.0 .0) } + + #[inline] + pub fn is_all(&self) -> bool { self.0 == ClusterFlagSet::ALL } + + #[inline] + pub fn is_empty(&self) -> bool { self.0.is_empty() } + + /// Whether every cluster in `other` is set. + #[inline] + pub fn contains(&self, other: Self) -> bool { self.0.contains_all(&other.0) } + + /// Whether any cluster in `other` is set. + #[inline] + pub fn intersects(&self, other: Self) -> bool { self.0.contains_any(&other.0) } + + #[inline] + pub fn insert(&mut self, other: Self) { self.0.union(other.0); } + + #[inline] + pub fn remove(&mut self, other: Self) { self.0 = ClusterFlagSet(self.0 .0 & !other.0 .0); } + + #[inline] + pub fn toggle(&mut self, other: Self) { self.0 = ClusterFlagSet(self.0 .0 ^ other.0 .0); } + + #[inline] + pub fn set(&mut self, other: Self, value: bool) { + if value { + self.insert(other); + } else { + self.remove(other); } } + pub fn has_cluster(&self, item_type: PlaylistItemType) -> bool { + self.0.contains(match item_type.cluster() { + XtreamCluster::Live => ClusterFlag::Live, + XtreamCluster::Video => ClusterFlag::Vod, + XtreamCluster::Series => ClusterFlag::Series, + }) + } + pub fn has_full_flags(&self) -> bool { self.is_all() } + fn parse_flag(item: &str) -> Result { + let item = item.trim(); + // ASCII-insensitive compare: no lowercase copy per item. + [("live", ClusterFlag::Live), ("vod", ClusterFlag::Vod), ("series", ClusterFlag::Series)] + .into_iter() + .find_map(|(name, flag)| name.eq_ignore_ascii_case(item).then_some(flag)) + .ok_or_else(|| format!("Invalid flag {}, allowed are live, vod, series", item.to_lowercase())) + } + fn from_items(items: I) -> Result where I: IntoIterator, S: AsRef, { - let mut result = ClusterFlags::empty(); - + let mut result = ClusterFlagSet::new(); for item in items { - match item.as_ref().trim().to_lowercase().as_str() { - "live" => result.set(ClusterFlags::Live, true), - "vod" => result.set(ClusterFlags::Vod, true), - "series" => result.set(ClusterFlags::Series, true), - invalid => return Err(format!("Invalid flag {invalid}, allowed are live, vod, series")), - } + result.set(Self::parse_flag(item.as_ref())?); } - - Ok(result) + Ok(Self(result)) } } +impl std::ops::BitOr for ClusterFlags { + type Output = Self; + + #[inline] + fn bitor(self, other: Self) -> Self { Self(ClusterFlagSet(self.0 .0 | other.0 .0)) } +} + +impl std::ops::BitOrAssign for ClusterFlags { + #[inline] + fn bitor_assign(&mut self, other: Self) { self.insert(other); } +} + impl From> for ClusterFlags { fn from(value: Option) -> Self { value.unwrap_or_else(ClusterFlags::all) } } impl fmt::Display for ClusterFlags { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - let mut flag_strings = Vec::new(); - if self.contains(ClusterFlags::Live) { - flag_strings.push("live"); + f.write_str("[")?; + let mut separator = ""; + for (flag, name) in [(ClusterFlag::Live, "live"), (ClusterFlag::Vod, "vod"), (ClusterFlag::Series, "series")] { + if self.0.contains(flag) { + f.write_str(separator)?; + f.write_str(name)?; + separator = ","; + } } - if self.contains(ClusterFlags::Vod) { - flag_strings.push("vod"); - } - if self.contains(ClusterFlags::Series) { - flag_strings.push("series"); - } - - write!(f, "[{}]", flag_strings.join(",")) + f.write_str("]") } } +impl fmt::Debug for ClusterFlags { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { write!(f, "ClusterFlags({self})") } +} + impl TryFrom<&str> for ClusterFlags { type Error = String; @@ -125,14 +196,77 @@ impl<'de> Deserialize<'de> for ClusterFlags { where A: SeqAccess<'de>, { - let mut values = Vec::new(); - while let Some(val) = seq.next_element::()? { - let entry = val.trim().to_lowercase(); - values.push(entry); + let mut result = ClusterFlagSet::new(); + while let Some(value) = seq.next_element::()? { + result.set(ClusterFlags::parse_flag(&value).map_err(A::Error::custom)?); } - ClusterFlags::try_from(values).map_err(A::Error::custom) + Ok(ClusterFlags(result)) } } deserializer.deserialize_any(ClusterFlagsVisitor) } } + +#[cfg(test)] +mod tests { + use super::ClusterFlags; + use crate::model::PlaylistItemType; + + #[test] + fn default_and_none_mean_all_clusters() { + assert_eq!(ClusterFlags::default(), ClusterFlags::all()); + assert_eq!(ClusterFlags::from(None), ClusterFlags::all()); + assert!(ClusterFlags::all().is_all()); + assert!(ClusterFlags::all().has_full_flags()); + assert!(ClusterFlags::empty().is_empty()); + assert!(!(ClusterFlags::Live | ClusterFlags::Vod).is_all()); + } + + #[test] + fn set_operations_behave_like_flags() { + let mut flags = ClusterFlags::empty(); + flags.insert(ClusterFlags::Live); + flags |= ClusterFlags::Series; + assert!(flags.contains(ClusterFlags::Live)); + assert!(flags.contains(ClusterFlags::Live | ClusterFlags::Series)); + assert!(!flags.contains(ClusterFlags::Live | ClusterFlags::Vod)); + assert!(flags.intersects(ClusterFlags::Live | ClusterFlags::Vod)); + assert!(!flags.intersects(ClusterFlags::Vod)); + flags.set(ClusterFlags::Live, false); + flags.set(ClusterFlags::Vod, true); + assert_eq!(flags, ClusterFlags::Vod | ClusterFlags::Series); + flags.toggle(ClusterFlags::Series); + flags.toggle(ClusterFlags::Live); + assert_eq!(flags, ClusterFlags::Live | ClusterFlags::Vod); + } + + #[test] + fn has_cluster_maps_item_types() { + let flags = ClusterFlags::Vod; + assert!(flags.has_cluster(PlaylistItemType::Video)); + assert!(!flags.has_cluster(PlaylistItemType::Live)); + } + + #[test] + fn display_and_parse_round_trip() { + assert_eq!((ClusterFlags::Live | ClusterFlags::Series).to_string(), "[live,series]"); + assert_eq!(ClusterFlags::empty().to_string(), "[]"); + assert_eq!(ClusterFlags::try_from("[ Vod , live ]"), Ok(ClusterFlags::Live | ClusterFlags::Vod)); + assert_eq!(ClusterFlags::try_from("()"), Ok(ClusterFlags::empty())); + assert!(ClusterFlags::try_from("live,radio").is_err()); + assert_eq!(ClusterFlags::try_from(vec!["series".to_string()]), Ok(ClusterFlags::Series)); + } + + /// The serialized form is persisted in configs and user databases, so it + /// must not change with the implementation underneath. + #[test] + fn serde_format_is_stable() { + let flags = ClusterFlags::Live | ClusterFlags::Vod; + assert_eq!(serde_json::to_string(&flags).unwrap(), r#""[live,vod]""#); + assert_eq!(serde_json::from_str::(r#""[live,vod]""#).unwrap(), flags); + assert_eq!(serde_json::from_str::(r#"["LIVE"," vod "]"#).unwrap(), flags); + assert_eq!(serde_saphyr::from_str::("[series]\n").unwrap(), ClusterFlags::Series); + let bin = rmp_serde::to_vec(&flags).unwrap(); + assert_eq!(rmp_serde::from_slice::(&bin).unwrap(), flags); + } +} diff --git a/shared/src/model/config/mod.rs b/shared/src/model/config/mod.rs index 80e34b7f1..103d18752 100644 --- a/shared/src/model/config/mod.rs +++ b/shared/src/model/config/mod.rs @@ -44,6 +44,7 @@ mod template_definition; mod time_units; mod tmdb_curation; mod trakt; +mod watch; mod web_auth; mod web_ui; @@ -108,5 +109,6 @@ pub use template_definition::*; pub use time_units::*; pub use tmdb_curation::*; pub use trakt::*; +pub use watch::*; pub use web_auth::*; pub use web_ui::*; diff --git a/shared/src/model/config/target.rs b/shared/src/model/config/target.rs index 51a9896f6..834141693 100644 --- a/shared/src/model/config/target.rs +++ b/shared/src/model/config/target.rs @@ -5,8 +5,8 @@ use crate::{ }, error::TuliproxError, model::{ - ConfigFavouritesDto, ConfigRenameDto, ConfigSortDto, CurationConfigDto, HdHomeRunDeviceOverview, - PatternTemplate, Prepare, PrepareAll, ProcessingOrder, TargetType, + ConfigFavouritesDto, ConfigRenameDto, ConfigSortDto, ConfigWatchDto, CurationConfigDto, + HdHomeRunDeviceOverview, PatternTemplate, Prepare, PrepareAll, ProcessingOrder, TargetType, }, }; @@ -47,7 +47,7 @@ pub struct ConfigTargetDto { #[serde(default, skip_serializing_if = "is_default_processing_order")] pub processing_order: ProcessingOrder, #[serde(default, skip_serializing_if = "Option::is_none")] - pub watch: Option>, + pub watch: Option>, #[serde(default, skip_serializing_if = "is_false")] pub use_memory_cache: bool, } @@ -236,13 +236,7 @@ impl ConfigTargetDto { self.favourites.prepare(templates)?; - if let Some(watch) = &self.watch { - for pat in watch { - if let Err(err) = crate::model::REGEX_CACHE.get_or_compile(pat) { - return Err(TuliproxError::ConfigTarget(format!("Invalid watch regular expression: {err}"))); - } - } - } + self.watch.prepare_all(())?; self.filter.prepare(templates)?; self.rename.prepare_all(templates)?; diff --git a/shared/src/model/config/target/tests.rs b/shared/src/model/config/target/tests.rs index 29aeee1fe..5b997d299 100644 --- a/shared/src/model/config/target/tests.rs +++ b/shared/src/model/config/target/tests.rs @@ -1,6 +1,6 @@ use super::{ - ConfigTargetDto, ConfigTargetFilterDto, ConfigTargetOptions, ConfigTargetShareLiveStreams, EpgOutputOptions, - M3uTargetOutputDto, StrmTargetOutputDto, TargetOutputDto, XtreamTargetOutputDto, + ConfigTargetDto, ConfigTargetFilterDto, ConfigTargetOptions, ConfigTargetShareLiveStreams, ConfigWatchDto, + EpgOutputOptions, M3uTargetOutputDto, StrmTargetOutputDto, TargetOutputDto, XtreamTargetOutputDto, }; fn target_with_outputs(output: Vec) -> ConfigTargetDto { @@ -452,3 +452,42 @@ fn target_options_mpeg_ts_helper_keeps_existing_stream_share_semantics() { assert!(!hls_only.share_live_mpeg_ts_enabled()); assert!(mpeg_ts.share_live_mpeg_ts_enabled()); } + +#[test] +fn watch_entry_with_empty_on_is_rejected() { + let mut target = target_with_outputs(vec![xtream_output()]); + target.watch = Some(vec![ConfigWatchDto { pattern: "^A".to_string(), on: crate::model::WatchContextSet::new() }]); + let err = target.prepare(1, None, None).unwrap_err().to_string(); + assert!(err.contains("empty 'on' list"), "{err}"); +} + +#[test] +fn watch_entry_with_contexts_prepares() { + let mut target = target_with_outputs(vec![xtream_output()]); + target.watch = + Some(vec![ConfigWatchDto::channels("^A"), ConfigWatchDto::new("^B", [crate::model::WatchContext::GroupAdded])]); + target.prepare(1, None, None).unwrap(); +} + +#[test] +fn watch_target_yaml_accepts_both_forms() { + let target: ConfigTargetDto = + serde_saphyr::from_str("name: watched\nwatch:\n - \"^A\"\n - pattern: \"^B\"\n on: [group_added]\n") + .expect("target yaml"); + let watch = target.watch.expect("watch set"); + assert_eq!(watch[0], ConfigWatchDto::channels("^A")); + assert!( + watch[1].has(crate::model::WatchContext::GroupAdded) && !watch[1].has(crate::model::WatchContext::Channels) + ); +} + +#[test] +fn every_bad_watch_entry_is_reported_at_once() { + let mut target = target_with_outputs(vec![xtream_output()]); + target.watch = Some(vec![ + ConfigWatchDto { pattern: "^A".to_string(), on: crate::model::WatchContextSet::new() }, + ConfigWatchDto::channels("^(B"), + ]); + let err = target.prepare(1, None, None).unwrap_err().to_string(); + assert!(err.contains("empty 'on' list") && err.contains("Invalid watch regular expression"), "{err}"); +} diff --git a/shared/src/model/config/watch.rs b/shared/src/model/config/watch.rs new file mode 100644 index 000000000..3af828c1a --- /dev/null +++ b/shared/src/model/config/watch.rs @@ -0,0 +1,303 @@ +//! One entry of a target's `watch` list. +//! +//! An entry is either a bare regex string, which watches the channels of +//! matching groups (the only form that existed before), or a map with +//! `pattern` and `on`, which also chooses whether groups appearing or +//! vanishing are reported. +//! +//! The contexts are a `u8` bitset built by `create_bitset!`, like `RoleSet` +//! and `PermissionSet`: no allocation per entry, and "does this entry apply +//! to X" is a `test` instruction in the per-group loop. + +use crate::{ + create_bitset, + error::TuliproxError, + model::{Prepare, REGEX_CACHE}, +}; +use serde::{de, Deserialize, Deserializer, Serialize, Serializer}; +use std::fmt; + +create_bitset!(u8, WatchContext, Channels, GroupAdded, GroupRemoved); + +/// The config spelling of a [`WatchContext`]. +/// +/// The macro-generated enum serialises its Rust variant names; the YAML +/// spelling is snake case. +#[derive(Clone, Copy, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +enum WireContext { + Channels, + GroupAdded, + GroupRemoved, +} + +const CONTEXTS: [(WatchContext, WireContext); 3] = [ + (WatchContext::Channels, WireContext::Channels), + (WatchContext::GroupAdded, WireContext::GroupAdded), + (WatchContext::GroupRemoved, WireContext::GroupRemoved), +]; + +impl From for WatchContext { + fn from(context: WireContext) -> Self { + match context { + WireContext::Channels => Self::Channels, + WireContext::GroupAdded => Self::GroupAdded, + WireContext::GroupRemoved => Self::GroupRemoved, + } + } +} + +/// The `on` list, read straight into the bitset without an intermediate `Vec`. +struct OnList(WatchContextSet); + +impl<'de> Deserialize<'de> for OnList { + fn deserialize>(deserializer: D) -> Result { + struct OnVisitor; + + impl<'de> de::Visitor<'de> for OnVisitor { + type Value = OnList; + + fn expecting(&self, f: &mut fmt::Formatter) -> fmt::Result { + f.write_str("a list of `channels`, `group_added`, `group_removed`") + } + + fn visit_seq>(self, mut seq: A) -> Result { + let mut set = WatchContextSet::new(); + while let Some(context) = seq.next_element::()? { + set.set(context.into()); + } + Ok(OnList(set)) + } + } + + deserializer.deserialize_seq(OnVisitor) + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ConfigWatchDto { + pub pattern: String, + /// Never empty once `ConfigTargetDto::prepare` accepted the target. + pub on: WatchContextSet, +} + +const CHANNELS_ONLY: WatchContextSet = WatchContextSet(1 << WatchContext::Channels as u8); + +impl ConfigWatchDto { + /// The short form: watch the channels of matching groups. + #[must_use] + pub fn channels(pattern: impl Into) -> Self { Self { pattern: pattern.into(), on: CHANNELS_ONLY } } + + #[must_use] + pub fn new(pattern: impl Into, on: impl IntoIterator) -> Self { + let mut set = WatchContextSet::new(); + for context in on { + set.set(context); + } + Self { pattern: pattern.into(), on: set } + } + + #[must_use] + pub fn has(&self, context: WatchContext) -> bool { self.on.contains(context) } +} + +/// One entry is usable when it reports at least one context and its pattern +/// compiles. Shared by config load and the web UI editor, so both reject the +/// same entries for the same reason. +impl Prepare for ConfigWatchDto { + type Ctx<'a> = (); + + fn prepare(&mut self, (): Self::Ctx<'_>) -> Result<(), TuliproxError> { + if self.on.is_empty() { + return Err(TuliproxError::ConfigTarget(format!("watch entry '{}' has an empty 'on' list", self.pattern))); + } + REGEX_CACHE + .get_or_compile(&self.pattern) + .map(|_| ()) + .map_err(|err| TuliproxError::ConfigTarget(format!("Invalid watch regular expression: {err}"))) + } +} + +impl Serialize for ConfigWatchDto { + fn serialize(&self, serializer: S) -> Result { + struct OnRef(WatchContextSet); + + impl Serialize for OnRef { + fn serialize(&self, serializer: S) -> Result { + serializer.collect_seq( + CONTEXTS.iter().filter(|(context, _)| self.0.contains(*context)).map(|(_, wire)| *wire), + ) + } + } + + #[derive(Serialize)] + struct FullFormRef<'a> { + pattern: &'a str, + on: OnRef, + } + + if self.on == CHANNELS_ONLY { + serializer.serialize_str(&self.pattern) + } else { + FullFormRef { pattern: &self.pattern, on: OnRef(self.on) }.serialize(serializer) + } + } +} + +impl<'de> Deserialize<'de> for ConfigWatchDto { + fn deserialize>(deserializer: D) -> Result { + // A visitor rather than `#[serde(untagged)]`: untagged reports every + // mistake in the map form as "did not match any variant", which hides + // the key the operator mistyped. + struct WatchVisitor; + + impl<'de> de::Visitor<'de> for WatchVisitor { + type Value = ConfigWatchDto; + + fn expecting(&self, f: &mut fmt::Formatter) -> fmt::Result { + f.write_str("a group regex string or a map with `pattern` and `on`") + } + + fn visit_str(self, value: &str) -> Result { + Ok(ConfigWatchDto::channels(value)) + } + + fn visit_string(self, value: String) -> Result { + Ok(ConfigWatchDto::channels(value)) + } + + // `Vec` used to accept unquoted YAML scalars such as + // `- 2024`; keep loading them as patterns. + fn visit_i64(self, value: i64) -> Result { + Ok(ConfigWatchDto::channels(value.to_string())) + } + + fn visit_u64(self, value: u64) -> Result { + Ok(ConfigWatchDto::channels(value.to_string())) + } + + fn visit_bool(self, value: bool) -> Result { + Ok(ConfigWatchDto::channels(value.to_string())) + } + + fn visit_map>(self, map: M) -> Result { + #[derive(Deserialize)] + #[serde(deny_unknown_fields)] + struct FullForm { + pattern: String, + #[serde(default)] + on: Option, + } + + let full = FullForm::deserialize(de::value::MapAccessDeserializer::new(map))?; + Ok(ConfigWatchDto { pattern: full.pattern, on: full.on.map_or(CHANNELS_ONLY, |on| on.0) }) + } + } + + deserializer.deserialize_any(WatchVisitor) + } +} + +#[cfg(test)] +mod tests { + use super::{ConfigWatchDto, WatchContext, WatchContextSet}; + use crate::model::Prepare; + + fn yaml(src: &str) -> Vec { serde_saphyr::from_str(src).expect("valid watch yaml") } + + fn set(contexts: &[WatchContext]) -> WatchContextSet { + let mut set = WatchContextSet::new(); + for context in contexts { + set.set(*context); + } + set + } + + #[test] + fn a_plain_string_watches_channels() { + let entries = yaml("- \"^DE Sport.*\"\n"); + assert_eq!(entries, vec![ConfigWatchDto::channels("^DE Sport.*")]); + } + + #[test] + fn the_object_form_reads_pattern_and_on() { + let entries = yaml("- pattern: \"^Event.*\"\n on: [group_added, channels]\n"); + assert_eq!(entries[0].pattern, "^Event.*"); + assert_eq!(entries[0].on, set(&[WatchContext::GroupAdded, WatchContext::Channels])); + } + + #[test] + fn the_object_form_without_on_watches_channels() { + let entries = yaml("- pattern: \"^News\"\n"); + assert_eq!(entries, vec![ConfigWatchDto::channels("^News")]); + } + + #[test] + fn duplicate_contexts_collapse() { + let entries = yaml("- pattern: x\n on: [group_removed, channels, group_removed]\n"); + assert_eq!(entries[0].on, set(&[WatchContext::GroupRemoved, WatchContext::Channels])); + } + + #[test] + fn an_empty_on_list_stays_empty_for_prepare_to_reject() { + let entries = yaml("- pattern: x\n on: []\n"); + assert!(entries[0].on.is_empty()); + } + + #[test] + fn an_unknown_key_is_named_in_the_error() { + let err = serde_saphyr::from_str::>("- group: \"^Sports\"\n").unwrap_err().to_string(); + assert!(err.contains("group"), "error must name the offending key: {err}"); + assert!(!err.contains("did not match any variant"), "untagged-style error leaked: {err}"); + } + + #[test] + fn an_unknown_context_is_rejected() { + assert!(serde_saphyr::from_str::>("- pattern: x\n on: [groups]\n").is_err()); + } + + #[test] + fn numeric_and_bool_scalars_still_load_as_patterns() { + let entries = yaml("- 2024\n- true\n"); + assert_eq!(entries, vec![ConfigWatchDto::channels("2024"), ConfigWatchDto::channels("true")]); + } + + #[test] + fn round_trip_keeps_short_form_and_writes_contexts_in_canonical_order() { + let entries = vec![ + ConfigWatchDto::channels("^A"), + ConfigWatchDto::new("^B", [WatchContext::GroupRemoved, WatchContext::Channels]), + ]; + let json = serde_json::to_string(&entries).expect("serialize"); + assert_eq!(json, r#"["^A",{"pattern":"^B","on":["channels","group_removed"]}]"#); + let back: Vec = serde_json::from_str(&json).expect("deserialize"); + assert_eq!(back, entries); + } + + #[test] + fn yaml_round_trip_keeps_both_forms() { + let entries = vec![ + ConfigWatchDto::channels("^A"), + ConfigWatchDto::new("^B", [WatchContext::GroupAdded, WatchContext::GroupRemoved]), + ]; + let yaml_text = serde_saphyr::to_string(&entries).expect("serialize yaml"); + assert_eq!(yaml(&yaml_text), entries, "{yaml_text}"); + } + + #[test] + fn prepare_accepts_a_valid_entry() { + assert!(ConfigWatchDto::new("^Event.*", [WatchContext::GroupAdded]).prepare(()).is_ok()); + } + + #[test] + fn prepare_rejects_an_empty_on_list() { + let err = ConfigWatchDto { pattern: "^A".to_string(), on: WatchContextSet::new() }.prepare(()).unwrap_err(); + assert!(err.to_string().contains("watch entry '^A' has an empty 'on' list"), "{err}"); + } + + #[test] + fn prepare_rejects_an_invalid_regex() { + let err = ConfigWatchDto::channels("^(Sport").prepare(()).unwrap_err(); + assert!(err.to_string().contains("Invalid watch regular expression"), "{err}"); + } +} diff --git a/shared/src/model/event/kind.rs b/shared/src/model/event/kind.rs index e2be56f79..fc60c9c7e 100644 --- a/shared/src/model/event/kind.rs +++ b/shared/src/model/event/kind.rs @@ -35,7 +35,6 @@ pub enum EventKind { DiskAlert, ConfigReloadFailed, PlaylistWatchChanged, - PlaylistGroupsChanged, PlaylistWatchDisabled, PlaylistWatchUnmatched, RecordingStarted, @@ -68,7 +67,7 @@ impl EventKind { /// /// The mask type below indexes into this, so the order is load-bearing: /// it is the bit order, not just a listing. - pub const ALL: [Self; 43] = [ + pub const ALL: [Self; 42] = [ Self::ServerError, Self::ServerStarted, Self::ServerShutdown, @@ -88,7 +87,6 @@ impl EventKind { Self::DiskAlert, Self::ConfigReloadFailed, Self::PlaylistWatchChanged, - Self::PlaylistGroupsChanged, Self::PlaylistWatchDisabled, Self::PlaylistWatchUnmatched, Self::RecordingStarted, @@ -139,7 +137,6 @@ impl EventKind { Self::PlaylistUpdate | Self::PlaylistUpdateProgress | Self::PlaylistWatchChanged - | Self::PlaylistGroupsChanged | Self::PlaylistWatchDisabled | Self::PlaylistWatchUnmatched => Permission::PlaylistWrite, Self::LibraryScanProgress | Self::LibraryScanFailed => Permission::LibraryWrite, @@ -254,7 +251,6 @@ impl EventKind { Self::DiskAlert => "system.disk.alert", Self::ConfigReloadFailed => "config.reload.failed", Self::PlaylistWatchChanged => "playlist.watch.changed", - Self::PlaylistGroupsChanged => "playlist.groups.changed", Self::PlaylistWatchDisabled => "playlist.watch.disabled", Self::PlaylistWatchUnmatched => "playlist.watch.unmatched", Self::RecordingStarted => "recording.started", @@ -283,7 +279,14 @@ impl EventKind { /// Parse a wire name back. Unknown names are `None` rather than a /// fallback variant, so a typo in a subscription is visible. #[must_use] - pub fn from_wire_name(name: &str) -> Option { Self::ALL.into_iter().find(|kind| kind.as_wire_name() == name) } + pub fn from_wire_name(name: &str) -> Option { + // `playlist.groups.changed` was folded into watch changes; existing + // plugin subscriptions keep working. + if name == "playlist.groups.changed" { + return Some(Self::PlaylistWatchChanged); + } + Self::ALL.into_iter().find(|kind| kind.as_wire_name() == name) + } } impl EventKindMask { diff --git a/shared/src/model/event/message.rs b/shared/src/model/event/message.rs index 0d1ffeb5d..deece278a 100644 --- a/shared/src/model/event/message.rs +++ b/shared/src/model/event/message.rs @@ -5,11 +5,10 @@ use crate::model::{ stats::SourceStats, ActiveUserConnectionChange, ConfigReloadFailure, ConfigType, ConnectionDenied, DiskAlert, LibraryScanProgressEvent, LibraryScanSummaryStatus, MetadataUpdateFailure, MsgKind, NotificationDeadLetter, Permission, - PlaylistGroupsChanged, PlaylistUpdateProgressEvent, PlaylistUpdateRunId, PlaylistUpdateRunOrder, - PlaylistUpdateState, ProviderAccountEvent, ProviderAccountState, ProviderFetchFailure, ProviderPoolExhausted, - ProviderPriorityFallback, RecordingLifecycleMessage, ScheduledTaskFailure, ServerLifecycleEvent, - ServerLifecycleState, StreamProbeFailure, SystemInfo, UserLifecycleEvent, UserLifecycleState, WatchChanges, - WatchDisabled, WatchUnmatched, + PlaylistUpdateProgressEvent, PlaylistUpdateRunId, PlaylistUpdateRunOrder, PlaylistUpdateState, + ProviderAccountEvent, ProviderAccountState, ProviderFetchFailure, ProviderPoolExhausted, ProviderPriorityFallback, + RecordingLifecycleMessage, ScheduledTaskFailure, ServerLifecycleEvent, ServerLifecycleState, StreamProbeFailure, + SystemInfo, UserLifecycleEvent, UserLifecycleState, WatchChanges, WatchDisabled, WatchUnmatched, }; use std::sync::Arc; @@ -56,7 +55,6 @@ pub enum EventMessage { /// A target's `watch` config saw its group membership change. PlaylistWatchChanged(WatchChanges), /// A target gained or lost whole groups between refreshes. - PlaylistGroupsChanged(PlaylistGroupsChanged), /// A target's `watch` config is configured but not working. PlaylistWatchDisabled(WatchDisabled), /// `watch` patterns that matched no group in the refreshed playlist. @@ -134,7 +132,6 @@ impl EventMessage { Self::DiskAlert(_) => EventKind::DiskAlert, Self::ConfigReloadFailed(_) => EventKind::ConfigReloadFailed, Self::PlaylistWatchChanged(_) => EventKind::PlaylistWatchChanged, - Self::PlaylistGroupsChanged(_) => EventKind::PlaylistGroupsChanged, Self::PlaylistWatchDisabled(_) => EventKind::PlaylistWatchDisabled, Self::PlaylistWatchUnmatched(_) => EventKind::PlaylistWatchUnmatched, // One payload, three kinds: a subscriber that only cares about @@ -232,7 +229,6 @@ impl EventMessage { Self::DiskAlert(_) => registry::SYSTEM_DISK_ALERT, Self::ConfigReloadFailed(_) => registry::CONFIG_RELOAD_FAILED, Self::PlaylistWatchChanged(_) => registry::PLAYLIST_WATCH_CHANGED, - Self::PlaylistGroupsChanged(_) => registry::PLAYLIST_GROUPS_CHANGED, Self::PlaylistWatchDisabled(_) => registry::PLAYLIST_WATCH_DISABLED, Self::PlaylistWatchUnmatched(_) => registry::PLAYLIST_WATCH_UNMATCHED, Self::RecordingLifecycle(msg) => match msg.event { @@ -309,7 +305,6 @@ impl EventMessage { Self::DiskAlert(alert) => encode(alert), Self::ConfigReloadFailed(failure) => encode(failure), Self::PlaylistWatchChanged(changes) => encode(changes), - Self::PlaylistGroupsChanged(changes) => encode(changes), Self::PlaylistWatchDisabled(disabled) => encode(disabled), Self::PlaylistWatchUnmatched(unmatched) => encode(unmatched), Self::RecordingLifecycle(msg) => encode(msg), diff --git a/shared/src/model/event/tests.rs b/shared/src/model/event/tests.rs index 286eb0943..db9bdfbe0 100644 --- a/shared/src/model/event/tests.rs +++ b/shared/src/model/event/tests.rs @@ -55,3 +55,10 @@ fn all_matches_every_kind() { assert!(EventKindMask::ALL.contains(kind)); } } + +#[test] +fn groups_changed_alias_resolves_for_kinds() { + // Folded into `playlist.watch.changed`; plugin manifests that subscribed + // to the old name keep receiving group changes. + assert_eq!(EventKind::from_wire_name("playlist.groups.changed"), Some(EventKind::PlaylistWatchChanged)); +} diff --git a/shared/src/model/messaging.rs b/shared/src/model/messaging.rs index cfd40ab52..85e266137 100644 --- a/shared/src/model/messaging.rs +++ b/shared/src/model/messaging.rs @@ -1,4 +1,5 @@ use crate::concat_string; +use std::sync::Arc; #[derive( Debug, @@ -60,7 +61,49 @@ impl MsgKind { #[cfg(test)] mod tests { - use super::MsgKind; + use super::{MsgKind, WatchChange, WatchChanges, WatchGroupItems, WatchItems}; + use std::sync::Arc; + + fn titles(names: &[&str]) -> Vec> { names.iter().map(|n| Arc::from(*n)).collect() } + + fn sample_changes() -> WatchChanges { + WatchChanges::new( + "sports".to_string(), + vec![ + WatchChange::GroupsAdded(WatchItems::complete(titles(&["Event Bundesliga"]))), + WatchChange::GroupsRemoved(WatchItems::complete(titles(&["DE Kids"]))), + WatchChange::ChannelsAdded(WatchGroupItems { + group: "DE Sport HD".into(), + items: WatchItems::complete(titles(&["Sky Sport 5"])), + }), + ], + ) + } + + #[test] + fn watch_changes_serialize_as_tagged_list() { + let json = serde_json::to_value(sample_changes()).expect("serialize"); + assert_eq!(json["target"], "sports"); + assert_eq!(json["changes"][0]["kind"], "groups_added"); + assert_eq!(json["changes"][0]["items"][0], "Event Bundesliga"); + assert_eq!(json["changes"][2]["kind"], "channels_added"); + assert_eq!(json["changes"][2]["group"], "DE Sport HD"); + assert_eq!(json["changes"][2]["total"], 1); + assert_eq!(json["changes"][2]["truncated"], false); + let back: WatchChanges = serde_json::from_value(json).expect("deserialize"); + assert_eq!(back, sample_changes()); + } + + #[test] + fn template_view_groups_changes_by_kind() { + let changes = sample_changes(); + let view = serde_json::to_value(changes.template_view()).expect("serialize view"); + assert_eq!(view["groups_added"]["items"][0], "Event Bundesliga"); + assert_eq!(view["groups_removed"]["total"], 1); + assert_eq!(view["channels_added"][0]["group"], "DE Sport HD"); + assert!(view.get("channels_removed").is_none(), "empty views must be absent"); + assert_eq!(view["changes"].as_array().map(Vec::len), Some(3)); + } #[test] fn wire_name_matches_serde_rename() { @@ -110,50 +153,98 @@ mod tests { } } -/// Group membership changes a target's `watch` config detected on refresh. +/// Every watch change one target refresh detected. /// /// Lives here rather than in `tuliprox-core` because it is an event payload: /// `EventMessage::PlaylistWatchChanged` carries it, and `shared` is the one -/// crate every emitter and every subscriber can name. +/// crate every emitter and every subscriber can name. One event per target +/// and refresh, so a reshuffle across many groups is one notification. #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] pub struct WatchChanges { pub target: String, - pub group: String, - /// Channel titles added, as a sample. Shorter than [`Self::added_total`] - /// when [`Self::truncated`] is set, and empty when the change was too - /// large to list at all - but never anything except channel titles. - pub added: Vec, - /// Channel titles removed. Same sampling rule as [`Self::added`]. - pub removed: Vec, - /// How many channels were added, whatever the list above carries. - /// - /// The lists used to absorb their own truncation notice: a synthesised - /// `"... 42 more added entries omitted"` pushed in beside real channel - /// titles. That was legible to the text template and to nothing else - - /// [`crate::model::EventMessage::payload`] serialises this struct straight - /// to JSON for plugins, and a plugin cannot tell a sentinel from a channel - /// actually named that. The counts live here so the lists stay honest. - #[serde(default)] - pub added_total: usize, - /// How many channels were removed. See [`Self::added_total`]. - #[serde(default)] - pub removed_total: usize, - /// Whether the lists are a sample rather than the whole change. - #[serde(default)] + /// Group changes first, then channel changes ordered by group title. + pub changes: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum WatchChange { + ChannelsAdded(WatchGroupItems), + ChannelsRemoved(WatchGroupItems), + GroupsAdded(WatchItems), + GroupsRemoved(WatchItems), +} + +/// A sampled list of titles. +/// +/// `items` is shorter than `total` when `truncated` is set, and empty when the +/// change was too large to list - but never anything except real titles. A +/// plugin reading the JSON cannot tell a synthesised "... 42 more" entry from a +/// channel actually named that, so the counts live beside the list. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct WatchItems { + pub items: Vec>, + pub total: usize, pub truncated: bool, } -impl WatchChanges { - /// A complete change set, nothing sampled. - /// - /// The totals follow the lists, so a caller that is not truncating cannot - /// get them out of step. +impl WatchItems { + /// A complete list, nothing sampled. #[must_use] - pub fn new(target: String, group: String, added: Vec, removed: Vec) -> Self { - Self { target, group, added_total: added.len(), removed_total: removed.len(), added, removed, truncated: false } + pub fn complete(items: Vec>) -> Self { Self { total: items.len(), items, truncated: false } } +} + +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct WatchGroupItems { + pub group: Arc, + #[serde(flatten)] + pub items: WatchItems, +} + +impl WatchChanges { + #[must_use] + pub fn new(target: String, changes: Vec) -> Self { Self { target, changes } } + + /// The shape notification templates see under `watch`. + #[must_use] + pub fn template_view(&self) -> WatchTemplateView<'_> { + let mut view = WatchTemplateView { + target: &self.target, + changes: &self.changes, + groups_added: None, + groups_removed: None, + channels_added: Vec::new(), + channels_removed: Vec::new(), + }; + for change in &self.changes { + match change { + WatchChange::GroupsAdded(items) => view.groups_added = Some(items), + WatchChange::GroupsRemoved(items) => view.groups_removed = Some(items), + WatchChange::ChannelsAdded(group) => view.channels_added.push(group), + WatchChange::ChannelsRemoved(group) => view.channels_removed.push(group), + } + } + view } } +/// `changes` regrouped by kind, so a template can say +/// `{{#each watch.groups_added.items}}` instead of filtering the raw list +/// with `eq`. Borrows from the payload; absent keys mean "none". +#[derive(Debug, serde::Serialize)] +pub struct WatchTemplateView<'a> { + pub target: &'a str, + pub changes: &'a [WatchChange], + #[serde(skip_serializing_if = "Option::is_none")] + pub groups_added: Option<&'a WatchItems>, + #[serde(skip_serializing_if = "Option::is_none")] + pub groups_removed: Option<&'a WatchItems>, + #[serde(skip_serializing_if = "Vec::is_empty")] + pub channels_added: Vec<&'a WatchGroupItems>, + #[serde(skip_serializing_if = "Vec::is_empty")] + pub channels_removed: Vec<&'a WatchGroupItems>, +} + /// One recording's lifecycle transition. /// /// `event` says which transition: `MsgKind::Recording{Started,Completed,Failed}`. diff --git a/shared/src/model/mod.rs b/shared/src/model/mod.rs index 85571883b..f3dd2c192 100644 --- a/shared/src/model/mod.rs +++ b/shared/src/model/mod.rs @@ -28,7 +28,6 @@ mod pagination; mod playlist; mod playlist_categories; mod playlist_document; -mod playlist_groups_changed; mod playlist_info_document; mod playlist_request; mod playlist_update_run; @@ -82,14 +81,14 @@ pub use self::{ custom_video_stream_type::*, epg::*, epg_grid::*, epg_request::*, event::*, input_refresh_policy::*, input_update_action::*, ip_check::*, item_field::*, library_request::*, log::*, mapping::*, media_properties::*, messaging::*, metadata_update_failure::*, notification::*, notification_dead_letter::*, pagination::*, playlist::*, - playlist_categories::*, playlist_groups_changed::*, playlist_info_document::*, playlist_request::*, - playlist_update_run::*, playlist_update_status::*, processing_order::*, progress::*, provider_fetch_failure::*, - provider_pool::*, recording::*, recording_math::*, regex_cache::*, resource::*, scheduled_task_failure::*, - search_fields::*, search_request::*, server_lifecycle::*, short_epg::*, stalker::*, stalker_item::*, stats::*, - status_check::*, stream_history::*, stream_history_record::*, stream_info::*, stream_meter::*, - stream_probe_failure::*, stream_properties::*, strm_export_style::*, system_info::*, target_bouquet::*, - target_type::*, transfer::*, ui_playlist_item::*, user_command::*, user_id::*, user_lifecycle::*, uuidtype::*, - watch_health::*, web_socket::*, xtream::*, + playlist_categories::*, playlist_info_document::*, playlist_request::*, playlist_update_run::*, + playlist_update_status::*, processing_order::*, progress::*, provider_fetch_failure::*, provider_pool::*, + recording::*, recording_math::*, regex_cache::*, resource::*, scheduled_task_failure::*, search_fields::*, + search_request::*, server_lifecycle::*, short_epg::*, stalker::*, stalker_item::*, stats::*, status_check::*, + stream_history::*, stream_history_record::*, stream_info::*, stream_meter::*, stream_probe_failure::*, + stream_properties::*, strm_export_style::*, system_info::*, target_bouquet::*, target_type::*, transfer::*, + ui_playlist_item::*, user_command::*, user_id::*, user_lifecycle::*, uuidtype::*, watch_health::*, web_socket::*, + xtream::*, }; pub use ids::*; pub use prepare::*; diff --git a/shared/src/model/notification/identity.rs b/shared/src/model/notification/identity.rs index 613860805..8cb459260 100644 --- a/shared/src/model/notification/identity.rs +++ b/shared/src/model/notification/identity.rs @@ -143,4 +143,5 @@ pub const LEGACY_ALIASES: &[(&str, EventId)] = &[ ("recordingcompleted", registry::RECORDING_COMPLETED), ("recording_failed", registry::RECORDING_FAILED), ("recordingfailed", registry::RECORDING_FAILED), + ("playlist.groups.changed", registry::PLAYLIST_WATCH_CHANGED), ]; diff --git a/shared/src/model/notification/registry.rs b/shared/src/model/notification/registry.rs index 8ff8a2991..3a490de7a 100644 --- a/shared/src/model/notification/registry.rs +++ b/shared/src/model/notification/registry.rs @@ -16,7 +16,6 @@ pub const SYSTEM_SHUTDOWN: EventId = EventId::new("system.shutdown"); pub const PLAYLIST_UPDATE_COMPLETED: EventId = EventId::new("playlist.update.completed"); pub const PLAYLIST_UPDATE_FAILED: EventId = EventId::new("playlist.update.failed"); pub const PLAYLIST_WATCH_CHANGED: EventId = EventId::new("playlist.watch.changed"); -pub const PLAYLIST_GROUPS_CHANGED: EventId = EventId::new("playlist.groups.changed"); pub const PLAYLIST_WATCH_DISABLED: EventId = EventId::new("playlist.watch.disabled"); pub const PLAYLIST_WATCH_UNMATCHED: EventId = EventId::new("playlist.watch.unmatched"); @@ -113,12 +112,7 @@ pub const ALL: &[EventDescriptor] = &[ EventDescriptor { id: PLAYLIST_WATCH_CHANGED, severity: Severity::Info, - description: "Channels were added to or removed from a watched group.", - }, - EventDescriptor { - id: PLAYLIST_GROUPS_CHANGED, - severity: Severity::Info, - description: "Groups were added to or removed from a target.", + description: "Channels or groups were added to or removed from a watched target.", }, EventDescriptor { id: PLAYLIST_WATCH_DISABLED, diff --git a/shared/src/model/notification/tests.rs b/shared/src/model/notification/tests.rs index 394bf4b22..857618638 100644 --- a/shared/src/model/notification/tests.rs +++ b/shared/src/model/notification/tests.rs @@ -206,3 +206,8 @@ fn registry_has_no_duplicate_ids() { assert!(seen.insert(descriptor.id), "duplicate registry entry for {}", descriptor.id); } } + +#[test] +fn groups_changed_alias_resolves_for_ids() { + assert_eq!(EventId::from_wire("playlist.groups.changed"), Some(registry::PLAYLIST_WATCH_CHANGED)); +} diff --git a/shared/src/model/playlist_groups_changed.rs b/shared/src/model/playlist_groups_changed.rs deleted file mode 100644 index 5fdb832e7..000000000 --- a/shared/src/model/playlist_groups_changed.rs +++ /dev/null @@ -1,37 +0,0 @@ -use serde::{Deserialize, Serialize}; - -/// Groups that appeared in or vanished from a target between refreshes. -/// -/// The `watch` feature tracks channels *inside* named groups, and is blind to -/// the group set itself in both directions. A group appearing is silent: -/// `process_group_watch` finds no baseline file, writes one, and emits -/// nothing, so the group's entire channel list reads as "not new" from then -/// on. A group vanishing is worse - it is absent from the refreshed playlist, -/// so nothing iterates it and no code path observes the disappearance at all. -/// -/// Sampled the same way [`WatchChanges`](crate::model::WatchChanges) is: the -/// lists carry group titles only, and the counts stay true whatever the lists -/// hold. -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -pub struct PlaylistGroupsChanged { - pub target: String, - /// Group titles that were not in the previous refresh. - pub added: Vec, - /// Group titles that were in the previous refresh and are gone now. - pub removed: Vec, - #[serde(default)] - pub added_total: usize, - #[serde(default)] - pub removed_total: usize, - /// Whether the lists are a sample rather than the whole change. - #[serde(default)] - pub truncated: bool, -} - -impl PlaylistGroupsChanged { - /// A complete change set, nothing sampled. - #[must_use] - pub fn new(target: String, added: Vec, removed: Vec) -> Self { - Self { target, added_total: added.len(), removed_total: removed.len(), added, removed, truncated: false } - } -}