diff --git a/CONTAINERS.md b/CONTAINERS.md index 87c0dbf..c27659d 100644 --- a/CONTAINERS.md +++ b/CONTAINERS.md @@ -25,6 +25,7 @@ | **Open WebUI** | 82 | http://192.168.86.149:82 | LAN | No | ✅ Running | | **Core API** | 8083 | http://192.168.86.149:8083 | LAN | No | ✅ Running | | **Core AI** | 8086 | http://192.168.86.149:8086 | LAN | No | ✅ Running | +| **Scheduler** | 8090 | http://192.168.86.149:8090 | LAN | No | 🔧 Deploying | | **SearXNG** | 8087 | http://192.168.86.149:8087 | LAN | No | ✅ Running | | **Jellyfin** | 8096, 8920, 7359, 1900 | https://media.schweitz.net | Internet | Yes (RTX 2080 Ti) | ✅ Running | | **Nextcloud** | 8082 | https://cloud.schweitz.net | Internet | No | ⚠️ Stopped | diff --git a/services/library-desk/LIBRARIAN_INTEGRATION.md b/services/library-desk/LIBRARIAN_INTEGRATION.md new file mode 100644 index 0000000..afc04ae --- /dev/null +++ b/services/library-desk/LIBRARIAN_INTEGRATION.md @@ -0,0 +1,473 @@ +# Librarian Integration Guide + +**How The Scheduler (Librarian) interacts with Library Desk for documentation management** + +## Overview + +The Scheduler's documentation mirroring tasks feed content into The Library system via Library Desk API. This creates a knowledge graph and vector index of all documentation for semantic search and relationship discovery. + +## Architecture Flow + +``` +┌─────────────────┐ +│ The Scheduler │ (The Librarian) +│ (scheduler) │ +└────────┬────────┘ + │ 1. Mirror docs from sources + │ (GitHub, Gitea, etc.) + ↓ +┌─────────────────┐ +│ Gitea Repo │ +│ docs-mirror/* │ +└────────┬────────┘ + │ 2. Ingest to Library + ↓ +┌─────────────────┐ +│ Library Desk │ (Coordination API) +│ (library-desk) │ +└────────┬────────┘ + │ 3. Process & Index + ├──→ Neo4j (relationships) + ├──→ Qdrant (embeddings) + └──→ Wiki.js (dossiers) +``` + +## Required Endpoints + +### 1. Document Ingestion + +**POST /ingest/document** +```json +{ + "source": "github", + "repository": "anthropics/anthropic-cookbook", + "path": "skills/citation/guide.md", + "content": "# Citation Guide\n...", + "metadata": { + "commit_sha": "abc123", + "author": "Anthropic", + "updated_at": "2025-12-08T10:30:00Z", + "gitea_mirror_path": "docs-mirror/anthropic-cookbook/skills/citation/guide.md", + "language": "markdown", + "tags": ["skills", "citation", "prompting"] + } +} +``` + +**Response:** +```json +{ + "document_id": "doc_abc123xyz", + "status": "processing", + "operations": { + "chunking": "pending", + "embedding": "pending", + "entity_extraction": "pending", + "graph_indexing": "pending" + }, + "estimated_completion": "2025-12-08T10:30:15Z" +} +``` + +### 2. Batch Ingestion + +**POST /ingest/batch** +```json +{ + "source": "github", + "repository": "anthropics/anthropic-cookbook", + "documents": [ + { + "path": "skills/citation/guide.md", + "content": "...", + "metadata": {...} + }, + { + "path": "skills/summarization/techniques.md", + "content": "...", + "metadata": {...} + } + ] +} +``` + +**Response:** +```json +{ + "batch_id": "batch_xyz789", + "total_documents": 2, + "status": "processing", + "documents": [ + {"document_id": "doc_1", "status": "queued"}, + {"document_id": "doc_2", "status": "queued"} + ] +} +``` + +### 3. Document Status Check + +**GET /ingest/status/{document_id}** + +**Response:** +```json +{ + "document_id": "doc_abc123xyz", + "status": "completed", + "operations": { + "chunking": "completed", + "embedding": "completed", + "entity_extraction": "completed", + "graph_indexing": "completed" + }, + "results": { + "chunks_created": 12, + "vectors_indexed": 12, + "entities_extracted": 8, + "relationships_created": 15 + }, + "completed_at": "2025-12-08T10:30:14Z" +} +``` + +### 4. Content Update Detection + +**POST /ingest/check-updates** +```json +{ + "documents": [ + { + "path": "docs-mirror/anthropic-cookbook/skills/citation/guide.md", + "content_hash": "sha256:abc123...", + "updated_at": "2025-12-08T10:30:00Z" + } + ] +} +``` + +**Response:** +```json +{ + "updates_needed": [ + { + "path": "docs-mirror/anthropic-cookbook/skills/citation/guide.md", + "reason": "content_changed", + "last_indexed": "2025-12-07T10:30:00Z", + "action": "re-index" + } + ], + "up_to_date": [], + "new_documents": [] +} +``` + +### 5. Deduplication Check + +**POST /deduplicate/check** +```json +{ + "document_id": "doc_abc123xyz", + "similarity_threshold": 0.85 +} +``` + +**Response:** +```json +{ + "duplicates": [ + { + "document_id": "doc_def456", + "similarity": 0.92, + "path": "docs-mirror/claude-docs/citation-best-practices.md", + "overlap_summary": "Both documents cover citation formatting" + } + ], + "suggestions": { + "action": "merge_or_cross_reference", + "confidence": 0.88 + } +} +``` + +### 6. Repository Sync Status + +**GET /ingest/repo-status/{repository_name}** + +**Response:** +```json +{ + "repository": "anthropic-cookbook", + "total_documents": 156, + "indexed_documents": 156, + "pending_updates": 0, + "failed_documents": 0, + "last_sync": "2025-12-08T03:00:00Z", + "next_scheduled_sync": "2025-12-09T03:00:00Z" +} +``` + +## Scheduler Integration Workflow + +### Phase 1: Mirror Documentation (Current) +```python +# This already exists in the Scheduler +async def mirror_documentation(): + """Mirror docs from external sources to Gitea""" + repos = [ + "anthropics/anthropic-cookbook", + "anthropics/prompt-eng-interactive-tutorial", + # ... etc + ] + + for repo in repos: + # Clone/pull to /docs-mirror/{repo-name} + await git_sync(repo, f"/docs-mirror/{repo}") +``` + +### Phase 2: Index to Library (New) +```python +async def index_to_library(): + """Send mirrored docs to Library Desk for indexing""" + + # Get list of documents in docs-mirror + docs_path = Path("/docs-mirror") + + for repo_dir in docs_path.iterdir(): + if not repo_dir.is_dir(): + continue + + # Find all markdown files + md_files = list(repo_dir.rglob("*.md")) + + # Check what needs updating + update_check = await check_library_updates(md_files) + + if update_check["updates_needed"]: + # Batch ingest updated documents + await batch_ingest_documents( + repository=repo_dir.name, + documents=update_check["updates_needed"] + ) + + # Wait for processing to complete + await wait_for_batch_completion(batch_id) + + # Check for duplicates + await check_and_resolve_duplicates(repo_dir.name) +``` + +### Phase 3: Monitor & Maintain +```python +async def maintain_library_index(): + """Periodic maintenance of Library index""" + + # Check for orphaned entries (deleted from source) + await cleanup_orphaned_documents() + + # Update embeddings if model changed + await refresh_embeddings_if_needed() + + # Generate relationship maps for new content + await discover_document_relationships() +``` + +## Scheduler Task Definition + +**New Task: `library_sync`** + +```yaml +Task Name: library_sync +Description: Sync mirrored documentation to Library for indexing and search +Schedule: Daily at 03:30 (after doc mirroring at 03:00) +Priority: 15 (user maintenance) +Service: library +Executor: scheduler.tasks.library_tasks.sync_library_index +Dependencies: + - docs_mirror (must complete first) +Configuration: + - LIBRARY_DESK_URL: http://library-desk:8089 + - LIBRARY_API_KEY: ${LIBRARY_API_KEY} + - BATCH_SIZE: 50 + - CHECK_UPDATES_ONLY: true +Outputs: + - Documents indexed + - Duplicates found + - Relationships created +``` + +## API Client Example + +```python +# scheduler/src/clients/library_desk.py + +import httpx +from typing import List, Dict, Any +from pathlib import Path + +class LibraryDeskClient: + def __init__(self, base_url: str, api_key: str): + self.base_url = base_url + self.headers = { + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json" + } + + async def ingest_document( + self, + source: str, + repository: str, + path: str, + content: str, + metadata: Dict[str, Any] + ) -> Dict[str, Any]: + """Ingest a single document""" + async with httpx.AsyncClient() as client: + response = await client.post( + f"{self.base_url}/ingest/document", + headers=self.headers, + json={ + "source": source, + "repository": repository, + "path": path, + "content": content, + "metadata": metadata + } + ) + response.raise_for_status() + return response.json() + + async def batch_ingest( + self, + source: str, + repository: str, + documents: List[Dict[str, Any]] + ) -> Dict[str, Any]: + """Ingest multiple documents""" + async with httpx.AsyncClient() as client: + response = await client.post( + f"{self.base_url}/ingest/batch", + headers=self.headers, + json={ + "source": source, + "repository": repository, + "documents": documents + }, + timeout=300.0 # 5 minutes for large batches + ) + response.raise_for_status() + return response.json() + + async def check_updates( + self, + documents: List[Dict[str, str]] + ) -> Dict[str, Any]: + """Check which documents need updating""" + async with httpx.AsyncClient() as client: + response = await client.post( + f"{self.base_url}/ingest/check-updates", + headers=self.headers, + json={"documents": documents} + ) + response.raise_for_status() + return response.json() + + async def get_repo_status(self, repository: str) -> Dict[str, Any]: + """Get indexing status for a repository""" + async with httpx.AsyncClient() as client: + response = await client.get( + f"{self.base_url}/ingest/repo-status/{repository}", + headers=self.headers + ) + response.raise_for_status() + return response.json() +``` + +## Typical Workflow Sequence + +### Daily Documentation Sync (03:00-03:45) + +1. **03:00** - Scheduler runs `docs_mirror` task + - Pulls latest from all configured repos + - Writes to `/docs-mirror/*` + +2. **03:30** - Scheduler runs `library_sync` task + - Scans `/docs-mirror/` for changes + - Calls `POST /ingest/check-updates` with file hashes + - Gets list of updated/new documents + +3. **03:31-03:40** - Batch ingestion + - Groups documents by repo + - Calls `POST /ingest/batch` for each repo + - Monitors `GET /ingest/status/{batch_id}` + +4. **03:41-03:44** - Post-processing + - Calls `POST /deduplicate/check` for new docs + - Reviews duplicate suggestions + - Logs statistics to Scheduler database + +5. **03:45** - Completion + - Scheduler marks task complete + - Sends summary to logs + - Updates next run time + +## Error Handling + +### Retry Strategy +```python +async def ingest_with_retry(document: Dict, max_retries: int = 3): + """Ingest with exponential backoff""" + for attempt in range(max_retries): + try: + result = await library_client.ingest_document(**document) + return result + except httpx.TimeoutException: + if attempt < max_retries - 1: + await asyncio.sleep(2 ** attempt) # 1s, 2s, 4s + else: + # Log failure and continue + logger.error(f"Failed to ingest {document['path']} after {max_retries} attempts") + return None +``` + +### Graceful Degradation +- If Library Desk is down, queue documents for later ingestion +- Store failed ingestions in Scheduler database +- Retry failed ingestions on next run + +## Metrics to Track + +The Scheduler should track: +- Documents mirrored vs. documents indexed +- Average ingestion time per document +- Deduplication rate +- Failed ingestions +- Library Desk response times + +These can be displayed in the Scheduler UI dashboard. + +## Environment Variables + +Add to Scheduler's environment: +```bash +# Library Integration +LIBRARY_DESK_URL=http://library-desk:8089 +LIBRARY_API_KEY=${LIBRARY_API_KEY} +LIBRARY_BATCH_SIZE=50 +LIBRARY_SYNC_ENABLED=true +``` + +## Next Steps + +1. Implement ingestion endpoints in Library Desk +2. Add LibraryDeskClient to Scheduler +3. Create `library_sync` task in Scheduler +4. Test with small batch of documents +5. Monitor and tune performance +6. Expand to full documentation corpus + +## Benefits + +- **Automatic Knowledge Base**: All mirrored docs automatically indexed +- **Semantic Search**: Find docs by meaning, not just keywords +- **Relationship Discovery**: Understand connections between docs +- **Deduplication**: Identify overlapping content across repos +- **HybridRAG Ready**: Knowledge graph + vectors enable advanced AI queries diff --git a/services/library-desk/README.md b/services/library-desk/README.md index b5c688b..1bb5bd2 100644 --- a/services/library-desk/README.md +++ b/services/library-desk/README.md @@ -193,6 +193,16 @@ The container: - **Logs**: `docker logs library-desk` - **Stats**: `GET /stats` (requires API key) +## Scheduler Integration + +See [LIBRARIAN_INTEGRATION.md](./LIBRARIAN_INTEGRATION.md) for details on how The Scheduler (Librarian) integrates with Library Desk for automated documentation indexing. + +**Key Workflow:** +1. Scheduler mirrors docs to Gitea (daily 03:00) +2. Scheduler syncs to Library Desk (daily 03:30) +3. Library Desk ingests, chunks, embeds, and indexes +4. Content becomes searchable via HybridRAG + ## Future Enhancements - [ ] Implement HybridRAG query logic @@ -202,6 +212,7 @@ The container: - [ ] Add entity extraction (spaCy/NLP) - [ ] Implement mind map generation - [ ] Add deduplication logic +- [ ] **Implement Scheduler integration endpoints** (see LIBRARIAN_INTEGRATION.md) - [ ] Add comprehensive tests - [ ] Add rate limiting - [ ] Add request tracing diff --git a/services/library-desk/pytest.ini b/services/library-desk/pytest.ini new file mode 100644 index 0000000..dcb2c1c --- /dev/null +++ b/services/library-desk/pytest.ini @@ -0,0 +1,30 @@ +[pytest] +# Pytest configuration for Library Desk + +# Test discovery +python_files = test_*.py +python_classes = Test* +python_functions = test_* + +# Asyncio settings +asyncio_mode = auto +asyncio_default_fixture_loop_scope = function + +# Output options +addopts = + -v + --strict-markers + --tb=short + --disable-warnings + +# Markers for test categorization +markers = + unit: Unit tests (no external dependencies) + integration: Integration tests (require services) + slow: Slow tests that take more than 1 second + +# Test paths +testpaths = tests + +# Minimum Python version +minversion = 3.12 diff --git a/services/library-desk/requirements.txt b/services/library-desk/requirements.txt index 2261824..ab59ec0 100644 --- a/services/library-desk/requirements.txt +++ b/services/library-desk/requirements.txt @@ -23,3 +23,7 @@ python-multipart~=0.0.20 # Utilities python-dateutil~=2.9.0 + +# Testing +pytest~=8.3.0 +pytest-asyncio~=0.24.0 diff --git a/services/library-desk/src/clients/__init__.py b/services/library-desk/src/clients/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/services/library-desk/src/clients/neo4j_client.py b/services/library-desk/src/clients/neo4j_client.py new file mode 100644 index 0000000..fe93545 --- /dev/null +++ b/services/library-desk/src/clients/neo4j_client.py @@ -0,0 +1,413 @@ +""" +Neo4j async client for Library Desk. + +Provides async Neo4j operations with: +- Connection pooling via AsyncGraphDatabase +- Session management with context managers +- Multi-tenancy support via user labels +- Automatic retry on transient failures +""" + +from neo4j import AsyncGraphDatabase, AsyncDriver, AsyncSession +from typing import Optional, List, Dict, Any +import logging + +from src.core.multi_tenancy import get_neo4j_user_label + +logger = logging.getLogger(__name__) + + +class Neo4jClient: + """ + Async Neo4j client with connection pooling. + + Features: + - Singleton driver instance (expensive to create) + - Session-per-request pattern (lightweight) + - Automatic transaction retry + - Multi-tenancy via user-specific labels + """ + + def __init__(self, uri: str, user: str, password: str): + """ + Initialize Neo4j client. + + Args: + uri: Neo4j Bolt URI (e.g., "bolt://neo4j:7687") + user: Neo4j username + password: Neo4j password + """ + self.uri = uri + self._driver: Optional[AsyncDriver] = None + self._auth = (user, password) + + async def connect(self): + """ + Initialize connection pool. + + Should be called once at app startup. + Driver handles connection pooling internally. + """ + if not self._driver: + self._driver = AsyncGraphDatabase.driver( + self.uri, + auth=self._auth, + max_connection_pool_size=50, + connection_timeout=30.0, + max_transaction_retry_time=30.0 + ) + # Verify connectivity + await self._driver.verify_connectivity() + logger.info(f"Connected to Neo4j at {self.uri}") + + async def close(self): + """ + Close connection pool. + + Should be called once at app shutdown. + """ + if self._driver: + await self._driver.close() + self._driver = None + logger.info("Closed Neo4j connection") + + async def execute_query( + self, + cypher: str, + parameters: Optional[Dict[str, Any]] = None + ) -> List[Dict[str, Any]]: + """ + Execute Cypher query and return results. + + Args: + cypher: Cypher query string + parameters: Query parameters + + Returns: + List of result records as dictionaries + + Raises: + Exception: If driver not initialized or query fails + """ + if not self._driver: + await self.connect() + + async with self._driver.session() as session: + result = await session.run(cypher, parameters or {}) + records = await result.data() + return records + + async def execute_write( + self, + cypher: str, + parameters: Optional[Dict[str, Any]] = None + ) -> List[Dict[str, Any]]: + """ + Execute write transaction with automatic retry. + + Args: + cypher: Cypher query string + parameters: Query parameters + + Returns: + List of result records as dictionaries + """ + if not self._driver: + await self.connect() + + async def write_tx(tx): + result = await tx.run(cypher, parameters or {}) + return await result.data() + + async with self._driver.session() as session: + return await session.execute_write(write_tx) + + # Multi-tenancy helpers + + def get_user_label(self, user: str) -> str: + """ + Get Neo4j label for user's documents. + + Args: + user: User identifier + + Returns: + Neo4j label string (e.g., "User_Jpmschweitzer_Document") + """ + return get_neo4j_user_label(user) + + # Document node operations + + async def create_document_node( + self, + user: str, + doc_id: str, + properties: Dict[str, Any] + ) -> Optional[Dict[str, Any]]: + """ + Create document node with user label. + + Node structure: + (doc:Document:User_{user}_Document { + id: "doc_123", + source: "github", + repository: "anthropic-cookbook", + path: "skills/citation/guide.md", + title: "Citation Guide", + created_at: timestamp(), + updated_at: timestamp(), + content_hash: "sha256:..." + }) + + Args: + user: User identifier + doc_id: Unique document ID + properties: Document properties + + Returns: + Created node properties or None on failure + """ + user_label = self.get_user_label(user) + + # Ensure required properties + properties["id"] = doc_id + if "created_at" not in properties: + properties["created_at"] = "timestamp()" + + cypher = f""" + CREATE (doc:Document:{user_label}) + SET doc = $properties + SET doc.created_at = timestamp() + SET doc.updated_at = timestamp() + RETURN doc + """ + + try: + result = await self.execute_write(cypher, {"properties": properties}) + return result[0]["doc"] if result else None + except Exception as e: + logger.error(f"Failed to create document node: {e}", exc_info=True) + return None + + async def get_document_node( + self, + user: str, + doc_id: str + ) -> Optional[Dict[str, Any]]: + """ + Get document node by ID. + + Args: + user: User identifier + doc_id: Document ID + + Returns: + Document node properties or None if not found + """ + user_label = self.get_user_label(user) + + cypher = f""" + MATCH (doc:Document:{user_label} {{id: $doc_id}}) + RETURN doc + """ + + try: + result = await self.execute_query(cypher, {"doc_id": doc_id}) + return result[0]["doc"] if result else None + except Exception as e: + logger.error(f"Failed to get document node: {e}", exc_info=True) + return None + + async def delete_document_node( + self, + user: str, + doc_id: str + ) -> bool: + """ + Delete document node and all its relationships. + + Args: + user: User identifier + doc_id: Document ID + + Returns: + True if deleted, False otherwise + """ + user_label = self.get_user_label(user) + + cypher = f""" + MATCH (doc:Document:{user_label} {{id: $doc_id}}) + DETACH DELETE doc + RETURN count(doc) as deleted + """ + + try: + result = await self.execute_write(cypher, {"doc_id": doc_id}) + return result[0]["deleted"] > 0 if result else False + except Exception as e: + logger.error(f"Failed to delete document node: {e}", exc_info=True) + return False + + async def find_similar_documents( + self, + user: str, + doc_ids: List[str], + max_depth: int = 2, + limit: int = 20 + ) -> List[Dict[str, Any]]: + """ + Find documents similar to given docs via graph traversal. + + Uses: Shared concepts, shared entities, citation links. + + Args: + user: User identifier + doc_ids: List of source document IDs + max_depth: Maximum traversal depth + limit: Maximum results to return + + Returns: + List of similar documents with connection strength + """ + user_label = self.get_user_label(user) + + cypher = f""" + MATCH (source:Document:{user_label}) + WHERE source.id IN $doc_ids + MATCH (source)-[*1..{max_depth}]-(related:Document:{user_label}) + WHERE related.id <> source.id AND NOT related.id IN $doc_ids + WITH related, count(*) as connection_strength + ORDER BY connection_strength DESC + LIMIT $limit + RETURN related, connection_strength + """ + + try: + result = await self.execute_query( + cypher, + {"doc_ids": doc_ids, "limit": limit} + ) + return result + except Exception as e: + logger.error(f"Failed to find similar documents: {e}", exc_info=True) + return [] + + async def list_user_documents( + self, + user: str, + limit: int = 100, + offset: int = 0 + ) -> List[Dict[str, Any]]: + """ + List all documents for a user. + + Args: + user: User identifier + limit: Maximum results to return + offset: Number of results to skip + + Returns: + List of document nodes + """ + user_label = self.get_user_label(user) + + cypher = f""" + MATCH (doc:Document:{user_label}) + RETURN doc + ORDER BY doc.created_at DESC + SKIP $offset + LIMIT $limit + """ + + try: + result = await self.execute_query( + cypher, + {"offset": offset, "limit": limit} + ) + return [r["doc"] for r in result] + except Exception as e: + logger.error(f"Failed to list documents: {e}", exc_info=True) + return [] + + # Concept/entity operations + + async def create_concept_node( + self, + concept_name: str, + concept_type: str, + properties: Optional[Dict[str, Any]] = None + ) -> Optional[Dict[str, Any]]: + """ + Create or update concept node. + + Args: + concept_name: Concept name + concept_type: Concept type (Technique, Tool, Pattern, etc.) + properties: Additional properties + + Returns: + Concept node properties + """ + cypher = """ + MERGE (concept:Concept {name: $name}) + ON CREATE SET + concept.type = $type, + concept.first_seen = timestamp(), + concept.mention_count = 1 + ON MATCH SET + concept.mention_count = concept.mention_count + 1 + SET concept += $properties + RETURN concept + """ + + try: + result = await self.execute_write( + cypher, + { + "name": concept_name, + "type": concept_type, + "properties": properties or {} + } + ) + return result[0]["concept"] if result else None + except Exception as e: + logger.error(f"Failed to create concept node: {e}", exc_info=True) + return None + + async def link_document_to_concept( + self, + user: str, + doc_id: str, + concept_name: str + ) -> bool: + """ + Create MENTIONS relationship between document and concept. + + Args: + user: User identifier + doc_id: Document ID + concept_name: Concept name + + Returns: + True if link created + """ + user_label = self.get_user_label(user) + + cypher = f""" + MATCH (doc:Document:{user_label} {{id: $doc_id}}) + MATCH (concept:Concept {{name: $concept_name}}) + MERGE (doc)-[r:MENTIONS]->(concept) + ON CREATE SET r.count = 1 + ON MATCH SET r.count = r.count + 1 + RETURN r + """ + + try: + result = await self.execute_write( + cypher, + {"doc_id": doc_id, "concept_name": concept_name} + ) + return len(result) > 0 + except Exception as e: + logger.error(f"Failed to link document to concept: {e}", exc_info=True) + return False diff --git a/services/library-desk/src/clients/ollama_client.py b/services/library-desk/src/clients/ollama_client.py new file mode 100644 index 0000000..e166826 --- /dev/null +++ b/services/library-desk/src/clients/ollama_client.py @@ -0,0 +1,366 @@ +""" +Ollama client for embeddings generation. + +Provides async embedding operations via Ollama API: +- Text embedding generation +- Batch embedding support +- Model management +""" + +import httpx +from typing import List, Dict, Any, Optional +import logging + +logger = logging.getLogger(__name__) + + +class OllamaClient: + """ + Ollama API client for embeddings. + + Documentation: https://github.com/ollama/ollama/blob/main/docs/api.md + Default model: nomic-embed-text (768-dimensional embeddings) + """ + + def __init__(self, base_url: str, model: str = "nomic-embed-text"): + """ + Initialize Ollama client. + + Args: + base_url: Ollama server URL (e.g., "http://ollama:11434") + model: Embedding model name (default: "nomic-embed-text") + """ + self.base_url = base_url.rstrip("/") + self.model = model + self.embeddings_url = f"{self.base_url}/api/embeddings" + self.generate_url = f"{self.base_url}/api/generate" + self.tags_url = f"{self.base_url}/api/tags" + self.client = httpx.AsyncClient(timeout=120.0) # Embeddings can be slow + logger.info(f"Initialized Ollama client: {base_url}, model: {model}") + + async def close(self): + """Close HTTP client""" + await self.client.aclose() + + async def embed(self, text: str) -> Optional[List[float]]: + """ + Generate embedding for single text. + + Args: + text: Text to embed + + Returns: + Embedding vector (768-dimensional for nomic-embed-text) or None on failure + + Example: + >>> embedding = await client.embed("Hello world") + >>> len(embedding) + 768 + """ + try: + payload = { + "model": self.model, + "prompt": text + } + + response = await self.client.post( + self.embeddings_url, + json=payload + ) + response.raise_for_status() + data = response.json() + + embedding = data.get("embedding") + if not embedding: + logger.error(f"No embedding in response: {data}") + return None + + return embedding + + except httpx.HTTPStatusError as e: + logger.error(f"HTTP error: {e.response.status_code} - {e.response.text}") + return None + except Exception as e: + logger.error(f"Embedding failed: {e}", exc_info=True) + return None + + async def embed_batch( + self, + texts: List[str], + show_progress: bool = False + ) -> List[Optional[List[float]]]: + """ + Generate embeddings for multiple texts. + + Args: + texts: List of texts to embed + show_progress: Log progress for large batches + + Returns: + List of embedding vectors (same order as input) + None entries for texts that failed to embed + + Example: + >>> texts = ["Hello", "World", "Test"] + >>> embeddings = await client.embed_batch(texts) + >>> len(embeddings) + 3 + """ + embeddings = [] + + for i, text in enumerate(texts): + if show_progress and i % 10 == 0: + logger.info(f"Embedding progress: {i}/{len(texts)}") + + embedding = await self.embed(text) + embeddings.append(embedding) + + if show_progress: + logger.info(f"Embedding complete: {len(embeddings)}/{len(texts)}") + + return embeddings + + async def embed_batch_filtered( + self, + texts: List[str], + show_progress: bool = False + ) -> List[List[float]]: + """ + Generate embeddings for multiple texts, filtering out failures. + + Args: + texts: List of texts to embed + show_progress: Log progress for large batches + + Returns: + List of successful embedding vectors (may be shorter than input) + + Example: + >>> texts = ["Hello", "World", "Test"] + >>> embeddings = await client.embed_batch_filtered(texts) + >>> all(e is not None for e in embeddings) + True + """ + all_embeddings = await self.embed_batch(texts, show_progress) + return [e for e in all_embeddings if e is not None] + + async def embed_documents( + self, + documents: List[Dict[str, Any]], + content_field: str = "content", + show_progress: bool = False + ) -> List[Dict[str, Any]]: + """ + Embed documents with metadata preservation. + + Args: + documents: List of document dictionaries + content_field: Field name containing text to embed + show_progress: Log progress for large batches + + Returns: + List of documents with added "embedding" field + + Example: + >>> docs = [ + ... {"content": "Hello world", "id": 1}, + ... {"content": "Test doc", "id": 2} + ... ] + >>> embedded = await client.embed_documents(docs) + >>> "embedding" in embedded[0] + True + """ + texts = [doc.get(content_field, "") for doc in documents] + embeddings = await self.embed_batch(texts, show_progress) + + results = [] + for doc, embedding in zip(documents, embeddings): + doc_copy = doc.copy() + doc_copy["embedding"] = embedding + results.append(doc_copy) + + return results + + async def get_embedding_dimension(self) -> Optional[int]: + """ + Get embedding dimension for current model. + + Returns: + Embedding dimension (e.g., 768 for nomic-embed-text) or None on failure + + Example: + >>> dim = await client.get_embedding_dimension() + >>> dim + 768 + """ + # Generate a test embedding to determine dimension + test_embedding = await self.embed("test") + if test_embedding: + return len(test_embedding) + return None + + async def list_models(self) -> List[Dict[str, Any]]: + """ + List available Ollama models. + + Returns: + List of model information dictionaries + + Example: + >>> models = await client.list_models() + >>> any(m["name"] == "nomic-embed-text" for m in models) + True + """ + try: + response = await self.client.get(self.tags_url) + response.raise_for_status() + data = response.json() + return data.get("models", []) + except Exception as e: + logger.error(f"Failed to list models: {e}") + return [] + + async def check_model_available(self, model_name: Optional[str] = None) -> bool: + """ + Check if a model is available. + + Args: + model_name: Model name to check (defaults to self.model) + + Returns: + True if model is available, False otherwise + """ + check_model = model_name or self.model + models = await self.list_models() + return any(m.get("name") == check_model for m in models) + + async def generate_text( + self, + prompt: str, + model: Optional[str] = None, + stream: bool = False + ) -> Optional[str]: + """ + Generate text completion (for non-embedding use cases). + + Args: + prompt: Input prompt + model: Model name (defaults to self.model) + stream: Enable streaming response + + Returns: + Generated text or None on failure + + Note: This is primarily for debugging/testing. Use specialized + LLM services for production text generation. + """ + try: + payload = { + "model": model or self.model, + "prompt": prompt, + "stream": stream + } + + response = await self.client.post( + self.generate_url, + json=payload + ) + response.raise_for_status() + + if stream: + # For streaming, return first chunk + # Full streaming implementation would need async generator + return response.text + else: + data = response.json() + return data.get("response") + + except Exception as e: + logger.error(f"Text generation failed: {e}") + return None + + async def health_check(self) -> bool: + """ + Check if Ollama server is reachable and model is available. + + Returns: + True if healthy, False otherwise + """ + try: + # Check server is up + response = await self.client.get(self.tags_url, timeout=5.0) + response.raise_for_status() + + # Check model is available + models_available = await self.check_model_available() + if not models_available: + logger.warning(f"Model '{self.model}' not found in Ollama") + return False + + return True + + except Exception as e: + logger.error(f"Health check failed: {e}") + return False + + def estimate_tokens(self, text: str) -> int: + """ + Rough estimate of token count for text. + + Uses simple heuristic: ~4 characters per token. + + Args: + text: Input text + + Returns: + Estimated token count + """ + return len(text) // 4 + + def chunk_text_for_embedding( + self, + text: str, + max_tokens: int = 512, + overlap: int = 50 + ) -> List[str]: + """ + Chunk text into segments suitable for embedding. + + Args: + text: Input text + max_tokens: Maximum tokens per chunk + overlap: Token overlap between chunks + + Returns: + List of text chunks + + Example: + >>> chunks = client.chunk_text_for_embedding(long_text, max_tokens=512) + >>> all(client.estimate_tokens(c) <= 512 for c in chunks) + True + """ + # Convert tokens to approximate character count + max_chars = max_tokens * 4 + overlap_chars = overlap * 4 + + if len(text) <= max_chars: + return [text] + + chunks = [] + start = 0 + + while start < len(text): + end = start + max_chars + chunk = text[start:end] + + # Try to break at sentence boundary + if end < len(text): + last_period = chunk.rfind(". ") + if last_period > max_chars * 0.5: # Only break if > 50% through chunk + end = start + last_period + 1 + chunk = text[start:end] + + chunks.append(chunk.strip()) + start = end - overlap_chars + + return chunks diff --git a/services/library-desk/src/clients/qdrant_client.py b/services/library-desk/src/clients/qdrant_client.py new file mode 100644 index 0000000..ab05017 --- /dev/null +++ b/services/library-desk/src/clients/qdrant_client.py @@ -0,0 +1,412 @@ +""" +Qdrant client wrapper for Library Desk. + +Provides async vector operations with: +- Collection-per-user multi-tenancy +- Document chunk storage with embeddings +- Semantic search +- Similarity queries +""" + +from qdrant_client import QdrantClient +from qdrant_client.models import ( + Distance, VectorParams, PointStruct, + Filter, FieldCondition, MatchValue +) +from typing import List, Dict, Any, Optional +import uuid +import logging + +from src.core.multi_tenancy import get_qdrant_collection_name + +logger = logging.getLogger(__name__) + + +class QdrantClientWrapper: + """ + Qdrant client wrapper with multi-tenancy support. + + Pattern: Collection per user (from qdrant_memory.py) + Each user has isolated vector collection for their documents. + """ + + def __init__(self, url: str, embedding_dim: int = 768): + """ + Initialize Qdrant client. + + Args: + url: Qdrant server URL (e.g., "http://qdrant:6333") + embedding_dim: Vector embedding dimension (default 768 for nomic-embed-text) + """ + self.client = QdrantClient(url=url) + self.embedding_dim = embedding_dim + logger.info(f"Initialized Qdrant client: {url}") + + def get_collection_name(self, user: str) -> str: + """ + Get collection name for user. + + Args: + user: User identifier + + Returns: + Collection name (e.g., "library_desk_jpmschweitzer") + """ + return get_qdrant_collection_name(user) + + async def ensure_collection(self, user: str): + """ + Create user's collection if it doesn't exist. + + Args: + user: User identifier + """ + collection_name = self.get_collection_name(user) + + try: + collections = self.client.get_collections() + existing = [c.name for c in collections.collections] + + if collection_name not in existing: + logger.info(f"Creating Qdrant collection: {collection_name}") + self.client.create_collection( + collection_name=collection_name, + vectors_config=VectorParams( + size=self.embedding_dim, + distance=Distance.COSINE + ) + ) + logger.info(f"Created collection: {collection_name}") + except Exception as e: + logger.error(f"Error ensuring collection: {e}", exc_info=True) + raise + + async def upsert_document_chunks( + self, + user: str, + doc_id: str, + chunks: List[Dict[str, Any]], + embeddings: List[List[float]] + ) -> int: + """ + Upsert document chunks with embeddings. + + Point structure: + { + id: "doc_123_chunk_0", + vector: [...], + payload: { + doc_id: "doc_123", + chunk_index: 0, + content: "text content", + metadata: {...} + } + } + + Args: + user: User identifier + doc_id: Document ID + chunks: List of chunk dictionaries with content + embeddings: List of embedding vectors + + Returns: + Number of chunks upserted + + Raises: + ValueError: If chunks and embeddings length mismatch + """ + if len(chunks) != len(embeddings): + raise ValueError( + f"Chunks ({len(chunks)}) and embeddings ({len(embeddings)}) length mismatch" + ) + + collection_name = self.get_collection_name(user) + await self.ensure_collection(user) + + points = [] + for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)): + # Generate deterministic point ID + point_id = str(uuid.uuid5( + uuid.NAMESPACE_DNS, + f"{doc_id}_chunk_{i}" + )) + + points.append(PointStruct( + id=point_id, + vector=embedding, + payload={ + "doc_id": doc_id, + "chunk_index": i, + "content": chunk.get("content", ""), + "metadata": chunk.get("metadata", {}) + } + )) + + try: + self.client.upsert( + collection_name=collection_name, + points=points + ) + logger.info(f"Upserted {len(points)} chunks for {doc_id}") + return len(points) + except Exception as e: + logger.error(f"Failed to upsert chunks: {e}", exc_info=True) + raise + + async def search( + self, + user: str, + query_vector: List[float], + limit: int = 10, + score_threshold: float = 0.7, + filter_dict: Optional[Dict[str, Any]] = None + ) -> List[Dict[str, Any]]: + """ + Semantic search in user's collection. + + Args: + user: User identifier + query_vector: Query embedding vector + limit: Maximum results to return + score_threshold: Minimum similarity score (0.0-1.0) + filter_dict: Optional payload filters + + Returns: + List of matches with scores and payloads + + Example: + results = await client.search( + user="jpmschweitzer", + query_vector=[0.1, 0.2, ...], + limit=5, + filter_dict={"doc_id": "doc_123"} + ) + """ + collection_name = self.get_collection_name(user) + + # Build filter if provided + query_filter = None + if filter_dict: + conditions = [] + for key, value in filter_dict.items(): + conditions.append( + FieldCondition(key=key, match=MatchValue(value=value)) + ) + query_filter = Filter(must=conditions) + + try: + results = self.client.search( + collection_name=collection_name, + query_vector=query_vector, + limit=limit, + score_threshold=score_threshold, + query_filter=query_filter, + with_payload=True + ) + + return [ + { + "id": str(result.id), + "score": result.score, + "doc_id": result.payload["doc_id"], + "chunk_index": result.payload["chunk_index"], + "content": result.payload["content"], + "metadata": result.payload.get("metadata", {}) + } + for result in results + ] + except Exception as e: + logger.error(f"Search failed: {e}", exc_info=True) + return [] + + async def delete_document(self, user: str, doc_id: str) -> bool: + """ + Delete all chunks for a document. + + Args: + user: User identifier + doc_id: Document ID + + Returns: + True if deleted successfully + """ + collection_name = self.get_collection_name(user) + + try: + self.client.delete( + collection_name=collection_name, + points_selector=Filter( + must=[ + FieldCondition( + key="doc_id", + match=MatchValue(value=doc_id) + ) + ] + ) + ) + logger.info(f"Deleted chunks for {doc_id}") + return True + except Exception as e: + logger.error(f"Failed to delete document: {e}", exc_info=True) + return False + + async def get_document_chunks( + self, + user: str, + doc_id: str + ) -> List[Dict[str, Any]]: + """ + Get all chunks for a document. + + Args: + user: User identifier + doc_id: Document ID + + Returns: + List of chunks with content and metadata + """ + collection_name = self.get_collection_name(user) + + try: + # Scroll through points with doc_id filter + points, _ = self.client.scroll( + collection_name=collection_name, + scroll_filter=Filter( + must=[ + FieldCondition( + key="doc_id", + match=MatchValue(value=doc_id) + ) + ] + ), + limit=1000, + with_payload=True, + with_vectors=False + ) + + return [ + { + "id": str(point.id), + "chunk_index": point.payload["chunk_index"], + "content": point.payload["content"], + "metadata": point.payload.get("metadata", {}) + } + for point in points + ] + except Exception as e: + logger.error(f"Failed to get document chunks: {e}", exc_info=True) + return [] + + async def count_documents(self, user: str) -> int: + """ + Count total number of unique documents in user's collection. + + Args: + user: User identifier + + Returns: + Number of unique documents + """ + collection_name = self.get_collection_name(user) + + try: + # Get collection info + collection_info = self.client.get_collection(collection_name) + # This gives total points, not unique docs + # For unique docs, would need to aggregate by doc_id + return collection_info.points_count + except Exception as e: + logger.error(f"Failed to count documents: {e}", exc_info=True) + return 0 + + async def delete_collection(self, user: str) -> bool: + """ + Delete user's entire collection. + + Warning: This removes all data for the user! + + Args: + user: User identifier + + Returns: + True if deleted successfully + """ + collection_name = self.get_collection_name(user) + + try: + self.client.delete_collection(collection_name) + logger.warning(f"Deleted collection: {collection_name}") + return True + except Exception as e: + logger.error(f"Failed to delete collection: {e}", exc_info=True) + return False + + async def find_similar_chunks( + self, + user: str, + doc_id: str, + limit: int = 10 + ) -> List[Dict[str, Any]]: + """ + Find chunks similar to those in a given document. + + Strategy: Get all chunks from doc, use their vectors to find similar chunks. + + Args: + user: User identifier + doc_id: Source document ID + limit: Maximum results per chunk + + Returns: + List of similar chunks from other documents + """ + collection_name = self.get_collection_name(user) + + try: + # Get source document chunks with vectors + source_points, _ = self.client.scroll( + collection_name=collection_name, + scroll_filter=Filter( + must=[ + FieldCondition( + key="doc_id", + match=MatchValue(value=doc_id) + ) + ] + ), + limit=10, # Sample first 10 chunks + with_payload=True, + with_vectors=True + ) + + if not source_points: + return [] + + # Search using first chunk's vector + # (could aggregate multiple chunks for better results) + first_vector = source_points[0].vector + + results = self.client.search( + collection_name=collection_name, + query_vector=first_vector, + limit=limit * 2, # Get more to filter out same doc + with_payload=True + ) + + # Filter out chunks from same document + similar = [ + { + "id": str(r.id), + "score": r.score, + "doc_id": r.payload["doc_id"], + "content": r.payload["content"] + } + for r in results + if r.payload["doc_id"] != doc_id + ] + + return similar[:limit] + + except Exception as e: + logger.error(f"Failed to find similar chunks: {e}", exc_info=True) + return [] \ No newline at end of file diff --git a/services/library-desk/src/clients/searxng_client.py b/services/library-desk/src/clients/searxng_client.py new file mode 100644 index 0000000..2c4b82e --- /dev/null +++ b/services/library-desk/src/clients/searxng_client.py @@ -0,0 +1,332 @@ +""" +SearXNG search client for Library Desk. + +Provides async web search via SearXNG instance: +- General web search +- Category-specific search +- Language filtering +- Result formatting +""" + +import httpx +from typing import Optional, List, Dict, Any +import logging + +logger = logging.getLogger(__name__) + + +class SearXNGClient: + """ + SearXNG search API client. + + Documentation: https://docs.searxng.org/dev/search_api.html + API Format: JSON (?format=json) + """ + + def __init__(self, base_url: str): + """ + Initialize SearXNG client. + + Args: + base_url: SearXNG base URL (e.g., "http://searxng:8080") + """ + self.base_url = base_url.rstrip("/") + self.search_url = f"{self.base_url}/search" + self.client = httpx.AsyncClient(timeout=30.0) + logger.info(f"Initialized SearXNG client: {base_url}") + + async def close(self): + """Close HTTP client""" + await self.client.aclose() + + async def search( + self, + query: str, + categories: Optional[List[str]] = None, + language: str = "en", + time_range: Optional[str] = None, + safesearch: int = 0, + pageno: int = 1, + limit: int = 10 + ) -> Dict[str, Any]: + """ + Search via SearXNG. + + Args: + query: Search query string + categories: List of categories (e.g., ["general", "images", "news"]) + language: Language code (e.g., "en", "fr", "auto") + time_range: Time range filter ("day", "week", "month", "year", None for all) + safesearch: Safe search level (0=off, 1=moderate, 2=strict) + pageno: Page number (starts at 1) + limit: Maximum results to return + + Returns: + Search results dictionary with: + - query: Original query + - results: List of result items + - number_of_results: Total results found + - suggestions: Query suggestions + + Raises: + Exception: If search fails + """ + params = { + "q": query, + "format": "json", + "language": language, + "safesearch": safesearch, + "pageno": pageno + } + + # Add optional parameters + if categories: + params["categories"] = ",".join(categories) + if time_range: + params["time_range"] = time_range + + try: + response = await self.client.get(self.search_url, params=params) + response.raise_for_status() + data = response.json() + + # Limit results if requested + results = data.get("results", []) + if limit: + results = results[:limit] + + return { + "query": data.get("query", query), + "results": results, + "number_of_results": data.get("number_of_results", 0), + "suggestions": data.get("suggestions", []), + "answers": data.get("answers", []), + "infoboxes": data.get("infoboxes", []) + } + + except httpx.HTTPStatusError as e: + logger.error(f"HTTP error: {e.response.status_code} - {e.response.text}") + raise Exception(f"SearXNG search failed: {e.response.status_code}") + except Exception as e: + logger.error(f"Search failed: {e}", exc_info=True) + raise + + async def search_general( + self, + query: str, + language: str = "en", + limit: int = 10 + ) -> List[Dict[str, Any]]: + """ + Simplified general web search. + + Args: + query: Search query + language: Language code + limit: Maximum results + + Returns: + List of search results with: + - title: Page title + - url: Page URL + - content: Page description/snippet + - engine: Search engine that provided result + - score: Relevance score (if available) + """ + try: + result = await self.search( + query=query, + categories=["general"], + language=language, + limit=limit + ) + return result.get("results", []) + except Exception as e: + logger.error(f"General search failed: {e}") + return [] + + async def search_with_context( + self, + query: str, + context: Optional[str] = None, + language: str = "en", + limit: int = 10 + ) -> List[Dict[str, Any]]: + """ + Search with additional context appended to query. + + Useful for RAG queries where you want to enhance search with context. + + Args: + query: Primary search query + context: Additional context to append + language: Language code + limit: Maximum results + + Returns: + List of search results + """ + enhanced_query = f"{query} {context}" if context else query + return await self.search_general(enhanced_query, language, limit) + + async def search_documentation( + self, + query: str, + limit: int = 10 + ) -> List[Dict[str, Any]]: + """ + Search for technical documentation. + + Enhances query with "documentation" and filters for technical content. + + Args: + query: Search query (e.g., "FastAPI") + limit: Maximum results + + Returns: + List of documentation search results + """ + enhanced_query = f"{query} documentation" + return await self.search_general(enhanced_query, limit=limit) + + async def search_recent( + self, + query: str, + time_range: str = "month", + limit: int = 10 + ) -> List[Dict[str, Any]]: + """ + Search for recent content only. + + Args: + query: Search query + time_range: Time range ("day", "week", "month", "year") + limit: Maximum results + + Returns: + List of recent search results + """ + try: + result = await self.search( + query=query, + categories=["general"], + time_range=time_range, + limit=limit + ) + return result.get("results", []) + except Exception as e: + logger.error(f"Recent search failed: {e}") + return [] + + async def get_suggestions(self, query: str) -> List[str]: + """ + Get search query suggestions. + + Args: + query: Partial query + + Returns: + List of suggested queries + """ + try: + result = await self.search(query=query, limit=1) + return result.get("suggestions", []) + except Exception as e: + logger.error(f"Failed to get suggestions: {e}") + return [] + + async def search_images( + self, + query: str, + safesearch: int = 1, + limit: int = 10 + ) -> List[Dict[str, Any]]: + """ + Search for images. + + Args: + query: Search query + safesearch: Safe search level (0=off, 1=moderate, 2=strict) + limit: Maximum results + + Returns: + List of image results with: + - title: Image title + - url: Image URL + - thumbnail_src: Thumbnail URL + - img_src: Full image URL + - content: Description + """ + try: + result = await self.search( + query=query, + categories=["images"], + safesearch=safesearch, + limit=limit + ) + return result.get("results", []) + except Exception as e: + logger.error(f"Image search failed: {e}") + return [] + + async def search_news( + self, + query: str, + time_range: str = "week", + limit: int = 10 + ) -> List[Dict[str, Any]]: + """ + Search for news articles. + + Args: + query: Search query + time_range: Time range ("day", "week", "month") + limit: Maximum results + + Returns: + List of news results + """ + try: + result = await self.search( + query=query, + categories=["news"], + time_range=time_range, + limit=limit + ) + return result.get("results", []) + except Exception as e: + logger.error(f"News search failed: {e}") + return [] + + async def format_results_for_rag( + self, + results: List[Dict[str, Any]], + max_snippet_length: int = 500 + ) -> List[Dict[str, str]]: + """ + Format search results for RAG context. + + Extracts relevant fields and truncates content. + + Args: + results: Raw search results + max_snippet_length: Maximum length of content snippet + + Returns: + Formatted results with title, url, snippet + """ + formatted = [] + for result in results: + content = result.get("content", "") + if len(content) > max_snippet_length: + content = content[:max_snippet_length] + "..." + + formatted.append({ + "title": result.get("title", ""), + "url": result.get("url", ""), + "snippet": content, + "engine": result.get("engine", ""), + "score": result.get("score", 0.0) + }) + + return formatted diff --git a/services/library-desk/src/clients/wikijs_client.py b/services/library-desk/src/clients/wikijs_client.py new file mode 100644 index 0000000..86597b8 --- /dev/null +++ b/services/library-desk/src/clients/wikijs_client.py @@ -0,0 +1,502 @@ +""" +Wiki.js GraphQL client for Library Desk. + +Provides async Wiki.js operations via GraphQL API: +- Page CRUD (create, read, update, delete) +- Search and listing +- Tag management +- Multi-tenancy via path namespaces +""" + +import httpx +from typing import Optional, List, Dict, Any +import logging + +logger = logging.getLogger(__name__) + + +class WikiJSClient: + """ + Wiki.js GraphQL API client. + + Documentation: https://docs.requarks.io/dev/api + Authentication: Bearer token in Authorization header + """ + + def __init__(self, base_url: str, api_key: str): + """ + Initialize Wiki.js client. + + Args: + base_url: Wiki.js base URL (e.g., "http://wiki:3000") + api_key: Wiki.js API key (from Admin → API Access) + """ + self.base_url = base_url.rstrip("/") + self.graphql_url = f"{self.base_url}/graphql" + self.headers = { + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json" + } + self.client = httpx.AsyncClient(timeout=30.0) + logger.info(f"Initialized Wiki.js client: {base_url}") + + async def close(self): + """Close HTTP client""" + await self.client.aclose() + + async def _execute_query( + self, + query: str, + variables: Optional[Dict] = None + ) -> Dict[str, Any]: + """ + Execute GraphQL query. + + Args: + query: GraphQL query string + variables: Query variables + + Returns: + Query result data + + Raises: + Exception: If query fails or returns errors + """ + payload = { + "query": query, + "variables": variables or {} + } + + try: + response = await self.client.post( + self.graphql_url, + headers=self.headers, + json=payload + ) + response.raise_for_status() + result = response.json() + + if "errors" in result: + logger.error(f"GraphQL errors: {result['errors']}") + raise Exception(f"GraphQL errors: {result['errors']}") + + return result.get("data", {}) + + except httpx.HTTPStatusError as e: + logger.error(f"HTTP error: {e.response.status_code} - {e.response.text}") + raise + except Exception as e: + logger.error(f"GraphQL query failed: {e}", exc_info=True) + raise + + async def list_pages( + self, + path_prefix: str = "", + tags: Optional[List[str]] = None, + limit: int = 50 + ) -> List[Dict[str, Any]]: + """ + List pages with optional filtering. + + Multi-tenancy: Use path_prefix to filter by user namespace. + + Args: + path_prefix: Filter by path prefix (e.g., "/users/jpmschweitzer") + tags: Filter by tags (e.g., ["projects"]) + limit: Maximum results + + Returns: + List of page objects + """ + query = """ + query ListPages($limit: Int, $orderBy: PageOrderBy) { + pages { + list(limit: $limit, orderBy: $orderBy) { + id + path + title + description + tags + createdAt + updatedAt + isPublished + } + } + } + """ + + variables = { + "limit": limit, + "orderBy": "TITLE" + } + + data = await self._execute_query(query, variables) + pages = data.get("pages", {}).get("list", []) + + # Filter by path prefix (client-side if API doesn't support) + if path_prefix: + pages = [p for p in pages if p["path"].startswith(path_prefix)] + + # Filter by tags (dossiers) + if tags: + pages = [ + p for p in pages + if any(tag in (p.get("tags") or []) for tag in tags) + ] + + return pages + + async def get_page(self, page_id: int) -> Optional[Dict[str, Any]]: + """ + Get single page by ID. + + Args: + page_id: Page ID + + Returns: + Page object or None if not found + """ + query = """ + query GetPage($id: Int!) { + pages { + single(id: $id) { + id + path + title + description + content + tags + createdAt + updatedAt + isPublished + editor + } + } + } + """ + + try: + data = await self._execute_query(query, {"id": page_id}) + return data.get("pages", {}).get("single") + except Exception as e: + logger.error(f"Failed to get page {page_id}: {e}") + return None + + async def create_page( + self, + path: str, + title: str, + content: str, + description: str = "", + tags: Optional[List[str]] = None, + is_published: bool = True, + editor: str = "markdown" + ) -> Dict[str, Any]: + """ + Create new page. + + Multi-tenancy: Ensure path starts with user namespace. + + Args: + path: Page path (e.g., "/users/jpmschweitzer/projects/library-desk") + title: Page title + content: Page content (markdown) + description: Short description + tags: List of tags (for dossier organization) + is_published: Whether page is published + editor: Editor type (markdown, wysiwyg, etc.) + + Returns: + Created page object + + Raises: + Exception: If creation fails + """ + mutation = """ + mutation CreatePage( + $content: String!, + $description: String!, + $editor: String!, + $isPublished: Boolean!, + $locale: String!, + $path: String!, + $tags: [String]!, + $title: String! + ) { + pages { + create( + content: $content, + description: $description, + editor: $editor, + isPublished: $isPublished, + locale: $locale, + path: $path, + tags: $tags, + title: $title + ) { + responseResult { + succeeded + errorCode + message + } + page { + id + path + title + } + } + } + } + """ + + variables = { + "path": path, + "title": title, + "content": content, + "description": description, + "tags": tags or [], + "isPublished": is_published, + "editor": editor, + "locale": "en" + } + + data = await self._execute_query(mutation, variables) + result = data.get("pages", {}).get("create", {}) + + if not result.get("responseResult", {}).get("succeeded"): + error = result.get("responseResult", {}) + raise Exception(f"Failed to create page: {error}") + + logger.info(f"Created page: {path}") + return result.get("page") + + async def update_page( + self, + page_id: int, + content: Optional[str] = None, + title: Optional[str] = None, + description: Optional[str] = None, + tags: Optional[List[str]] = None + ) -> Dict[str, Any]: + """ + Update existing page. + + Args: + page_id: Page ID + content: New content (optional) + title: New title (optional) + description: New description (optional) + tags: New tags (optional) + + Returns: + Updated page object + + Raises: + Exception: If update fails + """ + mutation = """ + mutation UpdatePage( + $id: Int!, + $content: String, + $title: String, + $description: String, + $tags: [String] + ) { + pages { + update( + id: $id, + content: $content, + title: $title, + description: $description, + tags: $tags + ) { + responseResult { + succeeded + errorCode + message + } + page { + id + updatedAt + } + } + } + } + """ + + variables = {"id": page_id} + if content is not None: + variables["content"] = content + if title is not None: + variables["title"] = title + if description is not None: + variables["description"] = description + if tags is not None: + variables["tags"] = tags + + data = await self._execute_query(mutation, variables) + result = data.get("pages", {}).get("update", {}) + + if not result.get("responseResult", {}).get("succeeded"): + error = result.get("responseResult", {}) + raise Exception(f"Failed to update page: {error}") + + logger.info(f"Updated page: {page_id}") + return result.get("page") + + async def delete_page(self, page_id: int): + """ + Delete page. + + Args: + page_id: Page ID + + Raises: + Exception: If deletion fails + """ + mutation = """ + mutation DeletePage($id: Int!) { + pages { + delete(id: $id) { + responseResult { + succeeded + errorCode + message + } + } + } + } + """ + + data = await self._execute_query(mutation, {"id": page_id}) + result = data.get("pages", {}).get("delete", {}) + + if not result.get("responseResult", {}).get("succeeded"): + error = result.get("responseResult", {}) + raise Exception(f"Failed to delete page: {error}") + + logger.info(f"Deleted page: {page_id}") + + async def search_pages( + self, + query: str, + path_prefix: Optional[str] = None + ) -> List[Dict[str, Any]]: + """ + Search pages by content. + + Args: + query: Search query + path_prefix: Optional path prefix filter + + Returns: + List of matching pages + """ + gql_query = """ + query SearchPages($query: String!) { + pages { + search(query: $query) { + results { + id + path + title + description + } + } + } + } + """ + + data = await self._execute_query(gql_query, {"query": query}) + results = data.get("pages", {}).get("search", {}).get("results", []) + + # Filter by path prefix if provided + if path_prefix: + results = [r for r in results if r["path"].startswith(path_prefix)] + + return results + + async def get_page_tree(self, path: str = "/") -> List[Dict[str, Any]]: + """ + Get page tree structure. + + Args: + path: Root path + + Returns: + Tree structure of pages + """ + query = """ + query GetPageTree($parent: Int, $mode: String!) { + pages { + tree(parent: $parent, mode: $mode) { + id + path + title + isFolder + pageId + } + } + } + """ + + try: + data = await self._execute_query( + query, + {"parent": 0, "mode": "all"} + ) + return data.get("pages", {}).get("tree", []) + except Exception as e: + logger.error(f"Failed to get page tree: {e}") + return [] + + async def move_page( + self, + page_id: int, + new_path: str, + locale: str = "en" + ) -> bool: + """ + Move/rename page. + + Args: + page_id: Page ID + new_path: New page path + locale: Page locale + + Returns: + True if successful + """ + mutation = """ + mutation MovePage($id: Int!, $destinationPath: String!, $destinationLocale: String!) { + pages { + move(id: $id, destinationPath: $destinationPath, destinationLocale: $destinationLocale) { + responseResult { + succeeded + errorCode + message + } + } + } + } + """ + + try: + data = await self._execute_query( + mutation, + { + "id": page_id, + "destinationPath": new_path, + "destinationLocale": locale + } + ) + result = data.get("pages", {}).get("move", {}) + success = result.get("responseResult", {}).get("succeeded", False) + + if success: + logger.info(f"Moved page {page_id} to {new_path}") + return success + + except Exception as e: + logger.error(f"Failed to move page: {e}") + return False diff --git a/services/library-desk/src/config.py b/services/library-desk/src/config.py index 8ea459e..bedd96b 100644 --- a/services/library-desk/src/config.py +++ b/services/library-desk/src/config.py @@ -41,10 +41,10 @@ class Settings(BaseSettings): ollama_url: str = Field(default="http://ollama:11434", description="Ollama URL") ollama_model: str = Field(default="nomic-embed-text", description="Ollama embedding model") - # Redis Configuration + # Redis Configuration (for job tracking - separate DB from wiki) redis_host: str = Field(default="redis-shared", description="Redis host") redis_port: int = Field(default=6379, description="Redis port") - redis_db: int = Field(default=2, description="Redis database number") + redis_db: int = Field(default=4, description="Redis database number (4 for library-desk jobs)") # Application app_name: str = Field(default="Library Desk", description="Application name") diff --git a/services/library-desk/src/core/__init__.py b/services/library-desk/src/core/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/services/library-desk/src/core/dependencies.py b/services/library-desk/src/core/dependencies.py new file mode 100644 index 0000000..7a36f15 --- /dev/null +++ b/services/library-desk/src/core/dependencies.py @@ -0,0 +1,286 @@ +""" +Dependency injection for Library Desk. + +Provides FastAPI dependencies for service clients with: +- Singleton pattern via @lru_cache +- Lazy initialization +- Proper lifecycle management +- Type aliases for clean endpoint signatures +""" + +from functools import lru_cache +from typing import Annotated +from fastapi import Depends +import logging + +from src.config import Settings, get_settings +from src.clients.neo4j_client import Neo4jClient +from src.clients.qdrant_client import QdrantClientWrapper +from src.clients.wikijs_client import WikiJSClient +from src.clients.searxng_client import SearXNGClient +from src.clients.ollama_client import OllamaClient + +logger = logging.getLogger(__name__) + + +# Settings dependency +SettingsDep = Annotated[Settings, Depends(get_settings)] + + +# Client factory functions with @lru_cache for singletons +@lru_cache +def get_neo4j_client() -> Neo4jClient: + """ + Get Neo4j client singleton. + + Returns: + Initialized Neo4j client (not yet connected) + + Note: Call client.connect() during app startup + """ + settings = get_settings() + client = Neo4jClient( + uri=settings.neo4j_uri, + user=settings.neo4j_user, + password=settings.neo4j_password + ) + logger.debug("Created Neo4j client instance") + return client + + +@lru_cache +def get_qdrant_client() -> QdrantClientWrapper: + """ + Get Qdrant client singleton. + + Returns: + Initialized Qdrant client + + Note: Collections are created lazily per-user + """ + settings = get_settings() + client = QdrantClientWrapper( + url=settings.qdrant_url, + embedding_dim=768 # nomic-embed-text default + ) + logger.debug("Created Qdrant client instance") + return client + + +@lru_cache +def get_wikijs_client() -> WikiJSClient: + """ + Get Wiki.js client singleton. + + Returns: + Initialized Wiki.js GraphQL client + """ + settings = get_settings() + client = WikiJSClient( + base_url=settings.wikijs_url, + api_key=settings.wikijs_api_key + ) + logger.debug("Created Wiki.js client instance") + return client + + +@lru_cache +def get_searxng_client() -> SearXNGClient: + """ + Get SearXNG client singleton. + + Returns: + Initialized SearXNG search client + """ + settings = get_settings() + client = SearXNGClient(base_url=settings.searxng_url) + logger.debug("Created SearXNG client instance") + return client + + +@lru_cache +def get_ollama_client() -> OllamaClient: + """ + Get Ollama client singleton. + + Returns: + Initialized Ollama embeddings client + """ + settings = get_settings() + client = OllamaClient( + base_url=settings.ollama_url, + model=settings.ollama_model + ) + logger.debug("Created Ollama client instance") + return client + + +# Type aliases for FastAPI endpoint dependencies +# Usage: def my_endpoint(neo4j: Neo4jDep): +Neo4jDep = Annotated[Neo4jClient, Depends(get_neo4j_client)] +QdrantDep = Annotated[QdrantClientWrapper, Depends(get_qdrant_client)] +WikiJSDep = Annotated[WikiJSClient, Depends(get_wikijs_client)] +SearXNGDep = Annotated[SearXNGClient, Depends(get_searxng_client)] +OllamaDep = Annotated[OllamaClient, Depends(get_ollama_client)] + + +# Lifecycle management functions +async def startup_clients(): + """ + Initialize all service clients at application startup. + + Should be called in FastAPI lifespan or startup event. + + Performs: + - Neo4j connection pool initialization + - Neo4j connectivity verification + - Ollama model availability check + """ + logger.info("Starting up service clients...") + + # Initialize Neo4j connection pool + neo4j = get_neo4j_client() + try: + await neo4j.connect() + logger.info("✓ Neo4j connected") + except Exception as e: + logger.error(f"✗ Neo4j connection failed: {e}") + # Don't fail startup - allow degraded operation + pass + + # Check Ollama availability + ollama = get_ollama_client() + try: + is_healthy = await ollama.health_check() + if is_healthy: + logger.info(f"✓ Ollama ready (model: {ollama.model})") + else: + logger.warning(f"✗ Ollama model '{ollama.model}' not available") + except Exception as e: + logger.error(f"✗ Ollama health check failed: {e}") + pass + + # Qdrant, Wiki.js, SearXNG are lazy-initialized + logger.info("Service clients startup complete") + + +async def shutdown_clients(): + """ + Cleanup all service clients at application shutdown. + + Should be called in FastAPI lifespan or shutdown event. + + Performs: + - Close Neo4j connection pool + - Close HTTP clients + """ + logger.info("Shutting down service clients...") + + # Close Neo4j driver + neo4j = get_neo4j_client() + try: + await neo4j.close() + logger.info("✓ Neo4j closed") + except Exception as e: + logger.error(f"Error closing Neo4j: {e}") + + # Close HTTP clients + clients_to_close = [ + ("Wiki.js", get_wikijs_client()), + ("SearXNG", get_searxng_client()), + ("Ollama", get_ollama_client()) + ] + + for name, client in clients_to_close: + try: + await client.close() + logger.info(f"✓ {name} client closed") + except Exception as e: + logger.error(f"Error closing {name} client: {e}") + + logger.info("Service clients shutdown complete") + + +async def check_service_health() -> dict: + """ + Check health of all service clients. + + Returns: + Dictionary with health status of each service: + { + "neo4j": bool, + "qdrant": bool, + "wikijs": bool, + "searxng": bool, + "ollama": bool + } + + Usage: + >>> health = await check_service_health() + >>> health["neo4j"] + True + """ + health = {} + + # Neo4j + try: + neo4j = get_neo4j_client() + # Simple query to check connectivity + await neo4j.execute_query("RETURN 1 as test", {}) + health["neo4j"] = True + except Exception as e: + logger.error(f"Neo4j health check failed: {e}") + health["neo4j"] = False + + # Qdrant + try: + qdrant = get_qdrant_client() + # Check if we can list collections + collections = qdrant.client.get_collections() + health["qdrant"] = True + except Exception as e: + logger.error(f"Qdrant health check failed: {e}") + health["qdrant"] = False + + # Wiki.js + try: + wikijs = get_wikijs_client() + # Try a simple query (list pages with limit 1) + await wikijs.list_pages(limit=1) + health["wikijs"] = True + except Exception as e: + logger.error(f"Wiki.js health check failed: {e}") + health["wikijs"] = False + + # SearXNG + try: + searxng = get_searxng_client() + # Try a simple search + await searxng.search_general("test", limit=1) + health["searxng"] = True + except Exception as e: + logger.error(f"SearXNG health check failed: {e}") + health["searxng"] = False + + # Ollama + try: + ollama = get_ollama_client() + is_healthy = await ollama.health_check() + health["ollama"] = is_healthy + except Exception as e: + logger.error(f"Ollama health check failed: {e}") + health["ollama"] = False + + return health + + +# Utility: Get default user from settings or multi_tenancy +def get_default_user() -> str: + """ + Get default user for operations. + + Returns: + Default user identifier + """ + from src.core.multi_tenancy import DEFAULT_USER + return DEFAULT_USER diff --git a/services/library-desk/src/core/multi_tenancy.py b/services/library-desk/src/core/multi_tenancy.py new file mode 100644 index 0000000..228f5d3 --- /dev/null +++ b/services/library-desk/src/core/multi_tenancy.py @@ -0,0 +1,182 @@ +""" +Multi-tenancy helpers for Library Desk. + +Provides utilities for user namespace management across: +- Wiki.js (path-based namespaces) +- Neo4j (user-specific labels) +- Qdrant (collection per user) +""" + +import re + +# Default user for all operations +DEFAULT_USER = "jpmschweitzer" + + +def sanitize_user_id(user_id: str) -> str: + """ + Sanitize user ID for use in collection names, labels, and paths. + + Converts special characters to underscores and ensures alphanumeric safety. + + Args: + user_id: Raw user identifier (email, username, etc.) + + Returns: + Sanitized user ID safe for use in identifiers + + Examples: + >>> sanitize_user_id("john@example.com") + 'john_at_example_com' + >>> sanitize_user_id("user.name") + 'user_name' + >>> sanitize_user_id("User Name") + 'user_name' + """ + sanitized = user_id.lower() + + # Convert @ to _at_ + sanitized = sanitized.replace("@", "_at_") + + # Convert dots to underscores + sanitized = sanitized.replace(".", "_") + + # Replace any non-alphanumeric characters with underscores + sanitized = re.sub(r'[^a-z0-9_]', '_', sanitized) + + # Remove consecutive underscores + sanitized = re.sub(r'_+', '_', sanitized) + + # Remove leading/trailing underscores + sanitized = sanitized.strip('_') + + return sanitized + + +def get_qdrant_collection_name(user_id: str) -> str: + """ + Get Qdrant collection name for user. + + Pattern: library_desk_{sanitized_user_id} + + Args: + user_id: User identifier + + Returns: + Qdrant collection name + + Examples: + >>> get_qdrant_collection_name("jpmschweitzer") + 'library_desk_jpmschweitzer' + >>> get_qdrant_collection_name("john@example.com") + 'library_desk_john_at_example_com' + """ + sanitized = sanitize_user_id(user_id) + return f"library_desk_{sanitized}" + + +def get_wikijs_namespace(user_id: str) -> str: + """ + Get Wiki.js namespace (path prefix) for user. + + Pattern: /users/{sanitized_user_id} + + All wiki pages for a user will be under this namespace. + + Args: + user_id: User identifier + + Returns: + Wiki.js path prefix + + Examples: + >>> get_wikijs_namespace("jpmschweitzer") + '/users/jpmschweitzer' + >>> get_wikijs_namespace("john@example.com") + '/users/john_at_example_com' + """ + sanitized = sanitize_user_id(user_id) + return f"/users/{sanitized}" + + +def get_neo4j_user_label(user_id: str) -> str: + """ + Get Neo4j label for user's documents. + + Pattern: User_{Sanitized}_Document + + Uses title case for Neo4j label convention. + + Args: + user_id: User identifier + + Returns: + Neo4j label for user's document nodes + + Examples: + >>> get_neo4j_user_label("jpmschweitzer") + 'User_Jpmschweitzer_Document' + >>> get_neo4j_user_label("john@example.com") + 'User_John_At_Example_Com_Document' + """ + sanitized = sanitize_user_id(user_id) + # Title case each segment for Neo4j label convention + parts = sanitized.split('_') + titled = '_'.join(part.capitalize() for part in parts if part) + return f"User_{titled}_Document" + + +def validate_user_id(user_id: str) -> bool: + """ + Validate that a user ID is acceptable. + + Checks: + - Not empty + - Not too long (max 100 chars) + - Contains some alphanumeric characters + + Args: + user_id: User identifier to validate + + Returns: + True if valid, False otherwise + + Examples: + >>> validate_user_id("jpmschweitzer") + True + >>> validate_user_id("") + False + >>> validate_user_id("a" * 101) + False + """ + if not user_id or len(user_id) > 100: + return False + + # Must contain at least one alphanumeric character + if not re.search(r'[a-zA-Z0-9]', user_id): + return False + + return True + + +def is_path_in_user_namespace(path: str, user_id: str) -> bool: + """ + Check if a Wiki.js path belongs to user's namespace. + + Args: + path: Wiki.js page path + user_id: User identifier + + Returns: + True if path is in user's namespace + + Examples: + >>> is_path_in_user_namespace("/users/jpmschweitzer/projects", "jpmschweitzer") + True + >>> is_path_in_user_namespace("/users/other/projects", "jpmschweitzer") + False + >>> is_path_in_user_namespace("/public/docs", "jpmschweitzer") + False + """ + namespace = get_wikijs_namespace(user_id) + return path.startswith(namespace) diff --git a/services/library-desk/src/jobs/__init__.py b/services/library-desk/src/jobs/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/services/library-desk/src/jobs/job_manager.py b/services/library-desk/src/jobs/job_manager.py new file mode 100644 index 0000000..895ac55 --- /dev/null +++ b/services/library-desk/src/jobs/job_manager.py @@ -0,0 +1,426 @@ +""" +Redis-based job tracking for Library Desk. + +Provides background job management with: +- Job creation and status tracking +- Progress updates +- Result storage +- Auto-expiration after 24 hours +- User-scoped job queries +""" + +import redis.asyncio as redis +import json +import uuid +from datetime import datetime, timezone +from typing import Optional, Dict, Any, List +from enum import Enum +import logging + +logger = logging.getLogger(__name__) + + +class JobStatus(str, Enum): + """Job status enumeration.""" + QUEUED = "queued" + PROCESSING = "processing" + COMPLETED = "completed" + FAILED = "failed" + CANCELLED = "cancelled" + + +class JobType(str, Enum): + """Job type enumeration.""" + DOCUMENT_INGESTION = "document_ingestion" + BATCH_INGESTION = "batch_ingestion" + GRAPH_UPDATE = "graph_update" + VECTOR_UPDATE = "vector_update" + EMBEDDINGS_GENERATION = "embeddings_generation" + + +class JobManager: + """ + Redis-based job tracking manager. + + Key structure: + - library:job:{job_id} - Job data (JSON, TTL 24h) + - library:user_jobs:{user} - Set of job IDs for user + - library:active_jobs - Set of active (non-completed) job IDs + + Job data format: + { + "job_id": "uuid", + "job_type": "document_ingestion", + "status": "processing", + "user": "jpmschweitzer", + "parameters": {...}, + "result": {...}, + "error": "error message", + "progress": 50, + "created_at": "2024-01-15T10:30:00Z", + "started_at": "2024-01-15T10:30:05Z", + "completed_at": "2024-01-15T10:35:00Z" + } + """ + + def __init__(self, redis_url: str): + """ + Initialize job manager. + + Args: + redis_url: Redis connection URL (e.g., "redis://redis-shared:6379/4") + """ + self.redis_url = redis_url + self._redis: Optional[redis.Redis] = None + self.ttl = 86400 # 24 hours + logger.info(f"Initialized JobManager: {redis_url}") + + async def connect(self): + """Initialize Redis connection.""" + if not self._redis: + self._redis = await redis.from_url( + self.redis_url, + encoding="utf-8", + decode_responses=True + ) + logger.info("Connected to Redis for job tracking") + + async def close(self): + """Close Redis connection.""" + if self._redis: + await self._redis.close() + self._redis = None + logger.info("Closed Redis connection") + + def _job_key(self, job_id: str) -> str: + """Get Redis key for job.""" + return f"library:job:{job_id}" + + def _user_jobs_key(self, user: str) -> str: + """Get Redis key for user's jobs.""" + return f"library:user_jobs:{user}" + + def _active_jobs_key(self) -> str: + """Get Redis key for active jobs.""" + return "library:active_jobs" + + async def create_job( + self, + job_type: JobType, + user: str, + parameters: Dict[str, Any] + ) -> str: + """ + Create new job. + + Args: + job_type: Type of job + user: User identifier + parameters: Job parameters + + Returns: + Job ID (UUID) + + Example: + >>> job_id = await manager.create_job( + ... JobType.DOCUMENT_INGESTION, + ... "jpmschweitzer", + ... {"doc_url": "https://example.com/doc.pdf"} + ... ) + """ + if not self._redis: + await self.connect() + + job_id = str(uuid.uuid4()) + job_data = { + "job_id": job_id, + "job_type": job_type.value, + "status": JobStatus.QUEUED.value, + "user": user, + "parameters": parameters, + "result": None, + "error": None, + "progress": 0, + "created_at": datetime.now(timezone.utc).isoformat() + } + + # Store job data + await self._redis.setex( + self._job_key(job_id), + self.ttl, + json.dumps(job_data) + ) + + # Add to user's job set + await self._redis.sadd(self._user_jobs_key(user), job_id) + await self._redis.expire(self._user_jobs_key(user), self.ttl) + + # Add to active jobs set + await self._redis.sadd(self._active_jobs_key(), job_id) + + logger.info(f"Created job {job_id}: {job_type.value} for user {user}") + return job_id + + async def get_job(self, job_id: str) -> Optional[Dict[str, Any]]: + """ + Get job by ID. + + Args: + job_id: Job ID + + Returns: + Job data dictionary or None if not found + + Example: + >>> job = await manager.get_job(job_id) + >>> job["status"] + 'processing' + """ + if not self._redis: + await self.connect() + + job_json = await self._redis.get(self._job_key(job_id)) + if not job_json: + return None + + return json.loads(job_json) + + async def update_job_status( + self, + job_id: str, + status: JobStatus, + progress: Optional[int] = None, + result: Optional[Dict[str, Any]] = None, + error: Optional[str] = None + ): + """ + Update job status. + + Args: + job_id: Job ID + status: New status + progress: Progress percentage (0-100) + result: Result data (for completed jobs) + error: Error message (for failed jobs) + + Example: + >>> await manager.update_job_status( + ... job_id, + ... JobStatus.PROCESSING, + ... progress=50 + ... ) + """ + if not self._redis: + await self.connect() + + job = await self.get_job(job_id) + if not job: + logger.error(f"Job {job_id} not found") + return + + # Update fields + job["status"] = status.value + if progress is not None: + job["progress"] = progress + if result is not None: + job["result"] = result + if error is not None: + job["error"] = error + + # Update timestamps + now = datetime.now(timezone.utc).isoformat() + if status == JobStatus.PROCESSING and not job.get("started_at"): + job["started_at"] = now + if status in (JobStatus.COMPLETED, JobStatus.FAILED, JobStatus.CANCELLED): + job["completed_at"] = now + + # Save updated job + await self._redis.setex( + self._job_key(job_id), + self.ttl, + json.dumps(job) + ) + + # Remove from active jobs if completed + if status in (JobStatus.COMPLETED, JobStatus.FAILED, JobStatus.CANCELLED): + await self._redis.srem(self._active_jobs_key(), job_id) + + logger.info(f"Updated job {job_id}: status={status.value}, progress={progress}") + + async def get_user_jobs( + self, + user: str, + limit: int = 50 + ) -> List[Dict[str, Any]]: + """ + Get all jobs for a user. + + Args: + user: User identifier + limit: Maximum jobs to return + + Returns: + List of job dictionaries (most recent first) + + Example: + >>> jobs = await manager.get_user_jobs("jpmschweitzer", limit=10) + >>> len(jobs) + 10 + """ + if not self._redis: + await self.connect() + + # Get job IDs for user + job_ids = await self._redis.smembers(self._user_jobs_key(user)) + if not job_ids: + return [] + + # Fetch job data + jobs = [] + for job_id in job_ids: + job = await self.get_job(job_id) + if job: + jobs.append(job) + + # Sort by created_at descending + jobs.sort(key=lambda j: j.get("created_at", ""), reverse=True) + + return jobs[:limit] + + async def get_active_jobs(self, limit: int = 100) -> List[Dict[str, Any]]: + """ + Get all active (non-completed) jobs. + + Args: + limit: Maximum jobs to return + + Returns: + List of active job dictionaries + + Example: + >>> active = await manager.get_active_jobs() + >>> all(j["status"] in ["queued", "processing"] for j in active) + True + """ + if not self._redis: + await self.connect() + + job_ids = await self._redis.smembers(self._active_jobs_key()) + if not job_ids: + return [] + + jobs = [] + for job_id in job_ids: + job = await self.get_job(job_id) + if job: + jobs.append(job) + + return jobs[:limit] + + async def cancel_job(self, job_id: str): + """ + Cancel a job. + + Args: + job_id: Job ID + + Note: This only marks the job as cancelled. The actual job + worker must check status and stop processing. + """ + await self.update_job_status( + job_id, + JobStatus.CANCELLED + ) + logger.info(f"Cancelled job {job_id}") + + async def delete_job(self, job_id: str, user: str): + """ + Delete a job. + + Args: + job_id: Job ID + user: User identifier (for authorization) + + Returns: + True if deleted, False if not found or unauthorized + """ + if not self._redis: + await self.connect() + + job = await self.get_job(job_id) + if not job: + return False + + # Check user authorization + if job.get("user") != user: + logger.warning(f"User {user} attempted to delete job {job_id} owned by {job.get('user')}") + return False + + # Delete job data + await self._redis.delete(self._job_key(job_id)) + + # Remove from user's jobs + await self._redis.srem(self._user_jobs_key(user), job_id) + + # Remove from active jobs + await self._redis.srem(self._active_jobs_key(), job_id) + + logger.info(f"Deleted job {job_id}") + return True + + async def cleanup_expired_jobs(self): + """ + Clean up expired jobs from sets. + + Redis will auto-expire job data, but set memberships need manual cleanup. + Should be called periodically (e.g., hourly). + """ + if not self._redis: + await self.connect() + + # Cleanup active jobs set + job_ids = await self._redis.smembers(self._active_jobs_key()) + for job_id in job_ids: + exists = await self._redis.exists(self._job_key(job_id)) + if not exists: + await self._redis.srem(self._active_jobs_key(), job_id) + + logger.info("Cleaned up expired jobs") + + async def get_job_stats(self, user: Optional[str] = None) -> Dict[str, int]: + """ + Get job statistics. + + Args: + user: Optional user filter + + Returns: + Statistics dictionary: + { + "total": 100, + "queued": 5, + "processing": 10, + "completed": 80, + "failed": 5 + } + """ + if user: + jobs = await self.get_user_jobs(user, limit=1000) + else: + jobs = await self.get_active_jobs(limit=1000) + + stats = { + "total": len(jobs), + "queued": 0, + "processing": 0, + "completed": 0, + "failed": 0, + "cancelled": 0 + } + + for job in jobs: + status = job.get("status", "") + if status in stats: + stats[status] += 1 + + return stats diff --git a/services/library-desk/src/main.py b/services/library-desk/src/main.py index 367d1ac..7146b62 100644 --- a/services/library-desk/src/main.py +++ b/services/library-desk/src/main.py @@ -69,9 +69,7 @@ class HealthResponse(BaseModel): status: str app_name: str version: str - neo4j: str - qdrant: str - wiki: str + services: Dict[str, Any] class StatsResponse(BaseModel): @@ -98,13 +96,42 @@ async def health(settings: Settings = Depends(get_settings)) -> HealthResponse: Health check endpoint. Returns status of all connected services. """ + from src.core.dependencies import check_service_health + + # Check service connectivity + service_health = await check_service_health() + + # Overall status is healthy if at least Neo4j and Qdrant are up + all_healthy = service_health.get("neo4j", False) and service_health.get("qdrant", False) + overall_status = "healthy" if all_healthy else "degraded" + return HealthResponse( - status="healthy", + status=overall_status, app_name=settings.app_name, version=settings.app_version, - neo4j=settings.neo4j_uri, - qdrant=settings.qdrant_url, - wiki=settings.wikijs_url + services={ + "neo4j": { + "url": settings.neo4j_uri, + "healthy": service_health.get("neo4j", False) + }, + "qdrant": { + "url": settings.qdrant_url, + "healthy": service_health.get("qdrant", False) + }, + "wikijs": { + "url": settings.wikijs_url, + "healthy": service_health.get("wikijs", False) + }, + "searxng": { + "url": settings.searxng_url, + "healthy": service_health.get("searxng", False) + }, + "ollama": { + "url": settings.ollama_url, + "model": settings.ollama_model, + "healthy": service_health.get("ollama", False) + } + } ) @@ -128,6 +155,114 @@ async def stats( ) +# Ingestion endpoints (for Scheduler integration) +@app.post("/ingest/document", tags=["Ingestion"]) +async def ingest_document( + document: Dict[str, Any], + api_key: str = Depends(verify_api_key) +) -> Dict[str, Any]: + """ + Ingest a single document for indexing. + Used by The Scheduler to add mirrored documentation to the knowledge base. + + Expected fields: + - source: str (e.g., "github", "gitea") + - repository: str (e.g., "anthropic-cookbook") + - path: str (file path) + - content: str (document content) + - metadata: dict (commit, author, tags, etc.) + + TODO: Implement document ingestion pipeline: + 1. Chunk content + 2. Generate embeddings (Ollama) + 3. Extract entities (NLP) + 4. Index in Qdrant + 5. Create graph nodes/relationships in Neo4j + """ + return { + "message": "Document ingestion not yet implemented", + "document_id": f"doc_{document.get('path', 'unknown')}", + "status": "stub" + } + + +@app.post("/ingest/batch", tags=["Ingestion"]) +async def batch_ingest( + batch: Dict[str, Any], + api_key: str = Depends(verify_api_key) +) -> Dict[str, Any]: + """ + Ingest multiple documents in a batch. + More efficient than individual ingestion for large syncs. + + TODO: Implement batch processing with task queue + """ + document_count = len(batch.get("documents", [])) + return { + "message": "Batch ingestion not yet implemented", + "batch_id": "batch_stub", + "total_documents": document_count, + "status": "stub" + } + + +@app.post("/ingest/check-updates", tags=["Ingestion"]) +async def check_updates( + documents: Dict[str, Any], + api_key: str = Depends(verify_api_key) +) -> Dict[str, Any]: + """ + Check which documents need updating based on content hashes. + Used by Scheduler to determine what changed since last sync. + + TODO: Implement update detection: + 1. Query existing documents by path + 2. Compare content hashes + 3. Return list of updates needed + """ + return { + "message": "Update checking not yet implemented", + "updates_needed": [], + "up_to_date": [], + "new_documents": [] + } + + +@app.get("/ingest/status/{document_id}", tags=["Ingestion"]) +async def get_ingestion_status( + document_id: str, + api_key: str = Depends(verify_api_key) +) -> Dict[str, Any]: + """ + Get processing status for a document. + + TODO: Implement status tracking + """ + return { + "message": "Status tracking not yet implemented", + "document_id": document_id, + "status": "unknown" + } + + +@app.get("/ingest/repo-status/{repository}", tags=["Ingestion"]) +async def get_repo_status( + repository: str, + api_key: str = Depends(verify_api_key) +) -> Dict[str, Any]: + """ + Get indexing status for an entire repository. + + TODO: Implement repository-level statistics + """ + return { + "message": "Repository status not yet implemented", + "repository": repository, + "total_documents": 0, + "indexed_documents": 0 + } + + # Query endpoints (stubs for future implementation) @app.post("/query/hybrid", tags=["Query"]) async def hybrid_query( @@ -180,20 +315,62 @@ async def graph_query( } +# Deduplication endpoints +@app.post("/deduplicate/check", tags=["Deduplication"]) +async def check_duplicates( + request: Dict[str, Any], + api_key: str = Depends(verify_api_key) +) -> Dict[str, Any]: + """ + Check for duplicate or highly similar documents. + Uses vector similarity and graph analysis. + + Expected fields: + - document_id: str + - similarity_threshold: float (default 0.85) + + TODO: Implement deduplication: + 1. Get document embedding from Qdrant + 2. Find similar vectors above threshold + 3. Check graph relationships + 4. Return candidates with similarity scores + """ + document_id = request.get("document_id") + threshold = request.get("similarity_threshold", 0.85) + + return { + "message": "Deduplication not yet implemented", + "document_id": document_id, + "threshold": threshold, + "duplicates": [], + "suggestions": None + } + + # Application lifecycle @app.on_event("startup") async def startup_event(): """Initialize connections and resources on startup.""" + from src.core.dependencies import startup_clients + settings = get_settings() logger.info(f"Starting {settings.app_name} v{settings.app_version}") logger.info(f"Neo4j: {settings.neo4j_uri}") logger.info(f"Qdrant: {settings.qdrant_url}") logger.info(f"Wiki.js: {settings.wikijs_url}") - # TODO: Initialize database connections + logger.info(f"SearXNG: {settings.searxng_url}") + logger.info(f"Ollama: {settings.ollama_url}") + + # Initialize all service clients + await startup_clients() @app.on_event("shutdown") async def shutdown_event(): """Clean up resources on shutdown.""" + from src.core.dependencies import shutdown_clients + logger.info("Shutting down Library Desk API") - # TODO: Close database connections + + # Close all service clients + await shutdown_clients() diff --git a/services/library-desk/src/models/__init__.py b/services/library-desk/src/models/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/services/library-desk/src/routers/__init__.py b/services/library-desk/src/routers/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/services/library-desk/src/services/__init__.py b/services/library-desk/src/services/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/services/library-desk/tests/__init__.py b/services/library-desk/tests/__init__.py new file mode 100644 index 0000000..c0ad886 --- /dev/null +++ b/services/library-desk/tests/__init__.py @@ -0,0 +1 @@ +"""Tests for Library Desk service.""" diff --git a/services/library-desk/tests/conftest.py b/services/library-desk/tests/conftest.py new file mode 100644 index 0000000..050f3bb --- /dev/null +++ b/services/library-desk/tests/conftest.py @@ -0,0 +1,109 @@ +"""Pytest configuration and shared fixtures for Library Desk tests.""" + +import pytest +import pytest_asyncio +from typing import AsyncGenerator + +# Test configuration +pytest_plugins = ("pytest_asyncio",) + + +@pytest.fixture +def test_user() -> str: + """Default test user.""" + return "test_user" + + +@pytest.fixture +def neo4j_test_uri() -> str: + """Test Neo4j URI.""" + return "bolt://neo4j:7687" + + +@pytest.fixture +def neo4j_test_auth() -> tuple: + """Test Neo4j authentication.""" + return ("neo4j", "test_password") + + +@pytest.fixture +def qdrant_test_url() -> str: + """Test Qdrant URL.""" + return "http://qdrant:6333" + + +@pytest.fixture +def wikijs_test_config() -> dict: + """Test Wiki.js configuration.""" + return { + "base_url": "http://wiki:3000", + "api_key": "test_api_key" + } + + +@pytest.fixture +def searxng_test_url() -> str: + """Test SearXNG URL.""" + return "http://searxng:8080" + + +@pytest.fixture +def ollama_test_config() -> dict: + """Test Ollama configuration.""" + return { + "base_url": "http://ollama:11434", + "model": "nomic-embed-text" + } + + +@pytest.fixture +def redis_test_url() -> str: + """Test Redis URL.""" + return "redis://redis-shared:6379/4" + + +@pytest.fixture +def sample_document() -> dict: + """Sample document for testing.""" + return { + "id": "test_doc_1", + "title": "Test Document", + "content": "This is a test document for unit testing.", + "metadata": { + "source": "test", + "author": "test_user" + } + } + + +@pytest.fixture +def sample_chunks() -> list: + """Sample document chunks for testing.""" + return [ + { + "content": "First chunk of text.", + "metadata": {"chunk_index": 0} + }, + { + "content": "Second chunk of text.", + "metadata": {"chunk_index": 1} + }, + { + "content": "Third chunk of text.", + "metadata": {"chunk_index": 2} + } + ] + + +@pytest.fixture +def sample_embeddings() -> list: + """Sample embeddings for testing (768-dimensional for nomic-embed-text).""" + import random + random.seed(42) # Reproducible embeddings + + # Generate 3 sample 768-dimensional embeddings + return [ + [random.random() for _ in range(768)], + [random.random() for _ in range(768)], + [random.random() for _ in range(768)] + ] diff --git a/services/library-desk/tests/test_integration.py b/services/library-desk/tests/test_integration.py new file mode 100644 index 0000000..e02c83a --- /dev/null +++ b/services/library-desk/tests/test_integration.py @@ -0,0 +1,313 @@ +"""Integration tests for Library Desk service clients. + +These tests require actual service connectivity: +- Neo4j running at bolt://neo4j:7687 +- Qdrant running at http://qdrant:6333 +- Wiki.js running at http://wiki:3000 +- SearXNG running at http://searxng:8080 +- Ollama running at http://ollama:11434 + +Run with: pytest tests/test_integration.py -v +""" + +import pytest +import pytest_asyncio +from typing import AsyncGenerator + +from src.clients.neo4j_client import Neo4jClient +from src.clients.qdrant_client import QdrantClientWrapper +from src.clients.wikijs_client import WikiJSClient +from src.clients.searxng_client import SearXNGClient +from src.clients.ollama_client import OllamaClient +from src.jobs.job_manager import JobManager, JobType, JobStatus +from src.config import get_settings + + +@pytest.fixture +def settings(): + """Get application settings.""" + return get_settings() + + +@pytest_asyncio.fixture +async def neo4j_client(settings) -> AsyncGenerator[Neo4jClient, None]: + """Get connected Neo4j client.""" + client = Neo4jClient( + uri=settings.neo4j_uri, + user=settings.neo4j_user, + password=settings.neo4j_password + ) + await client.connect() + yield client + await client.close() + + +@pytest.fixture +def qdrant_client(settings) -> QdrantClientWrapper: + """Get Qdrant client.""" + return QdrantClientWrapper(url=settings.qdrant_url) + + +@pytest_asyncio.fixture +async def wikijs_client(settings) -> AsyncGenerator[WikiJSClient, None]: + """Get Wiki.js client.""" + client = WikiJSClient( + base_url=settings.wikijs_url, + api_key=settings.wikijs_api_key + ) + yield client + await client.close() + + +@pytest_asyncio.fixture +async def searxng_client(settings) -> AsyncGenerator[SearXNGClient, None]: + """Get SearXNG client.""" + client = SearXNGClient(base_url=settings.searxng_url) + yield client + await client.close() + + +@pytest_asyncio.fixture +async def ollama_client(settings) -> AsyncGenerator[OllamaClient, None]: + """Get Ollama client.""" + client = OllamaClient( + base_url=settings.ollama_url, + model=settings.ollama_model + ) + yield client + await client.close() + + +@pytest_asyncio.fixture +async def job_manager(settings) -> AsyncGenerator[JobManager, None]: + """Get job manager.""" + manager = JobManager(redis_url=settings.redis_url) + await manager.connect() + yield manager + await manager.close() + + +class TestNeo4jIntegration: + """Test Neo4j connectivity and basic operations.""" + + @pytest.mark.asyncio + async def test_connection(self, neo4j_client): + """Test Neo4j connection.""" + result = await neo4j_client.execute_query("RETURN 1 as test", {}) + assert result + assert result[0]["test"] == 1 + + @pytest.mark.asyncio + async def test_create_and_get_document(self, neo4j_client, test_user): + """Test creating and retrieving a document node.""" + doc_id = "test_doc_integration" + + # Create document + doc = await neo4j_client.create_document_node( + user=test_user, + doc_id=doc_id, + properties={ + "title": "Integration Test Document", + "source": "test" + } + ) + assert doc is not None + assert doc["id"] == doc_id + + # Retrieve document + retrieved = await neo4j_client.get_document_node(test_user, doc_id) + assert retrieved is not None + assert retrieved["id"] == doc_id + assert retrieved["title"] == "Integration Test Document" + + # Cleanup + deleted = await neo4j_client.delete_document_node(test_user, doc_id) + assert deleted is True + + +class TestQdrantIntegration: + """Test Qdrant connectivity and basic operations.""" + + @pytest.mark.asyncio + async def test_collection_creation(self, qdrant_client, test_user): + """Test creating a collection.""" + await qdrant_client.ensure_collection(test_user) + + collection_name = qdrant_client.get_collection_name(test_user) + collections = qdrant_client.client.get_collections() + collection_names = [c.name for c in collections.collections] + + assert collection_name in collection_names + + @pytest.mark.asyncio + async def test_upsert_and_search(self, qdrant_client, test_user): + """Test upserting and searching chunks.""" + await qdrant_client.ensure_collection(test_user) + + # Create test chunks with 768-dimensional embeddings + chunks = [ + {"content": "Test chunk 1", "metadata": {}}, + {"content": "Test chunk 2", "metadata": {}} + ] + embeddings = [ + [0.1] * 768, + [0.2] * 768 + ] + + # Upsert + count = await qdrant_client.upsert_document_chunks( + user=test_user, + doc_id="test_doc_qdrant", + chunks=chunks, + embeddings=embeddings + ) + assert count == 2 + + # Search + results = await qdrant_client.search( + user=test_user, + query_vector=[0.1] * 768, + limit=5, + score_threshold=0.0 + ) + assert len(results) > 0 + + # Cleanup + deleted = await qdrant_client.delete_document(test_user, "test_doc_qdrant") + assert deleted is True + + +class TestWikiJSIntegration: + """Test Wiki.js connectivity and basic operations.""" + + @pytest.mark.asyncio + async def test_list_pages(self, wikijs_client): + """Test listing pages.""" + pages = await wikijs_client.list_pages(limit=10) + assert isinstance(pages, list) + # May be empty if wiki is new + + @pytest.mark.asyncio + async def test_search_pages(self, wikijs_client): + """Test searching pages.""" + results = await wikijs_client.search_pages("test") + assert isinstance(results, list) + + +class TestSearXNGIntegration: + """Test SearXNG connectivity and search.""" + + @pytest.mark.asyncio + async def test_general_search(self, searxng_client): + """Test general web search.""" + results = await searxng_client.search_general("python programming", limit=5) + assert isinstance(results, list) + if results: + assert "title" in results[0] + assert "url" in results[0] + + @pytest.mark.asyncio + async def test_search_with_suggestions(self, searxng_client): + """Test getting search suggestions.""" + suggestions = await searxng_client.get_suggestions("pytho") + assert isinstance(suggestions, list) + + +class TestOllamaIntegration: + """Test Ollama connectivity and embeddings.""" + + @pytest.mark.asyncio + async def test_health_check(self, ollama_client): + """Test Ollama health check.""" + is_healthy = await ollama_client.health_check() + # May be False if model not pulled + assert isinstance(is_healthy, bool) + + @pytest.mark.asyncio + async def test_list_models(self, ollama_client): + """Test listing available models.""" + models = await ollama_client.list_models() + assert isinstance(models, list) + + @pytest.mark.asyncio + @pytest.mark.skipif( + True, # Skip by default as embeddings can be slow + reason="Embedding generation is slow - enable manually if needed" + ) + async def test_generate_embedding(self, ollama_client): + """Test generating a single embedding.""" + embedding = await ollama_client.embed("test text") + if embedding: + assert isinstance(embedding, list) + assert len(embedding) > 0 + + +class TestJobManagerIntegration: + """Test Job Manager with Redis.""" + + @pytest.mark.asyncio + async def test_create_and_get_job(self, job_manager, test_user): + """Test creating and retrieving a job.""" + # Create job + job_id = await job_manager.create_job( + job_type=JobType.DOCUMENT_INGESTION, + user=test_user, + parameters={"doc_url": "https://example.com/doc.pdf"} + ) + assert job_id + + # Get job + job = await job_manager.get_job(job_id) + assert job is not None + assert job["job_id"] == job_id + assert job["user"] == test_user + assert job["status"] == JobStatus.QUEUED.value + + # Update job + await job_manager.update_job_status( + job_id, + JobStatus.PROCESSING, + progress=50 + ) + + updated_job = await job_manager.get_job(job_id) + assert updated_job["status"] == JobStatus.PROCESSING.value + assert updated_job["progress"] == 50 + + # Cleanup + deleted = await job_manager.delete_job(job_id, test_user) + assert deleted is True + + +class TestDependencyInjection: + """Test dependency injection and lifecycle management.""" + + @pytest.mark.asyncio + async def test_startup_clients(self): + """Test client startup.""" + from src.core.dependencies import startup_clients + + # Should not raise exceptions + await startup_clients() + + @pytest.mark.asyncio + async def test_check_service_health(self): + """Test health check for all services.""" + from src.core.dependencies import check_service_health + + health = await check_service_health() + + assert isinstance(health, dict) + assert "neo4j" in health + assert "qdrant" in health + assert "wikijs" in health + assert "searxng" in health + assert "ollama" in health + + @pytest.mark.asyncio + async def test_shutdown_clients(self): + """Test client shutdown.""" + from src.core.dependencies import shutdown_clients + + # Should not raise exceptions + await shutdown_clients() diff --git a/services/library-desk/tests/test_multi_tenancy.py b/services/library-desk/tests/test_multi_tenancy.py new file mode 100644 index 0000000..d4e277b --- /dev/null +++ b/services/library-desk/tests/test_multi_tenancy.py @@ -0,0 +1,115 @@ +"""Tests for multi-tenancy helpers.""" + +import pytest +from src.core.multi_tenancy import ( + sanitize_user_id, + get_qdrant_collection_name, + get_wikijs_namespace, + get_neo4j_user_label, + validate_user_id, + is_path_in_user_namespace, + DEFAULT_USER +) + + +class TestSanitizeUserId: + """Test user ID sanitization.""" + + def test_lowercase_conversion(self): + assert sanitize_user_id("JohnDoe") == "johndoe" + + def test_email_conversion(self): + assert sanitize_user_id("john@example.com") == "john_at_example_com" + + def test_dot_conversion(self): + assert sanitize_user_id("john.doe") == "john_doe" + + def test_space_conversion(self): + assert sanitize_user_id("John Doe") == "john_doe" + + def test_special_chars_removal(self): + assert sanitize_user_id("john-doe!") == "john_doe" + + def test_consecutive_underscores(self): + assert sanitize_user_id("john__doe") == "john_doe" + + def test_leading_trailing_underscores(self): + assert sanitize_user_id("_john_") == "john" + + +class TestQdrantCollectionName: + """Test Qdrant collection name generation.""" + + def test_simple_user(self): + assert get_qdrant_collection_name("jpmschweitzer") == "library_desk_jpmschweitzer" + + def test_email_user(self): + assert get_qdrant_collection_name("john@example.com") == "library_desk_john_at_example_com" + + def test_default_user(self): + assert get_qdrant_collection_name(DEFAULT_USER) == f"library_desk_{DEFAULT_USER}" + + +class TestWikijsNamespace: + """Test Wiki.js namespace generation.""" + + def test_simple_user(self): + assert get_wikijs_namespace("jpmschweitzer") == "/users/jpmschweitzer" + + def test_email_user(self): + assert get_wikijs_namespace("john@example.com") == "/users/john_at_example_com" + + def test_starts_with_slash(self): + namespace = get_wikijs_namespace("testuser") + assert namespace.startswith("/") + + +class TestNeo4jUserLabel: + """Test Neo4j user label generation.""" + + def test_simple_user(self): + assert get_neo4j_user_label("jpmschweitzer") == "User_Jpmschweitzer_Document" + + def test_email_user(self): + result = get_neo4j_user_label("john@example.com") + # Should be title case + assert result == "User_John_At_Example_Com_Document" + + def test_title_case(self): + result = get_neo4j_user_label("john_doe") + assert result == "User_John_Doe_Document" + + +class TestValidateUserId: + """Test user ID validation.""" + + def test_valid_simple(self): + assert validate_user_id("jpmschweitzer") is True + + def test_valid_email(self): + assert validate_user_id("john@example.com") is True + + def test_empty_invalid(self): + assert validate_user_id("") is False + + def test_too_long_invalid(self): + assert validate_user_id("a" * 101) is False + + def test_no_alphanumeric_invalid(self): + assert validate_user_id("___") is False + + +class TestPathInNamespace: + """Test path namespace checking.""" + + def test_path_in_namespace(self): + assert is_path_in_user_namespace("/users/jpmschweitzer/projects", "jpmschweitzer") is True + + def test_path_not_in_namespace(self): + assert is_path_in_user_namespace("/users/other/projects", "jpmschweitzer") is False + + def test_public_path_not_in_namespace(self): + assert is_path_in_user_namespace("/public/docs", "jpmschweitzer") is False + + def test_root_path(self): + assert is_path_in_user_namespace("/users/test", "test") is True diff --git a/stacks/library-desk.yml b/stacks/library-desk.yml index 8339232..c23ff0a 100644 --- a/stacks/library-desk.yml +++ b/stacks/library-desk.yml @@ -66,7 +66,7 @@ services: # Redis Configuration - REDIS_HOST=redis-shared - REDIS_PORT=6379 - - REDIS_DB=2 + - REDIS_DB=4 # Python Configuration - PYTHONUNBUFFERED=1