[HSC-288] recommendation-server-v0.0.2

app/analysis_server/main.py

@@ -0,0 +1,64 @@
+"""FastAPI entrypoint for the ephemeral analysis server."""
+
+from __future__ import annotations
+
+import os
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI, HTTPException
+
+from app.core.config import get_settings
+from app.core.logging import configure_logging
+from app.services.kafka_analysis_consumer_service import KafkaAnalysisConsumerService
+
+settings = get_settings()
+configure_logging(settings.debug)
+
+
+@asynccontextmanager
+async def lifespan(application: FastAPI):
+    consumer_service = KafkaAnalysisConsumerService(settings)
+    await consumer_service.start()
+    application.state.analysis_consumer_service = consumer_service
+    try:
+        yield
+    finally:
+        await consumer_service.stop()
+
+
+def create_app() -> FastAPI:
+    application = FastAPI(title=f"{settings.app_name}-analysis-server", lifespan=lifespan)
+
+    @application.get("/")
+    async def root() -> dict[str, str]:
+        return {"app": settings.app_name, "mode": "analysis-server", "health": "/health", "ready": "/ready"}
+
+    @application.get("/health")
+    async def health() -> dict[str, object]:
+        return application.state.analysis_consumer_service.health_payload()
+
+    @application.get("/ready")
+    async def ready() -> dict[str, object]:
+        payload = application.state.analysis_consumer_service.readiness_payload()
+        if payload["ready"]:
+            return payload
+        raise HTTPException(status_code=503, detail=payload)
+
+    return application
+
+
+app = create_app()
+
+
+def run() -> None:
+    import uvicorn
+
+    uvicorn.run(
+        "app.analysis_server.main:app",
+        host=os.getenv("APP_HOST", "0.0.0.0"),
+        port=int(os.getenv("APP_PORT", "8000")),
+    )
+
+
+if __name__ == "__main__":
+    run()

app/batch/main.py

@@ -10,6 +10,7 @@
 
 from app.core.config import get_settings
 from app.core.logging import configure_logging
+from app.infra.kafka.client_options import build_kafka_client_options
 from app.infra.postgres.analysis_repository import AnalysisRepository
 from app.infra.postgres.client import create_postgres_pool
 from app.infra.postgres.dispatch_outbox_repository import DispatchOutboxRepository
@@ -32,11 +33,11 @@ async def run_once() -> int:
 
     consumer = AIOKafkaConsumer(
         settings.kafka_analysis_request_topic,
-        bootstrap_servers=[s.strip() for s in settings.kafka_bootstrap_servers.split(",") if s.strip()],
         group_id=settings.kafka_consumer_group_id,
         auto_offset_reset=settings.kafka_auto_offset_reset,
         enable_auto_commit=False,
         value_deserializer=lambda value: json.loads(value.decode("utf-8")),
+        **build_kafka_client_options(settings),
     )
 
     processed_count = 0

app/core/config.py

@@ -20,6 +20,7 @@ class Settings(BaseSettings):
     database_ssl: bool = Field(default=False, validation_alias=AliasChoices("DATABASE_SSL", "DB_SSL"))
     openai_api_key: str = ""
     openai_chat_model: str = "gpt-4o-mini"
+    # 쿼리 임베딩용. product.embedding_vector 저장 시 사용한 모델과 동일해야 함 (기본 1536차원, DB VECTOR(1536)).
     openai_embedding_model: str = "text-embedding-3-small"
     recommend_top_k: int = 3
     cache_ttl_days: int = 7
@@ -32,12 +33,24 @@ class Settings(BaseSettings):
     kafka_consumer_enabled: bool = False
     kafka_bootstrap_servers: str = "localhost:9092"
     kafka_analysis_request_topic: str = "analysis.request.v1"
