package worker

import (
	"context"
	"dataservices/commons/grpc"
	"dataservices/commons/http"
	"dataservices/commons/logger"
	"dataservices/commons/utils"
	dto "dataservices/internal/dto/dataapi_control"
	"dataservices/internal/entity"
	"dataservices/internal/repository"
	"encoding/json"
	"fmt"
	"net/url"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/avast/retry-go"
	"github.com/tidbcloud/tiinf/errs"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.opentelemetry.io/otel/trace"
	"go.uber.org/zap"
)

const (
	JOB_TYPE_HOURLY_DATAAPP_TRAFFIC  = "HOURLY_dataapp_traffic"
	JOB_TYPE_HOURLY_ENDPOINT_TRAFFIC = "HOURLY_endpoint_traffic"
	JOB_TYPE_HOURLY_ENDPOINT_LATENCY = "HOURLY_endpoint_latency"
	JOB_TYPE_HOURLY_REGION_TRAFFIC   = "HOURLY_region_traffic"
	JOB_TYPE_HOURLY_APIKEY_TRAFFIC   = "HOURLY_apikey_traffic"
	JOB_TYPE_DAILY_KEY_FEATURE_USAGE = "DAILY_key_feature_usage"

	JOB_SUB_TYPE_KEY_FEATURE_CreateDataApp   = "create_data_app"
	JOB_SUB_TYPE_KEY_FEATURE_CreateEndpoint  = "create_endpoint"
	JOB_SUB_TYPE_KEY_FEATURE_PublishEndpoint = "publish_endpoint"
	JOB_SUB_TYPE_KEY_FEATURE_TestEndpoint    = "test_endpoint"
)

const (
	LIMIT_NUM = 2000
)

var errTaskDone = errs.NewConflict("task already has been done", nil)

type DataAppBriefInfo struct {
	Id           uint64
	ExternalType string
	AppType      string
}

type EndpointBriefInfo struct {
	Id          uint64
	DataAppId   uint64
	DataAppInfo *DataAppBriefInfo
	Region      string
}

type etlWorker struct {
	ctx                  context.Context
	jobStatusMutex       sync.Mutex
	jobStatusMap         map[uint64]bool
	appBriefInfoMap      map[uint64]DataAppBriefInfo
	endpointBriefInfoMap map[uint64]EndpointBriefInfo
	endpointRegionMap    map[uint64]string
}

func NewEtlWorker(ctx context.Context) Worker {
	return NewEtlInstance(ctx)
}

func NewEtlInstance(ctx context.Context) *etlWorker {
	var worker = etlWorker{
		ctx:                  ctx,
		jobStatusMap:         make(map[uint64]bool),
		appBriefInfoMap:      make(map[uint64]DataAppBriefInfo),
		endpointBriefInfoMap: make(map[uint64]EndpointBriefInfo),
		endpointRegionMap:    make(map[uint64]string),
	}
	return &worker
}

func (worker *etlWorker) Name() string {
	return "etl_worker"
}

func (worker *etlWorker) writeKeyFeatureUsageDailyToDB(ctx context.Context, secAt int64, appType string, subType string, email string, num int) error {
	record := entity.EtlCombinedData{
		JobType:         JOB_TYPE_DAILY_KEY_FEATURE_USAGE,
		RecordTimestamp: utils.NewJsonTime(time.Unix(secAt, 0)),
		Key1:            utils.GetExternalTypeByEmail(email),
		Key2:            appType,
		Key3:            email,
		Key4:            subType,
		Value:           float64(num),
	}
	return repository.RecordEtlCombinedData(ctx, &record)
}

func (worker *etlWorker) writeUserNumToDB(ctx context.Context, secAt int64, jobType string, externalType string, appType string, subType string, value int64) error {
	record := entity.EtlCombinedData{
		JobType:         jobType,
		RecordTimestamp: utils.NewJsonTime(time.Unix(secAt, 0)),
		Key1:            externalType,
		Key2:            appType,
		Key3:            subType,
		Key4:            "",
		Value:           float64(value),
	}
	return repository.RecordEtlCombinedData(ctx, &record)
}

