perf(withings): write a measure sync in sets and refold only what changed

A manual full sync re-reads the whole history, and nearly all of it is
already stored. Each reading cost a findFirst plus an update, and every
re-read day was marked dirty, so the run refolded the DAY tier and queued
three coarse recomputes for every day of the account: about thirty
seconds of database work behind 1.4 s of Withings calls.

The measures leg now reads the stored rows of the window once, skips the
unchanged ones, updates the changed or tombstoned ones and inserts the
new ones in batches. Only inserted or changed readings count as touched.
recomputeBucketsForKeys refolds exactly the touched DAY buckets in one
aggregate and one transaction, and queues one coarse job per run of
adjacent buckets instead of three per day.

src/lib/rollups/measurement-rollups.ts

@@ -358,6 +358,143 @@ export async function recomputeBucketsForMeasurement(
   );
 }
 
+/** Keys one set-based DAY refold carries; bounds the statement's arrays. */
+const KEYED_REFOLD_CHUNK = 2_000;
+
+/**
+ * The set-based sibling of {@link recomputeBucketsForMeasurement}, for a write
+ * that touched many `(type, day)` buckets at once (a sync, an import, a batch).
+ *
+ * The per-key hook costs an aggregate, a transaction and three queue sends per
+ * bucket, so a sync that re-read two years of history paid that seven thousand
+ * times over. Here the DAY tier is refolded for exactly the touched buckets in
+ * one aggregate and one transaction per chunk, and the coarser tiers are
+ * enqueued once per contiguous run of buckets rather than once per day. The
+ * result is the one the per-key loop produced: every touched day re-derived
+ * from its live rows, a day left empty dropped, untouched days left alone.
+ */
+export async function recomputeBucketsForKeys(
+  userId: string,
+  rows: Array<{ type: MeasurementType; measuredAt: Date }>,
+): Promise<void> {
+  const keys = collapseToTypeDayKeys(rows);
+  for (let i = 0; i < keys.length; i += KEYED_REFOLD_CHUNK) {
+    await refoldDayBuckets(userId, keys.slice(i, i + KEYED_REFOLD_CHUNK));
+  }
+  for (const granularity of ASYNC_GRANULARITIES) {
+    for (const run of bucketRuns(keys, granularity)) {
+      await enqueueRollupRecompute({ userId, granularity, ...run });
+    }
+  }
+}
+
+async function refoldDayBuckets(
+  userId: string,
+  keys: Array<{ type: MeasurementType; measuredAt: Date }>,
+): Promise<void> {
+  if (keys.length === 0) return;
+  const types = [...new Set(keys.map((k) => k.type))];
+  const days = [...new Set(keys.map((k) => k.measuredAt.getTime()))].sort(
+    (a, b) => a - b,
+  );
+  const wanted = new Set(
+    keys.map((k) => `${k.type}|${k.measuredAt.getTime()}`),
+  );
+  const aggregate = await runRollupAggregate({
+    userId,
+    types,
+    granularity: "DAY",
+    from: new Date(days[0]!),
+    to: new Date(days[days.length - 1]! + 24 * 60 * 60 * 1000),
+    days: days.map((d) => new Date(d)),
+  });
+  // The day filter is per day, not per (type, day): another touched type's
+  // day comes back too. Its aggregate would be right, but it was not touched,
+  // so it is not rewritten.
+  const ordered = aggregate
+    .filter((r) =>
+      wanted.has(`${r.type}|${new Date(r.bucket_start).getTime()}`),
+    )
+    .sort((a, b) =>
+      a.type !== b.type
+        ? a.type < b.type
+          ? -1
+          : 1
+        : a.source !== b.source
+          ? a.source < b.source
+            ? -1
+            : 1
+          : a.bucket_start.getTime() - b.bucket_start.getTime(),
+    );
+  // Delete exactly the touched buckets (every source, so a source that left a
+  // day does not strand its row, and a day that emptied out is dropped), then
+  // upsert the fresh aggregate, in one transaction like `persistRollupRows`.
+  await prisma.$transaction(
+    async (tx) => {
+      await tx.$executeRawUnsafe(
+        `DELETE FROM measurement_rollups
+          WHERE user_id = $1
+            AND granularity = 'DAY'::"measurement_rollup_granularity"
+            AND (type, bucket_start) IN (
+              SELECT k.t::"measurement_type", k.b
+                FROM unnest($2::text[], $3::timestamptz[]) AS k(t, b)
+            )`,
+        userId,
+        keys.map((k) => k.type),
+        keys.map((k) => k.measuredAt.toISOString()),
+      );
+      for (let i = 0; i < ordered.length; i += 500) {
+        const { sql, params } = buildRollupUpsert(
+          userId,
+          "DAY",
+          ordered.slice(i, i + 500),
+        );
+        await tx.$executeRawUnsafe(sql, ...params);
+      }
+    },
+    { maxWait: 10_000, timeout: 120_000 },
+  );
+}
+
+/**
+ * The coarse buckets the touched days fall in, merged per type into runs of
+ * adjacent buckets: one recompute job per run, each covering whole buckets.
+ */
+function bucketRuns(
+  keys: Array<{ type: MeasurementType; measuredAt: Date }>,
+  granularity: RollupGranularity,
+): Array<{ type: MeasurementType; from: Date; to: Date }> {
+  const spans = new Map<
+    string,
+    { type: MeasurementType; from: Date; to: Date }
+  >();
+  for (const k of keys) {
+    const span = bucketSpan(k.measuredAt, granularity);
+    spans.set(`${k.type}|${span.from.getTime()}`, { type: k.type, ...span });
+  }
+  const sorted = [...spans.values()].sort((a, b) =>
+    a.type !== b.type
+      ? a.type < b.type
+        ? -1
+        : 1
+      : a.from.getTime() - b.from.getTime(),
+  );
+  const runs: Array<{ type: MeasurementType; from: Date; to: Date }> = [];
+  for (const span of sorted) {
+    const last = runs[runs.length - 1];
+    if (
+      last &&
+      last.type === span.type &&
+      last.to.getTime() === span.from.getTime()
+    ) {
+      last.to = span.to;
+    } else {
+      runs.push({ ...span });
+    }
+  }
+  return runs;
+}
+
 /**
  * Best-effort pg-boss enqueue. Silent no-op when no boss is attached
  * (route is running in a context without the worker — typically the
@@ -399,7 +536,16 @@ export async function enqueueRollupRecompute(input: {
     // is still QUEUED — safe here precisely because the handler re-reads the
     // bucket's live rows when it starts, so the queued job still sees writes
     // that land after it was enqueued.
-    singletonKey: `${input.userId}|${input.type}|${input.granularity}|${input.from.toISOString()}`,
+    //
+    // A job covering a run of several buckets (`recomputeBucketsForKeys`)
+    // carries its end in the key too: under the bare start it would collapse
+    // into a queued single-bucket job and the rest of the run would be lost.
+    singletonKey:
+      `${input.userId}|${input.type}|${input.granularity}|${input.from.toISOString()}` +
+      (bucketSpan(input.from, input.granularity).to.getTime() ===
+      input.to.getTime()
+        ? ""
+        : `|${input.to.toISOString()}`),
   });
 }
 
@@ -480,10 +626,21 @@ async function runRollupAggregate(
     granularity: RollupGranularity;
     from: Date;
     to: Date;
+    /**
+     * Only the readings of these UTC days, inside `[from, to)`. A keyed
+     * refold (`recomputeBucketsForKeys`) passes the touched days so a handful
+     * spread across years does not aggregate everything in between.
+     */
+    days?: Date[];
   },
   db: Pick<typeof prisma, "$queryRawUnsafe"> = prisma,
 ): Promise<RollupRow[]> {
   if (input.types && input.types.length === 0) return [];
+  // Bound as the fourth parameter, never spliced.
+  const dayFilter = input.days
+    ? `AND date_trunc('day', m."measured_at") = ANY($4::timestamptz[])`
+    : "";
+  const dayParams = input.days ? [input.days.map((d) => d.toISOString())] : [];
 
   // Postgres rejects `date_trunc($1::text, m."measured_at")` in a
   // GROUP BY because the planner sees the parameterised expression
@@ -553,6 +710,7 @@ async function runRollupAggregate(
         AND m."measured_at" >= $2
         AND m."measured_at" <  $3
         AND m."deleted_at" IS NULL
+        ${dayFilter}
       GROUP BY m."type", m."source", ${dateTrunc}
     `;
     return applyHourlyMeanDay(
@@ -563,6 +721,7 @@ async function runRollupAggregate(
         input.userId,
         input.from,
         input.to,
+        ...dayParams,
       ),
     );
   }
@@ -600,6 +759,7 @@ async function runRollupAggregate(
       AND m."measured_at" >= $2
       AND m."measured_at" <  $3
       AND m."deleted_at" IS NULL
+      ${dayFilter}
     GROUP BY m."type", m."source", ${dateTrunc}
   `;
   return applyHourlyMeanDay(
@@ -610,6 +770,7 @@ async function runRollupAggregate(
       input.userId,
       input.from,
       input.to,
+      ...dayParams,
     ),
   );
 }
@@ -635,6 +796,7 @@ async function applyHourlyMeanDay(
     granularity: RollupGranularity;
     from: Date;
     to: Date;
+    days?: Date[];
   },
   rows: RollupRow[],
 ): Promise<RollupRow[]> {
@@ -685,6 +847,11 @@ async function applyHourlyMeanDay(
           AND m."measured_at" >= $2
           AND m."measured_at" <  $3
           AND m."deleted_at" IS NULL
+          ${
+            input.days
+              ? `AND date_trunc('day', m."measured_at") = ANY($4::timestamptz[])`
+              : ""
+          }
         GROUP BY m."type", m."source", date_trunc('hour', m."measured_at")
       ) h
       GROUP BY 1, 2, 3
@@ -694,6 +861,7 @@ async function applyHourlyMeanDay(
     input.userId,
     input.from,
     input.to,
+    ...(input.days ? [input.days.map((d) => d.toISOString())] : []),
   );
   const byKey = new Map(
     means.map((m) => [

src/lib/withings/__tests__/sync-measure-watermark.test.ts

@@ -16,8 +16,8 @@ import { beforeEach, describe, expect, it, vi } from "vitest";
 vi.mock("@/lib/db", () => ({
   prisma: {
     measurement: {
-      findFirst: vi.fn(),
-      create: vi.fn(),
+      findMany: vi.fn(),
+      createManyAndReturn: vi.fn(),
       update: vi.fn(),
     },
     withingsConnection: {
@@ -66,7 +66,7 @@ vi.mock("@/lib/rollups/measurement-rollups", async () => {
   >("@/lib/rollups/measurement-rollups");
   return {
     ...actual,
-    recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined),
+    recomputeBucketsForKeys: vi.fn().mockResolvedValue(undefined),
   };
 });
 
@@ -96,10 +96,6 @@ import {
 
 import { getValidToken, syncUserMeasurements } from "../sync";
 
-const P2002 = Object.assign(new Error("Unique constraint failed"), {
-  code: "P2002",
-});
-
 beforeEach(() => {
   vi.clearAllMocks();
   vi.mocked(prisma.$transaction).mockImplementation(async (run) =>
@@ -116,8 +112,15 @@ beforeEach(() => {
     lastSyncedAt: new Date(Date.now() - 60 * 60 * 1000),
   } as never);
   vi.mocked(prisma.withingsConnection.update).mockResolvedValue({} as never);
-  vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
-  vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
+  vi.mocked(prisma.measurement.findMany).mockResolvedValue([]);
+  vi.mocked(prisma.measurement.createManyAndReturn).mockImplementation(
+    (async (args: { data: Array<{ type: string; measuredAt: Date }> }) =>
+      args.data.map((row, i) => ({
+        id: `m-${i}`,
+        type: row.type,
+        measuredAt: row.measuredAt,
+      }))) as never,
+  );
   vi.mocked(prisma.measurement.update).mockResolvedValue({} as never);
 });
 
@@ -168,10 +171,20 @@ describe("syncUserMeasurements — watermark hold on hard row failure (F4)", ()
         measuredAt: new Date("2026-05-16T08:05:00Z"),
       },
     ] as never);
-    // The first write hard-fails (a transient DB error), the second succeeds.
-    vi.mocked(prisma.measurement.create)
-      .mockRejectedValueOnce(new Error("connection reset by peer"))
-      .mockResolvedValue({} as never);
+    // The weight is stored with an older value; its update hard-fails (a
+    // transient DB error). The pulse is new and inserts.
+    vi.mocked(prisma.measurement.findMany).mockResolvedValue([
+      {
+        id: "m-weight",
+        type: "WEIGHT",
+        measuredAt: new Date("2026-05-16T08:00:00Z"),
+        value: 79,
+        deletedAt: null,
+      },
+    ] as never);
+    vi.mocked(prisma.measurement.update).mockRejectedValueOnce(
+      new Error("connection reset by peer"),
+    );
 
     const result = await syncUserMeasurements("user-1");
 
@@ -187,7 +200,7 @@ describe("syncUserMeasurements — watermark hold on hard row failure (F4)", ()
     );
   });
 
-  it("does NOT hold the watermark for a benign P2002 idempotent-write collision", async () => {
+  it("does NOT hold the watermark for a duplicate a concurrent run inserted first", async () => {
     vi.mocked(fetchMeasurements).mockResolvedValue([
       {
         type: "WEIGHT",
@@ -195,11 +208,15 @@ describe("syncUserMeasurements — watermark hold on hard row failure (F4)", ()
         measuredAt: new Date("2026-05-16T08:00:00Z"),
       },
     ] as never);
-    vi.mocked(prisma.measurement.create).mockRejectedValueOnce(P2002);
+    // A concurrent run inserted the same reading first: the batched insert
+    // skips it as a duplicate.
+    vi.mocked(prisma.measurement.createManyAndReturn).mockResolvedValueOnce(
+      [] as never,
+    );
 
     const { imported } = await syncUserMeasurements("user-1");
 
-    // P2002 means the row is already present — nothing lost — so the cycle is
+    // A skipped duplicate is already present — nothing lost — so the cycle is
     // still healthy: stamp the watermark + success, record no failure.
     expect(imported).toBe(0);
     expect(prisma.withingsConnection.update).toHaveBeenCalledTimes(1);

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

@@ -10,8 +10,8 @@ import { beforeEach, describe, expect, it, vi } from "vitest";
 vi.mock("@/lib/db", () => ({
   prisma: {
     measurement: {
-      findFirst: vi.fn(),
-      create: vi.fn(),
+      findMany: vi.fn(),
+      createManyAndReturn: vi.fn(),
       update: vi.fn(),
     },
     withingsConnection: {
@@ -61,7 +61,7 @@ vi.mock("@/lib/rollups/measurement-rollups", async () => {
   >("@/lib/rollups/measurement-rollups");
   return {
     ...actual,
-    recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined),
+    recomputeBucketsForKeys: vi.fn().mockResolvedValue(undefined),
   };
 });
 
@@ -179,8 +179,10 @@ function persistedLegs(): Array<{ leg: unknown; value: unknown }> {
 
 function createdTypes(): string[] {
   return vi
-    .mocked(prisma.measurement.create)
-    .mock.calls.map((c) => (c[0] as { data: { type: string } }).data.type);
+    .mocked(prisma.measurement.createManyAndReturn)
+    .mock.calls.flatMap((c) =>
+      (c[0] as { data: Array<{ type: string }> }).data.map((d) => d.type),
+    );
 }
 
 beforeEach(() => {
@@ -193,14 +195,15 @@ beforeEach(() => {
     connection() as never,
   );
   vi.mocked(prisma.withingsConnection.update).mockResolvedValue({} as never);
-  vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
-  vi.mocked(prisma.measurement.create).mockImplementation((async (args: {
-    data: { type: string; measuredAt: Date };
-  }) => ({
-    id: `m-${args.data.type}`,
-    type: args.data.type,
-    measuredAt: args.data.measuredAt,
-  })) as never);
+  vi.mocked(prisma.measurement.findMany).mockResolvedValue([]);
+  vi.mocked(prisma.measurement.createManyAndReturn).mockImplementation(
+    (async (args: { data: Array<{ type: string; measuredAt: Date }> }) =>
+      args.data.map((d) => ({
+        id: `m-${d.type}`,
+        type: d.type,
+        measuredAt: d.measuredAt,
+      }))) as never,
+  );
   vi.mocked(prisma.$executeRaw).mockResolvedValue(1 as never);
 });
 
@@ -338,7 +341,7 @@ describe("syncUserMeasurements inside the Withings API plan", () => {
 
     for (let run = 0; run < 4; run++) {
       vi.mocked(prisma.$executeRaw).mockClear();
-      vi.mocked(prisma.measurement.create).mockClear();
+      vi.mocked(prisma.measurement.createManyAndReturn).mockClear();
       requestedStartdates.length = 0;
       await syncUserMeasurements("user-1");
       const [persisted] = persistedLegs();

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

@@ -14,10 +14,9 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 vi.mock("@/lib/db", () => ({
   prisma: {
     measurement: {
-      findFirst: vi.fn(),
       findMany: vi.fn(),
-      create: vi.fn(),
       createMany: vi.fn(),
+      createManyAndReturn: vi.fn(),
       update: vi.fn(),
     },
     withingsConnection: {
@@ -67,6 +66,7 @@ vi.mock("@/lib/rollups/measurement-rollups", async () => {
   return {
     ...actual,
     recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined),
+    recomputeBucketsForKeys: vi.fn().mockResolvedValue(undefined),
   };
 });
 
@@ -125,13 +125,15 @@ beforeEach(() => {
     lastSyncedAt: new Date(Date.now() - 60 * 60 * 1000),
   } as never);
   vi.mocked(prisma.withingsConnection.update).mockResolvedValue({} as never);
-  vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
   vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never);
-  vi.mocked(prisma.measurement.create).mockResolvedValue({
-    id: "m-1",
-    type: "PULSE",
-    measuredAt: new Date(),
-  } as never);
+  vi.mocked(prisma.measurement.createManyAndReturn).mockImplementation(
+    (async (args: { data: Array<{ type: string; measuredAt: Date }> }) =>
+      args.data.map((d, i) => ({
+        id: `m-${i}`,
+        type: d.type,
+        measuredAt: d.measuredAt,
+      }))) as never,
+  );
   vi.mocked(prisma.measurement.createMany).mockResolvedValue({
     count: 1,
   } as never);
@@ -157,7 +159,11 @@ describe("syncUserMeasurements — plausibility gate", () => {
     );
 
     expect(result).toEqual({ imported: 1, failed: false });
-    expect(prisma.measurement.create).toHaveBeenCalledTimes(1);
+    expect(prisma.measurement.createManyAndReturn).toHaveBeenCalledTimes(1);
+    expect(
+      vi.mocked(prisma.measurement.createManyAndReturn).mock.calls[0]?.[0]
+        ?.data,
+    ).toHaveLength(1);
     expect(tally()).toBeUndefined();
   });
 
@@ -182,7 +188,11 @@ describe("syncUserMeasurements — plausibility gate", () => {
     // Only the real reading landed, and the run is still a clean success: one
     // impossible sample must not strand the rest behind a held watermark.
     expect(result).toEqual({ imported: 1, failed: false });
-    expect(prisma.measurement.create).toHaveBeenCalledTimes(1);
+    expect(prisma.measurement.createManyAndReturn).toHaveBeenCalledTimes(1);
+    expect(
+      vi.mocked(prisma.measurement.createManyAndReturn).mock.calls[0]?.[0]
+        ?.data,
+    ).toHaveLength(1);
     expect(tally()).toEqual({
       total: 1,
       buckets: [
@@ -196,9 +206,15 @@ describe("syncUserMeasurements — plausibility gate", () => {
     // The slot already holds a real pulse; the provider re-sends the same
     // instant carrying a decode slip. Ungated, the update branch rewrote the
     // good value with the bad one.
-    vi.mocked(prisma.measurement.findFirst).mockResolvedValue({
-      id: "existing-row",
-    } as never);
+    vi.mocked(prisma.measurement.findMany).mockResolvedValueOnce([
+      {
+        id: "existing-row",
+        type: "PULSE",
+        measuredAt: new Date("2026-05-16T08:00:00Z"),
+        value: 61,
+        deletedAt: null,
+      },
+    ] as never);
     vi.mocked(fetchMeasurements).mockResolvedValue([
       {
         type: "PULSE",

src/lib/withings/sync.ts

@@ -29,7 +29,6 @@ import {
 } from "./api-plan";
 import { getUserWithingsCredentials } from "./credentials";
 import { annotate, getEvent } from "@/lib/logging/context";
-import { isP2002 } from "@/lib/prisma-errors";
 import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit";
 import { dropImplausibleMeasurements } from "@/lib/measurements/plausibility-gate";
 import {
@@ -45,7 +44,7 @@ import {
 } from "./response-classifier";
 import {
   collapseToTypeDayKeys,
-  recomputeBucketsForMeasurement,
+  recomputeBucketsForKeys,
 } from "@/lib/rollups/measurement-rollups";
 import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate";
 import {
@@ -349,7 +348,8 @@ export async function syncUserMeasurements(
   // through Withings. We collect first, fold once at the end so a 100-
   // row monthly catch-up costs at most ~N (type, day) recomputes rather
   // than N per-row hooks. Best-effort: a populator hiccup never fails
-  // the user's sync.
+  // the user's sync. Only readings this run inserted or changed count as
+  // touched: a re-read reading that is already stored dirties nothing.
   const touched: Array<{ type: MeasurementType; measuredAt: Date }> = [];
   const insertedArrivals: Array<{
     id: string;
@@ -367,69 +367,118 @@ export async function syncUserMeasurements(
   let anyRowFailed = false;
   let firstRowError: unknown = null;
 
+  // The run reads a window (the whole history on a full sync) and most of it
+  // is usually stored already. One read of the stored rows in that window
+  // splits it three ways: unchanged (nothing to write, nothing to refold),
+  // changed or tombstoned (an update), and new (one batched insert). The
+  // per-row `findFirst` + `update` this replaced cost two round trips per
+  // reading and marked every re-read day dirty, which is where a full sync's
+  // forty seconds went.
+  const latest = new Map<string, (typeof measures)[number]>();
   for (const m of measures) {
-    const measType = m.type as MeasurementType;
-    try {
-      // v1.4.25 — Migration 0055 added `sleepStage` to the
-      // composite unique with `NULLS NOT DISTINCT`, so a non-sleep
-      // upsert keyed on (userId, type, measuredAt, source) with
-      // sleepStage IS NULL is still safe. But Prisma's typed compound
-      // input requires a non-null `sleepStage`, so we model the
-      // idempotent write as a `findFirst` + `create`/`update`. The
-      // unique index still serializes concurrent inserts, so the worst
-      // case is a Prisma P2002 we catch and ignore.
-      const existing = await prisma.measurement.findFirst({
-        where: {
-          userId,
-          type: measType,
-          measuredAt: m.measuredAt,
-          source: "WITHINGS",
-          sleepStage: null,
+    latest.set(`${m.type}|${m.measuredAt.getTime()}`, m);
+  }
+  const incoming = [...latest.values()];
+  const existingByKey = new Map<
+    string,
+    { id: string; value: number; deletedAt: Date | null }
+  >();
+  if (incoming.length > 0) {
+    let first = incoming[0]!.measuredAt;
+    let last = first;
+    for (const m of incoming) {
+      if (m.measuredAt < first) first = m.measuredAt;
+      if (m.measuredAt > last) last = m.measuredAt;
+    }
+    const stored = await prisma.measurement.findMany({
+      where: {
+        userId,
+        source: "WITHINGS",
+        sleepStage: null,
+        type: {
+          in: [...new Set(incoming.map((m) => m.type as MeasurementType))],
         },
-        select: { id: true },
+        measuredAt: { gte: first, lte: last },
+      },
+      select: {
+        id: true,
+        type: true,
+        measuredAt: true,
+        value: true,
+        deletedAt: true,
+      },
+    });
+    for (const row of stored) {
+      existingByKey.set(`${row.type}|${row.measuredAt.getTime()}`, row);
+    }
+  }
+
+  const toCreate: typeof incoming = [];
+  for (const m of incoming) {
+    const measType = m.type as MeasurementType;
+    const existing = existingByKey.get(`${m.type}|${m.measuredAt.getTime()}`);
+    if (!existing) {
+      toCreate.push(m);
+      continue;
+    }
+    if (existing.value === m.value && existing.deletedAt === null) {
+      imported++;
+      continue;
+    }
+    try {
+      await prisma.measurement.update({
+        where: { id: existing.id },
+        // `deletedAt: null` is a no-op on a live row, a deliberate
+        // RESURRECTION on a tombstoned one — Withings is the source of
+        // truth for its own rows, so a re-fetched measure brings the row
+        // back (mirrors Google / Fitbit).
+        data: { value: m.value, deletedAt: null },
       });
-      if (existing) {
-        await prisma.measurement.update({
-          where: { id: existing.id },
-          // `deletedAt: null` is a no-op on a live row, a deliberate
-          // RESURRECTION on a tombstoned one — Withings is the source of
-          // truth for its own rows, so a re-fetched measure brings the row
-          // back (mirrors Google / Fitbit).
-          data: { value: m.value, deletedAt: null },
-        });
-      } else {
-        const created = await prisma.measurement.create({
-          data: {
-            userId,
-            type: measType,
-            value: m.value,
-            unit: getUnitForType(m.type),
-            measuredAt: m.measuredAt,
-            source: "WITHINGS",
-          },
-          select: { id: true, type: true, measuredAt: true },
-        });
-        insertedArrivals.push(created);
-      }
       touched.push({ type: measType, measuredAt: m.measuredAt });
       imported++;
     } catch (err) {
-      if (isP2002(err)) {
-        // Benign: the composite unique index serialised a concurrent insert of
-        // this exact reading — the row is present, nothing is lost. Warn and
-        // continue WITHOUT holding the watermark (this is the expected
-        // idempotent-write race the findFirst+create models around).
-        getEvent()?.addWarning(
-          `Withings measure upsert hit an idempotent P2002 (row already present): ${err}`,
-        );
-      } else {
-        // A genuine write failure strands this reading. Flag the cycle so the
-        // watermark is held and the next scheduled tick refetches the overlap
-        // window and retries.
-        anyRowFailed = true;
-        if (firstRowError === null) firstRowError = err;
-        getEvent()?.addWarning(`Failed to upsert measure: ${err}`);
+      // A genuine write failure strands this reading. Flag the cycle so the
+      // watermark is held and the next scheduled tick refetches the overlap
+      // window and retries.
+      anyRowFailed = true;
+      if (firstRowError === null) firstRowError = err;
+      getEvent()?.addWarning(`Failed to update measure: ${err}`);
+    }
+  }
+
+  // New readings in one statement per chunk. `skipDuplicates` absorbs the
+  // idempotent-write race the unique index serialises (a concurrent webhook
+  // run inserting the same reading): that row is present and nothing is lost,
+  // exactly the P2002 the per-row path caught and ignored.
+  for (let i = 0; i < toCreate.length; i += 1_000) {
+    const chunk = toCreate.slice(i, i + 1_000);
+    try {
+      const created = await prisma.measurement.createManyAndReturn({
+        data: chunk.map((m) => ({
+          userId,
+          type: m.type as MeasurementType,
+          value: m.value,
+          unit: getUnitForType(m.type),
+          measuredAt: m.measuredAt,
+          source: "WITHINGS" as const,
+        })),
+        skipDuplicates: true,
+        select: { id: true, type: true, measuredAt: true },
+      });
+      for (const row of created) {
+        insertedArrivals.push(row);
+        touched.push({ type: row.type, measuredAt: row.measuredAt });
       }
+      imported += created.length;
+      if (created.length < chunk.length) {
+        getEvent()?.addWarning(
+          `Withings measure insert skipped ${chunk.length - created.length} rows already present (idempotent race)`,
+        );
+      }
+    } catch (err) {
+      anyRowFailed = true;
+      if (firstRowError === null) firstRowError = err;
+      getEvent()?.addWarning(`Failed to insert measures: ${err}`);
     }
   }
 
@@ -446,9 +495,7 @@ export async function syncUserMeasurements(
   // chart at the 30-day range under-counted recent days.
   try {
     const keys = collapseToTypeDayKeys(touched);
-    for (const k of keys) {
-      await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt);
-    }
+    await recomputeBucketsForKeys(userId, keys);
 
     // v1.8.0 — a Withings sync that lands new weight / BP / pulse /
     // body-comp rows dirties the matching per-metric assessment caches.