from __future__ import annotations

import os
import queue
import hashlib
import sqlite3
import ssl
import threading
import time
from collections.abc import Iterable, Iterator, Mapping, Sequence
from decimal import Decimal
from pathlib import Path
from typing import Any
from urllib.parse import parse_qs, unquote, urlparse


def finding_run_idempotency_key(
    org_id: str, cluster_id: str, source: str, source_run_id: str
) -> bytes:
    payload = bytearray()
    for value in (org_id, cluster_id, source, source_run_id):
        encoded = value.encode("utf-8")
        payload.extend(str(len(encoded)).encode("ascii"))
        payload.extend(b":")
        payload.extend(encoded)
    return hashlib.sha256(payload).digest()


class DatabaseRow(dict[str, Any]):
    """Mapping row with sqlite.Row-compatible numeric access."""

    def __init__(self, columns: Sequence[str], values: Sequence[Any]) -> None:
        super().__init__(zip(columns, values, strict=True))
        self._values = tuple(values)

    def __getitem__(self, key: str | int) -> Any:
        if isinstance(key, int):
            return self._values[key]
        return super().__getitem__(key)


def _normalize_mysql_value(value: Any) -> Any:
    """Match SQLite numeric result types for TiDB aggregate expressions."""

    if not isinstance(value, Decimal):
        return value
    if value == value.to_integral_value():
        return int(value)
    return float(value)


class MySQLCursor:
    def __init__(self, cursor: Any) -> None:
        self._cursor = cursor
        self._columns = tuple(column[0] for column in (cursor.description or ()))
        self._rowcount = int(cursor.rowcount)
        self._closed = False
        if cursor.description is None:
            self.close()

    @property
    def rowcount(self) -> int:
        return self._rowcount

    def close(self) -> None:
        if self._closed:
            return
        self._cursor.close()
        self._closed = True

    def _convert(self, row: Sequence[Any] | None) -> DatabaseRow | None:
        if row is None:
            return None
        return DatabaseRow(
            self._columns,
            tuple(_normalize_mysql_value(value) for value in row),
        )

    def fetchone(self) -> DatabaseRow | None:
        try:
            return self._convert(self._cursor.fetchone())
        finally:
            self.close()

    def fetchall(self) -> list[DatabaseRow]:
        try:
            return [self._convert(row) for row in self._cursor.fetchall()]  # type: ignore[misc]
        finally:
            self.close()

    def __iter__(self) -> Iterator[DatabaseRow]:
        try:
            while raw_row := self._cursor.fetchone():
                row = self._convert(raw_row)
                if row is not None:
                    yield row
        finally:
            self.close()


def _qmark_to_mysql(sql: str) -> str:
    """Convert DB-API qmark placeholders without touching quoted SQL text."""

    output: list[str] = []
    index = 0
    quote: str | None = None
    line_comment = False
    block_comment = False
    while index < len(sql):
        char = sql[index]
        following = sql[index + 1] if index + 1 < len(sql) else ""
        if line_comment:
            output.append(char)
            if char == "\n":
                line_comment = False
            index += 1
            continue
        if block_comment:
            output.append(char)
            if char == "*" and following == "/":
                output.append(following)
                index += 2
                block_comment = False
            else:
                index += 1
            continue
        if quote:
            output.append(char)
            if char == "\\" and following:
                output.append(following)
                index += 2
                continue
            if char == quote:
                if following == quote and quote in {"'", '"'}:
                    output.append(following)
                    index += 2
                    continue
                quote = None
            index += 1
            continue
        if char == "-" and following == "-":
            output.extend((char, following))
            index += 2
            line_comment = True
            continue
        if char == "#":
            output.append(char)
            index += 1
            line_comment = True
            continue
        if char == "/" and following == "*":
            output.extend((char, following))
            index += 2
            block_comment = True
            continue
        if char in {"'", '"', "`"}:
            quote = char
            output.append(char)
        elif char == "?":
            output.append("%s")
        else:
            output.append(char)
        index += 1
    return "".join(output)


def split_sql_script(script: str) -> list[str]:
    statements: list[str] = []
    current: list[str] = []
    quote: str | None = None
    index = 0
    while index < len(script):
        char = script[index]
        following = script[index + 1] if index + 1 < len(script) else ""
        if quote:
            current.append(char)
            if char == "\\" and following:
                current.append(following)
                index += 2
                continue
            if char == quote:
                if following == quote and quote in {"'", '"'}:
                    current.append(following)
                    index += 2
                    continue
                quote = None
        elif char in {"'", '"', "`"}:
            quote = char
            current.append(char)
        elif char == ";":
            statement = "".join(current).strip()
            if statement:
                statements.append(statement)
            current = []
        else:
            current.append(char)
        index += 1
    statement = "".join(current).strip()
    if statement:
        statements.append(statement)
    return statements


