diff --git a/CHANGELOG.md b/CHANGELOG.md index ae667caed..154891c6c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -44,6 +44,18 @@ meaning: an app that posts only a five-point label still writes a complete entry, because the server derives the pleasantness value from that label. +- Workouts recorded before the phone learned to send heart-rate detail can + pick that detail up afterwards. When the app sends a session the server + already holds and the payload carries the reading-by-reading heart rate + this time, the curve is attached to the workout that is already there and + shows up on its detail page. The session itself is untouched: duration, + calories, the average and maximum heart rate, the times, all stay as they + were first recorded. A workout that already has its curve is left alone, + so the phone can walk back through years of history as often as it likes + without anything being written twice. If one session's readings arrive + unusable, that session alone is reported back and the rest of the batch + still lands. + ### Fixed - Marking an older missed dose as taken from the intake history now records diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index 026452bff..70132f280 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -3405,12 +3405,16 @@ paths: tags: - Measurements summary: Typed workout batch ingest (v1.4.25 W16b) - description: Server-side ingest endpoint for HKWorkout records (iOS) and Withings activity rows (server-to-server). Up + description: 'Server-side ingest endpoint for HKWorkout records (iOS) and Withings activity rows (server-to-server). Up to 100 workouts per call; nested route geometry (GeoJSON LineString) is capped at 20 000 points and stored in a 1:1 `WorkoutRoute` row keyed by `workoutId`. Idempotent via the `Idempotency-Key` header (replay window 24h). - Per-entry status (`inserted | duplicate | skipped`) lets the iOS sync cursor checkpoint accurately. The request - body ceiling is 5 MB enforced at the HTTP layer — clients above the ceiling receive a 413 with - `workout.batch.payload_too_large`. + Per-entry status (`inserted | duplicate | enriched | skipped`) lets the iOS sync cursor checkpoint accurately. + An entry matching a stored workout is first-write-wins for the workout row; when it carries a `samples` array + and the stored workout has no heart-rate series, the series is attached and the entry reports `enriched` + (repeatable: a workout that already has a series is a no-op). An entry whose `samples` array fails validation is + refused on its own with `reason: "invalid_samples"` — the rest of the batch still lands. The request body + ceiling is 5 MB enforced at the HTTP layer — clients above the ceiling receive a 413 with + `workout.batch.payload_too_large`.' requestBody: required: true content: @@ -21923,14 +21927,19 @@ components: type: integer minimum: 0 maximum: 9007199254740991 + description: Entries in the request. Equals `inserted + duplicates + skipped.length`. inserted: type: integer minimum: 0 maximum: 9007199254740991 + description: Workout rows this request wrote. duplicates: type: integer minimum: 0 maximum: 9007199254740991 + description: Entries that wrote no workout row because the workout was already stored. The top-level counters count + workout rows, so an entry reported as `enriched` counts here as well even though its heart-rate series was + attached. skipped: type: array items: @@ -21969,8 +21978,15 @@ components: enum: - inserted - duplicate + - enriched - skipped + description: "`inserted`: a new workout row. `duplicate`: the workout was already stored and nothing was written. + `enriched`: the workout was already stored and the heart-rate series this entry carried was attached to it; + the workout row itself was not written, and the entry also counts towards the top-level `duplicates`. + `skipped`: the server refused the entry, see `reason`." reason: + description: "Why a `skipped` entry was refused: `unstable_external_id` for an id that does not survive a client + restart, `invalid_samples` for a heart-rate series that failed validation." type: string required: - index diff --git a/src/app/api/workouts/__tests__/batch-create.test.ts b/src/app/api/workouts/__tests__/batch-create.test.ts index 1b8742e5b..05a5dba93 100644 --- a/src/app/api/workouts/__tests__/batch-create.test.ts +++ b/src/app/api/workouts/__tests__/batch-create.test.ts @@ -23,7 +23,7 @@ vi.mock("@/lib/db", () => ({ createMany: vi.fn(), }, workoutSamples: { - createMany: vi.fn(), + createManyAndReturn: vi.fn(), }, user: { findUnique: vi.fn(), @@ -109,6 +109,23 @@ function makeRequest( }); } +/** + * `workout.findMany` serves two callers in this route: the pre-flight + * probe for rows that already exist (it passes a `where`), and — inside + * the fake above — the stand-in for `INSERT ... RETURNING` (`select` + * only). Keying on `select.id` alone stopped telling the two apart once + * the probe began selecting `id` so an enrichment can address the stored + * workout, so the RETURNING arm is the one WITHOUT a `where`. + */ +function mockInsertReturning( + rows: Array<{ id: string; source: string; externalId: string }>, +) { + vi.mocked(prisma.workout.findMany).mockImplementation((async (args: { + where?: unknown; + select?: Record; + }) => (args.where === undefined && args.select?.id ? rows : [])) as never); +} + function validWorkout( externalId: string, overrides: Record = {}, @@ -161,7 +178,13 @@ beforeEach(() => { ); vi.mocked(prisma.workout.findFirst).mockResolvedValue(null); vi.mocked(prisma.workoutRoute.createMany).mockResolvedValue({ count: 0 }); - vi.mocked(prisma.workoutSamples.createMany).mockResolvedValue({ count: 0 }); + // The series write reports which rows landed, so an enrichment can tell + // an attach from a row a concurrent writer got in first. The fake + // echoes every row back as landed. + vi.mocked(prisma.workoutSamples.createManyAndReturn).mockImplementation( + (async (args: { data: Array<{ workoutId: string }> }) => + args.data.map((row) => ({ workoutId: row.workoutId }))) as never, + ); // v1.4.43 W9 — default to no per-user source-priority override so // the write-time picker walks the canonical default ladder. Tests // that exercise a custom ladder override this in-line. @@ -434,25 +457,18 @@ describe("POST /api/workouts/batch — per-entry status envelope", () => { createdAt: new Date("2026-05-14T10:00:01.000Z"), }, ] as never); - vi.mocked(prisma.workout.findMany).mockImplementation((async (args: { - select?: Record; - }) => { - if (args.select?.id) { - return [ - { - id: "apple-id", - source: "APPLE_HEALTH", - externalId: "shared-external-id", - }, - { - id: "manual-id", - source: "MANUAL", - externalId: "shared-external-id", - }, - ]; - } - return []; - }) as never); + mockInsertReturning([ + { + id: "apple-id", + source: "APPLE_HEALTH", + externalId: "shared-external-id", + }, + { + id: "manual-id", + source: "MANUAL", + externalId: "shared-external-id", + }, + ]); const res = await POST( makeRequest({ @@ -480,20 +496,13 @@ describe("POST /api/workouts/batch — nested route attachment", () => { vi.mocked(prisma.workout.createMany).mockResolvedValue({ count: 1 }); // Post-insert lookup returns the newly-inserted workout id so the // route can be attached by FK. - vi.mocked(prisma.workout.findMany).mockImplementation((async (args: { - select?: { id?: true; source?: true; externalId?: true }; - }) => { - if (args.select?.id) { - return [ - { - id: "wkt-fresh-id", - source: "APPLE_HEALTH", - externalId: "uuid-with-route", - }, - ]; - } - return []; - }) as never); + mockInsertReturning([ + { + id: "wkt-fresh-id", + source: "APPLE_HEALTH", + externalId: "uuid-with-route", + }, + ]); const res = await POST( makeRequest({ @@ -523,20 +532,13 @@ describe("POST /api/workouts/batch — nested route attachment", () => { describe("POST /api/workouts/batch — per-workout HR series (v1.10.0)", () => { it("creates a WorkoutSamples row for an indoor workout (samples, no route)", async () => { vi.mocked(prisma.workout.createMany).mockResolvedValue({ count: 1 }); - vi.mocked(prisma.workout.findMany).mockImplementation((async (args: { - select?: { id?: true }; - }) => { - if (args.select?.id) { - return [ - { - id: "wkt-indoor-id", - source: "APPLE_HEALTH", - externalId: "uuid-indoor", - }, - ]; - } - return []; - }) as never); + mockInsertReturning([ + { + id: "wkt-indoor-id", + source: "APPLE_HEALTH", + externalId: "uuid-indoor", + }, + ]); const res = await POST( makeRequest({ @@ -556,8 +558,9 @@ describe("POST /api/workouts/batch — per-workout HR series (v1.10.0)", () => { // The route table is untouched — an indoor workout has no GPS route. expect(prisma.workoutRoute.createMany).not.toHaveBeenCalled(); // The HR series lands in the dedicated child table by FK. - expect(prisma.workoutSamples.createMany).toHaveBeenCalledTimes(1); - const call = vi.mocked(prisma.workoutSamples.createMany).mock.calls[0]?.[0]; + expect(prisma.workoutSamples.createManyAndReturn).toHaveBeenCalledTimes(1); + const call = vi.mocked(prisma.workoutSamples.createManyAndReturn).mock + .calls[0]?.[0]; const rows = ( call as { data: Array<{ @@ -575,20 +578,13 @@ describe("POST /api/workouts/batch — per-workout HR series (v1.10.0)", () => { it("attaches both a route and an HR series on one workout", async () => { vi.mocked(prisma.workout.createMany).mockResolvedValue({ count: 1 }); - vi.mocked(prisma.workout.findMany).mockImplementation((async (args: { - select?: { id?: true }; - }) => { - if (args.select?.id) { - return [ - { - id: "wkt-both-id", - source: "APPLE_HEALTH", - externalId: "uuid-both", - }, - ]; - } - return []; - }) as never); + mockInsertReturning([ + { + id: "wkt-both-id", + source: "APPLE_HEALTH", + externalId: "uuid-both", + }, + ]); const res = await POST( makeRequest({ @@ -610,14 +606,15 @@ describe("POST /api/workouts/batch — per-workout HR series (v1.10.0)", () => { ); expect(res.status).toBe(200); expect(prisma.workoutRoute.createMany).toHaveBeenCalledTimes(1); - expect(prisma.workoutSamples.createMany).toHaveBeenCalledTimes(1); + expect(prisma.workoutSamples.createManyAndReturn).toHaveBeenCalledTimes(1); const routeRows = ( vi.mocked(prisma.workoutRoute.createMany).mock.calls[0]?.[0] as { data: Array<{ workoutId: string }>; } ).data; const sampleRows = ( - vi.mocked(prisma.workoutSamples.createMany).mock.calls[0]?.[0] as { + vi.mocked(prisma.workoutSamples.createManyAndReturn).mock + .calls[0]?.[0] as { data: Array<{ workoutId: string }>; } ).data; @@ -632,7 +629,7 @@ describe("POST /api/workouts/batch — per-workout HR series (v1.10.0)", () => { makeRequest({ workouts: [validWorkout("uuid-no-series")] }), ); expect(res.status).toBe(200); - expect(prisma.workoutSamples.createMany).not.toHaveBeenCalled(); + expect(prisma.workoutSamples.createManyAndReturn).not.toHaveBeenCalled(); }); }); diff --git a/src/app/api/workouts/batch/route.ts b/src/app/api/workouts/batch/route.ts index ac9f1c65d..61cfec22c 100644 --- a/src/app/api/workouts/batch/route.ts +++ b/src/app/api/workouts/batch/route.ts @@ -30,8 +30,8 @@ * - `Content-Length` ceiling at 5 MB before parsing — anything * larger returns 413 so the iOS client falls back to one workout * per call. - * - Per-entry status (`inserted | duplicate | skipped`) so the iOS - * sync cursor can checkpoint accurately. + * - Per-entry status (`inserted | duplicate | enriched | skipped`) so + * the iOS sync cursor can checkpoint accurately. * * Idempotency contract: * - HTTP-level via `Idempotency-Key` (replays cached envelope). @@ -39,6 +39,13 @@ * composite index. Re-posting the same batch surfaces duplicates * rather than failing the call. * + * Heart-rate series enrichment: a re-post that matches a stored + * workout and carries a `samples` array attaches that series when the + * workout has none. The workout row itself stays first-write-wins — + * nothing on it is written, not one column — and a workout that already + * holds a series is a no-op, so the client's history sweep is repeatable + * as often as it likes. See the enrichment block below. + * * Race reconciliation uses PostgreSQL's `INSERT ... RETURNING` through * Prisma's `createManyAndReturn`. The returned composite keys identify the * exact rows this request inserted, so per-entry statuses, child rows, and @@ -66,10 +73,11 @@ import { checkRateLimit } from "@/lib/rate-limit"; import { enqueuePrDetection } from "@/lib/jobs/pr-detection"; import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; import { emitDataArrival } from "@/lib/arrivals/emit-shared"; +import { MAX_WORKOUTS_PER_BATCH } from "@/lib/validations/workout"; import { - createBatchWorkoutSchema, - MAX_WORKOUTS_PER_BATCH, -} from "@/lib/validations/workout"; + INVALID_SAMPLES_REASON, + parseWorkoutBatch, +} from "@/lib/workouts/batch-ingest-parse"; import { classifyExternalId, unstableExternalIdMeta, @@ -106,16 +114,14 @@ const MAX_BODY_BYTES = 5 * 1024 * 1024; /** * Per-entry outcome the iOS client uses to advance its sync cursor. - * `inserted` and `duplicate` both indicate the row landed (or was - * already present); the client can checkpoint past them identically. - * `skipped` covers entries the server cannot durably store — currently - * unused by this endpoint (the Zod schema rejects malformed entries - * with a 422 before we reach the per-entry pass), but reserved on the - * envelope so the response shape parallels the measurements batch and - * future server-side validation can surface fine-grained skips without - * breaking the iOS DTO. + * `inserted`, `duplicate` and `enriched` all indicate the row landed (or + * was already present); the client can checkpoint past them identically. + * `enriched` additionally says the entry's heart-rate series was attached + * to a workout that had none, which is what lets a history sweep count + * its own progress. `skipped` covers entries the server refused: an + * unusable external id, or a `samples` array that failed validation. */ -type EntryStatus = "inserted" | "duplicate" | "skipped"; +type EntryStatus = "inserted" | "duplicate" | "enriched" | "skipped"; interface EntryResult { index: number; status: EntryStatus; @@ -251,15 +257,18 @@ async function postBatch(request: NextRequest): Promise { ); } - const parsed = createBatchWorkoutSchema.safeParse(rawBody); - if (!parsed.success) { - const issue = parsed.error.issues[0]; - return apiError(issue?.message ?? "Invalid batch", 400, { + // Strict on every field, per-entry on `samples` alone — see + // `parseWorkoutBatch` for why the series is the one field whose + // failure must not take the rest of a history sweep down with it. + const parsed = parseWorkoutBatch(rawBody); + if (!parsed.ok) { + return apiError(parsed.message, 400, { errorCode: "workout.batch.invalid", }); } const { workouts } = parsed.data; + const refusedSampleIndices = new Set(parsed.refusedSampleIndices); // Pre-flight duplicate detection so we can return per-entry status. // The composite unique index is `(userId, source, externalId)` — @@ -289,6 +298,18 @@ async function postBatch(request: NextRequest): Promise { const results: EntryResult[] = new Array(workouts.length); const unstableShapes: UnstableExternalIdShape[] = []; const prepared: Prepared[] = workouts.flatMap((w, index) => { + // An entry whose series failed validation is refused whole. Storing + // the workout without the curve the caller asked us to keep would be + // a silent loss, and the row may well already exist anyway. + if (refusedSampleIndices.has(index)) { + results[index] = { + index, + status: "skipped", + reason: INVALID_SAMPLES_REASON, + }; + return []; + } + // The `(userId, source, externalId)` dedup key is only idempotent // while the id is STABLE across client launches. An id that rotates // per process (an object description carrying a memory address) @@ -408,6 +429,7 @@ async function postBatch(request: NextRequest): Promise { const survivors = prepared.filter((p) => survivingIndices.has(p.index)); let duplicateCount = droppedByWriteDedup.length; let insertedCount = 0; + let enrichedCount = 0; const insertedIdByIndex = new Map(); if (survivors.length > 0) { @@ -418,6 +440,9 @@ async function postBatch(request: NextRequest): Promise { .map((p) => p.dedupKey) .filter((k): k is { source: string; externalId: string } => k !== null); + // The same probe answers both questions: does the row exist, and + // does it already carry a series. `id` comes along so an enrichment + // can address the stored workout without a second round trip. const existing = dedupCandidates.length ? await prisma.workout.findMany({ where: { @@ -427,30 +452,65 @@ async function postBatch(request: NextRequest): Promise { externalId: k.externalId, })), }, - select: { source: true, externalId: true }, + select: { + id: true, + source: true, + externalId: true, + samples: { select: { workoutId: true } }, + }, }) : []; - const existingSet = new Set( - existing.map((row) => `${row.source}::${row.externalId}`), - ); + const existingByKey = new Map(); + for (const row of existing) { + if (row.externalId === null) continue; + existingByKey.set(`${row.source}::${row.externalId}`, { + id: row.id, + hasSeries: row.samples !== null, + }); + } const toInsert: Prepared[] = []; + /** Stored workouts this batch hands a heart-rate series to. */ + const toEnrich: Array<{ + index: number; + workoutId: string; + samples: NonNullable; + }> = []; + /** Workouts already claimed by an earlier entry of THIS batch. */ + const enrichClaimed = new Set(); for (const p of survivors) { const keyTuple = p.dedupKey ? `${p.dedupKey.source}::${p.dedupKey.externalId}` : null; - if (keyTuple !== null && existingSet.has(keyTuple)) { - results[p.index] = { index: p.index, status: "duplicate" }; - duplicateCount += 1; + const known = keyTuple !== null ? existingByKey.get(keyTuple) : undefined; + if (known) { + if (p.samples && !known.hasSeries && !enrichClaimed.has(known.id)) { + enrichClaimed.add(known.id); + // Enrichment. The status is assigned after the write, because + // a concurrent writer may get the series in first — in which + // case this entry is the plain duplicate it always was. A + // second entry of this batch naming the same workout takes the + // duplicate arm above: one workout, one series, one `enriched`. + toEnrich.push({ + index: p.index, + workoutId: known.id, + samples: p.samples, + }); + } else { + results[p.index] = { index: p.index, status: "duplicate" }; + duplicateCount += 1; + } } else { results[p.index] = { index: p.index, status: "inserted" }; toInsert.push(p); } } - if (toInsert.length > 0) { + if (toInsert.length > 0 || toEnrich.length > 0) { const CHUNK = 100; + /** Workout ids whose series row this request actually wrote. */ + const seriesLanded = new Set(); await prisma.$transaction(async (tx) => { const withExternalId = toInsert.filter((p) => p.dedupKey !== null); @@ -517,6 +577,19 @@ async function postBatch(request: NextRequest): Promise { } } + // Enrichment rides the SAME child write as a fresh workout's + // series — one mechanism, one place where a series is stored. + // The only difference is where the `workoutId` came from: the + // pre-flight probe instead of this request's INSERT ... RETURNING. + // Nothing here writes to the workout row. + for (const e of toEnrich) { + samplesToInsert.push({ + workoutId: e.workoutId, + samples: e.samples.samples as Prisma.InputJsonValue, + sampleCount: e.samples.sampleCount, + }); + } + for (let i = 0; i < routesToInsert.length; i += CHUNK) { await tx.workoutRoute.createMany({ data: routesToInsert.slice(i, i + CHUNK), @@ -524,10 +597,16 @@ async function postBatch(request: NextRequest): Promise { }); } for (let i = 0; i < samplesToInsert.length; i += CHUNK) { - await tx.workoutSamples.createMany({ + // RETURNING names the rows that actually landed. `workoutId` is + // unique, so a workout that gained a series between the probe + // and this write is skipped here and reports as a duplicate — + // the re-post stays a no-op and never overwrites a stored curve. + const landed = await tx.workoutSamples.createManyAndReturn({ data: samplesToInsert.slice(i, i + CHUNK), skipDuplicates: true, + select: { workoutId: true }, }); + for (const row of landed) seriesLanded.add(row.workoutId); } }); @@ -540,6 +619,18 @@ async function postBatch(request: NextRequest): Promise { duplicateCount += 1; } } + + // An enriched entry is still a duplicate of the workout row: the + // top-level counters count rows, and no row was inserted for it. + // The per-entry status carries what happened to the series. + for (const e of toEnrich) { + results[e.index] = { + index: e.index, + status: seriesLanded.has(e.workoutId) ? "enriched" : "duplicate", + }; + if (seriesLanded.has(e.workoutId)) enrichedCount += 1; + duplicateCount += 1; + } } } @@ -554,6 +645,7 @@ async function postBatch(request: NextRequest): Promise { processed: workouts.length, inserted: insertedCount, duplicates: duplicateCount, + enriched: enrichedCount, skipped: skipped.length, }, }); @@ -584,12 +676,14 @@ async function postBatch(request: NextRequest): Promise { } } + // Counts only. The series itself never reaches a wide event. annotate({ action: { name: "workout.batch.ingest" }, meta: { processed: workouts.length, inserted: insertedCount, duplicates: duplicateCount, + enriched: enrichedCount, skipped: skipped.length, }, }); @@ -598,11 +692,17 @@ async function postBatch(request: NextRequest): Promise { // caches when at least one row landed. Workouts ride on the // measurements bucket because achievements / analytics also touch // workout-derived metrics. - if (insertedCount > 0) { + // An enrichment changes what the workout reads as — the detail seam + // gains its curve, the list gains its glyph — so it busts the cache + // too, even though no workout row was written. + if (insertedCount > 0 || enrichedCount > 0) { invalidateUserMeasurements(user.id); + } + if (insertedCount > 0) { // One arrival per exact INSERT ... RETURNING winner. Historical rows still // stop at the shared salience classifier before any queue work. + // An enrichment raises none: the workout arrived when it was stored. void emitWorkoutArrivals(user.id, prepared, insertedIdByIndex).catch( () => {}, ); diff --git a/src/lib/openapi/routes/workouts.ts b/src/lib/openapi/routes/workouts.ts index 9c4b2f910..42b312b27 100644 --- a/src/lib/openapi/routes/workouts.ts +++ b/src/lib/openapi/routes/workouts.ts @@ -20,22 +20,55 @@ import { // v1.4.25 W16b — typed workout batch ingest response envelope. Mirrors // the measurements batch shape but reports the `workouts` count rather -// than the entries count, and the `skipped` field is reserved (the -// Zod schema rejects malformed entries with a 400 before the per-entry -// pass, so today's responses always carry an empty array). +// than the entries count. `skipped` carries the entries the server +// refused: an unusable external id, or a `samples` array that failed +// validation (`reason: "invalid_samples"`). +// +// `enriched` says the entry matched a stored workout that had no +// heart-rate series and the series it carried was attached. The +// workout row itself was not written, so such an entry also counts +// towards `duplicates`. That rule is part of the published contract, +// not just a note here: it rides the field descriptions below, because +// a client summing the counters has to know which bucket an +// enrichment lands in. const workoutBatchEntryResult = z .object({ index: z.number().int().nonnegative(), - status: z.enum(["inserted", "duplicate", "skipped"]), - reason: z.string().optional(), + status: z + .enum(["inserted", "duplicate", "enriched", "skipped"]) + .describe( + "`inserted`: a new workout row. `duplicate`: the workout was already stored and nothing was written. `enriched`: the workout was already stored and the heart-rate series this entry carried was attached to it; the workout row itself was not written, and the entry also counts towards the top-level `duplicates`. `skipped`: the server refused the entry, see `reason`.", + ), + reason: z + .string() + .optional() + .describe( + "Why a `skipped` entry was refused: `unstable_external_id` for an id that does not survive a client restart, `invalid_samples` for a heart-rate series that failed validation.", + ), }) .meta({ id: "WorkoutBatchEntryResult" }); const workoutBatchResponse = z .object({ - processed: z.number().int().nonnegative(), - inserted: z.number().int().nonnegative(), - duplicates: z.number().int().nonnegative(), + processed: z + .number() + .int() + .nonnegative() + .describe( + "Entries in the request. Equals `inserted + duplicates + skipped.length`.", + ), + inserted: z + .number() + .int() + .nonnegative() + .describe("Workout rows this request wrote."), + duplicates: z + .number() + .int() + .nonnegative() + .describe( + "Entries that wrote no workout row because the workout was already stored. The top-level counters count workout rows, so an entry reported as `enriched` counts here as well even though its heart-rate series was attached.", + ), skipped: z.array( z.object({ index: z.number().int().nonnegative(), @@ -275,7 +308,7 @@ export const workoutPaths: NonNullable = { tags: ["Measurements"], summary: "Typed workout batch ingest (v1.4.25 W16b)", description: - "Server-side ingest endpoint for HKWorkout records (iOS) and Withings activity rows (server-to-server). Up to 100 workouts per call; nested route geometry (GeoJSON LineString) is capped at 20 000 points and stored in a 1:1 `WorkoutRoute` row keyed by `workoutId`. Idempotent via the `Idempotency-Key` header (replay window 24h). Per-entry status (`inserted | duplicate | skipped`) lets the iOS sync cursor checkpoint accurately. The request body ceiling is 5 MB enforced at the HTTP layer — clients above the ceiling receive a 413 with `workout.batch.payload_too_large`.", + 'Server-side ingest endpoint for HKWorkout records (iOS) and Withings activity rows (server-to-server). Up to 100 workouts per call; nested route geometry (GeoJSON LineString) is capped at 20 000 points and stored in a 1:1 `WorkoutRoute` row keyed by `workoutId`. Idempotent via the `Idempotency-Key` header (replay window 24h). Per-entry status (`inserted | duplicate | enriched | skipped`) lets the iOS sync cursor checkpoint accurately. An entry matching a stored workout is first-write-wins for the workout row; when it carries a `samples` array and the stored workout has no heart-rate series, the series is attached and the entry reports `enriched` (repeatable: a workout that already has a series is a no-op). An entry whose `samples` array fails validation is refused on its own with `reason: "invalid_samples"` — the rest of the batch still lands. The request body ceiling is 5 MB enforced at the HTTP layer — clients above the ceiling receive a 413 with `workout.batch.payload_too_large`.', requestBody: { required: true, content: { "application/json": { schema: createBatchWorkoutSchema } }, diff --git a/src/lib/workouts/__tests__/batch-ingest-parse.test.ts b/src/lib/workouts/__tests__/batch-ingest-parse.test.ts new file mode 100644 index 000000000..7be52ce3d --- /dev/null +++ b/src/lib/workouts/__tests__/batch-ingest-parse.test.ts @@ -0,0 +1,70 @@ +import { describe, expect, it } from "vitest"; + +import { + INVALID_SAMPLES_REASON, + parseWorkoutBatch, +} from "@/lib/workouts/batch-ingest-parse"; + +function entry(overrides: Record = {}) { + return { + sportType: "running", + startedAt: "2026-05-14T06:30:00.000Z", + endedAt: "2026-05-14T07:15:00.000Z", + source: "APPLE_HEALTH", + externalId: "hk-uuid-1", + ...overrides, + }; +} + +describe("parseWorkoutBatch", () => { + it("parses a clean batch and refuses nothing", () => { + const result = parseWorkoutBatch({ + workouts: [ + entry({ samples: [{ t: "2026-05-14T06:30:00.000Z", hr: 120 }] }), + ], + }); + expect(result.ok).toBe(true); + if (!result.ok) return; + expect(result.refusedSampleIndices).toEqual([]); + expect(result.data.workouts[0]?.samples).toHaveLength(1); + }); + + it("refuses only the entries whose samples array is malformed", () => { + const result = parseWorkoutBatch({ + workouts: [ + entry({ externalId: "hk-a", samples: [{ t: "nope", hr: 4000 }] }), + entry({ externalId: "hk-b" }), + entry({ externalId: "hk-c", samples: [] }), + ], + }); + expect(result.ok).toBe(true); + if (!result.ok) return; + expect(result.refusedSampleIndices).toEqual([0, 2]); + // The survivors keep their full typed shape. + expect(result.data.workouts).toHaveLength(3); + expect(result.data.workouts[1]?.externalId).toBe("hk-b"); + expect(result.data.workouts[0]?.samples).toBeUndefined(); + }); + + it("keeps failing the whole batch when any other field is wrong", () => { + const result = parseWorkoutBatch({ + workouts: [ + entry({ endedAt: "2026-05-14T06:00:00.000Z" }), + entry({ externalId: "hk-b", samples: [{ t: "nope" }] }), + ], + }); + expect(result.ok).toBe(false); + if (result.ok) return; + expect(result.message).toMatch(/endedAt/i); + }); + + it("fails the batch when the envelope itself is not a batch", () => { + expect(parseWorkoutBatch({ workouts: [] }).ok).toBe(false); + expect(parseWorkoutBatch(null).ok).toBe(false); + expect(parseWorkoutBatch({ workouts: "nope" }).ok).toBe(false); + }); + + it("names the per-entry reason code the route reports", () => { + expect(INVALID_SAMPLES_REASON).toBe("invalid_samples"); + }); +}); diff --git a/src/lib/workouts/batch-ingest-parse.ts b/src/lib/workouts/batch-ingest-parse.ts new file mode 100644 index 000000000..56b9ac7cf --- /dev/null +++ b/src/lib/workouts/batch-ingest-parse.ts @@ -0,0 +1,109 @@ +/** + * Batch-ingest body parsing for `POST /api/workouts/batch`. + * + * The wire contract is `createBatchWorkoutSchema` and stays exactly that: + * this module never widens what the endpoint accepts. What it changes is + * WHERE one specific failure lands. + * + * A malformed field anywhere in the batch fails the whole call with a 400, + * which is right for a client that got the shape wrong. The `samples` + * array is the exception, because of how it arrives: the native client + * re-posts workouts the server already stores purely to hand over their + * heart-rate curve, sweeping years of history in batches of up to a + * hundred. One session whose series came out of HealthKit malformed would + * otherwise stall the entire sweep at a 400 the client cannot repair by + * retrying — every other workout in that batch is fine and has nothing to + * do with the bad one. + * + * So an entry whose `samples` array fails validation is refused on its + * own, exactly the way an entry with an unusable `externalId` already is + * (see `classifyExternalId` in the route), and the rest of the batch + * lands. The refused entry is NOT stripped down and stored without its + * series: the caller asked for the curve to be saved, so a half-write + * would be a silent loss. It is reported back with its index so the + * client knows which session to fix. + * + * Every other validation failure keeps failing the batch. + */ +import { + createBatchWorkoutSchema, + type CreateBatchWorkoutInput, +} from "@/lib/validations/workout"; + +/** Per-entry `reason` code for a `samples` array that failed validation. */ +export const INVALID_SAMPLES_REASON = "invalid_samples"; + +export type WorkoutBatchParseResult = + | { + ok: true; + data: CreateBatchWorkoutInput; + /** Batch indices whose `samples` array was refused. */ + refusedSampleIndices: number[]; + } + | { ok: false; message: string }; + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +/** + * Indices of entries whose ONLY problem is the `samples` array. An issue + * path of `["workouts", 3, "samples", ...]` names entry 3; anything else + * means the batch is broken in a way this path does not forgive. + */ +function collectSampleFailureIndices( + issues: ReadonlyArray<{ path: PropertyKey[] }>, +): number[] | null { + const indices = new Set(); + for (const issue of issues) { + const [root, index, field] = issue.path; + if (root !== "workouts" || typeof index !== "number" || field !== "samples") + return null; + indices.add(index); + } + return indices.size > 0 ? [...indices].sort((a, b) => a - b) : null; +} + +/** + * Parses a batch body, refusing per entry when the only failures are + * `samples` arrays. On that path the offending entries are re-parsed + * without their series so the surviving entries keep their fully typed + * shape; the caller drops the refused indices before any write. + */ +export function parseWorkoutBatch(rawBody: unknown): WorkoutBatchParseResult { + const strict = createBatchWorkoutSchema.safeParse(rawBody); + if (strict.success) { + return { ok: true, data: strict.data, refusedSampleIndices: [] }; + } + + const refusedSampleIndices = collectSampleFailureIndices(strict.error.issues); + if ( + refusedSampleIndices === null || + !isRecord(rawBody) || + !Array.isArray(rawBody.workouts) + ) { + return { + ok: false, + message: strict.error.issues[0]?.message ?? "Invalid batch", + }; + } + + const refused = new Set(refusedSampleIndices); + const stripped = { + ...rawBody, + workouts: rawBody.workouts.map((entry, index) => { + if (!refused.has(index) || !isRecord(entry)) return entry; + const { samples: _samples, ...rest } = entry; + return rest; + }), + }; + + const relaxed = createBatchWorkoutSchema.safeParse(stripped); + if (!relaxed.success) { + return { + ok: false, + message: relaxed.error.issues[0]?.message ?? "Invalid batch", + }; + } + return { ok: true, data: relaxed.data, refusedSampleIndices }; +} diff --git a/tests/integration/workout-batch-enrich.test.ts b/tests/integration/workout-batch-enrich.test.ts new file mode 100644 index 000000000..3b80f526e --- /dev/null +++ b/tests/integration/workout-batch-enrich.test.ts @@ -0,0 +1,460 @@ +/** + * Integration suite for the heart-rate series enrichment path of + * `POST /api/workouts/batch`, against a real Postgres in a testcontainer. + * + * The case: the native client posted a workout before it learned to send + * the per-workout HR series. The row exists, the curve does not. A later + * re-post of the SAME `(source, externalId)` carrying a `samples` array + * attaches the series to the stored workout and reports the entry as + * `enriched`. + * + * What the suite pins: + * - The attach happens once and reports `enriched` + * - A second re-post over an existing series is a no-op and never + * reports `enriched` twice + * - The workout row is byte-identical across the enrichment — every + * column, `updatedAt` included, compared as a whole row + * - A malformed `samples` array is refused per entry; the rest of the + * batch still lands + * - A batch mixing insert / enrich / duplicate / skip reports each + * entry with its own status + */ +import { NextRequest } from "next/server"; +import { beforeEach, describe, expect, it, vi } from "vitest"; + +import { cookieJar, headerJar } from "./mock-next-headers"; +import { getPrismaClient, truncateAllTables } from "./setup"; + +process.env.API_TOKEN_HMAC_KEY ??= + "test-hmac-key-workout-enrich-integration-32-bytes-min-1234567890"; +process.env.ENCRYPTION_KEY = + "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; + +const TEST_USER_ID = "user-workout-enrich-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(); + headerJar.clear(); + await getPrismaClient().user.create({ + data: { + id: TEST_USER_ID, + username: "workout-enrich", + email: "workout-enrich@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 WorkoutFixture { + sportType?: string; + startedAt?: string; + endedAt?: string; + totalEnergyKcal?: number; + totalDistanceM?: number; + avgHeartRate?: number; + maxHeartRate?: number; + minHeartRate?: number; + source?: string; + externalId?: string; + metadata?: Record; + samples?: unknown; +} + +interface BatchResponse { + processed: number; + inserted: number; + duplicates: number; + skipped: Array<{ index: number; reason: string }>; + entries: Array<{ index: number; status: string; reason?: string }>; +} + +function makeRequest(body: { workouts: WorkoutFixture[] }): NextRequest { + return new NextRequest("http://localhost/api/workouts/batch", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(body), + }); +} + +function baseWorkout( + externalId: string, + overrides: WorkoutFixture = {}, +): WorkoutFixture { + return { + sportType: "running", + startedAt: "2026-05-14T06:30:00.000Z", + endedAt: "2026-05-14T07:15:00.000Z", + source: "APPLE_HEALTH", + externalId, + ...overrides, + }; +} + +const SERIES = [ + { t: "2026-05-14T06:30:00.000Z", hr: 121 }, + { t: "2026-05-14T06:30:05.000Z", hr: 128, cadence: 168 }, + { t: "2026-05-14T06:30:10.000Z", hr: 133 }, +]; + +async function post(body: { + workouts: WorkoutFixture[]; +}): Promise<{ status: number; data: BatchResponse }> { + const { POST } = await import("@/app/api/workouts/batch/route"); + const res = await POST(makeRequest(body)); + const json = (await res.json()) as { data: BatchResponse }; + return { status: res.status, data: json.data }; +} + +describe("POST /api/workouts/batch — HR series enrichment (real Postgres)", () => { + it("attaches the series to a stored workout that has none and reports enriched", async () => { + const first = await post({ + workouts: [baseWorkout("hk-uuid-enrich-001")], + }); + expect(first.status).toBe(200); + expect(first.data.inserted).toBe(1); + expect( + await getPrismaClient().workoutSamples.count({ + where: { workout: { userId: TEST_USER_ID } }, + }), + ).toBe(0); + + const second = await post({ + workouts: [baseWorkout("hk-uuid-enrich-001", { samples: SERIES })], + }); + expect(second.status).toBe(200); + expect(second.data.inserted).toBe(0); + expect(second.data.entries).toEqual([{ index: 0, status: "enriched" }]); + + const stored = await getPrismaClient().workout.findFirst({ + where: { userId: TEST_USER_ID, externalId: "hk-uuid-enrich-001" }, + include: { samples: true }, + }); + expect(stored?.samples).not.toBeNull(); + expect(stored?.samples?.sampleCount).toBe(3); + expect(stored?.samples?.samples).toEqual(SERIES); + + // The workout row count is unchanged — enrichment never mints a row. + expect( + await getPrismaClient().workout.count({ + where: { userId: TEST_USER_ID }, + }), + ).toBe(1); + }); + + it("treats a re-post over an existing series as a no-op and never reports enriched twice", async () => { + await post({ workouts: [baseWorkout("hk-uuid-enrich-002")] }); + const enriching = await post({ + workouts: [baseWorkout("hk-uuid-enrich-002", { samples: SERIES })], + }); + expect(enriching.data.entries[0]?.status).toBe("enriched"); + + const storedSeries = await getPrismaClient().workoutSamples.findFirst({ + where: { workout: { externalId: "hk-uuid-enrich-002" } }, + }); + expect(storedSeries).not.toBeNull(); + + const again = await post({ + workouts: [ + baseWorkout("hk-uuid-enrich-002", { + samples: [ + ...SERIES, + { t: "2026-05-14T06:30:15.000Z", hr: 141 }, + { t: "2026-05-14T06:30:20.000Z", hr: 145 }, + ], + }), + ], + }); + expect(again.status).toBe(200); + expect(again.data.entries[0]?.status).toBe("duplicate"); + expect(again.data.duplicates).toBe(1); + + // The stored series is the first one, untouched — the no-op does not + // overwrite a curve the server already holds. + const afterSeries = await getPrismaClient().workoutSamples.findFirst({ + where: { workout: { externalId: "hk-uuid-enrich-002" } }, + }); + expect(afterSeries).toEqual(storedSeries); + expect( + await getPrismaClient().workoutSamples.count({ + where: { workout: { userId: TEST_USER_ID } }, + }), + ).toBe(1); + }); + + it("hands the attach to exactly one of two concurrent enrichments", async () => { + await post({ workouts: [baseWorkout("hk-uuid-enrich-008")] }); + + // Both requests probe a workout with no series, so both mean to + // attach one. The unique index on `workout_id` decides: one write + // lands, the other conflicts and is skipped. The loser must report + // the duplicate it turned out to be rather than claim an attach the + // database never made — that is the arm this case exists for, and + // the outcome is deterministic even though the winner is not. + const { POST } = await import("@/app/api/workouts/batch/route"); + const body = { + workouts: [baseWorkout("hk-uuid-enrich-008", { samples: SERIES })], + }; + const [resA, resB] = await Promise.all([ + POST(makeRequest(body)), + POST(makeRequest(body)), + ]); + expect(resA.status).toBe(200); + expect(resB.status).toBe(200); + + const [jsonA, jsonB] = (await Promise.all([resA.json(), resB.json()])) as [ + { data: BatchResponse }, + { data: BatchResponse }, + ]; + const statuses = [ + jsonA.data.entries[0]?.status, + jsonB.data.entries[0]?.status, + ].sort(); + expect(statuses).toEqual(["duplicate", "enriched"]); + // Both responses still count the entry as a duplicate row. + expect(jsonA.data.duplicates).toBe(1); + expect(jsonB.data.duplicates).toBe(1); + expect(jsonA.data.inserted + jsonB.data.inserted).toBe(0); + + const series = await getPrismaClient().workoutSamples.findMany({ + where: { workout: { userId: TEST_USER_ID } }, + }); + expect(series).toHaveLength(1); + expect(series[0]?.sampleCount).toBe(3); + expect( + await getPrismaClient().workout.count({ + where: { userId: TEST_USER_ID }, + }), + ).toBe(1); + }); + + it("reports one enrichment when two entries in one batch aim at the same workout", async () => { + await post({ workouts: [baseWorkout("hk-uuid-enrich-007")] }); + + // Same external id twice, far enough apart in time that the + // write-time twin dedup keeps both entries. Only one series can land. + const res = await post({ + workouts: [ + baseWorkout("hk-uuid-enrich-007", { samples: SERIES }), + baseWorkout("hk-uuid-enrich-007", { + startedAt: "2026-05-14T09:00:00.000Z", + endedAt: "2026-05-14T09:45:00.000Z", + samples: [{ t: "2026-05-14T09:00:00.000Z", hr: 155 }], + }), + ], + }); + + expect(res.status).toBe(200); + expect( + res.data.entries.filter((e) => e.status === "enriched"), + ).toHaveLength(1); + expect( + await getPrismaClient().workoutSamples.count({ + where: { workout: { userId: TEST_USER_ID } }, + }), + ).toBe(1); + }); + + it("leaves every workout column byte-identical across an enrichment", async () => { + await post({ + workouts: [ + baseWorkout("hk-uuid-enrich-003", { + totalEnergyKcal: 388, + totalDistanceM: 7100, + avgHeartRate: 141, + maxHeartRate: 169, + minHeartRate: 96, + metadata: { device: "watch" }, + }), + ], + }); + + const before = await getPrismaClient().workout.findFirstOrThrow({ + where: { userId: TEST_USER_ID, externalId: "hk-uuid-enrich-003" }, + }); + + // A re-post whose every field contradicts the stored row. First-write- + // wins still governs the workout itself; only the series is attached. + const enriched = await post({ + workouts: [ + baseWorkout("hk-uuid-enrich-003", { + sportType: "cycling", + startedAt: "2026-05-14T09:00:00.000Z", + endedAt: "2026-05-14T10:30:00.000Z", + totalEnergyKcal: 999, + totalDistanceM: 42000, + avgHeartRate: 111, + maxHeartRate: 198, + minHeartRate: 55, + metadata: { device: "phone" }, + samples: SERIES, + }), + ], + }); + expect(enriched.data.entries[0]?.status).toBe("enriched"); + + const after = await getPrismaClient().workout.findFirstOrThrow({ + where: { userId: TEST_USER_ID, externalId: "hk-uuid-enrich-003" }, + }); + // Whole-row comparison — a spot check would miss a column nobody + // thought to name, and `updatedAt` proves no write touched the row. + expect(after).toEqual(before); + }); + + it("refuses a malformed samples array per entry without failing the batch", async () => { + await post({ workouts: [baseWorkout("hk-uuid-enrich-004")] }); + + const mixed = await post({ + workouts: [ + baseWorkout("hk-uuid-enrich-004", { + samples: [{ t: "not-a-timestamp", hr: 4000 }], + }), + baseWorkout("hk-uuid-enrich-005", { + startedAt: "2026-05-15T06:30:00.000Z", + endedAt: "2026-05-15T07:15:00.000Z", + }), + ], + }); + + expect(mixed.status).toBe(200); + expect(mixed.data.entries[0]?.status).toBe("skipped"); + expect(mixed.data.entries[0]?.reason).toBe("invalid_samples"); + expect(mixed.data.entries[1]?.status).toBe("inserted"); + expect(mixed.data.inserted).toBe(1); + expect(mixed.data.skipped).toEqual([ + { index: 0, reason: "invalid_samples" }, + ]); + + // Nothing attached to the workout the bad entry addressed. + expect( + await getPrismaClient().workoutSamples.count({ + where: { workout: { userId: TEST_USER_ID } }, + }), + ).toBe(0); + // And the second entry did land. + expect( + await getPrismaClient().workout.count({ + where: { userId: TEST_USER_ID }, + }), + ).toBe(2); + }); + + it("refuses an over-cap samples array per entry rather than the whole batch", async () => { + await post({ workouts: [baseWorkout("hk-uuid-enrich-006")] }); + + const oversized = Array.from({ length: 30_001 }, (_v, i) => ({ + t: new Date(Date.UTC(2026, 4, 14, 6, 30, 0) + i * 1000).toISOString(), + hr: 120, + })); + + const res = await post({ + workouts: [baseWorkout("hk-uuid-enrich-006", { samples: oversized })], + }); + expect(res.status).toBe(200); + expect(res.data.entries[0]?.status).toBe("skipped"); + expect(res.data.entries[0]?.reason).toBe("invalid_samples"); + expect( + await getPrismaClient().workoutSamples.count({ + where: { workout: { userId: TEST_USER_ID } }, + }), + ).toBe(0); + }); + + it("reports insert, enrich, duplicate and skip side by side in one batch", async () => { + await post({ + workouts: [ + baseWorkout("hk-uuid-mix-known-series", { + startedAt: "2026-05-10T06:00:00.000Z", + endedAt: "2026-05-10T06:45:00.000Z", + }), + baseWorkout("hk-uuid-mix-known-plain", { + startedAt: "2026-05-11T06:00:00.000Z", + endedAt: "2026-05-11T06:45:00.000Z", + }), + ], + }); + + const mixed = await post({ + workouts: [ + // 0 — new workout, inserts + baseWorkout("hk-uuid-mix-new", { + startedAt: "2026-05-12T06:00:00.000Z", + endedAt: "2026-05-12T06:45:00.000Z", + }), + // 1 — known workout with a series in the payload, enriches + baseWorkout("hk-uuid-mix-known-series", { + startedAt: "2026-05-10T06:00:00.000Z", + endedAt: "2026-05-10T06:45:00.000Z", + samples: SERIES, + }), + // 2 — known workout, no series in the payload, unchanged + baseWorkout("hk-uuid-mix-known-plain", { + startedAt: "2026-05-11T06:00:00.000Z", + endedAt: "2026-05-11T06:45:00.000Z", + }), + // 3 — unusable external id, refused per entry + baseWorkout(" ", { + startedAt: "2026-05-13T06:00:00.000Z", + endedAt: "2026-05-13T06:45:00.000Z", + }), + ], + }); + + expect(mixed.status).toBe(200); + expect(mixed.data.processed).toBe(4); + expect(mixed.data.entries.map((e) => e.status)).toEqual([ + "inserted", + "enriched", + "duplicate", + "skipped", + ]); + expect(mixed.data.inserted).toBe(1); + // An enriched entry is still a duplicate of the workout row itself — + // the counters keep counting rows, the entry status carries the detail. + expect(mixed.data.duplicates).toBe(2); + expect(mixed.data.skipped).toHaveLength(1); + + const series = await getPrismaClient().workoutSamples.findMany({ + where: { workout: { userId: TEST_USER_ID } }, + include: { workout: { select: { externalId: true } } }, + }); + expect(series).toHaveLength(1); + expect(series[0]?.workout.externalId).toBe("hk-uuid-mix-known-series"); + expect( + await getPrismaClient().workout.count({ + where: { userId: TEST_USER_ID }, + }), + ).toBe(3); + }); +});