coreected worker manager

app/core/worker_manager.py

@@ -1,31 +1,44 @@
-# app/core/worker_manager.py – UPSTASH-FREE-TIER-COMPATIBLE
-import os
+# app/core/worker_manager.py – UPSTASH-COMPATIBLE v4.1
+
 import asyncio
 import json
+import os
 import time
-from typing import Dict
+from typing import Dict, List, Optional, Any
 import logging
-
+import  datetime
 from app.core.event_hub import event_hub
-from app.tasks.analytics_worker import trigger_kpi_computation
+from app.tasks.analytics_worker import AnalyticsWorker
 
 logger = logging.getLogger(__name__)
 
 
+def _safe_redis_decode(value: Any) -> str:
+    """Safely decode Redis values that might be bytes or str"""
+    if isinstance(value, bytes):
+        return value.decode('utf-8')
+    return str(value)
+
+
 class WorkerManager:
+    """
+    🎛️ Manages worker lifecycle and prevents Redis hammering
+    Uses ONLY Upstash-safe HTTP commands: GET, SET, EXISTS, DEL, XREVRANGE
+    """
+    
     def __init__(self):
         self.active_workers: Dict[str, asyncio.Task] = {}
         self._shutdown = False
         
         # ⚡ ADAPTIVE POLLING (configurable via env vars)
