package service

import (
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/common"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/dto"
)

const (
	TopDigest = 100

	// DiffCPUThreshold = 50.0
	// DiffCntThreshold = 50.0
)

type ClusterDiffParam struct {
	ClusterA       ClusterInfo
	ClusterB       ClusterInfo
	TopN           int
	DigestDiffKeys []string
	DigestSources  []string
}
type ClusterInfo struct {
	ClusterId   string
	TimeRange   dto.TimeRange
	Alias       string
	IsDelete    bool
	Provider    string
	Region      string
	DeployType  string
	TenantID    string
	ProjectID   string
	DataPlaneID string
}
type ClusterCount struct {
	DigestCount int64
	TotalCount  int64
}

type ClusterDiffService struct {
	commonDAO common.CommonDAO
}

func NewClusterDiffService() *ClusterDiffService {
	return &ClusterDiffService{
		commonDAO: common.NewCommonDAO(),
	}
}

func DefaultTopN() int { return TopDigest }

// func (s *ClusterDiffService) TopDigesDiff(ctx context.Context, diffParam ClusterDiffParam) error {
// 	clusterA := config.GetClusterByID(diffParam.ClusterA.ClusterId)
// 	clusterB := config.GetClusterByID(diffParam.ClusterB.ClusterId)
// 	// org := config.GetOrgByClusterID(clusterA.ClusterID)

// 	// 1. query clusterA top X digest from deltalake
// 	deltaScanA, err := duckdb.GetDeltaScan(clusterA.ClusterID)
// 	if err != nil {
// 		log.PlainError("get delta scan failed: %v", zap.Error(err))
// 		return err
// 	}

// 	query := fmt.Sprintf(duckdb.SQLTopADigest, deltaScanA)

// 	args := []any{
// 		diffParam.ClusterA.TimeRange.Start.UTC().Unix(),
// 		diffParam.ClusterA.TimeRange.End.UTC().Unix()}

// 	// clusterASlowLogs, err := duckdb.QuerySlowLogBySql(clusterA.ClusterID, query, args...)
// 	req := duckdb.RunDuckDBQueryReq{
// 		ClusterId: clusterA.ClusterID,
// 		SQL:       query,
// 		IsLocal:   true,
// 		Args:      args,
// 	}
// 	clusterASlowLogs, err := duckdb.RunDuckDBQuery[duckdb.DigestInfo](req)
// 	if err != nil {
// 		log.PlainError("query slow logs failed: %v", zap.Error(err))
// 		return err
// 	}
// 	// log.PlainInfo("clusterASlowLogs", zap.Any("clusterASlowLogs", clusterASlowLogs))
// 	log.PlainInfo("clusterASlowLogs", zap.Any("clusterASlowLogs size", len(clusterASlowLogs)))

// 	//2、 get top100 digest
// 	topXDigestInfos := getTopXDigestInfos(clusterASlowLogs)

// 	log.PlainInfo("topXDigestInfos", zap.Any("topXDigestInfos size", len(topXDigestInfos)))

// 	// 3. delete old diff digest
// 	err = s.commonDAO.DeleteAll(ctx, &model.DiffDigest{})
// 	if err != nil {
// 		log.PlainError("Failed to delete diff digest", zap.Error(err))
// 		return err
// 	}
// 	// 4. save clusterA top X digest
// 	if err = s.saveDiffDigest(ctx, diffParam.ClusterA, topXDigestInfos); err != nil {
// 		log.PlainError("Failed to save diff digest", zap.Error(err))
// 		return err
// 	}
// 	digestArr := make([]string, 0, len(topXDigestInfos))
// 	for _, digestInfo := range topXDigestInfos {
// 		digestArr = append(digestArr, digestInfo.Digest)
// 	}

// 	inClause, inArgs := BuildInClause(digestArr)
// 	// Use Cluster B when building its delta scan source
// 	deltaScanB, err := duckdb.GetDeltaScan(clusterB.ClusterID)
// 	if err != nil {
// 		log.PlainError("get delta scan failed: %v", zap.Error(err))
// 		return err
// 	}

