[KM-565] sync catalog on document and db client deletion

- add CatalogStore.remove_source() to drop a single source by source_id
- add on_tabular_deleted / on_db_deleted triggers
- call triggers from delete endpoints so catalog stays in sync

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

src/api/v1/db_client.py

@@ -406,6 +406,8 @@ async def delete_database_client(
         raise HTTPException(status_code=403, detail="Access denied")
 
     await database_client_service.delete(db, client_id)
+    from src.pipeline.triggers import on_db_deleted
+    await on_db_deleted(client_id, user_id)
     return {"status": "success", "message": "Database client deleted successfully"}
 
 

src/api/v1/document.py

@@ -92,6 +92,8 @@ async def delete_document(
 ):
     """Delete a document."""
     await document_pipeline.delete(document_id, user_id, db)
+    from src.pipeline.triggers import on_tabular_deleted
+    await on_tabular_deleted(document_id, user_id)
     return {"status": "success", "message": "Document deleted successfully"}
  
  

src/catalog/store.py

@@ -63,6 +63,18 @@ class CatalogStore:
             sources=len(catalog.sources),
         )
 
+    async def remove_source(self, user_id: str, source_id: str) -> None:
+        existing = await self.get(user_id)
+        if existing is None:
+            logger.info("remove_source: no catalog found", user_id=user_id, source_id=source_id)
+            return
+        filtered = [s for s in existing.sources if s.source_id != source_id]
+        if len(filtered) == len(existing.sources):
+            logger.info("remove_source: source not in catalog", user_id=user_id, source_id=source_id)
+            return
+        await self.upsert(existing.model_copy(update={"sources": filtered}))
+        logger.info("remove_source: source removed", user_id=user_id, source_id=source_id)
+
     async def delete(self, user_id: str) -> None:
         async with AsyncSessionLocal() as session:
             await session.execute(delete(CatalogRow).where(CatalogRow.user_id == user_id))

src/pipeline/triggers.py

@@ -69,6 +69,22 @@ async def on_document_uploaded(document_id: str, user_id: str) -> None:
         await document_pipeline.process(document_id, user_id, db)
 
 
+async def on_tabular_deleted(document_id: str, user_id: str) -> None:
+    """Remove a tabular source from the user's catalog when its document is deleted."""
+    from src.catalog.store import CatalogStore
+
+    logger.info("on_tabular_deleted triggered", user_id=user_id, document_id=document_id)
+    await CatalogStore().remove_source(user_id, source_id=document_id)
+
+
+async def on_db_deleted(client_id: str, user_id: str) -> None:
+    """Remove a schema source from the user's catalog when its DB client is deleted."""
+    from src.catalog.store import CatalogStore
+
+    logger.info("on_db_deleted triggered", user_id=user_id, client_id=client_id)
+    await CatalogStore().remove_source(user_id, source_id=client_id)
+
+
 async def on_catalog_rebuild_requested(user_id: str) -> None:
     """Re-introspect every source in the user's catalog and upsert the result.