feat(import): Apple Health export.zip endpoints + multipart streamer

Three new routes wire the streaming parser to the wire:

* `POST /api/import/apple-health-export` — user-facing kick-off.
  Streams the multipart upload straight to `/tmp` (no in-memory
  buffer for the 1.5 GB hard cap), hashes the bytes inline for
  content-based idempotency, creates an `ImportJob` row in
  `queued`, and enqueues the `apple-health-import` pg-boss job.
  Returns `202 { jobId }` so the client can poll for progress.
* `GET /api/import/apple-health-export/[jobId]/status` — polling
  endpoint returning the canonical `ImportJobStatusResponse`
  envelope (R-1 §8). Authorises on owner-id OR triggering admin.
* `POST /api/admin/import-apple-health-export` — admin variant
  (cookie-only via `requireAdmin()`; Bearer never elevates) with
  a `userId` text field naming the target user. Same content-hash
  idempotency scoped per target user.

The kick-off path uses a hand-rolled multipart boundary walker
(`src/lib/multipart/stream-to-disk.ts`) so a multi-gigabyte
`export.zip` upload never lands in V8 heap — Next.js's
`request.formData()` buffers the full body. Captured bytes
stream to disk while a SHA-256 hash accumulates in the same pass.

Locks per `.planning/research/v1434-r-1-xml-import.md` §5.3, §5.4,
§5.5, §9.

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

