added redis emiter

app/mapper.py

@@ -162,10 +162,14 @@ def _fallback_detection(org_id: str, source_id: str) -> dict:
         entity_info = {"entity_type": entity_type, "confidence": confidence}
         print(f"[fallback] ✅ Direct detection: {entity_type} ({confidence:.2%})")
     
-    # Cache and write to Redis so next time it works
+    # ✅ CRITICAL: Write to Redis BEFORE returning
+    redis_key = f"entity:{org_id}:{source_id}"
+    event_hub.setex(redis_key, 3600, json.dumps(entity_info))
+    print(f"[fallback] 💾 WRITTEN TO REDIS: {redis_key}")
+    
+    # ✅ Also populate module cache for immediate reuse
     cache_key = (org_id, source_id)
     _ENTITY_CACHE[cache_key] = entity_info
-    event_hub.setex(f"entity:{org_id}:{source_id}", 3600, json.dumps(entity_info))
     
     return entity_info
     #poll for industry from redis
@@ -258,15 +262,22 @@ def _fallback_industry_detection(org_id: str, source_id: str) -> dict:
             industry_info = {"industry": industry, "confidence": confidence}
             print(f"[fallback_industry] ✅ Direct detection: {industry} ({confidence:.2%})")
         
-        # Cache and write to Redis via hub
+        # ✅ CRITICAL: Write to Redis BEFORE returning
+        redis_key = f"industry:{org_id}:{source_id}"
+        event_hub.setex(redis_key, 3600, json.dumps(industry_info))
+        print(f"[fallback_industry] 💾 WRITTEN TO REDIS: {redis_key}")
+        
+        # ✅ Also populate module cache
         cache_key = (org_id, source_id)
         _INDUSTRY_CACHE[cache_key] = industry_info
-        event_hub.setex(f"industry:{org_id}:{source_id}", 3600, json.dumps(industry_info))
         
         return industry_info
         
     except Exception as e:
         print(f"[fallback_industry] ❌ Failed: {e}")
+        # ✅ Even on error, write UNKNOWN to Redis so worker doesn't hang
+        redis_key = f"industry:{org_id}:{source_id}"
+        event_hub.setex(redis_key, 3600, json.dumps({"industry": "UNKNOWN", "confidence": 0.0}))
         return {"industry": "UNKNOWN", "confidence": 0.0}
 
 
@@ -521,8 +532,8 @@ def canonify_df(org_id: str, source_id: str, hours_window: int = 24) -> tuple[pd
                 event_hub.emit_analytics_trigger(org_id, source_id, {
                     "type": "kpi_compute",
                     "entity_type": entity_type,
-                    "industry": industry
-                })
+                    "industry": industry,
+                    "rows_inserted": len(df)                  })
                 print(f"[canonify] 🚀 Triggered analytics for {source_id}")
             except Exception as e:
                 print(f"[canonify] ⚠️ Analytics trigger failed (non-critical): {e}")