feat: 실시간 알림(Realtime) 및 AI 메시징 인프라 구축

.env.example

@@ -16,3 +16,10 @@ MINIO_ROOT_PASSWORD=minioadmin
 MINIO_API_PORT=9000
 MINIO_CONSOLE_PORT=9001
 MINIO_BUCKET=stackup
+
+# AI server
+AI_PORT=8000
+
+# RealTime server
+REALTIME_PORT=8081
+REALTIME_LOG_LEVEL=info

.gitignore

@@ -1,2 +1,6 @@
 .vscode
-.claude
+.claude
+
+.worktrees/
+.env
+!.env.example

CLAUDE.md

@@ -41,11 +41,10 @@ PostgreSQL+pgvector  ·  S3/MinIO
 |--------|------|------|
 | 프론트엔드 | [`frontend/CLAUDE.md`](./frontend/CLAUDE.md) | React 19 + TypeScript + Vite, FSD 구조 |
 | 백엔드 (Core) | [`backend/CLAUDE.md`](./backend/CLAUDE.md) | Java 21 + Spring Boot 4 + JPA + QueryDSL |
+| RealTime 서버 | [`realtime/CLAUDE.md`](./realtime/CLAUDE.md) | Go 1.26 + chi + amqp091-go, SSE/WebSocket |
 | AI 서버 | [`ai/CLAUDE.md`](./ai/CLAUDE.md) | Python 3.13 + FastAPI + LangChain |
 | 인프라 | [`infra/CLAUDE.md`](./infra/CLAUDE.md) | Docker Compose: PG(+pgvector), RabbitMQ, MinIO |
 
-> **RealTime Server (Go)** 는 아직 구현 디렉토리 없음. 추가 시 `realtime/CLAUDE.md` 생성.
-
 각 레이어 하위 슬라이스/도메인 패키지에도 `CLAUDE.md`가 배치되어 있을 수 있다. 작업할 디렉토리에서 가장 가까운 것을 우선 읽는다.
 
 ---
@@ -144,9 +143,9 @@ cd frontend && npm install && npm run dev
 - Phase 1 (MVP) 진행 중. US-01 ~ US-20 우선.
 - 디자인 시스템 토큰 파일 미생성 — 디자인 적용 첫 PR에서 `frontend/src/app/styles/tokens.css` 생성.
 - Spring Security, RabbitMQ starter, Flyway는 백엔드 기능 작성 PR과 함께 도입.
-- `infra/rabbitmq/definitions.json`은 현재 큐 6개 정의 (resume/repo 분석, questions/followup 생성, 콜백 2개). 피드백 큐는 US-24 작업 시 추가.
+- `infra/rabbitmq/definitions.json`은 현재 exchange 3개 (`stackup.core-to-ai`, `stackup.ai-to-core`, `stackup.realtime`), 큐 7개 정의. 피드백 큐는 US-24 작업 시 추가.
 - **Redis 미사용** — 휘발성 데이터(OAuth state, 멱등 키, 질문 풀 캐시)는 PostgreSQL의 short-lived 레코드 또는 Core 서버 인메모리로 처리.
-- 실시간 푸시는 **SSE 단일화** (양방향 WebSocket 미사용). 미디어 스트림만 WebRTC.
-- `RealTime Server (Go)`는 아직 디렉토리 없음. SSE/WebRTC가 필요한 시점에 작성.
+- 실시간 푸시는 **SSE + WebSocket 병행**. SSE는 작업 상태(RT2), WebSocket은 라이브 면접 메시지(RT1) 및 음성 스트림(RT3). 미디어 시그널링은 추후 WebRTC 도입 시점에 검토.
+- `RealTime Server (Go)`는 `realtime/` 디렉토리에 부트스트랩 완료. 현재 SSE만 활성, WS 엔드포인트는 US-Session-03 / US-Voice-01에서 도입.
 
 문서 변경 원칙: 코드와 함께 같은 PR에 포함. 단일 출처 원칙(SSOT) 유지.

ai/CLAUDE.md

@@ -290,12 +290,13 @@ docker run --env-file .env -p 8000:8000 stackup-ai
 
 ---
 
-## 16. 현재 상태 (2026-04 기준)
-
-- FastAPI 부트스트랩 + 헬스체크만 구현
-- RabbitMQ consumer 미구현 → US-09(이력서 분석) 작업 시 도입
-- LangChain 기본 import만 있음, 체인·프롬프트 정의 0
-- pgvector 연동 미구현
-- 음성 모듈은 Phase 2에서 본격 작성
+## 16. 현재 상태 (2026-05 기준)
+
+- FastAPI 부트스트랩 + 헬스체크
+- RabbitMQ consumer **echo만** 구현: `ai.analyze.resume` → `callback.analysis` 즉시 echo 응답 (멱등 LRU + structlog)
+- 비즈니스 로직 (PDF 파싱·LLM·임베딩·S3·pgvector) 미구현 — US-09 본 구현 PR에서 추가
+- `analyze.repository` / `generate.questions` / `generate.followup` consumer는 큐 정의만 존재, 코드 없음
+- LangChain import만 있음, 체인·프롬프트 정의 0
+- 음성 모듈은 Phase 2
 
 각 도입 시 본 문서 갱신.

