package service

import (
	"context"
	"dataservices/commons/config"
	dto "dataservices/internal/dto/dataapi_control"
	"fmt"
	"io"
	"math"
	"net/http"
	"net/url"
	"time"

	jsoniter "github.com/json-iterator/go"
	"github.com/prometheus/common/model"
	tierr "github.com/tidbcloud/tiinf/errs"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
)

type dataappMetricSvc struct {
	cnf        config.DataAppMetricsQueryConfig
	httpClient *http.Client
}

var globalDataappMetricSvc dataappMetricSvc

func InitDataAppMetricsService(cnf *config.DataAppMetricsQueryConfig) {
	globalDataappMetricSvc.cnf = *cnf
	globalDataappMetricSvc.httpClient = &http.Client{Timeout: time.Duration(cnf.QueryPromTimeoutSec) * time.Second}
}

func (c *dataappMetricSvc) doPromQuery(ctx context.Context, queryUrl url.URL, org string) (*dto.PromQueryRes, error) {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, queryUrl.String(), nil)
	if err != nil {
		return nil, tierr.NewInternalServerError("req prom build req failed", err)
	}
	if len(org) > 0 {
		req.Header.Set("X-Scope-OrgID", org)
	}
	tilog.Info(ctx, "prom query url", zap.String("url", req.URL.String()), zap.String("org_id", org))
	respRaw, err := c.httpClient.Do(req)
	if nil != err {
		tilog.Error(ctx, "query prom failed", zap.Error(err))
		return nil, tierr.NewInternalServerError("req prom http req failed", err)
	}
	if respRaw.StatusCode != http.StatusOK {
		return nil, tierr.NewInternalServerError(fmt.Sprintf("prom http code %d", respRaw.StatusCode), nil)
	}
	defer respRaw.Body.Close()
	data, err := io.ReadAll(respRaw.Body)
	if err != nil {
		return nil, tierr.NewInternalServerError(fmt.Sprintf("prom read resp body failed, %v", err), err)
	}
	var resp dto.PromQueryRes
	if err := jsoniter.Unmarshal(data, &resp); err != nil {
		return nil, tierr.NewInternalServerError(fmt.Sprintf("unmarshal json failed, %v", err), err)
	}
	if resp.Status != "success" {
		return nil, tierr.NewInternalServerError(fmt.Sprintf("prom query not success, status is %s", resp.Status), nil)
	}
	return &resp, nil
}

func (c *dataappMetricSvc) doRangeQueryProm(ctx context.Context, param *dto.PromQueryParam) (*dto.PromQueryRes, error) {
	queryUrl := url.URL{
		Scheme: "http",
		Host:   fmt.Sprintf("%s:%d", c.cnf.PromAddr, c.cnf.PromPort),
		Path:   "/api/v1/query_range",
	}
	query := queryUrl.Query()
	query.Add("step", param.Step.String())
	query.Add("start", uint64ToStr(param.Start))
	query.Add("end", uint64ToStr(param.End))
	query.Add("query", param.PromQl)
	queryUrl.RawQuery = query.Encode()
	return c.doPromQuery(ctx, queryUrl, "")
}

func (c *dataappMetricSvc) doInstantQueryProm(ctx context.Context, param *dto.PromQueryParam) (*dto.PromQueryRes, error) {
	queryUrl := url.URL{
		Scheme: "http",
		Host:   fmt.Sprintf("%s:%d", c.cnf.PromAddr, c.cnf.PromPort),
		Path:   "/api/v1/query",
	}
	query := queryUrl.Query()
	query.Add("time", uint64ToStr(param.End))
	query.Add("query", param.PromQl)
	queryUrl.RawQuery = query.Encode()
	return c.doPromQuery(ctx, queryUrl, "")
}

const (
	ConstMetricDataAppReqCounter    = "dataservice_custommetrics_latency_ms_count"
	ConstMetricDataAppLatencySum    = "dataservice_custommetrics_latency_ms_sum"
	ConstMetricDataAppLatencyBucket = "dataservice_custommetrics_latency_ms_bucket"
	ConstMetricDataAppErrorCounter  = "dataservice_custommetrics_error_counter"
	ConstMetricDataAppUnauthCounter = "dataservice_custommetrics_unauth_req_counter"
)

