Something went wrong. Try again.
Monorepo for the nibling.run platform
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208import assert from 'node:assert/strict';import { test } from 'node:test';import { backfillCollection, MAX_RECORD_PAGES, type BackfillDeps, type BackfillItem } from '../src/index/backfill.ts';import { upstream, type Upstream } from '../src/upstream.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 });}
test('backfillCollection pages listReposByCollection and listRecords, and validates each record', async () => { const calls: string[] = []; const fakeFetch = (async (input: string | URL | Request) => { const url = new URL(typeof input === 'string' ? input : input.toString()); calls.push(url.pathname + '?' + url.searchParams.toString());
if (url.pathname === '/xrpc/com.atproto.sync.listReposByCollection') { if (!url.searchParams.get('cursor')) { return jsonResponse({ repos: [{ did: 'did:plc:alice' }], cursor: 'page2' }); } return jsonResponse({ repos: [{ did: 'did:plc:bob' }] }); }
if (url.pathname === '/xrpc/com.atproto.repo.listRecords') { const repo = url.searchParams.get('repo'); if (repo === 'did:plc:alice') { return jsonResponse({ records: [ { uri: 'at://did:plc:alice/dev.nibling.project/3la1', cid: 'bafy1', value: { $type: 'dev.nibling.project', name: 'Alice Project', files: [], createdAt: '2026-11-01T00:00:00.000Z' }, }, { uri: 'at://did:plc:alice/dev.nibling.project/3la2', cid: 'bafy2', value: { $type: 'dev.nibling.project', files: [], createdAt: '2026-11-01T00:00:00.000Z' }, }, ], }); } throw new Error('network down'); }
throw new Error(`unexpected URL ${url}`); }) as typeof fetch;
const deps: BackfillDeps = { relayUrl: 'https://relay1.us-east.bsky.network', fetch: upstream(fakeFetch), resolvePds: async (did) => (did === 'did:plc:alice' ? 'https://alice.pds.example' : 'https://bob.pds.example'), };
const items = await collect(backfillCollection('dev.nibling.project', deps));
assert.deepStrictEqual(items.map((item) => (item.kind === 'event' ? item.event.uri : { repo: item.did, uris: [...item.uris] })), [ 'at://did:plc:alice/dev.nibling.project/3la1', { repo: 'did:plc:alice', uris: ['at://did:plc:alice/dev.nibling.project/3la1', 'at://did:plc:alice/dev.nibling.project/3la2'] }, ]); assert.deepStrictEqual(calls, [ '/xrpc/com.atproto.sync.listReposByCollection?collection=dev.nibling.project&limit=1000', '/xrpc/com.atproto.repo.listRecords?repo=did%3Aplc%3Aalice&collection=dev.nibling.project&limit=100', '/xrpc/com.atproto.sync.listReposByCollection?collection=dev.nibling.project&limit=1000&cursor=page2', '/xrpc/com.atproto.repo.listRecords?repo=did%3Aplc%3Abob&collection=dev.nibling.project&limit=100', ]);});
test('backfillCollection skips a repo whose PDS cannot be resolved', async () => { const deps: BackfillDeps = { relayUrl: 'https://relay1.us-east.bsky.network', fetch: upstream((async () => jsonResponse({ repos: [{ did: 'did:plc:ghost' }] })) as typeof fetch), resolvePds: async () => { throw new Error('did not found'); }, }; assert.deepStrictEqual(await collect(backfillCollection('dev.nibling.project', deps)), []);});
test('backfillCollection drops records whose URI names another repo or collection', async () => { const deps: BackfillDeps = { relayUrl: 'https://relay1.us-east.bsky.network', fetch: upstream((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: 'dev.nibling.project', name: 'Hijack', files: [], createdAt: '2026-11-01T00:00:00.000Z' }; return jsonResponse({ records: [ { uri: 'at://did:plc:victim/dev.nibling.project/3abc', cid: 'bafy1', value }, { uri: 'at://did:plc:evil/dev.nibling.feed.like/3abc', cid: 'bafy2', value }, { uri: 'at://did:plc:evil/dev.nibling.project/3own', cid: 'bafy3', value }, ], }); }) as typeof fetch), resolvePds: async () => 'https://evil.pds.example', }; const items = await collect(backfillCollection('dev.nibling.project', deps)); assert.deepStrictEqual( 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/dev.nibling.project/3own', did: 'did:plc:evil' }, 'repo'], );});
function relayThen(listRecords: (url: URL, init?: RequestInit) => Response | Promise<Response>, dids = ['did:plc:alice']): Upstream { return upstream(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); });}
const aliceRecord = (rkey: string) => ({ uri: `at://did:plc:alice/dev.nibling.project/${rkey}`, cid: `bafy-${rkey}`, value: { $type: 'dev.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('dev.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('dev.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: upstream((async () => { calls++; return jsonResponse({ repos: [], cursor: 'same' }); }) as typeof fetch), resolvePds: async () => 'https://alice.pds.example', }; assert.deepStrictEqual(await collect(backfillCollection('dev.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('dev.nibling.project', deps)); assert.deepStrictEqual(eventUris(items), ['at://did:plc:alice/dev.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('dev.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('dev.nibling.project', deps)), []); assert.deepStrictEqual(fetched, []);});