package service

import (
	"context"
	"fmt"
	"net/url"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/duke-git/lancet/v2/maputil"
	"github.com/duke-git/lancet/v2/slice"
	clinic_internal_api "github.com/pingcap-inc/clinic/clinic_data_service/pkg/clinic-internal-api"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/common"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/constant"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/dto"

	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/duckdb"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/log"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/prometheus"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/utils"
	"go.uber.org/zap"
)

var sqlUnit = map[string]string{
	"query_time":           "s",
	"topsql_tidb_cpu_time": "minute",
	"topsql_tidb_io_time":  "minute",
	"topsql_avg_latency":   "ms",
}

const maxInt64AsUint64 = uint64(^uint64(0) >> 1)

type DataProxyQuerier interface {
	Query(ctx context.Context, req *dto.DataProxyQueryRequest) (*dto.DataProxyQueryResult, error)
}

// CanonicalClusterDataURLsReader returns Clinic's authoritative storage and
// observability targets for a cluster.
type CanonicalClusterDataURLsReader interface {
	GetCanonicalClusterDataURLs(ctx context.Context, clusterID string) (*clinic_internal_api.ClusterDataURLs, error)
}

type ClusterDiffApiService struct {
	dataProxy    DataProxyQuerier
	targetReader CanonicalClusterDataURLsReader
}

func NewClusterDiffApiService(dataProxy DataProxyQuerier, targetReader CanonicalClusterDataURLsReader) *ClusterDiffApiService {
	return &ClusterDiffApiService{dataProxy: dataProxy, targetReader: targetReader}
}

func (s *ClusterDiffApiService) getSlowQueryDataSourceForCluster(ctx context.Context, c dto.ClusterSideDTO) (string, error) {
	if c.ClusterUrls.SlowQueryS3URL != "" {
		log.Info(ctx, "resolved cluster comparison slow query target",
			zap.String("operation", "resolve_cluster_comparison_slow_query_target"),
			zap.String("cluster_id", c.ClusterId),
			zap.String("source", "request"))
		return c.ClusterUrls.SlowQueryS3URL, nil
	}
	if s == nil || s.targetReader == nil {
		return "", fmt.Errorf("canonical cluster target reader is unavailable")
	}

	log.Info(ctx, "resolving canonical cluster comparison slow query target",
		zap.String("operation", "resolve_cluster_comparison_slow_query_target"),
		zap.String("cluster_id", c.ClusterId),
		zap.String("source", "clinic_internal_api"))
	target, err := s.targetReader.GetCanonicalClusterDataURLs(ctx, c.ClusterId)
	if err != nil {
		log.Error(ctx, "failed to resolve canonical cluster comparison slow query target",
			zap.String("operation", "resolve_cluster_comparison_slow_query_target"),
			zap.String("cluster_id", c.ClusterId),
			zap.Error(err))
		return "", fmt.Errorf("resolve canonical slow query target for cluster %s: %w", c.ClusterId, err)
	}
	if target == nil {
		return "", fmt.Errorf("canonical slow query target is empty for cluster %s", c.ClusterId)
	}
	dataSource, scheme, err := validateCanonicalSlowQueryDataSource(target.SlowQueryS3URL)
	if err != nil {
		return "", fmt.Errorf("invalid canonical slow query target for cluster %s: %w", c.ClusterId, err)
	}
	log.Info(ctx, "resolved cluster comparison slow query target",
		zap.String("operation", "resolve_cluster_comparison_slow_query_target"),
		zap.String("cluster_id", c.ClusterId),
		zap.String("source", "clinic_internal_api"),
		zap.String("scheme", scheme))
	return dataSource, nil
}

func validateCanonicalSlowQueryDataSource(raw string) (string, string, error) {
	dataSource := strings.TrimSpace(raw)
	if dataSource == "" {
		return "", "", fmt.Errorf("slow query data source is empty")
	}
	parsed, err := url.Parse(dataSource)
	if err != nil || parsed.Host == "" {
		return "", "", fmt.Errorf("slow query data source URL is malformed")
	}
	scheme := strings.ToLower(parsed.Scheme)
	switch scheme {
	case "s3", "gs", "oss", "abfss", "http", "https":
		return dataSource, scheme, nil
	default:
		return "", "", fmt.Errorf("slow query data source scheme is unsupported")
	}
}

type QueryComparisonResult struct {
	ClusterATop     []duckdb.DigestInfo
	ClusterBMatched []duckdb.DigestInfo
	TopSqlTidbA     []dtoTopSqlDigest
	TopSqlTidbB     []dtoTopSqlDigest
	TopSqlTikvA     []dtoTopSqlTikvDigest
	TopSqlTikvB     []dtoTopSqlTikvDigest
}

func (s *ClusterDiffApiService) QuerySlowQueryTopDigests(ctx context.Context, diffParam dto.QueryComparisonRequest) (QueryComparisonResult, error) {
	log.Info(ctx, "QueryComparison start")

	topN := diffParam.Options.TopN
	if topN <= 0 {
		topN = DefaultTopN()
	}
	sqlA := buildSlowQueryTopDigestSQL(diffParam.ClusterA.TimeRange.Start, diffParam.ClusterA.TimeRange.End, topN)
	log.Info(ctx, fmt.Sprintf("start query clusterA top N digests, topN: %d", topN))
	respA, err := s.querySlowQuery(ctx, diffParam.ClusterA, sqlA)
	if err != nil {
		return QueryComparisonResult{}, err
	}
	clusterADigests, err := parseRowsToDigest(respA)
	if err != nil {
		return QueryComparisonResult{}, err
	}
	// Build IN clause digests from A side; if empty, avoid invalid SQL like "IN ()"
	digestArr := make([]string, 0, len(clusterADigests))
	for _, digestInfo := range clusterADigests {
		digestArr = append(digestArr, digestInfo.Digest)
	}
	if len(digestArr) == 0 {
		return QueryComparisonResult{ClusterATop: clusterADigests, ClusterBMatched: []duckdb.DigestInfo{}}, nil
	}
	inClause := BuildInValuesLiteral(digestArr)
	sqlB := buildSlowQueryDigestInSQL(diffParam.ClusterB.TimeRange.Start, diffParam.ClusterB.TimeRange.End, inClause)
	log.Info(ctx, fmt.Sprintf("start query clusterB digests in A, inClause size: %d", len(digestArr)))
	respB, err := s.querySlowQuery(ctx, diffParam.ClusterB, sqlB)
	if err != nil {
		return QueryComparisonResult{}, err
	}
	clusterBSlowLogs, err := parseRowsToDigest(respB)
	if err != nil {
		return QueryComparisonResult{}, err
	}
	log.Info(ctx, "QueryTopDigests success", zap.Any("clusterB top digest size", len(clusterBSlowLogs)))
	return QueryComparisonResult{ClusterATop: clusterADigests, ClusterBMatched: clusterBSlowLogs}, nil
}