+    kafka_recommendation_topic: str = Field(
+        default="recommendation",
+        validation_alias=AliasChoices("KAFKA_RECOMMENDATION_TOPIC", "RECOMMENDATION_TOPIC"),
+    )
+    kafka_analysis_response_topic: str = "analysis.response.v1"
     kafka_consumer_group_id: str = "counseling-analytics-consumer"
     kafka_auto_offset_reset: str = "earliest"
     kafka_batch_size: int = 1000
     kafka_poll_timeout_ms: int = 1000
     kafka_log_each_message: bool = False
     kafka_log_result_limit: int = 20
+    kafka_response_max_attempts: int = 3
+    kafka_security_protocol: str = Field(default="PLAINTEXT", validation_alias=AliasChoices("KAFKA_SECURITY_PROTOCOL"))
+    kafka_sasl_mechanism: str = Field(default="", validation_alias=AliasChoices("KAFKA_SASL_MECHANISM"))
+    kafka_aws_region: str = Field(default="", validation_alias=AliasChoices("KAFKA_AWS_REGION", "AWS_REGION", "AWS_DEFAULT_REGION"))
+    kafka_max_poll_interval_ms: int = Field(default=1800000, validation_alias=AliasChoices("KAFKA_MAX_POLL_INTERVAL_MS"))
+    kafka_session_timeout_ms: int = Field(default=60000, validation_alias=AliasChoices("KAFKA_SESSION_TIMEOUT_MS"))
+    kafka_heartbeat_interval_ms: int = Field(default=15000, validation_alias=AliasChoices("KAFKA_HEARTBEAT_INTERVAL_MS"))
 
     # PostgreSQL connection for bulk lookup
     postgres_dsn: str = Field(default="", validation_alias=AliasChoices("POSTGRES_DSN", "DB_DSN"))

app/infra/kafka/client_options.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+import asyncio
+import time
+from typing import Any
+
+from aiokafka.abc import AbstractTokenProvider
+from aiokafka.helpers import create_ssl_context
+from aws_msk_iam_sasl_signer import MSKAuthTokenProvider
+
+from app.core.config import Settings
+
+
+class MskIamTokenProvider(AbstractTokenProvider):
+    def __init__(self, region: str) -> None:
+        self._region = region
+        self._token = ""
+        self._expiry_ms = 0
+        self._lock = asyncio.Lock()
+
+    async def token(self) -> str:
+        async with self._lock:
+            now_ms = int(time.time() * 1000)
+            if self._token and now_ms < self._expiry_ms - 60_000:
+                return self._token
+
+            token, expiry_ms = await asyncio.get_running_loop().run_in_executor(
+                None,
+                MSKAuthTokenProvider.generate_auth_token,
+                self._region,
+            )
+            self._token = token
+            self._expiry_ms = int(expiry_ms)
+            return token
+
+
+def build_kafka_client_options(settings: Settings) -> dict[str, Any]:
+    options: dict[str, Any] = {
+        "bootstrap_servers": [server.strip() for server in settings.kafka_bootstrap_servers.split(",") if server.strip()],
+    }
+    security_protocol = settings.kafka_security_protocol.strip().upper()
+    if not security_protocol:
+        return options
+
+    options["security_protocol"] = security_protocol
+    if security_protocol in {"SSL", "SASL_SSL"}:
+        options["ssl_context"] = create_ssl_context()
+
+    if security_protocol.startswith("SASL"):
+        sasl_mechanism = settings.kafka_sasl_mechanism.strip().upper()
+        if not sasl_mechanism:
+            raise RuntimeError("KAFKA_SASL_MECHANISM must be set when using a SASL security protocol.")
+
+        options["sasl_mechanism"] = sasl_mechanism
+        if sasl_mechanism == "OAUTHBEARER":
+            region = settings.kafka_aws_region.strip()
+            if not region:
+                raise RuntimeError("KAFKA_AWS_REGION or AWS_REGION must be set for SASL/OAUTHBEARER.")
+            options["sasl_oauth_token_provider"] = MskIamTokenProvider(region)
+
+    return options

