feat(preview): add unstable_observeDocument(s) to preview store

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,
 ): {

packages/sanity/src/core/preview/createGlobalListener.ts

@@ -0,0 +1,39 @@
+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) {
+  const allEvents$ = defer(() =>
+    client.listen(
+      '*[!(_id in path("_.**"))]',
+      {},
+      {
+        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',
+        tag: 'preview.global',
+      },
+    ),
+  ).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({refCount: true, bufferSize: 1}),
+  )
+  const mutations$ = allEvents$.pipe(filter((event) => event.type === 'mutation')).pipe(share())
+  return merge(welcome$, mutations$)
+}

packages/sanity/src/core/preview/createObserveDocument.ts

@@ -0,0 +1,86 @@
+import {type MutationEvent, type SanityClient, type WelcomeEvent} from '@sanity/client'
+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({
+  mutationChannel,
+  client,
+}: {
+  client: SanityClient
+  mutationChannel: Observable<WelcomeEvent | MutationEvent>
+}) {
+  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 MEMO: Record<string, Observable<SanityDocument | undefined>> = {}
+
+  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' as const, document})))
+        }
+        return event.documentId === id ? of(event) : EMPTY
+      }),
+      scan((current: SanityDocument | undefined, event) => {
+        if (event.type === 'sync') {
+          return event.document
+        }
+        if (event.type === 'mutation') {
+          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}
+}

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(

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,
   )

packages/sanity/src/core/preview/documentPreviewStore.ts

@@ -1,14 +1,17 @@
 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 {create_preview_availability} from './availability'
+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'
-import {create_preview_observeFields} from './observeFields'
+import {createObserveFields} from './observeFields'
 import {
   type ApiConfig,
   type DraftsModelDocument,
@@ -30,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 {
@@ -50,30 +58,56 @@ 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 */
 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)),
+  )
 
-  // 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 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} = create_preview_observeFields({
-    observePaths: __proxy_observePaths,
-    versionedClient,
-  })
+  const observeFields = getObserveFields()
 
   const {observePaths} = createPathObserver({observeFields})
 
@@ -87,21 +121,23 @@ 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!
+
   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,
   }