fix(import): bound what an export may inflate to, allow one import per account, and never leave the upload behind

The extractor held the inflated export.xml to a flat 8 GiB whatever the
member occupied, and the header ratio check reads sizes the archive
writes itself: a member declaring an honest ratio could inflate at
1000:1, so an 8 MB upload could fill the temp directory. The streamed
byte count is now held to 200 times the member's compressed size on
disk (never less than 64 MiB, never more than 8 GiB), and a stored
member may not grow at all.

An account may have one Apple Health import queued or running. A
second upload with different bytes is refused with 409
import.apple_health.busy and the running job's id; the check and the
new row are one transaction under an advisory lock.

The staged upload is the person's export in plain text. It is now
removed on every exit that leaves no worker to read it: a refused or
broken multipart body, no worker running, a queue that takes no job
(the row is failed as well), the admin route's missing or unknown
target account, and an account deleted before the worker reached it.
The worker removes the files before recording a failure, so a row that
went with a deleted account no longer strands them. The 15-minute
import reconcile pass also sweeps staged files older than the longest
an import may run.

src/app/api/admin/import-apple-health-export/__tests__/route.test.ts

@@ -25,6 +25,7 @@ vi.mock("next/headers", () => ({
 }));
 
 const mocks = vi.hoisted(() => ({
+  runningImportFindFirst: vi.fn(async () => null),
   checkRateLimit: vi.fn().mockResolvedValue({ allowed: true }),
   bossSend: vi.fn().mockResolvedValue("boss-job-1"),
   getGlobalBoss: vi.fn(),
@@ -60,6 +61,16 @@ vi.mock("@/lib/db", () => ({
       findFirst: mocks.importJobFindFirst,
       update: mocks.importJobUpdate,
     },
+    // The one-running-import check and the new row, in one transaction.
+    // The check finds nothing running unless a test says otherwise.
+    $transaction: async (fn: (tx: unknown) => unknown) =>
+      fn({
+        $executeRaw: async () => 0,
+        importJob: {
+          findFirst: mocks.runningImportFindFirst,
+          create: mocks.importJobCreate,
+        },
+      }),
     user: {
       findUnique: mocks.userFindUnique,
     },

src/app/api/admin/import-apple-health-export/route.ts

@@ -27,6 +27,11 @@ import {
   type AppleHealthImportPayload,
 } from "@/lib/jobs/apple-health-import-worker";
 import { streamMultipartToDisk } from "@/lib/multipart/stream-to-disk";
+import {
+  createImportJobUnlessBusy,
+  discardStagedUpload,
+  IMPORT_BUSY_CODE,
+} from "@/lib/import/apple-health-staging";
 import { unlink } from "node:fs/promises";
 
 export const dynamic = "force-dynamic";
@@ -96,6 +101,7 @@ export const POST = apiHandler(async (request: NextRequest) => {
       ipAddress: getClientIp(request),
       details: { reason: "missing_user_id" },
     });
+    await discardStagedUpload(uploaded.filePath);
     return apiError("Multipart field 'userId' is required", 422);
   }
 
@@ -109,6 +115,7 @@ export const POST = apiHandler(async (request: NextRequest) => {
       ipAddress: getClientIp(request),
       details: { reason: "target_user_missing", targetUserId },
     });
+    await discardStagedUpload(uploaded.filePath);
     return apiError(`Target user '${targetUserId}' does not exist`, 422);
   }
 
@@ -149,19 +156,27 @@ export const POST = apiHandler(async (request: NextRequest) => {
       ipAddress: getClientIp(request),
       details: { reason: "worker_not_running" },
     });
+    await discardStagedUpload(uploaded.filePath);
     throw new HttpError(503, "Background worker is not running");
   }
 
-  const importJob = await prisma.importJob.create({
-    data: {
-      userId: targetUser.id,
-      triggeredByAdminId: admin.id,
-      status: "queued",
-      uploadBytes: uploaded.bytes,
-      uploadSha256: uploaded.sha256,
-      parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
-    },
+  const staged = await createImportJobUnlessBusy(prisma, {
+    userId: targetUser.id,
+    triggeredByAdminId: admin.id,
+    status: "queued",
+    uploadBytes: uploaded.bytes,
+    uploadSha256: uploaded.sha256,
+    parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
   });
+  if ("busy" in staged) {
+    await discardStagedUpload(uploaded.filePath);
+    return apiError(
+      "An Apple Health import is already running for this account. Wait for it to finish, then upload again.",
+      409,
+      { errorCode: IMPORT_BUSY_CODE, jobId: staged.busy.id },
+    );
+  }
+  const importJob = staged.created;
 
   const payload: AppleHealthImportPayload = {
     userId: targetUser.id,
@@ -178,6 +193,19 @@ export const POST = apiHandler(async (request: NextRequest) => {
     payload,
     APPLE_HEALTH_IMPORT_SEND_OPTIONS,
   );
+  if (!bossJobId) {
+    await prisma.importJob.update({
+      where: { id: importJob.id },
+      data: {
+        status: "failed",
+        failureReason:
+          "The import could not be queued. Upload the export again.",
+        completedAt: new Date(),
+      },
+    });
+    await discardStagedUpload(uploaded.filePath);
+    throw new HttpError(503, "The import could not be queued");
+  }
 
   await prisma.importJob.update({
     where: { id: importJob.id },

src/app/api/import/apple-health-export/__tests__/route.test.ts

@@ -27,6 +27,7 @@ vi.mock("next/headers", () => ({
 }));
 
 const mocks = vi.hoisted(() => ({
+  runningImportFindFirst: vi.fn(async () => null),
   checkRateLimit: vi.fn().mockResolvedValue({ allowed: true }),
   bossSend: vi.fn().mockResolvedValue("boss-job-1"),
   getGlobalBoss: vi.fn(),
@@ -35,6 +36,7 @@ const mocks = vi.hoisted(() => ({
   importJobUpdate: vi.fn(),
   streamToDisk: vi.fn(),
   unlink: vi.fn().mockResolvedValue(undefined),
+  rm: vi.fn().mockResolvedValue(undefined),
 }));
 
 vi.mock("@/lib/rate-limit", () => ({
@@ -51,7 +53,7 @@ vi.mock("@/lib/multipart/stream-to-disk", () => ({
 
 vi.mock("node:fs/promises", async (importOriginal) => {
   const actual = await importOriginal<typeof import("node:fs/promises")>();
-  return { ...actual, unlink: mocks.unlink };
+  return { ...actual, unlink: mocks.unlink, rm: mocks.rm };
 });
 
 vi.mock("@/lib/db", () => ({
@@ -61,6 +63,16 @@ vi.mock("@/lib/db", () => ({
       findFirst: mocks.importJobFindFirst,
       update: mocks.importJobUpdate,
     },
+    // The one-running-import check and the new row, in one transaction.
+    // The check finds nothing running unless a test says otherwise.
+    $transaction: async (fn: (tx: unknown) => unknown) =>
+      fn({
+        $executeRaw: async () => 0,
+        importJob: {
+          findFirst: mocks.runningImportFindFirst,
+          create: mocks.importJobCreate,
+        },
+      }),
   },
 }));
 
@@ -90,6 +102,7 @@ beforeEach(() => {
     textFields: {},
   });
   mocks.unlink.mockResolvedValue(undefined);
+  mocks.rm.mockResolvedValue(undefined);
 });
 
 function multipartReq(): NextRequest {
@@ -220,3 +233,51 @@ describe("POST /api/import/apple-health-export — dedup by content hash (issue
     );
   });
 });
+
+// The staged upload is the person's export in plain text. Every exit that
+// leaves no worker to consume it removes it, and a second import for the same
+// account is refused while one is running.
+describe("POST /api/import/apple-health-export — the staged upload on every exit", () => {
+  const removed = () =>
+    mocks.rm.mock.calls.some(([path]) => path === STAGED_PATH);
+
+  it("refuses a second import while one is running, and removes the upload", async () => {
+    vi.mocked(getSession).mockResolvedValue(SESSION_OK as never);
+    mocks.runningImportFindFirst.mockResolvedValue({
+      id: "ij-running",
+      status: "parsing",
+    } as never);
+    const res = await POST(multipartReq());
+    expect(res.status).toBe(409);
+    const body = await res.json();
+    expect(body.meta).toMatchObject({
+      errorCode: "import.apple_health.busy",
+      jobId: "ij-running",
+    });
+    expect(mocks.importJobCreate).not.toHaveBeenCalled();
+    expect(mocks.bossSend).not.toHaveBeenCalled();
+    expect(removed()).toBe(true);
+  });
+
+  it("removes the upload when no worker is running", async () => {
+    vi.mocked(getSession).mockResolvedValue(SESSION_OK as never);
+    mocks.getGlobalBoss.mockReturnValue(null);
+    const res = await POST(multipartReq());
+    expect(res.status).toBe(503);
+    expect(removed()).toBe(true);
+  });
+
+  it("fails the row and removes the upload when the queue takes no job", async () => {
+    vi.mocked(getSession).mockResolvedValue(SESSION_OK as never);
+    mocks.bossSend.mockResolvedValue(null);
+    const res = await POST(multipartReq());
+    expect(res.status).toBe(503);
+    expect(mocks.importJobUpdate).toHaveBeenCalledWith(
+      expect.objectContaining({
+        where: { id: "ij-1" },
+        data: expect.objectContaining({ status: "failed" }),
+      }),
+    );
+    expect(removed()).toBe(true);
+  });
+});

src/app/api/import/apple-health-export/route.ts

@@ -35,6 +35,11 @@ import {
   type AppleHealthImportPayload,
 } from "@/lib/jobs/apple-health-import-worker";
 import { streamMultipartToDisk } from "@/lib/multipart/stream-to-disk";
+import {
+  createImportJobUnlessBusy,
+  discardStagedUpload,
+  IMPORT_BUSY_CODE,
+} from "@/lib/import/apple-health-staging";
 import { unlink } from "node:fs/promises";
 
 export const dynamic = "force-dynamic";
@@ -147,18 +152,28 @@ export const POST = apiHandler(async (request: NextRequest) => {
       ipAddress: getClientIp(request),
       details: { reason: "worker_not_running" },
     });
+    // Nothing will ever read the staged export; it must not stay in /tmp.
+    await discardStagedUpload(uploaded.filePath);
     throw new HttpError(503, "Background worker is not running");
   }
 
-  const importJob = await prisma.importJob.create({
-    data: {
-      userId: user.id,
-      status: "queued",
-      uploadBytes: uploaded.bytes,
-      uploadSha256: uploaded.sha256,
-      parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
-    },
+  // One running import per account (see `apple-health-staging.ts`).
+  const staged = await createImportJobUnlessBusy(prisma, {
+    userId: user.id,
+    status: "queued",
+    uploadBytes: uploaded.bytes,
+    uploadSha256: uploaded.sha256,
+    parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
   });
+  if ("busy" in staged) {
+    await discardStagedUpload(uploaded.filePath);
+    return apiError(
+      "An Apple Health import is already running for this account. Wait for it to finish, then upload again.",
+      409,
+      { errorCode: IMPORT_BUSY_CODE, jobId: staged.busy.id },
+    );
+  }
+  const importJob = staged.created;
 
   const payload: AppleHealthImportPayload = {
     userId: user.id,
@@ -174,6 +189,21 @@ export const POST = apiHandler(async (request: NextRequest) => {
     payload,
     APPLE_HEALTH_IMPORT_SEND_OPTIONS,
   );
+  if (!bossJobId) {
+    // The queue did not take the job, so no worker will consume the upload
+    // or move the row on. Fail the row now and remove the export.
+    await prisma.importJob.update({
+      where: { id: importJob.id },
+      data: {
+        status: "failed",
+        failureReason:
+          "The import could not be queued. Upload the export again.",
+        completedAt: new Date(),
+      },
+    });
+    await discardStagedUpload(uploaded.filePath);
+    throw new HttpError(503, "The import could not be queued");
+  }
 
   await prisma.importJob.update({
     where: { id: importJob.id },