func (s *ClusterDiffApiService) querySlowQuery(ctx context.Context, c dto.ClusterSideDTO, sql string) (*dto.DataProxyQueryResult, error) {
	if s == nil || s.dataProxy == nil {
		return nil, fmt.Errorf("data proxy service is unavailable")
	}
	dataSource, err := s.getSlowQueryDataSourceForCluster(ctx, c)
	if err != nil {
		return nil, err
	}
	return s.dataProxy.Query(ctx, &dto.DataProxyQueryRequest{
		ClusterId: c.ClusterId,
		SQL:       sql,
		Options: dto.DataProxyQueryOptions{
			DataSource: dataSource,
			Region:     c.Region,
			Vendor:     c.Provider,
		},
		Timeout: 15 * time.Minute,
	})
}

func buildSlowQueryTopDigestSQL(start, end time.Time, topN int) string {
	return fmt.Sprintf(`
		SELECT
			COALESCE(digest, '') AS digest,
			COALESCE(ROUND(AVG(mem_max), 0), 0) AS avg_mem,
			COALESCE(ROUND(MAX(mem_max), 0), 0) AS max_mem,
			COALESCE(ROUND(AVG(query_time), 2), 0) AS avg_query_time,
			COALESCE(ROUND(MAX(query_time), 2), 0) AS max_query_time,
			COUNT(*) AS total_count
		FROM clinic_data_proxy.slow_query_logs
		WHERE date IN (%s)
			AND time BETWEEN %d AND %d
			AND lower(query) LIKE 'select%%'
			AND lower(query) NOT LIKE 'analyze%%'
			AND lower(query) NOT LIKE 'create%%'
			AND lower(query) NOT LIKE 'alter%%'
			AND lower(query) NOT LIKE 'drop%%'
			AND lower(query) NOT LIKE 'truncate%%'
			AND lower(query) NOT LIKE 'rename%%'
			AND lower(query) NOT LIKE '%%force index%%'
			AND NOT (lower(query) LIKE '%%read_from_storage(tiflash%%' AND lower(query) NOT LIKE '%%/*+ read_from_storage(tiflash%%')
		GROUP BY digest
		ORDER BY total_count DESC
		LIMIT %d
	`, buildSlowQueryDateInClause(start, end), start.UTC().Unix(), end.UTC().Unix(), topN)
}

func buildSlowQueryDigestInSQL(start, end time.Time, inClause string) string {
	return fmt.Sprintf(`
		SELECT
			COALESCE(digest, '') AS digest,
			COALESCE(ROUND(AVG(mem_max), 0), 0) AS avg_mem,
			COALESCE(ROUND(MAX(mem_max), 0), 0) AS max_mem,
			COALESCE(ROUND(AVG(query_time), 2), 0) AS avg_query_time,
			COALESCE(ROUND(MAX(query_time), 2), 0) AS max_query_time,
			COUNT(*) AS total_count
		FROM clinic_data_proxy.slow_query_logs
		WHERE date IN (%s)
			AND time BETWEEN %d AND %d
			AND digest IN (%s)
			AND lower(query) LIKE 'select%%'
			AND lower(query) NOT LIKE 'analyze%%'
			AND lower(query) NOT LIKE 'create%%'
			AND lower(query) NOT LIKE 'alter%%'
			AND lower(query) NOT LIKE 'drop%%'
			AND lower(query) NOT LIKE 'truncate%%'
			AND lower(query) NOT LIKE 'rename%%'
			AND lower(query) NOT LIKE '%%force index%%'
			AND NOT (lower(query) LIKE '%%read_from_storage(tiflash%%' AND lower(query) NOT LIKE '%%/*+ read_from_storage(tiflash%%')
		GROUP BY digest
		ORDER BY total_count DESC
	`, buildSlowQueryDateInClause(start, end), start.UTC().Unix(), end.UTC().Unix(), inClause)
}

func buildSlowQueryDigestCountSQL(start, end time.Time) string {
	return fmt.Sprintf(`
		SELECT
			COUNT(*) AS total_records,
			COUNT(DISTINCT digest) AS distinct_digests
		FROM clinic_data_proxy.slow_query_logs
		WHERE date IN (%s)
			AND time BETWEEN %d AND %d
			AND lower(query) LIKE 'select%%'
			AND lower(query) NOT LIKE 'analyze%%'
			AND lower(query) NOT LIKE 'create%%'
			AND lower(query) NOT LIKE 'alter%%'
			AND lower(query) NOT LIKE 'drop%%'
			AND lower(query) NOT LIKE 'truncate%%'
			AND lower(query) NOT LIKE 'rename%%'
			AND lower(query) NOT LIKE '%%force index%%'
			AND NOT (lower(query) LIKE '%%read_from_storage(tiflash%%' AND lower(query) NOT LIKE '%%/*+ read_from_storage(tiflash%%')
	`, buildSlowQueryDateInClause(start, end), start.UTC().Unix(), end.UTC().Unix())
}

func buildSlowQueryDateInClause(start, end time.Time) string {
	startDay := dateOnlyUTC(start)
	endDay := dateOnlyUTC(end)
	if endDay.Before(startDay) {
		startDay, endDay = endDay, startDay
	}
	dates := make([]string, 0, int(endDay.Sub(startDay).Hours()/24)+1)
	for d := startDay; !d.After(endDay); d = d.AddDate(0, 0, 1) {
		dates = append(dates, d.Format("20060102"))
	}
	return BuildInValuesLiteral(dates)
}

func dateOnlyUTC(t time.Time) time.Time {
	utc := t.UTC()
	return time.Date(utc.Year(), utc.Month(), utc.Day(), 0, 0, 0, 0, time.UTC)
}

func parseRowsToDigest(result *dto.DataProxyQueryResult) ([]duckdb.DigestInfo, error) {
	rows := dataProxyRows(result)
	out := make([]duckdb.DigestInfo, 0, len(rows))
	for i, row := range rows {
		totalCount, err := int64Value(row["total_count"])
		if err != nil {
			return nil, fmt.Errorf("row %d total_count: %w", i, err)
		}
		avgMem, err := float64Value(row["avg_mem"])
		if err != nil {
			return nil, fmt.Errorf("row %d avg_mem: %w", i, err)
		}
		maxMem, err := float64Value(row["max_mem"])
		if err != nil {
			return nil, fmt.Errorf("row %d max_mem: %w", i, err)
		}
		avgQueryTime, err := float64Value(row["avg_query_time"])
		if err != nil {
			return nil, fmt.Errorf("row %d avg_query_time: %w", i, err)
		}
		maxQueryTime, err := float64Value(row["max_query_time"])
		if err != nil {
			return nil, fmt.Errorf("row %d max_query_time: %w", i, err)
		}
		out = append(out, duckdb.DigestInfo{
			Digest:       stringValue(row["digest"]),
			AvgMem:       avgMem,
			MaxMem:       maxMem,
			AvgQueryTime: avgQueryTime,
			MaxQueryTime: maxQueryTime,
			TotalCount:   totalCount,
		})
	}
	return out, nil
}

