From 8b00c6301915bc45aec392411e13ba8ddc6b9f6b Mon Sep 17 00:00:00 2001
From: Nate Moore <git@natemoo.re>
Date: Fri, 6 Nov 2026 01:30:06 -0500
Subject: [PATCH 01/10] Arm the index alarm on startup and re-arm it after a
 failed pass

Nothing scheduled the first alarm, so the index never ran. The Durable
Object now arms one from its constructor, and alarm() re-arms in a
finally so a throwing pass doesn't stop ingestion once Cloudflare's
retries run out.
---
 packages/api/README.md           |  8 ++++--
 packages/api/src/alarm.ts        | 26 ++++++++++++++++++++
 packages/api/src/worker.ts       | 12 +++------
 packages/api/tests/alarm.test.ts | 42 ++++++++++++++++++++++++++++++++
 4 files changed, 77 insertions(+), 11 deletions(-)
 create mode 100644 packages/api/src/alarm.ts
 create mode 100644 packages/api/tests/alarm.test.ts

diff --git a/packages/api/README.md b/packages/api/README.md
index 18a4334..dbb43c3 100644
--- a/packages/api/README.md
+++ b/packages/api/README.md
@@ -50,8 +50,12 @@ Every 60 seconds, a Durable Object alarm (`Index.alarm`, `src/worker.ts`) runs o
 2. Open a live-tail WebSocket to Jetstream and drain it until an event's `time` is within 5
    seconds of now, or a 20-second budget runs out, applying events in order and saving the
    cursor after each one.
-3. The alarm reschedules itself for 60 seconds later. No outgoing WebSocket is held open between
-   passes, so the Durable Object isn't billed for an idle connection.
+3. The alarm reschedules itself for 60 seconds later, in a `finally`, so a pass that throws still
+   re-arms (`src/alarm.ts`). No outgoing WebSocket is held open between passes, so the Durable
+   Object isn't billed for an idle connection.
+
+The first alarm is armed from the Durable Object's constructor (`blockConcurrencyWhile`), so the
+first request that wakes the object starts ingestion.
 
 Every record is validated against its generated `@nibling/lexicons` schema at the boundary
 (`toPutEvent` in `src/index/jetstream.ts`, shared by the live tail and the backfill path); a
diff --git a/packages/api/src/alarm.ts b/packages/api/src/alarm.ts
new file mode 100644
index 0000000..7106122
--- /dev/null
+++ b/packages/api/src/alarm.ts
@@ -0,0 +1,26 @@
+/** The slice of `ctx.storage` the alarm lifecycle needs, so tests can pass a fake. */
+export interface AlarmStorage {
+  getAlarm(): Promise<number | null>;
+  setAlarm(scheduledTime: number): Promise<void>;
+}
+
+export const INGEST_INTERVAL_MS = 60_000;
+
+/** Arms the first alarm when none is set. The DO calls this from its constructor, so the
+ * first request that wakes the object starts the recurring ingestion. */
+export async function ensureAlarm(storage: AlarmStorage, now: number): Promise<void> {
+  if ((await storage.getAlarm()) === null) await storage.setAlarm(now);
+}
+
+/** Runs one pass and always re-arms. A throw is logged, not rethrown: Cloudflare's own retry
+ * gives up after six attempts, which would stop ingestion for good, while the next scheduled
+ * pass retries the same work a minute later. */
+export async function runAndRearm(storage: AlarmStorage, run: () => Promise<void>, now: () => number): Promise<void> {
+  try {
+    await run();
+  } catch (error) {
+    console.error('ingest pass failed', error);
+  } finally {
+    await storage.setAlarm(now() + INGEST_INTERVAL_MS);
+  }
+}
diff --git a/packages/api/src/worker.ts b/packages/api/src/worker.ts
index 99fd175..652b573 100644
--- a/packages/api/src/worker.ts
+++ b/packages/api/src/worker.ts
@@ -1,4 +1,5 @@
 import { DurableObject } from 'cloudflare:workers';
+import { ensureAlarm, runAndRearm } from './alarm.ts';
 import { ingest, type WebSocketLike } from './index/ingest.ts';
 import { IndexModel, type RecentOptions, type TrendingOptions } from './index/model.ts';
 import { handleRequest } from './router.ts';
@@ -32,6 +33,7 @@ export class Index extends DurableObject<Env> {
       exec: <T = Record<string, unknown>>(query: string, ...bindings: unknown[]) =>
         ctx.storage.sql.exec(query, ...bindings).toArray() as T[],
     });
+    void ctx.blockConcurrencyWhile(() => ensureAlarm(ctx.storage, Date.now()));
   }
 
   async getRecent(options: RecentOptions) {
@@ -52,15 +54,7 @@ export class Index extends DurableObject<Env> {
   }
 
   async alarm(): Promise<void> {
-    await this.run();
-    await this.ctx.storage.setAlarm(Date.now() + 60_000);
-  }
-
-  /** Exposed over RPC so `wrangler dev` (or a smoke request) can trigger one ingestion pass
-   * without waiting for the alarm, and so the first request schedules the recurring one. */
-  async ensureAlarmScheduled(): Promise<void> {
-    const existing = await this.ctx.storage.getAlarm();
-    if (existing === null) await this.ctx.storage.setAlarm(Date.now() + 1000);
+    await runAndRearm(this.ctx.storage, () => this.run(), Date.now);
   }
 
   async run(): Promise<void> {
diff --git a/packages/api/tests/alarm.test.ts b/packages/api/tests/alarm.test.ts
new file mode 100644
index 0000000..96862b6
--- /dev/null
+++ b/packages/api/tests/alarm.test.ts
@@ -0,0 +1,42 @@
+import assert from 'node:assert/strict';
+import { test } from 'node:test';
+import { ensureAlarm, runAndRearm, type AlarmStorage } from '../src/alarm.ts';
+
+function fakeStorage(initial: number | null): AlarmStorage & { alarm: number | null } {
+  return {
+    alarm: initial,
+    async getAlarm() {
+      return this.alarm;
+    },
+    async setAlarm(time: number) {
+      this.alarm = time;
+    },
+  };
+}
+
+test('ensureAlarm arms an alarm when none is set', async () => {
+  const storage = fakeStorage(null);
+  await ensureAlarm(storage, 1000);
+  assert.equal(storage.alarm, 1000);
+});
+
+test('ensureAlarm leaves an existing alarm alone', async () => {
+  const storage = fakeStorage(5000);
+  await ensureAlarm(storage, 1000);
+  assert.equal(storage.alarm, 5000);
+});
+
+test('runAndRearm re-arms 60s out after a successful pass', async () => {
+  const storage = fakeStorage(null);
+  await runAndRearm(storage, async () => {}, () => 1000);
+  assert.equal(storage.alarm, 61_000);
+});
+
+test('runAndRearm re-arms even when the pass throws', async (t) => {
+  t.mock.method(console, 'error', () => {});
+  const storage = fakeStorage(null);
+  await runAndRearm(storage, async () => {
+    throw new Error('relay down');
+  }, () => 1000);
+  assert.equal(storage.alarm, 61_000);
+});
-- 
knotmirror 0.0.1


From 799ffb06bccc5799f2b399621202b519e0e13624 Mon Sep 17 00:00:00 2001
From: Nate Moore <git@natemoo.re>
Date: Fri, 6 Nov 2026 01:30:44 -0500
Subject: [PATCH 02/10] Send Slingshot's at_uri parameter and reject malformed
 oEmbed urls

getRecordByUri wants at_uri; with uri it answers 400, so every link
card, embed, and oEmbed request was a 404. /oembed?url=x also threw
from new URL and returned 500 instead of 400. Router tests now drive
the card routes through a fake Slingshot.
---
 packages/api/src/router.ts        |  4 +-
 packages/api/tests/router.test.ts | 71 +++++++++++++++++++++++++++++++
 2 files changed, 73 insertions(+), 2 deletions(-)

diff --git a/packages/api/src/router.ts b/packages/api/src/router.ts
index 97f371b..74353d5 100644
--- a/packages/api/src/router.ts
+++ b/packages/api/src/router.ts
@@ -53,7 +53,7 @@ async function resolveMiniDoc(fetchFn: typeof fetch, slingshotUrl: string, actor
 /** `blue.microcosm.repo.getRecordByUri` via Slingshot. */
 async function getRecordByUri(fetchFn: typeof fetch, slingshotUrl: string, uri: string): Promise<{ cid: string; value: Record<string, unknown> } | undefined> {
   const url = new URL('/xrpc/blue.microcosm.repo.getRecordByUri', slingshotUrl);
-  url.searchParams.set('uri', uri);
+  url.searchParams.set('at_uri', uri);
   const res = await fetchFn(url.toString());
   if (!res.ok) return undefined;
   const body = (await res.json()) as { cid?: string; value?: Record<string, unknown> };
@@ -131,7 +131,7 @@ export async function handleRequest(request: Request, env: RouterEnv): Promise<R
 
   if (url.pathname === '/oembed') {
     const pageUrl = url.searchParams.get('url');
-    const match = pageUrl ? new URL(pageUrl).pathname.match(/^\/@([^/]+)\/([^/]+)$/) : null;
+    const match = pageUrl && URL.canParse(pageUrl) ? new URL(pageUrl).pathname.match(/^\/@([^/]+)\/([^/]+)$/) : null;
     if (!match) return xrpcError('InvalidRequest', 'url must point at a project page', 400);
     const [, actor, rkey] = match as unknown as [string, string, string];
     const card = await loadProjectCard(env, actor, rkey);
diff --git a/packages/api/tests/router.test.ts b/packages/api/tests/router.test.ts
index 601e369..89f2255 100644
--- a/packages/api/tests/router.test.ts
+++ b/packages/api/tests/router.test.ts
@@ -65,3 +65,74 @@ test('oauth client-metadata.json is served with the exact D19 scope string', asy
   );
 });
 
+
+const ALICE_DID = 'did:plc:alice';
+const PROJECT_URI = `at://${ALICE_DID}/run.nibling.project/3la1`;
+
+/** Slingshot as it answers today: `resolveMiniDoc?identifier=` and `getRecordByUri?at_uri=`
+ * (a request with `uri=` gets HTTP 400, as the live service does). */
+function fakeSlingshot(calls: string[]): typeof fetch {
+  return (async (input: string | URL | Request) => {
+    const url = new URL(typeof input === 'string' ? input : input.toString());
+    calls.push(url.pathname + url.search);
+    if (url.pathname === '/xrpc/blue.microcosm.identity.resolveMiniDoc') {
+      if (url.searchParams.get('identifier') !== 'alice.test') return new Response('{}', { status: 400 });
+      return Response.json({ did: ALICE_DID, handle: 'alice.test', pds: 'https://pds.example.com' });
+    }
+    if (url.pathname === '/xrpc/blue.microcosm.repo.getRecordByUri') {
+      if (url.searchParams.get('at_uri') !== PROJECT_URI) return new Response('{}', { status: 400 });
+      return Response.json({
+        uri: PROJECT_URI,
+        cid: 'bafy-project1',
+        value: { $type: 'run.nibling.project', name: 'Alice <Project>', files: [], createdAt: '2026-11-01T00:00:00.000Z' },
+      });
+    }
+    return new Response('{}', { status: 404 });
+  }) as typeof fetch;
+}
+
+function cardEnv(calls: string[]): RouterEnv {
+  return { ...fakeEnv(), fetch: fakeSlingshot(calls) };
+}
+
+test('/@actor/rkey reads the record through Slingshot with at_uri and serves its og:title', async () => {
+  const calls: string[] = [];
+  const res = await handleRequest(new Request('https://api.nibling.run/@alice.test/3la1'), cardEnv(calls));
+  assert.equal(res.status, 200);
+  const html = await res.text();
+  assert.equal(html.includes('<meta property="og:title" content="Alice &lt;Project&gt;">'), true);
+  assert.deepStrictEqual(calls, [
+    '/xrpc/blue.microcosm.identity.resolveMiniDoc?identifier=alice.test',
+    `/xrpc/blue.microcosm.repo.getRecordByUri?at_uri=${encodeURIComponent(PROJECT_URI)}`,
+  ]);
+});
+
+test('/@actor/rkey is a 404 when the record does not exist', async () => {
+  const res = await handleRequest(new Request('https://api.nibling.run/@alice.test/missing'), cardEnv([]));
+  assert.equal(res.status, 404);
+});
+
+test('/embed/actor/rkey serves the same og:title in the embed shell', async () => {
+  const res = await handleRequest(new Request('https://api.nibling.run/embed/alice.test/3la1'), cardEnv([]));
+  assert.equal(res.status, 200);
+  const html = await res.text();
+  assert.equal(html.includes('<meta property="og:title" content="Alice &lt;Project&gt;">'), true);
+  assert.equal(html.includes(`src="https://nibling.run/embed.html#${PROJECT_URI}"`), true);
+});
+
+test('/oembed returns rich JSON with the exact title', async () => {
+  const res = await handleRequest(
+    new Request(`https://api.nibling.run/oembed?url=${encodeURIComponent('https://nibling.run/@alice.test/3la1')}`),
+    cardEnv([]),
+  );
+  assert.equal(res.status, 200);
+  const body = (await res.json()) as { type: string; title: string };
+  assert.equal(body.type, 'rich');
+  assert.equal(body.title, 'Alice <Project>');
+});
+
+test('/oembed with a malformed url is a 400, not a 500', async () => {
+  const res = await handleRequest(new Request('https://api.nibling.run/oembed?url=x'), cardEnv([]));
+  assert.equal(res.status, 400);
+  assert.deepStrictEqual(await res.json(), { error: 'InvalidRequest', message: 'url must point at a project page' });
+});
-- 
knotmirror 0.0.1


From 81a4d4eecdfe64143670a4079723b27b25f98b36 Mon Sep 17 00:00:00 2001
From: Nate Moore <git@natemoo.re>
Date: Fri, 6 Nov 2026 01:31:11 -0500
Subject: [PATCH 03/10] Derive backfilled record identity from the repo that
 was listed

listRecords answers come from an untrusted PDS, and backfill indexed
the URI it returned, so one account's PDS could overwrite another
account's rows. Records now must name the listed DID and collection,
and the URI is rebuilt from them.
---
 packages/api/src/index/backfill.ts   | 16 +++++++++++++++-
 packages/api/tests/backfill.test.ts  | 27 +++++++++++++++++++++++++++
 packages/api/tests/jetstream.test.ts | 10 ++++++++++
 3 files changed, 52 insertions(+), 1 deletion(-)

diff --git a/packages/api/src/index/backfill.ts b/packages/api/src/index/backfill.ts
index 08a4d68..ee21053 100644
--- a/packages/api/src/index/backfill.ts
+++ b/packages/api/src/index/backfill.ts
@@ -35,6 +35,18 @@ async function* listReposByCollection(collection: string, deps: BackfillDeps): A
   } while (cursor);
 }
 
+/** The record's URI rebuilt from the repo we asked for, or `undefined` when the PDS claims a
+ * record in some other repo or collection. A PDS is untrusted: without this, one account's PDS
+ * could overwrite index rows that belong to another account. */
+export function ownedRecordUri(uri: unknown, did: string, collection: string): string | undefined {
+  if (typeof uri !== 'string' || !uri.startsWith('at://')) return undefined;
+  const parts = uri.slice('at://'.length).split('/');
+  if (parts.length !== 3) return undefined;
+  const [authority, recordCollection, rkey] = parts as [string, string, string];
+  if (authority !== did || recordCollection !== collection || rkey === '') return undefined;
+  return `at://${did}/${collection}/${rkey}`;
+}
+
 /** `com.atproto.repo.listRecords` against one repo's own PDS, paged to exhaustion. A repo whose
  * PDS can't be reached is skipped rather than failing the whole backfill. */
 async function* listRecords(did: string, collection: string, deps: BackfillDeps): AsyncGenerator<IndexEvent> {
@@ -60,7 +72,9 @@ async function* listRecords(did: string, collection: string, deps: BackfillDeps)
     if (!res.ok) return;
     const body = (await res.json()) as ListRecordsResponse;
     for (const record of body.records) {
-      const event = toPutEvent(collection, record.uri, record.cid, did, record.value);
+      const uri = ownedRecordUri(record.uri, did, collection);
+      if (!uri) continue;
+      const event = toPutEvent(collection, uri, record.cid, did, record.value);
       if (event) yield event;
     }
     cursor = body.cursor;
diff --git a/packages/api/tests/backfill.test.ts b/packages/api/tests/backfill.test.ts
index d87d055..3a8e60f 100644
--- a/packages/api/tests/backfill.test.ts
+++ b/packages/api/tests/backfill.test.ts
@@ -78,3 +78,30 @@ test('backfillCollection skips a repo whose PDS cannot be resolved', async () =>
   for await (const event of backfillCollection('run.nibling.project', deps)) events.push(event);
   assert.deepStrictEqual(events, []);
 });
+
+test('backfillCollection drops records whose URI names another repo or collection', async () => {
+  const deps: BackfillDeps = {
+    relayUrl: 'https://relay1.us-east.bsky.network',
+    fetch: (async (input: string | URL | Request) => {
+      const url = new URL(typeof input === 'string' ? input : input.toString());
+      if (url.pathname === '/xrpc/com.atproto.sync.listReposByCollection') {
+        return jsonResponse({ repos: [{ did: 'did:plc:evil' }] });
+      }
+      const value = { $type: 'run.nibling.project', name: 'Hijack', files: [], createdAt: '2026-11-01T00:00:00.000Z' };
+      return jsonResponse({
+        records: [
+          { uri: 'at://did:plc:victim/run.nibling.project/3abc', cid: 'bafy1', value },
+          { uri: 'at://did:plc:evil/run.nibling.feed.like/3abc', cid: 'bafy2', value },
+          { uri: 'at://did:plc:evil/run.nibling.project/3own', cid: 'bafy3', value },
+        ],
+      });
+    }) as typeof fetch,
+    resolvePds: async () => 'https://evil.pds.example',
+  };
+  const events = [];
+  for await (const event of backfillCollection('run.nibling.project', deps)) events.push(event);
+  assert.deepStrictEqual(
+    events.map((e) => ({ uri: e.uri, did: e.action === 'put' ? e.did : undefined })),
+    [{ uri: 'at://did:plc:evil/run.nibling.project/3own', did: 'did:plc:evil' }],
+  );
+});
diff --git a/packages/api/tests/jetstream.test.ts b/packages/api/tests/jetstream.test.ts
index 551f42f..92fc379 100644
--- a/packages/api/tests/jetstream.test.ts
+++ b/packages/api/tests/jetstream.test.ts
@@ -70,3 +70,13 @@ test('a delete carries no record and needs no validation', () => {
     uri: 'at://did:plc:alice/run.nibling.project/3la1',
   });
 });
