feat(be): add websocket progress server

apps/progress/README.md

@@ -0,0 +1,92 @@
+# Progress Server
+
+Iris가 테스트케이스를 하나 채점할 때마다 보내는 `progress` 메시지를 받아,
+채점 진행률을 WebSocket으로 프론트에 실시간 push하는 서버입니다.
+
+```
+Iris ──(RabbitMQ)──▶ Progress Server ──(WebSocket)──▶ Frontend
+```
+
+## 동작 방식
+
+1. `iris.e.direct.judge` exchange(`judge.result` routing key)에 전용 큐
+   `progress.q.judge.result`를 bind합니다.
+   백엔드 큐(`iris.q.judge.result`)와 이름이 달라서, 같은 메시지의 복사본을 각각 받습니다.
+2. `properties.type`이 `progress`인 메시지만 처리하고, 나머지 타입은 ack 후 무시합니다.
+3. 제출별로 받은 `current`를 기록해 `완료 개수 / total`로 진행률을 계산합니다.
+   같은 `current`가 다시 오면 무시합니다(중복 집계 방지).
+4. 계산 결과를 해당 제출을 구독 중인 WebSocket 클라이언트에게 보냅니다.
+5. 모든 테스트케이스를 받으면 `done`, 컴파일 에러·서버 에러면 `error`로 끝납니다.
+
+## Iris 메시지 스펙
+
+| 위치       | 필드         | 설명                             |
+| ---------- | ------------ | -------------------------------- |
+| properties | `type`       | `progress`                       |
+| properties | `message_id` | submission id                    |
+| body       | `total`      | 테스트케이스 개수                |
+| body       | `current`    | 현재 테스트케이스 순서           |
+| body       | `statusCode` | sandbox 실행 결과 코드 (아래 표) |
+
+| 코드 | 상태                     |
+| ---- | ------------------------ |
+| 0    | RUN_SUCCESS              |
+| 1    | CPU_TIME_LIMIT_EXCEEDED  |
+| 2    | REAL_TIME_LIMIT_EXCEEDED |
+| 3    | MEMORY_LIMIT_EXCEEDED    |
+| 4    | RUNTIME_ERROR            |
+| 5    | COMPILE_ERROR            |
+| 6    | SEGMENTATION_FAULT_ERROR |
+| 7    | SERVER_ERROR             |
+
+> `RUN_SUCCESS`는 프로세스가 정상 종료됐다는 뜻일 뿐 정답(Accepted)이 아닙니다.
+> 정답/오답 판정은 기존처럼 `GET submission/{id}`로 확인합니다.
+
+## WebSocket 프로토콜
+
+클라이언트 → 서버
+
+```json
+{ "type": "subscribe", "submissionId": 123 }
+{ "type": "unsubscribe", "submissionId": 123 }
+```
+
+서버 → 클라이언트
+
+```json
+{
+  "type": "progress",
+  "submissionId": 123,
+  "status": "judging",
+  "completed": 2,
+  "total": 5,
+  "percentage": 40,
+  "lastStatusCode": 0,
+  "updatedAt": 1791425556973
+}
+```
+
+- `status`: `judging` | `done` | `error`
+- 구독 시점에 이미 진행 중이거나 막 끝난 채점이면 현재 상태를 즉시 한 번 보냅니다.
+
+## 실행
+
+루트에서 RabbitMQ를 띄운 뒤 실행합니다.
+
+```bash
+docker compose up -d rabbitmq
+pnpm --filter @codedang/progress dev
+```
+
+환경변수는 백엔드와 같은 이름을 쓰며, 값이 없으면 루트 `docker-compose.yml`의 로컬 설정을 사용합니다.
+
+| 변수                     | 기본값      |
+| ------------------------ | ----------- |
+| `WS_PORT`                | `8080`      |
+| `RABBITMQ_HOST`          | `localhost` |
+| `RABBITMQ_PORT`          | `5672`      |
+| `RABBITMQ_DEFAULT_USER`  | `skku`      |
+| `RABBITMQ_DEFAULT_PASS`  | `1234`      |
+| `RABBITMQ_DEFAULT_VHOST` | `vh`        |
+| `RABBITMQ_SSL`           | `false`     |
+.

apps/progress/package.json

