Prepre for serializaion/deserialization

Author: wu-hui

packages/firestore/src/core/target.ts

@@ -52,6 +52,7 @@ import {
   orderByEquals,
   stringifyOrderBy
 } from './order_by';
+import { Pipeline } from '../pipelines/api/pipeline';
 
 /**
  * A Target represents the WatchTarget representation of a Query, which is used
@@ -215,6 +216,12 @@ export function targetEquals(left: Target, right: Target): boolean {
   return boundEquals(left.endAt, right.endAt);
 }
 
+export function targetIsPipelineTarget(
+  target: Target | Pipeline
+): target is Pipeline {
+  return target instanceof Pipeline;
+}
+
 export function targetIsDocumentTarget(target: Target): boolean {
   return (
     DocumentKey.isDocumentKey(target.path) &&

packages/firestore/src/local/indexeddb_schema.ts

@@ -22,6 +22,7 @@ import {
   Document as ProtoDocument,
   DocumentsTarget as ProtoDocumentsTarget,
   QueryTarget as ProtoQueryTarget,
+  PipelineQueryTarget as ProtoPipelineQueryTarget,
   Write as ProtoWrite
 } from '../protos/firestore_proto_api';
 
@@ -253,7 +254,10 @@ export interface DbRemoteDocumentGlobal {
  * IndexedDb. We use the proto definitions for these two kinds of queries in
  * order to avoid writing extra serialization logic.
  */
