diff --git a/src/lib/rollups/__tests__/after-measurement-mutation.test.ts b/src/lib/rollups/__tests__/after-measurement-mutation.test.ts new file mode 100644 index 000000000..6a983e08e --- /dev/null +++ b/src/lib/rollups/__tests__/after-measurement-mutation.test.ts @@ -0,0 +1,129 @@ +/** + * `afterMeasurementMutation` — the shared post-mutation tail every + * measurement write surface calls (rollup recompute + status-insight + * invalidation). + * + * Watched red: with the recompute loop removed from the helper the + * first test fails naming the missing `recomputeBucketsForMeasurement` + * call; with the `.catch` isolation removed the failure-isolation test + * throws. Both verified red against a deliberately broken helper before + * the real implementation landed. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const mocks = vi.hoisted(() => ({ + recompute: vi.fn(), + invalidate: vi.fn(), +})); + +// The helper pulls `collapseToTypeDayKeys` (pure) from the rollup +// module; keep the original for it so the collapse behaviour under +// test is the real one, and stub only the impure recompute. `@/lib/db` +// is stubbed so importing the original module never touches Prisma. +vi.mock("@/lib/db", () => ({ prisma: {} })); + +vi.mock("@/lib/rollups/measurement-rollups", async (importOriginal) => { + const actual = + await importOriginal(); + return { + ...actual, + recomputeBucketsForMeasurement: mocks.recompute, + }; +}); + +vi.mock("@/lib/insights/status-invalidation", () => ({ + invalidateStatusInsightsForTypes: mocks.invalidate, +})); + +import { afterMeasurementMutation } from "../after-measurement-mutation"; +import type { MeasurementType } from "@/generated/prisma/client"; + +beforeEach(() => { + mocks.recompute.mockReset().mockResolvedValue(undefined); + mocks.invalidate.mockReset().mockResolvedValue(undefined); +}); + +const day = (iso: string) => new Date(iso); + +describe("afterMeasurementMutation", () => { + it("recomputes one bucket per distinct (type, day) and invalidates each type once", async () => { + await afterMeasurementMutation("user-1", [ + { + type: "WEIGHT" as MeasurementType, + measuredAt: day("2026-05-10T08:00:00Z"), + }, + { + type: "WEIGHT" as MeasurementType, + measuredAt: day("2026-05-10T21:00:00Z"), + }, + { + type: "PULSE" as MeasurementType, + measuredAt: day("2026-05-10T09:00:00Z"), + }, + ]); + + // Two rows on the same (WEIGHT, day) collapse into one recompute. + expect(mocks.recompute).toHaveBeenCalledTimes(2); + expect(mocks.recompute).toHaveBeenCalledWith( + "user-1", + "WEIGHT", + day("2026-05-10T00:00:00Z"), + ); + expect(mocks.recompute).toHaveBeenCalledWith( + "user-1", + "PULSE", + day("2026-05-10T00:00:00Z"), + ); + + expect(mocks.invalidate).toHaveBeenCalledTimes(1); + const [userId, types] = mocks.invalidate.mock.calls[0]; + expect(userId).toBe("user-1"); + expect([...types].sort()).toEqual(["PULSE", "WEIGHT"]); + }); + + it("is a no-op for an empty row set", async () => { + await afterMeasurementMutation("user-1", []); + expect(mocks.recompute).not.toHaveBeenCalled(); + expect(mocks.invalidate).not.toHaveBeenCalled(); + }); + + it("still invalidates status insights when the rollup recompute fails", async () => { + mocks.recompute.mockRejectedValue(new Error("populator down")); + const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); + try { + await afterMeasurementMutation( + "user-1", + [ + { + type: "WEIGHT" as MeasurementType, + measuredAt: day("2026-05-10T08:00:00Z"), + }, + ], + "test.surface", + ); + } finally { + warn.mockRestore(); + } + // The rollup failure never bubbles AND never gates the second leg. + expect(mocks.invalidate).toHaveBeenCalledTimes(1); + }); + + it("never rejects when the status-insight invalidation fails", async () => { + mocks.invalidate.mockRejectedValue(new Error("cache down")); + const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); + try { + await expect( + afterMeasurementMutation("user-1", [ + { + type: "WEIGHT" as MeasurementType, + measuredAt: day("2026-05-10T08:00:00Z"), + }, + ]), + ).resolves.toBeUndefined(); + // Let the fire-and-forget rejection settle inside the helper's catch. + await new Promise((resolve) => setImmediate(resolve)); + } finally { + warn.mockRestore(); + } + }); +}); diff --git a/src/lib/rollups/__tests__/measurement-rollups.test.ts b/src/lib/rollups/__tests__/measurement-rollups.test.ts index d8792c49d..f67bc8490 100644 --- a/src/lib/rollups/__tests__/measurement-rollups.test.ts +++ b/src/lib/rollups/__tests__/measurement-rollups.test.ts @@ -759,6 +759,30 @@ describe("ensureUserRollupsFresh", () => { expect(queryRawUnsafe).toHaveBeenCalledTimes(1); }); + // Watched red: restoring `deletedAt: null` to the probe's `where` + // makes the exact-match assertion below fail — that filter was the + // bug (a soft-delete bumps `updatedAt` on the tombstone, and hiding + // tombstones from the probe left ghost aggregates reading as fresh). + it("keeps tombstones visible to the freshness probe", async () => { + const rollupAt = new Date("2026-05-10T10:00:00.000Z"); + const deletionAt = new Date("2026-05-10T12:00:00.000Z"); + findFirst.mockResolvedValueOnce({ computedAt: rollupAt }); + // Newest row is a tombstone whose soft-delete bumped `updatedAt` + // past the rollup watermark — the probe must see it as staleness. + findFirstMeasurement.mockResolvedValueOnce({ + updatedAt: deletionAt, + measuredAt: new Date("2026-05-01T08:00:00.000Z"), + }); + queryRawUnsafe.mockResolvedValueOnce([]); + const result = await ensureUserRollupsFresh("user-1"); + expect(result.recomputed).toBe(true); + // Exact `where` match: the probe scopes by user only — no + // `deletedAt` filter that would blind it to deletions. + expect(findFirstMeasurement).toHaveBeenCalledWith( + expect.objectContaining({ where: { userId: "user-1" } }), + ); + }); + it("swallows populator errors so the read path never fails", async () => { findFirst.mockRejectedValueOnce(new Error("pool exhausted")); findFirstMeasurement.mockRejectedValueOnce(new Error("pool exhausted")); diff --git a/src/lib/rollups/after-measurement-mutation.ts b/src/lib/rollups/after-measurement-mutation.ts new file mode 100644 index 000000000..9d121f47a --- /dev/null +++ b/src/lib/rollups/after-measurement-mutation.ts @@ -0,0 +1,55 @@ +/** + * Shared post-mutation tail for every surface that creates, updates, + * deletes, or restores `Measurement` rows. + * + * Two consumers hang off every measurement write and both were being + * re-remembered by hand at each write surface (the documented two-ended + * failure class): + * + * 1. the `(type, day)` rollup buckets the mutation dirtied — recomputed + * inline (DAY) + enqueued (WEEK/MONTH/YEAR) via + * `recomputeBucketsForMeasurement`. The recompute handles deletions + * inherently: a day left empty drops its rollup row. + * 2. the cached per-metric status assessments — re-warmed via + * `invalidateStatusInsightsForTypes` (fire-and-forget). + * + * Both legs are best-effort: a populator hiccup never fails the user's + * mutation. Callers collect the affected `(type, measuredAt)` pairs + * (for deletes: BEFORE the destructive statement, while the rows are + * still readable) and hand them here. + */ +import { + collapseToTypeDayKeys, + recomputeBucketsForMeasurement, +} from "@/lib/rollups/measurement-rollups"; +import { invalidateStatusInsightsForTypes } from "@/lib/insights/status-invalidation"; +import type { MeasurementType } from "@/generated/prisma/client"; + +export async function afterMeasurementMutation( + userId: string, + rows: Array<{ type: MeasurementType; measuredAt: Date }>, + /** Log prefix identifying the calling surface in the warn lines. */ + label = "measurements", +): Promise { + if (rows.length === 0) return; + + // Collapse to the distinct `(type, day)` set BEFORE recomputing so a + // 500-row batch fires at most one DAY recompute per touched bucket. + const keys = collapseToTypeDayKeys(rows); + try { + for (const k of keys) { + await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt); + } + } catch (err) { + console.warn(`[${label}] rollup recompute failed`, err); + } + + // Status-insight re-warm is fire-and-forget — it must run even when + // the rollup leg above failed, so the two legs never gate each other. + const types = Array.from(new Set(keys.map((k) => k.type))); + if (types.length > 0) { + invalidateStatusInsightsForTypes(userId, types).catch((err) => { + console.warn(`[${label}] status-insight invalidate failed`, err); + }); + } +} diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index c3e5d6552..c2aa7c74d 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -911,8 +911,12 @@ async function ensureUserRollupsFreshImpl( orderBy: { computedAt: "desc" }, select: { computedAt: true }, }), + // Deliberately NOT filtered to `deletedAt: null`: a soft-delete + // bumps `updatedAt` on the tombstone, and that bump is exactly the + // staleness signal — filtering tombstones out made a deletion + // invisible to the probe and left the ghost aggregate "fresh". prisma.measurement.findFirst({ - where: { userId, deletedAt: null }, + where: { userId }, orderBy: { updatedAt: "desc" }, select: { updatedAt: true, measuredAt: true }, }), diff --git a/src/lib/whoop/__tests__/webhook-notification.test.ts b/src/lib/whoop/__tests__/webhook-notification.test.ts index bfe90f9cf..2e9db8f19 100644 --- a/src/lib/whoop/__tests__/webhook-notification.test.ts +++ b/src/lib/whoop/__tests__/webhook-notification.test.ts @@ -9,21 +9,37 @@ */ import { beforeEach, describe, expect, it, vi } from "vitest"; -const { findUnique, bossSend, updateMany, deleteMany } = vi.hoisted(() => ({ +const { + findUnique, + bossSend, + updateMany, + deleteMany, + findMany, + afterMutation, +} = vi.hoisted(() => ({ findUnique: vi.fn(), bossSend: vi.fn(), updateMany: vi.fn(), deleteMany: vi.fn(), + findMany: vi.fn(), + afterMutation: vi.fn(), })); vi.mock("@/lib/db", () => ({ prisma: { whoopConnection: { findUnique: (...a: unknown[]) => findUnique(...a) }, - measurement: { updateMany: (...a: unknown[]) => updateMany(...a) }, + measurement: { + updateMany: (...a: unknown[]) => updateMany(...a), + findMany: (...a: unknown[]) => findMany(...a), + }, workout: { deleteMany: (...a: unknown[]) => deleteMany(...a) }, }, })); +vi.mock("@/lib/rollups/after-measurement-mutation", () => ({ + afterMeasurementMutation: (...a: unknown[]) => afterMutation(...a), +})); + vi.mock("@/lib/jobs/boss-instance", () => ({ getGlobalBoss: () => ({ send: (...a: unknown[]) => bossSend(...a) }), })); @@ -44,6 +60,8 @@ beforeEach(() => { vi.clearAllMocks(); findUnique.mockResolvedValue({ userId: "user-1" }); bossSend.mockResolvedValue("job-id"); + findMany.mockResolvedValue([]); + afterMutation.mockResolvedValue(undefined); }); describe("processWhoopNotification — fetch-by-id enqueue", () => { @@ -101,4 +119,32 @@ describe("processWhoopNotification — fetch-by-id enqueue", () => { expect(bossSend).not.toHaveBeenCalled(); expect(deleteMany).toHaveBeenCalled(); }); + + // Watched red: with the `afterMeasurementMutation` call removed from + // the delete branch this test fails on the hook assertion — the + // pre-fix behaviour (soft-delete with no rollup recompute) is exactly + // the ghost-aggregate bug. + it("collects the affected (type, measuredAt) pairs before the soft-delete and fires the rollup hook", async () => { + const measuredAt = new Date("2026-05-10T08:00:00.000Z"); + findMany.mockResolvedValue([{ type: "SLEEP_DURATION", measuredAt }]); + updateMany.mockResolvedValue({ count: 1 }); + + const res = await processWhoopNotification({ + user_id: 42, + id: "s-9", + type: "sleep.deleted", + }); + + expect(res.status).toBe(200); + // The pre-image read happens BEFORE the destructive updateMany — + // afterwards the (type, measuredAt) pairs are unreadable. + expect(findMany.mock.invocationCallOrder[0]).toBeLessThan( + updateMany.mock.invocationCallOrder[0], + ); + expect(afterMutation).toHaveBeenCalledWith( + "user-1", + [{ type: "SLEEP_DURATION", measuredAt }], + "whoop.webhook", + ); + }); }); diff --git a/src/lib/whoop/webhook-handler.ts b/src/lib/whoop/webhook-handler.ts index 9b9eb8e0f..55790ef77 100644 --- a/src/lib/whoop/webhook-handler.ts +++ b/src/lib/whoop/webhook-handler.ts @@ -31,6 +31,8 @@ import { checkRateLimit, rateLimitHeaders } from "@/lib/rate-limit"; import { getClientIp } from "@/lib/api-response"; import { annotate, getEvent } from "@/lib/logging/context"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation"; +import type { MeasurementType } from "@/generated/prisma/client"; /** * Maximum age of a webhook timestamp before it is treated as a replay. Five @@ -163,6 +165,19 @@ export async function processWhoopNotification( // match catches every measurement derived from the deleted resource. if (resourceId) { const now = new Date(); + // Collect the affected `(type, measuredAt)` pairs BEFORE the + // soft-delete — the rollup tail below needs to know which + // `(type, day)` buckets lose rows, and after the updateMany the + // pre-image is gone. + const affected = await prisma.measurement.findMany({ + where: { + userId: connection.userId, + source: "WHOOP", + externalId: { startsWith: `${resourceId}:` }, + deletedAt: null, + }, + select: { type: true, measuredAt: true }, + }); await prisma.measurement.updateMany({ where: { userId: connection.userId, @@ -172,6 +187,18 @@ export async function processWhoopNotification( }, data: { deletedAt: now, syncVersion: { increment: 1 } }, }); + // Recompute the dirtied rollup buckets + drop the affected status + // insights — without this the aggregates keep counting the + // tombstoned rows forever (ghost-aggregate class). Best-effort + // inside the helper; never fails the 200 to WHOOP. + await afterMeasurementMutation( + connection.userId, + affected.map((row) => ({ + type: row.type as MeasurementType, + measuredAt: row.measuredAt, + })), + "whoop.webhook", + ); if (resource === "workout") { // The Workout model carries no soft-delete column, so a deleted // WHOOP workout is removed outright (keyed by the canonical