package prometheus

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/dto"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/log"
	"github.com/prometheus/common/model"
	"go.uber.org/zap"
)

// GetTotalSQLCount calculates the total execution count of SQL statements for a given day.
// If sqlDigest is provided, it calculates the count for that specific SQL digest.
// Otherwise, it calculates the total count for all SQL statements.
// It queries data in 10-minute intervals to avoid Prometheus query limits.
func GetTotalSQLCount(clusterId string, dayStr string, sqlDigest ...string) (float64, error) {
	pc, err := CreatePrometheusClient(context.Background(), CreatePrometheusClientReq{ClusterID: clusterId})
	if err != nil {
		return 0, err
	}

	startTime, err := time.ParseInLocation("20060102", dayStr, time.UTC)
	if err != nil {
		return 0, fmt.Errorf("failed to parse day string: %w", err)
	}

	var totalCount float64
	var query string

	// Define the smaller interval and query range
	interval := 10 * time.Minute
	queryRange := "[10m]"

	// Based on whether sqlDigest is provided, construct the appropriate query.
	if len(sqlDigest) > 0 && sqlDigest[0] != "" {
		query = fmt.Sprintf(`sum(sum_over_time(topsql_stmt_exec_count{sql_digest="%s"}%s))`, sqlDigest[0], queryRange)
	} else {
		query = fmt.Sprintf(`sum(sum_over_time(topsql_stmt_exec_count%s))`, queryRange)
	}

	// Loop through the day in 10-minute intervals. A day has 24 * 6 = 144 intervals.
	for i := 0; i < 24*6; i++ {
		// The query time should be the end of each interval.
		queryTime := startTime.Add(time.Duration(i+1) * interval)
		ctx, cancel := context.WithTimeout(context.Background(), 40*time.Second)

		log.PlainInfo("Querying 10-minute execution count",
			zap.Time("queryTime", queryTime),
			zap.String("query", query))

		result, err := pc.Query(ctx, query, queryTime)
		cancel() // Release resources in a timely manner.
		if err != nil {
			log.PlainError("Query failed", zap.Time("queryTime", queryTime), zap.Error(err))
			continue // If a query for one interval fails, log it and continue to the next.
		}

		if result == nil {
			log.PlainWarn("Query result is nil", zap.Time("queryTime", queryTime))
			continue
		}

		// The result of a sum query is a vector with a single element.
		if vector, ok := result.(model.Vector); ok {
			if len(vector) > 0 {
				totalCount += float64(vector[0].Value)
			}
		} else {
			log.PlainWarn("Query result is not a vector", zap.Time("queryTime", queryTime))
		}
	}

	return totalCount, nil
}

func GetSum24HoursValue(clusterId string, dayStr string, promSql string) (float64, error) {
	pc, err := CreatePrometheusClient(context.Background(), CreatePrometheusClientReq{ClusterID: clusterId})
	if err != nil {
		return 0, err
	}

	startTime, err := time.ParseInLocation("20060102", dayStr, time.UTC)
	if err != nil {
		return 0, fmt.Errorf("failed to parse day string: %w", err)
	}

	var sumResult float64
	// var query string

	// Define the smaller interval and query range

	// Loop through the day in 10-minute intervals. A day has 24 * 6 = 144 intervals.
	for i := 0; i < 24; i++ {
		// The query time should be the end of each interval.
		queryTime := startTime.Add(time.Duration(i+1) * time.Hour)
		ctx, cancel := context.WithTimeout(context.Background(), 40*time.Second)

		log.PlainInfo("Querying 1 hour execution count",
			zap.Time("queryTime", queryTime),
			zap.String("query", promSql))

		result, err := pc.Query(ctx, promSql, queryTime)
		cancel() // Release resources in a timely manner.
		if err != nil {
			log.PlainError("Query failed", zap.Time("queryTime", queryTime), zap.Error(err))
			continue // If a query for one interval fails, log it and continue to the next.
		}

		if result == nil {
			log.PlainWarn("Query result is nil", zap.Time("queryTime", queryTime))
			continue
		}

		// The result of a sum query is a vector with a single element.
		if vector, ok := result.(model.Vector); ok {
			if len(vector) > 0 {
				sumResult += float64(vector[0].Value)
			}
		} else {
			log.PlainWarn("Query result is not a vector", zap.Time("queryTime", queryTime))
		}
	}

	return sumResult, nil
}

// GetSumValueForDay calculates the sum of a PromQL query over a given day,
// splitting the day into intervals (e.g., 30m, 1h) to avoid Prometheus query limits.
// dayStr is in format "20060102", interval is the time slice per query.
// promSql must contain the full PromQL expression with the desired range selector (e.g. [30m], [1h]).
func GetSumValueForDay(clusterId, dayStr, promSql string, interval time.Duration) (float64, error) {
	pc, err := CreatePrometheusClient(context.Background(), CreatePrometheusClientReq{ClusterID: clusterId})
	if err != nil {
		return 0, err
	}

	startTime, err := time.ParseInLocation("20060102", dayStr, time.UTC)
	if err != nil {
		return 0, fmt.Errorf("failed to parse day string: %w", err)
	}
	log.PlainInfo("Querying interval execution count", zap.Duration("interval", interval), zap.String("query", promSql))
	var sumResult float64
	// Calculate how many intervals fit into 24h
	steps := int((24 * time.Hour) / interval)

	for i := 0; i < steps; i++ {
		// Query time is the end of each interval
		queryTime := startTime.Add(time.Duration(i+1) * interval)
		ctx, cancel := context.WithTimeout(context.Background(), 40*time.Second)

		result, err := pc.Query(ctx, promSql, queryTime)
		cancel()
		if err != nil {
			log.PlainError("Query failed", zap.Time("queryTime", queryTime), zap.Error(err))
			continue
		}
		if result == nil {
			log.PlainWarn("Query result is nil", zap.Time("queryTime", queryTime))
			continue
		}

		if vector, ok := result.(model.Vector); ok {
			if len(vector) > 0 {
				val := float64(vector[0].Value)
				sumResult += val
				msg := fmt.Sprintf("%v ， value: %f", queryTime, val)
				log.PlainInfo(msg)
			}
		} else {
			log.PlainWarn("Query result is not a vector", zap.Time("queryTime", queryTime))
		}

	}
	msg := fmt.Sprintf("%s sumResult: %f", dayStr, sumResult)
	log.PlainInfo(msg)
	return sumResult, nil
}

