mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-15 15:02:04 +00:00
fix(backup): export, upload and restore a large account as streams
An account of 1.25 million measurements could not export its data or be restored from any backup (#1031). Measured in a 1 GB container: - The full export and the passphrase export built the whole payload and its JSON string in memory; the app ran out of heap and restarted. - The weekly backup's JSON is 662 MB, longer than any string V8 can hold, so `unpackBackupBlob` threw on every read. Restore, download and the restore preview all answered "could not be decrypted". - Upload accepted 10 MB, a fraction of a large export. Every reader and writer of a backup file now streams it: - `scanBackupJson` reads a backup's bytes once, keeps every section but the named bulk arrays, and hands their elements out one at a time. `readStreamedBackup` uses it twice: a first pass that validates each measurement against the element schema and keeps the rest of the file, and a second pass that hands the measurements to the writer in batches. Stored copies and off-host objects are authenticated whole on their compressed ciphertext before any plaintext is read; only the decompression is streamed. - Restore writes the measurements with one `INSERT … FROM unnest` per batch. Prisma caches a query plan per `createMany` row count, and one 1 000-row plan kept about 145 MB alive, which alone put the restore past the heap limit. The transaction budget grows with the measurement count. The awards and ECG sections are handed the ids they reference that were written, not every written id. - The export routes write the file into the response as it is produced. The passphrase export spools its ciphertext to a temporary file because the HLX1 format puts the tag first; the format is unchanged, so existing archives open as before. - Upload accepts the raw body, JSON or gzip, up to the 512 MB request ceiling, checks it as it arrives and stores it in the same pass; the admin page compresses the file in the browser before sending it. Multipart stays for small files. - Download streams the stored JSON; the preview counts measurements from the stream. The monthly restore drill and `scripts/restore-backup.ts` read the off-host object the same way; the script can write `.json.gz`. On the seeded account in a 1 GB container: export 25 s at 389 MB peak, passphrase export 25 s, both uploads 19 s, restore of the weekly backup 94 s at 554 MB peak, and export, upload and restore round-trip to exactly the same rows. The encrypted-column guard now also recognises the raw SQL statement that writes `DataBackup.data`.
This commit is contained in:
Marc-André Bombeck committed
1 parent
79ed62b7f9
commit
237acc0002
38 files changed
+2250
-341
No files matched your search
@@ -0,0 +1,185 @@
|
||||
/**
|
||||
* `scanBackupJson` must read a backup exactly as `JSON.parse` of the whole
|
||||
* document would, with the streamed members handed out one element at a time
|
||||
* (issue #1031: a 662 MB backup cannot exist as one string). These tests hold
|
||||
* it to that over awkward values and arbitrary chunk boundaries, including
|
||||
* boundaries inside strings, escapes and multi-byte characters.
|
||||
*/
|
||||
import { describe, expect, it } from "vitest";
|
||||
|
||||
import { BackupJsonError, scanBackupJson } from "../backup-json-scan";
|
||||
|
||||
async function* chunked(text: string, sizes: number[]) {
|
||||
const bytes = Buffer.from(text, "utf8");
|
||||
let at = 0;
|
||||
let i = 0;
|
||||
while (at < bytes.length) {
|
||||
const size = sizes[i++ % sizes.length]!;
|
||||
yield bytes.subarray(at, at + size);
|
||||
at += size;
|
||||
}
|
||||
}
|
||||
|
||||
function rng(seed: number) {
|
||||
let x = seed;
|
||||
return () => {
|
||||
x = (x * 1103515245 + 12345) % 2 ** 31;
|
||||
return x / 2 ** 31;
|
||||
};
|
||||
}
|
||||
|
||||
const AWKWARD = [
|
||||
'quote " inside',
|
||||
"back\\slash and \\n",
|
||||
"brackets ] } [ { , : in a string",
|
||||
"umlaut ä ö ü and emoji 🫀",
|
||||
"",
|
||||
"\u0000\u001f control",
|
||||
];
|
||||
|
||||
function randomValue(rand: () => number, depth: number): unknown {
|
||||
const r = rand();
|
||||
if (depth > 2 || r < 0.35) {
|
||||
const p = rand();
|
||||
if (p < 0.2) return null;
|
||||
if (p < 0.3) return rand() < 0.5;
|
||||
if (p < 0.55) return Math.round(rand() * 1e6) / 100 - 500;
|
||||
return (
|
||||
AWKWARD[Math.floor(rand() * AWKWARD.length)]! + Math.floor(rand() * 9)
|
||||
);
|
||||
}
|
||||
if (r < 0.65) {
|
||||
return Array.from({ length: Math.floor(rand() * 4) }, () =>
|
||||
randomValue(rand, depth + 1),
|
||||
);
|
||||
}
|
||||
const obj: Record<string, unknown> = {};
|
||||
for (let i = 0; i < Math.floor(rand() * 4); i++) {
|
||||
obj[`${AWKWARD[i % AWKWARD.length]}-${i}`] = randomValue(rand, depth + 1);
|
||||
}
|
||||
return obj;
|
||||
}
|
||||
|
||||
function randomDocument(seed: number) {
|
||||
const rand = rng(seed);
|
||||
return {
|
||||
schemaVersion: "2",
|
||||
userId: "user-ä",
|
||||
measurements: Array.from({ length: Math.floor(rand() * 30) }, (_, i) =>
|
||||
i % 5 === 4
|
||||
? randomValue(rand, 1)
|
||||
: { id: `m${i}`, notes: AWKWARD[i % 6] },
|
||||
),
|
||||
medications: [randomValue(rand, 0), randomValue(rand, 0)],
|
||||
moodEntries: [],
|
||||
count: 42,
|
||||
flag: false,
|
||||
nothing: null,
|
||||
text: AWKWARD[seed % AWKWARD.length],
|
||||
nested: randomValue(rand, 0),
|
||||
};
|
||||
}
|
||||
|
||||
async function scan(text: string, sizes: number[], streamKeys: string[]) {
|
||||
const elements: Record<string, unknown[]> = {};
|
||||
const result = await scanBackupJson(chunked(text, sizes), {
|
||||
streamKeys: new Set(streamKeys),
|
||||
onElement: (key, element, index) => {
|
||||
(elements[key] ??= [])[index] = element;
|
||||
},
|
||||
});
|
||||
return { ...result, elements };
|
||||
}
|
||||
|
||||
describe("scanBackupJson", () => {
|
||||
it("equals JSON.parse over random documents, formats and chunk splits", async () => {
|
||||
for (let seed = 1; seed <= 200; seed++) {
|
||||
const doc = randomDocument(seed);
|
||||
const text =
|
||||
seed % 3 === 0
|
||||
? JSON.stringify(doc, null, seed % 2 ? 2 : "\t")
|
||||
: JSON.stringify(doc);
|
||||
const sizes = [1 + (seed % 7), 2, 13, 1, 64].slice(seed % 3);
|
||||
const { document, streamedCounts, elements } = await scan(text, sizes, [
|
||||
"measurements",
|
||||
"moodEntries",
|
||||
]);
|
||||
const expected = JSON.parse(text) as Record<string, unknown>;
|
||||
|
||||
expect(elements.measurements ?? [], `seed ${seed}`).toEqual(
|
||||
expected.measurements,
|
||||
);
|
||||
expect(streamedCounts).toEqual({
|
||||
measurements: (expected.measurements as unknown[]).length,
|
||||
moodEntries: 0,
|
||||
});
|
||||
expect(document, `seed ${seed}`).toEqual({
|
||||
...expected,
|
||||
measurements: [],
|
||||
moodEntries: [],
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
it("hands out streamed elements before it has read the rest of the document", async () => {
|
||||
const text = JSON.stringify({
|
||||
a: 1,
|
||||
measurements: [{ id: "x" }, { id: "y" }],
|
||||
tail: "z".repeat(100),
|
||||
});
|
||||
const bytes = Buffer.from(text);
|
||||
let delivered = 0;
|
||||
const seenAt: number[] = [];
|
||||
async function* source() {
|
||||
for (let i = 0; i < bytes.length; i += 4) {
|
||||
delivered = i + 4;
|
||||
yield bytes.subarray(i, i + 4);
|
||||
}
|
||||
}
|
||||
await scanBackupJson(source(), {
|
||||
streamKeys: new Set(["measurements"]),
|
||||
onElement: () => {
|
||||
seenAt.push(delivered);
|
||||
},
|
||||
});
|
||||
expect(seenAt).toHaveLength(2);
|
||||
expect(seenAt[1]!).toBeLessThan(bytes.length - 100);
|
||||
});
|
||||
|
||||
it("skips a member it was told to skip without keeping it", async () => {
|
||||
const { document } = await scanBackupJson(
|
||||
chunked('{"a":[1,{"b":"]"}],"big":{"x":[1,2,3]},"c":"d"}', [3]),
|
||||
{ skipKeys: new Set(["big"]) },
|
||||
);
|
||||
expect(document).toEqual({ a: [1, { b: "]" }], c: "d" });
|
||||
});
|
||||
|
||||
it("refuses a truncated or malformed document", async () => {
|
||||
for (const bad of [
|
||||
'{"a":1',
|
||||
'{"a":[1,2}',
|
||||
"[1,2]",
|
||||
'{"measurements":{"x":1}}',
|
||||
'{"a":1}{',
|
||||
'{"measurements":[{"id":1} {"id":2}]}',
|
||||
]) {
|
||||
await expect(
|
||||
scanBackupJson(chunked(bad, [2]), {
|
||||
streamKeys: new Set(["measurements"]),
|
||||
}),
|
||||
bad,
|
||||
).rejects.toBeInstanceOf(BackupJsonError);
|
||||
}
|
||||
});
|
||||
|
||||
it("propagates a failure from the element consumer", async () => {
|
||||
await expect(
|
||||
scanBackupJson(chunked('{"measurements":[1,2,3]}', [5]), {
|
||||
streamKeys: new Set(["measurements"]),
|
||||
onElement: (_key, _element, index) => {
|
||||
if (index === 1) throw new Error("consumer failed");
|
||||
},
|
||||
}),
|
||||
).rejects.toThrow("consumer failed");
|
||||
});
|
||||
});
|
||||
@@ -2,6 +2,7 @@ import { describe, it, expect } from "vitest";
|
||||
import { Buffer } from "node:buffer";
|
||||
import {
|
||||
encryptArchive,
|
||||
encryptArchiveToFile,
|
||||
decryptArchive,
|
||||
parseArchiveHeader,
|
||||
EXPORT_ARGON2_PARAMS,
|
||||
@@ -78,3 +79,45 @@ describe("passphrase-encrypted export archive (HLX1)", () => {
|
||||
expect(await decryptArchive(archive, PASSPHRASE)).toBe(PAYLOAD);
|
||||
});
|
||||
});
|
||||
|
||||
/**
|
||||
* #1031: the encrypted export is sealed as it is produced and spooled to a
|
||||
* file, because holding it did not fit a 1 GB container for a large account.
|
||||
* The format must not change: an archive written this way opens with the
|
||||
* same `decryptArchive` every existing archive opens with.
|
||||
*/
|
||||
describe("encryptArchiveToFile", () => {
|
||||
it("writes an archive decryptArchive opens, byte for byte the same document", async () => {
|
||||
const { mkdtemp, readFile, rm } = await import("node:fs/promises");
|
||||
const { tmpdir } = await import("node:os");
|
||||
const { join } = await import("node:path");
|
||||
const dir = await mkdtemp(join(tmpdir(), "hlx-spool-"));
|
||||
try {
|
||||
const pieces = Array.from({ length: 500 }, (_, i) =>
|
||||
JSON.stringify({ i, note: "ümlaut 🫀 ".repeat(i % 7) }),
|
||||
);
|
||||
const bodyPath = join(dir, "body");
|
||||
const spooled = await encryptArchiveToFile(
|
||||
async (write) => {
|
||||
await write("[");
|
||||
for (let i = 0; i < pieces.length; i++) {
|
||||
await write(i === 0 ? pieces[i]! : `,${pieces[i]}`);
|
||||
}
|
||||
await write(Buffer.from("]"));
|
||||
},
|
||||
PASSPHRASE,
|
||||
bodyPath,
|
||||
);
|
||||
const archive = Buffer.concat([spooled.prefix, await readFile(bodyPath)]);
|
||||
expect(archive.byteLength).toBe(spooled.byteLength);
|
||||
expect(await decryptArchive(archive, PASSPHRASE)).toBe(
|
||||
`[${pieces.join(",")}]`,
|
||||
);
|
||||
await expect(
|
||||
decryptArchive(archive, "wrong passphrase!!"),
|
||||
).rejects.toThrow();
|
||||
} finally {
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -37,7 +37,8 @@
|
||||
*/
|
||||
import { Buffer } from "node:buffer";
|
||||
import v8 from "node:v8";
|
||||
import { createGzip, gunzipSync, gzipSync } from "node:zlib";
|
||||
import { Readable } from "node:stream";
|
||||
import { createGunzip, createGzip, gunzipSync, gzipSync } from "node:zlib";
|
||||
|
||||
import {
|
||||
createStreamEncryptor,
|
||||
@@ -125,7 +126,7 @@ export class BackupBlobTooLargeError extends Error {
|
||||
* counts) is the caller's business, not the envelope's.
|
||||
*/
|
||||
export type BackupJsonProducer = (
|
||||
write: (chunk: string) => Promise<void>,
|
||||
write: (chunk: string | Buffer) => Promise<void>,
|
||||
) => Promise<unknown>;
|
||||
|
||||
export interface PackBackupBlobOptions {
|
||||
@@ -236,7 +237,7 @@ export async function packBackupBlobInto(
|
||||
await sink(piece);
|
||||
};
|
||||
|
||||
const write = async (chunk: string): Promise<void> => {
|
||||
const write = async (chunk: string | Buffer): Promise<void> => {
|
||||
if (failure) throw failure;
|
||||
await flush(false);
|
||||
if (gzip.write(chunk, "utf8")) return;
|
||||
@@ -313,3 +314,34 @@ export function unpackBackupBlob(stored: string): string {
|
||||
Buffer.from(plaintext.slice(GZIP_MARKER.length), "base64"),
|
||||
).toString("utf8");
|
||||
}
|
||||
|
||||
/**
|
||||
* A stored `DataBackup.data` string → a source of its JSON as byte chunks,
|
||||
* which can be opened as many times as the caller needs to read it.
|
||||
*
|
||||
* `unpackBackupBlob` returns the JSON as one string, and a string is what a
|
||||
* large record cannot be: the disaster-recovery JSON of an account with 1.25
|
||||
* million measurements is 662 MB, past the 536 870 888 characters V8 allows
|
||||
* in any string, so that call threw on every such backup (#1031). Here the
|
||||
* authentication happens first and whole, on the compressed ciphertext, which
|
||||
* stays small (48 MB for that record): no plaintext byte is released before
|
||||
* the tag has verified. Only the decompression is streamed, on every open.
|
||||
*
|
||||
* Fail-closed exactly like `unpackBackupBlob`: a bad key or a tag that does
|
||||
* not verify throws here, before a source exists; a truncated gzip member
|
||||
* errors the stream.
|
||||
*/
|
||||
export function openBackupBlob(stored: string): () => AsyncIterable<Buffer> {
|
||||
if (isStreamCiphertext(stored)) {
|
||||
const gz = decryptStream(stored);
|
||||
return () => Readable.from([gz]).pipe(createGunzip());
|
||||
}
|
||||
const plaintext = decrypt(stored);
|
||||
if (!plaintext.startsWith(GZIP_MARKER)) {
|
||||
// A pre-envelope copy was written from one string, so it is one.
|
||||
const bytes = Buffer.from(plaintext, "utf8");
|
||||
return () => Readable.from([bytes]);
|
||||
}
|
||||
const gz = Buffer.from(plaintext.slice(GZIP_MARKER.length), "base64");
|
||||
return () => Readable.from([gz]).pipe(createGunzip());
|
||||
}
|
||||
@@ -0,0 +1,336 @@
|
||||
/**
|
||||
* Read a backup's JSON as a stream, without ever holding the document.
|
||||
*
|
||||
* Why. A backup is one JSON object, and every reader used to turn it into
|
||||
* one string and `JSON.parse` it. For an account of 1.25 million measurements
|
||||
* (issue #1031) the disaster-recovery JSON is 662 MB. V8 cannot hold a string
|
||||
* longer than 536 870 888 characters at all, so `unpackBackupBlob` threw
|
||||
* before a byte was parsed: the weekly backup of that account could not be
|
||||
* restored, downloaded or even summarised, and the restore route reported the
|
||||
* copy as undecryptable. A portable export of the same account fits under
|
||||
* the limit but not beside the rest of a 1 GB container.
|
||||
*
|
||||
* What this does. It scans the bytes once, tracking only string and nesting
|
||||
* state. Every top-level member is parsed on its own with `JSON.parse` and
|
||||
* kept, except the members named in `streamKeys`: those must be arrays, and
|
||||
* each element is parsed and handed to `onElement` instead, so the caller
|
||||
* sees them one at a time and the kept object holds an empty array in their
|
||||
* place. A backup's bulk tables (measurements, intake events, mood entries)
|
||||
* are the only members that grow with a record, so everything else is small.
|
||||
*
|
||||
* Structural characters are ASCII and UTF-8 never uses an ASCII byte inside a
|
||||
* multi-byte sequence, so scanning bytes is exact. A value is decoded only
|
||||
* once it is complete, which is where a character split across two chunks is
|
||||
* put back together.
|
||||
*
|
||||
* The output is what `JSON.parse` of the whole document would give, with the
|
||||
* streamed arrays emptied: the tests hold it to that over arbitrary splits.
|
||||
*/
|
||||
|
||||
/** Characters the scanner cares about, as bytes. */
|
||||
const QUOTE = 0x22;
|
||||
const BACKSLASH = 0x5c;
|
||||
const OPEN_BRACE = 0x7b;
|
||||
const CLOSE_BRACE = 0x7d;
|
||||
const OPEN_BRACKET = 0x5b;
|
||||
const CLOSE_BRACKET = 0x5d;
|
||||
const COMMA = 0x2c;
|
||||
const COLON = 0x3a;
|
||||
|
||||
function isWhitespace(byte: number): boolean {
|
||||
return byte === 0x20 || byte === 0x0a || byte === 0x0d || byte === 0x09;
|
||||
}
|
||||
|
||||
export interface ScanBackupJsonOptions {
|
||||
/** Top-level members whose array elements are streamed, not kept. */
|
||||
streamKeys?: ReadonlySet<string>;
|
||||
/**
|
||||
* Receives each element of a streamed member, parsed, with its index.
|
||||
* Awaited, so a consumer that writes to a database applies backpressure.
|
||||
*/
|
||||
onElement?: (key: string, element: unknown, index: number) => unknown;
|
||||
/**
|
||||
* Members to skip entirely: scanned past, never parsed or kept. For a
|
||||
* reader that only needs a few sections of the document.
|
||||
*/
|
||||
skipKeys?: ReadonlySet<string>;
|
||||
}
|
||||
|
||||
export interface ScanBackupJsonResult {
|
||||
/** The document, with streamed members as empty arrays and skipped ones absent. */
|
||||
document: Record<string, unknown>;
|
||||
/** How many elements each streamed member held. */
|
||||
streamedCounts: Record<string, number>;
|
||||
}
|
||||
|
||||
/** Thrown for input that is not a JSON object the scanner can walk. */
|
||||
export class BackupJsonError extends Error {
|
||||
constructor(message: string) {
|
||||
super(message);
|
||||
this.name = "BackupJsonError";
|
||||
}
|
||||
}
|
||||
|
||||
/** Collects the bytes of one value across chunks. */
|
||||
class Collector {
|
||||
private parts: Buffer[] = [];
|
||||
private bytes = 0;
|
||||
add(chunk: Buffer, from: number, to: number): void {
|
||||
if (to > from) {
|
||||
this.parts.push(chunk.subarray(from, to));
|
||||
this.bytes += to - from;
|
||||
}
|
||||
}
|
||||
take(): string {
|
||||
const text =
|
||||
this.parts.length === 1
|
||||
? this.parts[0]!.toString("utf8")
|
||||
: Buffer.concat(this.parts, this.bytes).toString("utf8");
|
||||
this.parts = [];
|
||||
this.bytes = 0;
|
||||
return text;
|
||||
}
|
||||
}
|
||||
|
||||
type Phase =
|
||||
| "start" // before the opening brace
|
||||
| "key" // expecting a key or the closing brace
|
||||
| "colon"
|
||||
| "value" // expecting the start of a member value
|
||||
| "in-value" // inside a kept or skipped member value
|
||||
| "array-start" // a streamed member: expecting its opening bracket
|
||||
| "element" // inside a streamed array, expecting an element or `]`
|
||||
| "in-element" // inside one element
|
||||
| "after" // after a member, expecting `,` or `}`
|
||||
| "done";
|
||||
|
||||
/**
|
||||
* Scan a backup document delivered as byte chunks. Resolves to the document
|
||||
* (see the module comment) once the closing brace has been read; rejects with
|
||||
* `BackupJsonError` on malformed input or with whatever `onElement` threw.
|
||||
*/
|
||||
export async function scanBackupJson(
|
||||
chunks: AsyncIterable<Uint8Array>,
|
||||
options: ScanBackupJsonOptions = {},
|
||||
): Promise<ScanBackupJsonResult> {
|
||||
const streamKeys = options.streamKeys ?? new Set<string>();
|
||||
const skipKeys = options.skipKeys ?? new Set<string>();
|
||||
const document: Record<string, unknown> = {};
|
||||
const streamedCounts: Record<string, number> = {};
|
||||
|
||||
let phase: Phase = "start";
|
||||
let inString = false;
|
||||
let escaped = false;
|
||||
// Nesting depth inside the current value or element (0 = not inside one).
|
||||
let depth = 0;
|
||||
let key = "";
|
||||
let skipping = false;
|
||||
// Inside a key string.
|
||||
let keyOpen = false;
|
||||
// Inside a streamed array, after an element: the next byte must be `,` or `]`.
|
||||
let elementNeedsComma = false;
|
||||
const keyBytes = new Collector();
|
||||
const value = new Collector();
|
||||
|
||||
const finishValue = () => {
|
||||
const text = value.take();
|
||||
if (skipping) return;
|
||||
try {
|
||||
document[key] = JSON.parse(text);
|
||||
} catch {
|
||||
throw new BackupJsonError(`Backup member '${key}' is not valid JSON`);
|
||||
}
|
||||
};
|
||||
const finishElement = async () => {
|
||||
const text = value.take();
|
||||
let element: unknown;
|
||||
try {
|
||||
element = JSON.parse(text);
|
||||
} catch {
|
||||
throw new BackupJsonError(
|
||||
`Element ${streamedCounts[key]} of backup member '${key}' is not valid JSON`,
|
||||
);
|
||||
}
|
||||
const index = streamedCounts[key]!;
|
||||
streamedCounts[key] = index + 1;
|
||||
await options.onElement?.(key, element, index);
|
||||
};
|
||||
|
||||
for await (const raw of chunks) {
|
||||
const chunk = Buffer.isBuffer(raw)
|
||||
? raw
|
||||
: Buffer.from(raw.buffer, raw.byteOffset, raw.byteLength);
|
||||
// Where the value or key being collected started in this chunk.
|
||||
let mark = 0;
|
||||
for (let i = 0; i < chunk.length; i++) {
|
||||
const byte = chunk[i]!;
|
||||
|
||||
if (phase === "in-value" || phase === "in-element") {
|
||||
if (inString) {
|
||||
if (escaped) escaped = false;
|
||||
else if (byte === BACKSLASH) escaped = true;
|
||||
else if (byte === QUOTE) inString = false;
|
||||
continue;
|
||||
}
|
||||
if (byte === QUOTE) {
|
||||
inString = true;
|
||||
continue;
|
||||
}
|
||||
if (byte === OPEN_BRACE || byte === OPEN_BRACKET) {
|
||||
depth += 1;
|
||||
continue;
|
||||
}
|
||||
if (byte === CLOSE_BRACE || byte === CLOSE_BRACKET) {
|
||||
if (depth > 0) {
|
||||
depth -= 1;
|
||||
if (depth > 0) continue;
|
||||
// The value's own closing bracket: it belongs to the value.
|
||||
value.add(chunk, mark, i + 1);
|
||||
if (phase === "in-value") {
|
||||
finishValue();
|
||||
phase = "after";
|
||||
} else {
|
||||
await finishElement();
|
||||
phase = "element";
|
||||
elementNeedsComma = true;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
// A primitive value ended by the enclosing bracket.
|
||||
value.add(chunk, mark, i);
|
||||
if (phase === "in-value") {
|
||||
if (byte !== CLOSE_BRACE) {
|
||||
throw new BackupJsonError("Unbalanced bracket in backup");
|
||||
}
|
||||
finishValue();
|
||||
phase = "done";
|
||||
} else {
|
||||
if (byte !== CLOSE_BRACKET) {
|
||||
throw new BackupJsonError("Unbalanced bracket in backup");
|
||||
}
|
||||
await finishElement();
|
||||
phase = "after";
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if (depth === 0 && (byte === COMMA || isWhitespace(byte))) {
|
||||
// A primitive value ended.
|
||||
value.add(chunk, mark, i);
|
||||
if (phase === "in-value") {
|
||||
finishValue();
|
||||
phase = byte === COMMA ? "key" : "after";
|
||||
} else {
|
||||
await finishElement();
|
||||
phase = "element";
|
||||
elementNeedsComma = byte !== COMMA;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
if (phase === "key" && keyOpen) {
|
||||
if (escaped) escaped = false;
|
||||
else if (byte === BACKSLASH) escaped = true;
|
||||
else if (byte === QUOTE) {
|
||||
keyBytes.add(chunk, mark, i + 1);
|
||||
key = JSON.parse(keyBytes.take()) as string;
|
||||
keyOpen = false;
|
||||
phase = "colon";
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
if (isWhitespace(byte)) continue;
|
||||
|
||||
switch (phase) {
|
||||
case "start":
|
||||
if (byte !== OPEN_BRACE) {
|
||||
throw new BackupJsonError("Backup is not a JSON object");
|
||||
}
|
||||
phase = "key";
|
||||
break;
|
||||
case "key":
|
||||
if (byte === CLOSE_BRACE) {
|
||||
phase = "done";
|
||||
} else if (byte === QUOTE) {
|
||||
keyOpen = true;
|
||||
escaped = false;
|
||||
mark = i;
|
||||
} else {
|
||||
throw new BackupJsonError("Malformed backup: expected a key");
|
||||
}
|
||||
break;
|
||||
case "colon":
|
||||
if (byte !== COLON) {
|
||||
throw new BackupJsonError("Malformed backup: expected ':'");
|
||||
}
|
||||
phase = streamKeys.has(key) ? "array-start" : "value";
|
||||
skipping = skipKeys.has(key);
|
||||
break;
|
||||
case "value":
|
||||
phase = "in-value";
|
||||
mark = i;
|
||||
depth = 0;
|
||||
inString = false;
|
||||
escaped = false;
|
||||
// Re-read this byte as the first byte of the value.
|
||||
i -= 1;
|
||||
break;
|
||||
case "array-start":
|
||||
if (byte !== OPEN_BRACKET) {
|
||||
throw new BackupJsonError(
|
||||
`Backup member '${key}' must be an array`,
|
||||
);
|
||||
}
|
||||
streamedCounts[key] = 0;
|
||||
document[key] = [];
|
||||
phase = "element";
|
||||
elementNeedsComma = false;
|
||||
break;
|
||||
case "element":
|
||||
if (byte === CLOSE_BRACKET) {
|
||||
phase = "after";
|
||||
} else if (byte === COMMA && elementNeedsComma) {
|
||||
elementNeedsComma = false;
|
||||
} else if (!elementNeedsComma) {
|
||||
phase = "in-element";
|
||||
mark = i;
|
||||
depth = 0;
|
||||
inString = false;
|
||||
escaped = false;
|
||||
i -= 1;
|
||||
} else {
|
||||
throw new BackupJsonError(
|
||||
`Malformed backup: expected ',' in '${key}'`,
|
||||
);
|
||||
}
|
||||
break;
|
||||
case "after":
|
||||
if (byte === COMMA) {
|
||||
phase = "key";
|
||||
} else if (byte === CLOSE_BRACE) {
|
||||
phase = "done";
|
||||
} else {
|
||||
throw new BackupJsonError("Malformed backup: expected ',' or '}'");
|
||||
}
|
||||
break;
|
||||
case "done":
|
||||
throw new BackupJsonError("Unexpected content after the backup");
|
||||
}
|
||||
}
|
||||
// Carry the unfinished key or value into the next chunk.
|
||||
if (phase === "in-value" || phase === "in-element") {
|
||||
value.add(chunk, mark, chunk.length);
|
||||
} else if (phase === "key" && keyOpen) {
|
||||
keyBytes.add(chunk, mark, chunk.length);
|
||||
}
|
||||
}
|
||||
|
||||
// Anything but the closing brace as the last structural byte is a
|
||||
// truncated file, including one that ends inside a member value.
|
||||
if ((phase as Phase) !== "done") {
|
||||
throw new BackupJsonError("Backup ends before its closing brace");
|
||||
}
|
||||
return { document, streamedCounts };
|
||||
}
|
||||
@@ -0,0 +1,136 @@
|
||||
/**
|
||||
* Insert many measurements in one statement, as column arrays.
|
||||
*
|
||||
* Why not `createMany`. Prisma compiles and caches a query plan per query
|
||||
* shape, and a `createMany` of N rows is its own shape: measured, one
|
||||
* 1 000-row `createMany` of measurements leaves about 145 MB of retained
|
||||
* heap behind, a 2 000-row one about 290 MB, and the smaller last batch of a
|
||||
* restore adds its own plan on top. In a 1 GB container that alone took a
|
||||
* restore of 1.25 million measurements past the heap limit (#1031). One
|
||||
* `INSERT … SELECT FROM unnest(…)` with an array per column is a single small
|
||||
* statement whatever the batch size, and about four times faster.
|
||||
*
|
||||
* `COLUMNS` maps every field of `MeasurementInsertRow` to its column and
|
||||
* type. It is typed as a complete record over the row's keys, so a field
|
||||
* added to the row without a column here does not compile, and the restore
|
||||
* builds its rows as `MeasurementInsertRow`, so a field it starts writing has
|
||||
* to be added to the row first.
|
||||
*/
|
||||
import type { Prisma } from "@/generated/prisma/client";
|
||||
|
||||
/** One measurement as the restore writes it. */
|
||||
export interface MeasurementInsertRow {
|
||||
id: string;
|
||||
userId: string;
|
||||
type: string;
|
||||
value: number;
|
||||
valueMin: number | null;
|
||||
valueMax: number | null;
|
||||
unit: string;
|
||||
source: string;
|
||||
measuredAt: Date;
|
||||
notes: string | null;
|
||||
notesEncrypted: Uint8Array | null;
|
||||
externalId: string | null;
|
||||
externalSourceVersion: string | null;
|
||||
aggregationProvenance: string | null;
|
||||
glucoseContext: string | null;
|
||||
sleepStage: string | null;
|
||||
rhythmClassification: string | null;
|
||||
deviceType: string | null;
|
||||
syncVersion: number;
|
||||
deletedAt: Date | null;
|
||||
createdAt?: Date;
|
||||
updatedAt?: Date;
|
||||
}
|
||||
|
||||
type Kind =
|
||||
"text" | "float8" | "int4" | "timestamp" | "bytea" | `enum:${string}`;
|
||||
|
||||
/** Column and element type per field. Complete by construction. */
|
||||
const COLUMNS: { [K in keyof MeasurementInsertRow]-?: [string, Kind] } = {
|
||||
id: ["id", "text"],
|
||||
userId: ["user_id", "text"],
|
||||
type: ["type", "enum:measurement_type"],
|
||||
value: ["value", "float8"],
|
||||
valueMin: ["value_min", "float8"],
|
||||
valueMax: ["value_max", "float8"],
|
||||
unit: ["unit", "text"],
|
||||
source: ["source", "enum:measurement_source"],
|
||||
measuredAt: ["measured_at", "timestamp"],
|
||||
notes: ["notes", "text"],
|
||||
notesEncrypted: ["notes_encrypted", "bytea"],
|
||||
externalId: ["external_id", "text"],
|
||||
externalSourceVersion: ["external_source_version", "text"],
|
||||
aggregationProvenance: [
|
||||
"aggregation_provenance",
|
||||
"enum:measurement_aggregation_provenance",
|
||||
],
|
||||
glucoseContext: ["glucose_context", "enum:glucose_context"],
|
||||
sleepStage: ["sleep_stage", "enum:sleep_stage"],
|
||||
rhythmClassification: ["rhythm_classification", "enum:rhythm_classification"],
|
||||
deviceType: ["device_type", "text"],
|
||||
syncVersion: ["sync_version", "int4"],
|
||||
deletedAt: ["deleted_at", "timestamp"],
|
||||
createdAt: ["created_at", "timestamp"],
|
||||
updatedAt: ["updated_at", "timestamp"],
|
||||
};
|
||||
|
||||
const FIELDS = Object.keys(COLUMNS) as Array<keyof MeasurementInsertRow>;
|
||||
|
||||
/** How each array travels, and how its element is turned back into a value. */
|
||||
function arrayType(kind: Kind): string {
|
||||
if (kind === "float8" || kind === "int4") return `${kind}[]`;
|
||||
return "text[]"; // timestamps as ISO text, bytes as hex, enums as labels
|
||||
}
|
||||
|
||||
function selectExpression(
|
||||
field: keyof MeasurementInsertRow,
|
||||
kind: Kind,
|
||||
): string {
|
||||
const ref = `u."${field}"`;
|
||||
if (kind === "timestamp") {
|
||||
// Stored as UTC in `timestamp(3)`. ISO text with a `Z` casts to the same
|
||||
// wall-clock instant; the zone suffix is ignored by the cast.
|
||||
const cast = `${ref}::timestamp(3)`;
|
||||
// Prisma fills both on create when the row does not carry them.
|
||||
return field === "createdAt" || field === "updatedAt"
|
||||
? `COALESCE(${cast}, CURRENT_TIMESTAMP)`
|
||||
: cast;
|
||||
}
|
||||
if (kind === "bytea") return `decode(${ref}, 'hex')`;
|
||||
if (kind.startsWith("enum:")) return `${ref}::${kind.slice(5)}`;
|
||||
return ref;
|
||||
}
|
||||
|
||||
/**
|
||||
* The statement, built once. Its text depends on nothing but `COLUMNS`, a
|
||||
* compile-time table: no input reaches it, and every value travels as a bound
|
||||
* array parameter.
|
||||
*/
|
||||
export const MEASUREMENT_BULK_INSERT_SQL = `
|
||||
INSERT INTO measurements (${FIELDS.map((f) => COLUMNS[f][0]).join(", ")})
|
||||
SELECT ${FIELDS.map((f) => selectExpression(f, COLUMNS[f][1])).join(", ")}
|
||||
FROM unnest(${FIELDS.map((f, i) => `$${i + 1}::${arrayType(COLUMNS[f][1])}`).join(", ")})
|
||||
AS u(${FIELDS.map((f) => `"${f}"`).join(", ")})
|
||||
`;
|
||||
|
||||
function cell(value: unknown, kind: Kind): unknown {
|
||||
if (value === null || value === undefined) return null;
|
||||
if (kind === "timestamp") return (value as Date).toISOString();
|
||||
if (kind === "bytea") return Buffer.from(value as Uint8Array).toString("hex");
|
||||
return value;
|
||||
}
|
||||
|
||||
/** Insert `rows` in one statement. Resolves to the number inserted. */
|
||||
export async function insertMeasurementRows(
|
||||
tx: Prisma.TransactionClient,
|
||||
rows: readonly MeasurementInsertRow[],
|
||||
): Promise<number> {
|
||||
if (rows.length === 0) return 0;
|
||||
const arrays = FIELDS.map((field) => {
|
||||
const kind = COLUMNS[field][1];
|
||||
return rows.map((row) => cell(row[field], kind));
|
||||
});
|
||||
return tx.$executeRawUnsafe(MEASUREMENT_BULK_INSERT_SQL, ...arrays);
|
||||
}
|
||||
@@ -39,6 +39,8 @@
|
||||
* hashing params so tuning one never silently changes the other.
|
||||
*/
|
||||
import { hashRaw, type Algorithm } from "@node-rs/argon2";
|
||||
import { once } from "node:events";
|
||||
import { createWriteStream } from "node:fs";
|
||||
import {
|
||||
createCipheriv,
|
||||
createDecipheriv,
|
||||
@@ -128,6 +130,11 @@ export async function encryptArchive(
|
||||
const ciphertext = Buffer.concat([cipher.update(body), cipher.final()]);
|
||||
const tag = cipher.getAuthTag();
|
||||
|
||||
return Buffer.concat([archiveHeader(), salt, iv, tag, ciphertext]);
|
||||
}
|
||||
|
||||
/** The fixed `HLX1` header for the current KDF parameters. */
|
||||
function archiveHeader(): Buffer {
|
||||
const header = Buffer.alloc(4 + 1 + 1 + 4 + 4 + 1 + 1);
|
||||
header.write(MAGIC, 0, "ascii");
|
||||
header.writeUInt8(VERSION, 4);
|
||||
@@ -136,8 +143,88 @@ export async function encryptArchive(
|
||||
header.writeUInt32BE(EXPORT_ARGON2_PARAMS.timeCost, 10);
|
||||
header.writeUInt8(EXPORT_ARGON2_PARAMS.parallelism, 14);
|
||||
header.writeUInt8(SALT_LENGTH, 15);
|
||||
return header;
|
||||
}
|
||||
|
||||
return Buffer.concat([header, salt, iv, tag, ciphertext]);
|
||||
/** Where `encryptArchiveToFile` left an archive, in two parts. */
|
||||
export interface SpooledArchive {
|
||||
/** Header, salt, IV and tag: everything in front of the ciphertext. */
|
||||
prefix: Buffer;
|
||||
/** File holding the ciphertext, which follows the prefix. */
|
||||
bodyPath: string;
|
||||
/** Size of the whole archive in bytes. */
|
||||
byteLength: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Seal a JSON document produced in pieces into an `HLX1` archive without
|
||||
* holding it: the ciphertext goes to `bodyPath` as it is produced.
|
||||
*
|
||||
* Why a file. The format puts the GCM tag in front of the ciphertext, and the
|
||||
* tag exists only once the last byte has been encrypted, so the archive
|
||||
* cannot be sent as it is written. Holding it in memory instead is what the
|
||||
* route did, and for an account of 1.25 million measurements that did not fit
|
||||
* a 1 GB container (#1031). Spooling the ciphertext keeps the format, so every
|
||||
* archive already on someone's disk still opens the same way. Only ciphertext
|
||||
* ever touches the disk; the caller deletes the file once it is sent.
|
||||
*
|
||||
* Decrypts with `decryptArchive` exactly like an archive `encryptArchive`
|
||||
* wrote; the tests hold the two to that.
|
||||
*/
|
||||
export async function encryptArchiveToFile(
|
||||
producer: (
|
||||
write: (chunk: string | Buffer) => Promise<void>,
|
||||
) => Promise<unknown>,
|
||||
passphrase: string,
|
||||
bodyPath: string,
|
||||
): Promise<SpooledArchive> {
|
||||
if (passphrase.length < MIN_EXPORT_PASSPHRASE_LENGTH) {
|
||||
throw new Error(
|
||||
`Passphrase must be at least ${MIN_EXPORT_PASSPHRASE_LENGTH} characters`,
|
||||
);
|
||||
}
|
||||
const salt = randomBytes(SALT_LENGTH);
|
||||
const key = await deriveKey(passphrase, EXPORT_ARGON2_PARAMS, salt);
|
||||
const iv = randomBytes(IV_LENGTH);
|
||||
const cipher = createCipheriv(ALGORITHM, key, iv);
|
||||
const file = createWriteStream(bodyPath, { mode: 0o600 });
|
||||
let bodyBytes = 0;
|
||||
const fileError = new Promise<never>((_, reject) => {
|
||||
file.once("error", reject);
|
||||
});
|
||||
fileError.catch(() => {});
|
||||
|
||||
const put = async (bytes: Buffer): Promise<void> => {
|
||||
if (bytes.length === 0) return;
|
||||
bodyBytes += bytes.length;
|
||||
if (!file.write(bytes)) {
|
||||
await Promise.race([once(file, "drain"), fileError]);
|
||||
}
|
||||
};
|
||||
|
||||
try {
|
||||
await producer(async (chunk) => {
|
||||
await put(
|
||||
cipher.update(
|
||||
typeof chunk === "string" ? Buffer.from(chunk, "utf8") : chunk,
|
||||
),
|
||||
);
|
||||
});
|
||||
await put(cipher.final());
|
||||
file.end();
|
||||
await Promise.race([once(file, "finish"), fileError]);
|
||||
} catch (err) {
|
||||
file.destroy();
|
||||
throw err;
|
||||
}
|
||||
|
||||
const prefix = Buffer.concat([
|
||||
archiveHeader(),
|
||||
salt,
|
||||
iv,
|
||||
cipher.getAuthTag(),
|
||||
]);
|
||||
return { prefix, bodyPath, byteLength: prefix.length + bodyBytes };
|
||||
}
|
||||
|
||||
/** Parse + validate the fixed `HLX1` header. Throws on any structural fault. */
|
||||
|
||||
@@ -0,0 +1,82 @@
|
||||
/**
|
||||
* Turn a producer of text or bytes into a response body, with backpressure
|
||||
* and cancellation.
|
||||
*
|
||||
* The export routes used to build their whole file as one string and hand it
|
||||
* to the response. For an account of 1.25 million measurements that string
|
||||
* and the object graph behind it did not fit a 1 GB container: one export
|
||||
* took the app down for every user of the instance (#1031). The file is now
|
||||
* written into the response as it is produced.
|
||||
*
|
||||
* Backpressure: the producer's `write` waits while the client is slower than
|
||||
* the database, so nothing piles up in between. Cancellation: when the client
|
||||
* goes away, the next `write` throws `ResponseStreamClosedError`, which stops
|
||||
* the producer instead of leaving it to finish a file nobody reads.
|
||||
*/
|
||||
import { PassThrough, Readable } from "node:stream";
|
||||
|
||||
/** The client closed the response before the producer finished. */
|
||||
export class ResponseStreamClosedError extends Error {
|
||||
constructor() {
|
||||
super("The response was closed before the file was complete");
|
||||
this.name = "ResponseStreamClosedError";
|
||||
}
|
||||
}
|
||||
|
||||
export type ResponseProducer = (
|
||||
write: (chunk: string | Buffer) => Promise<void>,
|
||||
) => Promise<unknown>;
|
||||
|
||||
export interface ResponseStreamHooks {
|
||||
/** Runs once the producer has written everything. */
|
||||
onComplete?: () => unknown;
|
||||
/**
|
||||
* Runs when the producer failed or the client left. The response has
|
||||
* already started, so the client receives a truncated file; a truncated
|
||||
* JSON file does not parse, and a truncated archive fails its tag.
|
||||
*/
|
||||
onError?: (err: unknown) => unknown;
|
||||
}
|
||||
|
||||
/** Start `producer` and return the web stream its output flows into. */
|
||||
export function streamToResponseBody(
|
||||
producer: ResponseProducer,
|
||||
hooks: ResponseStreamHooks = {},
|
||||
): ReadableStream<Uint8Array> {
|
||||
const out = new PassThrough();
|
||||
|
||||
const write = async (chunk: string | Buffer): Promise<void> => {
|
||||
if (out.destroyed || out.writableEnded)
|
||||
throw new ResponseStreamClosedError();
|
||||
if (out.write(chunk)) return;
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const onDrain = () => {
|
||||
out.off("close", onClose);
|
||||
resolve();
|
||||
};
|
||||
const onClose = () => {
|
||||
out.off("drain", onDrain);
|
||||
reject(new ResponseStreamClosedError());
|
||||
};
|
||||
out.once("drain", onDrain);
|
||||
out.once("close", onClose);
|
||||
});
|
||||
};
|
||||
|
||||
void (async () => {
|
||||
try {
|
||||
await producer(write);
|
||||
out.end();
|
||||
await hooks.onComplete?.();
|
||||
} catch (err) {
|
||||
out.destroy(err instanceof Error ? err : new Error(String(err)));
|
||||
try {
|
||||
await hooks.onError?.(err);
|
||||
} catch {
|
||||
// Reporting a failure must not raise another one.
|
||||
}
|
||||
}
|
||||
})();
|
||||
|
||||
return Readable.toWeb(out) as ReadableStream<Uint8Array>;
|
||||
}
|
||||
@@ -57,14 +57,18 @@ export interface StoreBackupBlobInput {
|
||||
|
||||
/**
|
||||
* Pack `producer`'s JSON into the stored envelope and upsert it as the
|
||||
* account's `(userId, type)` backup. Resolves to the stored size in bytes.
|
||||
* `(userId, type)` backup. Resolves to the row's id and the stored size.
|
||||
*
|
||||
* `input` may be a function, read once the producer has finished: an
|
||||
* uploaded file names its owner somewhere inside itself, and the upload
|
||||
* route only knows who that is once it has read the file through.
|
||||
*/
|
||||
export async function storeBackupBlob(
|
||||
prisma: PrismaClient,
|
||||
input: StoreBackupBlobInput,
|
||||
input: StoreBackupBlobInput | (() => StoreBackupBlobInput),
|
||||
producer: BackupJsonProducer,
|
||||
options: PackBackupBlobOptions = {},
|
||||
): Promise<number> {
|
||||
): Promise<{ id: string; bytes: number }> {
|
||||
return prisma.$transaction(
|
||||
async (tx) => {
|
||||
await tx.$executeRawUnsafe(
|
||||
@@ -91,22 +95,24 @@ export async function storeBackupBlob(
|
||||
options,
|
||||
);
|
||||
|
||||
const target = typeof input === "function" ? input() : input;
|
||||
// The row first, through Prisma, so a new account's backup gets its id
|
||||
// the same way every other row does; the data it briefly carries is
|
||||
// never visible outside this transaction.
|
||||
await tx.dataBackup.upsert({
|
||||
where: { userId_type: { userId: input.userId, type: input.type } },
|
||||
const row = await tx.dataBackup.upsert({
|
||||
where: { userId_type: { userId: target.userId, type: target.type } },
|
||||
update: { createdAt: new Date() },
|
||||
create: { userId: input.userId, type: input.type, data: "" },
|
||||
create: { userId: target.userId, type: target.type, data: "" },
|
||||
select: { id: true },
|
||||
});
|
||||
await tx.$executeRaw`
|
||||
UPDATE data_backups
|
||||
SET data = (
|
||||
SELECT string_agg(piece, '' ORDER BY seq) FROM backup_blob_parts
|
||||
)
|
||||
WHERE user_id = ${input.userId} AND type = ${input.type}
|
||||
WHERE id = ${row.id}
|
||||
`;
|
||||
return bytes;
|
||||
return { id: row.id, bytes };
|
||||
},
|
||||
{ timeout: STORE_TRANSACTION_TIMEOUT_MS, maxWait: 60_000 },
|
||||
);
|
||||
|
||||
@@ -0,0 +1,138 @@
|
||||
/**
|
||||
* A backup file read as a stream: every section but the measurements in
|
||||
* memory, the measurements handed over in batches.
|
||||
*
|
||||
* Why. The measurements are the one section that grows with a record without
|
||||
* bound. For an account with 1.25 million of them the disaster-recovery JSON
|
||||
* is 662 MB, which cannot exist as one JavaScript string, and the parsed rows
|
||||
* alone would not fit a 1 GB container beside the running app (#1031). Every
|
||||
* other section of the same file is a few megabytes.
|
||||
*
|
||||
* So the file is read twice. The first read keeps everything except the
|
||||
* measurements, and checks each measurement against the same element schema
|
||||
* the whole-file schema uses, counting them. Nothing is written on the
|
||||
* strength of an unchecked row: a restore reads the file this way before it
|
||||
* deletes anything. The second read, from `forEachMeasurementBatch`, hands
|
||||
* the measurements over in order, in batches, for the writer.
|
||||
*
|
||||
* `raw` is the file exactly as `JSON.parse` would give it, with
|
||||
* `measurements` replaced by an empty array when the file has that section
|
||||
* (and still absent when it does not, which is what the section-presence
|
||||
* check reads). Run it through `parseBackupPayload` for the rest of the
|
||||
* validation; together that is the whole-file check.
|
||||
*/
|
||||
import { scanBackupJson } from "@/lib/export/backup-json-scan";
|
||||
import {
|
||||
BACKUP_SCHEMA_VERSION,
|
||||
backupMeasurementSchema,
|
||||
type BackupMeasurement,
|
||||
} from "@/lib/validations/backup";
|
||||
|
||||
/** The file's bytes, from the start, every time it is called. */
|
||||
export type BackupSource = () => AsyncIterable<Uint8Array>;
|
||||
|
||||
const MEASUREMENTS = "measurements";
|
||||
const STREAMED = new Set([MEASUREMENTS]);
|
||||
|
||||
/** A measurement the element schema refused, with where it is in the file. */
|
||||
export class StreamedBackupInvalidError extends Error {
|
||||
readonly path: string;
|
||||
constructor(path: string, message: string) {
|
||||
super(`${path}: ${message}`);
|
||||
this.name = "StreamedBackupInvalidError";
|
||||
this.path = path;
|
||||
}
|
||||
}
|
||||
|
||||
export interface StreamedBackup {
|
||||
/** The file without its measurements (see the module comment). */
|
||||
raw: Record<string, unknown>;
|
||||
/** How many measurements the file carries. */
|
||||
measurementCount: number;
|
||||
/** The file's size in bytes. */
|
||||
bytes: number;
|
||||
/**
|
||||
* Read the file again and hand its measurements over in file order, in
|
||||
* batches of at most `size`, each row parsed by the element schema.
|
||||
* Awaits `fn` before reading further.
|
||||
*/
|
||||
forEachMeasurementBatch(
|
||||
size: number,
|
||||
fn: (batch: BackupMeasurement[]) => Promise<void>,
|
||||
): Promise<void>;
|
||||
}
|
||||
|
||||
function parseMeasurement(element: unknown, index: number): BackupMeasurement {
|
||||
const parsed = backupMeasurementSchema.safeParse(element);
|
||||
if (!parsed.success) {
|
||||
const issue = parsed.error.issues[0];
|
||||
throw new StreamedBackupInvalidError(
|
||||
[MEASUREMENTS, index, ...(issue?.path ?? [])].join("."),
|
||||
issue?.message ?? "invalid measurement",
|
||||
);
|
||||
}
|
||||
return parsed.data;
|
||||
}
|
||||
|
||||
/**
|
||||
* First read of a backup file. Rejects with `BackupJsonError` for input that
|
||||
* is not a JSON object and with `StreamedBackupInvalidError` for a
|
||||
* measurement the schema refuses.
|
||||
*/
|
||||
export async function readStreamedBackup(
|
||||
source: BackupSource,
|
||||
): Promise<StreamedBackup> {
|
||||
let firstWithoutId: number | null = null;
|
||||
let bytes = 0;
|
||||
async function* counted() {
|
||||
for await (const chunk of source()) {
|
||||
bytes += chunk.byteLength;
|
||||
yield chunk;
|
||||
}
|
||||
}
|
||||
const { document, streamedCounts } = await scanBackupJson(counted(), {
|
||||
streamKeys: STREAMED,
|
||||
onElement: (_key, element, index) => {
|
||||
const row = parseMeasurement(element, index);
|
||||
if (!row.id && firstWithoutId === null) firstWithoutId = index;
|
||||
},
|
||||
});
|
||||
|
||||
// The whole-file schema's cross-row rule, applied the same way: a canonical
|
||||
// v2 file must give every measurement a stable id.
|
||||
if (
|
||||
document.schemaVersion === BACKUP_SCHEMA_VERSION &&
|
||||
firstWithoutId !== null
|
||||
) {
|
||||
throw new StreamedBackupInvalidError(
|
||||
`${MEASUREMENTS}.${firstWithoutId}.id`,
|
||||
"Canonical v2 measurements require a stable id",
|
||||
);
|
||||
}
|
||||
|
||||
const skipKeys = new Set(
|
||||
Object.keys(document).filter((key) => key !== MEASUREMENTS),
|
||||
);
|
||||
|
||||
return {
|
||||
raw: document,
|
||||
measurementCount: streamedCounts[MEASUREMENTS] ?? 0,
|
||||
bytes,
|
||||
async forEachMeasurementBatch(size, fn) {
|
||||
let batch: BackupMeasurement[] = [];
|
||||
await scanBackupJson(source(), {
|
||||
streamKeys: STREAMED,
|
||||
skipKeys,
|
||||
onElement: async (_key, element, index) => {
|
||||
batch.push(parseMeasurement(element, index));
|
||||
if (batch.length >= size) {
|
||||
const full = batch;
|
||||
batch = [];
|
||||
await fn(full);
|
||||
}
|
||||
},
|
||||
});
|
||||
if (batch.length > 0) await fn(batch);
|
||||
},
|
||||
};
|
||||
}
|
||||
Reference in new issue
Block a user