chadtmiller.com/knotless
git that lives in your Atmosphere account
1
10
The iOS app gets push notifications, sent by knotless-serve
.knotless/workflows/deploy.yml+4−0
| @@ -46,6 +46,10 @@ env: | ||
# proxy holds once deploy/setup.sh has run with this and the zone's | ||
# token. | ||
KNOTLESS_PAGES_DOMAIN: knotless.page | ||
# The iOS app's notifications, once the key is on the host as | ||
# deploy/README.md says. Neither of these is a secret. | ||
KNOTLESS_APNS_KEY_ID: J53A95HDW7 | ||
KNOTLESS_APNS_TEAM: YN68LN9T7Z | ||
KNOTLESS_DATA: /mnt/data/knotless | ||
KNOTLESS_ADMIN_DIDS: did:plc:bcgltzqazw5tb6k2g3ttenbj | ||
# The runner's account (ci.knotless.dev), the triager's | ||
crates/knotless-serve/src/web/notifications.rs+8−6
| @@ -110,16 +110,16 @@ pub async fn list_notifications( | ||
/// Which page of notifications: at most `limit`, after where the one | ||
/// before ended, of one `kind` or all. | ||
#[derive(Clone, Copy)] | ||
struct Page<'a> { | ||
limit: u32, | ||
after: Option<&'a str>, | ||
kind: Option<&'a str>, | ||
pub(super) struct Page<'a> { | ||
pub(super) limit: u32, | ||
pub(super) after: Option<&'a str>, | ||
pub(super) kind: Option<&'a str>, | ||
} | ||
| ||
/// A page of `did`'s notifications, as [`Index::notifications`] gives | ||
/// them: from the index, and from this server's index of the spaces of | ||
/// private repositories, of those `did` can still read, in one order. | ||
async fn page( | ||
pub(super) async fn page( | ||
app: &App, | ||
index: &Index, | ||
did: &str, | ||
| @@ -146,7 +146,9 @@ async fn page( | ||
.heard(did, private) | ||
.context("reading who hears of what") | ||
.map_err(AppError::internal)?; | ||
found.extend(readable(app, spaces, did, &orgs, private).await); | ||
let mut private = readable(app, spaces, did, &orgs, private).await; | ||
private.iter_mut().for_each(|n| n.private = true); | ||
found.extend(private); | ||
} | ||
found.sort_by(|a, b| (&b.at, &b.uri).cmp(&(&a.at, &a.uri))); | ||
more |= found.len() > limit as usize; | ||
crates/knotless-serve/src/web/push.rsnew+491−0
491 lines changed: not loaded until asked for.
crates/knotless-serve/src/index.rs+136−0
| @@ -632,6 +632,10 @@ pub(crate) fn connect(path: &Path) -> rusqlite::Result<(Connection, bool)> { | ||
space TEXT NOT NULL, did TEXT NOT NULL, collection TEXT NOT NULL, | ||
rkey TEXT NOT NULL, PRIMARY KEY (space, did, collection, rkey)); | ||
CREATE TABLE IF NOT EXISTS seen (did TEXT PRIMARY KEY, at TEXT NOT NULL); | ||
CREATE TABLE IF NOT EXISTS push_device ( | ||
token TEXT PRIMARY KEY, did TEXT NOT NULL, environment TEXT NOT NULL, | ||
at TEXT NOT NULL); | ||
CREATE INDEX IF NOT EXISTS push_device_did ON push_device (did, at); | ||
DROP TABLE IF EXISTS notify_pref; | ||
CREATE TABLE IF NOT EXISTS pref ( | ||
did TEXT NOT NULL, kind TEXT NOT NULL, value TEXT NOT NULL, | ||
| @@ -914,6 +918,7 @@ const PURGE: &[&str] = &[ | ||
"DELETE FROM org_member WHERE org = ?1 OR substr(space, 1, length(?2)) = ?2", | ||
"DELETE FROM membership WHERE did = ?1 OR substr(space, 1, length(?2)) = ?2", | ||
"DELETE FROM seen WHERE did = ?1", | ||
"DELETE FROM push_device WHERE did = ?1", | ||
"DELETE FROM pref WHERE did = ?1", | ||
"DELETE FROM subscription WHERE did = ?1", | ||
"DELETE FROM profile_kept WHERE did = ?1", | ||
| @@ -2425,6 +2430,7 @@ impl Index { | ||
text: clipped(row.get(13)?), | ||
reviewed: row.get::<_, Option<bool>>(14)?.unwrap_or_default(), | ||
state: None, | ||
private: false, | ||
}) | ||
}); | ||
let mut rows = rows?.collect::<rusqlite::Result<Vec<_>>>()?; | ||
| @@ -2468,6 +2474,62 @@ impl Index { | ||
stmt.query_row([did], |row| row.get(0)).optional() | ||
} | ||
| ||
/// Keeps a device `did` hears on, told of nothing before `at`. One held | ||
/// already stays as far as it was told, unless another account held it; | ||
/// those of `did` past the latest [`PUSH_DEVICES`] are let go. | ||
pub fn register_push( | ||
&self, | ||
did: &str, | ||
token: &str, | ||
environment: &str, | ||
at: &str, | ||
) -> rusqlite::Result<()> { | ||
let db = self.db.lock().unwrap(); | ||
db.execute( | ||
"INSERT INTO push_device VALUES (?1, ?2, ?3, ?4) | ||
ON CONFLICT (token) DO UPDATE SET | ||
environment = excluded.environment, | ||
at = CASE WHEN did = excluded.did THEN at ELSE excluded.at END, | ||
did = excluded.did", | ||
[token, did, environment, at], | ||
)?; | ||
db.execute( | ||
"DELETE FROM push_device WHERE did = ?1 AND token NOT IN ( | ||
SELECT token FROM push_device WHERE did = ?1 ORDER BY at DESC LIMIT ?2)", | ||
params![did, PUSH_DEVICES], | ||
)?; | ||
Ok(()) | ||
} | ||
| ||
/// Forgets a device of `did`. | ||
pub fn unregister_push(&self, did: &str, token: &str) -> rusqlite::Result<()> { | ||
let db = self.db.lock().unwrap(); | ||
db.execute( | ||
"DELETE FROM push_device WHERE token = ?1 AND did = ?2", | ||
[token, did], | ||
)?; | ||
Ok(()) | ||
} | ||
| ||
/// Every device, those of one account together. | ||
pub fn push_devices(&self) -> rusqlite::Result<Vec<PushDevice>> { | ||
let db = self.readers.take(&self.db); | ||
let mut stmt = | ||
db.prepare_cached("SELECT token, did, environment, at FROM push_device ORDER BY did")?; | ||
let rows = stmt.query_map([], push_device); | ||
rows?.collect() | ||
} | ||
| ||
/// Keeps that the device was told of what came up to `at`. | ||
pub fn pushed(&self, token: &str, at: &str) -> rusqlite::Result<()> { | ||
let db = self.db.lock().unwrap(); | ||
db.execute( | ||
"UPDATE push_device SET at = max(at, ?2) WHERE token = ?1", | ||
[token, at], | ||
)?; | ||
Ok(()) | ||
} | ||
| ||
/// Keeps that `did` saw its notifications up to `at`, unless it saw | ||
/// them later already; gives when it saw them now. | ||
pub fn see(&self, did: &str, at: &str) -> rusqlite::Result<String> { | ||
| @@ -4596,8 +4658,36 @@ pub struct Notified { | ||
/// The state the pull or issue it is about is in: `open`, `closed` or | ||
/// `merged`. | ||
pub state: Option<String>, | ||
/// Whether it is about a private repository: found in the index of its | ||
/// space, and so kept from what a push sends through Apple. | ||
pub private: bool, | ||
} | ||
| ||
/// A device an account hears on, by Apple's push service. | ||
#[derive(Clone, Debug, PartialEq)] | ||
pub struct PushDevice { | ||
pub token: String, | ||
pub did: String, | ||
/// `production` or `sandbox`: which of Apple's services it is known to. | ||
pub environment: String, | ||
/// The time of the latest notification it was told of, or when it was | ||
/// registered. | ||
pub at: String, | ||
} | ||
| ||
/// A row of `push_device`, as [`Index::push_devices`] reads it. | ||
fn push_device(row: &rusqlite::Row) -> rusqlite::Result<PushDevice> { | ||
Ok(PushDevice { | ||
token: row.get(0)?, | ||
did: row.get(1)?, | ||
environment: row.get(2)?, | ||
at: row.get(3)?, | ||
}) | ||
} | ||
| ||
/// The most devices an account hears on. | ||
pub const PUSH_DEVICES: u32 = 20; | ||
| ||
/// The most of what a notification's record says that it carries. | ||
pub const TEXT_MAX: usize = 1000; | ||
| ||
| @@ -5551,6 +5641,52 @@ mod tests { | ||
assert_eq!(states("did:a"), [("pull".into(), Some("merged".into()))]); | ||
} | ||
| ||
#[test] | ||
fn devices_an_account_hears_on() { | ||
let dir = tempfile::tempdir().unwrap(); | ||
let index = Index::open(&dir.path().join("i.db")).unwrap(); | ||
let (early, late) = ("2026-01-01T00:00:00.000Z", "2026-02-01T00:00:00.000Z"); | ||
let at = |token: &str| { | ||
let all = index.push_devices().unwrap(); | ||
let found = all.into_iter().find(|d| d.token == token); | ||
found.map(|d| (d.did, d.environment, d.at)) | ||
}; | ||
let held = |did: &str, env: &str, at: &str| Some((did.into(), env.into(), at.into())); | ||
index | ||
.register_push("did:a", "aa", "sandbox", early) | ||
.unwrap(); | ||
assert_eq!(at("aa"), held("did:a", "sandbox", early)); | ||
// Registered again, as the app does each launch, it stays as far as | ||
// it was told; its time only goes on. | ||
index.pushed("aa", late).unwrap(); | ||
index.pushed("aa", early).unwrap(); | ||
let later = "2026-03-01T00:00:00.000Z"; | ||
index | ||
.register_push("did:a", "aa", "production", later) | ||
.unwrap(); | ||
assert_eq!(at("aa"), held("did:a", "production", late)); | ||
// Another account's now, it hears nothing from before. | ||
index | ||
.register_push("did:b", "aa", "production", later) | ||
.unwrap(); | ||
assert_eq!(at("aa"), held("did:b", "production", later)); | ||
// Only its own account forgets it. | ||
index.unregister_push("did:a", "aa").unwrap(); | ||
assert!(at("aa").is_some()); | ||
index.unregister_push("did:b", "aa").unwrap(); | ||
assert_eq!(at("aa"), None); | ||
// The oldest of too many go. | ||
for n in 0..=PUSH_DEVICES { | ||
let when = format!("2026-01-01T00:00:{n:02}.000Z"); | ||
let token = format!("{n:02}"); | ||
index | ||
.register_push("did:a", &token, "sandbox", &when) | ||
.unwrap(); | ||
} | ||
assert_eq!(index.push_devices().unwrap().len(), PUSH_DEVICES as usize); | ||
assert_eq!(at("00"), None); | ||
} | ||
| ||
#[test] | ||
fn notifications_as_their_account_prefers() { | ||
let dir = tempfile::tempdir().unwrap(); | ||
crates/knotless-serve/src/lib.rs+26−1
| @@ -1048,6 +1048,8 @@ const SWEEP: Duration = Duration::from_secs(600); | ||
/// written there, besides as their authorities tell of writes: for those | ||
/// whose notifications went missing. | ||
const SPACES: Duration = Duration::from_secs(30); | ||
/// How often devices are told of what is new. | ||
const PUSH_ROUND: Duration = Duration::from_secs(30); | ||
| ||
/// How often what was counted is summed up in the log. | ||
const SUMMARY: Duration = Duration::from_secs(60); | ||
| @@ -1321,6 +1323,11 @@ pub struct Options { | ||
/// The domain repositories' sites are answered on, each owner at a | ||
/// label of it, as `knotless.page`: none answers none. | ||
pub pages_domain: Option<String>, | ||
/// The key Apple gave for its push service, a `.p8` file, which with | ||
/// its id and team tells devices of their accounts' notifications. | ||
pub apns_key: Option<std::path::PathBuf>, | ||
pub apns_key_id: Option<String>, | ||
pub apns_team: Option<String>, | ||
} | ||
| ||
/// What `--private`, which is gone, says instead. | ||
| @@ -1354,6 +1361,9 @@ pub fn parse_args(args: &[String]) -> Result<Options, String> { | ||
oauth_url: None, | ||
code_index: None, | ||
pages_domain: None, | ||
apns_key: None, | ||
apns_key_id: None, | ||
apns_team: None, | ||
}; | ||
let mut it = args.iter().skip(1); | ||
while let Some(flag) = it.next() { | ||
| @@ -1373,6 +1383,11 @@ pub fn parse_args(args: &[String]) -> Result<Options, String> { | ||
"--jetstream" => opts.jetstream = Some(value()?).filter(|v| !v.is_empty()), | ||
"--public-url" => opts.public_url = Some(value()?), | ||
"--pages-domain" => opts.pages_domain = Some(value()?).filter(|v| !v.is_empty()), | ||
"--apns-key" => { | ||
opts.apns_key = Some(value()?).filter(|v| !v.is_empty()).map(Into::into) | ||
} | ||
"--apns-key-id" => opts.apns_key_id = Some(value()?).filter(|v| !v.is_empty()), | ||
"--apns-team" => opts.apns_team = Some(value()?).filter(|v| !v.is_empty()), | ||
"--index" => opts.index = Some(value()?.into()), | ||
"--backfill" => opts.backfill = Some(value()?).filter(|v| !v.is_empty()), | ||
"--relay" => opts.relay = Some(value()?), | ||
| @@ -1426,7 +1441,7 @@ pub fn parse_args(args: &[String]) -> Result<Options, String> { | ||
"--code-index" => opts.code_index = Some(value()?.into()), | ||
other => { | ||
return Err(format!( | ||
"unexpected {other}\nusage: knotless-serve [--listen <addr>] [--ttl <secs>] [--firehose <wss://relay or pds>] [--jetstream <wss://host>] [--public-url <https://host>] [--oauth-url <https://host>] [--index <file>] [--backfill <https://pds or relay>] [--relay <https://relay>] [--labeler <did>]... [--default-runner <did>] [--client-minimum <version> [--client-minimum-reason <text>]] [--web <http://app>] [--releases <dir or knotless://repo>] [--stale <secs>] [--usage <file>] [--logs <file> [--logs-keep <lines>]] [--check-index <file>] [--metrics <addr>] [--bots <dir>] [--sign-up-pds <https://pds>] [--code-index <dir>] [--pages-domain <domain>]" | ||
"unexpected {other}\nusage: knotless-serve [--listen <addr>] [--ttl <secs>] [--firehose <wss://relay or pds>] [--jetstream <wss://host>] [--public-url <https://host>] [--oauth-url <https://host>] [--index <file>] [--backfill <https://pds or relay>] [--relay <https://relay>] [--labeler <did>]... [--default-runner <did>] [--client-minimum <version> [--client-minimum-reason <text>]] [--web <http://app>] [--releases <dir or knotless://repo>] [--stale <secs>] [--usage <file>] [--logs <file> [--logs-keep <lines>]] [--check-index <file>] [--metrics <addr>] [--bots <dir>] [--sign-up-pds <https://pds>] [--code-index <dir>] [--apns-key <file> --apns-key-id <id> --apns-team <id>] [--pages-domain <domain>]" | ||
)); | ||
} | ||
} | ||
| @@ -1465,6 +1480,13 @@ pub async fn run(opts: Options, ctx: Ctx) -> Result<()> { | ||
"--pages-domain takes --public-url, where a site's owner is sent from its root" | ||
); | ||
} | ||
let apns = match &opts.apns_key { | ||
Some(path) => { | ||
let (key_id, team) = (opts.apns_key_id.as_deref(), opts.apns_team.as_deref()); | ||
Some(web::push::Apns::open(path, key_id, team)?) | ||
} | ||
None => None, | ||
}; | ||
let gateway = match &opts.public_url { | ||
Some(url) => { | ||
let mut gw = gateway::Gateway::new(url).map_err(anyhow::Error::msg)?; | ||
| @@ -1539,6 +1561,9 @@ pub async fn run(opts: Options, ctx: Ctx) -> Result<()> { | ||
app.watching.store(true, Ordering::Relaxed); | ||
tokio::spawn(jetstream::watch(app.clone(), url)); | ||
} | ||
if let (Some(apns), Some(index)) = (apns, app.index.clone()) { | ||
tokio::spawn(web::push::follow(app.clone(), index, apns, PUSH_ROUND)); | ||
} | ||
tokio::spawn(spaces::follow(app.clone(), SPACES)); | ||
tokio::spawn(languages::follow(app.clone())); | ||
if let Some(follow) = code::follow(app.clone()) { | ||
crates/knotless-serve/src/web.rs+5−0
| @@ -60,6 +60,7 @@ mod opening; | ||
mod orgs; | ||
mod pages; | ||
mod preview; | ||
pub mod push; | ||
mod queue; | ||
mod reactions; | ||
mod releases; | ||
| @@ -86,6 +87,7 @@ use follows::{follow_actor, get_followers, get_follows, unfollow_actor}; | ||
use items::{get_pull_diff, get_thread, list_items}; | ||
use notifications::{get_unread_count, list_notifications, update_seen}; | ||
pub use pages::pages; | ||
use push::{register_push, unregister_push}; | ||
use reactions::{add_reaction, get_reactions, remove_reaction}; | ||
use repo::{ | ||
get_blame, get_commit, get_file, get_last_commits, get_repo, get_tree, list_branches, | ||
| @@ -201,6 +203,8 @@ pub fn queries() -> Router<Arc<App>> { | ||
) | ||
.route("/xrpc/dev.knotless.getUnreadCount", get(get_unread_count)) | ||
.route("/xrpc/dev.knotless.updateSeen", post(update_seen)) | ||
.route("/xrpc/dev.knotless.registerPush", post(register_push)) | ||
.route("/xrpc/dev.knotless.unregisterPush", post(unregister_push)) | ||
.route("/xrpc/dev.knotless.getTree", get(get_tree)) | ||
.route("/xrpc/dev.knotless.getLastCommits", get(get_last_commits)) | ||
.route("/xrpc/dev.knotless.listPaths", get(list_paths)) | ||
| @@ -367,6 +371,7 @@ fn answer<T: Serialize>(value: T) -> Response { | ||
} | ||
| ||
/// A failure, as the pages show it: a status and why. | ||
#[derive(Debug)] | ||
pub struct Shown(StatusCode, String); | ||
| ||
impl From<AppError> for Shown { | ||
crates/knotless-serve/tests/gateway/push.rsnew+268−0
| @@ -0,0 +1,268 @@ | ||
//! 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()); | ||
} | ||
crates/knotless-serve/tests/gateway.rs+2−0
| @@ -23,6 +23,8 @@ mod org_index; | ||
mod orgs; | ||
#[path = "gateway/private_flow.rs"] | ||
mod private_flow; | ||
#[path = "gateway/push.rs"] | ||
mod push; | ||
#[path = "gateway/reviewed.rs"] | ||
mod reviewed; | ||
#[path = "gateway/subscriptions.rs"] | ||
crates/knotless-serve/tests/serve.rs+47−0
| @@ -300,6 +300,29 @@ fn binary_lifecycle() { | ||
.unwrap(); | ||
assert_eq!(out.status.code(), Some(1)); | ||
assert!(String::from_utf8_lossy(&out.stderr).contains("--pages-domain takes --public-url")); | ||
// Apple's push key takes its id and team, and has to be one. | ||
let apns = |args: &[&str]| { | ||
let out = s.command(exe, s.dir.path()).args(args).output().unwrap(); | ||
assert_eq!(out.status.code(), Some(1)); | ||
String::from_utf8_lossy(&out.stderr).to_string() | ||
}; | ||
let key = s.path("apns.p8"); | ||
let key = key.to_string_lossy(); | ||
assert!(apns(&["--apns-key", &key]).contains("--apns-key-id and --apns-team")); | ||
let named = [ | ||
"--apns-key", | ||
&*key, | ||
"--apns-key-id", | ||
"K", | ||
"--apns-team", | ||
"T", | ||
]; | ||
assert!(apns(&named).contains("reading")); | ||
std::fs::write(s.path("apns.p8"), "not a key").unwrap(); | ||
assert!(apns(&named).contains("apns.p8")); | ||
let signing = p256::ecdsa::SigningKey::from_bytes((&[7u8; 32]).into()).unwrap(); | ||
let pem = p256::pkcs8::EncodePrivateKey::to_pkcs8_pem(&signing, Default::default()).unwrap(); | ||
std::fs::write(s.path("apns.p8"), pem.as_bytes()).unwrap(); | ||
| ||
std::fs::create_dir_all(s.path("releases")).unwrap(); | ||
std::fs::write(s.path("releases/latest"), "0.1.0\n").unwrap(); | ||
| @@ -342,6 +365,12 @@ fn binary_lifecycle() { | ||
&s.path("code").to_string_lossy(), | ||
"--pages-domain", | ||
"knotless.page", | ||
"--apns-key", | ||
&key, | ||
"--apns-key-id", | ||
"K", | ||
"--apns-team", | ||
"T", | ||
"--client-minimum", | ||
"0.1.0", | ||
"--client-minimum-reason", | ||
| @@ -631,6 +660,24 @@ fn arguments() { | ||
assert_eq!(o.pages_domain, None); | ||
let o = args(&["--pages-domain", "knotless.page"]).unwrap(); | ||
assert_eq!(o.pages_domain.as_deref(), Some("knotless.page")); | ||
assert_eq!(o.apns_key, None); | ||
let o = args(&[ | ||
"--apns-key", | ||
"/data/apns.p8", | ||
"--apns-key-id", | ||
"K", | ||
"--apns-team", | ||
"T", | ||
]) | ||
.unwrap(); | ||
assert_eq!( | ||
o.apns_key.as_deref(), | ||
Some(std::path::Path::new("/data/apns.p8")) | ||
); | ||
assert_eq!(o.apns_key_id.as_deref(), Some("K")); | ||
assert_eq!(o.apns_team.as_deref(), Some("T")); | ||
let o = args(&["--apns-key", "", "--apns-key-id", "", "--apns-team", ""]).unwrap(); | ||
assert_eq!((o.apns_key, o.apns_key_id, o.apns_team), (None, None, None)); | ||
let o = args(&["--oauth-url", "https://gateway.example"]).unwrap(); | ||
assert_eq!(o.oauth_url.as_deref(), Some("https://gateway.example")); | ||
} | ||
crates/knotless-serve/Cargo.toml+1−0
| @@ -8,6 +8,7 @@ rust-version.workspace = true | ||
| ||
[dependencies] | ||
reqwest.workspace = true | ||
p256.workspace = true | ||
knotless.workspace = true | ||
knotless-mcp = { path = "../knotless-mcp" } | ||
anyhow.workspace = true | ||
deploy/deploy.sh+15−0
| @@ -223,6 +223,16 @@ if [ -n "$REVIEW_REPOS" ]; then | ||
say "not starting the reviewer: it needs KNOTLESS_REVIEW_IDENTIFIER, KNOTLESS_REVIEW_PASSWORD and CLAUDE_CODE_OAUTH_TOKEN or ANTHROPIC_API_KEY once" | ||
fi | ||
fi | ||
# Apple's push key, once it is on the host: without it, no notification is | ||
# sent, rather than the server refusing to start for a file not there. | ||
APNS_KEY="" | ||
if [ -n "${KNOTLESS_APNS_KEY_ID:-}" ] && [ -n "${KNOTLESS_APNS_TEAM:-}" ]; then | ||
if $SSH "[ -f '$DATA/apns.p8' ]"; then | ||
APNS_KEY=/data/apns.p8 | ||
else | ||
say "sending no push notifications: $DATA/apns.p8 is not on the host" | ||
fi | ||
fi | ||
{ | ||
echo "KNOTLESS_DOMAIN=$DOMAIN" | ||
echo "KNOTLESS_PROXY_NETWORK=$NETWORK" | ||
| @@ -239,6 +249,11 @@ fi | ||
[ -z "${KNOTLESS_BOT_DIDS:-}" ] || echo "KNOTLESS_BOT_DIDS=$KNOTLESS_BOT_DIDS" | ||
[ -z "${KNOTLESS_LOG:-}" ] || echo "KNOTLESS_LOG=$KNOTLESS_LOG" | ||
[ -z "${KNOTLESS_PAGES_DOMAIN:-}" ] || echo "KNOTLESS_PAGES_DOMAIN=$KNOTLESS_PAGES_DOMAIN" | ||
if [ -n "$APNS_KEY" ]; then | ||
echo "KNOTLESS_APNS_KEY=$APNS_KEY" | ||
echo "KNOTLESS_APNS_KEY_ID=$KNOTLESS_APNS_KEY_ID" | ||
echo "KNOTLESS_APNS_TEAM=$KNOTLESS_APNS_TEAM" | ||
fi | ||
[ -z "$PROFILES" ] || echo "COMPOSE_PROFILES=$PROFILES" | ||
if [ -n "$TRIAGE_REPOS" ]; then | ||
echo "KNOTLESS_TRIAGE_REPOS=$TRIAGE_REPOS" | ||