func parseRowsToClusterCount(result *dto.DataProxyQueryResult) (duckdb.ClusterCount, error) {
	rows := dataProxyRows(result)
	if len(rows) == 0 {
		return duckdb.ClusterCount{}, nil
	}
	totalRecords, err := int64Value(rows[0]["total_records"])
	if err != nil {
		return duckdb.ClusterCount{}, fmt.Errorf("total_records: %w", err)
	}
	distinctDigests, err := int64Value(rows[0]["distinct_digests"])
	if err != nil {
		return duckdb.ClusterCount{}, fmt.Errorf("distinct_digests: %w", err)
	}
	return duckdb.ClusterCount{TotalRecords: totalRecords, DistinctDigest: distinctDigests}, nil
}

func dataProxyRows(result *dto.DataProxyQueryResult) []map[string]interface{} {
	if result == nil {
		return nil
	}
	rows := make([]map[string]interface{}, 0, len(result.Rows))
	for _, row := range result.Rows {
		item := make(map[string]interface{}, len(result.Columns))
		for i, col := range result.Columns {
			if i >= len(row) {
				break
			}
			item[normalizeColumnName(col)] = row[i]
		}
		rows = append(rows, item)
	}
	return rows
}

func normalizeColumnName(col string) string {
	return strings.ToLower(strings.Trim(strings.TrimSpace(col), "`\""))
}

func stringValue(v interface{}) string {
	if v == nil {
		return ""
	}
	if s, ok := v.(string); ok {
		return s
	}
	return fmt.Sprint(v)
}

func float64Value(v interface{}) (float64, error) {
	switch x := v.(type) {
	case nil:
		return 0, nil
	case float64:
		return x, nil
	case float32:
		return float64(x), nil
	case int:
		return float64(x), nil
	case int8:
		return float64(x), nil
	case int16:
		return float64(x), nil
	case int32:
		return float64(x), nil
	case int64:
		return float64(x), nil
	case uint:
		return float64(x), nil
	case uint8:
		return float64(x), nil
	case uint16:
		return float64(x), nil
	case uint32:
		return float64(x), nil
	case uint64:
		return float64(x), nil
	case string:
		if x == "" {
			return 0, nil
		}
		return strconv.ParseFloat(x, 64)
	default:
		return 0, fmt.Errorf("unsupported numeric type %T", v)
	}
}

func int64Value(v interface{}) (int64, error) {
	switch x := v.(type) {
	case nil:
		return 0, nil
	case int:
		return int64(x), nil
	case int8:
		return int64(x), nil
	case int16:
		return int64(x), nil
	case int32:
		return int64(x), nil
	case int64:
		return x, nil
	case uint:
		if uint64(x) > maxInt64AsUint64 {
			return 0, fmt.Errorf("value overflows int64: %d", x)
		}
		return int64(x), nil
	case uint8:
		return int64(x), nil
	case uint16:
		return int64(x), nil
	case uint32:
		return int64(x), nil
	case uint64:
		if x > maxInt64AsUint64 {
			return 0, fmt.Errorf("value overflows int64: %d", x)
		}
		return int64(x), nil
	case float32:
		return int64(x), nil
	case float64:
		return int64(x), nil
	case string:
		if x == "" {
			return 0, nil
		}
		return strconv.ParseInt(x, 10, 64)
	default:
		return 0, fmt.Errorf("unsupported integer type %T", v)
	}
}

// dtoTopSqlDigest captures TiDB TopSQL summary for one digest.
type dtoTopSqlDigest struct {
	Digest           string
	Count            float64
	AvgLatency       float64
	TotalCpuTimeTidb float64
	TotalCpuTimeTikv float64
	Cluster          string
}

// dtoTopSqlTikvDigest captures TiKV TopSQL summary for one digest.
type dtoTopSqlTikvDigest struct {
	Digest     string
	CpuTime    float64
	CpuPercent float64
	Cluster    string
}

// slowQueryDigestDiff 构造集合：
// - A→B 正向： aTopMap （A TopN）， bMatchMap （B 的匹配）
// - B→A 反向： bTopMap （B TopN）， aMatchMap （A 的匹配）
// - union 使用 aTopMap ∪ bTopMap

