package service

import (
	"context"
	"errors"
	"fmt"
	"net/http"
	"sort"
	"strings"
	"time"

	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/common"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/config"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/dao"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/log"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/model"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/prometheus"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/s3"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/utils"
	"go.uber.org/zap"
	"gorm.io/gorm"
)

type TikvLatencyService struct {
	Day         string
	CommonDAO   common.CommonDAO
	clusterData map[string]model.ClusterData
	// TableName   string
	isAlert bool
}
type TimeWindow struct {
	Start time.Time
	End   time.Time
}

type TikvGrpcLatencyArr struct {
	TikvLatencyBeans []*TikvLatencyJsonBean
}
type TikvLatencyJsonBean struct {
	Instance               string
	MaxLatencyWindow       TimeWindow
	LatencyWindowTopDigest map[string][]*prometheus.DigestInfo
}

func NewTikvLatencyService(day ...string) *TikvLatencyService {
	s := &TikvLatencyService{
		CommonDAO:   common.NewCommonDAO(),
		clusterData: make(map[string]model.ClusterData),
	}
	if len(day) > 0 {
		s.Day = day[0]
	}
	return s
}

func (s *TikvLatencyService) mergeTimeWindows(ctx context.Context, beans []*prometheus.TikvLatencyBean) ([]string, error) {
	if len(beans) == 0 {
		return nil, nil
	}

	// Use map to store hour times, avoid duplicates
	hourMap := make(map[time.Time]bool)

	// Iterate through all beans
	for _, bean := range beans {

		// Process all TimeWindows in LatencyWindowTopDigest
		for timeWindow := range bean.LatencyWindowTopDigest {
			startHour := time.Date(
				timeWindow.Start.Year(),
				timeWindow.Start.Month(),
				timeWindow.Start.Day(),
				timeWindow.Start.Hour(),
				0, 0, 0, timeWindow.Start.UTC().Location(),
			)
			hourMap[startHour] = true

			endHour := time.Date(
				timeWindow.End.Year(),
				timeWindow.End.Month(),
				timeWindow.End.Day(),
				timeWindow.End.Hour(),
				0, 0, 0, timeWindow.End.UTC().Location(),
			)
			if !startHour.Equal(endHour) {
				hourMap[endHour] = true
			}
		}
	}

	// Convert map hour times to slice
	var hourSlice []time.Time
	for hour := range hourMap {
		hourSlice = append(hourSlice, hour)
	}

	// Sort by time
	sort.Slice(hourSlice, func(i, j int) bool {
		return hourSlice[i].Before(hourSlice[j])
	})
	var hours []string
	for _, hour := range hourSlice {

		log.PlainInfo("hour", zap.Any("hour", hour))
		hours = append(hours, hour.Format("2006010215"))
	}

	// Format hour times as string return (if needed)
	// if len(hourSlice) > 0 {
	// 	return hourSlice[0].Format("2006-01-02 15"), nil
	// }

	return hours, nil
}