// getTopXTotalCount gets the top x SQLs with the highest CPU usage within a day
// Note: This function depends on the `CreatePrometheusClient` function defined elsewhere in the project.
func getTopXTotalCount(clusterId string, dayStr string, top int) ([]DigestInfo, error) {
	digestArr := make([]DigestInfo, 0)
	if top <= 0 {
		return nil, fmt.Errorf("top must be greater than 0")
	}

	pc, err := CreatePrometheusClient(context.Background(), CreatePrometheusClientReq{ClusterID: clusterId})
	if err != nil {
		return nil, err
	}

	queryTime, err := time.ParseInLocation("20060102", dayStr, time.UTC)
	if err != nil {
		return nil, fmt.Errorf("failed to parse day string: %w", err)
	}
	queryTime = queryTime.Add(24 * time.Hour)
	// Use map to store cumulative CPU time for each sql_digest
	// topRes := make(map[string]float64)

	query := fmt.Sprintf(`topk(%d, sum by (sql_digest) (sum_over_time(topsql_stmt_exec_count{instance_type="tidb", sql_digest!=""}[1d])))`, top)

	ctx, cancel := context.WithTimeout(context.Background(), 40*time.Second)

	log.PlainInfo("", zap.String("query", query), zap.Time("queryTime", queryTime))
	result, err := pc.Query(ctx, query, queryTime)
	cancel() // Release resources in a timely manner
	if err != nil {
		log.PlainError("Query failed", zap.Time("queryTime", queryTime), zap.Error(err))
		return digestArr, err // If query for an hour fails, log it and continue to the next hour
	}

	if result == nil {
		log.PlainWarn("Query result is nil", zap.Time("queryTime", queryTime))
		return digestArr, err
	}

	if vector, ok := result.(model.Vector); ok {
		for _, sample := range vector {
			// Get sql_digest from Metric labels
			digest := string(sample.Metric[model.LabelName("sql_digest")])
			// Get the sample value (CPU time)
			value := float64(sample.Value)
			// Accumulate in map
			digestArr = append(digestArr, DigestInfo{
				Digest: digest,
				Value:  value,
			})
		}
	} else {
		log.PlainWarn("Query result is not a vector", zap.Time("queryTime", queryTime))
	}

	return digestArr, nil
}

// func GetTotalSQLCount(clusterId string, dayStr string, sqlDigest ...string) (float64, error) {
// 	pc, err := CreatePrometheusClient(clusterId)
// 	if err != nil {
// 		return 0, err
// 	}

// 	startTime, err := time.ParseInLocation("20060102", dayStr, time.UTC)
// 	if err != nil {
// 		return 0, fmt.Errorf("failed to parse day string: %w", err)
// 	}

// 	var totalCount float64
// 	var query string

// 	// Define the smaller interval and query range
// 	interval := 10 * time.Minute
// 	queryRange := "[10m]"

// 	// Based on whether sqlDigest is provided, construct the appropriate query.
// 	if len(sqlDigest) > 0 && sqlDigest[0] != "" {
// 		query = fmt.Sprintf(`sum(increase(topsql_stmt_exec_count{sql_digest="%s"}%s))`, sqlDigest[0], queryRange)
// 	} else {
// 		query = fmt.Sprintf(`sum(increase(topsql_stmt_exec_count%s))`, queryRange)
// 	}

// 	// Loop through the day in 10-minute intervals. A day has 24 * 6 = 144 intervals.
// 	for i := 0; i < 24*6; i++ {
// 		// The query time should be the end of each interval.
// 		queryTime := startTime.Add(time.Duration(i+1) * interval)
// 		ctx, cancel := context.WithTimeout(context.Background(), 40*time.Second)

// 		log.PlainInfo("Querying 10-minute execution count",
// 			zap.Time("queryTime", queryTime),
// 			zap.String("query", query))

// 		result, err := pc.Query(ctx, query, queryTime)
// 		cancel() // Release resources in a timely manner.
// 		if err != nil {
// 			log.PlainError("Query failed", zap.Time("queryTime", queryTime), zap.Error(err))
// 			continue // If a query for one interval fails, log it and continue to the next.
// 		}

// 		if result == nil {
// 			log.PlainWarn("Query result is nil", zap.Time("queryTime", queryTime))
// 			continue
// 		}

// 		// The result of a sum query is a vector with a single element.
// 		if vector, ok := result.(model.Vector); ok {
// 			if len(vector) > 0 {
// 				totalCount += float64(vector[0].Value)
// 			}
// 		} else {
// 			log.PlainWarn("Query result is not a vector", zap.Time("queryTime", queryTime))
// 		}
// 	}

// 	return totalCount, nil
// }

// func GetSum24HoursValue(clusterId string, dayStr string, promSql string) (float64, error) {
// 	pc, err := CreatePrometheusClient(clusterId)
// 	if err != nil {
// 		return 0, err
// 	}

// 	startTime, err := time.ParseInLocation("20060102", dayStr, time.UTC)
// 	if err != nil {
// 		return 0, fmt.Errorf("failed to parse day string: %w", err)
// 	}

// 	var sumResult float64
// 	// var query string

// 	// Define the smaller interval and query range

// 	// Loop through the day in 10-minute intervals. A day has 24 * 6 = 144 intervals.
// 	for i := 0; i < 24; i++ {
// 		// The query time should be the end of each interval.
// 		queryTime := startTime.Add(time.Duration(i+1) * time.Hour)
// 		ctx, cancel := context.WithTimeout(context.Background(), 40*time.Second)

// 		log.PlainInfo("Querying 1 hour execution count",
// 			zap.Time("queryTime", queryTime),
// 			zap.String("query", promSql))

// 		result, err := pc.Query(ctx, promSql, queryTime)
// 		cancel() // Release resources in a timely manner.
// 		if err != nil {
// 			log.PlainError("Query failed", zap.Time("queryTime", queryTime), zap.Error(err))
// 			continue // If a query for one interval fails, log it and continue to the next.
// 		}

// 		if result == nil {
// 			log.PlainWarn("Query result is nil", zap.Time("queryTime", queryTime))
// 			continue
// 		}

// 		// The result of a sum query is a vector with a single element.
// 		if vector, ok := result.(model.Vector); ok {
// 			if len(vector) > 0 {
// 				sumResult += float64(vector[0].Value)
// 			}
// 		} else {
// 			log.PlainWarn("Query result is not a vector", zap.Time("queryTime", queryTime))
// 		}
// 	}

// 	return sumResult, nil
// }

// GetSumValueForDay calculates the sum of a PromQL query over a given day,
// splitting the day into intervals (e.g., 30m, 1h) to avoid Prometheus query limits.
// dayStr is in format "20060102", interval is the time slice per query.
// promSql must contain the full PromQL expression with the desired range selector (e.g. [30m], [1h]).
// func GetSumValueForDay(clusterId, dayStr, promSql string, interval time.Duration) (float64, error) {
// 	pc, err := CreatePrometheusClient(clusterId)
// 	if err != nil {
// 		return 0, err
// 	}

// 	startTime, err := time.ParseInLocation("20060102", dayStr, time.UTC)
// 	if err != nil {
// 		return 0, fmt.Errorf("failed to parse day string: %w", err)
// 	}
// 	log.PlainInfo("Querying interval execution count", zap.Duration("interval", interval), zap.String("query", promSql))
// 	var sumResult float64
// 	// Calculate how many intervals fit into 24h
// 	steps := int((24 * time.Hour) / interval)

// 	for i := 0; i < steps; i++ {
// 		// Query time is the end of each interval
// 		queryTime := startTime.Add(time.Duration(i+1) * interval)
// 		ctx, cancel := context.WithTimeout(context.Background(), 40*time.Second)

// 		result, err := pc.Query(ctx, promSql, queryTime)
// 		cancel()
// 		if err != nil {
// 			log.PlainError("Query failed", zap.Time("queryTime", queryTime), zap.Error(err))
// 			continue
// 		}
// 		if result == nil {
// 			log.PlainWarn("Query result is nil", zap.Time("queryTime", queryTime))
// 			continue
// 		}

// 		if vector, ok := result.(model.Vector); ok {
// 			if len(vector) > 0 {
// 				val := float64(vector[0].Value)
// 				sumResult += val
// 				msg := fmt.Sprintf("%v ， value: %f", queryTime, val)
// 				log.PlainInfo(msg)
// 			}
// 		} else {
// 			log.PlainWarn("Query result is not a vector", zap.Time("queryTime", queryTime))
// 		}

// 	}
// 	msg := fmt.Sprintf("%s sumResult: %f", dayStr, sumResult)
// 	log.PlainInfo(msg)
// 	return sumResult, nil
// }

