StackUp RealTime 서버. Go 1.26 + chi + amqp091-go. RabbitMQ consumer + SSE/WebSocket 서버. 본 문서 작성 시점에는 SSE만 활성, WS는 US-Session-03 / US-Voice-01에서 도입.
상위: /CLAUDE.md · 횡단 관심사: /docs/
| 영역 | 기술 |
|---|---|
| Language | Go 1.26.3 (go.mod) |
| HTTP framework | github.com/go-chi/chi/v5 |
| WebSocket | github.com/coder/websocket (현재 미사용, 모듈만 등록) |
| AMQP | github.com/rabbitmq/amqp091-go |
| Config | github.com/caarlos0/env/v11 + godotenv |
| Logging | stdlib log/slog (JSON handler) |
| Validation | github.com/go-playground/validator/v10 |
| 빌드 | Makefile |
realtime/
├── go.mod go.sum Makefile Dockerfile .env.example
├── cmd/
│ └── realtime/
│ └── main.go 조립 (config → 컴포넌트 → run)
└── internal/
├── config/ env-tag 기반 Config struct
├── transport/ chi 라우터 + 미들웨어 + SSE 핸들러
├── session/ sessionId → Subscriber 레지스트리
├── bridge/ MQ envelope 파싱 + 디스패처
├── messaging/ AMQP connection + reconnecting consumer
└── trace/ X-Trace-Id 미들웨어 + context helper
PDF의 Logical Architecture (Transport / Session State / Bridge / Messaging) 에 1:1 매핑.
cmd/realtime ──→ transport ──→ session
├─→ bridge ──→ session
└─→ messaging ──→ (amqp091-go)
trace는 transport, cmd가 import
config는 cmd, internal/* 모두에서 import 가능
원칙:
internal/하위 패키지는 다른 프로젝트에서 import 불가 (Go 표준)transport는 도메인 로직(session,bridge)에 의존하되 역방향 금지messaging은 AMQP 라이브러리만 의존, 도메인 로직 모름
| 패키지 | 책임 |
|---|---|
cmd/realtime |
조립. config 로드, 컴포넌트 wiring, signal-aware shutdown |
internal/config |
환경변수 → 타입 안전 Config |
internal/trace |
X-Trace-Id 추출/생성, context propagation |
internal/session |
sessionId → []*Subscriber. fan-out + slow consumer drop |
internal/transport |
chi 라우터, 미들웨어, /health, /realtime/sessions/{id} SSE |
internal/bridge |
RabbitMQ Envelope 파싱 + Dispatcher (sessionId 라우팅) |
internal/messaging |
AMQP connection + 무한 reconnect 가능한 consumer |
- ❌ PostgreSQL 직접 접근 — 데이터가 필요하면 Core
/internal/*API - ❌ JWT 발급 — Core 책임. RealTime은 Stream 토큰 서명 검증만
- ❌ 비즈니스 로직 (질문 생성, 분석) — AI 또는 Core
- ❌ RabbitMQ에 publish — Core를 통해서만 (
architecture.md §4.1) - ❌ PostgreSQL 접근 — WS 답변도 Core 내부 REST(
POST /api/internal/sessions/{id}/messages)로 프록시. RealTime은 PG·MQ publish 모두 미접근 - ❌ AI WS 오디오 프록시(RT3) — 오디오 전용 순수 바이트 파이프, 비즈니스 로직 없음. 음성 인식·메트릭·콜백 발행은 모두 AI 책임. RealTime은 브라우저↔AI WS 양방향 복사만
| ID | Method | Path | 책임 | 상태 |
|---|---|---|---|---|
| - | GET | /health |
헬스체크 | 활성 |
| RT2 | GET | /realtime/stream/me |
user 채널 SSE (분석 상태). userId는 토큰에서 | 활성 |
| RT2 | GET | /realtime/stream/documents/{id} |
document 채널 SSE (DOCUMENT 스코프 토큰 필요 — 현재 발급처 없음) | 활성 |
| RT2 | GET | /realtime/stream/sessions/{id} |
session 채널 SSE (feedback.ready 등 비-라이브) | 활성 |
| RT1 | WS | /realtime/sessions/{id} |
라이브 텍스트 면접 (서버→클라 push + 클라→서버 답변) | 활성 |
| RT3 | WS | /realtime/sessions/{id}/audio |
실시간 음성 답변 스트림 (오디오 업 ↔ 자막 다운, AI WS 프록시) | 활성 |
모든
/realtime/stream/*(SSE) 및/realtime/sessions/{id}(WS) 는 인증 필요 —?access_token=<stream-token>쿼리로 Core 발급 토큰 검증 (EventSource/WS 헤더 한계 우회).internal/auth미들웨어가 처리. WS는 SSE(/realtime/stream/*)와 다른 path라 Upgrade 헤더 분기가 필요 없다. WS 핸들러(coder/websocket)는 session 채널 fan-out을 그대로 구독(서버→클라)하고, 수신한 답변({type:"answer",content,idempotencyKey?})을internal/core클라이언트로 Core 내부 REST(POST /api/internal/sessions/{id}/messages)에 프록시한다. RT3(/realtime/sessions/{id}/audio)는WSAudioHandler(transport/ws_audio.go)가 브라우저 WS를 AI WS(REALTIME_AI_WS_URL+?sessionId=&messageId=,X-Internal-API-Key헤더)로 양방향 프록시한다.messageId는 사전에 CorePOST /api/sessions/{id}/messages/voice/stream-begin이 만든 placeholder. 오디오 내용은 해석하지 않고 프레임 타입을 보존하며 복사만(copyWS). 브라우저의?contentType=을 AI 로 전달한다 — AI 가 이 값으로 STT 세션의 디코더를 고르므로, 넘기지 않으면 무엇을 보내든audio/webm으로 가정되어 Safari(mp4) 같은 경우 틀어진다. 사용자 제어 값이라 화이트리스트(webm/ogg/mp4/mpeg/wav)를 통과한 base MIME 만 붙이고, 그 외에는 기존 기본값으로 떨어진다.
| Queue | Bind | Consumer | DLQ |
|---|---|---|---|
q.realtime.session.notify |
stackup.realtime exchange, routing keys realtime.session.* · realtime.user.* · realtime.document.* |
RealTime | dlq.q.realtime.session.notify (via stackup.dlx) |
단일 큐가 세 채널(session/user/document) 라우팅 키를 모두 바인딩한다. 채널 판별은 envelope
messageType(realtime.{kind}.notify) →bridge.Envelope.Channel()가context에서 id를 꺼낸다.
발행자: Core 서버. envelope 스키마는 /docs/messaging.md §5.
q.realtime.session.notify 는 x-dead-letter-exchange=stackup.dlx 인자를 가지므로 본 컨슈머가 Nack(false, false) 로 reject 한 메시지는 자동으로 DLQ 로 격리된다. 큐 자체는 infra/rabbitmq/definitions.json 가 import 시점에 declare 하며, Go 측은 QueueDeclarePassive 로 메타데이터만 확인한다.
id: <messageId>
event: <eventType>
data: {"data": <payload data>, "traceId": "<traceId>"}
Heartbeat (proxy keepalive):
: ping <unix-ts>
session.Registry는 sync.RWMutex로 보호.- 한 sessionId에 여러 SSE 연결(다중 디바이스/탭) 가능 → slice.
- slow consumer 처리:
Dispatch가slowTimeout을 초과하면 그 구독자만 drop, 다른 구독자는 영향 없음. - 단일 인스턴스 가정. 다중 인스턴스 전환 시 fanout exchange + 인스턴스별 자기 큐 패턴 필요.
| 변수 | 기본값 | 용도 |
|---|---|---|
REALTIME_LISTEN_ADDR |
:38020 |
HTTP 리슨 주소 |
REALTIME_RABBITMQ_URL |
amqp://stackup:stackup@localhost:38050/ |
AMQP 연결 |
REALTIME_LOG_LEVEL |
info |
slog 레벨 |
REALTIME_QUEUE_NAME |
q.realtime.session.notify |
구독 큐 |
REALTIME_SSE_PING_INTERVAL |
30s |
SSE heartbeat 주기 |
REALTIME_SSE_SLOW_CONSUMER_TIMEOUT |
5s |
구독자 send timeout |
REALTIME_SSE_BUFFER_SIZE |
16 |
구독자별 채널 버퍼 |
REALTIME_JWT_SECRET |
local-development-jwt-secret-must-be-replaced |
Core JWT_SECRET과 동일값 필수 (Stream 토큰 검증 키. 키=SHA-256(secret), HS256). 불일치 시 SSE/WS 인증 401 |
REALTIME_CORE_BASE_URL |
http://localhost:38010 |
Core 내부 REST base URL (WS 답변 프록시). compose 내부는 http://backend:38010 |
REALTIME_INTERNAL_API_KEY |
local-development-internal-api-key |
Core CORE_INTERNAL_API_KEY와 동일값 필수 (AI 서버도 공유). X-Internal-API-Key |
REALTIME_AI_WS_URL |
ws://localhost:8000/internal/voice/stream |
AI 음성 스트림 WS base URL (RT3 오디오 프록시 업스트림). compose 내부는 ws://ai:8000/internal/voice/stream |
REALTIME_WS_WRITE_TIMEOUT |
10s |
WS write 타임아웃 |
make build # bin/realtime
make run # 호스트에서 실행
make test # go test ./...
go test ./... -race # 동시성 검증
docker build -t stackup-realtime ./realtimedocker compose up -d realtime은 루트 compose에 등록되어 있음.
gofmt(Makefilemake fmt)- 패키지명은 단복수 단수
- 외부에 노출할 타입만
PascalCase, 내부는camelCase internal/활용으로 외부 import 차단- 에러는 wrap (
fmt.Errorf("...: %w", err))
- ❌ session.Registry 외부에서
subs맵 직접 조작 → API만 사용 - ❌ goroutine 누수 → 모든 long-running goroutine은
context.Context기반 종료 - ❌ Subscriber 채널을 두 번 close → Unsubscribe는 한 번만
- ❌ AMQP delivery에서 panic → recover middleware 또는 handler에서 catch + Nack
- ❌ blocking write to subscriber.Ch without timeout → 항상 select + slowTimeout
새 SSE 이벤트 타입 추가:
messaging.md §5에 envelope 추가- Core 서버가 publish (별도 PR)
- RealTime은
bridge.Dispatch가 자동 처리 (event_type만 다르고 라우팅 동일)
새 WebSocket 엔드포인트 (RT1/RT3):
transport/ws_*.go추가coder/websocket로 upgradeinternal/session.Registry를 protocol-agnostic으로 그대로 활용
새 환경 변수:
internal/config/config.go에 필드 +env/envDefault태그realtime/.env.example갱신- 본 문서 §10 갱신
- HTTP 서버 +
/health활성 - 멀티채널 SSE 활성 —
/realtime/stream/{me,documents/{id},sessions/{id}}(session/user/document) - Stream 토큰 인증 활성 —
?access_token=검증 (internal/auth, Core와 동일 HS256 규약) - AMQP
q.realtime.session.notifyconsumer 활성 —messageType기반 채널 라우팅 → fan-out - WebSocket(RT1 라이브 면접) 활성 —
/realtime/sessions/{id}서버→클라 push + 클라→서버 답변 프록시(Core 내부 REST) - WebSocket(RT3 실시간 음성) 활성 —
/realtime/sessions/{id}/audio브라우저↔AI WS 오디오 프록시(WSAudioHandler,REALTIME_AI_WS_URL). 오디오 전용 순수 파이프, STT·메트릭·callback.voice는 AI 책임 - 리소스 스코프 검증 활성 — 토큰 진위뿐 아니라
resourceType/resourceId가 요청 경로의 대상과 일치하는지 확인한다(transport/router.go). RealTime 은 DB 를 보지 않으므로, Core 가 소유권을 확인해 발급한 토큰의 범위를 강제하는 것이 유일한 소유권 검사다.sessions/{id}(SSE·WS·audio) →SESSION+ 같은 iddocuments/{id}→DOCUMENT+ 같은 id. 이 채널로는 분석 요약·기술스택·문서 경로가 흐르므로(= 남의 이력서 내용) 검증 없이 열어두면 id 를 바꿔가며 긁을 수 있다. 현재 Core 는 DOCUMENT 스코프 토큰을 발급하지 않고 프론트도 이 채널을 쓰지 않는다 (분석 상태는 user 채널/realtime/stream/me로 받는다)stream/me→ 경로에 id 가 없고 토큰의userId로 채널을 만든다(조작 여지 없음)
- DLQ 활성 — handler 실패 메시지는
dlq.q.realtime.session.notify로 격리 - Prometheus 노출 미구현
각 도입 시 본 문서 갱신.