func (s *ClusterDiffApiService) SqlDiff(ctx context.Context, diffParam dto.QueryComparisonRequest) (dto.SqlDiffResponse, error) {
	comments := "ClusterA vs ClusterB, diff is percent, e.g. 12 means clusterB is 12% faster than clusterA"
	keys := diffParam.Options.DigestDiffKeys
	includeCluster := slice.Contain(keys, "slowQueryCountDiff")
	includeSlowQuery := slice.Contain(keys, "slowQueryDigestDiff")
	includeTopSql := slice.Contain(keys, "topSqlDigestDiff")

	// if err != nil {
	// 	return dto.QueryComparisonResponse{}, err
	// }
	// clusterB, err := common.GetCluster(ctx, diffParam.ClusterB.ClusterId)
	// if err != nil {
	// 	return dto.SqlDiffResponse{}, err
	// }

	var clusterDiff []dto.DiffInfo
	var summary *dto.SummaryOuter
	var slowQueryGroupPtr *dto.DiffGroup
	var slowQuerySummary *dto.DiffGroupSummary
	var tidbGroupPtr *dto.DiffGroup
	var tikvGroupPtr *dto.DiffGroup
	var tidbSummaryPtr *dto.DiffGroupSummary

	diffClusters := dto.DiffClusters{
		ClusterA: diffParam.ClusterA,
		ClusterB: diffParam.ClusterB,
	}

	// clusterB := config.GetClusterByID(diffParam.ClusterB.ClusterId)
	isSkip := (diffParam.ClusterA.Provider == string(constant.ProviderAliCloud) || diffParam.ClusterB.Provider == string(constant.ProviderAliCloud) || diffParam.ClusterB.Provider == string(constant.ProviderGCP))
	// 1、slowQueryCountDiff
	if includeCluster && !isSkip {
		clusterCountArr, err := s.QueryDigestCount(ctx, diffParam)
		if err != nil {
			log.Error(ctx, "QueryDigestCount error", zap.Error(err), zap.String("clusterId", diffParam.ClusterA.ClusterId))

		} else {
			if len(clusterCountArr) == 2 {
				aSlorQueryCount := clusterCountArr[0].TotalRecords
				aSlowqueryDigest := clusterCountArr[0].DistinctDigest
				bSlorQueryCount := clusterCountArr[1].TotalRecords
				bSlowqueryDigest := clusterCountArr[1].DistinctDigest
				slowqueryCountDiff := dto.DiffInfo{
					Key:      "slowquery_count",
					ClusterA: float64(aSlorQueryCount),
					ClusterB: float64(bSlorQueryCount),
					// Diff: *utils.CalculateDiff(float64(bSlorQueryCount), float64(aSlorQueryCount)),
				}

				if aSlorQueryCount != 0 && bSlorQueryCount != 0 {
					slowqueryCountDiff.Diff = *utils.CalculateDiff(float64(bSlorQueryCount), float64(aSlorQueryCount))
				}
				clusterDiff = append(clusterDiff, slowqueryCountDiff)
				slowqueryDigestDiff := dto.DiffInfo{
					Key:      "slowquery_digest",
					ClusterA: float64(aSlowqueryDigest),
					ClusterB: float64(bSlowqueryDigest),
					// Diff: *utils.CalculateDiff(float64(bSlowqueryDigest), float64(aSlowqueryDigest)),
				}
				if aSlowqueryDigest != 0 && bSlowqueryDigest != 0 {
					slowqueryDigestDiff.Diff = *utils.CalculateDiff(float64(bSlowqueryDigest), float64(aSlowqueryDigest))
				}
				clusterDiff = append(clusterDiff, slowqueryDigestDiff)
			}
		}
	}
	//  only query slowQueryCountDiff
	if includeCluster && !(includeSlowQuery || includeTopSql) {

		return dto.SqlDiffResponse{
			DiffClusters: diffClusters,
			Comments:     comments,
			ClusterDiff:  clusterDiff,
			Units:        sqlUnit,
			Result:       dto.Result{IsSuccess: true},
			// Summary:              nil,
			// SlowQueryDigestDiff:  nil,
			// TopSqlDigestDiff: nil,
		}, nil
	}

	// 2、slowQueryDigestDiff
	if includeSlowQuery && !isSkip {
		res, err := s.QuerySlowQueryTopDigests(ctx, diffParam)
		if err != nil {
			log.Error(ctx, "Query slow query TopDigests error", zap.Error(err), zap.String("clusterId", diffParam.ClusterA.ClusterId))

			// return dto.SqlDiffResponse{}, err
		} else {
			aStats := make([]dto.DigestStat, 0, len(res.ClusterATop))
			for _, v := range res.ClusterATop {
				aStats = append(aStats, dto.DigestStat{Digest: v.Digest, Query: v.Query, AvgMem: v.AvgMem, MaxMem: v.MaxMem, AvgQueryTime: v.AvgQueryTime, MaxQueryTime: v.MaxQueryTime, TotalCount: v.TotalCount})
			}
			bStats := make([]dto.DigestStat, 0, len(res.ClusterBMatched))
			for _, v := range res.ClusterBMatched {
				bStats = append(bStats, dto.DigestStat{Digest: v.Digest, Query: v.Query, AvgMem: v.AvgMem, MaxMem: v.MaxMem, AvgQueryTime: v.AvgQueryTime, MaxQueryTime: v.MaxQueryTime, TotalCount: v.TotalCount})
			}
			aTopMap := make(map[string]dto.DigestStat, len(aStats))
			for _, v := range aStats {
				aTopMap[v.Digest] = v
			}
			bMatchMap := make(map[string]dto.DigestStat, len(bStats))
			for _, v := range bStats {
				bMatchMap[v.Digest] = v
			}

			revParam := diffParam
			revParam.ClusterA, revParam.ClusterB = diffParam.ClusterB, diffParam.ClusterA
			revRes, err := s.QuerySlowQueryTopDigests(ctx, revParam)
			if err != nil {
				return dto.SqlDiffResponse{}, err
			}
			bTopStats := make([]dto.DigestStat, 0, len(revRes.ClusterATop))
			for _, v := range revRes.ClusterATop {
				bTopStats = append(bTopStats, dto.DigestStat{Digest: v.Digest, Query: v.Query, AvgMem: v.AvgMem, MaxMem: v.MaxMem, AvgQueryTime: v.AvgQueryTime, MaxQueryTime: v.MaxQueryTime, TotalCount: v.TotalCount})
			}
			bTopMap := make(map[string]dto.DigestStat, len(bTopStats))
			for _, v := range bTopStats {
				bTopMap[v.Digest] = v
			}
			aMatchStats := make([]dto.DigestStat, 0, len(revRes.ClusterBMatched))
			for _, v := range revRes.ClusterBMatched {
				aMatchStats = append(aMatchStats, dto.DigestStat{Digest: v.Digest, Query: v.Query, AvgMem: v.AvgMem, MaxMem: v.MaxMem, AvgQueryTime: v.AvgQueryTime, MaxQueryTime: v.MaxQueryTime, TotalCount: v.TotalCount})
			}
			aMatchMap := make(map[string]dto.DigestStat, len(aMatchStats))
			for _, v := range aMatchStats {
				aMatchMap[v.Digest] = v
			}

			union := make(map[string]struct{})
			for d := range aTopMap {
				union[d] = struct{}{}
			}
			for d := range bTopMap {
				union[d] = struct{}{}
			}

			digestDiff := make([]dto.DigestDiffEntry, 0, len(union))
			// addedCnt, removedCnt, improvedCnt, degradedCnt := 0, 0, 0, 0

			var addedArr, removedArr, improvedArr, degradedArr []string
			// var dec50PlusDigests, dec50To20Digests, dec20To0Digests, neg5To5Digests, inc0To20Digests, inc20To50Digests, inc50PlusDigests []string

			dec50PlusDigests, dec50To20Digests, dec20To0Digests, neg5To5Digests, inc0To20Digests, inc20To50Digests, inc50PlusDigests := []string{}, []string{}, []string{}, []string{}, []string{}, []string{}, []string{}

			for d := range union {
				aStat, okA := aTopMap[d]
				if !okA {
					if v, ok := aMatchMap[d]; ok {
						aStat = v
						okA = true
					}
				}
				bStat, okB := bTopMap[d]
				if !okB {
					if v, ok := bMatchMap[d]; ok {
						bStat = v
						okB = true
					}
				}
				aCnt := float64(aStat.TotalCount)
				bCnt := float64(bStat.TotalCount)
				var diffCnt float64
				if aCnt == 0 {
					diffCnt = 0
				} else {
					diffCnt = *utils.CalculateDiff(float64(bCnt), float64(aCnt))
				}
				aAvgT := aStat.AvgQueryTime
				bAvgT := bStat.AvgQueryTime
				var diffAvgT float64
				if aAvgT == 0 {
					diffAvgT = 0
				} else {
					diffAvgT = *utils.CalculateDiff(bAvgT, aAvgT)
				}
				aAvgM := aStat.AvgMem
				bAvgM := bStat.AvgMem
				var diffAvgM float64
				if aAvgM == 0 {
					diffAvgM = 0
				} else {
					diffAvgM = *utils.CalculateDiff(bAvgM, aAvgM)
				}
				label := ""
				if _, inBTop := bTopMap[d]; inBTop {
					if _, inAMatch := aMatchMap[d]; !inAMatch {
						label = "Added"
						// addedCnt++
						addedArr = append(addedArr, d)
					}
				}
				if label == "" {
					if _, inATop := aTopMap[d]; inATop {
						if _, inBMatch := bMatchMap[d]; !inBMatch {
							label = "Removed"
							// removedCnt++
							removedArr = append(removedArr, d)
						}
					}
				}
				if okA && okB {
					if diffAvgT <= -50 {
						dec50PlusDigests = append(dec50PlusDigests, d)
					} else if diffAvgT <= -20 {
						dec50To20Digests = append(dec50To20Digests, d)
					} else if diffAvgT < -5 {
						dec20To0Digests = append(dec20To0Digests, d)
					} else if diffAvgT <= 5 {
						neg5To5Digests = append(neg5To5Digests, d)
					} else if diffAvgT <= 20 {
						inc0To20Digests = append(inc0To20Digests, d)
					} else if diffAvgT <= 50 {
						inc20To50Digests = append(inc20To50Digests, d)
					} else {
						inc50PlusDigests = append(inc50PlusDigests, d)
					}
					if label == "" {
						if diffAvgT >= -5 && diffAvgT <= 5 {
							label = "Stable"
						} else if diffAvgT < 0 {
							label = "ImprovedPerformance"
							improvedArr = append(improvedArr, d)
						} else if diffAvgT > 0 {
							label = "DegradedPerformance"
							degradedArr = append(degradedArr, d)
						}
					}
				}

				infos := []dto.DiffInfo{{Key: "slowquery_count", ClusterA: float64(aCnt), ClusterB: float64(bCnt), Diff: diffCnt}, {Key: "slowquery_avg_querytime", ClusterA: utils.RoundFloat(aAvgT, 2), ClusterB: utils.RoundFloat(bAvgT, 2), Diff: diffAvgT}, {Key: "slowquery_avg_mem", ClusterA: utils.RoundFloat(aAvgM, 2), ClusterB: utils.RoundFloat(bAvgM, 2), Diff: diffAvgM}}
				digestDiff = append(digestDiff, dto.DigestDiffEntry{Digest: d, Label: label, DiffInfos: infos})
			}
			buckets := []map[string]any{
				{"[<-50]": len(dec50PlusDigests), "digests": dec50PlusDigests},
				{"[-50~-20]": len(dec50To20Digests), "digests": dec50To20Digests},
				{"[-20~-5]": len(dec20To0Digests), "digests": dec20To0Digests},
				{"[-5~5]": len(neg5To5Digests), "digests": neg5To5Digests},
				{"[5~20]": len(inc0To20Digests), "digests": inc0To20Digests},
				{"[20~50]": len(inc20To50Digests), "digests": inc20To50Digests},
				{"[>50]": len(inc50PlusDigests), "digests": inc50PlusDigests},
			}
			slowQuerySummary = &dto.DiffGroupSummary{
				Added: len(addedArr), Removed: len(removedArr),
				ImprovedPerformance: len(improvedArr), DegradedPerformance: len(degradedArr), Stable: len(neg5To5Digests),
				DiffDetail: &dto.DiffDetail{Cloumn: "slowquery_avg_querytime", Buckets: buckets},
			}
			SortDigestByKeyDesc(digestDiff, "slowquery_avg_querytime")
			slowQueryGroup := dto.DiffGroup{DigestDiff: digestDiff}
			slowQueryGroupPtr = &slowQueryGroup
			if summary == nil {
				summary = &dto.SummaryOuter{}
			}
			summary.SlowQueryDigest = slowQuerySummary
		}
	}

	// 3、topSqlTidbDigestDiff

	if includeTopSql {

		pcClientMap, err := getClusterPrometheusClient(ctx, diffParam.ClusterA, diffParam.ClusterB)
		if err != nil {
			// return dto.SqlDiffResponse{}, err
			return dto.SqlDiffResponse{}, nil
		}

		aTopDigests, bTopDigests, aTopCount, bTopCount, err := s.fetchTopSqlTidbSets(ctx, pcClientMap, diffParam)
		if err != nil {
			return dto.SqlDiffResponse{}, nil
			// return dto.SqlDiffResponse{}, err
		}
		// group, sum := buildTopSqlDiffGroupFromSets(ctx,pcClientMap,diffParam, aTopDigests, bTopDigests, aTopCount, bTopCount)
		group, sum := buildTopSqlDiffGroupFromSets(ctx, pcClientMap, diffParam, aTopDigests, bTopDigests, aTopCount, bTopCount)
		tidbSummaryPtr = &sum
		tidbGroupPtr = group
		if summary == nil {
			summary = &dto.SummaryOuter{}
		}
		summary.TopSqlDigest = tidbSummaryPtr
	}
	// removed residual block; logic moved inside includeSlowQuery

	return dto.SqlDiffResponse{
		DiffClusters:         diffClusters,
		Comments:             comments,
		Units:                sqlUnit,
		Result:               dto.Result{IsSuccess: true},
		ClusterDiff:          clusterDiff,
		Summary:              summary,
		SlowQueryDigestDiff:  slowQueryGroupPtr,
		TopSqlTidbDigestDiff: tidbGroupPtr,
		TopSqlTikvDigestDiff: tikvGroupPtr,
	}, nil
}

