diff --git a/src/lib/whoop/__tests__/sync-dead-token.test.ts b/src/lib/whoop/__tests__/sync-dead-token.test.ts index c2e9f4c52..a5d51027e 100644 --- a/src/lib/whoop/__tests__/sync-dead-token.test.ts +++ b/src/lib/whoop/__tests__/sync-dead-token.test.ts @@ -32,6 +32,7 @@ const { whoopConnection: { findUnique: vi.fn(), update: vi.fn(async () => ({})), + updateMany: vi.fn(async () => ({ count: 1 })), }, $transaction: vi.fn(), }, @@ -131,6 +132,9 @@ beforeEach(() => { isReauthRequiredMock.mockResolvedValue(false); prismaMock.whoopConnection.findUnique.mockReset(); prismaMock.whoopConnection.update.mockReset().mockResolvedValue({}); + prismaMock.whoopConnection.updateMany + .mockReset() + .mockResolvedValue({ count: 1 }); getUserWhoopCredentialsMock.mockReset().mockResolvedValue(null); refreshAccessTokenMock.mockReset(); // The advisory-lock transaction runs its callback against the same mock. diff --git a/src/lib/whoop/__tests__/sync.test.ts b/src/lib/whoop/__tests__/sync.test.ts index 969bc0ee3..85462399e 100644 --- a/src/lib/whoop/__tests__/sync.test.ts +++ b/src/lib/whoop/__tests__/sync.test.ts @@ -169,7 +169,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"); @@ -178,15 +178,28 @@ describe("getValidToken — rotating refresh", () => { clientId: "cid", clientSecret: "csecret", }); - const updateArg = prismaMock.whoopConnection.update.mock.calls[0]![0]; - expect(updateArg.where.id).toBe("conn1"); + // Written only while the stored refresh token is still the one spent. + const updateArg = prismaMock.whoopConnection.updateMany.mock.calls[0]![0]; + expect(updateArg.where).toEqual({ + userId: "user1", + refreshToken: "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("re-reads token state under an advisory lock and reuses a peer winner", async () => { + const stale = { + id: "conn1", + whoopUserId: "42", + accessToken: "enc(old-access)", + refreshToken: "enc(old-refresh)", + tokenExpiresAt: new Date(Date.now() - 1000), + }; prismaMock.whoopConnection.findUnique + // The connection lookup, then the token read before the lock. + .mockResolvedValueOnce(stale) .mockResolvedValueOnce({ id: "conn1", whoopUserId: "42", diff --git a/src/lib/whoop/oauth-descriptor.ts b/src/lib/whoop/oauth-descriptor.ts new file mode 100644 index 000000000..8a86ac2a4 --- /dev/null +++ b/src/lib/whoop/oauth-descriptor.ts @@ -0,0 +1,51 @@ +/** + * Where a WHOOP grant lives and how its token endpoint is called, for the + * shared token lifecycle (`src/lib/integrations/oauth-kit/token.ts`). + * + * WHOOP keeps an expiry on `WhoopConnection`, so the token is refreshed ahead + * of it, and rotates the refresh token on every use. A connection without a + * WHOOP user id is a connect still in flight and has no usable grant yet. + */ +import type { OAuthTokenDescriptor } from "@/lib/integrations/oauth-kit/token"; +import { refreshAccessToken, type WhoopCredentials } from "./client"; +import { getUserWhoopCredentials } from "./credentials"; + +/** Refresh this long before the stored expiry. */ +export const WHOOP_TOKEN_REFRESH_BUFFER_MS = 5 * 60 * 1000; + +export const whoopTokenDescriptor: OAuthTokenDescriptor = { + provider: "whoop", + tracksExpiry: true, + expiryBufferMs: WHOOP_TOKEN_REFRESH_BUFFER_MS, + async readGrant(db, userId) { + const row = await db.whoopConnection.findUnique({ + where: { userId }, + select: { + whoopUserId: true, + accessToken: true, + refreshToken: true, + tokenExpiresAt: true, + }, + }); + if (!row?.whoopUserId) return null; + return { + accessCiphertext: row.accessToken, + refreshCiphertext: row.refreshToken, + expiresAt: row.tokenExpiresAt, + }; + }, + async writeGrant(db, userId, spent, next) { + const { count } = await db.whoopConnection.updateMany({ + where: { userId, refreshToken: spent }, + data: { + accessToken: next.accessCiphertext, + refreshToken: next.refreshCiphertext, + ...(next.expiresAt ? { tokenExpiresAt: next.expiresAt } : {}), + }, + }); + return count; + }, + loadCredentials: (userId) => getUserWhoopCredentials(userId), + refresh: (refreshToken, credentials) => + refreshAccessToken(refreshToken, credentials), +}; diff --git a/src/lib/whoop/sync-core.ts b/src/lib/whoop/sync-core.ts index 4ba22baeb..48f88adcb 100644 --- a/src/lib/whoop/sync-core.ts +++ b/src/lib/whoop/sync-core.ts @@ -3,12 +3,11 @@ * Measurement upsert/rollup-fold helpers, cursors, and sync status handling. * * Mirrors `src/lib/withings/sync.ts`: - * - `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). A - * provider/user advisory lock serializes refresh, and token state is - * re-read inside the transaction before the one-time token is spent. + * - `getValidToken` resolves the token through the shared OAuth token + * lifecycle (`src/lib/integrations/oauth-kit/token.ts`, descriptor in + * `./oauth-descriptor`): refreshed at `tokenExpiresAt - 5 min` under the + * provider/user advisory lock, BOTH rotated tokens persisted (WHOOP + * invalidates the prior access AND refresh token on every refresh). * - 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 @@ -27,7 +26,6 @@ import { AsyncLocalStorage } from "node:async_hooks"; import { prisma } from "@/lib/db"; import type { MeasurementType } from "@/generated/prisma/client"; -import { encrypt, decrypt } from "@/lib/crypto"; import { getEvent } from "@/lib/logging/context"; import { markSyncFailureRecorded, @@ -44,16 +42,12 @@ import { emitInsertedMeasurementArrivals, type InsertedMeasurementArrivalRow, } from "@/lib/arrivals/measurement-emit"; -import { - acquireProviderTokenRefreshLock, - PROVIDER_REFRESH_TRANSACTION_OPTIONS, -} from "@/lib/integrations/oauth-refresh"; +import { obtainAccessToken } from "@/lib/integrations/oauth-kit/token"; import { MeasurementReconciliationError, reconcileExternalMeasurement, } from "@/lib/measurements/reconcile-external-measurement"; -import { refreshAccessToken } from "./client"; -import { getUserWhoopCredentials } from "./credentials"; +import { whoopTokenDescriptor } from "./oauth-descriptor"; import { WhoopApiError, classifyWhoopError, @@ -164,9 +158,6 @@ export async function handleCollectionFetchError( throw markSyncFailureRecorded(err); } -/** Refresh the access token this many ms before `tokenExpiresAt`. */ -const TOKEN_REFRESH_BUFFER_MS = 5 * 60 * 1000; - /** * Overlap window for the incremental sync, in ms. WHOOP re-scores recovery * and sleep after the fact, and a night can reach the WHOOP cloud DAYS late @@ -259,94 +250,36 @@ export async function getValidToken( ): Promise { const connection = await prisma.whoopConnection.findUnique({ where: { userId }, + select: { id: true, whoopUserId: true }, }); if (!connection?.whoopUserId) return null; - - if ( - connection.tokenExpiresAt.getTime() - TOKEN_REFRESH_BUFFER_MS >= - Date.now() - ) { - return { - accessToken: decrypt(connection.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", - }); - // A dead token means every resource "succeeds" with 0 rows — without a - // ledger entry the cycle reads as clean and `recordSyncSuccess` un-parks - // `error_reauth` while the hourly cohort hammers the dead refresh token. - // Fail the cycle's verdict so the park holds. - noteHardFailure(WHOOP_TOKEN_HARD_FAIL); - return null; - } + const meta = { id: connection.id, whoopUserId: connection.whoopUserId }; try { - return await prisma.$transaction(async (tx) => { - await acquireProviderTokenRefreshLock(tx, "whoop", userId); - - const current = await tx.whoopConnection.findUnique({ - where: { userId }, - }); - if (!current?.whoopUserId) { - // The connection vanished on the re-read inside the advisory lock - // (disconnect/webhook race). Register on the ledger so the cycle does - // not stamp success over a connection that no longer exists — the same - // invariant as the refresh branch below. - noteHardFailure(WHOOP_TOKEN_HARD_FAIL); - 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 result = await obtainAccessToken(whoopTokenDescriptor, userId); + if (result.kind === "token") { + return { accessToken: result.accessToken, connection: meta }; + } + if (result.kind === "no_credentials") { + getEvent()?.addWarning( + `No WHOOP credentials found for user ${userId} during token refresh`, ); - 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, - }, + await recordSyncFailure({ + userId, + integration: "whoop", + kind: "reauth_required", + message: "WHOOP credentials missing — token refresh skipped", + errorCode: "credentials_missing", }); - - return { - accessToken: newTokens.access_token, - connection: { - id: current.id, - whoopUserId: current.whoopUserId, - }, - }; - }, PROVIDER_REFRESH_TRANSACTION_OPTIONS); + } + // No credentials, or the connection vanished between the read above and + // the token read (disconnect/webhook race). A dead token means every + // resource "succeeds" with 0 rows — without a ledger entry the cycle reads + // as clean and `recordSyncSuccess` un-parks `error_reauth` while the hourly + // cohort hammers the dead refresh token. Fail the cycle's verdict so the + // park holds. + noteHardFailure(WHOOP_TOKEN_HARD_FAIL); + return null; } catch (err) { getEvent()?.addWarning( `WHOOP token refresh failed for user ${userId}: ${err}`,