From be457962878ff14ba2c515c2b9d1abc3e73993aa Mon Sep 17 00:00:00 2001 From: Josh Date: Thu, 3 Sep 2026 13:48:23 +0100 Subject: [PATCH 1/7] feat: accept an OAuth token setup as `token` for transparent refresh --- .changeset/pr-1323.md | 6 + src/assets/AssetsClient.ts | 15 +- src/config.ts | 5 +- src/data/dataMethods.ts | 34 +++- src/data/listen.ts | 7 +- src/data/live.ts | 30 +++- src/data/reconnectOnConnectionFailure.ts | 39 ++++- src/data/resolveEventSourceFetch.ts | 34 +++- src/http/oauthRefreshHandler.ts | 180 +++++++++++++++++++++ src/http/requestOptions.ts | 5 +- src/types.ts | 22 ++- test/oauthRefreshHandler.node.test.ts | 48 ++++++ test/oauthRefreshHandler.test.ts | 195 +++++++++++++++++++++++ test/resumability.test.ts | 100 ++++++++++++ 14 files changed, 697 insertions(+), 23 deletions(-) create mode 100644 .changeset/pr-1323.md create mode 100644 src/http/oauthRefreshHandler.ts create mode 100644 test/oauthRefreshHandler.node.test.ts create mode 100644 test/oauthRefreshHandler.test.ts create mode 100644 test/resumability.test.ts diff --git a/.changeset/pr-1323.md b/.changeset/pr-1323.md new file mode 100644 index 000000000..a9998b0e3 --- /dev/null +++ b/.changeset/pr-1323.md @@ -0,0 +1,6 @@ + +--- +'@sanity/client': minor +--- + +feat: accept an OAuth token setup as `token` for transparent refresh \ No newline at end of file diff --git a/src/assets/AssetsClient.ts b/src/assets/AssetsClient.ts index 6a7828ff8..eefafefee 100644 --- a/src/assets/AssetsClient.ts +++ b/src/assets/AssetsClient.ts @@ -1,7 +1,8 @@ import {defer, lastValueFrom, type Observable} from 'rxjs' -import {filter, map, mergeAll} from 'rxjs/operators' +import {catchError, filter, map, mergeAll} from 'rxjs/operators' import {_prepareRequest, _uploadObservable} from '../data/dataMethods' +import {applyOAuthToken, getOAuthTokenSetup, refreshOnAuthError} from '../http/oauthRefreshHandler' import type {FetchRequest} from '../http/requestOptions' import type {ObservableSanityClient, SanityClient} from '../SanityClient' import type { @@ -262,10 +263,15 @@ function _upload< // credentials and timeout are identical across both upload transports. // The XHR API needs the query baked into the URL, though. const req = _prepareRequest(client, {...baseRequest}) - return uploadWithProgress({ + // XHR uploads bypass the request handler, so the OAuth token is resolved + // (and a 401 refreshed, without auto-retry) here — same rules as the + // fetch-path `_uploadObservable`. + const oauth = getOAuthTokenSetup(config.token) + const reqHeaders = oauth ? await applyOAuthToken(oauth, req.headers) : req.headers + const upload = uploadWithProgress({ url: appendQuery(req.url, req.query), method: req.method ?? 'POST', - headers: req.headers, + headers: reqHeaders, body, withCredentials: req.credentials === 'include', // XHR only has a single total-deadline timer, so a structured @@ -273,6 +279,9 @@ function _upload< timeout: typeof req.timeout === 'object' ? req.timeout.total : req.timeout, signal: req.signal, }) + return oauth + ? upload.pipe(catchError((err) => refreshOnAuthError(oauth, reqHeaders, err))) + : upload }).pipe(mergeAll()) } diff --git a/src/config.ts b/src/config.ts index 926d6dd46..a458f1aee 100644 --- a/src/config.ts +++ b/src/config.ts @@ -126,7 +126,10 @@ export const initConfig = ( newConfig.withCredentials = false } - if (isBrowser && isLocalhost && hasToken && newConfig.ignoreBrowserTokenWarning !== true) { + // The browser warning targets a secret baked into client-side code. An + // OAuthTokenSetup obtains tokens per user at runtime, so it is exempt. + const hasStringToken = typeof newConfig.token === 'string' && newConfig.token !== '' + if (isBrowser && isLocalhost && hasStringToken && newConfig.ignoreBrowserTokenWarning !== true) { warnings.printBrowserTokenWarning() } else if (typeof newConfig.useCdn === 'undefined') { warnings.printCdnWarning() diff --git a/src/data/dataMethods.ts b/src/data/dataMethods.ts index c6771c5c7..fab02b190 100644 --- a/src/data/dataMethods.ts +++ b/src/data/dataMethods.ts @@ -1,9 +1,15 @@ import {getDraftId, getVersionFromId, getVersionId, isDraftId} from '@sanity/client/csm' import {anySignal} from 'get-it/any-signal' -import {type MonoTypeOperatorFunction, Observable} from 'rxjs' -import {filter, map} from 'rxjs/operators' +import {defer, type MonoTypeOperatorFunction, Observable} from 'rxjs' +import {catchError, filter, map, mergeMap} from 'rxjs/operators' import {validateApiPerspective} from '../config' +import { + applyOAuthToken, + getOAuthTokenSetup, + refreshOnAuthError, + resolveRequestHandler, +} from '../http/oauthRefreshHandler' import {type FetchRequest, requestOptions} from '../http/requestOptions' import type {ObservableSanityClient, SanityClient} from '../SanityClient' import {stegaClean, type StegaCleaned} from '../stega/stegaClean' @@ -1157,7 +1163,7 @@ export function _observe( */ export function _request(client: Client, httpRequest: HttpRequest, options: Any): Promise { const reqOptions = _prepareRequest(client, options) - return httpRequest(reqOptions, client.config().requestHandler).then((body) => body as R) + return httpRequest(reqOptions, resolveRequestHandler(client.config())).then((body) => body as R) } /** @@ -1191,9 +1197,25 @@ export function _uploadObservable( options: RequestObservableOptions, ): Observable> { const reqOptions = _prepareRequest(client, options) - const requester = client.config().requester - const request = new Observable((subscriber) => - requester(reqOptions).subscribe(subscriber), + const config = client.config() + const requester = config.requester + // This path bypasses the request handler, so the OAuth token is resolved + // here — per subscription, so a resubscribe picks up a refreshed token. A + // 401 refreshes but does not auto-retry (the body may be a consumed stream); + // the error surfaces and a caller-level retry gets the fresh token. + const oauth = getOAuthTokenSetup(config.token) + const upload = (req: FetchRequest) => + new Observable((subscriber) => requester(req).subscribe(subscriber)) + const request = ( + oauth + ? defer(() => applyOAuthToken(oauth, reqOptions.headers)).pipe( + mergeMap((headers) => + upload({...reqOptions, headers}).pipe( + catchError((err) => refreshOnAuthError(oauth, headers, err)), + ), + ), + ) + : upload(reqOptions) ).pipe( filter((event: Any) => event?.type === 'progress' || event?.type === 'response'), map((event: Any): UploadEvent => diff --git a/src/data/listen.ts b/src/data/listen.ts index 90b1bd1e2..86a15f1bd 100644 --- a/src/data/listen.ts +++ b/src/data/listen.ts @@ -2,6 +2,7 @@ import {EventSource} from 'eventsource' import {Observable, throwError} from 'rxjs' import {filter, map} from 'rxjs/operators' +import {getOAuthRefresher, getOAuthTokenSetup} from '../http/oauthRefreshHandler' import type {ObservableSanityClient, SanityClient} from '../SanityClient' import { type Any, @@ -166,23 +167,25 @@ export function _connectListenEventSource( const {token, withCredentials, headers: configHeaders} = config const headers: Record = {} - if (token) { + if (typeof token === 'string') { headers.Authorization = `Bearer ${token}` } if (configHeaders) { Object.assign(headers, configHeaders) } + const tokenSetup = getOAuthTokenSetup(token) const initEventSource = () => new EventSource(uri, { fetch: resolveEventSourceFetch(config, { headers: Object.keys(headers).length ? headers : undefined, + tokenSetup, withCredentials, }), }) return connectEventSource(initEventSource, listenFor).pipe( - reconnectOnConnectionFailure(), + reconnectOnConnectionFailure(tokenSetup && getOAuthRefresher(tokenSetup)), filter((event) => listenFor.includes(event.type)), map((event) => ({ type: event.type, diff --git a/src/data/live.ts b/src/data/live.ts index b8df72ccf..279c81f69 100644 --- a/src/data/live.ts +++ b/src/data/live.ts @@ -4,6 +4,7 @@ import {catchError, mergeMap, Observable, of, throwError} from 'rxjs' import {finalize, map} from 'rxjs/operators' import {CorsOriginError} from '../http/errors' +import {getOAuthRefresher, getOAuthTokenSetup} from '../http/oauthRefreshHandler' import type {ObservableSanityClient, SanityClient} from '../SanityClient' import type { InitializedClientConfig, @@ -13,6 +14,7 @@ import type { LiveEventReconnect, LiveEventRestart, LiveEventWelcome, + OAuthTokenSetup, SyncTag, } from '../types' import {isRecord} from '../util/isRecord' @@ -90,12 +92,16 @@ export class LiveClient { url.searchParams.set('waitFor', waitFor) } const eventSourceHeaders: Record = {} - if (includeDrafts && token) { + if (includeDrafts && typeof token === 'string') { eventSourceHeaders.Authorization = `Bearer ${token}` } if (configHeaders) { Object.assign(eventSourceHeaders, configHeaders) } + // An OAuth token setup can't be baked into the headers — it is resolved + // per request inside the EventSource fetch, so reconnects pick up + // refreshed tokens. + const tokenSetup = includeDrafts ? getOAuthTokenSetup(token) : undefined const eventSourceWithCredentials = Boolean(includeDrafts && withCredentials) let transportCache = eventsCache.get(config.resolveFetch) @@ -108,6 +114,10 @@ export class LiveClient { typeof config.proxy === 'string' ? config.proxy : null, eventSourceHeaders, eventSourceWithCredentials, + // A string-token connection is distinguished by its Authorization header + // above; an OAuth setup contributes its identity instead, so two clients + // with different setups never share a stream. + tokenSetup ? tokenSetupCacheKey(tokenSetup) : null, ]) const existing = transportCache.get(cacheKey) @@ -119,6 +129,7 @@ export class LiveClient { new EventSource(url.href, { fetch: resolveEventSourceFetch(config, { headers: Object.keys(eventSourceHeaders).length ? eventSourceHeaders : undefined, + tokenSetup, withCredentials: eventSourceWithCredentials, }), }) @@ -140,7 +151,7 @@ export class LiveClient { const observable = events .pipe( - reconnectOnConnectionFailure(), + reconnectOnConnectionFailure(tokenSetup && getOAuthRefresher(tokenSetup)), mergeMap((event) => { if (event.type === 'reconnect') { // Check for CORS on reconnect events (which happen on 403s) @@ -289,3 +300,18 @@ const eventsCache = new Map< InitializedClientConfig['resolveFetch'], Map> >() + +/** + * Stable identity for an OAuth token setup in the string cache key — the setup + * object itself can't be stringified (its functions serialise to nothing). + */ +let nextTokenSetupId = 0 +const tokenSetupIds = new WeakMap() +function tokenSetupCacheKey(setup: OAuthTokenSetup): number { + let id = tokenSetupIds.get(setup) + if (id === undefined) { + id = nextTokenSetupId++ + tokenSetupIds.set(setup, id) + } + return id +} diff --git a/src/data/reconnectOnConnectionFailure.ts b/src/data/reconnectOnConnectionFailure.ts index e0aef00a9..764b290d9 100644 --- a/src/data/reconnectOnConnectionFailure.ts +++ b/src/data/reconnectOnConnectionFailure.ts @@ -1,6 +1,7 @@ import { catchError, concat, + from, mergeMap, Observable, of, @@ -13,13 +14,30 @@ import {ConnectionFailedError} from './eventsource' const RETRYABLE_STATUSES = new Set([408, 429]) +/** + * Minimum spacing between 401-triggered auth refreshes. A refresh → reconnect + * → 401 cycle completes within seconds, so a second 401 inside this window + * means the server is rejecting freshly refreshed tokens — surface it rather + * than rotate refresh tokens forever. A 401 after the window (a token that + * expired hours into a healthy connection) gets its own refresh. + */ +const AUTH_RETRY_WINDOW = 30_000 + /** * Note: connection failure is not the same as network disconnect which may happen more frequent. * The EventSource instance will automatically reconnect in case of a network disconnect, however, * in some rare cases a ConnectionFailed Error will be thrown and this operator explicitly retries these + * + * @param refreshAuth - When the connection authenticates via an OAuth token + * setup, the setup's single-flight refresher. A connection rejected with a 401 + * then refreshes and reconnects once, mirroring the request handler's 401 + * semantics. */ -export function reconnectOnConnectionFailure(): OperatorFunction { +export function reconnectOnConnectionFailure( + refreshAuth?: () => Promise, +): OperatorFunction { return function (source: Observable) { + let lastAuthRetryAt = -Infinity return source.pipe( catchError((err, caught) => { // Only reconnect on transient connection failures. A 4xx response is a @@ -37,6 +55,25 @@ export function reconnectOnConnectionFailure(): OperatorFunction caught))) } + if ( + refreshAuth && + err instanceof ConnectionFailedError && + err.status === 401 && + Date.now() - lastAuthRetryAt >= AUTH_RETRY_WINDOW + ) { + lastAuthRetryAt = Date.now() + return concat( + of({type: 'reconnect' as const}), + from(refreshAuth()).pipe( + // Surface the original connection error, not the refresh failure + // — `onAuthError` has already fired inside the refresher. Placed + // before `mergeMap` so it only catches the refresh promise, never + // errors from the resubscribed stream. + catchError(() => throwError(() => err)), + mergeMap(() => caught), + ), + ) + } return throwError(() => err) }), ) diff --git a/src/data/resolveEventSourceFetch.ts b/src/data/resolveEventSourceFetch.ts index dd9a918cb..48be81a94 100644 --- a/src/data/resolveEventSourceFetch.ts +++ b/src/data/resolveEventSourceFetch.ts @@ -1,7 +1,7 @@ import type {EventSourceFetchInit, FetchLikeResponse} from 'eventsource' import type {FetchFunction, FetchInit} from 'get-it' -import type {InitializedClientConfig} from '../types' +import type {InitializedClientConfig, OAuthTokenSetup} from '../types' /** @internal */ export interface EventSourceFetchOptions { @@ -11,6 +11,15 @@ export interface EventSourceFetchOptions { * etc. — things the native EventSource API has no equivalent for. */ headers?: Record + /** + * OAuth token setup to resolve an `Authorization` header from. Resolved via + * `getToken()` on every request — not once per connection — so the + * `eventsource` package's reconnects pick up a refreshed token. This fetch + * only reads; 401-driven `refresh()` lives upstream in + * `reconnectOnConnectionFailure`. Config `headers` take precedence, + * mirroring the string-token merge order. + */ + tokenSetup?: OAuthTokenSetup /** * If the client was configured with `withCredentials: true`, the * resolved fetch forwards `credentials: 'include'` so the browser @@ -50,19 +59,25 @@ export function resolveEventSourceFetch( options: EventSourceFetchOptions = {}, ): EventSourceFetch { const extraHeaders = options.headers + const tokenSetup = options.tokenSetup const credentials: FetchInit['credentials'] = options.withCredentials ? 'include' : undefined - return function eventSourceFetch(url, init) { + return async function eventSourceFetch(url, init) { const baseFetch = pickBaseFetch(config) // Extra `EventSourceFetchInit` fields get-it's `FetchInit` doesn't // declare (`mode`, `cache`) survive the spread and reach whichever // fetch implementation is effective. const mergedInit: FetchInit = {...init} - if (extraHeaders) { + if (extraHeaders || tokenSetup) { const headers = new Headers(init?.headers) - for (const [key, value] of Object.entries(extraHeaders)) { - headers.set(key, value) + if (tokenSetup) { + headers.set('Authorization', `Bearer ${await tokenSetup.getToken()}`) + } + if (extraHeaders) { + for (const [key, value] of Object.entries(extraHeaders)) { + headers.set(key, value) + } } mergedInit.headers = headers } @@ -71,7 +86,14 @@ export function resolveEventSourceFetch( } // get-it's `FetchResponse` is a structural superset of the package's // `FetchLikeResponse`, so it can be handed over as-is. - return baseFetch(typeof url === 'string' ? url : url.href, mergedInit) + const response = baseFetch(typeof url === 'string' ? url : url.href, mergedInit) + // Returning a promise from an async function attaches its rejection + // handler one microtask later (thenable adoption), and workerd's + // unhandled-rejection tracker flags a rejected promise in that gap. + // Attach a no-op handler synchronously; the rejection still propagates + // through the async return to the `eventsource` package's catch. + response.catch(() => {}) + return response } } diff --git a/src/http/oauthRefreshHandler.ts b/src/http/oauthRefreshHandler.ts new file mode 100644 index 000000000..73fbec848 --- /dev/null +++ b/src/http/oauthRefreshHandler.ts @@ -0,0 +1,180 @@ +import type { + InitializedClientConfig, + OAuthTokenSetup, + RequestHandler, + RequestHandlerOptions, +} from '../types' +import {ClientError} from './errors' + +/** + * The single spot deciding "is this token an OAuth setup" — shared by request + * handler resolution, SSE and asset uploads so the paths can never disagree on + * what counts as one. + * + * @internal + */ +export function getOAuthTokenSetup( + token: InitializedClientConfig['token'], +): OAuthTokenSetup | undefined { + return token && typeof token === 'object' ? token : undefined +} + +/** + * Resolve `getToken()` into an `Authorization` header for request paths that + * bypass the request handler (asset uploads). A pre-existing `Authorization` + * header (config `headers`) wins, matching the refresh handler's pass-through. + * + * @internal + */ +export async function applyOAuthToken( + setup: OAuthTokenSetup, + headers: Record, +): Promise> { + // `new Headers()` gives a case-insensitive lookup over the plain record. + if (new Headers(headers).has('authorization')) return headers + return {...headers, Authorization: `Bearer ${await setup.getToken()}`} +} + +/** + * Per-setup memo of single-flight refreshers, keyed on the setup object itself + * so every path that can refresh (the request handler, SSE reconnects) and + * every client holding the same setup (clones, `withConfig()` children, the + * observable twin) share one in-flight refresh. That is a correctness + * requirement, not an optimisation: OAuth 2.1 refresh tokens are single-use, + * so two concurrent refreshes would present the same consumed token and trip + * reuse detection, which can revoke the whole token family. + */ +const refreshers = new WeakMap Promise>() + +/** + * The setup's single-flight `refresh()`: concurrent callers share one attempt, + * and `onAuthError` fires once per attempt, not once per waiting caller. Only + * dedupes within this process — cross-tab serialisation is the provider's job. + * + * @internal + */ +export function getOAuthRefresher(setup: OAuthTokenSetup): () => Promise { + let refresher = refreshers.get(setup) + if (!refresher) { + // `??=` is synchronous, so two refreshes can never start at once. + let inFlight: Promise | null = null + refresher = () => + (inFlight ??= setup + .refresh() + .catch((error) => { + setup.onAuthError?.(error) + throw error + }) + .finally(() => { + inFlight = null + })) + refreshers.set(setup, refresher) + } + return refresher +} + +/** + * 401 handling for paths that can't safely retry — uploads, where a + * `NodeJS.ReadableStream` body is consumed by the first attempt. Refreshes so + * `onAuthError` can fire on an unrecoverable refresh and a caller-level retry + * gets a fresh token, then rethrows the original error: the caller decides + * whether its body is replayable, not the client. + * + * Mirrors the request handler's re-read guard: no refresh when the token has + * already moved on since the request was built, or when the `Authorization` + * header wasn't ours to begin with (explicit-header pass-through). + * + * @internal + */ +export async function refreshOnAuthError( + setup: OAuthTokenSetup, + sentHeaders: Record, + error: unknown, +): Promise { + if (error instanceof ClientError && error.statusCode === 401) { + try { + if (sentHeaders.Authorization === `Bearer ${await setup.getToken()}`) { + await getOAuthRefresher(setup)() + } + } catch { + // onAuthError already fired inside the refresher; surface the 401 below. + } + } + throw error +} + +/** + * Resolve the effective request handler for a request. The OAuth refresh + * handler wraps any user-supplied `requestHandler` (rather than being stored + * in `config.requestHandler`) so a later `withConfig({requestHandler})` swap + * can't silently drop auth. + * + * Resolved from the live config on every request, so reconfiguring the token + * via `client.config({token})` / `withConfig({token})` swaps refresh behaviour + * in or out like any other config change. + * + * @internal + */ +export function resolveRequestHandler( + config: InitializedClientConfig, +): RequestHandler | undefined { + const setup = getOAuthTokenSetup(config.token) + if (!setup) return config.requestHandler + const oauthHandler = createOAuthRefreshHandler(setup) + const userHandler = config.requestHandler + if (!userHandler) return oauthHandler + return (request, next) => oauthHandler(request, (r) => userHandler(r, next)) +} + +function withToken(request: RequestHandlerOptions, token: string): RequestHandlerOptions { + // `new Headers()` normalises every `FetchHeaders` shape, so no assertion. + const headers = new Headers(request.headers) + headers.set('Authorization', `Bearer ${token}`) + return {...request, headers} +} + +/** + * Build the request handler that keeps a client authenticated from an + * {@link OAuthTokenSetup}: it applies the current token to every request and, + * on a 401, refreshes and retries once. + * + * On a 401 it re-reads the token before refreshing: if another request — or + * another tab, via the provider's storage — already refreshed, it retries with + * the current token instead of rotating a still-valid refresh token again. + */ +function createOAuthRefreshHandler(setup: OAuthTokenSetup): RequestHandler { + const refreshOnce = getOAuthRefresher(setup) + + return async function oauthRefreshHandler(request, next) { + // An explicit `Authorization` header (a per-request `token` override or a + // config `headers` entry) wins: pass through, with no refresh semantics — + // a 401 against a token this handler didn't supply isn't its to fix. + if (new Headers(request.headers).has('authorization')) return next(request) + + const token = await setup.getToken() + try { + return await next(withToken(request, token)) + } catch (error) { + if (!(error instanceof ClientError) || error.statusCode !== 401) throw error + + // Refresh only if the current token is still the one that just failed; + // otherwise someone already refreshed and we retry with what's current. + let nextToken = await setup.getToken() + if (nextToken === token) { + try { + nextToken = await refreshOnce() + } catch { + // onAuthError already fired inside refreshOnce. Surface the 401, the + // error the caller's request produced, not the refresh failure. + throw error + } + } + + // A logged-out provider yields no token; don't retry with an empty bearer. + if (!nextToken) throw error + + // Retry once. A second 401 with a fresh token is a real authz failure. + return next(withToken(request, nextToken)) + } + } +} diff --git a/src/http/requestOptions.ts b/src/http/requestOptions.ts index cc47ac381..d071dc4d0 100644 --- a/src/http/requestOptions.ts +++ b/src/http/requestOptions.ts @@ -34,8 +34,11 @@ export function requestOptions(config: Any, overrides: Any = {}): FetchRequest { Object.assign(headers, config.headers) } + // A string token becomes a Bearer header here; an OAuthTokenSetup object is + // resolved by the OAuth refresh handler (see `resolveRequestHandler`), which + // sets the header itself — never stringify the object into one. const token = overrides.token || config.token - if (token) { + if (typeof token === 'string') { headers['Authorization'] = `Bearer ${token}` } diff --git a/src/types.ts b/src/types.ts index 36d55b523..addeefbc5 100644 --- a/src/types.ts +++ b/src/types.ts @@ -70,6 +70,26 @@ export type RequestHandler = ( next: (request: RequestHandlerOptions) => Promise, ) => Promise +/** + * An OAuth token setup, accepted as `token` in place of a static string so the + * client can refresh a short-lived access token transparently. + * @public + */ +export interface OAuthTokenSetup { + /** + * Resolve the access token for the next request. A provider may refresh + * proactively here on expiry skew, so the common path avoids a 401. + */ + getToken: () => Promise + /** + * Force a refresh after a 401 and resolve to the new access token. Once + * resolved, subsequent `getToken()` calls must return the refreshed token. + */ + refresh: () => Promise + /** Called when `refresh()` rejects */ + onAuthError?: (error: unknown) => void +} + /** * @public * @deprecated – The `r`-prefix is not required, use `string` instead @@ -132,7 +152,7 @@ export interface ClientConfig { dataset?: string /** @defaultValue true */ useCdn?: boolean - token?: string + token?: string | OAuthTokenSetup /** * Configure the client to work with a specific Sanity resource (Media Library, Canvas, etc.) diff --git a/test/oauthRefreshHandler.node.test.ts b/test/oauthRefreshHandler.node.test.ts new file mode 100644 index 000000000..641a3bff3 --- /dev/null +++ b/test/oauthRefreshHandler.node.test.ts @@ -0,0 +1,48 @@ +import {ClientError, type OAuthTokenSetup} from '@sanity/client' +import {describe, expect, test, vi} from 'vitest' + +import {getClient, projectHost} from './client/helpers' +import {getActiveMock} from './helpers/mockFetch' + +// Where `XMLHttpRequest` is a global (browsers, happy-dom) `assets.upload()` +// takes the XHR path (`src/http/browserUpload.ts`), which bypasses the client's +// `resolveFetch` seam and therefore the get-it fetch mock — so these mock-based +// assertions can only run where XHR is absent and uploads use the fetch path. +// The XHR path has its own coverage in `browserUpload.browser.test.ts`. + +const oauthClient = (setup: OAuthTokenSetup) => getClient({token: setup}) + +function authHeaders(): Array { + return getActiveMock() + .getRequests() + .map((request) => request.headers.get('authorization')) +} + +describe('OAuth auto-refresh (token as OAuthTokenSetup), fetch upload path', () => { + test('upload: a 401 refreshes but surfaces the error; the retry uses the fresh token', async () => { + getActiveMock() + .scope(projectHost()) + .on('POST', '/v1/assets/images/foo') + .respond({status: 401, body: {error: {description: 'Token expired'}}}) + .respond({status: 201, body: {document: {url: 'https://some.asset.url'}}}) + + let currentToken = 'expired-token' + const refresh = vi.fn(() => { + currentToken = 'fresh-token' + return Promise.resolve(currentToken) + }) + const client = oauthClient({getToken: async () => currentToken, refresh}) + + // No auto-retry: the 401 surfaces (the body may be a consumed stream)... + const error = await client.assets.upload('image', Buffer.from('img')).catch((e: unknown) => e) + expect(error).toBeInstanceOf(ClientError) + if (!(error instanceof ClientError)) throw error + expect(error.statusCode).toBe(401) + // ...but the refresh already happened, so the caller's retry succeeds. + expect(refresh).toHaveBeenCalledTimes(1) + await expect(client.assets.upload('image', Buffer.from('img'))).resolves.toMatchObject({ + url: 'https://some.asset.url', + }) + expect(authHeaders()).toEqual(['Bearer expired-token', 'Bearer fresh-token']) + }) +}) diff --git a/test/oauthRefreshHandler.test.ts b/test/oauthRefreshHandler.test.ts new file mode 100644 index 000000000..3f8d16ff2 --- /dev/null +++ b/test/oauthRefreshHandler.test.ts @@ -0,0 +1,195 @@ +import {ClientError, ConnectionFailedError, type OAuthTokenSetup} from '@sanity/client' +import {encode} from 'eventsource-encoder' +import {firstValueFrom} from 'rxjs' +import {describe, expect, test, vi} from 'vitest' + +import {getClient, projectHost} from './client/helpers' +import {getActiveMock} from './helpers/mockFetch' + +const usersPath = '/v1/users/me' + +// The DX under test: an OAuth setup handed straight to `token` on the real client. +const oauthClient = (setup: OAuthTokenSetup) => getClient({token: setup}) + +function authHeaders(): Array { + return getActiveMock() + .getRequests() + .map((request) => request.headers.get('authorization')) +} + +describe('OAuth auto-refresh (token as OAuthTokenSetup)', () => { + test('proactive: applies the token from getToken() to every request', async () => { + getActiveMock().scope(projectHost()).on('GET', usersPath).respond({status: 200, body: {id: 'me'}}) + + const setup: OAuthTokenSetup = { + getToken: async () => 'proactive-token', + refresh: () => Promise.reject(new Error('should not refresh')), + } + const client = oauthClient(setup) + + await expect(client.users.getById('me')).resolves.toEqual({id: 'me'}) + expect(authHeaders()).toEqual(['Bearer proactive-token']) + }) + + test('reactive: a 401 refreshes then retries once with the new token', async () => { + getActiveMock() + .scope(projectHost()) + .on('GET', usersPath) + .respond({status: 401, body: {error: {description: 'Token expired'}}}) + .respond({status: 200, body: {id: 'me'}}) + + const refresh = vi.fn(() => Promise.resolve('fresh-token')) + const setup: OAuthTokenSetup = {getToken: async () => 'expired-token', refresh} + const client = oauthClient(setup) + + await expect(client.users.getById('me')).resolves.toEqual({id: 'me'}) + expect(refresh).toHaveBeenCalledTimes(1) + expect(authHeaders()).toEqual(['Bearer expired-token', 'Bearer fresh-token']) + }) + + test('single-flight: concurrent 401s share one refresh, then each retries', async () => { + const concurrency = 3 + const route = getActiveMock().scope(projectHost()).on('GET', usersPath) + for (let i = 0; i < concurrency; i++) { + route.respond({status: 401, body: {error: {description: 'Token expired'}}}) + } + for (let i = 0; i < concurrency; i++) { + route.respond({status: 200, body: {id: 'me'}}) + } + + // Gate refresh open until all requests reach it; a straggler starting a + // second refresh would fail the dedupe. + const gate = Promise.withResolvers() + const refresh = vi.fn(() => gate.promise) + const setup: OAuthTokenSetup = {getToken: async () => 'expired-token', refresh} + const client = oauthClient(setup) + + const inflight = Promise.all( + Array.from({length: concurrency}, () => client.users.getById('me')), + ) + + // Let the first attempts 401 and land on the shared refresh. + await new Promise((resolve) => setTimeout(resolve, 0)) + expect(refresh).toHaveBeenCalledTimes(1) + + gate.resolve('fresh-token') + await expect(inflight).resolves.toEqual([{id: 'me'}, {id: 'me'}, {id: 'me'}]) + + const headers = authHeaders() + expect(headers.filter((h) => h === 'Bearer expired-token')).toHaveLength(concurrency) + expect(headers.filter((h) => h === 'Bearer fresh-token')).toHaveLength(concurrency) + }) + + test('unrecoverable refresh: calls onAuthError and surfaces the original 401', async () => { + getActiveMock() + .scope(projectHost()) + .on('GET', usersPath) + .respond({status: 401, body: {error: {description: 'Token expired'}}}) + + const refreshError = new Error('refresh token expired') + const onAuthError = vi.fn() + const setup: OAuthTokenSetup = { + getToken: async () => 'expired-token', + refresh: () => Promise.reject(refreshError), + onAuthError, + } + const client = oauthClient(setup) + + const error = await client.users.getById('me').catch((e: unknown) => e) + expect(error).toBeInstanceOf(ClientError) + if (!(error instanceof ClientError)) throw error + expect(error.statusCode).toBe(401) + expect(onAuthError).toHaveBeenCalledTimes(1) + expect(onAuthError).toHaveBeenCalledWith(refreshError) + }) + + test('already refreshed: retries with the current token without refreshing', async () => { + getActiveMock() + .scope(projectHost()) + .on('GET', usersPath) + .respond({status: 401, body: {error: {description: 'Token expired'}}}) + .respond({status: 200, body: {id: 'me'}}) + + // First read is the token we send; by the 401 the provider already holds a + // newer one (another request or tab refreshed it). + const getToken = vi.fn(async () => 'current-token') + getToken.mockResolvedValueOnce('stale-token') + const refresh = vi.fn(() => Promise.resolve('unused-token')) + const client = oauthClient({getToken, refresh}) + + await expect(client.users.getById('me')).resolves.toEqual({id: 'me'}) + expect(refresh).not.toHaveBeenCalled() + expect(authHeaders()).toEqual(['Bearer stale-token', 'Bearer current-token']) + }) + + test('listen: a 401-rejected connection refreshes then reconnects with the new token', async () => { + getActiveMock() + .scope(projectHost()) + .on('GET', '/v1/data/listen/foo') + .respond({status: 401, body: 'Unauthorized'}) + .respond({ + status: 200, + body: encode({event: 'welcome', data: '{}'}), + headers: {'Content-Type': 'text/event-stream'}, + }) + + let currentToken = 'expired-token' + const refresh = vi.fn(() => { + currentToken = 'fresh-token' + return Promise.resolve(currentToken) + }) + const client = oauthClient({getToken: async () => currentToken, refresh}) + + const event = await firstValueFrom(client.listen('*', {}, {events: ['welcome']})) + expect(event).toEqual({type: 'welcome'}) + expect(refresh).toHaveBeenCalledTimes(1) + expect(authHeaders()).toEqual(['Bearer expired-token', 'Bearer fresh-token']) + }) + + test('listen: a second consecutive 401 surfaces without another refresh', async () => { + getActiveMock() + .scope(projectHost()) + .on('GET', '/v1/data/listen/foo') + .respondPersist({status: 401, body: 'Unauthorized'}) + + const refresh = vi.fn(() => Promise.resolve('fresh-but-rejected-token')) + const client = oauthClient({getToken: async () => 'expired-token', refresh}) + + const error = await firstValueFrom(client.listen('*', {}, {events: ['welcome']})).catch( + (e: unknown) => e, + ) + expect(error).toBeInstanceOf(ConnectionFailedError) + if (!(error instanceof ConnectionFailedError)) throw error + expect(error.status).toBe(401) + expect(refresh).toHaveBeenCalledTimes(1) + }) + + test('unrecoverable refresh: fires onAuthError once across concurrent 401s', async () => { + const concurrency = 3 + const route = getActiveMock().scope(projectHost()).on('GET', usersPath) + for (let i = 0; i < concurrency; i++) { + route.respond({status: 401, body: {error: {description: 'Token expired'}}}) + } + + // Gate the shared refresh open until all three 401s have joined it, so the + // rejection is observed by all three but onAuthError fires once. + const gate = Promise.withResolvers() + const refresh = vi.fn(() => gate.promise) + const onAuthError = vi.fn() + const client = oauthClient({getToken: async () => 'expired-token', refresh, onAuthError}) + + const inflight = Promise.allSettled( + Array.from({length: concurrency}, () => client.users.getById('me')), + ) + await new Promise((resolve) => setTimeout(resolve, 0)) + expect(refresh).toHaveBeenCalledTimes(1) + + const refreshError = new Error('refresh token expired') + gate.reject(refreshError) + const results = await inflight + + expect(results.every((r) => r.status === 'rejected')).toBe(true) + expect(onAuthError).toHaveBeenCalledTimes(1) + expect(onAuthError).toHaveBeenCalledWith(refreshError) + }) +}) diff --git a/test/resumability.test.ts b/test/resumability.test.ts new file mode 100644 index 000000000..45bd794ea --- /dev/null +++ b/test/resumability.test.ts @@ -0,0 +1,100 @@ +import {ConnectionFailedError, type OAuthTokenSetup} from '@sanity/client' +import {encode} from 'eventsource-encoder' +import {firstValueFrom, take, toArray} from 'rxjs' +import {expect, test} from 'vitest' + +import {getClient, projectHost} from './client/helpers' +import {getActiveMock} from './helpers/mockFetch' + +const sse = (body: string) => ({status: 200, body, headers: {'Content-Type': 'text/event-stream'}}) +const oauthClient = (setup: OAuthTokenSetup) => getClient({token: setup}) + +test('given a listener with a string token, when the server drops the connection, then the eventsource lib reconnects with Last-Event-ID and the same token', async () => { + getActiveMock() + .scope(projectHost()) + .on('GET', '/v1/data/listen/foo') + .respond(sse(`retry: 1\n\n` + encode({event: 'mutation', id: 'evt-1', data: '{}'}))) + .respond(sse(encode({event: 'mutation', id: 'evt-2', data: '{}'}))) + + const client = getClient({token: 'static-token'}) + expect(await firstValueFrom(client.listen('*').pipe(take(2), toArray()))).toHaveLength(2) + + const requests = getActiveMock().getRequests() + expect(requests.map((r) => r.headers.get('last-event-id'))).toEqual([null, 'evt-1']) + expect(requests.map((r) => r.headers.get('authorization'))).toEqual([ + 'Bearer static-token', + 'Bearer static-token', + ]) +}) + +test('given a listener whose OAuth token rotates mid-stream, when the server drops the connection, then the eventsource lib reconnects with Last-Event-ID and the current token', async () => { + getActiveMock() + .scope(projectHost()) + .on('GET', '/v1/data/listen/foo') + // `retry: 1` so the eventsource lib reconnects in 1ms instead of 3s; body then ends (server drop) + .respond(sse(`retry: 1\n\n` + encode({event: 'mutation', id: 'evt-1', data: '{}'}))) + .respond(sse(encode({event: 'mutation', id: 'evt-2', data: '{}'}))) + + let currentToken = 'token-a' + const client = oauthClient({ + getToken: async () => currentToken, + refresh: () => Promise.reject(new Error('not needed')), + }) + + const events = firstValueFrom(client.listen('*').pipe(take(2), toArray())) + // rotate between the two connections, as a provider's background refresh would + await new Promise((r) => setTimeout(r, 0)) + currentToken = 'token-b' + expect(await events).toHaveLength(2) + + const requests = getActiveMock().getRequests() + expect(requests.map((r) => r.headers.get('last-event-id'))).toEqual([null, 'evt-1']) + expect(requests.map((r) => r.headers.get('authorization'))).toEqual([ + 'Bearer token-a', + 'Bearer token-b', + ]) +}) + +test('given a listener with a string token, when the reconnect is rejected with a 401, then the error surfaces and nothing reconnects', async () => { + getActiveMock() + .scope(projectHost()) + .on('GET', '/v1/data/listen/foo') + .respond(sse(`retry: 1\n\n` + encode({event: 'mutation', id: 'evt-1', data: '{}'}))) + .respond({status: 401, body: 'Unauthorized'}) + + const client = getClient({token: 'static-token'}) + const error = await firstValueFrom(client.listen('*').pipe(take(2), toArray())).catch( + (e: unknown) => e, + ) + + expect(error).toBeInstanceOf(ConnectionFailedError) + if (!(error instanceof ConnectionFailedError)) throw error + expect(error.status).toBe(401) + expect(getActiveMock().getRequests()).toHaveLength(2) +}) + +test('given a listener whose token has expired, when the reconnect is rejected with a 401, then the EventSource closes and a refreshed one opens without Last-Event-ID', async () => { + getActiveMock() + .scope(projectHost()) + .on('GET', '/v1/data/listen/foo') + .respond(sse(`retry: 1\n\n` + encode({event: 'mutation', id: 'evt-1', data: '{}'}))) + .respond({status: 401, body: 'Unauthorized'}) + .respond(sse(encode({event: 'mutation', id: 'evt-2', data: '{}'}))) + + let currentToken = 'expired' + const client = oauthClient({ + getToken: async () => currentToken, + refresh: async () => (currentToken = 'fresh'), + }) + + expect(await firstValueFrom(client.listen('*').pipe(take(2), toArray()))).toHaveLength(2) + + const requests = getActiveMock().getRequests() + expect(requests.map((r) => r.headers.get('authorization'))).toEqual([ + 'Bearer expired', + 'Bearer expired', + 'Bearer fresh', + ]) + // Last-Event-ID survives the lib's own reconnect (2nd request) but not ours (3rd) + expect(requests.map((r) => r.headers.get('last-event-id'))).toEqual([null, 'evt-1', null]) +}) From dcbafe4ea517778ac52af83947333aee6e55b234 Mon Sep 17 00:00:00 2001 From: Josh Date: Mon, 7 Sep 2026 16:02:46 +0100 Subject: [PATCH 2/7] feat: add proactive refresh --- src/data/resolveEventSourceFetch.ts | 13 ++-- src/http/oauthRefreshHandler.ts | 29 ++++++-- src/types.ts | 5 ++ test/oauthRefreshHandler.test.ts | 103 +++++++++++++++++++++++++--- 4 files changed, 130 insertions(+), 20 deletions(-) diff --git a/src/data/resolveEventSourceFetch.ts b/src/data/resolveEventSourceFetch.ts index 48be81a94..58b844fb1 100644 --- a/src/data/resolveEventSourceFetch.ts +++ b/src/data/resolveEventSourceFetch.ts @@ -1,6 +1,7 @@ import type {EventSourceFetchInit, FetchLikeResponse} from 'eventsource' import type {FetchFunction, FetchInit} from 'get-it' +import {resolveOAuthToken} from '../http/oauthRefreshHandler' import type {InitializedClientConfig, OAuthTokenSetup} from '../types' /** @internal */ @@ -13,11 +14,11 @@ export interface EventSourceFetchOptions { headers?: Record /** * OAuth token setup to resolve an `Authorization` header from. Resolved via - * `getToken()` on every request — not once per connection — so the - * `eventsource` package's reconnects pick up a refreshed token. This fetch - * only reads; 401-driven `refresh()` lives upstream in - * `reconnectOnConnectionFailure`. Config `headers` take precedence, - * mirroring the string-token merge order. + * `resolveOAuthToken()` on every request — not once per connection — so the + * `eventsource` package's reconnects pick up a refreshed token, and a token + * about to expire is refreshed proactively. 401-driven `refresh()` still + * lives upstream in `reconnectOnConnectionFailure`. Config `headers` take + * precedence, mirroring the string-token merge order. */ tokenSetup?: OAuthTokenSetup /** @@ -72,7 +73,7 @@ export function resolveEventSourceFetch( if (extraHeaders || tokenSetup) { const headers = new Headers(init?.headers) if (tokenSetup) { - headers.set('Authorization', `Bearer ${await tokenSetup.getToken()}`) + headers.set('Authorization', `Bearer ${await resolveOAuthToken(tokenSetup)}`) } if (extraHeaders) { for (const [key, value] of Object.entries(extraHeaders)) { diff --git a/src/http/oauthRefreshHandler.ts b/src/http/oauthRefreshHandler.ts index 73fbec848..2e619b28e 100644 --- a/src/http/oauthRefreshHandler.ts +++ b/src/http/oauthRefreshHandler.ts @@ -32,7 +32,7 @@ export async function applyOAuthToken( ): Promise> { // `new Headers()` gives a case-insensitive lookup over the plain record. if (new Headers(headers).has('authorization')) return headers - return {...headers, Authorization: `Bearer ${await setup.getToken()}`} + return {...headers, Authorization: `Bearer ${await resolveOAuthToken(setup)}`} } /** @@ -73,6 +73,27 @@ export function getOAuthRefresher(setup: OAuthTokenSetup): () => Promise return refresher } +/** + * In ms how early before the expiry do we attempt an automatic refresh. + * TODO: define this properly based on what the end lifespan of a token is. + */ +const REFRESH_SKEW_MS = 30_000 + +/** + * `getToken()`, or a single-flight `refresh()` when the token is about to + * expire. A failed proactive refresh falls back to the current token and + * lets the 401 path decide (onAuthError already fired inside the refresher). + * + * @internal + */ +export async function resolveOAuthToken(setup: OAuthTokenSetup): Promise { + const expiresAt = setup.getExpiresAt?.() + if (expiresAt !== undefined && Date.now() >= expiresAt - REFRESH_SKEW_MS) { + return getOAuthRefresher(setup)().catch(() => setup.getToken()) + } + return setup.getToken() +} + /** * 401 handling for paths that can't safely retry — uploads, where a * `NodeJS.ReadableStream` body is consumed by the first attempt. Refreshes so @@ -115,9 +136,7 @@ export async function refreshOnAuthError( * * @internal */ -export function resolveRequestHandler( - config: InitializedClientConfig, -): RequestHandler | undefined { +export function resolveRequestHandler(config: InitializedClientConfig): RequestHandler | undefined { const setup = getOAuthTokenSetup(config.token) if (!setup) return config.requestHandler const oauthHandler = createOAuthRefreshHandler(setup) @@ -151,7 +170,7 @@ function createOAuthRefreshHandler(setup: OAuthTokenSetup): RequestHandler { // a 401 against a token this handler didn't supply isn't its to fix. if (new Headers(request.headers).has('authorization')) return next(request) - const token = await setup.getToken() + const token = await resolveOAuthToken(setup) try { return await next(withToken(request, token)) } catch (error) { diff --git a/src/types.ts b/src/types.ts index addeefbc5..ddbea2a7b 100644 --- a/src/types.ts +++ b/src/types.ts @@ -86,6 +86,11 @@ export interface OAuthTokenSetup { * resolved, subsequent `getToken()` calls must return the refreshed token. */ refresh: () => Promise + /** + * Epoch ms the current access token expires. When provided, the client + * refreshes shortly before expiry instead of waiting for a 401. + */ + getExpiresAt?: () => number | undefined /** Called when `refresh()` rejects */ onAuthError?: (error: unknown) => void } diff --git a/test/oauthRefreshHandler.test.ts b/test/oauthRefreshHandler.test.ts index 3f8d16ff2..38aff676a 100644 --- a/test/oauthRefreshHandler.test.ts +++ b/test/oauthRefreshHandler.test.ts @@ -18,8 +18,11 @@ function authHeaders(): Array { } describe('OAuth auto-refresh (token as OAuthTokenSetup)', () => { - test('proactive: applies the token from getToken() to every request', async () => { - getActiveMock().scope(projectHost()).on('GET', usersPath).respond({status: 200, body: {id: 'me'}}) + test('applies the token from getToken() to every request', async () => { + getActiveMock() + .scope(projectHost()) + .on('GET', usersPath) + .respond({status: 200, body: {id: 'me'}}) const setup: OAuthTokenSetup = { getToken: async () => 'proactive-token', @@ -31,7 +34,89 @@ describe('OAuth auto-refresh (token as OAuthTokenSetup)', () => { expect(authHeaders()).toEqual(['Bearer proactive-token']) }) - test('reactive: a 401 refreshes then retries once with the new token', async () => { + test('an expiring token refreshes before sending, without a 401', async () => { + getActiveMock() + .scope(projectHost()) + .on('GET', usersPath) + .respond({status: 200, body: {id: 'me'}}) + + const refresh = vi.fn(() => Promise.resolve('fresh-token')) + const setup: OAuthTokenSetup = { + getToken: async () => 'expiring-token', + refresh, + getExpiresAt: () => Date.now() + 10_000, + } + const client = oauthClient(setup) + + await expect(client.users.getById('me')).resolves.toEqual({id: 'me'}) + expect(refresh).toHaveBeenCalledTimes(1) + expect(authHeaders()).toEqual(['Bearer fresh-token']) + }) + + test('concurrent requests with an expiring token refresh once', async () => { + const concurrency = 2 + const route = getActiveMock().scope(projectHost()).on('GET', usersPath) + for (let i = 0; i < concurrency; i++) { + route.respond({status: 200, body: {id: 'me'}}) + } + + const refresh = vi.fn(() => Promise.resolve('fresh-token')) + const setup: OAuthTokenSetup = { + getToken: async () => 'expiring-token', + refresh, + getExpiresAt: () => Date.now() + 10_000, + } + const client = oauthClient(setup) + + await expect( + Promise.all(Array.from({length: concurrency}, () => client.users.getById('me'))), + ).resolves.toEqual([{id: 'me'}, {id: 'me'}]) + expect(refresh).toHaveBeenCalledTimes(1) + expect(authHeaders()).toEqual(['Bearer fresh-token', 'Bearer fresh-token']) + }) + + test('a far-future expiry does not refresh', async () => { + getActiveMock() + .scope(projectHost()) + .on('GET', usersPath) + .respond({status: 200, body: {id: 'me'}}) + + const refresh = vi.fn(() => Promise.reject(new Error('should not refresh'))) + const setup: OAuthTokenSetup = { + getToken: async () => 'valid-token', + refresh, + getExpiresAt: () => Date.now() + 60 * 60 * 1000, + } + const client = oauthClient(setup) + + await expect(client.users.getById('me')).resolves.toEqual({id: 'me'}) + expect(refresh).not.toHaveBeenCalled() + expect(authHeaders()).toEqual(['Bearer valid-token']) + }) + + test('a rejected refresh fires onAuthError and sends the getToken() token', async () => { + getActiveMock() + .scope(projectHost()) + .on('GET', usersPath) + .respond({status: 200, body: {id: 'me'}}) + + const refreshError = new Error('refresh token expired') + const onAuthError = vi.fn() + const setup: OAuthTokenSetup = { + getToken: async () => 'expiring-token', + refresh: () => Promise.reject(refreshError), + getExpiresAt: () => Date.now() + 10_000, + onAuthError, + } + const client = oauthClient(setup) + + await expect(client.users.getById('me')).resolves.toEqual({id: 'me'}) + expect(onAuthError).toHaveBeenCalledTimes(1) + expect(onAuthError).toHaveBeenCalledWith(refreshError) + expect(authHeaders()).toEqual(['Bearer expiring-token']) + }) + + test('a 401 refreshes then retries once with the new token', async () => { getActiveMock() .scope(projectHost()) .on('GET', usersPath) @@ -47,7 +132,7 @@ describe('OAuth auto-refresh (token as OAuthTokenSetup)', () => { expect(authHeaders()).toEqual(['Bearer expired-token', 'Bearer fresh-token']) }) - test('single-flight: concurrent 401s share one refresh, then each retries', async () => { + test('concurrent 401s share one refresh, then each retries', async () => { const concurrency = 3 const route = getActiveMock().scope(projectHost()).on('GET', usersPath) for (let i = 0; i < concurrency; i++) { @@ -80,7 +165,7 @@ describe('OAuth auto-refresh (token as OAuthTokenSetup)', () => { expect(headers.filter((h) => h === 'Bearer fresh-token')).toHaveLength(concurrency) }) - test('unrecoverable refresh: calls onAuthError and surfaces the original 401', async () => { + test('calls onAuthError and surfaces the original 401 for an unrecoverable refresh', async () => { getActiveMock() .scope(projectHost()) .on('GET', usersPath) @@ -103,7 +188,7 @@ describe('OAuth auto-refresh (token as OAuthTokenSetup)', () => { expect(onAuthError).toHaveBeenCalledWith(refreshError) }) - test('already refreshed: retries with the current token without refreshing', async () => { + test('retries with the current token without refreshing', async () => { getActiveMock() .scope(projectHost()) .on('GET', usersPath) @@ -122,7 +207,7 @@ describe('OAuth auto-refresh (token as OAuthTokenSetup)', () => { expect(authHeaders()).toEqual(['Bearer stale-token', 'Bearer current-token']) }) - test('listen: a 401-rejected connection refreshes then reconnects with the new token', async () => { + test('a 401-rejected connection refreshes then reconnects with the new token', async () => { getActiveMock() .scope(projectHost()) .on('GET', '/v1/data/listen/foo') @@ -146,7 +231,7 @@ describe('OAuth auto-refresh (token as OAuthTokenSetup)', () => { expect(authHeaders()).toEqual(['Bearer expired-token', 'Bearer fresh-token']) }) - test('listen: a second consecutive 401 surfaces without another refresh', async () => { + test('a second consecutive 401 surfaces without another refresh', async () => { getActiveMock() .scope(projectHost()) .on('GET', '/v1/data/listen/foo') @@ -164,7 +249,7 @@ describe('OAuth auto-refresh (token as OAuthTokenSetup)', () => { expect(refresh).toHaveBeenCalledTimes(1) }) - test('unrecoverable refresh: fires onAuthError once across concurrent 401s', async () => { + test('fires onAuthError once across concurrent 401s for an unrecoverable refresh', async () => { const concurrency = 3 const route = getActiveMock().scope(projectHost()).on('GET', usersPath) for (let i = 0; i < concurrency; i++) { From a7982d03c8faafd1b4faa67ab9d1d6eec02e320c Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 10 Sep 2026 17:48:07 +0000 Subject: [PATCH 3/7] fix: preserve SSE resume position across OAuth reconnects Co-authored-by: joshuaellis <37798644+joshuaellis@users.noreply.github.com> --- src/data/listen.ts | 9 +++++- src/data/live.ts | 10 ++++++- src/data/resolveEventSourceFetch.ts | 14 ++++++++- test/resumability.test.ts | 44 +++++++++++++++++++++++------ 4 files changed, 65 insertions(+), 12 deletions(-) diff --git a/src/data/listen.ts b/src/data/listen.ts index 86a15f1bd..9ccf9f18a 100644 --- a/src/data/listen.ts +++ b/src/data/listen.ts @@ -1,6 +1,6 @@ import {EventSource} from 'eventsource' import {Observable, throwError} from 'rxjs' -import {filter, map} from 'rxjs/operators' +import {filter, map, tap} from 'rxjs/operators' import {getOAuthRefresher, getOAuthTokenSetup} from '../http/oauthRefreshHandler' import type {ObservableSanityClient, SanityClient} from '../SanityClient' @@ -174,17 +174,24 @@ export function _connectListenEventSource( Object.assign(headers, configHeaders) } const tokenSetup = getOAuthTokenSetup(token) + let lastEventId: string | undefined const initEventSource = () => new EventSource(uri, { fetch: resolveEventSourceFetch(config, { headers: Object.keys(headers).length ? headers : undefined, + lastEventId, tokenSetup, withCredentials, }), }) return connectEventSource(initEventSource, listenFor).pipe( + tap((event) => { + if ('id' in event && typeof event.id === 'string' && event.id) { + lastEventId = event.id + } + }), reconnectOnConnectionFailure(tokenSetup && getOAuthRefresher(tokenSetup)), filter((event) => listenFor.includes(event.type)), map((event) => ({ diff --git a/src/data/live.ts b/src/data/live.ts index 279c81f69..894c8db1d 100644 --- a/src/data/live.ts +++ b/src/data/live.ts @@ -1,7 +1,7 @@ import {EventSource} from 'eventsource' import type {FetchFunction} from 'get-it' import {catchError, mergeMap, Observable, of, throwError} from 'rxjs' -import {finalize, map} from 'rxjs/operators' +import {finalize, map, tap} from 'rxjs/operators' import {CorsOriginError} from '../http/errors' import {getOAuthRefresher, getOAuthTokenSetup} from '../http/oauthRefreshHandler' @@ -125,10 +125,13 @@ export class LiveClient { return existing } + let lastEventId: string | undefined + const initEventSource = () => new EventSource(url.href, { fetch: resolveEventSourceFetch(config, { headers: Object.keys(eventSourceHeaders).length ? eventSourceHeaders : undefined, + lastEventId, tokenSetup, withCredentials: eventSourceWithCredentials, }), @@ -151,6 +154,11 @@ export class LiveClient { const observable = events .pipe( + tap((event) => { + if ('id' in event && typeof event.id === 'string' && event.id) { + lastEventId = event.id + } + }), reconnectOnConnectionFailure(tokenSetup && getOAuthRefresher(tokenSetup)), mergeMap((event) => { if (event.type === 'reconnect') { diff --git a/src/data/resolveEventSourceFetch.ts b/src/data/resolveEventSourceFetch.ts index 58b844fb1..27f668e31 100644 --- a/src/data/resolveEventSourceFetch.ts +++ b/src/data/resolveEventSourceFetch.ts @@ -12,6 +12,14 @@ export interface EventSourceFetchOptions { * etc. — things the native EventSource API has no equivalent for. */ headers?: Record + /** + * Last event id from a prior EventSource instance. Used when the client has + * to construct a fresh EventSource itself (eg after an OAuth refresh on a + * rejected reconnect) so the new instance can resume from the previous + * position. The `eventsource` package's own reconnect header, when present, + * still wins. + */ + lastEventId?: string /** * OAuth token setup to resolve an `Authorization` header from. Resolved via * `resolveOAuthToken()` on every request — not once per connection — so the @@ -60,6 +68,7 @@ export function resolveEventSourceFetch( options: EventSourceFetchOptions = {}, ): EventSourceFetch { const extraHeaders = options.headers + const lastEventId = options.lastEventId const tokenSetup = options.tokenSetup const credentials: FetchInit['credentials'] = options.withCredentials ? 'include' : undefined @@ -70,8 +79,11 @@ export function resolveEventSourceFetch( // declare (`mode`, `cache`) survive the spread and reach whichever // fetch implementation is effective. const mergedInit: FetchInit = {...init} - if (extraHeaders || tokenSetup) { + if (extraHeaders || tokenSetup || lastEventId) { const headers = new Headers(init?.headers) + if (lastEventId && !headers.has('last-event-id')) { + headers.set('Last-Event-ID', lastEventId) + } if (tokenSetup) { headers.set('Authorization', `Bearer ${await resolveOAuthToken(tokenSetup)}`) } diff --git a/test/resumability.test.ts b/test/resumability.test.ts index 45bd794ea..1926105e1 100644 --- a/test/resumability.test.ts +++ b/test/resumability.test.ts @@ -4,7 +4,7 @@ import {firstValueFrom, take, toArray} from 'rxjs' import {expect, test} from 'vitest' import {getClient, projectHost} from './client/helpers' -import {getActiveMock} from './helpers/mockFetch' +import {getActiveMock, streamBody, streamStall} from './helpers/mockFetch' const sse = (body: string) => ({status: 200, body, headers: {'Content-Type': 'text/event-stream'}}) const oauthClient = (setup: OAuthTokenSetup) => getClient({token: setup}) @@ -73,13 +73,27 @@ test('given a listener with a string token, when the reconnect is rejected with expect(getActiveMock().getRequests()).toHaveLength(2) }) -test('given a listener whose token has expired, when the reconnect is rejected with a 401, then the EventSource closes and a refreshed one opens without Last-Event-ID', async () => { - getActiveMock() - .scope(projectHost()) +test('given a resumable listener whose token has expired, when the reconnect is rejected with a 401, then the refreshed EventSource resumes from the last event id', async () => { + const scope = getActiveMock().scope(projectHost()) + scope .on('GET', '/v1/data/listen/foo') - .respond(sse(`retry: 1\n\n` + encode({event: 'mutation', id: 'evt-1', data: '{}'}))) + .respond( + sse( + `retry: 1\n\n` + + encode({event: 'welcome', data: JSON.stringify({listenerName: 'foo-1'})}) + + encode({event: 'mutation', id: 'evt-1', data: '{}'}), + ), + ) .respond({status: 401, body: 'Unauthorized'}) - .respond(sse(encode({event: 'mutation', id: 'evt-2', data: '{}'}))) + scope.on('GET', '/v1/data/listen/foo', {headers: {'Last-Event-ID': 'evt-1'}}).respond({ + status: 200, + headers: {'Content-Type': 'text/event-stream'}, + body: streamBody( + encode({event: 'welcomeback', data: JSON.stringify({listenerName: 'foo-2'})}) + + encode({event: 'mutation', id: 'evt-2', data: '{}'}), + streamStall(), + ), + }) let currentToken = 'expired' const client = oauthClient({ @@ -87,7 +101,20 @@ test('given a listener whose token has expired, when the reconnect is rejected w refresh: async () => (currentToken = 'fresh'), }) - expect(await firstValueFrom(client.listen('*').pipe(take(2), toArray()))).toHaveLength(2) + expect( + await firstValueFrom( + client + .listen('*', {}, {enableResume: true, events: ['welcome', 'welcomeback', 'reconnect', 'mutation']}) + .pipe(take(6), toArray()), + ), + ).toEqual([ + {type: 'welcome', listenerName: 'foo-1'}, + {type: 'mutation'}, + {type: 'reconnect'}, + {type: 'reconnect'}, + {type: 'welcomeback', listenerName: 'foo-2'}, + {type: 'mutation'}, + ]) const requests = getActiveMock().getRequests() expect(requests.map((r) => r.headers.get('authorization'))).toEqual([ @@ -95,6 +122,5 @@ test('given a listener whose token has expired, when the reconnect is rejected w 'Bearer expired', 'Bearer fresh', ]) - // Last-Event-ID survives the lib's own reconnect (2nd request) but not ours (3rd) - expect(requests.map((r) => r.headers.get('last-event-id'))).toEqual([null, 'evt-1', null]) + expect(requests.map((r) => r.headers.get('last-event-id'))).toEqual([null, 'evt-1', 'evt-1']) }) From 3e0750fc0a6b2c57822ddd4e23ae8bf6418af29f Mon Sep 17 00:00:00 2001 From: Josh Date: Mon, 21 Sep 2026 07:59:13 +0100 Subject: [PATCH 4/7] chore: pr amends --- src/assets/AssetsClient.ts | 2 +- src/config.ts | 10 +++--- src/data/dataMethods.ts | 2 +- src/data/listen.ts | 4 +-- src/data/live.ts | 4 +-- src/data/reconnectOnConnectionFailure.ts | 4 +-- src/data/resolveEventSourceFetch.ts | 9 +----- src/http/oauthRefreshHandler.ts | 10 +++--- src/http/requestOptions.ts | 5 +-- src/types.ts | 13 ++++++-- test/oauthRefreshHandler.node.test.ts | 2 +- test/oauthRefreshHandler.test.ts | 15 +++++++-- test/resumability.test.ts | 40 ++++++++++++++++++++++-- 13 files changed, 81 insertions(+), 39 deletions(-) diff --git a/src/assets/AssetsClient.ts b/src/assets/AssetsClient.ts index eefafefee..0fa3e3d19 100644 --- a/src/assets/AssetsClient.ts +++ b/src/assets/AssetsClient.ts @@ -266,7 +266,7 @@ function _upload< // XHR uploads bypass the request handler, so the OAuth token is resolved // (and a 401 refreshed, without auto-retry) here — same rules as the // fetch-path `_uploadObservable`. - const oauth = getOAuthTokenSetup(config.token) + const oauth = getOAuthTokenSetup(config) const reqHeaders = oauth ? await applyOAuthToken(oauth, req.headers) : req.headers const upload = uploadWithProgress({ url: appendQuery(req.url, req.query), diff --git a/src/config.ts b/src/config.ts index a458f1aee..f6bd9ff4e 100644 --- a/src/config.ts +++ b/src/config.ts @@ -121,15 +121,15 @@ export const initConfig = ( const isLocalhost = isBrowser && isLocal(window.location.hostname) const hasToken = Boolean(newConfig.token) - if (newConfig.withCredentials && hasToken) { + if (hasToken && newConfig.auth?.oauth) { + throw new Error('`token` and `auth.oauth` are mutually exclusive, configure one or the other') + } + if (newConfig.withCredentials && (hasToken || newConfig.auth?.oauth)) { warnings.printCredentialedTokenWarning() newConfig.withCredentials = false } - // The browser warning targets a secret baked into client-side code. An - // OAuthTokenSetup obtains tokens per user at runtime, so it is exempt. - const hasStringToken = typeof newConfig.token === 'string' && newConfig.token !== '' - if (isBrowser && isLocalhost && hasStringToken && newConfig.ignoreBrowserTokenWarning !== true) { + if (isBrowser && isLocalhost && hasToken && newConfig.ignoreBrowserTokenWarning !== true) { warnings.printBrowserTokenWarning() } else if (typeof newConfig.useCdn === 'undefined') { warnings.printCdnWarning() diff --git a/src/data/dataMethods.ts b/src/data/dataMethods.ts index fab02b190..a31834d5c 100644 --- a/src/data/dataMethods.ts +++ b/src/data/dataMethods.ts @@ -1203,7 +1203,7 @@ export function _uploadObservable( // here — per subscription, so a resubscribe picks up a refreshed token. A // 401 refreshes but does not auto-retry (the body may be a consumed stream); // the error surfaces and a caller-level retry gets the fresh token. - const oauth = getOAuthTokenSetup(config.token) + const oauth = getOAuthTokenSetup(config) const upload = (req: FetchRequest) => new Observable((subscriber) => requester(req).subscribe(subscriber)) const request = ( diff --git a/src/data/listen.ts b/src/data/listen.ts index 9ccf9f18a..1c379dca1 100644 --- a/src/data/listen.ts +++ b/src/data/listen.ts @@ -167,13 +167,13 @@ export function _connectListenEventSource( const {token, withCredentials, headers: configHeaders} = config const headers: Record = {} - if (typeof token === 'string') { + if (token) { headers.Authorization = `Bearer ${token}` } if (configHeaders) { Object.assign(headers, configHeaders) } - const tokenSetup = getOAuthTokenSetup(token) + const tokenSetup = getOAuthTokenSetup(config) let lastEventId: string | undefined const initEventSource = () => diff --git a/src/data/live.ts b/src/data/live.ts index 894c8db1d..e9f8635ab 100644 --- a/src/data/live.ts +++ b/src/data/live.ts @@ -92,7 +92,7 @@ export class LiveClient { url.searchParams.set('waitFor', waitFor) } const eventSourceHeaders: Record = {} - if (includeDrafts && typeof token === 'string') { + if (includeDrafts && token) { eventSourceHeaders.Authorization = `Bearer ${token}` } if (configHeaders) { @@ -101,7 +101,7 @@ export class LiveClient { // An OAuth token setup can't be baked into the headers — it is resolved // per request inside the EventSource fetch, so reconnects pick up // refreshed tokens. - const tokenSetup = includeDrafts ? getOAuthTokenSetup(token) : undefined + const tokenSetup = includeDrafts ? getOAuthTokenSetup(config) : undefined const eventSourceWithCredentials = Boolean(includeDrafts && withCredentials) let transportCache = eventsCache.get(config.resolveFetch) diff --git a/src/data/reconnectOnConnectionFailure.ts b/src/data/reconnectOnConnectionFailure.ts index 764b290d9..4b7439466 100644 --- a/src/data/reconnectOnConnectionFailure.ts +++ b/src/data/reconnectOnConnectionFailure.ts @@ -1,7 +1,7 @@ import { catchError, concat, - from, + defer, mergeMap, Observable, of, @@ -64,7 +64,7 @@ export function reconnectOnConnectionFailure( lastAuthRetryAt = Date.now() return concat( of({type: 'reconnect' as const}), - from(refreshAuth()).pipe( + defer(refreshAuth).pipe( // Surface the original connection error, not the refresh failure // — `onAuthError` has already fired inside the refresher. Placed // before `mergeMap` so it only catches the refresh promise, never diff --git a/src/data/resolveEventSourceFetch.ts b/src/data/resolveEventSourceFetch.ts index 27f668e31..fc0247b40 100644 --- a/src/data/resolveEventSourceFetch.ts +++ b/src/data/resolveEventSourceFetch.ts @@ -99,14 +99,7 @@ export function resolveEventSourceFetch( } // get-it's `FetchResponse` is a structural superset of the package's // `FetchLikeResponse`, so it can be handed over as-is. - const response = baseFetch(typeof url === 'string' ? url : url.href, mergedInit) - // Returning a promise from an async function attaches its rejection - // handler one microtask later (thenable adoption), and workerd's - // unhandled-rejection tracker flags a rejected promise in that gap. - // Attach a no-op handler synchronously; the rejection still propagates - // through the async return to the `eventsource` package's catch. - response.catch(() => {}) - return response + return await baseFetch(typeof url === 'string' ? url : url.href, mergedInit) } } diff --git a/src/http/oauthRefreshHandler.ts b/src/http/oauthRefreshHandler.ts index 2e619b28e..c5894346a 100644 --- a/src/http/oauthRefreshHandler.ts +++ b/src/http/oauthRefreshHandler.ts @@ -7,16 +7,16 @@ import type { import {ClientError} from './errors' /** - * The single spot deciding "is this token an OAuth setup" — shared by request + * The single spot reading the OAuth setup off a config — shared by request * handler resolution, SSE and asset uploads so the paths can never disagree on - * what counts as one. + * where it lives. * * @internal */ export function getOAuthTokenSetup( - token: InitializedClientConfig['token'], + config: Pick, ): OAuthTokenSetup | undefined { - return token && typeof token === 'object' ? token : undefined + return config.auth?.oauth } /** @@ -137,7 +137,7 @@ export async function refreshOnAuthError( * @internal */ export function resolveRequestHandler(config: InitializedClientConfig): RequestHandler | undefined { - const setup = getOAuthTokenSetup(config.token) + const setup = getOAuthTokenSetup(config) if (!setup) return config.requestHandler const oauthHandler = createOAuthRefreshHandler(setup) const userHandler = config.requestHandler diff --git a/src/http/requestOptions.ts b/src/http/requestOptions.ts index d071dc4d0..cc47ac381 100644 --- a/src/http/requestOptions.ts +++ b/src/http/requestOptions.ts @@ -34,11 +34,8 @@ export function requestOptions(config: Any, overrides: Any = {}): FetchRequest { Object.assign(headers, config.headers) } - // A string token becomes a Bearer header here; an OAuthTokenSetup object is - // resolved by the OAuth refresh handler (see `resolveRequestHandler`), which - // sets the header itself — never stringify the object into one. const token = overrides.token || config.token - if (typeof token === 'string') { + if (token) { headers['Authorization'] = `Bearer ${token}` } diff --git a/src/types.ts b/src/types.ts index ddbea2a7b..1e5c20101 100644 --- a/src/types.ts +++ b/src/types.ts @@ -71,8 +71,8 @@ export type RequestHandler = ( ) => Promise /** - * An OAuth token setup, accepted as `token` in place of a static string so the - * client can refresh a short-lived access token transparently. + * An OAuth token setup, configured as `auth.oauth` so the client can refresh a + * short-lived access token transparently instead of using a static `token`. * @public */ export interface OAuthTokenSetup { @@ -157,7 +157,14 @@ export interface ClientConfig { dataset?: string /** @defaultValue true */ useCdn?: boolean - token?: string | OAuthTokenSetup + token?: string + auth?: { + /** + * OAuth token setup for transparent access-token refresh. Mutually + * exclusive with `token`. + */ + oauth?: OAuthTokenSetup + } /** * Configure the client to work with a specific Sanity resource (Media Library, Canvas, etc.) diff --git a/test/oauthRefreshHandler.node.test.ts b/test/oauthRefreshHandler.node.test.ts index 641a3bff3..5ffdbdac9 100644 --- a/test/oauthRefreshHandler.node.test.ts +++ b/test/oauthRefreshHandler.node.test.ts @@ -10,7 +10,7 @@ import {getActiveMock} from './helpers/mockFetch' // assertions can only run where XHR is absent and uploads use the fetch path. // The XHR path has its own coverage in `browserUpload.browser.test.ts`. -const oauthClient = (setup: OAuthTokenSetup) => getClient({token: setup}) +const oauthClient = (setup: OAuthTokenSetup) => getClient({auth: {oauth: setup}}) function authHeaders(): Array { return getActiveMock() diff --git a/test/oauthRefreshHandler.test.ts b/test/oauthRefreshHandler.test.ts index 38aff676a..4e6d9f2ce 100644 --- a/test/oauthRefreshHandler.test.ts +++ b/test/oauthRefreshHandler.test.ts @@ -8,8 +8,8 @@ import {getActiveMock} from './helpers/mockFetch' const usersPath = '/v1/users/me' -// The DX under test: an OAuth setup handed straight to `token` on the real client. -const oauthClient = (setup: OAuthTokenSetup) => getClient({token: setup}) +// The DX under test: an OAuth setup configured as `auth.oauth` on the real client. +const oauthClient = (setup: OAuthTokenSetup) => getClient({auth: {oauth: setup}}) function authHeaders(): Array { return getActiveMock() @@ -17,7 +17,16 @@ function authHeaders(): Array { .map((request) => request.headers.get('authorization')) } -describe('OAuth auto-refresh (token as OAuthTokenSetup)', () => { +describe('OAuth auto-refresh (auth.oauth)', () => { + test('rejects a config with both a static token and an OAuth setup', () => { + expect(() => + getClient({ + token: 'static', + auth: {oauth: {getToken: async () => 'x', refresh: async () => 'x'}}, + }), + ).toThrow('`token` and `auth.oauth` are mutually exclusive') + }) + test('applies the token from getToken() to every request', async () => { getActiveMock() .scope(projectHost()) diff --git a/test/resumability.test.ts b/test/resumability.test.ts index 1926105e1..2b71aa2f8 100644 --- a/test/resumability.test.ts +++ b/test/resumability.test.ts @@ -7,7 +7,7 @@ import {getClient, projectHost} from './client/helpers' import {getActiveMock, streamBody, streamStall} from './helpers/mockFetch' const sse = (body: string) => ({status: 200, body, headers: {'Content-Type': 'text/event-stream'}}) -const oauthClient = (setup: OAuthTokenSetup) => getClient({token: setup}) +const oauthClient = (setup: OAuthTokenSetup) => getClient({auth: {oauth: setup}}) test('given a listener with a string token, when the server drops the connection, then the eventsource lib reconnects with Last-Event-ID and the same token', async () => { getActiveMock() @@ -104,7 +104,11 @@ test('given a resumable listener whose token has expired, when the reconnect is expect( await firstValueFrom( client - .listen('*', {}, {enableResume: true, events: ['welcome', 'welcomeback', 'reconnect', 'mutation']}) + .listen( + '*', + {}, + {enableResume: true, events: ['welcome', 'welcomeback', 'reconnect', 'mutation']}, + ) .pipe(take(6), toArray()), ), ).toEqual([ @@ -124,3 +128,35 @@ test('given a resumable listener whose token has expired, when the reconnect is ]) expect(requests.map((r) => r.headers.get('last-event-id'))).toEqual([null, 'evt-1', 'evt-1']) }) + +test('given a listener that reconnected after an OAuth refresh, when the fresh EventSource drops, then its own newer Last-Event-ID wins over the id it was seeded with', async () => { + const scope = getActiveMock().scope(projectHost()) + scope + .on('GET', '/v1/data/listen/foo') + .respond(sse(`retry: 1\n\n` + encode({event: 'mutation', id: 'evt-1', data: '{}'}))) + .respond({status: 401, body: 'Unauthorized'}) + // the fresh EventSource, seeded with evt-1; its body ends after evt-2 so + // the eventsource lib reconnects on its own + scope + .on('GET', '/v1/data/listen/foo', {headers: {'Last-Event-ID': 'evt-1'}}) + .respond(sse(`retry: 1\n\n` + encode({event: 'mutation', id: 'evt-2', data: '{}'}))) + scope + .on('GET', '/v1/data/listen/foo', {headers: {'Last-Event-ID': 'evt-2'}}) + .respond(sse(encode({event: 'mutation', id: 'evt-3', data: '{}'}))) + + let currentToken = 'expired' + const client = oauthClient({ + getToken: async () => currentToken, + refresh: async () => (currentToken = 'fresh'), + }) + + expect(await firstValueFrom(client.listen('*').pipe(take(3), toArray()))).toHaveLength(3) + + const requests = getActiveMock().getRequests() + expect(requests.map((r) => r.headers.get('last-event-id'))).toEqual([ + null, + 'evt-1', + 'evt-1', + 'evt-2', + ]) +}) From 0c4b1439b1a0a5c9e28da33ab87b83f2a8556250 Mon Sep 17 00:00:00 2001 From: "squiggler-app[bot]" <265501495+squiggler-app[bot]@users.noreply.github.com> Date: Mon, 21 Sep 2026 07:00:12 +0000 Subject: [PATCH 5/7] chore: update auto-generated changeset for PR #1323 --- .changeset/pr-1323.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.changeset/pr-1323.md b/.changeset/pr-1323.md index a9998b0e3..8c65191e2 100644 --- a/.changeset/pr-1323.md +++ b/.changeset/pr-1323.md @@ -3,4 +3,4 @@ '@sanity/client': minor --- -feat: accept an OAuth token setup as `token` for transparent refresh \ No newline at end of file +feat: configure transparent OAuth token refresh via auth.oauth \ No newline at end of file From 28fe7d00c118168ec3745dc7ff3f162e6a674b1d Mon Sep 17 00:00:00 2001 From: Josh Date: Mon, 21 Sep 2026 15:11:50 +0100 Subject: [PATCH 6/7] fix: accept auth.oauth as credentials for live drafts --- src/data/live.ts | 12 ++++++------ test/live.test.ts | 31 ++++++++++++++++++++++++++++++- 2 files changed, 36 insertions(+), 7 deletions(-) diff --git a/src/data/live.ts b/src/data/live.ts index e9f8635ab..947cd3b1a 100644 --- a/src/data/live.ts +++ b/src/data/live.ts @@ -74,9 +74,13 @@ export class LiveClient { `Please update your API version to use this feature.`, ) } - if (includeDrafts && !token && !withCredentials) { + // An OAuth token setup can't be baked into the headers — it is resolved + // per request inside the EventSource fetch, so reconnects pick up + // refreshed tokens. + const tokenSetup = includeDrafts ? getOAuthTokenSetup(config) : undefined + if (includeDrafts && !token && !tokenSetup && !withCredentials) { throw new Error( - `The live events API requires a token or withCredentials when 'includeDrafts: true'. Please update your client configuration. The token should have the lowest possible access role.`, + `The live events API requires a token, auth.oauth or withCredentials when 'includeDrafts: true'. Please update your client configuration. The token should have the lowest possible access role.`, ) } const path = _getDataUrl(this.#client, 'live/events') @@ -98,10 +102,6 @@ export class LiveClient { if (configHeaders) { Object.assign(eventSourceHeaders, configHeaders) } - // An OAuth token setup can't be baked into the headers — it is resolved - // per request inside the EventSource fetch, so reconnects pick up - // refreshed tokens. - const tokenSetup = includeDrafts ? getOAuthTokenSetup(config) : undefined const eventSourceWithCredentials = Boolean(includeDrafts && withCredentials) let transportCache = eventsCache.get(config.resolveFetch) diff --git a/test/live.test.ts b/test/live.test.ts index f0177699a..b327b6fe8 100644 --- a/test/live.test.ts +++ b/test/live.test.ts @@ -110,7 +110,7 @@ describe('.live.events()', () => { test('requires token when includeDrafts is true', () => { const client = createClient({projectId: 'abc123', dataset: 'prod', apiVersion: 'vX'}) expect(() => client.live.events({includeDrafts: true})).toThrowErrorMatchingInlineSnapshot( - `[Error: The live events API requires a token or withCredentials when 'includeDrafts: true'. Please update your client configuration. The token should have the lowest possible access role.]`, + `[Error: The live events API requires a token, auth.oauth or withCredentials when 'includeDrafts: true'. Please update your client configuration. The token should have the lowest possible access role.]`, ) }) test('allows apiVersion 2021-03-26 when includeDrafts is true', () => { @@ -256,6 +256,35 @@ describe('.live.events()', () => { expect(request.init?.credentials).toBe('include') }) + test('authenticates includeDrafts with an OAuth token setup', async () => { + expect.assertions(2) + + getActiveMock() + .scope('https://abc123.api.sanity.io') + .on('GET', '/vX/data/live/events/oauth-drafts') + .respond({ + status: 200, + body: encode({id: '123', event: 'welcome', data: '{}'}), + headers: {'Access-Control-Allow-Origin': '*', 'Content-Type': 'text/event-stream'}, + }) + + const client = createClient({ + projectId: 'abc123', + dataset: 'oauth-drafts', + useCdn: false, + apiVersion: 'X', + auth: {oauth: {getToken: async () => 'oauth-token'}}, + }) + + // `auth.oauth` is mutually exclusive with `token` and switches + // `withCredentials` off, so it has to count as credentials on its own. + await firstValueFrom(client.live.events({includeDrafts: true})) + + const [request] = getActiveMock().getRequests() + expect(request.query).toMatchObject({includeDrafts: 'true'}) + expect(request).toHaveHeader('authorization', 'Bearer oauth-token') + }) + test('does not send cookies when withCredentials is set but drafts are not requested', async () => { expect.assertions(1) From 3cbe9833cf8f8ac7a21a166217af23a5876d8a42 Mon Sep 17 00:00:00 2001 From: Josh Date: Mon, 21 Sep 2026 15:21:27 +0100 Subject: [PATCH 7/7] test: satisfy OAuthTokenSetup type in live drafts test --- test/live.test.ts | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/test/live.test.ts b/test/live.test.ts index b327b6fe8..0b79f2f4d 100644 --- a/test/live.test.ts +++ b/test/live.test.ts @@ -273,7 +273,12 @@ describe('.live.events()', () => { dataset: 'oauth-drafts', useCdn: false, apiVersion: 'X', - auth: {oauth: {getToken: async () => 'oauth-token'}}, + auth: { + oauth: { + getToken: async () => 'oauth-token', + refresh: () => Promise.reject(new Error('not needed')), + }, + }, }) // `auth.oauth` is mutually exclusive with `token` and switches