From 6ab72aec3efad32c56a5b8210dab98d0cf023e12 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 8 Oct 2026 11:40:02 +0200 Subject: [PATCH] Fold Health Connect heart rate like Apple Health The nightly dense-intraday fold scanned Apple Health rows only, so the per-minute heart rate a Health Connect import writes for the last 90 days never aged into hourly means and every new export added another quarter of raw samples. The fold now walks each fold source in turn, Apple Health first and unchanged, and folds Health Connect samples into hourly stats: rows under their own source, deleting the raw rows outright. The folded_window rule covers both sources for the hour window and looks a covering row up under the sample's own source; the day window stays Apple Health only. The Health Connect importer applies the same rule to raw samples, so an export that still holds folded minutes does not put them back. --- docs/integrations/health-connect.md | 6 + src/lib/import/health-connect/import.ts | 46 +- src/lib/jobs/dense-intraday-retention.ts | 7 +- .../dense-intraday-pulse-summary.test.ts | 8 +- .../dense-intraday-resting-mint.test.ts | 9 +- .../dense-intraday-retention.test.ts | 30 +- .../dense-intraday-spo2-summary.test.ts | 9 +- .../__tests__/folded-window.test.ts | 77 +++ .../measurements/dense-intraday-retention.ts | 516 ++++++++++-------- src/lib/measurements/folded-window.ts | 105 ++-- .../dense-intraday-retention.test.ts | 85 +++ .../integration/health-connect-import.test.ts | 48 ++ 12 files changed, 649 insertions(+), 297 deletions(-) diff --git a/docs/integrations/health-connect.md b/docs/integrations/health-connect.md index 5ebf70738..df370ea98 100644 --- a/docs/integrations/health-connect.md +++ b/docs/integrations/health-connect.md @@ -135,6 +135,12 @@ export again writes nothing; a newer export that overlaps an older one adds only what is new. A value that changed on the phone is updated; a value you deleted in HealthLog stays deleted. +Heart rate imported per minute ages the way Apple Health heart rate +does: once a minute is older than 90 days, the nightly pass folds its +hour into one hourly mean and removes the minutes. A later import of +an export that still holds those minutes leaves them out, so an hour +is never counted twice. + ## Privacy - The uploaded ZIP and the extracted database are deleted as soon as diff --git a/src/lib/import/health-connect/import.ts b/src/lib/import/health-connect/import.ts index 320f27745..ec20132fc 100644 --- a/src/lib/import/health-connect/import.ts +++ b/src/lib/import/health-connect/import.ts @@ -19,8 +19,9 @@ * would count a step the phone and the watch both saw twice. * - Heart rate: per-minute samples of the last 90 days as raw rows, older * ones as hourly means (min and max kept), one app per hour by the vitals - * priority. The Apple Health retention fold only folds Apple Health rows, - * so this import folds its own history as it writes it. + * priority. The nightly retention fold folds the raw samples into the + * same hourly `stats:` rows once they age past 90 days, and a raw sample + * whose hour is already folded is left out (`folded_window`). * - Sleep: one row per stage (`sleep-stage-map.ts`), a session without * stages as one asleep row; of two overlapping sessions from different * apps the higher-ranked app's is kept. @@ -58,6 +59,12 @@ import { valuesMatch, } from "@/lib/measurements/cross-source-merge"; import { reconcileExternalMeasurement } from "@/lib/measurements/reconcile-external-measurement"; +import { + FOLDED_WINDOW_REASON, + findFoldedWindowDuplicates, + isFoldedWindowCandidate, + loadFoldTimezone, +} from "@/lib/measurements/folded-window"; import { canonicalDailyTimestamp } from "@/lib/measurements/consolidation-tz"; import { validateMeasurementRange } from "@/lib/validations/measurement"; import { isModuleEnabled } from "@/lib/modules/gate"; @@ -214,6 +221,8 @@ class ImportRun { private readonly userId: string; private readonly tz: string; private readonly now: Date; + /** The account zone the fold keys its hours in, read once on first need. */ + private foldTz: string | undefined; private readonly apps = new Map(); private readonly leftOutApps = new Set(); private readonly devices = new Map(); @@ -1335,6 +1344,7 @@ class ImportRun { }); if (sameReadingCheck) batch = await this.withoutOtherSourceTwins(batch); if (hourlyPulse) batch = await this.withoutRawPulseHours(batch); + else batch = await this.withoutFoldedHours(batch); if (batch.length === 0) return; const inserted = await insertNewMeasurementRows( @@ -1521,6 +1531,38 @@ class ImportRun { }); } + /** + * A raw sample of a folded type is left out when the nightly fold has + * already turned its hour into a `stats:` mean under this source. The fold + * deletes the samples it summarises, so without this a later import of an + * export that still holds them would put them back beside their own mean. + * Same rule as the Apple Health batch route and ZIP import + * (`folded-window.ts`); rows too young to have been folded cost no query. + */ + private async withoutFoldedHours(rows: PendingRow[]): Promise { + if (rows.length === 0) return rows; + const candidates = rows.map((row) => ({ + type: row.type, + source: SOURCE, + externalId: row.externalId, + measuredAt: row.measuredAt, + })); + if (!candidates.some((row) => isFoldedWindowCandidate(row))) return rows; + this.foldTz ??= await loadFoldTimezone(this.prisma, this.userId); + const folded = await findFoldedWindowDuplicates( + this.prisma, + this.userId, + candidates, + { tz: this.foldTz }, + ); + if (folded.size === 0) return rows; + return rows.filter((row, index) => { + if (!folded.has(index)) return true; + this.skip(FOLDED_WINDOW_REASON, 1, row.type); + return false; + }); + } + /** Called by the collapse when two apps write the same instant. */ noteCollision(type: string): void { this.skip("same_instant_other_app", 1, type); diff --git a/src/lib/jobs/dense-intraday-retention.ts b/src/lib/jobs/dense-intraday-retention.ts index a0c52a11a..7cbcdfad1 100644 --- a/src/lib/jobs/dense-intraday-retention.ts +++ b/src/lib/jobs/dense-intraday-retention.ts @@ -25,6 +25,7 @@ import { runDenseIntradayRetention, DENSE_INTRADAY_RETENTION_TYPES, DENSE_INTRADAY_RETENTION_DAYS, + DENSE_INTRADAY_FOLD_SOURCES, } from "@/lib/measurements/dense-intraday-retention"; export const DENSE_INTRADAY_RETENTION_QUEUE = "dense-intraday-retention"; @@ -140,8 +141,8 @@ export async function runDenseIntradayRetentionForUser( /** * Boot-time discovery. Finds every user holding at least one LIVE - * per-sample dense-tier row (an `APPLE_HEALTH` row of a - * `DENSE_INTRADAY_RETENTION_TYPES` type whose externalId does NOT start + * per-sample dense-tier row (a row of a `DENSE_INTRADAY_FOLD_SOURCES` source + * and a `DENSE_INTRADAY_RETENTION_TYPES` type whose externalId does NOT start * with the daily-stats prefix, that is not tombstoned, AND whose * `measuredAt` is older than the retention window) and enqueues one * retention job per account. @@ -187,7 +188,7 @@ export async function enqueueBootTimeDenseIntradayRetention(): Promise<{ const users = await prisma.measurement.groupBy({ by: ["userId"], where: { - source: "APPLE_HEALTH", + source: { in: [...DENSE_INTRADAY_FOLD_SOURCES] }, type: { in: types }, deletedAt: null, NOT: { externalId: { startsWith: "stats:" } }, diff --git a/src/lib/measurements/__tests__/dense-intraday-pulse-summary.test.ts b/src/lib/measurements/__tests__/dense-intraday-pulse-summary.test.ts index aee9a1c02..714fd0819 100644 --- a/src/lib/measurements/__tests__/dense-intraday-pulse-summary.test.ts +++ b/src/lib/measurements/__tests__/dense-intraday-pulse-summary.test.ts @@ -84,8 +84,12 @@ function buildPrismaMock(opts: { }; // Top-level client. - const findMany = vi.fn(async (args: { where: { type: string } }) => - args.where.type === "PULSE" ? opts.pulseRows : [], + // The seeded rows are Apple Health's; the Health Connect walk finds none. + const findMany = vi.fn( + async (args: { where: { type: string; source: string } }) => + args.where.type === "PULSE" && args.where.source === "APPLE_HEALTH" + ? opts.pulseRows + : [], ); // Native-resting probe is the only top-level `findFirst` call. const topFindFirst = vi diff --git a/src/lib/measurements/__tests__/dense-intraday-resting-mint.test.ts b/src/lib/measurements/__tests__/dense-intraday-resting-mint.test.ts index e72ae5db1..9c9c0ce2a 100644 --- a/src/lib/measurements/__tests__/dense-intraday-resting-mint.test.ts +++ b/src/lib/measurements/__tests__/dense-intraday-resting-mint.test.ts @@ -106,8 +106,13 @@ function buildPrismaMock(existingResting: Array<{ source: string }>) { .mockResolvedValue([{ id: "user-1", timezone: "Europe/Berlin" }]), }, measurement: { - findMany: vi.fn(async (args: { where: { type: string } }) => - args.where.type === "PULSE" ? pulseRows() : [], + // The seeded rows are Apple Health's; the Health Connect walk + // finds none. + findMany: vi.fn( + async (args: { where: { type: string; source: string } }) => + args.where.type === "PULSE" && args.where.source === "APPLE_HEALTH" + ? pulseRows() + : [], ), findFirst: probeFindFirst, upsert, diff --git a/src/lib/measurements/__tests__/dense-intraday-retention.test.ts b/src/lib/measurements/__tests__/dense-intraday-retention.test.ts index 740f40d21..e660d5731 100644 --- a/src/lib/measurements/__tests__/dense-intraday-retention.test.ts +++ b/src/lib/measurements/__tests__/dense-intraday-retention.test.ts @@ -62,9 +62,12 @@ function buildPrismaMock( ); const updateMany = vi.fn().mockResolvedValue({ count: 0 }); const deleteMany = vi.fn().mockResolvedValue({ count: 0 }); + // The seeded rows are Apple Health's; the Health Connect walk finds none. const findManyMeasurement = vi.fn( - async (args: { where: { type: string } }) => - rowsByType[args.where.type] ?? [], + async (args: { where: { type: string; source: string } }) => + args.where.source === "APPLE_HEALTH" + ? (rowsByType[args.where.type] ?? []) + : [], ); const tx = { measurement: { @@ -140,20 +143,21 @@ describe("runDenseIntradayRetention — retention bound", () => { } }); - it("scans exactly the dense-tier types, source-scoped to APPLE_HEALTH", async () => { + it("scans exactly the dense-tier types, once per fold source, Apple Health first", async () => { const { mock, findManyMeasurement } = buildPrismaMock({}); await runDenseIntradayRetention(mock, { log: () => {} }); - const scannedTypes = findManyMeasurement.mock.calls.map( - (c) => (c[0] as { where: { type: string } }).where.type, - ); - expect(scannedTypes.sort()).toEqual( - Array.from(DENSE_INTRADAY_RETENTION_TYPES).sort(), - ); - for (const call of findManyMeasurement.mock.calls) { - const where = (call[0] as unknown as { where: { source: string } }).where; - expect(where.source).toBe("APPLE_HEALTH"); - } + const scans = findManyMeasurement.mock.calls.map((c) => { + const where = (c[0] as { where: { type: string; source: string } }).where; + return `${where.source}:${where.type}`; + }); + const types = Array.from(DENSE_INTRADAY_RETENTION_TYPES); + // One walk per source, each scoped to its one source: two sources + // never share a mean, and manual or Withings rows are never scanned. + expect(scans).toEqual([ + ...types.map((type) => `APPLE_HEALTH:${type}`), + ...types.map((type) => `HEALTH_CONNECT:${type}`), + ]); }); }); diff --git a/src/lib/measurements/__tests__/dense-intraday-spo2-summary.test.ts b/src/lib/measurements/__tests__/dense-intraday-spo2-summary.test.ts index 6ae4152e5..23c4e2fa7 100644 --- a/src/lib/measurements/__tests__/dense-intraday-spo2-summary.test.ts +++ b/src/lib/measurements/__tests__/dense-intraday-spo2-summary.test.ts @@ -75,8 +75,13 @@ function buildPrismaMock(opts: { spo2Rows: PerSampleRow[] }) { }, }; - const findMany = vi.fn(async (args: { where: { type: string } }) => - args.where.type === "OXYGEN_SATURATION" ? opts.spo2Rows : [], + // The seeded rows are Apple Health's; the Health Connect walk finds none. + const findMany = vi.fn( + async (args: { where: { type: string; source: string } }) => + args.where.type === "OXYGEN_SATURATION" && + args.where.source === "APPLE_HEALTH" + ? opts.spo2Rows + : [], ); // Native-resting probe + derived-resting upsert: PULSE-only surfaces. const topFindFirst = vi.fn().mockResolvedValue(null); diff --git a/src/lib/measurements/__tests__/folded-window.test.ts b/src/lib/measurements/__tests__/folded-window.test.ts index f75359642..4c9bc9b0d 100644 --- a/src/lib/measurements/__tests__/folded-window.test.ts +++ b/src/lib/measurements/__tests__/folded-window.test.ts @@ -226,3 +226,80 @@ describe("findCompactionTombstones", () => { expect(found.size).toBe(0); }); }); + +/** + * Health Connect folds its raw heart-rate history the way Apple Health does: + * past the 90-day window into one `stats:` row per local hour, under its own + * source. The mean-day rule stays Apple Health only, because the nightly + * mean consolidation never reads another source. + */ +describe("Health Connect", () => { + const old = new Date("2026-03-01T10:15:00.000Z"); + const hc = ( + type: string, + at: Date, + overrides: Record = {}, + ) => + raw(type, at, { + source: "HEALTH_CONNECT", + externalId: "hc:00000000-0000-0000-0000-000000000001:1", + ...overrides, + }); + + /** A lookup that answers per source, as the database does. */ + function sourceAwareClient( + live: Array<{ type: string; source: string; externalId: string }>, + ) { + const findMany = vi.fn(async (args: { where: { source: string } }) => + live.filter((row) => row.source === args.where.source), + ); + return { + findMany, + client: { + measurement: { findMany }, + } as unknown as Parameters[0], + }; + } + + it("takes a raw dense sample past the window, and not a mean-type one", () => { + expect(isFoldedWindowCandidate(hc("PULSE", old), NOW)).toBe(true); + expect( + isFoldedWindowCandidate( + hc("PULSE", new Date(NOW.getTime() - 89 * DAY)), + NOW, + ), + ).toBe(false); + expect(isFoldedWindowCandidate(hc("RESPIRATORY_RATE", old), NOW)).toBe( + false, + ); + expect( + isFoldedWindowCandidate( + hc("PULSE", old, { + externalId: "stats:HKQuantityTypeIdentifierHeartRate:2026-03-01T10", + }), + NOW, + ), + ).toBe(false); + }); + + it("is covered only by a `stats:` row of its own source", async () => { + const hour = "stats:HKQuantityTypeIdentifierHeartRate:2026-03-01T10"; + const { client: c, findMany } = sourceAwareClient([ + { type: "PULSE", source: "HEALTH_CONNECT", externalId: hour }, + ]); + const found = await findFoldedWindowDuplicates( + c, + "u1", + [hc("PULSE", old), raw("PULSE", old)], + { now: NOW, tz: "UTC" }, + ); + // The Health Connect sample is a duplicate; the Apple Health sample of + // the same hour is not, because no Apple Health `stats:` row covers it. + expect([...found]).toEqual([0]); + const sources = findMany.mock.calls.map( + (call) => + (call as unknown as [{ where: { source: string } }])[0].where.source, + ); + expect(sources.sort()).toEqual(["APPLE_HEALTH", "HEALTH_CONNECT"]); + }); +}); diff --git a/src/lib/measurements/dense-intraday-retention.ts b/src/lib/measurements/dense-intraday-retention.ts index f13ec2fb9..1cbf1ab9d 100644 --- a/src/lib/measurements/dense-intraday-retention.ts +++ b/src/lib/measurements/dense-intraday-retention.ts @@ -50,6 +50,7 @@ */ import { holdAccountAgainstRestore } from "@/lib/export/restore-lock"; import type { + MeasurementSource, MeasurementType, Prisma, PrismaClient, @@ -154,6 +155,23 @@ export const DENSE_INTRADAY_RETENTION_TYPES: ReadonlySet = */ export const DENSE_INTRADAY_RETENTION_DAYS = 90; +/** + * The sources whose raw dense samples fold (v1.42). Apple Health from the + * start; Health Connect because its importer writes the samples of the last + * 90 days raw, exactly like an Apple Health sync, and without a fold every + * new export would add another quarter of minute-by-minute heart rate that + * never ages into hourly means. Each source folds into its OWN `stats:` rows, + * so an hour two sources both covered keeps one mean per source, the same + * way the raw samples kept their source. + */ +export const DENSE_INTRADAY_FOLD_SOURCES = [ + "APPLE_HEALTH", + "HEALTH_CONNECT", +] as const satisfies readonly MeasurementSource[]; + +export type DenseIntradayFoldSource = + (typeof DENSE_INTRADAY_FOLD_SOURCES)[number]; + /** The stats externalId prefix marks an already-collapsed row. */ const DAILY_STATS_PREFIX = "stats:"; @@ -249,10 +267,13 @@ export async function adoptOrMintHourlyRows( input: { userId: string; type: MeasurementType; + /** The source the folded samples carry; the hourly rows carry it too. */ + source?: DenseIntradayFoldSource; unit: string; slots: readonly HourlySlot[]; }, ): Promise { + const source = input.source ?? "APPLE_HEALTH"; if (input.slots.length === 0) return []; // Every row that either index could put in the way, lowest id first — the @@ -263,7 +284,7 @@ export async function adoptOrMintHourlyRows( where: { userId: input.userId, type: input.type, - source: "APPLE_HEALTH", + source, OR: [ { externalId: { in: input.slots.map((slot) => slot.externalId) } }, { @@ -338,7 +359,7 @@ export async function adoptOrMintHourlyRows( type: input.type, value: slot.value, unit: input.unit, - source: "APPLE_HEALTH" as const, + source, measuredAt: slot.anchor, externalId: slot.externalId, })), @@ -538,263 +559,280 @@ export async function runDenseIntradayRetention( // for the dry-run hourly fan-out estimate (writeDay never runs on dry-run). let currentTz = DEFAULT_TIMEZONE; - const walk = await runConsolidation({ - prismaClient, - options: { ...options, cutoffHours }, - types: DENSE_INTRADAY_RETENTION_TYPES, - hkIdentifierForType, - dailyStatsExternalId, - statsPrefix: DAILY_STATS_PREFIX, - // The base's per-day reducedValue is the day MEAN; the hourly fold - // recomputes per-hour means itself (same `meanBucketValue` reducer, so - // the two surfaces can never drift), and the day-level value only - // feeds the summary/dry-run reporting. - reduce: meanBucketValue, - // Live per-sample rows for the type, source-scoped to APPLE_HEALTH so - // manual + Withings spot rows survive. The minted hourly stats rows are - // excluded by the NOT-startsWith predicate so they are never re-folded; - // soft-deleted rows are excluded so a re-run converges. `cutoffAt` is - // the retention boundary — rows newer than it (inside the window) are - // NOT scanned, so the dense intra-day shape is preserved. - buildScanWhere: ({ userId, type, cutoffAt, statsPrefix }) => ({ - userId, - source: "APPLE_HEALTH", - type, - deletedAt: null, - NOT: { externalId: { startsWith: statsPrefix } }, - ...(cutoffAt ? { measuredAt: { lt: cutoffAt } } : {}), - }), - scanSelect: { - id: true, - type: true, - value: true, - measuredAt: true, - externalId: true, - unit: true, - }, - writeDay: async ({ - prismaClient: pc, - userId, - type, - externalId, - canonicalTimestamp, - dayRows, - sourceRowIds, - tz, - dateKey, - }): Promise => { - const unit = dayRows[0]?.unit ?? "unknown"; - const isPulse = type === "PULSE"; - const hkIdentifier = hkIdentifierForType(type); - // Unreachable: the base skips types without an HK identifier before - // scanning; asserted so a future type-set edit fails loud, not by - // minting a malformed externalId. - if (!hkIdentifier) { - throw new Error(`no HK identifier for dense-tier type ${type}`); - } + // Awaited inside, not returned bare: the restore-lock guard finds the + // consolidation passes by their `await runConsolidation`. + const walkSource = async (source: DenseIntradayFoldSource) => + await runConsolidation({ + prismaClient, + options: { ...options, cutoffHours }, + types: DENSE_INTRADAY_RETENTION_TYPES, + hkIdentifierForType, + dailyStatsExternalId, + statsPrefix: DAILY_STATS_PREFIX, + // The base's per-day reducedValue is the day MEAN; the hourly fold + // recomputes per-hour means itself (same `meanBucketValue` reducer, so + // the two surfaces can never drift), and the day-level value only + // feeds the summary/dry-run reporting. + reduce: meanBucketValue, + // Live per-sample rows for the type, scoped to the one source this walk + // folds so manual + Withings spot rows survive and two sources never + // share a mean. The minted hourly stats rows are + // excluded by the NOT-startsWith predicate so they are never re-folded; + // soft-deleted rows are excluded so a re-run converges. `cutoffAt` is + // the retention boundary — rows newer than it (inside the window) are + // NOT scanned, so the dense intra-day shape is preserved. + buildScanWhere: ({ userId, type, cutoffAt, statsPrefix }) => ({ + userId, + source, + type, + deletedAt: null, + NOT: { externalId: { startsWith: statsPrefix } }, + ...(cutoffAt ? { measuredAt: { lt: cutoffAt } } : {}), + }), + scanSelect: { + id: true, + type: true, + value: true, + measuredAt: true, + externalId: true, + unit: true, + }, + writeDay: async ({ + prismaClient: pc, + userId, + type, + externalId, + canonicalTimestamp, + dayRows, + sourceRowIds, + tz, + dateKey, + }): Promise => { + const unit = dayRows[0]?.unit ?? "unknown"; + const isPulse = type === "PULSE"; + const hkIdentifier = hkIdentifierForType(type); + // Unreachable: the base skips types without an HK identifier before + // scanning; asserted so a future type-set edit fails loud, not by + // minting a malformed externalId. + if (!hkIdentifier) { + throw new Error(`no HK identifier for dense-tier type ${type}`); + } - // DAY-rollup fidelity — mean AND min/max (all dense-tier types). The - // persistent `MeasurementRollup` DAY bucket aggregates LIVE rows; - // recomputing it AFTER the fold would aggregate the hourly means: the - // min/max collapse to the extreme hourly MEANS (for SpO2 hiding the - // overnight-desaturation nadir; for PULSE the resting floor + workout - // peak) and the mean becomes an unweighted mean-of-hourly-means that - // drifts whenever per-hour sample counts differ. So the DAY bucket is - // recomputed from the RAW rows BEFORE the fold, while they are still - // live, and the post-fold DAY recompute is skipped entirely. The - // WEEK/MONTH/YEAR recomputes this call enqueues run post-commit - // against whatever is live then, as before. - await recomputeBucketsForMeasurement(userId, type, canonicalTimestamp); + // DAY-rollup fidelity — mean AND min/max (all dense-tier types). The + // persistent `MeasurementRollup` DAY bucket aggregates LIVE rows; + // recomputing it AFTER the fold would aggregate the hourly means: the + // min/max collapse to the extreme hourly MEANS (for SpO2 hiding the + // overnight-desaturation nadir; for PULSE the resting floor + workout + // peak) and the mean becomes an unweighted mean-of-hourly-means that + // drifts whenever per-hour sample counts differ. So the DAY bucket is + // recomputed from the RAW rows BEFORE the fold, while they are still + // live, and the post-fold DAY recompute is skipped entirely. The + // WEEK/MONTH/YEAR recomputes this call enqueues run post-commit + // against whatever is live then, as before. + await recomputeBucketsForMeasurement(userId, type, canonicalTimestamp); - // Fold the day into per-LOCAL-HOUR mean rows, anchored at local - // HH:30. Sorted for a deterministic write order across runs. - const byHour = [...bucketRowsByLocalHour(dayRows, tz).entries()].sort( - (a, b) => a[0] - b[0], - ); + // Fold the day into per-LOCAL-HOUR mean rows, anchored at local + // HH:30. Sorted for a deterministic write order across runs. + const byHour = [...bucketRowsByLocalHour(dayRows, tz).entries()].sort( + (a, b) => a[0] - b[0], + ); - const foldOnce = async (): Promise<{ - removed: number; - retiredDaily: boolean; - }> => { - let removed = 0; - let retiredDaily = false; - await pc.$transaction(async (tx) => { - // First, before any reading is touched: see `restore-lock.ts`. - await holdAccountAgainstRestore(tx, userId); - const canonicalRowIds = await adoptOrMintHourlyRows(tx, { - userId, - type, - unit, - slots: byHour.map(([hour, hourRows]) => ({ - externalId: hourlyStatsExternalId(hkIdentifier, dateKey, hour), - anchor: canonicalHourlyTimestamp(dateKey, hour, tz), - value: meanBucketValue(hourRows), - })), - }); - - // Retire a live pre-hourly DAILY `stats:` row for this day (the - // late-sync path: raw samples arriving for a day folded to the - // daily grain before v1.28.31). Same transaction as the hourly - // mint — an AVG-over-live-rows reader must never see the daily - // row and the hourly rows live at once (double count). - const dailyRow = await tx.measurement.findFirst({ - where: { + const foldOnce = async (): Promise<{ + removed: number; + retiredDaily: boolean; + }> => { + let removed = 0; + let retiredDaily = false; + await pc.$transaction(async (tx) => { + // First, before any reading is touched: see `restore-lock.ts`. + await holdAccountAgainstRestore(tx, userId); + const canonicalRowIds = await adoptOrMintHourlyRows(tx, { userId, type, - source: "APPLE_HEALTH", - externalId, - deletedAt: null, - }, - select: { id: true }, - orderBy: { id: "asc" }, - }); - if (dailyRow && !canonicalRowIds.includes(dailyRow.id)) { - await tx.measurement.update({ - where: { id: dailyRow.id }, - data: { deletedAt: new Date() }, + source, + unit, + slots: byHour.map(([hour, hourRows]) => ({ + externalId: hourlyStatsExternalId(hkIdentifier, dateKey, hour), + anchor: canonicalHourlyTimestamp(dateKey, hour, tz), + value: meanBucketValue(hourRows), + })), }); - retiredDaily = true; - } - // Delete the out-of-window per-sample rows in the same - // transaction. EXCLUDE the adopted canonical rows: a per-sample row - // that happened to fall on an hourly anchor is the row just adopted - // as that hour's mean, so removing it would erase the fold. - // - // v1.42 — deleted outright, like the cumulative drain's rows, - // where they used to be tombstoned for the 75-day retention. No - // client ever pulled these rows, so the tombstone told nobody - // anything; it only held the re-upload of a folded sample off, - // and the `folded_window` ingest guard (`folded-window.ts`) does - // that now from the live hourly row. The tombstones made up most - // of the table and its indexes. - const del = await tx.measurement.deleteMany({ - where: { - id: { in: sourceRowIds, notIn: canonicalRowIds }, - deletedAt: null, - }, + // Retire a live pre-hourly DAILY `stats:` row for this day (the + // late-sync path: raw samples arriving for a day folded to the + // daily grain before v1.28.31). Same transaction as the hourly + // mint — an AVG-over-live-rows reader must never see the daily + // row and the hourly rows live at once (double count). + const dailyRow = await tx.measurement.findFirst({ + where: { + userId, + type, + source, + externalId, + deletedAt: null, + }, + select: { id: true }, + orderBy: { id: "asc" }, + }); + if (dailyRow && !canonicalRowIds.includes(dailyRow.id)) { + await tx.measurement.update({ + where: { id: dailyRow.id }, + data: { deletedAt: new Date() }, + }); + retiredDaily = true; + } + + // Delete the out-of-window per-sample rows in the same + // transaction. EXCLUDE the adopted canonical rows: a per-sample row + // that happened to fall on an hourly anchor is the row just adopted + // as that hour's mean, so removing it would erase the fold. + // + // v1.42 — deleted outright, like the cumulative drain's rows, + // where they used to be tombstoned for the 75-day retention. No + // client ever pulled these rows, so the tombstone told nobody + // anything; it only held the re-upload of a folded sample off, + // and the `folded_window` ingest guard (`folded-window.ts`) does + // that now from the live hourly row. The tombstones made up most + // of the table and its indexes. + const del = await tx.measurement.deleteMany({ + where: { + id: { in: sourceRowIds, notIn: canonicalRowIds }, + deletedAt: null, + }, + }); + removed = del.count; }); - removed = del.count; - }); - return { removed, retiredDaily }; - }; + return { removed, retiredDaily }; + }; - let foldResult: { removed: number; retiredDaily: boolean }; - try { - foldResult = await foldOnce(); - } catch (err) { - if (!isUniqueConstraintViolation(err)) throw err; - // A concurrent writer won a slot mid-transaction. Retry once: the - // deterministic lookup now resolves the winning row and adopts it, - // so this pass cannot create a duplicate. - foldResult = await foldOnce(); - } - // Counters mutate OUTSIDE the retryable transaction so a P2002 retry - // cannot double-count. - summary.totals.hourlyRowsUpserted += byHour.length; - if (foldResult.retiredDaily) summary.totals.dailyRowsRetired += 1; + let foldResult: { removed: number; retiredDaily: boolean }; + try { + foldResult = await foldOnce(); + } catch (err) { + if (!isUniqueConstraintViolation(err)) throw err; + // A concurrent writer won a slot mid-transaction. Retry once: the + // deterministic lookup now resolves the winning row and adopts it, + // so this pass cannot create a duplicate. + foldResult = await foldOnce(); + } + // Counters mutate OUTSIDE the retryable transaction so a P2002 retry + // cannot double-count. + summary.totals.hourlyRowsUpserted += byHour.length; + if (foldResult.retiredDaily) summary.totals.dailyRowsRetired += 1; - // iOS#34 — preserve the resting signal for PROXY users. For a user - // with zero native `RESTING_HEART_RATE` rows, the read-path resolver - // derives resting from the 20th-percentile of each day's RAW PULSE; the - // fold has just deleted those rows, so that day's resting figure would - // vanish (the hourly means flatten the low tail the percentile reads). - // Mint one derived `RESTING_HEART_RATE` row from the same percentile, - // sourced `COMPUTED` (distinct from Apple's own resting rows on both - // unique indexes), so the day reads back as the clean resting series - // the resolver prefers. Skipped for users WITH native resting rows - // (the resolver ignores the proxy entirely for them, so a derived row - // would only double-count). - if (isPulse) { - const restingValue = deriveDailyRestingFromPulse(dayRows); - if (restingValue !== null && !(await userHasNativeResting(userId))) { - const restingExternalId = dailyStatsExternalId( - RESTING_HK_IDENTIFIER, - // The fold's canonical timestamp is local-noon of the day; reuse - // its UTC date for the resting row's `stats:` key so a re-run - // upserts the same row instead of minting a sibling. - // eslint-disable-next-line healthlog/no-utc-day-key -- UTC by design: stable stats: externalId key; changing it would mint a sibling row on re-run - canonicalTimestamp.toISOString().slice(0, 10), - ); - await pc.measurement.upsert({ - where: { - userId_type_source_externalId: { + // iOS#34 — preserve the resting signal for PROXY users. For a user + // with zero native `RESTING_HEART_RATE` rows, the read-path resolver + // derives resting from the 20th-percentile of each day's RAW PULSE; the + // fold has just deleted those rows, so that day's resting figure would + // vanish (the hourly means flatten the low tail the percentile reads). + // Mint one derived `RESTING_HEART_RATE` row from the same percentile, + // sourced `COMPUTED` (distinct from Apple's own resting rows on both + // unique indexes), so the day reads back as the clean resting series + // the resolver prefers. Skipped for users WITH native resting rows + // (the resolver ignores the proxy entirely for them, so a derived row + // would only double-count). + if (isPulse) { + const restingValue = deriveDailyRestingFromPulse(dayRows); + if (restingValue !== null && !(await userHasNativeResting(userId))) { + const restingExternalId = dailyStatsExternalId( + RESTING_HK_IDENTIFIER, + // The fold's canonical timestamp is local-noon of the day; reuse + // its UTC date for the resting row's `stats:` key so a re-run + // upserts the same row instead of minting a sibling. + // eslint-disable-next-line healthlog/no-utc-day-key -- UTC by design: stable stats: externalId key; changing it would mint a sibling row on re-run + canonicalTimestamp.toISOString().slice(0, 10), + ); + await pc.measurement.upsert({ + where: { + userId_type_source_externalId: { + userId, + type: "RESTING_HEART_RATE", + source: "COMPUTED", + externalId: restingExternalId, + }, + }, + create: { userId, type: "RESTING_HEART_RATE", + value: restingValue, + unit: "bpm", source: "COMPUTED", + measuredAt: canonicalTimestamp, externalId: restingExternalId, }, - }, - create: { + update: { value: restingValue, deletedAt: null }, + }); + summary.totals.derivedRestingRowsUpserted += 1; + // The derived row is its own (type, day); recompute its DAY bucket + // so the rollup tier serves it on a covered read. + await recomputeBucketsForMeasurement( userId, - type: "RESTING_HEART_RATE", - value: restingValue, - unit: "bpm", - source: "COMPUTED", - measuredAt: canonicalTimestamp, - externalId: restingExternalId, - }, - update: { value: restingValue, deletedAt: null }, - }); - summary.totals.derivedRestingRowsUpserted += 1; - // The derived row is its own (type, day); recompute its DAY bucket - // so the rollup tier serves it on a covered read. - await recomputeBucketsForMeasurement( - userId, - "RESTING_HEART_RATE", - canonicalTimestamp, - ); + "RESTING_HEART_RATE", + canonicalTimestamp, + ); + } } - } - // No post-fold DAY recompute for ANY dense-tier type — it would - // overwrite the true raw-derived mean/min/max the pre-fold recompute - // above captured with hourly-mean-derived approximations (see the - // fidelity note there). + // No post-fold DAY recompute for ANY dense-tier type — it would + // overwrite the true raw-derived mean/min/max the pre-fold recompute + // above captured with hourly-mean-derived approximations (see the + // fidelity note there). - return { kind: "written", sourceRowsRemoved: foldResult.removed }; - }, - recordBucket: ({ dayRows, outcome }) => { - summary.totals.daysConsolidated += 1; - if (outcome === null) { - // Dry-run: writeDay never ran, so estimate the hourly fan-out here - // from the same bucketing the real fold would use. - summary.totals.hourlyRowsUpserted += bucketRowsByLocalHour( - dayRows, - currentTz, - ).size; - } - summary.totals.perSampleRowsSoftDeleted += - outcome?.kind === "written" - ? outcome.sourceRowsRemoved - : dayRows.length; - }, - onUserStart: ({ tz }) => { - currentTz = tz; - }, - onUserComplete: ({ userId, tz, dryRun }) => { - log( - `[dense-intraday-retention] user=${userId} tz=${tz}${dryRun ? " (dry-run)" : ""}`, - ); - }, - // Per-day failure boundary (v1.18.10 lesson). A day whose fold throws — - // e.g. a residual unique-index collision the adopt-in-place path could - // not absorb, or a derived-resting upsert that races a native row — is - // logged and STEPPED OVER so the global walk keeps draining every other - // user / type / day. The failed day keeps its raw rows live (the fold - // deletes only on a committed transaction), so the next nightly run - // retries it. Without this, one poisoned day aborted the whole walk and - // stranded every later account. - onBucketError: ({ userId, type, dateKey, error }) => { - const reason = error instanceof Error ? error.message : String(error); - log( - `[dense-intraday-retention] user=${userId} type=${type} day=${dateKey} failed — ${reason}; continuing with the next bucket`, - ); - }, - }); + return { kind: "written", sourceRowsRemoved: foldResult.removed }; + }, + recordBucket: ({ dayRows, outcome }) => { + summary.totals.daysConsolidated += 1; + if (outcome === null) { + // Dry-run: writeDay never ran, so estimate the hourly fan-out here + // from the same bucketing the real fold would use. + summary.totals.hourlyRowsUpserted += bucketRowsByLocalHour( + dayRows, + currentTz, + ).size; + } + summary.totals.perSampleRowsSoftDeleted += + outcome?.kind === "written" + ? outcome.sourceRowsRemoved + : dayRows.length; + }, + onUserStart: ({ tz }) => { + currentTz = tz; + }, + onUserComplete: ({ userId, tz, dryRun }) => { + log( + `[dense-intraday-retention] user=${userId} source=${source} tz=${tz}${dryRun ? " (dry-run)" : ""}`, + ); + }, + // Per-day failure boundary (v1.18.10 lesson). A day whose fold throws — + // e.g. a residual unique-index collision the adopt-in-place path could + // not absorb, or a derived-resting upsert that races a native row — is + // logged and STEPPED OVER so the global walk keeps draining every other + // user / type / day. The failed day keeps its raw rows live (the fold + // deletes only on a committed transaction), so the next nightly run + // retries it. Without this, one poisoned day aborted the whole walk and + // stranded every later account. + onBucketError: ({ userId, type, dateKey, error }) => { + const reason = error instanceof Error ? error.message : String(error); + log( + `[dense-intraday-retention] user=${userId} source=${source} type=${type} day=${dateKey} failed — ${reason}; continuing with the next bucket`, + ); + }, + }); - summary.totals.usersScanned = walk.usersScanned; - summary.stoppedEarly = walk.stoppedEarly; + // One walk per source, Apple Health first so its behaviour and its order + // are exactly what they were before Health Connect folded too. A walk that + // stopped early stops the pass: the next run resumes where it left off. + for (const source of DENSE_INTRADAY_FOLD_SOURCES) { + const walk = await walkSource(source); + summary.totals.usersScanned = Math.max( + summary.totals.usersScanned, + walk.usersScanned, + ); + if (walk.stoppedEarly) { + summary.stoppedEarly = true; + break; + } + } log( `[dense-intraday-retention] done — usersScanned=${summary.totals.usersScanned} daysConsolidated=${summary.totals.daysConsolidated} perSampleRowsSoftDeleted=${summary.totals.perSampleRowsSoftDeleted} hourlyRowsUpserted=${summary.totals.hourlyRowsUpserted} dailyRowsRetired=${summary.totals.dailyRowsRetired} derivedRestingRowsUpserted=${summary.totals.derivedRestingRowsUpserted}${options.dryRun ? " (dry-run)" : ""}`, diff --git a/src/lib/measurements/folded-window.ts b/src/lib/measurements/folded-window.ts index 2aeff4167..b6fbdb227 100644 --- a/src/lib/measurements/folded-window.ts +++ b/src/lib/measurements/folded-window.ts @@ -1,14 +1,13 @@ /** * The `folded_window` ingest guard and the compaction-tombstone class (v1.42). * - * The dense-intraday consolidation folds the raw Apple Health samples of a - * DENSE type into one `stats:` row per local hour once they are older than + * The dense-intraday consolidation folds the raw samples of a DENSE type into one `stats:` row per local hour once they are older than * the retention window, and a MEAN type into one `stats:` row per local day * once the consolidation grace has passed. Since v1.42 the folded raw rows are * deleted outright rather than left as tombstones, so a client that re-uploads * an old sample after losing its anchor would put it back beside the hour or - * day that already accounts for it. A raw (non-`stats:`) Apple Health value of - * such a type, old enough to have been folded, is therefore a duplicate when a + * day that already accounts for it. A raw (non-`stats:`) value of such a source + * and type, old enough to have been folded, is therefore a duplicate when a * live `stats:` row covers its hour or day. The batch route and the ZIP import * apply the same rule. * @@ -29,8 +28,12 @@ * the other zone; such a row is then simply not recognised, which is the * pre-v1.42 behaviour, never a wrong refusal. * - * Health Connect does not use this: its importer writes the folded shape - * itself. + * Health Connect folds its raw heart-rate history the same way under its own + * source (the importer writes samples of the last 90 days raw, and the + * nightly pass folds them once they age out), so the hour rule covers both + * sources and a covering row is always looked up under the sample's own + * source. The day rule stays Apple Health only: the nightly mean + * consolidation reads no other source. */ import type { MeasurementSource, @@ -49,6 +52,7 @@ import { hourOfDayForUserTz, } from "@/lib/measurements/consolidation-tz"; import { + DENSE_INTRADAY_FOLD_SOURCES, DENSE_INTRADAY_RETENTION_DAYS, DENSE_INTRADAY_RETENTION_TYPES, hourlyStatsExternalId, @@ -60,15 +64,17 @@ export const FOLDED_WINDOW_REASON: FoldedWindowReason = "folded_window"; const HOUR_MS = 3_600_000; -/** Which types fold into which window, and from what age. */ +/** Which sources and types fold into which window, and from what age. */ export const FOLDED_WINDOW_RULES = [ { window: "hour", + sources: DENSE_INTRADAY_FOLD_SOURCES, types: DENSE_INTRADAY_RETENTION_TYPES, foldedAfterMs: DENSE_INTRADAY_RETENTION_DAYS * 24 * HOUR_MS, }, { window: "day", + sources: ["APPLE_HEALTH"] as const, types: HIGH_FREQUENCY_MEAN_TYPES, foldedAfterMs: CONSOLIDATION_GRACE_CUTOFF_HOURS * HOUR_MS, }, @@ -98,7 +104,6 @@ export function isFoldedWindowCandidate( row: FoldedWindowCandidateRow, now: Date = new Date(), ): boolean { - if (row.source !== "APPLE_HEALTH") return false; if (row.externalId === null) return false; if (row.externalId.startsWith("stats:")) return false; // A collision retirement (`reconcile-external-measurement.ts`) parks the @@ -106,7 +111,10 @@ export function isFoldedWindowCandidate( if (row.externalId.startsWith("retired:")) return false; const age = now.getTime() - row.measuredAt.getTime(); return FOLDED_WINDOW_RULES.some( - (rule) => rule.types.has(row.type) && age > rule.foldedAfterMs, + (rule) => + (rule.sources as readonly MeasurementSource[]).includes(row.source) && + rule.types.has(row.type) && + age > rule.foldedAfterMs, ); } @@ -153,8 +161,9 @@ export async function loadFoldTimezone( } /** - * Indexes (into `rows`) whose covering `stats:` row is live. One query over - * the unique `(user_id, type, source, external_id)` index for the whole set. + * Indexes (into `rows`) whose covering `stats:` row is live under the row's + * own source. One query per source present, each over the unique + * `(user_id, type, source, external_id)` index for the whole set. */ async function coveredIndexes( client: Pick, @@ -163,35 +172,49 @@ async function coveredIndexes( rows: ReadonlyArray<{ index: number; type: MeasurementType; + source: MeasurementSource; measuredAt: Date; }>, ): Promise> { const covered = new Set(); if (rows.length === 0) return covered; - const keyOf = new Map(); - const ids = new Set(); - const types = new Set(); + const bySource = new Map< + MeasurementSource, + { + keyOf: Map; + ids: Set; + types: Set; + } + >(); for (const row of rows) { const id = coveringStatsExternalId(row, tz); if (!id) continue; - keyOf.set(row.index, `${row.type}|${id}`); - ids.add(id); - types.add(row.type); + let group = bySource.get(row.source); + if (!group) { + group = { keyOf: new Map(), ids: new Set(), types: new Set() }; + bySource.set(row.source, group); + } + group.keyOf.set(row.index, `${row.type}|${id}`); + group.ids.add(id); + group.types.add(row.type); } - if (ids.size === 0) return covered; - const live = await client.measurement.findMany({ - where: { - userId, - type: { in: [...types] }, - source: "APPLE_HEALTH", - externalId: { in: [...ids] }, - deletedAt: null, - }, - select: { type: true, externalId: true }, - }); - const liveKeys = new Set(live.map((row) => `${row.type}|${row.externalId}`)); - for (const [index, key] of keyOf) { - if (liveKeys.has(key)) covered.add(index); + for (const [source, group] of bySource) { + const live = await client.measurement.findMany({ + where: { + userId, + type: { in: [...group.types] }, + source, + externalId: { in: [...group.ids] }, + deletedAt: null, + }, + select: { type: true, externalId: true }, + }); + const liveKeys = new Set( + live.map((row) => `${row.type}|${row.externalId}`), + ); + for (const [index, key] of group.keyOf) { + if (liveKeys.has(key)) covered.add(index); + } } return covered; } @@ -211,7 +234,14 @@ export async function findFoldedWindowDuplicates( const now = options.now ?? new Date(); const candidates = rows.flatMap((row, index) => isFoldedWindowCandidate(row, now) - ? [{ index, type: row.type, measuredAt: row.measuredAt }] + ? [ + { + index, + type: row.type, + source: row.source, + measuredAt: row.measuredAt, + }, + ] : [], ); if (candidates.length === 0) return new Set(); @@ -230,7 +260,7 @@ export interface TombstoneCandidateRow extends FoldedWindowCandidateRow { /** * Indexes (into `rows`) of the compaction tombstones ("class A"): soft-deleted - * raw Apple Health samples of a folded type that were already past the fold + * raw samples of a folded source and type that were already past the fold * threshold when they were deleted, and whose hour or day a live `stats:` row * covers. Everything else, in particular a person's own deletions of recent * samples, is class B and keeps the 75-day tombstone retention. @@ -243,7 +273,14 @@ export async function findCompactionTombstones( ): Promise> { const candidates = rows.flatMap((row, index) => row.deletedAt !== null && isFoldedWindowCandidate(row, row.deletedAt) - ? [{ index, type: row.type, measuredAt: row.measuredAt }] + ? [ + { + index, + type: row.type, + source: row.source, + measuredAt: row.measuredAt, + }, + ] : [], ); return coveredIndexes(client, userId, tz, candidates); diff --git a/tests/integration/dense-intraday-retention.test.ts b/tests/integration/dense-intraday-retention.test.ts index db955362e..06a7d6fba 100644 --- a/tests/integration/dense-intraday-retention.test.ts +++ b/tests/integration/dense-intraday-retention.test.ts @@ -377,4 +377,89 @@ describe("runDenseIntradayRetention (real Postgres)", () => { // Both raw samples survive — the intra-day shape is preserved. expect(live).toHaveLength(2); }); + + it("folds Health Connect raw heart rate past the window into its own hourly rows, beside Apple Health's", async () => { + const prisma = getPrismaClient(); + // Berlin UTC+2 on 1 May: 08:xx UTC is local hour 10 — far past 90 days. + const at = (minute: number) => + new Date(`2026-05-01T08:${String(minute).padStart(2, "0")}:00.000Z`); + await prisma.measurement.createMany({ + data: [ + ...[60, 70, 80].map((value, i) => ({ + userId: TEST_USER_ID, + type: "PULSE" as const, + value, + unit: "bpm", + source: "HEALTH_CONNECT" as const, + measuredAt: at(i * 10 + 1), + externalId: `hc:00000000-0000-0000-0000-00000000000a:${i}`, + })), + { + userId: TEST_USER_ID, + type: "PULSE" as const, + value: 100, + unit: "bpm", + source: "APPLE_HEALTH" as const, + measuredAt: at(5), + externalId: "hk-pulse-1", + }, + ], + }); + + const summary = await runDenseIntradayRetention(prisma, { + userId: TEST_USER_ID, + // The default 90-day window: the samples are well past it. + log: () => {}, + }); + expect(summary.totals.perSampleRowsSoftDeleted).toBe(4); + + const rows = await prisma.measurement.findMany({ + where: { userId: TEST_USER_ID, type: "PULSE" }, + orderBy: { source: "asc" }, + }); + // Raw rows of both sources are gone outright, not tombstoned. + expect(rows.every((row) => row.externalId?.startsWith("stats:"))).toBe( + true, + ); + expect(rows.every((row) => row.deletedAt === null)).toBe(true); + const id = "stats:HKQuantityTypeIdentifierHeartRate:2026-05-01T10"; + // One mean per source: the two sources never share a row. + expect(rows.map((row) => [row.source, row.externalId, row.value])).toEqual([ + ["APPLE_HEALTH", id, 100], + ["HEALTH_CONNECT", id, 70], + ]); + + // A second run finds nothing left to fold. + const again = await runDenseIntradayRetention(prisma, { + userId: TEST_USER_ID, + log: () => {}, + }); + expect(again.totals.daysConsolidated).toBe(0); + }); + + it("keeps Health Connect raw heart rate inside the window raw", async () => { + const prisma = getPrismaClient(); + const recent = new Date(Date.now() - 2 * 24 * 60 * 60 * 1000); + await prisma.measurement.create({ + data: { + userId: TEST_USER_ID, + type: "PULSE", + value: 64, + unit: "bpm", + source: "HEALTH_CONNECT", + measuredAt: recent, + externalId: "hc:00000000-0000-0000-0000-00000000000b:1", + }, + }); + const summary = await runDenseIntradayRetention(prisma, { + userId: TEST_USER_ID, + log: () => {}, + }); + expect(summary.totals.daysConsolidated).toBe(0); + expect( + await prisma.measurement.count({ + where: { userId: TEST_USER_ID, source: "HEALTH_CONNECT" }, + }), + ).toBe(1); + }); }); diff --git a/tests/integration/health-connect-import.test.ts b/tests/integration/health-connect-import.test.ts index 9cd102bc9..87cd33fd9 100644 --- a/tests/integration/health-connect-import.test.ts +++ b/tests/integration/health-connect-import.test.ts @@ -30,6 +30,7 @@ import { zipHealthConnectDb, } from "../fixtures/health-connect/build-fixture"; import { importHealthConnectExport } from "@/lib/import/health-connect/import"; +import { runDenseIntradayRetention } from "@/lib/measurements/dense-intraday-retention"; import { HEALTH_CONNECT_IMPORT_KIND, HEALTH_CONNECT_IMPORT_QUEUE, @@ -406,6 +407,53 @@ describe("Health Connect import, heart rate history", () => { }); }); +describe("Health Connect import, folded heart rate", () => { + it("folds raw heart rate once it ages out, and a later import does not put it back", async () => { + const prisma = getPrismaClient(); + const user = await createUser("hc-folded"); + // A day of minute-by-minute heart rate, 120 days back, imported while it + // was recent so it lands raw. + const { path, endUtc } = fixtureDb("folded", { + days: 1, + endUtc: today - 120 * DAY, + }); + const args = { + prisma, + dbPath: path, + userId: user.id, + userTimezone: "Europe/Berlin", + connectedIntegrations: [], + now: new Date(endUtc), + } as const; + await importHealthConnectExport(args); + expect((await countBy(user.id)).PULSE).toBe(24 * 60); + + // The nightly fold, at today's clock: the samples are past the window. + await runDenseIntradayRetention(prisma, { + userId: user.id, + log: () => {}, + }); + const folded = await prisma.measurement.findMany({ + where: { userId: user.id, type: "PULSE", source: "HEALTH_CONNECT" }, + }); + expect(folded).toHaveLength(24); + expect( + folded.every( + (row) => + row.deletedAt === null && + row.externalId?.startsWith( + "stats:HKQuantityTypeIdentifierHeartRate:", + ), + ), + ).toBe(true); + + // The same export again: every raw sample's hour is folded already. + const again = await importHealthConnectExport(args); + expect(again.skipped["PULSE::folded_window"]).toBe(24 * 60); + expect((await countBy(user.id)).PULSE).toBe(24); + }); +}); + describe("Health Connect import, worker", () => { async function stagedJob(name: string, zipBuilder: (zip: string) => void) { const prisma = getPrismaClient();