feat(crypto): rotate the document blob column in bounded, codec-aware batches

The registry entry for InboundDocument.contentEncrypted gains a codecField
marker: the column now holds two layouts (base64v1 string codec | binary2
binary codec), recorded per row, and its rows are up to cap-sized blobs.
scan/rotate walk such a column with id-cursor pagination (25 rows per
page — never an unbounded blob findMany) and re-encrypt each row under its
OWN codec; an unknown codec value is counted as an error and the row left
untouched, same fail-closed posture as a missing key. The walk is
idempotent, so an interrupted run resumes on re-invocation with the
already-active rows skipped.

Both rotation paths share the logic: the admin-triggered pg-boss job via
rotateCorpus() and the CLI (scripts/rotate-encryption-key.ts), which now
calls the shared batched walk for this column. The registry guard test
keeps schema, registry, and script in lock-step as before.

Unit coverage over a mocked 500-row corpus asserts the batch boundedness
(take + advancing cursor per page), both-codec rotation, per-row
fail-closed on unknown codecs, and interrupted-run resume.

Task: W1-T11.

scripts/rotate-encryption-key.ts

@@ -24,6 +24,10 @@ import { Buffer } from "node:buffer";
 import { PrismaClient } from "@/generated/prisma/client";
 import { PrismaPg } from "@prisma/adapter-pg";
 import { decrypt, encrypt, extractKeyId, getActiveKeyId } from "@/lib/crypto";
+import {
+  rotateColumn,
+  type CorpusClient,
+} from "@/lib/crypto/encryption-corpus";
 
 const DATABASE_URL = process.env.DATABASE_URL;
 if (!DATABASE_URL) {
@@ -484,17 +488,31 @@ async function main() {
     ),
   );
 
