diff --git a/src/app/api/medications/[id]/intake/import/route.ts b/src/app/api/medications/[id]/intake/import/route.ts index fb18c1866..90fa2ac87 100644 --- a/src/app/api/medications/[id]/intake/import/route.ts +++ b/src/app/api/medications/[id]/intake/import/route.ts @@ -11,16 +11,13 @@ import { sanitiseZodIssues, } from "@/lib/api-response"; import { auditLog } from "@/lib/auth/audit"; -import { prisma, toJson } from "@/lib/db"; -import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { prisma } from "@/lib/db"; import { - MEDICATION_INTAKE_IMPORT_QUEUE, - MEDICATION_INTAKE_IMPORT_SEND_OPTIONS, - MEDICATION_INTAKE_IMPORT_STALE_AFTER_MS, type MedicationImportPayload, type MedicationImportProgress, } from "@/lib/jobs/medication-intake-import"; import { annotate } from "@/lib/logging/context"; +import { admitIntakeImportJob } from "@/lib/medications/intake-import-admission"; import { buildMedicationImportEntries } from "@/lib/medications/intake-import-payload"; import { assertMedicationOwnership } from "@/lib/medications/route-guards"; import { checkRateLimit } from "@/lib/rate-limit"; @@ -181,111 +178,26 @@ export const POST = apiHandler( touchedDays: [], rollupProcessed: 0, }; - const admission = await prisma.$transaction(async (tx) => { - await tx.$queryRaw` - SELECT "id" - FROM "medications" - WHERE "id" = ${id} - AND "user_id" = ${user.id} - FOR UPDATE - `; - const now = new Date(); - const staleBefore = new Date( - now.getTime() - MEDICATION_INTAKE_IMPORT_STALE_AFTER_MS, - ); - await tx.medicationIntakeImportJob.updateMany({ - where: { - userId: user.id, - medicationId: id, - status: { in: ["queued", "running"] }, - OR: [ - { heartbeatAt: { lt: staleBefore } }, - { - heartbeatAt: null, - startedAt: { lt: staleBefore }, - }, - { - heartbeatAt: null, - startedAt: null, - createdAt: { lt: staleBefore }, - }, - ], - }, - data: { - status: "failed", - failureReason: "Medication intake import abandoned", - heartbeatAt: now, - completedAt: now, - }, - }); - const activeJob = await tx.medicationIntakeImportJob.findFirst({ - where: { - userId: user.id, - medicationId: id, - status: { in: ["queued", "running"] }, - }, - select: { id: true }, - }); - if (activeJob) return { active: true as const }; - - const importJob = await tx.medicationIntakeImportJob.create({ - data: { - userId: user.id, - medicationId: id, - status: "queued", - payload: toJson(normalized), - progress: toJson(progress), - }, - }); - return { active: false as const, importJob }; + const admission = await admitIntakeImportJob({ + userId: user.id, + medicationId: id, + payload: normalized, + progress, + ipAddress: getClientIp(request), }); - if (admission.active) { + if (!admission.admitted) { return apiError("Medication intake import already in progress", 409); } - const { importJob } = admission; - - let bossJobId: string | null = null; - try { - const boss = getGlobalBoss(); - if (!boss) throw new Error("worker unavailable"); - bossJobId = await boss.send( - MEDICATION_INTAKE_IMPORT_QUEUE, - { jobId: importJob.id }, - MEDICATION_INTAKE_IMPORT_SEND_OPTIONS, - ); - if (!bossJobId) throw new Error("queue rejected job"); - await prisma.medicationIntakeImportJob.update({ - where: { id: importJob.id }, - data: { pgBossJobId: bossJobId }, - }); - } catch { - const completedAt = new Date(); - await prisma.medicationIntakeImportJob.update({ - where: { id: importJob.id }, - data: { - status: "failed", - failureReason: "Background worker enqueue failed", - heartbeatAt: completedAt, - completedAt, - }, - }); - await auditLog("medication.intake.import.kickoff.denied", { - userId: user.id, - ipAddress: getClientIp(request), - details: { - jobId: importJob.id, - medicationId: id, - reason: "enqueue_failed", - }, - }); + if (!admission.enqueued) { return apiError("Background worker is not available", 503); } + const { jobId, bossJobId } = admission; await auditLog("medication.intake.import.kickoff", { userId: user.id, ipAddress: getClientIp(request), details: { - jobId: importJob.id, + jobId, bossJobId, medicationId: id, total: normalized.entries.length, @@ -298,16 +210,16 @@ export const POST = apiHandler( entity_id: id, }, meta: { - job_id: importJob.id, + job_id: jobId, total: normalized.entries.length, }, }); return apiSuccess( { - jobId: importJob.id, + jobId, status: "queued" as const, - statusUrl: `/api/medications/${id}/intake/import/${importJob.id}/status`, + statusUrl: `/api/medications/${id}/intake/import/${jobId}/status`, }, 202, ); diff --git a/src/app/api/medications/intake/dose-history-import/route.ts b/src/app/api/medications/intake/dose-history-import/route.ts index df4150bd6..362840a42 100644 --- a/src/app/api/medications/intake/dose-history-import/route.ts +++ b/src/app/api/medications/intake/dose-history-import/route.ts @@ -32,18 +32,15 @@ import { NextRequest } from "next/server"; import { apiHandler, requireAuth } from "@/lib/api-handler"; import { apiError, apiSuccess, getClientIp } from "@/lib/api-response"; import { auditLog } from "@/lib/auth/audit"; -import { prisma, toJson } from "@/lib/db"; -import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { prisma } from "@/lib/db"; import { - MEDICATION_INTAKE_IMPORT_QUEUE, - MEDICATION_INTAKE_IMPORT_SEND_OPTIONS, - MEDICATION_INTAKE_IMPORT_STALE_AFTER_MS, groupMedicationImportSkips, totalMedicationImportSkips, type MedicationImportPayload, type MedicationImportProgress, } from "@/lib/jobs/medication-intake-import"; import { annotate } from "@/lib/logging/context"; +import { admitIntakeImportJob } from "@/lib/medications/intake-import-admission"; import { parseAutoExportCsv, parseAutoExportJson, @@ -190,104 +187,28 @@ export const POST = apiHandler(async (request: NextRequest) => { rollupProcessed: 0, }; - const admission = await prisma.$transaction(async (tx) => { - await tx.$queryRaw` - SELECT "id" - FROM "users" - WHERE "id" = ${user.id} - FOR UPDATE - `; - const now = new Date(); - const staleBefore = new Date( - now.getTime() - MEDICATION_INTAKE_IMPORT_STALE_AFTER_MS, - ); - await tx.medicationIntakeImportJob.updateMany({ - where: { - userId: user.id, - medicationId: null, - status: { in: ["queued", "running"] }, - OR: [ - { heartbeatAt: { lt: staleBefore } }, - { heartbeatAt: null, startedAt: { lt: staleBefore } }, - { - heartbeatAt: null, - startedAt: null, - createdAt: { lt: staleBefore }, - }, - ], - }, - data: { - status: "failed", - failureReason: "Medication intake import abandoned", - heartbeatAt: now, - completedAt: now, - }, - }); - const activeJob = await tx.medicationIntakeImportJob.findFirst({ - where: { - userId: user.id, - medicationId: null, - status: { in: ["queued", "running"] }, - }, - select: { id: true }, - }); - if (activeJob) return { active: true as const }; - - const importJob = await tx.medicationIntakeImportJob.create({ - data: { - userId: user.id, - medicationId: null, - status: "queued", - payload: toJson(payload), - progress: toJson(progress), - }, - }); - return { active: false as const, importJob }; + const admission = await admitIntakeImportJob({ + userId: user.id, + medicationId: null, + payload, + progress, + ipAddress: getClientIp(request), }); - if (admission.active) { + if (!admission.admitted) { return apiError("A dose-history import is already in progress", 409); } - const { importJob } = admission; - - const statusUrl = `/api/medications/intake/dose-history-import/${importJob.id}/status`; - let bossJobId: string | null = null; - try { - const boss = getGlobalBoss(); - if (!boss) throw new Error("worker unavailable"); - bossJobId = await boss.send( - MEDICATION_INTAKE_IMPORT_QUEUE, - { jobId: importJob.id }, - MEDICATION_INTAKE_IMPORT_SEND_OPTIONS, - ); - if (!bossJobId) throw new Error("queue rejected job"); - await prisma.medicationIntakeImportJob.update({ - where: { id: importJob.id }, - data: { pgBossJobId: bossJobId }, - }); - } catch { - const completedAt = new Date(); - await prisma.medicationIntakeImportJob.update({ - where: { id: importJob.id }, - data: { - status: "failed", - failureReason: "Background worker enqueue failed", - heartbeatAt: completedAt, - completedAt, - }, - }); - await auditLog("medication.intake.import.kickoff.denied", { - userId: user.id, - ipAddress: getClientIp(request), - details: { jobId: importJob.id, reason: "enqueue_failed" }, - }); + if (!admission.enqueued) { return apiError("Background worker is not available", 503); } + const { jobId, bossJobId } = admission; + + const statusUrl = `/api/medications/intake/dose-history-import/${jobId}/status`; await auditLog("medication.intake.history.import.kickoff", { userId: user.id, ipAddress: getClientIp(request), details: { - jobId: importJob.id, + jobId, bossJobId, rowsRead: file.rowsRead, queued: file.queued, @@ -297,7 +218,7 @@ export const POST = apiHandler(async (request: NextRequest) => { annotate({ action: { name: "medication.intake.history.import.kickoff" }, meta: { - job_id: importJob.id, + job_id: jobId, rows_read: file.rowsRead, queued: file.queued, refused: file.refused, @@ -307,7 +228,7 @@ export const POST = apiHandler(async (request: NextRequest) => { return apiSuccess( { dryRun: false, - jobId: importJob.id, + jobId, status: "queued" as const, statusUrl, file, diff --git a/src/lib/medications/__tests__/intake-import-idempotency-key.test.ts b/src/lib/medications/__tests__/intake-import-idempotency-key.test.ts new file mode 100644 index 000000000..c981d5012 --- /dev/null +++ b/src/lib/medications/__tests__/intake-import-idempotency-key.test.ts @@ -0,0 +1,67 @@ +/** + * Both intake-import surfaces must derive the same replay key. + * + * The key is what makes a re-run cost nothing: it matches the live-row unique + * on `(user_id, medication_id, scheduled_for, source)`, so a second submission + * of the same dose collides instead of writing a second ledger row. The two + * surfaces — the hand-listed per-medication body and the export reader — used + * to build the string separately, which meant a change to one silently stopped + * the other from deduplicating its replays. They now share one function; this + * pins that they agree for the same medication and instant, so a future split + * fails here rather than in a user's ledger. + */ +import { describe, expect, it } from "vitest"; + +import { planAutoExportImport } from "../import/auto-export-plan"; +import { parseAutoExportCsv } from "../import/auto-export-parse"; +import { + buildMedicationImportEntries, + medicationImportIdempotencyKey, +} from "../intake-import-payload"; + +const MEDICATION_ID = "med_shared"; +/** The instant both surfaces are asked about, expressed each one's own way. */ +const LOCAL_DATE = "2025-03-09"; +const LOCAL_TIME = "07:30"; +const INSTANT = new Date(`${LOCAL_DATE}T${LOCAL_TIME}`); + +const CSV_HEADER = + "Date,Scheduled Date,Medication,Nickname,Dose Taken,Dose,Unit,Status,Skipped,Notes"; + +function offsetSuffix(at: Date): string { + const minutes = -at.getTimezoneOffset(); + const sign = minutes < 0 ? "-" : "+"; + const abs = Math.abs(minutes); + const pad = (n: number) => String(n).padStart(2, "0"); + return `${sign}${pad(Math.floor(abs / 60))}${pad(abs % 60)}`; +} + +describe("intake-import replay keys", () => { + it("derives the key from the medication and the slot instant only", () => { + expect(medicationImportIdempotencyKey(MEDICATION_ID, INSTANT)).toBe( + `import-${MEDICATION_ID}-${INSTANT.getTime()}`, + ); + }); + + it("gives the hand-listed body and the export reader the same key", () => { + const [fromBody] = buildMedicationImportEntries(MEDICATION_ID, [ + { datum: LOCAL_DATE, uhrzeit: LOCAL_TIME }, + ]); + + // The export names its instants with an explicit offset, so the same wall + // time is written here with the local offset the body row implies. + const stamp = `${LOCAL_DATE} ${LOCAL_TIME}:00 ${offsetSuffix(INSTANT)}`; + const parsed = parseAutoExportCsv( + `${CSV_HEADER}\n${stamp},${stamp},Ramipril,,1.0,1.0,count,Taken,No,\n`, + ); + const plan = planAutoExportImport(parsed, [ + { id: MEDICATION_ID, name: "Ramipril", externalSource: null }, + ]); + + expect(plan.entries).toHaveLength(1); + expect(plan.entries[0].idempotencyKey).toBe(fromBody.idempotencyKey); + expect(fromBody.idempotencyKey).toBe( + medicationImportIdempotencyKey(MEDICATION_ID, INSTANT), + ); + }); +}); diff --git a/src/lib/medications/import/auto-export-plan.ts b/src/lib/medications/import/auto-export-plan.ts index 5ca81b81c..a2c3f41ea 100644 --- a/src/lib/medications/import/auto-export-plan.ts +++ b/src/lib/medications/import/auto-export-plan.ts @@ -14,6 +14,7 @@ import type { MedicationImportSkipReason, } from "@/lib/jobs/medication-intake-import"; +import { medicationImportIdempotencyKey } from "@/lib/medications/intake-import-payload"; import { normaliseMedicationName } from "./auto-export-format"; import type { AutoExportDose, @@ -62,22 +63,6 @@ export interface AutoExportPlan { unknownColumns: string[]; } -/** - * An imported dose is identified by its medication and its slot instant. - * - * That is the grain the database already enforces through the live-row unique on - * `(user_id, medication_id, scheduled_for, source)`, so the replay key is a - * function of exactly the same facts. It was previously built from an optional - * counter field in the request body, which collapsed a month of distinct doses - * onto one key as soon as an export put something non-unique there. - */ -export function autoExportIdempotencyKey( - medicationId: string, - scheduledFor: Date, -): string { - return `import-${medicationId}-${scheduledFor.getTime()}`; -} - function bump( counts: MedicationImportSkipCounts, reason: MedicationImportSkipReason, @@ -178,7 +163,7 @@ export function planAutoExportImport( continue; } - const idempotencyKey = autoExportIdempotencyKey( + const idempotencyKey = medicationImportIdempotencyKey( match.id, dose.scheduledFor, ); diff --git a/src/lib/medications/intake-import-admission.ts b/src/lib/medications/intake-import-admission.ts new file mode 100644 index 000000000..4feacd5d6 --- /dev/null +++ b/src/lib/medications/intake-import-admission.ts @@ -0,0 +1,182 @@ +/** + * The one admission path for an intake-import job. + * + * Two routes queue this work — the per-medication import under + * `/api/medications/[id]/…` and the account-wide dose-history import — and both + * have to do the same five things in the same order: take a row lock so two + * submissions cannot race past each other, fail out any job the worker + * abandoned, refuse if one is genuinely still running, write the job row, and + * hand it to pg-boss with a fallback that marks the row failed when the queue + * is not there. Written twice, those five steps drift: one route grows a + * staleness clause the other does not, and the difference only shows up as an + * import that hangs forever on one surface and not the other. + * + * `readMedicationIntakeImportJob` took the read half of this pair out of the + * routes for the same reason. This is the write half. + * + * The only real difference between the two callers is scope, and it is one + * argument: `medicationId` narrows the lock, the active-job check and the job + * row to a single medication, `null` means the whole account. Everything the + * caller still owns — the 409 wording, the success envelope, the kickoff audit + * line — stays at the call site, because those are the parts the two surfaces + * legitimately say differently. + */ +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { + MEDICATION_INTAKE_IMPORT_QUEUE, + MEDICATION_INTAKE_IMPORT_SEND_OPTIONS, + MEDICATION_INTAKE_IMPORT_STALE_AFTER_MS, + type MedicationImportPayload, + type MedicationImportProgress, +} from "@/lib/jobs/medication-intake-import"; +import { auditLog } from "@/lib/auth/audit"; +import { prisma, toJson } from "@/lib/db"; + +export interface IntakeImportAdmissionInput { + userId: string; + /** `null` = the account-wide dose-history import. */ + medicationId: string | null; + payload: MedicationImportPayload; + progress: MedicationImportProgress; + /** Threaded onto the enqueue-failure audit row; `null` when unresolved. */ + ipAddress: string | null; +} + +export type IntakeImportAdmission = + /** An import for this scope is already queued or running. */ + | { admitted: false } + /** + * The job row exists but the queue would not take it, so the row is already + * marked failed and the audit line is written. The caller answers 503. + */ + | { admitted: true; enqueued: false; jobId: string } + | { admitted: true; enqueued: true; jobId: string; bossJobId: string }; + +/** + * Claim the import slot for one scope and hand the job to the worker. + * + * The lock is taken on the row the scope belongs to — the medication for a + * per-medication import, the account for the dose-history import — so two + * submissions serialise against each other without blocking an unrelated one. + */ +export async function admitIntakeImportJob({ + userId, + medicationId, + payload, + progress, + ipAddress, +}: IntakeImportAdmissionInput): Promise { + const admission = await prisma.$transaction(async (tx) => { + if (medicationId === null) { + await tx.$queryRaw` + SELECT "id" + FROM "users" + WHERE "id" = ${userId} + FOR UPDATE + `; + } else { + await tx.$queryRaw` + SELECT "id" + FROM "medications" + WHERE "id" = ${medicationId} + AND "user_id" = ${userId} + FOR UPDATE + `; + } + + const now = new Date(); + const staleBefore = new Date( + now.getTime() - MEDICATION_INTAKE_IMPORT_STALE_AFTER_MS, + ); + // A worker that died mid-run leaves a row that says "running" forever, and + // the active-job check below would then refuse every later submission. The + // three clauses cover the three points a run can stop at: after a + // heartbeat, after the start, and before either. + await tx.medicationIntakeImportJob.updateMany({ + where: { + userId, + medicationId, + status: { in: ["queued", "running"] }, + OR: [ + { heartbeatAt: { lt: staleBefore } }, + { heartbeatAt: null, startedAt: { lt: staleBefore } }, + { + heartbeatAt: null, + startedAt: null, + createdAt: { lt: staleBefore }, + }, + ], + }, + data: { + status: "failed", + failureReason: "Medication intake import abandoned", + heartbeatAt: now, + completedAt: now, + }, + }); + + const activeJob = await tx.medicationIntakeImportJob.findFirst({ + where: { + userId, + medicationId, + status: { in: ["queued", "running"] }, + }, + select: { id: true }, + }); + if (activeJob) return { active: true as const }; + + const importJob = await tx.medicationIntakeImportJob.create({ + data: { + userId, + medicationId, + status: "queued", + payload: toJson(payload), + progress: toJson(progress), + }, + }); + return { active: false as const, importJob }; + }); + + if (admission.active) return { admitted: false }; + const jobId = admission.importJob.id; + + try { + const boss = getGlobalBoss(); + if (!boss) throw new Error("worker unavailable"); + const bossJobId = await boss.send( + MEDICATION_INTAKE_IMPORT_QUEUE, + { jobId }, + MEDICATION_INTAKE_IMPORT_SEND_OPTIONS, + ); + if (!bossJobId) throw new Error("queue rejected job"); + await prisma.medicationIntakeImportJob.update({ + where: { id: jobId }, + data: { pgBossJobId: bossJobId }, + }); + return { admitted: true, enqueued: true, jobId, bossJobId }; + } catch { + // The row is already written, so leaving it queued would park the scope + // behind a job nobody will ever pick up. Fail it here and the next + // submission is admitted immediately. + const completedAt = new Date(); + await prisma.medicationIntakeImportJob.update({ + where: { id: jobId }, + data: { + status: "failed", + failureReason: "Background worker enqueue failed", + heartbeatAt: completedAt, + completedAt, + }, + }); + await auditLog("medication.intake.import.kickoff.denied", { + userId, + ipAddress, + details: { + jobId, + ...(medicationId === null ? {} : { medicationId }), + reason: "enqueue_failed", + }, + }); + return { admitted: true, enqueued: false, jobId }; + } +} diff --git a/src/lib/medications/intake-import-payload.ts b/src/lib/medications/intake-import-payload.ts index da60d6944..a1d13d85e 100644 --- a/src/lib/medications/intake-import-payload.ts +++ b/src/lib/medications/intake-import-payload.ts @@ -19,8 +19,15 @@ export interface MedicationImportQueueEntry { } /** - * An imported dose is identified by its medication and its instant, and by - * nothing else. + * The replay key for one imported dose: its medication and its slot instant, + * and nothing else. + * + * That is the grain the database already enforces through the live-row unique + * on `(user_id, medication_id, scheduled_for, source)`, so the key is a + * function of exactly the same facts. Both import surfaces call it — the + * hand-listed per-medication body and the export reader — because a key + * written out twice is a key that can disagree with itself, and two callers + * that disagree stop deduplicating each other's replays. * * The key used to be built from the row's optional `zaehler` field whenever it * was present. That reads as a per-row counter, but nothing required it to be @@ -33,6 +40,20 @@ export interface MedicationImportQueueEntry { * and matches the live-row unique on `(user_id, medication_id, scheduled_for, * source)` that the database enforces regardless. */ +export function medicationImportIdempotencyKey( + medicationId: string, + scheduledFor: Date, +): string { + return `import-${medicationId}-${scheduledFor.getTime()}`; +} + +/** + * Derive the queued entries for a hand-listed import body. + * + * It lives here rather than inline in the route so the test that proves a + * reported export lands its rows can drive the real derivation instead of + * restating it. A test that mirrors the production formula proves the mirror. + */ export function buildMedicationImportEntries( medicationId: string, rows: readonly MedicationImportRow[], @@ -46,7 +67,7 @@ export function buildMedicationImportEntries( // The export-reading route is the one that has both facts to give. scheduledFor: takenAt.toISOString(), takenAt: takenAt.toISOString(), - idempotencyKey: `import-${medicationId}-${takenAt.getTime()}`, + idempotencyKey: medicationImportIdempotencyKey(medicationId, takenAt), }; }); }