firebase-js-sdk/packages/firestore/test/unit/specs/spec_test_runner.ts

/**
 * @license
 * Copyright 2017 Google LLC
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import { assert, expect, use } from 'chai';
import chaiExclude from 'chai-exclude';

import { LoadBundleTask } from '../../../src/api/bundle';
import {
  EmptyAppCheckTokenProvider,
  EmptyAuthCredentialsProvider
} from '../../../src/api/credentials';
import {
  IndexConfiguration,
  parseIndexes
} from '../../../src/api/index_configuration';
import { User } from '../../../src/auth/user';
import { ComponentConfiguration } from '../../../src/core/component_provider';
import { DatabaseInfo } from '../../../src/core/database_info';
import {
  addSnapshotsInSyncListener,
  EventManager,
  eventManagerListen,
  eventManagerUnlisten,
  ListenerDataSource as Source,
  ListenOptions,
  Observer,
  QueryListener,
  removeSnapshotsInSyncListener
} from '../../../src/core/event_manager';
import { CorePipeline } from '../../../src/core/pipeline';
import {
  canonifyQueryOrPipeline,
  pipelineEq,
  QueryOrPipeline,
  queryOrPipelineEqual,
  TargetOrPipeline,
  toCorePipeline,
  toPipelineStages
} from '../../../src/core/pipeline-util';
import {
  LimitType,
  newQueryForCollectionGroup,
  Query,
  QueryImpl,
  queryToTarget,
  queryWithAddedFilter,
  queryWithAddedOrderBy,
  queryWithLimit
} from '../../../src/core/query';
import { SnapshotVersion } from '../../../src/core/snapshot_version';
import { SyncEngine } from '../../../src/core/sync_engine';
import {
  syncEngineGetActiveLimboDocumentResolutions,
  syncEngineGetEnqueuedLimboDocumentResolutions,
  syncEngineListen,
  syncEngineLoadBundle,
  syncEngineRegisterPendingWritesCallback,
  syncEngineUnlisten,
  syncEngineWrite,
  triggerRemoteStoreListen,
  triggerRemoteStoreUnlisten
} from '../../../src/core/sync_engine_impl';
import { targetIsPipelineTarget } from '../../../src/core/target';
import { RemoteTargetId, TargetId } from '../../../src/core/types';
import {
  ChangeType,
  DocumentViewChange
} from '../../../src/core/view_snapshot';
import { IndexedDbLruDelegateImpl } from '../../../src/local/indexeddb_lru_delegate_impl';
import { IndexedDbPersistence } from '../../../src/local/indexeddb_persistence';
import {
  DbPrimaryClient,
  SCHEMA_VERSION
} from '../../../src/local/indexeddb_schema';
import { SchemaConverter } from '../../../src/local/indexeddb_schema_converter';
import {
  DbPrimaryClientKey,
  DbPrimaryClientStore
} from '../../../src/local/indexeddb_sentinels';
import { fromPipelineTarget } from '../../../src/local/local_serializer';
import { LocalStore } from '../../../src/local/local_store';
import { localStoreConfigureFieldIndexes } from '../../../src/local/local_store_impl';
import { LruGarbageCollector } from '../../../src/local/lru_garbage_collector';
import { MemoryLruDelegate } from '../../../src/local/memory_persistence';
import {
  ClientId,
  SharedClientState
} from '../../../src/local/shared_client_state';
import { SimpleDb } from '../../../src/local/simple_db';
import { TargetData, TargetPurpose } from '../../../src/local/target_data';
import { DocumentKey } from '../../../src/model/document_key';
import { FieldIndex } from '../../../src/model/field_index';
import { Mutation } from '../../../src/model/mutation';
import { JsonObject } from '../../../src/model/object_value';
import { encodeBase64 } from '../../../src/platform/base64';
import { toByteStreamReader } from '../../../src/platform/byte_stream_reader';
import { newTextEncoder } from '../../../src/platform/text_serializer';
import * as api from '../../../src/protos/firestore_proto_api';
import { ExistenceFilter } from '../../../src/remote/existence_filter';
import {
  fillWritePipeline,
  outstandingWrites,
  RemoteStore,
  remoteStoreDisableNetwork,
  remoteStoreEnableNetwork,
  remoteStoreHandleCredentialChange,
  remoteStoreGetRemoteTargetId,
  remoteStoreShutdown
} from '../../../src/remote/remote_store';
import { mapCodeFromRpcCode } from '../../../src/remote/rpc_error';
import {
  JsonProtoSerializer,
  toListenRequestLabels,
  toMutation,
  toTarget,
  toVersion
} from '../../../src/remote/serializer';
import {
  DocumentWatchChange,
  ExistenceFilterChange,
  WatchChange,
  WatchTargetChange,
  WatchTargetChangeState
} from '../../../src/remote/watch_change';
import { debugAssert, fail } from '../../../src/util/assert';
import { TimerId } from '../../../src/util/async_queue';
import {
  AsyncQueueImpl,
  newAsyncQueue
} from '../../../src/util/async_queue_impl';
import { newBundleReader } from '../../../src/util/bundle_reader_impl';
import { ByteString } from '../../../src/util/byte_string';
import { FirestoreError } from '../../../src/util/error';
import { logWarn } from '../../../src/util/log';
import { primitiveComparator } from '../../../src/util/misc';
import { forEach, objectSize } from '../../../src/util/obj';
import { ObjectMap } from '../../../src/util/obj_map';
import { Deferred, sequence } from '../../../src/util/promise';
import { newTestFirestore } from '../../util/api_helpers';
import {
  byteStringFromString,
  deletedDoc,
  deleteMutation,
  doc,
  filter,
  key,
  orderBy,
  patchMutation,
  query,
  setMutation,
  stringFromBase64String,
  TestSnapshotVersion,
  validateFirestoreError,
  version
} from '../../util/helpers';
import { pipelineFromStages } from '../../util/pipelines';
import { encodeWatchChange } from '../../util/spec_test_helpers';
import {
  FakeDocument,
  SharedFakeWebStorage,
  testWindow
} from '../../util/test_platform';
import {
  clearTestPersistence,
  INDEXEDDB_TEST_DATABASE_NAME,
  TEST_APP_ID,
  TEST_DATABASE_ID,
  TEST_PERSISTENCE_KEY,
  TEST_SERIALIZER
} from '../local/persistence_test_helpers';

import { MULTI_CLIENT_TAG } from './describe_spec';
import { ActiveTargetMap, ActiveTargetSpec } from './spec_builder';
import {
  EventAggregator,
  MockConnection,
  MockIndexedDbPersistence,
  MockMemoryOfflineComponentProvider,
  MockMemoryPersistence,
  MockMultiTabOfflineComponentProvider,
  MockOnlineComponentProvider,
  QueryEvent,
  SharedWriteTracker
} from './spec_test_components';

use(chaiExclude);

const ARBITRARY_SEQUENCE_NUMBER = 2;

interface DocumentOptions {
  hasLocalMutations?: boolean;
  hasCommittedMutations?: boolean;
}

export function parseQuery(querySpec: string | SpecQuery): Query {
  if (typeof querySpec === 'string') {
    return query(querySpec);
  } else {
    let query1 = querySpec.collectionGroup
      ? newQueryForCollectionGroup(querySpec.collectionGroup)
      : query(querySpec.path);
    if (querySpec.limit) {
      query1 = queryWithLimit(
        query1,
        querySpec.limit,
        querySpec.limitType === 'LimitToFirst'
          ? LimitType.First
          : LimitType.Last
      );
    }
    if (querySpec.filters) {
      querySpec.filters.forEach(([field, op, value]) => {
        query1 = queryWithAddedFilter(query1, filter(field, op, value));
      });
    }
    if (querySpec.orderBys) {
      querySpec.orderBys.forEach(([filter, direction]) => {
        query1 = queryWithAddedOrderBy(query1, orderBy(filter, direction));
      });
    }
    return query1;
  }
}

abstract class TestRunner {
  protected queue: AsyncQueueImpl;

  // Initialized asynchronously via start().
  private connection!: MockConnection;
  private eventManager!: EventManager;
  private syncEngine!: SyncEngine;

  private eventList: QueryEvent[] = [];
  private acknowledgedDocs: string[];
  private rejectedDocs: string[];
  private waitForPendingWritesEvents = 0;
  private snapshotsInSyncListeners: Array<Observer<void>>;
  private snapshotsInSyncEvents = 0;

  protected document = new FakeDocument();
  private queryListeners = new ObjectMap<QueryOrPipeline, QueryListener>(
    canonifyQueryOrPipeline,
    queryOrPipelineEqual
  );
  /**
   * Maps SDK target IDs (assigned by the spec builder based on the query) to
   * the list of remote target IDs generated by `RemoteStore`.
   *
   * In spec tests, the spec builder reuses the same SDK ID if the user unlistens
   * and listens again to the same query. However, `RemoteStore` generates a
   * brand new, unique remote ID every time a new listen is opened on the stream.
   * To track these correctly, we map the SDK ID to a list of remote IDs and use
   * a FIFO heuristic to resolve which remote ID an event (like ACK or REMOVE)
   * applies to.
   */
  private sdkToRemoteTargetIds = new Map<TargetId, RemoteTargetId[]>();
  private currentRemoteTargetIndex?: number;

  private expectedActiveLimboDocs: DocumentKey[];
  private expectedEnqueuedLimboDocs: DocumentKey[];
  private expectedActiveTargets: Map<TargetId, ActiveTargetSpec>;

  private networkEnabled = true;

  // Initialized asynchronously via start().
  private localStore!: LocalStore;
  private remoteStore!: RemoteStore;
  private persistence!: MockMemoryPersistence | MockIndexedDbPersistence;
  private lruGarbageCollector!: LruGarbageCollector;
  protected sharedClientState!: SharedClientState;

  private useEagerGCForMemory: boolean;
  private numClients: number;
  private maxConcurrentLimboResolutions?: number;
  private databaseInfo: DatabaseInfo;

  private convertToPipeline: boolean;

  protected user = User.UNAUTHENTICATED;
  protected clientId: ClientId;

  private started = false;
  private serializer: JsonProtoSerializer;

  constructor(
    private sharedWrites: SharedWriteTracker,
    clientIndex: number,
    protected config: SpecConfig
  ) {
    this.clientId = `client${clientIndex}`;
    this.databaseInfo = new DatabaseInfo(
      TEST_DATABASE_ID,
      TEST_APP_ID,
      TEST_PERSISTENCE_KEY,
      'host',
      /*ssl=*/ false,
      /*forceLongPolling=*/ false,
      /*autoDetectLongPolling=*/ false,
      /*longPollingOptions=*/ {},
      /*useFetchStreams=*/ false,
      /*isUsingEmulator=*/ false,
      'test-api-key'
    );

    // TODO(mrschmidt): During client startup in `firestore_client`, we block
    // the AsyncQueue from executing any operation. We should mimic this in the
    // setup of the spec tests.
    this.queue = newAsyncQueue() as AsyncQueueImpl;
    this.queue.skipDelaysForTimerId(TimerId.ListenStreamConnectionBackoff);

    this.serializer = new JsonProtoSerializer(
      this.databaseInfo.databaseId,
      /* useProto3Json= */ true
    );

    this.useEagerGCForMemory = config.useEagerGCForMemory;
    this.numClients = config.numClients;
    this.maxConcurrentLimboResolutions = config.maxConcurrentLimboResolutions;
    this.convertToPipeline = config.convertToPipeline ?? false;
    this.expectedActiveLimboDocs = [];
    this.expectedEnqueuedLimboDocs = [];
    this.expectedActiveTargets = new Map<TargetId, ActiveTargetSpec>();
    this.acknowledgedDocs = [];
    this.rejectedDocs = [];
    this.snapshotsInSyncListeners = [];
  }

  async start(): Promise<void> {
    const configuration = {
      asyncQueue: this.queue,
      databaseInfo: this.databaseInfo,
      authCredentials: new EmptyAuthCredentialsProvider(),
      appCheckCredentials: new EmptyAppCheckTokenProvider(),
      clientId: this.clientId,
      initialUser: this.user,
      maxConcurrentLimboResolutions:
        this.maxConcurrentLimboResolutions ?? Number.MAX_SAFE_INTEGER
    };

    this.connection = new MockConnection(this.queue);

    const onlineComponentProvider = new MockOnlineComponentProvider(
      this.connection
    );
    const offlineComponentProvider =
      await this.initializeOfflineComponentProvider(
        onlineComponentProvider,
        configuration,
        this.useEagerGCForMemory
      );
    await onlineComponentProvider.initialize(
      offlineComponentProvider,
      configuration
    );

    this.sharedClientState = offlineComponentProvider.sharedClientState;
    this.persistence = offlineComponentProvider.persistence;
    // Setting reference to lru collector for manual runs
    if (
      this.persistence.referenceDelegate instanceof MemoryLruDelegate ||
      this.persistence.referenceDelegate instanceof IndexedDbLruDelegateImpl
    ) {
      this.lruGarbageCollector =
        this.persistence.referenceDelegate.garbageCollector;
    }

    this.localStore = offlineComponentProvider.localStore;
    this.remoteStore = onlineComponentProvider.remoteStore;
    this.syncEngine = onlineComponentProvider.syncEngine;
    this.eventManager = onlineComponentProvider.eventManager;

    this.eventManager.onListen = syncEngineListen.bind(null, this.syncEngine);
    this.eventManager.onUnlisten = syncEngineUnlisten.bind(
      null,
      this.syncEngine
    );

    this.eventManager.onFirstRemoteStoreListen = triggerRemoteStoreListen.bind(
      null,
      this.syncEngine
    );
    this.eventManager.onLastRemoteStoreUnlisten =
      triggerRemoteStoreUnlisten.bind(null, this.syncEngine);

    await this.persistence.setDatabaseDeletedListener(async () => {
      await this.shutdown();
    });

    this.started = true;
  }

  protected abstract initializeOfflineComponentProvider(
    onlineComponentProvider: MockOnlineComponentProvider,
    configuration: ComponentConfiguration,
    eagerGcEnabled: boolean
  ): Promise<
    MockMultiTabOfflineComponentProvider | MockMemoryOfflineComponentProvider
  >;

  get isPrimaryClient(): boolean {
    return this.syncEngine.isPrimaryClient;
  }

  async shutdown(): Promise<void> {
    this.queue.enterRestrictedMode();
    const deferred = new Deferred();
    this.queue.enqueueAndForgetEvenWhileRestricted(async () => {
      if (this.started) {
        await this.doShutdown();
      }
      deferred.resolve();
    });
    return deferred.promise;
  }

  /** Runs a single SpecStep on this runner. */
  async run(step: SpecStep): Promise<void> {
    await this.doStep(step);
    await this.queue.drain();
    this.validateExpectedSnapshotEvents(step.expectedSnapshotEvents!);
    await this.validateExpectedState(step.expectedState!);
    this.validateSnapshotsInSyncEvents(step.expectedSnapshotsInSyncEvents);
    this.validateWaitForPendingWritesEvents(
      step.expectedWaitForPendingWritesEvents
    );
    this.eventList = [];
    this.rejectedDocs = [];
    this.acknowledgedDocs = [];
  }

  private doStep(step: SpecStep): Promise<void> {
    if ('userListen' in step) {
      return this.doListen(step.userListen!);
    } else if ('userUnlisten' in step) {
      return this.doUnlisten(step.userUnlisten!);
    } else if ('userSet' in step) {
      return this.doSet(step.userSet!);
    } else if ('userPatch' in step) {
      return this.doPatch(step.userPatch!);
    } else if ('userDelete' in step) {
      return this.doDelete(step.userDelete!);
    } else if ('addSnapshotsInSyncListener' in step) {
      return this.doAddSnapshotsInSyncListener();
    } else if ('removeSnapshotsInSyncListener' in step) {
      return this.doRemoveSnapshotsInSyncListener();
    } else if ('loadBundle' in step) {
      return this.doLoadBundle(step.loadBundle!);
    } else if ('setIndexConfiguration' in step) {
      return this.doSetIndexConfiguration(step.setIndexConfiguration!);
    } else if ('watchAck' in step) {
      return this.doWatchAck(step.watchAck!);
    } else if ('watchCurrent' in step) {
      return this.doWatchCurrent(step.watchCurrent!);
    } else if ('watchRemove' in step) {
      return this.doWatchRemove(step.watchRemove!);
    } else if ('watchEntity' in step) {
      return this.doWatchEntity(step.watchEntity!);
    } else if ('watchFilter' in step) {
      return this.doWatchFilter(step.watchFilter!);
    } else if ('watchSnapshot' in step) {
      return this.doWatchSnapshot(step.watchSnapshot!);
    } else if ('watchReset' in step) {
      return this.doWatchReset(step.watchReset!);
    } else if ('watchStreamClose' in step) {
      return this.doWatchStreamClose(step.watchStreamClose!);
    } else if ('writeAck' in step) {
      return this.doWriteAck(step.writeAck!);
    } else if ('failWrite' in step) {
      return this.doFailWrite(step.failWrite!);
    } else if ('waitForPendingWrites' in step) {
      return this.doWaitForPendingWrites();
    } else if ('runTimer' in step) {
      return this.doRunTimer(step.runTimer!);
    } else if ('drainQueue' in step) {
      return this.doDrainQueue();
    } else if ('enableNetwork' in step) {
      return step.enableNetwork!
        ? this.doEnableNetwork()
        : this.doDisableNetwork();
    } else if ('clearPersistence' in step) {
      return this.doClearPersistence();
    } else if ('restart' in step) {
      return this.doRestart();
    } else if ('shutdown' in step) {
      return this.doShutdown();
    } else if ('watchUsesTargetIndex' in step) {
      return this.doUseTargetIndex(step.watchUsesTargetIndex!);
    } else if ('applyClientState' in step) {
      // PORTING NOTE: Only used by web multi-tab tests.
      return this.doApplyClientState(step.applyClientState!);
    } else if ('triggerLruGC' in step) {
      return this.doTriggerLruGC(step.triggerLruGC!);
    } else if ('changeUser' in step) {
      return this.doChangeUser(step.changeUser!);
    } else if ('failDatabase' in step) {
      return step.failDatabase
        ? this.doFailDatabase(step.failDatabase!)
        : this.doRecoverDatabase();
    } else {
      return fail(0x6bb3, 'Unknown step: ' + JSON.stringify(step));
    }
  }

  private doUseTargetIndex(index: number | 'latest'): Promise<void> {
    this.currentRemoteTargetIndex = index === 'latest' ? undefined : index;
    return Promise.resolve();
  }

  private async doListen(listenSpec: SpecUserListen): Promise<void> {
    let targetFailed = false;

    const querySpec = listenSpec.query;
    const query =
      querySpec instanceof CorePipeline
        ? querySpec
        : this.convertToPipeline
        ? toCorePipeline(
            pipelineFromStages(
              toPipelineStages(parseQuery(querySpec), newTestFirestore())
            )
          )
        : parseQuery(querySpec);

    const aggregator = new EventAggregator(query, e => {
      if (e.error) {
        targetFailed = true;
      }
      this.pushEvent(e);
    });

    const options = {
      includeMetadataChanges:
        listenSpec.options?.includeMetadataChanges ?? true,
      waitForSyncWhenOnline: false,
      source: listenSpec.options?.source ?? Source.Default
    };

    const queryListener = new QueryListener(query, aggregator, options);
    this.queryListeners.set(query, queryListener);

    await this.queue.enqueue(() =>
      eventManagerListen(this.eventManager, queryListener)
    );

    const sdkTargetId = listenSpec.targetId;

    // The remote store will map the query listener to a new remoteTargetID
    const remoteTargetId = remoteStoreGetRemoteTargetId(
      this.remoteStore,
      sdkTargetId
    );
    if (remoteTargetId !== undefined) {
      // Add this new remoteTargetId to a list mapping the sdk target ID to the
      // remote target ID
      const list = this.sdkToRemoteTargetIds.get(sdkTargetId) || [];
      list.push(remoteTargetId);
      this.sdkToRemoteTargetIds.set(sdkTargetId, list);
    }

    if (targetFailed) {
      expect(this.persistence.injectFailures).contains('Allocate target');
    } else {
      // Skip the backoff that may have been triggered by a previous call to
      // `watchStreamCloses()`.
      if (
        this.queue.containsDelayedOperation(
          TimerId.ListenStreamConnectionBackoff
        )
      ) {
        await this.queue.runAllDelayedOperationsUntil(
          TimerId.ListenStreamConnectionBackoff
        );
      }

      if (
        this.isPrimaryClient &&
        this.networkEnabled &&
        options.source !== Source.Cache
      ) {
        // Unless listened to cache, open always have happened after a listen.
        await this.connection.waitForWatchOpen();
      }
    }
  }

  private async doUnlisten(listenSpec: SpecUserUnlisten): Promise<void> {
    // TODO(dimond): make sure correct target IDs are assigned
    // let targetId = listenSpec[0];
    const querySpec = listenSpec[1];
    const query =
      querySpec instanceof CorePipeline
        ? querySpec
        : this.convertToPipeline
        ? toCorePipeline(
            pipelineFromStages(
              toPipelineStages(parseQuery(querySpec), newTestFirestore())
            )
          )
        : parseQuery(querySpec);
    const eventEmitter = this.queryListeners.get(query);
    debugAssert(!!eventEmitter, 'There must be a query to unlisten too!');
    this.queryListeners.delete(query);
    await this.queue.enqueue(() =>
      eventManagerUnlisten(this.eventManager, eventEmitter!)
    );
  }

  private doSet(setSpec: SpecUserSet): Promise<void> {
    return this.doMutations([setMutation(setSpec[0], setSpec[1])]);
  }

  private doPatch(patchSpec: SpecUserPatch): Promise<void> {
    return this.doMutations([patchMutation(patchSpec[0], patchSpec[1])]);
  }

  private doDelete(deleteSpec: SpecUserDelete): Promise<void> {
    const key: string = deleteSpec;
    return this.doMutations([deleteMutation(key)]);
  }

  private doAddSnapshotsInSyncListener(): Promise<void> {
    const observer = {
      next: () => {
        this.snapshotsInSyncEvents += 1;
      },
      error: () => {}
    };
    this.snapshotsInSyncListeners.push(observer);
    addSnapshotsInSyncListener(this.eventManager, observer);
    return Promise.resolve();
  }

  private doRemoveSnapshotsInSyncListener(): Promise<void> {
    const removeObs = this.snapshotsInSyncListeners.pop();
    if (removeObs) {
      removeSnapshotsInSyncListener(this.eventManager, removeObs);
    } else {
      throw new Error('There must be a listener to unlisten to');
    }
    return Promise.resolve();
  }

  private async doLoadBundle(bundle: string): Promise<void> {
    const reader = newBundleReader(
      toByteStreamReader(newTextEncoder().encode(bundle)),
      this.serializer
    );
    const task = new LoadBundleTask();
    return this.queue.enqueue(async () => {
      syncEngineLoadBundle(this.syncEngine, reader, task);
      await task.catch(e => {
        logWarn(`Loading bundle failed with ${e}`);
      });
    });
  }

  private async doSetIndexConfiguration(
    jsonOrConfiguration: string | IndexConfiguration
  ): Promise<void> {
    return this.queue.enqueue(async () => {
      const parsedIndexes = parseIndexes(jsonOrConfiguration);
      return localStoreConfigureFieldIndexes(this.localStore, parsedIndexes);
    });
  }

  private doMutations(mutations: Mutation[]): Promise<void> {
    const documentKeys = mutations.map(val => val.key.path.toString());
    const syncEngineCallback = new Deferred<void>();
    syncEngineCallback.promise.then(
      () => this.acknowledgedDocs.push(...documentKeys),
      () => this.rejectedDocs.push(...documentKeys)
    );

    if (
      this.persistence.injectFailures.indexOf('Locally write mutations') === -1
    ) {
      this.sharedWrites.push(mutations);
    }

    return this.queue.enqueue(() =>
      syncEngineWrite(this.syncEngine, mutations, syncEngineCallback)
    );
  }

  /**
   * Resolves an SDK target ID to the appropriate remote target ID.
   *
   * Because multiple remote target IDs can be associated with a single SDK target ID over time,
   * this method allows the caller to specify the index of the remote target ID
   * to lookup for the SDK target ID. If sdkTargetIndex is undefined, then
   * the latest remote target ID is used.
   */
  private getRemoteTargetId(
    sdkTargetId: TargetId,
    isWatchStep: boolean = false
  ): RemoteTargetId {
    // Get and update the list with the current mapping from RemoteStore
    // if the remote target ID is not already in the sdkToRemoteTargetIds list
    const remoteStoreId = remoteStoreGetRemoteTargetId(
      this.remoteStore,
      sdkTargetId
    );
    const list = this.sdkToRemoteTargetIds.get(sdkTargetId) || [];
    if (remoteStoreId !== undefined && !list.includes(remoteStoreId)) {
      console.warn(
        `The sdk target ID ${sdkTargetId} was mapped to remote target id ${remoteStoreId} but was not found in the TestRunner sdkToRemoteTargetIds list ${this.sdkToRemoteTargetIds}`
      );
      list.push(remoteStoreId);
      this.sdkToRemoteTargetIds.set(sdkTargetId, list);
    }

    if (list.length === 0) {
      console.warn(
        `Mapping not found for SDK ID ${JSON.stringify(sdkTargetId)}`
      );
      return sdkTargetId as RemoteTargetId;
    }

    if (isWatchStep && this.currentRemoteTargetIndex !== undefined) {
      return list[this.currentRemoteTargetIndex];
    }

    // Default: return the latest one.
    return list[list.length - 1];
  }

  private doWatchAck(ackedTargets: SpecWatchAck): Promise<void> {
    const remoteTargetIds = ackedTargets.map(ackedTarget =>
      this.getRemoteTargetId(ackedTarget, true)
    );
    const change = new WatchTargetChange(
      WatchTargetChangeState.Added,
      remoteTargetIds
    );
    return this.doWatchEvent(change);
  }

  private doWatchCurrent(currentTargets: SpecWatchCurrent): Promise<void> {
    const targets = currentTargets[0];
    const remoteTargetIds = targets.map(id => this.getRemoteTargetId(id, true));
    const resumeToken = byteStringFromString(currentTargets[1]);
    const change = new WatchTargetChange(
      WatchTargetChangeState.Current,
      remoteTargetIds,
      resumeToken
    );
    return this.doWatchEvent(change);
  }

  private doWatchReset(targetIds: SpecWatchReset): Promise<void> {
    const remoteTargetIds = targetIds.map(id =>
      this.getRemoteTargetId(id, true)
    );
    const change = new WatchTargetChange(
      WatchTargetChangeState.Reset,
      remoteTargetIds
    );
    return this.doWatchEvent(change);
  }

  private doWatchRemove(removed: SpecWatchRemove): Promise<void> {
    const cause =
      removed.cause &&
      new FirestoreError(
        mapCodeFromRpcCode(removed.cause.code),
        removed.cause.message
      );
    const mappedIds = new Map<TargetId, RemoteTargetId>();
    const remoteTargetIds = removed.targetIds.map(id => {
      const remoteId = this.getRemoteTargetId(id, true);
      mappedIds.set(id, remoteId);
      return remoteId;
    });
    const change = new WatchTargetChange(
      WatchTargetChangeState.Removed,
      remoteTargetIds,
      ByteString.EMPTY_BYTE_STRING,
      cause || null
    );
    if (cause) {
      // Make sure that the target is active and can be removed.
      // Technically removing an unknown target is valid (e.g. it could race
      // with a server-side removal), but we want to pay extra careful
      // attention in tests that we only remove targets we listened too.
      removed.targetIds.forEach(targetId => {
        const remoteTargetId = mappedIds.get(targetId)!;
        if (!this.config.allowUnlistedTargetRemoval) {
          expect(
            this.connection.activeTargets[remoteTargetId],
            'Removing a non-active target'
          ).to.exist;
        }
        delete this.connection.activeTargets[remoteTargetId];
      });
    }

    return this.doWatchEvent(change);
  }

  private doWatchEntity(watchEntity: SpecWatchEntity): Promise<void> {
    const remoteTargets = (watchEntity.targets || []).map(id =>
      this.getRemoteTargetId(id, true)
    );
    const remoteRemovedTargets = (watchEntity.removedTargets || []).map(id =>
      this.getRemoteTargetId(id, true)
    );

    if (watchEntity.docs) {
      debugAssert(
        !watchEntity.doc,
        'Exactly one of `doc` or `docs` needs to be set'
      );
      return sequence(watchEntity.docs, (specDocument: SpecDocument) => {
        return this.doWatchEntity({
          doc: specDocument,
          targets: watchEntity.targets,
          removedTargets: watchEntity.removedTargets
        });
      });
    } else if (watchEntity.doc) {
      const document = watchEntity.doc.value
        ? doc(
            watchEntity.doc.key,
            watchEntity.doc.version,
            watchEntity.doc.value,
            watchEntity.doc.createTime
          )
        : deletedDoc(watchEntity.doc.key, watchEntity.doc.version);
      if (watchEntity.doc.options?.hasCommittedMutations) {
        document.setHasCommittedMutations();
      } else if (watchEntity.doc.options?.hasLocalMutations) {
        document.setHasLocalMutations();
      }
      const change = new DocumentWatchChange(
        remoteTargets,
        remoteRemovedTargets,
        document.key,
        document
      );
      return this.doWatchEvent(change);
    } else if (watchEntity.key) {
      const documentKey = key(watchEntity.key);
      const change = new DocumentWatchChange(
        remoteTargets,
        remoteRemovedTargets,
        documentKey,
        null
      );
      return this.doWatchEvent(change);
    } else {
      return fail(0xdcc3, 'Either doc or docs must be set');
    }
  }

  private doWatchFilter(watchFilter: SpecWatchFilter): Promise<void> {
    const { targetIds, keys, bloomFilter } = watchFilter;
    debugAssert(
      targetIds.length === 1,
      'ExistenceFilters currently support exactly one target only.'
    );
    const remoteTargetId = this.getRemoteTargetId(targetIds[0], true);
    const filter = new ExistenceFilter(keys.length, bloomFilter);
    const change = new ExistenceFilterChange(remoteTargetId, filter);
    return this.doWatchEvent(change);
  }

  private doWatchSnapshot(watchSnapshot: SpecWatchSnapshot): Promise<void> {
    // The client will only respond to watchSnapshots if they are on a target
    // change with an empty set of target IDs. So we should be sure to send a
    // separate event.
    const remoteTargetIds = watchSnapshot.targetIds.map(id =>
      this.getRemoteTargetId(id, true)
    );
    const protoJSON: api.ListenResponse = {
      targetChange: {
        readTime: toVersion(this.serializer, version(watchSnapshot.version)),
        // Convert to base64 string so it can later be parsed into ByteString.
        resumeToken: encodeBase64(watchSnapshot.resumeToken || ''),
        targetIds: remoteTargetIds
      }
    };
    this.connection.watchStream!.callOnMessage(protoJSON);

    // Put a no-op in the queue so that we know when any outstanding RemoteStore
    // writes on the network are complete.
    return this.queue.enqueue(async () => {});
  }

  private async doWatchEvent(watchChange: WatchChange): Promise<void> {
    const protoJSON = encodeWatchChange(watchChange);
    this.connection.watchStream!.callOnMessage(protoJSON);

    // Put a no-op in the queue so that we know when any outstanding RemoteStore
    // writes on the network are complete.
    return this.queue.enqueue(async () => {});
  }

  private async doWatchStreamClose(spec: SpecWatchStreamClose): Promise<void> {
    this.connection.failWatchStream(
      new FirestoreError(
        mapCodeFromRpcCode(spec.error.code),
        spec.error.message
      )
    );
    // The watch stream should re-open if we have active listeners.
    if (spec.runBackoffTimer && !this.queryListeners.isEmpty()) {
      await this.queue.runAllDelayedOperationsUntil(
        TimerId.ListenStreamConnectionBackoff
      );
      await this.connection.waitForWatchOpen();
    }
  }

  /** Validates that a write was sent and matches the expected write. */
  private validateNextWriteRequest(mutations: Mutation[]): Promise<void> {
    // Make sure this write was sent on the wire and it matches the expected
    // write.
    return this.connection.waitForWriteRequest().then(request => {
      const writes = request.writes!;
      expect(writes.length).to.equal(mutations.length);
      for (let i = 0; i < writes.length; ++i) {
        expect(writes[i]).to.deep.equal(
          toMutation(this.serializer, mutations[i])
        );
      }
    });
  }

  private doWriteAck(writeAck: SpecWriteAck): Promise<void> {
    const updateTime = toVersion(this.serializer, version(writeAck.version));
    const nextMutation = writeAck.keepInQueue
      ? this.sharedWrites.peek()
      : this.sharedWrites.shift();
    return this.validateNextWriteRequest(nextMutation).then(() => {
      this.connection.ackWrite(updateTime, [{ updateTime }]);
    });
  }

  private async doFailWrite(writeFailure: SpecWriteFailure): Promise<void> {
    const specError: SpecError = writeFailure.error;
    const error = new FirestoreError(
      mapCodeFromRpcCode(specError.code),
      specError.message
    );
    const nextMutation = writeFailure.keepInQueue
      ? this.sharedWrites.peek()
      : this.sharedWrites.shift();
    return this.validateNextWriteRequest(nextMutation).then(() => {
      this.connection.failWrite(error);
    });
  }

  private async doWaitForPendingWrites(): Promise<void> {
    const deferred = new Deferred();
    void deferred.promise.then(() => ++this.waitForPendingWritesEvents);
    return this.queue.enqueue(() =>
      syncEngineRegisterPendingWritesCallback(this.syncEngine, deferred)
    );
  }

  private async doRunTimer(timer: string): Promise<void> {
    // We assume the timer string is a valid TimerID enum value, but if it's
    // not, then there won't be a matching item on the queue and
    // runDelayedOperationsEarly() will throw.
    const timerId = timer as TimerId;
    await this.queue.runAllDelayedOperationsUntil(timerId);
  }

  private async doDisableNetwork(): Promise<void> {
    this.networkEnabled = false;
    // Make sure to execute all writes that are currently queued. This allows us
    // to assert on the total number of requests sent before shutdown.
    await fillWritePipeline(this.remoteStore);
    this.persistence.setNetworkEnabled(false);
    await remoteStoreDisableNetwork(this.remoteStore);
  }

  private async doDrainQueue(): Promise<void> {
    await this.queue.drain();
  }

  private async doEnableNetwork(): Promise<void> {
    this.networkEnabled = true;
    this.persistence.setNetworkEnabled(true);
    await remoteStoreEnableNetwork(this.remoteStore);
  }

  private async doShutdown(): Promise<void> {
    await remoteStoreShutdown(this.remoteStore);
    await this.sharedClientState.shutdown();
    // We don't delete the persisted data here since multi-clients may still
    // be accessing it. Instead, we manually remove it at the end of the
    // test run.
    await this.persistence.shutdown();
    this.started = false;
  }

  private async doClearPersistence(): Promise<void> {
    await clearTestPersistence();
  }

  private async doRestart(): Promise<void> {
    // Reinitialize everything.
    await this.doShutdown();
    this.sdkToRemoteTargetIds.clear();
    this.currentRemoteTargetIndex = undefined;

    // We have to schedule the starts, otherwise we could end up with
    // interleaved events.
    await this.queue.enqueue(() => this.start());
  }

  private async doApplyClientState(state: SpecClientState): Promise<void> {
    if (state.visibility) {
      this.document.raiseVisibilityEvent(state.visibility!);
    }

    if (state.primary) {
      await clearCurrentPrimaryLease();
      await this.queue.runAllDelayedOperationsUntil(
        TimerId.ClientMetadataRefresh
      );
    }

    return Promise.resolve();
  }

  private async doChangeUser(user: string | null): Promise<void> {
    this.user = new User(user);
    // We don't block on `handleCredentialChange` as it may not get executed
    // during an IndexedDb failure. Non-recovery tests will pick up the user
    // change when the AsyncQueue is drained.
    this.queue.enqueueRetryable(() =>
      remoteStoreHandleCredentialChange(this.remoteStore, new User(user))
    );
  }

  private async doTriggerLruGC(cacheThreshold: number): Promise<void> {
    return this.queue.enqueue(async () => {
      if (!!this.lruGarbageCollector) {
        const params = this.lruGarbageCollector.params as {
          cacheSizeCollectionThreshold: number;
          percentileToCollect: number;
        };
        params.cacheSizeCollectionThreshold = cacheThreshold;
        params.percentileToCollect = 100;
        await this.localStore.collectGarbage(this.lruGarbageCollector);
      }
    });
  }

  private async doFailDatabase(
    failActions: PersistenceAction[]
  ): Promise<void> {
    this.persistence.injectFailures = failActions;
  }

  private async doRecoverDatabase(): Promise<void> {
    this.persistence.injectFailures = [];
  }

  private validateExpectedSnapshotEvents(
    expectedEvents: SnapshotEvent[]
  ): void {
    if (expectedEvents) {
      expect(this.eventList.length).to.equal(
        expectedEvents.length,
        'Number of expected and actual events mismatch'
      );
      const actualEventsSorted: QueryEvent[] = this.eventList.sort((a, b) =>
        primitiveComparator(
          canonifyQueryOrPipeline(a.query),
          canonifyQueryOrPipeline(b.query)
        )
      );

      let expectedEventsSorted: SnapshotEvent[] = [];
      if (this.convertToPipeline) {
        expectedEventsSorted = expectedEvents.sort((a, b) =>
          primitiveComparator(
            canonifyQueryOrPipeline(
              toCorePipeline(
                pipelineFromStages(
                  toPipelineStages(
                    parseQuery(a.query as SpecQuery),
                    newTestFirestore()
                  )
                )
              )
            ),
            canonifyQueryOrPipeline(
              toCorePipeline(
                pipelineFromStages(
                  toPipelineStages(
                    parseQuery(b.query as SpecQuery),
                    newTestFirestore()
                  )
                )
              )
            )
          )
        );
      } else {
        expectedEventsSorted = expectedEvents.sort((a, b) =>
          primitiveComparator(
            canonifyQueryOrPipeline(parseQuery(a.query as SpecQuery)),
            canonifyQueryOrPipeline(parseQuery(b.query as SpecQuery))
          )
        );
      }

      for (let i = 0; i < expectedEventsSorted.length; i++) {
        const actual = actualEventsSorted[i];
        const expected = expectedEventsSorted[i];
        this.validateWatchExpectation(expected, actual);
      }
    } else {
      expect(this.eventList.length).to.equal(
        0,
        'Unexpected events: ' + JSON.stringify(this.eventList, null, 2)
      );
    }
  }

  private async validateExpectedState(
    expectedState: StateExpectation
  ): Promise<void> {
    if (expectedState) {
      if ('numOutstandingWrites' in expectedState) {
        expect(outstandingWrites(this.remoteStore)).to.equal(
          expectedState.numOutstandingWrites
        );
      }
      if ('numActiveClients' in expectedState) {
        debugAssert(
          this.persistence instanceof IndexedDbPersistence,
          'numActiveClients() requires IndexedDbPersistence'
        );
        const activeClients = await this.persistence.getActiveClients();
        expect(activeClients.length).to.equal(expectedState.numActiveClients);
      }
      if ('writeStreamRequestCount' in expectedState) {
        expect(this.connection.writeStreamRequestCount).to.equal(
          expectedState.writeStreamRequestCount
        );
      }
      if ('watchStreamRequestCount' in expectedState) {
        expect(this.connection.watchStreamRequestCount).to.equal(
          expectedState.watchStreamRequestCount
        );
      }
      if ('activeLimboDocs' in expectedState) {
        this.expectedActiveLimboDocs = expectedState.activeLimboDocs!.map(key);
      }
      if ('enqueuedLimboDocs' in expectedState) {
        this.expectedEnqueuedLimboDocs =
          expectedState.enqueuedLimboDocs!.map(key);
      }
      if ('activeTargets' in expectedState) {
        this.expectedActiveTargets.clear();
        forEach(expectedState.activeTargets!, (key, value) => {
          this.expectedActiveTargets.set(Number(key), value);
        });
      }
      if ('isPrimary' in expectedState) {
        expect(this.isPrimaryClient).to.eq(
          expectedState.isPrimary!,
          'isPrimary'
        );
      }
      if ('isShutdown' in expectedState) {
        expect(this.started).to.equal(!expectedState.isShutdown);
      }
      if ('indexes' in expectedState) {
        const fieldIndexes: FieldIndex[] =
          await this.persistence.runTransaction(
            'getFieldIndexes ',
            'readonly',
            transaction =>
              this.localStore.indexManager.getFieldIndexes(transaction)
          );

        assert.deepEqualExcluding(
          fieldIndexes,
          expectedState.indexes!,
          'indexId'
        );
      }
    }

    if (expectedState && expectedState.userCallbacks) {
      expect(this.acknowledgedDocs).to.have.members(
        expectedState.userCallbacks.acknowledgedDocs
      );
      expect(this.rejectedDocs).to.have.members(
        expectedState.userCallbacks.rejectedDocs
      );
    } else {
      expect(this.acknowledgedDocs).to.be.empty;
      expect(this.rejectedDocs).to.be.empty;
    }

    if (this.numClients === 1) {
      expect(this.isPrimaryClient).to.eq(true, 'isPrimary');
    }

    // Clients don't reset their limbo docs on shutdown, so any validation will
    // likely fail.
    if (this.started) {
      // Always validate that the expected limbo docs match the actual limbo
      // docs
      this.validateActiveLimboDocs();
      this.validateEnqueuedLimboDocs();
      // Always validate that the expected active targets match the actual
      // active targets
      await this.validateActiveTargets();
    }
  }

  private validateWaitForPendingWritesEvents(
    expectedCount: number | undefined
  ): void {
    expect(this.waitForPendingWritesEvents).to.eq(
      expectedCount || 0,
      'for waitForPendingWritesEvents'
    );
    this.waitForPendingWritesEvents = 0;
  }

  private validateSnapshotsInSyncEvents(
    expectedCount: number | undefined
  ): void {
    expect(this.snapshotsInSyncEvents).to.eq(
      expectedCount || 0,
      'for snapshotsInSyncEvents'
    );
    this.snapshotsInSyncEvents = 0;
  }

  private validateActiveLimboDocs(): void {
    let actualLimboDocs = syncEngineGetActiveLimboDocumentResolutions(
      this.syncEngine
    );

    if (this.connection.isWatchOpen) {
      // Validate that each active limbo doc has an expected active target
      actualLimboDocs.forEach((key, targetId) => {
        const targetIds = new Array(this.expectedActiveTargets.keys()).map(
          n => '' + n
        );
        expect(this.expectedActiveTargets.has(targetId)).to.equal(
          true,
          `Found limbo doc ${key.toString()}, but its target ID ${targetId} ` +
            `was not in the set of expected active target IDs ` +
            `(${targetIds.join(', ')})`
        );
      });
    }

    for (const expectedLimboDoc of this.expectedActiveLimboDocs) {
      expect(actualLimboDocs.get(expectedLimboDoc)).to.not.equal(
        null,
        'Expected doc to be in limbo, but was not: ' +
          expectedLimboDoc.toString()
      );
      actualLimboDocs = actualLimboDocs.remove(expectedLimboDoc);
    }
    expect(actualLimboDocs.size).to.equal(
      0,
      'Unexpected active docs in limbo: ' + actualLimboDocs.toString()
    );
  }

  private validateEnqueuedLimboDocs(): void {
    const actualLimboDocs = Array.from(
      syncEngineGetEnqueuedLimboDocumentResolutions(this.syncEngine)
    );
    const expectedLimboDocs = Array.from(this.expectedEnqueuedLimboDocs, key =>
      key.path.canonicalString()
    );
    expect(actualLimboDocs).to.have.members(
      expectedLimboDocs,
      'The set of enqueued limbo documents is incorrect'
    );
  }

  private async validateActiveTargets(): Promise<void> {
    if (!this.isPrimaryClient || !this.networkEnabled) {
      expect(this.connection.activeTargets).to.be.empty;
      return;
    }

    // In multi-tab mode, we cannot rely on the `waitForWatchOpen` call in
    // `doUserListen` since primary tabs may execute queries from other tabs
    // without any direct user interaction.

    // TODO(mrschmidt): Refactor so this is only executed after primary tab
    // change
    if (this.expectedActiveTargets.size > 0) {
      await this.connection.waitForWatchOpen();
      await this.queue.drain();
    }

    const actualTargets = { ...this.connection.activeTargets };
    this.expectedActiveTargets.forEach((expected, sdkTargetId) => {
      const remoteTargetId = this.getRemoteTargetId(sdkTargetId);
      expect(actualTargets[remoteTargetId]).to.not.equal(
        undefined,
        'Expected active target not found: ' + JSON.stringify(expected)
      );
      const { target: actualTarget, labels: actualLabels } =
        actualTargets[remoteTargetId];

      let targetData = new TargetData(
        this.specToTarget(expected.queries[0]),
        remoteTargetId,
        expected.targetPurpose ?? TargetPurpose.Listen,
        ARBITRARY_SEQUENCE_NUMBER
      );
      if (expected.resumeToken && expected.resumeToken !== '') {
        targetData = targetData.withResumeToken(
          byteStringFromString(expected.resumeToken),
          SnapshotVersion.min()
        );
      } else {
        targetData = targetData.withResumeToken(
          ByteString.EMPTY_BYTE_STRING,
          version(expected.readTime!)
        );
      }
      if (expected.expectedCount !== undefined) {
        targetData = targetData.withExpectedCount(expected.expectedCount);
      }

      const expectedLabels =
        toListenRequestLabels(this.serializer, targetData) ?? undefined;
      expect(actualLabels).to.deep.equal(expectedLabels);

      let expectedTarget: api.Target;
      if (
        (this.convertToPipeline || targetIsPipelineTarget(targetData.target)) &&
        targetData.purpose !== TargetPurpose.LimboResolution
      ) {
        expectedTarget = toTarget(
          this.serializer,
          new TargetData(
            targetData.target,
            targetData.targetId,
            targetData.purpose,
            targetData.sequenceNumber,
            targetData.snapshotVersion,
            targetData.lastLimboFreeSnapshotVersion,
            targetData.resumeToken
          )
        );
        const actualPipeline = fromPipelineTarget(
          actualTarget.pipelineQuery!,
          this.serializer
        );

        if (targetIsPipelineTarget(targetData.target)) {
          expect(queryOrPipelineEqual(actualPipeline, targetData.target)).to.be
            .true;
        } else {
          const expectedQuery = new QueryImpl(
            targetData.target.path,
            targetData.target.collectionGroup,
            targetData.target.orderBy,
            targetData.target.filters,
            targetData.target.limit,
            LimitType.First,
            targetData.target.startAt,
            targetData.target.endAt
          );
          const expectedPipeline = toCorePipeline(
            pipelineFromStages(
              toPipelineStages(expectedQuery, newTestFirestore())
            )
          );
          expect(queryOrPipelineEqual(actualPipeline, expectedPipeline)).to.be
            .true;
        }
      } else {
        expectedTarget = toTarget(this.serializer, targetData);
        expect(actualTarget.query).to.deep.equal(expectedTarget.query);
      }

      expect(actualTarget.targetId).to.equal(expectedTarget.targetId);
      expect(actualTarget.readTime).to.equal(expectedTarget.readTime);
      expect(actualTarget.resumeToken).to.equal(
        expectedTarget.resumeToken,
        `ResumeToken does not match - expected:
         ${stringFromBase64String(
           expectedTarget.resumeToken
         )}, actual: ${stringFromBase64String(actualTarget.resumeToken)}`
      );
      if (expected.expectedCount !== undefined) {
        expect(actualTarget.expectedCount).to.equal(
          expectedTarget.expectedCount
        );
      }
      delete actualTargets[remoteTargetId];
    });
    expect(objectSize(actualTargets)).to.equal(
      0,
      'Unexpected active targets: ' + JSON.stringify(actualTargets)
    );
  }

  private specToTarget(spec: SpecQuery | CorePipeline): TargetOrPipeline {
    if (spec instanceof CorePipeline) {
      return spec;
    }
    return queryToTarget(parseQuery(spec));
  }

  private validateWatchExpectation(
    expected: SnapshotEvent,
    actual: QueryEvent
  ): void {
    if (this.convertToPipeline) {
      expect(
        pipelineEq(
          toCorePipeline(
            pipelineFromStages(
              toPipelineStages(parseQuery(expected.query), newTestFirestore())
            )
          ),
          actual.query as CorePipeline
        )
      );
    } else {
      expect(parseQuery(expected.query)).to.deep.equal(actual.query);
    }

    if (expected.errorCode) {
      validateFirestoreError(
        mapCodeFromRpcCode(expected.errorCode),
        actual.error!
      );
    } else {
      const expectedChanges: DocumentViewChange[] = [];
      if (expected.removed) {
        expected.removed.forEach(change => {
          expectedChanges.push(this.parseChange(ChangeType.Removed, change));
        });
      }
      if (expected.added) {
        expected.added.forEach(change => {
          expectedChanges.push(this.parseChange(ChangeType.Added, change));
        });
      }
      if (expected.modified) {
        expected.modified.forEach(change => {
          expectedChanges.push(this.parseChange(ChangeType.Modified, change));
        });
      }

      if (expected.metadata) {
        expected.metadata.forEach(change => {
          expectedChanges.push(this.parseChange(ChangeType.Metadata, change));
        });
      }

      const actualChangesSorted = Array.from(actual.view!.docChanges).sort(
        (a, b) => primitiveComparator(a.doc, b.doc)
      );
      const expectedChangesSorted = Array.from(expectedChanges).sort((a, b) =>
        primitiveComparator(a.doc, b.doc)
      );
      expect(actualChangesSorted).to.deep.equal(expectedChangesSorted);

      expect(actual.view!.hasPendingWrites).to.equal(
        expected.hasPendingWrites,
        'hasPendingWrites'
      );
      expect(actual.view!.fromCache).to.equal(expected.fromCache, 'fromCache');

      if (actual && !expected) {
        expect(expected, 'Got an actual event without expecting one').to.be.ok;
      }
    }
  }

  private pushEvent(e: QueryEvent): void {
    this.eventList.push(e);
  }

  private parseChange(
    type: ChangeType,
    change: SpecDocument
  ): DocumentViewChange {
    const document = doc(
      change.key,
      change.version,
      change.value || {},
      change.createTime
    );
    if (change.options?.hasCommittedMutations) {
      document.setHasCommittedMutations();
    } else if (change.options?.hasLocalMutations) {
      document.setHasLocalMutations();
    }
    return {
      type,
      doc: document
    };
  }
}