-  // ───── v1.25 (W-DOCS-IN) inbound clinical document (Bytes column) ─────
-  // The raw uploaded doctor report / discharge letter (base64-of-binary →
-  // AES-256-GCM string → UTF-8 bytes). Always encrypted on write (NOT NULL),
-  // so every row rotates.
-  results.push(
-    await rotateBytesColumn(
-      "InboundDocument",
-      "contentEncrypted",
-      prisma.inboundDocument,
-    ),
-  );
+  // ───── Inbound clinical document (Bytes column, codec-dispatched) ─────
+  // The raw uploaded document. Two layouts recorded per row in
+  // `contentCodec` ("base64v1" string codec | "binary2" binary codec), so
+  // rotation goes through the shared codec-aware corpus walk: bounded
+  // id-cursor batches (rows are up to cap-sized blobs — never an unbounded
+  // findMany) re-encrypted under each row's OWN codec. Idempotent, so an
+  // interrupted run resumes on re-invocation.
+  {
+    const docResult = await rotateColumn(
+      { inboundDocument: prisma.inboundDocument } as unknown as CorpusClient,
+      {
+        model: "InboundDocument",
+        field: "contentEncrypted",
+        kind: "bytes",
+        codecField: "contentCodec",
+      },
+    );
+    results.push({
+      table: docResult.model,
+      field: docResult.field,
+      scanned: docResult.scanned,
+      rotated: docResult.rotated,
+      errors: docResult.errors,
+    });
+  }
   // The staged extracted-fact payloads: the FHIR-staged clinical values and the
   // verbatim source-span provenance. Both NOT NULL, so every staged row rotates.
   results.push(

src/lib/crypto/__tests__/encryption-corpus-blob.test.ts

@@ -0,0 +1,276 @@
+import { describe, it, expect, beforeEach } from "vitest";
+import { Buffer } from "node:buffer";
+import {
+  encrypt,
+  encryptBytes,
+  extractKeyIdFromBytes,
+  getActiveKeyId,
+  _resetCryptoCacheForTests,
+} from "@/lib/crypto";
+import {
+  rotateColumn,
+  scanColumn,
+  BLOB_ROTATION_BATCH_SIZE,
+  type CorpusClient,
+} from "@/lib/crypto/encryption-corpus";
+import type { EncryptedColumn } from "@/lib/crypto/encrypted-columns";
+
+/**
+ * Codec-dispatched blob rotation (the document vault's `contentEncrypted`):
+ * the walk is id-cursor paginated with a bounded batch size — never an
+ * unbounded blob `findMany` — re-encrypts each row under its OWN codec,
+ * fails closed per row on an unknown codec, and resumes cleanly after an
+ * interrupted run because already-active rows are skipped.
+ */
+
+const KEY_V1 =
+  "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
+const KEY_V2 =
+  "fedcba9876543210fedcba9876543210fedcba9876543210fedcba9876543210";
+
+const DOC_COLUMN: EncryptedColumn = {
+  model: "InboundDocument",
+  field: "contentEncrypted",
+  kind: "bytes",
+  codecField: "contentCodec",
+};
+
+interface DocRow {
+  id: string;
+  contentEncrypted: Uint8Array;
+  contentCodec: string;
+}
+
+interface FindManyArgs {
+  select: Record<string, true>;
+  orderBy?: Record<string, "asc" | "desc">;
+  take?: number;
+  cursor?: Record<string, unknown>;
+  skip?: number;
+}
+
+/**
+ * Pagination-honouring fake delegate: applies orderBy(id) + cursor + skip +
+ * take exactly like Prisma keyset pagination, and records every findMany
+ * call so the test can assert the batch boundedness.
+ */
+function makeDocClient(rows: DocRow[]): {
+  client: CorpusClient;
+  store: DocRow[];
+  findManyCalls: FindManyArgs[];
+  failUpdatesFor: Set<string>;
+} {
+  const store = [...rows];
+  const findManyCalls: FindManyArgs[] = [];
+  const failUpdatesFor = new Set<string>();
+  const client = {
+    inboundDocument: {
+      findMany: async (args: FindManyArgs) => {
+        findManyCalls.push(args);
+        let list = [...store].sort((a, b) => a.id.localeCompare(b.id));
+        if (args.cursor) {
+          const idx = list.findIndex((r) => r.id === args.cursor!.id);
+          list = list.slice(idx + (args.skip ?? 0));
+        }
+        if (args.take !== undefined) list = list.slice(0, args.take);
+        return list.map((row) => {
+          const out: Record<string, unknown> = {};
+          for (const f of Object.keys(args.select)) {
+            out[f] = row[f as keyof DocRow];
+          }
+          return out;
+        });
+      },
+      update: async ({
+        where,
+        data,
+      }: {
+        where: { id: string };
+        data: Record<string, unknown>;
+      }) => {
+        if (failUpdatesFor.has(where.id)) {
+          throw new Error("simulated interruption");
+        }
+        const row = store.find((r) => r.id === where.id);
+        if (row) Object.assign(row, data);
+        return row;
+      },
+    },
+  } as unknown as CorpusClient;
+  return { client, store, findManyCalls, failUpdatesFor };
+}
+
+function binaryUnder(keyId: "v1" | "v2", plaintext: Buffer): Uint8Array {
+  process.env.ENCRYPTION_ACTIVE_KEY_ID = keyId;
+  _resetCryptoCacheForTests();
+  const out = encryptBytes(plaintext);
+  process.env.ENCRYPTION_ACTIVE_KEY_ID = "v2";
+  _resetCryptoCacheForTests();
+  return new Uint8Array(out);
+}
+
+function legacyUnder(keyId: "v1" | "v2", plaintext: string): Uint8Array {
+  process.env.ENCRYPTION_ACTIVE_KEY_ID = keyId;
+  _resetCryptoCacheForTests();
+  const ct = encrypt(Buffer.from(plaintext).toString("base64"));
+  process.env.ENCRYPTION_ACTIVE_KEY_ID = "v2";
+  _resetCryptoCacheForTests();
+  return new Uint8Array(Buffer.from(ct, "utf8"));
+}
+
+function pad(i: number): string {
+  return `doc-${String(i).padStart(5, "0")}`;
+}
+
+beforeEach(() => {
+  process.env.ENCRYPTION_KEYS = JSON.stringify({ v1: KEY_V1, v2: KEY_V2 });
+  process.env.ENCRYPTION_ACTIVE_KEY_ID = "v2";
+  delete process.env.ENCRYPTION_KEY;
+  _resetCryptoCacheForTests();
+});
+
+describe("codec-dispatched blob rotation (bounded batches)", () => {
+  it("walks a 500-row corpus in bounded id-cursor batches and rotates both codecs", async () => {
+    const rows: DocRow[] = [];
+    // 200 binary2 under the stale v1, 150 legacy base64v1 under v1,
+    // 100 binary2 already on v2 (skipped), 50 legacy already on v2 (skipped).
+    for (let i = 0; i < 200; i++) {
+      rows.push({
+        id: pad(i),
+        contentEncrypted: binaryUnder("v1", Buffer.from(`bin-${i}`)),
+        contentCodec: "binary2",
+      });
+    }
+    for (let i = 200; i < 350; i++) {
+      rows.push({
+        id: pad(i),
+        contentEncrypted: legacyUnder("v1", `legacy-${i}`),
+        contentCodec: "base64v1",
+      });
+    }
+    for (let i = 350; i < 450; i++) {
+      rows.push({
+        id: pad(i),
+        contentEncrypted: binaryUnder("v2", Buffer.from(`bin-${i}`)),
+        contentCodec: "binary2",
+      });
+    }
+    for (let i = 450; i < 500; i++) {
+      rows.push({
+        id: pad(i),
+        contentEncrypted: legacyUnder("v2", `legacy-${i}`),
+        contentCodec: "base64v1",
+      });
+    }
+
+    const { client, store, findManyCalls } = makeDocClient(rows);
+    const result = await rotateColumn(client, DOC_COLUMN);
+
+    expect(result.scanned).toBe(500);
+    expect(result.rotated).toBe(350);
+    expect(result.errors).toBe(0);
+
+    // BOUNDED: every page requested at most the batch size, and the walk
+    // needed ceil(500 / batch) full pages (+1 terminating empty page, since
+    // 500 is an exact multiple) — never one unbounded findMany.
+    expect(findManyCalls.length).toBe(
+      Math.ceil(500 / BLOB_ROTATION_BATCH_SIZE) + 1,
+    );
+    for (const call of findManyCalls) {
+      expect(call.take).toBe(BLOB_ROTATION_BATCH_SIZE);
+      expect(call.orderBy).toEqual({ id: "asc" });
+    }
+    // Cursor advanced between pages.
+    expect(findManyCalls[0].cursor).toBeUndefined();
+    expect(findManyCalls[1].cursor).toEqual({ id: pad(24) });
+
+    // Every row now reads under the active key, each in its OWN codec.
+    for (const row of store) {
+      if (row.contentCodec === "binary2") {
+        expect(extractKeyIdFromBytes(Buffer.from(row.contentEncrypted))).toBe(
+          getActiveKeyId(),
+        );
+      } else {
+        expect(
+          Buffer.from(row.contentEncrypted).toString("utf8").startsWith("v2."),
+        ).toBe(true);
+      }
+    }
+
+    // IDEMPOTENT: the second pass rotates nothing.
+    const second = await rotateColumn(client, DOC_COLUMN);
+    expect(second.rotated).toBe(0);
+    expect(second.errors).toBe(0);
+  });
+
+  it("resumes after an interrupted run (already-rotated rows are skipped)", async () => {
+    const rows: DocRow[] = [];
+    for (let i = 0; i < 60; i++) {
+      rows.push({
+        id: pad(i),
+        contentEncrypted: binaryUnder("v1", Buffer.from(`bin-${i}`)),
+        contentCodec: "binary2",
+      });
+    }
+    const { client, failUpdatesFor } = makeDocClient(rows);
+
+    // First run: the last 20 updates fail (simulated interruption) — they
+    // are counted, never dropped.
+    for (let i = 40; i < 60; i++) failUpdatesFor.add(pad(i));
+    const first = await rotateColumn(client, DOC_COLUMN);
+    expect(first.rotated).toBe(40);
+    expect(first.errors).toBe(20);
+
+    // Second run with the interruption cleared: exactly the remainder
+    // rotates — the 40 already-active rows are skipped.
+    failUpdatesFor.clear();
+    const second = await rotateColumn(client, DOC_COLUMN);
+    expect(second.rotated).toBe(20);
+    expect(second.errors).toBe(0);
+  });
+
+  it("fails closed per row on an unknown codec and leaves the row untouched", async () => {
+    const strange = new Uint8Array(Buffer.from("???strange???"));
+    const { client, store } = makeDocClient([
+      {
+        id: pad(0),
+        contentEncrypted: binaryUnder("v1", Buffer.from("fine")),
+        contentCodec: "binary2",
+      },
+      { id: pad(1), contentEncrypted: strange, contentCodec: "codec9" },
+    ]);
+
+    const result = await rotateColumn(client, DOC_COLUMN);
+    expect(result.rotated).toBe(1);
+    expect(result.errors).toBe(1);
+    expect(store.find((r) => r.id === pad(1))!.contentEncrypted).toBe(strange);
+  });
+
+  it("scanColumn buckets blob rows per codec-aware key id in bounded batches", async () => {
+    const { client, findManyCalls } = makeDocClient([
+      {
+        id: pad(0),
+        contentEncrypted: binaryUnder("v1", Buffer.from("a")),
+        contentCodec: "binary2",
+      },
+      {
+        id: pad(1),
+        contentEncrypted: legacyUnder("v1", "b"),
+        contentCodec: "base64v1",
+      },
+      {
+        id: pad(2),
+        contentEncrypted: binaryUnder("v2", Buffer.from("c")),
+        contentCodec: "binary2",
+      },
+    ]);
+    const scan = await scanColumn(client, DOC_COLUMN);
+    expect(scan.total).toBe(3);
+    expect(scan.byKeyId.v1).toBe(2);
+    expect(scan.byKeyId.v2).toBe(1);
+    expect(scan.legacy).toBe(0);
+    for (const call of findManyCalls) {
+      expect(call.take).toBe(BLOB_ROTATION_BATCH_SIZE);
+    }
+  });
+});

src/lib/crypto/encrypted-columns.ts

@@ -34,6 +34,15 @@ export interface EncryptedColumn {
   /** Prisma field name (camelCase, as declared in schema.prisma). */
   readonly field: string;
   readonly kind: EncryptedColumnKind;
+  /**
+   * Present when the column's ciphertext layout is dispatched by a sibling
+   * codec column (the document vault's `contentCodec`: "base64v1" = the
+   * `encrypt()`-string-as-UTF-8 shape every other Bytes column uses,
+   * "binary2" = the binary `encryptBytes()` layout). A codec-dispatched
+   * column also rotates in bounded id-cursor batches — its rows are
+   * multi-megabyte blobs, so an unbounded `findMany` would balloon memory.
+   */
+  readonly codecField?: string;
 }
 
 /**
@@ -179,11 +188,19 @@ export const ENCRYPTED_COLUMNS: readonly EncryptedColumn[] = [
   { model: "Allergy", field: "notesEncrypted", kind: "bytes" },
   { model: "FamilyHistoryEntry", field: "notesEncrypted", kind: "bytes" },
 
-  // ───── v1.25 (W-DOCS-IN) inbound clinical document (Bytes column) ─────
-  // The raw uploaded doctor report / discharge letter, base64-of-binary →
-  // AES-256-GCM string → UTF-8 bytes. The most sensitive blob in the wave
-  // (a full clinical document); never logged, never in wide-event meta.
-  { model: "InboundDocument", field: "contentEncrypted", kind: "bytes" },
+  // ───── Inbound clinical document (Bytes column, codec-dispatched) ─────
+  // The raw uploaded document (the most sensitive blob in the corpus; never
+  // logged, never in wide-event meta). Two layouts, recorded per row in
+  // `contentCodec`: "base64v1" (pre-vault rows — base64-of-binary →
+  // AES-256-GCM string → UTF-8 bytes) and "binary2" (vault uploads — the
+  // binary `encryptBytes()` layout). Rotation walks this column in bounded
+  // id-cursor batches and re-encrypts under the row's own codec.
+  {
+    model: "InboundDocument",
+    field: "contentEncrypted",
+    kind: "bytes",
+    codecField: "contentCodec",
+  },
   // The staged extracted-fact payloads: the FHIR-staged clinical values
   // (diagnosis text / lab values / medication names / stated codes) and the
   // verbatim source-span provenance. Both are PHI transcribed from the source

src/lib/crypto/encryption-corpus.ts

@@ -30,7 +30,14 @@
  *    protects against dropping a legacy key too early.
  */
 import { Buffer } from "node:buffer";
-import { decrypt, encrypt, extractKeyId, getActiveKeyId } from "@/lib/crypto";
+import {
+  decrypt,
+  encrypt,
+  extractKeyId,
+  extractKeyIdFromBytes,
+  getActiveKeyId,
+  reencryptBytesToActive,
+} from "@/lib/crypto";
 import {
   ENCRYPTED_COLUMNS,
   encryptedColumnKey,
@@ -40,10 +47,21 @@ import {
 /** Sentinel bucket for legacy (unversioned) ciphertext under `byKeyId`. */
 export const LEGACY_BUCKET = "legacy";
 
+/**
+ * Batch size for codec-dispatched blob columns (`codecField` present). The
+ * document vault's rows are up to cap-sized ciphertexts, so the walk is
+ * id-cursor paginated — at most this many blobs are in memory at once.
+ */
+export const BLOB_ROTATION_BATCH_SIZE = 25;
+
 /** Minimal Prisma delegate shape this module needs. */
 interface ColumnDelegate {
   findMany: (args: {
     select: Record<string, true>;
+    orderBy?: Record<string, "asc" | "desc">;
+    take?: number;
+    cursor?: Record<string, unknown>;
+    skip?: number;
   }) => Promise<Array<Record<string, unknown>>>;
   update: (args: {
     where: { id: string };
@@ -98,6 +116,78 @@ function shouldRotate(ciphertext: string): boolean {
   return extractKeyId(ciphertext) !== getActiveKeyId();
 }
 
+// ─── Codec-dispatched blob columns (document vault) ─────────────────────────
+
+/**
+ * The key id a codec-dispatched blob row was written under, or null when
+ * legacy/unparsable. "binary2" parses the binary header; every other codec
+ * value (notably "base64v1") is the `encrypt()`-string-as-UTF-8 shape.
+ */
+function blobKeyId(value: Uint8Array, codec: string): string | null {
+  const buf = Buffer.from(value);
+  if (codec === "binary2") return extractKeyIdFromBytes(buf);
+  return extractKeyId(buf.toString("utf8"));
+}
+
+/** Re-encrypt one codec-dispatched blob under its OWN codec (never converts). */
+function reencryptBlob(value: Uint8Array, codec: string): Uint8Array {
+  const buf = Buffer.from(value);
+  if (codec === "binary2") {
+    const rotated = reencryptBytesToActive(buf);
+    const next = new Uint8Array(new ArrayBuffer(rotated.byteLength));
+    next.set(rotated);
+    return next;
+  }
+  if (codec === "base64v1") {
+    const rotated = encrypt(decrypt(buf.toString("utf8")));
+    const encoded = Buffer.from(rotated, "utf8");
+    const next = new Uint8Array(new ArrayBuffer(encoded.byteLength));
+    next.set(encoded);
+    return next;
+  }
+  // FAIL-CLOSED: an unknown codec is counted as an error and left untouched.
+  throw new Error(`Unknown content codec '${codec}'`);
+}
+
+/**
+ * Walk a codec-dispatched blob column in bounded id-cursor batches, invoking
+ * `onRow` per non-empty row. At most `BLOB_ROTATION_BATCH_SIZE` blobs are in
+ * memory per step; an interrupted run resumes safely on re-invocation because
+ * processing is idempotent (already-active rows are skipped by the callers).
+ */
+async function walkBlobColumn(
+  delegate: ColumnDelegate,
+  col: EncryptedColumn,
+  onRow: (row: {
+    id: string;
+    value: Uint8Array;
+    codec: string;
+  }) => Promise<void> | void,
+): Promise<void> {
+  const codecField = col.codecField!;
+  let cursor: string | null = null;
+  for (;;) {
+    const rows = await delegate.findMany({
+      select: { id: true, [col.field]: true, [codecField]: true },
+      orderBy: { id: "asc" },
+      take: BLOB_ROTATION_BATCH_SIZE,
+      ...(cursor ? { cursor: { id: cursor }, skip: 1 } : {}),
+    });
+    if (rows.length === 0) break;
+    for (const row of rows) {
+      const value = row[col.field] as Uint8Array | null;
+      if (!value || value.byteLength === 0) continue;
+      await onRow({
+        id: row.id as string,
+        value,
+        codec: String(row[codecField] ?? ""),
+      });
+    }
+    cursor = rows[rows.length - 1]!.id as string;
+    if (rows.length < BLOB_ROTATION_BATCH_SIZE) break;
+  }
+}
+
 export interface ColumnScan {
   model: string;
   field: string;
@@ -116,17 +206,27 @@ export async function scanColumn(
   col: EncryptedColumn,
 ): Promise<ColumnScan> {
   const delegate = getDelegate(client, col.model);
-  const rows = await delegate.findMany({
-    select: { id: true, [col.field]: true },
-  });
   const byKeyId: Record<string, number> = {};
   let total = 0;
-  for (const row of rows) {
-    const ciphertext = toCiphertext(row[col.field], col.kind);
-    if (ciphertext == null) continue;
-    total += 1;
-    const id = extractKeyId(ciphertext) ?? LEGACY_BUCKET;
-    byKeyId[id] = (byKeyId[id] ?? 0) + 1;
+
+  if (col.codecField) {
+    // Codec-dispatched blob column: bounded batches, per-row codec.
+    await walkBlobColumn(delegate, col, ({ value, codec }) => {
+      total += 1;
+      const id = blobKeyId(value, codec) ?? LEGACY_BUCKET;
+      byKeyId[id] = (byKeyId[id] ?? 0) + 1;
+    });
+  } else {
+    const rows = await delegate.findMany({
+      select: { id: true, [col.field]: true },
+    });
+    for (const row of rows) {
+      const ciphertext = toCiphertext(row[col.field], col.kind);
+      if (ciphertext == null) continue;
+      total += 1;
+      const id = extractKeyId(ciphertext) ?? LEGACY_BUCKET;
+      byKeyId[id] = (byKeyId[id] ?? 0) + 1;
+    }
   }
   return {
     model: col.model,
@@ -192,16 +292,41 @@ export async function rotateColumn(
   col: EncryptedColumn,
 ): Promise<RotationResult> {
   const delegate = getDelegate(client, col.model);
-  const rows = await delegate.findMany({
-    select: { id: true, [col.field]: true },
-  });
   const result: RotationResult = {
     model: col.model,
     field: col.field,
-    scanned: rows.length,
+    scanned: 0,
     rotated: 0,
     errors: 0,
   };
+
+  if (col.codecField) {
+    // Codec-dispatched blob column: bounded id-cursor batches (never an
+    // unbounded blob findMany), re-encrypted under each row's OWN codec.
+    // Idempotent — rows already on the active key are skipped, so an
+    // interrupted run resumes cleanly on the next invocation.
+    await walkBlobColumn(delegate, col, async ({ id, value, codec }) => {
+      result.scanned += 1;
+      if (blobKeyId(value, codec) === getActiveKeyId()) return;
+      try {
+        await delegate.update({
+          where: { id },
+          data: { [col.field]: reencryptBlob(value, codec) },
+        });
+        result.rotated += 1;
+      } catch {
+        // FAIL-CLOSED: unknown codec or a no-longer-configured key throws;
+        // count it and leave the row untouched rather than dropping data.
+        result.errors += 1;
+      }
+    });
+    return result;
+  }
+
+  const rows = await delegate.findMany({
+    select: { id: true, [col.field]: true },
+  });
+  result.scanned = rows.length;
   for (const row of rows) {
     const ciphertext = toCiphertext(row[col.field], col.kind);
     if (ciphertext == null || !shouldRotate(ciphertext)) continue;