1use chrono::{
2 DateTime,
3 Utc,
4};
5use serde::{
6 Deserialize,
7 Serialize,
8};
9
10#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
11#[serde(rename_all = "snake_case")]
12pub enum AlertKind {
13 Failed,
14 Degraded,
15 LongRun,
16 Outage,
17 Recovered,
18 Finished,
19 Test,
20}
21
22impl AlertKind {
23 pub fn as_str(self) -> &'static str {
24 match self {
25 Self::Failed => "failed",
26 Self::Degraded => "degraded",
27 Self::LongRun => "long_run",
28 Self::Outage => "outage",
29 Self::Recovered => "recovered",
30 Self::Finished => "finished",
31 Self::Test => "test",
32 }
33 }
34
35 pub fn parse(value: &str) -> Option<Self> {
36 Some(match value {
37 "failed" => Self::Failed,
38 "degraded" => Self::Degraded,
39 "long_run" => Self::LongRun,
40 "outage" => Self::Outage,
41 "recovered" => Self::Recovered,
42 "finished" => Self::Finished,
43 "test" => Self::Test,
44 _ => return None,
45 })
46 }
47
48 pub fn tone(self) -> &'static str {
49 match self {
50 Self::Failed | Self::Degraded => "danger",
51 Self::LongRun | Self::Outage => "warning",
52 Self::Recovered | Self::Finished | Self::Test => "success",
53 }
54 }
55}
56
57#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
58pub struct RunFinish {
59 pub at: DateTime<Utc>,
60 pub previous: Option<DateTime<Utc>>,
61}
62
63#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
64pub struct OnceWatch {
65 pub pipeline_id: String,
66 pub since: DateTime<Utc>,
67 #[serde(default, skip_serializing_if = "Option::is_none")]
68 pub fired_at: Option<DateTime<Utc>>,
69}
70
71impl OnceWatch {
72 fn matches(&self, pipeline_id: &str, finish: &RunFinish) -> bool {
73 self.pipeline_id == pipeline_id
74 && match self.fired_at {
75 Some(fired_at) => finish.at == fired_at,
76 None => {
77 finish.at >= self.since
78 && finish.previous.is_none_or(|previous| previous < self.since)
79 }
80 }
81 }
82
83 fn same_request(&self, other: &Self) -> bool {
84 self.pipeline_id == other.pipeline_id && self.since == other.since
85 }
86}
87
88#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
89#[serde(default)]
90pub struct NotificationPreferences {
91 pub failed: bool,
92 pub degraded: bool,
93 pub long_run: bool,
94 pub outage: bool,
95 pub recovered: bool,
96 pub muted_providers: Vec<i64>,
97 pub muted_pipelines: Vec<String>,
98 pub watched_pipelines: Vec<String>,
99 pub watch_once: Vec<OnceWatch>,
100}
101
102impl Default for NotificationPreferences {
103 fn default() -> Self {
104 Self {
105 failed: true,
106 degraded: true,
107 long_run: true,
108 outage: true,
109 recovered: true,
110 muted_providers: Vec::new(),
111 muted_pipelines: Vec::new(),
112 watched_pipelines: Vec::new(),
113 watch_once: Vec::new(),
114 }
115 }
116}
117
118impl NotificationPreferences {
119 pub fn allows(
120 &self, kind: AlertKind, provider_ids: &[i64], pipeline_id: Option<&str>,
121 finish: Option<&RunFinish>,
122 ) -> bool {
123 if let Some(id) = pipeline_id {
124 if kind != AlertKind::Test && self.muted_pipelines.iter().any(|muted| muted == id) {
125 return false;
126 }
127 if self.watched_pipelines.iter().any(|watched| watched == id)
128 || finish.is_some_and(|finish| {
129 self.watch_once.iter().any(|once| once.matches(id, finish))
130 })
131 {
132 return true;
133 }
134 }
135 let wanted = match kind {
136 AlertKind::Failed => self.failed,
137 AlertKind::Degraded => self.degraded,
138 AlertKind::LongRun => self.long_run,
139 AlertKind::Outage => self.outage,
140 AlertKind::Recovered => self.recovered,
141 AlertKind::Finished => false,
142 AlertKind::Test => return true,
143 };
144
145 let providers_muted = !provider_ids.is_empty()
146 && provider_ids
147 .iter()
148 .all(|id| self.muted_providers.contains(id));
149
150 wanted && !providers_muted
151 }
152
153 pub fn fire_once(&mut self, pipeline_id: &str, finish: &RunFinish) -> Vec<OnceWatch> {
154 self.watch_once
155 .iter_mut()
156 .filter(|once| once.fired_at.is_none() && once.matches(pipeline_id, finish))
157 .map(|once| {
158 once.fired_at = Some(finish.at);
159 once.clone()
160 })
161 .collect()
162 }
163
164 pub fn keep_fired(&mut self, stored: &Self) {
165 for once in &mut self.watch_once {
166 if let Some(fired) = stored
167 .watch_once
168 .iter()
169 .find(|saved| saved.same_request(once))
170 {
171 once.fired_at = once.fired_at.or(fired.fired_at);
172 }
173 }
174 }
175
176 pub fn fired_watches(&self) -> Vec<OnceWatch> {
177 self.watch_once
178 .iter()
179 .filter(|once| once.fired_at.is_some())
180 .cloned()
181 .collect()
182 }
183
184 pub fn kinds_only(&self) -> Self {
185 Self {
186 muted_providers: Vec::new(),
187 muted_pipelines: Vec::new(),
188 watched_pipelines: Vec::new(),
189 watch_once: Vec::new(),
190 ..self.clone()
191 }
192 }
193}
194
195#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
196pub struct DeviceAlert {
197 pub seq: i64,
198 pub id: String,
199 pub key: String,
200 pub kind: AlertKind,
201 pub title: String,
202 pub body: String,
203 pub pipeline_id: Option<String>,
204 pub provider_id: Option<i64>,
205 #[serde(default)]
206 pub provider_ids: Vec<i64>,
207 #[serde(default, skip_serializing_if = "Option::is_none")]
208 pub finish: Option<RunFinish>,
209 pub created_at: DateTime<Utc>,
210}
211
212impl DeviceAlert {
213 pub fn affected_providers(&self) -> Vec<i64> {
214 if self.provider_ids.is_empty() {
215 self.provider_id.into_iter().collect()
216 } else {
217 self.provider_ids.clone()
218 }
219 }
220}
221
222#[derive(Debug, Clone, PartialEq, Eq)]
223pub struct NewDeviceAlert {
224 pub key: String,
225 pub kind: AlertKind,
226 pub title: String,
227 pub body: String,
228 pub pipeline_id: Option<String>,
229 pub provider_id: Option<i64>,
230 pub provider_ids: Vec<i64>,
231 pub finish: Option<RunFinish>,
232}
233
234#[derive(Debug, Clone, PartialEq, Eq)]
235pub struct DevicePush {
236 pub device_id: String,
237 pub relay_subscription_id: String,
238 pub created_at: DateTime<Utc>,
239 pub last_success_at: Option<DateTime<Utc>>,
240 pub undelivered_seq: Option<i64>,
241 pub undelivered_max_seq: Option<i64>,
242}
243
244pub fn new_alert_id() -> String {
245 format!("{:032x}", rand::random::<u128>())
246}
247
248#[cfg(test)]
249mod tests {
250 use super::*;
251
252 #[test]
253 fn kinds_round_trip_through_their_wire_names() {
254 for kind in [
255 AlertKind::Failed,
256 AlertKind::Degraded,
257 AlertKind::LongRun,
258 AlertKind::Outage,
259 AlertKind::Recovered,
260 AlertKind::Finished,
261 AlertKind::Test,
262 ] {
263 assert_eq!(AlertKind::parse(kind.as_str()), Some(kind));
264 assert_eq!(
265 serde_json::to_value(kind).unwrap(),
266 serde_json::json!(kind.as_str())
267 );
268 }
269 assert_eq!(AlertKind::parse("nope"), None);
270 }
271
272 #[test]
273 fn tones_group_kinds_by_severity() {
274 assert_eq!(AlertKind::Failed.tone(), "danger");
275 assert_eq!(AlertKind::Degraded.tone(), "danger");
276 assert_eq!(AlertKind::LongRun.tone(), "warning");
277 assert_eq!(AlertKind::Outage.tone(), "warning");
278 assert_eq!(AlertKind::Recovered.tone(), "success");
279 assert_eq!(AlertKind::Test.tone(), "success");
280 }
281
282 #[test]
283 fn preferences_filter_by_kind_provider_and_pipeline() {
284 let preferences = NotificationPreferences {
285 long_run: false,
286 muted_providers: vec![2],
287 muted_pipelines: vec!["github__1__api".into()],
288 ..NotificationPreferences::default()
289 };
290
291 assert!(preferences.allows(AlertKind::Failed, &[1], Some("github__1__web"), None));
292 assert!(!preferences.allows(AlertKind::LongRun, &[1], Some("github__1__web"), None));
293 assert!(!preferences.allows(AlertKind::Failed, &[1], Some("github__1__api"), None));
294 assert!(!preferences.allows(AlertKind::Outage, &[2], None, None));
295 assert!(preferences.allows(AlertKind::Outage, &[2, 3], None, None));
296 assert!(preferences.allows(AlertKind::Outage, &[3], None, None));
297 assert!(preferences.allows(AlertKind::Outage, &[], None, None));
298 assert!(preferences.allows(AlertKind::Test, &[2], Some("github__1__api"), None));
299 assert!(preferences.kinds_only().allows(
300 AlertKind::Failed,
301 &[2],
302 Some("github__1__api"),
303 None
304 ));
305 }
306
307 #[test]
308 fn a_watched_pipeline_beats_provider_mutes_and_switches_but_not_its_own_mute() {
309 let preferences = NotificationPreferences {
310 failed: false,
311 long_run: false,
312 muted_providers: vec![2],
313 muted_pipelines: vec!["github__2__muted".into()],
314 watched_pipelines: vec!["github__2__api".into(), "github__2__muted".into()],
315 ..NotificationPreferences::default()
316 };
317 let watched = Some("github__2__api");
318
319 for kind in [AlertKind::Failed, AlertKind::LongRun, AlertKind::Finished] {
320 assert!(preferences.allows(kind, &[2], watched, None), "{kind:?}");
321 }
322 assert!(!preferences.allows(AlertKind::Finished, &[1], Some("github__1__web"), None));
323 assert!(!preferences.allows(AlertKind::Finished, &[], None, None));
324 assert!(!preferences.allows(AlertKind::Finished, &[2], Some("github__2__muted"), None));
325 assert!(!preferences
326 .kinds_only()
327 .allows(AlertKind::Finished, &[2], watched, None));
328 }
329
330 #[test]
331 fn a_one_time_watch_lets_only_the_first_finish_after_it_through() {
332 let since = DateTime::parse_from_rfc3339("2026-11-04T10:00:00Z")
333 .unwrap()
334 .with_timezone(&Utc);
335 let minutes = |m: i64| since + chrono::Duration::minutes(m);
336 let preferences = NotificationPreferences {
337 failed: false,
338 muted_providers: vec![1],
339 watch_once: vec![OnceWatch {
340 pipeline_id: "github__1__deploy".into(),
341 since,
342 fired_at: None,
343 }],
344 ..NotificationPreferences::default()
345 };
346 let allows = |kind, finish: RunFinish| {
347 preferences.allows(kind, &[1], Some("github__1__deploy"), Some(&finish))
348 };
349
350 assert!(allows(
351 AlertKind::Finished,
352 RunFinish {
353 at: minutes(5),
354 previous: Some(minutes(-30)),
355 }
356 ));
357 assert!(allows(
358 AlertKind::Failed,
359 RunFinish {
360 at: minutes(5),
361 previous: None,
362 }
363 ));
364 assert!(!allows(
365 AlertKind::Finished,
366 RunFinish {
367 at: minutes(9),
368 previous: Some(minutes(5)),
369 }
370 ));
371 assert!(!allows(
372 AlertKind::Finished,
373 RunFinish {
374 at: minutes(-1),
375 previous: None,
376 }
377 ));
378 assert!(!preferences.allows(AlertKind::Finished, &[1], Some("github__1__deploy"), None));
379 assert!(!preferences.allows(
380 AlertKind::Finished,
381 &[1],
382 Some("github__1__other"),
383 Some(&RunFinish {
384 at: minutes(5),
385 previous: None,
386 })
387 ));
388 }
389
390 #[test]
391 fn a_used_one_time_watch_only_lets_its_own_finish_through_even_after_a_restart() {
392 let since = DateTime::parse_from_rfc3339("2026-11-04T10:00:00Z")
393 .unwrap()
394 .with_timezone(&Utc);
395 let minutes = |m: i64| since + chrono::Duration::minutes(m);
396 let request = OnceWatch {
397 pipeline_id: "github__1__deploy".into(),
398 since,
399 fired_at: None,
400 };
401 let mut stored = NotificationPreferences {
402 watch_once: vec![request.clone()],
403 ..NotificationPreferences::default()
404 };
405 let first = RunFinish {
406 at: minutes(5),
407 previous: None,
408 };
409 let after_restart = RunFinish {
410 at: minutes(40),
411 previous: None,
412 };
413
414 assert_eq!(stored.fire_once("github__1__deploy", &first).len(), 1);
415 assert!(stored
416 .fire_once("github__1__deploy", &after_restart)
417 .is_empty());
418
419 let mut resent = NotificationPreferences {
420 watch_once: vec![
421 request,
422 OnceWatch {
423 pipeline_id: "github__1__deploy".into(),
424 since: minutes(50),
425 fired_at: None,
426 },
427 ],
428 ..NotificationPreferences::default()
429 };
430 resent.keep_fired(&stored);
431 let allows = |finish: &RunFinish| {
432 resent.watch_once[..1]
433 .iter()
434 .any(|once| once.matches("github__1__deploy", finish))
435 };
436
437 assert!(allows(&first));
438 assert!(!allows(&after_restart));
439 assert_eq!(resent.watch_once[1].fired_at, None);
440 }
441
442 #[test]
443 fn missing_preference_fields_keep_their_defaults() {
444 let preferences: NotificationPreferences =
445 serde_json::from_value(serde_json::json!({ "outage": false, "future": true })).unwrap();
446
447 assert_eq!(
448 preferences,
449 NotificationPreferences {
450 outage: false,
451 ..NotificationPreferences::default()
452 }
453 );
454 }
455
456 #[test]
457 fn alert_ids_are_32_hex_characters() {
458 let id = new_alert_id();
459 assert_eq!(id.len(), 32);
460 assert!(id.bytes().all(|b| b.is_ascii_hexdigit()));
461 assert_ne!(id, new_alert_id());
462 }
463}