Merge integration tz + oauth-refresh fixes into release/v1.20.0

Parse Fitbit's offset-less classic sleep/activity wall-clock strings
against the user's timezone (DST-safe) so a non-UTC user's segments land
on the right wake-day. Add a shared rotated-refresh-token persist that
compare-and-swaps on the stored ciphertext and reuses a peer's rotated
token on a lost race, so concurrent syncs across Fitbit, WHOOP and Oura
no longer park a connection at a spurious reauth.

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

@@ -503,6 +503,62 @@ describe("sleep mapping (1.2 levels.data)", () => {
     expect(mapSleepSession({ levels: { data: [] } })).toHaveLength(0);
     expect(mapSleepSession({})).toHaveLength(0);
   });
+
+  it("anchors a near-midnight segment END to the user's timezone, not the process zone", () => {
+    // A Berlin user (UTC+2 in May / CEST) whose last segment ENDS at 00:30
+    // local on 2026-05-11. The offset-less wall clock `2026-05-11T00:30:00`
+    // must resolve to 22:30 UTC on 2026-05-10 — NOT 00:30 UTC (which a bare
+    // `new Date(iso)` in a UTC process would produce, flipping the wake-day).
+    const sessions = readSleepSessions({
+      sleep: [
+        {
+          logId: 4242,
+          startTime: "2026-05-10T23:00:00.000",
+          endTime: "2026-05-11T00:30:00.000",
+          levels: {
+            data: [
+              {
+                dateTime: "2026-05-11T00:00:00.000",
+                level: "deep",
+                seconds: 1800, // ends 00:30 Berlin local
+              },
+            ],
+          },
+        },
+      ],
+    });
+
+    const out = mapSleepSession(sessions[0]!, "Europe/Berlin");
+    expect(out).toHaveLength(1);
+    // 00:30 CEST (UTC+2) → 22:30 UTC on the PRIOR civil day.
+    expect(out[0]!.measuredAt.toISOString()).toBe("2026-05-10T22:30:00.000Z");
+  });
+
+  it("anchors a logId-less session anchor against the user's timezone", () => {
+    const sessions = readSleepSessions({
+      sleep: [
+        {
+          // No logId → the anchor falls back to the END instant, which must
+          // also be tz-resolved so the externalId is stable across syncs.
+          endTime: "2026-05-11T00:30:00.000",
+          levels: {
+            data: [
+              {
+                dateTime: "2026-05-11T00:00:00.000",
+                level: "rem",
+                seconds: 1800,
+              },
+            ],
+          },
+        },
+      ],
+    });
+
+    const out = mapSleepSession(sessions[0]!, "Europe/Berlin");
+    expect(out).toHaveLength(1);
+    // anchor = end instant ISO (UTC) → fieldTag starts with that instant.
+    expect(out[0]!.fieldTag).toBe("2026-05-10T22:30:00.000Z:sleep_rem:0");
+  });
 });
 
 describe("workout mapping (activities list)", () => {

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

@@ -14,6 +14,7 @@ const { prismaMock, refreshAccessTokenMock, recordSyncFailure } = vi.hoisted(
       fitbitConnection: {
         findUnique: vi.fn(),
         update: vi.fn(),
+        updateMany: vi.fn(),
       },
       measurement: {
         findMany: vi.fn<(arg: Record<string, unknown>) => Promise<unknown[]>>(
@@ -103,7 +104,7 @@ describe("getValidToken — rotating refresh", () => {
       access_token: "new-access",
       expires_in: 3600,
     });
-    prismaMock.fitbitConnection.update.mockResolvedValue({});
+    prismaMock.fitbitConnection.updateMany.mockResolvedValue({ count: 1 });
 
     const result = await getValidToken("user1");
 
@@ -112,7 +113,7 @@ describe("getValidToken — rotating refresh", () => {
       clientId: "cid",
       clientSecret: "csecret",
     });
-    const updateArg = prismaMock.fitbitConnection.update.mock.calls[0]![0];
+    const updateArg = prismaMock.fitbitConnection.updateMany.mock.calls[0]![0];
     expect(updateArg.data.accessToken).toBe("enc(new-access)");
     // The stored refresh token must NOT be wiped when the response omits one.
     expect(updateArg.data).not.toHaveProperty("refreshToken");
@@ -131,15 +132,42 @@ describe("getValidToken — rotating refresh", () => {
       refresh_token: "rotated-refresh",
       expires_in: 3600,
     });
-    prismaMock.fitbitConnection.update.mockResolvedValue({});
+    prismaMock.fitbitConnection.updateMany.mockResolvedValue({ count: 1 });
 
     await getValidToken("user1");
 
-    const updateArg = prismaMock.fitbitConnection.update.mock.calls[0]![0];
+    const updateArg = prismaMock.fitbitConnection.updateMany.mock.calls[0]![0];
+    // CAS guard scopes the write to the connection AND the spent ciphertext.
+    expect(updateArg.where.id).toBe("conn1");
+    expect(updateArg.where.refreshToken).toBe("enc(old-refresh)");
     expect(updateArg.data.accessToken).toBe("enc(new-access)");
     expect(updateArg.data.refreshToken).toBe("enc(rotated-refresh)");
   });
 
+  it("reuses the peer's rotated token on a lost CAS race (no spurious reauth)", async () => {
+    prismaMock.fitbitConnection.findUnique
+      .mockResolvedValueOnce({
+        id: "conn1",
+        fitbitUserId: "abc123",
+        accessToken: "enc(old-access)",
+        refreshToken: "enc(old-refresh)",
+        tokenExpiresAt: new Date(Date.now() - 1000),
+      })
+      .mockResolvedValueOnce({ accessToken: "enc(peer-access)" });
+    refreshAccessTokenMock.mockResolvedValue({
+      access_token: "new-access",
+      refresh_token: "rotated-refresh",
+      expires_in: 3600,
+    });
+    // A concurrent sync rotated first → zero rows match the CAS guard.
+    prismaMock.fitbitConnection.updateMany.mockResolvedValue({ count: 0 });
+
+    const result = await getValidToken("user1");
+
+    expect(result?.accessToken).toBe("peer-access");
+    expect(recordSyncFailure).not.toHaveBeenCalled();
+  });
+
   it("returns the stored token without refresh when not near expiry", async () => {
     prismaMock.fitbitConnection.findUnique.mockResolvedValue({
       id: "conn1",

src/lib/fitbit/client.ts

@@ -39,6 +39,7 @@
 import { createHash, randomBytes } from "node:crypto";
 import { getEvent } from "@/lib/logging/context";
 import { safeFetch } from "@/lib/safe-fetch";
+import { zonedWallClockToUtc } from "@/lib/tz/wall-clock";
 import { FitbitApiError, classifyFitbitResponse } from "./response-classifier";
 
 /** Classic Fitbit Web API base. */
@@ -929,13 +930,13 @@ interface FitbitSleepSession {
  * The stable session anchor for a sleep externalId — the logId, else the end (or
  * start) instant. A re-scored night reuses the same logId → overwrites in place.
  */
-function sleepSessionAnchor(s: FitbitSleepSession): string {
+function sleepSessionAnchor(s: FitbitSleepSession, tz?: string): string {
   if (typeof s.logId === "number" && Number.isFinite(s.logId)) {
     return String(s.logId);
   }
   for (const t of [s.endTime, s.startTime]) {
     if (typeof t === "string") {
-      const d = parseLocalInstant(t);
+      const d = parseLocalInstant(t, tz);
       if (d) return d.toISOString();
     }
   }
@@ -943,12 +944,40 @@ function sleepSessionAnchor(s: FitbitSleepSession): string {
 }
 
 /**
- * Parse a Fitbit sleep timestamp. The 1.2 sleep log emits LOCAL wall-clock ISO
- * strings WITHOUT an offset (e.g. `2020-01-26T03:02:30.000`); `new Date(...)`
- * treats an offset-less ISO string as local time, which is the user's intent
- * here (the night belongs to the user's local clock). Returns null on a miss.
+ * Matches an offset-less local ISO wall-clock string Fitbit emits on the classic
+ * sleep + activity logs (`2020-01-26T03:02:30` / `...T03:02:30.000`). No trailing
+ * `Z`, no `±hh:mm` — those denote an absolute instant and are honoured verbatim.
  */
-function parseLocalInstant(iso: string): Date | null {
+const OFFSET_LESS_LOCAL_ISO =
+  /^(\d{4})-(\d{2})-(\d{2})[T ](\d{2}):(\d{2})(?::(\d{2}))?(?:\.\d+)?$/;
+
+/**
+ * Parse a Fitbit timestamp into a UTC instant. The classic 1.2 sleep log and the
+ * activities list emit LOCAL wall-clock strings WITHOUT an offset (e.g.
+ * `2020-01-26T03:02:30.000`) — the night/activity belongs to the user's local
+ * clock, so the wall clock must be resolved against the USER'S timezone, not the
+ * process zone. A bare `new Date(iso)` parses an offset-less string in the host
+ * zone (UTC in production), which shifts a non-UTC user's segment by their UTC
+ * offset and can flip the wake-day in the night reconstruction. When `tz` is
+ * omitted (no user zone on the path) the host-local fallback preserves the prior
+ * behaviour. Strings that DO carry an offset/`Z` are absolute and parsed as-is.
+ * Returns null on a miss.
+ */
+function parseLocalInstant(iso: string, tz?: string): Date | null {
+  const m = OFFSET_LESS_LOCAL_ISO.exec(iso.trim());
+  if (m) {
+    return zonedWallClockToUtc(
+      {
+        year: Number(m[1]),
+        month: Number(m[2]),
+        day: Number(m[3]),
+        hour: Number(m[4]),
+        minute: Number(m[5]),
+        second: m[6] ? Number(m[6]) : 0,
+      },
+      tz,
+    );
+  }
   const d = new Date(iso);
   return Number.isNaN(d.getTime()) ? null : d;
 }
@@ -959,14 +988,20 @@ function parseLocalInstant(iso: string): Date | null {
  * Each segment carries a stage-scoped, INDEXED fieldTag so the several segments
  * of one stage stay distinct under the `(userId, type, source, externalId)`
  * dedup key. Unknown stage labels are skipped.
+ *
+ * The 1.2 sleep log emits OFFSET-LESS local wall-clock timestamps, so `tz` (the
+ * user's stored zone) anchors them to the correct UTC instant — without it a
+ * non-UTC user's near-midnight segment END would shift by their UTC offset and
+ * could land on the wrong wake-day in the night reconstruction.
  */
 export function mapSleepSession(
   session: FitbitSleepSession,
+  tz?: string,
 ): FitbitMappedMeasurement[] {
   const data = session.levels?.data;
   if (!Array.isArray(data) || data.length === 0) return [];
 
-  const anchor = sleepSessionAnchor(session);
+  const anchor = sleepSessionAnchor(session, tz);
   const out: FitbitMappedMeasurement[] = [];
   let segIndex = 0;
   for (const seg of data) {
@@ -975,7 +1010,7 @@ export function mapSleepSession(
     if (!positive(seg.seconds)) continue;
     const startStr = typeof seg.dateTime === "string" ? seg.dateTime : null;
     if (!startStr) continue;
-    const start = parseLocalInstant(startStr);
+    const start = parseLocalInstant(startStr, tz);
     if (!start) continue;
     const minutes = seg.seconds / 60;
     const end = new Date(start.getTime() + seg.seconds * 1000);
@@ -1083,13 +1118,18 @@ export interface FitbitMappedWorkout {
  * there is no usable start + positive duration. The externalId anchors on the
  * `logId` (stable) so a re-fetch overwrites the same row. The classic
  * activities-list endpoint does not surface min/max HR, so those stay null.
+ *
+ * `startTime` is an offset-less local wall-clock string, so `tz` (the user's
+ * stored zone) anchors it to the correct UTC instant rather than the process
+ * zone.
  */
 export function mapWorkout(
   entry: FitbitActivityLogEntry,
+  tz?: string,
 ): FitbitMappedWorkout | null {
   const start =
     typeof entry.startTime === "string"
-      ? parseLocalInstant(entry.startTime)
+      ? parseLocalInstant(entry.startTime, tz)
       : null;
   if (!start) return null;
   if (!positive(entry.duration)) return null;

src/lib/fitbit/sync-sleep.ts

@@ -32,6 +32,7 @@ import {
   type FitbitResourceSyncOptions,
 } from "./sync";
 import { annotate } from "@/lib/logging/context";
+import { resolveUserTimezone } from "@/lib/tz/resolver";
 
 export async function syncUserSleep(
   userId: string,
@@ -40,6 +41,11 @@ export async function syncUserSleep(
   const tokenInfo = await getValidToken(userId);
   if (!tokenInfo) return 0;
 
+  // The classic 1.2 sleep log emits offset-less local wall-clock timestamps;
+  // anchor them against the user's stored zone so a near-midnight segment END
+  // lands on the correct wake-day rather than being shifted by the process zone.
+  const tz = await resolveUserTimezone(userId);
+
   const start = opts.start ?? new Date(Date.now() - 30 * 24 * 60 * 60 * 1000);
   const end = opts.end ?? new Date();
   const windows = chunkDateRanges(start, end, FITBIT_SLEEP_RANGE_DAYS);
@@ -53,7 +59,7 @@ export async function syncUserSleep(
       return handleCollectionFetchError("fetchSleep", userId, err);
     }
     for (const session of readSleepSessions(body)) {
-      for (const m of mapSleepSession(session)) {
+      for (const m of mapSleepSession(session, tz)) {
         readings.push({
           type: m.type,
           value: m.value,

src/lib/fitbit/sync-workout.ts

@@ -29,6 +29,7 @@ import {
 } from "./sync";
 import { prisma } from "@/lib/db";
 import { annotate, getEvent } from "@/lib/logging/context";
+import { resolveUserTimezone } from "@/lib/tz/resolver";
 
 /** Activities per page on the `activities/list` walk. */
 const WORKOUT_PAGE_SIZE = 100;
@@ -42,6 +43,10 @@ export async function syncUserWorkout(
   const tokenInfo = await getValidToken(userId);
   if (!tokenInfo) return 0;
 
+  // `startTime` on the activities list is offset-less local wall-clock; anchor
+  // it to the user's stored zone rather than the process zone.
+  const tz = await resolveUserTimezone(userId);
+
   const afterDate =
     opts.start ?? new Date(Date.now() - 30 * 24 * 60 * 60 * 1000);
 
@@ -67,7 +72,7 @@ export async function syncUserWorkout(
 
   let imported = 0;
   for (const entry of entries) {
-    const w: FitbitMappedWorkout | null = mapWorkout(entry);
+    const w: FitbitMappedWorkout | null = mapWorkout(entry, tz);
     if (!w) continue; // no usable time span — not a workout
 
     try {

src/lib/fitbit/sync.ts

@@ -5,7 +5,9 @@
  *     `tokenExpiresAt - 5 min`, and persists the new access token + expiry AND
  *     the rotated refresh token. Classic Fitbit ROTATES refresh tokens
  *     (one-time use), so the new `refresh_token` MUST replace the stored one —
- *     the inverse of the prior Google transport, which did not rotate.
+ *     the inverse of the prior Google transport, which did not rotate. The
+ *     persist is a compare-and-swap on the stored ciphertext so two overlapping
+ *     syncs cannot park the connection at reauth (see `persistRotatedToken`).
  *   - `recordFitbitSyncFailure` / `classificationToFailureKind` map a
  *     classified API error onto the shared integration-status ledger.
  *   - `upsertFitbitMeasurements` + `syncUserFitbit` own the write tail (dedup,
@@ -33,6 +35,7 @@ import {
   recomputeUserRollups,
 } from "@/lib/rollups/measurement-rollups";
 import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate";
+import { persistRotatedToken } from "@/lib/integrations/oauth-refresh";
 import { refreshAccessToken } from "./client";
 import { getUserFitbitCredentials } from "./credentials";
 import {
@@ -115,22 +118,47 @@ export async function getValidToken(
 
       // Classic Fitbit ROTATES refresh tokens: the refresh response carries a
       // fresh `refresh_token` that replaces the (now-invalid) stored one. Persist
-      // it unconditionally. Guard the malformed-reply case: if the response omits
-      // the new refresh token, keep the existing one rather than writing a blank,
-      // which would brick every future refresh.
-      await prisma.fitbitConnection.update({
-        where: { id: connection.id },
-        data: {
-          accessToken: encrypt(newTokens.access_token),
-          tokenExpiresAt: expiresAt,
-          ...(newTokens.refresh_token
-            ? { refreshToken: encrypt(newTokens.refresh_token) }
-            : {}),
+      // with a compare-and-swap on the stored ciphertext so a concurrent sync
+      // that already rotated the token does not get its work clobbered — and so
+      // THIS caller, if it lost the race, reuses the peer's rotated access token
+      // instead of parking the connection at reauth. Guard the malformed-reply
+      // case: if the response omits the new refresh token, keep the existing one
+      // rather than writing a blank, which would brick every future refresh.
+      const persistedAccessToken = await persistRotatedToken(
+        newTokens.access_token,
+        {
+          conditionalUpdate: async () => {
+            const { count } = await prisma.fitbitConnection.updateMany({
+              // `connection.refreshToken` is the exact ciphertext we read +
+              // spent; matching it is the CAS guard (a re-encrypt would differ).
+              where: {
+                id: connection.id,
+                refreshToken: connection.refreshToken,
+              },
+              data: {
+                accessToken: encrypt(newTokens.access_token),
+                tokenExpiresAt: expiresAt,
+                ...(newTokens.refresh_token
+                  ? { refreshToken: encrypt(newTokens.refresh_token) }
+                  : {}),
+              },
+            });
+            return count;
+          },
+          readPeerAccessToken: async () => {
+            const fresh = await prisma.fitbitConnection.findUnique({
+              where: { id: connection.id },
+              select: { accessToken: true },
+            });
+            return fresh ? decrypt(fresh.accessToken) : null;
+          },
         },
-      });
+      );
+
+      if (!persistedAccessToken) return null;
 
       return {
-        accessToken: newTokens.access_token,
+        accessToken: persistedAccessToken,
         connection: {
           id: connection.id,
           fitbitUserId: connection.fitbitUserId,

src/lib/integrations/__tests__/oauth-refresh.test.ts

@@ -0,0 +1,41 @@
+import { describe, expect, it, vi } from "vitest";
+import { persistRotatedToken } from "../oauth-refresh";
+
+describe("persistRotatedToken", () => {
+  it("returns the freshly minted access token when the CAS write wins (1 row)", async () => {
+    const conditionalUpdate = vi.fn(async () => 1);
+    const readPeerAccessToken = vi.fn(async () => "peer");
+
+    const token = await persistRotatedToken("mine", {
+      conditionalUpdate,
+      readPeerAccessToken,
+    });
+
+    expect(token).toBe("mine");
+    expect(conditionalUpdate).toHaveBeenCalledTimes(1);
+    // No re-read on a win.
+    expect(readPeerAccessToken).not.toHaveBeenCalled();
+  });
+
+  it("reuses the peer's rotated token on a lost race (0 rows)", async () => {
+    const conditionalUpdate = vi.fn(async () => 0);
+    const readPeerAccessToken = vi.fn(async () => "peer");
+
+    const token = await persistRotatedToken("mine", {
+      conditionalUpdate,
+      readPeerAccessToken,
+    });
+
+    expect(token).toBe("peer");
+    expect(readPeerAccessToken).toHaveBeenCalledTimes(1);
+  });
+
+  it("returns null when the row vanished after a lost race", async () => {
+    const token = await persistRotatedToken("mine", {
+      conditionalUpdate: async () => 0,
+      readPeerAccessToken: async () => null,
+    });
+
+    expect(token).toBeNull();
+  });
+});

src/lib/integrations/oauth-refresh.ts

@@ -0,0 +1,69 @@
+/**
+ * Shared OAuth rotating-refresh-token persistence helper.
+ *
+ * Fitbit, WHOOP, and Oura all issue ONE-TIME-USE refresh tokens: every refresh
+ * response carries a fresh `refresh_token` that invalidates the one just spent.
+ * The naive read → refresh → persist sequence is racy. When two syncs for the
+ * same user overlap (a scheduled tick collides with a manual sync, or two
+ * resource syncs each call `getValidToken`), both read the SAME stored refresh
+ * token, both POST it, and the provider honours only the first. The loser's POST
+ * comes back 401, the caller classifies it as `reauth_required`, and the
+ * connection is parked at `error_reauth` — spuriously, because the peer already
+ * rotated the token successfully and the connection is perfectly healthy.
+ *
+ * The fix is a compare-and-swap on persistence: write the rotated tokens with a
+ * `WHERE refreshToken = <the exact ciphertext we read>` guard. The tokens are
+ * AES-256-GCM encrypted with a random IV, so the guard matches the STORED
+ * ciphertext (passed in verbatim), never a re-encryption of the plaintext —
+ * re-encrypting the same plaintext yields different bytes. If the conditional
+ * update touches zero rows a peer refresh already won; re-read the row and reuse
+ * the peer's freshly-rotated access token instead of erroring.
+ *
+ * Both racing callers thus end up with a usable access token and neither parks
+ * the connection. The window is small but real on the shared scheduler, and a
+ * single spurious reauth forces the user back through the consent flow.
+ */
+
+/** Tokens returned by a provider's refresh endpoint (the fields we persist). */
+export interface RotatedTokens {
+  accessToken: string;
+  /** The new refresh token. Omitted only on a malformed reply (kept as-is). */
+  refreshToken?: string | null;
+  /** Absolute expiry of the new access token, if the provider supplies one. */
+  expiresAt?: Date;
+}
+
+export interface PersistRotatedTokenOptions {
+  /**
+   * Conditional persist: update the connection row ONLY when its stored refresh
+   * token still equals the ciphertext this caller read (compare-and-swap on the
+   * stored bytes). Returns the number of rows written — 0 means a peer rotated
+   * first. The closure owns the encryption + the `WHERE` clause so the helper
+   * stays storage-agnostic across the three provider tables.
+   */
+  conditionalUpdate: () => Promise<number>;
+  /**
+   * Re-read the connection AFTER a lost race and return the peer's freshly
+   * rotated, DECRYPTED access token. Returns null only if the row vanished
+   * (connection deleted mid-flight) — the caller then treats it as no token.
+   */
+  readPeerAccessToken: () => Promise<string | null>;
+}
+
+/**
+ * Persist a rotated token pair with compare-and-swap semantics, falling back to
+ * the peer's token on a lost race. Returns the access token the caller should
+ * use for the current sync — the one it just minted on a win, or the peer's on a
+ * loss — or null when the connection no longer exists.
+ */
+export async function persistRotatedToken(
+  freshAccessToken: string,
+  opts: PersistRotatedTokenOptions,
+): Promise<string | null> {
+  const rows = await opts.conditionalUpdate();
+  if (rows > 0) return freshAccessToken;
+  // Lost the race: a concurrent refresh already rotated the stored token, which
+  // invalidated the one we spent. Reuse the peer's freshly rotated access token
+  // rather than parking the connection at reauth.
+  return opts.readPeerAccessToken();
+}

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

@@ -83,11 +83,17 @@ vi.mock("../client", async (importOriginal) => {
 import { syncUserOura } from "../sync";
 import { OuraApiError } from "../response-classifier";
 
-const CONN = { accessToken: "acc", refreshToken: "ref" };
+const CONN = {
+  accessToken: "acc",
+  refreshToken: "ref",
+  refreshTokenCiphertext: "enc(ref)",
+};
 
 beforeEach(() => {
   getConnMock.mockReset();
-  storeTokensMock.mockReset().mockResolvedValue(undefined);
+  // The compare-and-swap persist returns the access token the caller should use
+  // (its own on a CAS win) — default to the freshly minted one.
+  storeTokensMock.mockReset().mockResolvedValue("newAcc");
   getCredsMock
     .mockReset()
     .mockResolvedValue({ clientId: "c", clientSecret: "s" });
@@ -155,11 +161,40 @@ describe("syncUserOura", () => {
     });
     const imported = await syncUserOura("u1");
     expect(refreshMock).toHaveBeenCalledWith("ref", expect.anything());
-    expect(storeTokensMock).toHaveBeenCalledWith("u1", "newAcc", "newRef");
+    expect(storeTokensMock).toHaveBeenCalledWith(
+      "u1",
+      "newAcc",
+      "newRef",
+      "enc(ref)",
+    );
     expect(imported).toBe(1);
     expect(recordSuccessMock).toHaveBeenCalledWith("u1", "oura");
   });
 
+  it("reuses the peer's rotated token on a lost CAS race and still completes (no reauth)", async () => {
+    getConnMock.mockResolvedValue(CONN);
+    fetchReadinessMock
+      .mockRejectedValueOnce(err401())
+      .mockResolvedValueOnce([{ id: "x", day: "2026-06-10", score: 70 }]);
+    refreshMock.mockResolvedValue({
+      access_token: "newAcc",
+      refresh_token: "newRef",
+      expires_in: 86400,
+    });
+    // A concurrent sync rotated first → the persist returns the PEER's token.
+    storeTokensMock.mockResolvedValue("peerAcc");
+
+    const imported = await syncUserOura("u1");
+
+    // The retry runs with the peer's token rather than the invalidated one.
+    expect(fetchReadinessMock).toHaveBeenLastCalledWith(
+      "peerAcc",
+      expect.anything(),
+    );
+    expect(imported).toBe(1);
+    expect(recordFailureMock).not.toHaveBeenCalled();
+  });
+
   it("records reauth_required when the refresh itself fails", async () => {
     getConnMock.mockResolvedValue(CONN);
     fetchReadinessMock.mockRejectedValue(err401());

src/lib/oura/credentials.ts

@@ -12,6 +12,7 @@
  */
 import { prisma } from "@/lib/db";
 import { decrypt, encrypt } from "@/lib/crypto";
+import { persistRotatedToken } from "@/lib/integrations/oauth-refresh";
 import { getOuraCredentials, type OuraCredentials } from "./client";
 
 /**
@@ -77,6 +78,12 @@ export async function clearOuraClientCredentials(
 export interface OuraConnection {
   accessToken: string;
   refreshToken: string;
+  /**
+   * The exact stored refresh-token ciphertext (NOT the decrypted value). The
+   * rotating-refresh compare-and-swap in `storeOuraTokens` matches on this so a
+   * concurrent sync that already rotated the token is not clobbered.
+   */
+  refreshTokenCiphertext: string;
 }
 
 export async function getOuraConnection(
@@ -95,20 +102,49 @@ export async function getOuraConnection(
   return {
     accessToken: decrypt(user.ouraAccessTokenEncrypted),
     refreshToken: decrypt(user.ouraRefreshTokenEncrypted),
+    refreshTokenCiphertext: user.ouraRefreshTokenEncrypted,
   };
 }
 
-/** Persist a rotated token pair after a successful refresh. */
+/**
+ * Persist a rotated Oura token pair with compare-and-swap on the stored refresh
+ * ciphertext (the inverse of Fitbit / WHOOP, which key by connection id). Oura's
+ * tokens live on `User` and the refresh is REACTIVE on a 401, so two overlapping
+ * syncs can race the same one-time-use refresh token. The CAS guard writes only
+ * when the stored ciphertext still equals the one this caller spent; on a lost
+ * race (0 rows) it re-reads and returns the peer's freshly rotated access token.
+ *
+ * Returns the access token the caller should use for the current sync (its own
+ * on a win, the peer's on a loss), or null if the connection vanished.
+ */
 export async function storeOuraTokens(
   userId: string,
   accessToken: string,
   refreshToken: string,
-): Promise<void> {
-  await prisma.user.update({
-    where: { id: userId },
-    data: {
-      ouraAccessTokenEncrypted: encrypt(accessToken),
-      ouraRefreshTokenEncrypted: encrypt(refreshToken),
+  expectedRefreshCiphertext: string,
+): Promise<string | null> {
+  return persistRotatedToken(accessToken, {
+    conditionalUpdate: async () => {
+      const { count } = await prisma.user.updateMany({
+        where: {
+          id: userId,
+          ouraRefreshTokenEncrypted: expectedRefreshCiphertext,
+        },
+        data: {
+          ouraAccessTokenEncrypted: encrypt(accessToken),
+          ouraRefreshTokenEncrypted: encrypt(refreshToken),
+        },
+      });
+      return count;
+    },
+    readPeerAccessToken: async () => {
+      const fresh = await prisma.user.findUnique({
+        where: { id: userId },
+        select: { ouraAccessTokenEncrypted: true },
+      });
+      return fresh?.ouraAccessTokenEncrypted
+        ? decrypt(fresh.ouraAccessTokenEncrypted)
+        : null;
     },
   });
 }

src/lib/oura/sync.ts

@@ -124,6 +124,7 @@ async function fetchAll(
   userId: string,
   accessToken: string,
   refreshToken: string,
+  refreshTokenCiphertext: string,
   lookbackDays: number,
 ): Promise<OuraMeasurementUpsert[]> {
   const now = new Date();
@@ -179,8 +180,17 @@ async function fetchAll(
     if (!creds) throw err;
 
     const rotated = await refreshAccessToken(refreshToken, creds);
-    await storeOuraTokens(userId, rotated.access_token, rotated.refresh_token);
-    return run(rotated.access_token);
+    // Compare-and-swap persist: on a lost race against a concurrent sync this
+    // returns the peer's freshly rotated access token rather than the (now
+    // invalidated) one we just minted, so neither sync parks the connection.
+    const usableToken = await storeOuraTokens(
+      userId,
+      rotated.access_token,
+      rotated.refresh_token,
+      refreshTokenCiphertext,
+    );
+    if (!usableToken) throw err;
+    return run(usableToken);
   }
 }
 
@@ -202,6 +212,7 @@ export async function syncUserOura(
       userId,
       conn.accessToken,
       conn.refreshToken,
+      conn.refreshTokenCiphertext,
       opts.lookbackDays ?? OURA_SYNC_LOOKBACK_DAYS,
     );
   } catch (err) {

src/lib/tz/__tests__/local-day.test.ts

@@ -1,6 +1,46 @@
 import { describe, expect, it } from "vitest";
 import { startOfLocalDayInTz } from "@/lib/tz/local-day";
-import { wallClockInTz } from "@/lib/tz/wall-clock";
+import { wallClockInTz, zonedWallClockToUtc } from "@/lib/tz/wall-clock";
+
+describe("zonedWallClockToUtc", () => {
+  it("resolves a wall clock in a positive-offset zone to the right UTC instant", () => {
+    // 00:30 CEST (UTC+2) on 2026-05-11 → 22:30 UTC on 2026-05-10.
+    const utc = zonedWallClockToUtc(
+      { year: 2026, month: 5, day: 11, hour: 0, minute: 30 },
+      "Europe/Berlin",
+    );
+    expect(utc.toISOString()).toBe("2026-05-10T22:30:00.000Z");
+  });
+
+  it("round-trips through wallClockInTz across the spring-forward boundary", () => {
+    // 03:30 local on Berlin's spring-forward day (clocks jumped 02:00→03:00).
+    const utc = zonedWallClockToUtc(
+      { year: 2026, month: 3, day: 29, hour: 3, minute: 30 },
+      "Europe/Berlin",
+    );
+    const parts = wallClockInTz(utc, "Europe/Berlin");
+    expect(parts.hour).toBe(3);
+    expect(parts.minute).toBe(30);
+    expect(parts.day).toBe(29);
+  });
+
+  it("resolves a negative-offset zone", () => {
+    // 00:30 NZST (UTC+12) on 2026-06-15 → 12:30 UTC on 2026-06-14.
+    const utc = zonedWallClockToUtc(
+      { year: 2026, month: 6, day: 15, hour: 0, minute: 30 },
+      "Pacific/Auckland",
+    );
+    expect(utc.toISOString()).toBe("2026-06-14T12:30:00.000Z");
+  });
+
+  it("falls back to host-local when tz is undefined", () => {
+    const utc = zonedWallClockToUtc(
+      { year: 2026, month: 6, day: 15, hour: 14, minute: 30, second: 15 },
+      undefined,
+    );
+    expect(utc.getTime()).toBe(new Date(2026, 5, 15, 14, 30, 15).getTime());
+  });
+});
 
 describe("startOfLocalDayInTz", () => {
   it("anchors at the user's local midnight, not UTC midnight", () => {

src/lib/tz/wall-clock.ts

@@ -64,6 +64,75 @@ const WALL_CLOCK_OPTIONS: Omit<Intl.DateTimeFormatOptions, "timeZone"> =
     weekday: "short",
   });
 
+/**
+ * The wall-clock fields a `zonedWallClockToUtc` caller supplies. Months are
+ * 1-based to match `WallClockParts`; second defaults to 0 when omitted.
+ */
+export interface WallClockInput {
+  year: number;
+  /** 1–12 (January = 1, December = 12). */
+  month: number;
+  /** 1–31. */
+  day: number;
+  /** 0–23. */
+  hour: number;
+  minute: number;
+  second?: number;
+}
+
+/**
+ * Inverse of `wallClockInTz`: given the wall-clock an observer in `tz` reads
+ * (e.g. an offset-less `2020-01-26T03:02:30` Fitbit sleep timestamp), return the
+ * UTC instant it denotes. DST-safe via the same two-pass converge as
+ * `startOfLocalDayInTz` — the second pass settles the zone offset across a DST
+ * transition for every IANA zone. When `tz` is omitted the parts are read as the
+ * host's system-local clock so offset-less callers keep their legacy shape.
+ */
+export function zonedWallClockToUtc(
+  parts: WallClockInput,
+  tz: string | undefined,
+): Date {
+  const second = parts.second ?? 0;
+  if (!tz) {
+    return new Date(
+      parts.year,
+      parts.month - 1,
+      parts.day,
+      parts.hour,
+      parts.minute,
+      second,
+      0,
+    );
+  }
+  // Treat the wall clock as if it were UTC, then correct by the zone offset at
+  // that approximate instant; one extra pass settles the DST boundary.
+  const asIfUtc = Date.UTC(
+    parts.year,
+    parts.month - 1,
+    parts.day,
+    parts.hour,
+    parts.minute,
+    second,
+    0,
+  );
+  let guess = new Date(asIfUtc);
+  for (let i = 0; i < 2; i++) {
+    const at = wallClockInTz(guess, tz);
+    const guessAsIfUtc = Date.UTC(
+      at.year,
+      at.month - 1,
+      at.day,
+      at.hour,
+      at.minute,
+      at.second,
+      0,
+    );
+    const offsetMin = Math.round((guessAsIfUtc - guess.getTime()) / 60_000);
+    guess = new Date(asIfUtc - offsetMin * 60_000);
+  }
+  return guess;
+}
+
 export function wallClockInTz(
   date: Date,
   tz: string | undefined,

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

@@ -20,6 +20,7 @@ const {
     whoopConnection: {
       findUnique: vi.fn(),
       update: vi.fn(),
+      updateMany: vi.fn(),
       findMany: vi.fn(),
     },
     measurement: {
@@ -118,7 +119,7 @@ describe("getValidToken — rotating refresh", () => {
       refresh_token: "new-refresh",
       expires_in: 3600,
     });
-    prismaMock.whoopConnection.update.mockResolvedValue({});
+    prismaMock.whoopConnection.updateMany.mockResolvedValue({ count: 1 });
 
     const result = await getValidToken("user1");
 
@@ -127,11 +128,39 @@ describe("getValidToken — rotating refresh", () => {
       clientId: "cid",
       clientSecret: "csecret",
     });
-    const updateArg = prismaMock.whoopConnection.update.mock.calls[0]![0];
+    const updateArg = prismaMock.whoopConnection.updateMany.mock.calls[0]![0];
+    // CAS guard: scoped to the connection AND the exact stored ciphertext spent.
+    expect(updateArg.where.id).toBe("conn1");
+    expect(updateArg.where.refreshToken).toBe("enc(old-refresh)");
     expect(updateArg.data.accessToken).toBe("enc(new-access)");
     expect(updateArg.data.refreshToken).toBe("enc(new-refresh)");
   });
 
+  it("reuses the peer's rotated token on a lost CAS race (no spurious reauth)", async () => {
+    prismaMock.whoopConnection.findUnique
+      .mockResolvedValueOnce({
+        id: "conn1",
+        whoopUserId: "42",
+        accessToken: "enc(old-access)",
+        refreshToken: "enc(old-refresh)",
+        tokenExpiresAt: new Date(Date.now() - 1000),
+      })
+      // Re-read after the lost race surfaces the peer's freshly rotated access.
+      .mockResolvedValueOnce({ accessToken: "enc(peer-access)" });
+    refreshAccessTokenMock.mockResolvedValue({
+      access_token: "new-access",
+      refresh_token: "new-refresh",
+      expires_in: 3600,
+    });
+    // A concurrent sync already rotated the token → zero rows match the guard.
+    prismaMock.whoopConnection.updateMany.mockResolvedValue({ count: 0 });
+
+    const result = await getValidToken("user1");
+
+    expect(result?.accessToken).toBe("peer-access");
+    expect(recordSyncFailure).not.toHaveBeenCalled();
+  });
+
   it("returns the stored token without refresh when not near expiry", async () => {
     prismaMock.whoopConnection.findUnique.mockResolvedValue({
       id: "conn1",

src/lib/whoop/sync.ts

@@ -6,7 +6,9 @@
  *   - `getValidToken` decrypts the stored pair, refreshes at
  *     `tokenExpiresAt - 5 min`, and persists BOTH rotated tokens (WHOOP
  *     invalidates the prior access AND refresh token on every refresh — the
- *     same discipline Withings uses for its rotating refresh token).
+ *     same discipline Withings uses for its rotating refresh token). The persist
+ *     is a compare-and-swap on the stored ciphertext so two overlapping syncs
+ *     cannot park the connection at reauth (see `persistRotatedToken`).
  *   - Each per-resource sync (`sync-recovery` / `sync-sleep` / `sync-cycle` /
  *     `sync-workout`) upserts into `Measurement` / `Workout` keyed on
  *     `(userId, type, source = WHOOP, externalId)` so a re-post (WHOOP
@@ -38,6 +40,7 @@ import {
   recomputeBucketsForMeasurement,
 } from "@/lib/rollups/measurement-rollups";
 import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate";
+import { persistRotatedToken } from "@/lib/integrations/oauth-refresh";
 import { refreshAccessToken } from "./client";
 import { getUserWhoopCredentials } from "./credentials";
 import {
@@ -217,20 +220,44 @@ export async function getValidToken(
       const newTokens = await refreshAccessToken(refreshToken, creds);
       const expiresAt = new Date(Date.now() + newTokens.expires_in * 1000);
 
-      // WHOOP rotates the refresh token on every refresh — persist BOTH the
-      // new access token AND the new refresh token, or the next refresh
-      // reuses an invalidated token and the connection drops to reauth.
-      await prisma.whoopConnection.update({
-        where: { id: connection.id },
-        data: {
-          accessToken: encrypt(newTokens.access_token),
-          refreshToken: encrypt(newTokens.refresh_token),
-          tokenExpiresAt: expiresAt,
+      // WHOOP rotates the refresh token on every refresh. Persist BOTH the new
+      // access token AND the new refresh token with a compare-and-swap on the
+      // stored ciphertext: a concurrent sync that already rotated the token is
+      // not clobbered, and THIS caller — if it lost the race — reuses the peer's
+      // rotated access token instead of dropping the connection to reauth.
+      const persistedAccessToken = await persistRotatedToken(
+        newTokens.access_token,
+        {
+          conditionalUpdate: async () => {
+            const { count } = await prisma.whoopConnection.updateMany({
+              // `connection.refreshToken` is the exact ciphertext we read +
+              // spent; matching it is the CAS guard (a re-encrypt would differ).
+              where: {
+                id: connection.id,
+                refreshToken: connection.refreshToken,
+              },
+              data: {
+                accessToken: encrypt(newTokens.access_token),
+                refreshToken: encrypt(newTokens.refresh_token),
+                tokenExpiresAt: expiresAt,
+              },
+            });
+            return count;
+          },
+          readPeerAccessToken: async () => {
+            const fresh = await prisma.whoopConnection.findUnique({
+              where: { id: connection.id },
+              select: { accessToken: true },
+            });
+            return fresh ? decrypt(fresh.accessToken) : null;
+          },
         },
-      });
+      );
+
+      if (!persistedAccessToken) return null;
 
       return {
-        accessToken: newTokens.access_token,
+        accessToken: persistedAccessToken,
         connection: {
           id: connection.id,
           whoopUserId: connection.whoopUserId,