package diagnosis

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/http/httputil"
	"net/url"
	"path"
	"strconv"
	"strings"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/prometheus/common/model"
	"go.uber.org/zap"

	apierr "github.com/pingcap-inc/clinic/pkg/api/error"
	"github.com/pingcap-inc/clinic/pkg/config"
)

var EdgeMatrixV2 = map[int64][]int64{
	// Write too slow -> Some instances write too slow (11271)
	8637: {11271},
	// Some instances write too slow -> Check write stall (9875)
	11271: {9875},
	// Check write stall -> Check Total compaction flow (9100)
	9875: {9100}, // Check Total compaction flow -> Check RocksDB compaction flow (8945, 8946, 9876, 11281, 11282)
	9100: {8945, 8946, 9876, 11281, 11282},
	// Check RocksDB compaction flow -> Check RocksDB write latency (10486, 11258)
	8945:  {10486, 11258},
	8946:  {10486, 11258},
	9876:  {10486, 11258},
	11281: {10486, 11258},
	11282: {10486, 11258},
	// Check RocksDB write latency -> Check RocksDB WAL latency (9102, 11285)
	// Check RocksDB write latency -> Check the Disk Write latency (8025)
	10486: {9102, 11285, 8025},
	11258: {9102, 11285, 8025},
	// Check the Disk Write latency -> Check write batch size (11261, 11262)
	// Check the Disk Write latency -> Check the RocksDB CPU usage (11270)
	// Check the Disk Write latency -> Check the Frontend flow (9099, 9101)
	8025: {11261, 11262, 11270, 9099, 9101},
	//  Check RocksDB WAL latency -> Check out Async Write
	9102:  {11284},
	11285: {11284},
	// Check out Async Write -> Check out RaftStore Threads (9407, 9408)
	// Check out Async Write -> Check out latch
	11284: {9407, 9408, 11008},
	// Check out RaftStore Threads -> Check out Wait for RaftStore Threads (9407, 9408) not defined in ppt
	// Check out latch -> Check out Scheduler Threads
	11008: {9255, 8947},
	// Check out Scheduler Threads -> Check Perf Context Mutex
	// Check Perf Context Mutex -> Check Perf Context Thread wait
	9255: {11263, 9571},
	8947: {11263, 9571},
	// Check Perf Context Mutex / Check Perf Context Thread wait -> Check PD Scheduling
	11263: {11276},
	9571:  {11276},
	// Check write batch size -> Check Perf Context Mutex
	11261: {11263, 9571},
	11262: {11263, 9571},

	// Read too slow -> Some instances read too slow (11272)
	9254: {11272},
	// Some instances read too slow -> Get too slow (11278)
	// Some instances read too slow -> Coprocessor too slow (11260)
	11272: {11278, 11260},
	// Coprocessor too slow -> Coprocessor handle too slow (10334)
	// Coprocessor too slow -> Coprocessor-RPC QPS Follow Write-RPC? (11279)
	11260: {10334, 11279},
	// Coprocessor handle too slow -> Check coprocessor threads
	10334: {9563, 10790},
	// Check coprocessor threads -> Check scanned data count
	9563:  {9561},
	10790: {9561},
	// Get too slow -> Check scanned data count (9561)
	// Get too slow -> BatchGet-RPC & Get-RPC QPS Follow Write-RPC? (10638) 11278: {9561, 10638},
	// Check scanned data count -> Check RPC count
	// Check scanned data count -> Check scanned Rocksdb tombstone count
	9561: {10942, 10182},
	// Check scanned Rocksdb tombstone count -> Check KVDB Seek and Get latency (9567, 9568, 10030)
	10182: {9567, 9568, 10030},
	// Check KVDB Seek and Get latency -> Check in-lease-read rate
	// Check KVDB Seek and Get latency -> Check memtable hit count and block-cache hit rate
	9567:  {11259, 11287, 9570},
	9568:  {11259, 11287, 9570},
	10030: {11259, 11287, 9570},
	// Check in-lease-read rate -> Check async-snap
	11259: {11286},
	// Check async-snap -> Check PD leader scheduling
	11286: {11276},
	// Check memtable hit count and block-cache hit rate -> Check SST read count
	11287: {9566},
	9570:  {9566},
	// Check SST read count -> Check SST read latency
	9566: {9569},
	// Check SST read latency -> Check Disk read latency
}