app/infra/postgres/dispatch_outbox_repository.py

@@ -5,21 +5,79 @@ class DispatchOutboxRepository:
     def __init__(self, pool: Pool) -> None:
         self._pool = pool
 
-    async def mark_acked_by_request_ids(self, request_ids: list[str]) -> set[str]:
+    async def load_metadata_by_request_ids(self, request_ids: list[str]) -> dict[str, dict[str, str | None]]:
         if not request_ids:
-            return set()
+            return {}
 
+        sql = """
+        SELECT
+            request_id,
+            chunk_id,
+            type::text AS type,
+            dispatch_status::text AS dispatch_status
+        FROM analysis_dispatch_outbox
+        WHERE request_id = ANY($1::text[])
+        """
+
+        async with self._pool.acquire() as conn:
+            rows = await conn.fetch(sql, request_ids)
+
+        return {
+            str(row["request_id"]): {
+                "chunkId": row["chunk_id"],
+                "type": row["type"],
+                "dispatchStatus": row["dispatch_status"],
+            }
+            for row in rows
+        }
+
+    async def prepare_response_dispatch(self, request_id: str, analysis_status: str) -> bool:
         sql = """
         UPDATE analysis_dispatch_outbox
         SET
-            dispatch_status = 'ACKED'::dispatch_status,
-            analysis_status = 'READY'::analysis_status,
+            type = 'RESPONSE'::dispatch_outbox_type,
+            dispatch_status = 'SENT'::dispatch_status,
+            analysis_status = $2::analysis_status,
+            last_error = NULL,
             updated_at = NOW()
-        WHERE request_id = ANY($1::text[])
+        WHERE request_id = $1
+          AND dispatch_status <> 'ACKED'::dispatch_status
         RETURNING request_id
         """
 
         async with self._pool.acquire() as conn:
-            rows = await conn.fetch(sql, request_ids)
+            row = await conn.fetchrow(sql, request_id, analysis_status)
+
+        return row is not None
+
+    async def mark_response_retry(
+        self,
+        request_id: str,
+        last_error: str,
+        max_attempts: int,
+        analysis_status: str,
+    ) -> str:
+        sql = """
+        UPDATE analysis_dispatch_outbox
+        SET
+            type = 'RESPONSE'::dispatch_outbox_type,
+            dispatch_status = CASE
+                WHEN attempt_count + 1 >= $3 THEN 'DEAD'::dispatch_status
+                ELSE 'RETRY'::dispatch_status
+            END,
+            analysis_status = $4::analysis_status,
+            attempt_count = attempt_count + 1,
+            next_retry_at = CASE
+                WHEN attempt_count + 1 >= $3 THEN NULL
+                ELSE NOW() + INTERVAL '5 minutes'
+            END,
+            last_error = LEFT($2, 1000),
+            updated_at = NOW()
+        WHERE request_id = $1
+        RETURNING dispatch_status::text
+        """
+
+        async with self._pool.acquire() as conn:
+            row = await conn.fetchrow(sql, request_id, last_error, max_attempts, analysis_status)
 
-        return {str(row["request_id"]) for row in rows}
+        return str(row["dispatch_status"]) if row is not None else "RETRY"

app/realtime/api/v1/recommendation.py

@@ -1,20 +1,24 @@
-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, BackgroundTasks, Depends
+from fastapi.responses import Response
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.database import get_db_session
-from app.schemas.recommendation import RecommendationRequest, RecommendationResponse
-from app.services.recommendation_service import get_recommendation
+from app.schemas.recommendation import RecommendationRequest
+from app.services.recommendation_service import run_recommendation_and_publish_to_kafka
 
 router = APIRouter()
 
 
