diff --git a/services/library-desk/src/routers/ingestion.py b/services/library-desk/src/routers/ingestion.py new file mode 100644 index 0000000..9fcdafd --- /dev/null +++ b/services/library-desk/src/routers/ingestion.py @@ -0,0 +1,169 @@ +""" +Document Ingestion API Router + +Endpoints for ingesting wiki pages into the knowledge base (vectors + graph). +""" +from fastapi import APIRouter, Depends, HTTPException, Query +from typing import Optional + +from src.services.ingestion_service import IngestionService +from src.models.ingestion import ( + IngestionRequest, + IngestionResult, + BatchIngestionRequest, + BatchIngestionResult +) +from src.core.dependencies import get_ingestion_service, verify_api_key + +router = APIRouter(prefix="/ingest", tags=["Document Ingestion"]) + + +@router.post("/page", response_model=IngestionResult) +async def ingest_page( + request: IngestionRequest, + ingestion: IngestionService = Depends(get_ingestion_service), + api_key: str = Depends(verify_api_key) +): + """ + Ingest a single wiki page into the knowledge base. + + This endpoint: + 1. Fetches page content from Wiki.js + 2. Chunks content and generates embeddings (Qdrant) + 3. Extracts entities and updates knowledge graph (Neo4j) + + ## Use Cases + + - **After page creation**: Automatically called by consolidation service + - **Manual re-indexing**: Force refresh a page after manual edits + - **Selective ingestion**: Skip vectors or graph if only one is needed + + ## Performance + + - Typical page: 1-3 seconds + - Large page (>5000 words): 5-10 seconds + - Vector and graph ingestion run in parallel + + ## Example + + ```bash + curl -X POST http://192.168.86.149:8089/ingest/page \ + -H "Authorization: Bearer $API_KEY" \ + -H "Content-Type: application/json" \ + -d '{ + "page_id": 19, + "user": "jpmschweitzer", + "force_refresh": false + }' + ``` + """ + result = await ingestion.ingest_page( + page_id=request.page_id, + user=request.user, + force_refresh=request.force_refresh, + skip_vectors=request.skip_vectors, + skip_graph=request.skip_graph + ) + + if not result.success: + raise HTTPException( + status_code=500, + detail=f"Ingestion failed: {result.error}" + ) + + return result + + +@router.post("/batch", response_model=BatchIngestionResult) +async def ingest_batch( + request: BatchIngestionRequest, + ingestion: IngestionService = Depends(get_ingestion_service), + api_key: str = Depends(verify_api_key) +): + """ + Ingest multiple wiki pages concurrently. + + ## Concurrency Control + + The `max_concurrent` parameter controls how many pages are processed simultaneously: + - **Low (1-2)**: Safer for resource-constrained systems + - **Medium (3-5)**: Good balance of speed and stability + - **High (6-10)**: Maximum speed, requires good resources + + ## Batch Size Recommendations + + - **Small batches (<10 pages)**: Use max_concurrent=3-5 + - **Medium batches (10-50 pages)**: Use max_concurrent=3 + - **Large batches (>50 pages)**: Use max_concurrent=2, consider splitting + + ## Example + + ```bash + curl -X POST http://192.168.86.149:8089/ingest/batch \ + -H "Authorization: Bearer $API_KEY" \ + -H "Content-Type: application/json" \ + -d '{ + "page_ids": [19, 20, 21, 22], + "user": "jpmschweitzer", + "max_concurrent": 3 + }' + ``` + """ + result = await ingestion.ingest_batch( + page_ids=request.page_ids, + user=request.user, + force_refresh=request.force_refresh, + skip_vectors=request.skip_vectors, + skip_graph=request.skip_graph, + max_concurrent=request.max_concurrent + ) + + return result + + +@router.post("/all", response_model=BatchIngestionResult) +async def ingest_all_pages( + user: str = Query(default="jpmschweitzer", description="User identifier"), + path_prefix: Optional[str] = Query(None, description="Path prefix filter (e.g., 'users/jpmschweitzer/tech')"), + force_refresh: bool = Query(False, description="Force re-ingestion of all pages"), + max_concurrent: int = Query(3, ge=1, le=10, description="Maximum concurrent ingestion tasks"), + ingestion: IngestionService = Depends(get_ingestion_service), + api_key: str = Depends(verify_api_key) +): + """ + Ingest all wiki pages for a user (bulk re-indexing). + + ## Use Cases + + - **Initial setup**: Index all existing pages + - **Full re-index**: After major schema changes + - **Path-specific**: Re-index a specific section (e.g., all tech docs) + + ## Performance + + - **Small wiki (<50 pages)**: 2-5 minutes + - **Medium wiki (50-200 pages)**: 5-20 minutes + - **Large wiki (>200 pages)**: 20+ minutes + + **Recommendation**: Run as background job for large wikis + + ## Example + + ```bash + # Ingest all pages for user + curl -X POST "http://192.168.86.149:8089/ingest/all?user=jpmschweitzer" \ + -H "Authorization: Bearer $API_KEY" + + # Ingest only tech docs + curl -X POST "http://192.168.86.149:8089/ingest/all?user=jpmschweitzer&path_prefix=users/jpmschweitzer/tech" \ + -H "Authorization: Bearer $API_KEY" + ``` + """ + result = await ingestion.ingest_all_pages( + user=user, + path_prefix=path_prefix, + force_refresh=force_refresh, + max_concurrent=max_concurrent + ) + + return result diff --git a/services/library-desk/src/routers/vector.py b/services/library-desk/src/routers/vector.py new file mode 100644 index 0000000..ebb121f --- /dev/null +++ b/services/library-desk/src/routers/vector.py @@ -0,0 +1,173 @@ +""" +Vector router for Library Desk API. + +Endpoints for semantic search and vector operations. +""" + +from fastapi import APIRouter, HTTPException, Depends, Query +from typing import Optional +import logging + +from src.models.vector import ( + SearchRequest, SearchResponse, + VectorUpdateRequest, VectorUpdateSummary, + CollectionListResponse, + DeletePageChunksRequest, DeletePageChunksResponse +) +from src.services.vector_service import VectorService +from src.clients.qdrant_client import QdrantClientWrapper +from src.clients.wikijs_client import WikiJSClient +from src.clients.ollama_client import OllamaClient +from src.core.dependencies import QdrantDep, WikiJSDep, OllamaDep, verify_api_key +from src.core.multi_tenancy import DEFAULT_USER + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/vector", tags=["Vector"]) + + +# Dependency to get vector service +def get_vector_service( + qdrant_client: QdrantDep, + wiki_client: WikiJSDep, + ollama_client: OllamaDep +) -> VectorService: + """Get vector service instance.""" + return VectorService(qdrant_client, wiki_client, ollama_client) + + +@router.post("/search", response_model=SearchResponse) +async def semantic_search( + request: SearchRequest, + vector_service: VectorService = Depends(get_vector_service), + api_key: str = Depends(verify_api_key) +): + """ + Perform semantic search across user's documents. + + Uses Ollama to generate query embedding, then searches Qdrant + for similar document chunks. + + **Example Request:** + ```json + { + "query": "how to configure docker", + "user": "jpmschweitzer", + "limit": 10, + "score_threshold": 0.5 + } + ``` + + **Returns:** List of matching chunks with similarity scores + """ + try: + return await vector_service.search( + query=request.query, + user=request.user, + limit=request.limit, + score_threshold=request.score_threshold + ) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + except Exception as e: + logger.error(f"Semantic search failed: {e}", exc_info=True) + raise HTTPException(status_code=500, detail="Search failed") + + +@router.post("/update-from-page/{page_id}", response_model=VectorUpdateSummary) +async def update_vectors_from_page( + page_id: int, + user: str = Query(default=DEFAULT_USER, description="User identifier"), + force_refresh: bool = Query(default=False, description="Force re-embedding"), + vector_service: VectorService = Depends(get_vector_service), + api_key: str = Depends(verify_api_key) +): + """ + Update vector embeddings from a wiki page. + + This endpoint: + 1. Fetches the page from Wiki.js + 2. Chunks the content (500 tokens with 50 token overlap) + 3. Generates embeddings via Ollama + 4. Upserts chunks to Qdrant with metadata + + **Use Cases:** + - Called automatically after page creation/update (via BackgroundTasks) + - Called manually by user/Librarian to refresh vectors + - Called by Scheduler for batch processing + + **Example:** `POST /vector/update-from-page/5?user=jpmschweitzer` + + **Returns:** Summary with chunks created and processing time + """ + try: + summary = await vector_service.update_from_page( + page_id=page_id, + user=user, + force_refresh=force_refresh + ) + + if not summary.success: + raise HTTPException( + status_code=500, + detail=f"Vector update failed: {summary.error_message}" + ) + + return summary + + except HTTPException: + raise + except Exception as e: + logger.error(f"Failed to update vectors from page {page_id}: {e}", exc_info=True) + raise HTTPException(status_code=500, detail="Vector update failed") + + +@router.delete("/pages/{page_id}", response_model=DeletePageChunksResponse) +async def delete_page_chunks( + page_id: int, + user: str = Query(default=DEFAULT_USER, description="User identifier"), + vector_service: VectorService = Depends(get_vector_service), + api_key: str = Depends(verify_api_key) +): + """ + Delete all vector chunks for a wiki page. + + This is automatically called when a page is deleted from the wiki. + + **Example:** `DELETE /vector/pages/5?user=jpmschweitzer` + """ + try: + deleted_count = await vector_service.delete_page_chunks( + page_id=page_id, + user=user + ) + + return DeletePageChunksResponse( + page_id=page_id, + chunks_deleted=deleted_count, + success=deleted_count > 0, + message=f"Deleted {deleted_count} chunks for page {page_id}" + ) + + except Exception as e: + logger.error(f"Failed to delete chunks for page {page_id}: {e}", exc_info=True) + raise HTTPException(status_code=500, detail="Failed to delete chunks") + + +@router.get("/collections", response_model=CollectionListResponse) +async def list_collections( + vector_service: VectorService = Depends(get_vector_service), + api_key: str = Depends(verify_api_key) +): + """ + List all Qdrant collections with statistics. + + Returns collection names, vector counts, and point counts. + + **Example:** `GET /vector/collections` + """ + try: + return await vector_service.list_collections() + except Exception as e: + logger.error(f"Failed to list collections: {e}", exc_info=True) + raise HTTPException(status_code=500, detail="Failed to list collections") diff --git a/services/library-desk/src/routers/wiki.py b/services/library-desk/src/routers/wiki.py new file mode 100644 index 0000000..c164e12 --- /dev/null +++ b/services/library-desk/src/routers/wiki.py @@ -0,0 +1,409 @@ +""" +Wiki router for Library Desk API. + +Endpoints for wiki page and dossier management. +All operations are scoped to user namespaces for multi-tenancy. +""" + +from fastapi import APIRouter, HTTPException, Depends, Query, Security, BackgroundTasks +from fastapi.security import HTTPAuthorizationCredentials +from typing import Optional +import logging + +from src.models.wiki import ( + WikiPage, WikiPageList, WikiPageCreate, WikiPageUpdate, WikiPageMove, + WikiOperationResponse, WikiSearchResponse, + DossierList, WikiSearchResult +) +from src.services.wiki_service import WikiService +from src.services.graph_service import GraphService +from src.services.vector_service import VectorService +from src.clients.wikijs_client import WikiJSClient +from src.clients.neo4j_client import Neo4jClient +from src.clients.qdrant_client import QdrantClientWrapper +from src.clients.ollama_client import OllamaClient +from src.core.dependencies import WikiJSDep, Neo4jDep, QdrantDep, OllamaDep, verify_api_key +from src.core.multi_tenancy import DEFAULT_USER + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/wiki", tags=["Wiki"]) + + +# Dependency to get wiki service +def get_wiki_service(wiki_client: WikiJSDep) -> WikiService: + """Get wiki service instance.""" + return WikiService(wiki_client) + + +# Dependency to get graph service +def get_graph_service(neo4j_client: Neo4jDep, wiki_client: WikiJSDep) -> GraphService: + """Get graph service instance.""" + return GraphService(neo4j_client, wiki_client) + + +# Dependency to get vector service +def get_vector_service( + qdrant_client: QdrantDep, + wiki_client: WikiJSDep, + ollama_client: OllamaDep +) -> VectorService: + """Get vector service instance.""" + return VectorService(qdrant_client, wiki_client, ollama_client) + + +# Page operations +@router.get("/pages", response_model=WikiPageList) +async def list_pages( + user: str = Query(default=DEFAULT_USER, description="User identifier"), + tag: Optional[str] = Query(default=None, description="Filter by tag (dossier)"), + limit: int = Query(default=50, ge=1, le=200, description="Maximum pages to return"), + wiki_service: WikiService = Depends(get_wiki_service), + api_key: str = Depends(verify_api_key) +): + """ + List wiki pages for a user. + + Optionally filter by tag (dossier). Pages are scoped to user's namespace. + + **Example:** `/wiki/pages?user=jpmschweitzer&tag=projects&limit=20` + """ + try: + return await wiki_service.list_pages(user=user, tag=tag, limit=limit) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + except Exception as e: + logger.error(f"Failed to list pages: {e}", exc_info=True) + raise HTTPException(status_code=500, detail="Internal server error") + + +@router.get("/pages/{page_id}", response_model=WikiPage) +async def get_page( + page_id: int, + user: str = Query(default=DEFAULT_USER, description="User identifier"), + wiki_service: WikiService = Depends(get_wiki_service), + api_key: str = Depends(verify_api_key) +): + """ + Get a single wiki page by ID. + + Access is restricted to pages within the user's namespace. + + **Example:** `/wiki/pages/123?user=jpmschweitzer` + """ + try: + page = await wiki_service.get_page(page_id=page_id, user=user) + if not page: + raise HTTPException(status_code=404, detail=f"Page {page_id} not found or access denied") + return page + except HTTPException: + raise + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + except Exception as e: + logger.error(f"Failed to get page {page_id}: {e}", exc_info=True) + raise HTTPException(status_code=500, detail="Internal server error") + + +@router.post("/pages", response_model=WikiPage, status_code=201) +async def create_page( + page_data: WikiPageCreate, + background_tasks: BackgroundTasks, + wiki_service: WikiService = Depends(get_wiki_service), + graph_service: GraphService = Depends(get_graph_service), + vector_service: VectorService = Depends(get_vector_service), + api_key: str = Depends(verify_api_key) +): + """ + Create a new wiki page. + + The page will be created in the user's namespace. If path doesn't start + with namespace, it will be automatically prefixed. + + **Auto-updates knowledge graph and vector embeddings**: After creating + the page, the graph and vectors are automatically updated in the + background to extract entities/relationships and generate semantic embeddings. + + **Example Request:** + ```json + { + "title": "Library Desk Architecture", + "path": "/projects/library-desk/architecture", + "content": "# Architecture\\n\\nThis describes...", + "description": "Architecture documentation", + "tags": ["projects", "architecture"], + "user": "jpmschweitzer" + } + ``` + """ + try: + page = await wiki_service.create_page(page_data) + + user = page_data.user or DEFAULT_USER + + # Schedule BOTH graph and vector updates in background (non-blocking) + background_tasks.add_task( + graph_service.update_from_page, + page_id=page.id, + user=user + ) + background_tasks.add_task( + vector_service.update_from_page, + page_id=page.id, + user=user + ) + + logger.info(f"Page {page.id} created, graph and vector updates scheduled") + return page + + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + except Exception as e: + logger.error(f"Failed to create page: {e}", exc_info=True) + raise HTTPException(status_code=500, detail="Internal server error") + + +@router.put("/pages/{page_id}", response_model=WikiPage) +async def update_page( + page_id: int, + page_data: WikiPageUpdate, + background_tasks: BackgroundTasks, + user: str = Query(default=DEFAULT_USER, description="User identifier"), + wiki_service: WikiService = Depends(get_wiki_service), + graph_service: GraphService = Depends(get_graph_service), + vector_service: VectorService = Depends(get_vector_service), + api_key: str = Depends(verify_api_key) +): + """ + Update an existing wiki page. + + Only pages within the user's namespace can be updated. + Partial updates are supported - only provided fields will be updated. + + **Auto-updates knowledge graph and vector embeddings**: After updating + the page, the graph and vectors are automatically refreshed in the + background to reflect the changes. + + **Example Request:** + ```json + { + "title": "Updated Title", + "tags": ["projects", "updated"] + } + ``` + """ + try: + page = await wiki_service.update_page( + page_id=page_id, + page_data=page_data, + user=user + ) + + # Schedule BOTH graph and vector updates in background (non-blocking) + background_tasks.add_task( + graph_service.update_from_page, + page_id=page_id, + user=user, + force_refresh=True # Force refresh on updates + ) + background_tasks.add_task( + vector_service.update_from_page, + page_id=page_id, + user=user, + force_refresh=True # Force refresh on updates + ) + + logger.info(f"Page {page_id} updated, graph and vector refresh scheduled") + return page + + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + except Exception as e: + logger.error(f"Failed to update page {page_id}: {e}", exc_info=True) + raise HTTPException(status_code=500, detail="Internal server error") + + +@router.delete("/pages/{page_id}", response_model=WikiOperationResponse) +async def delete_page( + page_id: int, + background_tasks: BackgroundTasks, + user: str = Query(default=DEFAULT_USER, description="User identifier"), + wiki_service: WikiService = Depends(get_wiki_service), + vector_service: VectorService = Depends(get_vector_service), + graph_service: GraphService = Depends(get_graph_service), + api_key: str = Depends(verify_api_key) +): + """ + Delete a wiki page. + + Only pages within the user's namespace can be deleted. + This operation cannot be undone. + + **Auto-cleanup**: Vector chunks and graph nodes for this page are + automatically deleted in the background. + """ + try: + success = await wiki_service.delete_page(page_id=page_id, user=user) + + # Schedule vector cleanup in background + background_tasks.add_task( + vector_service.delete_page_chunks, + page_id=page_id, + user=user + ) + + # Schedule graph cleanup in background + background_tasks.add_task( + graph_service.delete_page, + page_id=page_id, + user=user + ) + + logger.info(f"Page {page_id} deleted, vector and graph cleanup scheduled") + return WikiOperationResponse( + success=success, + message=f"Page {page_id} deleted successfully", + page_id=page_id + ) + except ValueError as e: + raise HTTPException(status_code=404, detail=str(e)) + except Exception as e: + logger.error(f"Failed to delete page {page_id}: {e}", exc_info=True) + raise HTTPException(status_code=500, detail="Internal server error") + + +@router.post("/pages/{page_id}/move", response_model=WikiOperationResponse) +async def move_page( + page_id: int, + move_data: WikiPageMove, + user: str = Query(default=DEFAULT_USER, description="User identifier"), + wiki_service: WikiService = Depends(get_wiki_service), + api_key: str = Depends(verify_api_key) +): + """ + Move or rename a wiki page. + + The new path must be within the user's namespace. + + **Example Request:** + ```json + { + "new_path": "/projects/library-desk/docs/architecture" + } + ``` + """ + try: + success = await wiki_service.move_page( + page_id=page_id, + new_path=move_data.new_path, + user=user + ) + if not success: + raise HTTPException(status_code=500, detail="Failed to move page") + + return WikiOperationResponse( + success=True, + message=f"Page {page_id} moved successfully", + page_id=page_id, + page_path=move_data.new_path + ) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + except HTTPException: + raise + except Exception as e: + logger.error(f"Failed to move page {page_id}: {e}", exc_info=True) + raise HTTPException(status_code=500, detail="Internal server error") + + +# Search operations +@router.get("/search", response_model=WikiSearchResponse) +async def search_pages( + q: str = Query(..., min_length=1, description="Search query"), + user: str = Query(default=DEFAULT_USER, description="User identifier"), + limit: int = Query(default=20, ge=1, le=100, description="Maximum results"), + wiki_service: WikiService = Depends(get_wiki_service), + api_key: str = Depends(verify_api_key) +): + """ + Search wiki pages within user's namespace. + + **Example:** `/wiki/search?q=architecture&user=jpmschweitzer&limit=10` + """ + try: + results = await wiki_service.search_pages( + query=q, + user=user, + limit=limit + ) + + return WikiSearchResponse( + results=[ + WikiSearchResult( + id=r.id, + path=r.path, + title=r.title, + description=r.description, + relevance=None + ) + for r in results + ], + query=q, + total=len(results) + ) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + except Exception as e: + logger.error(f"Search failed: {e}", exc_info=True) + raise HTTPException(status_code=500, detail="Internal server error") + + +# Dossier operations +@router.get("/dossiers", response_model=DossierList) +async def list_dossiers( + user: str = Query(default=DEFAULT_USER, description="User identifier"), + wiki_service: WikiService = Depends(get_wiki_service), + api_key: str = Depends(verify_api_key) +): + """ + List all dossiers (unique tags) for a user. + + Dossiers are tag-based collections of pages. + + **Example:** `/wiki/dossiers?user=jpmschweitzer` + """ + try: + return await wiki_service.list_dossiers(user=user) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + except Exception as e: + logger.error(f"Failed to list dossiers: {e}", exc_info=True) + raise HTTPException(status_code=500, detail="Internal server error") + + +@router.get("/dossiers/{dossier_name}/pages", response_model=WikiPageList) +async def get_dossier_pages( + dossier_name: str, + user: str = Query(default=DEFAULT_USER, description="User identifier"), + limit: int = Query(default=100, ge=1, le=500, description="Maximum pages"), + wiki_service: WikiService = Depends(get_wiki_service), + api_key: str = Depends(verify_api_key) +): + """ + Get all pages in a dossier. + + Returns pages tagged with the dossier name. + + **Example:** `/wiki/dossiers/projects/pages?user=jpmschweitzer` + """ + try: + return await wiki_service.get_dossier_pages( + dossier_name=dossier_name, + user=user, + limit=limit + ) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + except Exception as e: + logger.error(f"Failed to get dossier pages: {e}", exc_info=True) + raise HTTPException(status_code=500, detail="Internal server error")