workers/queue_processor.py

"""

Queue Processor — 메시지 큐 비동기 워커



PostgreSQL 테이블 기반 메시지 큐를 주기적으로 polling하여

배지 확인, 챌린지 진행도, 데이터 정리 등을 비동기 처리.



HuggingFace Spaces의 FastAPI 서버에서 백그라운드 태스크로 실행됨.



@changelog

- v1.1.0 (2026-03-07): 적응형 폴링 (adaptive backoff)

  - 큐 비어있으면 간격 2배씩 증가 (5s → 10s → ... → 300s max)

  - 메시지 발견 시 즉시 최소 간격으로 리셋

  - 유휴 시 시간당 ~36건 (기존 ~2,160건 대비 98% 감소)

- v1.0.0 (2026-03-07): 초기 구현

  - badge_check 큐: 배지 자격 확인 + 토큰 보상 지급

  - challenge_check 큐: 챌린지 진행도 업데이트

  - data_cleanup 큐: 만료 토큰/캐시/디바이스 데이터 정리

  - FOR UPDATE SKIP LOCKED 기반 동시 소비자 안전

  - 재시도 (max 3) + dead letter 처리

"""
import asyncio
import logging
import os
import time
from typing import Any, Callable, Coroutine, Dict, List, Optional

logger = logging.getLogger(__name__)

# 큐 설정
POLL_INTERVAL_MIN = int(os.getenv("MQ_POLL_INTERVAL_MIN", "5"))
POLL_INTERVAL_MAX = int(os.getenv("MQ_POLL_INTERVAL_MAX", "300"))
BATCH_SIZE = int(os.getenv("MQ_BATCH_SIZE", "10"))
VISIBILITY_TIMEOUT_SECONDS = int(os.getenv("MQ_VISIBILITY_TIMEOUT", "60"))
CLEANUP_INTERVAL_HOURS = int(os.getenv("MQ_CLEANUP_INTERVAL_HOURS", "6"))

# 싱글톤
_processor: Optional["QueueProcessor"] = None


def get_queue_processor() -> "QueueProcessor":
    """QueueProcessor 싱글톤 반환"""
    global _processor
    if _processor is None:
        _processor = QueueProcessor()
    return _processor


