feat(import): read Health Connect exports

Upload the ZIP the Android Health Connect app writes and import its
`health_connect_export.db` in the background, the same way an Apple
Health export is imported: streamed to disk with a SHA-256, capped at
1 GiB, one import per account, a job row of kind `health_connect` that
the shared reconcile cron and staging sweep now look up on its own
queue.

The worker extracts the database (4 GiB ceiling, zip-bomb checks) and
opens it read-only with `immutable=1`, `trusted_schema=OFF` and
`query_only`. A view in place of a record table refuses the file,
columns are found by name, and schema versions below 9 are refused.
Records are read in batches of 500 and aggregated in SQLite: steps,
energy and distance as one total per day from the app the person's
Health Connect priority list ranks first, heart rate per minute for 90
days and as hourly means before that, sleep stages with the unknown
stage counted only in sessions without real stages, workouts, cycle
days, water and micronutrients. Apps whose integration the account has
connected are left out, and a reading already stored under another
source is skipped. Rows carry source HEALTH_CONNECT, `hc:<uuid>` or the
`stats:` ids, so a second import of the same export writes nothing.

Both staged files are removed on every way out; the job result holds
counts per type and per app only. Settings gain a Health Connect card
with the export steps, progress and the outcome, in all seven
languages, and the import is documented in
docs/integrations/health-connect.md.

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