diff --git a/src/__tests__/measurement-write-tail-guard.test.ts b/src/__tests__/measurement-write-tail-guard.test.ts index abca40d7d..23a600ddd 100644 --- a/src/__tests__/measurement-write-tail-guard.test.ts +++ b/src/__tests__/measurement-write-tail-guard.test.ts @@ -79,6 +79,8 @@ const WRITE_SITES: Record = { "a primitive; every caller runs the tail over the verdicts' dirty identities", "lib/measurements/import-apple-health-export.ts": "the import worker refolds the import's span and re-warms every imported type", + "lib/measurements/import-unchanged-rows.ts": + "the Apple export import's natural-key rescue, inside the import; its worker runs both legs once over the whole import", "lib/import/health-connect/import.ts": "the Health Connect import worker refolds the import's span and re-warms every imported type", "lib/export/restore-backup.ts": diff --git a/src/lib/measurements/import-apple-health-export.ts b/src/lib/measurements/import-apple-health-export.ts index d91f34fac..931e62f75 100644 --- a/src/lib/measurements/import-apple-health-export.ts +++ b/src/lib/measurements/import-apple-health-export.ts @@ -19,6 +19,10 @@ * * Locks per `.planning/research/v1434-r-1-xml-import.md` ยง6. */ +import { + adoptNaturalKeyTwin, + storedUnchangedKeys, +} from "@/lib/measurements/import-unchanged-rows"; import { createReadStream } from "node:fs"; import { createHash } from "node:crypto"; import { StringDecoder } from "node:string_decoder"; @@ -856,6 +860,15 @@ export async function streamParseExportXml( else createdByKey.set(key, [created]); } + // Rows already stored as they are: counted, not written (see module). + const unchanged = await storedUnchangedKeys( + prisma, + userId, + chunk, + (r, i) => + failedInsertIndexes.has(i) || + createdByKey.has(`${r.type}::${r.externalId}`), + ); for (let index = 0; index < chunk.length; index += 1) { const row = chunk[index]; const stat = bumpStat(row.type); @@ -868,6 +881,11 @@ export async function streamParseExportXml( } const key = `${row.type}::${row.externalId}`; + if (unchanged.has(key)) { + stat.updated += 1; + rowsUpserted += 1; + continue; + } const inserted = createdByKey.get(key)?.shift(); if (inserted) { stat.inserted += 1; @@ -906,38 +924,13 @@ export async function streamParseExportXml( err instanceof Prisma.PrismaClientKnownRequestError && (err.code === "P2002" || err.code === "P2025") ) { - try { - const twin = await prisma.measurement.findFirst({ - where: { - userId, - type: row.type, - source: "APPLE_HEALTH", - measuredAt: row.measuredAt, - sleepStage: row.sleepStage ?? null, - }, - select: { id: true }, - }); - if (twin) { - await prisma.measurement.update({ - where: { id: twin.id }, - data: { - value: row.value, - unit: row.unit, - externalId: row.externalId, - externalSourceVersion: row.externalSourceVersion, - deviceType: row.deviceType, - deletedAt: null, - }, - }); - stat.updated += 1; - rowsUpserted += 1; - } else { - unknown[`${row.type}::natural_key_unresolved`] = - (unknown[`${row.type}::natural_key_unresolved`] ?? 0) + 1; - } - } catch { - unknown[`${row.type}::natural_key_rescue_failed`] = - (unknown[`${row.type}::natural_key_rescue_failed`] ?? 0) + 1; + const adopted = await adoptNaturalKeyTwin(prisma, userId, row); + if (adopted === "updated") { + stat.updated += 1; + rowsUpserted += 1; + } else { + unknown[`${row.type}::${adopted}`] = + (unknown[`${row.type}::${adopted}`] ?? 0) + 1; } } else { unknown[`${row.type}::upsert_failed`] = diff --git a/src/lib/measurements/import-unchanged-rows.ts b/src/lib/measurements/import-unchanged-rows.ts new file mode 100644 index 000000000..b3cae79d3 --- /dev/null +++ b/src/lib/measurements/import-unchanged-rows.ts @@ -0,0 +1,124 @@ +/** + * Rows of an Apple Health import chunk that the account already holds with + * exactly the values the import's update would write. + * + * A re-import of an export the account already holds sent one UPDATE per + * row the insert skipped, about two million statements for a large export, + * nearly all of them writing back what was stored. The import reads the + * stored rows of a chunk once and leaves those out; it still counts them as + * updated. A key that occurs twice in the chunk is never left out: its rows + * are written in order, as before, so the last one stands. + */ +import type { + MeasurementType, + PrismaClient, + SleepStage, +} from "@/generated/prisma/client"; + +export interface ImportUpdateRow { + type: MeasurementType; + externalId: string; + value: number; + measuredAt: Date; + externalSourceVersion: string | null; + sleepStage?: SleepStage | null; + deviceType: string | null; +} + +/** `type::externalId` of every row whose update would change nothing. */ +export async function storedUnchangedKeys( + db: Pick, + userId: string, + chunk: readonly ImportUpdateRow[], + /** Rows the chunk already settled (inserted, or failed to insert). */ + settled: (row: ImportUpdateRow, index: number) => boolean = () => false, +): Promise> { + const out = new Set(); + const rows = chunk.filter((row, index) => !settled(row, index)); + if (rows.length === 0) return out; + const counts = new Map(); + for (const row of rows) { + const key = `${row.type}::${row.externalId}`; + counts.set(key, (counts.get(key) ?? 0) + 1); + } + const single = rows.filter( + (row) => counts.get(`${row.type}::${row.externalId}`) === 1, + ); + if (single.length === 0) return out; + const stored = await db.measurement.findMany({ + where: { + userId, + source: "APPLE_HEALTH", + externalId: { in: single.map((row) => row.externalId) }, + }, + select: { + type: true, + externalId: true, + value: true, + measuredAt: true, + externalSourceVersion: true, + sleepStage: true, + deviceType: true, + }, + }); + const byKey = new Map( + stored.map((row) => [`${row.type}::${row.externalId}`, row]), + ); + for (const row of single) { + const key = `${row.type}::${row.externalId}`; + const was = byKey.get(key); + if ( + was && + was.value === row.value && + was.measuredAt.getTime() === row.measuredAt.getTime() && + (was.externalSourceVersion ?? null) === + (row.externalSourceVersion ?? null) && + (was.sleepStage ?? null) === (row.sleepStage ?? null) && + (was.deviceType ?? null) === (row.deviceType ?? null) + ) { + out.add(key); + } + } + return out; +} + +/** + * The rescue when an imported row's update finds no row at its external id, + * or collides on the natural identity: a skipped insert can mean another + * external id holds the row's `(type, measuredAt, source, sleepStage)`. + * Adopt that row in place, as the old upsert rescue did. Answers what + * happened, as the import's per-type tally names it. + */ +export async function adoptNaturalKeyTwin( + db: Pick, + userId: string, + row: ImportUpdateRow & { unit: string }, +): Promise<"updated" | "natural_key_unresolved" | "natural_key_rescue_failed"> { + try { + const twin = await db.measurement.findFirst({ + where: { + userId, + type: row.type, + source: "APPLE_HEALTH", + measuredAt: row.measuredAt, + sleepStage: row.sleepStage ?? null, + }, + select: { id: true }, + }); + if (!twin) return "natural_key_unresolved"; + await db.measurement.update({ + where: { id: twin.id }, + data: { + value: row.value, + unit: row.unit, + externalId: row.externalId, + externalSourceVersion: row.externalSourceVersion, + deviceType: row.deviceType, + deletedAt: null, + }, + }); + return "updated"; + } catch { + return "natural_key_rescue_failed"; + } +} diff --git a/tests/integration/apple-health-reimport-unchanged.test.ts b/tests/integration/apple-health-reimport-unchanged.test.ts new file mode 100644 index 000000000..7e05324ed --- /dev/null +++ b/tests/integration/apple-health-reimport-unchanged.test.ts @@ -0,0 +1,111 @@ +/** + * Re-importing an Apple Health export the account already holds, against a + * real Postgres. + * + * Every row the insert skips used to cost one UPDATE writing back what was + * stored. A row whose stored values already equal what the update would + * write is now left out: the per-type counts are the same as before (each + * such row counts as updated), the stored rows are the same, and the import + * sends no UPDATE for them. A row whose value changed is still updated, and + * a key the chunk carries twice is still written in order. + */ +import { mkdtempSync, writeFileSync } from "node:fs"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; + +import { beforeEach, describe, expect, it } from "vitest"; + +import { streamParseExportXml } from "@/lib/measurements/import-apple-health-export"; + +import { makeUser } from "../kit/factories"; +import { countStatements } from "./query-counter"; +import { getPrismaClient, truncateAllTables } from "./setup"; + +const prisma = getPrismaClient(); + +function writeXml(records: string): string { + const dir = mkdtempSync(join(tmpdir(), "healthlog-reimport-")); + const path = join(dir, "export.xml"); + writeFileSync( + path, + ` +]> + +${records} +`, + ); + return path; +} + +function weights(values: number[]): string { + return values + .map( + (value, day) => ` `, + ) + .join("\n"); +} + +async function stored(userId: string) { + return prisma.measurement.findMany({ + where: { userId, type: "WEIGHT" }, + orderBy: { measuredAt: "asc" }, + select: { value: true, measuredAt: true, externalId: true }, + }); +} + +describe("Apple Health re-import of unchanged rows", () => { + let userId: string; + const values = Array.from({ length: 20 }, (_, i) => 70 + i / 10); + + beforeEach(async () => { + await truncateAllTables(prisma); + userId = (await makeUser()).id; + const first = await streamParseExportXml({ + xmlPath: writeXml(weights(values)), + userId, + userTimezone: "Europe/Berlin", + prisma, + }); + expect(first.perType.WEIGHT?.inserted).toBe(values.length); + }); + + it("counts every row as updated, as before, and sends no UPDATE", async () => { + const before = await stored(userId); + const measured = await countStatements(() => + streamParseExportXml({ + xmlPath: writeXml(weights(values)), + userId, + userTimezone: "Europe/Berlin", + prisma, + }), + ); + expect(measured.result.perType.WEIGHT?.inserted).toBe(0); + expect(measured.result.perType.WEIGHT?.updated).toBe(values.length); + expect(await stored(userId)).toEqual(before); + const updates = measured.statements.filter((sql) => + /^\s*UPDATE "public"\."measurements"/i.test(sql), + ); + expect(updates).toEqual([]); + }); + + it("still updates a row whose value changed", async () => { + const changed = [...values]; + changed[3] = 99.9; + const result = await streamParseExportXml({ + xmlPath: writeXml(weights(changed)), + userId, + userTimezone: "Europe/Berlin", + prisma, + }); + expect(result.perType.WEIGHT?.updated).toBe(values.length); + const rows = await stored(userId); + // The externalId hashes the sample, so a new value is a new sample of + // the same instant: it adopts the stored row through the natural key. + expect(rows).toHaveLength(values.length); + expect(rows[3]!.value).toBe(99.9); + }); +});