diff --git a/fluxer_push/src/delivery.rs b/fluxer_push/src/delivery.rs index 6bf6619e5..362f45e7d 100644 --- a/fluxer_push/src/delivery.rs +++ b/fluxer_push/src/delivery.rs @@ -279,14 +279,11 @@ async fn run_subject( let job_transport = transport.clone(); running.spawn(async move { let _permit = permit; - let mut result = run_job(&job_state, &job_sends, job).await; + let result = run_job(&job_state, &job_sends, job).await; if matches!(result, Answer::Done) { for claim in claims { claim.done(); } - if recipients_running { - result = Answer::NotDone(RUNNING); - } } answer(&job_transport, reply_to, result).await; }); diff --git a/fluxer_push/src/payload.rs b/fluxer_push/src/payload.rs index b74b72f57..449ee2f45 100644 --- a/fluxer_push/src/payload.rs +++ b/fluxer_push/src/payload.rs @@ -11,7 +11,8 @@ const CLEAR_ACTION: &str = "clear_channel"; const RING_TYPE: &str = "call_ring"; const FALLBACK_TAG: &str = "fluxer-message"; const FALLBACK_TITLE: &str = "Fluxer"; -const APNS_CATEGORY: &str = "FLUXER_MESSAGE"; +const APNS_CATEGORY: &str = "fluxer_message"; +const FCM_CLICK_ACTION: &str = "FLUXER_MESSAGE"; const APNS_SOUND: &str = "default"; const APNS_ALERT_EXPIRATION_SECONDS: i64 = 86_400; const APNS_BACKGROUND_EXPIRATION_SECONDS: i64 = 86_400; @@ -25,7 +26,9 @@ const RING_CALLER_NAME_KEY: &str = "caller_name"; const RING_CALLER_AVATAR_KEY: &str = "caller_avatar_url"; const RING_AVATAR_KEYS: [&str; 1] = [RING_CALLER_AVATAR_KEY]; const RING_IDENTITY_KEYS: [&str; 2] = [RING_CALLER_ID_KEY, RING_CALLER_NAME_KEY]; -const MINIMAL_DATA_KEYS: [&str; 7] = [ +const MINIMAL_DATA_KEYS: [&str; 9] = [ + "type", + "action", "channel_id", "message_id", "guild_id", @@ -201,7 +204,7 @@ fn fcm_notification_message(device_token: &str, envelope: &Value) -> Value { let mut android_notification = json!({ "channel_id": "fluxer_default_push", "tag": tag, - "click_action": APNS_CATEGORY, + "click_action": FCM_CLICK_ACTION, }); put_image(&mut android_notification, image_url); json!({ @@ -262,9 +265,7 @@ pub fn apns_payload(envelope: &Value) -> Value { if let Some(badge) = badge_number(data.and_then(|data| data.get("badge_count"))) { aps.insert("badge".to_owned(), badge.into()); } - if image_url.is_some() { - aps.insert("mutable-content".to_owned(), 1.into()); - } + aps.insert("mutable-content".to_owned(), 1.into()); payload.insert("title".to_owned(), title.into()); payload.insert("body".to_owned(), body.into()); payload.remove("url"); @@ -591,3 +592,95 @@ fn non_empty(value: &Value) -> Option<&str> { fn serialize(value: &Value) -> Vec { serde_json::to_vec(value).expect("a json value serialises") } + +#[cfg(test)] +mod tests { + use super::*; + use crate::job::{ClearJob, MessageJob, NotificationFields}; + + const CHANNEL_ID: &str = "9876543210987654321"; + const MESSAGE_ID: &str = "1122334455667788990"; + const USER_ID: &str = "1234567890123456789"; + + fn message_job(image_url: Option<&str>) -> MessageJob { + MessageJob { + v: 1, + config_version: 7, + guild_id: "0".to_owned(), + channel_id: CHANNEL_ID.to_owned(), + message_id: MESSAGE_ID.to_owned(), + notification: NotificationFields { + title: "Elias".to_owned(), + body: "see you tomorrow".to_owned(), + icon: "https://media.fluxer.app/avatars/1/a.webp".to_owned(), + badge: "https://media.fluxer.app/badge.png".to_owned(), + tag: format!("channel:{CHANNEL_ID}:{MESSAGE_ID}"), + notification_tag: format!("channel:{CHANNEL_ID}"), + url: format!("/channels/@me/{CHANNEL_ID}"), + image_url: image_url.map(str::to_owned), + }, + user_ids: vec![USER_ID.to_owned()], + } + } + + fn clear_job() -> ClearJob { + ClearJob { + v: 1, + config_version: 7, + user_id: USER_ID.to_owned(), + channel_id: CHANNEL_ID.to_owned(), + message_id: MESSAGE_ID.to_owned(), + } + } + + fn apns_for(image_url: Option<&str>) -> Value { + apns_payload(&web_push_message(&message_job(image_url), USER_ID, 3)) + } + + #[test] + fn a_message_without_an_image_still_runs_the_notification_service_extension() { + assert_eq!(apns_for(None)["aps"]["mutable-content"], json!(1)); + } + + #[test] + fn a_message_with_an_image_still_runs_the_notification_service_extension() { + assert_eq!( + apns_for(Some("https://media.fluxer.app/proxy/attachment.webp"))["aps"]["mutable-content"], + json!(1) + ); + } + + #[test] + fn a_background_clear_never_runs_the_notification_service_extension() { + let payload = apns_payload(&web_push_clear(&clear_job(), 3)); + assert_eq!(payload["aps"], json!({"content-available": 1})); + } + + #[test] + fn the_apns_category_is_the_identifier_the_app_registers() { + assert_eq!(apns_for(None)["aps"]["category"], json!("fluxer_message")); + } + + #[test] + fn the_android_click_action_keeps_its_own_identifier() { + let message = fcm_message( + "device-token", + &web_push_message(&message_job(None), USER_ID, 3), + ); + assert_eq!( + message["message"]["android"]["notification"]["click_action"], + json!("FLUXER_MESSAGE") + ); + } + + #[test] + fn a_shrunk_clear_still_tells_the_client_it_is_a_clear() { + let envelope = web_push_clear(&clear_job(), 3); + let budget = serialize(&envelope).len() - 1; + let (bytes, step) = fit(&envelope, budget); + assert_eq!(step, Some(PayloadShrink::Minimal)); + let shrunk: Value = serde_json::from_slice(&bytes).expect("the shrunk payload is json"); + assert_eq!(shrunk["data"]["type"], json!(CLEAR_TYPE)); + assert_eq!(shrunk["data"]["action"], json!(CLEAR_ACTION)); + } +} diff --git a/fluxer_push/src/providers/mod.rs b/fluxer_push/src/providers/mod.rs index ef224aeeb..9f4b27cfb 100644 --- a/fluxer_push/src/providers/mod.rs +++ b/fluxer_push/src/providers/mod.rs @@ -91,15 +91,12 @@ pub async fn send(state: &AppState, sub: &Subscription, envelope: &Value) -> Sen return SendOutcome::permanent("unsupported_platform"); }; let started_ms = now_ms(); - let direct = own_relay::parse(&sub.endpoint, &state.cfg.own_relay_hosts); + let direct = in_process_hop(&sub.endpoint, &state.cfg.own_relay_hosts); let outcome = match (route, direct) { (Route::WebPush, Some(hop)) => { let hopped = hop.as_subscription(sub); state.metrics.record_own_relay_shortcut(); - match hop.leg { - own_relay::Leg::Fcm => fcm::send(state, &hopped, envelope).await, - _ => apns::send(state, &hopped, envelope).await, - } + apns::send(state, &hopped, envelope).await } (Route::WebPush, None) => web_push::send(state, sub, envelope).await, (Route::LegacyApns, _) => apns::send(state, sub, envelope).await, @@ -116,6 +113,10 @@ pub async fn send(state: &AppState, sub: &Subscription, envelope: &Value) -> Sen outcome } +fn in_process_hop(endpoint: &str, hosts: &[String]) -> Option { + own_relay::parse(endpoint, hosts).filter(|hop| matches!(hop.leg, own_relay::Leg::Apns)) +} + fn route_label(route: Route) -> DeliveryRoute { match route { Route::WebPush => DeliveryRoute::WebPush, @@ -142,3 +143,32 @@ fn result_of(outcome: &SendOutcome) -> SendResult { SendOutcome::Transient { .. } => SendResult::Transient, } } + +#[cfg(test)] +mod hop_tests { + use super::*; + + const TOKEN: &str = "3dbc5a5ef1a1c1666afc26f466e1b3ebaaf4c66d92dddeb0fd1b69c49641d4cd"; + + fn ours() -> Vec { + vec!["push.fluxer.com".to_owned()] + } + + #[test] + fn only_the_plain_apns_leg_is_delivered_in_process() { + let apns = format!("https://push.fluxer.com/relay/v1/apns/canary/production/{TOKEN}"); + assert!(in_process_hop(&apns, &ours()).is_some()); + } + + #[test] + fn an_fcm_relay_endpoint_keeps_its_encrypted_network_hop() { + let fcm = "https://push.fluxer.com/relay/v1/fcm/canary/tok%3AAPA91bExample"; + assert!(in_process_hop(fcm, &ours()).is_none()); + } + + #[test] + fn a_pushkit_relay_endpoint_keeps_its_voip_topic() { + let voip = format!("https://push.fluxer.com/relay/v1/apns-voip/canary/production/{TOKEN}"); + assert!(in_process_hop(&voip, &ours()).is_none()); + } +} diff --git a/fluxer_push/src/providers/own_relay.rs b/fluxer_push/src/providers/own_relay.rs index 251a32e96..427cef732 100644 --- a/fluxer_push/src/providers/own_relay.rs +++ b/fluxer_push/src/providers/own_relay.rs @@ -140,6 +140,27 @@ mod tests { assert_eq!(hop.device_token, "dYC_x9gXTjyyrG8_Aw3nUM:APA91bExample"); } + #[test] + fn a_voip_endpoint_on_our_own_relay_keeps_its_own_leg() { + let hop = parse( + &format!("https://push.fluxer.com/relay/v1/apns-voip/canary/production/{TOKEN}"), + &ours(), + ) + .expect("a voip endpoint parses"); + assert_eq!(hop.leg, Leg::ApnsVoip); + } + + #[test] + fn only_the_apns_alert_leg_is_taken_in_process() { + let shortcut = |path: &str| { + parse(&format!("https://push.fluxer.com/relay/v1/{path}"), &ours()) + .filter(|hop| matches!(hop.leg, Leg::Apns)) + }; + assert!(shortcut(&format!("apns/canary/production/{TOKEN}")).is_some()); + assert!(shortcut(&format!("apns-voip/canary/production/{TOKEN}")).is_none()); + assert!(shortcut("fcm/canary/dYC_x9gXTjyyrG8_Aw3nUM%3AAPA91bExample").is_none()); + } + #[test] fn a_relay_we_do_not_operate_is_left_on_the_network_path() { let endpoint = format!("https://push.example.org/relay/v1/apns/canary/production/{TOKEN}"); diff --git a/fluxer_push/src/providers/web_push.rs b/fluxer_push/src/providers/web_push.rs index 46b50de16..579e98719 100644 --- a/fluxer_push/src/providers/web_push.rs +++ b/fluxer_push/src/providers/web_push.rs @@ -37,6 +37,7 @@ const OCTET_STREAM: &str = "application/octet-stream"; const AES128GCM: &str = "aes128gcm"; const NOT_FOUND: u16 = 404; const GONE: u16 = 410; +const INSUFFICIENT_STORAGE: u16 = 507; const MAX_HOSTNAME_BYTES: usize = 253; const MAX_LABEL_BYTES: usize = 63; @@ -110,7 +111,7 @@ pub async fn send(state: &AppState, sub: &Subscription, envelope: &Value) -> Sen continue; } }; - if is_transient_status(status) && attempt < MAX_TRANSIENT_RETRIES { + if should_retry(status, attempt) { tokio::time::sleep(retry_delay(attempt)).await; attempt += 1; continue; @@ -127,6 +128,10 @@ fn delivery_headers(envelope: &Value) -> (&'static str, &'static str) { } } +fn should_retry(status: u16, attempt: u32) -> bool { + is_transient_status(status) && status != INSUFFICIENT_STORAGE && attempt < MAX_TRANSIENT_RETRIES +} + fn classify(status: u16) -> SendOutcome { match status { 200..=299 => SendOutcome::Accepted, @@ -134,6 +139,7 @@ fn classify(status: u16) -> SendOutcome { NOT_FOUND => SendOutcome::TokenInvalid { reason: "not_found", }, + INSUFFICIENT_STORAGE => SendOutcome::permanent("http_507"), _ if is_transient_status(status) => SendOutcome::transient(format!("http_{status}")), _ => SendOutcome::permanent(format!("http_{status}")), } @@ -208,3 +214,43 @@ fn retry_delay(attempt: u32) -> Duration { let jitter = rand::rng().random_range(1..=(base / 4).max(1)); Duration::from_millis(MAX_RETRY_DELAY_MS.min(base + jitter - 1)) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_unified_push_topic_with_no_listener_is_permanent() { + assert_eq!( + classify(INSUFFICIENT_STORAGE), + SendOutcome::permanent("http_507") + ); + } + + #[test] + fn an_unavailable_push_service_stays_retryable() { + assert_eq!(classify(503), SendOutcome::transient("http_503")); + } +} + +#[cfg(test)] +mod retry_tests { + use super::*; + + #[test] + fn a_topic_with_no_listener_is_not_retried() { + assert!(!should_retry(INSUFFICIENT_STORAGE, 0)); + } + + #[test] + fn an_unavailable_push_service_is_retried_until_the_budget_runs_out() { + assert!(should_retry(503, 0)); + assert!(!should_retry(503, MAX_TRANSIENT_RETRIES)); + } + + #[test] + fn a_permanent_status_is_never_retried() { + assert!(!should_retry(400, 0)); + assert!(!should_retry(410, 0)); + } +} diff --git a/fluxer_push/src/vendor.rs b/fluxer_push/src/vendor.rs index 4d7b3c12c..a47881225 100644 --- a/fluxer_push/src/vendor.rs +++ b/fluxer_push/src/vendor.rs @@ -221,7 +221,6 @@ fn apns_dead_token(status: u16, reason: &str) -> Option { (_, "Unregistered") => Some(DeadToken::Gone("unregistered")), (410, _) => Some(DeadToken::Gone("gone")), (400, "BadDeviceToken") => Some(DeadToken::Invalid("bad_device_token")), - (400, "DeviceTokenNotForTopic") => Some(DeadToken::Invalid("device_token_not_for_topic")), _ => None, } } @@ -308,6 +307,19 @@ mod tests { assert!(!error.without_url().to_string().contains(TOKEN)); } + #[test] + fn a_wrong_topic_is_not_a_dead_token() { + assert_eq!(apns_dead_token(400, "DeviceTokenNotForTopic"), None); + } + + #[test] + fn a_bad_device_token_is_still_a_dead_token() { + assert_eq!( + apns_dead_token(400, "BadDeviceToken"), + Some(DeadToken::Invalid("bad_device_token")) + ); + } + #[tokio::test] async fn a_refused_connection_stays_retryable() { let error = error_for("http://127.0.0.1:1/").await;