// GetTopXCPUConsumingSQLForDay 获取一天内CPU使用量最高的Top x SQL
// 注意：此函数依赖于一个在项目中别处定义的 `CreatePrometheusClient` 函数。
func (pc *PrometheusClient) GetSqlText(clusterId string, digest string, timeRange dto.TimeRange) (string, error) {
	digest = strings.ToUpper(digest)
	sql := ""
	// pc, err := CreatePrometheusClient(context.Background(), CreatePrometheusClientReq{ClusterID: clusterId})
	// if err != nil {
	// 	return sql, err
	// }
	startTime := timeRange.Start
	endTime := timeRange.End

	// startTime, err := time.ParseInLocation("20060102", dayStr, time.UTC)
	// if err != nil {
	// 	return sql, fmt.Errorf("failed to parse day string: %w", err)
	// }
	// endTime := startTime.Add(time.Duration(48) * time.Hour)
	query := fmt.Sprintf(`topk(1, topsql_sql_meta{cluster_id="%s",sql_digest="%s"})`, clusterId, digest)

	ctx, cancel := context.WithTimeout(context.Background(), 40*time.Second)
	log.PlainInfo("Querying GetSqlText", zap.Time("queryTime", startTime))

	step := 6 * time.Hour
	result, err := pc.QueryRange(ctx, query, startTime, endTime, step)
	cancel() // 及时释放资源
	if err != nil {
		log.PlainError("Query failed", zap.Time("queryTime", startTime), zap.Error(err))
		return sql, err
	}

	if result == nil {
		log.PlainWarn("Query result is nil", zap.Time("queryTime", startTime))
		return sql, err
	}

	// if t, ok := result.(model.Value); ok {

	// 	a:=t.Type()
	// 	log.PlainInfo("Query result type", zap.String("type", a.String()))

	if mtx, ok := result.(model.Matrix); ok {
		if len(mtx) == 0 {
			log.PlainWarn("Matrix result is empty", zap.Time("queryTime", startTime))
			return sql, nil
		}
		// 取第一条时间序列的 metric 标签
		first := mtx[0]
		if val, ok := first.Metric[model.LabelName("normalized_sql")]; ok {
			return string(val), nil
		}
		// Fallback: traverse all series to find normalized_sql
		for _, ss := range mtx {
			if val, ok := ss.Metric[model.LabelName("normalized_sql")]; ok {
				return string(val), nil
			}
		}
		log.PlainWarn("normalized_sql label not found in matrix result", zap.Time("queryTime", startTime))
		return sql, nil

	} else {
		log.PlainWarn("Query result is not a vector", zap.Time("queryTime", startTime))
	}

	return sql, nil
}

// func GetTopXCountInfo( clusterId string, dayStr string, top int) (TopCountResult, error) {
// 	pc, err := CreatePrometheusClient(ctx, CreatePrometheusClientReq{ClusterID: clusterId})
// 	if err != nil {
// 		return TopCountResult{}, err
// 	}
// 	digestArr := make([]CountDigest, 0)
// 	// get top x sql by count
// 	res, err := getTopXTotalCount(clusterId, dayStr, top)

// 	if err != nil {
// 		return TopCountResult{}, err
// 	}

// 	var totalCount float64
// 	var totalCpuSum float64

// 	//
// 	for i := range res {
// 		digest := res[i].Digest
// 		timeRange, err := utils.GetRangeTime(dayStr)
// 		if err != nil {
// 			return TopCountResult{}, err
// 		}
// 		sql, err := GetSqlText(clusterId, digest, timeRange)

// 		if err != nil {
// 			log.PlainError("GetSqlText failed", zap.String("digest", digest), zap.Error(err))
// 			return TopCountResult{}, err
// 		}

// 		queryTime, err := time.Parse(time.RFC3339, dayStr)
// 		if err != nil {
// 			return TopCountResult{}, err
// 		}
// 		// latency, err := GetAvgLatency(clusterId, queryTime, "1d", digest)
// 		// if err != nil {
// 		// 	log.PlainError("GetAvgLatency failed", zap.String("digest", digest), zap.Error(err))
// 		// 	return TopCountResult{}, err
// 		// }
// 		// qps, err := GetQps(clusterId, dayStr, digest)
// 		// if err != nil {
// 		// 	log.PlainError("GetQps failed", zap.String("digest", digest), zap.Error(err))
// 		// 	return TopSqlResult{}, err
// 		// }
// 		cpu, err := GetTotalCpu(clusterId, dayStr, digest)
// 		if err != nil {
// 			log.PlainError("GetTotalCpu failed", zap.String("digest", digest), zap.Error(err))
// 			return TopCountResult{}, err
// 		}
// 		maxCpu, err := GetMaxCpu(clusterId, dayStr, digest)
// 		if err != nil {
// 			log.PlainError("GetMaxCpu failed", zap.String("digest", digest), zap.Error(err))
// 			return TopCountResult{}, err
// 		}
// 		avgCpu, err := GetAvgCpu(clusterId, dayStr, digest)
// 		if err != nil {
// 			log.PlainError("GetAvgCpu failed", zap.String("digest", digest), zap.Error(err))
// 			return TopCountResult{}, err
// 		}
// 		// h
// 		totalCpu := cpu / 1000 / 60 / 60
// 		//（mil）
// 		count := math.Round((res[i].Value/1e6)*100) / 100

// 		totalCount += count
// 		totalCpuSum += totalCpu

// 		digestArr = append(digestArr, CountDigest{
// 			Digest:     digest,
// 			Sql:        sql,
// 			TotalCpu:   math.Round(totalCpu*1000) / 1000,
// 			Count:      count,
// 			// AvgLatency: math.Round(latency*100) / 100,
// 			// Qps:        qps,
// 			AvgCpu: math.Round(avgCpu*100) / 100,
// 			MaxCpu: math.Round(maxCpu*100) / 100,
// 		})
// 	}
// 	// sory by  Count desc
// 	sort.Slice(digestArr, func(i, j int) bool { return digestArr[i].TotalCpu > digestArr[j].TotalCpu })

// 	result := TopCountResult{
// 		Day:        dayStr,
// 		ClusterId:  clusterId,
// 		DataDesc:   fmt.Sprintf("top %d sql by count", top),
// 		TotalCount: math.Round(totalCount*1000) / 1000,
// 		TotalCpu:   math.Round(totalCpuSum*1000) / 1000,
// 		TopSqls:    digestArr,
// 		Unit:       []string{"count:mil", "cpu:h", "latency:ms"},
// 	}

// 	// log.PlainInfo("GetTopXTotalCount for 24 hours", zap.Any("res", res))

// 	return result, nil
// }

// func GetTopXCountInfo2(clusterId string, dayStr string, top int) (TopCountResult, error) {
// 	digestArr := make([]CountDigest, 0)
// 	// get top x sql by count
// 	res, err := getTopXTotalCount(clusterId, dayStr, top)

// 	if err != nil {
// 		return TopCountResult{}, err
// 	}

// 	var totalCount float64
// 	var totalCpuSum float64

// 	//
// 	for i := range res {
// 		digest := res[i].Digest
// 		timeRange, err := utils.GetRangeTime(dayStr)
// 		if err != nil {
// 			return TopCountResult{}, err
// 		}
// 		sql, err := GetSqlText(clusterId, digest, timeRange)

// 		if err != nil {
// 			log.PlainError("GetSqlText failed", zap.String("digest", digest), zap.Error(err))
// 			return TopCountResult{}, err
// 		}

