added fallback entity and industry to be streamed to reddis

.gitignore

@@ -2,3 +2,6 @@ node_modules
 client-nextjs/googlecalendar.json
 .env.local
 analytics-service/.env.analytics
+analytics-data/duckdb/*.duckdb
+analytics-data/duckdb/*.wal
+analytics-data/duckdb/*

app/main.py

@@ -329,6 +329,21 @@ def debug_stream(
         "entity_data": event_hub.get_key(entity_key),
         "industry_data": event_hub.get_key(industry_key),
     }
+@app.post("/api/v1/cache/clear")
+def clear_cache(org_id: str, source_id: str, api_key: str = Depends(verify_api_key)):
+    """Clear entity/industry caches to force fresh reads"""
+    cache_key = (org_id, source_id)
+    
+    # Import the cache dicts
+    from app.mapper import _ENTITY_CACHE, _INDUSTRY_CACHE
+    
+    if cache_key in _ENTITY_CACHE:
+        del _ENTITY_CACHE[cache_key]
+    if cache_key in _INDUSTRY_CACHE:
+        del _INDUSTRY_CACHE[cache_key]
+    
+    return {"status": "cleared", "cache_key": str(cache_key)}
+
 # ─── Root Endpoint ─────────────────────────────────────────────────────────────
 @app.get("/", tags=["root"])
 def read_root():

app/mapper.py

@@ -119,7 +119,20 @@ def poll_for_entity(org_id: str, source_id: str, timeout: int = 30) -> dict:
     
     # 5. Emergency fallback (worker is dead)
     print("[poll] ⚠️ Both attempts failed - using direct detection")
-    return _fallback_detection(org_id, source_id)
+    entity_info = _fallback_detection(org_id, source_id)
+    
+    # 🎯 NEW: Force write to Redis (ensure it's there)
+    event_hub.setex(
+        f"entity:{org_id}:{source_id}", 
+        3600, 
+        json.dumps(entity_info)
+    )
+    
+    # 🎯 NEW: Clear stale cache so next read is fresh
+    if (org_id, source_id) in _ENTITY_CACHE:
+        del _ENTITY_CACHE[(org_id, source_id)]
+    
+    return entity_info
 
 
 def _fallback_detection(org_id: str, source_id: str) -> dict:
@@ -197,7 +210,20 @@ def poll_for_industry(org_id: str, source_id: str, timeout: int = 10) -> dict:
     
     # 5. Emergency fallback (worker is dead)
     print("[poll_industry] ⚠️ Both attempts failed - using direct detection")
-    return _fallback_industry_detection(org_id, source_id)
+    industry_info = _fallback_industry_detection(org_id, source_id)
+    
+    # 🎯 NEW: Force write to Redis (ensure it's there)
+    event_hub.setex(
+        f"industry:{org_id}:{source_id}", 
+        3600, 
+        json.dumps(industry_info)
+    )
+    
+    # 🎯 NEW: Clear stale cache so next read is fresh
+    if (org_id, source_id) in _INDUSTRY_CACHE:
+        del _INDUSTRY_CACHE[(org_id, source_id)]
+    
+    return industry_info
     #fallback industry detection
 def _fallback_industry_detection(org_id: str, source_id: str) -> dict:
     """

app/tasks/analytics_worker.py

@@ -54,7 +54,8 @@ class AnalyticsWorker:
         """
         start_time = datetime.now()
         logger.info(f"\n[WORKER] 🚀 STARTING {self.org_id}/{self.source_id}")
-        
+        # 🎯 NEW: Wait for entity/industry keys to exist
+        await self._wait_for_entity_and_industry()
         try:
             # 1️⃣ LOAD DATA (handles missing tables)
             df = await self._load_dataframe()
@@ -304,7 +305,23 @@ class AnalyticsWorker:
             return "supermarket"
         except:
             return "supermarket"
-    
+    async def _wait_for_entity_and_industry(self):
+        """Block until entity and industry are detected (max 30s)"""
+        max_wait = 30
+        start = time.time()
+        
+        while (time.time() - start) < max_wait:
+            entity_key = f"entity:{self.org_id}:{self.source_id}"
+            industry_key = f"industry:{self.org_id}:{self.source_id}"
+            
+            if event_hub.exists(entity_key) and event_hub.exists(industry_key):
+                logger.info(f"[WORKER] ✅ Entity & industry keys found")
+                return
+            
+            logger.info(f"[WORKER] ⏳ Waiting for entity/industry keys...")
+            await asyncio.sleep(2)
+        
+        logger.warning(f"[WORKER] ⚠️ Timeout waiting for keys, proceeding anyway")
     # Change _publish() method to use streams
 async def _publish(self, results: Dict[str, Any]):
     try: