cleaned worker manager

app/core/worker_manager.py

@@ -2,70 +2,77 @@
 
 import asyncio
 import json
+from typing import Dict
 from app.tasks.analytics_worker import AnalyticsWorker
-from app.redis_client import redis
+from app.redis_client import redis  # Use your existing Upstash client
 import logging
 
 logger = logging.getLogger(__name__)
 
 class WorkerManager:
     def __init__(self):
-        self.active_workers: dict = {}
+        self.active_workers: Dict[str, AnalyticsWorker] = {}
         self.lock_ttl = 300
-        self.stream_key = "stream:analytics_triggers"  # Single stream for all triggers
-        self.last_id = "0"  # Resume point
+        self.stream_key = "stream:analytics_triggers"
+        self.last_id = "0"
+        self.poll_interval = 1.0  # Seconds between polls
     
     async def start_listener(self):
         """🎧 Poll Upstash HTTP API for trigger events"""
-        logger.info("[manager] 🎧 worker manager listening via HTTP polling")
+        logger.info(f"[manager] 🎧 listening via HTTP polling (interval: {self.poll_interval}s)")
         
         while True:
             try:
-                # XREAD without 'block' - Upstash doesn't support it
+                # CRITICAL: Remove 'block' parameter - not supported by Upstash
                 result = redis.xread({self.stream_key: self.last_id}, count=10)
                 
                 if result:
-                    # Upstash returns: {stream: [(msg_id, data), ...]}
+                    # Parse messages
                     messages = result.get(self.stream_key, [])
+                    logger.info(f"[manager] 📥 received {len(messages)} messages")
                     
                     for msg_id, msg_data in messages:
                         try:
-                            # Parse payload from message body (not stream name!)
                             payload = json.loads(msg_data.get("message", "{}"))
                             org_id = payload.get("org_id")
                             source_id = payload.get("source_id")
                             
                             if org_id and source_id:
-                                logger.info(f"[manager] 📥 trigger: {org_id}:{source_id} (msg: {msg_id})")
+                                logger.info(f"[manager] 🚀 processing {org_id}:{source_id}")
                                 await self.spawn_worker(org_id, source_id)
                                 self.last_id = msg_id  # Update resume point
                             else:
-                                logger.warning(f"[manager] ⚠️ invalid payload: {msg_data}")
+                                logger.warning(f"[manager] ⚠️ missing IDs in payload: {payload}")
                         except json.JSONDecodeError as e:
                             logger.error(f"[manager] ❌ JSON error: {e}")
+                    
+                    # Small delay after processing batch
+                    await asyncio.sleep(0.1)
                 else:
-                    await asyncio.sleep(1)  # Polling interval
+                    # CRITICAL: Sleep when no messages
+                    logger.debug("[manager] 💤 no messages, sleeping")
+                    await asyncio.sleep(self.poll_interval)
                     
             except Exception as e:
                 logger.error(f"[manager] ❌ polling error: {e}", exc_info=True)
-                await asyncio.sleep(5)
+                await asyncio.sleep(5)  # Back off on error
     
     async def spawn_worker(self, org_id: str, source_id: str):
-        """Spawn a worker with distributed locking"""
+        """Spawn worker with distributed lock"""
         worker_key = f"{org_id}:{source_id}"
         lock_key = f"worker_lock:{worker_key}"
         
         try:
-            # Check if worker is already running
+            # Check existing lock
             if redis.exists(lock_key):
-                logger.debug(f"[manager] ⏭️ worker already active: {worker_key}")
+                logger.debug(f"[manager] ⏭️ worker locked: {worker_key}")
                 return
             
-            # Set lock with TTL
+            # Set lock
             redis.setex(lock_key, self.lock_ttl, "1")
-            logger.info(f"[manager] 🚀 spawning worker for {worker_key}")
+            logger.info(f"[manager] 🔒 lock acquired: {lock_key}")
             
-            # Create and track worker
+            # Create worker
             worker = AnalyticsWorker(org_id, source_id)
             self.active_workers[worker_key] = worker
             
@@ -77,15 +84,16 @@ class WorkerManager:
             redis.delete(lock_key)
     
     async def _run_worker(self, worker: AnalyticsWorker, worker_key: str, lock_key: str):
-        """Run worker and cleanup"""
+        """Execute worker and cleanup"""
         try:
             results = await worker.run()
-            logger.info(f"[manager] ✅ worker completed: {worker_key}")
+            logger.info(f"[manager] ✅ worker complete: {worker_key}")
         except Exception as e:
             logger.error(f"[manager] ❌ worker failed: {worker_key} - {e}", exc_info=True)
         finally:
             # Cleanup
             self.active_workers.pop(worker_key, None)
             redis.delete(lock_key)
+            logger.info(f"[manager] 🧹 cleaned up: {worker_key}")
 
 worker_manager = WorkerManager()