removed redis hammering

app/main.py

@@ -14,7 +14,7 @@ import pathlib
 import json 
 
 # ─── Third-Party ──────────────────────────────────────────────────────────────
-# from fastapi import FastAPI, Depends, HTTPException, Request, Query, BackgroundTasks
+from fastapi import FastAPI, Depends, HTTPException, Request, Query, BackgroundTasks
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
 from contextlib import asynccontextmanager
@@ -280,12 +280,14 @@ async def continuous_kpi_refresh():
     """
     Auto-refresh KPIs every 5 minutes for active organizations.
     """
+    await asyncio.sleep(10)  # Let app startup complete
+    
     while True:
         try:
             logger.debug("🔄 KPI scheduler tick...")
             
             active_keys = event_hub.keys("entity:*")
-            for key in active_keys:
+            for key in active_keys[:10]:  # Max 10 per batch
                 key_parts = safe_redis_decode(key).split(":")
                 if len(key_parts) >= 3:
                     org_id, source_id = key_parts[1], key_parts[2]
@@ -295,17 +297,19 @@ async def continuous_kpi_refresh():
                     if event_hub.exists(cache_key):
                         continue
 
-                    # Trigger computation (non-blocking)
+                    # Skip if worker already running
+                    if event_hub.exists(f"worker:lock:{org_id}:{source_id}"):
+                        continue
+
+                    # Trigger computation
                     logger.info(f"⏰ Auto-triggering KPIs for {org_id}/{source_id}")
-                    # NOTE: Ensure trigger_kpi_computation is imported/defined
                     await trigger_kpi_computation(org_id, source_id)
-        
+                    await asyncio.sleep(1)  # 1s gap between triggers
+            
         except Exception as e:
             logger.error(f"❌ Scheduler error: {e}")
         
-        await asyncio.sleep(300)  # 5 minutes
-
-
+        await asyncio.sleep(300)  # ⭐ CRITICAL: Sleep 5 minutes between cycles
 @app.get("/debug/stream-content")
 def debug_stream(
     org_id: str = Query(...),

app/mapper.py

@@ -1,5 +1,5 @@
 import os
-# import json
+import json
 import pandas as pd
 import numpy as np
 from datetime import datetime, timedelta
@@ -7,8 +7,8 @@ from concurrent.futures import ThreadPoolExecutor
 import time
 
 from app.db import get_conn, ensure_raw_table, transactional_conn, ensure_schema_versions_table
-# from app.hybrid_entity_detector import hybrid_detect_entity_type
-# from app.core.event_hub import event_hub
+from app.hybrid_entity_detector import hybrid_detect_entity_type
+from app.core.event_hub import event_hub
 
 # ---------------------- Canonical Schema ---------------------- #
 CANONICAL = {

app/tasks/analytics_worker.py

@@ -547,27 +547,40 @@ class WorkerManager:
         self._shutdown = False
     
     async def start_listener(self):
-        """🎧 Listen to Redis pubsub for triggers"""
+        """
+        🎧 TRUE Redis pubsub listener - push-based, efficient, resilient
+        Uses async listen() instead of polling get_message()
+        """
         pubsub = event_hub.redis.pubsub()
         pubsub.psubscribe("analytics_trigger:*")
-        
-        logger.info("🎧 Worker Manager: Einstein+Elon mode ENGAGED")
-        
+
+        logger.info("🎧 Worker Manager: Einstein+Elon mode ENGAGED (true pubsub)")
+
+        # Track last cleanup time to avoid running it every iteration
+        last_cleanup = time.time()
+
         try:
-            while not self._shutdown:
-                # Wait for message with timeout (prevents blocking forever)
-                message = pubsub.get_message(timeout=5.0)
-                
-                if message and message["type"] == "pmessage":
+            # Use async listen() for push-based notifications (Zero Redis ops when idle)
+            async for message in pubsub.listen():
+                if message["type"] == "pmessage":
                     await self._handle_trigger(message)
-                
-                # Cleanup completed tasks
-                self._cleanup_completed_workers()
-                
+
+                # Cleanup every 10 seconds (not every message)
+                if time.time() - last_cleanup > 10:
+                    self._cleanup_completed_workers()
+                    last_cleanup = time.time()
+
         except asyncio.CancelledError:
-            logger.info("[MANAGER] 🛑 Listener cancelled")
+            logger.info("[MANAGER] 🛑 Listener cancelled by shutdown")
+        except Exception as e:
+            logger.error(f"[MANAGER] ❌ Listener crashed: {e}", exc_info=True)
+            # Auto-restart on failure (prevents silent death)
+            logger.info("[MANAGER] 🔄 Attempting to restart listener in 5s...")
+            await asyncio.sleep(5)
+            asyncio.create_task(self.start_listener(), name="worker-manager-restart")
         finally:
             pubsub.close()
+            logger.info("[MANAGER] 🔌 Pubsub connection closed")
     
     async def _handle_trigger(self, message: Dict[str, Any]):
         """Process a single trigger message"""