// struct to define vmstorage, vminsert and vmselect, vmstorage equals to native prometheus
type vmServiceType int

//nolint:varcheck
const (
	vmStorage vmServiceType = iota
	vmSelect
	vmInsert
)

type vmEndpoint struct {
	Host        string
	Port        uint16
	ServiceType vmServiceType
	AccountID   int
}

func (ep *vmEndpoint) Authority() string {
	if ep.Port == 0 {
		return ep.Host
	}
	return fmt.Sprintf("%s:%d", ep.Host, ep.Port)
}

// https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html#url-format
func (ep *vmEndpoint) PromQueryPath(suffix string) (string, error) {
	switch ep.ServiceType {
	case vmInsert:
		return "", errors.New("vminsert not support prometheus query api")
	case vmSelect:
		return path.Join(fmt.Sprintf("/select/%d/prometheus", ep.AccountID), suffix), nil
	default:
		return suffix, nil
	}
}

// https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html#url-format
func (ep *vmEndpoint) InsertPath(suffix string) (string, error) {
	switch ep.ServiceType {
	case vmInsert:
		return path.Join(fmt.Sprintf("/insert/%d", ep.AccountID), suffix), nil
	case vmSelect:
		return "", errors.New("vmselect not support insert api")
	default:
		return suffix, nil
	}
}

// func InsertMetricsSampleGinHandlerFunc(api *DiagnosisAPI) gin.HandlerFunc {
// 	return func(c *gin.Context) {
// 		param := &InsertMetricsSampleParam{}
// 		if err := c.ShouldBindJSON(param); err != nil {
// 			_ = c.Error(apierr.ErrBadRequest.WrapWithNoMessage(err))
// 			return
// 		}
// 		if err := param.Validate(); err != nil {
// 			_ = c.Error(apierr.ErrBadRequest.WrapWithNoMessage(err))
// 			return
// 		}
// 		log.Debug("InsertMetricsSample", zap.Any("param", param))
// 		data, err := api.InsertMetricsSample(c.Request.Context(), param)
// 		if err != nil {
// 			log.Error("insert sample failed", zap.Error(err))
// 			_ = c.Error(apierr.ErrInternal.WrapWithNoMessage(err))
// 			return
// 		}
// 		c.JSON(http.StatusNoContent, data)
// 	}
// }

func AnnotationGinHandlerFunc(api *DiagnosisAPI) gin.HandlerFunc {
	return func(c *gin.Context) {
		param := &QueryAnnotationsParam{}
		if ts, err := strconv.ParseInt(c.Query("start_ts"), 10, 64); err != nil {
			_ = c.Error(apierr.ErrBadRequest.WrapWithNoMessage(err))
			return
		} else {
			param.StartTS = ts
		}
		if ts, err := strconv.ParseInt(c.Query("end_ts"), 10, 64); err != nil {
			_ = c.Error(apierr.ErrBadRequest.WrapWithNoMessage(err))
			return
		} else {
			param.EndTS = ts
		}
		param.TiDBCluster = c.Query("tidb_cluster")
		param.Measurement = c.Query("measurement")
		if err := param.Validate(); err != nil {
			_ = c.Error(apierr.ErrBadRequest.WrapWithNoMessage(err))
			return
		}
		log.Debug("QueryAnnotations", zap.Any("param", param))
		data, err := api.QueryAnnotations(c.Request.Context(), param)
		if err != nil {
			log.Error("query annotations failed", zap.Error(err))
			_ = c.Error(apierr.ErrInternal.WrapWithNoMessage(err))
			return
		}
		c.JSON(http.StatusOK, data)
	}
}

