diff --git a/CHANGELOG.md b/CHANGELOG.md index adbbd7f..39fa0cf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,36 @@ All notable changes to Library Desk will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [1.4.3] - 2025-12-24 + +### Changed + +- **Volatile Cache System Refactored to Vector Storage** + - Backend migrated from Redis to Qdrant for semantic search capability + - Data converted to natural language for embedding and semantic retrieval + - Collection naming: `volatile_{user}` for per-user isolation + - TTL implemented via `ttl_expiry` timestamp in vector payload + - Simplified endpoints: + - `GET /volatile/search?q=...` - Semantic search across volatile data + - `POST /volatile/store?namespace=...&key=...` - Store with query params + - `GET /volatile/{namespace}/{key}` - Get specific record + - `DELETE /volatile/{namespace}/{key}` - Delete record + - Removed namespace-specific URL patterns (simpler API for LLM tool use) + +### Added + +- **HybridRAG Volatile Integration** - Volatile cache now included in multi-source search + - Volatile results get priority boost in RRF fusion (current data ranks higher) + - New config options: `enable_volatile`, `volatile_limit` (default 1), `volatile_threshold` + - Timing breakdown includes `volatile_ms` +- **Volatile Cleanup Endpoint** - `POST /maintenance/cleanup/volatile` + - Purges expired records across all `volatile_*` collections + - Scheduler task for every 10 minutes recommended + - Returns per-collection cleanup counts +- **Natural Language Conversion** - Structured data converted for embedding + - Template-based conversion for each namespace (weather, news, financial, etc.) + - Fallback for custom namespaces + ## [1.4.2] - 2025-12-24 ### Added diff --git a/pyproject.toml b/pyproject.toml index 780a8eb..795e8fb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "library-desk" -version = "1.4.2" +version = "1.4.3" description = "Coordination service for The Library system - HybridRAG queries, document ingestion, entity extraction, and knowledge consolidation" readme = "README.md" requires-python = ">=3.12" diff --git a/src/clients/qdrant_client.py b/src/clients/qdrant_client.py index d153bf9..93e3d1c 100644 --- a/src/clients/qdrant_client.py +++ b/src/clients/qdrant_client.py @@ -11,7 +11,7 @@ Provides async vector operations with: from qdrant_client import QdrantClient from qdrant_client.models import ( Distance, VectorParams, PointStruct, - Filter, FieldCondition, MatchValue + Filter, FieldCondition, MatchValue, Range ) from typing import List, Dict, Any, Optional import uuid @@ -676,4 +676,134 @@ class QdrantClientWrapper: except Exception as e: logger.error(f"Failed to list collections: {e}", exc_info=True) + return [] + + # ========== Volatile Data Methods ========== + + async def search_with_expiry_filter( + self, + collection_name: str, + query_vector: List[float], + current_timestamp: int, + limit: int = 10, + score_threshold: float = 0.7 + ) -> List[Dict[str, Any]]: + """ + Search vectors filtering out expired records. + + Args: + collection_name: Collection name + query_vector: Query embedding vector + current_timestamp: Current time in milliseconds + limit: Maximum results + score_threshold: Minimum similarity score + + Returns: + List of non-expired search results + """ + # Filter: ttl_expiry > current_timestamp (not expired) + expiry_filter = Filter( + must=[ + FieldCondition( + key="ttl_expiry", + range=Range(gt=current_timestamp) + ) + ] + ) + + try: + response = self.client.query_points( + collection_name=collection_name, + query=query_vector, + limit=limit, + score_threshold=score_threshold, + query_filter=expiry_filter, + with_payload=True + ) + + return [ + { + "id": str(point.id), + "score": point.score, + "payload": dict(point.payload) + } + for point in response.points + ] + except Exception as e: + logger.error(f"Volatile search failed: {e}", exc_info=True) + return [] + + async def delete_expired_vectors( + self, + collection_name: str, + current_timestamp: int + ) -> int: + """ + Delete all vectors where ttl_expiry < current_timestamp. + + Args: + collection_name: Collection name + current_timestamp: Current time in milliseconds + + Returns: + Number of points deleted (approximate) + """ + # Filter: ttl_expiry < current_timestamp (expired) + expiry_filter = Filter( + must=[ + FieldCondition( + key="ttl_expiry", + range=Range(lt=current_timestamp) + ) + ] + ) + + try: + # First count how many will be deleted (scroll to count) + count = 0 + offset = None + while True: + points, next_offset = self.client.scroll( + collection_name=collection_name, + scroll_filter=expiry_filter, + limit=100, + offset=offset, + with_payload=False + ) + count += len(points) + if next_offset is None: + break + offset = next_offset + + if count == 0: + return 0 + + # Delete expired points + self.client.delete( + collection_name=collection_name, + points_selector=expiry_filter + ) + + logger.info(f"Deleted {count} expired vectors from {collection_name}") + return count + + except Exception as e: + logger.error(f"Failed to delete expired vectors: {e}", exc_info=True) + return 0 + + async def get_volatile_collections(self) -> List[str]: + """ + Get all volatile collections (prefixed with 'volatile_'). + + Returns: + List of volatile collection names + """ + try: + collections = self.client.get_collections() + return [ + c.name for c in collections.collections + if c.name.startswith("volatile_") + ] + except Exception as e: + logger.error(f"Failed to list volatile collections: {e}", exc_info=True) return [] \ No newline at end of file diff --git a/src/models/hybrid_rag.py b/src/models/hybrid_rag.py index 5102d61..bc2d7af 100644 --- a/src/models/hybrid_rag.py +++ b/src/models/hybrid_rag.py @@ -14,13 +14,16 @@ class HybridRAGConfig(BaseModel): vector_limit: int = Field(default=10, ge=1, le=50, description="Max vector results") graph_limit: int = Field(default=10, ge=1, le=50, description="Max graph results") web_limit: int = Field(default=5, ge=1, le=20, description="Max web results") + volatile_limit: int = Field(default=1, ge=1, le=5, description="Max volatile results (typically 1)") enable_vector: bool = Field(default=True, description="Enable vector search") enable_graph: bool = Field(default=True, description="Enable graph search") enable_web: bool = Field(default=True, description="Enable web search") + enable_volatile: bool = Field(default=True, description="Enable volatile cache search") enable_reranking: bool = Field(default=True, description="Enable LLM re-ranking") enable_enrichment: bool = Field(default=True, description="Enable graph enrichment") final_result_count: int = Field(default=10, ge=1, le=50, description="Final results to return") rrf_k: int = Field(default=60, ge=1, le=100, description="RRF constant") + volatile_threshold: float = Field(default=0.8, ge=0.5, le=1.0, description="Volatile similarity threshold") class RelatedDossier(BaseModel): @@ -34,7 +37,7 @@ class RelatedDossier(BaseModel): class HybridRAGResult(BaseModel): """Single result from HybridRAG query.""" - source_type: str = Field(..., description="Source: 'vector', 'graph', 'web'") + source_type: str = Field(..., description="Source: 'wiki', 'web', 'volatile'") title: str content: str url: Optional[str] = Field(None, description="URL for web results") @@ -53,6 +56,7 @@ class TimingBreakdown(BaseModel): vector_ms: float = Field(..., description="Phase 1: Vector search") graph_ms: float = Field(..., description="Phase 1: Graph search") web_ms: float = Field(..., description="Phase 1: Web search") + volatile_ms: float = Field(default=0, description="Phase 1: Volatile cache search") fusion_ms: float = Field(..., description="Phase 2: RRF fusion") enrichment_ms: float = Field(..., description="Phase 3: Graph enrichment") reranking_ms: float = Field(..., description="Phase 4: LLM re-ranking") diff --git a/src/routers/hybrid_rag.py b/src/routers/hybrid_rag.py index 82f0fa3..89f44d5 100644 --- a/src/routers/hybrid_rag.py +++ b/src/routers/hybrid_rag.py @@ -1,7 +1,7 @@ """ HybridRAG router for multi-source search API. -Provides endpoint for combining vector, graph, and web search +Provides endpoint for combining vector, graph, volatile cache, and web search with RRF fusion and LLM re-ranking. """ @@ -34,10 +34,12 @@ def get_hybrid_rag_service( """Get HybridRAG service instance with all dependencies.""" from src.services.vector_service import VectorService from src.services.graph_service import GraphService + from src.services.volatile_service import VolatileCacheService # Create component services vector_service = VectorService(qdrant_client, wiki_client, ollama_client) graph_service = GraphService(neo4j_client, wiki_client) + volatile_service = VolatileCacheService(qdrant_client, ollama_client, settings) # Create HybridRAG service return HybridRAGService( @@ -46,7 +48,8 @@ def get_hybrid_rag_service( searxng_client=searxng_client, ollama_client=ollama_client, content_extractor=content_extractor, - settings=settings + settings=settings, + volatile_service=volatile_service ) @@ -58,26 +61,28 @@ async def hybrid_search( api_key: str = Depends(verify_api_key) ): """ - Execute HybridRAG query combining vector, graph, and web search. + Execute HybridRAG query combining vector, graph, volatile cache, and web search. **6-Phase Pipeline:** 1. **Query Enhancement**: Extract keywords/synonyms with LLM - 2. **Parallel Retrieval**: Search vector (Qdrant), graph (Neo4j), web (SearXNG) - 3. **RRF Fusion**: Merge results with Reciprocal Rank Fusion + 2. **Parallel Retrieval**: Search vector (Qdrant), graph (Neo4j), volatile cache, web (SearXNG) + 3. **RRF Fusion**: Merge results with Reciprocal Rank Fusion (volatile gets priority boost) 4. **Enrichment**: Add related documents via shared entities - 5. **LLM Re-ranking**: Re-rank with mistral-nemo for relevance + 5. **LLM Re-ranking**: Re-rank with configured model for relevance 6. **Context Formatting**: Format for LLM consumption 7. **Persistence**: Store for Librarian knowledge consolidation **Example Request:** ```json { - "query": "How does Docker orchestration work with Kubernetes?", + "query": "What's the weather in Rotterdam?", "user": "jpmschweitzer", "config": { "vector_limit": 10, "graph_limit": 10, "web_limit": 5, + "volatile_limit": 5, + "enable_volatile": true, "enable_reranking": true, "final_result_count": 10 } @@ -85,7 +90,7 @@ async def hybrid_search( ``` **Returns:** - - Ranked results from all sources + - Ranked results from all sources (wiki, volatile, web) - Extracted keywords/synonyms - Related dossiers (via graph) - Formatted context for LLM diff --git a/src/routers/maintenance.py b/src/routers/maintenance.py index 08f5d2d..38aa296 100644 --- a/src/routers/maintenance.py +++ b/src/routers/maintenance.py @@ -16,10 +16,12 @@ import time from src.services.vector_service import VectorService from src.services.graph_service import GraphService +from src.services.volatile_service import VolatileCacheService from src.core.dependencies import ( VectorServiceDep, GraphServiceDep, WikiJSDep, RedisDep, - verify_api_key + QdrantDep, OllamaDep, verify_api_key ) +from src.config import get_settings from datetime import datetime, timezone logger = logging.getLogger(__name__) @@ -209,6 +211,15 @@ class ReconcileIndexResponse(BaseModel): total_duration_ms: float +class VolatileCleanupResponse(BaseModel): + """Response from volatile cache cleanup operation.""" + success: bool + collections_processed: int + total_expired_purged: int + by_collection: Dict[str, int] = Field(default_factory=dict) + duration_ms: float + + # ========== Endpoints ========== @router.post("/cleanup/vectors", response_model=VectorCleanupResponse) @@ -490,6 +501,61 @@ async def cleanup_all( raise HTTPException(status_code=500, detail=str(e)) +@router.post("/cleanup/volatile", response_model=VolatileCleanupResponse) +async def cleanup_volatile( + qdrant: QdrantDep = None, + ollama: OllamaDep = None, + api_key: str = Depends(verify_api_key) +): + """ + Purge expired volatile cache records across all users. + + Loops through all volatile_* collections and removes records where + ttl_expiry < current_timestamp. + + **Scheduler Task** - Recommended to run every 10 minutes. + + **Scheduler Integration:** + ```json + { + "task_name": "volatile_cleanup", + "schedule": "*/10 * * * *", + "endpoint": "POST /maintenance/cleanup/volatile", + "description": "Purge expired volatile cache records" + } + ``` + """ + start_time = time.time() + + try: + settings = get_settings() + service = VolatileCacheService( + qdrant_client=qdrant, + ollama_client=ollama, + settings=settings + ) + + # Purge expired from all volatile collections + results = await service.purge_all_expired() + + total_purged = sum(results.values()) + duration_ms = (time.time() - start_time) * 1000 + + logger.info(f"Volatile cleanup complete: {total_purged} expired records purged from {len(results)} collections") + + return VolatileCleanupResponse( + success=True, + collections_processed=len(results), + total_expired_purged=total_purged, + by_collection=results, + duration_ms=duration_ms + ) + + except Exception as e: + logger.error(f"Volatile cleanup failed: {e}", exc_info=True) + raise HTTPException(status_code=500, detail=str(e)) + + @router.get("/health", response_model=HealthCheckResponse) async def maintenance_health( user: str = Query(..., description="User identifier"), diff --git a/src/routers/volatile.py b/src/routers/volatile.py index eb3845c..1adb018 100644 --- a/src/routers/volatile.py +++ b/src/routers/volatile.py @@ -2,6 +2,7 @@ Volatile cache router for Library Desk API. Endpoints for ephemeral cached data with TTL - weather, news, financial, etc. +Data is stored as vectors in Qdrant for semantic search retrieval. """ from fastapi import APIRouter, HTTPException, Depends, Query @@ -14,11 +15,11 @@ from src.models.volatile import ( VolatileScheduledResponse, VolatileStatsResponse, VolatileDeleteResponse, - VolatileBulkDeleteResponse, VolatileNamespace, + NAMESPACE_DEFAULT_TTL, ) from src.services.volatile_service import VolatileCacheService -from src.core.dependencies import verify_api_key, RedisDep +from src.core.dependencies import verify_api_key, QdrantDep, OllamaDep from src.core.multi_tenancy import DEFAULT_USER from src.config import get_settings @@ -27,16 +28,21 @@ logger = logging.getLogger(__name__) router = APIRouter(prefix="/volatile", tags=["Volatile Cache"]) -def get_volatile_service(redis: RedisDep) -> VolatileCacheService: +def get_volatile_service(qdrant: QdrantDep, ollama: OllamaDep) -> VolatileCacheService: """Get volatile cache service instance.""" settings = get_settings() - return VolatileCacheService(redis_client=redis, settings=settings) + return VolatileCacheService( + qdrant_client=qdrant, + ollama_client=ollama, + settings=settings + ) @router.get("/stats", response_model=VolatileStatsResponse) async def get_stats( user: str = Query(default=DEFAULT_USER, description="User identifier"), - redis: RedisDep = None, + qdrant: QdrantDep = None, + ollama: OllamaDep = None, api_key: str = Depends(verify_api_key) ): """ @@ -44,14 +50,14 @@ async def get_stats( Returns counts of records by namespace and scheduled refresh info. """ - service = get_volatile_service(redis) + service = get_volatile_service(qdrant, ollama) stats = await service.get_stats(user) return VolatileStatsResponse( total_records=stats["total_records"], by_namespace=stats["by_namespace"], scheduled_count=stats["scheduled_count"], - total_memory_bytes=stats.get("total_memory_bytes"), + total_memory_bytes=None, user=user, ) @@ -59,7 +65,8 @@ async def get_stats( @router.get("/scheduled", response_model=VolatileScheduledResponse) async def get_scheduled( user: str = Query(default=DEFAULT_USER, description="User identifier"), - redis: RedisDep = None, + qdrant: QdrantDep = None, + ollama: OllamaDep = None, api_key: str = Depends(verify_api_key) ): """ @@ -68,7 +75,7 @@ async def get_scheduled( Used by scheduler to determine what volatile data needs refreshing. Returns all records that have a refresh_schedule cron expression set. """ - service = get_volatile_service(redis) + service = get_volatile_service(qdrant, ollama) records = await service.get_scheduled(user) return VolatileScheduledResponse( @@ -86,10 +93,7 @@ async def list_namespaces( List available namespaces and their default TTLs. Returns predefined namespaces with their default TTL values. - Custom namespaces can also be used with the default TTL. """ - from src.models.volatile import NAMESPACE_DEFAULT_TTL - return { "namespaces": [ { @@ -120,103 +124,62 @@ def _get_namespace_description(ns: VolatileNamespace) -> str: return descriptions.get(ns, "Custom namespace") -@router.get("/{namespace}", response_model=VolatileListResponse) -async def list_keys( - namespace: str, +@router.get("/search") +async def search_volatile( + q: str = Query(..., min_length=1, description="Search query"), user: str = Query(default=DEFAULT_USER, description="User identifier"), - redis: RedisDep = None, + limit: int = Query(default=5, ge=1, le=20, description="Maximum results"), + threshold: float = Query(default=0.75, ge=0.5, le=1.0, description="Minimum similarity score"), + qdrant: QdrantDep = None, + ollama: OllamaDep = None, api_key: str = Depends(verify_api_key) ): """ - List all keys in a namespace. + Semantic search across volatile data. - Returns the list of keys stored in the specified namespace. - """ - service = get_volatile_service(redis) - keys = await service.list_namespace(user, namespace) - - return VolatileListResponse( - namespace=namespace, - keys=keys, - count=len(keys), - user=user, - ) - - -@router.delete("/{namespace}", response_model=VolatileBulkDeleteResponse) -async def delete_namespace( - namespace: str, - user: str = Query(default=DEFAULT_USER, description="User identifier"), - redis: RedisDep = None, - api_key: str = Depends(verify_api_key) -): - """ - Delete all records in a namespace. - - Removes all volatile data for the specified namespace. - """ - service = get_volatile_service(redis) - deleted = await service.delete_namespace(user, namespace) - - return VolatileBulkDeleteResponse( - namespace=namespace, - deleted_count=deleted, - user=user, - ) - - -@router.get("/{namespace}/{key}", response_model=VolatileRecordResponse) -async def get_record( - namespace: str, - key: str, - user: str = Query(default=DEFAULT_USER, description="User identifier"), - redis: RedisDep = None, - api_key: str = Depends(verify_api_key) -): - """ - Get a volatile record. - - Returns the record if it exists and has not expired. + Searches all volatile data for semantically similar content. + Higher threshold = stricter matching. **Example:** ``` - GET /volatile/weather/rotterdam?user=jpmschweitzer + GET /volatile/search?q=weather%20rotterdam&user=jpmschweitzer ``` """ - service = get_volatile_service(redis) - record = await service.get(user, namespace, key) + service = get_volatile_service(qdrant, ollama) + results = await service.search(user, q, limit=limit, score_threshold=threshold) - if not record: - raise HTTPException( - status_code=404, - detail=f"Record '{key}' not found in namespace '{namespace}'" - ) - - return record + return { + "query": q, + "results": results, + "count": len(results), + "user": user, + } -@router.post("/{namespace}/{key}", response_model=VolatileRecordResponse) -async def set_record( - namespace: str, - key: str, - request: VolatileRecordCreate, +@router.post("/store", response_model=VolatileRecordResponse) +async def store_volatile( + namespace: str = Query(..., description="Data namespace (weather, news, etc.)"), + key: str = Query(..., description="Record key (e.g., 'rotterdam', 'nos-headlines')"), + request: VolatileRecordCreate = None, user: str = Query(default=DEFAULT_USER, description="User identifier"), - redis: RedisDep = None, + qdrant: QdrantDep = None, + ollama: OllamaDep = None, api_key: str = Depends(verify_api_key) ): """ - Store or update a volatile record. + Store volatile data. - Creates or updates a record with the specified TTL. - If TTL is not provided, the namespace default is used. + Data is converted to natural language and embedded for semantic search. + If the same namespace+key already exists, it will be updated. **Example Request:** ```json + POST /volatile/store?namespace=weather&key=rotterdam { "data": { - "temperature": 18, - "conditions": "Partly cloudy", - "humidity": 65 + "temperature": 8, + "conditions": "Cloudy", + "humidity": 85 }, "source": "openweathermap", "ttl": 1800, @@ -228,10 +191,21 @@ async def set_record( Optional cron expression for automatic refresh. The scheduler will query `/volatile/scheduled` and trigger refreshes. """ - service = get_volatile_service(redis) + # Validate namespace if not custom + if namespace != VolatileNamespace.CUSTOM: + try: + VolatileNamespace(namespace) + except ValueError: + valid = [ns.value for ns in VolatileNamespace] + raise HTTPException( + status_code=400, + detail=f"Invalid namespace '{namespace}'. Valid: {valid}" + ) + + service = get_volatile_service(qdrant, ollama) try: - record = await service.set( + record = await service.store( user=user, namespace=namespace, key=key, @@ -244,7 +218,36 @@ async def set_record( except Exception as e: logger.error(f"Failed to store volatile record: {e}") - raise HTTPException(status_code=500, detail="Failed to store record") + raise HTTPException(status_code=500, detail=f"Failed to store record: {str(e)}") + + +@router.get("/{namespace}/{key}", response_model=VolatileRecordResponse) +async def get_record( + namespace: str, + key: str, + user: str = Query(default=DEFAULT_USER, description="User identifier"), + qdrant: QdrantDep = None, + ollama: OllamaDep = None, + api_key: str = Depends(verify_api_key) +): + """ + Get a specific volatile record by namespace and key. + + **Example:** + ``` + GET /volatile/weather/rotterdam?user=jpmschweitzer + ``` + """ + service = get_volatile_service(qdrant, ollama) + record = await service.get(user, namespace, key) + + if not record: + raise HTTPException( + status_code=404, + detail=f"Record '{key}' not found in namespace '{namespace}'" + ) + + return record @router.delete("/{namespace}/{key}", response_model=VolatileDeleteResponse) @@ -252,15 +255,14 @@ async def delete_record( namespace: str, key: str, user: str = Query(default=DEFAULT_USER, description="User identifier"), - redis: RedisDep = None, + qdrant: QdrantDep = None, + ollama: OllamaDep = None, api_key: str = Depends(verify_api_key) ): """ - Delete a volatile record. - - Removes the record from the cache. + Delete a specific volatile record. """ - service = get_volatile_service(redis) + service = get_volatile_service(qdrant, ollama) deleted = await service.delete(user, namespace, key) return VolatileDeleteResponse( diff --git a/src/services/hybrid_rag_service.py b/src/services/hybrid_rag_service.py index 3c3b9be..d8ee28e 100644 --- a/src/services/hybrid_rag_service.py +++ b/src/services/hybrid_rag_service.py @@ -20,6 +20,7 @@ import logging from src.services.vector_service import VectorService from src.services.graph_service import GraphService +from src.services.volatile_service import VolatileCacheService from src.clients.searxng_client import SearXNGClient from src.clients.ollama_client import OllamaClient from src.clients.content_extractor import ContentExtractor @@ -46,7 +47,8 @@ class HybridRAGService: searxng_client: SearXNGClient, ollama_client: OllamaClient, content_extractor: ContentExtractor, - settings: Settings + settings: Settings, + volatile_service: Optional[VolatileCacheService] = None ): """ Initialize HybridRAG service. @@ -58,6 +60,7 @@ class HybridRAGService: ollama_client: Client for LLM (keyword extraction, re-ranking) content_extractor: Client for extracting full content from URLs settings: Application settings + volatile_service: Service for volatile cache search (optional) """ self.vector = vector_service self.graph = graph_service @@ -65,6 +68,7 @@ class HybridRAGService: self.ollama = ollama_client self.content_extractor = content_extractor self.settings = settings + self.volatile = volatile_service self.reranker_model = settings.ollama_model async def search( @@ -104,8 +108,9 @@ class HybridRAGService: timing["vector_ms"] = raw_results.get("timing", {}).get("vector_ms", 0) timing["graph_ms"] = raw_results.get("timing", {}).get("graph_ms", 0) timing["web_ms"] = raw_results.get("timing", {}).get("web_ms", 0) + timing["volatile_ms"] = raw_results.get("timing", {}).get("volatile_ms", 0) - # Phase 2: Two-Stage RRF Fusion + # Phase 2: Three-Source RRF Fusion phase2_start = time.time() # Stage 1: Merge wiki sources (vector + graph) into single ranking @@ -115,10 +120,12 @@ class HybridRAGService: k=config.rrf_k ) - # Stage 2: Final RRF between wiki and web (equal footing) + # Stage 2: Final RRF between wiki, volatile, and web + # Volatile gets priority boost (smaller k = higher contribution per rank) fused_results = self._reciprocal_rank_fusion( wiki_results=wiki_merged, web_results=raw_results.get("web", []), + volatile_results=raw_results.get("volatile", []), k=config.rrf_k ) timing["fusion_ms"] = (time.time() - phase2_start) * 1000 @@ -389,6 +396,37 @@ JSON:""" tasks["web"] = web_search() + # Volatile cache search + if config.enable_volatile and self.volatile: + async def volatile_search(): + start = time.time() + try: + results = await self.volatile.search( + user=user, + query=query, + limit=config.volatile_limit, + score_threshold=config.volatile_threshold + ) + formatted = [ + { + "key": r.key, + "namespace": r.namespace, + "title": f"{r.namespace}: {r.key}", + "content": r.data.get("text", "") if isinstance(r.data, dict) else str(r.data), + "raw_data": r.data, + "source_api": r.source, + "ttl_remaining": r.ttl_remaining, + "source": "volatile" + } + for r in results + ] + return formatted, (time.time() - start) * 1000 + except Exception as e: + logger.error(f"Volatile search failed: {e}", exc_info=True) + return [], (time.time() - start) * 1000 + + tasks["volatile"] = volatile_search() + # Execute all searches in parallel results_dict = await asyncio.gather(*tasks.values()) @@ -401,7 +439,8 @@ JSON:""" logger.info( f"Parallel retrieval: vector={len(output.get('vector', []))}, " - f"graph={len(output.get('graph', []))}, web={len(output.get('web', []))}" + f"graph={len(output.get('graph', []))}, web={len(output.get('web', []))}, " + f"volatile={len(output.get('volatile', []))}" ) return output @@ -491,23 +530,42 @@ JSON:""" self, wiki_results: List[Dict], web_results: List[Dict], + volatile_results: Optional[List[Dict]] = None, k: int = 60 ) -> List[Dict[str, Any]]: """ - Stage 2: Final RRF between wiki (single source) and web. + Stage 2: Final RRF between wiki, volatile, and web. - Wiki results are pre-merged from vector+graph, so wiki and web - now compete on equal footing. + Wiki results are pre-merged from vector+graph. Volatile results + get a priority boost (smaller effective k) since they represent + current, time-sensitive information. Args: wiki_results: Pre-merged wiki results from _merge_wiki_sources() web_results: Results from web search + volatile_results: Results from volatile cache (fresh data) k: RRF constant (default 60) Returns: Final merged and sorted results """ rrf_scores = {} + volatile_results = volatile_results or [] + + # Volatile results get priority boost (k/2 = stronger score per rank) + volatile_k = k // 2 + for rank, result in enumerate(volatile_results, start=1): + key = result.get("key") + namespace = result.get("namespace", "unknown") + if not key: + continue + result_id = f"volatile_{namespace}_{key}" + rrf_scores[result_id] = { + "result": result, + "rrf_score": 1 / (volatile_k + rank), # Priority boost + "sources": ["volatile"], + "source_type": "volatile" + } # Wiki results (single source, already merged) for rank, result in enumerate(wiki_results, start=1): @@ -542,7 +600,8 @@ JSON:""" reverse=True ) - logger.info(f"Final RRF: {len(sorted_results)} results (wiki + web)") + volatile_count = len([r for r in sorted_results if r["source_type"] == "volatile"]) + logger.info(f"Final RRF: {len(sorted_results)} results (wiki + volatile[{volatile_count}] + web)") return sorted_results diff --git a/src/services/volatile_service.py b/src/services/volatile_service.py index 9391bd7..274a5df 100644 --- a/src/services/volatile_service.py +++ b/src/services/volatile_service.py @@ -1,23 +1,24 @@ """ Volatile Cache service for Library Desk. -Provides ephemeral data storage with TTL for time-sensitive information: +Provides ephemeral data storage with TTL using Qdrant vectors: - Weather, news, financial data - Transit schedules, traffic conditions - System status, social notifications + +Data is stored as embedded vectors for semantic search retrieval. """ -import json -import logging import hashlib +import logging +import time from datetime import datetime from typing import List, Optional, Dict, Any -import redis.asyncio as aioredis - +from src.clients.qdrant_client import QdrantClientWrapper +from src.clients.ollama_client import OllamaClient from src.config import Settings from src.models.volatile import ( - VolatileRecord, VolatileRecordResponse, VolatileNamespace, NAMESPACE_DEFAULT_TTL, @@ -28,47 +29,46 @@ logger = logging.getLogger(__name__) class VolatileCacheService: """ - Service for volatile data with TTL. + Service for volatile data with TTL stored in Qdrant. - Stores ephemeral data in Redis with automatic expiration. - Supports multiple namespaces with configurable TTLs. + Stores ephemeral data as vectors for semantic search retrieval. + Each user has an isolated volatile collection. """ - # Redis key prefix for volatile data - KEY_PREFIX = "volatile" + COLLECTION_PREFIX = "volatile_" def __init__( self, - redis_client: aioredis.Redis, + qdrant_client: QdrantClientWrapper, + ollama_client: OllamaClient, settings: Settings ): """ Initialize volatile cache service. Args: - redis_client: Async Redis client + qdrant_client: Qdrant client for vector storage + ollama_client: Ollama client for embeddings settings: Application settings """ - self.redis = redis_client + self.qdrant = qdrant_client + self.ollama = ollama_client self.settings = settings - logger.info("Initialized VolatileCacheService") + logger.info("Initialized VolatileCacheService (Qdrant backend)") - def _build_key(self, user: str, namespace: str, key: str) -> str: + def _collection_name(self, user: str) -> str: + """Get volatile collection name for user.""" + return f"{self.COLLECTION_PREFIX}{user}" + + def _make_vector_id(self, namespace: str, key: str) -> str: """ - Build Redis key for volatile record. + Generate deterministic vector ID for namespace/key. - Pattern: {user}:volatile:{namespace}:{key_hash} - Uses hash to ensure safe key characters and consistent length. + Same namespace+key always produces same ID for upsert behavior. """ - key_hash = hashlib.md5(key.encode()).hexdigest()[:12] - return f"{user}:{self.KEY_PREFIX}:{namespace}:{key_hash}" - - def _build_pattern(self, user: str, namespace: Optional[str] = None) -> str: - """Build pattern for key scanning.""" - if namespace: - return f"{user}:{self.KEY_PREFIX}:{namespace}:*" - return f"{user}:{self.KEY_PREFIX}:*" + combined = f"{namespace}:{key}" + return hashlib.md5(combined.encode()).hexdigest() def _get_default_ttl(self, namespace: str) -> int: """Get default TTL for a namespace.""" @@ -78,84 +78,116 @@ class VolatileCacheService: except ValueError: return self.settings.volatile_default_ttl - def _serialize_record(self, record: VolatileRecord) -> str: - """Serialize record to JSON for storage.""" - return json.dumps({ - "key": record.key, - "namespace": record.namespace, - "data": record.data, - "source": record.source, - "created_at": record.created_at.isoformat(), - "updated_at": record.updated_at.isoformat(), - "ttl": record.ttl, - "refresh_schedule": record.refresh_schedule, - "user": record.user, - }) + def _current_timestamp_ms(self) -> int: + """Get current timestamp in milliseconds.""" + return int(time.time() * 1000) - def _deserialize_record(self, data: str) -> VolatileRecord: - """Deserialize record from JSON.""" - obj = json.loads(data) - return VolatileRecord( - key=obj["key"], - namespace=obj["namespace"], - data=obj["data"], - source=obj.get("source"), - created_at=datetime.fromisoformat(obj["created_at"]), - updated_at=datetime.fromisoformat(obj["updated_at"]), - ttl=obj["ttl"], - refresh_schedule=obj.get("refresh_schedule"), - user=obj["user"], - ) - - async def get( + def _to_natural_language( self, - user: str, namespace: str, - key: str - ) -> Optional[VolatileRecordResponse]: + key: str, + data: Dict[str, Any] + ) -> str: """ - Get a volatile record. + Convert structured data to natural language for embedding. - Args: - user: User identifier - namespace: Data namespace - key: Record key - - Returns: - Record if found and not expired, None otherwise + This creates a text representation that embeds well semantically. """ - redis_key = self._build_key(user, namespace, key) + # Template-based conversion for known namespaces + if namespace == VolatileNamespace.WEATHER: + temp = data.get("temperature", data.get("temp", "unknown")) + conditions = data.get("conditions", data.get("weather", "")) + humidity = data.get("humidity", "") + text = f"Current weather in {key}: {temp}°C" + if conditions: + text += f", {conditions}" + if humidity: + text += f", humidity {humidity}%" + return text - try: - data = await self.redis.get(redis_key) - if not data: - return None + elif namespace == VolatileNamespace.NEWS: + title = data.get("title", data.get("headline", "")) + summary = data.get("summary", data.get("description", "")) + source = data.get("source", "") + text = f"News: {title}" + if summary: + text += f". {summary}" + if source: + text += f" (Source: {source})" + return text - record = self._deserialize_record(data) + elif namespace == VolatileNamespace.FINANCIAL: + symbol = data.get("symbol", key) + price = data.get("price", "") + change = data.get("change", data.get("change_percent", "")) + text = f"Financial data for {symbol}" + if price: + text += f": price {price}" + if change: + text += f", change {change}%" + return text - # Get TTL remaining - ttl_remaining = await self.redis.ttl(redis_key) - if ttl_remaining < 0: - return None + elif namespace == VolatileNamespace.TRANSIT: + route = data.get("route", data.get("line", key)) + status = data.get("status", "") + delay = data.get("delay", data.get("delay_minutes", "")) + text = f"Transit {route}" + if status: + text += f": {status}" + if delay: + text += f", delay {delay} minutes" + return text - return VolatileRecordResponse( - key=record.key, - namespace=record.namespace, - data=record.data, - source=record.source, - created_at=record.created_at, - updated_at=record.updated_at, - ttl=record.ttl, - ttl_remaining=max(0, ttl_remaining), - refresh_schedule=record.refresh_schedule, - user=record.user, - ) + elif namespace == VolatileNamespace.TRAFFIC: + location = data.get("location", key) + duration = data.get("duration", data.get("travel_time", "")) + congestion = data.get("congestion", "") + text = f"Traffic for {location}" + if duration: + text += f": {duration} minutes" + if congestion: + text += f", congestion level {congestion}" + return text - except Exception as e: - logger.error(f"Failed to get volatile record {redis_key}: {e}") - return None + elif namespace == VolatileNamespace.AIR_QUALITY: + location = data.get("location", key) + aqi = data.get("aqi", data.get("index", "")) + quality = data.get("quality", "") + text = f"Air quality in {location}" + if aqi: + text += f": AQI {aqi}" + if quality: + text += f" ({quality})" + return text - async def set( + elif namespace == VolatileNamespace.SPORTS: + event = data.get("event", data.get("match", key)) + score = data.get("score", "") + status = data.get("status", "") + text = f"Sports: {event}" + if score: + text += f" - Score: {score}" + if status: + text += f" ({status})" + return text + + elif namespace == VolatileNamespace.SYSTEM: + service = data.get("service", key) + status = data.get("status", "unknown") + message = data.get("message", "") + text = f"System status for {service}: {status}" + if message: + text += f". {message}" + return text + + # Fallback: serialize key fields + text_parts = [f"{namespace} data for {key}:"] + for k, v in data.items(): + if isinstance(v, (str, int, float, bool)): + text_parts.append(f"{k}: {v}") + return " ".join(text_parts) + + async def store( self, user: str, namespace: str, @@ -166,13 +198,13 @@ class VolatileCacheService: refresh_schedule: Optional[str] = None ) -> VolatileRecordResponse: """ - Store or update a volatile record. + Store volatile data as an embedded vector. Args: user: User identifier - namespace: Data namespace - key: Record key - data: Content to store + namespace: Data namespace (from controlled list) + key: Record key (normalized slug) + data: Structured data to store source: Origin API/service ttl: TTL in seconds (uses namespace default if not set) refresh_schedule: Optional cron expression for refresh @@ -180,49 +212,158 @@ class VolatileCacheService: Returns: The stored record """ - redis_key = self._build_key(user, namespace, key) + collection = self._collection_name(user) - # Use provided TTL or namespace default + # Ensure collection exists + await self.qdrant.ensure_collection(collection) + + # Calculate TTL and expiry effective_ttl = ttl if ttl is not None else self._get_default_ttl(namespace) + now_ms = self._current_timestamp_ms() + expiry_ms = now_ms + (effective_ttl * 1000) - # Check if record exists (for created_at) - existing = await self.get(user, namespace, key) + # Convert to natural language for embedding + text = self._to_natural_language(namespace, key, data) + + # Generate embedding + embedding = await self.ollama.embed(text) + if not embedding: + raise ValueError("Failed to generate embedding for volatile data") + + # Build payload now = datetime.utcnow() + payload = { + "doc_type": "volatile", + "namespace": namespace, + "key": key, + "text": text, + "raw_data": data, + "source": source, + "created_at": now.isoformat(), + "updated_at": now.isoformat(), + "ttl": effective_ttl, + "ttl_expiry": expiry_ms, + "refresh_schedule": refresh_schedule, + "user": user, + } - record = VolatileRecord( + # Upsert vector (same namespace+key = same ID = update) + vector_id = self._make_vector_id(namespace, key) + success = await self.qdrant.upsert_vector( + collection_name=collection, + vector_id=vector_id, + vector=embedding, + payload=payload + ) + + if not success: + raise ValueError("Failed to store volatile vector") + + logger.debug(f"Stored volatile {namespace}:{key} with TTL {effective_ttl}s") + + return VolatileRecordResponse( key=key, namespace=namespace, data=data, source=source, - created_at=existing.created_at if existing else now, + created_at=now, updated_at=now, ttl=effective_ttl, + ttl_remaining=effective_ttl, refresh_schedule=refresh_schedule, user=user, ) - try: - serialized = self._serialize_record(record) - await self.redis.setex(redis_key, effective_ttl, serialized) + async def search( + self, + user: str, + query: str, + limit: int = 5, + score_threshold: float = 0.75 + ) -> List[VolatileRecordResponse]: + """ + Semantic search across volatile data. - logger.debug(f"Stored volatile record {redis_key} with TTL {effective_ttl}s") + Args: + user: User identifier + query: Search query + limit: Maximum results + score_threshold: Minimum similarity score (higher = stricter) - return VolatileRecordResponse( - key=record.key, - namespace=record.namespace, - data=record.data, - source=record.source, - created_at=record.created_at, - updated_at=record.updated_at, - ttl=record.ttl, - ttl_remaining=effective_ttl, - refresh_schedule=record.refresh_schedule, - user=record.user, - ) + Returns: + List of matching volatile records + """ + collection = self._collection_name(user) - except Exception as e: - logger.error(f"Failed to store volatile record {redis_key}: {e}") - raise + # Check if collection exists + if not await self.qdrant.collection_exists(collection): + return [] + + # Generate query embedding + query_embedding = await self.ollama.embed(query) + if not query_embedding: + logger.error("Failed to embed query for volatile search") + return [] + + # Search with expiry filter + now_ms = self._current_timestamp_ms() + results = await self.qdrant.search_with_expiry_filter( + collection_name=collection, + query_vector=query_embedding, + current_timestamp=now_ms, + limit=limit, + score_threshold=score_threshold + ) + + # Convert to response models + responses = [] + for result in results: + payload = result["payload"] + ttl_expiry = payload.get("ttl_expiry", 0) + ttl_remaining = max(0, (ttl_expiry - now_ms) // 1000) + + responses.append(VolatileRecordResponse( + key=payload["key"], + namespace=payload["namespace"], + data=payload.get("raw_data", {}), + source=payload.get("source"), + created_at=datetime.fromisoformat(payload["created_at"]), + updated_at=datetime.fromisoformat(payload["updated_at"]), + ttl=payload.get("ttl", 0), + ttl_remaining=ttl_remaining, + refresh_schedule=payload.get("refresh_schedule"), + user=payload["user"], + )) + + return responses + + async def get( + self, + user: str, + namespace: str, + key: str + ) -> Optional[VolatileRecordResponse]: + """ + Get a specific volatile record by namespace and key. + + Args: + user: User identifier + namespace: Data namespace + key: Record key + + Returns: + Record if found and not expired, None otherwise + """ + # Use search with high threshold to find exact match + query = self._to_natural_language(namespace, key, {"key": key}) + results = await self.search(user, query, limit=10, score_threshold=0.5) + + # Find exact namespace+key match + for result in results: + if result.namespace == namespace and result.key == key: + return result + + return None async def delete( self, @@ -231,7 +372,7 @@ class VolatileCacheService: key: str ) -> bool: """ - Delete a volatile record. + Delete a specific volatile record. Args: user: User identifier @@ -239,51 +380,24 @@ class VolatileCacheService: key: Record key Returns: - True if record was deleted, False if not found + True if deleted, False if not found """ - redis_key = self._build_key(user, namespace, key) + collection = self._collection_name(user) - try: - deleted = await self.redis.delete(redis_key) - if deleted: - logger.debug(f"Deleted volatile record {redis_key}") - return deleted > 0 - - except Exception as e: - logger.error(f"Failed to delete volatile record {redis_key}: {e}") + if not await self.qdrant.collection_exists(collection): return False - async def list_namespace( - self, - user: str, - namespace: str - ) -> List[str]: - """ - List all keys in a namespace. - - Args: - user: User identifier - namespace: Data namespace - - Returns: - List of keys (original keys, not Redis keys) - """ - pattern = self._build_pattern(user, namespace) + vector_id = self._make_vector_id(namespace, key) try: - keys = [] - async for redis_key in self.redis.scan_iter(match=pattern): - # Get the record to retrieve original key - data = await self.redis.get(redis_key) - if data: - record = self._deserialize_record(data) - keys.append(record.key) - - return keys - + deleted = await self.qdrant.delete_by_ids( + collection_name=collection, + point_ids=[vector_id] + ) + return deleted > 0 except Exception as e: - logger.error(f"Failed to list namespace {namespace}: {e}") - return [] + logger.error(f"Failed to delete volatile {namespace}:{key}: {e}") + return False async def get_scheduled( self, @@ -300,33 +414,49 @@ class VolatileCacheService: Returns: List of records with refresh_schedule set """ - pattern = self._build_pattern(user) + collection = self._collection_name(user) + if not await self.qdrant.collection_exists(collection): + return [] + + now_ms = self._current_timestamp_ms() + scheduled = [] + + # Scroll through all non-expired records try: - scheduled = [] - async for redis_key in self.redis.scan_iter(match=pattern): - data = await self.redis.get(redis_key) - if data: - record = self._deserialize_record(data) - if record.refresh_schedule: - ttl_remaining = await self.redis.ttl(redis_key) - scheduled.append(VolatileRecordResponse( - key=record.key, - namespace=record.namespace, - data=record.data, - source=record.source, - created_at=record.created_at, - updated_at=record.updated_at, - ttl=record.ttl, - ttl_remaining=max(0, ttl_remaining), - refresh_schedule=record.refresh_schedule, - user=record.user, - )) + all_points = await self.qdrant.scroll_all_points( + collection_name=collection, + with_payload=True + ) + + for point in all_points: + payload = point.get("payload", {}) + ttl_expiry = payload.get("ttl_expiry", 0) + + # Skip expired + if ttl_expiry <= now_ms: + continue + + # Only include if has refresh schedule + if payload.get("refresh_schedule"): + ttl_remaining = max(0, (ttl_expiry - now_ms) // 1000) + scheduled.append(VolatileRecordResponse( + key=payload["key"], + namespace=payload["namespace"], + data=payload.get("raw_data", {}), + source=payload.get("source"), + created_at=datetime.fromisoformat(payload["created_at"]), + updated_at=datetime.fromisoformat(payload["updated_at"]), + ttl=payload.get("ttl", 0), + ttl_remaining=ttl_remaining, + refresh_schedule=payload["refresh_schedule"], + user=payload["user"], + )) return scheduled except Exception as e: - logger.error(f"Failed to get scheduled records: {e}") + logger.error(f"Failed to get scheduled volatile records: {e}") return [] async def get_stats( @@ -342,92 +472,93 @@ class VolatileCacheService: Returns: Statistics dict """ - pattern = self._build_pattern(user) + collection = self._collection_name(user) + + if not await self.qdrant.collection_exists(collection): + return { + "total_records": 0, + "by_namespace": {}, + "scheduled_count": 0, + "expired_count": 0, + } + + now_ms = self._current_timestamp_ms() + by_namespace: Dict[str, int] = {} + total = 0 + scheduled = 0 + expired = 0 try: - by_namespace: Dict[str, int] = {} - total = 0 - scheduled = 0 + all_points = await self.qdrant.scroll_all_points( + collection_name=collection, + with_payload=True + ) - async for redis_key in self.redis.scan_iter(match=pattern): - data = await self.redis.get(redis_key) - if data: - record = self._deserialize_record(data) + for point in all_points: + payload = point.get("payload", {}) + namespace = payload.get("namespace", "unknown") + ttl_expiry = payload.get("ttl_expiry", 0) + + if ttl_expiry <= now_ms: + expired += 1 + else: total += 1 - by_namespace[record.namespace] = by_namespace.get(record.namespace, 0) + 1 - if record.refresh_schedule: + by_namespace[namespace] = by_namespace.get(namespace, 0) + 1 + if payload.get("refresh_schedule"): scheduled += 1 return { "total_records": total, "by_namespace": by_namespace, "scheduled_count": scheduled, - "total_memory_bytes": None, # Could implement with DEBUG MEMORY + "expired_count": expired, } except Exception as e: - logger.error(f"Failed to get stats: {e}") + logger.error(f"Failed to get volatile stats: {e}") return { "total_records": 0, "by_namespace": {}, "scheduled_count": 0, - "total_memory_bytes": None, + "expired_count": 0, } - async def delete_namespace( - self, - user: str, - namespace: str - ) -> int: - """ - Delete all records in a namespace. - - Args: - user: User identifier - namespace: Data namespace - - Returns: - Number of records deleted - """ - pattern = self._build_pattern(user, namespace) - - try: - deleted = 0 - async for redis_key in self.redis.scan_iter(match=pattern): - await self.redis.delete(redis_key) - deleted += 1 - - logger.info(f"Deleted {deleted} records from namespace {namespace}") - return deleted - - except Exception as e: - logger.error(f"Failed to delete namespace {namespace}: {e}") - return 0 - - async def delete_all( + async def purge_expired( self, user: str ) -> int: """ - Delete all volatile records for user. + Purge all expired volatile records for user. Args: user: User identifier Returns: - Number of records deleted + Number of records purged """ - pattern = self._build_pattern(user) + collection = self._collection_name(user) - try: - deleted = 0 - async for redis_key in self.redis.scan_iter(match=pattern): - await self.redis.delete(redis_key) - deleted += 1 - - logger.info(f"Deleted all {deleted} volatile records for user {user}") - return deleted - - except Exception as e: - logger.error(f"Failed to delete all records: {e}") + if not await self.qdrant.collection_exists(collection): return 0 + + now_ms = self._current_timestamp_ms() + return await self.qdrant.delete_expired_vectors(collection, now_ms) + + async def purge_all_expired(self) -> Dict[str, int]: + """ + Purge expired records from all volatile collections. + + Returns: + Dict of collection -> purged count + """ + collections = await self.qdrant.get_volatile_collections() + results = {} + now_ms = self._current_timestamp_ms() + + for collection in collections: + purged = await self.qdrant.delete_expired_vectors(collection, now_ms) + if purged > 0: + results[collection] = purged + logger.info(f"Purged {purged} expired from {collection}") + + return results diff --git a/tests/test_volatile.py b/tests/test_volatile.py index 2e64562..cf71fbb 100644 --- a/tests/test_volatile.py +++ b/tests/test_volatile.py @@ -1,11 +1,13 @@ """ -Tests for volatile cache router and service. +Tests for volatile cache router and service (Qdrant backend). Tests: - Volatile record CRUD operations - Namespace listing and management - Scheduled record retrieval -- TTL behavior +- TTL behavior and expiry filtering +- Semantic search +- Natural language conversion """ import pytest @@ -106,6 +108,10 @@ class TestVolatileNamespaces: """Test sports namespace default TTL (fast updates).""" assert NAMESPACE_DEFAULT_TTL[VolatileNamespace.SPORTS] == 60 # 1 min + def test_namespace_count(self): + """Test we have the expected number of namespaces.""" + assert len(VolatileNamespace) == 11 + class TestVolatileListResponse: """Test list response models.""" @@ -210,18 +216,28 @@ class TestVolatileDeleteResponses: class TestVolatileService: - """Test VolatileCacheService functionality.""" + """Test VolatileCacheService functionality (Qdrant backend).""" @pytest.fixture - def mock_redis(self): - """Create mock Redis client.""" - redis = AsyncMock() - redis.get = AsyncMock(return_value=None) - redis.setex = AsyncMock() - redis.delete = AsyncMock(return_value=1) - redis.ttl = AsyncMock(return_value=1500) - redis.scan_iter = MagicMock(return_value=iter([])) - return redis + def mock_qdrant(self): + """Create mock Qdrant client.""" + qdrant = AsyncMock() + qdrant.ensure_collection = AsyncMock() + qdrant.collection_exists = AsyncMock(return_value=True) + qdrant.upsert_vector = AsyncMock(return_value=True) + qdrant.delete_by_ids = AsyncMock(return_value=1) + qdrant.search_with_expiry_filter = AsyncMock(return_value=[]) + qdrant.scroll_all_points = AsyncMock(return_value=[]) + qdrant.delete_expired_vectors = AsyncMock(return_value=0) + qdrant.get_volatile_collections = AsyncMock(return_value=[]) + return qdrant + + @pytest.fixture + def mock_ollama(self): + """Create mock Ollama client.""" + ollama = AsyncMock() + ollama.embed = AsyncMock(return_value=[0.1] * 768) # Return 768-dim embedding + return ollama @pytest.fixture def mock_settings(self): @@ -231,29 +247,29 @@ class TestVolatileService: return settings @pytest.fixture - def volatile_service(self, mock_redis, mock_settings): + def volatile_service(self, mock_qdrant, mock_ollama, mock_settings): """Create VolatileCacheService with mocks.""" from src.services.volatile_service import VolatileCacheService return VolatileCacheService( - redis_client=mock_redis, + qdrant_client=mock_qdrant, + ollama_client=mock_ollama, settings=mock_settings ) - def test_build_key(self, volatile_service): - """Test Redis key building.""" - key = volatile_service._build_key("jpmschweitzer", "weather", "rotterdam") - assert key.startswith("jpmschweitzer:volatile:weather:") - assert len(key) > 30 # Has hash suffix + def test_collection_name(self, volatile_service): + """Test collection naming pattern.""" + name = volatile_service._collection_name("jpmschweitzer") + assert name == "volatile_jpmschweitzer" - def test_build_pattern(self, volatile_service): - """Test pattern building.""" - pattern = volatile_service._build_pattern("jpmschweitzer", "weather") - assert pattern == "jpmschweitzer:volatile:weather:*" + def test_make_vector_id(self, volatile_service): + """Test deterministic vector ID generation.""" + id1 = volatile_service._make_vector_id("weather", "rotterdam") + id2 = volatile_service._make_vector_id("weather", "rotterdam") + id3 = volatile_service._make_vector_id("weather", "amsterdam") - def test_build_pattern_all(self, volatile_service): - """Test pattern building for all namespaces.""" - pattern = volatile_service._build_pattern("jpmschweitzer") - assert pattern == "jpmschweitzer:volatile:*" + assert id1 == id2 # Same namespace+key = same ID + assert id1 != id3 # Different key = different ID + assert len(id1) == 32 # MD5 hex length def test_get_default_ttl_known_namespace(self, volatile_service): """Test default TTL for known namespace.""" @@ -265,23 +281,287 @@ class TestVolatileService: ttl = volatile_service._get_default_ttl("unknown_namespace") assert ttl == 3600 # Falls back to settings default - @pytest.mark.asyncio - async def test_get_not_found(self, volatile_service, mock_redis): - """Test get when record not found.""" - mock_redis.get.return_value = None - result = await volatile_service.get("jpmschweitzer", "weather", "rotterdam") - assert result is None + def test_to_natural_language_weather(self, volatile_service): + """Test natural language conversion for weather data.""" + text = volatile_service._to_natural_language( + namespace="weather", + key="rotterdam", + data={"temperature": 18, "conditions": "Cloudy", "humidity": 75} + ) + assert "rotterdam" in text.lower() + assert "18" in text + assert "Cloudy" in text + assert "75" in text + + def test_to_natural_language_news(self, volatile_service): + """Test natural language conversion for news data.""" + text = volatile_service._to_natural_language( + namespace="news", + key="nos-headlines", + data={"title": "Breaking News", "summary": "Something happened", "source": "NOS"} + ) + assert "Breaking News" in text + assert "Something happened" in text + assert "NOS" in text + + def test_to_natural_language_financial(self, volatile_service): + """Test natural language conversion for financial data.""" + text = volatile_service._to_natural_language( + namespace="financial", + key="AAPL", + data={"symbol": "AAPL", "price": 150.50, "change": 2.3} + ) + assert "AAPL" in text + assert "price" in text.lower() + assert "change" in text.lower() + + def test_to_natural_language_transit(self, volatile_service): + """Test natural language conversion for transit data.""" + text = volatile_service._to_natural_language( + namespace="transit", + key="ns-intercity", + data={"route": "Amsterdam-Rotterdam", "status": "On time", "delay": 0} + ) + assert "Amsterdam-Rotterdam" in text or "ns-intercity" in text.lower() + assert "On time" in text + + def test_to_natural_language_fallback(self, volatile_service): + """Test natural language fallback for unknown namespace.""" + text = volatile_service._to_natural_language( + namespace="custom", + key="test-key", + data={"foo": "bar", "count": 42} + ) + assert "custom" in text.lower() + assert "foo" in text or "bar" in text @pytest.mark.asyncio - async def test_delete_success(self, volatile_service, mock_redis): + async def test_store_success(self, volatile_service, mock_qdrant, mock_ollama): + """Test successful store operation.""" + result = await volatile_service.store( + user="jpmschweitzer", + namespace="weather", + key="rotterdam", + data={"temperature": 18, "conditions": "Sunny"}, + source="openweathermap", + ttl=1800 + ) + + assert result.key == "rotterdam" + assert result.namespace == "weather" + assert result.ttl == 1800 + mock_qdrant.ensure_collection.assert_called_once() + mock_ollama.embed.assert_called_once() + mock_qdrant.upsert_vector.assert_called_once() + + @pytest.mark.asyncio + async def test_store_uses_namespace_default_ttl(self, volatile_service, mock_qdrant, mock_ollama): + """Test store uses namespace default TTL when not specified.""" + result = await volatile_service.store( + user="jpmschweitzer", + namespace="weather", + key="amsterdam", + data={"temperature": 16}, + source="openweathermap", + ttl=None # Not specified + ) + + assert result.ttl == 1800 # Weather default + + @pytest.mark.asyncio + async def test_search_empty_collection(self, volatile_service, mock_qdrant, mock_ollama): + """Test search when collection doesn't exist.""" + mock_qdrant.collection_exists.return_value = False + + results = await volatile_service.search( + user="jpmschweitzer", + query="weather rotterdam" + ) + + assert results == [] + mock_ollama.embed.assert_not_called() + + @pytest.mark.asyncio + async def test_search_with_results(self, volatile_service, mock_qdrant, mock_ollama): + """Test search returns results.""" + import time + now_ms = int(time.time() * 1000) + + mock_qdrant.search_with_expiry_filter.return_value = [ + { + "score": 0.95, + "payload": { + "key": "rotterdam", + "namespace": "weather", + "raw_data": {"temperature": 18}, + "source": "openweathermap", + "created_at": datetime.utcnow().isoformat(), + "updated_at": datetime.utcnow().isoformat(), + "ttl": 1800, + "ttl_expiry": now_ms + 900000, # 15 min remaining + "refresh_schedule": None, + "user": "jpmschweitzer" + } + } + ] + + results = await volatile_service.search( + user="jpmschweitzer", + query="weather rotterdam" + ) + + assert len(results) == 1 + assert results[0].key == "rotterdam" + assert results[0].namespace == "weather" + + @pytest.mark.asyncio + async def test_delete_success(self, volatile_service, mock_qdrant): """Test successful delete.""" - mock_redis.delete.return_value = 1 + mock_qdrant.delete_by_ids.return_value = 1 + result = await volatile_service.delete("jpmschweitzer", "weather", "rotterdam") + assert result is True + mock_qdrant.delete_by_ids.assert_called_once() @pytest.mark.asyncio - async def test_delete_not_found(self, volatile_service, mock_redis): + async def test_delete_not_found(self, volatile_service, mock_qdrant): """Test delete when record not found.""" - mock_redis.delete.return_value = 0 + mock_qdrant.delete_by_ids.return_value = 0 + result = await volatile_service.delete("jpmschweitzer", "weather", "nonexistent") + assert result is False + + @pytest.mark.asyncio + async def test_get_stats_empty(self, volatile_service, mock_qdrant): + """Test stats with no records.""" + mock_qdrant.collection_exists.return_value = False + + stats = await volatile_service.get_stats("jpmschweitzer") + + assert stats["total_records"] == 0 + assert stats["by_namespace"] == {} + assert stats["scheduled_count"] == 0 + + @pytest.mark.asyncio + async def test_get_stats_with_records(self, volatile_service, mock_qdrant): + """Test stats with records.""" + import time + now_ms = int(time.time() * 1000) + + mock_qdrant.scroll_all_points.return_value = [ + {"payload": {"namespace": "weather", "ttl_expiry": now_ms + 100000}}, + {"payload": {"namespace": "weather", "ttl_expiry": now_ms + 100000, "refresh_schedule": "0 * * * *"}}, + {"payload": {"namespace": "news", "ttl_expiry": now_ms + 100000}}, + {"payload": {"namespace": "weather", "ttl_expiry": now_ms - 100000}}, # Expired + ] + + stats = await volatile_service.get_stats("jpmschweitzer") + + assert stats["total_records"] == 3 # Excludes expired + assert stats["by_namespace"]["weather"] == 2 + assert stats["by_namespace"]["news"] == 1 + assert stats["scheduled_count"] == 1 + assert stats["expired_count"] == 1 + + @pytest.mark.asyncio + async def test_purge_expired(self, volatile_service, mock_qdrant): + """Test purging expired records.""" + mock_qdrant.delete_expired_vectors.return_value = 5 + + result = await volatile_service.purge_expired("jpmschweitzer") + + assert result == 5 + mock_qdrant.delete_expired_vectors.assert_called_once() + + @pytest.mark.asyncio + async def test_purge_all_expired(self, volatile_service, mock_qdrant): + """Test purging expired from all collections.""" + mock_qdrant.get_volatile_collections.return_value = [ + "volatile_user1", + "volatile_user2" + ] + mock_qdrant.delete_expired_vectors.side_effect = [3, 2] + + results = await volatile_service.purge_all_expired() + + assert results["volatile_user1"] == 3 + assert results["volatile_user2"] == 2 + + @pytest.mark.asyncio + async def test_get_scheduled(self, volatile_service, mock_qdrant): + """Test getting scheduled records.""" + import time + now_ms = int(time.time() * 1000) + + mock_qdrant.scroll_all_points.return_value = [ + { + "payload": { + "key": "nos-headlines", + "namespace": "news", + "raw_data": {"headlines": []}, + "source": "nos.nl", + "created_at": datetime.utcnow().isoformat(), + "updated_at": datetime.utcnow().isoformat(), + "ttl": 3600, + "ttl_expiry": now_ms + 1800000, + "refresh_schedule": "0 */6 * * *", + "user": "jpmschweitzer" + } + }, + { + "payload": { + "key": "rotterdam", + "namespace": "weather", + "raw_data": {"temperature": 18}, + "source": "openweathermap", + "created_at": datetime.utcnow().isoformat(), + "updated_at": datetime.utcnow().isoformat(), + "ttl": 1800, + "ttl_expiry": now_ms + 900000, + "refresh_schedule": None, # Not scheduled + "user": "jpmschweitzer" + } + } + ] + + scheduled = await volatile_service.get_scheduled("jpmschweitzer") + + assert len(scheduled) == 1 + assert scheduled[0].key == "nos-headlines" + assert scheduled[0].refresh_schedule == "0 */6 * * *" + + +class TestVolatileCleanupEndpoint: + """Test volatile cleanup in maintenance router.""" + + @pytest.mark.asyncio + async def test_cleanup_volatile(self): + """Test volatile cleanup endpoint.""" + from src.routers.maintenance import cleanup_volatile, VolatileCleanupResponse + + mock_qdrant = AsyncMock() + mock_qdrant.get_volatile_collections = AsyncMock(return_value=[ + "volatile_user1", + "volatile_user2" + ]) + mock_qdrant.delete_expired_vectors = AsyncMock(side_effect=[3, 2]) + + mock_ollama = AsyncMock() + + mock_settings = MagicMock() + mock_settings.volatile_default_ttl = 3600 + + with patch('src.routers.maintenance.get_settings', return_value=mock_settings): + result = await cleanup_volatile( + qdrant=mock_qdrant, + ollama=mock_ollama, + api_key="test" + ) + + assert result.success is True + assert result.collections_processed == 2 + assert result.total_expired_purged == 5 + assert result.by_collection["volatile_user1"] == 3 + assert result.by_collection["volatile_user2"] == 2