diff --git a/ai/src/ai_server/analyzer/_progress.py b/ai/src/ai_server/analyzer/_progress.py new file mode 100644 index 00000000..ad48e07d --- /dev/null +++ b/ai/src/ai_server/analyzer/_progress.py @@ -0,0 +1,8 @@ +from __future__ import annotations + +from collections.abc import Awaitable, Callable + +# analyzer 가 단계마다 호출하는 진행 emitter. (phase, message) 만 받으며, +# 실제 발행(user/target/trace 바인딩 + RealTime 직접 publish)은 messaging 레이어가 주입한다. +# analyzer 가 messaging 을 import 하지 않도록(레이어 규칙) 타입만 여기 둔다. +ProgressEmitter = Callable[[str, str], Awaitable[None]] diff --git a/ai/src/ai_server/analyzer/repository_analyzer.py b/ai/src/ai_server/analyzer/repository_analyzer.py index f02b9c27..ca0d867f 100644 --- a/ai/src/ai_server/analyzer/repository_analyzer.py +++ b/ai/src/ai_server/analyzer/repository_analyzer.py @@ -8,6 +8,7 @@ EmbeddingStepError, chunk_embed_and_upsert, ) +from ai_server.analyzer._progress import ProgressEmitter from ai_server.analyzer.sources.github_repo import ( GitHubRepoSourceExtractor, RepositoryFetchError, @@ -66,7 +67,12 @@ async def analyze( default_branch: str = "main", user_id: int | None, analyzed_document_id: int, + progress: ProgressEmitter | None = None, ) -> RepositoryAnalysisResult: + async def emit(phase: str, message: str) -> None: + if progress is not None: + await progress(phase, message) + if user_id is None: raise RepositoryAnalyzeError( code="MISSING_USER_ID", @@ -86,6 +92,7 @@ async def analyze( code=err.code, message=err.message, retriable=err.retriable ) from err + await emit("EXTRACTING", "레포지토리 코드를 가져오는 중…") log.info( "repository.extract.start", repository_id=repository_id, @@ -109,6 +116,7 @@ async def analyze( retriable=False, ) + await emit("SUMMARIZING", "AI가 코드를 분석·요약하는 중…") log.info( "repository.llm.start", repository_id=repository_id, @@ -128,6 +136,7 @@ async def analyze( md_chars=len(analysis.markdown), ) + await emit("EMBEDDING", "분석 결과를 임베딩하는 중…") try: chunk_count = await chunk_embed_and_upsert( document_id=analyzed_document_id, diff --git a/ai/src/ai_server/analyzer/resume_analyzer.py b/ai/src/ai_server/analyzer/resume_analyzer.py index c82f0309..cc46ed52 100644 --- a/ai/src/ai_server/analyzer/resume_analyzer.py +++ b/ai/src/ai_server/analyzer/resume_analyzer.py @@ -8,6 +8,7 @@ EmbeddingStepError, chunk_embed_and_upsert, ) +from ai_server.analyzer._progress import ProgressEmitter from ai_server.analyzer.sources.base import SourceExtractor from ai_server.chain.document_analysis_chain import DocumentAnalyzer from ai_server.core.client import CoreClient @@ -61,7 +62,13 @@ async def analyze( resume_id: int, file_path: str, analyzed_document_id: int, + progress: ProgressEmitter | None = None, ) -> ResumeAnalysisResult: + async def emit(phase: str, message: str) -> None: + if progress is not None: + await progress(phase, message) + + await emit("EXTRACTING", "이력서 텍스트를 추출하는 중…") log.info( "resume.extract.start", resume_id=resume_id, @@ -76,6 +83,7 @@ async def analyze( retriable=False, ) + await emit("SUMMARIZING", "AI가 이력서를 분석·요약하는 중…") log.info( "resume.llm.start", resume_id=resume_id, @@ -96,6 +104,7 @@ async def analyze( md_chars=len(analysis.markdown), ) + await emit("EMBEDDING", "분석 결과를 임베딩하는 중…") try: chunk_count = await chunk_embed_and_upsert( document_id=analyzed_document_id, diff --git a/ai/src/ai_server/config/settings.py b/ai/src/ai_server/config/settings.py index b4d57bfb..86a9ca22 100644 --- a/ai/src/ai_server/config/settings.py +++ b/ai/src/ai_server/config/settings.py @@ -32,6 +32,9 @@ class Settings(BaseSettings): ai_callback_routing_questions: str = "callback.questions" ai_callback_routing_feedback: str = "callback.feedback" ai_callback_routing_voice: str = "callback.voice" + # AI -> RealTime 직접 발행 (분석 단계 진행 상황). Core 를 거치지 않는 휘발성 알림. + ai_realtime_exchange: str = "stackup.realtime" + ai_realtime_routing_user: str = "realtime.user.notify" feedback_rag_top_k: int = 5 # 질문 풀 초기 크기. Core 의 applyPool 이 questions[0] 만 INSERT 하므로 1 로 고정해 토큰 낭비 차단. # 후속 작업에서 풀 저장 도입 시 늘리기 (예: 5). diff --git a/ai/src/ai_server/messaging/consumers/repository_consumer.py b/ai/src/ai_server/messaging/consumers/repository_consumer.py index 97f50ce2..60ef7cc0 100644 --- a/ai/src/ai_server/messaging/consumers/repository_consumer.py +++ b/ai/src/ai_server/messaging/consumers/repository_consumer.py @@ -8,6 +8,7 @@ RepositoryAnalyzer, ) from ai_server.messaging.idempotency import LruIdempotencyStore +from ai_server.messaging.progress import AnalysisProgressNotifier from ai_server.messaging.publisher import CallbackPublisher from ai_server.model.envelope import Envelope from ai_server.model.messages.analyze import ( @@ -26,11 +27,13 @@ def __init__( publisher: CallbackPublisher, idempotency: LruIdempotencyStore, callback_routing_key: str, + progress_notifier: AnalysisProgressNotifier | None = None, ) -> None: self._analyzer = analyzer self._publisher = publisher self._idempotency = idempotency self._callback_routing_key = callback_routing_key + self._progress = progress_notifier async def handle(self, message: AbstractIncomingMessage) -> None: async with message.process(requeue=False): @@ -92,6 +95,16 @@ async def _run_and_build_payload( user_id: int | None, trace_id: str, ) -> AnalysisCallbackPayload: + progress = ( + self._progress.emitter_for( + user_id=user_id, + target_type="REPOSITORY", + target_id=req.repository_id, + trace_id=trace_id, + ) + if self._progress is not None + else None + ) try: result = await self._analyzer.analyze( repository_id=req.repository_id, @@ -99,6 +112,7 @@ async def _run_and_build_payload( default_branch=req.default_branch, user_id=user_id, analyzed_document_id=req.analyzed_document_id, + progress=progress, ) except RepositoryAnalyzeError as err: log.warning( diff --git a/ai/src/ai_server/messaging/consumers/resume_consumer.py b/ai/src/ai_server/messaging/consumers/resume_consumer.py index 8aa90944..f22ed3a0 100644 --- a/ai/src/ai_server/messaging/consumers/resume_consumer.py +++ b/ai/src/ai_server/messaging/consumers/resume_consumer.py @@ -5,6 +5,7 @@ from ai_server.analyzer.resume_analyzer import ResumeAnalyzeError, ResumeAnalyzer from ai_server.messaging.idempotency import LruIdempotencyStore +from ai_server.messaging.progress import AnalysisProgressNotifier from ai_server.messaging.publisher import CallbackPublisher from ai_server.model.envelope import Envelope from ai_server.model.messages.analyze import ( @@ -23,11 +24,13 @@ def __init__( publisher: CallbackPublisher, idempotency: LruIdempotencyStore, callback_routing_key: str, + progress_notifier: AnalysisProgressNotifier | None = None, ) -> None: self._analyzer = analyzer self._publisher = publisher self._idempotency = idempotency self._callback_routing_key = callback_routing_key + self._progress = progress_notifier async def handle(self, message: AbstractIncomingMessage) -> None: async with message.process(requeue=False): @@ -59,7 +62,9 @@ async def handle(self, message: AbstractIncomingMessage) -> None: trace_id=envelope.trace_id, ) - payload = await self._run_and_build_payload(req, envelope.trace_id) + payload = await self._run_and_build_payload( + req, envelope.trace_id, user_id=envelope.context.user_id + ) await self._publisher.publish( routing_key=self._callback_routing_key, @@ -81,12 +86,25 @@ async def _run_and_build_payload( self, req: ResumeAnalyzeRequest, trace_id: str, + *, + user_id: int | None, ) -> AnalysisCallbackPayload: + progress = ( + self._progress.emitter_for( + user_id=user_id, + target_type="RESUME", + target_id=req.resume_id, + trace_id=trace_id, + ) + if self._progress is not None + else None + ) try: result = await self._analyzer.analyze( resume_id=req.resume_id, file_path=req.file_path, analyzed_document_id=req.analyzed_document_id, + progress=progress, ) except ResumeAnalyzeError as err: log.warning( diff --git a/ai/src/ai_server/messaging/progress.py b/ai/src/ai_server/messaging/progress.py new file mode 100644 index 00000000..bef22348 --- /dev/null +++ b/ai/src/ai_server/messaging/progress.py @@ -0,0 +1,86 @@ +from __future__ import annotations + +import structlog + +from ai_server.analyzer._progress import ProgressEmitter +from ai_server.messaging.publisher import CallbackPublisher +from ai_server.model.envelope import MessageContext +from ai_server.model.messages.realtime import ( + AnalysisProgressData, + RealtimeNotifyPayload, +) + +log = structlog.get_logger(__name__) + +ANALYSIS_PROGRESS_EVENT = "ANALYSIS_PROGRESS" + + +class AnalysisProgressNotifier: + """AI -> RealTime 직접 발행. 분석 단계 진행 상황을 user 채널 SSE 로 흘린다. + + 진행 정보는 휘발성이라 발행 실패가 분석을 막아선 안 된다 — 모든 예외를 삼키고 경고만 남긴다. + """ + + def __init__(self, *, publisher: CallbackPublisher, routing_key: str) -> None: + self._publisher = publisher + self._routing_key = routing_key + + async def emit( + self, + *, + user_id: int | None, + target_type: str, + target_id: int, + phase: str, + message: str, + trace_id: str, + ) -> None: + if user_id is None: + return + + payload = RealtimeNotifyPayload( + event_type=ANALYSIS_PROGRESS_EVENT, + data=AnalysisProgressData( + target_type=target_type, + target_id=target_id, + phase=phase, + message=message, + ), + ) + try: + await self._publisher.publish( + routing_key=self._routing_key, + message_type=self._routing_key, + payload=payload, + trace_id=trace_id, + correlation_id=f"progress-{target_type}-{target_id}-{phase}", + context=MessageContext(user_id=user_id), + ) + except Exception: + log.warning( + "analysis.progress.publish_failed", + target_type=target_type, + target_id=target_id, + phase=phase, + trace_id=trace_id, + ) + + def emitter_for( + self, + *, + user_id: int | None, + target_type: str, + target_id: int, + trace_id: str, + ) -> ProgressEmitter: + async def _emit(phase: str, message: str) -> None: + await self.emit( + user_id=user_id, + target_type=target_type, + target_id=target_id, + phase=phase, + message=message, + trace_id=trace_id, + ) + + return _emit diff --git a/ai/src/ai_server/messaging/runner.py b/ai/src/ai_server/messaging/runner.py index eb953790..0e2df2f1 100644 --- a/ai/src/ai_server/messaging/runner.py +++ b/ai/src/ai_server/messaging/runner.py @@ -39,6 +39,7 @@ from ai_server.messaging.consumers.resume_consumer import ResumeConsumer from ai_server.messaging.consumers.web_consumer import WebResumeConsumer from ai_server.messaging.idempotency import LruIdempotencyStore +from ai_server.messaging.progress import AnalysisProgressNotifier from ai_server.messaging.publisher import CallbackPublisher from ai_server.storage.factory import build_storage @@ -59,6 +60,16 @@ def __init__(self, settings: Settings) -> None: exchange_name=settings.ai_callback_exchange, publisher_name=settings.ai_publisher_name, ) + # AI -> RealTime 직접 발행용 (분석 단계 진행 상황). Core 콜백 익스체인지와 별개. + self._realtime_publisher = CallbackPublisher( + connection=self._connection, + exchange_name=settings.ai_realtime_exchange, + publisher_name=settings.ai_publisher_name, + ) + self._progress_notifier = AnalysisProgressNotifier( + publisher=self._realtime_publisher, + routing_key=settings.ai_realtime_routing_user, + ) self._idempotency = LruIdempotencyStore( max_size=settings.ai_idempotency_lru_size, ) @@ -130,12 +141,14 @@ def __init__(self, settings: Settings) -> None: publisher=self._publisher, idempotency=self._idempotency, callback_routing_key=settings.ai_callback_routing_analysis, + progress_notifier=self._progress_notifier, ) self._repository_consumer = RepositoryConsumer( analyzer=repo_analyzer, publisher=self._publisher, idempotency=self._idempotency, callback_routing_key=settings.ai_callback_routing_analysis, + progress_notifier=self._progress_notifier, ) self._web_consumer = WebResumeConsumer( analyzer=web_analyzer, @@ -202,6 +215,7 @@ def __init__(self, settings: Settings) -> None: async def start(self) -> None: await self._connection.open() await self._publisher.open() + await self._realtime_publisher.open() channel = self._connection.channel diff --git a/ai/src/ai_server/model/messages/realtime.py b/ai/src/ai_server/model/messages/realtime.py new file mode 100644 index 00000000..2abf28ae --- /dev/null +++ b/ai/src/ai_server/model/messages/realtime.py @@ -0,0 +1,23 @@ +from pydantic import BaseModel + +from ai_server.model._config import camel_config + + +# AI -> RealTime 직접 발행되는 분석 진행 상황. RealTime 이 user 채널 SSE 로 fan-out 한다. +# 휘발성(진행 표시용)이라 Core/DB 를 거치지 않는다. 종료 상태(ANALYZED/FAILED)는 +# 기존대로 Core 콜백 -> REPO_STATE/DOC_STATE 로 전달된다. +class AnalysisProgressData(BaseModel): + model_config = camel_config() + + target_type: str # "REPOSITORY" | "RESUME" + target_id: int + phase: str # "EXTRACTING" | "SUMMARIZING" | "EMBEDDING" + message: str + + +# RealTime Envelope.payload 형태 ({eventType, data}). Core 의 RealtimeNotifyPayload 와 동일. +class RealtimeNotifyPayload(BaseModel): + model_config = camel_config() + + event_type: str + data: AnalysisProgressData diff --git a/ai/tests/test_progress.py b/ai/tests/test_progress.py new file mode 100644 index 00000000..17ea0882 --- /dev/null +++ b/ai/tests/test_progress.py @@ -0,0 +1,145 @@ +from unittest.mock import AsyncMock + +import pytest + +from ai_server.analyzer.repository_analyzer import RepositoryAnalyzer +from ai_server.analyzer.sources.base import ExtractedSource +from ai_server.chain.document_analysis_chain import DocumentAnalysisResult +from ai_server.messaging.progress import ( + ANALYSIS_PROGRESS_EVENT, + AnalysisProgressNotifier, +) +from ai_server.model.messages.realtime import RealtimeNotifyPayload +from ai_server.rag.chunker import MarkdownChunker +from ai_server.rag.embedder import MockEmbeddingProvider + + +def _notifier() -> tuple[AnalysisProgressNotifier, AsyncMock]: + publisher = AsyncMock() + notifier = AnalysisProgressNotifier( + publisher=publisher, routing_key="realtime.user.notify" + ) + return notifier, publisher + + +@pytest.mark.asyncio +async def test_emit_publishes_progress_envelope() -> None: + notifier, publisher = _notifier() + + await notifier.emit( + user_id=5, + target_type="REPOSITORY", + target_id=4, + phase="EMBEDDING", + message="임베딩 중", + trace_id="trace-1", + ) + + publisher.publish.assert_awaited_once() + kwargs = publisher.publish.await_args.kwargs + assert kwargs["routing_key"] == "realtime.user.notify" + assert kwargs["message_type"] == "realtime.user.notify" + assert kwargs["trace_id"] == "trace-1" + assert kwargs["context"].user_id == 5 + + payload = kwargs["payload"] + assert isinstance(payload, RealtimeNotifyPayload) + assert payload.event_type == ANALYSIS_PROGRESS_EVENT + assert payload.data.target_type == "REPOSITORY" + assert payload.data.target_id == 4 + assert payload.data.phase == "EMBEDDING" + # RealTime Envelope.payload 형태로 직렬화되는지 확인. + dumped = payload.model_dump(by_alias=True) + assert dumped["eventType"] == ANALYSIS_PROGRESS_EVENT + assert dumped["data"]["targetType"] == "REPOSITORY" + + +@pytest.mark.asyncio +async def test_emit_without_user_id_is_noop() -> None: + notifier, publisher = _notifier() + + await notifier.emit( + user_id=None, + target_type="RESUME", + target_id=1, + phase="EXTRACTING", + message="x", + trace_id="t", + ) + + publisher.publish.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_emit_swallows_publish_failure() -> None: + notifier, publisher = _notifier() + publisher.publish = AsyncMock(side_effect=RuntimeError("broker down")) + + # 진행 정보는 휘발성 — 발행 실패가 예외로 전파되어 분석을 막으면 안 된다. + await notifier.emit( + user_id=5, + target_type="REPOSITORY", + target_id=4, + phase="EMBEDDING", + message="x", + trace_id="t", + ) + + +@pytest.mark.asyncio +async def test_emitter_for_binds_context() -> None: + notifier, publisher = _notifier() + + emit = notifier.emitter_for( + user_id=9, target_type="RESUME", target_id=3, trace_id="tr" + ) + await emit("SUMMARIZING", "요약 중") + + kwargs = publisher.publish.await_args.kwargs + assert kwargs["context"].user_id == 9 + assert kwargs["payload"].data.target_type == "RESUME" + assert kwargs["payload"].data.target_id == 3 + assert kwargs["payload"].data.phase == "SUMMARIZING" + + +@pytest.mark.asyncio +async def test_repository_analyzer_emits_phases_in_order() -> None: + extractor = AsyncMock() + extractor.extract = AsyncMock( + return_value=ExtractedSource(text="x" * 300, source_type="REPOSITORY") + ) + chain = AsyncMock() + chain.analyze = AsyncMock( + return_value=DocumentAnalysisResult( + summary="s", tech_stack=["Go"], markdown="## a\n" + "y" * 300 + ) + ) + core_client = AsyncMock() + core_client.fetch_github_token = AsyncMock(return_value="tok") + core_client.upsert_embeddings = AsyncMock(return_value=2) + + analyzer = RepositoryAnalyzer( + extractor=extractor, + core_client=core_client, + chain=chain, + storage=AsyncMock(), + chunker=MarkdownChunker(chunk_size=200, chunk_overlap=50), + embedder=MockEmbeddingProvider(dim=16), + analyzed_key_template="analyzed/repository/{repository_id}/summary.md", + ) + + phases: list[str] = [] + + async def progress(phase: str, _message: str) -> None: + phases.append(phase) + + await analyzer.analyze( + repository_id=7, + repo_full_name="u/r", + default_branch="main", + user_id=42, + analyzed_document_id=88, + progress=progress, + ) + + assert phases == ["EXTRACTING", "SUMMARIZING", "EMBEDDING"] diff --git a/ai/tests/test_repository_consumer.py b/ai/tests/test_repository_consumer.py index 909085b8..9c518e67 100644 --- a/ai/tests/test_repository_consumer.py +++ b/ai/tests/test_repository_consumer.py @@ -90,6 +90,7 @@ async def test_happy_path_publishes_analyzed_callback() -> None: default_branch="main", user_id=1, analyzed_document_id=88, + progress=None, ) payload = _captured_payload(publisher) assert payload.status == "ANALYZED" diff --git a/ai/tests/test_resume_consumer.py b/ai/tests/test_resume_consumer.py index 1f00f471..e1f5d306 100644 --- a/ai/tests/test_resume_consumer.py +++ b/ai/tests/test_resume_consumer.py @@ -89,7 +89,10 @@ async def test_happy_path_publishes_analyzed_callback() -> None: await consumer.handle(_incoming_message(_request_envelope())) analyzer.analyze.assert_awaited_once_with( - resume_id=42, file_path="resumes/raw/123/abc.pdf", analyzed_document_id=77 + resume_id=42, + file_path="resumes/raw/123/abc.pdf", + analyzed_document_id=77, + progress=None, ) payload = _captured_payload(publisher) assert payload.status == "ANALYZED" diff --git a/docs/event-stream.md b/docs/event-stream.md index d87ad8ab..6158190b 100644 --- a/docs/event-stream.md +++ b/docs/event-stream.md @@ -79,6 +79,28 @@ data: } ``` +### 3.2-1 분석 단계 진행 (`event: ANALYSIS_PROGRESS`) + +분석이 끝나기 전, 진행 단계를 사용자에게 보여주기 위한 **휘발성** 이벤트. +종료 상태(`DOC_STATE`/`REPO_STATE`)와 달리 DB/쿼리 캐시를 건드리지 않고 화면의 진행 문구만 갱신한다. + +```json +{ + "data": { + "targetType": "REPOSITORY", + "targetId": 4, + "phase": "EMBEDDING", + "message": "분석 결과를 임베딩하는 중…" + }, + "traceId": "…" +} +``` + +- `targetType` ∈ `REPOSITORY | RESUME` +- `phase` ∈ `EXTRACTING | SUMMARIZING | EMBEDDING` +- **발행 경로**: AI 서버 → (`stackup.realtime` exchange, `realtime.user.notify`) → RealTime → user 채널 SSE. Core 를 거치지 않는다(진행 정보는 영속 대상이 아님). 종료 상태만 기존대로 AI → Core 콜백 → REPO_STATE/DOC_STATE 로 전달. +- 프론트는 `ANALYSIS_PROGRESS` 수신 시 진행 store 갱신(쿼리 무효화 X), `REPO_STATE`/`DOC_STATE`(종료) 수신 시 진행 store clear + 목록 쿼리 무효화. + ### 3.3 세션 메시지 푸시 (`event: session.message`) ```json { diff --git a/docs/messaging.md b/docs/messaging.md index 09997bf7..c20b68d9 100644 --- a/docs/messaging.md +++ b/docs/messaging.md @@ -12,7 +12,7 @@ |----------|------|------| | `stackup.core-to-ai` | topic | Core → AI 작업 요청 | | `stackup.ai-to-core` | topic | AI → Core 결과 회신 | -| `stackup.realtime` | topic | Core → RealTime 세션 알림 | +| `stackup.realtime` | topic | Core·AI → RealTime 알림 (세션/상태, AI 분석 진행) | | `stackup.dlx` | direct | 처리 실패 메시지 격리 (Dead Letter Exchange) | ### Queues (durable) @@ -338,6 +338,14 @@ > 단일 큐 `q.realtime.session.notify` 가 세 라우팅 키(`realtime.session.*`/`realtime.user.*`/`realtime.document.*`)를 모두 바인딩한다. +#### AI → RealTime 직접 발행 (분석 단계 진행) + +분석 **종료** 상태(`ANALYZED`/`FAILED`)는 AI→Core 콜백(`callback.analysis`) → Core 가 `REPO_STATE`/`DOC_STATE` 로 RealTime 에 전달한다. 반면 분석 **진행 중** 단계(휘발성, DB 영속 불필요)는 AI 서버가 `stackup.realtime` exchange 의 `realtime.user.notify` 로 **직접** 발행해 user 채널 SSE 로 흘린다(Core 미경유). + +- 발행: AI `AnalysisProgressNotifier` (envelope 구조는 Core `RealtimeNotifyPublisher` 와 동일 — `payload.eventType` + `context.userId`). +- `payload.eventType` = `ANALYSIS_PROGRESS`, `payload.data` = `{ targetType, targetId, phase, message }`. phase ∈ `EXTRACTING | SUMMARIZING | EMBEDDING`. +- 상세 SSE 스펙: [`event-stream.md §3.2-1`](./event-stream.md). + --- ## 6. 재시도·DLQ 정책 diff --git a/frontend/src/features/repo/ui/RepoList.tsx b/frontend/src/features/repo/ui/RepoList.tsx index 1346a9b2..13b9286b 100644 --- a/frontend/src/features/repo/ui/RepoList.tsx +++ b/frontend/src/features/repo/ui/RepoList.tsx @@ -1,10 +1,11 @@ import { isApiError } from '@/shared/api' +import { useAnalysisProgress } from '@/shared/hooks' import { StatusBadge, type StatusTone } from '@/shared/ui' import { useDeleteRepository, useRegisteredRepositories, } from '../model/useRepositories' -import type { RepositoryStatus } from '../model/types' +import type { RegisteredRepository, RepositoryStatus } from '../model/types' const STATUS_META: Record< RepositoryStatus, @@ -43,40 +44,59 @@ export function RepoList() { return ( ) } + +function RepoRow({ + repo, + deleting, + onDelete, +}: { + repo: RegisteredRepository + deleting: boolean + onDelete: () => void +}) { + const meta = STATUS_META[repo.status] + const progress = useAnalysisProgress('REPOSITORY', repo.id) + // 진행 문구는 분석 진행 중일 때만 의미 있다. 완료/실패 시 store 가 clear 되지만 방어적으로 가드. + const showProgress = + !!progress && (repo.status === 'ANALYZING' || repo.status === 'PENDING') + + return ( +
  • +
    + + {repo.repoFullName} + +

    + {showProgress ? progress.message : (repo.defaultBranch ?? 'main')} +

    +
    +
    + {meta.label} + +
    +
    • + ) +} diff --git a/frontend/src/features/resume/ui/ResumeList.tsx b/frontend/src/features/resume/ui/ResumeList.tsx index ee86bd93..0450f9e6 100644 --- a/frontend/src/features/resume/ui/ResumeList.tsx +++ b/frontend/src/features/resume/ui/ResumeList.tsx @@ -1,8 +1,9 @@ import { isApiError } from '@/shared/api' +import { useAnalysisProgress } from '@/shared/hooks' import { StatusBadge, type StatusTone } from '@/shared/ui' import { useDeleteResume, useResumes } from '../model/useResumes' import { formatFileSize } from '../lib/format' -import type { ResumeStatus } from '../model/types' +import type { Resume, ResumeStatus } from '../model/types' const STATUS_META: Record = { PENDING: { tone: 'info', label: '대기 중' }, @@ -38,35 +39,54 @@ export function ResumeList() { return ( ) } + +function ResumeRow({ + resume, + deleting, + onDelete, +}: { + resume: Resume + deleting: boolean + onDelete: () => void +}) { + const meta = STATUS_META[resume.status] + const progress = useAnalysisProgress('RESUME', resume.id) + const showProgress = + !!progress && + (resume.status === 'ANALYZING' || resume.status === 'PENDING') + + return ( +
  • +
    +

    + {resume.originalFilename} +

    +

    + {showProgress ? progress.message : formatFileSize(resume.fileSize)} +

    +
    +
    + {meta.label} + +
    +
    • + ) +} diff --git a/frontend/src/pages/Workspace/model/useWorkspaceAnalysisStream.ts b/frontend/src/pages/Workspace/model/useWorkspaceAnalysisStream.ts index 77d65b1e..1d328754 100644 --- a/frontend/src/pages/Workspace/model/useWorkspaceAnalysisStream.ts +++ b/frontend/src/pages/Workspace/model/useWorkspaceAnalysisStream.ts @@ -4,26 +4,61 @@ import { createStreamToken } from '@/features/auth' import { resumeKeys } from '@/features/resume' import { repoKeys } from '@/features/repo' import { documentKeys } from '@/features/analysis' -import { useEventStream } from '@/shared/hooks' +import { analysisProgress, useEventStream } from '@/shared/hooks' + +// RealTime SSE data 봉투: { data: , traceId }. +type StreamData = { data?: T; traceId?: string | null } + +type StatePayload = { targetType?: string; targetId?: number } +type ProgressPayload = { + targetType?: string + targetId?: number + phase?: string + message?: string +} + +function unwrap(raw: unknown): T | undefined { + return (raw as StreamData | null)?.data +} // 이력서·레포·문서 쿼리를 무효화 → 화면이 자동으로 최신 상태로 갱신된다. +// 추가로 ANALYSIS_PROGRESS(단계별 진행)는 쿼리 무효화 없이 진행 store 만 갱신한다. export function useWorkspaceAnalysisStream() { const queryClient = useQueryClient() const getToken = useCallback(() => createStreamToken(), []) - const invalidateAll = useCallback(() => { - void queryClient.invalidateQueries({ queryKey: resumeKeys.all }) - void queryClient.invalidateQueries({ queryKey: repoKeys.registered }) - void queryClient.invalidateQueries({ queryKey: documentKeys.all }) - }, [queryClient]) + // 종료 상태(REPO_STATE/DOC_STATE): 진행 표시 제거 + 목록 갱신. + const onStateChange = useCallback( + (raw: unknown) => { + const data = unwrap(raw) + if (data?.targetType && typeof data.targetId === 'number') { + analysisProgress.clear(data.targetType, data.targetId) + } + void queryClient.invalidateQueries({ queryKey: resumeKeys.all }) + void queryClient.invalidateQueries({ queryKey: repoKeys.registered }) + void queryClient.invalidateQueries({ queryKey: documentKeys.all }) + }, + [queryClient], + ) + + // 진행 단계(휘발성): 쿼리 무효화 없이 store 만 갱신. + const onProgress = useCallback((raw: unknown) => { + const data = unwrap(raw) + if (!data?.targetType || typeof data.targetId !== 'number') return + analysisProgress.set(data.targetType, data.targetId, { + phase: data.phase ?? '', + message: data.message ?? '', + }) + }, []) useEventStream({ path: '/realtime/stream/me', getToken, handlers: { - DOC_STATE: invalidateAll, - REPO_STATE: invalidateAll, + DOC_STATE: onStateChange, + REPO_STATE: onStateChange, + ANALYSIS_PROGRESS: onProgress, }, }) } diff --git a/frontend/src/shared/hooks/index.ts b/frontend/src/shared/hooks/index.ts index d85629f0..970ef857 100644 --- a/frontend/src/shared/hooks/index.ts +++ b/frontend/src/shared/hooks/index.ts @@ -1,3 +1,5 @@ export { useTypewriter } from './useTypewriter' export type { UseTypewriterOptions } from './useTypewriter' export { useEventStream } from './useEventStream' +export { useAnalysisProgress, analysisProgress } from './useAnalysisProgress' +export type { AnalysisProgress } from './useAnalysisProgress' diff --git a/frontend/src/shared/hooks/useAnalysisProgress.ts b/frontend/src/shared/hooks/useAnalysisProgress.ts new file mode 100644 index 00000000..7296cd0b --- /dev/null +++ b/frontend/src/shared/hooks/useAnalysisProgress.ts @@ -0,0 +1,48 @@ +import { useSyncExternalStore } from 'react' + +// 분석 단계 진행 상황(휘발성). AI 서버가 RealTime 으로 직접 보내는 ANALYSIS_PROGRESS SSE 를 +// 받아 여기에 저장하고, 레포·이력서 목록이 현재 단계 문구를 표시한다. +// 서버 쿼리 캐시와 무관(쿼리 무효화 없이 갱신)하며, 분석 완료/실패 시 clear 된다. +export type AnalysisProgress = { + phase: string + message: string +} + +const keyOf = (targetType: string, targetId: number) => `${targetType}:${targetId}` + +const store = new Map() +const listeners = new Set<() => void>() + +function emitChange(): void { + for (const listener of listeners) listener() +} + +export const analysisProgress = { + set(targetType: string, targetId: number, value: AnalysisProgress): void { + store.set(keyOf(targetType, targetId), value) + emitChange() + }, + clear(targetType: string, targetId: number): void { + if (store.delete(keyOf(targetType, targetId))) emitChange() + }, + get(targetType: string, targetId: number): AnalysisProgress | undefined { + return store.get(keyOf(targetType, targetId)) + }, + subscribe(listener: () => void): () => void { + listeners.add(listener) + return () => { + listeners.delete(listener) + } + }, +} + +// 동일 (targetType,targetId)의 현재 진행 단계. 없으면 undefined. +export function useAnalysisProgress( + targetType: string, + targetId: number, +): AnalysisProgress | undefined { + return useSyncExternalStore( + analysisProgress.subscribe, + () => analysisProgress.get(targetType, targetId), + ) +}