"""Core inspection handler for the TableFullScan health check item."""
import json
import logging
import time
import uuid
from datetime import datetime, timezone
from typing import Dict, Any
from urllib.parse import quote

from service.datasource import DataProxyQueryError, ServiceDataSource
from service.s3_uploader import upload_result_to_s3

logger = logging.getLogger(__name__)

ITEM_NAME = "tablefullscan_slow_query"
ITEM_DESCRIPTION = (
    "Detects SQL queries with table full scan operations "
    "that scan large amounts of rows"
)
DEFAULT_SCAN_ROWS_THRESHOLD = 10000
DEFAULT_SAMPLE_TOP_SIZE = 10
DEFAULT_TOP_N = 100


def parse_iso_time(time_str: str) -> datetime:
    """Parse ISO 8601 time string to timezone-aware datetime (UTC)."""
    if time_str.endswith("Z"):
        time_str = time_str[:-1] + "+00:00"
    dt = datetime.fromisoformat(time_str)
    # Assume UTC if no timezone
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return dt


def build_s3_key(cluster_id: str, report_id: str) -> str:
    return f"healthcheck-report/{cluster_id}_{report_id}_{ITEM_NAME}.json"


def _parse_params(raw: Any) -> Dict[str, Any]:
    if raw is None:
        return {}
    if isinstance(raw, str):
        if not raw.strip():
            return {}
        try:
            parsed = json.loads(raw)
        except json.JSONDecodeError as exc:
            raise ValueError("params must be valid JSON") from exc
        if not isinstance(parsed, dict):
            raise ValueError("params must be a JSON object")
        return parsed
    if isinstance(raw, dict):
        return raw
    return {}


def _parse_int_param(value: Any, *, default: int, min_value: int, name: str) -> int:
    if value is None:
        return default
    if isinstance(value, bool):
        raise ValueError(f"{name} must be an integer")
    if isinstance(value, int):
        parsed = value
    elif isinstance(value, float):
        if not value.is_integer():
            raise ValueError(f"{name} must be an integer")
        parsed = int(value)
    elif isinstance(value, str):
        text = value.strip()
        if not text or not text.lstrip("-").isdigit():
            raise ValueError(f"{name} must be an integer")
        parsed = int(text)
    else:
        raise ValueError(f"{name} must be an integer")
    if parsed < min_value:
        raise ValueError(f"{name} must be >= {min_value}")
    return parsed


def _coerce_int(value: Any, default: int = 0) -> int:
    if value is None:
        return default
    if isinstance(value, int):
        return value
    try:
        return int(float(str(value)))
    except (TypeError, ValueError):
        return default


def _coerce_float(value: Any, default: float = 0.0) -> float:
    if value is None:
        return default
    if isinstance(value, (int, float)):
        return float(value)
    try:
        return float(value)
    except (TypeError, ValueError):
        return default


def _format_slow_log_time(value: Any) -> str:
    time_value = _coerce_float(value)
    time_dt = datetime.fromtimestamp(time_value, timezone.utc)
    return time_dt.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"


def _resolve_inspection_params(body: Dict[str, Any]) -> tuple[Dict[str, Any], int, bool, bool, int, int]:
    params = _parse_params(body.get("params", "{}"))
    top_n_raw = params.get("topN", body.get("topN"))
    threshold = _parse_int_param(
        params.get("scan_rows_threshold"),
        default=DEFAULT_SCAN_ROWS_THRESHOLD,
        min_value=1,
        name="params.scan_rows_threshold",
    )
    ignore_dml = params.get("ignore_dml", False)
    ignore_tiflash = params.get("ignore_tiflash", False)
    sample_top_size = _parse_int_param(
        params.get("sample_top_size"),
        default=DEFAULT_SAMPLE_TOP_SIZE,
        min_value=0,
        name="params.sample_top_size",
    )
    top_n = _parse_int_param(
        top_n_raw,
        default=DEFAULT_TOP_N,
        min_value=1,
        name="params.topN",
    )
    return params, threshold, ignore_dml, ignore_tiflash, sample_top_size, top_n


def validate_request_params(body: Dict[str, Any]) -> None:
    _resolve_inspection_params(body)


