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();