baukit_push/

expo.rs

//! Expo push adapter implementing the ticket then receipt protocol.

use std::collections::{BTreeMap, HashMap};

use axum::http::{HeaderMap, StatusCode};
use baukit_http::{classify_http_status, classify_transport_error};
use serde::{Deserialize, Serialize};

use crate::{
    MAX_RECEIPT_BATCH_SIZE, PushDeliveryStatus, PushError, PushFuture, PushMessage, PushOptions,
    PushOptionsError, PushOutcome, PushReceipt, PushReceiptFuture, PushReceiptSource,
    PushRejection, PushSender, PushTicketId,
};

const OK_STATUS: &str = "ok";

/// Expo push adapter.
///
/// Expo answers `/push/send` with one *ticket* per notification. A ticket only
/// says Expo accepted the notification; delivery is confirmed later through
/// `/push/getReceipts`. This adapter runs both phases per batch, so one
/// [`PushSender::send`] call returns settled outcomes wherever Expo has them and
/// [`PushDeliveryStatus::Accepted`] for the receipts that are not ready yet.
/// As a [`PushReceiptSource`] it reads those receipts later, up to
/// [`MAX_RECEIPT_BATCH_SIZE`] IDs per request.
///
/// Cloning is cheap; the inner `reqwest` client shares its connection pool.
#[derive(Clone, Debug)]
pub struct ExpoPushSender {
    client: reqwest::Client,
    options: PushOptions,
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct ExpoMessage<'a> {
    to: &'a str,
    title: &'a str,
    body: &'a str,
    sound: &'static str,
    data: &'a BTreeMap<String, String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    channel_id: Option<&'a str>,
}

#[derive(Deserialize)]
struct ExpoTicketResponse {
    data: Option<Vec<ExpoTicket>>,
}

#[derive(Deserialize)]
struct ExpoTicket {
    status: String,
    id: Option<String>,
    details: Option<ExpoErrorDetails>,
}

#[derive(Deserialize)]
struct ExpoReceiptResponse {
    data: Option<HashMap<String, ExpoReceipt>>,
}

#[derive(Deserialize)]
struct ExpoReceipt {
    status: String,
    details: Option<ExpoErrorDetails>,
}

#[derive(Deserialize)]
struct ExpoErrorDetails {
    error: Option<String>,
}

impl ExpoPushSender {
    /// Builds a sender against the public Expo API with default options.
    pub fn new() -> Result<Self, PushOptionsError> {
        Self::with_options(PushOptions::default())
    }

    /// Builds a sender from validated options.
    pub fn with_options(options: PushOptions) -> Result<Self, PushOptionsError> {
        let client = reqwest::Client::builder()
            .timeout(options.request_timeout())
            .build()
            .map_err(|error| PushOptionsError::InvalidEndpoint(error.to_string()))?;
        Ok(Self { client, options })
    }

    /// Returns the options this sender was built from.
    #[must_use]
    pub const fn options(&self) -> &PushOptions {
        &self.options
    }

    async fn send_chunk(&self, messages: &[PushMessage]) -> Result<Vec<PushOutcome>, PushError> {
        let payload = messages.iter().map(expo_message).collect::<Vec<_>>();
        let response: ExpoTicketResponse = self.post(self.options.endpoint(), &payload).await?;
        let tickets = response
            .data
            .ok_or_else(|| PushError::InvalidResponse("ticket response has no data".to_owned()))?;
        if tickets.len() != messages.len() {
            return Err(PushError::InvalidResponse(format!(
                "expected {} tickets, got {}",
                messages.len(),
                tickets.len()
            )));
        }

        let mut outcomes = Vec::with_capacity(messages.len());
        let mut pending = Vec::new();
        for (message, ticket) in messages.iter().zip(tickets) {
            match rejection(ticket.details.as_ref()) {
                Some(rejection) => outcomes.push(PushOutcome {
                    token: message.token.clone(),
                    status: PushDeliveryStatus::Rejected(rejection),
                }),
                None if ticket.status == OK_STATUS => {
                    pending.push((ticket_id(ticket.id)?, message.token.clone()));
                }
                None => outcomes.push(PushOutcome {
                    token: message.token.clone(),
                    status: PushDeliveryStatus::Rejected(PushRejection::ProviderError),
                }),
            }
        }
        if pending.is_empty() {
            return Ok(outcomes);
        }

        let ids = pending.iter().map(|(id, _)| id.clone()).collect::<Vec<_>>();
        let mut receipts = self.fetch_receipts(&ids).await?;
        outcomes.extend(pending.into_iter().map(|(id, token)| {
            let status = match receipts.remove(&id) {
                Some(receipt) => receipt.into(),
                // Expo has not settled this notification yet. Callers must
                // not resend on it; a deferred receipt poll reports the final
                // state.
                None => PushDeliveryStatus::Accepted(id),
            };
            PushOutcome { token, status }
        }));
        Ok(outcomes)
    }

