mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-12 22:26:13 +00:00
Marc-André Bombeck
5eb7956bad
The GPS track of an outdoor workout was stored readable, and its first and last points are usually the person's front door. The phone number and address in the address book of doctors and practices were readable too, while the note beside them was already encrypted. Migration 0361 adds workout_routes.geometry_encrypted (the geometry loses its NOT NULL) and practitioners.phone_encrypted / location_encrypted, plus partial indexes on the backfill's discovery predicates. A track is sealed as one binary AES-256-GCM value bound to its purpose; the contact fields use the same Bytes codec as the note. The practitioner's name, practice and specialty stay readable, because the picker searches and sorts on them. Every writer stores ciphertext only: workout batch ingest, practitioner create and edit, and both restore formats. Readers prefer the ciphertext and fall back to the old column for rows not yet migrated: the workout detail (map, splits, GPX), the activity insight's climb figure, the practitioner DTO behind the address book, visits and vaccinations. The wire shape does not change. An edit's audit row now names a changed phone number or address without recording either value. free-text-encryption-backfill covers the three columns in the same pass, finding a route through its workout. All three are registered for key rotation (registry, CLI script, and the in-app rotation). A disaster-recovery backup carries the contact fields as ciphertext only; a portable export carries them readable. Workout routes are still not part of a backup, as before. The old columns are marked for removal in a later release.
722 lines
28 KiB
TypeScript
722 lines
28 KiB
TypeScript
/**
|
||
* `POST /api/workouts/batch` — typed workout ingest with nested route.
|
||
*
|
||
* Consumers (v1.5):
|
||
* 1. The native iOS app draining its HealthKit observer queue —
|
||
* maps `HKWorkout` + `HKWorkoutRoute` samples into the request
|
||
* shape locked by `createBatchWorkoutSchema` in
|
||
* `src/lib/validations/workout.ts`.
|
||
* 2. The Withings activity sync (W17b, deferred) — server-to-server
|
||
* ingest with `externalId` formatted as
|
||
* `"${withings.id}:${withings.model_id ?? "0"}:${withings.startdate}"`.
|
||
* Withings ships no route geometry, so the nested `route` is
|
||
* always absent for WITHINGS-sourced entries. NOTE: the request
|
||
* schema's `source` is narrowed to the client-writable set
|
||
* (MANUAL / APPLE_HEALTH), so this server path must write its
|
||
* `WITHINGS`-sourced rows directly via `prisma.workout.upsert`
|
||
* (as the WHOOP / Fitbit syncs already do), not through this route.
|
||
*
|
||
* Cross-cutting concerns mirror `/api/measurements/batch` so the iOS
|
||
* sync engine can re-use the same retry / cursor plumbing:
|
||
* - `requireAuth()` — cookie + Bearer (narrow-scope-token safe per
|
||
* v1.4.25 W10 fix-C: when the route declares no scope, any
|
||
* authenticated token passes).
|
||
* - `withIdempotency` — `Idempotency-Key` header replays the cached
|
||
* response on retry (24h window).
|
||
* - `checkRateLimit("workouts:batch:${user.id}", 60, 60s)` — same
|
||
* window as the measurements batch. 60 batches/min × 100
|
||
* workouts/batch = 6 000 workouts/min headroom (orders of
|
||
* magnitude past any healthy iOS sync).
|
||
* - `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 | enriched | skipped`) so
|
||
* the iOS sync cursor can checkpoint accurately.
|
||
*
|
||
* Idempotency contract:
|
||
* - HTTP-level via `Idempotency-Key` (replays cached envelope).
|
||
* - Per-entry via the `@@unique([userId, source, externalId])`
|
||
* 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
|
||
* arrival references cannot drift to a racing or same-external-id workout.
|
||
*
|
||
* Prompt-injection surface: `Workout.metadata` is opaque — the Coach
|
||
* pipeline reads typed columns only (`sportType`, `durationSec`,
|
||
* distance, HR). A future change that pulls metadata into the Coach
|
||
* prompt must add deliberate escaping; the ingest path stores it raw.
|
||
*/
|
||
import { NextRequest } from "next/server";
|
||
|
||
import { prisma } from "@/lib/db";
|
||
import { apiHandler, requireAuth } from "@/lib/api-handler";
|
||
import { annotate } from "@/lib/logging/context";
|
||
import { auditLog } from "@/lib/auth/audit";
|
||
import {
|
||
apiError,
|
||
apiSuccess,
|
||
getClientIp,
|
||
safeJson,
|
||
} from "@/lib/api-response";
|
||
import { withIdempotency } from "@/lib/idempotency";
|
||
import { 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 {
|
||
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 { dedupeWorkoutBatch } from "@/lib/workouts/canonical-rows";
|
||
import { encryptRouteGeometry } from "@/lib/workouts/route-geometry-cipher";
|
||
import { Prisma, type MeasurementSource } from "@/generated/prisma/client";
|
||
|
||
// v1.4.25 W16c — 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 W10 reconcile (security H-2 parity): cap batch ingest at 60
|
||
// batches per user per minute. Healthy iOS sync drains its HealthKit
|
||
// observer queue in well under one batch per minute (the observer
|
||
// pattern coalesces); 60/min × 100 workouts/batch = 6 000 workouts/
|
||
// min headroom — generous for legitimate use, and a hard stop for a
|
||
// leaked wildcard token trying to saturate the write pipeline.
|
||
const BATCH_RATE_LIMIT_MAX = 60;
|
||
const BATCH_RATE_LIMIT_WINDOW_MS = 60 * 1000;
|
||
|
||
// 5 MB request-body ceiling (pre-existing). The route geometry is the
|
||
// largest tail in a workout payload; a typical batch sits comfortably
|
||
// under this, but the 20 000-point LineString cap combined with the
|
||
// 100-workout batch cap allows a theoretical tens-of-MB body for an
|
||
// extreme GPS-heavy backlog. Such a body returns 413 so the iOS client
|
||
// falls back to smaller / one-workout-per-call batches — which is also
|
||
// 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.
|
||
*/
|
||
type EntryStatus = "inserted" | "duplicate" | "enriched" | "skipped";
|
||
interface EntryResult {
|
||
index: number;
|
||
status: EntryStatus;
|
||
reason?: string;
|
||
}
|
||
|
||
/**
|
||
* v1.31.1 — emit one data arrival per workout this batch actually inserted.
|
||
*
|
||
* `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<number, string>,
|
||
): Promise<void> {
|
||
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<Response> {
|
||
const { user } = await requireAuth();
|
||
|
||
// v1.4.43 W9 — resolve the per-user source-priority blob once at
|
||
// the start of the handler so the write-time canonical-row picker
|
||
// walks the SAME ladder the read-time picker walks. Without this
|
||
// lookup a user who customised Settings → Sources (e.g. MANUAL >
|
||
// APPLE_HEALTH) would see their preferred row dropped at write-
|
||
// time before the read-time picker ever ran. One indexed
|
||
// findUnique per batch — well under the cost of letting the
|
||
// duplicate persist and carrying it forward on every read.
|
||
const userPriority = await prisma.user.findUnique({
|
||
where: { id: user.id },
|
||
select: { sourcePriorityJson: true },
|
||
});
|
||
|
||
// Content-Length pre-flight. The body cap exists to bound the
|
||
// request-pipeline cost BEFORE Node has to read the full payload —
|
||
// a misbehaving client shipping a 50 MB body shouldn't tie up the
|
||
// worker for a parse it'd reject anyway. We trust the header here
|
||
// because Next.js / undici populates it for any client that sets a
|
||
// body length; for chunked uploads (no Content-Length) we fall
|
||
// through and rely on the schema-level point / workout caps to keep
|
||
// memory bounded.
|
||
const contentLength = request.headers.get("content-length");
|
||
if (contentLength !== null) {
|
||
const parsedLength = Number.parseInt(contentLength, 10);
|
||
if (Number.isFinite(parsedLength) && parsedLength > MAX_BODY_BYTES) {
|
||
annotate({
|
||
action: { name: "workout.batch.ingest" },
|
||
meta: { outcome: "payload_too_large", bytes: parsedLength },
|
||
});
|
||
return apiError(
|
||
`Request body exceeds the ${MAX_BODY_BYTES}-byte limit`,
|
||
413,
|
||
{ errorCode: "workout.batch.payload_too_large" },
|
||
);
|
||
}
|
||
}
|
||
|
||
// v1.4.25 W10 reconcile (security H-2 parity): per-user rate limit.
|
||
// Without it, a leaked wildcard iOS token can sustain unbounded
|
||
// writes (100 workouts/batch × N batches/sec) and degrade the
|
||
// database for every other user on the host. 60 batches/min/user is
|
||
// generous for healthy iOS sync and tight enough that a misbehaving
|
||
// client bottoms out within a minute.
|
||
const rl = await checkRateLimit(
|
||
`workouts:batch:${user.id}`,
|
||
BATCH_RATE_LIMIT_MAX,
|
||
BATCH_RATE_LIMIT_WINDOW_MS,
|
||
);
|
||
if (!rl.allowed) {
|
||
annotate({
|
||
action: { name: "workout.batch.ingest" },
|
||
meta: { outcome: "rate_limited" },
|
||
});
|
||
return apiError("Too many batch submissions, try again later", 429);
|
||
}
|
||
|
||
// The Content-Length pre-flight above bounds the common case, but a
|
||
// chunked upload carries no Content-Length. The `safeJson` body cap
|
||
// backstops that gap: the raw text is measured before `JSON.parse`,
|
||
// so an over-limit chunked body returns a clean 413 before the parse
|
||
// cost. Reuses the same 5 MB ceiling — a typical iOS batch sits under
|
||
// it; an extreme GPS-heavy backlog that exceeds it falls back to
|
||
// smaller batches client-side.
|
||
const { data: rawBody, error: jsonError } = await safeJson<unknown>(request, {
|
||
maxBytes: MAX_BODY_BYTES,
|
||
});
|
||
if (jsonError) return jsonError;
|
||
|
||
// Distinguish "too many entries" from "validation failed" so the
|
||
// client surfaces a useful diagnostic. The cap stops a buggy client
|
||
// from saturating the request pipeline; the iOS app paginates above
|
||
// this threshold.
|
||
if (
|
||
typeof rawBody === "object" &&
|
||
rawBody !== null &&
|
||
"workouts" in rawBody &&
|
||
Array.isArray((rawBody as { workouts: unknown }).workouts) &&
|
||
(rawBody as { workouts: unknown[] }).workouts.length >
|
||
MAX_WORKOUTS_PER_BATCH
|
||
) {
|
||
return apiError(
|
||
`Batch exceeds the ${MAX_WORKOUTS_PER_BATCH}-workout limit`,
|
||
400,
|
||
{ errorCode: "workout.batch.too_large" },
|
||
);
|
||
}
|
||
|
||
// 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)` —
|
||
// 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
|
||
? {
|
||
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,
|
||
elevationM: w.elevationM ?? null,
|
||
pauseDurationSec: w.pauseDurationSec ?? null,
|
||
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: w.source, externalId: w.externalId }
|
||
: null,
|
||
},
|
||
];
|
||
});
|
||
|
||
if (unstableShapes.length > 0) {
|
||
annotate({
|
||
meta: unstableExternalIdMeta("workout.batch", unstableShapes),
|
||
});
|
||
}
|
||
|
||
// v1.4.42 W5 — write-time cross-source dedup. The v1.5 iOS app
|
||
// drains a HealthKit observer queue that frequently carries the
|
||
// SAME logical workout from two paired sensors (Apple Watch +
|
||
// Withings ScanWatch). Without this pass, both rows persist (their
|
||
// `externalId`s differ → the `(userId, source, externalId)` unique
|
||
// index never fires) and the read-time canonical-picker has to
|
||
// carry the duplicate forward on every dashboard render.
|
||
//
|
||
// The helper is pure (no Prisma, no user lookups). It groups by
|
||
// `(userId, activityType, startedAt ± 90 s)`, prefers the canonical
|
||
// source ladder (APPLE_HEALTH > WITHINGS > MANUAL > IMPORT), and
|
||
// breaks ties on calories > earliest createdAt > input order. Rows
|
||
// dropped here surface to the iOS client as `duplicate` so the
|
||
// sync cursor advances past them identically to the externalId
|
||
// case below.
|
||
const canonicalPrepared = dedupeWorkoutBatch(
|
||
prepared.map((p) => ({
|
||
userId: p.row.userId,
|
||
activityType: p.row.sportType,
|
||
startedAt: p.row.startedAt as Date,
|
||
// Zod's `measurementSourceEnum.optional().default("MANUAL")`
|
||
// guarantees `source` is set post-parse, but Prisma's
|
||
// `WorkoutCreateManyInput` types it optional. Default through
|
||
// the same fallback the validator uses so the picker's source
|
||
// ladder still resolves on the rare row where TypeScript
|
||
// can't prove the union.
|
||
source: (p.row.source ?? "MANUAL") as MeasurementSource,
|
||
caloriesKcal: p.row.totalEnergyKcal ?? null,
|
||
// No createdAt yet — the row is pre-insert; the tie-breaker
|
||
// falls through to input order via the `index` field below.
|
||
index: p.index,
|
||
})),
|
||
userPriority?.sourcePriorityJson ?? null,
|
||
);
|
||
const survivingIndices = new Set(canonicalPrepared.map((r) => r.index ?? -1));
|
||
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;
|
||
const insertedIdByIndex = new Map<number, string>();
|
||
|
||
if (survivors.length > 0) {
|
||
// v1.4.42 W5 — 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 } },
|
||
},
|
||
})
|
||
: [];
|
||
|
||
const existingByKey = new Map<string, { id: string; hasSeries: boolean }>();
|
||
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<Prepared["samples"]>;
|
||
}> = [];
|
||
/** Workouts already claimed by an earlier entry of THIS batch. */
|
||
const enrichClaimed = new Set<string>();
|
||
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) {
|
||
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) {
|
||
const CHUNK = 100;
|
||
/** Workout ids whose series row this request actually wrote. */
|
||
const seriesLanded = new Set<string>();
|
||
|
||
await prisma.$transaction(async (tx) => {
|
||
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<string, Prepared[]>();
|
||
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;
|
||
}
|
||
}
|
||
}
|
||
|
||
const skipped = results
|
||
.filter((r) => r.status === "skipped")
|
||
.map((r) => ({ index: r.index, reason: r.reason ?? "unknown" }));
|
||
|
||
await auditLog("workout.batch.ingest", {
|
||
userId: user.id,
|
||
ipAddress: getClientIp(request),
|
||
details: {
|
||
processed: workouts.length,
|
||
inserted: insertedCount,
|
||
duplicates: duplicateCount,
|
||
enriched: enrichedCount,
|
||
skipped: skipped.length,
|
||
},
|
||
});
|
||
|
||
// v1.4.25 W16c — 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 });
|
||
await auditLog("personal_records.detection_enqueued", {
|
||
userId: user.id,
|
||
details: {
|
||
source: "workout.batch",
|
||
batchSize: workouts.length,
|
||
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,
|
||
skipped: skipped.length,
|
||
},
|
||
});
|
||
|
||
// v1.4.34 IW-G — 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) {
|
||
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(
|
||
() => {},
|
||
);
|
||
}
|
||
|
||
return apiSuccess({
|
||
processed: workouts.length,
|
||
inserted: insertedCount,
|
||
duplicates: duplicateCount,
|
||
skipped,
|
||
entries: results,
|
||
});
|
||
}
|