fix: retain import actor attribution

Persist record and actor identities through job completion.\nWrite one bounded transactional completion audit per job.

prisma/migrations/0299_medication_intake_import_actor_attribution/migration.sql

@@ -0,0 +1,28 @@
+-- Keep the record subject and the actor distinct for durable intake-import
+-- work. Historical rows predate delegated imports, so their actor is the
+-- record subject: the only truthful attribution available for those jobs.
+ALTER TABLE "medication_intake_import_jobs"
+  RENAME COLUMN "user_id" TO "record_user_id";
+
+ALTER TABLE "medication_intake_import_jobs"
+  RENAME CONSTRAINT "medication_intake_import_jobs_user_id_fkey"
+  TO "medication_intake_import_jobs_record_user_id_fkey";
+
+ALTER TABLE "medication_intake_import_jobs"
+  ADD COLUMN "actor_user_id" TEXT;
+
+UPDATE "medication_intake_import_jobs"
+  SET "actor_user_id" = "record_user_id"
+  WHERE "actor_user_id" IS NULL;
+
+ALTER TABLE "medication_intake_import_jobs"
+  ALTER COLUMN "actor_user_id" SET NOT NULL;
+
+-- No actor foreign key: a deleted delegate must remain identifiable in the
+-- record's durable job and audit trail, just like audit_logs.actor_user_id.
+
+DROP INDEX "medication_intake_import_jobs_user_id_created_at_idx";
+CREATE INDEX "medication_intake_import_jobs_record_user_id_created_at_idx"
+  ON "medication_intake_import_jobs"("record_user_id", "created_at");
+CREATE INDEX "medication_intake_import_jobs_actor_user_id_created_at_idx"
+  ON "medication_intake_import_jobs"("actor_user_id", "created_at");

prisma/schema.prisma

