From f756ca94901b8179eaa41a3a6f604654cc6d1ba1 Mon Sep 17 00:00:00 2001 From: Jeroen Schweitzer Date: Wed, 24 Dec 2025 16:22:39 +0100 Subject: [PATCH] feat: add maintenance system with index reconciliation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Complete maintenance subsystem for index health and cleanup: Endpoints: - GET /maintenance/health - lightweight (or detailed) health check - POST /maintenance/cleanup/all - full orphan cleanup - POST /maintenance/cleanup/vectors - purge orphan vector chunks - POST /maintenance/cleanup/graph - purge orphan graph nodes - POST /maintenance/reconcile-index - cleanup + reindex missing pages Bidirectional orphan detection: - find_documents_without_vectors() in GraphService - find_chunks_without_graph_nodes() in VectorService Redis integration: - Tracks last_cleanup timestamp for scheduler visibility Config additions: - Document store, volatile cache, and maintenance settings - VectorServiceDep and GraphServiceDep type aliases 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 --- src/config.py | 21 +- src/core/dependencies.py | 14 +- src/main.py | 4 +- src/routers/maintenance.py | 794 +++++++++++++++++++++++++++++++++ src/services/graph_service.py | 350 +++++++++++++++ src/services/vector_service.py | 186 ++++++++ tests/test_maintenance.py | 488 ++++++++++++++++++++ 7 files changed, 1851 insertions(+), 6 deletions(-) create mode 100644 src/routers/maintenance.py create mode 100644 tests/test_maintenance.py diff --git a/src/config.py b/src/config.py index a8d65b5..efb90c1 100644 --- a/src/config.py +++ b/src/config.py @@ -43,8 +43,10 @@ class Settings(BaseSettings): # Wiki.js Configuration wikijs_url: str = Field(default="http://wiki:3000", description="Wiki.js URL") - wikijs_username: str = Field(..., description="Wiki.js username") - wikijs_password: str = Field(..., description="Wiki.js password") + wiki_graphql_api: str = Field(..., description="Wiki.js GraphQL API token (JWT)") + # Legacy auth fields - kept for backwards compatibility but deprecated + wikijs_username: str = Field(default="", description="Wiki.js username (deprecated, use wiki_graphql_api)") + wikijs_password: str = Field(default="", description="Wiki.js password (deprecated, use wiki_graphql_api)") # Wiki.js Database Configuration (for change listener) wikijs_db_host: str = Field(default="postgres-shared", description="Wiki.js PostgreSQL host") @@ -99,6 +101,21 @@ class Settings(BaseSettings): content_extraction_timeout: int = Field(default=5, ge=1, le=30, description="Trafilatura per-URL timeout in seconds") content_max_length: int = Field(default=2000, ge=500, le=10000, description="Max extracted content length per result") + # Document Store Configuration + document_store_enabled: bool = Field(default=True, description="Enable document store feature") + document_catalog_path_prefix: str = Field(default="docs", description="Wiki path prefix for catalog pages") + + # Volatile Cache Configuration + volatile_cache_enabled: bool = Field(default=True, description="Enable volatile cache feature") + volatile_default_ttl: int = Field(default=3600, ge=60, le=86400, description="Default TTL in seconds") + volatile_weather_ttl: int = Field(default=1800, ge=60, le=7200, description="Weather data TTL in seconds") + volatile_news_ttl: int = Field(default=7200, ge=300, le=86400, description="News data TTL in seconds") + volatile_financial_ttl: int = Field(default=300, ge=60, le=3600, description="Financial data TTL in seconds") + + # Maintenance Configuration + maintenance_orphan_cleanup_enabled: bool = Field(default=True, description="Enable automatic orphan cleanup") + maintenance_cleanup_batch_size: int = Field(default=100, ge=10, le=1000, description="Cleanup batch size") + @property def qdrant_url(self) -> str: """Computed Qdrant URL.""" diff --git a/src/core/dependencies.py b/src/core/dependencies.py index 8ecc3dd..b3e6492 100644 --- a/src/core/dependencies.py +++ b/src/core/dependencies.py @@ -76,13 +76,12 @@ def get_wikijs_client() -> WikiJSClient: Get Wiki.js client singleton. Returns: - Initialized Wiki.js GraphQL client with username/password auth + Initialized Wiki.js GraphQL client with API token auth """ settings = get_settings() client = WikiJSClient( base_url=settings.wikijs_url, - username=settings.wikijs_username, - password=settings.wikijs_password + api_token=settings.wiki_graphql_api ) logger.debug("Created Wiki.js client instance") return client @@ -425,3 +424,12 @@ async def verify_api_key( detail="Invalid API key" ) return credentials.credentials + + +# Service type aliases for FastAPI endpoint dependencies +# These are defined after the factory functions +from src.services.vector_service import VectorService +from src.services.graph_service import GraphService + +VectorServiceDep = Annotated[VectorService, Depends(get_vector_service)] +GraphServiceDep = Annotated[GraphService, Depends(get_graph_service)] diff --git a/src/main.py b/src/main.py index ae5ece4..c8d8751 100644 --- a/src/main.py +++ b/src/main.py @@ -50,7 +50,8 @@ app.add_middleware( # Register routers from src.routers import ( wiki, tools, graph, vector, hybrid_rag, consolidation, - ingestion, entity_linking, webhooks, rag_search, content + ingestion, entity_linking, webhooks, rag_search, content, + maintenance ) app.include_router(wiki.router) @@ -64,6 +65,7 @@ app.include_router(entity_linking.router) app.include_router(webhooks.router) app.include_router(rag_search.router) app.include_router(content.router) +app.include_router(maintenance.router) # Mount static files directory for Wiki.js integration scripts static_dir = Path(__file__).parent.parent / "static" diff --git a/src/routers/maintenance.py b/src/routers/maintenance.py new file mode 100644 index 0000000..08f5d2d --- /dev/null +++ b/src/routers/maintenance.py @@ -0,0 +1,794 @@ +""" +Maintenance router for Library Desk cleanup operations. + +Provides endpoints to clean up orphaned data in vectors and graph: +- Orphan vector chunks (no matching page/document in graph) +- Orphan entities (no MENTIONS relationships) +- Stale documents (graph nodes with no matching wiki page) +- Broken relationships +""" + +from fastapi import APIRouter, HTTPException, Depends, Query +from pydantic import BaseModel, Field +from typing import Optional, List, Dict, Any +import logging +import time + +from src.services.vector_service import VectorService +from src.services.graph_service import GraphService +from src.core.dependencies import ( + VectorServiceDep, GraphServiceDep, WikiJSDep, RedisDep, + verify_api_key +) +from datetime import datetime, timezone + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/maintenance", tags=["Maintenance"]) + +# Redis key for tracking last cleanup timestamp +LAST_CLEANUP_KEY = "library:maintenance:last_cleanup:{user}" + + +async def _get_last_cleanup(redis, user: str) -> Optional[str]: + """Get last cleanup timestamp from Redis.""" + try: + key = LAST_CLEANUP_KEY.format(user=user) + return await redis.get(key) + except Exception as e: + logger.warning(f"Failed to get last cleanup timestamp: {e}") + return None + + +async def _set_last_cleanup(redis, user: str) -> None: + """Store current timestamp as last cleanup time.""" + try: + key = LAST_CLEANUP_KEY.format(user=user) + timestamp = datetime.now(timezone.utc).isoformat() + # Keep for 30 days + await redis.setex(key, 86400 * 30, timestamp) + logger.info(f"Recorded cleanup timestamp: {timestamp}") + except Exception as e: + logger.warning(f"Failed to store cleanup timestamp: {e}") + + +async def _find_unindexed_pages( + wiki_pages: List[Dict], + chunk_refs: List[Dict], + graph_docs: List[Dict] +) -> tuple[List[int], List[int]]: + """ + Find wiki pages that are missing from vectors or graph. + + Returns: + Tuple of (pages_without_vectors, pages_without_graph) + """ + # Build sets of indexed page IDs + vectorized_page_ids = { + ref.get("page_id") for ref in chunk_refs + if ref.get("doc_type") == "wiki" and ref.get("page_id") + } + graphed_page_ids = { + doc.get("page_id") for doc in graph_docs + if doc.get("doc_type") == "wiki" and doc.get("page_id") + } + + # Find wiki pages missing from each store + pages_without_vectors = [] + pages_without_graph = [] + + for page in wiki_pages: + page_id = page.get("id") + if not page_id: + continue + + if page_id not in vectorized_page_ids: + pages_without_vectors.append(page_id) + if page_id not in graphed_page_ids: + pages_without_graph.append(page_id) + + return pages_without_vectors, pages_without_graph + + +async def _reindex_missing_pages( + page_ids: List[int], + user: str, + vector_service, + graph_service +) -> tuple[int, int, List[int]]: + """ + Reindex pages that are missing from vectors or graph. + + Returns: + Tuple of (pages_reindexed, pages_failed, failed_page_ids) + """ + reindexed = 0 + failed = 0 + failed_ids = [] + + for page_id in page_ids: + try: + # Index to both stores + vector_result = await vector_service.update_from_page(page_id, user, force_refresh=True) + graph_result = await graph_service.update_from_page(page_id, user, force_refresh=True) + + if vector_result.success and graph_result.success: + reindexed += 1 + logger.info(f"Reindexed missing page {page_id}") + else: + failed += 1 + failed_ids.append(page_id) + logger.warning(f"Failed to reindex page {page_id}: vector={vector_result.success}, graph={graph_result.success}") + + except Exception as e: + failed += 1 + failed_ids.append(page_id) + logger.error(f"Error reindexing page {page_id}: {e}") + + return reindexed, failed, failed_ids + + +# ========== Response Models ========== + +class CleanupResult(BaseModel): + """Result of a cleanup operation.""" + orphans_found: int = Field(default=0, description="Number of orphans detected") + orphans_purged: int = Field(default=0, description="Number of orphans deleted") + duration_ms: float = Field(description="Operation duration in milliseconds") + + +class VectorCleanupResponse(BaseModel): + """Response from vector cleanup operation.""" + success: bool + wiki_chunks: CleanupResult + document_chunks: CleanupResult + chunks_without_graph: CleanupResult # Vectors with no graph node + total_chunks_scanned: int + total_orphans_purged: int + duration_ms: float + + +class GraphCleanupResponse(BaseModel): + """Response from graph cleanup operation.""" + success: bool + orphan_entities: CleanupResult + stale_wiki_documents: CleanupResult + stale_store_documents: CleanupResult + docs_without_vectors: CleanupResult # Graph nodes with no vectors + broken_relationships_cleaned: int + duration_ms: float + + +class FullCleanupResponse(BaseModel): + """Response from full cleanup operation.""" + success: bool + vector_cleanup: VectorCleanupResponse + graph_cleanup: GraphCleanupResponse + total_duration_ms: float + + +class HealthCheckResponse(BaseModel): + """Response from maintenance health check.""" + status: str = Field(description="Health status: healthy, degraded, or unhealthy") + orphan_vector_count: int = Field(description="Number of orphan vector chunks (no source)") + orphan_entity_count: int = Field(description="Number of orphan entities") + stale_document_count: int = Field(description="Number of stale document nodes") + vectors_without_graph: int = Field(default=0, description="Vector chunks with no graph node") + docs_without_vectors: int = Field(default=0, description="Graph docs with no vectors") + unindexed_pages: int = Field(default=0, description="Wiki pages missing from indexes") + last_cleanup: Optional[str] = Field(default=None, description="Timestamp of last cleanup") + recommendations: List[str] = Field(default_factory=list) + + +class ReindexResponse(BaseModel): + """Response from reindex operation.""" + success: bool + page_id: int + vectors_deleted: int + vectors_created: int + graph_updated: bool + duration_ms: float + error: Optional[str] = None + + +class ReindexMissingResult(BaseModel): + """Result of reindexing missing pages.""" + pages_without_vectors: int = Field(description="Wiki pages with no vector embeddings") + pages_without_graph: int = Field(description="Wiki pages with no graph Document node") + pages_reindexed: int = Field(description="Pages successfully reindexed") + pages_failed: int = Field(description="Pages that failed to reindex") + failed_page_ids: List[int] = Field(default_factory=list) + duration_ms: float + + +class ReconcileIndexResponse(BaseModel): + """Response from reconcile-index operation (cleanup + reindex-missing).""" + success: bool + cleanup: FullCleanupResponse + reindex_missing: ReindexMissingResult + total_duration_ms: float + + +# ========== Endpoints ========== + +@router.post("/cleanup/vectors", response_model=VectorCleanupResponse) +async def cleanup_vectors( + user: str = Query(..., description="User identifier"), + dry_run: bool = Query(False, description="If true, only count orphans without deleting"), + vector_service: VectorServiceDep = None, + graph_service: GraphServiceDep = None, + wiki_client: WikiJSDep = None, + api_key: str = Depends(verify_api_key) +): + """ + Find and purge orphan vector chunks. + + Orphan chunks are vector embeddings that reference: + - Wiki pages that no longer exist + - Document Store documents that no longer exist + - Chunks with no corresponding graph Document node (bidirectional check) + + **Scheduler Task** - Recommended to run daily. + """ + start_time = time.time() + + try: + # Get all vector chunk references + chunk_refs = await vector_service.get_all_chunk_references(user) + total_scanned = len(chunk_refs) + + # Get all valid page IDs from wiki + wiki_pages = await wiki_client.list_all_pages() + valid_page_ids = {p.get("id") for p in wiki_pages if p.get("id")} + + # Get all valid document references from graph + graph_docs = await graph_service.get_all_document_references(user) + valid_doc_ids = {d["document_id"] for d in graph_docs if d.get("document_id")} + + # Find orphan wiki chunks (page_id not in wiki) + wiki_orphan_ids = [] + doc_orphan_ids = [] + + for ref in chunk_refs: + doc_type = ref.get("doc_type", "wiki") + + if doc_type == "wiki": + page_id = ref.get("page_id") + if page_id and page_id not in valid_page_ids: + wiki_orphan_ids.append(ref["chunk_id"]) + else: + document_id = ref.get("document_id") + if document_id and document_id not in valid_doc_ids: + doc_orphan_ids.append(ref["chunk_id"]) + + # Bidirectional check: chunks with no graph node + chunks_without_graph = vector_service.find_chunks_without_graph_nodes( + chunk_refs, graph_docs + ) + + # Purge orphans if not dry run + wiki_purged = 0 + doc_purged = 0 + graph_orphans_purged = 0 + + if not dry_run: + if wiki_orphan_ids: + wiki_purged = await vector_service.purge_chunks_by_ids(user, wiki_orphan_ids) + if doc_orphan_ids: + doc_purged = await vector_service.purge_chunks_by_ids(user, doc_orphan_ids) + if chunks_without_graph: + graph_orphans_purged = await vector_service.purge_chunks_by_ids( + user, chunks_without_graph + ) + + duration_ms = (time.time() - start_time) * 1000 + + return VectorCleanupResponse( + success=True, + wiki_chunks=CleanupResult( + orphans_found=len(wiki_orphan_ids), + orphans_purged=wiki_purged, + duration_ms=duration_ms / 3 + ), + document_chunks=CleanupResult( + orphans_found=len(doc_orphan_ids), + orphans_purged=doc_purged, + duration_ms=duration_ms / 3 + ), + chunks_without_graph=CleanupResult( + orphans_found=len(chunks_without_graph), + orphans_purged=graph_orphans_purged, + duration_ms=duration_ms / 3 + ), + total_chunks_scanned=total_scanned, + total_orphans_purged=wiki_purged + doc_purged + graph_orphans_purged, + duration_ms=duration_ms + ) + + except Exception as e: + logger.error(f"Vector cleanup failed: {e}", exc_info=True) + raise HTTPException(status_code=500, detail=str(e)) + + +@router.post("/cleanup/graph", response_model=GraphCleanupResponse) +async def cleanup_graph( + user: str = Query(..., description="User identifier"), + dry_run: bool = Query(False, description="If true, only count orphans without deleting"), + vector_service: VectorServiceDep = None, + graph_service: GraphServiceDep = None, + wiki_client: WikiJSDep = None, + api_key: str = Depends(verify_api_key) +): + """ + Find and purge orphan entities and stale documents from the graph. + + Cleans up: + - Orphan entities (no MENTIONS relationships) + - Stale wiki Document nodes (page deleted from Wiki.js) + - Stale Document Store nodes (document deleted) + - Graph Document nodes with no corresponding vectors (bidirectional check) + - Broken FOUND relationships from SearchQuery nodes + """ + start_time = time.time() + + try: + # 1. Find orphan entities + orphan_entities = await graph_service.find_orphan_entities(user) + entities_purged = 0 + + if not dry_run and orphan_entities: + entities_purged = await graph_service.purge_orphan_entities(user) + + # 2. Find stale wiki documents + graph_docs = await graph_service.get_all_document_references(user) + wiki_docs = [d for d in graph_docs if d.get("doc_type") == "wiki" and d.get("page_id")] + + # Get valid wiki page IDs + wiki_pages = await wiki_client.list_all_pages() + valid_page_ids = {p.get("id") for p in wiki_pages if p.get("id")} + + stale_wiki_ids = [d["page_id"] for d in wiki_docs if d["page_id"] not in valid_page_ids] + wiki_docs_purged = 0 + + if not dry_run and stale_wiki_ids: + wiki_docs_purged = await graph_service.purge_stale_documents_by_ids( + user, page_ids=stale_wiki_ids + ) + + # 3. Find stale Document Store documents (these would be detected differently) + # For now, Document Store docs are only stale if the collection is deleted + # This will be more relevant once DocumentService exists + stale_store_docs = 0 + store_docs_purged = 0 + + # 4. Bidirectional check: graph docs with no vectors + chunk_refs = await vector_service.get_all_chunk_references(user) + docs_without_vectors = await graph_service.find_documents_without_vectors( + user, chunk_refs + ) + docs_without_vectors_purged = 0 + + if not dry_run and docs_without_vectors: + # Purge wiki docs without vectors + wiki_orphans = [d["page_id"] for d in docs_without_vectors + if d.get("doc_type") == "wiki" and d.get("page_id")] + doc_orphans = [d["document_id"] for d in docs_without_vectors + if d.get("doc_type") != "wiki" and d.get("document_id")] + + if wiki_orphans: + docs_without_vectors_purged += await graph_service.purge_stale_documents_by_ids( + user, page_ids=wiki_orphans + ) + if doc_orphans: + docs_without_vectors_purged += await graph_service.purge_stale_documents_by_ids( + user, document_ids=doc_orphans + ) + + # 5. Clean broken relationships + broken_rels_cleaned = 0 + if not dry_run: + broken_rels_cleaned = await graph_service.cleanup_broken_relationships(user) + + duration_ms = (time.time() - start_time) * 1000 + + return GraphCleanupResponse( + success=True, + orphan_entities=CleanupResult( + orphans_found=len(orphan_entities), + orphans_purged=entities_purged, + duration_ms=duration_ms / 5 + ), + stale_wiki_documents=CleanupResult( + orphans_found=len(stale_wiki_ids), + orphans_purged=wiki_docs_purged, + duration_ms=duration_ms / 5 + ), + stale_store_documents=CleanupResult( + orphans_found=stale_store_docs, + orphans_purged=store_docs_purged, + duration_ms=duration_ms / 5 + ), + docs_without_vectors=CleanupResult( + orphans_found=len(docs_without_vectors), + orphans_purged=docs_without_vectors_purged, + duration_ms=duration_ms / 5 + ), + broken_relationships_cleaned=broken_rels_cleaned, + duration_ms=duration_ms + ) + + except Exception as e: + logger.error(f"Graph cleanup failed: {e}", exc_info=True) + raise HTTPException(status_code=500, detail=str(e)) + + +@router.post("/cleanup/all", response_model=FullCleanupResponse) +async def cleanup_all( + user: str = Query(..., description="User identifier"), + dry_run: bool = Query(False, description="If true, only count orphans without deleting"), + vector_service: VectorServiceDep = None, + graph_service: GraphServiceDep = None, + wiki_client: WikiJSDep = None, + redis: RedisDep = None, + api_key: str = Depends(verify_api_key) +): + """ + Full cleanup of vectors and graph. + + Runs both vector and graph cleanup in sequence. + + **Scheduler Task** - Recommended to run daily at low-traffic time. + + **Scheduler Integration:** + ```json + { + "task_name": "library_maintenance", + "schedule": "0 4 * * *", + "endpoint": "POST /maintenance/cleanup/all?user=jpmschweitzer", + "description": "Daily cleanup of orphan vectors and graph nodes" + } + ``` + """ + start_time = time.time() + + try: + # Run vector cleanup + vector_result = await cleanup_vectors( + user=user, + dry_run=dry_run, + vector_service=vector_service, + graph_service=graph_service, + wiki_client=wiki_client, + api_key=api_key + ) + + # Run graph cleanup + graph_result = await cleanup_graph( + user=user, + dry_run=dry_run, + vector_service=vector_service, + graph_service=graph_service, + wiki_client=wiki_client, + api_key=api_key + ) + + total_duration_ms = (time.time() - start_time) * 1000 + + # Record cleanup timestamp (only if not dry run) + if not dry_run and redis: + await _set_last_cleanup(redis, user) + + return FullCleanupResponse( + success=True, + vector_cleanup=vector_result, + graph_cleanup=graph_result, + total_duration_ms=total_duration_ms + ) + + except Exception as e: + logger.error(f"Full cleanup failed: {e}", exc_info=True) + raise HTTPException(status_code=500, detail=str(e)) + + +@router.get("/health", response_model=HealthCheckResponse) +async def maintenance_health( + user: str = Query(..., description="User identifier"), + detailed: bool = Query(False, description="If true, run full orphan analysis (slower)"), + vector_service: VectorServiceDep = None, + graph_service: GraphServiceDep = None, + wiki_client: WikiJSDep = None, + redis: RedisDep = None, + api_key: str = Depends(verify_api_key) +): + """ + Health check for maintenance status. + + **Lightweight mode (default)**: Returns last cleanup timestamp and basic status. + Use for frequent uptime checks (every 30s). + + **Detailed mode (?detailed=true)**: Runs full orphan/unindexed analysis. + Use for dashboards or before running reconcile-index. + """ + try: + # Get last cleanup timestamp from Redis (lightweight) + last_cleanup = None + if redis: + last_cleanup = await _get_last_cleanup(redis, user) + + # Lightweight mode - just return basic status + if not detailed: + return HealthCheckResponse( + status="healthy" if last_cleanup else "unknown", + orphan_vector_count=0, + orphan_entity_count=0, + stale_document_count=0, + vectors_without_graph=0, + docs_without_vectors=0, + unindexed_pages=0, + last_cleanup=last_cleanup, + recommendations=[] if last_cleanup else ["No cleanup recorded. Run POST /maintenance/reconcile-index"] + ) + + # Detailed mode - full analysis + recommendations = [] + + # Count orphan vector chunks + chunk_refs = await vector_service.get_all_chunk_references(user) + wiki_pages = await wiki_client.list_all_pages() + valid_page_ids = {p.get("id") for p in wiki_pages if p.get("id")} + + orphan_vector_count = sum( + 1 for ref in chunk_refs + if ref.get("doc_type") == "wiki" + and ref.get("page_id") not in valid_page_ids + ) + + if orphan_vector_count > 10: + recommendations.append( + f"Found {orphan_vector_count} orphan vector chunks. " + "Consider running POST /maintenance/cleanup/vectors" + ) + + # Count orphan entities + orphan_entities = await graph_service.find_orphan_entities(user) + orphan_entity_count = len(orphan_entities) + + if orphan_entity_count > 5: + recommendations.append( + f"Found {orphan_entity_count} orphan entities. " + "Consider running POST /maintenance/cleanup/graph" + ) + + # Count stale documents + graph_docs = await graph_service.get_all_document_references(user) + wiki_docs = [d for d in graph_docs if d.get("doc_type") == "wiki" and d.get("page_id")] + stale_document_count = sum(1 for d in wiki_docs if d["page_id"] not in valid_page_ids) + + if stale_document_count > 0: + recommendations.append( + f"Found {stale_document_count} stale Document nodes. " + "Consider running POST /maintenance/cleanup/graph" + ) + + # Bidirectional: vectors without graph nodes + vectors_without_graph = len(vector_service.find_chunks_without_graph_nodes( + chunk_refs, graph_docs + )) + + if vectors_without_graph > 5: + recommendations.append( + f"Found {vectors_without_graph} vectors without graph nodes. " + "Consider running POST /maintenance/cleanup/vectors" + ) + + # Bidirectional: graph docs without vectors + docs_without_vectors_list = await graph_service.find_documents_without_vectors( + user, chunk_refs + ) + docs_without_vectors = len(docs_without_vectors_list) + + if docs_without_vectors > 5: + recommendations.append( + f"Found {docs_without_vectors} graph docs without vectors. " + "Consider running POST /maintenance/cleanup/graph" + ) + + # Unindexed pages: wiki pages missing from vectors or graph + pages_without_vectors, pages_without_graph = await _find_unindexed_pages( + wiki_pages, chunk_refs, graph_docs + ) + unindexed_pages = len(set(pages_without_vectors + pages_without_graph)) + + if unindexed_pages > 0: + recommendations.append( + f"Found {unindexed_pages} wiki pages not in indexes. " + "Consider running POST /maintenance/reconcile-index" + ) + + # Determine overall status + total_issues = (orphan_vector_count + orphan_entity_count + stale_document_count + + vectors_without_graph + docs_without_vectors + unindexed_pages) + if total_issues == 0: + status = "healthy" + elif total_issues < 20: + status = "degraded" + else: + status = "unhealthy" + + # Get last cleanup timestamp from Redis + last_cleanup = None + if redis: + last_cleanup = await _get_last_cleanup(redis, user) + + return HealthCheckResponse( + status=status, + orphan_vector_count=orphan_vector_count, + orphan_entity_count=orphan_entity_count, + stale_document_count=stale_document_count, + vectors_without_graph=vectors_without_graph, + docs_without_vectors=docs_without_vectors, + unindexed_pages=unindexed_pages, + last_cleanup=last_cleanup, + recommendations=recommendations + ) + + except Exception as e: + logger.error(f"Health check failed: {e}", exc_info=True) + return HealthCheckResponse( + status="unhealthy", + orphan_vector_count=-1, + orphan_entity_count=-1, + stale_document_count=-1, + vectors_without_graph=-1, + docs_without_vectors=-1, + unindexed_pages=-1, + recommendations=[f"Health check failed: {str(e)}"] + ) + + +@router.post("/reindex/{page_id}", response_model=ReindexResponse) +async def reindex_page( + page_id: int, + user: str = Query(..., description="User identifier"), + vector_service: VectorServiceDep = None, + graph_service: GraphServiceDep = None, + api_key: str = Depends(verify_api_key) +): + """ + Force re-index a wiki page. + + Deletes existing vectors and graph data, then re-ingests. + Useful for fixing corrupted or stale data for a specific page. + """ + start_time = time.time() + + try: + # Delete existing vectors + vectors_deleted = await vector_service.delete_page_chunks(page_id, user) + + # Delete and recreate graph node + await graph_service.delete_page(page_id, user) + + # Re-ingest + vector_result = await vector_service.update_from_page(page_id, user, force_refresh=True) + graph_result = await graph_service.update_from_page(page_id, user, force_refresh=True) + + duration_ms = (time.time() - start_time) * 1000 + + return ReindexResponse( + success=vector_result.success and graph_result.success, + page_id=page_id, + vectors_deleted=vectors_deleted, + vectors_created=vector_result.chunks_created, + graph_updated=graph_result.success, + duration_ms=duration_ms, + error=vector_result.error_message or graph_result.error_message + ) + + except Exception as e: + duration_ms = (time.time() - start_time) * 1000 + logger.error(f"Reindex failed for page {page_id}: {e}", exc_info=True) + return ReindexResponse( + success=False, + page_id=page_id, + vectors_deleted=0, + vectors_created=0, + graph_updated=False, + duration_ms=duration_ms, + error=str(e) + ) + + +@router.post("/reconcile-index", response_model=ReconcileIndexResponse) +async def reconcile_index( + user: str = Query(..., description="User identifier"), + dry_run: bool = Query(False, description="If true, only detect issues without fixing"), + vector_service: VectorServiceDep = None, + graph_service: GraphServiceDep = None, + wiki_client: WikiJSDep = None, + redis: RedisDep = None, + api_key: str = Depends(verify_api_key) +): + """ + Full index reconciliation: cleanup orphans + reindex missing pages. + + This is the recommended daily maintenance endpoint. It: + 1. Cleans up orphan vectors and graph nodes (data without sources) + 2. Reindexes wiki pages that are missing from vectors or graph + + **Scheduler Task** - Recommended to run daily at low-traffic time. + + **Scheduler Integration:** + ```json + { + "task_name": "library_reconcile_index", + "schedule": "0 4 * * *", + "endpoint": "POST /maintenance/reconcile-index?user=jpmschweitzer", + "description": "Daily index reconciliation - cleanup + reindex missing" + } + ``` + """ + start_time = time.time() + + try: + # Phase 1: Run full cleanup + cleanup_result = await cleanup_all( + user=user, + dry_run=dry_run, + vector_service=vector_service, + graph_service=graph_service, + wiki_client=wiki_client, + redis=redis, + api_key=api_key + ) + + # Phase 2: Find and reindex missing pages + reindex_start = time.time() + + # Get current state + wiki_pages = await wiki_client.list_all_pages() + chunk_refs = await vector_service.get_all_chunk_references(user) + graph_docs = await graph_service.get_all_document_references(user) + + # Find pages missing from indexes + pages_without_vectors, pages_without_graph = await _find_unindexed_pages( + wiki_pages, chunk_refs, graph_docs + ) + + # Combine unique page IDs that need reindexing + missing_page_ids = list(set(pages_without_vectors + pages_without_graph)) + + # Reindex missing pages (unless dry run) + reindexed = 0 + failed = 0 + failed_ids = [] + + if not dry_run and missing_page_ids: + reindexed, failed, failed_ids = await _reindex_missing_pages( + missing_page_ids, user, vector_service, graph_service + ) + + reindex_duration = (time.time() - reindex_start) * 1000 + total_duration = (time.time() - start_time) * 1000 + + # Record reconciliation timestamp + if not dry_run and redis: + await _set_last_cleanup(redis, user) + + return ReconcileIndexResponse( + success=cleanup_result.success and failed == 0, + cleanup=cleanup_result, + reindex_missing=ReindexMissingResult( + pages_without_vectors=len(pages_without_vectors), + pages_without_graph=len(pages_without_graph), + pages_reindexed=reindexed, + pages_failed=failed, + failed_page_ids=failed_ids, + duration_ms=reindex_duration + ), + total_duration_ms=total_duration + ) + + except Exception as e: + logger.error(f"Reconcile-index failed: {e}", exc_info=True) + raise HTTPException(status_code=500, detail=str(e)) diff --git a/src/services/graph_service.py b/src/services/graph_service.py index 34f8bc2..15378c6 100644 --- a/src/services/graph_service.py +++ b/src/services/graph_service.py @@ -1261,3 +1261,353 @@ Feel free to expand it with more details! except Exception as e: logger.error(f"Failed to create entity mentions: {e}", exc_info=True) return 0 + + # ========== Cleanup Methods ========== + + async def delete_document_node( + self, + document_id: str, + user: str + ) -> int: + """ + Delete a Document Store document node and all its relationships. + + Args: + document_id: Document UUID (Document Store) + user: User identifier + + Returns: + Number of nodes deleted (1 if successful, 0 if not found) + """ + user_doc_label = get_neo4j_user_label(user) + + delete_query = f""" + MATCH (d:{user_doc_label}:Document {{document_id: $document_id}}) + DETACH DELETE d + RETURN count(d) as deleted_count + """ + + try: + result = await self.neo4j.execute_query( + delete_query, + {"document_id": document_id} + ) + + deleted_count = result[0]["deleted_count"] if result else 0 + + if deleted_count > 0: + logger.info(f"Deleted Document node for document {document_id}") + else: + logger.warning(f"No Document node found for document {document_id}") + + return deleted_count + + except Exception as e: + logger.error(f"Failed to delete document {document_id} from graph: {e}", exc_info=True) + return 0 + + async def delete_collection_node( + self, + collection_id: str, + user: str + ) -> int: + """ + Delete a DocumentCollection node and all contained documents. + + Args: + collection_id: Collection UUID + user: User identifier + + Returns: + Number of nodes deleted (collection + documents) + """ + user_doc_label = get_neo4j_user_label(user) + + # Delete collection and all documents it contains + delete_query = f""" + MATCH (c:{user_doc_label}:DocumentCollection {{id: $collection_id}}) + OPTIONAL MATCH (c)-[:CONTAINS]->(d:Document) + DETACH DELETE c, d + RETURN count(c) + count(d) as deleted_count + """ + + try: + result = await self.neo4j.execute_query( + delete_query, + {"collection_id": collection_id} + ) + + deleted_count = result[0]["deleted_count"] if result else 0 + logger.info(f"Deleted collection {collection_id} with {deleted_count} total nodes") + return deleted_count + + except Exception as e: + logger.error(f"Failed to delete collection {collection_id}: {e}", exc_info=True) + return 0 + + async def find_orphan_entities( + self, + user: str + ) -> List[Dict[str, Any]]: + """ + Find entities with no MENTIONS relationships (orphaned). + + Args: + user: User identifier + + Returns: + List of orphaned entities {id, name, type} + """ + from src.core.multi_tenancy import get_neo4j_user_base_label + + user_base_label = get_neo4j_user_base_label(user) + + query = f""" + MATCH (e:{user_base_label}) + WHERE NOT e:Document + AND NOT e:DocumentCollection + AND NOT EXISTS {{ (d:Document)-[:MENTIONS]->(e) }} + RETURN elementId(e) as id, e.name as name, labels(e) as labels + """ + + try: + results = await self.neo4j.execute_query(query, {}) + + orphans = [] + for r in results: + labels = r.get("labels", []) + entity_type = next( + (l for l in labels if l != user_base_label), + "Unknown" + ) + orphans.append({ + "id": r["id"], + "name": r["name"], + "type": entity_type + }) + + logger.info(f"Found {len(orphans)} orphan entities for user {user}") + return orphans + + except Exception as e: + logger.error(f"Failed to find orphan entities: {e}", exc_info=True) + return [] + + async def purge_orphan_entities( + self, + user: str + ) -> int: + """ + Delete all orphaned entities (entities with no MENTIONS relationships). + + Args: + user: User identifier + + Returns: + Number of entities purged + """ + from src.core.multi_tenancy import get_neo4j_user_base_label + + user_base_label = get_neo4j_user_base_label(user) + + query = f""" + MATCH (e:{user_base_label}) + WHERE NOT e:Document + AND NOT e:DocumentCollection + AND NOT EXISTS {{ (d:Document)-[:MENTIONS]->(e) }} + DETACH DELETE e + RETURN count(e) as purged_count + """ + + try: + results = await self.neo4j.execute_query(query, {}) + purged_count = results[0]["purged_count"] if results else 0 + + logger.info(f"Purged {purged_count} orphan entities for user {user}") + return purged_count + + except Exception as e: + logger.error(f"Failed to purge orphan entities: {e}", exc_info=True) + return 0 + + async def get_all_document_references( + self, + user: str + ) -> List[Dict[str, Any]]: + """ + Get all Document node references for orphan detection. + + Returns page_id for wiki docs and document_id for Document Store docs. + + Args: + user: User identifier + + Returns: + List of document references {page_id, document_id, doc_type, title} + """ + user_doc_label = get_neo4j_user_label(user) + + query = f""" + MATCH (d:{user_doc_label}:Document) + RETURN d.page_id as page_id, + d.document_id as document_id, + COALESCE(d.doc_type, 'wiki') as doc_type, + d.title as title + """ + + try: + results = await self.neo4j.execute_query(query, {}) + + references = [] + for r in results: + references.append({ + "page_id": r.get("page_id"), + "document_id": r.get("document_id"), + "doc_type": r.get("doc_type", "wiki"), + "title": r.get("title") + }) + + logger.info(f"Found {len(references)} document references for user {user}") + return references + + except Exception as e: + logger.error(f"Failed to get document references: {e}", exc_info=True) + return [] + + async def purge_stale_documents_by_ids( + self, + user: str, + page_ids: List[int] = None, + document_ids: List[str] = None + ) -> int: + """ + Delete specific stale Document nodes by their IDs. + + Args: + user: User identifier + page_ids: List of wiki page IDs to delete + document_ids: List of Document Store document IDs to delete + + Returns: + Number of documents purged + """ + user_doc_label = get_neo4j_user_label(user) + total_purged = 0 + + try: + # Purge by page_id (wiki docs) + if page_ids: + query = f""" + MATCH (d:{user_doc_label}:Document) + WHERE d.page_id IN $page_ids + DETACH DELETE d + RETURN count(d) as purged_count + """ + results = await self.neo4j.execute_query(query, {"page_ids": page_ids}) + count = results[0]["purged_count"] if results else 0 + total_purged += count + logger.info(f"Purged {count} wiki Document nodes") + + # Purge by document_id (Document Store docs) + if document_ids: + query = f""" + MATCH (d:{user_doc_label}:Document) + WHERE d.document_id IN $document_ids + DETACH DELETE d + RETURN count(d) as purged_count + """ + results = await self.neo4j.execute_query(query, {"document_ids": document_ids}) + count = results[0]["purged_count"] if results else 0 + total_purged += count + logger.info(f"Purged {count} Document Store Document nodes") + + return total_purged + + except Exception as e: + logger.error(f"Failed to purge stale documents: {e}", exc_info=True) + return 0 + + async def cleanup_broken_relationships( + self, + user: str + ) -> int: + """ + Clean up broken FOUND relationships from SearchQuery nodes. + + Removes relationships pointing to deleted documents. + + Args: + user: User identifier + + Returns: + Number of relationships cleaned + """ + query = """ + MATCH (sq:SearchQuery)-[r:FOUND]->(d) + WHERE NOT EXISTS { (d) } + DELETE r + RETURN count(r) as cleaned_count + """ + + try: + results = await self.neo4j.execute_query(query, {}) + cleaned_count = results[0]["cleaned_count"] if results else 0 + + if cleaned_count > 0: + logger.info(f"Cleaned {cleaned_count} broken FOUND relationships") + + return cleaned_count + + except Exception as e: + logger.error(f"Failed to cleanup broken relationships: {e}", exc_info=True) + return 0 + + async def find_documents_without_vectors( + self, + user: str, + vector_references: List[Dict[str, Any]] + ) -> List[Dict[str, Any]]: + """ + Find Document nodes that have no corresponding vectors. + + Used for bidirectional orphan detection - graph nodes without vector data. + + Args: + user: User identifier + vector_references: List of vector refs from VectorService.get_all_chunk_references() + + Returns: + List of orphan documents {page_id, document_id, doc_type, title} + """ + # Get all graph document references + graph_docs = await self.get_all_document_references(user) + + if not graph_docs: + return [] + + # Build sets of IDs that have vectors + vector_page_ids = { + ref.get("page_id") for ref in vector_references + if ref.get("doc_type") == "wiki" and ref.get("page_id") + } + vector_doc_ids = { + ref.get("document_id") for ref in vector_references + if ref.get("doc_type") != "wiki" and ref.get("document_id") + } + + # Find graph docs with no vectors + orphans = [] + for doc in graph_docs: + doc_type = doc.get("doc_type", "wiki") + + if doc_type == "wiki": + page_id = doc.get("page_id") + if page_id and page_id not in vector_page_ids: + orphans.append(doc) + else: + document_id = doc.get("document_id") + if document_id and document_id not in vector_doc_ids: + orphans.append(doc) + + logger.info(f"Found {len(orphans)} graph documents without vectors for user {user}") + return orphans diff --git a/src/services/vector_service.py b/src/services/vector_service.py index 475b562..f7413b1 100644 --- a/src/services/vector_service.py +++ b/src/services/vector_service.py @@ -356,3 +356,189 @@ class VectorService: collections=[], total=0 ) + + # ========== Cleanup Methods ========== + + async def delete_document_chunks( + self, + document_id: str, + user: str + ) -> int: + """ + Delete all chunks for a document (Document Store). + + Args: + document_id: Document UUID + user: User identifier + + Returns: + Number of chunks deleted + """ + collection_name = get_qdrant_collection_name(user) + + try: + deleted_count = await self.qdrant.delete_by_filter( + collection_name=collection_name, + filter_conditions={"document_id": document_id} + ) + + logger.info(f"Deleted chunks for document {document_id}") + return deleted_count + + except Exception as e: + logger.error(f"Failed to delete chunks for document {document_id}: {e}", exc_info=True) + return 0 + + async def delete_collection_chunks( + self, + collection_id: str, + user: str + ) -> int: + """ + Delete all chunks for a document collection. + + Args: + collection_id: Collection UUID + user: User identifier + + Returns: + Number of chunks deleted + """ + collection_name = get_qdrant_collection_name(user) + + try: + deleted_count = await self.qdrant.delete_by_filter( + collection_name=collection_name, + filter_conditions={"collection_id": collection_id} + ) + + logger.info(f"Deleted chunks for collection {collection_id}") + return deleted_count + + except Exception as e: + logger.error(f"Failed to delete chunks for collection {collection_id}: {e}", exc_info=True) + return 0 + + async def get_all_chunk_references( + self, + user: str + ) -> List[Dict[str, Any]]: + """ + Get all chunk references for orphan detection. + + Returns list of {id, page_id, document_id} for all chunks. + + Args: + user: User identifier + + Returns: + List of chunk references + """ + collection_name = get_qdrant_collection_name(user) + + try: + # Check if collection exists + exists = await self.qdrant.collection_exists(collection_name) + if not exists: + return [] + + all_points = await self.qdrant.scroll_all_points( + collection_name=collection_name, + batch_size=100, + with_payload=True + ) + + references = [] + for point in all_points: + payload = point.get("payload", {}) + references.append({ + "chunk_id": point["id"], + "page_id": payload.get("page_id"), + "document_id": payload.get("document_id"), + "collection_id": payload.get("collection_id"), + "doc_type": payload.get("doc_type", "wiki") + }) + + logger.info(f"Found {len(references)} chunks for user {user}") + return references + + except Exception as e: + logger.error(f"Failed to get chunk references: {e}", exc_info=True) + return [] + + async def purge_chunks_by_ids( + self, + user: str, + chunk_ids: List[str] + ) -> int: + """ + Delete specific chunks by their IDs. + + Args: + user: User identifier + chunk_ids: List of chunk IDs to delete + + Returns: + Number of chunks deleted + """ + if not chunk_ids: + return 0 + + collection_name = get_qdrant_collection_name(user) + + try: + deleted_count = await self.qdrant.delete_by_ids( + collection_name=collection_name, + point_ids=chunk_ids + ) + + logger.info(f"Purged {deleted_count} orphan chunks for user {user}") + return deleted_count + + except Exception as e: + logger.error(f"Failed to purge chunks: {e}", exc_info=True) + return 0 + + def find_chunks_without_graph_nodes( + self, + chunk_references: List[Dict[str, Any]], + graph_references: List[Dict[str, Any]] + ) -> List[str]: + """ + Find vector chunks that have no corresponding graph Document node. + + Used for bidirectional orphan detection - vectors without graph representation. + + Args: + chunk_references: List from get_all_chunk_references() + graph_references: List from GraphService.get_all_document_references() + + Returns: + List of orphan chunk IDs + """ + # Build sets of IDs that have graph nodes + graph_page_ids = { + ref.get("page_id") for ref in graph_references + if ref.get("doc_type") == "wiki" and ref.get("page_id") + } + graph_doc_ids = { + ref.get("document_id") for ref in graph_references + if ref.get("doc_type") != "wiki" and ref.get("document_id") + } + + # Find chunks with no graph node + orphan_ids = [] + for chunk in chunk_references: + doc_type = chunk.get("doc_type", "wiki") + + if doc_type == "wiki": + page_id = chunk.get("page_id") + if page_id and page_id not in graph_page_ids: + orphan_ids.append(chunk["chunk_id"]) + else: + document_id = chunk.get("document_id") + if document_id and document_id not in graph_doc_ids: + orphan_ids.append(chunk["chunk_id"]) + + logger.info(f"Found {len(orphan_ids)} vector chunks without graph nodes") + return orphan_ids diff --git a/tests/test_maintenance.py b/tests/test_maintenance.py new file mode 100644 index 0000000..95cac07 --- /dev/null +++ b/tests/test_maintenance.py @@ -0,0 +1,488 @@ +""" +Tests for maintenance router and cleanup functionality. + +Tests cleanup of: +- Orphan vector chunks +- Orphan entities in graph +- Stale document nodes +""" + +import pytest +from unittest.mock import AsyncMock, MagicMock, patch +from src.routers.maintenance import ( + cleanup_vectors, + cleanup_graph, + cleanup_all, + maintenance_health, + reindex_page, + CleanupResult, + VectorCleanupResponse, + GraphCleanupResponse, + FullCleanupResponse, + HealthCheckResponse, + ReindexResponse +) + + +class TestCleanupResult: + """Test CleanupResult model.""" + + def test_cleanup_result_defaults(self): + """Test CleanupResult with default values.""" + result = CleanupResult(duration_ms=100.0) + assert result.orphans_found == 0 + assert result.orphans_purged == 0 + assert result.duration_ms == 100.0 + + def test_cleanup_result_with_values(self): + """Test CleanupResult with actual values.""" + result = CleanupResult( + orphans_found=10, + orphans_purged=8, + duration_ms=250.5 + ) + assert result.orphans_found == 10 + assert result.orphans_purged == 8 + assert result.duration_ms == 250.5 + + +class TestVectorCleanupResponse: + """Test VectorCleanupResponse model.""" + + def test_vector_cleanup_response(self): + """Test VectorCleanupResponse structure.""" + response = VectorCleanupResponse( + success=True, + wiki_chunks=CleanupResult(orphans_found=5, orphans_purged=5, duration_ms=50), + document_chunks=CleanupResult(orphans_found=3, orphans_purged=3, duration_ms=50), + chunks_without_graph=CleanupResult(orphans_found=2, orphans_purged=2, duration_ms=50), + total_chunks_scanned=100, + total_orphans_purged=10, + duration_ms=100 + ) + assert response.success is True + assert response.wiki_chunks.orphans_found == 5 + assert response.document_chunks.orphans_found == 3 + assert response.chunks_without_graph.orphans_found == 2 + assert response.total_orphans_purged == 10 + + +class TestGraphCleanupResponse: + """Test GraphCleanupResponse model.""" + + def test_graph_cleanup_response(self): + """Test GraphCleanupResponse structure.""" + response = GraphCleanupResponse( + success=True, + orphan_entities=CleanupResult(orphans_found=10, orphans_purged=10, duration_ms=25), + stale_wiki_documents=CleanupResult(orphans_found=2, orphans_purged=2, duration_ms=25), + stale_store_documents=CleanupResult(orphans_found=0, orphans_purged=0, duration_ms=25), + docs_without_vectors=CleanupResult(orphans_found=1, orphans_purged=1, duration_ms=25), + broken_relationships_cleaned=5, + duration_ms=100 + ) + assert response.success is True + assert response.orphan_entities.orphans_found == 10 + assert response.docs_without_vectors.orphans_found == 1 + assert response.broken_relationships_cleaned == 5 + + +class TestHealthCheckResponse: + """Test HealthCheckResponse model.""" + + def test_health_check_healthy(self): + """Test healthy status.""" + response = HealthCheckResponse( + status="healthy", + orphan_vector_count=0, + orphan_entity_count=0, + stale_document_count=0 + ) + assert response.status == "healthy" + assert response.recommendations == [] + + def test_health_check_degraded(self): + """Test degraded status with recommendations.""" + response = HealthCheckResponse( + status="degraded", + orphan_vector_count=15, + orphan_entity_count=3, + stale_document_count=0, + recommendations=[ + "Found 15 orphan vector chunks. Consider running POST /maintenance/cleanup/vectors" + ] + ) + assert response.status == "degraded" + assert len(response.recommendations) == 1 + + +@pytest.mark.asyncio +class TestVectorCleanup: + """Test vector cleanup endpoint.""" + + async def test_cleanup_vectors_no_orphans(self): + """Test cleanup when no orphans exist.""" + # Mock services + vector_service = AsyncMock() + vector_service.get_all_chunk_references.return_value = [ + {"chunk_id": "c1", "page_id": 1, "doc_type": "wiki"} + ] + # find_chunks_without_graph_nodes is not async + vector_service.find_chunks_without_graph_nodes = MagicMock(return_value=[]) + + graph_service = AsyncMock() + graph_service.get_all_document_references.return_value = [ + {"page_id": 1, "doc_type": "wiki", "title": "Test"} + ] + + wiki_client = AsyncMock() + wiki_client.list_all_pages.return_value = [{"id": 1, "path": "test"}] + + # Call cleanup + result = await cleanup_vectors( + user="testuser", + dry_run=False, + vector_service=vector_service, + graph_service=graph_service, + wiki_client=wiki_client, + api_key="test" + ) + + assert result.success is True + assert result.wiki_chunks.orphans_found == 0 + assert result.chunks_without_graph.orphans_found == 0 + assert result.total_orphans_purged == 0 + + async def test_cleanup_vectors_with_orphans(self): + """Test cleanup when orphans exist.""" + # Mock services + vector_service = AsyncMock() + vector_service.get_all_chunk_references.return_value = [ + {"chunk_id": "c1", "page_id": 1, "doc_type": "wiki"}, + {"chunk_id": "c2", "page_id": 999, "doc_type": "wiki"}, # Orphan + {"chunk_id": "c3", "page_id": 999, "doc_type": "wiki"}, # Orphan + ] + vector_service.purge_chunks_by_ids.return_value = 2 + # find_chunks_without_graph_nodes is not async + vector_service.find_chunks_without_graph_nodes = MagicMock(return_value=[]) + + graph_service = AsyncMock() + graph_service.get_all_document_references.return_value = [ + {"page_id": 1, "doc_type": "wiki", "title": "Test"} + ] + + wiki_client = AsyncMock() + wiki_client.list_all_pages.return_value = [{"id": 1, "path": "test"}] + + # Call cleanup + result = await cleanup_vectors( + user="testuser", + dry_run=False, + vector_service=vector_service, + graph_service=graph_service, + wiki_client=wiki_client, + api_key="test" + ) + + assert result.success is True + assert result.wiki_chunks.orphans_found == 2 + assert result.wiki_chunks.orphans_purged == 2 + assert result.total_orphans_purged == 2 + + async def test_cleanup_vectors_dry_run(self): + """Test cleanup dry run doesn't purge.""" + # Mock services + vector_service = AsyncMock() + vector_service.get_all_chunk_references.return_value = [ + {"chunk_id": "c1", "page_id": 999, "doc_type": "wiki"}, # Orphan + ] + # find_chunks_without_graph_nodes is not async + vector_service.find_chunks_without_graph_nodes = MagicMock(return_value=[]) + + graph_service = AsyncMock() + graph_service.get_all_document_references.return_value = [] + + wiki_client = AsyncMock() + wiki_client.list_all_pages.return_value = [] + + # Call cleanup in dry run mode + result = await cleanup_vectors( + user="testuser", + dry_run=True, + vector_service=vector_service, + graph_service=graph_service, + wiki_client=wiki_client, + api_key="test" + ) + + assert result.success is True + assert result.wiki_chunks.orphans_found == 1 + assert result.wiki_chunks.orphans_purged == 0 # Not purged due to dry run + vector_service.purge_chunks_by_ids.assert_not_called() + + +@pytest.mark.asyncio +class TestGraphCleanup: + """Test graph cleanup endpoint.""" + + async def test_cleanup_graph_no_orphans(self): + """Test cleanup when no orphans exist.""" + vector_service = AsyncMock() + vector_service.get_all_chunk_references.return_value = [ + {"chunk_id": "c1", "page_id": 1, "doc_type": "wiki"} + ] + + graph_service = AsyncMock() + graph_service.find_orphan_entities.return_value = [] + graph_service.get_all_document_references.return_value = [ + {"page_id": 1, "doc_type": "wiki", "title": "Test"} + ] + graph_service.find_documents_without_vectors.return_value = [] + graph_service.cleanup_broken_relationships.return_value = 0 + + wiki_client = AsyncMock() + wiki_client.list_all_pages.return_value = [{"id": 1, "path": "test"}] + + result = await cleanup_graph( + user="testuser", + dry_run=False, + vector_service=vector_service, + graph_service=graph_service, + wiki_client=wiki_client, + api_key="test" + ) + + assert result.success is True + assert result.orphan_entities.orphans_found == 0 + assert result.stale_wiki_documents.orphans_found == 0 + assert result.docs_without_vectors.orphans_found == 0 + + async def test_cleanup_graph_with_orphan_entities(self): + """Test cleanup of orphan entities.""" + vector_service = AsyncMock() + vector_service.get_all_chunk_references.return_value = [] + + graph_service = AsyncMock() + graph_service.find_orphan_entities.return_value = [ + {"id": "e1", "name": "Orphan1", "type": "Person"}, + {"id": "e2", "name": "Orphan2", "type": "Technology"}, + ] + graph_service.purge_orphan_entities.return_value = 2 + graph_service.get_all_document_references.return_value = [] + graph_service.find_documents_without_vectors.return_value = [] + graph_service.cleanup_broken_relationships.return_value = 0 + + wiki_client = AsyncMock() + wiki_client.list_all_pages.return_value = [] + + result = await cleanup_graph( + user="testuser", + dry_run=False, + vector_service=vector_service, + graph_service=graph_service, + wiki_client=wiki_client, + api_key="test" + ) + + assert result.success is True + assert result.orphan_entities.orphans_found == 2 + assert result.orphan_entities.orphans_purged == 2 + + async def test_cleanup_graph_with_stale_documents(self): + """Test cleanup of stale document nodes.""" + vector_service = AsyncMock() + vector_service.get_all_chunk_references.return_value = [ + {"chunk_id": "c1", "page_id": 1, "doc_type": "wiki"} + ] + + graph_service = AsyncMock() + graph_service.find_orphan_entities.return_value = [] + graph_service.get_all_document_references.return_value = [ + {"page_id": 1, "doc_type": "wiki", "title": "Exists"}, + {"page_id": 999, "doc_type": "wiki", "title": "Deleted"}, # Stale + ] + graph_service.find_documents_without_vectors.return_value = [] + graph_service.purge_stale_documents_by_ids.return_value = 1 + graph_service.cleanup_broken_relationships.return_value = 0 + + wiki_client = AsyncMock() + wiki_client.list_all_pages.return_value = [{"id": 1, "path": "test"}] + + result = await cleanup_graph( + user="testuser", + dry_run=False, + vector_service=vector_service, + graph_service=graph_service, + wiki_client=wiki_client, + api_key="test" + ) + + assert result.success is True + assert result.stale_wiki_documents.orphans_found == 1 + assert result.stale_wiki_documents.orphans_purged == 1 + + +@pytest.mark.asyncio +class TestFullCleanup: + """Test full cleanup endpoint.""" + + async def test_full_cleanup(self): + """Test full cleanup runs both vector and graph cleanup.""" + vector_service = AsyncMock() + vector_service.get_all_chunk_references.return_value = [] + # find_chunks_without_graph_nodes is not async + vector_service.find_chunks_without_graph_nodes = MagicMock(return_value=[]) + + graph_service = AsyncMock() + graph_service.find_orphan_entities.return_value = [] + graph_service.get_all_document_references.return_value = [] + graph_service.find_documents_without_vectors.return_value = [] + graph_service.cleanup_broken_relationships.return_value = 0 + + wiki_client = AsyncMock() + wiki_client.list_all_pages.return_value = [] + + result = await cleanup_all( + user="testuser", + dry_run=False, + vector_service=vector_service, + graph_service=graph_service, + wiki_client=wiki_client, + api_key="test" + ) + + assert result.success is True + assert result.vector_cleanup.success is True + assert result.graph_cleanup.success is True + + +@pytest.mark.asyncio +class TestMaintenanceHealth: + """Test maintenance health endpoint.""" + + async def test_health_healthy(self): + """Test healthy status when no orphans.""" + vector_service = AsyncMock() + vector_service.get_all_chunk_references.return_value = [] + # find_chunks_without_graph_nodes is not async + vector_service.find_chunks_without_graph_nodes = MagicMock(return_value=[]) + + graph_service = AsyncMock() + graph_service.find_orphan_entities.return_value = [] + graph_service.get_all_document_references.return_value = [] + graph_service.find_documents_without_vectors.return_value = [] + + wiki_client = AsyncMock() + wiki_client.list_all_pages.return_value = [] + + result = await maintenance_health( + user="testuser", + vector_service=vector_service, + graph_service=graph_service, + wiki_client=wiki_client, + api_key="test" + ) + + assert result.status == "healthy" + assert result.orphan_vector_count == 0 + assert result.orphan_entity_count == 0 + assert result.vectors_without_graph == 0 + assert result.docs_without_vectors == 0 + + async def test_health_degraded(self): + """Test degraded status with orphans.""" + vector_service = AsyncMock() + vector_service.get_all_chunk_references.return_value = [ + {"chunk_id": f"c{i}", "page_id": 999, "doc_type": "wiki"} + for i in range(15) + ] + # find_chunks_without_graph_nodes is not async + vector_service.find_chunks_without_graph_nodes = MagicMock(return_value=[]) + + graph_service = AsyncMock() + graph_service.find_orphan_entities.return_value = [ + {"id": f"e{i}", "name": f"Entity{i}", "type": "Entity"} + for i in range(3) + ] + graph_service.get_all_document_references.return_value = [] + graph_service.find_documents_without_vectors.return_value = [] + + wiki_client = AsyncMock() + wiki_client.list_all_pages.return_value = [] + + result = await maintenance_health( + user="testuser", + vector_service=vector_service, + graph_service=graph_service, + wiki_client=wiki_client, + api_key="test" + ) + + assert result.status == "degraded" + assert result.orphan_vector_count == 15 + assert result.orphan_entity_count == 3 + assert len(result.recommendations) >= 1 + + +@pytest.mark.asyncio +class TestReindexPage: + """Test reindex page endpoint.""" + + async def test_reindex_success(self): + """Test successful page reindex.""" + vector_service = AsyncMock() + vector_service.delete_page_chunks.return_value = 5 + vector_service.update_from_page.return_value = MagicMock( + success=True, + chunks_created=6, + error_message=None + ) + + graph_service = AsyncMock() + graph_service.delete_page.return_value = 1 + graph_service.update_from_page.return_value = MagicMock( + success=True, + error_message=None + ) + + result = await reindex_page( + page_id=123, + user="testuser", + vector_service=vector_service, + graph_service=graph_service, + api_key="test" + ) + + assert result.success is True + assert result.page_id == 123 + assert result.vectors_deleted == 5 + assert result.vectors_created == 6 + assert result.graph_updated is True + + async def test_reindex_failure(self): + """Test reindex with failure.""" + vector_service = AsyncMock() + vector_service.delete_page_chunks.return_value = 0 + vector_service.update_from_page.return_value = MagicMock( + success=False, + chunks_created=0, + error_message="Page not found" + ) + + graph_service = AsyncMock() + graph_service.delete_page.return_value = 0 + graph_service.update_from_page.return_value = MagicMock( + success=False, + error_message="Page not found" + ) + + result = await reindex_page( + page_id=999, + user="testuser", + vector_service=vector_service, + graph_service=graph_service, + api_key="test" + ) + + assert result.success is False + assert result.error == "Page not found"