auto-stock/backend/app/services/db_worker.py

"""Background DB Writer Worker

ARCHITECTURE (CLAUDE.md 3.2):
  - 실시간/콜백 스레드에서는 절대 DB 세션을 직접 열지 않는다.
  - 모든 DB 쓰기 작업은 이 모듈의 db_task_queue 로 위임한다.
  - 워커가 큐를 소비하며 단일 세션에서 순차 처리 → 락 경합 제거.

Supported task types:
  - insert_trade_history       : dict  → TradeHistory INSERT
  - upsert_stock_analysis      : dict  → StockAnalysis single UPSERT
  - bulk_upsert_stock_analysis : list  → StockAnalysis batch UPSERT
  - upsert_stock_ban           : dict  → StockBan UPSERT (loss_tracker)
  - log_mcp_deny               : dict  → MCP 거부 이벤트 로그 (DB 저장 없이 로그만)
"""
import asyncio
import logging
from datetime import date

from app.core.database import AsyncSessionLocal
from app.models.models import TradeHistory

logger = logging.getLogger(__name__)

# ─────────────────────────────────────────────────────────────────────────────
# 전역 큐 — 앱 어디서나 import 하여 put() 호출
# ─────────────────────────────────────────────────────────────────────────────
db_task_queue: asyncio.Queue = asyncio.Queue()


# ─────────────────────────────────────────────────────────────────────────────
# 편의 함수 — 호출 측에서 dict 구성 없이 바로 사용
# ─────────────────────────────────────────────────────────────────────────────
async def enqueue_trade_history(data: dict) -> None:
    """매매일지 INSERT 를 큐에 등록."""
    await db_task_queue.put({"type": "insert_trade_history", "data": data})


async def enqueue_upsert_analysis(row: dict) -> None:
    """단건 StockAnalysis UPSERT 를 큐에 등록."""
    await db_task_queue.put({"type": "upsert_stock_analysis", "data": row})


async def enqueue_bulk_upsert_analysis(rows: list[dict]) -> None:
    """다건 StockAnalysis batch UPSERT 를 큐에 등록."""
    if rows:
        await db_task_queue.put({"type": "bulk_upsert_stock_analysis", "data": rows})


async def enqueue_db_write(task: dict) -> None:
    """범용 DB 쓰기 태스크를 큐에 등록.

    task 는 반드시 'type' 키를 포함해야 하며, 워커가 지원하는 task type 중 하나여야 한다.
    예: {"type": "log_mcp_deny", "reason": "...", "trade_plan": {...}}
    """
    await db_task_queue.put(task)


async def enqueue_stock_ban_upsert(stock_code: str, consecutive_losses: int, ban_until) -> None:
    """StockBan UPSERT 를 큐에 등록 (loss_tracker에서 호출)."""
    await db_task_queue.put({
        "type": "upsert_stock_ban",
        "data": {
            "stock_code": stock_code,
            "consecutive_losses": consecutive_losses,
            "ban_until": ban_until,
        }
    })


# ─────────────────────────────────────────────────────────────────────────────
# 워커 본체
# ─────────────────────────────────────────────────────────────────────────────
async def db_writer_worker() -> None:
    """lifespan 에서 asyncio.create_task() 로 실행되는 영구 백그라운드 워커.

    큐에서 task 를 꺼내 단일 AsyncSession 안에서 처리한 뒤
    commit / rollback 을 자동 관리한다.
    """
    logger.info("🛠️  Background DB Writer Worker started")
    while True:
        task = await db_task_queue.get()
        try:
            task_type = task.get("type")
            data = task.get("data")

            async with AsyncSessionLocal() as session:
                # ── 1. 매매일지 INSERT ─────────────────────────────────────
                if task_type == "insert_trade_history":
                    session.add(TradeHistory(**data))
                    await session.commit()
                    logger.info(
                        "✅ [Worker] TradeHistory saved: %s",
                        data.get("stock_code", "?"),
                    )

                # ── 2. 단건 StockAnalysis UPSERT ──────────────────────────
                elif task_type == "upsert_stock_analysis":
                    from app.services.analysis_db_service import _execute_upsert
                    await _execute_upsert(session, [data])
                    await session.commit()
                    logger.debug(
                        "✅ [Worker] StockAnalysis upsert: %s",
                        data.get("stock_code", "?"),
                    )

                # ── 3. 다건 StockAnalysis batch UPSERT ────────────────────
                elif task_type == "bulk_upsert_stock_analysis":
                    from app.services.analysis_db_service import _execute_upsert
                    await _execute_upsert(session, data)          # data = list[dict]
                    await session.commit()
                    logger.info(
                        "✅ [Worker] Bulk StockAnalysis upsert: %d rows", len(data)
                    )

                # ── 4. StockBan UPSERT ────────────────────────────────────────
                elif task_type == "upsert_stock_ban":
                    from app.models.models import StockBan
                    from sqlalchemy import select

                    stock_code = data.get("stock_code")
                    consecutive_losses = data.get("consecutive_losses")
                    ban_until = data.get("ban_until")

                    result = await session.execute(select(StockBan).where(StockBan.stock_code == stock_code))
                    record = result.scalar_one_or_none()

                    if record is None:
                        record = StockBan(
                            stock_code=stock_code,
                            consecutive_losses=consecutive_losses,
                            ban_until=ban_until,
                        )
                        session.add(record)
                    else:
                        record.consecutive_losses = consecutive_losses
                        record.ban_until = ban_until

                    await session.commit()
                    logger.debug(
                        "✅ [Worker] StockBan upsert: %s (losses=%d, ban_until=%s)",
                        stock_code,
                        consecutive_losses,
                        ban_until.isoformat() if ban_until else "None",
                    )

                # ── 5. MCP 거부 이벤트 로그 ───────────────────────────────────
                elif task_type == "log_mcp_deny":
                    # DB 저장 없이 로그만 기록 (필요 시 별도 테이블 확장 가능)
                    logger.warning(
                        "🛑 [Worker] MCP DENY logged: reason=%s, trade_plan=%s",
                        task.get("reason", "unknown"),
                        task.get("trade_plan"),
                    )

                else:
                    logger.warning("⚠️  [Worker] Unknown task type: %s", task_type)

        except asyncio.CancelledError:
            logger.info("⏹️  DB Writer Worker stopped")
            break
        except Exception as exc:
            logger.error("❌ [Worker] DB task failed: %s", exc, exc_info=True)
            # 실패한 task 는 드롭 — 재시도가 필요하면 retry_count 추가 가능
        finally:
            db_task_queue.task_done()