func (c *dataappMetricSvc) QueryDataappReqCount(ctx context.Context, req *dto.PromQueryReq) (*dto.PromQueryRes, error) {
	promql := fmt.Sprintf(`sum(increase(%s{dataapp_external_id="%s"}[%s]))`, ConstMetricDataAppReqCounter, req.DataAppExtId, req.RangeSelector)
	if req.RangeSelector <= model.Duration(time.Duration(c.cnf.HighResolutionThresholdSec)*time.Second) {
		stream := fmt.Sprintf(`%s{dataapp_external_id="%s"}`, ConstMetricDataAppReqCounter, req.DataAppExtId)
		promql = fmt.Sprintf(`sum(%s) - (sum(%s offset %s) or vector(0)) >=0`, stream, stream, req.RangeSelector)
	}
	tilog.Info(ctx, "QueryDataappReqCount", zap.String("promql", promql))
	res, err := c.doRangeQueryProm(ctx, &dto.PromQueryParam{
		PromQl:       promql,
		PromQueryReq: *req,
	})
	if err != nil {
		tilog.Error(ctx, "QueryDataappReqCount failed", zap.Error(err), zap.String("req", fmt.Sprintf("%+v", req)), zap.String("promql", promql))
		return nil, err
	}
	return res, nil
}

func (c *dataappMetricSvc) QueryDataappErrorCount(ctx context.Context, req *dto.PromQueryReq) (*dto.PromQueryRes, error) {
	promql := fmt.Sprintf(`sum(increase(%s{dataapp_external_id="%s"}[%s]))`, ConstMetricDataAppErrorCounter, req.DataAppExtId, req.RangeSelector)
	if req.RangeSelector <= model.Duration(time.Duration(c.cnf.HighResolutionThresholdSec)*time.Second) {
		stream := fmt.Sprintf(`%s{dataapp_external_id="%s"}`, ConstMetricDataAppErrorCounter, req.DataAppExtId)
		promql = fmt.Sprintf(`sum(%s) - (sum(%s offset %s) or vector(0)) >=0`, stream, stream, req.RangeSelector)
	}
	tilog.Info(ctx, "QueryDataappErrorCount", zap.String("promql", promql))
	res, err := c.doRangeQueryProm(ctx, &dto.PromQueryParam{
		PromQl:       promql,
		PromQueryReq: *req,
	})
	if err != nil {
		tilog.Error(ctx, "QueryDataappErrorCount failed", zap.Error(err), zap.String("req", fmt.Sprintf("%+v", req)), zap.String("promql", promql))
		return nil, err
	}
	return res, nil
}

func (c *dataappMetricSvc) QueryDataappUnauthCount(ctx context.Context, req *dto.PromQueryReq) (*dto.PromQueryRes, error) {
	promql := fmt.Sprintf(`sum(increase(%s{dataapp_external_id="%s"}[%s]))`, ConstMetricDataAppUnauthCounter, req.DataAppExtId, req.RangeSelector)
	if req.RangeSelector <= model.Duration(time.Duration(c.cnf.HighResolutionThresholdSec)*time.Second) {
		stream := fmt.Sprintf(`%s{dataapp_external_id="%s"}`, ConstMetricDataAppUnauthCounter, req.DataAppExtId)
		promql = fmt.Sprintf(`sum(%s) - (sum(%s offset %s) or vector(0)) >=0`, stream, stream, req.RangeSelector)
	}
	tilog.Info(ctx, "QueryDataappUnauthCount", zap.String("promql", promql))
	res, err := c.doRangeQueryProm(ctx, &dto.PromQueryParam{
		PromQl:       promql,
		PromQueryReq: *req,
	})
	if err != nil {
		tilog.Error(ctx, "QueryDataappUnauthCount failed", zap.Error(err), zap.String("req", fmt.Sprintf("%+v", req)), zap.String("promql", promql))
		return nil, err
	}
	return res, nil
}