-        self.active_interval = float(os.getenv("WORKER_POLL_ACTIVE", "1.0"))  # 1s when busy
-        self.idle_interval = float(os.getenv("WORKER_POLL_IDLE", "30.0"))     # 30s when idle
+        self.active_interval = float(os.getenv("WORKER_POLL_ACTIVE", "1.0"))
+        self.idle_interval = float(os.getenv("WORKER_POLL_IDLE", "30.0"))
         self.consecutive_empty = 0
     
     async def start_listener(self):
         """
         🎧 UPSTASH-SAFE: No pubsub, no blocking xread, just smart async polling
-        Redis ops: ~0.03/sec idle, ~2/sec under load (well within free tier)
+        Redis ops: ~0.03/sec idle, ~2/sec under load
         """
         logger.info(
             f"🎧 Worker Manager: Einstein+Elon mode ENGAGED "
@@ -47,7 +60,7 @@ class WorkerManager:
                 
                 # Log state changes
                 if self.consecutive_empty == 5:
-                    logger.info(f"[MANAGER] 🛌 Idle mode activated (poll: {interval}s)")
+                    logger.info(f"[MANAGER] 🛌 Idle mode activated (poll: {interval:.1f}s)")
                 
                 await asyncio.sleep(interval)
                 
@@ -56,25 +69,43 @@ class WorkerManager:
                 break
             except Exception as e:
                 logger.error(f"[MANAGER] ❌ Error: {e}", exc_info=True)
-                await asyncio.sleep(5)  # Back off on errors
+                await asyncio.sleep(5)
     
-    async def _fetch_pending_triggers(self) -> list:
+    async def _fetch_pending_triggers(self) -> List[tuple]:
         """
         Fetch pending triggers in a SINGLE Redis call
         Uses xrevrange to get newest messages without blocking
+        Returns: [(msg_id, {field: value}), ...]
         """
         try:
-            # Get last 10 messages from stream (non-blocking, minimal ops)
+            # Get last 10 messages from stream (non-blocking)
             result = event_hub.redis.xrevrange(
                 "stream:analytics_triggers",
                 count=10
             )
             
-            # If using Upstash HTTP client, result might be dict
+            # Handle different response formats from Upstash
+            messages = []
             if isinstance(result, dict):
-                messages = list(result.items()) if result else []
-            else:
-                messages = result or []
+                # Format: {msg_id: {field: value}, ...}
+                for msg_id, data in result.items():
+                    messages.append((msg_id, data))
+            elif isinstance(result, list):
+                # Format: [(msg_id, [field, value, field, value]), ...]
+                for item in result:
+                    if isinstance(item, (list, tuple)) and len(item) == 2:
+                        msg_id, data = item
+                        # Convert flat list to dict if needed
+                        if isinstance(data, list):
+                            data_dict = {}
+                            for i in range(0, len(data), 2):
+                                if i + 1 < len(data):
+                                    key = _safe_redis_decode(data[i])
+                                    value = _safe_redis_decode(data[i + 1])
+                                    data_dict[key] = value
+                            messages.append((msg_id, data_dict))
+                        else:
+                            messages.append((msg_id, data))
             
             return messages
             
@@ -82,13 +113,30 @@ class WorkerManager:
             logger.debug(f"[MANAGER] Fetch failed: {e}")
             return []
     
-    async def _process_batch(self, messages: list):
+    async def _process_batch(self, messages: List[tuple]):
         """Process multiple triggers efficiently"""
         logger.info(f"[MANAGER] 📥 Processing {len(messages)} triggers")
         
         for msg_id, msg_data in messages:
             try:
-                payload = json.loads(msg_data.get("message", "{}"))
+                # Handle different data formats
+                if isinstance(msg_data, dict):
+                    # Already a dict
+                    message_str = msg_data.get("message", "{}")
+                elif isinstance(msg_data, list):
+                    # Flat list: [field, value, field, value]
+                    message_str = "{}"
+                    for i in range(0, len(msg_data), 2):
+                        if i + 1 < len(msg_data):
+                            key = _safe_redis_decode(msg_data[i])
+                            if key == "message":
+                                message_str = _safe_redis_decode(msg_data[i + 1])
+                                break
+                else:
+                    logger.warning(f"[MANAGER] Unknown msg_data format: {type(msg_data)}")
+                    continue
+                
+                payload = json.loads(message_str)
                 await self._handle_trigger(payload)
                 
                 # Acknowledge: delete processed message
@@ -124,7 +172,8 @@ class WorkerManager:
     async def _run_worker(self, worker_id: str, org_id: str, source_id: str):
         """Execute KPI computation with automatic cleanup"""
         try:
-            await trigger_kpi_computation(org_id, source_id)
+            worker = AnalyticsWorker(org_id, source_id)
+            await worker.run()
             logger.info(f"[MANAGER] ✅ Complete: {worker_id}")
         except Exception as e:
             logger.error(f"[MANAGER] ❌ Failed: {worker_id} - {e}", exc_info=True)
@@ -146,5 +195,80 @@ class WorkerManager:
         logger.info("[MANAGER] 🛑 Shutdown initiated")
 
 
-# Global instance
-worker_manager = WorkerManager()
+# ==================== FASTAPI INTEGRATION ====================
+
+# Global manager instance
+_worker_manager: Optional[WorkerManager] = None
+
+
+async def get_worker_manager() -> WorkerManager:
+    """Get or create worker manager singleton"""
+    global _worker_manager
+    if _worker_manager is None:
+        _worker_manager = WorkerManager()
+    return _worker_manager
+
+
+async def trigger_kpi_computation(org_id: str, source_id: str):
+    """
+    🎯 FastAPI endpoint handler - triggers worker via Redis stream
+    Idempotent: multiple calls won't spawn duplicate workers
+    """
+    try:
+        # Write to stream (HTTP-safe)
+        event_hub.redis.xadd(
+            "stream:analytics_triggers",
+            {
+                "message": json.dumps({
+                    "org_id": org_id,
+                    "source_id": source_id,
+                    "type": "kpi_compute",
+                    "timestamp": datetime.now().isoformat()
+                })
+            }
+        )
+        logger.info(f"🎯 Triggered KPI computation: {org_id}/{source_id}")
+        return {"status": "triggered", "org_id": org_id, "source_id": source_id}
+        
+    except Exception as e:
+        logger.error(f"Trigger failed: {e}", exc_info=True)
+        return {"status": "error", "message": str(e)}
+
+
+# ==================== BACKGROUND REFRESH (Optional) ====================
+
+async def continuous_kpi_refresh(manager: WorkerManager):
+    """
+    🎛️ Gentle background refresh - runs every 5 minutes
+    Only triggers for stale data (no active worker, no fresh cache)
+    """
+    await asyncio.sleep(10)  # Let app startup complete
+    
+    while True:
+        try:
+            # Get all entity keys (HTTP-safe)
+            entity_keys = event_hub.redis.keys("entity:*:*")
+            
+            for key in entity_keys[:10]:  # Max 10 per cycle
+                key_str = key.decode() if isinstance(key, bytes) else key
+                _, org_id, source_id = key_str.split(":")
+                
+                worker_id = f"{org_id}:{source_id}"
+                
+                # Skip if worker already running
+                if worker_id in manager.active_workers:
+                    continue
+                
+                # Skip if KPIs are fresh (< 5 min old)
+                cache_key = f"kpi_cache:{org_id}:{source_id}"
+                if event_hub.redis.exists(cache_key):
+                    continue
+                
+                # Trigger refresh
+                await trigger_kpi_computation(org_id, source_id)
+                await asyncio.sleep(1)  # 1s gap
+            
+        except Exception as e:
+            logger.error(f"[AUTO] Error: {e}", exc_info=True)
+        
+        await asyncio.sleep(300)  # ⭐ Sleep 5 minutes