johnandersen777.bsky.social/knotless
git that lives in your Atmosphere account
Forked from chadtmiller.com/knotless
0
0
//! Push notifications: `dev.knotless.registerPush` and `unregisterPush`,
//! which keep the devices an account hears on, and the round that tells
//! each device of what is new, through Apple's push service. What is new
//! is what `listNotifications` would answer, so an account's choices,
//! blocks and mutes hold here too.
use super::notifications::{Page, asking, page};
use super::*;
use crate::index::{Notified, PushDevice};
use crate::viewer::Viewer;
use anyhow::Context;
use axum::Extension;
use base64::Engine;
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use p256::ecdsa::signature::Signer;
use p256::ecdsa::{Signature, SigningKey};
use p256::pkcs8::DecodePrivateKey;
use serde_json::{Value, json};
use std::sync::Mutex;
use std::time::Instant;
/// The app a notification is for: the topic Apple asks of each.
const TOPIC: &str = "dev.knotless.app";
/// How long the token signed for Apple is used: it refuses one over an
/// hour old, and one signed anew more often than every twenty minutes.
const SIGNED_FOR: Duration = Duration::from_secs(50 * 60);
/// How long Apple is given to answer one notification.
const WAIT: Duration = Duration::from_secs(15);
/// How many notifications an account reads each round, which is also the
/// most its badge counts.
const READ: u32 = 100;
/// The most a device is told of one by one in a round: past that, it is
/// told how many there are.
const ONE_BY_ONE: usize = 3;
/// The most of a title or a line a notification carries, as Apple takes
/// four kilobytes of it.
const CLIP: usize = 200;
/// A device's token: its bytes in hexadecimal, as it goes into a URL.
fn is_token(token: &str) -> bool {
!token.is_empty() && token.len() <= 200 && token.bytes().all(|b| b.is_ascii_hexdigit())
}
/// Who asks to change its devices, as `update_seen` allows it: with a
/// token, or the pages' cookie from the site itself.
fn asking_to_write(app: &App, viewer: Viewer, headers: &HeaderMap) -> Result<(String, Arc<Index>)> {
let token = headers.contains_key(header::AUTHORIZATION);
if !token && !crate::login::same_site(headers) {
return Err(Shown(StatusCode::FORBIDDEN, "not from this site".into()));
}
asking(app, viewer)
}
#[derive(Deserialize)]
pub struct RegisterInput {
token: String,
environment: String,
}
/// `dev.knotless.registerPush`.
pub async fn register_push(
State(app): State<Arc<App>>,
Extension(viewer): Extension<Viewer>,
headers: HeaderMap,
axum::Json(input): axum::Json<RegisterInput>,
) -> Result<Response> {
let (did, index) = asking_to_write(&app, viewer, &headers)?;
if !is_token(&input.token) {
let why = "a push token is hexadecimal, of at most 200 digits";
return Err(Shown(StatusCode::BAD_REQUEST, why.into()));
}
if !["production", "sandbox"].contains(&input.environment.as_str()) {
let why = format!("{:?} is not production or sandbox", input.environment);
return Err(Shown(StatusCode::BAD_REQUEST, why));
}
let now = knotless::records::now();
let kept = index.register_push(&did, &input.token, &input.environment, &now);
kept.map_err(anyhow::Error::from)
.map_err(AppError::internal)?;
Ok(answer(json!({})))
}
#[derive(Deserialize)]
pub struct UnregisterInput {
token: String,
}
/// `dev.knotless.unregisterPush`.
pub async fn unregister_push(
State(app): State<Arc<App>>,
Extension(viewer): Extension<Viewer>,
headers: HeaderMap,
axum::Json(input): axum::Json<UnregisterInput>,
) -> Result<Response> {
let (did, index) = asking_to_write(&app, viewer, &headers)?;
let forgot = index.unregister_push(&did, &input.token);
forgot
.map_err(anyhow::Error::from)
.map_err(AppError::internal)?;
Ok(answer(json!({})))
}
/// Apple's push service, and what this server signs in to it with.
pub struct Apns {
key: SigningKey,
key_id: String,
team: String,
/// Where a device's notification goes, by the environment it says it
/// is in. Other than Apple's only in tests.
pub production: String,
pub sandbox: String,
client: reqwest::Client,
signed: Mutex<Option<(Instant, String)>>,
}
impl Apns {
/// With the key Apple gave, in the `.p8` file at `path`.
pub fn open(
path: &std::path::Path,
key_id: Option<&str>,
team: Option<&str>,
) -> anyhow::Result<Apns> {
let Some((key_id, team)) = key_id.zip(team) else {
anyhow::bail!("--apns-key takes --apns-key-id and --apns-team");
};
let pem = std::fs::read_to_string(path);
let pem = pem.with_context(|| format!("reading {}", path.display()))?;
let key = SigningKey::from_pkcs8_pem(&pem);
let key = key.map_err(|err| anyhow::anyhow!("{}: {err}", path.display()))?;
let client = reqwest::Client::builder().timeout(WAIT).build()?;
Ok(Apns {
key,
key_id: key_id.into(),
team: team.into(),
production: "https://api.push.apple.com".into(),
sandbox: "https://api.sandbox.push.apple.com".into(),
client,
signed: Mutex::new(None),
})
}
/// The token Apple takes for now: the last one, until it is too old.
fn signed(&self) -> String {
let mut kept = self.signed.lock().unwrap();
let fresh = kept.take().filter(|(at, _)| at.elapsed() < SIGNED_FOR);
let fresh = fresh.unwrap_or_else(|| (Instant::now(), self.sign()));
let token = fresh.1.clone();
*kept = Some(fresh);
token
}
/// A token for Apple: ES256, naming the key and the team.
fn sign(&self) -> String {
let part = |value: Value| URL_SAFE_NO_PAD.encode(value.to_string());
let head = part(json!({"alg": "ES256", "kid": self.key_id}));
let claims = part(json!({"iss": self.team, "iat": now_secs()}));
let message = format!("{head}.{claims}");
let signature: Signature = self.key.sign(message.as_bytes());
let signature = URL_SAFE_NO_PAD.encode(signature.to_bytes());
format!("{message}.{signature}")
}
/// Tells `device`, saying what became of it.
async fn send(&self, device: &PushDevice, message: &Value) -> Sent {
let base = match device.environment.as_str() {
"sandbox" => &self.sandbox,
_ => &self.production,
};
let asked = self
.client
.post(format!("{base}/3/device/{}", device.token))
.bearer_auth(self.signed())
.header("apns-topic", TOPIC)
.header("apns-push-type", "alert")
.json(message)
.send()
.await;
let res = match asked {
Ok(res) => res,
Err(err) => return failed(&device.did, &err.to_string()),
};
let status = res.status();
if status.is_success() {
return Sent::Told;
}
let said = res.text().await.unwrap_or_default();
let gone = ["BadDeviceToken", "DeviceTokenNotForTopic"];
if status == StatusCode::GONE || gone.iter().any(|why| said.contains(why)) {
return Sent::Gone;
}
failed(&device.did, &format!("{status}: {said}"))
}
}
/// What became of a notification.
#[derive(Debug, PartialEq)]
enum Sent {
Told,
/// The device no longer takes them: the app is gone from it, say.
Gone,
/// It may do later, so it is asked again next round.
Failed,
}
/// Logs a notification Apple did not take.
fn failed(did: &str, why: &str) -> Sent {
let fields = knotless::telemetry::Fields {
did: Some(did),
error: Some(why.to_string()),
..Default::default()
};
knotless::telemetry::note(knotless::telemetry::Level::WARN, "push not sent", &fields);
Sent::Failed
}
/// What a read gave, or nothing if it failed, logged: the next round
/// reads it again.
fn or_nothing<T: Default, E: std::fmt::Debug>(read: std::result::Result<T, E>) -> T {
read.unwrap_or_else(|err| {
failed("", &format!("reading: {err:?}"));
T::default()
})
}
/// Tells every device of what is new, every `every`, for as long as the
/// server runs.
pub async fn follow(app: Arc<App>, index: Arc<Index>, apns: Apns, every: Duration) {
loop {
round(&app, &index, &apns).await;
tokio::time::sleep(every).await;
}
}
/// Tells every device once of what is new since it was last told.
pub async fn round(app: &App, index: &Index, apns: &Apns) {
let devices = or_nothing(index.push_devices());
for same in devices.chunk_by(|a, b| a.did == b.did) {
let did = &same[0].did;
let all = Page {
limit: READ,
after: None,
kind: None,
};
let (found, _) = or_nothing(page(app, index, did, all).await);
let seen = or_nothing(index.seen(did));
let now = knotless::records::now();
let badge = found.iter().filter(|n| unread(n, seen.as_deref())).count();
for device in same {
let due = due(&found, device, seen.as_deref(), &now);
tell(app, index, apns, device, &due, badge).await;
}
}
}
/// Whether `n` came after its account last looked.
fn unread(n: &Notified, seen: Option<&str>) -> bool {
seen.is_none_or(|seen| n.at.as_str() > seen)
}
/// Of `found`, the latest first, those `device` has yet to hear of, the
/// oldest first. One dated after `now` waits for then, as moving the
/// device's time on for it would silence all before that date.
fn due<'a>(
found: &'a [Notified],
device: &PushDevice,
seen: Option<&str>,
now: &str,
) -> Vec<&'a Notified> {
let new = |n: &&Notified| n.at > device.at && n.at.as_str() <= now && unread(n, seen);
found.iter().rev().filter(new).collect()
}
/// Tells `device` of `due`, the oldest first, and keeps how far it got.
async fn tell(
app: &App,
index: &Index,
apns: &Apns,
device: &PushDevice,
due: &[&Notified],
badge: usize,
) {
let mut messages = Vec::new();
if due.len() > ONE_BY_ONE {
let newest = due.iter().map(|n| n.at.as_str()).max().unwrap_or_default();
let body = format!("{} new notifications", due.len());
messages.push((newest, plain(&body, badge)));
} else {
for n in due {
messages.push((n.at.as_str(), message(app, n, badge).await));
}
}
for (at, message) in messages {
match apns.send(device, &message).await {
Sent::Told => or_nothing(index.pushed(&device.token, at)),
Sent::Gone => return or_nothing(index.unregister_push(&device.did, &device.token)),
Sent::Failed => return,
}
}
}
/// A notification that names nothing, which opens the inbox.
fn plain(body: &str, badge: usize) -> Value {
json!({"aps": {
"alert": {"title": "Knotless", "body": body},
"badge": badge,
"sound": "default",
}})
}
/// The notification for `n`. One about a private repository says nothing
/// of it: its title, its text and who is in it stay away from Apple.
pub async fn message(app: &App, n: &Notified, badge: usize) -> Value {
if n.private {
return plain("New activity in a private repository", badge);
}
let author = handle(app, &n.author).await;
let repo = match &n.repo {
Some((did, name)) => Some((did.as_str(), handle(app, did).await, name.as_str())),
None => None,
};
let title = match &repo {
Some((_, owner, name)) => format!("{owner}/{name}"),
None => author.clone(),
};
let mut alert = json!({"title": title, "body": body(n, &author)});
if let Some(subtitle) = &n.title {
alert["subtitle"] = json!(clip(subtitle));
}
let mut aps = json!({"alert": alert, "badge": badge, "sound": "default"});
if let Some(subject) = &n.subject {
aps["thread-id"] = json!(subject);
}
let mut message = json!({"aps": aps, "author": n.author});
if let Some(subject) = &n.subject {
message["subject"] = json!(subject);
}
if let Some((did, owner, name)) = repo {
message["repo"] = json!({"did": did, "handle": owner, "name": name});
}
message
}
/// What a notification says: who did what, then the start of what they
/// said.
fn body(n: &Notified, author: &str) -> String {
let what = match (what(n), n.reason.as_str()) {
(what, "checks-failed" | "ejected") => what.to_string(),
(what, _) => format!("{author} {what}"),
};
let line = n.text.as_deref().and_then(|t| t.lines().next());
match line {
Some(line) => format!("{what}\n{}", clip(line)),
None => what,
}
}
/// What `n` says happened, as the app's inbox says it.
fn what(n: &Notified) -> &'static str {
let kind = n.uri.split('/').nth(3).unwrap_or_default();
match (n.reason.as_str(), n.verdict.as_deref()) {
("star", _) => "starred your repository",
("follow", _) => "followed you",
("invite", _) => "invited you to join",
("checks-failed", _) => "Checks failed",
("ejected", _) => "Left the merge queue",
("review-requested", _) => "asked for your review",
("assigned", _) => "assigned you",
(_, Some("approve")) => "approved",
(_, Some("changes")) => "asked for changes",
_ if n.mentioned => "mentioned you",
_ if kind == knotless::records::nsid::PULL => "opened a pull",
_ if kind == knotless::records::nsid::ISSUE => "opened an issue",
_ if kind == knotless::records::nsid::RELEASE => "published a release",
_ => "commented",
}
}
/// `text`, no longer than [`CLIP`] characters.
fn clip(text: &str) -> String {
text.chars().take(CLIP).collect()
}
#[cfg(test)]
mod tests {
use super::*;
fn notified(uri: &str, reason: &str, at: &str) -> Notified {
Notified {
uri: uri.into(),
author: "did:a".into(),
reason: reason.into(),
subject: None,
repo: None,
title: None,
at: at.into(),
verdict: None,
mentioned: false,
line_comments: None,
role: None,
checks: Vec::new(),
text: None,
reviewed: false,
state: None,
private: false,
}
}
#[test]
fn a_token_is_hexadecimal() {
assert!(is_token("0aF9"));
assert!(!is_token(""));
assert!(!is_token("../../3/device"));
assert!(!is_token(&"a".repeat(201)));
}
#[test]
fn what_is_due_to_a_device() {
let device = PushDevice {
token: "aa".into(),
did: "did:a".into(),
environment: "production".into(),
at: "2026-01-02T00:00:00.000Z".into(),
};
let found = [
notified("future", "star", "2030-01-01T00:00:00.000Z"),
notified("newer", "star", "2026-01-04T00:00:00.000Z"),
notified("new", "star", "2026-01-03T00:00:00.000Z"),
notified("told", "star", "2026-01-01T00:00:00.000Z"),
];
let now = "2026-06-01T00:00:00.000Z";
let uris = |seen: Option<&str>| {
let due = due(&found, &device, seen, now);
due.iter().map(|n| n.uri.clone()).collect::<Vec<_>>()
};
assert_eq!(uris(None), ["new", "newer"]);
assert_eq!(uris(Some("2026-01-03T12:00:00.000Z")), ["newer"]);
assert!(uris(Some("2026-02-01T00:00:00.000Z")).is_empty());
}
#[test]
fn what_each_says() {
let says = |reason: &str, uri: &str, verdict: Option<&str>, mentioned: bool| {
let mut n = notified(uri, reason, "");
n.verdict = verdict.map(String::from);
n.mentioned = mentioned;
what(&n)
};
let comment = "at://did:a/dev.knotless.comment/1";
let cases = [
("star", "starred your repository"),
("follow", "followed you"),
("invite", "invited you to join"),
("checks-failed", "Checks failed"),
("ejected", "Left the merge queue"),
("review-requested", "asked for your review"),
("assigned", "assigned you"),
("comment", "commented"),
];
for (reason, said) in cases {
assert_eq!(says(reason, comment, None, false), said);
}
assert_eq!(says("comment", comment, Some("approve"), false), "approved");
let changes = says("comment", comment, Some("changes"), false);
assert_eq!(changes, "asked for changes");
assert_eq!(says("mention", comment, None, true), "mentioned you");
let pull = says("pull", "at://did:a/dev.knotless.pull/1", None, false);
assert_eq!(pull, "opened a pull");
let issue = says("issue", "at://did:a/dev.knotless.issue/1", None, false);
assert_eq!(issue, "opened an issue");
let release = says("watching", "at://did:a/dev.knotless.release/1", None, false);
assert_eq!(release, "published a release");
}
#[test]
fn what_a_notification_reads() {
let mut n = notified("at://did:a/dev.knotless.comment/1", "comment", "");
assert_eq!(body(&n, "ann.test"), "ann.test commented");
n.text = Some(format!("{}\nand more", "a".repeat(300)));
let read = body(&n, "ann.test");
assert_eq!(read, format!("ann.test commented\n{}", "a".repeat(200)));
let failed = notified("at://did:a/dev.knotless.pull/1", "checks-failed", "");
assert_eq!(body(&failed, "ann.test"), "Checks failed");
}
#[test]
fn a_failed_read_is_nothing() {
let read: Vec<u32> = or_nothing(Err("no"));
assert!(read.is_empty());
assert_eq!(or_nothing::<_, &str>(Ok(vec![1])), [1]);
}
}