Merge remote-tracking branch 'origin/fix/backup-streaming' into integration/post-1385

# Conflicts:
#	CHANGELOG.md

src/app/api/admin/backups/[id]/restore/__tests__/owner-mismatch.test.ts

@@ -36,7 +36,12 @@ vi.mock("@/lib/db", () => ({
   toJson: <T>(value: T) => value,
 }));
 
-vi.mock("@/lib/crypto", () => ({ decrypt: vi.fn(() => "{}") }));
+vi.mock("@/lib/crypto", () => ({
+  decrypt: vi.fn(() => "{}"),
+  // The envelope reader asks which stored shape it was handed first.
+  isStreamCiphertext: () => false,
+  decryptStream: vi.fn(),
+}));
 vi.mock("@/lib/crypto/note-cipher", () => ({ encryptNote: vi.fn() }));
 vi.mock("@/lib/validations/backup", () => ({
   parseBackupPayload: vi.fn(),

src/app/api/admin/backups/[id]/summary/__tests__/route.test.ts

@@ -30,8 +30,12 @@ vi.mock("@/lib/db", () => ({
 }));
 
 const decryptMock = vi.fn();
+// The envelope reader asks which of the stored shapes it was handed before it
+// decrypts anything, so a stub of this module has to answer that too.
 vi.mock("@/lib/crypto", () => ({
   decrypt: (...a: unknown[]) => decryptMock(...a),
+  isStreamCiphertext: () => false,
+  decryptStream: vi.fn(),
 }));
 
 const parseMock = vi.fn();

src/lib/crypto.ts

@@ -396,3 +396,139 @@ export function extractKeyId(encoded: string): string | null {
   if (!/^[A-Za-z0-9_-]{1,32}$/.test(id)) return null;
   return id;
 }
+
+// ─── Streaming codec ("stream1") ─────────────────────────────────────────────
+//
+// AES-256-GCM over a plaintext that is produced INCREMENTALLY and consumed as
+// one base64 string. Written for the weekly backup blob, whose plaintext is
+// hundreds of megabytes on a long-lived record: the string codec above has to
+// hold the whole plaintext, the whole ciphertext AND the whole base64 at once,
+// which is three full copies of a record that already does not fit.
+//
+// Layout: "~hlgcm1." + keyId + "." + base64(iv | ciphertext | authTag)
+//
+// Two deliberate differences from the string codec, and no others:
+//
+//   - The tag rides at the END rather than in front of the ciphertext. GCM
+//     only produces it once the last block is in, so a leading tag is what
+//     makes the string codec un-streamable. Nothing about the authentication
+//     changes: the tag still covers every ciphertext byte and `decryptStream`
+//     still verifies it BEFORE returning a single plaintext byte.
+//   - The `~` prefix. Key ids match [A-Za-z0-9_-]{1,32} and a legacy payload is
+//     bare base64, so no value the string codec can ever write starts with a
+//     tilde. The two formats are disjoint by construction rather than by
+//     sniffing, which is why `decrypt()` below is left untouched.
+//
+// The write side never materialises the ciphertext: base64 is emitted in
+// 3-byte-aligned pieces as the cipher produces them, with the sub-group
+// remainder carried across chunks. The read side is deliberately NOT
+// incremental — a caller that reads a backup parses it as one document
+// anyway, and releasing unverified plaintext to save a copy would trade the
+// authentication for memory.
+
+/** Marker prefix of a streamed ciphertext. Disjoint from every other format. */
+const STREAM_CODEC_PREFIX = "~hlgcm1.";
+
+/** Incremental AES-256-GCM writer. Emits base64 pieces; concatenate in order. */
+export interface StreamEncryptor {
+  /** The header, including base64(iv). Emit before any `update()` output. */
+  readonly header: string;
+  /** Encrypt one plaintext chunk into a base64 piece (may be empty). */
+  update(chunk: Buffer): string;
+  /** Flush the cipher and the auth tag. No further calls after this. */
+  final(): string;
+}
+
+/**
+ * Open a streaming encryptor under the ACTIVE key.
+ *
+ * The 12-byte IV is exactly four base64 groups, so it encodes standalone and
+ * the ciphertext continues on a group boundary — which is what lets the rest
+ * be emitted piecewise without ever holding the whole thing.
+ */
+export function createStreamEncryptor(): StreamEncryptor {
+  const { id, key } = getActiveKey();
+  const iv = randomBytes(IV_LENGTH);
+  const cipher = createCipheriv(ALGORITHM, key, iv);
+  // Bytes that did not fill a 3-byte base64 group in the previous chunk.
+  let carry = Buffer.alloc(0);
+  let done = false;
+
+  const emit = (bytes: Buffer): string => {
+    const buf = carry.byteLength > 0 ? Buffer.concat([carry, bytes]) : bytes;
+    const aligned = buf.byteLength - (buf.byteLength % 3);
+    carry = Buffer.from(buf.subarray(aligned));
+    return aligned === 0 ? "" : buf.subarray(0, aligned).toString("base64");
+  };
+
+  return {
+    header: `${STREAM_CODEC_PREFIX}${id}.${iv.toString("base64")}`,
+    update(chunk: Buffer): string {
+      if (done) throw new Error("Stream encryptor already finalised");
+      return emit(cipher.update(chunk));
+    },
+    final(): string {
+      if (done) throw new Error("Stream encryptor already finalised");
+      done = true;
+      // The tag is plaintext-independent trailing data, so it simply joins the
+      // byte stream; the last group is padded exactly once, here.
+      const tail = Buffer.concat([carry, cipher.final(), cipher.getAuthTag()]);
+      carry = Buffer.alloc(0);
+      return tail.toString("base64");
+    },
+  };
+}
+
+/** True when `stored` was written by `createStreamEncryptor`. */
+export function isStreamCiphertext(stored: string): boolean {
+  return stored.startsWith(STREAM_CODEC_PREFIX);
+}
+
+/**
+ * Decrypt a streamed ciphertext back to its raw plaintext bytes.
+ *
+ * Fail-closed on every arm — a malformed header, an unconfigured key id, a
+ * truncated payload or a tag that does not verify all throw, and no plaintext
+ * is returned until `final()` has authenticated the whole ciphertext.
+ */
+export function decryptStream(stored: string): Buffer {
+  if (!isStreamCiphertext(stored)) {
+    throw new Error("Not a streamed ciphertext");
+  }
+  const rest = stored.slice(STREAM_CODEC_PREFIX.length);
+  const dot = rest.indexOf(".");
+  if (dot <= 0 || dot >= rest.length - 1) {
+    throw new Error("Streamed ciphertext has a malformed header");
+  }
+  const keyId = rest.slice(0, dot);
+  if (!/^[A-Za-z0-9_-]{1,32}$/.test(keyId)) {
+    throw new Error("Streamed ciphertext carries a malformed key id");
+  }
+  const key = getKeyById(keyId);
+  if (!key) {
+    throw new Error(
+      `Encryption key id '${keyId}' is not configured. Add it to ` +
+        `ENCRYPTION_KEYS before decrypting rows written under that key.`,
+    );
+  }
+  const packed = Buffer.from(rest.slice(dot + 1), "base64");
+  if (packed.byteLength < IV_LENGTH + AUTH_TAG_LENGTH) {
+    throw new Error("Streamed ciphertext is truncated");
+  }
+  const iv = packed.subarray(0, IV_LENGTH);
+  const tag = packed.subarray(packed.byteLength - AUTH_TAG_LENGTH);
+  const ct = packed.subarray(IV_LENGTH, packed.byteLength - AUTH_TAG_LENGTH);
+  const dec = createDecipheriv(ALGORITHM, key, iv);
+  dec.setAuthTag(tag);
+  return Buffer.concat([dec.update(ct), dec.final()]);
+}
+
+/** The key id a streamed ciphertext was written under, or null when unparsable. */
+export function extractStreamKeyId(stored: string): string | null {
+  if (!isStreamCiphertext(stored)) return null;
+  const rest = stored.slice(STREAM_CODEC_PREFIX.length);
+  const dot = rest.indexOf(".");
+  if (dot <= 0) return null;
+  const keyId = rest.slice(0, dot);
+  return /^[A-Za-z0-9_-]{1,32}$/.test(keyId) ? keyId : null;
+}

src/lib/export/__tests__/backup-blob.test.ts

@@ -20,7 +20,20 @@ process.env.ENCRYPTION_KEY ??=
 import { describe, expect, it } from "vitest";
 
 import { encrypt } from "@/lib/crypto";
-import { packBackupBlob, unpackBackupBlob } from "../backup-blob";
+import {
+  packBackupBlob,
+  packBackupBlobStreaming,
+  unpackBackupBlob,
+} from "../backup-blob";
+
+/** Pack a whole string through the streaming writer, in small pieces. */
+async function packStreamed(json: string, pieceSize = 4_096): Promise<string> {
+  return packBackupBlobStreaming(async (write) => {
+    for (let at = 0; at < json.length; at += pieceSize) {
+      await write(json.slice(at, at + pieceSize));
+    }
+  });
+}
 
 /** A record-shaped string: many rows, few distinct keys. */
 function sampleJson(rows: number): string {
@@ -62,4 +75,62 @@ describe("backup blob envelope", () => {
     const mangled = `${packed.slice(0, -8)}AAAAAAAA`;
     expect(() => unpackBackupBlob(mangled)).toThrow();
   });
+
+  it("reads back a streamed blob byte for byte", async () => {
+    const json = sampleJson(2_000);
+    expect(unpackBackupBlob(await packStreamed(json))).toBe(json);
+  });
+
+  it("is indifferent to where the producer's pieces fall", async () => {
+    // Base64 encodes three bytes at a time, so a writer that carries its
+    // sub-group remainder across chunks wrongly still round-trips at some
+    // piece sizes and corrupts at others. Cover every residue class.
+    // Small on purpose: a one-byte piece size means one write per character,
+    // and the point of the case is the residue class, not the volume.
+    const json = sampleJson(20);
+    for (const pieceSize of [1, 2, 3, 7, 64, 1_000, 1_001, json.length]) {
+      expect(unpackBackupBlob(await packStreamed(json, pieceSize))).toBe(json);
+    }
+  });
+
+  it("compresses the streamed form as hard as the buffered one", async () => {
+    const json = sampleJson(5_000);
+    expect((await packStreamed(json)).length).toBeLessThan(json.length / 4);
+  });
+
+  it("keeps the three stored shapes readable side by side", async () => {
+    // An operator's newest usable copy may predate either change, and the
+    // restore has to work for exactly that person. Plain, compressed and
+    // streamed all decode through the one entry point.
+    const json = sampleJson(50);
+    expect(unpackBackupBlob(encrypt(json))).toBe(json);
+    expect(unpackBackupBlob(packBackupBlob(json))).toBe(json);
+    expect(unpackBackupBlob(await packStreamed(json))).toBe(json);
+  });
+
+  it("cannot be confused for a legacy blob", async () => {
+    // The marker starts with a tilde, which is in neither the key-id charset
+    // nor the base64 alphabet, so no value the older writers can produce
+    // begins with it and the two families never have to be sniffed apart.
+    const streamed = await packStreamed(sampleJson(10));
+    expect(streamed.startsWith("~")).toBe(true);
+    expect(packBackupBlob(sampleJson(10)).startsWith("~")).toBe(false);
+    expect(encrypt("x").startsWith("~")).toBe(false);
+  });
+
+  it("refuses a streamed blob whose authentication tag was altered", async () => {
+    const streamed = await packStreamed(sampleJson(20));
+    // The tag is the last 16 bytes of the payload, so the tail of the base64.
+    const mangled = `${streamed.slice(0, -6)}${streamed.slice(-6) === "AAAAAA" ? "BBBBBB" : "AAAAAA"}`;
+    expect(() => unpackBackupBlob(mangled)).toThrow();
+  });
+
+  it("propagates a producer failure instead of storing a truncated blob", async () => {
+    await expect(
+      packBackupBlobStreaming(async (write) => {
+        await write('{"measurements":[');
+        throw new Error("row source failed");
+      }),
+    ).rejects.toThrow("row source failed");
+  });
 });

src/lib/export/backup-blob.ts

@@ -13,22 +13,38 @@
  *
  * Compressing first is what makes the whole tail of that pipeline cheap. A
  * health record's JSON is extremely repetitive — the same twenty keys per row,
- * timestamps sharing a prefix — so gzip takes that 242 MB to 14 MB, and every
- * copy after it shrinks with it. The same run then completes in about eight
- * seconds with a peak heap of 434 MB. It is also the reason the stored row
- * stops being a liability of its own: the backup lives INSIDE the database it
- * would be needed to restore, so its size is not a cosmetic concern.
+ * timestamps sharing a prefix — so gzip takes that 242 MB to a few tens of
+ * megabytes, and every copy after it shrinks with it. It is also the reason
+ * the stored row stops being a liability of its own: the backup lives INSIDE
+ * the database it would be needed to restore, so its size is not a cosmetic
+ * concern.
  *
- * Both directions. Rows written before this envelope existed are plain
- * `encrypt(json)` and have to keep restoring — an operator whose newest usable
- * copy predates the fix is exactly the person who needs it to work. The marker
- * sits inside the ciphertext rather than in front of it so the stored column
- * keeps the one shape `decrypt()` already understands, key ids and all.
+ * Compressing first was not enough on its own. Even with a small stored blob,
+ * `packBackupBlob` still needs the whole JSON as one argument, and building
+ * that string is what exhausts the heap — measured under
+ * `--max-old-space-size=450` on the seeded account: `FATAL ERROR: Reached heap
+ * limit`. So the writer the weekly job actually uses is
+ * `packBackupBlobStreaming`, which never sees a complete copy of the JSON, the
+ * gzip output or the ciphertext: rows go in a page at a time, gzip and the
+ * cipher consume them as they arrive, and only the base64 answer accumulates,
+ * because the destination is a single `text` column and one value is what the
+ * column takes.
+ *
+ * Every direction reads. Three shapes exist in the wild and all three restore:
+ * the original `encrypt(json)`, the compressed `encrypt("HLZ1:" + gz)`, and
+ * the streamed `~hlgcm1.…` form written from now on. An operator whose newest
+ * usable copy predates any of this is exactly the person who needs it to work.
  */
 import { Buffer } from "node:buffer";
-import { gunzipSync, gzipSync } from "node:zlib";
+import { createGzip, gunzipSync, gzipSync } from "node:zlib";
 
-import { decrypt, encrypt } from "@/lib/crypto";
+import {
+  createStreamEncryptor,
+  decrypt,
+  decryptStream,
+  encrypt,
+  isStreamCiphertext,
+} from "@/lib/crypto";
 
 /**
  * Prefix of the DECRYPTED plaintext when the body is gzipped-then-base64'd.
@@ -43,14 +59,98 @@ export function packBackupBlob(json: string): string {
   return encrypt(`${GZIP_MARKER}${compressed}`);
 }
 
+/**
+ * Produces the backup JSON in pieces. Every piece is written in order.
+ *
+ * Whatever it resolves to is ignored — the writer's own return value (the row
+ * counts) is the caller's business, not the envelope's.
+ */
+export type BackupJsonProducer = (
+  write: (chunk: string) => Promise<void>,
+) => Promise<unknown>;
+
+/**
+ * Serialised backup JSON, produced in pieces → the stored string.
+ *
+ * The pipeline is JSON piece → gzip → AES-256-GCM → base64, with nothing
+ * buffered end to end but the base64 answer. `producer` decides how big its
+ * pieces are; the gzip stream applies backpressure through the promise this
+ * hands back, so a fast producer cannot outrun the compressor and pile up
+ * chunks in the stream's internal queue.
+ *
+ * The one copy that cannot be avoided is the answer itself. `data_backups.data`
+ * is a single `text` column, so the row has to be one value, and one value has
+ * to exist as one string before the driver can bind it.
+ */
+export async function packBackupBlobStreaming(
+  producer: BackupJsonProducer,
+): Promise<string> {
+  const encryptor = createStreamEncryptor();
+  const pieces: string[] = [encryptor.header];
+  const gzip = createGzip();
+
+  let failure: unknown = null;
+  gzip.on("data", (chunk: Buffer) => {
+    try {
+      const piece = encryptor.update(chunk);
+      if (piece !== "") pieces.push(piece);
+    } catch (err) {
+      failure ??= err;
+      gzip.destroy(err as Error);
+    }
+  });
+
+  const finished = new Promise<void>((resolve, reject) => {
+    gzip.on("end", resolve);
+    gzip.on("error", reject);
+  });
+
+  const write = async (chunk: string): Promise<void> => {
+    if (failure) throw failure;
+    if (gzip.write(chunk, "utf8")) return;
+    await new Promise<void>((resolve, reject) => {
+      const onDrain = () => {
+        gzip.off("error", onError);
+        resolve();
+      };
+      const onError = (err: Error) => {
+        gzip.off("drain", onDrain);
+        reject(err);
+      };
+      gzip.once("drain", onDrain);
+      gzip.once("error", onError);
+    });
+  };
+
+  try {
+    await producer(write);
+  } catch (err) {
+    gzip.destroy();
+    throw err;
+  }
+  gzip.end();
+  await finished;
+  if (failure) throw failure;
+
+  pieces.push(encryptor.final());
+  return pieces.join("");
+}
+
 /**
  * A stored `DataBackup.data` string → the backup JSON.
  *
- * Fails closed on every arm: a bad key, a mangled ciphertext or a truncated
- * gzip member throws rather than returning a partial document, because every
- * caller goes on to parse the result as a whole account.
+ * Fails closed on every arm: a bad key, a mangled ciphertext, a tag that does
+ * not verify or a truncated gzip member throws rather than returning a partial
+ * document, because every caller goes on to parse the result as a whole
+ * account.
  */
 export function unpackBackupBlob(stored: string): string {
+  // Streamed form. Always gzipped — the streaming writer has no other mode —
+  // and the tag is verified over the whole ciphertext before a byte of this
+  // is unpacked.
+  if (isStreamCiphertext(stored)) {
+    return gunzipSync(decryptStream(stored)).toString("utf8");
+  }
   const plaintext = decrypt(stored);
   if (!plaintext.startsWith(GZIP_MARKER)) return plaintext;
   return gunzipSync(

src/lib/export/full-backup-payload.ts

@@ -9,6 +9,16 @@
  *
  * The decrypted notes are surfaced here (the backup is the human-readable
  * artefact); an admin restore re-encrypts them on re-insert.
+ *
+ * Two writers read this file, not one. `buildFullBackupPayload` materialises
+ * the whole payload, which is what the two export routes and every test want.
+ * `streamFullBackupJson` (`full-backup-stream.ts`) asks for the same payload
+ * with `deferBulk`, which leaves the three unbounded tables as markers and
+ * pulls their rows through itself — the weekly job takes that path because
+ * materialising a large record is what used to exhaust the container's heap.
+ * Both go through the same per-row serialisers below, so a column that reaches
+ * one reaches the other; the integration suite pins the two outputs byte for
+ * byte rather than trusting that sentence.
  */
 import { Buffer } from "node:buffer";
 import type { Prisma, PrismaClient } from "@/generated/prisma/client";
@@ -146,6 +156,14 @@ export interface FullBackupResult {
 export interface FullBackupOptions {
   purpose?: "portable-export" | "disaster-recovery";
   exportedAt?: Date;
+  /**
+   * Declare the three unbounded tables instead of reading them.
+   *
+   * `measurements`, `intakeEvents` and `moodEntries` come back as
+   * `DeferredRows` markers and their counts as 0. Only `streamFullBackupJson`
+   * sets this; every other caller wants the arrays and gets them.
+   */
+  deferBulk?: boolean;
 }
 /**
  * Every restorable `Measurement` scalar. `userId` is deliberately absent — the
@@ -266,6 +284,35 @@ type BackupMoodContext = Prisma.MoodContextGetPayload<{
   select: typeof MOOD_CONTEXT_BACKUP_SELECT;
 }>;
 
+type BackupMeasurement = Prisma.MeasurementGetPayload<{
+  select: typeof MEASUREMENT_BACKUP_SELECT;
+}>;
+
+type BackupIntakeEvent = Prisma.MedicationIntakeEventGetPayload<{
+  include: { medication: { select: { name: true } } };
+}>;
+
+/**
+ * The mood read, select and joins together. Named because two readers issue
+ * it — the materialising one and the paged one — and a select that lived in
+ * only one of them is exactly the drift the completeness guard cannot see.
+ */
+const MOOD_ENTRY_QUERY_SELECT = {
+  ...MOOD_ENTRY_BACKUP_SELECT,
+  // Every link, not only the rated ones. This read used to filter
+  // `rating: { not: null }`, and a BINARY link carries a NULL rating by
+  // definition — so the present/absent tags, the kind most people pick,
+  // were absent from the file and could not come back from it.
+  tagLinks: { select: MOOD_ENTRY_TAG_LINK_SELECT },
+  // The day context, when the entry has one. Sparse by design — most
+  // entries carry none, and the join answers null on those.
+  context: { select: MOOD_CONTEXT_BACKUP_SELECT },
+} as const satisfies Prisma.MoodEntrySelect;
+
+type BackupMoodEntry = Prisma.MoodEntryGetPayload<{
+  select: typeof MOOD_ENTRY_QUERY_SELECT;
+}>;
+
 /**
  * One context row, in whichever of the two wire shapes the caller asked for.
  *
@@ -316,24 +363,339 @@ function serialiseMoodContext(
   };
 }
 
+/**
+ * One measurement, in whichever of the two wire shapes the caller asked for.
+ *
+ * Named rather than inlined so the paged reader below and the materialising
+ * reader beside it cannot drift: both call this, so a column that reaches one
+ * shape reaches the other by construction. `measurement-backup-completeness.
+ * test.ts` proves a selected column reaches a payload by matching
+ * `<key>: measurement.`, which is why the parameter keeps that spelling.
+ */
+function serialiseMeasurement(
+  measurement: BackupMeasurement,
+  disasterRecovery: boolean,
+) {
+  return disasterRecovery
+    ? {
+        id: measurement.id,
+        type: measurement.type,
+        value: measurement.value,
+        valueMin: measurement.valueMin,
+        valueMax: measurement.valueMax,
+        unit: measurement.unit,
+        measuredAt: measurement.measuredAt.toISOString(),
+        source: measurement.source,
+        notes: measurement.notes,
+        notesEncrypted: measurement.notesEncrypted
+          ? Buffer.from(measurement.notesEncrypted).toString("base64")
+          : null,
+        externalId: measurement.externalId,
+        externalSourceVersion: measurement.externalSourceVersion,
+        aggregationProvenance: measurement.aggregationProvenance,
+        glucoseContext: measurement.glucoseContext,
+        sleepStage: measurement.sleepStage,
+        rhythmClassification: measurement.rhythmClassification,
+        deviceType: measurement.deviceType,
+        syncVersion: measurement.syncVersion,
+        deletedAt: measurement.deletedAt?.toISOString() ?? null,
+        createdAt: measurement.createdAt.toISOString(),
+        updatedAt: measurement.updatedAt.toISOString(),
+      }
+    : {
+        id: measurement.id,
+        type: measurement.type,
+        value: measurement.value,
+        unit: measurement.unit,
+        measuredAt: measurement.measuredAt.toISOString(),
+        source: measurement.source,
+        notes: readNote(measurement.notesEncrypted, measurement.notes),
+        deletedAt: measurement.deletedAt?.toISOString() ?? null,
+      };
+}
+
+/** One intake event, as it rides the wire. */
+function serialiseIntakeEvent(e: BackupIntakeEvent, disasterRecovery: boolean) {
+  return {
+    ...(disasterRecovery
+      ? {
+          id: e.id,
+          medicationId: e.medicationId,
+          autoMissed: e.autoMissed,
+          attributionSource: e.attributionSource,
+          idempotencyKey: e.idempotencyKey,
+          createdAt: e.createdAt.toISOString(),
+          injectionSite: e.injectionSite,
+          doseTaken: e.doseTaken,
+          inventoryConsumption: e.inventoryConsumption,
+          externalId: e.externalId,
+          updatedAt: e.updatedAt.toISOString(),
+          syncVersion: e.syncVersion,
+          deletedAt: e.deletedAt?.toISOString() ?? null,
+        }
+      : {}),
+    medication: e.medication.name,
+    scheduledFor: e.scheduledFor.toISOString(),
+    takenAt: e.takenAt?.toISOString() ?? null,
+    skipped: e.skipped,
+    source: e.source,
+  };
+}
+
+/**
+ * One mood entry, with its links partitioned and its day context folded in.
+ *
+ * The parameter is spelled out rather than the `e` every other mapper here
+ * uses, and deliberately: `mood-backup-completeness.test.ts` proves a
+ * selected column reaches a payload by matching `: moodEntry.`. A matcher
+ * on `: e.` would also match the intake-event mapper above it and could
+ * therefore never fail.
+ */
+function serialiseMoodEntry(
+  moodEntry: BackupMoodEntry,
+  disasterRecovery: boolean,
+) {
+  // The two link arms partition the entry's links exhaustively: a link is
+  // a rated factor when its tag is RATED and it actually carries a score,
+  // and a structured tag otherwise. Total on purpose — a malformed link
+  // (a RATED tag whose rating went missing) rides as a tag key and comes
+  // back NAMED as unresolvable on restore, rather than falling between the
+  // two arms and disappearing the way every BINARY link used to.
+  const isRatedFactor = (link: (typeof moodEntry.tagLinks)[number]) =>
+    link.moodTag.kind === "RATED" && link.rating !== null;
+
+  const factors = moodEntry.tagLinks.filter(isRatedFactor).map((tagLink) => ({
+    key: tagLink.moodTag.key,
+    rating: tagLink.rating as number,
+  }));
+  // BINARY link keys. Their own array rather than a nullable rating inside
+  // `factors`: the restore resolves `factors` against `kind: "RATED"` and
+  // reports what it cannot resolve, so a rating-less entry in that array
+  // would collide with both. A separate key leaves every backup file
+  // written before this parsing and restoring exactly as it did.
+  const structuredTags = moodEntry.tagLinks
+    .filter((link) => !isRatedFactor(link))
+    .map((tagLink) => tagLink.moodTag.key);
+
+  return {
+    ...(disasterRecovery
+      ? {
+          note: moodEntry.note,
+          noteEncrypted: moodEntry.noteEncrypted
+            ? Buffer.from(moodEntry.noteEncrypted).toString("base64")
+            : null,
+          syncedAt: moodEntry.syncedAt.toISOString(),
+          syncVersion: moodEntry.syncVersion,
+          createdAt: moodEntry.createdAt.toISOString(),
+          updatedAt: moodEntry.updatedAt.toISOString(),
+        }
+      : { note: readNote(moodEntry.noteEncrypted, moodEntry.note) }),
+    id: moodEntry.id,
+    date: moodEntry.date,
+    mood: moodEntry.mood,
+    score: moodEntry.score,
+    // The five level-A values, under the keys the write path takes. Both
+    // arms carry them: they are the entry's own answers, not a storage
+    // detail, and a portable export that dropped them would hand back a
+    // file describing the day less well than the row it came from.
+    a1: moodEntry.moodA1,
+    a2: moodEntry.stressA2,
+    a3: moodEntry.energyA3,
+    a4: moodEntry.connectionA4,
+    a5: moodEntry.stabilityA5,
+    tags: moodEntry.tags,
+    source: moodEntry.source,
+    loggedAt: moodEntry.moodLoggedAt.toISOString(),
+    externalId: moodEntry.externalId,
+    // The zone the `date` string is anchored to. Without it a restored row
+    // falls back to the legacy Europe/Berlin reading and an account that
+    // logged elsewhere gets its day boundaries quietly moved.
+    tz: moodEntry.tz,
+    deletedAt: moodEntry.deletedAt?.toISOString() ?? null,
+    factors,
+    structuredTags,
+    // The day context, or null. Null rather than an empty object, because
+    // "no context" and "a context whose every field is empty" are
+    // different facts and the restore has to keep them apart.
+    context: moodEntry.context
+      ? serialiseMoodContext(moodEntry.context, disasterRecovery)
+      : null,
+  };
+}
+
+/**
+ * How many rows one page of a bulk table carries.
+ *
+ * Smaller than the measurement pager's 10 000 on purpose: an intake event
+ * drags its medication's name along and a mood entry drags its links and its
+ * day context, so a page of either costs several times what a page of
+ * measurements costs. The number is a memory budget, not a throughput knob.
+ */
+const BULK_PAGE_SIZE = 5_000;
+
+/**
+ * Every backup measurement, one serialised row at a time.
+ *
+ * Keyset-paged through `iterateMeasurementPages`, so the caller decides
+ * whether to keep the rows (the materialising payload does) or hand each
+ * straight to a writer and forget it (the streaming writer does). Nothing here
+ * holds more than one page.
+ */
+export async function* iterateBackupMeasurements(
+  prisma: PrismaClient,
+  userId: string,
+  disasterRecovery: boolean,
+): AsyncGenerator<Record<string, unknown>, void, void> {
+  const pages = iterateMeasurementPages(
+    prisma,
+    disasterRecovery ? { userId } : { userId, deletedAt: null },
+    MEASUREMENT_BACKUP_SELECT,
+  );
+  for await (const page of pages) {
+    for (const measurement of page) {
+      yield serialiseMeasurement(measurement, disasterRecovery);
+    }
+  }
+}
+
+/**
+ * Every backup intake event, one serialised row at a time.
+ *
+ * The sort is TOTAL — `scheduledFor desc` then `id desc` — where the
+ * materialising read used to order by the timestamp alone. A keyset cursor
+ * needs a total order to page correctly, and a partial one also let two
+ * exports of the same account emit two different files whenever two doses
+ * shared a scheduled minute, which every multi-dose day has.
+ */
+export async function* iterateBackupIntakeEvents(
+  prisma: PrismaClient,
+  userId: string,
+  disasterRecovery: boolean,
+): AsyncGenerator<Record<string, unknown>, void, void> {
+  let cursorId: string | null = null;
+  for (;;) {
+    const page: BackupIntakeEvent[] =
+      await prisma.medicationIntakeEvent.findMany({
+        where: disasterRecovery ? { userId } : { userId, deletedAt: null },
+        include: { medication: { select: { name: true } } },
+        orderBy: [{ scheduledFor: "desc" }, { id: "desc" }],
+        take: BULK_PAGE_SIZE,
+        ...(cursorId !== null ? { cursor: { id: cursorId }, skip: 1 } : {}),
+      });
+    if (page.length === 0) return;
+    for (const e of page) yield serialiseIntakeEvent(e, disasterRecovery);
+    if (page.length < BULK_PAGE_SIZE) return;
+    cursorId = page[page.length - 1]!.id;
+  }
+}
+
+/**
+ * Every backup mood entry, one serialised row at a time.
+ *
+ * Same total-order argument as the intake events above: `moodLoggedAt desc`
+ * alone is not a cursor, and two entries logged in the same millisecond are
+ * exactly what a bulk import produces.
+ */
+export async function* iterateBackupMoodEntries(
+  prisma: PrismaClient,
+  userId: string,
+  disasterRecovery: boolean,
+): AsyncGenerator<Record<string, unknown>, void, void> {
+  let cursorId: string | null = null;
+  for (;;) {
+    const page: BackupMoodEntry[] = await prisma.moodEntry.findMany({
+      where: disasterRecovery ? { userId } : { userId, deletedAt: null },
+      orderBy: [{ moodLoggedAt: "desc" }, { id: "desc" }],
+      take: BULK_PAGE_SIZE,
+      ...(cursorId !== null ? { cursor: { id: cursorId }, skip: 1 } : {}),
+      select: MOOD_ENTRY_QUERY_SELECT,
+    });
+    if (page.length === 0) return;
+    for (const moodEntry of page) {
+      yield serialiseMoodEntry(moodEntry, disasterRecovery);
+    }
+    if (page.length < BULK_PAGE_SIZE) return;
+    cursorId = page[page.length - 1]!.id;
+  }
+}
+
+/** Drain an async row source into an array. The materialising arm's only use. */
+async function collect<T>(rows: AsyncIterable<T>): Promise<T[]> {
+  const out: T[] = [];
+  for await (const row of rows) out.push(row);
+  return out;
+}
+
+/**
+ * A bulk table the payload declares but has NOT read.
+ *
+ * `buildFullBackupPayload({ deferBulk: true })` puts one of these where the
+ * row array would sit. The streaming writer recognises it, opens the JSON
+ * array itself and pulls rows through page by page, so a table with half a
+ * million rows never exists as an array at all. Every other caller leaves
+ * `deferBulk` unset and gets the arrays, unchanged.
+ *
+ * The marker key is a symbol so it cannot collide with a payload field and so
+ * the deferred payload can never be handed to `JSON.stringify` by accident and
+ * come back looking merely empty — `streamFullBackupJson` is the only reader.
+ */
+const DEFERRED_ROWS = Symbol("healthlog.backup.deferredRows");
+
+export interface DeferredRows {
+  readonly [DEFERRED_ROWS]: true;
+  /** Open the row source. Called exactly once, by the streaming writer. */
+  rows(): AsyncIterable<Record<string, unknown>>;
+}
+
+function deferredRows(
+  rows: () => AsyncIterable<Record<string, unknown>>,
+): DeferredRows {
+  return { [DEFERRED_ROWS]: true, rows };
+}
+
+/** True when a payload value is a deferred bulk table rather than an array. */
+export function isDeferredRows(value: unknown): value is DeferredRows {
+  return (
+    typeof value === "object" &&
+    value !== null &&
+    (value as Record<symbol, unknown>)[DEFERRED_ROWS] === true
+  );
+}
+
 /**
  * The same payload, already serialised, with the object graph released.
  *
- * Its own frame on purpose, and both backup jobs go through it. The payload
- * graph is the largest thing either job allocates — on an account with a few
- * hundred thousand measurements it is several hundred megabytes of small
- * objects — and neither job wants anything but the string. Returning from here
- * makes the graph unreachable before the compress-and-encrypt step allocates
- * anything of its own, which is part of why the weekly pass now fits in
- * memory. Inlining these two statements back into a caller quietly undoes it.
+ * Built through the streaming writer rather than by stringifying a finished
+ * payload, so the object graph and the string never coexist: the three
+ * unbounded tables go past a page at a time and every other section is
+ * released as soon as its JSON exists. What is left is the answer itself,
+ * which a caller that wants a string necessarily holds.
+ *
+ * A caller that does NOT need the string — the weekly job, which only wants
+ * the compressed ciphertext — should go through `packBackupBlobStreaming` and
+ * `streamFullBackupJson` directly and never let the whole document exist at
+ * all. This function is for the callers that genuinely hand a JSON body on.
  */
 export async function buildFullBackupJson(
   prisma: PrismaClient,
   userId: string,
   options: FullBackupOptions = {},
 ): Promise<string> {
-  const { payload } = await buildFullBackupPayload(prisma, userId, options);
-  return JSON.stringify(payload);
+  // Imported here rather than at the top of the file: the writer imports this
+  // module for `buildFullBackupPayload`, and a static edge in both directions
+  // is a cycle. The call is once per backup, so the cost is nil.
+  const { streamFullBackupJson } =
+    await import("@/lib/export/full-backup-stream");
+  const pieces: string[] = [];
+  await streamFullBackupJson(
+    prisma,
+    userId,
+    (chunk) => {
+      pieces.push(chunk);
+    },
+    options,
+  );
+  return pieces.join("");
 }
 
 /**
@@ -347,6 +709,7 @@ export async function buildFullBackupPayload(
   options: FullBackupOptions = {},
 ): Promise<FullBackupResult> {
   const disasterRecovery = options.purpose === "disaster-recovery";
+  const deferBulk = options.deferBulk === true;
   const [
     appSettings,
     measurements,
@@ -376,60 +739,9 @@ export async function buildFullBackupPayload(
     disasterRecovery
       ? prisma.appSettings.findUnique({ where: { id: "singleton" } })
       : Promise.resolve(null),
-    (async () => {
-      const rows: Array<Record<string, unknown>> = [];
-      const pages = iterateMeasurementPages(
-        prisma,
-        disasterRecovery ? { userId } : { userId, deletedAt: null },
-        MEASUREMENT_BACKUP_SELECT,
-      );
-      for await (const page of pages) {
-        for (const measurement of page) {
-          rows.push(
-            disasterRecovery
-              ? {
-                  id: measurement.id,
-                  type: measurement.type,
-                  value: measurement.value,
-                  valueMin: measurement.valueMin,
-                  valueMax: measurement.valueMax,
-                  unit: measurement.unit,
-                  measuredAt: measurement.measuredAt.toISOString(),
-                  source: measurement.source,
-                  notes: measurement.notes,
-                  notesEncrypted: measurement.notesEncrypted
-                    ? Buffer.from(measurement.notesEncrypted).toString("base64")
-                    : null,
-                  externalId: measurement.externalId,
-                  externalSourceVersion: measurement.externalSourceVersion,
-                  aggregationProvenance: measurement.aggregationProvenance,
-                  glucoseContext: measurement.glucoseContext,
-                  sleepStage: measurement.sleepStage,
-                  rhythmClassification: measurement.rhythmClassification,
-                  deviceType: measurement.deviceType,
-                  syncVersion: measurement.syncVersion,
-                  deletedAt: measurement.deletedAt?.toISOString() ?? null,
-                  createdAt: measurement.createdAt.toISOString(),
-                  updatedAt: measurement.updatedAt.toISOString(),
-                }
-              : {
-                  id: measurement.id,
-                  type: measurement.type,
-                  value: measurement.value,
-                  unit: measurement.unit,
-                  measuredAt: measurement.measuredAt.toISOString(),
-                  source: measurement.source,
-                  notes: readNote(
-                    measurement.notesEncrypted,
-                    measurement.notes,
-                  ),
-                  deletedAt: measurement.deletedAt?.toISOString() ?? null,
-                },
-          );
-        }
-      }
-      return rows;
-    })(),
+    deferBulk
+      ? null
+      : collect(iterateBackupMeasurements(prisma, userId, disasterRecovery)),
     prisma.medication.findMany({
       where: { userId },
       // Side effects ride INSIDE their medication, exactly as the schedules
@@ -469,26 +781,12 @@ export async function buildFullBackupPayload(
         },
       },
     }),
-    prisma.medicationIntakeEvent.findMany({
-      where: disasterRecovery ? { userId } : { userId, deletedAt: null },
-      include: { medication: { select: { name: true } } },
-      orderBy: { scheduledFor: "desc" },
-    }),
-    prisma.moodEntry.findMany({
-      where: disasterRecovery ? { userId } : { userId, deletedAt: null },
-      orderBy: { moodLoggedAt: "desc" },
-      select: {
-        ...MOOD_ENTRY_BACKUP_SELECT,
-        // Every link, not only the rated ones. This read used to filter
-        // `rating: { not: null }`, and a BINARY link carries a NULL rating by
-        // definition — so the present/absent tags, the kind most people pick,
-        // were absent from the file and could not come back from it.
-        tagLinks: { select: MOOD_ENTRY_TAG_LINK_SELECT },
-        // The day context, when the entry has one. Sparse by design — most
-        // entries carry none, and the join answers null on those.
-        context: { select: MOOD_CONTEXT_BACKUP_SELECT },
-      },
-    }),
+    deferBulk
+      ? null
+      : collect(iterateBackupIntakeEvents(prisma, userId, disasterRecovery)),
+    deferBulk
+      ? null
+      : collect(iterateBackupMoodEntries(prisma, userId, disasterRecovery)),
     // The account's OWN mood tags. The seeded catalogue is reference data every
     // instance has; a tag the user made lives only here. Without it the restore
     // cannot resolve the entry's factor keys — and unlike the cycle path, which
@@ -646,6 +944,27 @@ export async function buildFullBackupPayload(
   const environmentSection: EnvironmentBackupSection = environment;
   const ecgSection: EcgBackupSection = ecg;
 
+  // The three unbounded tables, either read into arrays or left as markers
+  // the streaming writer will pull through. One place decides, so the payload
+  // literal below reads the same in both modes.
+  const bulk = {
+    measurements: deferBulk
+      ? deferredRows(() =>
+          iterateBackupMeasurements(prisma, userId, disasterRecovery),
+        )
+      : measurements!,
+    intakeEvents: deferBulk
+      ? deferredRows(() =>
+          iterateBackupIntakeEvents(prisma, userId, disasterRecovery),
+        )
+      : intakeEvents!,
+    moodEntries: deferBulk
+      ? deferredRows(() =>
+          iterateBackupMoodEntries(prisma, userId, disasterRecovery),
+        )
+      : moodEntries!,
+  };
+
   // Where each archived era sits in its OWN drug's ordered list. Built once
   // here rather than per row, and scoped per medication because the supersede
   // pointer never crosses a drug — every route that writes it scopes the
@@ -674,7 +993,7 @@ export async function buildFullBackupPayload(
             documentQuotaBytes: appSettings.documentQuotaBytes.toString(),
           }
         : null,
-    measurements,
+    measurements: bulk.measurements,
     medications: medications.map((m) => ({
       ...(disasterRecovery
         ? {
@@ -903,105 +1222,8 @@ export async function buildFullBackupPayload(
         primary: t.primary,
       })),
     })),
-    intakeEvents: intakeEvents.map((e) => ({
-      ...(disasterRecovery
-        ? {
-            id: e.id,
-            medicationId: e.medicationId,
-            autoMissed: e.autoMissed,
-            attributionSource: e.attributionSource,
-            idempotencyKey: e.idempotencyKey,
-            createdAt: e.createdAt.toISOString(),
-            injectionSite: e.injectionSite,
-            doseTaken: e.doseTaken,
-            inventoryConsumption: e.inventoryConsumption,
-            externalId: e.externalId,
-            updatedAt: e.updatedAt.toISOString(),
-            syncVersion: e.syncVersion,
-            deletedAt: e.deletedAt?.toISOString() ?? null,
-          }
-        : {}),
-      medication: e.medication.name,
-      scheduledFor: e.scheduledFor.toISOString(),
-      takenAt: e.takenAt?.toISOString() ?? null,
-      skipped: e.skipped,
-      source: e.source,
-    })),
-    // The parameter is spelled out rather than the `e` every other mapper here
-    // uses, and deliberately: `mood-backup-completeness.test.ts` proves a
-    // selected column reaches a payload by matching `: moodEntry.`. A matcher
-    // on `: e.` would also match the intake-event mapper above it and could
-    // therefore never fail.
-    moodEntries: moodEntries.map((moodEntry) => {
-      // The two link arms partition the entry's links exhaustively: a link is
-      // a rated factor when its tag is RATED and it actually carries a score,
-      // and a structured tag otherwise. Total on purpose — a malformed link
-      // (a RATED tag whose rating went missing) rides as a tag key and comes
-      // back NAMED as unresolvable on restore, rather than falling between the
-      // two arms and disappearing the way every BINARY link used to.
-      const isRatedFactor = (link: (typeof moodEntry.tagLinks)[number]) =>
-        link.moodTag.kind === "RATED" && link.rating !== null;
-
-      const factors = moodEntry.tagLinks
-        .filter(isRatedFactor)
-        .map((tagLink) => ({
-          key: tagLink.moodTag.key,
-          rating: tagLink.rating as number,
-        }));
-      // BINARY link keys. Their own array rather than a nullable rating inside
-      // `factors`: the restore resolves `factors` against `kind: "RATED"` and
-      // reports what it cannot resolve, so a rating-less entry in that array
-      // would collide with both. A separate key leaves every backup file
-      // written before this parsing and restoring exactly as it did.
-      const structuredTags = moodEntry.tagLinks
-        .filter((link) => !isRatedFactor(link))
-        .map((tagLink) => tagLink.moodTag.key);
-
-      return {
-        ...(disasterRecovery
-          ? {
-              note: moodEntry.note,
-              noteEncrypted: moodEntry.noteEncrypted
-                ? Buffer.from(moodEntry.noteEncrypted).toString("base64")
-                : null,
-              syncedAt: moodEntry.syncedAt.toISOString(),
-              syncVersion: moodEntry.syncVersion,
-              createdAt: moodEntry.createdAt.toISOString(),
-              updatedAt: moodEntry.updatedAt.toISOString(),
-            }
-          : { note: readNote(moodEntry.noteEncrypted, moodEntry.note) }),
-        id: moodEntry.id,
-        date: moodEntry.date,
-        mood: moodEntry.mood,
-        score: moodEntry.score,
-        // The five level-A values, under the keys the write path takes. Both
-        // arms carry them: they are the entry's own answers, not a storage
-        // detail, and a portable export that dropped them would hand back a
-        // file describing the day less well than the row it came from.
-        a1: moodEntry.moodA1,
-        a2: moodEntry.stressA2,
-        a3: moodEntry.energyA3,
-        a4: moodEntry.connectionA4,
-        a5: moodEntry.stabilityA5,
-        tags: moodEntry.tags,
-        source: moodEntry.source,
-        loggedAt: moodEntry.moodLoggedAt.toISOString(),
-        externalId: moodEntry.externalId,
-        // The zone the `date` string is anchored to. Without it a restored row
-        // falls back to the legacy Europe/Berlin reading and an account that
-        // logged elsewhere gets its day boundaries quietly moved.
-        tz: moodEntry.tz,
-        deletedAt: moodEntry.deletedAt?.toISOString() ?? null,
-        factors,
-        structuredTags,
-        // The day context, or null. Null rather than an empty object, because
-        // "no context" and "a context whose every field is empty" are
-        // different facts and the restore has to keep them apart.
-        context: moodEntry.context
-          ? serialiseMoodContext(moodEntry.context, disasterRecovery)
-          : null,
-      };
-    }),
+    intakeEvents: bulk.intakeEvents,
+    moodEntries: bulk.moodEntries,
     customMoodTagCategories: customMoodTagCategories.map((category) => ({
       id: category.id,
       key: category.key,
@@ -1079,9 +1301,9 @@ export async function buildFullBackupPayload(
   return {
     payload,
     counts: {
-      measurements: measurements.length,
+      measurements: measurements?.length ?? 0,
       medications: medications.length,
-      intakeEvents: intakeEvents.length,
+      intakeEvents: intakeEvents?.length ?? 0,
       medicationSideEffects: medications.reduce(
         (total, m) => total + m.sideEffects.length,
         0,
@@ -1111,7 +1333,7 @@ export async function buildFullBackupPayload(
         (total, m) => total + m.scheduleRevisions.length,
         0,
       ),
-      moodEntries: moodEntries.length,
+      moodEntries: moodEntries?.length ?? 0,
       customMoodTagCategories: customMoodTagCategories.length,
       hiddenMoodTags: hiddenMoodTags.length,
       cycles: cycle.cycles.length,

src/lib/export/full-backup-stream.ts

@@ -0,0 +1,160 @@
+/**
+ * The full-backup payload, written out incrementally instead of built.
+ *
+ * Why it exists. `buildFullBackupJson` has to hold two things at once that
+ * both scale with the record: the payload object graph, and the JSON string
+ * `JSON.stringify` makes of it. On a seeded account of 445 000 measurements,
+ * 30 000 mood entries and 60 000 intake events, that pair alone exhausts a
+ * 546 MB heap and the process dies with `Reached heap limit` — measured, not
+ * inferred, and reproducible with `--max-old-space-size=450`. The production
+ * container is capped at 1 GB, which puts V8's limit at 524 MB, so the weekly
+ * pass took the whole app down with it: every other user of that instance lost
+ * their session because one account's backup did not fit.
+ *
+ * What changed. The three tables that scale with a long-lived record —
+ * measurements, intake events, mood entries — are declared rather than read
+ * (`deferBulk`), and this writer pulls them through page by page, serialising
+ * one row at a time straight into the sink. Every other section is stringified
+ * and then RELEASED from the payload before the next one is touched, so the
+ * peak is the largest single section rather than the sum of all of them.
+ *
+ * The output is byte-identical to `JSON.stringify(payload)`. That is not a
+ * hope: `Object.entries` walks a plain object in insertion order, which is the
+ * order `JSON.stringify` uses, and `JSON.stringify([a, b])` is exactly
+ * `"[" + JSON.stringify(a) + "," + JSON.stringify(b) + "]"` for plain rows.
+ * The integration suite pins it against the materialising builder rather than
+ * trusting that paragraph.
+ */
+import v8 from "node:v8";
+
+import type { PrismaClient } from "@/generated/prisma/client";
+import {
+  buildFullBackupPayload,
+  isDeferredRows,
+  type FullBackupCounts,
+  type FullBackupOptions,
+} from "@/lib/export/full-backup-payload";
+
+/** Where the writer puts its pieces. Awaited, so a sink can apply backpressure. */
+export type BackupJsonSink = (chunk: string) => void | Promise<void>;
+
+/**
+ * How much serialised JSON to gather before handing it to the sink.
+ *
+ * Row-at-a-time writes would push half a million tiny strings through gzip and
+ * spend more time in framing than in compression; a batch this size keeps the
+ * resident buffer trivial and the write count in the low thousands.
+ */
+const FLUSH_BYTES = 256 * 1024;
+
+/**
+ * Fraction of V8's heap limit the writer refuses to cross.
+ *
+ * The point is not to make an oversized account succeed — it is to make it
+ * FAIL AS A JOB. A backup that runs out of heap takes the process with it, and
+ * a pg-boss worker sharing the app process means one account's record can
+ * restart the instance for everybody. Because the work is now incremental
+ * there is a checkpoint every few hundred kilobytes, so the guard actually
+ * gets to run before V8 hits the wall; the old single `JSON.stringify` had no
+ * point at which anything could intervene.
+ */
+const HEAP_HIGH_WATER = 0.8;
+
+/** Thrown when the writer stops itself short of the heap limit. */
+export class BackupHeapBudgetExceededError extends Error {
+  constructor(usedBytes: number, ceilingBytes: number) {
+    super(
+      `Backup aborted at ${Math.round(usedBytes / 1024 / 1024)} MB of heap ` +
+        `(budget ${Math.round(ceilingBytes / 1024 / 1024)} MB). The record is ` +
+        `too large for this container's memory limit; raise the container's ` +
+        `memory or NODE_OPTIONS=--max-old-space-size.`,
+    );
+    this.name = "BackupHeapBudgetExceededError";
+  }
+}
+
+export interface StreamFullBackupOptions extends FullBackupOptions {
+  /**
+   * Heap ceiling in bytes. Defaults to 80 % of V8's limit for this process.
+   * Tests pass an explicit value; nothing else should need to.
+   */
+  heapCeilingBytes?: number;
+}
+
+function defaultHeapCeiling(): number {
+  return Math.floor(v8.getHeapStatistics().heap_size_limit * HEAP_HIGH_WATER);
+}
+
+/**
+ * Write the full-backup JSON for `userId` into `sink`, and answer the counts.
+ *
+ * The counts for the three deferred tables are filled in as their rows go
+ * past, so a caller gets the same numbers the materialising builder reports.
+ */
+export async function streamFullBackupJson(
+  prisma: PrismaClient,
+  userId: string,
+  sink: BackupJsonSink,
+  options: StreamFullBackupOptions = {},
+): Promise<FullBackupCounts> {
+  const ceiling = options.heapCeilingBytes ?? defaultHeapCeiling();
+  const { payload, counts } = await buildFullBackupPayload(prisma, userId, {
+    ...options,
+    deferBulk: true,
+  });
+
+  let pending = "";
+  let pendingBytes = 0;
+
+  const flush = async (force: boolean): Promise<void> => {
+    if (pending === "" || (!force && pendingBytes < FLUSH_BYTES)) return;
+    const chunk = pending;
+    pending = "";
+    pendingBytes = 0;
+    await sink(chunk);
+    const used = process.memoryUsage().heapUsed;
+    if (used > ceiling) throw new BackupHeapBudgetExceededError(used, ceiling);
+  };
+
+  const write = async (piece: string): Promise<void> => {
+    pending += piece;
+    pendingBytes += piece.length;
+    await flush(false);
+  };
+
+  const bulkCounts: Record<string, number> = {};
+
+  await write("{");
+  let first = true;
+  // The walk is destructive: each section is released from the payload as soon
+  // as its JSON is in the sink, which is what keeps the peak at one section
+  // rather than at all of them. Nothing else reads this object.
+  for (const key of Object.keys(payload)) {
+    const value = payload[key];
+    // `JSON.stringify` omits an undefined-valued key; so does this.
+    if (value === undefined) continue;
+    if (!first) await write(",");
+    first = false;
+    await write(`${JSON.stringify(key)}:`);
+
+    if (isDeferredRows(value)) {
+      await write("[");
+      let rows = 0;
+      for await (const row of value.rows()) {
+        await write(
+          rows === 0 ? JSON.stringify(row) : `,${JSON.stringify(row)}`,
+        );
+        rows++;
+      }
+      await write("]");
+      bulkCounts[key] = rows;
+    } else {
+      await write(JSON.stringify(value));
+    }
+    delete payload[key];
+  }
+  await write("}");
+  await flush(true);
+
+  return { ...counts, ...bulkCounts } as FullBackupCounts;
+}

src/lib/jobs/job-outcome.ts

@@ -103,6 +103,7 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet<string> = new Set([
   "geo_backfill_skipped",
   "geo_backfill_still_unresolved",
   "geolite2_fetch_status",
+  "heap_budget_trips",
   "host_metric_pruned",
   "hourly_rows_upserted",
   "imported",

src/lib/jobs/reminder/__tests__/backup-handlers.test.ts

@@ -9,23 +9,29 @@ const mocks = vi.hoisted(() => ({
   upsert: vi.fn(),
 }));
 
-// The handler takes the already-serialised form; the stand-in serialises
-// whatever the payload mock was told to return, so this file keeps stubbing
-// and asserting the PAYLOAD, which is what it is about.
+// Only the payload builder is stubbed. The REAL streaming writer runs on top
+// of it, so this file keeps stubbing and asserting the PAYLOAD — which is what
+// it is about — while the framing that turns it into stored bytes is the
+// framing the job actually uses. `isDeferredRows` rides along because the
+// writer asks it about every section; nothing here defers.
 vi.mock("@/lib/export/full-backup-payload", () => ({
   buildFullBackupPayload: mocks.buildFullBackupPayload,
-  buildFullBackupJson: async (...args: unknown[]) =>
-    JSON.stringify(
-      ((await mocks.buildFullBackupPayload(...args)) as { payload: unknown })
-        .payload,
-    ),
+  isDeferredRows: () => false,
 }));
 
 // The envelope is exercised end-to-end in
 // `src/lib/export/__tests__/backup-blob.test.ts`; here it stays transparent so
 // the stored bytes can be read back as JSON.
 vi.mock("@/lib/export/backup-blob", () => ({
-  packBackupBlob: mocks.packBlob,
+  packBackupBlobStreaming: async (
+    produce: (write: (chunk: string) => Promise<void>) => Promise<void>,
+  ) => {
+    let out = "";
+    await produce(async (chunk) => {
+      out += chunk;
+    });
+    return mocks.packBlob(out);
+  },
 }));
 
 vi.mock("@/lib/logging/background", () => ({
@@ -95,7 +101,12 @@ describe("handleDataBackup canonical DR payload", () => {
     expect(mocks.buildFullBackupPayload).toHaveBeenCalledWith(
       prisma,
       "user-dr",
-      { purpose: "disaster-recovery" },
+      // `deferBulk` is the writer's own ask: it declares the unbounded tables
+      // rather than reading them, and pulls their rows through itself.
+      expect.objectContaining({
+        purpose: "disaster-recovery",
+        deferBulk: true,
+      }),
     );
     expect(mocks.upsert).toHaveBeenCalledOnce();
     const encrypted = mocks.upsert.mock.calls[0]![0].create.data as string;

src/lib/jobs/reminder/backup-handlers.ts

@@ -6,8 +6,11 @@
  */
 import { type Job } from "pg-boss";
 import { recordError } from "@/lib/jobs/worker-status";
-import { buildFullBackupJson } from "@/lib/export/full-backup-payload";
-import { packBackupBlob } from "@/lib/export/backup-blob";
+import { packBackupBlobStreaming } from "@/lib/export/backup-blob";
+import {
+  BackupHeapBudgetExceededError,
+  streamFullBackupJson,
+} from "@/lib/export/full-backup-stream";
 import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome";
 import { withBackgroundEvent } from "@/lib/logging/background";
 import {
@@ -79,14 +82,19 @@ export async function handleDataBackup(
 
       let backed = 0;
       let usersFailed = 0;
+      let heapBudgetTrips = 0;
       let largestBlobBytes = 0;
       for (const user of users) {
         try {
-          // Compressed, then encrypted (the record contains sensitive health
-          // information). Both legs live in `packBackupBlob`, which is also
-          // what the restore path reads back through.
-          const encryptedBackup = packBackupBlob(
-            await buildFullBackupJson(prisma, user.id, {
+          // Streamed, compressed, then encrypted (the record contains
+          // sensitive health information). Nothing between the database rows
+          // and this string exists as a whole: the payload goes into gzip a
+          // page at a time and the cipher consumes gzip's output as it comes,
+          // because a materialised copy of a large record is what used to take
+          // the process down. Both legs live in `packBackupBlobStreaming`,
+          // which is also what the restore path reads back through.
+          const encryptedBackup = await packBackupBlobStreaming((write) =>
+            streamFullBackupJson(prisma, user.id, write, {
               purpose: "disaster-recovery",
             }),
           );
@@ -114,7 +122,14 @@ export async function handleDataBackup(
           // One user's payload failing is that user's problem, not the pass's:
           // it rides out as a count so the weekly run is not retried for the
           // whole cohort.
+          //
+          // A record too large for this container's heap lands here too, and
+          // that is the whole point of the budget the writer enforces: before
+          // it existed the same condition was an uncatchable V8 abort that
+          // restarted the instance, so one account's size was a denial of
+          // service on every other account on the host.
           usersFailed++;
+          if (err instanceof BackupHeapBudgetExceededError) heapBudgetTrips++;
           evt.addWarning(`Failed for user ${user.id}: ${err}`);
         }
       }
@@ -122,6 +137,9 @@ export async function handleDataBackup(
       // The stored size is the one number that says whether this pass is
       // heading back towards the wall it hit before: it tracks the record.
       evt.addMeta("data_backup_largest_blob_bytes", largestBlobBytes);
+      // How many accounts the writer stopped rather than let the process die.
+      // Non-zero means this host needs more memory, not a retry.
+      evt.addMeta("data_backup_heap_budget_trips", heapBudgetTrips);
       evt.setBackground({
         task_name: "job.data_backup",
         result: { backed, total: users.length },
@@ -130,6 +148,7 @@ export async function handleDataBackup(
         backed,
         total: users.length,
         users_failed: usersFailed,
+        heap_budget_trips: heapBudgetTrips,
       });
     } catch (err) {
       evt.setError(err);