feat: configure transparent OAuth token refresh via auth.oauth

.changeset/pr-1323.md

@@ -0,0 +1,6 @@
+<!-- auto-generated -->
+---
+'@sanity/client': minor
+---
+
+feat: configure transparent OAuth token refresh via auth.oauth

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,17 +263,25 @@ 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<T>({
+      // 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)
+      const reqHeaders = oauth ? await applyOAuthToken(oauth, req.headers) : req.headers
+      const upload = uploadWithProgress<T>({
         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
         // get-it timeout collapses to its `total` component here.
         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())
   }
 

src/config.ts

@@ -121,7 +121,10 @@ 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
   }

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<R>(
  */
 export function _request<R>(client: Client, httpRequest: HttpRequest, options: Any): Promise<R> {
   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<T>(
   options: RequestObservableOptions,
 ): Observable<UploadEvent<T>> {
   const reqOptions = _prepareRequest(client, options)
-  const requester = client.config().requester
-  const request = new Observable<Any>((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)
+  const upload = (req: FetchRequest) =>
+    new Observable<Any>((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<T> =>

src/data/listen.ts

@@ -1,7 +1,8 @@
 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'
 import {
   type Any,
@@ -172,17 +173,26 @@ export function _connectListenEventSource<TEvent extends {type: string}>(
   if (configHeaders) {
     Object.assign(headers, configHeaders)
   }
+  const tokenSetup = getOAuthTokenSetup(config)
+  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(
-    reconnectOnConnectionFailure(),
+    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) => ({
       type: event.type,

src/data/live.ts

@@ -1,9 +1,10 @@
 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'
 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'
@@ -72,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')
@@ -108,17 +114,25 @@ 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)
 
     if (existing) {
       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,
         }),
       })
@@ -140,7 +154,12 @@ export class LiveClient {
 
     const observable = events
       .pipe(
-        reconnectOnConnectionFailure(),
+        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') {
             // Check for CORS on reconnect events (which happen on 403s)
@@ -289,3 +308,18 @@ const eventsCache = new Map<
   InitializedClientConfig['resolveFetch'],
   Map<string, Observable<LiveEvent>>
 >()
+
+/**
+ * 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<OAuthTokenSetup, number>()
+function tokenSetupCacheKey(setup: OAuthTokenSetup): number {
+  let id = tokenSetupIds.get(setup)
+  if (id === undefined) {
+    id = nextTokenSetupId++
+    tokenSetupIds.set(setup, id)
+  }
+  return id
+}

src/data/reconnectOnConnectionFailure.ts

@@ -1,6 +1,7 @@
 import {
   catchError,
   concat,
+  defer,
   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<T>(): OperatorFunction<T, T | {type: 'reconnect'}> {
+export function reconnectOnConnectionFailure<T>(
+  refreshAuth?: () => Promise<unknown>,
+): OperatorFunction<T, T | {type: 'reconnect'}> {
   return function (source: Observable<T>) {
+    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<T>(): OperatorFunction<T, T | {type
         ) {
           return concat(of({type: 'reconnect' as const}), timer(1000).pipe(mergeMap(() => 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}),
+            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
+              // errors from the resubscribed stream.
+              catchError(() => throwError(() => err)),
+              mergeMap(() => caught),
+            ),
+          )
+        }
         return throwError(() => err)
       }),
     )

src/data/resolveEventSourceFetch.ts

@@ -1,7 +1,8 @@
 import type {EventSourceFetchInit, FetchLikeResponse} from 'eventsource'
 import type {FetchFunction, FetchInit} from 'get-it'
 
-import type {InitializedClientConfig} from '../types'
+import {resolveOAuthToken} from '../http/oauthRefreshHandler'
+import type {InitializedClientConfig, OAuthTokenSetup} from '../types'
 
 /** @internal */
 export interface EventSourceFetchOptions {
@@ -11,6 +12,23 @@ export interface EventSourceFetchOptions {
    * etc. — things the native EventSource API has no equivalent for.
    */
   headers?: Record<string, string>
+  /**
+   * 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
+   * `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
   /**
    * If the client was configured with `withCredentials: true`, the
    * resolved fetch forwards `credentials: 'include'` so the browser
@@ -50,19 +68,29 @@ 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
 
-  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 || lastEventId) {
       const headers = new Headers(init?.headers)
-      for (const [key, value] of Object.entries(extraHeaders)) {
-        headers.set(key, value)
+      if (lastEventId && !headers.has('last-event-id')) {
+        headers.set('Last-Event-ID', lastEventId)
+      }
+      if (tokenSetup) {
+        headers.set('Authorization', `Bearer ${await resolveOAuthToken(tokenSetup)}`)
+      }
+      if (extraHeaders) {
+        for (const [key, value] of Object.entries(extraHeaders)) {
+          headers.set(key, value)
+        }
       }
       mergedInit.headers = headers
     }
@@ -71,7 +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.
-    return baseFetch(typeof url === 'string' ? url : url.href, mergedInit)
+    return await baseFetch(typeof url === 'string' ? url : url.href, mergedInit)
   }
 }