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
  - update_trade_history       : dict  → TradeHistory UPDATE by id (체결 확인 후 슬리피지/수량 반영)
  - update_trade_history_by_order_id : dict → TradeHistory UPDATE by order_id (KIS 주문번호 기준)
  - upsert_stock_analysis      : dict  → StockAnalysis single UPSERT
  - bulk_upsert_stock_analysis : list  → StockAnalysis batch UPSERT
  - upsert_stock_ban           : dict  → StockBan UPSERT (loss_tracker)
  - insert_news                 : dict  → News INSERT (뉴스 감성 분석 결과 저장)
  - log_mcp_deny               : dict  → MCP 거부 이벤트 로그 (DB 저장 없이 로그만)
  - upsert_portfolio           : dict  → Portfolio INSERT/UPDATE (strategy_type 포함)
  - upsert_active_trade        : dict  → ActiveTrade UPSERT (strategy_name 포함)
  - close_active_trade         : dict  → ActiveTrade 상태를 'closed'로 변경
  - upsert_stock_ban_v2        : dict  → StockBanV2 UPSERT (user_id, stock_code 스코프 — 멀티테넌트)

멀티테넌트(사용자 스코프) 확장 (docs/multi_tenant_design.md §3.4):
  - insert_trade_history / update_trade_history(_by_order_id) / upsert_portfolio /
    upsert_active_trade / close_active_trade / insert_trade_journal 는 payload
    data["user_id"] 가 있으면 조회 필터·생성 컬럼에 반영하고, 없으면(None) 기존과
    100% 동일하게 동작한다(하위 호환). 새 task 타입을 추가하지 않는다.
  - stock_bans 는 PK가 stock_code 단일이라 사용자 스코프로 확장 불가 → 신규 테이블
    stock_bans_v2(user_id, stock_code UNIQUE 쌍) + 신규 task upsert_stock_ban_v2 사용.
