fix: 채팅 서버 고도화 v2.3.0

chat-server/src/common/modules/queue-map/queue-map.module.ts

@@ -0,0 +1,8 @@
+import { Module } from '@nestjs/common';
+import { QueueMapService } from './queue-map.service';
+
+@Module({
+  providers: [Map, QueueMapService],
+  exports: [QueueMapService],
+})
+export class QueueMapModule {}

chat-server/src/common/modules/queue-map/queue-map.service.ts

@@ -0,0 +1,22 @@
+import { Inject } from '@nestjs/common';
+import Bull from 'bull';
+
+export class QueueMapService {
+  constructor(@Inject(Map) private map: Map<string, Bull.Queue>) {}
+
+  set(key: string, value: Bull.Queue) {
+    this.map.set(key, value);
+  }
+
+  get(key: string) {
+    return this.map.get(key);
+  }
+
+  delete(key: string) {
+    this.map.delete(key);
+  }
+
+  keys() {
+    return this.map.keys();
+  }
+}

chat-server/src/common/modules/socket-map/socket-map.module.ts

@@ -0,0 +1,8 @@
+import { Module } from '@nestjs/common';
+import { SocketMapService } from './socket-map.service';
+
+@Module({
+  providers: [Map, SocketMapService],
+  exports: [SocketMapService],
+})
+export class SocketMapModule {}

chat-server/src/common/modules/socket-map/socket-map.service.ts

@@ -0,0 +1,22 @@
+import { Inject } from '@nestjs/common';
+import { Socket } from 'socket.io';
+
+export class SocketMapService {
+  constructor(@Inject(Map) private map: Map<string, Socket>) {}
+
+  set(key: string, value: Socket) {
+    this.map.set(key, value);
+  }
+
+  get(key: string) {
+    return this.map.get(key);
+  }
+
+  delete(key: string) {
+    this.map.delete(key);
+  }
+
+  keys() {
+    return this.map.keys();
+  }
+}

chat-server/src/common/modules/unread-map/unread-map.module.ts

@@ -0,0 +1,8 @@
+import { Module } from '@nestjs/common';
+import { UnReadMapService } from './unread-map.service';
+
+@Module({
+  providers: [Map, UnReadMapService],
+  exports: [UnReadMapService],
+})
+export class UnReadMapModule {}

chat-server/src/common/modules/unread-map/unread-map.service.ts

@@ -0,0 +1,21 @@
+import { Inject } from '@nestjs/common';
+
+export class UnReadMapService {
+  constructor(@Inject(Map) private map: Map<string, number>) {}
+
+  set(key: string, value: number) {
+    this.map.set(key, value);
+  }
+
+  get(key: string) {
+    return this.map.get(key);
+  }
+
+  delete(key: string) {
+    this.map.delete(key);
+  }
+
+  keys() {
+    return this.map.keys();
+  }
+}

chat-server/src/common/types/get-many.type.ts

@@ -0,0 +1,3 @@
+export type GetChatsDto = {
+  recruitId: string;
+};

chat-server/src/common/types/get-one.type.ts

@@ -0,0 +1,4 @@
+export type GetChatDto = {
+  recruitId: string;
+  userId: string;
+};

chat-server/src/common/types/history.type.ts

@@ -0,0 +1,6 @@
+export type ChatHistoryDto = {
+  recruitId: string;
+  userId: string;
+  page: number;
+  paused?: number;
+};

chat-server/src/common/types/index.ts

@@ -0,0 +1,5 @@
+import { GetChatDto } from './get-one.type';
+import { GetChatsDto } from './get-many.type';
+import { ChatHistoryDto } from './history.type';
+
+export { GetChatDto, GetChatsDto, ChatHistoryDto };

chat-server/src/main.ts

@@ -3,6 +3,7 @@ import { SocketModule } from './socket.module';
 
 async function bootstrap() {
   const app = await NestFactory.create(SocketModule);
+  app.enableCors({ origin: '*', credentials: true });
   await app.listen(8080);
 }
 bootstrap();