class MemoryTestRunner extends TestRunner {
  protected async initializeOfflineComponentProvider(
    onlineComponentProvider: MockOnlineComponentProvider,
    configuration: ComponentConfiguration,
    eagerGCEnabled: boolean
  ): Promise<MockMemoryOfflineComponentProvider> {
    const componentProvider = new MockMemoryOfflineComponentProvider(
      eagerGCEnabled
    );
    await componentProvider.initialize(configuration);
    return componentProvider;
  }
}

/**
 * Runs the specs using IndexedDbPersistence, the creator must ensure that it is
 * enabled for the platform.
 */
class IndexedDbTestRunner extends TestRunner {
  constructor(
    sharedWrites: SharedWriteTracker,
    private sharedFakeWebStorage: SharedFakeWebStorage,
    clientIndex: number,
    config: SpecConfig
  ) {
    super(sharedWrites, clientIndex, config);
  }

  protected async initializeOfflineComponentProvider(
    onlineComponentProvider: MockOnlineComponentProvider,
    configuration: ComponentConfiguration,
    _: boolean
  ): Promise<MockMultiTabOfflineComponentProvider> {
    const offlineComponentProvider = new MockMultiTabOfflineComponentProvider(
      testWindow(this.sharedFakeWebStorage),
      this.document,
      onlineComponentProvider
    );
    await offlineComponentProvider.initialize(configuration);
    return offlineComponentProvider;
  }

