diff --git a/prisma/migrations/0115_v1111_source_aware_rollups/migration.sql b/prisma/migrations/0115_v1111_source_aware_rollups/migration.sql new file mode 100644 index 000000000..68e74729f --- /dev/null +++ b/prisma/migrations/0115_v1111_source_aware_rollups/migration.sql @@ -0,0 +1,32 @@ +-- v1.11.1 — source-aware measurement rollups. +-- +-- `measurement_rollups` is a derived cache; the authoritative readings live in +-- `measurements`. Existing rollup rows are SOURCE-BLIND aggregates: they group +-- by (type, day) only, so two sources reporting the same standard vital for one +-- day (e.g. WHOOP + Apple Watch resting heart rate) were AVG-blended into one +-- row. Re-grain the table to (type, day, source) so the read path can collapse +-- overlapping sources to the ladder-canonical reading while cumulative metrics +-- (steps, energy) still sum per source. +-- +-- The existing rows cannot be re-grained in place (their source is unknown — +-- they already blend N sources), so purge them. This is non-destructive of user +-- data: the boot-time rollup backfill (`rollup-full-backfill`) and the +-- read-time self-heal (`ensureUserRollupsFresh` + the /api/measurements +-- coverage fallback) re-mint per-source rows on next worker boot / first read. +-- Bounded, idempotent, multi-tenant-safe; no operator action required. +DELETE FROM "measurement_rollups"; + +-- Additive column. DEFAULT only to satisfy NOT NULL during the (now empty) +-- table rewrite, then dropped so every writer must supply an explicit source. +ALTER TABLE "measurement_rollups" + ADD COLUMN "source" "measurement_source" NOT NULL DEFAULT 'MANUAL'; +ALTER TABLE "measurement_rollups" ALTER COLUMN "source" DROP DEFAULT; + +-- Re-grain the primary key to include source. The hot descending index +-- (user_id, type, granularity, bucket_start DESC) is unchanged: readers fetch +-- every source for a (user, type) range and collapse in application code, so +-- they never filter by source in SQL. +ALTER TABLE "measurement_rollups" DROP CONSTRAINT "measurement_rollups_pkey"; +ALTER TABLE "measurement_rollups" + ADD CONSTRAINT "measurement_rollups_pkey" + PRIMARY KEY ("user_id", "type", "granularity", "bucket_start", "source"); diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 232978bfb..8cb4c38e8 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -821,6 +821,13 @@ model MeasurementRollup { type MeasurementType granularity RollupGranularity bucketStart DateTime @map("bucket_start") @db.Timestamptz(3) + /// v1.11.1 — source-aware grain. Rollup rows are minted per source so the + /// read path can collapse overlapping sources (e.g. WHOOP + Apple Watch + /// resting heart rate) to the ladder-canonical reading, while cumulative + /// metrics (steps, energy) sum per source. Always non-null; migration 0115 + /// purged the legacy source-blind aggregates and the boot backfill re-mints + /// per-source rows. + source MeasurementSource count Int mean Float minValue Float @map("min_value") @@ -839,7 +846,7 @@ model MeasurementRollup { user User @relation(fields: [userId], references: [id], onDelete: Cascade) - @@id([userId, type, granularity, bucketStart]) + @@id([userId, type, granularity, bucketStart, source]) @@index([userId, type, granularity, bucketStart(sort: Desc)]) @@map("measurement_rollups") } diff --git a/src/lib/rollups/__tests__/measurement-rollups.test.ts b/src/lib/rollups/__tests__/measurement-rollups.test.ts index 7a5891898..f96c0018e 100644 --- a/src/lib/rollups/__tests__/measurement-rollups.test.ts +++ b/src/lib/rollups/__tests__/measurement-rollups.test.ts @@ -23,6 +23,7 @@ const mocks = vi.hoisted(() => ({ queryRawUnsafe: vi.fn(), transaction: vi.fn(), upsert: vi.fn(), + createMany: vi.fn(), deleteMany: vi.fn(), findFirst: vi.fn(), findMany: vi.fn(), @@ -38,6 +39,7 @@ vi.mock("@/lib/db", () => ({ $transaction: mocks.transaction, measurementRollup: { upsert: mocks.upsert, + createMany: mocks.createMany, deleteMany: mocks.deleteMany, findFirst: mocks.findFirst, findMany: mocks.findMany, @@ -66,6 +68,7 @@ const { queryRawUnsafe, transaction, upsert, + createMany, deleteMany, findFirst, findFirstMeasurement, @@ -96,6 +99,7 @@ beforeEach(() => { queryRawUnsafe.mockReset(); transaction.mockReset(); upsert.mockReset(); + createMany.mockReset(); deleteMany.mockReset(); findFirst.mockReset(); mocks.findMany.mockReset(); @@ -107,9 +111,14 @@ beforeEach(() => { // assertion. _resetEnsureUserRollupsFreshInFlightForTests(); // Default: $transaction takes an array of pre-built promises (the - // populator passes `slice.map(prisma.measurementRollup.upsert(...))`) - // and returns them awaited. - transaction.mockImplementation(async (operations: unknown[]) => operations); + // populator passes `[deleteMany(...), createMany(...)]`) and resolves + // them like the real client so the destructured create result is the + // resolved value, not a pending promise. + transaction.mockImplementation(async (operations: unknown[]) => + Promise.all(operations), + ); + // v1.11.1 — persistRollupRows writes via createMany; default the count. + createMany.mockResolvedValue({ count: 1 }); }); afterEach(() => { @@ -146,10 +155,11 @@ describe("collapseToTypeDayKeys", () => { }); describe("recomputeBucketsForMeasurement", () => { - it("upserts the DAY rollup synchronously and enqueues WEEK/MONTH/YEAR", async () => { + it("writes the DAY rollup synchronously and enqueues WEEK/MONTH/YEAR", async () => { queryRawUnsafe.mockResolvedValueOnce([ { type: "WEIGHT", + source: "MANUAL", bucket_start: new Date("2026-05-10T00:00:00.000Z"), count: BigInt(3), mean: 82.5, @@ -173,26 +183,24 @@ describe("recomputeBucketsForMeasurement", () => { new Date("2026-05-10T14:30:00.000Z"), ); - // DAY pass — single upsert in a transaction. + // DAY pass — v1.11.1 delete-then-insert the day partition in one tx. expect(transaction).toHaveBeenCalledTimes(1); - expect(upsert).toHaveBeenCalledTimes(1); - const upsertArg = upsert.mock.calls[0][0]; - expect(upsertArg.where.userId_type_granularity_bucketStart.userId).toBe( - "user-1", - ); - expect(upsertArg.where.userId_type_granularity_bucketStart.type).toBe( - "WEIGHT", - ); - expect( - upsertArg.where.userId_type_granularity_bucketStart.granularity, - ).toBe("DAY"); - expect(upsertArg.create.count).toBe(3); - expect(upsertArg.create.mean).toBe(82.5); - // v1.4.39 W-SUM — sumValue flows through on both create + update - // halves of the upsert. Cumulative read paths (steps, flights, - // distance, daylight, active-energy) consume this directly. - expect(upsertArg.create.sumValue).toBe(247.5); - expect(upsertArg.update.sumValue).toBe(247.5); + expect(deleteMany).toHaveBeenCalledTimes(1); + expect(deleteMany.mock.calls[0][0].where.userId).toBe("user-1"); + expect(deleteMany.mock.calls[0][0].where.granularity).toBe("DAY"); + expect(deleteMany.mock.calls[0][0].where.type.in).toContain("WEIGHT"); + expect(createMany).toHaveBeenCalledTimes(1); + const created = createMany.mock.calls[0][0].data; + expect(created).toHaveLength(1); + expect(created[0].userId).toBe("user-1"); + expect(created[0].type).toBe("WEIGHT"); + expect(created[0].granularity).toBe("DAY"); + expect(created[0].source).toBe("MANUAL"); + expect(created[0].count).toBe(3); + expect(created[0].mean).toBe(82.5); + // v1.4.39 W-SUM — sumValue flows through. Cumulative read paths + // (steps, flights, distance, daylight, active-energy) consume it. + expect(created[0].sumValue).toBe(247.5); // WEEK / MONTH / YEAR — three enqueues against the worker queue. expect(bossSend).toHaveBeenCalledTimes(3); @@ -210,6 +218,7 @@ describe("recomputeBucketsForMeasurement", () => { queryRawUnsafe.mockResolvedValueOnce([ { type: "ACTIVITY_STEPS", + source: "APPLE_HEALTH", bucket_start: new Date("2026-05-10T00:00:00.000Z"), count: BigInt(5), mean: 2496, @@ -230,13 +239,13 @@ describe("recomputeBucketsForMeasurement", () => { new Date("2026-05-10T14:30:00.000Z"), ); - expect(upsert).toHaveBeenCalledTimes(1); - const arg = upsert.mock.calls[0][0]; - expect(arg.create.sumValue).toBe(12480); + expect(createMany).toHaveBeenCalledTimes(1); + const arg = createMany.mock.calls[0][0].data[0]; + expect(arg.sumValue).toBe(12480); // mean × count algebraic equivalence — the writer carries SUM // directly because the aggregator computed it once. Asserting // both gives parity coverage for the consumer-side fallback. - expect(arg.create.mean * arg.create.count).toBe(12480); + expect(arg.mean * arg.count).toBe(12480); }); it("passes through null sum_value when the aggregator returns NULL", async () => { @@ -247,6 +256,7 @@ describe("recomputeBucketsForMeasurement", () => { queryRawUnsafe.mockResolvedValueOnce([ { type: "WEIGHT", + source: "MANUAL", bucket_start: new Date("2026-05-10T00:00:00.000Z"), count: BigInt(1), mean: 82.5, @@ -266,9 +276,8 @@ describe("recomputeBucketsForMeasurement", () => { new Date("2026-05-10T14:30:00.000Z"), ); - const arg = upsert.mock.calls[0][0]; - expect(arg.create.sumValue).toBeNull(); - expect(arg.update.sumValue).toBeNull(); + const arg = createMany.mock.calls[0][0].data[0]; + expect(arg.sumValue).toBeNull(); }); it("deletes the DAY row when the post-mutation aggregate is empty", async () => { diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index e8c95bc94..8457ed781 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -32,6 +32,7 @@ import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { annotate } from "@/lib/logging/context"; import { startOfUtcDay } from "@/lib/tz/start-of-utc-day"; import type { + MeasurementSource, MeasurementType, RollupGranularity, } from "@/generated/prisma/client"; @@ -102,6 +103,7 @@ const DATE_TRUNC_UNIT: Record = { /** Raw row shape returned by the aggregate `$queryRaw`. */ interface RollupRow { type: string; + source: string; bucket_start: Date; count: bigint; mean: number | null; @@ -118,10 +120,10 @@ interface RollupRow { * `types` / `granularities` / `from` / `to` filters; defaults cover * every type, every granularity, and the trailing 5 years. * - * Idempotent: rows are upserted under the - * `(userId, type, granularity, bucketStart)` composite primary key, - * so re-running the call is a no-op for unchanged buckets and a - * refresh for changed ones. + * Idempotent: rows are delete-then-inserted per source under the + * `(userId, type, granularity, bucketStart, source)` composite primary + * key, so re-running the call refreshes the affected buckets and drops + * any per-source row whose source no longer reports that bucket. */ export async function recomputeUserRollups( userId: string, @@ -347,6 +349,7 @@ async function runRollupAggregate(input: { const sql = ` SELECT m."type"::text AS type, + m."source"::text AS source, ${dateTrunc} AS bucket_start, COUNT(*) AS count, AVG(m."value")::double precision AS mean, @@ -368,7 +371,7 @@ async function runRollupAggregate(input: { AND m."measured_at" >= $2 AND m."measured_at" < $3 AND m."deleted_at" IS NULL - GROUP BY m."type", ${dateTrunc} + GROUP BY m."type", m."source", ${dateTrunc} `; return prisma.$queryRawUnsafe( sql, @@ -381,6 +384,7 @@ async function runRollupAggregate(input: { const sql = ` SELECT m."type"::text AS type, + m."source"::text AS source, ${dateTrunc} AS bucket_start, COUNT(*) AS count, AVG(m."value")::double precision AS mean, @@ -401,7 +405,7 @@ async function runRollupAggregate(input: { AND m."measured_at" >= $2 AND m."measured_at" < $3 AND m."deleted_at" IS NULL - GROUP BY m."type", ${dateTrunc} + GROUP BY m."type", m."source", ${dateTrunc} `; return prisma.$queryRawUnsafe( sql, @@ -423,61 +427,77 @@ async function persistRollupRows( rows: RollupRow[], ): Promise { if (rows.length === 0) return 0; - // Chunk to keep the parameter count below Postgres' 65k cap and - // the transaction round-trip count bounded for large backfills. - // For the typical hot path (a single DAY recompute = 1 row, a - // multi-week WEEK recompute = a few rows) the round-trip count is - // negligible. + + // v1.11.1 — rows are now minted per source. Delete-then-insert the affected + // (type, bucket) partitions across ALL sources before writing, so a source + // that disappeared from a bucket (the user disconnected a provider, or + // deleted the last reading from one device) does not strand a stale + // per-source row that a plain upsert would never revisit. The delete is + // scoped to the [min, max] bucket range of the rows being written, per type + // — an indexed range delete, cheap on the single-day write hook and bounded + // on the full backfill. + const types = [...new Set(rows.map((r) => r.type as MeasurementType))]; + let minBucket = rows[0].bucket_start; + let maxBucket = rows[0].bucket_start; + for (const r of rows) { + if (r.bucket_start < minBucket) minBucket = r.bucket_start; + if (r.bucket_start > maxBucket) maxBucket = r.bucket_start; + } + const deleteWhere = { + userId, + granularity, + type: { in: types }, + bucketStart: { gte: minBucket, lte: maxBucket }, + }; + + const toData = (row: RollupRow) => ({ + userId, + type: row.type as MeasurementType, + granularity, + bucketStart: row.bucket_start, + source: row.source as MeasurementSource, + count: Number(row.count), + mean: row.mean ?? 0, + minValue: row.min_value ?? 0, + maxValue: row.max_value ?? 0, + // v1.4.39 W-SUM — populate for every type. Cumulative read paths (steps, + // flights, distance, daylight, active-energy) consume this column + // directly; spot metrics carry it for free because the underlying SUM + // aggregator runs in the same query as AVG / MIN / MAX. + sumValue: row.sum_value ?? null, + sd: row.sd, + slope: row.slope, + r2: row.r2, + computedAt: new Date(), + }); + + // Chunk to keep the parameter count below Postgres' 65k cap and the + // round-trip count bounded for large backfills. const CHUNK = 500; + + // Common path — the entire write fits one chunk (every incremental hook + // and most bucket-span recomputes). Atomic delete-then-insert so a + // concurrent read never sees a half-rewritten partition. + if (rows.length <= CHUNK) { + const [, created] = await prisma.$transaction([ + prisma.measurementRollup.deleteMany({ where: deleteWhere }), + prisma.measurementRollup.createMany({ data: rows.map(toData) }), + ]); + return created.count; + } + + // Large backfill — delete the partition range once, then insert in chunks. + // The brief gap between delete and the first insert can only make a + // concurrent reader fall back to live SQL for those buckets (correct + // value), never serve a wrong one. + await prisma.measurementRollup.deleteMany({ where: deleteWhere }); let touched = 0; for (let i = 0; i < rows.length; i += CHUNK) { const slice = rows.slice(i, i + CHUNK); - await prisma.$transaction( - slice.map((row) => - prisma.measurementRollup.upsert({ - where: { - userId_type_granularity_bucketStart: { - userId, - type: row.type as MeasurementType, - granularity, - bucketStart: row.bucket_start, - }, - }, - create: { - userId, - type: row.type as MeasurementType, - granularity, - bucketStart: row.bucket_start, - count: Number(row.count), - mean: row.mean ?? 0, - minValue: row.min_value ?? 0, - maxValue: row.max_value ?? 0, - // v1.4.39 W-SUM — populate for every type. Cumulative read - // paths (steps, flights, distance, daylight, active-energy) - // consume this column directly; spot metrics carry it for - // free because the underlying SUM aggregator runs in the - // same query as AVG / MIN / MAX. - sumValue: row.sum_value ?? null, - sd: row.sd, - slope: row.slope, - r2: row.r2, - computedAt: new Date(), - }, - update: { - count: Number(row.count), - mean: row.mean ?? 0, - minValue: row.min_value ?? 0, - maxValue: row.max_value ?? 0, - sumValue: row.sum_value ?? null, - sd: row.sd, - slope: row.slope, - r2: row.r2, - computedAt: new Date(), - }, - }), - ), - ); - touched += slice.length; + const res = await prisma.measurementRollup.createMany({ + data: slice.map(toData), + }); + touched += res.count; } return touched; } diff --git a/tests/integration/measurement-soft-delete.test.ts b/tests/integration/measurement-soft-delete.test.ts index e36810713..c33a78bbe 100644 --- a/tests/integration/measurement-soft-delete.test.ts +++ b/tests/integration/measurement-soft-delete.test.ts @@ -289,14 +289,15 @@ describe("Measurement.deletedAt — tombstone invisibility (v1.4.40 W-DELETED)", noonUtc.getUTCDate(), ), ); - const bucket = await prisma.measurementRollup.findUnique({ + // v1.11.1 — rollups are source-aware; both seeded rows share the default + // MANUAL source so they still collapse to one DAY row. findFirst keeps the + // assertion source-agnostic. + const bucket = await prisma.measurementRollup.findFirst({ where: { - userId_type_granularity_bucketStart: { - userId: user.id, - type: "WEIGHT", - granularity: "DAY", - bucketStart: dayStartUtc, - }, + userId: user.id, + type: "WEIGHT", + granularity: "DAY", + bucketStart: dayStartUtc, }, }); expect(bucket).not.toBeNull();