kvitt/server/crates/kvitt-server/src/push/fcm.rs

//! The FCM HTTP v1 client (SPEC §14.6, §6.3): data-only messages, OAuth2 access tokens from a service-account JWT
//! (RS256) exchanged at Google's token endpoint and cached until 5 minutes before expiry.
//!
//! Endpoints are code constants; [`Endpoints`] exists so tests can point the client at a local stub server.

use std::path::Path;
use std::sync::Arc;
use std::time::Duration as StdDuration;

use chrono::{DateTime, Duration, Utc};
use jsonwebtoken::{Algorithm, EncodingKey, Header};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use tokio::sync::Mutex;

use super::{PushMessage, PushType};
use crate::error::AppError;
use crate::time::Clock;

/// Google's OAuth2 token endpoint.
pub const TOKEN_URL: &str = "https://oauth2.googleapis.com/token";
/// The FCM API host; the send URL is `<base>/v1/projects/<project>/messages:send`.
pub const SEND_BASE_URL: &str = "https://fcm.googleapis.com";
/// The OAuth2 scope of the HTTP v1 API.
pub const SCOPE: &str = "https://www.googleapis.com/auth/firebase.messaging";
/// The JWT bearer grant type.
pub const GRANT_TYPE: &str = "urn:ietf:params:oauth:grant-type:jwt-bearer";
/// JWT lifetime: Google's maximum, one hour.
pub const JWT_LIFETIME: Duration = Duration::hours(1);
/// A cached access token is replaced this long before it expires.
pub const REFRESH_MARGIN: Duration = Duration::minutes(5);
/// Per-request timeout of the HTTP client.
pub const HTTP_TIMEOUT: StdDuration = StdDuration::from_secs(30);
/// `android.ttl` of every message (SPEC §14.6).
pub const MESSAGE_TTL: &str = "86400s";
/// Payload version (`data.v`).
pub const PAYLOAD_VERSION: &str = "1";

/// Where the client sends. [`Endpoints::google`] in production.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Endpoints {
    pub token_url: String,
    pub send_base_url: String,
}

impl Endpoints {
    pub fn google() -> Self {
        Self {
            token_url: TOKEN_URL.to_owned(),
            send_base_url: SEND_BASE_URL.to_owned(),
        }
    }
}

/// The fields of a service-account key file the client uses.
#[derive(Clone, Deserialize)]
pub struct ServiceAccount {
    pub client_email: String,
    pub private_key: String,
    #[serde(default)]
    pub private_key_id: Option<String>,
}

impl std::fmt::Debug for ServiceAccount {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        // Never print the key.
        f.debug_struct("ServiceAccount")
            .field("client_email", &self.client_email)
            .finish_non_exhaustive()
    }
}

#[derive(Serialize)]
struct Claims<'a> {
    iss: &'a str,
    scope: &'a str,
    aud: &'a str,
    iat: i64,
    exp: i64,
}

#[derive(Deserialize)]
struct TokenResponse {
    access_token: String,
    expires_in: i64,
}

/// The result of one send attempt.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SendOutcome {
    /// 2xx.
    Sent,
    /// `UNREGISTERED` or HTTP 404: the token is dead; clear it.
    Unregistered,
    /// 429, 5xx or a network error: try again.
    Retry(String),
    /// Anything else: give up on this message for this device.
    Failed(String),
}

/// The `entity_type` of a push type (SPEC §14.6 table).
pub fn entity_type_of(push_type: PushType) -> &'static str {
    match push_type {
        PushType::ExpenseCreated
        | PushType::ExpenseUpdated
        | PushType::ExpenseDeleted
        | PushType::RecurringExpenseCreated => "expense",
        PushType::SettlementCreated | PushType::SettlementUpdated | PushType::SettlementDeleted => {
            "settlement"
        }
        PushType::SyncBatch => "",
    }
}

/// The HTTP v1 request body for one device (SPEC §14.6): data only, every value a string, ids only (AC-13.8).
pub fn message_body(fcm_token: &str, message: &PushMessage) -> Value {
    json!({
        "message": {
            "token": fcm_token,
            "android": { "priority": "high", "ttl": MESSAGE_TTL },
            "data": {
                "v": PAYLOAD_VERSION,
                "type": message.push_type.as_str(),
                "entity_type": entity_type_of(message.push_type),
                "entity_id": message.entity_id.map(|id| id.to_string()).unwrap_or_default(),
                "actor_user_id": message.actor_user_id.to_string(),
                "count": message.count.to_string(),
            }
        }
    })
}

/// Whether an FCM error body carries `errorCode` `UNREGISTERED` in a `google.firebase.fcm.v1.FcmError` detail.
pub fn is_unregistered(body: &Value) -> bool {
    body.pointer("/error/details")
        .and_then(Value::as_array)
        .is_some_and(|details| {
            details
                .iter()
                .any(|d| d.get("errorCode").and_then(Value::as_str) == Some("UNREGISTERED"))
        })
}

/// Sends FCM HTTP v1 messages for one project.
pub struct FcmClient {
    http: reqwest::Client,
    account: ServiceAccount,
    key: EncodingKey,
    project_id: String,
    endpoints: Endpoints,
    clock: Arc<dyn Clock>,
    cache: Mutex<Option<(String, DateTime<Utc>)>>,
}

