corrected datasource and canonify

app/mapper.py

@@ -490,14 +490,18 @@ def canonify_df(org_id: str, source_id: str, hours_window: int = 24) -> tuple[pd
 
     # After line: print(f"[canonify] ✅ Pipeline complete in {duration_ms:.2f}ms")
     if not df.empty:
-            redis.publish(
-                f"analytics_trigger:{org_id}:{source_id}",
-                json.dumps({
-                    "type": "kpi_compute",
-                    "entity_type": entity_type,
-                    "industry": industry
-                })
-            )
-    print(f"[canonify] 🚀 Triggered analytics for {source_id}")
+            # At the end of the canonify pipeline
+            try:
+                redis.publish(
+                    f"analytics_trigger:{org_id}:{source_id}",
+                    json.dumps({
+                        "type": "kpi_compute",
+                        "entity_type": entity_type,
+                        "industry": industry
+                    })
+                )
+                 print(f"[canonify] 🚀 Triggered analytics for {source_id}")
+            except Exception as e:
+            print(f"[canonify] ⚠️ Analytics trigger failed (non-critical): {e}")
     
     return df, industry, industry_confidence

app/routers/datasources.py

@@ -3,7 +3,7 @@ from typing import Dict, Any, List, Union
 from fastapi.responses import JSONResponse
 from pydantic import BaseModel
 from typing import List, Any, Dict, Union
-from app.deps import verify_api_key,get_current_user
+from app.deps import verify_api_key,get_current_user,APP_URL
 from app.db import get_conn, ensure_raw_table, bootstrap
 from app.mapper import canonify_df
 from app.routers.socket import sio
@@ -80,22 +80,36 @@ 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(...)
+
+        try:
+            # Use APP_URL from deps (no need for settings)
+            trigger_data = {
+                "type": "kpi_compute",
+                "org_id": org_id,
+                "source_id": source_id
+            }
+    
+            # Try QStash if available
+            if is_qstash_available():
+                try:
+                     from app.qstash_client import get_qstash_client
+                     client = get_qstash_client()
+                     result = client.message.publish(
+                         url=f"{APP_URL}/api/v1/analytics/callback",
+                         body=trigger_data
+                        )
+                        logger.info(f"📤 QStash queued: {result.message_id}")
+                except Exception as e:
+                     logger.warning(f"⚠️ QStash failed, using Redis: {e}")
+                     redis.publish(f"analytics_trigger:{org_id}:{source_id}", json.dumps(trigger_data))
+            else:
+            # ✅ Fallback: Direct Redis publish
+            redis.publish(f"analytics_trigger:{org_id}:{source_id}", json.dumps(trigger_data))
             logger.info("📡 Redis fallback for analytics trigger")
+
+        except Exception as e:
+        logger.warning(f"⚠️ Analytics trigger failed (non-critical): {e}")
+        # ✅ DON'T raise - ingestion should succeed even if analytics fails
         # 4. 📡 Broadcast to connected dashboards
         await sio.emit(
             "datasource:new-rows",