johnandersen777.bsky.social/knotless

Switch repository

Loading…

//! Push notifications: devices registered through the XRPC methods, told by
//! a round of what their account hears of, through a stand-in for Apple's
//! push service.

use super::*;
use axum::extract::Path as UrlPath;
use axum::http::HeaderMap;
use base64::Engine;
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use knotless_serve::index::Notified;
use knotless_serve::web::push::{Apns, follow, message, round};
use p256::ecdsa::signature::Verifier;
use p256::ecdsa::{Signature, SigningKey};
use p256::pkcs8::EncodePrivateKey;
use std::sync::Mutex;

/// What the stand-in was sent.
#[derive(Clone)]
struct Told {
    token: String,
    topic: String,
    kind: String,
    bearer: String,
    message: Value,
}

/// A stand-in for Apple, which answers by the token's first digit: `b` is
/// gone, `c` is a token it never gave, and `d` fails once.
fn apple(rt: &tokio::runtime::Runtime) -> (String, Arc<Mutex<Vec<Told>>>) {
    let told = Arc::new(Mutex::new(Vec::<Told>::new()));
    let kept = told.clone();
    let answer = move |UrlPath(token): UrlPath<String>,
                       headers: HeaderMap,
                       axum::Json(message): axum::Json<Value>| {
        let kept = kept.clone();
        async move {
            let header = |name: &str| headers[name].to_str().unwrap().to_string();
            let mut told = kept.lock().unwrap();
            let again = told.iter().any(|t: &Told| t.token == token);
            told.push(Told {
                token: token.clone(),
                topic: header("apns-topic"),
                kind: header("apns-push-type"),
                bearer: header("authorization"),
                message,
            });
            match token.chars().next() {
                Some('b') => (axum::http::StatusCode::GONE, String::new()),
                Some('c') => (
                    axum::http::StatusCode::BAD_REQUEST,
                    r#"{"reason":"BadDeviceToken"}"#.to_string(),
                ),
                Some('d') if !again => (axum::http::StatusCode::INTERNAL_SERVER_ERROR, "no".into()),
                _ => (axum::http::StatusCode::OK, String::new()),
            }
        }
    };
    let router = axum::Router::new().route("/3/device/{token}", axum::routing::post(answer));
    let listener = rt
        .block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
        .unwrap();
    let base = format!("http://{}", listener.local_addr().unwrap());
    rt.spawn(async move { axum::serve(listener, router).await.unwrap() });
    (base, told)
}

fn until(what: &str, mut done: impl FnMut() -> bool) {
    for _ in 0..600 {
        if done() {
            return;
        }
        std::thread::sleep(Duration::from_millis(50));
    }
    panic!("never: {what}");
}

