fix(integrations): make rotating-refresh persistence race-safe

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.
This commit is contained in:
Marc-André Bombeck committed 2026-06-22 08:16:04 +02:00
1 parent 68c67e4700
commit 0f5e563403
9 files changed
+348 -44

No files matched your search

+32 -4
View File
@@ -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",
+42 -14
View File
@@ -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,
@@ -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();
});
});
+69
View File
@@ -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();
}
+38 -3
View File
@@ -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());
+43 -7
View File
@@ -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;
},
});
}
+13 -2
View File
@@ -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) {
+31 -2
View File
@@ -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",
+39 -12
View File
@@ -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,