M4: sync pull and push (E-31, E-32), audit writer and diff, FCM sender, push batching, scheduler, pruning, rotate-epoch

- src/sync/: pull with UNION ALL paging and cursor_ahead; push processing of SPEC §14.4.4 steps 1 to 12 in one BEGIN IMMEDIATE with a savepoint per mutation, LWW with delete-wins, tombstones, clock_skew, replay from sync_mutations, sync_batch for more than 3 notifiable changes; 90-day pruning. - src/audit/: SqlAuditWriter and the §14.4.7 diff (now the default in AppState::new). - src/push/: bounded PushHub (100, drop oldest), FCM HTTP v1 client with service-account JWT (RS256, jsonwebtoken on aws-lc-rs) and token cache, sender with 1/5/25 s retries and token clearing; disabled without credentials. - src/tasks/: start installs push and spawns the 06:00 generation (and at start) and 04:00 pruning loops on the injectable clock. - cli: rotate-epoch; serve uses into_make_service_with_connect_info; stale doc comments updated; /i/ empty code gets the HTML 404. - tests: sync_api, push_fcm (local stub of Google's endpoints), scheduler_daily; skeleton routes test now asserts no route is 501. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TQd3pEP9UeN2uifbHknQ7B
2026-10-17 22:27:23 +02:00
parent 853c038c58
commit 15056dc7df
37 changed files with 6312 additions and 102 deletions
@@ -0,0 +1,309 @@
+//! 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}"))
+    }
+}
@@ -1,11 +1,20 @@
-//! The push notification seam (SPEC §14.6). M2a defines the trait and a no-op; M4 owns `src/push/**` and replaces
-//! the implementation with the bounded queue and FCM sender.
+//! Push notifications (SPEC §14.6). M2a defined the trait; M4 owns `src/push/**` and implements it with the bounded
+//! queue [`PushHub`] drained by the FCM [`Sender`] (started by `tasks::start`). [`NoopPushQueue`] remains for push
+//! disabled (no credentials file).
 //!
 //! [`PushQueue::enqueue`] never blocks and never fails the caller: sending happens on a background task and never
 //! delays an HTTP response (SPEC §14.6). Call it after the write transaction has committed.

 use uuid::Uuid;

+pub mod fcm;
+pub mod queue;
+pub mod sender;
+
+pub use fcm::{Endpoints, FcmClient};
+pub use queue::PushHub;
+pub use sender::{RetryPolicy, Sender};
+
 /// The notification types of SPEC §14.6 (exhaustive).
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub enum PushType {
@@ -52,7 +61,7 @@ pub trait PushQueue: Send + Sync + 'static {
    fn enqueue(&self, message: PushMessage);
 }

-/// Drops every message. Used until M4 lands the sender, and when push is disabled.
+/// Drops every message: push disabled (no FCM credentials).
 #[derive(Debug, Clone, Copy, Default)]
 pub struct NoopPushQueue;