func (worker *etlWorker) countKeyFeatureUsageByTimeRange(ctx context.Context, featureType string, startSecAt int64, endSecAt int64) (map[string]map[string]int, error) {
	res := make(map[string]map[string]int)
	if featureType == JOB_SUB_TYPE_KEY_FEATURE_CreateDataApp {
		dataAppArray, err := repository.ListDataAppsByTimeRange(ctx, int64(startSecAt), int64(endSecAt))
		if err != nil {
			tilog.Error(ctx, "ListDataAppsByTimeRange Failed.", zap.Any("err", err))
			return nil, err
		}
		for _, item := range dataAppArray {
			briefInfo, tmpErr := worker.GetDataAppBriefInfo(ctx, item.ID)
			if tmpErr != nil {
				tilog.Error(ctx, "GetDataAppBriefInfo Failed.", zap.Any("err", err))
				return nil, err
			}
			if _, ok := res[briefInfo.AppType]; !ok {
				res[briefInfo.AppType] = make(map[string]int)
			}
			if _, ok := res[briefInfo.AppType][item.CreatedBy]; !ok {
				res[briefInfo.AppType][item.CreatedBy] = 0
			}
			res[briefInfo.AppType][item.CreatedBy] += 1
		}
	} else if featureType == JOB_SUB_TYPE_KEY_FEATURE_CreateEndpoint {
		endpointArray, err := repository.ListEndpointsByTimeRange(ctx, int64(startSecAt), int64(endSecAt))
		if err != nil {
			tilog.Error(ctx, "ListEndpointsByTimeRange Failed.", zap.Any("err", err))
			return nil, err
		}
		for _, item := range endpointArray {
			briefInfo, tmpErr := worker.GetDataAppBriefInfo(ctx, item.DataappId)
			if tmpErr != nil {
				tilog.Error(ctx, "GetDataAppBriefInfo Failed.", zap.Any("err", err))
				return nil, err
			}
			appType := briefInfo.AppType
			if _, ok := res[appType]; !ok {
				res[appType] = make(map[string]int)
			}
			if _, ok := res[appType][item.CreatedBy]; !ok {
				res[appType][item.CreatedBy] = 0
			}
			res[appType][item.CreatedBy] += 1
		}
	} else if featureType == JOB_SUB_TYPE_KEY_FEATURE_PublishEndpoint {
		endpointArray, err := repository.ListPublishEndpointsByTimeRange(ctx, int64(startSecAt), int64(endSecAt))
		if err != nil {
			tilog.Error(ctx, "ListPublishEndpointsByTimeRange Failed.", zap.Any("err", err))
			return nil, err
		}
		for _, item := range endpointArray {
			briefInfo, tmpErr := worker.GetDataAppBriefInfo(ctx, item.DataappId)
			if tmpErr != nil {
				tilog.Error(ctx, "GetDataAppBriefInfo Failed.", zap.Any("err", err))
				return nil, err
			}
			appType := briefInfo.AppType
			if _, ok := res[appType]; !ok {
				res[appType] = make(map[string]int)
			}
			if _, ok := res[appType][item.CreatedBy]; !ok {
				res[appType][item.CreatedBy] = 0
			}
			res[appType][item.CreatedBy] += 1
		}
	} else if featureType == JOB_SUB_TYPE_KEY_FEATURE_TestEndpoint {
		endpointArray, err := repository.ListTestEndpointsByTimeRange(ctx, int64(startSecAt), int64(endSecAt))
		if err != nil {
			tilog.Error(ctx, "ListPublishEndpointsByTimeRange Failed.", zap.Any("err", err))
			return nil, err
		}
		for _, item := range endpointArray {
			briefInfo, tmpErr := worker.GetDataAppBriefInfo(ctx, item.DataAppID)
			if tmpErr != nil {
				tilog.Error(ctx, "GetDataAppBriefInfo Failed.", zap.Any("err", err))
				return nil, err
			}
			appType := briefInfo.AppType
			if _, ok := res[appType]; !ok {
				res[appType] = make(map[string]int)
			}
			if _, ok := res[appType][item.Caller]; !ok {
				res[appType][item.Caller] = 0
			}
			res[appType][item.Caller] += 1
		}
	}
	return res, nil
}

func (worker *etlWorker) countKeyFeatureUsage(ctx context.Context, featureType string, startSecAt int64, endSecAt int64, totalActiveUserMap map[string]map[string]bool) error {
	res, err := worker.countKeyFeatureUsageByTimeRange(ctx, featureType, startSecAt, endSecAt)
	if err != nil {
		return err
	}
	for appType, userMap := range res {
		if _, ok := totalActiveUserMap[appType]; !ok {
			totalActiveUserMap[appType] = make(map[string]bool)
		}
		for email, num := range userMap {
			worker.writeKeyFeatureUsageDailyToDB(ctx, startSecAt, appType, featureType, email, num)
			totalActiveUserMap[appType][email] = true
		}
	}
	return nil
}

