refactor(be): move chatting filter batch jobs from API server to Socket server (#114)

* refactor: remove batch logic in api server

* refactor: remove socket server chat controller

* feat: add chatting batch in socket server

* test: update chatting unit test

* refactor: remove socket http end point

* feat: add event handler in socket broadcast

* chore: auto-fix linting and formatting issues

---------

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

Author: wlgh1553

apps/server/src/app.module.ts

@@ -1,7 +1,6 @@
 import { MiddlewareConsumer, Module, NestModule } from '@nestjs/common';
 import { ConfigModule } from '@nestjs/config';
 import { APP_FILTER } from '@nestjs/core';
-import { ScheduleModule } from '@nestjs/schedule';
 
 import { AiModule } from '@ai/ai.module';
 import { AuthModule } from '@auth/auth.module';
@@ -16,7 +15,6 @@ import { QuestionsModule } from '@questions/questions.module';
 import { RepliesModule } from '@replies/replies.module';
 import { SessionsModule } from '@sessions/sessions.module';
 import { SessionsAuthModule } from '@sessions-auth/sessions-auth.module';
-import { SocketModule } from '@socket/socket.module';
 import { UploadModule } from '@upload/upload.module';
 import { UsersModule } from '@users/users.module';
 
@@ -27,7 +25,6 @@ import { UsersModule } from '@users/users.module';
       isGlobal: true,
       envFilePath: '.env',
     }),
-    ScheduleModule.forRoot(),
     UsersModule,
     PrismaModule,
     SessionsModule,
@@ -36,7 +33,6 @@ import { UsersModule } from '@users/users.module';
     RepliesModule,
     AuthModule,
     UploadModule,
-    SocketModule,
     ChatsModule,
     LoggerModule,
     AiModule,

apps/server/src/chats/chats.module.ts

@@ -4,11 +4,10 @@ import { ChatsController } from './chats.controller';
 import { ChatsRepository } from './chats.repository';
 import { ChatsService } from './chats.service';
 
