package client

import (
	"context"
	"crypto/tls"
	"encoding/json"
	"errors"
	"fmt"
	"io/ioutil"
	"net/http"
	"net/url"
	"strconv"
	"strings"
	"time"

	"github.com/pingcap/log"
	"github.com/tidbcloud/metrics-integration/pipeline/central/config"
	"github.com/tidbcloud/metrics-integration/pipeline/central/metrics"
	"github.com/tidbcloud/metrics-integration/pipeline/internal/datasource"
	"go.uber.org/zap"
)

var (
	QueryRequestStatusEmpty   = "empty"
	QueryRequestStatusError   = "error"
	QueryRequestStatusSuccess = "success"
)

// localTestHTTPClient logs every redirect hop and skips TLS verification for
// local port-forward tunnels (e.g. mirrord -L 29082:host:443 →
// https://127.0.0.1:29082). It must never be used outside local test mode.
var localTestHTTPClient = &http.Client{
	Transport: &http.Transport{
		TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, //nolint:gosec // local test tunnel only
	},
	CheckRedirect: func(req *http.Request, via []*http.Request) error {
		log.Debug("local test http redirect",
			zap.String("from", via[len(via)-1].URL.String()),
			zap.String("to", req.URL.String()),
			zap.String("host_header", req.Host),
		)
		if len(via) >= 10 {
			return errors.New("stopped after 10 redirects")
		}
		return nil
	},
}

func useLocalTestHTTPClient() bool {
	return config.Global().O11yGlobalServerConfig.LocalTestEnabled
}

func getHTTPClient() *http.Client {
	if useLocalTestHTTPClient() {
		return localTestHTTPClient
	}
	return http.DefaultClient
}

func logMetricRequest(url, host string) {
	if !useLocalTestHTTPClient() {
		return
	}
	log.Debug("local test metric request",
		zap.String("url", url),
		zap.String("host_header", host),
	)
}

func logNonOKMetricResponse(resp *http.Response) {
	if !useLocalTestHTTPClient() {
		return
	}
	log.Debug("local test non-200 response",
		zap.String("status", resp.Status),
		zap.String("final_url", resp.Request.URL.String()),
		zap.String("location", resp.Header.Get("Location")),
	)
}

type LabelSet struct {
	MetricName          string `json:"__name__,omitempty"`
	ClusterID           string `json:"cluster_id,omitempty"`
	Component           string `json:"component,omitempty"`
	Instance            string `json:"instance,omitempty"`
	Job                 string `json:"job,omitempty"`
	KeyspaceID          string `json:"keyspace_id,omitempty"`
	KeyspaceName        string `json:"keyspace_name,omitempty"`
	KubernetesNamespace string `json:"kubernetes_namespace,omitempty"`
	Name                string `json:"name,omitempty"`
	Replica             string `json:"replica,omitempty"`
	SeedAPI             string `json:"seed_api,omitempty"`
	SeedProvider        string `json:"seed_provider,omitempty"`
	SeedRegion          string `json:"seed_region,omitempty"`
	ShootInfra          string `json:"shoot_infra,omitempty"`
	ShootName           string `json:"shoot_name,omitempty"`
	Tenant              string `json:"tenant,omitempty"`
	TenantID            string `json:"tenant_id,omitempty"`
	TidbCluster         string `json:"tidb_cluster,omitempty"`
	TidbClusterID       string `json:"tidb_cluster_id,omitempty"`
	Type                string `json:"type,omitempty"`
	SqlType             string `json:"sql_type,omitempty"`
	Le                  string `json:"le,omitempty"`
	Result              string `json:"result,omitempty"`
	InTxn               string `json:"in_txn,omitempty"`
	TxnMode             string `json:"txn_mode,omitempty"`
	Changefeed          string `json:"changefeed,omitempty"`
	ResourceGroup       string `json:"resource_group,omitempty"`
	Device              string `json:"device"`
	Task                string `json:"task"`
	JobID               string `json:"job_id"`
	SourceID            string `json:"source_id"`
}

type Result struct {
	LabelSet `json:"metric"`
	Value    []interface{}   `json:"value"`
	Values   [][]interface{} `json:"values"`
}

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

func shouldUseProjectDatasource(metric string) bool {
	switch metric {
	case "tidb_cloud.dm_task_state",
		"tidb_cloud.dm_syncer_replication_lag",
		"tidb_cloud.dm_syncer_replication_lag_bucket",
		"tidb_cloud.dm_syncer_replication_lag_gauge",
		"tidb_cloud.dm_relay_read_error_count",
		"tidbcloud_dm_task_status",
		"tidbcloud_dm_syncer_replication_lag",
		"tidbcloud_dm_syncer_replication_lag_bucket",
		"tidbcloud_dm_syncer_replication_lag_gauge",
		"tidbcloud_dm_relay_read_error_count":
		return true
	default:
		return false
	}
}

