diff --git a/src/lib/measurements/import-apple-health-export.ts b/src/lib/measurements/import-apple-health-export.ts index 43608a994..dc280d285 100644 --- a/src/lib/measurements/import-apple-health-export.ts +++ b/src/lib/measurements/import-apple-health-export.ts @@ -122,11 +122,17 @@ export interface ImportJobResult { clinical: { skipped: number }; /** * Records left out because HealthLog wrote them into Apple Health itself and - * already holds them under their own source: `byMarker` carry the app's - * origin marker, `matchedManual` match a manual entry (same type and value, - * within 2 s) from before the marker existed. + * already holds them under their own source, each counted once however often + * the export repeats it: `byMarker` carry the app's origin marker, + * `byExternalId` carry a HealthLog row id as `HKExternalUUID` (samples from + * before the marker existed), `matchedManual` match a manual entry (same type + * and value, within 2 s) without either. */ - writtenByHealthLog: { byMarker: number; matchedManual: number }; + writtenByHealthLog: { + byMarker: number; + byExternalId: number; + matchedManual: number; + }; /** * v1.15.0 — reproductive HealthKit samples routed into CYCLE day-logs * (NOT Measurement). Absent / zeroed when the account has no cycle @@ -175,6 +181,29 @@ interface PreparedMeasurement { deviceType: string | null; } +/** + * A quantity `` read up to its open tag. Its contribution is committed + * at the close tag, once its `` children have been read. + */ +interface PendingRecord { + /** Queue the record; `healthLogId` is the `HKExternalUUID` it carried. */ + commit: (healthLogId: string | null) => void; + ownOrigin: boolean; + healthLogId: string | null; + /** The record's sample identity, computed only when it is left out. */ + sampleKey: () => string; +} + +/** Apple's export name for `HKMetadataKeyExternalUUID`. */ +const HK_EXTERNAL_UUID_METADATA_KEY = "HKExternalUUID"; + +/** The two halves of a blood pressure reading are one kind for an id match. */ +function kindOf(type: MeasurementType): string { + return type === "BLOOD_PRESSURE_SYS" || type === "BLOOD_PRESSURE_DIA" + ? "BLOOD_PRESSURE" + : type; +} + /** Flush row written to the `Workout` table. */ interface PreparedWorkout { userId: string; @@ -417,6 +446,8 @@ export async function streamParseExportXml( dayKey: string; rawValue: string | undefined; protectionUsed?: boolean; + sampleKey: string; + ownOrigin: boolean; } | null = null; // Cumulative fold: type -> local day -> hashed source identity -> subtotal. @@ -427,14 +458,27 @@ export async function streamParseExportXml( // A record's contribution is committed at its close tag, once its child // `` rows are read: one of them can mark the sample as written // by HealthLog itself, which has to keep it out of the import. - let pendingRecord: { commit: () => void; ownOrigin: boolean } | null = null; - const writtenByHealthLog = { byMarker: 0, matchedManual: 0 }; + let pendingRecord: PendingRecord | null = null; + const writtenByHealthLog = { byMarker: 0, byExternalId: 0, matchedManual: 0 }; + // An export can carry the same sample twice (a correlation's records also + // stand on their own), so a left-out sample is counted by its sample key. + const countedOwn = new Set(); + const countOwn = ( + reason: keyof typeof writtenByHealthLog, + sampleKey: string, + ): void => { + if (countedOwn.has(sampleKey)) return; + countedOwn.add(sampleKey); + writtenByHealthLog[reason] += 1; + }; + // The `HKExternalUUID` a queued spot row carried, checked at flush. + const healthLogIdOf = new Map(); // A `` (a blood pressure reading) wraps its records, and the // marker may sit on the correlation itself, before or after them. Its records // are held until it closes, then all kept or all left out together. let currentCorrelation: { ownOrigin: boolean; - held: Array<{ commit: () => void; ownOrigin: boolean }>; + held: PendingRecord[]; } | null = null; // Spot-row batch awaiting flush. const spotBatch: PreparedMeasurement[] = []; @@ -492,54 +536,134 @@ export async function streamParseExportXml( }; // ── Flush helpers ────────────────────────────────────────── - // A manual entry the app mirrored into Apple Health before it stamped its - // origin marker comes back in the export as a second copy of the reading. - // Leave it out when a MANUAL row of the same type and value sits within 2 s. - // Only MANUAL: Withings and import rows reach Apple Health through the app's - // mirror, which always sets the marker, so a value-and-time match against - // them would only add false positives. Same rule and tolerance as the sync - // path (`cross-source-merge.ts`). + // A sample the app wrote before it stamped its origin marker still carries + // the HealthLog row id as `HKExternalUUID`. A hit on one of this account's + // rows (deleted ones included: the sample is ours either way) of the same + // kind is an exact match. Blood pressure is one row per half, while the app + // stamps both samples of a reading with one id (the systolic row's, the + // diastolic row's in older builds), so either half matches either id. + const withoutHealthLogIds = async ( + rows: PreparedMeasurement[], + ): Promise => { + const ids = new Set(); + for (const row of rows) { + const id = healthLogIdOf.get(row); + if (id) ids.add(id); + } + if (ids.size === 0) return rows; + const known = await prisma.measurement.findMany({ + where: { userId, id: { in: [...ids] } }, + select: { id: true, type: true }, + }); + if (known.length === 0) return rows; + const typeOf = new Map(known.map((row) => [row.id, row.type])); + return rows.filter((row) => { + const id = healthLogIdOf.get(row); + const rowType = id ? typeOf.get(id) : undefined; + if (!rowType || kindOf(rowType) !== kindOf(row.type)) return true; + countOwn("byExternalId", row.externalId); + return false; + }); + }; + + // A manual entry mirrored with neither the marker nor an id comes back as a + // second copy of the reading. Leave it out when a MANUAL row of the same + // type and value sits within 2 s. Only MANUAL: Withings and import rows are + // written into Apple Health by the server-to-Health mirror, which has always + // stamped the row id, so a value-and-time match against them would only add + // false positives. Same rule and tolerance as the sync path + // (`cross-source-merge.ts`). + // + // The types that hold MANUAL rows are read once per import, so a batch of a + // type the account never typed in asks nothing, and a batch that does asks + // one time window per type rather than one clause per row. + let manualTypes: Promise> | null = null; const withoutManualMirrors = async ( rows: PreparedMeasurement[], ): Promise => { if (rows.length === 0) return rows; + manualTypes ??= prisma.measurement + .groupBy({ + by: ["type"], + where: { userId, source: "MANUAL", deletedAt: null }, + }) + .then((groups) => new Set(groups.map((group) => group.type))); + const typed = await manualTypes; + const windows = new Map(); + for (const row of rows) { + if (!typed.has(row.type)) continue; + const at = row.measuredAt.getTime(); + const window = windows.get(row.type); + if (!window) windows.set(row.type, { from: at, to: at }); + else { + if (at < window.from) window.from = at; + if (at > window.to) window.to = at; + } + } + if (windows.size === 0) return rows; const candidates = await prisma.measurement.findMany({ where: { userId, source: "MANUAL", deletedAt: null, - OR: rows.map((row) => ({ - type: row.type, + OR: [...windows].map(([type, window]) => ({ + type, measuredAt: { - gte: new Date(row.measuredAt.getTime() - MEASURED_AT_TOLERANCE_MS), - lte: new Date(row.measuredAt.getTime() + MEASURED_AT_TOLERANCE_MS), + gte: new Date(window.from - MEASURED_AT_TOLERANCE_MS), + lte: new Date(window.to + MEASURED_AT_TOLERANCE_MS), }, })), }, select: { type: true, source: true, value: true, measuredAt: true }, + orderBy: { measuredAt: "asc" }, }); if (candidates.length === 0) return rows; + const byType = new Map(); + for (const candidate of candidates) { + const list = byType.get(candidate.type); + if (list) list.push(candidate); + else byType.set(candidate.type, [candidate]); + } return rows.filter((row) => { - const mirrored = candidates.some((candidate) => - isSameReadingAcrossSource( + const list = byType.get(row.type); + if (!list) return true; + const at = row.measuredAt.getTime(); + // First candidate not earlier than the tolerance window. + let lo = 0; + let hi = list.length; + while (lo < hi) { + const mid = (lo + hi) >> 1; + if (list[mid].measuredAt.getTime() < at - MEASURED_AT_TOLERANCE_MS) + lo = mid + 1; + else hi = mid; + } + for (let i = lo; i < list.length; i++) { + if (list[i].measuredAt.getTime() > at + MEASURED_AT_TOLERANCE_MS) break; + const mirrored = isSameReadingAcrossSource( { type: row.type, source: "APPLE_HEALTH", value: row.value, measuredAt: row.measuredAt, }, - candidate, - ), - ); - if (mirrored) writtenByHealthLog.matchedManual += 1; - return !mirrored; + list[i], + ); + if (mirrored) { + countOwn("matchedManual", row.externalId); + return false; + } + } + return true; }); }; const flushSpotBatch = async (): Promise => { if (spotBatch.length === 0) return; const incoming = spotBatch.splice(0, spotBatch.length); - const chunk = await withoutManualMirrors(incoming); + const chunk = await withoutManualMirrors( + await withoutHealthLogIds(incoming), + ); + for (const row of incoming) healthLogIdOf.delete(row); for (const row of chunk) widenSpan(row.measuredAt); const insertedArrivals: Array<{ id: string; @@ -916,7 +1040,18 @@ export async function streamParseExportXml( Date.parse(attrs.endDate ?? attrs.startDate ?? ""), ) ? null - : { hkType, dayKey, rawValue: attrs.value }; + : { + hkType, + dayKey, + rawValue: attrs.value, + sampleKey: hashSampleKey( + hkType, + attrs.value ?? "", + attrs.startDate, + attrs.endDate, + ), + ownOrigin: false, + }; return; } @@ -978,7 +1113,15 @@ export async function streamParseExportXml( pendingRecord = { ownOrigin: false, - commit: () => { + healthLogId: null, + sampleKey: () => + hashSampleKey( + hkType, + attrs.value ?? "", + attrs.startDate, + attrs.endDate, + ), + commit: (healthLogId) => { if (CUMULATIVE_HK_TYPES.has(mapped.type)) { const dayKey = dayKeyForUserTz(mapped.takenAt, userTimezone); const sourceHash = hashCumulativeSourceIdentity( @@ -1007,7 +1150,7 @@ export async function streamParseExportXml( attrs.startDate, attrs.endDate, ); - spotBatch.push({ + const row: PreparedMeasurement = { userId, type: mapped.type, value: mapped.value, @@ -1017,7 +1160,9 @@ export async function streamParseExportXml( externalSourceVersion: attrs.sourceVersion ?? null, sleepStage: mapped.sleepStage ?? null, deviceType: null, - }); + }; + spotBatch.push(row); + if (healthLogId) healthLogIdOf.set(row, healthLogId); } }, }; @@ -1107,8 +1252,12 @@ export async function streamParseExportXml( if (isHealthLogOriginEntry(attrs.key, attrs.value)) { // On a record, or on the correlation that wraps it. if (pendingRecord) pendingRecord.ownOrigin = true; + else if (currentCycleRecord) currentCycleRecord.ownOrigin = true; else if (currentCorrelation) currentCorrelation.ownOrigin = true; } + if (pendingRecord && attrs.key === HK_EXTERNAL_UUID_METADATA_KEY) { + pendingRecord.healthLogId = attrs.value || null; + } // Attach the SexualActivity protection flag to the open cycle record. // Apple writes `HKMetadataKeySexualActivityProtectionUsed` with a // `"0"`/`"1"` (or `"true"`/`"false"`) value. @@ -1157,17 +1306,17 @@ export async function streamParseExportXml( const rec = pendingRecord; pendingRecord = null; if (currentCorrelation) currentCorrelation.held.push(rec); - else if (rec.ownOrigin) writtenByHealthLog.byMarker += 1; - else rec.commit(); + else if (rec.ownOrigin) countOwn("byMarker", rec.sampleKey()); + else rec.commit(rec.healthLogId); } if (tagName === "Correlation" && currentCorrelation) { const correlation = currentCorrelation; currentCorrelation = null; for (const rec of correlation.held) { if (correlation.ownOrigin || rec.ownOrigin) { - writtenByHealthLog.byMarker += 1; + countOwn("byMarker", rec.sampleKey()); } else { - rec.commit(); + rec.commit(rec.healthLogId); } } } @@ -1178,6 +1327,11 @@ export async function streamParseExportXml( if (tagName === "Record" && currentCycleRecord) { const rec = currentCycleRecord; currentCycleRecord = null; + if (rec.ownOrigin) { + // A day HealthLog mirrored into Apple Health from its own day-log. + countOwn("byMarker", rec.sampleKey); + return; + } const consumed = cycleAccumulator.consume( rec.hkType, rec.dayKey, diff --git a/tests/integration/apple-health-import-own-samples.test.ts b/tests/integration/apple-health-import-own-samples.test.ts index 8d48dc0b6..a24b663de 100644 --- a/tests/integration/apple-health-import-own-samples.test.ts +++ b/tests/integration/apple-health-import-own-samples.test.ts @@ -13,11 +13,17 @@ * - a record without it lands as before; * - the marker is read from a child `` that follows the * record's open tag, so the record is committed at its close tag; - * - a manual entry mirrored before the marker existed (a MANUAL row with the - * same type and value within 2 s) is left out and counted separately; + * - a sample from before the marker carries the HealthLog row id as + * `HKExternalUUID`; a hit on one of the account's rows of the same kind + * leaves it out, either id of a blood pressure pair covering both halves; + * - a manual entry mirrored with neither (a MANUAL row with the same type and + * value within 2 s) is left out and counted separately; * - a MANUAL row with another value, or 3 s away, is not a match; * - a Withings row of the same value and time is not a match either (those - * are mirrored with the marker, so a match would only add false positives). + * are mirrored with their row id, so a match would only add false + * positives); + * - the value match asks nothing for a type the account has no MANUAL rows + * of, and each sample left out is counted once. */ import { mkdtempSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; @@ -82,6 +88,7 @@ describe("a record HealthLog wrote itself (the origin marker)", () => { expect(rows.map((r) => r.value)).toEqual([80.2, 80.3]); expect(result.writtenByHealthLog).toEqual({ byMarker: 1, + byExternalId: 0, matchedManual: 0, }); }); @@ -243,6 +250,7 @@ describe("the synthetic export fixture", () => { expect(rows.map((r) => r.value)).toEqual([80.2, 80.4, 80.5]); expect(result.writtenByHealthLog).toEqual({ byMarker: 2, + byExternalId: 0, matchedManual: 0, }); }); @@ -286,6 +294,7 @@ describe("a manual entry mirrored before the marker existed", () => { expect(rows).toHaveLength(0); expect(result.writtenByHealthLog).toEqual({ byMarker: 0, + byExternalId: 0, matchedManual: 1, }); }); @@ -316,3 +325,248 @@ describe("a manual entry mirrored before the marker existed", () => { expect(result.writtenByHealthLog.matchedManual).toBe(0); }); }); + +/** + * Count every `measurement.findMany` the import issues against MANUAL rows, + * by wrapping the client the import is handed. + */ +function countingManualQueries() { + const manualQueries: Array<{ where?: Record }> = []; + const measurement = new Proxy(prisma.measurement, { + get(target, prop, receiver) { + const value = Reflect.get(target, prop, receiver); + if (prop !== "findMany" || typeof value !== "function") return value; + return (args: { where?: Record }) => { + if (args?.where?.source === "MANUAL") manualQueries.push(args); + return value.call(target, args); + }; + }, + }); + const client = new Proxy(prisma, { + get(target, prop, receiver) { + if (prop === "measurement") return measurement; + const value = Reflect.get(target, prop, receiver); + return typeof value === "function" ? value.bind(target) : value; + }, + }) as typeof prisma; + return { client, manualQueries }; +} + +describe("the manual-mirror lookup stays off the hot path", () => { + const AT = (i: number) => + `2026-05-14 ${String(8 + Math.floor(i / 60)).padStart(2, "0")}:${String(i % 60).padStart(2, "0")}:00 +0200`; + const heartRate = (value: number, at: string) => + ` `; + + async function runCounting(records: string, manualType?: "WEIGHT") { + const user = await prisma.user.create({ + data: { username: "own-hot", email: "hot@example.test", role: "USER" }, + }); + if (manualType) { + await prisma.measurement.create({ + data: { + userId: user.id, + type: manualType, + unit: "kg", + source: "MANUAL", + value: 80.4, + measuredAt: new Date("2026-05-14T06:00:00.000Z"), + }, + }); + } + const { client, manualQueries } = countingManualQueries(); + const result = await streamParseExportXml({ + xmlPath: writeXml(records), + userId: user.id, + userTimezone: "Europe/Berlin", + prisma: client, + spotBatchSize: 10, + }); + return { result, manualQueries, user }; + } + + it("asks nothing per batch when the account has no MANUAL rows", async () => { + const records = Array.from({ length: 40 }, (_, i) => + heartRate(60 + (i % 20), AT(i)), + ).join("\n"); + const { manualQueries, result } = await runCounting(records); + expect(manualQueries).toHaveLength(0); + expect(result.writtenByHealthLog.matchedManual).toBe(0); + }); + + it("asks nothing for a type without MANUAL rows, and one bounded window per batch for one with them", async () => { + const records = [ + ...Array.from({ length: 40 }, (_, i) => heartRate(60 + (i % 20), AT(i))), + weight(80.4, "2026-05-14 08:00:00 +0200"), + weight(80.6, "2026-05-14 09:00:00 +0200"), + ].join("\n"); + const { manualQueries, result, user } = await runCounting( + records, + "WEIGHT", + ); + // The heart rates ask nothing; the weights ask once, with one window for + // their type, not one clause per row. + expect(manualQueries).toHaveLength(1); + const or = manualQueries[0].where?.OR as Array<{ type: string }>; + expect(or).toHaveLength(1); + expect(or[0].type).toBe("WEIGHT"); + // The mirror is still caught; the other weight still lands. + expect(result.writtenByHealthLog.matchedManual).toBe(1); + const weights = await prisma.measurement.findMany({ + where: { userId: user.id, type: "WEIGHT", source: "APPLE_HEALTH" }, + }); + expect(weights.map((w) => w.value)).toEqual([80.6]); + }); +}); + +describe("a sample whose HKExternalUUID is a HealthLog row id", () => { + const AT = "2026-05-14 08:00:00 +0200"; + const ext = (id: string) => + ``; + + async function runWith( + seed: (userId: string) => Promise>, + records: (ids: Record) => string, + ) { + const user = await prisma.user.create({ + data: { username: "own-ext", email: "ext@example.test", role: "USER" }, + }); + const ids = await seed(user.id); + const result = await streamParseExportXml({ + xmlPath: writeXml(records(ids)), + userId: user.id, + userTimezone: "Europe/Berlin", + prisma, + }); + const rows = await prisma.measurement.findMany({ + where: { userId: user.id, source: "APPLE_HEALTH" }, + orderBy: [{ type: "asc" }], + }); + return { result, rows }; + } + + const seedRow = async ( + userId: string, + type: "WEIGHT" | "BLOOD_PRESSURE_SYS" | "BLOOD_PRESSURE_DIA", + value: number, + source: "MANUAL" | "WITHINGS" = "MANUAL", + ) => + ( + await prisma.measurement.create({ + data: { + userId, + type, + value, + unit: type === "WEIGHT" ? "kg" : "mmHg", + source, + // Ten minutes off the sample: only the id can tie them together. + measuredAt: new Date("2026-05-14T06:10:00.000Z"), + }, + }) + ).id; + + it("is left out without the marker, for a Withings row the value match never covers", async () => { + const { result, rows } = await runWith( + async (userId) => ({ + w: await seedRow(userId, "WEIGHT", 81, "WITHINGS"), + }), + (ids) => weight(80.4, AT, ext(ids.w)), + ); + expect(rows).toHaveLength(0); + expect(result.writtenByHealthLog).toEqual({ + byMarker: 0, + byExternalId: 1, + matchedManual: 0, + }); + }); + + it("leaves out both halves of a pressure pair stamped with either half's id", async () => { + const { result, rows } = await runWith( + async (userId) => ({ + sys: await seedRow(userId, "BLOOD_PRESSURE_SYS", 121), + dia: await seedRow(userId, "BLOOD_PRESSURE_DIA", 79), + }), + (ids) => + [ + // Current builds stamp both samples with the systolic id ... + bpRecord("Systolic", 121, AT, ext(ids.sys)), + bpRecord("Diastolic", 79, AT, ext(ids.sys)), + // ... older builds stamped both with the diastolic id. + bpRecord("Systolic", 122, "2026-05-14 09:00:00 +0200", ext(ids.dia)), + bpRecord("Diastolic", 80, "2026-05-14 09:00:00 +0200", ext(ids.dia)), + ].join("\n"), + ); + expect(rows).toHaveLength(0); + expect(result.writtenByHealthLog.byExternalId).toBe(4); + }); + + it("is imported when the id is unknown, belongs to another account, or names another kind", async () => { + const other = await prisma.user.create({ + data: { username: "own-ext-2", email: "ext2@example.test", role: "USER" }, + }); + const foreignId = await seedRow(other.id, "WEIGHT", 80.4); + const { result, rows } = await runWith( + async (userId) => ({ + sys: await seedRow(userId, "BLOOD_PRESSURE_SYS", 121), + }), + (ids) => + [ + weight(80.1, AT, ext("00000000-0000-4000-8000-000000000001")), + weight(80.2, "2026-05-14 09:00:00 +0200", ext(foreignId)), + weight(80.3, "2026-05-14 10:00:00 +0200", ext(ids.sys)), + ].join("\n"), + ); + expect(rows.map((r) => r.value)).toEqual([80.1, 80.2, 80.3]); + expect(result.writtenByHealthLog.byExternalId).toBe(0); + }); +}); + +describe("each sample left out is counted once", () => { + it("does not count a marked record twice when it appears top level and inside its correlation", async () => { + const AT = "2026-05-14 08:00:00 +0200"; + const sys = bpRecord("Systolic", 121, AT, `\n ${OWN}\n `); + const dia = bpRecord("Diastolic", 79, AT, `\n ${OWN}\n `); + const user = await prisma.user.create({ + data: { username: "own-once", email: "once@example.test", role: "USER" }, + }); + const result = await streamParseExportXml({ + xmlPath: writeXml( + [sys, dia, correlation([sys, dia].join("\n"))].join("\n"), + ), + userId: user.id, + userTimezone: "Europe/Berlin", + prisma, + }); + expect(result.writtenByHealthLog.byMarker).toBe(2); + }); +}); + +describe("a cycle sample HealthLog wrote itself", () => { + it("is left out of the day-log import", async () => { + const user = await prisma.user.create({ + data: { + username: "own-cycle", + email: "cycle@example.test", + role: "USER", + gender: "FEMALE", + timezone: "Europe/Berlin", + }, + }); + const flow = (day: string, children = "") => + ` ${children}`; + const result = await streamParseExportXml({ + xmlPath: writeXml( + [flow("2026-03-02", OWN), flow("2026-03-03")].join("\n"), + ), + userId: user.id, + userTimezone: "Europe/Berlin", + prisma, + }); + const days = await prisma.cycleDayLog.findMany({ + where: { userId: user.id }, + orderBy: { date: "asc" }, + }); + expect(days.map((d) => d.date)).toEqual(["2026-03-03"]); + expect(result.writtenByHealthLog.byMarker).toBe(1); + }); +});