corrected logging

app/routers/analytics_stream.py

@@ -8,7 +8,7 @@ from app.deps import get_current_user
 from app.redis_client import redis
 import uuid
 from app.qstash_client import publish_message, is_qstash_available
-
+from fastapi import BackgroundTasks, Body, Depends
 router = APIRouter(prefix="/api/v1/analytics/stream", tags=["analytics"])
 
 class AnalyticsStreamManager:
@@ -125,19 +125,20 @@ async def trigger_kpi_computation(
     except Exception as e:
         raise HTTPException(status_code=500, detail=f"QStash error: {str(e)}")
 
+# app/routers/analytics_stream.py
+from fastapi import BackgroundTasks, Body, Depends  # ✅ Add imports
+
 @router.post("/callback")
 async def qstash_kpi_callback(
-    payload: Dict = Body(...),
-    background_tasks: BackgroundTasks
+    background_tasks: BackgroundTasks,  # ✅ First (no default)
+    payload: Dict = Body(...),  # ✅ Second (has default)
 ):
     """QStash calls this to compute KPIs"""
     org_id = payload["org_id"]
     source_id = payload["source_id"]
     
     # Trigger background computation
-    background_tasks.add_task(
-        run_analytics_worker, org_id, source_id
-    )
+    background_tasks.add_task(run_analytics_worker, org_id, source_id)
     
     return {"status": "accepted"}
 

app/routers/datasources.py

@@ -13,6 +13,7 @@ from datetime import datetime, timedelta
 from app.redis_client import redis 
 # Add this import
 from app.tasks.analytics_worker import trigger_kpi_computation
+from app.qstash_client import get_qstash_client, is_qstash_available
 router = APIRouter(tags=["datasources"])  # Remove 
 
 
@@ -79,7 +80,22 @@ async def create_source_json(
                preview_df[col] = preview_df[col].astype(str)
 
         preview_rows = preview_df.to_dict("records") if not preview_df.empty else []
-        
+        if is_qstash_available():
+            try:
+                client = get_qstash_client()
+                result = client.message.publish(
+                    url=f"{settings.APP_URL}/api/v1/analytics/callback",
+                    body={"org_id": org_id, "source_id": source_id}
+                )
+                logger.info(f"📤 QStash queued: {result.message_id}")
+            except Exception as e:
+             logger.warning(f"⚠️ QStash failed, using Redis: {e}")
+                # Fallback to Redis publish
+            redis.publish(...)
+        else:
+            # ✅ Fallback when QStash not configured
+            redis.publish(...)
+            logger.info("📡 Redis fallback for analytics trigger")
         # 4. 📡 Broadcast to connected dashboards
         await sio.emit(
             "datasource:new-rows",