message-sync.mixin.ts

import { hook, Mixin } from './mixin';
import { RealtimeClient } from '../realtime.client';
import { Topics } from '../../constants';
import { tryUnzipAsync } from '../../shared';
import { IrisParserData } from '../parsers';
import { IllegalStateError } from 'mqtts';

export class MessageSyncMixin extends Mixin {
   apply(client: RealtimeClient): void {
      hook(client, 'connect', {
         post: () => {
            if (!client.mqtt) {
               throw new IllegalStateError('No mqtt client created');
            }
            client.mqtt.listen(
               {
                  topic: Topics.MESSAGE_SYNC.id,
                  transformer: async ({ payload }) =>
                     Topics.MESSAGE_SYNC.parser
                        .parseMessage(Topics.MESSAGE_SYNC, await tryUnzipAsync(payload))
                        .map(msg => msg.data),
               },
               data => this.handleMessageSync(client, data),
            );
         },
      });
   }

   private handleMessageSync(client: RealtimeClient, syncData: IrisParserData[]) {
      for (const element of syncData) {
         const data = element.data;
         if (!data) {
            client.emit('iris', element);
            continue;
         }
         delete element.data;
         data.forEach(e => {
            if (!e.path) {
               client.emit('iris', { ...element, ...e });
            }
            if (e.path.startsWith('/direct_v2/threads') && e.value) {
               client.emit('message', {
                  ...element,
                  message: {
                     path: e.path,
                     op: e.op,
                     thread_id: MessageSyncMixin.getThreadIdFromPath(e.path),
                     ...JSON.parse(e.value),
                  },
               });
            } else {
               client.emit('threadUpdate', {
                  ...element,
                  meta: {
                     path: e.path,
                     op: e.op,
                     thread_id: MessageSyncMixin.getThreadIdFromPath(e.path),
                  },
                  update: {
                     ...JSON.parse(e.value),
                  },
               });
            }
         });
      }
   }

   private static getThreadIdFromPath(path: string): string | undefined {
      const itemMatch = path.match(/^\/direct_v2\/threads\/(\d+)/);
      if (itemMatch) return itemMatch[1];
      const inboxMatch = path.match(/^\/direct_v2\/inbox\/threads\/(\d+)/);
      if (inboxMatch) return inboxMatch[1];

      return undefined;
   }

   get name(): string {
      return 'Message Sync';
   }
}