// 		cpu, err := GetTotalCpu(clusterId, dayStr, digest)
// 		if err != nil {
// 			log.PlainError("GetTotalCpu failed", zap.String("digest", digest), zap.Error(err))
// 			return TopCountResult{}, err
// 		}

// 		// h
// 		totalCpu := cpu / 1000 / 60 / 60
// 		//（mil）
// 		count := math.Round((res[i].Value/1e6)*100) / 100

// 		totalCount += count
// 		totalCpuSum += totalCpu

// 		digestArr = append(digestArr, CountDigest{
// 			Digest:   digest,
// 			Sql:      sql,
// 			TotalCpu: math.Round(totalCpu*1000) / 1000,
// 			Count:    count,
// 		})
// 	}
// 	// sory by  Count desc
// 	sort.Slice(digestArr, func(i, j int) bool { return digestArr[i].TotalCpu > digestArr[j].TotalCpu })

// 	result := TopCountResult{
// 		Day:        dayStr,
// 		ClusterId:  clusterId,
// 		DataDesc:   fmt.Sprintf("top %d sql by count", top),
// 		TotalCount: math.Round(totalCount*1000) / 1000,
// 		TotalCpu:   math.Round(totalCpuSum*1000) / 1000,
// 		TopSqls:    digestArr,
// 		Unit:       []string{"count:mil", "cpu:h"},
// 	}

// 	// log.PlainInfo("GetTopXTotalCount for 24 hours", zap.Any("res", res))

// 	return result, nil
// }

// // GetTopXCountSQL get top 20 sql by count [prod]
// func GetTopXCountSQL(clusterId string, days []string, top int) {
// 	topSqlResult := make([]TopCountResult, 0)
// 	for _, day := range days {
// 		res, err := GetTopXCountInfo(clusterId, day, top)
// 		if err != nil {
// 			log.PlainError("GetTopXCountInfo failed", zap.String("day", day), zap.Error(err))
// 		}
// 		topSqlResult = append(topSqlResult, res)
// 	}
// 	topSqlResult = DiffTopCount(topSqlResult)

// 	// cntDiff > 50%
// 	err := SaveDiffCntDigests(topSqlResult, top, days[0], 50.0)
// 	if err != nil {
// 		log.PlainError("WriteFilteredCountDigest failed", zap.Error(err))
// 		panic(err)
// 	}

// 	b, err := json.MarshalIndent(topSqlResult, "", "  ")
// 	if err != nil {
// 		log.PlainError("json marshal failed", zap.Error(err))
// 		panic(err)
// 	}
// 	str := string(b)
// 	if err := os.MkdirAll("output", 0755); err != nil {
// 		log.PlainError("create output dir failed", zap.Error(err))
// 		panic(err)
// 	}
// 	fileName := fmt.Sprintf("output/Top%dCount_%s.json", top, days[0])
// 	if err := os.WriteFile(fileName, []byte(str), 0644); err != nil {
// 		log.PlainError("write output file failed", zap.Error(err))
// 		panic(err)
// 	}
// 	fmt.Println(str)
// }

// func GetTopXCpuSQL(clusterId string, days []string, top int) {
// 	topSqlResult := make([]TopCpuResult, 0)
// 	for _, day := range days {
// 		res, err := GetTopXCPUInfo(clusterId, day, top)
// 		if err != nil {
// 			log.PlainError("GetTopXCPUInfo failed", zap.String("day", day), zap.Error(err))
// 		}
// 		topSqlResult = append(topSqlResult, res)
// 	}
// 	topSqlResult = DiffTopCpu(topSqlResult)

// 	// cpuDiff > 50%
// 	err := SaveDiffCpuDigests(topSqlResult, top, days[0], 20.0)
// 	if err != nil {
// 		log.PlainError("WriteFilteredCpuDigest failed", zap.Error(err))
// 		panic(err)
// 	}

// 	b, err := json.MarshalIndent(topSqlResult, "", "  ")
// 	if err != nil {
// 		log.PlainError("json marshal failed", zap.Error(err))
// 		panic(err)
// 	}
// 	str := string(b)
// 	if err := os.MkdirAll("output", 0755); err != nil {
// 		log.PlainError("create output dir failed", zap.Error(err))
// 		panic(err)
// 	}
// 	fileName := fmt.Sprintf("output/Top%dCpu_%s.json", top, days[0])
// 	if err := os.WriteFile(fileName, []byte(str), 0644); err != nil {
// 		log.PlainError("write output file failed", zap.Error(err))
// 		panic(err)
// 	}

// 	fmt.Println(str)
// }

func DiffTopCount(results []TopCountResult) []TopCountResult {
	for i := 1; i < len(results); i++ {
		prev := results[i-1]
		curr := &results[i]
		prevMap := make(map[string]CountDigest)
		for _, d := range prev.TopSqls {
			prevMap[d.Digest] = d
		}
		for j := range curr.TopSqls {
			d := &curr.TopSqls[j]
			if prevDigest, ok := prevMap[d.Digest]; ok {
				if prevDigest.Count != 0 {
					rate := (d.Count - prevDigest.Count) / prevDigest.Count * 100
					d.CntDiff = fmt.Sprintf("%.2f%%", rate)
				} else {
					d.CntDiff = "0.00%"
				}
				d.IsNew = false
			} else {
				// d.CntDiff = "new"
				d.IsNew = true
			}
		}
	}
	// 第一组全部为新
	// for j := range results[0].TopSqls {
	// 	results[0].TopSqls[j].CntDiff = "new"
	// 	results[0].TopSqls[j].IsNew = true
	// }
	return results
}

func DiffTopCpu(results []TopCpuResult) []TopCpuResult {
	for i := 1; i < len(results); i++ {
		prev := results[i-1]
		curr := &results[i]
		prevMap := make(map[string]CpuDigest)
		for _, d := range prev.TopSqls {
			prevMap[d.Digest] = d
		}
		for j := range curr.TopSqls {
			d := &curr.TopSqls[j]
			if prevDigest, ok := prevMap[d.Digest]; ok {
				if prevDigest.TotalCpu != 0 {
					rate := (d.TotalCpu - prevDigest.TotalCpu) / prevDigest.TotalCpu * 100
					d.CpuDiff = fmt.Sprintf("%.2f%%", rate)
				} else {
					d.CpuDiff = "0.00%"
				}
				d.IsNew = false
			} else {
				// d.CpuDiff = "new"
				d.IsNew = true
			}
		}
	}
	// for j := range results[0].TopSqls {
	// 	results[0].TopSqls[j].CpuDiff = "new"
	// 	results[0].TopSqls[j].IsNew = true
	// }
	return results
}

// func GetTopXCPUInfo(clusterId string, dayStr string, top int) (TopCpuResult, error) {

// 	var topTotalCount float64
// 	var topTotalCpuSum float64
// 	var allCpuDay float64
// 	digestArr := make([]CpuDigest, 0)
// 	allCpuDay, err := GetClusterCpuByDay(clusterId, dayStr)
// 	if err != nil {
// 		return TopCpuResult{}, err
// 	}

// 	// get top x sql by cpu time
// 	res, err := getTopXTotalCPUTime(clusterId, dayStr, top)
// 	if err != nil {
// 		return TopCpuResult{}, err
// 	}

// 	//
// 	for i := range res {
// 		digest := res[i].Digest
// 		timeRange, err := utils.GetRangeTime(dayStr)
// 		if err != nil {
// 			return TopCpuResult{}, err
// 		}
// 		sql, err := GetSqlText(clusterId, digest, timeRange)