def run_inspection(body: Dict[str, Any]) -> Dict[str, Any]:
    """Run TableFullScan inspection based on request body."""
    started_at = time.monotonic()
    cluster_id = body["clusterId"]
    report_id = body.get("reportId", str(uuid.uuid4()))
    time_range = body["timeRange"]
    data_source = body.get("dataSource")
    options = body.get("options", {})
    region = options.get("region", "us-east-1")
    cluster_meta = body.get("clusterMeta")

    _, threshold, ignore_dml, ignore_tiflash, sample_top_size, top_n = _resolve_inspection_params(body)

    # Parse time range
    start_time = parse_iso_time(time_range["start"])
    end_time = parse_iso_time(time_range["end"])
    vendor = body.get("provider", "aws")

    logger.info(
        "Starting inspection item=%s cluster=%s report_id=%s provider=%s time=%s~%s threshold=%d top_n=%d sample_top_size=%d ignore_dml=%s ignore_tiflash=%s source=clinic-data-proxy request_data_source=%s",
        ITEM_NAME,
        cluster_id,
        report_id,
        vendor,
        start_time,
        end_time,
        threshold,
        top_n,
        sample_top_size,
        ignore_dml,
        ignore_tiflash,
        data_source,
    )

    # Build cluster proxy for clinic URL generation (optional)
    cluster_proxy = None
    if cluster_meta:
        cluster_proxy = ClusterProxy(cluster_id, cluster_meta, region)

    datasource_endpoint = None
    datasource_timeout = options.get("dataProxyTimeoutSeconds")
    datasource_page_size = options.get("dataProxyPageSize")
    deprecated_window = options.get("dataProxyWindowSeconds")
    if isinstance(data_source, dict):
        datasource_endpoint = data_source.get("endpoint")
        datasource_timeout = data_source.get("timeoutSeconds", data_source.get("timeout_seconds", datasource_timeout))
        datasource_page_size = data_source.get("pageSize", data_source.get("page_size", datasource_page_size))
        deprecated_window = data_source.get("windowSeconds", data_source.get("window_seconds", deprecated_window))
    elif data_source:
        logger.info("Ignoring legacy S3 dataSource value for cluster=%s", cluster_id)

    if deprecated_window is not None:
        logger.info("Ignoring deprecated dataProxyWindowSeconds setting for cluster=%s value=%s", cluster_id, deprecated_window)

    # Query aggregated TableFullScan stats from SQL engine (pushed down)
    datasource = ServiceDataSource(
        cluster_id=cluster_id,
        endpoint=datasource_endpoint,
        timeout_seconds=datasource_timeout,
        page_size=datasource_page_size,
    )
    inspection_error: DataProxyQueryError | None = None
    overview_summary: Dict[str, Any] = {}
    digest_summaries = []
    samples_by_digest: Dict[str, list[Dict[str, Any]]] = {}
    fetch_stage = "initialize"
    try:
        datasource.initialize()
        try:
            fetch_stage = "summary"
            logger.info(
                "Inspection data fetch started item=%s stage=%s cluster=%s report_id=%s provider=%s",
                ITEM_NAME,
                fetch_stage,
                cluster_id,
                report_id,
                vendor,
            )
            digest_summaries = datasource.query_table_full_scan_summary(
                start_time,
                end_time,
                threshold=threshold,
                vendor=vendor,
                top_n=top_n,
                ignore_dml=ignore_dml,
                ignore_tiflash=ignore_tiflash,
            )
            logger.info(
                "Inspection data fetch completed item=%s stage=%s cluster=%s report_id=%s digests=%d",
                ITEM_NAME,
                fetch_stage,
                cluster_id,
                report_id,
                len(digest_summaries),
            )
            if digest_summaries:
                overview_summary = digest_summaries[0].get("overview", {})
            else:
                fetch_stage = "overview"
                logger.info(
                    "Inspection data fetch started item=%s stage=%s cluster=%s report_id=%s provider=%s",
                    ITEM_NAME,
                    fetch_stage,
                    cluster_id,
                    report_id,
                    vendor,
                )
                overview_summary = datasource.query_table_full_scan_overview(
                    start_time,
                    end_time,
                    threshold=threshold,
                    vendor=vendor,
                    ignore_dml=ignore_dml,
                    ignore_tiflash=ignore_tiflash,
                )
                logger.info(
                    "Inspection data fetch completed item=%s stage=%s cluster=%s report_id=%s has_overview=%s",
                    ITEM_NAME,
                    fetch_stage,
                    cluster_id,
                    report_id,
                    bool(overview_summary),
                )
            if digest_summaries and sample_top_size > 0:
                fetch_stage = "samples"
                sample_digests = [summary["digest"] for summary in digest_summaries if summary.get("digest")]
                logger.info(
                    "Inspection data fetch started item=%s stage=%s cluster=%s report_id=%s digests=%d sample_top_size=%d",
                    ITEM_NAME,
                    fetch_stage,
                    cluster_id,
                    report_id,
                    len(sample_digests),
                    sample_top_size,
                )
                samples_by_digest = datasource.query_table_full_scan_samples(
                    start_time,
                    end_time,
                    sample_digests,
                    sample_size=sample_top_size,
                    vendor=vendor,
                    ignore_dml=ignore_dml,
                    ignore_tiflash=ignore_tiflash,
                )
                logger.info(
                    "Inspection data fetch completed item=%s stage=%s cluster=%s report_id=%s sample_digest_groups=%d",
                    ITEM_NAME,
                    fetch_stage,
                    cluster_id,
                    report_id,
                    len(samples_by_digest),
                )
        except DataProxyQueryError as exc:
            inspection_error = exc
            logger.error(
                "Inspection data fetch failed item=%s stage=%s cluster=%s report_id=%s provider=%s error=%s",
                ITEM_NAME,
                fetch_stage,
                cluster_id,
                report_id,
                vendor,
                exc,
            )
    finally:
        datasource.close()

    logger.info(
        "Inspection report build started item=%s cluster=%s report_id=%s digests=%d status=%s",
        ITEM_NAME,
        cluster_id,
        report_id,
        len(digest_summaries),
        "failed" if inspection_error else "success",
    )

    # Build report from aggregated SQL results
    report = _build_report_from_summary(
        digest_summaries,
        threshold,
        cluster_proxy,
        top_n=top_n,
        overview_summary=overview_summary,
        samples_by_digest=samples_by_digest,
    )

    # Build result in the required output format
    result = {
        "name": ITEM_NAME,
        "description": ITEM_DESCRIPTION,
        "status": "failed" if inspection_error else "success",
        "params": {
            "tablefullscan_anomaly_threshold": threshold,
            "topN": top_n,
            "sample_top_size": sample_top_size,
            "ignore_dml": ignore_dml,
            "ignore_tiflash": ignore_tiflash,
        },
        "summary": report["overview"],
        "results": report["results"],
    }
    if inspection_error:
        result["error"] = str(inspection_error)

    # Upload full result to S3
    s3_key = build_s3_key(cluster_id, report_id)
    logger.info(
        "Inspection S3 upload started item=%s cluster=%s report_id=%s status=%s s3_key=%s",
        ITEM_NAME,
        cluster_id,
        report_id,
        result["status"],
        s3_key,
    )
    try:
        s3_url = upload_result_to_s3(result, s3_key)
    except Exception as exc:
        logger.error(
            "Inspection S3 upload failed item=%s cluster=%s report_id=%s s3_key=%s error=%s",
            ITEM_NAME,
            cluster_id,
            report_id,
            s3_key,
            exc,
        )
        raise
    elapsed_seconds = time.monotonic() - started_at
    logger.info(
        "Inspection completed item=%s cluster=%s report_id=%s status=%s digests=%d duration_s=%.3f s3_url=%s",
        ITEM_NAME,
        cluster_id,
        report_id,
        result["status"],
        len(digest_summaries),
        elapsed_seconds,
        s3_url,
    )
    return result


