perf(withings): batch the activity sync reads and writes

The activity sync ran a per-(entry, field) findFirst followed by an
update or create, roughly entries x 3 x 2 round-trips — about 2,190 on
a 365-day backfill. Flatten the entry/field grid into the concrete rows
the sync wants to land, resolve every overwrite-vs-insert decision in
one findMany keyed on the same four dedup columns, then run a single
createMany for fresh slots and one transaction for the changed-value
overwrites. A backfill is now a bounded handful of queries regardless of
the day count, and an unchanged re-sync is a no-op on the write path.

The overwrite contract is unchanged: an existing WITHINGS row for a
(type, measuredAt) slot keeps its row and only rewrites value; a fresh
slot inserts with the stable externalId. createMany uses skipDuplicates
so a concurrent sibling sync landing the same slot is tolerated by the
four-column unique.

src/lib/withings/__tests__/sync-activity.test.ts

@@ -7,18 +7,30 @@
  */
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 
-vi.mock("@/lib/db", () => ({
-  prisma: {
-    measurement: {
-      findFirst: vi.fn(),
-      create: vi.fn(),
-      update: vi.fn(),
+vi.mock("@/lib/db", () => {
+  // v1.20.2 F3 — the write path batches: one `findMany` resolves the
+  // overwrite-vs-insert decision for the whole sync, `createMany` lands
+  // every fresh slot, and changed-value overwrites run inside a
+  // `$transaction([...update])`. The mock `$transaction` awaits the array
+  // of update promises so the per-row `update` mocks still fire and the
+  // call counts assert the batched shape.
+  const measurement = {
+    findMany: vi.fn(),
+    createMany: vi.fn(),
+    update: vi.fn(),
+  };
+  return {
+    prisma: {
+      measurement,
+      withingsConnection: {
+        findUnique: vi.fn(),
+      },
+      $transaction: vi.fn((ops: unknown) =>
+        Array.isArray(ops) ? Promise.all(ops) : Promise.resolve(),
+      ),
     },
-    withingsConnection: {
-      findUnique: vi.fn(),
-    },
-  },
-}));
+  };
+});
 
 vi.mock("@/lib/integrations/status", () => ({
   isReauthRequired: vi.fn().mockResolvedValue(false),
@@ -155,22 +167,35 @@ describe("fetchWithingsActivity", () => {
   });
 });
 
+/**
+ * v1.20.2 F3 — helper: pull the flat `data[]` the batched `createMany`
+ * received (the writer always passes a single call with every fresh
+ * slot). Returns `[]` when `createMany` never fired.
+ */
+function createManyRows(): Array<Record<string, unknown>> {
+  const calls = vi.mocked(prisma.measurement.createMany).mock.calls;
+  if (calls.length === 0) return [];
+  return (calls[0][0] as { data: Array<Record<string, unknown>> }).data;
+}
+
 describe("syncUserActivity — field mapping + idempotency", () => {
   it("writes one row per (date, metric) the first time a day shows up", async () => {
     installFetchMock([
       { date: "2026-05-12", steps: 8420, distance: 6720, calories: 412 },
     ]);
-    vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
-    vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
+    // No existing slot rows → every planned row is an insert.
+    vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never);
+    vi.mocked(prisma.measurement.createMany).mockResolvedValue({
+      count: 3,
+    } as never);
 
     const imported = await syncUserActivity("user-1");
 
     expect(imported).toBe(3);
-    expect(prisma.measurement.create).toHaveBeenCalledTimes(3);
+    // v1.20.2 F3 — one batched insert carrying all three metric rows.
+    expect(prisma.measurement.createMany).toHaveBeenCalledTimes(1);
 
-    const types = vi
-      .mocked(prisma.measurement.create)
-      .mock.calls.map((c) => (c[0].data as { type: string }).type);
+    const types = createManyRows().map((r) => r.type as string);
     expect(types.sort()).toEqual([
       "ACTIVE_ENERGY_BURNED",
       "ACTIVITY_STEPS",
@@ -178,33 +203,67 @@ describe("syncUserActivity — field mapping + idempotency", () => {
     ]);
   });
 
-  it("updates instead of inserting when the same (date, metric) already exists", async () => {
+  it("updates instead of inserting when the same (date, metric) already exists with a different value", async () => {
     installFetchMock([
       { date: "2026-05-12", steps: 9001, distance: 7000, calories: 450 },
     ]);
-    vi.mocked(prisma.measurement.findFirst).mockResolvedValue({
-      id: "row-existing",
-    } as never);
+    const measuredAt = new Date("2026-05-12T12:00:00.000Z");
+    // Each slot already holds a row with a DIFFERENT value → all three
+    // overwrite (the value moved). Keyed by the same four columns the
+    // batched read matches on.
+    vi.mocked(prisma.measurement.findMany).mockResolvedValue([
+      { id: "row-steps", type: "ACTIVITY_STEPS", measuredAt, value: 1 },
+      {
+        id: "row-distance",
+        type: "WALKING_RUNNING_DISTANCE",
+        measuredAt,
+        value: 1,
+      },
+      {
+        id: "row-calories",
+        type: "ACTIVE_ENERGY_BURNED",
+        measuredAt,
+        value: 1,
+      },
+    ] as never);
     vi.mocked(prisma.measurement.update).mockResolvedValue({} as never);
 
     const imported = await syncUserActivity("user-1");
 
     expect(imported).toBe(3);
-    expect(prisma.measurement.create).not.toHaveBeenCalled();
+    expect(prisma.measurement.createMany).not.toHaveBeenCalled();
+    // One batched `$transaction` carrying three per-row updates.
+    expect(prisma.$transaction).toHaveBeenCalledTimes(1);
     expect(prisma.measurement.update).toHaveBeenCalledTimes(3);
   });
 
+  it("skips the write entirely when an existing slot already holds the same value (idempotent re-sync)", async () => {
+    installFetchMock([{ date: "2026-05-12", steps: 8420 }]);
+    const measuredAt = new Date("2026-05-12T12:00:00.000Z");
+    vi.mocked(prisma.measurement.findMany).mockResolvedValue([
+      { id: "row-steps", type: "ACTIVITY_STEPS", measuredAt, value: 8420 },
+    ] as never);
+
+    const imported = await syncUserActivity("user-1");
+
+    // The slot is still "touched" (rollup re-fold + imported count), but
+    // no value moved so neither the insert nor the update path runs.
+    expect(imported).toBe(1);
+    expect(prisma.measurement.createMany).not.toHaveBeenCalled();
+    expect(prisma.measurement.update).not.toHaveBeenCalled();
+    expect(prisma.$transaction).not.toHaveBeenCalled();
+  });
+
   it("anchors measuredAt at noon UTC so the instant lands inside the local day for every supported tz", async () => {
     installFetchMock([{ date: "2026-05-12", steps: 1000 }]);
-    vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
-    vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
+    vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never);
+    vi.mocked(prisma.measurement.createMany).mockResolvedValue({
+      count: 1,
+    } as never);
 
     await syncUserActivity("user-1");
 
-    const createArg = vi.mocked(prisma.measurement.create).mock.calls[0][0] as {
-      data: { measuredAt: Date };
-    };
-    const measuredAt = createArg.data.measuredAt;
+    const measuredAt = createManyRows()[0].measuredAt as Date;
     expect(measuredAt.toISOString()).toBe("2026-05-12T12:00:00.000Z");
 
     // Regression: anchoring at noon UTC keeps the row inside the
@@ -233,15 +292,15 @@ describe("syncUserActivity — field mapping + idempotency", () => {
     // `distance` absent — Withings drops the field rather than emitting 0
     // when there's no GPS. Steps + calories still write.
     installFetchMock([{ date: "2026-05-12", steps: 8000, calories: 380 }]);
-    vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
-    vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
+    vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never);
+    vi.mocked(prisma.measurement.createMany).mockResolvedValue({
+      count: 2,
+    } as never);
 
     const imported = await syncUserActivity("user-1");
 
     expect(imported).toBe(2);
-    const types = vi
-      .mocked(prisma.measurement.create)
-      .mock.calls.map((c) => (c[0].data as { type: string }).type);
+    const types = createManyRows().map((r) => r.type as string);
     expect(types).not.toContain("WALKING_RUNNING_DISTANCE");
   });
 
@@ -252,8 +311,10 @@ describe("syncUserActivity — field mapping + idempotency", () => {
     installFetchMock([
       { date: "2026-05-12", steps: 0, distance: 0, calories: 0 },
     ]);
-    vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
-    vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
+    vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never);
+    vi.mocked(prisma.measurement.createMany).mockResolvedValue({
+      count: 3,
+    } as never);
 
     const imported = await syncUserActivity("user-1");
     expect(imported).toBe(3);
@@ -261,8 +322,10 @@ describe("syncUserActivity — field mapping + idempotency", () => {
 
   it("calls recordSyncSuccess after a clean round-trip", async () => {
     installFetchMock([{ date: "2026-05-12", steps: 1000 }]);
-    vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
-    vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
+    vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never);
+    vi.mocked(prisma.measurement.createMany).mockResolvedValue({
+      count: 1,
+    } as never);
 
     await syncUserActivity("user-1");
     expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings");
@@ -270,18 +333,52 @@ describe("syncUserActivity — field mapping + idempotency", () => {
 
   it("tags every row with the canonical externalId so future replays dedup", async () => {
     installFetchMock([{ date: "2026-05-12", steps: 8420 }]);
-    vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
-    vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
+    vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never);
+    vi.mocked(prisma.measurement.createMany).mockResolvedValue({
+      count: 1,
+    } as never);
 
     await syncUserActivity("user-1");
 
-    const createArg = vi.mocked(prisma.measurement.create).mock.calls[0][0] as {
-      data: { externalId: string };
-    };
-    expect(createArg.data.externalId).toBe(
+    expect(createManyRows()[0].externalId).toBe(
       "withings:activity:user-1:2026-05-12:steps",
     );
   });
+
+  it("batches a multi-day backfill into one read and one insert (F3)", async () => {
+    // 5 days × 3 metrics = 15 planned rows. The pre-v1.20.2 shape ran
+    // 15 findFirst + 15 create = 30 round-trips; the batched path is one
+    // findMany + one createMany regardless of the day count.
+    installFetchMock(
+      Array.from({ length: 5 }, (_, i) => ({
+        date: `2026-05-1${i}`,
+        steps: 1000 + i,
+        distance: 2000 + i,
+        calories: 300 + i,
+      })),
+    );
+    vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never);
+    vi.mocked(prisma.measurement.createMany).mockResolvedValue({
+      count: 15,
+    } as never);
+
+    const imported = await syncUserActivity("user-1");
+
+    expect(imported).toBe(15);
+    expect(prisma.measurement.findMany).toHaveBeenCalledTimes(1);
+    expect(prisma.measurement.createMany).toHaveBeenCalledTimes(1);
+    expect(createManyRows()).toHaveLength(15);
+    // The single read is keyed by the full distinct (type, measuredAt)
+    // sets the batch will write — `skipDuplicates` guards the insert.
+    const readArg = vi.mocked(prisma.measurement.findMany).mock.calls[0][0] as {
+      where: { type: { in: unknown[] }; measuredAt: { in: unknown[] } };
+    };
+    expect(readArg.where.type.in).toHaveLength(3);
+    expect(readArg.where.measuredAt.in).toHaveLength(5);
+    const createArg = vi.mocked(prisma.measurement.createMany).mock
+      .calls[0][0] as { skipDuplicates: boolean };
+    expect(createArg.skipDuplicates).toBe(true);
+  });
 });
 
 describe("syncUserActivity — scope-skip guard (v1.4.26)", () => {
@@ -300,7 +397,7 @@ describe("syncUserActivity — scope-skip guard (v1.4.26)", () => {
 
     expect(imported).toBe(0);
     expect(fetchMock).not.toHaveBeenCalled();
-    expect(prisma.measurement.create).not.toHaveBeenCalled();
+    expect(prisma.measurement.createMany).not.toHaveBeenCalled();
     expect(recordSyncSuccess).not.toHaveBeenCalled();
   });
 
@@ -391,8 +488,10 @@ describe("syncUserActivity — measurement rollup hook (v1.4.39.1)", () => {
       { date: "2026-05-12", steps: 8420, distance: 6720, calories: 412 },
       { date: "2026-05-13", steps: 9100, distance: 7100, calories: 430 },
     ]);
-    vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
-    vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
+    vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never);
+    vi.mocked(prisma.measurement.createMany).mockResolvedValue({
+      count: 0,
+    } as never);
 
     await syncUserActivity("user-1");
 
@@ -418,8 +517,10 @@ describe("syncUserActivity — measurement rollup hook (v1.4.39.1)", () => {
     installFetchMock([
       { date: "2026-05-12", steps: 8420, distance: 6720, calories: 412 },
     ]);
-    vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
-    vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
+    vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never);
+    vi.mocked(prisma.measurement.createMany).mockResolvedValue({
+      count: 0,
+    } as never);
 
     await syncUserActivity("user-1");
 
@@ -431,8 +532,10 @@ describe("syncUserActivity — measurement rollup hook (v1.4.39.1)", () => {
     // hiccup must never fail the user's sync. The sync's `imported`
     // return value should still reflect the rows written.
     installFetchMock([{ date: "2026-05-12", steps: 8420 }]);
-    vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
-    vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
+    vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never);
+    vi.mocked(prisma.measurement.createMany).mockResolvedValue({
+      count: 0,
+    } as never);
     vi.mocked(recomputeBucketsForMeasurement).mockRejectedValueOnce(
       new Error("simulated rollup failure"),
     );

src/lib/withings/sync-activity.ts

@@ -316,6 +316,22 @@ export async function syncUserActivity(
   // See sync.ts header for the full rationale; the chart's
   // `source=rollup` fast-path otherwise misses Withings activity days.
   const touched: Array<{ type: MeasurementType; measuredAt: Date }> = [];
+
+  // v1.20.2 F3 — flatten (entry × field) into the concrete rows the sync
+  // wants to land, then batch the reads + writes. The pre-v1.20.2 shape
+  // ran a per-(entry, field) `findFirst` + `update`/`create`, i.e.
+  // ~entries × 3 × 2 round-trips (≈ 2,190 on a 365-day backfill). The
+  // overwrite contract is unchanged: an existing WITHINGS row for the
+  // `(type, measuredAt)` slot has its `value` rewritten (the four-column
+  // dedup keys on `(userId, type, measuredAt, source, sleepStage=null)`),
+  // a fresh slot is inserted with the stable `externalId`.
+  interface PlannedRow {
+    type: MeasurementType;
+    measuredAt: Date;
+    value: number;
+    externalId: string;
+  }
+  const planned: PlannedRow[] = [];
   for (const entry of entries) {
     if (!entry.date) continue;
     const measuredAt = activityMeasuredAt(entry.date);
@@ -325,42 +341,99 @@ export async function syncUserActivity(
       // valid (a day of rest still records 0 steps). Only skip
       // undefined / null / NaN.
       if (raw == null || !Number.isFinite(raw)) continue;
-      try {
-        const existing = await prisma.measurement.findFirst({
-          where: {
-            userId,
-            type,
-            measuredAt,
-            source: "WITHINGS",
-            sleepStage: null,
-          },
-          select: { id: true },
+      planned.push({
+        type,
+        measuredAt,
+        value: raw,
+        externalId: `withings:activity:${userId}:${entry.date}:${field}`,
+      });
+    }
+  }
+
+  if (planned.length > 0) {
+    const slotKey = (type: MeasurementType, measuredAt: Date) =>
+      `${type}|${measuredAt.getTime()}`;
+
+    try {
+      // One read for the whole batch: every existing WITHINGS / non-sleep
+      // row that occupies a slot we intend to write. Keyed by the same
+      // four columns the per-row `findFirst` matched on, so the
+      // overwrite-vs-insert decision is identical — just resolved in
+      // memory instead of N round-trips.
+      const distinctMeasuredAt = [
+        ...new Map(planned.map((p) => [p.measuredAt.getTime(), p.measuredAt])),
+      ].map(([, d]) => d);
+      const distinctTypes = [...new Set(planned.map((p) => p.type))];
+      const existingRows = await prisma.measurement.findMany({
+        where: {
+          userId,
+          source: "WITHINGS",
+          sleepStage: null,
+          type: { in: distinctTypes },
+          measuredAt: { in: distinctMeasuredAt },
+        },
+        select: { id: true, type: true, measuredAt: true, value: true },
+      });
+      const existingBySlot = new Map<string, { id: string; value: number }>();
+      for (const row of existingRows) {
+        existingBySlot.set(slotKey(row.type, row.measuredAt), {
+          id: row.id,
+          value: row.value,
         });
+      }
+
+      const toCreate: PlannedRow[] = [];
+      const toUpdate: Array<{ id: string; value: number }> = [];
+      for (const p of planned) {
+        const existing = existingBySlot.get(slotKey(p.type, p.measuredAt));
         if (existing) {
-          await prisma.measurement.update({
-            where: { id: existing.id },
-            data: { value: raw },
-          });
+          // Only write when the value actually moved — a re-sync of an
+          // unchanged day is then a pure no-op on the write path.
+          if (existing.value !== p.value) {
+            toUpdate.push({ id: existing.id, value: p.value });
+          }
         } else {
-          await prisma.measurement.create({
-            data: {
-              userId,
-              type,
-              value: raw,
-              unit: getUnitForType(type),
-              measuredAt,
-              source: "WITHINGS",
-              externalId: `withings:activity:${userId}:${entry.date}:${field}`,
-            },
-          });
+          toCreate.push(p);
         }
-        touched.push({ type, measuredAt });
+        touched.push({ type: p.type, measuredAt: p.measuredAt });
         imported++;
-      } catch (err) {
-        getEvent()?.addWarning(
-          `Failed to upsert activity row (${type}, ${entry.date}): ${err}`,
+      }
+
+      // One insert for every fresh slot. `skipDuplicates` tolerates a
+      // concurrent sibling sync landing the same slot between the read
+      // and this write (the four-column unique catches it).
+      if (toCreate.length > 0) {
+        await prisma.measurement.createMany({
+          data: toCreate.map((p) => ({
+            userId,
+            type: p.type,
+            value: p.value,
+            unit: getUnitForType(p.type),
+            measuredAt: p.measuredAt,
+            source: "WITHINGS" as const,
+            externalId: p.externalId,
+          })),
+          skipDuplicates: true,
+        });
+      }
+
+      // Values differ per row, so the overwrites can't collapse into one
+      // `updateMany`; run them in a single transaction so the batch is one
+      // round-trip rather than one per row.
+      if (toUpdate.length > 0) {
+        await prisma.$transaction(
+          toUpdate.map((u) =>
+            prisma.measurement.update({
+              where: { id: u.id },
+              data: { value: u.value },
+            }),
+          ),
         );
       }
+    } catch (err) {
+      getEvent()?.addWarning(
+        `Failed to batch-upsert ${planned.length} activity rows for ${userId}: ${err}`,
+      );
     }
   }