Method onMessageHandler

packages/client/src/sync-engine.ts:851–983

(session: SyncSession)

  // TODO: add an onError handler to gracefully handle errors in message handlers
  private onMessageHandler(session: SyncSession) {
    return async (evt: any) => {
      const message: ServerSyncMessage = JSON.parse(evt.data);
      this.logger.debug('received', message);
      for (const handler of this.messageReceivedSubscribers) {
        handler(message);
      }
      if (message.type === 'ERROR') {
        await this.handleErrorMessage(message);
      }
      if (message.type === 'ENTITY_DATA') {
        const {
          changes: stringifiedChanges,
          timestamp,
          forQueries: queryIds,
        } = message.payload;
        const changes = SuperJSON.deserialize<DBChanges>(stringifiedChanges);
        // first apply changes
        // the db will push these onto IVMs buffer
        if (this.client.awaitReady) await this.client.awaitReady;
        await this.client.db.applyChangesWithTimestamp(changes, timestamp, {
          skipRules: true,
        });

        // TODO do in same transaction
        await this.client.db.setMetadata(
          ['latest_server_timestamp'],
          timestamp
        );

        // then update the query fulfillment state so that
        // the client can signal in the results handler
        // that the next time IVM fires, it's because
        // of the server's response
        for (const qId of queryIds) {
          const query = this.queries.get(qId);
          if (!query) continue;
          if (query.syncState !== 'FULFILLED') {
            await this.markQueryAsSeen(qId);
            query.syncState = 'FULFILLED';
          }
          // this.queryFulfillmentCallbacks.delete(qId);
        }

        // update IVM
        await this.client.db.updateQueryViews();
        this.client.db.broadcastToQuerySubscribers();

        // finally, run the query fulfillment callbacks
        for (const qId of queryIds) {
          const query = this.queries.get(qId);
          if (!query) continue;
          for (const callback of query.syncStateCallbacks) {
            callback('FULFILLED', message.payload);
          }
        }
      }