func NodeGraphGinHandlerFunc(api *DiagnosisAPI) gin.HandlerFunc {
	return func(c *gin.Context) {
		param := &QueryNodeGraphParam{}
		if ts, err := strconv.ParseInt(c.Query("start_ts"), 10, 64); err != nil {
			_ = c.Error(apierr.ErrBadRequest.WrapWithNoMessage(err))
			return
		} else {
			param.StartTS = ts
		}
		if ts, err := strconv.ParseInt(c.Query("end_ts"), 10, 64); err != nil {
			_ = c.Error(apierr.ErrBadRequest.WrapWithNoMessage(err))
			return
		} else {
			param.EndTS = ts
		}
		param.TiDBCluster = c.Query("tidb_cluster")
		param.Measurement = c.Query("measurement")
		if err := param.Validate(); err != nil {
			_ = c.Error(apierr.ErrBadRequest.WrapWithNoMessage(err))
			return
		}
		log.Debug("QueryNodeGraph", zap.Any("param", param))
		data, err := api.QueryNodeGraph(c.Request.Context(), param)
		if err != nil {
			log.Error("query node graph failed", zap.Error(err))
			_ = c.Error(apierr.ErrInternal.WrapWithNoMessage(err))
			return
		}
		c.JSON(http.StatusOK, data)
	}
}

func DynamicTextValueGinHandlerFunc(api *DiagnosisAPI) gin.HandlerFunc {
	return func(c *gin.Context) {
		param := &QueryDynamicTextValueParam{}
		param.TiDBCluster = c.Query("tidb_cluster")
		param.StartTS, _ = strconv.ParseInt(c.Query("start_ts"), 10, 64)
		param.EndTS, _ = strconv.ParseInt(c.Query("end_ts"), 10, 64)
		param.Measurement = c.Query("measurement")
		param.Default1 = c.Query("default_1")

		if err := param.Validate(); err != nil {
			log.Error("param validate failed", zap.Error(err))
			_ = c.Error(apierr.ErrBadRequest.WrapWithNoMessage(err))
			return
		}
		log.Debug("QueryDynamicTextValue", zap.Any("param", param))
		data, err := api.QueryDynamicTextValue(c.Request.Context(), param)
		if err != nil {
			log.Error("query dynamic text value v2 failed", zap.Error(err))
			_ = c.Error(apierr.ErrInternal.WrapWithNoMessage(err))
			return
		}
		c.JSON(http.StatusOK, data)
	}
}

func InsertLogSampleGinHandlerFunc(api *DiagnosisAPI) gin.HandlerFunc {
	return func(c *gin.Context) {
		log.Debug("request arrived", zap.String("url", c.Request.URL.String()))
		api.logProxy.ServeHTTP(c.Writer, c.Request)
	}
}

func customizedTransport() http.RoundTripper {
	defaultTransPtr, ok := http.DefaultTransport.(*http.Transport)
	if !ok {
		return http.DefaultTransport
	}
	return &http.Transport{
		Proxy: http.ProxyFromEnvironment,
		DialContext: (&net.Dialer{
			Timeout:   3 * time.Second,
			KeepAlive: 30 * time.Second,
		}).DialContext,
		MaxIdleConns: 100,
		// increase conn count for better concurrency
		MaxIdleConnsPerHost:   10,
		ResponseHeaderTimeout: 30 * time.Second,
		IdleConnTimeout:       defaultTransPtr.IdleConnTimeout,
		TLSHandshakeTimeout:   defaultTransPtr.TLSHandshakeTimeout,
		ExpectContinueTimeout: defaultTransPtr.ExpectContinueTimeout,
	}
}

type DiagnosisAPI struct {
	vmInsertEndpoint vmEndpoint
	vmSelectEndpoint vmEndpoint
	httpCli          *http.Client
	logProxy         *reverseProxyWrapper
}

