Add ReadTime to MutableDocument

Author: schmidt-sebastian

packages/firestore/src/local/indexeddb_remote_document_cache.ts

@@ -53,6 +53,7 @@ import { PersistenceTransaction } from './persistence_transaction';
 import { RemoteDocumentCache } from './remote_document_cache';
 import { RemoteDocumentChangeBuffer } from './remote_document_change_buffer';
 import { IterateOptions, SimpleDbStore } from './simple_db';
+import { version } from '../../test/util/helpers';
 
 export interface DocumentSizeEntry {
   document: MutableDocument;
@@ -289,7 +290,10 @@ class IndexedDbRemoteDocumentCacheImpl implements IndexedDbRemoteDocumentCache {
           return;
         }
 
-        const document = fromDbRemoteDocument(this.serializer, dbRemoteDoc);
+        const document = this.maybeDecodeDocument(
+          DocumentKey.fromSegments(key),
+          dbRemoteDoc
+        );
         if (!query.path.isPrefixOf(document.key.path)) {
           control.done();
         } else if (queryMatches(query, document)) {
@@ -334,13 +338,12 @@ class IndexedDbRemoteDocumentCacheImpl implements IndexedDbRemoteDocumentCache {
    * Decodes `remoteDoc` and returns the document (or null, if the document
    * corresponds to the format used for sentinel deletes).
    */
-  private maybeDecodeDocument(
+  maybeDecodeDocument(
     documentKey: DocumentKey,
     dbRemoteDoc: DbRemoteDocument | null
   ): MutableDocument {
     if (dbRemoteDoc) {
-      const doc = fromDbRemoteDocument(this.serializer, dbRemoteDoc);
-
+      let doc = fromDbRemoteDocument(this.serializer, dbRemoteDoc);
       // Whether the document is a sentinel removal and should only be used in the
       // `getNewDocumentChanges()`
       const isSentinelRemoval =
@@ -392,11 +395,11 @@ export function remoteDocumentCacheGetNewDocumentChanges(
   return documentsStore
     .iterate(
       { index: DbRemoteDocument.readTimeIndex, range },
-      (_, dbRemoteDoc) => {
+      (key, dbRemoteDoc) => {
         // Unlike `getEntry()` and others, `getNewDocumentChanges()` parses
         // the documents directly since we want to keep sentinel deletes.
-        const doc = fromDbRemoteDocument(
-          remoteDocumentCacheImpl.serializer,
+        const doc = remoteDocumentCacheImpl.maybeDecodeDocument(
+          DocumentKey.fromSegments(key),
           dbRemoteDoc
         );
         changedDocs = changedDocs.insert(doc.key, doc);
@@ -480,15 +483,14 @@ class IndexedDbRemoteDocumentChangeBuffer extends RemoteDocumentChangeBuffer {
         previousSize !== undefined,
         `Cannot modify a document that wasn't read (for ${key})`
       );
-      if (documentChange.document.isValidDocument()) {
+      if (documentChange.isValidDocument()) {
         debugAssert(
-          !this.getReadTime(key).isEqual(SnapshotVersion.min()),
+          !documentChange.readTime.isEqual(SnapshotVersion.min()),
           'Cannot add a document with a read time of zero'
         );
         const doc = toDbRemoteDocument(
           this.documentCache.serializer,
-          documentChange.document,
-          this.getReadTime(key)
+          documentChange
         );
         collectionParents = collectionParents.add(key.path.popLast());
 
@@ -504,8 +506,7 @@ class IndexedDbRemoteDocumentChangeBuffer extends RemoteDocumentChangeBuffer {
           // preserved in `getNewDocumentChanges()`.
           const deletedDoc = toDbRemoteDocument(
             this.documentCache.serializer,
-            MutableDocument.newNoDocument(key, SnapshotVersion.min()),
-            this.getReadTime(key)
+            documentChange.convertToNoDocument(SnapshotVersion.min())
           );
           promises.push(
             this.documentCache.addEntry(transaction, key, deletedDoc)

packages/firestore/src/local/local_serializer.ts

@@ -69,35 +69,40 @@ export function fromDbRemoteDocument(
   localSerializer: LocalSerializer,
   remoteDoc: DbRemoteDocument
 ): MutableDocument {
+  let doc: MutableDocument;
   if (remoteDoc.document) {
-    return fromDocument(
+    doc = fromDocument(
       localSerializer.remoteSerializer,
       remoteDoc.document,
       !!remoteDoc.hasCommittedMutations
     );
   } else if (remoteDoc.noDocument) {
     const key = DocumentKey.fromSegments(remoteDoc.noDocument.path);
     const version = fromDbTimestamp(remoteDoc.noDocument.readTime);
-    const document = MutableDocument.newNoDocument(key, version);
-    return remoteDoc.hasCommittedMutations
-      ? document.setHasCommittedMutations()
-      : document;
+    doc = MutableDocument.newNoDocument(key, version);
+    if (remoteDoc.hasCommittedMutations) {
+      doc.setHasCommittedMutations();
+    }
   } else if (remoteDoc.unknownDocument) {
     const key = DocumentKey.fromSegments(remoteDoc.unknownDocument.path);
     const version = fromDbTimestamp(remoteDoc.unknownDocument.version);
-    return MutableDocument.newUnknownDocument(key, version);
+    doc = MutableDocument.newUnknownDocument(key, version);
   } else {
     return fail('Unexpected DbRemoteDocument');
   }
+
+  if (remoteDoc.readTime) {
+    doc.setReadTime(fromDbTimestampKey(remoteDoc.readTime));
+  }
+  return doc;
 }
 
 /** Encodes a document for storage locally. */
 export function toDbRemoteDocument(
   localSerializer: LocalSerializer,
-  document: MutableDocument,
-  readTime: SnapshotVersion
+  document: MutableDocument
 ): DbRemoteDocument {
-  const dbReadTime = toDbTimestampKey(readTime);
+  const dbReadTime = toDbTimestampKey(document.readTime);
   const parentPath = document.key.path.popLast().toArray();
   if (document.isFoundDocument()) {
     const doc = toDocument(localSerializer.remoteSerializer, document);
@@ -112,11 +117,11 @@ export function toDbRemoteDocument(
     );
   } else if (document.isNoDocument()) {
     const path = document.key.path.toArray();
-    const readTime = toDbTimestamp(document.version);
+    const version = toDbTimestamp(document.version);
     const hasCommittedMutations = document.hasCommittedMutations;
     return new DbRemoteDocument(
       /* unknownDocument= */ null,
-      new DbNoDocument(path, readTime),
+      new DbNoDocument(path, version),
       /* document= */ null,
       hasCommittedMutations,
       dbReadTime,

packages/firestore/src/local/local_store_impl.ts

@@ -533,7 +533,7 @@ export function localStoreApplyRemoteEventToLocalCache(
       });
 
       let changedDocs = mutableDocumentMap();
-      remoteEvent.documentUpdates.forEach((key, doc) => {
+      remoteEvent.documentUpdates.forEach(key => {
         if (remoteEvent.resolvedLimboDocuments.has(key)) {
           promises.push(
             localStoreImpl.persistence.referenceDelegate.updateLimboDocument(
@@ -550,9 +550,7 @@ export function localStoreApplyRemoteEventToLocalCache(
         populateDocumentChangeBuffer(
           txn,
           documentBuffer,
-          remoteEvent.documentUpdates,
-          remoteVersion,
-          undefined
+          remoteEvent.documentUpdates
         ).next(result => {
           changedDocs = result;
         })
@@ -617,19 +615,14 @@ export function localStoreApplyRemoteEventToLocalCache(
 function populateDocumentChangeBuffer(
   txn: PersistenceTransaction,
   documentBuffer: RemoteDocumentChangeBuffer,
-  documents: MutableDocumentMap,
-  globalVersion: SnapshotVersion,
-  // TODO(wuandy): We could add `readTime` to MaybeDocument instead to remove
-  // this parameter.
-  documentVersions: DocumentVersionMap | undefined
+  documents: MutableDocumentMap
 ): PersistencePromise<MutableDocumentMap> {
   let updatedKeys = documentKeySet();
   documents.forEach(k => (updatedKeys = updatedKeys.add(k)));
   return documentBuffer.getEntries(txn, updatedKeys).next(existingDocs => {
     let changedDocs = mutableDocumentMap();
     documents.forEach((key, doc) => {
       const existingDoc = existingDocs.get(key)!;
-      const docReadTime = documentVersions?.get(key) || globalVersion;
 
       // Note: The order of the steps below is important, since we want
       // to ensure that rejected limbo resolutions (which fabricate
@@ -639,19 +632,15 @@ function populateDocumentChangeBuffer(
         // NoDocuments with SnapshotVersion.min() are used in manufactured
         // events. We remove these documents from cache since we lost
         // access.
-        documentBuffer.removeEntry(key, docReadTime);
+        documentBuffer.removeEntry(key, doc.readTime);
         changedDocs = changedDocs.insert(key, doc);
       } else if (
         !existingDoc.isValidDocument() ||
         doc.version.compareTo(existingDoc.version) > 0 ||
         (doc.version.compareTo(existingDoc.version) === 0 &&
           existingDoc.hasPendingWrites)
       ) {
-        debugAssert(
-          !SnapshotVersion.min().isEqual(docReadTime),
-          'Cannot add a document when the remote version is zero'
-        );
-        documentBuffer.addEntry(doc, docReadTime);
+        documentBuffer.addEntry(doc);
         changedDocs = changedDocs.insert(key, doc);
       } else {
         logDebug(
@@ -1043,7 +1032,8 @@ function applyWriteToRemoteDocuments(
             // We use the commitVersion as the readTime rather than the
             // document's updateTime since the updateTime is not advanced
             // for updates that do not modify the underlying document.
-            documentBuffer.addEntry(doc, batchResult.commitVersion);
+            doc.setReadTime(batchResult.commitVersion);
+            documentBuffer.addEntry(doc);
           }
         }
       });
@@ -1207,20 +1197,16 @@ export async function localStoreApplyBundledDocuments(
   const localStoreImpl = debugCast(localStore, LocalStoreImpl);
   let documentKeys = documentKeySet();
   let documentMap = mutableDocumentMap();
-  let versionMap = documentVersionMap();
   for (const bundleDoc of documents) {
     const documentKey = bundleConverter.toDocumentKey(bundleDoc.metadata.name!);
     if (bundleDoc.document) {
       documentKeys = documentKeys.add(documentKey);
     }
-    documentMap = documentMap.insert(
-      documentKey,
-      bundleConverter.toMutableDocument(bundleDoc)
-    );
-    versionMap = versionMap.insert(
-      documentKey,
+    const document = bundleConverter.toMutableDocument(bundleDoc);
+    document.setReadTime(
       bundleConverter.toSnapshotVersion(bundleDoc.metadata.readTime!)
     );
+    documentMap = documentMap.insert(documentKey, document);
   }
 
   const documentBuffer = localStoreImpl.remoteDocuments.newChangeBuffer({
@@ -1237,13 +1223,7 @@ export async function localStoreApplyBundledDocuments(
     'Apply bundle documents',
     'readwrite',
     txn => {
-      return populateDocumentChangeBuffer(
-        txn,
-        documentBuffer,
-        documentMap,
-        SnapshotVersion.min(),
-        versionMap
-      )
+      return populateDocumentChangeBuffer(txn, documentBuffer, documentMap)
         .next(changedDocs => {
           documentBuffer.apply(txn);
           return changedDocs;

packages/firestore/src/local/memory_remote_document_cache.ts

@@ -39,7 +39,6 @@ export type DocumentSizer = (doc: Document) => number;
 interface MemoryRemoteDocumentCacheEntry {
   document: Document;
   size: number;
-  readTime: SnapshotVersion;
 }
 
 type DocumentEntryMap = SortedMap<DocumentKey, MemoryRemoteDocumentCacheEntry>;
@@ -85,11 +84,10 @@ class MemoryRemoteDocumentCacheImpl implements MemoryRemoteDocumentCache {
    */
   addEntry(
     transaction: PersistenceTransaction,
-    doc: MutableDocument,
-    readTime: SnapshotVersion
+    doc: MutableDocument
   ): PersistencePromise<void> {
     debugAssert(
-      !readTime.isEqual(SnapshotVersion.min()),
+      !doc.readTime.isEqual(SnapshotVersion.min()),
       'Cannot add a document with a read time of zero'
     );
 
@@ -100,8 +98,7 @@ class MemoryRemoteDocumentCacheImpl implements MemoryRemoteDocumentCache {
 
     this.docs = this.docs.insert(key, {
       document: doc.mutableCopy(),
-      size: currentSize,
-      readTime
+      size: currentSize
     });
 
     this.size += currentSize - previousSize;
@@ -173,12 +170,12 @@ class MemoryRemoteDocumentCacheImpl implements MemoryRemoteDocumentCache {
     while (iterator.hasNext()) {
       const {
         key,
-        value: { document, readTime }
+        value: { document }
       } = iterator.getNext();
       if (!query.path.isPrefixOf(key.path)) {
         break;
       }
-      if (readTime.compareTo(sinceReadTime) <= 0) {
+      if (document.readTime.compareTo(sinceReadTime) <= 0) {
         continue;
       }
       if (!queryMatches(query, document)) {
@@ -237,14 +234,8 @@ class MemoryRemoteDocumentChangeBuffer extends RemoteDocumentChangeBuffer {
   ): PersistencePromise<void> {
     const promises: Array<PersistencePromise<void>> = [];
     this.changes.forEach((key, doc) => {
-      if (doc.document.isValidDocument()) {
-        promises.push(
-          this.documentCache.addEntry(
-            transaction,
-            doc.document,
-            this.getReadTime(key)
-          )
-        );
+      if (doc.isValidDocument()) {
+        promises.push(this.documentCache.addEntry(transaction, doc));
       } else {
         this.documentCache.removeEntry(key);
       }