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