"""
Accès MySQL partagé par tous les modules (même pile que hec-photo-app : SQLAlchemy 2.0 + PyMySQL,
utf8mb4 / InnoDB, variables MYSQL_*).

Pourquoi un moteur SYNCHRONE derrière `get_db()` : le nettoyage RGPD, l'organizer, l'élection du leader et
le rate limiting tournent dans des threads (`asyncio.to_thread`) et appelaient déjà `get_db()`. Le pool de
SQLAlchemy fournit les connexions réutilisables ; l'interface (`conn.execute(sql, params)`, lignes accessibles
par nom, `with get_db() as conn`) ne change pas pour les ~220 requêtes existantes.

Dialecte : le code garde ses placeholders `?` ; ils sont traduits ici (avec `%` échappé). `INSERT OR REPLACE`
→ `REPLACE`, `INSERT OR IGNORE` → `INSERT IGNORE`, `BEGIN IMMEDIATE` → `START TRANSACTION`, la colonne `key`
(mot réservé MySQL) est citée automatiquement et `COLLATE NOCASE` est ignoré (la collation par défaut
utf8mb4_unicode_ci est déjà insensible à la casse, comme LIKE sous SQLite).
"""
import logging
import re
import threading
from contextlib import contextmanager
from typing import Any, Iterator, Optional, Sequence

from sqlalchemy import create_engine
from sqlalchemy.engine import URL, Engine

from core import config

log = logging.getLogger("meeconic.db")

_engine: Optional[Engine] = None
_engine_lock = threading.Lock()

TABLE_OPTIONS = "ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"


def _build_engine() -> Engine:
    url = URL.create(
        "mysql+pymysql", username=config.MYSQL_USER, password=config.MYSQL_PASSWORD or None,
        host=config.MYSQL_HOST, port=config.MYSQL_PORT, database=config.MYSQL_DB,
        query={"charset": "utf8mb4"})
    return create_engine(
        url, pool_size=config.MYSQL_POOL_SIZE, max_overflow=config.MYSQL_POOL_SIZE, pool_recycle=1800,
        pool_pre_ping=True,   # a MySQL restart / idle timeout must not surface as a 500 on the next request
        isolation_level="READ COMMITTED")


def engine() -> Engine:
    global _engine
    if _engine is None:
        with _engine_lock:
            if _engine is None:
                _engine = _build_engine()
    return _engine


def dispose_engine() -> None:
    """Closes the pool (tests switch database; the service does it on shutdown)."""
    global _engine
    with _engine_lock:
        if _engine is not None:
            _engine.dispose()
            _engine = None


# ── Traduction du dialecte ─────────────────────────────────────────────────────

_STRING_RE = re.compile(r"('(?:[^']|'')*')")
_KEY_RE = re.compile(r"(?<![\w`.])key(?![\w`])", re.IGNORECASE)
_OR_REPLACE_RE = re.compile(r"\bINSERT\s+OR\s+REPLACE\s+INTO\b", re.IGNORECASE)
_OR_IGNORE_RE = re.compile(r"\bINSERT\s+OR\s+IGNORE\s+INTO\b", re.IGNORECASE)
_NOCASE_RE = re.compile(r"\s+COLLATE\s+NOCASE\b", re.IGNORECASE)
_BEGIN_RE = re.compile(r"^\s*BEGIN(\s+IMMEDIATE|\s+TRANSACTION)?\s*$", re.IGNORECASE)
_cache: dict[str, str] = {}


def translate(sql: str) -> str:
    """SQL écrit pour sqlite3 (placeholders `?`) → SQL MySQL pour PyMySQL (`%s`)."""
    hit = _cache.get(sql)
    if hit is not None:
        return hit
    if _BEGIN_RE.match(sql):
        out = "START TRANSACTION"
    else:
        # Le DDL est déjà écrit en MySQL (PRIMARY KEY, KEY idx…) : on n'y touche pas à `key`.
        is_ddl = sql.lstrip()[:6].upper() in ("CREATE", "ALTER ", "DROP T")
        # Les littéraux entre quotes restent tels quels (sauf % → %% pour PyMySQL).
        parts = _STRING_RE.split(sql)
        for i, part in enumerate(parts):
            if i % 2:                      # littéral
                parts[i] = part.replace("%", "%%")
                continue
            part = _OR_REPLACE_RE.sub("REPLACE INTO", part)
            part = _OR_IGNORE_RE.sub("INSERT IGNORE INTO", part)
            part = _NOCASE_RE.sub("", part)
            if not is_ddl:
                part = _KEY_RE.sub("`key`", part)
            part = part.replace("%", "%%").replace("?", "%s")
            parts[i] = part
        out = "".join(parts)
    if len(_cache) < 2000:
        _cache[sql] = out
    return out


# ── Lignes et curseur ──────────────────────────────────────────────────────────

class Row(dict):
    """Ligne accessible par nom ET par position, comme sqlite3.Row."""
    __slots__ = ()

    def __getitem__(self, k):
        if isinstance(k, int):
            return list(self.values())[k]
        return super().__getitem__(k)


class Cursor:
    def __init__(self, cur):
        self._cur = cur

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

    @property
    def lastrowid(self) -> Optional[int]:
        return self._cur.lastrowid

    def fetchone(self) -> Optional[Row]:
        r = self._cur.fetchone()
        return Row(r) if r is not None else None

    def fetchall(self) -> list[Row]:
        return [Row(r) for r in self._cur.fetchall()]

    def __iter__(self) -> Iterator[Row]:
        for r in self._cur.fetchall():
            yield Row(r)