  static destroyPersistence(): Promise<void> {
    return SimpleDb.delete(INDEXEDDB_TEST_DATABASE_NAME);
  }
}

/**
 * Runs a spec test case.
 *
 * The spec consists of an array of individual steps to run in sequence.
 */
export async function runSpec(
  name: string,
  tags: string[],
  usePersistence: boolean,
  config: SpecConfig,
  steps: SpecStep[]
): Promise<void> {
  const sharedMockStorage = new SharedFakeWebStorage();

  // PORTING NOTE: Non multi-client SDKs only support a single test runner.
  const runners: TestRunner[] = [];
  const outstandingMutations = new SharedWriteTracker();

  const ensureRunner = async (clientIndex: number): Promise<TestRunner> => {
    if (!runners[clientIndex]) {
      if (usePersistence) {
        runners[clientIndex] = new IndexedDbTestRunner(
          outstandingMutations,
          sharedMockStorage,
          clientIndex,
          config
        );
      } else {
        runners[clientIndex] = new MemoryTestRunner(
          outstandingMutations,
          clientIndex,
          config
        );
      }
      await runners[clientIndex].start();
    }
    return runners[clientIndex];
  };

  let lastStep: SpecStep | null = null;
  let count = 0;
  try {
    await sequence(steps, async step => {
      debugAssert(
        step.clientIndex === undefined || tags.indexOf(MULTI_CLIENT_TAG) !== -1,
        "Cannot use 'client()' to initialize a test that is not tagged with " +
          "'multi-client'. Did you mean to use 'spec()'?"
      );

      ++count;
      lastStep = step;
      return ensureRunner(step.clientIndex || 0).then(runner =>
        runner.run(step)
      );
    });
  } catch (err) {
    console.warn(
      `Spec test failed at step ${count}: ${JSON.stringify(lastStep, null, 2)}`
    );
    throw err;
  } finally {
    for (const runner of runners) {
      await runner.shutdown();
    }
    if (usePersistence) {
      await IndexedDbTestRunner.destroyPersistence();
    }
  }
}

