Method connectQuery

packages/client/src/sync-engine.ts:672–757

(queryId: string)

  }

  private async connectQuery(queryId: string) {
    if (this.client.awaitReady) await this.client.awaitReady;
    if (!this.queries.has(queryId)) return;

    const queryMetadata = this.queries.get(queryId);
    /**
     * Do not send CONNECT_QUERY message if:
     * - query no longer exists
     * - query has already been sent in this session
     * - query has been aborted
     * - server is not ready
     */
    if (
      !queryMetadata ||
      queryMetadata.hasSent ||
      queryMetadata.abortController.signal.aborted ||
      !this.serverReady
    ) {
      return;
    }

    const latestServerTimestamp = (await this.client.db.getMetadata([
      'latest_server_timestamp',
    ])) as SyncTimestamp | undefined;

    let queryState: QueryState | undefined = undefined;

    if (latestServerTimestamp) {
      const queryWithRelationalInclusions =
        createQueryWithRelationalOrderAddedToIncludes(
          createQueryWithExistsAddedToIncludes(
            prepareQuery(
              queryMetadata.params,
              this.client.db.schema?.['collections'],
              {},
              undefined,
              {
                applyPermission: undefined,
              }
            )
          )
        );
      // We should only consider your cache data for checkpointed fetch
      // We dont have a great API for this right now, so using the DB's query engine directly
      const queryEngine = new EntityStoreQueryEngine(
        this.client.db.kv,
        this.client.db.entityStore.dataStore
      );
      const syncedResults = await queryEngine.fetch(
        queryWithRelationalInclusions
      );
      const changesFromResults = queryResultsToChanges(
        syncedResults,
        queryWithRelationalInclusions
      );

      const entityIds = changesToEntityIds(changesFromResults);