From 5cc0a00eafaa8fe985e3819a0a653124dcd631bb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Mon, 11 May 2026 00:20:37 +0200 Subject: [PATCH] feat(api): batch measurement ingest for the iOS Apple Health sync `POST /api/measurements/batch` accepts up to 500 HealthKit-shaped entries per call, runs each through the new mapping helper, and writes them under `source = APPLE_HEALTH` with the inbound sample UUID as the external id. The composite unique index on `(user_id, type, source, external_id)` does the per-entry dedup; `Idempotency-Key` covers the batch-level retry case so the iOS observer queue can replay in flight without double-writing. Per-entry status (`inserted` / `duplicate` / `skipped`) lets the client advance its sync cursor accurately even when individual rows land on different paths. Skip reasons cover unknown HealthKit identifiers and out-of-plausible-range values so a single bogus sensor reading never rejects an otherwise clean batch. Insertion uses chunked `createMany` with `skipDuplicates: true` so a race between two concurrent retries reconciles cleanly via the unique index instead of failing the transaction. Wide Event annotation logs `processed / inserted / duplicates / skipped` for ops visibility. Six integration tests against a real Postgres testcontainer cover the happy path, the per-entry duplicate path, the over-cap 422, mixed unmappable + out-of-range entries, per-stage sleep rows, and `Idempotency-Key` replay. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/app/api/measurements/batch/route.ts | 297 +++++++++++++++ tests/integration/measurements-batch.test.ts | 363 +++++++++++++++++++ 2 files changed, 660 insertions(+) create mode 100644 src/app/api/measurements/batch/route.ts create mode 100644 tests/integration/measurements-batch.test.ts diff --git a/src/app/api/measurements/batch/route.ts b/src/app/api/measurements/batch/route.ts new file mode 100644 index 000000000..5b89ea3bb --- /dev/null +++ b/src/app/api/measurements/batch/route.ts @@ -0,0 +1,297 @@ +/** + * `POST /api/measurements/batch` — Apple Health batch ingest. + * + * The iOS HealthLog app drains its HealthKit observer queue into this + * endpoint. Each entry is a HealthKit sample dressed in a thin + * envelope (identifier, value, unit, start/end dates, optional sleep + * stage codepoint, opaque external id). The server maps each entry + * through `mapAppleHealthEntry()`, inserts it as a `Measurement` row + * with `source = APPLE_HEALTH`, and returns a per-entry status so the + * client can advance its sync cursor accurately. + * + * Idempotency contract: + * - A single batch is dedupable end-to-end via `Idempotency-Key` — + * an HTTP-level retry replays the cached response. + * - Individual entries are dedupable via the + * `(user_id, type, source, external_id)` composite unique index — + * a partial-success retry, or two devices uploading the same + * HealthKit sample, surface as `duplicate` per entry instead of + * hard-failing the batch. + */ +import { NextRequest } from "next/server"; +import { z } from "zod/v4"; + +import { prisma } from "@/lib/db"; +import { apiHandler, requireAuth } from "@/lib/api-handler"; +import { annotate } from "@/lib/logging/context"; +import { auditLog } from "@/lib/auth/audit"; +import { apiError, apiSuccess, getClientIp, safeJson } from "@/lib/api-response"; +import { withIdempotency } from "@/lib/idempotency"; +import { mapAppleHealthEntry } from "@/lib/measurements/apple-health-mapping"; +import { validateMeasurementRange } from "@/lib/validations/measurement"; +import { Prisma } from "@/generated/prisma/client"; + +const MAX_BATCH_ENTRIES = 500; + +const batchEntrySchema = z.object({ + hkIdentifier: z.string().min(1).max(120), + value: z.number().finite(), + unit: z.string().min(1).max(60), + startDate: z.iso.datetime({ offset: true }), + endDate: z.iso.datetime({ offset: true }), + sleepStage: z.number().int().min(0).max(20).optional(), + externalId: z.string().min(1).max(120), + externalSourceVersion: z.string().min(1).max(120).optional(), +}); + +const batchPayloadSchema = z.object({ + entries: z.array(batchEntrySchema).min(1), +}); + +type BatchEntry = z.infer; + +/** + * Per-entry outcome the iOS client uses to advance its sync cursor. + * `inserted` and `duplicate` both indicate the row landed (or was + * already present), so the client can checkpoint past them. + * `skipped` is the only case where the client may want to surface a + * diagnostic — typically because Apple introduced a new identifier the + * server doesn't know about yet. + */ +type EntryStatus = "inserted" | "duplicate" | "skipped"; +interface EntryResult { + index: number; + status: EntryStatus; + reason?: string; +} + +export const POST = apiHandler(withIdempotency<[NextRequest]>(postBatch)); + +async function postBatch(request: NextRequest): Promise { + const { user } = await requireAuth(); + + const { data: rawBody, error: jsonError } = await safeJson(request); + if (jsonError) return jsonError; + + // Distinguish "too many entries" from "validation failed" so the + // client surfaces a useful diagnostic. The cap stops a buggy client + // from saturating the request pipeline; the iOS app paginates above + // this threshold. + if ( + typeof rawBody === "object" && + rawBody !== null && + "entries" in rawBody && + Array.isArray((rawBody as { entries: unknown }).entries) && + (rawBody as { entries: unknown[] }).entries.length > MAX_BATCH_ENTRIES + ) { + return apiError( + `Batch exceeds the ${MAX_BATCH_ENTRIES}-entry limit`, + 422, + { errorCode: "coach.batch.too_large" }, + ); + } + + const parsed = batchPayloadSchema.safeParse(rawBody); + if (!parsed.success) { + return apiError(parsed.error.issues[0]?.message ?? "Invalid batch", 422); + } + + const { entries } = parsed.data; + + // First pass — map each inbound entry; record skipped entries. + type Prepared = { + index: number; + entry: BatchEntry; + row: Prisma.MeasurementCreateManyInput; + }; + const prepared: Prepared[] = []; + const results: EntryResult[] = new Array(entries.length); + + for (let index = 0; index < entries.length; index++) { + const entry = entries[index]; + const mapped = mapAppleHealthEntry({ + hkIdentifier: entry.hkIdentifier, + value: entry.value, + unit: entry.unit, + startDate: entry.startDate, + endDate: entry.endDate, + sleepStage: entry.sleepStage, + }); + + if (!mapped) { + results[index] = { + index, + status: "skipped", + reason: "unmappable_identifier", + }; + continue; + } + + // Plausibility-range guard. We tolerate "out of plausible range" + // by skipping rather than failing the batch — a single bogus + // sensor reading shouldn't poison an otherwise clean ingest. + const rangeError = validateMeasurementRange(mapped.type, mapped.value); + if (rangeError !== null) { + results[index] = { + index, + status: "skipped", + reason: "value_out_of_range", + }; + continue; + } + + prepared.push({ + index, + entry, + row: { + userId: user.id, + type: mapped.type, + value: mapped.value, + unit: mapped.unit, + source: "APPLE_HEALTH", + measuredAt: mapped.takenAt, + externalId: entry.externalId, + externalSourceVersion: entry.externalSourceVersion ?? null, + sleepStage: mapped.sleepStage ?? null, + }, + }); + } + + let insertedCount = 0; + let duplicateCount = 0; + + if (prepared.length > 0) { + // Pre-flight duplicate detection so we can return per-entry status + // (createMany with skipDuplicates does the dedup but doesn't tell + // us *which* rows it skipped). We look up existing rows under the + // composite unique key first, then createMany the survivors. + const incomingKeys = prepared.map((p) => ({ + type: p.row.type as Prisma.MeasurementCreateManyInput["type"], + externalId: p.row.externalId as string, + })); + + const existing = await prisma.measurement.findMany({ + where: { + userId: user.id, + source: "APPLE_HEALTH", + OR: incomingKeys.map((k) => ({ + type: k.type, + externalId: k.externalId, + })), + }, + select: { type: true, externalId: true }, + }); + + const existingSet = new Set( + existing.map((row) => `${row.type}::${row.externalId}`), + ); + + const toInsert: Prisma.MeasurementCreateManyInput[] = []; + for (const p of prepared) { + const key = `${p.row.type}::${p.row.externalId}`; + if (existingSet.has(key)) { + results[p.index] = { index: p.index, status: "duplicate" }; + duplicateCount += 1; + } else { + results[p.index] = { index: p.index, status: "inserted" }; + toInsert.push(p.row); + } + } + + if (toInsert.length > 0) { + // Chunk to keep individual SQL statements bounded — 200 rows per + // chunk leaves headroom under Postgres' 65k-parameter cap even + // with 9-column inserts. `skipDuplicates: true` guards against + // races where two batch retries land in the same tick — the + // composite unique index does the actual dedup; skipDuplicates + // turns the duplicate-PG error into a no-op insert. + const CHUNK = 200; + const chunks: Prisma.MeasurementCreateManyInput[][] = []; + for (let i = 0; i < toInsert.length; i += CHUNK) { + chunks.push(toInsert.slice(i, i + CHUNK)); + } + await prisma.$transaction(async (tx) => { + for (const chunk of chunks) { + const result = await tx.measurement.createMany({ + data: chunk, + skipDuplicates: true, + }); + insertedCount += result.count; + } + }); + + // If `skipDuplicates` quietly absorbed a row (race with another + // batch), reconcile the per-entry status so the inserted + + // duplicate counts still equal `prepared.length`. + const racedDuplicates = toInsert.length - insertedCount; + if (racedDuplicates > 0) { + // We can't tell *which* rows raced from the createMany return + // value — recheck the DB and downgrade matching `inserted` + // rows to `duplicate`. This is a rare path so the extra + // round-trip is acceptable. + const recheck = await prisma.measurement.findMany({ + where: { + userId: user.id, + source: "APPLE_HEALTH", + OR: toInsert.map((row) => ({ + type: row.type, + externalId: row.externalId as string, + })), + }, + select: { type: true, externalId: true }, + }); + const stored = new Set( + recheck.map((row) => `${row.type}::${row.externalId}`), + ); + // Anything `prepared` flagged as `inserted` but isn't in + // `stored` was never written and was raced by another batch — + // surface as duplicate. + let downgraded = 0; + for (const p of prepared) { + if ( + results[p.index]?.status === "inserted" && + !stored.has(`${p.row.type}::${p.row.externalId}`) + ) { + results[p.index] = { index: p.index, status: "duplicate" }; + duplicateCount += 1; + downgraded += 1; + } + } + insertedCount -= downgraded; + } + } + } + + const skipped = results + .filter((r) => r.status === "skipped") + .map((r) => ({ index: r.index, reason: r.reason ?? "unknown" })); + + await auditLog("measurement.batch.ingest", { + userId: user.id, + ipAddress: getClientIp(request), + details: { + processed: entries.length, + inserted: insertedCount, + duplicates: duplicateCount, + skipped: skipped.length, + }, + }); + + annotate({ + action: { name: "measurement.batch.ingest" }, + meta: { + processed: entries.length, + inserted: insertedCount, + duplicates: duplicateCount, + skipped: skipped.length, + }, + }); + + return apiSuccess({ + processed: entries.length, + inserted: insertedCount, + duplicates: duplicateCount, + skipped, + entries: results, + }); +} diff --git a/tests/integration/measurements-batch.test.ts b/tests/integration/measurements-batch.test.ts new file mode 100644 index 000000000..9facd7fbf --- /dev/null +++ b/tests/integration/measurements-batch.test.ts @@ -0,0 +1,363 @@ +/** + * Integration suite for `POST /api/measurements/batch` — the iOS Apple + * Health ingest endpoint. Asserts the contract the iOS client relies on: + * - Mixed batches succeed end-to-end + * - Re-posting the same batch yields an idempotent outcome + * - Over-cap batches return 422 with the documented error code + * - Unmappable identifiers and out-of-range values surface as `skipped` + * - The composite unique index dedupes by `(user_id, type, source, external_id)` + * - Idempotency-Key replay returns the cached envelope + */ +import { NextRequest } from "next/server"; +import { beforeEach, describe, expect, it, vi } from "vitest"; + +import { cookieJar } from "./mock-next-headers"; +import { getPrismaClient, truncateAllTables } from "./setup"; + +const TEST_USER_ID = "user-batch-ingest-test"; + +vi.mock("next/headers", async () => { + const { cookieJar, headerJar } = await import("./mock-next-headers"); + return { + headers: vi.fn(async () => ({ + get: (name: string) => headerJar.get(name.toLowerCase()) ?? null, + })), + cookies: vi.fn(async () => ({ + get: (name: string) => { + const value = cookieJar.get(name); + return value ? { name, value } : undefined; + }, + set: (name: string, value: string) => { + cookieJar.set(name, value); + }, + delete: (name: string) => { + cookieJar.delete(name); + }, + })), + }; +}); + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +beforeEach(async () => { + await truncateAllTables(getPrismaClient()); + cookieJar.clear(); + await getPrismaClient().user.create({ + data: { + id: TEST_USER_ID, + username: "batch-ingest", + email: "batch-ingest@example.test", + }, + }); + const session = await getPrismaClient().session.create({ + data: { + userId: TEST_USER_ID, + expiresAt: new Date(Date.now() + 60 * 60 * 1000), + }, + }); + cookieJar.set("healthlog_session", session.id); +}); + +interface BatchEntryFixture { + hkIdentifier: string; + value: number; + unit: string; + startDate: string; + endDate: string; + externalId: string; + externalSourceVersion?: string; + sleepStage?: number; +} + +function makeRequest( + body: { entries: BatchEntryFixture[] }, + opts: { idempotencyKey?: string } = {}, +): NextRequest { + const headers: Record = { + "content-type": "application/json", + }; + if (opts.idempotencyKey) { + headers["idempotency-key"] = opts.idempotencyKey; + } + return new NextRequest("http://localhost/api/measurements/batch", { + method: "POST", + headers, + body: JSON.stringify(body), + }); +} + +describe("POST /api/measurements/batch (real Postgres)", () => { + it("inserts a well-formed batch and returns per-entry status", async () => { + const { POST } = await import("@/app/api/measurements/batch/route"); + + const body = { + entries: [ + { + hkIdentifier: "HKQuantityTypeIdentifierBodyMass", + value: 81.4, + unit: "kg", + startDate: "2026-05-09T07:30:00.000Z", + endDate: "2026-05-09T07:30:00.000Z", + externalId: "uuid-weight-001", + }, + { + hkIdentifier: "HKQuantityTypeIdentifierHeartRate", + value: 64, + unit: "count/min", + startDate: "2026-05-09T08:00:00.000Z", + endDate: "2026-05-09T08:00:00.000Z", + externalId: "uuid-pulse-001", + }, + { + hkIdentifier: "HKQuantityTypeIdentifierOxygenSaturation", + value: 0.97, + unit: "fraction", + startDate: "2026-05-09T08:05:00.000Z", + endDate: "2026-05-09T08:05:00.000Z", + externalId: "uuid-spo2-001", + }, + ], + }; + + const response = await POST(makeRequest(body)); + expect(response.status).toBe(200); + const json = (await response.json()) as { + data: { + processed: number; + inserted: number; + duplicates: number; + skipped: Array<{ index: number; reason: string }>; + }; + }; + + expect(json.data.processed).toBe(3); + expect(json.data.inserted).toBe(3); + expect(json.data.duplicates).toBe(0); + expect(json.data.skipped).toHaveLength(0); + + const stored = await getPrismaClient().measurement.findMany({ + where: { userId: TEST_USER_ID }, + orderBy: { measuredAt: "asc" }, + }); + expect(stored).toHaveLength(3); + expect(stored.every((r) => r.source === "APPLE_HEALTH")).toBe(true); + + const spo2 = stored.find((r) => r.type === "OXYGEN_SATURATION"); + expect(spo2?.value).toBeCloseTo(97); + expect(spo2?.unit).toBe("%"); + }); + + it("dedupes a re-posted batch into duplicate status without inserting twice", async () => { + const { POST } = await import("@/app/api/measurements/batch/route"); + + const body = { + entries: [ + { + hkIdentifier: "HKQuantityTypeIdentifierBodyMass", + value: 81.5, + unit: "kg", + startDate: "2026-05-09T07:30:00.000Z", + endDate: "2026-05-09T07:30:00.000Z", + externalId: "uuid-dup-001", + }, + { + hkIdentifier: "HKQuantityTypeIdentifierBodyMass", + value: 81.6, + unit: "kg", + startDate: "2026-05-09T08:00:00.000Z", + endDate: "2026-05-09T08:00:00.000Z", + externalId: "uuid-dup-002", + }, + ], + }; + + const first = await POST(makeRequest(body)); + expect(first.status).toBe(200); + const firstJson = (await first.json()) as { data: { inserted: number } }; + expect(firstJson.data.inserted).toBe(2); + + // Re-post the SAME body with NO idempotency-key. Per-entry dedup + // should kick in via the composite unique index, returning all + // duplicates and inserting nothing new. + const second = await POST(makeRequest(body)); + expect(second.status).toBe(200); + const secondJson = (await second.json()) as { + data: { inserted: number; duplicates: number }; + }; + expect(secondJson.data.inserted).toBe(0); + expect(secondJson.data.duplicates).toBe(2); + + const stored = await getPrismaClient().measurement.findMany({ + where: { userId: TEST_USER_ID }, + }); + expect(stored).toHaveLength(2); + }); + + it("returns 422 with coach.batch.too_large when entries exceed the cap", async () => { + const { POST } = await import("@/app/api/measurements/batch/route"); + + const entries: BatchEntryFixture[] = []; + for (let i = 0; i < 501; i++) { + entries.push({ + hkIdentifier: "HKQuantityTypeIdentifierBodyMass", + value: 80 + (i % 10) * 0.1, + unit: "kg", + startDate: "2026-05-09T07:30:00.000Z", + endDate: "2026-05-09T07:30:00.000Z", + externalId: `uuid-cap-${i}`, + }); + } + + const response = await POST(makeRequest({ entries })); + expect(response.status).toBe(422); + const json = (await response.json()) as { + error: string; + meta?: { errorCode?: string }; + }; + expect(json.error).toMatch(/500/); + expect(json.meta?.errorCode).toBe("coach.batch.too_large"); + }); + + it("flags unmappable identifiers and out-of-range values as skipped", async () => { + const { POST } = await import("@/app/api/measurements/batch/route"); + + const body = { + entries: [ + { + hkIdentifier: "HKQuantityTypeIdentifierHallucinated", + value: 42, + unit: "kg", + startDate: "2026-05-09T07:30:00.000Z", + endDate: "2026-05-09T07:30:00.000Z", + externalId: "uuid-skip-unknown", + }, + { + hkIdentifier: "HKQuantityTypeIdentifierBodyMass", + value: 9999, // out of plausible range + unit: "kg", + startDate: "2026-05-09T07:30:00.000Z", + endDate: "2026-05-09T07:30:00.000Z", + externalId: "uuid-skip-range", + }, + { + hkIdentifier: "HKQuantityTypeIdentifierBodyMass", + value: 81.5, + unit: "kg", + startDate: "2026-05-09T07:30:00.000Z", + endDate: "2026-05-09T07:30:00.000Z", + externalId: "uuid-good-001", + }, + ], + }; + + const response = await POST(makeRequest(body)); + expect(response.status).toBe(200); + const json = (await response.json()) as { + data: { + processed: number; + inserted: number; + skipped: Array<{ index: number; reason: string }>; + }; + }; + + expect(json.data.processed).toBe(3); + expect(json.data.inserted).toBe(1); + expect(json.data.skipped).toEqual( + expect.arrayContaining([ + { index: 0, reason: "unmappable_identifier" }, + { index: 1, reason: "value_out_of_range" }, + ]), + ); + + const stored = await getPrismaClient().measurement.findMany({ + where: { userId: TEST_USER_ID }, + }); + expect(stored).toHaveLength(1); + expect(stored[0]?.externalId).toBe("uuid-good-001"); + }); + + it("inserts per-stage sleep rows with the sleepStage column populated", async () => { + const { POST } = await import("@/app/api/measurements/batch/route"); + + const body = { + entries: [ + { + hkIdentifier: "HKCategoryTypeIdentifierSleepAnalysis", + value: 60, // minutes + unit: "min", + startDate: "2026-05-09T01:00:00.000Z", + endDate: "2026-05-09T02:00:00.000Z", + externalId: "uuid-sleep-core-001", + sleepStage: 3, // CORE + }, + { + hkIdentifier: "HKCategoryTypeIdentifierSleepAnalysis", + value: 90, + unit: "min", + startDate: "2026-05-09T02:00:00.000Z", + endDate: "2026-05-09T03:30:00.000Z", + externalId: "uuid-sleep-deep-001", + sleepStage: 4, // DEEP + }, + { + hkIdentifier: "HKCategoryTypeIdentifierSleepAnalysis", + value: 25, + unit: "min", + startDate: "2026-05-09T03:30:00.000Z", + endDate: "2026-05-09T03:55:00.000Z", + externalId: "uuid-sleep-rem-001", + sleepStage: 5, // REM + }, + ], + }; + + const response = await POST(makeRequest(body)); + expect(response.status).toBe(200); + + const stored = await getPrismaClient().measurement.findMany({ + where: { userId: TEST_USER_ID, type: "SLEEP_DURATION" }, + orderBy: { measuredAt: "asc" }, + }); + expect(stored).toHaveLength(3); + expect(stored.map((r) => r.sleepStage)).toEqual(["CORE", "DEEP", "REM"]); + expect(stored.every((r) => r.unit === "minutes")).toBe(true); + }); + + it("replays a cached response when the same Idempotency-Key is reused", async () => { + const { POST } = await import("@/app/api/measurements/batch/route"); + + const body = { + entries: [ + { + hkIdentifier: "HKQuantityTypeIdentifierBodyMass", + value: 81.7, + unit: "kg", + startDate: "2026-05-09T07:30:00.000Z", + endDate: "2026-05-09T07:30:00.000Z", + externalId: "uuid-idem-001", + }, + ], + }; + + const key = "ios-batch-12345678"; + const first = await POST(makeRequest(body, { idempotencyKey: key })); + expect(first.status).toBe(200); + const firstJson = (await first.json()) as { data: { inserted: number } }; + expect(firstJson.data.inserted).toBe(1); + + const second = await POST(makeRequest(body, { idempotencyKey: key })); + expect(second.status).toBe(200); + expect(second.headers.get("X-Idempotent-Replay")).toBe("true"); + const secondJson = (await second.json()) as { data: { inserted: number } }; + // Same envelope as the first call — even though it would otherwise + // have surfaced as a duplicate. + expect(secondJson.data.inserted).toBe(1); + + const stored = await getPrismaClient().measurement.findMany({ + where: { userId: TEST_USER_ID, externalId: "uuid-idem-001" }, + }); + expect(stored).toHaveLength(1); + }); +});