fix(fitbit): skip tombstones on sync, batch backfill writes, bound poll concurrency

Three hardening + performance fixes to the Fitbit/Google Health sync.

Stop soft-delete resurrection. The measurement write probed the existing
row by the `(userId, type, source, externalId)` unique key without excluding
soft-deleted rows, so a reading the user deleted came back on the next hourly
sync (the upsert matched the tombstone and took the update branch — silently
undoing the delete). The write now probes only live rows (`deletedAt: null`)
and treats a tombstone as absent, minting a fresh insert instead. The fresh
insert relies on the partial unique index over `deleted_at IS NULL` to keep
live-row uniqueness; the tombstone sits outside it, and `skipDuplicates` guards
a live-row race.

Batch the backfill writes + collapse the rollup recompute. The write loop
was one upsert per reading (N+1) followed by a per-(type,day) DAY-rollup
recompute, so a multi-year backfill paid thousands of serial round-trips on a
concurrency-1 worker. The existence probe is now a single `findMany` over the
batch; fresh rows go through chunked `createMany`; only already-live rows take a
per-row update for their differing values. On a `fullSync` the per-write rollup
hook is deferred and the touched type-days collapse into one
`recomputeUserRollups(from, to)` pass at the end of the cycle. The incremental
path keeps the inline per-day hook. The `stats:` overwrite contract, the
single-watermark `markSynced`, and the all-403 soft-skip guard are unchanged.

Bound the hourly poll concurrency. The poll synced every connection in one
serial loop, so a single slow Google response stalled the whole cohort. The
cohort now fans out through a bounded `p-limit` pool with per-user error
isolation, extracted into `runFitbitPollCohort` so the contract is unit testable
without exporting worker internals. The accumulators are read after the await to
avoid a compound-assignment lost-update across overlapping pool tasks.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
(cherry picked from commit 8369af2471fb8da04a938b5a784433318d81a248)

src/lib/fitbit/__tests__/poll-cohort.test.ts

@@ -0,0 +1,117 @@
+/**
+ * v1.12.1 — `runFitbitPollCohort` bounded-concurrency + per-user isolation.
+ *
+ * The hourly Fitbit cron carries no `userId`, so the worker resolves every
+ * connection and hands the cohort to `runFitbitPollCohort`. These tests pin the
+ * contract that matters for a growing cohort:
+ *   - the pool never runs more than `concurrency` users at once (one slow
+ *     Google response can't stall the whole pass — the others proceed);
+ *   - a single user's failure is isolated (reported, not thrown) and the rest
+ *     of the cohort still syncs;
+ *   - the per-cohort totals add up across the pool.
+ */
+import { describe, expect, it, vi } from "vitest";
+
+// `runFitbitPollCohort` is exercised with an injected `sync` fn so the real
+// per-resource modules + db are never touched.
+vi.mock("@/lib/db", () => ({ prisma: {} }));
+vi.mock("@/lib/crypto", () => ({
+  encrypt: (s: string) => s,
+  decrypt: (s: string) => s,
+}));
+vi.mock("@/lib/logging/context", () => ({
+  getEvent: () => null,
+  annotate: () => {},
+}));
+vi.mock("@/lib/integrations/status", () => ({
+  recordSyncFailure: vi.fn(async () => {}),
+  recordSyncSuccess: vi.fn(async () => {}),
+  isReauthRequired: vi.fn(async () => false),
+}));
+vi.mock("@/lib/rollups/measurement-rollups", () => ({
+  collapseToTypeDayKeys: (r: unknown) => r,
+  recomputeBucketsForMeasurement: vi.fn(async () => {}),
+  recomputeUserRollups: vi.fn(async () => ({ rowsUpserted: 0, durationMs: 0 })),
+}));
+vi.mock("@/lib/insights/comprehensive-generate", () => ({
+  invalidateStatusInsightsForTypes: vi.fn(async () => {}),
+}));
+
+import { FITBIT_POLL_CONCURRENCY, runFitbitPollCohort } from "../sync";
+
+/**
+ * A sync fn that records the peak number of simultaneously-in-flight calls.
+ * Each call yields across several microtask/timer ticks so the pool genuinely
+ * overlaps work — if `runFitbitPollCohort` ran serially, peak would stay 1; if
+ * it ran unbounded, peak would reach the cohort size.
+ */
+function trackingSync(perCall = 1) {
+  let inFlight = 0;
+  const state = { peak: 0 };
+  const sync = vi.fn(async () => {
+    inFlight++;
+    state.peak = Math.max(state.peak, inFlight);
+    // Several real-timer ticks so overlapping calls actually coexist.
+    await new Promise((r) => setTimeout(r, 5));
+    inFlight--;
+    return perCall;
+  });
+  return { sync, state };
+}
+
+describe("runFitbitPollCohort — bounded concurrency", () => {
+  it("overlaps work but never runs more than the concurrency cap at once", async () => {
+    const { sync, state } = trackingSync(1);
+
+    const { usersSynced, measurementsImported } = await runFitbitPollCohort(
+      Array.from({ length: 10 }, (_, i) => String(i)),
+      { concurrency: 3, sync },
+    );
+
+    // Bounded: never more than 3 at once …
+    expect(state.peak).toBeLessThanOrEqual(3);
+    // … yet genuinely parallel (a serial loop would peak at 1).
+    expect(state.peak).toBeGreaterThan(1);
+    expect(usersSynced).toBe(10);
+    expect(measurementsImported).toBe(10);
+    expect(sync).toHaveBeenCalledTimes(10);
+  });
+
+  it("isolates a per-user failure — the rest of the cohort still syncs", async () => {
+    const onUserError = vi.fn();
+    const sync = vi.fn(async (userId: string) => {
+      if (userId === "boom") throw new Error("google 500");
+      return 2;
+    });
+
+    const { usersSynced, measurementsImported } = await runFitbitPollCohort(
+      ["a", "boom", "b"],
+      { concurrency: 2, sync, onUserError },
+    );
+
+    // The failing user is captured, not thrown; the other two complete.
+    expect(usersSynced).toBe(2);
+    expect(measurementsImported).toBe(4);
+    expect(onUserError).toHaveBeenCalledTimes(1);
+    expect(onUserError.mock.calls[0]![0]).toBe("boom");
+  });
+
+  it("defaults the concurrency cap to FITBIT_POLL_CONCURRENCY", async () => {
+    const { sync, state } = trackingSync(0);
+
+    await runFitbitPollCohort(
+      Array.from({ length: 12 }, (_, i) => String(i)),
+      { sync },
+    );
+
+    expect(state.peak).toBeLessThanOrEqual(FITBIT_POLL_CONCURRENCY);
+    expect(sync).toHaveBeenCalledTimes(12);
+  });
+
+  it("returns zero totals for an empty cohort without running anything", async () => {
+    const sync = vi.fn(async () => 1);
+    const res = await runFitbitPollCohort([], { sync });
+    expect(res).toEqual({ usersSynced: 0, measurementsImported: 0 });
+    expect(sync).not.toHaveBeenCalled();
+  });
+});

