feat(rollups): make measurement rollups source-aware (migration 0115)

Rollup rows were grouped 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 a single row instead of resolving to
the source-priority ladder's canonical reading.

Re-grain the rollup table to (type, day, source): the writer now groups
by source and delete-then-inserts each affected partition (so a source
that stops reporting a bucket cannot strand a stale row), and the read
path will collapse overlapping sources to the canonical reading in a
follow-up commit. Cumulative metrics (steps, energy) keep summing per
source. Migration 0115 purges the legacy source-blind aggregates; the
boot backfill + read-time self-heal re-mint per-source rows.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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");

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")
 }

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 () => {

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<RollupGranularity, string> = {
 /** 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<RollupRow[]>(
       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<RollupRow[]>(
     sql,
@@ -423,61 +427,77 @@ async function persistRollupRows(
   rows: RollupRow[],
 ): Promise<number> {
   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;
 }

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();