refactored analytics stream

app/deps.py

@@ -43,6 +43,8 @@ HF_API_TOKEN = get_secret("HF_API_TOKEN", required=False)
 
 # QStash Token (optional, for advanced queue features)
 QSTASH_TOKEN = get_secret("QSTASH_TOKEN", required=False)
+# Application URL (where this HF Space is hosted)
+APP_URL = get_secret("SPACE_HOST", required=True) 
 
 # ── Singleton Database Connections ──────────────────────────────────────────────
 _org_db_connections = {}

app/main.py

@@ -25,6 +25,7 @@ from app.tasks.analytics_worker import redis_listener, trigger_kpi_computation
 from app.service.vector_service import cleanup_expired_vectors
 from app.routers import health, datasources, reports, flags, scheduler, run, socket, analytics_stream,ai_query,schema
 from app.service.llm_service import load_llm_service
+from app.deps import get_qstash_client
 # ─── Logger Configuration ───────────────────────────────────────────────────────
 logging.basicConfig(
     level=logging.INFO,
@@ -112,7 +113,12 @@ async def lifespan(app: FastAPI):
         logger.info("🤖 LLM service loading in background...")
     except Exception as e:
         logger.error(f"❌ LLM load failed: {e}")
-        # Continue anyway - LLM is optional for some features
+    # Continue anyway - LLM is optional for some features
+    try:
+        get_qstash_client()  # This creates the singleton if not exists
+        logger.info("✅ QStash ready")
+    except RuntimeError as e:
+        logger.warning(f"⚠️ QStash disabled: {e}")
     yield
     
     # ─── Shutdown ──────────────────────────────────────────────────────────────

app/qstash_client.py

@@ -1,69 +1,37 @@
 # app/qstash_client.py
-import os
 import logging
-from typing import Optional
+from typing import Optional, Dict, Any
+from app.deps import get_qstash_client  # ✅ Import from existing logic
 
 logger = logging.getLogger(__name__)
 
-# ✅ Module-level singleton (survives imports)
-_qstash_client: Optional["Client"] = None
-
-def init_qstash_client() -> bool:
+def is_qstash_available() -> bool:
     """
-    EXPLICITLY initialize QStash client on startup.
-    Call this AFTER HF secrets are loaded.
-    
-    Returns:
-        bool: True if successful, False if QStash not configured
+    Check if QStash is available without raising errors.
+    Uses the singleton from deps.py
     """
-    global _qstash_client
-    
-    if _qstash_client is not None:
-        logger.info("✅ QStash client already initialized")
-        return True
-    
-    token = os.getenv("QSTASH_TOKEN")
-    if not token:
-        logger.warning("⚠️ QSTASH_TOKEN not found - QStash features disabled")
-        return False
-    
     try:
-        from upstash_qstash import Client
-        
-        qstash_url = os.getenv("QSTASH_URL")
-        if qstash_url:
-            _qstash_client = Client(token=token, url=qstash_url)
-            logger.info(f"✅ QStash client initialized (custom URL: {qstash_url})")
-        else:
-            _qstash_client = Client(token=token)
-            logger.info("✅ QStash client initialized")
-        
+        get_qstash_client()
         return True
-        
-    except ImportError:
-        logger.error("❌ upstash_qstash not installed. pip install upstash-qstash")
-        return False
-    except Exception as e:
-        logger.error(f"❌ QStash initialization failed: {e}")
+    except RuntimeError:
         return False
 
-def get_qstash_client() -> Optional["Client"]:
+def publish_message(url: str, body: Dict[str, Any], callback: Optional[str] = None) -> Dict[str, Any]:
     """
-    Get the singleton QStash client.
+    Publish a message to QStash using the singleton client from deps.
+    
+    Args:
+        url: Endpoint URL to call
+        body: JSON payload
+        callback: Optional callback URL
     
     Returns:
-        Client instance or None if not initialized
+        Dict with message_id
         
     Raises:
-        RuntimeError: If called before init_qstash_client()
+        RuntimeError: If QStash not initialized
     """
-    if _qstash_client is None:
-        raise RuntimeError(
-            "QStash client not initialized. "
-            "Call init_qstash_client() in startup sequence first."
-        )
-    return _qstash_client
-
-def is_qstash_available() -> bool:
-    """Check if QStash is configured and ready"""
-    return _qstash_client is not None
+    client = get_qstash_client()
+    result = client.message.publish(url=url, body=body, callback=callback)
+    
+    return {"message_id": result.message_id}

app/routers/analytics_stream.py

@@ -4,7 +4,8 @@ from typing import List, Dict
 import json
 import asyncio
 from datetime import datetime
-from app.deps import get_current_user
+from app.deps import get_current_user,APP_URL
+
 from app.redis_client import redis
 import uuid
 from app.qstash_client import publish_message, is_qstash_available
@@ -85,6 +86,9 @@ async def get_recent_analytics(
         "timestamp": datetime.utcnow().isoformat()
     }
 
+# Add import at top
+from app.deps import APP_URL
+
 @router.post("/trigger")
 async def trigger_kpi_computation(
     source_id: str = Query(...),
@@ -99,21 +103,25 @@ async def trigger_kpi_computation(
             detail="QStash not configured. Check HF secrets."
         )
     
-    # Check cache (your existing logic)
+    # Check cache
     cached = redis.get(f"kpi_cache:{org_id}:{source_id}")
     if cached:
         return {"status": "cached", "data": json.loads(cached)}
     
+    # ✅ Use APP_URL from deps
+    callback_url = f"{APP_URL}/api/v1/analytics/callback"
+    notify_url = f"{APP_URL}/api/v1/analytics/notify"
+    
     # Publish to QStash
     try:
         result = publish_message(
-            url=f"{settings.APP_URL}/api/v1/analytics/callback",
+            url=callback_url,
             body={
                 "org_id": org_id,
                 "source_id": source_id,
                 "user_id": current_user["user_id"]
             },
-            callback=f"{settings.APP_URL}/api/v1/analytics/notify"
+            callback=notify_url
         )
         
         return {
@@ -123,8 +131,8 @@ async def trigger_kpi_computation(
         }
         
     except Exception as e:
-        raise HTTPException(status_code=500, detail=f"QStash error: {str(e)}")
-
+        logger.error(f"QStash publish failed: {e}")
+        raise HTTPException(status_code=500, detail="Failed to queue analytics")
 # app/routers/analytics_stream.py
 from fastapi import BackgroundTasks, Body, Depends  # ✅ Add imports