feat(watch): group added/removed contexts per watch entry (#912)

* feat(watch): group added/removed contexts per watch entry - watch entries: regex string or {pattern, on: [channels|group_added|group_removed]} - one playlist.watch.changed per refresh with WatchChange list; playlist.groups.changed aliased - Telegram/ntfy message length limits - web UI: Mapping and Watch sub pages, new Mapping icon - ClusterFlags on create_bitset!, bitflags removed BREAKING CHANGE: playlist.watch.changed templates must use the new watch.* fields * feat(watch): group added/removed contexts per watch entry - watch entries: regex string or {pattern, on: [channels|group_added|group_removed]} - one playlist.watch.changed per refresh with WatchChange list; playlist.groups.changed aliased - Telegram/ntfy message length limits - web UI: Mapping and Watch sub pages, new Mapping icon - ClusterFlags on create_bitset!, bitflags removed BREAKING CHANGE: playlist.watch.changed templates must use the new watch.* fields
2026-10-20 06:42:20 +02:00 · 2026-10-12 21:05:55 +02:00
parent 45c10edd3d
commit b6bab647df
45 changed files with 1981 additions and 546 deletions
@@ -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},
 };
@@ -1105,34 +1105,48 @@ pub(crate) async fn process_watch<E: EventSink>(
        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<String> = 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<E: EventSink>(
        )));
    }

-    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
 }
@@ -22,6 +22,8 @@ mod mapping_stage;

 mod curation_effect_gate;

+mod watch_contexts;
+
 #[cfg(test)]
 mod quality_rejection_fallback;

@@ -181,7 +181,7 @@ async fn assert_curation_preserves_empty_mixed_output_guard(canonical: Option<sh
                ..M3uTargetOutputDto::default()
            }),
        ],
-        watch: Some(vec![".*".to_string()]),
+        watch: Some(vec![shared::model::ConfigWatchDto::channels(".*")]),
        use_memory_cache: true,
        ..ConfigTargetDto::default()
    });
@@ -282,7 +282,7 @@ async fn complete_empty_curation_publishes_empty_watch_group_state() {
            }),
            ..XtreamTargetOutputDto::default()
        })],
-        watch: Some(vec![".*".to_string()]),
+        watch: Some(vec![shared::model::ConfigWatchDto::channels(".*")]),
        ..ConfigTargetDto::default()
    });
    let mut seeded = vec![PlaylistGroup {
@@ -0,0 +1,127 @@
+// `use super::*` brings `process_watch` into scope the same way `curation_effect_gate.rs` gets it.
+use super::{curation_effect_gate::app_config, support::PlaylistRunCollectSink, *};
+use shared::model::{ConfigTargetDto, ConfigWatchDto, WatchChange, WatchContext};
+use tempfile::tempdir;
+
+fn group(title: &str, channels: &[&str]) -> 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<ConfigWatchDto>) -> ConfigTarget {
+    ConfigTarget::from(&ConfigTargetDto { name: "sports".to_string(), watch: Some(watch), ..Default::default() })
+}
+
+fn watch_events(sink: &PlaylistRunCollectSink) -> Vec<EventMessage> {
+    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:?}"
+    );
+}