From 19edda66c047715baeafbe158c0a856255f855b1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sat, 26 Sep 2026 22:06:39 +0200 Subject: [PATCH] fix(import): re-fold every day a CSV import touched, even when it fails A re-import that moved a reading to another instant re-folded only the day it moved to, so the day it left kept the reading in its rollups. And a write chunk that failed part-way ended the request before the re-fold ran; a retry then read the landed rows as duplicates and touched nothing, so those days kept their pre-import rollups for good. The re-fold now covers the old instant of a moved row and runs whether or not the writes finish. --- .../api/import/csv/__tests__/route.test.ts | 53 +++++- src/app/api/import/csv/route.ts | 158 ++++++++++-------- 2 files changed, 138 insertions(+), 73 deletions(-) diff --git a/src/app/api/import/csv/__tests__/route.test.ts b/src/app/api/import/csv/__tests__/route.test.ts index 0d2761b00..3ca886bea 100644 --- a/src/app/api/import/csv/__tests__/route.test.ts +++ b/src/app/api/import/csv/__tests__/route.test.ts @@ -227,7 +227,11 @@ describe("POST /api/import/csv — batched write + per-row envelope", () => { it("surfaces an externalId row as updated when it already existed (updateMany)", async () => { mMeasurement().findMany.mockResolvedValue([ - { type: "WEIGHT", externalId: "ext-1" }, + { + type: "WEIGHT", + externalId: "ext-1", + measuredAt: new Date("2026-05-01T08:00:00Z"), + }, ]); const res = await POST( @@ -400,7 +404,11 @@ describe("POST /api/import/csv — contextless blood glucose", () => { it("keeps a re-upload of the same external id idempotent", async () => { // The key already exists under (userId, type, source=IMPORT, externalId). mMeasurement().findMany.mockResolvedValue([ - { type: "BLOOD_GLUCOSE", externalId: "sensor-1" }, + { + type: "BLOOD_GLUCOSE", + externalId: "sensor-1", + measuredAt: new Date("2024-04-03T02:15:00Z"), + }, ]); const res = await POST( @@ -513,6 +521,47 @@ describe("POST /api/import/csv — re-import", () => { expect(mMeasurement().updateMany).toHaveBeenCalledTimes(1); }); + it("re-folds the day a moved reading left as well as the day it moved to", async () => { + mMeasurement().findMany.mockResolvedValue([stored("ext-1")]); + await POST( + csvRequest( + [HEADER, "WEIGHT,80.5,kg,2026-05-03T08:00:00Z,,after run,ext-1"].join( + "\n", + ), + ), + ); + const days = vi + .mocked(recomputeBucketsForMeasurement) + .mock.calls.map( + ([, type, at]) => `${type}@${at.toISOString().slice(0, 10)}`, + ) + .sort(); + expect(days).toEqual(["WEIGHT@2026-05-01", "WEIGHT@2026-05-03"]); + }); + + it("re-folds what it touched when a write chunk fails part-way", async () => { + mMeasurement().findMany.mockResolvedValue([]); + mMeasurement().createManyAndReturn.mockRejectedValue( + new Error("connection reset"), + ); + await expect( + POST( + csvRequest( + [ + HEADER, + "WEIGHT,80.5,kg,2026-05-01T08:00:00Z,,,", + "WEIGHT,80.7,kg,2026-05-02T08:00:00Z,,,", + ].join("\n"), + ), + ), + ).rejects.toThrow("connection reset"); + const days = vi + .mocked(recomputeBucketsForMeasurement) + .mock.calls.map(([, , at]) => at.toISOString().slice(0, 10)) + .sort(); + expect(days).toEqual(["2026-05-01", "2026-05-02"]); + }); + it("probes per type in bounded chunks and writes updates in bounded batches", async () => { const lines = Array.from({ length: 2500 }, (_, i) => { const at = new Date(Date.UTC(2026, 0, 1) + i * 60_000).toISOString(); diff --git a/src/app/api/import/csv/route.ts b/src/app/api/import/csv/route.ts index 369599d4b..37aafbf5e 100644 --- a/src/app/api/import/csv/route.ts +++ b/src/app/api/import/csv/route.ts @@ -284,6 +284,15 @@ export const POST = apiHandler(async (request: NextRequest) => { type: m.type as MeasurementType, measuredAt: m.measuredAt, }); + // An update that moves the reading to another instant leaves the day + // it came from: that day is re-folded too, or the old day keeps a + // reading that no longer exists there. + if (stored && stored.measuredAt.getTime() !== m.measuredAt.getTime()) { + touchedMeasurements.push({ + type: m.type as MeasurementType, + measuredAt: stored.measuredAt, + }); + } } // --- externalId-less rows: create-only, deduped on the natural unique key @@ -353,83 +362,90 @@ export const POST = apiHandler(async (request: NextRequest) => { type: MeasurementType; measuredAt: Date; }> = []; - for (const chunk of chunked(toCreate, WRITE_CHUNK)) { - const rows = await prisma.measurement.createManyAndReturn({ - data: chunk, - skipDuplicates: true, - select: { id: true, type: true, measuredAt: true }, - }); - createdRows.push(...rows); - createdCount += rows.length; - } - for (const chunk of chunked([...extUpdates.values()], WRITE_CHUNK)) { - await prisma.$transaction( - chunk.map((m) => - prisma.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, - // No-op on a live row, a deliberate RESURRECTION on a - // tombstoned one — IMPORT rows are re-importable by design, so - // a re-imported externalId brings the row back (mirrors the - // source-owned sync resurrect rule). - deletedAt: null, - }, - }), - ), - ); - } + // The re-fold runs whatever happens to the writes. A chunk that fails + // part-way leaves the earlier chunks written, and a retry reads those + // rows as duplicates and touches nothing: without the re-fold here their + // days would keep the rollups from before the import for good. Re-folding + // a (type, day) nothing landed on is harmless. + try { + for (const chunk of chunked(toCreate, WRITE_CHUNK)) { + const rows = await prisma.measurement.createManyAndReturn({ + data: chunk, + skipDuplicates: true, + select: { id: true, type: true, measuredAt: true }, + }); + createdRows.push(...rows); + createdCount += rows.length; + } + for (const chunk of chunked([...extUpdates.values()], WRITE_CHUNK)) { + await prisma.$transaction( + chunk.map((m) => + prisma.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, + // No-op on a live row, a deliberate RESURRECTION on a + // tombstoned one — IMPORT rows are re-importable by design, so + // a re-imported externalId brings the row back (mirrors the + // source-owned sync resurrect rule). + deletedAt: null, + }, + }), + ), + ); + } - void emitInsertedMeasurementArrivals( - userId, - createdRows, - "csv_import", - ).catch(() => {}); - void maybeEnqueueMorningRefresh( - userId, - createdRows - .filter((row) => row.type === "SLEEP_DURATION") - .map((row) => row.measuredAt), - ).catch(() => {}); + void emitInsertedMeasurementArrivals( + userId, + createdRows, + "csv_import", + ).catch(() => {}); + void maybeEnqueueMorningRefresh( + userId, + createdRows + .filter((row) => row.type === "SLEEP_DURATION") + .map((row) => row.measuredAt), + ).catch(() => {}); - // Reconcile `inserted` against what `createMany` actually wrote - // (mirrors the batch route's raced-duplicate downgrade). Under - // `skipDuplicates` a conflicting row is silently absorbed — the key is - // present in the table either way, so we cannot identify the SPECIFIC - // raced lines; we only need the counters and per-line statuses to sum - // to the truth. Downgrade enough `inserted` lines to `duplicate` so - // the envelope matches the DB write count. - const racedDuplicates = toCreate.length - createdCount; - if (racedDuplicates > 0) { - let downgraded = 0; - for (const [line, outcome] of writeOutcome) { - if (downgraded >= racedDuplicates) break; - if (outcome === "inserted") { - writeOutcome.set(line, "duplicate"); - inserted--; - skipped++; - downgraded++; + // Reconcile `inserted` against what `createMany` actually wrote + // (mirrors the batch route's raced-duplicate downgrade). Under + // `skipDuplicates` a conflicting row is silently absorbed — the key is + // present in the table either way, so we cannot identify the SPECIFIC + // raced lines; we only need the counters and per-line statuses to sum + // to the truth. Downgrade enough `inserted` lines to `duplicate` so + // the envelope matches the DB write count. + const racedDuplicates = toCreate.length - createdCount; + if (racedDuplicates > 0) { + let downgraded = 0; + for (const [line, outcome] of writeOutcome) { + if (downgraded >= racedDuplicates) break; + if (outcome === "inserted") { + writeOutcome.set(line, "duplicate"); + inserted--; + skipped++; + downgraded++; + } } } + } finally { + // 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 + // — and the status re-warm for the touched types. Best-effort: a + // populator hiccup never fails the importer. + await afterMeasurementMutation(userId, touchedMeasurements, "import.csv"); } - // 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 — and - // the status re-warm for the touched types. Best-effort: a populator - // hiccup never fails the importer. - await afterMeasurementMutation(userId, touchedMeasurements, "import.csv"); - await auditLog("import.csv.upload", { userId, ipAddress: getClientIp(request),