perf(rollups): probe coverage with a loose index scan

probeRollupCoverage runs ahead of the dashboard snapshot, the analytics
slices and the comprehensive insights. It read every live row of the
account for SELECT DISTINCT type and counted every DAY bucket per type,
which the planner served with a sequential scan of measurement_rollups.
Walk the distinct types with a recursive CTE along the live index and
test coverage with EXISTS. On a copy of a long-lived account the probe
drops from 24-50 ms to under 2 ms and returns the same map; an
integration test keeps the old query as the oracle.

src/lib/rollups/measurement-coverage.ts

@@ -36,12 +36,20 @@ export type RollupCoverageMap = Map<string, boolean>;
 
 /**
  * Probe DAY-bucket coverage for every type the user has measurements
- * for. The join is anchored on the smaller `DISTINCT type FROM
- * measurements` set so the planner picks the per-type
- * `(user_id, type, measured_at)` index path; the LEFT JOIN onto the
- * rollup table uses the `(user_id, type, granularity, bucket_start)`
- * composite primary key. A `COUNT > 0` per partition keeps the result
- * shape stable even when a type has zero buckets.
+ * for, in one round-trip.
+ *
+ * The distinct-type set is read as a loose index scan: a recursive CTE
+ * that hops from one type to the next along the partial
+ * `(user_id, type, measured_at)` live index, one index descent per type
+ * instead of a walk over every live row the user has. Coverage is an
+ * `EXISTS` per type on the rollup index, which stops at the first DAY
+ * bucket rather than counting them all.
+ *
+ * The earlier shape, `SELECT DISTINCT type` joined to a `COUNT(r.*)`
+ * over every DAY bucket, read all of a user's live rows and all of their
+ * DAY buckets on each call; on a tenant with years of Apple Health data
+ * the planner chose a sequential scan of `measurement_rollups` for the
+ * join. It returns the same map; only the reads changed.
  */
 export async function probeRollupCoverage(
   userId: string,
@@ -49,20 +57,39 @@ export async function probeRollupCoverage(
   const rows = await prisma.$queryRaw<
     Array<{ type: string; has_buckets: boolean }>
   >`
+    WITH RECURSIVE live_types AS (
+      (
+        SELECT "type"
+        FROM measurements
+        WHERE user_id = ${userId}
+          AND "deleted_at" IS NULL
+        ORDER BY "type"
+        LIMIT 1
+      )
+      UNION ALL
+      SELECT (
+        SELECT m."type"
+        FROM measurements m
+        WHERE m.user_id = ${userId}
+          AND m."deleted_at" IS NULL
+          AND m."type" > live_types."type"
+        ORDER BY m."type"
+        LIMIT 1
+      )
+      FROM live_types
+      WHERE live_types."type" IS NOT NULL
+    )
     SELECT
-      m."type"::text                 AS type,
-      COUNT(r.*) > 0                 AS has_buckets
-    FROM (
-      SELECT DISTINCT "type"
-      FROM measurements
-      WHERE user_id = ${userId}
-        AND "deleted_at" IS NULL
-    ) m
-    LEFT JOIN measurement_rollups r
-      ON  r.user_id     = ${userId}
-      AND r."type"      = m."type"
-      AND r.granularity = 'DAY'
-    GROUP BY m."type"
+      t."type"::text AS type,
+      EXISTS (
+        SELECT 1
+        FROM measurement_rollups r
+        WHERE r.user_id     = ${userId}
+          AND r."type"      = t."type"
+          AND r.granularity = 'DAY'
+      ) AS has_buckets
+    FROM live_types t
+    WHERE t."type" IS NOT NULL
   `;
   const coverage: RollupCoverageMap = new Map();
   for (const row of rows) {

tests/integration/rollup-coverage-probe.test.ts

@@ -0,0 +1,128 @@
+/**
+ * The rollup coverage probe reads the distinct live types as a loose index
+ * scan and coverage as a per-type `EXISTS`. The previous shape (a DISTINCT
+ * over every live row joined to a COUNT over every DAY bucket) is kept here
+ * as the oracle: both must return the same map on the same fixture, including
+ * the edges that decide the read-swap — tombstoned types, types with only
+ * coarser buckets, another account's buckets, and an account with nothing.
+ */
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+import { getPrismaClient, truncateAllTables } from "./setup";
+import { probeRollupCoverage } from "@/lib/rollups/measurement-coverage";
+import type { MeasurementType } from "@/generated/prisma/client";
+
+vi.mock("@/lib/db-compat", () => ({
+  ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
+}));
+vi.mock("@/lib/jobs/boss-instance", () => ({
+  getGlobalBoss: vi.fn(() => null),
+}));
+
+const USER = "user-coverage-probe";
+const OTHER = "user-coverage-probe-other";
+const EMPTY = "user-coverage-probe-empty";
+
+async function oracle(userId: string): Promise<Map<string, boolean>> {
+  const rows = await getPrismaClient().$queryRaw<
+    Array<{ type: string; has_buckets: boolean }>
+  >`
+    SELECT m."type"::text AS type, COUNT(r.*) > 0 AS has_buckets
+    FROM (
+      SELECT DISTINCT "type" FROM measurements
+      WHERE user_id = ${userId} AND "deleted_at" IS NULL
+    ) m
+    LEFT JOIN measurement_rollups r
+      ON r.user_id = ${userId} AND r."type" = m."type" AND r.granularity = 'DAY'
+    GROUP BY m."type"
+  `;
+  return new Map(rows.map((r) => [r.type, Boolean(r.has_buckets)]));
+}
+
+beforeEach(async () => {
+  const prisma = getPrismaClient();
+  await truncateAllTables(prisma);
+  for (const id of [USER, OTHER, EMPTY]) {
+    await prisma.user.create({
+      data: { id, username: id, email: `${id}@example.test`, timezone: "UTC" },
+    });
+  }
+});
+
+describe("probeRollupCoverage", () => {
+  it("matches the full-scan oracle across tombstones, granularities and accounts", async () => {
+    const prisma = getPrismaClient();
+    const day = 86_400_000;
+    const base = Date.UTC(2026, 0, 10);
+    const reading = (
+      userId: string,
+      type: MeasurementType,
+      i: number,
+      deleted = false,
+    ) => ({
+      userId,
+      type,
+      value: 60 + i,
+      unit: "x",
+      measuredAt: new Date(base + i * day),
+      deletedAt: deleted ? new Date(base) : null,
+    });
+    await prisma.measurement.createMany({
+      data: [
+        // Live rows with DAY buckets → covered.
+        ...[0, 1, 2].map((i) => reading(USER, "PULSE", i)),
+        // Live rows, no buckets at all → uncovered.
+        ...[0, 1].map((i) => reading(USER, "WEIGHT", i)),
+        // Live rows with only a WEEK bucket → uncovered.
+        reading(USER, "BLOOD_PRESSURE_SYS", 0),
+        // Only tombstones → absent from the map, even with a DAY bucket.
+        ...[0, 1].map((i) => reading(USER, "SLEEP_DURATION", i, true)),
+        // A mix of tombstoned and live rows → present.
+        reading(USER, "ACTIVITY_STEPS", 0, true),
+        reading(USER, "ACTIVITY_STEPS", 1),
+        // Another account's rows and buckets never leak in.
+        reading(OTHER, "WEIGHT", 0),
+        reading(OTHER, "OXYGEN_SATURATION", 0),
+      ],
+    });
+    const bucket = (
+      userId: string,
+      type: MeasurementType,
+      granularity: "DAY" | "WEEK",
+    ) => ({
+      userId,
+      type,
+      granularity,
+      bucketStart: new Date(base),
+      source: "MANUAL" as const,
+      count: 1,
+      mean: 1,
+      minValue: 1,
+      maxValue: 1,
+    });
+    await prisma.measurementRollup.createMany({
+      data: [
+        bucket(USER, "PULSE", "DAY"),
+        bucket(USER, "BLOOD_PRESSURE_SYS", "WEEK"),
+        bucket(USER, "SLEEP_DURATION", "DAY"),
+        bucket(USER, "ACTIVITY_STEPS", "DAY"),
+        bucket(OTHER, "WEIGHT", "DAY"),
+        bucket(USER, "OXYGEN_SATURATION", "DAY"),
+      ],
+    });
+
+    const coverage = await probeRollupCoverage(USER);
+    expect(Object.fromEntries(coverage)).toEqual({
+      ACTIVITY_STEPS: true,
+      BLOOD_PRESSURE_SYS: false,
+      PULSE: true,
+      WEIGHT: false,
+    });
+    expect(coverage).toEqual(await oracle(USER));
+    expect(await probeRollupCoverage(OTHER)).toEqual(await oracle(OTHER));
+
+    const empty = await probeRollupCoverage(EMPTY);
+    expect(empty.size).toBe(0);
+    expect(empty).toEqual(await oracle(EMPTY));
+  });
+});