feat(listen)! enable resume by default

Author: bjoerge

src/data/listen.ts

@@ -12,8 +12,6 @@ import {
   type OpenEvent,
   type ReconnectEvent,
   type ResetEvent,
-  type ResumableListenEventNames,
-  type ResumableListenOptions,
   type WelcomeBackEvent,
   type WelcomeEvent,
 } from '../types'
@@ -37,7 +35,6 @@ const possibleOptions = [
   'includeAllVersions',
   'visibility',
   'effectFormat',
-  'enableResume',
   'tag',
 ]
 
@@ -56,10 +53,7 @@ const defaultOptions = {
  */
 export type MapListenEventNamesToListenEvents<
   R extends Record<string, Any> = Record<string, Any>,
-  Events extends (ResumableListenEventNames | ListenEventName)[] = (
-    | ResumableListenEventNames
-    | ListenEventName
-  )[],
+  Events extends ListenEventName[] = ListenEventName[],
 > = Events extends (infer E)[]
   ? E extends 'welcome'
     ? WelcomeEvent
@@ -87,13 +81,13 @@ export type MapListenEventNamesToListenEvents<
  */
 export type ListenEventFromOptions<
   R extends Record<string, Any> = Record<string, Any>,
-  Opts extends ListenOptions | ResumableListenOptions | undefined = undefined,
-> = Opts extends ListenOptions | ResumableListenOptions
-  ? Opts['events'] extends (ResumableListenEventNames | ListenEventName)[]
+  Opts extends ListenOptions | undefined = undefined,
+> = Opts extends ListenOptions
+  ? Opts['events'] extends ListenEventName[]
     ? MapListenEventNamesToListenEvents<R, Opts['events']>
     : // fall back to ListenEvent if opts events is present, but we can't infer the literal event names
       ListenEvent<R>
-  : MutationEvent<R>
+  : MutationEvent<R> | ResetEvent
 
 /**
  * Set up a listener that will be notified when mutations occur on documents matching the provided query/filter.
@@ -107,7 +101,7 @@ export function _listen<R extends Record<string, Any> = Record<string, Any>>(
   this: SanityClient | ObservableSanityClient,
   query: string,
   params?: ListenParams,
-): Observable<MutationEvent<R>>
+): Observable<MutationEvent<R> | ResetEvent>
 /**
  * Set up a listener that will be notified when mutations occur on documents matching the provided query/filter.
  *
@@ -118,7 +112,7 @@ export function _listen<R extends Record<string, Any> = Record<string, Any>>(
  */
 export function _listen<
   R extends Record<string, Any> = Record<string, Any>,
-  Opts extends ListenOptions | ResumableListenOptions = ListenOptions,
+  Opts extends ListenOptions = ListenOptions,
 >(
   this: SanityClient | ObservableSanityClient,
   query: string,
@@ -139,14 +133,16 @@ export function _listen<
   const tag = opts.tag && requestTagPrefix ? [requestTagPrefix, opts.tag].join('.') : opts.tag
   const options = {...defaults(opts, defaultOptions), tag}
   const listenOpts = pick(options, possibleOptions)
-  const qs = encodeQueryString({query, params, options: {tag, ...listenOpts}})
+  const qs = encodeQueryString({query, params, options: {tag, ...listenOpts, enableResume: true}})
 
   const uri = `${url}${_getDataUrl(this, 'listen', qs)}`
   if (uri.length > MAX_URL_LENGTH) {
     return throwError(() => new Error('Query too large for listener'))
   }
 
-  const listenFor = (options.events ? options.events : ['mutation']) satisfies Opts['events']
+  const listenFor = (
+    options.events ? options.events : ['mutation', 'reset']
+  ) satisfies Opts['events']
 
   const esOptions: EventSourceInit & {headers?: Record<string, string>} = {}
   if (withCredentials) {

src/types.ts

@@ -1176,20 +1176,16 @@ export type ListenEventName =
   | 'welcome'
   /** The listener has been disconnected, and a reconnect attempt is scheduled */
   | 'reconnect'
+  /** The listener has reconnected and successfully resumed from where it left off */
+  | 'welcomeback'
+  /** The listener can't be resumed or otherwise need to reset its local state */
+  | 'reset'
   /**
    * The listener connection has been established
    * note: it's usually a better option to use the 'welcome' event
    */
   | 'open'
 
-/** @public */
-export type ResumableListenEventNames =
-  | ListenEventName
-  /** The listener has reconnected and successfully resumed from where it left off */
-  | 'welcomeback'
-  /** The listener can't be resumed or otherwise need to reset its local state */
-  | 'reset'
-
 /** @public */
 export type ListenParams = {[key: string]: Any}
 
@@ -1240,8 +1236,7 @@ export interface ListenOptions {
   visibility?: 'transaction' | 'query'
 
   /**
-   * Array of event names to include in the observable. By default, only mutation events are included.
-   * Note: `welcomeback` and `reset` events requires `enableResume: true`
+   * Array of event names to include in the observable. By default, only mutation and reset events are included.
    * @defaultValue `['mutation']`
    */
   events?: ListenEventName[]
@@ -1263,32 +1258,6 @@ export interface ListenOptions {
    * @defaultValue `undefined`
    */
   tag?: string
-
-  /**
-   * If this is enabled, the client will normally resume events upon reconnect
-   * Note that you should also handle to `reset`-events and handle the case where the backend is unable to resume.
-   * @beta
-   * @defaultValue `false`
-   */
-  enableResume?: false
-}
-
-/** @public */
-export interface ResumableListenOptions extends Omit<ListenOptions, 'events' | 'enableResume'> {
-  /**
-   * If this is enabled, the client will normally resume events upon reconnect
-   * Note that you should also subscribe to `reset`-events and handle the case where the backend is unable to resume
-   * @beta
-   * @defaultValue `false`
-   */
-  enableResume: true
-
-  /**
-   * Array of event names to include in the observable. By default, only mutation events are included.
-   *
-   * @defaultValue `['mutation']`
-   */
-  events?: ResumableListenEventNames[]
 }
 
 /** @public */

test/client.test.ts

@@ -653,7 +653,9 @@ describe('client', async () => {
           ].join('\n')
 
           nock(`https://${apiHost}`)
-            .get('/v1/media-libraries/res-id/listen?query=foo.bar&includeResult=true')
+            .get(
+              '/v1/media-libraries/res-id/listen?query=foo.bar&includeResult=true&enableResume=true',
+            )
             .reply(200, response, {
               'cache-control': 'no-cache',
               'content-type': 'text/event-stream; charset=utf-8',
@@ -665,7 +667,7 @@ describe('client', async () => {
               'foo.bar',
             ),
           )
-          expect(evt.result).toEqual(doc)
+          expect(evt.type === 'mutation' && evt.result).toEqual(doc)
         },
       )
     })
@@ -3665,15 +3667,15 @@ describe('client', async () => {
       ].join('\n')
 
       nock(projectHost())
-        .get('/v1/data/listen/foo?query=foo.bar&includeResult=true')
+        .get('/v1/data/listen/foo?query=foo.bar&includeResult=true&enableResume=true')
         .reply(200, response, {
           'cache-control': 'no-cache',
           'content-type': 'text/event-stream; charset=utf-8',
           'transfer-encoding': 'chunked',
         })
 
       const evt = await firstValueFrom(getClient().listen('foo.bar'))
-      expect(evt.result).toEqual(doc)
+      expect(evt.type === 'mutation' && evt.result).toEqual(doc)
     })
 
     test('listeners connect to listen endpoint with request tag, emits events', async () => {
@@ -3695,7 +3697,7 @@ describe('client', async () => {
 
       nock(projectHost())
         .get(
-          '/v1/data/listen/foo?tag=sfcraft.checkins&query=*%5B_type%20%3D%3D%20%22checkin%22%5D&includeResult=true',
+          '/v1/data/listen/foo?tag=sfcraft.checkins&query=*%5B_type%20%3D%3D%20%22checkin%22%5D&includeResult=true&enableResume=true',
         )
         .reply(200, response, {
           'cache-control': 'no-cache',
@@ -3728,7 +3730,7 @@ describe('client', async () => {
 
       nock(projectHost())
         .get(
-          '/v1/data/listen/foo?tag=sf.craft.checkins&query=*%5B_type%20%3D%3D%20%22checkin%22%5D&includeResult=true',
+          '/v1/data/listen/foo?tag=sf.craft.checkins&query=*%5B_type%20%3D%3D%20%22checkin%22%5D&includeResult=true&enableResume=true',
         )
         .reply(200, response, {
           'cache-control': 'no-cache',
@@ -3762,7 +3764,7 @@ describe('client', async () => {
 
       let didRequest = false
       nock(projectHost())
-        .get('/v1/data/listen/foo?query=foo.bar&includeResult=true')
+        .get('/v1/data/listen/foo?query=foo.bar&includeResult=true&enableResume=true')
         .reply(() => {
           didRequest = true
           return [200, response]
@@ -3789,7 +3791,7 @@ describe('client', async () => {
 
       let requestCount = 0
       nock(projectHost())
-        .get('/v1/data/listen/foo?query=foo.bar&includeResult=true')
+        .get('/v1/data/listen/foo?query=foo.bar&includeResult=true&enableResume=true')
         .twice()
         .reply(() => {
           requestCount++

test/listen.test-d.ts

@@ -14,8 +14,8 @@ import {describe, expectTypeOf, test} from 'vitest'
 describe('client.listen', () => {
   const client = createClient({})
   test('event types', async () => {
-    // mutation event is default
-    expectTypeOf(client.listen('*')).toEqualTypeOf<Observable<MutationEvent>>()
+    // mutation and reset is default
+    expectTypeOf(client.listen('*')).toEqualTypeOf<Observable<MutationEvent | ResetEvent>>()
 
     // @ts-expect-error - WelcomeEvent should not be emitted
     expectTypeOf(client.listen('*')).toEqualTypeOf<Observable<WelcomeEvent>>()
@@ -34,14 +34,17 @@ describe('client.listen', () => {
 
     expectTypeOf(client.listen('*', {}, {events: []})).toEqualTypeOf<Observable<never>>()
 
-    //@ts-expect-error – welcomeback and reset requires `enableResume`
-    expectTypeOf(client.listen('*', {}, {events: ['welcomeback', 'reset']})).toEqualTypeOf<
-      Observable<ListenEvent>
+    expectTypeOf(client.listen('*', {}, {events: ['mutation', 'reset']})).toEqualTypeOf<
+      Observable<MutationEvent | ResetEvent>
     >()
 
     expectTypeOf(
-      client.listen('*', {}, {enableResume: true, events: ['welcome', 'mutation']}),
-    ).toEqualTypeOf<Observable<WelcomeEvent | MutationEvent>>()
+      client.listen(
+        '*',
+        {},
+        {enableResume: true, events: ['welcome', 'mutation', 'welcomeback', 'reset']},
+      ),
+    ).toEqualTypeOf<Observable<WelcomeEvent | MutationEvent | WelcomeBackEvent | ResetEvent>>()
 
     const observable = client.listen(
       '*',

test/listen.test.ts

@@ -52,6 +52,7 @@ describe.skipIf(typeof EdgeRuntime === 'string' || typeof document !== 'undefine
           query: '*[_type == "beer" && title == $beerName]',
           $beerName: '"Headroom Double IPA"',
           includeResult: 'true',
+          enableResume: 'true',
         })
         expect(request.url, 'url should be correct').toEqual(`/v1/data/listen/prod?${searchParams}`)