"""Read-only execution against remote databases via pyodbc or pymysql.

Used ONLY for data sync operations — never for user queries.
All user queries execute against local PostgreSQL.
"""

from __future__ import annotations

import asyncio
import logging
import re
import time
from typing import Any

logger = logging.getLogger(__name__)

try:
    import pyodbc
except ImportError:
    pyodbc = None  # pyodbc requires ODBC drivers (Windows target)

try:
    import pymysql
except ImportError:
    pymysql = None


def _parse_mysql_url(connection_string: str) -> dict:
    """Parse mysql://user:pass@host:port/database into components."""
    m = re.match(
        r"mysql://(?:(?P<user>[^:@]+)(?::(?P<password>[^@]*))?@)?(?P<host>[^:/]+)(?::(?P<port>\d+))?(?:/(?P<database>\w+))?",
        connection_string,
    )
    if not m:
        raise ValueError(f"Invalid MySQL connection string: {connection_string}")
    return {
        "host": m.group("host") or "127.0.0.1",
        "port": int(m.group("port") or 3306),
        "user": m.group("user") or "root",
        "password": m.group("password") or "",
        "database": m.group("database") or "",
    }


class RemoteExecutor:
    """Execute read-only queries against remote databases for sync purposes."""

    def __init__(self, connection_string: str, db_type: str = "sqlserver") -> None:
        self._connection_string = connection_string
        self._db_type = db_type

    def _get_connection(self):
        """Get a database connection using the appropriate driver."""
        if self._db_type == "mysql":
            return self._get_mysql_connection()
        return self._get_odbc_connection()

    def _ensure_driver(self) -> str:
        """Check connection string has a valid DRIVER= or DSN=. Auto-fix if possible."""
        cs = self._connection_string
        cs_upper = cs.upper()

        # Already has DRIVER= or DSN= → use as-is
        if "DRIVER=" in cs_upper or "DSN=" in cs_upper:
            return cs

        if pyodbc is None:
            return cs

        # Use preset-based driver detection
        from database.db_presets import detect_driver

        available = pyodbc.drivers()
        driver = detect_driver(self._db_type, available)

        if driver:
            cs = f"DRIVER={{{driver}}};{cs}"

        return cs

    def _get_odbc_connection(self):
        if pyodbc is None:
            raise ImportError("pyodbc is required for SQL Server/HANA connections")
        cs = self._ensure_driver()
        logger.info(f"[REMOTE] Connecting via ODBC (db_type={self._db_type}, cs={cs[:60]}...)")

        # Sybase ASE can hang or error with connect timeout param
        connect_kwargs = {}
        if self._db_type != "sybase":
            connect_kwargs["timeout"] = 30

        t0 = time.perf_counter()
        try:
            conn = pyodbc.connect(cs, **connect_kwargs)
        except pyodbc.InterfaceError as e:
            elapsed = time.perf_counter() - t0
            available = pyodbc.drivers() if pyodbc else []
            drivers_list = ", ".join(available) if available else "(none found)"
            logger.error(f"[REMOTE] ODBC connect FAILED after {elapsed:.1f}s: {e}")
            raise pyodbc.InterfaceError(
                f"ODBC connection failed: {e}\n"
                f"Connection string starts with: {cs[:80]}...\n"
                f"Available ODBC drivers: {drivers_list}"
            ) from e

        elapsed = time.perf_counter() - t0
        logger.info(f"[REMOTE] ODBC connected in {elapsed:.2f}s")

        # Sybase ASE drivers handle encoding internally — setdecoding
        # can cause garbled text or connection errors with some drivers
        if self._db_type != "sybase":
            conn.setdecoding(pyodbc.SQL_CHAR, encoding="utf-8")
            conn.setdecoding(pyodbc.SQL_WCHAR, encoding="utf-8")
        else:
            logger.debug("[REMOTE] Skipping setdecoding for Sybase ASE")

        return conn

    def _get_mysql_connection(self):
        if pymysql is None:
            raise ImportError("pymysql is required for MySQL connections (pip install pymysql)")
        params = _parse_mysql_url(self._connection_string)
        conn = pymysql.connect(
            host=params["host"],
            port=params["port"],
            user=params["user"],
            password=params["password"],
            database=params["database"] or None,
            charset="utf8mb4",
            connect_timeout=30,
            cursorclass=pymysql.cursors.DictCursor,
        )
        return conn

    def _run_session_init(self, cursor) -> None:
        """Run session init commands from the preset for this db_type."""
        from database.db_presets import get_preset

        preset = get_preset(self._db_type)
        if preset:
            for cmd in preset.session_init:
                cursor.execute(cmd)

    def _execute_sync(
        self, query: str, params: tuple | None = None
    ) -> tuple[list[dict], list[str]]:
        """Synchronous execution — called via asyncio.to_thread."""
        logger.debug(f"[REMOTE] execute_sync: {query[:120]}...")
        conn = self._get_connection()
        try:
            cursor = conn.cursor()
            self._run_session_init(cursor)

            t0 = time.perf_counter()
            cursor.execute(query, params or ())
            exec_elapsed = time.perf_counter() - t0

            if self._db_type == "mysql":
                rows = cursor.fetchall()
                columns = [desc[0] for desc in cursor.description] if cursor.description else []
                logger.debug(f"[REMOTE] execute_sync returned {len(rows)} rows in {exec_elapsed:.2f}s")
                return list(rows), columns
            else:
                columns = [desc[0] for desc in cursor.description] if cursor.description else []
                rows = [dict(zip(columns, row)) for row in cursor.fetchall()]
                logger.debug(f"[REMOTE] execute_sync returned {len(rows)} rows in {exec_elapsed:.2f}s")
                return rows, columns
        finally:
            conn.close()

    async def execute(
        self, query: str, params: tuple | None = None
    ) -> tuple[list[dict], list[str]]:
        """Execute a read-only query asynchronously (runs in thread pool)."""
        return await asyncio.to_thread(self._execute_sync, query, params)

    def _fetch_batched_sync(
        self, query: str, batch_size: int = 10000
    ):
        """Generator that yields batches of rows for large table sync."""
        logger.info(f"[REMOTE] fetch_batched_sync: batch_size={batch_size}, query={query[:120]}...")
        conn = self._get_connection()
        try:
            cursor = conn.cursor()
            self._run_session_init(cursor)

            # Sybase needs QUOTED_IDENTIFIER for tables with special chars
            if self._db_type == "sybase":
                cursor.execute("SET QUOTED_IDENTIFIER ON")

            t0 = time.perf_counter()
            cursor.execute(query)
            exec_elapsed = time.perf_counter() - t0
            columns = [desc[0] for desc in cursor.description] if cursor.description else []
            logger.info(f"[REMOTE] Query executed in {exec_elapsed:.2f}s, {len(columns)} columns: {columns[:10]}{'...' if len(columns) > 10 else ''}")

            batch_num = 0
            total_rows = 0
            fetch_start = time.perf_counter()

            if self._db_type == "mysql":
                while True:
                    t_fetch = time.perf_counter()
                    rows = cursor.fetchmany(batch_size)
                    if not rows:
                        break
                    batch_num += 1
                    total_rows += len(rows)
                    fetch_elapsed = time.perf_counter() - t_fetch
                    overall_elapsed = time.perf_counter() - fetch_start
                    rate = total_rows / overall_elapsed if overall_elapsed > 0 else 0
                    logger.info(
                        f"[REMOTE] Batch {batch_num}: fetched {len(rows)} rows in {fetch_elapsed:.2f}s "
                        f"(total: {total_rows:,} rows, {overall_elapsed:.1f}s elapsed, {rate:,.0f} rows/s)"
                    )
                    yield list(rows), columns
            else:
                while True:
                    t_fetch = time.perf_counter()
                    rows = cursor.fetchmany(batch_size)
                    if not rows:
                        break
                    batch_num += 1
                    total_rows += len(rows)
                    fetch_elapsed = time.perf_counter() - t_fetch
                    overall_elapsed = time.perf_counter() - fetch_start
                    rate = total_rows / overall_elapsed if overall_elapsed > 0 else 0
                    logger.info(
                        f"[REMOTE] Batch {batch_num}: fetched {len(rows)} rows in {fetch_elapsed:.2f}s "
                        f"(total: {total_rows:,} rows, {overall_elapsed:.1f}s elapsed, {rate:,.0f} rows/s)"
                    )
                    yield [dict(zip(columns, row)) for row in rows], columns

            total_elapsed = time.perf_counter() - fetch_start
            final_rate = total_rows / total_elapsed if total_elapsed > 0 else 0
            logger.info(
                f"[REMOTE] Fetch complete: {total_rows:,} rows in {batch_num} batches, "
                f"{total_elapsed:.1f}s total, {final_rate:,.0f} rows/s avg"
            )
        finally:
            conn.close()

    async def fetch_batched(self, query: str, batch_size: int = 10000):
        """Async generator that streams batches as they arrive from the remote DB.

        Uses an asyncio.Queue so each batch is yielded for insertion immediately
        after it's fetched — not buffered in memory until the entire table is read.
        This is critical for million-row tables where fetching alone takes hours.

        Yields (batch_rows, columns) tuples.
        """
        logger.info(
            f"[REMOTE] fetch_batched: starting streaming pipeline (batch_size={batch_size}, "
            f"queue_depth=2, query={query[:80]}...)"
        )
        queue: asyncio.Queue = asyncio.Queue(maxsize=2)  # backpressure: max 2 batches buffered
        _DONE = object()
        t_stream_start = time.perf_counter()

        # Capture the running event loop BEFORE entering the thread
        loop = asyncio.get_running_loop()

        def _producer():
            """Runs in thread pool — feeds batches into the queue."""
            try:
                logger.info(f"[REMOTE] Producer thread entered (loop={loop!r})")
                for batch in self._fetch_batched_sync(query, batch_size):
                    # Block until queue has space (backpressure)
                    asyncio.run_coroutine_threadsafe(queue.put(batch), loop).result()
            except Exception as e:
                logger.error(f"[REMOTE] Producer thread error: {type(e).__name__}: {e}", exc_info=True)
                try:
                    asyncio.run_coroutine_threadsafe(queue.put(e), loop).result(timeout=30)
                except Exception as put_err:
                    logger.critical(f"[REMOTE] Producer could not push error to queue: {put_err}")
            finally:
                try:
                    asyncio.run_coroutine_threadsafe(queue.put(_DONE), loop).result(timeout=30)
                except Exception as done_err:
                    logger.critical(f"[REMOTE] Producer could not push DONE sentinel: {done_err}")

        # Start producer in background thread
        future = loop.run_in_executor(None, _producer)
        logger.info("[REMOTE] Producer thread started, streaming batches to consumer...")

        # Consume batches as they arrive
        batches_yielded = 0
        while True:
            t_wait = time.perf_counter()
            item = await queue.get()
            wait_elapsed = time.perf_counter() - t_wait
            if item is _DONE:
                break
            if isinstance(item, Exception):
                logger.error(f"[REMOTE] Producer raised exception, re-raising: {item}")
                raise item
            batches_yielded += 1
            logger.info(
                f"[REMOTE] Streaming batch {batches_yielded} to consumer "
                f"(waited {wait_elapsed:.2f}s for producer)"
            )
            yield item

        # Ensure thread is done
        await asyncio.wrap_future(future)
        stream_elapsed = time.perf_counter() - t_stream_start
        logger.info(
            f"[REMOTE] Streaming pipeline finished: {batches_yielded} batches "
            f"in {stream_elapsed:.1f}s total"
        )

    async def test_connection(self) -> dict:
        """Test if the remote connection works.

        Returns: {"ok": bool, "error": str | None}
        """
        try:
            await self.execute("SELECT 1 AS test")
            return {"ok": True, "error": None}
        except Exception as e:
            return {"ok": False, "error": str(e)}