def _build_report_from_summary(
    digest_summaries: list[dict],
    threshold: int,
    cluster_proxy=None,
    *,
    top_n: int = DEFAULT_TOP_N,
    overview_summary: Dict[str, Any] | None = None,
    samples_by_digest: Dict[str, list[Dict[str, Any]]] | None = None,
) -> dict:
    """Build inspection report from SQL-aggregated per-digest summaries."""
    total_tfs_queries = 0
    total_anomaly_queries = 0
    digests_with_tfs = 0
    digests_with_anomaly = 0
    overall_max_rows = 0

    digest_results = []
    for summary in digest_summaries:
        tfs_count = summary["tfs_count"]
        max_act_rows = summary["max_act_rows"]
        anomaly_count = summary["anomaly_count"]
        total_scanned_rows = _coerce_int(summary.get("total_scanned_rows"))

        if tfs_count == 0:
            continue

        digests_with_tfs += 1
        total_tfs_queries += tfs_count
        total_anomaly_queries += anomaly_count
        overall_max_rows = max(overall_max_rows, max_act_rows)

        if anomaly_count == 0:
            continue

        digests_with_anomaly += 1
        max_qt = round(summary["max_query_time"], 3)
        total_qt = round(summary["total_query_time"], 3)
        digest_results.append({
            "digest": summary["digest"],
            "query": summary["query"],
            "slow_count": tfs_count,
            "with_table_full_scan_count": tfs_count,
            "without_table_full_scan_count": 0,
            f"anomaly_logs(scan>={threshold}rows)": anomaly_count,
            "total_scanned_rows": total_scanned_rows,
            "max_scanned_rows": max_act_rows,
            "p50_scanned_rows": summary.get("p50_act_rows", max_act_rows),
            "p80_scanned_rows": summary.get("p80_act_rows", max_act_rows),
            "p95_scanned_rows": summary.get("p95_act_rows", max_act_rows),
            "p99_scanned_rows": summary.get("p99_act_rows", max_act_rows),
            "total_latency_s": total_qt,
            "max_query_time_s": max_qt,
            "p50_latency_s": round(summary.get("p50_query_time", max_qt), 3),
            "p80_latency_s": round(summary.get("p80_query_time", max_qt), 3),
            "p95_latency_s": round(summary.get("p95_query_time", max_qt), 3),
            "p99_latency_s": round(summary.get("p99_query_time", max_qt), 3),
            "samples": _build_samples(
                summary["digest"],
                samples_by_digest.get(summary["digest"], []) if samples_by_digest else [],
                cluster_proxy,
            ),
        })

    digest_results.sort(key=lambda x: x["total_scanned_rows"], reverse=True)
    digest_results = digest_results[:top_n]

    if overview_summary:
        overview = {
            "queries_with_table_full_scan": _coerce_int(overview_summary.get("tfs_count")),
            "queries_without_plan": 0,
            f"anomaly_queries(scan>={threshold}rows)": _coerce_int(overview_summary.get("anomaly_count")),
            "digests_with_table_full_scan": _coerce_int(overview_summary.get("digests_with_tfs")),
            "digests_with_anomaly_table_full_scan": _coerce_int(overview_summary.get("digests_with_anomaly")),
        }
        overall_max_rows = _coerce_int(overview_summary.get("max_act_rows"))
    else:
        overview = {
            "queries_with_table_full_scan": total_tfs_queries,
            "queries_without_plan": 0,
            f"anomaly_queries(scan>={threshold}rows)": total_anomaly_queries,
            "digests_with_table_full_scan": digests_with_tfs,
            "digests_with_anomaly_table_full_scan": digests_with_anomaly,
        }
    if overall_max_rows > 0:
        overview["max_scanned_rows"] = overall_max_rows

    return {
        "overview": overview,
        "results": digest_results,
    }