class MySQLConnection:
    dialect = "mysql"

    def __init__(self, connection: Any) -> None:
        self._connection = connection

    def execute(
        self, sql: str, parameters: Sequence[Any] | Mapping[str, Any] = ()
    ) -> MySQLCursor:
        cursor = self._connection.cursor()
        try:
            if parameters:
                cursor.execute(_qmark_to_mysql(sql), parameters)
            else:
                cursor.execute(sql)
        except Exception:
            cursor.close()
            raise
        return MySQLCursor(cursor)

    def executemany(
        self, sql: str, parameters: Iterable[Sequence[Any]]
    ) -> MySQLCursor:
        cursor = self._connection.cursor()
        try:
            cursor.executemany(_qmark_to_mysql(sql), parameters)
        except Exception:
            cursor.close()
            raise
        return MySQLCursor(cursor)

    def executescript(self, script: str) -> None:
        for statement in split_sql_script(script):
            self.execute(statement)

    def commit(self) -> None:
        self._connection.commit()

    def rollback(self) -> None:
        self._connection.rollback()

    def close(self) -> None:
        self._connection.close()

    def __enter__(self) -> MySQLConnection:
        return self

    def __exit__(self, exc_type: Any, exc: Any, traceback: Any) -> None:
        try:
            if exc_type is None:
                self.commit()
            else:
                self.rollback()
        finally:
            self.close()


class MySQLConnectionPool:
    def __init__(self, create_connection: Any, max_size: int) -> None:
        self._create_connection = create_connection
        self._max_size = max_size
        self._available: queue.LifoQueue[Any] = queue.LifoQueue(maxsize=max_size)
        self._created = 0
        self._lock = threading.Lock()

    def acquire(self) -> Any:
        try:
            connection, released_at = self._available.get_nowait()
        except queue.Empty:
            connection = None
            released_at = 0.0

        if connection is None:
            with self._lock:
                if self._created < self._max_size:
                    self._created += 1
                    create_new = True
                else:
                    create_new = False
            if create_new:
                try:
                    return self._create_connection()
                except Exception:
                    with self._lock:
                        self._created -= 1
                    raise
            connection, released_at = self._available.get(
                timeout=float(os.environ.get("SQL_CATALOG_DB_POOL_TIMEOUT", "10"))
            )

        pre_ping_interval = max(
            0.0,
            float(os.environ.get("SQL_CATALOG_DB_PRE_PING_INTERVAL", "30")),
        )
        if time.monotonic() - released_at >= pre_ping_interval:
            try:
                connection.ping(reconnect=False)
            except Exception:
                try:
                    connection.close()
                finally:
                    with self._lock:
                        self._created -= 1
                return self.acquire()
        return connection

    def warm(self, target_size: int | None = None) -> None:
        target = min(target_size or self._max_size, self._max_size)
        acquired: list[Any] = []
        try:
            for _ in range(target):
                acquired.append(self.acquire())
        finally:
            for connection in acquired:
                self.release(connection, transaction_resolved=True)

    def release(self, connection: Any, *, transaction_resolved: bool) -> None:
        try:
            if not transaction_resolved:
                connection.rollback()
            self._available.put_nowait((connection, time.monotonic()))
        except Exception:
            try:
                connection.close()
            finally:
                with self._lock:
                    self._created -= 1


class PooledMySQLConnection(MySQLConnection):
    def __init__(self, connection: Any, pool: MySQLConnectionPool) -> None:
        super().__init__(connection)
        self._pool = pool
        self._released = False
        self._transaction_resolved = False

    def execute(
        self, sql: str, parameters: Sequence[Any] | Mapping[str, Any] = ()
    ) -> MySQLCursor:
        self._transaction_resolved = False
        return super().execute(sql, parameters)

    def executemany(
        self, sql: str, parameters: Iterable[Sequence[Any]]
    ) -> MySQLCursor:
        self._transaction_resolved = False
        return super().executemany(sql, parameters)

    def commit(self) -> None:
        super().commit()
        self._transaction_resolved = True

    def rollback(self) -> None:
        super().rollback()
        self._transaction_resolved = True

    def close(self) -> None:
        if self._released:
            return
        self._released = True
        self._pool.release(
            self._connection,
            transaction_resolved=self._transaction_resolved,
        )