#[test]
fn a_device_hears_what_its_account_does() {
    let server = gateway_with(None, true);
    let rt = tokio::runtime::Runtime::new().unwrap();
    let owner = person(account());
    let fan = person(account());
    let url = owner.url("rung");
    let work = owner.work("w", 1);
    owner.git(&work, &["push", "-q", &url, "main"]);
    let xrpc = |nsid: &str| format!("{}/xrpc/dev.knotless.{nsid}", server.base);
    let write = |of: &Sandbox, nsid: &str, input: Value| {
        let token = token(of, &server.did, &format!("dev.knotless.{nsid}"));
        let body = serde_json::to_vec(&input).unwrap();
        post(&xrpc(nsid), Some(&token), body, true)
    };
    let register = |of: &Sandbox, token: &str, environment: &str| {
        let input = json!({"token": token, "environment": environment});
        write(of, "registerPush", input)
    };

    // Only a token and an environment it knows, as an account, from here.
    assert_eq!(register(&owner, "../../up", "production").0, 400);
    assert_eq!(register(&owner, "ab12", "moon").0, 400);
    let input = json!({"token": "ab12", "environment": "production"});
    let away = Some("https://elsewhere.example");
    let path = xrpc("registerPush");
    assert_eq!(
        browse("POST", &path, None, away, Some(input.clone())).0,
        403
    );
    assert_eq!(browse("POST", &path, None, None, Some(input)).0, 401);

    let device = |digit: &str| digit.repeat(64);
    assert_eq!(
        register(&owner, &device("a"), "production"),
        (200, "{}".into())
    );
    for digit in ["b", "c", "d"] {
        assert_eq!(register(&owner, &device(digit), "production").0, 200);
    }
    // One no one answers for, on the service that is down.
    assert_eq!(register(&owner, &device("e"), "sandbox").0, 200);
    // One the account gives up hears nothing.
    assert_eq!(register(&owner, &device("f"), "production").0, 200);
    let input = json!({"token": device("f")});
    assert_eq!(write(&owner, "unregisterPush", input).1, "{}");

    let (base, told) = apple(&rt);
    let dir = tempfile::tempdir().unwrap();
    let signing = SigningKey::from_bytes((&[7u8; 32]).into()).unwrap();
    let pem = signing.to_pkcs8_pem(Default::default()).unwrap();
    let key = dir.path().join("apns.p8");
    std::fs::write(&key, pem.as_bytes()).unwrap();
    let mut apns = Apns::open(&key, Some("KEY"), Some("TEAM")).unwrap();
    apns.production = base;
    apns.sandbox = "http://127.0.0.1:9".into();
    let index = server.app.index.clone().unwrap();
    let tokens = || {
        let all = index.push_devices().unwrap();
        let mut tokens: Vec<_> = all.into_iter().map(|d| d.token).collect();
        tokens.sort();
        tokens
    };
    let unread = || {
        let url = xrpc("getUnreadCount");
        let token = token(&owner, &server.did, "dev.knotless.getUnreadCount");
        let client = reqwest::blocking::Client::new();
        let res = client.get(url).bearer_auth(token).send().unwrap();
        res.json::<Value>().unwrap()["count"].as_u64().unwrap()
    };
    let to = |digit: &str| {
        let told = told.lock().unwrap();
        let of = told.iter().filter(|t| t.token == device(digit));
        of.cloned().collect::<Vec<Told>>()
    };

    // Nothing yet, nothing told.
    rt.block_on(round(&server.app, &index, &apns));
    assert!(told.lock().unwrap().is_empty());

    // A follow reaches each device; those Apple no longer knows are let go,
    // and the one it failed stays for the next round.
    let handle = owner.account.handle.clone();
    assert_eq!(fan.cli(&["account", "follow", &handle], b"").0, 0);
    until("the follow is indexed", || unread() == 1);
    rt.block_on(round(&server.app, &index, &apns));
    let first = to("a");
    assert_eq!(first.len(), 1);
    let first = &first[0];
    assert_eq!(first.topic, "dev.knotless.app");
    assert_eq!(first.kind, "alert");
    let alert = &first.message["aps"]["alert"];
    let them = &fan.account.handle;
    assert_eq!(alert["title"], them.as_str());
    assert_eq!(alert["body"], format!("{them} followed you"));
    assert_eq!(first.message["aps"]["badge"], 1);
    assert_eq!(first.message["author"], fan.account.did);
    // Signed as Apple asks, with the key it gave.
    let bearer = first.bearer.strip_prefix("Bearer ").unwrap();
    let (signed, signature) = bearer.rsplit_once('.').unwrap();
    let signature = URL_SAFE_NO_PAD.decode(signature).unwrap();
    let signature = Signature::from_slice(&signature).unwrap();
    signing
        .verifying_key()
        .verify(signed.as_bytes(), &signature)
        .unwrap();
    let (head, claims) = signed.split_once('.').unwrap();
    let part = |p: &str| {
        let bytes = URL_SAFE_NO_PAD.decode(p).unwrap();
        serde_json::from_slice::<Value>(&bytes).unwrap()
    };
    assert_eq!(part(head), json!({"alg": "ES256", "kid": "KEY"}));
    assert_eq!(part(claims)["iss"], "TEAM");
    assert!(to("f").is_empty());
    assert_eq!(tokens(), [device("a"), device("d"), device("e")]);

    // Told once: only the one that failed hears of it again.
    rt.block_on(round(&server.app, &index, &apns));
    assert_eq!(to("a").len(), 1);
    assert_eq!(to("d").len(), 2);
    rt.block_on(round(&server.app, &index, &apns));
    assert_eq!(to("d").len(), 2);

    // An issue says where it is, and what it says.
    let issue = ["issue", "create", &url, "Slow", "-m", "It is.\nVery."];
    let (code, uri) = fan.cli(&issue, b"");
    assert_eq!(code, 0, "{uri}");
    until("the issue is indexed", || unread() == 2);
    rt.block_on(round(&server.app, &index, &apns));
    let second = to("a");
    assert_eq!(second.len(), 2);
    let second = &second[1].message;
    let alert = &second["aps"]["alert"];
    assert_eq!(alert["title"], format!("{handle}/rung"));
    assert_eq!(alert["subtitle"], "Slow");
    assert_eq!(alert["body"], format!("{them} opened an issue\nIt is."));
    assert_eq!(second["aps"]["badge"], 2);
    assert_eq!(second["aps"]["thread-id"], uri.trim());
    assert_eq!(second["subject"], uri.trim());
    let repo = json!({"did": owner.account.did, "handle": handle, "name": "rung"});
    assert_eq!(second["repo"], repo);

    // Many at once are told as many.
    for n in 0..4 {
        let (code, said) = fan.cli(&["issue", "create", &url, &format!("Slow {n}")], b"");
        assert_eq!(code, 0, "{said}");
    }
    until("the issues are indexed", || unread() == 6);
    rt.block_on(round(&server.app, &index, &apns));
    let third = to("a");
    assert_eq!(third.len(), 3);
    let third = &third[2].message;
    assert_eq!(third["aps"]["alert"]["body"], "4 new notifications");
    assert_eq!(third["aps"]["badge"], 6);
    assert_eq!(third.get("subject"), None);

    // One about a private repository says nothing of it.
    let private = Notified {
        uri: "at://did:a/dev.knotless.issue/1".into(),
        author: "did:a".into(),
        reason: "issue".into(),
        subject: Some("at://did:a/dev.knotless.issue/1".into()),
        repo: Some(("did:a".into(), "secret".into())),
        title: Some("A secret".into()),
        at: "2026-01-01T00:00:00.000Z".into(),
        verdict: None,
        mentioned: false,
        line_comments: None,
        role: None,
        checks: Vec::new(),
        text: Some("Secrets".into()),
        reviewed: false,
        state: None,
        private: true,
    };
    let said = rt.block_on(message(&server.app, &private, 1)).to_string();
    assert!(
        said.contains("New activity in a private repository"),
        "{said}"
    );
    assert!(!said.to_lowercase().contains("secret"), "{said}");

    // Round after round, for as long as it runs.
    let rounds = follow(
        server.app.clone(),
        index.clone(),
        apns,
        Duration::from_millis(1),
    );
    let ran = rt.block_on(async { tokio::time::timeout(Duration::from_millis(200), rounds).await });
    assert!(ran.is_err());
}