From 73cd44e009ec8b7a61aca463c64f73ed3cc62a4a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sat, 16 May 2026 20:35:12 +0200 Subject: [PATCH] feat(import): Apple Health export.zip endpoints + multipart streamer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../admin/import-apple-health-export/route.ts | 192 ++++++++++++ .../[jobId]/status/route.ts | 77 +++++ .../api/import/apple-health-export/route.ts | 176 +++++++++++ src/lib/multipart/stream-to-disk.ts | 289 ++++++++++++++++++ 4 files changed, 734 insertions(+) create mode 100644 src/app/api/admin/import-apple-health-export/route.ts create mode 100644 src/app/api/import/apple-health-export/[jobId]/status/route.ts create mode 100644 src/app/api/import/apple-health-export/route.ts create mode 100644 src/lib/multipart/stream-to-disk.ts diff --git a/src/app/api/admin/import-apple-health-export/route.ts b/src/app/api/admin/import-apple-health-export/route.ts new file mode 100644 index 000000000..c8c01446e --- /dev/null +++ b/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, + ); +}); diff --git a/src/app/api/import/apple-health-export/[jobId]/status/route.ts b/src/app/api/import/apple-health-export/[jobId]/status/route.ts new file mode 100644 index 000000000..41e3e11f4 --- /dev/null +++ b/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; + result: Record | 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) ?? {}, + result: (row.result as Record | null) ?? null, + failureReason: row.failureReason, + }; + + return apiSuccess(body, 200); + }, +); diff --git a/src/app/api/import/apple-health-export/route.ts b/src/app/api/import/apple-health-export/route.ts new file mode 100644 index 000000000..4ff136a93 --- /dev/null +++ b/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, + ); +}); diff --git a/src/lib/multipart/stream-to-disk.ts b/src/lib/multipart/stream-to-disk.ts new file mode 100644 index 000000000..9e37f46ca --- /dev/null +++ b/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` + * 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; +} + +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, + contentType: string | null, + opts: MultipartStreamOptions, +): Promise { + 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 = {}; + + // 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 => { + 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 => { + 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 => { + 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, + }; +}