Merge pull request #789 from stenioanibal/v2.0.0

V2.0.0 - Refactor websocket structure

src/api/integrations/websocket/controllers/websocket.controller.ts

@@ -1,49 +1,217 @@
-import { InstanceDto } from '@api/dto/instance.dto';
 import { WebsocketDto } from '@api/integrations/websocket/dto/websocket.dto';
-import { WebsocketService } from '@api/integrations/websocket/services/websocket.service';
+import { PrismaRepository } from '@api/repository/repository.service';
+import { WAMonitoringService } from '@api/services/monitor.service';
+import { wa } from '@api/types/wa.types';
+import { configService, Cors, HttpServer, Log, Websocket } from '@config/env.config';
+import { Logger } from '@config/logger.config';
+import { NotFoundException } from '@exceptions';
+import { Server } from 'http';
+import { Server as SocketIO } from 'socket.io';
 
 export class WebsocketController {
-  constructor(private readonly websocketService: WebsocketService) {}
+  private io: SocketIO;
+  private prismaRepository: PrismaRepository;
+  private waMonitor: WAMonitoringService;
+  private corsConfig: Array<any>;
+  private readonly logger = new Logger('SocketStartupService');
+  public readonly monitorEvents = ['REMOVE_INSTANCE', 'LOGOUT_INSTANCE'];
+  public readonly events = [
+    'APPLICATION_STARTUP',
+    'QRCODE_UPDATED',
+    'MESSAGES_SET',
+    'MESSAGES_UPSERT',
+    'MESSAGES_EDITED',
+    'MESSAGES_UPDATE',
+    'MESSAGES_DELETE',
+    'SEND_MESSAGE',
+    'CONTACTS_SET',
+    'CONTACTS_UPSERT',
+    'CONTACTS_UPDATE',
+    'PRESENCE_UPDATE',
+    'CHATS_SET',
+    'CHATS_UPSERT',
+    'CHATS_UPDATE',
+    'CHATS_DELETE',
+    'GROUPS_UPSERT',
+    'GROUP_UPDATE',
+    'GROUP_PARTICIPANTS_UPDATE',
+    'CONNECTION_UPDATE',
+    'LABELS_EDIT',
+    'LABELS_ASSOCIATION',
+    'CALL',
+    'TYPEBOT_START',
+    'TYPEBOT_CHANGE_STATUS',
+  ];
 
-  public async createWebsocket(instance: InstanceDto, data: WebsocketDto) {
+  constructor(prismaRepository: PrismaRepository, waMonitor: WAMonitoringService) {
+    this.prisma = prismaRepository;
+    this.monitor = waMonitor;
+    this.cors = configService.get<Cors>('CORS').ORIGIN;
+  }
+
+  public init(httpServer: Server): void {
+    if (!configService.get<Websocket>('WEBSOCKET')?.ENABLED) {
+      return;
+    }
+
+    this.socket = new SocketIO(httpServer, {
+      cors: {
+        origin: this.cors,
+      },
+    });
+
+    this.socket.on('connection', (socket) => {
+      this.logger.info('User connected');
+
+      socket.on('disconnect', () => {
+        this.logger.info('User disconnected');
+      });
+    });
+
+    this.logger.info('Socket.io initialized');
+  }
+
+  private set prisma(prisma: PrismaRepository) {
+    this.prismaRepository = prisma;
+  }
+
+  private get prisma() {
+    return this.prismaRepository;
+  }
+
+  private set monitor(waMonitor: WAMonitoringService) {
+    this.waMonitor = waMonitor;
+  }
+
+  private get monitor() {
+    return this.waMonitor;
+  }
+
+  private set cors(cors: Array<any>) {
+    this.corsConfig = cors;
+  }
+
+  private get cors(): string | Array<any> {
+    return this.corsConfig?.includes('*') ? '*' : this.corsConfig;
+  }
+
+  private set socket(socket: SocketIO) {
+    this.io = socket;
+  }
+
+  public get socket(): SocketIO {
+    return this.io;
+  }
+
+  public async set(instanceName: string, data: WebsocketDto): Promise<wa.LocalWebsocket> {
     if (!data.enabled) {
       data.events = [];
+    } else {
+      if (0 === data.events.length) {
+        data.events = this.events;
+      }
     }
 
-    if (data.events.length === 0) {
-      data.events = [
-        'APPLICATION_STARTUP',
-        'QRCODE_UPDATED',
-        'MESSAGES_SET',
-        'MESSAGES_UPSERT',
-        'MESSAGES_EDITED',
-        'MESSAGES_UPDATE',
-        'MESSAGES_DELETE',
-        'SEND_MESSAGE',
-        'CONTACTS_SET',
-        'CONTACTS_UPSERT',
-        'CONTACTS_UPDATE',
-        'PRESENCE_UPDATE',
-        'CHATS_SET',
-        'CHATS_UPSERT',
-        'CHATS_UPDATE',
-        'CHATS_DELETE',
-        'GROUPS_UPSERT',
-        'GROUP_UPDATE',
-        'GROUP_PARTICIPANTS_UPDATE',
-        'CONNECTION_UPDATE',
-        'LABELS_EDIT',
-        'LABELS_ASSOCIATION',
-        'CALL',
-        'TYPEBOT_START',
-        'TYPEBOT_CHANGE_STATUS',
-      ];
-    }
+    try {
+      await this.get(instanceName);
 
-    return this.websocketService.create(instance, data);
+      return this.prisma.websocket.update({
+        where: {
+          instanceId: this.monitor.waInstances[instanceName].instanceId,
+        },
+        data,
+      });
+    } catch (err) {
+      return this.prisma.websocket.create({
+        data: {
+          enabled: data.enabled,
+          events: data.events,
+          instanceId: this.monitor.waInstances[instanceName].instanceId,
+        },
+      });
+    }
   }
 
-  public async findWebsocket(instance: InstanceDto) {
-    return this.websocketService.find(instance);
+  public async get(instanceName: string): Promise<wa.LocalWebsocket> {
+    if (undefined === this.monitor.waInstances[instanceName]) {
+      throw new NotFoundException('Instance not found');
+    }
+
+    const data = await this.prisma.websocket.findUnique({
+      where: {
+        instanceId: this.monitor.waInstances[instanceName].instanceId,
+      },
+    });
+
+    if (!data) {
+      throw new NotFoundException('Websocket not found');
+    }
+
+    return data;
+  }
+
+  public async emit({
+    instanceName,
+    origin,
+    event,
+    data,
+  }: {
+    instanceName: string;
+    origin: string;
+    event: string;
+    data: Object;
+  }): Promise<void> {
+    if (!configService.get<Websocket>('WEBSOCKET')?.ENABLED) {
+      return;
+    }
+
+    const configEv = event.replace(/[.-]/gm, '_').toUpperCase();
+    const logEnabled = configService.get<Log>('LOG').LEVEL.includes('WEBSOCKET');
+    const serverUrl = configService.get<HttpServer>('SERVER').URL;
+    const date = new Date(Date.now() - new Date().getTimezoneOffset() * 60000).toISOString();
+    const message = {
+      event,
+      instanceName,
+      data,
+      serverUrl,
+      date,
+    };
+
+    if (configService.get<Websocket>('WEBSOCKET')?.GLOBAL_EVENTS) {
+      this.socket.emit(event, message);
+
+      if (logEnabled) {
+        this.logger.log({
+          local: `${origin}.sendData-WebsocketGlobal`,
+          ...message,
+        });
+      }
+    }
+
+    try {
+      const instanceSocket = await this.get(instanceName);
+
+      if (!instanceSocket.enabled) {
+        return;
+      }
+
+      if (
+        this.monitorEvents.includes(configEv) ||
+        (Array.isArray(instanceSocket.events) && instanceSocket.events.includes(configEv))
+      ) {
+        this.socket.of(`/${instanceName}`).emit(event, message);
+
+        if (logEnabled) {
+          this.logger.log({
+            local: `${origin}.sendData-Websocket`,
+            ...message,
+          });
+        }
+      }
+    } catch (err) {
+      if (logEnabled) {
+        this.logger.log(err);
+      }
+    }
   }
 }

src/api/integrations/websocket/libs/socket.server.ts

@@ -1,41 +0,0 @@
-import { configService, Cors, Websocket } from '@config/env.config';
-import { Logger } from '@config/logger.config';
-import { Server } from 'http';
-import { Server as SocketIO } from 'socket.io';
-
-const logger = new Logger('Socket');
-
-let io: SocketIO;
-
-const cors = configService.get<Cors>('CORS').ORIGIN;
-
-export const initIO = (httpServer: Server) => {
-  if (configService.get<Websocket>('WEBSOCKET')?.ENABLED) {
-    io = new SocketIO(httpServer, {
-      cors: {
-        origin: cors,
-      },
-    });
-
-    io.on('connection', (socket) => {
-      logger.info('User connected');
-
-      socket.on('disconnect', () => {
-        logger.info('User disconnected');
-      });
-    });
-
-    logger.info('Socket.io initialized');
-    return io;
-  }
-  return null;
-};
-
-export const getIO = (): SocketIO => {
-  if (!io) {
-    logger.error('Socket.io not initialized');
-    throw new Error('Socket.io not initialized');
-  }
-
-  return io;
-};

src/api/integrations/websocket/routes/websocket.router.ts

@@ -15,7 +15,7 @@ export class WebsocketRouter extends RouterBroker {
           request: req,
           schema: websocketSchema,
           ClassRef: WebsocketDto,
-          execute: (instance, data) => websocketController.createWebsocket(instance, data),
+          execute: (instance, data) => websocketController.set(instance.instanceName, data),
         });
 
         res.status(HttpStatus.CREATED).json(response);
@@ -25,7 +25,7 @@ export class WebsocketRouter extends RouterBroker {
           request: req,
           schema: instanceSchema,
           ClassRef: InstanceDto,
-          execute: (instance) => websocketController.findWebsocket(instance),
+          execute: (instance) => websocketController.get(instance.instanceName),
         });
 
         res.status(HttpStatus.OK).json(response);

src/api/integrations/websocket/services/websocket.service.ts

@@ -1,31 +0,0 @@
-import { InstanceDto } from '@api/dto/instance.dto';
-import { WebsocketDto } from '@api/integrations/websocket/dto/websocket.dto';
-import { WAMonitoringService } from '@api/services/monitor.service';
-import { Logger } from '@config/logger.config';
-import { Websocket } from '@prisma/client';
-
-export class WebsocketService {
-  constructor(private readonly waMonitor: WAMonitoringService) {}
-
-  private readonly logger = new Logger('WebsocketService');
-
-  public create(instance: InstanceDto, data: WebsocketDto) {
-    this.waMonitor.waInstances[instance.instanceName].setWebsocket(data);
-
-    return { websocket: { ...instance, websocket: data } };
-  }
-
-  public async find(instance: InstanceDto): Promise<Websocket> {
-    try {
-      const result = await this.waMonitor.waInstances[instance.instanceName].findWebsocket();
-
-      if (Object.keys(result).length === 0) {
-        throw new Error('Websocket not found');
-      }
-
-      return result;
-    } catch (error) {
-      return null;
-    }
-  }
-}