package handlers

import (
	"context"
	stderrors "errors"
	"fmt"
	"math"
	"net"
	"net/http"
	"net/url"
	"regexp"
	"strconv"
	"time"

	"github.com/gin-gonic/gin"
	gozap "go.uber.org/zap"

	clinic_internal_api "github.com/pingcap-inc/clinic/clinic_data_service/pkg/clinic-internal-api"
	apierrors "github.com/pingcap-inc/clinic/clinic_data_service/pkg/errors"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/log"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/observability"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/service"
)

const (
	maxBYOCTopSQLQueryBytes       = 16 << 10
	maxBYOCTopSQLTimestamp        = 253402300799
	maxBYOCTopSQLTimeRangeSeconds = 15 * 24 * 60 * 60
	maxBYOCTopSQLWindow           = 15 * 24 * time.Hour
	maxBYOCTopSQLBuckets          = 4096
	maxBYOCTopSQLTop              = 100
	maxBYOCTopSQLInstanceBytes    = 512
)

var (
	byocTopSQLDecimalPattern  = regexp.MustCompile(`^[0-9]+(?:\.[0-9]+)?$`)
	byocTopSQLIntegerPattern  = regexp.MustCompile(`^[0-9]+$`)
	byocTopSQLInstancePattern = regexp.MustCompile(`^[A-Za-z0-9_.:\-\[\]]+$`)

	byocTopSQLInstancesKeys = map[string]struct{}{
		"start": {}, "end": {},
	}
	byocTopSQLSummaryKeys = map[string]struct{}{
		"start": {}, "end": {}, "window": {}, "top": {}, "instance": {}, "instance_type": {},
	}
	byocTopSQLInstanceTypes = map[string]struct{}{
		"tidb": {}, "tikv": {}, "pd": {},
	}
	byocTopSQLRoutingHeaders = map[string]struct{}{
		"xenv": {}, "xbiztype": {}, "xregionid": {}, "xapplicationid": {},
		"xdeploytype": {}, "xdeploytypev2": {}, "xo11yclusterid": {},
	}
)

type byocTopSQLQuerier interface {
	Query(context.Context, uint64, string, service.BYOCTopSQLOperation, url.Values) (*service.BYOCTopSQLResult, error)
}

// BYOCTopSQLHandler exposes BYOC Top SQL data read directly from O11y. It has
// no warehouse, SQL, S3, Athena, or table-management dependency.
type BYOCTopSQLHandler struct {
	service byocTopSQLQuerier
}

// byocTopSQLResponse documents the stable O11y envelope. Runtime responses are
// forwarded byte-for-byte, including fields added by O11y in the future.
type byocTopSQLResponse struct {
	Status string                   `json:"status" binding:"required" minLength:"1"`
	Data   []map[string]interface{} `json:"data" binding:"required"`
}

func NewBYOCTopSQLHandler(svc byocTopSQLQuerier) *BYOCTopSQLHandler {
	return &BYOCTopSQLHandler{service: svc}
}

// Instances returns Top SQL instances from the resolved BYOC O11y target.
//
//	@Summary		List BYOC Top SQL instances
//	@ID			canonical-byoc-top-sql-instances
//	@Tags			data-proxy-top-sql
//	@Produce		json
//	@Param			orgId		path		string	true	"TiDB Cloud/DBAAS org ID"
//	@Param			clusterId	path		string	true	"Logical BYOC or physical byoc-host cluster ID"
//	@Param			start		query		number	true	"Strict decimal Unix seconds; the start/end span is at most 15 days"	minimum(0)	maximum(253402300799)
//	@Param			end			query		number	true	"Strict decimal Unix seconds; the start/end span is at most 15 days"	minimum(0)	maximum(253402300799)
//	@Success		200			{object}	byocTopSQLResponse
//	@Failure		400			{object}	apierrors.APIError
//	@Failure		401			{object}	apierrors.APIError
//	@Failure		403			{object}	apierrors.APIError
//	@Failure		404			{object}	apierrors.APIError
//	@Failure		413			{object}	apierrors.APIError
//	@Failure		429			{object}	apierrors.APIError
//	@Failure		502			{object}	apierrors.APIError
//	@Failure		503			{object}	apierrors.APIError
//	@Failure		504			{object}	apierrors.APIError
//	@Security		BearerAuth
//	@Router			/data-proxy/api/v1/orgs/{orgId}/clusters/{clusterId}/top-sql/instances [get]
func (h *BYOCTopSQLHandler) Instances(c *gin.Context) {
	h.handle(c, "canonical_byoc_top_sql_instances", service.BYOCTopSQLOperationInstances, canonicalBYOCTopSQLInstances)
}

