from __future__ import annotations

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

import pymysql
from google.cloud import bigquery

from .config import PipelineConfig
from .models import Column, Table


def inspect_tables(config: PipelineConfig, selected: list[str] | None = None) -> list[Table]:
    if config.source_type == "bigquery":
        return inspect_bigquery_tables(config, selected=selected)
    if config.source_type == "tidbcloud":
        return inspect_tidb_tables(config, selected=selected)
    if config.source_type in {"s3", "databend", "mysql_query"}:
        return inspect_external_table(config, selected=selected)
    raise ValueError(f"unsupported source type: {config.source_type}")


def selected_table_names(config: PipelineConfig, selected: list[str] | None = None) -> list[str]:
    return [table.name for table in inspect_tables(config, selected=selected)]


def normalize_exact_tidb_selection(config: PipelineConfig, selected: list[str]) -> list[str]:
    normalized: list[str] = []
    seen: set[str] = set()
    for raw_name in selected:
        value = raw_name.strip()
        if not value:
            raise ValueError("TiDB Cloud explicit table selection contains an empty name")
        if any(character in value for character in "*?["):
            raise ValueError(
                f"TiDB Cloud explicit table selection must use exact table names, not patterns: {value!r}"
            )
        if "." in value:
            database, table_name = value.split(".", 1)
            if database != config.source_database:
                raise ValueError(
                    f"TiDB Cloud explicit table {value!r} targets database {database!r}; "
                    f"expected {config.source_database!r}"
                )
        else:
            table_name = value
        if not table_name or "." in table_name:
            raise ValueError(f"invalid TiDB Cloud explicit table name: {value!r}")
        if table_name in seen:
            raise ValueError(f"duplicate TiDB Cloud explicit table selection: {table_name}")
        seen.add(table_name)
        normalized.append(table_name)
    if not normalized:
        raise ValueError("TiDB Cloud explicit table selection must not be empty")
    return normalized


def validate_exact_tidb_selection(requested: list[str], tables: list[Table]) -> None:
    resolved = [table.name for table in tables]
    duplicate_resolved = sorted({name for name in resolved if resolved.count(name) > 1})
    if duplicate_resolved:
        raise ValueError(f"TiDB Cloud source inspection returned duplicate tables: {duplicate_resolved}")
    missing = sorted(set(requested) - set(resolved))
    unexpected = sorted(set(resolved) - set(requested))
    if missing or unexpected or len(resolved) != len(requested):
        raise ValueError(
            "TiDB Cloud explicit table selection did not resolve exactly; "
            f"missing={missing}, unexpected={unexpected}, "
            f"requested_count={len(requested)}, resolved_count={len(resolved)}"
        )


def inspect_exact_tidb_tables(config: PipelineConfig, selected: list[str]) -> list[Table]:
    requested = normalize_exact_tidb_selection(config, selected)
    tables = inspect_tidb_tables(config, selected=requested)
    validate_exact_tidb_selection(requested, tables)
    return tables


def table_selection_contract(
    config: PipelineConfig,
    table_names: list[str],
    *,
    requested: list[str] | None,
) -> dict[str, object]:
    resolved = sorted(table_names)
    qualified_patterns = [f"{config.source_database}.{name}" for name in resolved]
    canonical = "\n".join(qualified_patterns).encode()
    return {
        "mode": "explicit_subset" if requested is not None else "configured",
        "requested_tables": normalize_exact_tidb_selection(config, requested) if requested is not None else None,
        "resolved_tables": resolved,
        "qualified_patterns": qualified_patterns,
        "qualified_patterns_sha256": hashlib.sha256(canonical).hexdigest(),
        "canonicalization": "utf8_newline_joined_sorted_qualified_patterns",
    }


def inspect_bigquery_tables(config: PipelineConfig, selected: list[str] | None = None) -> list[Table]:
    client = bigquery.Client(project=config.target_project, location=config.location)
    names = kept_bigquery_table_names(client, config, selected=selected)
    if not names:
        return []

    columns_query = f"""
SELECT table_name, column_name, data_type, is_nullable, ordinal_position
FROM `{config.source_project}.{config.source_dataset}.INFORMATION_SCHEMA.COLUMNS`
WHERE table_name IN UNNEST(@table_names)
ORDER BY table_name, ordinal_position
"""
    rows = client.query(
        columns_query,
        job_config=bigquery.QueryJobConfig(
            query_parameters=[bigquery.ArrayQueryParameter("table_names", "STRING", names)]
        ),
        location=config.location,
    ).result()

    columns_by_table: dict[str, list[Column]] = {name: [] for name in names}
    for row in rows:
        columns_by_table[row.table_name].append(
            Column(
                name=row.column_name,
                data_type=str(row.data_type).split("(")[0].lower(),
                column_type=str(row.data_type),
                nullable=str(row.is_nullable).upper() == "YES",
                ordinal_position=int(row.ordinal_position),
            )
        )

    estimates = bigquery_row_estimates(client, config, names)
    return [
        Table(
            database=config.source_dataset,
            name=name,
            rows_estimate=estimates.get(name),
            columns=columns_by_table[name],
            primary_key=config.primary_key_for(name),
            watermark_candidates=config.watermark_candidates,
        )
        for name in names
    ]


