hyungi_document_server/app/services/study/session_analysis_enqueue.py

"""Phase 4-B v1 세션 분석 enqueue 헬퍼 — finalize/fallback/manual 공유.

3 가지 진입점이 다른 정책으로 동작:
  - auto (finalize/fallback): wrong/unsure < 5 도 enqueue 허용 (worker 가 skipped 처리).
    GET fallback 은 best-effort + idempotent + non-blocking.
  - manual (regenerate endpoint): wrong/unsure < 5 또는 active job 존재 시 즉시 차단.
    사용자 즉시 안내 (UX).

is_stale 정책:
  - active job 있음: is_stale 변경 X
  - active 없음 + 기존 analysis 있음: is_stale=TRUE 박기 + 새 job
  - active 없음 + 기존 analysis 없음: is_stale 처리 X + 새 job
"""

from __future__ import annotations

import logging
from typing import Literal

from sqlalchemy import func, select, update
from sqlalchemy.ext.asyncio import AsyncSession

from models.study_question import StudyQuestionAttempt
from models.study_quiz_session import StudyQuizSession
from models.study_quiz_session_analysis import StudyQuizSessionAnalysis
from models.study_quiz_session_job import StudyQuizSessionJob, enqueue_session_analysis_job

logger = logging.getLogger(__name__)

MIN_ATTEMPTS_FOR_ANALYSIS = 5


async def _count_wrong_unsure(session: AsyncSession, study_quiz_session_id: int) -> int:
    row = await session.execute(
        select(func.count())
        .select_from(StudyQuestionAttempt)
        .where(
            StudyQuestionAttempt.quiz_session_id == study_quiz_session_id,
            StudyQuestionAttempt.outcome.in_(("wrong", "unsure")),
        )
    )
    return int(row.scalar() or 0)


async def _has_active_job(session: AsyncSession, study_quiz_session_id: int) -> bool:
    row = await session.execute(
        select(func.count())
        .select_from(StudyQuizSessionJob)
        .where(
            StudyQuizSessionJob.study_quiz_session_id == study_quiz_session_id,
            StudyQuizSessionJob.status.in_(("pending", "processing")),
        )
    )
    return int(row.scalar() or 0) > 0


async def _has_existing_analysis(session: AsyncSession, study_quiz_session_id: int) -> bool:
    row = await session.execute(
        select(StudyQuizSessionAnalysis.study_quiz_session_id).where(
            StudyQuizSessionAnalysis.study_quiz_session_id == study_quiz_session_id
        )
    )
    return row.scalar_one_or_none() is not None


async def enqueue_session_analysis_auto(
    session: AsyncSession,
    *,
    user_id: int,
    study_quiz_session_id: int,
) -> dict:
    """finalize_session 끝 + GET fallback 공통 자동 트리거.

    wrong/unsure 카운트 무관 enqueue (worker 가 < 5 면 insufficient_attempts skipped).
    active 행은 partial unique idx 가 차단. is_stale 은 안 건드림 (자동 트리거라 UX 무관).
    """
    enqueued = await enqueue_session_analysis_job(
        session,
        study_quiz_session_id=study_quiz_session_id,
        user_id=user_id,
    )
    return {"enqueued": enqueued}


async def request_session_analysis_regenerate(
    session: AsyncSession,
    *,
    user_id: int,
    study_quiz_session_id: int,
) -> dict:
    """사용자 [재생성] 버튼 — manual endpoint 가 호출.

    Returns:
      {'enqueued': bool, 'reason': Literal['insufficient_attempts','already_active'] | None}
    """
    qs = await session.get(StudyQuizSession, study_quiz_session_id)
    if qs is None or qs.user_id != user_id:
        return {"enqueued": False, "reason": "not_found"}
    if qs.status != "done":
        return {"enqueued": False, "reason": "not_done"}

    # 1. wrong/unsure < 5 즉시 차단
    cnt = await _count_wrong_unsure(session, study_quiz_session_id)
    if cnt < MIN_ATTEMPTS_FOR_ANALYSIS:
        return {"enqueued": False, "reason": "insufficient_attempts"}

    # 2. active job 있으면 즉시 차단 (is_stale 변경 X)
    if await _has_active_job(session, study_quiz_session_id):
        return {"enqueued": False, "reason": "already_active"}

    # 3. 기존 analysis 있으면 is_stale=TRUE
    if await _has_existing_analysis(session, study_quiz_session_id):
        await session.execute(
            update(StudyQuizSessionAnalysis)
            .where(StudyQuizSessionAnalysis.study_quiz_session_id == study_quiz_session_id)
            .values(is_stale=True)
        )

    # 4. 새 job
    enqueued = await enqueue_session_analysis_job(
        session,
        study_quiz_session_id=study_quiz_session_id,
        user_id=user_id,
        payload={"trigger": "manual"},
    )
    return {"enqueued": enqueued, "reason": None if enqueued else "race_lost"}