Merge Health Connect import into release/v1.42.0

# Conflicts:
#	src/lib/jobs/apple-health-import-worker.ts

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

@@ -22,6 +22,7 @@ import { checkRateLimit } from "@/lib/rate-limit";
 import { getGlobalBoss } from "@/lib/jobs/boss-instance";
 import {
   APPLE_HEALTH_IMPORT_V2_QUEUE,
+  APPLE_HEALTH_IMPORT_KIND,
   APPLE_HEALTH_IMPORT_PARSER_REVISION,
   APPLE_HEALTH_IMPORT_SEND_OPTIONS,
   type AppleHealthImportPayload,
@@ -31,6 +32,7 @@ import {
   createImportJobUnlessBusy,
   discardStagedUpload,
   IMPORT_BUSY_CODE,
+  IMPORT_BUSY_MESSAGE,
 } from "@/lib/import/apple-health-staging";
 import { unlink } from "node:fs/promises";
 
@@ -128,6 +130,7 @@ export const POST = apiHandler(async (request: NextRequest) => {
   const existing = await prisma.importJob.findFirst({
     where: {
       userId: targetUser.id,
+      kind: APPLE_HEALTH_IMPORT_KIND,
       uploadSha256: uploaded.sha256,
       parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
       status: { not: "failed" },
@@ -170,11 +173,10 @@ export const POST = apiHandler(async (request: NextRequest) => {
   });
   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 },
-    );
+    return apiError(IMPORT_BUSY_MESSAGE, 409, {
+      errorCode: IMPORT_BUSY_CODE,
+      jobId: staged.busy.id,
+    });
   }
   const importJob = staged.created;
 

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

@@ -30,6 +30,7 @@ import { checkRateLimit } from "@/lib/rate-limit";
 import { getGlobalBoss } from "@/lib/jobs/boss-instance";
 import {
   APPLE_HEALTH_IMPORT_V2_QUEUE,
+  APPLE_HEALTH_IMPORT_KIND,
   APPLE_HEALTH_IMPORT_PARSER_REVISION,
   APPLE_HEALTH_IMPORT_SEND_OPTIONS,
   type AppleHealthImportPayload,
@@ -39,6 +40,7 @@ import {
   createImportJobUnlessBusy,
   discardStagedUpload,
   IMPORT_BUSY_CODE,
+  IMPORT_BUSY_MESSAGE,
 } from "@/lib/import/apple-health-staging";
 import { unlink } from "node:fs/promises";
 
@@ -123,6 +125,7 @@ export const POST = apiHandler(async (request: NextRequest) => {
   const existing = await prisma.importJob.findFirst({
     where: {
       userId: user.id,
+      kind: APPLE_HEALTH_IMPORT_KIND,
       uploadSha256: uploaded.sha256,
       parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
       status: { not: "failed" },
@@ -167,11 +170,10 @@ export const POST = apiHandler(async (request: NextRequest) => {
   });
   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 },
-    );
+    return apiError(IMPORT_BUSY_MESSAGE, 409, {
+      errorCode: IMPORT_BUSY_CODE,
+      jobId: staged.busy.id,
+    });
   }
   const importJob = staged.created;
 

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

@@ -0,0 +1,252 @@
+/**
+ * The Health Connect upload and status routes: the size cap, the staging
+ * name the sweep knows, the job's `kind`, the content-hash dedupe, the one
+ * import per account rule and the status of the account's latest job.
+ */
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import { NextRequest } from "next/server";
+
+vi.mock("@/lib/auth/session", () => ({ getSession: vi.fn() }));
+vi.mock("@/lib/auth/audit", () => ({
+  auditLog: vi.fn().mockResolvedValue(undefined),
+}));
+vi.mock("@/lib/logging/transports", () => ({ emitIfSampled: vi.fn() }));
+vi.mock("@/lib/db-compat", () => ({
+  ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
+}));
+vi.mock("next/headers", () => ({
+  headers: vi.fn(async () => ({ get: () => null })),
+  cookies: vi.fn(async () => ({
+    get: () => undefined,
+    set: () => {},
+    delete: () => {},
+  })),
+}));
+
+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(),
+  importJobCreate: vi.fn(),
+  importJobFindFirst: vi.fn().mockResolvedValue(null),
+  importJobUpdate: vi.fn(),
+  streamToDisk: vi.fn(),
+  unlink: vi.fn().mockResolvedValue(undefined),
+  rm: vi.fn().mockResolvedValue(undefined),
+}));
+
+vi.mock("@/lib/rate-limit", () => ({
+  checkRateLimit: mocks.checkRateLimit,
+}));
+
+vi.mock("@/lib/jobs/boss-instance", () => ({
+  getGlobalBoss: mocks.getGlobalBoss,
+}));
+
+vi.mock("@/lib/multipart/stream-to-disk", () => ({
+  streamMultipartToDisk: mocks.streamToDisk,
+}));
+
+vi.mock("node:fs/promises", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("node:fs/promises")>();
+  return { ...actual, unlink: mocks.unlink, rm: mocks.rm };
+});
+
+vi.mock("@/lib/db", () => ({
+  prisma: {
+    importJob: {
+      create: mocks.importJobCreate,
+      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,
+        },
+      }),
+  },
+}));
+
+import { POST } from "../route";
+import { GET as STATUS } from "../status/route";
+import { getSession } from "@/lib/auth/session";
+
+const SESSION_OK = {
+  session: { id: "sess-1", expiresAt: new Date(Date.now() + 3_600_000) },
+  user: { id: "user-1", username: "testuser", role: "USER" as const },
+};
+
+const STAGED_PATH =
+  "/tmp/healthlog-health-connect-import-0b7f3c2a-1d4e-4f5a-9b8c-7d6e5f4a3b2c.bin";
+
+beforeEach(() => {
+  vi.resetAllMocks();
+  vi.mocked(getSession).mockResolvedValue(SESSION_OK as never);
+  mocks.checkRateLimit.mockResolvedValue({ allowed: true });
+  mocks.bossSend.mockResolvedValue("boss-job-1");
+  mocks.getGlobalBoss.mockReturnValue({ send: mocks.bossSend });
+  mocks.importJobCreate.mockResolvedValue({ id: "ij-1" });
+  mocks.importJobFindFirst.mockResolvedValue(null);
+  mocks.runningImportFindFirst.mockResolvedValue(null);
+  mocks.importJobUpdate.mockResolvedValue({ id: "ij-1" });
+  mocks.streamToDisk.mockResolvedValue({
+    filePath: STAGED_PATH,
+    bytes: 100,
+    sha256: "deadbeef",
+    originalFilename: "Health Connect.zip",
+    textFields: {},
+  });
+  mocks.unlink.mockResolvedValue(undefined);
+  mocks.rm.mockResolvedValue(undefined);
+});
+
+function upload(headers: Record<string, string> = {}): NextRequest {
+  return new NextRequest("http://localhost/api/import/health-connect-export", {
+    method: "POST",
+    headers: {
+      "content-type": "multipart/form-data; boundary=---x",
+      ...headers,
+    },
+    body: "stub",
+  });
+}
+
+describe("POST /api/import/health-connect-export", () => {
+  it("returns 401 when unauthenticated", async () => {
+    vi.mocked(getSession).mockResolvedValue(null);
+    expect((await POST(upload())).status).toBe(401);
+  });
+
+  it("refuses a declared body above 1 GiB before reading it", async () => {
+    const res = await POST(
+      upload({ "content-length": String(1024 * 1024 * 1024 + 1) }),
+    );
+    expect(res.status).toBe(413);
+    expect(mocks.streamToDisk).not.toHaveBeenCalled();
+  });
+
+  it("caps the streamed body at 1 GiB and stages it under its own name", async () => {
+    await POST(upload());
+    expect(mocks.streamToDisk).toHaveBeenCalledWith(
+      expect.anything(),
+      expect.any(String),
+      expect.objectContaining({
+        maxBytes: 1024 * 1024 * 1024,
+        tmpPrefix: "healthlog-health-connect-import",
+      }),
+    );
+  });
+
+  it("creates a health_connect job and queues it on its own queue", async () => {
+    const res = await POST(upload());
+    expect(res.status).toBe(202);
+    expect((await res.json()).data).toEqual({
+      jobId: "ij-1",
+      status: "queued",
+    });
+    expect(mocks.importJobCreate).toHaveBeenCalledWith(
+      expect.objectContaining({
+        data: expect.objectContaining({
+          userId: "user-1",
+          kind: "health_connect",
+          uploadSha256: "deadbeef",
+        }),
+      }),
+    );
+    expect(mocks.bossSend).toHaveBeenCalledWith(
+      "health-connect-import",
+      { userId: "user-1", importJobId: "ij-1", uploadPath: STAGED_PATH },
+      { retryLimit: 0, expireInSeconds: 21600 },
+    );
+  });
+
+  it("resolves the same bytes to the earlier job and drops the new copy", async () => {
+    mocks.importJobFindFirst.mockResolvedValue({ id: "ij-0", status: "done" });
+    const res = await POST(upload());
+    expect(res.status).toBe(202);
+    expect((await res.json()).data).toEqual({
+      jobId: "ij-0",
+      status: "done",
+      idempotent: true,
+    });
+    expect(mocks.importJobFindFirst).toHaveBeenCalledWith(
+      expect.objectContaining({
+        where: expect.objectContaining({
+          kind: "health_connect",
+          status: { not: "failed" },
+        }),
+      }),
+    );
+    expect(mocks.unlink).toHaveBeenCalledWith(STAGED_PATH);
+    expect(mocks.bossSend).not.toHaveBeenCalled();
+  });
+
+  it("refuses a second import while one runs and discards the upload", async () => {
+    mocks.runningImportFindFirst.mockResolvedValue({
+      id: "ij-running",
+      status: "parsing",
+    } as never);
+    const res = await POST(upload());
+    expect(res.status).toBe(409);
+    const body = await res.json();
+    expect(body.meta).toMatchObject({
+      errorCode: "import.apple_health.busy",
+      jobId: "ij-running",
+    });
+    expect(mocks.rm).toHaveBeenCalledWith(STAGED_PATH, { force: true });
+    expect(mocks.bossSend).not.toHaveBeenCalled();
+  });
+
+  it("answers 503 and removes the upload when no worker runs", async () => {
+    mocks.getGlobalBoss.mockReturnValue(null);
+    const res = await POST(upload());
+    expect(res.status).toBe(503);
+    expect(mocks.rm).toHaveBeenCalledWith(STAGED_PATH, { force: true });
+  });
+});
+
+describe("GET /api/import/health-connect-export/status", () => {
+  it("answers null when the account never imported", async () => {
+    const res = await STATUS(
+      new NextRequest(
+        "http://localhost/api/import/health-connect-export/status",
+      ),
+    );
+    expect(res.status).toBe(200);
+    expect((await res.json()).data).toEqual({ job: null });
+  });
+
+  it("answers the account's latest Health Connect job", async () => {
+    mocks.importJobFindFirst.mockResolvedValue({
+      id: "ij-9",
+      status: "done",
+      startedAt: new Date("2026-10-01T10:00:00Z"),
+      completedAt: new Date("2026-10-01T10:01:00Z"),
+      uploadBytes: 1234,
+      progress: { rowsUpserted: 5 },
+      result: { kind: "health_connect", totals: { rowsUpserted: 5 } },
+      failureReason: null,
+    });
+    const res = await STATUS(
+      new NextRequest(
+        "http://localhost/api/import/health-connect-export/status",
+      ),
+    );
+    expect((await res.json()).data.job).toMatchObject({
+      jobId: "ij-9",
+      status: "done",
+      completedAt: "2026-10-01T10:01:00.000Z",
+    });
+    expect(mocks.importJobFindFirst).toHaveBeenCalledWith(
+      expect.objectContaining({
+        where: { userId: "user-1", kind: "health_connect" },
+      }),
+    );
+  });
+});

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

