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

/**
 * The full-backup payload, written out incrementally instead of built.
 *
 * Why it exists. the old whole-string builder (removed with the streaming path) had 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. The two sections whose rows carry
 * stored bytes are deferred the same way (#1239): the ECG traces a page at a
 * time, and the documents one at a time with each document's bytes arriving
 * in slices (`StreamedField`), because a vault of 190 MB read whole cost about
 * five times that in heap and took a 524 MB process down on its own. Every
 * other section is written an element at a time and then RELEASED from the
 * payload before the next one is touched, so the largest string this writer
 * makes is one row's.
 *
 * 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 { randomUUID } from "node:crypto";

import type { PrismaClient } from "@/generated/prisma/client";
import { reportJobProgress } from "@/lib/jobs/job-observer";
import {
  buildFullBackupPayload,
  isDeferredRows,
  type FullBackupCounts,
  type FullBackupOptions,
} from "@/lib/export/full-backup-payload";
import {
  isStreamedField,
  type StreamedField,
} from "@/lib/export/streamed-field";

/**
 * Write a row whose one large field arrives in pieces, exactly as
 * `JSON.stringify` writes the whole row: the row is serialised with a marker
 * in the field's place, and the marker is replaced by the quoted pieces on the
 * way out. The marker is random per call, so no value in the row can equal it.
 */
async function writeStreamedRow(
  streamed: StreamedField,
  write: (piece: string) => Promise<void>,
): Promise<void> {
  const marker = `\u0000streamed:${randomUUID()}`;
  const json = JSON.stringify({ ...streamed.row, [streamed.field]: marker });
  const quoted = JSON.stringify(marker);
  const at = json.indexOf(quoted);
  if (at < 0) {
    throw new Error(`Streamed field ${streamed.field} is not in its row`);
  }
  await write(json.slice(0, at));
  await write('"');
  for await (const piece of streamed.pieces()) await write(piece);
  await write('"');
  await write(json.slice(at + quoted.length));
}

/** 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;

/** How often a bulk section reports how far it has got to a running job. */
const PROGRESS_EVERY_ROWS = 50_000;

/**
 * There is deliberately no heap reading in this writer.
 *
 * There used to be: every flush compared the process's live heap usage against
 * 80 % of V8's heap limit and aborted the backup above it. That reading is not
 * the backup's footprint — it is the whole process's, garbage included, and a
 * long-lived Next.js server sits at 400 MB of largely COLLECTABLE heap. On a
 * 1 GB container (a 524 MB V8 limit, so a 419 MB budget) the check therefore
 * fired on the first flush of every account: a weekly pass over four accounts
 * aborted all four in seven seconds, and the smallest of them was a demo
 * record whose entire stored copy is 1.2 MB. It turned "this process has been
 * up a while" into "your record is too large", and then told the operator to
 * buy memory.
 *
 * What this writer holds is bounded by its own shape and needs no gauge to
 * prove: the three tables that scale with a record are pulled a page at a
 * time, every other section is released as soon as its JSON is in the sink,
 * and at most `FLUSH_BYTES` of text is ever pending. The sink does not hold
 * the record either: the stored copy goes to the database in sealed pieces
 * (`packBackupChunks`), and the one limit on its size is a storage limit it
 * counts against its own bytes, not against a heap the rest of the process
 * shares.
 */

/**
 * 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 interface StreamFullBackupOptions extends FullBackupOptions {
  /**
   * Shown each top-level member's value just before it is written, and each
   * row of the three bulk tables as it passes, under the member's name. For
   * a reader that needs to see the copy as it is written (the stored copy's
   * preview, `backup-preview.ts`) without parsing the JSON a second time.
   * It must not keep what it is shown: the writer releases each section as
   * soon as it is written, and that is what bounds its memory.
   */
  observe?: (member: string, value: unknown) => void;
}

export async function streamFullBackupJson(
  prisma: PrismaClient,
  userId: string,
  sink: BackupJsonSink,
  options: StreamFullBackupOptions = {},
): Promise<FullBackupCounts> {
  const { observe, ...payloadOptions } = options;
  const { payload, counts } = await buildFullBackupPayload(prisma, userId, {
    ...payloadOptions,
    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 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()) {
        if (rows > 0) await write(",");
        if (isStreamedField(row)) {
          observe?.(key, { ...row.row, [row.field]: row.observed });
          await writeStreamedRow(row, write);
        } else {
          observe?.(key, row);
          await write(JSON.stringify(row));
        }
        rows++;
        if (rows % PROGRESS_EVERY_ROWS === 0) {
          reportJobProgress({ backup_section: key, backup_section_rows: rows });
        }
      }
      await write("]");
      bulkCounts[key] = rows;
    } else if (Array.isArray(value)) {
      // An element at a time, so the largest string this writer makes is one
      // row's, never a section's: `JSON.stringify` of a whole section was a
      // second copy of it on top of the rows (#1239). The same bytes either
      // way; an element `JSON.stringify` cannot write becomes `null`, as it
      // does inside an array.
      observe?.(key, value);
      await write("[");
      for (let at = 0; at < value.length; at++) {
        if (at > 0) await write(",");
        await write(JSON.stringify(value[at]) ?? "null");
      }
      await write("]");
    } else {
      observe?.(key, value);
      await write(JSON.stringify(value));
    }
    delete payload[key];
  }
  await write("}");
  await flush(true);

  return { ...counts, ...bulkCounts } as FullBackupCounts;
}