@@ -0,0 +1,20 @@
+{
+  "name": "@codedang/progress",
+  "version": "0.0.1",
+  "private": true,
+  "scripts": {
+    "dev": "ts-node-dev --respawn src/server.ts",
+    "typecheck": "tsc --noEmit"
+  },
+  "dependencies": {
+    "amqplib": "^0.10.4",
+    "ws": "^8.18.0"
+  },
+  "devDependencies": {
+    "@types/amqplib": "^0.10.5",
+    "@types/node": "^20.14.0",
+    "@types/ws": "^8.5.10",
+    "ts-node-dev": "^2.0.0",
+    "typescript": "^5.5.0"
+  }
+}

apps/progress/src/progressStore.ts

@@ -0,0 +1,104 @@
+import {
+  SandboxStatusCode,
+  type ProgressEvent,
+  type ProgressStatus,
+  type ProgressUpdateMessage
+} from './types'
+
+interface SubmissionState {
+  submissionId: number
+  total: number
+  /** 재전송 메시지 중복 집계 방지용 */
+  seenCurrents: Set<number>
+  lastStatusCode: SandboxStatusCode
+  status: ProgressStatus
+  updatedAt: number
+}
+
+// 끝난 직후 구독한 클라이언트도 최종 상태를 받을 수 있도록 잠시 보관
+const FINISHED_TTL_MS = 60_000
+// 메시지가 끊긴 제출(Iris 장애, 채점 중단 등) 정리용
+const IDLE_TTL_MS = 5 * 60_000
+
+export class ProgressStore {
+  private readonly states = new Map<number, SubmissionState>()
+  private readonly timers = new Map<number, NodeJS.Timeout>()
+
+  /** 무시한 메시지(중복, 이미 끝난 제출)면 null을 반환한다. */
+  handleProgress(event: ProgressEvent): ProgressUpdateMessage | null {
+    const state = this.states.get(event.submissionId) ?? this.createState(event)
+
+    if (state.status !== 'judging' || state.seenCurrents.has(event.current)) {
+      return null
+    }
+
+    state.seenCurrents.add(event.current)
+    state.total = event.total
+    state.lastStatusCode = event.statusCode
+    state.status = this.resolveStatus(state)
+    state.updatedAt = Date.now()
+
+    this.states.set(state.submissionId, state)
+    this.scheduleCleanup(
+      state.submissionId,
+      state.status === 'judging' ? IDLE_TTL_MS : FINISHED_TTL_MS
+    )
+
+    return this.toMessage(state)
+  }
+
+  getSnapshot(submissionId: number): ProgressUpdateMessage | null {
+    const state = this.states.get(submissionId)
+    return state ? this.toMessage(state) : null
+  }
+
+  private resolveStatus(state: SubmissionState): ProgressStatus {
+    // 컴파일 에러·서버 에러는 이후 테스트케이스가 실행되지 않는다.
+    if (
+      state.lastStatusCode === SandboxStatusCode.COMPILE_ERROR ||
+      state.lastStatusCode === SandboxStatusCode.SERVER_ERROR
+    ) {
+      return 'error'
+    }
+    // current의 시작값(0 or 1)과 무관하도록 받은 개수로 판단
+    if (state.seenCurrents.size >= state.total) {
+      return 'done'
+    }
+    return 'judging'
+  }
+
+  private toMessage(state: SubmissionState): ProgressUpdateMessage {
+    const completed = Math.min(state.seenCurrents.size, state.total)
+    return {
+      type: 'progress',
+      submissionId: state.submissionId,
+      status: state.status,
+      completed,
+      total: state.total,
+      percentage: Math.floor((completed / state.total) * 100),
+      lastStatusCode: state.lastStatusCode,
+      updatedAt: state.updatedAt
+    }
+  }
+
+  private createState(event: ProgressEvent): SubmissionState {
+    return {
+      submissionId: event.submissionId,
+      total: event.total,
+      seenCurrents: new Set(),
+      lastStatusCode: event.statusCode,
+      status: 'judging',
+      updatedAt: Date.now()
+    }
+  }
+
+  private scheduleCleanup(submissionId: number, delayMs: number) {
+    clearTimeout(this.timers.get(submissionId))
+    const timer = setTimeout(() => {
+      this.states.delete(submissionId)
+      this.timers.delete(submissionId)
+    }, delayMs)
+    timer.unref()
+    this.timers.set(submissionId, timer)
+  }
+}

apps/progress/src/rabbitConsumer.ts