func (worker *etlWorker) recordKeyFeatureUsageDaily(ctx context.Context, timestamp int64) error {
	startSecAt := timestamp / 86400 * 86400
	endSecAt := startSecAt + 86400

	totalActiveUserMap := make(map[string]map[string]bool)

	err := worker.countKeyFeatureUsage(ctx, JOB_SUB_TYPE_KEY_FEATURE_CreateDataApp, startSecAt, endSecAt, totalActiveUserMap)
	if err != nil {
		tilog.Error(ctx, "CountCreateDataAppUsers Failed.", zap.Any("err", err))
	}
	err = worker.countKeyFeatureUsage(ctx, JOB_SUB_TYPE_KEY_FEATURE_CreateEndpoint, startSecAt, endSecAt, totalActiveUserMap)
	if err != nil {
		tilog.Error(ctx, "CountCreateEndpointUsers Failed.", zap.Any("err", err))
	}
	err = worker.countKeyFeatureUsage(ctx, JOB_SUB_TYPE_KEY_FEATURE_PublishEndpoint, startSecAt, endSecAt, totalActiveUserMap)
	if err != nil {
		tilog.Error(ctx, "CountPublishEndpointUsers Failed.", zap.Any("err", err))
	}
	err = worker.countKeyFeatureUsage(ctx, JOB_SUB_TYPE_KEY_FEATURE_TestEndpoint, startSecAt, endSecAt, totalActiveUserMap)
	if err != nil {
		tilog.Error(ctx, "CountTestEndpointUsers Failed.", zap.Any("err", err))
	}

	for appType, userArray := range totalActiveUserMap {
		for user, _ := range userArray {
			err = repository.RecordEtlActiveUser(ctx, appType, user, time.Unix(startSecAt, 0))
			if err != nil {
				tilog.Error(ctx, "RecordEtlActiveUser Failed.", zap.Any("err", err))
			}
		}
	}
	return nil
}
func (worker *etlWorker) RunEtlJobDaily(ctx context.Context, timestamp int64) error {
	tilog.Info(ctx, "Run Etl Job Daily start. ", zap.Int64("timestamp", timestamp))
	worker.recordKeyFeatureUsageDaily(ctx, timestamp)
	worker.countActiveUserNumDaily(ctx, timestamp)
	return nil
}
func (worker *etlWorker) countActiveUserNumByType(ctx context.Context, jobType string, beforeStartSec int64, beforeEndSec int64,
	curStartSec int64, curEndSec int64) error {
	newUserFilter := entity.EtlActiveUserCountInput{
		CreatedRange:    true,
		UpdatedRange:    false,
		CreatedStartSec: curStartSec,
		CreatedEndSec:   curEndSec,
	}
	newUserArray, err := repository.CountActiveUserNum(ctx, newUserFilter)
	if err != nil {
		tilog.Error(ctx, "countActiveUserNumByType failed", zap.Any("err", err), zap.String("type", jobType))
		return err
	}
	for _, newUserItem := range newUserArray {
		externalType := newUserItem.Key1
		appType := newUserItem.Key2
		newUserNum := newUserItem.Value
		worker.writeUserNumToDB(ctx, curStartSec, "USER_NUM_COUNT__"+jobType, externalType, appType, "NewUser", newUserNum)
	}

	totalUserFilter := entity.EtlActiveUserCountInput{
		CreatedRange:    false,
		UpdatedRange:    true,
		UpdatedStartSec: curStartSec,
		UpdatedEndSec:   curEndSec,
	}
	totalUserArray, err := repository.CountActiveUserNum(ctx, totalUserFilter)
	if err != nil {
		tilog.Error(ctx, "countActiveUserNumByType failed", zap.Any("err", err), zap.String("type", jobType))
		return err
	}

	for _, totalUserItem := range totalUserArray {
		externalType := totalUserItem.Key1
		appType := totalUserItem.Key2
		totalUserNum := totalUserItem.Value
		worker.writeUserNumToDB(ctx, curStartSec, "USER_NUM_COUNT__"+jobType, externalType, appType, "TotalUser", totalUserNum)
	}

	lostUserFilter := entity.EtlActiveUserCountInput{
		CreatedRange:    true,
		UpdatedRange:    true,
		CreatedStartSec: beforeStartSec,
		CreatedEndSec:   beforeEndSec,
		UpdatedStartSec: 0,
		UpdatedEndSec:   curStartSec,
	}
	lostUserArray, err := repository.CountActiveUserNum(ctx, lostUserFilter)
	if err != nil {
		tilog.Error(ctx, "countActiveUserNumByType failed", zap.Any("err", err), zap.String("type", jobType))
		return err
	}

	for _, lostUserItem := range lostUserArray {
		externalType := lostUserItem.Key1
		appType := lostUserItem.Key2
		lostUserNum := lostUserItem.Value
		worker.writeUserNumToDB(ctx, curStartSec, "USER_NUM_COUNT__"+jobType, externalType, appType, "LostUser", lostUserNum)
	}
	return nil
}
func (worker *etlWorker) countActiveUserNumDaily(ctx context.Context, timestamp int64) error {
	dailyCurDate := timestamp
	worker.countActiveUserNumByType(ctx, "Daily", dailyCurDate-86400, dailyCurDate, dailyCurDate, dailyCurDate+86400)
	weeklyCurDate := timestamp - 6*86400
	worker.countActiveUserNumByType(ctx, "Weekly", weeklyCurDate-86400*7, weeklyCurDate, weeklyCurDate, weeklyCurDate+86400*7)

	monthlyCurDate := timestamp - 29*86400
	worker.countActiveUserNumByType(ctx, "Monthly", monthlyCurDate-86400*30, monthlyCurDate, monthlyCurDate, monthlyCurDate+86400*30)
	return nil
}
func (worker *etlWorker) RunEtlJobHourly(ctx context.Context, startTimestamp int64, refresh bool) (err error) {
	start := time.Now()
	defer func() {
		logger.LogWithError(ctx, err, "RunEtlJobHourly completed", zap.Int64("start", startTimestamp), zap.Bool("refresh", refresh), zap.Duration("cost", time.Since(start)))
	}()
	startSecAt := startTimestamp
	endSecAt := startTimestamp + 3600

	// check if the task has been completed, except for refreshing
	if !refresh {
		if count, err := repository.CountEtlCombinedData(ctx, JOB_TYPE_HOURLY_APIKEY_TRAFFIC, time.Unix(startSecAt, 0)); err != nil {
			logger.Error(ctx, "failed to count apikey traffic etl combined data", zap.Int64("startSecAt", startSecAt), zap.Error(err))
			return err
		} else if count > 0 {
			logger.Info(ctx, "RunEtlJobHourly has already been done", zap.Int64("startSecAt", startSecAt))
			return errTaskDone
		}
	}

	statisticsResult, err := worker.StatisticsEndpointLog(ctx, startSecAt, endSecAt)
	if err != nil {
		logger.LogWithError(ctx, err, "StatisticsEndpointLog failed", zap.Int64("startSecAt", startSecAt), zap.Error(err))
		return err
	}

	if _, err := worker.analyzeDataappRegion(ctx, startSecAt, statisticsResult.EndpointMap); err != nil {
		tilog.Error(ctx, "Run EtlJobHourly failed - analyzeDataappRegion", zap.Int64("startSecAt", startSecAt), zap.Any("err", err))
		return err
	}

	worker.countDataappTrafficHourly(ctx, startSecAt, statisticsResult.DataappTraffic)
	worker.countEndpointTrafficHourly(ctx, startSecAt, statisticsResult.EndpointTraffic)
	worker.countEndpointLatencyHourly(ctx, startSecAt, statisticsResult.EndpointLatency)
	worker.countRegionTrafficHourly(ctx, startSecAt, statisticsResult.EndpointTraffic)
	worker.countApiKeyTrafficHourly(ctx, startSecAt, statisticsResult.ApikeyTraffic)
	return nil
}

