package handlers

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

	"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"
	"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 (
	maxBYOCContinuousProfilingQueryBytes       = 16 << 10
	maxBYOCContinuousProfilingTimestamp        = 253402300799
	maxBYOCContinuousProfilingTimeRangeSeconds = 3 * 60 * 60
)

var (
	byocContinuousProfilingPositiveInteger = regexp.MustCompile(`^[0-9]+$`)
	byocContinuousProfilingListKeys        = map[string]struct{}{
		"begin_time": {},
		"end_time":   {},
	}
	byocContinuousProfilingRoutingHeaders = map[string]struct{}{
		"xenv": {}, "xbiztype": {}, "xregionid": {}, "xapplicationid": {},
		"xdeploytype": {}, "xdeploytypev2": {}, "xo11yclusterid": {},
	}
)

type byocContinuousProfilingQuerier interface {
	Query(
		context.Context,
		uint64,
		string,
		service.BYOCContinuousProfilingOperation,
		url.Values,
	) (*service.BYOCContinuousProfilingResult, error)
}

// BYOCContinuousProfilingHandler exposes the read-only Continuous Profiling
// inventory stored in a physical byoc-host target. It does not expose profile
// tokens, bytes, downloads, views, config, object storage, or warehouse paths.
type BYOCContinuousProfilingHandler struct {
	service byocContinuousProfilingQuerier
}

type byocContinuousProfilingGroupProfile struct {
	TS                  int64          `json:"ts"`
	ProfileDurationSecs int            `json:"profile_duration_secs"`
	State               string         `json:"state"`
	ComponentNum        map[string]int `json:"component_num"`
}

type byocContinuousProfilingGroupProfileDetail struct {
	TS                  int64                                           `json:"ts"`
	ProfileDurationSecs int                                             `json:"profile_duration_secs"`
	State               string                                          `json:"state"`
	TargetProfiles      []byocContinuousProfilingGroupProfileDetailItem `json:"target_profiles"`
}

type byocContinuousProfilingGroupProfileDetailItem struct {
	State       string                                          `json:"state"`
	Error       string                                          `json:"error"`
	ProfileType string                                          `json:"profile_type"`
	Target      byocContinuousProfilingGroupProfileDetailTarget `json:"target"`
}

type byocContinuousProfilingGroupProfileDetailTarget struct {
	Address   string `json:"address"`
	Component string `json:"component"`
}

func NewBYOCContinuousProfilingHandler(svc byocContinuousProfilingQuerier) *BYOCContinuousProfilingHandler {
	return &BYOCContinuousProfilingHandler{service: svc}
}

// GroupProfiles lists Continuous Profiling groups from the physical BYOC O11y target.
//
//	@Summary		List physical BYOC Continuous Profiling groups
//	@ID			canonical-byoc-continuous-profiling-group-profiles
//	@Tags			data-proxy-continuous-profiling
//	@Produce		json
//	@Param			orgId		path		string	true	"TiDB Cloud/DBAAS org ID"
//	@Param			clusterId	path		string	true	"Physical byoc-host cluster ID"
//	@Param			begin_time	query		int		true	"Inclusive window start in positive Unix seconds; O11y rounds down to the minute"	minimum(1)	maximum(253402300799)
//	@Param			end_time	query		int		true	"Inclusive window end in positive Unix seconds; minute-rounded span is at most 3 hours"	minimum(1)	maximum(253402300799)
//	@Success		200			{array}		byocContinuousProfilingGroupProfile
//	@Failure		400			{object}	errors.APIError
//	@Failure		401			{object}	errors.APIError
//	@Failure		403			{object}	errors.APIError
//	@Failure		404			{object}	errors.APIError
//	@Failure		413			{object}	errors.APIError
//	@Failure		429			{object}	errors.APIError
//	@Failure		502			{object}	errors.APIError
//	@Failure		503			{object}	errors.APIError
//	@Failure		504			{object}	errors.APIError
//	@Security		BearerAuth
//	@Router			/data-proxy/api/v1/orgs/{orgId}/clusters/{clusterId}/continuous-profiling/group-profiles [get]
func (h *BYOCContinuousProfilingHandler) GroupProfiles(c *gin.Context) {
	h.handle(
		c,
		"canonical_byoc_continuous_profiling_group_profiles",
		service.BYOCContinuousProfilingOperationGroupProfiles,
		canonicalBYOCContinuousProfilingList,
	)
}