func (s *TikvLatencyService) QueryTikvDuration30Min(ctx context.Context, startTime time.Time, clusterId ...string) error {
	var targetClusterId string
	// var targetOrg string

	if len(clusterId) > 0 {
		targetClusterId = clusterId[0]

	}

	windowEnd := startTime.Add(30 * time.Minute)
	window := TimeWindow{
		Start: startTime,
		End:   windowEnd,
	}
	// Query tikv grpc latency history range time
	historyWindowMap, err := s.queryTikvLatencyHistoryWindow(ctx, window)
	if err != nil {
		log.Error(ctx, "queryTikvLatencyHistoryWindow failed", zap.Error(err))
		return err
	}

	orgClusters, err := common.ListOrgClusters(ctx)
	if err != nil {
		log.Error(ctx, "ListEntities failed", zap.Error(err))
		return err
	}

	for orgId, clusters := range orgClusters {
		// skip other org when test
		// if orgId != targetOrg {
		// 	continue
		// }

		log.Info(ctx, fmt.Sprintf("==============[%s] org check start ==================", orgId))

		// 1. Get tikv grpc latency
		var tikvGrpcLatencyRespArr []*prometheus.TikvGrpcLatencyResp
		for _, cluster := range clusters {
			// respect skipCheck from config
			if cluster.SkipCheck {
				log.Info(ctx, fmt.Sprintf("skip cluster %s (%s)", cluster.ClusterID, cluster.Alias))
				continue
			}
			// skip other cluster when test
			if len(clusterId) > 0 && cluster.ClusterID != targetClusterId {
				continue
			}

			log.Info(ctx, fmt.Sprintf("----------[%s] tikv metric check  start ---------------", cluster.Alias))
			clusterID := cluster.ClusterID
			var hoursArr []string
			req, err := BuildPrometheusClinetReq(ctx, clusterID)
			if err != nil {
				return fmt.Errorf("failed to build prometheus client req: %w", err)
			}
			pc, err := prometheus.CreatePrometheusClient(ctx, req)
			if err != nil {
				return fmt.Errorf("failed to create prometheus client: %w", err)
			}

			tikvGrpcLatencyResp, err := pc.GetTikvGrpcLatencyByTime(clusterID, window.Start, window.End, historyWindowMap)
			if err != nil {
				log.Error(ctx, "getTikvGrpcLatencyByTime failed", zap.Error(err))
				return err
			}
			if len(tikvGrpcLatencyResp.TikvLatencyBeans) == 0 {
				continue
			}
			tikvGrpcLatencyRespArr = append(tikvGrpcLatencyRespArr, tikvGrpcLatencyResp)

			hours, _ := s.mergeTimeWindows(ctx, tikvGrpcLatencyResp.TikvLatencyBeans)
			if len(hours) > 0 {
				hoursArr = append(hoursArr, hours...)
			}

			if len(hoursArr) == 0 {
				continue
			}
			// 2. Download s3 file
			err = s.downloadS3File(ctx, clusterID, hoursArr)
			if err != nil {
				log.Error(ctx, "downloadS3File failed", zap.Error(err))
				return err
			}
			log.Info(ctx, "hoursArr", zap.Any("hoursArr", hoursArr))
			log.Info(ctx, fmt.Sprintf("----------[%s] tikv metric check  end ---------------", cluster.Alias))
		}

		if len(tikvGrpcLatencyRespArr) == 0 {
			continue
		}
		// 3. Process s3 file
		tableName := fmt.Sprintf("tikv_latency_%s", orgId)
		s3Stats, err := s3.ProcessLocalS3Logs(ctx, orgId, tableName)
		if err != nil {
			log.Error(ctx, "Failed to process local S3 logs", zap.Error(err))
			return err
		}
		log.Info(ctx, "s3Stats", zap.Any("s3Stats", s3Stats))

		// 4. Query slow query
		err = s.setSlowQuery(ctx, tableName, tikvGrpcLatencyRespArr)
		if err != nil {
			log.Error(ctx, "setSlowQuery failed", zap.Error(err))
			return err
		}
		// Save to db
		err = s.saveTikvLatency(ctx, orgId, tikvGrpcLatencyRespArr, window)
		if err != nil {
			log.Error(ctx, "saveTikvLatency failed", zap.Error(err))
			return err
		}
		log.Info(ctx, fmt.Sprintf("==============[%s] org check end ==================", orgId))
	}
	if s.isAlert {
		// 5. Call back trigger
		callBackTrigger(ctx)

	}

	// Delete old files in tikvInputDir
	if !config.IsDev() {
		err = utils.DeleteOldFiles(config.AppConfig.TikvInputDir, 2*time.Hour)
		if err != nil {
			log.Error(ctx, "Failed to delete old files in tikvInputDir", zap.Error(err))
			return err
		}
		err = utils.DeleteOldFiles(config.AppConfig.Database.TikvDbPath, 12*time.Hour)
		if err != nil {
			log.Error(ctx, "Failed to delete old files in tikvDbPath", zap.Error(err))
			return err
		}
	}
	return nil
}

// func (s *TikvLatencyService) QueryTikvDuration30MinDuckDB(ctx context.Context, startTime time.Time, clusterId ...string) error {
// 	var targetClusterId string
// 	var targetOrg string

// 	if len(clusterId) > 0 {
// 		targetClusterId = clusterId[0]
// 		targetOrg = config.GetOrgByClusterID(targetClusterId).OrgID
// 	}

// 	windowEnd := startTime.Add(30 * time.Minute)
// 	window := TimeWindow{
// 		Start: startTime,
// 		End:   windowEnd,
// 	}
// 	// Query tikv grpc latency history range time
// 	historyWindowMap, err := s.queryTikvLatencyHistoryWindow(ctx, window)
// 	if err != nil {
// 		log.Error(ctx, "queryTikvLatencyHistoryWindow failed", zap.Error(err))
// 		return err
// 	}
// 	for _, org := range config.OrgsConfig.Orgs {
// 		// skip other org when test
// 		if len(clusterId) > 0 && org.OrgID != targetOrg {
// 			continue
// 		}

// 		log.PlainInfo(fmt.Sprintf("==============[%s] org check start ==================", org.OrgName))

// 		// 1. Get tikv grpc latency
// 		var tikvGrpcLatencyRespArr []*prometheus.TikvGrpcLatencyResp
// 		for _, cluster := range org.Clusters {
// 			// respect skipCheck from config
// 			if cluster.SkipCheck {
// 				log.PlainInfo(fmt.Sprintf("skip cluster %s (%s)", cluster.ClusterID, cluster.Alias))
// 				continue
// 			}
// 			// skip other cluster when test
// 			if len(clusterId) > 0 && cluster.ClusterID != targetClusterId {
// 				continue
// 			}

// 			log.PlainInfo(fmt.Sprintf("----------[%s] tikv metric check  start ---------------", cluster.Alias))
// 			clusterID := cluster.ClusterID
// 			var hoursArr []string

