Merge pull request #3 from waterbustech/feat/recorder

Feat/recorder

Author: lambiengcode

.gitignore

@@ -34,4 +34,6 @@ lerna-debug.log*
 !.vscode/launch.json
 !.vscode/extensions.json
 
-.env
+.env
+rec
+credentials.json

ecosystem.config.js

@@ -14,6 +14,7 @@ module.exports = {
       error_file: 'logs/error.log',
       out_file: 'logs/out.log',
       log_date_format: 'YYYY-MM-DD HH:mm:ss',
+      kill_timeout: 30000,
     },
     {
       name: 'waterbus.sfu.ws.02',
@@ -31,6 +32,7 @@ module.exports = {
       error_file: 'logs/error.log',
       out_file: 'logs/out.log',
       log_date_format: 'YYYY-MM-DD HH:mm:ss',
+      kill_timeout: 30000,
     },
   ],
 };

env-example

@@ -0,0 +1,17 @@
+# App
+PORT=5985
+HOSTNAME=waterbus01
+
+# GRPC
+WEBSOCKET_GRPC_ADDRESS=0.0.0.0:50054
+AUTH_GRPC_ADDRESS=0.0.0.0:50055
+MEETING_GRPC_ADDRESS=0.0.0.0:50056
+
+# Microservices
+IS_MICROSERVICES=true
+# IS_USE_PM2=true
+REDIS_URL=redis://localhost:6379
+
+# Ice Servers
+TURN_USERNAME=waterbus
+TURN_CREDENTIAL=lambiengcode

example.env

@@ -22,6 +22,8 @@
     "pm2:start": "pm2 start ecosystem.config.js"
   },
   "dependencies": {
+    "@discordjs/opus": "^0.8.0",
+    "@google-cloud/speech": "^6.5.0",
     "@grpc/grpc-js": "^1.10.5",
     "@nestjs/common": "^9.0.0",
     "@nestjs/config": "^3.2.1",
@@ -33,8 +35,11 @@
     "@nestjs/websockets": "^10.3.7",
     "@socket.io/redis-adapter": "^8.3.0",
     "@types/mime": "3.0.4",
+    "audio-decode": "^2.2.0",
+    "blob": "^0.1.0",
     "dotenv": "^16.4.5",
     "google-protobuf": "^3.21.2",
+    "multicast-dns": "^7.2.5",
     "redis": "^4.6.13",
     "reflect-metadata": "^0.1.13",
     "rxjs": "^7.8.1",

package.json

@@ -22,15 +22,20 @@ const SocketEvent = {
   setAudioEnabledSSC: 'SET_AUDIO_ENABLED_SSC',
   setScreenSharingSSC: 'SET_SCREEN_SHARING_SSC',
   setScreenSharingCSS: 'SET_SCREEN_SHARING_CSS',
+  handRaisingSSC: 'HAND_RAISING_SSC',
+  handRaisingCSS: 'HAND_RAISING_CSS',
+  subtitleSSC: 'SUBTITLE_SSC',
+  setSubscribeSubtitleCSS: 'SET_SUBSCRIBE_SUBTITLE_CSS',
 
   // Chats
   sendMessageSSC: 'SEND_MESSAGE_SSC',
   updateMessageSSC: 'UPDATE_MESSAGE_SSC',
   deleteMessageSSC: 'DELETE_MESSAGE_SSC',
 
-  // Default
+  // System
   connection: 'connection',
   disconnect: 'disconnect',
+  destroy: 'destroy_',
 } as const;
 
 export default SocketEvent;

src/domain/constants/socket_events.ts

@@ -42,7 +42,7 @@ export class RedisIoAdapter extends IoAdapter {
             token: accessToken,
           });
 