func getClusterPrometheusClient(ctx context.Context, clusterA, clusterB dto.ClusterSideDTO) (map[string]*prometheus.PrometheusClient, error) {
	clientMap := make(map[string]*prometheus.PrometheusClient)
	clusterAId := clusterA.ClusterId
	clusterBId := clusterB.ClusterId
	log.Info(ctx, "get cluster prometheus clients", zap.Any("clusterID", clusterAId))
	// 1. Create Prometheus clients for prod cluster
	pcClusterA, err := prometheus.CreatePrometheusClient(ctx, prometheus.CreatePrometheusClientReq{
		ClusterID:     clusterA.ClusterId,
		PrometheusURL: clusterA.ClusterUrls.ClusterMetricsURL,
		Provider:      clusterA.Provider,
		Region:        clusterA.Region,
		DeployType:    clusterA.DeployType,
		TenantID:      clusterA.TenantID,
		ProjectID:     clusterA.ProjectID,
	})
	if err != nil {
		log.Error(ctx, "Failed to create prod Prometheus client", zap.Error(err), zap.String("clusterId", clusterAId))
		return nil, fmt.Errorf("failed to create prod Prometheus client: %w", err)
	}
	clientMap[clusterAId] = pcClusterA
	log.Info(ctx, "get cluster prometheus clients", zap.Any("clusterA", clusterAId))
	pcClusterB, err := prometheus.CreatePrometheusClient(ctx, prometheus.CreatePrometheusClientReq{
		ClusterID:     clusterB.ClusterId,
		PrometheusURL: clusterB.ClusterUrls.ClusterMetricsURL,
		Provider:      clusterB.Provider,
		Region:        clusterB.Region,
		DeployType:    clusterB.DeployType,
		TenantID:      clusterB.TenantID,
		ProjectID:     clusterB.ProjectID,
	})
	if err != nil {
		log.Error(ctx, "Failed to create test Prometheus client", zap.Error(err), zap.String("clusterId", clusterBId))
		return nil, fmt.Errorf("failed to create test Prometheus client: %w", err)
	}
	clientMap[clusterBId] = pcClusterB
	return clientMap, nil
}