func (worker *etlWorker) Handle(tracer trace.Tracer) {
	ctx, cancel := context.WithTimeout(worker.ctx, time.Hour)
	defer cancel()
	newctx, span := tracer.Start(ctx, "etl-worker")
	defer span.End()
	// Running the job every hour
	formatTimeStamp := uint64(time.Now().Unix()/3600*3600) - 3600
	// Check whether the running job exists in current id
	if worker.jobRunning(formatTimeStamp) {
		tilog.Info(ctx, "task already running, skip once")
		return
	}
	defer worker.jobDone(formatTimeStamp)
	if err := worker.RunEtlJobHourly(newctx, int64(formatTimeStamp), false); err == nil {
		if formatTimeStamp%86400 == 0 {
			worker.RunEtlJobDaily(newctx, int64(formatTimeStamp-86400))
		}
	}
}

func (worker *etlWorker) jobRunning(id uint64) bool {
	worker.jobStatusMutex.Lock()
	defer worker.jobStatusMutex.Unlock()
	if worker.jobStatusMap[id] {
		return true
	}
	worker.jobStatusMap[id] = true
	return false
}

func (worker *etlWorker) jobDone(id uint64) {
	worker.jobStatusMutex.Lock()
	defer worker.jobStatusMutex.Unlock()
	delete(worker.jobStatusMap, id)
}

func (worker *etlWorker) RunningMsInterval() time.Duration {
	return time.Minute * 10
}

func (worker *etlWorker) IsEnable() bool {
	return true
}

func (worker *etlWorker) GetEndpointBriefInfo(ctx context.Context, endpointId uint64) (*EndpointBriefInfo, error) {
	if info, ok := worker.endpointBriefInfoMap[endpointId]; ok {
		return &info, nil
	}
	endpointInfo, err := repository.GetEndpointRaw(ctx, endpointId)
	if err != nil {
		return nil, err
	}
	tilog.Info(ctx, "endpointInfo", zap.Any("info", endpointInfo))

	dataAppBriefInfo, err := worker.GetDataAppBriefInfo(ctx, endpointInfo.DataappId)
	if err != nil {
		return nil, err
	}

	region, ok := worker.endpointRegionMap[endpointId]
	if !ok {
		region = ""
	}
	briefInfo := EndpointBriefInfo{
		Id:          endpointInfo.ID,
		DataAppId:   endpointInfo.DataappId,
		DataAppInfo: dataAppBriefInfo,
		Region:      region,
	}
	worker.endpointBriefInfoMap[endpointId] = briefInfo
	return &briefInfo, nil
}