@@ -0,0 +1,81 @@
+import amqp from 'amqplib'
+import { isIrisProgressPayload, type ProgressEvent } from './types'
+
+// NestJS 백엔드와 동일한 exchange/routing key (apps/backend/libs/constants/src/rabbitmq.constants.ts)
+const EXCHANGE = 'iris.e.direct.judge'
+const ROUTING_KEY = 'judge.result'
+// 백엔드 큐와 이름이 달라야 같은 메시지의 복사본을 각각 받는다.
+const QUEUE = 'progress.q.judge.result'
+
+const PROGRESS_MESSAGE_TYPE = 'progress'
+
+// 서버가 꺼져 있던 동안 쌓인 오래된 진행률은 버린다.
+const MESSAGE_TTL_MS = 60_000
+
+function toProgressEvent(raw: amqp.ConsumeMessage): ProgressEvent | null {
+  // submission id는 body가 아니라 messageId에 문자열로 온다.
+  const submissionId = Number(raw.properties.messageId)
+  if (!Number.isInteger(submissionId) || submissionId <= 0) {
+    return null
+  }
+
+  const payload: unknown = JSON.parse(raw.content.toString())
+  if (!isIrisProgressPayload(payload)) {
+    return null
+  }
+
+  return { submissionId, ...payload }
+}
+
+export async function startProgressConsumer(
+  amqpUrl: string,
+  onProgress: (event: ProgressEvent) => void
+) {
+  const connection = await amqp.connect(amqpUrl)
+  const channel = await connection.createChannel()
+
+  await channel.assertExchange(EXCHANGE, 'direct', { durable: true })
+  await channel.assertQueue(QUEUE, {
+    durable: true,
+    messageTtl: MESSAGE_TTL_MS
+  })
+  await channel.bindQueue(QUEUE, EXCHANGE, ROUTING_KEY)
+  await channel.prefetch(20)
+
+  await channel.consume(QUEUE, (raw) => {
+    if (!raw) {
+      return
+    }
+
+    if (raw.properties.type !== PROGRESS_MESSAGE_TYPE) {
+      channel.ack(raw)
+      return
+    }
+
+    try {
+      const event = toProgressEvent(raw)
+      if (!event) {
+        console.warn(
+          '[progress-server] invalid progress message',
+          raw.properties.messageId,
+          raw.content.toString()
+        )
+        channel.nack(raw, false, false)
+        return
+      }
+
+      onProgress(event)
+      channel.ack(raw)
+    } catch (e) {
+      console.error('[progress-server] failed to process message', e)
+      // 재시도해도 같은 에러라 requeue하지 않는다.
+      channel.nack(raw, false, false)
+    }
+  })
+
+  console.log(
+    `[progress-server] consuming ${QUEUE} bound to ${EXCHANGE}/${ROUTING_KEY}`
+  )
+
+  return { connection, channel }
+}

apps/progress/src/server.ts

@@ -0,0 +1,39 @@
+import { ProgressStore } from './progressStore'
+import { startProgressConsumer } from './rabbitConsumer'
+import { startWebSocketServer } from './wsServer'
+
+// 백엔드(libs/amqp/src/amqp.module.ts)와 같은 환경변수를 쓴다.
+// 기본값은 루트 docker-compose.yml의 로컬 RabbitMQ 설정이다.
+function getAmqpUrl() {
+  const protocol = process.env.RABBITMQ_SSL === 'true' ? 'amqps' : 'amqp'
+  const user = process.env.RABBITMQ_DEFAULT_USER ?? 'skku'
+  const pass = process.env.RABBITMQ_DEFAULT_PASS ?? '1234'
+  const host = process.env.RABBITMQ_HOST ?? 'localhost'
+  const port = process.env.RABBITMQ_PORT ?? '5672'
+  const vhost = process.env.RABBITMQ_DEFAULT_VHOST ?? 'vh'
+
+  return `${protocol}://${encodeURIComponent(user)}:${encodeURIComponent(pass)}@${host}:${port}/${encodeURIComponent(vhost)}`
+}
+
+async function main() {
+  const wsPort = Number(process.env.WS_PORT ?? 8080)
+
+  const store = new ProgressStore()
+  const { broadcast } = startWebSocketServer(wsPort, store)
+
+  await startProgressConsumer(getAmqpUrl(), (event) => {
+    const update = store.handleProgress(event)
+    if (update) {
+      broadcast(update)
+    }
+  })
+
+  console.log(
+    `[progress-server] websocket listening on ws://localhost:${wsPort}`
+  )
+}
+
+main().catch((e) => {
+  console.error('[progress-server] fatal error', e)
+  process.exit(1)
+})