From 60c8a76b2c9330a39930174bbe4fb89841403087 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 21 Jun 2026 10:31:27 +0200 Subject: [PATCH] feat(measurements): fold out-of-window SpO2 to the dense intra-day tier MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Apple Watch Blood-Oxygen samples OXYGEN_SATURATION through the day and overnight. The readings map to latest aggregation and sit in neither the cumulative SUM drain nor the daily-mean drain, so every sample piled up raw forever. Fold out-of-window per-sample SpO2 rows to one daily-mean stats: row, soft-delete the raw rows, and preserve the daily min/max on the rollup DAY bucket via a pre-fold recompute — the lowest sample is the overnight desaturation nadir, the clinically meaningful figure the mean would hide. Generalise the PULSE min/max preservation to a min/max-preserving set so the pre-fold recompute and the skipped post-fold recompute cover both PULSE and SpO2; resting-HR derivation stays PULSE-only. RESPIRATORY_RATE is left to the existing daily-mean drain it already belongs to. --- .../dense-intraday-drain-exemption.test.ts | 14 +- .../dense-intraday-spo2-summary.test.ts | 257 ++++++++++++++++++ .../measurements/dense-intraday-retention.ts | 83 ++++-- 3 files changed, 332 insertions(+), 22 deletions(-) create mode 100644 src/lib/measurements/__tests__/dense-intraday-spo2-summary.test.ts diff --git a/src/lib/measurements/__tests__/dense-intraday-drain-exemption.test.ts b/src/lib/measurements/__tests__/dense-intraday-drain-exemption.test.ts index 1f9d82faa..fb0311aa7 100644 --- a/src/lib/measurements/__tests__/dense-intraday-drain-exemption.test.ts +++ b/src/lib/measurements/__tests__/dense-intraday-drain-exemption.test.ts @@ -47,6 +47,16 @@ describe("dense intra-day retention — drain exemption (WX-E)", () => { ); }); + it("EXEMPTS OXYGEN_SATURATION from the destructive daily-mean drain", () => { + // SpO2 is folded ONLY by the dense tier (bounded window, min/max + // preserved). If it were added to HIGH_FREQUENCY_MEAN_TYPES too, both + // drains would fold it and the overnight-desaturation nadir (the daily + // MIN) would be lost the next night the mean drain ran. + expect( + HIGH_FREQUENCY_MEAN_TYPES.has("OXYGEN_SATURATION" as MeasurementType), + ).toBe(false); + }); + it("EXEMPTS both dense-tier types from the cumulative (SUM) drain too", () => { // HRV / HR are spot signals, never cumulative — they must not be in the // SUM drain either, which would also collapse + hard-delete them. @@ -55,9 +65,9 @@ describe("dense intra-day retention — drain exemption (WX-E)", () => { } }); - it("scopes the dense tier to exactly HEART_RATE_VARIABILITY + PULSE", () => { + it("scopes the dense tier to exactly HEART_RATE_VARIABILITY + PULSE + OXYGEN_SATURATION", () => { expect(Array.from(DENSE_INTRADAY_RETENTION_TYPES).sort()).toEqual( - ["HEART_RATE_VARIABILITY", "PULSE"].sort(), + ["HEART_RATE_VARIABILITY", "PULSE", "OXYGEN_SATURATION"].sort(), ); }); diff --git a/src/lib/measurements/__tests__/dense-intraday-spo2-summary.test.ts b/src/lib/measurements/__tests__/dense-intraday-spo2-summary.test.ts new file mode 100644 index 000000000..5ce32dbb8 --- /dev/null +++ b/src/lib/measurements/__tests__/dense-intraday-spo2-summary.test.ts @@ -0,0 +1,257 @@ +/** + * v1.19.0 — server-side consolidation of high-frequency blood oxygen (SpO2). + * + * Apple Watch's Blood-Oxygen app samples OXYGEN_SATURATION periodically + * through the day and overnight; the readings map to `aggregation: "latest"` + * and belong to NEITHER `CUMULATIVE_HK_TYPES` nor `HIGH_FREQUENCY_MEAN_TYPES`, + * so every sample piled up raw forever. The dense intra-day retention drain + * folds the out-of-window per-sample rows to one daily-MEAN `stats:` row, + * mirroring the PULSE facet's min/max preservation: the daily MIN is the + * overnight-desaturation nadir — the single most clinically meaningful SpO2 + * figure — so the DAY rollup bucket is recomputed from the RAW rows BEFORE the + * fold and the post-fold recompute that would collapse min == max == mean is + * skipped. + * + * This suite pins the SpO2 facet: the pre-fold (and only) DAY-rollup recompute, + * the MEAN fold + soft-delete, the no-derived-resting-row guard (resting HR is + * a PULSE-only concern), idempotency, and the per-day failure boundary. + */ +import { describe, it, expect, vi, beforeEach } from "vitest"; + +const { recomputeBucketsForMeasurement } = vi.hoisted(() => ({ + recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined), +})); +vi.mock("@/lib/rollups/measurement-rollups", () => ({ + recomputeBucketsForMeasurement, +})); + +import { runDenseIntradayRetention } from "../dense-intraday-retention"; +import type { PerSampleRow } from "../drain-per-sample-cumulative"; +import type { MeasurementType, PrismaClient } from "@/generated/prisma/client"; + +function spo2Row(id: string, value: number, iso: string): PerSampleRow { + return { + id, + type: "OXYGEN_SATURATION" as MeasurementType, + value, + measuredAt: new Date(iso), + externalId: null, + unit: "%", + }; +} + +/** + * Prisma mock covering the SpO2 fold flow: the scan (per-type rows), the fold + * transaction (`create`/`update`/`findFirst`/`updateMany`), and the top-level + * surfaces a PULSE day would touch (`findFirst` for the native-resting probe, + * `upsert` for the derived resting row) — both must stay UNCALLED for SpO2. + */ +function buildPrismaMock(opts: { spo2Rows: PerSampleRow[] }) { + const txCreate = vi.fn().mockResolvedValue({ id: "minted-daily" }); + const txUpdate = vi.fn().mockResolvedValue({}); + const txFindFirst = vi.fn().mockResolvedValue(null); + const txUpdateMany = vi + .fn() + .mockResolvedValue({ count: opts.spo2Rows.length }); + const tx = { + measurement: { + create: txCreate, + update: txUpdate, + findFirst: txFindFirst, + updateMany: txUpdateMany, + }, + }; + + const findMany = vi.fn(async (args: { where: { type: string } }) => + args.where.type === "OXYGEN_SATURATION" ? opts.spo2Rows : [], + ); + // Native-resting probe + derived-resting upsert: PULSE-only surfaces. + const topFindFirst = vi.fn().mockResolvedValue(null); + const upsert = vi.fn().mockResolvedValue({ id: "derived-rhr" }); + + const mock = { + user: { + findMany: vi + .fn() + .mockResolvedValue([{ id: "user-1", timezone: "Europe/Berlin" }]), + }, + measurement: { findMany, findFirst: topFindFirst, upsert }, + $transaction: vi.fn(async (cb: (t: unknown) => Promise) => cb(tx)), + } as unknown as PrismaClient; + + return { mock, txCreate, txUpdate, txUpdateMany, upsert, topFindFirst }; +} + +beforeEach(() => { + recomputeBucketsForMeasurement.mockClear(); +}); + +describe("SpO2 fold — MEAN collapse + soft-delete", () => { + it("folds the day to one MEAN row (not the latest, not the min) and soft-deletes the raw rows", async () => { + // Overnight dip to 91, daytime band 96..98. Mean ~95.2, NOT the 91 nadir. + const rows = [ + spo2Row("a", 91, "2026-05-01T03:00:00.000Z"), + spo2Row("b", 96, "2026-05-01T09:00:00.000Z"), + spo2Row("c", 97, "2026-05-01T13:00:00.000Z"), + spo2Row("d", 98, "2026-05-01T19:00:00.000Z"), + ]; + const { mock, txCreate, txUpdateMany } = buildPrismaMock({ + spo2Rows: rows, + }); + + const summary = await runDenseIntradayRetention(mock, { + retentionDays: 0, + log: () => {}, + }); + + const createArg = txCreate.mock.calls[0]?.[0] as { + data: { value: number; type: string; source: string; unit: string }; + }; + expect(createArg.data.type).toBe("OXYGEN_SATURATION"); + expect(createArg.data.source).toBe("APPLE_HEALTH"); + expect(createArg.data.unit).toBe("%"); + expect(createArg.data.value).toBeCloseTo((91 + 96 + 97 + 98) / 4, 6); + + // Soft-delete, never hard delete. + const updArg = txUpdateMany.mock.calls[0]?.[0] as { + data: { deletedAt: Date }; + }; + expect(updArg.data.deletedAt).toBeInstanceOf(Date); + expect(summary.totals.daysConsolidated).toBe(1); + expect(summary.totals.perSampleRowsSoftDeleted).toBe(rows.length); + }); +}); + +describe("SpO2 fold — min/max preservation (overnight desat nadir)", () => { + it("recomputes the SpO2 DAY rollup from RAW rows BEFORE the fold, and NOT after", async () => { + const rows = [ + spo2Row("a", 90, "2026-05-01T03:00:00.000Z"), + spo2Row("b", 96, "2026-05-01T12:00:00.000Z"), + spo2Row("c", 99, "2026-05-01T18:00:00.000Z"), + ]; + const { mock, txCreate } = buildPrismaMock({ spo2Rows: rows }); + + await runDenseIntradayRetention(mock, { retentionDays: 0, log: () => {} }); + + expect(txCreate).toHaveBeenCalledTimes(1); + + // EXACTLY one SpO2 recompute — the pre-fold capture. A second (post-fold) + // recompute would aggregate the single mean row and collapse the daily MIN + // (the overnight-desaturation nadir) into the mean. + const spo2Recomputes = recomputeBucketsForMeasurement.mock.calls.filter( + (c) => c[1] === "OXYGEN_SATURATION", + ); + expect(spo2Recomputes).toHaveLength(1); + }); +}); + +describe("SpO2 fold — no resting derivation (PULSE-only concern)", () => { + it("never probes native resting nor mints a derived RESTING_HEART_RATE row", async () => { + const rows = [ + spo2Row("a", 95, "2026-05-01T03:00:00.000Z"), + spo2Row("b", 96, "2026-05-01T09:00:00.000Z"), + spo2Row("c", 97, "2026-05-01T19:00:00.000Z"), + ]; + const { mock, upsert, topFindFirst } = buildPrismaMock({ spo2Rows: rows }); + + const summary = await runDenseIntradayRetention(mock, { + retentionDays: 0, + log: () => {}, + }); + + expect(upsert).not.toHaveBeenCalled(); + expect(topFindFirst).not.toHaveBeenCalled(); + expect(summary.totals.derivedRestingRowsUpserted).toBe(0); + }); +}); + +describe("SpO2 fold — idempotency + dry-run + no double-collapse", () => { + it("converges to zero work when the scan returns no live rows (re-run)", async () => { + const { mock, txCreate, txUpdateMany } = buildPrismaMock({ spo2Rows: [] }); + const summary = await runDenseIntradayRetention(mock, { + retentionDays: 0, + log: () => {}, + }); + expect(txCreate).not.toHaveBeenCalled(); + expect(txUpdateMany).not.toHaveBeenCalled(); + expect(summary.totals.daysConsolidated).toBe(0); + expect(recomputeBucketsForMeasurement).not.toHaveBeenCalled(); + }); + + it("writes nothing on a dry-run (no fold, no pre-fold recompute)", async () => { + const rows = [ + spo2Row("a", 95, "2026-05-01T03:00:00.000Z"), + spo2Row("b", 96, "2026-05-01T09:00:00.000Z"), + spo2Row("c", 97, "2026-05-01T19:00:00.000Z"), + ]; + const { mock, txCreate, upsert } = buildPrismaMock({ spo2Rows: rows }); + await runDenseIntradayRetention(mock, { + retentionDays: 0, + dryRun: true, + log: () => {}, + }); + expect(txCreate).not.toHaveBeenCalled(); + expect(upsert).not.toHaveBeenCalled(); + expect(recomputeBucketsForMeasurement).not.toHaveBeenCalled(); + }); + + it("does not double-collapse already-folded `stats:` rows (skipped by the bucketing predicate)", async () => { + // A previously-folded daily row carrying the `stats:` externalId is the + // only "row" left on a re-run; the bucketing skips it, so nothing folds. + const folded: PerSampleRow = { + id: "folded", + type: "OXYGEN_SATURATION" as MeasurementType, + value: 95, + measuredAt: new Date("2026-05-01T12:00:00.000Z"), + externalId: "stats:HKQuantityTypeIdentifierOxygenSaturation:2026-05-01", + unit: "%", + }; + const { mock, txCreate } = buildPrismaMock({ spo2Rows: [folded] }); + const summary = await runDenseIntradayRetention(mock, { + retentionDays: 0, + log: () => {}, + }); + expect(txCreate).not.toHaveBeenCalled(); + expect(summary.totals.daysConsolidated).toBe(0); + }); +}); + +describe("SpO2 fold — per-day failure boundary", () => { + it("steps over a day whose fold throws and keeps draining the rest", async () => { + const day1 = [ + spo2Row("a", 95, "2026-05-01T03:00:00.000Z"), + spo2Row("b", 96, "2026-05-01T12:00:00.000Z"), + ]; + const day2 = [ + spo2Row("c", 94, "2026-05-02T03:00:00.000Z"), + spo2Row("d", 97, "2026-05-02T12:00:00.000Z"), + ]; + const { mock } = buildPrismaMock({ spo2Rows: [...day1, ...day2] }); + + // First day's fold transaction throws a non-P2002 error; the boundary + // must absorb it and continue to the second day. + let call = 0; + (mock.$transaction as ReturnType).mockImplementation( + async (cb: (t: unknown) => Promise) => { + call += 1; + if (call === 1) throw new Error("boom — poisoned day"); + return cb({ + measurement: { + create: vi.fn().mockResolvedValue({ id: "minted" }), + update: vi.fn(), + findFirst: vi.fn().mockResolvedValue(null), + updateMany: vi.fn().mockResolvedValue({ count: 2 }), + }, + }); + }, + ); + + const summary = await runDenseIntradayRetention(mock, { + retentionDays: 0, + log: () => {}, + }); + + // The second day still folded despite the first day's failure. + expect(summary.totals.daysConsolidated).toBe(1); + }); +}); diff --git a/src/lib/measurements/dense-intraday-retention.ts b/src/lib/measurements/dense-intraday-retention.ts index b8934410c..dfbdd6802 100644 --- a/src/lib/measurements/dense-intraday-retention.ts +++ b/src/lib/measurements/dense-intraday-retention.ts @@ -99,14 +99,54 @@ import type { PerSampleRow } from "./drain-per-sample-cumulative"; */ /** - * The dense-tier types. These are the daytime intra-day signals the Stress - * engine reads as a shape. Both are deliberately EXEMPT from the - * destructive daily-mean drain (`HIGH_FREQUENCY_MEAN_TYPES`); the - * disjointness from that set is the load-bearing invariant a regression - * test pins. + * The dense-tier types. These are the high-frequency intra-day signals that + * accumulate raw per-sample rows the destructive daily-mean drain never + * touches: + * + * - `HEART_RATE_VARIABILITY` + `PULSE` — the daytime shape the Stress + * engine reads (v1.10.0 / v1.18.11). + * - `OXYGEN_SATURATION` — Apple Watch's Blood-Oxygen app samples SpO2 + * periodically through the day and overnight; the readings map to + * `aggregation: "latest"` and sit in NEITHER `CUMULATIVE_HK_TYPES` nor + * `HIGH_FREQUENCY_MEAN_TYPES`, so every sample piled up raw forever. + * Folded here to one daily-mean row with the daily min/max preserved on + * the rollup tier (the lowest sample is the overnight-desaturation + * nadir — clinically the meaningful figure, not the mean). + * + * Every member is deliberately EXEMPT from the destructive daily-mean drain + * (`HIGH_FREQUENCY_MEAN_TYPES`); the disjointness from that set is the + * load-bearing invariant a regression test pins — a type in both would be + * folded by two drains at once and corrupt the value. + * + * `RESPIRATORY_RATE` is deliberately NOT here: it is already a member of + * `HIGH_FREQUENCY_MEAN_TYPES`, so the nightly mean-consolidation already + * collapses it to a daily-mean row. Adding it here would double-fold it. */ export const DENSE_INTRADAY_RETENTION_TYPES: ReadonlySet = - new Set(["HEART_RATE_VARIABILITY", "PULSE"]); + new Set([ + "HEART_RATE_VARIABILITY", + "PULSE", + "OXYGEN_SATURATION", + ]); + +/** + * Dense-tier types whose daily MIN / MAX are clinically meaningful and must + * survive the fold. For these, the DAY rollup bucket is recomputed from the + * RAW rows BEFORE the fold (while min/max are still computable from live + * rows) and the post-fold recompute that would collapse min == max == mean is + * SKIPPED — identical to the PULSE facet (iOS#34), generalised: + * + * - `PULSE` — daily min/max bound the resting floor + workout peak. + * - `OXYGEN_SATURATION` — the daily MIN is the overnight-desaturation + * nadir, the single most clinically meaningful SpO2 figure; the mean + * alone would hide a transient nocturnal dip. + * + * `HEART_RATE_VARIABILITY` is absent: out-of-window HRV has no min/max + * consumer (the Stress engine reads the IN-window shape this drain never + * folds), so it keeps the simpler post-fold recompute against its mean row. + */ +const DENSE_MIN_MAX_PRESERVING_TYPES: ReadonlySet = + new Set(["PULSE", "OXYGEN_SATURATION"]); /** * Retention bound (days). Per-sample HRV / HR rows older than this window @@ -302,15 +342,18 @@ export async function runDenseIntradayRetention( }): Promise => { const unit = dayRows[0]?.unit ?? "unknown"; const isPulse = type === "PULSE"; + const preservesMinMax = DENSE_MIN_MAX_PRESERVING_TYPES.has(type); - // iOS#34 — PULSE min/max preservation. The persistent `MeasurementRollup` - // DAY bucket stores min/max/mean but aggregates LIVE rows; recomputing it - // AFTER the fold (one mean row) collapses min == max == mean. So for - // PULSE, recompute the DAY bucket from the RAW rows BEFORE the fold, - // while they are still live — the bucket then keeps the TRUE intraday - // min/max/mean — and the post-fold recompute is SKIPPED below. (HRV has - // no min/max consumer, so it keeps the simpler post-fold recompute.) - if (isPulse && !options.dryRun) { + // iOS#34 — min/max preservation (PULSE + SpO2). The persistent + // `MeasurementRollup` DAY bucket stores min/max/mean but aggregates LIVE + // rows; recomputing it AFTER the fold (one mean row) collapses + // min == max == mean. So for the min/max-preserving types, recompute the + // DAY bucket from the RAW rows BEFORE the fold, while they are still live + // — the bucket then keeps the TRUE intraday min/max/mean (for SpO2 the + // MIN is the overnight-desaturation nadir) — and the post-fold recompute + // is SKIPPED below. HRV has no min/max consumer, so it keeps the simpler + // post-fold recompute. + if (preservesMinMax && !options.dryRun) { await recomputeBucketsForMeasurement(userId, type, canonicalTimestamp); } @@ -509,12 +552,12 @@ export async function runDenseIntradayRetention( // the stale DAY bucket must re-aggregate against the single // consolidated row, exactly as the mean-consolidation drain does. // - // iOS#34 — for PULSE this would COLLAPSE the true intraday min/max the - // pre-fold recompute above just captured (the single folded row makes - // min == max == mean), so the PULSE DAY bucket is left as the pre-fold - // recompute set it. HRV (no min/max consumer) keeps the post-fold - // recompute against its single mean row. - if (!isPulse) { + // iOS#34 — for the min/max-preserving types (PULSE + SpO2) this would + // COLLAPSE the true intraday min/max the pre-fold recompute above just + // captured (the single folded row makes min == max == mean), so their + // DAY bucket is left as the pre-fold recompute set it. HRV (no min/max + // consumer) keeps the post-fold recompute against its single mean row. + if (!preservesMinMax) { await recomputeBucketsForMeasurement(userId, type, canonicalTimestamp); }