""" 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