diff --git a/src/app/api/import/csv/__tests__/route.test.ts b/src/app/api/import/csv/__tests__/route.test.ts index b663f67c3..a1bb577a0 100644 --- a/src/app/api/import/csv/__tests__/route.test.ts +++ b/src/app/api/import/csv/__tests__/route.test.ts @@ -8,11 +8,23 @@ vi.mock("@/lib/api-handler", () => ({ })), })); -vi.mock("@/lib/db", () => ({ - prisma: { - measurement: { create: vi.fn(), upsert: vi.fn() }, - }, -})); +vi.mock("@/lib/db", () => { + const measurement = { + findMany: vi.fn(), + createMany: vi.fn(), + updateMany: vi.fn(), + }; + return { + prisma: { + measurement, + // Run the batched write callback against the same measurement mock so + // createMany / updateMany calls are observable. + $transaction: vi.fn(async (fn: (tx: unknown) => unknown) => + fn({ measurement }), + ), + }, + }; +}); vi.mock("@/lib/rate-limit", () => ({ checkRateLimit: vi.fn(), @@ -83,6 +95,23 @@ beforeEach(() => { remaining: 4, resetAt: new Date(Date.now() + 1000), } as never); + // Default: nothing pre-exists; the transaction runner is re-stubbed by the + // db mock factory (resetAllMocks clears its implementation), so restore it. + const measurement = prisma.measurement as unknown as { + findMany: ReturnType; + createMany: ReturnType; + updateMany: ReturnType; + }; + measurement.findMany.mockResolvedValue([]); + measurement.createMany.mockResolvedValue({ count: 0 }); + measurement.updateMany.mockResolvedValue({ count: 0 }); + ( + prisma.$transaction as unknown as { + mockImplementation: (f: unknown) => void; + } + ).mockImplementation(async (fn: (tx: unknown) => unknown) => + fn({ measurement }), + ); }); describe("POST /api/import/csv — rate limit", () => { @@ -103,18 +132,23 @@ describe("POST /api/import/csv — rate limit", () => { }); }); +const mMeasurement = () => + prisma.measurement as unknown as { + findMany: ReturnType; + createMany: ReturnType; + updateMany: ReturnType; + }; + describe("POST /api/import/csv — fatal header error", () => { it("returns 422 when a required column is missing", async () => { const res = await POST(csvRequest("type,value,unit\nWEIGHT,80,kg")); expect(res.status).toBe(422); - expect(vi.mocked(prisma.measurement.create)).not.toHaveBeenCalled(); + expect(mMeasurement().createMany).not.toHaveBeenCalled(); }); }); -describe("POST /api/import/csv — write loop + per-row envelope", () => { +describe("POST /api/import/csv — batched write + per-row envelope", () => { it("inserts valid rows, skips invalid ones, returns per-row status", async () => { - vi.mocked(prisma.measurement.create).mockResolvedValue({} as never); - const res = await POST( csvRequest( [ @@ -135,12 +169,29 @@ describe("POST /api/import/csv — write loop + per-row envelope", () => { const reasons = body.data?.rows.map((r) => r.reason); expect(reasons).toContain("unknown_type"); expect(reasons).toContain("missing_timezone_offset"); - // One create call for the single valid row. - expect(vi.mocked(prisma.measurement.create)).toHaveBeenCalledTimes(1); + // One bulk createMany carrying the single valid survivor. + expect(mMeasurement().createMany).toHaveBeenCalledTimes(1); + const arg = mMeasurement().createMany.mock.calls[0][0]; + expect(arg.data).toHaveLength(1); // Rollup re-fold fired for the touched (type, day). expect(vi.mocked(recomputeBucketsForMeasurement)).toHaveBeenCalled(); }); + it("batches many rows into a single createMany (no per-row round-trip)", async () => { + const rows = [HEADER]; + for (let i = 0; i < 50; i++) { + const mm = String(i).padStart(2, "0"); // unique minute → unique measuredAt + rows.push(`WEIGHT,${80 + i * 0.1},kg,2026-05-01T08:${mm}:00Z,,,`); + } + const res = await POST(csvRequest(rows.join("\n"))); + expect(res.status).toBe(200); + const body = (await res.json()) as CsvEnvelope; + expect(body.data?.inserted).toBe(50); + // 50 rows, one createMany call — the whole point of the batch. + expect(mMeasurement().createMany).toHaveBeenCalledTimes(1); + expect(mMeasurement().createMany.mock.calls[0][0].data).toHaveLength(50); + }); + it("dryRun previews without writing and reports projected inserts", async () => { const res = await POST( csvRequest( @@ -153,17 +204,14 @@ describe("POST /api/import/csv — write loop + per-row envelope", () => { expect(body.data?.dryRun).toBe(true); expect(body.data?.inserted).toBe(1); expect(body.data?.rows[0].status).toBe("inserted"); - expect(vi.mocked(prisma.measurement.create)).not.toHaveBeenCalled(); + expect(mMeasurement().createMany).not.toHaveBeenCalled(); expect(vi.mocked(recomputeBucketsForMeasurement)).not.toHaveBeenCalled(); }); - it("upserts an externalId row and surfaces it as updated when it already existed", async () => { - const created = new Date("2026-05-01T08:00:00Z"); - const bumped = new Date("2026-05-02T08:00:00Z"); - vi.mocked(prisma.measurement.upsert).mockResolvedValue({ - createdAt: created, - updatedAt: bumped, - } as never); + it("surfaces an externalId row as updated when it already existed (updateMany)", async () => { + mMeasurement().findMany.mockResolvedValue([ + { type: "WEIGHT", externalId: "ext-1" }, + ]); const res = await POST( csvRequest( @@ -174,13 +222,28 @@ describe("POST /api/import/csv — write loop + per-row envelope", () => { expect(body.data?.updated).toBe(1); expect(body.data?.inserted).toBe(0); expect(body.data?.rows[0].status).toBe("updated"); - expect(vi.mocked(prisma.measurement.upsert)).toHaveBeenCalledTimes(1); + expect(mMeasurement().updateMany).toHaveBeenCalledTimes(1); + expect(mMeasurement().createMany).not.toHaveBeenCalled(); }); - it("counts a unique-constraint duplicate as skipped/duplicate", async () => { - vi.mocked(prisma.measurement.create).mockRejectedValue( - new Error("Unique constraint failed"), + it("inserts an externalId row when it does not exist yet", async () => { + mMeasurement().findMany.mockResolvedValue([]); + const res = await POST( + csvRequest( + [HEADER, "WEIGHT,80.5,kg,2026-05-01T08:00:00Z,,,ext-2"].join("\n"), + ), ); + const body = (await res.json()) as CsvEnvelope; + expect(body.data?.inserted).toBe(1); + expect(body.data?.updated).toBe(0); + expect(body.data?.rows[0].status).toBe("inserted"); + expect(mMeasurement().createMany).toHaveBeenCalledTimes(1); + }); + + it("counts a pre-existing natural-key row as skipped/duplicate", async () => { + mMeasurement().findMany.mockResolvedValue([ + { type: "WEIGHT", measuredAt: new Date("2026-05-01T08:00:00Z") }, + ]); const res = await POST( csvRequest([HEADER, "WEIGHT,80.5,kg,2026-05-01T08:00:00Z,,,"].join("\n")), @@ -192,5 +255,23 @@ describe("POST /api/import/csv — write loop + per-row envelope", () => { status: "skipped", reason: "duplicate", }); + expect(mMeasurement().createMany).not.toHaveBeenCalled(); + }); + + it("collapses an in-file duplicate (same type+measuredAt) to one insert + one duplicate", async () => { + const res = await POST( + csvRequest( + [ + HEADER, + "WEIGHT,80.5,kg,2026-05-01T08:00:00Z,,,", + "WEIGHT,80.6,kg,2026-05-01T08:00:00Z,,,", + ].join("\n"), + ), + ); + const body = (await res.json()) as CsvEnvelope; + expect(body.data?.inserted).toBe(1); + expect(body.data?.skipped).toBe(1); + // Only the first survivor reaches the bulk insert. + expect(mMeasurement().createMany.mock.calls[0][0].data).toHaveLength(1); }); }); diff --git a/src/app/api/import/csv/route.ts b/src/app/api/import/csv/route.ts index ae6a53de9..7fdd868d7 100644 --- a/src/app/api/import/csv/route.ts +++ b/src/app/api/import/csv/route.ts @@ -19,6 +19,8 @@ import type { MeasurementType, GlucoseContext, } from "@/generated/prisma/client"; +import { Prisma } from "@/generated/prisma/client"; +import type { NormalisedMeasurementRow } from "@/lib/import/csv-measurements"; /** * CSV measurement import (v1.17.1). @@ -106,84 +108,165 @@ export const POST = apiHandler(async (request: NextRequest) => { const writeOutcome = new Map(); if (!dryRun) { - for (const result of okRows) { - const m = result.row; - try { - if (m.externalId) { - // Idempotent re-import keyed on the source-stable id — re-uploading - // the same file updates in place rather than minting duplicates. - const res = await prisma.measurement.upsert({ + // Batched write. The previous shape ran one serial `upsert` (or `create`) + // per accepted row — N round-trips for an N-row file, which on a + // few-thousand-row import approaches the request timeout. This adopts the + // fast shape `/measurements/batch` uses: pre-fetch the existing rows under + // the IMPORT source in one query per dedup-key shape, partition into bulk + // insert / update sets, then `createMany` + grouped `updateMany`. The + // per-line inserted/updated/duplicate envelope is reconstructed from the + // partition, so the response contract is unchanged. + + const buildCreateData = ( + m: NormalisedMeasurementRow, + ): Prisma.MeasurementCreateManyInput => ({ + userId, + type: m.type as MeasurementType, + value: m.value, + unit: m.unit, + source: "IMPORT", + ...(m.externalId ? { externalId: m.externalId } : {}), + measuredAt: m.measuredAt, + notes: null, + notesEncrypted: encryptNote(m.notes ?? null), + glucoseContext: (m.glucoseContext as GlucoseContext | undefined) ?? null, + }); + + const extRows = okRows.filter((r) => r.row.externalId); + const plainRows = okRows.filter((r) => !r.row.externalId); + + // --- externalId rows: idempotent upsert on + // (userId, type, source=IMPORT, externalId). One probe fetches every + // pre-existing key; a hit means the row updates in place. + const extExisting = + extRows.length > 0 + ? await prisma.measurement.findMany({ where: { - userId_type_source_externalId: { - userId, - type: m.type as MeasurementType, - source: "IMPORT", - externalId: m.externalId, - }, - }, - update: { - value: m.value, - unit: m.unit, - measuredAt: m.measuredAt, - notes: null, - notesEncrypted: encryptNote(m.notes ?? null), - glucoseContext: - (m.glucoseContext as GlucoseContext | undefined) ?? null, - }, - create: { userId, - type: m.type as MeasurementType, - value: m.value, - unit: m.unit, source: "IMPORT", - externalId: m.externalId, - measuredAt: m.measuredAt, - notes: null, - notesEncrypted: encryptNote(m.notes ?? null), - glucoseContext: - (m.glucoseContext as GlucoseContext | undefined) ?? null, + OR: extRows.map((r) => ({ + type: r.row.type as MeasurementType, + externalId: r.row.externalId as string, + })), }, - select: { createdAt: true, updatedAt: true }, - }); - // A fresh create has createdAt === updatedAt; an in-place update - // bumps updatedAt past createdAt. - if (res.updatedAt.getTime() > res.createdAt.getTime()) { - updated++; - writeOutcome.set(result.line, "updated"); - } else { - inserted++; - writeOutcome.set(result.line, "inserted"); - } - } else { - await prisma.measurement.create({ - data: { + select: { type: true, externalId: true }, + }) + : []; + const extExistingSet = new Set( + extExisting.map((e) => `${e.type}::${e.externalId}`), + ); + + // Partition in file order, honouring in-file repeats: the first sighting + // of a brand-new key inserts, every later sighting (or any sighting of a + // key already in the DB) updates in place. The last value for a key wins, + // matching the old sequential-upsert overwrite. Insert/update maps are + // keyed by (type, externalId) so a repeated key collapses to one write. + const seenExt = new Set(); + const extInserts = new Map(); + const extUpdates = new Map(); + for (const result of extRows) { + const m = result.row; + const key = `${m.type}::${m.externalId}`; + if (extExistingSet.has(key)) { + extUpdates.set(key, m); + updated++; + writeOutcome.set(result.line, "updated"); + } else if (seenExt.has(key)) { + // New-to-DB key seen again in this file: still a single create with + // the latest value, but this line reads as an update just as the + // sequential upsert did. + extInserts.set(key, m); + updated++; + writeOutcome.set(result.line, "updated"); + } else { + seenExt.add(key); + extInserts.set(key, m); + inserted++; + writeOutcome.set(result.line, "inserted"); + } + touchedMeasurements.push({ + type: m.type as MeasurementType, + measuredAt: m.measuredAt, + }); + } + + // --- externalId-less rows: create-only, deduped on the natural unique key + // (userId, type, measuredAt, source=IMPORT, sleepStage=null). A collision + // — against the DB or an earlier row in this same file — is a duplicate, + // never an overwrite (each sample is a canonical reading). + const plainExisting = + plainRows.length > 0 + ? await prisma.measurement.findMany({ + where: { userId, - type: m.type as MeasurementType, - value: m.value, - unit: m.unit, source: "IMPORT", - measuredAt: m.measuredAt, - notes: null, - notesEncrypted: encryptNote(m.notes ?? null), - glucoseContext: - (m.glucoseContext as GlucoseContext | undefined) ?? null, + OR: plainRows.map((r) => ({ + type: r.row.type as MeasurementType, + measuredAt: r.row.measuredAt, + })), }, - }); - inserted++; - writeOutcome.set(result.line, "inserted"); - } + select: { type: true, measuredAt: true }, + }) + : []; + const plainExistingSet = new Set( + plainExisting.map((e) => `${e.type}::${e.measuredAt.getTime()}`), + ); + const seenPlain = new Set(); + const plainInserts: NormalisedMeasurementRow[] = []; + for (const result of plainRows) { + const m = result.row; + const key = `${m.type}::${m.measuredAt.getTime()}`; + if (plainExistingSet.has(key) || seenPlain.has(key)) { + skipped++; + writeOutcome.set(result.line, "duplicate"); + } else { + seenPlain.add(key); + plainInserts.push(m); + inserted++; + writeOutcome.set(result.line, "inserted"); touchedMeasurements.push({ type: m.type as MeasurementType, measuredAt: m.measuredAt, }); - } catch { - // Unique-constraint violation (a NULL-externalId duplicate against an - // existing row) — count as skipped rather than failing the file. - skipped++; - writeOutcome.set(result.line, "duplicate"); } } + // Bulk write: createMany the survivors (chunked under the PG parameter + // cap, skipDuplicates to absorb a concurrent double-submit race), then + // grouped updateMany for the pre-existing externalId rows. + const toCreate: Prisma.MeasurementCreateManyInput[] = [ + ...[...extInserts.values()].map(buildCreateData), + ...plainInserts.map(buildCreateData), + ]; + const CREATE_CHUNK = 200; + await prisma.$transaction(async (tx) => { + for (let i = 0; i < toCreate.length; i += CREATE_CHUNK) { + await tx.measurement.createMany({ + data: toCreate.slice(i, i + CREATE_CHUNK), + skipDuplicates: true, + }); + } + for (const m of extUpdates.values()) { + await tx.measurement.updateMany({ + where: { + userId, + source: "IMPORT", + type: m.type as MeasurementType, + externalId: m.externalId as string, + }, + data: { + value: m.value, + unit: m.unit, + measuredAt: m.measuredAt, + notes: null, + notesEncrypted: encryptNote(m.notes ?? null), + glucoseContext: + (m.glucoseContext as GlucoseContext | undefined) ?? null, + }, + }); + } + }); + // One bounded rollup re-fold per touched (type, day) — a 10 000-row CSV // pays at most ~N (type, day) recomputes, not 10 000 per-row hooks. // Best-effort: a populator hiccup never fails the importer.