// Summary returns the Top SQL summary from the resolved BYOC O11y target.
//
//	@Summary		Get BYOC Top SQL summary
//	@ID			canonical-byoc-top-sql-summary
//	@Tags			data-proxy-top-sql
//	@Produce		json
//	@Param			orgId			path		string	true	"TiDB Cloud/DBAAS org ID"
//	@Param			clusterId		path		string	true	"Logical BYOC or physical byoc-host cluster ID"
//	@Param			start			query		number	true	"Strict decimal Unix seconds; the start/end span is at most 15 days"	minimum(0)	maximum(253402300799)
//	@Param			end				query		number	true	"Strict decimal Unix seconds; the start/end span is at most 15 days"	minimum(0)	maximum(253402300799)
//	@Param			window			query		string	true	"Whole-second Go duration from 1s through 360h; at most 4096 windows"
//	@Param			top				query		int		true	"Top N SQL statements"	minimum(1)	maximum(100)
//	@Param			instance		query		string	true	"Top SQL instance"	minlength(1)	maxlength(512)
//	@Param			instance_type	query		string	true	"Top SQL instance type"	Enums(tidb,tikv,pd)
//	@Success		200				{object}	byocTopSQLResponse
//	@Failure		400				{object}	apierrors.APIError
//	@Failure		401				{object}	apierrors.APIError
//	@Failure		403				{object}	apierrors.APIError
//	@Failure		404				{object}	apierrors.APIError
//	@Failure		413				{object}	apierrors.APIError
//	@Failure		429				{object}	apierrors.APIError
//	@Failure		502				{object}	apierrors.APIError
//	@Failure		503				{object}	apierrors.APIError
//	@Failure		504				{object}	apierrors.APIError
//	@Security		BearerAuth
//	@Router			/data-proxy/api/v1/orgs/{orgId}/clusters/{clusterId}/top-sql/summary [get]
func (h *BYOCTopSQLHandler) Summary(c *gin.Context) {
	h.handle(c, "canonical_byoc_top_sql_summary", service.BYOCTopSQLOperationSummary, canonicalBYOCTopSQLSummary)
}

type byocTopSQLParamsParser func(*gin.Context) (url.Values, error)

func (h *BYOCTopSQLHandler) handle(
	c *gin.Context,
	endpoint string,
	operation service.BYOCTopSQLOperation,
	parseParams byocTopSQLParamsParser,
) {
	start, rec := beginCanonicalDataProxyRequest(c, endpoint)
	rec.QueryType = "topsql"
	rec.Engine = "o11y"
	defer func() { observability.RecordDataProxy(c, start, rec) }()

	// This check deliberately precedes all helpers that parse the query.
	if len(c.Request.URL.RawQuery) > maxBYOCTopSQLQueryBytes {
		rec.ErrCode = "INVALID_TOPSQL_PARAMETERS"
		writeResourceError(c, http.StatusRequestEntityTooLarge, "PayloadTooLarge", errBYOCTopSQLParametersTooLarge.Error(), nil)
		return
	}
	if !requireNoCanonicalBody(c, rec) || !rejectCanonicalPlacement(c, rec) || !rejectBYOCTopSQLRoutingHeaders(c, rec) {
		return
	}
	orgID, clusterID, ok := parseCanonicalTelemetryScope(c, rec)
	if !ok {
		return
	}
	query, err := parseParams(c)
	if err != nil {
		rec.ErrCode = "INVALID_TOPSQL_PARAMETERS"
		status := http.StatusBadRequest
		code := "InvalidArgument"
		if stderrors.Is(err, errBYOCTopSQLParametersTooLarge) {
			status = http.StatusRequestEntityTooLarge
			code = "PayloadTooLarge"
		}
		writeResourceError(c, status, code, err.Error(), nil)
		return
	}
	if h == nil || h.service == nil {
		rec.ErrCode = "TOPSQL_UNAVAILABLE"
		writeResourceError(c, http.StatusServiceUnavailable, "Unavailable", "BYOC Top SQL service is temporarily unavailable", nil)
		return
	}

	result, err := h.service.Query(c.Request.Context(), orgID, clusterID, operation, query)
	if err != nil {
		h.writeError(c, rec, err, clusterID)
		return
	}
	if result == nil {
		rec.ErrCode = "TOPSQL_INVALID_RESPONSE"
		writeResourceError(c, http.StatusBadGateway, "BadGateway", "BYOC Top SQL returned an invalid response", nil)
		return
	}
	rec.Provider = result.Provider
	rec.Region = result.Region
	rec.DeployTypeV2 = result.DeployTypeV2
	c.Data(http.StatusOK, "application/json; charset=utf-8", result.Body)
}

