"""File-based content cache for L2 storage.

Stores cached content on disk with metadata sidecar files:
- Binary content: {hash}.bin
- Parsed text: {hash}.txt
- Metadata: {hash}.meta (JSON — replaces DB ContentCacheModel per data-storage-matrix.md)

Content is stored in: {cache_dir}/.cache/{zone_id}/{hash[:2]}/{hash[2:4]}/{hash}.*

Performance optimization:
- Uses Bloom filter for fast cache miss detection (avoids disk I/O)
- 99%+ of cache misses skip disk access entirely

Usage:
    file_cache = FileContentCache("/app/data")

    # Write content with metadata
    file_cache.write("zone1", "/mnt/gcs/file.txt", b"content",
                     text_content="parsed text",
                     meta={"content_hash": "abc", "content_type": "full"})

    # Read content
    content = file_cache.read("zone1", "/mnt/gcs/file.txt")

    # Read metadata
    meta = file_cache.read_meta("zone1", "/mnt/gcs/file.txt")

    # Delete content
    file_cache.delete("zone1", "/mnt/gcs/file.txt")
"""

import json as _json
import logging
import shutil
import threading
import time
from pathlib import Path
from typing import TYPE_CHECKING, Any

import blake3

# RUST_FALLBACK: BloomFilter (read_file / read_files_bulk are required)
if TYPE_CHECKING:
    from nexus_runtime import BloomFilter

logger = logging.getLogger(__name__)

# Default Bloom filter capacity (can be overridden)
DEFAULT_BLOOM_CAPACITY = 100_000
DEFAULT_BLOOM_FP_RATE = 0.01  # 1% false positive rate

# Default cache directory name
CACHE_DIR_NAME = ".cache"


