"""Async-to-Sync Bridge for ReBAC Operations.

This module provides a bridge between synchronous server code
(like ThreadingHTTPServer) and async ReBAC operations.

It manages a background event loop that runs async database operations,
allowing the sync server to benefit from async database pooling and
non-blocking I/O.

Usage:
    from nexus.bricks.rebac.async_bridge import AsyncReBACBridge

    # Initialize once at server startup (prefer injecting engine via DI)
    bridge = AsyncReBACBridge(database_url, engine=record_store.engine)
    bridge.start()

    # Use in sync request handlers
    result = bridge.rebac_check(subject, permission, object, zone_id)

    # Shutdown when done
    bridge.stop()
"""

import asyncio
import logging
import threading
from concurrent.futures import Future
from typing import TYPE_CHECKING, Any

if TYPE_CHECKING:
    from nexus.bricks.rebac.manager import AsyncReBACManager

logger = logging.getLogger(__name__)


class AsyncReBACBridge:
    """Bridge sync code to async ReBAC operations.

    Creates a dedicated event loop in a background thread for running
    async database operations. This allows the sync ThreadingHTTPServer
    to benefit from async database pooling.

    Key benefits:
    - Connection pooling across threads (asyncpg/aiosqlite pools are thread-safe)
    - Non-blocking database I/O
    - 5-10x throughput improvement under concurrent load
    """

    def __init__(
        self,
        database_url: str,
        cache_ttl_seconds: int = 300,
        max_depth: int = 50,
        enable_l1_cache: bool = True,
        l1_cache_size: int = 10000,
        l1_cache_ttl: int = 300,
        *,
        engine: Any | None = None,
    ):
        """Initialize the async bridge.

        Args:
            database_url: Database connection URL (postgresql:// or sqlite://)
            cache_ttl_seconds: L2 cache TTL
            max_depth: Max graph traversal depth
            enable_l1_cache: Enable in-memory L1 cache
            l1_cache_size: L1 cache max entries
            l1_cache_ttl: L1 cache TTL in seconds
            engine: Pre-built SQLAlchemy Engine from RecordStoreABC (Issue #621).
                When provided, skips creating a private engine.
        """
        self.database_url = database_url
        self.cache_ttl_seconds = cache_ttl_seconds
        self.max_depth = max_depth
        self.enable_l1_cache = enable_l1_cache
        self.l1_cache_size = l1_cache_size
        self.l1_cache_ttl = l1_cache_ttl
        self._injected_engine = engine

        self._loop: asyncio.AbstractEventLoop | None = None
        self._thread: threading.Thread | None = None
        self._manager: AsyncReBACManager | None = None
        self._record_store: Any | None = None  # SQLAlchemyRecordStore fallback
        self._started = False
        self._lock = threading.Lock()

    def start(self) -> None:
        """Start the background event loop thread.

        This method is thread-safe and idempotent.
        """
        with self._lock:
            if self._started:
                return

            # Create and start background thread with event loop
            self._loop = asyncio.new_event_loop()
            self._thread = threading.Thread(target=self._run_loop, daemon=True)
            self._thread.start()

            # Initialize async manager in the loop (allow before _started is set)
            future = self._run_coro(self._init_manager(), _allow_startup=True)
            future.result(timeout=30)  # Wait for initialization

            self._started = True
            logger.info("AsyncReBACBridge started")

    def stop(self) -> None:
        """Stop the background event loop thread."""
        with self._lock:
            if not self._started:
                return

            if self._loop:
                self._loop.call_soon_threadsafe(self._loop.stop)

            if self._thread:
                self._thread.join(timeout=5)

            self._loop = None
            self._thread = None
            self._manager = None
            if self._record_store is not None:
                self._record_store.close()
                self._record_store = None
            self._started = False
            logger.info("AsyncReBACBridge stopped")

    def _run_loop(self) -> None:
        """Run the event loop in the background thread."""
        asyncio.set_event_loop(self._loop)
        assert self._loop is not None
        try:
            self._loop.run_forever()
        finally:
            self._loop.close()

    def _run_coro(self, coro: Any, *, _allow_startup: bool = False) -> Future[Any]:
        """Run a coroutine in the background loop and return a Future.

        Args:
            coro: Coroutine to run
            _allow_startup: Internal flag — skip the ``_started`` check during
                ``start()`` so that ``_init_manager()`` can run before the flag
                is set.

        Returns:
            concurrent.futures.Future that will contain the result
        """
        if not self._loop or (not self._started and not _allow_startup):
            raise RuntimeError("AsyncReBACBridge not started")

        return asyncio.run_coroutine_threadsafe(coro, self._loop)

    async def _init_manager(self) -> None:
        """Initialize the async ReBAC manager.

        Issue #1385: AsyncReBACManager is now a thin asyncio.to_thread() wrapper
        around the sync ReBACManager. We create a sync manager and wrap it.

        Issue #621: Engine creation delegates to RecordStoreABC instead of
        calling create_engine() directly.
        """
        from nexus.bricks.rebac.manager import AsyncReBACManager, ReBACManager

        if self._injected_engine is not None:
            sync_engine = self._injected_engine
        else:
            from nexus.storage.record_store import SQLAlchemyRecordStore

            self._record_store = SQLAlchemyRecordStore(db_url=self.database_url)
            sync_engine = self._record_store.engine

        sync_manager = ReBACManager(
            engine=sync_engine,
            cache_ttl_seconds=self.cache_ttl_seconds,
            max_depth=self.max_depth,
        )
        self._manager = AsyncReBACManager(sync_manager)

    def rebac_check(
        self,
        subject: tuple[str, str],
        permission: str,
        object: tuple[str, str],
        zone_id: str | None = None,
        timeout: float = 5.0,
    ) -> bool:
        """Check permission synchronously (bridges to async).

        Args:
            subject: (subject_type, subject_id) tuple
            permission: Permission to check
            object: (object_type, object_id) tuple
            zone_id: Zone ID
            timeout: Max wait time in seconds

        Returns:
            True if permission granted
        """
        if not self._manager:
            raise RuntimeError("AsyncReBACBridge not started")

        future = self._run_coro(
            self._manager.rebac_check(subject, permission, object, zone_id=zone_id)
        )
        result: bool = future.result(timeout=timeout)
        return result

    def rebac_check_bulk(
        self,
        checks: list[tuple[tuple[str, str], str, tuple[str, str]]],
        zone_id: str,
        timeout: float = 30.0,
    ) -> dict[tuple[tuple[str, str], str, tuple[str, str]], bool]:
        """Check multiple permissions synchronously (bridges to async).

        Args:
            checks: List of (subject, permission, object) tuples
            zone_id: Zone ID
            timeout: Max wait time in seconds

        Returns:
            Dict mapping each check to its result
        """
        if not self._manager:
            raise RuntimeError("AsyncReBACBridge not started")

        future = self._run_coro(self._manager.rebac_check_bulk(checks, zone_id))
        result: dict[tuple[tuple[str, str], str, tuple[str, str]], bool] = future.result(
            timeout=timeout
        )
        return result

    def write_tuple(
        self,
        subject: tuple[str, str],
        relation: str,
        object: tuple[str, str],
        zone_id: str | None = None,
        subject_relation: str | None = None,
        timeout: float = 5.0,
    ) -> str:
        """Create a relationship tuple synchronously.

        Args:
            subject: (subject_type, subject_id) tuple
            relation: Relation name
            object: (object_type, object_id) tuple
            zone_id: Zone ID
            subject_relation: For userset subjects
            timeout: Max wait time

        Returns:
            tuple_id of created tuple
        """
        if not self._manager:
            raise RuntimeError("AsyncReBACBridge not started")

        future = self._run_coro(
            self._manager.write_tuple(subject, relation, object, zone_id, subject_relation)
        )
        result: str = future.result(timeout=timeout)
        return result

    def delete_tuple(
        self,
        subject: tuple[str, str],
        relation: str,
        object: tuple[str, str],
        zone_id: str | None = None,
        timeout: float = 5.0,
    ) -> bool:
        """Delete a relationship tuple synchronously.

        Args:
            subject: (subject_type, subject_id) tuple
            relation: Relation name
            object: (object_type, object_id) tuple
            zone_id: Zone ID
            timeout: Max wait time

        Returns:
            True if deleted
        """
        if not self._manager:
            raise RuntimeError("AsyncReBACBridge not started")

        future = self._run_coro(self._manager.delete_tuple(subject, relation, object, zone_id))
        result: bool = future.result(timeout=timeout)
        return result

    def get_cache_stats(self) -> dict[str, Any]:
        """Get L1 cache statistics."""
        if not self._manager:
            return {}
        stats: dict[str, Any] = self._manager.get_l1_cache_stats()
        return stats


# Global bridge instance (optional singleton pattern)
_global_bridge: AsyncReBACBridge | None = None


def get_async_rebac_bridge(
    database_url: str | None = None,
    *,
    engine: Any | None = None,
    **kwargs: Any,
) -> AsyncReBACBridge:
    """Get or create the global async ReBAC bridge.

    Args:
        database_url: Database URL (required on first call unless engine given)
        engine: Pre-built SQLAlchemy Engine from RecordStoreABC (Issue #621).
        **kwargs: Additional arguments for AsyncReBACBridge

    Returns:
        AsyncReBACBridge instance
    """
    global _global_bridge

    if _global_bridge is None:
        if database_url is None:
            raise ValueError("database_url required on first call")
        _global_bridge = AsyncReBACBridge(database_url, engine=engine, **kwargs)
        _global_bridge.start()

    return _global_bridge


def shutdown_async_rebac_bridge() -> None:
    """Shutdown the global async ReBAC bridge."""
    global _global_bridge

    if _global_bridge is not None:
        _global_bridge.stop()
        _global_bridge = None