@@ -1,15 +1,191 @@
 /**
  * `POST /api/import/health-connect-export` — upload an Android Health Connect
- * export ZIP for background import (v1.42, #972). Outside the proxy matcher
- * like the Apple Health upload (`src/lib/http/proxy-bypass-routes.ts`), so the
- * body is streamed to disk after authentication instead of being buffered.
+ * export ZIP for background import (v1.42, #972).
  *
- * Contract stub: authenticates as the finished route will, then answers 501.
+ * Mirrors the Apple Health upload (`../apple-health-export/route.ts`): the
+ * route sits outside the proxy matcher (`src/lib/http/proxy-bypass-routes.ts`)
+ * so the body is streamed to disk after authentication instead of buffered,
+ * hashed with SHA-256 on the way, refused past the size cap, and handed to
+ * the `health-connect-import` queue. One import runs per account at a time,
+ * whatever its kind (`createImportJobUnlessBusy`). Re-uploading the same
+ * bytes resolves to the still-viable job instead of importing twice; a failed
+ * job never blocks a retry of the same file.
+ *
+ * The audit row carries the upload's size and hash and nothing from inside
+ * the file.
  */
-import { apiHandler, requireAuth } from "@/lib/api-handler";
-import { notImplemented } from "@/lib/http/not-implemented";
+import type { NextRequest } from "next/server";
+import { unlink } from "node:fs/promises";
 
