cache.py

"""
Caching module for Silver Table Assistant.
Provides in-memory caching for performance optimization.
"""

import time
import hashlib
import json
from typing import Any, Optional, Dict, Callable
from functools import wraps
import asyncio


class Cache:
    """Simple in-memory cache with TTL support."""
    
    def __init__(self, default_ttl: int = 3600):  # 1 hour default TTL
        self._cache: Dict[str, Dict[str, Any]] = {}
        self.default_ttl = default_ttl
    
    def _generate_key(self, prefix: str, *args, **kwargs) -> str:
        """Generate a cache key from function arguments."""
        # Convert args and kwargs to a sorted string representation
        key_data = {
            "args": args,
            "kwargs": kwargs
        }
        key_string = json.dumps(key_data, sort_keys=True, default=str)
        key_hash = hashlib.md5(key_string.encode()).hexdigest()
        return f"{prefix}:{key_hash}"
    
    def get(self, key: str) -> Optional[Any]:
        """Get value from cache if not expired."""
        if key in self._cache:
            cache_entry = self._cache[key]
            if time.time() < cache_entry["expires_at"]:
                return cache_entry["value"]
            else:
                # Remove expired entry
                del self._cache[key]
        return None
    
    def set(self, key: str, value: Any, ttl: Optional[int] = None) -> None:
        """Set value in cache with TTL."""
        ttl = ttl or self.default_ttl
        self._cache[key] = {
            "value": value,
            "expires_at": time.time() + ttl
        }
    
    def delete(self, key: str) -> None:
        """Delete key from cache."""
        if key in self._cache:
            del self._cache[key]
    
    def clear(self) -> None:
        """Clear all cache entries."""
        self._cache.clear()
    
    def cleanup_expired(self) -> int:
        """Remove all expired entries and return count removed."""
        current_time = time.time()
        expired_keys = [
            key for key, entry in self._cache.items()
            if current_time >= entry["expires_at"]
        ]
        
        for key in expired_keys:
            del self._cache[key]
        
        return len(expired_keys)
    
    def get_stats(self) -> Dict[str, Any]:
        """Get cache statistics."""
        current_time = time.time()
        total_entries = len(self._cache)
        expired_entries = sum(
            1 for entry in self._cache.values()
            if current_time >= entry["expires_at"]
        )
        
        return {
            "total_entries": total_entries,
            "active_entries": total_entries - expired_entries,
            "expired_entries": expired_entries,
            "cache_hit_potential": "high" if total_entries > 100 else "medium" if total_entries > 10 else "low"
        }


# Global cache instances
document_cache = Cache(default_ttl=1800)  # 30 minutes for document queries
nutrition_cache = Cache(default_ttl=3600)  # 1 hour for nutrition calculations
user_context_cache = Cache(default_ttl=900)  # 15 minutes for user context


def cache_result(cache_instance: Cache, prefix: str, ttl: Optional[int] = None):
    """Decorator to cache function results."""
    
    def decorator(func: Callable):
        @wraps(func)
        async def async_wrapper(*args, **kwargs):
            # Generate cache key
            cache_key = cache_instance._generate_key(prefix, *args, **kwargs)
            
            # Try to get from cache
            cached_result = cache_instance.get(cache_key)
            if cached_result is not None:
                return cached_result
            
            # Execute function and cache result
            if asyncio.iscoroutinefunction(func):
                result = await func(*args, **kwargs)
            else:
                result = func(*args, **kwargs)
            
            cache_instance.set(cache_key, result, ttl)
            return result
        
        @wraps(func)
        def sync_wrapper(*args, **kwargs):
            # Generate cache key
            cache_key = cache_instance._generate_key(prefix, *args, **kwargs)
            
            # Try to get from cache
            cached_result = cache_instance.get(cache_key)
            if cached_result is not None:
                return cached_result
            
            # Execute function and cache result
            result = func(*args, **kwargs)
            cache_instance.set(cache_key, result, ttl)
            return result
        
        # Choose wrapper based on function type
        if asyncio.iscoroutinefunction(func):
            return async_wrapper
        else:
            return sync_wrapper
    
    return decorator


