mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-11 22:02:16 +00:00
fix(import): make re-importing the same CSV idempotent and fast
Re-uploading a large export with an externalId column ran one updateMany per row inside a single interactive transaction; on a 9 000-row file that outlived Prisma's 5 s timeout and the request ended in a 500. A re-imported externalId row whose stored value, unit, instant, glucose context and note already match the line is now a duplicate: not written, reported as skipped / duplicate. Changed rows still update, a tombstoned one is still resurrected. The existence probes run per measurement type in chunks of 1 000 keys, and the writes commit in chunks of 200 (creates via createManyAndReturn, updates as one batch transaction per chunk) instead of one transaction around the whole file. A failure part-way leaves earlier chunks written, and re-uploading the same file completes the rest. The OpenAPI description now says what a re-upload does, including the natural-key dedup for rows without an externalId, which it used to call a duplicate.
This commit is contained in:
1 parent
46eb8c8b94
commit
24050d84c7
4 files changed
+462
-86
No files matched your search
@@ -17,10 +17,13 @@ vi.mock("@/lib/db", () => {
|
||||
return {
|
||||
prisma: {
|
||||
measurement,
|
||||
// Run the batched write callback against the same measurement mock so
|
||||
// createManyAndReturn / updateMany calls are observable.
|
||||
$transaction: vi.fn(async (fn: (tx: unknown) => unknown) =>
|
||||
fn({ measurement }),
|
||||
// Both transaction forms against the same measurement mock so
|
||||
// createManyAndReturn / updateMany calls are observable: a callback
|
||||
// gets the mock as `tx`, a batch array is simply awaited.
|
||||
$transaction: vi.fn(async (arg: unknown) =>
|
||||
typeof arg === "function"
|
||||
? (arg as (tx: unknown) => unknown)({ measurement })
|
||||
: Promise.all(arg as Promise<unknown>[]),
|
||||
),
|
||||
},
|
||||
};
|
||||
@@ -37,6 +40,10 @@ vi.mock("@/lib/crypto/note-cipher", () => ({
|
||||
s === null || s === undefined || s.length === 0
|
||||
? null
|
||||
: new Uint8Array(Buffer.from(`enc:${s}`, "utf8")),
|
||||
readNote: (c: Uint8Array | null | undefined, p: string | null | undefined) =>
|
||||
c && c.byteLength > 0
|
||||
? Buffer.from(c).toString("utf8").slice(4)
|
||||
: (p ?? null),
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/auth/audit", () => ({
|
||||
@@ -115,8 +122,10 @@ beforeEach(() => {
|
||||
prisma.$transaction as unknown as {
|
||||
mockImplementation: (f: unknown) => void;
|
||||
}
|
||||
).mockImplementation(async (fn: (tx: unknown) => unknown) =>
|
||||
fn({ measurement }),
|
||||
).mockImplementation(async (arg: unknown) =>
|
||||
typeof arg === "function"
|
||||
? (arg as (tx: unknown) => unknown)({ measurement })
|
||||
: Promise.all(arg as Promise<unknown>[]),
|
||||
);
|
||||
});
|
||||
|
||||
@@ -239,7 +248,19 @@ describe("POST /api/import/csv — batched write + per-row envelope", () => {
|
||||
// row matches like a live one; the update must carry the resurrection
|
||||
// (IMPORT rows are re-importable by design).
|
||||
mMeasurement().findMany.mockResolvedValue([
|
||||
{ type: "WEIGHT", externalId: "ext-tomb" },
|
||||
{
|
||||
type: "WEIGHT",
|
||||
externalId: "ext-tomb",
|
||||
// Identical values: only the tombstone differs, and that alone
|
||||
// makes the line an update rather than a duplicate.
|
||||
value: 81.0,
|
||||
unit: "kg",
|
||||
measuredAt: new Date("2026-05-01T08:00:00Z"),
|
||||
glucoseContext: null,
|
||||
notes: null,
|
||||
notesEncrypted: null,
|
||||
deletedAt: new Date("2026-05-02T08:00:00Z"),
|
||||
},
|
||||
]);
|
||||
|
||||
const res = await POST(
|
||||
@@ -424,3 +445,105 @@ describe("POST /api/import/csv — contextless blood glucose", () => {
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
/**
|
||||
* v1.39.3 — re-importing the same file. A 9 000-row re-upload with an
|
||||
* externalId column used to run one update per row inside a single
|
||||
* interactive transaction and ended in a 500 when that outlived Prisma's 5 s
|
||||
* timeout. An unchanged row is now a duplicate with no write, a changed one
|
||||
* still updates, and lookups and writes run in bounded chunks.
|
||||
*/
|
||||
describe("POST /api/import/csv — re-import", () => {
|
||||
beforeEach(() => {
|
||||
vi.mocked(checkRateLimit).mockResolvedValue({
|
||||
allowed: true,
|
||||
remaining: 4,
|
||||
resetAt: Date.now() + 3_600_000,
|
||||
} as never);
|
||||
mMeasurement().findMany.mockReset();
|
||||
mMeasurement().createManyAndReturn.mockReset();
|
||||
mMeasurement().createManyAndReturn.mockResolvedValue([]);
|
||||
mMeasurement().updateMany.mockReset();
|
||||
mMeasurement().updateMany.mockResolvedValue({ count: 1 });
|
||||
vi.mocked(prisma.$transaction).mockClear();
|
||||
});
|
||||
|
||||
function stored(externalId: string, overrides: Record<string, unknown> = {}) {
|
||||
return {
|
||||
type: "WEIGHT",
|
||||
externalId,
|
||||
value: 80.5,
|
||||
unit: "kg",
|
||||
measuredAt: new Date("2026-05-01T08:00:00Z"),
|
||||
glucoseContext: null,
|
||||
notes: null,
|
||||
notesEncrypted: new Uint8Array(Buffer.from("enc:after run", "utf8")),
|
||||
deletedAt: null,
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
it("writes nothing for a row that is already stored exactly so", async () => {
|
||||
mMeasurement().findMany.mockResolvedValue([stored("ext-1")]);
|
||||
const res = await POST(
|
||||
csvRequest(
|
||||
[HEADER, "WEIGHT,80.5,kg,2026-05-01T08:00:00Z,,after run,ext-1"].join(
|
||||
"\n",
|
||||
),
|
||||
),
|
||||
);
|
||||
const body = (await res.json()) as CsvEnvelope;
|
||||
expect(body.data).toMatchObject({ inserted: 0, updated: 0, skipped: 1 });
|
||||
expect(body.data?.rows).toEqual([
|
||||
{ line: 2, status: "skipped", reason: "duplicate" },
|
||||
]);
|
||||
expect(mMeasurement().updateMany).not.toHaveBeenCalled();
|
||||
expect(mMeasurement().createManyAndReturn).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it.each([
|
||||
["value", "WEIGHT,81,kg,2026-05-01T08:00:00Z,,after run,ext-1"],
|
||||
["note", "WEIGHT,80.5,kg,2026-05-01T08:00:00Z,,before run,ext-1"],
|
||||
["instant", "WEIGHT,80.5,kg,2026-05-01T09:00:00Z,,after run,ext-1"],
|
||||
])("still updates a row whose %s changed", async (_what, line) => {
|
||||
mMeasurement().findMany.mockResolvedValue([stored("ext-1")]);
|
||||
const res = await POST(csvRequest([HEADER, line].join("\n")));
|
||||
const body = (await res.json()) as CsvEnvelope;
|
||||
expect(body.data).toMatchObject({ updated: 1, skipped: 0 });
|
||||
expect(mMeasurement().updateMany).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
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();
|
||||
return `WEIGHT,${70 + (i % 10)},kg,${at},,,ext-${i}`;
|
||||
});
|
||||
// Every key exists, with a different value, so every line updates.
|
||||
mMeasurement().findMany.mockImplementation((async (args: {
|
||||
where: { externalId: { in: string[] } };
|
||||
}) =>
|
||||
args.where.externalId.in.map((externalId) =>
|
||||
stored(externalId, { value: 1 }),
|
||||
)) as never);
|
||||
|
||||
const res = await POST(csvRequest([HEADER, ...lines].join("\n")));
|
||||
const body = (await res.json()) as CsvEnvelope;
|
||||
expect(body.data?.updated).toBe(2500);
|
||||
|
||||
const probes = mMeasurement().findMany.mock.calls.map(
|
||||
(c) => (c[0] as { where: { externalId: { in: string[] } } }).where,
|
||||
);
|
||||
expect(probes).toHaveLength(3);
|
||||
expect(probes.every((w) => w.externalId.in.length <= 1000)).toBe(true);
|
||||
// 2 500 updates in batches of at most 200, each its own transaction:
|
||||
// no single transaction carries the whole file.
|
||||
const batches = vi
|
||||
.mocked(prisma.$transaction)
|
||||
.mock.calls.map((c) => c[0] as unknown as unknown[]);
|
||||
expect(batches).toHaveLength(13);
|
||||
expect(batches.every((b) => Array.isArray(b) && b.length <= 200)).toBe(
|
||||
true,
|
||||
);
|
||||
expect(mMeasurement().updateMany).toHaveBeenCalledTimes(2500);
|
||||
});
|
||||
});
|
||||
+183
-78
@@ -3,7 +3,7 @@ import { apiHandler, requireAuth } from "@/lib/api-handler";
|
||||
import { annotate } from "@/lib/logging/context";
|
||||
import { auditLog } from "@/lib/auth/audit";
|
||||
import { apiSuccess, apiError, getClientIp } from "@/lib/api-response";
|
||||
import { encryptNote } from "@/lib/crypto/note-cipher";
|
||||
import { encryptNote, readNote } from "@/lib/crypto/note-cipher";
|
||||
import { checkRateLimit } from "@/lib/rate-limit";
|
||||
import { NextRequest } from "next/server";
|
||||
|
||||
@@ -24,6 +24,63 @@ import type { NormalisedMeasurementRow } from "@/lib/import/csv-measurements";
|
||||
import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit";
|
||||
import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger";
|
||||
|
||||
/** Keys per dedup probe: well under Postgres's bind-parameter cap. */
|
||||
const LOOKUP_CHUNK = 1000;
|
||||
/** Rows per create call / per update batch transaction. */
|
||||
const WRITE_CHUNK = 200;
|
||||
|
||||
function chunked<T>(items: readonly T[], size: number): T[][] {
|
||||
const out: T[][] = [];
|
||||
for (let i = 0; i < items.length; i += size) {
|
||||
out.push(items.slice(i, i + size));
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
/** Group rows' probe keys by measurement type, de-duplicated. */
|
||||
function groupByType<K>(
|
||||
rows: ReadonlyArray<{ row: NormalisedMeasurementRow }>,
|
||||
keyOf: (r: { row: NormalisedMeasurementRow }) => K,
|
||||
): Map<MeasurementType, K[]> {
|
||||
const byType = new Map<MeasurementType, Map<string, K>>();
|
||||
for (const r of rows) {
|
||||
const type = r.row.type as MeasurementType;
|
||||
const key = keyOf(r);
|
||||
const bucket = byType.get(type) ?? new Map<string, K>();
|
||||
bucket.set(key instanceof Date ? String(key.getTime()) : String(key), key);
|
||||
byType.set(type, bucket);
|
||||
}
|
||||
return new Map([...byType].map(([type, keys]) => [type, [...keys.values()]]));
|
||||
}
|
||||
|
||||
/**
|
||||
* True when a stored IMPORT row already holds exactly what this line would
|
||||
* write: same value, unit, instant, glucose context and note, and live. Such
|
||||
* a line is a duplicate, not an update.
|
||||
*/
|
||||
function sameStoredRow(
|
||||
stored: {
|
||||
value: number;
|
||||
unit: string;
|
||||
measuredAt: Date;
|
||||
glucoseContext: GlucoseContext | null;
|
||||
notes: string | null;
|
||||
notesEncrypted: Uint8Array | null;
|
||||
deletedAt: Date | null;
|
||||
},
|
||||
m: NormalisedMeasurementRow,
|
||||
): boolean {
|
||||
return (
|
||||
stored.deletedAt === null &&
|
||||
stored.value === m.value &&
|
||||
stored.unit === m.unit &&
|
||||
stored.measuredAt.getTime() === m.measuredAt.getTime() &&
|
||||
stored.glucoseContext ===
|
||||
((m.glucoseContext as GlucoseContext | undefined) ?? null) &&
|
||||
readNote(stored.notesEncrypted, stored.notes) === (m.notes || null)
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* CSV measurement import (v1.17.1).
|
||||
*
|
||||
@@ -138,38 +195,72 @@ export const POST = apiHandler(async (request: NextRequest) => {
|
||||
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,
|
||||
source: "IMPORT",
|
||||
OR: extRows.map((r) => ({
|
||||
type: r.row.type as MeasurementType,
|
||||
externalId: r.row.externalId as string,
|
||||
})),
|
||||
},
|
||||
select: { type: true, externalId: true },
|
||||
})
|
||||
: [];
|
||||
const extExistingSet = new Set(
|
||||
extExisting.map((e) => `${e.type}::${e.externalId}`),
|
||||
);
|
||||
// (userId, type, source=IMPORT, externalId). One probe per type and chunk
|
||||
// fetches every pre-existing key with the columns an update would write,
|
||||
// so a row whose stored values already match is left alone.
|
||||
const extExisting = new Map<
|
||||
string,
|
||||
{
|
||||
value: number;
|
||||
unit: string;
|
||||
measuredAt: Date;
|
||||
glucoseContext: GlucoseContext | null;
|
||||
notes: string | null;
|
||||
notesEncrypted: Uint8Array | null;
|
||||
deletedAt: Date | null;
|
||||
}
|
||||
>();
|
||||
for (const [type, ids] of groupByType(
|
||||
extRows,
|
||||
(r) => r.row.externalId as string,
|
||||
)) {
|
||||
for (const chunk of chunked(ids, LOOKUP_CHUNK)) {
|
||||
const found = await prisma.measurement.findMany({
|
||||
where: {
|
||||
userId,
|
||||
source: "IMPORT",
|
||||
type,
|
||||
externalId: { in: chunk },
|
||||
},
|
||||
select: {
|
||||
externalId: true,
|
||||
value: true,
|
||||
unit: true,
|
||||
measuredAt: true,
|
||||
glucoseContext: true,
|
||||
notes: true,
|
||||
notesEncrypted: true,
|
||||
deletedAt: true,
|
||||
},
|
||||
});
|
||||
for (const row of found) {
|
||||
extExisting.set(`${type}::${row.externalId}`, row);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 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.
|
||||
// A key already in the DB whose stored row equals this line is a
|
||||
// duplicate: no write, reported like the natural-key duplicates below.
|
||||
// That is what makes re-importing the same file cheap — a second upload
|
||||
// of an unchanged 9 000-row export writes nothing.
|
||||
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)) {
|
||||
const stored = extExisting.get(key);
|
||||
if (stored && !extUpdates.has(key) && sameStoredRow(stored, m)) {
|
||||
skipped++;
|
||||
writeOutcome.set(result.line, "duplicate");
|
||||
continue;
|
||||
}
|
||||
if (stored) {
|
||||
extUpdates.set(key, m);
|
||||
updated++;
|
||||
writeOutcome.set(result.line, "updated");
|
||||
@@ -195,24 +286,28 @@ export const POST = apiHandler(async (request: NextRequest) => {
|
||||
// --- 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,
|
||||
source: "IMPORT",
|
||||
OR: plainRows.map((r) => ({
|
||||
type: r.row.type as MeasurementType,
|
||||
measuredAt: r.row.measuredAt,
|
||||
})),
|
||||
},
|
||||
select: { type: true, measuredAt: true },
|
||||
})
|
||||
: [];
|
||||
const plainExistingSet = new Set(
|
||||
plainExisting.map((e) => `${e.type}::${e.measuredAt.getTime()}`),
|
||||
);
|
||||
// never an overwrite (each sample is a canonical reading). Probed per type
|
||||
// and chunk, like the externalId rows.
|
||||
const plainExistingSet = new Set<string>();
|
||||
for (const [type, instants] of groupByType(
|
||||
plainRows,
|
||||
(r) => r.row.measuredAt,
|
||||
)) {
|
||||
for (const chunk of chunked(instants, LOOKUP_CHUNK)) {
|
||||
const found = await prisma.measurement.findMany({
|
||||
where: {
|
||||
userId,
|
||||
source: "IMPORT",
|
||||
type,
|
||||
measuredAt: { in: chunk },
|
||||
},
|
||||
select: { measuredAt: true },
|
||||
});
|
||||
for (const row of found) {
|
||||
plainExistingSet.add(`${type}::${row.measuredAt.getTime()}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
const seenPlain = new Set<string>();
|
||||
const plainInserts: NormalisedMeasurementRow[] = [];
|
||||
for (const result of plainRows) {
|
||||
@@ -233,55 +328,65 @@ export const POST = apiHandler(async (request: NextRequest) => {
|
||||
}
|
||||
}
|
||||
|
||||
// Bulk write: createManyAndReturn the survivors (chunked under the PG
|
||||
// parameter cap, skipDuplicates to absorb a concurrent double-submit
|
||||
// race), then grouped updateMany for pre-existing externalId rows.
|
||||
// Bulk write in bounded chunks, each its own short transaction.
|
||||
//
|
||||
// This used to be ONE interactive transaction around every create chunk
|
||||
// AND one `updateMany` per re-imported row. Re-importing a 9 000-row file
|
||||
// with an externalId column meant 9 000 serial updates inside it, which
|
||||
// outlived Prisma's 5 s interactive-transaction timeout and ended in a
|
||||
// 500. A chunk now commits on its own: creates go through
|
||||
// `createManyAndReturn` (skipDuplicates absorbs a concurrent
|
||||
// double-submit), updates through one batch transaction per chunk. A
|
||||
// failure part-way leaves the earlier chunks written, and that is safe to
|
||||
// retry: a re-upload of the same file skips what landed (unchanged
|
||||
// externalId rows and natural-key duplicates are not written again).
|
||||
const toCreate: Prisma.MeasurementCreateManyInput[] = [
|
||||
...[...extInserts.values()].map(buildCreateData),
|
||||
...plainInserts.map(buildCreateData),
|
||||
];
|
||||
const CREATE_CHUNK = 200;
|
||||
let createdCount = 0;
|
||||
const createdRows: Array<{
|
||||
id: string;
|
||||
type: MeasurementType;
|
||||
measuredAt: Date;
|
||||
}> = [];
|
||||
await prisma.$transaction(async (tx) => {
|
||||
for (let i = 0; i < toCreate.length; i += CREATE_CHUNK) {
|
||||
const rows = await tx.measurement.createManyAndReturn({
|
||||
data: toCreate.slice(i, i + CREATE_CHUNK),
|
||||
skipDuplicates: true,
|
||||
select: { id: true, type: true, measuredAt: true },
|
||||
});
|
||||
createdRows.push(...rows);
|
||||
createdCount += rows.length;
|
||||
}
|
||||
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,
|
||||
// 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,
|
||||
},
|
||||
});
|
||||
}
|
||||
});
|
||||
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,
|
||||
|
||||
@@ -183,7 +183,7 @@ export const importPaths: NonNullable<ZodOpenApiObject["paths"]> = {
|
||||
tags: ["Import"],
|
||||
summary: "Import measurements from CSV",
|
||||
description:
|
||||
"Bulk-import measurements from a CSV file (web cold-start escape hatch). Body is raw `text/csv` (≤ 16 MB, ≤ 10 000 valid rows). Header (order-independent): `type,value,unit,measuredAt[,glucoseContext,notes,externalId]`. `measuredAt` must carry an explicit ISO-8601 offset and is bounded (no future beyond a 5-min skew, no instant before 1900). Glucose accepts `mmol/L` (converted to canonical `mg/dL`); weight accepts `lb` (converted to `kg`). `glucoseContext` is optional on a `BLOOD_GLUCOSE` row — a blank cell stores no context, which is the normal shape of a continuous-sensor export; a non-empty value is still checked against the enum, and the column is refused on any other type. An `externalId` column makes re-upload idempotent (upsert on `(userId, type, source=IMPORT, externalId)`); without it a re-upload duplicates. `?dryRun=1` validates + previews without writing. Shares the 5/hour `import:` rate bucket.",
|
||||
"Bulk-import measurements from a CSV file (web cold-start escape hatch). Body is raw `text/csv` (≤ 16 MB, ≤ 10 000 valid rows). Header (order-independent): `type,value,unit,measuredAt[,glucoseContext,notes,externalId]`. `measuredAt` must carry an explicit ISO-8601 offset and is bounded (no future beyond a 5-min skew, no instant before 1900). Glucose accepts `mmol/L` (converted to canonical `mg/dL`); weight accepts `lb` (converted to `kg`). `glucoseContext` is optional on a `BLOOD_GLUCOSE` row — a blank cell stores no context, which is the normal shape of a continuous-sensor export; a non-empty value is still checked against the enum, and the column is refused on any other type. Re-uploading a file is idempotent. With an `externalId` column a row upserts on `(userId, type, source=IMPORT, externalId)`: a changed row is `updated`, an unchanged one is `skipped` with reason `duplicate` and not written. Without it a row is deduplicated on `(type, measuredAt)` and a repeat is `skipped` / `duplicate`. Writes commit in chunks rather than as one transaction; a failure part-way leaves earlier chunks written, and re-uploading the same file completes the rest. `?dryRun=1` validates + previews without writing. Shares the 5/hour `import:` rate bucket.",
|
||||
parameters: [
|
||||
{
|
||||
name: "dryRun",
|
||||
|
||||
@@ -0,0 +1,148 @@
|
||||
/**
|
||||
* v1.39.3 — importing the same large CSV twice.
|
||||
*
|
||||
* A re-upload of a 9 000-row export with an externalId column ended in a 500:
|
||||
* every row already existed, so the route ran one update per row, serially,
|
||||
* inside a single interactive transaction, and that outlived Prisma's 5 s
|
||||
* timeout. Only a real database shows it. This imports the same file twice and
|
||||
* then a lightly edited copy, and reads the table back after each pass.
|
||||
*/
|
||||
import { NextRequest } from "next/server";
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
import { cookieJar } from "./mock-next-headers";
|
||||
import { getPrismaClient, truncateAllTables } from "./setup";
|
||||
|
||||
const USER = "user-csv-reimport";
|
||||
|
||||
vi.mock("next/headers", async () => {
|
||||
const { cookieJar, headerJar } = await import("./mock-next-headers");
|
||||
return {
|
||||
headers: vi.fn(async () => ({
|
||||
get: (name: string) => headerJar.get(name.toLowerCase()) ?? null,
|
||||
})),
|
||||
cookies: vi.fn(async () => ({
|
||||
get: (name: string) => {
|
||||
const value = cookieJar.get(name);
|
||||
return value ? { name, value } : undefined;
|
||||
},
|
||||
set: (name: string, value: string) => {
|
||||
cookieJar.set(name, value);
|
||||
},
|
||||
delete: (name: string) => {
|
||||
cookieJar.delete(name);
|
||||
},
|
||||
})),
|
||||
};
|
||||
});
|
||||
|
||||
vi.mock("@/lib/db-compat", () => ({
|
||||
ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
|
||||
}));
|
||||
|
||||
const HEADER = "type,value,unit,measuredAt,glucoseContext,notes,externalId";
|
||||
const ROWS = 9000;
|
||||
|
||||
/** 8 000 externalId rows and 1 000 without one, a minute apart. */
|
||||
function fileLines(changed: ReadonlySet<number> = new Set()): string[] {
|
||||
return Array.from({ length: ROWS }, (_, i) => {
|
||||
const at = new Date(Date.UTC(2025, 0, 1) + i * 60_000).toISOString();
|
||||
const value = (70 + (i % 50) / 10 + (changed.has(i) ? 1 : 0)).toFixed(1);
|
||||
const note = i % 100 === 0 ? "after the long ride" : "";
|
||||
const externalId = i < 8000 ? `export-${i}` : "";
|
||||
return `WEIGHT,${value},kg,${at},,${note},${externalId}`;
|
||||
});
|
||||
}
|
||||
|
||||
function csvRequest(lines: string[]): NextRequest {
|
||||
return new NextRequest("http://localhost/api/import/csv", {
|
||||
method: "POST",
|
||||
headers: { "content-type": "text/csv" },
|
||||
body: [HEADER, ...lines].join("\n"),
|
||||
});
|
||||
}
|
||||
|
||||
interface CsvEnvelope {
|
||||
data: {
|
||||
inserted: number;
|
||||
updated: number;
|
||||
skipped: number;
|
||||
total: number;
|
||||
} | null;
|
||||
}
|
||||
|
||||
async function signIn() {
|
||||
await getPrismaClient().user.create({
|
||||
data: {
|
||||
id: USER,
|
||||
username: "csv-reimport",
|
||||
email: "csv-reimport@example.test",
|
||||
timezone: "Europe/Berlin",
|
||||
},
|
||||
});
|
||||
const session = await getPrismaClient().session.create({
|
||||
data: { userId: USER, expiresAt: new Date(Date.now() + 60 * 60 * 1000) },
|
||||
});
|
||||
cookieJar.set("healthlog_session", session.id);
|
||||
}
|
||||
|
||||
beforeEach(async () => {
|
||||
await truncateAllTables(getPrismaClient());
|
||||
cookieJar.clear();
|
||||
});
|
||||
|
||||
describe("POST /api/import/csv — re-importing a 9 000-row file (real Postgres)", () => {
|
||||
it("imports once, writes nothing the second time, and updates only what changed", async () => {
|
||||
await signIn();
|
||||
const { POST } = await import("@/app/api/import/csv/route");
|
||||
const prisma = getPrismaClient();
|
||||
|
||||
const first = await POST(csvRequest(fileLines()));
|
||||
expect(first.status).toBe(200);
|
||||
expect(((await first.json()) as CsvEnvelope).data).toMatchObject({
|
||||
inserted: ROWS,
|
||||
updated: 0,
|
||||
skipped: 0,
|
||||
});
|
||||
expect(await prisma.measurement.count({ where: { userId: USER } })).toBe(
|
||||
ROWS,
|
||||
);
|
||||
const before = await prisma.measurement.findMany({
|
||||
where: { userId: USER },
|
||||
select: { id: true, value: true, notesEncrypted: true },
|
||||
orderBy: { measuredAt: "asc" },
|
||||
});
|
||||
|
||||
const second = await POST(csvRequest(fileLines()));
|
||||
expect(second.status).toBe(200);
|
||||
expect(((await second.json()) as CsvEnvelope).data).toMatchObject({
|
||||
inserted: 0,
|
||||
updated: 0,
|
||||
skipped: ROWS,
|
||||
});
|
||||
const after = await prisma.measurement.findMany({
|
||||
where: { userId: USER },
|
||||
select: { id: true, value: true, notesEncrypted: true },
|
||||
orderBy: { measuredAt: "asc" },
|
||||
});
|
||||
// Same rows, untouched: not even the note ciphertext was re-sealed.
|
||||
expect(after).toEqual(before);
|
||||
|
||||
const changed = new Set([0, 1, 4999, 7999]);
|
||||
const third = await POST(csvRequest(fileLines(changed)));
|
||||
expect(third.status).toBe(200);
|
||||
expect(((await third.json()) as CsvEnvelope).data).toMatchObject({
|
||||
inserted: 0,
|
||||
updated: changed.size,
|
||||
skipped: ROWS - changed.size,
|
||||
});
|
||||
expect(await prisma.measurement.count({ where: { userId: USER } })).toBe(
|
||||
ROWS,
|
||||
);
|
||||
const edited = await prisma.measurement.findFirst({
|
||||
where: { userId: USER, externalId: "export-4999" },
|
||||
select: { value: true },
|
||||
});
|
||||
expect(edited?.value).toBeCloseTo(70 + (4999 % 50) / 10 + 1, 5);
|
||||
}, 120_000);
|
||||
});
|
||||
Reference in new issue
Block a user