func NewDiagnosisAPI(metricConfig *config.MetricSingleEndpointConfig, logConfig *config.LogConfig) (*DiagnosisAPI, error) {
	logProxy, err := buildLogProxy(logConfig)
	if err != nil {
		return nil, err
	}
	api := &DiagnosisAPI{
		httpCli: &http.Client{
			Transport: customizedTransport(),
		},
		vmInsertEndpoint: vmEndpoint{
			Host:      metricConfig.Host,
			Port:      metricConfig.Port,
			AccountID: 1,
		},
		vmSelectEndpoint: vmEndpoint{
			Host:      metricConfig.SelectAddr.Host,
			Port:      metricConfig.SelectAddr.Port,
			AccountID: 1,
		},
		logProxy: logProxy,
	}
	if metricConfig.InVMCluster() {
		api.vmInsertEndpoint.ServiceType = vmInsert
	}
	if metricConfig.SelectAddr.InVMCluster() {
		api.vmSelectEndpoint.ServiceType = vmSelect
	}

	return api, nil
}

// func (api *DiagnosisAPI) InsertMetricsSample(ctx context.Context, param *InsertMetricsSampleParam) (*InsertMetricsSampleData, error) {
// 	if _, ok := param.Tags["tidb_cluster"]; !ok {
// 		param.Tags["tidb_cluster"] = param.TiDBCluster
// 	}
// 	ts := time.Unix(param.Timestamp, 0)
// 	point := influxdb2.NewPoint(param.Measurement, param.GetTags(), param.Fields, ts)
// 	payload, err := encodePoints(point)
// 	if err != nil {
// 		return nil, err
// 	}
// 	p, err := api.vmInsertEndpoint.InsertPath("/influx/api/v2/write")
// 	if err != nil {
// 		return nil, err
// 	}
// 	u := fmt.Sprintf("http://%s%s", api.vmInsertEndpoint.Authority(), p)
// 	req, err := http.NewRequestWithContext(ctx, http.MethodPost, u, strings.NewReader(payload))
// 	if err != nil {
// 		log.Error("new request failed", zap.Error(err))
// 		return nil, err
// 	}
// 	resp, err := api.httpCli.Do(req)
// 	if err != nil {
// 		log.Error("do request failed", zap.Error(err))
// 		return nil, err
// 	}
// 	defer resp.Body.Close()
// 	// if success, vm response with 204
// 	if resp.StatusCode/100 != 2 {
// 		log.Error("response is not ok", zap.String("status", resp.Status))
// 		return nil, fmt.Errorf("response status is %v", resp.StatusCode)
// 	}
// 	return &InsertMetricsSampleData{}, nil
// }

func (api *DiagnosisAPI) QueryAnnotations(ctx context.Context, param *QueryAnnotationsParam) (QueryAnnotationsData, error) {
	if len(param.Measurement) == 0 {
		param.Measurement = "fast_tune_anomaly"
	}
	ts, interval := param.GetRollUpParam()
	queryExpr := fmt.Sprintf(`{__name__=~"%s.*",tidb_cluster="%s"}[%s]`, param.Measurement, param.TiDBCluster, interval)
	v, err := api.instantQueryMetrics(ctx, queryExpr, ts)
	if err != nil {
		return nil, err
	}
	matrix, ok := v.(model.Matrix)
	if !ok {
		log.Error("convert to matrix failed", zap.Any("value", v))
		return nil, fmt.Errorf("")
	}
	data := make(QueryAnnotationsData, 0)
	for _, sample := range matrix {
		for _, pair := range sample.Values {
			item := QueryAnnotationItem{
				Annotation: DefaultAnomalyAnnotation(),
				Title:      "anomaly title",
				Tags:       "anomaly tags",
				Text:       "anomaly text",
			}
			if panelID, ok := sample.Metric["panel_id"]; ok {
				item.PanelID, _ = strconv.ParseInt(string(panelID), 0, 64)
			}
			item.Title = string(sample.Metric["title"])
			item.Tags = string(sample.Metric["tags"])
			item.Text = string(sample.Metric["text"])
			item.Time = pair.Timestamp.UnixNano() / 1e6
			if pair.Value > 0 {
				item.TimeEnd = item.Time + int64(pair.Value)*1e3
			}
			data = append(data, item)
		}
	}
	return data, nil
}