    async fn fetch_receipts(
        &self,
        ids: &[PushTicketId],
    ) -> Result<HashMap<PushTicketId, PushReceipt>, PushError> {
        let mut settled = HashMap::new();
        for chunk in ids.chunks(MAX_RECEIPT_BATCH_SIZE) {
            let request = chunk.iter().map(PushTicketId::as_str).collect::<Vec<_>>();
            let response: ExpoReceiptResponse = self
                .post(
                    self.options.receipts_endpoint(),
                    &serde_json::json!({ "ids": request }),
                )
                .await?;
            let mut receipts = response.data.unwrap_or_default();
            settled.extend(chunk.iter().filter_map(|id| {
                receipts
                    .remove(id.as_str())
                    .map(|receipt| (id.clone(), push_receipt(&receipt)))
            }));
        }
        Ok(settled)
    }

    async fn post<T: serde::de::DeserializeOwned>(
        &self,
        endpoint: &str,
        payload: &impl Serialize,
    ) -> Result<T, PushError> {
        let mut request = self.client.post(endpoint).json(payload);
        if let Some(token) = self.options.access_token() {
            request = request.bearer_auth(token);
        }
        let response = request.send().await.map_err(|error| {
            tracing::debug!(target: "baukit_push", %error, endpoint, "expo request failed");
            PushError::Transport {
                class: classify_transport_error(error.is_timeout()),
            }
        })?;
        let status = response.status();
        if !status.is_success() {
            return Err(transport_error(status, response.headers()));
        }
        response.json::<T>().await.map_err(|error| {
            PushError::InvalidResponse(format!("{endpoint} response did not parse: {error}"))
        })
    }
}

fn transport_error(status: StatusCode, headers: &HeaderMap) -> PushError {
    // Expo rejects a whole request for a bad token or a rate limit, so a 4xx
    // that is not one of those still classifies as permanent and the caller
    // learns not to retry it.
    let class = classify_http_status(status, headers, &[]);
    PushError::Transport { class }
}

fn ticket_id(id: Option<String>) -> Result<PushTicketId, PushError> {
    id.ok_or_else(|| PushError::InvalidResponse("accepted ticket has no id".to_owned()))
        .and_then(|id| {
            PushTicketId::new(id).map_err(|error| PushError::InvalidResponse(error.to_string()))
        })
}

fn push_receipt(receipt: &ExpoReceipt) -> PushReceipt {
    match rejection(receipt.details.as_ref()) {
        Some(rejection) => PushReceipt::Rejected(rejection),
        None if receipt.status == OK_STATUS => PushReceipt::Delivered,
        None => PushReceipt::Rejected(PushRejection::ProviderError),
    }
}

fn expo_message(message: &PushMessage) -> ExpoMessage<'_> {
    ExpoMessage {
        to: message.token.expose(),
        title: &message.title,
        body: &message.body,
        sound: "default",
        data: &message.data,
        channel_id: message.channel_id.as_deref(),
    }
}

fn rejection(details: Option<&ExpoErrorDetails>) -> Option<PushRejection> {
    details
        .and_then(|details| details.error.as_deref())
        .map(PushRejection::from_code)
}

