kvitt/server/crates/kvitt-server/tests/push_fcm/fcm.rs

//! The FCM sender against the stub (SPEC §14.6).

use std::sync::Arc;
use std::time::{Duration, Instant};

use chrono::Duration as ChronoDuration;
use jsonwebtoken::{Algorithm, DecodingKey, Validation};
use kvitt_server::push::fcm::{SCOPE, TOKEN_URL, message_body};
use kvitt_server::push::queue::QUEUE_CAPACITY;
use kvitt_server::push::sender::Delivery;
use kvitt_server::push::{
    FcmClient, PushHub, PushMessage, PushQueue, PushType, RetryPolicy, Sender,
};
use kvitt_server::time::{format_ts, parse_ts};
use serde_json::{Value, json};
use uuid::Uuid;

use crate::common::*;
use crate::stub::*;
use crate::sync_fixtures;

const EXPENSE: &str = "01929c5e-1111-7aaa-8bbb-000000000001";
const FAST: [u64; 3] = [10, 20, 40];

struct Setup {
    app: TestApp,
    stub: Stub,
    key: TestKey,
    hub: Arc<PushHub>,
}

impl Setup {
    async fn new() -> Self {
        let app = sync_fixtures::app().await;
        for (device, tok) in [
            (ADMIN_DEVICE_ID, "tok-admin"),
            (MEMBER_DEVICE_ID, "tok-member"),
            (ADMIN_OLD_DEVICE_ID, "tok-revoked"),
        ] {
            set_fcm_token(&app, device, Some(tok)).await;
        }
        Self {
            app,
            stub: Stub::start().await,
            key: service_account(),
            hub: Arc::new(PushHub::new()),
        }
    }

    fn client(&self) -> FcmClient {
        FcmClient::new(
            &self.key.json,
            "kvitt-1337",
            self.stub.endpoints(),
            self.app.state.clock.clone(),
        )
        .unwrap()
    }

    fn sender(&self) -> Sender {
        Sender::new(
            self.app.state.db.clone(),
            self.hub.clone(),
            self.client(),
            RetryPolicy {
                delays: FAST.iter().map(|ms| Duration::from_millis(*ms)).collect(),
            },
        )
    }
}

async fn set_fcm_token(app: &TestApp, device: &str, tok: Option<&str>) {
    sqlx::query("UPDATE devices SET fcm_token = ? WHERE id = ?")
        .bind(tok)
        .bind(device)
        .execute(&app.state.db)
        .await
        .unwrap();
}

async fn fcm_token(app: &TestApp, device: &str) -> Option<String> {
    sqlx::query_scalar("SELECT fcm_token FROM devices WHERE id = ?")
        .bind(device)
        .fetch_one(&app.state.db)
        .await
        .unwrap()
}

fn message(push_type: PushType, actor: &str) -> PushMessage {
    PushMessage {
        push_type,
        entity_id: (push_type != PushType::SyncBatch).then(|| EXPENSE.parse().unwrap()),
        actor_user_id: actor.parse().unwrap(),
        count: 1,
    }
}

#[test]
fn payload_of_every_type_equals_the_contract_example() {
    let all = [
        PushType::ExpenseCreated,
        PushType::ExpenseUpdated,
        PushType::ExpenseDeleted,
        PushType::SettlementCreated,
        PushType::SettlementUpdated,
        PushType::SettlementDeleted,
        PushType::RecurringExpenseCreated,
        PushType::SyncBatch,
    ];
    for push_type in all {
        let expected = example(&format!("push/{}.json", push_type.as_str()));
        let data = &expected["message"]["data"];
        let entity_id = data["entity_id"].as_str().unwrap();
        let msg = PushMessage {
            push_type,
            entity_id: (!entity_id.is_empty()).then(|| entity_id.parse::<Uuid>().unwrap()),
            actor_user_id: data["actor_user_id"].as_str().unwrap().parse().unwrap(),
            count: data["count"].as_str().unwrap().parse().unwrap(),
        };
        let body = message_body(expected["message"]["fid"].as_str().unwrap(), &msg);
        assert_eq!(body, expected, "{}", push_type.as_str());
        // Data only, every value a string (SPEC §14.6), ids only (AC-13.8).
        assert!(body["message"].get("notification").is_none());
        assert!(
            body["message"]["data"]
                .as_object()
                .unwrap()
                .values()
                .all(Value::is_string)
        );
    }
}