func (c *dataappMetricSvc) QueryDataappLatencyTotal(ctx context.Context, req *dto.PromQueryReq) (*dto.PromQueryRes, error) {
	promql := fmt.Sprintf(`sum(increase(%s{dataapp_external_id="%s",endpoint_id!="unknown_endpoint"}[%s]))`, ConstMetricDataAppLatencySum, req.DataAppExtId, req.RangeSelector)
	if req.RangeSelector <= model.Duration(time.Duration(c.cnf.HighResolutionThresholdSec)*time.Second) {
		stream := fmt.Sprintf(`%s{dataapp_external_id="%s",endpoint_id!="unknown_endpoint"}`, ConstMetricDataAppLatencySum, req.DataAppExtId)
		promql = fmt.Sprintf(`sum(%s) - (sum(%s offset %s) or vector(0)) >=0`, stream, stream, req.RangeSelector)
	}
	tilog.Info(ctx, "QueryDataappLatencyTotal", zap.String("promql", promql))
	res, err := c.doRangeQueryProm(ctx, &dto.PromQueryParam{
		PromQl:       promql,
		PromQueryReq: *req,
	})
	if err != nil {
		tilog.Error(ctx, "QueryDataappLatencyTotal failed", zap.Error(err), zap.String("req", fmt.Sprintf("%+v", req)), zap.String("promql", promql))
		return nil, err
	}
	return res, nil
}

func (c *dataappMetricSvc) QueryDataappLatencyPT99(ctx context.Context, req *dto.PromQueryReq) (*dto.PromQueryRes, error) {
	promql := fmt.Sprintf(`histogram_quantile(0.99,sum by (le) (rate(%s{dataapp_external_id="%s",endpoint_id!="unknown_endpoint"}[%s])))`, ConstMetricDataAppLatencyBucket, req.DataAppExtId, req.RangeSelector)
	tilog.Info(ctx, "QueryDataappLatencyPT99", zap.String("promql", promql))
	res, err := c.doRangeQueryProm(ctx, &dto.PromQueryParam{
		PromQl:       promql,
		PromQueryReq: *req,
	})
	if err != nil {
		tilog.Error(ctx, "QueryDataappLatencyPT99 failed", zap.Error(err), zap.String("req", fmt.Sprintf("%+v", req)), zap.String("promql", promql))
		return nil, err
	}
	if len(res.Data.Result) > 0 {
		for i, v := range res.Data.Result[0].Values {
			if math.IsNaN(float64(v.Value)) {
				res.Data.Result[0].Values[i].Value = 0
			}
		}
	}
	return res, nil
}

// ==========topk

func (c *dataappMetricSvc) QueryDataappTopKReqEndpoints(ctx context.Context, req *dto.PromQueryReq) (*dto.PromQueryRes, error) {
	if time.Duration(req.RangeSelector) <= time.Second*time.Duration(c.cnf.UseLokiThresSec) {
		return c.QueryDataappTopKReqEndpointsLokiVer(ctx, req)
	}
	promql := fmt.Sprintf(`topk(%d,sum by (endpoint_id) (increase(%s{dataapp_external_id="%s",endpoint_id!="unknown_endpoint"}[%s])))`, req.TopK, ConstMetricDataAppReqCounter, req.DataAppExtId, req.RangeSelector)
	if req.RangeSelector <= model.Duration(time.Duration(c.cnf.HighResolutionThresholdSec)*time.Second) {
		stream := fmt.Sprintf(`%s{dataapp_external_id="%s",endpoint_id!="unknown_endpoint"}`, ConstMetricDataAppReqCounter, req.DataAppExtId)
		promql = fmt.Sprintf(`topk(%d,sum by (endpoint_id) (%s - %s offset %s >=0))`, req.TopK, stream, stream, req.RangeSelector)
	}
	tilog.Info(ctx, "QueryDataappTopKReqEndpoints", zap.String("promql", promql))
	res, err := c.doInstantQueryProm(ctx, &dto.PromQueryParam{
		PromQl:       promql,
		PromQueryReq: *req,
	})
	if err != nil {
		tilog.Error(ctx, "QueryDataappTopKReqEndpoints failed", zap.Error(err), zap.String("req", fmt.Sprintf("%+v", req)), zap.String("promql", promql))
		return nil, err
	}
	return res, nil
}