@@ -0,0 +1,75 @@
+//! The bounded push queue (SPEC §14.6): capacity 100, the oldest message dropped with a warning when full.
+//! [`PushQueue::enqueue`] never blocks; the sender task waits on [`PushHub::next`].
+
+use std::collections::VecDeque;
+use std::sync::{Mutex, MutexGuard};
+
+use tokio::sync::Notify;
+
+use super::{PushMessage, PushQueue};
+
+/// Queue capacity (SPEC §14.6).
+pub const QUEUE_CAPACITY: usize = 100;
+
+/// The queue between the request handlers (and the scheduler) and the sender task.
+#[derive(Debug, Default)]
+pub struct PushHub {
+    queue: Mutex<VecDeque<PushMessage>>,
+    notify: Notify,
+}
+
+impl PushHub {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    fn lock(&self) -> MutexGuard<'_, VecDeque<PushMessage>> {
+        match self.queue.lock() {
+            Ok(guard) => guard,
+            Err(poisoned) => poisoned.into_inner(),
+        }
+    }
+
+    /// Messages waiting.
+    pub fn len(&self) -> usize {
+        self.lock().len()
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.len() == 0
+    }
+
+    /// The oldest waiting message, if any.
+    pub fn pop(&self) -> Option<PushMessage> {
+        self.lock().pop_front()
+    }
+
+    /// Waits for the next message.
+    pub async fn next(&self) -> PushMessage {
+        loop {
+            if let Some(message) = self.pop() {
+                return message;
+            }
+            self.notify.notified().await;
+        }
+    }
+}
+
+impl PushQueue for PushHub {
+    fn enqueue(&self, message: PushMessage) {
+        {
+            let mut queue = self.lock();
+            if queue.len() >= QUEUE_CAPACITY
+                && let Some(dropped) = queue.pop_front()
+            {
+                tracing::warn!(
+                    push_type = dropped.push_type.as_str(),
+                    capacity = QUEUE_CAPACITY,
+                    "push queue full: oldest message dropped"
+                );
+            }
+            queue.push_back(message);
+        }
+        self.notify.notify_one();
+    }
+}
@@ -0,0 +1,176 @@
+//! The push sender task (SPEC §14.6): takes messages from the [`PushHub`], resolves the target devices, sends one
+//! FCM message per device with retries, and clears dead tokens.
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use sqlx::Row;
+
+use super::PushMessage;
+use super::PushType;
+use super::fcm::{FcmClient, SendOutcome, message_body};
+use super::queue::PushHub;
+use crate::db::{self, Db};
+use crate::error::AppError;
+
+/// Delays before the retries of a send that failed with 429, 5xx or a network error (SPEC §14.6: 1 s, 5 s, 25 s).
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct RetryPolicy {
+    pub delays: Vec<Duration>,
+}
+
+impl Default for RetryPolicy {
+    fn default() -> Self {
+        Self {
+            delays: vec![
+                Duration::from_secs(1),
+                Duration::from_secs(5),
+                Duration::from_secs(25),
+            ],
+        }
+    }
+}
+
+/// One target device.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Recipient {
+    pub device_id: String,
+    pub fcm_token: String,
+}
+
+/// What happened to one message for one device.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Delivery {
+    Sent { attempts: usize },
+    TokenCleared,
+    GaveUp { attempts: usize, detail: String },
+}
+
+/// The background sender.
+pub struct Sender {
+    db: Db,
+    hub: Arc<PushHub>,
+    fcm: FcmClient,
+    retry: RetryPolicy,
+}
+
+impl Sender {
+    pub fn new(db: Db, hub: Arc<PushHub>, fcm: FcmClient, retry: RetryPolicy) -> Self {
+        Self {
+            db,
+            hub,
+            fcm,
+            retry,
+        }
+    }
+
+    /// Runs forever, one message at a time.
+    pub async fn run(self) {
+        loop {
+            let message = self.hub.next().await;
+            if let Err(e) = self.deliver(&message).await {
+                tracing::warn!(
+                    push_type = message.push_type.as_str(),
+                    error_code = e.code.as_str(),
+                    detail = %e.message,
+                    "push not delivered"
+                );
+            }
+        }
+    }
+
+    /// The devices `message` goes to (SPEC §14.6, AC-13.1, AC-13.2): non-revoked devices with an FCM token of the
+    /// other user, or of both users for `recurring_expense_created`.
+    pub async fn recipients(&self, message: &PushMessage) -> Result<Vec<Recipient>, AppError> {
+        let rows = if message.push_type == PushType::RecurringExpenseCreated {
+            sqlx::query(
+                "SELECT id, fcm_token FROM devices WHERE revoked_at IS NULL AND fcm_token IS NOT NULL \
+                 ORDER BY created_at, id",
+            )
+            .fetch_all(&self.db)
+            .await?
+        } else {
+            sqlx::query(
+                "SELECT id, fcm_token FROM devices WHERE revoked_at IS NULL AND fcm_token IS NOT NULL \
+                 AND user_id <> ? ORDER BY created_at, id",
+            )
+            .bind(message.actor_user_id.to_string())
+            .fetch_all(&self.db)
+            .await?
+        };
+        rows.into_iter()
+            .map(|row| {
+                Ok(Recipient {
+                    device_id: row.try_get("id")?,
+                    fcm_token: row.try_get("fcm_token")?,
+                })
+            })
+            .collect()
+    }
+
+    /// Sends `message` to every recipient. Returns one delivery per recipient, in recipient order.
+    pub async fn deliver(&self, message: &PushMessage) -> Result<Vec<Delivery>, AppError> {
+        let mut deliveries = Vec::new();
+        for recipient in self.recipients(message).await? {
+            let delivery = self.send_to(&recipient, message).await?;
+            match &delivery {
+                Delivery::Sent { attempts } => tracing::debug!(
+                    push_type = message.push_type.as_str(),
+                    device_id = %recipient.device_id,
+                    attempts,
+                    "push sent"
+                ),
+                Delivery::TokenCleared => tracing::info!(
+                    device_id = %recipient.device_id,
+                    "FCM token unregistered: cleared"
+                ),
+                Delivery::GaveUp { attempts, detail } => tracing::warn!(
+                    push_type = message.push_type.as_str(),
+                    device_id = %recipient.device_id,
+                    attempts,
+                    detail = %detail,
+                    "push failed"
+                ),
+            }
+            deliveries.push(delivery);
+        }
+        Ok(deliveries)
+    }
+
+    async fn send_to(
+        &self,
+        recipient: &Recipient,
+        message: &PushMessage,
+    ) -> Result<Delivery, AppError> {
+        let body = message_body(&recipient.fcm_token, message);
+        let mut attempts = 0;
+        let mut delays = self.retry.delays.iter();
+        loop {
+            attempts += 1;
+            match self.fcm.send(&body).await {
+                SendOutcome::Sent => return Ok(Delivery::Sent { attempts }),
+                SendOutcome::Unregistered => {
+                    self.clear_token(recipient).await?;
+                    return Ok(Delivery::TokenCleared);
+                }
+                SendOutcome::Failed(detail) => return Ok(Delivery::GaveUp { attempts, detail }),
+                SendOutcome::Retry(detail) => match delays.next() {
+                    Some(delay) => tokio::time::sleep(*delay).await,
+                    None => return Ok(Delivery::GaveUp { attempts, detail }),
+                },
+            }
+        }
+    }
+
+    /// Clears the device's token, unless the app registered a new one meanwhile.
+    async fn clear_token(&self, recipient: &Recipient) -> Result<(), AppError> {
+        let mut tx = db::begin_immediate(&self.db).await?;
+        sqlx::query("UPDATE devices SET fcm_token = NULL WHERE id = ? AND fcm_token = ?")
+            .bind(&recipient.device_id)
+            .bind(&recipient.fcm_token)
+            .execute(&mut *tx)
+            .await?;
+        tx.commit().await?;
+        Ok(())
+    }
+}