diff --git a/src/lib/__tests__/db.test.ts b/src/lib/__tests__/db.test.ts index baf6e00f0..8ed282b60 100644 --- a/src/lib/__tests__/db.test.ts +++ b/src/lib/__tests__/db.test.ts @@ -2,7 +2,7 @@ * Pool configuration stays unit-tested here so connection ceilings and wait * timeouts cannot silently fall back to separate library defaults. */ -import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { buildSessionOptions, @@ -129,3 +129,32 @@ describe("buildSessionOptions", () => { expect(buildSessionOptions()).toBeUndefined(); }); }); + +describe("process-wide Prisma client", () => { + const globalSlot = globalThis as unknown as { prisma?: unknown }; + + afterEach(() => { + vi.unstubAllEnvs(); + delete globalSlot.prisma; + vi.resetModules(); + }); + + it("shares one client across separate copies of the module in production", async () => { + // A production build loads this module once per server bundle; each copy + // must resolve the same client rather than constructing its own. + vi.stubEnv("NODE_ENV", "production"); + vi.stubEnv( + "DATABASE_URL", + "postgresql://healthlog:test@db:5432/healthlog?connection_limit=20", + ); + delete globalSlot.prisma; + + vi.resetModules(); + const first = (await import("../db")).prisma; + vi.resetModules(); + const second = (await import("../db")).prisma; + + expect(second).toBe(first); + expect(globalSlot.prisma).toBe(first); + }); +}); diff --git a/src/lib/cache/invalidate.ts b/src/lib/cache/invalidate.ts index e770551f3..40157bb1b 100644 --- a/src/lib/cache/invalidate.ts +++ b/src/lib/cache/invalidate.ts @@ -18,6 +18,8 @@ * the write endpoint's `apiSuccess()` return. No `await` overhead. */ +import { forgetCoverageProbe } from "@/lib/rollups/coverage-inflight"; + import { caches } from "./server-cache"; /** @@ -68,6 +70,9 @@ export function invalidateUserMeasurements( userId: string, opts?: { evict?: boolean }, ): void { + // A coverage probe already running read the set before this write; the + // next reader must not join it. + forgetCoverageProbe(userId); // The `${userId}|` prefix covers the slim / thick analytics cells, the // iOS summary cell, AND the v1.7.0 dashboard snapshot // (`${userId}|dashboard-snapshot`) in one pass. diff --git a/src/lib/db.ts b/src/lib/db.ts index 7b46e69ad..86b45c035 100644 --- a/src/lib/db.ts +++ b/src/lib/db.ts @@ -106,11 +106,16 @@ function createPrismaClient() { return new PrismaClient({ adapter }); } +// One client per process, in production too. A production build gives every +// server bundle its own copy of this module (the route handlers, the RSC/SSR +// renderer, and the instrumentation bundle that boots the workers), so a +// module-scoped client alone meant three of them: three query-compiler states +// of roughly 28 MB each on a heap that V8 caps near 520 MB in a 1 GiB +// container, and three pg pools each sized to the whole per-process +// connection budget above. Parking the client on `globalThis` makes every +// copy of the module resolve the same one. export const prisma = globalForPrisma.prisma ?? createPrismaClient(); - -if (process.env.NODE_ENV !== "production") { - globalForPrisma.prisma = prisma; -} +globalForPrisma.prisma = prisma; // Prisma's `InputJsonValue` requires an explicit index signature that // typed application shapes (Zod-validated, hand-written interfaces) diff --git a/src/lib/rollups/__tests__/coverage-inflight.test.ts b/src/lib/rollups/__tests__/coverage-inflight.test.ts new file mode 100644 index 000000000..55154e91f --- /dev/null +++ b/src/lib/rollups/__tests__/coverage-inflight.test.ts @@ -0,0 +1,83 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const { queryRaw } = vi.hoisted(() => ({ queryRaw: vi.fn() })); +vi.mock("@/lib/db", () => ({ prisma: { $queryRaw: queryRaw } })); + +import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; +import { probeRollupCoverage } from "../measurement-coverage"; + +function deferred() { + let resolve!: (value: T) => void; + let reject!: (err: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +} + +beforeEach(() => { + queryRaw.mockReset(); +}); + +describe("probeRollupCoverage single-flight", () => { + it("shares one query between concurrent probes of an account", async () => { + const rows = deferred>(); + queryRaw.mockReturnValueOnce(rows.promise); + + const a = probeRollupCoverage("u1"); + const b = probeRollupCoverage("u1"); + rows.resolve([ + { type: "PULSE", has_buckets: true }, + { type: "WEIGHT", has_buckets: false }, + ]); + const [ma, mb] = await Promise.all([a, b]); + + expect(queryRaw).toHaveBeenCalledTimes(1); + expect(Object.fromEntries(ma)).toEqual({ PULSE: true, WEIGHT: false }); + expect(mb).toEqual(ma); + // Each caller owns its map. + expect(mb).not.toBe(ma); + ma.set("PULSE", false); + expect(mb.get("PULSE")).toBe(true); + }); + + it("keeps nothing once the query settles, and never joins across accounts", async () => { + queryRaw.mockResolvedValue([{ type: "PULSE", has_buckets: true }]); + await probeRollupCoverage("u1"); + await probeRollupCoverage("u1"); + await Promise.all([probeRollupCoverage("u1"), probeRollupCoverage("u2")]); + expect(queryRaw).toHaveBeenCalledTimes(4); + }); + + it("starts afresh after a measurement write", async () => { + const first = deferred>(); + queryRaw + .mockReturnValueOnce(first.promise) + .mockResolvedValueOnce([{ type: "WEIGHT", has_buckets: false }]); + + const before = probeRollupCoverage("u1"); + invalidateUserMeasurements("u1"); + const after = probeRollupCoverage("u1"); + first.resolve([]); + + expect((await before).size).toBe(0); + expect(Object.fromEntries(await after)).toEqual({ WEIGHT: false }); + expect(queryRaw).toHaveBeenCalledTimes(2); + }); + + it("passes a failure to every joined caller and does not keep it", async () => { + const failing = deferred>(); + queryRaw + .mockReturnValueOnce(failing.promise) + .mockResolvedValueOnce([{ type: "PULSE", has_buckets: true }]); + + const a = probeRollupCoverage("u1"); + const b = probeRollupCoverage("u1"); + failing.reject(new Error("db down")); + await expect(a).rejects.toThrow("db down"); + await expect(b).rejects.toThrow("db down"); + + expect((await probeRollupCoverage("u1")).get("PULSE")).toBe(true); + }); +}); diff --git a/src/lib/rollups/coverage-inflight.ts b/src/lib/rollups/coverage-inflight.ts new file mode 100644 index 000000000..f62666469 --- /dev/null +++ b/src/lib/rollups/coverage-inflight.ts @@ -0,0 +1,48 @@ +/** + * In-flight registry for `probeRollupCoverage`, kept in its own module so the + * cache invalidators can reach it without importing the rollup read path. + * + * A single Coach snapshot, derived-metric batch or Insights page fans out into + * a dozen readers that each probe the same account's coverage at the same + * moment; one MCP `get_metrics` call was seen holding 59 of these probes open + * on the database at once. Concurrent probes for one account now share the + * query that is already running. Nothing is kept once it settles, and a + * measurement write drops the entry, so a probe that starts after a write + * always reads afresh. + */ + +// Parked on `globalThis` like the Prisma client: a production build loads this +// module once per server bundle, and a write handled in one bundle must reach +// a probe started in another. +const registry = globalThis as unknown as { + __healthlogCoverageInFlight?: Map>>; +}; +const inFlight: Map< + string, + Promise> +> = (registry.__healthlogCoverageInFlight ??= new Map()); + +/** Join the probe already running for `userId`, or start `run` as the one. */ +export function joinCoverageProbe( + userId: string, + run: () => Promise>, +): Promise> { + let shared = inFlight.get(userId); + if (!shared) { + const started = run(); + shared = started; + inFlight.set(userId, started); + const release = () => { + if (inFlight.get(userId) === started) inFlight.delete(userId); + }; + started.then(release, release); + } + // Each caller gets its own map: a reader that adjusts its copy must not + // change what the others see. + return shared.then((coverage) => new Map(coverage)); +} + +/** Called on a measurement write: the next probe for the account starts afresh. */ +export function forgetCoverageProbe(userId: string): void { + inFlight.delete(userId); +} diff --git a/src/lib/rollups/measurement-coverage.ts b/src/lib/rollups/measurement-coverage.ts index 04f19bc36..9a28ff997 100644 --- a/src/lib/rollups/measurement-coverage.ts +++ b/src/lib/rollups/measurement-coverage.ts @@ -20,6 +20,7 @@ * read regardless of how many types the user has logged. */ import { prisma } from "@/lib/db"; +import { joinCoverageProbe } from "@/lib/rollups/coverage-inflight"; import { ROLLUP_FOLD_WINDOW_MS } from "@/lib/rollups/measurement-rollups"; /** @@ -36,33 +37,67 @@ 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. + * + * Concurrent probes for one account share the query already in flight (see + * `coverage-inflight.ts`); each caller still receives its own map. */ -export async function probeRollupCoverage( +export function probeRollupCoverage( userId: string, ): Promise { + return joinCoverageProbe(userId, () => readRollupCoverage(userId)); +} + +async function readRollupCoverage(userId: string): Promise { 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/src/lib/tz/__tests__/zone-offset.test.ts b/src/lib/tz/__tests__/zone-offset.test.ts new file mode 100644 index 000000000..6c411738f --- /dev/null +++ b/src/lib/tz/__tests__/zone-offset.test.ts @@ -0,0 +1,164 @@ +/** + * The hourly offset memo must give the same wall clock as asking Intl + * directly, for every instant, including the hours around a transition and + * zones whose transitions do not fall on a UTC hour. + */ +import { beforeEach, describe, expect, it } from "vitest"; + +import { formatInUserTz, tzOffsetMinutes, userDayKey } from "../format"; +import { wallClockInTz } from "../wall-clock"; +import { __resetZoneOffsetCacheForTests, zoneOffsetMs } from "../zone-offset"; + +const ZONES = [ + "UTC", + "Europe/Berlin", + "America/New_York", + "America/St_Johns", + "Australia/Lord_Howe", + "Pacific/Chatham", + "Asia/Kathmandu", + "Asia/Kolkata", + "Africa/Casablanca", + "America/Sao_Paulo", +]; + +const REFERENCE_OPTIONS: Intl.DateTimeFormatOptions = { + year: "numeric", + month: "2-digit", + day: "2-digit", + hour: "2-digit", + minute: "2-digit", + second: "2-digit", + hour12: false, + weekday: "short", +}; +const WEEKDAYS = ["Sun", "Mon", "Tue", "Wed", "Thu", "Fri", "Sat"]; + +function reference(date: Date, tz: string) { + const parts = new Intl.DateTimeFormat("en-US", { + ...REFERENCE_OPTIONS, + timeZone: tz, + }).formatToParts(date); + const get = (type: string) => parts.find((p) => p.type === type)!.value; + const hour = Number(get("hour")) % 24; + const wall = { + year: Number(get("year")), + month: Number(get("month")), + day: Number(get("day")), + hour, + minute: Number(get("minute")), + second: Number(get("second")), + weekday: WEEKDAYS.indexOf(get("weekday")), + }; + const asIfUtc = Date.UTC( + wall.year, + wall.month - 1, + wall.day, + wall.hour, + wall.minute, + wall.second, + ); + const pad = (n: number) => String(n).padStart(2, "0"); + return { + wall, + offsetMinutes: Math.round((asIfUtc - date.getTime()) / 60000), + dayKey: `${wall.year}-${pad(wall.month)}-${pad(wall.day)}`, + datetime: `${wall.year}-${pad(wall.month)}-${pad(wall.day)} ${pad(wall.hour)}:${pad(wall.minute)}`, + }; +} + +/** Instants every 7 min 13.5 s across two days around each 2026 transition. */ +function instantsAroundTransitions(): Date[] { + const anchors = [ + Date.UTC(2026, 2, 28), // EU spring forward (29 Mar) + Date.UTC(2026, 9, 24), // EU fall back (25 Oct) + Date.UTC(2026, 2, 7), // US spring forward (8 Mar) + Date.UTC(2026, 9, 31), // US fall back (1 Nov) + Date.UTC(2026, 3, 4), // Lord Howe / Chatham autumn (4-5 Apr) + Date.UTC(2026, 8, 26), // Chatham spring (27 Sep) + Date.UTC(2026, 9, 3), // Lord Howe spring (3 Oct 15:30 UTC) / Chatham (27 Sep) + ]; + const out: Date[] = []; + for (const anchor of anchors) { + for (let t = anchor; t < anchor + 2 * 86_400_000; t += 433_500) { + out.push(new Date(t)); + } + } + return out; +} + +beforeEach(() => { + __resetZoneOffsetCacheForTests(); +}); + +describe("zone offset memo", () => { + it("matches Intl for every instant around the transitions", () => { + const instants = instantsAroundTransitions(); + for (const tz of ZONES) { + for (const date of instants) { + const ref = reference(date, tz); + expect(wallClockInTz(date, tz)).toEqual(ref.wall); + expect(tzOffsetMinutes(date, tz)).toBe(ref.offsetMinutes); + expect(userDayKey(date, tz)).toBe(ref.dayKey); + expect(formatInUserTz(date, tz, "datetime")).toBe(ref.datetime); + } + } + }); + + it("matches Intl on a warm memo for scattered instants across decades", () => { + let seed = 7; + const next = () => { + seed = (seed * 1_103_515_245 + 12_345) % 2_147_483_648; + return seed / 2_147_483_648; + }; + const from = Date.UTC(1972, 0, 1); + const span = Date.UTC(2060, 0, 1) - from; + for (let i = 0; i < 4000; i++) { + const date = new Date(from + Math.floor(next() * span)); + const tz = ZONES[i % ZONES.length]; + // Twice: the first read fills the hour, the second reads the memo. + for (let pass = 0; pass < 2; pass++) { + const ref = reference(date, tz); + expect(wallClockInTz(date, tz)).toEqual(ref.wall); + expect(tzOffsetMinutes(date, tz)).toBe(ref.offsetMinutes); + expect(userDayKey(date, tz)).toBe(ref.dayKey); + } + } + }); + + it("refuses an hour that contains a transition", () => { + // Europe/Berlin springs forward at 01:00 UTC on 29 March 2026. + expect( + zoneOffsetMs(new Date(Date.UTC(2026, 2, 29, 1, 30)), "Europe/Berlin"), + ).toBe(2 * 3_600_000); + // Lord Howe springs forward half an hour at 15:30 UTC on 3 October: + // the 15:00 UTC hour straddles it and stays on the formatter path, + // on both sides of the change. + expect( + zoneOffsetMs( + new Date(Date.UTC(2026, 9, 3, 15, 10)), + "Australia/Lord_Howe", + ), + ).toBeNull(); + for (const minute of [10, 29, 30, 45]) { + const at = new Date(Date.UTC(2026, 9, 3, 15, minute)); + expect(wallClockInTz(at, "Australia/Lord_Howe")).toEqual( + reference(at, "Australia/Lord_Howe").wall, + ); + } + }); + + it("leaves out-of-range and invalid inputs to the formatter path", () => { + expect( + zoneOffsetMs(new Date(Date.UTC(1950, 0, 1)), "Europe/Berlin"), + ).toBeNull(); + expect(zoneOffsetMs(new Date(Number.NaN), "Europe/Berlin")).toBeNull(); + expect(() => wallClockInTz(new Date(), "Not/A_Zone")).toThrow(RangeError); + // The formatter reads a missing date as "now"; the memo must not turn + // that into a TypeError. + const now = reference(new Date(), "Europe/Berlin").wall; + const loose = wallClockInTz(undefined as unknown as Date, "Europe/Berlin"); + expect(loose.year).toBe(now.year); + expect(loose.day).toBe(now.day); + }); +}); diff --git a/src/lib/tz/format.ts b/src/lib/tz/format.ts index 5e6bde119..01809a9c2 100644 --- a/src/lib/tz/format.ts +++ b/src/lib/tz/format.ts @@ -13,6 +13,7 @@ */ import { getDateTimeFormat } from "./intl-cache"; +import { wallClockFromOffset, zoneOffsetMs } from "./zone-offset"; export const DEFAULT_TIMEZONE = "Europe/Berlin"; @@ -52,6 +53,9 @@ const WALL_CLOCK_PARTS_OPTIONS: Omit = */ const OFFSET_ZONE = /^[+-]\d/; +const validityByZone = new Map(); +const MAX_VALIDITY_ENTRIES = 256; + /** * Validate a timezone string: a named IANA zone (or `UTC`) the runtime * accepts. Returns `false` for anything else. Cheap (microseconds) — call @@ -67,6 +71,12 @@ export function isValidTimezone(tz: string): boolean { if (!tz || typeof tz !== "string" || tz.length === 0 || tz.length > 64) { return false; } + // The answer for a given name never changes within a process, and the + // day-key helpers below ask it once per row; `resolvedOptions()` allocates + // a fresh options object on every call. + const memo = validityByZone.get(tz); + if (memo !== undefined) return memo; + let valid: boolean; try { // The memo only caches successful constructions, so probing an // invalid zone through it cannot poison the formatter map — and a @@ -76,10 +86,13 @@ export function isValidTimezone(tz: string): boolean { tz, VALIDATION_OPTIONS, ).resolvedOptions().timeZone; - return !OFFSET_ZONE.test(resolved); + valid = !OFFSET_ZONE.test(resolved); } catch { - return false; + valid = false; } + if (validityByZone.size >= MAX_VALIDITY_ENTRIES) validityByZone.clear(); + validityByZone.set(tz, valid); + return valid; } /** @@ -275,7 +288,23 @@ interface WallClockParts { second: string; } +function pad2(value: number): string { + return value < 10 ? `0${value}` : String(value); +} + function wallClockParts(date: Date, tz: string): WallClockParts { + const offsetMs = zoneOffsetMs(date, tz); + if (offsetMs !== null) { + const at = wallClockFromOffset(date, offsetMs); + return { + year: String(at.year), + month: pad2(at.month), + day: pad2(at.day), + hour: pad2(at.hour), + minute: pad2(at.minute), + second: pad2(at.second), + }; + } const fmt = getDateTimeFormat("en-CA", tz, WALL_CLOCK_PARTS_OPTIONS); const parts = fmt.formatToParts(date); const get = (type: Intl.DateTimeFormatPartTypes): string => @@ -319,6 +348,12 @@ function formatIsoWithOffset(date: Date, tz: string): string { * difference between the wall-clock parts and the UTC parts. */ export function tzOffsetMinutes(date: Date, tz: string): number { + const offsetMs = zoneOffsetMs(date, tz); + if (offsetMs !== null) { + // Same arithmetic as below: the wall clock is read to the whole second, + // so the sub-second part of the instant shifts the rounded result. + return Math.round((offsetMs - (date.getTime() % 1000)) / 60000); + } const fmt = getDateTimeFormat("en-CA", tz, WALL_CLOCK_PARTS_OPTIONS); const parts = fmt.formatToParts(date); const get = (type: Intl.DateTimeFormatPartTypes): number => diff --git a/src/lib/tz/wall-clock.ts b/src/lib/tz/wall-clock.ts index 917e8f595..7ec7f1617 100644 --- a/src/lib/tz/wall-clock.ts +++ b/src/lib/tz/wall-clock.ts @@ -19,6 +19,7 @@ * silent drift a real failure mode. */ import { getDateTimeFormat } from "./intl-cache"; +import { wallClockFromOffset, zoneOffsetMs } from "./zone-offset"; export interface WallClockParts { year: number; @@ -190,6 +191,8 @@ export function wallClockInTz( weekday: date.getDay(), }; } + const offsetMs = zoneOffsetMs(date, tz); + if (offsetMs !== null) return wallClockFromOffset(date, offsetMs); const parts = getDateTimeFormat( "en-US", tz, diff --git a/src/lib/tz/zone-offset.ts b/src/lib/tz/zone-offset.ts new file mode 100644 index 000000000..12db19e02 --- /dev/null +++ b/src/lib/tz/zone-offset.ts @@ -0,0 +1,147 @@ +/** + * Per-hour memo of a zone's UTC offset, so wall-clock conversions in hot loops + * stop paying for `Intl.DateTimeFormat#formatToParts` on every row. + * + * A day-key or wall-clock read is pure arithmetic once the offset in force at + * the instant is known. The offset is probed through the same cached + * formatter the slow path uses, at both ends of the UTC hour that contains the + * instant; only when both ends agree (and the offset is a whole number of + * seconds) is it memoised for that hour. An hour that straddles a transition + * is stored as "not constant" and every instant inside it keeps going through + * the formatter, so a transition at any minute — Lord Howe's half-hour shift + * at 15:30 UTC, a historical LMT offset with seconds — still resolves exactly. + * + * The fast path is limited to 1971..2199 so the arithmetic never meets the + * Julian / proleptic-Gregorian split or a non-four-digit year, and so `t % 1000` + * is never negative. Outside that range, for an invalid date, or for a zone + * the formatter rejects, callers fall through to the formatter unchanged + * (an invalid zone throws from the probe exactly as it throws from the + * formatter path, and a value that is not a Date is left to the formatter). + */ +import { getDateTimeFormat } from "./intl-cache"; + +const PROBE_OPTIONS: Omit = + Object.freeze({ + year: "numeric", + month: "2-digit", + day: "2-digit", + hour: "2-digit", + minute: "2-digit", + second: "2-digit", + hour12: false, + }); + +const BUCKET_MS = 3_600_000; +const FAST_FROM_MS = Date.UTC(1971, 0, 1); +const FAST_UNTIL_MS = Date.UTC(2200, 0, 1); +/** Hours memoised across all zones before the memo starts over (~2 MB). */ +const MAX_BUCKETS = 20_000; +/** Stored for an hour whose offset changes inside it. */ +const NOT_CONSTANT = Number.NaN; + +const offsetsByZone = new Map>(); +let bucketCount = 0; + +/** Exact offset (ms) at an instant that falls on a whole second. */ +function probeOffsetMs(instantMs: number, tz: string): number { + const parts = getDateTimeFormat("en-CA", tz, PROBE_OPTIONS).formatToParts( + new Date(instantMs), + ); + let year = 0; + let month = 0; + let day = 0; + let hour = 0; + let minute = 0; + let second = 0; + for (const part of parts) { + switch (part.type) { + case "year": + year = Number(part.value); + break; + case "month": + month = Number(part.value); + break; + case "day": + day = Number(part.value); + break; + case "hour": + hour = Number(part.value); + break; + case "minute": + minute = Number(part.value); + break; + case "second": + second = Number(part.value); + break; + } + } + if (hour === 24) hour = 0; + return Date.UTC(year, month - 1, day, hour, minute, second) - instantMs; +} + +/** + * The zone's offset from UTC in milliseconds at `date`, or `null` when the + * caller must use the formatter (out of range, invalid date, or an hour that + * contains a transition). Throws for a zone `Intl` rejects. + */ +export function zoneOffsetMs(date: Date, tz: string): number | null { + // A caller that slips a non-Date through (an absent field read off a loose + // row) keeps the formatter's own handling: `formatToParts(undefined)` + // formats the current instant rather than throwing. + if (!(date instanceof Date)) return null; + const t = date.getTime(); + if (!(t >= FAST_FROM_MS && t < FAST_UNTIL_MS)) return null; + const bucket = Math.floor(t / BUCKET_MS); + let zone = offsetsByZone.get(tz); + const memo = zone?.get(bucket); + if (memo !== undefined) return Number.isNaN(memo) ? null : memo; + + const start = bucket * BUCKET_MS; + const atStart = probeOffsetMs(start, tz); + const atEnd = probeOffsetMs(start + BUCKET_MS - 1000, tz); + const offset = + atStart === atEnd && atStart % 1000 === 0 ? atStart : NOT_CONSTANT; + + if (bucketCount >= MAX_BUCKETS) { + offsetsByZone.clear(); + bucketCount = 0; + zone = undefined; + } + if (!zone) { + zone = new Map(); + offsetsByZone.set(tz, zone); + } + zone.set(bucket, offset); + bucketCount += 1; + return Number.isNaN(offset) ? null : offset; +} + +/** Wall-clock fields at `date` in the zone whose offset is `offsetMs`. */ +export function wallClockFromOffset( + date: Date, + offsetMs: number, +): { + year: number; + month: number; + day: number; + hour: number; + minute: number; + second: number; + weekday: number; +} { + const local = new Date(date.getTime() + offsetMs); + return { + year: local.getUTCFullYear(), + month: local.getUTCMonth() + 1, + day: local.getUTCDate(), + hour: local.getUTCHours(), + minute: local.getUTCMinutes(), + second: local.getUTCSeconds(), + weekday: local.getUTCDay(), + }; +} + +export function __resetZoneOffsetCacheForTests(): void { + offsetsByZone.clear(); + bucketCount = 0; +} 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)); + }); +});