func (worker *etlWorker) GetDataAppBriefInfo(ctx context.Context, appId uint64) (*DataAppBriefInfo, error) {
	if info, ok := worker.appBriefInfoMap[appId]; ok {
		return &info, nil
	}
	dataAppInfo, err := repository.GetDataAppRaw(ctx, appId)
	if err != nil {
		return nil, err
	}
	externalType := utils.GetExternalTypeByEmail(dataAppInfo.CreatedBy)
	briefInfo := DataAppBriefInfo{Id: dataAppInfo.ID, AppType: string(dataAppInfo.AppType), ExternalType: externalType}
	worker.appBriefInfoMap[appId] = briefInfo
	return &briefInfo, nil
}
func (worker *etlWorker) analyzeDataappRegion(ctx context.Context, startSecAt int64, endpointMap map[string]*EndpointRegion) ([]*entity.EtlEndpointRegion, error) {
	var records []*entity.EtlEndpointRegion
	if endpointMap == nil {
		return records, nil
	}
	for _, item := range endpointMap {
		record := entity.EtlEndpointRegion{
			EndpointID: item.EndpointID,
			DataappID:  item.DataappID,
			Region:     item.Region,
			CreatedAt:  utils.NewJsonTime(time.Unix(startSecAt, 0)),
			UpdatedAt:  utils.NewJsonTime(time.Unix(startSecAt, 0)),
		}
		tilog.Info(ctx, "item", zap.Any("result", record))
		err := repository.RecordEtlEndpointRegion(ctx, &record)
		if err != nil {
			return nil, err
		}
		records = append(records, &record)
		worker.endpointRegionMap[item.EndpointID] = item.Region
	}
	return records, nil
}
func (worker *etlWorker) countDataappTrafficHourly(ctx context.Context, startSecAt int64, traffics []entity.AnalysisIntKVResult) error {
	if len(traffics) == 0 {
		return nil
	}
	tilog.Info(ctx, "result", zap.Any("result", traffics))
	repository.DeleteEtlCombinedData(ctx, JOB_TYPE_HOURLY_DATAAPP_TRAFFIC, time.Unix(startSecAt, 0))
	for _, item := range traffics {
		briefInfo, err := worker.GetDataAppBriefInfo(ctx, uint64(item.Key))
		if err != nil {
			tilog.Error(ctx, "GetDataAppBriefInfo Failed.", zap.Any("err", err))
			continue
		}
		tilog.Info(ctx, "brief info", zap.Any("info", briefInfo))
		record := entity.EtlCombinedData{
			JobType:         JOB_TYPE_HOURLY_DATAAPP_TRAFFIC,
			RecordTimestamp: utils.NewJsonTime(time.Unix(startSecAt, 0)),
			Key1:            briefInfo.ExternalType,
			Key2:            briefInfo.AppType,
			Key3:            strconv.FormatInt(item.Key, 10),
			Key4:            "",
			Value:           float64(item.Value),
		}
		err = repository.RecordEtlCombinedData(ctx, &record)
		if err != nil {
			return err
		}
	}
	return nil
}
func (worker *etlWorker) countEndpointLatencyHourly(ctx context.Context, startSecAt int64, traffics []entity.AnalysisIntKVResult) error {
	if len(traffics) == 0 {
		return nil
	}
	tilog.Info(ctx, "countEndpointLatencyHourly result", zap.Any("result", traffics))

	repository.DeleteEtlCombinedData(ctx, JOB_TYPE_HOURLY_ENDPOINT_LATENCY, time.Unix(startSecAt, 0))

	for _, item := range traffics {
		briefInfo, err := worker.GetEndpointBriefInfo(ctx, uint64(item.Key))
		if err != nil {
			tilog.Error(ctx, "GetDataAppBriefInfo Failed.", zap.Any("err", err))
			continue
		}
		tilog.Info(ctx, "brief info", zap.Any("info", briefInfo))
		record := entity.EtlCombinedData{
			JobType:         JOB_TYPE_HOURLY_ENDPOINT_LATENCY,
			RecordTimestamp: utils.NewJsonTime(time.Unix(startSecAt, 0)),
			Key1:            briefInfo.DataAppInfo.ExternalType,
			Key2:            briefInfo.DataAppInfo.AppType,
			Key3:            strconv.FormatInt(item.Key, 10),
			Key4:            "",
			Value:           float64(item.Value),
		}
		err = repository.RecordEtlCombinedData(ctx, &record)
		if err != nil {
			return err
		}
	}
	return nil
}
func (worker *etlWorker) countEndpointTrafficHourly(ctx context.Context, startSecAt int64, traffics []entity.AnalysisIntKVResult) error {
	if len(traffics) == 0 {
		return nil
	}
	tilog.Info(ctx, "result", zap.Any("result", traffics))

	repository.DeleteEtlCombinedData(ctx, JOB_TYPE_HOURLY_ENDPOINT_TRAFFIC, time.Unix(startSecAt, 0))

	for _, item := range traffics {
		briefInfo, err := worker.GetEndpointBriefInfo(ctx, uint64(item.Key))
		if err != nil {
			tilog.Error(ctx, "GetDataAppBriefInfo Failed.", zap.Any("err", err))
			continue
		}
		tilog.Info(ctx, "brief info", zap.Any("info", briefInfo))
		record := entity.EtlCombinedData{
			JobType:         JOB_TYPE_HOURLY_ENDPOINT_TRAFFIC,
			RecordTimestamp: utils.NewJsonTime(time.Unix(startSecAt, 0)),
			Key1:            briefInfo.DataAppInfo.ExternalType,
			Key2:            briefInfo.DataAppInfo.AppType,
			Key3:            strconv.FormatInt(item.Key, 10),
			Key4:            "",
			Value:           float64(item.Value),
		}
		err = repository.RecordEtlCombinedData(ctx, &record)
		if err != nil {
			return err
		}
	}
	return nil
}

