added https instead of pub/redis

app/core/event_hub.py

@@ -6,10 +6,10 @@ and read recent stream entries without importing `redis` directly.
 import json
 from datetime import datetime
 from typing import Any, Dict
-
+import logging
 from app.deps import get_redis
 
-
+logger = logging.getLogger(__name__)
 class EventHub:
     def __init__(self):
         self.redis = get_redis()
@@ -72,27 +72,23 @@ class EventHub:
         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 both to pubsub (for immediate listeners) and add to a small stream
-        so systems listening to keyspace events (xadd) can also react.
-        """
+        """Publish trigger to stream (worker manager polls this)"""
         payload = {
-            "org_id": org_id,
+             "org_id": org_id,
             "source_id": source_id,
             "timestamp": datetime.utcnow().isoformat(),
         }
         if extra:
-            payload.update(extra)
+          payload.update(extra)
 
-        # PubSub for backward compatibility
+        stream_key = f"stream:analytics_trigger:{org_id}:{source_id}"
+    
         try:
-            self.redis.publish(self.trigger_channel(org_id, source_id), json.dumps(payload))
-        except Exception:
-            pass
-
-        # Also write to a stream so consumers using streams can react
-        try:
-            return self.redis.xadd(f"stream:analytics_trigger:{org_id}:{source_id}", {"message": json.dumps(payload)})
-        except Exception:
+            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
 
     def ensure_consumer_group(self, stream_key: str, group: str):

app/core/worker_manager.py

@@ -1,9 +1,8 @@
-# app/core/worker_manager.py
 import asyncio
 import json
-from app.redis_async_client import redis_async  # ✅ NEW: Import async client
+import aiohttp  # Add to requirements.txt
 from app.tasks.analytics_worker import AnalyticsWorker
-from app.redis_client import redis  # ✅ Keep Upstash client for locks
+from app.redis_client import redis  # Use Upstash client
 import logging
 
 logger = logging.getLogger(__name__)
@@ -14,30 +13,39 @@ class WorkerManager:
         self.lock_ttl = 300
     
     async def start_listener(self):
-        logger.info("[manager] 🎧 worker manager listening for events")
+        """🎧 Poll Upstash HTTP API for trigger events instead of pubsub"""
+        logger.info("[manager] 🎧 worker manager listening for events via HTTP polling")
         
-        # ✅ FIX: Use async client for pubsub
-        pubsub = redis_async.pubsub()
-        
-        # Subscribe to pattern
-        await pubsub.psubscribe("analytics_trigger:*")
-        
-        # Listen async (this requires async client)
-        async for message in pubsub.listen():
-            if message["type"] == "pmessage":
-                try:
-                    channel = message["channel"].decode()
-                    if "analytics_trigger" in channel:
-                        parts = channel.split(":")
-                        org_id, source_id = parts[2], parts[3]
-                        await self.spawn_worker(org_id, source_id)
-                except Exception as e:
-                    logger.error(f"[manager] ❌ failed: {e}", exc_info=True)
+        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)
+                
+                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]
+                                await self.spawn_worker(org_id, source_id)
+                else:
+                    # No messages, sleep before next poll
+                    await asyncio.sleep(1)
+                    
+            except Exception as e:
+                logger.error(f"[manager] ❌ polling error: {e}", exc_info=True)
+                await asyncio.sleep(5)  # Back off on error
     
     async def spawn_worker(self, org_id: str, source_id: str):
         worker_key = f"{org_id}:{source_id}"
         
-        # ✅ Keep using Upstash sync client for simple lock operations
+        # Use Upstash client for locks
         if redis.exists(f"worker_lock:{worker_key}"):
             return
         
@@ -51,7 +59,7 @@ class WorkerManager:
             asyncio.create_task(self._run_worker(worker, worker_key))
             
         except Exception as e:
-            logger.error(f"[manager] ❌ failed: {e}", exc_info=True)
+            logger.error(f"[manager] ❌ failed to spawn: {e}", exc_info=True)
             redis.delete(f"worker_lock:{worker_key}")
     
     async def _run_worker(self, worker: AnalyticsWorker, worker_key: str):