push.rs 86.3%

63 of 73 lines run, 9 of 30 functions · run #540 on main · run missed

//! Phone alerts (oxygenhub-android#16, oxygenhub-gcp decision 0008): each
//! notification an account is told also goes to its phones through Firebase
//! Cloud Messaging, sealed to each phone's key (`flow_core::alert`), so the
//! push service carries only ciphertext and a wake-up.
//!
//! The hub sends as the service account its VM runs as: the metadata
//! server's token, kept until a minute before it expires. Off while
//! `[push] fcm_project` is empty.

use crate::db::push as pdb;
use crate::Shared;
use std::sync::Mutex;
use std::time::{Duration, Instant};

/// Most phones one account's alert goes to.
const MOST_DEVICES: usize = 20;

static TOKEN: Mutex<Option<(String, Instant)>> = Mutex::new(None);

/// What a phone is told, before it is sealed: small, as a message's data
/// is at most 4 KB.
#[derive(serde::Serialize, serde::Deserialize, Debug, PartialEq)]
pub struct Alert {
    pub kind: String,
    pub title: String,
    pub url: String,
}

/// Whether alerts are on.
pub fn enabled(app: &Shared) -> bool {
    !app.config.push.fcm_project.is_empty()
}

/// Send `alert` to every phone of `account_id`, in the background.
pub fn send(app: &Shared, account_id: i64, alert: Alert) {
    if !enabled(app) {
        return;
    }
    let app = app.clone();
    tokio::spawn(async move {
        if let Err(e) = send_now(&app, account_id, &alert).await {
            tracing::warn!("push: alert for account {account_id}: {e}");
        }
    });
}

async fn access_token(app: &Shared, client: &reqwest::Client) -> Result<String, String> {
    if let Some((t, until)) = TOKEN.lock().unwrap().clone() {
        if Instant::now() < until {
            return Ok(t);
        }
    }
    #[derive(serde::Deserialize)]
    struct Got {
        access_token: String,
        expires_in: u64,
    }
    let got: Got = client
        .get(&app.config.push.token_url)
        .header("Metadata-Flavor", "Google")
        .send()
        .await
        .and_then(|r| r.error_for_status())
        .map_err(|e| format!("the metadata server's token: {e}"))?
        .json()
        .await
        .map_err(|e| format!("the metadata server's token: {e}"))?;
    let until = Instant::now() + Duration::from_secs(got.expires_in.saturating_sub(60));
    *TOKEN.lock().unwrap() = Some((got.access_token.clone(), until));
    Ok(got.access_token)
}