// GroupProfileDetail returns one physical BYOC Continuous Profiling group.
//
//	@Summary		Get a physical BYOC Continuous Profiling group
//	@Description	Returns JSON null with HTTP 200 when O11y has no group for the requested timestamp.
//	@ID			canonical-byoc-continuous-profiling-group-profile-detail
//	@Tags			data-proxy-continuous-profiling
//	@Produce		json
//	@Param			orgId		path		string	true	"TiDB Cloud/DBAAS org ID"
//	@Param			clusterId	path		string	true	"Physical byoc-host cluster ID"
//	@Param			ts			path		int		true	"Profiling group timestamp in positive Unix seconds"	minimum(1)	maximum(253402300799)
//	@Success		200			{object}	byocContinuousProfilingGroupProfileDetail	"Group detail, or JSON null when no group exists"
//	@Failure		400			{object}	errors.APIError
//	@Failure		401			{object}	errors.APIError
//	@Failure		403			{object}	errors.APIError
//	@Failure		404			{object}	errors.APIError
//	@Failure		413			{object}	errors.APIError
//	@Failure		429			{object}	errors.APIError
//	@Failure		502			{object}	errors.APIError
//	@Failure		503			{object}	errors.APIError
//	@Failure		504			{object}	errors.APIError
//	@Security		BearerAuth
//	@Router			/data-proxy/api/v1/orgs/{orgId}/clusters/{clusterId}/continuous-profiling/group-profiles/{ts} [get]
func (h *BYOCContinuousProfilingHandler) GroupProfileDetail(c *gin.Context) {
	h.handle(
		c,
		"canonical_byoc_continuous_profiling_group_profile_detail",
		service.BYOCContinuousProfilingOperationGroupProfileDetail,
		canonicalBYOCContinuousProfilingDetail,
	)
}

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