-export const POST = apiHandler(async () => {
-  await requireAuth();
-  return notImplemented();
+import { prisma } from "@/lib/db";
+import { apiHandler, requireAuth, HttpError } from "@/lib/api-handler";
+import { apiError, apiSuccess, getClientIp } from "@/lib/api-response";
+import { annotate } from "@/lib/logging/context";
+import { auditLog } from "@/lib/auth/audit";
+import { checkRateLimit } from "@/lib/rate-limit";
+import { getGlobalBoss } from "@/lib/jobs/boss-instance";
+import {
+  HEALTH_CONNECT_IMPORT_KIND,
+  HEALTH_CONNECT_IMPORT_PARSER_REVISION,
+  HEALTH_CONNECT_IMPORT_QUEUE,
+  HEALTH_CONNECT_IMPORT_SEND_OPTIONS,
+  type HealthConnectImportPayload,
+} from "@/lib/jobs/health-connect-import-worker";
+import { streamMultipartToDisk } from "@/lib/multipart/stream-to-disk";
+import {
+  createImportJobUnlessBusy,
+  discardStagedUpload,
+  IMPORT_BUSY_CODE,
+  IMPORT_BUSY_MESSAGE,
+} from "@/lib/import/apple-health-staging";
+
+export const dynamic = "force-dynamic";
+
+/**
+ * Hard cap on the upload. The ZIP holds one SQLite file that compresses to
+ * about half; a multi-year export with minute-by-minute heart rate sits in
+ * the low hundreds of megabytes, and 1 GiB leaves room for the long tail.
+ */
+const HEALTH_CONNECT_MAX_UPLOAD_BYTES = 1024 * 1024 * 1024;
+
+/** Three uploads per minute per account, like the Apple Health upload. */
+const RATE_LIMIT_MAX = 3;
+const RATE_LIMIT_WINDOW_MS = 60_000;
+
+export const POST = apiHandler(async (request: NextRequest) => {
+  const { user } = await requireAuth();
+  annotate({ action: { name: "import.health-connect.kickoff" } });
+
+  const rl = await checkRateLimit(
+    `import:health-connect:${user.id}`,
+    RATE_LIMIT_MAX,
+    RATE_LIMIT_WINDOW_MS,
+  );
+  if (!rl.allowed) {
+    throw new HttpError(429, "Too many import uploads, try again later");
+  }
+
+  const declaredBytes = Number(request.headers.get("content-length") ?? 0);
+  if (declaredBytes > HEALTH_CONNECT_MAX_UPLOAD_BYTES) {
+    await auditLog("import.health-connect.kickoff.denied", {
+      userId: user.id,
+      ipAddress: getClientIp(request),
+      details: {
+        reason: "content_length_exceeded",
+        contentLength: declaredBytes,
+      },
+    });
+    return apiError("Upload exceeds the 1 GB limit", 413);
+  }
+
+  const body = request.body;
+  if (!body) return apiError("Request body is required", 400);
+
+  let uploaded;
+  try {
+    uploaded = await streamMultipartToDisk(
+      body,
+      request.headers.get("content-type"),
+      {
+        maxBytes: HEALTH_CONNECT_MAX_UPLOAD_BYTES,
+        fieldName: "file",
+        tmpPrefix: "healthlog-health-connect-import",
+      },
+    );
+  } catch (err) {
+    const message = err instanceof Error ? err.message : "upload_failed";
+    await auditLog("import.health-connect.kickoff.denied", {
+      userId: user.id,
+      ipAddress: getClientIp(request),
+      details: { reason: "stream_to_disk_failed" },
+    });
+    return apiError(`Multipart upload failed: ${message}`, 422);
+  }
+
+  const existing = await prisma.importJob.findFirst({
+    where: {
+      userId: user.id,
+      kind: HEALTH_CONNECT_IMPORT_KIND,
+      uploadSha256: uploaded.sha256,
+      parserRevision: HEALTH_CONNECT_IMPORT_PARSER_REVISION,
+      status: { not: "failed" },
+    },
+    orderBy: { startedAt: "desc" },
+  });
+  if (existing) {
+    await unlink(uploaded.filePath).catch(() => {});
+    annotate({ meta: { idempotent_hit: true, job_id: existing.id } });
+    return apiSuccess(
+      { jobId: existing.id, status: existing.status, idempotent: true },
+      202,
+    );
+  }
+
+  const boss = getGlobalBoss();
+  if (!boss) {
+    await auditLog("import.health-connect.kickoff.denied", {
+      userId: user.id,
+      ipAddress: getClientIp(request),
+      details: { reason: "worker_not_running" },
+    });
+    await discardStagedUpload(uploaded.filePath);
+    throw new HttpError(503, "Background worker is not running");
+  }
+
+  const staged = await createImportJobUnlessBusy(prisma, {
+    userId: user.id,
+    kind: HEALTH_CONNECT_IMPORT_KIND,
+    status: "queued",
+    uploadBytes: uploaded.bytes,
+    uploadSha256: uploaded.sha256,
+    parserRevision: HEALTH_CONNECT_IMPORT_PARSER_REVISION,
+  });
+  if ("busy" in staged) {
+    await discardStagedUpload(uploaded.filePath);
+    return apiError(IMPORT_BUSY_MESSAGE, 409, {
+      errorCode: IMPORT_BUSY_CODE,
+      jobId: staged.busy.id,
+    });
+  }
+  const importJob = staged.created;
+
+  const payload: HealthConnectImportPayload = {
+    userId: user.id,
+    importJobId: importJob.id,
+    uploadPath: uploaded.filePath,
+  };
+  const bossJobId = await boss.send(
+    HEALTH_CONNECT_IMPORT_QUEUE,
+    payload,
+    HEALTH_CONNECT_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 },
+    data: { pgBossJobId: bossJobId },
+  });
+
+  await auditLog("import.health-connect.kickoff", {
+    userId: user.id,
+    ipAddress: getClientIp(request),
+    details: {
+      jobId: importJob.id,
+      uploadBytes: uploaded.bytes,
+      uploadSha256: uploaded.sha256,
+    },
+  });
+
+  return apiSuccess({ jobId: importJob.id, status: "queued" as const }, 202);
 });

