mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-09 13:01:00 +00:00
perf(import): batch CSV measurement upserts instead of one round-trip per row
The CSV importer ran a serial prisma.upsert/create per accepted row, so an N-row file cost N round-trips and a few-thousand-row import could approach the request timeout. Adopt the /measurements/batch shape: pre-fetch existing rows under the IMPORT source in one query per dedup-key shape (externalId and natural-key paths), partition into bulk createMany + grouped updateMany, and reconstruct the per-line inserted/updated/duplicate envelope from the partition. In-file repeats and re-import idempotency preserved; response contract unchanged.
This commit is contained in:
Marc-André Bombeck committed
1 parent
50031911c2
commit
329f78d295
2 files changed
+251
-87
No files matched your search
@@ -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<typeof vi.fn>;
|
||||
createMany: ReturnType<typeof vi.fn>;
|
||||
updateMany: ReturnType<typeof vi.fn>;
|
||||
};
|
||||
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<typeof vi.fn>;
|
||||
createMany: ReturnType<typeof vi.fn>;
|
||||
updateMany: ReturnType<typeof vi.fn>;
|
||||
};
|
||||
|
||||
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);
|
||||
});
|
||||
});
|
||||
+147
-64
@@ -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<number, "inserted" | "updated" | "duplicate">();
|
||||
|
||||
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<string>();
|
||||
const extInserts = new Map<string, NormalisedMeasurementRow>();
|
||||
const extUpdates = new Map<string, NormalisedMeasurementRow>();
|
||||
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<string>();
|
||||
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.
|
||||
|
||||
Reference in new issue
Block a user