#[tokio::test]
async fn oauth_jwt_and_send_request_follow_the_http_v1_flow() {
    let s = Setup::new().await;
    let deliveries = s
        .sender()
        .deliver(&message(PushType::ExpenseCreated, ADMIN_ID))
        .await
        .unwrap();
    assert_eq!(deliveries, [Delivery::Sent { attempts: 1 }]);

    let tokens = s.stub.tokens();
    assert_eq!(tokens.len(), 1);
    let form = String::from_utf8(tokens[0].body.to_vec()).unwrap();
    let (grant, assertion) = form.split_once('&').unwrap();
    assert_eq!(
        grant,
        "grant_type=urn%3Aietf%3Aparams%3Aoauth%3Agrant%2Dtype%3Ajwt%2Dbearer"
    );
    let jwt = assertion.strip_prefix("assertion=").unwrap();
    let header = jsonwebtoken::decode_header(jwt).unwrap();
    assert_eq!(header.alg, Algorithm::RS256);
    assert_eq!(header.kid.as_deref(), Some("test-key-1"));
    let mut validation = Validation::new(Algorithm::RS256);
    validation.set_audience(&[TOKEN_URL]);
    validation.set_issuer(&["kvitt-fcm@kvitt-1337.iam.gserviceaccount.com"]);
    // The scripted clock is in the future relative to the machine's: check exp by hand.
    validation.validate_exp = false;
    let claims: Value = jsonwebtoken::decode(
        jwt,
        &DecodingKey::from_rsa_der(&s.key.public_der),
        &validation,
    )
    .unwrap()
    .claims;
    assert_eq!(claims["scope"], SCOPE);
    let now = parse_ts(NOW).unwrap().timestamp();
    assert_eq!(claims["iat"], now);
    assert_eq!(claims["exp"], now + 3600);

    // One message to the other user's one non-revoked device.
    let sends = s.stub.sends();
    assert_eq!(sends.len(), 1);
    assert_eq!(sends[0].path, "/v1/projects/kvitt-1337/messages:send");
    assert_eq!(
        sends[0].authorization.as_deref(),
        Some("Bearer ya29.stub-1")
    );
    let expected = json!({"message": {"fid": "tok-member", "android": {"priority": "high", "ttl": "86400s"},
        "data": {"v": "1", "type": "expense_created", "entity_type": "expense", "entity_id": EXPENSE,
                 "actor_user_id": ADMIN_ID, "count": "1"}}});
    assert_eq!(sends[0].json(), expected);
}

#[tokio::test]
async fn access_token_is_cached_until_5_minutes_before_expiry() {
    let s = Setup::new().await;
    let sender = s.sender();
    let msg = message(PushType::ExpenseUpdated, ADMIN_ID);
    sender.deliver(&msg).await.unwrap();
    s.app.set_now(&format_ts(
        parse_ts(NOW).unwrap() + ChronoDuration::minutes(54),
    ));
    sender.deliver(&msg).await.unwrap();
    assert_eq!(s.stub.tokens().len(), 1);
    s.app.set_now(&format_ts(
        parse_ts(NOW).unwrap() + ChronoDuration::minutes(55),
    ));
    sender.deliver(&msg).await.unwrap();
    assert_eq!(s.stub.tokens().len(), 2);
    assert_eq!(
        s.stub.sends()[2].authorization.as_deref(),
        Some("Bearer ya29.stub-2")
    );
}

