"""Service for reading Delta Lake tables via DuckDB with cloud/local support."""

from __future__ import annotations

import re
from typing import Any, Callable, Optional

import duckdb

from src.models.read_delta import Vendor
from src.utils.errors import DeltaError

# Default row limit when none specified
DEFAULT_LIMIT = 10_000

# Allowed SQL for optional query: must be a single SELECT ... FROM __t__ with optional LIMIT
# We inject delta_scan as __t__ and only allow this pattern to avoid injection.
_SAFE_QUERY_PATTERN = re.compile(
    r"^\s*SELECT\s+.+FROM\s+__t__(?:\s+WHERE\s+.+)?(?:\s+ORDER\s+BY\s+.+)?(?:\s+LIMIT\s+\d+)?\s*$",
    re.IGNORECASE | re.DOTALL,
)


def _escape_sql_string(s: str) -> str:
    """Escape single quotes for use inside SQL single-quoted string."""
    return s.replace("'", "''")


def _create_secret_aws(
    conn: duckdb.DuckDBPyConnection,
    credentials: Optional[dict[str, Any]] = None,
) -> None:
    key_id = credentials.get("access_key_id") if credentials else None
    secret = credentials.get("secret_access_key") if credentials else None
    region = credentials.get("region") if credentials else None
    session_token = credentials.get("session_token") if credentials else None
    if not key_id or not secret or not region or not session_token:
        raise DeltaError(
            "AWS credentials are required from auth service. "
            "Configure AUTH_SERVICE_URL and GITHUB_TOKEN and retry."
        )
    conn.execute(
        f"""
        CREATE OR REPLACE SECRET delta_aws (
            TYPE S3,
            PROVIDER config,
            KEY_ID '{_escape_sql_string(key_id)}',
            SECRET '{_escape_sql_string(secret)}',
            REGION '{_escape_sql_string(region)}',
            SESSION_TOKEN '{_escape_sql_string(session_token)}'
        )
        """
    )


def _create_secret_alicloud(
    conn: duckdb.DuckDBPyConnection,
    credentials: Optional[dict[str, Any]] = None,
) -> None:
    key_id = credentials.get("access_key_id") if credentials else None
    secret = credentials.get("secret_access_key") if credentials else None
    region = credentials.get("region") if credentials else None
    session_token = credentials.get("session_token") if credentials else None
    if not key_id or not secret or not region or not session_token:
        raise DeltaError(
            "Alicloud credentials are required from auth service. "
            "Configure AUTH_SERVICE_URL and GITHUB_TOKEN and retry."
            f"credentials={credentials}"
        )

    region_name = str(region).strip()
    endpoint = f"oss-{region_name}.aliyuncs.com"
    conn.execute(
        f"""
        CREATE OR REPLACE SECRET delta_alicloud (
            TYPE S3,
            PROVIDER config,
            KEY_ID '{_escape_sql_string(key_id)}',
            SECRET '{_escape_sql_string(secret)}',
            REGION '{_escape_sql_string(region)}',
            SESSION_TOKEN '{_escape_sql_string(session_token)}',
            ENDPOINT '{_escape_sql_string(endpoint)}'
        )
        """
    )


def _ensure_delta_extension(conn: duckdb.DuckDBPyConnection) -> None:
    try:
        conn.execute("INSTALL delta")
        conn.execute("LOAD delta")
    except Exception as e:
        raise DeltaError(f"Failed to load Delta extension: {e}") from e


# Cloud -> secret setup; local needs no secret.
# Setup receives optional credentials for AWS/Alicloud.
_SecretSetup = Callable[[duckdb.DuckDBPyConnection, Optional[dict[str, Any]]], None] | None
_CLOUD_SECRET_SETUP: dict[Vendor, _SecretSetup] = {
    "aws": _create_secret_aws,
    "alicloud": _create_secret_alicloud,
    "local": None,
}


def read_delta(
    table_path: str,
    vendor: Vendor,
    query: str | None = None,
    limit: int | None = None,
    credentials: Optional[dict[str, Any]] = None,
) -> list[dict[str, Any]]:
    conn = duckdb.connect(":memory:")
    try:
        _ensure_delta_extension(conn)
        setup = _CLOUD_SECRET_SETUP.get(vendor)
        if setup is not None:
            setup(conn, credentials)

        path_escaped = _escape_sql_string(table_path)
        effective_limit = limit if limit is not None else DEFAULT_LIMIT

        if query and query.strip():
            if not _SAFE_QUERY_PATTERN.match(query.strip()):
                raise DeltaError(
                    "Optional query must be of the form: SELECT ... FROM __t__ "
                    "[WHERE ...] [ORDER BY ...] [LIMIT n]"
                )
            # Replace __t__ with delta_scan(...) subquery
            safe_from = f"(SELECT * FROM delta_scan('{path_escaped}') AS __t__)"
            user_sql = query.strip()
            # Replace FROM __t__ with FROM (SELECT * FROM delta_scan(...) AS __t__)
            sql = re.sub(
                r"\bFROM\s+__t__\b",
                "FROM " + safe_from,
                user_sql,
                count=1,
                flags=re.IGNORECASE,
            )
            if sql == user_sql:
                raise DeltaError("Query must contain FROM __t__ to reference the Delta table.")
        else:
            sql = f"SELECT * FROM delta_scan('{path_escaped}') LIMIT {effective_limit}"

        try:
            result = conn.execute(sql)
            df = result.fetchdf()
        except Exception as e:
            raise DeltaError(f"DuckDB/Delta read failed: {e}") from e

        if df is None or df.empty:
            return []
        return df.to_dict("records")
    finally:
        conn.close()