// 		if err != nil {
// 			log.PlainError("GetSqlText failed", zap.String("digest", digest), zap.Error(err))
// 			return TopCpuResult{}, err
// 		}
// 		queryTime, err := time.Parse(time.RFC3339, dayStr)
// 		if err != nil {
// 			return TopCpuResult{}, err
// 		}
// 		latency, err := GetAvgLatency(clusterId, queryTime, "1d", digest)
// 		if err != nil {
// 			log.PlainError("GetAvgLatency failed", zap.String("digest", digest), zap.Error(err))
// 			return TopCpuResult{}, err
// 		}

// 		totalCpu, err := GetTotalCpu(clusterId, dayStr, digest)
// 		if err != nil {
// 			log.PlainError("GetTotalCpu failed", zap.String("digest", digest), zap.Error(err))
// 			return TopCpuResult{}, err
// 		}
// 		maxCpu, err := GetMaxCpu(clusterId, dayStr, digest)
// 		if err != nil {
// 			log.PlainError("GetMaxCpu failed", zap.String("digest", digest), zap.Error(err))
// 			return TopCpuResult{}, err
// 		}
// 		avgCpu, err := GetAvgCpu(clusterId, dayStr, digest)
// 		if err != nil {
// 			log.PlainError("GetAvgCpu failed", zap.String("digest", digest), zap.Error(err))
// 			return TopCpuResult{}, err
// 		}
// 		// tikbAvgCpuIn5Min, err := GetTikvAvgCpuIn5Min(clusterId, dayStr, digest)
// 		// if err != nil {
// 		// 	log.PlainError("tikbAvgCpuIn5Min failed", zap.String("digest", digest), zap.Error(err))
// 		// 	return TopCpuResult{}, err
// 		// }

// 		// tidbAvgCpuIn5Min, err := GetTiDBAvgCpuIn5Min(clusterId, dayStr, digest)
// 		// if err != nil {
// 		// 	log.PlainError("tikbAvgCpuIn5Min failed", zap.String("digest", digest), zap.Error(err))
// 		// 	return TopCpuResult{}, err
// 		// }

// 		// isInSlowQuery, err := slowquery.IsInSlowQuery(clusterId, dayStr, digest)
// 		// if err != nil {
// 		// 	log.PlainError("isInSlowQuery failed", zap.String("digest", digest), zap.Error(err))
// 		// 	return TopCpuResult{}, err
// 		// }

// 		// s
// 		maxCpu = maxCpu / 1000
// 		avgCpu = avgCpu / 1000

// 		// min
// 		// tikbAvgCpuIn5Min = tikbAvgCpuIn5Min / 1000 / 60
// 		// tidbAvgCpuIn5Min = tidbAvgCpuIn5Min / 1000 / 60
// 		// hour
// 		totalCpu = totalCpu / 1000 / 60 / 60
// 		// （mil）
// 		count := math.Round((res[i].Value/1e6)*100) / 100

// 		topTotalCount += count
// 		topTotalCpuSum += totalCpu

// 		digestArr = append(digestArr, CpuDigest{
// 			Digest:        digest,
// 			Sql:           sql,
// 			TotalCpu:      math.Round(totalCpu*100) / 100,
// 			Count:         count,
// 			AvgLatency:    math.Round(latency*100) / 100,
// 			AvgCpu:        math.Round(avgCpu*100) / 100,
// 			MaxCpu:        math.Round(maxCpu*100) / 100,
// 			CpuDayPercent: math.Round((totalCpu/allCpuDay)*10000) / 100,
// 			// IsInSlowQuery:    isInSlowQuery,
// 			// TikvAvgCpuIn5Min: math.Round(tikbAvgCpuIn5Min*100) / 100,
// 			// TidbAvgCpuIn5Min: math.Round(tidbAvgCpuIn5Min*100) / 100,
// 		})
// 	}
// 	sort.Slice(digestArr, func(i, j int) bool { return digestArr[i].TotalCpu > digestArr[j].TotalCpu })

// 	result := TopCpuResult{
// 		Day:           dayStr,
// 		ClusterId:     clusterId,
// 		DataDesc:      fmt.Sprintf("top %d sql by Cpu time", top),
// 		TopTotalCount: math.Round(topTotalCount*1000) / 1000,
// 		TopTotalCpu:   math.Round(topTotalCpuSum*1000) / 1000,
// 		AllSqlCpu:     math.Round(allCpuDay*1000) / 1000,
// 		TopSqls:       digestArr,
// 		Unit:          []string{"count:mil", "cpu:h", "latency:ms", "AvgCpuIn5Min:min"},
// 	}

// 	// log.PlainInfo("GetTopXTotalCount for 24 hours", zap.Any("res", res))
// 	// 返回 TopSqlResult 结构体
// 	return result, nil
// }
// func GetTop20CpuInfo(clusterId string, dayStr string, top int) (TopCpuResult, error) {

// 	var topTotalCount float64
// 	var topTotalCpuSum float64
// 	var allCpuDay float64
// 	digestArr := make([]CpuDigest, 0)
// 	allCpuDay, err := GetClusterCpuByDay(clusterId, dayStr)
// 	if err != nil {
// 		return TopCpuResult{}, err
// 	}

// 	// get top x sql by cpu time
// 	res, err := getTopXTotalCPUTime(clusterId, dayStr, top)
// 	if err != nil {
// 		return TopCpuResult{}, err
// 	}

// 	//
// 	for i := range res {
// 		digest := res[i].Digest
// 		timeRange, err := utils.GetRangeTime(dayStr)
// 		if err != nil {
// 			return TopCpuResult{}, err
// 		}
// 		sql, err := GetSqlText(clusterId, digest, timeRange)

// 		if err != nil {
// 			log.PlainError("GetSqlText failed", zap.String("digest", digest), zap.Error(err))
// 			return TopCpuResult{}, err
// 		}
// 		// latency, err := GetAvgLatency(clusterId, dayStr, digest)
// 		// if err != nil {
// 		// 	log.PlainError("GetAvgLatency failed", zap.String("digest", digest), zap.Error(err))
// 		// 	return TopCpuResult{}, err
// 		// }

// 		totalCpu, err := GetTotalCpu(clusterId, dayStr, digest)
// 		if err != nil {
// 			log.PlainError("GetTotalCpu failed", zap.String("digest", digest), zap.Error(err))
// 			return TopCpuResult{}, err
// 		}

// 		// hour
// 		totalCpu = totalCpu / 1000 / 60 / 60
// 		// （mil）
// 		count := math.Round((res[i].Value/1e6)*100) / 100

// 		topTotalCount += count
// 		topTotalCpuSum += totalCpu

// 		digestArr = append(digestArr, CpuDigest{
// 			Digest:        digest,
// 			Sql:           sql,
// 			TotalCpu:      math.Round(totalCpu*100) / 100,
// 			Count:         count,
// 			CpuDayPercent: math.Round((totalCpu/allCpuDay)*10000) / 100,
// 			// IsInSlowQuery:    isInSlowQuery,
// 			// TikvAvgCpuIn5Min: math.Round(tikbAvgCpuIn5Min*100) / 100,
// 			// TidbAvgCpuIn5Min: math.Round(tidbAvgCpuIn5Min*100) / 100,
// 		})
// 	}
// 	sort.Slice(digestArr, func(i, j int) bool { return digestArr[i].TotalCpu > digestArr[j].TotalCpu })