func (h *BYOCContinuousProfilingHandler) handle(
	c *gin.Context,
	endpoint string,
	operation service.BYOCContinuousProfilingOperation,
	parseParams byocContinuousProfilingParamsParser,
) {
	start, rec := beginCanonicalDataProxyRequest(c, endpoint)
	rec.QueryType = "continuous_profiling"
	rec.Engine = "o11y"
	defer func() { observability.RecordDataProxy(c, start, rec) }()

	if len(c.Request.URL.RawQuery) > maxBYOCContinuousProfilingQueryBytes {
		rec.ErrCode = "INVALID_CONTINUOUS_PROFILING_PARAMETERS"
		writeResourceError(c, http.StatusRequestEntityTooLarge, "PayloadTooLarge", "query parameters exceed the allowed size", nil)
		return
	}
	if !requireNoCanonicalBody(c, rec) ||
		!rejectCanonicalPlacement(c, rec) ||
		!rejectBYOCContinuousProfilingRoutingHeaders(c, rec) {
		return
	}
	orgID, clusterID, ok := parseCanonicalTelemetryScope(c, rec)
	if !ok {
		return
	}
	query, err := parseParams(c)
	if err != nil {
		rec.ErrCode = "INVALID_CONTINUOUS_PROFILING_PARAMETERS"
		writeResourceError(c, http.StatusBadRequest, "InvalidArgument", err.Error(), nil)
		return
	}
	if h == nil || h.service == nil {
		rec.ErrCode = "CONTINUOUS_PROFILING_UNAVAILABLE"
		writeResourceError(c, http.StatusServiceUnavailable, "Unavailable", "BYOC Continuous Profiling 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 = "CONTINUOUS_PROFILING_INVALID_RESPONSE"
		writeResourceError(c, http.StatusBadGateway, "BadGateway", "BYOC Continuous Profiling 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 (h *BYOCContinuousProfilingHandler) RegisterRoutes(r *gin.Engine) {
	group := r.Group("/data-proxy/api/v1/orgs/:orgId/clusters/:clusterId/continuous-profiling")
	group.GET("/group-profiles", h.GroupProfiles)
	group.GET("/group-profiles/:ts", h.GroupProfileDetail)
}

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

func canonicalBYOCContinuousProfilingList(c *gin.Context) (url.Values, error) {
	values, err := parseBYOCContinuousProfilingValues(c, byocContinuousProfilingListKeys)
	if err != nil {
		return nil, err
	}
	begin, err := canonicalBYOCContinuousProfilingTimestamp("begin_time", values["begin_time"])
	if err != nil {
		return nil, err
	}
	end, err := canonicalBYOCContinuousProfilingTimestamp("end_time", values["end_time"])
	if err != nil {
		return nil, err
	}
	if end < begin {
		return nil, fmt.Errorf("end_time must be greater than or equal to begin_time")
	}
	beginMinute := begin - begin%60
	endMinute := end - end%60
	if endMinute-beginMinute > maxBYOCContinuousProfilingTimeRangeSeconds {
		return nil, fmt.Errorf("minute-rounded time range must not exceed %d seconds", maxBYOCContinuousProfilingTimeRangeSeconds)
	}
	// Preserve the caller's validated spelling because the established O11y
	// contract owns the minute-floor operation.
	return cloneURLValues(values), nil
}

func canonicalBYOCContinuousProfilingDetail(c *gin.Context) (url.Values, error) {
	if c.Request.URL.RawQuery != "" {
		return nil, fmt.Errorf("query parameters are not supported")
	}
	tsValues := []string{c.Param("ts")}
	if _, err := canonicalBYOCContinuousProfilingTimestamp("ts", tsValues); err != nil {
		return nil, err
	}
	return url.Values{"ts": tsValues}, nil
}

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

func canonicalBYOCContinuousProfilingTimestamp(name string, values []string) (int64, error) {
	if len(values) != 1 {
		return 0, fmt.Errorf("%s must be provided exactly once", name)
	}
	raw := values[0]
	if !byocContinuousProfilingPositiveInteger.MatchString(raw) {
		return 0, fmt.Errorf("%s must be a positive integer Unix timestamp", name)
	}
	value, err := strconv.ParseInt(raw, 10, 64)
	if err != nil || value <= 0 || value > maxBYOCContinuousProfilingTimestamp {
		return 0, fmt.Errorf("%s must be no greater than %d", name, int64(maxBYOCContinuousProfilingTimestamp))
	}
	return value, nil
}

func (h *BYOCContinuousProfilingHandler) writeError(
	c *gin.Context,
	rec *observability.DataProxyRecord,
	err error,
	clusterID string,
) {
	switch {
	case stderrors.Is(err, context.Canceled):
		rec.ErrCode = "CONTINUOUS_PROFILING_CANCELED"
		writeResourceError(c, clientClosedStatus, "Canceled", "BYOC Continuous Profiling request was canceled", nil)
		return
	case stderrors.Is(err, context.DeadlineExceeded):
		rec.ErrCode = "CONTINUOUS_PROFILING_TIMEOUT"
		writeResourceError(c, http.StatusGatewayTimeout, "DeadlineExceeded", "BYOC Continuous Profiling request timed out", nil)
		return
	case stderrors.Is(err, service.ErrBYOCContinuousProfilingUnsupported):
		rec.ErrCode = "CONTINUOUS_PROFILING_UNSUPPORTED"
		writeResourceError(c, http.StatusBadRequest, "InvalidArgument", "Continuous Profiling is supported only for physical byoc-host clusters", nil)
		return
	case stderrors.Is(err, service.ErrBYOCContinuousProfilingUnavailable):
		rec.ErrCode = "CONTINUOUS_PROFILING_UNAVAILABLE"
		writeResourceError(c, http.StatusServiceUnavailable, "Unavailable", "BYOC Continuous Profiling service is temporarily unavailable", nil)
		return
	case stderrors.Is(err, service.ErrInvalidBYOCContinuousProfilingTarget),
		stderrors.Is(err, service.ErrInvalidBYOCContinuousProfilingResponse):
		rec.ErrCode = "CONTINUOUS_PROFILING_INVALID_UPSTREAM"
		writeResourceError(c, http.StatusBadGateway, "BadGateway", "BYOC Continuous Profiling returned an invalid response", nil)
		return
	}

	var networkError net.Error
	if stderrors.As(err, &networkError) {
		if networkError.Timeout() {
			rec.ErrCode = "CONTINUOUS_PROFILING_TIMEOUT"
			writeResourceError(c, http.StatusGatewayTimeout, "DeadlineExceeded", "BYOC Continuous Profiling request timed out", nil)
			return
		}
		rec.ErrCode = "CONTINUOUS_PROFILING_UNAVAILABLE"
		writeResourceError(c, http.StatusServiceUnavailable, "Unavailable", "BYOC Continuous Profiling 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 = "CONTINUOUS_PROFILING_UNSUPPORTED"
			writeResourceError(c, http.StatusBadRequest, "InvalidArgument", "Continuous Profiling is supported only for physical byoc-host clusters", 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 = "CONTINUOUS_PROFILING_TARGET_TIMEOUT"
			writeResourceError(c, http.StatusGatewayTimeout, "DeadlineExceeded", "BYOC Continuous Profiling target resolution timed out", nil)
		case http.StatusTooManyRequests:
			rec.ErrCode = "CONTINUOUS_PROFILING_TARGET_RATE_LIMITED"
			writeResourceError(c, http.StatusTooManyRequests, "ResourceExhausted", "BYOC Continuous Profiling target rate limit exceeded", nil)
		case http.StatusServiceUnavailable:
			rec.ErrCode = "CONTINUOUS_PROFILING_TARGET_UNAVAILABLE"
			writeResourceError(c, http.StatusServiceUnavailable, "Unavailable", "BYOC Continuous Profiling target is temporarily unavailable", nil)
		default:
			rec.ErrCode = "CONTINUOUS_PROFILING_TARGET_ERROR"
			writeResourceError(c, http.StatusBadGateway, "BadGateway", "failed to resolve BYOC Continuous Profiling target", nil)
		}
		return
	}

	var upstreamError *service.BYOCContinuousProfilingUpstreamError
	if stderrors.As(err, &upstreamError) {
		switch upstreamError.StatusCode {
		case http.StatusBadRequest:
			rec.ErrCode = "INVALID_CONTINUOUS_PROFILING_PARAMETERS"
			writeResourceError(c, http.StatusBadRequest, "InvalidArgument", "invalid BYOC Continuous Profiling request", nil)
		case http.StatusNotFound:
			rec.ErrCode = "CONTINUOUS_PROFILING_NOT_FOUND"
			writeResourceError(c, http.StatusNotFound, "NotFound", "BYOC Continuous Profiling data is not available for this cluster", nil)
		case http.StatusTooManyRequests:
			rec.ErrCode = "CONTINUOUS_PROFILING_RATE_LIMITED"
			writeResourceError(c, http.StatusTooManyRequests, "ResourceExhausted", "BYOC Continuous Profiling rate limit exceeded", nil)
		case http.StatusRequestTimeout, http.StatusGatewayTimeout:
			rec.ErrCode = "CONTINUOUS_PROFILING_TIMEOUT"
			writeResourceError(c, http.StatusGatewayTimeout, "DeadlineExceeded", "BYOC Continuous Profiling request timed out", nil)
		case http.StatusServiceUnavailable:
			rec.ErrCode = "CONTINUOUS_PROFILING_UNAVAILABLE"
			writeResourceError(c, http.StatusServiceUnavailable, "Unavailable", "BYOC Continuous Profiling service is temporarily unavailable", nil)
		default:
			rec.ErrCode = "CONTINUOUS_PROFILING_UPSTREAM_ERROR"
			writeResourceError(c, http.StatusBadGateway, "BadGateway", "BYOC Continuous Profiling upstream request failed", nil)
		}
		return
	}

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

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