package proxy

import (
	"bytes"
	"compress/gzip"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"strconv"
	"strings"

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

	"github.com/pingcap-inc/clinic/pkg/api/common"
	error2 "github.com/pingcap-inc/clinic/pkg/api/error"
	"github.com/pingcap-inc/clinic/pkg/log"
	"github.com/pingcap-inc/clinic/pkg/o11y"
	o11yurls "github.com/pingcap-inc/clinic/pkg/o11y/urls"
	rbac "github.com/pingcap-inc/clinic/pkg/rbac"
)

// CloudNodeMetricHandler check which metrics backend to proxy. <vm|thanos|o11y>
type CloudNodeMetricHandler struct {
	rbac    rbac.Auth
	o11yCli *o11y.Client
	// reuse o11y cache
	O11yProxyHandler *O11yProxyHandler
	env              string
}

func NewCloudNodeMetricHandler(rbac rbac.Auth, O11yProxyHandler *O11yProxyHandler, env string) *CloudNodeMetricHandler {
	return &CloudNodeMetricHandler{
		rbac:             rbac,
		O11yProxyHandler: O11yProxyHandler,
		env:              env,
	}
}

func (handler *CloudNodeMetricHandler) SetO11y(o11yCli *o11y.Client) {
	handler.o11yCli = o11yCli
}

type PrometheusResponse struct {
	Status string `json:"status"`
	Data   struct {
		ResultType string       `json:"resultType"`
		Result     []ResultItem `json:"result"`
	} `json:"data"`
}

type ResultItem struct {
	Metric map[string]string `json:"metric"`
	Values []any             `json:"values,omitempty"`
}

type SeriesResponse struct {
	Status string              `json:"status"`
	Data   []map[string]string `json:"data"`
}

const (
	clinicCustomerPodMetric = "CLINIC_CUSTOMER_POD"
	customerPodAllowRegex   = "db-tikv.*|db-tidb.*|db-tiflash.*|.*-ticdc.*|db-tiproxy.*"
	customerPodExcludeRegex = "db-tidb-extra.*"
	noMatchNamespace        = "__clinic_no_match__"
)

type podLabelMatcher struct {
	label       string
	operator    string
	quotedValue string
	value       string
}

func rewriteCustomerPodMatch(match, clusterID string) (string, bool) {
	if match == clinicCustomerPodMetric {
		return fmt.Sprintf("kube_pod_info{namespace=~\".*%s\",pod=~\"%s\",pod!~\"%s\"}", clusterID, customerPodAllowRegex, customerPodExcludeRegex), true
	}
	if !strings.HasPrefix(match, clinicCustomerPodMetric+"{") || !strings.HasSuffix(match, "}") {
		return "", false
	}

	selector := strings.TrimSuffix(strings.TrimPrefix(match, clinicCustomerPodMetric+"{"), "}")
	matchers, ok := parsePodLabelMatchers(selector)
	if !ok {
		return "", false
	}

	var namespaceMatcher string
	var podMatcher string
	for _, matcher := range matchers {
		switch matcher.label {
		case "namespace":
			if matcher.operator != "=" && matcher.operator != "=~" {
				return "", false
			}
			if !strings.Contains(matcher.value, clusterID) {
				namespaceMatcher = fmt.Sprintf("namespace=\"%s\"", noMatchNamespace)
			} else {
				namespaceMatcher = fmt.Sprintf("namespace%s%s", matcher.operator, matcher.quotedValue)
			}
		case "pod":
			if matcher.operator != "=" && matcher.operator != "=~" {
				return "", false
			}
			podMatcher = fmt.Sprintf("pod%s%s", matcher.operator, matcher.quotedValue)
		default:
			return "", false
		}
	}

	if namespaceMatcher == "" {
		namespaceMatcher = fmt.Sprintf("namespace=~\".*%s\"", clusterID)
	}

	selectors := []string{
		namespaceMatcher,
		fmt.Sprintf("pod=~\"%s\"", customerPodAllowRegex),
		fmt.Sprintf("pod!~\"%s\"", customerPodExcludeRegex),
	}
	if podMatcher != "" {
		selectors = append(selectors, podMatcher)
	}

	return fmt.Sprintf("kube_pod_info{%s}", strings.Join(selectors, ",")), true
}

func parsePodLabelMatchers(selector string) ([]podLabelMatcher, bool) {
	parts, ok := splitPrometheusLabelMatchers(selector)
	if !ok {
		return nil, false
	}

	matchers := make([]podLabelMatcher, 0, len(parts))
	for _, part := range parts {
		matcher, ok := parsePodLabelMatcher(part)
		if !ok {
			return nil, false
		}
		matchers = append(matchers, matcher)
	}
	return matchers, true
}