func (worker *etlWorker) countApiKeyTrafficHourly(ctx context.Context, startSecAt int64, traffics []entity.AnalysisStrIntKKVResult) error {
	if len(traffics) == 0 {
		return nil
	}
	tilog.Info(ctx, "result", zap.Any("result", traffics))

	repository.DeleteEtlCombinedData(ctx, JOB_TYPE_HOURLY_APIKEY_TRAFFIC, time.Unix(startSecAt, 0))

	for _, item := range traffics {
		record := entity.EtlCombinedData{
			JobType:         JOB_TYPE_HOURLY_APIKEY_TRAFFIC,
			RecordTimestamp: utils.NewJsonTime(time.Unix(startSecAt, 0)),
			Key1:            strconv.FormatInt(item.Key2, 10),
			Key2:            item.Key1,
			Key3:            "",
			Key4:            "",
			Value:           float64(item.Value),
		}
		if err := repository.RecordEtlCombinedData(ctx, &record); err != nil {
			return err
		}
	}
	return nil
}
func (worker *etlWorker) countRegionTrafficHourly(ctx context.Context, startSecAt int64, traffics []entity.AnalysisIntKVResult) error {
	if len(traffics) == 0 {
		return nil
	}
	tilog.Info(ctx, "result", zap.Any("result", traffics))
	regionCountMap := make(map[string]map[string]map[string]uint64)

	for _, item := range traffics {
		briefInfo, err := worker.GetEndpointBriefInfo(ctx, uint64(item.Key))
		if err != nil {
			tilog.Error(ctx, "GetDataAppBriefInfo Failed.", zap.Any("err", err))
			continue
		}
		externalType := briefInfo.DataAppInfo.ExternalType
		appType := briefInfo.DataAppInfo.AppType
		region := briefInfo.Region
		if _, ok := regionCountMap[externalType]; !ok {
			regionCountMap[externalType] = make(map[string]map[string]uint64)
		}
		if _, ok := regionCountMap[externalType][appType]; !ok {
			regionCountMap[externalType][appType] = make(map[string]uint64)
		}
		if _, ok := regionCountMap[externalType][appType][region]; !ok {
			regionCountMap[externalType][appType][region] = 0
		}
		regionCountMap[externalType][appType][region] += uint64(item.Value)
	}
	repository.DeleteEtlCombinedData(ctx, JOB_TYPE_HOURLY_REGION_TRAFFIC, time.Unix(startSecAt, 0))
	for externalType, map1 := range regionCountMap {
		for appType, map2 := range map1 {
			for region, val := range map2 {
				record := entity.EtlCombinedData{
					JobType:         JOB_TYPE_HOURLY_REGION_TRAFFIC,
					RecordTimestamp: utils.NewJsonTime(time.Unix(startSecAt, 0)),
					Key1:            externalType,
					Key2:            appType,
					Key3:            region,
					Key4:            "",
					Value:           float64(val),
				}
				if err := repository.RecordEtlCombinedData(ctx, &record); err != nil {
					return err
				}
			}
		}
	}

	return nil
}

type EndpointRegion struct {
	DataappID  uint64
	EndpointID uint64
	Region     string
}
type ApikeyTraffic struct {
	DataappID uint64
	Apikey    string
	Traffic   int
}
type StatisticsRawResult struct {
	EndpointMap     map[string]*EndpointRegion
	DataappTraffic  map[uint64]int
	EndpointTraffic map[uint64]int
	EndpointLatency map[uint64]int64
	ApikeyTraffic   map[string]*ApikeyTraffic
}

type StatisticsResult struct {
	EndpointMap     map[string]*EndpointRegion
	DataappTraffic  []entity.AnalysisIntKVResult
	EndpointTraffic []entity.AnalysisIntKVResult
	EndpointLatency []entity.AnalysisIntKVResult
	ApikeyTraffic   []entity.AnalysisStrIntKKVResult
}

