fix(rollups): let a recompute replace every bucket in its window, not just the ones it recomputed

persistRollupRows scoped its delete to the [min, max] bucket range of the rows
the aggregate RETURNED. That made the repair self-limiting in exactly the case
that needs it: a recompute could only overwrite a bucket the live aggregate
still produces, so a bucket whose source rows had all gone was never in the
delete range and kept its stale aggregate.

Three shapes of the same hole. An empty aggregate skipped the write entirely
(the caller short-circuited on rows.length === 0). A returned range narrower
than the requested window left the edge buckets untouched. A type that lost
every reading in the window dropped out of the returned rows, so the per-type
delete filter could never reach it.

The DAY write hook masked it — it deletes its own bucket by hand when the day
empties — but nothing covered the WEEK / MONTH / YEAR worker, which recomputes
one bucket span at a time: deleting the last reading of a type in a week
dropped the DAY bucket and left the coarser ones stale forever. The coverage
probe still reported hasBuckets = true, so the read-swap served them.

A recompute now prunes the window it was asked for and writes its rows in the
same transaction. The single-bucket write hook keeps its own scope, which is
correct there because it already deletes its bucket when the aggregate empties.
Still one tier: the rollup replaces the legacy read and falls back to live SQL
only on a coverage miss.

src/lib/rollups/__tests__/measurement-rollups.test.ts

@@ -567,6 +567,124 @@ describe("persistRollupRows — large path (via recomputeUserRollups)", () => {
   });
 });
 
