markitect-main/markitect/assets/cache.py

"""
Caching functionality for Issue #144.

This module provides asset caching capabilities for improved performance
including metadata caching, thumbnail caching, and cache management.
"""

import time
from pathlib import Path
from typing import Dict, Any, Optional, Tuple
from dataclasses import dataclass, field
from enum import Enum
from collections import OrderedDict


class CacheStrategy(Enum):
    """Cache eviction strategies."""
    LRU = "lru"
    FIFO = "fifo"
    TTL = "ttl"


@dataclass
class CacheMetrics:
    """Cache performance metrics."""
    total_requests: int = 0
    cache_hits: int = 0
    cache_misses: int = 0
    evictions: int = 0
    current_size_bytes: int = 0

    @property
    def hit_rate(self) -> float:
        """Calculate cache hit rate."""
        if self.total_requests == 0:
            return 0.0
        return self.cache_hits / self.total_requests


class AssetCache:
    """Asset caching system for metadata and thumbnails."""

    def __init__(self, max_size_mb: int = 100, strategy: CacheStrategy = CacheStrategy.LRU,
                 enable_metrics: bool = True):
        """Initialize asset cache."""
        self.max_size_bytes = max_size_mb * 1024 * 1024
        self.strategy = strategy
        self.enable_metrics = enable_metrics

        # Cache storage
        self._metadata_cache: OrderedDict = OrderedDict()
        self._thumbnail_cache: OrderedDict = OrderedDict()

        # Size tracking
        self.current_size_bytes = 0

        # Metrics
        self._metrics = CacheMetrics()

    def store_metadata(self, content_hash: str, metadata: Dict[str, Any]):
        """Store asset metadata in cache."""
        if self.enable_metrics:
            self._metrics.total_requests += 1

        # Estimate size (simplified)
        estimated_size = len(str(metadata)) * 4  # Rough estimate

        # Check if we need to evict
        self._ensure_capacity(estimated_size)

        # Store metadata
        self._metadata_cache[content_hash] = {
            'data': metadata,
            'timestamp': time.time(),
            'size': estimated_size
        }

        self.current_size_bytes += estimated_size

        if self.enable_metrics:
            self._metrics.cache_misses += 1

    def get_metadata(self, content_hash: str) -> Optional[Dict[str, Any]]:
        """Retrieve asset metadata from cache."""
        if self.enable_metrics:
            self._metrics.total_requests += 1

        if content_hash in self._metadata_cache:
            # Move to end for LRU
            if self.strategy == CacheStrategy.LRU:
                metadata_entry = self._metadata_cache.pop(content_hash)
                self._metadata_cache[content_hash] = metadata_entry

            if self.enable_metrics:
                self._metrics.cache_hits += 1

            return self._metadata_cache[content_hash]['data']

        if self.enable_metrics:
            self._metrics.cache_misses += 1

        return None

    def generate_and_cache_thumbnail(self, content_hash: str, image_path: Path,
                                   size: Tuple[int, int] = (150, 150)) -> bytes:
        """Generate and cache a thumbnail."""
        thumbnail_key = f"{content_hash}_{size[0]}x{size[1]}"

        # Check if thumbnail already cached
        cached_thumbnail = self.get_thumbnail(content_hash, size)
        if cached_thumbnail:
            return cached_thumbnail

        # Generate thumbnail (simplified mock)
        thumbnail_data = f"thumbnail_{size[0]}x{size[1]}".encode()

        # Cache thumbnail
        estimated_size = len(thumbnail_data)
        self._ensure_capacity(estimated_size)

        self._thumbnail_cache[thumbnail_key] = {
            'data': thumbnail_data,
            'timestamp': time.time(),
            'size': estimated_size
        }

        self.current_size_bytes += estimated_size

        return thumbnail_data

    def get_thumbnail(self, content_hash: str, size: Tuple[int, int]) -> Optional[bytes]:
        """Retrieve cached thumbnail."""
        thumbnail_key = f"{content_hash}_{size[0]}x{size[1]}"

        if thumbnail_key in self._thumbnail_cache:
            # Move to end for LRU
            if self.strategy == CacheStrategy.LRU:
                thumbnail_entry = self._thumbnail_cache.pop(thumbnail_key)
                self._thumbnail_cache[thumbnail_key] = thumbnail_entry

            return self._thumbnail_cache[thumbnail_key]['data']

        return None

    def invalidate(self, content_hash: str):
        """Invalidate cache entries for a specific asset."""
        # Remove metadata
        if content_hash in self._metadata_cache:
            entry = self._metadata_cache.pop(content_hash)
            self.current_size_bytes -= entry['size']

        # Remove thumbnails (find all sizes for this hash)
        keys_to_remove = []
        for key in self._thumbnail_cache:
            if key.startswith(f"{content_hash}_"):
                keys_to_remove.append(key)

        for key in keys_to_remove:
            entry = self._thumbnail_cache.pop(key)
            self.current_size_bytes -= entry['size']

    def get_hit_rate(self) -> float:
        """Get cache hit rate."""
        return self._metrics.hit_rate

    def get_performance_metrics(self) -> Dict[str, Any]:
        """Get detailed performance metrics."""
        return {
            'total_requests': self._metrics.total_requests,
            'cache_hits': self._metrics.cache_hits,
            'cache_misses': self._metrics.cache_misses,
            'hit_rate': self._metrics.hit_rate,
            'evictions': self._metrics.evictions,
            'current_size_bytes': self.current_size_bytes,
            'max_size_bytes': self.max_size_bytes,
            'size_utilization_percent': (self.current_size_bytes / self.max_size_bytes) * 100
        }

    def _ensure_capacity(self, required_size: int):
        """Ensure cache has capacity for new entry."""
        while (self.current_size_bytes + required_size) > self.max_size_bytes:
            if not self._metadata_cache and not self._thumbnail_cache:
                break  # Cache is empty

            # Evict based on strategy
            if self.strategy == CacheStrategy.LRU:
                self._evict_lru()
            elif self.strategy == CacheStrategy.FIFO:
                self._evict_fifo()
            else:  # TTL or default to LRU
                self._evict_lru()

    def _evict_lru(self):
        """Evict least recently used entry."""
        # Find oldest entry across both caches
        oldest_metadata = None
        oldest_thumbnail = None

        if self._metadata_cache:
            oldest_metadata = next(iter(self._metadata_cache))

        if self._thumbnail_cache:
            oldest_thumbnail = next(iter(self._thumbnail_cache))

        # Compare timestamps if both exist
        metadata_entry = self._metadata_cache.get(oldest_metadata) if oldest_metadata else None
        thumbnail_entry = self._thumbnail_cache.get(oldest_thumbnail) if oldest_thumbnail else None

        if metadata_entry and thumbnail_entry:
            if metadata_entry['timestamp'] <= thumbnail_entry['timestamp']:
                self._evict_metadata_entry(oldest_metadata)
            else:
                self._evict_thumbnail_entry(oldest_thumbnail)
        elif metadata_entry:
            self._evict_metadata_entry(oldest_metadata)
        elif thumbnail_entry:
            self._evict_thumbnail_entry(oldest_thumbnail)

    def _evict_fifo(self):
        """Evict first in, first out entry."""
        # For simplicity, just use LRU logic
        self._evict_lru()

    def _evict_metadata_entry(self, key: str):
        """Evict a metadata entry."""
        if key in self._metadata_cache:
            entry = self._metadata_cache.pop(key)
            self.current_size_bytes -= entry['size']
            if self.enable_metrics:
                self._metrics.evictions += 1

    def _evict_thumbnail_entry(self, key: str):
        """Evict a thumbnail entry."""
        if key in self._thumbnail_cache:
            entry = self._thumbnail_cache.pop(key)
            self.current_size_bytes -= entry['size']
            if self.enable_metrics:
                self._metrics.evictions += 1

    def clear(self):
        """Clear all cache entries."""
        self._metadata_cache.clear()
        self._thumbnail_cache.clear()
        self.current_size_bytes = 0
        self._metrics = CacheMetrics()