mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-08 20:42:43 +00:00
Merge pull request #1147 from MBombeck/perf/dashboard-server
perf(server): one Prisma client, indexed coverage probe, memoised timezone offsets
This commit is contained in:
11 files changed
+709
-27
No files matched your search
@@ -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);
|
||||
});
|
||||
});
|
||||
Vendored
+5
@@ -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.
|
||||
|
||||
+9
-4
@@ -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)
|
||||
|
||||
@@ -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<T>() {
|
||||
let resolve!: (value: T) => void;
|
||||
let reject!: (err: unknown) => void;
|
||||
const promise = new Promise<T>((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<Array<{ type: string; has_buckets: boolean }>>();
|
||||
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<Array<{ type: string; has_buckets: boolean }>>();
|
||||
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<Array<{ type: string; has_buckets: boolean }>>();
|
||||
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);
|
||||
});
|
||||
});
|
||||
@@ -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<string, Promise<Map<string, boolean>>>;
|
||||
};
|
||||
const inFlight: Map<
|
||||
string,
|
||||
Promise<Map<string, boolean>>
|
||||
> = (registry.__healthlogCoverageInFlight ??= new Map());
|
||||
|
||||
/** Join the probe already running for `userId`, or start `run` as the one. */
|
||||
export function joinCoverageProbe(
|
||||
userId: string,
|
||||
run: () => Promise<Map<string, boolean>>,
|
||||
): Promise<Map<string, boolean>> {
|
||||
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);
|
||||
}
|
||||
@@ -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<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.
|
||||
*
|
||||
* 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<RollupCoverageMap> {
|
||||
return joinCoverageProbe(userId, () => readRollupCoverage(userId));
|
||||
}
|
||||
|
||||
async function readRollupCoverage(userId: string): Promise<RollupCoverageMap> {
|
||||
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) {
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
+37
-2
@@ -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<Intl.DateTimeFormatOptions, "timeZone"> =
|
||||
*/
|
||||
const OFFSET_ZONE = /^[+-]\d/;
|
||||
|
||||
const validityByZone = new Map<string, boolean>();
|
||||
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 =>
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<Intl.DateTimeFormatOptions, "timeZone"> =
|
||||
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<string, Map<number, number>>();
|
||||
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;
|
||||
}
|
||||
@@ -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));
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user