func (s *ClusterDiffApiService) QueryDigestCount(ctx context.Context, diffParam dto.QueryComparisonRequest) ([]duckdb.ClusterCount, error) {
	log.Info(ctx, "QueryDigestCount ", zap.Any("diffParam", diffParam))

	clustersArr := []dto.ClusterSideDTO{diffParam.ClusterA, diffParam.ClusterB}
	var countArr []duckdb.ClusterCount
	for _, c := range clustersArr {
		// 	// cluster := config.GetClusterByID(c.ClusterId)
		// 	cluster, _ := common.GetCluster(ctx, c.ClusterId)
		// 	if cluster == nil {
		// 		return nil, fmt.Errorf("cluster not found in db, clusterId: %s", c.ClusterId)
		// 	}

		sql := buildSlowQueryDigestCountSQL(c.TimeRange.Start, c.TimeRange.End)

		resp, err := s.querySlowQuery(ctx, c, sql)
		if err != nil {
			log.Error(ctx, "data proxy digest count failed", zap.Error(err))
			return nil, err
		}
		clusterCount, err := parseRowsToClusterCount(resp)
		if err != nil {
			log.Error(ctx, "parse digest count failed", zap.Error(err))
			return nil, err
		}
		clusterCount.ClusterId = c.ClusterId
		// log.PlainInfo("clusterASlowLogs", zap.Any("clusterASlowLogs", clusterASlowLogs))
		countArr = append(countArr, clusterCount)
		log.Info(ctx, "clusterCount  success", zap.Any("clusterCount ", clusterCount))
	}
	return countArr, nil
}

func (s *ClusterDiffApiService) fetchTopSqlTidbSets(ctx context.Context, pcClientMap map[string]*prometheus.PrometheusClient, diffParam dto.QueryComparisonRequest) ([]string, []string, map[string]float64, map[string]float64, error) {
	a := diffParam.ClusterA
	b := diffParam.ClusterB
	pcClientA := pcClientMap[a.ClusterId]
	pcClientB := pcClientMap[b.ClusterId]
	aInterval := fmt.Sprintf("%dm", int(a.TimeRange.End.Sub(a.TimeRange.Start).Minutes()))
	bInterval := fmt.Sprintf("%dm", int(b.TimeRange.End.Sub(b.TimeRange.Start).Minutes()))
	aQueryTime := a.TimeRange.End.UTC()
	bQueryTime := b.TimeRange.End.UTC()
	topN := diffParam.Options.TopN
	if topN <= 0 {
		topN = DefaultTopN()
	}

	topA, err := pcClientA.GetTopXCount(a.ClusterId, aInterval, topN, aQueryTime)
	if err != nil {
		return nil, nil, nil, nil, err
	}
	if len(topA) == 0 {
		topA, err = pcClientA.GetTopXCountHistoryData(a.ClusterId, aInterval, topN, aQueryTime)
		if err != nil {
			return nil, nil, nil, nil, err
		}
	}
	aTopDigests := make([]string, 0, len(topA))
	aTopCount := make(map[string]float64, len(topA))
	for _, t := range topA {
		aTopDigests = append(aTopDigests, t.Digest)
		aTopCount[t.Digest] = t.Value
	}
	topB, err := pcClientB.GetTopXCount(b.ClusterId, bInterval, topN, bQueryTime)
	if err != nil {
		return nil, nil, nil, nil, err
	}
	if len(topB) == 0 {
		topB, err = pcClientB.GetTopXCountHistoryData(b.ClusterId, bInterval, topN, bQueryTime)
		if err != nil {
			return nil, nil, nil, nil, err
		}
	}
	bTopDigests := make([]string, 0, len(topB))
	bTopCount := make(map[string]float64, len(topB))
	for _, t := range topB {
		bTopDigests = append(bTopDigests, t.Digest)
		bTopCount[t.Digest] = t.Value
	}
	return aTopDigests, bTopDigests, aTopCount, bTopCount, nil
}