-export type DbQuery = ProtoQueryTarget | ProtoDocumentsTarget;
+export type DbQuery =
+  | ProtoQueryTarget
+  | ProtoDocumentsTarget
+  | ProtoPipelineQueryTarget;
 
 /**
  * An object to be stored in the 'targets' store in IndexedDb.

packages/firestore/src/local/indexeddb_target_cache.ts

@@ -268,7 +268,8 @@ export class IndexedDbTargetCache implements TargetCache {
           const found = fromDbTarget(value);
           // After finding a potential match, check that the target is
           // actually equal to the requested target.
-          if (targetEquals(target, found.target)) {
+          // TODO(pipeline): This needs to handle pipeline properly.
+          if (targetEquals(target, found.target as Target)) {
             result = found;
             control.done();
           }

packages/firestore/src/local/local_serializer.ts

@@ -19,7 +19,12 @@ import { Timestamp } from '../api/timestamp';
 import { BundleMetadata, NamedQuery } from '../core/bundle';
 import { LimitType, Query, queryWithLimit } from '../core/query';
 import { SnapshotVersion } from '../core/snapshot_version';
-import { canonifyTarget, Target, targetIsDocumentTarget } from '../core/target';
+import {
+  canonifyTarget,
+  Target,
+  targetIsDocumentTarget,
+  targetIsPipelineTarget
+} from '../core/target';
 import { MutableDocument } from '../model/document';
 import { DocumentKey } from '../model/document_key';
 import {
@@ -36,18 +41,23 @@ import {
   BundleMetadata as ProtoBundleMetadata,
   NamedQuery as ProtoNamedQuery
 } from '../protos/firestore_bundle_proto';
-import { DocumentsTarget as PublicDocumentsTarget } from '../protos/firestore_proto_api';
+import {
+  DocumentsTarget as PublicDocumentsTarget,
+  PipelineQueryTarget as PublicPipelineQueryTarget
+} from '../protos/firestore_proto_api';
 import {
   convertQueryTargetToQuery,
   fromDocument,
   fromDocumentsTarget,
   fromMutation,
+  fromPipelineTarget,
   fromQueryTarget,
   fromVersion,
   JsonProtoSerializer,
   toDocument,
   toDocumentsTarget,
   toMutation,
+  toPipelineTarget,
   toQueryTarget
 } from '../remote/serializer';
 import { debugAssert, fail } from '../util/assert';
@@ -71,6 +81,7 @@ import {
 } from './indexeddb_schema';
 import { DbDocumentOverlayKey, DbTimestampKey } from './indexeddb_sentinels';
 import { TargetData, TargetPurpose } from './target_data';
+import { Pipeline } from '../pipelines/api/pipeline';
 
 /** Serializer for values stored in the LocalStore. */
 export class LocalSerializer {
@@ -241,8 +252,10 @@ export function fromDbTarget(dbTarget: DbTarget): TargetData {
       ? fromDbTimestamp(dbTarget.lastLimboFreeSnapshotVersion)
       : SnapshotVersion.min();
 
-  let target: Target;
-  if (isDocumentQuery(dbTarget.query)) {
+  let target: Target | Pipeline;
+  if (isPipelineQueryTarget(dbTarget.query)) {
+    target = fromPipelineTarget(dbTarget.query);
+  } else if (isDocumentQuery(dbTarget.query)) {
     target = fromDocumentsTarget(dbTarget.query);
   } else {
     target = fromQueryTarget(dbTarget.query);
@@ -275,7 +288,21 @@ export function toDbTarget(
     targetData.lastLimboFreeSnapshotVersion
   );
   let queryProto: DbQuery;
-  if (targetIsDocumentTarget(targetData.target)) {
+  if (targetIsPipelineTarget(targetData.target)) {
+    queryProto = toPipelineTarget(
+      localSerializer.remoteSerializer,
+      targetData.target
+    );
+    return {
+      targetId: targetData.targetId,
+      canonicalId: '',
+      readTime: dbTimestamp,
+      resumeToken: '',
+      lastListenSequenceNumber: targetData.sequenceNumber,
+      lastLimboFreeSnapshotVersion: dbLastLimboFreeTimestamp,
+      query: queryProto
+    };
+  } else if (targetIsDocumentTarget(targetData.target)) {
     queryProto = toDocumentsTarget(
       localSerializer.remoteSerializer,
       targetData.target
@@ -303,6 +330,12 @@ export function toDbTarget(
   };
 }
 
+function isPipelineQueryTarget(
+  dbQuery: DbQuery
+): dbQuery is PublicPipelineQueryTarget {
+  return (dbQuery as PublicPipelineQueryTarget).pipeline !== undefined;
+}
+
 /**
  * A helper function for figuring out what kind of query has been stored.
  */

packages/firestore/src/local/local_store_impl.ts

@@ -1063,7 +1063,8 @@ export async function localStoreReleaseTarget(
 
   localStoreImpl.targetDataByTarget =
     localStoreImpl.targetDataByTarget.remove(targetId);
-  localStoreImpl.targetIdByTarget.delete(targetData!.target);
+  // TODO(pipeline): This needs to handle pipeline properly.
+  localStoreImpl.targetIdByTarget.delete(targetData!.target as Target);
 }
 
 /**
@@ -1220,15 +1221,21 @@ export function localStoreGetCachedTarget(
   );
   const cachedTargetData = localStoreImpl.targetDataByTarget.get(targetId);
   if (cachedTargetData) {
-    return Promise.resolve(cachedTargetData.target);
+    // TODO(pipeline): This needs to handle pipeline properly.
+    return Promise.resolve(cachedTargetData.target as Target);
   } else {
     return localStoreImpl.persistence.runTransaction(
       'Get target data',
       'readonly',
       txn => {
-        return targetCacheImpl
-          .getTargetDataForTarget(txn, targetId)
-          .next(targetData => (targetData ? targetData.target : null));
+        return (
+          targetCacheImpl
+            .getTargetDataForTarget(txn, targetId)
+            // TODO(pipeline): This needs to handle pipeline properly.
+            .next(targetData =>
+              targetData ? (targetData.target as Target) : null
+            )
+        );
       }
     );
   }

packages/firestore/src/local/memory_target_cache.ts

@@ -101,7 +101,8 @@ export class MemoryTargetCache implements TargetCache {
   }
 
   private saveTargetData(targetData: TargetData): void {
-    this.targets.set(targetData.target, targetData);
+    // TODO(pipeline): This needs to handle pipeline properly.
+    this.targets.set(targetData.target as Target, targetData);
     const targetId = targetData.targetId;
     if (targetId > this.highestTargetId) {
       this.targetIdGenerator = new TargetIdGenerator(targetId);
@@ -117,7 +118,8 @@ export class MemoryTargetCache implements TargetCache {
     targetData: TargetData
   ): PersistencePromise<void> {
     debugAssert(
-      !this.targets.has(targetData.target),
+      // TODO(pipeline): This needs to handle pipeline properly.
+      !this.targets.has(targetData.target as Target),
       'Adding a target that already exists'
     );
     this.saveTargetData(targetData);
@@ -130,7 +132,8 @@ export class MemoryTargetCache implements TargetCache {
     targetData: TargetData
   ): PersistencePromise<void> {
     debugAssert(
-      this.targets.has(targetData.target),
+      // TODO(pipeline): This needs to handle pipeline properly.
+      this.targets.has(targetData.target as Target),
       'Updating a nonexistent target'
     );
     this.saveTargetData(targetData);
@@ -143,10 +146,11 @@ export class MemoryTargetCache implements TargetCache {
   ): PersistencePromise<void> {
     debugAssert(this.targetCount > 0, 'Removing a target from an empty cache');
     debugAssert(
-      this.targets.has(targetData.target),
+      // TODO(pipeline): This needs to handle pipeline properly.
+      this.targets.has(targetData.target as Target),
       'Removing a nonexistent target from the cache'
     );
-    this.targets.delete(targetData.target);
+    this.targets.delete(targetData.target as Target);
     this.references.removeReferencesForId(targetData.targetId);
     this.targetCount -= 1;
     return PersistencePromise.resolve();

packages/firestore/src/local/target_data.ts

@@ -19,6 +19,7 @@ import { SnapshotVersion } from '../core/snapshot_version';
 import { Target } from '../core/target';
 import { ListenSequenceNumber, TargetId } from '../core/types';
 import { ByteString } from '../util/byte_string';
+import { Pipeline } from '../pipelines/api/pipeline';
 
 /** An enumeration of the different purposes we have for targets. */
 export const enum TargetPurpose {
@@ -47,7 +48,7 @@ export const enum TargetPurpose {
 export class TargetData {
   constructor(
     /** The target being listened to. */
-    readonly target: Target,
+    readonly target: Target | Pipeline,
     /**
      * The target ID to which the target corresponds; Assigned by the
      * LocalStore for user listens and by the SyncEngine for limbo watches.

packages/firestore/src/protos/firestore_proto_api.ts

@@ -559,6 +559,8 @@ export declare type Pipeline = firestoreV1ApiClientInterfaces.Pipeline;
 export declare type Precondition = firestoreV1ApiClientInterfaces.Precondition;
 export declare type Projection = firestoreV1ApiClientInterfaces.Projection;
 export declare type QueryTarget = firestoreV1ApiClientInterfaces.QueryTarget;
+export declare type PipelineQueryTarget =
+  firestoreV1ApiClientInterfaces.PipelineQueryTarget;
 export declare type ReadOnly = firestoreV1ApiClientInterfaces.ReadOnly;
 export declare type ReadWrite = firestoreV1ApiClientInterfaces.ReadWrite;
 export declare type RollbackRequest =

packages/firestore/src/remote/serializer.ts

@@ -35,7 +35,11 @@ import {
   queryToTarget
 } from '../core/query';
 import { SnapshotVersion } from '../core/snapshot_version';
-import { targetIsDocumentTarget, Target } from '../core/target';
+import {
+  targetIsDocumentTarget,
+  Target,
+  targetIsPipelineTarget
+} from '../core/target';
 import { TargetId } from '../core/types';
 import { Bytes } from '../lite-api/bytes';
 import { GeoPoint } from '../lite-api/geo_point';
@@ -84,6 +88,7 @@ import {
   OrderDirection as ProtoOrderDirection,
   Precondition as ProtoPrecondition,
   QueryTarget as ProtoQueryTarget,
+  PipelineQueryTarget as ProtoPipelineQueryTarget,
   RunAggregationQueryRequest as ProtoRunAggregationQueryRequest,
   Aggregation as ProtoAggregation,
   Status as ProtoStatus,
@@ -111,6 +116,7 @@ import {
   WatchTargetChange,
   WatchTargetChangeState
 } from './watch_change';
+import { Pipeline } from '../api';
 
 const DIRECTIONS = (() => {
   const dirs: { [dir: string]: ProtoOrderDirection } = {};
@@ -1087,14 +1093,28 @@ export function toLabel(purpose: TargetPurpose): string | null {
   }
 }
 
+export function fromPipelineTarget(target: ProtoPipelineQueryTarget): Pipeline {
+  return {} as Pipeline;
+}
+
+export function toPipelineTarget(
+  serializer: JsonProtoSerializer,
+  target: Pipeline
+): ProtoPipelineQueryTarget {
+  return {
+    pipeline: {}
+  };
+}
+
 export function toTarget(
   serializer: JsonProtoSerializer,
   targetData: TargetData
 ): ProtoTarget {
   let result: ProtoTarget;
   const target = targetData.target;
-
-  if (targetIsDocumentTarget(target)) {
+  if (targetIsPipelineTarget(target)) {
+    result = { pipelineQuery: toPipelineTarget(serializer, target) };
+  } else if (targetIsDocumentTarget(target)) {
     result = { documents: toDocumentsTarget(serializer, target) };
   } else {
     result = { query: toQueryTarget(serializer, target).queryTarget };

packages/firestore/src/remote/watch_change.ts

@@ -17,7 +17,7 @@
 
 import { DatabaseId } from '../core/database_info';
 import { SnapshotVersion } from '../core/snapshot_version';
-import { targetIsDocumentTarget } from '../core/target';
+import { targetIsDocumentTarget, targetIsPipelineTarget } from '../core/target';
 import { TargetId } from '../core/types';
 import { ChangeType } from '../core/view_snapshot';
 import { TargetData, TargetPurpose } from '../local/target_data';
@@ -414,7 +414,9 @@ export class WatchChangeAggregator {
     const targetData = this.targetDataForActiveTarget(targetId);
     if (targetData) {
       const target = targetData.target;
-      if (targetIsDocumentTarget(target)) {
+      if (targetIsPipelineTarget(target)) {
+        //TODO(pipeline): handle existence filter correctly for pipelines
+      } else if (targetIsDocumentTarget(target)) {
         if (expectedCount === 0) {
           // The existence filter told us the document does not exist. We deduce
           // that this document does not exist and apply a deleted document to
@@ -584,7 +586,11 @@ export class WatchChangeAggregator {
     this.targetStates.forEach((targetState, targetId) => {
       const targetData = this.targetDataForActiveTarget(targetId);
       if (targetData) {
-        if (targetState.current && targetIsDocumentTarget(targetData.target)) {
+        if (
+          targetState.current &&
+          !targetIsPipelineTarget(targetData.target) &&
+          targetIsDocumentTarget(targetData.target)
+        ) {
           // Document queries for document that don't exist can produce an empty
           // result set. To update our local cache, we synthesize a document
           // delete if we have not previously received the document. This