[Queue WAS] 큐 스케줄링

queue-backend/package.json

@@ -28,6 +28,7 @@
     "@nestjs/config": "^4.0.2",
     "@nestjs/core": "^11.0.1",
     "@nestjs/platform-express": "^11.0.1",
+    "@nestjs/schedule": "^6.1.0",
     "@nestjs/swagger": "^11.2.4",
     "@nestjs/typeorm": "^11.0.0",
     "cookie-parser": "^1.4.7",

queue-backend/src/queue/queue.module.ts

@@ -1,9 +1,13 @@
 import { Module } from '@nestjs/common';
 import { QueueService } from './queue.service';
 import { QueueController } from './queue.controller';
+import { ScheduleModule } from '@nestjs/schedule';
+import { QueueWorker } from './queue.worker';
+import { QueueTrigger } from './queue.trigger';
 
 @Module({
-  providers: [QueueService],
+  imports: [ScheduleModule.forRoot()],
+  providers: [QueueService, QueueWorker, QueueTrigger],
   controllers: [QueueController],
 })
 export class QueueModule {}

queue-backend/src/queue/queue.trigger.spec.ts

@@ -0,0 +1,52 @@
+import { Test, TestingModule } from '@nestjs/testing';
+import { QueueTrigger } from './queue.trigger';
+import { QueueWorker } from './queue.worker';
+import { PROVIDERS } from '@beastcamp/shared-constants';
+
+describe('QueueTrigger', () => {
+  let trigger: QueueTrigger;
+  let workerMock: Partial<QueueWorker>;
+  let redisMock: Record<string, jest.Mock>;
+  let subClientMock: Record<string, jest.Mock>;
+
+  beforeEach(async () => {
+    workerMock = {
+      processQueueTransfer: jest.fn().mockResolvedValue(undefined),
+    };
+
+    subClientMock = {
+      subscribe: jest.fn().mockResolvedValue(null),
+      on: jest.fn(),
+      quit: jest.fn(),
+    };
+
+    redisMock = {
+      duplicate: jest.fn().mockReturnValue(subClientMock),
+    };
+
+    const module: TestingModule = await Test.createTestingModule({
+      providers: [
+        QueueTrigger,
+        { provide: QueueWorker, useValue: workerMock },
+        { provide: PROVIDERS.REDIS_QUEUE, useValue: redisMock },
+      ],
+    }).compile();
+
+    trigger = module.get<QueueTrigger>(QueueTrigger);
+  });
+
+  it('handleCron 실행 시 worker의 이동 로직을 호출해야 한다', async () => {
+    await trigger.handleCron();
+    expect(workerMock.processQueueTransfer).toHaveBeenCalled();
+  });
+
+  it('onModuleInit 시 Redis 구독을 설정해야 한다', async () => {
+    await trigger.onModuleInit();
+
+    expect(subClientMock.subscribe).toHaveBeenCalledWith('channel:finish');
+    expect(subClientMock.on).toHaveBeenCalledWith(
+      'message',
+      expect.any(Function),
+    );
+  });
+});

queue-backend/src/queue/queue.trigger.ts

@@ -0,0 +1,48 @@
+import { PROVIDERS } from '@beastcamp/shared-constants';
+import {
+  Inject,
+  Injectable,
+  OnModuleDestroy,
+  OnModuleInit,
+  Logger,
+} from '@nestjs/common';
+import Redis from 'ioredis';
+import { QueueWorker } from './queue.worker';
+import { Cron, CronExpression } from '@nestjs/schedule';
+
+@Injectable()
+export class QueueTrigger implements OnModuleInit, OnModuleDestroy {
+  private readonly logger = new Logger(QueueTrigger.name);
+  private subClient: Redis;
+
+  constructor(
+    @Inject(PROVIDERS.REDIS_QUEUE) private readonly redis: Redis,
+    private readonly worker: QueueWorker,
+  ) {}
+
+  async onModuleInit() {
+    this.subClient = this.redis.duplicate();
+    await this.subClient.subscribe('channel:finish');
+
+    this.subClient.on('message', (channel: string, message: string) => {
+      if (channel === 'channel:finish') {
+        this.logger.log('🔔 작업 완료 메시지 수신 - 즉시 이동 시도');
+        void (async () => {
+          await this.worker.removeActiveUser(message);
+          await this.worker.processQueueTransfer();
+        })().catch((err: Error) => {
+          this.logger.error(`🚨 [트리거 오류] message: ${message}`, err.stack);
+        });
+      }
+    });
+  }
+
+  @Cron(CronExpression.EVERY_MINUTE)
+  async handleCron() {
+    await this.worker.processQueueTransfer();
+  }
+
+  async onModuleDestroy() {
+    await this.subClient?.quit();
+  }
+}

queue-backend/src/queue/queue.worker.spec.ts

