refactored for self hosted redis

Dockerfile

@@ -1,4 +1,3 @@
-
 # ---- 1. base image ---------------------------------------------------------
 FROM python:3.11-slim
 
@@ -13,30 +12,33 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
         ca-certificates \
         wget \
         unzip \
+        redis-server \      # ✅ ADD THIS
+        supervisor \        # ✅ ADD THIS
     && rm -rf /var/lib/apt/lists/*
 
 # ---- 3. upgrade pip & enable pre-built wheels ------------------------------
 RUN pip install --no-cache-dir --upgrade pip setuptools wheel
 
-
-# ---- 4. install Python deps (+ DuckDB driver) ------------------------------
 # ---- 4. install Python deps (+ DuckDB driver) ------------------------------
 COPY requirements.txt /tmp/requirements.txt
 RUN pip install --no-cache-dir --prefer-binary -r /tmp/requirements.txt && \
     pip install --no-cache-dir "duckdb>=1.0.0"
 
-# ---- 5. Pre-download VSS extension (matches DuckDB v1.0.0) ---------------
-# This prevents runtime download failures and startup crashes
-RUN mkdir -p /root/.duckdb/extensions/v1.0.0/linux_amd64 && \
-    wget -q https://extensions.duckdb.org/v1.0.0/linux_amd64/vss.duckdb_extension.gz  \
-         -O /root/.duckdb/extensions/v1.0.0/linux_amd64/vss.duckdb_extension.gz && \
-    gunzip /root/.duckdb/extensions/v1.0.0/linux_amd64/vss.duckdb_extension.gz
+# ---- 5. Pre-download VSS extension (matches DuckDB v1.1.3) ---------------
+# ✅ CHANGED: v1.1.3 instead of v1.0.0
+RUN mkdir -p /root/.duckdb/extensions/v1.1.3/linux_amd64 && \
+    wget -q https://extensions.duckdb.org/v1.1.3/linux_amd64/vss.duckdb_extension.gz \
+         -O /root/.duckdb/extensions/v1.1.3/linux_amd64/vss.duckdb_extension.gz && \
+    gunzip /root/.duckdb/extensions/v1.1.3/linux_amd64/vss.duckdb_extension.gz
 # ---- 6. copy source --------------------------------------------------------
 COPY . /app
 WORKDIR /app
 
 # ---- 7. scheduler loop ----------------------------------------------------
 COPY scheduler_loop.py /app/scheduler_loop.py
-#  corrected duck db
-# ---- 8. start both services -----------------------------------------------
-CMD sh -c "python -m uvicorn app.main:app --host 0.0.0.0 --port 7860 & python /app/scheduler_loop.py"
+
+# ---- 8. copy supervisord config -------------------------------------------
+COPY supervisord.conf /etc/supervisor/conf.d/supervisord.conf  # ✅ ADD THIS
+
+# ---- 9. start all services via supervisord --------------------------------
+CMD ["/usr/bin/supervisord", "-c", "/etc/supervisor/conf.d/supervisord.conf"]

app/core/event_hub.py

@@ -13,7 +13,7 @@ logger = logging.getLogger(__name__)
 class EventHub:
     def __init__(self):
         self.redis = get_redis()
-
+        self.is_rest_api = not hasattr(self.redis, 'pubsub')
     # Generic key helpers
     def get_key(self, key: str):
         return self.redis.get(key)
@@ -30,6 +30,23 @@ class EventHub:
 
     def delete(self, key: str):
         return self.redis.delete(key)
+    
+    # ✅ ADD: Raw command execution compatibility
+    def execute_command(self, *args):
+        """
+        Execute raw Redis command (works for both TCP and Upstash)
+        Usage: execute_command("XADD", "stream", "*", "field", "value")
+        """
+        try:
+            if self.is_rest_api:
+                # Upstash: pass as list to execute()
+                return self.redis.execute(list(args))
+            else:
+                # TCP Redis: native execute_command
+                return self.redis.execute_command(*args)
+        except Exception as e:
+            logger.error(f"[hub] ❌ Command failed {args}: {e}")
+            raise
 
     # Stream & pub/sub helpers
     def stream_key(self, org_id: str, source_id: str) -> str:
@@ -82,7 +99,7 @@ class EventHub:
     def emit_analytics_trigger(self, org_id: str, source_id: str, extra: dict | None = None):
         """Write trigger to centralized stream"""
         stream_key = "stream:analytics_triggers"
-    
+        
         payload = {
             "org_id": org_id,
             "source_id": source_id,
@@ -92,15 +109,15 @@ class EventHub:
             payload.update(extra)
 
         try:
-            # ✅ FIX: Pass command as a LIST, not multiple arguments
-            msg_id = self.redis.execute([
+            # ✅ Use compatibility wrapper
+            msg_id = self.execute_command(
                 "XADD", 
-                stream_key, 
-             "*",  # Auto-generate ID
-                "message", 
+                stream_key,
+                "*",  # Auto-generate ID
+                "message",
                 json.dumps(payload)
-            ])
-        
+            )
+            
             logger.info(f"[hub] 📤 trigger emitted: {org_id}:{source_id} (msg: {msg_id})")
             return msg_id
         except Exception as e:

app/deps.py

@@ -130,7 +130,7 @@ def get_duckdb(org_id: str) -> duckdb.DuckDBPyConnection:
             
             try:
                 conn = duckdb.connect(str(db_file), read_only=False)
-                conn.execute("SET hnsw_enable_experimental_persistence = true")
+ 
                 # Enable VSS
                 conn.execute("INSTALL vss;")
                 conn.execute("LOAD vss;")
@@ -222,91 +222,69 @@ def get_vector_db(org_id: Optional[str] = None) -> duckdb.DuckDBPyConnection:
     return _vector_db_connections[org_id]
 
 
-# ── Redis Client (TCP + Upstash Compatible) ─────────────────────────────────────
+# ── Redis Client (self hosted  TCP + Upstash Compatible) ─────────────────────────────────────
 _redis_client = None
-_redis_config_cache: Dict[str, Any] = {}
-
+_redis_lock = threading.Lock()
 def get_redis():
     """
-    🔄 Returns Redis client (TCP or Upstash HTTP)
-    Singleton pattern with config caching
+    🎯 Redis connection with clear priority:
+    1. Self-hosted (TCP) - HF Spaces with supervisord
+    2. Upstash (HTTP) - Fallback only
+    3. Local dev mock - Last resort
     """
-    global _redis_client, _redis_config_cache
-    
-    if _redis_client is not None:
-        return _redis_client
+    global _redis_client
     
-    # Check for TCP Redis first
-    redis_host = os.getenv("REDIS_HOST")
-    if redis_host:
-        logger.info("[REDIS] 🔌 Initializing TCP Redis client")
-        
-        import redis as redis_py
+    with _redis_lock:
+        if _redis_client is not None:
+            return _redis_client
         
-        redis_url = os.getenv("REDIS_URL")
-        if redis_url and redis_url.startswith("redis://"):
-            from urllib.parse import urlparse
-            parsed = urlparse(redis_url)
-            
-            _redis_client = redis_py.Redis(
-                host=parsed.hostname or redis_host,
-                port=parsed.port or int(os.getenv("REDIS_PORT", 6379)),
-                password=parsed.password or os.getenv("REDIS_PASSWORD"),
-                username=parsed.username or os.getenv("REDIS_USER"),
-                decode_responses=True,
-                ssl=bool(os.getenv("REDIS_SSL", False)),
-                ssl_cert_reqs=None,
-                socket_keepalive=True,
-                socket_connect_timeout=5,
-                socket_timeout=5,
-                connection_pool=redis_py.ConnectionPool(
-                    max_connections=int(os.getenv("REDIS_MAX_CONNECTIONS", "10")),
-                    retry_on_timeout=True,
-                    socket_keepalive=True,
-                )
-            )
-        else:
-            _redis_client = redis_py.Redis(
-                host=redis_host,
-                port=int(os.getenv("REDIS_PORT", 6379)),
-                password=os.getenv("REDIS_PASSWORD", None),
-                decode_responses=True,
-                socket_keepalive=True,
-                connection_pool=redis_py.ConnectionPool(
-                    max_connections=int(os.getenv("REDIS_MAX_CONNECTIONS", "10")),
+        # 1. Self-hosted Redis (HF Spaces)
+        redis_url = os.getenv("REDIS_URL", "redis://localhost:6379")
+        if redis_url.startswith("redis://"):
+            try:
+                import redis as redis_py
+                _redis_client = redis_py.from_url(
+                    redis_url,
+                    decode_responses=True,
+                    socket_connect_timeout=2,
+                    socket_timeout=2,
+                    retry_on_timeout=True
                 )
-            )
+                # Test connection immediately
+                _redis_client.ping()
+                logger.info(f"✅ Redis connected: {redis_url} (TCP)")
+                return _redis_client
+            except Exception as e:
+                logger.warning(f"⚠️ TCP Redis failed: {e}")
         
-        _redis_config_cache["type"] = "tcp"
-        return _redis_client
-    
-    # Fallback to Upstash HTTP
-    if REDIS_URL and REDIS_TOKEN:
-        logger.info("[REDIS] 🔌 Initializing Upstash HTTP Redis client")
+        # 2. Upstash fallback (only if explicit)
+        upstash_url = os.getenv("UPSTASH_REDIS_REST_URL")
+        upstash_token = os.getenv("UPSTASH_REDIS_REST_TOKEN")
+        
+        if upstash_url and upstash_token:
+            _redis_client = Redis(url=upstash_url, token=upstash_token)
+            logger.info("📡 Redis connected: Upstash (HTTP)")
+            return _redis_client
         
-        _redis_client = Redis(url=REDIS_URL, token=REDIS_TOKEN)
-        _redis_config_cache["type"] = "upstash"
+        # 3. Mock for local dev
+        logger.error("❌ No Redis available, using mock!")
+        from unittest.mock import Mock
+        _redis_client = Mock()
         return _redis_client
-    
-    # Local dev fallback
-    logger.warning("[REDIS] ⚠️ No config, using localhost:6379")
-    import redis as redis_py
-    _redis_client = redis_py.Redis(host="localhost", port=6379, decode_responses=True)
-    _redis_config_cache["type"] = "local"
-    return _redis_client
 
 
-def reset_redis_client():
-    """SRE: Reset connection pool if needed"""
+def reset_redis():
+    """SRE: Reset Redis connection (for testing)"""
     global _redis_client
-    if _redis_client:
-        try:
-            _redis_client.close()
-        except Exception:
-            pass
     _redis_client = None
 
 
+# ── Event Hub Connection Type Detection ─────────────────────────────────────────
+def is_tcp_redis() -> bool:
+    """Check if using TCP Redis (pub/sub capable)"""
+    redis_url = os.getenv("REDIS_URL", "")
+    return redis_url.startswith("redis://")
+
 # ── QStash (Optional) ───────────────────────────────────────────────────────────
 _qstash_client = None
 _qstash_verifier = None

app/redis_async_client.py

@@ -1,39 +0,0 @@
-# app/redis_async_client.py
-"""
-Async Redis client for Upstash using native Redis protocol.
-This is separate from the Upstash HTTP client used elsewhere.
-"""
-
-import redis.asyncio as aioredis
-import os
-from urllib.parse import urlparse
-
-# Load from HF Secrets (exact names)
-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 not found in HF Secrets")
-
-# Parse the REST URL to get host and port
-parsed = urlparse(UPSTASH_REDIS_REST_URL)
-
-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
-redis_async = aioredis.from_url(
-    REDIS_URL,
-    password=UPSTASH_REDIS_REST_TOKEN,
-    decode_responses=True,
-    retry_on_timeout=True,
-    socket_keepalive=True,
-    socket_connect_timeout=10,
-    max_connections=10,
-)
-
-# Export for use
-__all__ = ["redis_async"]

app/redis_client.py

@@ -1,13 +1,13 @@
-# app/redis_client.py
+# app/redis_client.py – Lazy Singleton (No Startup Crash)
 from app.deps import get_redis
 
-# Export the singleton instance
+# Export the singleton instance (lazy, doesn't connect until first use)
 redis = get_redis()
 
-# Test on import
-try:
-    redis.ping()
-    print("✅ Redis bridge connected")
-except Exception as e:
-    print(f"❌ Redis connection failed: {e}")
-    raise RuntimeError(f"Redis not available: {e}")
+# ✅ REMOVE: Don't ping on import - causes startup race condition
+# try:
+#     redis.ping()
+#     print("✅ Redis bridge connected")
+# except Exception as e:
+#     print(f"❌ Redis connection failed: {e}")
+#     raise RuntimeError(f"Redis not available: {e}")

supervisord.conf

@@ -0,0 +1,32 @@
+[supervisord]
+nodaemon=true
+logfile=/var/log/supervisor/supervisord.log
+pidfile=/var/run/supervisord.pid
+
+[program:redis]
+command=/usr/bin/redis-server /etc/redis/redis.conf
+autostart=true
+autorestart=true
+priority=1
+stdout_logfile=/var/log/supervisor/redis.log
+stderr_logfile=/var/log/supervisor/redis_error.log
+startsecs=2  # Wait 2 seconds to ensure Redis is ready
+
+[program:uvicorn]
+command=python -m uvicorn app.main:app --host 0.0.0.0 --port 7860 --workers 1
+directory=/app
+autostart=true
+autorestart=true
+priority=2
+stdout_logfile=/var/log/supervisor/uvicorn.log
+stderr_logfile=/var/log/supervisor/uvicorn_error.log
+startsecs=3  # Wait for Redis to be ready
+
+[program:scheduler]
+command=python /app/scheduler_loop.py
+directory=/app
+autostart=true
+autorestart=true
+priority=3
+stdout_logfile=/var/log/supervisor/scheduler.log
+stderr_logfile=/var/log/supervisor/scheduler_error.log