class QueueProcessor:
    """

    메시지 큐 워커 프로세서.



    FastAPI lifespan에서 start()/stop()을 호출하여

    백그라운드 태스크로 실행.



    각 큐별 핸들러를 등록하고, 주기적으로 polling하여

    메시지를 처리합니다.

    """

    def __init__(self):
        self._running = False
        self._task: Optional[asyncio.Task] = None
        self._cleanup_task: Optional[asyncio.Task] = None
        self._handlers: Dict[str, Callable] = {}
        self._current_interval = POLL_INTERVAL_MIN
        self._stats = {
            "total_processed": 0,
            "total_errors": 0,
            "total_dead": 0,
            "queues": {},
        }

        # 기본 핸들러 등록
        self.register_handler("badge_check", self._handle_badge_check)
        self.register_handler("challenge_check", self._handle_challenge_check)
        self.register_handler("data_cleanup", self._handle_data_cleanup)

    def register_handler(

        self,

        queue_name: str,

        handler: Callable[

            [Dict[str, Any]], Coroutine[Any, Any, None]

        ],

    ):
        """큐별 핸들러 등록"""
        self._handlers[queue_name] = handler
        self._stats["queues"][queue_name] = {
            "processed": 0,
            "errors": 0,
            "last_processed_at": None,
        }
        logger.info(f"[mq] Registered handler for queue: {queue_name}")

    def start(self):
        """백그라운드 워커 시작"""
        if self._running:
            logger.warning("[mq] Queue processor already running")
            return

        self._running = True
        self._current_interval = POLL_INTERVAL_MIN
        self._task = asyncio.create_task(self._poll_loop())
        self._cleanup_task = asyncio.create_task(self._cleanup_loop())
        logger.info(
            f"[mq] Queue processor started "
            f"(poll={POLL_INTERVAL_MIN}~{POLL_INTERVAL_MAX}s adaptive, "
            f"batch={BATCH_SIZE}, timeout={VISIBILITY_TIMEOUT_SECONDS}s)"
        )

    async def stop(self):
        """백그라운드 워커 중지 (graceful)"""
        if not self._running:
            return

        self._running = False
        logger.info("[mq] Queue processor stopping...")

        for name, task in [("poll", self._task), ("cleanup", self._cleanup_task)]:
            if task is not None and not task.done():
                task.cancel()
                try:
                    await task
                except asyncio.CancelledError:
                    pass
                logger.info(f"[mq] {name} task stopped")

        self._task = None
        self._cleanup_task = None
        logger.info("[mq] Queue processor stopped")

    def get_stats(self) -> Dict[str, Any]:
        """워커 통계 반환"""
        return {
            "running": self._running,
            "poll_interval_seconds": self._current_interval,
            "poll_interval_range": f"{POLL_INTERVAL_MIN}~{POLL_INTERVAL_MAX}s",
            "batch_size": BATCH_SIZE,
            "registered_queues": list(self._handlers.keys()),
            **self._stats,
        }

    # ==========================================
    # 메인 루프
    # ==========================================

    async def _poll_loop(self):
        """적응형 폴링: 메시지 있으면 빠르게, 없으면 간격을 점진적으로 늘림"""
        # 서버 시작 후 안정화 대기
        await asyncio.sleep(10)

        while self._running:
            found_any = False
            try:
                for queue_name in self._handlers:
                    if not self._running:
                        break
                    count = await self._process_queue(queue_name)
                    if count > 0:
                        found_any = True
            except asyncio.CancelledError:
                raise
            except Exception as e:
                logger.error(
                    f"[mq] Poll loop error (will continue): "
                    f"{type(e).__name__}: {str(e)[:200]}"
                )

            # 적응형 백오프: 메시지 발견 시 최소 간격으로 리셋, 없으면 2배씩 증가
            if found_any:
                self._current_interval = POLL_INTERVAL_MIN
            else:
                self._current_interval = min(
                    self._current_interval * 2, POLL_INTERVAL_MAX
                )

            await asyncio.sleep(self._current_interval)

    async def _cleanup_loop(self):
        """주기적으로 완료/dead 메시지 정리"""
        # 첫 정리는 1시간 후
        await asyncio.sleep(3600)

        while self._running:
            try:
                await self._run_mq_cleanup()
            except asyncio.CancelledError:
                raise
            except Exception as e:
                logger.error(
                    f"[mq] Cleanup loop error (will continue): "
                    f"{type(e).__name__}: {str(e)[:200]}"
                )

            await asyncio.sleep(CLEANUP_INTERVAL_HOURS * 3600)

    # ==========================================
    # 큐 처리 핵심 로직
    # ==========================================

    async def _process_queue(self, queue_name: str) -> int:
        """특정 큐의 메시지를 배치로 수신하고 처리. 반환: 처리된 메시지 수"""
        from db import get_supabase

        supabase = get_supabase()
        handler = self._handlers.get(queue_name)
        if not handler:
            return 0

        try:
            # mq_receive RPC 호출 — FOR UPDATE SKIP LOCKED
            result = supabase.rpc(
                "mq_receive",
                {
                    "p_queue_name": queue_name,
                    "p_batch_size": BATCH_SIZE,
                    "p_visibility_timeout_seconds": VISIBILITY_TIMEOUT_SECONDS,
                },
            ).execute()

            messages = result.data or []
            if not messages:
                return 0

            logger.debug(
                f"[mq:{queue_name}] Received {len(messages)} messages"
            )

            completed_ids: List[int] = []
            for msg in messages:
                msg_id = msg["msg_id"]
                msg_body = msg["msg_message"]

                try:
                    await handler(msg_body)
                    completed_ids.append(msg_id)
                    self._stats["total_processed"] += 1
                    self._stats["queues"][queue_name]["processed"] += 1
                except Exception as e:
                    # 개별 메시지 실패 — mq_fail로 재시도/dead 처리
                    self._stats["total_errors"] += 1
                    self._stats["queues"][queue_name]["errors"] += 1
                    error_msg = f"{type(e).__name__}: {str(e)[:500]}"
                    logger.warning(
                        f"[mq:{queue_name}] Message {msg_id} failed: {error_msg}"
                    )

                    try:
                        fail_result = supabase.rpc(
                            "mq_fail",
                            {
                                "p_message_id": msg_id,
                                "p_error_msg": error_msg,
                            },
                        ).execute()
                        new_status = fail_result.data
                        if new_status == "dead":
                            self._stats["total_dead"] += 1
                            logger.error(
                                f"[mq:{queue_name}] Message {msg_id} moved to dead letter"
                            )
                    except Exception as fail_err:
                        logger.error(
                            f"[mq:{queue_name}] mq_fail call failed for {msg_id}: {fail_err}"
                        )

            # 성공한 메시지들을 일괄 완료 처리
            if completed_ids:
                try:
                    supabase.rpc(
                        "mq_complete",
                        {"p_message_ids": completed_ids},
                    ).execute()
                    self._stats["queues"][queue_name]["last_processed_at"] = (
                        time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
                    )
                    logger.info(
                        f"[mq:{queue_name}] Completed {len(completed_ids)} messages"
                    )
                except Exception as e:
                    logger.error(
                        f"[mq:{queue_name}] mq_complete failed: {e}"
                    )

            return len(messages)

        except Exception as e:
            # mq_receive 자체 실패 (DB 연결 등)
            logger.error(
                f"[mq:{queue_name}] Failed to receive messages: "
                f"{type(e).__name__}: {str(e)[:200]}"
            )
            return 0

    async def _run_mq_cleanup(self):
        """완료/dead 메시지 정리 RPC 호출"""
        from db import get_supabase

        supabase = get_supabase()
        try:
            result = supabase.rpc(
                "mq_cleanup", {"p_max_age_days": 7}
            ).execute()
            cleanup_data = result.data
            logger.info(f"[mq] Cleanup completed: {cleanup_data}")
        except Exception as e:
            logger.error(f"[mq] Cleanup failed: {e}")

    # ==========================================
    # 큐 핸들러: badge_check
    # ==========================================

    async def _handle_badge_check(self, message: Dict[str, Any]):
        """

        배지 자격 확인 — check_and_grant_badges의 비동기 버전



        기존 동기 트리거가 매 INSERT마다 N개 배지를 순회하며

        COUNT 쿼리를 실행하던 것을, 큐 워커에서 처리.



        최적화:

        - 이미 보유한 배지 제외

        - 활동 유형에 해당하는 배지만 확인

        - user_activities 대신 가능하면 집계 캐시 활용

        """
        from db import get_supabase

        user_id = message.get("user_id")
        activity_type = message.get("activity_type")

        if not user_id or not activity_type:
            logger.warning(f"[mq:badge_check] Invalid message: {message}")
            return

        supabase = get_supabase()

        # activity_type → condition_type 매핑
        activity_to_condition = {
            "story_unlocked": "stories",
            "course_viewed": "visits",
            "course_generated": "courses",
            "spot_visited": "spots",
            "spot_liked": "spots",
        }
        condition_type = activity_to_condition.get(activity_type)
        if not condition_type:
            return  # 매핑되지 않는 활동은 무시

        try:
            # UUID 형식 확인
            import uuid
            user_uuid = str(uuid.UUID(user_id))
        except (ValueError, AttributeError):
            return

        # 1. 해당 condition_type의 활성 배지 중 미보유 배지만 조회
        badges_result = supabase.table("badge_definitions").select("*").eq(
            "condition_type", condition_type
        ).eq("is_active", True).execute()

        if not badges_result.data:
            return

        # 이미 보유한 배지 조회
        owned_result = supabase.table("user_badges").select("badge_type").eq(
            "user_id", user_uuid
        ).execute()
        owned_types = {b["badge_type"] for b in (owned_result.data or [])}

        # 미보유 배지만 필터
        candidates = [
            b for b in badges_result.data
            if b["badge_type"] not in owned_types
        ]
        if not candidates:
            return

        # 2. 활동 카운트 한 번만 조회 (condition_type별)
        count = await self._get_activity_count(
            supabase, user_id, condition_type
        )

        # 3. 조건 충족 배지 부여
        for badge in candidates:
            if count >= badge["condition_value"]:
                await self._grant_badge(supabase, user_uuid, badge)

    async def _get_activity_count(

        self, supabase, user_id: str, condition_type: str

    ) -> int:
        """활동 유형별 카운트 조회"""
        activity_type_map = {
            "stories": ("story_unlocked", False),
            "visits": ("course_viewed", False),
            "courses": ("course_generated", False),
            "spots": ("spot_visited", True),  # DISTINCT spot_id
        }

        act_type, use_distinct = activity_type_map.get(
            condition_type, (None, False)
        )
        if not act_type:
            return 0

        if use_distinct:
            # DISTINCT spot_id 카운트는 직접 쿼리로 해야 함
            # PostgREST에서 DISTINCT COUNT가 어려우므로 전체 조회 후 카운트
            result = supabase.table("user_activities").select(
                "details"
            ).eq(
                "user_id", user_id
            ).eq(
                "activity_type", act_type
            ).execute()

            if not result.data:
                return 0

            spot_ids = set()
            for row in result.data:
                details = row.get("details") or {}
                spot_id = details.get("spot_id")
                if spot_id:
                    spot_ids.add(spot_id)
            return len(spot_ids)
        else:
            # 단순 COUNT
            result = supabase.table("user_activities").select(
                "id", count="exact"
            ).eq(
                "user_id", user_id
            ).eq(
                "activity_type", act_type
            ).execute()
            return result.count or 0

    async def _grant_badge(

        self, supabase, user_uuid: str, badge: Dict[str, Any]

    ):
        """배지 부여 + 보상 토큰 지급"""
        try:
            # 배지 부여 (ON CONFLICT DO NOTHING으로 중복 방지)
            supabase.table("user_badges").upsert(
                {
                    "user_id": user_uuid,
                    "badge_type": badge["badge_type"],
                    "badge_name": badge["badge_name"],
                    "badge_description": badge.get("badge_description", ""),
                    "badge_icon": badge.get("badge_icon", ""),
                    "earned_at": "now()",
                },
                on_conflict="user_id,badge_type",
                # DO NOTHING과 동일 효과: 기존 행 유지
                ignore_duplicates=True,
            ).execute()

            logger.info(
                f"[mq:badge_check] Granted badge '{badge['badge_type']}' "
                f"to user {user_uuid}"
            )

            # 보상 토큰 지급
            reward = badge.get("reward_tokens", 0)
            if reward > 0:
                await self._grant_badge_reward(supabase, user_uuid, badge, reward)

        except Exception as e:
            logger.error(
                f"[mq:badge_check] Failed to grant badge "
                f"'{badge['badge_type']}' to {user_uuid}: {e}"
            )
            raise

    async def _grant_badge_reward(

        self,

        supabase,

        user_uuid: str,

        badge: Dict[str, Any],

        reward: int,

    ):
        """

        배지 보상 토큰 지급.



        기존 SQL 트리거와 동일한 로직:

        1. user_tokens에 balance += reward (없으면 INSERT)

        2. token_transactions에 거래 내역 기록

        """
        try:
            # 1. 현재 잔액 조회
            existing = supabase.table("user_tokens").select(
                "balance"
            ).eq("user_id", user_uuid).maybe_single().execute()

            if existing.data:
                # 기존 사용자: balance += reward
                old_balance = existing.data.get("balance", 0)
                new_balance = old_balance + reward
                supabase.table("user_tokens").update(
                    {
                        "balance": new_balance,
                        "updated_at": "now()",
                    }
                ).eq("user_id", user_uuid).execute()
            else:
                # 신규 사용자: INSERT
                new_balance = reward
                supabase.table("user_tokens").insert(
                    {
                        "user_id": user_uuid,
                        "balance": new_balance,
                        "updated_at": "now()",
                    }
                ).execute()

            # 2. 토큰 거래 내역 기록
            supabase.table("token_transactions").insert(
                {
                    "user_id": user_uuid,
                    "type": "bonus",
                    "amount": reward,
                    "balance_after": new_balance,
                    "description": f"배지 보상: {badge['badge_name']}",
                }
            ).execute()

            logger.info(
                f"[mq:badge_check] Granted {reward} tokens for badge "
                f"'{badge['badge_type']}' to {user_uuid} "
                f"(new_balance={new_balance})"
            )

        except Exception as e:
            # 토큰 지급 실패해도 배지 부여는 유지 (기존 동작과 동일)
            logger.warning(
                f"[mq:badge_check] Token grant failed for "
                f"{user_uuid}/{badge['badge_type']}: {e}"
            )

    # ==========================================
    # 큐 핸들러: challenge_check
    # ==========================================

    async def _handle_challenge_check(self, message: Dict[str, Any]):
        """

        챌린지 진행도 비동기 업데이트 — update_challenge_progress의 비동기 버전

        """
        from db import get_supabase

        user_id = message.get("user_id")
        activity_type = message.get("activity_type")

        if not user_id or not activity_type:
            return

        # activity_type → challenge target_type 매핑
        target_type_map = {
            "course_completed": "courses_completed",
            "spot_visited": "spots_visited",
            "story_unlocked": "stories_unlocked",
            "review_written": "reviews_written",
            "course_shared": "courses_shared",
        }
        target_type = target_type_map.get(activity_type)
        if not target_type:
            return

        try:
            import uuid
            user_uuid = str(uuid.UUID(user_id))
        except (ValueError, AttributeError):
            return

        supabase = get_supabase()

        # 해당 target_type의 활성 챌린지 조회
        challenges_result = supabase.table("challenges").select("*").eq(
            "target_type", target_type
        ).eq("is_active", True).execute()

        if not challenges_result.data:
            return

        from datetime import date

        today = date.today().isoformat()

        for challenge in challenges_result.data:
            # 기간 확인
            start_date = challenge.get("start_date")
            end_date = challenge.get("end_date")
            if start_date and start_date > today:
                continue
            if end_date and end_date < today:
                continue

            challenge_id = challenge["id"]

            # user_challenges UPSERT (current_value +1)
            try:
                # 기존 진행 조회
                existing = supabase.table("user_challenges").select("*").eq(
                    "user_id", user_uuid
                ).eq("challenge_id", challenge_id).maybe_single().execute()

                if existing.data:
                    current = existing.data
                    if current.get("is_completed"):
                        continue  # 이미 완료된 챌린지 건너뜀

                    new_value = current["current_value"] + 1
                    update_data = {
                        "current_value": new_value,
                        "updated_at": "now()",
                    }

                    # 완료 체크
                    if new_value >= challenge["target_value"]:
                        update_data["is_completed"] = True
                        update_data["completed_at"] = "now()"

                    supabase.table("user_challenges").update(
                        update_data
                    ).eq("user_id", user_uuid).eq(
                        "challenge_id", challenge_id
                    ).execute()
                else:
                    # 신규 생성
                    new_value = 1
                    insert_data = {
                        "user_id": user_uuid,
                        "challenge_id": challenge_id,
                        "current_value": new_value,
                    }
                    if new_value >= challenge["target_value"]:
                        insert_data["is_completed"] = True
                        insert_data["completed_at"] = "now()"

                    supabase.table("user_challenges").insert(
                        insert_data
                    ).execute()

                logger.debug(
                    f"[mq:challenge_check] Updated challenge "
                    f"{challenge_id} for {user_uuid}: value={new_value}"
                )

            except Exception as e:
                logger.warning(
                    f"[mq:challenge_check] Failed to update challenge "
                    f"{challenge_id} for {user_uuid}: {e}"
                )

    # ==========================================
    # 큐 핸들러: data_cleanup
    # ==========================================

    async def _handle_data_cleanup(self, message: Dict[str, Any]):
        """

        데이터 정리 작업 — cleanup_type별 분기



        cleanup_type:

        - "device_data": 오래된 디바이스 데이터 정리

        - "expired_tokens": 만료 토큰 처리

        - "expired_cache": 캐시 정리

        - "expired_rate_limits": 만료 rate limit 정리

        - "mq_cleanup": 메시지 큐 자체 정리

        """
        from db import get_supabase

        cleanup_type = message.get("cleanup_type")
        if not cleanup_type:
            logger.warning(f"[mq:data_cleanup] Missing cleanup_type: {message}")
            return

        supabase = get_supabase()

        try:
            if cleanup_type == "device_data":
                days = message.get("days_old", 90)
                result = supabase.rpc(
                    "cleanup_old_device_data", {"p_days_old": days}
                ).execute()
                logger.info(
                    f"[mq:data_cleanup] device_data cleanup: {result.data}"
                )

            elif cleanup_type == "expired_tokens":
                result = supabase.rpc("expire_stale_tokens").execute()
                logger.info(
                    f"[mq:data_cleanup] expired_tokens: {result.data}"
                )

            elif cleanup_type == "expired_cache":
                result = supabase.rpc("cleanup_expired_cache").execute()
                logger.info(
                    f"[mq:data_cleanup] expired_cache: {result.data}"
                )

            elif cleanup_type == "expired_rate_limits":
                result = supabase.rpc(
                    "cleanup_expired_rate_limits"
                ).execute()
                logger.info(
                    f"[mq:data_cleanup] expired_rate_limits: {result.data}"
                )

            elif cleanup_type == "mq_cleanup":
                max_age = message.get("max_age_days", 7)
                result = supabase.rpc(
                    "mq_cleanup", {"p_max_age_days": max_age}
                ).execute()
                logger.info(f"[mq:data_cleanup] mq_cleanup: {result.data}")

            else:
                logger.warning(
                    f"[mq:data_cleanup] Unknown cleanup_type: {cleanup_type}"
                )

        except Exception as e:
            logger.error(
                f"[mq:data_cleanup] {cleanup_type} failed: "
                f"{type(e).__name__}: {str(e)[:200]}"
            )
            raise