ai/src/ai_server/config/settings.py

@@ -16,6 +16,14 @@ class Settings(BaseSettings):
     # RabbitMQ
     rabbitmq_url: str = "amqp://stackup:stackup@localhost:5672/"
 
+    # AI consumer
+    ai_queue_resume: str = "ai.analyze.resume"
+    ai_queue_prefetch: int = 10
+    ai_callback_exchange: str = "stackup.ai-to-core"
+    ai_callback_routing_analysis: str = "callback.analysis"
+    ai_publisher_name: str = "ai-server"
+    ai_idempotency_lru_size: int = 1024
+
     # S3 / MinIO
     s3_endpoint_url: str = "http://localhost:9000"
     s3_access_key: str = ""

ai/src/ai_server/main.py

@@ -1,17 +1,28 @@
 from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
 
+import structlog
 from fastapi import FastAPI
 
 from ai_server.api.health import router as health_router
 from ai_server.config.settings import Settings, get_settings
+from ai_server.messaging.runner import MessagingRuntime
 
+log = structlog.get_logger(__name__)
 
-@asynccontextmanager
-async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
-    # Startup
-    yield
-    # Shutdown
+
+def _build_lifespan(settings: Settings):
+    @asynccontextmanager
+    async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
+        runtime = MessagingRuntime(settings)
+        app.state.messaging = runtime
+        try:
+            await runtime.start()
+            yield
+        finally:
+            await runtime.stop()
+
+    return lifespan
 
 
 def create_app(settings: Settings | None = None) -> FastAPI:
@@ -21,7 +32,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
     app = FastAPI(
         title=settings.app_name,
         version=settings.app_version,
-        lifespan=lifespan,
+        lifespan=_build_lifespan(settings),
     )
 
     app.include_router(health_router)

ai/src/ai_server/messaging/connection.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+import structlog
+from aio_pika import connect_robust
+from aio_pika.abc import AbstractRobustChannel, AbstractRobustConnection
+
+log = structlog.get_logger(__name__)
+
+
+class RabbitConnection:
+    """Thin wrapper around aio-pika's connect_robust.
+
+    Holds one connection + one channel. Auto-reconnect is provided by
+    aio-pika; we just expose lifecycle hooks for FastAPI lifespan.
+    """
+
+    def __init__(self, url: str, prefetch: int) -> None:
+        self._url = url
+        self._prefetch = prefetch
+        self._connection: AbstractRobustConnection | None = None
+        self._channel: AbstractRobustChannel | None = None
+
+    async def open(self) -> None:
+        log.info("rabbit.connect.start", url_masked=_mask(self._url))
+        self._connection = await connect_robust(self._url)
+        channel = await self._connection.channel()
+        await channel.set_qos(prefetch_count=self._prefetch)
+        self._channel = channel  # type: ignore[assignment]
+        log.info("rabbit.connect.ok", prefetch=self._prefetch)
+
+    async def close(self) -> None:
+        if self._connection is not None and not self._connection.is_closed:
+            await self._connection.close()
+            log.info("rabbit.connect.closed")
+
+    @property
+    def channel(self) -> AbstractRobustChannel:
+        if self._channel is None:
+            raise RuntimeError("RabbitConnection not opened")
+        return self._channel
+
+
+def _mask(url: str) -> str:
+    # amqp://user:pass@host:port/  →  amqp://***@host:port/
+    if "@" not in url:
+        return url
+    scheme, rest = url.split("://", 1) if "://" in url else ("", url)
+    creds, hostpart = rest.split("@", 1)
+    return f"{scheme}://***@{hostpart}" if scheme else f"***@{hostpart}"

ai/src/ai_server/messaging/consumers/resume_consumer.py