def kept_bigquery_table_names(
    client: bigquery.Client,
    config: PipelineConfig,
    selected: list[str] | None = None,
) -> list[str]:
    selected = config.selected_tables if selected is None else selected
    rows = client.query(
        f"""
SELECT table_name
FROM `{config.source_project}.{config.source_dataset}.INFORMATION_SCHEMA.TABLES`
WHERE table_type IN ('BASE TABLE', 'VIEW')
""",
        location=config.location,
    ).result()
    names = [
        row.table_name
        for row in rows
        if include_table(row.table_name, selected=selected, excluded=config.excluded_tables)
    ]
    if config.require_primary_key:
        names = [name for name in names if config.primary_key_for(name)]
    return sorted(names)


def bigquery_row_estimates(client: bigquery.Client, config: PipelineConfig, names: list[str]) -> dict[str, int | None]:
    rows = client.query(
        f"""
SELECT table_id, row_count
FROM `{config.source_project}.{config.source_dataset}.__TABLES__`
WHERE table_id IN UNNEST(@table_names)
""",
        job_config=bigquery.QueryJobConfig(
            query_parameters=[bigquery.ArrayQueryParameter("table_names", "STRING", names)]
        ),
        location=config.location,
    ).result()
    return {row.table_id: int(row.row_count) if row.row_count is not None else None for row in rows}


def bigquery_table_types(client: bigquery.Client, config: PipelineConfig, names: list[str]) -> dict[str, str]:
    rows = client.query(
        f"""
SELECT table_name, table_type
FROM `{config.source_project}.{config.source_dataset}.INFORMATION_SCHEMA.TABLES`
WHERE table_name IN UNNEST(@table_names)
""",
        job_config=bigquery.QueryJobConfig(
            query_parameters=[bigquery.ArrayQueryParameter("table_names", "STRING", names)]
        ),
        location=config.location,
    ).result()
    return {row.table_name: row.table_type for row in rows}


def inspect_tidb_tables(config: PipelineConfig, selected: list[str] | None = None) -> list[Table]:
    with tidb_connect(config) as conn:
        with conn.cursor() as cursor:
            names = kept_tidb_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),
            )
            estimates = {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=str(row["DATA_TYPE"]).lower(),
                        column_type=str(row["COLUMN_TYPE"]),
                        nullable=row["IS_NULLABLE"] == "YES",
                        ordinal_position=int(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=estimates.get(name),
            columns=columns_by_table[name],
            primary_key=primary_keys[name],
            watermark_candidates=config.watermark_candidates,
        )
        for name in names
    ]


def inspect_external_table(config: PipelineConfig, selected: list[str] | None = None) -> list[Table]:
    configured_tables = config.query_tables if config.source_type == "mysql_query" else [config.external]
    selected_tables = (config.selected_tables or ["*"]) if selected is None else selected
    tables: list[Table] = []
    for configured in configured_tables:
        table_name = str(configured["table"])
        if not include_table(table_name, selected=selected_tables, excluded=config.excluded_tables):
            continue
        columns = [
            Column(
                name=str(column["name"]),
                data_type=str(column.get("data_type", column.get("type", "string"))),
                column_type=str(column.get("column_type", column.get("type", "STRING"))),
                nullable=bool(column.get("nullable", True)),
                ordinal_position=int(column.get("ordinal_position", index)),
            )
            for index, column in enumerate(configured.get("columns", []), start=1)
        ]
        row_estimate = configured.get("row_estimate")
        tables.append(
            Table(
                database=config.source_database,
                name=table_name,
                rows_estimate=int(row_estimate) if row_estimate is not None else None,
                columns=columns,
                primary_key=config.primary_key_for(table_name),
                watermark_candidates=config.watermark_candidates,
            )
        )
    return tables


def tidb_connect(config: PipelineConfig):
    dsn = os.environ.get(config.source_dsn_env)
    if not dsn:
        raise RuntimeError(f"required environment variable is not set: {config.source_dsn_env}")
    parsed = urlparse(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"}:
        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_tidb_table_names(cursor, config: PipelineConfig, selected: list[str] | None = None) -> list[str]:
    selected = config.selected_tables if selected is None else selected
    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=selected, excluded=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 tidb_export_patterns(config: PipelineConfig, selected: list[str] | None = None) -> list[str]:
    if selected is not None:
        return [
            f"{config.source_database}.{table.name}"
            for table in inspect_exact_tidb_tables(config, selected)
        ]
    selected = config.selected_tables
    has_wildcard = not selected or any(any(character in pattern for character in "*?[") for pattern in selected)
    if config.tidb_export_fault_domains and has_wildcard:
        return [f"{config.source_database}.{name}" for name in selected_table_names(config)]
    if not config.excluded_tables and not config.require_primary_key:
        if not config.selected_tables or config.selected_tables == ["*"]:
            return [f"{config.source_database}.*"]
        return [
            table if "." in table else f"{config.source_database}.{table}" for table in config.selected_tables
        ]
    return [f"{config.source_database}.{name}" for name in selected_table_names(config)]


def tidb_export_filter(config: PipelineConfig) -> str:
    return ",".join(tidb_export_patterns(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
    return any(
        fnmatch.fnmatchcase(table_name, pattern.split(".", 1)[1] if "." in pattern else pattern)
        for pattern in selected
    )