/** Specifies initial configuration information for the test. */
export interface SpecConfig {
  /** A boolean to enable / disable eager GC for memory persistence. */
  useEagerGCForMemory: boolean;

  /** The number of active clients for this test run. */
  numClients: number;

  /** A boolean to allow removal of non-active targets by the server. */
  allowUnlistedTargetRemoval?: boolean;

  /**
   * The maximum number of concurrently-active listens for limbo resolutions.
   * This value must be strictly greater than zero, or undefined to use the
   * default value.
   */
  maxConcurrentLimboResolutions?: number;

  convertToPipeline?: boolean;
}

/**
 * The cumulative list of actions run against Persistence. This is used by the
 * Spec tests to fail specific types of actions.
 */
export type PersistenceAction =
  | 'Get next mutation batch'
  | 'read document'
  | 'Allocate target'
  | 'Release target'
  | 'Execute query'
  | 'Handle user change'
  | 'Locally write mutations'
  | 'Acknowledge batch'
  | 'Reject batch'
  | 'Get highest unacknowledged batch id'
  | 'Get last stream token'
  | 'Set last stream token'
  | 'Get last remote snapshot version'
  | 'Set last remote snapshot version'
  | 'Apply remote event'
  | 'notifyLocalViewChanges'
  | 'Remote document keys'
  | 'Collect garbage'
  | 'maybeGarbageCollectMultiClientState'
  | 'Lookup mutation documents'
  | 'Get target data'
  | 'Get new document changes'
  | 'Synchronize last document change read time'
  | 'updateClientMetadataAndTryBecomePrimary'
  | 'getHighestListenSequenceNumber';