func getQueryURLByMetric(addr datasource.MetricDataAddress, metric string) string {
	if shouldUseProjectDatasource(metric) && len(addr.ProjectQueryUrl) != 0 {
		return addr.ProjectQueryUrl
	}

	return addr.QueryUrl
}

func getQueryHostByMetric(addr datasource.MetricDataAddress, metric string) string {
	if shouldUseProjectDatasource(metric) && len(addr.ProjectQueryHost) != 0 {
		return addr.ProjectQueryHost
	}

	return addr.QueryHost
}

// unify range query and instant query results, transfer vector results to matrix results
func (r *Response) Unify() {
	if r.Data.ResultType == "vector" {
		for i := range r.Data.Result {
			r.Data.Result[i].Values = append(r.Data.Result[i].Values, r.Data.Result[i].Value)
		}
	}
}

func DoInstantQueryWithTime(ctx context.Context, q string, metric string, t time.Time) (resp *Response, err error) {
	addr := datasource.MetricDataAddress{}
	defer func() {
		clusterId := strconv.FormatUint(addr.TidbClusterInfo.ClusterID.Uint64(), 10)
		logger := log.With(zap.String("tidb_cluster_vendor", addr.TidbClusterInfo.Vendor),
			zap.String("tidb_cluster_region", addr.TidbClusterInfo.Region),
			zap.Uint64("tidb_cluster_tenant_id", addr.TidbClusterInfo.TenantID.Uint64()),
			zap.Uint64("tidb_cluster_project_id", addr.TidbClusterInfo.ProjectID.Uint64()),
			zap.String("tidb_cluster_id", clusterId),
			zap.String("IntegrationType", addr.IntegrationType),
			zap.String("QueryUrl", addr.QueryUrl),
			zap.String("ProjectQueryUrl", addr.ProjectQueryUrl),
			zap.String("PromQL", q),
			zap.String("ExportedMetric", metric),
			zap.Time("time", t),
		)
		if err != nil || resp.Status != "success" {
			// use metric instead of promql as label to avoid high cardinality.
			metrics.O11yServerQueryRequest.WithLabelValues(addr.QueryUrl, clusterId, metric, QueryRequestStatusError).Inc()
			logger.Sugar().Errorf("doInstantQueryWithTime fail, err %v, result: %+v", zap.Error(err), resp)
		} else if len(resp.Data.Result) == 0 {
			metrics.O11yServerQueryRequest.WithLabelValues(addr.QueryUrl, clusterId, metric, QueryRequestStatusEmpty).Inc()
			logger.Info("doInstantQueryWithTime empty result")
		} else {
			// FIXME: we don't know how to cleanup those metrics when the cluster is deleted.
			metrics.O11yServerQueryRequest.WithLabelValues(addr.QueryUrl, clusterId, metric, QueryRequestStatusSuccess).Inc()
			logger.Info("doInstantQueryWithTime success", zap.Any("result.len", len(resp.Data.Result)))
		}
	}()

	addr, err = datasource.GetMetricDataAddressFromContext(ctx)
	if err != nil {
		return nil, err
	}

	queryUrl := getQueryURLByMetric(addr, metric)
	queryHost := getQueryHostByMetric(addr, metric)
	reqCnt := 3
	for i := 0; i < reqCnt; i++ {
		result, canRetry, err := QueryMetricDataWithTime(ctx, queryUrl, queryHost, q, t)
		if err == nil {
			return result, nil
		}

		if i == reqCnt-1 {
			return nil, err
		}

		if canRetry {
			time.Sleep(100 * time.Millisecond * time.Duration(i+1))
			continue
		}

		return nil, err
	}

	return nil, errors.New("unknown error for instant query")
}

func QueryMetricDataWithTime(ctx context.Context, queryUrl, queryHost, urlParams string, t time.Time) (thanosResp *Response, canRetry bool, err error) {
	ts := t.Unix()
	now := time.Now()
	path := "/query"
	method := "GET"
	defer func() {
		duration := float64(time.Since(now)) / float64(time.Millisecond)
		metrics.O11yServerRequestDuration.WithLabelValues(queryUrl, path, method).Observe(duration)
		if err != nil {
			metrics.O11yServerRequestErrCnt.WithLabelValues(queryUrl, path, method).Inc()
		}
	}()

	urlString := fmt.Sprintf("%s%s?time=%d&query=%s", strings.TrimRight(queryUrl, "/"), path, ts, url.QueryEscape(urlParams))
	req, err := http.NewRequestWithContext(ctx, method, urlString, nil)
	if err != nil {
		return nil, false, err
	}
	if queryHost != "" {
		req.Host = queryHost
	}

	logMetricRequest(req.URL.String(), req.Host)
	resp, err := getHTTPClient().Do(req)
	if err != nil {
		return nil, true, err
	}
	defer resp.Body.Close() //nolint

	if resp.StatusCode != http.StatusOK {
		body, _ := ioutil.ReadAll(resp.Body)
		logNonOKMetricResponse(resp)
		return nil, true, fmt.Errorf("http response status %s from metric data server, body: %s", resp.Status, body)
	}

	thanosNodeMemoryRespResult := Response{}
	if err := json.NewDecoder(resp.Body).Decode(&thanosNodeMemoryRespResult); err != nil {
		return nil, false, err
	}
	thanosNodeMemoryRespResult.Unify()
	return &thanosNodeMemoryRespResult, false, nil
}

