mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-09 04:50:59 +00:00
Marc-André Bombeck
f8c17ee9bb
Every accepted row was its own insert, so a 20 000-row file meant 20 000 round trips and an import cut off part way left an arbitrary prefix behind. Each 200-row chunk is now one createMany: it lands whole or not at all, and uploading the same file again skips the written chunks as duplicates. The importers report each committed write, and the audit row is written in a finally block, marked partial with the rows written when the import did not finish.
370 lines
11 KiB
TypeScript
370 lines
11 KiB
TypeScript
/**
|
|
* The lab and medication CSV imports against a real Postgres (#1215).
|
|
*
|
|
* - a dry run writes nothing (no reading, no marker, no medication) and
|
|
* reports what the real run then does;
|
|
* - names resolve to existing markers, new analytes mint one marker each,
|
|
* a reading in another unit is converted and keeps what the file said;
|
|
* - re-uploading the same file writes nothing: every row is a duplicate;
|
|
* - a refused row is reported by line and does not stop the rest;
|
|
* - medications take a daily schedule, reminders off, as-needed without
|
|
* schedule, and a dated window opens a course.
|
|
*/
|
|
import { beforeAll, describe, expect, it, vi } from "vitest";
|
|
|
|
import { getPrismaClient, truncateAllTables } from "./setup";
|
|
import { textToStream } from "@/lib/import/csv-stream";
|
|
import { importLabsCsv } from "@/lib/import/labs-csv";
|
|
import { importMedicationsCsv } from "@/lib/import/medications-csv";
|
|
import {
|
|
labsCsvQuerySchema,
|
|
medicationsCsvQuerySchema,
|
|
} from "@/lib/validations/record-import";
|
|
|
|
vi.mock("@/lib/db-compat", () => ({
|
|
ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
|
|
}));
|
|
vi.mock("@/lib/jobs/boss-instance", () => ({
|
|
getGlobalBoss: vi.fn(() => null),
|
|
}));
|
|
|
|
const prisma = getPrismaClient();
|
|
const USER = "csv-import-user";
|
|
const TZ = "Europe/Berlin";
|
|
|
|
const LABS = [
|
|
"Analyt;Wert;Einheit;Datum;Referenz",
|
|
"Glucose;5,5;mmol/L;01.03.2025;3,9 - 5,5",
|
|
"Glukose;99;mg/dL;02.03.2025;",
|
|
"HbA1c;5,4;%;01.03.2025;4,0-5,6",
|
|
"Urinstatus Eiweiß;negativ;;01.03.2025;",
|
|
"Ferritin;1.234;ng/mL;01.03.2025;",
|
|
"Ferritin;85;;01.03.2025;",
|
|
"Neuer Marker;12;U/l;03.03.2025;",
|
|
].join("\n");
|
|
|
|
function labsQuery(dryRun: boolean) {
|
|
return labsCsvQuerySchema.parse({
|
|
analyte: "0",
|
|
value: "1",
|
|
unit: "2",
|
|
date: "3",
|
|
referenceRange: "4",
|
|
decimal: "comma",
|
|
dateOrder: "dmy",
|
|
dryRun: dryRun ? "1" : "0",
|
|
});
|
|
}
|
|
|
|
beforeAll(async () => {
|
|
await truncateAllTables(prisma);
|
|
await prisma.user.create({
|
|
data: { id: USER, username: USER, timezone: TZ },
|
|
});
|
|
await prisma.biomarker.create({
|
|
// No analyte key stored: the import derives it the way the write does.
|
|
data: { userId: USER, name: "Glucose", unit: "mg/dL" },
|
|
});
|
|
await prisma.biomarkerAlias.create({
|
|
data: {
|
|
userId: USER,
|
|
biomarkerId: (
|
|
await prisma.biomarker.findFirstOrThrow({ where: { userId: USER } })
|
|
).id,
|
|
name: "Glukose",
|
|
nameFolded: "glukose",
|
|
source: "MANUAL",
|
|
},
|
|
});
|
|
});
|
|
|
|
describe("lab CSV import", () => {
|
|
it("a dry run writes nothing and reports the plan", async () => {
|
|
const result = await importLabsCsv({
|
|
userId: USER,
|
|
tz: TZ,
|
|
stream: textToStream(LABS, 16),
|
|
maxBytes: 1_000_000,
|
|
query: labsQuery(true),
|
|
});
|
|
expect(result).toMatchObject({
|
|
rows: 7,
|
|
valid: 5,
|
|
inserted: 5,
|
|
failed: 2,
|
|
duplicates: 0,
|
|
});
|
|
expect(result.errors).toEqual([
|
|
{ line: 6, reason: "ambiguous_number", field: "value" },
|
|
{ line: 7, reason: "missing_unit", field: "unit" },
|
|
]);
|
|
expect(result.markers?.created.sort()).toEqual([
|
|
"HbA1c",
|
|
"Neuer Marker",
|
|
"Urinstatus Eiweiß",
|
|
]);
|
|
expect(result.markers?.matched.map((m) => m.how).sort()).toEqual([
|
|
"alias",
|
|
"name",
|
|
]);
|
|
expect(await prisma.labResult.count({ where: { userId: USER } })).toBe(0);
|
|
expect(await prisma.biomarker.count({ where: { userId: USER } })).toBe(1);
|
|
});
|
|
|
|
it("the import converts, mints and keeps what the file said", async () => {
|
|
const result = await importLabsCsv({
|
|
userId: USER,
|
|
tz: TZ,
|
|
stream: textToStream(LABS, 16),
|
|
maxBytes: 1_000_000,
|
|
query: labsQuery(false),
|
|
});
|
|
expect(result).toMatchObject({ inserted: 5, failed: 2 });
|
|
const glucose = await prisma.labResult.findMany({
|
|
where: { userId: USER, analyte: "Glucose" },
|
|
orderBy: { takenAt: "asc" },
|
|
});
|
|
expect(glucose).toHaveLength(2);
|
|
expect(glucose[0].unit).toBe("mg/dL");
|
|
expect(glucose[0].value).toBeCloseTo(99.1, 0);
|
|
expect(glucose[0].sourceValue).toBe(5.5);
|
|
expect(glucose[0].sourceUnit).toBe("mmol/L");
|
|
expect(glucose[0].source).toBe("IMPORT");
|
|
expect(glucose[1].value).toBe(99);
|
|
const text = await prisma.labResult.findFirstOrThrow({
|
|
where: { userId: USER, valueText: "negativ" },
|
|
});
|
|
expect(text.value).toBeNull();
|
|
expect(await prisma.biomarker.count({ where: { userId: USER } })).toBe(4);
|
|
});
|
|
|
|
it("re-uploading the same file writes nothing", async () => {
|
|
const before = await prisma.labResult.count({ where: { userId: USER } });
|
|
const result = await importLabsCsv({
|
|
userId: USER,
|
|
tz: TZ,
|
|
stream: textToStream(LABS, 7),
|
|
maxBytes: 1_000_000,
|
|
query: labsQuery(false),
|
|
});
|
|
expect(result).toMatchObject({ inserted: 0, duplicates: 5, failed: 2 });
|
|
expect(await prisma.labResult.count({ where: { userId: USER } })).toBe(
|
|
before,
|
|
);
|
|
});
|
|
|
|
it("refuses a unit that does not convert into the marker's", async () => {
|
|
const result = await importLabsCsv({
|
|
userId: USER,
|
|
tz: TZ,
|
|
stream: textToStream(
|
|
"Analyt;Wert;Einheit;Datum\nGlucose;7;mmHg;05.03.2025\n",
|
|
),
|
|
maxBytes: 1_000_000,
|
|
query: labsCsvQuerySchema.parse({
|
|
analyte: "0",
|
|
value: "1",
|
|
unit: "2",
|
|
date: "3",
|
|
decimal: "comma",
|
|
dateOrder: "dmy",
|
|
}),
|
|
});
|
|
expect(result.errors).toEqual([
|
|
{ line: 2, reason: "unit_mismatch", field: "unit" },
|
|
]);
|
|
});
|
|
});
|
|
|
|
describe("lab CSV import in a zone of the file's own", () => {
|
|
it("cuts the duplicate day in the zone the file is read in", async () => {
|
|
// Read in New York, these are draws on two days (1 and 2 March). In the
|
|
// account's Berlin they both fall on 2 March, which used to make the
|
|
// second a duplicate of the first.
|
|
const csv = [
|
|
"Analyt;Wert;Einheit;Datum",
|
|
"Zone marker;7;mg/dL;2025-03-01T22:30",
|
|
"Zone marker;7;mg/dL;2025-03-02T02:00",
|
|
].join("\n");
|
|
const result = await importLabsCsv({
|
|
userId: USER,
|
|
tz: TZ,
|
|
stream: textToStream(csv),
|
|
maxBytes: 1_000_000,
|
|
query: labsCsvQuerySchema.parse({
|
|
analyte: "0",
|
|
value: "1",
|
|
unit: "2",
|
|
date: "3",
|
|
timezone: "America/New_York",
|
|
}),
|
|
});
|
|
expect(result).toMatchObject({ inserted: 2, duplicates: 0 });
|
|
});
|
|
});
|
|
|
|
describe("lab CSV import at the row cap", () => {
|
|
const BIG_USER = "csv-import-big-user";
|
|
// 20 000 rows: 40 markers, 500 draw days each, one reading per day.
|
|
const lines = ["Analyt;Wert;Einheit;Datum"];
|
|
for (let i = 0; i < 20_000; i++) {
|
|
const marker = `Marker ${i % 40}`;
|
|
const day = new Date(Date.UTC(2020, 0, 1) + Math.floor(i / 40) * 86_400_000)
|
|
.toISOString()
|
|
.slice(0, 10);
|
|
lines.push(`${marker};${(i % 97) + 1};U/l;${day}`);
|
|
}
|
|
const BIG = lines.join("\n");
|
|
const query = () =>
|
|
labsCsvQuerySchema.parse({
|
|
analyte: "0",
|
|
value: "1",
|
|
unit: "2",
|
|
date: "3",
|
|
});
|
|
|
|
/** The file, failing like a dropped connection after `bytes`. */
|
|
function cutAt(bytes: number): ReadableStream<Uint8Array> {
|
|
const all = new TextEncoder().encode(BIG);
|
|
let offset = 0;
|
|
return new ReadableStream<Uint8Array>({
|
|
pull(controller) {
|
|
if (offset >= bytes) {
|
|
controller.error(new Error("connection reset"));
|
|
return;
|
|
}
|
|
controller.enqueue(all.slice(offset, Math.min(offset + 64_000, bytes)));
|
|
offset += 64_000;
|
|
},
|
|
});
|
|
}
|
|
|
|
beforeAll(async () => {
|
|
await prisma.user.create({
|
|
data: { id: BIG_USER, username: BIG_USER, timezone: TZ },
|
|
});
|
|
});
|
|
|
|
it("an import cut off part way leaves whole chunks, and a retry writes the rest once", async () => {
|
|
let written = 0;
|
|
await expect(
|
|
importLabsCsv({
|
|
userId: BIG_USER,
|
|
tz: TZ,
|
|
stream: cutAt(Math.floor(BIG.length / 3)),
|
|
maxBytes: 16 * 1024 * 1024,
|
|
query: query(),
|
|
onWritten: (n) => {
|
|
written += n;
|
|
},
|
|
}),
|
|
).rejects.toThrow("connection reset");
|
|
const afterCut = await prisma.labResult.count({
|
|
where: { userId: BIG_USER },
|
|
});
|
|
expect(afterCut).toBe(written);
|
|
expect(afterCut).toBeGreaterThan(0);
|
|
// Whole chunks only: nothing half-written.
|
|
expect(afterCut % 200).toBe(0);
|
|
|
|
const started = Date.now();
|
|
const result = await importLabsCsv({
|
|
userId: BIG_USER,
|
|
tz: TZ,
|
|
stream: textToStream(BIG),
|
|
maxBytes: 16 * 1024 * 1024,
|
|
query: query(),
|
|
});
|
|
const elapsed = Date.now() - started;
|
|
expect(result).toMatchObject({
|
|
rows: 20_000,
|
|
inserted: 20_000 - afterCut,
|
|
duplicates: afterCut,
|
|
failed: 0,
|
|
});
|
|
expect(await prisma.labResult.count({ where: { userId: BIG_USER } })).toBe(
|
|
20_000,
|
|
);
|
|
expect(await prisma.biomarker.count({ where: { userId: BIG_USER } })).toBe(
|
|
40,
|
|
);
|
|
// Generous: the chunked write takes a few seconds here; the bound only
|
|
// catches a return to one statement per row.
|
|
expect(elapsed).toBeLessThan(120_000);
|
|
}, 300_000);
|
|
});
|
|
|
|
describe("medication CSV import", () => {
|
|
const MEDS = [
|
|
"name,dose,times,as needed,start,end",
|
|
"Ramipril,5 mg,08:00,,2025-01-01,",
|
|
"Metformin,500 mg,08:00 20:00,,,",
|
|
"Ibuprofen,400 mg,,yes,,",
|
|
"Vitamin D,1000 IE,,,,",
|
|
"ramipril,5 MG,08:00,,,",
|
|
].join("\n");
|
|
const query = (dryRun: boolean) =>
|
|
medicationsCsvQuerySchema.parse({
|
|
name: "0",
|
|
dose: "1",
|
|
times: "2",
|
|
asNeeded: "3",
|
|
startsOn: "4",
|
|
endsOn: "5",
|
|
dryRun: dryRun ? "1" : "0",
|
|
});
|
|
|
|
it("checks, then imports, then skips everything on a second upload", async () => {
|
|
const dry = await importMedicationsCsv({
|
|
userId: USER,
|
|
tz: TZ,
|
|
stream: textToStream(MEDS),
|
|
maxBytes: 100_000,
|
|
query: query(true),
|
|
});
|
|
expect(dry).toMatchObject({
|
|
rows: 5,
|
|
inserted: 3,
|
|
duplicates: 1,
|
|
failed: 1,
|
|
});
|
|
expect(dry.errors).toEqual([
|
|
{ line: 5, reason: "missing_times", field: "times" },
|
|
{ line: 6, reason: "duplicate" },
|
|
]);
|
|
expect(await prisma.medication.count({ where: { userId: USER } })).toBe(0);
|
|
|
|
const real = await importMedicationsCsv({
|
|
userId: USER,
|
|
tz: TZ,
|
|
stream: textToStream(MEDS),
|
|
maxBytes: 100_000,
|
|
query: query(false),
|
|
});
|
|
expect(real).toMatchObject({ inserted: 3, duplicates: 1, failed: 1 });
|
|
const meds = await prisma.medication.findMany({
|
|
where: { userId: USER },
|
|
include: { schedules: true, courses: true },
|
|
orderBy: { name: "asc" },
|
|
});
|
|
const byName = new Map(meds.map((m) => [m.name, m]));
|
|
expect(byName.get("Metformin")?.schedules[0]).toMatchObject({
|
|
rrule: "FREQ=DAILY",
|
|
timesOfDay: ["08:00", "20:00"],
|
|
});
|
|
expect(byName.get("Metformin")?.notificationsEnabled).toBe(false);
|
|
expect(byName.get("Ibuprofen")).toMatchObject({ asNeeded: true });
|
|
expect(byName.get("Ibuprofen")?.schedules).toHaveLength(0);
|
|
expect(byName.get("Ramipril")?.courses).toHaveLength(1);
|
|
|
|
const again = await importMedicationsCsv({
|
|
userId: USER,
|
|
tz: TZ,
|
|
stream: textToStream(MEDS),
|
|
maxBytes: 100_000,
|
|
query: query(false),
|
|
});
|
|
expect(again).toMatchObject({ inserted: 0, duplicates: 4 });
|
|
});
|
|
});
|