func (api *DiagnosisAPI) QueryNodeGraph(ctx context.Context, param *QueryNodeGraphParam) (*QueryNodeGraphData, error) {
	if len(param.Measurement) == 0 {
		param.Measurement = "fast_tune_similarity"
	}
	ts, interval := param.GetRollUpParam()
	queryExpr := fmt.Sprintf(`first_over_time({__name__=~"%s.*",tidb_cluster="%s"}[%s])`, param.Measurement, param.TiDBCluster, interval)
	v, err := api.instantQueryMetrics(ctx, queryExpr, ts)
	if err != nil {
		return nil, err
	}
	vector, ok := v.(model.Vector)
	if !ok {
		log.Error("convert to vector failed", zap.Any("value", v))
		return nil, fmt.Errorf("")
	}
	data := QueryNodeGraphData{
		Nodes: make([]*Node, 0),
		Edges: make([]*Edge, 0),
	}

	nodesLookup := make(map[int64]struct{})
	if len(vector) == 0 {
		return &data, nil
	}
	for _, sample := range vector {
		similarity := float64(sample.Value)
		idStr := string(sample.Metric["id"])
		id, err := strconv.ParseInt(idStr, 0, 64)
		if err != nil {
			log.Warn("parse int fail skip the node", zap.String("id", idStr), zap.Error(err))
			continue
		}
		node := DefaultNode()

		node.ID = idStr
		node.Title = idStr
		node.SubTitle = string(sample.Metric["title"])
		node.MainStat = fmt.Sprintf("%.3f", similarity)
		node.ArcPositive = similarity
		node.ArcNegative = 1 - similarity

		data.Nodes = append(data.Nodes, node)
		nodesLookup[id] = struct{}{}
	}
	for _, node := range data.Nodes {
		id, _ := strconv.ParseInt(node.ID, 0, 64)
		if targets, ok := EdgeMatrixV2[id]; ok {
			for _, target := range targets {
				if _, ok := nodesLookup[target]; !ok {
					continue
				}
				edge := DefaultEdge()
				edge.ID = fmt.Sprintf("%v%v", id, target)
				edge.Source = node.ID
				edge.Target = fmt.Sprintf("%v", target)
				data.Edges = append(data.Edges, edge)
			}
		}
	}
	return &data, nil
}