func (worker *etlWorker) StatisticsEndpointLog(ctx context.Context, startSecAt, endSecAt int64) (statisticsRes *StatisticsResult, err error) {
	start := time.Now()
	defer func() {
		if err != nil {
			logger.LogWithError(ctx, err, "statistics endpoint log failed", zap.Int64("start", startSecAt))
		} else {
			logger.Info(ctx, "statistics endpoint log completed", zap.Int64("start", startSecAt),
				zap.Int64("end", endSecAt), zap.Duration("cost", time.Since(start)),
				zap.Int("EndpointMapSize", len(statisticsRes.EndpointMap)),
				zap.Int("DataappTrafficSize", len(statisticsRes.DataappTraffic)),
				zap.Int("EndpointLatencySize", len(statisticsRes.EndpointLatency)),
				zap.Int("EndpointTrafficSize", len(statisticsRes.EndpointTraffic)),
				zap.Int("ApikeyTrafficSize", len(statisticsRes.ApikeyTraffic)),
			)
		}
	}()

	it := &EndpointAllLogIterator{
		ctx:       ctx,
		startTime: uint64(startSecAt),
		endTime:   uint64(endSecAt),
	}
	result := &StatisticsRawResult{
		EndpointMap:     make(map[string]*EndpointRegion),
		DataappTraffic:  make(map[uint64]int),
		EndpointTraffic: make(map[uint64]int),
		EndpointLatency: make(map[uint64]int64),
		ApikeyTraffic:   make(map[string]*ApikeyTraffic),
	}
	for it.Next() {
		rawLog, err := it.Item()
		if err != nil {
			logger.Warn(ctx, "failed to get item", zap.Error(err))
			continue
		}
		endpointID, err := strconv.ParseUint(rawLog.EndpointId, 10, 64)
		if err != nil {
			logger.Error(ctx, "failed to parse endpoint id",
				zap.String("endpoint_id", rawLog.EndpointId),
				zap.String("trace_id", rawLog.TraceID), zap.Error(err))
			continue
		}
		dataappID, _ := strconv.ParseUint(rawLog.DataappID, 10, 64)
		if endpointID == 0 || dataappID == 0 {
			continue
		}
		var headers map[string]any
		if err := json.Unmarshal([]byte(rawLog.ReqHeader), &headers); err != nil {
			logger.Error(ctx, "failed to unmarshal endpoint request header",
				zap.String("endpoint_id", rawLog.EndpointId),
				zap.String("header", rawLog.ReqHeader),
				zap.String("trace_id", rawLog.TraceID), zap.Error(err))
			continue
		}
		authContentStr, _ := headers["X-Auth-Content"].(string)
		var authContent map[string]any
		if err := json.Unmarshal([]byte(authContentStr), &authContent); err != nil {
			logger.Error(ctx, "failed to unmarshal endpoint request auth content",
				zap.String("endpoint_id", rawLog.EndpointId),
				zap.String("header", rawLog.ReqHeader),
				zap.String("trace_id", rawLog.TraceID), zap.Error(err))
			continue
		}
		apikey, _ := authContent["public_key"].(string)
		urlInfo, err := url.Parse(rawLog.ReqUrl)
		if err != nil {
			logger.Error(ctx, "failed to parse endpoint request url",
				zap.String("endpoint_id", rawLog.EndpointId),
				zap.String("url", rawLog.ReqUrl),
				zap.String("trace_id", rawLog.TraceID), zap.Error(err))
			continue
		}
		hostname := urlInfo.Hostname()
		pos := strings.Index(hostname, ".")
		region := hostname[:pos]
		latency, err := time.ParseDuration(rawLog.Latency)
		if err != nil {
			logger.Error(ctx, "failed to parse endpoint request latency",
				zap.String("endpoint_id", rawLog.EndpointId),
				zap.String("latency", rawLog.Latency),
				zap.String("trace_id", rawLog.TraceID), zap.Error(err))
			continue
		}

		endpointKey := fmt.Sprintf("%d:%d:%s", dataappID, endpointID, region)
		if _, ok := result.EndpointMap[endpointKey]; !ok {
			result.EndpointMap[endpointKey] = &EndpointRegion{
				DataappID:  dataappID,
				EndpointID: endpointID,
				Region:     region,
			}
		}
		result.DataappTraffic[dataappID]++
		result.EndpointTraffic[endpointID]++
		result.EndpointLatency[endpointID] += latency.Milliseconds()
		apikeyTrafficKey := fmt.Sprintf("%d_%s", dataappID, apikey)
		if _, ok := result.ApikeyTraffic[apikeyTrafficKey]; !ok {
			result.ApikeyTraffic[apikeyTrafficKey] = &ApikeyTraffic{
				DataappID: dataappID,
				Apikey:    apikey,
			}
		}
		result.ApikeyTraffic[apikeyTrafficKey].Traffic++
	}
	if err := it.Error(); err != nil {
		return nil, err
	}
	for endpointID := range result.EndpointLatency {
		result.EndpointLatency[endpointID] /= int64(result.EndpointTraffic[endpointID])
	}

	statisticsRes = new(StatisticsResult)
	statisticsRes.EndpointMap = result.EndpointMap
	for dataappID, traffic := range result.DataappTraffic {
		statisticsRes.DataappTraffic = append(statisticsRes.DataappTraffic, entity.AnalysisIntKVResult{
			Key:   int64(dataappID),
			Value: int64(traffic),
		})
	}
	for endpointID, traffic := range result.EndpointTraffic {
		statisticsRes.EndpointTraffic = append(statisticsRes.EndpointTraffic, entity.AnalysisIntKVResult{
			Key:   int64(endpointID),
			Value: int64(traffic),
		})
	}
	for endpointID, latency := range result.EndpointLatency {
		statisticsRes.EndpointLatency = append(statisticsRes.EndpointLatency, entity.AnalysisIntKVResult{
			Key:   int64(endpointID),
			Value: int64(latency),
		})
	}
	for _, traffic := range result.ApikeyTraffic {
		statisticsRes.ApikeyTraffic = append(statisticsRes.ApikeyTraffic, entity.AnalysisStrIntKKVResult{
			Key1:  traffic.Apikey,
			Key2:  int64(traffic.DataappID),
			Value: int64(traffic.Traffic),
		})
	}
	return statisticsRes, nil
}

type EndpointAllLogIterator struct {
	ctx             context.Context
	orgs            []uint64
	orgRegions      [][]string
	currentIndex    [2]int
	currentIterator *EndpointRegionLogIterator
	startTime       uint64
	endTime         uint64
	initOnce        sync.Once
}

func (it *EndpointAllLogIterator) Next() bool {
	it.initOnce.Do(func() {
		it.initRegions()
	})
	if len(it.orgs) == 0 {
		return false
	}
	if it.currentIterator != nil && it.currentIterator.Next() {
		return true
	}
	orgIdx, regionIdx := it.currentIndex[0], it.currentIndex[1]
	for ; orgIdx < len(it.orgs); orgIdx++ {
		if regionIdx >= len(it.orgRegions[orgIdx]) {
			regionIdx = 0
			continue
		}
		orgID, region := it.orgs[orgIdx], it.orgRegions[orgIdx][regionIdx]
		it.currentIterator = &EndpointRegionLogIterator{
			ctx:    it.ctx,
			region: region,
			currentLogReq: &dto.QueryUserExecutionLogReq{
				Page:     1,
				PageSize: 50,
				QueryUserExecutionLogParam: dto.QueryUserExecutionLogParam{
					Org:    strconv.FormatUint(orgID, 10),
					Start:  it.startTime,
					End:    it.endTime,
					Source: "elt-worker",
				},
			},
		}
		regionIdx++
		break
	}
	it.currentIndex[0], it.currentIndex[1] = orgIdx, regionIdx
	return it.currentIterator.Next()
}