func DoInstantQuery(ctx context.Context, q string, metric string) (resp *Response, err error) {
	addr := datasource.MetricDataAddress{}
	defer func() {
		clusterId := strconv.FormatUint(addr.TidbClusterInfo.ClusterID.Uint64(), 10)
		logger := log.With(zap.String("tidb_cluster_vendor", addr.TidbClusterInfo.Vendor),
			zap.String("tidb_cluster_region", addr.TidbClusterInfo.Region),
			zap.Uint64("tidb_cluster_tenant_id", addr.TidbClusterInfo.TenantID.Uint64()),
			zap.Uint64("tidb_cluster_project_id", addr.TidbClusterInfo.ProjectID.Uint64()),
			zap.String("tidb_cluster_id", clusterId),
			zap.String("IntegrationType", addr.IntegrationType),
			zap.String("QueryUrl", addr.QueryUrl),
			zap.String("PromQL", q),
			zap.String("ExportedMetric", metric))
		if err != nil || resp.Status != "success" {
			// use metric instead of promql as label to avoid high cardinality.
			metrics.O11yServerQueryRequest.WithLabelValues(addr.QueryUrl, clusterId, metric, QueryRequestStatusError).Inc()
			logger.Sugar().Errorf("doInstantQuery fail, err %v, result: %+v", zap.Error(err), resp)
		} else if len(resp.Data.Result) == 0 {
			metrics.O11yServerQueryRequest.WithLabelValues(addr.QueryUrl, clusterId, metric, QueryRequestStatusEmpty).Inc()
			logger.Info("doInstantQuery empty result")
		} else {
			// FIXME: we don't know how to cleanup those metrics when the cluster is deleted.
			metrics.O11yServerQueryRequest.WithLabelValues(addr.QueryUrl, clusterId, metric, QueryRequestStatusSuccess).Inc()
			logger.Debug("doInstantQuery success", zap.Any("result", resp.Data.Result))
		}
	}()

	addr, err = datasource.GetMetricDataAddressFromContext(ctx)
	if err != nil {
		return nil, err
	}

	queryUrl := getQueryURLByMetric(addr, metric)
	queryHost := getQueryHostByMetric(addr, metric)
	reqCnt := 3
	for i := 0; i < reqCnt; i++ {
		result, canRetry, err := queryMetricData(ctx, queryUrl, queryHost, q)
		if err == nil {
			return result, nil
		}

		if i == reqCnt-1 {
			return nil, err
		}

		if canRetry {
			time.Sleep(100 * time.Millisecond * time.Duration(i+1))
			continue
		}

		return nil, err
	}

	return nil, errors.New("unknown error for instant query")
}

func queryMetricData(ctx context.Context, queryUrl, queryHost, urlParams string) (thanosResp *Response, canRetry bool, err error) {
	startTime := time.Now()
	path := "/query"
	method := "GET"
	defer func() {
		duration := float64(time.Since(startTime)) / float64(time.Millisecond)
		metrics.O11yServerRequestDuration.WithLabelValues(queryUrl, path, method).Observe(duration)
		if err != nil {
			metrics.O11yServerRequestErrCnt.WithLabelValues(queryUrl, path, method).Inc()
		}
	}()

	urlString := fmt.Sprintf("%s%s?query=%s", strings.TrimRight(queryUrl, "/"), path, url.QueryEscape(urlParams))
	req, err := http.NewRequestWithContext(ctx, method, urlString, nil)
	if err != nil {
		return nil, false, err
	}
	if queryHost != "" {
		req.Host = queryHost
	}

	logMetricRequest(req.URL.String(), req.Host)
	resp, err := getHTTPClient().Do(req)
	if err != nil {
		return nil, true, err
	}
	defer resp.Body.Close() //nolint

	if resp.StatusCode != http.StatusOK {
		body, _ := ioutil.ReadAll(resp.Body)
		logNonOKMetricResponse(resp)
		return nil, true, fmt.Errorf("http response status %s from metric data server, body: %s", resp.Status, body)
	}

	thanosNodeMemoryRespResult := Response{}
	if err := json.NewDecoder(resp.Body).Decode(&thanosNodeMemoryRespResult); err != nil {
		return nil, false, err
	}
	thanosNodeMemoryRespResult.Unify()
	return &thanosNodeMemoryRespResult, false, nil
}

