From 0f5e5634031a42cb756eaf73c98c10ee8fab224a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Mon, 22 Jun 2026 08:16:04 +0200 Subject: [PATCH] fix(integrations): make rotating-refresh persistence race-safe MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fitbit, WHOOP, and Oura all issue one-time-use refresh tokens. The token persistence did an unguarded read → refresh → write, so two overlapping syncs for one user both read the same stored refresh token and both POST it. The provider honours only the first; the loser's refresh comes back 401, gets classified as reauth, and the connection is parked at `error_reauth` spuriously even though the peer rotated the token successfully. Add a shared `persistRotatedToken` helper that writes the rotated pair with a compare-and-swap on the stored ciphertext (matched verbatim, since AES-GCM re-encryption of the same plaintext differs byte-for-byte). When the conditional update touches zero rows a peer already won the race; re-read the row and reuse the peer's freshly rotated access token instead of erroring. Wire it into all three providers — Fitbit and WHOOP `getValidToken` (proactive expiry refresh, keyed by connection id) and Oura `storeOuraTokens` (reactive 401 refresh, keyed by user id). Both racing callers now end with a usable token and neither drops the connection. --- src/lib/fitbit/__tests__/sync.test.ts | 36 ++++++++-- src/lib/fitbit/sync.ts | 56 +++++++++++---- .../__tests__/oauth-refresh.test.ts | 41 +++++++++++ src/lib/integrations/oauth-refresh.ts | 69 +++++++++++++++++++ src/lib/oura/__tests__/sync.test.ts | 41 ++++++++++- src/lib/oura/credentials.ts | 50 ++++++++++++-- src/lib/oura/sync.ts | 15 +++- src/lib/whoop/__tests__/sync.test.ts | 33 ++++++++- src/lib/whoop/sync.ts | 51 ++++++++++---- 9 files changed, 348 insertions(+), 44 deletions(-) create mode 100644 src/lib/integrations/__tests__/oauth-refresh.test.ts create mode 100644 src/lib/integrations/oauth-refresh.ts diff --git a/src/lib/fitbit/__tests__/sync.test.ts b/src/lib/fitbit/__tests__/sync.test.ts index 58be146a8..6d4047964 100644 --- a/src/lib/fitbit/__tests__/sync.test.ts +++ b/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) => Promise>( @@ -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", diff --git a/src/lib/fitbit/sync.ts b/src/lib/fitbit/sync.ts index f85eda2ad..6abf49dcd 100644 --- a/src/lib/fitbit/sync.ts +++ b/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, diff --git a/src/lib/integrations/__tests__/oauth-refresh.test.ts b/src/lib/integrations/__tests__/oauth-refresh.test.ts new file mode 100644 index 000000000..8f7185107 --- /dev/null +++ b/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(); + }); +}); diff --git a/src/lib/integrations/oauth-refresh.ts b/src/lib/integrations/oauth-refresh.ts new file mode 100644 index 000000000..2298724d8 --- /dev/null +++ b/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 = ` 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; + /** + * 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; +} + +/** + * 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 { + 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(); +} diff --git a/src/lib/oura/__tests__/sync.test.ts b/src/lib/oura/__tests__/sync.test.ts index 52cd7742a..e452322f1 100644 --- a/src/lib/oura/__tests__/sync.test.ts +++ b/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()); diff --git a/src/lib/oura/credentials.ts b/src/lib/oura/credentials.ts index 32e9df380..bcf74f3a6 100644 --- a/src/lib/oura/credentials.ts +++ b/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 { - await prisma.user.update({ - where: { id: userId }, - data: { - ouraAccessTokenEncrypted: encrypt(accessToken), - ouraRefreshTokenEncrypted: encrypt(refreshToken), + expectedRefreshCiphertext: string, +): Promise { + 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; }, }); } diff --git a/src/lib/oura/sync.ts b/src/lib/oura/sync.ts index cefddf77b..363b3c485 100644 --- a/src/lib/oura/sync.ts +++ b/src/lib/oura/sync.ts @@ -124,6 +124,7 @@ async function fetchAll( userId: string, accessToken: string, refreshToken: string, + refreshTokenCiphertext: string, lookbackDays: number, ): Promise { 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) { diff --git a/src/lib/whoop/__tests__/sync.test.ts b/src/lib/whoop/__tests__/sync.test.ts index 33d5afd2f..bec3e8df8 100644 --- a/src/lib/whoop/__tests__/sync.test.ts +++ b/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", diff --git a/src/lib/whoop/sync.ts b/src/lib/whoop/sync.ts index baf009137..5015819ea 100644 --- a/src/lib/whoop/sync.ts +++ b/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,