tom/HealthLog
1
0 Fork 0
mirror of https://github.com/MBombeck/HealthLog.git synced 2026-11-08 20:42:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(import): re-fold every day a CSV import touched, even when it fails

Browse Source 
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.
This commit is contained in:
Marc-André Bombeck committed 2026-09-26 22:06:39 +02:00
1 parent 96b98cb719
commit 19edda66c0
2 files changed
+138 -73
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/app/api/import/csv/__tests__/route.test.ts
+51 -2
View File
@@ -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();
src/app/api/import/csv/route.ts
+87 -71
View File
@@ -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),