import asyncio
from typing import Any

from pydantic import ValidationError as PydanticValidationError

from src.models.read_delta import ReadDeltaRequest
from src.services.auth import (
    get_delta_credentials,
    get_gcp_access_token,
    get_valid_server_auth,
)
from src.services.duckdb_delta import read_delta
from src.services.gcp_parquet import read_gcp_delta_parquet
from src.services.observability_resolve import get_cluster_info
from src.tools.query_range import mcp
from src.utils.errors import AuthServiceError, DeltaError, NetworkError, ValidationError
from src.utils.logging_utils import get_logger
from src.utils.tool_response import format_tool_error, tool_json_response, tool_text_response

logger = get_logger("tool.read_delta")


@mcp.tool()
async def read_delta_tool(
    table_path: str,
    env: str,
    deploy_type: str,
    cluster_id: str,
    query: str | None = None,
    limit: int | None = None,
) -> list[dict[str, Any]]:
    """
    Read the slowlog and sql_statement in Delta Lake table format via DuckDB and return rows as JSON.

    Supports local paths and cloud object storage for AWS/Alicloud (`s3://`)
    and GCP (`gs://`).
    Cloud credentials are resolved automatically from server-owned auth.
    Vendor and region are auto-resolved from GS API using env/deploy_type.
    Reference the table as `__t__` in custom SQL queries.
    For GCP, query must include exactly one date partition filter:
    `date='yyyymmdd'`.
    In GCP flow, that predicate is consumed by the download step and removed before
    local parquet query execution.
    GCP responses include `local_cache_path` alongside `rows`.

    Args:
        table_path: Delta table URI (e.g. 's3://bucket/path', 'gs://bucket/path',
            'file:///local/path').
        env: Environment name (e.g. 'prod', 'staging', 'dev').
        deploy_type: Cluster deploy type (e.g. 'dedicated', 'premium', 'starter').
        cluster_id: TiDB Cloud cluster ID (string). Empty for local.
        query: DuckDB SQL using `__t__` (e.g. 'SELECT * FROM __t__ LIMIT 50').
            Only SELECT allowed. Returns all columns if omitted.
        limit: Max rows (default 10000, max 100000).
    """
    try:
        logger.info(
            "tool_call_start name=read_delta has_query=%s limit=%s",
            bool(query),
            limit,
        )
        if deploy_type.strip().lower() == "local":
            cluster = {"vendor": "local", "region": None}
        else:
            cluster, _ = await get_cluster_info(
                cluster_id=cluster_id,
                env=env,
                deploy_type=deploy_type,
            )
        try:
            request = ReadDeltaRequest.model_validate(
                {
                    "table_path": table_path.strip(),
                    "region": cluster.get("region"),
                    "vendor": cluster.get("vendor"),
                    "cluster_id": str(cluster_id).strip(),
                    "query": query.strip() if query else None,
                    "limit": limit,
                }
            )
        except (ValueError, PydanticValidationError) as e:
            raise ValidationError(str(e)) from e

        if request.vendor != "local":
            await get_valid_server_auth(vendor=request.vendor)

        # GCP currently receives OAuth access_token (not S3-compatible key/secret),
        # so reads go through GCS object listing/downloading and local parquet scan.
        if request.vendor == "gcp":
            token = await get_gcp_access_token(vendor="gcp")
            if not token:
                raise AuthServiceError(
                    "GCP access_token is missing from auth service response."
                )
            gcp_result = await asyncio.to_thread(
                read_gcp_delta_parquet,
                request.table_path,
                token,
                request.query,
                request.limit,
            )
            rows = gcp_result.get("rows", [])
        else:
            # AWS/Alicloud continue using DuckDB delta_scan with S3-compatible secrets.
            credentials = None
            try:
                if request.vendor != "local":
                    credentials = await get_delta_credentials(vendor=request.vendor)
            except (AuthServiceError, NetworkError) as e:
                return tool_text_response(
                    format_tool_error(e)
                    + "\n\nS3 credentials could not be obtained from the auth service. "
                    "Check AUTH_SERVICE_URL and the server-owned GitHub token."
                )

            if credentials is not None:
                credentials["region"] = request.region

            rows = await asyncio.to_thread(
                read_delta,
                request.table_path,
                request.vendor,
                request.query,
                request.limit,
                credentials,
            )
        logger.info(
            "tool_call_success name=read_delta vendor=%s row_count=%s",
            request.vendor,
            len(rows),
        )
        if request.vendor == "gcp":
            return tool_json_response(
                {
                    "local_cache_path": gcp_result.get("local_cache_path"),
                    "rows": rows,
                }
            )
        return tool_json_response(rows)
    except (ValidationError, DeltaError) as e:
        logger.warning("tool_call_error name=read_delta type=%s", e.__class__.__name__)
        return tool_text_response(format_tool_error(e))
    except Exception as e:
        logger.exception("tool_call_unexpected_error name=read_delta")
        return tool_text_response(format_tool_error(e))