/**
 * Union type for each step. The step consists of exactly one `field`
 * set and optionally expected events in the `expect` field.
 */
export interface SpecStep {
  /** The index of the local client for multi-client spec tests. */
  clientIndex?: number; // PORTING NOTE: Only used by web multi-tab tests
  /** Listen to a new query (must be unique) */
  userListen?: SpecUserListen;
  /** Unlisten from a query (must be listened to) */
  userUnlisten?: SpecUserUnlisten;
  /** Perform a user initiated set */
  userSet?: SpecUserSet;
  /** Perform a user initiated patch */
  userPatch?: SpecUserPatch;
  /** Perform a user initiated delete */
  userDelete?: SpecUserDelete;
  /** Listens to a SnapshotsInSync event. */
  addSnapshotsInSyncListener?: true;
  /** Unlistens from a SnapshotsInSync event. */
  removeSnapshotsInSyncListener?: true;
  /** Loads a bundle from a string. */
  loadBundle?: string;

  /** Ack for a query in the watch stream */
  watchAck?: SpecWatchAck;
  /** Marks the query results as current */
  watchCurrent?: SpecWatchCurrent;
  /** Reset the results of a query */
  watchReset?: SpecWatchReset;
  /** Ack for remove or rejection of a query in the watch stream */
  watchRemove?: SpecWatchRemove;
  /** Document update in the watch stream */
  watchEntity?: SpecWatchEntity;
  /** Existence filter in the watch stream */
  watchFilter?: SpecWatchFilter;
  /** Snapshot ("NO_CHANGE") event in the watch stream. */
  watchSnapshot?: SpecWatchSnapshot;
  /** A step that the watch stream restarts. */
  watchStreamClose?: SpecWatchStreamClose;