// 	queryB := fmt.Sprintf(duckdb.SQLTopBDigest, deltaScanB, inClause)
// 	// Flatten IN clause arguments: pass each digest as its own arg
// 	args = make([]any, 0, len(inArgs)+2)
// 	args = append(args, inArgs...)
// 	args = append(args, diffParam.ClusterB.TimeRange.Start.UTC().Unix())
// 	args = append(args, diffParam.ClusterB.TimeRange.End.UTC().Unix())
// 	req = duckdb.RunDuckDBQueryReq{
// 		ClusterId: clusterB.ClusterID,
// 		SQL:       queryB,
// 		IsLocal:   true,
// 		Args:      args,
// 	}
// 	clusterBSlowLogs, err := duckdb.RunDuckDBQuery[duckdb.DigestInfo](req)
// 	if err != nil {
// 		log.PlainError("query slow logs failed: %v", zap.Error(err))
// 		return err
// 	}
// 	log.PlainInfo("clusterBSlowLogs", zap.Any("clusterBSlowLogs size", len(clusterBSlowLogs)))
// 	// log.PlainInfo("clusterBSlowLogs", zap.Any("clusterBSlowLogs", clusterBSlowLogs))

// 	if err := s.saveDiffDigest(ctx, diffParam.ClusterB, clusterBSlowLogs); err != nil {
// 		log.PlainError("Failed to save diff digest", zap.Error(err))
// 		return err
// 	}

// 	return nil
// }
// func (s *ClusterDiffService) TopSqlTopDigesDiff(ctx context.Context, diffParam ClusterDiffParam) error {
// 	clusterA := config.GetClusterByID(diffParam.ClusterA.ClusterId)
// 	clusterB := config.GetClusterByID(diffParam.ClusterB.ClusterId)
// 	startTimeA := diffParam.ClusterA.TimeRange.Start
// 	endTimeA := diffParam.ClusterA.TimeRange.End
// 	endTimeB := diffParam.ClusterB.TimeRange.End

// 	subTime := endTimeA.Sub(startTimeA)
// 	// org := config.GetOrgByClusterID(clusterA.ClusterID)
// 	interval := fmt.Sprintf("%dm", int(subTime.Minutes()))
// 	// 1. get top100 digest from prometheus  (tidb)
// 	topXDigestInfos, err := prometheus.GetTopXCPUTime(clusterA.ClusterID, interval, TopDigest, endTimeA)
// 	if err != nil {
// 		log.PlainError("query slow logs failed: %v", zap.Error(err))
// 		return err
// 	}
// 	// log.PlainInfo("clusterASlowLogs", zap.Any("clusterASlowLogs", clusterASlowLogs))
// 	log.PlainInfo("clusterASlowLogs", zap.Any("clusterASlowLogs size", len(topXDigestInfos)))

// 	var TopDiges []model.TopSqlDiff
// 	//2、 query execution count，avg latency
// 	for _, digestInfo := range topXDigestInfos {
// 		digest := digestInfo.Digest
// 		// clusterId := clusterA.ClusterID

// 		countA, err := prometheus.GetTopSqlCount(clusterA.ClusterID, digestInfo.Digest, endTimeA, interval)
// 		if err != nil {
// 			log.PlainError("query prometheus failed: %v", zap.Error(err))
// 			return err
// 		}

// 		latencyA, err := prometheus.GetAvgLatency(clusterA.ClusterID, endTimeA, interval, digest)
// 		if err != nil {
// 			log.PlainError("GetAvgLatency failed", zap.String("digest", digest), zap.Error(err))
// 			return err
// 		}

// 		cpuTimeA := digestInfo.Value / 1000 / 60 // minute
// 		cpuTimeA = utils.RoundFloat(cpuTimeA)
// 		digestA := model.TopSqlDiff{
// 			ClusterId:    clusterA.ClusterID,
// 			ClusterAlias: clusterA.Alias,
// 			Digest:       digestInfo.Digest,
// 			Count:        int64(countA),
// 			AvgLatency:   utils.RoundFloat(latencyA), //ms
// 			TotalCpuTime: cpuTimeA,                   // minute

// 		}

