chore(release): v1.28.32 — Fitbit re-keyed re-imports can no longer be silently dropped

Single-fix release mirroring the Google Health natural-key rescue from
v1.28.28 into the Fitbit sync. The measurements natural-key unique index
(user, type, measured_at, source, sleep_stage — NULLS NOT DISTINCT,
migration 0055) covers tombstoned rows, so a re-import carrying NEW
externalIds for previously-removed rows collided on the natural key and
skipDuplicates dropped the insert silently. Every planned create now gets
a second chunked probe by natural key with no deletedAt filter; a match —
live or tombstoned — is updated in place with the fresh value, the new
externalId, and deletedAt: null. Probe failure notes the hard-failure
ledger so the watermark holds and the next tick retries. No other code
path touched.

src/lib/fitbit/__tests__/sync-upsert-resurrect.test.ts

@@ -11,11 +11,14 @@
  */
 import { beforeEach, describe, expect, it, vi } from "vitest";
 
-const { findManyMock, createManyMock, updateMock } = vi.hoisted(() => ({
-  findManyMock: vi.fn(async () => [] as unknown[]),
-  createManyMock: vi.fn(async () => ({ count: 0 })),
-  updateMock: vi.fn(async () => ({})),
-}));
+const { findManyMock, createManyMock, updateMock, addWarningMock } = vi.hoisted(
+  () => ({
+    findManyMock: vi.fn(async () => [] as unknown[]),
+    createManyMock: vi.fn(async () => ({ count: 0 })),
+    updateMock: vi.fn(async () => ({})),
+    addWarningMock: vi.fn(),
+  }),
+);
 
 vi.mock("@/lib/db", () => ({
   prisma: {
@@ -27,6 +30,10 @@ vi.mock("@/lib/db", () => ({
   },
 }));
 vi.mock("@/lib/crypto", () => ({ encrypt: vi.fn(), decrypt: vi.fn() }));
+vi.mock("@/lib/logging/context", () => ({
+  getEvent: () => ({ addWarning: addWarningMock }),
+  annotate: () => {},
+}));
 vi.mock("@/lib/integrations/status", () => ({
   isReauthRequired: vi.fn(async () => false),
   recordSyncFailure: vi.fn(async () => {}),
@@ -62,6 +69,7 @@ beforeEach(() => {
   findManyMock.mockReset().mockResolvedValue([]);
   createManyMock.mockReset().mockResolvedValue({ count: 0 });
   updateMock.mockReset().mockResolvedValue({});
+  addWarningMock.mockReset();
 });
 
 describe("upsertFitbitMeasurements — tombstones resurrect", () => {
@@ -121,3 +129,127 @@ describe("upsertFitbitMeasurements — tombstones resurrect", () => {
     expect(imported).toBe(1);
   });
 });
+
+describe("upsertFitbitMeasurements — natural-key migration rescue", () => {
+  const SLEEP_READING = {
+    type: "SLEEP_DURATION",
+    value: 45,
+    unit: "minutes",
+    measuredAt: new Date("2026-07-08T06:30:00.000Z"),
+    externalId: "log-42:sleep:2026-07-08T05:45:00.000Z",
+    sleepStage: "DEEP" as const,
+  };
+
+  it("re-keys a tombstoned natural-key twin instead of dropping the insert", async () => {
+    // externalId probe: no match (the key FORMAT changed); natural-key probe:
+    // the old-key row (tombstoned by the sweep) occupies the same
+    // (type, measuredAt, sleepStage) slot — without the rescue the insert
+    // would silently die on the 0055 unique via skipDuplicates.
+    findManyMock
+      .mockResolvedValueOnce([]) // externalId probe
+      .mockResolvedValueOnce([
+        {
+          id: "old-row",
+          type: "SLEEP_DURATION",
+          measuredAt: new Date("2026-07-08T06:30:00.000Z"),
+          sleepStage: "DEEP",
+        },
+      ]); // natural-key rescue probe
+
+    const { imported } = await upsertFitbitMeasurements(
+      "user-1",
+      [SLEEP_READING],
+      { deferRollup: true },
+    );
+
+    // The rescue probe must see tombstoned rows too: no deletedAt filter.
+    const probe = (findManyMock.mock.calls[1] as unknown[])[0] as {
+      where: Record<string, unknown>;
+    };
+    expect(probe.where).toEqual({
+      userId: "user-1",
+      source: "FITBIT",
+      type: { in: ["SLEEP_DURATION"] },
+      measuredAt: { in: [new Date("2026-07-08T06:30:00.000Z")] },
+    });
+    expect(probe.where).not.toHaveProperty("deletedAt");
+
+    expect(createManyMock).not.toHaveBeenCalled();
+    expect(updateMock).toHaveBeenCalledTimes(1);
+    const upd = (updateMock.mock.calls[0] as unknown[])[0] as {
+      where: { id: string };
+      data: Record<string, unknown>;
+    };
+    expect(upd.where).toEqual({ id: "old-row" });
+    expect(upd.data.externalId).toBe("log-42:sleep:2026-07-08T05:45:00.000Z");
+    expect(upd.data.deletedAt).toBeNull();
+    expect(upd.data.value).toBe(45);
+    expect(imported).toBe(1);
+  });
+
+  it("re-keys a LIVE natural-key twin in place (update, never a duplicate insert)", async () => {
+    findManyMock
+      .mockResolvedValueOnce([]) // externalId probe
+      .mockResolvedValueOnce([
+        {
+          id: "live-row",
+          type: "ACTIVITY_STEPS",
+          measuredAt: new Date("2026-07-08T00:00:00.000Z"),
+          sleepStage: null,
+        },
+      ]); // natural-key rescue probe
+
+    const reading = { ...STEPS_READING, externalId: "log-99:steps:2026-07-08" };
+    const { imported } = await upsertFitbitMeasurements("user-1", [reading], {
+      deferRollup: true,
+    });
+
+    expect(createManyMock).not.toHaveBeenCalled();
+    expect(updateMock).toHaveBeenCalledTimes(1);
+    const upd = (updateMock.mock.calls[0] as unknown[])[0] as {
+      where: { id: string };
+      data: Record<string, unknown>;
+    };
+    expect(upd.where).toEqual({ id: "live-row" });
+    expect(upd.data.externalId).toBe("log-99:steps:2026-07-08");
+    expect(upd.data.deletedAt).toBeNull();
+    expect(imported).toBe(1);
+  });
+
+  it("creates normally when no natural-key twin exists", async () => {
+    findManyMock.mockResolvedValueOnce([]).mockResolvedValueOnce([]);
+    createManyMock.mockResolvedValue({ count: 1 });
+    const { imported } = await upsertFitbitMeasurements(
+      "user-1",
+      [STEPS_READING],
+      { deferRollup: true },
+    );
+    expect(updateMock).not.toHaveBeenCalled();
+    expect(createManyMock).toHaveBeenCalledTimes(1);
+    expect(imported).toBe(1);
+  });
+
+  it("a rescue-probe failure warns and still attempts the insert (never throws)", async () => {
+    // The ledger note that fails the cycle's verdict is pinned at the
+    // `syncUserFitbit` level (sync-user-fitbit.test.ts) — the ambient
+    // hard-fail scope only exists inside the orchestrator.
+    findManyMock
+      .mockResolvedValueOnce([]) // externalId probe
+      .mockRejectedValueOnce(new Error("db down")); // natural-key rescue probe
+    createManyMock.mockResolvedValue({ count: 1 });
+
+    const { imported } = await upsertFitbitMeasurements(
+      "user-1",
+      [STEPS_READING],
+      { deferRollup: true },
+    );
+
+    expect(addWarningMock).toHaveBeenCalledWith(
+      expect.stringContaining("natural-key rescue probe failed"),
+    );
+    // The planned create still goes to createMany — skipDuplicates absorbs a
+    // twin collision and the held watermark retries the rescue next tick.
+    expect(createManyMock).toHaveBeenCalledTimes(1);
+    expect(imported).toBe(1);
+  });
+});

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

@@ -93,6 +93,7 @@ import {
   upsertFitbitMeasurements,
 } from "../sync";
 import { FitbitApiError } from "../response-classifier";
+import { prisma } from "@/lib/db";
 
 /** A resource sync that 403s its collection and soft-skips, like the real one. */
 async function softSkip403(...args: unknown[]): Promise<number> {
@@ -238,6 +239,42 @@ describe("syncUserFitbit — hard-fail ledger (F-SYNC-4)", () => {
     // The hard failure was recorded on the status ledger.
     expect(recordSyncFailure).toHaveBeenCalled();
   });
+
+  it("a failed natural-key rescue probe notes the ledger: no success stamp, no watermark", async () => {
+    // The rescue probe (second measurement.findMany inside the upsert) dies.
+    // The rows still go to createMany (skipDuplicates absorbs a twin
+    // collision), but a twin that WAS wedged would be dropped silently — so
+    // the ledger entry must hold the watermark for a retry next tick.
+    const measurementFindMany = vi.mocked(prisma.measurement.findMany);
+    measurementFindMany
+      .mockResolvedValueOnce([]) // externalId probe
+      .mockRejectedValueOnce(new Error("db down")); // natural-key rescue probe
+    syncUserMetrics.mockImplementation(async (...args: unknown[]) => {
+      const { imported } = await upsertFitbitMeasurements(
+        String(args[0]),
+        [
+          {
+            type: "ACTIVITY_STEPS",
+            value: 8123,
+            unit: "steps",
+            measuredAt: new Date("2026-07-08T00:00:00.000Z"),
+            externalId: "stats:steps:2026-07-08",
+          },
+        ],
+        { deferRollup: true },
+      );
+      return imported;
+    });
+
+    // Resolves (no rethrow) — the insert was still attempted.
+    const total = await syncUserFitbit("user1");
+    expect(total).toBe(1);
+
+    // The ledger flips anyFailed → success is NOT stamped and the watermark is
+    // not advanced, so the next tick retries the rescue.
+    expect(recordSyncSuccess).not.toHaveBeenCalled();
+    expect(update).not.toHaveBeenCalled();
+  });
 });
 
 describe("syncUserFitbit — single cycle-wide watermark", () => {

src/lib/fitbit/sync.ts

@@ -458,7 +458,12 @@ export async function upsertFitbitMeasurements(
     externalId: string;
     sleepStage: FitbitMeasurementUpsert["sleepStage"];
   }> = [];
-  const toUpdate: Array<{ id: string; r: FitbitMeasurementUpsert }> = [];
+  const toUpdate: Array<{
+    id: string;
+    r: FitbitMeasurementUpsert;
+    /** Set when a key-format migration re-keys the row in place (see below). */
+    reKeyTo?: string;
+  }> = [];
   const touched: Array<{ type: MeasurementType; measuredAt: Date }> = [];
 
   // A batch can carry the same (type, externalId) twice (an overlap re-fetch);
@@ -499,6 +504,92 @@ export async function upsertFitbitMeasurements(
     }
   }
 
+  // ── KEY-FORMAT MIGRATION RESCUE ──────────────────────────────────────────
+  // measurements carries a SECOND full unique index — the natural key
+  // `(user_id, type, measured_at, source, sleep_stage)` NULLS NOT DISTINCT
+  // (migration 0055) — and it covers tombstoned rows too. When an externalId
+  // FORMAT changes (a re-keyed re-import after an anchor-scheme fix), a
+  // re-fetched night's fresh rows carry NEW externalIds but the SAME natural
+  // key as the rows a replace-by-window sweep just tombstoned: the externalId
+  // probe finds nothing, the insert collides with the tombstone on the natural
+  // index, and `skipDuplicates` drops it SILENTLY — old row gone, new row
+  // never written (live incident on the Google transport: a full-history
+  // re-sync erased a user's sleep). So every planned CREATE gets a second
+  // probe by natural key; a match — live or tombstoned — is UPDATED in place
+  // instead: fresh value, the NEW externalId (the migration itself),
+  // `deletedAt: null`.
+  if (toCreate.length > 0) {
+    try {
+      const rescued = new Set<number>();
+      const naturalKeyOf = (
+        type: string,
+        measuredAt: Date,
+        stage: FitbitMeasurementUpsert["sleepStage"],
+      ): string => `${type}|${measuredAt.getTime()}|${stage ?? ""}`;
+      for (let i = 0; i < toCreate.length; i += FITBIT_CREATE_CHUNK) {
+        const chunk = toCreate.slice(i, i + FITBIT_CREATE_CHUNK);
+        const rows = await prisma.measurement.findMany({
+          where: {
+            userId,
+            source: "FITBIT",
+            type: { in: [...new Set(chunk.map((c) => c.type))] },
+            measuredAt: { in: chunk.map((c) => c.measuredAt) },
+          },
+          select: {
+            id: true,
+            type: true,
+            measuredAt: true,
+            sleepStage: true,
+          },
+        });
+        const byNaturalKey = new Map(
+          rows.map((row) => [
+            naturalKeyOf(
+              row.type,
+              row.measuredAt,
+              row.sleepStage as FitbitMeasurementUpsert["sleepStage"],
+            ),
+            row.id,
+          ]),
+        );
+        for (let j = 0; j < chunk.length; j++) {
+          const c = chunk[j];
+          const id = byNaturalKey.get(
+            naturalKeyOf(c.type, c.measuredAt, c.sleepStage),
+          );
+          if (id) {
+            rescued.add(i + j);
+            toUpdate.push({
+              id,
+              r: {
+                type: c.type,
+                value: c.value,
+                unit: c.unit,
+                measuredAt: c.measuredAt,
+                externalId: c.externalId,
+                sleepStage: c.sleepStage,
+              },
+              reKeyTo: c.externalId,
+            });
+          }
+        }
+      }
+      if (rescued.size > 0) {
+        for (let i = toCreate.length - 1; i >= 0; i--) {
+          if (rescued.has(i)) toCreate.splice(i, 1);
+        }
+      }
+    } catch (err) {
+      // Best-effort: an unresolved natural-key twin only means the insert is
+      // dropped by `skipDuplicates` as before — note the failure on the
+      // ambient ledger so the watermark holds and the next tick retries the
+      // rescue rather than losing the rows for good.
+      getEvent()?.addWarning(`Fitbit: natural-key rescue probe failed: ${err}`);
+      const hardTracker = hardFailStorage.getStore();
+      if (hardTracker) hardTracker.failures.push("measurements:rescue");
+    }
+  }
+
   let imported = 0;
 
   // Fresh inserts: chunked `createMany` (server-owned rows, field-by-field).
@@ -534,7 +625,7 @@ export async function upsertFitbitMeasurements(
   // `syncVersion`. `deletedAt: null` rides along unconditionally — a no-op on
   // a live row, a deliberate RESURRECTION on a tombstoned one (Fitbit is the
   // source of truth for its own rows; see TOMBSTONES RESURRECT above).
-  for (const { id, r } of toUpdate) {
+  for (const { id, r, reKeyTo } of toUpdate) {
     try {
       await prisma.measurement.update({
         where: { id },
@@ -544,6 +635,8 @@ export async function upsertFitbitMeasurements(
           measuredAt: r.measuredAt,
           sleepStage: r.sleepStage ?? null,
           deletedAt: null,
+          // Key-format migration: adopt the fresh externalId in place.
+          ...(reKeyTo ? { externalId: reKeyTo } : {}),
           // Surface the server-side mutation to the iOS LWW reconciler.
           syncVersion: { increment: 1 },
         },