chat-server/src/queue-manager/manager.module.ts

@@ -0,0 +1,29 @@
+import { CacheModule, Module } from '@nestjs/common';
+import * as redisStore from 'cache-manager-ioredis';
+import { ManagerService } from './manager.service';
+import { BullModule } from '@nestjs/bull';
+import { QueueMapModule } from 'src/common/modules/queue-map/queue-map.module';
+import { SocketMapModule } from 'src/common/modules/socket-map/socket-map.module';
+import { UnReadMapModule } from 'src/common/modules/unread-map/unread-map.module';
+
+@Module({
+  imports: [
+    CacheModule.registerAsync({
+      useFactory: () => {
+        return {
+          store: redisStore,
+          host: 'localhost',
+          port: 6379,
+          ttl: 0,
+        };
+      },
+    }),
+    BullModule,
+    QueueMapModule,
+    SocketMapModule,
+    UnReadMapModule,
+  ],
+  providers: [ManagerService],
+  exports: [ManagerService],
+})
+export class ManagerModule {}

chat-server/src/queue-manager/manager.service.ts

@@ -0,0 +1,103 @@
+import { CACHE_MANAGER, Inject, Injectable, Scope } from '@nestjs/common';
+import { Cache } from 'cache-manager';
+import * as Bull from 'bull';
+import { Socket } from 'socket.io';
+import { QueueMapService } from 'src/common/modules/queue-map/queue-map.service';
+import { SocketMapService } from 'src/common/modules/socket-map/socket-map.service';
+import { UnReadMapService } from 'src/common/modules/unread-map/unread-map.service';
+
+@Injectable({ scope: Scope.DEFAULT })
+export class ManagerService {
+  constructor(
+    @Inject(CACHE_MANAGER) private redisCache: Cache,
+    private queueMap: QueueMapService,
+    private socketMap: SocketMapService,
+    private unReadCountMap: UnReadMapService,
+  ) {}
+
+  async generateQueue(name: string) {
+    const queue = new Bull(name);
+    queue.pause();
+    this.queueMap.set(name, queue);
+    return queue;
+  }
+
+  async deleteOneQueue(name: string) {
+    const deleteWork = [];
+    const keyArr = await this.redisCache.store.keys(`bull:${name}:*`);
+    keyArr.map((key: string) => {
+      deleteWork.push(this.redisCache.del(key));
+    }); // bull.js Queue 지우는용
+    this.queueMap.delete(name); // 매핑된 인스턴스 지우는용
+    return Promise.all(deleteWork);
+  }
+
+  async deleteManyQueue(recruitId: string) {
+    const deleteWork = [];
+    const keyArr = await this.redisCache.store.keys(`bull:${recruitId}:*`);
+    keyArr.map((key: string) => {
+      deleteWork.push(this.redisCache.del(key));
+    }); // bull.js Queue 지우는용
+    const keys = Array.from(this.queueMap.keys()).filter(
+      (key: string) => key.split(':')[0] === recruitId,
+    );
+    keys.map((key: string) => this.queueMap.delete(key)); // 매핑된 모든 인스턴스 지우는용
+  }
+
+  // 서버 메모리에서, name(key) 값으로 Queue Instance 가져와서 반환해주기
+  getQueue(name: string): Bull.Queue {
+    return this.queueMap.get(name);
+  }
+
+  getQueueList(recruitId: string) {
+    const keys = Array.from(this.queueMap.keys()).filter(
+      (key: string) => key.split(':')[0] === recruitId,
+    );
+    return keys.map((key) => this.queueMap.get(key));
+  }
+
+  async getQueueSize(name: string) {
+    const queue = this.queueMap.get(name);
+    if (!queue) return 0;
+    const { paused } = (await queue.getJobCounts()) as any;
+    return paused;
+  }
+
+  getSocket(name: string): Socket {
+    return this.socketMap.get(name);
+  }
+  // 온라인인 유저 상태 관리 -> 그래야 send 이벤트를 발생을 시켜서 값을 가져오니깐
+  // 온라인이면 -> 바로 전송
+  // 오프라인이면 -> 큐에 누적
+
+  // 지금 Cache Module에서 사용하는건, 레디스인데, Cache Module 메모리와 레디스를 함께 쓸 순 없을까?
+
+  setSocket(userId: string, socket: Socket): void {
+    this.socketMap.set(userId, socket);
+  }
+
+  deleteSocket(userId: string): void {
+    this.socketMap.delete(userId);
+  }
+
+  getUnReadCount(name: string): number {
+    return this.unReadCountMap.get(name);
+  }
+
+  setUnReadCount(name: string, unReadCount: number): void {
+    this.unReadCountMap.set(name, unReadCount);
+  }
+
+  deleteUnReadCount(name: string): void {
+    this.unReadCountMap.delete(name);
+  }
+
+  addUnReadCount(recruitId: string): void {
+    const keys = Array.from(this.unReadCountMap.keys()).filter(
+      (key: string) => key.split(':')[0] === recruitId,
+    );
+    keys.map((key) => {
+      this.unReadCountMap.set(key, this.unReadCountMap.get(key) + 1);
+    });
+  }
+}