-import { LoggerModule } from '@logger/logger.module';
 import { PrismaModule } from '@prisma-alias/prisma.module';
 
 @Module({
-  imports: [PrismaModule, LoggerModule],
+  imports: [PrismaModule],
   providers: [ChatsRepository, ChatsService],
   exports: [ChatsService, ChatsRepository],
   controllers: [ChatsController],

apps/server/src/chats/chats.repository.ts

@@ -1,32 +1,10 @@
 import { Injectable } from '@nestjs/common';
-import { AbuseState } from '@prisma/client';
-
-import { ChatSaveDto } from './chats.service';
 
 import { PrismaService } from '@prisma-alias/prisma.service';
 
 @Injectable()
 export class ChatsRepository {
   constructor(private readonly prisma: PrismaService) {}
-  async save({ sessionId, token, body }: ChatSaveDto) {
-    return await this.prisma.chatting.create({
-      data: { sessionId, createUserToken: token, body },
-      include: {
-        createUserTokenEntity: {
-          select: {
-            user: true,
-          },
-        },
-      },
-    });
-  }
-
-  async update(chattingId: number, abuse: AbuseState) {
-    await this.prisma.chatting.update({
-      where: { chattingId },
-      data: { abuse },
-    });
-  }
 
   async getChatsForInfiniteScroll(sessionId: string, count: number, chatId?: number) {
     return await this.prisma.chatting.findMany({
@@ -47,14 +25,4 @@ export class ChatsRepository {
       take: count,
     });
   }
-
-  async getChatsForFilter(count: number, chatId: number) {
-    return await this.prisma.chatting.findMany({
-      where: {
-        chattingId: { gt: chatId },
-        abuse: AbuseState.PENDING,
-      },
-      take: count,
-    });
-  }
 }

apps/server/src/chats/chats.service.spec.ts

@@ -1,21 +1,12 @@
 import { Test, TestingModule } from '@nestjs/testing';
-import { AbuseState } from '@prisma/client';
 
 import { ChatsRepository } from './chats.repository';
-import { ChatSaveDto, ChatsService } from './chats.service';
-import {
-  MOCK_CHAT_DATA,
-  MOCK_CHAT_DATA_NO_NICKNAME,
-  MOCK_SAVED_CHAT,
-  MOCK_SAVED_CHAT_NO_NICKNAME,
-} from './test-chats.mock';
-
-import { LoggerService } from '@logger/logger.service';
+import { ChatsService } from './chats.service';
+import { MOCK_CHAT_DATA, MOCK_CHAT_DATA_NO_NICKNAME } from './test-chats.mock';
 
 describe('ChatsService', () => {
   let service: ChatsService;
   let chatsRepository: jest.Mocked<ChatsRepository>;
-  let fetchMock: jest.SpyInstance;
 
   beforeEach(async () => {
     const module: TestingModule = await Test.createTestingModule({
@@ -26,78 +17,19 @@ describe('ChatsService', () => {
           useValue: {
             save: jest.fn(),
             getChatsForInfiniteScroll: jest.fn(),
-            getChatsForFilter: jest.fn(),
-            update: jest.fn(),
-          },
-        },
-        {
-          provide: LoggerService,
-          useValue: {
-            log: jest.fn(),
-            error: jest.fn(),
-            warn: jest.fn(),
-            debug: jest.fn(),
           },
         },
       ],
     }).compile();
 
     service = module.get<ChatsService>(ChatsService);
     chatsRepository = module.get(ChatsRepository);
-
-    fetchMock = jest.spyOn(global, 'fetch').mockImplementation(() =>
-      Promise.resolve({
-        ok: true,
-        text: () => Promise.resolve(JSON.stringify({ predicted: '일반어', probability: 0.9 })),
-      } as Response),
-    );
   });
 
   it('서비스가 정의되어 있어야 한다', () => {
     expect(service).toBeDefined();
   });
 
-  describe('saveChat', () => {
-    it('채팅을 저장하고 저장된 데이터를 반환해야 한다', async () => {
-      const data: ChatSaveDto = {
-        sessionId: '123',
-        token: 'mockToken',
-        body: 'Test message',
-      };
-
-      chatsRepository.save.mockResolvedValue(MOCK_SAVED_CHAT);
-
-      const result = await service.saveChat(data);
-      expect(chatsRepository.save).toHaveBeenCalledWith(data);
-      expect(result).toEqual({
-        chattingId: 1,
-        nickname: 'TestUser',
-        content: 'Test chat message',
-        abuse: false,
-      });
-    });
-
-    it('사용자의 닉네임이 없는 경우 "익명"을 반환해야 한다', async () => {
-      const data: ChatSaveDto = {
-        sessionId: '123',
-        token: 'mockToken',
-        body: 'Test message',
-      };
-
-      chatsRepository.save.mockResolvedValue(MOCK_SAVED_CHAT_NO_NICKNAME);
-
-      const result = await service.saveChat(data);
-
-      expect(chatsRepository.save).toHaveBeenCalledWith(data);
-      expect(result).toEqual({
-        chattingId: 1,
-        nickname: '익명',
-        content: 'Test message',
-        abuse: false,
-      });
-    });
-  });
-
   describe('getChatsForInfiniteScroll', () => {
     it('무한 스크롤을 위한 채팅 목록을 조회해야 한다', async () => {
       const sessionId = '123';
@@ -126,73 +58,4 @@ describe('ChatsService', () => {
       expect(result).toEqual([{ chattingId: 10, nickname: '익명', content: 'Message 1', abuse: false }]);
     });
   });
-
-  describe('detectAbuseBatch', () => {
-    it('새로운 채팅들을 필터링하고 상태를 업데이트해야 한다', async () => {
-      const SAFE_WORD = 'Hello';
-      const BAD_WORD = 'Bad word';
-
-      const mockChats = [
-        { ...MOCK_SAVED_CHAT, chattingId: 1, body: SAFE_WORD },
-        { ...MOCK_SAVED_CHAT, chattingId: 2, body: BAD_WORD },
-      ];
-
-      chatsRepository.getChatsForFilter.mockResolvedValue(mockChats);
-
-      fetchMock
-        .mockResolvedValueOnce({
-          ok: true,
-          text: () => Promise.resolve(JSON.stringify({ predicted: '일반어', probability: 0.9 })),
-        } as Response)
-        .mockResolvedValueOnce({
-          ok: true,
-          text: () => Promise.resolve(JSON.stringify({ predicted: '욕설', probability: 0.9 })),
-        } as Response);
-
-      await service.detectAbuseBatch();
-
-      expect(chatsRepository.update).toHaveBeenCalledWith(1, AbuseState.SAFE);
-      expect(chatsRepository.update).toHaveBeenCalledWith(2, AbuseState.BLOCKED);
-    });
-  });
-
-  describe('checkAbuse', () => {
-    it('욕설이 감지되면 true를 반환해야 한다', async () => {
-      fetchMock.mockResolvedValueOnce({
-        ok: true,
-        text: () => Promise.resolve(JSON.stringify({ predicted: '욕설', probability: 0.9 })),
-      } as Response);
-
-      const result = await service.checkAbuse('욕설 내용');
-      expect(result).toBe(true);
-    });
-
-    it('일반어가 감지되면 false를 반환해야 한다', async () => {
-      fetchMock.mockResolvedValueOnce({
-        ok: true,
-        text: () => Promise.resolve(JSON.stringify({ predicted: '일반어', probability: 0.9 })),
-      } as Response);
-
-      const result = await service.checkAbuse('일반 내용');
-      expect(result).toBe(false);
-    });
-
-    it('API 요청이 실패하면 false를 반환해야 한다', async () => {
-      fetchMock.mockResolvedValueOnce({
-        ok: false,
-        status: 500,
-        statusText: 'Internal Server Error',
-      } as Response);
-
-      const result = await service.checkAbuse('테스트 내용');
-      expect(result).toBe(false);
-    });
-
-    it('네트워크 에러가 발생하면 false를 반환해야 한다', async () => {
-      fetchMock.mockRejectedValueOnce(new Error('Network error'));
-
-      const result = await service.checkAbuse('테스트 내용');
-      expect(result).toBe(false);
-    });
-  });
 });

apps/server/src/chats/chats.service.ts

@@ -1,42 +1,10 @@
 import { Injectable } from '@nestjs/common';
-import { Cron, CronExpression } from '@nestjs/schedule';
-import { AbuseState } from '@prisma/client';
 
 import { ChatsRepository } from './chats.repository';
 
-import { LoggerService } from '@logger/logger.service';
-
-export interface ChatSaveDto {
-  sessionId: string;
-  token: string;
-  body: string;
-}
-
-interface SlangPredictResult {
-  predicted: '욕설' | '일반어';
-  probability: number;
-}
-
 @Injectable()
 export class ChatsService {
-  private lastProcessedChattingId = 0;
-  private readonly BATCH_SIZE = 10;
-
-  constructor(
-    private readonly chatsRepository: ChatsRepository,
-    private readonly logger: LoggerService,
-  ) {}
-
-  async saveChat(data: ChatSaveDto) {
-    const chat = await this.chatsRepository.save(data);
-    const { chattingId, createUserTokenEntity, body: content } = chat;
-    return {
-      chattingId,
-      nickname: createUserTokenEntity?.user?.nickname || '익명',
-      content,
-      abuse: chat.abuse === 'BLOCKED' ? true : false,
-    };
-  }
+  constructor(private readonly chatsRepository: ChatsRepository) {}
 
   async getChatsForInfiniteScroll(sessionId: string, count: number, chatId?: number) {
     const chats = await this.chatsRepository.getChatsForInfiniteScroll(sessionId, count, chatId);
@@ -53,69 +21,4 @@ export class ChatsService {
       };
     });
   }
-
-  @Cron(CronExpression.EVERY_MINUTE, { name: 'chatting-abuse-detection' })
-  async detectAbuseBatch() {
-    const startTime = new Date();
-
-    const newChats = await this.chatsRepository.getChatsForFilter(this.BATCH_SIZE, this.lastProcessedChattingId);
-    this.lastProcessedChattingId = newChats.reduce(
-      (max, { chattingId }) => Math.max(max, chattingId),
-      this.lastProcessedChattingId,
-    );
-    const abuseResult = await Promise.all(
-      newChats.map(async ({ chattingId, body, sessionId }) => {
-        const abuse = (await this.checkAbuse(body)) ? AbuseState.BLOCKED : AbuseState.SAFE;
-        this.chatsRepository.update(chattingId, abuse);
-        return { chattingId, sessionId, abuse };
-      }),
-    );
-
-    this.broadcast(abuseResult.filter(({ abuse }) => abuse === AbuseState.BLOCKED));
-
-    const endTime = new Date();
-    const executionTime = endTime.getTime() - startTime.getTime();
-    this.logger.log(
-      `[BATCH] last chatting id: ${this.lastProcessedChattingId} chatting count: ${newChats.length} execution time: ${executionTime}ms `,
-    );
-  }
-
-  async broadcast(abuseChattings: { chattingId: number; sessionId: string }[]) {
-    const SOCKET_SERVER_HOST = process.env.SOCKET_SERVER_HOST;
-    const SOCKET_SERVER_PORT = process.env.SOCKET_SERVER_PORT;
-
-    try {
-      fetch(`${SOCKET_SERVER_HOST}:${SOCKET_SERVER_PORT}/api/socket/abuse-chattings`, {
-        method: 'POST',
-        headers: { 'Content-Type': 'application/json' },
-        body: JSON.stringify({ abuseChattings }),
-      });
-    } catch (error) {
-      this.logger.error('Failed to fetch /api/socket/abuse-chattings', error.stack);
-    }
-  }
-
-  async checkAbuse(content: string) {
-    const CLASSIFIER_SERVER_HOST = process.env.CLASSIFIER_SERVER_HOST;
-    const CLASSIFIER_SERVER_PORT = process.env.CLASSIFIER_SERVER_PORT;
-
-    try {
-      const response = await fetch(`${CLASSIFIER_SERVER_HOST}:${CLASSIFIER_SERVER_PORT}/slang-predict`, {
-        method: 'POST',
-        headers: { 'Content-Type': 'application/json' },
-        body: JSON.stringify({ input: content }),
-      });
-
-      if (!response.ok) {
-        this.logger.error(`Classifier server response: ${response.status}`, response.statusText);
-        return false;
-      }
-
-      const data: SlangPredictResult = JSON.parse(await response.text());
-      return data.predicted === '욕설';
-    } catch (error) {
-      this.logger.error(`Unexpected error while requesting classifier server`, error.stack);
-      return false;
-    }
-  }
 }

apps/server/src/socket/socket.constant.ts

@@ -19,4 +19,5 @@ export const SOCKET_EVENTS = {
 
   HOST_CHANGED: 'hostChanged',
   SESSION_ENDED: 'sessionEnded',
+  CHATTING_FILTERED: 'chattingFiltered',
 } as const;