// 			tikvGrpcLatencyResp, err := prometheus.GetTikvGrpcLatencyByTime(clusterID, window.Start, window.End, historyWindowMap)
// 			if err != nil {
// 				log.Error(ctx, "getTikvGrpcLatencyByTime failed", zap.Error(err))
// 				return err
// 			}
// 			if len(tikvGrpcLatencyResp.TikvLatencyBeans) == 0 {
// 				continue
// 			}
// 			tikvGrpcLatencyRespArr = append(tikvGrpcLatencyRespArr, tikvGrpcLatencyResp)

// 			hours, _ := s.mergeTimeWindows(ctx, tikvGrpcLatencyResp.TikvLatencyBeans)
// 			if len(hours) > 0 {
// 				hoursArr = append(hoursArr, hours...)
// 			}

// 			log.PlainInfo("hoursArr", zap.Any("hoursArr", hoursArr))
// 			log.PlainInfo(fmt.Sprintf("----------[%s] tikv metric check  end ---------------", cluster.Alias))
// 		}

// 		if len(tikvGrpcLatencyRespArr) == 0 {
// 			continue
// 		}

// 		// 3. Query slow query
// 		err = s.setSlowQueryDuckDB(ctx, tikvGrpcLatencyRespArr)
// 		if err != nil {
// 			log.Error(ctx, "setSlowQueryDuckDB failed", zap.Error(err))
// 			return err
// 		}
// 		// 4. Save to db
// 		err = s.saveTikvLatency(ctx, org.OrgID, tikvGrpcLatencyRespArr, window)
// 		if err != nil {
// 			log.Error(ctx, "saveTikvLatency failed", zap.Error(err))
// 			return err
// 		}
// 		log.PlainInfo(fmt.Sprintf("==============[%s] org check end ==================", org.OrgName))
// 	}
// 	// 5. Call back trigger
// 	if s.isAlert {
// 		callBackTrigger()

// 	}

// 	// Delete old files in tikvInputDir
// 	if !config.IsDev() {
// 		err = utils.DeleteOldFiles(config.AppConfig.TikvInputDir, 2*time.Hour)
// 		if err != nil {
// 			log.PlainError("Failed to delete old files in tikvInputDir", zap.Error(err))
// 			return err
// 		}
// 		err = utils.DeleteOldFiles(config.AppConfig.Database.TikvDbPath, 12*time.Hour)
// 		if err != nil {
// 			log.PlainError("Failed to delete old files in tikvDbPath", zap.Error(err))
// 			return err
// 		}
// 	}
// 	return nil
// }

func (s *TikvLatencyService) queryTikvLatencyHistoryWindow(ctx context.Context, window TimeWindow) (map[string][]prometheus.TimeWindow, error) {
	var tikvLatencyArr []model.TikvLatency
	// Query TikvLatency that intersects with window
	param := common.ListParams{
		ModelFilters: model.TikvLatency{},
		AdvancedFilters: map[string]any{
			"range_end_time < (?)": window.Start.Unix(),
		},
	}
	if err := s.CommonDAO.ListEntities(ctx, param, &tikvLatencyArr); err != nil {
		log.Error(ctx, "ListEntities failed for TikvLatency", zap.Error(err))
		return nil, err
	}
	historyWindowMap := make(map[string][]prometheus.TimeWindow)
	for _, latency := range tikvLatencyArr {
		tw := prometheus.TimeWindow{
			Start: time.Unix(latency.RangeStartTime, 0).UTC(),
			End:   time.Unix(latency.RangeEndTime, 0).UTC(),
		}
		historyWindowMap[latency.ClusterId] = append(historyWindowMap[latency.ClusterId], tw)
	}
	return historyWindowMap, nil
}