chat-server/src/socket.controller.ts

@@ -0,0 +1,69 @@
+import { Body, Controller, Get, Post, Query } from '@nestjs/common';
+import { ManagerService } from './queue-manager/manager.service';
+import { SocketService } from './socket.service';
+import { GetChatDto, GetChatsDto, ChatHistoryDto } from './common/types';
+
+@Controller()
+export class SocketController {
+  constructor(
+    private managerService: ManagerService,
+    private socketService: SocketService,
+  ) {}
+
+  @Get('chat')
+  async getRecentMessage(@Query() bodyDto: ChatHistoryDto) {
+    const { userId, recruitId, page, paused } = bodyDto;
+    const unReadCount =
+      this.managerService.getUnReadCount(`${recruitId}:${userId}`) ||
+      Number(paused);
+
+    const data = await this.socketService.getRecentMessage(
+      parseInt(recruitId),
+      page,
+      unReadCount,
+    );
+    return {
+      statusCode: 200,
+      data,
+    };
+  }
+
+  @Get('unread')
+  async getUnreadMessage(@Query() bodyDto: GetChatDto) {
+    const { recruitId, userId } = bodyDto;
+    const queue = this.managerService.getQueue(`${recruitId}:${userId}`);
+
+    if (queue === undefined) {
+      console.log(`GET unread/ ${recruitId}:${userId}: Queue is not defined!`);
+      return { statusCode: 201, data: { paused: 0 } };
+    }
+
+    const response = (await queue.getJobCounts()) as any;
+    console.log(`GET unread/ ${recruitId}:${userId}: ${response.paused}`);
+    return { statusCode: 201, data: { paused: response.paused } };
+  }
+
+  // POST localhost:8080/queue {recruitId, userId}
+  @Post('queue')
+  async generate(@Body() bodyDto: GetChatDto) {
+    const { recruitId, userId } = bodyDto;
+    await this.managerService.generateQueue(`${recruitId}:${userId}`);
+    return { statusCode: 201 };
+  }
+
+  // POST localhost:8080/queue/delete/one {recruitId, userId}
+  @Post('queue/delete/one')
+  async deleteOne(@Body() bodyDto: GetChatDto) {
+    const { recruitId, userId } = bodyDto;
+    await this.managerService.deleteOneQueue(`${recruitId}:${userId}`);
+    return { statusCode: 201 };
+  }
+
+  // POST localhost:8080/queue/delete/many {recruitId}
+  @Post('queue/delete/many')
+  async deleteMany(@Body() bodyDto: GetChatsDto) {
+    const { recruitId } = bodyDto;
+    await this.managerService.deleteManyQueue(recruitId);
+    return { statusCode: 201 };
+  }
+}