class CleanupScheduler:
    """

    정기 정리 작업 스케줄러.



    cleanup 큐에 주기적으로 정리 메시지를 전송하여

    데이터 정리 작업을 트리거합니다.



    GitHub Actions cron 대체 (B-M6/B-M7 해결).

    """

    # 스케줄 정의: (cleanup_type, interval_hours, extra_params)
    SCHEDULES = [
        ("device_data", 24, {"days_old": 90}),
        ("expired_tokens", 12, {}),
        ("expired_cache", 6, {}),
        ("expired_rate_limits", 6, {}),
        ("mq_cleanup", 24, {"max_age_days": 7}),
    ]

    def __init__(self):
        self._task: Optional[asyncio.Task] = None
        self._running = False

    def start(self):
        """스케줄러 시작"""
        if self._running:
            return
        self._running = True
        self._task = asyncio.create_task(self._schedule_loop())
        logger.info("[mq:scheduler] Cleanup scheduler started")

    async def stop(self):
        """스케줄러 중지"""
        self._running = False
        if self._task and not self._task.done():
            self._task.cancel()
            try:
                await self._task
            except asyncio.CancelledError:
                pass
        self._task = None
        logger.info("[mq:scheduler] Cleanup scheduler stopped")

    async def _schedule_loop(self):
        """스케줄에 따라 cleanup 메시지 전송"""
        # 서버 시작 후 5분 대기 (안정화)
        await asyncio.sleep(300)

        # 각 스케줄별 마지막 실행 시간 추적
        last_run: Dict[str, float] = {}

        while self._running:
            try:
                now = time.time()
                for cleanup_type, interval_hours, params in self.SCHEDULES:
                    last = last_run.get(cleanup_type, 0)
                    if now - last >= interval_hours * 3600:
                        await self._enqueue_cleanup(cleanup_type, params)
                        last_run[cleanup_type] = now
            except asyncio.CancelledError:
                raise
            except Exception as e:
                logger.error(
                    f"[mq:scheduler] Schedule loop error: "
                    f"{type(e).__name__}: {str(e)[:200]}"
                )

            # 1시간마다 확인
            await asyncio.sleep(3600)

    async def _enqueue_cleanup(

        self, cleanup_type: str, params: Dict[str, Any]

    ):
        """cleanup 큐에 메시지 전송"""
        from db import get_supabase

        supabase = get_supabase()
        message = {"cleanup_type": cleanup_type, **params}

        try:
            supabase.rpc(
                "mq_send",
                {
                    "p_queue_name": "data_cleanup",
                    "p_message": message,
                    "p_priority": 0,
                    "p_max_retries": 2,
                },
            ).execute()
            logger.info(
                f"[mq:scheduler] Enqueued cleanup: {cleanup_type}"
            )
        except Exception as e:
            logger.error(
                f"[mq:scheduler] Failed to enqueue {cleanup_type}: {e}"
            )


# 스케줄러 싱글톤
_scheduler: Optional[CleanupScheduler] = None


def get_cleanup_scheduler() -> CleanupScheduler:
    """CleanupScheduler 싱글톤 반환"""
    global _scheduler
    if _scheduler is None:
        _scheduler = CleanupScheduler()
    return _scheduler