func (api *DiagnosisAPI) QueryDynamicTextValue(ctx context.Context, param *QueryDynamicTextValueParam) (QueryDynamicTextValueData, error) {
	ts, interval := param.GetRollUpParam()
	queryExpr := fmt.Sprintf(`{__name__=~"%s.*",tidb_cluster="%s"}[%s]`, param.Measurement, param.TiDBCluster, interval)
	v, err := api.instantQueryMetrics(ctx, queryExpr, ts)
	if err != nil {
		return nil, err
	}
	matrix, ok := v.(model.Matrix)
	if !ok {
		log.Error("convert to matrix failed", zap.Any("value", v))
		return nil, fmt.Errorf("expect response to be matrix, got %T", v)
	}
	data := make(QueryDynamicTextValueData)
	if len(param.Default1) > 0 {
		for _, field := range strings.Split(param.Default1, ",") {
			data[field] = int64(1)
		}
	}
	if len(matrix) == 0 {
		return data, nil
	}
	perfix := param.Measurement
	if !strings.HasSuffix(perfix, "_") {
		perfix = fmt.Sprintf("%s_", perfix)
	}
	instanceCnt := make(map[string]int)
	startTsToTimeIdx := make(map[int64]int)
	// iterate twice to build data
	for _, sample := range matrix {
		if sample.Metric["type"] != "duration_seconds" {
			continue
		}
		for idx, pair := range sample.Values {
			startTsToTimeIdx[pair.Timestamp.Unix()] = idx + 1
		}
	}

	for _, sample := range matrix {
		metricsName := string(sample.Metric["__name__"])
		fieldName := strings.TrimPrefix(metricsName, perfix)
		if len(sample.Values) == 0 {
			continue
		}
		if sample.Metric["aggr"] == "first" {
			value := sample.Values[0].Value
			data[fieldName] = float64(value)
			continue
		}

		for _, pair := range sample.Values {
			value := pair.Value
			idx, ok := startTsToTimeIdx[pair.Timestamp.Unix()]
			if !ok {
				continue
			}
			switch sample.Metric["type"] {
			case "float":
				key := fmt.Sprintf("%s_time_%d", fieldName, idx)
				data[key] = float64(value)
			case "int":
				key := fmt.Sprintf("%s_time_%d", fieldName, idx)
				data[key] = int64(value)
			case "duration_seconds":
				key := fmt.Sprintf("%s_%d", fieldName, idx)
				seconds := int64(value)
				startUnix := pair.Timestamp.Unix()
				endUnix := startUnix + seconds
				data[key] = int64(1)
				data[fmt.Sprintf("%s_start_unix", key)] = startUnix
				data[fmt.Sprintf("%s_end_unix", key)] = endUnix
				data[fmt.Sprintf("%s_start_unix_rfc3339", key)] = time.Unix(startUnix, 0).Format(time.RFC3339)
				data[fmt.Sprintf("%s_end_unix_rfc3339", key)] = time.Unix(endUnix, 0).Format(time.RFC3339)
			case "unix_seconds":
				key := fmt.Sprintf("%s_time_%d", fieldName, idx)
				data[key] = int64(value)
				data[fmt.Sprintf("%s_rfc3339", key)] = time.Unix(int64(value), 0).Format(time.RFC3339)
			case "address":
				if value == 0 {
					continue
				}
				if _, ok := instanceCnt[fieldName]; !ok {
					instanceCnt[fieldName] = 1
				} else {
					instanceCnt[fieldName]++
				}
				cnt := instanceCnt[fieldName]
				instanceKey := fmt.Sprintf("%s_time_%d_%d", fieldName, idx, cnt)
				instanceAddKey := fmt.Sprintf("%s_time_%d_%d_addr", fieldName, idx, cnt)
				instanceAddr := sample.Metric["instance"]
				data[instanceAddKey] = string(instanceAddr)
				data[instanceKey] = int64(value)
			default:
				key := fmt.Sprintf("%s_time_%d", fieldName, idx)
				data[key] = value
			}
		}
	}
	return data, nil
}

// instantQueryMetrics use prometheus instant query to get metrics
func (api *DiagnosisAPI) instantQueryMetrics(ctx context.Context, queryExpr string, ts int64) (model.Value, error) {
	p, err := api.vmSelectEndpoint.PromQueryPath("/api/v1/query")
	if err != nil {
		return nil, err
	}
	u := fmt.Sprintf("http://%s%s", api.vmSelectEndpoint.Authority(), p)
	payload := url.Values{
		"query": {queryExpr},
		"time":  {strconv.FormatInt(ts, 10)},
	}

	req, err := http.NewRequestWithContext(ctx, http.MethodPost, u, strings.NewReader(payload.Encode()))
	if err != nil {
		return nil, err
	}
	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
	resp, err := api.httpCli.Do(req)
	if err != nil {
		return nil, err
	}
	defer func() {
		_, _ = io.Copy(io.Discard, resp.Body)
		_ = resp.Body.Close()
	}()
	if resp.StatusCode != http.StatusOK {
		bs, _ := io.ReadAll(resp.Body)
		return nil, fmt.Errorf("response status: %s, message: %s", resp.Status, string(bs))
	}

	mResp := metricsResp{}
	if err := json.NewDecoder(resp.Body).Decode(&mResp); err != nil {
		return nil, err
	}
	return mResp.Data.v, nil
}

