fixed analytics event emmiter

app/core/event_hub.py

@@ -71,25 +71,27 @@ class EventHub:
         channel = f"analytics:{org_id}:{source_id}:error"
         return self.redis.publish(channel, json.dumps(payload))
 
-    def emit_analytics_trigger(self, org_id: str, source_id: str, extra: Dict | None = None):
-        """Publish trigger to stream (worker manager polls this)"""
-        payload = {
-             "org_id": org_id,
-            "source_id": source_id,
-            "timestamp": datetime.utcnow().isoformat(),
-        }
-        if extra:
-          payload.update(extra)
+    # app/core/event_hub.py
 
-        stream_key = f"stream:analytics_trigger:{org_id}:{source_id}"
+def emit_analytics_trigger(self, org_id: str, source_id: str, extra: dict | None = None):
+    """Write trigger to single stream (worker manager polls this)"""
+    stream_key = "stream:analytics_triggers"
     
-        try:
-            msg_id = self.redis.xadd(stream_key, {"message": json.dumps(payload)})
-            logger.info(f"[hub] 📤 emitted analytics trigger to {stream_key} (id: {msg_id})")
-            return msg_id
-        except Exception as e:
-            logger.error(f"[hub] ❌ failed to emit trigger: {e}", exc_info=True)
-            return None
+    payload = {
+        "org_id": org_id,
+        "source_id": source_id,
+        "timestamp": datetime.utcnow().isoformat(),
+    }
+    if extra:
+        payload.update(extra)
+
+    try:
+        msg_id = self.redis.xadd(stream_key, {"message": json.dumps(payload)})
+        logger.info(f"[hub] 📤 emitted trigger to {stream_key} (id: {msg_id})")
+        return msg_id
+    except Exception as e:
+        logger.error(f"[hub] ❌ failed to emit trigger: {e}", exc_info=True)
+        return None
 
     def ensure_consumer_group(self, stream_key: str, group: str):
         try:

app/core/worker_manager.py

@@ -1,77 +1,91 @@
+# app/core/worker_manager.py
+
 import asyncio
 import json
-import aiohttp  # Add to requirements.txt
 from app.tasks.analytics_worker import AnalyticsWorker
-from app.redis_client import redis  # Use Upstash client
+from app.redis_client import redis
 import logging
 
 logger = logging.getLogger(__name__)
 
 class WorkerManager:
     def __init__(self):
-        self.active_workers = {}
+        self.active_workers: dict = {}
         self.lock_ttl = 300
+        self.stream_key = "stream:analytics_triggers"  # Single stream for all triggers
+        self.last_id = "0"  # Resume point
     
     async def start_listener(self):
-        """🎧 Poll Upstash HTTP API for trigger events instead of pubsub"""
-        logger.info("[manager] 🎧 worker manager listening for events via HTTP polling")
+        """🎧 Poll Upstash HTTP API for trigger events"""
+        logger.info("[manager] 🎧 worker manager listening via HTTP polling")
         
         while True:
             try:
-                # Poll Redis stream for trigger messages
-                # Stream key: stream:analytics_trigger:{orgId}:{sourceId}
-                stream_key = "stream:analytics_trigger:*"
-                
-                # Use Upstash HTTP client to read from stream
-                # XREAD with BLOCK 1000ms (1 second)
-                result = redis.xread({stream_key: "0"}, count=10, block=1000)
+                # XREAD without 'block' - Upstash doesn't support it
+                result = redis.xread({self.stream_key: self.last_id}, count=10)
                 
                 if result:
-                    for stream_name, messages in result:
-                        for msg_id, msg_data in messages:
-                            # Extract org_id and source_id from stream name
-                            parts = stream_name.decode().split(":")
-                            if len(parts) >= 4:
-                                org_id, source_id = parts[2], parts[3]
+                    # Upstash returns: {stream: [(msg_id, data), ...]}
+                    messages = result.get(self.stream_key, [])
+                    
+                    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})")
                                 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}")
+                        except json.JSONDecodeError as e:
+                            logger.error(f"[manager] ❌ JSON error: {e}")
                 else:
-                    # No messages, sleep before next poll
-                    await asyncio.sleep(1)
+                    await asyncio.sleep(1)  # Polling interval
                     
             except Exception as e:
                 logger.error(f"[manager] ❌ polling error: {e}", exc_info=True)
-                await asyncio.sleep(5)  # Back off on error
+                await asyncio.sleep(5)
     
     async def spawn_worker(self, org_id: str, source_id: str):
+        """Spawn a worker with distributed locking"""
         worker_key = f"{org_id}:{source_id}"
-        
-        # Use Upstash client for locks
-        if redis.exists(f"worker_lock:{worker_key}"):
-            return
+        lock_key = f"worker_lock:{worker_key}"
         
         try:
-            redis.setex(f"worker_lock:{worker_key}", self.lock_ttl, "1")
+            # Check if worker is already running
+            if redis.exists(lock_key):
+                logger.debug(f"[manager] ⏭️ worker already active: {worker_key}")
+                return
             
+            # Set lock with TTL
+            redis.setex(lock_key, self.lock_ttl, "1")
             logger.info(f"[manager] 🚀 spawning worker for {worker_key}")
+            
+            # Create and track worker
             worker = AnalyticsWorker(org_id, source_id)
             self.active_workers[worker_key] = worker
             
-            asyncio.create_task(self._run_worker(worker, worker_key))
+            # Run in background
+            asyncio.create_task(self._run_worker(worker, worker_key, lock_key))
             
         except Exception as e:
-            logger.error(f"[manager] ❌ failed to spawn: {e}", exc_info=True)
-            redis.delete(f"worker_lock:{worker_key}")
+            logger.error(f"[manager] ❌ spawn failed: {e}", exc_info=True)
+            redis.delete(lock_key)
     
-    async def _run_worker(self, worker: AnalyticsWorker, worker_key: str):
+    async def _run_worker(self, worker: AnalyticsWorker, worker_key: str, lock_key: str):
+        """Run worker and cleanup"""
         try:
             results = await worker.run()
-            logger.info(f"[manager] ✅ worker completed for {worker_key}")
+            logger.info(f"[manager] ✅ worker completed: {worker_key}")
         except Exception as e:
-            logger.error(f"[manager] ❌ worker failed for {worker_key}: {e}", exc_info=True)
+            logger.error(f"[manager] ❌ worker failed: {worker_key} - {e}", exc_info=True)
         finally:
-            if worker_key in self.active_workers:
-                del self.active_workers[worker_key]
-            
-            redis.delete(f"worker_lock:{worker_key}")
+            # Cleanup
+            self.active_workers.pop(worker_key, None)
+            redis.delete(lock_key)
 
 worker_manager = WorkerManager()