diff --git a/messages/de.json b/messages/de.json index 3af8f13ca..a95ac2bc6 100644 --- a/messages/de.json +++ b/messages/de.json @@ -7167,7 +7167,7 @@ "uploadTitle": "Backup hochladen", "uploadDescription": "Eine Backup-Datei validieren und als manuellen Snapshot speichern. Das Wiederherstellen ist ein separater Schritt.", "uploadButton": "Datei hochladen", - "uploadHelp": "JSON-Datei passend zum aktuellen Backup-Schema. Max. 10 MB.", + "uploadHelp": "HealthLog-Backup als .json oder .json.gz. Die Datei wird vor dem Hochladen komprimiert; komprimiert darf sie bis zu 512 MB groß sein.", "uploadSuccess": "Backup hochgeladen ({count} Datensätze)", "uploadFailed": "Upload fehlgeschlagen", "restore": "Wiederherstellen", diff --git a/messages/en.json b/messages/en.json index 9cd82bd03..ecbe39876 100644 --- a/messages/en.json +++ b/messages/en.json @@ -7167,7 +7167,7 @@ "uploadTitle": "Upload backup", "uploadDescription": "Validate a backup file and store it as a manual snapshot. Restoring is a separate step.", "uploadButton": "Upload file", - "uploadHelp": "JSON file matching the current backup schema. Max 10 MB.", + "uploadHelp": "A HealthLog backup as .json or .json.gz. The file is compressed before upload; compressed, it may be up to 512 MB.", "uploadSuccess": "Backup uploaded ({count} records)", "uploadFailed": "Upload failed", "restore": "Restore", diff --git a/messages/es.json b/messages/es.json index b69c2d44f..5efbc6d9e 100644 --- a/messages/es.json +++ b/messages/es.json @@ -7167,7 +7167,7 @@ "uploadTitle": "Subir copia de seguridad", "uploadDescription": "Valida un archivo de copia de seguridad y guárdalo como instantánea manual. La restauración es un paso aparte.", "uploadButton": "Subir archivo", - "uploadHelp": "Archivo JSON que coincida con el esquema de copia de seguridad actual. Máx. 10 MB.", + "uploadHelp": "Una copia de seguridad de HealthLog en .json o .json.gz. El archivo se comprime antes de subirlo; comprimido puede ocupar hasta 512 MB.", "uploadSuccess": "Copia de seguridad subida ({count} registros)", "uploadFailed": "La subida falló", "restore": "Restaurar", diff --git a/messages/fr.json b/messages/fr.json index e832e094e..6d7ae95e6 100644 --- a/messages/fr.json +++ b/messages/fr.json @@ -7167,7 +7167,7 @@ "uploadTitle": "Téléverser une sauvegarde", "uploadDescription": "Validez un fichier de sauvegarde et stockez-le comme instantané manuel. La restauration est une étape distincte.", "uploadButton": "Téléverser un fichier", - "uploadHelp": "Fichier JSON correspondant au schéma de sauvegarde actuel. Max 10 Mo.", + "uploadHelp": "Une sauvegarde HealthLog au format .json ou .json.gz. Le fichier est compressé avant l’envoi ; compressé, il peut atteindre 512 Mo.", "uploadSuccess": "Sauvegarde téléversée ({count} enregistrements)", "uploadFailed": "Échec du téléversement", "restore": "Restaurer", diff --git a/messages/it.json b/messages/it.json index 237f44644..bb6e9b094 100644 --- a/messages/it.json +++ b/messages/it.json @@ -7167,7 +7167,7 @@ "uploadTitle": "Carica backup", "uploadDescription": "Convalida un file di backup e archivialo come istantanea manuale. Il ripristino è un passaggio separato.", "uploadButton": "Carica file", - "uploadHelp": "File JSON conforme allo schema di backup attuale. Max 10 MB.", + "uploadHelp": "Un backup di HealthLog in formato .json o .json.gz. Il file viene compresso prima del caricamento; compresso può arrivare a 512 MB.", "uploadSuccess": "Backup caricato ({count} record)", "uploadFailed": "Caricamento non riuscito", "restore": "Ripristina", diff --git a/messages/ko.json b/messages/ko.json index 91b12f477..fcb2faf95 100644 --- a/messages/ko.json +++ b/messages/ko.json @@ -7146,7 +7146,7 @@ "uploadTitle": "백업 업로드", "uploadDescription": "백업 파일을 검사한 뒤 수동 스냅샷으로 저장해요. 복원은 별도 단계예요.", "uploadButton": "파일 업로드", - "uploadHelp": "현재 백업 스키마에 맞는 JSON 파일이에요. 최대 10 MB.", + "uploadHelp": "HealthLog 백업 파일(.json 또는 .json.gz)이에요. 업로드 전에 압축되며, 압축된 크기는 최대 512 MB까지 가능해요.", "uploadSuccess": "백업을 업로드했어요 (기록 {count}개)", "uploadFailed": "업로드하지 못했어요", "restore": "복원", diff --git a/messages/pl.json b/messages/pl.json index ad7f00f51..de7694e66 100644 --- a/messages/pl.json +++ b/messages/pl.json @@ -7167,7 +7167,7 @@ "uploadTitle": "Prześlij kopię zapasową", "uploadDescription": "Sprawdź plik kopii zapasowej i zapisz go jako ręczną migawkę. Przywracanie to osobny krok.", "uploadButton": "Prześlij plik", - "uploadHelp": "Plik JSON zgodny z bieżącym schematem kopii zapasowej. Maks. 10 MB.", + "uploadHelp": "Kopia zapasowa HealthLog w formacie .json lub .json.gz. Plik jest kompresowany przed wysłaniem; po kompresji może mieć do 512 MB.", "uploadSuccess": "Przesłano kopię zapasową ({count} rekordów)", "uploadFailed": "Przesyłanie nie powiodło się", "restore": "Przywróć", diff --git a/scripts/restore-backup.ts b/scripts/restore-backup.ts index 354da0fff..c6f4a51fe 100644 --- a/scripts/restore-backup.ts +++ b/scripts/restore-backup.ts @@ -15,12 +15,23 @@ * BACKUP_S3_REGION=auto \ * BACKUP_ENCRYPTION_KEY= \ * pnpm dlx tsx scripts/restore-backup.ts 2026-05-08/user-clx123.json.enc /tmp/restored.json + * + * An output name ending in `.gz` writes the JSON compressed, which is the + * form to hand to the admin upload for a large account: the JSON of 1.25 + * million measurements is 662 MB, the compressed file about a tenth of it. + * The file is written as a stream in either case and never held whole. */ -import { writeFileSync } from "node:fs"; +import { createWriteStream } from "node:fs"; +import { rename } from "node:fs/promises"; +import { PassThrough, Readable } from "node:stream"; +import { pipeline } from "node:stream/promises"; +import { createGzip } from "node:zlib"; + +import { scanBackupJson } from "@/lib/export/backup-json-scan"; import { loadOffhostConfig, getS3Client, - decryptBackup, + openBackupObject, OffhostBackupNotConfiguredError, } from "@/lib/jobs/offhost-backup"; @@ -98,28 +109,43 @@ async function main() { `Downloading s3://${cfg.bucket}/${key} from ${cfg.endpoint} (region=${cfg.region})`, ); const ciphertext = await s3.getObject(key); - const plaintext = decryptBackup(ciphertext, cfg.encryptionKey); + const source = openBackupObject(ciphertext, cfg.encryptionKey); - let parsed: unknown; - try { - parsed = JSON.parse(plaintext); - } catch (err) { - throw new Error( - `Decrypted payload is not valid JSON: ${(err as Error).message}`, - ); - } - const body = validateBackupShape(parsed); + // Checked first, in one streamed pass that keeps only the small sections, + // so nothing is written for an object that is not a backup. + const { document, streamedCounts } = await scanBackupJson(source(), { + streamKeys: new Set(["measurements", "intakeEvents", "moodEntries"]), + }); + const body = validateBackupShape({ + ...document, + measurements: [], + intakeEvents: [], + moodEntries: [], + }); if (expectedUserId && body.userId !== expectedUserId) { throw new Error( `Refusing to write: backup.userId='${body.userId}' does not match --user-id='${expectedUserId}'`, ); } - // The restored file is decrypted PHI in plaintext. Write it 0o600 so it is - // not world-readable on the operator's disk while it sits there. - writeFileSync(out, plaintext, { encoding: "utf8", mode: 0o600 }); + // The restored file is decrypted PHI in plaintext. Written 0o600 so it is + // not world-readable on the operator's disk, to a temporary name first so + // an interrupted run never leaves a truncated file under the real one. + const partial = `${out}.part`; + let bytes = 0; + const counter = new PassThrough(); + counter.on("data", (chunk: Buffer) => { + bytes += chunk.length; + }); + await pipeline( + Readable.from(source()), + counter, + ...(out.endsWith(".gz") ? [createGzip()] : []), + createWriteStream(partial, { mode: 0o600 }), + ); + await rename(partial, out); console.log( - `Restored userId=${body.userId} (${plaintext.length} bytes) -> ${out}`, + `Restored userId=${body.userId} (${bytes} bytes of JSON, ${streamedCounts.measurements ?? 0} measurements) -> ${out}`, ); } diff --git a/src/__tests__/backup-readers-stream-guard.test.ts b/src/__tests__/backup-readers-stream-guard.test.ts new file mode 100644 index 000000000..8798c6586 --- /dev/null +++ b/src/__tests__/backup-readers-stream-guard.test.ts @@ -0,0 +1,45 @@ +/** + * Every reader of a stored or uploaded backup reads it as a stream (#1031). + * + * A backup's JSON grows with the record, and past 536 870 888 characters it + * cannot exist as one JavaScript string at all: the disaster-recovery JSON of + * an account with 1.25 million measurements is 662 MB. Each of the readers + * below used to turn the file into one string first, and so none of them + * could open that account's backup. This pins that none of them goes back to + * it: no whole-document decrypt-to-string, no whole-body text read. + */ +import { readFileSync } from "node:fs"; +import { describe, expect, it } from "vitest"; + +const READERS = [ + "src/app/api/admin/backups/[id]/restore/route.ts", + "src/app/api/admin/backups/[id]/summary/route.ts", + "src/app/api/admin/backups/[id]/download/route.ts", + "src/app/api/admin/backups/upload/route.ts", + "src/lib/jobs/restore-drill.ts", + "scripts/restore-backup.ts", +] as const; + +/** Calls that hand back the whole document as one string or parse it whole. */ +const WHOLE_DOCUMENT = [ + /\bunpackBackupBlob\s*\(/, + /\bdecryptBackup\s*\(/, + /\bJSON\.parse\s*\(\s*plaintext\b/, + /\bparseBackupPayload\s*\(\s*plaintext\b/, + /\bfile\.text\s*\(/, + /\bawait\s+request\.text\s*\(\s*\)[\s\S]{0,200}parseBackupPayload/, +]; + +describe("backup readers stream the file", () => { + it.each(READERS)("%s never holds the whole document", (path) => { + const source = readFileSync(path, "utf8"); + for (const pattern of WHOLE_DOCUMENT) { + expect(source, `${path} matches ${pattern}`).not.toMatch(pattern); + } + }); + + it.each(READERS)("%s reads through the streaming reader", (path) => { + const source = readFileSync(path, "utf8"); + expect(source).toMatch(/\b(readStreamedBackup|scanBackupJson)\b/); + }); +}); diff --git a/src/app/api/admin/backups/[id]/download/route.ts b/src/app/api/admin/backups/[id]/download/route.ts index 644313726..bb2a01736 100644 --- a/src/app/api/admin/backups/[id]/download/route.ts +++ b/src/app/api/admin/backups/[id]/download/route.ts @@ -14,6 +14,8 @@ * at-rest contract: the ciphertext stays in the DB, the plaintext is only * materialised inside the request handler and streamed to the admin. */ +import { Readable } from "node:stream"; + import { NextRequest, NextResponse } from "next/server"; import { prisma } from "@/lib/db"; import { apiHandler, HttpError, requireAdmin } from "@/lib/api-handler"; @@ -22,8 +24,13 @@ import { auditLog } from "@/lib/auth/audit"; import { BACKUP_UNDECRYPTABLE_CODE, BACKUP_UNDECRYPTABLE_ERROR, - unpackBackupBlob, + openBackupBlob, } from "@/lib/export/backup-blob"; +import { + readStreamedBackup, + type BackupSource, + type StreamedBackup, +} from "@/lib/export/streamed-backup"; import { annotate } from "@/lib/logging/context"; import { parseBackupPayload } from "@/lib/validations/backup"; @@ -54,9 +61,11 @@ export const GET = apiHandler( throw new HttpError(404, "Backup not found"); } - let plaintext: string; + // Opened and streamed, never unpacked into one string: a large record's + // JSON is longer than any string V8 can hold (#1031). + let source: BackupSource; try { - plaintext = unpackBackupBlob(backup.data); + source = openBackupBlob(backup.data); } catch (err) { // A rotated or dropped key, or a stored copy that is no longer the one // that was written. Both are bad stored input rather than a fault in @@ -80,8 +89,10 @@ export const GET = apiHandler( // pathological case where the worker wrote something the upload // route would later reject — better to fail loudly here than to ship // a junk file that breaks an admin's restore plan. + let streamed: StreamedBackup; try { - parseBackupPayload(plaintext); + streamed = await readStreamedBackup(source); + parseBackupPayload(streamed.raw); } catch (err) { await auditLog("admin.backups.download.failed", { userId: admin.id, @@ -107,22 +118,26 @@ export const GET = apiHandler( ownerId: backup.userId, ownerUsername: backup.user.username, type: backup.type, - sizeBytes: Buffer.byteLength(plaintext, "utf8"), + sizeBytes: streamed.bytes, }, }); - // Returning the plaintext as-is keeps the response a faithful copy of + // Returning the stored bytes as-is keeps the response a faithful copy of // what the worker wrote (formatting, key order). The file is the // canonical artefact admins will store / re-upload. - return new NextResponse(plaintext, { - status: 200, - headers: { - "content-type": "application/json; charset=utf-8", - "content-disposition": `attachment; filename="${filename}"`, - // Backups contain sensitive health data — keep them out of any - // shared cache (CDN, browser disk cache). - "cache-control": "no-store, max-age=0", + return new NextResponse( + Readable.toWeb(Readable.from(source())) as ReadableStream, + { + status: 200, + headers: { + "content-type": "application/json; charset=utf-8", + "content-disposition": `attachment; filename="${filename}"`, + // Backups contain sensitive health data — keep them out of any + // shared cache (CDN, browser disk cache). + "cache-control": "no-store, max-age=0", + "content-length": String(streamed.bytes), + }, }, - }); + ); }, ); diff --git a/src/app/api/admin/backups/[id]/restore/__tests__/owner-mismatch.test.ts b/src/app/api/admin/backups/[id]/restore/__tests__/owner-mismatch.test.ts index 5d52faa1a..70053fb02 100644 --- a/src/app/api/admin/backups/[id]/restore/__tests__/owner-mismatch.test.ts +++ b/src/app/api/admin/backups/[id]/restore/__tests__/owner-mismatch.test.ts @@ -47,6 +47,9 @@ vi.mock("@/lib/validations/backup", () => ({ parseBackupPayload: vi.fn(), isCompatibleSchemaVersion: vi.fn(() => true), summarizeBackup: vi.fn(() => ({})), + // The streamed reader checks measurements one at a time; this file has none. + BACKUP_SCHEMA_VERSION: "2", + backupMeasurementSchema: { safeParse: vi.fn() }, })); vi.mock("@/lib/auth/audit", () => ({ auditLog: vi.fn() })); vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn() })); diff --git a/src/app/api/admin/backups/[id]/restore/route.ts b/src/app/api/admin/backups/[id]/restore/route.ts index 7e45e49f9..7f8f2b4af 100644 --- a/src/app/api/admin/backups/[id]/restore/route.ts +++ b/src/app/api/admin/backups/[id]/restore/route.ts @@ -26,8 +26,17 @@ import { auditLog } from "@/lib/auth/audit"; import { BACKUP_UNDECRYPTABLE_CODE, BACKUP_UNDECRYPTABLE_ERROR, - unpackBackupBlob, + openBackupBlob, } from "@/lib/export/backup-blob"; +import { + insertMeasurementRows, + type MeasurementInsertRow, +} from "@/lib/export/measurement-bulk-insert"; +import { + readStreamedBackup, + type BackupSource, + type StreamedBackup, +} from "@/lib/export/streamed-backup"; import { encryptNote } from "@/lib/crypto/note-cipher"; import { encryptToBytes } from "@/lib/ai/coach/bytes-codec"; import { encryptContextToBytes } from "@/lib/labs/biomarker-store"; @@ -38,6 +47,7 @@ import { parseBackupPayload, isCompatibleSchemaVersion, summarizeBackup, + type BackupMeasurement, type BackupSummary, } from "@/lib/validations/backup"; import { recomputeUserMoodRollups } from "@/lib/rollups/mood-rollups"; @@ -137,6 +147,19 @@ interface RestoreResponse { }; } +/** + * How long the restore transaction may run. Two minutes was the whole budget + * when a record was a few hundred thousand rows; deleting and rewriting 1.25 + * million measurements in one transaction takes longer than that (#1031). On + * the test host the whole restore of that account, reading the file twice + * included, took 94 s. The budget grows by a second per 5 000 measurements, + * which gives that account six minutes, about four times what it needed, and + * never drops below the old two minutes. + */ +function restoreTransactionTimeoutMs(measurementCount: number): number { + return 120_000 + Math.ceil(measurementCount / 5); +} + function decodeEncryptedBytes(encoded: string): Uint8Array { const decoded = Buffer.from(encoded, "base64"); const bytes = new Uint8Array(new ArrayBuffer(decoded.byteLength)); @@ -198,9 +221,12 @@ const handler = apiHandler( throw new HttpError(404, "Backup not found"); } - let plaintext: string; + // Opened, not unpacked: the JSON of a large record is longer than any + // string V8 can hold (#1031), so it is read as a stream below. Opening + // authenticates the whole ciphertext first, as unpacking did. + let source: BackupSource; try { - plaintext = unpackBackupBlob(backup.data); + source = openBackupBlob(backup.data); } catch (err) { await auditLog("admin.backups.restore.failed", { userId: admin.id, @@ -224,10 +250,16 @@ const handler = apiHandler( // Parsed once and kept, because the schema's per-section `.default([])` // erases the difference between a section that is absent and one that is // empty — and that difference is what the completeness check below reads. + // + // The measurements are the exception: `readStreamedBackup` checks each one + // against the element schema as it goes and keeps only their count, and + // the transaction reads them a second time, in batches, as it writes them. let raw: unknown; let payload; + let streamed: StreamedBackup; try { - raw = JSON.parse(plaintext); + streamed = await readStreamedBackup(source); + raw = streamed.raw; payload = parseBackupPayload(raw); } catch (err) { await auditLog("admin.backups.restore.failed", { @@ -548,7 +580,7 @@ const handler = apiHandler( } const toRestoredMeasurementData = ( - measurement: (typeof payload.measurements)[number], + measurement: BackupMeasurement, ) => ({ type: measurement.type, value: measurement.value, @@ -600,55 +632,80 @@ const handler = apiHandler( : {}), }); - const stableRows = payload.measurements.flatMap((measurement) => - measurement.id - ? [ - { - id: measurement.id, - userId: ownerId, - ...toRestoredMeasurementData(measurement), - }, - ] - : [], - ); - const measurementBatchSize = 1_000; - for ( - let offset = 0; - offset < stableRows.length; - offset += measurementBatchSize - ) { - await tx.measurement.createMany({ - data: stableRows.slice(offset, offset + measurementBatchSize), - }); - } - - // v1 payloads did not require stable ids. Preserve their historical - // natural-key reconciliation without routing canonical v2 rows - // through it. - for (const measurement of payload.measurements) { - if (measurement.id) continue; - const restoredData = toRestoredMeasurementData(measurement); - const existing = await tx.measurement.findFirst({ - where: { - userId: ownerId, - type: measurement.type, - source: restoredData.source, - measuredAt: restoredData.measuredAt, - sleepStage: restoredData.sleepStage, - }, - select: { id: true }, - }); - if (existing) { - await tx.measurement.update({ - where: { id: existing.id, userId: ownerId }, - data: restoredData, - }); - } else { - await tx.measurement.create({ - data: { userId: ownerId, ...restoredData }, - }); + // The measurements, a batch at a time from a second read of the + // file, so the restore never holds more than one batch of them. + // + // Two later sections point INTO the measurements (a record's + // `sourceMeasurementId`, an ECG strip's `measurementId`) and have + // to know which of those ids this transaction actually wrote. They + // used to be handed every written id; they only ever ask about the + // ones they reference, so only those are collected. + const referencedMeasurementIds = new Set(); + for (const record of payload.personalRecords) { + if (record.sourceMeasurementId) { + referencedMeasurementIds.add(record.sourceMeasurementId); } } + for (const recording of payload.ecgRecordings) { + if (recording.measurementId) { + referencedMeasurementIds.add(recording.measurementId); + } + } + const writtenReferencedIds = new Set(); + const measurementBatchSize = 1_000; + await streamed.forEachMeasurementBatch( + measurementBatchSize, + async (batch) => { + const stableRows = batch.flatMap( + (measurement): MeasurementInsertRow[] => + measurement.id + ? [ + { + id: measurement.id, + userId: ownerId, + ...toRestoredMeasurementData(measurement), + }, + ] + : [], + ); + // One statement per batch, not `createMany`: see + // `insertMeasurementRows` for what the latter cost here. + await insertMeasurementRows(tx, stableRows); + for (const row of stableRows) { + if (referencedMeasurementIds.has(row.id)) { + writtenReferencedIds.add(row.id); + } + } + + // v1 payloads did not require stable ids. Preserve their + // historical natural-key reconciliation without routing + // canonical v2 rows through it. + for (const measurement of batch) { + if (measurement.id) continue; + const restoredData = toRestoredMeasurementData(measurement); + const existing = await tx.measurement.findFirst({ + where: { + userId: ownerId, + type: measurement.type, + source: restoredData.source, + measuredAt: restoredData.measuredAt, + sleepStage: restoredData.sleepStage, + }, + select: { id: true }, + }); + if (existing) { + await tx.measurement.update({ + where: { id: existing.id, userId: ownerId }, + data: restoredData, + }); + } else { + await tx.measurement.create({ + data: { userId: ownerId, ...restoredData }, + }); + } + } + }, + ); const medByName = new Map(); const restoredMedicationIds = new Set(); @@ -1932,7 +1989,7 @@ const handler = apiHandler( tx, ownerId, payload, - new Set(stableRows.map((row) => row.id)), + writtenReferencedIds, skips, ); @@ -1949,7 +2006,7 @@ const handler = apiHandler( tx, ownerId, payload, - new Set(stableRows.map((row) => row.id)), + writtenReferencedIds, skips, ); @@ -2024,7 +2081,7 @@ const handler = apiHandler( }, { maxWait: 10_000, - timeout: 120_000, + timeout: restoreTransactionTimeoutMs(streamed.measurementCount), }, ); } catch (err) { @@ -2071,7 +2128,7 @@ const handler = apiHandler( // outside the transaction (the 5-year fold would otherwise hold a // long write lock) and is best-effort so a populator hiccup never // undoes the restore. The boot-time backfill is the safety net. - if (payload.measurements.length > 0) { + if (streamed.measurementCount > 0) { try { await recomputeUserRollups(ownerId); } catch (err) { @@ -2131,7 +2188,10 @@ const handler = apiHandler( } } - const summary = summarizeBackup(payload); + const summary = { + ...summarizeBackup(payload), + measurements: streamed.measurementCount, + }; await auditLog("admin.backups.restore", { userId: admin.id, diff --git a/src/app/api/admin/backups/[id]/summary/__tests__/route.test.ts b/src/app/api/admin/backups/[id]/summary/__tests__/route.test.ts index 766e0d4de..16f704733 100644 --- a/src/app/api/admin/backups/[id]/summary/__tests__/route.test.ts +++ b/src/app/api/admin/backups/[id]/summary/__tests__/route.test.ts @@ -44,6 +44,11 @@ vi.mock("@/lib/validations/backup", () => ({ parseBackupPayload: (...a: unknown[]) => parseMock(...a), summarizeBackup: (...a: unknown[]) => summarizeMock(...a), isCompatibleSchemaVersion: vi.fn(() => true), + // The measurements are read as a stream and counted one by one. + BACKUP_SCHEMA_VERSION: "2", + backupMeasurementSchema: { + safeParse: (value: unknown) => ({ success: true, data: value }), + }, })); import { GET } from "../route"; @@ -62,9 +67,14 @@ beforeEach(() => { createdAt: new Date("2026-08-01T00:00:00Z"), user: { id: "u1", username: "self-hoster" }, }); - decryptMock.mockReturnValue("{}"); + // Twelve measurements in the file: the preview counts them from the stream. + decryptMock.mockReturnValue( + JSON.stringify({ + measurements: Array.from({ length: 12 }, (_, i) => ({ id: `m${i}` })), + }), + ); parseMock.mockReturnValue({ schemaVersion: 3 }); - summarizeMock.mockReturnValue({ measurements: 12, moodEntries: 3 }); + summarizeMock.mockReturnValue({ measurements: 0, moodEntries: 3 }); }); describe("GET /api/admin/backups/[id]/summary", () => { diff --git a/src/app/api/admin/backups/[id]/summary/route.ts b/src/app/api/admin/backups/[id]/summary/route.ts index 4b79fcc79..cb40b404c 100644 --- a/src/app/api/admin/backups/[id]/summary/route.ts +++ b/src/app/api/admin/backups/[id]/summary/route.ts @@ -19,8 +19,12 @@ import { apiError, apiSuccess } from "@/lib/api-response"; import { BACKUP_UNDECRYPTABLE_CODE, BACKUP_UNDECRYPTABLE_ERROR, - unpackBackupBlob, + openBackupBlob, } from "@/lib/export/backup-blob"; +import { + readStreamedBackup, + type BackupSource, +} from "@/lib/export/streamed-backup"; import { annotate } from "@/lib/logging/context"; import { isCompatibleSchemaVersion, @@ -46,9 +50,11 @@ export const GET = apiHandler( throw new HttpError(404, "Backup not found"); } - let plaintext: string; + // Opened and read as a stream: a large record's JSON is longer than any + // string V8 can hold (#1031). + let source: BackupSource; try { - plaintext = unpackBackupBlob(backup.data); + source = openBackupBlob(backup.data); } catch { // Same refusal the restore and the download give, for the same reason: // a copy this instance cannot open is bad stored input, and the preview @@ -59,8 +65,11 @@ export const GET = apiHandler( } let payload; + let measurementCount: number; try { - payload = parseBackupPayload(plaintext); + const streamed = await readStreamedBackup(source); + measurementCount = streamed.measurementCount; + payload = parseBackupPayload(streamed.raw); } catch { return apiError("Backup payload failed schema validation", 422); } @@ -72,7 +81,10 @@ export const GET = apiHandler( ); } - const summary = summarizeBackup(payload); + const summary = { + ...summarizeBackup(payload), + measurements: measurementCount, + }; annotate({ action: { name: "admin.backups.summary" }, diff --git a/src/app/api/admin/backups/upload/route.ts b/src/app/api/admin/backups/upload/route.ts index b7b6443ea..8110b4bac 100644 --- a/src/app/api/admin/backups/upload/route.ts +++ b/src/app/api/admin/backups/upload/route.ts @@ -1,17 +1,25 @@ /** * POST /api/admin/backups/upload — admin-only ingest of a backup file. * - * Accepts a `multipart/form-data` body with a single `file` field that - * carries a JSON backup matching `backupPayloadSchema`. The route: + * Accepts the backup file either as the raw request body (JSON, or JSON + * compressed with gzip) or, for a small file, as a `multipart/form-data` + * body with a single `file` field. The route: * * 1. Authenticates the caller as an admin (cookie session only). * 2. Reads the file (size capped — see `MAX_UPLOAD_BYTES`). * 3. Parses and validates the JSON against the canonical schema. * 4. Rejects files written by a future schema version (incompatible). - * 5. Encrypts the plaintext and inserts a new `DataBackup` row of - * type `MANUAL_UPLOAD_` so multiple uploads coexist with + * 5. Encrypts the file as it arrives and inserts a new `DataBackup` row + * of type `MANUAL_UPLOAD_` so multiple uploads coexist with * the rolling `WEEKLY_AUTO` snapshot AND with each other. * + * The file is read as a stream and never held whole (#1031): an export of an + * account with 1.25 million measurements is a few hundred megabytes of JSON, + * which neither fits a 1 GB container beside the running app nor, in the + * disaster-recovery shape, fits in one JavaScript string at all. It is + * checked as it goes and stored in the same pass; a file that fails any check + * rolls the store back, so nothing of it remains. + * * Crucially, this route does NOT execute a restore. It only stores the * file. The separate `POST /api/admin/backups/[id]/restore` endpoint * (criterion 3) actually replaces user data — keeping the two phases @@ -20,46 +28,120 @@ * * Phase B1 / criterion 2 of the v1.4.15 backup-completeness work. */ +import { Readable } from "node:stream"; +import { createGunzip } from "node:zlib"; + import { NextRequest } from "next/server"; import { ZodError } from "zod/v4"; import { prisma } from "@/lib/db"; import { apiHandler, HttpError, requireAdmin } from "@/lib/api-handler"; import { apiError, apiSuccess, getClientIp } from "@/lib/api-response"; import { auditLog } from "@/lib/auth/audit"; -import { packBackupBlob } from "@/lib/export/backup-blob"; +import { BackupJsonError, scanBackupJson } from "@/lib/export/backup-json-scan"; +import { storeBackupBlob } from "@/lib/export/store-backup-blob"; +import { StreamedBackupInvalidError } from "@/lib/export/streamed-backup"; import { annotate } from "@/lib/logging/context"; import { checkRateLimit } from "@/lib/rate-limit"; import { + BACKUP_SCHEMA_VERSION, + backupMeasurementSchema, backupPayloadSchema, isCompatibleSchemaVersion, summarizeBackup, + type BackupPayload, type BackupSummary, } from "@/lib/validations/backup"; export const dynamic = "force-dynamic"; /** - * Cap on uploaded file size. The largest production backups today are - * ~2 MB; 10 MB leaves headroom while keeping a malicious admin client - * from blowing memory by streaming a 4 GB file. Enforced by reading the - * `Content-Length` header AND the actual buffered size after decoding — - * a missing/wrong content-length doesn't bypass the limit. + * Cap on a file sent as a multipart form. The form parser holds the whole + * file in memory, so this path stays for small files only; anything larger + * is sent as the raw request body, which is read as a stream. */ const MAX_UPLOAD_BYTES = 10 * 1024 * 1024; +/** + * Cap on a file sent as the raw request body, compressed or not. The same + * ceiling the app sets for any request body (`middlewareClientMaxBodySize` in + * `next.config.ts`), beyond which the body would arrive truncated. A + * disaster-recovery file of 1.25 million measurements is 662 MB as plain + * JSON and 64 MB compressed, so large files go up compressed. + */ +const MAX_RAW_UPLOAD_BYTES = 512 * 1024 * 1024; + interface UploadResponse { id: string; valid: true; summary: BackupSummary; } +/** A file the route refuses: the answer, and what the audit row records. */ +class UploadRefused extends Error { + constructor( + readonly status: number, + readonly answer: string, + readonly details: Record, + readonly meta?: Record, + ) { + super(answer); + this.name = "UploadRefused"; + } +} + +/** Bytes → bytes, gunzipped when the file starts with the gzip magic. */ +async function* decodedBytes( + source: AsyncIterable, + limit: number, +): AsyncGenerator { + let seen = 0; + async function* counted() { + for await (const chunk of source) { + seen += chunk.byteLength; + if (seen > limit) { + throw new UploadRefused( + 413, + `Upload exceeds ${Math.round(limit / 1024 / 1024)} MB limit`, + { reason: "file_size_exceeded", size: seen }, + ); + } + yield chunk; + } + } + const iterator = counted()[Symbol.asyncIterator](); + const first = await iterator.next(); + if (first.done) return; + const rest = (async function* () { + yield first.value; + for (;;) { + const next = await iterator.next(); + if (next.done) return; + yield next.value; + } + })(); + const head = first.value; + if (head.byteLength >= 2 && head[0] === 0x1f && head[1] === 0x8b) { + const gunzip = Readable.from(rest).pipe(createGunzip()); + try { + for await (const chunk of gunzip) yield chunk as Buffer; + } catch (err) { + if (err instanceof UploadRefused) throw err; + throw new UploadRefused(422, "Uploaded file is not valid gzip", { + reason: "invalid_gzip", + }); + } + return; + } + yield* rest; +} + export const POST = apiHandler(async (request: NextRequest) => { const { user: admin } = await requireAdmin(); annotate({ action: { name: "admin.backups.upload" } }); + const ipAddress = getClientIp(request); - // Rate-limit identical to the manual-run endpoint. An admin uploading - // a 10 MB blob 60× per minute would just be wasteful — three is plenty - // for a deliberate restore-prep workflow. + // Rate-limit identical to the manual-run endpoint. Three is plenty for a + // deliberate restore-prep workflow. const rl = await checkRateLimit( `admin-backups-upload:${admin.id}`, 3, @@ -69,120 +151,56 @@ export const POST = apiHandler(async (request: NextRequest) => { throw new HttpError(429, "Too many backup uploads"); } - // Cheap pre-flight on the declared content length. The actual buffer - // size check below is the hard limit; this just rejects obvious abuse - // before allocating the FormData parser. + const denied = (reason: string, extra: Record = {}) => + auditLog("admin.backups.upload.denied", { + userId: admin.id, + ipAddress, + details: { reason, ...extra }, + }); + + const contentType = request.headers.get("content-type") ?? ""; + const multipart = contentType.startsWith("multipart/form-data"); + const limit = multipart ? MAX_UPLOAD_BYTES : MAX_RAW_UPLOAD_BYTES; + + // Cheap pre-flight on the declared content length. The byte count while + // reading is the hard limit; this just rejects obvious abuse early. const contentLength = Number(request.headers.get("content-length") ?? 0); - if (contentLength > MAX_UPLOAD_BYTES) { - await auditLog("admin.backups.upload.denied", { - userId: admin.id, - ipAddress: getClientIp(request), - details: { reason: "content_length_exceeded", contentLength }, - }); - return apiError("Upload exceeds 10 MB limit", 413); - } - - let formData: FormData; - try { - formData = await request.formData(); - } catch (err) { + if (contentLength > limit) { + await denied("content_length_exceeded", { contentLength }); return apiError( - `Invalid multipart body: ${err instanceof Error ? err.message : "unknown"}`, - 400, + `Upload exceeds ${Math.round(limit / 1024 / 1024)} MB limit`, + 413, ); } - const file = formData.get("file"); - if (!(file instanceof File)) { - return apiError("Field 'file' must be a file", 422); - } - - if (file.size > MAX_UPLOAD_BYTES) { - await auditLog("admin.backups.upload.denied", { - userId: admin.id, - ipAddress: getClientIp(request), - details: { reason: "file_size_exceeded", size: file.size }, - }); - return apiError("Upload exceeds 10 MB limit", 413); - } - - let plaintext: string; - try { - plaintext = await file.text(); - } catch { - return apiError("Failed to read uploaded file", 400); - } - - let parsedJson: unknown; - try { - parsedJson = JSON.parse(plaintext); - } catch (err) { - await auditLog("admin.backups.upload.denied", { - userId: admin.id, - ipAddress: getClientIp(request), - details: { - reason: "invalid_json", - message: err instanceof Error ? err.message : "parse_failed", - }, - }); - return apiError("Uploaded file is not valid JSON", 422); - } - - let payload; - try { - payload = backupPayloadSchema.parse(parsedJson); - } catch (err) { - const issues = - err instanceof ZodError - ? err.issues.slice(0, 10).map((i) => ({ - path: i.path.join("."), - message: i.message, - })) - : []; - await auditLog("admin.backups.upload.denied", { - userId: admin.id, - ipAddress: getClientIp(request), - details: { reason: "schema_invalid", issues }, - }); - return apiError("Backup payload failed schema validation", 422, { - issues, - }); - } - - if (!isCompatibleSchemaVersion(payload.schemaVersion)) { - await auditLog("admin.backups.upload.denied", { - userId: admin.id, - ipAddress: getClientIp(request), - details: { - reason: "incompatible_schema_version", - schemaVersion: payload.schemaVersion, - }, - }); - return apiError( - `Backup schema version '${payload.schemaVersion}' is not supported by this server`, - 422, + let source: AsyncIterable; + if (multipart) { + let formData: FormData; + try { + formData = await request.formData(); + } catch (err) { + return apiError( + `Invalid multipart body: ${err instanceof Error ? err.message : "unknown"}`, + 400, + ); + } + const file = formData.get("file"); + if (!(file instanceof File)) { + return apiError("Field 'file' must be a file", 422); + } + if (file.size > MAX_UPLOAD_BYTES) { + await denied("file_size_exceeded", { size: file.size }); + return apiError("Upload exceeds 10 MB limit", 413); + } + source = Readable.fromWeb( + file.stream() as Parameters[0], ); - } - - // Make sure the userId on the file points at a user that exists in - // this DB. Restore would fail later anyway, but failing here gives - // the admin a precise error before any side-effect. - const owner = await prisma.user.findUnique({ - where: { id: payload.userId }, - select: { id: true, username: true }, - }); - if (!owner) { - await auditLog("admin.backups.upload.denied", { - userId: admin.id, - ipAddress: getClientIp(request), - details: { - reason: "owner_not_found", - ownerId: payload.userId, - }, - }); - return apiError( - `Backup is for user '${payload.userId}' which does not exist in this DB`, - 422, + } else { + if (!request.body) { + return apiError("The backup file is missing from the request body", 422); + } + source = Readable.fromWeb( + request.body as Parameters[0], ); } @@ -190,25 +208,146 @@ export const POST = apiHandler(async (request: NextRequest) => { // intact so multiple uploads can coexist for the same user without // overwriting each other or the rolling WEEKLY_AUTO snapshot. const uploadType = `MANUAL_UPLOAD_${Date.now()}`; - const encrypted = packBackupBlob(JSON.stringify(payload)); + let payload: BackupPayload | undefined; + let measurementCount = 0; + let owner: { id: string; username: string } | null = null; - const created = await prisma.dataBackup.create({ - data: { - userId: owner.id, - type: uploadType, - data: encrypted, - }, - }); + let created: { id: string }; + try { + created = await storeBackupBlob( + prisma, + () => ({ userId: owner!.id, type: uploadType }), + async (write) => { + // One pass: every chunk is stored as it is checked, and a check that + // fails throws, which rolls the store back. + async function* storedAsRead() { + for await (const chunk of decodedBytes(source, limit)) { + await write(Buffer.from(chunk)); + yield chunk; + } + } + let firstWithoutId: number | null = null; + let scanned; + try { + scanned = await scanBackupJson(storedAsRead(), { + streamKeys: new Set(["measurements"]), + onElement: (_key, element, index) => { + 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", + ); + } + if (!parsed.data.id && firstWithoutId === null) { + firstWithoutId = index; + } + }, + }); + } catch (err) { + if (err instanceof BackupJsonError) { + throw new UploadRefused(422, "Uploaded file is not valid JSON", { + reason: "invalid_json", + message: err.message, + }); + } + if (err instanceof StreamedBackupInvalidError) { + const issues = [{ path: err.path, message: err.message }]; + throw new UploadRefused( + 422, + "Backup payload failed schema validation", + { reason: "schema_invalid", issues }, + { issues }, + ); + } + throw err; + } + measurementCount = scanned.streamedCounts.measurements ?? 0; - const summary = summarizeBackup(payload); + try { + payload = backupPayloadSchema.parse(scanned.document); + } catch (err) { + const issues = + err instanceof ZodError + ? err.issues.slice(0, 10).map((i) => ({ + path: i.path.join("."), + message: i.message, + })) + : []; + throw new UploadRefused( + 422, + "Backup payload failed schema validation", + { reason: "schema_invalid", issues }, + { issues }, + ); + } + if ( + payload.schemaVersion === BACKUP_SCHEMA_VERSION && + firstWithoutId !== null + ) { + const issues = [ + { + path: `measurements.${firstWithoutId}.id`, + message: "Canonical v2 measurements require a stable id", + }, + ]; + throw new UploadRefused( + 422, + "Backup payload failed schema validation", + { reason: "schema_invalid", issues }, + { issues }, + ); + } + + if (!isCompatibleSchemaVersion(payload.schemaVersion)) { + throw new UploadRefused( + 422, + `Backup schema version '${payload.schemaVersion}' is not supported by this server`, + { + reason: "incompatible_schema_version", + schemaVersion: payload.schemaVersion, + }, + ); + } + + // Make sure the userId on the file points at a user that exists in + // this DB. Restore would fail later anyway, but failing here gives + // the admin a precise error before any side-effect. + owner = await prisma.user.findUnique({ + where: { id: payload.userId }, + select: { id: true, username: true }, + }); + if (!owner) { + throw new UploadRefused( + 422, + `Backup is for user '${payload.userId}' which does not exist in this DB`, + { reason: "owner_not_found", ownerId: payload.userId }, + ); + } + }, + ); + } catch (err) { + if (err instanceof UploadRefused) { + await denied(err.details.reason as string, err.details); + return apiError(err.answer, err.status, err.meta); + } + throw err; + } + + const summary: BackupSummary = { + ...summarizeBackup(payload!), + measurements: measurementCount, + }; + const storedOwner = owner as { id: string; username: string } | null; await auditLog("admin.backups.upload", { userId: admin.id, - ipAddress: getClientIp(request), + ipAddress, details: { backupId: created.id, - ownerId: owner.id, - ownerUsername: owner.username, + ownerId: storedOwner?.id, + ownerUsername: storedOwner?.username, type: uploadType, schemaVersion: summary.schemaVersion, counts: { diff --git a/src/app/api/export/__tests__/soft-delete-filter.test.ts b/src/app/api/export/__tests__/soft-delete-filter.test.ts index bd4a5669c..ff386b3ec 100644 --- a/src/app/api/export/__tests__/soft-delete-filter.test.ts +++ b/src/app/api/export/__tests__/soft-delete-filter.test.ts @@ -64,6 +64,29 @@ vi.mock("@/lib/db", () => ({ documentConditionLink: { findMany: vi.fn() }, extractedFact: { findMany: vi.fn() }, ecgRecording: { findMany: vi.fn() }, + onboardingRecord: { findUnique: vi.fn().mockResolvedValue(null) }, + environmentTravelLocation: { findMany: vi.fn().mockResolvedValue([]) }, + environmentContext: { findMany: vi.fn().mockResolvedValue([]) }, + personalRecord: { findMany: vi.fn().mockResolvedValue([]) }, + coachReminder: { findMany: vi.fn().mockResolvedValue([]) }, + coachPlan: { findMany: vi.fn().mockResolvedValue([]) }, + coachFact: { findMany: vi.fn().mockResolvedValue([]) }, + coachConversation: { findMany: vi.fn().mockResolvedValue([]) }, + measurementReminderEvent: { findMany: vi.fn().mockResolvedValue([]) }, + measurementReminder: { findMany: vi.fn().mockResolvedValue([]) }, + vaccinationDocumentLink: { findMany: vi.fn().mockResolvedValue([]) }, + vaccinationRecord: { findMany: vi.fn().mockResolvedValue([]) }, + encounterConditionLink: { findMany: vi.fn().mockResolvedValue([]) }, + encounterLabLink: { findMany: vi.fn().mockResolvedValue([]) }, + encounterDocumentLink: { findMany: vi.fn().mockResolvedValue([]) }, + encounter: { findMany: vi.fn().mockResolvedValue([]) }, + practitioner: { findMany: vi.fn().mockResolvedValue([]) }, + healthScoreRecord: { findMany: vi.fn().mockResolvedValue([]) }, + intradayCumulativeProfile: { findMany: vi.fn().mockResolvedValue([]) }, + correlationPattern: { findMany: vi.fn().mockResolvedValue([]) }, + healthProfileFactRevision: { findMany: vi.fn().mockResolvedValue([]) }, + customMetric: { findMany: vi.fn().mockResolvedValue([]) }, + cycleSymptom: { findMany: vi.fn().mockResolvedValue([]) }, }, })); @@ -177,6 +200,7 @@ beforeEach(() => { ); vi.mocked(prisma.extractedFact.findMany).mockResolvedValue([] as never); vi.mocked(prisma.ecgRecording.findMany).mockResolvedValue([] as never); + vi.mocked(prisma.userAchievement.findMany).mockResolvedValue([] as never); }); describe("v1.4.41 W-DELETED-2 — soft-delete invisibility", () => { @@ -192,7 +216,13 @@ describe("v1.4.41 W-DELETED-2 — soft-delete invisibility", () => { it("/api/export/full-backup scopes the measurement read to deletedAt: null", async () => { const { GET } = await import("../full-backup/route"); - await GET(mkReq("http://localhost/api/export/full-backup")); + // The file is written into the response as it is produced, so the reads + // happen while the body is consumed. The mocked client does not cover + // every section the builder reads, so the body errors part-way (as the + // whole response did before it streamed); the reads asserted here have + // been issued by then. + const res = await GET(mkReq("http://localhost/api/export/full-backup")); + await res.text(); expect(prisma.measurement.findMany).toHaveBeenCalledWith( expect.objectContaining({ where: expect.objectContaining({ deletedAt: null }), @@ -202,7 +232,10 @@ describe("v1.4.41 W-DELETED-2 — soft-delete invisibility", () => { it("/api/export/full-backup scopes every v1.28 records domain to userId + deletedAt: null", async () => { const { GET } = await import("../full-backup/route"); - await GET(mkReq("http://localhost/api/export/full-backup")); + // The file is written into the response as it is produced, so the reads + // happen while the body is consumed. + const res = await GET(mkReq("http://localhost/api/export/full-backup")); + await res.text(); expect(prisma.labResult.findMany).toHaveBeenCalledWith( expect.objectContaining({ diff --git a/src/app/api/export/encrypted/route.ts b/src/app/api/export/encrypted/route.ts index e68f4f987..bd19ae009 100644 --- a/src/app/api/export/encrypted/route.ts +++ b/src/app/api/export/encrypted/route.ts @@ -42,13 +42,44 @@ import { safeJson, } from "@/lib/api-response"; import { checkRateLimit } from "@/lib/rate-limit"; -import { buildFullBackupPayload } from "@/lib/export/full-backup-payload"; +import { randomBytes } from "node:crypto"; +import { createReadStream } from "node:fs"; +import { readdir, rm, stat } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import type { FullBackupCounts } from "@/lib/export/full-backup-payload"; +import { streamFullBackupJson } from "@/lib/export/full-backup-stream"; import { - encryptArchive, + encryptArchiveToFile, MIN_EXPORT_PASSPHRASE_LENGTH, + type SpooledArchive, } from "@/lib/export/passphrase-archive"; +import { streamToResponseBody } from "@/lib/export/response-stream"; import { NextRequest, NextResponse } from "next/server"; +const SPOOL_PREFIX = "healthlog-export-"; + +/** + * Remove spool files an earlier export left behind: a process that died + * mid-export never reached its cleanup. An hour is far longer than any export + * takes, so nothing still being written or sent is touched. + */ +async function removeStaleSpools(): Promise { + try { + const dir = tmpdir(); + const cutoff = Date.now() - 60 * 60 * 1000; + for (const name of await readdir(dir)) { + if (!name.startsWith(SPOOL_PREFIX)) continue; + const path = join(dir, name); + const info = await stat(path).catch(() => null); + if (info && info.mtimeMs < cutoff) await rm(path, { force: true }); + } + } catch { + // Housekeeping only; an export must not fail over it. + } +} + const encryptedExportSchema = z .object({ passphrase: z @@ -89,10 +120,31 @@ export const POST = apiHandler(async (request: NextRequest) => { } const { passphrase } = parsed.data; - const { payload, counts } = await buildFullBackupPayload(prisma, user.id); - - // Encrypt the JSON bytes. The passphrase goes no further than the KDF. - const archive = await encryptArchive(JSON.stringify(payload), passphrase); + // Sealed as it is produced and spooled to a temporary file, then sent: the + // format puts the tag in front of the ciphertext, so the archive cannot go + // out before its last byte is encrypted, and holding it in memory is what + // took the app down on an account of 1.25 million measurements (#1031). Only + // ciphertext touches the disk, and the file is removed once it is sent or + // the client leaves. + let counts: FullBackupCounts | undefined; + await removeStaleSpools(); + const bodyPath = join( + tmpdir(), + `${SPOOL_PREFIX}${randomBytes(12).toString("hex")}.hlx.part`, + ); + let archive: SpooledArchive; + try { + archive = await encryptArchiveToFile( + async (write) => { + counts = await streamFullBackupJson(prisma, user.id, write); + }, + passphrase, + bodyPath, + ); + } catch (err) { + await rm(bodyPath, { force: true }); + throw err; + } await auditLog("user.export.encrypted", { userId: user.id, @@ -102,38 +154,36 @@ export const POST = apiHandler(async (request: NextRequest) => { annotate({ meta: { - export_measurements_count: counts.measurements, - export_medications_count: counts.medications, - export_intake_count: counts.intakeEvents, - export_medication_side_effect_count: counts.medicationSideEffects, - export_mood_count: counts.moodEntries, - export_cycle_count: counts.cycles, - export_cycle_day_log_count: counts.cycleDayLogs, - export_lab_result_count: counts.labResults, - export_biomarker_count: counts.biomarkers, - export_illness_episode_count: counts.illnessEpisodes, - export_illness_day_log_count: counts.illnessDayLogs, - export_allergy_count: counts.allergies, - export_family_history_count: counts.familyHistory, - export_workout_count: counts.workouts, - export_document_count: counts.documents, - export_nutrient_day_count: counts.nutrientDays, - export_health_profile_count: counts.healthProfile, - export_custom_metric_count: counts.customMetrics, - export_custom_metric_entry_count: counts.customMetricEntries, - export_intraday_profile_count: counts.intradayProfiles, + export_measurements_count: counts?.measurements, + export_medications_count: counts?.medications, + export_intake_count: counts?.intakeEvents, + export_mood_count: counts?.moodEntries, export_archive_bytes: archive.byteLength, }, }); + const archiveBody = streamToResponseBody( + async (write) => { + await write(archive.prefix); + for await (const chunk of createReadStream(archive.bodyPath)) { + await write(chunk as Buffer); + } + }, + { + onComplete: () => rm(archive.bodyPath, { force: true }), + onError: () => rm(archive.bodyPath, { force: true }), + }, + ); + const stamp = new Date().toISOString().slice(0, 10); // Return the raw binary archive (NOT the apiSuccess envelope). The file is a // self-contained `.hlx` archive openable with the user's passphrase via // scripts/decrypt-export.ts. - return new NextResponse(new Uint8Array(archive), { + return new NextResponse(archiveBody, { status: 200, headers: { "Content-Type": "application/octet-stream", + "Content-Length": String(archive.byteLength), "Content-Disposition": `attachment; filename="healthlog-backup-${user.id}-${stamp}.hlx"`, "Cache-Control": "no-store", }, diff --git a/src/app/api/export/full-backup/route.ts b/src/app/api/export/full-backup/route.ts index c7e83a262..b36201c92 100644 --- a/src/app/api/export/full-backup/route.ts +++ b/src/app/api/export/full-backup/route.ts @@ -35,7 +35,9 @@ import { annotate } from "@/lib/logging/context"; import { auditLog } from "@/lib/auth/audit"; import { apiError, getClientIp } from "@/lib/api-response"; import { checkRateLimit } from "@/lib/rate-limit"; -import { buildFullBackupPayload } from "@/lib/export/full-backup-payload"; +import type { FullBackupCounts } from "@/lib/export/full-backup-payload"; +import { streamFullBackupJson } from "@/lib/export/full-backup-stream"; +import { streamToResponseBody } from "@/lib/export/response-stream"; import { NextRequest, NextResponse } from "next/server"; export const GET = apiHandler(async (request: NextRequest) => { @@ -47,48 +49,44 @@ export const GET = apiHandler(async (request: NextRequest) => { return apiError("Maximum 10 exports per hour", 429); } - // v1.23 — the payload builder is shared with the passphrase-encrypted - // export route so both emit the byte-for-byte same shape (the measurement/ - // medication/mood/cycle domains are restore-compatible; the newer records - // domains are export-only for now — see the module doc comment above). - const { payload, counts } = await buildFullBackupPayload(prisma, user.id); - - await auditLog("user.export.full-backup", { - userId: user.id, - ipAddress: getClientIp(request), - details: { counts }, - }); - - annotate({ - meta: { - export_measurements_count: counts.measurements, - export_medications_count: counts.medications, - export_intake_count: counts.intakeEvents, - export_medication_side_effect_count: counts.medicationSideEffects, - export_mood_count: counts.moodEntries, - export_cycle_count: counts.cycles, - export_cycle_day_log_count: counts.cycleDayLogs, - export_lab_result_count: counts.labResults, - export_biomarker_count: counts.biomarkers, - export_illness_episode_count: counts.illnessEpisodes, - export_illness_day_log_count: counts.illnessDayLogs, - export_allergy_count: counts.allergies, - export_family_history_count: counts.familyHistory, - export_workout_count: counts.workouts, - export_document_count: counts.documents, - export_nutrient_day_count: counts.nutrientDays, - export_health_profile_count: counts.healthProfile, - export_custom_metric_count: counts.customMetrics, - export_custom_metric_entry_count: counts.customMetricEntries, - export_intraday_profile_count: counts.intradayProfiles, + // Written into the response as it is produced (`streamFullBackupJson`), not + // built and stringified first: for an account of 1.25 million measurements + // the built payload and its JSON string did not fit a 1 GB container, and + // the export took the whole app down (#1031). The file is byte for byte + // what `JSON.stringify` of the payload builder's object gives; the passphrase + // export beside this route writes the same bytes, encrypted. + // + // The counts are only known once the last row has gone out, so the audit + // row is written then. A file that stops early is audited as a failure. + const ipAddress = getClientIp(request); + let counts: FullBackupCounts | undefined; + const body = streamToResponseBody( + async (write) => { + counts = await streamFullBackupJson(prisma, user.id, write); }, - }); + { + onComplete: () => + auditLog("user.export.full-backup", { + userId: user.id, + ipAddress, + details: { counts }, + }), + onError: (err) => + auditLog("user.export.full-backup.failed", { + userId: user.id, + ipAddress, + details: { + reason: err instanceof Error ? err.message : String(err), + }, + }), + }, + ); const stamp = new Date().toISOString().slice(0, 10); // Stream the JSON directly (NOT wrapped in the apiSuccess envelope) so // the file is a self-contained backup — admin upload + restore expect // the raw payload, not `{ data: { ... } }`. - return new NextResponse(JSON.stringify(payload), { + return new NextResponse(body, { status: 200, headers: { "Content-Type": "application/json; charset=utf-8", diff --git a/src/components/admin/__tests__/backups-section-empty.test.tsx b/src/components/admin/__tests__/backups-section-empty.test.tsx index a29f39741..85e81600d 100644 --- a/src/components/admin/__tests__/backups-section-empty.test.tsx +++ b/src/components/admin/__tests__/backups-section-empty.test.tsx @@ -130,10 +130,10 @@ describe("BackupsSection — empty state", () => { const className = helpElement?.match(/\bclass="([^"]*)"/)?.[1] ?? ""; expect(english).toContain( - "JSON file matching the current backup schema. Max 10 MB.", + "A HealthLog backup as .json or .json.gz. The file is compressed before upload; compressed, it may be up to 512 MB.", ); expect(german).toContain( - "JSON-Datei passend zum aktuellen Backup-Schema. Max. 10 MB.", + "HealthLog-Backup als .json oder .json.gz. Die Datei wird vor dem Hochladen komprimiert; komprimiert darf sie bis zu 512 MB groß sein.", ); expect(helpElement).toBeDefined(); expect(className.split(/\s+/)).toContain("text-muted-foreground"); diff --git a/src/components/admin/backups-section.tsx b/src/components/admin/backups-section.tsx index fb065f101..5fc81ab58 100644 --- a/src/components/admin/backups-section.tsx +++ b/src/components/admin/backups-section.tsx @@ -454,6 +454,20 @@ function missingSectionLabel( * the server may reword, and matching on it would turn a copy edit into a * silent loss of the panel. */ + +/** + * The upload body for a backup file: the file itself when it is already + * gzip, otherwise the file compressed in the browser. The server accepts + * both; compressing keeps a large export well inside the request size limit. + */ +async function uploadBody(file: File): Promise { + const head = new Uint8Array(await file.slice(0, 2).arrayBuffer()); + if (head[0] === 0x1f && head[1] === 0x8b) return file; + return new Response( + file.stream().pipeThrough(new CompressionStream("gzip")), + ).blob(); +} + export function missingSectionsOf(err: unknown): MissingBackupSection[] { if (!(err instanceof ApiError)) return []; if (err.meta?.errorCode !== "backup.section.missing") return []; @@ -697,10 +711,11 @@ export function BackupsSection() { const fileInputRef = useRef(null); const upload = useMutation({ mutationFn: async (file: File) => { - const fd = new FormData(); - fd.append("file", file); - // apiFetch with a raw init: the multipart body must NOT be JSON - // re-encoded, so the verb helper (which JSON-stringifies) is out. + // Sent as the raw body, compressed on the way unless it already is: the + // server reads it as a stream, and a large export (hundreds of MB of + // JSON) shrinks about tenfold. No request timeout, because a file of + // that size takes longer to check and store than the default allows. + const body = await uploadBody(file); return apiFetch<{ id: string; valid: true; @@ -734,7 +749,12 @@ export function BackupsSection() { customMetricEntries?: number; intradayProfiles?: number; }; - }>("/api/admin/backups/upload", { method: "POST", body: fd }); + }>("/api/admin/backups/upload", { + method: "POST", + body, + headers: { "Content-Type": "application/gzip" }, + signal: null, + }); }, onSuccess: (data) => { const total = @@ -818,7 +838,8 @@ export function BackupsSection() { return apiPost<{ restored: true; skipped?: RestoreSkipSummary }>( `/api/admin/backups/${row.id}/restore`, { confirm: "RESTORE", restoreInstanceSettings }, - { headers: { "Idempotency-Key": idempotencyKey } }, + // No request timeout: rebuilding a large account takes minutes. + { headers: { "Idempotency-Key": idempotencyKey }, signal: null }, ); }, onSuccess: (data) => { @@ -959,7 +980,7 @@ export function BackupsSection() { = new Set(); +/** + * Ciphertext written by raw SQL, which the Prisma-call walk below cannot see. + * + * `DataBackup.data` is the case: the stored backup is produced as ciphertext + * pieces (`packBackupBlobInto`) and assembled into the row by one SQL + * statement, because holding it as one string in the process is what a large + * record could not afford (#1031). Each entry names the file and the + * statement, and the scan fails if the statement is no longer there, so a + * moved or removed raw write cannot leave a stale entry that still passes. + */ +const RAW_CIPHERTEXT_WRITES: ReadonlyArray<{ + key: string; + file: string; + statement: RegExp; +}> = [ + { + key: "DataBackup.data", + file: "src/lib/export/store-backup-blob.ts", + statement: + /UPDATE\s+data_backups\s+SET\s+data\s*=\s*\(\s*SELECT\s+string_agg/, + }, +]; + function sourceFiles(dir: string, out: string[] = []): string[] { for (const entry of readdirSync(dir)) { const p = join(dir, entry); @@ -388,6 +411,12 @@ function scanCiphertextWrites(): Map> { `\\b(?:prisma|tx|db)\\s*\\.\\s*(\\w+)\\s*\\.\\s*(?:${WRITE_METHODS})\\s*\\(`, "g", ); + for (const raw of RAW_CIPHERTEXT_WRITES) { + if (rel !== raw.file) continue; + const hit = raw.statement.exec(src); + if (hit) record(...(raw.key.split(".") as [string, string]), hit.index); + } + for (const m of src.matchAll(writes)) { const model = m[1][0].toUpperCase() + m[1].slice(1); if (!schema.models.has(model)) continue; @@ -425,6 +454,17 @@ describe("ciphertext columns derived from the code", () => { } }); + it("still finds every raw-SQL ciphertext write it was told about", () => { + for (const raw of RAW_CIPHERTEXT_WRITES) { + expect( + [...(written.get(raw.key) ?? [])].some((at) => + at.startsWith(`${raw.file}:`), + ), + `${raw.key}: no matching statement left in ${raw.file}`, + ).toBe(true); + } + }); + it("finds ciphertext writes, including the non-suffixed columns", () => { // Anchored non-zero match count. An empty or collapsed result set is the // failure mode this whole file exists to make impossible, so name the diff --git a/src/lib/export/__tests__/backup-json-scan.test.ts b/src/lib/export/__tests__/backup-json-scan.test.ts new file mode 100644 index 000000000..773e8080e --- /dev/null +++ b/src/lib/export/__tests__/backup-json-scan.test.ts @@ -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 = {}; + 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 = {}; + 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; + + 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"); + }); +}); diff --git a/src/lib/export/__tests__/passphrase-archive.test.ts b/src/lib/export/__tests__/passphrase-archive.test.ts index 2cd158c97..dcdb84c5b 100644 --- a/src/lib/export/__tests__/passphrase-archive.test.ts +++ b/src/lib/export/__tests__/passphrase-archive.test.ts @@ -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 }); + } + }); +}); diff --git a/src/lib/export/backup-blob.ts b/src/lib/export/backup-blob.ts index f305a67b4..0c152ba45 100644 --- a/src/lib/export/backup-blob.ts +++ b/src/lib/export/backup-blob.ts @@ -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, + write: (chunk: string | Buffer) => Promise, ) => Promise; export interface PackBackupBlobOptions { @@ -236,7 +237,7 @@ export async function packBackupBlobInto( await sink(piece); }; - const write = async (chunk: string): Promise => { + const write = async (chunk: string | Buffer): Promise => { 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 { + 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()); +} diff --git a/src/lib/export/backup-json-scan.ts b/src/lib/export/backup-json-scan.ts new file mode 100644 index 000000000..60c3c967a --- /dev/null +++ b/src/lib/export/backup-json-scan.ts @@ -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; + /** + * 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; +} + +export interface ScanBackupJsonResult { + /** The document, with streamed members as empty arrays and skipped ones absent. */ + document: Record; + /** How many elements each streamed member held. */ + streamedCounts: Record; +} + +/** 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, + options: ScanBackupJsonOptions = {}, +): Promise { + const streamKeys = options.streamKeys ?? new Set(); + const skipKeys = options.skipKeys ?? new Set(); + const document: Record = {}; + const streamedCounts: Record = {}; + + 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 }; +} diff --git a/src/lib/export/measurement-bulk-insert.ts b/src/lib/export/measurement-bulk-insert.ts new file mode 100644 index 000000000..f1db771d9 --- /dev/null +++ b/src/lib/export/measurement-bulk-insert.ts @@ -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; + +/** 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 { + 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); +} diff --git a/src/lib/export/passphrase-archive.ts b/src/lib/export/passphrase-archive.ts index e30ee171b..f1793bdaf 100644 --- a/src/lib/export/passphrase-archive.ts +++ b/src/lib/export/passphrase-archive.ts @@ -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, + ) => Promise, + passphrase: string, + bodyPath: string, +): Promise { + 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((_, reject) => { + file.once("error", reject); + }); + fileError.catch(() => {}); + + const put = async (bytes: Buffer): Promise => { + 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. */ diff --git a/src/lib/export/response-stream.ts b/src/lib/export/response-stream.ts new file mode 100644 index 000000000..964bb2c41 --- /dev/null +++ b/src/lib/export/response-stream.ts @@ -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, +) => Promise; + +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 { + const out = new PassThrough(); + + const write = async (chunk: string | Buffer): Promise => { + if (out.destroyed || out.writableEnded) + throw new ResponseStreamClosedError(); + if (out.write(chunk)) return; + await new Promise((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; +} diff --git a/src/lib/export/store-backup-blob.ts b/src/lib/export/store-backup-blob.ts index cf7989269..82c8f0fae 100644 --- a/src/lib/export/store-backup-blob.ts +++ b/src/lib/export/store-backup-blob.ts @@ -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 { +): 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 }, ); diff --git a/src/lib/export/streamed-backup.ts b/src/lib/export/streamed-backup.ts new file mode 100644 index 000000000..1e65d6150 --- /dev/null +++ b/src/lib/export/streamed-backup.ts @@ -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; + +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; + /** 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, + ): Promise; +} + +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 { + 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); + }, + }; +} diff --git a/src/lib/jobs/offhost-backup.ts b/src/lib/jobs/offhost-backup.ts index 3627b1d2b..19bb54f84 100644 --- a/src/lib/jobs/offhost-backup.ts +++ b/src/lib/jobs/offhost-backup.ts @@ -23,8 +23,8 @@ */ import { Buffer } from "node:buffer"; import { createCipheriv, createDecipheriv, randomBytes } from "node:crypto"; -import { Transform, type Readable } from "node:stream"; -import { createGzip, gunzipSync, gzipSync } from "node:zlib"; +import { Readable, Transform } from "node:stream"; +import { createGunzip, createGzip, gunzipSync, gzipSync } from "node:zlib"; import type { PrismaClient } from "@/generated/prisma/client"; import { createRawStreamEncryptor, decryptRawStream } from "@/lib/crypto"; import { streamFullBackupJson } from "@/lib/export/full-backup-stream"; @@ -193,6 +193,48 @@ export function decryptBackup(buf: Buffer, key: Buffer): string { : plaintext.toString("utf8"); } +/** + * An off-host object → a source of its JSON as byte chunks, openable as often + * as needed. `decryptBackup` answers one string, which the JSON of a large + * record cannot be: 662 MB for an account of 1.25 million measurements, past + * the longest string V8 allows (#1031). The tag is verified whole on the + * compressed ciphertext before a source exists, exactly as `decryptBackup` + * does; only the decompression is streamed. + */ +export function openBackupObject( + buf: Buffer, + key: Buffer, +): () => AsyncIterable { + const magic = buf.subarray(0, 4).toString("binary"); + const version = buf[PREAMBLE_LENGTH - 1]; + if ( + magic !== MAGIC || + (version !== BACKUP_ENVELOPE_PLAIN && + version !== BACKUP_ENVELOPE_GZIP && + version !== BACKUP_ENVELOPE_STREAM) + ) { + throw new Error("Invalid backup envelope (bad magic or version)"); + } + let plaintext: Buffer; + if (version === BACKUP_ENVELOPE_STREAM) { + plaintext = decryptRawStream(buf.subarray(PREAMBLE_LENGTH), key); + } else { + const iv = buf.subarray(PREAMBLE_LENGTH, PREAMBLE_LENGTH + IV_LENGTH); + const tag = buf.subarray( + PREAMBLE_LENGTH + IV_LENGTH, + PREAMBLE_LENGTH + IV_LENGTH + TAG_LENGTH, + ); + const ct = buf.subarray(PREAMBLE_LENGTH + IV_LENGTH + TAG_LENGTH); + const dec = createDecipheriv(ALGORITHM, key, iv); + dec.setAuthTag(tag); + plaintext = Buffer.concat([dec.update(ct), dec.final()]); + } + if (version === BACKUP_ENVELOPE_PLAIN) { + return () => Readable.from([plaintext]); + } + return () => Readable.from([plaintext]).pipe(createGunzip()); +} + /** * How much of one object the upload holds at a time, and how many of those * windows are in flight. `@aws-sdk/lib-storage` buffers `partSize` bytes per diff --git a/src/lib/jobs/reminder/__tests__/backup-handlers.test.ts b/src/lib/jobs/reminder/__tests__/backup-handlers.test.ts index fe2a94b55..743eae2a6 100644 --- a/src/lib/jobs/reminder/__tests__/backup-handlers.test.ts +++ b/src/lib/jobs/reminder/__tests__/backup-handlers.test.ts @@ -34,7 +34,7 @@ vi.mock("@/lib/export/store-backup-blob", () => ({ out += chunk; }); await mocks.store(input, out); - return out.length; + return { id: "backup-row", bytes: out.length }; }, })); diff --git a/src/lib/jobs/reminder/backup-handlers.ts b/src/lib/jobs/reminder/backup-handlers.ts index 13c258894..41a114b1d 100644 --- a/src/lib/jobs/reminder/backup-handlers.ts +++ b/src/lib/jobs/reminder/backup-handlers.ts @@ -132,7 +132,7 @@ export async function handleDataBackup( // assembles into the row (`storeBackupBlob`). The stored copy used // to be built here as one string first, which on a large record // added several hundred megabytes to a container capped at 1 GB. - const storedBytes = await storeBackupBlob( + const { bytes: storedBytes } = await storeBackupBlob( prisma, { userId: user.id, type: "WEEKLY_AUTO" }, (write) => diff --git a/src/lib/jobs/restore-drill.ts b/src/lib/jobs/restore-drill.ts index 36056c56f..50450c630 100644 --- a/src/lib/jobs/restore-drill.ts +++ b/src/lib/jobs/restore-drill.ts @@ -22,8 +22,9 @@ * no other cron uses. */ import type { Job } from "pg-boss"; +import { BackupJsonError, scanBackupJson } from "@/lib/export/backup-json-scan"; import { - decryptBackup, + openBackupObject, getS3Client, loadOffhostConfig, OffhostBackupNotConfiguredError, @@ -103,19 +104,36 @@ export async function runRestoreDrill( const dateKey = BACKUP_KEY_PATTERN.exec(objectKey)![1]; const ciphertext = await s3.getObject(objectKey); - const plaintext = decryptBackup(ciphertext, cfg.encryptionKey); - const parsed: unknown = JSON.parse(plaintext); - - if (typeof parsed !== "object" || parsed === null) { - throw new Error( - `Restore drill: backup object "${objectKey}" decrypted and parsed but is not a JSON object.`, - ); + // Read as a stream, never as one string: the JSON of a large record is + // longer than any string V8 can hold (#1031). The bulk tables are counted, + // not kept. + const source = openBackupObject(ciphertext, cfg.encryptionKey); + let plaintextBytes = 0; + async function* counted() { + for await (const chunk of source()) { + plaintextBytes += chunk.byteLength; + yield chunk; + } } - const payload = parsed as Record; + let scanned; + try { + scanned = await scanBackupJson(counted(), { + streamKeys: new Set(["measurements", "intakeEvents", "moodEntries"]), + }); + } catch (err) { + if (err instanceof BackupJsonError) { + throw new Error( + `Restore drill: backup object "${objectKey}" decrypted but is not a JSON object: ${err.message}`, + ); + } + throw err; + } + const payload = scanned.document; + const streamedCounts = scanned.streamedCounts; if ( typeof payload.exportedAt !== "string" || typeof payload.userId !== "string" || - !Array.isArray(payload.measurements) + streamedCounts.measurements === undefined ) { throw new Error( `Restore drill: backup object "${objectKey}" parses but is missing core fields (exportedAt / userId / measurements).`, @@ -133,12 +151,12 @@ export async function runRestoreDrill( ageDays, stale: ageDays > MAX_BACKUP_AGE_DAYS, ciphertextBytes: ciphertext.length, - plaintextBytes: Buffer.byteLength(plaintext, "utf8"), + plaintextBytes, recordCounts: { - measurements: payload.measurements.length, + measurements: streamedCounts.measurements, medications: countArray(payload.medications), - intakeEvents: countArray(payload.intakeEvents), - moodEntries: countArray(payload.moodEntries), + intakeEvents: streamedCounts.intakeEvents ?? 0, + moodEntries: streamedCounts.moodEntries ?? 0, }, }; } diff --git a/src/lib/validations/backup.ts b/src/lib/validations/backup.ts index 9873bbfbc..c2c9fab6e 100644 --- a/src/lib/validations/backup.ts +++ b/src/lib/validations/backup.ts @@ -1773,6 +1773,16 @@ export const backupPayloadSchema = z export type BackupPayload = z.infer; +/** + * One element of `measurements`, validated on its own. The streamed readers + * (`src/lib/export/streamed-backup.ts`) check a large file's measurements + * one at a time with this, and the rest of the file with + * `backupPayloadSchema` over a document whose measurements are empty, which + * together is the same check `backupPayloadSchema` makes of the whole file. + */ +export const backupMeasurementSchema = measurementSchema; +export type BackupMeasurement = z.infer; + /** * Numeric counts of each backed-up record kind. Returned in the * upload + restore API responses so the admin sees what they diff --git a/tests/integration/large-account-jobs.test.ts b/tests/integration/large-account-jobs.test.ts index 9568c0235..c897ec973 100644 --- a/tests/integration/large-account-jobs.test.ts +++ b/tests/integration/large-account-jobs.test.ts @@ -214,7 +214,7 @@ describe("storeBackupBlob (#1031)", () => { it("stores a copy that reads back as exactly the JSON produced", async () => { const prisma = getPrismaClient(); - const bytes = await storeBackupBlob( + const { bytes } = await storeBackupBlob( prisma, { userId: RUN_A, type: "WEEKLY_AUTO" }, producer(20_000), diff --git a/tests/integration/large-backup-round-trip.test.ts b/tests/integration/large-backup-round-trip.test.ts new file mode 100644 index 000000000..61301ac42 --- /dev/null +++ b/tests/integration/large-backup-round-trip.test.ts @@ -0,0 +1,203 @@ +/** + * #1031 — an account too large to export or restore. + * + * For an account of 1.25 million measurements the full export took the app + * down, and its weekly backup could not be restored, downloaded or summarised + * at all: the backup's JSON (662 MB) is longer than any string V8 can hold, + * and every reader turned the file into one string first. Measured on a + * seeded account of that size in a 1 GB container, the streamed paths now + * export in 25 s, upload in under 20 s and restore in 94 s, and the restored + * rows match the file exactly. + * + * A million rows do not fit a test run. This file runs the same paths at a + * size that spans many pages and batches, and checks the rows come back + * exactly: the export streamed into the response, uploaded as gzip through + * the raw-body path, and restored through the streamed two-pass reader; and a + * weekly backup restored the same way, every column intact, with a personal + * record still pointing at the measurement it was found in. + */ +import { gzipSync } from "node:zlib"; + +import { beforeEach, describe, expect, it, vi } from "vitest"; + +process.env.ENCRYPTION_KEY ??= + "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; + +import { cookieJar, headerJar } from "./mock-next-headers"; +import { getPrismaClient, truncateAllTables } from "./setup"; +import { streamFullBackupJson } from "@/lib/export/full-backup-stream"; +import { storeBackupBlob } from "@/lib/export/store-backup-blob"; +import { encryptNote } from "@/lib/crypto/note-cipher"; + +vi.mock("next/headers", async () => { + const { cookieJar, headerJar } = await import("./mock-next-headers"); + return { + headers: vi.fn(async () => ({ + get: (name: string) => headerJar.get(name.toLowerCase()) ?? null, + })), + cookies: vi.fn(async () => ({ + get: (name: string) => { + const value = cookieJar.get(name); + return value ? { name, value } : undefined; + }, + set: (name: string, value: string) => { + cookieJar.set(name, value); + }, + delete: (name: string) => { + cookieJar.delete(name); + }, + })), + }; +}); +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +const COUNT = 7_300; // more than one read page and several write batches + +beforeEach(async () => { + await truncateAllTables(getPrismaClient()); + cookieJar.clear(); + headerJar.clear(); +}); + +async function seedAccount() { + const prisma = getPrismaClient(); + const user = await prisma.user.create({ + data: { + username: "large-account", + email: "large-account@example.test", + role: "ADMIN", + }, + }); + const session = await prisma.session.create({ + data: { userId: user.id, expiresAt: new Date(Date.now() + 600_000) }, + }); + cookieJar.set("healthlog_session", session.id); + + const start = Date.UTC(2024, 0, 1); + await prisma.measurement.createMany({ + data: Array.from({ length: COUNT }, (_, i) => ({ + id: `m-${String(i).padStart(5, "0")}`, + userId: user.id, + type: i % 3 === 0 ? ("PULSE" as const) : ("SLEEP_DURATION" as const), + value: 50 + (i % 40) + 0.25, + valueMax: i % 3 === 0 ? 120 : null, + unit: i % 3 === 0 ? "bpm" : "minutes", + source: "APPLE_HEALTH" as const, + measuredAt: new Date(start + i * 300_000), + externalId: `hk-${i}`, + sleepStage: i % 3 === 0 ? null : ("CORE" as const), + deviceType: "watch", + syncVersion: 1 + (i % 4), + notesEncrypted: i === 17 ? encryptNote("after the long run") : null, + deletedAt: i % 97 === 0 ? new Date(start) : null, + })), + }); + await prisma.personalRecord.create({ + data: { + userId: user.id, + metricType: "PULSE", + direction: "MAX", + value: 89.25, + unit: "bpm", + achievedAt: new Date(start + 3 * 300_000), + sourceMeasurementId: "m-00003", + }, + }); + return user; +} + +async function measurementsOf(userId: string) { + return getPrismaClient().measurement.findMany({ + where: { userId }, + orderBy: { id: "asc" }, + omit: { createdAt: true, updatedAt: true }, + }); +} + +async function restore(id: string) { + const { POST } = await import("@/app/api/admin/backups/[id]/restore/route"); + const res = await POST( + new Request(`http://localhost/api/admin/backups/${id}/restore`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ confirm: "RESTORE" }), + }) as never, + { params: Promise.resolve({ id }) }, + ); + return res; +} + +describe("large backups round-trip through the streamed paths (#1031)", () => { + it("export → gzip upload → restore gives back every live measurement", async () => { + const user = await seedAccount(); + const live = (await measurementsOf(user.id)).filter((m) => !m.deletedAt); + + const { GET } = await import("@/app/api/export/full-backup/route"); + const exported = await GET( + new Request("http://localhost/api/export/full-backup") as never, + ); + expect(exported.status).toBe(200); + const file = Buffer.from(await exported.arrayBuffer()); + expect(JSON.parse(file.toString("utf8")).measurements).toHaveLength( + live.length, + ); + + const { POST: upload } = + await import("@/app/api/admin/backups/upload/route"); + const uploaded = await upload( + new Request("http://localhost/api/admin/backups/upload", { + method: "POST", + headers: { "content-type": "application/gzip" }, + body: gzipSync(file), + }) as never, + ); + expect(uploaded.status).toBe(201); + const { data } = (await uploaded.json()) as { + data: { id: string; summary: { measurements: number } }; + }; + expect(data.summary.measurements).toBe(live.length); + + expect((await restore(data.id)).status).toBe(200); + const restored = await measurementsOf(user.id); + // A portable file carries the fields a person reads, not the sync + // bookkeeping, so those are what must come back. + const readable = (rows: typeof restored) => + rows.map((m) => ({ + id: m.id, + type: m.type, + value: m.value, + unit: m.unit, + measuredAt: m.measuredAt.toISOString(), + source: m.source, + })); + expect(readable(restored)).toEqual(readable(live)); + }); + + it("a weekly backup restores every column, and a record's pointer survives", async () => { + const user = await seedAccount(); + const before = await measurementsOf(user.id); + const prisma = getPrismaClient(); + + const { id } = await storeBackupBlob( + prisma, + { userId: user.id, type: "WEEKLY_AUTO" }, + (write) => + streamFullBackupJson(prisma, user.id, write, { + purpose: "disaster-recovery", + }), + ); + + // Change the account so the restore has something to undo. + await prisma.measurement.deleteMany({ where: { userId: user.id } }); + + const res = await restore(id); + expect(res.status).toBe(200); + expect(await measurementsOf(user.id)).toEqual(before); + const record = await prisma.personalRecord.findFirstOrThrow({ + where: { userId: user.id }, + }); + expect(record.sourceMeasurementId).toBe("m-00003"); + }); +}); diff --git a/tests/integration/measurement-bulk-insert.test.ts b/tests/integration/measurement-bulk-insert.test.ts new file mode 100644 index 000000000..c0caae7b1 --- /dev/null +++ b/tests/integration/measurement-bulk-insert.test.ts @@ -0,0 +1,139 @@ +/** + * `insertMeasurementRows` replaced `createMany` in the restore because + * Prisma's cached plan for a 1 000-row `createMany` kept about 145 MB alive + * (#1031). It writes raw SQL, so this pins that a row written through it + * reads back exactly as the same row written through Prisma, every column + * set, and every column null where it may be. + */ +import { beforeEach, describe, expect, it } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; +import { + insertMeasurementRows, + type MeasurementInsertRow, +} from "@/lib/export/measurement-bulk-insert"; + +const USER = "user-bulk-insert"; + +beforeEach(async () => { + const prisma = getPrismaClient(); + await truncateAllTables(prisma); + await prisma.user.create({ + data: { id: USER, username: USER, email: `${USER}@example.test` }, + }); +}); + +function full(id: string, at: string): MeasurementInsertRow { + return { + id, + userId: USER, + type: "SLEEP_DURATION", + value: 47.25, + valueMin: 1.5, + valueMax: 99.75, + unit: "minutes", + source: "APPLE_HEALTH", + measuredAt: new Date(at), + notes: "legacy note", + notesEncrypted: new Uint8Array([0, 1, 2, 254, 255]), + externalId: `ext-${id}`, + externalSourceVersion: "17.4", + aggregationProvenance: "HEALTHKIT_STATISTICS", + glucoseContext: null, + sleepStage: "REM", + rhythmClassification: null, + deviceType: "watch", + syncVersion: 7, + deletedAt: new Date("2026-02-03T04:05:06.789Z"), + createdAt: new Date("2025-01-02T03:04:05.678Z"), + updatedAt: new Date("2025-06-07T08:09:10.111Z"), + }; +} + +function sparse(id: string, at: string): MeasurementInsertRow { + return { + id, + userId: USER, + type: "BLOOD_GLUCOSE", + value: 5.4, + valueMin: null, + valueMax: null, + unit: "mmol/L", + source: "MANUAL", + measuredAt: new Date(at), + notes: null, + notesEncrypted: null, + externalId: null, + externalSourceVersion: null, + aggregationProvenance: null, + glucoseContext: "FASTING", + sleepStage: null, + rhythmClassification: null, + deviceType: null, + syncVersion: 1, + deletedAt: null, + }; +} + +describe("insertMeasurementRows", () => { + it("writes every column exactly as Prisma create does", async () => { + const prisma = getPrismaClient(); + const rows = [ + [ + full("a-bulk", "2026-01-01T10:00:00.123Z"), + full("a-prisma", "2026-01-02T10:00:00.123Z"), + ], + [ + sparse("b-bulk", "2026-01-03T10:00:00.000Z"), + sparse("b-prisma", "2026-01-04T10:00:00.000Z"), + ], + ] as const; + + for (const [viaBulk, viaPrisma] of rows) { + await prisma.$transaction((tx) => insertMeasurementRows(tx, [viaBulk])); + await prisma.measurement.create({ data: viaPrisma as never }); + } + + for (const [viaBulk, viaPrisma] of rows) { + const a = await prisma.measurement.findUniqueOrThrow({ + where: { id: viaBulk.id }, + }); + const b = await prisma.measurement.findUniqueOrThrow({ + where: { id: viaPrisma.id }, + }); + const strip = (row: typeof a) => ({ + ...row, + id: undefined, + externalId: row.externalId?.replace(/-(bulk|prisma)$/, ""), + measuredAt: undefined, + // Filled with the current time when the row carries none. + ...(viaBulk.createdAt + ? {} + : { createdAt: undefined, updatedAt: undefined }), + }); + expect(strip(a)).toEqual(strip(b)); + expect(a.measuredAt.toISOString()).toBe(viaBulk.measuredAt.toISOString()); + } + const sparseRow = await prisma.measurement.findUniqueOrThrow({ + where: { id: "b-bulk" }, + }); + expect(Date.now() - sparseRow.createdAt.getTime()).toBeLessThan(60_000); + }); + + it("inserts a whole batch in one statement and reports the count", async () => { + const prisma = getPrismaClient(); + const batch = Array.from({ length: 2_500 }, (_, i) => + sparse( + `m-${i}`, + new Date(Date.UTC(2026, 0, 1) + i * 60_000).toISOString(), + ), + ); + const inserted = await prisma.$transaction((tx) => + insertMeasurementRows(tx, batch), + ); + expect(inserted).toBe(2_500); + expect(await prisma.measurement.count({ where: { userId: USER } })).toBe( + 2_500, + ); + }); +});