fix: prevent listener leak on unsubscribe before eventsource module l…

…oad (#783)

src/data/listen.ts

@@ -85,16 +85,13 @@ export function _listen<R extends Record<string, Any> = Record<string, Any>>(
 
   return new Observable((observer) => {
     let es: InstanceType<typeof import('@sanity/eventsource')>
-    getEventSource()
-      .then((eventSource) => {
-        es = eventSource
-      })
-      .catch((reason) => {
-        observer.error(reason)
-        stop()
-      })
     let reconnectTimer: NodeJS.Timeout
     let stopped = false
+    // Unsubscribe differs from stopped in that we will never reopen.
+    // Once it is`true`, it will never be `false` again.
+    let unsubscribed = false
+
+    open()
 
     function onError() {
       if (stopped) {
@@ -150,8 +147,17 @@ export function _listen<R extends Record<string, Any> = Record<string, Any>>(
       }
     }
 
-    async function getEventSource(): Promise<InstanceType<typeof import('@sanity/eventsource')>> {
+    async function getEventSource(): Promise<InstanceType<
+      typeof import('@sanity/eventsource')
+    > | void> {
       const {default: EventSource} = await import('@sanity/eventsource')
+
+      // If the listener has been unsubscribed from before we managed to load the module,
+      // do not set up the EventSource.
+      if (unsubscribed) {
+        return
+      }
+
       const evs = new EventSource(uri, esOptions)
       evs.addEventListener('error', onError)
       evs.addEventListener('channelError', onChannelError)
@@ -163,7 +169,9 @@ export function _listen<R extends Record<string, Any> = Record<string, Any>>(
     function open() {
       getEventSource()
         .then((eventSource) => {
-          es = eventSource
+          if (eventSource) {
+            es = eventSource
+          }
         })
         .catch((reason) => {
           observer.error(reason)
@@ -174,6 +182,7 @@ export function _listen<R extends Record<string, Any> = Record<string, Any>>(
     function stop() {
       stopped = true
       unsubscribe()
+      unsubscribed = true
     }
 
     return stop

test/listen.test.ts

@@ -1,7 +1,7 @@
 import type {AddressInfo} from 'node:net'
 
 import {type ClientConfig, createClient} from '@sanity/client'
-import {describe, expect, test} from 'vitest'
+import {describe, expect, test, vitest} from 'vitest'
 
 import {createSseServer, type OnRequest} from './helpers/sseServer'
 
@@ -212,5 +212,34 @@ describe.skipIf(typeof EdgeRuntime === 'string' || typeof document !== 'undefine
         server.close()
       }
     })
+
+    test('can immediately unsubscribe, does not connect to server', async () => {
+      const onMessage = vitest.fn()
+      const onError = vitest.fn()
+      const onRequest = vitest.fn(({channel}) => {
+        channel!.send({event: 'mutation', data: {foo: 'bar'}})
+        process.nextTick(() => channel!.close())
+      })
+
+      const {server, client} = await testSse(onRequest)
+
+      const query = '*[_type == "beer" && title == $beerName]'
+      const params = {beerName: 'Headroom Double IPA'}
+
+      client
+        .listen(query, params)
+        .subscribe({
+          next: onMessage,
+          error: onError,
+        })
+        .unsubscribe()
+
+      await new Promise((resolve) => setTimeout(resolve, 100))
+
+      expect(onMessage).not.toHaveBeenCalled()
+      expect(onError).not.toHaveBeenCalled()
+      expect(onRequest).not.toHaveBeenCalled()
+      server.close()
+    })
   },
 )