johnandersen777.bsky.social/knotless
git that lives in your Atmosphere account
Forked from chadtmiller.com/knotless
0
0
//! An appview fed by Jetstream: a stand-in server speaks subscribeEvents
//! frames, and the index fills from the records they carry without asking
//! any PDS.
use futures::SinkExt;
use knotless_serve::App;
use knotless_serve::index::Index;
use std::sync::atomic::Ordering;
use std::time::Duration;
use tokio_tungstenite::tungstenite::Message;
use tokio_tungstenite::tungstenite::handshake::server::{Request, Response};
const KIND: &str = "network.bsky.jetstream.subscribeEvents#";
const TIME: &str = "2026-10-20T00:00:00.000001Z";
const REPO: &str = "at://did:plc:owner/dev.knotless.repo/proj";
const KEY: &str = "3l3qo2vuowo2a";
fn message(payload: serde_json::Value) -> Message {
Message::Text(
serde_json::json!({"$type": "message", "payload": payload})
.to_string()
.into(),
)
}
fn commit(
seq: i64,
did: &str,
op: &str,
collection: &str,
rkey: &str,
record: Option<serde_json::Value>,
) -> Message {
let mut payload = serde_json::json!({
"$type": format!("{KIND}commit"), "seq": seq, "did": did, "time": TIME,
"rev": "3l3qo2vutsw2b", "operation": op, "collection": collection, "rkey": rkey,
});
if let Some(record) = record {
payload["record"] = record;
payload["cid"] = "bafyreia4w4bbvc5lqzkvexqlbwbagz5yfjfzzrqhfwqkhkk2xgekh4c6ai".into();
}
message(payload)
}
fn wait_until(mut f: impl FnMut() -> bool) {
for _ in 0..200 {
if f() {
return;
}
std::thread::sleep(Duration::from_millis(50));
}
panic!("condition never held");
}
// The handshake callback's error type is tungstenite's, not ours.
#[allow(clippy::result_large_err)]
#[test]
fn the_index_fills_from_jetstream() {
let dir = tempfile::tempdir().unwrap();
let rt = tokio::runtime::Runtime::new().unwrap();
let listener = rt
.block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
.unwrap();
let addr = listener.local_addr().unwrap();
let (asked, mut asks) = tokio::sync::mpsc::unbounded_channel();
let (go, mut gone) = tokio::sync::mpsc::unbounded_channel::<()>();
rt.spawn(async move {
for connection in 0..=u32::MAX {
let (tcp, _) = listener.accept().await.unwrap();
let asked = asked.clone();
let mut ws = tokio_tungstenite::accept_hdr_async(tcp, |req: &Request, mut res: Response| {
let protocol = req.headers()["sec-websocket-protocol"].clone();
asked.send((req.uri().to_string(), protocol.to_str().unwrap().to_string())).unwrap();
res.headers_mut().insert("sec-websocket-protocol", protocol);
Ok(res)
})
.await
.unwrap();
let frames = match connection {
0 => vec![
message(serde_json::json!({"$type": format!("{KIND}info"), "name": "OutdatedCursor", "message": "starting at seq 10"})),
Message::Text("not json".into()),
Message::Binary(vec![1].into()),
commit(10, "did:plc:owner", "create", "dev.knotless.repo", "proj", Some(serde_json::json!({
"$type": "dev.knotless.repo", "key": KEY, "description": "a project", "createdAt": TIME,
}))),
commit(11, "did:plc:other", "create", "dev.knotless.issue", "3l3qo2vuowo2b", Some(serde_json::json!({
"$type": "dev.knotless.issue", "repo": {"uri": REPO, "key": KEY}, "title": "It breaks", "createdAt": TIME,
}))),
commit(12, "did:plc:other", "create", "app.bsky.feed.post", "3l3qo2vuowo2c", Some(serde_json::json!({
"$type": "app.bsky.feed.post", "text": "elsewhere", "createdAt": TIME,
}))),
message(serde_json::json!({
"$type": format!("{KIND}identity"), "seq": 13, "did": "did:plc:moved", "time": TIME,
"identity": {"did": "did:plc:moved", "handle": "moved.test", "seq": 99, "time": TIME},
})),
Message::Text(r#"{"$type":"error","error":"ConsumerTooSlow"}"#.into()),
],
// Its author edits it, keeping what the index does not know.
1 => vec![commit(14, "did:plc:other", "update", "dev.knotless.issue", "3l3qo2vuowo2b", Some(serde_json::json!({
"$type": "dev.knotless.issue", "repo": {"uri": REPO, "key": KEY}, "title": "It breaks badly",
"dev.example.labels": ["bug"], "createdAt": TIME, "updatedAt": "2026-10-21T00:00:00.000Z",
})))],
// Once the test has seen the edit.
2 => {
gone.recv().await;
vec![commit(15, "did:plc:other", "delete", "dev.knotless.issue", "3l3qo2vuowo2b", None)]
}
_ => vec![],
};
for frame in frames {
ws.send(frame).await.unwrap();
}
ws.close(None).await.unwrap();
}
});
let env = knotless::env::Env::from_vars(|_| None);
let index = Index::open(&dir.path().join("index.db")).unwrap();
let app = App::build(
knotless::env::Ctx::new(env),
Duration::ZERO,
None,
Some(index),
);
rt.spawn(knotless_serve::jetstream::watch(
app.clone(),
format!("ws://{addr}"),
));
let index = app.index.clone().unwrap();
// It asks for knotless commits, blocks, and every account's identity, offering
// the one framing Jetstream speaks.
let (first, protocol) = rt.block_on(asks.recv()).unwrap();
assert_eq!(protocol, "xrpc.v1.json");
assert!(
first.starts_with(
"/xrpc/network.bsky.jetstream.subscribeEvents?collections=dev.knotless.*&kinds=commit"
),
"{first}"
);
assert!(!first.contains("cursor"), "{first}");
// The records it carried are indexed as they came.
wait_until(|| index.forge("did:plc:owner", KEY).unwrap().len() == 1);
assert_eq!(
index.repo(REPO).unwrap().unwrap()["description"],
"a project"
);
assert_eq!(
index.forge("did:plc:owner", KEY).unwrap()[0].value["title"],
"It breaks"
);
// The identity came after the issue, so it can still be on its way.
wait_until(|| app.events_for("did:plc:moved") == 1);
assert_eq!(app.events_for("did:plc:owner"), 1);
assert_eq!(app.events_for("did:plc:other"), 0);
// After a drop it resumes past the last event it saw. An update
// replaces what the index had, whole.
let (second, _) = rt.block_on(asks.recv()).unwrap();
assert!(second.ends_with("&cursor=14"), "{second}");
wait_until(|| {
index.forge("did:plc:owner", KEY).unwrap()[0].value["title"] == "It breaks badly"
});
let edited = index.forge("did:plc:owner", KEY).unwrap().remove(0).value;
assert_eq!(edited["updatedAt"], "2026-10-21T00:00:00.000Z");
assert_eq!(edited["dev.example.labels"][0], "bug");
let (third, _) = rt.block_on(asks.recv()).unwrap();
assert!(third.ends_with("&cursor=15"), "{third}");
go.send(()).unwrap();
wait_until(|| index.forge("did:plc:owner", KEY).unwrap().is_empty());
let (fourth, _) = rt.block_on(asks.recv()).unwrap();
assert!(fourth.ends_with("&cursor=16"), "{fourth}");
// The index keeps where it was fed to, for a restart to resume from.
assert_eq!(index.stream_seq(&format!("ws://{addr}")).unwrap(), Some(16));
// A Jetstream it cannot reach, or cannot even name, leaves the TTL in
// charge.
for url in ["ws://127.0.0.1:9", "not a url"] {
let lost = App::new(
knotless::env::Ctx::new(knotless::env::Env::from_vars(|_| None)),
Duration::ZERO,
);
lost.watching.store(true, Ordering::Relaxed);
rt.spawn(knotless_serve::jetstream::watch(lost.clone(), url.into()));
wait_until(|| !lost.watching.load(Ordering::Relaxed));
}
}
// The handshake callback's error type is tungstenite's, not ours.
#[allow(clippy::result_large_err)]
#[test]
fn a_silent_jetstream_is_left() {
let dir = tempfile::tempdir().unwrap();
let rt = tokio::runtime::Runtime::new().unwrap();
let listener = rt
.block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
.unwrap();
let addr = listener.local_addr().unwrap();
let (asked, mut asks) = tokio::sync::mpsc::unbounded_channel();
rt.spawn(async move {
let mut held = Vec::new();
loop {
let (tcp, _) = listener.accept().await.unwrap();
let asked = asked.clone();
let mut ws =
tokio_tungstenite::accept_hdr_async(tcp, |req: &Request, mut res: Response| {
asked.send(req.uri().to_string()).unwrap();
let protocol = req.headers()["sec-websocket-protocol"].clone();
res.headers_mut().insert("sec-websocket-protocol", protocol);
Ok(res)
})
.await
.unwrap();
ws.send(commit(
7,
"did:plc:owner",
"delete",
"dev.knotless.issue",
"3l3qo2vuowo2b",
None,
))
.await
.unwrap();
// Never read again, so its pings go unanswered: a connection
// gone half-open.
held.push(ws);
}
});
// Restarted, it starts where the index was last fed to.
let index = Index::open(&dir.path().join("index.db")).unwrap();
index.set_stream_seq(&format!("ws://{addr}"), 5).unwrap();
index.set_stream_seq(&format!("ws://{addr}"), 6).unwrap();
let app = App::build(
knotless::env::Ctx::new(knotless::env::Env::from_vars(|_| None)),
Duration::ZERO,
None,
Some(index),
);
rt.spawn(knotless_serve::jetstream::follow(
app.clone(),
format!("ws://{addr}"),
Duration::from_millis(100),
));
let first = rt.block_on(asks.recv()).unwrap();
assert!(first.ends_with("&cursor=6"), "{first}");
// Silent through a ping, it is given up on and followed again from
// past what it said.
let second = rt.block_on(asks.recv()).unwrap();
assert!(second.ends_with("&cursor=8"), "{second}");
assert_eq!(
app.index
.as_ref()
.unwrap()
.stream_seq(&format!("ws://{addr}"))
.unwrap(),
Some(8)
);
}
/// A stand-in Jetstream that sends `frames` once to the first client.
// The handshake callback's error type is tungstenite's, not ours.
#[allow(clippy::result_large_err)]
fn sends(rt: &tokio::runtime::Runtime, frames: Vec<Message>) -> String {
let listener = rt
.block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
.unwrap();
let url = format!("ws://{}", listener.local_addr().unwrap());
rt.spawn(async move {
let (tcp, _) = listener.accept().await.unwrap();
let mut ws =
tokio_tungstenite::accept_hdr_async(tcp, |req: &Request, mut res: Response| {
let protocol = req.headers()["sec-websocket-protocol"].clone();
res.headers_mut().insert("sec-websocket-protocol", protocol);
Ok(res)
})
.await
.unwrap();
for frame in frames {
ws.send(frame).await.unwrap();
}
// Held open, so the server does not ask again.
std::future::pending::<()>().await;
});
url
}
#[test]
fn checks_by_the_runners_a_repository_names() {
let dir = tempfile::tempdir().unwrap();
let rt = tokio::runtime::Runtime::new().unwrap();
let check = |seq: i64, runner: &str, rkey: &str, r: &str, sha: &str, at: &str| {
commit(
seq,
runner,
"create",
"dev.knotless.check",
rkey,
Some(serde_json::json!({
"$type": "dev.knotless.check", "repo": {"uri": REPO, "key": KEY}, "ref": r,
"commit": sha, "workflow": "ci", "status": "success", "startedAt": at,
})),
)
};
let main = "refs/heads/main";
let (a, b) = ("a".repeat(40), "b".repeat(40));
let url = sends(
&rt,
vec![
commit(
1,
"did:plc:owner",
"create",
"dev.knotless.repo",
"proj",
Some(serde_json::json!({
"$type": "dev.knotless.repo", "key": KEY, "name": "proj", "createdAt": TIME,
"runners": [{"did": "did:plc:ci"}],
})),
),
check(2, "did:plc:ci", "c1", main, &a, "2026-10-20T00:00:01.000Z"),
check(
3,
"did:plc:ci",
"c2",
"refs/heads/x",
&b,
"2026-10-20T00:00:02.000Z",
),
// Anyone can write a check; the repository does not name this one.
check(
4,
"did:plc:stranger",
"c3",
main,
&a,
"2026-10-20T00:00:03.000Z",
),
],
);
let index = Index::open(&dir.path().join("index.db")).unwrap();
let app = App::build(
knotless::env::Ctx::new(knotless::env::Env::from_vars(|_| None)),
Duration::ZERO,
None,
Some(index),
);
rt.spawn(knotless_serve::jetstream::watch(app.clone(), url));
let listener = rt
.block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
.unwrap();
let base = format!("http://{}", listener.local_addr().unwrap());
let router = knotless_serve::router(app);
rt.spawn(async move { axum::serve(listener, router).await.unwrap() });
let list = |query: &str| -> serde_json::Value {
reqwest::blocking::get(format!(
"{base}/xrpc/dev.knotless.listChecks?repo={REPO}&key={KEY}{query}"
))
.unwrap()
.json()
.unwrap()
};
wait_until(|| list("")["checks"].as_array().unwrap().len() == 2);
let all = list("");
assert_eq!(
all["checks"][0]["uri"],
"at://did:plc:ci/dev.knotless.check/c2"
);
assert_eq!(all["checks"][0]["runner"], "did:plc:ci");
assert_eq!(all["checks"][1]["value"]["commit"], a.as_str());
assert!(all.get("cursor").is_none());
assert_eq!(
list(&format!("&ref={main}"))["checks"]
.as_array()
.unwrap()
.len(),
1
);
assert_eq!(
list(&format!("&commit={b}"))["checks"][0]["value"]["ref"],
"refs/heads/x"
);
let first = list("&limit=1");
let cursor = first["cursor"].as_str().unwrap();
let second = list(&format!("&limit=1&cursor={}", cursor.replace(' ', "%20")));
assert_eq!(
second["checks"][0]["uri"],
"at://did:plc:ci/dev.knotless.check/c1"
);
// A repository the index has never seen names no runners.
let unknown = reqwest::blocking::get(format!(
"{base}/xrpc/dev.knotless.listChecks?repo=at://did:plc:nobody/dev.knotless.repo/x&key={KEY}"
))
.unwrap()
.json::<serde_json::Value>()
.unwrap();
assert_eq!(unknown["checks"], serde_json::json!([]));
let bad = reqwest::blocking::get(format!("{base}/xrpc/dev.knotless.listChecks?repo=x&key=k"))
.unwrap();
assert_eq!(bad.status().as_u16(), 400);
}