refactor(medications): one admission path and one replay key for intake imports

The per-medication import and the account-wide dose-history import queued
their work through ninety near-identical lines each: row lock, stale-job
sweep, active-job refusal, job row, pg-boss send, and the fallback that fails
the row when the queue is not there. Five steps written twice drift, and the
drift shows up as an import that hangs on one surface and not the other. The
status projection moved out of these routes for the same reason; this is the
write half.

Scope is the only real difference, and it is one argument. What the two
surfaces legitimately say differently — the 409 wording, the envelope, the
kickoff audit line — stays at the call site.

The replay key had the same shape of problem in smaller form: the export
reader owned a function for it and the hand-listed body inlined the same
template. Both now call one function, with a test that pins them equal for
the same medication and instant.

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,
     );

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,

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),
+    );
+  });
+});

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,
     );

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<IntakeImportAdmission> {
+  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 };
+  }
+}

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),
     };
   });
 }