diff --git a/apps/aggregator/migrations/0001_init.sql b/apps/aggregator/migrations/0001_init.sql index 251ea002bc..ed3a6a1f41 100644 --- a/apps/aggregator/migrations/0001_init.sql +++ b/apps/aggregator/migrations/0001_init.sql @@ -58,7 +58,16 @@ CREATE TABLE releases ( verified_at TEXT NOT NULL, tombstoned_at TEXT, -- soft delete (publisher deleted record) PRIMARY KEY (did, package, version), - FOREIGN KEY (did, package) REFERENCES packages(did, slug) + -- ON DELETE CASCADE because Jetstream events for a publisher can arrive + -- in arbitrary order under network reorder. A publisher who deletes their + -- profile (and all releases) emits the events in author-order, but the + -- profile-delete might land at the consumer before the release-deletes. + -- Without cascade, the consumer would have to either skip the profile + -- delete (leaving stale rows) or sequence retries, neither of which is + -- worth the complexity. Releases are version-immutable from a publishing + -- perspective, but a publisher is still entitled to remove their entire + -- package; cascade mirrors that intent. + FOREIGN KEY (did, package) REFERENCES packages(did, slug) ON DELETE CASCADE ); CREATE INDEX idx_releases_latest ON releases(did, package, version_sort DESC) WHERE tombstoned_at IS NULL; @@ -67,16 +76,90 @@ CREATE INDEX idx_releases_cts ON releases(cts); -- Audit trail for rejected duplicate-version attempts. FAIR PR #77 makes -- versions immutable: a second record at the same (did, package, version) is -- rejected at the SQL layer and logged here for forensics. +-- +-- The UNIQUE constraint dedupes attempts by content (CID), not by raw bytes. +-- CAR bytes include the publisher's commit + MST proof which churns whenever +-- the publisher writes any other record in the same repo, so byte-equality +-- would misclassify benign retries as new attempts and bloat the table. +-- The CID is content-addressed and stable for an unchanged record. +-- +-- The consumer's INSERT uses `ON CONFLICT … DO UPDATE SET rejected_at, +-- attempted_record_blob = excluded.{rejected_at, attempted_record_blob}` so +-- the row tracks the latest attempt timestamp + the latest envelope bytes +-- (newer proofs supersede older ones in the forensics column). CREATE TABLE release_duplicate_attempts ( did TEXT NOT NULL, package TEXT NOT NULL, version TEXT NOT NULL, + -- CID of the verified record (stable for content; changes only when the + -- record itself changes). Used as the dedup key. + attempted_cid TEXT NOT NULL, rejected_at TEXT NOT NULL, reason TEXT NOT NULL, - attempted_record_blob BLOB NOT NULL + -- Raw CAR bytes from the most recent attempt. Kept for forensics so + -- operators can inspect what was actually attempted even if the + -- publisher has since deleted the offending record. + attempted_record_blob BLOB NOT NULL, + UNIQUE (did, package, version, attempted_cid) ); -CREATE INDEX idx_release_duplicates ON release_duplicate_attempts(did, package, version); +-- The UNIQUE constraint creates an implicit index on +-- (did, package, version, attempted_record_blob); a separate index on the +-- (did, package, version) prefix is redundant for both lookups (the implicit +-- index handles prefix seeks) and inserts (one fewer index to maintain). + +------------------------------------------------------------------------------ +-- Publishers: identity-level publisher profiles + verification claims +------------------------------------------------------------------------------ + +-- One row per publisher DID (rkey is always literal `self`). Optional: a DID +-- may publish packages without ever publishing a publisher.profile, in which +-- case the row is absent and clients fall back to the handle. This table is +-- the canonical source for "who is publishing these packages?" — distinct from +-- `packages.authors`, which is per-package and remains authoritative for that +-- package. +CREATE TABLE publishers ( + did TEXT PRIMARY KEY, + display_name TEXT NOT NULL, -- bound by verification records — see publisher_verifications + description TEXT, + url TEXT, + contact TEXT, -- JSON array of { kind, url?, email? } + updated_at TEXT, + record_blob BLOB NOT NULL, + signature_metadata TEXT, -- JSON: head CID, signing key id + verified_at TEXT NOT NULL +); + +-- Verification claims: issuer DID vouches for subject DID as a trusted +-- publisher. The rkey is a TID, so an issuer can issue multiple claims (e.g. +-- delegated + official) and we store each as its own row. Validity is bound to +-- the subject's handle + publisher.profile.displayName at issuance time: +-- clients re-resolve those at read time and treat the claim as not in force if +-- either has changed. Ingest stores the facts; the validity check is a +-- query-time concern. +CREATE TABLE publisher_verifications ( + issuer_did TEXT NOT NULL, -- DID of the repo that wrote the record + rkey TEXT NOT NULL, -- TID + subject_did TEXT NOT NULL, + subject_handle TEXT NOT NULL, -- bound at issuance; query-time validity check compares against current + subject_display_name TEXT NOT NULL, -- bound at issuance; query-time validity check compares against current + created_at TEXT NOT NULL, + expires_at TEXT, + record_blob BLOB NOT NULL, + signature_metadata TEXT, + verified_at TEXT NOT NULL, + tombstoned_at TEXT, + PRIMARY KEY (issuer_did, rkey) +); + +-- Hot path: "show me all unexpired, non-tombstoned verifications for subject X". +-- Partial index keeps the index small by excluding tombstoned rows. +CREATE INDEX idx_publisher_verifications_subject ON publisher_verifications(subject_did) + WHERE tombstoned_at IS NULL; + +-- For periodic expiry sweeps. +CREATE INDEX idx_publisher_verifications_expires ON publisher_verifications(expires_at) + WHERE expires_at IS NOT NULL AND tombstoned_at IS NULL; ------------------------------------------------------------------------------ -- Mirror tracking (populated when the artifact mirror lands) @@ -204,10 +287,62 @@ CREATE TABLE ingest_state ( -- Known publisher DIDs we've seen via Jetstream or Constellation. Reconciliation -- iterates this table; cold-start backfill seeds it from Constellation. +-- +-- Doubles as the DID-document resolution cache: `pds`, `signing_key`, +-- `signing_key_id` are populated by the records consumer on first verification +-- and refreshed when `pds_resolved_at` is older than the consumer's TTL +-- (currently 24h, applied at query time as +-- `pds_resolved_at > datetime('now', '-1 day')`). Backfill may insert a row +-- with these fields null; the consumer's first event for that DID forces a +-- resolution and UPDATE. CREATE TABLE known_publishers ( did TEXT PRIMARY KEY, pds TEXT, -- cached PDS endpoint from DID document - pds_resolved_at TEXT, + signing_key TEXT, -- cached #atproto signing key (multibase) + signing_key_id TEXT, -- e.g. 'did:plc:xxx#atproto' + pds_resolved_at TEXT, -- last successful DID-doc resolution first_seen_at TEXT NOT NULL, last_seen_at TEXT NOT NULL ); + +------------------------------------------------------------------------------ +-- Verification-failure forensics +------------------------------------------------------------------------------ + +-- Records that failed PDS-verified ingest (signature, MST proof, AT-URI, +-- lexicon, content-mismatch). Written instead of retrying, because these +-- failures indicate malicious or broken upstream — retrying would just burn +-- PDS round trips. Operators query this table to investigate suspected attacks +-- or upstream regressions; it is NOT used as a retry queue. +-- +-- Distinct from the configured Cloudflare DLQ (`emdash-aggregator-records-dlq`, +-- see wrangler.jsonc), which receives messages after `max_retries` exhausted — +-- that is for transient failures (PDS down, profile-not-yet-arrived). Two +-- distinct failure modes, two distinct destinations. +-- +-- `payload` holds the unverified record bytes from the Jetstream event so an +-- operator can inspect what was attempted without going back to the source PDS. +CREATE TABLE dead_letters ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + did TEXT NOT NULL, + collection TEXT NOT NULL, + rkey TEXT NOT NULL, + -- Reason code; matches the `DeadLetterReason` union in records-consumer.ts. + -- Current values: 'RECORD_NOT_FOUND', 'RESPONSE_TOO_LARGE', 'INVALID_PROOF', + -- 'PDS_HTTP_ERROR', 'LEXICON_VALIDATION_FAILED', 'RKEY_MISMATCH', + -- 'CONTACT_VALIDATION_FAILED', 'INVALID_VERSION', 'UNKNOWN_COLLECTION', + -- 'UNEXPECTED_ERROR'. + reason TEXT NOT NULL, + -- Free-form context (which field, expected vs got, library error message, etc.). + detail TEXT, + -- UTF-8 encoded JSON bytes of `RecordsJob.jetstreamRecord` when present, or a + -- fallback envelope `{operation, cid}` for delete events that don't carry one. + -- Stored as BLOB so future formats (CBOR, raw record bytes) can land here + -- without a schema change; today operators must `CAST(payload AS TEXT)` to + -- read. + payload BLOB NOT NULL, + received_at TEXT NOT NULL DEFAULT (datetime('now')) +); + +CREATE INDEX idx_dead_letters_did ON dead_letters(did); +CREATE INDEX idx_dead_letters_received ON dead_letters(received_at); diff --git a/apps/aggregator/package.json b/apps/aggregator/package.json index 0b3054088b..aa5cc00397 100644 --- a/apps/aggregator/package.json +++ b/apps/aggregator/package.json @@ -22,6 +22,7 @@ "@atcute/client": "catalog:", "@atcute/crypto": "catalog:", "@atcute/firehose": "catalog:", + "@atcute/identity": "catalog:", "@atcute/identity-resolver": "catalog:", "@atcute/jetstream": "catalog:", "@atcute/lexicons": "catalog:", diff --git a/apps/aggregator/src/constants.ts b/apps/aggregator/src/constants.ts index 89b368851e..d0c13f80ee 100644 --- a/apps/aggregator/src/constants.ts +++ b/apps/aggregator/src/constants.ts @@ -9,8 +9,20 @@ * NSIDs the aggregator subscribes to via Jetstream and verifies via PDS * fetches. Will migrate to FAIR-namespaced equivalents once those NSIDs * stabilise. + * + * Two record families: + * - `package.*` — per-package metadata (profile + immutable releases) backing + * the discovery / install path. + * - `publisher.*` — identity-level metadata about the publishing entity + * (`publisher.profile`, rkey `self`) and verification claims about it + * (`publisher.verification`, rkey TID). Verifications are bound to the + * subject's handle + publisher.profile.displayName at issuance time; + * the consumer stores facts as observed and clients re-check validity at + * read time. */ export const WANTED_COLLECTIONS = [ "com.emdashcms.experimental.package.profile", "com.emdashcms.experimental.package.release", + "com.emdashcms.experimental.publisher.profile", + "com.emdashcms.experimental.publisher.verification", ] as const; diff --git a/apps/aggregator/src/did-resolver.ts b/apps/aggregator/src/did-resolver.ts new file mode 100644 index 0000000000..cbdffefd20 --- /dev/null +++ b/apps/aggregator/src/did-resolver.ts @@ -0,0 +1,234 @@ +/** + * DID document resolver with a TTL'd cache backed by `known_publishers`. + * + * The records consumer calls `resolve(did)` once per verification job to learn + * the publisher's PDS endpoint and `#atproto` signing key. The signing key is + * returned as a `PublicKey` instance (from `@atcute/crypto`) ready to hand to + * `verifyRecord` in `@atcute/repo`. + * + * Pure constructor injection — no D1 imports in the class itself, so tests + * pass an in-memory cache and a stub resolver. `createD1DidDocCache(db)` is + * the production binding to `known_publishers`. + */ + +import { + getPublicKeyFromDidController, + P256PublicKey, + Secp256k1PublicKey, + type PublicKey, +} from "@atcute/crypto"; +import { type DidDocument, getAtprotoVerificationMaterial, getPdsEndpoint } from "@atcute/identity"; +import type { Did } 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 + * `importKey` is fast enough that an in-memory `PublicKey` cache isn't worth + * the complexity. */ +export interface CachedDidDoc { + pds: string; + signingKey: string; // multibase + signingKeyId: string; // e.g. 'did:plc:xxx#atproto' + resolvedAt: Date; +} + +export interface DidDocCache { + read(did: string): Promise; + upsert(did: string, doc: Omit, now: Date): Promise; + /** Force the cached row to look stale without disturbing other timestamps + * the cache tracks (e.g. `last_seen_at` in the D1 binding). Used by + * `DidResolver.invalidate()` after a signature failure suggests a key + * rotation. The implementation chooses what "stale" means; the + * Map-backed test cache rewrites `resolvedAt` to epoch, the D1 binding + * sets `pds_resolved_at` only. */ + expire(did: string): Promise; +} + +export interface DidDocumentResolverLike { + resolve(did: Did): Promise; +} + +export interface DidResolverOptions { + cache: DidDocCache; + resolver: DidDocumentResolverLike; + /** Default 24 hours. Cache entries older than this are re-resolved. */ + ttlMs?: number; + /** Injected for deterministic tests. Defaults to `() => new Date()`. */ + now?: () => Date; +} + +export interface ResolvedDidDoc { + pds: string; + publicKey: PublicKey; + signingKeyId: string; +} + +export class DidResolver { + private readonly cache: DidDocCache; + private readonly resolver: DidDocumentResolverLike; + private readonly ttlMs: number; + private readonly now: () => Date; + + constructor(opts: DidResolverOptions) { + this.cache = opts.cache; + this.resolver = opts.resolver; + this.ttlMs = opts.ttlMs ?? DEFAULT_TTL_MS; + this.now = opts.now ?? (() => new Date()); + } + + async resolve(did: string): Promise { + const did_ = asDid(did); + const now = this.now(); + const cached = await this.cache.read(did_); + if (cached && now.getTime() - cached.resolvedAt.getTime() < this.ttlMs) { + return materialise(cached); + } + const doc = await this.resolver.resolve(did_); + const fresh = extractCacheable(doc); + await this.cache.upsert(did_, fresh, now); + return materialise({ ...fresh, resolvedAt: now }); + } + + /** Force a re-resolution next time. Used by the verification path on + * signature failure (the cached signing key may be stale after a + * publisher key rotation). Delegates to the cache's `expire` so other + * timestamps the cache tracks (e.g. `last_seen_at`) aren't disturbed — + * we shouldn't pretend the publisher hasn't been seen since 1970 just + * because we want to drop the cached crypto. */ + async invalidate(did: string): Promise { + await this.cache.expire(asDid(did)); + } +} + +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}`); + } + return did; +} + +function extractCacheable(doc: DidDocument): Omit { + const pds = getPdsEndpoint(doc); + if (!pds) { + throw new Error(`DID document has no atproto PDS service entry: ${doc.id}`); + } + const material = getAtprotoVerificationMaterial(doc); + if (!material) { + throw new Error(`DID document has no #atproto verification method: ${doc.id}`); + } + return { + pds, + signingKey: material.publicKeyMultibase, + // Verification method ids are returned by `getAtprotoVerificationMaterial` + // only as part of the wider doc; reconstruct the canonical id from the + // DID + the well-known fragment. + signingKeyId: `${doc.id}#atproto`, + }; +} + +async function materialise(cached: CachedDidDoc): Promise { + // `getPublicKeyFromDidController` only inspects `publicKeyMultibase`; the + // `type` field is ignored by the parser (the multibase prefix carries the + // curve). Pass a placeholder type — using the actual cached value would + // require persisting it in `known_publishers` for no benefit. + const found = getPublicKeyFromDidController({ + type: "Multikey", + publicKeyMultibase: cached.signingKey, + }); + let publicKey: PublicKey; + if (found.type === "p256") { + publicKey = await P256PublicKey.importRaw(found.publicKeyBytes); + } else if (found.type === "secp256k1") { + publicKey = await Secp256k1PublicKey.importRaw(found.publicKeyBytes); + } else { + // Exhaustiveness check — `FoundPublicKey` is a discriminated union of + // p256 and secp256k1 only. A new variant in a future @atcute/crypto + // release should be handled explicitly. + const _exhaustive: never = found; + throw new Error(`unsupported atproto signing key type`); + } + return { + pds: cached.pds, + publicKey, + signingKeyId: cached.signingKeyId, + }; +} + +/** + * D1-backed cache binding `known_publishers`. Used in production; tests pass + * an in-memory `Map`-backed `DidDocCache` instead. + * + * `first_seen_at` is set on the first insert and preserved on update. Tests + * confirm this — the consumer needs the discovery timestamp to be sticky for + * reconciliation reporting later. + */ +export function createD1DidDocCache(db: D1Database): DidDocCache { + return { + async read(did: string): Promise { + const row = await db + .prepare( + `SELECT pds, signing_key, signing_key_id, pds_resolved_at + FROM known_publishers + WHERE did = ?`, + ) + .bind(did) + .first<{ + pds: string | null; + signing_key: string | null; + signing_key_id: string | null; + pds_resolved_at: string | null; + }>(); + if ( + !row || + row.pds === null || + row.signing_key === null || + row.signing_key_id === null || + row.pds_resolved_at === null + ) { + return null; + } + return { + pds: row.pds, + signingKey: row.signing_key, + signingKeyId: row.signing_key_id, + resolvedAt: new Date(row.pds_resolved_at), + }; + }, + async upsert(did, doc, now): Promise { + const nowIso = now.toISOString(); + await db + .prepare( + `INSERT INTO known_publishers + (did, pds, signing_key, signing_key_id, pds_resolved_at, first_seen_at, last_seen_at) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(did) DO UPDATE SET + pds = excluded.pds, + signing_key = excluded.signing_key, + signing_key_id = excluded.signing_key_id, + pds_resolved_at = excluded.pds_resolved_at, + last_seen_at = excluded.last_seen_at`, + ) + .bind(did, doc.pds, doc.signingKey, doc.signingKeyId, nowIso, nowIso, nowIso) + .run(); + }, + async expire(did): Promise { + // Touches only `pds_resolved_at`; first_seen_at / last_seen_at / + // the cached crypto are intentionally untouched. Setting to epoch + // is unambiguous "older than any plausible TTL". No-op when the + // row doesn't exist. + await db + .prepare( + `UPDATE known_publishers SET pds_resolved_at = '1970-01-01T00:00:00.000Z' + WHERE did = ?`, + ) + .bind(did) + .run(); + }, + }; +} diff --git a/apps/aggregator/src/index.ts b/apps/aggregator/src/index.ts index 32dc1f958d..709f7da6a8 100644 --- a/apps/aggregator/src/index.ts +++ b/apps/aggregator/src/index.ts @@ -16,8 +16,12 @@ */ import type { RecordsJob } from "./env.js"; +import { drainDeadLetterBatch, processBatch } from "./records-consumer.js"; import { RECORDS_DO_NAME } from "./records-do.js"; +const RECORDS_QUEUE_NAME = "emdash-aggregator-records"; +const RECORDS_DLQ_NAME = "emdash-aggregator-records-dlq"; + export { RecordsJetstreamDO } from "./records-do.js"; /** @@ -50,8 +54,20 @@ export default { }); }, - async queue(_batch: MessageBatch, _env: Env, _ctx: ExecutionContext): Promise { - // PDS-verified ingest will land here. + 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. + switch (batch.queue) { + case RECORDS_QUEUE_NAME: + await processBatch(batch, env); + return; + case RECORDS_DLQ_NAME: + await drainDeadLetterBatch(batch, env); + return; + default: + console.error("[aggregator] unknown queue, acking batch", { queue: batch.queue }); + for (const m of batch.messages) m.ack(); + } }, async scheduled(_event: ScheduledEvent, env: Env, ctx: ExecutionContext): Promise { diff --git a/apps/aggregator/src/jetstream-client.ts b/apps/aggregator/src/jetstream-client.ts index 78b606b38d..c3fac47c02 100644 --- a/apps/aggregator/src/jetstream-client.ts +++ b/apps/aggregator/src/jetstream-client.ts @@ -139,11 +139,8 @@ export function wrapAtcuteSubscription( ]); if (result.done) return { value: undefined, done: true }; const event = result.value; - if (event.kind === "commit") { - // Cast within the function: by `kind === "commit"` we - // know the event is a commit; the generic `E` is too - // wide for the compiler to narrow automatically. - return { value: event as unknown as JetstreamCommitEvent, done: false }; + if (isCommitEvent(event)) { + return { value: event, done: false }; } // Skip identity/account events; loop until next commit. } @@ -157,3 +154,45 @@ export function wrapAtcuteSubscription( }, }; } + +/** + * Discriminator + structural predicate that narrows to `JetstreamCommitEvent`. + * + * The runtime check verifies BOTH `kind === "commit"` AND that `commit` is + * present and shaped enough for the ingestor's downstream access (it reads + * `event.commit.collection`, `event.commit.rkey`, `event.commit.operation`, + * `event.commit.cid`). Without the structural check, a producer emitting + * `{kind: "commit"}` with no `commit` field would crash the ingestor on + * access; the cursor wouldn't advance; Jetstream would replay the same + * malformed event forever. + */ +/** Wider parameter type than the bare `{ kind: string }` constraint so the + * predicate can inspect `commit` without an unsafe cast. Any producer + * conforming to `RawJetstreamSubscription` where `E extends { kind: string }` + * is assignable here because `commit` is optional. */ +type MaybeCommitEvent = { + kind: string; + commit?: { + collection?: unknown; + rkey?: unknown; + operation?: unknown; + cid?: unknown; + }; +}; + +const KNOWN_OPERATIONS = new Set(["create", "update", "delete"]); + +function isCommitEvent(event: MaybeCommitEvent): event is JetstreamCommitEvent { + if (event.kind !== "commit" || event.commit === undefined) return false; + const c = event.commit; + if (typeof c.collection !== "string" || typeof c.rkey !== "string") return false; + // Restrict to the operations the downstream RecordsJob + applyDelete + // dispatcher know about. An unknown operation slipping through would + // produce a job the consumer can't process and would land in + // dead_letters as UNEXPECTED_ERROR — better to drop it at the source. + if (typeof c.operation !== "string" || !KNOWN_OPERATIONS.has(c.operation)) return false; + // `cid` is required for create/update (the ingestor reads it into the + // RecordsJob); delete events legitimately have no cid. + if (c.operation !== "delete" && typeof c.cid !== "string") return false; + return true; +} diff --git a/apps/aggregator/src/pds-verify.ts b/apps/aggregator/src/pds-verify.ts new file mode 100644 index 0000000000..daf7ec2342 --- /dev/null +++ b/apps/aggregator/src/pds-verify.ts @@ -0,0 +1,228 @@ +/** + * Fetch + verify a single record from a publisher's PDS. + * + * Two-stage pipeline: + * + * 1. Fetch CAR bytes via `com.atproto.sync.getRecord` against the publisher's + * PDS endpoint (resolved upstream by `DidResolver`). + * 2. Hand the bytes to `@atcute/repo`'s `verifyRecord`, which does MST + * inclusion proof + commit signature verification in one call against the + * publisher's `#atproto` signing key. + * + * Failures are reported via a structured `PdsVerificationError` carrying a + * `reason` code. The consumer decides retry vs. forensics-and-ack based on the + * code (network/5xx → retry; 404, response too large, invalid proof → + * forensics + ack). Doing the classification here keeps the consumer's catch + * block readable and lets future call sites (backfill, reconciliation) reuse + * the same semantics. + */ + +import type { PublicKey } from "@atcute/crypto"; +import type { AtprotoDid } from "@atcute/lexicons/syntax"; +import { verifyRecord } from "@atcute/repo"; + +const DEFAULT_TIMEOUT_MS = 15_000; +/** 5 MB ceiling. Records and their proofs are tiny (sub-KB typical); this is + * a defence against a hostile or broken PDS streaming an unbounded body. */ +const DEFAULT_MAX_RESPONSE_BYTES = 5 * 1024 * 1024; + +export type VerificationFailureReason = + | "PDS_NETWORK_ERROR" + | "PDS_HTTP_ERROR" + | "RECORD_NOT_FOUND" + | "RESPONSE_TOO_LARGE" + | "INVALID_PROOF"; + +export class PdsVerificationError extends Error { + override readonly name = "PdsVerificationError"; + constructor( + readonly reason: VerificationFailureReason, + message: string, + readonly status?: number, + override readonly cause?: unknown, + ) { + super(message); + } +} + +export interface FetchAndVerifyOptions { + pds: string; + did: string; + collection: string; + rkey: string; + publicKey: PublicKey; + /** Default 15s. Aborts the fetch if the PDS is slow. */ + timeoutMs?: number; + /** Default 5 MB. Rejects with `RESPONSE_TOO_LARGE` if exceeded. */ + maxResponseBytes?: number; + /** Inject for tests; defaults to `globalThis.fetch`. */ + fetch?: typeof fetch; +} + +export interface VerifiedPdsRecord { + cid: string; + record: unknown; + /** Raw CAR bytes the PDS served. Stored verbatim in `*.record_blob` so the + * read API can passthrough the signed envelope to clients without re-fetching. */ + carBytes: Uint8Array; +} + +export async function fetchAndVerifyRecord( + opts: FetchAndVerifyOptions, +): Promise { + const fetchImpl = opts.fetch ?? fetch; + const timeoutMs = opts.timeoutMs ?? DEFAULT_TIMEOUT_MS; + const maxResponseBytes = opts.maxResponseBytes ?? DEFAULT_MAX_RESPONSE_BYTES; + + if (!isAtprotoDid(opts.did)) { + // Caller is expected to have validated this upstream (the resolver + // rejects non-DID strings before reaching here), but the verifier's + // type contract is narrower than `string` so guard explicitly. + throw new PdsVerificationError( + "INVALID_PROOF", + `unsupported DID method (expected did:plc or did:web): ${opts.did}`, + ); + } + const url = buildGetRecordUrl(opts.pds, opts.did, opts.collection, opts.rkey); + const carBytes = await fetchCar(fetchImpl, url, timeoutMs, maxResponseBytes); + + try { + const result = await verifyRecord({ + did: opts.did, + collection: opts.collection, + rkey: opts.rkey, + publicKey: opts.publicKey, + carBytes, + }); + return { cid: result.cid, record: result.record, carBytes }; + } catch (err) { + // `verifyRecord` rejects on signature failure, MST proof failure, + // malformed CAR, or rkey/collection mismatch. All four are "drop and + // log" outcomes — distinguishing them isn't load-bearing here, the + // detail goes into the dead_letters detail column for forensics. + throw new PdsVerificationError( + "INVALID_PROOF", + `verifyRecord failed: ${err instanceof Error ? err.message : String(err)}`, + undefined, + err, + ); + } +} + +function buildGetRecordUrl(pds: string, did: string, collection: string, rkey: string): string { + const url = new URL("/xrpc/com.atproto.sync.getRecord", pds); + url.searchParams.set("did", did); + url.searchParams.set("collection", collection); + url.searchParams.set("rkey", rkey); + return url.toString(); +} + +async function fetchCar( + fetchImpl: typeof fetch, + url: string, + timeoutMs: number, + maxBytes: number, +): Promise { + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), timeoutMs); + let response: Response; + try { + response = await fetchImpl(url, { + signal: controller.signal, + headers: { accept: "application/vnd.ipld.car" }, + }); + } catch (err) { + // Whether the fetch threw because we aborted (timeout) or because the + // network failed at the OS layer, the right caller behaviour is the + // same: retry. Lump them under PDS_NETWORK_ERROR. + throw new PdsVerificationError( + "PDS_NETWORK_ERROR", + err instanceof Error && err.name === "AbortError" + ? `PDS fetch aborted after ${timeoutMs}ms` + : `PDS fetch failed: ${err instanceof Error ? err.message : String(err)}`, + undefined, + err, + ); + } finally { + clearTimeout(timer); + } + + if (response.status === 404) { + // Distinct from a generic 4xx. The publisher may have deleted the + // record between Jetstream emitting and us fetching, which is the + // common cause; other 4xx (auth, bad request) suggest programming + // errors. Both end up dead-lettered by the consumer (the audit trail + // is useful even for legitimate races so operators can spot + // systematic Jetstream-vs-PDS skew); the distinct reason code keeps + // them queryable separately. + throw new PdsVerificationError("RECORD_NOT_FOUND", `PDS returned 404 for ${url}`, 404); + } + if (!response.ok) { + throw new PdsVerificationError( + "PDS_HTTP_ERROR", + `PDS returned ${response.status} for ${url}`, + response.status, + ); + } + + // Buffer the body up to the size limit. Don't trust Content-Length; a + // hostile or buggy PDS could under-report and stream more bytes than + // advertised. + const reader = response.body?.getReader(); + if (!reader) { + throw new PdsVerificationError("INVALID_PROOF", "PDS response body is null"); + } + const chunks: Uint8Array[] = []; + let total = 0; + try { + // biome-ignore lint/correctness/noConstantCondition: drains the stream + while (true) { + const { done, value } = await reader.read(); + if (done) break; + total += value.byteLength; + if (total > maxBytes) { + throw new PdsVerificationError( + "RESPONSE_TOO_LARGE", + `PDS response exceeded ${maxBytes} bytes`, + ); + } + chunks.push(value); + } + } catch (err) { + // Cancel the stream so the underlying socket isn't left dangling. + await reader.cancel().catch(() => { + /* swallow — we already have a primary error to surface */ + }); + throw err; + } finally { + reader.releaseLock(); + } + + const out = new Uint8Array(total); + let offset = 0; + for (const chunk of chunks) { + out.set(chunk, offset); + offset += chunk.byteLength; + } + return out; +} + +/** + * Map a `PdsVerificationError.reason` to "should the consumer retry?". Network + * blips and 5xx are transient; everything else is permanent (forensics + ack). + * + * Exposed so the consumer can write `if (isTransient(err.reason, err.status)) + * message.retry()` without re-encoding the policy in the catch block. + */ +function isAtprotoDid(value: string): value is AtprotoDid { + return value.startsWith("did:plc:") || value.startsWith("did:web:"); +} + +export function isTransient( + reason: VerificationFailureReason, + status: number | undefined, +): boolean { + if (reason === "PDS_NETWORK_ERROR") return true; + if (reason === "PDS_HTTP_ERROR" && status !== undefined && status >= 500) return true; + return false; +} diff --git a/apps/aggregator/src/records-consumer.ts b/apps/aggregator/src/records-consumer.ts new file mode 100644 index 0000000000..d52810e436 --- /dev/null +++ b/apps/aggregator/src/records-consumer.ts @@ -0,0 +1,1101 @@ +/** + * Records queue consumer. Replaces the no-op `queue()` handler in `index.ts`. + * + * For each `RecordsJob` from the Records Queue: + * + * 1. `delete` operations short-circuit to a tombstone / hard-delete write. + * 2. `create` / `update` go through the verification pipeline: + * a. Resolve publisher's PDS endpoint + signing key (cached in + * `known_publishers`). + * b. Fetch + verify the record via `com.atproto.sync.getRecord` + * (`@atcute/repo` does MST + signature in one call). + * c. Cross-check the verified record against the Jetstream-supplied copy + * (verified always wins; mismatch is logged as a Jetstream-correctness + * signal). + * d. Lexicon-validate against the generated runtime schema for the + * specific collection. + * e. Per-collection structural checks (rkey-vs-version for releases, + * rkey=='self' for publisher.profile, contact validation, etc.). + * f. Write to D1. + * + * Error policy: + * - Verification failure (signature, MST, lexicon, structural): write a + * `dead_letters` row with the structured reason + payload, ack the message. + * Never retry — these are malicious or broken upstream. + * - Transient PDS failure (network, timeout, 5xx): `retry()` so Cloudflare + * Queues backs off and retries. After `max_retries` (5) it lands in the + * configured DLQ. + * - Unexpected programming errors: log loud, write a `dead_letters` row, + * ack the message. Never crash the worker — that would block the queue. + */ + +import { + AtprotoWebDidDocumentResolver, + CompositeDidDocumentResolver, + PlcDidDocumentResolver, +} from "@atcute/identity-resolver"; +import { safeParse } from "@atcute/lexicons/validations"; +import { + NSID, + PackageProfile, + PackageRelease, + PackageReleaseExtension, + PublisherProfile, + PublisherVerification, +} from "@emdash-cms/registry-lexicons"; + +import { createD1DidDocCache, DidResolver } from "./did-resolver.js"; +import type { RecordsJob } from "./env.js"; +import { + fetchAndVerifyRecord, + isTransient, + PdsVerificationError, + type VerificationFailureReason, + type VerifiedPdsRecord, +} from "./pds-verify.js"; + +/** + * Deps the consumer needs at runtime. Constructed once per `processBatch` call + * (per workerd invocation). Tests inject their own. + */ +export interface ConsumerDeps { + db: D1Database; + resolver: DidResolver; + fetch?: typeof fetch; + now?: () => Date; + /** + * Optional override for the PDS-verification step. Used by tests to inject + * synthetic `VerifiedPdsRecord` payloads without standing up a real CAR + * fixture (the FakePublisher/MockPds toolkit can't run inside the workers + * test pool — see `@atproto/repo` lex-data incompatibility). Defaults to + * `fetchAndVerifyRecord`. + */ + verify?: (opts: { + pds: string; + did: string; + collection: string; + rkey: string; + publicKey: import("@atcute/crypto").PublicKey; + fetch?: typeof fetch; + }) => Promise; +} + +/** Subset of `cloudflare:workers` `Message` we use; defining inline so tests + * don't need to import workerd types. */ +export interface MessageController { + ack(): void; + retry(): void; +} + +/** Subset of a `MessageBatch`. Workers' real batch object satisfies this. */ +export interface MessageBatchLike { + readonly messages: ReadonlyArray; +} + +/** Reason codes written to `dead_letters.reason`. PDS-verification reasons + * pass through verbatim from `PdsVerificationError`; the rest are structural + * checks the consumer enforces locally. */ +export type DeadLetterReason = + // from pds-verify (only the permanent ones — transient ones retry) + | "RECORD_NOT_FOUND" + | "RESPONSE_TOO_LARGE" + | "INVALID_PROOF" + | "PDS_HTTP_ERROR" + // structural checks (consumer-enforced) + | "LEXICON_VALIDATION_FAILED" + | "RKEY_MISMATCH" + | "CONTACT_VALIDATION_FAILED" + | "INVALID_VERSION" + | "UNKNOWN_COLLECTION" + | "UNEXPECTED_ERROR"; + +/** + * Thrown by writers when the input is well-formed but the dependency it + * needs isn't yet present (e.g. a release whose parent profile hasn't been + * ingested). Distinct from `IngestError` because the right action is + * `controller.retry()` — the next attempt may succeed once the parent + * arrives. After Cloudflare Queues exhausts `max_retries` (5) the message + * lands in the configured DLQ and the reconciliation pass picks it up. + */ +export class MissingDependencyError extends Error { + override readonly name = "MissingDependencyError"; +} + +/** Thrown by writers and structural checks. Carries the reason for the + * `dead_letters` row plus an optional human-readable detail. */ +export class IngestError extends Error { + override readonly name = "IngestError"; + constructor( + readonly reason: DeadLetterReason, + message: string, + readonly detail?: string, + ) { + super(message); + } +} + +export async function processBatch( + batch: MessageBatchLike, + env: Env, + depsOverride?: ConsumerDeps, +): Promise { + const deps = depsOverride ?? createProductionDeps(env); + // Process jobs independently — a single failed verification must not fail + // the whole batch and trigger redeliveries for already-acked messages. + // Wrap each call: a `writeDeadLetter` throw inside `processMessage` + // (e.g. transient D1 hiccup mid-batch) would otherwise escape and halt + // the for-loop, leaving subsequent messages without ack/retry. Same + // shape as `drainDeadLetterBatch`. + for (const message of batch.messages) { + try { + await processMessage(message.body, message, deps); + } catch (err) { + console.error("[aggregator] processMessage threw unexpectedly", { + did: message.body.did, + collection: message.body.collection, + rkey: message.body.rkey, + error: err instanceof Error ? err.message : String(err), + }); + message.retry(); + } + } +} + +/** + * Drain the records DLQ. Today's policy is "log + ack" — we record the + * dead-lettered job to Workers logs (and `dead_letters` in D1, for + * structured queryability) so operators can observe drift, and ack the + * message so the DLQ doesn't grow unbounded. The reconciliation pass + * (Slice 1, not yet built) will replace this with retry-from-listRecords; + * until then this prevents legitimate-but-out-of-order messages + * (MissingDependencyError exhausting retries) from being permanently + * dropped without trace. + */ +export async function drainDeadLetterBatch( + batch: MessageBatchLike, + env: Env, +): Promise { + const now = new Date(); + for (const message of batch.messages) { + const job = message.body; + console.warn("[aggregator] DLQ drain: acking job", { + did: job.did, + collection: job.collection, + rkey: job.rkey, + operation: job.operation, + }); + try { + await writeDeadLetter(env.DB, job, "UNEXPECTED_ERROR", "drained from DLQ", now); + message.ack(); + } catch (err) { + // Don't ack on D1 failure — workerd will redeliver per the DLQ + // consumer's max_retries, and after exhaustion the message is + // dropped (no DLQ-of-DLQ). Better to retry than silently lose + // forensics on a transient hiccup. + console.error("[aggregator] DLQ drain: failed to write forensics row, retrying", { + did: job.did, + rkey: job.rkey, + error: err instanceof Error ? err.message : String(err), + }); + message.retry(); + } + } +} + +export async function processMessage( + job: RecordsJob, + controller: MessageController, + deps: ConsumerDeps, +): Promise { + const now = deps.now ?? (() => new Date()); + + if (job.operation === "delete") { + try { + await applyDelete(deps.db, job, now()); + controller.ack(); + } catch (err) { + if (err instanceof IngestError) { + // Structural: unknown collection. Don't retry — the schema + // problem won't fix itself across attempts. Forensics + ack. + await writeDeadLetter(deps.db, job, err.reason, err.detail ?? err.message, now()); + controller.ack(); + return; + } + // Transient (D1 unavailable, etc.) — retry. + console.error("[aggregator] delete failed", { + did: job.did, + collection: job.collection, + rkey: job.rkey, + error: err instanceof Error ? err.message : String(err), + }); + controller.retry(); + } + return; + } + + try { + await verifyAndIngest(job, deps); + controller.ack(); + return; + } catch (err) { + if (err instanceof PdsVerificationError) { + if (isTransient(err.reason, err.status)) { + controller.retry(); + return; + } + // Compute the mapped reason in its own try/catch — `mapPdsReason` + // throws on the supposedly-unreachable PDS_NETWORK_ERROR case; + // without this guard, the throw escapes the catch we're inside + // (function-arg evaluation runs before writeDeadLetter) and the + // whole batch crashes. Fall back to UNEXPECTED_ERROR loudly. + let mapped: DeadLetterReason; + try { + mapped = mapPdsReason(err.reason); + } catch (mapErr) { + console.error("[aggregator] mapPdsReason failed; falling back", { + reason: err.reason, + error: mapErr instanceof Error ? mapErr.message : String(mapErr), + }); + mapped = "UNEXPECTED_ERROR"; + } + await writeDeadLetter(deps.db, job, mapped, err.message, now()); + controller.ack(); + return; + } + if (err instanceof IngestError) { + await writeDeadLetter(deps.db, job, err.reason, err.detail ?? err.message, now()); + controller.ack(); + return; + } + if (err instanceof MissingDependencyError) { + // Out-of-order Jetstream delivery (release before its profile is a + // common case). Retry; after max_retries the message lands in the + // DLQ for the reconciliation pass to recover. + console.warn("[aggregator] missing dependency, retrying", { + did: job.did, + collection: job.collection, + rkey: job.rkey, + reason: err.message, + }); + controller.retry(); + return; + } + // Unexpected — log loud, dead-letter, ack so the queue isn't blocked. + // We don't retry because we have no evidence the next attempt will + // succeed and unbounded retries on a poison message stall the slot. + console.error("[aggregator] unexpected consumer error", { + did: job.did, + collection: job.collection, + rkey: job.rkey, + error: err instanceof Error ? (err.stack ?? err.message) : String(err), + }); + await writeDeadLetter( + deps.db, + job, + "UNEXPECTED_ERROR", + err instanceof Error ? err.message : String(err), + now(), + ); + controller.ack(); + } +} + +async function verifyAndIngest(job: RecordsJob, deps: ConsumerDeps): Promise { + const resolved = await deps.resolver.resolve(job.did); + const verifyFn = deps.verify ?? fetchAndVerifyRecord; + const verified = await verifyFn({ + pds: resolved.pds, + did: job.did, + collection: job.collection, + rkey: job.rkey, + publicKey: resolved.publicKey, + fetch: deps.fetch, + }); + + // Cross-check vs Jetstream copy intentionally omitted: the verified PDS + // copy is canonical and always wins, so the comparison is a monitoring + // signal only. JSON.stringify isn't a canonical comparator — key order + // and undefined-vs-missing differences fire false positives constantly. + // Add a CBOR-canonical comparator when Jetstream-correctness monitoring + // becomes load-bearing; for now the verified copy is what gets written. + + const now = (deps.now ?? (() => new Date()))(); + + switch (job.collection) { + case NSID.packageProfile: + return ingestPackageProfile(deps.db, job, verified, now); + case NSID.packageRelease: + return ingestPackageRelease(deps.db, job, verified, now); + case NSID.publisherProfile: + return ingestPublisherProfile(deps.db, job, verified, now); + case NSID.publisherVerification: + return ingestPublisherVerification(deps.db, job, verified, now); + default: + throw new IngestError( + "UNKNOWN_COLLECTION", + `unsupported collection: ${job.collection}`, + job.collection, + ); + } +} + +// ─── Writers ──────────────────────────────────────────────────────────────── + +export async function ingestPackageProfile( + db: D1Database, + job: RecordsJob, + verified: VerifiedPdsRecord, + now: Date, +): Promise { + const validation = safeParse(PackageProfile.mainSchema, verified.record); + if (!validation.ok) { + throw new IngestError( + "LEXICON_VALIDATION_FAILED", + "package.profile failed lexicon validation", + formatValidationIssues(validation.issues), + ); + } + const record = validation.value; + // Lexicon requires `id` to be the canonical AT URI of the record itself; + // aggregators MUST reject records where it disagrees with the URI we + // fetched from. verifyRecord binds the body to (did, collection, rkey) + // via the MST proof, but the publisher could put a bogus `id` value in + // the body and it would still verify — that's exactly what this check + // catches. + const expectedId = `at://${job.did}/${NSID.packageProfile}/${job.rkey}`; + if (record.id !== expectedId) { + throw new IngestError( + "RKEY_MISMATCH", + `package.profile record.id '${record.id}' does not match AT URI '${expectedId}'`, + ); + } + // Slug is optional — when absent, clients use the rkey as the display + // slug. When present, lexicon requires it to equal the rkey. + if (record.slug !== undefined && record.slug !== job.rkey) { + throw new IngestError( + "RKEY_MISMATCH", + `package.profile rkey '${job.rkey}' does not match record.slug '${record.slug}'`, + ); + } + // Lexicon-can't-express constraint (`profile.json` line 113): each + // security[] entry MUST carry at least one of url/email; aggregators + // MUST reject otherwise. authors[] only SHOULDs the same so we don't + // enforce there. + for (const c of record.security) { + if (!c.url && !c.email) { + throw new IngestError( + "CONTACT_VALIDATION_FAILED", + "package.profile security entry must include at least one of `url` or `email`", + ); + } + } + const slug = record.slug ?? job.rkey; + const sigMeta = JSON.stringify({ cid: verified.cid }); + await db + .prepare( + `INSERT INTO packages + (did, slug, type, name, description, license, authors, security, keywords, sections, + last_updated, latest_version, capabilities, record_blob, signature_metadata, verified_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(did, slug) DO UPDATE SET + type = excluded.type, + name = excluded.name, + description = excluded.description, + license = excluded.license, + authors = excluded.authors, + security = excluded.security, + keywords = excluded.keywords, + sections = excluded.sections, + last_updated = excluded.last_updated, + record_blob = excluded.record_blob, + signature_metadata = excluded.signature_metadata, + verified_at = excluded.verified_at`, + ) + .bind( + job.did, + slug, + record.type, + record.name ?? null, + record.description ?? null, + record.license, + JSON.stringify(record.authors), + JSON.stringify(record.security), + record.keywords ? JSON.stringify(record.keywords) : null, + record.sections ? JSON.stringify(record.sections) : null, + record.lastUpdated ?? null, + null, // latest_version — populated by release writer, not the profile writer + null, // capabilities — populated by release writer + verified.carBytes, + sigMeta, + now.toISOString(), + ) + .run(); +} + +export async function ingestPackageRelease( + db: D1Database, + job: RecordsJob, + verified: VerifiedPdsRecord, + now: Date, +): Promise { + const validation = safeParse(PackageRelease.mainSchema, verified.record); + if (!validation.ok) { + throw new IngestError( + "LEXICON_VALIDATION_FAILED", + "package.release failed lexicon validation", + formatValidationIssues(validation.issues), + ); + } + const record = validation.value; + // Lexicon describes the slug format ("ASCII letter followed by ASCII + // letters, digits, '-', or '_'") but doesn't pin a regex. `record.package` + // must obey the same shape since it's the slug of the parent profile; + // without this check, `package: "foo:bar"` builds an ambiguous rkey + // `foo:bar:1.0.0` indistinguishable from `package: "foo"` + `version: + // "bar:1.0.0"`. Two rows for what looks like the same release. + if (!PACKAGE_SLUG_RE.test(record.package)) { + throw new IngestError( + "RKEY_MISMATCH", + `package.release record.package '${record.package}' must match ${PACKAGE_SLUG_RE}`, + ); + } + const expectedRkey = `${record.package}:${encodeRkeyVersion(record.version)}`; + if (job.rkey !== expectedRkey) { + throw new IngestError( + "RKEY_MISMATCH", + `package.release rkey '${job.rkey}' does not match expected '${expectedRkey}'`, + ); + } + + const versionSort = computeVersionSort(record.version); + if (!versionSort) { + throw new IngestError( + "INVALID_VERSION", + `package.release version '${record.version}' is not parseable as semver`, + ); + } + + // Lexicon mandates releases of type emdash-plugin include a + // releaseExtension entry under the keyed open-union `extensions` map. + // `extensions` is typed as `unknown` in the generated schema so we + // validate the inner shape ourselves; without this, malformed extension + // payloads land in `releases.emdash_extension` and break the read API. + if (!isPlainObject(record.extensions)) { + throw new IngestError( + "LEXICON_VALIDATION_FAILED", + `package.release extensions field must be an object, got ${typeof record.extensions}`, + ); + } + const extension = record.extensions[NSID.packageReleaseExtension]; + if (!extension) { + throw new IngestError( + "LEXICON_VALIDATION_FAILED", + `package.release missing required extensions['${NSID.packageReleaseExtension}']`, + ); + } + const extValidation = safeParse(PackageReleaseExtension.mainSchema, extension); + if (!extValidation.ok) { + throw new IngestError( + "LEXICON_VALIDATION_FAILED", + "package.release releaseExtension failed lexicon validation", + formatValidationIssues(extValidation.issues), + ); + } + + // Parent-profile presence check. The schema's FK would catch this at + // INSERT time, but a raw FK violation surfaces as an opaque error that + // gets dead-lettered as UNEXPECTED_ERROR with no recovery path. Doing + // the lookup explicitly lets us throw MissingDependencyError → retry, + // so out-of-order Jetstream delivery (release before profile) recovers + // once the profile arrives. + const parent = await db + .prepare(`SELECT 1 FROM packages WHERE did = ? AND slug = ?`) + .bind(job.did, record.package) + .first(); + if (!parent) { + throw new MissingDependencyError( + `package.release ${job.rkey} requires profile ${record.package} which is not yet ingested`, + ); + } + + const sigMeta = JSON.stringify({ cid: verified.cid }); + const insertStmt = db + .prepare( + `INSERT INTO releases + (did, package, version, rkey, version_sort, artifacts, requires, suggests, + emdash_extension, repo_url, cts, record_blob, signature_metadata, verified_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(did, package, version) DO NOTHING`, + ) + .bind( + job.did, + record.package, + record.version, + job.rkey, + versionSort, + JSON.stringify(record.artifacts), + record.requires ? JSON.stringify(record.requires) : null, + record.suggests ? JSON.stringify(record.suggests) : null, + // Store only the validated releaseExtension contents, not the + // arbitrary `record.extensions` map. Schema comment for + // `emdash_extension` matches this shape. + JSON.stringify(extValidation.value), + record.repo ?? null, + // cts column intentionally mirrors verified_at: the release lexicon + // has no creation-timestamp field today and the atproto MST commit + // rev isn't surfaced by verifyRecord. Tracked: revisit if the + // lexicon adds a createdAt or @atcute/repo exposes commit metadata. + now.toISOString(), + verified.carBytes, + sigMeta, + now.toISOString(), + ); + + // Atomic with the refresh: D1 wraps a batch in a single transaction. If + // the insert succeeds but the refresh fails (transient D1 hiccup), both + // roll back together and the message retries to a clean state. Without + // the batch, an insert-success / refresh-failure could leave + // `packages.latest_version` permanently stale. + const batchResults = await db.batch([ + insertStmt, + refreshPackageLatestStmt(db, job.did, record.package), + ]); + const insertResult = batchResults[0]; + if (!insertResult) { + // Defensive: D1.batch() guarantees one result per statement; if it + // ever returns fewer, surface loudly rather than silently no-oping. + throw new Error("D1 batch returned no result for INSERT statement"); + } + + if (insertResult.meta.changes === 0) { + // Conflict path. Three sub-cases: + // 1. Same content, not tombstoned → legitimate idempotent replay, + // silent no-op (the refresh in the batch above is a no-op too). + // 2. Same content, tombstoned → publisher re-published a previously + // deleted release; clear the tombstone + refresh so it reappears + // in read results. + // 3. Different content → immutability violation; audit it. + // + // Comparison is on the verified record CID (content-addressed), NOT + // on raw CAR bytes. CARs include the publisher's commit + MST proof, + // which change whenever the publisher writes ANY other record in + // the same repo — so a benign re-fetch of an unchanged record + // produces different bytes. CIDs only change when the record itself + // changes. + const existing = await db + .prepare( + `SELECT signature_metadata, tombstoned_at + FROM releases WHERE did = ? AND package = ? AND version = ?`, + ) + .bind(job.did, record.package, record.version) + .first<{ signature_metadata: string; tombstoned_at: string | null }>(); + if (existing) { + const existingCid = parseCid(existing.signature_metadata); + const sameContent = existingCid === verified.cid; + if (sameContent && existing.tombstoned_at !== null) { + await db.batch([ + db + .prepare( + `UPDATE releases SET tombstoned_at = NULL + WHERE did = ? AND package = ? AND version = ?`, + ) + .bind(job.did, record.package, record.version), + refreshPackageLatestStmt(db, job.did, record.package), + ]); + } else if (!sameContent) { + // On true-duplicate (same hostile content pumped repeatedly, + // distinguished by CID), DO UPDATE so the audit row tracks + // the latest attempt rather than freezing at the first one. + // Operators querying "is this attack ongoing?" read + // `rejected_at` for freshness. The bytes are kept in + // `attempted_record_blob` for forensics — operators can see + // what was actually attempted even if the publisher has + // since deleted the offending record from their PDS. + await db + .prepare( + `INSERT INTO release_duplicate_attempts + (did, package, version, attempted_cid, rejected_at, reason, attempted_record_blob) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(did, package, version, attempted_cid) + DO UPDATE SET + rejected_at = excluded.rejected_at, + attempted_record_blob = excluded.attempted_record_blob`, + ) + .bind( + job.did, + record.package, + record.version, + verified.cid, + now.toISOString(), + "IMMUTABLE_VERSION", + verified.carBytes, + ) + .run(); + } + } + } + + // TODO(slice 3): enqueue artifact-mirror task for this release. Until then, + // `mirrored_artifacts` stays empty for new releases. +} + +/** + * Recompute `packages.latest_version` and `capabilities` from the current + * max-version-sort, non-tombstoned release. + * + * Single UPDATE with correlated subqueries — race-safe under concurrent + * release ingest because the read and write happen in one statement, which + * D1/SQLite serialises against other writers. A naïve SELECT-then-UPDATE + * pair could see two workers each read max=v1, then race to write — final + * value depends on commit order rather than actual current state. + * + * Capabilities are the keys of the latest release's `declaredAccess` map, + * extracted via SQLite's json1 functions. When no release exists or + * `declaredAccess` is missing/non-object, json_each over NULL yields an + * empty set and `json_group_array()` returns `'[]'`. App code treats `'[]'` + * and NULL equivalently for capability filtering. + */ +// The WHERE clause guard prevents the UPDATE from touching `packages` when +// the computed values match what's already stored. Without it, every +// idempotent refresh (the common case for Jetstream redelivery of an +// already-ingested release) fires `packages_au` AFTER UPDATE which DELETEs +// and re-INSERTs into `packages_fts` even though name/description/keywords +// /authors/sections didn't change. SQL-level no-op short-circuit avoids the +// trigger fire entirely. Subqueries duplicate by design — SQLite can share +// them via expression CSE; even un-shared the cost is two extra index seeks +// on a path that already does the same lookups. +// +// The capabilities subquery wraps `json_each` in `(SELECT key ORDER BY key)` +// so the resulting JSON array is order-stable. Without the ORDER BY, +// `json_each` enumerates in unspecified order — `{content, network}` could +// serialise as `["content","network"]` one run and `["network","content"]` +// the next, defeating the IS NOT short-circuit and re-firing the trigger +// on every idempotent refresh. +const REFRESH_PACKAGE_LATEST_SQL = ` + UPDATE packages SET + latest_version = ( + SELECT version FROM releases + WHERE did = packages.did AND package = packages.slug AND tombstoned_at IS NULL + ORDER BY version_sort DESC LIMIT 1 + ), + capabilities = ( + SELECT json_group_array(key) FROM ( + SELECT key FROM json_each( + (SELECT json_extract(emdash_extension, '$.declaredAccess') + FROM releases + WHERE did = packages.did AND package = packages.slug AND tombstoned_at IS NULL + ORDER BY version_sort DESC LIMIT 1) + ) ORDER BY key + ) + ) + WHERE did = ? AND slug = ? + AND ( + latest_version IS NOT ( + SELECT version FROM releases + WHERE did = packages.did AND package = packages.slug AND tombstoned_at IS NULL + ORDER BY version_sort DESC LIMIT 1 + ) + OR capabilities IS NOT ( + SELECT json_group_array(key) FROM ( + SELECT key FROM json_each( + (SELECT json_extract(emdash_extension, '$.declaredAccess') + FROM releases + WHERE did = packages.did AND package = packages.slug AND tombstoned_at IS NULL + ORDER BY version_sort DESC LIMIT 1) + ) ORDER BY key + ) + ) + ) +`; + +function refreshPackageLatestStmt(db: D1Database, did: string, pkg: string): D1PreparedStatement { + return db.prepare(REFRESH_PACKAGE_LATEST_SQL).bind(did, pkg); +} + +export async function ingestPublisherProfile( + db: D1Database, + job: RecordsJob, + verified: VerifiedPdsRecord, + now: Date, +): Promise { + const validation = safeParse(PublisherProfile.mainSchema, verified.record); + if (!validation.ok) { + throw new IngestError( + "LEXICON_VALIDATION_FAILED", + "publisher.profile failed lexicon validation", + formatValidationIssues(validation.issues), + ); + } + const record = validation.value; + if (job.rkey !== "self") { + throw new IngestError( + "RKEY_MISMATCH", + `publisher.profile rkey must be 'self', got '${job.rkey}'`, + ); + } + // Lexicon can't express "at least one of url|email" on contact entries. + // Enforce at the consumer. + for (const c of record.contact ?? []) { + if (!c.url && !c.email) { + throw new IngestError( + "CONTACT_VALIDATION_FAILED", + "publisher.profile contact entry must include at least one of `url` or `email`", + ); + } + } + const sigMeta = JSON.stringify({ cid: verified.cid }); + await db + .prepare( + `INSERT INTO publishers + (did, display_name, description, url, contact, updated_at, + record_blob, signature_metadata, verified_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(did) DO UPDATE SET + display_name = excluded.display_name, + description = excluded.description, + url = excluded.url, + contact = excluded.contact, + updated_at = excluded.updated_at, + record_blob = excluded.record_blob, + signature_metadata = excluded.signature_metadata, + verified_at = excluded.verified_at`, + ) + .bind( + job.did, + record.displayName, + record.description ?? null, + record.url ?? null, + record.contact ? JSON.stringify(record.contact) : null, + record.updatedAt ?? null, + verified.carBytes, + sigMeta, + now.toISOString(), + ) + .run(); +} + +export async function ingestPublisherVerification( + db: D1Database, + job: RecordsJob, + verified: VerifiedPdsRecord, + now: Date, +): Promise { + const validation = safeParse(PublisherVerification.mainSchema, verified.record); + if (!validation.ok) { + throw new IngestError( + "LEXICON_VALIDATION_FAILED", + "publisher.verification failed lexicon validation", + formatValidationIssues(validation.issues), + ); + } + const record = validation.value; + const sigMeta = JSON.stringify({ cid: verified.cid }); + await db + .prepare( + `INSERT INTO publisher_verifications + (issuer_did, rkey, subject_did, subject_handle, subject_display_name, + created_at, expires_at, record_blob, signature_metadata, verified_at, tombstoned_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL) + ON CONFLICT(issuer_did, rkey) DO UPDATE SET + subject_did = excluded.subject_did, + subject_handle = excluded.subject_handle, + subject_display_name = excluded.subject_display_name, + created_at = excluded.created_at, + expires_at = excluded.expires_at, + record_blob = excluded.record_blob, + signature_metadata = excluded.signature_metadata, + verified_at = excluded.verified_at, + tombstoned_at = NULL`, + ) + .bind( + job.did, + job.rkey, + record.subject, + record.handle, + record.displayName, + record.createdAt, + record.expiresAt ?? null, + verified.carBytes, + sigMeta, + now.toISOString(), + ) + .run(); +} + +// ─── Delete handling ──────────────────────────────────────────────────────── + +export async function applyDelete(db: D1Database, job: RecordsJob, now: Date): Promise { + switch (job.collection) { + case NSID.packageProfile: + // Hard-delete the profile. The releases FK is ON DELETE CASCADE + // (see `0001_init.sql`), so all of this publisher's releases for + // this slug are removed in the same statement. CASCADE is the + // right semantic when the publisher's intent is "the whole + // package goes away" — and crucially, it lets out-of-order + // Jetstream delivery work: a profile-delete arriving before its + // release-deletes doesn't fail with FK violation. Audit history + // for those releases lives only in `release_duplicate_attempts` + // (for prior immutability violations) and `dead_letters` (for + // prior verification failures); the canonical release rows are + // gone with the profile. + await db + .prepare(`DELETE FROM packages WHERE did = ? AND slug = ?`) + .bind(job.did, job.rkey) + .run(); + return; + case NSID.packageRelease: { + // Releases are version-immutable but a publisher CAN delete them + // (yanking from the source). Soft-delete: read APIs filter on + // `tombstoned_at IS NULL` so they disappear from listings. + // Parse rkey back to (package, version) so we hit the PK index + // instead of the partial idx_releases_latest, which has to scan + // for did=? then filter by rkey. + const parsed = parseReleaseRkey(job.rkey); + if (!parsed) { + // Surface the malformed delete so an operator investigating + // "why didn't this delete take effect?" has an audit trail. + // IngestError → dispatcher writes a dead_letters row + acks. + throw new IngestError( + "RKEY_MISMATCH", + `package.release delete with malformed rkey: '${job.rkey}'`, + ); + } + // Batch tombstone + refresh: D1 wraps in a single transaction so + // the visible-release set and the denormalised latest_version + // commit together. Refresh runs even on idempotent re-deletes + // (changes=0) — refresh is itself idempotent and the cost is one + // extra UPDATE that touches no row when state is already correct. + await db.batch([ + db + .prepare( + `UPDATE releases SET tombstoned_at = ? + WHERE did = ? AND package = ? AND version = ? AND tombstoned_at IS NULL`, + ) + .bind(now.toISOString(), job.did, parsed.pkg, parsed.version), + refreshPackageLatestStmt(db, job.did, parsed.pkg), + ]); + return; + } + case NSID.publisherProfile: + // Hard-delete; one-per-DID, no audit value in retaining it. + await db.prepare(`DELETE FROM publishers WHERE did = ?`).bind(job.did).run(); + return; + case NSID.publisherVerification: + // Soft-delete to preserve the audit trail. `(issuer_did, rkey)` + // is the AT-URI primary key. + await db + .prepare( + `UPDATE publisher_verifications SET tombstoned_at = ? + WHERE issuer_did = ? AND rkey = ? AND tombstoned_at IS NULL`, + ) + .bind(now.toISOString(), job.did, job.rkey) + .run(); + return; + default: + // Reach here only if a future collection is added to + // WANTED_COLLECTIONS without an applyDelete arm. Throw so the + // dispatcher writes a dead_letters row instead of silently + // dropping the delete — silent drops let the table drift out of + // sync with the publisher's repo until someone notices the + // inconsistency. + throw new IngestError( + "UNKNOWN_COLLECTION", + `delete for unhandled collection: ${job.collection}`, + job.collection, + ); + } +} + +// ─── Forensics ───────────────────────────────────────────────────────────── + +async function writeDeadLetter( + db: D1Database, + job: RecordsJob, + reason: DeadLetterReason, + detail: string | null, + now: Date, +): Promise { + // `payload` holds whatever Jetstream gave us, encoded as JSON. If the job + // didn't carry a jetstreamRecord (delete operations don't), store the + // envelope of operation+cid so the row is still inspectable. + const payload = JSON.stringify(job.jetstreamRecord ?? { operation: job.operation, cid: job.cid }); + const payloadBytes = new TextEncoder().encode(payload); + await db + .prepare( + `INSERT INTO dead_letters + (did, collection, rkey, reason, detail, payload, received_at) + VALUES (?, ?, ?, ?, ?, ?, ?)`, + ) + .bind(job.did, job.collection, job.rkey, reason, detail, payloadBytes, now.toISOString()) + .run(); +} + +// ─── Production wiring ───────────────────────────────────────────────────── + +function createProductionDeps(env: Env): ConsumerDeps { + const composite = new CompositeDidDocumentResolver({ + methods: { + plc: new PlcDidDocumentResolver(), + web: new AtprotoWebDidDocumentResolver(), + }, + }); + return { + db: env.DB, + resolver: new DidResolver({ + cache: createD1DidDocCache(env.DB), + resolver: composite, + }), + }; +} + +// ─── Helpers ─────────────────────────────────────────────────────────────── + +/** + * Translate a permanent `PdsVerificationError.reason` to its `DeadLetterReason` + * counterpart. The parameter type is the imported `VerificationFailureReason` + * union so a new variant added in `pds-verify.ts` becomes a compile-time + * error here. Transient reasons (PDS_NETWORK_ERROR today) are unreachable + * because the caller filters them via `isTransient`; we throw rather than + * silently dead-letter to surface the broken invariant loudly. + */ +function mapPdsReason(reason: VerificationFailureReason): DeadLetterReason { + switch (reason) { + case "RECORD_NOT_FOUND": + case "RESPONSE_TOO_LARGE": + case "INVALID_PROOF": + case "PDS_HTTP_ERROR": + return reason; + case "PDS_NETWORK_ERROR": + throw new Error( + "unreachable: PDS_NETWORK_ERROR should have been retried by isTransient before reaching mapPdsReason", + ); + default: { + const exhaustive: never = reason; + throw new Error(`unhandled PdsVerificationError reason: ${String(exhaustive)}`); + } + } +} + +/** + * Encode a semver version string for use in the release rkey per the lexicon's + * `:` rule. Atproto rkeys allow `[A-Za-z0-9._~-]`; + * semver versions can include `+` for build metadata which must be + * percent-encoded. Our lexicon disallows `+` so this is conservative. + */ +const PACKAGE_SLUG_RE = /^[a-zA-Z][a-zA-Z0-9_-]*$/; +const PLUS_RE = /\+/g; +function encodeRkeyVersion(version: string): string { + return version.replace(PLUS_RE, "%2B"); +} + +/** + * Parse a release rkey of the form `:` back into + * its components. Validates BOTH parts against the regexes the consumer uses + * elsewhere — a malformed rkey like `demo:1.0.0:extra` would otherwise split + * to `pkg="demo"`, `version="1.0.0:extra"` and silently no-op the delete (no + * row matches an illegal version), losing the audit trail. + * + * Returns null on any malformation. Callers decide what to do — `applyDelete` + * throws `IngestError("RKEY_MISMATCH", …)` to surface the malformed delete + * in `dead_letters`. + * + * `decodeURIComponent` throws URIError on malformed `%`-escapes (e.g. + * `1.0.0%XX`). Caught here so the function's contract is "returns null OR a + * validated pair, never throws". + */ +function parseReleaseRkey(rkey: string): { pkg: string; version: string } | null { + const idx = rkey.indexOf(":"); + if (idx <= 0 || idx === rkey.length - 1) return null; + const pkg = rkey.slice(0, idx); + if (!PACKAGE_SLUG_RE.test(pkg)) return null; + const encodedVersion = rkey.slice(idx + 1); + let version: string; + try { + version = decodeURIComponent(encodedVersion); + } catch { + return null; + } + if (!SEMVER_RE.test(version)) return null; + return { pkg, version }; +} + +const SEMVER_RE = /^(\d+)\.(\d+)\.(\d+)(?:-([0-9A-Za-z.-]+))?$/; +const NUMERIC_RE = /^\d+$/; +const pad = (s: string) => s.padStart(10, "0"); + +/** + * Pre-compute a fixed-width sortable string for a semver version. + * + * Format: `<10-digit-major>.<10-digit-minor>.<10-digit-patch>.` + * + * - Numeric components are zero-padded to 10 digits so lexicographic sort + * matches numeric order ('1.10.0' > '1.9.0'). + * - The final-release sentinel `~` is one character above the prerelease + * alphabet (`[0-9A-Za-z-]`, max char `z` at ASCII 122; `~` at 126), so + * any final release sorts after any prerelease at the same + * major.minor.patch — including pathological prereleases like + * `1.0.0-zzzz` that would beat a `zzz` sentinel by being longer. + * - Within a prerelease tag, numeric identifiers are zero-padded too. This + * matches semver precedence rules approximately; the "numeric < non-numeric" + * wrinkle isn't fully captured but the typical patterns ('rc.1', 'beta.2') + * sort correctly. + * + * Returns null when the input doesn't parse as our supported semver subset + * (the lexicon disallows build metadata '+...'). + */ +const FINAL_VERSION_SENTINEL = "~"; +function computeVersionSort(version: string): string | null { + const m = SEMVER_RE.exec(version); + if (!m) return null; + const major = m[1] ?? "0"; + const minor = m[2] ?? "0"; + const patch = m[3] ?? "0"; + const pre = m[4]; + if (major.length > 10 || minor.length > 10 || patch.length > 10) { + // 10-digit pad ceiling — versions past this can't be ordered correctly + // by simple zero-padding. ~10 billion is well past reasonable. + return null; + } + if (pre) { + const parts = pre.split("."); + const padded: string[] = []; + for (const p of parts) { + if (NUMERIC_RE.test(p)) { + if (p.length > 10) return null; // same ceiling for prerelease numerics + padded.push(pad(p)); + } else { + padded.push(p); + } + } + return `${pad(major)}.${pad(minor)}.${pad(patch)}.${padded.join(".")}`; + } + 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 + * malformed (which shouldn't happen in writer-controlled data, but the + * fallback keeps the comparison robust against future schema drift). + */ +function parseCid(signatureMetadata: string): string | null { + try { + const parsed: unknown = JSON.parse(signatureMetadata); + if (isPlainObject(parsed) && typeof parsed.cid === "string") return parsed.cid; + } catch { + // fall through + } + return null; +} + +function formatValidationIssues(issues: unknown): string { + try { + return JSON.stringify(issues); + } catch { + return String(issues); + } +} diff --git a/apps/aggregator/src/records-do.ts b/apps/aggregator/src/records-do.ts index 008c6d5c98..3d8b49cfad 100644 --- a/apps/aggregator/src/records-do.ts +++ b/apps/aggregator/src/records-do.ts @@ -22,7 +22,7 @@ import { JetstreamIngestor, type IngestorStorage } from "./jetstream-ingestor.js /** Singleton DO ID. There's exactly one ingestor per deployment. */ export const RECORDS_DO_NAME = "main"; -export class RecordsJetstreamDO extends DurableObject { +export class RecordsJetstreamDO extends DurableObject { private readonly ingestor: JetstreamIngestor; /** Held so the run loop isn't garbage-collected. */ private readonly runPromise: Promise; diff --git a/apps/aggregator/test/did-resolver.test.ts b/apps/aggregator/test/did-resolver.test.ts new file mode 100644 index 0000000000..dc17abce86 --- /dev/null +++ b/apps/aggregator/test/did-resolver.test.ts @@ -0,0 +1,367 @@ +/** + * DidResolver unit tests + the D1 binding's contract test. + * + * The class is exercised end-to-end with a real WebCrypto-backed signing key + * (generated once in `beforeAll`) and a Map-backed cache so cache behaviour is + * verified independently of D1 wiring. A separate suite runs the D1 binding + * against the test pool's in-memory D1 and re-runs the cache contract via + * the same scenarios — that way the contract is the same in tests and in + * production. + */ + +import { P256PrivateKeyExportable } from "@atcute/crypto"; +import type { DidDocument } from "@atcute/identity"; +import type { Did } from "@atcute/lexicons/syntax"; +import { applyD1Migrations, env } from "cloudflare:test"; +import { beforeAll, beforeEach, describe, expect, it } from "vitest"; + +import { + type CachedDidDoc, + createD1DidDocCache, + type DidDocCache, + type DidDocumentResolverLike, + DidResolver, +} from "../src/did-resolver.js"; + +interface TestEnv { + DB: D1Database; + TEST_MIGRATIONS: Parameters[1]; +} + +const testEnv = env as unknown as TestEnv; + +const TEST_DID = "did:plc:test00000000000000000000"; +const TEST_PDS = "https://pds.test.example"; + +let signingKeyMultibase: string; + +beforeAll(async () => { + const kp = await P256PrivateKeyExportable.createKeypair(); + signingKeyMultibase = await kp.exportPublicKey("multikey"); +}); + +class MapDidDocCache implements DidDocCache { + private readonly entries = new Map(); + readonly reads: string[] = []; + readonly upserts: Array<{ did: string; doc: Omit; now: Date }> = []; + readonly expires: string[] = []; + + read(did: string): Promise { + this.reads.push(did); + return Promise.resolve(this.entries.get(did) ?? null); + } + upsert(did: string, doc: Omit, now: Date): Promise { + this.upserts.push({ did, doc, now }); + this.entries.set(did, { ...doc, resolvedAt: now }); + return Promise.resolve(); + } + expire(did: string): Promise { + this.expires.push(did); + const entry = this.entries.get(did); + if (entry) { + this.entries.set(did, { ...entry, resolvedAt: new Date(0) }); + } + return Promise.resolve(); + } +} + +class StubResolver implements DidDocumentResolverLike { + readonly calls: Did[] = []; + private response: DidDocument; + private error: Error | null = null; + + constructor(response: DidDocument) { + this.response = response; + } + + resolve(did: Did): Promise { + this.calls.push(did); + if (this.error) return Promise.reject(this.error); + return Promise.resolve(this.response); + } + + setResponse(doc: DidDocument): void { + this.response = doc; + } + setError(err: Error): void { + this.error = err; + } +} + +function buildDidDoc(overrides: Partial = {}): DidDocument { + return { + id: TEST_DID as `did:${string}:${string}`, + verificationMethod: [ + { + id: `${TEST_DID}#atproto`, + type: "Multikey", + controller: TEST_DID as `did:${string}:${string}`, + publicKeyMultibase: signingKeyMultibase, + }, + ], + service: [ + { + id: "#atproto_pds", + type: "AtprotoPersonalDataServer", + serviceEndpoint: TEST_PDS, + }, + ], + ...overrides, + }; +} + +describe("DidResolver", () => { + describe("with in-memory cache", () => { + it("resolves on cache miss, writes the row, returns a usable PublicKey", async () => { + const cache = new MapDidDocCache(); + const resolver = new StubResolver(buildDidDoc()); + const subject = new DidResolver({ cache, resolver, now: () => new Date(1000) }); + + const result = await subject.resolve(TEST_DID); + + expect(result.pds).toBe(TEST_PDS); + expect(result.signingKeyId).toBe(`${TEST_DID}#atproto`); + expect(typeof result.publicKey.verify).toBe("function"); + expect(resolver.calls).toEqual([TEST_DID]); + expect(cache.upserts).toHaveLength(1); + expect(cache.upserts[0]).toMatchObject({ + did: TEST_DID, + doc: { pds: TEST_PDS, signingKey: signingKeyMultibase }, + }); + }); + + it("hits cache on second call within TTL — no resolver call", async () => { + const cache = new MapDidDocCache(); + const resolver = new StubResolver(buildDidDoc()); + const subject = new DidResolver({ + cache, + resolver, + ttlMs: 60_000, + now: () => new Date(1000), + }); + + await subject.resolve(TEST_DID); + await subject.resolve(TEST_DID); + + expect(resolver.calls).toHaveLength(1); + expect(cache.upserts).toHaveLength(1); + }); + + it("re-resolves when cached entry is past TTL", async () => { + const cache = new MapDidDocCache(); + const resolver = new StubResolver(buildDidDoc()); + let now = 1_000; + const subject = new DidResolver({ + cache, + resolver, + ttlMs: 60_000, + now: () => new Date(now), + }); + + await subject.resolve(TEST_DID); + now = 1_000 + 60_001; + await subject.resolve(TEST_DID); + + expect(resolver.calls).toHaveLength(2); + expect(cache.upserts).toHaveLength(2); + }); + + it("propagates resolver errors without writing to cache", async () => { + const cache = new MapDidDocCache(); + const resolver = new StubResolver(buildDidDoc()); + resolver.setError(new Error("plc unreachable")); + const subject = new DidResolver({ cache, resolver }); + + await expect(subject.resolve(TEST_DID)).rejects.toThrow("plc unreachable"); + expect(cache.upserts).toHaveLength(0); + }); + + it("rejects DID documents with no PDS service entry", async () => { + const cache = new MapDidDocCache(); + const resolver = new StubResolver(buildDidDoc({ service: [] })); + const subject = new DidResolver({ cache, resolver }); + + await expect(subject.resolve(TEST_DID)).rejects.toThrow(/no atproto PDS/i); + }); + + it("rejects DID documents with no #atproto verification method", async () => { + const cache = new MapDidDocCache(); + const resolver = new StubResolver(buildDidDoc({ verificationMethod: [] })); + const subject = new DidResolver({ cache, resolver }); + + await expect(subject.resolve(TEST_DID)).rejects.toThrow(/#atproto verification method/i); + }); + + it("rejects malformed DIDs without calling the resolver or cache", async () => { + const cache = new MapDidDocCache(); + const resolver = new StubResolver(buildDidDoc()); + const subject = new DidResolver({ cache, resolver }); + + await expect(subject.resolve("not-a-did")).rejects.toThrow(/invalid DID/i); + expect(resolver.calls).toHaveLength(0); + expect(cache.reads).toHaveLength(0); + }); + + it("invalidate() forces re-resolution on the next call", async () => { + const cache = new MapDidDocCache(); + const resolver = new StubResolver(buildDidDoc()); + // Use a real-world `now` so invalidate's "epoch" sentinel falls + // well outside the TTL window. + const subject = new DidResolver({ + cache, + resolver, + ttlMs: 60_000, + now: () => new Date("2026-05-09T12:00:00.000Z"), + }); + + await subject.resolve(TEST_DID); + await subject.invalidate(TEST_DID); + await subject.resolve(TEST_DID); + + expect(resolver.calls).toHaveLength(2); + }); + + it("invalidate() on an unknown DID is a no-op", async () => { + const cache = new MapDidDocCache(); + const resolver = new StubResolver(buildDidDoc()); + const subject = new DidResolver({ cache, resolver }); + + await expect(subject.invalidate(TEST_DID)).resolves.toBeUndefined(); + expect(cache.upserts).toHaveLength(0); + }); + }); + + describe("createD1DidDocCache", () => { + beforeAll(async () => { + await applyD1Migrations(testEnv.DB, testEnv.TEST_MIGRATIONS); + }); + + beforeEach(async () => { + await testEnv.DB.exec("DELETE FROM known_publishers"); + }); + + it("returns null when no row exists", async () => { + const cache = createD1DidDocCache(testEnv.DB); + expect(await cache.read(TEST_DID)).toBeNull(); + }); + + it("returns null when row exists but cache fields are unpopulated", async () => { + // Backfill or future code may insert a known_publishers row before the + // consumer has resolved the DID doc; the read should treat that as + // a cache miss, not a stale cache hit. + const now = new Date().toISOString(); + await testEnv.DB.prepare( + `INSERT INTO known_publishers (did, first_seen_at, last_seen_at) + VALUES (?, ?, ?)`, + ) + .bind(TEST_DID, now, now) + .run(); + + const cache = createD1DidDocCache(testEnv.DB); + expect(await cache.read(TEST_DID)).toBeNull(); + }); + + it("upsert + read round-trips a cache entry", async () => { + const cache = createD1DidDocCache(testEnv.DB); + const now = new Date("2026-05-09T12:00:00.000Z"); + await cache.upsert( + TEST_DID, + { + pds: TEST_PDS, + signingKey: signingKeyMultibase, + signingKeyId: `${TEST_DID}#atproto`, + }, + now, + ); + + const out = await cache.read(TEST_DID); + expect(out).not.toBeNull(); + expect(out?.pds).toBe(TEST_PDS); + expect(out?.signingKey).toBe(signingKeyMultibase); + expect(out?.signingKeyId).toBe(`${TEST_DID}#atproto`); + expect(out?.resolvedAt.toISOString()).toBe(now.toISOString()); + }); + + it("upsert preserves first_seen_at across updates", async () => { + const cache = createD1DidDocCache(testEnv.DB); + const t1 = new Date("2026-05-09T12:00:00.000Z"); + const t2 = new Date("2026-05-10T12:00:00.000Z"); + + await cache.upsert( + TEST_DID, + { + pds: TEST_PDS, + signingKey: signingKeyMultibase, + signingKeyId: `${TEST_DID}#atproto`, + }, + t1, + ); + await cache.upsert( + TEST_DID, + { + pds: TEST_PDS, + signingKey: signingKeyMultibase, + signingKeyId: `${TEST_DID}#atproto`, + }, + t2, + ); + + const row = await testEnv.DB.prepare( + `SELECT first_seen_at, last_seen_at, pds_resolved_at FROM known_publishers WHERE did = ?`, + ) + .bind(TEST_DID) + .first<{ first_seen_at: string; last_seen_at: string; pds_resolved_at: string }>(); + expect(row?.first_seen_at).toBe(t1.toISOString()); + expect(row?.last_seen_at).toBe(t2.toISOString()); + expect(row?.pds_resolved_at).toBe(t2.toISOString()); + }); + + it("end-to-end: resolver wired to D1 cache", async () => { + const cache = createD1DidDocCache(testEnv.DB); + const resolver = new StubResolver(buildDidDoc()); + const subject = new DidResolver({ cache, resolver }); + + await subject.resolve(TEST_DID); + await subject.resolve(TEST_DID); + + // One resolver call for the cache miss; the second resolve hits D1. + expect(resolver.calls).toHaveLength(1); + }); + + it("expire only touches pds_resolved_at; preserves first_seen_at and last_seen_at", async () => { + const cache = createD1DidDocCache(testEnv.DB); + const seenAt = new Date("2026-05-09T12:00:00.000Z"); + await cache.upsert( + TEST_DID, + { + pds: TEST_PDS, + signingKey: signingKeyMultibase, + signingKeyId: `${TEST_DID}#atproto`, + }, + seenAt, + ); + + await cache.expire(TEST_DID); + + const row = await testEnv.DB.prepare( + `SELECT first_seen_at, last_seen_at, pds_resolved_at FROM known_publishers WHERE did = ?`, + ) + .bind(TEST_DID) + .first<{ first_seen_at: string; last_seen_at: string; pds_resolved_at: string }>(); + expect(row?.first_seen_at).toBe(seenAt.toISOString()); + expect(row?.last_seen_at).toBe(seenAt.toISOString()); + // pds_resolved_at gets pushed to epoch so the next resolve() + // sees the row as stale per the TTL check. + expect(row?.pds_resolved_at).toBe("1970-01-01T00:00:00.000Z"); + }); + + it("expire on an unknown DID is a no-op (no row appears)", async () => { + const cache = createD1DidDocCache(testEnv.DB); + await cache.expire(TEST_DID); + const row = await testEnv.DB.prepare(`SELECT did FROM known_publishers WHERE did = ?`) + .bind(TEST_DID) + .first(); + expect(row).toBeNull(); + }); + }); +}); diff --git a/apps/aggregator/test/jetstream-client.test.ts b/apps/aggregator/test/jetstream-client.test.ts index b54e81e606..ea98fd179b 100644 --- a/apps/aggregator/test/jetstream-client.test.ts +++ b/apps/aggregator/test/jetstream-client.test.ts @@ -95,11 +95,24 @@ describe("wrapAtcuteSubscription", () => { }); it("filters non-commit events", async () => { - const events: Array<{ kind: string; commit?: { collection: string } }> = [ + // `isCommitEvent` requires the full commit shape (collection + rkey + + // operation) — a `{kind: "commit"}` envelope without a structurally + // valid `commit` object is correctly rejected as "malformed", so the + // stub must mirror what production producers emit. + const events: Array<{ + kind: string; + commit?: { collection: string; rkey: string; operation: string; cid?: string }; + }> = [ { kind: "identity" }, - { kind: "commit", commit: { collection: "x" } }, + { + kind: "commit", + commit: { collection: "x", rkey: "r1", operation: "create", cid: "bafyc1" }, + }, { kind: "account" }, - { kind: "commit", commit: { collection: "y" } }, + { + kind: "commit", + commit: { collection: "y", rkey: "r2", operation: "create", cid: "bafyc2" }, + }, ]; let i = 0; const sub: RawJetstreamSubscription<(typeof events)[number]> = { @@ -118,4 +131,87 @@ describe("wrapAtcuteSubscription", () => { expect(out).toHaveLength(2); expect(out.every((e) => (e as { kind: string }).kind === "commit")).toBe(true); }); + + it("rejects commits with missing cid on non-delete operations", async () => { + // `create`/`update` events without a `cid` would produce a RecordsJob + // with `cid: undefined`, breaking the consumer's verification step. + // Predicate must drop them at the source. + const events = [ + { kind: "commit", commit: { collection: "x", rkey: "r1", operation: "create" } }, + { + kind: "commit", + commit: { collection: "x", rkey: "r2", operation: "update" }, + }, + ]; + let i = 0; + const sub: RawJetstreamSubscription<(typeof events)[number]> = { + cursor: 0, + [Symbol.asyncIterator]: () => ({ + async next() { + if (i >= events.length) return { value: undefined, done: true }; + const value = events[i++]; + return { value: value as (typeof events)[number], done: false }; + }, + }), + }; + const handle = wrapAtcuteSubscription(sub); + const out: unknown[] = []; + for await (const event of handle) out.push(event); + expect(out).toHaveLength(0); + }); + + it("rejects commits whose operation isn't one of create/update/delete", async () => { + // A producer emitting an unknown operation would otherwise produce a + // RecordsJob the consumer can't handle, ending up as + // UNEXPECTED_ERROR in dead_letters. Better to drop at the source. + const events = [ + { + kind: "commit", + commit: { + collection: "x", + rkey: "r1", + operation: "rebase", // not a real atproto op + cid: "bafyc1", + }, + }, + ]; + let i = 0; + const sub: RawJetstreamSubscription<(typeof events)[number]> = { + cursor: 0, + [Symbol.asyncIterator]: () => ({ + async next() { + if (i >= events.length) return { value: undefined, done: true }; + const value = events[i++]; + return { value: value as (typeof events)[number], done: false }; + }, + }), + }; + const handle = wrapAtcuteSubscription(sub); + const out: unknown[] = []; + for await (const event of handle) out.push(event); + expect(out).toHaveLength(0); + }); + + it("accepts delete commits without cid", async () => { + // Delete events legitimately have no cid; predicate must let them + // through. + const events = [ + { kind: "commit", commit: { collection: "x", rkey: "r1", operation: "delete" } }, + ]; + let i = 0; + const sub: RawJetstreamSubscription<(typeof events)[number]> = { + cursor: 0, + [Symbol.asyncIterator]: () => ({ + async next() { + if (i >= events.length) return { value: undefined, done: true }; + const value = events[i++]; + return { value: value as (typeof events)[number], done: false }; + }, + }), + }; + const handle = wrapAtcuteSubscription(sub); + const out: unknown[] = []; + for await (const event of handle) out.push(event); + expect(out).toHaveLength(1); + }); }); diff --git a/apps/aggregator/test/pds-verify.test.ts b/apps/aggregator/test/pds-verify.test.ts new file mode 100644 index 0000000000..68c2055d10 --- /dev/null +++ b/apps/aggregator/test/pds-verify.test.ts @@ -0,0 +1,174 @@ +/** + * pds-verify unit tests. + * + * Cover the HTTP / error-shaping logic with a stub `fetch`. The actual + * verification handoff to `@atcute/repo`'s `verifyRecord` is NOT exercised + * end-to-end anywhere in this suite — building a valid signed CAR by hand + * would re-implement what `@atcute/repo` already tests internally, and the + * consumer-level test path stubs verification via `ConsumerDeps.verify` + * (the FakePublisher / MockPds fixture from `@emdash-cms/atproto-test-utils` + * can't load inside `@cloudflare/vitest-pool-workers` due to a transitive + * `@atproto/lex-data` incompatibility; see records-consumer test header). + * + * What we DO test here is the surface every reason code can be reached + * through, plus the `isTransient` policy mapping the consumer relies on. + */ + +import { P256PublicKey, P256PrivateKeyExportable } from "@atcute/crypto"; +import { beforeAll, describe, expect, it } from "vitest"; + +import { fetchAndVerifyRecord, isTransient, PdsVerificationError } from "../src/pds-verify.js"; + +const TEST_DID = "did:plc:test00000000000000000000"; +const TEST_PDS = "https://pds.test.example"; + +let publicKey: P256PublicKey; + +beforeAll(async () => { + const kp = await P256PrivateKeyExportable.createKeypair(); + const raw = await kp.exportPublicKey("raw"); + publicKey = await P256PublicKey.importRaw(raw); +}); + +async function captureRejection(promise: Promise): Promise { + try { + await promise; + } catch (err) { + if (err instanceof PdsVerificationError) return err; + throw err; + } + throw new Error("expected promise to reject with PdsVerificationError"); +} + +function buildOpts(overrides: { + fetch: typeof fetch; + timeoutMs?: number; + maxResponseBytes?: number; +}) { + return { + pds: TEST_PDS, + did: TEST_DID, + collection: "com.emdashcms.experimental.package.profile", + rkey: "demo", + publicKey, + ...overrides, + }; +} + +describe("fetchAndVerifyRecord — HTTP path", () => { + it("builds the canonical sync.getRecord URL with did/collection/rkey", async () => { + let observedUrl: string | undefined; + const fetchImpl: typeof fetch = async (input) => { + observedUrl = + typeof input === "string" ? input : input instanceof URL ? input.href : input.url; + return new Response(new Uint8Array([0]), { status: 200 }); + }; + await fetchAndVerifyRecord(buildOpts({ fetch: fetchImpl })).catch(() => { + /* verifyRecord rejects on the dummy bytes — we only care about the URL */ + }); + expect(observedUrl).toBe( + `${TEST_PDS}/xrpc/com.atproto.sync.getRecord?did=${encodeURIComponent(TEST_DID)}&collection=com.emdashcms.experimental.package.profile&rkey=demo`, + ); + }); + + it("maps a network error to PDS_NETWORK_ERROR", async () => { + const fetchImpl: typeof fetch = () => Promise.reject(new TypeError("connection refused")); + await expect(fetchAndVerifyRecord(buildOpts({ fetch: fetchImpl }))).rejects.toMatchObject({ + name: "PdsVerificationError", + reason: "PDS_NETWORK_ERROR", + }); + }); + + it("maps an aborted fetch (timeout) to PDS_NETWORK_ERROR with the timeout in the message", async () => { + const fetchImpl: typeof fetch = (_input, init) => { + return new Promise((_resolve, reject) => { + init?.signal?.addEventListener("abort", () => { + const err = new DOMException("aborted", "AbortError"); + reject(err); + }); + }); + }; + const err = await captureRejection( + fetchAndVerifyRecord(buildOpts({ fetch: fetchImpl, timeoutMs: 25 })), + ); + expect(err.reason).toBe("PDS_NETWORK_ERROR"); + expect(err.message).toMatch(/aborted after 25ms/); + }); + + it("maps a 404 to RECORD_NOT_FOUND with status", async () => { + const fetchImpl: typeof fetch = () => Promise.resolve(new Response("", { status: 404 })); + const err = await captureRejection(fetchAndVerifyRecord(buildOpts({ fetch: fetchImpl }))); + expect(err.reason).toBe("RECORD_NOT_FOUND"); + expect(err.status).toBe(404); + }); + + it("maps a 500 to PDS_HTTP_ERROR with status", async () => { + const fetchImpl: typeof fetch = () => Promise.resolve(new Response("", { status: 503 })); + const err = await captureRejection(fetchAndVerifyRecord(buildOpts({ fetch: fetchImpl }))); + expect(err.reason).toBe("PDS_HTTP_ERROR"); + expect(err.status).toBe(503); + }); + + it("maps a non-404 4xx to PDS_HTTP_ERROR with status", async () => { + const fetchImpl: typeof fetch = () => Promise.resolve(new Response("", { status: 401 })); + const err = await captureRejection(fetchAndVerifyRecord(buildOpts({ fetch: fetchImpl }))); + expect(err.reason).toBe("PDS_HTTP_ERROR"); + expect(err.status).toBe(401); + }); + + it("rejects responses larger than maxResponseBytes with RESPONSE_TOO_LARGE", async () => { + const big = new Uint8Array(64); + big.fill(0xff); + const fetchImpl: typeof fetch = () => Promise.resolve(new Response(big, { status: 200 })); + const err = await captureRejection( + fetchAndVerifyRecord(buildOpts({ fetch: fetchImpl, maxResponseBytes: 16 })), + ); + expect(err.reason).toBe("RESPONSE_TOO_LARGE"); + }); + + it("rejects a null body with INVALID_PROOF", async () => { + const fetchImpl: typeof fetch = () => { + // Construct a Response with a null body. The Response constructor + // allows null for HEAD-style responses; we never get null in + // practice but the guard is defensive. + return Promise.resolve(new Response(null, { status: 200 })); + }; + const err = await captureRejection(fetchAndVerifyRecord(buildOpts({ fetch: fetchImpl }))); + expect(err.reason).toBe("INVALID_PROOF"); + }); + + it("hands successful body bytes to verifyRecord (which rejects malformed input as INVALID_PROOF)", async () => { + // Random bytes are guaranteed not to parse as a valid CAR. The point + // of the test is that we got past HTTP and INTO verifyRecord — and + // that verifyRecord's rejection is wrapped as INVALID_PROOF. + const garbage = new Uint8Array([1, 2, 3, 4, 5]); + const fetchImpl: typeof fetch = () => Promise.resolve(new Response(garbage, { status: 200 })); + const err = await captureRejection(fetchAndVerifyRecord(buildOpts({ fetch: fetchImpl }))); + expect(err.reason).toBe("INVALID_PROOF"); + expect(err.cause).toBeDefined(); + }); +}); + +describe("isTransient policy", () => { + it("network errors retry", () => { + expect(isTransient("PDS_NETWORK_ERROR", undefined)).toBe(true); + }); + it("HTTP 5xx retries", () => { + expect(isTransient("PDS_HTTP_ERROR", 500)).toBe(true); + expect(isTransient("PDS_HTTP_ERROR", 503)).toBe(true); + }); + it("HTTP 4xx is permanent", () => { + expect(isTransient("PDS_HTTP_ERROR", 401)).toBe(false); + expect(isTransient("PDS_HTTP_ERROR", 400)).toBe(false); + }); + it("missing status on PDS_HTTP_ERROR is treated as permanent", () => { + // Defensive: PDS_HTTP_ERROR is always raised with a status, but the + // policy must not blow up if a future code path drops it. + expect(isTransient("PDS_HTTP_ERROR", undefined)).toBe(false); + }); + it("404, oversized response, and invalid proof are permanent", () => { + expect(isTransient("RECORD_NOT_FOUND", 404)).toBe(false); + expect(isTransient("RESPONSE_TOO_LARGE", undefined)).toBe(false); + expect(isTransient("INVALID_PROOF", undefined)).toBe(false); + }); +}); diff --git a/apps/aggregator/test/records-consumer.test.ts b/apps/aggregator/test/records-consumer.test.ts new file mode 100644 index 0000000000..48c07b53ad --- /dev/null +++ b/apps/aggregator/test/records-consumer.test.ts @@ -0,0 +1,1743 @@ +/** + * Records consumer tests. + * + * Three layers of coverage: + * + * 1. **Writer unit tests** call the per-collection ingest functions directly + * with synthetic `VerifiedPdsRecord` payloads. This sidesteps the PDS + * verification step (already tested in pds-verify.test.ts) and the + * in-workerd unavailability of the FakePublisher fixture (which depends + * on `@atproto/repo`, a Node-only package). What we test here is the + * structural validation + D1 write SQL, against a real D1 instance. + * + * 2. **Delete tests** call `applyDelete` with each collection and assert the + * right tombstone / hard-delete behaviour. + * + * 3. **Dispatcher tests** drive `processMessage` with stub deps to cover + * ack/retry decisions: transient PDS errors retry, permanent errors + * forensics+ack, IngestError forensics+ack, unexpected errors + * forensics+ack, success acks. The verify path is stubbed via a + * drop-in `DidResolver` and a `fetch` that throws controlled errors; + * end-to-end success-path verification will land in a follow-up PR + * once a node-pool integration test config is in place. + */ + +import { P256PrivateKeyExportable } from "@atcute/crypto"; +import type { DidDocument } from "@atcute/identity"; +import type { Did } from "@atcute/lexicons/syntax"; +import { NSID } from "@emdash-cms/registry-lexicons"; +import { applyD1Migrations, env } from "cloudflare:test"; +import { beforeAll, beforeEach, describe, expect, it } from "vitest"; + +import { + type DidDocCache, + type DidDocumentResolverLike, + DidResolver, +} from "../src/did-resolver.js"; +import type { RecordsJob } from "../src/env.js"; +import { PdsVerificationError, type VerifiedPdsRecord } from "../src/pds-verify.js"; +import { + applyDelete, + type ConsumerDeps, + IngestError, + ingestPackageProfile, + ingestPackageRelease, + ingestPublisherProfile, + ingestPublisherVerification, + type MessageController, + processMessage, +} 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"; + +let signingKeyMultibase: string; + +beforeAll(async () => { + const kp = await P256PrivateKeyExportable.createKeypair(); + signingKeyMultibase = await kp.exportPublicKey("multikey"); + await applyD1Migrations(testEnv.DB, testEnv.TEST_MIGRATIONS); +}); + +beforeEach(async () => { + for (const table of [ + "release_duplicate_attempts", + "releases", + "packages", + "publisher_verifications", + "publishers", + "known_publishers", + "dead_letters", + ]) { + await testEnv.DB.prepare(`DELETE FROM ${table}`).run(); + } +}); + +function fakeVerified(record: unknown): VerifiedPdsRecord { + return { + cid: "bafyreigtest00000000000000000000000000000000000000000000", + record, + carBytes: new Uint8Array([0xde, 0xad, 0xbe, 0xef]), + }; +} + +function jobFor( + did: string, + collection: string, + rkey: string, + overrides: Partial = {}, +): RecordsJob { + return { + did, + collection, + rkey, + operation: "create", + cid: "bafyreigtest00000000000000000000000000000000000000000000", + ...overrides, + }; +} + +const NOW = new Date("2026-05-09T12:00:00.000Z"); + +// ─── Writer: package.profile ──────────────────────────────────────────────── + +describe("ingestPackageProfile", () => { + const validRecord = { + $type: NSID.packageProfile, + id: `at://${DID_A}/${NSID.packageProfile}/demo`, + slug: "demo", + type: "emdash-plugin", + license: "MIT", + authors: [{ name: "Tester" }], + security: [{ email: "x@y.test" }], + }; + + it("inserts a row on first call", async () => { + const job = jobFor(DID_A, NSID.packageProfile, "demo"); + await ingestPackageProfile(testEnv.DB, job, fakeVerified(validRecord), NOW); + + const row = await testEnv.DB.prepare(`SELECT did, slug, license FROM packages WHERE did = ?`) + .bind(DID_A) + .first<{ did: string; slug: string; license: string }>(); + expect(row).toMatchObject({ did: DID_A, slug: "demo", license: "MIT" }); + }); + + it("upserts on second call with edited record", async () => { + const job = jobFor(DID_A, NSID.packageProfile, "demo"); + await ingestPackageProfile(testEnv.DB, job, fakeVerified(validRecord), NOW); + await ingestPackageProfile( + testEnv.DB, + job, + fakeVerified({ ...validRecord, license: "Apache-2.0" }), + NOW, + ); + + const row = await testEnv.DB.prepare(`SELECT license FROM packages WHERE did = ?`) + .bind(DID_A) + .first<{ license: string }>(); + expect(row?.license).toBe("Apache-2.0"); + }); + + it("rejects when rkey ≠ record.slug", async () => { + const job = jobFor(DID_A, NSID.packageProfile, "different"); + await expect( + ingestPackageProfile(testEnv.DB, job, fakeVerified(validRecord), NOW), + ).rejects.toMatchObject({ name: "IngestError", reason: "RKEY_MISMATCH" }); + }); + + it("rejects records that don't match the lexicon", async () => { + const job = jobFor(DID_A, NSID.packageProfile, "demo"); + await expect( + ingestPackageProfile( + testEnv.DB, + job, + fakeVerified({ slug: "demo" /* missing required */ }), + NOW, + ), + ).rejects.toMatchObject({ name: "IngestError", reason: "LEXICON_VALIDATION_FAILED" }); + }); +}); + +// ─── Writer: package.release ──────────────────────────────────────────────── + +describe("ingestPackageRelease", () => { + const validProfile = { + $type: NSID.packageProfile, + id: `at://${DID_A}/${NSID.packageProfile}/demo`, + slug: "demo", + type: "emdash-plugin", + license: "MIT", + authors: [{ name: "Tester" }], + security: [{ email: "x@y.test" }], + }; + + function makeRelease(version: string) { + return { + $type: NSID.packageRelease, + package: "demo", + version, + artifacts: { + package: { url: "https://example.com/demo.tgz", checksum: "bsha256-abc" }, + }, + extensions: { + "com.emdashcms.experimental.package.releaseExtension": { + $type: "com.emdashcms.experimental.package.releaseExtension", + declaredAccess: {}, + }, + }, + }; + } + + beforeEach(async () => { + // Releases reference packages via FK; seed the parent profile. + await ingestPackageProfile( + testEnv.DB, + jobFor(DID_A, NSID.packageProfile, "demo"), + fakeVerified(validProfile), + NOW, + ); + }); + + it("inserts a release with computed version_sort", async () => { + const release = makeRelease("1.10.0"); + const job = jobFor(DID_A, NSID.packageRelease, "demo:1.10.0"); + await ingestPackageRelease(testEnv.DB, job, fakeVerified(release), NOW); + + const row = await testEnv.DB.prepare( + `SELECT version, version_sort FROM releases WHERE did = ? AND version = ?`, + ) + .bind(DID_A, "1.10.0") + .first<{ version: string; version_sort: string }>(); + expect(row?.version).toBe("1.10.0"); + // 1.10.0 must sort after 1.9.0 — the whole point of version_sort. + expect(row?.version_sort.startsWith("0000000001.0000000010.")).toBe(true); + }); + + it("rejects when rkey ≠ ':'", async () => { + const release = makeRelease("1.0.0"); + const job = jobFor(DID_A, NSID.packageRelease, "wrong-rkey"); + await expect( + ingestPackageRelease(testEnv.DB, job, fakeVerified(release), NOW), + ).rejects.toMatchObject({ reason: "RKEY_MISMATCH" }); + }); + + it("rejects unparseable semver versions", async () => { + const release = makeRelease("not-a-version"); + const job = jobFor(DID_A, NSID.packageRelease, "demo:not-a-version"); + // Lexicon validation accepts any 1-64 char string in `version`; the + // semver parse failure is what catches non-semver strings. + await expect( + ingestPackageRelease(testEnv.DB, job, fakeVerified(release), NOW), + ).rejects.toMatchObject({ reason: "INVALID_VERSION" }); + }); + + it("silently no-ops on a same-content replay", async () => { + const release = makeRelease("1.0.0"); + const job = jobFor(DID_A, NSID.packageRelease, "demo:1.0.0"); + await ingestPackageRelease(testEnv.DB, job, fakeVerified(release), NOW); + await ingestPackageRelease(testEnv.DB, job, fakeVerified(release), NOW); + + const dups = await testEnv.DB.prepare( + `SELECT COUNT(*) as n FROM release_duplicate_attempts`, + ).first<{ n: number }>(); + expect(dups?.n).toBe(0); + }); + + it("audits a duplicate-version attempt with different content", async () => { + const job = jobFor(DID_A, NSID.packageRelease, "demo:1.0.0"); + await ingestPackageRelease(testEnv.DB, job, fakeVerified(makeRelease("1.0.0")), NOW); + + // Second call — same did/package/version, different carBytes (simulating + // a malicious republish or a publisher trying to mutate a version). + const tampered: VerifiedPdsRecord = { + cid: "bafyreigDIFFERENT00000000000000000000000000000000000000", + record: makeRelease("1.0.0"), + carBytes: new Uint8Array([0x01, 0x02, 0x03]), + }; + await ingestPackageRelease(testEnv.DB, job, tampered, NOW); + + const dup = await testEnv.DB.prepare( + `SELECT did, package, version, reason FROM release_duplicate_attempts`, + ).first<{ did: string; package: string; version: string; reason: string }>(); + expect(dup).toMatchObject({ + did: DID_A, + package: "demo", + version: "1.0.0", + reason: "IMMUTABLE_VERSION", + }); + }); +}); + +// ─── Writer: publisher.profile ────────────────────────────────────────────── + +describe("ingestPublisherProfile", () => { + const validRecord = { + $type: NSID.publisherProfile, + displayName: "Acme Plugin Co.", + description: "We make plugins", + contact: [{ kind: "general", email: "hi@acme.test" }], + }; + + it("inserts on first call, upserts on subsequent", async () => { + const job = jobFor(DID_A, NSID.publisherProfile, "self"); + await ingestPublisherProfile(testEnv.DB, job, fakeVerified(validRecord), NOW); + await ingestPublisherProfile( + testEnv.DB, + job, + fakeVerified({ ...validRecord, displayName: "Acme Inc." }), + NOW, + ); + + const row = await testEnv.DB.prepare(`SELECT display_name FROM publishers WHERE did = ?`) + .bind(DID_A) + .first<{ display_name: string }>(); + expect(row?.display_name).toBe("Acme Inc."); + }); + + it("rejects rkey ≠ 'self'", async () => { + const job = jobFor(DID_A, NSID.publisherProfile, "not-self"); + await expect( + ingestPublisherProfile(testEnv.DB, job, fakeVerified(validRecord), NOW), + ).rejects.toMatchObject({ reason: "RKEY_MISMATCH" }); + }); + + it("rejects contact entries with neither url nor email", async () => { + const job = jobFor(DID_A, NSID.publisherProfile, "self"); + await expect( + ingestPublisherProfile( + testEnv.DB, + job, + fakeVerified({ ...validRecord, contact: [{ kind: "general" }] }), + NOW, + ), + ).rejects.toMatchObject({ reason: "CONTACT_VALIDATION_FAILED" }); + }); +}); + +// ─── Writer: publisher.verification ───────────────────────────────────────── + +describe("ingestPublisherVerification", () => { + const validRecord = { + $type: NSID.publisherVerification, + subject: DID_B, + handle: "subject.test", + displayName: "Subject Co.", + createdAt: "2026-05-09T12:00:00.000Z", + }; + + it("inserts a verification, preserving the bound handle + displayName", async () => { + const job = jobFor(DID_A, NSID.publisherVerification, "3kifgtest00000"); + await ingestPublisherVerification(testEnv.DB, job, fakeVerified(validRecord), NOW); + + const row = await testEnv.DB.prepare( + `SELECT subject_did, subject_handle, subject_display_name, tombstoned_at + FROM publisher_verifications WHERE issuer_did = ? AND rkey = ?`, + ) + .bind(DID_A, "3kifgtest00000") + .first<{ + subject_did: string; + subject_handle: string; + subject_display_name: string; + tombstoned_at: string | null; + }>(); + expect(row).toMatchObject({ + subject_did: DID_B, + subject_handle: "subject.test", + subject_display_name: "Subject Co.", + tombstoned_at: null, + }); + }); + + it("upsert-on-conflict clears any tombstone (re-publish recovers)", async () => { + const job = jobFor(DID_A, NSID.publisherVerification, "3kifgtest00000"); + await ingestPublisherVerification(testEnv.DB, job, fakeVerified(validRecord), NOW); + await applyDelete(testEnv.DB, { ...job, operation: "delete" }, NOW); + await ingestPublisherVerification(testEnv.DB, job, fakeVerified(validRecord), NOW); + + const row = await testEnv.DB.prepare( + `SELECT tombstoned_at FROM publisher_verifications WHERE issuer_did = ? AND rkey = ?`, + ) + .bind(DID_A, "3kifgtest00000") + .first<{ tombstoned_at: string | null }>(); + expect(row?.tombstoned_at).toBeNull(); + }); +}); + +// ─── Delete handling ──────────────────────────────────────────────────────── + +describe("applyDelete", () => { + beforeEach(async () => { + await ingestPackageProfile( + testEnv.DB, + jobFor(DID_A, NSID.packageProfile, "demo"), + fakeVerified({ + $type: NSID.packageProfile, + id: `at://${DID_A}/${NSID.packageProfile}/demo`, + slug: "demo", + type: "emdash-plugin", + license: "MIT", + authors: [{ name: "Tester" }], + security: [{ email: "x@y.test" }], + }), + NOW, + ); + await ingestPackageRelease( + testEnv.DB, + jobFor(DID_A, NSID.packageRelease, "demo:1.0.0"), + fakeVerified({ + $type: NSID.packageRelease, + package: "demo", + version: "1.0.0", + artifacts: { package: { url: "https://x.test/d.tgz", checksum: "bsha-abc" } }, + extensions: { + "com.emdashcms.experimental.package.releaseExtension": { + $type: "com.emdashcms.experimental.package.releaseExtension", + declaredAccess: {}, + }, + }, + }), + NOW, + ); + }); + + it("hard-deletes a package.profile", async () => { + await applyDelete( + testEnv.DB, + jobFor(DID_A, NSID.packageProfile, "demo", { operation: "delete" }), + NOW, + ); + const row = await testEnv.DB.prepare(`SELECT did FROM packages WHERE did = ?`) + .bind(DID_A) + .first(); + expect(row).toBeNull(); + }); + + it("soft-deletes a release (sets tombstoned_at)", async () => { + await applyDelete( + testEnv.DB, + jobFor(DID_A, NSID.packageRelease, "demo:1.0.0", { operation: "delete" }), + NOW, + ); + const row = await testEnv.DB.prepare( + `SELECT tombstoned_at FROM releases WHERE did = ? AND rkey = ?`, + ) + .bind(DID_A, "demo:1.0.0") + .first<{ tombstoned_at: string | null }>(); + expect(row?.tombstoned_at).toBe(NOW.toISOString()); + }); + + it("hard-deletes a publisher.profile", async () => { + await ingestPublisherProfile( + testEnv.DB, + jobFor(DID_A, NSID.publisherProfile, "self"), + fakeVerified({ + $type: NSID.publisherProfile, + displayName: "Acme", + contact: [{ email: "a@b.test" }], + }), + NOW, + ); + await applyDelete( + testEnv.DB, + jobFor(DID_A, NSID.publisherProfile, "self", { operation: "delete" }), + NOW, + ); + const row = await testEnv.DB.prepare(`SELECT did FROM publishers WHERE did = ?`) + .bind(DID_A) + .first(); + expect(row).toBeNull(); + }); + + it("soft-deletes a publisher.verification", async () => { + await ingestPublisherVerification( + testEnv.DB, + jobFor(DID_A, NSID.publisherVerification, "tid001"), + fakeVerified({ + $type: NSID.publisherVerification, + subject: DID_B, + handle: "s.test", + displayName: "S", + createdAt: NOW.toISOString(), + }), + NOW, + ); + await applyDelete( + testEnv.DB, + jobFor(DID_A, NSID.publisherVerification, "tid001", { operation: "delete" }), + NOW, + ); + const row = await testEnv.DB.prepare( + `SELECT tombstoned_at FROM publisher_verifications WHERE issuer_did = ? AND rkey = ?`, + ) + .bind(DID_A, "tid001") + .first<{ tombstoned_at: string | null }>(); + expect(row?.tombstoned_at).toBe(NOW.toISOString()); + }); +}); + +// ─── Dispatcher (processMessage) ──────────────────────────────────────────── + +class StubResolver implements DidDocumentResolverLike { + resolve(_did: Did): Promise { + // processMessage tests inject a DidResolver that's wired to a stub DID + // doc — we never actually traverse this resolver because the cache + // always hits. + return Promise.reject(new Error("StubResolver should not be called")); + } +} + +class MapDidDocCache implements DidDocCache { + private readonly entries = new Map< + string, + { pds: string; signingKey: string; signingKeyId: string; resolvedAt: Date } + >(); + read(did: string) { + return Promise.resolve(this.entries.get(did) ?? null); + } + upsert(did: string, doc: { pds: string; signingKey: string; signingKeyId: string }, now: Date) { + this.entries.set(did, { ...doc, resolvedAt: now }); + return Promise.resolve(); + } + expire(did: string) { + const entry = this.entries.get(did); + if (entry) this.entries.set(did, { ...entry, resolvedAt: new Date(0) }); + return Promise.resolve(); + } + seed(did: string) { + this.entries.set(did, { + pds: "https://pds.test.example", + signingKey: signingKeyMultibase, + signingKeyId: `${did}#atproto`, + resolvedAt: NOW, + }); + } +} + +class FakeMessage implements MessageController { + acked = 0; + retried = 0; + ack() { + this.acked += 1; + } + retry() { + this.retried += 1; + } +} + +function buildDeps(opts: { fetch: typeof fetch }): { + deps: ConsumerDeps; + cache: MapDidDocCache; +} { + const cache = new MapDidDocCache(); + const resolver = new DidResolver({ + cache, + resolver: new StubResolver(), + // Long TTL so we never actually call StubResolver. + ttlMs: 1_000_000, + now: () => NOW, + }); + return { + deps: { db: testEnv.DB, resolver, fetch: opts.fetch, now: () => NOW }, + cache, + }; +} + +async function deadLetterCount(): Promise { + const r = await testEnv.DB.prepare(`SELECT COUNT(*) as n FROM dead_letters`).first<{ + n: number; + }>(); + return r?.n ?? 0; +} + +describe("processMessage dispatcher", () => { + it("acks and dead-letters on a permanent PDS error (404)", async () => { + const { deps, cache } = buildDeps({ + fetch: () => Promise.resolve(new Response("", { status: 404 })), + }); + cache.seed(DID_A); + const msg = new FakeMessage(); + const job = jobFor(DID_A, NSID.packageProfile, "missing"); + + await processMessage(job, msg, deps); + + expect(msg.acked).toBe(1); + expect(msg.retried).toBe(0); + expect(await deadLetterCount()).toBe(1); + const row = await testEnv.DB.prepare(`SELECT reason FROM dead_letters`).first<{ + reason: string; + }>(); + expect(row?.reason).toBe("RECORD_NOT_FOUND"); + }); + + it("retries on a transient PDS error (5xx)", async () => { + const { deps, cache } = buildDeps({ + fetch: () => Promise.resolve(new Response("", { status: 503 })), + }); + cache.seed(DID_A); + const msg = new FakeMessage(); + + await processMessage(jobFor(DID_A, NSID.packageProfile, "demo"), msg, deps); + + expect(msg.retried).toBe(1); + expect(msg.acked).toBe(0); + expect(await deadLetterCount()).toBe(0); + }); + + it("retries on a network error", async () => { + const { deps, cache } = buildDeps({ + fetch: () => Promise.reject(new TypeError("connection refused")), + }); + cache.seed(DID_A); + const msg = new FakeMessage(); + + await processMessage(jobFor(DID_A, NSID.packageProfile, "demo"), msg, deps); + + expect(msg.retried).toBe(1); + expect(await deadLetterCount()).toBe(0); + }); + + it("forensics + acks on garbage CAR bytes (verifyRecord rejects → INVALID_PROOF)", async () => { + const { deps, cache } = buildDeps({ + fetch: () => Promise.resolve(new Response(new Uint8Array([1, 2, 3, 4]), { status: 200 })), + }); + cache.seed(DID_A); + const msg = new FakeMessage(); + + await processMessage(jobFor(DID_A, NSID.packageProfile, "demo"), msg, deps); + + expect(msg.acked).toBe(1); + const row = await testEnv.DB.prepare(`SELECT reason FROM dead_letters`).first<{ + reason: string; + }>(); + expect(row?.reason).toBe("INVALID_PROOF"); + }); + + it("delete: acks immediately, no PDS fetch", async () => { + let fetchCalls = 0; + const { deps } = buildDeps({ + fetch: () => { + fetchCalls += 1; + return Promise.resolve(new Response("", { status: 500 })); + }, + }); + const msg = new FakeMessage(); + const job = jobFor(DID_A, NSID.packageProfile, "demo", { operation: "delete" }); + + await processMessage(job, msg, deps); + + expect(msg.acked).toBe(1); + expect(fetchCalls).toBe(0); + }); +}); + +// ─── Adversarial-review fixes: regression tests ───────────────────────────── + +describe("ingestPackageProfile: security[] contact validation", () => { + it("rejects security entries with neither url nor email", async () => { + const job = jobFor(DID_A, NSID.packageProfile, "demo"); + await expect( + ingestPackageProfile( + testEnv.DB, + job, + fakeVerified({ + $type: NSID.packageProfile, + id: `at://${DID_A}/${NSID.packageProfile}/demo`, + slug: "demo", + type: "emdash-plugin", + license: "MIT", + authors: [{ name: "Tester" }], + security: [{ kind: "security" }], + }), + NOW, + ), + ).rejects.toMatchObject({ reason: "CONTACT_VALIDATION_FAILED" }); + }); +}); + +describe("ingestPackageRelease: releaseExtension validation", () => { + const validProfile = { + $type: NSID.packageProfile, + id: `at://${DID_A}/${NSID.packageProfile}/demo`, + slug: "demo", + type: "emdash-plugin", + license: "MIT", + authors: [{ name: "Tester" }], + security: [{ email: "x@y.test" }], + }; + beforeEach(async () => { + await ingestPackageProfile( + testEnv.DB, + jobFor(DID_A, NSID.packageProfile, "demo"), + fakeVerified(validProfile), + NOW, + ); + }); + + it("rejects when extensions field is missing the releaseExtension key", async () => { + await expect( + ingestPackageRelease( + testEnv.DB, + jobFor(DID_A, NSID.packageRelease, "demo:1.0.0"), + fakeVerified({ + $type: NSID.packageRelease, + package: "demo", + version: "1.0.0", + artifacts: { package: { url: "https://x.test/d.tgz", checksum: "bsha-abc" } }, + extensions: {}, + }), + NOW, + ), + ).rejects.toMatchObject({ reason: "LEXICON_VALIDATION_FAILED" }); + }); + + it("rejects when extensions field is not an object", async () => { + await expect( + ingestPackageRelease( + testEnv.DB, + jobFor(DID_A, NSID.packageRelease, "demo:1.0.0"), + fakeVerified({ + $type: NSID.packageRelease, + package: "demo", + version: "1.0.0", + artifacts: { package: { url: "https://x.test/d.tgz", checksum: "bsha-abc" } }, + extensions: "lol", + }), + NOW, + ), + ).rejects.toMatchObject({ reason: "LEXICON_VALIDATION_FAILED" }); + }); + + it("rejects when releaseExtension fails its own lexicon validation", async () => { + await expect( + ingestPackageRelease( + testEnv.DB, + jobFor(DID_A, NSID.packageRelease, "demo:1.0.0"), + fakeVerified({ + $type: NSID.packageRelease, + package: "demo", + version: "1.0.0", + artifacts: { package: { url: "https://x.test/d.tgz", checksum: "bsha-abc" } }, + extensions: { + "com.emdashcms.experimental.package.releaseExtension": { + $type: "com.emdashcms.experimental.package.releaseExtension", + // missing required `declaredAccess` + }, + }, + }), + NOW, + ), + ).rejects.toMatchObject({ reason: "LEXICON_VALIDATION_FAILED" }); + }); + + it("stores only the validated releaseExtension contents in emdash_extension", async () => { + await ingestPackageRelease( + testEnv.DB, + jobFor(DID_A, NSID.packageRelease, "demo:1.0.0"), + fakeVerified({ + $type: NSID.packageRelease, + package: "demo", + version: "1.0.0", + artifacts: { package: { url: "https://x.test/d.tgz", checksum: "bsha-abc" } }, + extensions: { + "com.emdashcms.experimental.package.releaseExtension": { + $type: "com.emdashcms.experimental.package.releaseExtension", + declaredAccess: { network: { fetch: {} } }, + }, + // arbitrary extra key — must NOT land in the column + "com.example.someoneElse": { irrelevant: "data" }, + }, + }), + NOW, + ); + const row = await testEnv.DB.prepare( + `SELECT emdash_extension FROM releases WHERE did = ? AND package = ? AND version = ?`, + ) + .bind(DID_A, "demo", "1.0.0") + .first<{ emdash_extension: string }>(); + const stored = JSON.parse(row?.emdash_extension ?? "{}"); + expect(stored.declaredAccess).toBeDefined(); + expect(stored).not.toHaveProperty("com.example.someoneElse"); + }); +}); + +describe("ingestPackageRelease: package field charset", () => { + beforeEach(async () => { + await ingestPackageProfile( + testEnv.DB, + jobFor(DID_A, NSID.packageProfile, "demo"), + fakeVerified({ + $type: NSID.packageProfile, + id: `at://${DID_A}/${NSID.packageProfile}/demo`, + slug: "demo", + type: "emdash-plugin", + license: "MIT", + authors: [{ name: "Tester" }], + security: [{ email: "x@y.test" }], + }), + NOW, + ); + }); + + it("rejects record.package containing a colon", async () => { + // Ambiguous-rkey attack: `package: "foo:bar"` + `version: "1.0.0"` + // would build the same rkey as `package: "foo"` + `version: "bar:1.0.0"`. + await expect( + ingestPackageRelease( + testEnv.DB, + jobFor(DID_A, NSID.packageRelease, "demo:bad:1.0.0"), + fakeVerified({ + $type: NSID.packageRelease, + package: "demo:bad", + version: "1.0.0", + artifacts: { package: { url: "https://x.test/d.tgz", checksum: "bsha-abc" } }, + extensions: { + "com.emdashcms.experimental.package.releaseExtension": { + $type: "com.emdashcms.experimental.package.releaseExtension", + declaredAccess: {}, + }, + }, + }), + NOW, + ), + ).rejects.toMatchObject({ reason: "RKEY_MISMATCH" }); + }); +}); + +describe("ingestPackageRelease: parent profile pre-check", () => { + it("throws MissingDependencyError when no parent profile exists", async () => { + // No profile seeded — release event arriving before its profile. + await expect( + ingestPackageRelease( + testEnv.DB, + jobFor(DID_A, NSID.packageRelease, "demo:1.0.0"), + fakeVerified({ + $type: NSID.packageRelease, + package: "demo", + version: "1.0.0", + artifacts: { package: { url: "https://x.test/d.tgz", checksum: "bsha-abc" } }, + extensions: { + "com.emdashcms.experimental.package.releaseExtension": { + $type: "com.emdashcms.experimental.package.releaseExtension", + declaredAccess: {}, + }, + }, + }), + NOW, + ), + ).rejects.toMatchObject({ name: "MissingDependencyError" }); + }); + + // Dispatcher-level retry-on-MissingDependency coverage lives in the + // "processMessage dispatcher" suite further down — uses + // `ConsumerDeps.verify` injection so the writer's parent-profile + // pre-check actually fires and the dispatcher's retry branch runs + // end-to-end. +}); + +describe("ingestPackageRelease: latest_version + capabilities denormalisation", () => { + const validProfile = { + $type: NSID.packageProfile, + id: `at://${DID_A}/${NSID.packageProfile}/demo`, + slug: "demo", + type: "emdash-plugin", + license: "MIT", + authors: [{ name: "Tester" }], + security: [{ email: "x@y.test" }], + }; + beforeEach(async () => { + await ingestPackageProfile( + testEnv.DB, + jobFor(DID_A, NSID.packageProfile, "demo"), + fakeVerified(validProfile), + NOW, + ); + }); + + function release(version: string, declaredAccess: Record) { + return { + $type: NSID.packageRelease, + package: "demo", + version, + artifacts: { package: { url: "https://x.test/d.tgz", checksum: "bsha-abc" } }, + extensions: { + "com.emdashcms.experimental.package.releaseExtension": { + $type: "com.emdashcms.experimental.package.releaseExtension", + declaredAccess, + }, + }, + }; + } + + it("populates packages.latest_version after first release insert", async () => { + await ingestPackageRelease( + testEnv.DB, + jobFor(DID_A, NSID.packageRelease, "demo:1.0.0"), + fakeVerified(release("1.0.0", { content: { read: {} } })), + NOW, + ); + const row = await testEnv.DB.prepare( + `SELECT latest_version, capabilities FROM packages WHERE did = ?`, + ) + .bind(DID_A) + .first<{ latest_version: string; capabilities: string }>(); + expect(row?.latest_version).toBe("1.0.0"); + expect(JSON.parse(row?.capabilities ?? "[]")).toEqual(["content"]); + }); + + it("updates latest_version when a higher-version release lands", async () => { + await ingestPackageRelease( + testEnv.DB, + jobFor(DID_A, NSID.packageRelease, "demo:1.0.0"), + fakeVerified(release("1.0.0", { content: { read: {} } })), + NOW, + ); + await ingestPackageRelease( + testEnv.DB, + jobFor(DID_A, NSID.packageRelease, "demo:2.0.0"), + fakeVerified(release("2.0.0", { network: { fetch: {} } })), + NOW, + ); + const row = await testEnv.DB.prepare( + `SELECT latest_version, capabilities FROM packages`, + ).first<{ latest_version: string; capabilities: string }>(); + expect(row?.latest_version).toBe("2.0.0"); + expect(JSON.parse(row?.capabilities ?? "[]")).toEqual(["network"]); + }); + + it("does NOT downgrade latest_version when an older release lands", async () => { + await ingestPackageRelease( + testEnv.DB, + jobFor(DID_A, NSID.packageRelease, "demo:2.0.0"), + fakeVerified(release("2.0.0", { network: { fetch: {} } })), + NOW, + ); + await ingestPackageRelease( + testEnv.DB, + jobFor(DID_A, NSID.packageRelease, "demo:1.0.0"), + fakeVerified(release("1.0.0", { content: { read: {} } })), + NOW, + ); + const row = await testEnv.DB.prepare(`SELECT latest_version FROM packages`).first<{ + latest_version: string; + }>(); + expect(row?.latest_version).toBe("2.0.0"); + }); + + it("recomputes latest_version after a release tombstone", async () => { + await ingestPackageRelease( + testEnv.DB, + jobFor(DID_A, NSID.packageRelease, "demo:1.0.0"), + fakeVerified(release("1.0.0", {})), + NOW, + ); + await ingestPackageRelease( + testEnv.DB, + jobFor(DID_A, NSID.packageRelease, "demo:2.0.0"), + fakeVerified(release("2.0.0", {})), + NOW, + ); + // Tombstone the latest. + await applyDelete( + testEnv.DB, + jobFor(DID_A, NSID.packageRelease, "demo:2.0.0", { operation: "delete" }), + NOW, + ); + const row = await testEnv.DB.prepare(`SELECT latest_version FROM packages`).first<{ + latest_version: string; + }>(); + expect(row?.latest_version).toBe("1.0.0"); + }); +}); + +describe("ingestPackageRelease: same-content re-publish on tombstoned row", () => { + const validProfile = { + $type: NSID.packageProfile, + id: `at://${DID_A}/${NSID.packageProfile}/demo`, + slug: "demo", + type: "emdash-plugin", + license: "MIT", + authors: [{ name: "Tester" }], + security: [{ email: "x@y.test" }], + }; + const release = { + $type: NSID.packageRelease, + package: "demo", + version: "1.0.0", + artifacts: { package: { url: "https://x.test/d.tgz", checksum: "bsha-abc" } }, + extensions: { + "com.emdashcms.experimental.package.releaseExtension": { + $type: "com.emdashcms.experimental.package.releaseExtension", + declaredAccess: {}, + }, + }, + }; + + beforeEach(async () => { + await ingestPackageProfile( + testEnv.DB, + jobFor(DID_A, NSID.packageProfile, "demo"), + fakeVerified(validProfile), + NOW, + ); + }); + + it("clears tombstoned_at on a same-content republish", async () => { + const job = jobFor(DID_A, NSID.packageRelease, "demo:1.0.0"); + const verified = fakeVerified(release); + await ingestPackageRelease(testEnv.DB, job, verified, NOW); + await applyDelete(testEnv.DB, { ...job, operation: "delete" }, NOW); + // Confirm tombstoned. + const before = await testEnv.DB.prepare( + `SELECT tombstoned_at FROM releases WHERE did = ? AND package = ? AND version = ?`, + ) + .bind(DID_A, "demo", "1.0.0") + .first<{ tombstoned_at: string | null }>(); + expect(before?.tombstoned_at).not.toBeNull(); + // Re-publish identical bytes. + await ingestPackageRelease(testEnv.DB, job, verified, NOW); + const after = await testEnv.DB.prepare( + `SELECT tombstoned_at FROM releases WHERE did = ? AND package = ? AND version = ?`, + ) + .bind(DID_A, "demo", "1.0.0") + .first<{ tombstoned_at: string | null }>(); + expect(after?.tombstoned_at).toBeNull(); + }); + + it("does NOT audit a duplicate-attempt when same content lands on a tombstone", async () => { + const job = jobFor(DID_A, NSID.packageRelease, "demo:1.0.0"); + const verified = fakeVerified(release); + await ingestPackageRelease(testEnv.DB, job, verified, NOW); + await applyDelete(testEnv.DB, { ...job, operation: "delete" }, NOW); + await ingestPackageRelease(testEnv.DB, job, verified, NOW); + const dups = await testEnv.DB.prepare( + `SELECT COUNT(*) as n FROM release_duplicate_attempts`, + ).first<{ n: number }>(); + expect(dups?.n).toBe(0); + }); +}); + +describe("computeVersionSort + version overflow rejection", () => { + const validProfile = { + $type: NSID.packageProfile, + id: `at://${DID_A}/${NSID.packageProfile}/demo`, + slug: "demo", + type: "emdash-plugin", + license: "MIT", + authors: [{ name: "Tester" }], + security: [{ email: "x@y.test" }], + }; + beforeEach(async () => { + await ingestPackageProfile( + testEnv.DB, + jobFor(DID_A, NSID.packageProfile, "demo"), + fakeVerified(validProfile), + NOW, + ); + }); + + function release(version: string) { + return { + $type: NSID.packageRelease, + package: "demo", + version, + artifacts: { package: { url: "https://x.test/d.tgz", checksum: "bsha-abc" } }, + extensions: { + "com.emdashcms.experimental.package.releaseExtension": { + $type: "com.emdashcms.experimental.package.releaseExtension", + declaredAccess: {}, + }, + }, + }; + } + + it("rejects prerelease numerics longer than 10 digits", async () => { + await expect( + ingestPackageRelease( + testEnv.DB, + jobFor(DID_A, NSID.packageRelease, "demo:1.0.0-12345678901"), + fakeVerified(release("1.0.0-12345678901")), + NOW, + ), + ).rejects.toMatchObject({ reason: "INVALID_VERSION" }); + }); + + it("rejects major/minor/patch components longer than 10 digits", async () => { + await expect( + ingestPackageRelease( + testEnv.DB, + jobFor(DID_A, NSID.packageRelease, "demo:99999999999.0.0"), + fakeVerified(release("99999999999.0.0")), + NOW, + ), + ).rejects.toMatchObject({ reason: "INVALID_VERSION" }); + }); +}); + +describe("applyDelete unknown collection", () => { + it("throws IngestError UNKNOWN_COLLECTION instead of silently dropping", async () => { + await expect( + applyDelete( + testEnv.DB, + jobFor(DID_A, "com.example.unknown", "x", { operation: "delete" }), + NOW, + ), + ).rejects.toMatchObject({ name: "IngestError", reason: "UNKNOWN_COLLECTION" }); + }); +}); + +describe("processMessage dispatcher: MissingDependencyError → retry", () => { + it("retries the message when the release writer throws MissingDependencyError", async () => { + // No parent profile seeded — release writer's parent-profile check + // throws MissingDependencyError → dispatcher should map to retry(). + const cache = new MapDidDocCache(); + const resolver = new DidResolver({ + cache, + resolver: new StubResolver(), + ttlMs: 1_000_000, + now: () => NOW, + }); + cache.seed(DID_A); + const release = { + $type: NSID.packageRelease, + package: "demo", + version: "1.0.0", + artifacts: { package: { url: "https://x.test/d.tgz", checksum: "bsha-abc" } }, + extensions: { + "com.emdashcms.experimental.package.releaseExtension": { + $type: "com.emdashcms.experimental.package.releaseExtension", + declaredAccess: {}, + }, + }, + }; + const deps: ConsumerDeps = { + db: testEnv.DB, + resolver, + now: () => NOW, + // Inject a verifier that returns a real-shaped record without + // running the actual @atcute/repo verification chain. + verify: () => + Promise.resolve({ + cid: "bafyreigtest00000000000000000000000000000000000000000000", + record: release, + carBytes: new Uint8Array([0xde, 0xad, 0xbe, 0xef]), + }), + }; + const msg = new FakeMessage(); + await processMessage(jobFor(DID_A, NSID.packageRelease, "demo:1.0.0"), msg, deps); + + expect(msg.retried).toBe(1); + expect(msg.acked).toBe(0); + expect(await deadLetterCount()).toBe(0); + }); +}); + +describe("ingestPackageRelease: latest_version refresh atomicity", () => { + const validProfile = { + $type: NSID.packageProfile, + id: `at://${DID_A}/${NSID.packageProfile}/demo`, + slug: "demo", + type: "emdash-plugin", + license: "MIT", + authors: [{ name: "Tester" }], + security: [{ email: "x@y.test" }], + }; + function release(version: string) { + return { + $type: NSID.packageRelease, + package: "demo", + version, + artifacts: { package: { url: "https://x.test/d.tgz", checksum: "bsha-abc" } }, + extensions: { + "com.emdashcms.experimental.package.releaseExtension": { + $type: "com.emdashcms.experimental.package.releaseExtension", + declaredAccess: { content: { read: {} } }, + }, + }, + }; + } + beforeEach(async () => { + await ingestPackageProfile( + testEnv.DB, + jobFor(DID_A, NSID.packageProfile, "demo"), + fakeVerified(validProfile), + NOW, + ); + }); + + it("idempotent same-content insert still leaves latest_version correct", async () => { + const job = jobFor(DID_A, NSID.packageRelease, "demo:1.0.0"); + const verified = fakeVerified(release("1.0.0")); + await ingestPackageRelease(testEnv.DB, job, verified, NOW); + // Manually corrupt latest_version to simulate a refresh that + // somehow drifted out of sync with the underlying releases. + await testEnv.DB.prepare(`UPDATE packages SET latest_version = NULL`).run(); + // Replay same content. Refresh always runs in the batch — should fix. + await ingestPackageRelease(testEnv.DB, job, verified, NOW); + const row = await testEnv.DB.prepare(`SELECT latest_version FROM packages`).first<{ + latest_version: string; + }>(); + expect(row?.latest_version).toBe("1.0.0"); + }); + + it("uses single-statement UPDATE for refresh (race-safety check via SQL shape)", async () => { + // Insert v1, manually insert v2 directly (bypassing writer), then + // re-trigger refresh by re-inserting v1 (idempotent). The race-safe + // UPDATE should pick up v2 as the new latest because its subquery + // reads current max state, not a snapshot from before the manual + // insert. + await ingestPackageRelease( + testEnv.DB, + jobFor(DID_A, NSID.packageRelease, "demo:1.0.0"), + fakeVerified(release("1.0.0")), + NOW, + ); + // Manually insert v2 with a known version_sort that sorts after 1.0.0. + await testEnv.DB.prepare( + `INSERT INTO releases + (did, package, version, rkey, version_sort, artifacts, + emdash_extension, cts, record_blob, signature_metadata, verified_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .bind( + DID_A, + "demo", + "2.0.0", + "demo:2.0.0", + "0000000002.0000000000.0000000000.~", + "{}", + JSON.stringify({ declaredAccess: { network: { fetch: {} } } }), + NOW.toISOString(), + new Uint8Array([0xff]), + JSON.stringify({ cid: "x" }), + NOW.toISOString(), + ) + .run(); + // Re-trigger refresh by re-publishing v1 (same content → DO NOTHING, + // but the batched refresh-UPDATE still runs). + await ingestPackageRelease( + testEnv.DB, + jobFor(DID_A, NSID.packageRelease, "demo:1.0.0"), + fakeVerified(release("1.0.0")), + NOW, + ); + const row = await testEnv.DB.prepare( + `SELECT latest_version, capabilities FROM packages`, + ).first<{ latest_version: string; capabilities: string }>(); + expect(row?.latest_version).toBe("2.0.0"); + expect(JSON.parse(row?.capabilities ?? "[]")).toEqual(["network"]); + }); +}); + +describe("release_duplicate_attempts UNIQUE constraint", () => { + const validProfile = { + $type: NSID.packageProfile, + id: `at://${DID_A}/${NSID.packageProfile}/demo`, + slug: "demo", + type: "emdash-plugin", + license: "MIT", + authors: [{ name: "Tester" }], + security: [{ email: "x@y.test" }], + }; + function release(version: string) { + return { + $type: NSID.packageRelease, + package: "demo", + version, + artifacts: { package: { url: "https://x.test/d.tgz", checksum: "bsha-abc" } }, + extensions: { + "com.emdashcms.experimental.package.releaseExtension": { + $type: "com.emdashcms.experimental.package.releaseExtension", + declaredAccess: {}, + }, + }, + }; + } + beforeEach(async () => { + await ingestPackageProfile( + testEnv.DB, + jobFor(DID_A, NSID.packageProfile, "demo"), + fakeVerified(validProfile), + NOW, + ); + }); + + it("dedupes repeated identical duplicate-attempt payloads", async () => { + const job = jobFor(DID_A, NSID.packageRelease, "demo:1.0.0"); + await ingestPackageRelease(testEnv.DB, job, fakeVerified(release("1.0.0")), NOW); + + // Hostile-publisher pattern: pump the same different-content tampered + // payload many times. + const tampered: VerifiedPdsRecord = { + cid: "bafyreigDIFFERENT00000000000000000000000000000000000000", + record: release("1.0.0"), + carBytes: new Uint8Array([0x01, 0x02, 0x03]), + }; + for (let i = 0; i < 5; i++) { + await ingestPackageRelease(testEnv.DB, job, tampered, NOW); + } + const dups = await testEnv.DB.prepare( + `SELECT COUNT(*) as n FROM release_duplicate_attempts`, + ).first<{ n: number }>(); + expect(dups?.n).toBe(1); + }); + + it("audits distinct tampered payloads as separate attempts", async () => { + const job = jobFor(DID_A, NSID.packageRelease, "demo:1.0.0"); + await ingestPackageRelease(testEnv.DB, job, fakeVerified(release("1.0.0")), NOW); + + await ingestPackageRelease( + testEnv.DB, + job, + { + cid: "x", + record: release("1.0.0"), + carBytes: new Uint8Array([0x01]), + }, + NOW, + ); + await ingestPackageRelease( + testEnv.DB, + job, + { + cid: "y", + record: release("1.0.0"), + carBytes: new Uint8Array([0x02]), + }, + NOW, + ); + const dups = await testEnv.DB.prepare( + `SELECT COUNT(*) as n FROM release_duplicate_attempts`, + ).first<{ n: number }>(); + expect(dups?.n).toBe(2); + }); +}); + +describe("parseReleaseRkey: malformed %-encoding in delete rkey", () => { + it("throws IngestError so the dispatcher writes a dead_letters row before acking", async () => { + // Silently no-op'ing the parse failure would lose the audit trail — + // an operator investigating "why didn't this delete take effect?" + // would have nothing to look at. + await expect( + applyDelete( + testEnv.DB, + jobFor(DID_A, NSID.packageRelease, "demo:1.0.0%XX", { operation: "delete" }), + NOW, + ), + ).rejects.toMatchObject({ name: "IngestError", reason: "RKEY_MISMATCH" }); + }); +}); + +describe("drainDeadLetterBatch", () => { + it("acks each message and writes a forensics row", async () => { + const { drainDeadLetterBatch: drain } = await import("../src/records-consumer.js"); + const messages: Array = [ + { + body: jobFor(DID_A, NSID.packageRelease, "demo:1.0.0"), + ack: () => {}, + retry: () => {}, + }, + { + body: jobFor(DID_B, NSID.packageProfile, "other"), + ack: () => {}, + retry: () => {}, + }, + ]; + let acked = 0; + messages.forEach((m) => { + const orig = m.ack; + m.ack = () => { + acked += 1; + orig(); + }; + }); + await drain({ messages }, { DB: testEnv.DB } as unknown as Env); + + expect(acked).toBe(2); + const dl = await testEnv.DB.prepare(`SELECT COUNT(*) as n FROM dead_letters`).first<{ + n: number; + }>(); + expect(dl?.n).toBe(2); + }); +}); + +describe("drainDeadLetterBatch: D1 failure", () => { + it("retries the message when writeDeadLetter throws", async () => { + const { drainDeadLetterBatch: drain } = await import("../src/records-consumer.js"); + let acked = 0; + let retried = 0; + const message: MessageController & { body: RecordsJob } = { + body: jobFor(DID_A, NSID.packageRelease, "demo:1.0.0"), + ack: () => { + acked += 1; + }, + retry: () => { + retried += 1; + }, + }; + // Stub DB whose insert throws to simulate transient D1 failure. + const failingDb = { + prepare: () => ({ + bind: () => ({ + run: () => Promise.reject(new Error("D1 unavailable")), + }), + }), + } as unknown as D1Database; + await drain({ messages: [message] }, { DB: failingDb } as unknown as Env); + + expect(retried).toBe(1); + expect(acked).toBe(0); + }); +}); + +describe("refresh skips writes when values unchanged (avoids FTS-trigger thrashing)", () => { + const validProfile = { + $type: NSID.packageProfile, + id: `at://${DID_A}/${NSID.packageProfile}/demo`, + slug: "demo", + type: "emdash-plugin", + license: "MIT", + // name + description give FTS something to index so the test can + // observe trigger-induced reindexing. + name: "Demo Plugin", + description: "A searchable demo plugin", + authors: [{ name: "Tester" }], + security: [{ email: "x@y.test" }], + }; + const release = { + $type: NSID.packageRelease, + package: "demo", + version: "1.0.0", + artifacts: { package: { url: "https://x.test/d.tgz", checksum: "bsha-abc" } }, + extensions: { + "com.emdashcms.experimental.package.releaseExtension": { + $type: "com.emdashcms.experimental.package.releaseExtension", + declaredAccess: { content: { read: {} } }, + }, + }, + }; + + beforeEach(async () => { + await ingestPackageProfile( + testEnv.DB, + jobFor(DID_A, NSID.packageProfile, "demo"), + fakeVerified(validProfile), + NOW, + ); + }); + + it("does not re-fire packages_au trigger on idempotent refresh", async () => { + const job = jobFor(DID_A, NSID.packageRelease, "demo:1.0.0"); + const verified = fakeVerified(release); + await ingestPackageRelease(testEnv.DB, job, verified, NOW); + + // Capture FTS state after first ingest. + const ftsBefore = await testEnv.DB.prepare( + `SELECT rowid FROM packages_fts WHERE packages_fts MATCH ?`, + ) + .bind("demo") + .all(); + + // Replay same content many times. Each replay would normally fire + // the AFTER UPDATE trigger and re-index FTS even with same content. + // The WHERE-AND-IS-NOT guard short-circuits before the trigger. + for (let i = 0; i < 10; i++) { + await ingestPackageRelease(testEnv.DB, job, verified, NOW); + } + + const ftsAfter = await testEnv.DB.prepare( + `SELECT rowid FROM packages_fts WHERE packages_fts MATCH ?`, + ) + .bind("demo") + .all(); + + // FTS state must be unchanged (same single row). Asserting both row + // count and rowid stability — a re-trigger would delete + re-insert + // with the same rowid in this trigger's design, but if SQLite ever + // optimizes that to a no-op, this test still passes. + expect(ftsAfter.results).toHaveLength(1); + expect(ftsAfter.results).toEqual(ftsBefore.results); + }); +}); + +describe("release_duplicate_attempts.rejected_at tracks latest attempt", () => { + const validProfile = { + $type: NSID.packageProfile, + id: `at://${DID_A}/${NSID.packageProfile}/demo`, + slug: "demo", + type: "emdash-plugin", + license: "MIT", + authors: [{ name: "Tester" }], + security: [{ email: "x@y.test" }], + }; + const release = { + $type: NSID.packageRelease, + package: "demo", + version: "1.0.0", + artifacts: { package: { url: "https://x.test/d.tgz", checksum: "bsha-abc" } }, + extensions: { + "com.emdashcms.experimental.package.releaseExtension": { + $type: "com.emdashcms.experimental.package.releaseExtension", + declaredAccess: {}, + }, + }, + }; + + it("DO UPDATE refreshes rejected_at on repeated identical attempts", async () => { + await ingestPackageProfile( + testEnv.DB, + jobFor(DID_A, NSID.packageProfile, "demo"), + fakeVerified(validProfile), + NOW, + ); + const job = jobFor(DID_A, NSID.packageRelease, "demo:1.0.0"); + await ingestPackageRelease(testEnv.DB, job, fakeVerified(release), NOW); + + const tampered: VerifiedPdsRecord = { + cid: "x", + record: release, + carBytes: new Uint8Array([0x01, 0x02, 0x03]), + }; + const t1 = new Date("2026-05-09T12:00:00.000Z"); + const t2 = new Date("2026-05-10T18:00:00.000Z"); + await ingestPackageRelease(testEnv.DB, job, tampered, t1); + await ingestPackageRelease(testEnv.DB, job, tampered, t2); + + const row = await testEnv.DB.prepare( + `SELECT rejected_at FROM release_duplicate_attempts`, + ).first<{ rejected_at: string }>(); + expect(row?.rejected_at).toBe(t2.toISOString()); + }); +}); + +describe("processBatch isolates per-message failures", () => { + it("retries the failing message and continues processing the rest of the batch", async () => { + const { processBatch } = await import("../src/records-consumer.js"); + // First message's processMessage will throw (forensics-write fails); + // second message should still be processed. + const failingDeps: ConsumerDeps = { + db: { + prepare: () => ({ + bind: () => ({ + run: () => Promise.reject(new Error("D1 unavailable")), + first: () => Promise.reject(new Error("D1 unavailable")), + }), + }), + } as unknown as D1Database, + resolver: new DidResolver({ + cache: new MapDidDocCache(), + resolver: new StubResolver(), + ttlMs: 1_000_000, + now: () => NOW, + }), + now: () => NOW, + // verify that throws → processMessage tries to writeDeadLetter → + // that throws too because db is broken → escapes to processBatch. + verify: () => + Promise.reject(Object.assign(new Error("network down"), { name: "PdsVerificationError" })), + }; + + const messages: Array = []; + const acks: number[] = []; + const retries: number[] = []; + for (let i = 0; i < 3; i++) { + const idx = i; + messages.push({ + body: jobFor(`did:plc:b${i.toString().padStart(20, "0")}`, NSID.packageProfile, "x"), + ack: () => { + acks.push(idx); + }, + retry: () => { + retries.push(idx); + }, + }); + } + await processBatch({ messages }, {} as Env, failingDeps); + + // Each message either acks or retries — no message escapes the loop + // without being controlled. With the failing deps every message ends + // up retried via the catch in processBatch. + expect(retries).toEqual([0, 1, 2]); + expect(acks).toEqual([]); + }); +}); + +describe("duplicate detection compares CIDs, not CAR bytes", () => { + const validProfile = { + $type: NSID.packageProfile, + id: `at://${DID_A}/${NSID.packageProfile}/demo`, + slug: "demo", + type: "emdash-plugin", + license: "MIT", + authors: [{ name: "Tester" }], + security: [{ email: "x@y.test" }], + }; + const release = { + $type: NSID.packageRelease, + package: "demo", + version: "1.0.0", + artifacts: { package: { url: "https://x.test/d.tgz", checksum: "bsha-abc" } }, + extensions: { + "com.emdashcms.experimental.package.releaseExtension": { + $type: "com.emdashcms.experimental.package.releaseExtension", + declaredAccess: {}, + }, + }, + }; + + beforeEach(async () => { + await ingestPackageProfile( + testEnv.DB, + jobFor(DID_A, NSID.packageProfile, "demo"), + fakeVerified(validProfile), + NOW, + ); + }); + + it("treats same CID + different CAR bytes as a benign replay (no audit row)", async () => { + const job = jobFor(DID_A, NSID.packageRelease, "demo:1.0.0"); + // First ingest with one set of bytes. + await ingestPackageRelease( + testEnv.DB, + job, + { + cid: "bafyreigtest00000000000000000000000000000000000000000000", + record: release, + carBytes: new Uint8Array([0x01, 0x02, 0x03]), + }, + NOW, + ); + // Re-fetch produces different CAR bytes (publisher has written other + // records → MST proof differs) but the same record CID. + await ingestPackageRelease( + testEnv.DB, + job, + { + cid: "bafyreigtest00000000000000000000000000000000000000000000", + record: release, + carBytes: new Uint8Array([0xff, 0xfe, 0xfd, 0xfc]), + }, + NOW, + ); + const dups = await testEnv.DB.prepare( + `SELECT COUNT(*) as n FROM release_duplicate_attempts`, + ).first<{ n: number }>(); + expect(dups?.n).toBe(0); + }); + + it("treats different CID at same version as an immutability violation", async () => { + const job = jobFor(DID_A, NSID.packageRelease, "demo:1.0.0"); + await ingestPackageRelease( + testEnv.DB, + job, + { + cid: "bafyreigtest00000000000000000000000000000000000000000000", + record: release, + carBytes: new Uint8Array([0x01]), + }, + NOW, + ); + await ingestPackageRelease( + testEnv.DB, + job, + { + cid: "bafyreigtampered000000000000000000000000000000000000000", + record: release, + carBytes: new Uint8Array([0x02]), + }, + NOW, + ); + const row = await testEnv.DB.prepare( + `SELECT attempted_cid, reason FROM release_duplicate_attempts`, + ).first<{ attempted_cid: string; reason: string }>(); + expect(row?.attempted_cid).toBe("bafyreigtampered000000000000000000000000000000000000000"); + expect(row?.reason).toBe("IMMUTABLE_VERSION"); + }); +}); + +describe("computeVersionSort: final sentinel beats pathological prereleases", () => { + const validProfile = { + $type: NSID.packageProfile, + id: `at://${DID_A}/${NSID.packageProfile}/demo`, + slug: "demo", + type: "emdash-plugin", + license: "MIT", + authors: [{ name: "Tester" }], + security: [{ email: "x@y.test" }], + }; + function release(version: string) { + return { + $type: NSID.packageRelease, + package: "demo", + version, + artifacts: { package: { url: "https://x.test/d.tgz", checksum: "bsha-abc" } }, + extensions: { + "com.emdashcms.experimental.package.releaseExtension": { + $type: "com.emdashcms.experimental.package.releaseExtension", + declaredAccess: {}, + }, + }, + }; + } + beforeEach(async () => { + await ingestPackageProfile( + testEnv.DB, + jobFor(DID_A, NSID.packageProfile, "demo"), + fakeVerified(validProfile), + NOW, + ); + }); + + it("`1.0.0` (final) sorts after `1.0.0-zzzz` (prerelease longer than old `zzz` sentinel)", async () => { + await ingestPackageRelease( + testEnv.DB, + jobFor(DID_A, NSID.packageRelease, "demo:1.0.0-zzzz"), + fakeVerified(release("1.0.0-zzzz")), + NOW, + ); + await ingestPackageRelease( + testEnv.DB, + jobFor(DID_A, NSID.packageRelease, "demo:1.0.0"), + fakeVerified(release("1.0.0")), + NOW, + ); + const row = await testEnv.DB.prepare(`SELECT latest_version FROM packages`).first<{ + latest_version: string; + }>(); + expect(row?.latest_version).toBe("1.0.0"); + }); +}); + +describe("parseReleaseRkey component validation", () => { + it("applyDelete rejects `package:version:extra` rkey via IngestError", async () => { + // Splitting on the first `:` would parse as pkg=`demo`, version=`1.0.0:extra`. + // The semver regex must reject the colon-bearing version. + await expect( + applyDelete( + testEnv.DB, + jobFor(DID_A, NSID.packageRelease, "demo:1.0.0:extra", { operation: "delete" }), + NOW, + ), + ).rejects.toMatchObject({ name: "IngestError", reason: "RKEY_MISMATCH" }); + }); + + it("applyDelete rejects rkeys whose package portion violates the slug regex", async () => { + await expect( + applyDelete( + testEnv.DB, + jobFor(DID_A, NSID.packageRelease, "9bad-leading-digit:1.0.0", { operation: "delete" }), + NOW, + ), + ).rejects.toMatchObject({ name: "IngestError", reason: "RKEY_MISMATCH" }); + }); +}); + +// Anchors the imports so a future refactor that drops them gets flagged. The +// classes are referenced indirectly via toMatchObject({ name }) assertions. +const _imports: ReadonlyArray = [IngestError, PdsVerificationError]; +void _imports; diff --git a/apps/aggregator/wrangler.jsonc b/apps/aggregator/wrangler.jsonc index ff37163952..416856c222 100644 --- a/apps/aggregator/wrangler.jsonc +++ b/apps/aggregator/wrangler.jsonc @@ -34,6 +34,19 @@ "max_retries": 5, "dead_letter_queue": "emdash-aggregator-records-dlq", }, + { + // Drains the DLQ. Today the consumer logs each dead-lettered + // job to Workers logs and writes a `dead_letters` row, then + // acks. On D1 failure the handler calls `message.retry()` + // (configured `max_retries: 3` per below); after that, the + // DLQ has no DLQ-of-DLQ so workerd drops the message. Once + // the reconciliation pass lands it'll replace this with + // retry-from-listRecords. + "queue": "emdash-aggregator-records-dlq", + "max_batch_size": 25, + "max_batch_timeout": 30, + "max_retries": 3, + }, ], }, "durable_objects": { diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index f19d09cbae..1d2a091327 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -42,6 +42,9 @@ catalogs: '@atcute/firehose': specifier: ^1.0.0 version: 1.0.0 + '@atcute/identity': + specifier: ^2.0.0 + version: 2.0.0 '@atcute/identity-resolver': specifier: ^1.2.2 version: 1.2.2 @@ -305,6 +308,9 @@ importers: '@atcute/firehose': specifier: 'catalog:' version: 1.0.0(@atcute/cid@2.4.1)(@atcute/lexicons@1.3.0)(react@19.2.4) + '@atcute/identity': + specifier: 'catalog:' + version: 2.0.0(@atcute/lexicons@1.3.0)(typescript@5.9.3) '@atcute/identity-resolver': specifier: 'catalog:' version: 1.2.2(@atcute/identity@2.0.0(@atcute/lexicons@1.3.0)(typescript@5.9.3)) diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml index 9a6146c583..8db462e9e5 100644 --- a/pnpm-workspace.yaml +++ b/pnpm-workspace.yaml @@ -28,6 +28,7 @@ catalog: "@atcute/client": ^4.2.1 "@atcute/crypto": ^2.4.1 "@atcute/firehose": ^1.0.0 + "@atcute/identity": ^2.0.0 "@atcute/identity-resolver": ^1.2.2 "@atcute/jetstream": ^2.0.0 "@atcute/lex-cli": ^2.8.1