func (s *TikvLatencyService) saveTikvLatency(ctx context.Context, orgId string, tikvGrpcLatencyRespArr []*prometheus.TikvGrpcLatencyResp, window TimeWindow) error {
	// 1. Calculate query range
	rangeStartTime := window.Start.Unix()
	rangeEndTime := window.End.Unix()

	// 2. Convert to TikvLatency array
	tikvLatencyArr, err := convertToTikvLatencyArr(orgId, tikvGrpcLatencyRespArr, rangeStartTime, rangeEndTime)
	if err != nil {
		log.Error(ctx, "convertToTikvLatencyArr failed", zap.Error(err))
		return err
	}

	// 3. First delete old data, then save new data (TikvLatency)
	var delTikvLatencyArr []model.TikvLatency
	if len(tikvLatencyArr) > 0 {
		// set isAlert to true
		s.isAlert = true
		deleteBeans := model.TikvLatency{
			RangeStartTime: rangeStartTime,
			RangeEndTime:   rangeEndTime,
			OrgId:          orgId,
		}

		// save TikvLatencyArr
		if err := s.CommonDAO.ListEntities(ctx, common.ListParams{ModelFilters: deleteBeans}, &delTikvLatencyArr); err != nil {
			log.Error(ctx, "ListEntities failed for TikvLatency", zap.Error(err))
			return err
		}
		log.PlainInfo("delTikvLatencyArr", zap.Any("len", len(delTikvLatencyArr)))
		deleBeans := &model.TikvLatency{}
		if err := s.CommonDAO.DeleteEntity(ctx, common.ComFilter{ModelFilters: deleteBeans}, deleBeans); err != nil {
			log.Error(ctx, "DeleteEntity failed for TikvLatency", zap.Error(err))
			return err
		}

		if err := s.CommonDAO.CreateEntity(ctx, &tikvLatencyArr); err != nil {
			log.Error(ctx, "CreateEntity failed for TikvLatency", zap.Error(err))
			return err
		}
		log.Info(ctx, "save TikvLatency success", zap.Int("count", len(tikvLatencyArr)))
	}

	// 4. Construct (clusterId|instance|latencyStartTime) -> Id mapping
	idMap := make(map[string]int64)
	for _, l := range tikvLatencyArr {
		key := fmt.Sprintf("%s|%s|%d", l.ClusterId, l.Instance, l.LatencyStartTime)
		idMap[key] = l.Id
	}

	// 5. Convert TikvTopDigest and write foreign key
	topDigestArr, err := convertToTikvTopDigestArr(tikvGrpcLatencyRespArr, idMap)
	if err != nil {
		log.Error(ctx, "convertToTikvTopDigestArr failed", zap.Error(err))
		return err
	}

	if len(topDigestArr) > 0 {

		var deleteIds []int64
		for _, l := range delTikvLatencyArr {
			deleteIds = append(deleteIds, l.Id)
		}
		// filter := common.ComFilter{
		// 	AdvancedFilters: map[string]any{
		// 		"tikv_latency_id in (?)": deleteIds,
		// 	},
		// }

		tdFilter := common.ComFilter{AdvancedFilters: map[string]any{"check_type = ?": slow_tikv_check_type, "check_window_id in (?)": deleteIds}}
		if err := s.CommonDAO.DeleteEntity(ctx, tdFilter, &model.TikvTopDigest{}); err != nil {
			log.Error(ctx, "DeleteEntity failed for TikvTopDigest", zap.Error(err))
			return err
		}
		if err := s.CommonDAO.CreateEntity(ctx, &topDigestArr); err != nil {
			log.Error(ctx, "CreateEntity failed for TikvTopDigest", zap.Error(err))
			return err
		}
		log.Info(ctx, "save TikvTopDigest success", zap.Int("count", len(topDigestArr)))
	}
	return nil
}

// convertToTikvLatencyArr - Complete LatencyStart/EndTime fields
func convertToTikvLatencyArr(orgId string, tikvGrpcLatencyRespArr []*prometheus.TikvGrpcLatencyResp, rangeStartTime, rangeEndTime int64) ([]*model.TikvLatency, error) {
	var tikvLatencyArr []*model.TikvLatency
	for _, resp := range tikvGrpcLatencyRespArr {
		clusterId := resp.ClusterId
		for _, tikvLatencyBean := range resp.TikvLatencyBeans {
			for _, lw := range tikvLatencyBean.LatencyWindows {
				tikvLatencyArr = append(tikvLatencyArr, &model.TikvLatency{
					OrgId:                  orgId,
					ClusterId:              clusterId,
					RangeStartTime:         rangeStartTime,
					RangeEndTime:           rangeEndTime,
					LatencyStartTime:       lw.LatencyWindow.Start.Unix(),
					LatencyEndTime:         lw.LatencyWindow.End.Unix(),
					ClusterLatencyP95:      lw.ClusterLatencyP95,
					ClusterLatencyP95Diff:  lw.ClusterLatencyP95Diff,
					ClusterLatencyP99:      lw.ClusterLatencyP99,
					ClusterLatencyP99Diff:  lw.ClusterLatencyP99Diff,
					ClusterLatencyP999:     lw.ClusterLatencyP999,
					ClusterLatencyP999Diff: lw.ClusterLatencyP999Diff,
					Instance:               tikvLatencyBean.Instance,
					InstanceMaxLatency:     tikvLatencyBean.MaxLatencyWindow.WindowMaxLatency,
					InstanceMaxCpu:         lw.InstanceMaxCpu,
					InstanceMaxCpuDiff:     lw.InstanceMaxCpuDiff,
					MaxLatencyWindow:       fmt.Sprintf("%s ~ %s", tikvLatencyBean.MaxLatencyWindow.Start.Format("2006-01-02 15:04:05"), tikvLatencyBean.MaxLatencyWindow.End.Format("2006-01-02 15:04:05")),
				})
			}
		}
	}
	return tikvLatencyArr, nil
}