class FileContentCache:
    """File-based content cache for L2 storage.

    Stores content on disk in a structured directory layout:
    {base_dir}/.cache/{zone_id}/{path_hash[:2]}/{path_hash[2:4]}/{path_hash}.bin

    Uses hash-based sharding to avoid too many files in a single directory.

    Performance optimization:
    - Uses Bloom filter to avoid disk I/O for cache misses
    - Bloom filter has ~1% false positive rate (harmless - falls through to disk check)
    - No false negatives (never skips existing files)
    """

    _bloom: "BloomFilter | None"

    def __init__(
        self,
        base_dir: str | Path,
        bloom_capacity: int = DEFAULT_BLOOM_CAPACITY,
        bloom_fp_rate: float = DEFAULT_BLOOM_FP_RATE,
    ):
        """Initialize file cache.

        Args:
            base_dir: Base directory for cache storage (e.g., /app/data)
            bloom_capacity: Expected number of cached items (default: 100,000)
            bloom_fp_rate: Target false positive rate (default: 0.01 = 1%)
        """
        self.base_dir = Path(base_dir)
        self.cache_dir = self.base_dir / CACHE_DIR_NAME
        self._bloom = None
        self._bloom_capacity = bloom_capacity
        self._bloom_fp_rate = bloom_fp_rate

        # Lease-aware staleness tracking (Issue #3400).
        # _leased_paths: paths with active leases (content is known-fresh).
        # _stale_paths:  paths whose leases were revoked (content may be stale).
        # Both are keyed by "zone_id:path_hash" for O(1) lookup.
        self._lease_lock = threading.Lock()
        self._leased_paths: set[str] = set()
        self._stale_paths: set[str] = set()

        self._ensure_cache_dir()
        self._init_bloom_filter()

    def _ensure_cache_dir(self) -> None:
        """Create cache directory if it doesn't exist."""
        self.cache_dir.mkdir(parents=True, exist_ok=True)

    def _init_bloom_filter(self) -> None:
        """Initialize Bloom filter for fast cache miss detection.

        Degrades gracefully to `_bloom=None` when nexus_runtime is absent or stale.
        """
        # RUST_FALLBACK: BloomFilter (optional — stale/absent binary disables bloom)
        from nexus._rust_compat import BloomFilter

        if BloomFilter is None:
            logger.debug(
                "BloomFilter unavailable (stale or absent nexus_runtime) — "
                "cache miss detection will use disk checks"
            )
            self._bloom = None
            return

        self._bloom = BloomFilter(self._bloom_capacity, self._bloom_fp_rate)
        self._populate_bloom_from_disk()
        logger.debug(
            f"Bloom filter initialized: capacity={self._bloom_capacity}, "
            f"fp_rate={self._bloom_fp_rate}, memory={self._bloom.memory_bytes} bytes"
        )

    def _populate_bloom_from_disk(self) -> None:
        """Populate Bloom filter from existing cache entries on disk.

        Scans the cache directory and adds all cached paths to the Bloom filter.
        This is called on startup to ensure the filter reflects disk state.
        """
        if self._bloom is None or not self.cache_dir.exists():
            return

        keys: list[str] = []
        try:
            for zone_dir in self.cache_dir.iterdir():
                if not zone_dir.is_dir():
                    continue
                zone_id = zone_dir.name
                # Scan all .bin files in zone directory
                for cache_file in zone_dir.rglob("*.bin"):
                    # Extract hash from filename (remove .bin extension)
                    file_hash = cache_file.stem
                    # We store keys as "zone_id:hash" since we can't recover original path
                    # This works because _bloom_key uses the same format
                    keys.append(f"{zone_id}:{file_hash}")

            if keys:
                self._bloom.add_bulk(keys)
                logger.info(f"Bloom filter populated with {len(keys)} entries from disk")
        except Exception as e:
            logger.warning(f"Failed to populate Bloom filter from disk: {e}")

    # =================================================================
    # Lease-aware staleness (Issue #3400)
    # =================================================================

    def _lease_key(self, zone_id: str, virtual_path: str) -> str:
        """Compose a lease-tracking key from zone + path hash."""
        return f"{zone_id}:{self._path_hash(virtual_path)}"

    def mark_lease_acquired(self, zone_id: str, virtual_path: str) -> None:
        """Record that a lease has been acquired for a path.

        The cached content for this path is now known-fresh.  Clears any
        prior staleness marker so subsequent reads return cached data.
        """
        key = self._lease_key(zone_id, virtual_path)
        with self._lease_lock:
            self._leased_paths.add(key)
            self._stale_paths.discard(key)

    def mark_lease_revoked(self, zone_id: str, virtual_path: str) -> None:
        """Record that a lease has been revoked for a path.

        The cached content may now be stale — subsequent reads for this
        path will return ``None`` until fresh content is written.
        """
        key = self._lease_key(zone_id, virtual_path)
        with self._lease_lock:
            self._leased_paths.discard(key)
            self._stale_paths.add(key)

    def is_stale(self, zone_id: str, virtual_path: str) -> bool:
        """Check whether cached content for a path is marked stale."""
        key = self._lease_key(zone_id, virtual_path)
        with self._lease_lock:
            return key in self._stale_paths

    def has_active_lease(self, zone_id: str, virtual_path: str) -> bool:
        """Check whether a path has an active lease (content is fresh)."""
        key = self._lease_key(zone_id, virtual_path)
        with self._lease_lock:
            return key in self._leased_paths

    def _bloom_key(self, zone_id: str, virtual_path: str) -> str:
        """Generate Bloom filter key for a cache entry.

        Uses zone_id:path_hash format to match what we store on disk.
        """
        path_hash = self._path_hash(virtual_path)
        return f"{zone_id}:{path_hash}"

    def _bloom_check(self, zone_id: str, virtual_path: str) -> bool:
        """Check Bloom filter for possible existence.

        Returns:
            True if entry might exist (need to check disk)
            False if entry definitely does not exist (skip disk I/O)
        """
        if self._bloom is None:
            return True  # No Bloom filter, always check disk
        return bool(self._bloom.might_exist(self._bloom_key(zone_id, virtual_path)))

    def _bloom_add(self, zone_id: str, virtual_path: str) -> None:
        """Add entry to Bloom filter after writing to disk."""
        if self._bloom is not None:
            self._bloom.add(self._bloom_key(zone_id, virtual_path))

    def _path_hash(self, virtual_path: str) -> str:
        """Generate a hash for the virtual path.

        Uses BLAKE3 truncated to 32 chars for reasonable uniqueness
        while keeping filenames manageable. BLAKE3 is ~10x faster than SHA-256.
        """
        hash_hex: str = blake3.blake3(virtual_path.encode()).hexdigest()
        return hash_hex[:32]

    @staticmethod
    def _safe_zone_id(zone_id: str) -> str:
        """Validate zone_id is safe for filesystem path construction."""
        from nexus.lib.zone import validate_zone_id

        return validate_zone_id(zone_id)

    def _get_cache_path(self, zone_id: str, virtual_path: str) -> Path:
        """Get the file path for cached content.

        Uses hash-based sharding: {zone}/{hash[:2]}/{hash[2:4]}/{hash}.bin
        This prevents too many files in a single directory (max 256 per level).
        """
        path_hash = self._path_hash(virtual_path)
        safe_zone = self._safe_zone_id(zone_id)
        return self.cache_dir / safe_zone / path_hash[:2] / path_hash[2:4] / f"{path_hash}.bin"

    def _get_text_cache_path(self, zone_id: str, virtual_path: str) -> Path:
        """Get the file path for cached parsed text."""
        path_hash = self._path_hash(virtual_path)
        safe_zone = self._safe_zone_id(zone_id)
        return self.cache_dir / safe_zone / path_hash[:2] / path_hash[2:4] / f"{path_hash}.txt"

    def _get_meta_cache_path(self, zone_id: str, virtual_path: str) -> Path:
        """Get the file path for cached metadata (replaces DB ContentCacheModel)."""
        path_hash = self._path_hash(virtual_path)
        safe_zone = self._safe_zone_id(zone_id)
        return self.cache_dir / safe_zone / path_hash[:2] / path_hash[2:4] / f"{path_hash}.meta"

    def write(
        self,
        zone_id: str,
        virtual_path: str,
        content: bytes,
        text_content: str | None = None,
        meta: dict[str, Any] | None = None,
    ) -> Path:
        """Write content to cache.

        Args:
            zone_id: Zone ID
            virtual_path: Virtual file path
            content: Binary content to cache
            text_content: Optional parsed text content
            meta: Optional metadata dict (replaces DB ContentCacheModel).
                  Keys: content_hash, content_type, original_size, cached_size,
                  backend_version, synced_at, parsed_from, parse_metadata, zone_id.

        Returns:
            Path to the cached binary file
        """
        # Write binary content
        cache_path = self._get_cache_path(zone_id, virtual_path)
        cache_path.parent.mkdir(parents=True, exist_ok=True)

        try:
            cache_path.write_bytes(content)
            logger.debug(f"Cached {len(content)} bytes to {cache_path}")
        except Exception as e:
            logger.error(f"Failed to write cache file {cache_path}: {e}")
            raise

        # Clear staleness AFTER successful write — if write_bytes() failed
        # the stale marker must survive so reads don't serve old data.
        key = self._lease_key(zone_id, virtual_path)
        with self._lease_lock:
            self._stale_paths.discard(key)

        # Write text content if provided
        if text_content:
            text_path = self._get_text_cache_path(zone_id, virtual_path)
            try:
                text_path.write_text(text_content, encoding="utf-8")
            except Exception as e:
                logger.warning(f"Failed to write text cache {text_path}: {e}")

        # Write metadata sidecar if provided
        if meta:
            self.write_meta(zone_id, virtual_path, meta)

        # Add to Bloom filter for fast future lookups
        self._bloom_add(zone_id, virtual_path)

        return cache_path

    def write_meta(
        self,
        zone_id: str,
        virtual_path: str,
        meta: dict[str, Any],
    ) -> None:
        """Write metadata sidecar file (replaces DB ContentCacheModel).

        Args:
            zone_id: Zone ID
            virtual_path: Virtual file path
            meta: Metadata dict with cache entry properties
        """
        meta_path = self._get_meta_cache_path(zone_id, virtual_path)
        meta_path.parent.mkdir(parents=True, exist_ok=True)
        try:
            meta_path.write_text(_json.dumps(meta), encoding="utf-8")
        except Exception as e:
            logger.warning(f"Failed to write meta cache {meta_path}: {e}")

    def read_meta(self, zone_id: str, virtual_path: str) -> dict[str, Any] | None:
        """Read metadata sidecar file.

        Args:
            zone_id: Zone ID
            virtual_path: Virtual file path

        Returns:
            Metadata dict, or None if not cached or stale
        """
        if not self._bloom_check(zone_id, virtual_path):
            return None

        if self.is_stale(zone_id, virtual_path):
            return None

        meta_path = self._get_meta_cache_path(zone_id, virtual_path)
        if not meta_path.exists():
            return None

        try:
            data: dict[str, Any] = _json.loads(meta_path.read_text(encoding="utf-8"))
            return data
        except Exception as e:
            logger.warning(f"Failed to read meta cache {meta_path}: {e}")
            return None

    def read_meta_bulk(
        self,
        zone_id: str,
        virtual_paths: list[str],
    ) -> dict[str, dict[str, Any]]:
        """Read multiple metadata sidecar files.

        Args:
            zone_id: Zone ID
            virtual_paths: List of virtual file paths

        Returns:
            Dict mapping virtual_path to metadata dict (only for cached files)
        """
        result: dict[str, dict[str, Any]] = {}
        for path in virtual_paths:
            meta = self.read_meta(zone_id, path)
            if meta is not None:
                result[path] = meta
        return result

    def read(self, zone_id: str, virtual_path: str) -> bytes | None:
        """Read binary content from cache.

        Uses Bloom filter for fast cache miss detection - avoids disk I/O
        for entries that definitely don't exist.

        Uses mmap-based reading via nexus_runtime for better performance:
        - Leverages OS page cache efficiently
        - 20-70% faster for medium to large files

        Returns ``None`` (forcing a re-fetch from the backend) when:
        - The entry doesn't exist (Bloom filter negative)
        - The entry has been marked stale by lease revocation (Issue #3400)

        Args:
            zone_id: Zone ID
            virtual_path: Virtual file path

        Returns:
            Cached content bytes, or None if not cached or stale
        """
        # Fast path: Bloom filter says entry definitely doesn't exist
        if not self._bloom_check(zone_id, virtual_path):
            return None

        # Staleness check: lease was revoked → content may be stale
        if self.is_stale(zone_id, virtual_path):
            return None

        cache_path = self._get_cache_path(zone_id, virtual_path)

        from nexus._rust_compat import read_file

        try:
            if read_file is not None:
                data: bytes | None = read_file(str(cache_path))
                return data
            if not cache_path.exists():
                return None
            return cache_path.read_bytes()
        except Exception as e:
            logger.warning(f"Failed to read cache file {cache_path}: {e}")
            return None

    def read_if_fresh(
        self,
        zone_id: str,
        virtual_path: str,
        expected_fingerprint: str | None,
    ) -> bytes | None:
        """Read cached bytes only when metadata proves the entry is fresh."""
        meta = self.read_meta(zone_id, virtual_path)
        if meta is None:
            return None

        if expected_fingerprint is not None:
            if meta.get("fingerprint") != expected_fingerprint:
                return None
        else:
            expires_at = meta.get("expires_at")
            if expires_at is not None and expires_at < time.time():
                return None

        return self.read(zone_id, virtual_path)

    def read_text(self, zone_id: str, virtual_path: str) -> str | None:
        """Read parsed text content from cache.

        Uses Bloom filter for fast cache miss detection - avoids disk I/O
        for entries that definitely don't exist.

        Args:
            zone_id: Zone ID
            virtual_path: Virtual file path

        Returns:
            Cached text content, or None if not cached or stale
        """
        # Fast path: Bloom filter says entry definitely doesn't exist
        # (text file uses same path hash as binary file)
        if not self._bloom_check(zone_id, virtual_path):
            return None

        # Staleness check
        if self.is_stale(zone_id, virtual_path):
            return None

        text_path = self._get_text_cache_path(zone_id, virtual_path)

        if not text_path.exists():
            return None

        try:
            return text_path.read_text(encoding="utf-8")
        except Exception as e:
            logger.warning(f"Failed to read text cache {text_path}: {e}")
            return None

    def read_bulk(
        self,
        zone_id: str,
        virtual_paths: list[str],
    ) -> dict[str, bytes]:
        """Read multiple files from cache.

        Uses parallel mmap-based reading via nexus_runtime for better performance
        when reading many files (10+ files uses parallel I/O).

        Args:
            zone_id: Zone ID
            virtual_paths: List of virtual file paths

        Returns:
            Dict mapping virtual_path to content (only for cached files)
        """
        # Filter out paths that definitely don't exist (via Bloom filter)
        # and paths that are marked stale (lease revoked)
        paths_to_check = [
            p
            for p in virtual_paths
            if self._bloom_check(zone_id, p) and not self.is_stale(zone_id, p)
        ]

        if not paths_to_check:
            return {}

        # Build mapping: cache_path -> virtual_path
        cache_to_virtual: dict[str, str] = {}
        cache_paths: list[str] = []
        for vpath in paths_to_check:
            cache_path = str(self._get_cache_path(zone_id, vpath))
            cache_to_virtual[cache_path] = vpath
            cache_paths.append(cache_path)

        from nexus._rust_compat import read_files_bulk

        if read_files_bulk is not None:
            cache_contents = read_files_bulk(cache_paths)
        else:
            cache_contents = {}
            for cache_path in cache_paths:
                path = Path(cache_path)
                if path.exists():
                    cache_contents[cache_path] = path.read_bytes()

        # Map back to virtual paths
        result: dict[str, bytes] = {}
        for cache_path, content in cache_contents.items():
            virtual_path = cache_to_virtual.get(cache_path)
            if virtual_path:
                result[virtual_path] = content
        return result

    def read_text_bulk(
        self,
        zone_id: str,
        virtual_paths: list[str],
    ) -> dict[str, str]:
        """Read multiple text files from cache.

        Args:
            zone_id: Zone ID
            virtual_paths: List of virtual file paths

        Returns:
            Dict mapping virtual_path to text content (only for cached files)
        """
        result = {}
        for path in virtual_paths:
            text = self.read_text(zone_id, path)
            if text is not None:
                result[path] = text
        return result

    def get_disk_paths_bulk(
        self,
        zone_id: str,
        virtual_paths: list[str],
    ) -> dict[str, str]:
        """Get disk paths for cached files (Issue #893).

        Returns the actual disk paths where cached content is stored,
        enabling mmap-based operations to bypass Python.

        Args:
            zone_id: Zone ID
            virtual_paths: List of virtual file paths

        Returns:
            Dict mapping virtual_path to disk_path (only for existing cached files)
        """
        # Filter out paths that definitely don't exist (via Bloom filter)
        paths_to_check = [p for p in virtual_paths if self._bloom_check(zone_id, p)]

        if not paths_to_check:
            return {}

        result: dict[str, str] = {}
        for vpath in paths_to_check:
            cache_path = self._get_cache_path(zone_id, vpath)
            if cache_path.exists():
                result[vpath] = str(cache_path)

        return result

    def exists(self, zone_id: str, virtual_path: str) -> bool:
        """Check if content is cached.

        Uses Bloom filter for fast cache miss detection - avoids disk I/O
        for entries that definitely don't exist.

        Args:
            zone_id: Zone ID
            virtual_path: Virtual file path

        Returns:
            True if content is cached
        """
        # Fast path: Bloom filter says entry definitely doesn't exist
        if not self._bloom_check(zone_id, virtual_path):
            return False

        return self._get_cache_path(zone_id, virtual_path).exists()

    def delete(self, zone_id: str, virtual_path: str) -> bool:
        """Delete cached content (binary, text, and metadata sidecar).

        Args:
            zone_id: Zone ID
            virtual_path: Virtual file path

        Returns:
            True if content was deleted
        """
        deleted = False

        # Delete binary cache
        cache_path = self._get_cache_path(zone_id, virtual_path)
        if cache_path.exists():
            try:
                cache_path.unlink()
                deleted = True
            except Exception as e:
                logger.warning(f"Failed to delete cache file {cache_path}: {e}")

        # Delete text cache
        text_path = self._get_text_cache_path(zone_id, virtual_path)
        if text_path.exists():
            try:
                text_path.unlink()
            except Exception as e:
                logger.warning(f"Failed to delete text cache {text_path}: {e}")

        # Delete metadata sidecar
        meta_path = self._get_meta_cache_path(zone_id, virtual_path)
        if meta_path.exists():
            try:
                meta_path.unlink()
            except Exception as e:
                logger.warning(f"Failed to delete meta cache {meta_path}: {e}")

        # Clear staleness / lease state only AFTER files are deleted.
        # If deletion failed, preserve stale markers so reads don't
        # silently serve the old on-disk content.
        if deleted:
            key = self._lease_key(zone_id, virtual_path)
            with self._lease_lock:
                self._stale_paths.discard(key)
                self._leased_paths.discard(key)

        return deleted

    def delete_zone(self, zone_id: str) -> int:
        """Delete all cached content for a zone.

        Args:
            zone_id: Zone ID

        Returns:
            Number of files deleted
        """
        safe_zone = self._safe_zone_id(zone_id)
        zone_dir = self.cache_dir / safe_zone
        if not zone_dir.exists():
            return 0

        count = sum(1 for _ in zone_dir.rglob("*") if _.is_file())

        try:
            shutil.rmtree(zone_dir)
            logger.info(f"Deleted {count} cached files for zone {zone_id}")
        except Exception as e:
            logger.error(f"Failed to delete zone cache {zone_dir}: {e}")
            return 0

        return count

    def get_cache_stats(self) -> dict:
        """Get cache statistics.

        Returns:
            Dict with cache stats (file count, total size, by zone)
        """
        stats: dict[str, Any] = {
            "total_files": 0,
            "total_size_bytes": 0,
            "zones": {},
        }

        if not self.cache_dir.exists():
            return stats

        for zone_dir in self.cache_dir.iterdir():
            if not zone_dir.is_dir():
                continue

            zone_id = zone_dir.name
            zone_files = 0
            zone_size = 0

            for cache_file in zone_dir.rglob("*.bin"):
                zone_files += 1
                zone_size += cache_file.stat().st_size

            stats["zones"][zone_id] = {
                "files": zone_files,
                "size_bytes": zone_size,
            }
            stats["total_files"] += zone_files
            stats["total_size_bytes"] += zone_size

        return stats

    def get_zoekt_index_path(self) -> Path:
        """Get the path that Zoekt should index.

        Returns:
            Path to the cache directory for Zoekt indexing
        """
        return self.cache_dir