func (c *dataappMetricSvc) QueryDataappTopKReqEndpointsLokiVer(ctx context.Context, req *dto.PromQueryReq) (*dto.PromQueryRes, error) {
	lokiql := fmt.Sprintf(`topk(%d,sum by (endpoint_id) (count_over_time({dataapp="%s"}|json [%s])))`, req.TopK, req.DataAppExtId, req.RangeSelector)
	tilog.Info(ctx, "QueryDataappTopKReqEndpointsLokiVer", zap.String("lokiql", lokiql))
	res, err := c.doInstantQueryLokiMetric(ctx, &dto.PromQueryParam{
		PromQl:       lokiql,
		PromQueryReq: *req,
	})
	if err != nil {
		tilog.Error(ctx, "QueryDataappTopKReqEndpoints failed", zap.Error(err), zap.String("req", fmt.Sprintf("%+v", req)), zap.String("lokiql", lokiql))
		return nil, err
	}
	return res, nil
}

func (c *dataappMetricSvc) QueryDataappTopKErrorEndpoints(ctx context.Context, req *dto.PromQueryReq) (*dto.PromQueryRes, error) {
	if time.Duration(req.RangeSelector) <= time.Second*time.Duration(c.cnf.UseLokiThresSec) {
		return c.QueryDataappTopKErrorEndpointsLokiVer(ctx, req)
	}
	promql := fmt.Sprintf(`topk(%d,sum by (endpoint_id) (increase(%s{dataapp_external_id="%s",endpoint_id!="unknown_endpoint"}[%s])))`, req.TopK, ConstMetricDataAppErrorCounter, req.DataAppExtId, req.RangeSelector)
	if req.RangeSelector <= model.Duration(time.Duration(c.cnf.HighResolutionThresholdSec)*time.Second) {
		stream := fmt.Sprintf(`%s{dataapp_external_id="%s",endpoint_id!="unknown_endpoint"}`, ConstMetricDataAppErrorCounter, req.DataAppExtId)
		promql = fmt.Sprintf(`topk(%d,sum by (endpoint_id) (%s - %s offset %s >=0))`, req.TopK, stream, stream, req.RangeSelector)
	}
	tilog.Info(ctx, "QueryDataappTopKErrorEndpoints", zap.String("promql", promql))
	res, err := c.doInstantQueryProm(ctx, &dto.PromQueryParam{
		PromQl:       promql,
		PromQueryReq: *req,
	})
	if err != nil {
		tilog.Error(ctx, "QueryDataappTopKReqEndpoints failed", zap.Error(err), zap.String("req", fmt.Sprintf("%+v", req)), zap.String("promql", promql))
		return nil, err
	}
	return res, nil
}

func (c *dataappMetricSvc) QueryDataappTopKErrorEndpointsLokiVer(ctx context.Context, req *dto.PromQueryReq) (*dto.PromQueryRes, error) {
	lokiql := fmt.Sprintf(`topk(%d,sum by (endpoint_id) (count_over_time({dataapp="%s",log_level="ERROR"}|json [%s])))`, req.TopK, req.DataAppExtId, req.RangeSelector)
	tilog.Info(ctx, "QueryDataappTopKErrorEndpoints", zap.String("lokiql", lokiql))
	res, err := c.doInstantQueryLokiMetric(ctx, &dto.PromQueryParam{
		PromQl:       lokiql,
		PromQueryReq: *req,
	})
	if err != nil {
		tilog.Error(ctx, "QueryDataappTopKReqEndpoints failed", zap.Error(err), zap.String("req", fmt.Sprintf("%+v", req)), zap.String("lokiql", lokiql))
		return nil, err
	}
	return res, nil
}