@@ -0,0 +1,86 @@
+from __future__ import annotations
+
+import structlog
+from aio_pika.abc import AbstractIncomingMessage
+
+from ai_server.messaging.idempotency import LruIdempotencyStore
+from ai_server.messaging.publisher import CallbackPublisher
+from ai_server.model.envelope import Envelope
+from ai_server.model.messages.analyze import (
+    AnalysisCallbackPayload,
+    ResumeAnalyzeRequest,
+)
+
+log = structlog.get_logger(__name__)
+
+
+def build_echo_callback(
+    envelope: Envelope[ResumeAnalyzeRequest],
+) -> AnalysisCallbackPayload:
+    rid = envelope.payload.resume_id
+    return AnalysisCallbackPayload(
+        target_type="RESUME",
+        target_id=rid,
+        status="ANALYZED",
+        summary="[ECHO] not yet analyzed",
+        tech_stack=[],
+        document_s3_key=f"echo/resume/{rid}/echo.md",
+        embedding_chunk_count=0,
+    )
+
+
+class ResumeConsumer:
+    def __init__(
+        self,
+        *,
+        publisher: CallbackPublisher,
+        idempotency: LruIdempotencyStore,
+        callback_routing_key: str,
+    ) -> None:
+        self._publisher = publisher
+        self._idempotency = idempotency
+        self._callback_routing_key = callback_routing_key
+
+    async def handle(self, message: AbstractIncomingMessage) -> None:
+        async with message.process(requeue=False):
+            try:
+                envelope = Envelope[ResumeAnalyzeRequest].model_validate_json(
+                    message.body
+                )
+            except Exception as exc:  # parse error → DLQ-ready (auto NACK on raise)
+                log.error(
+                    "resume.parse.failed",
+                    error=str(exc),
+                    delivery_tag=message.delivery_tag,
+                )
+                raise
+
+            if self._idempotency.is_seen_then_mark(envelope.message_id):
+                log.info(
+                    "resume.idempotent.skip",
+                    message_id=envelope.message_id,
+                    trace_id=envelope.trace_id,
+                )
+                return
+
+            log.info(
+                "resume.echo.start",
+                message_id=envelope.message_id,
+                resume_id=envelope.payload.resume_id,
+                trace_id=envelope.trace_id,
+            )
+            callback = build_echo_callback(envelope)
+            await self._publisher.publish(
+                routing_key=self._callback_routing_key,
+                message_type="callback.analysis",
+                payload=callback,
+                trace_id=envelope.trace_id,
+                correlation_id=envelope.message_id,
+                context=envelope.context,
+            )
+            log.info(
+                "resume.echo.done",
+                message_id=envelope.message_id,
+                resume_id=envelope.payload.resume_id,
+                trace_id=envelope.trace_id,
+            )

ai/src/ai_server/messaging/idempotency.py

@@ -0,0 +1,27 @@
+from collections import OrderedDict
+from threading import Lock
+
+
+class LruIdempotencyStore:
+    """Thread-safe LRU set for messageId-based idempotency.
+
+    `is_seen_then_mark(key)` returns True if the key was already present
+    (caller should ACK + skip), or False if it's new (record it and process).
+    """
+
+    def __init__(self, max_size: int) -> None:
+        if max_size <= 0:
+            raise ValueError("max_size must be > 0")
+        self._max_size = max_size
+        self._store: OrderedDict[str, None] = OrderedDict()
+        self._lock = Lock()
+
+    def is_seen_then_mark(self, key: str) -> bool:
+        with self._lock:
+            if key in self._store:
+                self._store.move_to_end(key)
+                return True
+            self._store[key] = None
+            if len(self._store) > self._max_size:
+                self._store.popitem(last=False)
+            return False

ai/src/ai_server/messaging/publisher.py

@@ -0,0 +1,81 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from uuid import uuid4
+
+import structlog
+from aio_pika import DeliveryMode, ExchangeType, Message
+from pydantic import BaseModel
+
+from ai_server.messaging.connection import RabbitConnection
+from ai_server.model.envelope import Envelope, MessageContext
+
+log = structlog.get_logger(__name__)
+
+
+class CallbackPublisher:
+    def __init__(
+        self,
+        connection: RabbitConnection,
+        exchange_name: str,
+        publisher_name: str,
+    ) -> None:
+        self._connection = connection
+        self._exchange_name = exchange_name
+        self._publisher_name = publisher_name
+        self._exchange = None  # set in open()
+
+    async def open(self) -> None:
+        self._exchange = await self._connection.channel.declare_exchange(
+            self._exchange_name,
+            ExchangeType.TOPIC,
+            durable=True,
+            passive=True,  # 정의 파일이 이미 만들었음. 우리는 잡기만.
+        )
+
+    async def publish(
+        self,
+        *,
+        routing_key: str,
+        message_type: str,
+        payload: BaseModel,
+        trace_id: str,
+        correlation_id: str,
+        context: MessageContext,
+        version: str = "v1",
+    ) -> None:
+        if self._exchange is None:
+            raise RuntimeError("CallbackPublisher not opened")
+
+        envelope = Envelope[type(payload)].model_validate(  # type: ignore[misc]
+            {
+                "messageId": str(uuid4()),
+                "messageType": message_type,
+                "version": version,
+                "traceId": trace_id,
+                "publishedAt": datetime.now(timezone.utc),
+                "publisher": self._publisher_name,
+                "payload": payload.model_dump(by_alias=True),
+                "context": context.model_dump(by_alias=True),
+            }
+        )
+
+        body = envelope.model_dump_json(by_alias=True).encode("utf-8")
+        msg = Message(
+            body=body,
+            content_type="application/json",
+            content_encoding="utf-8",
+            delivery_mode=DeliveryMode.PERSISTENT,
+            message_id=envelope.message_id,
+            correlation_id=correlation_id,
+            headers={"x-trace-id": trace_id, "x-attempt": 0},
+        )
+        await self._exchange.publish(msg, routing_key=routing_key)
+        log.info(
+            "rabbit.publish",
+            exchange=self._exchange_name,
+            routing_key=routing_key,
+            message_id=envelope.message_id,
+            correlation_id=correlation_id,
+            trace_id=trace_id,
+        )