Something went wrong. Try again.
Monorepo for the nibling.run platform
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139import type { IndexCollection, IndexEvent } from './model.ts';import { toPutEvent, WANTED_COLLECTIONS } from './jetstream.ts';import { FETCH_TIMEOUT_MS, isPublicHttpsUrl } from '../net.ts';import type { Upstream } from '../upstream.ts';
/** Injected rather than `globalThis.fetch` so tests run with no network. */export interface BackfillDeps { relayUrl: string; fetch: Upstream; /** 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. */export type BackfillItem = | { kind: 'event'; event: IndexEvent } | { kind: 'repo'; collection: IndexCollection; did: string; uris: ReadonlySet<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; }}
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 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; 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 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 * 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. 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 { pds = await deps.resolvePds(did); } catch { return; } if (!isPublicHttpsUrl(pds)) return; const uris = new Set<string>(); const seen = new Set<string>(); let cursor: string | undefined; 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); 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 }; } 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 * stored cursor is missing or older than Jetstream's replay window. */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<BackfillItem> { for (const collection of WANTED_COLLECTIONS) { yield* backfillCollection(collection, deps); }}