perf(import): skip the update of a re-imported row that is already stored

Re-importing an Apple Health export the account already holds sent one
UPDATE per row the insert skipped, about two million statements for a
large export, nearly all writing back what was stored. One read per chunk
now finds the rows whose stored values equal what the update would write;
they count as updated as before, without the statement. A key the chunk
carries twice is written in order as before. The natural-key rescue moves
into the same module.

src/__tests__/measurement-write-tail-guard.test.ts

@@ -79,6 +79,8 @@ const WRITE_SITES: Record<string, string> = {
     "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":

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`] =

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<PrismaClient, "measurement">,
+  userId: string,
+  chunk: readonly ImportUpdateRow[],
+  /** Rows the chunk already settled (inserted, or failed to insert). */
+  settled: (row: ImportUpdateRow, index: number) => boolean = () => false,
+): Promise<Set<string>> {
+  const out = new Set<string>();
+  const rows = chunk.filter((row, index) => !settled(row, index));
+  if (rows.length === 0) return out;
+  const counts = new Map<string, number>();
+  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<string, (typeof stored)[number]>(
+    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<PrismaClient, "measurement">,
+  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";
+  }
+}