Add readTime to MutableDocument

Author: schmidt-sebastian

packages/firestore/src/local/indexeddb_remote_document_cache.ts

@@ -289,7 +289,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)) {
@@ -331,16 +334,15 @@ class IndexedDbRemoteDocumentCacheImpl implements IndexedDbRemoteDocumentCache {
   }
 
   /**
-   * Decodes `remoteDoc` and returns the document (or null, if the document
-   * corresponds to the format used for sentinel deletes).
+   * Decodes `remoteDoc` and returns the document (or an invalid document, if
+   * the document corresponds to the format used for sentinel deletes).
    */
   private maybeDecodeDocument(
     documentKey: DocumentKey,
     dbRemoteDoc: DbRemoteDocument | null
   ): MutableDocument {
     if (dbRemoteDoc) {
       const doc = fromDbRemoteDocument(this.serializer, dbRemoteDoc);
-
       // Whether the document is a sentinel removal and should only be used in the
       // `getNewDocumentChanges()`
       const isSentinelRemoval =
@@ -480,15 +482,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 +505,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,41 @@ 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 +118,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

@@ -26,8 +26,6 @@ import {
   documentKeySet,
   DocumentKeySet,
   DocumentMap,
-  DocumentVersionMap,
-  documentVersionMap,
   mutableDocumentMap,
   MutableDocumentMap
 } from '../model/collections';
@@ -533,7 +531,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 +548,7 @@ export function localStoreApplyRemoteEventToLocalCache(
         populateDocumentChangeBuffer(
           txn,
           documentBuffer,
-          remoteEvent.documentUpdates,
-          remoteVersion,
-          undefined
+          remoteEvent.documentUpdates
         ).next(result => {
           changedDocs = result;
         })
@@ -617,19 +613,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,7 +630,7 @@ 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() ||
@@ -648,10 +639,10 @@ function populateDocumentChangeBuffer(
           existingDoc.hasPendingWrites)
       ) {
         debugAssert(
-          !SnapshotVersion.min().isEqual(docReadTime),
+          !SnapshotVersion.min().isEqual(doc.readTime),
           '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 +1034,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 +1199,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 doc = bundleConverter.toMutableDocument(bundleDoc);
+    doc.setReadTime(
       bundleConverter.toSnapshotVersion(bundleDoc.metadata.readTime!)
     );
+    documentMap = documentMap.insert(documentKey, doc);
   }
 
   const documentBuffer = localStoreImpl.remoteDocuments.newChangeBuffer({
@@ -1237,13 +1225,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);
       }

packages/firestore/src/local/remote_document_change_buffer.ts

@@ -28,13 +28,6 @@ import { PersistenceTransaction } from './persistence_transaction';
 /**
  * Represents a document change to be applied to remote document cache.
  */
-interface RemoteDocumentChange {
-  // The document in this change. Contains an invalid document if it is a
-  // removed from the cache.
-  readonly document: MutableDocument;
-  // The timestamp when this change is read.
-  readonly readTime: SnapshotVersion | null;
-}
 
 /**
  * An in-memory buffer of entries to be written to a RemoteDocumentCache.
@@ -51,13 +44,11 @@ interface RemoteDocumentChange {
  * porting this class as part of that implementation work.
  */
 export abstract class RemoteDocumentChangeBuffer {
-  // A mapping of document key to the new cache entry that should be written (or null if any
-  // existing cache entry should be removed).
-  protected changes: ObjectMap<DocumentKey, RemoteDocumentChange> =
-    new ObjectMap(
-      key => key.toString(),
-      (l, r) => l.isEqual(r)
-    );
+  // A mapping of document key to the new cache entry that should be written.
+  protected changes: ObjectMap<DocumentKey, MutableDocument> = new ObjectMap(
+    key => key.toString(),
+    (l, r) => l.isEqual(r)
+  );
 
   private changesApplied = false;
 
@@ -75,27 +66,15 @@ export abstract class RemoteDocumentChangeBuffer {
     transaction: PersistenceTransaction
   ): PersistencePromise<void>;
 
-  protected getReadTime(key: DocumentKey): SnapshotVersion {
-    const change = this.changes.get(key);
-    if (change) {
-      debugAssert(
-        !!change.readTime,
-        `Read time is not set for ${key}. All removeEntry() calls must include a readTime if 'trackRemovals' is used.`
-      );
-      return change.readTime;
-    }
-    return SnapshotVersion.min();
-  }
-
   /**
    * Buffers a `RemoteDocumentCache.addEntry()` call.
    *
    * You can only modify documents that have already been retrieved via
    * `getEntry()/getEntries()` (enforced via IndexedDbs `apply()`).
    */
-  addEntry(document: MutableDocument, readTime: SnapshotVersion): void {
+  addEntry(document: MutableDocument): void {
     this.assertNotApplied();
-    this.changes.set(document.key, { document, readTime });
+    this.changes.set(document.key, document);
   }
 
   /**
@@ -104,12 +83,12 @@ export abstract class RemoteDocumentChangeBuffer {
    * You can only remove documents that have already been retrieved via
    * `getEntry()/getEntries()` (enforced via IndexedDbs `apply()`).
    */
-  removeEntry(key: DocumentKey, readTime: SnapshotVersion | null = null): void {
+  removeEntry(key: DocumentKey, readTime: SnapshotVersion): void {
     this.assertNotApplied();
-    this.changes.set(key, {
-      document: MutableDocument.newInvalidDocument(key),
-      readTime
-    });
+    this.changes.set(
+      key,
+      MutableDocument.newInvalidDocument(key).setReadTime(readTime)
+    );
   }
 
   /**
@@ -130,7 +109,7 @@ export abstract class RemoteDocumentChangeBuffer {
     this.assertNotApplied();
     const bufferedEntry = this.changes.get(documentKey);
     if (bufferedEntry !== undefined) {
-      return PersistencePromise.resolve(bufferedEntry.document);
+      return PersistencePromise.resolve(bufferedEntry);
     } else {
       return this.getFromCache(transaction, documentKey);
     }