+/**
+ * A recompute REPLACES every bucket in the window it was asked for.
+ *
+ * Scoping the delete to the [min, max] range of the rows the aggregate
+ * RETURNED made the repair self-limiting in exactly the case that needs it: a
+ * bucket whose source rows all disappeared is not in the returned range, so it
+ * survived with its stale aggregate while `probeRollupCoverage` still reported
+ * `hasBuckets = true` and the read-swap served it instead of falling back to
+ * live SQL. The write hook masked this for DAY (it deletes its own bucket by
+ * hand when the day empties) but nothing covered the WEEK / MONTH / YEAR
+ * worker, which recomputes one bucket span at a time.
+ */
+describe("recomputeUserRollups — prunes the whole recomputed window", () => {
+  function rowsInBucket(bucketStart: Date, type = "WEIGHT") {
+    return [
+      {
+        type,
+        source: "MANUAL",
+        bucket_start: bucketStart,
+        count: BigInt(1),
+        mean: 80,
+        min_value: 80,
+        max_value: 80,
+        sum_value: 80,
+        sd: 0,
+        slope: 0,
+        r2: 0,
+      },
+    ];
+  }
+
+  it("deletes the stale bucket when the window's last reading is gone", async () => {
+    // The exact damage: the user deleted their last WEIGHT reading of the
+    // week, so the WEEK aggregate for that span now returns nothing.
+    queryRawUnsafe.mockResolvedValueOnce([]);
+
+    await recomputeUserRollups("user-1", {
+      granularities: ["WEEK"],
+      types: ["WEIGHT"],
+      from: new Date(Date.UTC(2024, 2, 4)),
+      to: new Date(Date.UTC(2024, 2, 11)),
+    });
+
+    expect(deleteMany).toHaveBeenCalledTimes(1);
+    const where = deleteMany.mock.calls[0][0].where;
+    expect(where).toMatchObject({
+      userId: "user-1",
+      granularity: "WEEK",
+      type: { in: ["WEIGHT"] },
+    });
+    expect(where.bucketStart.gte.toISOString()).toBe(
+      "2024-03-04T00:00:00.000Z",
+    );
+    expect(where.bucketStart.lt.toISOString()).toBe("2024-03-11T00:00:00.000Z");
+    // Nothing to write — the window is authoritatively empty.
+    expect(txExecuteRawUnsafe).not.toHaveBeenCalled();
+  });
+
+  it("covers the requested window, not just the range the aggregate returned", async () => {
+    // Readings survive only in the middle of the window; the buckets at both
+    // edges lost theirs. The delete must still reach the edges.
+    queryRawUnsafe.mockResolvedValueOnce(
+      rowsInBucket(new Date(Date.UTC(2024, 2, 10))),
+    );
+    txExecuteRawUnsafe.mockResolvedValueOnce(1);
+
+    await recomputeUserRollups("user-1", {
+      granularities: ["DAY"],
+      from: new Date(Date.UTC(2024, 2, 1)),
+      to: new Date(Date.UTC(2024, 2, 20)),
+    });
+
+    const where = deleteMany.mock.calls[0][0].where;
+    expect(where.bucketStart.gte.toISOString()).toBe(
+      "2024-03-01T00:00:00.000Z",
+    );
+    expect(where.bucketStart.lt.toISOString()).toBe("2024-03-20T00:00:00.000Z");
+  });
+
+  it("prunes every type when the recompute was not type-scoped", async () => {
+    // A type that lost all its readings drops out of the returned rows
+    // entirely, so a `type: { in: <returned types> }` filter could never
+    // reach its buckets.
+    queryRawUnsafe.mockResolvedValueOnce(
+      rowsInBucket(new Date(Date.UTC(2024, 2, 10))),
+    );
+    txExecuteRawUnsafe.mockResolvedValueOnce(1);
+
+    await recomputeUserRollups("user-1", {
+      granularities: ["DAY"],
+      from: new Date(Date.UTC(2024, 2, 1)),
+      to: new Date(Date.UTC(2024, 2, 20)),
+    });
+
+    const where = deleteMany.mock.calls[0][0].where;
+    expect(where.type).toBeUndefined();
+  });
+
+  it("still writes the fresh rows it did compute", async () => {
+    queryRawUnsafe.mockResolvedValueOnce(
+      rowsInBucket(new Date(Date.UTC(2024, 2, 10))),
+    );
+    txExecuteRawUnsafe.mockResolvedValueOnce(1);
+
+    const result = await recomputeUserRollups("user-1", {
+      granularities: ["DAY"],
+      from: new Date(Date.UTC(2024, 2, 1)),
+      to: new Date(Date.UTC(2024, 2, 20)),
+    });
+
+    // Delete and upsert share the one interactive transaction, so a reader
+    // never observes the pruned window without its replacement rows.
+    expect(transaction).toHaveBeenCalledTimes(1);
+    expect(txExecuteRawUnsafe).toHaveBeenCalledTimes(1);
+    expect(result.rowsUpserted).toBe(1);
+  });
+});
+
 describe("enqueueRollupRecompute", () => {
   it("is a silent no-op when no boss is attached", async () => {
     getGlobalBossMock.mockReturnValue(null);

src/lib/rollups/measurement-rollups.ts

@@ -26,6 +26,13 @@
  *     side (`analytics`, `comprehensive`, `summaries-slice`) decides
  *     whether to fall through to the live aggregator + persist-on-miss
  *     via its own per-tier freshness probe.
+ *
+ * Repair contract: a recompute REPLACES every bucket in the window it was
+ * asked for. That includes buckets the live aggregate returns nothing for —
+ * without it, a bucket outlives the rows it was folded from. See the
+ * `deleteWhere` note in `persistRollupRows` for the three shapes that hole
+ * took. Nothing here reads both tiers; the rollup replaces the legacy read and
+ * falls back to live SQL only on a coverage miss.
  */
 import { prisma } from "@/lib/db";
 import { getGlobalBoss } from "@/lib/jobs/boss-instance";
@@ -217,9 +224,15 @@ export async function recomputeUserRollups(
       from: alignedFrom,
       to: alignedTo,
     });
-    if (rows.length === 0) continue;
-
-    rowsUpserted += await persistRollupRows(userId, granularity, rows);
+    // No `continue` on an empty aggregate: an empty result is a STATEMENT —
+    // this window holds no readings any more — and the prune below is what
+    // acts on it. Skipping the write here is what let a bucket outlive its
+    // source rows.
+    rowsUpserted += await persistRollupRows(userId, granularity, rows, {
+      from: alignedFrom,
+      to: alignedTo,
+      types: opts.types,
+    });
   }
 
   return { rowsUpserted, durationMs: Date.now() - startedAt };
@@ -263,7 +276,9 @@ export async function recomputeBucketsForMeasurement(
       },
     });
   } else {
-    await persistRollupRows(userId, "DAY", rows);
+    // No prune scope: this call owns exactly one (type, day) bucket and the
+    // branch above already deletes it by hand when the day emptied out.
+    await persistRollupRows(userId, "DAY", rows, null);
   }
 
   // WEEK / MONTH / YEAR — enqueue. Each granularity covers a range