func (it *EndpointAllLogIterator) Error() error {
	if it.currentIterator != nil {
		return it.currentIterator.err
	}
	return nil
}

func (it *EndpointAllLogIterator) initRegions() {

	// load all dataapp's org and cluster with draft endpoint
	page, pageSize := 0, 20
	orgClusterMap := make(map[uint64][]uint64)
	for {
		page++
		result, err := repository.ListDraftEndpointClusters(it.ctx, page, pageSize)
		if err != nil {
			logger.LogWithError(it.ctx, err, "failed to list draft endpoint cluster", zap.Int("page", page), zap.Error(err))
			continue
		}
		for _, item := range result {
			orgID := item.OrgID
			clusterID := item.ClusterID
			if orgID == 0 || clusterID == 0 {
				continue
			}
			orgClusterMap[uint64(orgID)] = append(orgClusterMap[uint64(orgID)], uint64(clusterID))
		}
		if len(result) < pageSize {
			break
		}
	}

	// load all dataapp's org and cluster with deployed endpoint
	page = 0
	for {
		page++
		result, err := repository.ListDeployedEndpointClusters(it.ctx, page, pageSize)
		if err != nil {
			logger.LogWithError(it.ctx, err, "failed to list deployed endpoint cluster", zap.Int("page", page), zap.Error(err))
			continue
		}
		for _, item := range result {
			orgID := item.OrgID
			clusterID := item.ClusterID
			orgClusterMap[uint64(orgID)] = append(orgClusterMap[uint64(orgID)], uint64(clusterID))
		}
		if len(result) < pageSize {
			break
		}
	}

	logger.Info(it.ctx, "load endpoint org and cluster from db completed", zap.Int("org_count", len(orgClusterMap)))

	// statistics org's regions
	orgRegions := make(map[uint64]map[string]struct{})
	for orgID, clusterIDs := range orgClusterMap {
		clusterInfos, err := grpc.ListClusterByIdsWPC(it.ctx, clusterIDs)
		if err != nil {
			logger.LogWithError(it.ctx, err, "failed to list cluster by ids", zap.Uint64("org_id", orgID), zap.Uint64s("cluster_ids", clusterIDs), zap.Error(err))
			continue
		}
		for _, clusterInfo := range clusterInfos {
			region := clusterInfo.RegionName
			if _, ok := orgRegions[orgID]; !ok {
				orgRegions[orgID] = make(map[string]struct{})
			}
			orgRegions[orgID][region] = struct{}{}
		}
	}
	for orgID := range orgRegions {
		regions := make([]string, 0)
		for region := range orgRegions[orgID] {
			regions = append(regions, region)
		}
		if len(regions) == 0 {
			continue
		}
		it.orgs = append(it.orgs, orgID)
		it.orgRegions = append(it.orgRegions, regions)
	}
	logger.Info(it.ctx, "load org and region completed", zap.Int("org_count", len(it.orgs)))
}

func (it *EndpointAllLogIterator) Item() (*dto.CustomExecLogRaw, error) {
	if len(it.orgs) == 0 {
		return nil, fmt.Errorf("no more endpoint log")
	}

	if it.currentIterator == nil {
		return nil, fmt.Errorf("no more endpoint log")
	}

	return it.currentIterator.Item()
}

type EndpointRegionLogIterator struct {
	ctx           context.Context
	region        string
	currentLogReq *dto.QueryUserExecutionLogReq
	currentPage   *dto.QueryUserExecutionLogRsp
	currentIndex  int
	err           error
}

func (it *EndpointRegionLogIterator) Next() bool {
	if it.err != nil {
		return false
	}
	if it.currentPage == nil || it.currentIndex >= len(it.currentPage.Data) {
		if it.currentPage != nil && !it.currentPage.HasMore {
			return false
		}
		err := retry.Do(func() error {
			it.currentPage, it.err = http.QueryDataAppLog(it.ctx, it.region, 0, it.currentLogReq)
			return it.err
		}, retry.Attempts(3),
			retry.Delay(100*time.Millisecond),
			retry.MaxDelay(3*time.Second),
			retry.LastErrorOnly(true),
			retry.OnRetry(func(n uint, err error) {
				logger.LogWithError(it.ctx, err, "failed to query dataapp log", zap.Uint("times", n), zap.String("region", it.region), zap.String("org_id", it.currentLogReq.Org), zap.Uint64("page_size", it.currentLogReq.Page), zap.Error(err))
			}),
		)
		if err != nil {
			logger.LogWithError(it.ctx, it.err, "failed to query dataapp log", zap.String("region", it.region), zap.String("org_id", it.currentLogReq.Org), zap.Uint64("page_size", it.currentLogReq.Page), zap.Error(err))
			return false
		}
		it.currentLogReq.Passback = it.currentPage.Passback
		it.currentLogReq.Page++
		it.currentIndex = 0
		return true
	}
	it.currentIndex++
	return true
}

func (it *EndpointRegionLogIterator) Item() (*dto.CustomExecLogRaw, error) {
	if it.err != nil {
		return nil, it.err
	}

	if it.currentPage == nil || it.currentIndex >= len(it.currentPage.Data) {
		return nil, fmt.Errorf("no more endpoint log, region:%s, org_id:%s", it.region, it.currentLogReq.Org)
	}

	return &it.currentPage.Data[it.currentIndex], nil
}