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}`,