// convertToTikvTopDigestArr - Use idMap to assign TikvLatencyId
func convertToTikvTopDigestArr(tikvGrpcLatencyRespArr []*prometheus.TikvGrpcLatencyResp, idMap map[string]int64) ([]model.TikvTopDigest, error) {
	var topDigestArr []model.TikvTopDigest
	for _, resp := range tikvGrpcLatencyRespArr {
		for _, tikvLatencyBean := range resp.TikvLatencyBeans {
			for timeWindow, digestInfoArr := range tikvLatencyBean.LatencyWindowTopDigest {
				if len(digestInfoArr) == 0 {
					continue
				}
				// Calculate key to get TikvLatencyId
				key := fmt.Sprintf("%s|%s|%d", resp.ClusterId, tikvLatencyBean.Instance, timeWindow.Start.Unix())
				tikvLatencyId := idMap[key]
				for _, digestInfo := range digestInfoArr {
					td := model.TikvTopDigest{
						TikvLatencyId: tikvLatencyId,
						CheckWindowId: tikvLatencyId,
						Digest:        digestInfo.Digest,
						CheckType:     slow_tikv_check_type,

						CpuTime:     digestInfo.Value,
						CpuPercent:  digestInfo.CpuPer,
						IsSlowQuery: digestInfo.SlowQueryInfo.IsInSlowQuery,
						TopSqlLink:  digestInfo.TopSqlLink,
					}
					if digestInfo.SlowQueryInfo.IsInSlowQuery {
						MAXMem := utils.RoundFloat(digestInfo.SlowQueryInfo.MAXMem)
						avgMem := utils.RoundFloat(digestInfo.SlowQueryInfo.AVGMem)
						maxQueryTime := utils.RoundFloat(digestInfo.SlowQueryInfo.MAXQueryTime)
						avgQueryTime := utils.RoundFloat(digestInfo.SlowQueryInfo.AVGQueryTime)
						td.SlowQueryCount = &digestInfo.SlowQueryInfo.TotalCount
						td.SlowQueryMaxMem = &MAXMem
						td.SlowQueryAvgMem = &avgMem
						td.SlowQueryMaxQueryTime = &maxQueryTime
						td.SlowQueryAvgQueryTime = &avgQueryTime
						td.SlowQueryDetailLink = digestInfo.SlowQueryInfo.DetailLink
						td.SlowQueryListLink = digestInfo.SlowQueryInfo.ListLink

					}

					topDigestArr = append(topDigestArr, td)
				}
			}
		}
	}
	return topDigestArr, nil
}

func (s *TikvLatencyService) setSlowQuery(ctx context.Context, tableName string, tikvGrpcLatencyRespArr []*prometheus.TikvGrpcLatencyResp) error {
	for _, resp := range tikvGrpcLatencyRespArr {
		clusterId := resp.ClusterId
		for _, bean := range resp.TikvLatencyBeans {
			if len(bean.LatencyWindowTopDigest) == 0 {
				continue
			}
			for timeWindow, digestInfoArr := range bean.LatencyWindowTopDigest {
				var digests []string
				for _, digestInfo := range digestInfoArr {
					digests = append(digests, strings.ToLower(digestInfo.Digest))
				}
				if len(digests) == 0 {
					continue
				}

				orgID, _ := common.GetOrgId(ctx, clusterId)
				dbPath := config.GetTikvDbPath(orgID)

				log.PlainInfo("dbPath:" + dbPath)
				if !utils.FileExists(dbPath) {
					log.PlainWarn("dbPath not exist", zap.String("dbPath", dbPath))
					return nil
				}
				// Initialize database manager, todo remove

				db, err := dao.InitSQLiteDB(dbPath)

				if err != nil {
					log.PlainError("Failed to initialize database", zap.Error(err))
					return err
				}

				startTime := timeWindow.Start.UTC().Unix()
				endTime := timeWindow.End.UTC().Unix()

				var slowQueryInfoArr []prometheus.SlowQueryInfo
				sqlStr := `
				SELECT 
					a.Digest, 
					ROUND(AVG(a.Mem_max), 0) AS AVGMem,
					MAX(a.Mem_max) AS MAXMem, 
					AVG(a.Query_time) AS AVGQueryTime, 
					MAX(a.Query_time) AS MAXQueryTime, 
					COUNT(*) AS TotalCount 
				 FROM %s a 
				WHERE a.Digest in (?) 
				and a.Timestamp >= ?
				and a.Timestamp <= ?
				and a.Cluster_Id = ?
				GROUP BY a.Digest
				`

				sql := fmt.Sprintf(sqlStr, tableName)
				db.Raw(sql, digests, startTime, endTime, clusterId).Scan(&slowQueryInfoArr)
				log.PlainInfo("slowQueryInfoArr", zap.Any("slowQueryInfoArr", slowQueryInfoArr))
				rangeTime := TimeWindow{
					Start: timeWindow.Start,
					End:   timeWindow.End,
				}
				for _, digestInfo := range digestInfoArr {
					topSqlLink := getTopSqlLink(ctx, clusterId, bean.Instance, rangeTime)
					if err != nil {
						log.PlainError("Failed to get digest link", zap.Error(err))
						continue
					}
					digestInfo.TopSqlLink = topSqlLink
					for _, slowQueryInfo := range slowQueryInfoArr {
						if strings.ToUpper(slowQueryInfo.Digest) == strings.ToUpper(digestInfo.Digest) {
							slowQueryInfo.IsInSlowQuery = true
							digestInfo.SlowQueryInfo = slowQueryInfo

							req := GetDetailLinkReq{

								Digest:       strings.ToLower(digestInfo.Digest),
								TimeWindow:   rangeTime,
								MaxQueryTime: digestInfo.SlowQueryInfo.MAXQueryTime,
							}

							// Set slow query DetailLink and
							linkResp, err := getDigestLink(ctx, db, tableName, req)
							if err != nil {
								log.PlainError("Failed to get digest link", zap.Error(err))
								continue
							}
							digestInfo.SlowQueryInfo.DetailLink = linkResp.DetailLink
							digestInfo.SlowQueryInfo.ListLink = linkResp.ListLink

						}
					}
				}
			}
		}
	}

	return nil
}

