Merge pull request #281 from boostcampwm-2024/fix/be/chats

[Fix] 채팅 필터 적용에 따른 지연 문제 수정
boostcampwm-2024 · Nov 27, 2024 · 87868a7 · 87868a7
2 parents 29027c6 + ad9c9b2
commit 87868a7
Show file tree

Hide file tree

Showing 3 changed files with 71 additions and 9 deletions.
diff --git a/Backend/apps/api/src/chats/chats.controller.ts b/Backend/apps/api/src/chats/chats.controller.ts
@@ -11,6 +11,6 @@ export class ChatsController {
   @Post()
   @UseGuards(JwtAuthGuard)
   async sendChat(@Body(ValidationPipe) sendChatDto: SendChatDto, @Req() req: Request & { user: UserEntity }) {
-    this.chatsService.publishChat({ ...sendChatDto, userId: req.user.id, nickname: req.user.nickname });
+    this.chatsService.ingestChat({ ...sendChatDto, userId: req.user.id, nickname: req.user.nickname });
   }
 }
diff --git a/Backend/apps/api/src/chats/chats.service.ts b/Backend/apps/api/src/chats/chats.service.ts
@@ -14,7 +14,7 @@ export class ChatsService {
     private readonly configService: ConfigService,
   ) {}
 
-  async publishChat({
+  async ingestChat({
     channelId,
     message,
     userId,
@@ -25,9 +25,19 @@ export class ChatsService {
     userId: number;
     nickname: string;
   }) {
-    const filteringResult = await this.clovaFiltering(message);
-    const chat = JSON.stringify({ content: message, userId, nickname, timestamp: new Date(), filteringResult });
-    await this.redisClient.multi().publish(`${channelId}:chat`, chat).rpush(`${channelId}:chats`, chat).exec();
+    const chatId = crypto.randomUUID();
+    const chat = {
+      content: message,
+      userId,
+      nickname,
+      timestamp: new Date(),
+      channelId,
+      chatId,
+      filteringResult: true,
+    };
+    const chatString = JSON.stringify(chat);
+    await this.redisClient.multi().publish(`${channelId}:chat`, chatString).rpush('chatQueue', chatId).exec();
+    this.clovaFiltering(chat);
   }
 
   async readViewers(channelId: UUID) {
@@ -38,7 +48,7 @@ export class ChatsService {
     this.redisClient.del(`${channelId}:chats`);
   }
 
-  async clovaFiltering(message: string) {
+  async clovaFiltering(chat) {
     const postData = {
       messages: [
         {
@@ -47,9 +57,16 @@ export class ChatsService {
         },
         {
           role: 'user',
-          content: message,
+          content: `채팅내용 : "${chat.content}"`,
         },
       ],
+      maxTokens: this.configService.get<number>('CLOVA_CHAT_FILTERING_MAX_TOKEN') || 10,
+      topP: 0.8,
+      topK: 1,
+      temperature: 0.1,
+      repeatPenalty: 1.0,
+      includeAiFilters: true,
+      seed: 0,
     };
     const { data } = await firstValueFrom(
       this.httpService.post(this.configService.get<string>('CLOVA_API_URL'), postData, {
@@ -61,6 +78,39 @@ export class ChatsService {
       }),
     );
 
-    return data?.result?.message?.content?.includes('true');
+    chat.filteringResult = data?.result?.message?.content?.includes('true');
+    await this.redisClient
+      .multi()
+      .publish(
+        `${chat.channelId}:filter`,
+        JSON.stringify({ chatId: chat.chatId, filteringResult: chat.filteringResult }),
+      )
+      .hset('chatCache', chat.chatId, JSON.stringify(chat))
+      .exec();
+    this.flushChat();
+  }
+
+  async flushChat() {
+    const lockKey = 'chat:flush:lock';
+    const lock = await this.redisClient.set(lockKey, 'lock', 'NX');
+
+    try {
+      if (lock) {
+        while (true) {
+          const frontChatId = await this.redisClient.lindex('chatQueue', 0);
+          const chatString = await this.redisClient.hget('chatCache', frontChatId);
+          if (!chatString) {
+            break;
+          } else {
+            const chat = JSON.parse(chatString);
+            await this.redisClient.multi().rpush(`${chat.channelId}:chats`, chatString).lpop('chatQueue').exec();
+          }
+        }
+      }
+    } catch (err) {
+      console.log(err);
+    } finally {
+      await this.redisClient.del(lockKey);
+    }
   }
 }
diff --git a/Backend/apps/chats/src/chats.gateway.ts b/Backend/apps/chats/src/chats.gateway.ts
@@ -25,13 +25,21 @@ export class ChatsGateway implements OnGatewayDisconnect, OnGatewayConnection {
   constructor(@InjectRedis() private redisClient: Redis) {
     const subscriber = this.redisClient.duplicate();
     subscriber.psubscribe('*:chat');
+    subscriber.psubscribe('*:filter');
     subscriber.on('pmessage', async (pattern, channel, message) => {
       if (pattern === '*:chat') {
-        const channelId = channel.slice(0, -5) as UUID;
+        const channelId = channel.split(':')[0];
         const chat = plainToInstance(ChatDto, message);
         this.emitChat({ channelId, chat });
       }
     });
+
+    subscriber.on('pmessage', async (pattern, channel, message) => {
+      if (pattern === '*:filter') {
+        const channelId = channel.split(':')[0];
+        this.emitFilter({ channelId, filteringResult: JSON.parse(message) });
+      }
+    });
   }
 
   async handleConnection(socket: Socket) {
@@ -67,4 +75,8 @@ export class ChatsGateway implements OnGatewayDisconnect, OnGatewayConnection {
   async emitChat({ channelId, chat }) {
     this.server.to(channelId).emit('chat', [chat]);
   }
+
+  async emitFilter({ channelId, filteringResult }) {
+    this.server.to(channelId).emit('filter', [filteringResult]);
+  }
 }