func rejectBYOCTopSQLRoutingHeaders(c *gin.Context, rec *observability.DataProxyRecord) bool {
	for key := range c.Request.Header {
		if _, blocked := byocTopSQLRoutingHeaders[normalizeCanonicalRoutingKey(key)]; blocked {
			return writeCanonicalPlacementError(c, rec)
		}
	}
	return true
}

func (h *BYOCTopSQLHandler) RegisterRoutes(r *gin.Engine) {
	group := r.Group("/data-proxy/api/v1/orgs/:orgId/clusters/:clusterId/top-sql")
	group.GET("/instances", h.Instances)
	group.GET("/summary", h.Summary)
}

var errBYOCTopSQLParametersTooLarge = stderrors.New("query parameters exceed the allowed size")

func canonicalBYOCTopSQLInstances(c *gin.Context) (url.Values, error) {
	values, err := parseBYOCTopSQLValues(c, byocTopSQLInstancesKeys)
	if err != nil {
		return nil, err
	}
	_, _, result, err := canonicalBYOCTopSQLTimeRange(values)
	if err != nil {
		return nil, err
	}
	return result, nil
}

func canonicalBYOCTopSQLSummary(c *gin.Context) (url.Values, error) {
	values, err := parseBYOCTopSQLValues(c, byocTopSQLSummaryKeys)
	if err != nil {
		return nil, err
	}
	start, end, result, err := canonicalBYOCTopSQLTimeRange(values)
	if err != nil {
		return nil, err
	}
	for _, key := range []string{"window", "top", "instance", "instance_type"} {
		if len(values[key]) != 1 {
			return nil, fmt.Errorf("%s must be provided exactly once", key)
		}
	}

	window, err := time.ParseDuration(values.Get("window"))
	if err != nil || window < time.Second || window > maxBYOCTopSQLWindow || window%time.Second != 0 {
		return nil, fmt.Errorf("window must be a whole-second duration between 1s and 360h")
	}
	if math.Ceil((end-start)/window.Seconds()) > maxBYOCTopSQLBuckets {
		return nil, fmt.Errorf("time range must not contain more than %d windows", maxBYOCTopSQLBuckets)
	}

	topRaw := values.Get("top")
	if !byocTopSQLIntegerPattern.MatchString(topRaw) {
		return nil, fmt.Errorf("top must be an integer between 1 and %d", maxBYOCTopSQLTop)
	}
	top, err := strconv.Atoi(topRaw)
	if err != nil || top < 1 || top > maxBYOCTopSQLTop {
		return nil, fmt.Errorf("top must be an integer between 1 and %d", maxBYOCTopSQLTop)
	}

	instance := values.Get("instance")
	if len(instance) == 0 || len(instance) > maxBYOCTopSQLInstanceBytes || !byocTopSQLInstancePattern.MatchString(instance) {
		return nil, fmt.Errorf("instance must be 1-%d bytes using only letters, digits, _, ., :, -, [, or ]", maxBYOCTopSQLInstanceBytes)
	}
	instanceType := values.Get("instance_type")
	if _, ok := byocTopSQLInstanceTypes[instanceType]; !ok {
		return nil, fmt.Errorf("instance_type must be tidb, tikv, or pd")
	}

	result.Set("top", strconv.Itoa(top))
	// Preserve the caller's valid Go duration spelling for compatibility with
	// the existing NGM/O11y contract (for example, 10m remains 10m).
	result.Set("window", values.Get("window"))
	result.Set("instance", instance)
	result.Set("instance_type", instanceType)
	return result, nil
}

