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.
This commit is contained in:
Marc-André Bombeck committed 2026-10-21 19:13:16 +02:00
1 parent 45da8ae971
commit 23f6fe4c6e
7 files changed
+570 -10

No files matched your search

+3 -6
View File
@@ -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)) {
@@ -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);
});
});
@@ -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<string>["refresh"];
creds?: string | null;
}) {
let stored = opts.grant;
const writes: NextGrant[] = [];
const reads: Array<StoredGrant | null> = [];
const descriptor: OAuthTokenDescriptor<string> = {
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> = {}): 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([]);
});
});
@@ -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<Response, "headers">,
now: number = Date.now(),
): number | undefined {
return parseRetryAfterMs(res.headers.get("retry-after"), now);
}
+246
View File
@@ -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<Credentials> {
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<StoredGrant | null>;
/**
* 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<number>;
/** The OAuth client the refresh authenticates as, or null when none is set. */
loadCredentials(userId: string): Promise<Credentials | null>;
/** Call the token endpoint. Null for a provider without refresh tokens. */
readonly refresh:
| ((
refreshToken: string,
credentials: Credentials,
) => Promise<RefreshedGrant>)
| 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<unknown>,
"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<Credentials>(
descriptor: OAuthTokenDescriptor<Credentials>,
userId: string,
opts: ObtainAccessTokenOptions = {},
): Promise<AccessTokenResult> {
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);
}
+8 -1
View File
@@ -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
+7 -3
View File
@@ -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);