//! 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>>) { let told = Arc::new(Mutex::new(Vec::::new())); let kept = told.clone(); let answer = move |UrlPath(token): UrlPath, headers: HeaderMap, axum::Json(message): axum::Json| { 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::().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::>() }; // 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::(&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()); }