mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-12 06:06:10 +00:00
Marc-André Bombeck
5fd4333017
The pulse and per-metric status snapshots read the latest 365 rows, which for a dense pulse stream is a few hours, so their 30-day figures were computed over part of one day; they now read at least the last 30 days. The general overview read the newest 5000 rows across every type, so a dense stream crowded the other types out within days; it now reads the latest 365 of each type with one index descent per type. The sleep summary in the analytics slice read one year of stage rows, so its all-time count and its year-ago average fell short; it now walks the whole history in chunks.
1315 lines
52 KiB
TypeScript
1315 lines
52 KiB
TypeScript
/**
|
|
* v1.4.36 — slim `summaries` slice for `/api/analytics?slice=summaries`.
|
|
*
|
|
* The default `/api/analytics` path fans out 30+ `Promise.all`-wrapped
|
|
* chunked findMany reads (one per `MeasurementType`) so it can hand
|
|
* each series to `summarize()` for slope7/30/90 + anomaly detection +
|
|
* lastMonth/lastYear means. The dashboard tile strip only consumes
|
|
* `count`, `latest`, `avg7`, `avg30`, `min`, `max`, `mean`, and the
|
|
* three slope tuples — every other field gets `null`-coalesced before
|
|
* it ever paints — yet the user waits on the full chunked walk.
|
|
*
|
|
* Read shape (v1.4.36)
|
|
* --------------------
|
|
* v1.4.35 partially read-swapped onto the persistent `measurement_rollups`
|
|
* table but kept the full-fat `$queryRaw` running in parallel as a
|
|
* parity check. The heavy COUNT/MIN/MAX/AVG aggregate scanned the
|
|
* measurements table every request even on warm rollups.
|
|
*
|
|
* v1.4.36 promotes the rollup table to the canonical source on the
|
|
* happy path: a single cheap COUNT against `measurement_rollups`
|
|
* decides whether to read the composable stats from buckets or fall
|
|
* back to the heavy live aggregate. When the rollup is populated:
|
|
*
|
|
* 1. **DAY-bucket read** — composes `count / min / max / mean` per
|
|
* type via `aggregateBuckets`. The four are linearly composable
|
|
* across DAY buckets so the composed value is mathematically
|
|
* identical to the live aggregate over the same rows.
|
|
*
|
|
* 2. **Narrow `$queryRaw`** — carries only the non-composable
|
|
* windowed columns (`avg7`, `avg30`, and the slope tuples). The
|
|
* heavy COUNT/MIN/MAX/AVG columns are NOT projected.
|
|
*
|
|
* 3. **Latest read** (`readLatestCanonical`) — most-recent `(type,
|
|
* value, measured_at)` triplet per type for the `latest` field
|
|
* and the `lastSeenByType` freshness map.
|
|
*
|
|
* On the cold fallback (no rollup rows yet) the slim slice runs the
|
|
* legacy heavy aggregate so the response shape is correct on the very
|
|
* first request.
|
|
*
|
|
* Fields the slim shape intentionally omits (the dashboard never
|
|
* reads them on first paint):
|
|
* - `anomalyCount`: would require an extra per-type read for the
|
|
* z-score loop. v1.37.19 removed it from the wire shape entirely —
|
|
* the insights pipeline computes its own via `summarize()` /
|
|
* the comprehensive aggregator, and no client component reads it.
|
|
* - `avg30LastMonth` / `avg30LastYear`: the dashboard tile delta
|
|
* callout uses them, but only when the comparison-baseline
|
|
* widget is enabled — that path already pre-fetches the default
|
|
* slice via the explicit comparison opt-in.
|
|
*
|
|
* Slope `confidence` is the SQL `regr_r2` value (R²) which matches
|
|
* the existing `trendSlope()` helper's `Math.round(rSquared * 100) /
|
|
* 100`. Slope `direction` is derived from `slope`'s sign with the
|
|
* same 0.01-units-per-day "stable" threshold the JS helper uses.
|
|
*/
|
|
import pLimit from "p-limit";
|
|
|
|
import type { MeasurementType } from "@/generated/prisma/client";
|
|
import { Prisma } from "@/generated/prisma/client";
|
|
|
|
import { prisma } from "@/lib/db";
|
|
import {
|
|
HOURLY_MEAN_DAY_TYPES,
|
|
usesHourlyMeanDay,
|
|
windowWeighting,
|
|
} from "@/lib/measurements/day-statistic";
|
|
import {
|
|
SESSION_DAY_FRAME,
|
|
dayValue,
|
|
dayWeightedRows,
|
|
dayWeightedRowsSql,
|
|
foldMeanSql,
|
|
isHourlyMeanTypeSql,
|
|
windowMeanSql,
|
|
} from "@/lib/measurements/day-mean";
|
|
import type { DataSummary } from "@/lib/analytics/trends";
|
|
import { measurementTypeEnum } from "@/lib/validations/measurement";
|
|
import { annotate } from "@/lib/logging/context";
|
|
import { caches, type ServerCache } from "@/lib/cache/server-cache";
|
|
import {
|
|
ensureUserRollupsFresh,
|
|
readRollupBuckets,
|
|
ROLLUP_FOLD_WINDOW_MS,
|
|
} from "@/lib/rollups/measurement-rollups";
|
|
import {
|
|
aggregateBuckets,
|
|
collapseRollupRowsBySource,
|
|
composeWindowedRegression,
|
|
hasPendingAccumulatorBackfill,
|
|
loadUserSourcePriority,
|
|
type AccumulatorBucketRow,
|
|
} from "@/lib/rollups/measurement-read";
|
|
import { startOfUtcDay } from "@/lib/tz/start-of-utc-day";
|
|
import {
|
|
buildSourceRankCase,
|
|
canonicalMeasurementsCte,
|
|
} from "@/lib/analytics/source-rank-sql";
|
|
import { readLatestCanonical } from "@/lib/measurements/latest-canonical";
|
|
import {
|
|
isFullyCovered,
|
|
probeRollupCoverage,
|
|
type RollupCoverageMap,
|
|
} from "@/lib/rollups/measurement-coverage";
|
|
import {
|
|
summarizeNights,
|
|
type SleepSourceDiscrepancy,
|
|
} from "@/lib/analytics/sleep-night";
|
|
import { resolveUserTimezone } from "@/lib/tz/resolver";
|
|
import { readSleepNights } from "@/lib/measurements/list/sleep-nights";
|
|
import { userDayKey } from "@/lib/tz/format";
|
|
import { localDayWindow } from "@/lib/tz/local-day";
|
|
import { canonicalRowsOfDay } from "@/lib/mood/linked-context";
|
|
|
|
const DAY_MS = 24 * 60 * 60 * 1000;
|
|
|
|
/** No account holds this many nights; the walk is uncapped in practice. */
|
|
const ALL_NIGHTS = 100_000;
|
|
|
|
/**
|
|
* v1.18.10 — upper bound on the cold-fallback `allTime` aggregate's
|
|
* scan window. Sits far beyond any real account's history (the earliest prod
|
|
* measurements are < 10 years old) so it never changes a real all-time
|
|
* count / min / max / mean; it exists purely so a coverage-miss cold read
|
|
* can't trigger an unbounded full-partition scan. Whitelisted by
|
|
* `canonicalMeasurementsFrom`'s interval regex (`<N> years`).
|
|
*/
|
|
const ALL_TIME_LIVE_CAP_INTERVAL = "15 years" as const;
|
|
|
|
/**
|
|
* v1.4.48 — all-time aggregate row used on the cold-mount fallback
|
|
* path. Holds the linearly composable columns (`count / min / max /
|
|
* mean`) that must reflect every row the user has ever logged for
|
|
* that type, capped at `ALL_TIME_LIVE_CAP_INTERVAL` (v1.18.10 P-7) as a
|
|
* scan-DoS floor.
|
|
*/
|
|
interface AllTimeAggregateRow {
|
|
type: string;
|
|
count: bigint;
|
|
min_value: number | null;
|
|
max_value: number | null;
|
|
mean_value: number | null;
|
|
}
|
|
|
|
/**
|
|
* Narrow aggregate row — only the columns DAY buckets can't compose
|
|
* linearly. The heavy `COUNT / MIN / MAX / AVG` columns are dropped
|
|
* so the rollup-fresh read path's SQL projection is half as wide as
|
|
* the legacy aggregate's.
|
|
*/
|
|
/**
|
|
* v1.4.37.2 — shape returned by the per-type GROUP BY over
|
|
* `measurement_rollups`. Replaces the v1.4.35 row-per-bucket transfer
|
|
* that materialised a six-figure row count on tenants with large
|
|
* measurement partitions; the SQL now does the aggregation
|
|
* server-side and hands back one row per type.
|
|
*/
|
|
interface RollupAggregateRow {
|
|
type: string;
|
|
count: number;
|
|
/** Days that hold a reading: the weight of the mean for an hourly-mean type. */
|
|
days: number;
|
|
min: number;
|
|
max: number;
|
|
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;
|
|
avg30: number | null;
|
|
/**
|
|
* v1.8.5 — 50th percentile over the trailing 90-day window. The
|
|
* windowed median is cheap (the narrow query already index-scans the
|
|
* 90-day partition) and represents the central reading the stat strip
|
|
* surfaces next to min / max / mean. An all-time median would require
|
|
* sorting every raw row per type, which the rollup tier cannot
|
|
* compose and which the perf posture (v1.8.3 anti-freeze) forbids on
|
|
* the read path. Null when the window holds no rows.
|
|
*
|
|
* NOTE: this window (trailing 90 days) is the contract for the stat
|
|
* strip's `DataSummary.median`. The JS `summarize()` in `trends.ts`
|
|
* computes its median over the *full* series it is handed (a
|
|
* caller-defined window), so the two share a field name but not a
|
|
* window — see the `median` doc on `DataSummary`.
|
|
*/
|
|
median: number | null;
|
|
avg30_last_month: number | null;
|
|
// v1.20.0 — slope7/30/90 + r2_7/30/90 have moved off this live query
|
|
// onto the rollup accumulators (composeWindowedRegression); they no longer
|
|
// scan the raw measurements partition on the warm path. `median` stays
|
|
// live (PERCENTILE_CONT is non-composable across buckets).
|
|
}
|
|
|
|
/**
|
|
* v1.20.0 — cold-fallback windowed row. The coverage-miss path keeps the
|
|
* live REGR_* regression (no rollup accumulators to compose from yet), so it
|
|
* carries the slope / r² columns the warm `NarrowAggregateRow` dropped.
|
|
*/
|
|
interface WindowedAggregateRow extends NarrowAggregateRow {
|
|
slope7: number | null;
|
|
r2_7: number | null;
|
|
slope30: number | null;
|
|
r2_30: number | null;
|
|
}
|
|
|
|
/** Same threshold/rounding contract as `trendSlope()` in `trends.ts`. */
|
|
function buildSlope(
|
|
slope: number | null,
|
|
rSquared: number | null,
|
|
): DataSummary["slope7"] {
|
|
if (slope === null) return null;
|
|
const threshold = 0.01;
|
|
const direction: "up" | "down" | "stable" =
|
|
Math.abs(slope) < threshold ? "stable" : slope > 0 ? "up" : "down";
|
|
return {
|
|
slope: Math.round(slope * 1000) / 1000,
|
|
direction,
|
|
confidence: rSquared === null ? 0 : Math.round(rSquared * 100) / 100,
|
|
};
|
|
}
|
|
|
|
function round2(value: number | null): number | null {
|
|
if (value === null || !Number.isFinite(value)) return null;
|
|
return Math.round(value * 100) / 100;
|
|
}
|
|
|
|
function emptySummary(): DataSummary {
|
|
return {
|
|
count: 0,
|
|
latest: null,
|
|
min: null,
|
|
max: null,
|
|
mean: null,
|
|
median: null,
|
|
avg7: null,
|
|
avg30: null,
|
|
slope7: null,
|
|
slope30: null,
|
|
avg30LastMonth: null,
|
|
avg30LastYear: null,
|
|
};
|
|
}
|
|
|
|
export interface SummariesSlice {
|
|
summaries: Record<string, DataSummary>;
|
|
/**
|
|
* `null` on the slim slice. The default route computes BMI from
|
|
* `summaries.WEIGHT.latest` + `user.heightCm`; consumers of the
|
|
* slim slice that need BMI re-derive client-side or fetch the
|
|
* default slice.
|
|
*/
|
|
bmi: null;
|
|
/**
|
|
* v1.4.34 — per-type freshness map matching the default-slice
|
|
* shape so the dashboard tile strip can render a "Letzter Wert vor
|
|
* Xd" caption on stale tiles regardless of which slice the client
|
|
* read from. Types the user has never logged report `null`.
|
|
*/
|
|
lastSeenByType: Record<
|
|
string,
|
|
{ lastSeenAt: string; daysAgo: number } | null
|
|
>;
|
|
/**
|
|
* v1.28.x — source-discrepancy annotation for the LATEST night's main
|
|
* session (see `SleepNight.sourceDiscrepancy`): non-null when two writer
|
|
* buckets reported clearly different asleep totals for the night behind
|
|
* `summaries.SLEEP_DURATION.latest`. Observational only — the served
|
|
* summary stays the winning writer's totals; the dashboard sleep tile
|
|
* marks the number with a discreet "sources disagree" hint. Minutes are
|
|
* whole (rounded), matching the `/api/sleep/night` serialisation.
|
|
* Additive on the wire; every consumer must tolerate `null`.
|
|
*/
|
|
sleepSourceDiscrepancy: SleepSourceDiscrepancy | null;
|
|
}
|
|
|
|
export async function computeSummariesSlice(
|
|
userId: string,
|
|
/**
|
|
* v1.12.7 — optional pre-probed coverage map. The dashboard
|
|
* snapshot already calls `probeRollupCoverage` up front to decide
|
|
* whether to run the thick `extras` phase, then called this slice
|
|
* which re-probed the identical query. Threading the map in lets the
|
|
* slice reuse it and saves one round-trip on the snapshot's hot path.
|
|
* Omit it (every other caller) to keep the internal probe as the
|
|
* fallback — byte-stable for those callers.
|
|
*/
|
|
precomputedCoverage?: RollupCoverageMap,
|
|
): Promise<SummariesSlice> {
|
|
// 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
|
|
// sample sets that keep the 90-day window slightly stale. The downstream
|
|
// coverage probe falls back to live SQL for uncovered types, so
|
|
// correctness is preserved; the read just doesn't block waiting
|
|
// for the background refresh.
|
|
void ensureUserRollupsFresh(userId);
|
|
|
|
// v1.4.36 — per-type coverage probe replaces the legacy global
|
|
// COUNT. The previous gate returned true as soon as ANY type had at
|
|
// least one DAY bucket. Pathology: a user with BP fully rolled up
|
|
// logs their first WEIGHT measurement → the sync write-hook upserts
|
|
// one WEIGHT DAY bucket → the global probe stayed >0 → the route
|
|
// flipped to the bucket-derived path for WEIGHT as well → all-time
|
|
// count collapsed to whatever the trailing window covers because the
|
|
// narrow aggregate's windowed columns + the single fresh bucket
|
|
// can't reconstruct the prior history. We now decide per type and
|
|
// only take the rollup path when EVERY type the user has logged is
|
|
// covered. Partial coverage falls back to the live aggregate so the
|
|
// brand-new-type case stays correct.
|
|
const coverage = precomputedCoverage ?? (await probeRollupCoverage(userId));
|
|
if (isFullyCovered(coverage)) {
|
|
return withLatestDayValues(
|
|
userId,
|
|
await withSleepNightTotals(userId, await computeFromRollups(userId)),
|
|
);
|
|
}
|
|
// v1.4.38.7 — annotate the live fallback with the per-type
|
|
// coverage map so the operator can see WHICH type stranded the
|
|
// request on the live aggregator. Without this the wide-event only
|
|
// says `path:"live"`, and "why" requires shell access to the DB.
|
|
const missing: string[] = [];
|
|
for (const [type, hasBuckets] of coverage.entries()) {
|
|
if (!hasBuckets) missing.push(type);
|
|
}
|
|
annotate({
|
|
meta: {
|
|
analytics: {
|
|
slim_summaries: {
|
|
fallback_reason: "partial_rollup_coverage",
|
|
missing_types: missing,
|
|
covered_count: coverage.size - missing.length,
|
|
total_types: coverage.size,
|
|
},
|
|
},
|
|
},
|
|
});
|
|
return withLatestDayValues(
|
|
userId,
|
|
await withSleepNightTotals(userId, await computeFromLiveAggregate(userId)),
|
|
);
|
|
}
|
|
|
|
/**
|
|
* The one cache cell of the summaries slice. The tile-strip route, the thick
|
|
* analytics body and the dashboard snapshot all read the slice; each used to
|
|
* compute it inside its own cell, so one dashboard mount built it up to three
|
|
* times over the same rows.
|
|
*
|
|
* `serveStale` is for the caller that answers the client with the slice
|
|
* itself. The composing callers are stale-while-revalidate cells of their
|
|
* own and read fresh-only: a background refresh after a write must not
|
|
* compose the pre-write slice and cache it again as fresh.
|
|
*/
|
|
export async function readSummariesSlice(
|
|
userId: string,
|
|
options: { coverage?: RollupCoverageMap; serveStale?: boolean } = {},
|
|
): Promise<SummariesSlice> {
|
|
const cache = caches.analytics as ServerCache<SummariesSlice>;
|
|
const key = summariesSliceCacheKey(userId);
|
|
const build = () => computeSummariesSlice(userId, options.coverage);
|
|
const { value, outcome } = options.serveStale
|
|
? await cache.wrapSwr(key, build)
|
|
: await cache.wrap(key, build);
|
|
annotate({ meta: { "cache.summaries_slice.outcome": outcome } });
|
|
return value;
|
|
}
|
|
|
|
export function summariesSliceCacheKey(userId: string): string {
|
|
return `${userId}|summaries`;
|
|
}
|
|
|
|
/**
|
|
* `latest` of an hourly-mean type (pulse) is the value of its latest local
|
|
* day, not its latest single reading.
|
|
*
|
|
* The dashboard's pulse tile led with the last reading a watch happened to
|
|
* take, while the day view showed the day's pulse as the mean of its hours'
|
|
* means (`day-mean.ts`). Both say "Pulse"; both now mean the same statistic:
|
|
* the latest day's value, read through the same canonical-source pick and
|
|
* the same day-mean helper the day view uses, in the person's own zone.
|
|
*/
|
|
async function withLatestDayValues(
|
|
userId: string,
|
|
slice: SummariesSlice,
|
|
): Promise<SummariesSlice> {
|
|
const types = [...HOURLY_MEAN_DAY_TYPES].filter(
|
|
(type) =>
|
|
slice.summaries[type]?.latest != null && slice.lastSeenByType[type],
|
|
);
|
|
if (types.length === 0) return slice;
|
|
const [tz, priorityJson] = await Promise.all([
|
|
resolveUserTimezone(userId),
|
|
loadUserSourcePriority(userId),
|
|
]);
|
|
await Promise.all(
|
|
types.map(async (type) => {
|
|
const seen = slice.lastSeenByType[type]!;
|
|
const day = userDayKey(new Date(seen.lastSeenAt), tz);
|
|
const { dayStart, dayEnd } = localDayWindow(day, tz);
|
|
const rows = await prisma.measurement.findMany({
|
|
where: {
|
|
userId,
|
|
type,
|
|
deletedAt: null,
|
|
measuredAt: { gte: dayStart, lt: dayEnd },
|
|
},
|
|
select: {
|
|
type: true,
|
|
value: true,
|
|
unit: true,
|
|
measuredAt: true,
|
|
source: true,
|
|
deviceType: true,
|
|
},
|
|
orderBy: [{ measuredAt: "asc" }, { id: "asc" }],
|
|
});
|
|
const canonical = canonicalRowsOfDay(rows, type, day, tz, priorityJson);
|
|
const value = dayValue(type, canonical, tz);
|
|
if (value !== null) slice.summaries[type].latest = value;
|
|
}),
|
|
);
|
|
return slice;
|
|
}
|
|
|
|
/**
|
|
* v1.11.4 — replace the per-stage `summaries.SLEEP_DURATION` with a
|
|
* per-NIGHT summary (iOS #1, HIGH).
|
|
*
|
|
* `SLEEP_DURATION` is stored one row per sleep STAGE per night (minutes).
|
|
* The slim slice's generic aggregate therefore treats each stage as an
|
|
* independent reading: `latest` is the most-recent STAGE (not the night),
|
|
* `mean` / `avg7` / `avg30` average across stage rows. For the dashboard
|
|
* sleep tile and the insights overview that is wrong — they want the
|
|
* per-night TIME-ASLEEP total.
|
|
*
|
|
* This post-pass reads the SLEEP_DURATION rows once and overwrites the
|
|
* summary with one computed over per-night asleep totals (one DataPoint
|
|
* per night). The unit stays MINUTES (canonical), so existing consumers
|
|
* that already treat the summary as minutes keep working — they now
|
|
* average nights instead of stages, which is strictly more correct. It
|
|
* covers the whole history, like every other type here (a year's cut left
|
|
* `avg30LastYear` and the all-time count and range short), and is skipped
|
|
* entirely when the user has logged no sleep.
|
|
*/
|
|
async function withSleepNightTotals(
|
|
userId: string,
|
|
slice: SummariesSlice,
|
|
): Promise<SummariesSlice> {
|
|
const existing = slice.summaries.SLEEP_DURATION;
|
|
// No sleep data at all → nothing to recompute (the empty summary stays).
|
|
if (!existing || existing.count === 0) return slice;
|
|
|
|
const [tz, priorityJson] = await Promise.all([
|
|
resolveUserTimezone(userId),
|
|
loadUserSourcePriority(userId),
|
|
]);
|
|
// The whole history, as every other type in the slice: the walk names the
|
|
// nights in Postgres and reconstructs them ninety at a time, so the heap
|
|
// holds a chunk's rows however long the history is.
|
|
const nights = await readSleepNights(userId, tz, priorityJson, {
|
|
maxNights: ALL_NIGHTS,
|
|
});
|
|
if (nights.length === 0) return slice;
|
|
const { summary, latestNight } = summarizeNights(nights);
|
|
// Preserve the slope tuples the night summary doesn't compute (the
|
|
// dashboard tile reads slope30); recompute them off the night series is
|
|
// out of scope here — the per-night `summarize()` already fills slope7 /
|
|
// slope30 from the night DataPoints, so use them directly.
|
|
slice.summaries.SLEEP_DURATION = summary;
|
|
// Additive, observational: the latest night's main session saw two
|
|
// writer buckets with clearly different asleep totals. Round to whole
|
|
// minutes, matching the `/api/sleep/night` serialisation.
|
|
slice.sleepSourceDiscrepancy = latestNight?.sourceDiscrepancy
|
|
? {
|
|
deltaMinutes: Math.round(latestNight.sourceDiscrepancy.deltaMinutes),
|
|
sources: latestNight.sourceDiscrepancy.sources.map((b) => ({
|
|
source: b.source,
|
|
deviceType: b.deviceType,
|
|
asleepMinutes: Math.round(b.asleepMinutes),
|
|
})),
|
|
}
|
|
: null;
|
|
return slice;
|
|
}
|
|
|
|
/**
|
|
* The seven-day mean per type, exactly as the slice computes `avg7` (the same
|
|
* canonical-source CTE, the same day weighting, the same rounding), without
|
|
* the rest of the slice. Sleep is absent: the slice summarises it per night. For a reader that needs only the dashboard's basis,
|
|
* such as the target verdicts: one aggregate query, no reading ever leaves
|
|
* the database as a row.
|
|
*/
|
|
export async function readSevenDayMeans(
|
|
userId: string,
|
|
): Promise<Record<string, number | null>> {
|
|
const priorityJson = await loadUserSourcePriority(userId);
|
|
const rankUnqualified = buildSourceRankCase(
|
|
priorityJson,
|
|
'"type"',
|
|
'"source"',
|
|
);
|
|
const rows = await prisma.$queryRawUnsafe<
|
|
Array<{ type: string; avg7: number | null }>
|
|
>(
|
|
`
|
|
WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, "90 days")}
|
|
)
|
|
SELECT
|
|
m."type"::text AS type,
|
|
${AVG7_SQL}::double precision AS avg7
|
|
FROM ${dayWeightedRows("cm", SESSION_DAY_FRAME)} m
|
|
GROUP BY m."type"
|
|
`,
|
|
userId,
|
|
);
|
|
const out: Record<string, number | null> = {};
|
|
for (const row of rows) {
|
|
// Stored per stage, summarised per night by `withSleepNightTotals`: a
|
|
// mean over stage rows is not the slice's sleep figure, so it is left
|
|
// out rather than served wrong. Callers fold the nights themselves.
|
|
if (row.type === "SLEEP_DURATION") continue;
|
|
out[row.type] = round2(row.avg7);
|
|
}
|
|
return out;
|
|
}
|
|
|
|
/**
|
|
* Happy path — DAY buckets carry `count / min / max / mean`; a narrow
|
|
* `$queryRaw` carries only the windowed avgs + regression columns the
|
|
* buckets cannot reconstruct.
|
|
*
|
|
* v1.4.47.1 — the `narrows` query takes an outer 90-day `measured_at`
|
|
* cap so the planner does an index range scan on
|
|
* `(user_id, type, measured_at)` instead of reading every row in the
|
|
* user's measurements partition. Every FILTER expression inside the
|
|
* SELECT already restricts to 7/30/90 days, so the cap excludes only
|
|
* rows that were already being aggregated to NULL. On tenants with
|
|
* 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<SummariesSlice> {
|
|
// 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
|
|
// the unqualified rollup + inner-subquery columns.
|
|
const priorityJson = await loadUserSourcePriority(userId);
|
|
const rankUnqualified = buildSourceRankCase(
|
|
priorityJson,
|
|
'"type"',
|
|
'"source"',
|
|
);
|
|
const rankPerSource = buildSourceRankCase(
|
|
priorityJson,
|
|
'p."type"',
|
|
'p."source"',
|
|
);
|
|
|
|
// v1.37.19 — where the fold window starts. DAY rollup buckets
|
|
// only exist inside the trailing `ROLLUP_FOLD_WINDOW_MS` (5 y), so the
|
|
// rollup-composed "all-time" figures silently truncated to that window
|
|
// for accounts with older history, while the cold live path reported
|
|
// the true unbounded numbers. The pre-fold remainder query below
|
|
// splices the older rows back in.
|
|
//
|
|
// The two halves meet at ONE instant, and it is a DAY bucket boundary:
|
|
// buckets from `foldSplit` on, raw readings before it. Buckets older than
|
|
// the window do exist — the per-write hook folds any day it touches, the
|
|
// import folded whole spans up to v1.39.0, every fold snaps its start down
|
|
// to the whole day, and nothing prunes a bucket the moving window has
|
|
// 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 [narrows, latests, dayBuckets, accumulatorRows, preFoldRemainder] =
|
|
await Promise.all([
|
|
// The FROM is restricted to the canonical-source rows per (type, day): the
|
|
// inner DISTINCT ON picks the ladder-winning source for each day, and the
|
|
// join keeps only that source's readings so the 90-day AVG / median / slope
|
|
// never blend two devices that both reported the same vital.
|
|
// The window means weigh each day once for an hourly-mean type
|
|
// (`windowedMean`); the median stays over every reading.
|
|
prisma.$queryRawUnsafe<NarrowAggregateRow[]>(
|
|
`
|
|
WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, "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,
|
|
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,
|
|
),
|
|
readLatestCanonical(userId, priorityJson),
|
|
// v1.4.37.2 — the v1.4.35 implementation read EVERY DAY
|
|
// rollup bucket for the user (`findMany` without a `bucketStart`
|
|
// window) and then composed `count / min / max / mean` in JS.
|
|
// On tenants with large measurement partitions that materialised
|
|
// as a six-figure row transfer + a JS loop = multi-second per
|
|
// cache miss, even with the rollup table hot. The slim slice's
|
|
// contract is the all-time count / min / max / mean per type —
|
|
// exactly what a SQL `GROUP BY type` returns in a single
|
|
// round-trip.
|
|
//
|
|
// v1.11.1 — rows are now per source. Collapse each (type, day) to the
|
|
// ladder-canonical source via DISTINCT ON before the all-time aggregate,
|
|
// so a dual-source vital is counted once. Still one server-side pass —
|
|
// the DISTINCT ON + GROUP BY runs in Postgres and returns one row/type.
|
|
prisma.$queryRawUnsafe<RollupAggregateRow[]>(
|
|
`
|
|
WITH collapsed AS (
|
|
SELECT DISTINCT ON ("type", "bucket_start")
|
|
"type" AS type,
|
|
"count" AS count,
|
|
"min_value" AS min_value,
|
|
"max_value" AS max_value,
|
|
"mean" AS mean
|
|
FROM "measurement_rollups"
|
|
WHERE "user_id" = $1
|
|
AND "granularity" = 'DAY'
|
|
AND "bucket_start" >= $2
|
|
ORDER BY "type", "bucket_start", (${rankUnqualified}), "source"::text
|
|
)
|
|
SELECT
|
|
"type"::text AS type,
|
|
SUM("count")::int AS count,
|
|
COUNT(*)::int AS days,
|
|
MIN("min_value")::double precision AS min,
|
|
MAX("max_value")::double precision AS max,
|
|
-- An hourly-mean type (pulse): every day once, the DAY mean already
|
|
-- being the mean of its hours' means. Every other type: every
|
|
-- reading once.
|
|
(CASE WHEN ${isHourlyMeanTypeSql('"type"')}
|
|
THEN AVG("mean")::double precision
|
|
ELSE (
|
|
SUM("count" * "mean")::double precision
|
|
/ NULLIF(SUM("count")::double precision, 0)
|
|
)
|
|
END) AS mean
|
|
FROM collapsed
|
|
GROUP BY "type"
|
|
`,
|
|
userId,
|
|
foldSplit,
|
|
),
|
|
// v1.20.0 — trailing 90-day DAY rollup rows carrying the per-bucket
|
|
// regression accumulators. The windowed slope / r² / sd compose from
|
|
// these (composeWindowedRegression) instead of the live REGR_* scan the
|
|
// narrow query used to run. Rows are per source; collapsed per type
|
|
// below so a dual-source vital contributes only its canonical source's
|
|
// accumulators. Bounded to 90 days (≤ ~90 rows/type), the same window
|
|
// the dropped slope columns covered.
|
|
prisma.measurementRollup.findMany({
|
|
where: {
|
|
userId,
|
|
granularity: "DAY",
|
|
bucketStart: {
|
|
gte: startOfUtcDay(new Date(Date.now() - 90 * DAY_MS)),
|
|
},
|
|
},
|
|
orderBy: [{ type: "asc" }, { bucketStart: "asc" }],
|
|
select: {
|
|
type: true,
|
|
source: true,
|
|
bucketStart: true,
|
|
count: true,
|
|
mean: true,
|
|
sumX: true,
|
|
sumXy: true,
|
|
sumXx: true,
|
|
sumYy: true,
|
|
},
|
|
}),
|
|
// v1.37.19 — all-time remainder OLDER than the fold window,
|
|
// spliced into the per-type figures below so "all-time" means all time
|
|
// on the rollup path too (empty for any account younger than the
|
|
// window). It keeps one source per (type, day) the way the rollup half
|
|
// and the live path do: without it, a vital recorded by two devices
|
|
// counted both of them on every pre-fold day. Each source is folded per
|
|
// day first (a hash aggregate, a few rows per day however dense the
|
|
// stream), each day then keeps its ladder-canonical source, so the
|
|
// sort never sees more than one row per source per day.
|
|
// An hourly-mean type folds each source's day as the mean of its
|
|
// hours' means (`wsum` / `wdays`, see `day-mean.ts`) and the days into
|
|
// the remainder once each, the statistic the DAY rollup stores.
|
|
prisma.$queryRaw<
|
|
Array<{
|
|
type: string;
|
|
count: number;
|
|
days: number;
|
|
min: number;
|
|
max: number;
|
|
mean: number;
|
|
}>
|
|
>`
|
|
WITH src AS (
|
|
SELECT *
|
|
FROM measurements m
|
|
WHERE m."user_id" = ${userId}
|
|
AND m."deleted_at" IS NULL
|
|
AND m."measured_at" < ${foldSplit}
|
|
),
|
|
per_source AS (
|
|
SELECT
|
|
m."type",
|
|
m."source",
|
|
date_trunc('day', m."measured_at") AS day,
|
|
COUNT(*)::int AS cnt,
|
|
SUM(m."value")::double precision AS total,
|
|
MIN(m."value")::double precision AS min_value,
|
|
MAX(m."value")::double precision AS max_value,
|
|
SUM(m."value" * m.day_weight)::double precision AS wsum,
|
|
SUM(m.day_weight)::double precision AS wdays
|
|
FROM ${dayWeightedRowsSql("src", null, { bySource: true })} m
|
|
GROUP BY m."type", m."source", 3
|
|
),
|
|
canon AS (
|
|
SELECT DISTINCT ON (p."type", p.day)
|
|
p."type", p.cnt, p.total, p.min_value, p.max_value, p.wsum, p.wdays
|
|
FROM per_source p
|
|
ORDER BY p."type", p.day, (${Prisma.raw(rankPerSource)}), p."source"::text
|
|
)
|
|
SELECT
|
|
c."type"::text AS type,
|
|
SUM(c.cnt)::int AS count,
|
|
COUNT(*)::int AS days,
|
|
MIN(c.min_value)::double precision AS min,
|
|
MAX(c.max_value)::double precision AS max,
|
|
${Prisma.raw(
|
|
foldMeanSql({
|
|
typeColumn: 'c."type"',
|
|
total: "c.total",
|
|
count: "c.cnt",
|
|
weightedSum: "c.wsum",
|
|
weightSum: "c.wdays",
|
|
}),
|
|
)}::double precision AS mean
|
|
FROM canon c
|
|
GROUP BY c."type"
|
|
`,
|
|
]);
|
|
|
|
const preFoldByType = new Map<
|
|
string,
|
|
{ count: number; days: number; min: number; max: number; mean: number }
|
|
>();
|
|
for (const row of preFoldRemainder) {
|
|
preFoldByType.set(row.type, {
|
|
count: Number(row.count),
|
|
days: Number(row.days),
|
|
min: Number(row.min),
|
|
max: Number(row.max),
|
|
mean: Number(row.mean),
|
|
});
|
|
}
|
|
|
|
// v1.20.0 — collapse the 90-day accumulator rows to the canonical
|
|
// source per (type, day), then build a per-type accumulator-bucket list so
|
|
// the windowed slope / r² / sd compose from the canonical source only.
|
|
const accBucketsByType = new Map<string, AccumulatorBucketRow[]>();
|
|
{
|
|
const byType = new Map<string, (typeof accumulatorRows)[number][]>();
|
|
for (const r of accumulatorRows) {
|
|
const list = byType.get(r.type) ?? [];
|
|
list.push(r);
|
|
byType.set(r.type, list);
|
|
}
|
|
for (const [type, rows] of byType) {
|
|
const collapsed = collapseRollupRowsBySource(
|
|
rows,
|
|
type as MeasurementType,
|
|
priorityJson,
|
|
);
|
|
accBucketsByType.set(
|
|
type,
|
|
collapsed.map((b) => ({
|
|
bucketStart: b.bucketStart,
|
|
count: b.count,
|
|
mean: b.mean,
|
|
sumX: b.sumX,
|
|
sumXy: b.sumXy,
|
|
sumXx: b.sumXx,
|
|
sumYy: b.sumYy,
|
|
})),
|
|
);
|
|
}
|
|
}
|
|
|
|
// 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();
|
|
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));
|
|
|
|
const latestByType = new Map<string, number>();
|
|
// v1.4.34 — capture the per-type `measured_at` alongside the
|
|
// value so we can surface the freshness map on the slim slice
|
|
// without an extra round-trip.
|
|
const lastSeenAtByType = new Map<string, Date>();
|
|
for (const row of latests) {
|
|
latestByType.set(row.type, Number(row.value));
|
|
if (row.measured_at) {
|
|
lastSeenAtByType.set(row.type, new Date(row.measured_at));
|
|
}
|
|
}
|
|
|
|
// Seed every enum option so consumers can read `summaries.PULSE` and
|
|
// get a deterministic empty shape rather than `undefined`. Mirrors
|
|
// the default route which iterates `measurementTypeEnum.options`.
|
|
const summaries: Record<string, DataSummary> = {};
|
|
const lastSeenByType: Record<
|
|
string,
|
|
{ lastSeenAt: string; daysAgo: number } | null
|
|
> = {};
|
|
for (const type of measurementTypeEnum.options) {
|
|
summaries[type] = emptySummary();
|
|
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),
|
|
);
|
|
lastSeenByType[type] = {
|
|
lastSeenAt: measuredAt.toISOString(),
|
|
daysAgo,
|
|
};
|
|
}
|
|
|
|
// v1.4.37.2 — the rollup aggregate is now shaped server-side
|
|
// (one row per type with count / min / max / mean already computed),
|
|
// so we map straight through instead of partitioning + composing in
|
|
// JS. The narrows query feeds the windowed avgs + slope columns the
|
|
// GROUP BY cannot reconstruct.
|
|
const narrowByType = new Map<string, NarrowAggregateRow>();
|
|
for (const row of narrows) {
|
|
narrowByType.set(row.type, row);
|
|
}
|
|
|
|
let totalRows = 0;
|
|
let typeCount = 0;
|
|
const typesWithData: string[] = [];
|
|
// 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 — honest, it converges after backfill — but the miss must be
|
|
// observable per the "no silent cap" rule, so it surfaces in the
|
|
// annotation below rather than passing silently as a plain rollup read.
|
|
let regressionPendingBackfill = false;
|
|
for (const row of dayBuckets) {
|
|
if (!row.count || row.count <= 0) continue;
|
|
typeCount += 1;
|
|
totalRows += row.count;
|
|
typesWithData.push(row.type);
|
|
const narrow = narrowByType.get(row.type);
|
|
const latest = latestByType.get(row.type) ?? null;
|
|
// v1.20.0 — windowed slope / r² compose from the per-bucket
|
|
// accumulators (canonical-source-collapsed above) rather than the live
|
|
// REGR_* scan. A full miss (NULL accumulators pre-backfill, or < 2
|
|
// readings) yields a null slope, the same shape the empty REGR_* FILTER
|
|
// returned.
|
|
const accBuckets = accBucketsByType.get(row.type) ?? [];
|
|
if (
|
|
!regressionPendingBackfill &&
|
|
hasPendingAccumulatorBackfill(accBuckets, since90)
|
|
) {
|
|
regressionPendingBackfill = true;
|
|
}
|
|
const reg7 = composeWindowedRegression(accBuckets, since7);
|
|
const reg30 = composeWindowedRegression(accBuckets, since30);
|
|
// v1.37.19 — splice the pre-fold remainder into the all-time
|
|
// figures: the rollup tier only covers the trailing fold window, so
|
|
// count / min / max / mean composed from it alone truncated "all
|
|
// time" to 5 y. Windowed fields (avg7/30, median, slopes) are
|
|
// unaffected — their windows sit far inside the fold window.
|
|
const pre = preFoldByType.get(row.type);
|
|
const allCount = row.count + (pre?.count ?? 0);
|
|
const allMin = pre ? Math.min(row.min, pre.min) : row.min;
|
|
const allMax = pre ? Math.max(row.max, pre.max) : row.max;
|
|
// Both halves weigh their days once for an hourly-mean type (pulse), so
|
|
// they combine by day count; every other type combines by readings.
|
|
const allMean = pre
|
|
? usesHourlyMeanDay(row.type)
|
|
? (row.mean * Number(row.days) + pre.mean * pre.days) /
|
|
(Number(row.days) + pre.days)
|
|
: (row.mean * row.count + pre.mean * pre.count) / allCount
|
|
: row.mean;
|
|
summaries[row.type] = {
|
|
count: allCount,
|
|
latest,
|
|
min: round2(allMin),
|
|
max: round2(allMax),
|
|
mean: round2(allMean),
|
|
median: round2(narrow?.median ?? null),
|
|
avg7: round2(narrow?.avg7 ?? null),
|
|
avg30: round2(narrow?.avg30 ?? null),
|
|
slope7: buildSlope(reg7.slope, reg7.r2),
|
|
slope30: buildSlope(reg30.slope, reg30.r2),
|
|
avg30LastMonth: round2(narrow?.avg30_last_month ?? null),
|
|
avg30LastYear: null,
|
|
};
|
|
if (pre) totalRows += pre.count;
|
|
}
|
|
|
|
// v1.37.19 — a type whose ONLY rows are older than the fold
|
|
// window has no DAY bucket at all; without this arm it read as "never
|
|
// logged" on the rollup path while the live path reported its history.
|
|
for (const [type, pre] of preFoldByType) {
|
|
if (typesWithData.includes(type)) continue;
|
|
typeCount += 1;
|
|
totalRows += pre.count;
|
|
typesWithData.push(type);
|
|
summaries[type] = {
|
|
count: pre.count,
|
|
latest: latestByType.get(type) ?? null,
|
|
min: round2(pre.min),
|
|
max: round2(pre.max),
|
|
mean: round2(pre.mean),
|
|
median: null,
|
|
avg7: null,
|
|
avg30: null,
|
|
slope7: null,
|
|
slope30: null,
|
|
avg30LastMonth: null,
|
|
avg30LastYear: null,
|
|
};
|
|
}
|
|
|
|
// Populate `avg30LastYear` per type from the DAY rollup tier. Only types
|
|
// 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);
|
|
let yearOverYearTypeCount = 0;
|
|
for (const [type, value] of avg30LastYearMap.entries()) {
|
|
if (value === null) continue;
|
|
const summary = summaries[type];
|
|
if (!summary) continue;
|
|
summary.avg30LastYear = round2(value);
|
|
yearOverYearTypeCount += 1;
|
|
}
|
|
|
|
annotate({
|
|
action: { name: "analytics.get.slim" },
|
|
meta: {
|
|
analytics: {
|
|
slim_summaries: {
|
|
row_count: totalRows,
|
|
type_count: typeCount,
|
|
path: "rollup",
|
|
year_over_year_types: yearOverYearTypeCount,
|
|
// v1.20.0 — slope / r² now compose from the rollup
|
|
// accumulators (no live REGR_* scan). The median stays live
|
|
// (PERCENTILE_CONT is non-composable); both sources are logged
|
|
// per the "no silent cap" rule so a regression is observable.
|
|
slope_source: "rollup",
|
|
median_source: "live",
|
|
// v1.20.0 — when a NULL accumulator (pre-0190 row, or boot
|
|
// re-fold pending) forced a slope miss, the read intentionally
|
|
// returns null without a live-SQL fallback (the design avoids the
|
|
// heavy scan). Surface the gap so the silent null is observable;
|
|
// it clears once the boot backfill refills the accumulators.
|
|
regression_source: regressionPendingBackfill
|
|
? "unavailable_pending_backfill"
|
|
: "rollup",
|
|
},
|
|
},
|
|
},
|
|
});
|
|
|
|
return { summaries, bmi: null, lastSeenByType, sleepSourceDiscrepancy: null };
|
|
}
|
|
|
|
/**
|
|
* Cold fallback — runs the legacy heavy aggregate when the rollup
|
|
* table is empty for this user. Subsequent reads pick up the populated
|
|
* rollup on `computeFromRollups`.
|
|
*
|
|
* v1.4.48 — split into two parallel queries:
|
|
*
|
|
* 1. `allTime` — the linearly composable columns (`count / min /
|
|
* max / mean`). v1.18.10 P-7 caps this at `ALL_TIME_LIVE_CAP_INTERVAL`
|
|
* (15 years) so an un-windowed cold-fallback read on a coverage-miss
|
|
* tenant can never scan an unbounded partition. The cap is a scan-DoS
|
|
* floor, not a semantic change: it sits far beyond any real account's
|
|
* history (the earliest prod rows are < 10 years old) and the steady
|
|
* state is rollup-backed anyway, so it only ever bounds the cold path.
|
|
*
|
|
*
|
|
* 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'`
|
|
* 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
|
|
* clauses. On tenants with large measurement partitions the
|
|
* cold fallback drops from multi-second to sub-second. Output is
|
|
* bit-identical because every row excluded by the new outer cap
|
|
* was already aggregating to NULL inside its FILTER clause.
|
|
*/
|
|
async function computeFromLiveAggregate(
|
|
userId: string,
|
|
): Promise<SummariesSlice> {
|
|
// 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
|
|
// parity). `canonicalMeasurementsFrom` swaps `FROM measurements m` for a
|
|
// canonical-source-filtered subquery; the latest read uses the ladder rank.
|
|
const priorityJson = await loadUserSourcePriority(userId);
|
|
const rankUnqualified = buildSourceRankCase(
|
|
priorityJson,
|
|
'"type"',
|
|
'"source"',
|
|
);
|
|
|
|
const [allTime, windowed, latests] = await Promise.all([
|
|
prisma.$queryRawUnsafe<AllTimeAggregateRow[]>(
|
|
`
|
|
WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, ALL_TIME_LIVE_CAP_INTERVAL)}
|
|
)
|
|
SELECT
|
|
m."type"::text AS type,
|
|
COUNT(*) AS count,
|
|
MIN(m."value")::double precision AS min_value,
|
|
MAX(m."value")::double precision AS max_value,
|
|
${windowMeanSql({ typeColumn: 'm."type"', value: 'm."value"', weight: "m.day_weight" })}::double precision
|
|
AS mean_value
|
|
FROM ${dayWeightedRows("cm", SESSION_DAY_FRAME)} m
|
|
GROUP BY m."type"
|
|
`,
|
|
userId,
|
|
),
|
|
prisma.$queryRawUnsafe<WindowedAggregateRow[]>(
|
|
`
|
|
WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, "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,
|
|
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
|
|
FROM ${dayWeightedRows("cm", SESSION_DAY_FRAME)} m
|
|
GROUP BY m."type"
|
|
`,
|
|
userId,
|
|
),
|
|
readLatestCanonical(userId, priorityJson),
|
|
]);
|
|
|
|
const latestByType = new Map<string, number>();
|
|
const lastSeenAtByType = new Map<string, Date>();
|
|
for (const row of latests) {
|
|
latestByType.set(row.type, Number(row.value));
|
|
if (row.measured_at) {
|
|
lastSeenAtByType.set(row.type, new Date(row.measured_at));
|
|
}
|
|
}
|
|
|
|
const summaries: Record<string, DataSummary> = {};
|
|
const lastSeenByType: Record<
|
|
string,
|
|
{ lastSeenAt: string; daysAgo: number } | null
|
|
> = {};
|
|
for (const type of measurementTypeEnum.options) {
|
|
summaries[type] = emptySummary();
|
|
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),
|
|
);
|
|
lastSeenByType[type] = {
|
|
lastSeenAt: measuredAt.toISOString(),
|
|
daysAgo,
|
|
};
|
|
}
|
|
|
|
// v1.4.48 — windowed columns are keyed by type alongside the
|
|
// all-time aggregate. Types that exist in `allTime` but have no
|
|
// measurements in the 90-day window leave their windowed columns at
|
|
// `null` (the same shape the pre-split query produced via FILTER
|
|
// returning NULL on an empty set).
|
|
const windowedByType = new Map<string, WindowedAggregateRow>();
|
|
for (const row of windowed) {
|
|
windowedByType.set(row.type, row);
|
|
}
|
|
|
|
let totalRows = 0;
|
|
const typesWithData: string[] = [];
|
|
for (const row of allTime) {
|
|
const count = Number(row.count);
|
|
totalRows += count;
|
|
typesWithData.push(row.type);
|
|
const latest = latestByType.get(row.type) ?? null;
|
|
const win = windowedByType.get(row.type);
|
|
summaries[row.type] = {
|
|
count,
|
|
latest,
|
|
min: round2(row.min_value),
|
|
max: round2(row.max_value),
|
|
mean: round2(row.mean_value),
|
|
median: round2(win?.median ?? null),
|
|
avg7: round2(win?.avg7 ?? null),
|
|
avg30: round2(win?.avg30 ?? null),
|
|
slope7: buildSlope(win?.slope7 ?? null, win?.r2_7 ?? null),
|
|
slope30: buildSlope(win?.slope30 ?? null, win?.r2_30 ?? null),
|
|
avg30LastMonth: round2(win?.avg30_last_month ?? null),
|
|
avg30LastYear: null,
|
|
};
|
|
}
|
|
|
|
// Even on the live-aggregate fallback (trailing DAY coverage incomplete),
|
|
// 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);
|
|
let yearOverYearTypeCount = 0;
|
|
for (const [type, value] of avg30LastYearMap.entries()) {
|
|
if (value === null) continue;
|
|
const summary = summaries[type];
|
|
if (!summary) continue;
|
|
summary.avg30LastYear = round2(value);
|
|
yearOverYearTypeCount += 1;
|
|
}
|
|
|
|
annotate({
|
|
action: { name: "analytics.get.slim" },
|
|
meta: {
|
|
analytics: {
|
|
slim_summaries: {
|
|
row_count: totalRows,
|
|
type_count: allTime.length,
|
|
path: "live",
|
|
year_over_year_types: yearOverYearTypeCount,
|
|
},
|
|
},
|
|
},
|
|
});
|
|
|
|
return { summaries, bmi: null, lastSeenByType, sleepSourceDiscrepancy: null };
|
|
}
|
|
|
|
/**
|
|
* Year-ago 30-day baseline per type, served from the DAY rollup tier.
|
|
*
|
|
* `DataSummary.avg30LastYear` is the mean over the 30 days that start 395
|
|
* days ago, which the dashboard tile uses for "current average vs the same
|
|
* time last year". The window is exactly those 30 UTC days (the DAY-tier
|
|
* grain), read as canonical-source DAY buckets and composed count-weighted,
|
|
* so it is the same per-reading mean `avg30` reports for the last 30 days.
|
|
*
|
|
* Up to v1.39.2 this read the MONTH tier and kept the months whose first day
|
|
* fell inside the window, so the baseline was a whole calendar month that
|
|
* could start up to a month off, or nothing at all when no month began inside
|
|
* the window.
|
|
*
|
|
* Returns `null` when no reading falls in the window; the UI treats that as
|
|
* "no comparison available".
|
|
*/
|
|
async function computeAvg30LastYearForType(
|
|
userId: string,
|
|
type: MeasurementType,
|
|
priorityJson: unknown,
|
|
now: number,
|
|
): Promise<number | null> {
|
|
const rows = await readRollupBuckets(
|
|
userId,
|
|
type,
|
|
"DAY",
|
|
startOfUtcDay(new Date(now - 395 * DAY_MS)),
|
|
startOfUtcDay(new Date(now - 365 * DAY_MS)),
|
|
priorityJson,
|
|
);
|
|
const composed = aggregateBuckets(
|
|
rows.map((r) => ({
|
|
day: r.bucketStart,
|
|
count: r.count,
|
|
mean: r.mean,
|
|
minValue: r.minValue,
|
|
maxValue: r.maxValue,
|
|
})),
|
|
windowWeighting(type),
|
|
);
|
|
return composed.count > 0 ? composed.mean : null;
|
|
}
|
|
|
|
/**
|
|
* v1.4.40 — fan out `computeAvg30LastYearForType` across
|
|
* the types the caller actually surfaces. Runs the per-type rollup reads
|
|
* in parallel so the long-tail of types doesn't serialise into a
|
|
* per-type round-trip stack.
|
|
*
|
|
* v1.4.43 — cap concurrent per-type WMY reads at 4.
|
|
*
|
|
* Pre-fix this helper ran an unbounded Promise.all over every type the
|
|
* user has data for. On a 15-type tenant the burst held 15+ Prisma
|
|
* slots simultaneously, and with the slim and thick analytics slices
|
|
* firing in parallel on dashboard mount the combined fan-out drowned
|
|
* the pg.Pool max=20 even after the v1.4.40 W-POOL raise. The cap
|
|
* mirrors the W-POOL `p-limit(4)` discipline v1.4.40 applied to the
|
|
* thick route's per-type live walk (`ANALYTICS_TYPE_FETCH_CONCURRENCY`
|
|
* in `src/app/api/analytics/route.ts`) so both slices behave the same
|
|
* way under burst.
|
|
*
|
|
* Returns a `type → mean | null` map; types with no coverage at any
|
|
* granularity (or no buckets in the year-ago slice) map to `null` so
|
|
* the caller can blanket-assign without per-type null-checking.
|
|
*/
|
|
export const WMY_FANOUT_CONCURRENCY = 4;
|
|
|
|
export async function computeAvg30LastYearMap(
|
|
userId: string,
|
|
types: ReadonlyArray<string>,
|
|
): Promise<Map<string, number | null>> {
|
|
const out = new Map<string, number | null>();
|
|
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 () => {
|
|
const value = await computeAvg30LastYearForType(
|
|
userId,
|
|
type as MeasurementType,
|
|
priorityJson,
|
|
now,
|
|
);
|
|
return [type, value] as const;
|
|
}),
|
|
),
|
|
);
|
|
for (const [type, value] of results) {
|
|
out.set(type, value);
|
|
}
|
|
return out;
|
|
}
|