@@ -0,0 +1,85 @@
+import { Test, TestingModule } from '@nestjs/testing';
+import { QueueWorker } from './queue.worker';
+import { PROVIDERS, REDIS_KEYS } from '@beastcamp/shared-constants';
+import { ConfigService } from '@nestjs/config';
+import { Logger } from '@nestjs/common';
+
+describe('QueueWorker', () => {
+  let worker: QueueWorker;
+  let redisMock: Record<string, jest.Mock>;
+
+  beforeEach(async () => {
+    redisMock = {
+      transferUser: jest.fn(),
+    };
+
+    const module: TestingModule = await Test.createTestingModule({
+      providers: [
+        QueueWorker,
+        {
+          provide: PROVIDERS.REDIS_QUEUE,
+          useValue: redisMock,
+        },
+        {
+          provide: ConfigService,
+          useValue: {
+            get: jest.fn().mockReturnValue(10),
+          },
+        },
+      ],
+    }).compile();
+
+    worker = module.get<QueueWorker>(QueueWorker);
+  });
+
+  it('대기열 스케줄링 로직 호출 시 커스텀 루아 명령어가 올바른 인자로 실행되어야 한다', async () => {
+    // 상황 설정: 루아 스크립트가 유저 2명을 이동시켰다고 가정
+    const movedUsers = ['user1', 'user2'];
+    redisMock.transferUser.mockResolvedValue(movedUsers);
+
+    await worker.processQueueTransfer();
+
+    // 검증: 루아 명령어가 한 번 호출되었는가?
+    expect(redisMock.transferUser).toHaveBeenCalledTimes(1);
+
+    // 검증: 인자가 순서대로 잘 들어갔는가? (KEYS[1], KEYS[2], ARGV[1], ARGV[2])
+    // 1: 대기큐, 2: 활성큐, 3: MAX_CAPACITY, 4: 타임스탬프(문자열)
+    expect(redisMock.transferUser).toHaveBeenCalledWith(
+      REDIS_KEYS.WAITING_QUEUE,
+      REDIS_KEYS.ACTIVE_QUEUE,
+      10, // MAX_CAPACITY
+      expect.any(String), // Date.now().toString()
+    );
+  });
+
+  it('이동된 유저가 있으면 로그를 남겨야 한다', async () => {
+    const movedUsers = ['user1'];
+    redisMock.transferUser.mockResolvedValue(movedUsers);
+
+    // Logger spy 생성 (선택 사항)
+    const loggerSpy = jest.spyOn(Logger.prototype, 'log').mockImplementation();
+
+    await worker.processQueueTransfer();
+
+    expect(loggerSpy).toHaveBeenCalledWith(
+      expect.stringContaining(
+        '🚀 [입장] 유저 user1님이 활성 큐로 이동했습니다.',
+      ),
+    );
+  });
+
+  it('에러 발생 시 에러 로그를 남겨야 한다', async () => {
+    // 상황 설정: Redis 실행 중 에러 발생
+    redisMock.transferUser.mockRejectedValue(new Error('Redis Error'));
+    const loggerErrorSpy = jest
+      .spyOn(Logger.prototype, 'error')
+      .mockImplementation();
+
+    await worker.processQueueTransfer();
+
+    expect(loggerErrorSpy).toHaveBeenCalledWith(
+      '대기열 스케줄링 중 오류 발생:',
+      expect.any(Error),
+    );
+  });
+});

queue-backend/src/queue/queue.worker.ts

@@ -0,0 +1,72 @@
+import { Injectable, Inject, Logger } from '@nestjs/common';
+import { Redis } from 'ioredis';
+import { REDIS_KEYS, PROVIDERS } from '@beastcamp/shared-constants';
+
+interface RedisWithCommands extends Redis {
+  transferUser(
+    waitQ: string,
+    activeQ: string,
+    maxCapacity: number,
+    now: string,
+  ): Promise<string[]>;
+}
+
+@Injectable()
+export class QueueWorker {
+  private readonly logger = new Logger(QueueWorker.name);
+  private isActive = false;
+  private MAX_CAPACITY = 10; // 활성 큐 최대 용량
+
+  constructor(
+    @Inject(PROVIDERS.REDIS_QUEUE) private readonly redis: RedisWithCommands,
+  ) {}
+
+  async processQueueTransfer() {
+    if (this.isActive) {
+      this.logger.debug('🚫 이미 활성 큐 처리 중입니다.');
+      return;
+    }
+
+    this.isActive = true;
+
+    try {
+      const movedUsers = await this.redis.transferUser(
+        REDIS_KEYS.WAITING_QUEUE,
+        REDIS_KEYS.ACTIVE_QUEUE,
+        this.MAX_CAPACITY,
+        Date.now().toString(),
+      );
+
+      if (movedUsers.length > 0) {
+        this.logger.log(
+          `🚀 [입장] 유저 ${movedUsers.join(', ')}님이 활성 큐로 이동했습니다.`,
+        );
+      }
+    } catch (error) {
+      this.logger.error('대기열 스케줄링 중 오류 발생:', error);
+    }
+
+    this.isActive = false;
+  }
+
+  async removeActiveUser(userId: string) {
+    if (!userId) {
+      return;
+    }
+
+    const statusKey = `status:active:${userId}`;
+
+    try {
+      const removed = await this.redis.zrem(REDIS_KEYS.ACTIVE_QUEUE, userId);
+      await this.redis.del(statusKey);
+
+      if (removed > 0) {
+        this.logger.log(
+          `🛑 [퇴장] 유저 ${userId}님을 활성 큐에서 제거했습니다.`,
+        );
+      }
+    } catch (error) {
+      this.logger.error('활성 큐 제거 중 오류 발생:', error);
+    }
+  }
+}