"""study_memo_card_jobs consumer — APScheduler 1분 간격 (공부 암기노트 Phase 1).

기존 study_queue_consumer / study_session_queue_consumer 와 별도 테이블·별도 consumer 라
자연 격리된다 (정본 제약: 기존 consumer 무수정). study_session_queue_consumer 골격 복제.
  BATCH_SIZE=1, MLX gate Semaphore(1) 공유 — explanation/session 처리 중이면 직렬 대기.
  STALE_MINUTES=10 자체 복구.
dispatch: kind=='card_extract' -> run_card_extract_job, 그 외 -> skipped(else 분기 silent loss 방지).
"""

from __future__ import annotations

from datetime import datetime, timedelta, timezone

from sqlalchemy import select, update
from sqlalchemy.exc import SQLAlchemyError

from core.config import settings
from core.database import async_session
from core.utils import setup_logger
from models.study_memo_card_job import StudyMemoCardJob
from workers.study_memo_card_worker import run_card_extract_job

logger = setup_logger("study_memo_card_jobs_consumer")

BATCH_SIZE = 1
STALE_MINUTES = 10


async def reset_stale_card_jobs() -> None:
    """processing 으로 STALE_MINUTES 이상 방치된 job 을 pending 으로 복구."""
    cutoff = datetime.now(timezone.utc) - timedelta(minutes=STALE_MINUTES)
    try:
        async with async_session() as session:
            stmt = (
                update(StudyMemoCardJob)
                .where(
                    StudyMemoCardJob.status == "processing",
                    StudyMemoCardJob.started_at.is_not(None),
                    StudyMemoCardJob.started_at < cutoff,
                )
                .values(status="pending", started_at=None)
            )
            result = await session.execute(stmt)
            await session.commit()
            n = result.rowcount or 0
            if n > 0:
                logger.warning("study_memo_card_jobs_stale_reset count=%s", n)
    except SQLAlchemyError as e:
        logger.exception("study_memo_card_jobs_stale_reset_failed: %s", e)


async def consume_study_memo_card_queue() -> None:
    """APScheduler 진입점. pending card_extract job 을 BATCH_SIZE 만큼 처리."""
    # 생성 LLM 홀드: claim 자체를 하지 않음 (1분 주기라 로그는 debug).
    if "study_memo_card" in settings.pipeline_held_stages:
        logger.debug("study_memo_card 보류 (pipeline.held_stages)")
        return
    await reset_stale_card_jobs()

    async with async_session() as session:
        rows = (
            await session.execute(
                select(StudyMemoCardJob)
                .where(StudyMemoCardJob.status == "pending")
                .order_by(StudyMemoCardJob.id.asc())
                .limit(BATCH_SIZE)
            )
        ).scalars().all()

    for job_row in rows:
        async with async_session() as s:
            try:
                job = await s.get(StudyMemoCardJob, job_row.id)
                if job is None or job.status != "pending":
                    continue
                if job.kind == "card_extract":
                    await run_card_extract_job(s, job)
                else:
                    # 미지원 kind — lost-in-queue 방지 위해 명시 skipped.
                    job.status = "skipped"
                    job.error_code = "unknown"
                    job.error_message = f"unsupported kind: {job.kind!r}"
                    job.completed_at = datetime.now(timezone.utc)
                await s.commit()
                logger.info(
                    "card_extract_processed id=%s src=%s/%s status=%s error_code=%s attempts=%s",
                    job.id, job.source_kind, job.source_id, job.status, job.error_code,
                    job.attempts,
                )
            except Exception as e:
                await s.rollback()
                logger.exception("card_extract_outer_failed job_id=%s: %s", job_row.id, e)