class Connection:
    """Connexion du pool avec l'interface de sqlite3.Connection que le code utilisait."""

    def __init__(self, raw):
        import pymysql.cursors
        self._raw = raw
        self._dict_cursor = pymysql.cursors.DictCursor
        self.isolation_level = ""   # compat : `conn.isolation_level = None` servait aux transactions manuelles

    def execute(self, sql: str, params: Sequence[Any] = ()) -> Cursor:
        cur = self._raw.cursor(self._dict_cursor)
        cur.execute(translate(sql), tuple(params))
        return Cursor(cur)

    def executemany(self, sql: str, seq: Sequence[Sequence[Any]]) -> None:
        cur = self._raw.cursor(self._dict_cursor)
        cur.executemany(translate(sql), [tuple(p) for p in seq])

    def executescript(self, script: str) -> None:
        for stmt in (s.strip() for s in script.split(";")):
            if stmt:
                self.execute(stmt)

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

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

    def close(self) -> None:
        try:
            self._raw.close()      # returns it to the pool (SQLAlchemy's pooled proxy)
        except Exception:
            pass

    def __enter__(self) -> "Connection":
        return self

    def __exit__(self, exc_type, exc, tb) -> bool:
        try:
            if exc_type is None:
                self._raw.commit()
            else:
                self._raw.rollback()
        finally:
            self.close()
        return False


def get_db() -> Connection:
    return Connection(engine().raw_connection())


@contextmanager
def named_lock(name: str, timeout: int = 5):
    """Verrou applicatif inter-processus (GET_LOCK). Remplace `BEGIN IMMEDIATE` de SQLite pour les
    lecture-puis-écriture qui doivent être atomiques (rate limiting, cooldown d'alertes)."""
    lock = f"meeconic:{name}"
    if len(lock) > 64:     # GET_LOCK names are limited to 64 chars
        import hashlib
        lock = "meeconic:" + hashlib.sha1(name.encode()).hexdigest()
    conn = get_db()
    try:
        got = conn.execute("SELECT GET_LOCK(?, ?) AS ok", (lock, timeout)).fetchone()["ok"]
        if got != 1:
            raise TimeoutError(f"lock {name!r} not acquired in {timeout}s")
        try:
            yield
        finally:
            conn.execute("SELECT RELEASE_LOCK(?)", (lock,))
    finally:
        conn.close()


# ── Introspection (remplace PRAGMA table_info / sqlite_master) ─────────────────────

def table_columns(conn: Connection, table: str) -> list[str]:
    rows = conn.execute(
        "SELECT COLUMN_NAME AS name FROM information_schema.COLUMNS "
        "WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=?", (table,)).fetchall()
    return [r["name"] for r in rows]


def table_exists(conn: Connection, table: str) -> bool:
    return bool(conn.execute(
        "SELECT 1 FROM information_schema.TABLES WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=?",
        (table,)).fetchone())


def ensure_column(conn: Connection, table: str, column: str, ddl: str) -> None:
    if column not in table_columns(conn, table):
        conn.execute(f"ALTER TABLE {table} ADD COLUMN {column} {ddl}")


def ensure_index(conn: Connection, table: str, name: str, columns: str) -> None:
    """CREATE INDEX IF NOT EXISTS n'existe pas en MySQL."""
    exists = conn.execute(
        "SELECT 1 FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=? AND INDEX_NAME=?",
        (table, name)).fetchone()
    if not exists:
        conn.execute(f"CREATE INDEX {name} ON {table} ({columns})")


def create_database_if_missing() -> None:
    """Crée la base (utf8mb4) si l'utilisateur MySQL en a le droit — pratique en dev (XAMPP)."""
    import pymysql
    try:
        conn = pymysql.connect(host=config.MYSQL_HOST, port=config.MYSQL_PORT, user=config.MYSQL_USER,
                               password=config.MYSQL_PASSWORD, charset="utf8mb4")
        try:
            with conn.cursor() as cur:
                cur.execute(f"CREATE DATABASE IF NOT EXISTS `{config.MYSQL_DB}` "
                            f"DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci")
            conn.commit()
        finally:
            conn.close()
    except Exception as e:
        # A production user normally owns just its own database and may not be allowed to create one: the
        # database then already exists and the real connection below tells whether it is reachable.
        log.warning("Could not ensure database %s exists (%s) — assuming it was created beforehand", config.MYSQL_DB, e)


def init_core_tables() -> None:
    create_database_if_missing()
    with get_db() as conn:
        conn.executescript(f"""
        CREATE TABLE IF NOT EXISTS app_kv (
            `key`   VARCHAR(191) NOT NULL PRIMARY KEY,
            value   MEDIUMTEXT
        ) {TABLE_OPTIONS};
        CREATE TABLE IF NOT EXISTS rate_hits (
            `key` VARCHAR(191) NOT NULL,
            ts    DOUBLE NOT NULL,
            KEY idx_rate_hits_key (`key`, ts)
        ) {TABLE_OPTIONS};
        CREATE TABLE IF NOT EXISTS leader_locks (
            name       VARCHAR(64) NOT NULL PRIMARY KEY,
            owner      VARCHAR(64) NOT NULL,
            expires_at DOUBLE NOT NULL
        ) {TABLE_OPTIONS}
        """)
        conn.commit()