loaded enity and industry from redis hub

app/tasks/analytics_worker.py

@@ -80,8 +80,8 @@ class AnalyticsWorker:
         try:
             logger.info(f"\n[WORKER] 🚀 STARTING {worker_id}")
             
-            # 🎯 STEP 2: Wait for entity/industry keys (exponential backoff)
-            await self._wait_for_entity_and_industry()
+            # ✅ STEP 2: INSTANT Redis read (no waiting, no polling)
+            entity_info = await self._load_entity_from_redis()
             
             # 🎯 STEP 3: Load data with retry logic
             df = await self._load_dataframe()
@@ -289,75 +289,34 @@ class AnalyticsWorker:
                 except:
                     pass
     
-    def _get_entity_info(self) -> Optional[Dict[str, Any]]:
-        """Get entity info from Redis with cache invalidation"""
+    async def _load_entity_from_redis(self) -> dict:
+        """Instantly load entity/industry from Redis (source of truth)"""
         try:
-            from app.mapper import _ENTITY_CACHE
-            cache_key = (self.org_id, self.source_id)
-            
+            # Read entity from Redis (written by mapper)
             entity_key = f"entity:{self.org_id}:{self.source_id}"
-            data = event_hub.get_key(entity_key)
-            
-            if data:
-                entity_info = json.loads(data)
-                _ENTITY_CACHE[cache_key] = entity_info
-                return entity_info
-            
-            _ENTITY_CACHE.pop(cache_key, None)
-            return None
-            
-        except Exception as e:
-            logger.error(f"[ENTITY] Error: {e}")
-            return None
-    
-    # ==================== ENTITY/INDUSTRY WAITING ====================
-    
-    async def _wait_for_entity_and_industry(self):
-        """Wait for entity and industry keys with exponential backoff"""
-        MAX_WAIT = 30.0
-        INITIAL_DELAY = 0.5
-        MAX_DELAY = 5.0
+            entity_data = await asyncio.to_thread(event_hub.get_key, entity_key)
         
-        entity_key = f"entity:{self.org_id}:{self.source_id}"
-        industry_key = f"industry:{self.org_id}:{self.source_id}"
+            if not entity_data:
+                raise ValueError(f"Entity key not found: {entity_key}")
+            
+            entity_info = json.loads(entity_data)
+            self._entity_type = entity_info["entity_type"]
         
-        delay = INITIAL_DELAY
-        start = time.time()
+            # Read industry from Redis
+            industry_key = f"industry:{self.org_id}:{self.source_id}"
+            industry_data = await asyncio.to_thread(event_hub.get_key, industry_key)
         
-        while (time.time() - start) < MAX_WAIT:
-            try:
-                # Use EXISTS (HTTP-safe)
-                ent_exists = event_hub.redis.exists(entity_key)
-                ind_exists = event_hub.redis.exists(industry_key)
-                
-                if ent_exists and ind_exists:
-                    # Update cache
-                    from app.mapper import _ENTITY_CACHE, _INDUSTRY_CACHE
-                    cache_key = (self.org_id, self.source_id)
-                    
-                    if cache_key not in _ENTITY_CACHE:
-                        data = event_hub.get_key(entity_key)
-                        if data:
-                            _ENTITY_CACHE[cache_key] = json.loads(data)
-                    
-                    if cache_key not in _INDUSTRY_CACHE:
-                        data = event_hub.get_key(industry_key)
-                        if data:
-                            _INDUSTRY_CACHE[cache_key] = json.loads(data)
-                    
-                    logger.info("[WORKER] ✅ Entity & industry keys found and validated")
-                    return
-                
-                logger.info(f"[WORKER] ⏳ Waiting for keys (entity={ent_exists}, industry={ind_exists})...")
-                
-            except Exception as e:
-                logger.debug(f"[WORKER] Redis check error: {e}")
+            if not industry_data:
+                raise ValueError(f"Industry key not found: {industry_key}")
             
-            await asyncio.sleep(delay)
-            delay = min(delay * 1.5, MAX_DELAY)
+            self._industry_info = json.loads(industry_data)
         
-        logger.warning(f"[WORKER] ⚠️ Timeout waiting for keys after {MAX_WAIT}s, proceeding anyway")
-    
+            logger.info(f"[WORKER] ✅ Loaded entity={self._entity_type}, industry={self._industry_info['industry']} from Redis")
+            return entity_info
+        
+        except Exception as e:
+            logger.error(f"[WORKER] ❌ Failed to load from Redis: {e}")
+            raise
     # ==================== SCHEMA & EMBEDDING ====================
     
     # app/tasks/analytics_worker.py - Replace your _discover_schema method