  /** Ack the last write */
  writeAck?: SpecWriteAck;
  /** Fail a write */
  failWrite?: SpecWriteFailure;
  /** Add a new `waitForPendingWrites` listener. */
  waitForPendingWrites?: true;

  /** Fails the listed database actions. */
  failDatabase?: false | PersistenceAction[];

  /**
   * Set Index Configuration
   */
  setIndexConfiguration?: string | IndexConfiguration;

  /**
   * Run a queued timer task (without waiting for the delay to expire). See
   * TimerId enum definition for possible values).
   */
  runTimer?: string;

  /**
   * Process all events currently enqueued in the AsyncQueue.
   */
  drainQueue?: true;

  /** Enable or disable RemoteStore's network connection. */
  enableNetwork?: boolean;

  /** Clears the persistent storage in IndexedDB. */
  clearPersistence?: true;

  /** Changes the metadata state of a client instance. */
  applyClientState?: SpecClientState; // PORTING NOTE: Only used by web multi-tab tests

  /** Change to a new active user (specified by uid or null for anonymous). */
  changeUser?: string | null;

  /** Trigger a GC event with given cache threshold in bytes. */
  triggerLruGC?: number;

  /**
   * Restarts the SyncEngine from scratch, except re-uses persistence and auth
   * components. This allows you to queue writes, get documents into cache,
   * etc. and then simulate an app restart.
   */
  restart?: true;

