use std::sync::Arc;
use reqwest::StatusCode;
use serde_json::json;
use web_push_native::{
jwt_simple::algorithms::ES256KeyPair, p256::PublicKey, Auth, WebPushBuilder,
};
use crate::db::{Db, Subscription, VapidKeys};
fn session_url(session: &str, window: Option<&str>) -> String {
match window {
Some(w) if !w.is_empty() => format!("s/{session}?w={w}"),
_ => format!("s/{session}"),
}
}
pub fn fire_bell(db: Arc<Db>, contact: String, session: &str, window: Option<&str>) {
let payload = Payload {
title: "mobux".to_string(),
body: format!("session {session}: 🔔"),
tag: Some(format!("bell-{session}")),
url: Some(session_url(session, window)),
};
tokio::spawn(notify(db, contact, payload));
}
pub struct Payload {
pub title: String,
pub body: String,
pub tag: Option<String>,
pub url: Option<String>,
}
pub const NO_SUBSCRIBED_DEVICE: &str =
"no subscribed device: open mobux on the phone and turn on notifications in Settings";
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct Delivery {
pub sent: usize,
pub failed: usize,
pub pruned: usize,
}
pub async fn send_to_devices(
db: Arc<Db>,
contact: String,
payload: Payload,
) -> anyhow::Result<Delivery> {
if db.list_subscriptions()?.is_empty() {
anyhow::bail!(NO_SUBSCRIBED_DEVICE);
}
let delivery = notify(db, contact, payload).await;
if delivery.sent == 0 {
anyhow::bail!(
"no device took the notification: failed={} pruned={}",
delivery.failed,
delivery.pruned
);
}
Ok(delivery)
}
pub async fn notify(db: Arc<Db>, contact: String, payload: Payload) -> Delivery {
let vapid = match db.vapid_keys() {
Ok(v) => v,
Err(e) => {
eprintln!("push: load vapid keys failed: {e:#}");
return Delivery::default();
}
};
let subs = match db.list_subscriptions() {
Ok(s) => s,
Err(e) => {
eprintln!("push: list subscriptions failed: {e:#}");
return Delivery::default();
}
};
if subs.is_empty() {
return Delivery::default();
}
let payload_bytes = json!({
"title": payload.title,
"body": payload.body,
"tag": payload.tag,
"url": payload.url.unwrap_or_else(|| "./".to_string()),
})
.to_string()
.into_bytes();
eprintln!(
"push: notify title={:?} subscribers={}",
payload.title,
subs.len()
);
let client = reqwest::Client::new();
let mut sent = 0usize;
let mut failed = 0usize;
let mut pruned = 0usize;
for sub in subs {
match deliver(&client, &vapid, &contact, &sub, payload_bytes.clone()).await {
DeliveryOutcome::Ok => sent += 1,
DeliveryOutcome::Gone => {
if let Err(e) = db.remove_subscription(&sub.endpoint) {
eprintln!(
"push: failed to prune dead subscription {}: {e:#}",
sub.endpoint
);
} else {
pruned += 1;
}
}
DeliveryOutcome::Failed => failed += 1,
}
}
eprintln!("push: notify sent={sent} failed={failed} pruned={pruned}");
Delivery {
sent,
failed,
pruned,
}
}
enum DeliveryOutcome {
Ok,
Gone,
Failed,
}
async fn deliver(
client: &reqwest::Client,
vapid: &VapidKeys,
contact: &str,
sub: &Subscription,
payload: Vec<u8>,
) -> DeliveryOutcome {
let key_pair = match ES256KeyPair::from_bytes(&vapid.private_key) {
Ok(k) => k,
Err(e) => {
eprintln!("push: invalid VAPID private key: {e}");
return DeliveryOutcome::Failed;
}
};
let endpoint_uri = match sub.endpoint.parse() {
Ok(u) => u,
Err(e) => {
eprintln!("push: bad endpoint {}: {e}", sub.endpoint);
return DeliveryOutcome::Failed;
}
};
let ua_public = match PublicKey::from_sec1_bytes(&sub.p256dh) {
Ok(p) => p,
Err(e) => {
eprintln!("push: bad p256dh for {}: {e}", sub.endpoint);
return DeliveryOutcome::Failed;
}
};
if sub.auth.len() != 16 {
eprintln!(
"push: bad auth length {} for {} (expected 16)",
sub.auth.len(),
sub.endpoint
);
return DeliveryOutcome::Failed;
}
#[allow(deprecated)]
let ua_auth = Auth::clone_from_slice(&sub.auth);
let builder =
WebPushBuilder::new(endpoint_uri, ua_public, ua_auth).with_vapid(&key_pair, contact);
let request = match builder.build(payload) {
Ok(r) => r,
Err(e) => {
eprintln!("push: build request for {} failed: {e}", sub.endpoint);
return DeliveryOutcome::Failed;
}
};
let (parts, body) = request.into_parts();
let url = parts.uri.to_string();
let mut req = client.post(&url).body(body);
for (name, value) in parts.headers.iter() {
req = req.header(name.as_str(), value.as_bytes());
}
match req.send().await {
Ok(resp) => {
let status = resp.status();
if status.is_success() {
DeliveryOutcome::Ok
} else if status == StatusCode::NOT_FOUND || status == StatusCode::GONE {
eprintln!("push: subscription gone ({}) for {}", status, sub.endpoint);
DeliveryOutcome::Gone
} else {
let body = resp.text().await.unwrap_or_default();
eprintln!(
"push: delivery failed ({}) for {}: {}",
status, sub.endpoint, body
);
DeliveryOutcome::Failed
}
}
Err(e) => {
eprintln!("push: HTTP error for {}: {e}", sub.endpoint);
DeliveryOutcome::Failed
}
}
}
#[cfg(test)]
mod tests {
use super::session_url;
fn resolved_against_scope(scope: &str, url: &str) -> String {
format!("{scope}{url}")
}
#[test]
fn session_url_is_relative() {
assert_eq!(session_url("work", None), "s/work");
assert_eq!(session_url("work", Some("3")), "s/work?w=3");
assert_eq!(session_url("work", Some("")), "s/work");
}
#[test]
fn session_url_lands_where_it_always_did_at_the_empty_prefix() {
let scope = "https://host/";
assert_eq!(
resolved_against_scope(scope, &session_url("work", None)),
"https://host/s/work"
);
assert_eq!(
resolved_against_scope(scope, &session_url("work", Some("3"))),
"https://host/s/work?w=3"
);
}
#[test]
fn session_url_stays_inside_a_path_prefix() {
let scope = "https://host/mobux/";
assert_eq!(
resolved_against_scope(scope, &session_url("work", Some("3"))),
"https://host/mobux/s/work?w=3"
);
}
}