@@ -816,7 +816,7 @@ model User {
   // pg-boss job state so the polling endpoint can read live per-
   // phase progress without round-tripping through pg-boss internals.
   importJobs                     ImportJob[]
-  medicationIntakeImportJobs     MedicationIntakeImportJob[]
+  medicationIntakeImportJobs     MedicationIntakeImportJob[] @relation("MedicationIntakeImportRecord")
   // v1.5.0 — persistent measurement rollup table. Carries the
   // count/min/max/mean/sd/slope/r2 stats per (type, granularity,
   // bucketStart) so the analytics + comprehensive read paths bypass
@@ -4715,7 +4715,8 @@ model ImportJob {
 // stays in Postgres so retries never depend on request-process memory.
 model MedicationIntakeImportJob {
   id            String    @id @default(cuid())
-  userId        String    @map("user_id")
+  recordUserId  String    @map("record_user_id")
+  actorUserId   String    @map("actor_user_id")
   /// v1.33.0 (migration 0278) — nullable. The per-medication import route
   /// scopes its job to one medication and names it here. A dose history
   /// exported from another tracker covers a whole regimen in one file — the
@@ -4736,10 +4737,11 @@ model MedicationIntakeImportJob {
   createdAt     DateTime  @default(now()) @map("created_at")
   updatedAt     DateTime  @updatedAt @map("updated_at")
 
-  user       User        @relation(fields: [userId], references: [id], onDelete: Cascade)
+  recordUser User        @relation("MedicationIntakeImportRecord", fields: [recordUserId], references: [id], onDelete: Cascade)
   medication Medication? @relation(fields: [medicationId], references: [id], onDelete: Cascade)
 
-  @@index([userId, createdAt])
+  @@index([recordUserId, createdAt])
+  @@index([actorUserId, createdAt])
   @@index([medicationId, createdAt])
   @@index([status, heartbeatAt])
   @@map("medication_intake_import_jobs")

src/app/api/medications/[id]/intake/import/[jobId]/status/__tests__/route.test.ts

@@ -61,7 +61,11 @@ describe("GET medication intake import status", () => {
 
     expect(response.status).toBe(404);
     expect(prisma.medicationIntakeImportJob.findFirst).toHaveBeenCalledWith({
-      where: { id: "job-1", medicationId: "med-1", userId: "user-1" },
+      where: {
+        id: "job-1",
+        medicationId: "med-1",
+        recordUserId: "user-1",
+      },
     });
   });
 

src/app/api/medications/[id]/intake/import/[jobId]/status/route.ts

@@ -1,6 +1,6 @@
 import type { NextRequest } from "next/server";
 
-import { apiHandler, requireAuth } from "@/lib/api-handler";
+import { apiHandler, requireRecordAuth } from "@/lib/api-handler";
 import { apiError, apiSuccess } from "@/lib/api-response";
 import { annotate } from "@/lib/logging/context";
 import { readMedicationIntakeImportJob } from "@/lib/medications/intake-import-job-status";
@@ -13,7 +13,7 @@ export const dynamic = "force-dynamic";
 
 export const GET = apiHandler(
   async (_request: NextRequest, { params }: RouteContext) => {
-    const { user } = await requireAuth();
+    const { user } = await requireRecordAuth("manage", "medications");
     const { id: medicationId, jobId } = await params;
     annotate({
       action: { name: "medication.intake.import.status" },

src/app/api/medications/[id]/intake/import/__tests__/route.test.ts

@@ -237,7 +237,8 @@ describe("POST /api/medications/[id]/intake/import — durable kickoff", () => {
 
     expect(prisma.medicationIntakeImportJob.create).toHaveBeenCalledWith({
       data: expect.objectContaining({
-        userId: "user-1",
+        recordUserId: "user-1",
+        actorUserId: "user-1",
         medicationId: "m1",
         status: "queued",
         // A legacy `zaehler` on the first row rides along and decides nothing:
@@ -303,7 +304,7 @@ describe("POST /api/medications/[id]/intake/import — durable kickoff", () => {
       expect(prisma.medicationIntakeImportJob.updateMany).toHaveBeenCalledWith({
         where: expect.objectContaining({
           medicationId: "m1",
-          userId: "user-1",
+          recordUserId: "user-1",
           status: { in: ["queued", "running"] },
           OR: expect.arrayContaining([
             {

src/app/api/medications/[id]/intake/import/route.ts

@@ -187,7 +187,8 @@ export const POST = apiHandler(
       rollupProcessed: 0,
     };
     const admission = await admitIntakeImportJob({
-      userId: user.id,
+      recordUserId: user.id,
+      actorUserId: actor.id,
       medicationId: id,
       payload: normalized,
       progress,

src/app/api/medications/intake/dose-history-import/[jobId]/status/route.ts

@@ -1,6 +1,6 @@
 import type { NextRequest } from "next/server";
 
-import { apiHandler, requireAuth } from "@/lib/api-handler";
+import { apiHandler, requireRecordAuth } from "@/lib/api-handler";
 import { apiError, apiSuccess } from "@/lib/api-response";
 import { annotate } from "@/lib/logging/context";
 import { readMedicationIntakeImportJob } from "@/lib/medications/intake-import-job-status";
@@ -13,7 +13,7 @@ export const dynamic = "force-dynamic";
 
 export const GET = apiHandler(
   async (_request: NextRequest, { params }: RouteContext) => {
-    const { user } = await requireAuth();
+    const { user } = await requireRecordAuth("manage", "medications");
     const { jobId } = await params;
     annotate({
       action: { name: "medication.intake.history.import.status" },

src/app/api/medications/intake/dose-history-import/__tests__/route.test.ts

@@ -148,7 +148,8 @@ describe("POST /api/medications/intake/dose-history-import", () => {
     await POST(postReq(CSV));
     expect(prisma.medicationIntakeImportJob.create).toHaveBeenCalledWith({
       data: {
-        userId: "user-1",
+        recordUserId: "user-1",
+        actorUserId: "user-1",
         // No medication owns a file covering a regimen; each entry names its own.
         medicationId: null,
         status: "queued",
@@ -258,7 +259,7 @@ describe("POST /api/medications/intake/dose-history-import", () => {
     // reach across and fail it.
     expect(prisma.medicationIntakeImportJob.findFirst).toHaveBeenCalledWith({
       where: {
-        userId: "user-1",
+        recordUserId: "user-1",
         medicationId: null,
         status: { in: ["queued", "running"] },
       },
@@ -267,7 +268,7 @@ describe("POST /api/medications/intake/dose-history-import", () => {
     expect(prisma.medicationIntakeImportJob.updateMany).toHaveBeenCalledWith(
       expect.objectContaining({
         where: expect.objectContaining({
-          userId: "user-1",
+          recordUserId: "user-1",
           medicationId: null,
         }),
       }),

src/app/api/medications/intake/dose-history-import/route.ts

@@ -197,7 +197,8 @@ export const POST = apiHandler(async (request: NextRequest) => {
   };
 
   const admission = await admitIntakeImportJob({
-    userId: user.id,
+    recordUserId: user.id,
+    actorUserId: actor.id,
     medicationId: null,
     payload,
     progress,

src/lib/auth/audit.ts

@@ -49,16 +49,24 @@ export async function auditLog(
   action: string,
   opts: {
     userId?: string | null;
+    /** Explicit worker attribution when no request context remains. */
+    actorUserId?: string | null;
+    /** Use the active transaction when the audit must commit atomically. */
+    client?: Pick<typeof prisma, "auditLog">;
     details?: Record<string, unknown>;
     ipAddress?: string | null;
   } = {},
 ) {
   const userId = opts.userId ?? null;
-  const entry = await prisma.auditLog.create({
+  const client = opts.client ?? prisma;
+  const entry = await client.auditLog.create({
     data: {
       action,
       userId,
-      actorUserId: actingActorFor(userId),
+      actorUserId:
+        opts.actorUserId === undefined
+          ? actingActorFor(userId)
+          : opts.actorUserId,
       details: opts.details ? JSON.stringify(opts.details) : null,
       ipAddress: opts.ipAddress ?? null,
     },

src/lib/jobs/__tests__/medication-intake-import.test.ts

@@ -125,13 +125,14 @@ describe("medication intake import worker bounds", () => {
           medicationIntakeImportJob: {
             findUnique: async () => ({
               id: "job-many-days",
-              userId: "user-1",
+              recordUserId: "user-1",
+              actorUserId: "user-1",
               medicationId: "medication-1",
               status,
               payload: { entries: [] },
               progress: persistedProgress,
               startedAt: new Date(),
-              user: { timezone: "UTC" },
+              recordUser: { timezone: "UTC" },
             }),
             update: async (args: {
               data: { progress?: unknown; status?: string };

src/lib/jobs/medication-intake-import.ts

@@ -2,6 +2,7 @@ import type { Job, JobWithMetadata } from "pg-boss";
 import { Prisma, type PrismaClient } from "@/generated/prisma/client";
 
 import { invalidateUserMedications } from "@/lib/cache/invalidate";
+import { auditLog } from "@/lib/auth/audit";
 import { prisma, toJson } from "@/lib/db";
 import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
 import { consumeImportedIntakesBatch } from "@/lib/medications/inventory/consumption";
@@ -584,14 +585,14 @@ async function processNextChunk(
 
       const row = await tx.medicationIntakeImportJob.findUnique({
         where: { id: jobId },
-        include: { user: { select: { timezone: true } } },
+        include: { recordUser: { select: { timezone: true } } },
       });
       if (!row || row.status === "done" || row.status === "failed") {
         return {
           terminal: true,
           finalized: false,
           result: null,
-          userId: row?.userId ?? null,
+          userId: row?.recordUserId ?? null,
         };
       }
 
@@ -604,9 +605,9 @@ async function processNextChunk(
         );
         await recomputeTouchedDays(
           tx,
-          row.userId,
+          row.recordUserId,
           row.medicationId,
-          row.user.timezone,
+          row.recordUser.timezone,
           rollupDays,
         );
         const finalizedProgress: MedicationImportProgress = {
@@ -629,7 +630,7 @@ async function processNextChunk(
             terminal: false,
             finalized: false,
             result: null,
-            userId: row.userId,
+            userId: row.recordUserId,
           };
         }
         const result: MedicationImportResult = {
@@ -644,18 +645,17 @@ async function processNextChunk(
             : { skippedDetailsOmitted: progress.skippedDetailsOmitted }),
         };
         const completedAt = new Date();
-        await tx.auditLog.create({
-          data: {
-            userId: row.userId,
-            action: "medication.intake.import",
-            details: JSON.stringify({
-              jobId: row.id,
-              medicationId: row.medicationId,
-              imported: result.imported,
-              skipped: result.skipped,
-              skipReasons: result.skipReasons,
-              total: progress.total,
-            }),
+        await auditLog("medication.intake.import", {
+          client: tx,
+          userId: row.recordUserId,
+          actorUserId: row.actorUserId,
+          details: {
+            jobId: row.id,
+            recordUserId: row.recordUserId,
+            actorUserId: row.actorUserId,
+            imported: result.imported,
+            skipped: result.skipped,
+            total: progress.total,
           },
         });
         await tx.medicationIntakeImportJob.update({
@@ -673,7 +673,7 @@ async function processNextChunk(
           terminal: true,
           finalized: true,
           result,
-          userId: row.userId,
+          userId: row.recordUserId,
         };
       }
 
@@ -714,7 +714,7 @@ async function processNextChunk(
           const takenAt =
             entry.takenAt === null ? null : new Date(entry.takenAt);
           return {
-            userId: row.userId,
+            userId: row.recordUserId,
             medicationId: resolveMedicationId(entry),
             // The scheduled slot and the time taken are two separate facts, and
             // the export states both. Writing the take into both columns — which
@@ -774,7 +774,7 @@ async function processNextChunk(
       for (const [medicationId, events] of consumedByMedication) {
         await consumeImportedIntakesBatch({
           client: tx,
-          userId: row.userId,
+          userId: row.recordUserId,
           medicationId,
           events,
         });
@@ -783,7 +783,7 @@ async function processNextChunk(
       const touchedDays = created.map((event) =>
         encodeTouchedDay(
           event.medicationId,
-          dayKeyForScheduledFor(event.scheduledFor, row.user.timezone),
+          dayKeyForScheduledFor(event.scheduledFor, row.recordUser.timezone),
         ),
       );
       // Two different facts, counted apart. The Map collapse above is entries
@@ -818,7 +818,7 @@ async function processNextChunk(
         terminal: false,
         finalized: false,
         result: null,
-        userId: row.userId,
+        userId: row.recordUserId,
       };
     },
     { timeout: 60_000 },

src/lib/medications/intake-import-admission.ts

@@ -33,7 +33,8 @@ import { auditLog } from "@/lib/auth/audit";
 import { prisma, toJson } from "@/lib/db";
 
 export interface IntakeImportAdmissionInput {
-  userId: string;
+  recordUserId: string;
+  actorUserId: string;
   /** `null` = the account-wide dose-history import. */
   medicationId: string | null;
   payload: MedicationImportPayload;
@@ -60,7 +61,8 @@ export type IntakeImportAdmission =
  * submissions serialise against each other without blocking an unrelated one.
  */
 export async function admitIntakeImportJob({
-  userId,
+  recordUserId,
+  actorUserId,
   medicationId,
   payload,
   progress,
@@ -71,7 +73,7 @@ export async function admitIntakeImportJob({
       await tx.$queryRaw`
         SELECT "id"
         FROM "users"
-        WHERE "id" = ${userId}
+        WHERE "id" = ${recordUserId}
         FOR UPDATE
       `;
     } else {
@@ -79,7 +81,7 @@ export async function admitIntakeImportJob({
         SELECT "id"
         FROM "medications"
         WHERE "id" = ${medicationId}
-          AND "user_id" = ${userId}
+          AND "user_id" = ${recordUserId}
         FOR UPDATE
       `;
     }
@@ -94,7 +96,7 @@ export async function admitIntakeImportJob({
     // heartbeat, after the start, and before either.
     await tx.medicationIntakeImportJob.updateMany({
       where: {
-        userId,
+        recordUserId,
         medicationId,
         status: { in: ["queued", "running"] },
         OR: [
@@ -117,7 +119,7 @@ export async function admitIntakeImportJob({
 
     const activeJob = await tx.medicationIntakeImportJob.findFirst({
       where: {
-        userId,
+        recordUserId,
         medicationId,
         status: { in: ["queued", "running"] },
       },
@@ -127,7 +129,8 @@ export async function admitIntakeImportJob({
 
     const importJob = await tx.medicationIntakeImportJob.create({
       data: {
-        userId,
+        recordUserId,
+        actorUserId,
         medicationId,
         status: "queued",
         payload: toJson(payload),
@@ -169,7 +172,8 @@ export async function admitIntakeImportJob({
       },
     });
     await auditLog("medication.intake.import.kickoff.denied", {
-      userId,
+      userId: recordUserId,
+      actorUserId,
       ipAddress,
       details: {
         jobId,

src/lib/medications/intake-import-job-status.ts

@@ -109,18 +109,18 @@ export function projectMedicationImportResult(
  * Read one job the caller owns, or `null`.
  *
  * `medicationId` narrows to a job scoped to that medication; omit it for the
- * account-wide import. `userId` is always in the `where`, so a job id belonging
+ * account-wide import. `recordUserId` is always in the `where`, so a job id belonging
  * to somebody else is a 404 and not a leak.
  */
 export async function readMedicationIntakeImportJob(
-  userId: string,
+  recordUserId: string,
   jobId: string,
   medicationId?: string,
 ): Promise<MedicationIntakeImportJobStatus | null> {
   const row = await prisma.medicationIntakeImportJob.findFirst({
     where: {
       id: jobId,
-      userId,
+      recordUserId,
       // An account-wide job carries no medication, and asking for one by id must
       // not resolve to the other kind. `null` is the explicit narrowing.
       medicationId: medicationId ?? null,

tests/integration/medication-dose-history-import.test.ts

@@ -98,7 +98,8 @@ async function enqueue(userId: string) {
   );
   const job = await prisma.medicationIntakeImportJob.create({
     data: {
-      userId,
+      recordUserId: userId,
+      actorUserId: userId,
       // Account-wide: the file covers sixteen medications, so no single one owns
       // the job and each entry names its own.
       medicationId: null,

tests/integration/medication-intake-import-job.test.ts

@@ -92,7 +92,8 @@ describe("medication intake import job — concurrent retry", () => {
     };
     const job = await prisma.medicationIntakeImportJob.create({
       data: {
-        userId: user.id,
+        recordUserId: user.id,
+        actorUserId: user.id,
         medicationId: medication.id,
         payload: toJson(payload),
         progress: toJson(progress),
@@ -218,7 +219,8 @@ describe("medication intake import job — concurrent retry", () => {
     };
     const job = await prisma.medicationIntakeImportJob.create({
       data: {
-        userId: user.id,
+        recordUserId: user.id,
+        actorUserId: user.id,
         medicationId: medication.id,
         payload: toJson(payload),
         progress: toJson(progress),

tests/integration/medication-intake-import-reported-export.test.ts

@@ -93,7 +93,8 @@ describe("medication intake import — the reported 28-dose export", () => {
 
     const job = await prisma.medicationIntakeImportJob.create({
       data: {
-        userId: user.id,
+        recordUserId: user.id,
+        actorUserId: user.id,
         medicationId: medication.id,
         payload: toJson(payload),
         progress: toJson({
@@ -140,7 +141,8 @@ describe("medication intake import — the reported 28-dose export", () => {
 
     const first = await prisma.medicationIntakeImportJob.create({
       data: {
-        userId: user.id,
+        recordUserId: user.id,
+        actorUserId: user.id,
         medicationId: medication.id,
         payload: toJson(payload),
         progress: toJson(progress),
@@ -150,7 +152,8 @@ describe("medication intake import — the reported 28-dose export", () => {
 
     const second = await prisma.medicationIntakeImportJob.create({
       data: {
-        userId: user.id,
+        recordUserId: user.id,
+        actorUserId: user.id,
         medicationId: medication.id,
         payload: toJson(payload),
         progress: toJson(progress),
@@ -211,7 +214,8 @@ describe("medication intake import — the reported 28-dose export", () => {
 
     const job = await prisma.medicationIntakeImportJob.create({
       data: {
-        userId: user.id,
+        recordUserId: user.id,
+        actorUserId: user.id,
         medicationId: medication.id,
         payload: toJson({ entries }),
         progress: toJson({

tests/integration/sharing-import-attribution.test.ts

@@ -1,5 +1,9 @@
-import { beforeEach, describe, expect, it } from "vitest";
+import { NextRequest } from "next/server";
+import { beforeEach, describe, expect, it, vi } from "vitest";
 
+import { GET as getImportStatus } from "@/app/api/medications/[id]/intake/import/[jobId]/status/route";
+import { ACCOUNT_SELECTOR_HEADER } from "@/lib/api-handler";
+import { hashToken } from "@/lib/auth/hmac";
 import { toJson } from "@/lib/db";
 import {
   _setMedicationIntakeImportPrismaForTests,
@@ -8,6 +12,22 @@ import {
 } from "@/lib/jobs/medication-intake-import";
 
 import { getPrismaClient, truncateAllTables } from "./setup";
+import { cookieJar, headerJar } from "./mock-next-headers";
+
+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;
+      },
+    })),
+  };
+});
 
 const EMPTY_PROGRESS: MedicationImportProgress = {
   processed: 0,
@@ -22,6 +42,8 @@ beforeEach(async () => {
   const prisma = getPrismaClient();
   _setMedicationIntakeImportPrismaForTests(prisma);
   await truncateAllTables(prisma);
+  cookieJar.clear();
+  headerJar.clear();
 });
 
 describe("shared intake import attribution", () => {
@@ -87,4 +109,72 @@ describe("shared intake import attribution", () => {
       total: 0,
     });
   });
+
+  it("lets the initiating manager poll the record-scoped job", async () => {
+    const prisma = getPrismaClient();
+    await prisma.user.createMany({
+      data: [
+        {
+          id: "status-record",
+          username: "status-record",
+          email: "status-record@example.test",
+          role: "USER",
+        },
+        {
+          id: "status-manager",
+          username: "status-manager",
+          email: "status-manager@example.test",
+          role: "USER",
+        },
+      ],
+    });
+    const medication = await prisma.medication.create({
+      data: {
+        userId: "status-record",
+        name: "Status medication",
+        dose: "1 unit",
+      },
+    });
+    await prisma.accountGrant.create({
+      data: {
+        id: "status-manage-grant",
+        grantorId: "status-record",
+        granteeId: "status-manager",
+        access: "MANAGE",
+        acceptedAt: new Date(),
+      },
+    });
+    const token = "shared-import-status-token".padEnd(28, "0");
+    await prisma.apiToken.create({
+      data: {
+        userId: "status-manager",
+        name: "shared import status",
+        tokenHash: hashToken(token),
+        permissions: ["*"],
+      },
+    });
+    headerJar.set("authorization", `Bearer ${token}`);
+    headerJar.set(ACCOUNT_SELECTOR_HEADER, "status-record");
+    const job = await prisma.medicationIntakeImportJob.create({
+      data: {
+        recordUserId: "status-record",
+        actorUserId: "status-manager",
+        medicationId: medication.id,
+        payload: toJson({ entries: [] }),
+        progress: toJson(EMPTY_PROGRESS),
+      },
+    });
+
+    const response = await getImportStatus(
+      new NextRequest(
+        `http://localhost/api/medications/${medication.id}/intake/import/${job.id}/status`,
+        {
+        },
+      ),
+      { params: Promise.resolve({ id: medication.id, jobId: job.id }) },
+    );
+
+    expect(response.status).toBe(200);
+    expect((await response.json()).data.jobId).toBe(job.id);
+  });
 });

tests/integration/sharing-manage-handler-matrix.test.ts

@@ -1603,7 +1603,7 @@ proveManageRoute("POST /api/medications/[id]/intake/import", {
   auditAction: "medication.intake.import.kickoff",
   applied: async ({ ownerId }) =>
     (await getPrismaClient().medicationIntakeImportJob.count({
-      where: { userId: ownerId },
+      where: { recordUserId: ownerId },
     })) === 1,
 });
 
@@ -1637,7 +1637,7 @@ manageContract("POST /api/medications/intake/dose-history-import", {
   auditAction: "medication.intake.history.import.kickoff",
   applied: async ({ ownerId }) =>
     (await getPrismaClient().medicationIntakeImportJob.count({
-      where: { userId: ownerId },
+      where: { recordUserId: ownerId },
     })) === 1,
 });