From ecd2f693cccf8ff57b2e00bb1f0961b39e872d0a Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Bj=C3=B8rge=20N=C3=A6ss?= <bjoerge@gmail.com>
Date: Tue, 16 Jul 2024 15:30:20 +0200
Subject: [PATCH 1/6] chore: remove futile circular dependency workaround

---
 .../src/core/preview/documentPreviewStore.ts     | 12 +-----------
 .../sanity/src/core/preview/observeFields.ts     | 16 +++-------------
 2 files changed, 4 insertions(+), 24 deletions(-)

diff --git a/packages/sanity/src/core/preview/documentPreviewStore.ts b/packages/sanity/src/core/preview/documentPreviewStore.ts
index 50249244f75..14ffb1ac640 100644
--- a/packages/sanity/src/core/preview/documentPreviewStore.ts
+++ b/packages/sanity/src/core/preview/documentPreviewStore.ts
@@ -63,17 +63,7 @@ export function createDocumentPreviewStore({
 }: DocumentPreviewStoreOptions): DocumentPreviewStore {
   const versionedClient = client.withConfig({apiVersion: '1'})
 
-  // NOTE: this is workaroudn for circumventing a circular dependency between `observePaths` and
-  // `observeFields`.
-  // eslint-disable-next-line camelcase
-  const __proxy_observePaths: ObservePathsFn = (value, paths, apiConfig) => {
-    return observePaths(value, paths, apiConfig)
-  }
-
-  const {observeFields} = create_preview_observeFields({
-    observePaths: __proxy_observePaths,
-    versionedClient,
-  })
+  const {observeFields} = create_preview_observeFields({versionedClient})
 
   const {observePaths} = createPathObserver({observeFields})
 
diff --git a/packages/sanity/src/core/preview/observeFields.ts b/packages/sanity/src/core/preview/observeFields.ts
index b86a038da20..7475dc2144e 100644
--- a/packages/sanity/src/core/preview/observeFields.ts
+++ b/packages/sanity/src/core/preview/observeFields.ts
@@ -25,14 +25,7 @@ import {
 } from 'rxjs/operators'
 
 import {INCLUDE_FIELDS} from './constants'
-import {
-  type ApiConfig,
-  type FieldName,
-  type Id,
-  type ObservePathsFn,
-  type PreviewPath,
-  type Selection,
-} from './types'
+import {type ApiConfig, type FieldName, type Id, type PreviewPath, type Selection} from './types'
 import {debounceCollect} from './utils/debounceCollect'
 import {hasEqualFields} from './utils/hasEqualFields'
 import {isUniqueBy} from './utils/isUniqueBy'
@@ -48,11 +41,8 @@ type Cache = {
   [id: string]: CachedFieldObserver[]
 }
 
-export function create_preview_observeFields(context: {
-  observePaths: ObservePathsFn
-  versionedClient: SanityClient
-}) {
-  const {observePaths, versionedClient} = context
+export function create_preview_observeFields(context: {versionedClient: SanityClient}) {
+  const {versionedClient} = context
 
   let _globalListener: any
 

From 6b5805b7df3ebbac274edb65208f4880fe63422c Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Bj=C3=B8rge=20N=C3=A6ss?= <bjoerge@gmail.com>
Date: Tue, 16 Jul 2024 15:44:01 +0200
Subject: [PATCH 2/6] refactor(core): lift global listener out of preview store

---
 .../src/core/preview/createGlobalListener.ts  | 37 ++++++++++++
 .../src/core/preview/documentPreviewStore.ts  |  8 +--
 .../sanity/src/core/preview/observeFields.ts  | 60 +++----------------
 3 files changed, 50 insertions(+), 55 deletions(-)
 create mode 100644 packages/sanity/src/core/preview/createGlobalListener.ts

diff --git a/packages/sanity/src/core/preview/createGlobalListener.ts b/packages/sanity/src/core/preview/createGlobalListener.ts
new file mode 100644
index 00000000000..15f64f51ef3
--- /dev/null
+++ b/packages/sanity/src/core/preview/createGlobalListener.ts
@@ -0,0 +1,37 @@
+import {type MutationEvent, type SanityClient, type WelcomeEvent} from '@sanity/client'
+import {defer, merge, timer} from 'rxjs'
+import {filter, share, shareReplay} from 'rxjs/operators'
+
+/**
+ * @internal
+ * Creates a listener that will emit 'welcome' for all new subscribers immediately, and thereafter emit at every mutation event
+ */
+export function createGlobalListener(client: SanityClient) {
+  return defer(() => {
+    const allEvents$ = client
+      .listen(
+        '*[!(_id in path("_.**"))]',
+        {},
+        {
+          events: ['welcome', 'mutation'],
+          includeResult: false,
+          visibility: 'query',
+          tag: 'preview.global',
+        },
+      )
+      .pipe(
+        filter(
+          (event): event is WelcomeEvent | MutationEvent =>
+            event.type === 'welcome' || event.type === 'mutation',
+        ),
+        share({resetOnRefCountZero: () => timer(2000)}),
+      )
+
+    const welcome$ = allEvents$.pipe(
+      filter((event) => event.type === 'welcome'),
+      shareReplay(1),
+    )
+    const mutations$ = allEvents$.pipe(filter((event) => event.type === 'mutation'))
+    return merge(welcome$, mutations$)
+  })
+}
diff --git a/packages/sanity/src/core/preview/documentPreviewStore.ts b/packages/sanity/src/core/preview/documentPreviewStore.ts
index 14ffb1ac640..7fd7b387aa5 100644
--- a/packages/sanity/src/core/preview/documentPreviewStore.ts
+++ b/packages/sanity/src/core/preview/documentPreviewStore.ts
@@ -5,10 +5,11 @@ import {distinctUntilChanged, map} from 'rxjs/operators'
 
 import {isRecord} from '../util'
 import {create_preview_availability} from './availability'
+import {createGlobalListener} from './createGlobalListener'
 import {createPathObserver} from './createPathObserver'
 import {createPreviewObserver} from './createPreviewObserver'
 import {create_preview_documentPair} from './documentPair'
-import {create_preview_observeFields} from './observeFields'
+import {createObserveFields} from './observeFields'
 import {
   type ApiConfig,
   type DraftsModelDocument,
@@ -62,9 +63,8 @@ export function createDocumentPreviewStore({
   client,
 }: DocumentPreviewStoreOptions): DocumentPreviewStore {
   const versionedClient = client.withConfig({apiVersion: '1'})
-
-  const {observeFields} = create_preview_observeFields({versionedClient})
-
+  const globalListener = createGlobalListener(versionedClient)
+  const {observeFields} = createObserveFields({versionedClient, globalListener})
   const {observePaths} = createPathObserver({observeFields})
 
   function observeDocumentTypeFromId(
diff --git a/packages/sanity/src/core/preview/observeFields.ts b/packages/sanity/src/core/preview/observeFields.ts
index 7475dc2144e..29bd0d776a2 100644
--- a/packages/sanity/src/core/preview/observeFields.ts
+++ b/packages/sanity/src/core/preview/observeFields.ts
@@ -1,11 +1,10 @@
-import {type SanityClient} from '@sanity/client'
+import {type MutationEvent, type SanityClient, type WelcomeEvent} from '@sanity/client'
 import {difference, flatten, memoize} from 'lodash'
 import {
   combineLatest,
   concat,
   defer,
   EMPTY,
-  from,
   fromEvent,
   merge,
   type Observable,
@@ -41,55 +40,14 @@ type Cache = {
   [id: string]: CachedFieldObserver[]
 }
 
-export function create_preview_observeFields(context: {versionedClient: SanityClient}) {
-  const {versionedClient} = context
-
-  let _globalListener: any
-
-  const getGlobalEvents = () => {
-    if (!_globalListener) {
-      const allEvents$ = from(
-        versionedClient.listen(
-          '*[!(_id in path("_.**"))]',
-          {},
-          {
-            events: ['welcome', 'mutation'],
-            includeResult: false,
-            visibility: 'query',
-            tag: 'preview.global',
-          },
-        ),
-      ).pipe(share())
-
-      // This is a stream of welcome events from the server, each telling us that we have established listener connection
-      // We map these to snapshot fetch/sync. It is good to wait for the first welcome event before fetching any snapshots as, we may miss
-      // events that happens in the time period after initial fetch and before the listener is established.
-      const welcome$ = allEvents$.pipe(
-        filter((event: any) => event.type === 'welcome'),
-        shareReplay({refCount: true, bufferSize: 1}),
-      )
-
-      // This will keep the listener active forever and in turn reduce the number of initial fetches
-      // as less 'welcome' events will be emitted.
-      // @todo: see if we can delay unsubscribing or connect with some globally defined shared listener
-      welcome$.subscribe()
-
-      const mutations$ = allEvents$.pipe(filter((event: any) => event.type === 'mutation'))
-
-      _globalListener = {
-        welcome$,
-        mutations$,
-      }
-    }
-
-    return _globalListener
-  }
-
+export function createObserveFields(context: {
+  versionedClient: SanityClient
+  globalListener: Observable<WelcomeEvent | MutationEvent>
+}) {
+  const {versionedClient, globalListener} = context
   function listen(id: Id) {
-    const globalEvents = getGlobalEvents()
-    return merge(
-      globalEvents.welcome$,
-      globalEvents.mutations$.pipe(filter((event: any) => event.documentId === id)),
+    return globalListener.pipe(
+      filter((event) => event.type === 'welcome' || event.documentId === id),
     )
   }
 
@@ -107,7 +65,7 @@ export function create_preview_observeFields(context: {versionedClient: SanityCl
 
   function currentDatasetListenFields(id: Id, fields: PreviewPath[]) {
     return listen(id).pipe(
-      switchMap((event: any) => {
+      switchMap((event) => {
         if (event.type === 'welcome' || event.visibility === 'query') {
           return fetchDocumentPathsFast(id, fields as any).pipe(
             mergeMap((result) => {

From 038d1c41c6c7aa189db9eb6dfc09607e1c1efeb4 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Bj=C3=B8rge=20N=C3=A6ss?= <bjoerge@gmail.com>
Date: Tue, 16 Jul 2024 16:24:42 +0200
Subject: [PATCH 3/6] fix(preview): cleanup code so it matches current
 conventions + add some docs

---
 .../sanity/src/core/preview/availability.ts   |  2 +-
 .../src/core/preview/createGlobalListener.ts  | 54 +++++++++----------
 .../src/core/preview/createObserveDocument.ts | 41 ++++++++++++++
 .../src/core/preview/createPathObserver.ts    |  9 +++-
 .../sanity/src/core/preview/documentPair.ts   |  4 +-
 .../src/core/preview/documentPreviewStore.ts  | 18 +++++--
 .../sanity/src/core/preview/observeFields.ts  | 49 ++++++++++-------
 packages/sanity/src/core/preview/types.ts     | 11 ++++
 8 files changed, 134 insertions(+), 54 deletions(-)
 create mode 100644 packages/sanity/src/core/preview/createObserveDocument.ts

diff --git a/packages/sanity/src/core/preview/availability.ts b/packages/sanity/src/core/preview/availability.ts
index e77537678e8..d339ff3d7aa 100644
--- a/packages/sanity/src/core/preview/availability.ts
+++ b/packages/sanity/src/core/preview/availability.ts
@@ -67,7 +67,7 @@ function mutConcat<T>(array: T[], chunks: T[]) {
   return array
 }
 
-export function create_preview_availability(
+export function createPreviewAvailabilityObserver(
   versionedClient: SanityClient,
   observePaths: ObservePathsFn,
 ): {
diff --git a/packages/sanity/src/core/preview/createGlobalListener.ts b/packages/sanity/src/core/preview/createGlobalListener.ts
index 15f64f51ef3..42ee47127b3 100644
--- a/packages/sanity/src/core/preview/createGlobalListener.ts
+++ b/packages/sanity/src/core/preview/createGlobalListener.ts
@@ -1,5 +1,5 @@
 import {type MutationEvent, type SanityClient, type WelcomeEvent} from '@sanity/client'
-import {defer, merge, timer} from 'rxjs'
+import {merge, timer} from 'rxjs'
 import {filter, share, shareReplay} from 'rxjs/operators'
 
 /**
@@ -7,31 +7,31 @@ import {filter, share, shareReplay} from 'rxjs/operators'
  * Creates a listener that will emit 'welcome' for all new subscribers immediately, and thereafter emit at every mutation event
  */
 export function createGlobalListener(client: SanityClient) {
-  return defer(() => {
-    const allEvents$ = client
-      .listen(
-        '*[!(_id in path("_.**"))]',
-        {},
-        {
-          events: ['welcome', 'mutation'],
-          includeResult: false,
-          visibility: 'query',
-          tag: 'preview.global',
-        },
-      )
-      .pipe(
-        filter(
-          (event): event is WelcomeEvent | MutationEvent =>
-            event.type === 'welcome' || event.type === 'mutation',
-        ),
-        share({resetOnRefCountZero: () => timer(2000)}),
-      )
-
-    const welcome$ = allEvents$.pipe(
-      filter((event) => event.type === 'welcome'),
-      shareReplay(1),
+  const allEvents$ = client
+    .listen(
+      '*[!(_id in path("_.**"))]',
+      {},
+      {
+        events: ['welcome', 'mutation'],
+        includeResult: false,
+        includePreviousRevision: false,
+        visibility: 'transaction',
+        effectFormat: 'mendoza',
+        tag: 'preview.global',
+      },
+    )
+    .pipe(
+      filter(
+        (event): event is WelcomeEvent | MutationEvent =>
+          event.type === 'welcome' || event.type === 'mutation',
+      ),
+      share({resetOnRefCountZero: () => timer(2000)}),
     )
-    const mutations$ = allEvents$.pipe(filter((event) => event.type === 'mutation'))
-    return merge(welcome$, mutations$)
-  })
+
+  const welcome$ = allEvents$.pipe(
+    filter((event) => event.type === 'welcome'),
+    shareReplay(1),
+  )
+  const mutations$ = allEvents$.pipe(filter((event) => event.type === 'mutation'))
+  return merge(welcome$, mutations$)
 }
diff --git a/packages/sanity/src/core/preview/createObserveDocument.ts b/packages/sanity/src/core/preview/createObserveDocument.ts
new file mode 100644
index 00000000000..e0649354595
--- /dev/null
+++ b/packages/sanity/src/core/preview/createObserveDocument.ts
@@ -0,0 +1,41 @@
+import {type MutationEvent, type SanityClient, type WelcomeEvent} from '@sanity/client'
+import {uniq} from 'lodash'
+import {EMPTY, type Observable, of} from 'rxjs'
+import {concatMap, map, scan} from 'rxjs/operators'
+
+import {debounceCollect} from './utils/debounceCollect'
+
+export function createObserveDocument({
+  mutationChannel,
+  client,
+}: {
+  client: SanityClient
+  mutationChannel: Observable<WelcomeEvent | MutationEvent>
+}) {
+  function batchFetchDocuments(ids: string[][]) {
+    return client.observable
+      .fetch(`*[_id in $ids]`, {ids: uniq(ids.flat())}, {tag: 'preview.observe-document'})
+      .pipe(map((result) => ids.map((id) => result.find(id))))
+  }
+
+  const fetchDocument = debounceCollect(batchFetchDocuments, 10)
+
+  return function observeDocument(id: string) {
+    return mutationChannel.pipe(
+      concatMap((event) => {
+        if (event.type === 'welcome') {
+          return fetchDocument(id).pipe(map((document) => ({type: 'sync', document})))
+        }
+        return event.documentId === id ? of(event) : EMPTY
+      }),
+      scan((current, event) => {
+        if (event.type === 'sync') {
+          return event.document
+        }
+        if (event.type === 'mutation') {
+          return current
+        }
+      }, undefined),
+    )
+  }
+}
diff --git a/packages/sanity/src/core/preview/createPathObserver.ts b/packages/sanity/src/core/preview/createPathObserver.ts
index 773a0338434..0c006b9af9f 100644
--- a/packages/sanity/src/core/preview/createPathObserver.ts
+++ b/packages/sanity/src/core/preview/createPathObserver.ts
@@ -119,8 +119,13 @@ function normalizePaths(path: (FieldName | PreviewPath)[]): PreviewPath[] {
   )
 }
 
-export function createPathObserver(context: {observeFields: ObserveFieldsFn}) {
-  const {observeFields} = context
+/**
+ * Creates a function that allows observing nested paths on a document.
+ * If the path includes a reference, the reference will be "followed", allowing for selecting paths within the referenced document.
+ * @param options - Options - Requires a function that can observe fields on a document
+ * */
+export function createPathObserver(options: {observeFields: ObserveFieldsFn}) {
+  const {observeFields} = options
 
   return {
     observePaths(
diff --git a/packages/sanity/src/core/preview/documentPair.ts b/packages/sanity/src/core/preview/documentPair.ts
index d797a147141..6c8e682258c 100644
--- a/packages/sanity/src/core/preview/documentPair.ts
+++ b/packages/sanity/src/core/preview/documentPair.ts
@@ -4,7 +4,7 @@ import {combineLatest, type Observable, of} from 'rxjs'
 import {map, switchMap} from 'rxjs/operators'
 
 import {getIdPair, isRecord} from '../util'
-import {create_preview_availability} from './availability'
+import {createPreviewAvailabilityObserver} from './availability'
 import {type DraftsModelDocument, type ObservePathsFn, type PreviewPath} from './types'
 
 export function create_preview_documentPair(
@@ -16,7 +16,7 @@ export function create_preview_documentPair(
     paths: PreviewPath[],
   ) => Observable<DraftsModelDocument<T>>
 } {
-  const {observeDocumentPairAvailability} = create_preview_availability(
+  const {observeDocumentPairAvailability} = createPreviewAvailabilityObserver(
     versionedClient,
     observePaths,
   )
diff --git a/packages/sanity/src/core/preview/documentPreviewStore.ts b/packages/sanity/src/core/preview/documentPreviewStore.ts
index 7fd7b387aa5..77dbfa84c35 100644
--- a/packages/sanity/src/core/preview/documentPreviewStore.ts
+++ b/packages/sanity/src/core/preview/documentPreviewStore.ts
@@ -4,7 +4,7 @@ import {type Observable} from 'rxjs'
 import {distinctUntilChanged, map} from 'rxjs/operators'
 
 import {isRecord} from '../util'
-import {create_preview_availability} from './availability'
+import {createPreviewAvailabilityObserver} from './availability'
 import {createGlobalListener} from './createGlobalListener'
 import {createPathObserver} from './createPathObserver'
 import {createPreviewObserver} from './createPreviewObserver'
@@ -64,7 +64,16 @@ export function createDocumentPreviewStore({
 }: DocumentPreviewStoreOptions): DocumentPreviewStore {
   const versionedClient = client.withConfig({apiVersion: '1'})
   const globalListener = createGlobalListener(versionedClient)
-  const {observeFields} = createObserveFields({versionedClient, globalListener})
+
+  const invalidationChannel = globalListener.pipe(
+    map((event) => (event.type === 'welcome' ? {type: 'connected' as const} : event)),
+  )
+
+  const {observeFields} = createObserveFields({
+    client: versionedClient,
+    invalidationChannel,
+  })
+
   const {observePaths} = createPathObserver({observeFields})
 
   function observeDocumentTypeFromId(
@@ -77,15 +86,16 @@ export function createDocumentPreviewStore({
     )
   }
 
-  // const {createPreviewObserver} = create_preview_createPreviewObserver(observeDocumentTypeFromId)
   const observeForPreview = createPreviewObserver({observeDocumentTypeFromId, observePaths})
-  const {observeDocumentPairAvailability} = create_preview_availability(
+  const {observeDocumentPairAvailability} = createPreviewAvailabilityObserver(
     versionedClient,
     observePaths,
   )
   const {observePathsDocumentPair} = create_preview_documentPair(versionedClient, observePaths)
 
   // @todo: explain why the API is like this now, and that it should not be like this in the future!
+  const observeDocument = createObserveDocument({client, mutationChannel: globalListener})
+
   return {
     observePaths,
     observeForPreview,
diff --git a/packages/sanity/src/core/preview/observeFields.ts b/packages/sanity/src/core/preview/observeFields.ts
index 29bd0d776a2..319dc37a5e6 100644
--- a/packages/sanity/src/core/preview/observeFields.ts
+++ b/packages/sanity/src/core/preview/observeFields.ts
@@ -1,4 +1,4 @@
-import {type MutationEvent, type SanityClient, type WelcomeEvent} from '@sanity/client'
+import {type SanityClient} from '@sanity/client'
 import {difference, flatten, memoize} from 'lodash'
 import {
   combineLatest,
@@ -24,7 +24,14 @@ import {
 } from 'rxjs/operators'
 
 import {INCLUDE_FIELDS} from './constants'
-import {type ApiConfig, type FieldName, type Id, type PreviewPath, type Selection} from './types'
+import {
+  type ApiConfig,
+  type FieldName,
+  type Id,
+  type InvalidationChannelEvent,
+  type PreviewPath,
+  type Selection,
+} from './types'
 import {debounceCollect} from './utils/debounceCollect'
 import {hasEqualFields} from './utils/hasEqualFields'
 import {isUniqueBy} from './utils/isUniqueBy'
@@ -40,14 +47,19 @@ type Cache = {
   [id: string]: CachedFieldObserver[]
 }
 
-export function createObserveFields(context: {
-  versionedClient: SanityClient
-  globalListener: Observable<WelcomeEvent | MutationEvent>
+/**
+ * Creates a function that allows observing individual fields on a document.
+ * It will automatically debounce and batch requests, and maintain an in-memory cache of the latest field values
+ * @param options - Options to use when creating the observer
+ */
+export function createObserveFields(options: {
+  client: SanityClient
+  invalidationChannel: Observable<InvalidationChannelEvent>
 }) {
-  const {versionedClient, globalListener} = context
+  const {client: currentDatasetClient, invalidationChannel} = options
   function listen(id: Id) {
-    return globalListener.pipe(
-      filter((event) => event.type === 'welcome' || event.documentId === id),
+    return invalidationChannel.pipe(
+      filter((event) => event.type === 'connected' || event.documentId === id),
     )
   }
 
@@ -60,13 +72,20 @@ export function createObserveFields(context: {
     }
   }
 
-  const fetchDocumentPathsFast = debounceCollect(fetchAllDocumentPathsWith(versionedClient), 100)
-  const fetchDocumentPathsSlow = debounceCollect(fetchAllDocumentPathsWith(versionedClient), 1000)
+  const fetchDocumentPathsFast = debounceCollect(
+    fetchAllDocumentPathsWith(currentDatasetClient),
+    100,
+  )
+
+  const fetchDocumentPathsSlow = debounceCollect(
+    fetchAllDocumentPathsWith(currentDatasetClient),
+    1000,
+  )
 
   function currentDatasetListenFields(id: Id, fields: PreviewPath[]) {
     return listen(id).pipe(
       switchMap((event) => {
-        if (event.type === 'welcome' || event.visibility === 'query') {
+        if (event.type === 'connected' || event.visibility === 'query') {
           return fetchDocumentPathsFast(id, fields as any).pipe(
             mergeMap((result) => {
               return concat(
@@ -85,17 +104,11 @@ export function createObserveFields(context: {
     )
   }
 
-  // keep for debugging purposes for now
-  // function fetchDocumentPaths(id, selection) {
-  //   return client.observable.fetch(`*[_id==$id]{_id,_type,${selection.join(',')}}`, {id})
-  //     .map(result => result[0])
-  // }
-
   const CACHE: Cache = {} // todo: use a LRU cache instead (e.g. hashlru or quick-lru)
 
   const getBatchFetcherForDataset = memoize(
     function getBatchFetcherForDataset(apiConfig: ApiConfig) {
-      const client = versionedClient.withConfig(apiConfig)
+      const client = currentDatasetClient.withConfig(apiConfig)
       const fetchAll = fetchAllDocumentPathsWith(client)
       return debounceCollect(fetchAll, 10)
     },
diff --git a/packages/sanity/src/core/preview/types.ts b/packages/sanity/src/core/preview/types.ts
index c4adb85bab6..dbd82de69fa 100644
--- a/packages/sanity/src/core/preview/types.ts
+++ b/packages/sanity/src/core/preview/types.ts
@@ -112,6 +112,17 @@ export interface DraftsModelDocument<T extends SanityDocumentLike = SanityDocume
   }
 }
 
+/**
+ * Event emitted to notify preview subscribers when they need to refetch a document being previewed
+ * - 'connected' will happen when the store is connected to the invalidation channel, both initially and after a reconnect after a connection loss
+ * - 'mutation' will happen when a document has been mutated and the store needs to refetch a document
+ * @hidden
+ * @beta
+ */
+export type InvalidationChannelEvent =
+  | {type: 'connected'}
+  | {type: 'mutation'; documentId: string; visibility: string}
+
 /**
  * @hidden
  * @beta */

From a89eddde9799385936fcf10f09d76298824876de Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Bj=C3=B8rge=20N=C3=A6ss?= <bjoerge@gmail.com>
Date: Tue, 16 Jul 2024 18:45:09 +0200
Subject: [PATCH 4/6] feat(preview): add unstable_observeDocument +
 unstable_observeDocuments to preview store

---
 .../src/core/preview/createGlobalListener.ts  | 26 +++----
 .../src/core/preview/createObserveDocument.ts | 71 +++++++++++++++----
 .../src/core/preview/documentPreviewStore.ts  | 52 +++++++++++---
 .../sanity/src/core/preview/observeFields.ts  |  3 +-
 .../core/preview/utils/applyMendozaPatch.ts   | 18 +++++
 5 files changed, 134 insertions(+), 36 deletions(-)
 create mode 100644 packages/sanity/src/core/preview/utils/applyMendozaPatch.ts

diff --git a/packages/sanity/src/core/preview/createGlobalListener.ts b/packages/sanity/src/core/preview/createGlobalListener.ts
index 42ee47127b3..883c25c177e 100644
--- a/packages/sanity/src/core/preview/createGlobalListener.ts
+++ b/packages/sanity/src/core/preview/createGlobalListener.ts
@@ -1,5 +1,5 @@
 import {type MutationEvent, type SanityClient, type WelcomeEvent} from '@sanity/client'
-import {merge, timer} from 'rxjs'
+import {defer, merge, timer} from 'rxjs'
 import {filter, share, shareReplay} from 'rxjs/operators'
 
 /**
@@ -7,8 +7,8 @@ import {filter, share, shareReplay} from 'rxjs/operators'
  * Creates a listener that will emit 'welcome' for all new subscribers immediately, and thereafter emit at every mutation event
  */
 export function createGlobalListener(client: SanityClient) {
-  const allEvents$ = client
-    .listen(
+  const allEvents$ = defer(() =>
+    client.listen(
       '*[!(_id in path("_.**"))]',
       {},
       {
@@ -19,19 +19,19 @@ export function createGlobalListener(client: SanityClient) {
         effectFormat: 'mendoza',
         tag: 'preview.global',
       },
-    )
-    .pipe(
-      filter(
-        (event): event is WelcomeEvent | MutationEvent =>
-          event.type === 'welcome' || event.type === 'mutation',
-      ),
-      share({resetOnRefCountZero: () => timer(2000)}),
-    )
+    ),
+  ).pipe(
+    filter(
+      (event): event is WelcomeEvent | MutationEvent =>
+        event.type === 'welcome' || event.type === 'mutation',
+    ),
+    share({resetOnRefCountZero: () => timer(2000), resetOnComplete: true}),
+  )
 
   const welcome$ = allEvents$.pipe(
     filter((event) => event.type === 'welcome'),
-    shareReplay(1),
+    shareReplay({refCount: true, bufferSize: 1}),
   )
-  const mutations$ = allEvents$.pipe(filter((event) => event.type === 'mutation'))
+  const mutations$ = allEvents$.pipe(filter((event) => event.type === 'mutation')).pipe(share())
   return merge(welcome$, mutations$)
 }
diff --git a/packages/sanity/src/core/preview/createObserveDocument.ts b/packages/sanity/src/core/preview/createObserveDocument.ts
index e0649354595..cf255f77c71 100644
--- a/packages/sanity/src/core/preview/createObserveDocument.ts
+++ b/packages/sanity/src/core/preview/createObserveDocument.ts
@@ -1,8 +1,11 @@
 import {type MutationEvent, type SanityClient, type WelcomeEvent} from '@sanity/client'
-import {uniq} from 'lodash'
-import {EMPTY, type Observable, of} from 'rxjs'
-import {concatMap, map, scan} from 'rxjs/operators'
+import {type SanityDocument} from '@sanity/types'
+import {memoize, uniq} from 'lodash'
+import {EMPTY, finalize, type Observable, of} from 'rxjs'
+import {concatMap, map, scan, shareReplay} from 'rxjs/operators'
 
+import {type ApiConfig} from './types'
+import {applyMendozaPatch} from './utils/applyMendozaPatch'
 import {debounceCollect} from './utils/debounceCollect'
 
 export function createObserveDocument({
@@ -12,30 +15,72 @@ export function createObserveDocument({
   client: SanityClient
   mutationChannel: Observable<WelcomeEvent | MutationEvent>
 }) {
-  function batchFetchDocuments(ids: string[][]) {
-    return client.observable
-      .fetch(`*[_id in $ids]`, {ids: uniq(ids.flat())}, {tag: 'preview.observe-document'})
-      .pipe(map((result) => ids.map((id) => result.find(id))))
-  }
+  const getBatchFetcher = memoize(
+    function getBatchFetcher(apiConfig: {dataset: string; projectId: string}) {
+      const _client = client.withConfig(apiConfig)
+
+      function batchFetchDocuments(ids: [string][]) {
+        return _client.observable
+          .fetch(`*[_id in $ids]`, {ids: uniq(ids.flat())}, {tag: 'preview.observe-document'})
+          .pipe(
+            // eslint-disable-next-line max-nested-callbacks
+            map((result) => ids.map(([id]) => result.find((r: {_id: string}) => r._id === id))),
+          )
+      }
+      return debounceCollect(batchFetchDocuments, 100)
+    },
+    (apiConfig) => apiConfig.dataset + apiConfig.projectId,
+  )
 
-  const fetchDocument = debounceCollect(batchFetchDocuments, 10)
+  const MEMO: Record<string, Observable<SanityDocument | undefined>> = {}
 
-  return function observeDocument(id: string) {
+  function observeDocument(id: string, apiConfig?: ApiConfig) {
+    const _apiConfig = apiConfig || {
+      dataset: client.config().dataset!,
+      projectId: client.config().projectId!,
+    }
+    const fetchDocument = getBatchFetcher(_apiConfig)
     return mutationChannel.pipe(
       concatMap((event) => {
         if (event.type === 'welcome') {
-          return fetchDocument(id).pipe(map((document) => ({type: 'sync', document})))
+          return fetchDocument(id).pipe(map((document) => ({type: 'sync' as const, document})))
         }
         return event.documentId === id ? of(event) : EMPTY
       }),
-      scan((current, event) => {
+      scan((current: SanityDocument | undefined, event) => {
         if (event.type === 'sync') {
           return event.document
         }
         if (event.type === 'mutation') {
-          return current
+          return applyMutationEvent(current, event)
         }
+        //@ts-expect-error - this should never happen
+        throw new Error(`Unexpected event type: "${event.type}"`)
       }, undefined),
     )
   }
+  return function memoizedObserveDocument(id: string, apiConfig?: ApiConfig) {
+    const key = apiConfig ? `${id}-${JSON.stringify(apiConfig)}` : id
+    if (!(key in MEMO)) {
+      MEMO[key] = observeDocument(id, apiConfig).pipe(
+        finalize(() => delete MEMO[key]),
+        shareReplay({bufferSize: 1, refCount: true}),
+      )
+    }
+    return MEMO[key]
+  }
+}
+
+function applyMutationEvent(current: SanityDocument | undefined, event: MutationEvent) {
+  if (event.previousRev !== current?._rev) {
+    console.warn('Document out of sync, skipping mutation')
+    return current
+  }
+  if (!event.effects) {
+    throw new Error(
+      'Mutation event is missing effects. Is the listener set up with effectFormat=mendoza?',
+    )
+  }
+  const next = applyMendozaPatch(current, event.effects.apply)
+  return {...next, _rev: event.resultRev}
 }
diff --git a/packages/sanity/src/core/preview/documentPreviewStore.ts b/packages/sanity/src/core/preview/documentPreviewStore.ts
index 77dbfa84c35..be921b68aab 100644
--- a/packages/sanity/src/core/preview/documentPreviewStore.ts
+++ b/packages/sanity/src/core/preview/documentPreviewStore.ts
@@ -1,11 +1,13 @@
 import {type SanityClient} from '@sanity/client'
 import {type PrepareViewOptions, type SanityDocument} from '@sanity/types'
-import {type Observable} from 'rxjs'
+import {pick} from 'lodash'
+import {combineLatest, type Observable} from 'rxjs'
 import {distinctUntilChanged, map} from 'rxjs/operators'
 
 import {isRecord} from '../util'
 import {createPreviewAvailabilityObserver} from './availability'
 import {createGlobalListener} from './createGlobalListener'
+import {createObserveDocument} from './createObserveDocument'
 import {createPathObserver} from './createPathObserver'
 import {createPreviewObserver} from './createPreviewObserver'
 import {create_preview_documentPair} from './documentPair'
@@ -31,6 +33,11 @@ export type ObserveForPreviewFn = (
 ) => Observable<PreparedSnapshot>
 
 /**
+ * The document preview store supports subscribing to content for previewing purposes.
+ * Documents observed by this store will be kept in sync and receive real-time updates from all collaborators,
+ * but has no support for optimistic updates, so any local edits will require a server round-trip before becoming visible,
+ * which means this store is less suitable for real-time editing scenarios.
+ *
  * @hidden
  * @beta */
 export interface DocumentPreviewStore {
@@ -51,6 +58,19 @@ export interface DocumentPreviewStore {
     id: string,
     paths: PreviewPath[],
   ) => Observable<DraftsModelDocument<T>>
+
+  /**
+   * Observe a complete document with the given ID
+   * @hidden
+   * @beta
+   */
+  unstable_observeDocument: (id: string) => Observable<SanityDocument | undefined>
+  /**
+   * Observe a list of complete documents with the given IDs
+   * @hidden
+   * @beta
+   */
+  unstable_observeDocuments: (ids: string[]) => Observable<(SanityDocument | undefined)[]>
 }
 
 /** @internal */
@@ -58,21 +78,36 @@ export interface DocumentPreviewStoreOptions {
   client: SanityClient
 }
 
+/** @internal
+ * Should the preview system fetch partial documents or full documents?
+ * Setting this to true will end up fetching full documents for everything that's currently being previewed in the studio
+ * This comes with an extra memory and initial transfer cost, but gives faster updating previews and less likelihood of displaying
+ * out-of-date previews as documents will be kept in sync by applying mendoza patches, instead of re-fetching preview queries
+ * */
+const PREVIEW_FETCH_FULL_DOCUMENTS = false
+
 /** @internal */
 export function createDocumentPreviewStore({
   client,
 }: DocumentPreviewStoreOptions): DocumentPreviewStore {
   const versionedClient = client.withConfig({apiVersion: '1'})
   const globalListener = createGlobalListener(versionedClient)
-
   const invalidationChannel = globalListener.pipe(
     map((event) => (event.type === 'welcome' ? {type: 'connected' as const} : event)),
   )
 
-  const {observeFields} = createObserveFields({
-    client: versionedClient,
-    invalidationChannel,
-  })
+  const observeDocument = createObserveDocument({client, mutationChannel: globalListener})
+
+  function getObserveFields() {
+    if (PREVIEW_FETCH_FULL_DOCUMENTS) {
+      return function observeFields(id: string, fields: string[], apiConfig?: ApiConfig) {
+        return observeDocument(id, apiConfig).pipe(map((doc) => pick(doc, fields)))
+      }
+    }
+    return createObserveFields({client: versionedClient, invalidationChannel})
+  }
+
+  const observeFields = getObserveFields()
 
   const {observePaths} = createPathObserver({observeFields})
 
@@ -94,14 +129,15 @@ export function createDocumentPreviewStore({
   const {observePathsDocumentPair} = create_preview_documentPair(versionedClient, observePaths)
 
   // @todo: explain why the API is like this now, and that it should not be like this in the future!
-  const observeDocument = createObserveDocument({client, mutationChannel: globalListener})
 
   return {
     observePaths,
     observeForPreview,
     observeDocumentTypeFromId,
 
-    // eslint-disable-next-line camelcase
+    unstable_observeDocument: observeDocument,
+    unstable_observeDocuments: (ids: string[]) =>
+      combineLatest(ids.map((id) => observeDocument(id))),
     unstable_observeDocumentPairAvailability: observeDocumentPairAvailability,
     unstable_observePathsDocumentPair: observePathsDocumentPair,
   }
diff --git a/packages/sanity/src/core/preview/observeFields.ts b/packages/sanity/src/core/preview/observeFields.ts
index 319dc37a5e6..de9ea564b53 100644
--- a/packages/sanity/src/core/preview/observeFields.ts
+++ b/packages/sanity/src/core/preview/observeFields.ts
@@ -187,8 +187,7 @@ export function createObserveFields(options: {
     )
   }
 
-  // API
-  return {observeFields: cachedObserveFields}
+  return cachedObserveFields
 
   function pickFrom(objects: Record<string, any>[], fields: string[]) {
     return [...INCLUDE_FIELDS, ...fields].reduce((result, fieldName) => {
diff --git a/packages/sanity/src/core/preview/utils/applyMendozaPatch.ts b/packages/sanity/src/core/preview/utils/applyMendozaPatch.ts
new file mode 100644
index 00000000000..0c1be69450c
--- /dev/null
+++ b/packages/sanity/src/core/preview/utils/applyMendozaPatch.ts
@@ -0,0 +1,18 @@
+import {type SanityDocument} from '@sanity/types'
+import {applyPatch, type RawPatch} from 'mendoza'
+
+function omitRev(document: SanityDocument | undefined) {
+  if (document === undefined) {
+    return undefined
+  }
+  const {_rev, ...doc} = document
+  return doc
+}
+
+export function applyMendozaPatch(
+  document: SanityDocument | undefined,
+  patch: RawPatch,
+): SanityDocument | undefined {
+  const next = applyPatch(omitRev(document), patch)
+  return next === null ? undefined : next
+}

From 31255345ec8331557dd58e6a46b47d4f34b5fa16 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Bj=C3=B8rge=20N=C3=A6ss?= <bjoerge@gmail.com>
Date: Thu, 18 Jul 2024 14:02:47 +0200
Subject: [PATCH 5/6] fix: use includeMutations: false

---
 packages/sanity/src/core/preview/createGlobalListener.ts | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/packages/sanity/src/core/preview/createGlobalListener.ts b/packages/sanity/src/core/preview/createGlobalListener.ts
index 883c25c177e..f1b68448144 100644
--- a/packages/sanity/src/core/preview/createGlobalListener.ts
+++ b/packages/sanity/src/core/preview/createGlobalListener.ts
@@ -15,7 +15,8 @@ export function createGlobalListener(client: SanityClient) {
         events: ['welcome', 'mutation'],
         includeResult: false,
         includePreviousRevision: false,
-        visibility: 'transaction',
+        includeMutations: false,
+        visibility: 'query',
         effectFormat: 'mendoza',
         tag: 'preview.global',
       },

From 971f3cf41551182eb41451589ff876c4aae9567d Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Bj=C3=B8rge=20N=C3=A6ss?= <bjoerge@gmail.com>
Date: Fri, 19 Jul 2024 15:32:03 +0200
Subject: [PATCH 6/6] fix: expect ts error on includeMutations: false for now

---
 packages/sanity/src/core/preview/createGlobalListener.ts | 1 +
 1 file changed, 1 insertion(+)

diff --git a/packages/sanity/src/core/preview/createGlobalListener.ts b/packages/sanity/src/core/preview/createGlobalListener.ts
index f1b68448144..a5a83dbc6a0 100644
--- a/packages/sanity/src/core/preview/createGlobalListener.ts
+++ b/packages/sanity/src/core/preview/createGlobalListener.ts
@@ -15,6 +15,7 @@ export function createGlobalListener(client: SanityClient) {
         events: ['welcome', 'mutation'],
         includeResult: false,
         includePreviousRevision: false,
+        // @ts-expect-error - will be enabled by https://github.com/sanity-io/client/pull/872
         includeMutations: false,
         visibility: 'query',
         effectFormat: 'mendoza',