// reverseProxyWrapper wrap native ReversePorxy for customization
type reverseProxyWrapper struct {
	*httputil.ReverseProxy
}

func (wrapper *reverseProxyWrapper) WithBasePath(basePath string) *reverseProxyWrapper {
	originDirector := wrapper.ReverseProxy.Director
	wrapper.ReverseProxy.Director = func(req *http.Request) {
		req.URL.Path = strings.TrimPrefix(req.URL.Path, basePath)
		originDirector(req)
	}
	return wrapper
}

func buildLogProxy(cfg *config.LogConfig) (*reverseProxyWrapper, error) {
	if len(cfg.Endpoint) == 0 {
		return nil, errors.New("endpoint is empty in log config")
	}
	u, err := url.Parse(cfg.Endpoint)
	if err != nil {
		return nil, err
	}
	rp := &httputil.ReverseProxy{
		Transport: customizedTransport(),
		Director: func(req *http.Request) {
			targetPath := strings.TrimPrefix(req.URL.Path, "/diagnosis/log_sample")
			req.URL.Scheme = u.Scheme
			req.URL.Host = u.Host
			req.URL.Path = targetPath
			// add basic auth
			if cfg.User != "" && cfg.Password != "" {
				req.SetBasicAuth(cfg.User, cfg.Password)
			}
			// deal with compatibility header
			// Treat 'application/vnd.elasticsearch+x-ndjson' as application/x-ndjson for compatibility
			// Treat 'application/vnd.elasticsearch+json' as application/json for compatibility
			if strings.HasPrefix(req.Header.Get("Content-Type"), "application/vnd.elasticsearch+json") {
				req.Header.Set("Content-Type", "application/json")
			} else if strings.HasPrefix(req.Header.Get("Content-Type"), "application/vnd.elasticsearch+x-ndjson") {
				req.Header.Set("Content-Type", "application/x-ndjson")
			}
		},
		ModifyResponse: func(response *http.Response) error {
			// https://github.com/elastic/elasticsearch-py/blob/main/elasticsearch/_sync/client/_base.py#L328
			response.Header.Set("x-elastic-product", "Elasticsearch")
			return nil
		},
	}
	return &reverseProxyWrapper{ReverseProxy: rp}, nil
}

// func encodePoints(point *write.Point) (string, error) {
// 	var buffer bytes.Buffer
// 	e := lp.NewEncoder(&buffer)
// 	e.SetFieldTypeSupport(lp.UintSupport)
// 	e.FailOnFieldErr(true)
// 	e.SetPrecision(time.Nanosecond)
// 	_, err := e.Encode(point)
// 	if err != nil {
// 		return "", err
// 	}
// 	return buffer.String(), nil
// }

// copy from prometheus client golang.
type metricsResp struct {
	Status string              `json:"status"`
	Data   *metricsQueryResult `json:"data"`
}

type metricsQueryResult struct {
	// The decoded value.
	v model.Value
}

func (qr *metricsQueryResult) UnmarshalJSON(b []byte) error {
	v := struct {
		Type   model.ValueType `json:"resultType"`
		Result json.RawMessage `json:"result"`
	}{}

	err := json.Unmarshal(b, &v)
	if err != nil {
		return err
	}

	switch v.Type {
	case model.ValScalar:
		var sv model.Scalar
		err = json.Unmarshal(v.Result, &sv)
		qr.v = &sv

	case model.ValVector:
		var vv model.Vector
		err = json.Unmarshal(v.Result, &vv)
		qr.v = vv

	case model.ValMatrix:
		var mv model.Matrix
		err = json.Unmarshal(v.Result, &mv)
		qr.v = mv

	default:
		err = fmt.Errorf("unexpected value type %q", v.Type)
	}
	return err
}