src/lib/fitbit/__tests__/sync-user-fitbit.test.ts

@@ -21,6 +21,7 @@ const {
   syncUserWorkout,
   findUnique,
   update,
+  recomputeUserRollups,
 } = vi.hoisted(() => ({
   recordSyncSuccess: vi.fn<(...a: unknown[]) => Promise<void>>(async () => {}),
   recordSyncFailure: vi.fn<(...a: unknown[]) => Promise<void>>(async () => {}),
@@ -33,10 +34,22 @@ const {
   syncUserWorkout: vi.fn<(...a: unknown[]) => Promise<number>>(),
   findUnique: vi.fn<(...a: unknown[]) => Promise<unknown>>(),
   update: vi.fn<(...a: unknown[]) => Promise<unknown>>(async () => ({})),
+  recomputeUserRollups: vi.fn<(...a: unknown[]) => Promise<unknown>>(
+    async () => ({ rowsUpserted: 0, durationMs: 0 }),
+  ),
 }));
 
 vi.mock("@/lib/db", () => ({
-  prisma: { fitbitConnection: { findUnique, update } },
+  prisma: {
+    fitbitConnection: { findUnique, update },
+    measurement: {
+      findMany: vi.fn(async () => []),
+      createMany: vi.fn(async (arg: { data: unknown[] }) => ({
+        count: arg.data.length,
+      })),
+      update: vi.fn(async () => ({})),
+    },
+  },
 }));
 vi.mock("@/lib/crypto", () => ({
   encrypt: (s: string) => s,
@@ -52,8 +65,10 @@ vi.mock("@/lib/integrations/status", () => ({
   isReauthRequired: (...a: unknown[]) => isReauthRequired(...a),
 }));
 vi.mock("@/lib/rollups/measurement-rollups", () => ({
-  collapseToTypeDayKeys: () => [],
+  collapseToTypeDayKeys: (rows: Array<{ type: string; measuredAt: Date }>) =>
+    rows,
   recomputeBucketsForMeasurement: vi.fn(async () => {}),
+  recomputeUserRollups: (...a: unknown[]) => recomputeUserRollups(...a),
 }));
 vi.mock("@/lib/insights/comprehensive-generate", () => ({
   invalidateStatusInsightsForTypes: vi.fn(async () => {}),
@@ -72,7 +87,11 @@ vi.mock("../sync-workout", () => ({
   syncUserWorkout: (...a: unknown[]) => syncUserWorkout(...a),
 }));
 
-import { handleCollectionFetchError, syncUserFitbit } from "../sync";
+import {
+  handleCollectionFetchError,
+  syncUserFitbit,
+  upsertFitbitMeasurements,
+} from "../sync";
 import { FitbitApiError } from "../response-classifier";
 
 /** A resource sync that 403s its collection and soft-skips, like the real one. */
@@ -224,7 +243,7 @@ describe("syncUserFitbit — single cycle-wide watermark", () => {
     expect(recordSyncSuccess).not.toHaveBeenCalled();
   });
 
-  it("passes fullSync through with an undefined start on a backfill cycle", async () => {
+  it("passes fullSync + deferRollup through with an undefined start on a backfill cycle", async () => {
     syncUserMetrics.mockResolvedValue(2);
 
     await syncUserFitbit("user1", { fullSync: true });
@@ -232,10 +251,79 @@ describe("syncUserFitbit — single cycle-wide watermark", () => {
     const opts = syncUserMetrics.mock.calls[0]![1] as {
       start?: Date;
       fullSync?: boolean;
+      deferRollup?: boolean;
     };
     expect(opts.fullSync).toBe(true);
     // Full sync walks deep history with no lower bound.
     expect(opts.start).toBeUndefined();
+    // The backfill defers the per-write rollup hook to one end-of-cycle pass.
+    expect(opts.deferRollup).toBe(true);
+  });
+
+  it("does NOT thread deferRollup on an incremental cycle", async () => {
+    syncUserMetrics.mockResolvedValue(1);
+
+    await syncUserFitbit("user1");
+
+    const opts = syncUserMetrics.mock.calls[0]![1] as { deferRollup?: boolean };
+    expect(opts.deferRollup).not.toBe(true);
+  });
+});
+
+describe("syncUserFitbit — backfill collapses rollup recompute to one pass", () => {
+  it("runs ONE recomputeUserRollups spanning the touched days, not a per-day loop", async () => {
+    // A resource that actually writes (via the real deferRollup-aware
+    // upsert) so the orchestrator's defer tracker accumulates touched keys.
+    syncUserActivity.mockImplementation(async (...args: unknown[]) => {
+      const opts = args[1] as { deferRollup?: boolean };
+      const { imported } = await upsertFitbitMeasurements(
+        String(args[0]),
+        [
+          {
+            type: "ACTIVITY_STEPS",
+            value: 1000,
+            unit: "count",
+            measuredAt: new Date("2026-05-01T00:00:00.000Z"),
+            externalId: "stats:steps:2026-05-01",
+          },
+          {
+            type: "ACTIVITY_STEPS",
+            value: 2000,
+            unit: "count",
+            measuredAt: new Date("2026-05-03T00:00:00.000Z"),
+            externalId: "stats:steps:2026-05-03",
+          },
+        ],
+        { deferRollup: opts.deferRollup },
+      );
+      return imported;
+    });
+
+    await syncUserFitbit("user1", { fullSync: true });
+
+    // Exactly one collapsed recompute for the whole backfill cycle.
+    expect(recomputeUserRollups).toHaveBeenCalledTimes(1);
+    const arg = recomputeUserRollups.mock.calls[0]![1] as {
+      types: string[];
+      from: Date;
+      to: Date;
+    };
+    expect(arg.types).toEqual(["ACTIVITY_STEPS"]);
+    // Range spans the first touched day through just past the last.
+    expect(arg.from.getTime()).toBe(
+      new Date("2026-05-01T00:00:00.000Z").getTime(),
+    );
+    expect(arg.to.getTime()).toBe(
+      new Date("2026-05-03T00:00:00.000Z").getTime() + 24 * 60 * 60 * 1000,
+    );
+  });
+
+  it("does NOT run the collapsed recompute on an incremental cycle", async () => {
+    syncUserMetrics.mockResolvedValue(3);
+
+    await syncUserFitbit("user1");
+
+    expect(recomputeUserRollups).not.toHaveBeenCalled();
   });
 
   it("returns 0 without running resources when the connection is gone", async () => {

src/lib/fitbit/__tests__/sync.test.ts

@@ -15,7 +15,16 @@ const { prismaMock, refreshAccessTokenMock, recordSyncFailure } = vi.hoisted(
         update: vi.fn(),
       },
       measurement: {
-        upsert: vi.fn<(arg: Record<string, unknown>) => Promise<unknown>>(
+        findMany:
+          vi.fn<(arg: Record<string, unknown>) => Promise<unknown[]>>(
+            async () => [],
+          ),
+        createMany: vi.fn<
+          (arg: Record<string, unknown>) => Promise<{ count: number }>
+        >(async (arg) => ({
+          count: (arg.data as unknown[]).length,
+        })),
+        update: vi.fn<(arg: Record<string, unknown>) => Promise<unknown>>(
           async () => ({}),
         ),
       },
@@ -28,8 +37,11 @@ const { prismaMock, refreshAccessTokenMock, recordSyncFailure } = vi.hoisted(
 vi.mock("@/lib/db", () => ({ prisma: prismaMock }));
 
 vi.mock("@/lib/rollups/measurement-rollups", () => ({
-  collapseToTypeDayKeys: () => [],
+  collapseToTypeDayKeys: (
+    rows: Array<{ type: string; measuredAt: Date }>,
+  ) => rows,
   recomputeBucketsForMeasurement: vi.fn(async () => {}),
+  recomputeUserRollups: vi.fn(async () => ({ rowsUpserted: 0, durationMs: 0 })),
 }));
 
 vi.mock("@/lib/insights/comprehensive-generate", () => ({
@@ -195,10 +207,12 @@ describe("getValidToken — non-rotating refresh", () => {
   });
 });
 
-describe("upsertFitbitMeasurements — idempotent FITBIT upsert", () => {
-  it("upserts each reading on the (userId, type, source=FITBIT, externalId) key with field-by-field data", async () => {
+describe("upsertFitbitMeasurements — batched, tombstone-safe write", () => {
+  it("inserts a fresh reading via createMany with field-by-field data, source pinned to FITBIT", async () => {
     const measuredAt = new Date("2026-05-10T07:00:00.000Z");
-    const imported = await upsertFitbitMeasurements("user1", [
+    prismaMock.measurement.findMany.mockResolvedValue([]); // nothing live yet
+
+    const { imported } = await upsertFitbitMeasurements("user1", [
       {
         type: "WEIGHT",
         value: 80.5,
@@ -209,35 +223,160 @@ describe("upsertFitbitMeasurements — idempotent FITBIT upsert", () => {
     ]);
 
     expect(imported).toBe(1);
-    const arg = prismaMock.measurement.upsert.mock.calls[0]![0] as {
-      where: { userId_type_source_externalId: Record<string, unknown> };
-      create: Record<string, unknown>;
-      update: { syncVersion: unknown };
+    expect(prismaMock.measurement.update).not.toHaveBeenCalled();
+
+    // The probe filters to LIVE FITBIT rows for the batch's externalIds.
+    const probeArg = prismaMock.measurement.findMany.mock.calls[0]![0] as {
+      where: Record<string, unknown>;
     };
-    // The composite unique key pins source = FITBIT — a re-post overwrites in
-    // place rather than minting a duplicate.
-    expect(arg.where.userId_type_source_externalId).toEqual({
+    expect(probeArg.where).toMatchObject({
       userId: "user1",
-      type: "WEIGHT",
       source: "FITBIT",
-      externalId: "2026-05-10T07:00:00.000Z:weight",
+      deletedAt: null,
+      externalId: { in: ["2026-05-10T07:00:00.000Z:weight"] },
     });
-    // create builds the data field-by-field (no mass-assignment spread).
-    expect(arg.create).toMatchObject({
+
+    const createArg = prismaMock.measurement.createMany.mock.calls[0]![0] as {
+      data: Record<string, unknown>[];
+      skipDuplicates: boolean;
+    };
+    expect(createArg.skipDuplicates).toBe(true);
+    expect(createArg.data[0]).toMatchObject({
       userId: "user1",
       type: "WEIGHT",
       source: "FITBIT",
       value: 80.5,
       unit: "kg",
+      externalId: "2026-05-10T07:00:00.000Z:weight",
     });
-    // update bumps syncVersion for the iOS LWW reconciler.
-    expect(arg.update.syncVersion).toEqual({ increment: 1 });
+  });
+
+  it("overwrites a matched LIVE row in place via update and bumps syncVersion", async () => {
+    const measuredAt = new Date("2026-05-10T07:00:00.000Z");
+    // A re-fetched daily summary: the live row already exists.
+    prismaMock.measurement.findMany.mockResolvedValue([
+      { id: "m-live", type: "WEIGHT", externalId: "stats:weight:2026-05-10" },
+    ]);
+
+    const { imported } = await upsertFitbitMeasurements("user1", [
+      {
+        type: "WEIGHT",
+        value: 79.9,
+        unit: "kg",
+        measuredAt,
+        externalId: "stats:weight:2026-05-10",
+      },
+    ]);
+
+    expect(imported).toBe(1);
+    // No fresh insert — the live row was overwritten in place.
+    expect(prismaMock.measurement.createMany).not.toHaveBeenCalled();
+    const updateArg = prismaMock.measurement.update.mock.calls[0]![0] as {
+      where: { id: string };
+      data: { value: number; syncVersion: unknown };
+    };
+    expect(updateArg.where).toEqual({ id: "m-live" });
+    expect(updateArg.data.value).toBe(79.9);
+    expect(updateArg.data.syncVersion).toEqual({ increment: 1 });
+  });
+
+  it("does NOT resurrect a soft-deleted row — a tombstone is absent from the live probe, so a fresh insert is created", async () => {
+    const measuredAt = new Date("2026-05-10T07:00:00.000Z");
+    // The user deleted this Fitbit reading: the tombstone (deletedAt set) is
+    // excluded from the live probe, so findMany returns NOTHING for the key.
+    prismaMock.measurement.findMany.mockResolvedValue([]);
+
+    const { imported } = await upsertFitbitMeasurements("user1", [
+      {
+        type: "WEIGHT",
+        value: 80.5,
+        unit: "kg",
+        measuredAt,
+        externalId: "stats:weight:2026-05-10",
+      },
+    ]);
+
+    // The deleted row is NOT updated/resurrected — a fresh insert is attempted
+    // (the partial unique index `WHERE deleted_at IS NULL` keeps it from
+    // colliding with the dead row, and skipDuplicates guards a live race).
+    expect(imported).toBe(1);
+    expect(prismaMock.measurement.update).not.toHaveBeenCalled();
+    expect(prismaMock.measurement.createMany).toHaveBeenCalledTimes(1);
+  });
+
+  it("splits a mixed batch: matched-live → update, unmatched → createMany", async () => {
+    const measuredAt = new Date("2026-05-10T07:00:00.000Z");
+    prismaMock.measurement.findMany.mockResolvedValue([
+      { id: "m-live", type: "WEIGHT", externalId: "live-key" },
+    ]);
+
+    const { imported } = await upsertFitbitMeasurements("user1", [
+      { type: "WEIGHT", value: 1, unit: "kg", measuredAt, externalId: "live-key" },
+      { type: "WEIGHT", value: 2, unit: "kg", measuredAt, externalId: "fresh-key" },
+    ]);
+
+    expect(imported).toBe(2);
+    expect(prismaMock.measurement.update).toHaveBeenCalledTimes(1);
+    expect(prismaMock.measurement.update.mock.calls[0]![0]).toMatchObject({
+      where: { id: "m-live" },
+    });
+    const createArg = prismaMock.measurement.createMany.mock.calls[0]![0] as {
+      data: Record<string, unknown>[];
+    };
+    expect(createArg.data).toHaveLength(1);
+    expect(createArg.data[0]).toMatchObject({ externalId: "fresh-key" });
+  });
+
+  it("collapses a duplicate fresh key inside one batch to a single create (last-write-wins)", async () => {
+    const measuredAt = new Date("2026-05-10T07:00:00.000Z");
+    prismaMock.measurement.findMany.mockResolvedValue([]);
+
+    await upsertFitbitMeasurements("user1", [
+      { type: "WEIGHT", value: 1, unit: "kg", measuredAt, externalId: "dup" },
+      { type: "WEIGHT", value: 9, unit: "kg", measuredAt, externalId: "dup" },
+    ]);
+
+    const createArg = prismaMock.measurement.createMany.mock.calls[0]![0] as {
+      data: Record<string, unknown>[];
+    };
+    expect(createArg.data).toHaveLength(1);
+    // Last write wins.
+    expect(createArg.data[0]!.value).toBe(9);
+  });
+
+  it("defers the rollup hook and returns touched keys when deferRollup is set", async () => {
+    const { recomputeBucketsForMeasurement } = await import(
+      "@/lib/rollups/measurement-rollups"
+    );
+    const measuredAt = new Date("2026-05-10T07:00:00.000Z");
+    prismaMock.measurement.findMany.mockResolvedValue([]);
+
+    const { imported, touched } = await upsertFitbitMeasurements(
+      "user1",
+      [
+        {
+          type: "WEIGHT",
+          value: 80.5,
+          unit: "kg",
+          measuredAt,
+          externalId: "k1",
+        },
+      ],
+      { deferRollup: true },
+    );
+
+    expect(imported).toBe(1);
+    expect(touched).toEqual([{ type: "WEIGHT", measuredAt }]);
+    // The inline per-day hook is skipped on the deferred (backfill) path.
+    expect(recomputeBucketsForMeasurement).not.toHaveBeenCalled();
   });
 
   it("returns 0 without touching prisma for an empty batch", async () => {
-    const imported = await upsertFitbitMeasurements("user1", []);
+    const { imported } = await upsertFitbitMeasurements("user1", []);
     expect(imported).toBe(0);
-    expect(prismaMock.measurement.upsert).not.toHaveBeenCalled();
+    expect(prismaMock.measurement.findMany).not.toHaveBeenCalled();
+    expect(prismaMock.measurement.createMany).not.toHaveBeenCalled();
+    expect(prismaMock.measurement.update).not.toHaveBeenCalled();
   });
 });
 

src/lib/fitbit/sync-activity.ts

@@ -113,7 +113,11 @@ export async function syncUserActivity(
         });
       }
     }
-    imported += await upsertFitbitMeasurements(userId, readings);
+    imported += (
+      await upsertFitbitMeasurements(userId, readings, {
+        deferRollup: opts.deferRollup,
+      })
+    ).imported;
   }
 
   // `markSynced` is owned by the orchestrator (`syncUserFitbit`).

src/lib/fitbit/sync-metrics.ts

@@ -135,7 +135,11 @@ export async function syncUserMetrics(
         });
       }
     }
-    imported += await upsertFitbitMeasurements(userId, readings);
+    imported += (
+      await upsertFitbitMeasurements(userId, readings, {
+        deferRollup: opts.deferRollup,
+      })
+    ).imported;
   }
 
   // Height → User.heightCm, only when the user has no height yet. Never mint a

src/lib/fitbit/sync-sleep.ts

@@ -68,7 +68,11 @@ export async function syncUserSleep(
     }
   }
 
-  const imported = await upsertFitbitMeasurements(userId, readings);
+  const imported = (
+    await upsertFitbitMeasurements(userId, readings, {
+      deferRollup: opts.deferRollup,
+    })
+  ).imported;
   // `markSynced` is owned by the orchestrator (`syncUserFitbit`).
   annotate({ action: { name: "fitbit.sleep.sync", details: { imported } } });
   return imported;

src/lib/fitbit/sync.ts

@@ -14,6 +14,7 @@
  * wave and extends this file.
  */
 import { AsyncLocalStorage } from "node:async_hooks";
+import pLimit from "p-limit";
 import { prisma } from "@/lib/db";
 import type { MeasurementType } from "@/generated/prisma/client";
 import { encrypt, decrypt } from "@/lib/crypto";
@@ -27,6 +28,7 @@ import {
 import {
   collapseToTypeDayKeys,
   recomputeBucketsForMeasurement,
+  recomputeUserRollups,
 } from "@/lib/rollups/measurement-rollups";
 import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate";
 import { refreshAccessToken } from "./client";
@@ -162,6 +164,19 @@ interface SoftSkipTracker {
 }
 const softSkipStorage = new AsyncLocalStorage<SoftSkipTracker>();
 
+/**
+ * Per-`syncUserFitbit`-cycle accumulator of the `(type, day)` keys every
+ * resource's writes touched. Only populated on a `fullSync` backfill (when the
+ * inline per-day rollup hook is deferred); the orchestrator drains it into ONE
+ * `recomputeUserRollups(from, to)` pass at the end of the cycle. Scoped by
+ * AsyncLocalStorage so concurrent per-user jobs never cross-pollinate, mirroring
+ * the soft-skip tracker.
+ */
+interface RollupDeferTracker {
+  keys: Array<{ type: MeasurementType; measuredAt: Date }>;
+}
+const rollupDeferStorage = new AsyncLocalStorage<RollupDeferTracker>();
+
 /**
  * Single-source the per-resource collection-fetch error handling. A 403 on one
  * data class soft-skips it (warn + return 0) so sibling resources still sync;
@@ -198,6 +213,17 @@ export interface FitbitResourceSyncOptions {
   fullSync?: boolean;
   /** The incremental lower bound, snapshotted once by the orchestrator. */
   start?: Date;
+  /**
+   * When true, `upsertFitbitMeasurements` writes the rows but SKIPS the inline
+   * per-(type,day) DAY-rollup recompute + status-insight invalidate, returning
+   * the touched type-days to the caller instead. The orchestrator collapses
+   * them into ONE `recomputeUserRollups(from, to)` pass at the end of a
+   * `fullSync` cycle, so a multi-year backfill pays a single range-recompute
+   * rather than thousands of per-day round-trips. The incremental path leaves
+   * this unset and keeps the inline per-day hook (small touched set, warm read
+   * on the next tick).
+   */
+  deferRollup?: boolean;
 }
 
 /**
@@ -236,12 +262,30 @@ export interface FitbitMeasurementUpsert {
   sleepStage?: "IN_BED" | "AWAKE" | "ASLEEP" | "REM" | "CORE" | "DEEP" | null;
 }
 
+/** Chunk size for the batched `createMany` insert of fresh readings. */
+const FITBIT_CREATE_CHUNK = 500;
+
 /**
- * Upsert a batch of mapped Fitbit readings for one user and fold the rollup tier
- * + invalidate status-insight caches once at the end (mirrors the WHOOP / Withings
- * sync tail). Idempotent: the `(userId, type, source, externalId)` unique key
- * makes a re-post (a re-fetched daily summary) overwrite in place rather than
- * minting a duplicate. Returns the count of rows written.
+ * Write a batch of mapped Fitbit readings for one user and (unless deferred)
+ * fold the rollup tier + invalidate status-insight caches once at the end
+ * (mirrors the WHOOP / Withings sync tail). Returns the count of rows written
+ * plus the distinct `(type, day)` keys the write touched.
+ *
+ * TOMBSTONE-SAFE: the live-row probe filters `deletedAt: null`, so a row the
+ * user soft-deleted (iOS LWW reconciler) is NOT matched — a fresh insert is
+ * created instead of resurrecting the tombstone on the next hourly sync. The
+ * partial unique index (`Measurement … WHERE deleted_at IS NULL`) keeps the
+ * fresh insert from colliding with the live set; the tombstone sits outside it.
+ *
+ * OVERWRITE CONTRACT preserved: a re-fetched daily summary (`stats:`-keyed and
+ * every other stable per-point anchor) maps to the SAME live `externalId`, so
+ * the probe finds the existing live row and takes the in-place update branch
+ * (bumping `syncVersion` for the iOS LWW reconciler) — never a duplicate.
+ *
+ * BATCHING: the existence probe is a single `findMany` over the batch's
+ * externalIds; fresh rows go through chunked `createMany` (collapsing the N+1
+ * insert round-trips a deep backfill used to pay), and only the (bounded)
+ * already-live rows take a per-row `update` for their differing values.
  *
  * Best-effort on the rollup fold + insight invalidate — a populator hiccup never
  * fails the user's sync.
@@ -249,35 +293,126 @@ export interface FitbitMeasurementUpsert {
 export async function upsertFitbitMeasurements(
   userId: string,
   readings: FitbitMeasurementUpsert[],
-): Promise<number> {
-  if (readings.length === 0) return 0;
+  opts: { deferRollup?: boolean } = {},
+): Promise<{
+  imported: number;
+  touched: Array<{ type: MeasurementType; measuredAt: Date }>;
+}> {
+  if (readings.length === 0) return { imported: 0, touched: [] };
 
-  let imported = 0;
+  // Probe the LIVE rows (deletedAt: null) for every externalId in the batch in
+  // a single query. A tombstoned row deliberately does NOT appear here, so it is
+  // treated as absent → a fresh insert, not a resurrecting update.
+  const externalIds = readings.map((r) => r.externalId);
+  let liveByKey = new Map<string, { id: string }>();
+  try {
+    const existing = await prisma.measurement.findMany({
+      where: {
+        userId,
+        source: "FITBIT",
+        deletedAt: null,
+        externalId: { in: externalIds },
+      },
+      select: { id: true, type: true, externalId: true },
+    });
+    liveByKey = new Map(
+      existing
+        .filter((e) => e.externalId !== null)
+        .map((e) => [`${e.type}${e.externalId}`, { id: e.id }]),
+    );
+  } catch (err) {
+    // A probe failure must not strand the whole batch; fall back to treating
+    // every row as fresh (the partial unique index still rejects a genuine
+    // live-row collision, which the per-create catch swallows).
+    getEvent()?.addWarning(`Fitbit: live-row probe failed: ${err}`);
+  }
+
+  const toCreate: Array<{
+    type: MeasurementType;
+    value: number;
+    unit: string;
+    measuredAt: Date;
+    externalId: string;
+    sleepStage: FitbitMeasurementUpsert["sleepStage"];
+  }> = [];
+  const toUpdate: Array<{ id: string; r: FitbitMeasurementUpsert }> = [];
   const touched: Array<{ type: MeasurementType; measuredAt: Date }> = [];
 
+  // A batch can carry the same (type, externalId) twice (an overlap re-fetch);
+  // collapse to last-write-wins so we never emit two creates for one live key.
+  const plannedCreateKeys = new Set<string>();
   for (const r of readings) {
     const type = r.type as MeasurementType;
-    try {
-      await prisma.measurement.upsert({
-        where: {
-          userId_type_source_externalId: {
-            userId,
-            type,
-            source: "FITBIT",
-            externalId: r.externalId,
-          },
-        },
-        create: {
-          userId,
+    const key = `${type}${r.externalId}`;
+    const live = liveByKey.get(key);
+    if (live) {
+      toUpdate.push({ id: live.id, r });
+    } else if (!plannedCreateKeys.has(key)) {
+      plannedCreateKeys.add(key);
+      toCreate.push({
+        type,
+        value: r.value,
+        unit: r.unit,
+        measuredAt: r.measuredAt,
+        externalId: r.externalId,
+        sleepStage: r.sleepStage ?? null,
+      });
+    } else {
+      // A duplicate fresh key inside the same batch — overwrite the planned
+      // create's payload so last-write-wins, matching the prior upsert loop.
+      const idx = toCreate.findIndex(
+        (c) => `${c.type}${c.externalId}` === key,
+      );
+      if (idx >= 0) {
+        toCreate[idx] = {
           type,
-          source: "FITBIT",
           value: r.value,
           unit: r.unit,
           measuredAt: r.measuredAt,
           externalId: r.externalId,
           sleepStage: r.sleepStage ?? null,
-        },
-        update: {
+        };
+      }
+    }
+  }
+
+  let imported = 0;
+
+  // Fresh inserts: chunked `createMany` (server-owned rows, field-by-field).
+  // `skipDuplicates` guards the partial-unique index in the rare race where a
+  // concurrent run inserted the same live key between the probe and the write.
+  for (let i = 0; i < toCreate.length; i += FITBIT_CREATE_CHUNK) {
+    const chunk = toCreate.slice(i, i + FITBIT_CREATE_CHUNK);
+    try {
+      const res = await prisma.measurement.createMany({
+        data: chunk.map((c) => ({
+          userId,
+          type: c.type,
+          source: "FITBIT" as const,
+          value: c.value,
+          unit: c.unit,
+          measuredAt: c.measuredAt,
+          externalId: c.externalId,
+          sleepStage: c.sleepStage,
+        })),
+        skipDuplicates: true,
+      });
+      imported += res.count;
+      for (const c of chunk) {
+        touched.push({ type: c.type, measuredAt: c.measuredAt });
+      }
+    } catch (err) {
+      getEvent()?.addWarning(`Fitbit: failed to create measurements: ${err}`);
+    }
+  }
+
+  // Live-row overwrites: per-row update (differing values) on the live id, so
+  // the re-fetched daily summary overwrites in place and bumps `syncVersion`.
+  for (const { id, r } of toUpdate) {
+    try {
+      await prisma.measurement.update({
+        where: { id },
+        data: {
           value: r.value,
           unit: r.unit,
           measuredAt: r.measuredAt,
@@ -286,13 +421,23 @@ export async function upsertFitbitMeasurements(
           syncVersion: { increment: 1 },
         },
       });
-      touched.push({ type, measuredAt: r.measuredAt });
+      touched.push({ type: r.type as MeasurementType, measuredAt: r.measuredAt });
       imported++;
     } catch (err) {
-      getEvent()?.addWarning(`Fitbit: failed to upsert measurement: ${err}`);
+      getEvent()?.addWarning(`Fitbit: failed to update measurement: ${err}`);
     }
   }
 
+  // On a `fullSync` backfill the caller defers the rollup fold: it collapses
+  // every resource's touched type-days into a SINGLE range-recompute at the end
+  // of the cycle (thousands of per-day round-trips → one pass). The incremental
+  // path keeps the inline per-day hook here (small touched set, warm next read).
+  if (opts.deferRollup) {
+    const tracker = rollupDeferStorage.getStore();
+    if (tracker) tracker.keys.push(...touched);
+    return { imported, touched };
+  }
+
   try {
     const keys = collapseToTypeDayKeys(touched);
     for (const k of keys) {
@@ -312,7 +457,7 @@ export async function upsertFitbitMeasurements(
     );
   }
 
-  return imported;
+  return { imported, touched };
 }
 
 /**
@@ -375,6 +520,10 @@ export async function syncUserFitbit(
   const resourceOpts: FitbitResourceSyncOptions = {
     fullSync: opts.fullSync,
     start,
+    // A backfill walks years of daily summaries — defer the per-(type,day)
+    // rollup hook on every write and run ONE range-recompute at the end of the
+    // cycle. The hourly incremental keeps the inline hook (small touched set).
+    deferRollup: opts.fullSync === true,
   };
 
   const [{ syncUserMetrics }, { syncUserActivity }, { syncUserSleep }, { syncUserWorkout }] =
@@ -393,19 +542,53 @@ export async function syncUserFitbit(
   ];
 
   const tracker: SoftSkipTracker = { count: 0 };
+  const deferTracker: RollupDeferTracker = { keys: [] };
   let total = 0;
   let anyFailed = false;
   await softSkipStorage.run(tracker, async () => {
-    for (const fn of resources) {
-      try {
-        total += await fn(userId, resourceOpts);
-      } catch (err) {
-        anyFailed = true;
-        getEvent()?.addWarning(`fitbit ${fn.name} failed for ${userId}: ${err}`);
+    await rollupDeferStorage.run(deferTracker, async () => {
+      for (const fn of resources) {
+        try {
+          total += await fn(userId, resourceOpts);
+        } catch (err) {
+          anyFailed = true;
+          getEvent()?.addWarning(
+            `fitbit ${fn.name} failed for ${userId}: ${err}`,
+          );
+        }
       }
-    }
+    });
   });
 
+  // On a `fullSync` backfill the per-write inline rollup hook was deferred; the
+  // accumulated touched type-days collapse into ONE range-recompute spanning the
+  // touched days. One pass replaces the thousands of per-(type,day) round-trips a
+  // deep backfill would otherwise pay (each = an aggregate SELECT + rollup upsert
+  // + 3 queue sends). Best-effort: a populator hiccup never fails the backfill.
+  if (opts.fullSync && deferTracker.keys.length > 0) {
+    try {
+      const days = collapseToTypeDayKeys(deferTracker.keys);
+      const types = Array.from(new Set(days.map((k) => k.type)));
+      const sorted = days
+        .map((k) => k.measuredAt.getTime())
+        .sort((a, b) => a - b);
+      const from = new Date(sorted[0]!);
+      // The keys are UTC day-starts; extend `to` past the last touched day so
+      // the aggregator's `< to` upper bound covers it.
+      const to = new Date(sorted[sorted.length - 1]! + 24 * 60 * 60 * 1000);
+      await recomputeUserRollups(userId, { types, from, to });
+      invalidateStatusInsightsForTypes(userId, types).catch((err) => {
+        getEvent()?.addWarning(
+          `fitbit: status-insight invalidate failed for ${userId}: ${err}`,
+        );
+      });
+    } catch (err) {
+      getEvent()?.addWarning(
+        `fitbit: backfill rollup recompute failed for ${userId}: ${err}`,
+      );
+    }
+  }
+
   // A genuine grant-revoke 403s EVERY collection: each resource soft-skips
   // (returns 0, records no failure), so `anyFailed` stays false and `total` is
   // 0 — yet the connection is dead until the token-refresh path next catches the
@@ -464,3 +647,54 @@ export function classificationToFailureKind(
       return "transient";
   }
 }
+
+/**
+ * Bounded fan-out width for the hourly Fitbit cohort poll. The cron tick carries
+ * no `userId`, so the worker resolves every connection and hands the cohort here.
+ * A small pool (rather than a strict serial loop) means one slow Google response
+ * can't stall the whole cohort, while the cap keeps the burst of per-user
+ * resource fetches + rollup writes from crowding the worker's DB pool — matching
+ * the `p-limit` ceilings the analytics fan-out + insight warm pass use elsewhere.
+ */
+export const FITBIT_POLL_CONCURRENCY = 4;
+
+/**
+ * Run a Fitbit hourly-poll cohort with bounded concurrency + per-user error
+ * isolation. Each user's `syncUserFitbit` runs inside the pool; a single user's
+ * failure (or a slow Google response) is captured through `onUserError` and never
+ * aborts the rest of the pass. Returns the per-cohort totals for the wide-event
+ * envelope. Extracted from the worker handler so the concurrency contract is unit
+ * testable without exporting worker internals.
+ */
+export async function runFitbitPollCohort(
+  userIds: string[],
+  opts: {
+    concurrency?: number;
+    sync?: (userId: string) => Promise<number>;
+    onUserError?: (userId: string, err: unknown) => void;
+  } = {},
+): Promise<{ usersSynced: number; measurementsImported: number }> {
+  const sync = opts.sync ?? ((userId: string) => syncUserFitbit(userId));
+  const limit = pLimit(opts.concurrency ?? FITBIT_POLL_CONCURRENCY);
+
+  let usersSynced = 0;
+  let measurementsImported = 0;
+  await Promise.all(
+    userIds.map((userId) =>
+      limit(async () => {
+        try {
+          // Read the accumulators AFTER the await resolves. A compound
+          // assignment (`x += await f()`) captures the left value BEFORE the
+          // await and would lose increments under overlapping pool tasks.
+          const n = await sync(userId);
+          measurementsImported = measurementsImported + n;
+          usersSynced = usersSynced + 1;
+        } catch (err) {
+          opts.onUserError?.(userId, err);
+        }
+      }),
+    ),
+  );
+
+  return { usersSynced, measurementsImported };
+}

src/lib/jobs/reminder-worker.ts

@@ -32,7 +32,7 @@ import {
   type WhoopBackfillPayload,
 } from "@/lib/jobs/whoop-backfill";
 import { cleanupExpiredWhoopOAuthStates } from "@/lib/jobs/whoop-oauth-state-cleanup";
-import { syncUserFitbit } from "@/lib/fitbit/sync";
+import { runFitbitPollCohort } from "@/lib/fitbit/sync";
 import {
   FITBIT_BACKFILL_QUEUE,
   FITBIT_BACKFILL_CONCURRENCY,
@@ -1752,16 +1752,16 @@ async function handleFitbitSync(jobs: Job<FitbitSyncPayload>[]) {
       }
       if (targets.length === 0) return;
 
-      let usersSynced = 0;
-      let measurementsImported = 0;
-      for (const { userId } of targets) {
-        try {
-          measurementsImported += await syncUserFitbit(userId);
-          usersSynced++;
-        } catch (err) {
-          evt.addWarning(`job.fitbit_sync failed for user ${userId}: ${err}`);
-        }
-      }
+      // Fan the cohort out with bounded concurrency + per-user error isolation:
+      // one slow Google response can't stall the whole cohort, and a single
+      // user's failure is warned without aborting the pass.
+      const { usersSynced, measurementsImported } = await runFitbitPollCohort(
+        targets.map((t) => t.userId),
+        {
+          onUserError: (userId, err) =>
+            evt.addWarning(`job.fitbit_sync failed for user ${userId}: ${err}`),
+        },
+      );
 
       evt.setBackground({
         task_name: "job.fitbit_sync",