diff --git a/src/lib/rollups/measurement-coverage.ts b/src/lib/rollups/measurement-coverage.ts index 04f19bc36..330e0df33 100644 --- a/src/lib/rollups/measurement-coverage.ts +++ b/src/lib/rollups/measurement-coverage.ts @@ -36,12 +36,20 @@ export type RollupCoverageMap = Map; /** * 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) { diff --git a/tests/integration/rollup-coverage-probe.test.ts b/tests/integration/rollup-coverage-probe.test.ts new file mode 100644 index 000000000..2d4b118ee --- /dev/null +++ b/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> { + 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)); + }); +});