tom/HealthLog
1
0 Fork 0
mirror of https://github.com/MBombeck/HealthLog.git synced 2026-11-09 04:50:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(workouts): move the batch ingest out of its route

Browse Source 
postBatch was a 700-line function. The route keeps authentication, the
per-account preference read, the body caps, the rate limit, the parse and
the bridge-source refusal; the ingest moves verbatim into
lib/workouts/batch-ingest/: the first pass (entries), the duplicate probe
and write plan with the manual-correction rules (plan), the transaction
that writes corrections, workouts, tracks and series (write), and the
orchestrator with the audit, record detection, wide event, cache and
arrivals (ingest). The route is 275 lines.

The parity walk recorded before the move is unchanged; the guards that
name the track writer, the Workout writer set and the per-entry
external-id check point at the service files.
This commit is contained in:
Marc-André Bombeck committed 2026-10-23 11:39:00 +02:00
1 parent dd52a8d92a
commit 899b86f4d4
9 files changed
+909 -699
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/__tests__/ai-egress-capability-guard.test.ts
+2 -2
View File
@@ -314,8 +314,8 @@ describe("workout route geometry never reaches an AI path", () => {
const ROUTE_READERS: Record<string, string> = {
"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":
src/__tests__/external-id-stability-guard.test.ts
+4 -1
View File
@@ -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": [],
};
src/__tests__/workout-sport-provenance-guard.test.ts
+3 -2
View File
@@ -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<string, string> = {
// 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",
src/app/api/workouts/batch/route.ts
+10 -693
View File
@@ -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<ManualOverwriteValues> | null {
const changes: Partial<ManualOverwriteValues> = {};
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<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> {
@@ -382,566 +248,17 @@ async function postBatch(request: NextRequest): Promise<Response> {
}
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<number, string>();
/** Workout ids of the manual edits this request actually wrote. */
const overwrittenIdByIndex = new Map<number, string>();
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<Prepared["samples"]>;
}> = [];
/** Hand-entered workouts this batch edits in place. */
const toOverwrite: Array<{
index: number;
workoutId: string;
externalId: string;
changes: Partial<ManualOverwriteValues>;
}> = [];
/** 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 &&
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<string>();
/** Batch indices whose manual edit this request actually wrote. */
const overwritten = new Set<number>();
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<string, unknown>)[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<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;
}
// 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. */
src/lib/validations/workout.ts
+1 -1
View File
@@ -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(),
src/lib/workouts/batch-ingest/entries.ts
+180
View File
@@ -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<number>;
}): {
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,
};
}
src/lib/workouts/batch-ingest/ingest.ts
+227
View File
@@ -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<number>;
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<WorkoutBatchResult> {
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<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",
});
}
}
src/lib/workouts/batch-ingest/plan.ts
+222
View File
@@ -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<ManualOverwriteValues> | null {
const changes: Partial<ManualOverwriteValues> = {};
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<PreparedWorkout["samples"]>;
}>;
/** Hand-entered workouts this batch edits in place. */
toOverwrite: Array<{
index: number;
workoutId: string;
externalId: string;
changes: Partial<ManualOverwriteValues>;
}>;
/** 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<Map<string, StoredWorkout>> {
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<string, StoredWorkout>();
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<WritePlan> {
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<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 &&
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;
}
src/lib/workouts/batch-ingest/write.ts
+260
View File
@@ -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<number, string>;
/** Workout ids of the manual edits this request actually wrote. */
overwrittenIdByIndex: Map<number, string>;
}
export async function writeWorkoutPlan(
userId: string,
plan: WritePlan,
results: EntryResult[],
): Promise<WriteOutcome> {
const { toInsert, toEnrich, toOverwrite } = plan;
const outcome: WriteOutcome = {
insertedCount: 0,
enrichedCount: 0,
updatedCount: 0,
duplicates: 0,
insertedIdByIndex: new Map<number, string>(),
overwrittenIdByIndex: new Map<number, string>(),
};
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<string>();
/** Batch indices whose manual edit this request actually wrote. */
const overwritten = new Set<number>();
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<number>,
): Promise<void> {
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<string, unknown>)[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<number, string>,
): Promise<void> {
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, PreparedWorkout[]>();
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<number, string>,
seriesLanded: Set<string>,
): Promise<void> {
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);
}
}