def connection_dialect(connection: Any) -> str:
    return str(getattr(connection, "dialect", "sqlite"))


def connect_sqlite(path: Path) -> sqlite3.Connection:
    path.parent.mkdir(parents=True, exist_ok=True)
    connection = sqlite3.connect(path)
    connection.row_factory = sqlite3.Row
    connection.execute("PRAGMA foreign_keys = ON")
    return connection


_MYSQL_POOLS: dict[tuple[str, int, str, str, bool], MySQLConnectionPool] = {}
_MYSQL_POOLS_LOCK = threading.Lock()


def warm_mysql_pool(
    database_url: str,
    password_env: str = "DB_PASSWD",
    *,
    target_size: int | None = None,
) -> None:
    connection = connect_mysql(database_url, password_env)
    pool = connection._pool
    connection.close()
    pool.warm(target_size)


def connect_mysql(database_url: str, password_env: str = "DB_PASSWD") -> MySQLConnection:
    try:
        import pymysql
    except ImportError as exc:  # pragma: no cover - exercised by deployment setup.
        raise RuntimeError(
            "PyMySQL is required for TiDB; install backend/requirements.txt"
        ) from exc

    parsed = urlparse(database_url)
    if parsed.scheme not in {"mysql", "tidb"}:
        raise ValueError("SQL_CATALOG_DATABASE_URL must use mysql:// or tidb://")
    database = unquote(parsed.path.lstrip("/"))
    if not parsed.hostname or not parsed.username or not database:
        raise ValueError(
            "SQL_CATALOG_DATABASE_URL must include user, host, port, and database"
        )
    password = unquote(parsed.password) if parsed.password else os.environ.get(password_env)
    if password is None:
        raise ValueError(f"Database password is missing; set {password_env}")
    query = parse_qs(parsed.query)
    tls_enabled = query.get("ssl", ["true"])[0].lower() not in {"0", "false", "off"}
    tls_context = ssl.create_default_context() if tls_enabled else None
    host = parsed.hostname
    port = parsed.port or 4000
    user = unquote(parsed.username)
    pool_key = (host, port, user, database, tls_enabled)

    def create_connection() -> Any:
        attempts = max(
            1,
            min(int(os.environ.get("SQL_CATALOG_DB_CONNECT_ATTEMPTS", "3")), 10),
        )
        retry_delay = max(
            0.0,
            float(os.environ.get("SQL_CATALOG_DB_CONNECT_RETRY_DELAY", "2")),
        )
        for attempt in range(attempts):
            try:
                return pymysql.connect(
                    host=host,
                    port=port,
                    user=user,
                    password=password,
                    database=database,
                    charset="utf8mb4",
                    autocommit=False,
                    connect_timeout=int(
                        os.environ.get("SQL_CATALOG_DB_CONNECT_TIMEOUT", "10")
                    ),
                    read_timeout=int(
                        os.environ.get("SQL_CATALOG_DB_READ_TIMEOUT", "30")
                    ),
                    write_timeout=int(
                        os.environ.get("SQL_CATALOG_DB_WRITE_TIMEOUT", "30")
                    ),
                    ssl=tls_context,
                )
            except pymysql.err.OperationalError as exc:
                error_code = exc.args[0] if exc.args else None
                retryable = error_code in {2003, 2013, 2055}
                if not retryable or attempt + 1 >= attempts:
                    raise
                time.sleep(min(retry_delay * (attempt + 1), 5.0))
        raise RuntimeError("unreachable MySQL connection retry state")

    with _MYSQL_POOLS_LOCK:
        pool = _MYSQL_POOLS.get(pool_key)
        if pool is None:
            pool_size = max(
                1, min(int(os.environ.get("SQL_CATALOG_DB_POOL_SIZE", "8")), 32)
            )
            pool = MySQLConnectionPool(create_connection, pool_size)
            _MYSQL_POOLS[pool_key] = pool
    return PooledMySQLConnection(pool.acquire(), pool)


def database_description(database_url: str | None, sqlite_path: Path) -> str:
    if not database_url:
        return f"sqlite:{sqlite_path}"
    parsed = urlparse(database_url)
    return f"{parsed.scheme}://{parsed.hostname}:{parsed.port or 4000}/{parsed.path.lstrip('/')}"