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",