added correct reddis embed storage

app/service/vector_service.py

@@ -212,32 +212,72 @@ class VectorService:
     
     # Replace the _upsert_redis method in VectorService
 
-    def _upsert_redis(
+    # Make _upsert_redis async and non-blocking
+
+    async def _upsert_redis(
         self,
         embeddings: List[List[float]],
         metadata: List[Dict[str, Any]],
         namespace: str
-        ):
-        """Store in Redis with 24h TTL (Upstash-compatible, no pipeline)"""
+    ):
+        """Store in Redis with 24h TTL (non-blocking)"""
         try:
             stored = 0
-            for idx, (emb, meta) in enumerate(zip(embeddings, metadata)):
-                key = f"vector:{namespace}:{idx}:{int(time.time())}"
-                event_hub.setex(
-                    key,
-                    86400,  # 24 hours
-                    json.dumps({
-                        "embedding": emb,
-                        "metadata": meta,
-                        "org_id": self.org_id
-                    })
-                )
-                stored += 1
         
-            logger.info(f"[✅ VECTOR] Redis: Stored {stored} vectors sequentially")
+            # Process in smaller batches to reduce HTTP calls
+            batch_size = 50
+        
+            for i in range(0, len(embeddings), batch_size):
+                batch_embeddings = embeddings[i:i + batch_size]
+                batch_metadata = metadata[i:i + batch_size]
+            
+                # Store batch sequentially but non-blocking
+                for idx, (emb, meta) in enumerate(zip(batch_embeddings, batch_metadata)):
+                    global_idx = i + idx
+                    key = f"vector:{namespace}:{global_idx}:{int(time.time())}"
+                
+                    # ✅ FIX: Run in thread pool to avoid blocking
+                    await asyncio.to_thread(
+                        event_hub.setex,
+                        key,
+                        86400,
+                        json.dumps({
+                            "embedding": emb,
+                            "metadata": meta,
+                            "org_id": self.org_id
+                        })
+                    )
+                    stored += 1
+            
+                # Small delay every batch to prevent overwhelming Redis
+                if i > 0 and i % 200 == 0:
+                    await asyncio.sleep(0.01)
+        
+            logger.info(f"[✅ VECTOR] Redis: Stored {stored} vectors (non-blocking)")
         
         except Exception as e:
             logger.error(f"[❌ VECTOR] Redis error: {e}")
+
+# Also update upsert_embeddings to be async:
+
+async def upsert_embeddings(
+    self,
+    embeddings: List[List[float]],
+    metadata: List[Dict[str, Any]],
+    namespace: str
+):
+    """Store in BOTH Redis (hot) and DuckDB VSS (cold) - ASYNC"""
+    try:
+        # Run both storage operations concurrently
+        redis_task = self._upsert_redis(embeddings, metadata, namespace)
+        vss_task = asyncio.to_thread(self._upsert_vss, embeddings, metadata, namespace)
+        
+        await asyncio.gather(redis_task, vss_task)
+        
+        logger.info(f"[✅ VECTOR] Dual-store complete: {len(embeddings)} vectors")
+        
+    except Exception as e:
+        logger.error(f"[❌ VECTOR] Dual upsert failed: {e}", exc_info=True)