diff --git a/apps/aggregator/.dev.vars.example b/apps/aggregator/.dev.vars.example new file mode 100644 index 0000000000..f093bab7a3 --- /dev/null +++ b/apps/aggregator/.dev.vars.example @@ -0,0 +1,8 @@ +# Local-dev secrets for the aggregator Worker. Copy to `.dev.vars` (which is +# gitignored) and customise. `wrangler dev` reads this file and binds the +# values into `env`. Production secrets are managed via +# `wrangler secret put ` and never appear in the repo. + +# Bearer token for the /_admin/* routes. Any non-empty string works locally; +# pick something you'd recognise in `wrangler tail` logs. +ADMIN_TOKEN=dev-only-not-for-production diff --git a/apps/aggregator/src/backfill-consumer.ts b/apps/aggregator/src/backfill-consumer.ts new file mode 100644 index 0000000000..3c6c9f536d --- /dev/null +++ b/apps/aggregator/src/backfill-consumer.ts @@ -0,0 +1,113 @@ +/** + * Backfill queue consumer. Pulls one `BackfillJob` at a time and walks + * `com.atproto.repo.listRecords` for that (DID, collection) pair, batching + * results onto the records queue for the standard verify-and-write path. + * + * Why a separate queue from records: per-pair work (PDS resolution + + * paginated listRecords + sendBatch onto the records queue) is bounded but + * non-trivial — running it inside the records-queue consumer would burn the + * sub-request budget for jobs that should just be writing to D1. Keeping + * the queues separate also lets the operator throttle backfill work + * independently of live ingest. + * + * Error policy: + * - Per-pair `processBackfillJob` throw → `message.retry()`. Cloudflare + * Queues backs off and retries; after `max_retries` (3, configured in + * wrangler.jsonc) the message lands in `emdash-aggregator-backfill-dlq`. + * - Unexpected throws inside the batch loop are caught per-message so one + * bad job can't poison the rest of the batch. + * + * DLQ drain (`drainBackfillDeadLetterBatch`): logs each dead-lettered + * pair at error level (so operators tailing `wrangler tail` see it loud) + * and acks so the DLQ doesn't accumulate unbounded. No D1 forensics row — + * the recovery action for a backfill pair that exhausted retries is + * "re-run backfill for the affected DID", which only needs the + * (did, collection) pair already on the log line. + */ + +import { + AtprotoWebDidDocumentResolver, + CompositeDidDocumentResolver, + PlcDidDocumentResolver, +} from "@atcute/identity-resolver"; + +import { processBackfillJob, type ProcessBackfillJobDeps } from "./backfill.js"; +import { createD1DidDocCache, DidResolver } from "./did-resolver.js"; +import type { BackfillJob } from "./env.js"; +import type { MessageBatchLike } from "./records-consumer.js"; + +/** + * Process one batch of backfill jobs. Mirrors `records-consumer.processBatch`'s + * shape: per-message try/catch, ack on success, retry on throw. + * + * `depsOverride` is the test seam — production calls without it and gets + * the standard composite resolver wired against `env.DB`. + */ +export async function processBackfillBatch( + batch: MessageBatchLike, + env: Env, + depsOverride?: ProcessBackfillJobDeps, +): Promise { + const deps = depsOverride ?? createProductionDeps(env); + for (const message of batch.messages) { + const job = message.body; + try { + const result = await processBackfillJob(job, deps); + console.log("[aggregator] backfill job complete", { + did: result.did, + collection: result.collection, + enqueued: result.enqueued, + }); + message.ack(); + } catch (err) { + // Resolution failures, listRecords 5xx, timeouts, and pagination + // runaway all land here. Retry — Cloudflare Queues backoff handles + // transient PDS failures; permanently broken DIDs land in the DLQ + // after max_retries. + console.error("[aggregator] backfill job failed, retrying", { + did: job.did, + collection: job.collection, + error: err instanceof Error ? err.message : String(err), + }); + message.retry(); + } + } +} + +/** + * Drain the backfill DLQ. Mirror of `records-consumer.drainDeadLetterBatch` + * but without the D1 forensics row — the recovery action for a backfill + * pair that exhausted retries is "re-run backfill for the affected DID", + * which only needs the (did, collection) pair from the log line. + * + * Logs at error level so operators tailing `wrangler tail` see the message + * loud, acks so the DLQ doesn't accumulate unbounded. + */ +export function drainBackfillDeadLetterBatch( + batch: MessageBatchLike, + _env: Env, +): void { + for (const message of batch.messages) { + console.error("[aggregator] backfill DLQ drain: pair exhausted retries", { + did: message.body.did, + collection: message.body.collection, + }); + message.ack(); + } +} + +function createProductionDeps(env: Env): ProcessBackfillJobDeps { + const composite = new CompositeDidDocumentResolver({ + methods: { + plc: new PlcDidDocumentResolver(), + web: new AtprotoWebDidDocumentResolver(), + }, + }); + return { + resolver: new DidResolver({ + cache: createD1DidDocCache(env.DB), + resolver: composite, + }), + queue: env.RECORDS_QUEUE, + }; +} diff --git a/apps/aggregator/src/backfill.ts b/apps/aggregator/src/backfill.ts new file mode 100644 index 0000000000..db0878ff81 --- /dev/null +++ b/apps/aggregator/src/backfill.ts @@ -0,0 +1,457 @@ +/** + * Cold-start discovery worker. + * + * Operator-triggered (via `POST /_admin/backfill`). Two trigger shapes: + * + * - `{ "dids": [...] }` — explicit list, primarily for testing or recovery + * of a known DID set. + * - `{}` (empty body) — production cold-start. Calls + * `com.atproto.sync.listReposByCollection` against the configured relay + * for each NSID in `WANTED_COLLECTIONS`, paginates the full DID set, + * dedupes, and feeds the union into the same backfill loop. + * + * Architecture: the POST handler synchronously discovers DIDs (or accepts an + * explicit list), then fans out one `BackfillJob = { did, collection }` per + * (DID × WANTED_COLLECTIONS) pair onto the dedicated `BACKFILL_QUEUE` via + * `sendBatch`. A separate consumer (`backfill-consumer.ts`) processes one + * pair at a time: resolve PDS, paginate `com.atproto.repo.listRecords`, + * batch-enqueue each returned record onto the existing Records Queue. + * + * Why a separate queue rather than running the per-DID loop inside the + * `ctx.waitUntil` of the POST handler: Cloudflare's hard 30-second + * wall-clock cap on `waitUntil` would limit a single backfill POST to + * ~15–25 DIDs before in-flight work was cancelled. The queue gives us + * automatic retry, concurrency, and per-pair invocation budgets that each + * fit comfortably under the sub-request ceiling. + * + * Live discovery (post-cold-start) is Jetstream's job, not this worker's; + * the consumer writes `known_publishers` opportunistically on any record + * event for an unseen DID. Backfill exists for the cold-start gap + * (publishers who published before the aggregator was listening) and for + * operator-triggered recovery after a known outage. There is deliberately + * no periodic scheduler — see plan §"Why no reconciliation cron". + */ + +import { parseCanonicalResourceUri } from "@atcute/lexicons/syntax"; + +import { WANTED_COLLECTIONS } from "./constants.js"; +import type { DidResolver } from "./did-resolver.js"; +import type { BackfillJob, RecordsJob } from "./env.js"; +import { isPlainObject } from "./utils.js"; + +const PAGE_SIZE = 100; +/** Per-listRecords-page timeout. A hostile or hung publisher PDS that + * accepts the connection but stalls the body would otherwise block the + * fetch until workerd's overall sub-request budget exhausts — starving + * every later page in the same consumer invocation. Same shape as + * `pds-verify.ts`'s fetchCar timeout. */ +const LIST_RECORDS_TIMEOUT_MS = 15_000; +/** Cap on listRecords pagination per (DID, collection) pair. A buggy or + * malicious PDS that echoes the same cursor would otherwise loop forever + * inside one consumer invocation. 1000 pages × 100 records = 100k records + * per pair, which is past anything we'd legitimately backfill in one shot. */ +const MAX_PAGES_PER_COLLECTION = 1000; +/** Defensive cap on records per page. Real PDSes honour the `limit` query + * param; this guards against a hostile PDS returning an enormous array. + * Capped at the same width as Cloudflare Queues' sendBatch (100) so a + * compliant page maps 1:1 to one batch send; oversize pages are rejected + * rather than chunked, surfacing the PDS's spec violation as a partial + * failure the operator can investigate. */ +const MAX_RECORDS_PER_PAGE = PAGE_SIZE; +/** Cloudflare Queues' hard cap on `sendBatch` size. Per-page enqueues are + * always ≤ this thanks to MAX_RECORDS_PER_PAGE; documented here so the + * relationship is visible at the call site. */ +export const QUEUE_SEND_BATCH_CAP = 100; +// Static guard: bumping MAX_RECORDS_PER_PAGE above the queue's batch cap +// would silently break sendBatch in production. Surface the violation at +// module load rather than at the first batch send. +if (MAX_RECORDS_PER_PAGE > QUEUE_SEND_BATCH_CAP) { + throw new Error( + `MAX_RECORDS_PER_PAGE (${MAX_RECORDS_PER_PAGE}) exceeds QUEUE_SEND_BATCH_CAP (${QUEUE_SEND_BATCH_CAP})`, + ); +} +/** Producer-side records queue surface. The production binding + * `env.RECORDS_QUEUE` satisfies this; tests pass an in-memory implementation. */ +export interface RecordsQueueProducer { + sendBatch(messages: ReadonlyArray<{ body: RecordsJob }>): Promise; +} + +/** Producer-side backfill-jobs queue surface. The production binding + * `env.BACKFILL_QUEUE` satisfies this; tests pass an in-memory + * implementation. Same shape as `RecordsQueueProducer`, separated so the + * type system catches accidental cross-wiring. */ +export interface BackfillQueueProducer { + sendBatch(messages: ReadonlyArray<{ body: BackfillJob }>): Promise; +} + +/** Cap on pages walked per relay collection. Same shape as + * MAX_PAGES_PER_COLLECTION but for the discovery side. At 100 repos/page, + * 100 pages = 10k publishers — past anything we'd legitimately discover at + * Slice 1 scale. */ +const MAX_DISCOVERY_PAGES_PER_COLLECTION = 100; +const DISCOVERY_PAGE_SIZE = 100; +const DISCOVERY_TIMEOUT_MS = 15_000; +/** Defensive cap on the union of discovered DIDs. A relay returning a + * runaway list (bug or hostile mirror) would otherwise let one POST fan + * out millions of jobs onto BACKFILL_QUEUE. At Slice 1 scale we expect a + * handful to a few hundred publishers. */ +export const MAX_DISCOVERED_DIDS = 1000; + +/** + * Discover all DIDs publishing any of `WANTED_COLLECTIONS` by querying the + * relay's `com.atproto.sync.listReposByCollection`. Returns the union of + * unique DIDs across all collections, in arbitrary order. + * + * Uses the same defenses as the per-pair listRecords loop: per-page + * timeout, max-page cap, cursor-equality check. Per-collection failures + * are logged and the loop continues — discovery via the relay is + * best-effort; a partial discovery list is better than none. Stops early + * once `MAX_DISCOVERED_DIDS` is reached so a runaway relay can't pump + * arbitrary fan-out into the queue. + */ +export async function discoverDids( + relayUrl: string, + opts: { fetch?: typeof fetch; timeoutMs?: number } = {}, +): Promise { + const fetchImpl = opts.fetch ?? fetch; + const timeoutMs = opts.timeoutMs ?? DISCOVERY_TIMEOUT_MS; + const dids = new Set(); + for (const collection of WANTED_COLLECTIONS) { + try { + await discoverCollection(relayUrl, collection, fetchImpl, timeoutMs, dids); + } catch (err) { + console.error("[aggregator] backfill discovery failed for collection", { + collection, + error: err instanceof Error ? err.message : String(err), + }); + } + if (dids.size >= MAX_DISCOVERED_DIDS) { + console.warn("[aggregator] backfill discovery hit DID cap, stopping early", { + cap: MAX_DISCOVERED_DIDS, + stoppedAfterCollection: collection, + }); + break; + } + } + return [...dids]; +} + +async function discoverCollection( + relayUrl: string, + collection: string, + fetchImpl: typeof fetch, + timeoutMs: number, + dids: Set, +): Promise { + let cursor: string | undefined; + let prevCursor: string | undefined; + let pages = 0; + do { + if (++pages > MAX_DISCOVERY_PAGES_PER_COLLECTION) { + throw new Error(`exceeded ${MAX_DISCOVERY_PAGES_PER_COLLECTION} discovery pages`); + } + if (cursor !== undefined && cursor === prevCursor) { + throw new Error("relay returned identical cursor twice"); + } + prevCursor = cursor; + + const url = new URL("/xrpc/com.atproto.sync.listReposByCollection", relayUrl); + url.searchParams.set("collection", collection); + url.searchParams.set("limit", String(DISCOVERY_PAGE_SIZE)); + if (cursor) url.searchParams.set("cursor", cursor); + + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), timeoutMs); + let response: Response; + try { + response = await fetchImpl(url.toString(), { + headers: { accept: "application/json" }, + signal: controller.signal, + }); + } catch (err) { + if (err instanceof Error && err.name === "AbortError") { + throw new Error(`listReposByCollection timed out after ${timeoutMs}ms`, { cause: err }); + } + throw err; + } finally { + clearTimeout(timer); + } + if (!response.ok) { + throw new Error(`listReposByCollection returned ${response.status}`); + } + const body: unknown = await response.json(); + if (!isPlainObject(body)) throw new Error("listReposByCollection returned non-object body"); + const repos = body["repos"]; + if (!Array.isArray(repos)) { + throw new Error("listReposByCollection response missing `repos` array"); + } + for (const repo of repos) { + if (!isPlainObject(repo)) continue; + const did = repo["did"]; + if (typeof did === "string") { + dids.add(did); + if (dids.size >= MAX_DISCOVERED_DIDS) return; + } + } + const nextCursor = body["cursor"]; + cursor = typeof nextCursor === "string" && nextCursor.length > 0 ? nextCursor : undefined; + } while (cursor); +} + +/** + * Fan out backfill work for a list of DIDs onto `BACKFILL_QUEUE`. Produces + * one `BackfillJob` per (DID × WANTED_COLLECTIONS) pair, sent in batches + * of up to `QUEUE_SEND_BATCH_CAP` to honour Cloudflare Queues' sendBatch + * limit. Returns the total number of jobs enqueued. + * + * Caller is responsible for capping `dids.length` (admin route enforces + * `MAX_BACKFILL_DIDS` for the explicit path; `discoverDids` enforces + * `MAX_DISCOVERED_DIDS` for the discovery path). This function trusts its + * input and just fans out — the per-call work is bounded by + * `dids.length * WANTED_COLLECTIONS.length` enqueues. + */ +export async function enqueueBackfillJobs( + dids: readonly string[], + queue: BackfillQueueProducer, +): Promise { + const messages: { body: BackfillJob }[] = []; + for (const did of dids) { + for (const collection of WANTED_COLLECTIONS) { + messages.push({ body: { did, collection } }); + } + } + // Fan the sendBatch calls in parallel. Each is an independent outbound + // sub-request and the orchestrator runs inside the POST handler's 30s + // `waitUntil` budget — serial awaits on a `MAX_DISCOVERED_DIDS`-sized + // fan-out (1000 DIDs × 4 collections / 100 = 40 batches) would + // noticeably eat into the cold-start budget on top of discovery's own + // fetches. + const sends: Promise[] = []; + for (let i = 0; i < messages.length; i += QUEUE_SEND_BATCH_CAP) { + sends.push(queue.sendBatch(messages.slice(i, i + QUEUE_SEND_BATCH_CAP))); + } + await Promise.all(sends); + return messages.length; +} + +export interface ProcessBackfillJobDeps { + resolver: DidResolver; + queue: RecordsQueueProducer; + /** Inject for tests; defaults to `globalThis.fetch`. */ + fetch?: typeof fetch; + /** Override for the per-listRecords-page timeout. Defaults to + * `LIST_RECORDS_TIMEOUT_MS`. Tests use a small value to exercise the + * abort path without burning the production budget. */ + listRecordsTimeoutMs?: number; +} + +export interface ProcessBackfillJobResult { + did: string; + collection: string; + enqueued: number; +} + +/** + * Process one (DID, collection) pair: resolve the DID's PDS, paginate + * `com.atproto.repo.listRecords` for the collection, and batch-enqueue + * each returned record onto the records queue. + * + * Throws on any failure (resolution, listRecords status, pagination + * runaway). The queue consumer translates a thrown result into + * `message.retry()`; messages that exhaust max_retries land in the + * backfill DLQ for the operator to inspect. + * + * 404 from the PDS on the first page is treated as "publisher does not + * host this collection" and returns 0 enqueues without throwing — same + * shape as the previous serial-loop semantics. + */ +export async function processBackfillJob( + job: BackfillJob, + deps: ProcessBackfillJobDeps, +): Promise { + const resolved = await deps.resolver.resolve(job.did); + const fetchImpl = deps.fetch ?? fetch; + const timeoutMs = deps.listRecordsTimeoutMs ?? LIST_RECORDS_TIMEOUT_MS; + const enqueued = await paginateAndEnqueue({ + did: job.did, + pds: resolved.pds, + collection: job.collection, + queue: deps.queue, + fetchImpl, + timeoutMs, + }); + return { did: job.did, collection: job.collection, enqueued }; +} + +interface PaginateOpts { + did: string; + pds: string; + collection: string; + queue: RecordsQueueProducer; + fetchImpl: typeof fetch; + timeoutMs: number; +} + +/** + * Walk one DID's records for a single collection, paginating through + * `listRecords` and enqueuing each result via `sendBatch` (one batch per + * page). Returns the total records enqueued. + * + * 404 from the PDS on the FIRST page means the repo doesn't host this + * collection — silently treated as zero records, not an error. A 404 + * mid-pagination is a partial-failure signal (the PDS is misrouting one + * page while the rest of the repo is fine) and throws. + * + * Pagination is capped at MAX_PAGES_PER_COLLECTION + cursor-equality check + * to defend against a PDS that echoes the same cursor forever. + * + * `MAX_RECORDS_PER_PAGE` matches Cloudflare Queues' `sendBatch` cap (100), + * so a compliant page maps 1:1 to one batch send. A PDS that ignores the + * `?limit=` query and returns more records than that throws — we'd rather + * surface the spec violation than silently chunk and hide the upstream bug. + */ +async function paginateAndEnqueue(opts: PaginateOpts): Promise { + let cursor: string | undefined; + let prevCursor: string | undefined; + let pages = 0; + let totalEnqueued = 0; + do { + if (++pages > MAX_PAGES_PER_COLLECTION) { + throw new Error(`exceeded ${MAX_PAGES_PER_COLLECTION} pages`); + } + if (cursor !== undefined && cursor === prevCursor) { + throw new Error("PDS returned identical cursor twice"); + } + prevCursor = cursor; + + const url = new URL("/xrpc/com.atproto.repo.listRecords", opts.pds); + url.searchParams.set("repo", opts.did); + url.searchParams.set("collection", opts.collection); + url.searchParams.set("limit", String(PAGE_SIZE)); + if (cursor) url.searchParams.set("cursor", cursor); + + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), opts.timeoutMs); + let response: Response; + try { + response = await opts.fetchImpl(url.toString(), { + headers: { accept: "application/json" }, + signal: controller.signal, + }); + } catch (err) { + if (err instanceof Error && err.name === "AbortError") { + throw new Error(`listRecords timed out after ${opts.timeoutMs}ms`, { cause: err }); + } + throw err; + } finally { + clearTimeout(timer); + } + if (response.status === 404) { + if (cursor === undefined) { + // First-page 404: publisher has no records of this collection. + return totalEnqueued; + } + // Mid-pagination 404 is a partial failure; surface it. + throw new Error(`listRecords returned 404 mid-pagination at cursor=${cursor}`); + } + if (!response.ok) { + throw new Error(`listRecords returned ${response.status}`); + } + + const body: unknown = await response.json(); + const records = extractListRecordsBody(body); + if (records.length > MAX_RECORDS_PER_PAGE) { + throw new Error( + `PDS returned ${records.length} records, exceeding per-page cap of ${MAX_RECORDS_PER_PAGE}`, + ); + } + cursor = extractCursor(body); + + const messages: { body: RecordsJob }[] = []; + for (const record of records) { + const parsed = parseCanonicalResourceUri(record.uri); + if (!parsed.ok) continue; + // Defence vs. a buggy/malicious PDS that returns records under + // a different DID (or a different collection) than the one we + // asked for. Such jobs would never verify (signature would be + // from a different key) and would just churn dead-letters; drop + // at the source. `parseCanonicalResourceUri` already validated + // the rkey grammar and the collection NSID for us, so we only + // need the cross-checks here. + if (parsed.value.repo !== opts.did) continue; + if (parsed.value.collection !== opts.collection) continue; + messages.push({ + body: { + did: opts.did, + collection: opts.collection, + rkey: parsed.value.rkey, + operation: "create", + cid: record.cid, + }, + }); + } + if (messages.length > 0) { + // Page size is capped at QUEUE_SEND_BATCH_CAP at module load via + // the static assertion above, so this sendBatch never exceeds + // Cloudflare's 100-message limit by construction. + await opts.queue.sendBatch(messages); + totalEnqueued += messages.length; + } + } while (cursor); + return totalEnqueued; +} + +interface ListRecordEntry { + uri: string; + cid: string; + value: unknown; +} + +/** + * Parse a `com.atproto.repo.listRecords` response body. Throws on any + * structural mismatch — a PDS that 200s with the wrong shape is upstream + * breakage, not "no records", and silently treating it as the latter + * causes operator-invisible partial backfills. The thrown error + * propagates out of `processBackfillJob` and the queue consumer retries + * (then DLQs) per the standard policy. + */ +function extractListRecordsBody(body: unknown): ListRecordEntry[] { + if (!isPlainObject(body)) { + throw new Error("listRecords response was not a JSON object"); + } + const records = body["records"]; + if (!Array.isArray(records)) { + throw new Error("listRecords response missing `records` array"); + } + const out: ListRecordEntry[] = []; + for (const r of records) { + if (!isPlainObject(r)) { + throw new Error("listRecords record entry was not a JSON object"); + } + const uri = r["uri"]; + const cid = r["cid"]; + if (typeof uri !== "string" || typeof cid !== "string") { + throw new Error("listRecords record entry missing string `uri` or `cid`"); + } + out.push({ uri, cid, value: r["value"] }); + } + return out; +} + +/** + * Pull the optional `cursor` out of a `listRecords` response. `undefined` + * (no key) is the spec-compliant signal for "end of pagination"; any other + * non-string value (number, object, etc.) is a PDS bug and throws so the + * pagination loop doesn't silently terminate on the wrong page. + */ +function extractCursor(body: unknown): string | undefined { + if (!isPlainObject(body)) { + throw new Error("listRecords response was not a JSON object"); + } + const cursor = body["cursor"]; + if (cursor === undefined) return undefined; + if (typeof cursor !== "string") { + throw new Error(`listRecords cursor was not a string (got ${typeof cursor})`); + } + return cursor; +} diff --git a/apps/aggregator/src/did-resolver.ts b/apps/aggregator/src/did-resolver.ts index cbdffefd20..8e4d5c58ed 100644 --- a/apps/aggregator/src/did-resolver.ts +++ b/apps/aggregator/src/did-resolver.ts @@ -18,10 +18,9 @@ import { type PublicKey, } from "@atcute/crypto"; import { type DidDocument, getAtprotoVerificationMaterial, getPdsEndpoint } from "@atcute/identity"; -import type { Did } from "@atcute/lexicons/syntax"; +import { type Did, isDid } from "@atcute/lexicons/syntax"; const DEFAULT_TTL_MS = 24 * 60 * 60 * 1000; -const DID_PATTERN = /^did:[a-z]+:[A-Za-z0-9._%:-]+$/; /** Cache entry shape; the multibase signing key is stored raw so the * `PublicKey` instance is reconstructed on each `resolve()`. WebCrypto @@ -102,10 +101,6 @@ export class DidResolver { } } -function isDid(value: string): value is Did { - return DID_PATTERN.test(value); -} - function asDid(did: string): Did { if (!isDid(did)) { throw new Error(`invalid DID: ${did}`); diff --git a/apps/aggregator/src/env.ts b/apps/aggregator/src/env.ts index 3053c6d33e..f85caadb5d 100644 --- a/apps/aggregator/src/env.ts +++ b/apps/aggregator/src/env.ts @@ -20,3 +20,18 @@ export interface RecordsJob { */ jetstreamRecord?: unknown; } + +/** + * One unit of cold-start backfill work: walk every record under `collection` + * for `did` via that DID's PDS, then enqueue each record onto the records + * queue for the standard verify-and-write path. + * + * Granularity is (DID, collection) rather than per-DID because the per-DID + * fan-out can exceed Cloudflare's 30s `ctx.waitUntil` wall-clock cap. One + * collection's worth of pagination fits comfortably under that ceiling and + * gets queue-level retry + concurrency for free. + */ +export interface BackfillJob { + did: string; + collection: string; +} diff --git a/apps/aggregator/src/index.ts b/apps/aggregator/src/index.ts index 709f7da6a8..ee364e5c1e 100644 --- a/apps/aggregator/src/index.ts +++ b/apps/aggregator/src/index.ts @@ -15,31 +15,177 @@ * Worker boots. */ -import type { RecordsJob } from "./env.js"; +import { isDid } from "@atcute/lexicons/syntax"; + +import { drainBackfillDeadLetterBatch, processBackfillBatch } from "./backfill-consumer.js"; +import { discoverDids, enqueueBackfillJobs } from "./backfill.js"; +import type { BackfillJob, RecordsJob } from "./env.js"; import { drainDeadLetterBatch, processBatch } from "./records-consumer.js"; import { RECORDS_DO_NAME } from "./records-do.js"; +import { isPlainObject } from "./utils.js"; const RECORDS_QUEUE_NAME = "emdash-aggregator-records"; const RECORDS_DLQ_NAME = "emdash-aggregator-records-dlq"; +const BACKFILL_QUEUE_NAME = "emdash-aggregator-backfill"; +const BACKFILL_DLQ_NAME = "emdash-aggregator-backfill-dlq"; export { RecordsJetstreamDO } from "./records-do.js"; /** - * Operational bootstrap route. Hitting `/_admin/start` once after deploy - * spins up the Records DO, which opens its outbound WebSocket and starts - * ingesting. The DO's WebSocket keeps it alive thereafter. The route is - * unauthenticated but returns no operational detail — just a fixed 204 — - * so a probing caller learns nothing useful. The action is idempotent on - * an already-running DO. Recommended deploy hook: + * Operational admin routes. Both gated by the `ADMIN_TOKEN` secret declared + * in `wrangler.jsonc`'s `secrets.required` and validated via constant-time + * compare against the `Authorization: Bearer ` header. Recommended + * deploy hook: + * + * wrangler deploy && curl -X POST -H "Authorization: Bearer $ADMIN_TOKEN" \ + * https://api.emdashcms.com/_admin/start + * + * `/_admin/start` spins up the Records DO (idempotent on an already-running DO). + * `/_admin/backfill` discovers publishers (or accepts an explicit DID list) + * and fans (DID, collection) jobs onto `BACKFILL_QUEUE`; the consumer in + * `backfill-consumer.ts` does the per-pair listRecords + records-queue + * fan-out work asynchronously, getting queue retry + concurrency for free + * and sidestepping the 30s `waitUntil` cap on the POST handler. * - * wrangler deploy && curl -X POST https://api.emdashcms.com/_admin/start + * Auth-gating both routes: backfill specifically introduces caller-chosen + * URLs into the worker's outbound fetches (via `did:web` resolution + the + * publisher's PDS endpoint), so the unauth posture would expose an SSRF-shaped + * surface. Same gate on `/_admin/start` for symmetry — anyone with the token + * can do operational things. */ const BOOTSTRAP_PATH = "/_admin/start"; +const BACKFILL_PATH = "/_admin/backfill"; + +/** + * Cap on the explicit DID list a single POST may submit. Lower than the + * old serial-loop cap (was 1000) because the queue-fan-out path amplifies + * a leaked-token attack: each submitted DID becomes + * `WANTED_COLLECTIONS.length` queue messages, each consuming a consumer + * invocation with its own outbound PDS fetches. 100 × 4 = 400 jobs from + * the explicit path is a meaningful operator-recovery batch but a + * tractable blast radius for the explicit path. + * + * The discovery path (empty body) has a separate ceiling at + * `MAX_DISCOVERED_DIDS = 1000` and is therefore actually the larger + * worst-case fan-out source — by design, since legitimate-publisher + * enumeration is the primary use case and we want the explicit list to + * be the tighter "operator types it themselves" bucket. Both paths share + * the same per-pair caps in the consumer. + */ +const MAX_BACKFILL_DIDS = 100; + +const tokenEncoder = new TextEncoder(); + +/** + * Constant-time string equality via workerd's audited + * `crypto.subtle.timingSafeEqual`. The primitive returns `false` immediately + * for length-mismatched buffers, so the *prefix*-comparison is constant-time + * but a length difference is still observable via timing — acceptable here + * because the protected secret (`ADMIN_TOKEN`) has a fixed configured length + * known only to the operator, and any realistic length-via-timing attack + * would require so many requests that other defences (rate-limiting, + * Cloudflare Bot Management, log review) catch it first. + */ +function timingSafeEqual(a: string, b: string): boolean { + const aBuf = tokenEncoder.encode(a); + const bBuf = tokenEncoder.encode(b); + if (aBuf.byteLength !== bBuf.byteLength) return false; + return crypto.subtle.timingSafeEqual(aBuf, bBuf); +} + +/** + * Validate the request's `Authorization: Bearer ` header against + * `env.ADMIN_TOKEN`. Returns null on success, or a 401 Response to return + * directly. Empty/missing token in env fails closed. + */ +function requireAdminAuth(request: Request, env: Env): Response | null { + const expected = env.ADMIN_TOKEN; + // `trim()` defends against a whitespace-only secret slipping past + // `secrets.required`'s presence check — `wrangler secret put ADMIN_TOKEN` + // followed by an accidental Enter would otherwise produce a working + // endpoint with a trivially guessable token. + if (!expected || expected.trim().length === 0) { + // Misconfigured production or unset dev — closed by default. + return new Response("admin endpoints not configured", { status: 503 }); + } + const auth = request.headers.get("authorization"); + const SCHEME_PREFIX = "bearer "; + // RFC 6750 §2.1: the auth scheme is case-insensitive. `curl -H + // "authorization: bearer ..."` and SDKs that don't canonicalise the + // scheme would otherwise fail with a confusing 401 even though the + // token is correct. + if ( + !auth || + auth.length < SCHEME_PREFIX.length || + auth.slice(0, SCHEME_PREFIX.length).toLowerCase() !== SCHEME_PREFIX + ) { + return new Response("unauthorized", { + status: 401, + headers: { "www-authenticate": "Bearer" }, + }); + } + const token = auth.slice(SCHEME_PREFIX.length); + if (!timingSafeEqual(token, expected)) { + return new Response("unauthorized", { + status: 401, + headers: { "www-authenticate": "Bearer" }, + }); + } + return null; +} + +type BackfillRequest = { mode: "explicit"; dids: string[] } | { mode: "discover" }; + +function parseBackfillBody(body: unknown): BackfillRequest | { error: string } { + if (!isPlainObject(body)) { + return { error: "request body must be a JSON object" }; + } + const rawDids = body["dids"]; + // Empty body OR `{ "dids": null }` OR `{ "dids": undefined }` ⇒ discovery + // mode. Production cold-start uses this path; the explicit list is the + // testing / recovery seam. + if (rawDids === undefined || rawDids === null) { + return { mode: "discover" }; + } + if (!Array.isArray(rawDids)) { + return { + error: "`dids` must be an array of DID strings, or omitted to discover via the relay", + }; + } + if (rawDids.length === 0) { + return { + error: "`dids` must not be empty (omit the field to discover via the relay)", + }; + } + if (rawDids.length > MAX_BACKFILL_DIDS) { + return { + error: `\`dids\` must contain at most ${MAX_BACKFILL_DIDS} entries (got ${rawDids.length})`, + }; + } + const seen = new Set(); + for (const did of rawDids) { + if (!isDid(did)) { + return { error: `invalid DID in list: ${JSON.stringify(did)}` }; + } + seen.add(did); + } + // Set iteration order matches insertion; dedup preserves first-seen order + // so the operator sees jobs run in the order they submitted. + return { mode: "explicit", dids: [...seen] }; +} export default { async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise { const url = new URL(request.url); if (url.pathname === BOOTSTRAP_PATH) { + if (request.method !== "POST") { + return new Response("method not allowed", { + status: 405, + headers: { allow: "POST" }, + }); + } + const denied = requireAdminAuth(request, env); + if (denied) return denied; const id = env.RECORDS_DO.idFromName(RECORDS_DO_NAME); const stub = env.RECORDS_DO.get(id); // Fire-and-forget so the response shape doesn't depend on the @@ -48,21 +194,72 @@ export default { ctx.waitUntil(stub.fetch("https://do.internal/bootstrap")); return new Response(null, { status: 204 }); } + if (url.pathname === BACKFILL_PATH) { + if (request.method !== "POST") { + return new Response("method not allowed", { + status: 405, + headers: { allow: "POST" }, + }); + } + const denied = requireAdminAuth(request, env); + if (denied) return denied; + // Empty / no body is the production discovery path. JSON-parse + // failures with no content (Content-Length: 0) come back as + // SyntaxError; we treat that as "discover" rather than 400ing + // so `curl -X POST ... /_admin/backfill` (no body) does the + // expected thing. + let body: unknown = {}; + const text = await request.text(); + if (text.length > 0) { + try { + body = JSON.parse(text); + } catch { + return new Response("request body must be valid JSON", { status: 400 }); + } + } + const parsed = parseBackfillBody(body); + if ("error" in parsed) { + return new Response(parsed.error, { status: 400 }); + } + // Fire-and-forget via waitUntil so the route returns 202 quickly. + // `runBackfill` only does discovery + queue fan-out (both fast); + // per-pair work runs in the BACKFILL_QUEUE consumer with its own + // invocation budget. Operator-facing observability: + // - this handler's logs ([aggregator] backfill discovery/enqueue) + // - per-pair logs from `backfill-consumer.ts` + // - DLQ inspection (`emdash-aggregator-backfill-dlq`) for + // pairs that exhausted retries + ctx.waitUntil(runBackfill(parsed, env)); + return new Response(null, { status: 202 }); + } return new Response("emdash-aggregator: not yet implemented", { status: 503, headers: { "content-type": "text/plain" }, }); }, - async queue(batch: MessageBatch, env: Env, _ctx: ExecutionContext): Promise { - // Workerd routes both consumers (records + records-dlq) here; dispatch - // by queue name. Adding a third queue requires updating this switch. + async queue(batch: MessageBatch, env: Env, _ctx: ExecutionContext): Promise { + // Workerd routes every consumer here; dispatch by queue name to the + // matching typed handler. The parameter is unparameterised + // (`MessageBatch` defaults to `MessageBatch`) because the + // binding is shared across queues — narrowing happens per-case via + // the queue name, which is a runtime tag the compiler can't see. switch (batch.queue) { case RECORDS_QUEUE_NAME: - await processBatch(batch, env); + // eslint-disable-next-line typescript-eslint(no-unsafe-type-assertion) -- narrowed by queue name + await processBatch(batch as MessageBatch, env); return; case RECORDS_DLQ_NAME: - await drainDeadLetterBatch(batch, env); + // eslint-disable-next-line typescript-eslint(no-unsafe-type-assertion) -- narrowed by queue name + await drainDeadLetterBatch(batch as MessageBatch, env); + return; + case BACKFILL_QUEUE_NAME: + // eslint-disable-next-line typescript-eslint(no-unsafe-type-assertion) -- narrowed by queue name + await processBackfillBatch(batch as MessageBatch, env); + return; + case BACKFILL_DLQ_NAME: + // eslint-disable-next-line typescript-eslint(no-unsafe-type-assertion) -- narrowed by queue name + drainBackfillDeadLetterBatch(batch as MessageBatch, env); return; default: console.error("[aggregator] unknown queue, acking batch", { queue: batch.queue }); @@ -82,3 +279,66 @@ export default { ctx.waitUntil(stub.fetch("https://do.internal/liveness")); }, }; + +type BackfillRequestParsed = { mode: "explicit"; dids: string[] } | { mode: "discover" }; + +/** + * Backfill orchestrator. Runs inside the POST handler's `ctx.waitUntil`. + * Two paths: + * + * - "discover" — call `com.atproto.sync.listReposByCollection` against + * `env.RELAY_URL` for each NSID in WANTED_COLLECTIONS, union the DIDs, + * then fan them out as backfill jobs. + * - "explicit" — operator-supplied DID list, used for testing or for + * recovery of a known DID set. + * + * In both cases the actual per-(DID, collection) work runs in the + * `BACKFILL_QUEUE` consumer (see `backfill-consumer.ts`). This orchestrator + * just discovers + enqueues — both fast operations that fit well within + * Cloudflare's 30s `waitUntil` ceiling regardless of fan-out size, even + * after a `MAX_DISCOVERED_DIDS`-sized union. + * + * Per-pair progress is logged from the consumer; this function only logs + * the discovery + fan-out result so an operator watching `wrangler tail` + * sees how many jobs were enqueued. + */ +async function runBackfill(req: BackfillRequestParsed, env: Env): Promise { + try { + let dids: readonly string[]; + if (req.mode === "discover") { + console.log("[aggregator] backfill discovery starting", { + relay: env.RELAY_URL, + }); + dids = await discoverDids(env.RELAY_URL); + console.log("[aggregator] backfill discovery complete", { + relay: env.RELAY_URL, + didCount: dids.length, + }); + } else { + dids = req.dids; + console.log("[aggregator] backfill enqueue starting", { + mode: "explicit", + didCount: dids.length, + }); + } + + if (dids.length === 0) { + console.warn("[aggregator] backfill produced zero DIDs, nothing to enqueue", { + mode: req.mode, + }); + return; + } + + const enqueued = await enqueueBackfillJobs(dids, env.BACKFILL_QUEUE); + console.log("[aggregator] backfill enqueue complete", { + mode: req.mode, + didCount: dids.length, + jobsEnqueued: enqueued, + }); + } catch (err) { + console.error("[aggregator] backfill aborted", { + mode: req.mode, + error: err instanceof Error ? (err.stack ?? err.message) : String(err), + }); + } +} diff --git a/apps/aggregator/src/pds-verify.ts b/apps/aggregator/src/pds-verify.ts index daf7ec2342..4b31bb0b0a 100644 --- a/apps/aggregator/src/pds-verify.ts +++ b/apps/aggregator/src/pds-verify.ts @@ -18,7 +18,7 @@ */ import type { PublicKey } from "@atcute/crypto"; -import type { AtprotoDid } from "@atcute/lexicons/syntax"; +import { type AtprotoDid, isDid } from "@atcute/lexicons/syntax"; import { verifyRecord } from "@atcute/repo"; const DEFAULT_TIMEOUT_MS = 15_000; @@ -215,6 +215,11 @@ async function fetchCar( * message.retry()` without re-encoding the policy in the catch block. */ function isAtprotoDid(value: string): value is AtprotoDid { + // Library `isDid` enforces the full grammar (length, terminator chars); + // the prefix check then narrows to the atproto-supported method subset + // (`did:plc:` or `did:web:`). A bare prefix check would accept things + // like `did:plc:` (empty body), so the library call carries weight here. + if (!isDid(value)) return false; return value.startsWith("did:plc:") || value.startsWith("did:web:"); } diff --git a/apps/aggregator/src/records-consumer.ts b/apps/aggregator/src/records-consumer.ts index d52810e436..5470c13409 100644 --- a/apps/aggregator/src/records-consumer.ts +++ b/apps/aggregator/src/records-consumer.ts @@ -53,6 +53,7 @@ import { type VerificationFailureReason, type VerifiedPdsRecord, } from "./pds-verify.js"; +import { isPlainObject } from "./utils.js"; /** * Deps the consumer needs at runtime. Constructed once per `processBatch` call @@ -1072,10 +1073,6 @@ function computeVersionSort(version: string): string | null { return `${pad(major)}.${pad(minor)}.${pad(patch)}.${FINAL_VERSION_SENTINEL}`; } -function isPlainObject(value: unknown): value is Record { - return value !== null && typeof value === "object" && !Array.isArray(value); -} - /** * Pull the verified record's CID out of the JSON-stringified * `signature_metadata` column. Returns null if the column is missing or diff --git a/apps/aggregator/src/utils.ts b/apps/aggregator/src/utils.ts new file mode 100644 index 0000000000..ed17cae1a6 --- /dev/null +++ b/apps/aggregator/src/utils.ts @@ -0,0 +1,13 @@ +/** + * Shared utility helpers used across multiple modules. + */ + +/** + * Type guard for narrowing `unknown` to `Record` so + * subsequent `value["key"]` accesses are typesafe without an `as` cast. + * Excludes arrays (which are also `typeof === "object"`) so consumers + * checking for "JSON-shaped object" get what they expect. + */ +export function isPlainObject(value: unknown): value is Record { + return value !== null && typeof value === "object" && !Array.isArray(value); +} diff --git a/apps/aggregator/test/backfill.test.ts b/apps/aggregator/test/backfill.test.ts new file mode 100644 index 0000000000..f8da7ff5d3 --- /dev/null +++ b/apps/aggregator/test/backfill.test.ts @@ -0,0 +1,1088 @@ +/** + * Backfill worker tests. + * + * The worker is plain `fetch` + queue producers; tests stub both. The DID + * resolver is also stubbed (in-memory cache + a simple stub upstream) so the + * tests don't depend on the workers test pool or live PLC. + * + * Architecture under test (post-restructure): + * - `enqueueBackfillJobs`: synchronous fan-out of (DID × WANTED_COLLECTIONS) + * pairs onto BACKFILL_QUEUE, batched at QUEUE_SEND_BATCH_CAP. + * - `processBackfillJob`: per-pair worker. Resolve PDS, paginate + * `com.atproto.repo.listRecords`, batch-enqueue records onto RECORDS_QUEUE. + * - `processBackfillBatch`: queue consumer that calls `processBackfillJob` + * and translates throws to `message.retry()`. + * - `discoverDids`: relay enumeration via `com.atproto.sync.listReposByCollection`. + */ + +import { P256PrivateKeyExportable } from "@atcute/crypto"; +import type { DidDocument } from "@atcute/identity"; +import type { Did } from "@atcute/lexicons/syntax"; +import { applyD1Migrations, env, SELF } from "cloudflare:test"; +import { beforeAll, beforeEach, describe, expect, it } from "vitest"; + +import { drainBackfillDeadLetterBatch, processBackfillBatch } from "../src/backfill-consumer.js"; +import { + type BackfillQueueProducer, + discoverDids, + enqueueBackfillJobs, + MAX_DISCOVERED_DIDS, + processBackfillJob, + QUEUE_SEND_BATCH_CAP, + type RecordsQueueProducer, +} from "../src/backfill.js"; +import { WANTED_COLLECTIONS } from "../src/constants.js"; +import { + type CachedDidDoc, + type DidDocCache, + type DidDocumentResolverLike, + DidResolver, +} from "../src/did-resolver.js"; +import type { BackfillJob, RecordsJob } from "../src/env.js"; +import type { MessageController } from "../src/records-consumer.js"; + +interface TestEnv { + DB: D1Database; + TEST_MIGRATIONS: Parameters[1]; +} +const testEnv = env as unknown as TestEnv; + +const DID_A = "did:plc:test00000000000000000000"; +const DID_B = "did:plc:test00000000000000000001"; +const PDS = "https://pds.test.example"; + +let signingKeyMultibase: string; + +beforeAll(async () => { + const kp = await P256PrivateKeyExportable.createKeypair(); + signingKeyMultibase = await kp.exportPublicKey("multikey"); + await applyD1Migrations(testEnv.DB, testEnv.TEST_MIGRATIONS); +}); + +beforeEach(async () => { + await testEnv.DB.prepare("DELETE FROM known_publishers").run(); +}); + +class CapturingRecordsQueue implements RecordsQueueProducer { + readonly sent: RecordsJob[] = []; + sendBatch(messages: ReadonlyArray<{ body: RecordsJob }>): Promise { + for (const m of messages) this.sent.push(m.body); + return Promise.resolve(); + } +} + +class CapturingBackfillQueue implements BackfillQueueProducer { + readonly sent: BackfillJob[] = []; + readonly batches: number[] = []; + sendBatch(messages: ReadonlyArray<{ body: BackfillJob }>): Promise { + this.batches.push(messages.length); + for (const m of messages) this.sent.push(m.body); + return Promise.resolve(); + } +} + +class MapDidDocCache implements DidDocCache { + private readonly entries = new Map(); + read(did: string): Promise { + return Promise.resolve(this.entries.get(did) ?? null); + } + upsert(did: string, doc: Omit, now: Date): Promise { + this.entries.set(did, { ...doc, resolvedAt: now }); + return Promise.resolve(); + } + expire(did: string): Promise { + const entry = this.entries.get(did); + if (entry) this.entries.set(did, { ...entry, resolvedAt: new Date(0) }); + return Promise.resolve(); + } +} + +class FakeMessage implements MessageController { + acked = 0; + retried = 0; + constructor(readonly body: T) {} + ack() { + this.acked += 1; + } + retry() { + this.retried += 1; + } +} + +function buildResolver(): DidResolver { + const cache = new MapDidDocCache(); + const resolver: DidDocumentResolverLike = { + resolve(did: Did): Promise { + return Promise.resolve({ + id: did as `did:${string}:${string}`, + verificationMethod: [ + { + id: `${did}#atproto`, + type: "Multikey", + controller: did as `did:${string}:${string}`, + publicKeyMultibase: signingKeyMultibase, + }, + ], + service: [ + { + id: "#atproto_pds", + type: "AtprotoPersonalDataServer", + serviceEndpoint: PDS, + }, + ], + }); + }, + }; + return new DidResolver({ cache, resolver, ttlMs: 1_000_000, now: () => new Date() }); +} + +interface MockListRecord { + uri: string; + cid: string; + value: Record; +} + +/** + * Build a fetch stub that returns canned `listRecords` responses keyed by + * collection. Records arrive as a single page; pagination is exercised in a + * dedicated test by passing pages of records explicitly. + */ +function makeFetch( + recordsByCollection: Record, + overrides?: { status?: Record }, +): typeof fetch { + return async (input) => { + const url = + typeof input === "string" + ? new URL(input) + : input instanceof URL + ? input + : new URL(input.url); + if (!url.pathname.endsWith("/xrpc/com.atproto.repo.listRecords")) { + return new Response("not stubbed", { status: 599 }); + } + const collection = url.searchParams.get("collection") ?? ""; + const status = overrides?.status?.[collection]; + if (status !== undefined) { + return new Response(JSON.stringify({ error: "X" }), { + status, + headers: { "content-type": "application/json" }, + }); + } + const records = recordsByCollection[collection] ?? []; + return new Response(JSON.stringify({ records }), { + status: 200, + headers: { "content-type": "application/json" }, + }); + }; +} + +describe("processBackfillJob", () => { + it("enqueues each listRecords result as a RecordsJob", async () => { + const queue = new CapturingRecordsQueue(); + const resolver = buildResolver(); + const collection = WANTED_COLLECTIONS[0]; + if (!collection) throw new Error("test assumes ≥1 collection"); + const fetchImpl = makeFetch({ + [collection]: [ + { + uri: `at://${DID_A}/${collection}/demo`, + cid: "bafyc1", + value: { foo: "bar" }, + }, + ], + }); + + const result = await processBackfillJob( + { did: DID_A, collection }, + { resolver, queue, fetch: fetchImpl }, + ); + + expect(result.enqueued).toBe(1); + expect(result.did).toBe(DID_A); + expect(result.collection).toBe(collection); + expect(queue.sent).toHaveLength(1); + expect(queue.sent[0]).toMatchObject({ + did: DID_A, + collection, + rkey: "demo", + operation: "create", + cid: "bafyc1", + }); + // jetstreamRecord intentionally not set on backfill jobs — the + // consumer's DLQ payload field would otherwise mislabel + // `listRecords` data as Jetstream-supplied data. + expect(queue.sent[0]?.jetstreamRecord).toBeUndefined(); + }); + + it("treats 404 from the PDS as 'no records of this collection', not an error", async () => { + const queue = new CapturingRecordsQueue(); + const resolver = buildResolver(); + const collection = WANTED_COLLECTIONS[0]; + if (!collection) throw new Error("test assumes ≥1 collection"); + const fetchImpl = makeFetch({}, { status: { [collection]: 404 } }); + + const result = await processBackfillJob( + { did: DID_A, collection }, + { resolver, queue, fetch: fetchImpl }, + ); + + expect(result.enqueued).toBe(0); + expect(queue.sent).toHaveLength(0); + }); + + it("throws on non-404 PDS errors so the consumer can retry", async () => { + const queue = new CapturingRecordsQueue(); + const resolver = buildResolver(); + const collection = WANTED_COLLECTIONS[0]; + if (!collection) throw new Error("test assumes ≥1 collection"); + const fetchImpl = makeFetch({}, { status: { [collection]: 503 } }); + + await expect( + processBackfillJob({ did: DID_A, collection }, { resolver, queue, fetch: fetchImpl }), + ).rejects.toThrow(/503/); + }); + + it("throws when the resolver fails (queue consumer translates to retry)", async () => { + const queue = new CapturingRecordsQueue(); + const resolver = new DidResolver({ + cache: new MapDidDocCache(), + resolver: { + resolve: () => Promise.reject(new Error("PLC unreachable")), + }, + }); + const collection = WANTED_COLLECTIONS[0]; + if (!collection) throw new Error("test assumes ≥1 collection"); + const fetchImpl = makeFetch({}); + + await expect( + processBackfillJob({ did: DID_A, collection }, { resolver, queue, fetch: fetchImpl }), + ).rejects.toThrow(/PLC unreachable/); + expect(queue.sent).toHaveLength(0); + }); + + it("paginates listRecords via cursor", async () => { + const queue = new CapturingRecordsQueue(); + const resolver = buildResolver(); + const collection = WANTED_COLLECTIONS[0]; + if (!collection) throw new Error("test assumes ≥1 collection"); + + let calls = 0; + const fetchImpl: typeof fetch = async (input) => { + const url = + typeof input === "string" + ? new URL(input) + : input instanceof URL + ? input + : new URL(input.url); + calls += 1; + const cursor = url.searchParams.get("cursor"); + if (!cursor) { + return new Response( + JSON.stringify({ + records: [{ uri: `at://${DID_A}/${collection}/p1`, cid: "c1", value: {} }], + cursor: "p2", + }), + { status: 200, headers: { "content-type": "application/json" } }, + ); + } + return new Response( + JSON.stringify({ + records: [{ uri: `at://${DID_A}/${collection}/p2`, cid: "c2", value: {} }], + }), + { status: 200, headers: { "content-type": "application/json" } }, + ); + }; + + const result = await processBackfillJob( + { did: DID_A, collection }, + { resolver, queue, fetch: fetchImpl }, + ); + + expect(calls).toBe(2); + expect(queue.sent.map((j) => j.rkey)).toEqual(["p1", "p2"]); + expect(result.enqueued).toBe(2); + }); + + it("skips records whose URI doesn't match the expected collection (defensive)", async () => { + const queue = new CapturingRecordsQueue(); + const resolver = buildResolver(); + const collection = WANTED_COLLECTIONS[0]; + if (!collection) throw new Error("test assumes ≥1 collection"); + const fetchImpl = makeFetch({ + [collection]: [ + { uri: `at://${DID_A}/${collection}/legit`, cid: "c1", value: {} }, + // Buggy PDS: returns a record under the wrong collection. + { uri: `at://${DID_A}/wrong.collection/x`, cid: "c2", value: {} }, + // Buggy URI shape (missing rkey). + { uri: `at://${DID_A}/${collection}/`, cid: "c3", value: {} }, + ], + }); + + const result = await processBackfillJob( + { did: DID_A, collection }, + { resolver, queue, fetch: fetchImpl }, + ); + + expect(queue.sent.map((j) => j.rkey)).toEqual(["legit"]); + expect(result.enqueued).toBe(1); + }); + + it("skips records whose URI references a different DID than the job (defensive)", async () => { + const queue = new CapturingRecordsQueue(); + const resolver = buildResolver(); + const collection = WANTED_COLLECTIONS[0]; + if (!collection) throw new Error("test assumes ≥1 collection"); + // Buggy/malicious PDS: returns a record under a *different* repo's DID. + // Even if everything else parses, that record's signature would be + // from the wrong key — enqueueing it would just churn dead-letters. + const fetchImpl = makeFetch({ + [collection]: [ + { uri: `at://${DID_A}/${collection}/legit`, cid: "c1", value: {} }, + { uri: `at://${DID_B}/${collection}/imposter`, cid: "c2", value: {} }, + ], + }); + + const result = await processBackfillJob( + { did: DID_A, collection }, + { resolver, queue, fetch: fetchImpl }, + ); + + expect(queue.sent.map((j) => j.rkey)).toEqual(["legit"]); + expect(result.enqueued).toBe(1); + }); + + it("rejects records with malformed rkey (atproto rkey grammar violation)", async () => { + const queue = new CapturingRecordsQueue(); + const resolver = buildResolver(); + const collection = WANTED_COLLECTIONS[0]; + if (!collection) throw new Error("test assumes ≥1 collection"); + const fetchImpl = makeFetch({ + [collection]: [ + { uri: `at://${DID_A}/${collection}/legit`, cid: "c1", value: {} }, + { uri: `at://${DID_A}/${collection}/has?queryparam`, cid: "c2", value: {} }, + { uri: `at://${DID_A}/${collection}/has#fragment`, cid: "c3", value: {} }, + { uri: `at://${DID_A}/${collection}/has space`, cid: "c4", value: {} }, + ], + }); + + const result = await processBackfillJob( + { did: DID_A, collection }, + { resolver, queue, fetch: fetchImpl }, + ); + expect(queue.sent.map((j) => j.rkey)).toEqual(["legit"]); + expect(result.enqueued).toBe(1); + }); + + it("end-to-end against the production D1 cache: DID is registered in known_publishers", async () => { + const queue = new CapturingRecordsQueue(); + const { createD1DidDocCache } = await import("../src/did-resolver.js"); + const cache = createD1DidDocCache(testEnv.DB); + const collection = WANTED_COLLECTIONS[0]; + if (!collection) throw new Error("test assumes ≥1 collection"); + const resolver = new DidResolver({ + cache, + resolver: { + resolve: (did) => + Promise.resolve({ + id: did as `did:${string}:${string}`, + verificationMethod: [ + { + id: `${did}#atproto`, + type: "Multikey", + controller: did as `did:${string}:${string}`, + publicKeyMultibase: signingKeyMultibase, + }, + ], + service: [ + { + id: "#atproto_pds", + type: "AtprotoPersonalDataServer", + serviceEndpoint: PDS, + }, + ], + }), + }, + }); + const fetchImpl = makeFetch({}); + + await processBackfillJob({ did: DID_A, collection }, { resolver, queue, fetch: fetchImpl }); + + const row = await testEnv.DB.prepare( + `SELECT did, pds, signing_key, signing_key_id FROM known_publishers WHERE did = ?`, + ) + .bind(DID_A) + .first<{ did: string; pds: string; signing_key: string; signing_key_id: string }>(); + expect(row).toMatchObject({ did: DID_A, pds: PDS }); + expect(row?.signing_key).toBe(signingKeyMultibase); + }); +}); + +describe("processBackfillJob: defenses against malicious / buggy PDS", () => { + it("aborts after MAX_PAGES_PER_COLLECTION when cursor never empties", async () => { + const queue = new CapturingRecordsQueue(); + const resolver = buildResolver(); + const collection = WANTED_COLLECTIONS[0]; + if (!collection) throw new Error("test assumes ≥1 collection"); + // Hostile PDS: returns a different non-empty cursor every call so the + // cursor-equality check doesn't fire — only the page cap stops us. + let counter = 0; + const fetchImpl: typeof fetch = async () => { + counter += 1; + return new Response(JSON.stringify({ records: [], cursor: `cursor-${counter}` }), { + status: 200, + headers: { "content-type": "application/json" }, + }); + }; + + await expect( + processBackfillJob({ did: DID_A, collection }, { resolver, queue, fetch: fetchImpl }), + ).rejects.toThrow(/exceeded/); + // Loop ran at most MAX_PAGES_PER_COLLECTION times. + expect(counter).toBeLessThanOrEqual(1001); + }); + + it("aborts when the PDS returns the identical cursor twice", async () => { + const queue = new CapturingRecordsQueue(); + const resolver = buildResolver(); + const collection = WANTED_COLLECTIONS[0]; + if (!collection) throw new Error("test assumes ≥1 collection"); + // Buggy PDS: echoes the cursor we sent. + let calls = 0; + const fetchImpl: typeof fetch = async () => { + calls += 1; + return new Response(JSON.stringify({ records: [], cursor: "stuck" }), { + status: 200, + headers: { "content-type": "application/json" }, + }); + }; + + await expect( + processBackfillJob({ did: DID_A, collection }, { resolver, queue, fetch: fetchImpl }), + ).rejects.toThrow(/identical cursor/); + expect(calls).toBe(2); // first page, then second page caught the dupe + }); + + it("treats 404 mid-pagination as a partial failure (throws)", async () => { + const queue = new CapturingRecordsQueue(); + const resolver = buildResolver(); + const collection = WANTED_COLLECTIONS[0]; + if (!collection) throw new Error("test assumes ≥1 collection"); + const fetchImpl: typeof fetch = async (input) => { + const url = + typeof input === "string" + ? new URL(input) + : input instanceof URL + ? input + : new URL(input.url); + const cursor = url.searchParams.get("cursor"); + if (!cursor) { + return new Response( + JSON.stringify({ + records: [{ uri: `at://${DID_A}/${collection}/p1`, cid: "c1", value: {} }], + cursor: "p2", + }), + { status: 200, headers: { "content-type": "application/json" } }, + ); + } + return new Response("not found", { status: 404 }); + }; + + await expect( + processBackfillJob({ did: DID_A, collection }, { resolver, queue, fetch: fetchImpl }), + ).rejects.toThrow(/404 mid-pagination/); + }); + + it("rejects pages with > MAX_RECORDS_PER_PAGE records (PDS oversize attack)", async () => { + const queue = new CapturingRecordsQueue(); + const resolver = buildResolver(); + const collection = WANTED_COLLECTIONS[0]; + if (!collection) throw new Error("test assumes ≥1 collection"); + const records = Array.from({ length: 250 }, (_, i) => ({ + uri: `at://${DID_A}/${collection}/r${i}`, + cid: `c${i}`, + value: {}, + })); + const fetchImpl = makeFetch({ [collection]: records }); + + await expect( + processBackfillJob({ did: DID_A, collection }, { resolver, queue, fetch: fetchImpl }), + ).rejects.toThrow(/per-page cap/); + expect(queue.sent).toHaveLength(0); + }); + + it("aborts a hung PDS fetch via the listRecords timeout", async () => { + const queue = new CapturingRecordsQueue(); + const resolver = buildResolver(); + const collection = WANTED_COLLECTIONS[0]; + if (!collection) throw new Error("test assumes ≥1 collection"); + const fetchImpl: typeof fetch = (_input, init) => + new Promise((_resolve, reject) => { + init?.signal?.addEventListener("abort", () => { + reject(new DOMException("aborted", "AbortError")); + }); + }); + + await expect( + processBackfillJob( + { did: DID_A, collection }, + { resolver, queue, fetch: fetchImpl, listRecordsTimeoutMs: 25 }, + ), + ).rejects.toThrow(/timed out after 25ms/); + }); + + it("throws when listRecords body isn't a JSON object (no silent zero)", async () => { + const queue = new CapturingRecordsQueue(); + const resolver = buildResolver(); + const collection = WANTED_COLLECTIONS[0]; + if (!collection) throw new Error("test assumes ≥1 collection"); + const fetchImpl: typeof fetch = async () => + new Response(JSON.stringify(["not", "an", "object"]), { + status: 200, + headers: { "content-type": "application/json" }, + }); + + await expect( + processBackfillJob({ did: DID_A, collection }, { resolver, queue, fetch: fetchImpl }), + ).rejects.toThrow(/not a JSON object/); + }); + + it("throws when listRecords body is missing the records array", async () => { + const queue = new CapturingRecordsQueue(); + const resolver = buildResolver(); + const collection = WANTED_COLLECTIONS[0]; + if (!collection) throw new Error("test assumes ≥1 collection"); + const fetchImpl: typeof fetch = async () => + new Response(JSON.stringify({ cursor: "x" }), { + status: 200, + headers: { "content-type": "application/json" }, + }); + + await expect( + processBackfillJob({ did: DID_A, collection }, { resolver, queue, fetch: fetchImpl }), + ).rejects.toThrow(/missing `records` array/); + }); + + it("throws when cursor is present but not a string (no silent end-of-pagination)", async () => { + const queue = new CapturingRecordsQueue(); + const resolver = buildResolver(); + const collection = WANTED_COLLECTIONS[0]; + if (!collection) throw new Error("test assumes ≥1 collection"); + const fetchImpl: typeof fetch = async () => + new Response(JSON.stringify({ records: [], cursor: 42 }), { + status: 200, + headers: { "content-type": "application/json" }, + }); + + await expect( + processBackfillJob({ did: DID_A, collection }, { resolver, queue, fetch: fetchImpl }), + ).rejects.toThrow(/cursor was not a string/); + }); +}); + +describe("enqueueBackfillJobs", () => { + it("emits one job per (DID × WANTED_COLLECTIONS) pair", async () => { + const queue = new CapturingBackfillQueue(); + const enqueued = await enqueueBackfillJobs([DID_A, DID_B], queue); + + const expected = 2 * WANTED_COLLECTIONS.length; + expect(enqueued).toBe(expected); + expect(queue.sent).toHaveLength(expected); + + // Cartesian shape: every DID appears with every collection exactly once. + for (const did of [DID_A, DID_B]) { + for (const collection of WANTED_COLLECTIONS) { + expect(queue.sent.filter((j) => j.did === did && j.collection === collection)).toHaveLength( + 1, + ); + } + } + }); + + it("batches sendBatch calls at QUEUE_SEND_BATCH_CAP", async () => { + const queue = new CapturingBackfillQueue(); + // Pick enough DIDs that the total job count exceeds the cap. With + // QUEUE_SEND_BATCH_CAP = 100 and 4 collections, 30 DIDs → 120 jobs + // → batches of [100, 20]. + const dids = Array.from( + { length: 30 }, + (_, i) => `did:plc:bulk${i.toString().padStart(20, "0")}`, + ); + const total = dids.length * WANTED_COLLECTIONS.length; + const expectedBatches: number[] = []; + for (let i = 0; i < total; i += QUEUE_SEND_BATCH_CAP) { + expectedBatches.push(Math.min(QUEUE_SEND_BATCH_CAP, total - i)); + } + + const enqueued = await enqueueBackfillJobs(dids, queue); + + expect(enqueued).toBe(total); + expect(queue.batches).toEqual(expectedBatches); + expect(queue.batches.every((n) => n <= QUEUE_SEND_BATCH_CAP)).toBe(true); + }); + + it("emits no batches for an empty DID list", async () => { + const queue = new CapturingBackfillQueue(); + const enqueued = await enqueueBackfillJobs([], queue); + expect(enqueued).toBe(0); + expect(queue.batches).toEqual([]); + expect(queue.sent).toHaveLength(0); + }); +}); + +describe("processBackfillBatch (consumer)", () => { + it("acks each message after a successful per-pair run", async () => { + const recordsQueue = new CapturingRecordsQueue(); + const resolver = buildResolver(); + const collection = WANTED_COLLECTIONS[0]; + if (!collection) throw new Error("test assumes ≥1 collection"); + const fetchImpl = makeFetch({ + [collection]: [{ uri: `at://${DID_A}/${collection}/r1`, cid: "c", value: {} }], + }); + + const message = new FakeMessage({ did: DID_A, collection }); + await processBackfillBatch({ messages: [message] }, {} as Env, { + resolver, + queue: recordsQueue, + fetch: fetchImpl, + }); + + expect(message.acked).toBe(1); + expect(message.retried).toBe(0); + expect(recordsQueue.sent).toHaveLength(1); + }); + + it("retries when processBackfillJob throws (transient PDS failure)", async () => { + const recordsQueue = new CapturingRecordsQueue(); + const resolver = buildResolver(); + const collection = WANTED_COLLECTIONS[0]; + if (!collection) throw new Error("test assumes ≥1 collection"); + const fetchImpl = makeFetch({}, { status: { [collection]: 503 } }); + + const message = new FakeMessage({ did: DID_A, collection }); + await processBackfillBatch({ messages: [message] }, {} as Env, { + resolver, + queue: recordsQueue, + fetch: fetchImpl, + }); + + expect(message.retried).toBe(1); + expect(message.acked).toBe(0); + }); + + it("does not let one failed job poison the rest of the batch", async () => { + const recordsQueue = new CapturingRecordsQueue(); + const resolver = buildResolver(); + const collection = WANTED_COLLECTIONS[0]; + const collection2 = WANTED_COLLECTIONS[1]; + if (!collection || !collection2) throw new Error("test assumes ≥2 collections"); + // First collection 503s; second succeeds with one record. + const fetchImpl: typeof fetch = async (input) => { + const url = + typeof input === "string" + ? new URL(input) + : input instanceof URL + ? input + : new URL(input.url); + const c = url.searchParams.get("collection"); + if (c === collection) return new Response("err", { status: 503 }); + return new Response( + JSON.stringify({ + records: [{ uri: `at://${DID_A}/${c}/r1`, cid: "c", value: {} }], + }), + { status: 200, headers: { "content-type": "application/json" } }, + ); + }; + + const failing = new FakeMessage({ did: DID_A, collection }); + const succeeding = new FakeMessage({ did: DID_A, collection: collection2 }); + await processBackfillBatch({ messages: [failing, succeeding] }, {} as Env, { + resolver, + queue: recordsQueue, + fetch: fetchImpl, + }); + + expect(failing.retried).toBe(1); + expect(failing.acked).toBe(0); + expect(succeeding.acked).toBe(1); + expect(succeeding.retried).toBe(0); + expect(recordsQueue.sent).toHaveLength(1); + expect(recordsQueue.sent[0]?.collection).toBe(collection2); + }); +}); + +describe("drainBackfillDeadLetterBatch", () => { + it("acks every dead-lettered job (DLQ doesn't accumulate)", () => { + const collection = WANTED_COLLECTIONS[0]; + if (!collection) throw new Error("test assumes ≥1 collection"); + const messages = [ + new FakeMessage({ did: DID_A, collection }), + new FakeMessage({ did: DID_B, collection }), + ]; + + drainBackfillDeadLetterBatch({ messages }, {} as Env); + + for (const m of messages) { + expect(m.acked).toBe(1); + expect(m.retried).toBe(0); + } + }); +}); + +describe("discoverDids: listReposByCollection enumeration", () => { + const RELAY = "https://relay.test.example"; + + function makeRelayFetch(reposByCollection: Record>): typeof fetch { + return async (input) => { + const url = + typeof input === "string" + ? new URL(input) + : input instanceof URL + ? input + : new URL(input.url); + if (!url.pathname.endsWith("/xrpc/com.atproto.sync.listReposByCollection")) { + return new Response("not stubbed", { status: 599 }); + } + const collection = url.searchParams.get("collection") ?? ""; + const repos = reposByCollection[collection] ?? []; + return new Response(JSON.stringify({ repos }), { + status: 200, + headers: { "content-type": "application/json" }, + }); + }; + } + + it("returns the union of distinct DIDs across all WANTED_COLLECTIONS", async () => { + const c0 = WANTED_COLLECTIONS[0]; + const c1 = WANTED_COLLECTIONS[1]; + if (!c0 || !c1) throw new Error("test assumes ≥2 collections"); + const fetchImpl = makeRelayFetch({ + [c0]: [{ did: "did:plc:a" }, { did: "did:plc:b" }], + [c1]: [{ did: "did:plc:b" }, { did: "did:plc:c" }], + }); + + const dids = await discoverDids(RELAY, { fetch: fetchImpl }); + + expect(new Set(dids)).toEqual(new Set(["did:plc:a", "did:plc:b", "did:plc:c"])); + }); + + it("paginates via cursor", async () => { + const c0 = WANTED_COLLECTIONS[0]; + if (!c0) throw new Error("test assumes ≥1 collection"); + let calls = 0; + const fetchImpl: typeof fetch = async (input) => { + const url = + typeof input === "string" + ? new URL(input) + : input instanceof URL + ? input + : new URL(input.url); + if (url.searchParams.get("collection") !== c0) { + return new Response(JSON.stringify({ repos: [] }), { status: 200 }); + } + calls += 1; + const cursor = url.searchParams.get("cursor"); + if (!cursor) { + return new Response(JSON.stringify({ repos: [{ did: "did:plc:a" }], cursor: "next" }), { + status: 200, + headers: { "content-type": "application/json" }, + }); + } + return new Response(JSON.stringify({ repos: [{ did: "did:plc:b" }] }), { + status: 200, + headers: { "content-type": "application/json" }, + }); + }; + + const dids = await discoverDids(RELAY, { fetch: fetchImpl }); + + expect(calls).toBe(2); + expect(new Set(dids)).toEqual(new Set(["did:plc:a", "did:plc:b"])); + }); + + it("logs and continues when one collection's listReposByCollection fails", async () => { + const c0 = WANTED_COLLECTIONS[0]; + const c1 = WANTED_COLLECTIONS[1]; + if (!c0 || !c1) throw new Error("test assumes ≥2 collections"); + const fetchImpl: typeof fetch = async (input) => { + const url = + typeof input === "string" + ? new URL(input) + : input instanceof URL + ? input + : new URL(input.url); + const collection = url.searchParams.get("collection"); + if (collection === c0) return new Response("relay broken", { status: 503 }); + if (collection === c1) { + return new Response(JSON.stringify({ repos: [{ did: "did:plc:c1" }] }), { + status: 200, + headers: { "content-type": "application/json" }, + }); + } + return new Response(JSON.stringify({ repos: [] }), { status: 200 }); + }; + + const dids = await discoverDids(RELAY, { fetch: fetchImpl }); + + expect(dids).toContain("did:plc:c1"); + }); + + it("aborts a hung relay fetch via the timeout", async () => { + const fetchImpl: typeof fetch = (_input, init) => + new Promise((_resolve, reject) => { + init?.signal?.addEventListener("abort", () => { + reject(new DOMException("aborted", "AbortError")); + }); + }); + + const dids = await discoverDids(RELAY, { fetch: fetchImpl, timeoutMs: 25 }); + + // Every collection's discovery throws "timed out"; the function returns + // an empty set rather than propagating the error. + expect(dids).toEqual([]); + }); + + it("stops enumerating once MAX_DISCOVERED_DIDS is hit (defense vs runaway relay)", async () => { + const c0 = WANTED_COLLECTIONS[0]; + if (!c0) throw new Error("test assumes ≥1 collection"); + // Relay returns MAX_DISCOVERED_DIDS+50 repos in one page for the first + // collection. discoverDids should add exactly MAX_DISCOVERED_DIDS DIDs + // then stop without paging further or hitting the next collection. + let pdsCalls = 0; + const fetchImpl: typeof fetch = async (input) => { + const url = + typeof input === "string" + ? new URL(input) + : input instanceof URL + ? input + : new URL(input.url); + pdsCalls += 1; + const collection = url.searchParams.get("collection"); + if (collection !== c0) { + // Should never be reached if the cap fires. + return new Response(JSON.stringify({ repos: [{ did: "did:plc:later" }] }), { + status: 200, + headers: { "content-type": "application/json" }, + }); + } + const repos = Array.from({ length: MAX_DISCOVERED_DIDS + 50 }, (_, i) => ({ + did: `did:plc:cap${i.toString().padStart(20, "0")}`, + })); + return new Response(JSON.stringify({ repos }), { + status: 200, + headers: { "content-type": "application/json" }, + }); + }; + + const dids = await discoverDids("https://relay.test.example", { fetch: fetchImpl }); + + expect(dids).toHaveLength(MAX_DISCOVERED_DIDS); + expect(dids).not.toContain("did:plc:later"); + // Only the first collection was queried; the cap fired before reaching + // any subsequent collection. + expect(pdsCalls).toBe(1); + }); +}); + +describe("backfill admin route: auth + input validation", () => { + it("returns 401 when Authorization header is missing", async () => { + const res = await SELF.fetch("https://test/_admin/backfill", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ dids: [DID_A] }), + }); + expect(res.status).toBe(401); + expect(res.headers.get("www-authenticate")).toBe("Bearer"); + }); + + it("returns 401 with wrong token", async () => { + const res = await SELF.fetch("https://test/_admin/backfill", { + method: "POST", + headers: { + "content-type": "application/json", + authorization: "Bearer wrong-token", + }, + body: JSON.stringify({ dids: [DID_A] }), + }); + expect(res.status).toBe(401); + }); + + it("returns 405 on GET", async () => { + const res = await SELF.fetch("https://test/_admin/backfill", { + method: "GET", + headers: { authorization: "Bearer test-admin-token" }, + }); + expect(res.status).toBe(405); + }); + + it("accepts an empty body and triggers discovery (production cold-start path)", async () => { + const res = await SELF.fetch("https://test/_admin/backfill", { + method: "POST", + headers: { + "content-type": "application/json", + authorization: "Bearer test-admin-token", + }, + body: JSON.stringify({}), + }); + expect(res.status).toBe(202); + }); + + it("accepts a literal empty request body (no JSON) and triggers discovery", async () => { + const res = await SELF.fetch("https://test/_admin/backfill", { + method: "POST", + headers: { authorization: "Bearer test-admin-token" }, + }); + expect(res.status).toBe(202); + }); + + it("returns 400 on a non-array `dids` value (string instead of array)", async () => { + const res = await SELF.fetch("https://test/_admin/backfill", { + method: "POST", + headers: { + "content-type": "application/json", + authorization: "Bearer test-admin-token", + }, + body: JSON.stringify({ dids: "did:plc:foo" }), + }); + expect(res.status).toBe(400); + expect(await res.text()).toContain("must be an array"); + }); + + it("returns 400 on empty dids array (suggests omitting the field for discovery)", async () => { + const res = await SELF.fetch("https://test/_admin/backfill", { + method: "POST", + headers: { + "content-type": "application/json", + authorization: "Bearer test-admin-token", + }, + body: JSON.stringify({ dids: [] }), + }); + expect(res.status).toBe(400); + const text = await res.text(); + expect(text).toContain("not be empty"); + expect(text).toMatch(/discover|omit/i); + }); + + it("returns 400 on dids list larger than the cap", async () => { + // Cap is currently 100 (lowered from 1000 because the queue-fan-out + // path amplifies a leaked-token attack). 101 DIDs → over cap. + const dids = Array.from( + { length: 101 }, + (_, i) => `did:plc:test${i.toString().padStart(20, "0")}`, + ); + const res = await SELF.fetch("https://test/_admin/backfill", { + method: "POST", + headers: { + "content-type": "application/json", + authorization: "Bearer test-admin-token", + }, + body: JSON.stringify({ dids }), + }); + expect(res.status).toBe(400); + expect(await res.text()).toContain("at most 100"); + }); + + it("returns 400 on malformed DID (caught by DID_PATTERN)", async () => { + const res = await SELF.fetch("https://test/_admin/backfill", { + method: "POST", + headers: { + "content-type": "application/json", + authorization: "Bearer test-admin-token", + }, + body: JSON.stringify({ dids: ["did:plc:has space"] }), + }); + expect(res.status).toBe(400); + expect(await res.text()).toContain("invalid DID"); + }); + + it("returns 202 with a valid token + body (fires backfill in waitUntil)", async () => { + const res = await SELF.fetch("https://test/_admin/backfill", { + method: "POST", + headers: { + "content-type": "application/json", + authorization: "Bearer test-admin-token", + }, + body: JSON.stringify({ dids: [DID_A] }), + }); + expect(res.status).toBe(202); + }); + + it("dedupes duplicate DIDs in input", async () => { + // We can't assert the dedup directly through SELF without race-y waits, + // but the route accepts the body and returns 202 — the dedup is exercised + // in parseBackfillBody, which is unit-tested via the 'invalid DID' path + // (a duplicate doesn't surface as an error). Smoke test only. + const res = await SELF.fetch("https://test/_admin/backfill", { + method: "POST", + headers: { + "content-type": "application/json", + authorization: "Bearer test-admin-token", + }, + body: JSON.stringify({ dids: [DID_A, DID_A, DID_A] }), + }); + expect(res.status).toBe(202); + }); +}); + +describe("admin auth: scheme + token edge cases", () => { + it("accepts canonical mixed-case Bearer (regression guard)", async () => { + const res = await SELF.fetch("https://test/_admin/start", { + method: "POST", + headers: { authorization: "Bearer test-admin-token" }, + }); + expect(res.status).toBe(204); + }); + + it("accepts uppercase BEARER scheme (RFC 6750 case-insensitive)", async () => { + const res = await SELF.fetch("https://test/_admin/start", { + method: "POST", + headers: { authorization: "BEARER test-admin-token" }, + }); + expect(res.status).toBe(204); + }); + + it("rejects an Authorization header with a non-Bearer scheme", async () => { + const res = await SELF.fetch("https://test/_admin/start", { + method: "POST", + headers: { authorization: "Basic dGVzdC1hZG1pbi10b2tlbjo=" }, + }); + expect(res.status).toBe(401); + }); + + it("rejects empty token after Bearer prefix", async () => { + const res = await SELF.fetch("https://test/_admin/start", { + method: "POST", + headers: { authorization: "Bearer " }, + }); + expect(res.status).toBe(401); + }); +}); + +describe("admin start route: auth + method", () => { + it("returns 405 on GET (POST-only route)", async () => { + const res = await SELF.fetch("https://test/_admin/start"); + expect(res.status).toBe(405); + expect(res.headers.get("allow")).toBe("POST"); + }); + + it("returns 401 on POST without token", async () => { + const res = await SELF.fetch("https://test/_admin/start", { method: "POST" }); + expect(res.status).toBe(401); + }); + + it("returns 204 on POST with valid token", async () => { + const res = await SELF.fetch("https://test/_admin/start", { + method: "POST", + headers: { authorization: "Bearer test-admin-token" }, + }); + expect(res.status).toBe(204); + }); + + it("accepts case-insensitive Bearer scheme (RFC 6750 §2.1)", async () => { + const res = await SELF.fetch("https://test/_admin/start", { + method: "POST", + headers: { authorization: "bearer test-admin-token" }, + }); + expect(res.status).toBe(204); + }); +}); diff --git a/apps/aggregator/vitest.config.ts b/apps/aggregator/vitest.config.ts index 0958ca0b5b..c44bb02004 100644 --- a/apps/aggregator/vitest.config.ts +++ b/apps/aggregator/vitest.config.ts @@ -33,7 +33,15 @@ export default defineConfig({ cloudflareTest({ wrangler: { configPath: "./wrangler.jsonc" }, miniflare: { - bindings: { TEST_MIGRATIONS: migrations }, + bindings: { + TEST_MIGRATIONS: migrations, + // Stub admin auth token so tests can exercise the auth-gated + // admin routes without needing a real secret in the test + // environment. Production deploys pull from + // `wrangler secret put ADMIN_TOKEN`; the value below only + // applies inside the workers test pool. + ADMIN_TOKEN: "test-admin-token", + }, }, }), ], diff --git a/apps/aggregator/worker-configuration.d.ts b/apps/aggregator/worker-configuration.d.ts index 732bb5f4ec..b16346f3b6 100644 --- a/apps/aggregator/worker-configuration.d.ts +++ b/apps/aggregator/worker-configuration.d.ts @@ -1,5 +1,5 @@ /* eslint-disable */ -// Generated by Wrangler by running `wrangler types` (hash: e774431debb87e9b7f5a2fa40582abe1) +// Generated by Wrangler by running `wrangler types` (hash: 9052900c7b8ec62b9504e6508846deb9) // Runtime types generated with workerd@1.20260507.1 2026-02-24 nodejs_compat declare namespace Cloudflare { interface GlobalProps { @@ -9,8 +9,10 @@ declare namespace Cloudflare { interface Env { DB: D1Database; RECORDS_QUEUE: Queue; + BACKFILL_QUEUE: Queue; JETSTREAM_URL: "wss://jetstream2.us-east.bsky.network/subscribe"; - CONSTELLATION_URL: "https://constellation.microcosm.blue"; + RELAY_URL: "https://bsky.network"; + ADMIN_TOKEN: string; RECORDS_DO: DurableObjectNamespace; } } @@ -20,7 +22,7 @@ type StringifyValues> = { }; declare namespace NodeJS { interface ProcessEnv extends StringifyValues< - Pick + Pick > {} } diff --git a/apps/aggregator/wrangler.jsonc b/apps/aggregator/wrangler.jsonc index 416856c222..dd2b164126 100644 --- a/apps/aggregator/wrangler.jsonc +++ b/apps/aggregator/wrangler.jsonc @@ -25,6 +25,16 @@ "binding": "RECORDS_QUEUE", "queue": "emdash-aggregator-records", }, + { + // Backfill orchestrator (POST /_admin/backfill) fans + // (DID, collection) pairs onto this queue. Per-pair work + // (resolve PDS → listRecords → sendBatch onto RECORDS_QUEUE) + // fits well within a single consumer invocation's waitUntil, + // solving the 30s wall-clock cap that would otherwise limit + // us to ~15–25 DIDs per backfill POST. + "binding": "BACKFILL_QUEUE", + "queue": "emdash-aggregator-backfill", + }, ], "consumers": [ { @@ -47,6 +57,35 @@ "max_batch_timeout": 30, "max_retries": 3, }, + { + // Backfill (DID, collection) consumer. Each job: resolve + // PDS, paginate listRecords for one collection, batch + // results onto RECORDS_QUEUE. `max_batch_size: 1` so each + // pair gets its own consumer invocation — listRecords can + // chain ~15s per page × `MAX_PAGES_PER_COLLECTION`, so + // batching multiple pairs per invocation risks blowing + // the consumer wall-clock budget. Throughput is fine + // because backfill is operator-triggered, not steady-state. + "queue": "emdash-aggregator-backfill", + "max_batch_size": 1, + "max_batch_timeout": 5, + "max_retries": 3, + "dead_letter_queue": "emdash-aggregator-backfill-dlq", + }, + { + // Drains the backfill DLQ. Logs each dead-lettered pair + // loud enough that operators tailing logs see it, then + // acks so the DLQ doesn't accumulate unbounded. We don't + // write a forensics row to D1 here (no `backfill_dead_letters` + // table) because the operator's recovery action is + // "re-trigger backfill for the affected DID" — they don't + // need the per-pair payload, just the (did, collection) + // pair, which is on the log line. + "queue": "emdash-aggregator-backfill-dlq", + "max_batch_size": 25, + "max_batch_timeout": 30, + "max_retries": 3, + }, ], }, "durable_objects": { @@ -77,8 +116,30 @@ // Jetstream endpoint. Override per environment for self-hosted relays // or staging backends. "JETSTREAM_URL": "wss://jetstream2.us-east.bsky.network/subscribe", - // Constellation HTTP API for cold-start backfill DID discovery. - "CONSTELLATION_URL": "https://constellation.microcosm.blue", + // Relay base URL implementing `com.atproto.sync.listReposByCollection`, + // used by `/_admin/backfill` to enumerate publishers of our NSIDs at + // cold-start time. Defaults to Bluesky's main relay (the canonical + // atproto relay; same trust orbit as our Jetstream endpoint). + // Override for self-hosted indexers (e.g. Microcosm's Lightrail) + // if/when we want a non-bsky.network discovery path. + "RELAY_URL": "https://bsky.network", + }, + // Required secrets, declared in config so `wrangler types` generates the + // typed binding and `wrangler deploy` validates the secret is set on the + // target Worker before publishing. `wrangler dev` warns at startup when a + // required secret isn't in `.dev.vars` / `.env`. + // + // Set in production with `wrangler secret put ADMIN_TOKEN`. Tests bind + // a stub via miniflare in vitest.config.ts. Local dev pulls from + // `.dev.vars` (gitignored; see `.dev.vars.example`). + // + // The `requireAdminAuth` runtime guard fails closed (503) if the binding + // is somehow empty at request time, so even a misconfigured deploy can't + // leave the admin routes unauth-passable. NOTE: this `secrets` block is + // not inherited by named environments — repeat it under each `env.` + // you add or that env's deploys won't enforce the secret. + "secrets": { + "required": ["ADMIN_TOKEN"], }, "observability": { "enabled": true,