// 		countB, err := prometheus.GetTopSqlCount(clusterB.ClusterID, digestInfo.Digest, endTimeB, interval)
// 		if err != nil {
// 			log.PlainError("query prometheus failed: %v", zap.Error(err))
// 			return err
// 		}
// 		if countB == 0 {
// 			continue
// 		}
// 		totalCpuB, err := prometheus.GetTotalCpuByTime(clusterB.ClusterID, endTimeB, interval, digest)
// 		if err != nil {
// 			log.PlainError("query prometheus failed: %v", zap.Error(err))
// 			return err
// 		}

// 		latencyB, err := prometheus.GetAvgLatency(clusterB.ClusterID, endTimeB, interval, digest)
// 		if err != nil {
// 			log.PlainError("GetAvgLatency failed", zap.String("digest", digest), zap.Error(err))
// 			return err
// 		}

// 		digestB := model.TopSqlDiff{
// 			ClusterId:    clusterB.ClusterID,
// 			ClusterAlias: clusterB.Alias,
// 			Digest:       digestInfo.Digest,
// 			TotalCpuTime: utils.RoundFloat(totalCpuB),
// 			Count:        int64(countB),
// 			AvgLatency:   utils.RoundFloat(latencyB), //ms
// 		}
// 		TopDiges = append(TopDiges, digestA)
// 		TopDiges = append(TopDiges, digestB)
// 	}

// 	log.PlainInfo("TopDiges", zap.Any("topXDigesTopDigestInfos size", len(TopDiges)))

// 	// 3. delete old diff digest
// 	err = s.commonDAO.DeleteAll(ctx, &model.TopSqlDiff{})
// 	if err != nil {
// 		log.PlainError("Failed to delete diff digest", zap.Error(err))
// 		return err
// 	}
// 	// 4. save clusterA top X digest
// 	if err = s.commonDAO.CreateEntity(ctx, &TopDiges); err != nil {
// 		log.PlainError("Failed to save diff digest", zap.Error(err))
// 		return err
// 	}

// 	return nil
// }

// func getTopXDigestInfos(clusterASlowLogs []duckdb.DigestInfo) []duckdb.DigestInfo {
// 	if len(clusterASlowLogs) <= TopDigest {
// 		return clusterASlowLogs
// 	}
// 	var topXSlowLogs []duckdb.DigestInfo
// 	for _, slowLog := range clusterASlowLogs {
// 		// filter  slow log
// 		if ShouldKeep(slowLog.Query) && len(topXSlowLogs) < TopDigest {
// 			topXSlowLogs = append(topXSlowLogs, slowLog)
// 		}
// 	}
// 	return topXSlowLogs
// }

// func (s *ClusterDiffService) QueryDigestCountDuckDB(ctx context.Context, diffParam ClusterDiffParam) error {
// 	log.Info(ctx, "QueryDigestCount ", zap.Any("diffParam", diffParam))

// 	clusterInfoArr := []ClusterInfo{diffParam.ClusterA, diffParam.ClusterB}

// 	var clusterACount duckdb.ClusterCount
// 	var clusterBCount duckdb.ClusterCount
// 	for _, c := range clusterInfoArr {

// 		cluster := config.GetClusterByID(c.ClusterId)

// 		deltaScan, err := duckdb.GetDeltaScan(cluster.ClusterID)
// 		if err != nil {
// 			log.PlainError("get delta scan failed: %v", zap.Error(err))
// 			return err
// 		}

// 		query := fmt.Sprintf(duckdb.DigestCountSQL, deltaScan)

// 		req := duckdb.RunDuckDBQueryReq{
// 			ClusterId: c.ClusterId,
// 			SQL:       query,
// 			IsLocal:   true,
// 			Args: []any{
// 				c.TimeRange.Start.UTC().Unix(),
// 				c.TimeRange.End.UTC().Unix()},
// 		}
// 		clusterCountArr, err := duckdb.RunDuckDBQuery[duckdb.ClusterCount](req)
// 		if err != nil {
// 			log.PlainError("query slow logs failed: %v", zap.Error(err))
// 			return err
// 		}
// 		clusterCount := clusterCountArr[0]
// 		clusterCount.ClusterId = c.ClusterId
// 		if c.ClusterId == diffParam.ClusterA.ClusterId {
// 			clusterACount = clusterCount
// 		} else {
// 			clusterBCount = clusterCount
// 		}
// 		// log.PlainInfo("clusterASlowLogs", zap.Any("clusterASlowLogs", clusterASlowLogs))