// 	result := TopCpuResult{
// 		Day:           dayStr,
// 		ClusterId:     clusterId,
// 		DataDesc:      fmt.Sprintf("top %d sql by Cpu time", top),
// 		TopTotalCount: math.Round(topTotalCount*1000) / 1000,
// 		TopTotalCpu:   math.Round(topTotalCpuSum*1000) / 1000,
// 		AllSqlCpu:     math.Round(allCpuDay*1000) / 1000,
// 		TopSqls:       digestArr,
// 		Unit:          []string{"count:mil", "cpu:h", "latency:ms", "AvgCpuIn5Min:min"},
// 	}

// 	// log.PlainInfo("GetTopXTotalCount for 24 hours", zap.Any("res", res))
// 	// 返回 TopSqlResult 结构体
// 	return result, nil
// }

func GetClusterCpuByDay(clusterId string, dayStr string) (float64, error) {
	queryTime := GetDayQueryTime(dayStr)
	interval := "1d"
	allCpuDay, err := GetAllSQLTotalCPUTime(clusterId, interval, queryTime, CpuTimeUnitHour)
	if err != nil {
		return 0, err
	}
	return allCpuDay, nil
}

// func GetTopXCPUInfo2(clusterId string, dayStr string, top int) (TopCpuResult, error) {

// 	var topTotalCount float64
// 	var topTotalCpuSum float64
// 	var allCpuDay float64
// 	digestArr := make([]CpuDigest, 0)
// 	queryTime := GetDayQueryTime(dayStr)
// 	interval := "1d"
// 	allCpuDay, err := GetAllSQLTotalCPUTime(clusterId, interval, queryTime, CpuTimeUnitHour)
// 	if err != nil {
// 		return TopCpuResult{}, err
// 	}

// 	// get top x sql by cpu time
// 	res, err := getTopXTotalCPUTime(clusterId, dayStr, top)
// 	if err != nil {
// 		return TopCpuResult{}, err
// 	}

// 	//
// 	for i := range res {
// 		digest := res[i].Digest
// 		timeRange, err := utils.GetRangeTime(dayStr)
// 		if err != nil {
// 			return TopCpuResult{}, err
// 		}
// 		sql, err := GetSqlText(clusterId, digest, timeRange)

// 		if err != nil {
// 			log.PlainError("GetSqlText failed", zap.String("digest", digest), zap.Error(err))
// 			return TopCpuResult{}, err
// 		}

// 		totalCpu, err := GetTotalCpu(clusterId, dayStr, digest)
// 		if err != nil {
// 			log.PlainError("GetTotalCpu failed", zap.String("digest", digest), zap.Error(err))
// 			return TopCpuResult{}, err
// 		}

// 		// hour
// 		totalCpu = totalCpu / 1000 / 60 / 60
// 		// （mil）
// 		count := math.Round((res[i].Value/1e6)*100) / 100

// 		topTotalCount += count
// 		topTotalCpuSum += totalCpu

// 		digestArr = append(digestArr, CpuDigest{
// 			Digest:   digest,
// 			Sql:      sql,
// 			TotalCpu: math.Round(totalCpu*100) / 100,
// 			Count:    count,

// 			CpuDayPercent: math.Round((totalCpu/allCpuDay)*10000) / 100,
// 		})
// 	}
// 	sort.Slice(digestArr, func(i, j int) bool { return digestArr[i].TotalCpu > digestArr[j].TotalCpu })

// 	result := TopCpuResult{
// 		Day:           dayStr,
// 		ClusterId:     clusterId,
// 		DataDesc:      fmt.Sprintf("top %d sql by Cpu time", top),
// 		TopTotalCount: math.Round(topTotalCount*1000) / 1000,
// 		TopTotalCpu:   math.Round(topTotalCpuSum*1000) / 1000,
// 		AllSqlCpu:     math.Round(allCpuDay*1000) / 1000,
// 		TopSqls:       digestArr,
// 		Unit:          []string{"count:mil", "cpu:h"},
// 	}

// 	// log.PlainInfo("GetTopXTotalCount for 24 hours", zap.Any("res", res))
// 	// 返回 TopSqlResult 结构体
// 	return result, nil
// }

func filterCountDigest(results []TopCountResult, threshold float64) ([]CountDigest, error) {
	filtered := make([]CountDigest, 0)
	for _, dayResult := range results {
		for _, d := range dayResult.TopSqls {
			if strings.HasSuffix(d.CntDiff, "%") {
				val := strings.TrimSuffix(d.CntDiff, "%")
				f, err := strconv.ParseFloat(val, 64)
				if err == nil && f > threshold {
					filtered = append(filtered, d)
				}
			}
		}
	}
	return filtered, nil
}

func filterCpuDigest(results []TopCpuResult, threshold float64) ([]CpuDigest, error) {
	filtered := make([]CpuDigest, 0)
	for _, dayResult := range results {
		for _, d := range dayResult.TopSqls {
			if strings.HasSuffix(d.CpuDiff, "%") {
				val := strings.TrimSuffix(d.CpuDiff, "%")
				f, err := strconv.ParseFloat(val, 64)
				if err == nil && f > threshold {
					filtered = append(filtered, d)
				}
			}
		}
	}
	return filtered, nil
}

func SaveDiffCntDigests(results []TopCountResult, top int, day string, threshold float64) error {
	diffDigests := make([]CountDigest, 0)
	newDigests := make([]CountDigest, 0)
	for _, dayResult := range results {
		for _, d := range dayResult.TopSqls {
			if strings.HasSuffix(d.CntDiff, "%") {
				val := strings.TrimSuffix(d.CntDiff, "%")
				f, err := strconv.ParseFloat(val, 64)
				if err == nil && f > threshold {
					diffDigests = append(diffDigests, d)
				}
			}
			if d.IsNew {
				newDigests = append(newDigests, d)
			}
		}
	}
	output := map[string]interface{}{
		"Day":                         day,
		"DataDesc":                    "diff count over threshold or new digest",
		"ClusterId":                   results[0].ClusterId,
		"DiffCntOverThresholdTotal":   len(diffDigests),
		"NewDigestTotal":              len(newDigests),
		"Unit":                        []string{"count:mil", "cpu:h", "latency:ms"},
		"DiffCntOverThresholdDigests": diffDigests,
		"NewDigests":                  newDigests,
	}
	b, err := json.MarshalIndent(output, "", "  ")
	if err != nil {
		return err
	}
	fileName := fmt.Sprintf("output/Top%dCountDiffOver%.0f_%s.json", top, threshold, day)
	return os.WriteFile(fileName, b, 0644)
}

func SaveDiffCpuDigests(results []TopCpuResult, top int, day string, threshold float64) error {
	diffDigests := make([]CpuDigest, 0)
	newDigests := make([]CpuDigest, 0)
	for _, dayResult := range results {
		for _, d := range dayResult.TopSqls {
			if strings.HasSuffix(d.CpuDiff, "%") {
				val := strings.TrimSuffix(d.CpuDiff, "%")
				f, err := strconv.ParseFloat(val, 64)
				if err == nil && f > threshold {
					diffDigests = append(diffDigests, d)
				}
			}
			if d.IsNew {
				newDigests = append(newDigests, d)
			}
		}
	}
	output := map[string]interface{}{
		"Day":                         day,
		"DataDesc":                    "diff cpu over threshold or new digest",
		"ClusterId":                   results[0].ClusterId,
		"DiffCpuOverThresholdTotal":   len(diffDigests),
		"NewDigestTotal":              len(newDigests),
		"Unit":                        []string{"count:mil", "cpu:h", "latency:ms"},
		"DiffCpuOverThresholdDigests": diffDigests,
		"NewDigests":                  newDigests,
	}
	b, err := json.MarshalIndent(output, "", "  ")
	if err != nil {
		return err
	}
	fileName := fmt.Sprintf("output/Top%dCpuDiffOver%.0f_%s.json", top, threshold, day)
	return os.WriteFile(fileName, b, 0644)
}

