app/services/worker_manager.py

"""
Worker Manager Service
Consumer process that pulls categories from Redis and executes them one by one.
Provides stability, reliability (RPOPLPUSH), and anti-ban pacing.
"""
import asyncio
import logging
import random
import signal
import gc
from datetime import datetime

from app.services.upstash_cache import get_upstash_cache
from app.services.news_processor import process_category
from app.utils.custom_logger import get_logger
from app.config import CATEGORIES

logger = get_logger(__name__)

class WorkerManager:
    def __init__(self):
        self.running = False
        # NOTE: Do NOT create a new NewsAggregator here.
        # We must use the process-wide singleton from scheduler.py so that
        # circuit breaker failure counts accumulate correctly across tasks.
        # A fresh NewsAggregator() resets all in-memory state and defeats
        # the circuit breaker pattern entirely.
        self._aggregator = None
        self.upstash = get_upstash_cache()
        self.pending_queue = "segmento:pending_news_queue"
        self.processing_queue = "segmento:processing_news_queue"
        self.dlq = "segmento:dead_letter_queue"
        self.visibility_map = "segmento:worker_visibility_tracker"
        self.visibility_timeout = 600  # 10 minutes

    @property
    def aggregator(self):
        """Lazy-load the shared aggregator singleton from scheduler."""
        if self._aggregator is None:
            from app.services.scheduler import _get_shared_aggregator
            self._aggregator = _get_shared_aggregator()
        return self._aggregator
        
    async def start(self):
        """Main consumer loop"""
        self.running = True
        self.polling_wait = 5  # Start with 5s polling wait
        logger.info("👷 [WORKER] Starting consumer loop...")
        
        # Initial jitter to stagger startup across multiple instances if they exist
        await asyncio.sleep(random.uniform(5, 15))
        
        while self.running:
            try:
                # 1. Atomic extraction (Pending -> Processing)
                # This ensures we don't lose the task if the worker crashes mid-run.
                category = await self.upstash.rpoplpush(self.pending_queue, self.processing_queue)
                
                if not category:
                    # 1b. Polite Polling: Exponential Backoff (5s -> 10s -> ... -> 60s)
                    self.polling_wait = min(self.polling_wait * 2, 60)
                    
                    # No new tasks, check for zombies occasionally (5% chance per empty loop)
                    if random.random() < 0.05:
                        await self.cleanup_zombie_tasks()
                    
                    logger.debug("📭 [WORKER] Queue empty. Polling backoff: %ds", self.polling_wait)
                    await asyncio.sleep(self.polling_wait)
                    continue
                
                # Reset polling wait when a task is found
                self.polling_wait = 5
                
                # 2. Track start time for visibility timeout
                start_time = int(datetime.now().timestamp())
                await self.upstash._execute_command(["HSET", self.visibility_map, category, start_time])

                # 3. Process the category
                logger.info("🎯 [WORKER] Processing task from queue: %s", category.upper())
                
                success = False
                try:
                    success = await process_category(category, self.aggregator)
                except Exception as proc_err:
                    logger.error("❌ [WORKER] Task failed: %s. Moving to DLQ.", category)
                    await self.upstash.lpush(self.dlq, f"{category} | {datetime.now().isoformat()} | {str(proc_err)}")
                finally:
                    # 4. Cleanup: Task ALWAYS removed from processing queue and visibility map
                    await self.upstash.lrem(self.processing_queue, 1, category)
                    await self.upstash._execute_command(["HDEL", self.visibility_map, category])
                    if success:
                        logger.info("✅ [WORKER] Task completed and cleaned: %s", category.upper())
                    else:
                        logger.warning("⚠️ [WORKER] Task cleaned from queue after failure: %s", category.upper())

                # 5. Mandatory inter-task spacing (HF-safe pacing)
                # Production logs show zero 429s from any free provider (Google RSS,
                # Medium, HackerNews all return 200). Inshorts fails with connection
                # reset (geo-block), not rate limit. No banning risk detected.
                #
                # Previous 30-90s was causing a 40+ minute full cycle time.
                # 8-18s provides enough jitter to avoid burst patterns while
                # keeping a full 22-category cycle under ~7 minutes.
                wait_time = random.uniform(8, 18)
                logger.info("💤 [WORKER] Pacing: sleeping for %.1fs...", wait_time)
                await asyncio.sleep(wait_time)
                
            except Exception as e:
                logger.error("❌ [WORKER] Fatal error in consumer loop: %s", e)
                # Should we move to DLQ? In Task 5 we will implement the Reaper.
                await asyncio.sleep(30) # Backoff on error
            finally:
                # Explicit garbage collection per worker cycle (OOM Fix)
                gc.collect()

    def stop(self, *args):
        logger.info("👷 [WORKER] Stopping gracefully...")
        self.running = False

    async def cleanup_zombie_tasks(self):
        """
        The Reaper: Scans the processing queue for tasks that timed out.
        If a worker died while processing, this moves the task back to pending.
        """
        try:
            # 1. Get all tasks in processing queue
            processing_tasks = await self.upstash._execute_command(["LRANGE", self.processing_queue, 0, -1])
            if not processing_tasks:
                return

            # 2. Get start times from visibility map
            # Note: we use individual HGETs or HGETALL if safe
            current_time = int(datetime.now().timestamp())
            
            for category in processing_tasks:
                start_time_str = await self.upstash._execute_command(["HGET", self.visibility_map, category])
                if not start_time_str:
                    # Edge case: Task in processing but no timestamp? Move back to pending.
                    logger.warning("👻 [REAPER] Ghost task found for %s. Re-queueing.", category)
                    await self.upstash.lrem(self.processing_queue, 1, category)
                    await self.upstash.lpush(self.pending_queue, category)
                    continue
                
                start_time = int(start_time_str)
                if (current_time - start_time) > self.visibility_timeout:
                    # Task timed out!
                    logger.warning("🧟 [REAPER] Zombie task detected for %s (timed out). Recovering...", category)
                    await self.upstash.lrem(self.processing_queue, 1, category)
                    await self.upstash._execute_command(["HDEL", self.visibility_map, category])
                    await self.upstash.lpush(self.pending_queue, category)

        except Exception as reaper_err:
            logger.error("❌ [REAPER] Error: %s", reaper_err)

async def run_worker():
    """Main entry point for the worker process"""
    worker = WorkerManager()
    await worker.start()

if __name__ == "__main__":
    asyncio.run(run_worker())