func (c *dataappMetricSvc) QueryDataappTopKLatencyEndpoints(ctx context.Context, req *dto.PromQueryReq) (*dto.PromQueryRes, error) {
	if time.Duration(req.RangeSelector) <= time.Second*time.Duration(c.cnf.UseLokiThresSec) {
		return c.QueryDataappTopKLatencyEndpointsLokiVer(ctx, req)
	}
	labelFilter := fmt.Sprintf(`dataapp_external_id="%s",endpoint_id!="unknown_endpoint"`, req.DataAppExtId)
	promql := fmt.Sprintf(`topk(%d, sum by (endpoint_id)(increase(%s{%s}[%s])) / (sum by (endpoint_id) (increase(%s{%s}[%s])) >0 ))`,
		req.TopK,
		ConstMetricDataAppLatencySum, labelFilter, req.RangeSelector,
		ConstMetricDataAppReqCounter, labelFilter, req.RangeSelector,
	)
	tilog.Info(ctx, "QueryDataappTopKLatencyEndpoints", zap.String("promql", promql))
	res, err := c.doInstantQueryProm(ctx, &dto.PromQueryParam{
		PromQl:       promql,
		PromQueryReq: *req,
	})
	if err != nil {
		tilog.Error(ctx, "QueryDataappTopKLatencyEndpoints failed", zap.Error(err), zap.String("req", fmt.Sprintf("%+v", req)), zap.String("promql", promql))
		return nil, err
	}
	return res, nil
}

func (c *dataappMetricSvc) QueryDataappTopKLatencyEndpointsLokiVer(ctx context.Context, req *dto.PromQueryReq) (*dto.PromQueryRes, error) {
	lokiql := fmt.Sprintf(`topk(%d,avg_over_time({dataapp="%s"}|json|unwrap latency [%s]) by (endpoint_id))`, req.TopK, req.DataAppExtId, req.RangeSelector)
	tilog.Info(ctx, "QueryDataappTopKErrorEndpoints", zap.String("lokiql", lokiql))
	res, err := c.doInstantQueryLokiMetric(ctx, &dto.PromQueryParam{
		PromQl:       lokiql,
		PromQueryReq: *req,
	})
	if err != nil {
		tilog.Error(ctx, "QueryDataappTopKReqEndpoints failed", zap.Error(err), zap.String("req", fmt.Sprintf("%+v", req)), zap.String("lokiql", lokiql))
		return nil, err
	}
	return res, nil
}

func (c *dataappMetricSvc) doInstantQueryLokiMetric(ctx context.Context, param *dto.PromQueryParam) (*dto.PromQueryRes, error) {
	queryUrl := url.URL{
		Scheme: "http",
		Host:   fmt.Sprintf("%s:%d", c.cnf.LokiAddr, c.cnf.LokiPort),
		Path:   "/loki/api/v1/query",
	}
	query := queryUrl.Query()
	query.Add("time", uint64ToStr(param.End*uint64(time.Second)))
	query.Add("query", param.PromQl)
	queryUrl.RawQuery = query.Encode()
	return c.doPromQuery(ctx, queryUrl, param.Org)
}

func MetricsQuery(ctx context.Context, req *dto.PromQueryReq) (*dto.PromQueryRes, error) {
	switch req.QueryTopic {
	case dto.ConstMetricsQueryTopicCountReq:
		return globalDataappMetricSvc.QueryDataappReqCount(ctx, req)
	case dto.ConstMetricsQueryTopicCountError:
		return globalDataappMetricSvc.QueryDataappErrorCount(ctx, req)
	case dto.ConstMetricsQueryTopicCountUnauth:
		return globalDataappMetricSvc.QueryDataappUnauthCount(ctx, req)
	case dto.ConstMetricsQueryTopicSumLatency:
		return globalDataappMetricSvc.QueryDataappLatencyTotal(ctx, req)
	case dto.ConstMetricsQueryTopicPT99Latency:
		return globalDataappMetricSvc.QueryDataappLatencyPT99(ctx, req)
	case dto.ConstMetricsQueryTopicTopKReq:
		return globalDataappMetricSvc.QueryDataappTopKReqEndpoints(ctx, req)
	case dto.ConstMetricsQueryTopicTopKError:
		return globalDataappMetricSvc.QueryDataappTopKErrorEndpoints(ctx, req)
	case dto.ConstMetricsQueryTopicTopKLantency:
		return globalDataappMetricSvc.QueryDataappTopKLatencyEndpoints(ctx, req)
	}
	return nil, tierr.NewBadRequest(fmt.Sprintf("invalid query topic %d", req.QueryTopic), nil)
}