// func (s *TikvLatencyService) setSlowQueryDuckDB(ctx context.Context, tikvGrpcLatencyRespArr []*prometheus.TikvGrpcLatencyResp) error {
// 	for _, resp := range tikvGrpcLatencyRespArr {
// 		clusterId := resp.ClusterId
// 		for _, bean := range resp.TikvLatencyBeans {
// 			if len(bean.LatencyWindowTopDigest) == 0 {
// 				continue
// 			}
// 			for timeWindow, digestInfoArr := range bean.LatencyWindowTopDigest {
// 				var digests []string
// 				for _, digestInfo := range digestInfoArr {
// 					digests = append(digests, strings.ToLower(digestInfo.Digest))
// 				}
// 				if len(digests) == 0 {
// 					continue
// 				}

// 				startTime := timeWindow.Start.UTC().Unix()
// 				endTime := timeWindow.End.UTC().Unix()

// 				// 1. query clusterA top X digest from deltalake
// 				deltaScanA, err := duckdb.GetDeltaScan(clusterId)
// 				if err != nil {
// 					log.PlainError("get delta scan failed: %v", zap.Error(err))
// 					return err
// 				}
// 				// test
// 				// digests = []string{"a79c25853ebd9531db257e76eba3f5d16fb919301fa9925b52c014103ce5bee9", "035b1dede2a9ce72b0e7acec9a24c7f05a5ed49a6fbc224e6806e5c3414fc803"}
// 				inClause, inArgs := BuildInClause(digests)

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

// 				args := make([]any, 0, len(inArgs)+2)
// 				args = append(args, startTime)
// 				args = append(args, endTime)
// 				args = append(args, inArgs...)
// 				req := duckdb.RunDuckDBQueryReq{
// 					ClusterId: clusterId,
// 					SQL:       query,
// 					// IsLocal:   true,
// 					Args: args,
// 				}
// 				log.PlainInfo("start to query slowquery by digest (topsql top5 digest)")
// 				slowQueryArr, err := duckdb.RunDuckDBQuery[duckdb.DigestInfo](req)
// 				if err != nil {
// 					log.PlainError("query slow logs failed: %v", zap.Error(err))
// 					return err
// 				}
// 				var slowQueryInfoArr []prometheus.SlowQueryInfo
// 				for _, slowQueryInfo := range slowQueryArr {
// 					slowQueryInfoArr = append(slowQueryInfoArr, prometheus.SlowQueryInfo{
// 						Digest:       slowQueryInfo.Digest,
// 						AVGMem:       slowQueryInfo.AvgMem,
// 						MAXMem:       slowQueryInfo.MaxMem,
// 						AVGQueryTime: slowQueryInfo.AvgQueryTime,
// 						MAXQueryTime: slowQueryInfo.MaxQueryTime,
// 						TotalCount:   slowQueryInfo.TotalCount,
// 					})
// 				}
// 				log.PlainInfo("slowQueryInfoArr", zap.Any("slowQueryInfoArr", slowQueryInfoArr))
// 				rangeTime := TimeWindow{
// 					Start: timeWindow.Start,
// 					End:   timeWindow.End,
// 				}
// 				for _, digestInfo := range digestInfoArr {
// 					topSqlLink := getTopSqlLink(clusterId, bean.Instance, rangeTime)
// 					if err != nil {
// 						log.PlainError("Failed to get digest link", zap.Error(err))
// 						continue
// 					}
// 					digestInfo.TopSqlLink = topSqlLink
// 					for _, slowQueryInfo := range slowQueryInfoArr {
// 						if strings.ToUpper(slowQueryInfo.Digest) == strings.ToUpper(digestInfo.Digest) {
// 							slowQueryInfo.IsInSlowQuery = true
// 							digestInfo.SlowQueryInfo = slowQueryInfo

// 							req := GetDetailLinkReq{
// 								ClusterId:    clusterId,
// 								Digest:       strings.ToLower(digestInfo.Digest),
// 								TimeWindow:   rangeTime,
// 								MaxQueryTime: digestInfo.SlowQueryInfo.MAXQueryTime,
// 							}