src/app/api/import/health-connect-export/status/route.ts

@@ -1,13 +1,53 @@
 /**
  * `GET /api/import/health-connect-export/status` — the caller's most recent
- * Health Connect import job (v1.42, #972).
+ * Health Connect import job (v1.42, #972), or `null` when there never was
+ * one. The settings card polls it while a job runs and shows the last
+ * outcome after a reload.
  *
- * Contract stub: authenticates as the finished route will, then answers 501.
+ * The result is counts only (per type, per app); the import never stores a
+ * value or a timestamp from the file on the job row.
  */
-import { apiHandler, requireAuth } from "@/lib/api-handler";
-import { notImplemented } from "@/lib/http/not-implemented";
+import type { NextRequest } from "next/server";
 
-export const GET = apiHandler(async () => {
-  await requireAuth();
-  return notImplemented();
+import { prisma } from "@/lib/db";
+import { apiHandler, requireAuth } from "@/lib/api-handler";
+import { apiSuccess } from "@/lib/api-response";
+import { annotate } from "@/lib/logging/context";
+import { HEALTH_CONNECT_IMPORT_KIND } from "@/lib/jobs/health-connect-import-worker";
+
+export const dynamic = "force-dynamic";
+
+export const GET = apiHandler(async (_request: NextRequest) => {
+  const { user } = await requireAuth();
+  annotate({ action: { name: "import.health-connect.status" } });
+
+  const row = await prisma.importJob.findFirst({
+    where: { userId: user.id, kind: HEALTH_CONNECT_IMPORT_KIND },
+    orderBy: { startedAt: "desc" },
+    select: {
+      id: true,
+      status: true,
+      startedAt: true,
+      completedAt: true,
+      uploadBytes: true,
+      progress: true,
+      result: true,
+      failureReason: true,
+    },
+  });
+
+  return apiSuccess({
+    job: row
+      ? {
+          jobId: row.id,
+          status: row.status,
+          startedAt: row.startedAt.toISOString(),
+          completedAt: row.completedAt?.toISOString() ?? null,
+          uploadBytes: row.uploadBytes,
+          progress: (row.progress as Record<string, unknown>) ?? {},
+          result: (row.result as Record<string, unknown> | null) ?? null,
+          failureReason: row.failureReason,
+        }
+      : null,
+  });
 });