[base] Inline document store

packages/@sanity/base/src/components/QueryContainer.js

@@ -17,7 +17,7 @@ import {
 import {combineLatest, concat, merge, of} from 'rxjs'
 import deepEquals from 'react-fast-compare'
 import {createEventHandler, streamingComponent} from 'react-props-stream'
-import {listenQuery} from '../util/listenQuery'
+import {listenQuery} from '../datastores/document/listenQuery'
 
 const INITIAL_CHILD_PROPS = {
   result: null,

packages/@sanity/base/src/components/WithReferringDocuments.js

@@ -9,13 +9,15 @@ const loadProps = receivedProps$ =>
     switchMap(receivedProps =>
       concat(
         of({...receivedProps, referringDocuments: [], isLoading: true}),
-        documentStore.query('*[references($docId)] [0...101]', {docId: receivedProps.id}).pipe(
-          map(event => ({
-            ...receivedProps,
-            referringDocuments: event.documents,
-            isLoading: false
-          }))
-        )
+        documentStore
+          .listenQuery('*[references($docId)] [0...101]', {docId: receivedProps.id})
+          .pipe(
+            map(docs => ({
+              ...receivedProps,
+              referringDocuments: docs,
+              isLoading: false
+            }))
+          )
       )
     )
   )

packages/@sanity/base/src/datastores/document/_createDeprecatedAPIs.ts

@@ -0,0 +1,141 @@
+/* eslint-disable @typescript-eslint/no-use-before-define */
+import {Observable, defer, of as observableOf} from 'rxjs'
+import {concatMap, map, share} from 'rxjs/operators'
+import {createBufferedDocument} from './buffered-doc/createBufferedDocument'
+import {doCommit} from './checkoutPair'
+import {SanityDocument, WelcomeEvent} from './types'
+
+function fetchDocumentSnapshot(client, id) {
+  return client.observable.getDocument(id).pipe(
+    map(document => ({
+      type: 'snapshot',
+      document: document
+    }))
+  )
+}
+
+export interface QuerySnapshotEvent {
+  type: 'snapshot'
+  documents: SanityDocument[]
+}
+
+type QueryEvent = WelcomeEvent | MutationEvent | QuerySnapshotEvent
+
+function _createDeprecatedAPIs(client) {
+  const _doCommit = mutations => doCommit(client, mutations)
+
+  function patchDoc(documentId, patches) {
+    const doc = checkout(documentId)
+    doc.patch(patches)
+    return doc.commit()
+  }
+
+  function deleteDoc(documentId) {
+    const doc = checkout(documentId)
+    doc.delete()
+    doc.commit()
+  }
+
+  function checkout(documentId) {
+    const serverEvents$ = client
+      .listen(
+        '*[_id == $id]',
+        {id: documentId},
+        {includeResult: false, events: ['welcome', 'mutation', 'reconnect']}
+      )
+      .pipe(
+        concatMap((event: any) => {
+          return event.type === 'welcome'
+            ? fetchDocumentSnapshot(client, documentId)
+            : observableOf(event)
+        }),
+        share()
+      )
+
+    return createBufferedDocument(documentId, serverEvents$, _doCommit)
+  }
+
+  function byId(documentId) {
+    return checkout(documentId).events
+  }
+
+  function byIds(documentIds) {
+    return new Observable(observer => {
+      const documentSubscriptions = documentIds.map(id => byId(id).subscribe(observer))
+
+      return () => {
+        documentSubscriptions.map(subscription => subscription.unsubscribe())
+      }
+    })
+  }
+
+  function create(document) {
+    return client.observable.create(document)
+  }
+
+  function createIfNotExists(document) {
+    return client.observable.createIfNotExists(document)
+  }
+
+  function createOrReplace(document) {
+    return client.observable.createOrReplace(document)
+  }
+
+  function fetchQuerySnapshot(groqQuery: string, params): Observable<QuerySnapshotEvent> {
+    return client.observable.fetch(groqQuery, params).pipe(
+      map(documents => ({
+        type: 'snapshot',
+        documents: documents
+      }))
+    )
+  }
+
+  function query(groqQuery: string, params: {}): Observable<QueryEvent> {
+    return defer(
+      () =>
+        client.observable.listen(groqQuery, params || {}, {
+          includeResult: false,
+          events: ['welcome', 'mutation', 'reconnect']
+        }) as Observable<WelcomeEvent | MutationEvent>
+    ).pipe(
+      concatMap(event => {
+        return event.type === 'welcome'
+          ? fetchQuerySnapshot(groqQuery, params)
+          : observableOf(event)
+      })
+    )
+  }
+
+  return {
+    byId,
+    byIds,
+    create,
+    checkout,
+    query,
+    patch: patchDoc,
+    delete: deleteDoc,
+    createOrReplace: createOrReplace,
+    createIfNotExists: createIfNotExists
+  }
+}
+
+function deprecate(name, fn) {
+  return (...args) => {
+    console.warn(
+      'The `documentStore.%s()-method is deprecated and should not be relied upon. Please use checkoutPair() or listenQuery() instead.',
+      name
+    )
+    return fn(...args)
+  }
+}
+
+function mapObj(obj, mapFn) {
+  return Object.keys(obj).reduce((acc, key) => {
+    acc[key] = mapFn(obj[key], key)
+    return acc
+  }, {})
+}
+
+export default function createDeprecatedAPIs(client) {
+  return mapObj(_createDeprecatedAPIs(client), (fn, key) => deprecate(key, fn))
+}

packages/@sanity/base/src/datastores/document/buffered-doc/createBufferedDocument.ts

@@ -0,0 +1,66 @@
+import {createObservableBufferedDocument} from './createObservableBufferedDocument'
+import {filter} from 'rxjs/operators'
+import {merge, Observable} from 'rxjs'
+import {ReconnectEvent} from '../types'
+import {
+  CommitFunction,
+  SnapshotEvent,
+  CommittedEvent,
+  DocumentRebaseEvent,
+  DocumentMutationEvent
+} from './types'
+import {ListenerEvent} from '../getPairListener'
+
+type BufferedDocumentEvent =
+  | ReconnectEvent
+  | SnapshotEvent
+  | DocumentRebaseEvent
+  | DocumentMutationEvent
+  | CommittedEvent
+
+export interface BufferedDocumentWrapper {
+  events: Observable<BufferedDocumentEvent>
+  patch: (patches) => void
+  create: (document) => void
+  createIfNotExists: (document) => void
+  createOrReplace: (document) => void
+  delete: () => void
+  commit: () => Observable<never>
+}
+
+function isReconnect(event: ListenerEvent): event is ReconnectEvent {
+  return event.type === 'reconnect'
+}
+export const createBufferedDocument = (
+  documentId: string,
+  serverEvents$: Observable<ListenerEvent>,
+  doCommit: CommitFunction
+): BufferedDocumentWrapper => {
+  const bufferedDocument = createObservableBufferedDocument(serverEvents$, doCommit)
+
+  const reconnects$ = serverEvents$.pipe(filter(isReconnect))
+
+  return {
+    events: merge(reconnects$, bufferedDocument.updates$),
+    patch(patches) {
+      bufferedDocument.addMutations(patches.map(patch => ({patch: {...patch, id: documentId}})))
+    },
+    create(document) {
+      bufferedDocument.addMutation({
+        create: Object.assign({id: documentId}, document)
+      })
+    },
+    createIfNotExists(document) {
+      bufferedDocument.addMutation({createIfNotExists: document})
+    },
+    createOrReplace(document) {
+      bufferedDocument.addMutation({createOrReplace: document})
+    },
+    delete() {
+      bufferedDocument.addMutation({delete: {id: documentId}})
+    },
+    commit() {
+      return bufferedDocument.commit()
+    }
+  }
+}