  /** Shut down the client and close it network connection. */
  shutdown?: true;

  /** Sets the remote target index to be used by following steps. */
  watchUsesTargetIndex?: number | 'latest';

  /**
   * Optional list of expected events.
   * If not provided, the test will fail if the step causes events to be raised.
   */
  expectedSnapshotEvents?: SnapshotEvent[];

  /**
   * Optional dictionary of expected states.
   */
  expectedState?: StateExpectation;

  /**
   * Optional expected number of onSnapshotsInSync callbacks to be called.
   * If not provided, the test will fail if the step causes events to be raised.
   */
  expectedSnapshotsInSyncEvents?: number;

  /**
   * Optional expected number of waitForPendingWrite callbacks to be called.
   * If not provided, the test will fail if the step causes events to be raised.
   */
  expectedWaitForPendingWritesEvents?: number;
}

export interface SpecUserListen {
  targetId: TargetId;
  query: string | SpecQuery | CorePipeline;
  options?: ListenOptions;
}

/** [<target-id>, <query-path>] */
export type SpecUserUnlisten = [TargetId, string | SpecQuery | CorePipeline];

/** [<key>, <value>] */
export type SpecUserSet = [string, JsonObject<unknown>];

/** [<key>, <patches>] */
export type SpecUserPatch = [string, JsonObject<unknown>];

