diff --git a/src/__tests__/baselines/source-file-lines.json b/src/__tests__/baselines/source-file-lines.json index 5238dd4b8..09fafac62 100644 --- a/src/__tests__/baselines/source-file-lines.json +++ b/src/__tests__/baselines/source-file-lines.json @@ -1,7 +1,7 @@ { "$comment": "Production files over 800 lines and their length. Maintained by src/__tests__/source-size-ratchet.test.ts.", "count": 89, - "total": 99364, + "total": 99252, "entries": { "app/medications/page-client.tsx": 839, "components/charts/mood-chart.tsx": 1166, @@ -35,7 +35,7 @@ "lib/ai/provider.ts": 1632, "lib/analytics/score/reader.ts": 998, "lib/analytics/sleep-night.ts": 1345, - "lib/analytics/summaries-slice.ts": 1231, + "lib/analytics/summaries-slice.ts": 1178, "lib/auth/session.ts": 811, "lib/cache/server-cache.ts": 848, "lib/cycle/prediction.ts": 883, @@ -58,7 +58,7 @@ "lib/illness/correlation.ts": 1059, "lib/import/health-connect/import.ts": 1689, "lib/import/unzip-export-xml.ts": 825, - "lib/insights/comprehensive-aggregator.ts": 927, + "lib/insights/comprehensive-aggregator.ts": 868, "lib/insights/comprehensive-generate.ts": 1674, "lib/insights/correlation-channel-series.ts": 1339, "lib/insights/correlation-discovery.ts": 1240, diff --git a/src/__tests__/baselines/source-function-lines.json b/src/__tests__/baselines/source-function-lines.json index 067438902..38bb22fd9 100644 --- a/src/__tests__/baselines/source-function-lines.json +++ b/src/__tests__/baselines/source-function-lines.json @@ -1,7 +1,7 @@ { "$comment": "Production functions over 200 lines, keyed \"file :: name\", and their length. Maintained by src/__tests__/source-size-ratchet.test.ts.", "count": 352, - "total": 113987, + "total": 113984, "entries": { "app/achievements/page.tsx :: AchievementsPage": 206, "app/api/admin/backups/upload/route.ts :: apiHandler(arg)": 314, @@ -262,7 +262,7 @@ "lib/analytics/correlations-fast-path.ts :: computeCorrelationHypothesesFastPath": 256, "lib/analytics/intraday-pulse-io.ts :: loadIntradayPulse": 206, "lib/analytics/score/reader.ts :: computeUserHealthScore": 324, - "lib/analytics/summaries-slice.ts :: computeFromRollups": 375, + "lib/analytics/summaries-slice.ts :: computeFromRollups": 373, "lib/api-handler.ts :: apiHandler": 207, "lib/api-handler.ts :: apiHandler > wrapped": 201, "lib/auth/profile-update.ts :: applyProfileUpdate": 353, @@ -295,7 +295,7 @@ "lib/illness/correlation.ts :: computeIllnessCorrelation": 233, "lib/import/labs-csv.ts :: importLabsCsv": 309, "lib/import/labs-csv.ts :: importLabsCsv > flush": 231, - "lib/insights/comprehensive-aggregator.ts :: buildFromRollups": 218, + "lib/insights/comprehensive-aggregator.ts :: buildFromRollups": 217, "lib/insights/comprehensive-generate.ts :: generateComprehensiveInsight": 876, "lib/insights/features.ts :: extractFeatures": 796, "lib/insights/mood-aggregates.ts :: fetchMoodAggregates": 246, diff --git a/src/app/api/activity/calendar/route.ts b/src/app/api/activity/calendar/route.ts index 9ee482f07..b6ff7a502 100644 --- a/src/app/api/activity/calendar/route.ts +++ b/src/app/api/activity/calendar/route.ts @@ -71,7 +71,6 @@ export const GET = apiHandler(async (request: NextRequest) => { tz, fromKey: bounds.readStart, toKey: lastRead, - now, includeWorkouts: true, }), prisma.user.findUnique({ diff --git a/src/lib/activity/calendar-read.ts b/src/lib/activity/calendar-read.ts index 8fc919d0d..2da42e8db 100644 --- a/src/lib/activity/calendar-read.ts +++ b/src/lib/activity/calendar-read.ts @@ -38,8 +38,6 @@ export const LOAD_TYPES: readonly MeasurementType[] = [ "STRAIN_SCORE", ]; -const DAY_MS = 86_400_000; - interface DayAggregateRow { day: string; total: number; @@ -54,15 +52,10 @@ async function dailyAggregate( type: MeasurementType, from: Date, to: Date, - now: Date, ): Promise> { - // `seriesRowsFrom` scopes its canonical pick to a trailing window counted - // from now; it is widened by a day so the edge day's source is picked from - // all of its readings, and the outer filter trims to the exact window. - const days = Math.min( - 3649, - Math.max(1, Math.ceil((now.getTime() - from.getTime()) / DAY_MS) + 1), - ); + // `seriesRowsFrom` scopes its canonical pick to the window from a day + // before `from`, so the edge day's source is picked from all of its + // readings; the outer filter trims to the exact window. const rows = await prisma.$queryRawUnsafe( ` SELECT @@ -72,7 +65,7 @@ async function dailyAggregate( ) AS day, SUM(m."value")::double precision AS total, COUNT(*)::int AS n - FROM ${seriesRowsFrom(priorityJson, type, days)} + FROM ${seriesRowsFrom(priorityJson, type, from)} WHERE m."measured_at" >= $3 AND m."measured_at" < $4 GROUP BY 1 @@ -103,21 +96,19 @@ export async function readActivityDays(args: { tz: string; fromKey: string; toKey: string; - now?: Date; includeWorkouts: boolean; }): Promise { const { userId, tz, fromKey, toKey, includeWorkouts } = args; - const now = args.now ?? new Date(); const from = startOfLocalDayKey(fromKey, tz); const to = startOfLocalDayKey(shiftDateKey(toKey, 1), tz); const priorityJson = await loadUserSourcePriority(userId); const [steps, exercise, loads, workoutRows] = await Promise.all([ - dailyAggregate(userId, tz, priorityJson, "ACTIVITY_STEPS", from, to, now), - dailyAggregate(userId, tz, priorityJson, "EXERCISE_MINUTES", from, to, now), + dailyAggregate(userId, tz, priorityJson, "ACTIVITY_STEPS", from, to), + dailyAggregate(userId, tz, priorityJson, "EXERCISE_MINUTES", from, to), Promise.all( LOAD_TYPES.map((type) => - dailyAggregate(userId, tz, priorityJson, type, from, to, now), + dailyAggregate(userId, tz, priorityJson, type, from, to), ), ), includeWorkouts diff --git a/src/lib/analytics/__tests__/source-rank-sql.test.ts b/src/lib/analytics/__tests__/source-rank-sql.test.ts index e3b2ff4f0..5a5f0cebc 100644 --- a/src/lib/analytics/__tests__/source-rank-sql.test.ts +++ b/src/lib/analytics/__tests__/source-rank-sql.test.ts @@ -9,21 +9,25 @@ * wrapped in `( … ) m` — so the single-reference call site is unchanged; * 2. the canonical body carries exactly ONE DISTINCT-ON pick — so a fold * that references the CTE twice runs the expensive pick once; - * 3. the interval whitelist still rejects an unsafe splice. + * 3. the interval whitelist still rejects an unsafe splice; + * 4. the window counts from an instant the caller took, never from the + * database clock. */ import { describe, expect, it } from "vitest"; +import { boundInstantSql, instantLiteralSql } from "../request-clock-sql"; import { canonicalMeasurementsCte, canonicalMeasurementsFrom, } from "../source-rank-sql"; const RANK = "90"; +const NINETY_DAYS = { from: boundInstantSql(2), interval: "90 days" }; describe("canonicalMeasurementsCte / canonicalMeasurementsFrom", () => { it("wraps the CTE body verbatim in the ( … ) m alias", () => { - const cte = canonicalMeasurementsCte(RANK, "90 days"); - const from = canonicalMeasurementsFrom(RANK, "90 days"); + const cte = canonicalMeasurementsCte(RANK, NINETY_DAYS); + const from = canonicalMeasurementsFrom(RANK, NINETY_DAYS); // The alias form is the CTE body plus the closing `) m` wrapper. Pinning // equality means the single-reference call sites keep byte-stable SQL // while the CTE body can be reused for the folded double-reference case. @@ -31,29 +35,65 @@ describe("canonicalMeasurementsCte / canonicalMeasurementsFrom", () => { }); it("runs exactly one DISTINCT-ON canonical pick in the body", () => { - const cte = canonicalMeasurementsCte(RANK, "90 days"); + const cte = canonicalMeasurementsCte(RANK, NINETY_DAYS); const matches = cte.match(/DISTINCT ON/g) ?? []; expect(matches).toHaveLength(1); }); it("binds the user id as $1 and scopes both the inner and outer windows", () => { - const cte = canonicalMeasurementsCte(RANK, "90 days"); + const cte = canonicalMeasurementsCte(RANK, NINETY_DAYS); // user id is always a bound parameter, never spliced. expect(cte).toContain('"user_id" = $1'); // The trailing-window bound appears on both the inner pick and the outer // filter so the folded references stay scoped identically. - const intervalMatches = cte.match(/INTERVAL '90 days'/g) ?? []; + const intervalMatches = + cte.match(/>= \$2::timestamp - INTERVAL '90 days'/g) ?? []; expect(intervalMatches).toHaveLength(2); }); + it("never cuts its window on the database clock", () => { + for (const cte of [ + canonicalMeasurementsCte(RANK, NINETY_DAYS), + canonicalMeasurementsCte(RANK, { from: boundInstantSql(3) }), + canonicalMeasurementsCte(RANK, { + from: instantLiteralSql(new Date("2026-03-01T06:30:00.000Z")), + interval: "1 day", + }), + ]) { + expect(cte).not.toMatch(/NOW\(\)|CURRENT_|LOCALTIMESTAMP/i); + } + expect( + canonicalMeasurementsCte(RANK, { + from: instantLiteralSql(new Date("2026-03-01T06:30:00.000Z")), + interval: "1 day", + }), + ).toContain(`>= '2026-03-01 06:30:00.000'::timestamp - INTERVAL '1 day'`); + }); + it("omits the window clause entirely when no interval is given", () => { const cte = canonicalMeasurementsCte(RANK); expect(cte).not.toContain("INTERVAL"); }); it("rejects an unsafe interval splice", () => { - expect(() => canonicalMeasurementsCte(RANK, "90 days; DROP TABLE")).toThrow( - /unsafe interval/, - ); + expect(() => + canonicalMeasurementsCte(RANK, { + from: boundInstantSql(2), + interval: "90 days; DROP TABLE", + }), + ).toThrow(/unsafe interval/); + }); + + it("rejects an instant that is neither a bound parameter nor a literal it wrote", () => { + for (const from of [ + "NOW()", + "$2", + "'2026-03-01'::timestamp", + "$2::timestamp; DROP TABLE", + ]) { + expect(() => + canonicalMeasurementsCte(RANK, { from, interval: "90 days" }), + ).toThrow(/unsafe instant/); + } }); }); diff --git a/src/lib/analytics/__tests__/summaries-slice.test.ts b/src/lib/analytics/__tests__/summaries-slice.test.ts index 9d3f85e30..f0a22ec1c 100644 --- a/src/lib/analytics/__tests__/summaries-slice.test.ts +++ b/src/lib/analytics/__tests__/summaries-slice.test.ts @@ -46,15 +46,15 @@ vi.mock("@/lib/logging/context", () => ({ // coupling the slice unit test to the rank builder's enum-whitelist // internals. vi.mock("@/lib/analytics/source-rank-sql", () => { - const cte = (_rank: string, sinceInterval?: string) => + const cte = (_rank: string, window?: { from: string; interval?: string }) => ` SELECT mm.* FROM measurements mm WHERE mm."user_id" = $1 AND mm."deleted_at" IS NULL ${ - sinceInterval - ? `AND mm."measured_at" >= NOW() - INTERVAL '${sinceInterval}'` + window + ? `AND mm."measured_at" >= ${window.from} - INTERVAL '${window.interval}'` : "" }`; return { @@ -605,20 +605,20 @@ describe("computeSummariesSlice", () => { // raw alias (`mm.`) rather than the outer `m.`. Pin that the cap // survives the source-rank refactor. const narrowsCap = joined.match( - /AND mm\."measured_at" >= NOW\(\) - INTERVAL '90 days'/g, + /AND mm\."measured_at" >= \$2::timestamp - INTERVAL '90 days'/g, ); expect(narrowsCap).not.toBeNull(); expect(narrowsCap?.length).toBeGreaterThanOrEqual(1); // v1.11.1 — the cold-fallback `windowed` scan delegates its outer - // 90-day cap to `canonicalMeasurementsCte(rank, "90 days")` (a CTE - // now, so the day weights can read it twice). The - // helper lives in `@/lib/analytics/source-rank-sql` (stubbed - // above), so pin the contract at the call boundary: the slice - // must ask for the 90-day window. + // 90-day cap to `canonicalMeasurementsCte` (a CTE now, so the day + // weights can read it twice). The helper lives in + // `@/lib/analytics/source-rank-sql` (stubbed above), so pin the + // contract at the call boundary: the slice must ask for the 90-day + // window, counted from the request's `now` bound at `$2`. expect(canonicalMeasurementsCte).toHaveBeenCalledWith( expect.any(String), - "90 days", + { from: "$2::timestamp", interval: "90 days" }, ); // v1.18.10 — the `allTime` aggregate is now capped at a generous // 15-year scan-DoS floor (far beyond any real history) so a @@ -626,8 +626,16 @@ describe("computeSummariesSlice", () => { // scan. It no longer calls the helper without an interval. expect(canonicalMeasurementsCte).toHaveBeenCalledWith( expect.any(String), - "15 years", + { from: "$2::timestamp", interval: "15 years" }, ); + // Every window is cut against the request's clock, bound with each + // statement; none reads the database clock. + expect(joined).not.toMatch(/NOW\(\)|CURRENT_TIMESTAMP|CURRENT_DATE/i); + for (const call of UNSAFE.mock.calls) { + if (String(call[0]).includes("$2::timestamp")) { + expect(call[2]).toBeInstanceOf(Date); + } + } }); }); @@ -736,8 +744,8 @@ describe("computeSummariesSlice", () => { * `computeFromRollups` path windows the regression on UTC-midnight * (`startOfUtcDay(now − N days)` + `composeWindowedRegression`). The * cold-fallback `windowed` query must anchor its REGR_* FILTERs on the - * SAME UTC-midnight boundary — `date_trunc('day', NOW() AT TIME ZONE - * 'UTC') − INTERVAL 'N days'` — so a warm→cold coverage transition + * SAME UTC-midnight boundary of the same `now` — `date_trunc('day', + * $2::timestamp) − INTERVAL 'N days'` — so a warm→cold coverage transition * returns identical boundary-day membership for slope7/30/90 instead of * a cache-dependent answer to the same request. */ @@ -766,11 +774,11 @@ describe("computeSummariesSlice", () => { // read the same rows the regression fits. for (const days of ["7 days", "30 days"]) { expect(sql).toContain( - `m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '${days}')`, + `m."measured_at" >= (date_trunc('day', $2::timestamp) - INTERVAL '${days}')`, ); } const anchored = sql.match( - /date_trunc\('day', NOW\(\) AT TIME ZONE 'UTC'\) - INTERVAL '(?:7|30) days'\)/g, + /date_trunc\('day', \$2::timestamp\) - INTERVAL '(?:7|30) days'\)/g, ); expect(anchored?.length).toBe(10); // The bound stays a naive UTC timestamp like measured_at: converting @@ -778,14 +786,14 @@ describe("computeSummariesSlice", () => { // time zone. expect(sql).not.toMatch(/INTERVAL '(?:7|30) days'\) AT TIME ZONE/); - // The regression windows must NOT fall back to the bare wall-clock - // `NOW() - INTERVAL 'N days'` bound the warm path never uses for the - // slope. Isolate the slope columns (everything from the first - // REGR_SLOPE onward) so the avg7/avg30 windows above — which DO stay - // wall-clock by design — don't trip the assertion. + // The regression windows must NOT fall back to the bare instant + // bound the warm path never uses for the slope. Isolate the slope + // columns (everything from the first REGR_SLOPE onward) so the + // avg7/avg30 windows above, which DO count from the instant by + // design, don't trip the assertion. const slopeBlock = sql.slice(sql.indexOf("REGR_SLOPE")); expect(slopeBlock).not.toMatch( - /WHERE m\."measured_at" >= \(?NOW\(\)( AT TIME ZONE 'UTC'\))? -/, + /WHERE m\."measured_at" >= \$2::timestamp -/, ); }); }); diff --git a/src/lib/analytics/__tests__/summaries-slice.wmy-cap.test.ts b/src/lib/analytics/__tests__/summaries-slice.wmy-cap.test.ts index c08caa5d3..b46bdbcb6 100644 --- a/src/lib/analytics/__tests__/summaries-slice.wmy-cap.test.ts +++ b/src/lib/analytics/__tests__/summaries-slice.wmy-cap.test.ts @@ -71,7 +71,7 @@ describe("computeAvg30LastYearMap — WMY fan-out cap (v1.4.43)", () => { }); const types = Array.from({ length: 15 }, (_, i) => `TYPE_${i}`); - const promise = computeAvg30LastYearMap("user-1", types); + const promise = computeAvg30LastYearMap("user-1", types, Date.now()); // Let the scheduler hand out the initial batch before we start // draining. Two macrotask flushes is enough for `p-limit` to fill @@ -103,7 +103,7 @@ describe("computeAvg30LastYearMap — WMY fan-out cap (v1.4.43)", () => { }); it("returns an empty map without invoking the reader when given no types", async () => { - const result = await computeAvg30LastYearMap("user-1", []); + const result = await computeAvg30LastYearMap("user-1", [], Date.now()); expect(result.size).toBe(0); expect(READ).not.toHaveBeenCalled(); }); diff --git a/src/lib/analytics/bp-in-target-fast-path.ts b/src/lib/analytics/bp-in-target-fast-path.ts index 58de8a1bd..0d25bcad6 100644 --- a/src/lib/analytics/bp-in-target-fast-path.ts +++ b/src/lib/analytics/bp-in-target-fast-path.ts @@ -58,6 +58,7 @@ import { buildSourceRankCase, canonicalMeasurementsFrom, } from "@/lib/analytics/source-rank-sql"; +import { boundInstantSql } from "@/lib/analytics/request-clock-sql"; import { probeRollupCoverage, type RollupCoverageMap, @@ -599,12 +600,11 @@ async function computeFromLive( async function fetchSeriesChunked( userId: string, type: "BLOOD_PRESSURE_SYS" | "BLOOD_PRESSURE_DIA", - // The 365-day window is expressed as a SQL interval inside the canonical - // subquery; the bound `Date` arg is retained for call-site symmetry. + // The window's start (365 days before the request's `now`), bound as the + // canonical pick's lower edge so the cut runs on the caller's clock. since: Date, rankUnqualified: string, ): Promise { - void since; const rows = await prisma.$queryRawUnsafe< Array<{ measured_at: Date; value: number }> >( @@ -612,11 +612,12 @@ async function fetchSeriesChunked( SELECT m."measured_at" AS measured_at, m."value"::double precision AS value - FROM ${canonicalMeasurementsFrom(rankUnqualified, "365 days")} + FROM ${canonicalMeasurementsFrom(rankUnqualified, { from: boundInstantSql(2) })} WHERE m."type" = '${type}' ORDER BY m."measured_at" ASC `, userId, + since, ); return rows.map((r) => ({ measuredAt: new Date(r.measured_at), diff --git a/src/lib/analytics/request-clock-sql.ts b/src/lib/analytics/request-clock-sql.ts new file mode 100644 index 000000000..c8d3e82e5 --- /dev/null +++ b/src/lib/analytics/request-clock-sql.ts @@ -0,0 +1,88 @@ +/** + * The request's clock inside a live aggregate. + * + * A summary read cuts some of its windows in JS (`startOfUtcDay(now - 7 d)` + * for the rollup slope, the `bucketStart` bound of the DAY buckets, the + * staleness of `lastSeenAt`) and others in SQL (the 7- and 30-day means, the + * 90-day canonical pick, the median). Those used to read `NOW()`, the + * database's clock, so one response mixed two clocks: a reading on a window's + * edge fell in or out with whichever host answered, and a process whose clock + * is pinned (a test, a replayed request) saw its SQL windows cut hours or + * days away from its JS ones. Every window is now cut against the one `now` + * the request took, bound as a parameter. + * + * `measured_at` is a naive UTC timestamp. The driver sends a `Date` parameter + * as an offset-free UTC string, so `$N::timestamp` keeps it as written, + * whatever the session time zone; `::timestamptz` would read it in the + * session zone and shift every bound outside UTC. + */ +import { windowMeanSql } from "@/lib/measurements/day-mean"; + +const BOUND_INSTANT_RE = /^\$[1-9][0-9]?::timestamp$/; +const INSTANT_LITERAL_RE = + /^'\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d{3}'::timestamp$/; + +/** The `Date` bound at `$param`, as the naive UTC timestamp `measured_at` holds. */ +export function boundInstantSql(param: number): string { + if (!Number.isInteger(param) || param < 1 || param > 99) { + throw new Error(`parameter index out of range for SQL splice: ${param}`); + } + return `$${param}::timestamp`; +} + +/** + * `at` as a naive UTC timestamp literal, for a fragment that several + * statements splice at different parameter positions (`seriesRowsFrom`). It + * is written from the `Date` itself, never from input, and checked against a + * closed format at every splice point. + */ +export function instantLiteralSql(at: Date): string { + if (Number.isNaN(at.getTime())) { + throw new Error("invalid instant for SQL splice"); + } + return `'${at.toISOString().slice(0, 23).replace("T", " ")}'::timestamp`; +} + +/** Refuse anything but a `boundInstantSql` / `instantLiteralSql` expression. */ +export function assertInstantSql(sql: string): string { + if (!BOUND_INSTANT_RE.test(sql) && !INSTANT_LITERAL_RE.test(sql)) { + throw new Error(`unsafe instant expression for SQL splice: ${sql}`); + } + return sql; +} + +function windowedMean(filter: string): string { + return windowMeanSql({ + typeColumn: 'm."type"', + value: 'm."value"', + weight: "m.day_weight", + filter, + }); +} + +/** + * The 7-day window mean over canonical readings carrying a `day_weight` (see + * `day-mean.ts`): the mean of the day values for an hourly-mean type (pulse), + * the unchanged `AVG(value)` otherwise. Days are UTC days, the frame the + * canonical-source pick and the rollup tier bucket by. + */ +export function avg7Sql(now: string): string { + return windowedMean( + `m."measured_at" >= ${assertInstantSql(now)} - INTERVAL '7 days'`, + ); +} + +/** + * `avg7`, `avg30` and `avg30_last_month` (days [30, 60) back, the tile's + * delta callout) as SELECT columns over rows aliased `m`, grouped per type. + */ +export function windowMeanColumnsSql(now: string): string { + const at = assertInstantSql(now); + const avg30 = windowedMean(`m."measured_at" >= ${at} - INTERVAL '30 days'`); + const lastMonth = windowedMean( + `m."measured_at" >= ${at} - INTERVAL '60 days' AND m."measured_at" < ${at} - INTERVAL '30 days'`, + ); + return `${avg7Sql(at)}::double precision AS avg7, + ${avg30}::double precision AS avg30, + ${lastMonth}::double precision AS avg30_last_month`; +} diff --git a/src/lib/analytics/source-rank-sql.ts b/src/lib/analytics/source-rank-sql.ts index 279413a99..58b015bf5 100644 --- a/src/lib/analytics/source-rank-sql.ts +++ b/src/lib/analytics/source-rank-sql.ts @@ -19,6 +19,7 @@ * The user id is always passed as a bound parameter, never spliced. */ import type { MeasurementType } from "@/generated/prisma/client"; +import { assertInstantSql } from "@/lib/analytics/request-clock-sql"; import { metricKeyForType } from "@/lib/measurements/cumulative-day-sum"; import { getSourceLadder, @@ -108,6 +109,15 @@ function assertInterval(value: string): string { return value; } +/** + * A trailing window: the readings from the instant `from` on, or from + * `interval` before it when one is given. + */ +export interface CanonicalWindow { + from: string; + interval?: string; +} + /** * Build the canonical-source SELECT body — the raw `measurements` rows * restricted to the ladder-winning source per (type, day) — WITHOUT the @@ -123,10 +133,13 @@ function assertInterval(value: string): string { * `canonicalMeasurementsFrom`, which wraps this body in the alias. * * The user id is bound as `$1`. `rankUnqualified` must be a CASE built with - * unqualified `"type"`/`"source"` columns. `sinceInterval` (e.g. `"90 days"`) - * is whitelisted and, when given, scopes both the inner pick and the outer - * filter to a trailing window. `onlyType`, when given, scopes both halves to - * that one metric. + * unqualified `"type"`/`"source"` columns. `window`, when given, scopes both + * the inner pick and the outer filter to the readings from `from` on, or from + * `interval` (e.g. `"90 days"`, whitelisted) before it. `from` is an instant + * the caller took (`boundInstantSql` or `instantLiteralSql`, see + * `request-clock-sql.ts`), the request's `now` for a trailing window, so the + * cut runs on the caller's clock and never on the database's. `onlyType`, when given, scopes both + * halves to that one metric. * * The pick groups before it sorts. Its order (rank, then source name) depends * only on a row's `(type, source)`, so every reading of one source on one day @@ -140,15 +153,16 @@ function assertInterval(value: string): string { */ export function canonicalMeasurementsCte( rankUnqualified: string, - sinceInterval?: string, + window?: CanonicalWindow, onlyType?: MeasurementType, ): string { - const sinceInner = sinceInterval - ? `AND "measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '${assertInterval(sinceInterval)}'` - : ""; - const sinceOuter = sinceInterval - ? `AND mm."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '${assertInterval(sinceInterval)}'` - : ""; + const bound = !window + ? null + : window.interval === undefined + ? assertInstantSql(window.from) + : `${assertInstantSql(window.from)} - INTERVAL '${assertInterval(window.interval)}'`; + const sinceInner = bound ? `AND "measured_at" >= ${bound}` : ""; + const sinceOuter = bound ? `AND mm."measured_at" >= ${bound}` : ""; // A single-type reader (a series for one metric) scopes both halves to that // type, so the pick scans one metric's rows instead of every row the user // has in the window. @@ -194,16 +208,14 @@ export function canonicalMeasurementsCte( * an overlapping vital never blends two devices — the same collapse the rollup * readers apply, kept in lockstep for live/rollup parity. * - * The user id is bound as `$1`. `rankUnqualified` must be a CASE built with - * unqualified `"type"`/`"source"` columns. `sinceInterval` (e.g. `"90 days"`) - * is whitelisted and, when given, scopes both the inner pick and the outer - * filter to a trailing window. + * The user id is bound as `$1`; `rankUnqualified`, `window` and `onlyType` are + * those of `canonicalMeasurementsCte`. */ export function canonicalMeasurementsFrom( rankUnqualified: string, - sinceInterval?: string, + window?: CanonicalWindow, onlyType?: MeasurementType, ): string { - return `(${canonicalMeasurementsCte(rankUnqualified, sinceInterval, onlyType)} + return `(${canonicalMeasurementsCte(rankUnqualified, window, onlyType)} ) m`; } diff --git a/src/lib/analytics/summaries-slice.ts b/src/lib/analytics/summaries-slice.ts index ffa9a976b..031022a05 100644 --- a/src/lib/analytics/summaries-slice.ts +++ b/src/lib/analytics/summaries-slice.ts @@ -73,10 +73,15 @@ import { } from "@/lib/measurements/day-mean"; import { wireSummary, type DataSummary } from "@/lib/analytics/trends"; import { - TREND_EVIDENCE_SQL, liveTrend, + liveTrendSql, type LiveTrendColumns, } from "@/lib/analytics/trend-evidence-sql"; +import { + avg7Sql, + boundInstantSql, + windowMeanColumnsSql, +} from "@/lib/analytics/request-clock-sql"; import { measurementTypeEnum } from "@/lib/validations/measurement"; import { annotate } from "@/lib/logging/context"; import { caches, type ServerCache } from "@/lib/cache/server-cache"; @@ -168,30 +173,6 @@ interface RollupAggregateRow { mean: number; } -/** - * The window means over canonical readings that carry a `day_weight` (see - * `day-mean.ts`): the mean of the day values for an hourly-mean type (pulse), - * the unchanged `AVG(value)` for every other type. Days are UTC days, the - * frame the canonical-source pick and the rollup tier bucket by. - */ -function windowedMean(filter: string): string { - return windowMeanSql({ - typeColumn: 'm."type"', - value: 'm."value"', - weight: "m.day_weight", - filter, - }); -} -const AVG7_SQL = windowedMean( - `m."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '7 days'`, -); -const AVG30_SQL = windowedMean( - `m."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days'`, -); -const AVG30_LAST_MONTH_SQL = windowedMean( - `m."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '60 days' AND m."measured_at" < (NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days'`, -); - interface NarrowAggregateRow { type: string; avg7: number | null; @@ -293,6 +274,9 @@ export async function computeSummariesSlice( */ precomputedCoverage?: RollupCoverageMap, ): Promise { + // One clock for every window of the slice, JS and SQL alike (see + // `request-clock-sql.ts`). + const now = new Date(); // v1.4.37.1 — fire-and-forget. See `src/app/api/analytics/route.ts` // for the full rationale: awaiting this on the read path can stall // the event loop for tens of seconds on tenants with large iOS step @@ -318,7 +302,7 @@ export async function computeSummariesSlice( if (isFullyCovered(coverage)) { return withLatestDayValues( userId, - await withSleepNightTotals(userId, await computeFromRollups(userId)), + await withSleepNightTotals(userId, await computeFromRollups(userId, now)), ); } // v1.4.38.7 — annotate the live fallback with the per-type @@ -343,7 +327,10 @@ export async function computeSummariesSlice( }); return withLatestDayValues( userId, - await withSleepNightTotals(userId, await computeFromLiveAggregate(userId)), + await withSleepNightTotals( + userId, + await computeFromLiveAggregate(userId, now), + ), ); } @@ -510,15 +497,16 @@ export async function readSevenDayMeans( Array<{ type: string; avg7: number | null }> >( ` - WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, "90 days")} + WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, { from: boundInstantSql(2), interval: "90 days" })} ) SELECT m."type"::text AS type, - ${AVG7_SQL}::double precision AS avg7 + ${avg7Sql(boundInstantSql(2))}::double precision AS avg7 FROM ${dayWeightedRows("cm", SESSION_DAY_FRAME)} m GROUP BY m."type" `, userId, + new Date(), ); const out: Record = {}; for (const row of rows) { @@ -545,7 +533,10 @@ export async function readSevenDayMeans( * large measurement partitions this lifts the slim slice from * multi-second cold to sub-second; output is bit-identical. */ -async function computeFromRollups(userId: string): Promise { +async function computeFromRollups( + userId: string, + now: Date, +): Promise { // v1.11.1 — resolve the user's source-priority ladders once, then build the // rank CASE expressions used to collapse overlapping sources to the canonical // reading per (type, day) inside each query. `"type"`/`"source"` variants for @@ -577,7 +568,8 @@ async function computeFromRollups(userId: string): Promise { // passed — so bounding only the raw half counted each reading on such a // day twice. Split on a boundary, a reading is in exactly one half // whichever buckets exist. - const foldSplit = startOfUtcDay(new Date(Date.now() - ROLLUP_FOLD_WINDOW_MS)); + const nowMs = now.getTime(); + const foldSplit = startOfUtcDay(new Date(nowMs - ROLLUP_FOLD_WINDOW_MS)); const [narrows, latests, dayBuckets, accumulatorRows, preFoldRemainder] = await Promise.all([ @@ -589,22 +581,20 @@ async function computeFromRollups(userId: string): Promise { // (`windowedMean`); the median stays over every reading. prisma.$queryRawUnsafe( ` - WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, "90 days")} + WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, { from: boundInstantSql(2), interval: "90 days" })} ) SELECT m."type"::text AS type, - ${AVG7_SQL}::double precision AS avg7, - ${AVG30_SQL}::double precision AS avg30, - ${AVG30_LAST_MONTH_SQL}::double precision AS avg30_last_month, + ${windowMeanColumnsSql(boundInstantSql(2))}, + -- The canonical set above is already the 90-day window. PERCENTILE_CONT(0.5) WITHIN GROUP ( ORDER BY m."value" - ) FILTER ( - WHERE m."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '90 days' )::double precision AS median FROM ${dayWeightedRows("cm", SESSION_DAY_FRAME)} m GROUP BY m."type" `, userId, + now, ), readLatestCanonical(userId, priorityJson), // v1.4.37.2 — the v1.4.35 implementation read EVERY DAY @@ -670,7 +660,7 @@ async function computeFromRollups(userId: string): Promise { userId, granularity: "DAY", bucketStart: { - gte: startOfUtcDay(new Date(Date.now() - 90 * DAY_MS)), + gte: startOfUtcDay(new Date(nowMs - 90 * DAY_MS)), }, }, orderBy: [{ type: "asc" }, { bucketStart: "asc" }], @@ -738,8 +728,7 @@ async function computeFromRollups(userId: string): Promise { } // Day-aligned 7/30/90-day anchors — the rollup-grain equivalent of the - // live `measured_at >= NOW() - INTERVAL 'N days'` filter. - const nowMs = Date.now(); + // live `measured_at >= now - INTERVAL 'N days'` filter. const since7 = startOfUtcDay(new Date(nowMs - 7 * DAY_MS)); const since30 = startOfUtcDay(new Date(nowMs - 30 * DAY_MS)); const since90 = startOfUtcDay(new Date(nowMs - 90 * DAY_MS)); @@ -769,11 +758,8 @@ async function computeFromRollups(userId: string): Promise { lastSeenByType[type] = null; } - const nowForStaleness = Date.now(); for (const [type, measuredAt] of lastSeenAtByType.entries()) { - const daysAgo = Math.floor( - (nowForStaleness - measuredAt.getTime()) / (24 * 60 * 60 * 1000), - ); + const daysAgo = Math.floor((nowMs - measuredAt.getTime()) / DAY_MS); lastSeenByType[type] = { lastSeenAt: measuredAt.toISOString(), daysAgo, @@ -880,9 +866,9 @@ async function computeFromRollups(userId: string): Promise { // with data in the current window are probed so we don't pay the per-type // round-trip on the long tail of unlogged measurement enums. Types with no // reading in the year-ago window surface as `null`. - const avg30LastYearMap = await computeAvg30LastYearMap(userId, typesWithData); + const lastYear = await computeAvg30LastYearMap(userId, typesWithData, nowMs); let yearOverYearTypeCount = 0; - for (const [type, value] of avg30LastYearMap.entries()) { + for (const [type, value] of lastYear.entries()) { if (value === null) continue; const summary = summaries[type]; if (!summary) continue; @@ -939,7 +925,7 @@ async function computeFromRollups(userId: string): Promise { * * 2. `windowed` — `avg7 / avg30` and the slope/r² tuples are all * already filtered to a 7/30/90 day window inside the SELECT. - * Adding an outer `measured_at >= NOW() - INTERVAL '90 days'` + * Adding an outer `measured_at >= now - INTERVAL '90 days'` * cap lets the planner do an index range scan on * `(user_id, type, measured_at)` instead of a full-partition * sequential scan and discarding 95 % of rows inside FILTER @@ -950,7 +936,9 @@ async function computeFromRollups(userId: string): Promise { */ async function computeFromLiveAggregate( userId: string, + now: Date, ): Promise { + const nowMs = now.getTime(); // v1.11.1 — collapse overlapping sources to the ladder-canonical reading per // (type, day) BEFORE aggregating, exactly as the rollup path does, so a // coverage-miss tenant gets the same numbers as a warm one (live/rollup @@ -966,7 +954,7 @@ async function computeFromLiveAggregate( const [allTime, windowed, latests] = await Promise.all([ prisma.$queryRawUnsafe( ` - WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, ALL_TIME_LIVE_CAP_INTERVAL)} + WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, { from: boundInstantSql(2), interval: ALL_TIME_LIVE_CAP_INTERVAL })} ) SELECT m."type"::text AS type, @@ -979,62 +967,24 @@ async function computeFromLiveAggregate( GROUP BY m."type" `, userId, + now, ), prisma.$queryRawUnsafe( ` - WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, "90 days")} + WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, { from: boundInstantSql(2), interval: "90 days" })} ) SELECT m."type"::text AS type, - ${AVG7_SQL}::double precision AS avg7, - ${AVG30_SQL}::double precision AS avg30, - ${AVG30_LAST_MONTH_SQL}::double precision AS avg30_last_month, + ${windowMeanColumnsSql(boundInstantSql(2))}, PERCENTILE_CONT(0.5) WITHIN GROUP ( ORDER BY m."value" - ) FILTER ( - WHERE m."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '90 days' )::double precision AS median, - -- A2-M2 — anchor the regression windows on the UTC-midnight day - -- boundary so this cold-fallback path returns the SAME boundary-day - -- membership the warm computeFromRollups composes via - -- startOfUtcDay(now - N days) + composeWindowedRegression. The - -- expression date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL - -- 'N days' is the SQL twin of startOfUtcDay. It stays a naive UTC - -- timestamp, like measured_at itself, so the comparison never - -- passes through the session time zone. The wall-clock NOW() - INTERVAL - -- bound let the boundary day flip in/out of the slope window across - -- a warm-to-cold coverage transition (cache-dependent slope on the - -- same request); the avg7/30/median windows above stay wall-clock - -- because the warm narrow query computes THOSE the same way. - REGR_SLOPE( - m."value", - EXTRACT(EPOCH FROM m."measured_at") / 86400.0 - ) FILTER ( - WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '7 days') - )::double precision AS slope7, - REGR_R2( - m."value", - EXTRACT(EPOCH FROM m."measured_at") / 86400.0 - ) FILTER ( - WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '7 days') - )::double precision AS r2_7, - REGR_SLOPE( - m."value", - EXTRACT(EPOCH FROM m."measured_at") / 86400.0 - ) FILTER ( - WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days') - )::double precision AS slope30, - REGR_R2( - m."value", - EXTRACT(EPOCH FROM m."measured_at") / 86400.0 - ) FILTER ( - WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days') - )::double precision AS r2_30, - ${TREND_EVIDENCE_SQL} + ${liveTrendSql(boundInstantSql(2))} FROM ${dayWeightedRows("cm", SESSION_DAY_FRAME)} m GROUP BY m."type" `, userId, + now, ), readLatestCanonical(userId, priorityJson), ]); @@ -1058,11 +1008,8 @@ async function computeFromLiveAggregate( lastSeenByType[type] = null; } - const nowForStaleness = Date.now(); for (const [type, measuredAt] of lastSeenAtByType.entries()) { - const daysAgo = Math.floor( - (nowForStaleness - measuredAt.getTime()) / (24 * 60 * 60 * 1000), - ); + const daysAgo = Math.floor((nowMs - measuredAt.getTime()) / DAY_MS); lastSeenByType[type] = { lastSeenAt: measuredAt.toISOString(), daysAgo, @@ -1107,9 +1054,9 @@ async function computeFromLiveAggregate( // the year-ago DAY buckets may already exist because the boot-time // backfill folds the whole history once per user, so a partial-coverage // cold mount still surfaces `avg30LastYear`. - const avg30LastYearMap = await computeAvg30LastYearMap(userId, typesWithData); + const lastYear = await computeAvg30LastYearMap(userId, typesWithData, nowMs); let yearOverYearTypeCount = 0; - for (const [type, value] of avg30LastYearMap.entries()) { + for (const [type, value] of lastYear.entries()) { if (value === null) continue; const summary = summaries[type]; if (!summary) continue; @@ -1205,12 +1152,12 @@ export const WMY_FANOUT_CONCURRENCY = 4; export async function computeAvg30LastYearMap( userId: string, types: ReadonlyArray, + now: number, ): Promise> { const out = new Map(); if (types.length === 0) return out; const limit = pLimit(WMY_FANOUT_CONCURRENCY); const priorityJson = await loadUserSourcePriority(userId); - const now = Date.now(); const results = await Promise.all( types.map((type) => limit(async () => { diff --git a/src/lib/analytics/trend-evidence-sql.ts b/src/lib/analytics/trend-evidence-sql.ts index ed91942ee..9157547d8 100644 --- a/src/lib/analytics/trend-evidence-sql.ts +++ b/src/lib/analytics/trend-evidence-sql.ts @@ -1,36 +1,46 @@ /** - * The trend evidence of the live-SQL slope windows, as SELECT columns. + * The live-SQL slope windows and their trend evidence, as SELECT columns. * * The cold-fallback summaries (`summaries-slice.ts`, the comprehensive * aggregator) fit `slope7` / `slope30` with `REGR_SLOPE` over the canonical * readings. Whether a window may publish that slope is decided in JS by * `toTrendSlope` from the window's distinct reading days and their span (see - * `MIN_TREND_DAYS` in `window-confidence.ts`); these columns hand it exactly - * that, over the same rows and the same UTC-midnight window bounds the - * `REGR_*` FILTER clauses use, so the live path withholds a slope on the - * same evidence the warm rollup path does (`composeWindowedTrend`). + * `MIN_TREND_DAYS` in `window-confidence.ts`); the evidence columns hand it + * exactly that, over the same rows and the same window bounds, so the live + * path withholds a slope on the same evidence the warm rollup path does + * (`composeWindowedTrend`). + * + * The windows anchor on the UTC-midnight day boundary of the request's `now` + * (`date_trunc('day', now) - N days`, the SQL twin of `startOfUtcDay(now - N + * days)`), not on the instant, so the cold path counts the same boundary day + * the warm path composes from: with an instant bound the boundary day flipped + * in and out of the slope window across a warm-to-cold coverage transition. + * `now` is a `boundInstantSql` expression (see `request-clock-sql.ts`). * * `measured_at` is a naive UTC timestamp, so `::date` is its UTC calendar - * day, the day the rollup tier buckets by; `date - date` is whole days. The - * text is a file literal with no input spliced in. It expects the row alias - * `m` and a `GROUP BY` per type. + * day, the day the rollup tier buckets by; `date - date` is whole days. It + * expects the row alias `m` and a `GROUP BY` per type. */ +import { assertInstantSql } from "@/lib/analytics/request-clock-sql"; import { toTrendSlope, type TrendSlope } from "@/lib/analytics/trends"; -function windowFilter(days: 7 | 30): string { - return `m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '${days} days')`; -} - -function evidenceColumns(days: 7 | 30): string { - const filter = windowFilter(days); - return `COUNT(DISTINCT m."measured_at"::date) FILTER (WHERE ${filter})::int AS days${days}, +function windowColumns(now: string, days: 7 | 30): string { + const filter = `m."measured_at" >= (date_trunc('day', ${now}) - INTERVAL '${days} days')`; + const x = `EXTRACT(EPOCH FROM m."measured_at") / 86400.0`; + const r2 = days === 7 ? "r2_7" : "r2_30"; + return `REGR_SLOPE(m."value", ${x}) FILTER (WHERE ${filter})::double precision AS slope${days}, + REGR_R2(m."value", ${x}) FILTER (WHERE ${filter})::double precision AS ${r2}, + COUNT(DISTINCT m."measured_at"::date) FILTER (WHERE ${filter})::int AS days${days}, (MAX(m."measured_at"::date) FILTER (WHERE ${filter}) - MIN(m."measured_at"::date) FILTER (WHERE ${filter}))::int AS span${days}`; } -/** `days7, span7, days30, span30` for a per-type live aggregate. */ -export const TREND_EVIDENCE_SQL = `${evidenceColumns(7)}, - ${evidenceColumns(30)}`; +/** `slope7, r2_7, days7, span7` and the 30-day four for a per-type live aggregate. */ +export function liveTrendSql(now: string): string { + const at = assertInstantSql(now); + return `${windowColumns(at, 7)}, + ${windowColumns(at, 30)}`; +} /** The slope, r² and evidence columns a live per-type aggregate returns. */ export interface LiveTrendColumns { diff --git a/src/lib/analytics/trends.ts b/src/lib/analytics/trends.ts index bedc2bd55..42fe57a1c 100644 --- a/src/lib/analytics/trends.ts +++ b/src/lib/analytics/trends.ts @@ -150,8 +150,8 @@ export interface DataSummary { * for the snapshot builder, …). It is NOT guaranteed to be the trailing * 90-day window. The insights stat strip does not read this value — its * median comes from the SQL `summaries-slice` path, which fixes the - * window to the trailing 90 days (`PERCENTILE_CONT(0.5) … FILTER (WHERE - * measured_at >= NOW() - INTERVAL '90 days')`). A consumer comparing the + * window to the trailing 90 days (`PERCENTILE_CONT(0.5)` over the readings + * from 90 days before the request's `now`). A consumer comparing the * two must account for the differing windows. The percentile *algorithm* * matches (linear-interpolated midpoint), only the window differs. */ diff --git a/src/lib/dashboard/summary-reads.ts b/src/lib/dashboard/summary-reads.ts index 8f8550160..bd1293b93 100644 --- a/src/lib/dashboard/summary-reads.ts +++ b/src/lib/dashboard/summary-reads.ts @@ -258,6 +258,12 @@ export async function readAllTimeCounts( * replaced shifted the column with a single `AT TIME ZONE`, which treats the * UTC value as local time and moves it the wrong way: in Berlin a reading at * 23:30 UTC landed on the day before instead of the day after. + * + * `since` and `now` arrive the same way, as offset-free UTC strings, so they + * are read as UTC (`::timestamp AT TIME ZONE 'UTC'`) before they move into + * `timeZone`. A bare `::timestamptz` read them in the session zone: outside + * a UTC session the series ended a day early and today's reading fell out of + * the streak. */ export async function readActivityDayKeys( db: PrismaClient, @@ -269,8 +275,8 @@ export async function readActivityDayKeys( const rows = await db.$queryRaw>` SELECT to_char(d, 'YYYY-MM-DD') AS day_key FROM generate_series( - (${since}::timestamptz AT TIME ZONE ${timeZone})::date, - (${now}::timestamptz AT TIME ZONE ${timeZone})::date, + ((${since}::timestamp AT TIME ZONE 'UTC') AT TIME ZONE ${timeZone})::date, + ((${now}::timestamp AT TIME ZONE 'UTC') AT TIME ZONE ${timeZone})::date, INTERVAL '1 day' ) AS d WHERE EXISTS ( diff --git a/src/lib/insights/comprehensive-aggregator.ts b/src/lib/insights/comprehensive-aggregator.ts index 7c7926ee6..0c9e035a5 100644 --- a/src/lib/insights/comprehensive-aggregator.ts +++ b/src/lib/insights/comprehensive-aggregator.ts @@ -100,10 +100,14 @@ import { prisma } from "@/lib/db"; import type { DataSummary } from "@/lib/analytics/trends"; import { - TREND_EVIDENCE_SQL, liveTrend, + liveTrendSql, type LiveTrendColumns, } from "@/lib/analytics/trend-evidence-sql"; +import { + boundInstantSql, + windowMeanColumnsSql, +} from "@/lib/analytics/request-clock-sql"; import { annotate } from "@/lib/logging/context"; import type { MeasurementSource, @@ -137,29 +141,9 @@ import { } from "@/lib/rollups/measurement-coverage"; import { dateOnlyKey } from "@/lib/tz/date-only"; -/** - * Window means over canonical readings carrying a `day_weight` (see - * `day-mean.ts`): the mean of the day values for an hourly-mean type (pulse), - * the unchanged `AVG(value)` otherwise. Days are UTC days, the frame the - * canonical-source pick buckets by. - */ -function windowedMean(filter: string): string { - return windowMeanSql({ - typeColumn: 'm."type"', - value: 'm."value"', - weight: "m.day_weight", - filter, - }); -} -const AVG7_SQL = windowedMean( - `m."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '7 days'`, -); -const AVG30_SQL = windowedMean( - `m."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days'`, -); -const AVG30_LAST_MONTH_SQL = windowedMean( - `m."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '60 days' AND m."measured_at" < (NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days'`, -); +/** The request's `now`, bound at `$2` of every live aggregate here. */ +const NOW_SQL = boundInstantSql(2); +const NINETY_DAYS = { from: NOW_SQL, interval: "90 days" } as const; /** * Heavy aggregate row — count/min/max/mean alongside the non-composable @@ -200,11 +184,6 @@ interface NarrowAggregateRow { // they no longer scan the raw measurements partition on the warm path. } -/** The latest canonical reading per type is read only inside the 90-day window. */ -function latestFloor(): Date { - return new Date(Date.now() - 90 * 24 * 60 * 60 * 1000); -} - /** * Result bundle. Everything the route handler needs to assemble the * comprehensive response without holding the 100k-row raw measurement @@ -265,7 +244,9 @@ function round2(value: number | null): number | null { export async function buildComprehensiveAggregate( userId: string, ): Promise { - const ninetyDaysAgo = new Date(Date.now() - 90 * 24 * 60 * 60 * 1000); + // One clock for every window, JS and SQL alike (see `request-clock-sql.ts`). + const now = new Date(); + const ninetyDaysAgo = new Date(now.getTime() - 90 * 24 * 60 * 60 * 1000); // v1.4.38 — per-sub-query wall-clock timing. Emitted under // `meta.insights.sub_*_ms` so the next perf-verify can attribute @@ -297,8 +278,8 @@ export async function buildComprehensiveAggregate( timings["insights.sub_coverage_ms"] = Date.now() - probeStart; const result = isFullyCovered(coverage) - ? await buildFromRollups(userId, ninetyDaysAgo, timings) - : await buildFromLiveAggregate(userId, ninetyDaysAgo, timings); + ? await buildFromRollups(userId, now, ninetyDaysAgo, timings) + : await buildFromLiveAggregate(userId, now, ninetyDaysAgo, timings); annotate({ meta: timings }); return result; @@ -351,10 +332,7 @@ interface BpRawRow { */ async function fetchBpRawRows( userId: string, - // The 90-day window is expressed as a SQL interval inside the canonical - // subquery (parity with the rollup fast-path's `"90 days"`); the bound - // `Date` arg is retained for call-site symmetry and is exactly 90 days. - ninetyDaysAgo: Date, + now: Date, // The source-rank CASE built once by the caller (over unqualified // `"type"`/`"source"` columns) — reused so the BP pull doesn't re-load the // user's source-priority blob the caller already resolved. @@ -364,7 +342,6 @@ async function fetchBpRawRows( dia: Array<{ measuredAt: Date; value: number }>; weight: Array<{ measuredAt: Date; value: number }>; }> { - void ninetyDaysAgo; // v1.32.17 — WEIGHT rides this same canonical-dedup pull (one extra // sparse type, no new round-trip) so the route can derive a local-day // mood × weight series without re-scanning raw rows. @@ -374,11 +351,12 @@ async function fetchBpRawRows( m."type"::text AS type, m."measured_at" AS measured_at, m."value"::double precision AS value - FROM ${canonicalMeasurementsFrom(rankUnqualified, "90 days")} + FROM ${canonicalMeasurementsFrom(rankUnqualified, NINETY_DAYS)} WHERE m."type" IN ('BLOOD_PRESSURE_SYS', 'BLOOD_PRESSURE_DIA', 'WEIGHT') ORDER BY m."measured_at" ASC `, userId, + now, ); const sys: Array<{ measuredAt: Date; value: number }> = []; const dia: Array<{ measuredAt: Date; value: number }> = []; @@ -403,6 +381,7 @@ async function fetchBpRawRows( */ async function buildFromRollups( userId: string, + now: Date, ninetyDaysAgo: Date, timings: Record, ): Promise { @@ -428,7 +407,7 @@ async function buildFromRollups( -- cost; bind it ONCE as a CTE and reference it from both window_stats -- and the outer aggregate so the DISTINCT-ON pick runs a single time -- instead of twice per rebuild. Output is identical. - WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, "90 days")} + WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, NINETY_DAYS)} ), -- The z-score's mean and spread stay over every reading: they measure -- how far one reading sits from the others, a per-reading dispersion, @@ -454,19 +433,17 @@ async function buildFromRollups( (ROUND(((m."value" - ws.mean_value) / ws.stddev_value)::numeric, 2)) ) > 2 ) AS anomaly_count, - ${AVG7_SQL}::double precision AS avg7, - ${AVG30_SQL}::double precision AS avg30, - -- Prior 30d window (days [30, 60) ago) used by tile delta callout. - ${AVG30_LAST_MONTH_SQL}::double precision AS avg30_last_month + ${windowMeanColumnsSql(NOW_SQL)} FROM ${dayWeightedRows("cm", SESSION_DAY_FRAME)} m JOIN window_stats ws ON ws."type" = m."type" GROUP BY m."type", ws.stddev_value `, userId, + now, ), ), timeSubquery(timings, "latests", () => - readLatestCanonical(userId, priorityJson, latestFloor()), + readLatestCanonical(userId, priorityJson, ninetyDaysAgo), ), // v1.4.35 — DAY rollup buckets for the trailing 90 days, every // type. Drives both the per-type `count / min / max / mean` @@ -489,7 +466,7 @@ async function buildFromRollups( // readings; the per-type partition in `fetchBpRawRows` preserves // the legacy `bpRawRows.sys` / `bpRawRows.dia` byte-shape. timeSubquery(timings, "bpRaw", () => - fetchBpRawRows(userId, ninetyDaysAgo, rankUnqualified), + fetchBpRawRows(userId, now, rankUnqualified), ), timeSubquery( timings, @@ -522,11 +499,11 @@ async function buildFromRollups( let totalMeasurements = 0; // v1.20.0 — day-aligned 7/30/90-day window anchors, matching the // DAY-rollup grain. `startOfUtcDay(now − N days)` is the day-aligned - // equivalent of the live `measured_at >= NOW() - INTERVAL 'N days'`. - const now = Date.now(); - const since7 = startOfUtcDay(new Date(now - 7 * 24 * 60 * 60 * 1000)); - const since30 = startOfUtcDay(new Date(now - 30 * 24 * 60 * 60 * 1000)); - const since90 = startOfUtcDay(new Date(now - 90 * 24 * 60 * 60 * 1000)); + // equivalent of the live `measured_at >= now - INTERVAL 'N days'`. + const nowMs = now.getTime(); + const since7 = startOfUtcDay(new Date(nowMs - 7 * 24 * 60 * 60 * 1000)); + const since30 = startOfUtcDay(new Date(nowMs - 30 * 24 * 60 * 60 * 1000)); + const since90 = startOfUtcDay(new Date(nowMs - 90 * 24 * 60 * 60 * 1000)); // v1.20.0 — set once any in-window bucket carries a NULL // accumulator (pre-0190 row, or boot re-fold pending). The slope stays // null then — it converges after backfill — but the miss must be @@ -629,6 +606,7 @@ async function buildFromRollups( */ async function buildFromLiveAggregate( userId: string, + now: Date, ninetyDaysAgo: Date, timings: Record, ): Promise { @@ -651,7 +629,7 @@ async function buildFromLiveAggregate( -- v1.18.11 perf#3a — bind the canonical-source self-join once (see the -- warm narrow query). window_stats + the outer aggregate both read cm, -- so the DISTINCT-ON pick runs once. Output is identical. - WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, "90 days")} + WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, NINETY_DAYS)} ), -- The z-score's mean and spread stay over every reading: they measure -- how far one reading sits from the others, a per-reading dispersion, @@ -679,55 +657,18 @@ async function buildFromLiveAggregate( (ROUND(((m."value" - ws.mean_value) / ws.stddev_value)::numeric, 2)) ) > 2 ) AS anomaly_count, - ${AVG7_SQL}::double precision AS avg7, - ${AVG30_SQL}::double precision AS avg30, - ${AVG30_LAST_MONTH_SQL}::double precision AS avg30_last_month, - -- A2-M2 — the regression windows anchor on the UTC-midnight day - -- boundary, NOT the wall-clock instant, so this cold-fallback path - -- returns the SAME boundary-day membership the warm rollup path - -- composes from (startOfUtcDay(now - N days) + - -- composeWindowedRegression). The expression - -- date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL 'N days' - -- is the SQL twin of startOfUtcDay. It stays a naive UTC timestamp, - -- like measured_at itself, so the comparison never passes through - -- the session time zone. Without this anchor an account that - -- crosses a coverage boundary (warm to cold) saw the boundary day - -- flip in/out of the slope window, giving a cache-dependent answer - -- to the same request. - REGR_SLOPE( - m."value", - EXTRACT(EPOCH FROM m."measured_at") / 86400.0 - ) FILTER ( - WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '7 days') - )::double precision AS slope7, - REGR_R2( - m."value", - EXTRACT(EPOCH FROM m."measured_at") / 86400.0 - ) FILTER ( - WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '7 days') - )::double precision AS r2_7, - REGR_SLOPE( - m."value", - EXTRACT(EPOCH FROM m."measured_at") / 86400.0 - ) FILTER ( - WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days') - )::double precision AS slope30, - REGR_R2( - m."value", - EXTRACT(EPOCH FROM m."measured_at") / 86400.0 - ) FILTER ( - WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days') - )::double precision AS r2_30, - ${TREND_EVIDENCE_SQL} + ${windowMeanColumnsSql(NOW_SQL)}, + ${liveTrendSql(NOW_SQL)} FROM ${dayWeightedRows("cm", SESSION_DAY_FRAME)} m JOIN window_stats ws ON ws."type" = m."type" GROUP BY m."type", ws.stddev_value `, userId, + now, ), ), timeSubquery(timings, "latests", () => - readLatestCanonical(userId, priorityJson, latestFloor()), + readLatestCanonical(userId, priorityJson, ninetyDaysAgo), ), // Buckets may exist for some types even when the COUNT probe came // back zero (race window between the probe and a sibling @@ -745,7 +686,7 @@ async function buildFromLiveAggregate( ), // v1.4.38 — bp sys + dia in one round-trip (see `fetchBpRawRows`). timeSubquery(timings, "bpRaw", () => - fetchBpRawRows(userId, ninetyDaysAgo, rankUnqualified), + fetchBpRawRows(userId, now, rankUnqualified), ), ]); diff --git a/src/lib/measurements/__tests__/series-canonical.test.ts b/src/lib/measurements/__tests__/series-canonical.test.ts index 6f0da6222..310f08f2d 100644 --- a/src/lib/measurements/__tests__/series-canonical.test.ts +++ b/src/lib/measurements/__tests__/series-canonical.test.ts @@ -1,12 +1,15 @@ import { describe, expect, it } from "vitest"; +import { boundInstantSql } from "@/lib/analytics/request-clock-sql"; import { canonicalMeasurementsCte } from "@/lib/analytics/source-rank-sql"; import { seriesRowsFrom } from "../series-canonical"; +const SINCE = new Date("2026-03-01T06:30:00.000Z"); + describe("seriesRowsFrom", () => { it("collapses a type that has a ladder to the ladder's source per day", () => { - const sql = seriesRowsFrom(null, "PULSE", 30); + const sql = seriesRowsFrom(null, "PULSE", SINCE); expect(sql).toContain("DISTINCT ON"); // Scoped to the one metric, in both halves of the helper. expect(sql).toContain(`"type" = 'PULSE'::"measurement_type"`); @@ -15,22 +18,24 @@ describe("seriesRowsFrom", () => { }); it("reads one day wider than the request, so the edge day is picked from all its readings", () => { - expect(seriesRowsFrom(null, "WEIGHT", 30)).toContain("INTERVAL '31 days'"); - expect(seriesRowsFrom(null, "WEIGHT", 3650)).toContain( - "INTERVAL '3651 days'", - ); + const sql = seriesRowsFrom(null, "WEIGHT", SINCE); + expect( + sql.match(/>= '2026-03-01 06:30:00.000'::timestamp - INTERVAL '1 day'/g), + ).toHaveLength(2); + // Counted from the caller's `since`, never from the database clock. + expect(sql).not.toMatch(/NOW\(\)/i); }); it("applies the person's ladder, not only the default one", () => { const first = seriesRowsFrom( { pulse: ["FITBIT", "APPLE_HEALTH"] }, "PULSE", - 7, + SINCE, ); const second = seriesRowsFrom( { pulse: ["APPLE_HEALTH", "FITBIT"] }, "PULSE", - 7, + SINCE, ); expect(first).not.toEqual(second); expect(first).toMatch(/WHEN 'FITBIT' THEN 0 WHEN 'APPLE_HEALTH' THEN 1/); @@ -40,7 +45,7 @@ describe("seriesRowsFrom", () => { it.each(["BLOOD_GLUCOSE", "TOTAL_BODY_WATER", "BONE_MASS"] as const)( "keeps every source for %s, which has no ladder", (type) => { - const sql = seriesRowsFrom(null, type, 30); + const sql = seriesRowsFrom(null, type, SINCE); expect(sql).not.toContain("DISTINCT ON"); expect(sql).toContain(`mm."type" = '${type}'::"measurement_type"`); expect(sql).toContain(`mm."user_id" = $1`); @@ -48,21 +53,24 @@ describe("seriesRowsFrom", () => { }, ); - it.each([0, -1, 1.5, 3651, Number.NaN])("refuses days = %s", (days) => { - expect(() => seriesRowsFrom(null, "PULSE", days)).toThrow(/days/); + it("refuses an invalid instant", () => { + expect(() => seriesRowsFrom(null, "PULSE", new Date(Number.NaN))).toThrow( + /invalid instant/, + ); }); }); describe("canonicalMeasurementsCte, one-metric scope", () => { const rank = `CASE "source" WHEN 'A' THEN 0 ELSE 90 END`; + const NINETY_DAYS = { from: boundInstantSql(2), interval: "90 days" }; it("is unchanged when no type is given", () => { - const sql = canonicalMeasurementsCte(rank, "90 days"); + const sql = canonicalMeasurementsCte(rank, NINETY_DAYS); expect(sql).not.toContain("measurement_type"); }); it("scopes the pick and the join to the type when one is given", () => { - const sql = canonicalMeasurementsCte(rank, "90 days", "PULSE"); + const sql = canonicalMeasurementsCte(rank, NINETY_DAYS, "PULSE"); expect(sql.match(/'PULSE'::"measurement_type"/g)).toHaveLength(2); }); diff --git a/src/lib/measurements/series-canonical.ts b/src/lib/measurements/series-canonical.ts index 963703aeb..422a972cc 100644 --- a/src/lib/measurements/series-canonical.ts +++ b/src/lib/measurements/series-canonical.ts @@ -22,20 +22,18 @@ import { buildSourceRankCase, canonicalMeasurementsFrom, } from "@/lib/analytics/source-rank-sql"; +import { instantLiteralSql } from "@/lib/analytics/request-clock-sql"; const ENUM_RE = /^[A-Z0-9_]+$/; export function seriesRowsFrom( priorityJson: unknown, type: MeasurementType, - days: number, + since: Date, ): string { if (!ENUM_RE.test(type)) { throw new Error(`unsafe enum literal for SQL splice: ${type}`); } - if (!Number.isInteger(days) || days < 1 || days > 3650) { - throw new Error(`days out of range for SQL splice: ${days}`); - } if (!RANKED_TYPES.includes(type)) { return `( SELECT mm.* @@ -47,10 +45,11 @@ export function seriesRowsFrom( } // One day wider than the request: the caller trims to its exact `since`, so // the edge day's source is picked from all of its readings, not from the - // slice the interval happens to include. + // slice the window happens to include. The window counts from the caller's + // `since`, its own clock, never from the database's. return canonicalMeasurementsFrom( buildSourceRankCase(priorityJson, '"type"', '"source"'), - `${days + 1} days`, + { from: instantLiteralSql(since), interval: "1 day" }, type, ); } diff --git a/src/lib/measurements/series/read-bucketed.ts b/src/lib/measurements/series/read-bucketed.ts index 9dc49a0d5..687571c5c 100644 --- a/src/lib/measurements/series/read-bucketed.ts +++ b/src/lib/measurements/series/read-bucketed.ts @@ -66,7 +66,7 @@ export async function readBucketedSeries( loadUserSourcePriority(userId), ]); const type = KIND_TO_TYPE[kind]; - const seriesRows = seriesRowsFrom(priorityJson, type, days); + const seriesRows = seriesRowsFrom(priorityJson, type, since); // A day of a type with an activity-dependent sampling rate (pulse) is the // weighted mean of its local hours' means, so a workout hour with hundreds // of times the readings does not outweigh the rest of the day (see diff --git a/src/lib/measurements/series/read-raw.ts b/src/lib/measurements/series/read-raw.ts index 39a1db0b6..212c101d1 100644 --- a/src/lib/measurements/series/read-raw.ts +++ b/src/lib/measurements/series/read-raw.ts @@ -51,7 +51,7 @@ export async function readRawSeries( >( ` SELECT m."id", m."value", m."measured_at", m."value_min", m."value_max" - FROM ${seriesRowsFrom(sourcePriority, type, days)} + FROM ${seriesRowsFrom(sourcePriority, type, since)} WHERE m."measured_at" >= $2 ORDER BY m."measured_at" ASC, m."id" ASC `, diff --git a/src/lib/rollups/measurement-read.ts b/src/lib/rollups/measurement-read.ts index 26780b4d1..660c80f87 100644 --- a/src/lib/rollups/measurement-read.ts +++ b/src/lib/rollups/measurement-read.ts @@ -331,7 +331,7 @@ export interface AccumulatorBucketRow extends RegressionAccumulators { * The window is DAY-aligned (a bucket is in-window iff its `bucketStart` * is `>= since`), matching the DAY-rollup grain. Callers anchor `since` on * a UTC-day boundary (`startOfUtcDay(now − N days)`) so the window is the - * day-aligned equivalent of the live `measured_at >= NOW() - INTERVAL 'N + * day-aligned equivalent of the live `measured_at >= now - INTERVAL 'N * days'` filter. The accumulator-composed result over a given bucket set * is bit-identical to live REGR_* / STDDEV_POP over the same buckets' raw * rows — the parity test pins this. @@ -354,7 +354,7 @@ export function composeWindowedRegression( * regression `composeWindowedRegression` composes, withheld by `toTrendSlope` * when the window's days with a reading are too few or span too little of * `windowDays` (`isTrendWindowSufficient`). Days are the buckets' UTC days, - * the same days the live path counts with `TREND_EVIDENCE_SQL`. + * the same days the live path counts with `liveTrendSql`. */ export function composeWindowedTrend( rows: ReadonlyArray, diff --git a/tests/integration/activity-calendar-read.test.ts b/tests/integration/activity-calendar-read.test.ts index 230281993..2c6c86c42 100644 --- a/tests/integration/activity-calendar-read.test.ts +++ b/tests/integration/activity-calendar-read.test.ts @@ -152,7 +152,6 @@ describe("readActivityDays", () => { tz: TZ, fromKey, toKey, - now: NOW, includeWorkouts: true, }); const by = new Map(days.map((d) => [d.date, d])); diff --git a/tests/integration/canonical-source-pick-parity.test.ts b/tests/integration/canonical-source-pick-parity.test.ts index 69eaad9de..da80698f1 100644 --- a/tests/integration/canonical-source-pick-parity.test.ts +++ b/tests/integration/canonical-source-pick-parity.test.ts @@ -2,7 +2,9 @@ * The canonical-source pick, against a real Postgres: the shipped statement * (group the window to `(type, day, source)` first, then `DISTINCT ON`) * against the statement it replaced (`DISTINCT ON` over every raw row), - * kept verbatim below as the reference. + * kept verbatim below as the reference. Only its window anchor changed: it + * counts from the same bound instant the shipped helper takes, where it read + * `NOW()`, so both cut one window. * * Both must return exactly the same rows, for every reader shape the helper * serves (all types or one type, a trailing window or none), under the @@ -18,6 +20,7 @@ import type { MeasurementSource, MeasurementType, } from "@/generated/prisma/client"; +import { boundInstantSql } from "@/lib/analytics/request-clock-sql"; import { buildSourceRankCase, canonicalMeasurementsCte, @@ -29,6 +32,8 @@ import { getPrismaClient, truncateAllTables } from "./setup"; const USER_ID = "user-canonical-parity"; const DAY_MS = 86_400_000; +/** The instant both statements count their window from, bound at `$2`. */ +const NOW_SQL = boundInstantSql(2); /** The pick as it stood before it grouped first. */ function legacyCanonicalMeasurementsCte( @@ -37,10 +42,10 @@ function legacyCanonicalMeasurementsCte( onlyType?: MeasurementType, ): string { const sinceInner = sinceInterval - ? `AND "measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '${sinceInterval}'` + ? `AND "measured_at" >= ${NOW_SQL} - INTERVAL '${sinceInterval}'` : ""; const sinceOuter = sinceInterval - ? `AND mm."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '${sinceInterval}'` + ? `AND mm."measured_at" >= ${NOW_SQL} - INTERVAL '${sinceInterval}'` : ""; const typeInner = onlyType ? `AND "type" = '${onlyType}'::"measurement_type"` @@ -200,15 +205,18 @@ describe("canonical-source pick parity", () => { it(`returns the same rows: ${ladderLabel}, ${shapeLabel}`, async () => { const prisma = getPrismaClient(); const rank = buildSourceRankCase(ladder, '"type"', '"source"'); + const at = new Date(); + const window = since ? { from: NOW_SQL, interval: since } : undefined; const select = (cte: string) => prisma.$queryRawUnsafe< Array<{ id: string; type: string; source: string }> >( `WITH cm AS (${cte}) SELECT id, type::text AS type, source::text AS source FROM cm ORDER BY id`, USER_ID, + ...(window ? [at] : []), ); const [now, before] = await Promise.all([ - select(canonicalMeasurementsCte(rank, since, onlyType)), + select(canonicalMeasurementsCte(rank, window, onlyType)), select(legacyCanonicalMeasurementsCte(rank, since, onlyType)), ]); expect(now.length).toBeGreaterThan(0); diff --git a/tests/integration/dense-series-read-parity.test.ts b/tests/integration/dense-series-read-parity.test.ts index 09f7c745b..d8d822420 100644 --- a/tests/integration/dense-series-read-parity.test.ts +++ b/tests/integration/dense-series-read-parity.test.ts @@ -158,13 +158,14 @@ describe("dense-series read parity", () => { for (const [type, grain, days] of CASES) { for (const userTz of ZONES) { it(`matches the two statements: ${type}, ${grain}, ${days} d, ${userTz}`, async () => { + const since = new Date(Date.now() - days * DAY_MS); const args = { userId, type, grain, userTz, - since: new Date(Date.now() - days * DAY_MS), - seriesRows: seriesRowsFrom(null, type, days), + since, + seriesRows: seriesRowsFrom(null, type, since), }; const [now, before] = await Promise.all([ readDenseSeries(args), @@ -191,13 +192,14 @@ describe("dense-series read parity", () => { it("answers no figures row for an hourly-mean type without readings, as before", async () => { const empty = await makeUser(); + const since = new Date(Date.now() - 365 * DAY_MS); const args = { userId: empty.id, type: "PULSE" as const, grain: "day" as const, userTz: "UTC", - since: new Date(Date.now() - 365 * DAY_MS), - seriesRows: seriesRowsFrom(null, "PULSE", 365), + since, + seriesRows: seriesRowsFrom(null, "PULSE", since), }; const [now, before] = await Promise.all([ readDenseSeries(args), diff --git a/tests/integration/request-clock-windows-session-tz.test.ts b/tests/integration/request-clock-windows-session-tz.test.ts new file mode 100644 index 000000000..3b0d9aea7 --- /dev/null +++ b/tests/integration/request-clock-windows-session-tz.test.ts @@ -0,0 +1,32 @@ +/** + * The summary windows count from the request's clock in a session outside + * UTC too. The app pins its sessions to UTC; this runs the pool in another + * zone on purpose, so a bound that went through `timestamptz` (and so through + * the session zone) would be cut hours off (see + * `request-clock-windows.cases.ts`). + */ +import { describe, vi } from "vitest"; + +vi.mock("@/lib/db-session", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, TIME_ZONE_OPTION: "-c TimeZone=Pacific/Auckland" }; +}); + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +// The read path's fire-and-forget refresh would build the rollup tier behind +// the live case; the cases build it themselves when they mean to. +vi.mock("@/lib/rollups/measurement-rollups", async (importOriginal) => ({ + ...(await importOriginal< + typeof import("@/lib/rollups/measurement-rollups") + >()), + ensureUserRollupsFresh: async () => {}, +})); + +import { defineRequestClockCases } from "./request-clock-windows.cases"; + +describe("summary windows on the request's clock, outside UTC", () => { + defineRequestClockCases("Pacific/Auckland"); +}); diff --git a/tests/integration/request-clock-windows.cases.ts b/tests/integration/request-clock-windows.cases.ts new file mode 100644 index 000000000..828ecd18e --- /dev/null +++ b/tests/integration/request-clock-windows.cases.ts @@ -0,0 +1,172 @@ +/** + * Shared cases for the request-clock suites: the summary windows count from + * the clock of the request, never from the database's. + * + * The summaries slice and the comprehensive aggregator cut some windows in JS + * and the rest in SQL. The SQL side read `NOW()`, so with the process clock + * pinned (as every parity walk does) the 7-, 30- and 60-day means and the + * 90-day canonical pick were cut hours or days away from the JS windows, and + * the walks passed or failed with the hour they ran at. + * + * Each case pins the process clock to a day long before the database's own + * clock, at four times of day, and lays readings ten minutes either side of + * every window edge. Counted from the pinned clock the figures are exact; + * counted from the database clock every window would be empty. Both read + * paths run: the live aggregate before any rollup exists, the rollup path + * after the DAY tier is built. The streak's local days ride along: their + * bounds are instants too, and a session outside UTC moved them. + * + * Not a test file. Only for tests under `tests/integration/`. + */ +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; + +import { + computeSummariesSlice, + readSevenDayMeans, +} from "@/lib/analytics/summaries-slice"; +import { readActivityDayKeys } from "@/lib/dashboard/summary-reads"; +import { buildComprehensiveAggregate } from "@/lib/insights/comprehensive-aggregator"; +import { settleBackgroundTasks } from "@/lib/logging/background-tasks"; +import { + isFullyCovered, + probeRollupCoverage, +} from "@/lib/rollups/measurement-coverage"; +import { userDayKey } from "@/lib/tz/format"; + +import { makeUser } from "../kit/factories"; +import { getPrismaClient, truncateAllTables } from "./setup"; + +const DAY = 86_400_000; +const MINUTE = 60_000; + +/** A day far from any database clock the suite will meet. */ +const PINNED_DAY = Date.UTC(2025, 2, 12); +const TIMES_OF_DAY: Array<[string, number]> = [ + ["00:30", 30 * MINUTE], + ["09:00", 9 * 60 * MINUTE], + ["12:00", 12 * 60 * MINUTE], + ["23:30", (23 * 60 + 30) * MINUTE], +]; + +/** + * Weight readings (one source, so the canonical pick keeps them all) placed + * ten minutes inside or outside an edge, as offsets from `now`. + */ +const READINGS: Array<[offsetMs: number, value: number]> = [ + [-60 * MINUTE, 80], + [-7 * DAY + 10 * MINUTE, 80], // inside the 7-day window + [-7 * DAY - 10 * MINUTE, 100], // outside it, inside the 30-day one + [-30 * DAY - 10 * MINUTE, 90], // the month before + [-60 * DAY + 10 * MINUTE, 90], // the month before, at its start + [-60 * DAY - 10 * MINUTE, 200], // older than the month before + [-90 * DAY + 10 * MINUTE, 70], // inside the 90-day window + [-90 * DAY - 10 * MINUTE, 300], // outside it +]; + +/** The figures the readings give when every window counts from `now`. */ +const EXPECTED = { + avg7: 80, + avg30: Math.round(((80 + 80 + 100) / 3) * 100) / 100, + avg30LastMonth: 90, + /** Median of 70, 80, 80, 90, 90, 100, 200. */ + median: 90, + /** Readings inside the 90-day window. */ + inNinetyDays: 7, +}; + +export function defineRequestClockCases(expectedZone: string): void { + const prisma = getPrismaClient(); + + afterAll(() => { + vi.useRealTimers(); + }); + + it(`runs in the ${expectedZone} session zone`, async () => { + const [{ zone }] = await prisma.$queryRaw>` + SELECT current_setting('TimeZone') AS zone + `; + expect(zone).toBe(expectedZone); + }); + + describe.each(TIMES_OF_DAY)("with the clock at %s UTC", (_label, time) => { + const now = new Date(PINNED_DAY + time); + let userId = ""; + + beforeAll(async () => { + vi.useFakeTimers({ toFake: ["Date"], now }); + await truncateAllTables(prisma); + const user = await makeUser({ timezone: "UTC" }); + userId = user.id; + await prisma.measurement.createMany({ + data: READINGS.map(([offset, value]) => { + const measuredAt = new Date(now.getTime() + offset); + return { + userId, + type: "WEIGHT" as const, + value, + unit: "kg", + measuredAt, + createdAt: measuredAt, + }; + }), + }); + }); + + afterAll(async () => { + await settleBackgroundTasks(); + vi.useRealTimers(); + }); + + async function expectWindows(path: "live" | "rollup"): Promise { + expect(isFullyCovered(await probeRollupCoverage(userId))).toBe( + path === "rollup", + ); + + const slice = await computeSummariesSlice(userId); + await settleBackgroundTasks(); + const weight = slice.summaries.WEIGHT; + expect(weight.avg7).toBe(EXPECTED.avg7); + expect(weight.avg30).toBe(EXPECTED.avg30); + expect(weight.avg30LastMonth).toBe(EXPECTED.avg30LastMonth); + expect(weight.median).toBe(EXPECTED.median); + // The staleness of the latest reading counts from the same clock. + expect(slice.lastSeenByType.WEIGHT?.daysAgo).toBe(0); + + const aggregate = await buildComprehensiveAggregate(userId); + await settleBackgroundTasks(); + const summary = aggregate.summaries.WEIGHT; + expect(summary?.avg7).toBe(EXPECTED.avg7); + expect(summary?.avg30).toBe(EXPECTED.avg30); + expect(summary?.avg30LastMonth).toBe(EXPECTED.avg30LastMonth); + expect(aggregate.weightRawRows).toHaveLength(EXPECTED.inNinetyDays); + + expect((await readSevenDayMeans(userId)).WEIGHT).toBe(EXPECTED.avg7); + } + + it("keys the streak's local days on the request's clock", async () => { + const keys = await readActivityDayKeys( + prisma, + userId, + "UTC", + new Date(now.getTime() - 3 * DAY), + now, + ); + expect(keys).toEqual([ + userDayKey(new Date(now.getTime() - 60 * MINUTE), "UTC"), + ]); + }); + + it("cuts every window of the live aggregate on the request's clock", async () => { + await expectWindows("live"); + }); + + it("cuts every window of the rollup path on the request's clock", async () => { + const rollups = await vi.importActual< + typeof import("@/lib/rollups/measurement-rollups") + >("@/lib/rollups/measurement-rollups"); + await rollups.ensureUserRollupsFresh(userId); + await settleBackgroundTasks(); + await expectWindows("rollup"); + }); + }); +} diff --git a/tests/integration/request-clock-windows.test.ts b/tests/integration/request-clock-windows.test.ts new file mode 100644 index 000000000..58e138433 --- /dev/null +++ b/tests/integration/request-clock-windows.test.ts @@ -0,0 +1,20 @@ +/** + * The summary windows count from the request's clock, in the UTC session + * every connection runs in (see `request-clock-windows.cases.ts`). + */ +import { describe, vi } from "vitest"; + +// The read path's fire-and-forget refresh would build the rollup tier behind +// the live case; the cases build it themselves when they mean to. +vi.mock("@/lib/rollups/measurement-rollups", async (importOriginal) => ({ + ...(await importOriginal< + typeof import("@/lib/rollups/measurement-rollups") + >()), + ensureUserRollupsFresh: async () => {}, +})); + +import { defineRequestClockCases } from "./request-clock-windows.cases"; + +describe("summary windows on the request's clock", () => { + defineRequestClockCases("UTC"); +}); diff --git a/tests/integration/routes/parity/analytics.parity.test.ts b/tests/integration/routes/parity/analytics.parity.test.ts index 18e8b83e9..72fb3c738 100644 --- a/tests/integration/routes/parity/analytics.parity.test.ts +++ b/tests/integration/routes/parity/analytics.parity.test.ts @@ -15,8 +15,8 @@ * event, the rollup refreshes the request handed off, and the stored health * scores and correlation patterns. * - * The clock is pinned (`Date` only) to noon of the day the test runs; see - * `NOW`. The record's rollups are built once + * The clock is pinned (`Date` only) to noon of a fixed day; see `NOW`. The + * record's rollups are built once * before the walk; the fire-and-forget refresh the route starts on every * request is recorded instead of run, so the wide event never depends on * whether it finished first. @@ -32,9 +32,9 @@ import { effectRecorder, eventFacts, expectParity, + FIXED_ANCHOR_DAY, relativeToAnchor, stable, - startOfUtcToday, wideEventRecorder, } from "../../../kit/parity"; import { getPrismaClient, truncateAllTables } from "../../setup"; @@ -66,13 +66,13 @@ vi.mock("@/lib/rollups/measurement-rollups", async (importOriginal) => { const prisma = getPrismaClient(); /** - * Noon of the current UTC day. The summaries read their 90-day windows in SQL - * against the database clock, which a pinned `Date` cannot move, so the clock - * is pinned to today and the snapshot records positions relative to today's - * start (`relativeToAnchor`). The record is under sixty days old, so no - * reading sits on a SQL window's edge whatever the time of day. + * Noon of a fixed day. Every window the route cuts, in JS and in SQL, counts + * from the request's clock, which the pin reaches, so the walk reads the same + * at any hour and on any date it runs; the weekday-keyed figures need the + * fixed date (`FIXED_ANCHOR_DAY`). The snapshot records positions relative to + * the day's start (`relativeToAnchor`). */ -const ANCHOR = startOfUtcToday(); +const ANCHOR = FIXED_ANCHOR_DAY; const NOW = new Date(ANCHOR.getTime() + 12 * 60 * 60 * 1000); const TZ = "UTC"; const OWNER_ID = parityId("analytics", 1); diff --git a/tests/integration/routes/parity/dashboard-summary.parity.test.ts b/tests/integration/routes/parity/dashboard-summary.parity.test.ts index e70743b6b..7467d68fe 100644 --- a/tests/integration/routes/parity/dashboard-summary.parity.test.ts +++ b/tests/integration/routes/parity/dashboard-summary.parity.test.ts @@ -9,9 +9,11 @@ * read it, without credentials, and for an account with no data. The * fire-and-forget rollup refresh is recorded instead of run. * - * The clock is pinned to noon of the current UTC day and the snapshot is - * written relative to the start of that day (`relativeToAnchor`), because - * the per-type reads cut their windows in SQL against the database clock. + * The clock is pinned to noon of a fixed day (`FIXED_ANCHOR_DAY`: the + * chronotype reads work and free nights by weekday) and the snapshot is + * written relative to the start of that day (`relativeToAnchor`). Every + * window the summary cuts, in JS and in SQL, counts from the request's + * clock, so the walk reads the same at any hour it runs. */ import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; @@ -24,9 +26,9 @@ import { effectRecorder, eventFacts, expectParity, + FIXED_ANCHOR_DAY, relativeToAnchor, stable, - startOfUtcToday, wideEventRecorder, } from "../../../kit/parity"; import { @@ -60,7 +62,7 @@ vi.mock("@/lib/rollups/measurement-rollups", async (importOriginal) => { }); const prisma = getPrismaClient(); -const ANCHOR = startOfUtcToday(); +const ANCHOR = FIXED_ANCHOR_DAY; const NOW = new Date(ANCHOR.getTime() + 12 * 60 * 60 * 1000); const TZ = "UTC"; const OWNER_ID = parityId("dashsum", 1); @@ -170,11 +172,16 @@ describe("GET /api/dashboard/summary parity", () => { modulePreferencesJson: { glucose: false, sleep: false, mood: false }, }, }); - await step("03-modules-off"); - await prisma.user.update({ - where: { id: OWNER_ID }, - data: { modulePreferencesJson: {} }, - }); + try { + await step("03-modules-off"); + } finally { + // Restored even when the case fails, so the cases after it are not + // read with the modules still off. + await prisma.user.update({ + where: { id: OWNER_ID }, + data: { modulePreferencesJson: {} }, + }); + } }); it("04 another language, served from the cache", async () => { diff --git a/tests/integration/routes/parity/insights-comprehensive.parity.test.ts b/tests/integration/routes/parity/insights-comprehensive.parity.test.ts index 1518d3c8b..e8275e128 100644 --- a/tests/integration/routes/parity/insights-comprehensive.parity.test.ts +++ b/tests/integration/routes/parity/insights-comprehensive.parity.test.ts @@ -14,12 +14,12 @@ * resting series, a medication with no schedule, an as-needed one, and no * height. An account with no data closes the walk. * - * The clock is pinned (`Date` only) to noon of the day the test runs. The - * aggregator cuts its 7-, 30- and 60-day windows in SQL against the - * database clock, which the pin cannot reach, so the record's measurements - * on exactly those days back are removed: a reading there would fall in or - * out of a window with the hour the test runs. Every instant and day key in - * the snapshot is written relative to the start of the day. + * The clock is pinned (`Date` only) to noon of a fixed day + * (`FIXED_ANCHOR_DAY`); every window the aggregator cuts, in JS and in SQL, + * counts from that clock. The record leaves out its measurements on the 7th, + * 30th and 60th day back, from when the aggregator cut those windows against + * the database clock; the snapshot was recorded so. Every instant and day key + * in the snapshot is written relative to the start of the day. * * The rollup refreshes the route starts and does not await are recorded * instead of run; the record's rollups are built once before the walk. @@ -42,9 +42,9 @@ import { effectRecorder, eventFacts, expectParity, + FIXED_ANCHOR_DAY, relativeToAnchor, stable, - startOfUtcToday, wideEventRecorder, } from "../../../kit/parity"; import { @@ -92,9 +92,9 @@ vi.mock("@/lib/rollups/mood-rollups", async (importOriginal) => { const prisma = getPrismaClient(); const DAY = 86_400_000; -const ANCHOR = startOfUtcToday(); +const ANCHOR = FIXED_ANCHOR_DAY; const NOW = new Date(ANCHOR.getTime() + 12 * 60 * 60 * 1000); -/** Days back on which a SQL window edge falls (see the header). */ +/** Days back the record leaves out (see the header). */ const EDGE_DAYS = [7, 30, 60]; const OWNER_ID = parityId("insightscomp", 1); diff --git a/tests/integration/source-ladder-new-sources.test.ts b/tests/integration/source-ladder-new-sources.test.ts index 0819ebff8..963ca9d00 100644 --- a/tests/integration/source-ladder-new-sources.test.ts +++ b/tests/integration/source-ladder-new-sources.test.ts @@ -20,6 +20,7 @@ import { buildSourceRankCase, canonicalMeasurementsFrom, } from "@/lib/analytics/source-rank-sql"; +import { boundInstantSql } from "@/lib/analytics/request-clock-sql"; vi.mock("@/lib/db-compat", () => ({ ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), @@ -56,9 +57,10 @@ async function canonicalSources(): Promise> { `SELECT DISTINCT m."type"::text AS type, date_trunc('day', m."measured_at") AS day, m."source"::text AS source - FROM ${canonicalMeasurementsFrom(rank, "90 days")} + FROM ${canonicalMeasurementsFrom(rank, { from: boundInstantSql(2), interval: "90 days" })} ORDER BY 1, 2, 3`, USER, + new Date(), ); const out = new Map(); for (const r of rows) { diff --git a/tests/integration/source-provenance.test.ts b/tests/integration/source-provenance.test.ts index 5e2c93052..14b04ae3c 100644 --- a/tests/integration/source-provenance.test.ts +++ b/tests/integration/source-provenance.test.ts @@ -15,6 +15,7 @@ import { buildSourceRankCase, canonicalMeasurementsCte, } from "@/lib/analytics/source-rank-sql"; +import { instantLiteralSql } from "@/lib/analytics/request-clock-sql"; import { readDayProvenance, readSourceComparison, @@ -102,7 +103,7 @@ beforeAll(async () => { async function canonicalSource(type: string, date: string, priority: unknown) { const rank = buildSourceRankCase(priority, '"type"', '"source"'); const found = await prisma.$queryRawUnsafe>( - `SELECT DISTINCT m."source"::text AS source FROM (${canonicalMeasurementsCte(rank, "30 days")}) m + `SELECT DISTINCT m."source"::text AS source FROM (${canonicalMeasurementsCte(rank, { from: instantLiteralSql(new Date()), interval: "30 days" })}) m WHERE m."type"::text = $2 AND to_char(date_trunc('day', m."measured_at"), 'YYYY-MM-DD') = $3`, USER, type, diff --git a/tests/kit/parity.ts b/tests/kit/parity.ts index a778ace23..d592155f4 100644 --- a/tests/kit/parity.ts +++ b/tests/kit/parity.ts @@ -110,17 +110,31 @@ export function startOfUtcToday(): Date { return new Date(Math.floor(Date.now() / DAY_MS) * DAY_MS); } +/** + * A fixed anchor day, for a walk whose route cuts every window on the + * request's clock (`src/lib/analytics/request-clock-sql.ts`), so a pinned + * `Date` reaches all of them. + * + * A record laid out from today is not the same record on every day: its + * weekdays move with the date. The chronotype's work and free nights and the + * weight-by-weekday points follow them, so a snapshot taken on a Friday + * failed on a Saturday. A Friday, the day those walks were first recorded. + * A walk that still meets `NOW()` keeps `startOfUtcToday`: against a fixed + * day, the database clock would sit months away. + */ +export const FIXED_ANCHOR_DAY = new Date("2026-10-23T00:00:00.000Z"); + /** * A copy of `value` with every ISO instant written as days and time from * `anchor` (`anchor-13d+06:30:00`) and every `YYYY-MM-DD` day key as days from * the anchor's day (`day-13`), inside any string. * - * For a walk whose route reads windows against the database clock, which a - * pinned `Date` cannot reach: the record is laid out from the start of the - * current UTC day, the clock is pinned to that day, and the snapshot holds - * only positions relative to it, so it reads the same on every day it runs. - * Keep the account in UTC, so a daylight-saving change between two runs - * cannot move a local day boundary against the anchor. + * The record is laid out from the anchor day (`FIXED_ANCHOR_DAY`, or + * `startOfUtcToday` for a walk whose route reads windows against the database + * clock, which a pinned `Date` cannot reach), the clock is pinned to that day, + * and the snapshot holds only positions relative to it. Keep the account in + * UTC, so a daylight-saving change between two runs cannot move a local day + * boundary against the anchor. */ export function relativeToAnchor(value: unknown, anchor: Date): unknown { const base = anchor.getTime();