func parseBYOCTopSQLValues(c *gin.Context, allowed map[string]struct{}) (url.Values, error) {
	if len(c.Request.URL.RawQuery) > maxBYOCTopSQLQueryBytes {
		return nil, errBYOCTopSQLParametersTooLarge
	}
	values, err := url.ParseQuery(c.Request.URL.RawQuery)
	if err != nil {
		return nil, fmt.Errorf("query parameters are malformed")
	}
	for key := range values {
		if _, ok := allowed[key]; !ok {
			return nil, fmt.Errorf("unsupported Top SQL parameter %q", key)
		}
		if len(values[key]) != 1 {
			return nil, fmt.Errorf("%s must be provided exactly once", key)
		}
	}
	return values, nil
}

func canonicalBYOCTopSQLTimeRange(values url.Values) (float64, float64, url.Values, error) {
	for _, key := range []string{"start", "end"} {
		if len(values[key]) != 1 {
			return 0, 0, nil, fmt.Errorf("%s must be provided exactly once", key)
		}
	}
	start, startText, err := canonicalBYOCTopSQLTimestamp("start", values.Get("start"))
	if err != nil {
		return 0, 0, nil, err
	}
	end, endText, err := canonicalBYOCTopSQLTimestamp("end", values.Get("end"))
	if err != nil {
		return 0, 0, nil, err
	}
	if end < start {
		return 0, 0, nil, fmt.Errorf("end must be greater than or equal to start")
	}
	if end-start > maxBYOCTopSQLTimeRangeSeconds {
		return 0, 0, nil, fmt.Errorf("time range must not exceed %d seconds", maxBYOCTopSQLTimeRangeSeconds)
	}
	result := cloneURLValues(values)
	result.Set("start", startText)
	result.Set("end", endText)
	return start, end, result, nil
}

func canonicalBYOCTopSQLTimestamp(key, raw string) (float64, string, error) {
	if !byocTopSQLDecimalPattern.MatchString(raw) {
		return 0, "", fmt.Errorf("%s must be a finite non-negative decimal Unix timestamp", key)
	}
	value, err := strconv.ParseFloat(raw, 64)
	if err != nil || math.IsNaN(value) || math.IsInf(value, 0) || value < 0 || value > maxBYOCTopSQLTimestamp {
		return 0, "", fmt.Errorf("%s must be a finite non-negative decimal Unix timestamp no greater than %d", key, int64(maxBYOCTopSQLTimestamp))
	}
	return value, strconv.FormatFloat(value, 'f', -1, 64), nil
}

