[NOTICKET] fix delete, now can filter by user

src/document/document_service.py

@@ -3,6 +3,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy import select, delete, text
 from src.db.postgres.models import Document
+from src.db.postgres.connection import _pgvector_engine
 from src.storage.az_blob.az_blob import blob_storage
 from src.middlewares.logging import get_logger
 from typing import List, Optional
@@ -77,11 +78,20 @@ class DocumentService:
         # Delete from blob storage
         await blob_storage.delete_file(document.blob_name)
 
-        # Delete vector embeddings from pgvector
-        await db.execute(
-            text("DELETE FROM langchain_pg_embedding WHERE cmetadata->'data'->>'document_id' = :doc_id"),
-            {"doc_id": document_id}
-        )
+        # Delete vector embeddings from pgvector (scoped to user + collection to avoid cross-user over-delete)
+        async with _pgvector_engine.begin() as conn:
+            await conn.execute(
+                text("""
+                    DELETE FROM langchain_pg_embedding
+                    WHERE cmetadata->>'user_id' = :user_id
+                      AND cmetadata->>'source_type' = 'document'
+                      AND cmetadata->'data'->>'document_id' = :doc_id
+                      AND collection_id = (
+                        SELECT uuid FROM langchain_pg_collection WHERE name = 'document_embeddings'
+                      )
+                """),
+                {"user_id": document.user_id, "doc_id": document_id},
+            )
 
         # Delete from database
         await db.execute(

src/pipeline/db_pipeline/db_pipeline_service.py

@@ -183,35 +183,39 @@ class DbPipelineService:
         vector_store = get_vector_store()
         logger.info("db pipeline start", user_id=user_id)
 
+        # Profile first — if this fails, old embeddings are untouched
+        schema = await asyncio.to_thread(get_schema, engine, exclude_tables)
+
+        updated_at = datetime.now(timezone(timedelta(hours=7))).isoformat()
+        all_docs: list = []
+        for table_name, columns in schema.items():
+            logger.info("profiling table", table=table_name, columns=len(columns))
+            entries = await asyncio.to_thread(profile_table, engine, table_name, columns)
+            docs = [self._to_document(user_id, client_id, table_name, e, updated_at) for e in entries]
+            all_docs.extend(docs)
+            logger.info("profiled table", table=table_name, count=len(docs))
+
+        # Delete only after all docs are ready
         async with _pgvector_engine.begin() as conn:
             result = await conn.execute(
                 text(
                     "DELETE FROM langchain_pg_embedding "
                     "WHERE cmetadata->>'user_id' = :user_id "
                     "  AND cmetadata->>'source_type' = 'database' "
+                    "  AND cmetadata->>'database_client_id' = :client_id "
                     "  AND collection_id = ("
                     "    SELECT uuid FROM langchain_pg_collection WHERE name = 'document_embeddings'"
                     "  )"
                 ),
-                {"user_id": user_id},
+                {"user_id": user_id, "client_id": client_id},
             )
             logger.info("cleared old db embeddings", user_id=user_id, deleted=result.rowcount)
 
-        schema = await asyncio.to_thread(get_schema, engine, exclude_tables)
-
-        updated_at = datetime.now(timezone(timedelta(hours=7))).isoformat()
-        total = 0
-        for table_name, columns in schema.items():
-            logger.info("profiling table", table=table_name, columns=len(columns))
-            entries = await asyncio.to_thread(profile_table, engine, table_name, columns)
-            docs = [self._to_document(user_id, client_id, table_name, e, updated_at) for e in entries]
-            if docs:
-                await vector_store.aadd_documents(docs)
-                total += len(docs)
-                logger.info("ingested chunks", table=table_name, count=len(docs))
+        if all_docs:
+            await vector_store.aadd_documents(all_docs)
 
-        logger.info("db pipeline complete", user_id=user_id, total=total)
-        return total
+        logger.info("db pipeline complete", user_id=user_id, total=len(all_docs))
+        return len(all_docs)
 
 
 db_pipeline_service = DbPipelineService()