-@router.post("/recommendations", response_model=RecommendationResponse)
+@router.post("/recommendations", status_code=202)
 async def post_recommendations(
     body: RecommendationRequest,
+    background_tasks: BackgroundTasks,
     session: AsyncSession = Depends(get_db_session),
-) -> RecommendationResponse:
-    return await get_recommendation(
-        session=session,
-        member_id=body.member_id,
-        profile_text=body.profile_text,
-    )
+) -> Response:
+    """
+    202 Accepted 즉시 반환. 백그라운드에서 추천 생성 후 Kafka recommendation-topic 발행.
+    Spring이 Kafka consume → persona_recommendation 적재 → CompletableFuture.complete(결과).
+    """
+    _ = session
+    background_tasks.add_task(run_recommendation_and_publish_to_kafka, body.member_id)
+    return Response(status_code=202, content=None)

app/realtime/main.py

@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
+import logging
 import os
+import re
 
 from fastapi import FastAPI
 
@@ -14,10 +16,22 @@
 configure_logging(settings.debug)
 
 
+def _mask_database_url(url: str) -> str:
+    """비밀번호만 마스킹한 URL (연결 대상 확인용)."""
+    if not url:
+        return "(empty)"
+    return re.sub(r"(:[^:@]+)(@)", r":****\2", url, count=1)
+
+
 def create_app() -> FastAPI:
     application = FastAPI(title=settings.app_name)
     application.include_router(api_router, prefix=settings.api_v1_prefix)
 
+    @application.on_event("startup")
+    async def log_db_target() -> None:
+        url = get_settings().effective_database_url
+        logging.info("DB 연결 대상: %s", _mask_database_url(url))
+
     @application.get("/")
     async def root() -> dict[str, str]:
         return {"app": settings.app_name, "docs": "/docs", "health": "/health"}

app/schemas/analysis_request_message.py

@@ -4,6 +4,7 @@
 class AnalysisRequestMessage(BaseModel):
     model_config = ConfigDict(populate_by_name=True, extra="ignore")
 
+    type: str | None = Field(default=None, alias="type")
     dispatch_request_id: str = Field(..., alias="dispatchRequestId", min_length=1)
     case_id: int = Field(..., alias="caseId", ge=1)
     analyzer_version: int = Field(..., alias="analyzerVersion", ge=1)

app/schemas/recommendation.py

@@ -10,13 +10,22 @@ class Segment(str, Enum):
 
 
 class RecommendationRequest(BaseModel):
-    member_id: int
-    profile_text: str | None = None  # 테스트용: 있으면 이 텍스트와 유사한 상품 top-k 추천
+    """POST /recommendations 요청. Body: {"memberId": number}"""
+    member_id: int = Field(..., alias="memberId", description="회원 PK")
+
+    model_config = ConfigDict(populate_by_name=True, serialize_by_alias=True)
 
 
 class RecommendedProductItem(BaseModel):
+    """명세 2.3: Spring/프론트 연동용 추천 상품 한 건. embedding_text는 내부용이라 응답에서 제외."""
+    rank: int = Field(..., serialization_alias="rank")
     product_id: int = Field(..., serialization_alias="productId")
-    reason: str
+    product_name: str = Field(..., serialization_alias="productName")
+    product_type: str = Field(..., serialization_alias="productType")
+    product_price: int = Field(..., serialization_alias="productPrice")
+    sale_price: int = Field(..., serialization_alias="salePrice")
+    tags: list[str] = Field(..., serialization_alias="tags")
+    llm_reason: str = Field(..., serialization_alias="llmReason")
 
     model_config = ConfigDict(serialize_by_alias=True)
 
@@ -27,5 +36,7 @@ class RecommendationResponse(BaseModel):
     recommended_products: list[RecommendedProductItem] = Field(
         ..., serialization_alias="recommendedProducts"
     )
+    source: str = Field(..., serialization_alias="source")  # CACHE | LIVE
+    updated_at: str = Field(..., serialization_alias="updatedAt")  # ISO 8601
 
     model_config = ConfigDict(serialize_by_alias=True)