[5/4] Listen Options and Server Timestamp

wu-hui · wu-hui · commit e7b831429594 · 2025-06-27T14:34:22.000-04:00
diff --git a/packages/firestore/src/api/pipeline_impl.ts b/packages/firestore/src/api/pipeline_impl.ts
@@ -42,7 +42,11 @@ import { ensureFirestoreConfigured, Firestore } from './database';
 import { Pipeline } from './pipeline'; // Keep this specific Pipeline import if needed alongside LitePipeline
 import { RealtimePipeline } from './realtime_pipeline';
 import { DocumentReference } from './reference';
-import { SnapshotListenOptions, Unsubscribe } from './reference_impl';
+import {
+  PipelineListenOptions,
+  SnapshotListenOptions,
+  Unsubscribe
+} from './reference_impl';
 import { RealtimePipelineSnapshot } from './snapshot';
 import { ExpUserDataWriter } from './user_data_writer';
 
@@ -165,7 +169,7 @@ export function _onRealtimePipelineSnapshot(
  */
 export function _onRealtimePipelineSnapshot(
   pipeline: RealtimePipeline,
-  options: SnapshotListenOptions,
+  options: PipelineListenOptions,
   observer: {
     next?: (snapshot: RealtimePipelineSnapshot) => void;
     error?: (error: FirestoreError) => void;
@@ -188,7 +192,7 @@ export function _onRealtimePipelineSnapshot(
  */
 export function _onRealtimePipelineSnapshot(
   pipeline: RealtimePipeline,
-  options: SnapshotListenOptions,
+  options: PipelineListenOptions,
   onNext: (snapshot: RealtimePipelineSnapshot) => void,
   onError?: (error: FirestoreError) => void,
   onComplete?: () => void
@@ -197,9 +201,10 @@ export function _onRealtimePipelineSnapshot(
   pipeline: RealtimePipeline,
   ...args: unknown[]
 ): Unsubscribe {
-  let options: SnapshotListenOptions = {
+  let options: PipelineListenOptions = {
     includeMetadataChanges: false,
-    source: 'default'
+    source: 'default',
+    serverTimestampBehavior: 'none'
   };
   let currArg = 0;
   if (typeof args[currArg] === 'object' && !isPartialObserver(args[currArg])) {
@@ -209,7 +214,8 @@ export function _onRealtimePipelineSnapshot(
 
   const internalOptions = {
     includeMetadataChanges: options.includeMetadataChanges,
-    source: options.source as ListenerDataSource
+    source: options.source as ListenerDataSource,
+    serverTimestampBehavior: options.serverTimestampBehavior
   };
 
   let userObserver: PartialObserver<RealtimePipelineSnapshot>;
@@ -227,7 +233,9 @@ export function _onRealtimePipelineSnapshot(
   const observer = {
     next: (snapshot: ViewSnapshot) => {
       if (userObserver.next) {
-        userObserver.next(new RealtimePipelineSnapshot(pipeline, snapshot));
+        userObserver.next(
+          new RealtimePipelineSnapshot(pipeline, snapshot, internalOptions)
+        );
       }
     },
     error: userObserver.error,
@@ -236,7 +244,7 @@ export function _onRealtimePipelineSnapshot(
 
   return firestoreClientListen(
     client,
-    toCorePipeline(pipeline),
+    toCorePipeline(pipeline, internalOptions),
     internalOptions, // Pass parsed options here
     observer
   );
diff --git a/packages/firestore/src/api/reference_impl.ts b/packages/firestore/src/api/reference_impl.ts
@@ -757,6 +757,21 @@ export function onSnapshot<AppModelType, DbModelType extends DocumentData>(
   );
 }
 
+export interface PipelineListenOptions {
+  /**
+   * Include a change even if only the metadata of the query or of a document
+   * changed. Default is false.
+   */
+  readonly includeMetadataChanges?: boolean;
+
+  /**
+   * Set the source the query listens to. Default to "default", which
+   * listens to both cache and server.
+   */
+  readonly source?: ListenSource;
+  readonly serverTimestampBehavior?: 'estimate' | 'previous' | 'none';
+}
+
 export function onPipelineSnapshot(
   query: RealtimePipeline,
   observer: {
@@ -767,7 +782,7 @@ export function onPipelineSnapshot(
 ): Unsubscribe;
 export function onPipelineSnapshot(
   query: RealtimePipeline,
-  options: SnapshotListenOptions,
+  options: PipelineListenOptions,
   observer: {
     next?: (snapshot: RealtimePipelineSnapshot) => void;
     error?: (error: FirestoreError) => void;
@@ -782,7 +797,7 @@ export function onPipelineSnapshot(
 ): Unsubscribe;
 export function onPipelineSnapshot(
   query: RealtimePipeline,
-  options: SnapshotListenOptions,
+  options: PipelineListenOptions,
   onNext: (snapshot: RealtimePipelineSnapshot) => void,
   onError?: (error: FirestoreError) => void,
   onCompletion?: () => void
@@ -793,9 +808,10 @@ export function onPipelineSnapshot(
 ): Unsubscribe {
   reference = getModularInstance(reference);
 
-  let options: SnapshotListenOptions = {
+  let options: PipelineListenOptions = {
     includeMetadataChanges: false,
-    source: 'default'
+    source: 'default',
+    serverTimestampBehavior: 'none'
   };
   let currArg = 0;
   if (typeof args[currArg] === 'object' && !isPartialObserver(args[currArg])) {
@@ -805,7 +821,8 @@ export function onPipelineSnapshot(
 
   const internalOptions = {
     includeMetadataChanges: options.includeMetadataChanges,
-    source: options.source as ListenerDataSource
+    source: options.source as ListenerDataSource,
+    serverTimestampBehavior: options.serverTimestampBehavior
   };
 
   if (isPartialObserver(args[currArg])) {
@@ -823,12 +840,16 @@ export function onPipelineSnapshot(
 
   // RealtimePipeline
   firestore = cast(reference._db, Firestore);
-  internalQuery = toCorePipeline(reference);
+  internalQuery = toCorePipeline(reference, internalOptions);
   observer = {
     next: snapshot => {
       if (args[currArg]) {
         (args[currArg] as NextFn<RealtimePipelineSnapshot>)(
-          new RealtimePipelineSnapshot(reference as RealtimePipeline, snapshot)
+          new RealtimePipelineSnapshot(
+            reference as RealtimePipeline,
+            snapshot,
+            internalOptions
+          )
         );
       }
     },
diff --git a/packages/firestore/src/api/snapshot.ts b/packages/firestore/src/api/snapshot.ts
@@ -15,6 +15,7 @@
  * limitations under the License.
  */
 
+import { ListenOptions } from '../core/event_manager';
 import { CorePipeline } from '../core/pipeline';
 import { isPipeline } from '../core/pipeline-util';
 import { newPipelineComparator } from '../core/pipeline_run';
@@ -847,7 +848,8 @@ export function resultChangesFromSnapshot(
         new SnapshotMetadata(
           querySnapshot._snapshot.mutatedKeys.has(change.doc.key),
           querySnapshot._snapshot.fromCache
-        )
+        ),
+        querySnapshot._listenOptions
       );
       lastDoc = change.doc;
       return {
@@ -877,7 +879,8 @@ export function resultChangesFromSnapshot(
           new SnapshotMetadata(
             querySnapshot._snapshot.mutatedKeys.has(change.doc.key),
             querySnapshot._snapshot.fromCache
-          )
+          ),
+          querySnapshot._listenOptions
         );
         let oldIndex = -1;
         let newIndex = -1;
@@ -917,7 +920,11 @@ export class RealtimePipelineSnapshot {
   private _cachedChangesIncludeMetadataChanges?: boolean;
 
   /** @hideconstructor */
-  constructor(pipeline: RealtimePipeline, readonly _snapshot: ViewSnapshot) {
+  constructor(
+    pipeline: RealtimePipeline,
+    readonly _snapshot: ViewSnapshot,
+    readonly _listenOptions: ListenOptions
+  ) {
     this.metadata = new SnapshotMetadata(
       _snapshot.hasPendingWrites,
       _snapshot.fromCache
@@ -929,7 +936,7 @@ export class RealtimePipelineSnapshot {
   get results(): PipelineResult[] {
     const result: PipelineResult[] = [];
     this._snapshot.docs.forEach(doc =>
-      result.push(toPipelineResult(doc, this.pipeline))
+      result.push(toPipelineResult(doc, this.pipeline, this._listenOptions))
     );
     return result;
   }
diff --git a/packages/firestore/src/core/event_manager.ts b/packages/firestore/src/core/event_manager.ts
@@ -408,6 +408,9 @@ export interface ListenOptions {
 
   /** Set the source events raised from. */
   readonly source?: ListenerDataSource;
+
+  /** Realtime Pipeline Only: Set server timestamp behavior. */
+  readonly serverTimestampBehavior?: 'estimate' | 'previous' | 'none';
 }
 
 /**
diff --git a/packages/firestore/src/core/expressions.ts b/packages/firestore/src/core/expressions.ts
@@ -57,6 +57,12 @@ import { objectSize } from '../util/obj';
 import { isNegativeZero } from '../util/types';
 
 import { EvaluationContext, PipelineInputOutput } from './pipeline_run';
+import {
+  getLocalWriteTime,
+  getPreviousValue,
+  isServerTimestamp
+} from '../model/server_timestamps';
+import { SnapshotVersion } from './snapshot_version';
 
 export type EvaluateResultType =
   | 'ERROR'
@@ -309,7 +315,34 @@ export class CoreField implements EvaluableExpr {
     }
     // Return 'UNSET' if the field doesn't exist, otherwise the Value.
     const result = input.data.field(this.expr._fieldPath);
+
+    function getServerTimestampValue(
+      context: EvaluationContext,
+      value: Value
+    ): Value {
+      if (context.serverTimestampBehavior === 'estimate') {
+        return {
+          timestampValue: toVersion(
+            context.serializer,
+            SnapshotVersion.fromTimestamp(getLocalWriteTime(value))
+          )
+        };
+      } else if (context.serverTimestampBehavior === 'previous') {
+        const previousValue = getPreviousValue(value);
+        if (previousValue) {
+          return previousValue;
+        }
+      }
+      return { nullValue: 'NULL_VALUE' };
+    }
+
     if (!!result) {
+      if (isServerTimestamp(result)) {
+        return EvaluateResult.newValue(
+          getServerTimestampValue(context, result)
+        );
+      }
+
       return EvaluateResult.newValue(result);
     } else {
       return EvaluateResult.newUnset();
diff --git a/packages/firestore/src/core/pipeline-util.ts b/packages/firestore/src/core/pipeline-util.ts
@@ -93,6 +93,7 @@ import {
   targetIsPipelineTarget
 } from './target';
 import { VectorValue } from '../api';
+import { ListenOptions } from './event_manager';
 
 /* eslint @typescript-eslint/no-explicit-any: 0 */
 
@@ -596,7 +597,12 @@ function addSystemFields(fields: Map<string, Expr>): Map<string, Expr> {
 }
 
 export function toCorePipeline(
-  p: ApiPipeline | RealtimePipeline
+  p: ApiPipeline | RealtimePipeline,
+  listenOptions?: ListenOptions
 ): CorePipeline {
-  return new CorePipeline(p.userDataReader.serializer, rewriteStages(p.stages));
+  return new CorePipeline(
+    p.userDataReader.serializer,
+    rewriteStages(p.stages),
+    listenOptions
+  );
 }
diff --git a/packages/firestore/src/core/pipeline.ts b/packages/firestore/src/core/pipeline.ts
@@ -28,12 +28,14 @@ import { JsonProtoSerializer } from '../remote/serializer';
 import { debugAssert } from '../util/assert';
 
 import { PipelineFlavor, PipelineSourceType } from './pipeline-util';
+import { ListenOptions } from './event_manager';
 
 export class CorePipeline {
   isCorePipeline = true;
   constructor(
     readonly serializer: JsonProtoSerializer,
-    readonly stages: Stage[]
+    readonly stages: Stage[],
+    readonly listenOptions?: ListenOptions
   ) {}
   getPipelineCollection(): string | undefined {
     return getPipelineCollection(this);
diff --git a/packages/firestore/src/core/pipeline_run.ts b/packages/firestore/src/core/pipeline_run.ts
@@ -45,6 +45,7 @@ export type PipelineInputOutput = MutableDocument;
 
 export interface EvaluationContext {
   serializer: JsonProtoSerializer;
+  serverTimestampBehavior?: 'estimate' | 'previous' | 'none';
 }
 
 export function runPipeline(
@@ -53,7 +54,14 @@ export function runPipeline(
 ): PipelineInputOutput[] {
   let current = input;
   for (const stage of pipeline.stages) {
-    current = evaluate({ serializer: pipeline.serializer }, stage, current);
+    current = evaluate(
+      {
+        serializer: pipeline.serializer,
+        serverTimestampBehavior: pipeline.listenOptions?.serverTimestampBehavior
+      },
+      stage,
+      current
+    );
   }
 
   return current;
diff --git a/packages/firestore/src/lite-api/pipeline-result.ts b/packages/firestore/src/lite-api/pipeline-result.ts
diff --git a/packages/firestore/test/integration/api/pipeline.listen.test.ts b/packages/firestore/test/integration/api/pipeline.listen.test.ts

Original file line number	Diff line number	Diff line change
`@@ -408,6 +408,9 @@ export interface ListenOptions {`
`408`	`408`
`409`	`409`	`/** Set the source events raised from. */`
`410`	`410`	`readonly source?: ListenerDataSource;`
	`411`	`+`
	`412`	`+ /** Realtime Pipeline Only: Set server timestamp behavior. */`
	`413`	`+ readonly serverTimestampBehavior?: 'estimate' \| 'previous' \| 'none';`
`411`	`414`	`}`
`412`	`415`
`413`	`416`	`/**`