from __future__ import annotations

import fnmatch
import ssl
from urllib.parse import parse_qs, unquote, urlparse

import pymysql

from ingestion.common.models import Column, Table

from .config import SourceConfig


def connect(config: SourceConfig):
    parsed = urlparse(config.source_dsn)
    query = parse_qs(parsed.query)
    kwargs = {
        "host": parsed.hostname,
        "port": parsed.port or 4000,
        "user": unquote(parsed.username or ""),
        "password": unquote(parsed.password or ""),
        "database": parsed.path.lstrip("/") or config.source_database,
        "cursorclass": pymysql.cursors.DictCursor,
    }
    if query.get("ssl", ["true"])[0].lower() not in {"0", "false", "no"}:
        # Verify the server certificate and hostname. The previous empty ssl dict
        # encrypted the connection but did not authenticate the server (PyMySQL leaves a
        # CA-less context at CERT_NONE / check_hostname=False), leaving it open to MITM.
        # create_default_context() loads the system CA bundle, sets CERT_REQUIRED and
        # check_hostname=True — correct for TiDB Cloud's publicly-trusted cert. A custom
        # CA bundle can be supplied via ?ssl_ca=<path> in the DSN.
        context = ssl.create_default_context()
        ssl_ca = query.get("ssl_ca", [None])[0]
        if ssl_ca:
            context.load_verify_locations(unquote(ssl_ca))
        kwargs["ssl"] = context
    return pymysql.connect(**kwargs)


def kept_base_table_names(cursor, config: SourceConfig, selected: list[str] | None = None) -> list[str]:
    """Single source of truth for which base tables this source processes.

    Applies the include/exclude patterns and the ``require_primary_key`` rule against
    ``information_schema`` using the given open cursor. Both schema inspection (which
    drives contracts + the manifest) and the export filter derive their table set from
    this one function, so the exported set and the contracted set cannot silently
    diverge.
    """
    selected = selected if selected is not None else config.selected_tables
    cursor.execute(
        "SELECT TABLE_NAME FROM information_schema.TABLES "
        "WHERE TABLE_SCHEMA = %s AND TABLE_TYPE = 'BASE TABLE'",
        (config.source_database,),
    )
    names = [
        row["TABLE_NAME"]
        for row in cursor.fetchall()
        if include_table(row["TABLE_NAME"], selected, config.excluded_tables)
    ]
    if names and config.require_primary_key:
        placeholders = ",".join(["%s"] * len(names))
        cursor.execute(
            f"""
SELECT DISTINCT TABLE_NAME FROM information_schema.TABLE_CONSTRAINTS
WHERE TABLE_SCHEMA = %s AND CONSTRAINT_TYPE = 'PRIMARY KEY'
  AND TABLE_NAME IN ({placeholders})
""",
            (config.source_database, *names),
        )
        with_pk = {row["TABLE_NAME"] for row in cursor.fetchall()}
        names = [name for name in names if name in with_pk]
    return sorted(names)


def inspect_tables(config: SourceConfig, selected: list[str] | None = None) -> list[Table]:
    with connect(config) as conn:
        with conn.cursor() as cursor:
            names = kept_base_table_names(cursor, config, selected=selected)
            if not names:
                return []

            placeholders = ",".join(["%s"] * len(names))
            cursor.execute(
                f"""
SELECT TABLE_NAME, TABLE_ROWS
FROM information_schema.TABLES
WHERE TABLE_SCHEMA = %s AND TABLE_NAME IN ({placeholders})
""",
                (config.source_database, *names),
            )
            table_rows = {row["TABLE_NAME"]: row["TABLE_ROWS"] for row in cursor.fetchall()}

            cursor.execute(
                f"""
SELECT TABLE_NAME, COLUMN_NAME, DATA_TYPE, COLUMN_TYPE, IS_NULLABLE, ORDINAL_POSITION
FROM information_schema.COLUMNS
WHERE TABLE_SCHEMA = %s AND TABLE_NAME IN ({placeholders})
ORDER BY TABLE_NAME, ORDINAL_POSITION
""",
                (config.source_database, *names),
            )
            columns_by_table: dict[str, list[Column]] = {name: [] for name in names}
            for row in cursor.fetchall():
                columns_by_table[row["TABLE_NAME"]].append(
                    Column(
                        name=row["COLUMN_NAME"],
                        data_type=row["DATA_TYPE"],
                        column_type=row["COLUMN_TYPE"],
                        nullable=row["IS_NULLABLE"] == "YES",
                        ordinal_position=row["ORDINAL_POSITION"],
                    )
                )

            cursor.execute(
                f"""
SELECT k.TABLE_NAME, k.COLUMN_NAME
FROM information_schema.TABLE_CONSTRAINTS t
JOIN information_schema.KEY_COLUMN_USAGE k
  ON t.CONSTRAINT_SCHEMA = k.CONSTRAINT_SCHEMA
 AND t.TABLE_SCHEMA = k.TABLE_SCHEMA
 AND t.TABLE_NAME = k.TABLE_NAME
 AND t.CONSTRAINT_NAME = k.CONSTRAINT_NAME
WHERE t.TABLE_SCHEMA = %s
  AND t.TABLE_NAME IN ({placeholders})
  AND t.CONSTRAINT_TYPE = 'PRIMARY KEY'
ORDER BY k.TABLE_NAME, k.ORDINAL_POSITION
""",
                (config.source_database, *names),
            )
            primary_keys: dict[str, list[str]] = {name: [] for name in names}
            for row in cursor.fetchall():
                primary_keys[row["TABLE_NAME"]].append(row["COLUMN_NAME"])

    return [
        Table(
            database=config.source_database,
            name=name,
            rows_estimate=table_rows.get(name),
            columns=columns_by_table[name],
            primary_key=primary_keys[name],
            watermark_candidates=config.watermark_candidates,
        )
        for name in names
    ]


def selected_table_names(config: SourceConfig) -> list[str]:
    """Names of base tables to process. Used by the export filter so it lists exactly the
    tables that get contracts — both derive from ``kept_base_table_names``."""
    with connect(config) as conn:
        with conn.cursor() as cursor:
            return kept_base_table_names(cursor, config)


def export_filter(config: SourceConfig) -> str:
    """ticloud export `--filter`. When exclusions/require_primary_key are inactive and
    everything is included, use the cheap `db.*` wildcard; otherwise enumerate the kept
    tables (ticloud filters are include-only, so exclusion must be expressed explicitly)."""
    if not config.excluded_tables and not config.require_primary_key:
        tables = config.selected_tables
        if not tables or tables == ["*"]:
            return f"{config.source_database}.*"
        return ",".join(
            table if "." in table else f"{config.source_database}.{table}" for table in tables
        )
    return ",".join(f"{config.source_database}.{name}" for name in selected_table_names(config))


def include_table(table_name: str, selected: list[str], excluded: list[str] | None = None) -> bool:
    for pattern in excluded or []:
        table_pattern = pattern.split(".", 1)[1] if "." in pattern else pattern
        if fnmatch.fnmatchcase(table_name, table_pattern):
            return False
    if not selected:
        return True
    for pattern in selected:
        table_pattern = pattern.split(".", 1)[1] if "." in pattern else pattern
        if fnmatch.fnmatchcase(table_name, table_pattern):
            return True
    return False