diff --git a/prisma/schema.prisma b/prisma/schema.prisma index b58af23c3..eb6a974d2 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -5343,11 +5343,14 @@ model AppSettings { // every hour alike; `weighted_hour_mean_v2` weighs an hour by // `min(1, n / 3)`. Switched only by the admin apply step after a dry run; // the dry-run result is the only thing a dry run writes. - /// @pending-consumer(#v143-B8) Active pulse day-mean method (CHECK in migration 0407). + /// Active pulse day-mean method (CHECK in migration 0407). Readers: every + /// process's mirror (`pulse-day-mean-method.ts`) and the rollup tier + /// (`storedHourWeightSql`); the admin status route. pulseDayMeanMethod String @default("hour_mean_v1") @map("pulse_day_mean_method") - /// @pending-consumer(#v143-B8) When the admin applied the current method. + /// When the admin applied the current method; shown on the admin card. pulseDayMeanAppliedAt DateTime? @map("pulse_day_mean_applied_at") - /// @pending-consumer(#v143-B8) Last dry-run status and result (account labels only, no ids). + /// Last dry-run status and result (account ordinals only, no ids); the apply + /// step's precondition and the admin card's result. pulseDayMeanDryRunJson Json? @map("pulse_day_mean_dry_run_json") @@map("app_settings") diff --git a/src/__tests__/contract-stub-inventory.test.ts b/src/__tests__/contract-stub-inventory.test.ts index 3c6b7c5b1..293b57bf1 100644 --- a/src/__tests__/contract-stub-inventory.test.ts +++ b/src/__tests__/contract-stub-inventory.test.ts @@ -24,10 +24,6 @@ const API_ROOT = join(process.cwd(), "src", "app", "api"); /** Route file (under `src/app/api`) to the operations it still stubs. */ const CONTRACT_STUBS: Readonly> = { - "admin/maintenance/pulse-day-mean/route.ts": [ - "admin.pulse_day_mean.get", - "admin.pulse_day_mean.post", - ], "anamnesis/audit-c/[id]/route.ts": ["anamnesis.audit_c.delete"], "anamnesis/audit-c/route.ts": ["anamnesis.audit_c.create"], "anamnesis/life-circumstances/ai-sections/route.ts": [ @@ -53,7 +49,7 @@ const CONTRACT_STUBS: Readonly> = { }; /** The size of the register when it was written; see the header. */ -const REGISTER_SIZE = 15; +const REGISTER_SIZE = 14; function routeFiles(): string[] { return walkSourceFiles(API_ROOT, { floor: 400, extensions: [".ts"] }) diff --git a/src/__tests__/pulse-day-mean-reader-guard.test.ts b/src/__tests__/pulse-day-mean-reader-guard.test.ts index df329bd84..a94277515 100644 --- a/src/__tests__/pulse-day-mean-reader-guard.test.ts +++ b/src/__tests__/pulse-day-mean-reader-guard.test.ts @@ -26,6 +26,11 @@ * imports the shared statistic, hands the type to a fold that does, or * is listed with the reason it never averages pulse readings. * + * A third matcher has no list: since the pulse day-mean method (v1.43) an + * hour weighs by the method, so no day may still be folded from its hours' + * means with every hour weighing one (`AVG(hour_mean)`, a sum over hours + * divided by their number). The weight lives in the helper alone. + * * Comments are stripped before matching, the matchers are whitespace-tolerant, * and the floors fail the guard if either matcher stops finding anything. The * planted cases run each matcher over an offender; dropping the type from the @@ -67,6 +72,15 @@ const TS_MEAN = const NAMES_PULSE = /\bPULSE\b/; +/** + * A day folded from its hours' means with every hour weighing one: SQL + * `AVG(hour_mean)` and an in-memory sum over hours divided by their number. + * Since v1.43 an hour weighs by the pulse day-mean method, which only the + * helper applies (`hourWeightSql`, `combineHourCells`), so none may remain. + */ +const UNWEIGHTED_HOURS = + /\bAVG\s*\(\s*(?:[a-z_]+\s*\.\s*)?"?hour_mean"?\s*\)|\/\s*hours\.(?:length|size)\b/g; + type Entry = /** Averages pulse and imports the shared statistic. */ | { helper: true } @@ -85,6 +99,8 @@ const SQL_READERS: Record = { "lib/doctor-report/dense-buckets.ts": HELPER, "lib/insights/comprehensive-aggregator.ts": HELPER, "lib/insights/comprehensive-generate.ts": HELPER, + // The dry run's hour means, weighed by both methods through the helper. + "lib/jobs/pulse-day-mean.ts": HELPER, "lib/measurements/daily-series-read.ts": HELPER, "lib/rollups/measurement-rollups.ts": HELPER, "lib/rollups/tiered-context.ts": HELPER, @@ -163,7 +179,11 @@ const TS_READERS: Record = { }, }; -function scan(): { sql: Map; ts: Map } { +function scan(): { + sql: Map; + ts: Map; + unweighted: string[]; +} { const files = walkSourceFiles(SRC, { floor: 1500 }).filter( (f) => !f.startsWith("generated/") && @@ -173,8 +193,11 @@ function scan(): { sql: Map; ts: Map } { ); const sql = new Map(); const ts = new Map(); + const unweighted: string[] = []; for (const f of files) { const text = stripComments(readFileSync(join(SRC, f), "utf8")); + if (UNWEIGHTED_HOURS.test(text)) unweighted.push(f); + UNWEIGHTED_HOURS.lastIndex = 0; const s = text.match(SQL_MEAN)?.length ?? 0; if (s > 0) sql.set(f, s); if (NAMES_PULSE.test(text)) { @@ -182,10 +205,10 @@ function scan(): { sql: Map; ts: Map } { if (t > 0) ts.set(f, t); } } - return { sql, ts }; + return { sql, ts, unweighted }; } -const { sql, ts } = scan(); +const { sql, ts, unweighted } = scan(); const source = (f: string) => readFileSync(join(SRC, f), "utf8"); function check(found: Map, listed: Record) { @@ -239,6 +262,22 @@ describe("pulse day-mean reader guard", () => { }); }); + it("leaves no day folded from equally weighted hours", () => { + expect(unweighted).toEqual([]); + }); + + it("catches a planted unweighted hour fold", () => { + expect(`AVG(h.hour_mean) AS day_mean`.match(UNWEIGHTED_HOURS)).toHaveLength( + 1, + ); + expect( + `hours.reduce((s, h) => s + h.total / h.count, 0) / hours.length`.match( + UNWEIGHTED_HOURS, + ), + ).toHaveLength(1); + expect(`SUM(h.hour_mean * w) / SUM(w)`.match(UNWEIGHTED_HOURS)).toBeNull(); + }); + it("catches a planted plain AVG and a planted fold", () => { expect( `SELECT AVG( m."value" ) FROM measurements m`.match(SQL_MEAN), diff --git a/src/app/api/admin/maintenance/pulse-day-mean/route.ts b/src/app/api/admin/maintenance/pulse-day-mean/route.ts index f5bb44f50..e13c592e4 100644 --- a/src/app/api/admin/maintenance/pulse-day-mean/route.ts +++ b/src/app/api/admin/maintenance/pulse-day-mean/route.ts @@ -3,30 +3,70 @@ * method (#1107 follow-up). * * GET answers the active method, when it was applied and the last dry run - * (status and result: per-account labels only, never an id). POST + * (status and result: per-account ordinal labels only, never an id). POST * `{ action: "dry_run" | "apply" }` answers 202 `{ enqueued }`: a dry run * computes both methods for every account and day and writes nothing but its * result; `apply` needs a successful dry run from the last seven days (409 * `maintenance.pulseDayMean.dryRunRequired`), switches the method and * rebuilds the pulse rollups. 409 `maintenance.pulseDayMean.busy` while a run - * is active. Admin only, cookie session only. + * is active. Admin only, cookie session only. The work itself runs in the + * background worker (`src/lib/jobs/pulse-day-mean.ts`). */ import { NextRequest } from "next/server"; import { apiHandler, requireAdmin } from "@/lib/api-handler"; -import { returnAllZodIssues, safeJson } from "@/lib/api-response"; -import { notImplemented } from "@/lib/http/not-implemented"; +import { + apiError, + apiSuccess, + getClientIp, + returnAllZodIssues, + safeJson, +} from "@/lib/api-response"; +import { auditLog } from "@/lib/auth/audit"; +import { prisma } from "@/lib/db"; +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { + hasFreshDryRun, + isPulseDayMeanRunActive, + PULSE_DAY_MEAN_QUEUE, + PULSE_DAY_MEAN_SEND_OPTIONS, + readPulseDayMeanStatus, + writeDryRun, + type PulseDayMeanPayload, +} from "@/lib/jobs/pulse-day-mean"; +import { annotate } from "@/lib/logging/context"; +import { + setActivePulseDayMeanMethod, + type PulseDayMeanMethod, +} from "@/lib/measurements/day-statistic"; import { pulseDayMeanActionSchema } from "@/lib/validations/pulse-day-mean"; export const dynamic = "force-dynamic"; +/** The method the apply step switches to. */ +const APPLIED_METHOD: PulseDayMeanMethod = "weighted_hour_mean_v2"; + +function busy() { + return apiError("A pulse day-mean run is already queued or running", 409, { + errorCode: "maintenance.pulseDayMean.busy", + }); +} + export const GET = apiHandler(async () => { await requireAdmin(); - return notImplemented("admin.pulse_day_mean.get"); + const status = await readPulseDayMeanStatus(); + annotate({ + action: { name: "admin.pulse_day_mean.status" }, + meta: { + method: status.method, + dry_run_status: status.lastDryRun?.status ?? null, + }, + }); + return apiSuccess(status); }); export const POST = apiHandler(async (request: NextRequest) => { - await requireAdmin(); + const { user } = await requireAdmin(); const { data: body, error: jsonError } = await safeJson(request, { maxBytes: 1024, @@ -34,6 +74,80 @@ export const POST = apiHandler(async (request: NextRequest) => { if (jsonError) return jsonError; const parsed = pulseDayMeanActionSchema.safeParse(body); if (!parsed.success) return returnAllZodIssues(parsed.error, 422); + const action = parsed.data.action; + annotate({ + action: { name: "admin.pulse_day_mean.requested" }, + meta: { pulse_day_mean_action: action }, + }); - return notImplemented("admin.pulse_day_mean.post"); + const boss = getGlobalBoss(); + if (!boss) return apiError("Background worker is not available", 503); + if (await isPulseDayMeanRunActive()) return busy(); + + if (action === "dry_run") { + const previous = (await readPulseDayMeanStatus()).lastDryRun; + await writeDryRun({ + status: "queued", + startedAt: new Date().toISOString(), + finishedAt: null, + result: null, + }); + const payload: PulseDayMeanPayload = { action: "dry_run" }; + const jobId = await boss.send( + PULSE_DAY_MEAN_QUEUE, + payload, + PULSE_DAY_MEAN_SEND_OPTIONS, + ); + if (jobId === null) { + // Another run took the singleton between the check and the send; the + // last dry run stays what it was. + if (previous) await writeDryRun(previous); + return busy(); + } + annotate({ meta: { pulse_day_mean_enqueued: true } }); + return apiSuccess({ enqueued: true }, 202); + } + + if (!hasFreshDryRun(await readPulseDayMeanStatus())) { + return apiError( + "Run a dry run first; apply needs one finished in the last seven days", + 409, + { errorCode: "maintenance.pulseDayMean.dryRunRequired" }, + ); + } + + // The method first, then the rebuild: the rollups read the stored method + // inside their own statements, so a rebuild that starts before the switch + // would write the old statistic again. + const appliedAt = new Date(); + await prisma.appSettings.upsert({ + where: { id: "singleton" }, + create: { + id: "singleton", + pulseDayMeanMethod: APPLIED_METHOD, + pulseDayMeanAppliedAt: appliedAt, + }, + update: { + pulseDayMeanMethod: APPLIED_METHOD, + pulseDayMeanAppliedAt: appliedAt, + }, + }); + setActivePulseDayMeanMethod(APPLIED_METHOD); + const payload: PulseDayMeanPayload = { action: "rebuild" }; + const jobId = await boss.send( + PULSE_DAY_MEAN_QUEUE, + payload, + PULSE_DAY_MEAN_SEND_OPTIONS, + ); + const enqueued = jobId !== null; + + await auditLog("admin.pulseDayMean.apply", { + userId: user.id, + ipAddress: getClientIp(request), + details: { method: APPLIED_METHOD, rebuildEnqueued: enqueued }, + }); + annotate({ + meta: { pulse_day_mean_enqueued: enqueued, method: APPLIED_METHOD }, + }); + return apiSuccess({ enqueued }, 202); }); diff --git a/src/app/api/measurements/series/route.ts b/src/app/api/measurements/series/route.ts index 72901ddb9..b3abf5504 100644 --- a/src/app/api/measurements/series/route.ts +++ b/src/app/api/measurements/series/route.ts @@ -35,7 +35,9 @@ import { loadUserSourcePriority } from "@/lib/rollups/measurement-read"; import { usesHourlyMeanDay } from "@/lib/measurements/day-statistic"; import { dayWeightedRows, + hourWeightSql, readingsMean, + weightedHourMeanSql, windowMeanSql, zoneDayFrame, } from "@/lib/measurements/day-mean"; @@ -485,8 +487,8 @@ export const GET = apiHandler(async (request: NextRequest) => { const type = KIND_TO_TYPE[kind]; const seriesRows = seriesRowsFrom(priorityJson, type, days); // A day of a type with an activity-dependent sampling rate (pulse) is the - // mean of its local hours' means, each hour once, so a workout hour with - // hundreds of times the readings does not outweigh the rest of the day (see + // 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 // `day-statistic.ts`). The hour grain is already one mean per hour. const hourWeighted = grain === "day" && usesHourlyMeanDay(type); const bucketRows = await prisma.$queryRawUnsafe< @@ -515,6 +517,7 @@ export const GET = apiHandler(async (request: NextRequest) => { SELECT date_trunc($2, local_hour) AS local_day, AVG("value") AS hour_mean, + COUNT(*) AS n, MIN(COALESCE("value_min", "value")) AS min_value, MAX(COALESCE("value_max", "value")) AS max_value FROM localized @@ -522,7 +525,7 @@ export const GET = apiHandler(async (request: NextRequest) => { ) SELECT local_day AT TIME ZONE $3 AS bucket_start, - AVG(hour_mean)::double precision AS mean, + ${weightedHourMeanSql("hour_mean", hourWeightSql("n"))}::double precision AS mean, MIN(min_value)::double precision AS min_value, MAX(max_value)::double precision AS max_value FROM hourly diff --git a/src/instrumentation.ts b/src/instrumentation.ts index 78981c5e6..7927effd9 100644 --- a/src/instrumentation.ts +++ b/src/instrumentation.ts @@ -39,6 +39,16 @@ export async function register() { // A diagnostic must never break boot. } + // Every pulse reader follows the operator's pulse day-mean method; mirror + // it in this process and keep the mirror current. + try { + const { startPulseDayMeanMethodSync } = + await import("@/lib/measurements/pulse-day-mean-method"); + await startPulseDayMeanMethodSync(); + } catch { + // Without the mirror the process keeps the default method. + } + // Every Node process watches its own event loop — web-only // containers included; a stall report from the process serving // requests is the whole point. diff --git a/src/lib/day/daily-stats.ts b/src/lib/day/daily-stats.ts index a9bcba033..d3fa3692f 100644 --- a/src/lib/day/daily-stats.ts +++ b/src/lib/day/daily-stats.ts @@ -23,6 +23,7 @@ import { } from "@/generated/prisma/client"; import { prisma } from "@/lib/db"; import { isCumulativeDaySumType } from "@/lib/measurements/cumulative-day-sum"; +import { combineHourCells } from "@/lib/measurements/day-mean"; import { HOURLY_MEAN_DAY_TYPES } from "@/lib/measurements/day-statistic"; import { dateOnlyAtNoonUtc, dateOnlyKey } from "@/lib/tz/date-only"; import { collapseRollupRowsBySource } from "@/lib/rollups/measurement-read"; @@ -169,8 +170,9 @@ export function foldDailyCells( if (cumulative) { value = hours.reduce((s, h) => s + h.total, 0); } else if (hourly) { - value = - hours.reduce((s, h) => s + h.total / h.count, 0) / hours.length; + value = combineHourCells( + hours.map((h) => ({ sum: h.total, n: h.count })), + ) as number; } else { value = hours.reduce((s, h) => s + h.total, 0) / count; } diff --git a/src/lib/doctor-report/__tests__/day-bucket-latest-and-count.test.ts b/src/lib/doctor-report/__tests__/day-bucket-latest-and-count.test.ts index b83ac5e05..54dc53f7a 100644 --- a/src/lib/doctor-report/__tests__/day-bucket-latest-and-count.test.ts +++ b/src/lib/doctor-report/__tests__/day-bucket-latest-and-count.test.ts @@ -45,8 +45,8 @@ function bucket( clinicalTar2Count: 0, clinicalLowRiskSum: null, clinicalHighRiskSum: null, - hourMeanSum: null, - hourCount: null, + hourWeightedSum: null, + hourWeight: null, }; } diff --git a/src/lib/doctor-report/dense-buckets.ts b/src/lib/doctor-report/dense-buckets.ts index 8d05fdab9..3810fcda7 100644 --- a/src/lib/doctor-report/dense-buckets.ts +++ b/src/lib/doctor-report/dense-buckets.ts @@ -13,7 +13,10 @@ */ import { Prisma } from "@/generated/prisma/client"; import { prisma } from "@/lib/db"; -import { isHourlyMeanTypeSql } from "@/lib/measurements/day-mean"; +import { + hourWeightSql, + isHourlyMeanTypeSql, +} from "@/lib/measurements/day-mean"; import type { GlucoseContext, MeasurementSource, @@ -47,11 +50,12 @@ export interface DenseMeasurementBucket { clinicalHighRiskSum: number | null; /** * For an hourly-mean type (pulse, see `day-mean.ts`): the sum of the - * bucket's local hours' means and the number of those hours, so the day - * value is their ratio. `null` for every other type. + * bucket's local hours' means, each times its hour weight, and the sum of + * those weights, so the day value is their ratio. `null` for every other + * type. */ - hourMeanSum: number | null; - hourCount: number | null; + hourWeightedSum: number | null; + hourWeight: number | null; } /** @@ -75,6 +79,8 @@ export async function loadDenseMeasurementBuckets(params: { ); // Closed code constant (the hourly-mean type list), spliced as literals. const hourlyMeanType = Prisma.raw(isHourlyMeanTypeSql('"type"')); + // Each hour's weight under the active pulse day-mean method (code-built). + const hourWeight = Prisma.raw(hourWeightSql("n")); return prisma.$queryRaw` WITH localized AS ( SELECT @@ -135,8 +141,9 @@ export async function loadDenseMeasurementBuckets(params: { "measured_at" DESC, "id" DESC ), - -- The hourly-mean day (pulse): each local hour's mean, then their sum - -- and count per bucket. Empty for every other type. + -- The hourly-mean day (pulse): each local hour's mean and reading + -- count, then the weighted sum of the means and the sum of the hour + -- weights per bucket. Empty for every other type. hour_means AS ( SELECT local_day, @@ -144,7 +151,8 @@ export async function loadDenseMeasurementBuckets(params: { "source", "device_type", "glucose_context", - AVG("value") AS hour_mean + AVG("value") AS hour_mean, + COUNT(*) AS n FROM localized WHERE ${hourlyMeanType} GROUP BY @@ -162,8 +170,8 @@ export async function loadDenseMeasurementBuckets(params: { "source", "device_type", "glucose_context", - SUM(hour_mean)::double precision AS hour_mean_sum, - COUNT(*)::int AS hour_count + SUM(hour_mean * ${hourWeight})::double precision AS hour_weighted_sum, + SUM(${hourWeight})::double precision AS hour_weight FROM hour_means GROUP BY local_day, @@ -248,8 +256,8 @@ export async function loadDenseMeasurementBuckets(params: { grouped.clinical_tar2_count AS "clinicalTar2Count", grouped.clinical_low_risk_sum AS "clinicalLowRiskSum", grouped.clinical_high_risk_sum AS "clinicalHighRiskSum", - hour_agg.hour_mean_sum AS "hourMeanSum", - hour_agg.hour_count AS "hourCount" + hour_agg.hour_weighted_sum AS "hourWeightedSum", + hour_agg.hour_weight AS "hourWeight" FROM grouped INNER JOIN latest_rows ON latest_rows.local_day = grouped.local_day diff --git a/src/lib/doctor-report/measurement-series.ts b/src/lib/doctor-report/measurement-series.ts index c00d76eb0..e353f3388 100644 --- a/src/lib/doctor-report/measurement-series.ts +++ b/src/lib/doctor-report/measurement-series.ts @@ -62,8 +62,8 @@ export function summariseDenseBuckets( max: number; latestAt: Date; latest: number; - hourMeanSum: number; - hourCount: number; + hourWeightedSum: number; + hourWeight: number; } >(); for (const row of canonical) { @@ -79,15 +79,15 @@ export function summariseDenseBuckets( max: row.maxValue, latestAt: row.latestAt, latest: row.latestValue, - hourMeanSum: row.hourMeanSum ?? 0, - hourCount: row.hourCount ?? 0, + hourWeightedSum: row.hourWeightedSum ?? 0, + hourWeight: row.hourWeight ?? 0, }); continue; } current.count += row.count; current.sum += row.sumValue; - current.hourMeanSum += row.hourMeanSum ?? 0; - current.hourCount += row.hourCount ?? 0; + current.hourWeightedSum += row.hourWeightedSum ?? 0; + current.hourWeight += row.hourWeight ?? 0; current.min = Math.min(current.min, row.minValue); current.max = Math.max(current.max, row.maxValue); if (row.latestAt > current.latestAt) { @@ -116,16 +116,16 @@ export function summariseDenseBuckets( } }; - // For an hourly-mean type (pulse, see `day-mean.ts`) a day is the mean of - // its local hours' means and the period average the mean of its days, so a - // workout hour's dense samples do not outweigh the rest of the day, nor a - // workout day the rest of the period. Count, min and max stay over every + // For an hourly-mean type (pulse, see `day-mean.ts`) a day is the weighted + // mean of its local hours' means and the period average the mean of its + // days, so a workout hour's dense samples do not outweigh the rest of the + // day, nor a workout day the rest of the period. Count, min and max stay over every // reading. Every other type keeps the mean of its readings. const dayMeans = new Map(); for (const row of byTypeDay.values()) { const value = - usesHourlyMeanDay(row.type) && row.hourCount > 0 - ? row.hourMeanSum / row.hourCount + usesHourlyMeanDay(row.type) && row.hourWeight > 0 + ? row.hourWeightedSum / row.hourWeight : row.sum / row.count; (byType[row.type] ??= []).push({ value, diff --git a/src/lib/insights/correlation-channel-series.ts b/src/lib/insights/correlation-channel-series.ts index a0f6d8fce..0d1db8cf8 100644 --- a/src/lib/insights/correlation-channel-series.ts +++ b/src/lib/insights/correlation-channel-series.ts @@ -67,7 +67,7 @@ import type { } from "@/generated/prisma/client"; import { TRACKED_INTAKE_WHERE } from "@/lib/medications/intake-tracking"; import { dateOnlyKey } from "@/lib/tz/date-only"; -import { hourMeanRows } from "@/lib/measurements/day-mean"; +import { dayValueRows } from "@/lib/measurements/day-mean"; import { usesHourlyMeanDay } from "@/lib/measurements/day-statistic"; import { decryptSymptomText } from "@/lib/symptoms/server"; import { @@ -750,11 +750,11 @@ export function buildMeasurementDailySeries( if (CUMULATIVE_HK_TYPES.has(type)) { return buildCumulativeDailySeries(type, rows, tz, priorityJson); } - // A day of pulse is the mean of its local hours' means (`day-mean.ts`): - // collapse each local hour to its mean first, so the daily mean below - // weighs every hour once. Every other type passes through unchanged. + // A day of pulse is the weighted mean of its local hours' means + // (`day-mean.ts`): collapse each local day to that value first, so the daily + // mean below passes it through. Every other type passes through unchanged. return toDailyMeans( - hourMeanRows( + dayValueRows( type, rows.map((r) => ({ value: r.value, measuredAt: r.at })), tz, diff --git a/src/lib/jobs/pulse-day-mean.ts b/src/lib/jobs/pulse-day-mean.ts new file mode 100644 index 000000000..3b613434a --- /dev/null +++ b/src/lib/jobs/pulse-day-mean.ts @@ -0,0 +1,441 @@ +/** + * The pulse day-mean method switch (#1107 follow-up): dry run, then apply. + * + * Both start from `POST /api/admin/maintenance/pulse-day-mean` (cookie-only + * admin) and run here, in the background worker, one at a time. + * + * - `dry_run` computes, for every account with pulse and every UTC day, the + * day value under both methods on the account's canonical source, and + * writes one thing: the result, to `app_settings.pulse_day_mean_dry_run_json` + * (how many days move by more than 1 bpm, the largest move, a distribution, + * and the same per account under an ordinal label). Nothing else is + * written, the rollup tier included. + * - `rebuild` follows the apply step, which has already stored the new + * method: it recomputes the PULSE rollups of every account, which read the + * stored method inside their own statement. + * + * Both work one account and one window at a time (`DRY_RUN_WINDOW_DAYS`, + * `REBUILD_WINDOW_DAYS`), so the process holds a window's aggregate rows and + * never an account's readings: the heap stays flat however long the history. + */ +import type { Job } from "pg-boss"; + +import { buildSourceRankCase } from "@/lib/analytics/source-rank-sql"; +import { prisma } from "@/lib/db"; +import { withJobLock } from "@/lib/jobs/job-lock"; +import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; +import { withBackgroundEvent } from "@/lib/logging/background"; +import { logCaught } from "@/lib/logging/signal"; +import { + hourWeightSql, + weightedHourMeanSql, +} from "@/lib/measurements/day-mean"; +import { + isPulseDayMeanMethod, + DEFAULT_PULSE_DAY_MEAN_METHOD, +} from "@/lib/measurements/day-statistic"; +import { loadUserSourcePriority } from "@/lib/rollups/measurement-read"; +import { + recomputeUserRollups, + ROLLUP_FOLD_WINDOW_MS, +} from "@/lib/rollups/measurement-rollups"; +import { + PULSE_DAY_MEAN_CHANGE_BUCKETS, + pulseDayMeanDryRunSchema, + type PulseDayMeanDryRun, + type PulseDayMeanDryRunResult, + type PulseDayMeanStatus, +} from "@/lib/validations/pulse-day-mean"; + +export const PULSE_DAY_MEAN_QUEUE = "pulse-day-mean"; + +/** + * One-shot and exclusive: a dry run or a rebuild that stops part way is + * started again by the operator, never retried beside a half-finished one. + */ +export const PULSE_DAY_MEAN_SEND_OPTIONS = { + retryLimit: 0, + expireInSeconds: 6 * 60 * 60, + singletonKey: PULSE_DAY_MEAN_QUEUE, +} as const; + +export interface PulseDayMeanPayload { + action: "dry_run" | "rebuild"; +} + +/** How old a finished dry run may be for the apply step to accept it. */ +export const DRY_RUN_FRESH_MS = 7 * 24 * 60 * 60 * 1000; + +const DAY_MS = 24 * 60 * 60 * 1000; +/** Days of one account's pulse a dry-run statement covers. */ +const DRY_RUN_WINDOW_DAYS = 90; +/** Days of one account's pulse a rebuild statement covers. */ +const REBUILD_WINDOW_DAYS = 366; +/** A window's statements may outlast the request limit, never this. */ +const STATEMENT_BUDGET_MS = 5 * 60 * 1000; + +/** A day "changed" when its value moves by more than this many bpm. */ +const CHANGED_OVER_BPM = 1; + +// ─── Status ─────────────────────────────────────────────── + +/** Whether a dry run or rebuild is queued or running. */ +export async function isPulseDayMeanRunActive(): Promise { + try { + const rows = await prisma.$queryRaw>` + SELECT count(*)::int AS n + FROM pgboss.job + WHERE name = ${PULSE_DAY_MEAN_QUEUE} + AND state IN ('created', 'retry', 'active') + `; + return (rows[0]?.n ?? 0) > 0; + } catch { + // No queue schema (web-only deployment): nothing can be running. + return false; + } +} + +function parseDryRun(value: unknown): PulseDayMeanDryRun | null { + const parsed = pulseDayMeanDryRunSchema.safeParse(value); + return parsed.success ? parsed.data : null; +} + +/** + * The stored method, when it was applied, and the last dry run. A dry run + * still marked queued or running with no live job behind it (the worker went + * away mid-run) reads as failed. + */ +export async function readPulseDayMeanStatus(): Promise { + const row = await prisma.appSettings.findUnique({ + where: { id: "singleton" }, + select: { + pulseDayMeanMethod: true, + pulseDayMeanAppliedAt: true, + pulseDayMeanDryRunJson: true, + }, + }); + let lastDryRun = parseDryRun(row?.pulseDayMeanDryRunJson ?? null); + if ( + lastDryRun && + (lastDryRun.status === "queued" || lastDryRun.status === "running") && + !(await isPulseDayMeanRunActive()) + ) { + lastDryRun = { ...lastDryRun, status: "failed" }; + } + return { + method: isPulseDayMeanMethod(row?.pulseDayMeanMethod) + ? row.pulseDayMeanMethod + : DEFAULT_PULSE_DAY_MEAN_METHOD, + appliedAt: row?.pulseDayMeanAppliedAt?.toISOString() ?? null, + lastDryRun, + }; +} + +/** A finished dry run from the last seven days, the apply step's condition. */ +export function hasFreshDryRun( + status: PulseDayMeanStatus, + now: Date = new Date(), +): boolean { + const run = status.lastDryRun; + if (!run || run.status !== "done" || run.finishedAt === null) return false; + const finished = Date.parse(run.finishedAt); + return Number.isFinite(finished) && now.getTime() - finished <= DRY_RUN_FRESH_MS; +} + +/** The only write a dry run makes. */ +export async function writeDryRun(run: PulseDayMeanDryRun): Promise { + await prisma.appSettings.upsert({ + where: { id: "singleton" }, + create: { id: "singleton", pulseDayMeanDryRunJson: run }, + update: { pulseDayMeanDryRunJson: run }, + }); +} + +// ─── Accounts ───────────────────────────────────────────── + +interface PulseAccount { + id: string; + first: Date; + last: Date; +} + +/** Accounts holding pulse, oldest account first, with their pulse span. */ +async function pulseAccounts(): Promise { + const rows = await prisma.$queryRaw< + Array<{ id: string; first: Date | null; last: Date | null }> + >` + SELECT u."id", + (SELECT MIN(m."measured_at") FROM measurements m + WHERE m."user_id" = u."id" AND m."type" = 'PULSE' + AND m."deleted_at" IS NULL) AS first, + (SELECT MAX(m."measured_at") FROM measurements m + WHERE m."user_id" = u."id" AND m."type" = 'PULSE' + AND m."deleted_at" IS NULL) AS last + FROM users u + ORDER BY u."created_at", u."id" + `; + return rows + .filter((r) => r.first !== null && r.last !== null) + .map((r) => ({ id: r.id, first: r.first!, last: r.last! })); +} + +/** `[from, to)` windows of `days` covering `[first, last]`, UTC-day aligned. */ +function windows( + first: Date, + last: Date, + days: number, +): Array<{ from: Date; to: Date }> { + const out: Array<{ from: Date; to: Date }> = []; + const start = Math.floor(first.getTime() / DAY_MS) * DAY_MS; + const end = (Math.floor(last.getTime() / DAY_MS) + 1) * DAY_MS; + for (let t = start; t < end; t += days * DAY_MS) { + out.push({ from: new Date(t), to: new Date(Math.min(end, t + days * DAY_MS)) }); + } + return out; +} + +// ─── Dry run ────────────────────────────────────────────── + +interface WindowComparison { + days: number; + over1: number; + maxDelta: number; + buckets: [number, number, number, number]; +} + +/** + * One account's UTC days in `[from, to)`: each day's value on its canonical + * source under both methods, compared. Aggregated in the statement; one row + * comes back however many readings the window holds. + */ +async function compareWindow( + userId: string, + rank: string, + from: Date, + to: Date, +): Promise { + const v1 = weightedHourMeanSql( + "h.hour_mean", + hourWeightSql("h.n", "hour_mean_v1"), + ); + const v2 = weightedHourMeanSql( + "h.hour_mean", + hourWeightSql("h.n", "weighted_hour_mean_v2"), + ); + // Whitelist splice: `v1` / `v2` are built from code constants and `rank` + // from the closed source enum (`buildSourceRankCase`); every value is a + // positional parameter. + const rows = await prisma.$transaction(async (tx) => { + await tx.$executeRaw`SELECT set_config('statement_timeout', ${String(STATEMENT_BUDGET_MS)}, true)`; + return tx.$queryRawUnsafe< + Array<{ + days: number; + over1: number; + max_delta: number; + b0: number; + b1: number; + b2: number; + b3: number; + }> + >( + ` + WITH h AS ( + SELECT m."source", + date_trunc('hour', m."measured_at") AS hr, + AVG(m."value") AS hour_mean, + COUNT(*) AS n + FROM measurements m + WHERE m."user_id" = $1 + AND m."type" = 'PULSE' + AND m."deleted_at" IS NULL + AND m."measured_at" >= $2 + AND m."measured_at" < $3 + GROUP BY 1, 2 + ), + d AS ( + SELECT h."source", date_trunc('day', h.hr) AS day, + ${v1} AS v1, + ${v2} AS v2 + FROM h + GROUP BY 1, 2 + ), + c AS ( + SELECT DISTINCT ON (d.day) ABS(d.v2 - d.v1) AS delta + FROM d + ORDER BY d.day, (${rank}), d."source"::text + ) + SELECT COUNT(*)::int AS days, + COUNT(*) FILTER (WHERE delta > $4)::int AS over1, + COALESCE(MAX(delta), 0)::double precision AS max_delta, + COUNT(*) FILTER (WHERE delta <= 1)::int AS b0, + COUNT(*) FILTER (WHERE delta > 1 AND delta <= 2)::int AS b1, + COUNT(*) FILTER (WHERE delta > 2 AND delta <= 5)::int AS b2, + COUNT(*) FILTER (WHERE delta > 5)::int AS b3 + FROM c + `, + userId, + from, + to, + CHANGED_OVER_BPM, + ); + }); + const r = rows[0]; + return { + days: Number(r?.days ?? 0), + over1: Number(r?.over1 ?? 0), + maxDelta: Number(r?.max_delta ?? 0), + buckets: [ + Number(r?.b0 ?? 0), + Number(r?.b1 ?? 0), + Number(r?.b2 ?? 0), + Number(r?.b3 ?? 0), + ], + }; +} + +/** Compare both methods over every account's pulse. Writes nothing. */ +export async function computePulseDayMeanDryRun(): Promise { + const buckets = [0, 0, 0, 0]; + let daysCompared = 0; + let over1 = 0; + let maxChange = 0; + const perAccount: PulseDayMeanDryRunResult["perAccount"] = []; + const accounts = await pulseAccounts(); + for (const [i, account] of accounts.entries()) { + const rank = buildSourceRankCase( + await loadUserSourcePriority(account.id), + "'PULSE'", + 'd."source"', + ); + let days = 0; + let changed = 0; + for (const w of windows(account.first, account.last, DRY_RUN_WINDOW_DAYS)) { + const c = await compareWindow(account.id, rank, w.from, w.to); + days += c.days; + changed += c.over1; + maxChange = Math.max(maxChange, c.maxDelta); + c.buckets.forEach((n, b) => (buckets[b] += n)); + } + daysCompared += days; + over1 += changed; + perAccount.push({ account: `#${i + 1}`, days, changedOver1Bpm: changed }); + } + return { + accounts: accounts.length, + daysCompared, + daysChangedOver1Bpm: over1, + maxChangeBpm: Math.round(maxChange * 10) / 10, + distribution: PULSE_DAY_MEAN_CHANGE_BUCKETS.map((bucket, b) => ({ + bucket, + days: buckets[b], + })), + perAccount, + }; +} + +// ─── Rebuild ────────────────────────────────────────────── + +/** + * Recompute every account's PULSE rollups inside the fold window, a year of + * one account at a time. Returns the accounts done and the rows written. + */ +export async function rebuildPulseRollups(): Promise<{ + accounts: number; + rows: number; +}> { + const floor = new Date(Date.now() - ROLLUP_FOLD_WINDOW_MS); + let rows = 0; + let accounts = 0; + for (const account of await pulseAccounts()) { + if (account.last < floor) continue; + const first = account.first < floor ? floor : account.first; + for (const w of windows(first, account.last, REBUILD_WINDOW_DAYS)) { + const done = await recomputeUserRollups(account.id, { + types: ["PULSE"], + from: w.from, + to: w.to, + statementTimeoutMs: STATEMENT_BUDGET_MS, + }); + rows += done.rowsUpserted; + } + accounts += 1; + } + return { accounts, rows }; +} + +// ─── Job ────────────────────────────────────────────────── + +async function runDryRun(): Promise { + const startedAt = new Date().toISOString(); + await writeDryRun({ + status: "running", + startedAt, + finishedAt: null, + result: null, + }); + try { + const result = await computePulseDayMeanDryRun(); + await writeDryRun({ + status: "done", + startedAt, + finishedAt: new Date().toISOString(), + result, + }); + return result; + } catch (err) { + await writeDryRun({ + status: "failed", + startedAt, + finishedAt: new Date().toISOString(), + result: null, + }).catch((writeErr) => + logCaught("pulse_day_mean.dry_run.status_write_failed", writeErr), + ); + throw err; + } +} + +export async function handlePulseDayMean( + jobs: Job[], +): Promise { + const action = jobs[0]?.data?.action === "rebuild" ? "rebuild" : "dry_run"; + return withBackgroundEvent("job.pulse_day_mean", async (evt) => { + const startedAt = Date.now(); + evt.addMeta("pulse_day_mean_action", action); + try { + const guarded = await withJobLock(PULSE_DAY_MEAN_QUEUE, async () => + action === "rebuild" + ? { kind: "rebuild" as const, ...(await rebuildPulseRollups()) } + : { kind: "dry_run" as const, result: await runDryRun() }, + ); + if (!guarded.ran) { + return jobDone({ outcome: "already_running" }); + } + const done = guarded.result; + if (done.kind === "rebuild") { + evt.addMeta("pulse_day_mean_accounts", done.accounts); + evt.addMeta("pulse_day_mean_rows", done.rows); + return jobDone({ + outcome: "rebuilt", + accounts_completed: done.accounts, + rows_upserted: done.rows, + duration_ms: Date.now() - startedAt, + }); + } + evt.addMeta("pulse_day_mean_accounts", done.result.accounts); + evt.addMeta("pulse_day_mean_days", done.result.daysCompared); + evt.addMeta("pulse_day_mean_changed", done.result.daysChangedOver1Bpm); + return jobDone({ + outcome: "compared", + accounts_completed: done.result.accounts, + processed: done.result.daysCompared, + duration_ms: Date.now() - startedAt, + }); + } catch (err) { + logCaught(`pulse_day_mean.${action}_failed`, err); + return jobFailed(`pulse day-mean ${action} failed`, err, { + duration_ms: Date.now() - startedAt, + }); + } + }); +} diff --git a/src/lib/jobs/queue-runtime.ts b/src/lib/jobs/queue-runtime.ts index 319df60e9..3b94cd1a9 100644 --- a/src/lib/jobs/queue-runtime.ts +++ b/src/lib/jobs/queue-runtime.ts @@ -41,6 +41,7 @@ import { APPLE_HEALTH_IMPORT_SEND_OPTIONS } from "@/lib/jobs/apple-health-import import { MEDICATION_INTAKE_IMPORT_SEND_OPTIONS } from "@/lib/jobs/medication-intake-import"; import { HEALTH_CONNECT_IMPORT_SEND_OPTIONS } from "@/lib/jobs/health-connect-import-worker"; import { MEASUREMENT_MAINTENANCE_SEND_OPTIONS } from "@/lib/jobs/measurement-maintenance"; +import { PULSE_DAY_MEAN_SEND_OPTIONS } from "@/lib/jobs/pulse-day-mean"; /** pg-boss's default job expiry, in seconds. */ export const DEFAULT_JOB_EXPIRE_SECONDS = 15 * 60; @@ -410,6 +411,23 @@ export const QUEUE_RUNTIME: Readonly> = { }, why: "Rebuilds every index of the largest table, one at a time.", }, + "pulse-day-mean": { + runtime: "long", + expireInSeconds: PULSE_DAY_MEAN_SEND_OPTIONS.expireInSeconds, + expiryVia: ["PULSE_DAY_MEAN_SEND_OPTIONS"], + stop: { + oneShot: + "A dry run or rebuild that stops part way is started again by the operator; the send sets retryLimit 0.", + }, + exclusive: { + marker: "retryLimit: 0", + at: { + file: "lib/jobs/pulse-day-mean.ts", + fn: "PULSE_DAY_MEAN_SEND_OPTIONS", + }, + }, + why: "Walks every account's pulse history a window at a time, twice over for a dry run.", + }, "medication-intake-import": { runtime: "long", expireInSeconds: MEDICATION_INTAKE_IMPORT_SEND_OPTIONS.expireInSeconds, diff --git a/src/lib/jobs/reminder/register-maintenance.ts b/src/lib/jobs/reminder/register-maintenance.ts index 2e8d413b0..76e4ac419 100644 --- a/src/lib/jobs/reminder/register-maintenance.ts +++ b/src/lib/jobs/reminder/register-maintenance.ts @@ -83,6 +83,11 @@ import { handleMeasurementMaintenance, type MeasurementMaintenancePayload, } from "@/lib/jobs/measurement-maintenance"; +import { + PULSE_DAY_MEAN_QUEUE, + handlePulseDayMean, + type PulseDayMeanPayload, +} from "@/lib/jobs/pulse-day-mean"; import { HEALTH_CONNECT_IMPORT_QUEUE, HEALTH_CONNECT_IMPORT_CONCURRENCY, @@ -496,6 +501,8 @@ const allQueues = [ // cron: the admin route enqueues it. Without this entry the trigger // silently no-ops. MEASUREMENT_MAINTENANCE_QUEUE, + // v1.43 — the pulse day-mean dry run and rebuild, admin-triggered only. + PULSE_DAY_MEAN_QUEUE, // v1.8.2 — one-time duplicate dose-slot cleanup. Boot discovery enqueues // one job per user holding two live intake rows that snap to the same // canonical slot (the pre-fix REMINDER-pending + API-taken pair). Also @@ -835,6 +842,12 @@ const queuePolicies: QueuePolicyTable = { reason: "Fixed singleton key, admin-triggered. Two maintenance passes would rebuild the same indexes side by side; the route reports a suppressed send back as enqueued: false.", }, + // v1.43 — the pulse day-mean dry run and rebuild. + [PULSE_DAY_MEAN_QUEUE]: { + policy: "exclusive", + reason: + "Fixed singleton key, admin-triggered. A dry run beside a rebuild would compare against rollups being rewritten; the route answers 409 busy instead.", + }, // v1.42 — compaction-tombstone backlog purge. `short`, not `exclusive`: // a run that stops early sends its own follow-up while it is still active, // and `exclusive` would swallow that send. A boot's send while a run is @@ -1228,6 +1241,13 @@ export async function registerMaintenanceQueues( { localConcurrency: 1 }, handleMeasurementMaintenance, ); + // v1.43 — pulse day-mean dry run and rebuild, operator-triggered only. + await createAndWork( + boss, + PULSE_DAY_MEAN_QUEUE, + { localConcurrency: 1 }, + handlePulseDayMean, + ); // v1.32.1 (issue #588) — periodic orphan-ImportJob sweep. Single-flight: // the underlying `updateMany` is idempotent and two ticks racing the same // read-then-write pass would just be wasted work. diff --git a/src/lib/jobs/subsystem-surface.ts b/src/lib/jobs/subsystem-surface.ts index ed4dfa171..d20e764ce 100644 --- a/src/lib/jobs/subsystem-surface.ts +++ b/src/lib/jobs/subsystem-surface.ts @@ -123,6 +123,7 @@ const subsystemSurface = { "compaction-tombstone-purge": { audience: "system" }, "measurement-fold-repair": { audience: "system" }, "measurement-maintenance": { audience: "system" }, + "pulse-day-mean": { audience: "system" }, "medication-intake-import": { audience: "system" }, "intake-slot-dedup": { audience: "system" }, "mood-reminder-cleanup": { audience: "system" }, diff --git a/src/lib/measurements/__tests__/day-mean.test.ts b/src/lib/measurements/__tests__/day-mean.test.ts index f2da9e6fc..db55a8ea8 100644 --- a/src/lib/measurements/__tests__/day-mean.test.ts +++ b/src/lib/measurements/__tests__/day-mean.test.ts @@ -1,17 +1,26 @@ -import { describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it } from "vitest"; import { + combineHourCells, dayValue, dayValueRows, dayWeightedRows, foldMeanSql, - hourMeanRows, + hourWeightSql, isHourlyMeanTypeSql, readingsMean, + storedHourWeightSql, + weightedHourMeanSql, windowMeanSql, zoneDayFrame, SESSION_DAY_FRAME, } from "../day-mean"; +import { + activePulseDayMeanMethod, + hourWeight, + isPulseDayMeanMethod, + setActivePulseDayMeanMethod, +} from "../day-statistic"; const at = (iso: string) => new Date(iso); @@ -93,24 +102,108 @@ describe("readingsMean", () => { }); }); -describe("hourMeanRows / dayValueRows", () => { - it("collapse pulse to one row per hour / per day", () => { +describe("dayValueRows", () => { + it("collapses pulse to one row per day", () => { const rows = workoutDay("2026-03-02"); - const hours = hourMeanRows("PULSE", rows, "UTC"); - expect(hours.map((h) => h.value)).toEqual([150, 60, 60, 60]); - expect(hours[0].measuredAt.toISOString()).toBe("2026-03-02T10:00:00.000Z"); const days = dayValueRows("PULSE", rows, "UTC"); expect(days).toHaveLength(1); expect(days[0].value).toBeCloseTo(82.5, 10); + expect(days[0].measuredAt.toISOString()).toBe("2026-03-02T10:00:00.000Z"); }); - it("hand every other type back untouched", () => { + it("hands every other type back untouched", () => { const rows = workoutDay("2026-03-02"); - expect(hourMeanRows("BLOOD_GLUCOSE", rows, "UTC")).toBe(rows); expect(dayValueRows("BLOOD_GLUCOSE", rows, "UTC")).toBe(rows); }); }); +/** One sparse hour (a single reading at 120), then 23 dense hours at 60. */ +function sparseHourDay(day: string) { + const rows = [{ value: 120, measuredAt: at(`${day}T00:30:00Z`) }]; + for (let h = 1; h < 24; h++) { + for (let m = 0; m < 60; m += 10) { + rows.push({ + value: 60, + measuredAt: at( + `${day}T${String(h).padStart(2, "0")}:${String(m).padStart(2, "0")}:00Z`, + ), + }); + } + } + return rows; +} + +describe("pulse day-mean method", () => { + afterEach(() => setActivePulseDayMeanMethod("hour_mean_v1")); + + it("weighs a lone reading's hour a third under weighted_hour_mean_v2", () => { + const rows = sparseHourDay("2026-03-02"); + expect(dayValue("PULSE", rows, "UTC", "hour_mean_v1")).toBeCloseTo( + 62.5, + 10, + ); + const v2 = dayValue("PULSE", rows, "UTC", "weighted_hour_mean_v2"); + expect(Number(v2!.toFixed(1))).toBe(60.9); + expect(v2).toBeCloseTo((23 * 60 + 120 / 3) / (23 + 1 / 3), 10); + }); + + it("leaves a dense day where it was", () => { + // Every hour holds at least three readings, so every hour weighs one. + const dense = sparseHourDay("2026-03-02").slice(1); + expect(dayValue("PULSE", dense, "UTC", "weighted_hour_mean_v2")).toBe( + dayValue("PULSE", dense, "UTC", "hour_mean_v1"), + ); + }); + + it("follows the active method when the caller names none", () => { + const rows = sparseHourDay("2026-03-02"); + expect(activePulseDayMeanMethod()).toBe("hour_mean_v1"); + expect(dayValue("PULSE", rows, "UTC")).toBeCloseTo(62.5, 10); + setActivePulseDayMeanMethod("weighted_hour_mean_v2"); + expect(dayValue("PULSE", rows, "UTC")).toBeCloseTo( + (23 * 60 + 40) / (23 + 1 / 3), + 10, + ); + expect(readingsMean("PULSE", rows, "UTC")).toBeCloseTo( + (23 * 60 + 40) / (23 + 1 / 3), + 10, + ); + expect(dayValueRows("PULSE", rows, "UTC")[0].value).toBeCloseTo( + (23 * 60 + 40) / (23 + 1 / 3), + 10, + ); + }); + + it("weighs hours by min(1, n / 3)", () => { + expect([1, 2, 3, 10].map((n) => hourWeight(n, "weighted_hour_mean_v2"))) + .toEqual([1 / 3, 2 / 3, 1, 1]); + expect([1, 2, 3].map((n) => hourWeight(n, "hour_mean_v1"))).toEqual([ + 1, 1, 1, + ]); + }); + + it("combines hour cells and skips empty ones", () => { + const cells = [ + { sum: 120, n: 1 }, + { sum: 360, n: 6 }, + { sum: 0, n: 0 }, + ]; + expect(combineHourCells(cells, "hour_mean_v1")).toBe(90); + expect(combineHourCells(cells, "weighted_hour_mean_v2")).toBeCloseTo( + (40 + 60) / (4 / 3), + 10, + ); + expect(combineHourCells([], "weighted_hour_mean_v2")).toBeNull(); + }); + + it("recognises only the two stored values", () => { + expect(isPulseDayMeanMethod("hour_mean_v1")).toBe(true); + expect(isPulseDayMeanMethod("weighted_hour_mean_v2")).toBe(true); + expect(isPulseDayMeanMethod("weighted")).toBe(false); + expect(isPulseDayMeanMethod(null)).toBe(false); + }); +}); + describe("SQL builders", () => { it("splice only the closed type list", () => { expect(isHourlyMeanTypeSql('m."type"')).toBe( @@ -142,4 +235,30 @@ describe("SQL builders", () => { }); expect(fold).toContain("ELSE SUM(c.total) / SUM(c.cnt) END"); }); + + it("build the hour weight per method and refuse a non-column", () => { + expect(hourWeightSql("h.n", "hour_mean_v1")).toBe("1.0"); + expect(hourWeightSql("h.n", "weighted_hour_mean_v2")).toBe( + "LEAST(1.0, (h.n)::double precision / 3.0)", + ); + expect(storedHourWeightSql("h.n")).toContain( + `"pulse_day_mean_method" FROM "app_settings"`, + ); + expect(weightedHourMeanSql("h.hour_mean", "1.0")).toBe( + "(SUM((h.hour_mean) * 1.0) / NULLIF(SUM(1.0), 0))", + ); + expect(() => hourWeightSql("n; DROP", "hour_mean_v1")).toThrow(); + expect(() => weightedHourMeanSql("x'", "1.0")).toThrow(); + }); + + it("weigh the per-row day weight by the method", () => { + const v1 = dayWeightedRows("cm", SESSION_DAY_FRAME, { + method: "hour_mean_v1", + }); + const v2 = dayWeightedRows("cm", SESSION_DAY_FRAME, { + method: "weighted_hour_mean_v2", + }); + expect(v1).not.toContain("LEAST"); + expect(v2).toContain("LEAST(1.0, (s.dm_hour_n)::double precision / 3.0)"); + }); }); diff --git a/src/lib/measurements/daily-series-read.ts b/src/lib/measurements/daily-series-read.ts index ef2b0659e..468921d95 100644 --- a/src/lib/measurements/daily-series-read.ts +++ b/src/lib/measurements/daily-series-read.ts @@ -16,6 +16,10 @@ import { type AggregateGrain, } from "@/lib/measurements/range-aggregation"; import { CUMULATIVE_HK_TYPES } from "@/lib/measurements/apple-health-mapping"; +import { + hourWeightSql, + weightedHourMeanSql, +} from "@/lib/measurements/day-mean"; import { HOURLY_MEAN_DAY_TYPES } from "@/lib/measurements/day-statistic"; import { readTieredRollupSeries } from "@/lib/rollups/measurement-read-wmy"; import { buildSourceRankCase } from "@/lib/analytics/source-rank-sql"; @@ -201,6 +205,11 @@ export async function readLiveBuckets(opts: { const rankRaw = Prisma.raw( buildSourceRankCase(priorityJson, 'p."type"', 'p."source"'), ); + // The day of an hourly-mean type: its hours' means, each hour weighed by the + // active pulse day-mean method (`day-mean.ts`). Code-built SQL, no input. + const dayOfHours = Prisma.raw( + weightedHourMeanSql("h.hour_mean", hourWeightSql("h.n")), + ); const typeFilter = type ? Prisma.sql`AND m."type" = ${type}::measurement_type` : Prisma.empty; @@ -242,20 +251,21 @@ export async function readLiveBuckets(opts: { ${typeFilter} GROUP BY m."type", m."source", 3 ), - -- A day of a type with an activity-dependent sampling rate is the mean of - -- its local hours' means, each hour once, not the mean of its readings: a - -- workout hour holds hundreds of times as many readings as a resting one. + -- A day of a type with an activity-dependent sampling rate is the weighted + -- mean of its local hours' means, not the mean of its readings: a workout + -- hour holds hundreds of times as many readings as a resting one. -- Empty for every other type. hour_mean AS ( SELECT h."type", h."source", h.day, - AVG(h.hour_mean)::double precision AS day_mean + ${dayOfHours}::double precision AS day_mean FROM ( SELECT m."type", m."source", date_trunc('day', (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${timeZone}) AS day, date_trunc('hour', (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${timeZone}) AS hr, - AVG(m."value") AS hour_mean + AVG(m."value") AS hour_mean, + COUNT(*) AS n FROM measurements m WHERE m."user_id" = ${userId} AND m."measured_at" >= ${from} diff --git a/src/lib/measurements/day-mean.ts b/src/lib/measurements/day-mean.ts index 9d90908bc..57ac46c89 100644 --- a/src/lib/measurements/day-mean.ts +++ b/src/lib/measurements/day-mean.ts @@ -2,10 +2,14 @@ * The one place that turns readings into a day value and a window mean for the * types in `HOURLY_MEAN_DAY_TYPES` (see `day-statistic.ts`). * - * For such a type a day is the mean of its hours' means, each hour weighing - * one, and a window of days is the mean of its day values, each day weighing - * one. Count, min and max stay over every reading. Every other type keeps the - * plain mean over its readings, computed exactly as before. + * For such a type a day is the weighted mean of its hours' means and a + * window of days the mean of its day values, each day weighing one. How an + * hour weighs is the pulse day-mean method (`day-statistic.ts`): one under + * `hour_mean_v1`, `min(1, n / 3)` for an hour of `n` readings under + * `weighted_hour_mean_v2`. Every helper applies this process's active method + * unless the caller names one. Count, min and max stay over every reading. + * Every other type keeps the plain mean over its readings, computed exactly as + * before. * * "Day" and "hour" are the reader's own frame: a reader that buckets by the * user's local day passes that zone, a reader that buckets by the session's @@ -15,14 +19,17 @@ * Two shapes are offered, one per kind of reader: * * - TypeScript, for readers that hold the readings in memory: - * `readingsMean`, `dayValue` and `dayValueRows`. + * `readingsMean`, `dayValue` and `dayValueRows`, and `combineHourCells` + * for a reader that already holds per-hour sums. * - SQL, for readers that aggregate in Postgres: `dayWeightedRows` attaches - * a `day_weight` to every reading (1 / (readings in its hour × hours in its - * day) for an hourly-mean type, 1 otherwise), so a day's weights add up to - * one and `SUM(value * day_weight) / SUM(day_weight)` is the mean of the - * day values over any window. `windowMeanSql` and `foldMeanSql` build the - * mean expressions, falling back to the unchanged plain mean for every - * other type. + * a `day_weight` to every reading (its hour's weight shared among the + * hour's readings, over the day's total hour weight, for an hourly-mean + * type; 1 otherwise), so a day's weights add up to one and + * `SUM(value * day_weight) / SUM(day_weight)` is the mean of the day + * values over any window. `windowMeanSql` and `foldMeanSql` build the mean + * expressions, falling back to the unchanged plain mean for every other + * type. A reader that groups by hour itself weighs its hours with + * `hourWeightSql` (or `storedHourWeightSql`) and `weightedHourMeanSql`. * * Type lists are closed code constants checked against an enum-shaped regex * before they are spliced; every value (user id, zone, bounds) stays a bound @@ -31,7 +38,14 @@ import { Prisma } from "@/generated/prisma/client"; import { tzOffsetMinutes } from "@/lib/tz/format"; -import { HOURLY_MEAN_DAY_TYPES, usesHourlyMeanDay } from "./day-statistic"; +import { + activePulseDayMeanMethod, + HOURLY_MEAN_DAY_TYPES, + hourWeight, + type PulseDayMeanMethod, + SPARSE_HOUR_FULL_WEIGHT, + usesHourlyMeanDay, +} from "./day-statistic"; const MS_PER_MINUTE = 60_000; const MS_PER_HOUR = 3_600_000; @@ -66,10 +80,32 @@ function plainMean(rows: readonly DayReading[]): number | null { return rows.reduce((s, r) => s + r.value, 0) / rows.length; } -/** Mean of the per-hour means of readings that share one day. */ +/** + * A day value from its hours' reading sums and counts: the mean of the hours' + * means, each hour weighed by `hourWeight` under `method`. `null` when no hour + * holds a reading. The one TypeScript place the weighting lives; a reader + * that already holds per-hour sums (`daily-stats.ts`) calls it directly. + */ +export function combineHourCells( + hours: Iterable<{ sum: number; n: number }>, + method: PulseDayMeanMethod = activePulseDayMeanMethod(), +): number | null { + let weighted = 0; + let weights = 0; + for (const h of hours) { + if (h.n <= 0) continue; + const w = hourWeight(h.n, method); + weighted += w * (h.sum / h.n); + weights += w; + } + return weights > 0 ? weighted / weights : null; +} + +/** The weighted mean of the per-hour means of readings that share one day. */ function meanOfHourMeans( rows: readonly DayReading[], localMs: (at: Date) => number, + method: PulseDayMeanMethod, ): number { const hours = new Map(); for (const r of rows) { @@ -82,9 +118,7 @@ function meanOfHourMeans( hours.set(hour, { sum: r.value, n: 1 }); } } - let sum = 0; - for (const h of hours.values()) sum += h.sum / h.n; - return sum / hours.size; + return combineHourCells(hours.values(), method) as number; } /** Group readings by local day (days since the epoch in `tz`). */ @@ -110,10 +144,11 @@ export function dayValue( type: string, rows: readonly DayReading[], tz: string, + method: PulseDayMeanMethod = activePulseDayMeanMethod(), ): number | null { if (rows.length === 0) return null; if (!usesHourlyMeanDay(type)) return plainMean(rows); - return meanOfHourMeans(rows, makeLocalMs(tz)); + return meanOfHourMeans(rows, makeLocalMs(tz), method); } /** @@ -126,6 +161,7 @@ export function readingsMean( type: string, rows: readonly DayReading[], tz: string, + method: PulseDayMeanMethod = activePulseDayMeanMethod(), ): number | null { if (rows.length === 0) return null; if (!usesHourlyMeanDay(type)) return plainMean(rows); @@ -133,43 +169,12 @@ export function readingsMean( let sum = 0; let days = 0; for (const dayRows of groupByLocalDay(rows, localMs).values()) { - sum += meanOfHourMeans(dayRows, localMs); + sum += meanOfHourMeans(dayRows, localMs, method); days += 1; } return sum / days; } -/** - * Readings of an hourly-mean type replaced by one row per local hour carrying - * the hour's mean, stamped at the hour's first reading. A reader that already - * folds its rows into day buckets by plain mean (and a window by mean of - * those) then yields the hourly-mean day without knowing the statistic. Every - * other type comes back untouched (the same array). - */ -export function hourMeanRows( - type: string, - rows: readonly T[], - tz: string, -): readonly DayReading[] { - if (!usesHourlyMeanDay(type) || rows.length === 0) return rows; - const localMs = makeLocalMs(tz); - const hours = new Map(); - for (const r of rows) { - const hour = Math.floor(localMs(r.measuredAt) / MS_PER_HOUR); - const acc = hours.get(hour); - if (acc) { - acc.sum += r.value; - acc.n += 1; - if (r.measuredAt < acc.at) acc.at = r.measuredAt; - } else { - hours.set(hour, { sum: r.value, n: 1, at: r.measuredAt }); - } - } - return [...hours.values()] - .map((h) => ({ value: h.sum / h.n, measuredAt: h.at })) - .sort((a, b) => a.measuredAt.getTime() - b.measuredAt.getTime()); -} - /** * Readings of an hourly-mean type replaced by one row per local day carrying * the day value, stamped at the day's first reading; a reader that folds rows @@ -180,6 +185,7 @@ export function dayValueRows( type: string, rows: readonly T[], tz: string, + method: PulseDayMeanMethod = activePulseDayMeanMethod(), ): readonly DayReading[] { if (!usesHourlyMeanDay(type) || rows.length === 0) return rows; const localMs = makeLocalMs(tz); @@ -187,7 +193,10 @@ export function dayValueRows( for (const dayRows of groupByLocalDay(rows, localMs).values()) { let first = dayRows[0].measuredAt; for (const r of dayRows) if (r.measuredAt < first) first = r.measuredAt; - out.push({ value: meanOfHourMeans(dayRows, localMs), measuredAt: first }); + out.push({ + value: meanOfHourMeans(dayRows, localMs, method), + measuredAt: first, + }); } return out.sort((a, b) => a.measuredAt.getTime() - b.measuredAt.getTime()); } @@ -237,10 +246,55 @@ function frameTs(frame: DayFrame, column: string): string { : `((${column} AT TIME ZONE 'UTC') AT TIME ZONE ${frame.tzSql})`; } +/** A column or expression of the caller's own query, never input. */ +const SQL_EXPR_RE = /^[A-Za-z0-9_."() ]+$/; + +/** + * SQL weight of an hour holding `n` readings under `method` (see + * `hourWeight`): `1.0` for `hour_mean_v1`, `LEAST(1, n / 3)` for + * `weighted_hour_mean_v2`. `n` is a column of the caller's query. + */ +export function hourWeightSql( + n: string, + method: PulseDayMeanMethod = activePulseDayMeanMethod(), +): string { + if (!SQL_EXPR_RE.test(n)) { + throw new Error(`hour count must be a column expression: ${n}`); + } + return method === "weighted_hour_mean_v2" + ? `LEAST(1.0, (${n})::double precision / ${SPARSE_HOUR_FULL_WEIGHT}.0)` + : "1.0"; +} + +/** + * `hourWeightSql` under the method stored in `app_settings`, read inside the + * statement itself. For a writer whose output persists (the rollup tier): it + * must follow the setting the moment the apply step changes it, not this + * process's mirror of it, which trails by up to a minute. + */ +export function storedHourWeightSql(n: string): string { + return `(CASE WHEN (SELECT st."pulse_day_mean_method" FROM "app_settings" st WHERE st."id" = 'singleton') = 'weighted_hour_mean_v2' + THEN ${hourWeightSql(n, "weighted_hour_mean_v2")} + ELSE 1.0 END)`; +} + +/** + * Aggregate over per-hour rows: the day value as the weighted mean of the + * hours' means, `SUM(hourMean * weight) / SUM(weight)`. `weight` comes from + * `hourWeightSql` or `storedHourWeightSql`. + */ +export function weightedHourMeanSql(hourMean: string, weight: string): string { + if (!SQL_EXPR_RE.test(hourMean)) { + throw new Error(`hour mean must be a column expression: ${hourMean}`); + } + return `(SUM((${hourMean}) * ${weight}) / NULLIF(SUM(${weight}), 0))`; +} + function weightedRowsSql( source: string, frame: DayFrame, bySource: boolean, + method: PulseDayMeanMethod, ): string { if (!/^[a-z_][a-z0-9_]*$/.test(source)) { throw new Error(`source must be a CTE name: ${source}`); @@ -251,8 +305,11 @@ function weightedRowsSql( // join as a nested loop and a year of pulse ran into the statement timeout. // // `dm_hour_n` counts the readings in a row's hour and `dm_hour_first` marks - // one row per hour, so the sum of `dm_hour_first` over a day is the day's - // hour count. Both columns ride along on the row; every caller aggregates. + // one row per hour, so the sum of `dm_hour_first * w` over a day is the + // day's total hour weight (its hour count under `hour_mean_v1`). A row's + // weight is its hour's weight shared among the hour's readings, divided by + // that total, so a day's weights still add up to one. Both columns ride + // along on the row; every caller aggregates. const partition = (alias: string, unit: "hour" | "day") => { const keys = bySource ? `${alias}."type", ${alias}."source"` @@ -263,7 +320,7 @@ function weightedRowsSql( return `( SELECT s.*, (CASE WHEN ${isHourlyMeanTypeSql('s."type"')} - THEN 1.0 / (s.dm_hour_n * SUM(s.dm_hour_first) OVER (${partition("s", "day")})) + THEN ${hourWeightSql("s.dm_hour_n", method)} / (s.dm_hour_n * SUM(s.dm_hour_first * ${hourWeightSql("s.dm_hour_n", method)}) OVER (${partition("s", "day")})) ELSE 1.0 END)::double precision AS day_weight FROM ( SELECT t.*, @@ -286,9 +343,14 @@ function weightedRowsSql( export function dayWeightedRows( source: string, frame: DayFrame, - opts: { bySource?: boolean } = {}, + opts: { bySource?: boolean; method?: PulseDayMeanMethod } = {}, ): string { - return weightedRowsSql(source, frame, opts.bySource === true); + return weightedRowsSql( + source, + frame, + opts.bySource === true, + opts.method ?? activePulseDayMeanMethod(), + ); } /** @@ -298,11 +360,17 @@ export function dayWeightedRows( export function dayWeightedRowsSql( source: string, tz: string | null, - opts: { bySource?: boolean } = {}, + opts: { bySource?: boolean; method?: PulseDayMeanMethod } = {}, ): Prisma.Sql { + const method = opts.method ?? activePulseDayMeanMethod(); if (tz === null) { return Prisma.raw( - weightedRowsSql(source, SESSION_DAY_FRAME, opts.bySource === true), + weightedRowsSql( + source, + SESSION_DAY_FRAME, + opts.bySource === true, + method, + ), ); } // Split around a placeholder so the zone binds as a parameter. @@ -311,6 +379,7 @@ export function dayWeightedRowsSql( source, { kind: "zone", tzSql: marker }, opts.bySource === true, + method, ).split(marker); let sql = Prisma.raw(pieces[0]); for (const piece of pieces.slice(1)) { diff --git a/src/lib/measurements/day-statistic.ts b/src/lib/measurements/day-statistic.ts index 2b42cf4f0..2baa12799 100644 --- a/src/lib/measurements/day-statistic.ts +++ b/src/lib/measurements/day-statistic.ts @@ -30,3 +30,73 @@ export type WindowWeighting = "count" | "day"; export function windowWeighting(type: string): WindowWeighting { return usesHourlyMeanDay(type) ? "day" : "count"; } + +// ─── Pulse day-mean method (#1107 follow-up) ────────────── + +/** + * How the hours of a pulse day weigh against each other. + * + * - `hour_mean_v1`: every hour that holds a reading weighs one, the + * statistic up to v1.42. One stray reading in an otherwise empty hour + * then counts as much as an hour of steady sampling. + * - `weighted_hour_mean_v2`: an hour with `n` readings weighs + * `min(1, n / SPARSE_HOUR_FULL_WEIGHT)`, so an hour reaches full weight + * at three readings and a lone reading weighs a third. A dense hour still + * weighs one, so the workout bias the hourly mean removed stays removed. + * + * The operator switches it once, after a dry run, from the admin console; + * `app_settings.pulse_day_mean_method` holds it. + */ +export const PULSE_DAY_MEAN_METHODS = [ + "hour_mean_v1", + "weighted_hour_mean_v2", +] as const; +export type PulseDayMeanMethod = (typeof PULSE_DAY_MEAN_METHODS)[number]; + +export const DEFAULT_PULSE_DAY_MEAN_METHOD: PulseDayMeanMethod = "hour_mean_v1"; + +/** Readings at which an hour reaches full weight under `weighted_hour_mean_v2`. */ +export const SPARSE_HOUR_FULL_WEIGHT = 3; + +export function isPulseDayMeanMethod( + value: unknown, +): value is PulseDayMeanMethod { + return ( + typeof value === "string" && + (PULSE_DAY_MEAN_METHODS as readonly string[]).includes(value) + ); +} + +/** The weight of an hour holding `n` readings under `method`. */ +export function hourWeight(n: number, method: PulseDayMeanMethod): number { + if (method === "weighted_hour_mean_v2") { + return Math.min(1, n / SPARSE_HOUR_FULL_WEIGHT); + } + return 1; +} + +/** + * The method this process applies. Every reader computes synchronously, so + * the stored setting is mirrored here: loaded at boot and refreshed every + * minute (`pulse-day-mean-method.ts`), set at once by the apply step in the + * process that ran it. Pinned on `globalThis` because the boot hook and the + * route handlers are separate bundles of one process. The default until the + * first load is the v1.42 statistic, which is also what the column defaults + * to. + */ +const METHOD_SYMBOL = Symbol.for("healthlog.pulseDayMeanMethod"); + +type GlobalWithMethod = typeof globalThis & { + [METHOD_SYMBOL]?: PulseDayMeanMethod; +}; + +export function activePulseDayMeanMethod(): PulseDayMeanMethod { + return ( + (globalThis as GlobalWithMethod)[METHOD_SYMBOL] ?? + DEFAULT_PULSE_DAY_MEAN_METHOD + ); +} + +export function setActivePulseDayMeanMethod(method: PulseDayMeanMethod): void { + (globalThis as GlobalWithMethod)[METHOD_SYMBOL] = method; +} diff --git a/src/lib/measurements/pulse-day-mean-method.ts b/src/lib/measurements/pulse-day-mean-method.ts new file mode 100644 index 000000000..91be70cbc --- /dev/null +++ b/src/lib/measurements/pulse-day-mean-method.ts @@ -0,0 +1,70 @@ +/** + * Keeps this process's mirror of the pulse day-mean method in step with + * `app_settings.pulse_day_mean_method` (#1107 follow-up). + * + * Every pulse reader computes synchronously through `day-mean.ts`, so the + * method cannot be read per call; it is mirrored instead + * (`activePulseDayMeanMethod`). `instrumentation.ts` loads it once at boot and + * refreshes it every minute in every process, web and worker alike. The apply + * step sets it at once in the process that ran it; any other process follows + * within a minute. The rollup tier does not use the mirror: it reads the + * setting inside its own statement (`storedHourWeightSql`). + * + * A failed read keeps the mirror as it is: the setting changes once, by hand, + * and a database blip must not flip a process back to the old statistic. + */ +import type { Prisma } from "@/generated/prisma/client"; +import { prisma } from "@/lib/db"; + +import { + DEFAULT_PULSE_DAY_MEAN_METHOD, + isPulseDayMeanMethod, + setActivePulseDayMeanMethod, + type PulseDayMeanMethod, +} from "./day-statistic"; + +/** How often a process re-reads the stored method. */ +const REFRESH_MS = 60_000; + +const TIMER_SYMBOL = Symbol.for("healthlog.pulseDayMeanMethodTimer"); + +type GlobalWithTimer = typeof globalThis & { + [TIMER_SYMBOL]?: ReturnType; +}; + +type SettingsReader = { + appSettings: Pick; +}; + +/** The stored method; the column default when there is no settings row. */ +export async function loadPulseDayMeanMethod( + db: SettingsReader = prisma, +): Promise { + const row = await db.appSettings.findUnique({ + where: { id: "singleton" }, + select: { pulseDayMeanMethod: true }, + }); + const stored = row?.pulseDayMeanMethod; + return isPulseDayMeanMethod(stored) ? stored : DEFAULT_PULSE_DAY_MEAN_METHOD; +} + +async function refresh(): Promise { + try { + setActivePulseDayMeanMethod(await loadPulseDayMeanMethod()); + } catch { + // Keep the mirror; the next tick tries again. + } +} + +/** + * Load the stored method now and keep re-reading it. Idempotent per process; + * the timer does not hold the process open. + */ +export async function startPulseDayMeanMethodSync(): Promise { + await refresh(); + const g = globalThis as GlobalWithTimer; + if (g[TIMER_SYMBOL]) return; + const timer = setInterval(() => void refresh(), REFRESH_MS); + timer.unref?.(); + g[TIMER_SYMBOL] = timer; +} diff --git a/src/lib/openapi/error-codes.ts b/src/lib/openapi/error-codes.ts index 8560c5a3f..83045e37e 100644 --- a/src/lib/openapi/error-codes.ts +++ b/src/lib/openapi/error-codes.ts @@ -256,6 +256,10 @@ export const ERROR_CODE_CATALOGUE: Readonly> = "managed_profile.handover.unknown_guardian", "managed_profile.not_found", ], + maintenance: [ + "maintenance.pulseDayMean.busy", + "maintenance.pulseDayMean.dryRunRequired", + ], measurement: [ "measurement.batch.invalid", "measurement.batch.source_not_permitted", diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index 8f6643007..18b335740 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -43,6 +43,10 @@ import { utcBucketEnd, utcBucketStart, } from "@/lib/rollups/measurement-read"; +import { + storedHourWeightSql, + weightedHourMeanSql, +} from "@/lib/measurements/day-mean"; import { HOURLY_MEAN_DAY_TYPES } from "@/lib/measurements/day-statistic"; import { startOfUtcDay } from "@/lib/tz/start-of-utc-day"; import type { @@ -588,16 +592,17 @@ async function runRollupAggregate( /** * For a type whose DAY value is the mean of its per-hour means (see - * `day-statistic.ts`), replace the bucket's `mean`: a DAY bucket's is the mean - * of its hours' means, a WEEK / MONTH / YEAR bucket's the mean of its days' - * values, each day once. The coarse rows are not read for their mean (every + * `day-statistic.ts`), replace the bucket's `mean`: a DAY bucket's is the + * weighted mean of its hours' means under the stored pulse day-mean method, + * a WEEK / MONTH / YEAR bucket's the mean of its days' values, each day once. The coarse rows are not read for their mean (every * coarse read folds the DAY tier, `readCanonicalRollupBuckets`), but a stored * row should not contradict the tier it is folded from. Only `mean` changes: * `count`, `min`, `max` and the sums the regression composes from stay over * every reading, so the band and the windowed slope are untouched. * * Hours and days are UTC, matching the UTC day this tier buckets by; migration - * 0372 applies the same statistic to rows stored before it. + * 0372 applies the same statistic to rows stored before it, and the admin + * apply step of the day-mean method rebuilds the PULSE rows it changes. */ async function applyHourlyMeanDay( db: Pick, @@ -627,6 +632,12 @@ async function applyHourlyMeanDay( } } const typeList = types.map((t) => `'${t}'::"measurement_type"`).join(","); + // The stored method, read inside the statement: a rollup outlives this + // process's mirror of the setting, so it follows the setting itself. + const dayOfHours = weightedHourMeanSql( + "h.hour_mean", + storedHourWeightSql("h.n"), + ); const means = await db.$queryRawUnsafe< Array<{ type: string; source: string; bucket_start: Date; mean: number }> >( @@ -638,11 +649,12 @@ async function applyHourlyMeanDay( FROM ( SELECT h."type", h."source", date_trunc('day', h.hr) AS day, - AVG(h.hour_mean) AS day_mean + ${dayOfHours} AS day_mean FROM ( SELECT m."type", m."source", date_trunc('hour', m."measured_at") AS hr, - AVG(m."value") AS hour_mean + AVG(m."value") AS hour_mean, + COUNT(*) AS n FROM measurements m WHERE m."user_id" = $1 AND m."type" IN (${typeList}) diff --git a/src/lib/validations/pulse-day-mean.ts b/src/lib/validations/pulse-day-mean.ts index 39a837534..170930d43 100644 --- a/src/lib/validations/pulse-day-mean.ts +++ b/src/lib/validations/pulse-day-mean.ts @@ -1,14 +1,58 @@ /** * v1.43 (#1107 follow-up) — the admin pulse day-mean dry run and apply step. + * + * The status shape is the admin console's own and stays out of the published + * API description: the admin routes are not part of the public contract. */ import { z } from "zod/v4"; -export const PULSE_DAY_MEAN_METHODS = [ - "hour_mean_v1", - "weighted_hour_mean_v2", -] as const; +import type { PulseDayMeanMethod } from "@/lib/measurements/day-statistic"; export const pulseDayMeanActionSchema = z.strictObject({ action: z.enum(["dry_run", "apply"]), }); export type PulseDayMeanActionInput = z.infer; + +/** Where a day's change between the two methods falls, in bpm. */ +export const PULSE_DAY_MEAN_CHANGE_BUCKETS = ["0-1", "1-2", "2-5", "5+"] as const; + +/** + * What a dry run found. Accounts are numbered in the order they were created + * (`"#1"`, `"#2"`, …), never named or identified. + */ +const dryRunResultSchema = z.object({ + accounts: z.number().int().nonnegative(), + daysCompared: z.number().int().nonnegative(), + daysChangedOver1Bpm: z.number().int().nonnegative(), + maxChangeBpm: z.number().nonnegative(), + distribution: z.array( + z.object({ + bucket: z.enum(PULSE_DAY_MEAN_CHANGE_BUCKETS), + days: z.number().int().nonnegative(), + }), + ), + perAccount: z.array( + z.object({ + account: z.string(), + days: z.number().int().nonnegative(), + changedOver1Bpm: z.number().int().nonnegative(), + }), + ), +}); +export type PulseDayMeanDryRunResult = z.infer; + +/** `app_settings.pulse_day_mean_dry_run_json`: the last dry run. */ +export const pulseDayMeanDryRunSchema = z.object({ + status: z.enum(["queued", "running", "done", "failed"]), + startedAt: z.string(), + finishedAt: z.string().nullable(), + result: dryRunResultSchema.nullable(), +}); +export type PulseDayMeanDryRun = z.infer; + +/** `GET /api/admin/maintenance/pulse-day-mean`. */ +export interface PulseDayMeanStatus { + method: PulseDayMeanMethod; + appliedAt: string | null; + lastDryRun: PulseDayMeanDryRun | null; +}