From f8c17ee9bb28029518f0a3e160a1d43a2566bbe6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Wed, 21 Oct 2026 23:49:23 +0200 Subject: [PATCH] fix(import): write lab CSV rows one chunk at a time and audit cut-off imports Every accepted row was its own insert, so a 20 000-row file meant 20 000 round trips and an import cut off part way left an arbitrary prefix behind. Each 200-row chunk is now one createMany: it lands whole or not at all, and uploading the same file again skips the written chunks as duplicates. The importers report each committed write, and the audit row is written in a finally block, marked partial with the rows written when the import did not finish. --- .../__tests__/record-import-route.test.ts | 110 ++++++++ src/lib/import/labs-csv.ts | 57 +++-- src/lib/import/medications-csv.ts | 234 +++++++++--------- src/lib/import/record-import-route.ts | 44 ++-- tests/integration/record-csv-import.test.ts | 91 +++++++ 5 files changed, 387 insertions(+), 149 deletions(-) create mode 100644 src/lib/import/__tests__/record-import-route.test.ts diff --git a/src/lib/import/__tests__/record-import-route.test.ts b/src/lib/import/__tests__/record-import-route.test.ts new file mode 100644 index 000000000..33b74f93b --- /dev/null +++ b/src/lib/import/__tests__/record-import-route.test.ts @@ -0,0 +1,110 @@ +/** + * The shared CSV record-import request side: the audit row is written for a + * finished import and for one cut off part way, with how many rows landed. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { z } from "zod/v4"; + +const { auditLog } = vi.hoisted(() => ({ auditLog: vi.fn() })); + +vi.mock("@/lib/api-handler", () => ({ + requireAuth: vi.fn(async () => ({ user: { id: "u1" } })), +})); +vi.mock("@/lib/modules/gate", () => ({ + requireModuleEnabled: vi.fn(async () => ({ enabled: true })), +})); +vi.mock("@/lib/rate-limit", () => ({ + checkRateLimit: vi.fn(async () => ({ allowed: true })), +})); +vi.mock("@/lib/tz/resolver", () => ({ + resolveUserTimezone: vi.fn(async () => "UTC"), +})); +vi.mock("@/lib/auth/audit", () => ({ auditLog })); +vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn() })); + +import { handleRecordImport } from "../record-import-route"; +import type { RecordImportResult } from "@/lib/validations/record-import"; + +const schema = z + .object({ dryRun: z.enum(["0", "1"]).transform((v) => v === "1") }) + .strict(); + +function request(dryRun = "0") { + return new Request(`http://localhost/api/import/labs-csv?dryRun=${dryRun}`, { + method: "POST", + headers: { "content-type": "text/csv" }, + body: "a,b\n1,2\n", + }) as never; +} + +const DONE = { + kind: "labs", + dryRun: false, + rows: 400, + valid: 400, + inserted: 400, + duplicates: 0, + failed: 0, +} as unknown as RecordImportResult; + +beforeEach(() => auditLog.mockReset()); + +describe("handleRecordImport audit", () => { + it("audits a finished import as complete", async () => { + await handleRecordImport(request(), { + kind: "labs", + module: "labs", + schema, + run: async ({ onWritten }) => { + onWritten(200); + onWritten(200); + return DONE; + }, + }); + expect(auditLog).toHaveBeenCalledTimes(1); + expect(auditLog.mock.calls[0][1].details).toMatchObject({ + inserted: 400, + partial: false, + written: 400, + }); + }); + + it("audits an import cut off part way with what it wrote", async () => { + await expect( + handleRecordImport(request(), { + kind: "labs", + module: "labs", + schema, + run: async ({ onWritten }) => { + onWritten(200); + throw new Error("client went away"); + }, + }), + ).rejects.toThrow("client went away"); + expect(auditLog).toHaveBeenCalledTimes(1); + expect(auditLog.mock.calls[0][1].details).toEqual({ + partial: true, + written: 200, + }); + }); + + it("writes no audit row for a dry run or an abort that wrote nothing", async () => { + await handleRecordImport(request("1"), { + kind: "labs", + module: "labs", + schema, + run: async () => ({ ...DONE, dryRun: true }), + }); + await expect( + handleRecordImport(request(), { + kind: "labs", + module: "labs", + schema, + run: async () => { + throw new Error("boom"); + }, + }), + ).rejects.toThrow("boom"); + expect(auditLog).not.toHaveBeenCalled(); + }); +}); diff --git a/src/lib/import/labs-csv.ts b/src/lib/import/labs-csv.ts index 3c874dbfe..e3308503a 100644 --- a/src/lib/import/labs-csv.ts +++ b/src/lib/import/labs-csv.ts @@ -13,7 +13,14 @@ * Memory: the file is read record by record (`readCsvRecords`) and written in * chunks of `CHUNK_ROWS`; what is held across chunks is the marker index, the * duplicate keys of rows already accepted and the bounded report. + * + * Writes: one `createMany` per chunk, so a chunk lands whole or not at all. + * An import cut off part way (the client went away, a proxy timed out) leaves + * whole chunks behind, and uploading the same file again skips them as + * duplicates and writes the rest: a retry is idempotent. `onWritten` reports + * each committed chunk so the caller can audit a partial import. */ +import type { Prisma } from "@/generated/prisma/client"; import { prisma } from "@/lib/db"; import { readCsvRecords, @@ -272,6 +279,8 @@ export async function importLabsCsv(args: { maxBytes: number; query: LabsCsvQuery; now?: Date; + /** Called after each committed chunk with the rows it wrote. */ + onWritten?: (rows: number) => void; }): Promise { const { userId, tz, query } = args; const dryRun = query.dryRun; @@ -411,7 +420,9 @@ export async function importLabsCsv(args: { } } - // 3. Write (or count) what is new. + // 3. Write (or count) what is new: rows are built here and written in + // one statement for the chunk. + const toWrite: Prisma.LabResultCreateManyInput[] = []; for (const { row, marker } of planned) { const probe = { biomarkerId: @@ -489,26 +500,24 @@ export async function importLabsCsv(args: { } // Field by field, never a spread of the parsed row. - await prisma.labResult.create({ - data: { - userId, - biomarkerId: biomarker.id, - panel: row.panel ?? biomarker.panel, - analyte: biomarker.name, - value: storedValue, - valueText: row.valueText, - unit: biomarker.unit, - sourceValue: convertedFrom?.sourceValue ?? null, - sourceUnit: convertedFrom?.sourceUnit ?? null, - referenceLow: biomarker.lowerBound, - referenceHigh: biomarker.upperBound, - sourceReferenceLow: storedLow, - sourceReferenceHigh: storedHigh, - sourceReferenceText: row.sourceReferenceText, - takenAt: row.takenAt, - source: "IMPORT", - noteEncrypted: row.note ? encryptNoteToBytes(row.note) : null, - }, + toWrite.push({ + userId, + biomarkerId: biomarker.id, + panel: row.panel ?? biomarker.panel, + analyte: biomarker.name, + value: storedValue, + valueText: row.valueText, + unit: biomarker.unit, + sourceValue: convertedFrom?.sourceValue ?? null, + sourceUnit: convertedFrom?.sourceUnit ?? null, + referenceLow: biomarker.lowerBound, + referenceHigh: biomarker.upperBound, + sourceReferenceLow: storedLow, + sourceReferenceHigh: storedHigh, + sourceReferenceText: row.sourceReferenceText, + takenAt: row.takenAt, + source: "IMPORT", + noteEncrypted: row.note ? encryptNoteToBytes(row.note) : null, }); for (const k of [ ...keys, @@ -516,7 +525,11 @@ export async function importLabsCsv(args: { ]) seen.add(k); report.valid += 1; - report.inserted += 1; + } + if (toWrite.length > 0) { + const { count } = await prisma.labResult.createMany({ data: toWrite }); + report.inserted += count; + args.onWritten?.(count); } }; diff --git a/src/lib/import/medications-csv.ts b/src/lib/import/medications-csv.ts index d616ce675..354bfab13 100644 --- a/src/lib/import/medications-csv.ts +++ b/src/lib/import/medications-csv.ts @@ -185,6 +185,8 @@ export async function importMedicationsCsv(args: { maxBytes: number; query: MedicationsCsvQuery; now?: Date; + /** Called after each medication written. */ + onWritten?: (rows: number) => void; }): Promise { const { userId, tz, query } = args; const dryRun = query.dryRun; @@ -207,125 +209,131 @@ export async function importMedicationsCsv(args: { ? "\t" : "auto"; - let first = true; - for await (const record of readCsvRecords(args.stream, { - maxBytes: args.maxBytes, - delimiter, - })) { - if (first && query.header) { + let written = 0; + try { + let first = true; + for await (const record of readCsvRecords(args.stream, { + maxBytes: args.maxBytes, + delimiter, + })) { + if (first && query.header) { + first = false; + continue; + } first = false; - continue; - } - first = false; - report.rows += 1; - if (report.rows > MEDICATIONS_CSV_MAX_ROWS) { - report.fail(record.line, "row_limit"); - continue; - } - if (record.oversized) { - report.fail(record.line, "oversized_row"); - continue; - } - const result = normaliseMedicationRow(record, query, timezone); - if (!result.ok) { - report.fail(record.line, result.reason, result.field); - continue; - } - const row = result.row; - const parsed = createMedicationSchema.safeParse( - medicationCreateBody(row, query.reminders), - ); - if (!parsed.success) { - report.fail(row.line, "invalid_medication"); - continue; - } - // The course the create route would open for this window. - let course: { startsOn: Date; endsOn: Date | null } | null = null; - if (parsed.data.startsOn || parsed.data.endsOn) { - const endKey = parsed.data.endsOn - ? dayKeyOfDate(parsed.data.endsOn) - : null; - course = { - startsOn: - parsed.data.startsOn ?? - dateOfDayKey(endKey && endKey < todayKey ? endKey : todayKey), - endsOn: parsed.data.endsOn ?? null, - }; - if (validateCourses([course], todayKey, false)) { - report.fail(row.line, "end_before_start", "endsOn"); + report.rows += 1; + if (report.rows > MEDICATIONS_CSV_MAX_ROWS) { + report.fail(record.line, "row_limit"); + continue; + } + if (record.oversized) { + report.fail(record.line, "oversized_row"); + continue; + } + const result = normaliseMedicationRow(record, query, timezone); + if (!result.ok) { + report.fail(record.line, result.reason, result.field); + continue; + } + const row = result.row; + const parsed = createMedicationSchema.safeParse( + medicationCreateBody(row, query.reminders), + ); + if (!parsed.success) { + report.fail(row.line, "invalid_medication"); + continue; + } + // The course the create route would open for this window. + let course: { startsOn: Date; endsOn: Date | null } | null = null; + if (parsed.data.startsOn || parsed.data.endsOn) { + const endKey = parsed.data.endsOn + ? dayKeyOfDate(parsed.data.endsOn) + : null; + course = { + startsOn: + parsed.data.startsOn ?? + dateOfDayKey(endKey && endKey < todayKey ? endKey : todayKey), + endsOn: parsed.data.endsOn ?? null, + }; + if (validateCourses([course], todayKey, false)) { + report.fail(row.line, "end_before_start", "endsOn"); + continue; + } + } + const key = identity(row.name, row.dose); + if (seen.has(key)) { + report.duplicate(row.line); + continue; + } + seen.add(key); + report.valid += 1; + report.inserted += 1; + if (dryRun) { + report.addPreview({ + line: row.line, + name: row.name, + dose: row.dose, + times: row.times.join(" "), + asNeeded: row.asNeeded, + startsOn: row.startsOn, + endsOn: row.endsOn, + }); continue; } - } - const key = identity(row.name, row.dose); - if (seen.has(key)) { - report.duplicate(row.line); - continue; - } - seen.add(key); - report.valid += 1; - report.inserted += 1; - if (dryRun) { - report.addPreview({ - line: row.line, - name: row.name, - dose: row.dose, - times: row.times.join(" "), - asNeeded: row.asNeeded, - startsOn: row.startsOn, - endsOn: row.endsOn, - }); - continue; - } - const data = parsed.data; - // Field by field, with the create route's invariants for the shapes an - // import produces: a daily schedule (FREQ=DAILY, times dual-written), an - // opening course for a dated window, no schedule for as-needed. - const medication = await prisma.medication.create({ - data: { - userId, - name: data.name, - dose: data.dose, - asNeeded: data.asNeeded ?? false, - notificationsEnabled: data.notificationsEnabled ?? false, - ...(course - ? { - startsOn: course.startsOn, - endsOn: course.endsOn, - courses: { - create: { - userId, - startsOn: course.startsOn, - endsOn: course.endsOn, + const data = parsed.data; + // Field by field, with the create route's invariants for the shapes an + // import produces: a daily schedule (FREQ=DAILY, times dual-written), an + // opening course for a dated window, no schedule for as-needed. + const medication = await prisma.medication.create({ + data: { + userId, + name: data.name, + dose: data.dose, + asNeeded: data.asNeeded ?? false, + notificationsEnabled: data.notificationsEnabled ?? false, + ...(course + ? { + startsOn: course.startsOn, + endsOn: course.endsOn, + courses: { + create: { + userId, + startsOn: course.startsOn, + endsOn: course.endsOn, + }, }, - }, - } - : {}), - schedules: { - create: (data.schedules ?? []).map((s) => ({ - windowStart: s.windowStart, - windowEnd: s.windowEnd, - label: null, - dose: null, - daysOfWeek: serializeScheduleRecurrence({ - daysOfWeek: [], - intervalWeeks: 1, - }), - timesOfDay: - s.timesOfDay && s.timesOfDay.length > 0 - ? s.timesOfDay - : [s.windowStart], - rrule: "FREQ=DAILY", - })), + } + : {}), + schedules: { + create: (data.schedules ?? []).map((s) => ({ + windowStart: s.windowStart, + windowEnd: s.windowEnd, + label: null, + dose: null, + daysOfWeek: serializeScheduleRecurrence({ + daysOfWeek: [], + intervalWeeks: 1, + }), + timesOfDay: + s.timesOfDay && s.timesOfDay.length > 0 + ? s.timesOfDay + : [s.windowStart], + rrule: "FREQ=DAILY", + })), + }, }, - }, - select: { id: true }, - }); - await setMedicationCategory(medication.id, undefined); - } - - if (!dryRun && report.inserted > 0) { - invalidateUserMedications(userId, { evict: true }); + select: { id: true }, + }); + await setMedicationCategory(medication.id, undefined); + written += 1; + args.onWritten?.(1); + } + } finally { + // Also after an import cut off part way: what was written is visible. + if (!dryRun && written > 0) { + invalidateUserMedications(userId, { evict: true }); + } } return report.toResult(); } diff --git a/src/lib/import/record-import-route.ts b/src/lib/import/record-import-route.ts index ded29bc6e..8035becd4 100644 --- a/src/lib/import/record-import-route.ts +++ b/src/lib/import/record-import-route.ts @@ -47,6 +47,7 @@ export async function handleRecordImport< stream: ReadableStream; maxBytes: number; query: z.infer; + onWritten: (rows: number) => void; }) => Promise; }, ): Promise { @@ -102,7 +103,8 @@ export async function handleRecordImport< } const tz = await resolveUserTimezone(user.id); - let result: RecordImportResult; + let result: RecordImportResult | null = null; + let written = 0; try { result = await opts.run({ userId: user.id, @@ -110,6 +112,9 @@ export async function handleRecordImport< stream: request.body, maxBytes: RECORD_IMPORT_MAX_BYTES, query, + onWritten: (rows) => { + written += rows; + }, }); } catch (err) { if (err instanceof CsvTooLargeError) { @@ -118,25 +123,36 @@ export async function handleRecordImport< }); } throw err; + } finally { + // The audit row is written whatever happened: an import cut off part way + // (the client went away, a proxy timed out, the cap was hit after some + // chunks) has already written rows, and the trail says how many. + if ( + !query.dryRun && + ((result !== null && result.rows > 0) || written > 0) + ) { + await auditLog(action, { + userId: user.id, + ipAddress: getClientIp(request), + details: + result !== null + ? { + rows: result.rows, + inserted: result.inserted, + duplicates: result.duplicates, + failed: result.failed, + partial: false, + written, + } + : { partial: true, written }, + }); + } } if (result.rows === 0) { return apiError("The CSV file has no data rows", 422, { errorCode: "import.csv.empty", }); } - - if (!query.dryRun) { - await auditLog(action, { - userId: user.id, - ipAddress: getClientIp(request), - details: { - rows: result.rows, - inserted: result.inserted, - duplicates: result.duplicates, - failed: result.failed, - }, - }); - } annotate({ meta: { import_dry_run: query.dryRun, diff --git a/tests/integration/record-csv-import.test.ts b/tests/integration/record-csv-import.test.ts index 7164cab39..7e28273ab 100644 --- a/tests/integration/record-csv-import.test.ts +++ b/tests/integration/record-csv-import.test.ts @@ -203,6 +203,97 @@ describe("lab CSV import in a zone of the file's own", () => { }); }); +describe("lab CSV import at the row cap", () => { + const BIG_USER = "csv-import-big-user"; + // 20 000 rows: 40 markers, 500 draw days each, one reading per day. + const lines = ["Analyt;Wert;Einheit;Datum"]; + for (let i = 0; i < 20_000; i++) { + const marker = `Marker ${i % 40}`; + const day = new Date(Date.UTC(2020, 0, 1) + Math.floor(i / 40) * 86_400_000) + .toISOString() + .slice(0, 10); + lines.push(`${marker};${(i % 97) + 1};U/l;${day}`); + } + const BIG = lines.join("\n"); + const query = () => + labsCsvQuerySchema.parse({ + analyte: "0", + value: "1", + unit: "2", + date: "3", + }); + + /** The file, failing like a dropped connection after `bytes`. */ + function cutAt(bytes: number): ReadableStream { + const all = new TextEncoder().encode(BIG); + let offset = 0; + return new ReadableStream({ + pull(controller) { + if (offset >= bytes) { + controller.error(new Error("connection reset")); + return; + } + controller.enqueue(all.slice(offset, Math.min(offset + 64_000, bytes))); + offset += 64_000; + }, + }); + } + + beforeAll(async () => { + await prisma.user.create({ + data: { id: BIG_USER, username: BIG_USER, timezone: TZ }, + }); + }); + + it("an import cut off part way leaves whole chunks, and a retry writes the rest once", async () => { + let written = 0; + await expect( + importLabsCsv({ + userId: BIG_USER, + tz: TZ, + stream: cutAt(Math.floor(BIG.length / 3)), + maxBytes: 16 * 1024 * 1024, + query: query(), + onWritten: (n) => { + written += n; + }, + }), + ).rejects.toThrow("connection reset"); + const afterCut = await prisma.labResult.count({ + where: { userId: BIG_USER }, + }); + expect(afterCut).toBe(written); + expect(afterCut).toBeGreaterThan(0); + // Whole chunks only: nothing half-written. + expect(afterCut % 200).toBe(0); + + const started = Date.now(); + const result = await importLabsCsv({ + userId: BIG_USER, + tz: TZ, + stream: textToStream(BIG), + maxBytes: 16 * 1024 * 1024, + query: query(), + }); + const elapsed = Date.now() - started; + expect(result).toMatchObject({ + rows: 20_000, + inserted: 20_000 - afterCut, + duplicates: afterCut, + failed: 0, + }); + expect(await prisma.labResult.count({ where: { userId: BIG_USER } })).toBe( + 20_000, + ); + expect(await prisma.biomarker.count({ where: { userId: BIG_USER } })).toBe( + 40, + ); + // Generous: the chunked write takes a few seconds here; the bound only + // catches a return to one statement per row. + expect(elapsed).toBeLessThan(120_000); + }, 300_000); +}); + describe("medication CSV import", () => { const MEDS = [ "name,dose,times,as needed,start,end",