""" 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.services.volatile_service import VolatileCacheService from src.core.dependencies import ( VectorServiceDep, GraphServiceDep, WikiJSDep, RedisDep, QdrantDep, OllamaDep, PaperlessDep, verify_api_key ) from src.config import get_settings from src.core.multi_tenancy import DEFAULT_USER 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 class VolatileCleanupResponse(BaseModel): """Response from volatile cache cleanup operation.""" success: bool collections_processed: int total_expired_purged: int by_collection: Dict[str, int] = Field(default_factory=dict) duration_ms: float class TestDataCleanupResponse(BaseModel): """Response from test data cleanup operation.""" success: bool dry_run: bool wiki_pages_deleted: int graph_nodes_deleted: int vector_chunks_deleted: int pages_found: List[Dict[str, Any]] = Field(default_factory=list) duration_ms: float class PaperlessCleanupResponse(BaseModel): """Response from Paperless orphan cleanup operation.""" success: bool dry_run: bool paperless_ids_checked: int = Field(description="Total Paperless IDs found in indexes") orphans_found: int = Field(description="Documents deleted from Paperless but still indexed") orphan_ids: List[int] = Field(default_factory=list, description="Paperless IDs that are orphans") vector_chunks_deleted: int = Field(description="Vector chunks removed") graph_nodes_deleted: int = Field(description="Graph Document nodes removed") duration_ms: float # Test data path patterns - restricted to test user namespace only # These are the only paths that can be cleaned up for safety TEST_USER_PATH_PREFIXES = [ "users/llm-tester/", "users/llm_tester/", ] def _matches_test_user_path(path: str) -> bool: """Check if a path is in the test user namespace. Only matches paths that START with test user prefixes for safety. This prevents accidental deletion of non-test data. """ path_lower = path.lower() return any(path_lower.startswith(prefix) for prefix in TEST_USER_PATH_PREFIXES) # ========== 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.post("/cleanup/volatile", response_model=VolatileCleanupResponse) async def cleanup_volatile( qdrant: QdrantDep = None, ollama: OllamaDep = None, api_key: str = Depends(verify_api_key) ): """ Purge expired volatile cache records across all users. Loops through all volatile_* collections and removes records where ttl_expiry < current_timestamp. **Scheduler Task** - Recommended to run every 10 minutes. **Scheduler Integration:** ```json { "task_name": "volatile_cleanup", "schedule": "*/10 * * * *", "endpoint": "POST /maintenance/cleanup/volatile", "description": "Purge expired volatile cache records" } ``` """ start_time = time.time() try: settings = get_settings() service = VolatileCacheService( qdrant_client=qdrant, ollama_client=ollama, settings=settings ) # Purge expired from all volatile collections results = await service.purge_all_expired() total_purged = sum(results.values()) duration_ms = (time.time() - start_time) * 1000 logger.info(f"Volatile cleanup complete: {total_purged} expired records purged from {len(results)} collections") return VolatileCleanupResponse( success=True, collections_processed=len(results), total_expired_purged=total_purged, by_collection=results, duration_ms=duration_ms ) except Exception as e: logger.error(f"Volatile cleanup failed: {e}", exc_info=True) raise HTTPException(status_code=500, detail=str(e)) @router.post("/cleanup/test-data", response_model=TestDataCleanupResponse) async def cleanup_test_data( dry_run: bool = Query(default=True, description="Preview only, don't delete"), wiki: WikiJSDep = None, vector_service: VectorServiceDep = None, graph_service: GraphServiceDep = None, api_key: str = Depends(verify_api_key) ): """ Purge LLM tester data from wiki, graph, and vectors. **Security**: Only deletes pages in the test user namespace: - users/llm-tester/* - users/llm_tester/* This endpoint cannot delete data outside these paths. **Use dry_run=true (default) to preview what would be deleted.** **Scheduler Integration:** ```json { "task_name": "test_data_cleanup", "schedule": "0 3 * * 0", "endpoint": "POST /maintenance/cleanup/test-data?dry_run=false", "description": "Weekly cleanup of LLM test data" } ``` """ start_time = time.time() try: # List all wiki pages all_pages = await wiki.list_all_pages(batch_size=500) # Filter for test user paths only (security: restricted to test namespace) test_pages = [ {"id": p["id"], "path": p["path"], "title": p.get("title", "")} for p in all_pages if _matches_test_user_path(p.get("path", "")) ] logger.info(f"Found {len(test_pages)} test pages matching patterns: {TEST_USER_PATH_PREFIXES}") wiki_deleted = 0 graph_deleted = 0 vector_deleted = 0 if not dry_run and test_pages: for page in test_pages: page_id = page["id"] page_path = page["path"] try: # Delete vector chunks for this page (using DEFAULT_USER collection) chunks_removed = await vector_service.delete_page_chunks(page_id, DEFAULT_USER) vector_deleted += chunks_removed # Delete graph node for this page (returns count, may be 0 if no node) graph_removed = await graph_service.delete_page(page_id, DEFAULT_USER) graph_deleted += graph_removed # Delete wiki page (raises exception on failure, returns None on success) await wiki.delete_page(page_id) wiki_deleted += 1 logger.info(f"Deleted test page: {page_path} (id={page_id})") except Exception as e: logger.error(f"Failed to delete page {page_path}: {e}") continue duration_ms = (time.time() - start_time) * 1000 return TestDataCleanupResponse( success=True, dry_run=dry_run, wiki_pages_deleted=wiki_deleted, graph_nodes_deleted=graph_deleted, vector_chunks_deleted=vector_deleted, pages_found=test_pages, duration_ms=duration_ms ) except Exception as e: logger.error(f"Test data cleanup failed: {e}", exc_info=True) raise HTTPException(status_code=500, detail=str(e)) @router.post("/cleanup/paperless", response_model=PaperlessCleanupResponse) async def cleanup_paperless_orphans( user: str = Query(..., description="User identifier"), dry_run: bool = Query(default=True, description="Preview only, don't delete"), vector_service: VectorServiceDep = None, graph_service: GraphServiceDep = None, paperless: PaperlessDep = None, api_key: str = Depends(verify_api_key) ): """ Find and clean up Paperless document orphans. Detects documents that were indexed in Library Desk but have since been deleted from Paperless-ngx. Removes orphaned vectors and graph nodes. **Use dry_run=true (default) to preview what would be deleted.** **Scheduler Integration:** ```json { "task_name": "paperless_orphan_cleanup", "schedule": "0 5 * * *", "endpoint": "POST /maintenance/cleanup/paperless?user=jpmschweitzer&dry_run=false", "description": "Daily cleanup of orphaned Paperless documents" } ``` """ start_time = time.time() try: settings = get_settings() if not settings.paperless_token: raise HTTPException(status_code=503, detail="Paperless not configured") # Get all document chunks from vectors with doc_type="document" chunk_refs = await vector_service.get_all_chunk_references(user) doc_chunks = [ref for ref in chunk_refs if ref.get("doc_type") == "document"] # Extract unique paperless_ids paperless_ids = list(set( ref.get("paperless_id") for ref in doc_chunks if ref.get("paperless_id") )) logger.info(f"Found {len(paperless_ids)} unique Paperless IDs in indexes") # Check each against Paperless API orphan_ids = [] for pid in paperless_ids: try: doc = await paperless.get_document(pid) if doc is None: orphan_ids.append(pid) except Exception as e: # Document not found or API error - treat as orphan logger.debug(f"Paperless document {pid} not found: {e}") orphan_ids.append(pid) logger.info(f"Found {len(orphan_ids)} orphaned Paperless documents") # Delete orphans if not dry run vectors_deleted = 0 graph_deleted = 0 if not dry_run and orphan_ids: for pid in orphan_ids: try: # Delete vector chunks for this paperless_id chunks_removed = await vector_service.delete_paperless_document_chunks(pid, user) vectors_deleted += chunks_removed # Delete graph node for this paperless_id graph_removed = await graph_service.delete_paperless_document(pid, user) graph_deleted += graph_removed logger.info(f"Cleaned up orphaned Paperless document {pid}: {chunks_removed} chunks, {graph_removed} nodes") except Exception as e: logger.error(f"Failed to cleanup Paperless document {pid}: {e}") duration_ms = (time.time() - start_time) * 1000 return PaperlessCleanupResponse( success=True, dry_run=dry_run, paperless_ids_checked=len(paperless_ids), orphans_found=len(orphan_ids), orphan_ids=orphan_ids, vector_chunks_deleted=vectors_deleted, graph_nodes_deleted=graph_deleted, duration_ms=duration_ms ) except HTTPException: raise except Exception as e: logger.error(f"Paperless orphan 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))