From c88754a0b50f7aa3a7d4a6a7f1b4c2e9921003bc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sat, 26 Sep 2026 16:07:51 +0200 Subject: [PATCH] fix(measurements): re-warm status assessments from every measurement write The MCP writes, the Telegram reply, the CSV and backup importers and the Apple export import refreshed the rollups by hand and never re-warmed the cached status assessments, so a reading logged there kept the day's old assessment until the nightly pass. They now run the shared post-write tail, and the export import re-warms every type it wrote. The WHOOP body sync removed its legacy weight row without refolding that day; it does now. A structural test lists every file that writes measurements and how it keeps both readers current, so a new write site fails until it says. --- .../measurement-write-tail-guard.test.ts | 171 ++++++++++++++++++ src/app/api/import/csv/route.ts | 27 +-- src/app/api/import/route.ts | 31 +--- src/lib/jobs/apple-health-import-worker.ts | 19 +- src/lib/mcp/__tests__/writes.test.ts | 38 +++- src/lib/mcp/writes.ts | 49 ++--- .../__tests__/create-from-telegram.test.ts | 12 +- src/lib/measurements/create-from-telegram.ts | 19 +- src/lib/whoop/__tests__/sync-body.test.ts | 31 +++- src/lib/whoop/sync-body.ts | 23 ++- 10 files changed, 318 insertions(+), 102 deletions(-) create mode 100644 src/__tests__/measurement-write-tail-guard.test.ts diff --git a/src/__tests__/measurement-write-tail-guard.test.ts b/src/__tests__/measurement-write-tail-guard.test.ts new file mode 100644 index 000000000..dc56b62f8 --- /dev/null +++ b/src/__tests__/measurement-write-tail-guard.test.ts @@ -0,0 +1,171 @@ +/** + * Every place that writes `Measurement` rows either runs the shared + * post-write tail or says here why it does not. + * + * Two readers hang off every measurement write: the rollup buckets the write + * dirtied, and the cached status assessments of the types it touched. The + * shared tail (`afterMeasurementMutation`) runs both. Several write paths + * refreshed the rollups by hand and never re-warmed the assessments (the MCP + * writes, the Telegram reply, both importers), so a reading logged there kept + * the day's old assessment until the nightly pass. This guard freezes the set + * of write sites: a new one fails until it is listed with its tail. + * + * A tripwire, not a proof: it cannot tell whether a listed reason is still + * true, only that the set did not change without someone editing this file. + */ +import { describe, it, expect } from "vitest"; +import { readFileSync } from "node:fs"; +import { join } from "node:path"; + +import { walkSourceFiles } from "./helpers/source-files"; + +const SRC = join(process.cwd(), "src"); + +/** + * `prisma.measurement.create(`, `tx.measurement.deleteMany(`, … Whitespace + * and line breaks between the parts are tolerated so a formatter reflow does + * not hide a site. + */ +const WRITE_RE = + /\.\s*measurement\s*\.\s*(?:create|createMany|createManyAndReturn|upsert|update|updateMany|delete|deleteMany)\s*\(/g; + +/** Runs `afterMeasurementMutation` on the rows it wrote. */ +const TAIL = "runs afterMeasurementMutation"; + +/** + * Every file with a measurement write, and how it keeps the two readers + * current. A file listed here must still contain a write (a stale entry + * fails too). + */ +const WRITE_SITES: Record = { + "app/api/import/csv/route.ts": TAIL, + "app/api/import/route.ts": TAIL, + "app/api/measurements/[id]/route.ts": TAIL, + "app/api/measurements/bulk-delete/route.ts": TAIL, + "app/api/measurements/by-external-ids/route.ts": TAIL, + "app/api/measurements/restore/route.ts": TAIL, + "app/api/measurements/route.ts": TAIL, + "app/api/mental-health/assessments/route.ts": TAIL, + "lib/insights/score-row.ts": TAIL, + "lib/mcp/writes.ts": TAIL, + "lib/measurements/create-from-telegram.ts": TAIL, + "lib/whoop/sync-body.ts": TAIL, + "lib/whoop/webhook-handler.ts": TAIL, + // Integration syncs: each collects the (type, day) pairs it touched and + // runs both legs itself at the end of the sync. + "lib/fitbit/sync-core.ts": + "refolds the touched days and re-warms the touched types at the end of the upsert", + "lib/google-health/sync-core.ts": + "refolds the touched days and re-warms the touched types at the end of the upsert", + "lib/nightscout/sync.ts": + "refolds the touched days and re-warms the touched types at the end of the sync", + "lib/withings/sync.ts": + "refolds the touched days and re-warms the touched types at the end of the sync", + "lib/withings/sync-activity.ts": + "refolds the touched days (and the day a moved row left) and re-warms the touched types", + "lib/withings/sync-ecg.ts": + "refolds the touched days and re-warms the touched types at the end of the sync", + // The reconcile primitive writes one row inside the caller's transaction; + // its callers (Polar, Oura, WHOOP, the Apple export import) collect the + // verdicts' dirty identities and run the tail once per batch. + "lib/measurements/reconcile-external-measurement.ts": + "a primitive; every caller runs the tail over the verdicts' dirty identities", + "lib/measurements/import-apple-health-export.ts": + "the import worker refolds the import's span and re-warms every imported type", + "lib/export/restore-backup.ts": + "a restore replaces the whole record: it refolds the whole window and drops every per-user cache", + // Value-preserving reshapes: the same readings folded into daily or hourly + // rows. The rollups are refolded; the figures an assessment reads do not + // change, so re-warming would only re-bill the same text. + "lib/measurements/consolidate-daily-mean.ts": + "value-preserving consolidation; refolds the touched days", + "lib/measurements/consolidate-legacy-steps.ts": + "value-preserving consolidation; refolds the touched days", + "lib/measurements/drain-per-sample-cumulative.ts": + "value-preserving consolidation; refolds the touched days", + "lib/measurements/dense-intraday-hourly-rebuild.ts": + "value-preserving hourly fold; refolds the touched spans", + "lib/measurements/dense-intraday-retention.ts": + "value-preserving retention fold; refolds the touched days", + "lib/jobs/sleep-timeline-backfill.ts": + "one-shot re-stamp of stored sleep rows onto their timeline; refolds the touched days", + "lib/jobs/step-consolidation-repair.ts": + "one-shot restore of step rows an old consolidation removed; refolds the touched days", + "lib/sleep/sweep-stale-segments.ts": + "tombstones re-scored sleep segments inside a sync; the calling sync refolds the night", + // Writes that do not change a reading. + "lib/jobs/note-encryption-backfill.ts": + "encrypts the note column in place; no value, instant or type changes", + "lib/jobs/measurement-tombstone-cleanup.ts": + "hard-deletes rows that were already soft-deleted and read nowhere", +}; + +function sourceFiles(): string[] { + return walkSourceFiles(SRC, { floor: 3000 }) + .filter((p) => !p.startsWith("generated/")) + .filter((p) => !p.includes("__tests__")) + .filter((p) => !p.endsWith(".test.ts") && !p.endsWith(".test.tsx")) + .sort(); +} + +function writeCounts(): Map { + const out = new Map(); + for (const rel of sourceFiles()) { + const matches = readFileSync(join(SRC, rel), "utf8").match(WRITE_RE); + if (matches && matches.length > 0) out.set(rel, matches.length); + } + return out; +} + +describe("measurement write tail", () => { + const counts = writeCounts(); + + it("finds the write sites at all", () => { + // An empty match set would agree with any allowlist. + expect(counts.size).toBeGreaterThanOrEqual(30); + }); + + it("lists every file that writes measurements", () => { + const unlisted = [...counts.keys()].filter((f) => !(f in WRITE_SITES)); + expect( + unlisted, + "a new measurement write site: run afterMeasurementMutation on the rows it writes, or list it here with the reason it does not", + ).toEqual([]); + }); + + it("has no stale entries", () => { + const stale = Object.keys(WRITE_SITES).filter((f) => !counts.has(f)); + expect(stale).toEqual([]); + }); + + it("the files that claim the shared tail call it", () => { + const missing = Object.entries(WRITE_SITES) + .filter(([, reason]) => reason === TAIL) + .map(([file]) => file) + .filter( + (file) => + !/afterMeasurementMutation\s*\(/.test( + readFileSync(join(SRC, file), "utf8"), + ), + ); + expect(missing).toEqual([]); + }); + + it("the syncs that run both legs by hand call both", () => { + const handRolled = Object.entries(WRITE_SITES) + .filter(([, reason]) => reason.includes("re-warms")) + .map(([file]) => file) + .filter( + (file) => file !== "lib/measurements/import-apple-health-export.ts", + ); + expect(handRolled.length).toBeGreaterThan(0); + const missing = handRolled.filter((file) => { + const src = readFileSync(join(SRC, file), "utf8"); + return !( + /recomputeBucketsForMeasurement|afterMeasurementMutation/.test(src) && + /invalidateStatusInsightsForTypes|afterMeasurementMutation/.test(src) + ); + }); + expect(missing).toEqual([]); + }); +}); diff --git a/src/app/api/import/csv/route.ts b/src/app/api/import/csv/route.ts index bbd46810c..e5535b9a9 100644 --- a/src/app/api/import/csv/route.ts +++ b/src/app/api/import/csv/route.ts @@ -11,10 +11,7 @@ import { parseCsvMeasurements, type CsvRowResult, } from "@/lib/import/csv-measurements"; -import { - collapseToTypeDayKeys, - recomputeBucketsForMeasurement, -} from "@/lib/rollups/measurement-rollups"; +import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation"; import type { MeasurementType, GlucoseContext, @@ -317,24 +314,10 @@ export const POST = apiHandler(async (request: NextRequest) => { } // One bounded rollup re-fold per touched (type, day) — a 10 000-row CSV - // pays at most ~N (type, day) recomputes, not 10 000 per-row hooks. - // Best-effort: a populator hiccup never fails the importer. - if (touchedMeasurements.length > 0) { - try { - const keys = collapseToTypeDayKeys(touchedMeasurements); - for (const k of keys) { - await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt); - } - } catch (err) { - annotate({ - meta: { - measurement_rollup_csv_import_failed: true, - measurement_rollup_csv_import_error: - err instanceof Error ? err.message : String(err), - }, - }); - } - } + // pays at most ~N (type, day) recomputes, not 10 000 per-row hooks — and + // the status re-warm for the touched types. Best-effort: a populator + // hiccup never fails the importer. + await afterMeasurementMutation(userId, touchedMeasurements, "import.csv"); await auditLog("import.csv.upload", { userId, diff --git a/src/app/api/import/route.ts b/src/app/api/import/route.ts index 02bad9389..88ea23e34 100644 --- a/src/app/api/import/route.ts +++ b/src/app/api/import/route.ts @@ -32,10 +32,7 @@ import { } from "@/lib/validations/external-id"; import { encryptNote } from "@/lib/crypto/note-cipher"; import { recomputeUserMoodRollups } from "@/lib/rollups/mood-rollups"; -import { - collapseToTypeDayKeys, - recomputeBucketsForMeasurement, -} from "@/lib/rollups/measurement-rollups"; +import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation"; import type { MeasurementType } from "@/generated/prisma/client"; import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit"; import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger"; @@ -481,27 +478,11 @@ export const POST = apiHandler(async (request: NextRequest) => { } } - // v1.4.39.1 — refresh the persistent measurement rollup table for - // every distinct (type, day) the import touched. Collapsed so a - // 10 000-row CSV restore pays at most ~N (type, day) recomputes - // rather than 10 000 per-row hooks. Best-effort: a populator hiccup - // never fails the importer. - if (touchedMeasurements.length > 0) { - try { - const keys = collapseToTypeDayKeys(touchedMeasurements); - for (const k of keys) { - await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt); - } - } catch (err) { - annotate({ - meta: { - measurement_rollup_import_failed: true, - measurement_rollup_import_error: - err instanceof Error ? err.message : String(err), - }, - }); - } - } + // Refresh the persistent measurement rollup table for every distinct + // (type, day) the import touched (collapsed, so a 10 000-row restore pays + // at most ~N recomputes) and re-warm the touched types' status + // assessments. Best-effort: a populator hiccup never fails the importer. + await afterMeasurementMutation(userId, touchedMeasurements, "import"); annotate({ meta: { diff --git a/src/lib/jobs/apple-health-import-worker.ts b/src/lib/jobs/apple-health-import-worker.ts index 257649e72..2875cddf2 100644 --- a/src/lib/jobs/apple-health-import-worker.ts +++ b/src/lib/jobs/apple-health-import-worker.ts @@ -26,7 +26,7 @@ */ import { unlinkSync } from "node:fs"; import { getHeapStatistics } from "node:v8"; -import type { PrismaClient } from "@/generated/prisma/client"; +import type { MeasurementType, PrismaClient } from "@/generated/prisma/client"; import { prisma, toJson } from "@/lib/db"; import type { Job } from "pg-boss"; @@ -35,6 +35,7 @@ import { parseAppleHealthEcgCsv } from "@/lib/apple-health/ecg-csv"; import { importAppleHealthEcg } from "@/lib/apple-health/ecg-import"; import { extractExportXml } from "@/lib/import/unzip-export-xml"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { invalidateStatusInsightsForTypes } from "@/lib/insights/status-invalidation"; import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; import { streamParseExportXml, @@ -533,6 +534,22 @@ export async function handleAppleHealthImport( ); } + // The imported types' cached status assessments describe data that just + // changed; re-warm them, the second leg every other measurement write + // runs (`afterMeasurementMutation`). The rollups above are refolded over + // the import's span rather than per day, so the leg is called directly. + const importedTypes = Object.entries(result.perType) + .filter(([, stat]) => stat.inserted + stat.updated > 0) + .map(([type]) => type as MeasurementType); + if (importedTypes.length > 0) { + invalidateStatusInsightsForTypes(userId, importedTypes).catch((err) => { + console.warn( + `[apple-health-import] status-insight invalidate failed for user ${userId}`, + err, + ); + }); + } + // Best-effort cleanup. A failed unlink is not fatal — `/tmp` is // periodically swept on the host. safeUnlink(unzip.xmlPath); diff --git a/src/lib/mcp/__tests__/writes.test.ts b/src/lib/mcp/__tests__/writes.test.ts index 6772d4f77..a762f1f4c 100644 --- a/src/lib/mcp/__tests__/writes.test.ts +++ b/src/lib/mcp/__tests__/writes.test.ts @@ -21,8 +21,8 @@ vi.mock("@/lib/cache/invalidate", () => ({ invalidateUserMood: vi.fn(), })); -vi.mock("@/lib/rollups/measurement-rollups", () => ({ - recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined), +vi.mock("@/lib/rollups/after-measurement-mutation", () => ({ + afterMeasurementMutation: vi.fn().mockResolvedValue(undefined), })); vi.mock("@/lib/rollups/mood-rollups", () => ({ @@ -41,6 +41,7 @@ vi.mock("@/lib/logging/context", () => ({ import { logMcpMeasurement, logMcpMood, logMcpBloodPressure } from "../writes"; import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; +import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation"; beforeEach(() => { vi.clearAllMocks(); @@ -465,3 +466,36 @@ describe("MCP writes — idempotency-key stability floor", () => { expect(measurement.create).toHaveBeenCalledTimes(1); }); }); + +describe("MCP writes — post-write tail", () => { + it("refolds and re-warms through the shared helper for a single reading", async () => { + await logMcpMeasurement({ + userId: "u-1", + type: "WEIGHT", + value: 80, + idempotencyKey: "tail-1", + }); + expect(afterMeasurementMutation).toHaveBeenCalledWith( + "u-1", + [expect.objectContaining({ type: "WEIGHT" })], + "mcp", + ); + }); + + it("refolds and re-warms both blood-pressure series", async () => { + await logMcpBloodPressure({ + userId: "u-1", + systolic: 120, + diastolic: 80, + idempotencyKey: "tail-2", + }); + expect(afterMeasurementMutation).toHaveBeenCalledWith( + "u-1", + [ + expect.objectContaining({ type: "BLOOD_PRESSURE_SYS" }), + expect.objectContaining({ type: "BLOOD_PRESSURE_DIA" }), + ], + "mcp", + ); + }); +}); diff --git a/src/lib/mcp/writes.ts b/src/lib/mcp/writes.ts index a25903f19..aa0047673 100644 --- a/src/lib/mcp/writes.ts +++ b/src/lib/mcp/writes.ts @@ -36,7 +36,7 @@ import { invalidateUserMeasurements, invalidateUserMood, } from "@/lib/cache/invalidate"; -import { recomputeBucketsForMeasurement } from "@/lib/rollups/measurement-rollups"; +import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation"; import { recomputeMoodBucketsForEntry } from "@/lib/rollups/mood-rollups"; import { MOOD_ENUM_BY_SCORE } from "@/lib/mood/labels"; import { getScoreForMood } from "@/lib/validations/mood"; @@ -349,15 +349,13 @@ export async function logMcpMeasurement(input: { invalidateUserMeasurements(input.userId, { evict: true }); - // Best-effort rollup refresh — a cache tier, never a write-path invariant. - try { - await recomputeBucketsForMeasurement(input.userId, input.type, measuredAt); - } catch (rollupErr) { - getEvent()?.addMeta( - "mcp_measurement_rollup_failed", - rollupErr instanceof Error ? rollupErr.message : String(rollupErr), - ); - } + // Rollup refresh plus the status re-warm, both best-effort — a cache + // tier, never a write-path invariant. + await afterMeasurementMutation( + input.userId, + [{ type: input.type, measuredAt }], + "mcp", + ); await auditLog("mcp.write.measurement", { userId: input.userId, @@ -522,27 +520,16 @@ export async function logMcpBloodPressure(input: { invalidateUserMeasurements(input.userId, { evict: true }); - // Best-effort rollup refresh for both series — a cache tier, never a - // write-path invariant. - try { - await Promise.all([ - recomputeBucketsForMeasurement( - input.userId, - "BLOOD_PRESSURE_SYS", - measuredAt, - ), - recomputeBucketsForMeasurement( - input.userId, - "BLOOD_PRESSURE_DIA", - measuredAt, - ), - ]); - } catch (rollupErr) { - getEvent()?.addMeta( - "mcp_blood_pressure_rollup_failed", - rollupErr instanceof Error ? rollupErr.message : String(rollupErr), - ); - } + // Rollup refresh plus the status re-warm for both series, best-effort — + // a cache tier, never a write-path invariant. + await afterMeasurementMutation( + input.userId, + [ + { type: "BLOOD_PRESSURE_SYS", measuredAt }, + { type: "BLOOD_PRESSURE_DIA", measuredAt }, + ], + "mcp", + ); await auditLog("mcp.write.blood_pressure", { userId: input.userId, diff --git a/src/lib/measurements/__tests__/create-from-telegram.test.ts b/src/lib/measurements/__tests__/create-from-telegram.test.ts index 7dfdf3141..54fcbd717 100644 --- a/src/lib/measurements/__tests__/create-from-telegram.test.ts +++ b/src/lib/measurements/__tests__/create-from-telegram.test.ts @@ -15,8 +15,8 @@ vi.mock("@/lib/cache/invalidate", () => ({ invalidateUserMeasurements: vi.fn(), })); -vi.mock("@/lib/rollups/measurement-rollups", () => ({ - recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined), +vi.mock("@/lib/rollups/after-measurement-mutation", () => ({ + afterMeasurementMutation: vi.fn().mockResolvedValue(undefined), })); vi.mock("@/lib/logging/context", () => ({ @@ -29,6 +29,7 @@ import { parseTelegramNumber, } from "@/lib/measurements/create-from-telegram"; import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; +import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation"; beforeEach(() => { vi.clearAllMocks(); @@ -90,6 +91,13 @@ describe("logTelegramMeasurement", () => { expect(invalidateUserMeasurements).toHaveBeenCalledWith("user-1", { evict: true, }); + // Rollups refold and the status assessments re-warm through the one + // shared tail. + expect(afterMeasurementMutation).toHaveBeenCalledWith( + "user-1", + [expect.objectContaining({ type: "WEIGHT" })], + "telegram-measurement", + ); }); it("rejects a non-numeric reply without writing", async () => { diff --git a/src/lib/measurements/create-from-telegram.ts b/src/lib/measurements/create-from-telegram.ts index 713a931f9..9bbe84543 100644 --- a/src/lib/measurements/create-from-telegram.ts +++ b/src/lib/measurements/create-from-telegram.ts @@ -24,8 +24,7 @@ import { validateMeasurementRange, } from "@/lib/validations/measurement"; import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; -import { recomputeBucketsForMeasurement } from "@/lib/rollups/measurement-rollups"; -import { getEvent } from "@/lib/logging/context"; +import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation"; import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit"; /** @@ -158,15 +157,13 @@ export async function logTelegramMeasurement(input: { invalidateUserMeasurements(input.userId, { evict: true }); - // Best-effort rollup refresh — a cache tier, never a write-path invariant. - try { - await recomputeBucketsForMeasurement(input.userId, input.type, measuredAt); - } catch (rollupErr) { - getEvent()?.addMeta( - "telegram_measurement_rollup_failed", - rollupErr instanceof Error ? rollupErr.message : String(rollupErr), - ); - } + // Rollup refresh plus the status re-warm, both best-effort — a cache + // tier, never a write-path invariant. + await afterMeasurementMutation( + input.userId, + [{ type: input.type, measuredAt }], + "telegram-measurement", + ); return { status: "ok" }; } diff --git a/src/lib/whoop/__tests__/sync-body.test.ts b/src/lib/whoop/__tests__/sync-body.test.ts index 80db3fd5f..2002fd38b 100644 --- a/src/lib/whoop/__tests__/sync-body.test.ts +++ b/src/lib/whoop/__tests__/sync-body.test.ts @@ -23,6 +23,7 @@ const { update: vi.fn<(...a: unknown[]) => Promise>(async () => ({})), }, measurement: { + findMany: vi.fn<(...a: unknown[]) => Promise>(async () => []), deleteMany: vi.fn<(...a: unknown[]) => Promise>(async () => ({ count: 0, })), @@ -45,6 +46,10 @@ const { vi.mock("@/lib/db", () => ({ prisma: prismaMock })); +vi.mock("@/lib/rollups/after-measurement-mutation", () => ({ + afterMeasurementMutation: vi.fn(async () => {}), +})); + vi.mock("@/lib/logging/context", () => ({ getEvent: () => null, annotate: () => {}, @@ -73,6 +78,7 @@ vi.mock("../sync-core", async (orig) => { import { mapBody } from "../client"; import { syncUserBody, WHOOP_BODY_WEIGHT_EXTERNAL_ID } from "../sync-body"; +import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation"; const TOKEN = { accessToken: "acc", @@ -129,12 +135,22 @@ describe("syncUserBody — profile weight is not ingested (v1.16.11)", () => { expect(upsertWhoopMeasurementsMock).not.toHaveBeenCalled(); }); - it("clears the legacy overwrite row on every pass (idempotent no-op after the first)", async () => { + it("clears the legacy overwrite row and refolds the day it sat on", async () => { fetchBodyMeasurementMock.mockResolvedValue({ weight_kilogram: 80 }); prismaMock.user.findUnique.mockResolvedValue({ heightCm: 170 }); + const legacyRow = { + type: "WEIGHT", + measuredAt: new Date("2026-01-02T08:00:00.000Z"), + }; + prismaMock.measurement.findMany.mockResolvedValueOnce([legacyRow]); await syncUserBody("user1"); + expect(afterMeasurementMutation).toHaveBeenCalledWith( + "user1", + [legacyRow], + "whoop-body", + ); expect(prismaMock.measurement.deleteMany).toHaveBeenCalledWith({ where: { userId: "user1", @@ -249,3 +265,16 @@ describe("syncUserBody — tier degradation", () => { expect(fetchBodyMeasurementMock).not.toHaveBeenCalled(); }); }); + +describe("syncUserBody — legacy row already gone", () => { + it("is a no-op once the legacy row has been removed", async () => { + fetchBodyMeasurementMock.mockResolvedValue({ weight_kilogram: 80 }); + prismaMock.user.findUnique.mockResolvedValue({ heightCm: 170 }); + prismaMock.measurement.findMany.mockResolvedValueOnce([]); + prismaMock.measurement.deleteMany.mockClear(); + + await syncUserBody("user1"); + + expect(prismaMock.measurement.deleteMany).not.toHaveBeenCalled(); + }); +}); diff --git a/src/lib/whoop/sync-body.ts b/src/lib/whoop/sync-body.ts index 0739fbe5a..2600b2687 100644 --- a/src/lib/whoop/sync-body.ts +++ b/src/lib/whoop/sync-body.ts @@ -28,6 +28,7 @@ import { } from "./sync-core"; import { prisma } from "@/lib/db"; import { getEvent } from "@/lib/logging/context"; +import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation"; /** * Stable externalId the retired profile-weight ingestion wrote under — @@ -63,14 +64,22 @@ export async function syncUserBody( // legacy overwrite row once; deleteMany is a no-op after the first // pass, keeping the sync idempotent. try { - await prisma.measurement.deleteMany({ - where: { - userId, - type: "WEIGHT", - source: "WHOOP", - externalId: WHOOP_BODY_WEIGHT_EXTERNAL_ID, - }, + const legacyWhere = { + userId, + type: "WEIGHT" as const, + source: "WHOOP" as const, + externalId: WHOOP_BODY_WEIGHT_EXTERNAL_ID, + }; + // Read the day it sat on before removing it, so that day's rollup is + // refolded without it. + const legacy = await prisma.measurement.findMany({ + where: legacyWhere, + select: { type: true, measuredAt: true }, }); + if (legacy.length > 0) { + await prisma.measurement.deleteMany({ where: legacyWhere }); + await afterMeasurementMutation(userId, legacy, "whoop-body"); + } } catch (err) { getEvent()?.addWarning( `whoop: failed to clear the legacy profile-weight row for ${userId}: ${err}`,