app/services/dispatch.py

"""Enqueue 디스패치 + 실패 복구 (next_retry_at 게이트 존중).

  - relay_outbox: 발행 대상 outbox 이벤트를 enqueue. next_retry_at 과 task 적격성을 확인.
  - repair_stuck_queued: 직접 enqueue 하지 않는다. QUEUED 인데 미발행 outbox 가 없는 task 에
    outbox 이벤트만 보충(next_retry_at 승계). 실제 dispatch 는 relay_outbox 가 담당.
"""
from __future__ import annotations

import logging
from collections.abc import Callable
from datetime import datetime, timezone
from typing import Any

from sqlalchemy import update
from sqlalchemy.orm import Session

from app.domain.states import TaskStatus
from app.models.outbox import OutboxEvent
from app.models.photo_task import PhotoTask
from app.services import task_repo

logger = logging.getLogger(__name__)

EnqueueFn = Callable[[str, int], Any]


def _utcnow() -> datetime:
    return datetime.now(timezone.utc)


def _aware(dt: datetime | None) -> datetime | None:
    """sqlite 는 tz 를 잃은 naive 를 돌려준다 — UTC 로 간주해 비교 안전성 확보."""
    if dt is not None and dt.tzinfo is None:
        return dt.replace(tzinfo=timezone.utc)
    return dt


def default_enqueue(task_id: str, dispatch_version: int = 0) -> Any:
    """실제 Celery enqueue. payload 에 dispatch_version 을 실어 stale dispatch 를 차단."""
    from app.workers.tasks import process_photo

    return process_photo.delay(task_id, dispatch_version)


def _job_id(async_result: Any) -> str | None:
    return getattr(async_result, "id", None)


def mark_published(db: Session, task_id: str) -> None:
    """해당 task 의 미발행 outbox 이벤트를 발행 완료로 마킹(upload 성공 직후)."""
    now = _utcnow()
    events = (
        db.query(OutboxEvent)
        .filter(OutboxEvent.task_id == task_id, OutboxEvent.published.is_(False))
        .all()
    )
    for event in events:
        event.published = True
        event.status = "PUBLISHED"
        event.published_at = now
        event.attempts += 1
    if events:
        db.commit()


def _task_dispatchable(task: PhotoTask | None, now: datetime) -> bool:
    """photo_task 가 지금 dispatch 가능한 상태인지(재시도 게이트 포함)."""
    if task is None:
        return False
    if task.status != TaskStatus.QUEUED:
        return False
    if task.dead_lettered_at is not None:
        return False
    locked_until = _aware(task.locked_until)
    if locked_until is not None and locked_until >= now:
        return False
    next_retry_at = _aware(task.next_retry_at)
    if next_retry_at is not None and next_retry_at > now:
        return False
    return True


def relay_outbox(
    db: Session,
    *,
    enqueue: EnqueueFn = default_enqueue,
    limit: int = 100,
    now: datetime | None = None,
) -> int:
    """미발행 outbox 이벤트를 enqueue. 게이트 미통과 이벤트는 그대로 남긴다.

    이벤트 게이트: published=False AND (next_retry_at IS NULL OR <= now)
    task 게이트:  _task_dispatchable() (QUEUED, dead-letter 아님, lock 만료, next_retry 도달)
    """
    now = now or _utcnow()
    events = (
        db.query(OutboxEvent)
        .filter(
            OutboxEvent.status.in_(("PENDING", "FAILED")),
            OutboxEvent.published.is_(False),
            (OutboxEvent.next_retry_at.is_(None))
            | (OutboxEvent.next_retry_at <= now),
        )
        .order_by(OutboxEvent.created_at)
        .limit(limit)
        .all()
    )
    # 1) 적격성 판정. 읽기 트랜잭션은 닫는다(enqueue 는 브로커 I/O — DB 락 미보유).
    #    publish 직전 task 의 현재 dispatch_version 을 재확인:
    #    event.dispatch_version != task.dispatch_version → stale → STALE 로 마킹 후 skip.
    eligible = []
    stale_ids = []
    for ev in events:
        task = db.get(PhotoTask, ev.task_id)
        if task is not None and ev.dispatch_version != task.dispatch_version:
            stale_ids.append(ev.id)
            logger.info(
                "stale_dispatch_ignored: event=%s task=%s event_dv=%s task_dv=%s",
                ev.id, ev.task_id, ev.dispatch_version, task.dispatch_version,
            )
            continue
        if _task_dispatchable(task, now):
            eligible.append((ev.id, ev.task_id, ev.attempts, task.dispatch_version))
    for ev_id in stale_ids:
        db.execute(
            update(OutboxEvent)
            .where(OutboxEvent.id == ev_id)
            .values(status="STALE", last_error="stale_dispatch_ignored")
            .execution_options(synchronize_session=False)
        )
    db.commit()

    # 2) enqueue(payload 에 dispatch_version) + 짧은 트랜잭션으로 published 마킹.
    published = 0
    for ev_id, task_id, attempts, dversion in eligible:
        async_result = enqueue(task_id, dversion)
        db.execute(
            update(OutboxEvent)
            .where(OutboxEvent.id == ev_id)
            .values(
                published=True,
                status="PUBLISHED",
                published_at=_utcnow(),
                attempts=attempts + 1,
            )
            .execution_options(synchronize_session=False)
        )
        db.execute(
            update(PhotoTask)
            .where(
                PhotoTask.task_id == task_id,
                PhotoTask.celery_job_id.is_(None),
            )
            .values(celery_job_id=_job_id(async_result))
            .execution_options(synchronize_session=False)
        )
        db.commit()
        published += 1
    return published


def repair_stuck_queued(db: Session, *, limit: int = 100, now: datetime | None = None) -> int:
    """dispatch 가능한데 현재 dispatch_version 의 outbox 가 없는 task 에 이벤트 보충.

    직접 enqueue 하지 않는다. 멱등 helper(ensure_dispatch_event, rearm_published=False)로
    같은 dispatch_version 중복 생성을 막는다. next_retry_at 은 task 값을 승계.
    반환: 새로 만든(=created) outbox 이벤트 수.
    """
    now = now or _utcnow()
    tasks = (
        db.query(PhotoTask)
        .filter(
            PhotoTask.status == TaskStatus.QUEUED,
            PhotoTask.dead_lettered_at.is_(None),
        )
        .limit(limit)
        .all()
    )
    created = 0
    for task in tasks:
        # 게이트: lock 만료 + next_retry 도달
        if not _task_dispatchable(task, now):
            continue
        outcome = task_repo.ensure_dispatch_event(
            db,
            task.task_id,
            task.dispatch_version,
            next_retry_at=task.next_retry_at,
            rearm_published=False,
        )
        if outcome == "created":
            created += 1
    db.commit()
    return created