func DoRangeQuery(ctx context.Context, q string, metric string, startTime, endTime time.Time) (resp *Response, err error) {
	addr := datasource.MetricDataAddress{}
	defer func() {
		clusterId := strconv.FormatUint(addr.TidbClusterInfo.ClusterID.Uint64(), 10)
		logger := log.With(zap.String("tidb_cluster_vendor", addr.TidbClusterInfo.Vendor),
			zap.String("tidb_cluster_region", addr.TidbClusterInfo.Region),
			zap.Uint64("tidb_cluster_tenant_id", addr.TidbClusterInfo.TenantID.Uint64()),
			zap.Uint64("tidb_cluster_project_id", addr.TidbClusterInfo.ProjectID.Uint64()),
			zap.String("tidb_cluster_id", clusterId),
			zap.String("IntegrationType", addr.IntegrationType),
			zap.String("QueryUrl", addr.QueryUrl),
			zap.String("PromQL", q),
			zap.String("ExportedMetric", metric),
			zap.Time("start_time", startTime),
			zap.Time("end_time", endTime),
		)
		if err != nil || resp.Status != "success" {
			// use metric instead of promql as label to avoid high cardinality.
			metrics.O11yServerQueryRequest.WithLabelValues(addr.QueryUrl, clusterId, metric, QueryRequestStatusError).Inc()
			logger.Sugar().Errorf("doRangeQuery fail, err %v, result: %+v", zap.Error(err), resp)
		} else if len(resp.Data.Result) == 0 {
			metrics.O11yServerQueryRequest.WithLabelValues(addr.QueryUrl, clusterId, metric, QueryRequestStatusEmpty).Inc()
			logger.Info("doRangeQuery empty result")
		} else {
			// FIXME: we don't know how to cleanup those metrics when the cluster is deleted.
			metrics.O11yServerQueryRequest.WithLabelValues(addr.QueryUrl, clusterId, metric, QueryRequestStatusSuccess).Inc()
			logger.Debug("doRangeQuery success", zap.Any("result", resp.Data.Result))
		}
	}()

	addr, err = datasource.GetMetricDataAddressFromContext(ctx)
	if err != nil {
		return nil, err
	}

	queryUrl := getQueryURLByMetric(addr, metric)
	queryHost := getQueryHostByMetric(addr, metric)
	reqCnt := 3
	for i := 0; i < reqCnt; i++ {
		result, canRetry, err := QueryMetricDataWithRange(ctx, queryUrl, queryHost, q, startTime, endTime)
		if err == nil {
			return result, nil
		}

		if i == reqCnt-1 {
			return nil, err
		}

		if canRetry {
			time.Sleep(100 * time.Millisecond * time.Duration(i+1))
			continue
		}

		return nil, err
	}

	return nil, errors.New("unknown error for instant query")
}

func QueryMetricDataWithRange(ctx context.Context, queryUrl, queryHost, urlParams string, start, end time.Time) (thanosResp *Response, canRetry bool, err error) {
	startTs := start.Unix()
	endTs := end.Unix()
	now := time.Now()
	path := "/query_range"
	method := "GET"
	defer func() {
		duration := float64(time.Since(now)) / float64(time.Millisecond)
		metrics.O11yServerRequestDuration.WithLabelValues(queryUrl, path, method).Observe(duration)
		if err != nil {
			metrics.O11yServerRequestErrCnt.WithLabelValues(queryUrl, path, method).Inc()
		}
	}()

	urlString := fmt.Sprintf("%s%s?start=%d&end=%d&step=15s&query=%s", strings.TrimRight(queryUrl, "/"), path, startTs, endTs, url.QueryEscape(urlParams))
	req, err := http.NewRequestWithContext(ctx, method, urlString, nil)
	if err != nil {
		return nil, false, err
	}
	if queryHost != "" {
		req.Host = queryHost
	}

	logMetricRequest(req.URL.String(), req.Host)
	resp, err := getHTTPClient().Do(req)
	if err != nil {
		return nil, true, err
	}
	defer resp.Body.Close() //nolint

	if resp.StatusCode != http.StatusOK {
		body, _ := ioutil.ReadAll(resp.Body)
		logNonOKMetricResponse(resp)
		return nil, true, fmt.Errorf("http response status %s from metric data server, body: %s", resp.Status, body)
	}

	thanosNodeMemoryRespResult := Response{}
	if err := json.NewDecoder(resp.Body).Decode(&thanosNodeMemoryRespResult); err != nil {
		return nil, false, err
	}
	return &thanosNodeMemoryRespResult, false, nil
}