#[tokio::test]
async fn recipients_are_the_other_users_live_devices_with_a_token_recurring_goes_to_both() {
    let s = Setup::new().await;
    let sender = s.sender();
    let ids = |r: Vec<kvitt_server::push::sender::Recipient>| {
        r.into_iter().map(|r| r.device_id).collect::<Vec<_>>()
    };
    // Admin acts: the member's device. Member acts: the admin's current device (the revoked one is skipped).
    assert_eq!(
        ids(sender
            .recipients(&message(PushType::SettlementCreated, ADMIN_ID))
            .await
            .unwrap()),
        [MEMBER_DEVICE_ID]
    );
    assert_eq!(
        ids(sender
            .recipients(&message(PushType::ExpenseDeleted, MEMBER_ID))
            .await
            .unwrap()),
        [ADMIN_DEVICE_ID]
    );
    assert_eq!(
        ids(sender
            .recipients(&message(PushType::SyncBatch, MEMBER_ID))
            .await
            .unwrap()),
        [ADMIN_DEVICE_ID]
    );
    // AC-13.2: recurring to every live device of both users, the actor's included.
    assert_eq!(
        ids(sender
            .recipients(&message(PushType::RecurringExpenseCreated, ADMIN_ID))
            .await
            .unwrap()),
        [ADMIN_DEVICE_ID, MEMBER_DEVICE_ID]
    );
    // A device without a token is skipped.
    set_fcm_token(&s.app, MEMBER_DEVICE_ID, None).await;
    assert!(
        sender
            .recipients(&message(PushType::ExpenseCreated, ADMIN_ID))
            .await
            .unwrap()
            .is_empty()
    );
}

#[test]
fn production_retry_delays_are_1_5_25_seconds() {
    assert_eq!(
        RetryPolicy::default().delays,
        [
            Duration::from_secs(1),
            Duration::from_secs(5),
            Duration::from_secs(25)
        ]
    );
}

#[tokio::test]
async fn retries_429_and_5xx_with_the_policy_delays_then_succeeds_or_gives_up() {
    let s = Setup::new().await;
    let sender = s.sender();
    let msg = message(PushType::ExpenseCreated, ADMIN_ID);
    s.stub.script_sends(&[
        (503, fcm_error(503, "UNAVAILABLE", "UNAVAILABLE")),
        (500, fcm_error(500, "INTERNAL", "INTERNAL")),
        (429, fcm_error(429, "RESOURCE_EXHAUSTED", "QUOTA_EXCEEDED")),
    ]);
    let started = Instant::now();
    assert_eq!(
        sender.deliver(&msg).await.unwrap(),
        [Delivery::Sent { attempts: 4 }]
    );
    assert!(started.elapsed() >= Duration::from_millis(FAST.iter().sum()));
    assert_eq!(s.stub.sends().len(), 4);

    s.stub.script_sends(&[
        (503, json!({})),
        (503, json!({})),
        (503, json!({})),
        (503, json!({})),
    ]);
    let d = sender.deliver(&msg).await.unwrap();
    assert!(
        matches!(&d[0], Delivery::GaveUp { attempts: 4, .. }),
        "{d:?}"
    );
    assert_eq!(s.stub.sends().len(), 8);

    // Other 4xx: no retry, token kept.
    s.stub
        .script_sends(&[(400, fcm_error(400, "INVALID_ARGUMENT", "INVALID_ARGUMENT"))]);
    let d = sender.deliver(&msg).await.unwrap();
    assert!(
        matches!(&d[0], Delivery::GaveUp { attempts: 1, .. }),
        "{d:?}"
    );
    assert_eq!(
        fcm_token(&s.app, MEMBER_DEVICE_ID).await.as_deref(),
        Some("tok-member")
    );
}