+
+test('the URI and did come from the frame envelope, never the record body', () => {
+  const frame = commitFrame({
+    record: { $type: 'run.nibling.project', name: 'x', files: [], createdAt: '2026-11-06T06:47:44.859Z', did: 'did:plc:victim', uri: 'at://did:plc:victim/run.nibling.project/3la1' },
+  });
+  const event = parseJetstreamMessage(frame);
+  assert.ok(event && event.action === 'put');
+  assert.equal(event.uri, 'at://did:plc:alice/run.nibling.project/3la1');
+  assert.equal(event.did, 'did:plc:alice');
+});
-- 
knotmirror 0.0.1


From d8d40980d7626827ec832d3b53115dc3a32767fd Mon Sep 17 00:00:00 2001
From: Nate Moore <git@natemoo.re>
Date: Fri, 6 Nov 2026 01:35:01 -0500
Subject: [PATCH 04/10] Keep the index cursor moving and make backfill lossless

The cursor is now a Jetstream time_us. A quiet drain advances it to
when the socket opened, capped at 10 minutes past its start, so a quiet
network no longer re-backfills every minute. A backfill stores its
start time as the cursor and the live tail replays from it, so events
during the backfill are kept. A repo listed completely has its unlisted
rows deleted, which catches deletions made while the index was down.
---
 packages/api/README.md              |  20 ++-
 packages/api/src/index/backfill.ts  |  26 ++-
 packages/api/src/index/ingest.ts    | 131 ++++++++++------
 packages/api/src/index/jetstream.ts |  31 ++--
 packages/api/src/index/model.ts     |  85 ++++++----
 packages/api/src/worker.ts          |   4 +-
 packages/api/tests/backfill.test.ts |  34 ++--
 packages/api/tests/ingest.test.ts   | 235 ++++++++++++++++++++--------
 packages/api/tests/model.test.ts    |  22 ++-
 9 files changed, 383 insertions(+), 205 deletions(-)

diff --git a/packages/api/README.md b/packages/api/README.md
index dbb43c3..e092b7f 100644
--- a/packages/api/README.md
+++ b/packages/api/README.md
@@ -46,10 +46,16 @@ Every 60 seconds, a Durable Object alarm (`Index.alarm`, `src/worker.ts`) runs o
    backfill first: `com.atproto.sync.listReposByCollection` on the relay for each of the four
    collections, then `com.atproto.repo.listRecords` against each repo's own PDS
    (`src/index/backfill.ts`). A repo whose PDS can't be resolved or reached is skipped, not
-   fatal.
-2. Open a live-tail WebSocket to Jetstream and drain it until an event's `time` is within 5
-   seconds of now, or a 20-second budget runs out, applying events in order and saving the
-   cursor after each one.
+   fatal. A repo that was listed completely has its other rows in that collection deleted, which
+   catches deletions made while the index wasn't listening. The time the backfill started
+   becomes the cursor, so the live tail replays everything that happened during it.
+2. Open a live-tail WebSocket to Jetstream from the cursor and drain it until an event's `time`
+   is within 5 seconds of now, or a 20-second budget runs out, applying events in order and
+   saving the cursor after each one. The cursor is a Jetstream `time_us`. A drain that stayed
+   open and heard nothing for its last 5 seconds moves the cursor up to when the socket opened
+   (minus 5 seconds), but never more than 10 minutes past where it started: Jetstream sends
+   nothing on a quiet filtered stream, and its replay scans about 12 minutes of network time per
+   second, so a silent drain has only proven that much.
 3. The alarm reschedules itself for 60 seconds later, in a `finally`, so a pass that throws still
    re-arms (`src/alarm.ts`). No outgoing WebSocket is held open between passes, so the Durable
    Object isn't billed for an idle connection.
@@ -72,9 +78,9 @@ The draft spec (`api-spec.md`) assumed Jetstream's legacy `/subscribe` endpoint
 a v2 API: `wss://<host>/xrpc/network.bsky.jetstream.subscribeEvents`, with a repeated
 `collections` param (confirmed live, documented as supporting an
 `app.bsky.feed.*`-style NSID-prefix wildcard, so `run.nibling.*` would also work), `kinds`, and a
-`cursor` that's a `seq` rather than a `time_us` (a `time_us`-shaped cursor is still accepted by
-magnitude on the live tail, per the docs, which is why this doesn't store or compare a typed
-timestamp). `src/index/jetstream.ts` targets the verified v2 shape; `wrangler.jsonc`'s
+`cursor` that takes either a `seq` or a `time_us`, told apart by magnitude. The index stores a
+`time_us`, since that also gives the cursor's age; a live probe on 2026-11-06 with a `time_us`
+two minutes old replayed from two minutes back. `src/index/jetstream.ts` targets the verified v2 shape; `wrangler.jsonc`'s
 `JETSTREAM_URL` points at `wss://jetstream.us-east.bsky.network` accordingly. A live, read-only
 connection to that URL with this worker's exact query string opened cleanly and returned zero
 `run.nibling.*` commits in an 8-second window, which is expected: nibling.run has no real users
diff --git a/packages/api/src/index/backfill.ts b/packages/api/src/index/backfill.ts
index ee21053..4172ea9 100644
--- a/packages/api/src/index/backfill.ts
+++ b/packages/api/src/index/backfill.ts
@@ -1,4 +1,4 @@
-import type { IndexEvent } from './model.ts';
+import type { IndexCollection, IndexEvent } from './model.ts';
 import { toPutEvent, WANTED_COLLECTIONS } from './jetstream.ts';
 
 /** Injected rather than `globalThis.fetch` so tests run with no network. */
@@ -9,6 +9,13 @@ export interface BackfillDeps {
   resolvePds(did: string): Promise<string>;
 }
 