// 							// Set slow query DetailLink and
// 							linkResp, err := GetDigestLinkDuckDB(req)
// 							if err != nil {
// 								log.PlainError("Failed to get digest link", zap.Error(err))
// 								continue
// 							}
// 							digestInfo.SlowQueryInfo.DetailLink = linkResp.DetailLink
// 							digestInfo.SlowQueryInfo.ListLink = linkResp.ListLink

// 						}
// 					}
// 				}
// 			}
// 		}
// 	}

// 	return nil
// }

func (s *TikvLatencyService) mergeTimeWindow(ctx context.Context, beans []*prometheus.TikvLatencyBean) ([]string, error) {
	if len(beans) == 0 {
		return nil, nil
	}

	// Use map to store hour times, avoid duplicates
	hourMap := make(map[time.Time]bool)

	// Iterate through all beans
	for _, bean := range beans {

		// Process all TimeWindows in LatencyWindowTopDigest
		for timeWindow := range bean.LatencyWindowTopDigest {
			startHour := time.Date(
				timeWindow.Start.Year(),
				timeWindow.Start.Month(),
				timeWindow.Start.Day(),
				timeWindow.Start.Hour(),
				0, 0, 0, timeWindow.Start.UTC().Location(),
			)
			hourMap[startHour] = true

			endHour := time.Date(
				timeWindow.End.Year(),
				timeWindow.End.Month(),
				timeWindow.End.Day(),
				timeWindow.End.Hour(),
				0, 0, 0, timeWindow.End.UTC().Location(),
			)
			if !startHour.Equal(endHour) {
				hourMap[endHour] = true
			}
		}
	}

	// Convert map hour times to slice
	var hourSlice []time.Time
	for hour := range hourMap {
		hourSlice = append(hourSlice, hour)
	}

	// Sort by time
	sort.Slice(hourSlice, func(i, j int) bool {
		return hourSlice[i].Before(hourSlice[j])
	})
	var hours []string
	for _, hour := range hourSlice {
		hours = append(hours, hour.Format("2006010215"))
	}
	return hours, nil
}

/*
	 func (s *TikvLatencyService) getTimeWindowArr() []TimeWindow {
		// 20250905
		day := s.Day
		startTime, _ := time.Parse("20060102", day)

		// 创建一个包含48个30分钟时间窗口的数组（一天24小时，每小时2个30分钟窗口）
		windows := make([]TimeWindow, 0, 48)

		// 遍历一天中的每个30分钟时间段
		for i := 0; i < 48; i++ {
			windowStart := startTime.Add(time.Duration(i*30) * time.Minute)
			windowEnd := windowStart.Add(30 * time.Minute)

			window := TimeWindow{
				Start: windowStart,
				End:   windowEnd,
			}

			windows = append(windows, window)
		}

		return windows
	}
*/
func (s *TikvLatencyService) downloadS3File(ctx context.Context, clusterId string, hours []string) error {

	// day := "20250817"
	// prefixHour := "21"
	for _, hour := range hours {
		// Extract date and hour from hour string
		day := hour[:8]     // Extract first 8 characters as date (20250817)
		hourStr := hour[8:] // Extract remaining characters as hour (17)
		// org := config.GetOrgByClusterID(clusterId)
		cluster, _ := common.GetCluster(ctx, clusterId)

		if cluster == nil {
			log.PlainError("Cluster configuration not found in db", zap.String("clusterId", clusterId))
			return errors.New("Cluster configuration not found")
		}
		saveveDir := fmt.Sprintf("%s/%s/%s/%s", config.AppConfig.TikvInputDir, day, cluster.OrgID, clusterId)
		err := s3.DownloadS3Files(ctx, clusterId, day, hourStr, saveveDir)
		if err != nil {
			log.Error(ctx, "DownloadS3Files failed", zap.Error(err))
			return err
		}

	}
	return nil
}

// func (s *TikvLatencyService) processLocalS3Log(ctx context.Context) error {
// 	org := config.OrgsConfig.Orgs[0]
// 	s3Stats, err := s3.ProcessLocalS3Logs(org)
// 	if err != nil {
// 		log.Error(ctx, "ProcessLocalS3Logs failed", zap.Error(err))
// 		return err
// 	}
// 	log.PlainInfo("s3Stats", zap.Any("s3Stats", s3Stats))
// 	return nil
// }

// func convertMapKeysToString(input map[prometheus.TimeWindow][]*prometheus.DigestInfo) map[string][]*prometheus.DigestInfo {
// 	output := make(map[string][]*prometheus.DigestInfo)
// 	for key, value := range input {

// 		keyStr := key.Start.Format("2006-01-02 15:04:05") + " ~ " + key.End.Format("2006-01-02 15:04:05")
// 		output[keyStr] = value
// 	}
// 	return output
// }

type GetDetailLinkReq struct {
	ClusterId    string
	Digest       string
	MaxQueryTime float64
	TimeWindow   TimeWindow
}
type GetDetailLinkResp struct {
	DetailLink string
	ListLink   string
}

