"""FUSE mount management for Nexus filesystem.

This module provides the high-level interface for mounting Nexus as a
FUSE filesystem, including mount mode management and lifecycle control.

Issue #1076: Added automatic cache warmup on mount for faster first access.
"""

import asyncio
import logging
import threading
from enum import Enum
from pathlib import Path
from typing import TYPE_CHECKING, Any, cast

from fuse import FUSE

if TYPE_CHECKING:
    from nexus.core.nexus_fs import NexusFS

from nexus.fuse.operations import NexusFUSEOperations
from nexus.fuse.passthrough import (
    PassthroughOptions,
    RustPassthroughMount,
    mount_is_passthrough_safe,
)

logger = logging.getLogger(__name__)


class MountMode(Enum):
    """Mount mode for FUSE filesystem.

    - BINARY: Return raw file content (no parsing)
    - TEXT: Parse all files and return text representation
    - SMART: Auto-detect file type and return appropriate format
    """

    BINARY = "binary"
    TEXT = "text"
    SMART = "smart"


class NexusFUSE:
    """FUSE mount manager for Nexus filesystem.

    This class manages the lifecycle of a FUSE mount, including starting,
    stopping, and monitoring the mount.

    Example:
        >>> from nexus import connect
        >>> from nexus.fuse import NexusFUSE, MountMode
        >>>
        >>> nx = connect(config={"data_dir": "./nexus-data"})
        >>> fuse = NexusFUSE(nx, "/mnt/nexus", mode=MountMode.SMART)
        >>> fuse.mount(foreground=False)
        >>> # ... use filesystem ...
        >>> fuse.unmount()
    """

    def __init__(
        self,
        nexus_fs: "NexusFS",
        mount_point: str,
        mode: MountMode = MountMode.SMART,
        cache_config: dict[str, int | bool] | None = None,
        warmup_depth: int = 2,
        warmup_max_files: int = 1000,
        agent_id: str | None = None,
        subject_type: str | None = None,
        owner_id: str | None = None,
        zone_id: str | None = None,
        use_rust: bool = False,
        passthrough_enabled: bool | None = None,
        passthrough_patterns: list[str] | None = None,
        passthrough_deny_patterns: list[str] | None = None,
        passthrough_threshold_bytes: int | None = None,
        passthrough_require: bool = False,
        passthrough_backing_dir: str | Path | None = None,
        lease_manager: Any | None = None,
        file_cache: Any | None = None,
    ) -> None:
        """Initialize FUSE mount manager.

        Args:
            nexus_fs: Nexus filesystem instance to mount
            mount_point: Local path to mount the filesystem
            mode: Mount mode (binary, text, or smart)
            cache_config: Optional cache configuration dict with keys:
                         - content_cache_bytes: int (default: 512*1024*1024)
                         - parsed_cache_bytes: int (default: 64*1024*1024)
                         - max_drain_bytes: int (default: 16*1024*1024)
                         - attr_cache_ttl: int (default: 60)
                         - dir_cache_ttl: int (default: 5)
                         - index_ttl_overrides: dict[str, int] (default: {})
                         - enable_metrics: bool (default: False)
            warmup_depth: Directory depth for automatic warmup (default: 2)
            warmup_max_files: Maximum files to warm (default: 1000)
            agent_id: Optional agent ID for namespace-scoped mounts (Issue #1305).
                     When set, the mount is scoped to this agent's namespace.
                     When None, the global view is used (backward compatible).
            subject_type: Subject type for OperationContext (A4-B: None sentinel).
                         When None and agent_id is set, defaults to "agent".
                         Pass explicit value to override (e.g., "service").
            owner_id: Human owner ID for the agent (defaults to agent_id).
            zone_id: Zone ID for multi-zone isolation.
            use_rust: Use high-performance Rust FUSE daemon (Issue #1569).
            passthrough_enabled: Route safe mounts to Rust-owned passthrough.
            passthrough_patterns: Include patterns for Rust passthrough.
            passthrough_deny_patterns: Exclude patterns for Rust passthrough.
            passthrough_threshold_bytes: Minimum passthrough size. Defaults to env.
            passthrough_require: Raise instead of falling back when unsafe.
            passthrough_backing_dir: Optional local backing directory.

        Note:
            Issue #1076: Cache warmup runs automatically after mount in background.
            This is always enabled as it's non-blocking and reduces first-access latency.
        """
        self.nexus_fs = nexus_fs
        self.mount_point = Path(mount_point)
        self.mode = mode
        self.cache_config = cache_config
        self.warmup_depth = warmup_depth
        self.warmup_max_files = warmup_max_files
        self._agent_id = agent_id
        self._subject_type = subject_type
        self._owner_id = owner_id
        self._zone_id = zone_id
        self._use_rust = use_rust
        env_options = PassthroughOptions.from_env()
        self._passthrough_options = PassthroughOptions(
            enabled=env_options.enabled if passthrough_enabled is None else passthrough_enabled,
            patterns=(
                passthrough_patterns if passthrough_patterns is not None else env_options.patterns
            ),
            deny_patterns=(
                passthrough_deny_patterns
                if passthrough_deny_patterns is not None
                else env_options.deny_patterns
            ),
            threshold_bytes=(
                passthrough_threshold_bytes
                if passthrough_threshold_bytes is not None
                else env_options.threshold_bytes
            ),
            require=passthrough_require or env_options.require,
            backing_dir=(
                Path(passthrough_backing_dir)
                if passthrough_backing_dir
                else env_options.backing_dir
            ),
        )
        self._rust_passthrough_mount: RustPassthroughMount | None = None
        self._lease_manager = lease_manager
        self._file_cache = file_cache
        # Generate unique mount ID for lease holder identity (Decision 2A)
        import uuid

        self._mount_id = f"mount-{uuid.uuid4().hex[:12]}"
        self.fuse: FUSE | None = None
        self._mount_thread: threading.Thread | None = None
        self._mounted = False
        self._warmup_thread: threading.Thread | None = None
        # Issue #1115: Event loop for async event dispatch from FUSE operations
        self._event_loop: asyncio.AbstractEventLoop | None = None
        self._event_loop_thread: threading.Thread | None = None

    def mount(
        self,
        foreground: bool = True,
        allow_other: bool = False,
        debug: bool = False,
    ) -> None:
        """Mount the Nexus filesystem.

        Args:
            foreground: If True, run in foreground (blocking); if False, run in background
            allow_other: If True, allow other users to access the mount
            debug: If True, enable FUSE debug output

        Raises:
            RuntimeError: If already mounted or mount fails
            FileNotFoundError: If mount point doesn't exist
        """
        if self._mounted:
            raise RuntimeError("Filesystem is already mounted")

        # Ensure mount point exists
        if not self.mount_point.exists():
            raise FileNotFoundError(f"Mount point does not exist: {self.mount_point}")

        # Check if mount point is a directory
        if not self.mount_point.is_dir():
            raise ValueError(f"Mount point is not a directory: {self.mount_point}")

        # Check if mount point is empty
        if list(self.mount_point.iterdir()):
            logger.warning(f"Mount point is not empty: {self.mount_point}")

        # Issue #1305: Build OperationContext for namespace-scoped mounts
        context = None
        namespace_manager = None
        if self._agent_id is not None:
            from nexus.contracts.types import OperationContext

            # A4-B: Use sentinel default — only override to "agent" when caller
            # left subject_type at its default value (None sentinel).
            subject_type = self._subject_type if self._subject_type is not None else "agent"

            context = OperationContext(
                user_id=self._owner_id or self._agent_id,
                agent_id=self._agent_id,
                subject_type=subject_type,
                subject_id=self._agent_id,
                zone_id=self._zone_id,
                groups=[],
            )
            logger.info(
                f"[FUSE] Namespace-scoped mount: agent_id={self._agent_id}, "
                f"subject_type={subject_type}, zone_id={self._zone_id}"
            )

            # A2-B: Try to obtain NamespaceManager for direct visibility checks
            namespace_manager = getattr(self.nexus_fs, "namespace_manager", None)

        if self._should_use_rust_passthrough(context):
            self._start_rust_passthrough_mount()
            return

        # Create FUSE operations
        # Issue #1771: event_bus via kernel-level _event_bus
        event_bus = getattr(self.nexus_fs, "_event_bus", None) if self.nexus_fs else None
        subscription_manager = getattr(self.nexus_fs, "subscription_manager", None)
        operations = NexusFUSEOperations(
            self.nexus_fs,
            self.mode,
            self.cache_config,
            context=context,
            namespace_manager=namespace_manager,
            use_rust=self._use_rust,
            event_bus=event_bus,
            subscription_manager=subscription_manager,
            lease_manager=self._lease_manager,
            mount_id=self._mount_id,
            file_cache=self._file_cache,
        )

        # Issue #1115: Set up event loop for async event dispatch
        # FUSE operations are synchronous, but event dispatch is async.
        # We create a dedicated event loop thread and pass it to operations.
        def start_event_loop() -> None:
            """Run event loop in dedicated thread for async event dispatch."""
            self._event_loop = asyncio.new_event_loop()
            asyncio.set_event_loop(self._event_loop)
            operations.set_event_loop(self._event_loop)
            logger.info("[FUSE] Event loop started for async event dispatch")
            self._event_loop.run_forever()

        self._event_loop_thread = threading.Thread(
            target=start_event_loop, daemon=True, name="fuse-event-loop"
        )
        self._event_loop_thread.start()

        # Wait for event loop to be ready
        import time

        for _ in range(50):  # 500ms timeout
            if self._event_loop is not None:
                break
            time.sleep(0.01)

        if self._event_loop is None:
            logger.warning("[FUSE] Event loop failed to start, events will be disabled")

        # Build FUSE options
        # Note: Always use foreground=True because we handle backgrounding ourselves via threading
        fuse_options: dict[str, bool | str | int] = {
            "nothreads": False,
            "foreground": True,  # Always run FUSE in foreground mode (within our thread)
            "debug": debug,
        }

        if allow_other:
            fuse_options["allow_other"] = True

        # Add macOS-specific options to prevent Spotlight indexing and reduce overhead
        # These options tell macOS not to index the filesystem and disable extended attributes
        import platform

        if platform.system() == "Darwin":  # macOS
            logger.info("Applying macOS-specific FUSE options to reduce Spotlight indexing")
            fuse_options.update(
                {
                    "volname": "Nexus",  # Custom volume name
                    "noappledouble": True,  # Disable ._* AppleDouble files
                    "noapplexattr": True,  # Disable extended attributes
                    "daemon_timeout": 600,  # Keep daemon alive longer
                    "auto_cache": True,  # Enable automatic kernel caching
                }
            )
            logger.info(f"FUSE options: {fuse_options}")

        # Mount filesystem
        logger.info(f"Mounting Nexus to {self.mount_point} (mode={self.mode.value})")

        if foreground:
            # Run in foreground (blocking)
            self._mounted = True
            try:
                self.fuse = FUSE(
                    operations,
                    str(self.mount_point),
                    **fuse_options,
                )
            finally:
                self._mounted = False
        else:
            # Run in background thread
            def mount_thread() -> None:
                try:
                    self.fuse = FUSE(
                        operations,
                        str(self.mount_point),
                        **fuse_options,
                    )
                except Exception as e:
                    logger.error(f"FUSE mount error: {e}")
                finally:
                    self._mounted = False

            # Use daemon=False so the thread keeps the process alive
            self._mount_thread = threading.Thread(target=mount_thread, daemon=False)
            self._mount_thread.start()
            self._mounted = True

            # Wait a bit to ensure mount succeeds
            import time

            time.sleep(1)

            if not self._mounted:
                raise RuntimeError("Failed to mount filesystem")

            logger.info(f"Mounted Nexus to {self.mount_point} (background)")

            # Issue #1076: Automatic cache warmup after mount
            # Always enabled - non-blocking, lightweight, reduces first-access latency
            self._start_warmup()

    def _should_use_rust_passthrough(self, context: Any | None) -> bool:
        if not (self._use_rust and self._passthrough_options.enabled):
            return False
        if self._rust_passthrough_credentials() is None:
            if self._passthrough_options.require:
                raise RuntimeError(
                    "FUSE passthrough requires a remote NexusFS with passthrough credentials"
                )
            return False
        if not mount_is_passthrough_safe(
            self.nexus_fs,
            mode_value=self.mode.value,
            context=context,
        ):
            if self._passthrough_options.require:
                raise RuntimeError("FUSE passthrough is not safe for this mount configuration")
            return False
        return True

    def _rust_passthrough_credentials(self) -> tuple[str, str] | None:
        missing = object()
        remote_fs = cast(Any, self.nexus_fs)

        nexus_url = getattr(remote_fs, "_base_url", missing)
        if nexus_url is missing:
            nexus_url = getattr(remote_fs, "_remote_base_url", missing)
        if nexus_url is missing:
            return None

        api_key = getattr(remote_fs, "_api_key", missing)
        if api_key is missing:
            api_key = getattr(remote_fs, "_remote_api_key", "")
        if api_key is None:
            api_key = ""

        return str(nexus_url), str(api_key)

    def _start_rust_passthrough_mount(self) -> None:
        credentials = self._rust_passthrough_credentials()
        if credentials is None:
            raise RuntimeError(
                "FUSE passthrough requires a remote NexusFS with passthrough credentials"
            )
        nexus_url, api_key = credentials
        self._rust_passthrough_mount = RustPassthroughMount.create(
            nexus_url=nexus_url,
            api_key=api_key,
            mount_point=self.mount_point,
            options=self._passthrough_options,
            agent_id=getattr(self.nexus_fs, "_agent_id", self._agent_id),
        )
        self._rust_passthrough_mount.start()
        self._mounted = True
        self._start_warmup()

    def _start_warmup(self) -> None:
        """Start background cache warmup (Issue #1076).

        Runs warmup in a separate thread to not block mount completion.
        """

        def warmup_thread() -> None:
            try:
                import importlib as _il

                _cw = _il.import_module("nexus.server.cache_warmer")
                CacheWarmer = _cw.CacheWarmer
                WarmupConfig = _cw.WarmupConfig

                logger.info(
                    f"[WARMUP] Starting automatic warmup for mount at / "
                    f"(depth={self.warmup_depth}, max_files={self.warmup_max_files})"
                )

                config = WarmupConfig(
                    max_files=self.warmup_max_files,
                    depth=self.warmup_depth,
                    include_content=False,  # Metadata only for fast warmup
                )

                warmer = CacheWarmer(nexus_fs=self.nexus_fs, config=config)

                # Run async warmup in sync context
                async def do_warmup() -> Any:
                    return await warmer.warmup_directory(
                        path="/",
                        depth=self.warmup_depth,
                        include_content=False,
                        max_files=self.warmup_max_files,
                    )

                # Create new event loop for this thread
                loop = asyncio.new_event_loop()
                asyncio.set_event_loop(loop)
                try:
                    stats = loop.run_until_complete(do_warmup())
                    logger.info(f"[WARMUP] Mount warmup complete: {stats.to_dict()}")
                finally:
                    loop.close()

            except Exception as e:
                logger.warning(f"[WARMUP] Mount warmup failed (non-critical): {e}")

        self._warmup_thread = threading.Thread(target=warmup_thread, daemon=True)
        self._warmup_thread.start()

    def _stop_rust_passthrough_mount(self) -> None:
        rust_mount = self._rust_passthrough_mount
        if rust_mount is None:
            return
        rust_mount.stop()
        self._rust_passthrough_mount = None

    def unmount(self) -> None:
        """Unmount the filesystem.

        Raises:
            RuntimeError: If not mounted
        """
        if not self._mounted:
            if self._rust_passthrough_mount is not None:
                self._stop_rust_passthrough_mount()
                return
            raise RuntimeError("Filesystem is not mounted")

        logger.info(f"Unmounting {self.mount_point}")

        # Use platform-specific unmount command
        import platform
        import subprocess

        system = platform.system()

        try:
            if system == "Darwin":  # macOS
                subprocess.run(
                    ["umount", str(self.mount_point)],
                    check=True,
                    capture_output=True,
                )
            elif system == "Linux":
                subprocess.run(
                    ["fusermount", "-u", str(self.mount_point)],
                    check=True,
                    capture_output=True,
                )
            else:
                raise RuntimeError(f"Unsupported platform: {system}")

            self._mounted = False
            logger.info(f"Unmounted {self.mount_point}")

            # Issue #1115: Stop event loop thread
            if self._event_loop is not None:
                self._event_loop.call_soon_threadsafe(self._event_loop.stop)
                self._event_loop = None
                logger.info("[FUSE] Event loop stopped")

            # Issue #3397: Close lease coordinator
            if hasattr(self, "_operations") and hasattr(self._operations, "_ctx"):
                coordinator = self._operations._ctx.cache
                if hasattr(coordinator, "close"):
                    coordinator.close()

            self._stop_rust_passthrough_mount()

        except subprocess.CalledProcessError as e:
            stderr = e.stderr.decode()
            logger.error(f"Failed to unmount: {stderr}")
            try:
                self._stop_rust_passthrough_mount()
            except Exception as stop_error:
                logger.error(f"Failed to stop Rust passthrough mount: {stop_error}")
            raise RuntimeError(f"Failed to unmount: {stderr}") from e

    def is_mounted(self) -> bool:
        """Check if filesystem is currently mounted.

        Returns:
            True if mounted, False otherwise
        """
        return self._mounted

    def wait(self) -> None:
        """Wait for the mount thread to finish.

        This is useful when running in background mode to keep the
        process alive.
        """
        if self._mount_thread and self._mount_thread.is_alive():
            self._mount_thread.join()

    def __enter__(self) -> "NexusFUSE":
        """Context manager entry."""
        return self

    def __exit__(self, exc_type: type, exc_val: Exception, exc_tb: object) -> None:
        """Context manager exit."""
        if self._mounted:
            try:
                self.unmount()
            except Exception as e:
                logger.error(f"Error unmounting in context manager: {e}")