func (pcClient *PrometheusClient) GetTopSqlCount(clusterId string, digest string, queryTime time.Time, interval string) (float64, error) {
	digest = strings.ToUpper(digest)
	query := fmt.Sprintf(`sum(sum_over_time(topsql_stmt_exec_count{cluster_id="%s",instance_type="tidb",sql_digest="%s"}[%s]))`, clusterId, digest, interval)
	return GetOneValueByTime(clusterId, queryTime, query)
}

func (pcClient *PrometheusClient) GetDigestsCount(clusterId string, queryTime time.Time, interval string, sqlDigests []string) (map[string]TopSqlDigest, error) {

	digestsRegex := buildDigestRegex(sqlDigests)
	query := fmt.Sprintf(`sum by (sql_digest)(sum_over_time(topsql_stmt_exec_count{cluster_id="%s",instance_type="tidb",sql_digest=~"%s"}[%s]))`, clusterId, digestsRegex, interval)

	digestArray, err := pcClient.GetDigestArray(clusterId, queryTime, query)
	if err != nil {
		return nil, err
	}
	digestMap := make(map[string]TopSqlDigest, len(digestArray))
	for _, v := range digestArray {
		digestMap[v.Digest] = TopSqlDigest{
			Digest: v.Digest,
			Value:  v.Value,
		}
	}
	log.PlainInfo("GetDigestsCount", zap.String("query", query), zap.Any("digestArray", digestArray))
	return digestMap, nil

}

/*
	curl -G \
	      'https://www.ds.us-east-1.aws.observability.tidbcloud.com/internal/metrics/ed869d81-91c5-4a87-a28d-20795c738d3f/api/v1/query' \
	      --data-urlencode 'query=topk(100, sum by (sql_digest) (sum_over_time(topsql_stmt_exec_count:sum_over_time_5m:sum_sql_digest_instance_type{instance_type="tidb",sql_digest!=""}[1d:5m])))'  \
	      --data-urlencode 'time=2025-11-26T10:00:00Z' |jq -r
*/
func GetTopXCount(clusterId string, interval string, top int, queryTime time.Time) ([]TopSqlDigest, error) {
	if top <= 0 {
		return nil, fmt.Errorf("top must be greater than 0")
	}

	query := fmt.Sprintf(`topk(%d, sum by (sql_digest) (sum_over_time(topsql_stmt_exec_count:sum_over_time_5m:sum_sql_digest_instance_type{sql_digest!="",instance_type="tidb"}[%s:5m])))`, top, interval)
	return GetDigestArray(clusterId, queryTime, query)
}

func (pcClient *PrometheusClient) GetTopXCount(clusterId string, interval string, top int, queryTime time.Time) ([]TopSqlDigest, error) {
	if top <= 0 {
		return nil, fmt.Errorf("top must be greater than 0")
	}

	query := fmt.Sprintf(`topk(%d, sum by (sql_digest) (sum_over_time(topsql_stmt_exec_count:sum_over_time_5m:sum_sql_digest_instance_type{sql_digest!="",instance_type="tidb"}[%s:5m])))`, top, interval)
	return pcClient.GetDigestArray(clusterId, queryTime, query)
}

/*
	curl -G \
	      'https://www.ds.us-east-1.aws.observability.tidbcloud.com/internal/metrics/ed869d81-91c5-4a87-a28d-20795c738d3f/api/v1/query' \
	      --data-urlencode 'query=topk(100, sum by (sql_digest) (sum_over_time(topsql_stmt_exec_count:sum_over_time_5m:sum_sql_digest_instance_type{instance_type="tidb",sql_digest!=""}[1d:5m])))'  \
	      --data-urlencode 'time=2025-11-26T10:00:00Z' |jq -r
*/
func GetTopXCountHistoryData(clusterId string, interval string, top int, queryTime time.Time) ([]TopSqlDigest, error) {
	if top <= 0 {
		return nil, fmt.Errorf("top must be greater than 0")
	}

	instances, err := GetTiDBInstances(clusterId, queryTime, interval)
	if err != nil {
		return nil, err
	}
	if len(instances) == 0 {
		return []TopSqlDigest{}, nil
	}

	// Aggregate top results across instances (sequentially) to avoid a single huge query.
	agg := make(map[string]float64)
	topInst := top * 2
	for _, inst := range instances {
		q := fmt.Sprintf(`topk(%d, sum by (sql_digest) (sum_over_time(topsql_stmt_exec_count{sql_digest!="", instance="%s", instance_type="tidb"}[%s])))`, topInst, inst, interval)
		arr, err := GetDigestArray(clusterId, queryTime, q)
		if err != nil {
			log.PlainError("instance query failed", zap.String("instance", inst), zap.Error(err))
			continue
		}
		for _, it := range arr {
			agg[it.Digest] += it.Value
		}
	}

	// Convert aggregation map to slice and compute global TopN
	res := make([]TopSqlDigest, 0, len(agg))
	for d, v := range agg {
		res = append(res, TopSqlDigest{Digest: d, Value: v})
	}
	sort.Slice(res, func(i, j int) bool { return res[i].Value > res[j].Value })
	if len(res) > top {
		res = res[:top]
	}
	return res, nil
}

/*
	curl -G \
	      'https://www.ds.us-east-1.aws.observability.tidbcloud.com/internal/metrics/ed869d81-91c5-4a87-a28d-20795c738d3f/api/v1/query' \
	      --data-urlencode 'query=topk(100, sum by (sql_digest) (sum_over_time(topsql_stmt_exec_count:sum_over_time_5m:sum_sql_digest_instance_type{instance_type="tidb",sql_digest!=""}[1d:5m])))'  \
	      --data-urlencode 'time=2025-11-26T10:00:00Z' |jq -r
*/
func (pcClient *PrometheusClient) GetTopXCountHistoryData(clusterId string, interval string, top int, queryTime time.Time) ([]TopSqlDigest, error) {
	if top <= 0 {
		return nil, fmt.Errorf("top must be greater than 0")
	}

	instances, err := pcClient.GetTiDBInstances(clusterId, queryTime, interval)
	if err != nil {
		return nil, err
	}
	if len(instances) == 0 {
		return []TopSqlDigest{}, nil
	}

	// Aggregate top results across instances (sequentially) to avoid a single huge query.
	agg := make(map[string]float64)
	topInst := top * 2
	for _, inst := range instances {
		q := fmt.Sprintf(`topk(%d, sum by (sql_digest) (sum_over_time(topsql_stmt_exec_count{sql_digest!="", instance="%s", instance_type="tidb"}[%s])))`, topInst, inst, interval)
		arr, err := pcClient.GetDigestArray(clusterId, queryTime, q)
		if err != nil {
			log.PlainError("instance query failed", zap.String("instance", inst), zap.Error(err))
			continue
		}
		for _, it := range arr {
			agg[it.Digest] += it.Value
		}
	}

	// Convert aggregation map to slice and compute global TopN
	res := make([]TopSqlDigest, 0, len(agg))
	for d, v := range agg {
		res = append(res, TopSqlDigest{Digest: d, Value: v})
	}
	sort.Slice(res, func(i, j int) bool { return res[i].Value > res[j].Value })
	if len(res) > top {
		res = res[:top]
	}
	return res, nil
}