+/** What a backfill produces. `repo` names every record one repo's PDS still holds in a
+ * collection, so the index can drop rows deleted while it wasn't listening. It is emitted only
+ * after that repo's listing finished, never after a partial one. */
+export type BackfillItem =
+  | { kind: 'event'; event: IndexEvent }
+  | { kind: 'repo'; collection: IndexCollection; did: string; uris: ReadonlySet<string> };
+
 interface ListReposByCollectionResponse {
   repos: { did: string }[];
   cursor?: string;
@@ -48,14 +55,16 @@ export function ownedRecordUri(uri: unknown, did: string, collection: string): s
 }
 
 /** `com.atproto.repo.listRecords` against one repo's own PDS, paged to exhaustion. A repo whose
- * PDS can't be reached is skipped rather than failing the whole backfill. */
-async function* listRecords(did: string, collection: string, deps: BackfillDeps): AsyncGenerator<IndexEvent> {
+ * PDS can't be reached is skipped rather than failing the whole backfill, and gets no `repo`
+ * item, so none of its rows are pruned. */
+async function* listRecords(did: string, collection: IndexCollection, deps: BackfillDeps): AsyncGenerator<BackfillItem> {
   let pds: string;
   try {
     pds = await deps.resolvePds(did);
   } catch {
     return;
   }
+  const uris = new Set<string>();
   let cursor: string | undefined;
   do {
     const url = new URL('/xrpc/com.atproto.repo.listRecords', pds);
@@ -74,23 +83,28 @@ async function* listRecords(did: string, collection: string, deps: BackfillDeps)
     for (const record of body.records) {
       const uri = ownedRecordUri(record.uri, did, collection);
       if (!uri) continue;
+      uris.add(uri);
       const event = toPutEvent(collection, uri, record.cid, did, record.value);
-      if (event) yield event;
+      if (event) yield { kind: 'event', event };
     }
     cursor = body.cursor;
   } while (cursor);
+  yield { kind: 'repo', collection, did, uris };
 }
 
 /** Rebuilds the index for one collection from the relay and each repo's PDS, used when the
  * stored cursor is missing or older than Jetstream's replay window. */
-export async function* backfillCollection(collection: string, deps: BackfillDeps): AsyncGenerator<IndexEvent> {
+export async function* backfillCollection(collection: IndexCollection, deps: BackfillDeps): AsyncGenerator<BackfillItem> {
+  const dids = new Set<string>();
   for await (const did of listReposByCollection(collection, deps)) {
+    if (dids.has(did)) continue;
+    dids.add(did);
     yield* listRecords(did, collection, deps);
   }
 }
 
 /** Backfills every collection the index cares about. */
-export async function* backfillAll(deps: BackfillDeps): AsyncGenerator<IndexEvent> {
+export async function* backfillAll(deps: BackfillDeps): AsyncGenerator<BackfillItem> {
   for (const collection of WANTED_COLLECTIONS) {
     yield* backfillCollection(collection, deps);
   }
diff --git a/packages/api/src/index/ingest.ts b/packages/api/src/index/ingest.ts
index 8342078..095fab8 100644
--- a/packages/api/src/index/ingest.ts
+++ b/packages/api/src/index/ingest.ts
@@ -1,4 +1,4 @@
-import type { IndexEvent } from './model.ts';
+import type { IndexCollection, IndexEvent } from './model.ts';
 import { parseJetstreamMessage, subscribeUrl } from './jetstream.ts';
 import { backfillAll, type BackfillDeps } from './backfill.ts';
 
@@ -7,101 +7,128 @@ import { backfillAll, type BackfillDeps } from './backfill.ts';
  * outgoing client) satisfies this. */
 export interface WebSocketLike {
   addEventListener(type: 'message', listener: (event: { data: unknown }) => void): void;
-  addEventListener(type: 'close' | 'error', listener: () => void): void;
+  addEventListener(type: 'open' | 'close' | 'error', listener: () => void): void;
   close(): void;
 }
 
+/** What ingestion writes to. `IndexModel` implements it. */
+export interface IngestIndex {
+  apply(event: IndexEvent): void;
+  pruneRepo(collection: IndexCollection, did: string, keep: ReadonlySet<string>): void;
+  /** Jetstream `time_us`: every event before it has been applied. */
+  cursor(): number | undefined;
+  setCursor(timeUs: number): void;
+}
+
 export interface IngestDeps {
   jetstreamUrl: string;
   connect(url: string): WebSocketLike;
-  apply(event: IndexEvent): void;
-  cursor(): string | undefined;
-  setCursor(value: string): void;
-  backfill: Omit<BackfillDeps, 'relayUrl'> & { relayUrl: string };
+  index: IngestIndex;
+  backfill: BackfillDeps;
   now(): number;
-  /** How long a stale-cursor replay window is treated as expired. Default 24h, per the spec. */
+  /** How old a cursor can get before Jetstream can no longer replay from it. Default 24h. */
   replayWindowMs?: number;
-  /** How long to drain the live tail before returning. Default 20s, per the spec. */
+  /** How long to drain the live tail before returning. Default 20s. */
   drainBudgetMs?: number;
-  /** How close to "now" an event's `time` must be to call the drain caught-up. Default 5s. */
+  /** How close to "now" an event's `time` must be to call the drain caught up, and how long the
+   * socket must stay silent to call it quiet. Default 5s. */
   caughtUpSkewMs?: number;
+  /** How far one quiet drain may move the cursor past where it started. Jetstream scans replay
+   * at roughly 12 minutes of network time per second (measured 2026-11-06), so a drain that
+   * heard nothing has certainly scanned this much. Default 10 minutes. */
+  maxQuietAdvanceMs?: number;
 }
 
-/** One ingestion pass: backfill if the cursor is missing or stale, then drain the live tail
- * until caught up or the budget runs out, applying events in order and saving the cursor as it
- * goes so a crash mid-drain doesn't replay from scratch. Does not hold the socket open past
- * this call — the DO's alarm (every 60s) is the only thing that keeps calling it. */
+/**
+ * One ingestion pass. If the cursor is missing or outside Jetstream's replay window, backfill
+ * from the relay and each PDS first, then replay Jetstream from the moment the backfill started,
+ * so events that land during the backfill are not lost. Then drain the live tail until caught
+ * up or out of budget, saving the cursor as it goes. The socket is not held open past this call;
+ * the DO's alarm (every 60s) is the only thing that keeps calling it.
+ */
 export async function ingest(deps: IngestDeps): Promise<void> {
   const replayWindowMs = deps.replayWindowMs ?? 24 * 60 * 60 * 1000;
-  const drainBudgetMs = deps.drainBudgetMs ?? 20_000;
-  const caughtUpSkewMs = deps.caughtUpSkewMs ?? 5_000;
 
-  let cursor = deps.cursor();
-  const cursorAgeMs = cursor ? deps.now() - cursorTimeMs(cursor) : Number.POSITIVE_INFINITY;
-  if (!cursor || cursorAgeMs > replayWindowMs) {
-    for await (const event of backfillAll(deps.backfill)) {
-      deps.apply(event);
+  let from = deps.index.cursor();
+  if (from === undefined || deps.now() - from / 1000 > replayWindowMs) {
+    const mark = deps.now() * 1000;
+    for await (const item of backfillAll(deps.backfill)) {
+      if (item.kind === 'event') deps.index.apply(item.event);
+      else deps.index.pruneRepo(item.collection, item.did, item.uris);
     }
-    cursor = undefined;
+    deps.index.setCursor(mark);
+    from = mark;
   }
 
-  await drainLiveTail(deps, cursor, { drainBudgetMs, caughtUpSkewMs });
-}
-
-/** A Jetstream `cursor` for `run.nibling.*` is a `seq`, an opaque increasing integer with no
- * fixed relationship to wall-clock time (see jetstream.ts). There's no way to derive its age
- * without tracking the `time` of the event it came from, so the cursor is stored as
- * `"<seq>:<time_ms>"` and only the time half is used for staleness. */
-function cursorTimeMs(cursor: string): number {
-  const [, timeMs] = cursor.split(':');
-  const parsed = timeMs ? Number.parseInt(timeMs, 10) : Number.NaN;
-  return Number.isFinite(parsed) ? parsed : Number.NEGATIVE_INFINITY;
+  await drainLiveTail(deps, from, {
+    drainBudgetMs: deps.drainBudgetMs ?? 20_000,
+    caughtUpSkewMs: deps.caughtUpSkewMs ?? 5_000,
+    maxQuietAdvanceMs: deps.maxQuietAdvanceMs ?? 10 * 60 * 1000,
+  });
 }
 
-function cursorSeq(cursor: string): string {
-  return cursor.split(':', 1)[0] ?? cursor;
+function eventTimeUs(raw: unknown): number | undefined {
+  const time = (raw as { payload?: { time?: unknown } } | null)?.payload?.time;
+  if (typeof time !== 'string') return undefined;
+  const ms = Date.parse(time);
+  return Number.isFinite(ms) ? ms * 1000 : undefined;
 }
 
 function drainLiveTail(
   deps: IngestDeps,
-  cursor: string | undefined,
-  options: { drainBudgetMs: number; caughtUpSkewMs: number },
+  fromUs: number,
+  options: { drainBudgetMs: number; caughtUpSkewMs: number; maxQuietAdvanceMs: number },
 ): Promise<void> {
   return new Promise((resolve) => {
-    const url = subscribeUrl(deps.jetstreamUrl, cursor ? cursorSeq(cursor) : undefined);
-    const ws = deps.connect(url);
+    const ws = deps.connect(subscribeUrl(deps.jetstreamUrl, String(fromUs)));
+    let latestUs = fromUs;
+    let openedAt: number | undefined;
+    let lastMessageAt: number | undefined;
     let done = false;
-    const finish = () => {
+
+    const advance = (timeUs: number) => {
+      if (timeUs <= latestUs) return;
+      latestUs = timeUs;
+      deps.index.setCursor(timeUs);
+    };
+
+    // A run.nibling.* stream is often silent for minutes, and Jetstream sends nothing on a
+    // filtered stream with no matches. Without this, a quiet network never moves the cursor and
+    // ages into a full backfill. Only a drain that stayed open and silent counts.
+    const finish = (budgetSpent: boolean) => {
       if (done) return;
       done = true;
       clearTimeout(timer);
+      if (budgetSpent && openedAt !== undefined && deps.now() - (lastMessageAt ?? openedAt) >= options.caughtUpSkewMs) {
+        const quietMs = Math.min(openedAt - options.caughtUpSkewMs, fromUs / 1000 + options.maxQuietAdvanceMs);
+        advance(quietMs * 1000);
+      }
       ws.close();
       resolve();
     };
-    const timer = setTimeout(finish, options.drainBudgetMs);
+    const timer = setTimeout(() => finish(true), options.drainBudgetMs);
 
+    ws.addEventListener('open', () => {
+      openedAt = deps.now();
+    });
     ws.addEventListener('message', (event) => {
       if (done) return;
+      lastMessageAt = deps.now();
       let raw: unknown;
       try {
         raw = typeof event.data === 'string' ? JSON.parse(event.data) : event.data;
       } catch {
         return;
       }
-      const time = typeof (raw as { payload?: { time?: unknown } })?.payload?.time === 'string'
-        ? Date.parse((raw as { payload: { time: string } }).payload.time)
-        : undefined;
-      const seq = (raw as { payload?: { seq?: unknown } })?.payload?.seq;
-
       const parsed = parseJetstreamMessage(raw);
-      if (parsed) deps.apply(parsed);
+      if (parsed) deps.index.apply(parsed);
 
-      if (typeof seq === 'number' && typeof time === 'number') {
-        deps.setCursor(`${seq}:${time}`);
-        if (deps.now() - time <= options.caughtUpSkewMs) finish();
-      }
+      const timeUs = eventTimeUs(raw);
+      if (timeUs === undefined) return;
+      advance(timeUs);
+      if (deps.now() - timeUs / 1000 <= options.caughtUpSkewMs) finish(false);
     });
-    ws.addEventListener('close', finish);
-    ws.addEventListener('error', finish);
+    ws.addEventListener('close', () => finish(false));
+    ws.addEventListener('error', () => finish(false));
   });
 }
diff --git a/packages/api/src/index/jetstream.ts b/packages/api/src/index/jetstream.ts
index ac3666a..94f3e12 100644
--- a/packages/api/src/index/jetstream.ts
+++ b/packages/api/src/index/jetstream.ts
@@ -1,14 +1,13 @@
 import { safeParse } from '@atcute/lexicons';
 import { RunNiblingActorProfile, RunNiblingFeedLike, RunNiblingGraphFollow, RunNiblingProject } from '@nibling/lexicons';
-import type { IndexEvent } from './model.ts';
+import { INDEX_COLLECTIONS, type IndexCollection, type IndexEvent } from './model.ts';
 
 /** The four collections the index cares about (D22). */
-export const WANTED_COLLECTIONS = [
-  'run.nibling.project',
-  'run.nibling.feed.like',
-  'run.nibling.graph.follow',
-  'run.nibling.actor.profile',
-] as const;
+export const WANTED_COLLECTIONS = INDEX_COLLECTIONS;
+
+function isIndexCollection(value: string): value is IndexCollection {
+  return (INDEX_COLLECTIONS as readonly string[]).includes(value);
+}
 
 /**
  * Builds the live-tail subscribe URL.
@@ -16,9 +15,8 @@ export const WANTED_COLLECTIONS = [
  * Deviation from the api-spec draft: the spec assumed Jetstream's legacy `/subscribe` endpoint
  * (`wantedCollections=...&cursor=<time_us>`). The current docs (bsky.network/docs/jetstream,
  * read 2026-11-06) describe a v2 API at `/xrpc/network.bsky.jetstream.subscribeEvents` with
- * repeated `collections` params, `kinds`, and a `cursor` that's a `seq` (a v1-style `time_us`
- * is still accepted by magnitude, per the docs, which is why `cursor()` below stays an opaque
- * string rather than a typed `time_us`). `collections` does document NSID-prefix wildcards
+ * repeated `collections` params, `kinds`, and a `cursor` that's either a `seq` or a `time_us`,
+ * told apart by magnitude. The index passes a `time_us` (see ingest.ts). `collections` does document NSID-prefix wildcards
  * ("an `app.bsky.feed.*`-style namespace wildcard"), so `run.nibling.*` would also work, but
  * since there are only four collections to ask for, this lists them explicitly instead.
  */
@@ -67,18 +65,7 @@ export function parseJetstreamMessage(raw: unknown, now: () => string = () => ne
   const uri = `at://${payload.did}/${payload.collection}/${payload.rkey}`;
 
   if (payload.operation === 'delete') {
-    switch (payload.collection) {
-      case 'run.nibling.project':
-        return { action: 'delete', collection: 'run.nibling.project', uri };
-      case 'run.nibling.feed.like':
-        return { action: 'delete', collection: 'run.nibling.feed.like', uri };
-      case 'run.nibling.graph.follow':
-        return { action: 'delete', collection: 'run.nibling.graph.follow', uri };
-      case 'run.nibling.actor.profile':
-        return { action: 'delete', collection: 'run.nibling.actor.profile', uri };
-      default:
-        return undefined;
-    }
+    return isIndexCollection(payload.collection) ? { action: 'delete', collection: payload.collection, uri } : undefined;
   }
 
   if (payload.operation !== 'create' && payload.operation !== 'update') return undefined;
diff --git a/packages/api/src/index/model.ts b/packages/api/src/index/model.ts
index 3177cdf..2af7f4b 100644
--- a/packages/api/src/index/model.ts
+++ b/packages/api/src/index/model.ts
@@ -35,13 +35,29 @@ interface ProfileRecord {
 /** One put or tombstone, already validated at the boundary (jetstream.ts, backfill.ts). */
 export type IndexEvent =
   | { action: 'put'; collection: 'run.nibling.project'; uri: string; cid: string; did: string; record: ProjectRecord; indexedAt: string }
-  | { action: 'delete'; collection: 'run.nibling.project'; uri: string }
   | { action: 'put'; collection: 'run.nibling.feed.like'; uri: string; cid: string; did: string; record: LikeRecord }
-  | { action: 'delete'; collection: 'run.nibling.feed.like'; uri: string }
   | { action: 'put'; collection: 'run.nibling.graph.follow'; uri: string; cid: string; did: string; record: FollowRecord }
-  | { action: 'delete'; collection: 'run.nibling.graph.follow'; uri: string }
   | { action: 'put'; collection: 'run.nibling.actor.profile'; uri: string; cid: string; did: string; record: ProfileRecord }
-  | { action: 'delete'; collection: 'run.nibling.actor.profile'; uri: string };
+  | { action: 'delete'; collection: IndexCollection; uri: string };
+
+type PutEvent<C extends IndexCollection> = Extract<IndexEvent, { action: 'put'; collection: C }>;
+
+export const INDEX_COLLECTIONS = [
+  'run.nibling.project',
+  'run.nibling.feed.like',
+  'run.nibling.graph.follow',
+  'run.nibling.actor.profile',
+] as const;
+
+export type IndexCollection = (typeof INDEX_COLLECTIONS)[number];
+
+/** Where each collection's rows live, keyed by the repo DID that owns them. */
+const TABLES: Record<IndexCollection, string> = {
+  'run.nibling.project': 'projects',
+  'run.nibling.feed.like': 'likes',
+  'run.nibling.graph.follow': 'follows',
+  'run.nibling.actor.profile': 'profiles',
+};
 
 export interface RecentOptions {
   limit?: number;
@@ -75,6 +91,7 @@ const SCHEMA = [
   )`,
   `CREATE INDEX IF NOT EXISTS projects_sortAt ON projects(sortAt DESC)`,
   `CREATE INDEX IF NOT EXISTS projects_forkOf ON projects(forkOf)`,
+  `CREATE INDEX IF NOT EXISTS projects_did ON projects(did)`,
   `CREATE TABLE IF NOT EXISTS project_tags (
     uri TEXT NOT NULL,
     tag TEXT NOT NULL,
@@ -88,12 +105,14 @@ const SCHEMA = [
     createdAtMs INTEGER NOT NULL
   )`,
   `CREATE INDEX IF NOT EXISTS likes_subject ON likes(subjectUri)`,
+  `CREATE INDEX IF NOT EXISTS likes_did ON likes(did)`,
   `CREATE TABLE IF NOT EXISTS follows (
     uri TEXT PRIMARY KEY,
     did TEXT NOT NULL,
     subject TEXT NOT NULL,
     createdAt TEXT NOT NULL
   )`,
+  `CREATE INDEX IF NOT EXISTS follows_did ON follows(did)`,
   `CREATE TABLE IF NOT EXISTS profiles (
     did TEXT PRIMARY KEY,
     uri TEXT NOT NULL,
@@ -105,11 +124,6 @@ const SCHEMA = [
   )`,
 ];
 
-function didFromUri(uri: string): string {
-  // at://<did>/<collection>/<rkey>
-  return uri.slice('at://'.length).split('/')[0] ?? '';
-}
-
 function encodeCursor(sortAt: string, uri: string): string {
   return Buffer.from(JSON.stringify([sortAt, uri])).toString('base64url');
 }
@@ -136,6 +150,10 @@ export class IndexModel {
   }
 
   apply(event: IndexEvent): void {
+    if (event.action === 'delete') {
+      this.remove(event.collection, event.uri);
+      return;
+    }
     switch (event.collection) {
       case 'run.nibling.project':
         this.applyProject(event);
@@ -152,12 +170,12 @@ export class IndexModel {
     }
   }
 
-  private applyProject(event: IndexEvent & { collection: 'run.nibling.project' }): void {
-    if (event.action === 'delete') {
-      this.sql.exec('DELETE FROM projects WHERE uri = ?', event.uri);
-      this.sql.exec('DELETE FROM project_tags WHERE uri = ?', event.uri);
-      return;
-    }
+  private remove(collection: IndexCollection, uri: string): void {
+    if (collection === 'run.nibling.project') this.sql.exec('DELETE FROM project_tags WHERE uri = ?', uri);
+    this.sql.exec(`DELETE FROM ${TABLES[collection]} WHERE uri = ?`, uri);
+  }
+
+  private applyProject(event: PutEvent<'run.nibling.project'>): void {
     const { uri, cid, did, record, indexedAt } = event;
     const sortAt = record.updatedAt ?? record.createdAt;
     this.sql.exec(
@@ -176,11 +194,7 @@ export class IndexModel {
     }
   }
 
-  private applyLike(event: IndexEvent & { collection: 'run.nibling.feed.like' }): void {
-    if (event.action === 'delete') {
-      this.sql.exec('DELETE FROM likes WHERE uri = ?', event.uri);
-      return;
-    }
+  private applyLike(event: PutEvent<'run.nibling.feed.like'>): void {
     const { uri, did, record } = event;
     this.sql.exec(
       `INSERT INTO likes (uri, did, subjectUri, createdAtMs) VALUES (?, ?, ?, ?)
@@ -189,11 +203,7 @@ export class IndexModel {
     );
   }
 
-  private applyFollow(event: IndexEvent & { collection: 'run.nibling.graph.follow' }): void {
-    if (event.action === 'delete') {
-      this.sql.exec('DELETE FROM follows WHERE uri = ?', event.uri);
-      return;
-    }
+  private applyFollow(event: PutEvent<'run.nibling.graph.follow'>): void {
     const { uri, did, record } = event;
     this.sql.exec(
       `INSERT INTO follows (uri, did, subject, createdAt) VALUES (?, ?, ?, ?)
@@ -202,11 +212,7 @@ export class IndexModel {
     );
   }
 
-  private applyProfile(event: IndexEvent & { collection: 'run.nibling.actor.profile' }): void {
-    if (event.action === 'delete') {
-      this.sql.exec('DELETE FROM profiles WHERE did = ?', didFromUri(event.uri));
-      return;
-    }
+  private applyProfile(event: PutEvent<'run.nibling.actor.profile'>): void {
     const { uri, did, record } = event;
     this.sql.exec(
       `INSERT INTO profiles (did, uri, createdAt) VALUES (?, ?, ?)
@@ -325,16 +331,27 @@ export class IndexModel {
     return dids.filter((did) => found.has(did));
   }
 
-  cursor(): string | undefined {
+  /** Deletes the rows one repo owns in a collection that its PDS no longer lists, which is how
+   * a backfill catches deletions that happened while the index wasn't listening. */
+  pruneRepo(collection: IndexCollection, did: string, keep: ReadonlySet<string>): void {
+    const rows = this.sql.exec<{ uri: string }>(`SELECT uri FROM ${TABLES[collection]} WHERE did = ?`, did);
+    for (const { uri } of rows) {
+      if (!keep.has(uri)) this.apply({ action: 'delete', collection, uri });
+    }
+  }
+
+  /** The Jetstream cursor, as `time_us` (microseconds since the epoch). */
+  cursor(): number | undefined {
     const rows = this.sql.exec<{ value: string }>("SELECT value FROM meta WHERE key = 'jetstreamCursor'");
-    return rows[0]?.value;
+    const value = rows[0] ? Number(rows[0].value) : Number.NaN;
+    return Number.isSafeInteger(value) ? value : undefined;
   }
 
-  setCursor(value: string): void {
+  setCursor(timeUs: number): void {
     this.sql.exec(
       `INSERT INTO meta (key, value) VALUES ('jetstreamCursor', ?)
        ON CONFLICT(key) DO UPDATE SET value = excluded.value`,
-      value,
+      String(timeUs),
     );
   }
 }
diff --git a/packages/api/src/worker.ts b/packages/api/src/worker.ts
index 652b573..8978fb3 100644
--- a/packages/api/src/worker.ts
+++ b/packages/api/src/worker.ts
@@ -61,9 +61,7 @@ export class Index extends DurableObject<Env> {
     await ingest({
       jetstreamUrl: this.env.JETSTREAM_URL,
       connect: (url) => new WebSocket(url) as unknown as WebSocketLike,
-      apply: (event) => this.#model.apply(event),
-      cursor: () => this.#model.cursor(),
-      setCursor: (value) => this.#model.setCursor(value),
+      index: this.#model,
       backfill: {
         relayUrl: this.env.RELAY_URL,
         fetch,
diff --git a/packages/api/tests/backfill.test.ts b/packages/api/tests/backfill.test.ts
index 3a8e60f..ce2c277 100644
--- a/packages/api/tests/backfill.test.ts
+++ b/packages/api/tests/backfill.test.ts
@@ -1,6 +1,16 @@
 import assert from 'node:assert/strict';
 import { test } from 'node:test';
-import { backfillCollection, type BackfillDeps } from '../src/index/backfill.ts';
+import { backfillCollection, type BackfillDeps, type BackfillItem } from '../src/index/backfill.ts';
+
+async function collect(items: AsyncIterable<BackfillItem>): Promise<BackfillItem[]> {
+  const out: BackfillItem[] = [];
+  for await (const item of items) out.push(item);
+  return out;
+}
+
+function eventUris(items: BackfillItem[]): string[] {
+  return items.flatMap((item) => (item.kind === 'event' ? [item.event.uri] : []));
+}
 
 function jsonResponse(body: unknown, ok = true): Response {
   return new Response(JSON.stringify(body), { status: ok ? 200 : 500 });
@@ -51,13 +61,12 @@ test('backfillCollection pages listReposByCollection and listRecords, and valida
     resolvePds: async (did) => (did === 'did:plc:alice' ? 'https://alice.pds.example' : 'https://bob.pds.example'),
   };
 
-  const events = [];
-  for await (const event of backfillCollection('run.nibling.project', deps)) events.push(event);
+  const items = await collect(backfillCollection('run.nibling.project', deps));
 
-  assert.deepStrictEqual(
-    events.map((e) => e.uri),
-    ['at://did:plc:alice/run.nibling.project/3la1'],
-  );
+  assert.deepStrictEqual(items.map((item) => (item.kind === 'event' ? item.event.uri : { repo: item.did, uris: [...item.uris] })), [
+    'at://did:plc:alice/run.nibling.project/3la1',
+    { repo: 'did:plc:alice', uris: ['at://did:plc:alice/run.nibling.project/3la1', 'at://did:plc:alice/run.nibling.project/3la2'] },
+  ]);
   assert.deepStrictEqual(calls, [
     '/xrpc/com.atproto.sync.listReposByCollection?collection=run.nibling.project&limit=1000',
     '/xrpc/com.atproto.repo.listRecords?repo=did%3Aplc%3Aalice&collection=run.nibling.project&limit=100',
@@ -74,9 +83,7 @@ test('backfillCollection skips a repo whose PDS cannot be resolved', async () =>
       throw new Error('did not found');
     },
   };
-  const events = [];
-  for await (const event of backfillCollection('run.nibling.project', deps)) events.push(event);
-  assert.deepStrictEqual(events, []);
+  assert.deepStrictEqual(await collect(backfillCollection('run.nibling.project', deps)), []);
 });
 
 test('backfillCollection drops records whose URI names another repo or collection', async () => {
@@ -98,10 +105,9 @@ test('backfillCollection drops records whose URI names another repo or collectio
     }) as typeof fetch,
     resolvePds: async () => 'https://evil.pds.example',
   };
-  const events = [];
-  for await (const event of backfillCollection('run.nibling.project', deps)) events.push(event);
+  const items = await collect(backfillCollection('run.nibling.project', deps));
   assert.deepStrictEqual(
-    events.map((e) => ({ uri: e.uri, did: e.action === 'put' ? e.did : undefined })),
-    [{ uri: 'at://did:plc:evil/run.nibling.project/3own', did: 'did:plc:evil' }],
+    items.map((item) => (item.kind === 'event' && item.event.action === 'put' ? { uri: item.event.uri, did: item.event.did } : item.kind)),
+    [{ uri: 'at://did:plc:evil/run.nibling.project/3own', did: 'did:plc:evil' }, 'repo'],
   );
 });
diff --git a/packages/api/tests/ingest.test.ts b/packages/api/tests/ingest.test.ts
index d8303ac..59b600e 100644
--- a/packages/api/tests/ingest.test.ts
+++ b/packages/api/tests/ingest.test.ts
@@ -1,32 +1,34 @@
 import assert from 'node:assert/strict';
 import { test } from 'node:test';
-import type { IndexEvent } from '../src/index/model.ts';
+import { IndexModel } from '../src/index/model.ts';
 import { ingest, type IngestDeps, type WebSocketLike } from '../src/index/ingest.ts';
+import { sqliteSql } from './sqlite-sql.ts';
 
-interface FakeFrame {
-  type: 'message' | 'close';
-  data?: unknown;
-}
+type FakeFrame = { type: 'open' } | { type: 'message'; data: unknown } | { type: 'close' } | { type: 'error' };
 
-/** A fake live tail that plays back a fixed script of frames, one per microtask, so the test
- * doesn't depend on real timers or a real socket. */
-function fakeSocket(frames: FakeFrame[]): WebSocketLike {
-  const listeners: Record<string, ((event: { data: unknown }) => void)[]> = {};
-  let index = 0;
-  const playNext = () => {
-    if (index >= frames.length) return;
-    const frame = frames[index++]!;
-    for (const listener of listeners[frame.type] ?? []) listener({ data: frame.data });
-    if (frame.type !== 'close') queueMicrotask(playNext);
-  };
-  queueMicrotask(playNext);
-  return {
-    addEventListener(type: string, listener: (event: { data: unknown }) => void) {
-      (listeners[type] ??= []).push(listener as never);
-    },
-    close() {
-      for (const listener of listeners.close ?? []) listener({ data: undefined });
-    },
+/** A fake live tail that plays back a fixed script of frames, one per microtask, and records
+ * the URL it was opened with. After the script runs out it stays open and silent. */
+function fakeSocket(frames: FakeFrame[], urls: string[]): (url: string) => WebSocketLike {
+  return (url) => {
+    urls.push(url);
+    const listeners: Record<string, ((event: { data: unknown }) => void)[]> = {};
+    let index = 0;
+    const playNext = () => {
+      if (index >= frames.length) return;
+      const frame = frames[index++]!;
+      const data = frame.type === 'message' ? frame.data : undefined;
+      for (const listener of listeners[frame.type] ?? []) listener({ data });
+      if (frame.type !== 'close' && frame.type !== 'error') queueMicrotask(playNext);
+    };
+    queueMicrotask(playNext);
+    return {
+      addEventListener(type: string, listener: (event: { data: unknown }) => void) {
+        (listeners[type] ??= []).push(listener as never);
+      },
+      close() {
+        for (const listener of listeners.close ?? []) listener({ data: undefined });
+      },
+    };
   };
 }
 
@@ -50,89 +52,198 @@ function commitFrame(seq: number, timeIso: string): FakeFrame {
   };
 }
 
-function baseDeps(overrides: Partial<IngestDeps> = {}): { applied: IndexEvent[]; deps: IngestDeps } {
-  const applied: IndexEvent[] = [];
-  let cursor: string | undefined;
+const NOW = Date.parse('2026-11-06T00:00:10.000Z');
+const us = (iso: string) => Date.parse(iso) * 1000;
+
+const noRepos = (async () => Response.json({ repos: [] })) as typeof fetch;
+
+function baseDeps(overrides: Partial<IngestDeps> = {}): { model: IndexModel; urls: string[]; deps: IngestDeps } {
+  const model = new IndexModel(sqliteSql());
+  const urls: string[] = [];
   const deps: IngestDeps = {
     jetstreamUrl: 'wss://jetstream.us-east.bsky.network',
-    connect: () => fakeSocket([]),
-    apply: (event) => applied.push(event),
-    cursor: () => cursor,
-    setCursor: (value) => {
-      cursor = value;
-    },
+    connect: fakeSocket([{ type: 'open' }], urls),
+    index: model,
     backfill: {
       relayUrl: 'https://relay1.us-east.bsky.network',
-      fetch: (async () => new Response(JSON.stringify({ repos: [] }))) as typeof fetch,
+      fetch: noRepos,
       resolvePds: async () => {
         throw new Error('not reached: no repos to resolve');
       },
     },
-    now: () => Date.parse('2026-11-06T00:00:10.000Z'),
-    drainBudgetMs: 1000,
+    now: () => NOW,
+    drainBudgetMs: 20,
     caughtUpSkewMs: 5000,
     ...overrides,
   };
-  return { applied, deps };
+  return { model, urls, deps };
+}
+
+function cursorParam(url: string | undefined): string | null {
+  return url ? new URL(url).searchParams.get('cursor') : null;
 }
 
 test('drains the live tail until an event lands within the caught-up skew, then closes', async () => {
-  const { applied, deps } = baseDeps({
-    connect: () =>
-      fakeSocket([
-        commitFrame(1, '2026-11-05T00:00:00.000Z'), // far behind "now"
-        commitFrame(2, '2026-11-06T00:00:08.000Z'), // within 5s of "now" (00:00:10): caught up
-        commitFrame(3, '2026-11-06T00:00:09.000Z'), // would arrive after close; never applied
-      ]),
+  const urls: string[] = [];
+  const { model, deps } = baseDeps({
+    connect: fakeSocket(
+      [
+        { type: 'open' },
+        commitFrame(1, '2026-11-05T12:00:00.000Z'),
+        commitFrame(2, '2026-11-06T00:00:08.000Z'),
+        commitFrame(3, '2026-11-06T00:00:09.000Z'),
+      ],
+      urls,
+    ),
   });
+  model.setCursor(us('2026-11-05T11:00:00.000Z'));
 
   await ingest(deps);
 
-  assert.deepStrictEqual(applied.map((e) => e.uri), [
-    'at://did:plc:alice/run.nibling.project/3la1',
+  assert.deepStrictEqual(model.recent().projects.map((p) => p.uri), [
     'at://did:plc:alice/run.nibling.project/3la2',
+    'at://did:plc:alice/run.nibling.project/3la1',
   ]);
-  assert.equal(deps.cursor(), `2:${Date.parse('2026-11-06T00:00:08.000Z')}`);
+  assert.equal(model.cursor(), us('2026-11-06T00:00:08.000Z'));
+  assert.equal(cursorParam(urls[0]), String(us('2026-11-05T11:00:00.000Z')));
 });
 
-test('backfills first when there is no stored cursor', async () => {
-  let backfillRan = false;
-  const { applied, deps } = baseDeps({
+test('a backfill records its start as the cursor and replays the live tail from it', async () => {
+  let clock = NOW;
+  const urls: string[] = [];
+  const { model, deps } = baseDeps({
+    now: () => clock,
+    backfill: {
+      relayUrl: 'https://relay1.us-east.bsky.network',
+      fetch: (async () => {
+        clock += 30_000;
+        return Response.json({ repos: [] });
+      }) as typeof fetch,
+      resolvePds: async () => 'https://example.pds',
+    },
+    connect: fakeSocket([{ type: 'close' }], urls),
+  });
+
+  await ingest(deps);
+
+  assert.equal(model.cursor(), NOW * 1000);
+  assert.equal(cursorParam(urls[0]), String(NOW * 1000));
+});
+
+test('a backfill prunes rows the repo no longer lists', async () => {
+  const { model, deps } = baseDeps({
     backfill: {
       relayUrl: 'https://relay1.us-east.bsky.network',
       fetch: (async (input: string | URL | Request) => {
-        backfillRan = true;
         const url = new URL(typeof input === 'string' ? input : input.toString());
         if (url.pathname === '/xrpc/com.atproto.sync.listReposByCollection') {
-          return new Response(JSON.stringify({ repos: [] }));
+          return Response.json({ repos: url.searchParams.get('collection') === 'run.nibling.project' ? [{ did: 'did:plc:alice' }] : [] });
         }
-        throw new Error('unexpected');
+        return Response.json({
+          records: [
+            {
+              uri: 'at://did:plc:alice/run.nibling.project/kept',
+              cid: 'bafy-kept',
+              value: { $type: 'run.nibling.project', name: 'Kept', files: [], createdAt: '2026-11-01T00:00:00.000Z' },
+            },
+          ],
+        });
       }) as typeof fetch,
-      resolvePds: async () => 'https://example.pds',
+      resolvePds: async () => 'https://alice.pds.example',
     },
-    connect: () => fakeSocket([{ type: 'close' }]),
+  });
+  for (const rkey of ['kept', 'deleted-while-down']) {
+    model.apply({
+      action: 'put',
+      collection: 'run.nibling.project',
+      uri: `at://did:plc:alice/run.nibling.project/${rkey}`,
+      cid: 'bafy-old',
+      did: 'did:plc:alice',
+      record: { name: rkey, createdAt: '2026-10-01T00:00:00.000Z' },
+      indexedAt: '2026-10-01T00:00:00.000Z',
+    });
+  }
+  model.apply({
+    action: 'put',
+    collection: 'run.nibling.project',
+    uri: 'at://did:plc:bob/run.nibling.project/other-repo',
+    cid: 'bafy-bob',
+    did: 'did:plc:bob',
+    record: { name: 'Bob', createdAt: '2026-10-01T00:00:00.000Z' },
+    indexedAt: '2026-10-01T00:00:00.000Z',
   });
 
   await ingest(deps);
-  assert.equal(backfillRan, true);
-  assert.deepStrictEqual(applied, []);
+
+  assert.deepStrictEqual(model.recent().projects.map((p) => p.uri).sort(), [
+    'at://did:plc:alice/run.nibling.project/kept',
+    'at://did:plc:bob/run.nibling.project/other-repo',
+  ]);
+});
+
+test('a quiet drain moves the cursor up to when the socket opened', async () => {
+  let clock = NOW;
+  const { model, deps } = baseDeps({ now: () => clock });
+  model.setCursor((NOW - 60_000) * 1000);
+  const run = ingest(deps);
+  await new Promise((r) => setTimeout(r, 5));
+  clock += 6000;
+  await run;
+  assert.equal(model.cursor(), (NOW - 5000) * 1000);
+});
+
+test('a quiet drain moves the cursor at most maxQuietAdvanceMs past where it started', async () => {
+  let clock = NOW;
+  const { model, deps } = baseDeps({ now: () => clock, maxQuietAdvanceMs: 10_000 });
+  const start = (NOW - 3_600_000) * 1000;
+  model.setCursor(start);
+  const run = ingest(deps);
+  await new Promise((r) => setTimeout(r, 5));
+  clock += 6000;
+  await run;
+  assert.equal(model.cursor(), start + 10_000 * 1000);
+});
+
+test('a drain that errors does not move the cursor', async () => {
+  const { model, deps } = baseDeps({ connect: fakeSocket([{ type: 'open' }, { type: 'error' }], []) });
+  model.setCursor((NOW - 60_000) * 1000);
+  await ingest(deps);
+  assert.equal(model.cursor(), (NOW - 60_000) * 1000);
 });
 
 test('skips backfill when the stored cursor is still inside the replay window', async () => {
   let backfillRan = false;
-  const { deps } = baseDeps({
-    cursor: () => `5:${Date.parse('2026-11-06T00:00:00.000Z')}`, // 10s before "now", well inside the 24h window
+  const { model, deps } = baseDeps({
     backfill: {
       relayUrl: 'https://relay1.us-east.bsky.network',
       fetch: (async () => {
         backfillRan = true;
-        return new Response(JSON.stringify({ repos: [] }));
+        return Response.json({ repos: [] });
       }) as typeof fetch,
       resolvePds: async () => 'https://example.pds',
     },
-    connect: () => fakeSocket([{ type: 'close' }]),
+    connect: fakeSocket([{ type: 'close' }], []),
   });
-
+  model.setCursor(us('2026-11-06T00:00:00.000Z'));
   await ingest(deps);
   assert.equal(backfillRan, false);
 });
+
+test('backfills when the stored cursor is older than the replay window', async () => {
+  let backfillRan = false;
+  const { model, deps } = baseDeps({
+    backfill: {
+      relayUrl: 'https://relay1.us-east.bsky.network',
+      fetch: (async () => {
+        backfillRan = true;
+        return Response.json({ repos: [] });
+      }) as typeof fetch,
+      resolvePds: async () => 'https://example.pds',
+    },
+    connect: fakeSocket([{ type: 'close' }], []),
+  });
+  model.setCursor(us('2026-11-04T00:00:00.000Z'));
+  await ingest(deps);
+  assert.equal(backfillRan, true);
+  assert.equal(model.cursor(), NOW * 1000);
+});
diff --git a/packages/api/tests/model.test.ts b/packages/api/tests/model.test.ts
index f35707e..7ae64a1 100644
--- a/packages/api/tests/model.test.ts
+++ b/packages/api/tests/model.test.ts
@@ -158,11 +158,23 @@ test('known returns exactly the DIDs with a project or profile record', () => {
   assert.deepStrictEqual(known, ['did:plc:alice', 'did:plc:dave', 'did:plc:erin']);
 });
 
-test('cursor round-trips', () => {
+test('cursor round-trips as time_us', () => {
   const model = new IndexModel(sqliteSql());
   assert.equal(model.cursor(), undefined);
-  model.setCursor('12345:1762387200000');
-  assert.equal(model.cursor(), '12345:1762387200000');
-  model.setCursor('99999:1762400000000');
-  assert.equal(model.cursor(), '99999:1762400000000');
+  model.setCursor(1762387200000000);
+  assert.equal(model.cursor(), 1762387200000000);
+  model.setCursor(1762400000123456);
+  assert.equal(model.cursor(), 1762400000123456);
+});
+
+test('pruneRepo deletes only that repo\'s rows missing from the kept set', () => {
+  const model = new IndexModel(sqliteSql());
+  for (const event of fixtureEvents()) model.apply(event);
+  model.pruneRepo('run.nibling.feed.like', 'did:plc:bob', new Set());
+  model.pruneRepo('run.nibling.project', 'did:plc:alice', new Set([project1Uri]));
+  model.pruneRepo('run.nibling.actor.profile', 'did:plc:erin', new Set());
+
+  const project1 = model.recent().projects.find((p) => p.uri === project1Uri)!;
+  assert.equal(project1.likeCount, 1);
+  assert.deepStrictEqual(model.known(['did:plc:alice', 'did:plc:erin']), ['did:plc:alice']);
 });
-- 
knotmirror 0.0.1


From a3655ca06ce443b22f06f1030284deee2c2cf345 Mon Sep 17 00:00:00 2001
From: Nate Moore <git@natemoo.re>
Date: Fri, 6 Nov 2026 01:35:37 -0500
Subject: [PATCH 05/10] Keep getKnown under Durable Object SQLite's
 100-parameter limit

known() bound each DID twice in one statement, so 50 or more DIDs
threw. It now queries in chunks of 50. The router rejects more than the
lexicon's 100 DIDs with a 400, and the test SQL harness enforces the
100-parameter limit for every model test.
---
 packages/api/src/index/model.ts   | 24 ++++++++++++++----------
 packages/api/src/router.ts        |  4 ++++
 packages/api/tests/model.test.ts  | 10 ++++++++++
 packages/api/tests/router.test.ts | 14 ++++++++++++++
 packages/api/tests/sqlite-sql.ts  |  7 +++++++
 5 files changed, 49 insertions(+), 10 deletions(-)

diff --git a/packages/api/src/index/model.ts b/packages/api/src/index/model.ts
index 2af7f4b..20d970f 100644
--- a/packages/api/src/index/model.ts
+++ b/packages/api/src/index/model.ts
@@ -74,6 +74,8 @@ export interface TrendingOptions {
 
 type ProjectView = RunNiblingFeedDefs.ProjectView;
 
+const KNOWN_CHUNK = 50;
+
 // One statement per array entry: `Sql.exec` runs exactly one statement per call (that's what
 // `ctx.storage.sql.exec` does in the DO), so the schema can't be a single multi-statement string.
 const SCHEMA = [
@@ -317,17 +319,19 @@ export class IndexModel {
   }
 
   known(dids: readonly string[]): string[] {
-    if (dids.length === 0) return [];
-    const placeholders = dids.map(() => '?').join(', ');
-    const rows = this.sql.exec<{ did: string }>(
-      `SELECT DISTINCT did FROM (
-         SELECT did FROM projects WHERE did IN (${placeholders})
+    const found = new Set<string>();
+    // Each DID binds twice, and Durable Object SQLite caps a query at 100 bound parameters.
+    for (let i = 0; i < dids.length; i += KNOWN_CHUNK) {
+      const chunk = dids.slice(i, i + KNOWN_CHUNK);
+      const placeholders = chunk.map(() => '?').join(', ');
+      const rows = this.sql.exec<{ did: string }>(
+        `SELECT did FROM projects WHERE did IN (${placeholders})
          UNION
-         SELECT did FROM profiles WHERE did IN (${placeholders})
-       )`,
-      ...dids, ...dids,
-    );
-    const found = new Set(rows.map((r) => r.did));
+         SELECT did FROM profiles WHERE did IN (${placeholders})`,
+        ...chunk, ...chunk,
+      );
+      for (const row of rows) found.add(row.did);
+    }
     return dids.filter((did) => found.has(did));
   }
 
diff --git a/packages/api/src/router.ts b/packages/api/src/router.ts
index 74353d5..1476317 100644
--- a/packages/api/src/router.ts
+++ b/packages/api/src/router.ts
@@ -24,6 +24,9 @@ const XRPC_HEADERS = {
   'cache-control': 'public, max-age=30',
 };
 
+/** `run.nibling.actor.getKnown`'s `dids` has `maxLength: 100` in the lexicon. */
+const MAX_KNOWN_DIDS = 100;
+
 function xrpcJson(body: unknown): Response {
   return new Response(JSON.stringify(body), { headers: XRPC_HEADERS });
 }
@@ -104,6 +107,7 @@ export async function handleRequest(request: Request, env: RouterEnv): Promise<R
   if (url.pathname === '/xrpc/run.nibling.actor.getKnown') {
     const dids = url.searchParams.getAll('dids').flatMap((v) => v.split(','));
     if (dids.length === 0) return xrpcError('InvalidRequest', 'dids is required', 400);
+    if (dids.length > MAX_KNOWN_DIDS) return xrpcError('InvalidRequest', `dids accepts at most ${MAX_KNOWN_DIDS}`, 400);
     return xrpcJson({ dids: await stub.getKnown(dids) });
   }
 
diff --git a/packages/api/tests/model.test.ts b/packages/api/tests/model.test.ts
index 7ae64a1..6369956 100644
--- a/packages/api/tests/model.test.ts
+++ b/packages/api/tests/model.test.ts
@@ -178,3 +178,13 @@ test('pruneRepo deletes only that repo\'s rows missing from the kept set', () =>
   assert.equal(project1.likeCount, 1);
   assert.deepStrictEqual(model.known(['did:plc:alice', 'did:plc:erin']), ['did:plc:alice']);
 });
+
+test('known answers for 100 DIDs within the Durable Object bound-parameter limit', () => {
+  const model = new IndexModel(sqliteSql());
+  for (const event of fixtureEvents()) model.apply(event);
+  const dids = Array.from({ length: 98 }, (_, i) => `did:plc:stranger${i}`);
+  dids.splice(10, 0, 'did:plc:alice');
+  dids.push('did:plc:erin');
+  assert.equal(dids.length, 100);
+  assert.deepStrictEqual(model.known(dids), ['did:plc:alice', 'did:plc:erin']);
+});
diff --git a/packages/api/tests/router.test.ts b/packages/api/tests/router.test.ts
index 89f2255..f2a42c5 100644
--- a/packages/api/tests/router.test.ts
+++ b/packages/api/tests/router.test.ts
@@ -136,3 +136,17 @@ test('/oembed with a malformed url is a 400, not a 500', async () => {
   assert.equal(res.status, 400);
   assert.deepStrictEqual(await res.json(), { error: 'InvalidRequest', message: 'url must point at a project page' });
 });
+
+test('getKnown with more than 100 DIDs is a 400', async () => {
+  const dids = Array.from({ length: 101 }, (_, i) => `did:plc:user${i}`).join(',');
+  const res = await handleRequest(new Request(`https://api.nibling.run/xrpc/run.nibling.actor.getKnown?dids=${dids}`), fakeEnv());
+  assert.equal(res.status, 400);
+  assert.deepStrictEqual(await res.json(), { error: 'InvalidRequest', message: 'dids accepts at most 100' });
+});
+
+test('getKnown with exactly 100 DIDs is answered', async () => {
+  const dids = ['did:plc:alice', ...Array.from({ length: 99 }, (_, i) => `did:plc:user${i}`)].join(',');
+  const res = await handleRequest(new Request(`https://api.nibling.run/xrpc/run.nibling.actor.getKnown?dids=${dids}`), fakeEnv());
+  assert.equal(res.status, 200);
+  assert.deepStrictEqual(await res.json(), { dids: ['did:plc:alice'] });
+});
diff --git a/packages/api/tests/sqlite-sql.ts b/packages/api/tests/sqlite-sql.ts
index 1029f6a..195b444 100644
--- a/packages/api/tests/sqlite-sql.ts
+++ b/packages/api/tests/sqlite-sql.ts
@@ -2,11 +2,18 @@
 import { DatabaseSync } from 'node:sqlite';
 import type { Sql } from '../src/index/model.ts';
 
+/** Durable Object SQLite allows at most 100 bound parameters per query
+ * (developers.cloudflare.com/durable-objects/platform/limits/); node:sqlite allows far more. */
+export const DO_MAX_BOUND_PARAMETERS = 100;
+
 /** Implements the model's `Sql` interface over `node:sqlite`, the way the spec asks for in
  * tests. The DO implements the same interface over `ctx.storage.sql.exec(...).toArray()`. */
 export function sqliteSql(db: DatabaseSync = new DatabaseSync(':memory:')): Sql {
   return {
     exec<T = Record<string, unknown>>(query: string, ...bindings: unknown[]): T[] {
+      if (bindings.length > DO_MAX_BOUND_PARAMETERS) {
+        throw new Error(`${bindings.length} bound parameters; Durable Object SQLite allows ${DO_MAX_BOUND_PARAMETERS}`);
+      }
       return db.prepare(query).all(...(bindings as never[])) as T[];
     },
   };
-- 
knotmirror 0.0.1


From 00f9bbdb7dcb04bd1dd40198e1ba6bed2b48f305 Mon Sep 17 00:00:00 2001
From: Nate Moore <git@natemoo.re>
Date: Fri, 6 Nov 2026 01:36:45 -0500
Subject: [PATCH 06/10] Decay trending scores toward the window edge, from
 server-seen time

The score was age divided by window, so a like from now scored 0 and a
week-old one scored 1. It is now (window - age) / window. A like or
fork counts from its createdAt clamped to when the index first saw it,
so a post-dated record can't stay fresh. A negative trending cursor is
clamped to 0.
---
 packages/api/README.md              |  4 ++-
 packages/api/src/index/jetstream.ts |  2 +-
 packages/api/src/index/model.ts     | 44 +++++++++++++++++------------
 packages/api/tests/model.test.ts    | 37 ++++++++++++++++++++++++
 4 files changed, 67 insertions(+), 20 deletions(-)

diff --git a/packages/api/README.md b/packages/api/README.md
index e092b7f..9f8faab 100644
--- a/packages/api/README.md
+++ b/packages/api/README.md
@@ -10,7 +10,9 @@ holds that data as the source of truth.
 - `GET /xrpc/run.nibling.feed.getRecent?limit=&cursor=&tag=` — recently created or updated
   projects, newest first.
 - `GET /xrpc/run.nibling.feed.getTrending?limit=&cursor=` — projects ranked by likes and forks
-  from the last 7 days, decayed linearly to zero at the window edge.
+  from the last 7 days, decayed linearly to zero at the window edge. A like or fork counts from
+  its `createdAt`, but never later than when the index first saw it, so a post-dated record
+  can't stay fresh.
 - `GET /xrpc/run.nibling.actor.getKnown?dids=` — which of the given DIDs have a
   `run.nibling.project` or `run.nibling.actor.profile` record. Used to narrow imported Bluesky
   or Tangled follows to accounts that already use nibling.
diff --git a/packages/api/src/index/jetstream.ts b/packages/api/src/index/jetstream.ts
index 94f3e12..da33641 100644
--- a/packages/api/src/index/jetstream.ts
+++ b/packages/api/src/index/jetstream.ts
@@ -97,7 +97,7 @@ export function toPutEvent(
     case 'run.nibling.feed.like': {
       const result = safeParse(RunNiblingFeedLike.mainSchema, record);
       if (!result.ok) return undefined;
-      return { action: 'put', collection: 'run.nibling.feed.like', uri, cid, did, record: result.value };
+      return { action: 'put', collection: 'run.nibling.feed.like', uri, cid, did, record: result.value, indexedAt: now() };
     }
     case 'run.nibling.graph.follow': {
       const result = safeParse(RunNiblingGraphFollow.mainSchema, record);
diff --git a/packages/api/src/index/model.ts b/packages/api/src/index/model.ts
index 20d970f..5b5c2a5 100644
--- a/packages/api/src/index/model.ts
+++ b/packages/api/src/index/model.ts
@@ -35,7 +35,7 @@ interface ProfileRecord {
 /** One put or tombstone, already validated at the boundary (jetstream.ts, backfill.ts). */
 export type IndexEvent =
   | { action: 'put'; collection: 'run.nibling.project'; uri: string; cid: string; did: string; record: ProjectRecord; indexedAt: string }
-  | { action: 'put'; collection: 'run.nibling.feed.like'; uri: string; cid: string; did: string; record: LikeRecord }
+  | { action: 'put'; collection: 'run.nibling.feed.like'; uri: string; cid: string; did: string; record: LikeRecord; indexedAt: string }
   | { action: 'put'; collection: 'run.nibling.graph.follow'; uri: string; cid: string; did: string; record: FollowRecord }
   | { action: 'put'; collection: 'run.nibling.actor.profile'; uri: string; cid: string; did: string; record: ProfileRecord }
   | { action: 'delete'; collection: IndexCollection; uri: string };
@@ -89,7 +89,8 @@ const SCHEMA = [
     createdAt TEXT NOT NULL,
     updatedAt TEXT,
     sortAt TEXT NOT NULL,
-    indexedAt TEXT NOT NULL
+    indexedAt TEXT NOT NULL,
+    createdMs INTEGER NOT NULL
   )`,
   `CREATE INDEX IF NOT EXISTS projects_sortAt ON projects(sortAt DESC)`,
   `CREATE INDEX IF NOT EXISTS projects_forkOf ON projects(forkOf)`,
@@ -104,7 +105,7 @@ const SCHEMA = [
     uri TEXT PRIMARY KEY,
     did TEXT NOT NULL,
     subjectUri TEXT NOT NULL,
-    createdAtMs INTEGER NOT NULL
+    atMs INTEGER NOT NULL
   )`,
   `CREATE INDEX IF NOT EXISTS likes_subject ON likes(subjectUri)`,
   `CREATE INDEX IF NOT EXISTS likes_did ON likes(did)`,
@@ -126,6 +127,13 @@ const SCHEMA = [
   )`,
 ];
 
+/** When something happened, as far as the index will believe: the author's `createdAt`, but
+ * never later than when the index first saw it. Authors can backdate a record (which only ages
+ * it) but can't post-date one to stay fresh. */
+function observedMs(authorTime: string, indexedAt: string): number {
+  return Math.min(Date.parse(authorTime), Date.parse(indexedAt));
+}
+
 function encodeCursor(sortAt: string, uri: string): string {
   return Buffer.from(JSON.stringify([sortAt, uri])).toString('base64url');
 }
@@ -181,14 +189,14 @@ export class IndexModel {
     const { uri, cid, did, record, indexedAt } = event;
     const sortAt = record.updatedAt ?? record.createdAt;
     this.sql.exec(
-      `INSERT INTO projects (uri, cid, did, name, description, forkOf, createdAt, updatedAt, sortAt, indexedAt)
-       VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+      `INSERT INTO projects (uri, cid, did, name, description, forkOf, createdAt, updatedAt, sortAt, indexedAt, createdMs)
+       VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        ON CONFLICT(uri) DO UPDATE SET
          cid = excluded.cid, did = excluded.did, name = excluded.name, description = excluded.description,
          forkOf = excluded.forkOf, createdAt = excluded.createdAt, updatedAt = excluded.updatedAt,
-         sortAt = excluded.sortAt, indexedAt = excluded.indexedAt`,
+         sortAt = excluded.sortAt, indexedAt = excluded.indexedAt, createdMs = MIN(projects.createdMs, excluded.createdMs)`,
       uri, cid, did, record.name, record.description ?? null, record.forkOf?.uri ?? null,
-      record.createdAt, record.updatedAt ?? null, sortAt, indexedAt,
+      record.createdAt, record.updatedAt ?? null, sortAt, indexedAt, observedMs(record.createdAt, indexedAt),
     );
     this.sql.exec('DELETE FROM project_tags WHERE uri = ?', uri);
     for (const tag of record.tags ?? []) {
@@ -197,11 +205,11 @@ export class IndexModel {
   }
 
   private applyLike(event: PutEvent<'run.nibling.feed.like'>): void {
-    const { uri, did, record } = event;
+    const { uri, did, record, indexedAt } = event;
     this.sql.exec(
-      `INSERT INTO likes (uri, did, subjectUri, createdAtMs) VALUES (?, ?, ?, ?)
-       ON CONFLICT(uri) DO UPDATE SET did = excluded.did, subjectUri = excluded.subjectUri, createdAtMs = excluded.createdAtMs`,
-      uri, did, record.subject.uri, Date.parse(record.createdAt),
+      `INSERT INTO likes (uri, did, subjectUri, atMs) VALUES (?, ?, ?, ?)
+       ON CONFLICT(uri) DO UPDATE SET did = excluded.did, subjectUri = excluded.subjectUri, atMs = MIN(likes.atMs, excluded.atMs)`,
+      uri, did, record.subject.uri, observedMs(record.createdAt, indexedAt),
     );
   }
 
@@ -290,20 +298,20 @@ export class IndexModel {
     const rows = this.sql.exec<{ uri: string; score: number }>(
       `SELECT p.uri as uri,
          (
-           (SELECT COALESCE(SUM(MAX(0.0, ? - l.createdAtMs) / ?), 0) FROM likes l
-             WHERE l.subjectUri = p.uri AND l.createdAtMs >= ?)
+           (SELECT COALESCE(SUM(MAX(0.0, ? - (? - l.atMs)) / ?), 0) FROM likes l
+             WHERE l.subjectUri = p.uri AND l.atMs >= ?)
            +
-           (SELECT COALESCE(SUM(MAX(0.0, ? - ((julianday(f.createdAt) - 2440587.5) * 86400000.0)) / ?), 0)
-             FROM projects f WHERE f.forkOf = p.uri AND (julianday(f.createdAt) - 2440587.5) * 86400000.0 >= ?)
+           (SELECT COALESCE(SUM(MAX(0.0, ? - (? - f.createdMs)) / ?), 0) FROM projects f
+             WHERE f.forkOf = p.uri AND f.createdMs >= ?)
          ) as score
        FROM projects p
        ORDER BY score DESC, p.sortAt DESC, p.uri DESC`,
-      options.now, windowMs, windowStart,
-      options.now, windowMs, windowStart,
+      windowMs, options.now, windowMs, windowStart,
+      windowMs, options.now, windowMs, windowStart,
     );
     // Trending is re-ranked on every call (scores decay continuously), so there is no stable sort
     // key to resume from. The cursor is the rank offset into that call's ranking.
-    const offset = options.cursor ? Number.parseInt(options.cursor, 10) || 0 : 0;
+    const offset = options.cursor ? Math.max(0, Number.parseInt(options.cursor, 10) || 0) : 0;
     const page = rows.slice(offset, offset + limit);
     const projects = page.map((r) => this.getProjectView(r.uri)).filter((v): v is ProjectView => v !== undefined);
     const cursor = offset + limit < rows.length ? String(offset + limit) : undefined;
diff --git a/packages/api/tests/model.test.ts b/packages/api/tests/model.test.ts
index 6369956..977b1a0 100644
--- a/packages/api/tests/model.test.ts
+++ b/packages/api/tests/model.test.ts
@@ -29,6 +29,7 @@ function fixtureEvents(): IndexEvent[] {
       cid: 'bafy-like1',
       did: 'did:plc:bob',
       record: { subject: { uri: project1Uri, cid: 'bafy-project1' }, createdAt: '2026-11-06T00:00:00.000Z' },
+      indexedAt: '2026-11-06T00:00:01.000Z',
     },
     {
       action: 'put',
@@ -37,6 +38,7 @@ function fixtureEvents(): IndexEvent[] {
       cid: 'bafy-like2',
       did: 'did:plc:carol',
       record: { subject: { uri: project1Uri, cid: 'bafy-project1' }, createdAt: '2026-11-02T12:00:00.000Z' },
+      indexedAt: '2026-11-02T12:00:01.000Z',
     },
     {
       action: 'put',
@@ -188,3 +190,38 @@ test('known answers for 100 DIDs within the Durable Object bound-parameter limit
   assert.equal(dids.length, 100);
   assert.deepStrictEqual(model.known(dids), ['did:plc:alice', 'did:plc:erin']);
 });
+
+function project(rkey: string): IndexEvent {
+  return {
+    action: 'put',
+    collection: 'run.nibling.project',
+    uri: `at://did:plc:owner/run.nibling.project/${rkey}`,
+    cid: `bafy-${rkey}`,
+    did: 'did:plc:owner',
+    record: { name: rkey, createdAt: '2026-10-01T00:00:00.000Z' },
+    indexedAt: '2026-10-01T00:00:00.000Z',
+  };
+}
+
+function like(liker: string, rkey: string, subject: string, createdAt: string, indexedAt: string): IndexEvent {
+  return {
+    action: 'put',
+    collection: 'run.nibling.feed.like',
+    uri: `at://${liker}/run.nibling.feed.like/${rkey}`,
+    cid: `bafy-${rkey}`,
+    did: liker,
+    record: { subject: { uri: `at://did:plc:owner/run.nibling.project/${subject}`, cid: `bafy-${subject}` }, createdAt },
+    indexedAt,
+  };
+}
+
+test('trending: a fresh like outranks a 6-day-old one, and a future createdAt counts from when it was seen', () => {
+  const model = new IndexModel(sqliteSql());
+  for (const rkey of ['fresh', 'old', 'future']) model.apply(project(rkey));
+  model.apply(like('did:plc:a', 'l1', 'fresh', '2026-11-05T23:00:00.000Z', '2026-11-05T23:00:01.000Z'));
+  model.apply(like('did:plc:b', 'l2', 'old', '2026-10-31T00:00:00.000Z', '2026-10-31T00:00:01.000Z'));
+  model.apply(like('did:plc:c', 'l3', 'future', '2030-01-01T00:00:00.000Z', '2026-11-03T00:00:00.000Z'));
+
+  const { projects } = model.trending({ now: NOW });
+  assert.deepStrictEqual(projects.map((p) => p.name), ['fresh', 'future', 'old']);
+});
-- 
knotmirror 0.0.1


From 7e4316aea8e711f3a07222100e2c210bf6bdf747 Mon Sep 17 00:00:00 2001
From: Nate Moore <git@natemoo.re>
Date: Fri, 6 Nov 2026 01:37:37 -0500
Subject: [PATCH 07/10] Rank recent projects by server-seen time and count
 distinct accounts

A far-future updatedAt pinned a project to the top of getRecent. Sort
time is now clamped to when the index saw that version, and replaying
an unchanged version is a no-op. Like and fork counts, and trending
scores, count each account once per project.
---
 packages/api/README.md           |  4 ++-
 packages/api/src/index/model.ts  | 32 ++++++++++++------
 packages/api/tests/model.test.ts | 58 ++++++++++++++++++++++++++++++--
 3 files changed, 79 insertions(+), 15 deletions(-)

diff --git a/packages/api/README.md b/packages/api/README.md
index 9f8faab..646d538 100644
--- a/packages/api/README.md
+++ b/packages/api/README.md
@@ -8,7 +8,9 @@ holds that data as the source of truth.
 ## Routes
 
 - `GET /xrpc/run.nibling.feed.getRecent?limit=&cursor=&tag=` — recently created or updated
-  projects, newest first.
+  projects, newest first. A project sorts by its `updatedAt` (or `createdAt`), clamped to when
+  the index saw that version, so a far-future date can't pin it to the top. Like and fork counts
+  count distinct accounts.
 - `GET /xrpc/run.nibling.feed.getTrending?limit=&cursor=` — projects ranked by likes and forks
   from the last 7 days, decayed linearly to zero at the window edge. A like or fork counts from
   its `createdAt`, but never later than when the index first saw it, so a post-dated record
diff --git a/packages/api/src/index/model.ts b/packages/api/src/index/model.ts
index 5b5c2a5..d2f4f10 100644
--- a/packages/api/src/index/model.ts
+++ b/packages/api/src/index/model.ts
@@ -187,7 +187,9 @@ export class IndexModel {
 
   private applyProject(event: PutEvent<'run.nibling.project'>): void {
     const { uri, cid, did, record, indexedAt } = event;
-    const sortAt = record.updatedAt ?? record.createdAt;
+    const existing = this.sql.exec<{ cid: string }>('SELECT cid FROM projects WHERE uri = ?', uri);
+    if (existing[0]?.cid === cid) return;
+    const sortAt = new Date(observedMs(record.updatedAt ?? record.createdAt, indexedAt)).toISOString();
     this.sql.exec(
       `INSERT INTO projects (uri, cid, did, name, description, forkOf, createdAt, updatedAt, sortAt, indexedAt, createdMs)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
@@ -232,12 +234,12 @@ export class IndexModel {
   }
 
   private likeCount(uri: string): number {
-    const rows = this.sql.exec<{ n: number }>('SELECT COUNT(*) as n FROM likes WHERE subjectUri = ?', uri);
+    const rows = this.sql.exec<{ n: number }>('SELECT COUNT(DISTINCT did) as n FROM likes WHERE subjectUri = ?', uri);
     return rows[0]?.n ?? 0;
   }
 
   private forkCount(uri: string): number {
-    const rows = this.sql.exec<{ n: number }>('SELECT COUNT(*) as n FROM projects WHERE forkOf = ?', uri);
+    const rows = this.sql.exec<{ n: number }>('SELECT COUNT(DISTINCT did) as n FROM projects WHERE forkOf = ?', uri);
     return rows[0]?.n ?? 0;
   }
 
@@ -295,16 +297,24 @@ export class IndexModel {
     const windowStart = options.now - windowMs;
     // Linear decay to zero at the edge of the 7-day window: a like or fork from just now
     // scores 1, one from 7 days ago scores 0. Deterministic given `now`.
+    // Each account counts once per project: its freshest like, plus its freshest fork.
     const rows = this.sql.exec<{ uri: string; score: number }>(
-      `SELECT p.uri as uri,
-         (
-           (SELECT COALESCE(SUM(MAX(0.0, ? - (? - l.atMs)) / ?), 0) FROM likes l
-             WHERE l.subjectUri = p.uri AND l.atMs >= ?)
-           +
-           (SELECT COALESCE(SUM(MAX(0.0, ? - (? - f.createdMs)) / ?), 0) FROM projects f
-             WHERE f.forkOf = p.uri AND f.createdMs >= ?)
-         ) as score
+      `WITH like_scores AS (
+         SELECT uri, SUM(s) AS score FROM (
+           SELECT subjectUri AS uri, MAX(MAX(0.0, ? - (? - atMs)) / ?) AS s
+           FROM likes WHERE atMs >= ? GROUP BY subjectUri, did
+         ) GROUP BY uri
+       ),
+       fork_scores AS (
+         SELECT uri, SUM(s) AS score FROM (
+           SELECT forkOf AS uri, MAX(MAX(0.0, ? - (? - createdMs)) / ?) AS s
+           FROM projects WHERE forkOf IS NOT NULL AND createdMs >= ? GROUP BY forkOf, did
+         ) GROUP BY uri
+       )
+       SELECT p.uri AS uri, COALESCE(ls.score, 0) + COALESCE(fs.score, 0) AS score
        FROM projects p
+       LEFT JOIN like_scores ls ON ls.uri = p.uri
+       LEFT JOIN fork_scores fs ON fs.uri = p.uri
        ORDER BY score DESC, p.sortAt DESC, p.uri DESC`,
       windowMs, options.now, windowMs, windowStart,
       windowMs, options.now, windowMs, windowStart,
diff --git a/packages/api/tests/model.test.ts b/packages/api/tests/model.test.ts
index 977b1a0..be1c2f3 100644
--- a/packages/api/tests/model.test.ts
+++ b/packages/api/tests/model.test.ts
@@ -191,15 +191,19 @@ test('known answers for 100 DIDs within the Durable Object bound-parameter limit
   assert.deepStrictEqual(model.known(dids), ['did:plc:alice', 'did:plc:erin']);
 });
 
-function project(rkey: string): IndexEvent {
+function project(
+  rkey: string,
+  times: { createdAt?: string; updatedAt?: string; indexedAt?: string } = {},
+): IndexEvent {
+  const record = { name: rkey, createdAt: times.createdAt ?? '2026-10-01T00:00:00.000Z' };
   return {
     action: 'put',
     collection: 'run.nibling.project',
     uri: `at://did:plc:owner/run.nibling.project/${rkey}`,
     cid: `bafy-${rkey}`,
     did: 'did:plc:owner',
-    record: { name: rkey, createdAt: '2026-10-01T00:00:00.000Z' },
-    indexedAt: '2026-10-01T00:00:00.000Z',
+    record: times.updatedAt ? { ...record, updatedAt: times.updatedAt } : record,
+    indexedAt: times.indexedAt ?? '2026-10-01T00:00:00.000Z',
   };
 }
 
@@ -225,3 +229,51 @@ test('trending: a fresh like outranks a 6-day-old one, and a future createdAt co
   const { projects } = model.trending({ now: NOW });
   assert.deepStrictEqual(projects.map((p) => p.name), ['fresh', 'future', 'old']);
 });
+
+test('recent: a far-future updatedAt sorts by when the index saw it, not where the author put it', () => {
+  const model = new IndexModel(sqliteSql());
+  model.apply(project('pinned', { createdAt: '2026-11-01T00:00:00.000Z', updatedAt: '9999-12-31T00:00:00.000Z', indexedAt: '2026-11-02T00:00:00.000Z' }));
+  model.apply(project('honest', { createdAt: '2026-11-03T00:00:00.000Z', indexedAt: '2026-11-03T00:00:01.000Z' }));
+
+  const { projects } = model.recent();
+  assert.deepStrictEqual(projects.map((p) => [p.name, p.updatedAt]), [
+    ['honest', '2026-11-03T00:00:00.000Z'],
+    ['pinned', '2026-11-02T00:00:00.000Z'],
+  ]);
+});
+
+test('likeCount and forkCount count distinct accounts, not records', () => {
+  const model = new IndexModel(sqliteSql());
+  model.apply(project('target'));
+  for (const rkey of ['l1', 'l2', 'l3']) {
+    model.apply(like('did:plc:spam', rkey, 'target', '2026-11-05T00:00:00.000Z', '2026-11-05T00:00:01.000Z'));
+  }
+  model.apply(like('did:plc:fan', 'l4', 'target', '2026-11-05T00:00:00.000Z', '2026-11-05T00:00:01.000Z'));
+  for (const rkey of ['f1', 'f2']) {
+    model.apply({
+      action: 'put',
+      collection: 'run.nibling.project',
+      uri: `at://did:plc:spam/run.nibling.project/${rkey}`,
+      cid: `bafy-${rkey}`,
+      did: 'did:plc:spam',
+      record: { name: rkey, createdAt: '2026-11-05T00:00:00.000Z', forkOf: { uri: 'at://did:plc:owner/run.nibling.project/target', cid: 'bafy-target' } },
+      indexedAt: '2026-11-05T00:00:01.000Z',
+    });
+  }
+
+  const target = model.recent().projects.find((p) => p.name === 'target')!;
+  assert.deepStrictEqual({ likeCount: target.likeCount, forkCount: target.forkCount }, { likeCount: 2, forkCount: 1 });
+});
+
+test('trending: one account liking many times counts once', () => {
+  const model = new IndexModel(sqliteSql());
+  model.apply(project('spammed'));
+  model.apply(project('liked'));
+  for (const rkey of ['l1', 'l2', 'l3', 'l4', 'l5']) {
+    model.apply(like('did:plc:spam', rkey, 'spammed', '2026-11-05T00:00:00.000Z', '2026-11-05T00:00:01.000Z'));
+  }
+  model.apply(like('did:plc:a', 'l6', 'liked', '2026-11-05T00:00:00.000Z', '2026-11-05T00:00:01.000Z'));
+  model.apply(like('did:plc:b', 'l7', 'liked', '2026-11-05T00:00:00.000Z', '2026-11-05T00:00:01.000Z'));
+
+  assert.deepStrictEqual(model.trending({ now: NOW }).projects.map((p) => p.name), ['liked', 'spammed']);
+});
-- 
knotmirror 0.0.1


From 7ba7202cd17fc5e1c080cdb708cb2e7b37410db2 Mon Sep 17 00:00:00 2001
From: Nate Moore <git@natemoo.re>
Date: Fri, 6 Nov 2026 01:40:22 -0500
Subject: [PATCH 08/10] Bound every call the index makes to an untrusted server

Backfill paging stops on a repeated or empty cursor and has page caps.
A repo that answers bad JSON or a malformed page is skipped instead of
aborting the backfill. Relay, PDS, PLC, and Slingshot calls time out
after 10 seconds. PDS endpoints must be public https URLs, and did:web
documents are fetched from the path the DID names.
---
 packages/api/README.md              |  7 +++
 packages/api/src/identity.ts        | 29 +++++++++
 packages/api/src/index/backfill.ts  | 90 ++++++++++++++++----------
 packages/api/src/net.ts             | 29 +++++++++
 packages/api/src/router.ts          | 39 +++++++-----
 packages/api/src/worker.ts          | 13 +---
 packages/api/tests/backfill.test.ts | 97 ++++++++++++++++++++++++++++-
 packages/api/tests/net.test.ts      | 69 ++++++++++++++++++++
 8 files changed, 313 insertions(+), 60 deletions(-)
 create mode 100644 packages/api/src/identity.ts
 create mode 100644 packages/api/src/net.ts
 create mode 100644 packages/api/tests/net.test.ts

diff --git a/packages/api/README.md b/packages/api/README.md
index 646d538..97a6e93 100644
--- a/packages/api/README.md
+++ b/packages/api/README.md
@@ -67,6 +67,13 @@ Every 60 seconds, a Durable Object alarm (`Index.alarm`, `src/worker.ts`) runs o
 The first alarm is armed from the Durable Object's constructor (`blockConcurrencyWhile`), so the
 first request that wakes the object starts ingestion.
 
+Relays, PDSes, PLC, and Slingshot are untrusted. Every call to them has a 10-second timeout
+(`src/net.ts`). Listings stop on an empty or repeated cursor, and after 1000 relay pages or 100
+PDS pages per repo and collection. A repo that answers with an error or a malformed page is
+skipped, and its rows are not pruned. A PDS endpoint must be a public `https:` URL
+(`isPublicHttpsUrl`), and `did:web` documents are fetched from the path the DID names
+(`src/identity.ts`). Backfilled records must name the repo and collection that were listed.
+
 Every record is validated against its generated `@nibling/lexicons` schema at the boundary
 (`toPutEvent` in `src/index/jetstream.ts`, shared by the live tail and the backfill path); a
 record that fails validation is dropped, not indexed. `IndexModel.apply()` is idempotent — in
diff --git a/packages/api/src/identity.ts b/packages/api/src/identity.ts
new file mode 100644
index 0000000..8dc1613
--- /dev/null
+++ b/packages/api/src/identity.ts
@@ -0,0 +1,29 @@
+import { FETCH_TIMEOUT_MS, isPublicHttpsUrl } from './net.ts';
+
+/** Where a DID's document lives: `plc.directory` for `did:plc:`, and for `did:web:` the host
+ * (with `%3A` for a port) plus any `:`-separated path, per the did:web spec. `undefined` for any
+ * other method or a host that isn't safe to fetch. */
+export function didDocumentUrl(did: string): string | undefined {
+  if (/^did:plc:[a-z2-7]{24}$/.test(did)) return `https://plc.directory/${did}`;
+  if (!did.startsWith('did:web:')) return undefined;
+  const [host, ...path] = did.slice('did:web:'.length).split(':');
+  if (!host) return undefined;
+  const url = path.length > 0
+    ? `https://${decodeURIComponent(host)}/${path.map(decodeURIComponent).join('/')}/did.json`
+    : `https://${decodeURIComponent(host)}/.well-known/did.json`;
+  return isPublicHttpsUrl(url) ? url : undefined;
+}
+
+/** Resolves a DID to its PDS endpoint. Throws when the document can't be fetched or names no
+ * PDS, or names one that isn't a public `https:` URL. */
+export async function resolveDidPds(did: string, fetchFn: typeof fetch = fetch, timeoutMs = FETCH_TIMEOUT_MS): Promise<string> {
+  const docUrl = didDocumentUrl(did);
+  if (!docUrl) throw new Error(`cannot resolve ${did}`);
+  const res = await fetchFn(docUrl, { signal: AbortSignal.timeout(timeoutMs) });
+  if (!res.ok) throw new Error(`DID document for ${did}: HTTP ${res.status}`);
+  const doc = (await res.json()) as { service?: unknown };
+  const services = Array.isArray(doc.service) ? (doc.service as { type?: unknown; serviceEndpoint?: unknown }[]) : [];
+  const pds = services.find((s) => s?.type === 'AtprotoPersonalDataServer')?.serviceEndpoint;
+  if (typeof pds !== 'string' || !isPublicHttpsUrl(pds)) throw new Error(`no usable PDS in DID document for ${did}`);
+  return pds;
+}
diff --git a/packages/api/src/index/backfill.ts b/packages/api/src/index/backfill.ts
index 4172ea9..59f1087 100644
--- a/packages/api/src/index/backfill.ts
+++ b/packages/api/src/index/backfill.ts
@@ -1,5 +1,6 @@
 import type { IndexCollection, IndexEvent } from './model.ts';
 import { toPutEvent, WANTED_COLLECTIONS } from './jetstream.ts';
+import { FETCH_TIMEOUT_MS, isPublicHttpsUrl } from '../net.ts';
 
 /** Injected rather than `globalThis.fetch` so tests run with no network. */
 export interface BackfillDeps {
@@ -7,8 +8,15 @@ export interface BackfillDeps {
   fetch: typeof fetch;
   /** Resolves a DID to its PDS base URL (normally via `plc.directory` or `did:web`). */
   resolvePds(did: string): Promise<string>;
+  /** Per-request timeout. Default `FETCH_TIMEOUT_MS`. */
+  timeoutMs?: number;
 }
 
+/** Page caps, so a server that never stops handing out cursors can't hold the alarm forever:
+ * 1000 relay pages of 1000 repos, and 100 PDS pages of 100 records per repo and collection. */
+export const MAX_REPO_PAGES = 1000;
+export const MAX_RECORD_PAGES = 100;
+
 /** What a backfill produces. `repo` names every record one repo's PDS still holds in a
  * collection, so the index can drop rows deleted while it wasn't listening. It is emitted only
  * after that repo's listing finished, never after a partial one. */
@@ -16,30 +24,49 @@ export type BackfillItem =
   | { kind: 'event'; event: IndexEvent }
   | { kind: 'repo'; collection: IndexCollection; did: string; uris: ReadonlySet<string> };
 
-interface ListReposByCollectionResponse {
-  repos: { did: string }[];
-  cursor?: string;
+/** One page of a paged XRPC listing, or `undefined` if the request failed or the body isn't
+ * JSON. */
+async function getJson(url: URL, deps: BackfillDeps): Promise<unknown> {
+  try {
+    const res = await deps.fetch(url.toString(), { signal: AbortSignal.timeout(deps.timeoutMs ?? FETCH_TIMEOUT_MS) });
+    if (!res.ok) return undefined;
+    return await res.json();
+  } catch {
+    return undefined;
+  }
 }
 
-interface ListRecordsResponse {
-  records: { uri: string; cid: string; value: unknown }[];
-  cursor?: string;
+type Next = { kind: 'page'; cursor: string } | { kind: 'end' } | { kind: 'repeat' };
+
+/** Where a listing goes next. A missing or empty cursor ends it; a cursor the server already
+ * handed out means it would loop forever. */
+function nextCursor(body: { cursor?: unknown }, seen: Set<string>): Next {
+  const cursor = body.cursor;
+  if (typeof cursor !== 'string' || cursor === '') return { kind: 'end' };
+  if (seen.has(cursor)) return { kind: 'repeat' };
+  seen.add(cursor);
+  return { kind: 'page', cursor };
 }
 
-/** `com.atproto.sync.listReposByCollection` on the relay, paged to exhaustion. */
+/** `com.atproto.sync.listReposByCollection` on the relay, paged until it ends, repeats a
+ * cursor, or hits `MAX_REPO_PAGES`. */
 async function* listReposByCollection(collection: string, deps: BackfillDeps): AsyncGenerator<string> {
+  const seen = new Set<string>();
   let cursor: string | undefined;
-  do {
+  for (let page = 0; page < MAX_REPO_PAGES; page++) {
     const url = new URL('/xrpc/com.atproto.sync.listReposByCollection', deps.relayUrl);
     url.searchParams.set('collection', collection);
     url.searchParams.set('limit', '1000');
     if (cursor) url.searchParams.set('cursor', cursor);
-    const res = await deps.fetch(url.toString());
-    if (!res.ok) throw new Error(`listReposByCollection ${collection}: HTTP ${res.status}`);
-    const body = (await res.json()) as ListReposByCollectionResponse;
-    for (const { did } of body.repos) yield did;
-    cursor = body.cursor;
-  } while (cursor);
+    const body = (await getJson(url, deps)) as { repos?: unknown; cursor?: unknown } | undefined;
+    if (!body || !Array.isArray(body.repos)) throw new Error(`listReposByCollection ${collection}: bad response`);
+    for (const repo of body.repos as { did?: unknown }[]) {
+      if (typeof repo?.did === 'string' && repo.did.startsWith('did:')) yield repo.did;
+    }
+    const next = nextCursor(body, seen);
+    if (next.kind !== 'page') return;
+    cursor = next.cursor;
+  }
 }
 
 /** The record's URI rebuilt from the repo we asked for, or `undefined` when the PDS claims a
@@ -54,9 +81,10 @@ export function ownedRecordUri(uri: unknown, did: string, collection: string): s
   return `at://${did}/${collection}/${rkey}`;
 }
 
-/** `com.atproto.repo.listRecords` against one repo's own PDS, paged to exhaustion. A repo whose
- * PDS can't be reached is skipped rather than failing the whole backfill, and gets no `repo`
- * item, so none of its rows are pruned. */
+/** `com.atproto.repo.listRecords` against one repo's own PDS. A repo whose PDS can't be
+ * resolved, isn't a public `https:` URL, or answers with an error or a malformed page is skipped
+ * rather than failing the whole backfill. Only a listing that ended normally yields a `repo`
+ * item, so a skipped, looping, or capped repo has none of its rows pruned. */
 async function* listRecords(did: string, collection: IndexCollection, deps: BackfillDeps): AsyncGenerator<BackfillItem> {
   let pds: string;
   try {
@@ -64,32 +92,30 @@ async function* listRecords(did: string, collection: IndexCollection, deps: Back
   } catch {
     return;
   }
+  if (!isPublicHttpsUrl(pds)) return;
   const uris = new Set<string>();
+  const seen = new Set<string>();
   let cursor: string | undefined;
-  do {
+  for (let page = 0; page < MAX_RECORD_PAGES; page++) {
     const url = new URL('/xrpc/com.atproto.repo.listRecords', pds);
     url.searchParams.set('repo', did);
     url.searchParams.set('collection', collection);
     url.searchParams.set('limit', '100');
     if (cursor) url.searchParams.set('cursor', cursor);
-    let res: Response;
-    try {
-      res = await deps.fetch(url.toString());
-    } catch {
-      return;
-    }
-    if (!res.ok) return;
-    const body = (await res.json()) as ListRecordsResponse;
-    for (const record of body.records) {
-      const uri = ownedRecordUri(record.uri, did, collection);
-      if (!uri) continue;
+    const body = (await getJson(url, deps)) as { records?: unknown; cursor?: unknown } | undefined;
+    if (!body || !Array.isArray(body.records)) return;
+    for (const record of body.records as { uri?: unknown; cid?: unknown; value?: unknown }[]) {
+      const uri = ownedRecordUri(record?.uri, did, collection);
+      if (!uri || typeof record.cid !== 'string') continue;
       uris.add(uri);
       const event = toPutEvent(collection, uri, record.cid, did, record.value);
       if (event) yield { kind: 'event', event };
     }
-    cursor = body.cursor;
-  } while (cursor);
-  yield { kind: 'repo', collection, did, uris };
+    const next = nextCursor(body, seen);
+    if (next.kind === 'end') yield { kind: 'repo', collection, did, uris };
+    if (next.kind !== 'page') return;
+    cursor = next.cursor;
+  }
 }
 
 /** Rebuilds the index for one collection from the relay and each repo's PDS, used when the
diff --git a/packages/api/src/net.ts b/packages/api/src/net.ts
new file mode 100644
index 0000000..d34271b
--- /dev/null
+++ b/packages/api/src/net.ts
@@ -0,0 +1,29 @@
+/** How long any one call to a relay, PDS, PLC, or Slingshot may take. */
+export const FETCH_TIMEOUT_MS = 10_000;
+
+const PRIVATE_SUFFIXES = ['.localhost', '.internal', '.local'];
+
+function isPrivateIPv4(host: string): boolean {
+  const parts = host.split('.');
+  if (parts.length !== 4 || !parts.every((p) => /^\d{1,3}$/.test(p))) return false;
+  const [a, b] = parts.map(Number) as [number, number];
+  return a === 0 || a === 10 || a === 127 || (a === 169 && b === 254) || (a === 172 && b >= 16 && b <= 31) || (a === 192 && b === 168);
+}
+
+function isPrivateIPv6(host: string): boolean {
+  if (!host.startsWith('[')) return false;
+  const addr = host.slice(1, -1);
+  return addr === '::1' || addr === '::' || addr.startsWith('fc') || addr.startsWith('fd') || addr.startsWith('fe80:') || addr.startsWith('::ffff:');
+}
+
+/** Whether a PDS endpoint (or any URL named by a DID document) is safe to fetch: `https:`, and
+ * not a host that obviously points inside a network. `URL` canonicalizes IPv4 spellings such as
+ * `2130706433` or `0x7f.1` to dotted decimal, so the checks below see the real address. */
+export function isPublicHttpsUrl(value: string): boolean {
+  if (!URL.canParse(value)) return false;
+  const url = new URL(value);
+  if (url.protocol !== 'https:' || url.username || url.password) return false;
+  const host = url.hostname.toLowerCase().replace(/\.$/, '');
+  if (host === 'localhost' || PRIVATE_SUFFIXES.some((suffix) => host.endsWith(suffix))) return false;
+  return !isPrivateIPv4(host) && !isPrivateIPv6(host);
+}
diff --git a/packages/api/src/router.ts b/packages/api/src/router.ts
index 1476317..7daeac9 100644
--- a/packages/api/src/router.ts
+++ b/packages/api/src/router.ts
@@ -1,6 +1,7 @@
 import { renderEmbedPage, renderOEmbedJson, renderProjectPage, type ProjectCardData } from './cards.ts';
 import type { RecentOptions, TrendingOptions } from './index/model.ts';
 import { clientMetadata } from '@nibling/sdk/metadata';
+import { FETCH_TIMEOUT_MS, isPublicHttpsUrl } from './net.ts';
 
 /** What the router needs from the Index DO, kept to just the RPC surface it calls. Isolating
  * this (rather than importing `cloudflare:workers` types here) is what lets this file run under
@@ -42,26 +43,32 @@ function parseLimit(raw: string | null, fallback = 30, max = 100): number {
   return Math.min(n, max);
 }
 
-/** `blue.microcosm.identity.resolveMiniDoc` via Slingshot: handle or DID in, `{did, pds}` out. */
+/** One Slingshot query, or `undefined` if it errors, times out, or isn't JSON. */
+async function slingshot(fetchFn: typeof fetch, slingshotUrl: string, nsid: string, params: Record<string, string>): Promise<Record<string, unknown> | undefined> {
+  const url = new URL(`/xrpc/${nsid}`, slingshotUrl);
+  for (const [key, value] of Object.entries(params)) url.searchParams.set(key, value);
+  try {
+    const res = await fetchFn(url.toString(), { signal: AbortSignal.timeout(FETCH_TIMEOUT_MS) });
+    if (!res.ok) return undefined;
+    const body: unknown = await res.json();
+    return typeof body === 'object' && body !== null ? (body as Record<string, unknown>) : undefined;
+  } catch {
+    return undefined;
+  }
+}
+
+/** `blue.microcosm.identity.resolveMiniDoc`: handle or DID in, `{did, pds}` out. */
 async function resolveMiniDoc(fetchFn: typeof fetch, slingshotUrl: string, actor: string): Promise<{ did: string; pds: string } | undefined> {
-  const url = new URL('/xrpc/blue.microcosm.identity.resolveMiniDoc', slingshotUrl);
-  url.searchParams.set('identifier', actor);
-  const res = await fetchFn(url.toString());
-  if (!res.ok) return undefined;
-  const body = (await res.json()) as { did?: string; pds?: string };
-  if (!body.did || !body.pds) return undefined;
+  const body = await slingshot(fetchFn, slingshotUrl, 'blue.microcosm.identity.resolveMiniDoc', { identifier: actor });
+  if (typeof body?.did !== 'string' || typeof body.pds !== 'string') return undefined;
   return { did: body.did, pds: body.pds };
 }
 
-/** `blue.microcosm.repo.getRecordByUri` via Slingshot. */
+/** `blue.microcosm.repo.getRecordByUri`. */
 async function getRecordByUri(fetchFn: typeof fetch, slingshotUrl: string, uri: string): Promise<{ cid: string; value: Record<string, unknown> } | undefined> {
-  const url = new URL('/xrpc/blue.microcosm.repo.getRecordByUri', slingshotUrl);
-  url.searchParams.set('at_uri', uri);
-  const res = await fetchFn(url.toString());
-  if (!res.ok) return undefined;
-  const body = (await res.json()) as { cid?: string; value?: Record<string, unknown> };
-  if (!body.cid || !body.value) return undefined;
-  return { cid: body.cid, value: body.value };
+  const body = await slingshot(fetchFn, slingshotUrl, 'blue.microcosm.repo.getRecordByUri', { at_uri: uri });
+  if (typeof body?.cid !== 'string' || typeof body.value !== 'object' || body.value === null) return undefined;
+  return { cid: body.cid, value: body.value as Record<string, unknown> };
 }
 
 async function loadProjectCard(env: RouterEnv, actor: string, rkey: string): Promise<ProjectCardData | undefined> {
@@ -75,7 +82,7 @@ async function loadProjectCard(env: RouterEnv, actor: string, rkey: string): Pro
   const name = typeof value.name === 'string' ? value.name : rkey;
   const description = typeof value.description === 'string' ? value.description : undefined;
   const thumbnail = value.thumbnail as { ref?: { $link?: string } } | undefined;
-  const thumbnailUrl = thumbnail?.ref?.$link
+  const thumbnailUrl = thumbnail?.ref?.$link && isPublicHttpsUrl(identity.pds)
     ? `${identity.pds}/xrpc/com.atproto.sync.getBlob?did=${identity.did}&cid=${thumbnail.ref.$link}`
     : undefined;
   const card: ProjectCardData = { actor, did: identity.did, rkey, name };
diff --git a/packages/api/src/worker.ts b/packages/api/src/worker.ts
index 8978fb3..c191608 100644
--- a/packages/api/src/worker.ts
+++ b/packages/api/src/worker.ts
@@ -3,6 +3,7 @@ import { ensureAlarm, runAndRearm } from './alarm.ts';
 import { ingest, type WebSocketLike } from './index/ingest.ts';
 import { IndexModel, type RecentOptions, type TrendingOptions } from './index/model.ts';
 import { handleRequest } from './router.ts';
+import { resolveDidPds } from './identity.ts';
 
 export interface Env {
   INDEX: DurableObjectNamespace<Index>;
@@ -65,19 +66,9 @@ export class Index extends DurableObject<Env> {
       backfill: {
         relayUrl: this.env.RELAY_URL,
         fetch,
-        resolvePds: async (did) => resolveDidPds(did),
+        resolvePds: (did) => resolveDidPds(did),
       },
       now: () => Date.now(),
     });
   }
 }
-
-/** Resolves a DID's PDS endpoint via `plc.directory` (for `did:plc:`) or the `did:web` document. */
-async function resolveDidPds(did: string): Promise<string> {
-  const doc: { service?: { id: string; type: string; serviceEndpoint: string }[] } = did.startsWith('did:web:')
-    ? await (await fetch(`https://${decodeURIComponent(did.slice('did:web:'.length))}/.well-known/did.json`)).json()
-    : await (await fetch(`https://plc.directory/${did}`)).json();
-  const pds = doc.service?.find((s) => s.type === 'AtprotoPersonalDataServer');
-  if (!pds) throw new Error(`no PDS service in DID document for ${did}`);
-  return pds.serviceEndpoint;
-}
diff --git a/packages/api/tests/backfill.test.ts b/packages/api/tests/backfill.test.ts
index ce2c277..ae0e0b1 100644
--- a/packages/api/tests/backfill.test.ts
+++ b/packages/api/tests/backfill.test.ts
@@ -1,6 +1,6 @@
 import assert from 'node:assert/strict';
 import { test } from 'node:test';
-import { backfillCollection, type BackfillDeps, type BackfillItem } from '../src/index/backfill.ts';
+import { backfillCollection, MAX_RECORD_PAGES, type BackfillDeps, type BackfillItem } from '../src/index/backfill.ts';
 
 async function collect(items: AsyncIterable<BackfillItem>): Promise<BackfillItem[]> {
   const out: BackfillItem[] = [];
@@ -111,3 +111,98 @@ test('backfillCollection drops records whose URI names another repo or collectio
     [{ uri: 'at://did:plc:evil/run.nibling.project/3own', did: 'did:plc:evil' }, 'repo'],
   );
 });
+
+function relayThen(listRecords: (url: URL, init?: RequestInit) => Response | Promise<Response>, dids = ['did:plc:alice']): typeof fetch {
+  return (async (input: string | URL | Request, init?: RequestInit) => {
+    const url = new URL(typeof input === 'string' ? input : input.toString());
+    if (url.pathname === '/xrpc/com.atproto.sync.listReposByCollection') return jsonResponse({ repos: dids.map((did) => ({ did })) });
+    return listRecords(url, init);
+  }) as typeof fetch;
+}
+
+const aliceRecord = (rkey: string) => ({
+  uri: `at://did:plc:alice/run.nibling.project/${rkey}`,
+  cid: `bafy-${rkey}`,
+  value: { $type: 'run.nibling.project', name: rkey, files: [], createdAt: '2026-11-01T00:00:00.000Z' },
+});
+
+test('a PDS that repeats a cursor stops the listing, and the repo is not pruned', async () => {
+  let calls = 0;
+  const deps: BackfillDeps = {
+    relayUrl: 'https://relay.example.com',
+    fetch: relayThen(() => {
+      calls++;
+      return jsonResponse({ records: [aliceRecord('3la1')], cursor: 'same' });
+    }),
+    resolvePds: async () => 'https://alice.pds.example',
+  };
+  const items = await collect(backfillCollection('run.nibling.project', deps));
+  assert.equal(calls, 2);
+  assert.deepStrictEqual(items.map((item) => item.kind), ['event', 'event']);
+});
+
+test('a PDS that never stops paging is cut off at MAX_RECORD_PAGES', async () => {
+  let calls = 0;
+  const deps: BackfillDeps = {
+    relayUrl: 'https://relay.example.com',
+    fetch: relayThen(() => jsonResponse({ records: [], cursor: `page${++calls}` })),
+    resolvePds: async () => 'https://alice.pds.example',
+  };
+  const items = await collect(backfillCollection('run.nibling.project', deps));
+  assert.equal(calls, MAX_RECORD_PAGES);
+  assert.deepStrictEqual(items, []);
+});
+
+test('a relay that repeats a cursor stops listing repos', async () => {
+  let calls = 0;
+  const deps: BackfillDeps = {
+    relayUrl: 'https://relay.example.com',
+    fetch: (async () => {
+      calls++;
+      return jsonResponse({ repos: [], cursor: 'same' });
+    }) as typeof fetch,
+    resolvePds: async () => 'https://alice.pds.example',
+  };
+  assert.deepStrictEqual(await collect(backfillCollection('run.nibling.project', deps)), []);
+  assert.equal(calls, 2);
+});
+
+test('a repo answering bad JSON or a non-array is skipped, and the next repo is still listed', async () => {
+  const deps: BackfillDeps = {
+    relayUrl: 'https://relay.example.com',
+    fetch: relayThen((url) => {
+      const repo = url.searchParams.get('repo');
+      if (repo === 'did:plc:broken') return new Response('<html>oops</html>');
+      if (repo === 'did:plc:weird') return jsonResponse({ records: 'nope' });
+      return jsonResponse({ records: [aliceRecord('3la1')] });
+    }, ['did:plc:broken', 'did:plc:weird', 'did:plc:alice']),
+    resolvePds: async () => 'https://pds.example',
+  };
+  const items = await collect(backfillCollection('run.nibling.project', deps));
+  assert.deepStrictEqual(eventUris(items), ['at://did:plc:alice/run.nibling.project/3la1']);
+  assert.deepStrictEqual(items.flatMap((item) => (item.kind === 'repo' ? [item.did] : [])), ['did:plc:alice']);
+});
+
+test('a PDS that hangs is abandoned after the timeout', async () => {
+  const deps: BackfillDeps = {
+    relayUrl: 'https://relay.example.com',
+    fetch: relayThen((_url, init) => new Promise((_, reject) => init?.signal?.addEventListener('abort', () => reject(init.signal!.reason)))),
+    resolvePds: async () => 'https://alice.pds.example',
+    timeoutMs: 20,
+  };
+  assert.deepStrictEqual(await collect(backfillCollection('run.nibling.project', deps)), []);
+});
+
+test('a repo whose PDS is not a public https URL is never fetched', async () => {
+  const fetched: string[] = [];
+  const deps: BackfillDeps = {
+    relayUrl: 'https://relay.example.com',
+    fetch: relayThen((url) => {
+      fetched.push(url.origin);
+      return jsonResponse({ records: [aliceRecord('3la1')] });
+    }),
+    resolvePds: async () => 'http://169.254.169.254',
+  };
+  assert.deepStrictEqual(await collect(backfillCollection('run.nibling.project', deps)), []);
+  assert.deepStrictEqual(fetched, []);
+});
diff --git a/packages/api/tests/net.test.ts b/packages/api/tests/net.test.ts
new file mode 100644
index 0000000..0b34b9f
--- /dev/null
+++ b/packages/api/tests/net.test.ts
@@ -0,0 +1,69 @@
+import assert from 'node:assert/strict';
+import { test } from 'node:test';
+import { isPublicHttpsUrl } from '../src/net.ts';
+import { didDocumentUrl, resolveDidPds } from '../src/identity.ts';
+
+const ENDPOINTS: [string, boolean][] = [
+  ['https://puffball.us-east.host.bsky.network', true],
+  ['https://pds.example.com:8443/', true],
+  ['https://172.32.0.1', true],
+  ['http://pds.example.com', false],
+  ['ftp://pds.example.com', false],
+  ['not a url', false],
+  ['https://user:pass@pds.example.com', false],
+  ['https://localhost', false],
+  ['https://LOCALHOST.', false],
+  ['https://api.localhost', false],
+  ['https://metadata.google.internal', false],
+  ['https://printer.local', false],
+  ['https://127.0.0.1', false],
+  ['https://127.255.0.9', false],
+  ['https://2130706433', false],
+  ['https://0x7f.1', false],
+  ['https://0.0.0.0', false],
+  ['https://10.1.2.3', false],
+  ['https://172.16.0.1', false],
+  ['https://172.31.255.255', false],
+  ['https://192.168.1.1', false],
+  ['https://169.254.169.254', false],
+  ['https://[::1]', false],
+  ['https://[fd00::1]', false],
+  ['https://[fe80::1]', false],
+  ['https://[::ffff:127.0.0.1]', false],
+  ['https://[2606:4700::1111]', true],
+];
+
+for (const [endpoint, expected] of ENDPOINTS) {
+  test(`isPublicHttpsUrl(${JSON.stringify(endpoint)}) is ${expected}`, () => {
+    assert.equal(isPublicHttpsUrl(endpoint), expected);
+  });
+}
+
+const DID_DOCUMENTS: [string, string | undefined][] = [
+  ['did:plc:z72i7hdynmk6r22z27h6tvur', 'https://plc.directory/did:plc:z72i7hdynmk6r22z27h6tvur'],
+  ['did:plc:../../etc', undefined],
+  ['did:web:example.com', 'https://example.com/.well-known/did.json'],
+  ['did:web:example.com:user:alice', 'https://example.com/user/alice/did.json'],
+  ['did:web:example.com%3A8443', 'https://example.com:8443/.well-known/did.json'],
+  ['did:web:localhost%3A8080', undefined],
+  ['did:web:10.0.0.1', undefined],
+  ['did:key:z6Mk', undefined],
+];
+
+for (const [did, expected] of DID_DOCUMENTS) {
+  test(`didDocumentUrl(${did}) is ${expected}`, () => {
+    assert.equal(didDocumentUrl(did), expected);
+  });
+}
+
+test('resolveDidPds refuses a DID document that names a private PDS', async () => {
+  const fakeFetch = (async () =>
+    Response.json({ service: [{ id: '#atproto_pds', type: 'AtprotoPersonalDataServer', serviceEndpoint: 'http://169.254.169.254' }] })) as typeof fetch;
+  await assert.rejects(resolveDidPds('did:web:example.com', fakeFetch), /no usable PDS/);
+});
+
+test('resolveDidPds returns a public https PDS', async () => {
+  const fakeFetch = (async () =>
+    Response.json({ service: [{ id: '#atproto_pds', type: 'AtprotoPersonalDataServer', serviceEndpoint: 'https://pds.example.com' }] })) as typeof fetch;
+  assert.equal(await resolveDidPds('did:web:example.com', fakeFetch), 'https://pds.example.com');
+});
-- 
knotmirror 0.0.1


From a71b2b73814b8a6e5f2a3e15696d24f19dc4a872 Mon Sep 17 00:00:00 2001
From: Nate Moore <git@natemoo.re>
Date: Fri, 6 Nov 2026 01:41:24 -0500
Subject: [PATCH 09/10] Test the Jetstream parser against captured v2 frames

The parser and ingest tests used hand-built frames shaped like the
parser's own assumptions. They now load frames captured from the live
v2 endpoint: like create and delete, identity, and account, plus
run.nibling.* copies that keep the captured envelope.
---
 packages/api/README.md                        |   5 +
 .../api/tests/fixtures/jetstream/account.json |   1 +
 .../jetstream/app-bsky-like-create.json       |   1 +
 .../jetstream/app-bsky-like-delete.json       |   1 +
 .../tests/fixtures/jetstream/identity.json    |   1 +
 .../tests/fixtures/jetstream/like-create.json |   1 +
 .../tests/fixtures/jetstream/like-delete.json |   1 +
 .../fixtures/jetstream/project-create.json    |   1 +
 packages/api/tests/ingest.test.ts             |  26 ++--
 packages/api/tests/jetstream.test.ts          | 112 ++++++++++--------
 10 files changed, 83 insertions(+), 67 deletions(-)
 create mode 100644 packages/api/tests/fixtures/jetstream/account.json
 create mode 100644 packages/api/tests/fixtures/jetstream/app-bsky-like-create.json
 create mode 100644 packages/api/tests/fixtures/jetstream/app-bsky-like-delete.json
 create mode 100644 packages/api/tests/fixtures/jetstream/identity.json
 create mode 100644 packages/api/tests/fixtures/jetstream/like-create.json
 create mode 100644 packages/api/tests/fixtures/jetstream/like-delete.json
 create mode 100644 packages/api/tests/fixtures/jetstream/project-create.json

diff --git a/packages/api/README.md b/packages/api/README.md
index 97a6e93..112818d 100644
--- a/packages/api/README.md
+++ b/packages/api/README.md
@@ -116,6 +116,11 @@ cards, jetstream-parsing, backfill, and ingest-loop tests all take their externa
 (the index stub, `fetch`, the WebSocket, `resolvePds`, the clock) as injected parameters, so they
 also run under plain Node with no network and no Workers runtime.
 
+The Jetstream parser and ingest tests run on frames captured from the live v2 endpoint
+(`tests/fixtures/jetstream/`): commit create and delete, identity, and account. Since nobody
+writes `run.nibling.*` yet, the `run.nibling.*` fixtures keep a captured envelope and change only
+`collection` and `record`.
+
 A smoke run of `wrangler dev --local` confirmed `GET /oauth/client-metadata.json` (200, the exact
 D19 scope string) and `GET /xrpc/run.nibling.actor.getKnown?dids=did:plc:test` (200,
 `{"dids":[]}`, with CORS and cache headers) both serve correctly end to end.
diff --git a/packages/api/tests/fixtures/jetstream/account.json b/packages/api/tests/fixtures/jetstream/account.json
new file mode 100644
index 0000000..3e9dbc2
--- /dev/null
+++ b/packages/api/tests/fixtures/jetstream/account.json
@@ -0,0 +1 @@
+{"$type":"message","payload":{"$type":"network.bsky.jetstream.subscribeEvents#account","account":{"active":true,"did":"did:plc:iblx2oxf3ebuxwgvej42gy2f","seq":35251014246,"time":"2026-11-06T06:23:31.107Z"},"did":"did:plc:iblx2oxf3ebuxwgvej42gy2f","seq":26874579015,"time":"2026-11-06T06:23:31.391400Z","witnessedAt":"2026-11-06T06:23:31.391400Z"}}
diff --git a/packages/api/tests/fixtures/jetstream/app-bsky-like-create.json b/packages/api/tests/fixtures/jetstream/app-bsky-like-create.json
new file mode 100644
index 0000000..d700572
--- /dev/null
+++ b/packages/api/tests/fixtures/jetstream/app-bsky-like-create.json
@@ -0,0 +1 @@
+{"$type":"message","payload":{"$type":"network.bsky.jetstream.subscribeEvents#commit","cid":"bafyreihjpamtabgxzbb3r3ghmdx23ruqwztqgprpqppkzsnzvnldrqa5fa","collection":"app.bsky.feed.like","did":"did:plc:6mjqiiyd6crl57wl3pfypqjr","operation":"create","record":{"$type":"app.bsky.feed.like","createdAt":"2026-11-06T06:23:21.775Z","subject":{"cid":"bafyreic3jcnyd4qzzxn6mipk3do2be4efi7u43kcmr4ubrk7h2u5larpse","uri":"at://did:plc:rfqxrozvm5in5cyw2vr3rnej/app.bsky.feed.post/3mzmpslzg722e"}},"rev":"3mzmqn3dbqz2c","rkey":"3mzmqn3cvzz2c","seq":26874575837,"time":"2026-11-06T06:23:21.496870Z","witnessedAt":"2026-11-06T06:23:21.496870Z"}}
diff --git a/packages/api/tests/fixtures/jetstream/app-bsky-like-delete.json b/packages/api/tests/fixtures/jetstream/app-bsky-like-delete.json
new file mode 100644
index 0000000..3f85103
--- /dev/null
+++ b/packages/api/tests/fixtures/jetstream/app-bsky-like-delete.json
@@ -0,0 +1 @@
+{"$type":"message","payload":{"$type":"network.bsky.jetstream.subscribeEvents#commit","collection":"app.bsky.feed.like","did":"did:plc:zpmeobhmhfhz4clqwknac5gp","operation":"delete","rev":"3mzmqn3hwav2k","rkey":"3mzmqmt2eb72k","seq":26874575891,"time":"2026-11-06T06:23:21.663408Z","witnessedAt":"2026-11-06T06:23:21.663408Z"}}
diff --git a/packages/api/tests/fixtures/jetstream/identity.json b/packages/api/tests/fixtures/jetstream/identity.json
new file mode 100644
index 0000000..b4c8165
--- /dev/null
+++ b/packages/api/tests/fixtures/jetstream/identity.json
@@ -0,0 +1 @@
+{"$type":"message","payload":{"$type":"network.bsky.jetstream.subscribeEvents#identity","did":"did:plc:iblx2oxf3ebuxwgvej42gy2f","identity":{"did":"did:plc:iblx2oxf3ebuxwgvej42gy2f","seq":35251014245,"time":"2026-11-06T06:23:31.107Z"},"seq":26874578981,"time":"2026-11-06T06:23:31.362139Z","witnessedAt":"2026-11-06T06:23:31.362139Z"}}
diff --git a/packages/api/tests/fixtures/jetstream/like-create.json b/packages/api/tests/fixtures/jetstream/like-create.json
new file mode 100644
index 0000000..ad4a62c
--- /dev/null
+++ b/packages/api/tests/fixtures/jetstream/like-create.json
@@ -0,0 +1 @@
+{"$type":"message","payload":{"$type":"network.bsky.jetstream.subscribeEvents#commit","cid":"bafyreihjpamtabgxzbb3r3ghmdx23ruqwztqgprpqppkzsnzvnldrqa5fa","collection":"run.nibling.feed.like","did":"did:plc:6mjqiiyd6crl57wl3pfypqjr","operation":"create","record":{"$type":"run.nibling.feed.like","createdAt":"2026-11-06T06:23:21.775Z","subject":{"cid":"bafyreic3jcnyd4qzzxn6mipk3do2be4efi7u43kcmr4ubrk7h2u5larpse","uri":"at://did:plc:rfqxrozvm5in5cyw2vr3rnej/run.nibling.project/3mzmpslzg722e"}},"rev":"3mzmqn3dbqz2c","rkey":"3mzmqn3cvzz2c","seq":26874575837,"time":"2026-11-06T06:23:21.496870Z","witnessedAt":"2026-11-06T06:23:21.496870Z"}}
diff --git a/packages/api/tests/fixtures/jetstream/like-delete.json b/packages/api/tests/fixtures/jetstream/like-delete.json
new file mode 100644
index 0000000..5b998bd
--- /dev/null
+++ b/packages/api/tests/fixtures/jetstream/like-delete.json
@@ -0,0 +1 @@
+{"$type":"message","payload":{"$type":"network.bsky.jetstream.subscribeEvents#commit","collection":"run.nibling.feed.like","did":"did:plc:zpmeobhmhfhz4clqwknac5gp","operation":"delete","rev":"3mzmqn3hwav2k","rkey":"3mzmqmt2eb72k","seq":26874575891,"time":"2026-11-06T06:23:21.663408Z","witnessedAt":"2026-11-06T06:23:21.663408Z"}}
diff --git a/packages/api/tests/fixtures/jetstream/project-create.json b/packages/api/tests/fixtures/jetstream/project-create.json
new file mode 100644
index 0000000..4a3b612
--- /dev/null
+++ b/packages/api/tests/fixtures/jetstream/project-create.json
@@ -0,0 +1 @@
+{"$type":"message","payload":{"$type":"network.bsky.jetstream.subscribeEvents#commit","cid":"bafyreihjpamtabgxzbb3r3ghmdx23ruqwztqgprpqppkzsnzvnldrqa5fa","collection":"run.nibling.project","did":"did:plc:6mjqiiyd6crl57wl3pfypqjr","operation":"create","record":{"$type":"run.nibling.project","createdAt":"2026-11-06T06:23:21.775Z","files":[],"name":"Captured envelope"},"rev":"3mzmqn3dbqz2c","rkey":"3mzmqn3cvzz2c","seq":26874575837,"time":"2026-11-06T06:23:21.496870Z","witnessedAt":"2026-11-06T06:23:21.496870Z"}}
diff --git a/packages/api/tests/ingest.test.ts b/packages/api/tests/ingest.test.ts
index 59b600e..d763ecc 100644
--- a/packages/api/tests/ingest.test.ts
+++ b/packages/api/tests/ingest.test.ts
@@ -1,4 +1,6 @@
+/// <reference types="node" />
 import assert from 'node:assert/strict';
+import { readFileSync } from 'node:fs';
 import { test } from 'node:test';
 import { IndexModel } from '../src/index/model.ts';
 import { ingest, type IngestDeps, type WebSocketLike } from '../src/index/ingest.ts';
@@ -32,24 +34,14 @@ function fakeSocket(frames: FakeFrame[], urls: string[]): (url: string) => WebSo
   };
 }
 
+const CAPTURED = readFileSync(`${import.meta.dirname}/fixtures/jetstream/project-create.json`, 'utf8');
+
+/** The captured project-create frame (see jetstream.test.ts), re-timed and re-keyed. */
 function commitFrame(seq: number, timeIso: string): FakeFrame {
-  return {
-    type: 'message',
-    data: JSON.stringify({
-      $type: 'message',
-      payload: {
-        $type: 'network.bsky.jetstream.subscribeEvents#commit',
-        did: 'did:plc:alice',
-        seq,
-        time: timeIso,
-        operation: 'create',
-        collection: 'run.nibling.project',
-        rkey: `3la${seq}`,
-        cid: `bafy${seq}`,
-        record: { $type: 'run.nibling.project', name: `Project ${seq}`, files: [], createdAt: timeIso },
-      },
-    }),
-  };
+  const frame = JSON.parse(CAPTURED);
+  Object.assign(frame.payload, { seq, time: timeIso, witnessedAt: timeIso, rkey: `3la${seq}`, cid: `bafy${seq}`, did: 'did:plc:alice' });
+  frame.payload.record.createdAt = timeIso;
+  return { type: 'message', data: JSON.stringify(frame) };
 }
 
 const NOW = Date.parse('2026-11-06T00:00:10.000Z');
diff --git a/packages/api/tests/jetstream.test.ts b/packages/api/tests/jetstream.test.ts
index 92fc379..9a0cea2 100644
--- a/packages/api/tests/jetstream.test.ts
+++ b/packages/api/tests/jetstream.test.ts
@@ -1,25 +1,22 @@
+/// <reference types="node" />
 import assert from 'node:assert/strict';
+import { readFileSync } from 'node:fs';
 import { test } from 'node:test';
 import { parseJetstreamMessage, subscribeUrl, WANTED_COLLECTIONS } from '../src/index/jetstream.ts';
 
-function commitFrame(overrides: Record<string, unknown> = {}) {
-  return {
-    $type: 'message',
-    payload: {
-      $type: 'network.bsky.jetstream.subscribeEvents#commit',
-      did: 'did:plc:alice',
-      seq: 24664288881,
-      time: '2026-11-06T06:47:43.959305Z',
-      operation: 'create',
-      collection: 'run.nibling.project',
-      rkey: '3la1',
-      cid: 'bafy-project1',
-      record: { $type: 'run.nibling.project', name: 'Alice Project', files: [], createdAt: '2026-11-06T06:47:44.859Z' },
-      ...overrides,
-    },
-  };
+/**
+ * Frames captured read-only from wss://jetstream.us-east.bsky.network/xrpc/
+ * network.bsky.jetstream.subscribeEvents on 2026-11-06. `app-bsky-*`, `identity`, and `account`
+ * are byte-for-byte as received. `like-create`, `like-delete`, and `project-create` keep a
+ * captured envelope and change only `collection` and `record` to run.nibling.* shapes, since
+ * nobody writes run.nibling.* yet.
+ */
+function fixture(name: string): { $type: string; payload: Record<string, unknown> } {
+  return JSON.parse(readFileSync(`${import.meta.dirname}/fixtures/jetstream/${name}.json`, 'utf8'));
 }
 
+const NOW = () => '2026-11-06T06:23:22.000Z';
+
 test('subscribeUrl builds the v2 live-tail URL with repeated collections and kinds=commit', () => {
   const url = new URL(subscribeUrl('wss://jetstream.us-east.bsky.network'));
   assert.equal(url.pathname, '/xrpc/network.bsky.jetstream.subscribeEvents');
@@ -29,54 +26,69 @@ test('subscribeUrl builds the v2 live-tail URL with repeated collections and kin
 });
 
 test('subscribeUrl threads the cursor through when given one', () => {
-  const url = new URL(subscribeUrl('wss://jetstream.us-east.bsky.network', '24664288881'));
-  assert.equal(url.searchParams.get('cursor'), '24664288881');
+  const url = new URL(subscribeUrl('wss://jetstream.us-east.bsky.network', '1793946099886000'));
+  assert.equal(url.searchParams.get('cursor'), '1793946099886000');
 });
 
-test('parses a valid run.nibling.project create into a put event', () => {
-  const event = parseJetstreamMessage(commitFrame());
-  assert.ok(event);
-  assert.equal(event.action, 'put');
-  assert.equal(event.collection, 'run.nibling.project');
-  if (event.action === 'put' && event.collection === 'run.nibling.project') {
-    assert.equal(event.uri, 'at://did:plc:alice/run.nibling.project/3la1');
-    assert.equal(event.cid, 'bafy-project1');
-    assert.equal(event.did, 'did:plc:alice');
-    assert.equal(event.record.name, 'Alice Project');
-  }
+test('parses a captured like create into a put event', () => {
+  assert.deepStrictEqual(parseJetstreamMessage(fixture('like-create'), NOW), {
+    action: 'put',
+    collection: 'run.nibling.feed.like',
+    uri: 'at://did:plc:6mjqiiyd6crl57wl3pfypqjr/run.nibling.feed.like/3mzmqn3cvzz2c',
+    cid: 'bafyreihjpamtabgxzbb3r3ghmdx23ruqwztqgprpqppkzsnzvnldrqa5fa',
+    did: 'did:plc:6mjqiiyd6crl57wl3pfypqjr',
+    record: {
+      $type: 'run.nibling.feed.like',
+      createdAt: '2026-11-06T06:23:21.775Z',
+      subject: {
+        cid: 'bafyreic3jcnyd4qzzxn6mipk3do2be4efi7u43kcmr4ubrk7h2u5larpse',
+        uri: 'at://did:plc:rfqxrozvm5in5cyw2vr3rnej/run.nibling.project/3mzmpslzg722e',
+      },
+    },
+    indexedAt: '2026-11-06T06:23:22.000Z',
+  });
 });
 
-test('drops a record that fails lexicon validation', () => {
-  const frame = commitFrame({ record: { $type: 'run.nibling.project', createdAt: '2026-11-06T06:47:44.859Z' } }); // missing required `name`
-  assert.equal(parseJetstreamMessage(frame), undefined);
+test('parses a captured project create into a put event', () => {
+  const event = parseJetstreamMessage(fixture('project-create'), NOW);
+  assert.ok(event && event.action === 'put' && event.collection === 'run.nibling.project');
+  assert.equal(event.uri, 'at://did:plc:6mjqiiyd6crl57wl3pfypqjr/run.nibling.project/3mzmqn3cvzz2c');
+  assert.equal(event.record.name, 'Captured envelope');
 });
 
-test('drops a collection it does not care about', () => {
-  const frame = commitFrame({ collection: 'app.bsky.feed.post', rkey: 'xyz' });
-  assert.equal(parseJetstreamMessage(frame), undefined);
+test('a captured delete carries no record and needs no validation', () => {
+  assert.deepStrictEqual(parseJetstreamMessage(fixture('like-delete')), {
+    action: 'delete',
+    collection: 'run.nibling.feed.like',
+    uri: 'at://did:plc:zpmeobhmhfhz4clqwknac5gp/run.nibling.feed.like/3mzmqmt2eb72k',
+  });
 });
 
-test('drops a non-commit event', () => {
-  const frame = { $type: 'message', payload: { $type: 'network.bsky.jetstream.subscribeEvents#identity', did: 'did:plc:alice' } };
+test('drops a record that fails lexicon validation', () => {
+  const frame = fixture('project-create');
+  frame.payload.record = { $type: 'run.nibling.project', files: [], createdAt: '2026-11-06T06:23:21.775Z' };
   assert.equal(parseJetstreamMessage(frame), undefined);
 });
 
-test('a delete carries no record and needs no validation', () => {
-  const frame = commitFrame({ operation: 'delete', record: undefined, cid: undefined });
-  const event = parseJetstreamMessage(frame);
-  assert.deepStrictEqual(event, {
-    action: 'delete',
-    collection: 'run.nibling.project',
-    uri: 'at://did:plc:alice/run.nibling.project/3la1',
-  });
+test('drops captured app.bsky commits, a collection it does not care about', () => {
+  assert.equal(parseJetstreamMessage(fixture('app-bsky-like-create')), undefined);
+  assert.equal(parseJetstreamMessage(fixture('app-bsky-like-delete')), undefined);
+});
+
+test('drops captured identity and account events', () => {
+  assert.equal(parseJetstreamMessage(fixture('identity')), undefined);
+  assert.equal(parseJetstreamMessage(fixture('account')), undefined);
 });
 
 test('the URI and did come from the frame envelope, never the record body', () => {
-  const frame = commitFrame({
-    record: { $type: 'run.nibling.project', name: 'x', files: [], createdAt: '2026-11-06T06:47:44.859Z', did: 'did:plc:victim', uri: 'at://did:plc:victim/run.nibling.project/3la1' },
-  });
+  const frame = fixture('project-create');
+  frame.payload.record = {
+    ...(frame.payload.record as object),
+    did: 'did:plc:victim',
+    uri: 'at://did:plc:victim/run.nibling.project/3mzmqn3cvzz2c',
+  };
   const event = parseJetstreamMessage(frame);
   assert.ok(event && event.action === 'put');
-  assert.equal(event.uri, 'at://did:plc:alice/run.nibling.project/3la1');
-  assert.equal(event.did, 'did:plc:alice');
+  assert.equal(event.uri, 'at://did:plc:6mjqiiyd6crl57wl3pfypqjr/run.nibling.project/3mzmqn3cvzz2c');
+  assert.equal(event.did, 'did:plc:6mjqiiyd6crl57wl3pfypqjr');
 });
-- 
knotmirror 0.0.1


From c527ba7520861708385d9788b2468c42f0f76771 Mon Sep 17 00:00:00 2001
From: Nate Moore <git@natemoo.re>
Date: Fri, 6 Nov 2026 01:48:06 -0500
Subject: [PATCH 10/10] Call fetch unbound from the index backfill

The worker passed the global fetch as a property of the backfill deps,
and workerd throws Illegal invocation when it's called as a method, so
every relay listing failed. A wrangler dev run against the live relay
and Jetstream now backfills and advances the cursor.
---
 packages/api/src/worker.ts | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/packages/api/src/worker.ts b/packages/api/src/worker.ts
index c191608..c15ecf4 100644
--- a/packages/api/src/worker.ts
+++ b/packages/api/src/worker.ts
@@ -65,7 +65,8 @@ export class Index extends DurableObject<Env> {
       index: this.#model,
       backfill: {
         relayUrl: this.env.RELAY_URL,
-        fetch,
+        // workerd throws "Illegal invocation" when fetch is called as a method of another object.
+        fetch: (input, init) => fetch(input, init),
         resolvePds: (did) => resolveDidPds(did),
       },
       now: () => Date.now(),
-- 
knotmirror 0.0.1