func buildTopSqlDiffGroupFromSets(ctx context.Context, pcClientMap map[string]*prometheus.PrometheusClient, diffParam dto.QueryComparisonRequest, aTopDigests []string, bTopDigests []string, aTopCount map[string]float64, bTopCount map[string]float64) (*dto.DiffGroup, dto.DiffGroupSummary) {
	pcClientA := pcClientMap[diffParam.ClusterA.ClusterId]
	pcClientB := pcClientMap[diffParam.ClusterB.ClusterId]
	aTopSet := make(map[string]struct{}, len(aTopDigests))
	for _, d := range aTopDigests {
		aTopSet[d] = struct{}{}
	}
	bTopSet := make(map[string]struct{}, len(bTopDigests))
	for _, d := range bTopDigests {
		bTopSet[d] = struct{}{}
	}

	union := make(map[string]struct{})
	for d := range aTopSet {
		union[d] = struct{}{}
	}
	for d := range bTopSet {
		union[d] = struct{}{}
	}

	a := diffParam.ClusterA
	b := diffParam.ClusterB
	aInterval := fmt.Sprintf("%dm", int(a.TimeRange.End.Sub(a.TimeRange.Start).Minutes()))
	bInterval := fmt.Sprintf("%dm", int(b.TimeRange.End.Sub(b.TimeRange.Start).Minutes()))
	aQueryTime := a.TimeRange.End.UTC()
	bQueryTime := b.TimeRange.End.UTC()
	instanceTypeTidb := "tidb"
	instanceTypeTikv := "tikv"
	digestDiff := make([]dto.DigestDiffEntry, 0, len(union))
	// added, removed, improved, degraded := 0, 0, 0, 0
	var addedArr, removedArr, improvedArr, degradedArr []string
	dec50PlusDigests, dec50To20Digests, dec20To0Digests, neg5To5Digests, inc0To20Digests, inc20To50Digests, inc50PlusDigests := []string{}, []string{}, []string{}, []string{}, []string{}, []string{}, []string{}
	// wlt test
	digests := maputil.Keys(union)
	aDigestTidbCpu, err := pcClientA.GetDigestsCpu(a.ClusterId, aQueryTime, aInterval, digests, instanceTypeTidb)
	if err != nil {
		log.Error(ctx, "aDigestTidbCpu failed", zap.Error(err))
	}
	bDigestTidbCpu, err := pcClientB.GetDigestsCpu(b.ClusterId, bQueryTime, bInterval, digests, instanceTypeTidb)
	if err != nil {
		log.Error(ctx, "bDigestTidbCpu failed", zap.Error(err))
	}
	aDigestTikvCpu, err := pcClientA.GetDigestsCpu(a.ClusterId, aQueryTime, aInterval, digests, instanceTypeTikv)
	if err != nil {
		log.Error(ctx, "aDigestTikvCpu failed", zap.Error(err))
	}
	bDigestTikvCpu, err := pcClientB.GetDigestsCpu(b.ClusterId, bQueryTime, bInterval, digests, instanceTypeTikv)
	if err != nil {
		log.Error(ctx, "bDigestTikvCpu failed", zap.Error(err))
	}

	aDigestAvgLatenc, err := pcClientA.GetDigestsAvgLatency(a.ClusterId, aQueryTime, aInterval, digests)
	if err != nil {
		log.Error(ctx, "aDigestAvgLatenc failed", zap.Error(err))
	}
	bDigestAvgLatenc, err := pcClientB.GetDigestsAvgLatency(b.ClusterId, bQueryTime, bInterval, digests)
	if err != nil {
		log.Error(ctx, "bDigestAvgLatenc failed", zap.Error(err))
	}
	aDigestCount, err := pcClientA.GetDigestsCount(a.ClusterId, aQueryTime, aInterval, digests)
	if err != nil {
		log.Error(ctx, "aDigestCount failed", zap.Error(err))
	}
	bDigestCount, err := pcClientB.GetDigestsCount(b.ClusterId, bQueryTime, bInterval, digests)
	if err != nil {
		log.Error(ctx, "bDigestCount failed", zap.Error(err))
	}
	aDigestSql, err := pcClientA.GetDigestsSql(a.ClusterId, aQueryTime, digests)
	if err != nil {
		log.Error(ctx, "aDigestSql failed", zap.Error(err))
	}
	bDigestSql, err := pcClientB.GetDigestsSql(b.ClusterId, bQueryTime, digests)
	if err != nil {
		log.Error(ctx, "bDigestSql failed", zap.Error(err))
	}

	// log.PlainInfo("aDigestTidbCpu success", zap.Any("aDigestTidbCpu", aDigestTidbCpu))
	// log.PlainInfo("bDigestTidbCpu success", zap.Any("bDigestTidbCpu", bDigestTidbCpu))
	// log.PlainInfo("aDigestTikvCpu success", zap.Any("aDigestTikvCpu", aDigestTikvCpu))
	// log.PlainInfo("bDigestTikvCpu success", zap.Any("bDigestTikvCpu", bDigestTikvCpu))
	// log.PlainInfo("aDigestAvgLatenc success", zap.Any("aDigestAvgLatenc", aDigestAvgLatenc))
	// log.PlainInfo("bDigestAvgLatenc success", zap.Any("bDigestAvgLatenc", bDigestAvgLatenc))
	// log.PlainInfo("aDigestCount success", zap.Any("aDigestCount", aDigestCount))
	// log.PlainInfo("bDigestCount success", zap.Any("bDigestCount", bDigestCount))

	for d := range union {
		// count prefer batch maps
		aCnt := aTopCount[d]
		if v, ok := aDigestCount[d]; ok {
			aCnt = v.Value
		} else if _, ok := aTopCount[d]; !ok {
			if v, err := pcClientA.GetTopSqlCount(a.ClusterId, d, aQueryTime, aInterval); err == nil {
				aCnt = v
			}
		}
		bCnt := bTopCount[d]
		if v, ok := bDigestCount[d]; ok {
			bCnt = v.Value
		} else if _, ok := bTopCount[d]; !ok {
			if v, err := pcClientB.GetTopSqlCount(b.ClusterId, d, bQueryTime, bInterval); err == nil {
				bCnt = v
			}
		}

		// cpu and latency from batch maps; missing digests default to 0
		var aCpu, bCpu, aCpuTikv, bCpuTikv, aLat, bLat float64
		if v, ok := aDigestTidbCpu[d]; ok {
			aCpu = v.Value
		}
		if v, ok := bDigestTidbCpu[d]; ok {
			bCpu = v.Value
		}
		if v, ok := aDigestTikvCpu[d]; ok {
			aCpuTikv = v.Value
		}
		if v, ok := bDigestTikvCpu[d]; ok {
			bCpuTikv = v.Value
		}
		if v, ok := aDigestAvgLatenc[d]; ok {
			aLat = v.Value
		}
		if v, ok := bDigestAvgLatenc[d]; ok {
			bLat = v.Value
		}
		var dCnt float64
		if aCnt == 0 {
			dCnt = 0
		} else {
			dCnt = *utils.CalculateDiff(bCnt, aCnt)
		}
		var dCpu float64
		if aCpu == 0 {
			dCpu = 0
		} else {
			dCpu = *utils.CalculateDiff(bCpu, aCpu)
		}
		var dCpuTikv float64
		if aCpuTikv == 0 {
			dCpuTikv = 0
		} else {
			dCpuTikv = *utils.CalculateDiff(bCpuTikv, aCpuTikv)
		}
		var dLat float64
		if aLat == 0 {
			dLat = 0
		} else {
			dLat = *utils.CalculateDiff(bLat, aLat)
		}
		label := ""
		if _, inBTop := bTopSet[d]; inBTop {
			if aCnt == 0 {
				label = "Added"
				// added++
				addedArr = append(addedArr, d)
			}
		}
		if label == "" {
			if _, inATop := aTopSet[d]; inATop {
				if bCnt == 0 {
					label = "Removed"
					// removed++
					removedArr = append(removedArr, d)
				}
			}
		}
		if aLat > 0 && bLat > 0 {
			if dLat <= -50 {
				dec50PlusDigests = append(dec50PlusDigests, d)
			} else if dLat <= -20 {
				dec50To20Digests = append(dec50To20Digests, d)
			} else if dLat < -5 {
				dec20To0Digests = append(dec20To0Digests, d)
			} else if dLat <= 5 {
				neg5To5Digests = append(neg5To5Digests, d)
			} else if dLat <= 20 {
				inc0To20Digests = append(inc0To20Digests, d)
			} else if dLat <= 50 {
				inc20To50Digests = append(inc20To50Digests, d)
			} else {
				inc50PlusDigests = append(inc50PlusDigests, d)
			}
			if label == "" {
				if dLat >= -5 && dLat <= 5 {
					label = "Stable"
				} else if dLat < 0 {
					label = "ImprovedPerformance"
					improvedArr = append(improvedArr, d)
				} else if dLat > 0 {
					label = "DegradedPerformance"
					degradedArr = append(degradedArr, d)
				}
			}
		}
		maxCpuTimeIn5MinWindow := dto.MaxCpuTimeIn5MinWindow{}
		if aCnt > 0 {
			if info, err := getMaxCpuTimeIn5MinWindow(ctx, *pcClientA, a, d); err == nil {
				maxCpuTimeIn5MinWindow.ClusterA = info
			}
		}
		if bCnt > 0 {
			if info, err := getMaxCpuTimeIn5MinWindow(ctx, *pcClientB, b, d); err == nil {
				maxCpuTimeIn5MinWindow.ClusterB = info
			}
		}
		sql := ""
		if v, ok := aDigestSql[d]; ok {
			sql = v.Sql
		}
		if sql == "" {
			if v, ok := bDigestSql[d]; ok {
				sql = v.Sql
			}
		}
		infos := []dto.DiffInfo{{Key: "topsql_count", ClusterA: aCnt, ClusterB: bCnt, Diff: dCnt}, {Key: "topsql_tidb_cpu_time", ClusterA: utils.RoundFloat(aCpu, 2), ClusterB: utils.RoundFloat(bCpu, 2), Diff: dCpu}, {Key: "topsql_tikv_cpu_time", ClusterA: utils.RoundFloat(aCpuTikv, 2), ClusterB: utils.RoundFloat(bCpuTikv, 2), Diff: dCpuTikv}, {Key: "topsql_avg_latency", ClusterA: utils.RoundFloat(aLat, 2), ClusterB: utils.RoundFloat(bLat, 2), Diff: dLat}}
		digestDiff = append(digestDiff, dto.DigestDiffEntry{Digest: d, SQL: sql, Label: label, DiffInfos: infos, MaxCpuTimeIn5MinWindow: &maxCpuTimeIn5MinWindow})
	}
	SortDigestByKeyDesc(digestDiff, "avg_latency")
	grp := dto.DiffGroup{DigestDiff: digestDiff}
	buckets := []map[string]any{
		{"[<-50]": len(dec50PlusDigests), "digests": dec50PlusDigests},
		{"[-50~-20]": len(dec50To20Digests), "digests": dec50To20Digests},
		{"[-20~-5]": len(dec20To0Digests), "digests": dec20To0Digests},
		{"[-5~5]": len(neg5To5Digests), "digests": neg5To5Digests},
		{"[5~20]": len(inc0To20Digests), "digests": inc0To20Digests},
		{"[20~50]": len(inc20To50Digests), "digests": inc20To50Digests},
		{"[>50]": len(inc50PlusDigests), "digests": inc50PlusDigests},
	}
	sum := dto.DiffGroupSummary{
		Added: len(addedArr),
		// AddedDigests:        addedArr,
		Removed:             len(removedArr),
		ImprovedPerformance: len(improvedArr),
		DegradedPerformance: len(degradedArr),
		Stable:              len(neg5To5Digests),
		DiffDetail:          &dto.DiffDetail{Cloumn: "topsql_avg_latency", Buckets: buckets},
	}
	return &grp, sum
}

