use std::sync::Arc;
use std::time::Duration;
use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use serde_json::Value;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Platform {
Ios,
Android,
Web,
}
impl Platform {
pub fn name(&self) -> &'static str {
match self {
Platform::Ios => "ios",
Platform::Android => "android",
Platform::Web => "web",
}
}
}
impl std::fmt::Display for Platform {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(self.name())
}
}
#[derive(Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Recipient {
Apns { device_token: String },
Fcm { registration_token: String },
WebPush {
endpoint: String,
p256dh: String,
auth: String,
},
}
impl Recipient {
pub fn apns(device_token: impl Into<String>) -> Self {
Recipient::Apns {
device_token: device_token.into(),
}
}
pub fn fcm(registration_token: impl Into<String>) -> Self {
Recipient::Fcm {
registration_token: registration_token.into(),
}
}
pub fn web_push(
endpoint: impl Into<String>,
p256dh: impl Into<String>,
auth: impl Into<String>,
) -> Self {
Recipient::WebPush {
endpoint: endpoint.into(),
p256dh: p256dh.into(),
auth: auth.into(),
}
}
pub fn platform(&self) -> Platform {
match self {
Recipient::Apns { .. } => Platform::Ios,
Recipient::Fcm { .. } => Platform::Android,
Recipient::WebPush { .. } => Platform::Web,
}
}
}
fn fingerprint(value: &str) -> String {
use sha2::{Digest as _, Sha256};
use std::fmt::Write as _;
let digest = Sha256::digest(value.as_bytes());
let mut hex = String::with_capacity(12);
for byte in digest.iter().take(6) {
let _ = write!(hex, "{byte:02x}");
}
hex
}
impl std::fmt::Debug for Recipient {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Recipient::Apns { device_token } => write!(
f,
"Recipient::Apns {{ device_token: fp:{} }}",
fingerprint(device_token)
),
Recipient::Fcm { registration_token } => write!(
f,
"Recipient::Fcm {{ registration_token: fp:{} }}",
fingerprint(registration_token)
),
Recipient::WebPush {
endpoint, p256dh, ..
} => write!(
f,
"Recipient::WebPush {{ endpoint: fp:{}, p256dh: fp:{}, auth: [redacted] }}",
fingerprint(endpoint),
fingerprint(p256dh)
),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum Priority {
#[default]
Immediate,
Conserve,
}
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct LocKeys {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub title_loc_key: Option<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub title_loc_args: Vec<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub body_loc_key: Option<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub body_loc_args: Vec<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Notification {
pub title: String,
pub body: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub category: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub thread_id: Option<String>,
#[serde(default)]
pub data: Value,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub collapse_id: Option<String>,
#[serde(default)]
pub priority: Priority,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub icon: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub url: Option<String>,
#[serde(
default,
skip_serializing_if = "Option::is_none",
with = "duration_secs"
)]
pub ttl: Option<Duration>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub badge: Option<u32>,
#[serde(default, skip_serializing_if = "is_false")]
pub silent: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub loc: Option<LocKeys>,
}
#[allow(clippy::trivially_copy_pass_by_ref)] fn is_false(value: &bool) -> bool {
!*value
}
#[must_use]
pub fn ttl_secs(ttl: Duration) -> u64 {
match ttl.as_secs() {
0 if !ttl.is_zero() => 1,
secs => secs,
}
}
const RETRY_AFTER: &str = "retry-after";
static IMF_FIXDATE: std::sync::LazyLock<
Vec<time::format_description::BorrowedFormatItem<'static>>,
> = std::sync::LazyLock::new(|| {
time::format_description::parse_borrowed::<2>(
"[weekday repr:short], [day] [month repr:short] [year] [hour]:[minute]:[second] GMT",
)
.expect("a format description that is a literal in this file")
});
fn parse_http_date(value: &str) -> Option<time::OffsetDateTime> {
time::PrimitiveDateTime::parse(value, IMF_FIXDATE.as_slice())
.ok()
.map(time::PrimitiveDateTime::assume_utc)
}
#[must_use]
pub fn retry_after(headers: &http::HeaderMap, clock: &dyn crate::Clock) -> Option<Duration> {
let value = headers.get(RETRY_AFTER)?.to_str().ok()?.trim();
if let Ok(seconds) = value.parse::<u64>() {
return Some(Duration::from_secs(seconds));
}
let when = parse_http_date(value)?;
let delta = when - clock.now();
if delta.is_negative() {
return Some(Duration::ZERO);
}
Duration::try_from(delta).ok()
}
mod duration_secs {
use serde::{Deserialize, Deserializer, Serializer};
use std::time::Duration;
#[allow(clippy::ref_option)]
pub fn serialize<S: Serializer>(
value: &Option<Duration>,
serializer: S,
) -> Result<S::Ok, S::Error> {
match value {
Some(duration) => serializer.serialize_some(&super::ttl_secs(*duration)),
None => serializer.serialize_none(),
}
}
pub fn deserialize<'de, D: Deserializer<'de>>(
deserializer: D,
) -> Result<Option<Duration>, D::Error> {
Ok(Option::<u64>::deserialize(deserializer)?.map(Duration::from_secs))
}
}
impl Notification {
#[must_use]
pub fn new(title: impl Into<String>, body: impl Into<String>) -> Self {
Self {
title: title.into(),
body: body.into(),
category: None,
thread_id: None,
data: Value::Null,
collapse_id: None,
priority: Priority::Immediate,
icon: None,
url: None,
ttl: None,
badge: None,
silent: false,
loc: None,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PushOutcome {
Delivered { id: Option<String> },
NotConfigured,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PushError {
Unregistered,
Rejected(String),
Transient {
message: String,
retry_after: Option<Duration>,
},
}
impl std::fmt::Display for PushError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let scrub = crate::logging::scrub_text;
match self {
Self::Unregistered => f.write_str("device token is no longer registered; delete it"),
Self::Rejected(message) => write!(f, "push rejected: {}", scrub(message)),
Self::Transient { message, .. } => {
write!(f, "push failed, retryable: {}", scrub(message))
}
}
}
}
impl std::error::Error for PushError {}
impl PushError {
pub fn transient(message: impl Into<String>) -> Self {
PushError::Transient {
message: message.into(),
retry_after: None,
}
}
pub fn transient_after(message: impl Into<String>, retry_after: Option<Duration>) -> Self {
PushError::Transient {
message: message.into(),
retry_after,
}
}
pub fn unsupported_recipient(recipient: &Recipient) -> Self {
PushError::Rejected(format!(
"unsupported recipient: this adapter does not serve {}",
recipient.platform()
))
}
pub fn retry_after(&self) -> Option<Duration> {
match self {
PushError::Transient { retry_after, .. } => *retry_after,
_ => None,
}
}
}
#[async_trait]
pub trait Push: Send + Sync {
async fn send(
&self,
to: &Recipient,
notification: &Notification,
) -> Result<PushOutcome, PushError>;
}
#[derive(Default, Clone)]
pub struct RoutingPush {
apns: Option<Arc<dyn Push>>,
fcm: Option<Arc<dyn Push>>,
web_push: Option<Arc<dyn Push>>,
}
impl RoutingPush {
#[must_use]
pub fn new() -> Self {
Self::default()
}
#[must_use]
pub fn apns(mut self, push: Arc<dyn Push>) -> Self {
self.apns = Some(push);
self
}
#[must_use]
pub fn fcm(mut self, push: Arc<dyn Push>) -> Self {
self.fcm = Some(push);
self
}
#[must_use]
pub fn web_push(mut self, push: Arc<dyn Push>) -> Self {
self.web_push = Some(push);
self
}
#[must_use]
pub fn route_for(&self, recipient: &Recipient) -> Option<&Arc<dyn Push>> {
match recipient {
Recipient::Apns { .. } => self.apns.as_ref(),
Recipient::Fcm { .. } => self.fcm.as_ref(),
Recipient::WebPush { .. } => self.web_push.as_ref(),
}
}
}
impl std::fmt::Debug for RoutingPush {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("RoutingPush")
.field("apns", &self.apns.is_some())
.field("fcm", &self.fcm.is_some())
.field("web_push", &self.web_push.is_some())
.finish()
}
}
#[async_trait]
impl Push for RoutingPush {
async fn send(
&self,
to: &Recipient,
notification: &Notification,
) -> Result<PushOutcome, PushError> {
match self.route_for(to) {
Some(push) => push.send(to, notification).await,
None => Ok(PushOutcome::NotConfigured),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_recipient_round_trips_through_json() {
for recipient in [
Recipient::apns("devicetoken"),
Recipient::fcm("registration"),
Recipient::web_push("https://fcm.googleapis.com/wp/x", "p256dh-key", "auth-key"),
] {
let json = serde_json::to_string(&recipient).expect("serialises");
let back: Recipient = serde_json::from_str(&json).expect("deserialises");
assert_eq!(recipient, back);
}
}
#[test]
fn the_recipient_tag_is_the_transport_name() {
let json = serde_json::to_value(Recipient::web_push("https://e", "p", "a")).unwrap();
assert_eq!(json["web_push"]["endpoint"], "https://e");
assert_eq!(json["web_push"]["p256dh"], "p");
assert_eq!(json["web_push"]["auth"], "a");
}
#[test]
fn debug_prints_no_credential_material() {
let web = Recipient::web_push(
"https://fcm.googleapis.com/wp/cAPABILITYtokenPATH",
"BP256dhPublicKeyValue",
"AuthSecretValue",
);
let printed = format!("{web:?}");
assert!(printed.contains("WebPush"), "{printed}");
assert!(!printed.contains("AuthSecretValue"), "{printed}");
assert!(!printed.contains("cAPABILITYtokenPATH"), "{printed}");
assert!(!printed.contains("fcm.googleapis.com"), "{printed}");
assert!(!printed.contains("BP256dhPublicKeyValue"), "{printed}");
let apns = format!("{:?}", Recipient::apns("deadbeefDEVICEtoken"));
assert!(apns.contains("Apns"), "{apns}");
assert!(!apns.contains("deadbeefDEVICEtoken"), "{apns}");
let fcm = format!("{:?}", Recipient::fcm("REGISTRATIONtokenValue"));
assert!(fcm.contains("Fcm"), "{fcm}");
assert!(!fcm.contains("REGISTRATIONtokenValue"), "{fcm}");
assert_eq!(printed, format!("{web:?}"));
assert_ne!(
printed,
format!("{:?}", Recipient::web_push("https://other", "p", "a"))
);
}
#[test]
fn a_recipient_inside_a_struct_is_redacted_too() {
#[derive(Debug)]
struct Row {
recipient: Recipient,
}
let row = Row {
recipient: Recipient::apns("deadbeefDEVICEtoken"),
};
let printed = format!("{row:?}");
assert!(!printed.contains("deadbeefDEVICEtoken"), "{printed}");
assert_eq!(row.recipient.platform(), Platform::Ios);
}
#[test]
fn platform_is_a_transport_fact() {
assert_eq!(Recipient::apns("t").platform(), Platform::Ios);
assert_eq!(Recipient::fcm("t").platform(), Platform::Android);
assert_eq!(
Recipient::web_push("https://ntfy.sh/x", "p", "a").platform(),
Platform::Web
);
}
#[test]
fn a_minimal_notification_serialises_to_title_body_data_priority() {
let json = serde_json::to_value(Notification::new("Hi", "there")).unwrap();
let object = json.as_object().expect("object");
let mut keys: Vec<&str> = object.keys().map(String::as_str).collect();
keys.sort_unstable();
assert_eq!(keys, ["body", "data", "priority", "title"]);
}
#[test]
fn a_full_notification_round_trips_and_ttl_is_seconds() {
let mut notification = Notification::new("Room starting", "Yoga in 10 min");
notification.category = Some("SESSION".to_owned());
notification.thread_id = Some("room-42".to_owned());
notification.data = serde_json::json!({ "room_id": "42" });
notification.collapse_id = Some("room-42".to_owned());
notification.priority = Priority::Conserve;
notification.icon = Some("https://example.test/icon.png".to_owned());
notification.url = Some("https://example.test/rooms/42".to_owned());
notification.ttl = Some(Duration::from_secs(3_600));
notification.badge = Some(3);
notification.silent = true;
notification.loc = Some(LocKeys {
title_loc_key: Some("ROOM_STARTING".to_owned()),
title_loc_args: vec!["Yoga".to_owned()],
body_loc_key: Some("ROOM_BODY".to_owned()),
body_loc_args: vec!["10".to_owned()],
});
let json = serde_json::to_value(¬ification).unwrap();
assert_eq!(json["ttl"], 3_600, "a TTL is whole seconds on the wire");
assert_eq!(json["silent"], true);
assert_eq!(json["loc"]["title_loc_key"], "ROOM_STARTING");
let back: Notification = serde_json::from_value(json).unwrap();
assert_eq!(back, notification);
}
#[test]
fn a_notification_from_the_pre_recipient_wire_form_still_parses() {
let old = serde_json::json!({
"title": "Hi",
"body": "there",
"data": {},
"priority": "immediate"
});
let notification: Notification = serde_json::from_value(old).unwrap();
assert_eq!(notification.title, "Hi");
assert_eq!(notification.ttl, None);
assert!(!notification.silent);
}
#[test]
fn a_sub_second_ttl_rounds_up_and_never_becomes_drop_now() {
assert_eq!(ttl_secs(Duration::from_millis(900)), 1);
assert_eq!(ttl_secs(Duration::from_nanos(1)), 1);
assert_eq!(ttl_secs(Duration::ZERO), 0, "zero is the deliberate drop");
assert_eq!(ttl_secs(Duration::from_secs(3_600)), 3_600);
assert_eq!(
ttl_secs(Duration::from_millis(1_900)),
1,
"whole seconds otherwise truncate, as every protocol does"
);
let mut notification = Notification::new("a", "b");
notification.ttl = Some(Duration::from_millis(900));
let json = serde_json::to_value(¬ification).unwrap();
assert_eq!(json["ttl"], 1, "sub-second serialises as one second");
let back: Notification = serde_json::from_value(json).unwrap();
assert_eq!(
back.ttl,
Some(Duration::from_secs(1)),
"and comes back meaning one second, not Duration::ZERO"
);
let mut zero = Notification::new("a", "b");
zero.ttl = Some(Duration::ZERO);
let json = serde_json::to_value(&zero).unwrap();
assert_eq!(json["ttl"], 0);
let back: Notification = serde_json::from_value(json).unwrap();
assert_eq!(back.ttl, Some(Duration::ZERO));
}
#[test]
fn transient_carries_an_optional_retry_after() {
let plain = PushError::transient("apns 503");
assert_eq!(plain.retry_after(), None);
let throttled = PushError::transient_after("apns 429", Some(Duration::from_secs(30)));
assert_eq!(throttled.retry_after(), Some(Duration::from_secs(30)));
assert_eq!(PushError::Unregistered.retry_after(), None);
}
#[test]
fn an_unsupported_recipient_names_the_transport() {
let error = PushError::unsupported_recipient(&Recipient::fcm("t"));
let message = error.to_string();
assert!(message.contains("unsupported recipient"), "{message}");
assert!(message.contains("android"), "{message}");
}
struct Recording {
label: &'static str,
seen: std::sync::atomic::AtomicUsize,
}
impl Recording {
fn new(label: &'static str) -> Arc<Self> {
Arc::new(Self {
label,
seen: std::sync::atomic::AtomicUsize::new(0),
})
}
fn count(&self) -> usize {
self.seen.load(std::sync::atomic::Ordering::Relaxed)
}
}
#[async_trait]
impl Push for Recording {
async fn send(
&self,
_to: &Recipient,
_notification: &Notification,
) -> Result<PushOutcome, PushError> {
self.seen.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
Ok(PushOutcome::Delivered {
id: Some(self.label.to_owned()),
})
}
}
#[test]
fn the_router_dispatches_by_variant() {
let apns = Recording::new("apns");
let web = Recording::new("web");
let router = RoutingPush::new().apns(apns.clone()).web_push(web.clone());
let notification = Notification::new("a", "b");
let outcome =
pollster::block_on(router.send(&Recipient::apns("t"), ¬ification)).unwrap();
assert_eq!(
outcome,
PushOutcome::Delivered {
id: Some("apns".to_owned())
}
);
let outcome = pollster::block_on(
router.send(&Recipient::web_push("https://e", "p", "a"), ¬ification),
)
.unwrap();
assert_eq!(
outcome,
PushOutcome::Delivered {
id: Some("web".to_owned())
}
);
assert_eq!(apns.count(), 1);
assert_eq!(web.count(), 1);
}
#[test]
fn a_transport_with_no_adapter_is_not_configured_not_rejected() {
let router = RoutingPush::new().apns(Recording::new("apns"));
let outcome =
pollster::block_on(router.send(&Recipient::fcm("t"), &Notification::new("a", "b")))
.unwrap();
assert_eq!(outcome, PushOutcome::NotConfigured);
}
#[test]
fn an_empty_router_is_not_configured_for_every_transport() {
let router = RoutingPush::new();
for recipient in [
Recipient::apns("t"),
Recipient::fcm("t"),
Recipient::web_push("https://e", "p", "a"),
] {
let outcome =
pollster::block_on(router.send(&recipient, &Notification::new("a", "b"))).unwrap();
assert_eq!(outcome, PushOutcome::NotConfigured);
assert!(router.route_for(&recipient).is_none());
}
}
#[test]
fn display_sanitizes_the_provider_text() {
let error = PushError::Rejected(
"web push 400 for https://push.example.test/wp/alice?auth=cap-abcdef".to_owned(),
);
let text = error.to_string();
assert!(!text.contains("cap-abcdef"), "{text}");
assert!(text.contains("?[redacted]"), "{text}");
let error = PushError::transient("fcm 503 for alice@example.test");
let text = error.to_string();
assert!(!text.contains('@'), "{text}");
assert!(text.contains("[subject_hash:"), "{text}");
assert_eq!(
PushError::Unregistered.to_string(),
"device token is no longer registered; delete it"
);
assert!(format!("{error:?}").contains("alice@example.test"));
}
}
#[cfg(test)]
mod retry_after_tests {
use super::{parse_http_date, retry_after};
use crate::Clock;
use std::time::Duration;
struct FixedClock(time::OffsetDateTime);
#[async_trait::async_trait]
impl Clock for FixedClock {
fn now(&self) -> time::OffsetDateTime {
self.0
}
}
fn at(unix: i64) -> FixedClock {
FixedClock(time::OffsetDateTime::from_unix_timestamp(unix).expect("a valid instant"))
}
fn headers(value: &str) -> http::HeaderMap {
let mut headers = http::HeaderMap::new();
headers.insert("retry-after", value.parse().expect("a header value"));
headers
}
#[test]
fn an_http_date_parses_and_a_nonsense_one_does_not() {
let parsed = parse_http_date("Sun, 06 Nov 1994 08:49:37 GMT").expect("IMF-fixdate");
assert_eq!(parsed.unix_timestamp(), 784_111_777);
assert!(parse_http_date("Sunday, 06-Nov-94 08:49:37 GMT").is_none());
assert!(parse_http_date("tomorrow").is_none());
}
#[test]
fn the_delta_seconds_form_is_read() {
assert_eq!(
retry_after(&headers("120"), &at(0)),
Some(Duration::from_secs(120))
);
assert_eq!(
retry_after(&headers(" 30 "), &at(0)),
Some(Duration::from_secs(30)),
"surrounding space is not a parse failure"
);
}
#[test]
fn the_http_date_form_is_read_too() {
let an_hour_before = 784_111_777 - 3_600;
assert_eq!(
retry_after(
&headers("Sun, 06 Nov 1994 08:49:37 GMT"),
&at(an_hour_before)
),
Some(Duration::from_secs(3_600))
);
}
#[test]
fn a_date_already_past_means_retry_now_not_no_delay() {
assert_eq!(
retry_after(
&headers("Sun, 06 Nov 1994 08:49:37 GMT"),
&at(784_111_777 + 60)
),
Some(Duration::ZERO)
);
}
#[test]
fn an_absent_or_unparseable_header_states_no_delay() {
assert_eq!(retry_after(&http::HeaderMap::new(), &at(0)), None);
assert_eq!(retry_after(&headers("soon"), &at(0)), None);
assert_eq!(
retry_after(&headers("-5"), &at(0)),
None,
"a negative delta is not a duration"
);
}
}