// 		log.PlainInfo("clusterCount  success", zap.Any("clusterCount ", clusterCount))
// 	}

// 	TotalRecordsDiff := utils.CalculateDiff(float64(clusterBCount.TotalRecords), float64(clusterACount.TotalRecords))
// 	distinctDigestDiff := utils.CalculateDiff(float64(clusterBCount.DistinctDigest), float64(clusterACount.DistinctDigest))
// 	log.PlainInfo(fmt.Sprintf("cluserA:%s,  cluserB:%s", diffParam.ClusterA.ClusterId, diffParam.ClusterB.ClusterId))
// 	log.PlainInfo(fmt.Sprintf("TotalRecordsDiff(B-A): %v%%", *TotalRecordsDiff))
// 	log.PlainInfo(fmt.Sprintf("distinctDigestDiff(B-A): %v%%", *distinctDigestDiff))
// 	return nil
// }

// /*
// func (s *ClusterDiffService) saveSlowLogs(ctx context.Context, slowLogs []duckdb.SlowLog) error {
// 	const batchSize = 5000  // insert 5000 per batch

// 	total := len(slowLogs)
// 	for start := 0; start < total; start += batchSize {
// 		end := start + batchSize
// 		if end > total {
// 			end = total
// 		}
// 		batch := slowLogs[start:end]

// 		if err := s.commonDAO.CreateEntity(ctx, &batch); err != nil {
// 			log.PlainError("Failed to save slow logs batch",
// 				zap.Int("batch_start", start),
// 				zap.Int("batch_end", end),
// 				zap.Error(err),
// 			)
// 			return err
// 		}
// 		log.PlainInfo("Saved slow logs batch successfully",
// 			zap.Int("batch_start", start),
// 			zap.Int("batch_end", end),
// 			zap.Int("batch_size", len(batch)),
// 		)
// 	}
// 	return nil
// }

// */

// func (s *ClusterDiffService) saveDiffDigest(ctx context.Context, clusterInfo ClusterInfo, slowLogs []duckdb.DigestInfo) error {
// 	log.PlainInfo("star to save DiffDigest", zap.Any("DiffDigest.len()", len(slowLogs)))
// 	if len(slowLogs) == 0 {
// 		log.PlainInfo("no slow logs to save:" + clusterInfo.ClusterId)
// 		return nil
// 	}
// 	cluster := config.GetClusterByID(clusterInfo.ClusterId)
// 	var diffDigestArr []model.DiffDigest
// 	// interval := "1d"
// 	// queryTimeA := clusterInfo.TimeRange.End.UTC()

// 	for _, slowQuery := range slowLogs {
// 		// digest := slowQuery.Digest
// 		// topSqlCpu, err := prometheus.GetTotalCpuByTime(clusterInfo.ClusterId, queryTimeA, interval, digest)
// 		// if err != nil {
// 		// 	return err
// 		// }

// 		// topSqlCount, err := prometheus.GetTopSqlCount(cluster.ClusterID, digest, queryTimeA, interval)
// 		// if err != nil {
// 		// 	log.PlainError("Failed to get total cpu by time", zap.Error(err))
// 		// 	return err
// 		// }

// 		diffDigest := model.DiffDigest{
// 			ClusterId:             cluster.ClusterID,
// 			ClusterAlias:          clusterInfo.Alias,
// 			Digest:                slowQuery.Digest,
// 			SlowQueryCount:        slowQuery.TotalCount,
// 			SlowQueryAvgQueryTime: slowQuery.AvgQueryTime,
// 			SlowQueryAvgMem:       slowQuery.AvgMem,
// 			Sql:                   slowQuery.Query,
// 			// TopSqlTotalCpu:        utils.RoundFloat(topSqlCpu),
// 			// TopSqlCount:           utils.RoundFloat(topSqlCount),
// 		}
// 		diffDigestArr = append(diffDigestArr, diffDigest)
// 	}
// 	if err := s.commonDAO.CreateEntity(ctx, &diffDigestArr); err != nil {
// 		log.PlainError("Failed to save diff digest", zap.Error(err))
// 		return err
// 	}
// 	log.PlainInfo("save diffDigest  success", zap.Any("diffDigest len", len(diffDigestArr)))

// 	return nil
// }