Merge remote-tracking branch 'origin/main'

fluxer_push/src/delivery.rs

@@ -279,14 +279,11 @@ async fn run_subject<T: Transport>(
             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;
             });

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<u8> {
     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));
+    }
+}

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 {
+            apns::send(state, &hopped, envelope).await
-                own_relay::Leg::Fcm => fcm::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::Hop> {
+    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<String> {
+        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());
+    }
+}

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}");

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));
+    }
+}

fluxer_push/src/vendor.rs

@@ -221,7 +221,6 @@ fn apns_dead_token(status: u16, reason: &str) -> Option<DeadToken> {
         (_, "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;