func splitPrometheusLabelMatchers(selector string) ([]string, bool) {
	var parts []string
	start := 0
	inString := false
	escaped := false

	for i, ch := range selector {
		if inString {
			if escaped {
				escaped = false
				continue
			}
			if ch == '\\' {
				escaped = true
				continue
			}
			if ch == '"' {
				inString = false
			}
			continue
		}

		if ch == '"' {
			inString = true
			continue
		}
		if ch == ',' {
			parts = append(parts, selector[start:i])
			start = i + 1
		}
	}
	if inString {
		return nil, false
	}

	parts = append(parts, selector[start:])
	return parts, true
}

func parsePodLabelMatcher(part string) (podLabelMatcher, bool) {
	part = strings.TrimSpace(part)
	for _, operator := range []string{"=~", "!~", "!=", "="} {
		idx := strings.Index(part, operator)
		if idx < 0 {
			continue
		}
		label := strings.TrimSpace(part[:idx])
		quotedValue := strings.TrimSpace(part[idx+len(operator):])
		if label == "" || len(quotedValue) < 2 || quotedValue[0] != '"' || quotedValue[len(quotedValue)-1] != '"' {
			return podLabelMatcher{}, false
		}
		return podLabelMatcher{
			label:       label,
			operator:    operator,
			quotedValue: quotedValue,
			value:       quotedValue[1 : len(quotedValue)-1],
		}, true
	}

	return podLabelMatcher{}, false
}