/*
	curl -G \
	      'https://www.ds.us-east-1.aws.observability.tidbcloud.com/internal/metrics/ed869d81-91c5-4a87-a28d-20795c738d3f/api/v1/query' \
	      --data-urlencode 'query=topk(100, sum by (sql_digest) (sum_over_time(topsql_stmt_exec_count:sum_over_time_5m:sum_sql_digest_instance_type{instance_type="tidb",sql_digest!=""}[1d:5m])))'  \
	      --data-urlencode 'time=2025-11-26T10:00:00Z' |jq -r
*/
func GetTopXCountHistoryDataBak(clusterId string, interval string, top int, queryTime time.Time) ([]TopSqlDigest, error) {
	if top <= 0 {
		return nil, fmt.Errorf("top must be greater than 0")
	}

	instances, err := GetTiDBInstances(clusterId, queryTime, interval)
	if err != nil {
		return nil, err
	}
	instanceOr := buildLabelFilter("instance", instances)

	query := fmt.Sprintf(`topk(%d, sum by (sql_digest) (sum_over_time(topsql_stmt_exec_count{%s,sql_digest!="",instance_type="tidb"}[%s])))`, top, instanceOr, interval)
	return GetDigestArray(clusterId, queryTime, query)
}

// example1:  topk(100, sum by (sql_digest) (sum_over_time(topsql_stmt_exec_count{instance_type="tidb",sql_digest!=""}[3h])))
// example2:  topk(100, sum by (sql_digest) (sum_over_time(topsql_stmt_exec_count{instance_type="tidb",sql_digest!=""}[300m])))
// example3:  topk(100, sum by (sql_digest) (sum_over_time(topsql_stmt_exec_count{instance_type="tidb",sql_digest!=""}[72000s])))
// example3:  topk(100, sum by (sql_digest) (sum_over_time(topsql_stmt_exec_count{instance_type="tidb",sql_digest!=""}[1d])))
// / interval>1h
// GetTopXCoutDigest splits a long PromQL range window into multiple smaller windows
// and aggregates digest counts across all sub-queries.
// GetTopXCountDigest 获取时间段内执行次数最多的 SQL Digest。
//
// 参数说明:
//   - queryEndTime: 查询的【结束时间】(即通常说的 "当前时间" 或 "到现在为止")。
//   - totalDuration: 需要往前查询的总时长 (例如 3h, 1d)。
//   - top: 需要返回的前几名。
func GetTopXCountDigest(clusterId string, queryEndTime time.Time, totalDuration time.Duration, top int) ([]TopSqlDigest, error) {
	if top <= 0 {
		return nil, fmt.Errorf("top must be greater than 0")
	}

	pc, err := CreatePrometheusClient(context.Background(), CreatePrometheusClientReq{ClusterID: clusterId})
	if err != nil {
		return nil, err
	}

	// 1. 计算【起始时间】：所有时间片将从这里开始向后推移
	queryStartTime := queryEndTime.Add(-totalDuration)

	// 结果聚合容器
	totalDigestMap := make(map[string]float64)

	// 2. 扩大系数
	// 关键逻辑：分片查询 TopK 时，如果只查 Top 100，容易漏掉那些在每个分片排名第 101 但总和很高的 SQL。
	// 所以分片查询时，我们获取更多的数据 (例如 Top 500)，聚合后再取 Top 100。
	stepLimit := int(float64(top) * 1.2)

	// 3. 循环推进时间片
	elapsed := time.Duration(0)

	for elapsed < totalDuration {
		remain := totalDuration - elapsed
		step := chooseStep(remain)

		// 关键修正：
		// 分片的结束点 = 起始时间 + 已经过去的时间 + 当前步长
		// Prometheus 查询时，该时间点代表 Lookbehind 窗口的终点。
		// Example: Start=09:00, Step=1h -> SliceEndTime=10:00 -> Query Range [09:00, 10:00]
		sliceEndTime := queryStartTime.Add(elapsed + step)

		// 执行单个分片的查询并聚合
		err := queryAndAggregateStep(*pc, sliceEndTime, step, stepLimit, totalDigestMap)
		if err != nil {
			// 策略：记录错误，但继续执行后续分片，尽可能返回部分数据
			log.PlainError("failed to query partial window",
				zap.Time("slice_end_time", sliceEndTime),
				zap.Duration("step", step),
				zap.Error(err),
			)
		}

		elapsed += step
	}

	// 4. 结果排序与截断
	return sortAndLimit(totalDigestMap, top)
}

// queryAndAggregateStep 负责单个时间窗口的查询逻辑
func queryAndAggregateStep(pc PrometheusClient, evalTime time.Time, stepDuration time.Duration, limit int, resultMap map[string]float64) error {
	rangeStr := formatPromRange(stepDuration)

	// 构造查询语句
	// 注意：这里使用扩大的 limit 进行查询
	query := buildTopKCountQuery(limit, rangeStr)

	log.PlainInfo("querying slice window",
		zap.Time("eval_time", evalTime),
		zap.String("range", rangeStr),
	)

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel() // 确保资源释放

	result, err := pc.Query(ctx, query, evalTime)
	if err != nil {
		return err
	}

	vector, ok := result.(model.Vector)
	if !ok {
		// 可能是 nil 或者其他类型，视为空数据处理
		return nil
	}

	// 聚合数据
	for _, sample := range vector {
		digest := string(sample.Metric[model.LabelName("sql_digest")])
		if digest == "" {
			continue
		}
		resultMap[digest] += float64(sample.Value)
	}

	return nil
}

// sortAndLimit 负责将 Map 转换为有序 Slice 并返回前 Top N
func sortAndLimit(data map[string]float64, top int) ([]TopSqlDigest, error) {
	if len(data) == 0 {
		return nil, nil
	}

	// 1. Map 转 Slice
	slice := make([]TopSqlDigest, 0, len(data))
	for k, v := range data {
		slice = append(slice, TopSqlDigest{Digest: k, Value: v})
	}

	// 2. 降序排序
	sort.Slice(slice, func(i, j int) bool {
		return slice[i].Value > slice[j].Value
	})

	// 3. 截取 Top N
	if len(slice) < top {
		return slice, nil
	}
	return slice[:top], nil
}

// --------------- 辅助函数 ---------------

// chooseStep 选择合适的时间步长，避免单次查询压力过大
// 默认最大步长为 1h，不足 1h 则按剩余时间查
func chooseStep(remain time.Duration) time.Duration {
	if remain >= time.Hour {
		return time.Hour
	}
	return remain
}

// formatPromRange 格式化时间段为 PromQL 支持的字符串
func formatPromRange(d time.Duration) string {
	// 优先匹配整小时或整分钟，生成的 SQL 更易读
	if d%time.Hour == 0 {
		return fmt.Sprintf("%dh", int(d.Hours()))
	}
	if d%time.Minute == 0 {
		return fmt.Sprintf("%dm", int(d.Minutes()))
	}
	return fmt.Sprintf("%ds", int(d.Seconds()))
}

// buildTopKCountQuery 构造 PromQL
func buildTopKCountQuery(top int, rangeStr string) string {
	// 这里的 instance_type="tidb" 是你的业务 Tag，如果需要通用化可以作为参数传入
	return fmt.Sprintf(`topk(%d, sum by (sql_digest) (sum_over_time(topsql_stmt_exec_count{instance_type="tidb",instance=~".*db-tidb-.*",sql_digest!=""}[%s])))`, top, rangeStr)
}