/** key */
export type SpecUserDelete = string;

/** [{sdkTargetId: <target-id>, remoteTargetIndex: <num?>}, ...] */
export type SpecWatchAck = TargetId[];

/** [[<target-id>, ...], <resume-token>] */
export type SpecWatchCurrent = [TargetId[], string];

/** [<target-id>, ...] */
export type SpecWatchReset = TargetId[];

export interface SpecError {
  code: number;
  message: string;
}

export interface SpecWatchRemove {
  targetIds: TargetId[];
  cause?: SpecError;
}

export interface SpecWatchSnapshot {
  version: TestSnapshotVersion;
  targetIds: TargetId[];
  resumeToken?: string;
}

export interface SpecWatchStreamClose {
  error: SpecError;
  runBackoffTimer: boolean;
}

export interface SpecWriteAck {
  /** The version the backend uses to ack the write. */
  version: TestSnapshotVersion;
  /**
   * Whether we should keep the write in our internal queue. This should only
   * be set to 'true' if the client ignores the write (e.g. a secondary client
   * which ignores write acknowledgments).
   *
   * Defaults to false.
   */
  // PORTING NOTE: Multi-Tab only.
  keepInQueue?: boolean;
}

export interface SpecWriteFailure {
  /** The error the backend uses to fail the write. */
  error: SpecError;
  /**
   * Whether we should keep the write in our internal queue. This should be set
   * to 'true' for transient errors or if the client ignores the failure
   * (e.g. a secondary client which ignores write rejections).
   *
   * Defaults to false.
   */
  keepInQueue?: boolean;
}

export interface SpecWatchEntity {
  // exactly one of key, doc or docs is set
  key?: string;
  /** [<key>, <version>, <value>] */
  doc?: SpecDocument;
  /** [<key>, <version>, <value>][] */
  docs?: SpecDocument[];
  /** [<target-id>, ...] */
  targets?: TargetId[];
  /** [<target-id>, ...] */
  removedTargets?: TargetId[];
}

// PORTING NOTE: Only used by web multi-tab tests.
export interface SpecClientState {
  /** The visibility state of the browser tab running the client. */
  visibility?: DocumentVisibilityState;
  /** Whether this tab should try to forcefully become primary. */
  primary?: true;
}

/**
 * The filter is based of a list of keys to match in the existence filter
 */
export interface SpecWatchFilter {
  targetIds: TargetId[];
  keys: string[];
  bloomFilter?: api.BloomFilter;
}

export type SpecLimitType = 'LimitToFirst' | 'LimitToLast';

/**
 * [field, op, value]
 * Op must be the `name` of an `Operator`.
 */
export type SpecQueryFilter = [string, string, unknown];

/**
 * [field, direction]
 * Direction can be 'asc' or 'desc'.
 */
export type SpecQueryOrderBy = [string, string];

/**
 * A representation of a query.
 */
export interface SpecQuery {
  path: string;
  collectionGroup?: string;
  limit?: number;
  limitType?: SpecLimitType;
  filters?: SpecQueryFilter[];
  orderBys?: SpecQueryOrderBy[];
}

/**
 * [<key>, <version>, <value>, <doc-options> (optional), ...]
 * Represents a document. <value> is null for deleted documents.
 * Doc options are:
 *   'local': document has local modifications
 */
export interface SpecDocument {
  key: string;
  version: TestSnapshotVersion;
  createTime: TestSnapshotVersion;
  value: JsonObject<unknown> | null;
  options?: DocumentOptions;
}

export interface SnapshotEvent {
  query: SpecQuery;
  errorCode?: number;
  fromCache?: boolean;
  hasPendingWrites?: boolean;
  added?: SpecDocument[];
  removed?: SpecDocument[];
  modified?: SpecDocument[];
  metadata?: SpecDocument[];
}

export interface StateExpectation {
  /** Number of outstanding writes in the datastore queue. */
  numOutstandingWrites?: number;
  /** Number of clients currently marked active. Used in multi-client tests. */
  numActiveClients?: number;
  /** Number of requests sent to the write stream. */
  writeStreamRequestCount?: number;
  /** Number of requests sent to the watch stream. */
  watchStreamRequestCount?: number;
  /**
   * Current documents in limbo that have an active target.
   * Verified in each step until overwritten.
   */
  activeLimboDocs?: string[];
  /**
   * Current documents in limbo that are enqueued and therefore do not have an
   * active target.
   * Verified in each step until overwritten.
   */
  enqueuedLimboDocs?: string[];
  /**
   * Whether the instance holds the primary lease. Used in multi-client tests.
   */
  isPrimary?: boolean;
  /** Whether the client is shutdown. */
  isShutdown?: boolean;
  /**
   * Current expected active targets. Verified in each step until overwritten.
   */
  activeTargets?: ActiveTargetMap;
  /**
   * Expected set of callbacks for previously written docs.
   */
  userCallbacks?: {
    acknowledgedDocs: string[];
    rejectedDocs: string[];
  };
  /** Indexes */
  indexes?: FieldIndex[];
}

async function clearCurrentPrimaryLease(): Promise<void> {
  const db = new SimpleDb(
    INDEXEDDB_TEST_DATABASE_NAME,
    SCHEMA_VERSION,
    new SchemaConverter(TEST_SERIALIZER)
  );
  await db.runTransaction(
    'clearCurrentPrimaryLease',
    'readwrite',
    [DbPrimaryClientStore],
    txn => {
      const primaryClientStore = txn.store<DbPrimaryClientKey, DbPrimaryClient>(
        DbPrimaryClientStore
      );
      return primaryClientStore.delete(DbPrimaryClientKey);
    }
  );
  db.close();
}