diff --git a/src/__tests__/ai-egress-capability-guard.test.ts b/src/__tests__/ai-egress-capability-guard.test.ts index a63ae2a69..2b84a83b4 100644 --- a/src/__tests__/ai-egress-capability-guard.test.ts +++ b/src/__tests__/ai-egress-capability-guard.test.ts @@ -314,8 +314,8 @@ describe("workout route geometry never reaches an AI path", () => { const ROUTE_READERS: Record = { "app/api/workouts/[id]/route.ts": "The workout detail read: serves the track to the person's own map.", - "app/api/workouts/batch/route.ts": - "Writes a kept track (opt-in) and derives its climb before storage.", + "lib/workouts/batch-ingest/write.ts": + "The workout batch ingest's write: stores a kept track (opt-in), sealed.", "lib/apple-health/route-import.ts": "Writes a kept track from the Apple Health export (opt-in).", "lib/crypto/encrypted-columns.ts": diff --git a/src/__tests__/external-id-stability-guard.test.ts b/src/__tests__/external-id-stability-guard.test.ts index 10e1a9f40..75b8eb974 100644 --- a/src/__tests__/external-id-stability-guard.test.ts +++ b/src/__tests__/external-id-stability-guard.test.ts @@ -404,7 +404,10 @@ describe("the batch surfaces refuse per entry, never whole-batch", () => { ], "app/api/mood-entries/bulk/route.ts": [], "app/api/medications/intake/bulk/route.ts": [], - "app/api/workouts/batch/route.ts": [], + "app/api/workouts/batch/route.ts": [ + "lib/workouts/batch-ingest/entries.ts", + "lib/workouts/batch-ingest/ingest.ts", + ], "app/api/cycle/day-logs/bulk/route.ts": [], "app/api/import/route.ts": [], }; diff --git a/src/__tests__/workout-sport-provenance-guard.test.ts b/src/__tests__/workout-sport-provenance-guard.test.ts index 280c7fe55..5bb08c40b 100644 --- a/src/__tests__/workout-sport-provenance-guard.test.ts +++ b/src/__tests__/workout-sport-provenance-guard.test.ts @@ -189,9 +189,10 @@ describe("T2 — the set of files writing Workout rows is frozen", () => { * already resolved, so the provenance rule does not reach them. */ const NON_MAPPING_WRITERS: Record = { - // The iOS / client batch endpoint. The client sends an enum value and MAY + // The iOS / client batch endpoint (its write, behind + // `/api/workouts/batch`). The client sends an enum value and MAY // send its own `metadata`; no server-side label mapping happens here. - "app/api/workouts/batch/route.ts": "client-supplied enum", + "lib/workouts/batch-ingest/write.ts": "client-supplied enum", // Backup restore replays rows this app previously wrote, provenance and // all. "lib/export/workouts-backup.ts": "restore replay", diff --git a/src/app/api/workouts/batch/route.ts b/src/app/api/workouts/batch/route.ts index cc07172c3..ba76b39a7 100644 --- a/src/app/api/workouts/batch/route.ts +++ b/src/app/api/workouts/batch/route.ts @@ -50,7 +50,7 @@ * 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. + * as often as it likes. See `src/lib/workouts/batch-ingest/plan.ts`. * * Race reconciliation uses PostgreSQL's `INSERT ... RETURNING` through * Prisma's `createManyAndReturn`. The returned composite keys identify the @@ -66,9 +66,7 @@ import { NextRequest } from "next/server"; import { prisma } from "@/lib/db"; import { apiHandler, isScopedCredential, requireAuth } from "@/lib/api-handler"; -import { routeClimbM } from "@/lib/workouts/route-climb"; import { annotate } from "@/lib/logging/context"; -import { auditLog } from "@/lib/auth/audit"; import { apiError, apiSuccess, @@ -77,33 +75,10 @@ import { } from "@/lib/api-response"; import { withIdempotency } from "@/lib/idempotency"; import { checkRateLimit } from "@/lib/rate-limit"; -import { enqueuePrDetection } from "@/lib/jobs/pr-detection"; -import { lockPersonalRecordsForUser } from "@/lib/personal-records/pr-detection-worker"; -import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; -import { emitDataArrival } from "@/lib/arrivals/emit-shared"; import { MAX_WORKOUTS_PER_BATCH } from "@/lib/validations/workout"; -import { - INVALID_SAMPLES_REASON, - parseWorkoutBatch, -} from "@/lib/workouts/batch-ingest-parse"; -import { - classifyExternalId, - unstableExternalIdMeta, - UNSTABLE_EXTERNAL_ID_REASON, - type UnstableExternalIdShape, -} from "@/lib/validations/external-id"; -import { batchWorkoutIndicesToKeep } from "@/lib/workouts/batch-dedup"; +import { parseWorkoutBatch } from "@/lib/workouts/batch-ingest-parse"; +import { ingestWorkoutBatch } from "@/lib/workouts/batch-ingest/ingest"; import { WORKOUTS_WRITE_SCOPE } from "@/lib/workouts/scopes"; -import { EXTERNAL_SOURCE } from "@/lib/measurements/external-source"; -import { encryptRouteGeometry } from "@/lib/workouts/route-geometry-cipher"; -import { Prisma } from "@/generated/prisma/client"; - -// v1.4.25 — push-suppression threshold for workout PRs. A batch -// larger than this fires the detection job with `silent: true` so a -// multi-year HKWorkout backfill doesn't spam the user with hundreds of -// pushes. Matches the measurements-batch threshold so behaviour stays -// uniform across both ingest paths. -const PR_DETECTION_SILENT_THRESHOLD = 50; // v1.4.25 (security H-2 parity): cap batch ingest at 60 // batches per user per minute. Healthy iOS sync drains its HealthKit @@ -123,115 +98,6 @@ const BATCH_RATE_LIMIT_WINDOW_MS = 60 * 1000; // the correct response to a misbehaving client or an attempted DoS. const MAX_BODY_BYTES = 5 * 1024 * 1024; -/** - * Per-entry outcome the iOS client uses to advance its sync cursor. - * `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. - * `updated` says a hand-entered workout (`manual:` id, source MANUAL) was - * sent again with different values and the stored row now carries them. - */ -type EntryStatus = - "inserted" | "duplicate" | "enriched" | "updated" | "skipped"; - -/** The workout columns a hand-entered workout's resubmit may overwrite. */ -const MANUAL_OVERWRITE_FIELDS = [ - "sportType", - "startedAt", - "endedAt", - "durationSec", - "totalEnergyKcal", - "totalDistanceM", - "avgHeartRate", - "maxHeartRate", - "minHeartRate", - "stepCount", - "elevationM", - "pauseDurationSec", -] as const; -type ManualOverwriteField = (typeof MANUAL_OVERWRITE_FIELDS)[number]; -type ManualOverwriteValues = { - [K in ManualOverwriteField]: string | number | Date | null; -}; - -/** - * A workout entered by hand carries one id per form opening, and the form - * resends it with the edited values when the person corrects a submit. Such a - * resubmit is an edit of the stored row, the way a `stats:` measurement - * re-post is. Every other id stays first-write-wins. - */ -function isManualOverwrite(source: unknown, externalId: string | null) { - return ( - source === "MANUAL" && - externalId !== null && - externalId.startsWith("manual:") - ); -} - -/** The changed columns of a manual resubmit, or null when nothing differs. */ -function manualOverwriteChanges( - stored: ManualOverwriteValues, - sent: Prisma.WorkoutCreateManyInput, -): Partial | null { - const changes: Partial = {}; - for (const field of MANUAL_OVERWRITE_FIELDS) { - const next = (sent[field] ?? null) as string | number | Date | null; - const prev = stored[field]; - const same = - next instanceof Date && prev instanceof Date - ? next.getTime() === prev.getTime() - : next === prev; - if (!same) changes[field] = next; - } - return Object.keys(changes).length > 0 ? changes : null; -} -interface EntryResult { - index: number; - status: EntryStatus; - reason?: string; -} - -/** - * v1.31.1 — emit one data arrival per workout this batch actually inserted, - * and per hand-entered workout whose edit it actually wrote. - * - * `createManyAndReturn` gives the transaction the exact inserted ids. The - * index-keyed map avoids a second lookup and keeps equal external ids from - * different sources attached to their own workout. - * - * Fully best-effort — every failure path returns quietly. A reaction is never - * worth failing an ingest that already succeeded. - */ -async function emitWorkoutArrivals( - userId: string, - prepared: ReadonlyArray<{ - index: number; - row: Prisma.WorkoutCreateManyInput; - }>, - insertedIdByIndex: ReadonlyMap, -): Promise { - for (const p of prepared) { - const refId = insertedIdByIndex.get(p.index); - if (!refId) continue; - const startedAt = - p.row.startedAt instanceof Date - ? p.row.startedAt - : new Date(p.row.startedAt as string); - if (Number.isNaN(startedAt.getTime())) continue; - await emitDataArrival({ - userId, - kind: "workout", - newestSampleAt: startedAt, - insertedCount: 1, - refId, - source: "batch", - }); - } -} - export const POST = apiHandler(withIdempotency<[NextRequest]>(postBatch)); async function postBatch(request: NextRequest): Promise { @@ -382,566 +248,17 @@ async function postBatch(request: NextRequest): Promise { } 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)` — - // entries without an `externalId` cannot collide on the key (NULL- - // distinct semantics), so they always insert as new rows. - type Prepared = { - index: number; - row: Prisma.WorkoutCreateManyInput; - route: { - geometry: unknown; - sampleTimestamps: unknown; - } | null; - /** Route-independent per-workout HR series. NULL when the entry - * ships no `samples` array. Persisted to the `WorkoutSamples` - * child by FK after the workout row lands. */ - samples: { - samples: unknown; - sampleCount: number; - } | null; - /** Composite-key probe for the pre-flight findMany. NULL when - * the entry has no externalId — those rows always pass through - * as "inserted" because the NULL-distinct unique index can never - * collide them. */ - dedupKey: { source: string; externalId: string } | null; - }; - - 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) - // never matches its own earlier row, so every re-sync mints another - // workout. Refused per entry — one bad row must not stop the rest of - // the batch landing. - const unstable = - w.externalId === undefined ? null : classifyExternalId(w.externalId); - if (unstable) { - unstableShapes.push(unstable); - results[index] = { - index, - status: "skipped", - reason: UNSTABLE_EXTERNAL_ID_REASON, - }; - return []; - } - - const durationSec = Math.max( - 0, - Math.round((w.endedAt.getTime() - w.startedAt.getTime()) / 1000), - ); - const route = - w.route && keepRoutes - ? { - geometry: w.route.geometry as unknown, - sampleTimestamps: (w.route.sampleTimestamps ?? null) as unknown, - } - : null; - const samples = w.samples - ? { - samples: w.samples as unknown, - sampleCount: w.samples.length, - } - : null; - const row: Prisma.WorkoutCreateManyInput = { - userId: user.id, - sportType: w.sportType, - startedAt: w.startedAt, - endedAt: w.endedAt, - durationSec, - totalEnergyKcal: w.totalEnergyKcal ?? null, - totalDistanceM: w.totalDistanceM ?? null, - avgHeartRate: w.avgHeartRate ?? null, - maxHeartRate: w.maxHeartRate ?? null, - minHeartRate: w.minHeartRate ?? null, - stepCount: w.stepCount ?? null, - // A kept track fills the climb when the source sent none, so the figure - // lives on the workout row and no reader has to open the track for it. - elevationM: w.elevationM ?? (route ? routeClimbM(route.geometry) : null), - pauseDurationSec: w.pauseDurationSec ?? null, - // `EXTERNAL` under a scoped credential, and nothing else: the guard - // above refused any body that named a source, so nothing asserted is - // discarded here. It also moves the dedup key into its own namespace, - // so a bridge's ids can never collide with the phone's. - source: scoped ? EXTERNAL_SOURCE : w.source, - externalId: w.externalId ?? null, - externalSourceVersion: w.externalSourceVersion ?? null, - metadata: w.metadata - ? (w.metadata as Prisma.InputJsonValue) - : Prisma.JsonNull, - }; - return [ - { - index, - row, - route, - samples, - dedupKey: w.externalId - ? { - source: scoped ? EXTERNAL_SOURCE : w.source, - externalId: w.externalId, - } - : null, - }, - ]; - }); - - if (unstableShapes.length > 0) { - annotate({ - meta: unstableExternalIdMeta("workout.batch", unstableShapes), - }); - } - - // Write-time cross-source dedup through the one workout clustering the - // reads use (see `batchWorkoutIndicesToKeep`): one logical workout from two - // sources in one batch, or one HealthKit session re-sent under a new id, - // lands once. Rows hidden here surface to the client as `duplicate`, so the - // sync cursor advances past them like the externalId case below. - const survivingIndices = await batchWorkoutIndicesToKeep( - user.id, - prepared, - userPriority?.sourcePriorityJson ?? null, - ); - const droppedByWriteDedup: number[] = []; - for (const p of prepared) { - if (!survivingIndices.has(p.index)) { - results[p.index] = { index: p.index, status: "duplicate" }; - droppedByWriteDedup.push(p.index); - } - } - const survivors = prepared.filter((p) => survivingIndices.has(p.index)); - let duplicateCount = droppedByWriteDedup.length; - let insertedCount = 0; - let enrichedCount = 0; - let updatedCount = 0; - const insertedIdByIndex = new Map(); - /** Workout ids of the manual edits this request actually wrote. */ - const overwrittenIdByIndex = new Map(); - - if (survivors.length > 0) { - // v1.4.42 — only the write-time dedup survivors are probed - // against existing rows; the dropped twins already carry a - // `duplicate` status and never reach the DB. - const dedupCandidates = survivors - .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: { - userId: user.id, - OR: dedupCandidates.map((k) => ({ - source: k.source as Prisma.WorkoutCreateManyInput["source"], - externalId: k.externalId, - })), - }, - select: { - id: true, - source: true, - externalId: true, - samples: { select: { workoutId: true } }, - sportType: true, - startedAt: true, - endedAt: true, - durationSec: true, - totalEnergyKcal: true, - totalDistanceM: true, - avgHeartRate: true, - maxHeartRate: true, - minHeartRate: true, - stepCount: true, - elevationM: true, - pauseDurationSec: true, - }, - }) - : []; - - const existingByKey = new Map< - string, - { id: string; hasSeries: boolean; values: ManualOverwriteValues } - >(); - for (const row of existing) { - if (row.externalId === null) continue; - existingByKey.set(`${row.source}::${row.externalId}`, { - id: row.id, - hasSeries: row.samples !== null, - values: { - sportType: row.sportType, - startedAt: row.startedAt, - endedAt: row.endedAt, - durationSec: row.durationSec, - totalEnergyKcal: row.totalEnergyKcal, - totalDistanceM: row.totalDistanceM, - avgHeartRate: row.avgHeartRate, - maxHeartRate: row.maxHeartRate, - minHeartRate: row.minHeartRate, - stepCount: row.stepCount, - elevationM: row.elevationM, - pauseDurationSec: row.pauseDurationSec, - }, - }); - } - - const toInsert: Prepared[] = []; - /** Stored workouts this batch hands a heart-rate series to. */ - const toEnrich: Array<{ - index: number; - workoutId: string; - samples: NonNullable; - }> = []; - /** Hand-entered workouts this batch edits in place. */ - const toOverwrite: Array<{ - index: number; - workoutId: string; - externalId: string; - changes: Partial; - }> = []; - /** 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; - const known = keyTuple !== null ? existingByKey.get(keyTuple) : undefined; - if ( - known && - p.dedupKey && - isManualOverwrite(p.dedupKey.source, p.dedupKey.externalId) - ) { - const changes = enrichClaimed.has(known.id) - ? null - : manualOverwriteChanges(known.values, p.row); - if (changes) { - enrichClaimed.add(known.id); - toOverwrite.push({ - index: p.index, - workoutId: known.id, - externalId: p.dedupKey.externalId, - changes, - }); - } else { - results[p.index] = { index: p.index, status: "duplicate" }; - duplicateCount += 1; - } - } else 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 || toEnrich.length > 0 || toOverwrite.length > 0) { - const CHUNK = 100; - /** Workout ids whose series row this request actually wrote. */ - const seriesLanded = new Set(); - /** Batch indices whose manual edit this request actually wrote. */ - const overwritten = new Set(); - - await prisma.$transaction(async (tx) => { - // A hand-entered workout sent again with other values. Columns are - // copied one by one from the validated entry; the records the old - // values set go with them and the detection pass below re-derives - // the honest best, as the DELETE does. - if (toOverwrite.length > 0) { - await lockPersonalRecordsForUser(tx, user.id); - } - for (const o of toOverwrite) { - const data: Prisma.WorkoutUpdateManyMutationInput = {}; - for (const field of MANUAL_OVERWRITE_FIELDS) { - if (field in o.changes) { - (data as Record)[field] = o.changes[field]; - } - } - const written = await tx.workout.updateMany({ - where: { id: o.workoutId, userId: user.id, source: "MANUAL" }, - data, - }); - if (written.count === 1) { - overwritten.add(o.index); - await tx.personalRecord.deleteMany({ - where: { - userId: user.id, - source: "MANUAL", - metricSlot: { not: null }, - externalId: o.externalId, - }, - }); - // The paragraph described the old values (a 12 km run corrected - // to 1.2 km must not keep its 12 km story). It goes with them, - // and the claim is let go of its workout the way a delete does - // (`ON DELETE SET NULL`): the row stays as the day's ledger, so - // the edit buys no slot back, and the arrival emitted below can - // claim a fresh one through the ordinary path, under the same cap. - await tx.workoutInsight.deleteMany({ - where: { userId: user.id, workoutId: o.workoutId }, - }); - await tx.workoutInsightGenerationClaim.updateMany({ - where: { userId: user.id, workoutId: o.workoutId }, - data: { workoutId: null }, - }); - } - } - - const withExternalId = toInsert.filter((p) => p.dedupKey !== null); - const withoutExternalId = toInsert.filter((p) => p.dedupKey === null); - - // PostgreSQL RETURNING is the only reliable way to identify the exact - // winner when another request races the same unique key. - for (let i = 0; i < withExternalId.length; i += CHUNK) { - const slice = withExternalId.slice(i, i + CHUNK); - const pendingByKey = new Map(); - for (const p of slice) { - const key = `${p.dedupKey!.source}::${p.dedupKey!.externalId}`; - const pending = pendingByKey.get(key); - if (pending) pending.push(p); - else pendingByKey.set(key, [p]); - } - - const inserted = await tx.workout.createManyAndReturn({ - data: slice.map((p) => p.row), - skipDuplicates: true, - select: { id: true, source: true, externalId: true }, - }); - for (const row of inserted) { - if (row.externalId === null) continue; - const key = `${row.source}::${row.externalId}`; - const preparedRow = pendingByKey.get(key)?.shift(); - if (preparedRow) insertedIdByIndex.set(preparedRow.index, row.id); - } - } - - // NULL external ids are intentionally distinct in PostgreSQL. Insert - // them one at a time so a route/HR series can never attach to a - // different otherwise-identical manual workout. - for (const p of withoutExternalId) { - const inserted = await tx.workout.createManyAndReturn({ - data: p.row, - select: { id: true }, - }); - const row = inserted[0]; - if (row) insertedIdByIndex.set(p.index, row.id); - } - - const routesToInsert: Prisma.WorkoutRouteCreateManyInput[] = []; - const samplesToInsert: Prisma.WorkoutSamplesCreateManyInput[] = []; - for (const p of toInsert) { - const id = insertedIdByIndex.get(p.index); - if (!id) continue; - if (p.route) { - routesToInsert.push({ - workoutId: id, - // v1.39.4 — the GPS track is stored sealed only; the readable - // `geometry` column stays NULL. - geometryEncrypted: encryptRouteGeometry(p.route.geometry), - sampleTimestamps: - p.route.sampleTimestamps === null - ? Prisma.JsonNull - : (p.route.sampleTimestamps as Prisma.InputJsonValue), - }); - } - if (p.samples) { - samplesToInsert.push({ - workoutId: id, - samples: p.samples.samples as Prisma.InputJsonValue, - sampleCount: p.samples.sampleCount, - }); - } - } - - // 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), - skipDuplicates: true, - }); - } - for (let i = 0; i < samplesToInsert.length; i += CHUNK) { - // 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); - } - }); - - insertedCount = insertedIdByIndex.size; - for (const p of toInsert) { - if (insertedIdByIndex.has(p.index)) { - results[p.index] = { index: p.index, status: "inserted" }; - } else { - results[p.index] = { index: p.index, status: "duplicate" }; - 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; - } - - // Like an enrichment, an edit inserts no row, so it counts towards - // `duplicates`; the per-entry status says the stored row changed. - for (const o of toOverwrite) { - const landed = overwritten.has(o.index); - results[o.index] = { - index: o.index, - status: landed ? "updated" : "duplicate", - }; - if (landed) { - updatedCount += 1; - overwrittenIdByIndex.set(o.index, o.workoutId); - } - duplicateCount += 1; - } - } - } - - const skipped = results - .filter((r) => r.status === "skipped") - .map((r) => ({ index: r.index, reason: r.reason ?? "unknown" })); - - await auditLog("workout.batch.ingest", { + const result = await ingestWorkoutBatch({ userId: user.id, + scoped, + keepRoutes, + sourcePriorityJson: userPriority?.sourcePriorityJson ?? null, + workouts, + refusedSampleIndices: new Set(parsed.refusedSampleIndices), ipAddress: getClientIp(request), - details: { - processed: workouts.length, - inserted: insertedCount, - duplicates: duplicateCount, - enriched: enrichedCount, - updated: updatedCount, - skipped: skipped.length, - }, }); - // v1.4.25 — kick off PR detection for this user. Suppress push - // notifications for historical-backfill batches so a multi-year - // HKWorkout import doesn't fire hundreds of pushes during initial - // sync. The detector also scans Measurement history on the same - // pass, so a workout-only batch still surfaces measurement-side PRs - // the user may have racked up since the last detection. - if (insertedCount > 0 || duplicateCount > 0) { - const silent = workouts.length > PR_DETECTION_SILENT_THRESHOLD; - try { - await enqueuePrDetection(user.id, { silent }); - // Telemetry, not an audit event: it rides the wide event. - annotate({ - meta: { pr_detection_enqueued: true, pr_detection_silent: silent }, - }); - } catch (err) { - annotate({ - action: { name: "personal_records.detection_enqueue_failed" }, - meta: { error: err instanceof Error ? err.message : String(err) }, - }); - } - } - - // 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, - updated: updatedCount, - skipped: skipped.length, - // Tracks the body carried but the opt-in told us not to keep. - routes_dropped: keepRoutes - ? 0 - : workouts.filter((w) => w.route != null).length, - }, - }); - - // v1.4.34 — bust per-user analytics + achievements + workouts - // caches when at least one row landed. Workouts ride on the - // measurements bucket because achievements / analytics also touch - // workout-derived metrics. - // 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 || updatedCount > 0) { - invalidateUserMeasurements(user.id); - } - - if (insertedCount > 0 || overwrittenIdByIndex.size > 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. - // A hand-entered workout whose edit landed raises one again, so its - // paragraph (dropped with the old values) is written from the new ones. - void emitWorkoutArrivals( - user.id, - prepared, - new Map([...insertedIdByIndex, ...overwrittenIdByIndex]), - ).catch(() => {}); - } - - return apiSuccess({ - processed: workouts.length, - inserted: insertedCount, - duplicates: duplicateCount, - skipped, - entries: results, - }); + return apiSuccess(result); } /** Whether any entry of a raw batch body names a `source` at all. */ diff --git a/src/lib/validations/workout.ts b/src/lib/validations/workout.ts index 3174f1d10..12f9a5661 100644 --- a/src/lib/validations/workout.ts +++ b/src/lib/validations/workout.ts @@ -178,7 +178,7 @@ export const createWorkoutSchema = z // The only writer of this schema is the batch ingest, which enforces // external-id stability per entry rather than at the field: a // schema-level refusal would fail the whole batch on one bad row. - // @external-id-checked-per-entry: src/app/api/workouts/batch/route.ts + // @external-id-checked-per-entry: src/lib/workouts/batch-ingest/entries.ts externalId: z.string().max(128).optional(), externalSourceVersion: z.string().max(64).optional(), metadata: z.record(z.string(), z.unknown()).optional(), diff --git a/src/lib/workouts/batch-ingest/entries.ts b/src/lib/workouts/batch-ingest/entries.ts new file mode 100644 index 000000000..de392b9c0 --- /dev/null +++ b/src/lib/workouts/batch-ingest/entries.ts @@ -0,0 +1,180 @@ +/** + * The first pass of the workout batch ingest (`POST /api/workouts/batch`): + * what the ingest answers per entry, and the row, track and series each + * accepted entry becomes. + */ +import { Prisma } from "@/generated/prisma/client"; +import { EXTERNAL_SOURCE } from "@/lib/measurements/external-source"; +import type { CreateBatchWorkoutInput } from "@/lib/validations/workout"; +import { + classifyExternalId, + UNSTABLE_EXTERNAL_ID_REASON, + type UnstableExternalIdShape, +} from "@/lib/validations/external-id"; +import { routeClimbM } from "@/lib/workouts/route-climb"; +import { INVALID_SAMPLES_REASON } from "@/lib/workouts/batch-ingest-parse"; + +export type WorkoutEntry = CreateBatchWorkoutInput["workouts"][number]; + +/** + * Per-entry outcome the iOS client uses to advance its sync cursor. + * `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. + * `updated` says a hand-entered workout (`manual:` id, source MANUAL) was + * sent again with different values and the stored row now carries them. + */ +export type EntryStatus = + "inserted" | "duplicate" | "enriched" | "updated" | "skipped"; + +export interface EntryResult { + index: number; + status: EntryStatus; + reason?: string; +} + +/** An accepted entry, ready for the duplicate probe and the write. */ +export type PreparedWorkout = { + index: number; + row: Prisma.WorkoutCreateManyInput; + route: { + geometry: unknown; + sampleTimestamps: unknown; + } | null; + /** Route-independent per-workout HR series. NULL when the entry + * ships no `samples` array. Persisted to the `WorkoutSamples` + * child by FK after the workout row lands. */ + samples: { + samples: unknown; + sampleCount: number; + } | null; + /** Composite-key probe for the pre-flight findMany. NULL when + * the entry has no externalId — those rows always pass through + * as "inserted" because the NULL-distinct unique index can never + * collide them. */ + dedupKey: { source: string; externalId: string } | null; +}; + +/** + * Pre-flight duplicate detection needs per-entry status. The composite + * unique index is `(userId, source, externalId)` — entries without an + * `externalId` cannot collide on the key (NULL-distinct semantics), so they + * always insert as new rows. + */ +export function prepareWorkouts(args: { + userId: string; + scoped: boolean; + keepRoutes: boolean; + workouts: WorkoutEntry[]; + refusedSampleIndices: ReadonlySet; +}): { + prepared: PreparedWorkout[]; + results: EntryResult[]; + unstableShapes: UnstableExternalIdShape[]; +} { + const { userId, scoped, keepRoutes, workouts, refusedSampleIndices } = args; + const results: EntryResult[] = new Array(workouts.length); + const unstableShapes: UnstableExternalIdShape[] = []; + const prepared: PreparedWorkout[] = 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) + // never matches its own earlier row, so every re-sync mints another + // workout. Refused per entry — one bad row must not stop the rest of + // the batch landing. + const unstable = + w.externalId === undefined ? null : classifyExternalId(w.externalId); + if (unstable) { + unstableShapes.push(unstable); + results[index] = { + index, + status: "skipped", + reason: UNSTABLE_EXTERNAL_ID_REASON, + }; + return []; + } + + return [preparedFor(userId, scoped, keepRoutes, w, index)]; + }); + return { prepared, results, unstableShapes }; +} + +function preparedFor( + userId: string, + scoped: boolean, + keepRoutes: boolean, + w: WorkoutEntry, + index: number, +): PreparedWorkout { + const durationSec = Math.max( + 0, + Math.round((w.endedAt.getTime() - w.startedAt.getTime()) / 1000), + ); + const route = + w.route && keepRoutes + ? { + geometry: w.route.geometry as unknown, + sampleTimestamps: (w.route.sampleTimestamps ?? null) as unknown, + } + : null; + const samples = w.samples + ? { + samples: w.samples as unknown, + sampleCount: w.samples.length, + } + : null; + const row: Prisma.WorkoutCreateManyInput = { + userId, + sportType: w.sportType, + startedAt: w.startedAt, + endedAt: w.endedAt, + durationSec, + totalEnergyKcal: w.totalEnergyKcal ?? null, + totalDistanceM: w.totalDistanceM ?? null, + avgHeartRate: w.avgHeartRate ?? null, + maxHeartRate: w.maxHeartRate ?? null, + minHeartRate: w.minHeartRate ?? null, + stepCount: w.stepCount ?? null, + // A kept track fills the climb when the source sent none, so the figure + // lives on the workout row and no reader has to open the track for it. + elevationM: w.elevationM ?? (route ? routeClimbM(route.geometry) : null), + pauseDurationSec: w.pauseDurationSec ?? null, + // `EXTERNAL` under a scoped credential, and nothing else: the route + // refused any body that named a source, so nothing asserted is + // discarded here. It also moves the dedup key into its own namespace, + // so a bridge's ids can never collide with the phone's. + source: scoped ? EXTERNAL_SOURCE : w.source, + externalId: w.externalId ?? null, + externalSourceVersion: w.externalSourceVersion ?? null, + metadata: w.metadata + ? (w.metadata as Prisma.InputJsonValue) + : Prisma.JsonNull, + }; + return { + index, + row, + route, + samples, + dedupKey: w.externalId + ? { + source: scoped ? EXTERNAL_SOURCE : w.source, + externalId: w.externalId, + } + : null, + }; +} diff --git a/src/lib/workouts/batch-ingest/ingest.ts b/src/lib/workouts/batch-ingest/ingest.ts new file mode 100644 index 000000000..f1073ae7c --- /dev/null +++ b/src/lib/workouts/batch-ingest/ingest.ts @@ -0,0 +1,227 @@ +/** + * The workout batch ingest behind `POST /api/workouts/batch`, from a parsed + * and admitted batch to the per-entry answer: the first pass, the write-time + * cross-source dedup, the duplicate probe and plan, the write, then the + * audit row, the record detection, the wide event, the cache and the + * arrivals. The route owns authentication, the rate limit, the body caps and + * the response envelope. + */ +import type { Prisma } from "@/generated/prisma/client"; +import { annotate } from "@/lib/logging/context"; +import { auditLog } from "@/lib/auth/audit"; +import { enqueuePrDetection } from "@/lib/jobs/pr-detection"; +import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; +import { emitDataArrival } from "@/lib/arrivals/emit-shared"; +import { unstableExternalIdMeta } from "@/lib/validations/external-id"; +import { batchWorkoutIndicesToKeep } from "@/lib/workouts/batch-dedup"; + +import { + prepareWorkouts, + type EntryResult, + type WorkoutEntry, +} from "./entries"; +import { planWorkoutWrites } from "./plan"; +import { writeWorkoutPlan, type WriteOutcome } from "./write"; + +// Push-suppression threshold for workout PRs. A batch larger than this fires +// the detection job with `silent: true` so a multi-year HKWorkout backfill +// doesn't spam the user with hundreds of pushes. Matches the +// measurements-batch threshold so behaviour stays uniform across both ingest +// paths. +const PR_DETECTION_SILENT_THRESHOLD = 50; + +export interface WorkoutBatchInput { + userId: string; + scoped: boolean; + /** The account's opt-in: a GPS track is kept only when it is on. */ + keepRoutes: boolean; + /** `User.sourcePriorityJson`, the ladder the write-time picker walks. */ + sourcePriorityJson: unknown; + workouts: WorkoutEntry[]; + refusedSampleIndices: ReadonlySet; + ipAddress: string | null; +} + +/** The response body of the workout batch route. */ +export interface WorkoutBatchResult { + processed: number; + inserted: number; + duplicates: number; + skipped: Array<{ index: number; reason: string }>; + entries: EntryResult[]; +} + +const NOTHING_WRITTEN: WriteOutcome = { + insertedCount: 0, + enrichedCount: 0, + updatedCount: 0, + duplicates: 0, + insertedIdByIndex: new Map(), + overwrittenIdByIndex: new Map(), +}; + +export async function ingestWorkoutBatch( + input: WorkoutBatchInput, +): Promise { + const { userId, workouts, keepRoutes } = input; + const { prepared, results, unstableShapes } = prepareWorkouts(input); + + if (unstableShapes.length > 0) { + annotate({ + meta: unstableExternalIdMeta("workout.batch", unstableShapes), + }); + } + + // Write-time cross-source dedup through the one workout clustering the + // reads use (see `batchWorkoutIndicesToKeep`): one logical workout from two + // sources in one batch, or one HealthKit session re-sent under a new id, + // lands once. Rows hidden here surface to the client as `duplicate`, so the + // sync cursor advances past them like the externalId case below. + const survivingIndices = await batchWorkoutIndicesToKeep( + userId, + prepared, + input.sourcePriorityJson ?? null, + ); + let duplicateCount = 0; + for (const p of prepared) { + if (!survivingIndices.has(p.index)) { + results[p.index] = { index: p.index, status: "duplicate" }; + duplicateCount += 1; + } + } + const survivors = prepared.filter((p) => survivingIndices.has(p.index)); + + let written = NOTHING_WRITTEN; + if (survivors.length > 0) { + const plan = await planWorkoutWrites(userId, survivors, results); + duplicateCount += plan.duplicates; + written = await writeWorkoutPlan(userId, plan, results); + duplicateCount += written.duplicates; + } + const { insertedCount, enrichedCount, updatedCount } = written; + + const skipped = results + .filter((r) => r.status === "skipped") + .map((r) => ({ index: r.index, reason: r.reason ?? "unknown" })); + + await auditLog("workout.batch.ingest", { + userId, + ipAddress: input.ipAddress, + details: { + processed: workouts.length, + inserted: insertedCount, + duplicates: duplicateCount, + enriched: enrichedCount, + updated: updatedCount, + skipped: skipped.length, + }, + }); + + // Kick off PR detection for this user. Suppress push notifications for + // historical-backfill batches so a multi-year HKWorkout import doesn't + // fire hundreds of pushes during initial sync. The detector also scans + // Measurement history on the same pass, so a workout-only batch still + // surfaces measurement-side PRs the user may have racked up since the + // last detection. + if (insertedCount > 0 || duplicateCount > 0) { + const silent = workouts.length > PR_DETECTION_SILENT_THRESHOLD; + try { + await enqueuePrDetection(userId, { silent }); + // Telemetry, not an audit event: it rides the wide event. + annotate({ + meta: { pr_detection_enqueued: true, pr_detection_silent: silent }, + }); + } catch (err) { + annotate({ + action: { name: "personal_records.detection_enqueue_failed" }, + meta: { error: err instanceof Error ? err.message : String(err) }, + }); + } + } + + // 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, + updated: updatedCount, + skipped: skipped.length, + // Tracks the body carried but the opt-in told us not to keep. + routes_dropped: keepRoutes + ? 0 + : workouts.filter((w) => w.route != null).length, + }, + }); + + // Bust per-user analytics + achievements + workouts caches when at least + // one row landed. Workouts ride on the measurements bucket because + // achievements / analytics also touch workout-derived metrics. 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 || updatedCount > 0) { + invalidateUserMeasurements(userId); + } + + if (insertedCount > 0 || written.overwrittenIdByIndex.size > 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. + // A hand-entered workout whose edit landed raises one again, so its + // paragraph (dropped with the old values) is written from the new ones. + void emitWorkoutArrivals( + userId, + prepared, + new Map([...written.insertedIdByIndex, ...written.overwrittenIdByIndex]), + ).catch(() => {}); + } + + return { + processed: workouts.length, + inserted: insertedCount, + duplicates: duplicateCount, + skipped, + entries: results, + }; +} + +/** + * Emit one data arrival per workout this batch actually inserted, and per + * hand-entered workout whose edit it actually wrote. + * + * `createManyAndReturn` gives the transaction the exact inserted ids. The + * index-keyed map avoids a second lookup and keeps equal external ids from + * different sources attached to their own workout. + * + * Fully best-effort — every failure path returns quietly. A reaction is never + * worth failing an ingest that already succeeded. + */ +async function emitWorkoutArrivals( + userId: string, + prepared: ReadonlyArray<{ + index: number; + row: Prisma.WorkoutCreateManyInput; + }>, + insertedIdByIndex: ReadonlyMap, +): Promise { + for (const p of prepared) { + const refId = insertedIdByIndex.get(p.index); + if (!refId) continue; + const startedAt = + p.row.startedAt instanceof Date + ? p.row.startedAt + : new Date(p.row.startedAt as string); + if (Number.isNaN(startedAt.getTime())) continue; + await emitDataArrival({ + userId, + kind: "workout", + newestSampleAt: startedAt, + insertedCount: 1, + refId, + source: "batch", + }); + } +} diff --git a/src/lib/workouts/batch-ingest/plan.ts b/src/lib/workouts/batch-ingest/plan.ts new file mode 100644 index 000000000..0092c2ae8 --- /dev/null +++ b/src/lib/workouts/batch-ingest/plan.ts @@ -0,0 +1,222 @@ +/** + * What each surviving entry of a workout batch does to the store: insert a + * new workout, hand a stored one its heart-rate series, correct a + * hand-entered one in place, or nothing (a duplicate). + */ +import { prisma } from "@/lib/db"; +import type { Prisma } from "@/generated/prisma/client"; + +import type { EntryResult, PreparedWorkout } from "./entries"; + +/** The workout columns a hand-entered workout's resubmit may overwrite. */ +export const MANUAL_OVERWRITE_FIELDS = [ + "sportType", + "startedAt", + "endedAt", + "durationSec", + "totalEnergyKcal", + "totalDistanceM", + "avgHeartRate", + "maxHeartRate", + "minHeartRate", + "stepCount", + "elevationM", + "pauseDurationSec", +] as const; +type ManualOverwriteField = (typeof MANUAL_OVERWRITE_FIELDS)[number]; +export type ManualOverwriteValues = { + [K in ManualOverwriteField]: string | number | Date | null; +}; + +/** + * A workout entered by hand carries one id per form opening, and the form + * resends it with the edited values when the person corrects a submit. Such a + * resubmit is an edit of the stored row, the way a `stats:` measurement + * re-post is. Every other id stays first-write-wins. + */ +function isManualOverwrite(source: unknown, externalId: string | null) { + return ( + source === "MANUAL" && + externalId !== null && + externalId.startsWith("manual:") + ); +} + +/** The changed columns of a manual resubmit, or null when nothing differs. */ +function manualOverwriteChanges( + stored: ManualOverwriteValues, + sent: Prisma.WorkoutCreateManyInput, +): Partial | null { + const changes: Partial = {}; + for (const field of MANUAL_OVERWRITE_FIELDS) { + const next = (sent[field] ?? null) as string | number | Date | null; + const prev = stored[field]; + const same = + next instanceof Date && prev instanceof Date + ? next.getTime() === prev.getTime() + : next === prev; + if (!same) changes[field] = next; + } + return Object.keys(changes).length > 0 ? changes : null; +} + +export interface WritePlan { + toInsert: PreparedWorkout[]; + /** Stored workouts this batch hands a heart-rate series to. */ + toEnrich: Array<{ + index: number; + workoutId: string; + samples: NonNullable; + }>; + /** Hand-entered workouts this batch edits in place. */ + toOverwrite: Array<{ + index: number; + workoutId: string; + externalId: string; + changes: Partial; + }>; + /** Survivors already settled as duplicates by the probe. */ + duplicates: number; +} + +type StoredWorkout = { + id: string; + hasSeries: boolean; + values: ManualOverwriteValues; +}; + +/** + * Only the write-time dedup survivors are probed against existing rows; the + * dropped twins already carry a `duplicate` status and never reach the DB. + * + * 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. + */ +async function readStoredWorkouts( + userId: string, + survivors: PreparedWorkout[], +): Promise> { + const dedupCandidates = survivors + .map((p) => p.dedupKey) + .filter((k): k is { source: string; externalId: string } => k !== null); + + const existing = dedupCandidates.length + ? await prisma.workout.findMany({ + where: { + userId, + OR: dedupCandidates.map((k) => ({ + source: k.source as Prisma.WorkoutCreateManyInput["source"], + externalId: k.externalId, + })), + }, + select: { + id: true, + source: true, + externalId: true, + samples: { select: { workoutId: true } }, + sportType: true, + startedAt: true, + endedAt: true, + durationSec: true, + totalEnergyKcal: true, + totalDistanceM: true, + avgHeartRate: true, + maxHeartRate: true, + minHeartRate: true, + stepCount: true, + elevationM: true, + pauseDurationSec: true, + }, + }) + : []; + + 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, + values: { + sportType: row.sportType, + startedAt: row.startedAt, + endedAt: row.endedAt, + durationSec: row.durationSec, + totalEnergyKcal: row.totalEnergyKcal, + totalDistanceM: row.totalDistanceM, + avgHeartRate: row.avgHeartRate, + maxHeartRate: row.maxHeartRate, + minHeartRate: row.minHeartRate, + stepCount: row.stepCount, + elevationM: row.elevationM, + pauseDurationSec: row.pauseDurationSec, + }, + }); + } + return existingByKey; +} + +/** Decide every survivor's write, settling the duplicates on the way. */ +export async function planWorkoutWrites( + userId: string, + survivors: PreparedWorkout[], + results: EntryResult[], +): Promise { + const existingByKey = await readStoredWorkouts(userId, survivors); + const plan: WritePlan = { + toInsert: [], + toEnrich: [], + toOverwrite: [], + duplicates: 0, + }; + /** 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; + const known = keyTuple !== null ? existingByKey.get(keyTuple) : undefined; + if ( + known && + p.dedupKey && + isManualOverwrite(p.dedupKey.source, p.dedupKey.externalId) + ) { + const changes = enrichClaimed.has(known.id) + ? null + : manualOverwriteChanges(known.values, p.row); + if (changes) { + enrichClaimed.add(known.id); + plan.toOverwrite.push({ + index: p.index, + workoutId: known.id, + externalId: p.dedupKey.externalId, + changes, + }); + } else { + results[p.index] = { index: p.index, status: "duplicate" }; + plan.duplicates += 1; + } + } else 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`. + plan.toEnrich.push({ + index: p.index, + workoutId: known.id, + samples: p.samples, + }); + } else { + results[p.index] = { index: p.index, status: "duplicate" }; + plan.duplicates += 1; + } + } else { + results[p.index] = { index: p.index, status: "inserted" }; + plan.toInsert.push(p); + } + } + return plan; +} diff --git a/src/lib/workouts/batch-ingest/write.ts b/src/lib/workouts/batch-ingest/write.ts new file mode 100644 index 000000000..4f9526d41 --- /dev/null +++ b/src/lib/workouts/batch-ingest/write.ts @@ -0,0 +1,260 @@ +/** + * The write of a planned workout batch, in one transaction: the manual + * corrections, the new workouts, then their tracks and series and the + * enrichments. Statuses are settled from what actually landed. + */ +import { prisma } from "@/lib/db"; +import { Prisma } from "@/generated/prisma/client"; +import { lockPersonalRecordsForUser } from "@/lib/personal-records/pr-detection-worker"; +import { encryptRouteGeometry } from "@/lib/workouts/route-geometry-cipher"; + +import type { EntryResult, PreparedWorkout } from "./entries"; +import { MANUAL_OVERWRITE_FIELDS, type WritePlan } from "./plan"; + +const CHUNK = 100; + +type Tx = Prisma.TransactionClient; + +/** What the write reports back for the counters and the arrivals. */ +export interface WriteOutcome { + insertedCount: number; + enrichedCount: number; + updatedCount: number; + /** Planned entries that ended as duplicates once the write had run. */ + duplicates: number; + insertedIdByIndex: Map; + /** Workout ids of the manual edits this request actually wrote. */ + overwrittenIdByIndex: Map; +} + +export async function writeWorkoutPlan( + userId: string, + plan: WritePlan, + results: EntryResult[], +): Promise { + const { toInsert, toEnrich, toOverwrite } = plan; + const outcome: WriteOutcome = { + insertedCount: 0, + enrichedCount: 0, + updatedCount: 0, + duplicates: 0, + insertedIdByIndex: new Map(), + overwrittenIdByIndex: new Map(), + }; + if (!(toInsert.length > 0 || toEnrich.length > 0 || toOverwrite.length > 0)) { + return outcome; + } + const { insertedIdByIndex } = outcome; + /** Workout ids whose series row this request actually wrote. */ + const seriesLanded = new Set(); + /** Batch indices whose manual edit this request actually wrote. */ + const overwritten = new Set(); + + await prisma.$transaction(async (tx) => { + await overwriteManualWorkouts(tx, userId, toOverwrite, overwritten); + await insertWorkouts(tx, toInsert, insertedIdByIndex); + await insertChildren(tx, plan, insertedIdByIndex, seriesLanded); + }); + + outcome.insertedCount = insertedIdByIndex.size; + for (const p of toInsert) { + if (insertedIdByIndex.has(p.index)) { + results[p.index] = { index: p.index, status: "inserted" }; + } else { + results[p.index] = { index: p.index, status: "duplicate" }; + outcome.duplicates += 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)) outcome.enrichedCount += 1; + outcome.duplicates += 1; + } + + // Like an enrichment, an edit inserts no row, so it counts towards + // `duplicates`; the per-entry status says the stored row changed. + for (const o of toOverwrite) { + const landed = overwritten.has(o.index); + results[o.index] = { + index: o.index, + status: landed ? "updated" : "duplicate", + }; + if (landed) { + outcome.updatedCount += 1; + outcome.overwrittenIdByIndex.set(o.index, o.workoutId); + } + outcome.duplicates += 1; + } + return outcome; +} + +/** + * A hand-entered workout sent again with other values. Columns are copied + * one by one from the validated entry; the records the old values set go + * with them and the detection pass re-derives the honest best, as the + * DELETE does. + */ +async function overwriteManualWorkouts( + tx: Tx, + userId: string, + toOverwrite: WritePlan["toOverwrite"], + overwritten: Set, +): Promise { + if (toOverwrite.length > 0) { + await lockPersonalRecordsForUser(tx, userId); + } + for (const o of toOverwrite) { + const data: Prisma.WorkoutUpdateManyMutationInput = {}; + for (const field of MANUAL_OVERWRITE_FIELDS) { + if (field in o.changes) { + (data as Record)[field] = o.changes[field]; + } + } + const written = await tx.workout.updateMany({ + where: { id: o.workoutId, userId, source: "MANUAL" }, + data, + }); + if (written.count === 1) { + overwritten.add(o.index); + await tx.personalRecord.deleteMany({ + where: { + userId, + source: "MANUAL", + metricSlot: { not: null }, + externalId: o.externalId, + }, + }); + // The paragraph described the old values (a 12 km run corrected + // to 1.2 km must not keep its 12 km story). It goes with them, + // and the claim is let go of its workout the way a delete does + // (`ON DELETE SET NULL`): the row stays as the day's ledger, so + // the edit buys no slot back, and the arrival emitted afterwards can + // claim a fresh one through the ordinary path, under the same cap. + await tx.workoutInsight.deleteMany({ + where: { userId, workoutId: o.workoutId }, + }); + await tx.workoutInsightGenerationClaim.updateMany({ + where: { userId, workoutId: o.workoutId }, + data: { workoutId: null }, + }); + } + } +} + +async function insertWorkouts( + tx: Tx, + toInsert: PreparedWorkout[], + insertedIdByIndex: Map, +): Promise { + const withExternalId = toInsert.filter((p) => p.dedupKey !== null); + const withoutExternalId = toInsert.filter((p) => p.dedupKey === null); + + // PostgreSQL RETURNING is the only reliable way to identify the exact + // winner when another request races the same unique key. + for (let i = 0; i < withExternalId.length; i += CHUNK) { + const slice = withExternalId.slice(i, i + CHUNK); + const pendingByKey = new Map(); + for (const p of slice) { + const key = `${p.dedupKey!.source}::${p.dedupKey!.externalId}`; + const pending = pendingByKey.get(key); + if (pending) pending.push(p); + else pendingByKey.set(key, [p]); + } + + const inserted = await tx.workout.createManyAndReturn({ + data: slice.map((p) => p.row), + skipDuplicates: true, + select: { id: true, source: true, externalId: true }, + }); + for (const row of inserted) { + if (row.externalId === null) continue; + const key = `${row.source}::${row.externalId}`; + const preparedRow = pendingByKey.get(key)?.shift(); + if (preparedRow) insertedIdByIndex.set(preparedRow.index, row.id); + } + } + + // NULL external ids are intentionally distinct in PostgreSQL. Insert + // them one at a time so a route/HR series can never attach to a + // different otherwise-identical manual workout. + for (const p of withoutExternalId) { + const inserted = await tx.workout.createManyAndReturn({ + data: p.row, + select: { id: true }, + }); + const row = inserted[0]; + if (row) insertedIdByIndex.set(p.index, row.id); + } +} + +async function insertChildren( + tx: Tx, + plan: WritePlan, + insertedIdByIndex: Map, + seriesLanded: Set, +): Promise { + const routesToInsert: Prisma.WorkoutRouteCreateManyInput[] = []; + const samplesToInsert: Prisma.WorkoutSamplesCreateManyInput[] = []; + for (const p of plan.toInsert) { + const id = insertedIdByIndex.get(p.index); + if (!id) continue; + if (p.route) { + routesToInsert.push({ + workoutId: id, + // The GPS track is stored sealed only; the readable `geometry` + // column stays NULL. + geometryEncrypted: encryptRouteGeometry(p.route.geometry), + sampleTimestamps: + p.route.sampleTimestamps === null + ? Prisma.JsonNull + : (p.route.sampleTimestamps as Prisma.InputJsonValue), + }); + } + if (p.samples) { + samplesToInsert.push({ + workoutId: id, + samples: p.samples.samples as Prisma.InputJsonValue, + sampleCount: p.samples.sampleCount, + }); + } + } + + // 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 plan.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), + skipDuplicates: true, + }); + } + for (let i = 0; i < samplesToInsert.length; i += CHUNK) { + // 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); + } +}