Merge w11b/dense into release/v1.19.0

This commit is contained in:
Marc-André Bombeck committed 2026-06-21 10:31:45 +02:00
commit 3faafef0df
3 files changed
+332 -22

No files matched your search

@@ -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(),
);
});
@@ -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<unknown>) => 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<typeof vi.fn>).mockImplementation(
async (cb: (t: unknown) => Promise<unknown>) => {
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);
});
});
@@ -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<MeasurementType> =
new Set<MeasurementType>(["HEART_RATE_VARIABILITY", "PULSE"]);
new Set<MeasurementType>([
"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<MeasurementType> =
new Set<MeasurementType>(["PULSE", "OXYGEN_SATURATION"]);
/**
* Retention bound (days). Per-sample HRV / HR rows older than this window
@@ -302,15 +342,18 @@ export async function runDenseIntradayRetention(
}): Promise<DayWriteOutcome> => {
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);
}