From 23f6fe4c6eef913b871341988e9ca8a00970ff5e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Wed, 21 Oct 2026 19:13:16 +0200 Subject: [PATCH] feat(integrations): add one token lifecycle for every OAuth provider A provider describes where its grant lives and how its token endpoint is called; the kit reads the grant, refreshes it under the per-user advisory lock, re-reads under the lock so a peer's refresh or a reconnect wins, and writes back only over the refresh token it spent. Retry-After is read in one place, both forms, with a cap. --- src/lib/google-health/client.ts | 9 +- .../oauth-kit/__tests__/retry-after.test.ts | 41 +++ .../oauth-kit/__tests__/token.test.ts | 223 ++++++++++++++++ src/lib/integrations/oauth-kit/retry-after.ts | 42 +++ src/lib/integrations/oauth-kit/token.ts | 246 ++++++++++++++++++ src/lib/integrations/oauth-refresh.ts | 9 +- src/lib/strava/client.ts | 10 +- 7 files changed, 570 insertions(+), 10 deletions(-) create mode 100644 src/lib/integrations/oauth-kit/__tests__/retry-after.test.ts create mode 100644 src/lib/integrations/oauth-kit/__tests__/token.test.ts create mode 100644 src/lib/integrations/oauth-kit/retry-after.ts create mode 100644 src/lib/integrations/oauth-kit/token.ts diff --git a/src/lib/google-health/client.ts b/src/lib/google-health/client.ts index a0e3a7aa4..dbdd331b3 100644 --- a/src/lib/google-health/client.ts +++ b/src/lib/google-health/client.ts @@ -56,6 +56,7 @@ import { noteGoogleHealthWalkPage, } from "./backfill-bounds"; import { walkPages } from "@/lib/integrations/page-walk"; +import { retryAfterMsOf } from "@/lib/integrations/oauth-kit/retry-after"; export * from "./mappers"; @@ -1026,12 +1027,8 @@ export function googleHealthRetryDelayMs( json: unknown, now: number = Date.now(), ): number | undefined { - const header = res.headers.get("retry-after")?.trim(); - if (header) { - if (/^\d+$/.test(header)) return Number.parseInt(header, 10) * 1000; - const at = Date.parse(header); - if (Number.isFinite(at)) return Math.max(0, at - now); - } + const fromHeader = retryAfterMsOf(res, now); + if (fromHeader !== undefined) return fromHeader; const details = (json as { error?: { details?: unknown } } | null)?.error ?.details; if (Array.isArray(details)) { diff --git a/src/lib/integrations/oauth-kit/__tests__/retry-after.test.ts b/src/lib/integrations/oauth-kit/__tests__/retry-after.test.ts new file mode 100644 index 000000000..3466fa14b --- /dev/null +++ b/src/lib/integrations/oauth-kit/__tests__/retry-after.test.ts @@ -0,0 +1,41 @@ +import { describe, expect, it } from "vitest"; +import { + parseRetryAfterMs, + RETRY_AFTER_CAP_MS, + retryAfterMsOf, +} from "../retry-after"; + +const NOW = Date.parse("2026-10-21T12:00:00.000Z"); + +describe("parseRetryAfterMs", () => { + it("reads delay seconds", () => { + expect(parseRetryAfterMs("30", NOW)).toBe(30_000); + expect(parseRetryAfterMs(" 1.5 ", NOW)).toBe(1_500); + }); + + it("reads an HTTP date, a past one as zero", () => { + expect(parseRetryAfterMs("Wed, 21 Oct 2026 12:01:00 GMT", NOW)).toBe( + 60_000, + ); + expect(parseRetryAfterMs("Wed, 21 Oct 2026 11:00:00 GMT", NOW)).toBe(0); + }); + + it("caps an absurd wait", () => { + expect(parseRetryAfterMs("999999999", NOW)).toBe(RETRY_AFTER_CAP_MS); + }); + + it("ignores what it cannot read", () => { + expect(parseRetryAfterMs(null, NOW)).toBeUndefined(); + expect(parseRetryAfterMs("", NOW)).toBeUndefined(); + expect(parseRetryAfterMs("soon", NOW)).toBeUndefined(); + expect(parseRetryAfterMs("-5", NOW)).toBeUndefined(); + }); + + it("reads it off a response", () => { + const res = new Response(null, { + status: 429, + headers: { "Retry-After": "7" }, + }); + expect(retryAfterMsOf(res, NOW)).toBe(7_000); + }); +}); diff --git a/src/lib/integrations/oauth-kit/__tests__/token.test.ts b/src/lib/integrations/oauth-kit/__tests__/token.test.ts new file mode 100644 index 000000000..18eb3bd35 --- /dev/null +++ b/src/lib/integrations/oauth-kit/__tests__/token.test.ts @@ -0,0 +1,223 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const { lock } = vi.hoisted(() => ({ lock: vi.fn(async () => undefined) })); + +vi.mock("@/lib/db", () => ({ + prisma: { + $transaction: async (fn: (tx: unknown) => unknown) => fn({ tx: true }), + }, +})); +vi.mock("@/lib/crypto", () => ({ + encrypt: (v: string) => `enc(${v})`, + decrypt: (v: string) => v.replace(/^enc\((.*)\)$/, "$1"), +})); +vi.mock("@/lib/integrations/oauth-refresh", () => ({ + acquireProviderTokenRefreshLock: lock, + PROVIDER_REFRESH_TRANSACTION_OPTIONS: {}, +})); + +import { + obtainAccessToken, + type NextGrant, + type OAuthTokenDescriptor, + type StoredGrant, +} from "../token"; + +const NOW = Date.parse("2026-10-21T12:00:00.000Z"); + +function harness(opts: { + grant: StoredGrant | null; + tracksExpiry?: boolean; + refresh?: OAuthTokenDescriptor["refresh"]; + creds?: string | null; +}) { + let stored = opts.grant; + const writes: NextGrant[] = []; + const reads: Array = []; + const descriptor: OAuthTokenDescriptor = { + provider: "withings", + tracksExpiry: opts.tracksExpiry ?? true, + expiryBufferMs: 5 * 60 * 1000, + readGrant: async () => { + reads.push(stored); + return stored; + }, + writeGrant: async (_db, _user, spent, next) => { + if (!stored || stored.refreshCiphertext !== spent) return 0; + writes.push(next); + stored = next; + return 1; + }, + loadCredentials: async () => + opts.creds === undefined ? "creds" : opts.creds, + refresh: + opts.refresh === undefined + ? vi.fn(async () => ({ + access_token: "new-access", + refresh_token: "new-refresh", + expires_in: 3600, + })) + : opts.refresh, + }; + return { + descriptor, + writes, + reads, + replace(g: StoredGrant | null) { + stored = g; + }, + }; +} + +const grant = (over: Partial = {}): StoredGrant => ({ + accessCiphertext: "enc(old-access)", + refreshCiphertext: "enc(old-refresh)", + expiresAt: new Date(NOW + 60 * 60 * 1000), + ...over, +}); + +beforeEach(() => lock.mockClear()); + +describe("obtainAccessToken", () => { + it("answers none without a stored grant", async () => { + const h = harness({ grant: null }); + await expect( + obtainAccessToken(h.descriptor, "u", { now: () => NOW }), + ).resolves.toEqual({ kind: "none" }); + }); + + it("returns a fresh token without taking the lock", async () => { + const h = harness({ grant: grant() }); + const res = await obtainAccessToken(h.descriptor, "u", { now: () => NOW }); + expect(res).toMatchObject({ kind: "token", accessToken: "old-access" }); + expect(lock).not.toHaveBeenCalled(); + expect(h.descriptor.refresh).not.toHaveBeenCalled(); + }); + + it("refreshes a stale token under the lock and writes both tokens", async () => { + const h = harness({ grant: grant({ expiresAt: new Date(NOW + 1000) }) }); + const res = await obtainAccessToken(h.descriptor, "u", { now: () => NOW }); + expect(lock).toHaveBeenCalledWith({ tx: true }, "withings", "u"); + expect(res).toMatchObject({ + kind: "token", + accessToken: "new-access", + refreshed: true, + }); + expect(h.writes).toEqual([ + { + accessCiphertext: "enc(new-access)", + refreshCiphertext: "enc(new-refresh)", + expiresAt: new Date(NOW + 3600 * 1000), + }, + ]); + }); + + it("keeps the stored refresh token when the response carries none", async () => { + const h = harness({ + grant: grant({ expiresAt: new Date(NOW - 1) }), + refresh: vi.fn(async () => ({ access_token: "a2", expires_in: 60 })), + }); + await obtainAccessToken(h.descriptor, "u", { now: () => NOW }); + expect(h.writes[0]?.refreshCiphertext).toBe("enc(old-refresh)"); + }); + + it("uses a peer's refresh it finds under the lock instead of spending the token again", async () => { + const stale = grant({ expiresAt: new Date(NOW - 1) }); + const h = harness({ grant: stale }); + // The peer commits between the first read and the read under the lock. + let first = true; + const read = h.descriptor.readGrant; + h.descriptor.readGrant = async (db, user) => { + const g = await read(db, user); + if (first) { + first = false; + h.replace( + grant({ + accessCiphertext: "enc(peer-access)", + refreshCiphertext: "enc(peer-refresh)", + }), + ); + } + return g; + }; + const res = await obtainAccessToken(h.descriptor, "u", { now: () => NOW }); + expect(res).toMatchObject({ accessToken: "peer-access", refreshed: false }); + expect(h.descriptor.refresh).not.toHaveBeenCalled(); + }); + + it("after a refusal refreshes once, and a later refusal of the same token reuses the result", async () => { + const h = harness({ + grant: grant({ expiresAt: null }), + tracksExpiry: false, + }); + const a = await obtainAccessToken(h.descriptor, "u", { + afterRefusalOf: "enc(old-refresh)", + now: () => NOW, + }); + const b = await obtainAccessToken(h.descriptor, "u", { + afterRefusalOf: "enc(old-refresh)", + now: () => NOW, + }); + expect(a).toMatchObject({ accessToken: "new-access", refreshed: true }); + expect(b).toMatchObject({ accessToken: "new-access", refreshed: false }); + expect(h.descriptor.refresh).toHaveBeenCalledTimes(1); + expect(h.writes[0]?.expiresAt).toBeNull(); + }); + + it("lets a reconnect that lands during the refresh win", async () => { + const h = harness({ grant: grant({ expiresAt: new Date(NOW - 1) }) }); + h.descriptor = { + ...h.descriptor, + refresh: vi.fn(async () => { + h.replace( + grant({ + accessCiphertext: "enc(reconnected)", + refreshCiphertext: "enc(reconnected-refresh)", + }), + ); + return { access_token: "late", refresh_token: "late-r", expires_in: 1 }; + }), + }; + const res = await obtainAccessToken(h.descriptor, "u", { now: () => NOW }); + expect(res).toMatchObject({ accessToken: "reconnected", refreshed: false }); + expect(h.writes).toEqual([]); + }); + + it("answers no_credentials when the client is gone", async () => { + const h = harness({ + grant: grant({ expiresAt: new Date(NOW - 1) }), + creds: null, + }); + await expect( + obtainAccessToken(h.descriptor, "u", { now: () => NOW }), + ).resolves.toEqual({ kind: "no_credentials" }); + }); + + it("never refreshes for a provider without refresh tokens", async () => { + const h = harness({ + grant: grant({ refreshCiphertext: null, expiresAt: null }), + tracksExpiry: false, + refresh: null, + }); + const res = await obtainAccessToken(h.descriptor, "u", { + force: true, + now: () => NOW, + }); + expect(res).toMatchObject({ kind: "token", accessToken: "old-access" }); + expect(lock).not.toHaveBeenCalled(); + }); + + it("rethrows the provider's refusal untouched", async () => { + const boom = new Error("invalid_grant"); + const h = harness({ + grant: grant({ expiresAt: new Date(NOW - 1) }), + refresh: vi.fn(async () => { + throw boom; + }), + }); + await expect( + obtainAccessToken(h.descriptor, "u", { now: () => NOW }), + ).rejects.toBe(boom); + expect(h.writes).toEqual([]); + }); +}); diff --git a/src/lib/integrations/oauth-kit/retry-after.ts b/src/lib/integrations/oauth-kit/retry-after.ts new file mode 100644 index 000000000..cf6a42f58 --- /dev/null +++ b/src/lib/integrations/oauth-kit/retry-after.ts @@ -0,0 +1,42 @@ +/** + * How long a provider asked to be left alone after a 429 or 503. + * + * `Retry-After` comes in two forms (RFC 9110 10.2.3): a number of seconds, or + * an HTTP date. Every provider client reads it through here, so a date form is + * not honoured by one provider and ignored by the next. + */ + +/** Upper bound on any wait a provider can ask for, so a bad header cannot park a job for days. */ +export const RETRY_AFTER_CAP_MS = 24 * 60 * 60 * 1000; + +/** + * The delay a `Retry-After` value asks for, in ms, or undefined when the value + * is absent or unreadable. A date in the past reads as 0. + */ +export function parseRetryAfterMs( + value: string | null | undefined, + now: number = Date.now(), +): number | undefined { + const header = value?.trim(); + if (!header) return undefined; + if (/^\d+(?:\.\d+)?$/.test(header)) { + return Math.min( + RETRY_AFTER_CAP_MS, + Math.round(Number.parseFloat(header) * 1000), + ); + } + // An HTTP date always names its weekday and month; a bare number that is not + // a delay (a negative one) is not a date either, whatever Date.parse thinks. + if (!/[a-z]/i.test(header)) return undefined; + const at = Date.parse(header); + if (!Number.isFinite(at)) return undefined; + return Math.min(RETRY_AFTER_CAP_MS, Math.max(0, at - now)); +} + +/** The `Retry-After` delay of a response, in ms. */ +export function retryAfterMsOf( + res: Pick, + now: number = Date.now(), +): number | undefined { + return parseRetryAfterMs(res.headers.get("retry-after"), now); +} diff --git a/src/lib/integrations/oauth-kit/token.ts b/src/lib/integrations/oauth-kit/token.ts new file mode 100644 index 000000000..d526c4f9e --- /dev/null +++ b/src/lib/integrations/oauth-kit/token.ts @@ -0,0 +1,246 @@ +/** + * The one token lifecycle every OAuth provider integration goes through. + * + * Six providers (Withings, WHOOP, Oura, Strava, Google Health, Polar) used to + * carry their own copy of "read the stored grant, decide whether it needs a + * refresh, refresh it, write it back". The copies drifted: two refreshed under + * the advisory lock, one only compare-and-swapped, one did neither, and the + * Withings status route refreshed with no guard at all. A provider now + * describes where its grant lives and how its token endpoint is called + * (`OAuthTokenDescriptor`), and this module owns the rest: + * + * - A fresh token is returned straight from the row, without a transaction. + * - A refresh happens under the per-provider, per-user advisory lock + * (`acquireProviderTokenRefreshLock`). Under the lock the row is read again: + * a peer that refreshed while this caller waited, or a reconnect that + * stored a new grant, wins, and its token is used instead of refreshing a + * second time. Spending a one-time refresh token twice is what an + * authorization server reads as theft. + * - The write is conditional on the refresh ciphertext the refresh spent, so + * a reconnect that lands while the provider answers (it takes no lock) is + * never overwritten by the older grant. + * - A provider that does not rotate refresh tokens keeps the stored one when + * the response carries none. + * + * Storage is unchanged by design: every descriptor reads and writes the + * columns the provider always used, so no connection has to reconnect. + * + * What stays with the provider: classifying a refused refresh, recording it on + * the integration ledger, and the provider's own reauth flag. The kit throws + * the provider's own error untouched, so those branches keep working. + */ +import { prisma } from "@/lib/db"; +import type { Prisma } from "@/generated/prisma/client"; +import { decrypt, encrypt } from "@/lib/crypto"; +import { + acquireProviderTokenRefreshLock, + PROVIDER_REFRESH_TRANSACTION_OPTIONS, +} from "@/lib/integrations/oauth-refresh"; + +/** The OAuth integrations whose tokens the kit manages. */ +export type OAuthKitProvider = + "withings" | "whoop" | "oura" | "strava" | "google-health" | "polar"; + +/** Read and write access the descriptor gets: the client or a transaction. */ +export type OAuthKitDb = Prisma.TransactionClient; + +/** A grant as stored, still sealed. */ +export interface StoredGrant { + /** Ciphertext of the access token. */ + readonly accessCiphertext: string; + /** Ciphertext of the refresh token; null for a provider without one. */ + readonly refreshCiphertext: string | null; + /** When the access token stops working; null when the provider gives none. */ + readonly expiresAt: Date | null; +} + +/** The sealed grant a refresh writes back. */ +export interface NextGrant { + readonly accessCiphertext: string; + readonly refreshCiphertext: string; + readonly expiresAt: Date | null; +} + +/** What a provider's token endpoint answered on a refresh. */ +export interface RefreshedGrant { + readonly access_token: string; + readonly refresh_token?: string | null; + readonly expires_in?: number | null; +} + +export interface OAuthTokenDescriptor { + readonly provider: OAuthKitProvider; + /** + * True when the provider stores an expiry and the token is refreshed ahead + * of it. False for the providers that refresh only after a refusal (Oura, + * Strava) or never (Polar). + */ + readonly tracksExpiry: boolean; + /** How long before the expiry a token counts as stale. */ + readonly expiryBufferMs: number; + /** Read the stored grant, or null when the user has no connection. */ + readGrant(db: OAuthKitDb, userId: string): Promise; + /** + * Write a refreshed grant, only while the stored refresh ciphertext still + * equals `spentRefreshCiphertext`. Returns the number of rows written. + */ + writeGrant( + db: OAuthKitDb, + userId: string, + spentRefreshCiphertext: string, + next: NextGrant, + ): Promise; + /** The OAuth client the refresh authenticates as, or null when none is set. */ + loadCredentials(userId: string): Promise; + /** Call the token endpoint. Null for a provider without refresh tokens. */ + readonly refresh: + | (( + refreshToken: string, + credentials: Credentials, + ) => Promise) + | null; +} + +export type AccessTokenResult = + | { readonly kind: "none" } + | { readonly kind: "no_credentials" } + | { + readonly kind: "token"; + readonly accessToken: string; + readonly expiresAt: Date | null; + /** The refresh ciphertext stored beside the token, for a later refusal. */ + readonly refreshCiphertext: string | null; + /** True when this call refreshed the token itself. */ + readonly refreshed: boolean; + }; + +export interface ObtainAccessTokenOptions { + /** + * The token was just refused. Refresh unless a peer already did: pass the + * refresh ciphertext the refused token was read with. A stored ciphertext + * that differs from it means the grant moved on since, and its token is + * used as it is. + */ + readonly afterRefusalOf?: string; + /** Refresh even when the stored expiry says the token is still good. */ + readonly force?: boolean; + /** Clock override for tests. */ + readonly now?: () => number; +} + +function isFresh( + descriptor: Pick< + OAuthTokenDescriptor, + "tracksExpiry" | "expiryBufferMs" + >, + grant: StoredGrant, + now: number, +): boolean { + if (!descriptor.tracksExpiry || grant.expiresAt === null) return true; + return grant.expiresAt.getTime() - descriptor.expiryBufferMs >= now; +} + +function tokenOf(grant: StoredGrant, refreshed: boolean): AccessTokenResult { + return { + kind: "token", + accessToken: decrypt(grant.accessCiphertext), + expiresAt: grant.expiresAt, + refreshCiphertext: grant.refreshCiphertext, + refreshed, + }; +} + +/** + * A usable access token for the user's connection, refreshing it when it is + * stale, forced, or was just refused. Throws whatever the provider's token + * endpoint threw, for the provider to classify and record. + */ +export async function obtainAccessToken( + descriptor: OAuthTokenDescriptor, + userId: string, + opts: ObtainAccessTokenOptions = {}, +): Promise { + const now = opts.now ?? Date.now; + const observed = await descriptor.readGrant(prisma, userId); + if (!observed) return { kind: "none" }; + + const wantsRefresh = + opts.force === true || + opts.afterRefusalOf !== undefined || + !isFresh(descriptor, observed, now()); + if ( + !wantsRefresh || + descriptor.refresh === null || + observed.refreshCiphertext === null + ) { + return tokenOf(observed, false); + } + + const credentials = await descriptor.loadCredentials(userId); + if (!credentials) return { kind: "no_credentials" }; + + const refresh = descriptor.refresh; + // What the caller's view of the grant was. A refusal names the ciphertext + // its token was read with; otherwise it is what was just read above. + const spentView = opts.afterRefusalOf ?? observed.refreshCiphertext; + + return prisma.$transaction(async (tx) => { + await acquireProviderTokenRefreshLock(tx, descriptor.provider, userId); + const current = await descriptor.readGrant(tx, userId); + if (!current || current.refreshCiphertext === null) { + return { kind: "none" } as const; + } + // The grant moved on while this caller waited for the lock or used its + // token: a peer refreshed it, or a reconnect replaced it. Use that one. + // A non-rotating provider keeps its refresh ciphertext, so its access + // ciphertext tells the same story. + const movedOn = + current.refreshCiphertext !== spentView || + (opts.afterRefusalOf === undefined && + current.accessCiphertext !== observed.accessCiphertext); + if (movedOn && isFresh(descriptor, current, now())) { + return tokenOf(current, false); + } + if ( + !movedOn && + opts.force !== true && + opts.afterRefusalOf === undefined && + isFresh(descriptor, current, now()) + ) { + return tokenOf(current, false); + } + + const rotated = await refresh( + decrypt(current.refreshCiphertext), + credentials, + ); + const expiresAt = descriptor.tracksExpiry + ? new Date(now() + (rotated.expires_in ?? 3600) * 1000) + : null; + const next: NextGrant = { + accessCiphertext: encrypt(rotated.access_token), + refreshCiphertext: rotated.refresh_token + ? encrypt(rotated.refresh_token) + : current.refreshCiphertext, + expiresAt, + }; + const written = await descriptor.writeGrant( + tx, + userId, + current.refreshCiphertext, + next, + ); + if (written > 0) { + return { + kind: "token", + accessToken: rotated.access_token, + expiresAt, + refreshCiphertext: next.refreshCiphertext, + refreshed: true, + } as const; + } + // A reconnect stored a new grant while the provider answered; it wins. + const replaced = await descriptor.readGrant(tx, userId); + return replaced ? tokenOf(replaced, false) : ({ kind: "none" } as const); + }, PROVIDER_REFRESH_TRANSACTION_OPTIONS); +} diff --git a/src/lib/integrations/oauth-refresh.ts b/src/lib/integrations/oauth-refresh.ts index 452caec8f..8acef990a 100644 --- a/src/lib/integrations/oauth-refresh.ts +++ b/src/lib/integrations/oauth-refresh.ts @@ -62,7 +62,14 @@ export async function persistRotatedToken( } export type SerializedRefreshProvider = - "whoop" | "withings" | "oura" | "codex" | "admin-codex"; + | "whoop" + | "withings" + | "oura" + | "strava" + | "google-health" + | "polar" + | "codex" + | "admin-codex"; /** * Provider refreshes hold a database transaction open around one bounded diff --git a/src/lib/strava/client.ts b/src/lib/strava/client.ts index 3bc4d1f5d..8ac54fd1a 100644 --- a/src/lib/strava/client.ts +++ b/src/lib/strava/client.ts @@ -23,6 +23,7 @@ */ import { getEvent } from "@/lib/logging/context"; import { safeFetch } from "@/lib/safe-fetch"; +import { parseRetryAfterMs } from "@/lib/integrations/oauth-kit/retry-after"; import type { Prisma } from "@/generated/prisma/client"; import { StravaApiError, classifyStravaResponse } from "./response-classifier"; import { mapStravaSportType } from "./sport-map"; @@ -282,9 +283,12 @@ export function stravaRateLimitRetryAfterSeconds( headers: Headers, now: Date = new Date(), ): number { - const retryAfter = Number(headers.get("retry-after")); - if (Number.isFinite(retryAfter) && retryAfter > 0) { - return Math.ceil(retryAfter); + const retryAfterMs = parseRetryAfterMs( + headers.get("retry-after"), + now.getTime(), + ); + if (retryAfterMs !== undefined && retryAfterMs > 0) { + return Math.ceil(retryAfterMs / 1000); } const pair = (name: string): [number, number] | null => { const parts = (headers.get(name) ?? "").split(",").map(Number);