-          if (verifyResponse.valid) {
+          if (verifyResponse && verifyResponse.valid) {
             request['userId'] = verifyResponse.userId;
             return allowFunction(null, true);
           }

src/infrastructure/config/socket/redis.adapter.ts

@@ -1,4 +1,4 @@
-import { Module } from '@nestjs/common';
+import { Module, forwardRef } from '@nestjs/common';
 import { SocketGateway } from './socket/socket.gateway';
 import { MeetingGateway } from './meeting/meeting.gateway';
 import { MeetingGrpcService } from 'src/infrastructure/services/meeting/meeting.service';

src/infrastructure/gateways/gateway.module.ts

@@ -0,0 +1,6 @@
+import { ApiProperty } from '@nestjs/swagger';
+
+export class SetSubscribeSubtitleDto {
+  @ApiProperty({ type: Boolean })
+  enabled: boolean;
+}

src/infrastructure/gateways/meeting/dtos/set_subscribe_subtitle.dto.ts

@@ -3,4 +3,10 @@ import { ApiProperty } from '@nestjs/swagger';
 export class SubscribeDto {
   @ApiProperty({ type: String })
   targetId: string;
+
+  @ApiProperty({ type: String })
+  roomId: string;
+
+  @ApiProperty({ type: String })
+  participantId: string;
 }

src/infrastructure/gateways/meeting/dtos/subscribe.dto.ts

@@ -21,6 +21,7 @@ import { MeetingGrpcService } from 'src/infrastructure/services/meeting/meeting.
 import RedisEvents from 'src/domain/constants/redis_events';
 import { MessageBroker } from 'src/infrastructure/services/message-broker/message-broker';
 import { EnvironmentConfigService } from 'src/infrastructure/config/environment/environments';
+import { SetSubscribeSubtitleDto } from './dtos/set_subscribe_subtitle.dto';
 
 @WebSocketGateway()
 export class MeetingGateway {
@@ -36,7 +37,7 @@ export class MeetingGateway {
   }
 
   @WebSocketServer() private server: Server;
-  private logger: Logger = new Logger('MeetingGateway');
+  private logger: Logger = new Logger(MeetingGateway.name);
 
   @SubscribeMessage(SocketEvent.publishCSS)
   async handleJoinRoom(client: ISocketClient, payload: JoinRoomDto) {
@@ -102,18 +103,12 @@ export class MeetingGateway {
           ...responsePayload,
         });
       } else {
-        const clientInfo = this.rtcManager.getClientBySocketId({
-          clientId: client.id,
-        });
-
-        if (!clientInfo) return;
-
         this.messageBroker.publishRedisChannel(
           participantInfo.ccu.podName,
           RedisEvents.SUBSCRIBE,
           {
-            participantId: clientInfo.participantId,
-            roomId: clientInfo.roomId,
+            participantId: payload.participantId,
+            roomId: payload.roomId,
             targetId: payload.targetId,
             clientId: client.id,
           },
@@ -321,6 +316,48 @@ export class MeetingGateway {
     });
   }
 
+  @SubscribeMessage(SocketEvent.handRaisingCSS)
+  handleSetHandRaising(client: ISocketClient, payload: any): any {
+    const clientInfo = this.rtcManager.getClientBySocketId({
+      clientId: client.id,
+    });
+
+    if (!clientInfo) return;
+
+    const roomId = clientInfo.roomId;
+    const targetId = clientInfo.participantId;
+
+    if (!roomId) return;
+
+    client.broadcast.to(roomId).emit(SocketEvent.handRaisingSSC, {
+      participantId: targetId,
+    });
+  }
+
+  @SubscribeMessage(SocketEvent.setSubscribeSubtitleCSS)
+  handleSetSubscribeSubtitle(
+    client: ISocketClient,
+    payload: SetSubscribeSubtitleDto,
+  ): any {
+    const clientInfo = this.rtcManager.getClientBySocketId({
+      clientId: client.id,
+    });
+
+    if (!clientInfo) return;
+
+    const roomId = clientInfo.roomId;
+
+    if (!roomId) return;
+
+    const subtitleChannel = SocketEvent.subtitleSSC + roomId;
+
+    if (payload.enabled) {
+      client.join(subtitleChannel);
+    } else {
+      client.leave(subtitleChannel);
+    }
+  }
+
   @SubscribeMessage(SocketEvent.leaveRoomCSS)
   async handleLeaveRoom(client: ISocketClient, payload: any) {
     const info = this.rtcManager.leaveRoom({ clientId: client.id });

src/infrastructure/gateways/meeting/meeting.gateway.ts

@@ -31,7 +31,7 @@ export class SocketGateway
 
   @WebSocketServer() public server: Server;
 
-  private logger: Logger = new Logger('SocketGateway');
+  private logger: Logger = new Logger(SocketGateway.name);
 
   afterInit(server: Server) {
     this.server = server;
@@ -40,15 +40,15 @@ export class SocketGateway
 
   async handleConnection(client: ISocketClient) {
     try {
-      this.logger.debug(`Client connected: ${client.id}`);
-
       const userId = client.request['userId'];
 
       this.authService.createCCU({
         socketId: client.id,
         podName: this.environment.getPodName(),
         userId,
       });
+
+      client.join(SocketEvent.destroy + this.environment.getPodName());
     } catch (error) {
       this.logger.error(error?.message, error?.stack);
       return client.disconnect(true);
@@ -86,6 +86,8 @@ export class SocketGateway
       try {
         this.logger.debug(`Pod is shutting down...`);
 
+        this.server.emit(SocketEvent.destroy + this.environment.getPodName());
+
         // Delete CCU & participants in this pod
         this.authService.shutDownPod({
           podName: this.environment.getPodName(),

src/infrastructure/gateways/socket/socket.gateway.ts

@@ -1,35 +1,35 @@
-import * as winston from 'winston';
+// import * as winston from 'winston';
 
-const customFormat = winston.format.printf(({ timestamp, level, message }) => {
-  let icon = '';
-  switch (level) {
-    case 'info':
-      icon = '🦋';
-      break;
-    case 'warn':
-      icon = '⚠️';
-      break;
-    case 'error':
-      icon = '🐞';
-      break;
-    default:
-      icon = '';
-      break;
-  }
-  return `${icon} [${timestamp}][${level}]: ${message}`;
-});
+// const customFormat = winston.format.printf(({ timestamp, level, message }) => {
+//   let icon = '';
+//   switch (level) {
+//     case 'info':
+//       icon = '🦋';
+//       break;
+//     case 'warn':
+//       icon = '⚠️';
+//       break;
+//     case 'error':
+//       icon = '🐞';
+//       break;
+//     default:
+//       icon = '';
+//       break;
+//   }
+//   return `${icon} [${timestamp}][${level}]: ${message}`;
+// });
 
-const logger = winston.createLogger({
-  level: 'info',
-  format: winston.format.combine(
-    winston.format.colorize(),
-    winston.format.timestamp(),
-    customFormat,
-  ),
-  transports: [
-    new winston.transports.File({ filename: 'app.log', level: 'info' }),
-    new winston.transports.Console(),
-  ],
-});
+// const logger = winston.createLogger({
+//   level: 'info',
+//   format: winston.format.combine(
+//     winston.format.colorize(),
+//     winston.format.timestamp(),
+//     customFormat,
+//   ),
+//   transports: [
+//     new winston.transports.File({ filename: 'app.log', level: 'info' }),
+//     new winston.transports.Console(),
+//   ],
+// });
 
-export default logger;
+// export default logger;

src/infrastructure/helpers/logger.ts

@@ -27,7 +27,7 @@ export class AuthGrpcService implements OnModuleInit {
   private $reconnect: Subscription;
 
   constructor(private readonly authClientProxy: ClientGrpc) {
-    this.logger = new Logger('AuthService');
+    this.logger = new Logger(AuthGrpcService.name);
   }
 
   onModuleInit() {

src/infrastructure/services/auth/auth.service.ts

@@ -27,7 +27,7 @@ export class MeetingGrpcService implements OnModuleInit {
   private $reconnect: Subscription;
 
   constructor(private readonly meetingClientProxy: ClientGrpc) {
-    this.logger = new Logger('MeetingService');
+    this.logger = new Logger(MeetingGrpcService.name);
   }
 
   onModuleInit() {