func getMaxCpuTimeIn5MinWindow(ctx context.Context, pcClient prometheus.PrometheusClient, cluster dto.ClusterSideDTO, digest string) (dto.MaxCpuTimeInfo, error) {
	interval := fmt.Sprintf("%ds", int(cluster.TimeRange.End.Sub(cluster.TimeRange.Start).Seconds()))
	queryTime := cluster.TimeRange.End.UTC()
	resp, err := pcClient.GetMaxCPUTimeIn5MinWindow(ctx, cluster.ClusterId, digest, interval, queryTime)
	if err != nil {
		return dto.MaxCpuTimeInfo{}, err
	}
	if resp == nil {
		return dto.MaxCpuTimeInfo{}, nil
	}
	return dto.MaxCpuTimeInfo{MaxCpuTime: resp.CpuTime, Instance: resp.Instance, Timestamp: resp.Timestamp.Format(time.RFC3339)}, nil
}

func extractByKey(d dto.DigestDiffEntry, key string) (float64, bool) {
	for _, info := range d.DiffInfos {
		if info.Key == key {
			return info.ClusterB, true
		}
	}
	return 0, false
}

func SortDigestByKeyDesc(digests []dto.DigestDiffEntry, key string) {
	sort.SliceStable(digests, func(i, j int) bool {
		vi, okI := extractByKey(digests[i], key)
		vj, okJ := extractByKey(digests[j], key)

		if okI && okJ {
			return vi > vj // desc
		}
		if okI != okJ {
			return okI
		}
		return false
	})
}

func SetClusterBasicInfo(ctx context.Context, diffParam *dto.QueryComparisonRequest) error {

	clusterA, err := common.GetCluster(ctx, diffParam.ClusterA.ClusterId)
	if err != nil {
		log.Error(ctx, "SetClusterBasicInfo failed", zap.Error(err))
		return err
	}
	// if diffParam.ClusterA.ClusterUrls.ClusterMetricsURL == "" {
	// 	diffParam.ClusterA.Region = clusterA.Region
	// }
	if diffParam.ClusterA.Provider == "" {
		diffParam.ClusterA.Provider = clusterA.Provider
	}
	if diffParam.ClusterA.TenantID == "" {
		diffParam.ClusterA.TenantID = clusterA.OrgID
	}

	if diffParam.ClusterA.ProjectID == "" {
		diffParam.ClusterA.ProjectID = clusterA.ProjectID
	}
	if diffParam.ClusterA.DeployType == "" {
		diffParam.ClusterA.DeployType = clusterA.DeployType
	}
	if diffParam.ClusterA.Region == "" {
		diffParam.ClusterA.Region = clusterA.Region
	}

	clusterB, err := common.GetCluster(ctx, diffParam.ClusterB.ClusterId)
	if err != nil {
		log.Error(ctx, "SetClusterBasicInfo failed", zap.Error(err))
		return err

	}
	if diffParam.ClusterB.Region == "" {
		diffParam.ClusterB.Region = clusterB.Region
	}
	if diffParam.ClusterB.Provider == "" {
		diffParam.ClusterB.Provider = clusterB.Provider
	}
	if diffParam.ClusterB.TenantID == "" {
		diffParam.ClusterB.TenantID = clusterB.OrgID
	}
	if diffParam.ClusterB.ProjectID == "" {
		diffParam.ClusterB.ProjectID = clusterB.ProjectID
	}

	if diffParam.ClusterB.DeployType == "" {
		diffParam.ClusterB.DeployType = clusterB.DeployType
	}

	return nil
}