func getDigestLink(ctx context.Context, gormDB *gorm.DB, tableName string, req GetDetailLinkReq) (GetDetailLinkResp, error) {
	resp := GetDetailLinkResp{}
	// ctx := context.Background()
	slowQueryLogMax := model.SlowQueryLog{}

	gormDB.Table(tableName).
		Where("Digest =? and Query_time = ?", req.Digest, req.MaxQueryTime).
		Group("Digest").
		Scan(&slowQueryLogMax)

	if slowQueryLogMax.QueryTime == 0 {
		return resp, nil
	}

	connID := fmt.Sprintf("%d", slowQueryLogMax.ConnID)
	formattedTimestamp := slowQueryLogMax.FormattedTimestamp
	cluster, _ := common.GetCluster(ctx, slowQueryLogMax.ClusterID)
	org := config.GetOrgByClusterID(slowQueryLogMax.ClusterID)

	deployType := cluster.DeployType
	if deployType == "" {
		deployType = "dedicated"
	}
	// Detail link
	resp.DetailLink = fmt.Sprintf("%s?provider=%s&region=%s&orgName=%s&orgId=%s&projectId=%s&clusterId=%s&clusterName=%s&deployType=%s#/slow_query/detail?digest=%s&connection_id=%s&timestamp=%s",
		config.AppConfig.Clinic.BaseUrl,
		cluster.Provider,
		cluster.Region,
		org.OrgName,
		org.OrgID,
		cluster.ProjectID,
		cluster.ClusterID,
		cluster.ClusterName,
		deployType,
		req.Digest,
		connID,
		formattedTimestamp)

	//   topSqlUrl: "https://clinic.pingcap.com/portal/dashboard/cloud/ngm.html?provider=%s&region=%s&orgId=%s&projectId=%s&clusterId=%s&clusterName=%s&deployType=dedicated#/topsql?from=%d&to=%d"

	// startTime := time.Date(year, time.Month(month), day, 0, 0, 0, 0, time.UTC)
	startTime := fmt.Sprintf("%d", req.TimeWindow.Start.Unix())
	endTime := fmt.Sprintf("%d", req.TimeWindow.End.Unix())

	// List link
	resp.ListLink = fmt.Sprintf("%s?provider=%s&region=%s&orgName=%s&orgId=%s&projectId=%s&clusterId=%s&clusterName=%s&deployType=%s#/slow_query?show_internal=true&digest=%s&from=%s&row=0&to=%s",
		config.AppConfig.Clinic.BaseUrl,
		cluster.Provider,
		cluster.Region,
		org.OrgName,
		org.OrgID,
		cluster.ProjectID,
		cluster.ClusterID,
		cluster.ClusterName,
		deployType,
		req.Digest,
		startTime,
		endTime)

	// log2.PlainInfo("Link: "+res.Link)

	return resp, nil
}

func getTopSqlLink(ctx context.Context, clusterId string, instance string, window TimeWindow) string {
	// cluster := config.GetClusterByID(clusterId)
	cluster, _ := common.GetCluster(ctx, clusterId)

	// org := config.GetOrgByClusterID(clusterId)

	deployType := cluster.DeployType
	if deployType == "" {
		deployType = "dedicated"
	}
	startTime := fmt.Sprintf("%d", window.Start.Unix())
	endTime := fmt.Sprintf("%d", window.End.Unix())
	// db-tikv-0.db-tikv-peer.tidb10648102463976334389.svc%3A20160
	instanceStr := fmt.Sprintf("%s.db-tikv-peer.tidb%s.svc%%3A20160", instance, clusterId)
	link := fmt.Sprintf("%s?provider=%s&region=%s&orgId=%s&projectId=%s&clusterId=%s&clusterName=%s&deployType=%s#/topsql?from=%s&to=%s&instance=%s",
		config.AppConfig.Clinic.BaseUrl,
		cluster.Provider,
		cluster.Region,
		cluster.OrgID,
		cluster.ProjectID,
		cluster.ClusterID,
		cluster.ClusterName,
		deployType,
		startTime,
		endTime,
		instanceStr)

	return link
}

func callBackTrigger(ctx context.Context) error {
	url := "/anycross/trigger/callback/MGMwOTNjMmNjYmIxZmE4ZmUyN2U4YmY1NDc1MDIyZDE1"
	host := "https://open.feishu.cn"
	method := http.MethodPost
	// data := `{
	// 		}`

	header := map[string]string{
		"Content-Type":  "application/json",
		"Authorization": "Basic c3FsX2luc3BlY3Rpb246eFIzdmV6ZWs2UmM=",
	}
	resp, err := utils.HttpRequest(ctx, *utils.NewHttpConfig(host, url, method, header))
	if err != nil {
		log.Error(ctx, "HttpRequest failed", zap.Error(err))
		return err
	}
	log.Info(ctx, "callBackTrigger success", zap.Any("resp", resp))
	return nil
}