Merge branch 'fix/encrypt-free-text-and-local-ai' into release/v1.39.3

# Conflicts:
#	src/app/api/import/csv/route.ts
This commit is contained in:
Marc-André Bombeck committed 2026-09-26 21:50:05 +02:00
commit 66b1aef40c
69 files changed
+3625 -367

No files matched your search

+130 -7
View File
@@ -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
View File
@@ -8,7 +8,7 @@ import {
getClientIp,
readBodyText,
} 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";
@@ -26,6 +26,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).
*
@@ -141,38 +198,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");
@@ -198,24 +289,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) {
@@ -236,55 +331,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,