def mount_nexus(
    nexus_fs: "NexusFS",
    mount_point: str,
    mode: str = "smart",
    foreground: bool = True,
    allow_other: bool = False,
    debug: bool = False,
    cache_config: dict[str, int | bool] | None = None,
    warmup_depth: int = 2,
    warmup_max_files: int = 1000,
    agent_id: str | None = None,
    subject_type: str | None = None,
    owner_id: str | None = None,
    zone_id: str | None = None,
    use_rust: bool = False,
    passthrough_enabled: bool | None = None,
    passthrough_patterns: list[str] | None = None,
    passthrough_deny_patterns: list[str] | None = None,
    passthrough_threshold_bytes: int | None = None,
    passthrough_require: bool = False,
    passthrough_backing_dir: str | Path | None = None,
    lease_manager: Any | None = None,
    file_cache: Any | None = None,
) -> "NexusFUSE":
    """Convenience function to mount Nexus filesystem.

    Args:
        nexus_fs: Nexus filesystem instance
        mount_point: Local path to mount
        mode: Mount mode ("binary", "text", or "smart")
        foreground: Run in foreground (blocking)
        allow_other: Allow other users to access the mount
        debug: Enable FUSE debug output
        cache_config: Optional cache configuration dict with keys:
                     - content_cache_bytes: int (default: 512*1024*1024)
                     - parsed_cache_bytes: int (default: 64*1024*1024)
                     - max_drain_bytes: int (default: 16*1024*1024)
                     - attr_cache_ttl: int (default: 60)
                     - dir_cache_ttl: int (default: 5)
                     - index_ttl_overrides: dict[str, int] (default: {})
                     - enable_metrics: bool (default: False)
        warmup_depth: Directory depth for automatic warmup (default: 2)
        warmup_max_files: Maximum files to warm (default: 1000)
        agent_id: Optional agent ID for namespace-scoped mounts (Issue #1305)
        subject_type: Subject type for OperationContext (None → "agent" when agent_id set)
        owner_id: Human owner ID for the agent
        zone_id: Zone ID for multi-zone isolation
        use_rust: Use high-performance Rust FUSE daemon (Issue #1569)
        passthrough_enabled: Route safe mounts to Rust-owned passthrough.
        passthrough_patterns: Include patterns for Rust passthrough.
        passthrough_deny_patterns: Exclude patterns for Rust passthrough.
        passthrough_threshold_bytes: Minimum passthrough size. Defaults to env.
        passthrough_require: Raise instead of falling back when unsafe.
        passthrough_backing_dir: Optional local backing directory.

    Returns:
        NexusFUSE instance

    Note:
        Issue #1076: Cache warmup runs automatically after mount in background.
        This is always enabled as it's non-blocking and reduces first-access latency.

    Example:
        >>> from nexus import connect
        >>> from nexus.fuse import mount_nexus
        >>>
        >>> nx = connect(config={"data_dir": "./nexus-data"})
        >>>
        >>> # Mount with virtual parsed views (warmup runs automatically)
        >>> fuse = mount_nexus(nx, "/mnt/nexus", mode="smart", foreground=False)
        >>> # cat /mnt/nexus/file.xlsx → binary content
        >>> # cat /mnt/nexus/file_parsed.xlsx.md → parsed markdown
        >>>
        >>> # Namespace-scoped mount for an agent (Issue #1305)
        >>> fuse = mount_nexus(nx, "/mnt/nexus", agent_id="agent_001", foreground=False)
        >>> # Agent only sees paths it has been granted access to
    """
    # Parse mode
    mode_enum = MountMode(mode.lower())

    # Create and mount (warmup runs automatically in background)
    fuse = NexusFUSE(
        nexus_fs,
        mount_point,
        mode=mode_enum,
        cache_config=cache_config,
        warmup_depth=warmup_depth,
        warmup_max_files=warmup_max_files,
        agent_id=agent_id,
        subject_type=subject_type,
        owner_id=owner_id,
        zone_id=zone_id,
        use_rust=use_rust,
        passthrough_enabled=passthrough_enabled,
        passthrough_patterns=passthrough_patterns,
        passthrough_deny_patterns=passthrough_deny_patterns,
        passthrough_threshold_bytes=passthrough_threshold_bytes,
        passthrough_require=passthrough_require,
        passthrough_backing_dir=passthrough_backing_dir,
        lease_manager=lease_manager,
        file_cache=file_cache,
    )
    fuse.mount(foreground=foreground, allow_other=allow_other, debug=debug)

    return fuse