impl FcmClient {
    /// A client from the service-account JSON (`key_json`).
    pub fn new(
        key_json: &str,
        project_id: &str,
        endpoints: Endpoints,
        clock: Arc<dyn Clock>,
    ) -> Result<Self, AppError> {
        let account: ServiceAccount = serde_json::from_str(key_json).map_err(|e| {
            AppError::internal(format!(
                "FCM credentials are not a service-account key: {e}"
            ))
        })?;
        let key = EncodingKey::from_rsa_pem(account.private_key.as_bytes())
            .map_err(|e| AppError::internal(format!("FCM credentials: private_key: {e}")))?;
        let http = reqwest::Client::builder()
            .timeout(HTTP_TIMEOUT)
            .build()
            .map_err(|e| AppError::internal(format!("HTTP client: {e}")))?;
        Ok(Self {
            http,
            account,
            key,
            project_id: project_id.to_owned(),
            endpoints,
            clock,
            cache: Mutex::new(None),
        })
    }

    /// A client from the key file at `path`; `Ok(None)` when the file does not exist (push disabled).
    pub fn from_file(
        path: &Path,
        project_id: &str,
        endpoints: Endpoints,
        clock: Arc<dyn Clock>,
    ) -> Result<Option<Self>, AppError> {
        let text = match std::fs::read_to_string(path) {
            Ok(text) => text,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(e) => {
                return Err(AppError::internal(format!(
                    "cannot read FCM credentials {}: {e}",
                    path.display()
                )));
            }
        };
        Self::new(&text, project_id, endpoints, clock).map(Some)
    }

    /// `<base>/v1/projects/<project>/messages:send`.
    pub fn send_url(&self) -> String {
        format!(
            "{}/v1/projects/{}/messages:send",
            self.endpoints.send_base_url, self.project_id
        )
    }

    fn assertion(&self, now: DateTime<Utc>) -> Result<String, String> {
        let mut header = Header::new(Algorithm::RS256);
        header.kid = self.account.private_key_id.clone();
        let claims = Claims {
            iss: &self.account.client_email,
            scope: SCOPE,
            aud: TOKEN_URL,
            iat: now.timestamp(),
            exp: (now + JWT_LIFETIME).timestamp(),
        };
        jsonwebtoken::encode(&header, &claims, &self.key)
            .map_err(|e| format!("JWT signing failed: {e}"))
    }

    /// A valid access token, from the cache or a new exchange.
    async fn access_token(&self) -> Result<String, SendOutcome> {
        let now = self.clock.now();
        let mut cache = self.cache.lock().await;
        if let Some((token, expires)) = cache.as_ref()
            && now < *expires - REFRESH_MARGIN
        {
            return Ok(token.clone());
        }
        let assertion = self.assertion(now).map_err(SendOutcome::Failed)?;
        let grant: String =
            percent_encoding::utf8_percent_encode(GRANT_TYPE, percent_encoding::NON_ALPHANUMERIC)
                .collect();
        let response = self
            .http
            .post(&self.endpoints.token_url)
            .header("content-type", "application/x-www-form-urlencoded")
            .body(format!("grant_type={grant}&assertion={assertion}"))
            .send()
            .await
            .map_err(|e| SendOutcome::Retry(format!("token request: {e}")))?;
        let status = response.status();
        if !status.is_success() {
            let detail = format!("token endpoint answered {}", status.as_u16());
            return Err(if status.as_u16() == 429 || status.is_server_error() {
                SendOutcome::Retry(detail)
            } else {
                SendOutcome::Failed(detail)
            });
        }
        let body: TokenResponse = response
            .json()
            .await
            .map_err(|e| SendOutcome::Retry(format!("token response: {e}")))?;
        let expires = now + Duration::seconds(body.expires_in);
        *cache = Some((body.access_token.clone(), expires));
        Ok(body.access_token)
    }

    /// Forgets the cached access token (after a 401).
    async fn forget_token(&self) {
        *self.cache.lock().await = None;
    }

    /// One send attempt of `body` (from [`message_body`]).
    pub async fn send(&self, body: &Value) -> SendOutcome {
        let token = match self.access_token().await {
            Ok(token) => token,
            Err(outcome) => return outcome,
        };
        let response = match self
            .http
            .post(self.send_url())
            .bearer_auth(token)
            .json(body)
            .send()
            .await
        {
            Ok(response) => response,
            Err(e) => return SendOutcome::Retry(format!("send: {e}")),
        };
        let status = response.status();
        if status.is_success() {
            return SendOutcome::Sent;
        }
        let code = status.as_u16();
        let error: Value = response.json().await.unwrap_or(Value::Null);
        if code == 404 || is_unregistered(&error) {
            return SendOutcome::Unregistered;
        }
        if code == 429 || status.is_server_error() {
            return SendOutcome::Retry(format!("FCM answered {code}"));
        }
        if code == 401 {
            self.forget_token().await;
        }
        let status_text = error
            .pointer("/error/status")
            .and_then(Value::as_str)
            .unwrap_or("");
        SendOutcome::Failed(format!("FCM answered {code} {status_text}"))
    }
}