func (h *BYOCTopSQLHandler) writeError(
	c *gin.Context,
	rec *observability.DataProxyRecord,
	err error,
	clusterID string,
) {
	switch {
	case stderrors.Is(err, context.Canceled):
		rec.ErrCode = "TOPSQL_CANCELED"
		writeResourceError(c, clientClosedStatus, "Canceled", "BYOC Top SQL request was canceled", nil)
		return
	case stderrors.Is(err, context.DeadlineExceeded):
		rec.ErrCode = "TOPSQL_TIMEOUT"
		writeResourceError(c, http.StatusGatewayTimeout, "DeadlineExceeded", "BYOC Top SQL request timed out", nil)
		return
	case stderrors.Is(err, service.ErrBYOCTopSQLUnavailable):
		rec.ErrCode = "TOPSQL_UNAVAILABLE"
		writeResourceError(c, http.StatusServiceUnavailable, "Unavailable", "BYOC Top SQL service is temporarily unavailable", nil)
		return
	case stderrors.Is(err, service.ErrInvalidBYOCTopSQLTarget), stderrors.Is(err, service.ErrInvalidBYOCTopSQLResponse):
		rec.ErrCode = "TOPSQL_INVALID_UPSTREAM"
		writeResourceError(c, http.StatusBadGateway, "BadGateway", "BYOC Top SQL returned an invalid response", nil)
		return
	}

	var networkError net.Error
	if stderrors.As(err, &networkError) {
		if networkError.Timeout() {
			rec.ErrCode = "TOPSQL_TIMEOUT"
			writeResourceError(c, http.StatusGatewayTimeout, "DeadlineExceeded", "BYOC Top SQL request timed out", nil)
			return
		}
		rec.ErrCode = "TOPSQL_UNAVAILABLE"
		writeResourceError(c, http.StatusServiceUnavailable, "Unavailable", "BYOC Top SQL service is temporarily unavailable", nil)
		return
	}

	var targetError *clinic_internal_api.UpstreamError
	if stderrors.As(err, &targetError) {
		switch targetError.StatusCode {
		case http.StatusBadRequest:
			rec.ErrCode = "TOPSQL_UNSUPPORTED"
			writeResourceError(c, http.StatusBadRequest, "InvalidArgument", "BYOC Top SQL is not supported for this cluster", nil)
		case http.StatusNotFound:
			rec.ErrCode = "CLUSTER_NOT_FOUND"
			writeResourceError(c, http.StatusNotFound, "NotFound", "cluster not found", nil)
		case http.StatusRequestTimeout, http.StatusGatewayTimeout:
			rec.ErrCode = "TOPSQL_TARGET_TIMEOUT"
			writeResourceError(c, http.StatusGatewayTimeout, "DeadlineExceeded", "BYOC Top SQL target resolution timed out", nil)
		case http.StatusTooManyRequests:
			rec.ErrCode = "TOPSQL_TARGET_RATE_LIMITED"
			writeResourceError(c, http.StatusTooManyRequests, "ResourceExhausted", "BYOC Top SQL target rate limit exceeded", nil)
		case http.StatusServiceUnavailable:
			rec.ErrCode = "TOPSQL_TARGET_UNAVAILABLE"
			writeResourceError(c, http.StatusServiceUnavailable, "Unavailable", "BYOC Top SQL target is temporarily unavailable", nil)
		default:
			rec.ErrCode = "TOPSQL_TARGET_ERROR"
			writeResourceError(c, http.StatusBadGateway, "BadGateway", "failed to resolve BYOC Top SQL target", nil)
		}
		return
	}

	var upstreamError *service.BYOCTopSQLUpstreamError
	if stderrors.As(err, &upstreamError) {
		switch upstreamError.StatusCode {
		case http.StatusBadRequest:
			rec.ErrCode = "INVALID_TOPSQL_PARAMETERS"
			writeResourceError(c, http.StatusBadRequest, "InvalidArgument", "invalid BYOC Top SQL request", nil)
		case http.StatusNotFound:
			rec.ErrCode = "TOPSQL_NOT_FOUND"
			writeResourceError(c, http.StatusNotFound, "NotFound", "BYOC Top SQL data is not available for this cluster", nil)
		case http.StatusTooManyRequests:
			rec.ErrCode = "TOPSQL_RATE_LIMITED"
			writeResourceError(c, http.StatusTooManyRequests, "ResourceExhausted", "BYOC Top SQL rate limit exceeded", nil)
		case http.StatusRequestTimeout, http.StatusGatewayTimeout:
			rec.ErrCode = "TOPSQL_TIMEOUT"
			writeResourceError(c, http.StatusGatewayTimeout, "DeadlineExceeded", "BYOC Top SQL request timed out", nil)
		case http.StatusServiceUnavailable:
			rec.ErrCode = "TOPSQL_UNAVAILABLE"
			writeResourceError(c, http.StatusServiceUnavailable, "Unavailable", "BYOC Top SQL service is temporarily unavailable", nil)
		default:
			rec.ErrCode = "TOPSQL_UPSTREAM_ERROR"
			writeResourceError(c, http.StatusBadGateway, "BadGateway", "BYOC Top SQL upstream request failed", nil)
		}
		return
	}

	rec.ErrCode = "TOPSQL_INTERNAL_ERROR"
	log.Error(c.Request.Context(), "BYOC Top SQL request failed",
		gozap.String("cluster_id", clusterID), gozap.String("error_type", fmt.Sprintf("%T", err)))
	writeResourceError(c, http.StatusBadGateway, "BadGateway", "BYOC Top SQL request failed", nil)
}

// Keep the canonical OpenAPI error type reachable by the annotation parser.
var _ apierrors.APIError