1use std::sync::Arc;
2use std::time::Duration;
3
4use futures_util::future::join_all;
5use pipedash_core::application::{
6 PipelineService,
7 ProviderService,
8};
9use pipedash_core::domain::{
10 DevicePush,
11 Pipeline,
12 PipelineRun,
13 PipelineStatus,
14 PushSubscription,
15};
16use pipedash_core::event::CoreEvent;
17use pipedash_core::infrastructure::database::PushRepository;
18use pipedash_core::infrastructure::{
19 ConfigKey,
20 ConfigState,
21 HttpClientManager,
22};
23use pipedash_core::CoreContext;
24use pipedash_plugin_api::ClientOptions;
25use serde::Serialize;
26use tokio::sync::broadcast::error::RecvError;
27
28use crate::error::{
29 ApiResult,
30 AppError,
31};
32use crate::state::AppState;
33use crate::ws::closed;
34
35const RELAY_TIMEOUT: Duration = Duration::from_secs(10);
36const RUN_LOOKUP_TIMEOUT: Duration = Duration::from_secs(10);
37const RUN_LOOKUP_LIMIT: usize = 3;
38const FAILED_SUFFIX: &str = " failed";
39const DEGRADED_SUFFIX: &str = " is degraded";
40const MAX_TITLE_CHARS: usize = 120;
41const MAX_BODY_CHARS: usize = 400;
42
43#[derive(Debug, Clone, Copy, PartialEq, Eq)]
44pub enum RelayOutcome {
45 Delivered,
46 Gone,
47 RateLimited,
48 Failed,
49}
50
51fn relay_outcome(status: u16) -> RelayOutcome {
52 match status {
53 202 => RelayOutcome::Delivered,
54 404 | 410 => RelayOutcome::Gone,
55 429 => RelayOutcome::RateLimited,
56 _ => RelayOutcome::Failed,
57 }
58}
59
60pub struct Alert {
61 pub title: String,
62 pub body: String,
63 pub thread_id: Option<String>,
64 pub pipeline_id: Option<String>,
65 pub provider_id: Option<u64>,
66 pub run_number: Option<u64>,
67}
68
69#[derive(Serialize)]
70struct RelayNotification<'a> {
71 subscription_id: &'a str,
72 title: &'a str,
73 body: &'a str,
74 #[serde(skip_serializing_if = "Option::is_none")]
75 thread_id: Option<&'a str>,
76 #[serde(skip_serializing_if = "Option::is_none")]
77 pipeline_id: Option<&'a str>,
78 #[serde(skip_serializing_if = "Option::is_none")]
79 provider_id: Option<u64>,
80 #[serde(skip_serializing_if = "Option::is_none")]
81 run_number: Option<u64>,
82}
83
84#[derive(Serialize)]
85struct RelayAlert<'a> {
86 subscription_id: &'a str,
87 alert_id: &'a str,
88}
89
90pub fn normalize_relay_url(url: &str) -> Option<String> {
91 let url = url.trim().trim_end_matches('/');
92
93 (!url.is_empty()).then(|| url.to_owned())
94}
95
96async fn post_relay<T: Serialize>(
97 http: &HttpClientManager, relay_url: &str, body: &T,
98) -> RelayOutcome {
99 let client = match http.client_for(&ClientOptions::default()) {
100 Ok(client) => client,
101 Err(error) => {
102 tracing::warn!(%error, "Couldn't build the push relay client");
103 return RelayOutcome::Failed;
104 }
105 };
106
107 let response = client
108 .post(format!("{relay_url}/v1/notifications"))
109 .timeout(RELAY_TIMEOUT)
110 .json(body)
111 .send()
112 .await;
113
114 match response {
115 Ok(response) => relay_outcome(response.status().as_u16()),
116 Err(error) => {
117 tracing::warn!(error = %error.without_url(), "Push relay request failed");
118 RelayOutcome::Failed
119 }
120 }
121}
122
123pub async fn send_alert(
124 http: &HttpClientManager, relay_url: &str, subscription_id: &str, alert_id: &str,
125) -> RelayOutcome {
126 post_relay(
127 http,
128 relay_url,
129 &RelayAlert {
130 subscription_id,
131 alert_id,
132 },
133 )
134 .await
135}
136
137fn truncate(text: &str, max_chars: usize) -> String {
138 if text.chars().count() <= max_chars {
139 return text.to_owned();
140 }
141
142 let mut truncated: String = text.chars().take(max_chars.saturating_sub(1)).collect();
143 truncated.push('…');
144
145 truncated
146}
147
148fn failure_alert(pipeline: &Pipeline, provider: &str, run: Option<&PipelineRun>) -> Alert {
149 let branch = run
150 .and_then(|run| run.branch.as_deref())
151 .or(pipeline.branch.as_deref());
152 let details = [
153 Some(provider.to_owned()),
154 branch.map(str::to_owned),
155 run.map(|run| format!("#{}", run.run_number)),
156 pipeline
157 .resource_state
158 .as_ref()
159 .and_then(|state| state.message.clone()),
160 ]
161 .into_iter()
162 .flatten()
163 .collect::<Vec<_>>()
164 .join(" · ");
165 let suffix = match pipeline.status {
166 PipelineStatus::Degraded => DEGRADED_SUFFIX,
167 _ => FAILED_SUFFIX,
168 };
169
170 Alert {
171 title: format!(
172 "{}{suffix}",
173 truncate(&pipeline.name, MAX_TITLE_CHARS - suffix.len())
174 ),
175 body: truncate(&details, MAX_BODY_CHARS),
176 thread_id: Some(pipeline.id.clone()),
177 pipeline_id: Some(pipeline.id.clone()),
178 provider_id: u64::try_from(pipeline.provider_id).ok(),
179 run_number: run.and_then(|run| u64::try_from(run.run_number).ok()),
180 }
181}
182
183pub struct PushServices {
184 pub subscriptions: PushRepository,
185 config: Arc<ConfigState>,
186 http: Arc<HttpClientManager>,
187 providers: Arc<ProviderService>,
188 pipelines: Arc<PipelineService>,
189}
190
191impl PushServices {
192 pub async fn from_state(state: &AppState) -> ApiResult<Self> {
193 let inner = state.inner.read().await;
194 let core = inner.core.as_ref().ok_or_else(AppError::not_initialized)?;
195
196 Ok(Self {
197 subscriptions: PushRepository::new(
198 core.provider_service.repository().cache_pool().clone(),
199 ),
200 config: Arc::clone(&core.config_state),
201 http: Arc::clone(&core.http_client_manager),
202 providers: Arc::clone(&core.provider_service),
203 pipelines: Arc::clone(&core.pipeline_service),
204 })
205 }
206
207 pub async fn relay_url(&self) -> Option<String> {
208 normalize_relay_url(&self.config.get_string(ConfigKey::PushRelayUrl).await)
209 }
210
211 pub async fn deliver(
212 &self, relay_url: &str, subscription: &PushSubscription, alert: &Alert,
213 ) -> RelayOutcome {
214 let outcome = self.send(relay_url, subscription, alert).await;
215 let result = match outcome {
216 RelayOutcome::Delivered => self.subscriptions.mark_success(&subscription.id).await,
217 RelayOutcome::Gone => self
218 .subscriptions
219 .delete(&subscription.id)
220 .await
221 .map(|_| ()),
222 RelayOutcome::RateLimited | RelayOutcome::Failed => {
223 tracing::warn!(?outcome, "The push relay did not accept a notification");
224 Ok(())
225 }
226 };
227 if let Err(error) = result {
228 tracing::warn!(%error, "Couldn't update a push subscription");
229 }
230
231 outcome
232 }
233
234 async fn send(
235 &self, relay_url: &str, subscription: &PushSubscription, alert: &Alert,
236 ) -> RelayOutcome {
237 post_relay(
238 &self.http,
239 relay_url,
240 &RelayNotification {
241 subscription_id: &subscription.relay_subscription_id,
242 title: &alert.title,
243 body: &alert.body,
244 thread_id: alert.thread_id.as_deref(),
245 pipeline_id: alert.pipeline_id.as_deref(),
246 provider_id: alert.provider_id,
247 run_number: alert.run_number,
248 },
249 )
250 .await
251 }
252
253 async fn failed_run(&self, pipeline: &Pipeline) -> Option<PipelineRun> {
254 let history = self
255 .pipelines
256 .fetch_run_history(&pipeline.id, RUN_LOOKUP_LIMIT);
257 let runs = match tokio::time::timeout(RUN_LOOKUP_TIMEOUT, history).await {
258 Ok(Ok(runs)) => runs,
259 Ok(Err(error)) => {
260 tracing::debug!(%error, "Couldn't look up the failed run for a push");
261 return None;
262 }
263 Err(_) => return None,
264 };
265
266 runs.into_iter()
267 .filter(|run| run.status == PipelineStatus::Failed)
268 .max_by_key(|run| run.run_number)
269 }
270
271 async fn failure_alert(&self, pipeline: &Pipeline) -> Alert {
272 let provider = match self
273 .providers
274 .repository()
275 .config_backend()
276 .get_provider(pipeline.provider_id)
277 .await
278 {
279 Ok(Some(config)) => config.name,
280 _ => pipeline.provider_type.clone(),
281 };
282 let run = match pipeline.status {
283 PipelineStatus::Failed => self.failed_run(pipeline).await,
284 _ => None,
285 };
286
287 failure_alert(pipeline, &provider, run.as_ref())
288 }
289
290 async fn notify_failures(&self, pipelines: &[Pipeline]) {
291 let Some(relay_url) = self.relay_url().await else {
292 return;
293 };
294 let subscriptions = match self.subscriptions.list().await {
295 Ok(subscriptions) if !subscriptions.is_empty() => subscriptions,
296 Ok(_) => return,
297 Err(error) => {
298 tracing::warn!(%error, "Couldn't load push subscriptions");
299 return;
300 }
301 };
302
303 for pipeline in pipelines {
304 let alert = self.failure_alert(pipeline).await;
305 join_all(
306 subscriptions
307 .iter()
308 .map(|subscription| self.deliver(&relay_url, subscription, &alert)),
309 )
310 .await;
311 }
312 }
313}
314
315pub fn spawn_failure_notifier(state: AppState) {
316 let mut events = state.ws_event_bus.subscribe();
317
318 tokio::spawn(async move {
319 loop {
320 match events.recv().await {
321 Ok(CoreEvent::PipelinesFailed { pipelines }) => {
322 let state = state.clone();
323 tokio::spawn(async move {
324 if let Ok(services) = PushServices::from_state(&state).await {
325 services.notify_failures(&pipelines).await;
326 }
327 });
328 }
329 Ok(_) => {}
330 Err(RecvError::Lagged(skipped)) => {
331 tracing::warn!(skipped, "The push notifier missed events");
332 }
333 Err(RecvError::Closed) => break,
334 }
335 }
336 });
337}
338
339pub async fn deliver_alert(
340 core: &CoreContext, relay_url: &str, push: &DevicePush, alert_id: &str, seq: Option<i64>,
341) -> RelayOutcome {
342 let outcome = send_alert(
343 &core.http_client_manager,
344 relay_url,
345 &push.relay_subscription_id,
346 alert_id,
347 )
348 .await;
349 let pushes = core.device_pushes();
350 let result = match outcome {
351 RelayOutcome::Delivered => pushes.mark_success(&push.device_id).await,
352 RelayOutcome::Gone => pushes
353 .delete_matching(&push.device_id, &push.relay_subscription_id)
354 .await
355 .map(|_| ()),
356 RelayOutcome::RateLimited | RelayOutcome::Failed => {
357 tracing::debug!(?outcome, "The push relay did not accept an alert");
358 match seq {
359 Some(seq) => {
360 pushes
361 .mark_undelivered(&push.device_id, &push.relay_subscription_id, seq)
362 .await
363 }
364 None => Ok(()),
365 }
366 }
367 };
368 if let Err(error) = result {
369 tracing::warn!(%error, "Couldn't update a device push registration");
370 }
371
372 outcome
373}
374
375async fn push_alert(state: &AppState, alert_id: &str, seq: i64) {
376 let Some(relay_url) = state.push_relay_url().await else {
377 return;
378 };
379 let Ok(core) = state.core().await else {
380 return;
381 };
382 let pushes = match core.device_pushes().list().await {
383 Ok(pushes) => pushes,
384 Err(error) => {
385 tracing::warn!(%error, "Couldn't load device push registrations");
386 return;
387 }
388 };
389
390 join_all(
391 pushes
392 .iter()
393 .map(|push| deliver_alert(&core, &relay_url, push, alert_id, Some(seq))),
394 )
395 .await;
396}
397
398pub fn spawn_alert_push(state: AppState) {
399 let mut events = state.ws_event_bus.subscribe();
400 let mut closing = state.closing.subscribe();
401
402 tokio::spawn(async move {
403 loop {
404 let event = tokio::select! {
405 event = events.recv() => event,
406 () = closed(&mut closing) => break,
407 };
408 match event {
409 Ok(CoreEvent::AlertRecorded { id, seq }) => {
410 let state = state.clone();
411 tokio::spawn(async move { push_alert(&state, &id, seq).await });
412 }
413 Ok(_) => {}
414 Err(RecvError::Lagged(skipped)) => {
415 tracing::warn!(skipped, "The alert push sender missed events");
416 }
417 Err(RecvError::Closed) => break,
418 }
419 }
420 });
421}
422
423#[cfg(test)]
424mod tests {
425 use std::future::Future;
426
427 use chrono::Utc;
428 use pipedash_core::event::EventBus;
429
430 use super::*;
431 use crate::test_support::{
432 hosted,
433 stub_server,
434 };
435
436 fn pipeline(name: &str, branch: Option<&str>) -> Pipeline {
437 serde_json::from_value(serde_json::json!({
438 "id": "github__1__acme/app/ci.yml",
439 "provider_id": 1,
440 "provider_type": "github",
441 "name": name,
442 "status": "failed",
443 "last_run": null,
444 "last_updated": Utc::now(),
445 "repository": "acme/app",
446 "branch": branch,
447 "workflow_file": null,
448 }))
449 .unwrap()
450 }
451
452 fn run(number: i64, branch: Option<&str>) -> PipelineRun {
453 serde_json::from_value(serde_json::json!({
454 "id": number.to_string(),
455 "pipeline_id": "github__1__acme/app/ci.yml",
456 "run_number": number,
457 "status": "failed",
458 "started_at": Utc::now(),
459 "logs_url": "",
460 "branch": branch,
461 }))
462 .unwrap()
463 }
464
465 #[test]
466 fn a_failure_names_the_pipeline_provider_branch_run_and_thread() {
467 let alert = failure_alert(
468 &pipeline("CI", Some("main")),
469 "GitHub",
470 Some(&run(42, None)),
471 );
472
473 assert_eq!(alert.title, "CI failed");
474 assert_eq!(alert.body, "GitHub · main · #42");
475 assert_eq!(
476 alert.thread_id.as_deref(),
477 Some("github__1__acme/app/ci.yml")
478 );
479 assert_eq!(
480 alert.pipeline_id.as_deref(),
481 Some("github__1__acme/app/ci.yml")
482 );
483 assert_eq!(alert.provider_id, Some(1));
484 assert_eq!(alert.run_number, Some(42));
485 }
486
487 #[test]
488 fn the_run_branch_wins_and_missing_parts_are_left_out() {
489 let with_run_branch = failure_alert(
490 &pipeline("CI", Some("main")),
491 "GitHub",
492 Some(&run(7, Some("release"))),
493 );
494 let provider_only = failure_alert(&pipeline("CI", None), "GitHub", None);
495
496 assert_eq!(with_run_branch.body, "GitHub · release · #7");
497 assert_eq!(provider_only.body, "GitHub");
498 }
499
500 #[test]
501 fn long_names_are_cut_so_the_relay_accepts_them() {
502 let alert = failure_alert(
503 &pipeline(&"é".repeat(500), None),
504 "p".repeat(1000).as_str(),
505 None,
506 );
507
508 assert_eq!(alert.title.chars().count(), MAX_TITLE_CHARS);
509 assert!(alert.title.ends_with("… failed"));
510 assert_eq!(alert.body.chars().count(), MAX_BODY_CHARS);
511 assert_eq!(truncate("short", 10), "short");
512 }
513
514 #[test]
515 fn a_degraded_pipeline_says_so_and_explains_why() {
516 let mut degraded = pipeline("web", None);
517 degraded.status = PipelineStatus::Degraded;
518 degraded.resource_state = Some(pipedash_core::domain::ResourceState::new(
519 "Degraded",
520 Some("Pod web/web-7f9c: back-off restarting failed container"),
521 ));
522
523 let alert = failure_alert(°raded, "Argo CD", None);
524
525 assert_eq!(alert.title, "web is degraded");
526 assert_eq!(
527 alert.body,
528 "Argo CD · Pod web/web-7f9c: back-off restarting failed container"
529 );
530 assert_eq!(alert.run_number, None);
531 }
532
533 #[test]
534 fn relay_statuses_map_to_outcomes() {
535 assert_eq!(relay_outcome(202), RelayOutcome::Delivered);
536 assert_eq!(relay_outcome(404), RelayOutcome::Gone);
537 assert_eq!(relay_outcome(410), RelayOutcome::Gone);
538 assert_eq!(relay_outcome(429), RelayOutcome::RateLimited);
539 assert_eq!(relay_outcome(400), RelayOutcome::Failed);
540 assert_eq!(relay_outcome(502), RelayOutcome::Failed);
541 }
542
543 #[test]
544 fn relay_urls_are_trimmed_and_blank_means_none() {
545 assert_eq!(
546 normalize_relay_url(" https://relay.example.com/ ").as_deref(),
547 Some("https://relay.example.com")
548 );
549 assert_eq!(normalize_relay_url(" / "), None);
550 assert_eq!(normalize_relay_url(""), None);
551 }
552
553 async fn eventually<F, Fut>(what: &str, mut check: F)
554 where
555 F: FnMut() -> Fut,
556 Fut: Future<Output = bool>,
557 {
558 let wait = async {
559 while !check().await {
560 tokio::time::sleep(Duration::from_millis(20)).await;
561 }
562 };
563
564 if tokio::time::timeout(Duration::from_secs(10), wait)
565 .await
566 .is_err()
567 {
568 panic!("timed out waiting for {what}");
569 }
570 }
571
572 async fn subscribed(state: &AppState, subscriptions: &[&str]) {
573 let core = state.core().await.unwrap();
574 for subscription in subscriptions {
575 let (device, _) = core.pair_device(subscription, "ios").await.unwrap();
576 core.device_pushes()
577 .set(&device.id, subscription)
578 .await
579 .unwrap();
580 }
581 }
582
583 fn recorded(id: &str) -> CoreEvent {
584 CoreEvent::AlertRecorded {
585 id: id.into(),
586 seq: 1,
587 }
588 }
589
590 #[tokio::test]
591 async fn a_recorded_alert_reaches_every_subscribed_device_with_only_its_id() {
592 let host = hosted("alert-push").await;
593 let relay = stub_server().await;
594 let state = host
595 .state
596 .clone()
597 .with_push_relay(&format!("{}/", relay.url));
598 subscribed(&state, &["sub-1", "sub-2"]).await;
599 spawn_alert_push(state.clone());
600
601 state.ws_event_bus.emit(recorded("a1")).await;
602 eventually("both pushes", || async { relay.requests().len() == 2 }).await;
603
604 let mut requests = relay.requests();
605 requests.sort_by_key(|request| request["subscription_id"].to_string());
606 assert_eq!(
607 requests,
608 vec![
609 serde_json::json!({ "subscription_id": "sub-1", "alert_id": "a1" }),
610 serde_json::json!({ "subscription_id": "sub-2", "alert_id": "a1" }),
611 ]
612 );
613 let core = state.core().await.unwrap();
614 eventually("both successes to be recorded", || async {
615 let pushes = core.device_pushes().list().await.unwrap();
616 pushes.len() == 2 && pushes.iter().all(|push| push.last_success_at.is_some())
617 })
618 .await;
619 }
620
621 #[tokio::test]
622 async fn a_device_the_relay_no_longer_knows_is_forgotten_after_an_alert_push() {
623 let host = hosted("alert-push-gone").await;
624 let relay = stub_server().await;
625 relay.reply_with(404);
626 let state = host.state.clone().with_push_relay(&relay.url);
627 subscribed(&state, &["sub-1"]).await;
628 spawn_alert_push(state.clone());
629
630 state.ws_event_bus.emit(recorded("a1")).await;
631
632 let core = state.core().await.unwrap();
633 eventually("the device to be forgotten", || async {
634 core.device_pushes().list().await.unwrap().is_empty()
635 })
636 .await;
637 assert_eq!(relay.requests().len(), 1);
638 }
639
640 #[tokio::test]
641 async fn nothing_is_sent_without_a_relay_or_after_the_host_stops() {
642 let host = hosted("alert-push-idle").await;
643 let relay = stub_server().await;
644 subscribed(&host.state, &["sub-1"]).await;
645 spawn_alert_push(host.state.clone());
646 host.state.ws_event_bus.emit(recorded("a1")).await;
647 tokio::time::sleep(Duration::from_millis(300)).await;
648 assert!(relay.requests().is_empty());
649
650 let state = host.state.clone().with_push_relay(&relay.url);
651 spawn_alert_push(state.clone());
652 state.closing.send_replace(true);
653 tokio::time::sleep(Duration::from_millis(100)).await;
654 state.ws_event_bus.emit(recorded("a2")).await;
655 tokio::time::sleep(Duration::from_millis(300)).await;
656 assert!(relay.requests().is_empty());
657 }
658
659 #[tokio::test]
660 async fn a_host_reads_the_relay_from_its_builder() {
661 let host = hosted("alert-relay-builder").await;
662 assert_eq!(host.state.push_relay_url().await, None);
663 let state = host
664 .state
665 .clone()
666 .with_push_relay("https://relay.example.org/");
667 assert_eq!(
668 state.push_relay_url().await.as_deref(),
669 Some("https://relay.example.org")
670 );
671 }
672}
673
674#[cfg(test)]
675mod gone_tests {
676 use super::*;
677 use crate::test_support::{
678 hosted,
679 stub_server,
680 };
681
682 #[tokio::test]
683 async fn a_stale_gone_answer_keeps_the_registration_that_replaced_it() {
684 let host = hosted("gone-replaced").await;
685 let relay = stub_server().await;
686 relay.reply_with(410);
687 let core = host.state.core().await.unwrap();
688 let (device, _) = core.pair_device("Pixel", "ios").await.unwrap();
689 core.device_pushes().set(&device.id, "sub-a").await.unwrap();
690 let stale = core.device_pushes().get(&device.id).await.unwrap().unwrap();
691 core.device_pushes().set(&device.id, "sub-b").await.unwrap();
692
693 let outcome = deliver_alert(&core, &relay.url, &stale, "a1", None).await;
694
695 assert_eq!(outcome, RelayOutcome::Gone);
696 assert_eq!(relay.requests().len(), 1);
697 let kept = core.device_pushes().get(&device.id).await.unwrap().unwrap();
698 assert_eq!(kept.relay_subscription_id, "sub-b");
699
700 let current = kept;
701 deliver_alert(&core, &relay.url, ¤t, "a2", None).await;
702 assert!(core
703 .device_pushes()
704 .get(&device.id)
705 .await
706 .unwrap()
707 .is_none());
708 }
709}