Change ProtoByteString to use its own class rather than String | Uint8Array

packages/firestore/src/core/types.ts

@@ -15,6 +15,8 @@
  * limitations under the License.
  */
 
+import { Blob } from '../api/blob';
+
 /**
  * BatchID is a locally assigned ID for a batch of mutations that have been
  * applied.
@@ -29,9 +31,7 @@ export type TargetId = number;
 
 export type ListenSequenceNumber = number;
 
-// TODO(b/35918695): In GRPC / node, tokens are Uint8Array. In WebChannel,
-// they're strings. We should probably (de-)serialize to a common internal type.
-export type ProtoByteString = Uint8Array | string;
+export type ProtoByteString = Blob;
 
 /** The different states of a mutation batch. */
 export type MutationBatchState = 'pending' | 'acknowledged' | 'rejected';

packages/firestore/src/local/indexeddb_mutation_queue.ts

@@ -47,7 +47,8 @@ import { LocalSerializer } from './local_serializer';
 import { MutationQueue } from './mutation_queue';
 import { PersistenceTransaction, ReferenceDelegate } from './persistence';
 import { PersistencePromise } from './persistence_promise';
-import { SimpleDb, SimpleDbStore, SimpleDbTransaction } from './simple_db';
+import { SimpleDbStore, SimpleDbTransaction } from './simple_db';
+import { Blob } from '../api/blob';
 
 /** A mutation queue for a specific user, backed by IndexedDB. */
 export class IndexedDbMutationQueue implements MutationQueue {
@@ -124,7 +125,9 @@ export class IndexedDbMutationQueue implements MutationQueue {
     streamToken: ProtoByteString
   ): PersistencePromise<void> {
     return this.getMutationQueueMetadata(transaction).next(metadata => {
-      metadata.lastStreamToken = convertStreamToken(streamToken);
+      // Convert the streamToken to base64 in order to store it as a string that can
+      // be reconstructed into a Blob type.
+      metadata.lastStreamToken = streamToken.toBase64();
 
       return mutationQueuesStore(transaction).put(metadata);
     });
@@ -134,7 +137,7 @@ export class IndexedDbMutationQueue implements MutationQueue {
     transaction: PersistenceTransaction
   ): PersistencePromise<ProtoByteString> {
     return this.getMutationQueueMetadata(transaction).next<ProtoByteString>(
-      metadata => metadata.lastStreamToken
+      metadata => Blob.fromBase64String(metadata.lastStreamToken)
     );
   }
 
@@ -143,7 +146,9 @@ export class IndexedDbMutationQueue implements MutationQueue {
     streamToken: ProtoByteString
   ): PersistencePromise<void> {
     return this.getMutationQueueMetadata(transaction).next(metadata => {
-      metadata.lastStreamToken = convertStreamToken(streamToken);
+      // Convert the streamToken to base64 in order to store it as a string that can
+      // be reconstructed into a Blob type.
+      metadata.lastStreamToken = streamToken.toBase64();
       return mutationQueuesStore(transaction).put(metadata);
     });
   }
@@ -671,19 +676,6 @@ export function removeMutationBatch(
   return PersistencePromise.waitFor(promises).next(() => removedDocuments);
 }
 
-function convertStreamToken(token: ProtoByteString): string {
-  if (token instanceof Uint8Array) {
-    // TODO(b/78771403): Convert tokens to strings during deserialization
-    assert(
-      SimpleDb.isMockPersistence(),
-      'Persisting non-string stream tokens is only supported with mock persistence.'
-    );
-    return token.toString();
-  } else {
-    return token;
-  }
-}
-
 /**
  * Helper to get a typed SimpleDbStore for the mutations object store.
  */

packages/firestore/src/local/local_serializer.ts

@@ -42,8 +42,8 @@ import {
   DbTimestampKey,
   DbUnknownDocument
 } from './indexeddb_schema';
-import { SimpleDb } from './simple_db';
 import { TargetData, TargetPurpose } from './target_data';
+import { Blob } from '../api/blob';
 
 /** Serializer for values stored in the LocalStore. */
 export class LocalSerializer {
@@ -207,8 +207,6 @@ export class LocalSerializer {
       dbTarget.lastLimboFreeSnapshotVersion !== undefined
         ? this.fromDbTimestamp(dbTarget.lastLimboFreeSnapshotVersion)
         : SnapshotVersion.MIN;
-    // TODO(b/140573486): Convert to platform representation
-    const resumeToken = dbTarget.resumeToken;
 
     let target: Target;
     if (isDocumentQuery(dbTarget.query)) {
@@ -223,7 +221,7 @@ export class LocalSerializer {
       dbTarget.lastListenSequenceNumber,
       version,
       lastLimboFreeSnapshotVersion,
-      resumeToken
+      Blob.fromBase64String(dbTarget.resumeToken)
     );
   }
 
@@ -247,18 +245,9 @@ export class LocalSerializer {
       queryProto = this.remoteSerializer.toQueryTarget(targetData.target);
     }
 
-    let resumeToken: string;
-
-    if (targetData.resumeToken instanceof Uint8Array) {
-      // TODO(b/78771403): Convert tokens to strings during deserialization
-      assert(
-        SimpleDb.isMockPersistence(),
-        'Persisting non-string stream tokens is only supported with mock persistence .'
-      );
-      resumeToken = targetData.resumeToken.toString();
-    } else {
-      resumeToken = targetData.resumeToken;
-    }
+    // We can't store the resumeToken as a Blob in IndexedDb, so we convert it
+    // to a base64 string for storage.
+    const resumeToken = targetData.resumeToken.toBase64();
 
     // lastListenSequenceNumber is always 0 until we do real GC.
     return new DbTarget(

packages/firestore/src/local/local_store.ts

@@ -551,7 +551,7 @@ export class LocalStore {
 
               const resumeToken = change.resumeToken;
               // Update the resume token if the change includes one.
-              if (resumeToken.length > 0) {
+              if (resumeToken._approximateByteSize() > 0) {
                 const newTargetData = oldTargetData
                   .withResumeToken(resumeToken, remoteVersion)
                   .withSequenceNumber(txn.currentSequenceNumber);
@@ -696,12 +696,12 @@ export class LocalStore {
     change: TargetChange
   ): boolean {
     assert(
-      newTargetData.resumeToken.length > 0,
+      newTargetData.resumeToken._approximateByteSize() > 0,
       'Attempted to persist target data with no resume token'
     );
 
     // Always persist target data if we don't already have a resume token.
-    if (oldTargetData.resumeToken.length === 0) {
+    if (oldTargetData.resumeToken._approximateByteSize() === 0) {
       return true;
     }
 

packages/firestore/src/local/memory_mutation_queue.ts

@@ -22,7 +22,6 @@ import { DocumentKeySet } from '../model/collections';
 import { DocumentKey } from '../model/document_key';
 import { Mutation } from '../model/mutation';
 import { MutationBatch, BATCHID_UNKNOWN } from '../model/mutation_batch';
-import { emptyByteString } from '../platform/platform';
 import { assert } from '../util/assert';
 import { primitiveComparator } from '../util/misc';
 import { SortedMap } from '../util/sorted_map';
@@ -33,6 +32,7 @@ import { MutationQueue } from './mutation_queue';
 import { PersistenceTransaction, ReferenceDelegate } from './persistence';
 import { PersistencePromise } from './persistence_promise';
 import { DocReference } from './reference_set';
+import { emptyByteString } from '../util/proto_byte_string';
 
 export class MemoryMutationQueue implements MutationQueue {
   /**

packages/firestore/src/local/target_data.ts

@@ -18,7 +18,7 @@
 import { SnapshotVersion } from '../core/snapshot_version';
 import { Target } from '../core/target';
 import { ListenSequenceNumber, ProtoByteString, TargetId } from '../core/types';
-import { emptyByteString } from '../platform/platform';
+import { emptyByteString } from '../util/proto_byte_string';
 
 /** An enumeration of the different purposes we have for targets. */
 export enum TargetPurpose {
@@ -128,7 +128,7 @@ export class TargetData {
       this.lastLimboFreeSnapshotVersion.isEqual(
         other.lastLimboFreeSnapshotVersion
       ) &&
-      this.resumeToken === other.resumeToken &&
+      this.resumeToken.isEqual(other.resumeToken) &&
       this.target.isEqual(other.target)
     );
   }

packages/firestore/src/platform/platform.ts

@@ -16,7 +16,6 @@
  */
 
 import { DatabaseId, DatabaseInfo } from '../core/database_info';
-import { ProtoByteString } from '../core/types';
 import { Connection } from '../remote/connection';
 import { JsonProtoSerializer } from '../remote/serializer';
 import { fail } from '../util/assert';
@@ -52,8 +51,6 @@ export interface Platform {
 
   /** True if and only if the Base64 conversion functions are available. */
   readonly base64Available: boolean;
-
-  readonly emptyByteString: ProtoByteString;
 }
 
 /**
@@ -77,11 +74,3 @@ export class PlatformSupport {
     return PlatformSupport.platform;
   }
 }
-
-/**
- * Returns the representation of an empty "proto" byte string for the
- * platform.
- */
-export function emptyByteString(): ProtoByteString {
-  return PlatformSupport.getPlatform().emptyByteString;
-}

packages/firestore/src/platform_browser/browser_platform.ts

@@ -28,8 +28,6 @@ import { WebChannelConnection } from './webchannel_connection';
 export class BrowserPlatform implements Platform {
   readonly base64Available: boolean;
 
-  readonly emptyByteString = '';
-
   constructor() {
     this.base64Available = typeof atob !== 'undefined';
   }

packages/firestore/src/platform_node/node_platform.ts

@@ -31,8 +31,6 @@ import { loadProtos } from './load_protos';
 export class NodePlatform implements Platform {
   readonly base64Available = true;
 
-  readonly emptyByteString = new Uint8Array(0);
-
   readonly document = null;
 
   get window(): Window | null {

packages/firestore/src/remote/persistent_stream.ts

@@ -32,7 +32,7 @@ import { ExponentialBackoff } from './backoff';
 import { Connection, Stream } from './connection';
 import { JsonProtoSerializer } from './serializer';
 import { WatchChange } from './watch_change';
-import { emptyByteString } from '../platform/platform';
+import { emptyByteString } from '../util/proto_byte_string';
 
 const LOG_TAG = 'PersistentStream';
 
@@ -698,7 +698,7 @@ export class PersistentWriteStream extends PersistentStream<
       !!responseProto.streamToken,
       'Got a write response without a stream token'
     );
-    this.lastStreamToken = responseProto.streamToken;
+    this.lastStreamToken = this.serializer.fromBytes(responseProto.streamToken);
 
     if (!this.handshakeComplete_) {
       // The first response is always the handshake response
@@ -748,14 +748,12 @@ export class PersistentWriteStream extends PersistentStream<
       'Handshake must be complete before writing mutations'
     );
     assert(
-      this.lastStreamToken.length > 0,
+      this.lastStreamToken._approximateByteSize() > 0,
       'Trying to write mutation without a token'
     );
 
     const request: WriteRequest = {
-      // Protos are typed with string, but we support UInt8Array on Node
-      // eslint-disable-next-line @typescript-eslint/no-explicit-any
-      streamToken: this.lastStreamToken as any,
+      streamToken: this.serializer.toBytes(this.lastStreamToken),
       writes: mutations.map(mutation => this.serializer.toMutation(mutation))
     };
 

packages/firestore/src/remote/remote_event.ts

@@ -24,8 +24,8 @@ import {
   MaybeDocumentMap,
   targetIdSet
 } from '../model/collections';
-import { emptyByteString } from '../platform/platform';
 import { SortedSet } from '../util/sorted_set';
+import { emptyByteString } from '../util/proto_byte_string';
 
 /**
  * An event from the RemoteStore. It is split into targetChanges (changes to the

packages/firestore/src/remote/remote_store.ts

@@ -26,7 +26,6 @@ import {
   MutationBatch,
   MutationBatchResult
 } from '../model/mutation_batch';
-import { emptyByteString } from '../platform/platform';
 import { assert } from '../util/assert';
 import { FirestoreError } from '../util/error';
 import * as log from '../util/log';
@@ -52,6 +51,7 @@ import {
   WatchTargetChange,
   WatchTargetChangeState
 } from './watch_change';
+import { emptyByteString } from '../util/proto_byte_string';
 
 const LOG_TAG = 'RemoteStore';
 
@@ -431,7 +431,7 @@ export class RemoteStore implements TargetMetadataProvider {
     // Update in-memory resume tokens. LocalStore will update the
     // persistent view of these when applying the completed RemoteEvent.
     objUtils.forEachNumber(remoteEvent.targetChanges, (targetId, change) => {
-      if (change.resumeToken.length > 0) {
+      if (change.resumeToken._approximateByteSize() > 0) {
         const targetData = this.listenTargets[targetId];
         // A watched target might have been removed already.
         if (targetData) {