@@ -0,0 +1,192 @@
+/**
+ * `POST /api/admin/import-apple-health-export` — admin-only Apple
+ * Health import on behalf of a target user.
+ *
+ * Same multipart streaming path as the user-facing kick-off, but with
+ * `requireAdmin()` (cookie-only — Bearer tokens never elevate) and a
+ * `userId` text field naming the target user. The kicked-off
+ * `ImportJob` row carries `triggeredByAdminId = admin.id` so the
+ * status endpoint admits both the target user AND the triggering
+ * admin.
+ *
+ * Locks per `.planning/research/v1434-r-1-xml-import.md` §9.
+ */
+import type { NextRequest } from "next/server";
+
+import { prisma } from "@/lib/db";
+import { apiHandler, requireAdmin, 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 {
+  APPLE_HEALTH_IMPORT_QUEUE,
+  type AppleHealthImportPayload,
+} from "@/lib/jobs/apple-health-import-worker";
+import { streamMultipartToDisk } from "@/lib/multipart/stream-to-disk";
+
+export const dynamic = "force-dynamic";
+
+const MAX_UPLOAD_BYTES = 1.5 * 1024 * 1024 * 1024;
+const RATE_LIMIT_MAX = 3;
+const RATE_LIMIT_WINDOW_MS = 60_000;
+
+export const POST = apiHandler(async (request: NextRequest) => {
+  const { user: admin } = await requireAdmin();
+  annotate({ action: { name: "admin.import-apple-health.kickoff" } });
+
+  const rl = await checkRateLimit(
+    `admin-import-apple-health:${admin.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 > MAX_UPLOAD_BYTES) {
+    await auditLog("admin.import-apple-health.kickoff.denied", {
+      userId: admin.id,
+      ipAddress: getClientIp(request),
+      details: {
+        reason: "content_length_exceeded",
+        contentLength: declaredBytes,
+      },
+    });
+    return apiError("Upload exceeds 1.5 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: MAX_UPLOAD_BYTES,
+        fieldName: "file",
+        tmpPrefix: "healthlog-admin-apple-health-import",
+      },
+    );
+  } catch (err) {
+    const message = err instanceof Error ? err.message : "upload_failed";
+    await auditLog("admin.import-apple-health.kickoff.denied", {
+      userId: admin.id,
+      ipAddress: getClientIp(request),
+      details: { reason: "stream_to_disk_failed", message },
+    });
+    return apiError(`Multipart upload failed: ${message}`, 422);
+  }
+
+  // Resolve the target user from the multipart text field. Mandatory:
+  // the admin variant is explicitly cross-user.
+  const targetUserId = uploaded.textFields.userId;
+  if (!targetUserId) {
+    await auditLog("admin.import-apple-health.kickoff.denied", {
+      userId: admin.id,
+      ipAddress: getClientIp(request),
+      details: { reason: "missing_user_id" },
+    });
+    return apiError("Multipart field 'userId' is required", 422);
+  }
+
+  const targetUser = await prisma.user.findUnique({
+    where: { id: targetUserId },
+    select: { id: true, username: true },
+  });
+  if (!targetUser) {
+    await auditLog("admin.import-apple-health.kickoff.denied", {
+      userId: admin.id,
+      ipAddress: getClientIp(request),
+      details: { reason: "target_user_missing", targetUserId },
+    });
+    return apiError(
+      `Target user '${targetUserId}' does not exist`,
+      422,
+    );
+  }
+
+  // Same content-hash idempotency as the user-facing route, but scoped
+  // by the target user — an admin re-uploading the same file for the
+  // same user resolves to the previous job.
+  const existing = await prisma.importJob.findFirst({
+    where: {
+      userId: targetUser.id,
+      uploadSha256: uploaded.sha256,
+    },
+    orderBy: { startedAt: "desc" },
+  });
+  if (existing) {
+    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("admin.import-apple-health.kickoff.denied", {
+      userId: admin.id,
+      ipAddress: getClientIp(request),
+      details: { reason: "worker_not_running" },
+    });
+    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,
+    },
+  });
+
+  const payload: AppleHealthImportPayload = {
+    userId: targetUser.id,
+    triggeredByAdminId: admin.id,
+    uploadPath: uploaded.filePath,
+    uploadBytes: uploaded.bytes,
+    enqueuedAt: new Date().toISOString(),
+  };
+  const bossJobId = await boss.send(APPLE_HEALTH_IMPORT_QUEUE, payload);
+
+  await prisma.importJob.update({
+    where: { id: importJob.id },
+    data: { pgBossJobId: bossJobId },
+  });
+
+  await auditLog("admin.import-apple-health.kickoff", {
+    userId: admin.id,
+    ipAddress: getClientIp(request),
+    details: {
+      jobId: importJob.id,
+      bossJobId: bossJobId ?? null,
+      targetUserId: targetUser.id,
+      targetUsername: targetUser.username,
+      uploadBytes: uploaded.bytes,
+      uploadSha256: uploaded.sha256,
+    },
+  });
+
+  return apiSuccess(
+    {
+      jobId: importJob.id,
+      status: "queued" as const,
+      targetUserId: targetUser.id,
+    },
+    202,
+  );
+});

src/app/api/import/apple-health-export/[jobId]/status/route.ts

@@ -0,0 +1,77 @@
+/**
+ * `GET /api/import/apple-health-export/[jobId]/status` — polling
+ * endpoint for an in-flight Apple Health import.
+ *
+ * Returns the canonical `ImportJobStatusResponse` envelope (§8 of
+ * `.planning/research/v1434-r-1-xml-import.md`). The polling cadence
+ * the iOS client uses:
+ *   - every 2 s while `queued | unpacking | parsing | upserting`
+ *   - every 30 s once `done | failed` (last-known state is fine)
+ *
+ * Authorisation: the row's `userId` must match the requester or the
+ * requester is the admin who kicked it off via the admin variant.
+ */
+import type { NextRequest } from "next/server";
+
+import { prisma } from "@/lib/db";
+import { apiHandler, requireAuth } from "@/lib/api-handler";
+import { apiError, apiSuccess } from "@/lib/api-response";
+import { annotate } from "@/lib/logging/context";
+
+export const dynamic = "force-dynamic";
+
+interface RouteContext {
+  params: Promise<{ jobId: string }>;
+}
+
+interface ImportJobStatusResponse {
+  jobId: string;
+  status: string;
+  startedAt: string;
+  completedAt: string | null;
+  uploadBytes: number;
+  exportedAt: string | null;
+  progress: Record<string, unknown>;
+  result: Record<string, unknown> | null;
+  failureReason: string | null;
+}
+
+export const GET = apiHandler(
+  async (_request: NextRequest, ctx: RouteContext) => {
+    const { user } = await requireAuth();
+    const { jobId } = await ctx.params;
+    annotate({
+      action: { name: "import.apple-health.status" },
+      meta: { job_id: jobId },
+    });
+
+    const row = await prisma.importJob.findUnique({ where: { id: jobId } });
+    if (!row) {
+      return apiError("Import job not found", 404);
+    }
+
+    // Authorisation: the owner can see their own job; the admin who
+    // triggered it can see it from the admin route. Any other caller
+    // gets a 404 (not 403 — we don't leak existence to outsiders).
+    const isOwner = row.userId === user.id;
+    const isTriggeringAdmin =
+      row.triggeredByAdminId !== null && row.triggeredByAdminId === user.id;
+    if (!isOwner && !isTriggeringAdmin) {
+      return apiError("Import job not found", 404);
+    }
+
+    const body: ImportJobStatusResponse = {
+      jobId: row.id,
+      status: row.status,
+      startedAt: row.startedAt.toISOString(),
+      completedAt: row.completedAt?.toISOString() ?? null,
+      uploadBytes: row.uploadBytes,
+      exportedAt: row.exportedAt?.toISOString() ?? null,
+      progress: (row.progress as Record<string, unknown>) ?? {},
+      result: (row.result as Record<string, unknown> | null) ?? null,
+      failureReason: row.failureReason,
+    };
+
+    return apiSuccess(body, 200);
+  },
+);

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

@@ -0,0 +1,176 @@
+/**
+ * `POST /api/import/apple-health-export` — user-facing kick-off for an
+ * Apple Health `export.zip` import.
+ *
+ * Streams the upload directly to disk (`/tmp/healthlog-upload-*.bin`)
+ * without buffering the full body in memory, creates an `ImportJob`
+ * row in `queued`, and enqueues an `apple-health-import` pg-boss job.
+ * Returns `{ jobId }` so the client can poll
+ * `GET /api/import/apple-health-export/[jobId]/status`.
+ *
+ * Idempotency: a re-upload of the exact same file (same SHA-256 of
+ * the bytes) short-circuits to the previous `ImportJob` id without
+ * re-queueing — the iOS client cannot reliably emit a stable
+ * `Idempotency-Key` for a 1 GB file, so we hash the bytes and dedup
+ * on content.
+ *
+ * Locks per `.planning/research/v1434-r-1-xml-import.md` §5.3.
+ */
+import type { NextRequest } from "next/server";
+
+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 {
+  APPLE_HEALTH_IMPORT_QUEUE,
+  type AppleHealthImportPayload,
+} from "@/lib/jobs/apple-health-import-worker";
+import { streamMultipartToDisk } from "@/lib/multipart/stream-to-disk";
+
+export const dynamic = "force-dynamic";
+
+/**
+ * Hard cap on the upload size. Apple's largest observed exports
+ * sit around 800 MB for a 10-year iCloud-synced account; 1.5 GB
+ * leaves comfortable margin for the long tail.
+ */
+const MAX_UPLOAD_BYTES = 1.5 * 1024 * 1024 * 1024;
+
+/** Rate-limit window: three uploads per minute per user. */
+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.apple-health.kickoff" } });
+
+  // Rate limit before touching the body — protects against a flood of
+  // 1 GB uploads from a single user.
+  const rl = await checkRateLimit(
+    `import:apple-health:${user.id}`,
+    RATE_LIMIT_MAX,
+    RATE_LIMIT_WINDOW_MS,
+  );
+  if (!rl.allowed) {
+    throw new HttpError(429, "Too many import uploads, try again later");
+  }
+
+  // Cheap pre-flight on the declared Content-Length. The streaming
+  // sink below enforces the hard cap as well — a missing/wrong
+  // content-length cannot bypass it.
+  const declaredBytes = Number(request.headers.get("content-length") ?? 0);
+  if (declaredBytes > MAX_UPLOAD_BYTES) {
+    await auditLog("import.apple-health.kickoff.denied", {
+      userId: user.id,
+      ipAddress: getClientIp(request),
+      details: {
+        reason: "content_length_exceeded",
+        contentLength: declaredBytes,
+      },
+    });
+    return apiError("Upload exceeds 1.5 GB limit", 413);
+  }
+
+  const body = request.body;
+  if (!body) {
+    return apiError("Request body is required", 400);
+  }
+
+  // Stream the upload to disk, computing a SHA-256 in the same pass.
+  let uploaded;
+  try {
+    uploaded = await streamMultipartToDisk(
+      body,
+      request.headers.get("content-type"),
+      {
+        maxBytes: MAX_UPLOAD_BYTES,
+        fieldName: "file",
+        tmpPrefix: "healthlog-apple-health-import",
+      },
+    );
+  } catch (err) {
+    const message = err instanceof Error ? err.message : "upload_failed";
+    await auditLog("import.apple-health.kickoff.denied", {
+      userId: user.id,
+      ipAddress: getClientIp(request),
+      details: { reason: "stream_to_disk_failed", message },
+    });
+    return apiError(`Multipart upload failed: ${message}`, 422);
+  }
+
+  // Content-hash idempotency: a re-upload of the same bytes resolves
+  // to the previous ImportJob row instead of re-queueing.
+  const existing = await prisma.importJob.findFirst({
+    where: {
+      userId: user.id,
+      uploadSha256: uploaded.sha256,
+    },
+    orderBy: { startedAt: "desc" },
+  });
+  if (existing) {
+    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.apple-health.kickoff.denied", {
+      userId: user.id,
+      ipAddress: getClientIp(request),
+      details: { reason: "worker_not_running" },
+    });
+    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,
+    },
+  });
+
+  const payload: AppleHealthImportPayload = {
+    userId: user.id,
+    uploadPath: uploaded.filePath,
+    uploadBytes: uploaded.bytes,
+    enqueuedAt: new Date().toISOString(),
+  };
+  const bossJobId = await boss.send(APPLE_HEALTH_IMPORT_QUEUE, payload);
+
+  await prisma.importJob.update({
+    where: { id: importJob.id },
+    data: { pgBossJobId: bossJobId },
+  });
+
+  await auditLog("import.apple-health.kickoff", {
+    userId: user.id,
+    ipAddress: getClientIp(request),
+    details: {
+      jobId: importJob.id,
+      bossJobId: bossJobId ?? null,
+      uploadBytes: uploaded.bytes,
+      uploadSha256: uploaded.sha256,
+    },
+  });
+
+  return apiSuccess(
+    {
+      jobId: importJob.id,
+      status: "queued" as const,
+    },
+    202,
+  );
+});

src/lib/multipart/stream-to-disk.ts

@@ -0,0 +1,289 @@
+/**
+ * Stream a `multipart/form-data` body to disk without buffering the
+ * whole payload in memory.
+ *
+ * Next.js 16's `request.formData()` is fine up to ~10 MB but allocates
+ * the full body on a 1 GB upload. The Apple Health export ingest
+ * therefore reads `request.body` directly (a `ReadableStream<Uint8Array>`
+ * per Web Streams) and walks the multipart boundary by hand.
+ *
+ * Only the named `file` field of the upload is captured; any other
+ * field-name parts are read and discarded so the stream drains
+ * cleanly. The captured field's bytes are written to a temp file and
+ * a SHA-256 digest is computed in the same pass — the digest powers
+ * the content-hash idempotency check in the kick-off endpoint.
+ *
+ * Locks per `.planning/research/v1434-r-1-xml-import.md` §5.4.
+ */
+import { createHash, randomUUID } from "node:crypto";
+import { createWriteStream } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { Writable } from "node:stream";
+
+export interface MultipartFileResult {
+  /** Absolute path the captured file landed at. */
+  filePath: string;
+  /** Total bytes streamed. */
+  bytes: number;
+  /** SHA-256 hex digest of the captured bytes. */
+  sha256: string;
+  /** Optional original filename pulled from `Content-Disposition`. */
+  originalFilename: string | null;
+  /** Other named text fields encountered in the multipart body. */
+  textFields: Record<string, string>;
+}
+
+export interface MultipartStreamOptions {
+  /** Maximum allowed file size in bytes (hard cap; throws on overflow). */
+  maxBytes: number;
+  /** Required multipart field name to capture. */
+  fieldName: string;
+  /** Temp directory; defaults to OS temp dir. */
+  tmpDir?: string;
+  /** Optional filename prefix; defaults to `healthlog-upload`. */
+  tmpPrefix?: string;
+}
+
+const CRLF = Buffer.from("\r\n");
+const DOUBLE_CRLF = Buffer.from("\r\n\r\n");
+
+/**
+ * Parse the multipart boundary from a `Content-Type` header. Returns
+ * `null` if the header is missing or unparseable.
+ */
+export function parseBoundary(contentType: string | null): string | null {
+  if (!contentType) return null;
+  const match = contentType.match(/boundary=(?:"([^"]+)"|([^;,\s]+))/i);
+  if (!match) return null;
+  return match[1] ?? match[2] ?? null;
+}
+
+/**
+ * Stream a multipart body from `req.body` to disk, returning the
+ * captured file's path + a SHA-256 digest. Throws on:
+ *   - Missing / unparseable boundary.
+ *   - Missing required field.
+ *   - Size cap exceeded.
+ */
+export async function streamMultipartToDisk(
+  body: ReadableStream<Uint8Array>,
+  contentType: string | null,
+  opts: MultipartStreamOptions,
+): Promise<MultipartFileResult> {
+  const boundary = parseBoundary(contentType);
+  if (!boundary) {
+    throw new Error("Multipart body missing boundary parameter");
+  }
+  const boundaryMarker = Buffer.from(`--${boundary}`);
+  const tmpDirectory = opts.tmpDir ?? tmpdir();
+  const filePath = join(
+    tmpDirectory,
+    `${opts.tmpPrefix ?? "healthlog-upload"}-${randomUUID()}.bin`,
+  );
+
+  const hash = createHash("sha256");
+  let bytesWritten = 0;
+  let originalFilename: string | null = null;
+  const textFields: Record<string, string> = {};
+
+  // Lazily created sink — opens once we have entered the matching part.
+  let fileSink: Writable | null = null;
+  let fileSinkOpen = false;
+
+  const ensureFileSink = (): Writable => {
+    if (!fileSink) {
+      fileSink = createWriteStream(filePath);
+      fileSinkOpen = true;
+    }
+    return fileSink;
+  };
+
+  // Buffer holding bytes awaiting boundary scan. Capped at twice the
+  // boundary length so we never accumulate more than necessary.
+  let buffer = Buffer.alloc(0);
+  // Current parser state:
+  //   "preamble"  — before the first boundary
+  //   "headers"   — inside a part's header block
+  //   "body-file" — inside the captured file's body
+  //   "body-text" — inside a text field's body
+  //   "epilogue"  — after the closing boundary
+  type ParserState = "preamble" | "headers" | "body-file" | "body-text" | "epilogue";
+  let state: ParserState = "preamble";
+  let currentFieldName: string | null = null;
+  let currentTextBuffer: Buffer[] = [];
+
+  const writeFileChunk = (chunk: Buffer): Promise<void> => {
+    return new Promise((resolve, reject) => {
+      if (bytesWritten + chunk.length > opts.maxBytes) {
+        reject(
+          new Error(
+            `Upload exceeds size cap of ${opts.maxBytes} bytes`,
+          ),
+        );
+        return;
+      }
+      bytesWritten += chunk.length;
+      hash.update(chunk);
+      const sink = ensureFileSink();
+      sink.write(chunk, (err) => {
+        if (err) reject(err);
+        else resolve();
+      });
+    });
+  };
+
+  const closeFileSink = (): Promise<void> => {
+    return new Promise((resolve) => {
+      if (!fileSink || !fileSinkOpen) {
+        resolve();
+        return;
+      }
+      fileSinkOpen = false;
+      fileSink.end(() => resolve());
+    });
+  };
+
+  const reader = body.getReader();
+  let streamDone = false;
+
+  // The classic SAX-style boundary scanner: accumulate into `buffer`,
+  // search for the boundary marker, emit bytes up to it.
+  const processBuffer = async (): Promise<void> => {
+    while (true) {
+      if (state === "preamble") {
+        const idx = buffer.indexOf(boundaryMarker);
+        if (idx === -1) {
+          // Need more bytes to find a boundary; retain a small tail
+          // so a boundary split across chunks is still detectable.
+          if (buffer.length > boundaryMarker.length) {
+            buffer = buffer.slice(buffer.length - boundaryMarker.length);
+          }
+          return;
+        }
+        // Drop everything up to and including the boundary marker.
+        buffer = buffer.slice(idx + boundaryMarker.length);
+        // After the first boundary we expect "\r\n" or "--" (closing).
+        if (buffer.length < 2) return;
+        if (buffer[0] === 0x2d && buffer[1] === 0x2d) {
+          state = "epilogue";
+          return;
+        }
+        if (buffer[0] === 0x0d && buffer[1] === 0x0a) {
+          buffer = buffer.slice(2);
+          state = "headers";
+          continue;
+        }
+        return;
+      }
+
+      if (state === "headers") {
+        const eoh = buffer.indexOf(DOUBLE_CRLF);
+        if (eoh === -1) return;
+        const headerBlock = buffer.slice(0, eoh).toString("utf8");
+        buffer = buffer.slice(eoh + DOUBLE_CRLF.length);
+
+        const disposition = headerBlock
+          .split(/\r?\n/)
+          .find((l) => l.toLowerCase().startsWith("content-disposition:"));
+        if (!disposition) {
+          // Unparseable part — skip until next boundary as text.
+          state = "body-text";
+          currentFieldName = null;
+          continue;
+        }
+        const nameMatch = disposition.match(/name="([^"]+)"/i);
+        const filenameMatch = disposition.match(/filename="([^"]*)"/i);
+        currentFieldName = nameMatch?.[1] ?? null;
+        if (currentFieldName === opts.fieldName) {
+          state = "body-file";
+          originalFilename = filenameMatch?.[1] ?? null;
+        } else {
+          state = "body-text";
+          currentTextBuffer = [];
+        }
+        continue;
+      }
+
+      // body-file / body-text: emit bytes up to the next CRLF--boundary.
+      const sep = Buffer.concat([CRLF, boundaryMarker]);
+      const idx = buffer.indexOf(sep);
+      if (idx === -1) {
+        // Emit everything except a trailing slice that might contain
+        // the start of the next boundary marker.
+        const keep = Math.min(buffer.length, sep.length - 1);
+        const emit = buffer.slice(0, buffer.length - keep);
+        if (emit.length > 0) {
+          if (state === "body-file") {
+            await writeFileChunk(emit);
+          } else {
+            currentTextBuffer.push(emit);
+          }
+        }
+        buffer = buffer.slice(buffer.length - keep);
+        return;
+      }
+
+      const emit = buffer.slice(0, idx);
+      if (emit.length > 0) {
+        if (state === "body-file") {
+          await writeFileChunk(emit);
+        } else {
+          currentTextBuffer.push(emit);
+        }
+      }
+      // Finalise the text field if applicable.
+      if (state === "body-text" && currentFieldName) {
+        textFields[currentFieldName] = Buffer.concat(
+          currentTextBuffer,
+        ).toString("utf8");
+        currentTextBuffer = [];
+      }
+      if (state === "body-file") {
+        await closeFileSink();
+      }
+      buffer = buffer.slice(idx + sep.length);
+      // After the boundary we expect "\r\n" or "--".
+      if (buffer.length < 2) {
+        state = "preamble";
+        return;
+      }
+      if (buffer[0] === 0x2d && buffer[1] === 0x2d) {
+        state = "epilogue";
+        return;
+      }
+      if (buffer[0] === 0x0d && buffer[1] === 0x0a) {
+        buffer = buffer.slice(2);
+        state = "headers";
+        continue;
+      }
+      return;
+    }
+  };
+
+  while (!streamDone) {
+    const { value, done } = await reader.read();
+    if (done) {
+      streamDone = true;
+      break;
+    }
+    buffer = Buffer.concat([buffer, Buffer.from(value)]);
+    await processBuffer();
+  }
+  await processBuffer();
+  await closeFileSink();
+
+  if (bytesWritten === 0) {
+    throw new Error(
+      `Multipart body did not include a '${opts.fieldName}' file field`,
+    );
+  }
+
+  return {
+    filePath,
+    bytes: bytesWritten,
+    sha256: hash.digest("hex"),
+    originalFilename,
+    textFields,
+  };
+}