/// Send `alert` to each of the account's phones now; a phone Cloud
/// Messaging no longer knows is forgotten.
pub async fn send_now(app: &Shared, account_id: i64, alert: &Alert) -> Result<usize, String> {
    let devices = pdb::of_account(&app.db, account_id).await.map_err(|e| e.to_string())?;
    if devices.is_empty() {
        return Ok(0);
    }
    let client = reqwest::Client::builder().timeout(Duration::from_secs(20)).build().map_err(|e| e.to_string())?;
    let token = access_token(app, &client).await?;
    let plain = serde_json::to_vec(alert).map_err(|e| e.to_string())?;
    let url = format!("{}/v1/projects/{}/messages:send", app.config.push.fcm_url.trim_end_matches('/'), app.config.push.fcm_project);
    let mut sent = 0;
    for d in devices.iter().take(MOST_DEVICES) {
        let Some(key) = hex::decode(&d.public_key).ok().and_then(|k| <[u8; 32]>::try_from(k).ok()) else { continue };
        let Some(sealed) = flow_core::alert::seal(&key, &plain) else { continue };
        // Data only, high priority: the app wakes, opens it and shows it;
        // the push service has no title or body to show or read.
        let body = serde_json::json!({ "message": {
            "token": d.fcm_token,
            "data": { "sealed": crate::oidc::b64url(&sealed) },
            "android": { "priority": "high" },
        }});
        let r = client.post(&url).bearer_auth(&token).json(&body).send().await.map_err(|e| e.to_string())?;
        let status = r.status();
        if status.is_success() {
            sent += 1;
        } else {
            let text = r.text().await.unwrap_or_default();
            if status == reqwest::StatusCode::NOT_FOUND || text.contains("UNREGISTERED") {
                pdb::forget_fcm(&app.db, &d.fcm_token).await.ok();
            } else {
                tracing::warn!("push: Cloud Messaging said {status}: {}", text.chars().take(200).collect::<String>());
            }
        }
    }
    Ok(sent)
}

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	113×	pub fn enabled(app: &Shared) -> bool {
31	113×	!app.config.push.fcm_project.is_empty()
32	113×	}
33
34		/// Send `alert` to every phone of `account_id`, in the background.
35	110×	pub fn send(app: &Shared, account_id: i64, alert: Alert) {
36	110×	if !enabled(app) {
37	109×	return;
38	1×	}
39	1×	let app = app.clone();
40	1×	tokio::spawn(async move {
41	1×	if let Err(e) = send_now(&app, account_id, &alert).await {
42	0×	tracing::warn!("push: alert for account {account_id}: {e}");
43	1×	}
44	1×	});
45	110×	}
46
47	2×	async fn access_token(app: &Shared, client: &reqwest::Client) -> Result<String, String> {
48	2×	if let Some((t, until)) = TOKEN.lock().unwrap().clone() {
49	1×	if Instant::now() < until {
50	1×	return Ok(t);
51	0×	}
52	1×	}
53		#[derive(serde::Deserialize)]
54		struct Got {
55		access_token: String,
56		expires_in: u64,
57		}
58	1×	let got: Got = client
59	1×	.get(&app.config.push.token_url)
60	1×	.header("Metadata-Flavor", "Google")
61	1×	.send()
62	1×	.await
63	1×	.and_then(\|r\| r.error_for_status())
64	1×	.map_err(\|e\| format!("the metadata server's token: {e}"))?
65	1×	.json()
66	1×	.await
67	1×	.map_err(\|e\| format!("the metadata server's token: {e}"))?;
68	1×	let until = Instant::now() + Duration::from_secs(got.expires_in.saturating_sub(60));
69	1×	*TOKEN.lock().unwrap() = Some((got.access_token.clone(), until));
70	1×	Ok(got.access_token)
71	2×	}
72
73		/// Send `alert` to each of the account's phones now; a phone Cloud
74		/// Messaging no longer knows is forgotten.
75	2×	pub async fn send_now(app: &Shared, account_id: i64, alert: &Alert) -> Result<usize, String> {
76	2×	let devices = pdb::of_account(&app.db, account_id).await.map_err(\|e\| e.to_string())?;
77	2×	if devices.is_empty() {
78	0×	return Ok(0);
79	2×	}
80	2×	let client = reqwest::Client::builder().timeout(Duration::from_secs(20)).build().map_err(\|e\| e.to_string())?;
81	2×	let token = access_token(app, &client).await?;
82	2×	let plain = serde_json::to_vec(alert).map_err(\|e\| e.to_string())?;
83	2×	let url = format!("{}/v1/projects/{}/messages:send", app.config.push.fcm_url.trim_end_matches('/'), app.config.push.fcm_project);
84	2×	let mut sent = 0;
85	3×	for d in devices.iter().take(MOST_DEVICES) {
86	3×	let Some(key) = hex::decode(&d.public_key).ok().and_then(\|k\| <[u8; 32]>::try_from(k).ok()) else { continue };
87	3×	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	3×	let body = serde_json::json!({ "message": {
91	3×	"token": d.fcm_token,
92	3×	"data": { "sealed": crate::oidc::b64url(&sealed) },
93	3×	"android": { "priority": "high" },
94		}});
95	3×	let r = client.post(&url).bearer_auth(&token).json(&body).send().await.map_err(\|e\| e.to_string())?;
96	3×	let status = r.status();
97	3×	if status.is_success() {
98	2×	sent += 1;
99	2×	} else {
100	1×	let text = r.text().await.unwrap_or_default();
101	1×	if status == reqwest::StatusCode::NOT_FOUND \|\| text.contains("UNREGISTERED") {
102	1×	pdb::forget_fcm(&app.db, &d.fcm_token).await.ok();
103		} else {
104	0×	tracing::warn!("push: Cloud Messaging said {status}: {}", text.chars().take(200).collect::<String>());
105		}
106		}
107		}
108	2×	Ok(sent)
109	2×	}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

Coverage / crates/hub/src/push.rs 86.3%

Keyboard