@@ -530,40 +545,102 @@ async function runRollupAggregate(input: {
 }
 
 /**
- * UPSERT the aggregate rows into `measurement_rollups`. Returns the
- * number of rows touched. Uses a chunked raw insert with
- * `ON CONFLICT DO UPDATE` so the round-trip count is bounded
- * regardless of fan-out width AND a concurrent writer on the same
- * partition can never raise a unique violation.
+ * The window a recompute authoritatively re-derived. Every rollup bucket inside
+ * it is replaced by the freshly computed rows — including buckets the aggregate
+ * returned NOTHING for, which is the only way a bucket whose source rows all
+ * disappeared can be removed. `types` narrows the prune to the types the
+ * aggregate was scoped to; `undefined` means it covered every type.
  */
-async function persistRollupRows(
+interface PruneScope {
+  from: Date;
+  to: Date;
+  types?: MeasurementType[];
+}
+
+/**
+ * Delete scope for a caller that wrote a single known bucket rather than a
+ * re-derived window: the [min, max] bucket range of the rows themselves, per
+ * type. Correct there precisely because the caller already knows which bucket
+ * it touched and deletes it by hand when the aggregate comes back empty.
+ */
+function buildReturnedRangeDeleteWhere(
   userId: string,
   granularity: RollupGranularity,
   rows: RollupRow[],
-): Promise<number> {
-  if (rows.length === 0) return 0;
-
-  // v1.11.1 — rows are now minted per source. Delete-then-upsert 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;
+  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 = {
+  return {
     userId,
     granularity,
     type: { in: types },
     bucketStart: { gte: minBucket, lte: maxBucket },
   };
+}
+
+/**
+ * UPSERT the aggregate rows into `measurement_rollups`. Returns the
+ * number of rows touched. Uses a chunked raw insert with
+ * `ON CONFLICT DO UPDATE` so the round-trip count is bounded
+ * regardless of fan-out width AND a concurrent writer on the same
+ * partition can never raise a unique violation.
+ *
+ * `prune` is the recomputed window when the caller re-derived a whole span
+ * (`recomputeUserRollups`), or `null` when it wrote a single known bucket and
+ * handles its own empty case (`recomputeBucketsForMeasurement`). See the
+ * `deleteWhere` note below for why the distinction matters.
+ */
+async function persistRollupRows(
+  userId: string,
+  granularity: RollupGranularity,
+  rows: RollupRow[],
+  prune: PruneScope | null,
+): Promise<number> {
+  // v1.11.1 — rows are now minted per source. Delete-then-upsert 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 RECOMPUTED WINDOW when the caller supplies one,
+  // not to the [min, max] range of the rows coming back. Scoping it to the
+  // returned rows made the repair self-limiting in exactly the case that needs
+  // it: a recompute can only ever overwrite a bucket the live aggregate still
+  // produces, so a bucket whose source rows all disappeared was never in the
+  // delete range and survived with its stale aggregate. Three shapes of the
+  // same hole — an empty aggregate skipped the write entirely, a returned range
+  // narrower than the requested window left the edges untouched, and a type
+  // that lost every reading dropped out of `type: { in: … }` altogether. The
+  // WEEK / MONTH / YEAR worker hit all three: deleting the last reading of a
+  // type in a week dropped the DAY bucket (the write hook deletes that one by
+  // hand) while the coarser buckets kept the stale aggregate forever, and the
+  // coverage probe still reported `hasBuckets = true` so the read-swap served
+  // them instead of falling back to live SQL.
+  const pruneWhere = prune
+    ? {
+        userId,
+        granularity,
+        ...(prune.types ? { type: { in: prune.types } } : {}),
+        bucketStart: { gte: prune.from, lt: prune.to },
+      }
+    : null;
+
+  if (rows.length === 0) {
+    // Nothing to write. With a prune scope there is still work to do: the
+    // window is authoritatively empty, so every bucket in it must go. No
+    // transaction needed — there is no insert leg that could leave a gap.
+    if (!pruneWhere) return 0;
+    await prisma.measurementRollup.deleteMany({ where: pruneWhere });
+    return 0;
+  }
+
+  const deleteWhere =
+    pruneWhere ?? buildReturnedRangeDeleteWhere(userId, granularity, rows);
 
   // v1.28.33 (issue #486) — the insert leg is a raw
   // `INSERT … ON CONFLICT (pk) DO UPDATE` instead of `createMany`. The