@sanity/client - npm Package Compare versions

Comparing version 6.17.3-canary.2 to 6.17.3-canary.3

package.json

 {
   "name": "@sanity/client",
-  "version": "6.17.3-canary.2",
+  "version": "6.17.3-canary.3",
   "description": "Client for retrieving, creating and patching data from Sanity.io",
@@ -5,0 +5,0 @@ "keywords": [

src/data/dataMethods.ts

@@ -70,7 +70,5 @@ import {from, type MonoTypeOperatorFunction, Observable} from 'rxjs'
   query: string,
-  __params: Q = {} as Q,
+  _params: Q = {} as Q,
   options: QueryOptions = {},
 ): Observable<RawQueryResponse<R> | R> {
-  const _params = options.livePos ? {...__params, _: options.livePos} : __params
-
   const stega =
@@ -263,3 +261,3 @@ 'stega' in options
   const returnFirst = options.returnFirst
-  const {timeout, token, tag, headers, returnQuery} = options
+  const {timeout, token, tag, headers, returnQuery, lastLiveEventId} = options
 
@@ -281,2 +279,3 @@ const uri = _getDataUrl(client, endpoint, stringQuery)
     resultSourceMap: options.resultSourceMap,
+    lastLiveEventId,
     canUseCdn: isQuery,
@@ -383,2 +382,6 @@ signal: options.signal,
 
+    if (options.lastLiveEventId) {
+      options.query = {...options.query, lastLiveEventId: options.lastLiveEventId}
+    }
+
     if (options.returnQuery === false) {
@@ -385,0 +388,0 @@ options.query = {returnQuery: 'false', ...options.query}

src/data/live.ts

 import {Observable} from 'rxjs'
 
 import type {ObservableSanityClient, SanityClient} from '../SanityClient'
-import type {LiveErrorEvent, LiveOptions, LiveRefreshEvent, LiveRestartEvent} from '../types'
+import type {Any, LiveEventError, LiveEventMessage, LiveEventRestart} from '../types'
 import {_getDataUrl} from './dataMethods'
 
-/** @public */
-export function _live(
-  this: SanityClient | ObservableSanityClient,
-  opts: LiveOptions = {},
-): Observable<LiveRefreshEvent | LiveErrorEvent | LiveRestartEvent> {
-  const path = _getDataUrl(this, 'sync-tags')
-  const url = new URL(this.getUrl(path, false))
-  if (opts.start) {
-    url.searchParams.append('start', opts.start)
+/**
+ * @internal
+ */
+export class LiveClient {
+  #client: SanityClient | ObservableSanityClient
+  constructor(client: SanityClient | ObservableSanityClient) {
+    this.#client = client
   }
 
-  return new Observable((observer) => {
-    const controller = new AbortController()
-    const {signal} = controller
-    fetch(url, {signal})
-      .then(async (res) => {
-        if (!res.body) {
-          throw new TypeError('Response body is not readable')
+  events(): Observable<LiveEventMessage | LiveEventError | LiveEventRestart> {
+    const path = _getDataUrl(this.#client, 'live/events')
+    const url = new URL(this.#client.getUrl(path, false))
+
+    const listenFor = ['restart', 'message']
+
+    return new Observable((observer) => {
+      let es: InstanceType<typeof import('@sanity/eventsource')>
+      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(evt: Any) {
+        if (stopped) {
+          return
         }
-        const reader = res.body
-          .pipeThrough(new TextDecoderStream())
-          .pipeThrough(splitStream('\n'))
-          .pipeThrough(parseJSON())
-          .getReader()
 
-        const results = {
-          [Symbol.asyncIterator]() {
-            return {
-              next: () => reader.read(),
-            }
-          },
+        observer.error(cooerceError(evt))
+
+        // Unless we've explicitly stopped the ES (in which case `stopped` should be true),
+        // we should never be in a disconnected state. By default, EventSource will reconnect
+        // automatically, in which case it sets readyState to `CONNECTING`, but in some cases
+        // (like when a laptop lid is closed), it closes the connection. In these cases we need
+        // to explicitly reconnect.
+        if (es.readyState === es.CLOSED) {
+          unsubscribe()
         }
+      }
 
-        for await (const chunk of results) {
-          if (signal.aborted) break
-          if (chunk.type === 'error') {
-            observer.error(chunk)
-            break
-          }
-          observer.next(chunk)
+      function onMessage(evt: Any) {
+        const event = parseEvent(evt)
+        return event instanceof Error ? observer.error(event) : observer.next(event)
+      }
+
+      function unsubscribe() {
+        if (!es) return
+        es.removeEventListener('error', onError)
+        listenFor.forEach((type: string) => es.removeEventListener(type, onMessage))
+        es.close()
+      }
+
+      async function getEventSource() {
+        const EventSourceImplementation: typeof import('@sanity/eventsource') =
+          typeof EventSource === 'undefined'
+            ? (await import('@sanity/eventsource')).default
+            : (globalThis.EventSource as unknown as Any)
+
+        // If the listener has been unsubscribed from before we managed to load the module,
+        // do not set up the EventSource.
+        if (unsubscribed) {
+          return
         }
-      })
-      .catch((err) => {
-        if (err?.name !== 'TimeoutError' && err?.name !== 'AbortError') {
-          observer.error(err)
-        }
-      })
-    return () => {
-      controller.abort()
-    }
-  })
+
+        const evs = new EventSourceImplementation(url.toString())
+        evs.addEventListener('error', onError)
+        listenFor.forEach((type: string) => evs.addEventListener(type, onMessage))
+        return evs
+      }
+
+      function open() {
+        getEventSource()
+          .then((eventSource) => {
+            if (eventSource) {
+              es = eventSource
+            }
+          })
+          .catch((reason) => {
+            observer.error(reason)
+            stop()
+          })
+      }
+
+      function stop() {
+        stopped = true
+        unsubscribe()
+        unsubscribed = true
+      }
+
+      return stop
+    })
+  }
 }
 
-function splitStream(splitOn: string) {
-  let buffer = ''
+function parseEvent(event: Any) {
+  try {
+    const data = (event.data && JSON.parse(event.data)) || {}
+    return {type: event.type, ...data}
+  } catch (err) {
+    return err
+  }
+}
 
-  return new TransformStream({
-    transform(chunk, controller) {
-      buffer += chunk
-      const parts = buffer.split(splitOn)
-      parts.slice(0, -1).forEach((part) => controller.enqueue(part))
-      buffer = parts[parts.length - 1]
-    },
-    flush(controller) {
-      if (buffer) controller.enqueue(buffer)
-    },
-  })
+function cooerceError(err: Any) {
+  if (err instanceof Error) {
+    return err
+  }
+
+  const evt = parseEvent(err)
+  return evt instanceof Error ? evt : new Error(extractErrorMessage(evt))
 }
 
-function parseJSON() {
-  return new TransformStream({
-    transform(chunk, controller) {
-      if (!chunk) return
-      controller.enqueue(JSON.parse(chunk))
-    },
-  })
+function extractErrorMessage(err: Any) {
+  if (!err.error) {
+    return err.message || 'Unknown listener error'
+  }
+
+  if (err.error.description) {
+    return err.error.description
+  }
+
+  return typeof err.error === 'string' ? err.error : JSON.stringify(err.error, null, 2)
 }

src/SanityClient.ts

@@ -7,3 +7,3 @@ import {lastValueFrom, Observable} from 'rxjs'
 import {_listen} from './data/listen'
-import {_live} from './data/live'
+import {LiveClient} from './data/live'
 import {ObservablePatch, Patch} from './data/patch'
@@ -46,5 +46,5 @@ import {ObservableTransaction, Transaction} from './data/transaction'
   _listen,
-  _live,
   AssetsClient,
   DatasetsClient,
+  LiveClient,
   ObservableAssetsClient,
@@ -62,2 +62,3 @@ ObservableDatasetsClient,
   datasets: ObservableDatasetsClient
+  live: LiveClient
   projects: ObservableProjectsClient
@@ -76,3 +77,2 @@ users: ObservableUsersClient
   listen = _listen
-  live = _live
 
@@ -86,2 +86,3 @@ constructor(httpRequest: HttpRequest, config: ClientConfig = defaultConfig) {
     this.datasets = new ObservableDatasetsClient(this, this.#httpRequest)
+    this.live = new LiveClient(this)
     this.projects = new ObservableProjectsClient(this, this.#httpRequest)
@@ -706,2 +707,3 @@ this.users = new ObservableUsersClient(this, this.#httpRequest)
   datasets: DatasetsClient
+  live: LiveClient
   projects: ProjectsClient
@@ -725,3 +727,2 @@ users: UsersClient
   listen = _listen
-  live = _live
 
@@ -735,2 +736,3 @@ constructor(httpRequest: HttpRequest, config: ClientConfig = defaultConfig) {
     this.datasets = new DatasetsClient(this, this.#httpRequest)
+    this.live = new LiveClient(this)
     this.projects = new ProjectsClient(this, this.#httpRequest)
@@ -737,0 +739,0 @@ this.users = new UsersClient(this, this.#httpRequest)

src/types.ts

@@ -309,2 +309,3 @@ // deno-lint-ignore-file no-empty-interface
   perspective?: ClientPerspective
+  lastLiveEventId?: string
 }
@@ -484,3 +485,3 @@
   /** @deprecated you're using a fetch option as a GROQ parameter, this is likely a mistake */
-  livePos?: never
+  lastLiveEventId?: never
   /* eslint-enable @typescript-eslint/no-explicit-any */
@@ -750,3 +751,3 @@ }
   next?: ('next' extends keyof RequestInit ? RequestInit : never)['next']
-  livePos?: string
+  lastLiveEventId?: string
 }
@@ -1010,7 +1011,5 @@
 /** @public */
-export interface LiveOptions {
-  start?: string
-}
+export type SyncTag = `s1:${string}`
 /** @public */
-export interface LiveErrorEvent {
+export interface LiveEventError {
   type: 'error'
@@ -1021,9 +1020,10 @@ status: number
 /** @public */
-export interface LiveRestartEvent {
+export interface LiveEventRestart {
   type: 'restart'
 }
 /** @public */
-export interface LiveRefreshEvent {
-  tags: `s1:${string}`[]
-  pos: string
+export interface LiveEventMessage {
+  type: 'message'
+  id: string
+  tags: SyncTag[]
 }
@@ -1030,0 +1030,0 @@

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

2180855

increased by0.87%

Lines of code

24414

increased by0.51%

Number of medium supply chain risk alerts

5

decreased by-54.55%

No dependency changes