impl PushSender for ExpoPushSender {
    fn send<'a>(&'a self, batch: Vec<PushMessage>) -> PushFuture<'a> {
        Box::pin(async move {
            let mut outcomes = Vec::with_capacity(batch.len());
            for chunk in batch.chunks(self.options.batch_size()) {
                outcomes.extend(self.send_chunk(chunk).await?);
            }
            Ok(outcomes)
        })
    }
}

impl PushReceiptSource for ExpoPushSender {
    fn receipts<'a>(&'a self, tickets: Vec<PushTicketId>) -> PushReceiptFuture<'a> {
        Box::pin(async move { self.fetch_receipts(&tickets).await })
    }
}

#[cfg(test)]
mod tests {
    use std::time::Duration;

    use baukit_http::RetryClass;

    use super::*;

    fn message(token: &str) -> PushMessage {
        let token = crate::DeviceToken::new(token).expect("valid test token");
        PushMessage::new(token, "Reminder", "Your session starts soon")
            .with_data("session_id", "abc")
    }

    #[test]
    fn a_message_serializes_into_expos_camel_case_shape() {
        let payload = serde_json::to_value(expo_message(
            &message("ExponentPushToken[x]").with_channel_id("reminders"),
        ))
        .expect("message should serialize");
        assert_eq!(payload["to"], "ExponentPushToken[x]");
        assert_eq!(payload["title"], "Reminder");
        assert_eq!(payload["body"], "Your session starts soon");
        assert_eq!(payload["sound"], "default");
        assert_eq!(payload["channelId"], "reminders");
        assert_eq!(payload["data"]["session_id"], "abc");
    }

    #[test]
    fn a_message_without_a_channel_omits_the_field_entirely() {
        let payload =
            serde_json::to_value(expo_message(&message("token"))).expect("message serializes");
        assert!(payload.get("channelId").is_none());
    }

    #[test]
    fn ticket_error_details_map_onto_neutral_rejections() {
        let parsed: ExpoTicket = serde_json::from_value(serde_json::json!({
            "status": "error",
            "details": {"error": "DeviceNotRegistered"}
        }))
        .expect("ticket should parse");
        assert_eq!(parsed.status, "error");
        assert_eq!(
            rejection(parsed.details.as_ref()),
            Some(PushRejection::DeviceNotRegistered)
        );
    }

    #[test]
    fn an_error_without_details_has_no_specific_rejection() {
        let parsed: ExpoTicket =
            serde_json::from_value(serde_json::json!({"status": "error"})).expect("ticket parses");
        assert_eq!(rejection(parsed.details.as_ref()), None);
    }

    #[test]
    fn receipts_parse_keyed_by_ticket_id() {
        let parsed: ExpoReceiptResponse = serde_json::from_value(serde_json::json!({
            "data": {
                "ticket-1": {"status": "ok"},
                "ticket-2": {"status": "error", "details": {"error": "MessageTooBig"}}
            }
        }))
        .expect("receipts should parse");
        let data = parsed.data.expect("receipts have data");
        assert_eq!(data["ticket-1"].status, OK_STATUS);
        assert_eq!(
            rejection(data["ticket-2"].details.as_ref()),
            Some(PushRejection::MessageTooBig)
        );
    }

    #[test]
    fn upstream_statuses_classify_into_transport_errors() {
        let mut headers = HeaderMap::new();
        headers.insert(
            axum::http::header::RETRY_AFTER,
            axum::http::HeaderValue::from_static("12"),
        );
        assert_eq!(
            transport_error(StatusCode::TOO_MANY_REQUESTS, &headers).retry_after(),
            Some(Duration::from_secs(12))
        );
        assert!(matches!(
            transport_error(StatusCode::UNAUTHORIZED, &HeaderMap::new()),
            PushError::Transport {
                class: RetryClass::Revoked
            }
        ));
        assert!(matches!(
            transport_error(StatusCode::BAD_GATEWAY, &HeaderMap::new()),
            PushError::Transport {
                class: RetryClass::Unavailable
            }
        ));
        assert!(!transport_error(StatusCode::BAD_REQUEST, &HeaderMap::new()).is_retryable());
    }
}