#[tokio::test]
async fn network_errors_are_retried() {
    let s = Setup::new().await;
    // A port nobody listens on.
    let dead = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
    let addr = dead.local_addr().unwrap();
    drop(dead);
    let endpoints = kvitt_server::push::Endpoints {
        token_url: format!("http://{addr}/token"),
        send_base_url: format!("http://{addr}"),
    };
    let client = FcmClient::new(
        &s.key.json,
        "kvitt-1337",
        endpoints,
        s.app.state.clock.clone(),
    )
    .unwrap();
    let sender = Sender::new(
        s.app.state.db.clone(),
        s.hub.clone(),
        client,
        RetryPolicy {
            delays: FAST.iter().map(|ms| Duration::from_millis(*ms)).collect(),
        },
    );
    let d = sender
        .deliver(&message(PushType::ExpenseCreated, ADMIN_ID))
        .await
        .unwrap();
    assert!(
        matches!(&d[0], Delivery::GaveUp { attempts: 4, .. }),
        "{d:?}"
    );
}

#[tokio::test]
async fn unregistered_or_404_clears_the_devices_token() {
    let s = Setup::new().await;
    let sender = s.sender();
    s.stub
        .script_sends(&[(404, fcm_error(404, "NOT_FOUND", "UNREGISTERED"))]);
    assert_eq!(
        sender
            .deliver(&message(PushType::ExpenseCreated, ADMIN_ID))
            .await
            .unwrap(),
        [Delivery::TokenCleared]
    );
    assert_eq!(fcm_token(&s.app, MEMBER_DEVICE_ID).await, None);
    assert_eq!(s.stub.sends().len(), 1, "no retry");
    // UNREGISTERED under another status, and a bare 404.
    s.stub.script_sends(&[
        (400, fcm_error(400, "INVALID_ARGUMENT", "UNREGISTERED")),
        (404, json!({})),
    ]);
    set_fcm_token(&s.app, MEMBER_DEVICE_ID, Some("tok-member")).await;
    sender
        .deliver(&message(PushType::ExpenseCreated, ADMIN_ID))
        .await
        .unwrap();
    assert_eq!(fcm_token(&s.app, MEMBER_DEVICE_ID).await, None);
    sender
        .deliver(&message(PushType::ExpenseCreated, MEMBER_ID))
        .await
        .unwrap();
    assert_eq!(fcm_token(&s.app, ADMIN_DEVICE_ID).await, None);
}

#[tokio::test]
async fn full_queue_drops_the_oldest() {
    let hub = PushHub::new();
    for i in 0..=QUEUE_CAPACITY {
        hub.enqueue(PushMessage {
            push_type: PushType::SyncBatch,
            entity_id: None,
            actor_user_id: ADMIN_ID.parse().unwrap(),
            count: u32::try_from(i).unwrap(),
        });
    }
    assert_eq!(QUEUE_CAPACITY, 100);
    assert_eq!(hub.len(), 100);
    assert_eq!(hub.pop().unwrap().count, 1, "message 0 was dropped");
}

#[tokio::test]
async fn missing_or_unusable_credentials_disable_push() {
    let s = Setup::new().await;
    let missing = s.app.data_dir().join("no-such-key.json");
    let client = FcmClient::from_file(
        &missing,
        "kvitt-1337",
        s.stub.endpoints(),
        s.app.state.clock.clone(),
    );
    assert!(client.unwrap().is_none());
    // tasks::start_push with the configured path (missing in the test config): the no-op queue; nothing is sent.
    let mut state = s.app.state.clone();
    let mut config = (*state.config).clone();
    config.fcm_credentials = missing;
    state.config = Arc::new(config.clone());
    let queue = kvitt_server::tasks::start_push(&state, s.stub.endpoints(), RetryPolicy::default());
    queue.enqueue(message(PushType::ExpenseCreated, ADMIN_ID));
    // An unparsable file: disabled too.
    let bad = s.app.data_dir().join("bad-key.json");
    std::fs::write(
        &bad,
        "{\"client_email\": \"x\", \"private_key\": \"not a key\"}",
    )
    .unwrap();
    config.fcm_credentials = bad;
    state.config = Arc::new(config);
    let queue = kvitt_server::tasks::start_push(&state, s.stub.endpoints(), RetryPolicy::default());
    queue.enqueue(message(PushType::ExpenseCreated, ADMIN_ID));
    tokio::time::sleep(Duration::from_millis(100)).await;
    assert!(s.stub.sends().is_empty());
    assert!(s.stub.tokens().is_empty());
}

