OXYGENHUB
# a22e4aec2e28 oxygenhub / crates / hub / src / push.rs
4.2 KiB · Rust · last changed Phone alerts: a phone registers its Cloud Messaging token and key (PUT /api/v1/user/push), and each notification its account is told goes to it sealed to that key, sent keyless as the VM's service account; pins OxygenFlow main (refs oxygenhub-android#16) 8 days ago source blame history raw
1
//! Phone alerts (oxygenhub-android#16, oxygenhub-gcp decision 0008): each
2
//! notification an account is told also goes to its phones through Firebase
3
//! Cloud Messaging, sealed to each phone's key (`flow_core::alert`), so the
4
//! push service carries only ciphertext and a wake-up.
5
//!
6
//! The hub sends as the service account its VM runs as: the metadata
7
//! server's token, kept until a minute before it expires. Off while
8
//! `[push] fcm_project` is empty.
9
10
use crate::db::push as pdb;
11
use crate::Shared;
12
use std::sync::Mutex;
13
use std::time::{Duration, Instant};
14
15
/// Most phones one account's alert goes to.
16
const MOST_DEVICES: usize = 20;
17
18
static TOKEN: Mutex<Option<(String, Instant)>> = Mutex::new(None);
19
20
/// What a phone is told, before it is sealed: small, as a message's data
21
/// is at most 4 KB.
22
#[derive(serde::Serialize, serde::Deserialize, Debug, PartialEq)]
23
pub struct Alert {
24
    pub kind: String,
25
    pub title: String,
26
    pub url: String,
27
}
28
29
/// Whether alerts are on.
30
pub fn enabled(app: &Shared) -> bool {
31
    !app.config.push.fcm_project.is_empty()
32
}
33
34
/// Send `alert` to every phone of `account_id`, in the background.
35
pub fn send(app: &Shared, account_id: i64, alert: Alert) {
36
    if !enabled(app) {
37
        return;
38
    }
39
    let app = app.clone();
40
    tokio::spawn(async move {
41
        if let Err(e) = send_now(&app, account_id, &alert).await {
42
            tracing::warn!("push: alert for account {account_id}: {e}");
43
        }
44
    });
45
}
46
47
async fn access_token(app: &Shared, client: &reqwest::Client) -> Result<String, String> {
48
    if let Some((t, until)) = TOKEN.lock().unwrap().clone() {
49
        if Instant::now() < until {
50
            return Ok(t);
51
        }
52
    }
53
    #[derive(serde::Deserialize)]
54
    struct Got {
55
        access_token: String,
56
        expires_in: u64,
57
    }
58
    let got: Got = client
59
        .get(&app.config.push.token_url)
60
        .header("Metadata-Flavor", "Google")
61
        .send()
62
        .await
63
        .and_then(|r| r.error_for_status())
64
        .map_err(|e| format!("the metadata server's token: {e}"))?
65
        .json()
66
        .await
67
        .map_err(|e| format!("the metadata server's token: {e}"))?;
68
    let until = Instant::now() + Duration::from_secs(got.expires_in.saturating_sub(60));
69
    *TOKEN.lock().unwrap() = Some((got.access_token.clone(), until));
70
    Ok(got.access_token)
71
}
72
73
/// Send `alert` to each of the account's phones now; a phone Cloud
74
/// Messaging no longer knows is forgotten.
75
pub async fn send_now(app: &Shared, account_id: i64, alert: &Alert) -> Result<usize, String> {
76
    let devices = pdb::of_account(&app.db, account_id).await.map_err(|e| e.to_string())?;
77
    if devices.is_empty() {
78
        return Ok(0);
79
    }
80
    let client = reqwest::Client::builder().timeout(Duration::from_secs(20)).build().map_err(|e| e.to_string())?;
81
    let token = access_token(app, &client).await?;
82
    let plain = serde_json::to_vec(alert).map_err(|e| e.to_string())?;
83
    let url = format!("{}/v1/projects/{}/messages:send", app.config.push.fcm_url.trim_end_matches('/'), app.config.push.fcm_project);
84
    let mut sent = 0;
85
    for d in devices.iter().take(MOST_DEVICES) {
86
        let Some(key) = hex::decode(&d.public_key).ok().and_then(|k| <[u8; 32]>::try_from(k).ok()) else { continue };
87
        let Some(sealed) = flow_core::alert::seal(&key, &plain) else { continue };
88
        // Data only, high priority: the app wakes, opens it and shows it;
89
        // the push service has no title or body to show or read.
90
        let body = serde_json::json!({ "message": {
91
            "token": d.fcm_token,
92
            "data": { "sealed": crate::oidc::b64url(&sealed) },
93
            "android": { "priority": "high" },
94
        }});
95
        let r = client.post(&url).bearer_auth(&token).json(&body).send().await.map_err(|e| e.to_string())?;
96
        let status = r.status();
97
        if status.is_success() {
98
            sent += 1;
99
        } else {
100
            let text = r.text().await.unwrap_or_default();
101
            if status == reqwest::StatusCode::NOT_FOUND || text.contains("UNREGISTERED") {
102
                pdb::forget_fcm(&app.db, &d.fcm_token).await.ok();
103
            } else {
104
                tracing::warn!("push: Cloud Messaging said {status}: {}", text.chars().take(200).collect::<String>());
105
            }
106
        }
107
    }
108
    Ok(sent)
109
}