"""
import asyncio
import json
import logging
import uuid
from datetime import date, datetime
from pathlib import Path

from sqlalchemy import select

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

logger = logging.getLogger(__name__)

# ─────────────────────────────────────────────────────────────────────────────
# DLQ (Dead Letter Queue) — 3회 재시도 실패 시 파일로 백업
# ─────────────────────────────────────────────────────────────────────────────
DLQ_DIR = Path.home() / ".cache" / "auto-stock" / "dlq"


async def _save_to_dlq(task: dict, error: str) -> None:
    """3회 재시도 실패한 task를 DLQ 파일로 저장.

    저장 위치: ~/.cache/auto-stock/dlq/{timestamp}_{task_type}_{uuid}.json
    파일 I/O는 asyncio.to_thread로 감싸 이벤트 루프 블로킹을 방지한다.
    """
    try:
        task_type = task.get("type", "unknown")
        filename = (
            f"{datetime.now().strftime('%Y%m%d_%H%M%S')}"
            f"_{task_type}"
            f"_{uuid.uuid4().hex[:8]}.json"
        )
        dlq_entry = {
            "task": task,
            "error": str(error),
            "failed_at": datetime.now().isoformat(),
            "retry_count": task.get("_retry_count", 0),
        }

        def _write():
            DLQ_DIR.mkdir(parents=True, exist_ok=True)
            (DLQ_DIR / filename).write_text(
                json.dumps(dlq_entry, default=str, ensure_ascii=False),
                encoding="utf-8",
            )

        await asyncio.to_thread(_write)
        logger.critical(
            "[DLQ] Task saved to DLQ: %s | type=%s | error=%s",
            filename, task_type, error,
        )
    except Exception as e:
        logger.critical(
            "[DLQ] Failed to save to DLQ: %s | task_type=%s",
            e, task.get("type", "unknown"),
        )


# ─────────────────────────────────────────────────────────────────────────────
# 전역 큐 — 앱 어디서나 import 하여 put() 호출
# [이슈8 수정] asyncio.Queue()를 모듈 로드 시점에 생성하면 이벤트 루프가 없는
# 환경(pytest, 스크립트 임포트 등)에서 RuntimeError가 발생할 수 있음.
# _db_task_queue를 None으로 두고, 첫 접근 시 lazy 초기화하는 패턴으로 변경.
# 외부 코드는 기존처럼 db_task_queue 이름으로 접근 가능 (get_db_task_queue() 내부 참조).
# ─────────────────────────────────────────────────────────────────────────────
_db_task_queue: asyncio.Queue | None = None


def get_db_task_queue() -> asyncio.Queue:
    """이벤트 루프 기동 후 최초 호출 시 Queue를 생성(lazy init)하여 반환."""
    global _db_task_queue
    if _db_task_queue is None:
        _db_task_queue = asyncio.Queue()
    return _db_task_queue


# 하위 호환: 기존 코드가 `db_task_queue.put()` 형태로 직접 참조하는 경우를 위해
# 프록시 객체 대신 get_db_task_queue()를 통해 접근하도록 모듈 수준 alias 제공.
# 단, 모듈 임포트 시점에 이벤트 루프가 없어도 오류가 발생하지 않음.
class _QueueProxy:
    """asyncio.Queue를 lazy init으로 감싸는 투명 프록시."""

    def __getattr__(self, name):
        return getattr(get_db_task_queue(), name)

    async def put(self, item):
        await get_db_task_queue().put(item)

    async def get(self):
        return await get_db_task_queue().get()

    def task_done(self):
        get_db_task_queue().task_done()

    def qsize(self):
        return get_db_task_queue().qsize()

    def empty(self):
        return get_db_task_queue().empty()


db_task_queue: asyncio.Queue = _QueueProxy()  # type: ignore[assignment]


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


async def enqueue_update_trade_history(trade_id: int, updates: dict) -> None:
    """기존 TradeHistory 레코드 UPDATE 를 큐에 등록.

    주로 체결 확인(verify_and_sync) 후 executed_price/quantity/status/slippage 반영 시 사용.

    Args:
        trade_id: 업데이트할 TradeHistory.id
        updates:  업데이트할 필드 dict (예: {"executed_quantity": 10, "status": "COMPLETED"})
    """
    await db_task_queue.put({
        "type": "update_trade_history",
        "data": {"id": trade_id, **updates},
    })


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_trade_journal_raw(data: dict) -> None:
    """TradeJournal INSERT 를 큐에 등록 (순수 dict 전달용)."""
    await db_task_queue.put({"type": "insert_trade_journal", "data": data})


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 enqueue_stock_ban_upsert_v2(
    user_id: int, stock_code: str, consecutive_losses: int, ban_until
) -> None:
    """[v2] StockBanV2 UPSERT 를 큐에 등록 (사용자 스코프 loss_tracker에서 호출).

    stock_bans(PK=stock_code, 단일 계정 전제)와 달리 stock_bans_v2 는
    UNIQUE(user_id, stock_code) 이므로 동일 종목을 사용자별로 독립 추적한다.
    """
    await db_task_queue.put({
        "type": "upsert_stock_ban_v2",
        "data": {
            "user_id": user_id,
            "stock_code": stock_code,
            "consecutive_losses": consecutive_losses,
            "ban_until": ban_until,
        }
    })


async def enqueue_portfolio_upsert(data: dict) -> None:
    """Portfolio INSERT/UPDATE 를 큐에 등록.

    data 필수 키: stock_code, quantity, average_price
    data 선택 키: stock_name, current_price, profit_loss, profit_loss_rate,
                  market, strategy_type ('swing' | 'daytrading')
    """
    await db_task_queue.put({"type": "upsert_portfolio", "data": data})


async def enqueue_active_trade_upsert(data: dict) -> None:
    """ActiveTrade UPSERT 를 큐에 등록.

    data 필수 키: stock_code, quantity, entry_price
    data 선택 키: strategy_name ('swing' | 'day_trade'), stop_loss,
                  take_profit, entry_reason, highest_since_entry
    """
    await db_task_queue.put({"type": "upsert_active_trade", "data": data})


async def enqueue_close_active_trade(stock_code: str, strategy_name: str | None = None) -> None:
    """ActiveTrade 를 'closed' 상태로 변경하는 태스크를 큐에 등록."""
    await db_task_queue.put({
        "type": "close_active_trade",
        "data": {"stock_code": stock_code, "strategy_name": strategy_name},
    })


async def enqueue_close_active_trade_v2(
    user_id: int, stock_code: str, strategy_name: str | None = None
) -> None:
    """[v2] 사용자 스코프 ActiveTrade 를 'closed' 상태로 변경하는 태스크를 큐에 등록.

    close_active_trade 와 동일 task type("close_active_trade")을 재사용한다 —
    핸들러가 data.get("user_id")로 필터를 추가하므로 새 타입이 필요 없다.
    """
    await db_task_queue.put({
        "type": "close_active_trade",
        "data": {"stock_code": stock_code, "strategy_name": strategy_name, "user_id": user_id},
    })


async def enqueue_update_trade_history_by_order_id(order_id: str, updates: dict) -> None:
    """KIS 주문번호(order_id)로 TradeHistory UPDATE 를 큐에 등록.

    trade_id 대신 order_id 로 레코드를 조회하여 업데이트한다.
    enqueue_trade_history() 로 INSERT 한 레코드의 id 를 모르는 경우 사용.

    Args:
        order_id: KIS 주문번호 (TradeHistory.order_id)
        updates:  업데이트할 필드 dict (예: {"executed_quantity": 10, "status": "COMPLETED"})
    """
    await db_task_queue.put({
        "type": "update_trade_history_by_order_id",
        "data": {"order_id": order_id, **updates},
    })


async def enqueue_trade_journal(trade, content: str) -> None:
    """TradeJournal INSERT 를 큐에 등록 (main.py post_market_scheduler 용).

    CLAUDE.md 3.2: 스케줄러(실시간 계층)에서 직접 DB 커밋 금지 → 큐 경유.
    """
    from datetime import datetime
    from app.services.trade_analyzer import _to_float, _parse_strategy, _parse_exit_type

    pl_rate = _to_float(trade.profit_loss_rate)
    strategy = _parse_strategy(trade.entry_reason)
    exit_type = _parse_exit_type(trade.exit_reason)
    mood = "positive" if pl_rate > 0 else ("neutral" if pl_rate > -3.0 else "negative")
    tags = [strategy, exit_type]
    if abs(pl_rate) > 5.0:
        tags.append("significant")

    await db_task_queue.put({
        "type": "insert_trade_journal",
        "data": {
            "trade_id": trade.id,
            "stock_code": trade.stock_code,
            "stock_name": trade.stock_name,
            "journal_date": trade.executed_at or trade.created_at or datetime.utcnow(),
            "content": content,
            "strategy_name": strategy,
            "mood": mood,
            "tags": tags,
        },
    })


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

    큐에서 task 를 꺼내 단일 AsyncSession 안에서 처리한 뒤
    commit / rollback 을 자동 관리한다.
    """
    logger.info("🛠️  Background DB Writer Worker started")
    queue = get_db_task_queue()  # 워커 진입 시 큐 인스턴스 고정
    while True:
        task = await 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. TradeHistory UPDATE (체결 확인 후 슬리피지/수량 반영) ──
                elif task_type == "update_trade_history":
                    trade_id = data.get("id")
                    user_id = data.get("user_id")
                    if trade_id:
                        query = select(TradeHistory).where(TradeHistory.id == trade_id)
                        if user_id is not None:
                            query = query.where(TradeHistory.user_id == user_id)
                        row = await session.execute(query)
                        trade = row.scalar_one_or_none()
                        if trade:
                            for key, val in data.items():
                                if key != "id" and val is not None and hasattr(trade, key):
                                    setattr(trade, key, val)
                            await session.commit()
                            logger.info(
                                "✅ [Worker] TradeHistory updated: id=%s status=%s",
                                trade_id,
                                data.get("status", "?"),
                            )

                # ── 2b. TradeHistory UPDATE by order_id (KIS 주문번호 기준) ──
                elif task_type == "update_trade_history_by_order_id":
                    order_id = data.get("order_id")
                    user_id = data.get("user_id")
                    if order_id:
                        query = select(TradeHistory).where(TradeHistory.order_id == order_id)
                        if user_id is not None:
                            query = query.where(TradeHistory.user_id == user_id)
                        row = await session.execute(query)
                        trade = row.scalar_one_or_none()
                        if trade:
                            for key, val in data.items():
                                if key != "order_id" and val is not None and hasattr(trade, key):
                                    setattr(trade, key, val)
                            await session.commit()
                            logger.info(
                                "✅ [Worker] TradeHistory updated by order_id=%s status=%s",
                                order_id,
                                data.get("status", "?"),
                            )
                        else:
                            logger.warning(
                                "⚠️ [Worker] TradeHistory not found for order_id=%s",
                                order_id,
                            )

                # ── 3. 단건 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 app.core.dt_utils import to_naive_utc

                    stock_code = data.get("stock_code")
                    consecutive_losses = data.get("consecutive_losses")
                    # stock_bans.ban_until은 tz-naive 컬럼 — tz-aware 값이 들어와도 안전하게 저장.
                    ban_until = to_naive_utc(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",
                    )

                # ── 4v2. StockBanV2 UPSERT (user_id, stock_code 스코프) ────────
                elif task_type == "upsert_stock_ban_v2":
                    from app.models.models import StockBanV2
                    from sqlalchemy.dialects.postgresql import insert as pg_insert
                    from app.core.dt_utils import to_naive_utc

                    user_id = data.get("user_id")
                    stock_code = data.get("stock_code")
                    consecutive_losses = data.get("consecutive_losses")
                    # stock_bans_v2.ban_until도 tz-naive 컬럼 — tz-aware 값이 들어와도 안전하게 저장.
                    ban_until = to_naive_utc(data.get("ban_until"))

                    # PostgreSQL INSERT ... ON CONFLICT DO UPDATE — 여러 워커/사용자
                    # 사이클이 동시에 같은 (user_id, stock_code)를 갱신해도 경쟁 안전.
                    stmt = pg_insert(StockBanV2).values(
                        user_id=user_id,
                        stock_code=stock_code,
                        consecutive_losses=consecutive_losses,
                        ban_until=ban_until,
                    )
                    stmt = stmt.on_conflict_do_update(
                        constraint="uq_stock_bans_v2_user_stock",
                        set_={
                            "consecutive_losses": stmt.excluded.consecutive_losses,
                            "ban_until": stmt.excluded.ban_until,
                            "updated_at": datetime.utcnow(),
                        },
                    )
                    await session.execute(stmt)
                    await session.commit()
                    logger.debug(
                        "✅ [Worker] StockBanV2 upsert: user=%s %s (losses=%s, ban_until=%s)",
                        user_id, stock_code, consecutive_losses,
                        ban_until.isoformat() if ban_until else "None",
                    )

                # ── 4b. News INSERT (중복 URL 무시) ────────────────────────────
                elif task_type == "insert_news":
                    from app.models.models import News

                    news_url = data.get("url")
                    is_dup = False
                    # URL 기반 중복 방지 (동일 URL은 skip)
                    if news_url:
                        dup_result = await session.execute(
                            select(News.id).where(News.url == news_url).limit(1)
                        )
                        if dup_result.scalar_one_or_none() is not None:
                            is_dup = True
                            logger.debug(
                                "⏭️ [Worker] News skip (dup URL): %s",
                                (data.get("title") or "?")[:40],
                            )

                    if not is_dup:
                        session.add(News(**data))
                        await session.commit()
                        logger.debug(
                            "✅ [Worker] News saved: %s",
                            (data.get("title") or "?")[:40],
                        )

                # ── 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"),
                    )

                # ── 6. Portfolio UPSERT ────────────────────────────────────────
                elif task_type == "upsert_portfolio":
                    from app.models.models import Portfolio

                    stock_code = data.get("stock_code")
                    user_id = data.get("user_id")
                    query = select(Portfolio).where(Portfolio.stock_code == stock_code)
                    if user_id is not None:
                        query = query.where(Portfolio.user_id == user_id)
                    result = await session.execute(query)
                    record = result.scalar_one_or_none()

                    if record is None:
                        record = Portfolio(**data)
                        session.add(record)
                    else:
                        for key, val in data.items():
                            if key != "stock_code" and val is not None:
                                setattr(record, key, val)

                    await session.commit()
                    logger.debug(
                        "✅ [Worker] Portfolio upsert: %s (strategy=%s)",
                        stock_code,
                        data.get("strategy_type", "swing"),
                    )

                # ── 7. ActiveTrade UPSERT ──────────────────────────────────────
                elif task_type == "upsert_active_trade":
                    from app.models.models import ActiveTrade

                    stock_code = data.get("stock_code")
                    strategy_name = data.get("strategy_name")
                    user_id = data.get("user_id")

                    query = select(ActiveTrade).where(
                        ActiveTrade.stock_code == stock_code,
                        ActiveTrade.status == "active",
                    )
                    if strategy_name:
                        query = query.where(ActiveTrade.strategy_name == strategy_name)
                    if user_id is not None:
                        query = query.where(ActiveTrade.user_id == user_id)
                        result = await session.execute(query)
                        record = result.scalar_one_or_none()
                    else:
                        # legacy(user_id=None) 경로: 사용자 다중화로 동일 종목 active
                        # 행이 2건 이상 존재할 수 있어 scalar_one_or_none() 이
                        # MultipleResultsFound 를 던질 수 있음 → 가장 최근 행 우선.
                        query = query.order_by(ActiveTrade.id.desc()).limit(1)
                        result = await session.execute(query)
                        record = result.scalars().first()

                    if record is None:
                        # remaining_pct_multiply는 신규 생성 시 무시
                        create_data = {k: v for k, v in data.items() if k != "remaining_pct_multiply"}
                        # 필수 컬럼(quantity/entry_price) 없이 신규 생성 요청이 오면
                        # IntegrityError로 3회 재시도만 낭비 → 경고 후 스킵
                        _missing = [k for k in ("quantity", "entry_price") if create_data.get(k) is None]
                        if _missing:
                            logger.warning(
                                "⚠️ [Worker] ActiveTrade upsert 스킵: %s active 행 없음 + 필수 키 누락 %s (keys=%s)",
                                stock_code, _missing, sorted(create_data.keys()),
                            )
                            continue
                        record = ActiveTrade(**create_data)
                        session.add(record)
                    else:
                        # remaining_pct_multiply: 기존 remaining_pct에 비율을 곱함
                        # (MDD 축소 시 기존 분할매도 이력 반영)
                        # data.get() 사용 — pop()으로 task dict를 파괴적으로 변경하면
                        # 재시도 시 이 키가 사라져 no-op으로 성공 처리되는 버그 방지
                        multiply_ratio = data.get("remaining_pct_multiply")
                        if multiply_ratio is not None:
                            current_remaining = getattr(record, "remaining_pct", 1.0) or 1.0
                            record.remaining_pct = float(current_remaining) * float(multiply_ratio)
                        for key, val in data.items():
                            if key not in ("stock_code", "remaining_pct_multiply") and val is not None:
                                setattr(record, key, val)

                    await session.commit()
                    logger.debug(
                        "✅ [Worker] ActiveTrade upsert: %s (strategy=%s)",
                        stock_code,
                        strategy_name,
                    )

                # ── 8. ActiveTrade close ───────────────────────────────────────
                elif task_type == "close_active_trade":
                    from app.models.models import ActiveTrade
                    from datetime import datetime as _dt

                    stock_code = data.get("stock_code")
                    strategy_name = data.get("strategy_name")
                    user_id = data.get("user_id")

                    query = select(ActiveTrade).where(
                        ActiveTrade.stock_code == stock_code,
                        ActiveTrade.status == "active",
                    )
                    if strategy_name:
                        query = query.where(ActiveTrade.strategy_name == strategy_name)
                    if user_id is not None:
                        query = query.where(ActiveTrade.user_id == user_id)
                        result = await session.execute(query)
                        record = result.scalar_one_or_none()
                    else:
                        # legacy(user_id=None) 경로: 사용자 다중화로 동일 종목 active
                        # 행이 2건 이상 존재할 수 있어 scalar_one_or_none() 이
                        # MultipleResultsFound 를 던질 수 있음 → 가장 최근 행 우선.
                        query = query.order_by(ActiveTrade.id.desc()).limit(1)
                        result = await session.execute(query)
                        record = result.scalars().first()

                    if record:
                        record.status = "closed"
                        record.updated_at = _dt.utcnow()
                        await session.commit()
                        logger.debug(
                            "✅ [Worker] ActiveTrade closed: %s (strategy=%s)",
                            stock_code,
                            strategy_name,
                        )

                # ── 9. TradeJournal INSERT ────────────────────────────────────
                elif task_type == "insert_trade_journal":
                    from app.models.models import TradeJournal

                    session.add(TradeJournal(**data))
                    await session.commit()
                    logger.debug(
                        "✅ [Worker] TradeJournal saved: %s",
                        data.get("stock_code", "?"),
                    )

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

        except asyncio.CancelledError:
            # 큐 drain: 종료 전 남은 태스크 처리하여 데이터 손실 방지
            remaining = 0
            while not queue.empty():
                try:
                    drain_task = queue.get_nowait()
                except asyncio.QueueEmpty:
                    # get_nowait()이 실패해 아이템을 꺼내지 못했으므로 task_done() 호출 금지
                    break
                try:
                    drain_type = drain_task.get("type", "unknown")
                    drain_data = drain_task.get("data")
                    async with AsyncSessionLocal() as drain_session:
                        if drain_type == "insert_trade_history":
                            drain_session.add(TradeHistory(**drain_data))
                            await drain_session.commit()
                        elif drain_type in (
                            "update_trade_history",
                            "update_trade_history_by_order_id",
                            "upsert_stock_analysis",
                            "bulk_upsert_stock_analysis",
                            "upsert_stock_ban",
                            "upsert_stock_ban_v2",
                            "upsert_portfolio",
                            "upsert_active_trade",
                            "close_active_trade",
                            "insert_trade_journal",
                        ):
                            # 단순 재큐잉 대신, 중요 태스크만 커밋 시도
                            # insert_trade_history 외에는 복잡한 로직이므로 DLQ 저장
                            await _save_to_dlq(drain_task, "shutdown_drain")
                        else:
                            await _save_to_dlq(drain_task, "shutdown_drain_unknown_type")
                    remaining += 1
                except Exception as drain_err:
                    logger.error(f"[DBWorker] Drain 실패: {drain_err}")
                    try:
                        await _save_to_dlq(drain_task, str(drain_err))
                    except Exception:
                        pass
                    remaining += 1
                finally:
                    queue.task_done()
            if remaining:
                logger.info(f"[DBWorker] 종료 시 {remaining}건 drain 완료")
            logger.info("⏹️  DB Writer Worker stopped")
            break
        except Exception as exc:
            retry_count = task.get("_retry_count", 0)
            max_retries = 3
            task_type = task.get("type", "unknown")

            if retry_count < max_retries:
                task["_retry_count"] = retry_count + 1
                delay = min(2 ** retry_count, 8)  # 1, 2, 4초 지수 백오프
                logger.warning(
                    "⚠️ [Worker] DB task retry %d/%d (type=%s): %s — %ds 후 재시도",
                    retry_count + 1, max_retries, task_type, exc, delay,
                )
                await asyncio.sleep(delay)
                await queue.put(task)  # 큐에 재등록
            else:
                logger.error(
                    "❌ [Worker] DB task DROPPED after %d retries (type=%s): %s",
                    max_retries, task_type, exc, exc_info=True,
                )
                # DLQ: 최종 실패 task를 파일로 백업하여 데이터 손실 방지
                await _save_to_dlq(task, str(exc))
        finally:
            queue.task_done()