From 5d784a639cb5ebb76e4a8a64bbc84f4f68f63300 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Mon, 20 Jul 2026 12:09:32 +0200 Subject: [PATCH] fix(integrations): serialize refresh and cursor writes --- src/lib/integrations/oauth-refresh.ts | 31 +++ src/lib/whoop/__tests__/sync.test.ts | 143 ++++++++++-- src/lib/whoop/sync.ts | 215 ++++++++--------- .../__tests__/sync-measure-watermark.test.ts | 49 +++- src/lib/withings/sync.ts | 99 ++++---- ...ovider-token-refresh-serialization.test.ts | 220 ++++++++++++++++++ 6 files changed, 590 insertions(+), 167 deletions(-) create mode 100644 tests/integration/provider-token-refresh-serialization.test.ts diff --git a/src/lib/integrations/oauth-refresh.ts b/src/lib/integrations/oauth-refresh.ts index 87276e76f..89c424051 100644 --- a/src/lib/integrations/oauth-refresh.ts +++ b/src/lib/integrations/oauth-refresh.ts @@ -1,3 +1,5 @@ +import type { Prisma } from "@/generated/prisma/client"; + /** * Shared OAuth rotating-refresh-token persistence helper. * @@ -58,3 +60,32 @@ export async function persistRotatedToken( // rather than parking the connection at reauth. return opts.readPeerAccessToken(); } + +export type SerializedRefreshProvider = "whoop" | "withings"; + +/** + * Provider refreshes hold a database transaction open around one bounded + * external request. `maxWait` bounds pool acquisition and `timeout` bounds the + * advisory-lock wait, provider call, and durable token replacement together. + */ +export const PROVIDER_REFRESH_TRANSACTION_OPTIONS = { + maxWait: 10_000, + timeout: 60_000, +} as const; + +/** + * Serialize one provider/user refresh attempt for the transaction lifetime. + * Callers must re-read token state after this returns: a prior contender may + * have committed a fresh token while this transaction waited for the lock. + */ +export async function acquireProviderTokenRefreshLock( + tx: Pick, + provider: SerializedRefreshProvider, + userId: string, +): Promise { + const key = `provider-token-refresh:${provider}:${userId}`; + await tx.$queryRaw` + SELECT 1 AS locked + FROM pg_advisory_xact_lock(hashtextextended(${key}, 0)) + `; +} diff --git a/src/lib/whoop/__tests__/sync.test.ts b/src/lib/whoop/__tests__/sync.test.ts index 91dff2c21..c8e56648d 100644 --- a/src/lib/whoop/__tests__/sync.test.ts +++ b/src/lib/whoop/__tests__/sync.test.ts @@ -31,6 +31,7 @@ const { findMany: vi.fn(), }, $executeRaw: vi.fn(), + $queryRaw: vi.fn(), $transaction: vi.fn(), }, refreshAccessTokenMock: vi.fn(), @@ -113,8 +114,8 @@ import { syncUserRecovery } from "../sync-recovery"; beforeEach(() => { vi.clearAllMocks(); - prismaMock.$transaction.mockImplementation(async (run: (tx: unknown) => unknown) => - run({}), + prismaMock.$transaction.mockImplementation( + async (run: (tx: unknown) => unknown) => run(prismaMock), ); reconcileMock.mockImplementation( async (_tx: unknown, input: { type: string; measuredAt: Date; externalId: string }) => ({ @@ -150,7 +151,7 @@ describe("getValidToken — rotating refresh", () => { refresh_token: "new-refresh", expires_in: 3600, }); - prismaMock.whoopConnection.updateMany.mockResolvedValue({ count: 1 }); + prismaMock.whoopConnection.update.mockResolvedValue({}); const result = await getValidToken("user1"); @@ -159,15 +160,15 @@ describe("getValidToken — rotating refresh", () => { clientId: "cid", clientSecret: "csecret", }); - const updateArg = prismaMock.whoopConnection.updateMany.mock.calls[0]![0]; - // CAS guard: scoped to the connection AND the exact stored ciphertext spent. + const updateArg = prismaMock.whoopConnection.update.mock.calls[0]![0]; 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)"); + expect(prismaMock.$queryRaw).toHaveBeenCalledTimes(1); }); - it("reuses the peer's rotated token on a lost CAS race (no spurious reauth)", async () => { + + it("re-reads token state under an advisory lock and reuses a peer winner", async () => { prismaMock.whoopConnection.findUnique .mockResolvedValueOnce({ id: "conn1", @@ -176,20 +177,26 @@ describe("getValidToken — rotating refresh", () => { 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 }); + .mockResolvedValueOnce({ + id: "conn1", + whoopUserId: "42", + accessToken: "enc(winner-access)", + refreshToken: "enc(winner-refresh)", + tokenExpiresAt: new Date(Date.now() + 60 * 60 * 1000), + }); const result = await getValidToken("user1"); - expect(result?.accessToken).toBe("peer-access"); - expect(recordSyncFailure).not.toHaveBeenCalled(); + expect(result?.accessToken).toBe("winner-access"); + expect(prismaMock.$queryRaw).toHaveBeenCalledTimes(1); + expect(refreshAccessTokenMock).not.toHaveBeenCalled(); + expect(prismaMock.$transaction).toHaveBeenCalledWith( + expect.any(Function), + expect.objectContaining({ + maxWait: expect.any(Number), + timeout: expect.any(Number), + }), + ); }); it("returns the stored token without refresh when not near expiry", async () => { @@ -418,6 +425,13 @@ describe("upsertWhoopMeasurements — exact insertion results", () => { source: "WHOOP", externalId: readings[1].externalId, }); + expect(prismaMock.$transaction).toHaveBeenCalledWith( + expect.any(Function), + expect.objectContaining({ + maxWait: expect.any(Number), + timeout: expect.any(Number), + }), + ); }); it("does not pre-probe and still emits a genuine insert", async () => { @@ -459,6 +473,99 @@ describe("upsertWhoopMeasurements — exact insertion results", () => { }); }); +describe("syncUserRecovery — durable cursor ordering", () => { + const recovery = { + cycle_id: 1, + sleep_id: "sleep-uuid", + user_id: 42, + created_at: "2026-06-01T06:00:00.000Z", + updated_at: "2026-06-01T07:00:00.000Z", + score_state: "SCORED", + score: { + user_calibrating: false, + recovery_score: 66, + resting_heart_rate: 52, + hrv_rmssd_milli: 48.7, + spo2_percentage: 97, + skin_temp_celsius: 33.4, + }, + }; + + beforeEach(() => { + prismaMock.whoopConnection.findUnique.mockResolvedValue({ + id: "conn1", + whoopUserId: "42", + accessToken: "enc(live-access)", + refreshToken: "enc(live-refresh)", + tokenExpiresAt: new Date(Date.now() + 60 * 60 * 1000), + lastSyncedAt: new Date("2026-06-01T00:00:00Z"), + resourceCursors: { + recovery: "2026-06-01T00:00:00.000Z", + }, + }); + fetchRecoveriesMock.mockResolvedValue([recovery]); + }); + + it("keeps the cursor unchanged when any represented measurement write fails", async () => { + reconcileMock + .mockResolvedValueOnce({ + status: "inserted", + row: { + id: "inserted:first", + type: "RECOVERY_SCORE", + measuredAt: new Date(recovery.updated_at), + externalId: "sleep-uuid:recovery_score", + }, + }) + .mockResolvedValueOnce({ + status: "failed", + reason: "db_error", + error: new Error("injected write failure"), + }); + + await expect(syncUserRecovery("user1")).rejects.toThrow(); + + expect(prismaMock.$executeRaw).not.toHaveBeenCalled(); + }); + + it("advances after success and replay updates identities without new arrivals", async () => { + expect(await syncUserRecovery("user1")).toBe(5); + expect(prismaMock.$executeRaw).toHaveBeenCalledTimes(1); + + vi.clearAllMocks(); + prismaMock.$transaction.mockImplementation( + async (run: (tx: unknown) => unknown) => run(prismaMock), + ); + prismaMock.whoopConnection.findUnique.mockResolvedValue({ + id: "conn1", + whoopUserId: "42", + accessToken: "enc(live-access)", + refreshToken: "enc(live-refresh)", + tokenExpiresAt: new Date(Date.now() + 60 * 60 * 1000), + lastSyncedAt: new Date("2026-06-01T00:00:00Z"), + resourceCursors: { + recovery: "2026-06-01T00:00:00.000Z", + }, + }); + fetchRecoveriesMock.mockResolvedValue([recovery]); + reconcileMock.mockImplementation( + async (_tx: unknown, input: { type: string; measuredAt: Date; externalId: string }) => ({ + status: "updated", + row: { + id: `existing:${input.externalId}`, + type: input.type, + measuredAt: input.measuredAt, + externalId: input.externalId, + }, + }), + ); + + expect(await syncUserRecovery("user1")).toBe(5); + expect(prismaMock.$executeRaw).toHaveBeenCalledTimes(1); + expect(emitArrivalMock).toHaveBeenCalledWith("user1", [], "whoop"); + }); +}); + describe("isCollectionForbidden — tier degradation gate", () => { it("is true only for a WhoopApiError carrying HTTP 403", () => { const forbidden = new WhoopApiError({ diff --git a/src/lib/whoop/sync.ts b/src/lib/whoop/sync.ts index c3802ba14..c275a49da 100644 --- a/src/lib/whoop/sync.ts +++ b/src/lib/whoop/sync.ts @@ -6,9 +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). The persist - * is a compare-and-swap on the stored ciphertext so two overlapping syncs - * cannot park the connection at reauth (see `persistRotatedToken`). + * same discipline Withings uses for its rotating refresh token). A + * provider/user advisory lock serializes refresh, and token state is + * re-read inside the transaction before the one-time token is spent. * - 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 @@ -44,7 +44,10 @@ import { emitInsertedMeasurementArrivals, type InsertedMeasurementArrivalRow, } from "@/lib/arrivals/measurement-emit"; -import { persistRotatedToken } from "@/lib/integrations/oauth-refresh"; +import { + acquireProviderTokenRefreshLock, + PROVIDER_REFRESH_TRANSACTION_OPTIONS, +} from "@/lib/integrations/oauth-refresh"; import { MeasurementReconciliationError, reconcileExternalMeasurement, @@ -202,91 +205,86 @@ export async function getValidToken( }); if (!connection) return null; - const accessToken = decrypt(connection.accessToken); - const refreshToken = decrypt(connection.refreshToken); - if ( - connection.tokenExpiresAt.getTime() - TOKEN_REFRESH_BUFFER_MS < + connection.tokenExpiresAt.getTime() - TOKEN_REFRESH_BUFFER_MS >= Date.now() ) { - try { - const creds = await getUserWhoopCredentials(userId); - if (!creds) { - getEvent()?.addWarning( - `No WHOOP credentials found for user ${userId} during token refresh`, - ); - await recordSyncFailure({ - userId, - integration: "whoop", - kind: "reauth_required", - message: "WHOOP credentials missing — token refresh skipped", - errorCode: "credentials_missing", - }); - return null; - } - - 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 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: persistedAccessToken, - connection: { - id: connection.id, - whoopUserId: connection.whoopUserId, - }, - }; - } catch (err) { - getEvent()?.addWarning( - `WHOOP token refresh failed for user ${userId}: ${err}`, - ); - await recordWhoopSyncFailure(userId, err); - return null; - } + return { + accessToken: decrypt(connection.accessToken), + connection: { + id: connection.id, + whoopUserId: connection.whoopUserId, + }, + }; } - return { - accessToken, - connection: { - id: connection.id, - whoopUserId: connection.whoopUserId, - }, - }; + const creds = await getUserWhoopCredentials(userId); + if (!creds) { + getEvent()?.addWarning( + `No WHOOP credentials found for user ${userId} during token refresh`, + ); + await recordSyncFailure({ + userId, + integration: "whoop", + kind: "reauth_required", + message: "WHOOP credentials missing — token refresh skipped", + errorCode: "credentials_missing", + }); + return null; + } + + try { + return await prisma.$transaction(async (tx) => { + await acquireProviderTokenRefreshLock(tx, "whoop", userId); + + const current = await tx.whoopConnection.findUnique({ + where: { userId }, + }); + if (!current) return null; + + if ( + current.tokenExpiresAt.getTime() - TOKEN_REFRESH_BUFFER_MS >= + Date.now() + ) { + return { + accessToken: decrypt(current.accessToken), + connection: { + id: current.id, + whoopUserId: current.whoopUserId, + }, + }; + } + + const newTokens = await refreshAccessToken( + decrypt(current.refreshToken), + creds, + ); + const expiresAt = new Date(Date.now() + newTokens.expires_in * 1000); + + await tx.whoopConnection.update({ + where: { id: current.id }, + data: { + accessToken: encrypt(newTokens.access_token), + refreshToken: encrypt(newTokens.refresh_token), + tokenExpiresAt: expiresAt, + }, + }); + + return { + accessToken: newTokens.access_token, + connection: { + id: current.id, + whoopUserId: current.whoopUserId, + }, + }; + }, PROVIDER_REFRESH_TRANSACTION_OPTIONS); + } catch (err) { + getEvent()?.addWarning( + `WHOOP token refresh failed for user ${userId}: ${err}`, + ); + await recordWhoopSyncFailure(userId, err); + return null; + } } /** @@ -350,30 +348,33 @@ export async function upsertWhoopMeasurements( InsertedMeasurementArrivalRow & { externalId: string | null } > = []; - const verdicts = await prisma.$transaction(async (tx) => { - const outcomes = []; - for (const r of readings) { - const verdict = await reconcileExternalMeasurement( - tx, - { - userId, - type: r.type as MeasurementType, - source: "WHOOP", - value: r.value, - unit: r.unit, - measuredAt: r.measuredAt, - externalId: r.externalId, - sleepStage: r.sleepStage ?? null, - }, - { exactExternalMatch: "update" }, - ); - if (verdict.status === "failed") { - throw new MeasurementReconciliationError(verdict); + const verdicts = await prisma.$transaction( + async (tx) => { + const outcomes = []; + for (const r of readings) { + const verdict = await reconcileExternalMeasurement( + tx, + { + userId, + type: r.type as MeasurementType, + source: "WHOOP", + value: r.value, + unit: r.unit, + measuredAt: r.measuredAt, + externalId: r.externalId, + sleepStage: r.sleepStage ?? null, + }, + { exactExternalMatch: "update" }, + ); + if (verdict.status === "failed") { + throw new MeasurementReconciliationError(verdict); + } + outcomes.push(verdict); } - outcomes.push(verdict); - } - return outcomes; - }, { timeout: 60_000 }); + return outcomes; + }, + { maxWait: 10_000, timeout: 60_000 }, + ); for (let index = 0; index < readings.length; index++) { const reading = readings[index]!; diff --git a/src/lib/withings/__tests__/sync-measure-watermark.test.ts b/src/lib/withings/__tests__/sync-measure-watermark.test.ts index 57a542171..c81a9d4a6 100644 --- a/src/lib/withings/__tests__/sync-measure-watermark.test.ts +++ b/src/lib/withings/__tests__/sync-measure-watermark.test.ts @@ -24,6 +24,8 @@ vi.mock("@/lib/db", () => ({ findUnique: vi.fn(), update: vi.fn(), }, + $queryRaw: vi.fn(), + $transaction: vi.fn(), }, })); @@ -38,6 +40,13 @@ vi.mock("../client", () => ({ subscribeWebhook: vi.fn(), })); +vi.mock("../credentials", () => ({ + getUserWithingsCredentials: vi.fn(async () => ({ + clientId: "cid", + clientSecret: "secret", + })), +})); + vi.mock("@/lib/integrations/status", async () => { const actual = await vi.importActual< typeof import("@/lib/integrations/status") @@ -77,13 +86,13 @@ vi.mock("@/lib/logging/context", () => ({ })); import { prisma } from "@/lib/db"; -import { fetchMeasurements } from "../client"; +import { fetchMeasurements, refreshAccessToken } from "../client"; import { recordSyncFailure, recordSyncSuccess, } from "@/lib/integrations/status"; -import { syncUserMeasurements } from "../sync"; +import { getValidToken, syncUserMeasurements } from "../sync"; const P2002 = Object.assign(new Error("Unique constraint failed"), { code: "P2002", @@ -91,6 +100,9 @@ const P2002 = Object.assign(new Error("Unique constraint failed"), { beforeEach(() => { vi.clearAllMocks(); + vi.mocked(prisma.$transaction).mockImplementation(async (run) => + run(prisma as never), + ); // A live, non-expired connection so `getValidToken` returns the token // without touching the refresh path. vi.mocked(prisma.withingsConnection.findUnique).mockResolvedValue({ @@ -107,6 +119,39 @@ beforeEach(() => { vi.mocked(prisma.measurement.update).mockResolvedValue({} as never); }); +describe("getValidToken — serialized rotating refresh", () => { + it("re-reads token state under an advisory lock and reuses a peer winner", async () => { + vi.mocked(prisma.withingsConnection.findUnique) + .mockResolvedValueOnce({ + id: "conn-1", + withingsUserId: "wu-1", + accessToken: "old-access", + refreshToken: "old-refresh", + tokenExpiresAt: new Date(Date.now() - 1000), + } as never) + .mockResolvedValueOnce({ + id: "conn-1", + withingsUserId: "wu-1", + accessToken: "winner-access", + refreshToken: "winner-refresh", + tokenExpiresAt: new Date(Date.now() + 60 * 60 * 1000), + } as never); + + const result = await getValidToken("user-1"); + + expect(result?.accessToken).toBe("winner-access"); + expect(prisma.$queryRaw).toHaveBeenCalledTimes(1); + expect(refreshAccessToken).not.toHaveBeenCalled(); + expect(prisma.$transaction).toHaveBeenCalledWith( + expect.any(Function), + expect.objectContaining({ + maxWait: expect.any(Number), + timeout: expect.any(Number), + }), + ); + }); +}); + describe("syncUserMeasurements — watermark hold on hard row failure (F4)", () => { it("holds lastSyncedAt and records a failure when a row hard-fails", async () => { vi.mocked(fetchMeasurements).mockResolvedValue([ diff --git a/src/lib/withings/sync.ts b/src/lib/withings/sync.ts index 50f660f7f..ac0e12561 100644 --- a/src/lib/withings/sync.ts +++ b/src/lib/withings/sync.ts @@ -32,6 +32,10 @@ import { } from "@/lib/rollups/measurement-rollups"; import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; import { invalidateUserDashboardSnapshot } from "@/lib/cache/invalidate"; +import { + acquireProviderTokenRefreshLock, + PROVIDER_REFRESH_TRANSACTION_OPTIONS, +} from "@/lib/integrations/oauth-refresh"; /** * Build the callback URL handed to Withings at `Notify.subscribe` time. @@ -74,35 +78,58 @@ export async function getValidToken(userId: string): Promise<{ if (!connection) return null; - const accessToken = decrypt(connection.accessToken); - const refreshToken = decrypt(connection.refreshToken); + if (connection.tokenExpiresAt.getTime() - 5 * 60 * 1000 >= Date.now()) { + return { + accessToken: decrypt(connection.accessToken), + connection: { + id: connection.id, + withingsUserId: connection.withingsUserId, + }, + }; + } - // Check if token is expired (with 5 min buffer) - if (connection.tokenExpiresAt.getTime() - 5 * 60 * 1000 < Date.now()) { - try { - const creds = await getUserWithingsCredentials(userId); - if (!creds) { - getEvent()?.addWarning( - `No credentials found for user ${userId} during token refresh`, - ); - // Without credentials we cannot refresh OR re-authenticate — the - // user has to re-enter the client_id/secret. Mark as reauth so - // scheduled syncs back off. - await recordSyncFailure({ - userId, - integration: "withings", - kind: "reauth_required", - message: "Withings credentials missing — token refresh skipped", - errorCode: "credentials_missing", - }); - return null; + const creds = await getUserWithingsCredentials(userId); + if (!creds) { + getEvent()?.addWarning( + `No credentials found for user ${userId} during token refresh`, + ); + await recordSyncFailure({ + userId, + integration: "withings", + kind: "reauth_required", + message: "Withings credentials missing — token refresh skipped", + errorCode: "credentials_missing", + }); + return null; + } + + try { + return await prisma.$transaction(async (tx) => { + await acquireProviderTokenRefreshLock(tx, "withings", userId); + + const current = await tx.withingsConnection.findUnique({ + where: { userId }, + }); + if (!current) return null; + + if (current.tokenExpiresAt.getTime() - 5 * 60 * 1000 >= Date.now()) { + return { + accessToken: decrypt(current.accessToken), + connection: { + id: current.id, + withingsUserId: current.withingsUserId, + }, + }; } - const newTokens = await refreshAccessToken(refreshToken, creds); - + const newTokens = await refreshAccessToken( + decrypt(current.refreshToken), + creds, + ); const expiresAt = new Date(Date.now() + newTokens.expires_in * 1000); - await prisma.withingsConnection.update({ - where: { id: connection.id }, + + await tx.withingsConnection.update({ + where: { id: current.id }, data: { accessToken: encrypt(newTokens.access_token), refreshToken: encrypt(newTokens.refresh_token), @@ -113,24 +140,16 @@ export async function getValidToken(userId: string): Promise<{ return { accessToken: newTokens.access_token, connection: { - id: connection.id, - withingsUserId: connection.withingsUserId, + id: current.id, + withingsUserId: current.withingsUserId, }, }; - } catch (err) { - getEvent()?.addWarning(`Token refresh failed for user ${userId}: ${err}`); - await recordWithingsSyncFailure(userId, err); - return null; - } + }, PROVIDER_REFRESH_TRANSACTION_OPTIONS); + } catch (err) { + getEvent()?.addWarning(`Token refresh failed for user ${userId}: ${err}`); + await recordWithingsSyncFailure(userId, err); + return null; } - - return { - accessToken, - connection: { - id: connection.id, - withingsUserId: connection.withingsUserId, - }, - }; } /** diff --git a/tests/integration/provider-token-refresh-serialization.test.ts b/tests/integration/provider-token-refresh-serialization.test.ts new file mode 100644 index 000000000..cc8afa424 --- /dev/null +++ b/tests/integration/provider-token-refresh-serialization.test.ts @@ -0,0 +1,220 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import type * as WhoopClientModule from "@/lib/whoop/client"; +import type * as WithingsClientModule from "@/lib/withings/client"; + +import { getPrismaClient, truncateAllTables } from "./setup"; + +process.env.ENCRYPTION_KEY = + "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; +delete process.env.ENCRYPTION_KEYS; +delete process.env.ENCRYPTION_ACTIVE_KEY_ID; + +const { whoopRefresh, withingsRefresh, whoopRecoveries } = vi.hoisted(() => ({ + whoopRefresh: vi.fn(), + withingsRefresh: vi.fn(), + whoopRecoveries: vi.fn(), +})); + +vi.mock("@/lib/whoop/client", async (importOriginal) => ({ + ...(await importOriginal()), + refreshAccessToken: whoopRefresh, + fetchRecoveries: whoopRecoveries, +})); + +vi.mock("@/lib/withings/client", async (importOriginal) => ({ + ...(await importOriginal()), + refreshAccessToken: withingsRefresh, +})); + +import { encrypt, decrypt } from "@/lib/crypto"; +import { getValidToken as getValidWhoopToken } from "@/lib/whoop/sync"; +import { syncUserRecovery } from "@/lib/whoop/sync-recovery"; +import { getValidToken as getValidWithingsToken } from "@/lib/withings/sync"; + +const CONTENDERS = 75; +const EXPIRED_AT = new Date("2026-01-01T00:00:00.000Z"); + +async function seedProviderUser(provider: "whoop" | "withings") { + const prisma = getPrismaClient(); + const userId = `refresh-${provider}`; + await prisma.user.create({ + data: { + id: userId, + username: `refresh-${provider}`, + email: `refresh-${provider}@example.test`, + ...(provider === "whoop" + ? { + whoopClientIdEncrypted: encrypt("whoop-client"), + whoopClientSecretEncrypted: encrypt("whoop-secret"), + } + : { + withingsClientIdEncrypted: encrypt("withings-client"), + withingsClientSecretEncrypted: encrypt("withings-secret"), + }), + }, + }); + + if (provider === "whoop") { + await prisma.whoopConnection.create({ + data: { + userId, + whoopUserId: "whoop-user", + accessToken: encrypt("stale-access"), + refreshToken: encrypt("stale-refresh"), + tokenExpiresAt: EXPIRED_AT, + }, + }); + } else { + await prisma.withingsConnection.create({ + data: { + userId, + withingsUserId: "withings-user", + accessToken: encrypt("stale-access"), + refreshToken: encrypt("stale-refresh"), + tokenExpiresAt: EXPIRED_AT, + }, + }); + } + + return userId; +} + +async function runContenders( + provider: "whoop" | "withings", + userId: string, +) { + const getToken = + provider === "whoop" ? getValidWhoopToken : getValidWithingsToken; + return Promise.all( + Array.from({ length: CONTENDERS }, () => getToken(userId)), + ); +} + +beforeEach(async () => { + await truncateAllTables(getPrismaClient()); + vi.clearAllMocks(); +}); + +describe.each([ + ["whoop", whoopRefresh], + ["withings", withingsRefresh], +] as const)("%s rotating token refresh serialization (real Postgres)", (provider, refresh) => { + it("lets exactly one contender refresh a stale token and shares the winner", async () => { + const userId = await seedProviderUser(provider); + refresh.mockImplementation(async (refreshToken: string) => { + expect(refreshToken).toBe("stale-refresh"); + return { + access_token: "winner-access", + refresh_token: "winner-refresh", + expires_in: 3600, + }; + }); + + const results = await runContenders(provider, userId); + + expect(refresh).toHaveBeenCalledTimes(1); + expect(results).toHaveLength(CONTENDERS); + expect(results.every((result) => result?.accessToken === "winner-access")).toBe( + true, + ); + + const prisma = getPrismaClient(); + const row = + provider === "whoop" + ? await prisma.whoopConnection.findUniqueOrThrow({ where: { userId } }) + : await prisma.withingsConnection.findUniqueOrThrow({ where: { userId } }); + expect(decrypt(row.accessToken)).toBe("winner-access"); + expect(decrypt(row.refreshToken)).toBe("winner-refresh"); + }); + + it("rolls back a failed winner so a later contender can retry", async () => { + const userId = await seedProviderUser(provider); + refresh + .mockRejectedValueOnce(new Error("injected refresh failure")) + .mockResolvedValue({ + access_token: "retry-access", + refresh_token: "retry-refresh", + expires_in: 3600, + }); + + const results = await runContenders(provider, userId); + + expect(refresh).toHaveBeenCalledTimes(2); + expect(results.filter((result) => result === null)).toHaveLength(1); + expect( + results.filter((result) => result?.accessToken === "retry-access"), + ).toHaveLength(CONTENDERS - 1); + + const prisma = getPrismaClient(); + const row = + provider === "whoop" + ? await prisma.whoopConnection.findUniqueOrThrow({ where: { userId } }) + : await prisma.withingsConnection.findUniqueOrThrow({ where: { userId } }); + expect(decrypt(row.accessToken)).toBe("retry-access"); + expect(decrypt(row.refreshToken)).toBe("retry-refresh"); + }); +}); + +describe("WHOOP recovery cursor durability (real Postgres)", () => { + const oldCursor = "2026-06-01T00:00:00.000Z"; + const recovery = { + cycle_id: 1, + sleep_id: "sleep-real", + user_id: 42, + created_at: "2026-06-01T06:00:00.000Z", + updated_at: "2026-06-01T07:00:00.000Z", + score_state: "SCORED", + score: { + user_calibrating: false, + recovery_score: 66, + resting_heart_rate: 52, + hrv_rmssd_milli: 48.7, + spo2_percentage: 97, + skin_temp_celsius: 33.4, + }, + }; + + it("holds the cursor on a durable write failure, then advances idempotently", async () => { + const userId = await seedProviderUser("whoop"); + const prisma = getPrismaClient(); + await prisma.whoopConnection.update({ + where: { userId }, + data: { + tokenExpiresAt: new Date(Date.now() + 60 * 60 * 1000), + lastSyncedAt: new Date(oldCursor), + resourceCursors: { recovery: oldCursor }, + }, + }); + whoopRecoveries.mockResolvedValue([recovery]); + + await prisma.$executeRaw` + ALTER TABLE "measurements" + ADD CONSTRAINT "test_whoop_recovery_write_failure" + CHECK ("external_id" <> 'sleep-real:hrv_rmssd') + `; + + try { + await expect(syncUserRecovery(userId)).rejects.toThrow(); + const failed = await prisma.whoopConnection.findUniqueOrThrow({ + where: { userId }, + }); + expect(failed.resourceCursors).toEqual({ recovery: oldCursor }); + expect(await prisma.measurement.count({ where: { userId } })).toBe(0); + } finally { + await prisma.$executeRaw` + ALTER TABLE "measurements" + DROP CONSTRAINT IF EXISTS "test_whoop_recovery_write_failure" + `; + } + + expect(await syncUserRecovery(userId)).toBe(5); + const succeeded = await prisma.whoopConnection.findUniqueOrThrow({ + where: { userId }, + }); + expect(succeeded.resourceCursors).not.toEqual({ recovery: oldCursor }); + expect(await prisma.measurement.count({ where: { userId } })).toBe(5); + + expect(await syncUserRecovery(userId)).toBe(5); + expect(await prisma.measurement.count({ where: { userId } })).toBe(5); + }); +});