#[tokio::test]
async fn sync_push_reaches_fcm_through_the_queue_and_the_sender_task() {
    let mut s = Setup::new().await;
    let key_file = s.app.data_dir().join("fcm.json");
    std::fs::write(&key_file, &s.key.json).unwrap();
    let mut config = (*s.app.state.config).clone();
    config.fcm_credentials = key_file;
    s.app.state.config = Arc::new(config);
    let queue = kvitt_server::tasks::start_push(
        &s.app.state,
        s.stub.endpoints(),
        RetryPolicy {
            delays: vec![Duration::from_millis(10)],
        },
    );
    sync_fixtures::with_push(&mut s.app, queue);
    let id = sync_fixtures::new_id();
    sync_fixtures::push_one(
        &s.app,
        &member_token(),
        sync_fixtures::exp(sync_fixtures::expense(&id, "2026-10-17T11:00:00.000Z")),
    )
    .await;
    let deadline = Instant::now() + Duration::from_secs(5);
    while s.stub.sends().is_empty() && Instant::now() < deadline {
        tokio::time::sleep(Duration::from_millis(10)).await;
    }
    let sends = s.stub.sends();
    assert_eq!(sends.len(), 1);
    let body = sends[0].json();
    assert_eq!(body["message"]["fid"], "tok-admin");
    assert!(body["message"].get("token").is_none());
    assert_eq!(body["message"]["data"]["entity_id"], id);
    assert_eq!(body["message"]["data"]["actor_user_id"], MEMBER_ID);
}

#[derive(Clone, Default)]
struct Captured(Arc<std::sync::Mutex<Vec<u8>>>);

impl std::io::Write for Captured {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        self.0.lock().unwrap().extend_from_slice(buf);
        Ok(buf.len())
    }
    fn flush(&mut self) -> std::io::Result<()> {
        Ok(())
    }
}

/// The only test of this binary that installs a subscriber. Global, not thread-local, because tracing caches callsite
/// interest process-wide; the other tests' lines are captured too, so only lines of this test's push type are read.
#[tokio::test]
async fn a_successful_send_logs_one_info_line_with_aggregates_only() {
    let captured = Captured::default();
    let writer = captured.clone();
    let subscriber = tracing_subscriber::fmt()
        .json()
        .with_max_level(tracing::Level::INFO)
        .with_writer(move || writer.clone())
        .finish();
    tracing::subscriber::set_global_default(subscriber).unwrap();

    let s = Setup::new().await;
    let deliveries = s
        .sender()
        .deliver(&message(PushType::SettlementDeleted, MEMBER_ID))
        .await
        .unwrap();
    assert_eq!(deliveries, [Delivery::Sent { attempts: 1 }]);

    let logs = String::from_utf8(captured.0.lock().unwrap().clone()).unwrap();
    let lines: Vec<Value> = logs
        .lines()
        .map(|l| serde_json::from_str::<Value>(l).unwrap())
        .filter(|l| {
            l["fields"]["message"] == "push sent"
                && l["fields"]["push_type"] == "settlement_deleted"
        })
        .collect();
    assert_eq!(lines.len(), 1, "{logs}");
    let line = &lines[0];
    assert_eq!(line["level"], "INFO");
    assert_eq!(line["fields"]["devices_sent"], 1);
    assert_eq!(line["fields"]["devices_targeted"], 1);
    // The member acted, so the admin's one live device got it: its id's last 8 characters only.
    assert_eq!(line["fields"]["devices"], "00000001");
    let text = line.to_string();
    for secret in ["tok-admin", ADMIN_DEVICE_ID, MEMBER_ID, EXPENSE] {
        assert!(!text.contains(secret), "{text}");
    }
}