isaaccorbrey.com / mnemosyne
A Rust reimplementation of the Tangled knotserver.
0
Fork 0

Configure Feed

Select the types of activity you want to include in your feed.

xrpc: Add sh.tangled.knot.listKeys + public_keys table

Knot-scoped endpoint listing the SSH public keys registered for
push authentication. Currently read-only — keys arrive into the
table via AT Proto firehose ingest of sh.tangled.publicKey records,
which isn't built yet. Operators can pre-seed manually; the SSH
push handler that depends on this comes next.

Schema (migration 0006):

- public_keys (did TEXT, key TEXT, created_at TEXT, PK (did, key))
- index on (created_at DESC, did, key) for the listing query

created_at is TEXT rather than TIMESTAMPTZ because we emit it on
the wire as the same RFC3339 string the record provided —
roundtripping through a typed timestamp would introduce
formatting drift between intake and emission. Lexical sort of
consistent RFC3339 strings is chronologically correct, and the
AT Proto record schema mandates RFC3339, so this is safe.

(did, key) primary key — a single DID can register multiple keys
(different machines, etc.); a specific (DID, key) pair should
land at most once.

Wire format follows the Go knotserver:

- cursor is a string-encoded integer offset (same lexicon
divergence as repo.log — the lexicon allows any cursor format,
every existing client speaks Go's offset convention)
- order is created_at DESC with (did, key) tiebreak
- next cursor returned only if there are more rows (over-fetch
by 1 to detect)
- output object omits 'name' (the AT Proto record has it but the
listKeys lexicon doesn't expose it)
- InternalServerError on DB failure; InvalidRequest on negative
cursor

Tests need a small unusual piece of infrastructure: a
serializing mutex (LIST_KEYS_LOCK). Unlike every other XRPC
fixture in this file, public_keys isn't repo-scoped, so parallel
tests would step on each other's seeded rows. Five tests
serialize through the mutex; the non-row-touching invalid-cursor
test doesn't. Performance-wise the cost is negligible against
the testcontainer startup tax.

5 integration tests cover: created_at DESC ordering, cursor
pagination across two pages, empty-table returning an empty
array (no cursor), negative-cursor rejection, and lexicon wire
shape (no 'name' field; createdAt key spelling is camelCase per
lexicon).

author
Isaac Corbrey date (May 15, 2026, 10:11 AM -0400) commit b12f491a parent cb0a9260 change-id nnoxnvzz
+312
+32
mnemosyne_postgres/migrations/0006_create_public_keys.sql
··· 1 + -- Public keys accepted for SSH-based git push authentication. 2 + -- 3 + -- Rows arrive via the AT Proto firehose ingest pipeline (subscribing 4 + -- to `sh.tangled.publicKey` records on user PDSs). Until that pipeline 5 + -- ships, the table is populated only by tests and manual operator 6 + -- inserts. The SSH push handler will look up `(key, did)` here to 7 + -- verify push attempts; until SSH itself lands the table is read-only 8 + -- via `sh.tangled.knot.listKeys`. 9 + -- 10 + -- created_at is stored as a TEXT RFC3339 timestamp rather than 11 + -- TIMESTAMPTZ because (a) we emit it on the wire as the same string 12 + -- the record provided, so a roundtrip through TIMESTAMPTZ + reformat 13 + -- would introduce formatting drift, and (b) consistent RFC3339 14 + -- strings sort lexically the same as their underlying datetimes for 15 + -- ordering purposes. Assumes upstream sources emit consistent 16 + -- RFC3339; the AT Proto record schema mandates it. 17 + -- 18 + -- (did, key) is the primary key — a single user can register multiple 19 + -- keys (different machines, etc.), but a specific (user, key) pair 20 + -- should land at most once. 21 + CREATE TABLE public_keys ( 22 + did TEXT NOT NULL, 23 + key TEXT NOT NULL, 24 + created_at TEXT NOT NULL, 25 + PRIMARY KEY (did, key) 26 + ); 27 + 28 + -- Listing index for `listKeys`: order by created_at DESC, with 29 + -- (did, key) as the tiebreaker for stable ordering across rows that 30 + -- share a timestamp (rare in practice but cheap insurance). 31 + CREATE INDEX public_keys_listing_idx 32 + ON public_keys (created_at DESC, did, key);
+41
mnemosyne_postgres/src/lib.rs
··· 200 200 Ok(row.map(|(payload,)| payload.0)) 201 201 } 202 202 203 + /// A public-key row as emitted by `sh.tangled.knot.listKeys`. Fields 204 + /// match the lexicon's `publicKey` ref shape so the handler can map 205 + /// directly to the wire output. 206 + #[derive(Debug, Clone)] 207 + pub struct PublicKey { 208 + pub did: String, 209 + pub key: String, 210 + /// RFC3339 timestamp string. Stored as TEXT (see 211 + /// `0006_create_public_keys.sql` for rationale). 212 + pub created_at: String, 213 + } 214 + 215 + /// Page `limit` rows of `public_keys` starting at byte offset 216 + /// `offset`, ordered by created_at DESC + (did, key) tiebreak. The 217 + /// handler over-fetches by 1 (passes `limit + 1`) to determine whether 218 + /// there's a next page; this function doesn't trim — it just returns 219 + /// whatever rows the LIMIT selected. 220 + pub async fn list_public_keys( 221 + pool: &PgPool, 222 + limit: i64, 223 + offset: i64, 224 + ) -> Result<Vec<PublicKey>, PgError> { 225 + let rows: Vec<(String, String, String)> = sqlx::query_as( 226 + "SELECT did, key, created_at FROM public_keys \ 227 + ORDER BY created_at DESC, did ASC, key ASC \ 228 + LIMIT $1 OFFSET $2", 229 + ) 230 + .bind(limit) 231 + .bind(offset) 232 + .fetch_all(pool) 233 + .await?; 234 + Ok(rows 235 + .into_iter() 236 + .map(|(did, key, created_at)| PublicKey { 237 + did, 238 + key, 239 + created_at, 240 + }) 241 + .collect()) 242 + } 243 + 203 244 /// Write the language-stats payload for a tree. First writer wins: 204 245 /// concurrent computes on the same uncached tree may both run, but 205 246 /// they're idempotent (same input → same output), and the loser's
+148
mnemosyne_tests/tests/xrpc_endpoints.rs
··· 385 385 assert_eq!(body["error"], "InvalidRequest"); 386 386 }); 387 387 } 388 + 389 + // --------------------------------------------------------------------------- 390 + // sh.tangled.knot.listKeys 391 + // --------------------------------------------------------------------------- 392 + 393 + // Unlike every other test fixture in this file, listKeys reads from a 394 + // table that isn't repo-scoped (`public_keys` has no repo_did column). 395 + // Tests serialize via this mutex so parallel runs don't trip over each 396 + // other's seeded rows. The cost is roughly 5 sequential tests instead 397 + // of parallel; not material against the testcontainer startup cost. 398 + static LIST_KEYS_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); 399 + 400 + async fn seed_keys(pool: &PgPool, dids_keys_times: &[(&str, &str, &str)]) { 401 + for (did, key, created_at) in dids_keys_times { 402 + sqlx::query( 403 + "INSERT INTO public_keys (did, key, created_at) VALUES ($1, $2, $3) \ 404 + ON CONFLICT (did, key) DO NOTHING", 405 + ) 406 + .bind(did) 407 + .bind(key) 408 + .bind(created_at) 409 + .execute(pool) 410 + .await 411 + .expect("seed public_keys"); 412 + } 413 + } 414 + 415 + async fn clear_keys(pool: &PgPool) { 416 + sqlx::query("DELETE FROM public_keys") 417 + .execute(pool) 418 + .await 419 + .expect("clear public_keys"); 420 + } 421 + 422 + #[test] 423 + fn list_keys_orders_by_created_at_desc() { 424 + let _guard = LIST_KEYS_LOCK.lock().unwrap(); 425 + run(async { 426 + let pool = common::shared_pool().await; 427 + clear_keys(&pool).await; 428 + seed_keys( 429 + &pool, 430 + &[ 431 + ("did:test:alice", "ssh-ed25519 AAAA...alice", "2026-05-10T12:00:00Z"), 432 + ("did:test:bob", "ssh-ed25519 AAAA...bob", "2026-05-12T12:00:00Z"), 433 + ("did:test:carol", "ssh-ed25519 AAAA...carol", "2026-05-11T12:00:00Z"), 434 + ], 435 + ) 436 + .await; 437 + 438 + let (status, body) = serve(state(pool.clone(), None), "/sh.tangled.knot.listKeys").await; 439 + assert_eq!(status, StatusCode::OK, "body={body}"); 440 + let keys = body["keys"].as_array().expect("keys array"); 441 + assert_eq!(keys.len(), 3); 442 + // bob (12th), carol (11th), alice (10th). 443 + assert_eq!(keys[0]["did"], "did:test:bob"); 444 + assert_eq!(keys[1]["did"], "did:test:carol"); 445 + assert_eq!(keys[2]["did"], "did:test:alice"); 446 + assert!(body.get("cursor").is_none(), "no next page expected"); 447 + 448 + clear_keys(&pool).await; 449 + }); 450 + } 451 + 452 + #[test] 453 + fn list_keys_paginates_via_cursor() { 454 + let _guard = LIST_KEYS_LOCK.lock().unwrap(); 455 + run(async { 456 + let pool = common::shared_pool().await; 457 + clear_keys(&pool).await; 458 + seed_keys( 459 + &pool, 460 + &[ 461 + ("did:test:a", "ssh-rsa AAAA...a", "2026-05-10T12:00:00Z"), 462 + ("did:test:b", "ssh-rsa AAAA...b", "2026-05-11T12:00:00Z"), 463 + ("did:test:c", "ssh-rsa AAAA...c", "2026-05-12T12:00:00Z"), 464 + ("did:test:d", "ssh-rsa AAAA...d", "2026-05-13T12:00:00Z"), 465 + ], 466 + ) 467 + .await; 468 + 469 + let (_, page1) = serve(state(pool.clone(), None), "/sh.tangled.knot.listKeys?limit=2").await; 470 + let keys1 = page1["keys"].as_array().unwrap(); 471 + assert_eq!(keys1.len(), 2); 472 + assert_eq!(keys1[0]["did"], "did:test:d"); 473 + assert_eq!(keys1[1]["did"], "did:test:c"); 474 + let next = page1["cursor"].as_str().expect("cursor present"); 475 + 476 + let uri2 = format!("/sh.tangled.knot.listKeys?limit=2&cursor={next}"); 477 + let (_, page2) = serve(state(pool.clone(), None), &uri2).await; 478 + let keys2 = page2["keys"].as_array().unwrap(); 479 + assert_eq!(keys2.len(), 2); 480 + assert_eq!(keys2[0]["did"], "did:test:b"); 481 + assert_eq!(keys2[1]["did"], "did:test:a"); 482 + // No more pages. 483 + assert!(page2.get("cursor").is_none(), "page 2 should be the last"); 484 + 485 + clear_keys(&pool).await; 486 + }); 487 + } 488 + 489 + #[test] 490 + fn list_keys_empty_table_returns_empty_array() { 491 + let _guard = LIST_KEYS_LOCK.lock().unwrap(); 492 + run(async { 493 + let pool = common::shared_pool().await; 494 + clear_keys(&pool).await; 495 + let (status, body) = serve(state(pool, None), "/sh.tangled.knot.listKeys").await; 496 + assert_eq!(status, StatusCode::OK); 497 + assert_eq!(body["keys"].as_array().map(|a| a.len()), Some(0)); 498 + assert!(body.get("cursor").is_none()); 499 + }); 500 + } 501 + 502 + #[test] 503 + fn list_keys_rejects_negative_cursor() { 504 + // No lock needed: this test never touches table contents. 505 + run(async { 506 + let pool = common::shared_pool().await; 507 + let (status, body) = serve(state(pool, None), "/sh.tangled.knot.listKeys?cursor=-1").await; 508 + assert_eq!(status, StatusCode::BAD_REQUEST, "body={body}"); 509 + assert_eq!(body["error"], "InvalidRequest"); 510 + }); 511 + } 512 + 513 + #[test] 514 + fn list_keys_emits_lexicon_wire_shape() { 515 + let _guard = LIST_KEYS_LOCK.lock().unwrap(); 516 + run(async { 517 + let pool = common::shared_pool().await; 518 + clear_keys(&pool).await; 519 + seed_keys( 520 + &pool, 521 + &[("did:test:lex", "ssh-ed25519 AAAA...lex", "2026-05-14T15:30:00Z")], 522 + ) 523 + .await; 524 + 525 + let (_, body) = serve(state(pool.clone(), None), "/sh.tangled.knot.listKeys").await; 526 + let key = &body["keys"][0]; 527 + assert_eq!(key["did"], "did:test:lex"); 528 + assert_eq!(key["key"], "ssh-ed25519 AAAA...lex"); 529 + assert_eq!(key["createdAt"], "2026-05-14T15:30:00Z"); 530 + // Per lexicon, no `name` field on the listing shape. 531 + assert!(key.get("name").is_none(), "name should not appear"); 532 + 533 + clear_keys(&pool).await; 534 + }); 535 + }
+89
mnemosyne_xrpc/src/knot/list_keys.rs
··· 1 + //! `sh.tangled.knot.listKeys` — list public keys registered on this 2 + //! knot for SSH push authentication. 3 + //! 4 + //! Cursor is an integer-offset-encoded-as-string per the Go reference's 5 + //! convention; the lexicon allows any string format but every existing 6 + //! client speaks Go's. Same divergence from the lexicon's loose 7 + //! "cursor" type as `repo.log`. 8 + 9 + use axum::extract::{Query, State}; 10 + use axum::http::StatusCode; 11 + use axum::Json; 12 + use mnemosyne_postgres::list_public_keys; 13 + use serde::{Deserialize, Serialize}; 14 + 15 + use crate::router::XrpcState; 16 + use crate::XrpcError; 17 + 18 + const MAX_LIMIT: i64 = 1000; 19 + const DEFAULT_LIMIT: i64 = 100; 20 + 21 + #[derive(Debug, Deserialize)] 22 + pub struct Params { 23 + #[serde(default)] 24 + pub limit: Option<i64>, 25 + #[serde(default)] 26 + pub cursor: Option<String>, 27 + } 28 + 29 + #[derive(Debug, Serialize)] 30 + pub struct Output { 31 + pub keys: Vec<PublicKey>, 32 + #[serde(skip_serializing_if = "Option::is_none")] 33 + pub cursor: Option<String>, 34 + } 35 + 36 + #[derive(Debug, Serialize)] 37 + pub struct PublicKey { 38 + pub did: String, 39 + pub key: String, 40 + #[serde(rename = "createdAt")] 41 + pub created_at: String, 42 + } 43 + 44 + pub async fn handler( 45 + State(state): State<XrpcState>, 46 + Query(params): Query<Params>, 47 + ) -> Result<Json<Output>, XrpcError> { 48 + let limit = params 49 + .limit 50 + .map(|l| l.clamp(1, MAX_LIMIT)) 51 + .unwrap_or(DEFAULT_LIMIT); 52 + 53 + let offset = match params.cursor.as_deref() { 54 + Some(s) if !s.is_empty() => s.parse::<i64>().ok().filter(|o| *o >= 0).ok_or_else(|| { 55 + XrpcError::invalid_request("cursor must be a non-negative integer offset") 56 + })?, 57 + _ => 0, 58 + }; 59 + 60 + // Over-fetch by one so we can tell whether a next page exists. 61 + let mut rows = list_public_keys(&state.pool, limit + 1, offset) 62 + .await 63 + .map_err(|e| { 64 + XrpcError::new( 65 + "InternalServerError", 66 + format!("failed to retrieve public keys: {e}"), 67 + StatusCode::INTERNAL_SERVER_ERROR, 68 + ) 69 + })?; 70 + 71 + let next_cursor = if rows.len() as i64 > limit { 72 + rows.truncate(limit as usize); 73 + Some((offset + limit).to_string()) 74 + } else { 75 + None 76 + }; 77 + 78 + Ok(Json(Output { 79 + keys: rows 80 + .into_iter() 81 + .map(|k| PublicKey { 82 + did: k.did, 83 + key: k.key, 84 + created_at: k.created_at, 85 + }) 86 + .collect(), 87 + cursor: next_cursor, 88 + })) 89 + }
+1
mnemosyne_xrpc/src/knot/mod.rs
··· 1 + pub mod list_keys; 1 2 pub mod version;
+1
mnemosyne_xrpc/src/router.rs
··· 25 25 26 26 pub fn router(state: XrpcState) -> Router { 27 27 Router::new() 28 + .route("/sh.tangled.knot.listKeys", get(knot::list_keys::handler)) 28 29 .route("/sh.tangled.knot.version", get(knot::version::handler)) 29 30 .route("/sh.tangled.owner", get(owner::handler)) 30 31 .route("/sh.tangled.repo.branch", get(repo::branch::handler))