func (handler *CloudNodeMetricHandler) GinHandle(
	c *gin.Context,
	getClusterID func(c *gin.Context) string,
	routerPathFunc func(string) string,
) {
	allowedMetrics := []string{
		// Disk iops
		"node_disk_reads_completed_total",
		"node_disk_writes_completed_total",
		// Disk I/O Util
		"node_disk_io_time_seconds_total",
		// Disk Bandwidth
		"node_disk_read_bytes_total",
		"node_disk_written_bytes_total",
		// Disk wait latency
		"node_disk_read_time_seconds_total",
		"node_disk_reads_completed_total",
		"node_disk_write_time_seconds_total",
		"node_disk_writes_completed_total",
	}
	allowedLabels := []string{
		"__name__",
		"tenant",
		"instance",
		"device",
		"node",
		"pod",
		"namespace",
		"k8s_cluster_info",
	}

	// directly return for cluster-unrelated metrics
	switch c.Request.URL.Path {
	case "/metric/node/query/api/v1/label/__name__/values":
		c.JSON(http.StatusOK, map[string]any{
			"status": "success",
			"data":   allowedMetrics,
		})
		return
	case "/metric/node/query/api/v1/metadata":
		c.JSON((http.StatusOK), map[string]any{
			"status": "success",
			"data":   map[string]any{},
		})
		return
	case "/metric/node/query/api/v1/labels":
		c.JSON(http.StatusOK, map[string]any{
			"status": "success",
			"data":   allowedLabels,
		})
		return
	}

	// get clusterID from cookie, must be used as metric query check and permission check
	clusterID := getClusterID(c)
	if clusterID == "" {
		_ = c.Error(error2.ErrNotFound.New("cluster is not found"))
		return
	}

	// include permission check
	_, cluster, ok, err := common.CanReadCloudCluster(c, handler.rbac, clusterID)
	if err != nil {
		_ = c.Error(error2.ErrNotFound.New("%s", err.Error()))
		return
	}
	if !ok {
		_ = c.Error(error2.ErrForbidden.New("access cluster failed"))
		return
	}
	vendor := cluster.Provider
	region := cluster.Region

	/*	try to map clusterID to k8s_cluster, but not work since it is empty in 3layer. Field value has been moved to separate api
		//  clusterID -> k8s_cluster
		projectID := cluster.ProjectID
		k8scluster := cluster.DataPlaneName
		log.Info("get cluster",
			zap.String("project_id", projectID),
			zap.String("k8s_cluster", k8scluster),
		)
	*/

	ctx := c.Request.Context()
	log.Info(ctx, "node metrics request validated",
		zap.String("operation", "proxy_node_metrics"),
		zap.String("cluster_id", clusterID),
		zap.String("deploy_type", cluster.DeployType),
		zap.String("deploy_type_v2", cluster.DeployTypeV2),
		zap.String("path", c.Request.URL.Path),
		zap.String("method", c.Request.Method),
	)

	if cluster.Provider == "aws" && cluster.Region == "us-west-2" && cluster.DeployType == o11y.DedicatedType && handler.env != "prod" && cluster.CreatedAt.After(o11yUswest2SwitchStagingDate) {
		region += "-staging"
	}
	target, err := o11yurls.NewClusterURLsResolver(handler.o11yCli).ResolveNodeMetricsTarget(ctx, cluster, region)
	if err != nil {
		log.Error(ctx, "failed to resolve node metrics target",
			zap.String("operation", "resolve_node_metrics_target"),
			zap.String("cluster_id", clusterID),
			zap.String("provider", vendor),
			zap.String("region", region),
			zap.Error(err))
		_ = c.Error(error2.ErrInternal.New("get node metrics target failed"))
		return
	}
	log.Info(ctx, "resolved node metrics target",
		zap.String("operation", "resolve_node_metrics_target"),
		zap.String("cluster_id", clusterID),
		zap.Bool("uses_internal_read_url", target.InternalReadURL != ""))

	var respModifyFunc func(resp *http.Response) error

	switch c.Request.URL.Path {
	case "/metric/node/query/api/v1/series":
		r := c.Request
		queries := r.URL.Query()
		match := queries.Get("match[]")
		if match == "CLINIC_CUSTOMER_K8S" {
			queries.Set("match[]", fmt.Sprintf("kube_node_labels{label_cluster=\"%s\"}", clusterID))
			r.URL.RawQuery = queries.Encode()
		} else if rewritten, ok := rewriteCustomerPodMatch(match, clusterID); ok {
			queries.Set("match[]", rewritten)
			r.URL.RawQuery = queries.Encode()
		} else if strings.HasPrefix(match, "CLINIC_CUSTOMER_INSTANCE_") {
			pod, _ := strings.CutPrefix(match, "CLINIC_CUSTOMER_INSTANCE_")
			queries.Set("match[]", fmt.Sprintf("kube_pod_info{namespace=~\".*%s\",pod=~\"%s\"}", clusterID, pod))
			r.URL.RawQuery = queries.Encode()
		} else {
			// do not allow free query
			log.Warn(ctx, "invalid node metrics series selector",
				zap.String("operation", "validate_node_metrics_selector"),
				zap.String("cluster_id", clusterID))
			_ = c.Error(error2.ErrBadRequest.New("invalid match[]"))
			return
		}

		respModifyFunc = func(resp *http.Response) error {
			resp.Header.Set("X-Metrics-From", "o11y")

			r := resp.Body
			defer resp.Body.Close()
			if resp.Header.Get("Content-Encoding") == "gzip" {
				r, err = gzip.NewReader(resp.Body)
				if err != nil {
					log.Error(resp.Request.Context(), "failed to create node metrics gzip reader", zap.Error(err))
					return err
				}
				defer r.Close()
			}

			body, err := io.ReadAll(r)
			if err != nil {
				log.Error(resp.Request.Context(), "failed to read node metrics response body", zap.Error(err))
				return err
			}

			var promResp SeriesResponse
			if err := json.Unmarshal(body, &promResp); err != nil {
				log.Error(resp.Request.Context(), "failed to parse node metrics series response", zap.Error(err))
				return fmt.Errorf("failed to parse prometheus response")
			}

			// Filter labels
			for i := range promResp.Data {
				labels := promResp.Data[i]
				filtered := make(map[string]string)

				// Only keep cluster_id and instance labels
				if v, ok := labels["pod"]; ok {
					filtered["pod"] = v
				}
				if v, ok := labels["namespace"]; ok {
					filtered["namespace"] = v
				}
				if v, ok := labels["instance"]; ok {
					filtered["instance"] = v
				}
				if v, ok := labels["device"]; ok {
					filtered["device"] = v
				}
				if v, ok := labels["k8s_cluster_info"]; ok {
					filtered["k8s_cluster_info"] = v
				}
				if v, ok := labels["node"]; ok {
					filtered["node"] = v
				}

				promResp.Data[i] = filtered
			}

			updated, err := json.Marshal(promResp)
			if err != nil {
				log.Error(resp.Request.Context(), "failed to marshal node metrics series response", zap.Error(err))
				return fmt.Errorf("failed to marshal prometheus response")
			}
			resp.Body = io.NopCloser(bytes.NewBuffer(updated))
			resp.ContentLength = int64(len(updated))
			resp.Header.Del("Content-Encoding")
			resp.Header.Set("Content-Length", fmt.Sprintf("%d", len(updated)))
			log.Info(resp.Request.Context(), "node metrics upstream request completed",
				zap.String("operation", "proxy_node_metrics_series"),
				zap.Int("status", resp.StatusCode),
				zap.Int("result_count", len(promResp.Data)))
			return nil
		}

	case "/metric/node/query/api/v1/query_range":
		if c.Request.Body != nil {
			// read body and log
			rawBody, err := io.ReadAll(c.Request.Body)
			if err != nil {
				log.Error(ctx, "failed to read node metrics request body", zap.Error(err))
				_ = c.Error(error2.ErrBadRequest.New("failed to read request body"))
				return
			}
			values, err := url.ParseQuery(string(rawBody))
			if err != nil {
				log.Error(ctx, "failed to parse node metrics request body", zap.Error(err))
				_ = c.Error(error2.ErrBadRequest.New("failed to parse request body"))
				return
			}
			log.Debug(ctx, "parsed node metrics query range request",
				zap.String("operation", "validate_node_metrics_query_range"),
				zap.Int("body_size", len(rawBody)),
				zap.Bool("has_query", values.Get("query") != ""))
			c.Request.Body = io.NopCloser(bytes.NewBuffer(rawBody))
		}
		respModifyFunc = func(resp *http.Response) error {
			resp.Header.Set("X-Metrics-From", "o11y")

			r := resp.Body
			defer resp.Body.Close()
			if resp.Header.Get("Content-Encoding") == "gzip" {
				r, err = gzip.NewReader(resp.Body)
				if err != nil {
					log.Error(resp.Request.Context(), "failed to create node metrics gzip reader", zap.Error(err))
					return err
				}
				defer r.Close()
			}

			body, err := io.ReadAll(r)
			if err != nil {
				log.Error(resp.Request.Context(), "failed to read node metrics response body", zap.Error(err))
				return err
			}

			var promResp PrometheusResponse
			if err := json.Unmarshal(body, &promResp); err != nil {
				log.Error(resp.Request.Context(), "failed to parse node metrics query range response", zap.Error(err))
				return fmt.Errorf("failed to parse prometheus query_range response")
			}

			// Filter labels
			result := []ResultItem{}
			for i := range promResp.Data.Result {
				metric := promResp.Data.Result[i].Metric
				filtered := make(map[string]string)
				for _, label := range allowedLabels {
					if v, ok := metric[label]; ok {
						filtered[label] = v
					}
				}

				if filtered["tenant"] == strconv.FormatUint(cluster.TenantID, 10) {
					result = append(result, ResultItem{
						Metric: filtered,
						Values: promResp.Data.Result[i].Values,
					})
				}
			}
			promResp.Data.Result = result

			updated, err := json.Marshal(promResp)
			if err != nil {
				log.Error(resp.Request.Context(), "failed to marshal node metrics query range response", zap.Error(err))
				return fmt.Errorf("failed to marshal query_range prometheus response")
			}
			resp.Body = io.NopCloser(bytes.NewBuffer(updated))
			resp.ContentLength = int64(len(updated))
			resp.Header.Del("Content-Encoding")
			resp.Header.Set("Content-Length", fmt.Sprintf("%d", len(updated)))
			log.Info(resp.Request.Context(), "node metrics upstream request completed",
				zap.String("operation", "proxy_node_metrics_query_range"),
				zap.Int("status", resp.StatusCode),
				zap.Int("result_count", len(promResp.Data.Result)))
			return nil
		}
	default:
		log.Warn(ctx, "invalid node metrics path", zap.String("path", c.Request.URL.Path))
		_ = c.Error(error2.ErrBadRequest.New("invalid path"))
		return
	}

	if target.InternalReadURL != "" {
		internalReadURL, parseErr := url.Parse(target.InternalReadURL)
		if parseErr != nil || internalReadURL.Host == "" {
			log.Error(ctx, "invalid node metrics internal read URL",
				zap.String("operation", "proxy_node_metrics"),
				zap.String("cluster_id", clusterID),
				zap.Error(parseErr))
			_ = c.Error(error2.ErrInternal.New("invalid o11y node metrics target"))
			return
		}
		scheme := internalReadURL.Scheme
		if scheme == "" {
			scheme = "https"
		}
		requestModifyFunc := buildModifyFuncByInternalReadURI(internalReadURL.Host, internalReadURL.Path, internalReadURL.RawPath, routerPathFunc)
		requestModifyFunc = withO11yClusterRequestHeaders(requestModifyFunc, target.O11yCluster)
		proxyModifyResponse(ctx, c, scheme, internalReadURL.Host, requestModifyFunc, respModifyFunc)
		return
	}

	targetHost := handler.o11yCli.GenDataServiceProxyHost(region, vendor)
	proxyModifyResponse(ctx, c, "https", targetHost, buildModifyFuncByO11yID(handler.o11yCli, targetHost, target.O11yCluster.ID, routerPathFunc, false), respModifyFunc)
}