class NutritionCache:
    """Specialized cache for nutrition calculations."""
    
    @staticmethod
    def get_menu_item_nutrition(menu_item_id: int) -> Optional[Dict[str, Any]]:
        """Get cached nutrition data for a menu item."""
        key = f"nutrition:menu_item:{menu_item_id}"
        return nutrition_cache.get(key)
    
    @staticmethod
    def set_menu_item_nutrition(menu_item_id: int, nutrition_data: Dict[str, Any], ttl: Optional[int] = None) -> None:
        """Cache nutrition data for a menu item."""
        key = f"nutrition:menu_item:{menu_item_id}"
        nutrition_cache.set(key, nutrition_data, ttl)
    
    @staticmethod
    def get_user_nutrition_summary(user_id: str, days: int = 7) -> Optional[Dict[str, Any]]:
        """Get cached nutrition summary for a user."""
        key = f"nutrition:summary:{user_id}:{days}"
        return nutrition_cache.get(key)
    
    @staticmethod
    def set_user_nutrition_summary(user_id: str, days: int, summary_data: Dict[str, Any], ttl: Optional[int] = None) -> None:
        """Cache nutrition summary for a user."""
        key = f"nutrition:summary:{user_id}:{days}"
        nutrition_cache.set(key, summary_data, ttl)
    
    @staticmethod
    def invalidate_user_nutrition(user_id: str) -> None:
        """Invalidate all nutrition cache for a user."""
        # Remove all nutrition-related entries for this user
        keys_to_remove = [
            key for key in nutrition_cache._cache.keys()
            if key.startswith(f"nutrition:summary:{user_id}")
        ]
        for key in keys_to_remove:
            nutrition_cache.delete(key)


class DocumentCache:
    """Specialized cache for document queries."""
    
    @staticmethod
    def get_relevant_documents(query: str, k: int, score_threshold: Optional[float] = None) -> Optional[list]:
        """Get cached document search results."""
        threshold_key = f":{score_threshold}" if score_threshold else ""
        key = f"documents:query:{hashlib.md5(query.encode()).hexdigest()}:k{k}{threshold_key}"
        return document_cache.get(key)
    
    @staticmethod
    def set_relevant_documents(query: str, k: int, documents: list, score_threshold: Optional[float] = None, ttl: Optional[int] = None) -> None:
        """Cache document search results."""
        threshold_key = f":{score_threshold}" if score_threshold else ""
        key = f"documents:query:{hashlib.md5(query.encode()).hexdigest()}:k{k}{threshold_key}"
        document_cache.set(key, documents, ttl)
    
    @staticmethod
    def invalidate_document_cache() -> None:
        """Invalidate all document cache entries."""
        document_cache.clear()


class UserContextCache:
    """Specialized cache for user context data."""
    
    @staticmethod
    def get_user_context(user_id: str) -> Optional[Dict[str, Any]]:
        """Get cached user context."""
        key = f"context:user:{user_id}"
        return user_context_cache.get(key)
    
    @staticmethod
    def set_user_context(user_id: str, context_data: Dict[str, Any], ttl: Optional[int] = None) -> None:
        """Cache user context data."""
        key = f"context:user:{user_id}"
        user_context_cache.set(key, context_data, ttl)
    
    @staticmethod
    def invalidate_user_context(user_id: str) -> None:
        """Invalidate user context cache."""
        key = f"context:user:{user_id}"
        user_context_cache.delete(key)


# Cache management utilities
def get_cache_stats() -> Dict[str, Any]:
    """Get statistics for all caches."""
    return {
        "document_cache": document_cache.get_stats(),
        "nutrition_cache": nutrition_cache.get_stats(),
        "user_context_cache": user_context_cache.get_stats()
    }


def cleanup_all_caches() -> Dict[str, int]:
    """Clean up expired entries from all caches."""
    return {
        "document_cache": document_cache.cleanup_expired(),
        "nutrition_cache": nutrition_cache.cleanup_expired(),
        "user_context_cache": user_context_cache.cleanup_expired()
    }


def invalidate_user_cache(user_id: str) -> None:
    """Invalidate all cache entries for a specific user."""
    NutritionCache.invalidate_user_nutrition(user_id)
    UserContextCache.invalidate_user_context(user_id)