diff --git a/src/app/api/workouts/[id]/route.ts b/src/app/api/workouts/[id]/route.ts index 21f04d8a5..a5ff667fa 100644 --- a/src/app/api/workouts/[id]/route.ts +++ b/src/app/api/workouts/[id]/route.ts @@ -36,6 +36,10 @@ import { annotate } from "@/lib/logging/context"; import { auditLog } from "@/lib/auth/audit"; import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; import { enqueuePrDetection } from "@/lib/jobs/pr-detection"; +import { + lockPersonalRecordsForUser, + workoutPrSlotsForSport, +} from "@/lib/personal-records/pr-detection-worker"; import { pickCanonicalWorkoutRows } from "@/lib/measurements/pick-canonical-workout-rows"; import { requireModuleEnabled } from "@/lib/modules/gate"; import { aiCapabilityToServe } from "@/lib/ai/capabilities/gate"; @@ -380,7 +384,9 @@ export const GET = apiHandler( * (`module-route-gate-inventory.test.ts`). * * The workout table carries no tombstone, so the row is removed outright; - * route, samples, insight and claim rows cascade with it. A personal record + * route, samples and insight rows cascade with it. The insight's generation + * claim stays behind with a null workout id: it is the day's cap ledger, and + * a delete must not hand the slot back. A personal record * the workout set would otherwise outlive it (the detector only ever raises * a stored best), so the records keyed on this workout go in the same * transaction and a silent detection pass re-derives the honest best. @@ -421,14 +427,25 @@ export const DELETE = apiHandler( } const removedRecords = await prisma.$transaction(async (tx) => { + // The detector reads the best workout and writes its record under this + // same lock, so it either wrote before this delete (and the record goes + // below) or reads after it (and never sees the workout). + await lockPersonalRecordsForUser(tx, user.id); const records = await tx.personalRecord.deleteMany({ where: { userId: user.id, - metricSlot: { not: null }, source: "MANUAL", ...(row.externalId !== null - ? { externalId: row.externalId } - : { externalId: null, achievedAt: row.startedAt }), + ? { metricSlot: { not: null }, externalId: row.externalId } + : { + // A legacy record carries no external id, so the start time + // is all that ties it to this workout. Scope it to the slots + // of this workout's sport so a record another sport set at + // the same instant stays. + metricSlot: { in: workoutPrSlotsForSport(row.sportType) }, + externalId: null, + achievedAt: row.startedAt, + }), }, }); await tx.workout.delete({ where: { id: row.id } }); diff --git a/src/app/api/workouts/__tests__/delete.test.ts b/src/app/api/workouts/__tests__/delete.test.ts index 340fd2cee..8af47cf6a 100644 --- a/src/app/api/workouts/__tests__/delete.test.ts +++ b/src/app/api/workouts/__tests__/delete.test.ts @@ -13,6 +13,7 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; import { NextRequest } from "next/server"; const tx = { + $queryRaw: vi.fn(async () => []), personalRecord: { deleteMany: vi.fn() }, workout: { delete: vi.fn() }, }; @@ -129,8 +130,16 @@ describe("DELETE /api/workouts/{id}", () => { expect(tx.personalRecord.deleteMany).toHaveBeenCalledWith({ where: { userId: "user-1", - metricSlot: { not: null }, source: "MANUAL", + // Only the running slots: a cycling record set at the same instant + // is not this workout's. + metricSlot: { + in: [ + "longest_run_duration", + "longest_distance_run", + "fastest_5km_time", + ], + }, externalId: null, achievedAt: MANUAL_ROW.startedAt, }, @@ -139,6 +148,21 @@ describe("DELETE /api/workouts/{id}", () => { expect(enqueuePrDetection).not.toHaveBeenCalled(); }); + it("takes the personal-records lock before it deletes anything", async () => { + vi.mocked(prisma.workout.findUnique).mockResolvedValue(MANUAL_ROW as never); + + await del(); + + expect(tx.$queryRaw).toHaveBeenCalledTimes(1); + const lockOrder = tx.$queryRaw.mock.invocationCallOrder[0]; + expect(lockOrder).toBeLessThan( + tx.personalRecord.deleteMany.mock.invocationCallOrder[0], + ); + expect(lockOrder).toBeLessThan( + tx.workout.delete.mock.invocationCallOrder[0], + ); + }); + it.each(["APPLE_HEALTH", "WITHINGS", "WHOOP", "EXTERNAL"])( "refuses a %s workout with 409 and touches nothing", async (source) => { diff --git a/src/lib/personal-records/__tests__/pr-detection-worker.test.ts b/src/lib/personal-records/__tests__/pr-detection-worker.test.ts index 9b3281d3d..cbaf12d2b 100644 --- a/src/lib/personal-records/__tests__/pr-detection-worker.test.ts +++ b/src/lib/personal-records/__tests__/pr-detection-worker.test.ts @@ -143,7 +143,7 @@ function makeFakePrisma(state: { return true; } - return { + const fake = { // v1.29.6 — the cumulative-kind path no longer runs a `$queryRaw` // UTC `date_trunc` aggregate; it reads every matching row via // `measurement.findMany` and does the source-collapse + tz-aware @@ -277,6 +277,13 @@ function makeFakePrisma(state: { ), }, } as unknown as PrismaClient; + // The workout-slot leg runs under a per-user advisory lock inside a + // transaction; the fake runs the callback against itself. + Object.assign(fake, { + $queryRaw: vi.fn(async () => []), + $transaction: vi.fn(async (cb: (tx: PrismaClient) => unknown) => cb(fake)), + }); + return fake; } const USER = "user-1"; diff --git a/src/lib/personal-records/pr-detection-worker.ts b/src/lib/personal-records/pr-detection-worker.ts index ae7ef6f80..d55e5c6dc 100644 --- a/src/lib/personal-records/pr-detection-worker.ts +++ b/src/lib/personal-records/pr-detection-worker.ts @@ -386,58 +386,65 @@ export async function detectPersonalRecordsForUser( [slot.field]: { gt: 0 } as { gt: number }, } as const; - const sampleCount = await prisma.workout.count({ - where: baseWhere, + // Read the best workout and write its record under the same per-user + // lock the workout DELETE takes. Without it a delete could land between + // the read and the write and leave a record for a workout that no longer + // exists (PersonalRecord has no FK to Workout to stop it). + const outcome = await prisma.$transaction(async (tx) => { + await lockPersonalRecordsForUser(tx, userId); + + const sampleCount = await tx.workout.count({ + where: baseWhere, + }); + if (sampleCount < PR_DETECTION_WARMUP_THRESHOLD) return null; + + const best = await findBestWorkout(tx, baseWhere, slot); + if (!best) return null; + + const value = + slot.field === "durationSec" + ? best.durationSec + : (best.totalDistanceM ?? 0); + if (!Number.isFinite(value) || value <= 0) return null; + + const currentPR = await tx.personalRecord.findFirst({ + where: { userId, metricType: slot.metricType, metricSlot: slot.slot }, + orderBy: + slot.direction === PersonalRecordDirection.MAX + ? { value: "desc" } + : { value: "asc" }, + }); + + const comparison = compareToCurrentBest( + value, + currentPR?.value, + slot.direction, + ); + if (comparison === "no-improvement") return null; + + const result = await tx.personalRecord.createMany({ + data: [ + { + userId, + metricType: slot.metricType, + metricSlot: slot.slot, + direction: slot.direction, + value, + unit: slot.unit, + achievedAt: best.startedAt, + sourceMeasurementId: null, + source: best.source, + externalId: best.externalId, + }, + ], + skipDuplicates: true, + }); + return result.count > 0 ? comparison : null; }); - if (sampleCount < PR_DETECTION_WARMUP_THRESHOLD) continue; - - const best = await findBestWorkout(prisma, baseWhere, slot); - if (!best) continue; - - const value = - slot.field === "durationSec" - ? best.durationSec - : (best.totalDistanceM ?? 0); - if (!Number.isFinite(value) || value <= 0) continue; - - const currentPR = await prisma.personalRecord.findFirst({ - where: { userId, metricType: slot.metricType, metricSlot: slot.slot }, - orderBy: - slot.direction === PersonalRecordDirection.MAX - ? { value: "desc" } - : { value: "asc" }, - }); - - const outcome = compareToCurrentBest( - value, - currentPR?.value, - slot.direction, - ); - if (outcome === "no-improvement") continue; - - const result = await prisma.personalRecord.createMany({ - data: [ - { - userId, - metricType: slot.metricType, - metricSlot: slot.slot, - direction: slot.direction, - value, - unit: slot.unit, - achievedAt: best.startedAt, - sourceMeasurementId: null, - source: best.source, - externalId: best.externalId, - }, - ], - skipDuplicates: true, - }); - if (result.count > 0) { - if (outcome === "tie") { - ties += 1; - } else { - inserted += 1; - } + if (outcome === "tie") { + ties += 1; + } else if (outcome === "improvement") { + inserted += 1; } } @@ -669,7 +676,7 @@ async function findBestCumulativeDay( } async function findBestWorkout( - prisma: PrismaClient, + prisma: Pick, where: Prisma.WorkoutWhereInput, slot: WorkoutSlotDefinition, ): Promise { @@ -704,6 +711,33 @@ async function findBestWorkout( }; } +/** + * The per-user lock shared by the workout-slot detection and the workout + * DELETE. Transaction-scoped: it releases on commit or rollback. + */ +export async function lockPersonalRecordsForUser( + tx: Pick, + userId: string, +): Promise { + await tx.$queryRaw` + SELECT pg_advisory_xact_lock( + hashtext('personal-records'), + hashtext(${userId}) + )::text AS locked + `; +} + +/** + * The workout-PR slot names the detector derives for one sport. The DELETE + * uses it to scope a legacy record match (no external id) to the sport of the + * workout being removed. + */ +export function workoutPrSlotsForSport(sportType: string): string[] { + return WORKOUT_SLOTS.filter((slot) => slot.sportType === sportType).map( + (slot) => slot.slot, + ); +} + /** Exposed for tests + the integration harness. */ export const __test__ = { WORKOUT_SLOTS,