fixed redis_async_client

app/core/worker_manager.py

@@ -1,7 +1,7 @@
 # app/core/worker_manager.py
 import asyncio
 import json
-from app.redis_async_client import redis_async  # ✅ Use async client
+from app.redis_async_client import redis_async  # ✅ NEW: Import async client
 from app.tasks.analytics_worker import AnalyticsWorker
 from app.redis_client import redis  # ✅ Keep Upstash client for locks
 import logging
@@ -16,10 +16,13 @@ class WorkerManager:
     async def start_listener(self):
         logger.info("[manager] 🎧 worker manager listening for events")
         
-        # ✅ Use async Redis client for pubsub
+        # ✅ 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:
@@ -34,13 +37,11 @@ class WorkerManager:
     async def spawn_worker(self, org_id: str, source_id: str):
         worker_key = f"{org_id}:{source_id}"
         
-        # ✅ Use Upstash sync client for simple lock operations
+        # ✅ Keep using Upstash sync client for simple lock operations
         if redis.exists(f"worker_lock:{worker_key}"):
-            logger.debug(f"[manager] ⏭️ worker already active for {worker_key}")
             return
         
         try:
-            # ✅ Use Upstash sync client for lock
             redis.setex(f"worker_lock:{worker_key}", self.lock_ttl, "1")
             
             logger.info(f"[manager] 🚀 spawning worker for {worker_key}")

app/redis_async_client.py

@@ -1,14 +1,7 @@
 # app/redis_async_client.py
 """
 Async Redis client for Upstash using native Redis protocol.
-
-This module creates an async connection pool for pubsub operations
-required by the worker manager, while using the same Upstash credentials
-from HF Secrets.
-
-Environment Variables (from HF Secrets):
-- UPSTASH_REDIS_REST_URL: The HTTP REST URL (e.g., https://host[:port])
-- UPSTASH_REDIS_REST_TOKEN: The authentication token
+This is separate from the Upstash HTTP client used elsewhere.
 """
 
 import redis.asyncio as aioredis
@@ -20,23 +13,18 @@ UPSTASH_REDIS_REST_URL = os.getenv("UPSTASH_REDIS_REST_URL")
 UPSTASH_REDIS_REST_TOKEN = os.getenv("UPSTASH_REDIS_REST_TOKEN")
 
 if not UPSTASH_REDIS_REST_URL:
-    raise RuntimeError("UPSTASH_REDIS_REST_URL is not set in HF Secrets")
+    raise RuntimeError("UPSTASH_REDIS_REST_URL not found in HF Secrets")
 
-# Parse the REST URL to extract host and port
+# Parse the REST URL to get host and port
 parsed = urlparse(UPSTASH_REDIS_REST_URL)
-host = parsed.hostname
 
-if not host:
-    raise RuntimeError(f"Could not parse hostname from UPSTASH_REDIS_REST_URL: {UPSTASH_REDIS_REST_URL}")
-
-# Default Redis port if not specified
+host = parsed.hostname
 port = parsed.port if parsed.port else 6379
 
 # Construct native Redis protocol URL
 REDIS_URL = f"redis://{host}:{port}"
 
-# Create async Redis client connection pool
-# Upstash token is used as the Redis AUTH password
+# Create async Redis client
 redis_async = aioredis.from_url(
     REDIS_URL,
     password=UPSTASH_REDIS_REST_TOKEN,
@@ -47,17 +35,5 @@ redis_async = aioredis.from_url(
     max_connections=10,
 )
 
-async def test_async_connection():
-    """
-    Test the async Redis connection.
-    Call this during app startup to verify connectivity.
-    """
-    try:
-        await redis_async.ping()
-        print(f"✅ Async Redis connected to {host}:{port}")
-        return True
-    except Exception as e:
-        print(f"❌ Async Redis connection failed: {e}")
-        return False
-
-__all__ = ["redis_async", "test_async_connection"]
+# Export for use
+__all__ = ["redis_async"]