pipedash_core/application/services/
device_alerts.rs

1use std::collections::{
2    HashMap,
3    VecDeque,
4};
5use std::sync::atomic::{
6    AtomicU64,
7    Ordering,
8};
9use std::sync::{
10    Arc,
11    Mutex,
12    PoisonError,
13};
14use std::time::Duration;
15
16use async_trait::async_trait;
17use chrono::Utc;
18use tokio::sync::mpsc::{
19    self,
20    error::TrySendError,
21};
22use tokio_util::sync::CancellationToken;
23
24use super::health_service::{
25    LocalNotification,
26    NotificationState,
27    ObservedNotification,
28};
29use super::pipeline_service::PipelineService;
30use super::provider_service::ProviderService;
31use crate::domain::{
32    AlertKind,
33    DomainError,
34    NewDeviceAlert,
35    NotificationPreferences,
36    RunFinish,
37};
38use crate::event::{
39    CoreEvent,
40    EventBus,
41};
42use crate::infrastructure::database::{
43    DeviceAlertRepository,
44    DeviceRepository,
45};
46
47const CHANNEL_CAPACITY: usize = 512;
48const PRUNE_INTERVAL: Duration = Duration::from_secs(3600);
49const RECENT_LIMIT: usize = 256;
50
51pub struct RecentNotifications {
52    entries: Mutex<VecDeque<ObservedNotification>>,
53    next_id: AtomicU64,
54}
55
56impl Default for RecentNotifications {
57    fn default() -> Self {
58        Self {
59            entries: Mutex::new(VecDeque::with_capacity(RECENT_LIMIT)),
60            next_id: AtomicU64::new(
61                u64::try_from(Utc::now().timestamp_micros()).unwrap_or_default(),
62            ),
63        }
64    }
65}
66
67impl RecentNotifications {
68    fn observe(&self, notice: LocalNotification) -> ObservedNotification {
69        let observed = ObservedNotification {
70            id: self.next_id.fetch_add(1, Ordering::Relaxed),
71            at: Utc::now(),
72            notice,
73        };
74        let mut entries = self.entries.lock().unwrap_or_else(PoisonError::into_inner);
75        if entries.len() == RECENT_LIMIT {
76            entries.pop_front();
77        }
78        entries.push_back(observed.clone());
79
80        observed
81    }
82
83    pub fn after(&self, after: u64) -> Vec<ObservedNotification> {
84        self.entries
85            .lock()
86            .unwrap_or_else(PoisonError::into_inner)
87            .iter()
88            .filter(|observed| observed.id > after)
89            .cloned()
90            .collect()
91    }
92}
93
94struct RecordingBus {
95    inner: Arc<dyn EventBus>,
96    tx: mpsc::Sender<CoreEvent>,
97}
98
99fn is_observed(event: &CoreEvent) -> bool {
100    matches!(
101        event,
102        CoreEvent::PipelinesUpdated { .. }
103            | CoreEvent::PipelinesFailed { .. }
104            | CoreEvent::ProviderHealthAlert { .. }
105            | CoreEvent::ProviderAdded { .. }
106            | CoreEvent::ProviderUpdated { .. }
107            | CoreEvent::ProviderStatusUpdated { .. }
108            | CoreEvent::ProviderRemoved { .. }
109    )
110}
111
112#[async_trait]
113impl EventBus for RecordingBus {
114    async fn emit(&self, event: CoreEvent) {
115        if is_observed(&event) {
116            if let Err(TrySendError::Full(_)) = self.tx.try_send(event.clone()) {
117                tracing::debug!("Alert queue is full; an observation was dropped");
118            }
119        }
120        self.inner.emit(event).await;
121    }
122
123    async fn emit_to(&self, target: &str, event: CoreEvent) {
124        self.inner.emit_to(target, event).await;
125    }
126}
127
128pub fn recording_bus(inner: Arc<dyn EventBus>) -> (Arc<dyn EventBus>, mpsc::Receiver<CoreEvent>) {
129    let (tx, rx) = mpsc::channel(CHANNEL_CAPACITY);
130
131    (Arc::new(RecordingBus { inner, tx }), rx)
132}
133
134pub async fn seed(pipelines: &PipelineService, providers: &ProviderService) -> NotificationState {
135    let mut state = NotificationState::default();
136    let cached = async {
137        Ok::<_, DomainError>((
138            pipelines.get_cached_pipelines(None).await?,
139            providers.list_providers().await?,
140        ))
141    };
142    match cached.await {
143        Ok((pipelines, providers)) => state.seed(&pipelines, &providers),
144        Err(error) => tracing::warn!(%error, "Failed to seed the alert recorder"),
145    }
146
147    state
148}
149
150pub async fn record(
151    rx: mpsc::Receiver<CoreEvent>, state: NotificationState, stop: CancellationToken,
152    alerts: DeviceAlertRepository, devices: DeviceRepository, bus: Arc<dyn EventBus>,
153    recent: Arc<RecentNotifications>,
154) {
155    record_pruning(
156        rx,
157        state,
158        stop,
159        alerts,
160        devices,
161        bus,
162        recent,
163        PRUNE_INTERVAL,
164    )
165    .await;
166}
167
168#[allow(clippy::too_many_arguments)]
169async fn record_pruning(
170    mut rx: mpsc::Receiver<CoreEvent>, mut state: NotificationState, stop: CancellationToken,
171    alerts: DeviceAlertRepository, devices: DeviceRepository, bus: Arc<dyn EventBus>,
172    recent: Arc<RecentNotifications>, prune_every: Duration,
173) {
174    let mut prune = tokio::time::interval(prune_every);
175    prune.tick().await;
176    loop {
177        let event = tokio::select! {
178            biased;
179            _ = stop.cancelled() => break,
180            _ = prune.tick() => {
181                if let Err(error) = alerts.prune().await {
182                    tracing::warn!(%error, "Failed to prune old device alerts");
183                }
184                continue;
185            }
186            event = rx.recv() => match event {
187                Some(event) => event,
188                None => break,
189            },
190        };
191        let notices = state.observe_event(&event);
192        if notices.is_empty() {
193            continue;
194        }
195        for notice in &notices {
196            bus.emit(CoreEvent::NotificationObserved {
197                notification: recent.observe(notice.clone()),
198            })
199            .await;
200        }
201        match devices.list().await {
202            Ok(paired) if paired.is_empty() => continue,
203            Ok(_) => {}
204            Err(error) => {
205                tracing::warn!(%error, "Failed to list paired devices for alerts");
206                continue;
207            }
208        }
209        let mut watchers = if notices.iter().any(|notice| notice.finish.is_some()) {
210            match devices.notification_preferences().await {
211                Ok(preferences) => preferences,
212                Err(error) => {
213                    tracing::warn!(%error, "Failed to load who watches finished runs");
214                    HashMap::new()
215                }
216            }
217        } else {
218            HashMap::new()
219        };
220        for notice in notices {
221            if notice.kind == AlertKind::Finished && !wanted_by_any(&watchers, &notice) {
222                continue;
223            }
224            if let (Some(pipeline_id), Some(finish)) = (&notice.pipeline_id, &notice.finish) {
225                fire_once(&devices, &mut watchers, pipeline_id, finish).await;
226            }
227            let alert = NewDeviceAlert {
228                key: notice.key,
229                kind: notice.kind,
230                title: notice.title,
231                body: notice.body,
232                pipeline_id: notice.pipeline_id,
233                provider_id: notice.provider_id,
234                provider_ids: notice.provider_ids,
235                finish: notice.finish,
236            };
237            match alerts.insert(&alert).await {
238                Ok(stored) => {
239                    bus.emit(CoreEvent::AlertRecorded {
240                        id: stored.id,
241                        seq: stored.seq,
242                    })
243                    .await;
244                }
245                Err(error) => tracing::warn!(%error, "Failed to record a device alert"),
246            }
247        }
248    }
249}
250
251fn wanted_by_any(
252    watchers: &HashMap<String, NotificationPreferences>, notice: &LocalNotification,
253) -> bool {
254    watchers.values().any(|preferences| {
255        preferences.allows(
256            notice.kind,
257            &notice.provider_ids,
258            notice.pipeline_id.as_deref(),
259            notice.finish.as_ref(),
260        )
261    })
262}
263
264async fn fire_once(
265    devices: &DeviceRepository, watchers: &mut HashMap<String, NotificationPreferences>,
266    pipeline_id: &str, finish: &RunFinish,
267) {
268    for (device_id, preferences) in watchers.iter_mut() {
269        if preferences.fire_once(pipeline_id, finish).is_empty() {
270            continue;
271        }
272        match devices
273            .update_notification_preferences(device_id, |stored| {
274                !stored.fire_once(pipeline_id, finish).is_empty()
275            })
276            .await
277        {
278            Ok(Some(saved)) => *preferences = saved,
279            Ok(None) => {}
280            Err(error) => tracing::warn!(%error, "Failed to mark a one-time watch as used"),
281        }
282    }
283}
284
285#[cfg(test)]
286mod tests {
287    use chrono::{
288        Duration,
289        TimeZone,
290        Utc,
291    };
292
293    use super::*;
294    use crate::application::test_support::{
295        pipeline,
296        Harness,
297    };
298    use crate::domain::{
299        AlertKind,
300        DeviceAlert,
301        PairedDevice,
302        Pipeline,
303        PipelineStatus,
304    };
305    use crate::infrastructure::database::DatabasePool;
306
307    fn run(status: PipelineStatus, minute: i64) -> Pipeline {
308        let mut pipeline = pipeline(1, "api", status);
309        pipeline.last_run =
310            Some(Utc.with_ymd_and_hms(2026, 11, 3, 12, 0, 0).unwrap() + Duration::minutes(minute));
311        pipeline
312    }
313
314    fn updated(pipeline: Pipeline) -> CoreEvent {
315        CoreEvent::PipelinesUpdated {
316            pipelines: vec![pipeline],
317            provider_id: 1,
318            timestamp: 0,
319        }
320    }
321
322    async fn recorded(h: &Harness, paired: bool, events: Vec<CoreEvent>) -> Vec<DeviceAlert> {
323        let pool = DatabasePool::Sqlite(h.pool.clone());
324        let devices = DeviceRepository::new(pool.clone());
325        if paired {
326            devices
327                .insert(&PairedDevice::issue("Pixel", "android").0)
328                .await
329                .unwrap();
330        }
331        let (bus, rx) = recording_bus(h.bus.clone());
332        let state = seed(&h.pipeline_service, &h.provider_service).await;
333        let task = tokio::spawn(record(
334            rx,
335            state,
336            CancellationToken::new(),
337            DeviceAlertRepository::new(pool.clone()),
338            devices,
339            h.bus.clone(),
340            Arc::default(),
341        ));
342        for event in events {
343            bus.emit(event).await;
344        }
345        drop(bus);
346        task.await.unwrap();
347
348        DeviceAlertRepository::new(pool).list().await.unwrap()
349    }
350
351    #[tokio::test]
352    async fn the_recorder_stops_when_cancelled_while_events_can_still_arrive() {
353        let h = Harness::new().await;
354        let pool = DatabasePool::Sqlite(h.pool.clone());
355        let (bus, rx) = recording_bus(h.bus.clone());
356        let stop = CancellationToken::new();
357        let task = tokio::spawn(record(
358            rx,
359            NotificationState::default(),
360            stop.clone(),
361            DeviceAlertRepository::new(pool.clone()),
362            DeviceRepository::new(pool),
363            h.bus.clone(),
364            Arc::default(),
365        ));
366
367        stop.cancel();
368
369        tokio::time::timeout(std::time::Duration::from_secs(5), task)
370            .await
371            .expect("recorder kept running after cancel")
372            .unwrap();
373        drop(bus);
374    }
375
376    #[tokio::test]
377    async fn the_recorder_prunes_expired_alerts_periodically() {
378        let h = Harness::new().await;
379        let pool = DatabasePool::Sqlite(h.pool.clone());
380        let alerts = DeviceAlertRepository::new(pool.clone());
381        let stored = alerts
382            .insert(&NewDeviceAlert {
383                key: "pipeline:p".into(),
384                kind: AlertKind::Failed,
385                title: "t".into(),
386                body: "b".into(),
387                pipeline_id: None,
388                provider_id: None,
389                provider_ids: Vec::new(),
390                finish: None,
391            })
392            .await
393            .unwrap();
394        sqlx::query("UPDATE device_alerts SET created_at = ? WHERE seq = ?")
395            .bind((Utc::now() - Duration::days(8)).to_rfc3339())
396            .bind(stored.seq)
397            .execute(&h.pool)
398            .await
399            .unwrap();
400        let (_bus, rx) = recording_bus(h.bus.clone());
401        let stop = CancellationToken::new();
402        let task = tokio::spawn(record_pruning(
403            rx,
404            NotificationState::default(),
405            stop.clone(),
406            DeviceAlertRepository::new(pool.clone()),
407            DeviceRepository::new(pool),
408            h.bus.clone(),
409            Arc::default(),
410            std::time::Duration::from_millis(300),
411        ));
412
413        assert_eq!(alerts.list().await.unwrap().len(), 1);
414        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
415        while !alerts.list().await.unwrap().is_empty() {
416            assert!(
417                std::time::Instant::now() < deadline,
418                "expired alert was never pruned"
419            );
420            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
421        }
422
423        stop.cancel();
424        task.await.unwrap();
425    }
426
427    #[tokio::test]
428    async fn the_seed_reflects_the_cached_pipelines() {
429        let h = Harness::new().await;
430        assert_eq!(h.add_provider("ci").await, 1);
431        h.repository
432            .update_pipelines_cache(1, &[run(PipelineStatus::Success, 0)])
433            .await
434            .unwrap();
435
436        let stored = recorded(&h, true, vec![updated(run(PipelineStatus::Failed, 1))]).await;
437
438        assert_eq!(stored.len(), 1);
439        assert_eq!(stored[0].kind, AlertKind::Failed);
440    }
441
442    #[tokio::test]
443    async fn a_failing_pipeline_is_recorded_once_and_announced() {
444        let h = Harness::new().await;
445        let events = vec![
446            updated(run(PipelineStatus::Success, 0)),
447            updated(run(PipelineStatus::Failed, 1)),
448        ];
449
450        let stored = recorded(&h, true, events).await;
451
452        assert_eq!(stored.len(), 1);
453        assert_eq!(stored[0].kind, AlertKind::Failed);
454        assert_eq!(stored[0].pipeline_id.as_deref(), Some("github__1__api"));
455        let announced = h
456            .bus
457            .events()
458            .into_iter()
459            .filter_map(|event| match event {
460                CoreEvent::AlertRecorded { id, seq } => Some((id, seq)),
461                _ => None,
462            })
463            .collect::<Vec<_>>();
464        assert_eq!(announced, vec![(stored[0].id.clone(), stored[0].seq)]);
465    }
466
467    #[tokio::test]
468    async fn a_pipeline_that_passes_again_is_recorded_once_as_a_recovery_with_its_finish() {
469        let h = Harness::new().await;
470        let events = vec![
471            updated(run(PipelineStatus::Failed, 0)),
472            updated(run(PipelineStatus::Running, 1)),
473            updated(run(PipelineStatus::Success, 1)),
474            updated(run(PipelineStatus::Success, 1)),
475        ];
476
477        let stored = recorded(&h, true, events).await;
478
479        assert_eq!(stored.len(), 1);
480        assert_eq!(stored[0].kind, AlertKind::Recovered);
481        assert_eq!(stored[0].title, "api passed again");
482        assert_eq!(stored[0].pipeline_id.as_deref(), Some("github__1__api"));
483        assert!(stored[0].finish.is_some());
484    }
485
486    #[tokio::test]
487    async fn nothing_is_recorded_without_a_paired_device() {
488        let h = Harness::new().await;
489        let events = vec![
490            updated(run(PipelineStatus::Success, 0)),
491            updated(run(PipelineStatus::Failed, 1)),
492        ];
493
494        assert!(recorded(&h, false, events).await.is_empty());
495        assert_eq!(h.bus.count("alert-recorded"), 0);
496    }
497
498    #[tokio::test]
499    async fn the_first_sight_of_a_failed_pipeline_is_not_recorded() {
500        let h = Harness::new().await;
501        let events = vec![updated(run(PipelineStatus::Failed, 1))];
502
503        assert!(recorded(&h, true, events).await.is_empty());
504    }
505
506    #[tokio::test]
507    async fn a_finished_run_is_recorded_only_while_a_device_watches_it() {
508        let h = Harness::new().await;
509        let devices = DeviceRepository::new(DatabasePool::Sqlite(h.pool.clone()));
510        let device = PairedDevice::issue("Pixel", "android").0;
511        devices.insert(&device).await.unwrap();
512        let finishing = |minute| {
513            vec![
514                updated(run(PipelineStatus::Running, minute)),
515                updated(run(PipelineStatus::Success, minute)),
516            ]
517        };
518
519        assert!(recorded(&h, false, finishing(1)).await.is_empty());
520
521        devices
522            .set_notification_preferences(
523                &device.id,
524                &NotificationPreferences {
525                    watched_pipelines: vec!["github__1__api".into()],
526                    ..NotificationPreferences::default()
527                },
528            )
529            .await
530            .unwrap();
531        let stored = recorded(&h, false, finishing(2)).await;
532
533        assert_eq!(stored.len(), 1);
534        assert_eq!(stored[0].kind, AlertKind::Finished);
535        assert_eq!(stored[0].title, "api succeeded");
536        assert!(stored[0].finish.is_some());
537    }
538
539    #[tokio::test]
540    async fn a_one_time_watch_is_used_once_even_when_the_server_restarts() {
541        let h = Harness::new().await;
542        let devices = DeviceRepository::new(DatabasePool::Sqlite(h.pool.clone()));
543        let device = PairedDevice::issue("Pixel", "android").0;
544        devices.insert(&device).await.unwrap();
545        devices
546            .set_notification_preferences(
547                &device.id,
548                &NotificationPreferences {
549                    watch_once: vec![crate::domain::OnceWatch {
550                        pipeline_id: "github__1__api".into(),
551                        since: Utc::now() - Duration::minutes(1),
552                        fired_at: None,
553                    }],
554                    ..NotificationPreferences::default()
555                },
556            )
557            .await
558            .unwrap();
559        let finishing = |minute| {
560            vec![
561                updated(run(PipelineStatus::Running, minute)),
562                updated(run(PipelineStatus::Success, minute)),
563            ]
564        };
565
566        let first = recorded(&h, false, finishing(1)).await;
567        let after_restart = recorded(&h, false, finishing(2)).await;
568
569        assert_eq!(first.len(), 1);
570        assert_eq!(after_restart, first);
571        let saved = devices.notification_preferences().await.unwrap()[&device.id].clone();
572        assert_eq!(
573            saved.watch_once[0].fired_at,
574            first[0].finish.map(|finish| finish.at)
575        );
576        assert!(saved.allows(
577            AlertKind::Finished,
578            &[1],
579            Some("github__1__api"),
580            first[0].finish.as_ref()
581        ));
582    }
583
584    #[tokio::test]
585    async fn every_notice_is_announced_with_a_rising_id_even_without_a_paired_device() {
586        let h = Harness::new().await;
587        let recent = Arc::new(RecentNotifications::default());
588        let (bus, rx) = recording_bus(h.bus.clone());
589        let state = seed(&h.pipeline_service, &h.provider_service).await;
590        let task = tokio::spawn(record(
591            rx,
592            state,
593            CancellationToken::new(),
594            DeviceAlertRepository::new(DatabasePool::Sqlite(h.pool.clone())),
595            DeviceRepository::new(DatabasePool::Sqlite(h.pool.clone())),
596            h.bus.clone(),
597            Arc::clone(&recent),
598        ));
599        for event in [
600            updated(run(PipelineStatus::Running, 0)),
601            updated(run(PipelineStatus::Success, 0)),
602            updated(run(PipelineStatus::Failed, 1)),
603        ] {
604            bus.emit(event).await;
605        }
606        drop(bus);
607        task.await.unwrap();
608
609        let announced: Vec<ObservedNotification> = h
610            .bus
611            .events()
612            .into_iter()
613            .filter_map(|event| match event {
614                CoreEvent::NotificationObserved { notification } => Some(notification),
615                _ => None,
616            })
617            .collect();
618        assert_eq!(
619            announced
620                .iter()
621                .map(|observed| observed.notice.kind)
622                .collect::<Vec<_>>(),
623            [AlertKind::Finished, AlertKind::Failed]
624        );
625        assert!(announced[0].id < announced[1].id);
626        assert_eq!(recent.after(announced[0].id), announced[1..]);
627        assert_eq!(h.bus.count("alert-recorded"), 0);
628    }
629}
pipedash_core/application/services/device_alerts.rs

pipedash_core/application/services/
device_alerts.rs