def _build_samples(
    digest: str,
    raw_samples: list[Dict[str, Any]],
    cluster_proxy=None,
) -> list[Dict[str, Any]]:
    samples = []
    for sample_record in raw_samples:
        time_value = _coerce_float(sample_record.get("time"))
        sample = {
            "time": time_value,
            "time_formatted": _format_slow_log_time(time_value),
            "conn_id": sample_record.get("conn_id", ""),
            "db": sample_record.get("db", ""),
            "act_rows": _coerce_int(sample_record.get("act_rows")),
            "latency_s": _coerce_float(sample_record.get("latency_s")),
            "full_slow_log": sample_record.get("full_slow_log", {}),
        }
        if cluster_proxy:
            sample["clinic_detail_url"] = cluster_proxy.get_clinic_url(
                digest,
                str(sample_record.get("conn_id", "")),
                str(time_value),
            )
        samples.append(sample)
    return samples


class ClusterProxy:
    """Minimal cluster-info object for clinic URL generation."""

    def __init__(self, cluster_id: str, meta: Dict[str, Any], region: str):
        self.id = cluster_id
        self.name = meta.get("clusterName", "")
        self.region = region
        self.org_name = meta.get("orgName", "")
        self.org_id = meta.get("orgId", "")
        self.project_id = meta.get("projectId", "")
        self.vendor = meta.get("vendor", "aws")

    def get_clinic_url(self, digest: str, conn_id: str, timestamp: str) -> str:
        base_url = "https://clinic.pingcap.com/portal/dashboard/cloud/ngm.html"
        try:
            ts_str = f"{float(timestamp):.6f}"
        except (ValueError, TypeError):
            ts_str = str(timestamp)
        params = [
            f"provider={quote(self.vendor)}",
            f"region={quote(self.region)}",
            f"orgName={quote(self.org_name)}",
            f"orgId={quote(self.org_id)}",
            f"projectId={quote(self.project_id)}",
            f"clusterId={quote(self.id)}",
            f"clusterName={quote(self.name)}",
            "deployType=dedicated",
        ]
        fragment = (
            f"#/slow_query/detail?digest={quote(digest)}"
            f"&connection_id={quote(conn_id)}&timestamp={ts_str}"
        )
        return f"{base_url}?{'&'.join(params)}{fragment}"