package resource_limit

import (
	"fmt"
	"strconv"
	"time"

	"github.com/pingcap/log"
	"github.com/robfig/cron/v3"
	"github.com/tidbcloud/account/idl/pbgen/proto/account"
	openapiv1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/common/v1beta1"
	mgmtv1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"

	"github.com/pingcap-inc/clinic/pkg/auth/session"
	"github.com/pingcap-inc/clinic/pkg/config"
	"github.com/pingcap-inc/clinic/pkg/datasource"
	model2 "github.com/pingcap-inc/clinic/pkg/datasource/model"
	"github.com/pingcap-inc/clinic/pkg/datasource/utils"
	"github.com/pingcap-inc/clinic/pkg/db"
	"github.com/pingcap-inc/clinic/pkg/db/model"
	"github.com/pingcap-inc/clinic/pkg/es"
	"github.com/pingcap-inc/clinic/pkg/o11y"
	"github.com/pingcap-inc/clinic/pkg/qps_report"
	"github.com/pingcap-inc/clinic/pkg/stream"
	"github.com/pingcap-inc/clinic/pkg/tsdb"
)

var expired = 3

type DeleteMetricDataJob struct {
	Name       string
	mc         config.MetricConfig
	ExpiredDay int
}

// http://<vmselect>:8481/delete/<accountID>/prometheus/api/v1/admin/tsdb/delete_series?match[]=<timeseries_selector_for_delete>
func (j *DeleteMetricDataJob) Run() {
	log.Info("corn job running", zap.String("name", "DeleteMetricDataJob"))
	defer func() {
		if err := recover(); err != nil {
			log.Error("DeleteMetricDataJob failed", zap.String("err", fmt.Sprintf("%v", err)))
		}
	}()

	// Generally files has smaller data than clusters
	files, err := model.GetExpiredFiles(j.ExpiredDay, "rebuild_metric_status")
	if err != nil {
		log.Error("get expired cluster package failed", zap.Error(err))
		return
	}

	if len(files) == 0 {
		return
	}
	log.Warn("found file with expired metrics", zap.Int("count", len(files)))

	// todo@toto fill the func
	clusters, err := model.GetExpiredClusters(j.ExpiredDay)
	if err != nil {
		log.Error("get expired cluster package failed", zap.Error(err))
		return
	}

	clusterMap := make(map[string]struct{})
	stream.OfSlice(clusters).Each(func(c db.Cluster) {
		clusterMap[c.ClusterID] = struct{}{}
	})
	var needFroceMerge bool
	for _, file := range files {
		if _, ok := clusterMap[file.ClusterID]; !ok {
			continue
		}

		log.Info("delete expired cluster from vm", zap.String("cluster", file.ClusterID))

		if err := tsdb.DeleteByCluster(file.ClusterID, file.OrgID, *file.BeginTime, *file.EndTime, j.mc); err != nil {
			log.Error("delete cluster failed, ", zap.Error(err))
		} else {
			needFroceMerge = true
		}
		err := model.SetRebuildMetricStatusByClusterID(file.ClusterID, db.StatusDeleted)
		if err != nil {
			log.Error("set status failed", zap.Error(err))
		}
	}
	// after calling deletion api, call force merge api on all vmstorage pods.
	// we use empty `partition_prefix` to let vmstorage apply the merge on all partitions.
	if needFroceMerge {
		if err = tsdb.ForceMerge(""); err != nil {
			log.Error("force merge vmstorage failed", zap.Error(err))
		}
	}
}

type DeleteLogDataJob struct {
	Name       string
	lc         config.LogConfig
	ExpiredDay int
}

func (j *DeleteLogDataJob) Run() {
	log.Info("corn job running", zap.String("name", "DeleteLogDataJob"))
	defer func() {
		if err := recover(); err != nil {
			log.Error("DeleteLogDataJob failed", zap.String("err", fmt.Sprintf("%v", err)))
		}
	}()
	// todo@toto fill the func
	files, err := model.GetExpiredFiles(j.ExpiredDay, "rebuild_log_status")
	if err != nil {
		log.Error("get expired cluster package failed", zap.Error(err))
		return
	}
	for _, file := range files {
		logIndex := fmt.Sprintf("org%d_log", file.OrgID)
		err := j.clean(logIndex, file.FileUUID)
		if err != nil {
			log.Error("delete es log failed", zap.String("index", logIndex), zap.Error(err))
			continue
		}
		slowlogIndex := fmt.Sprintf("org%d_slowlog", file.OrgID)
		err = j.clean(slowlogIndex, file.FileUUID)
		if err != nil {
			log.Error("delete es log failed", zap.String("index", logIndex), zap.Error(err))
			continue
		}

		err = model.SetRebuildLogStatus(file.FileUUID, db.StatusDeleted)
		if err != nil {
			log.Error("set status failed", zap.Error(err))
		}
	}
}

func (j *DeleteLogDataJob) clean(index, fileUUID string) error {
	err := es.DeleteByFileUUID(index, fileUUID, j.lc)
	if err != nil {
		return err
	}
	// todo@toto check if need delete index
	count, found, err := es.QueryDataNumByIndex(index, j.lc)
	if err != nil {
		return err
	}
	// if count is not zero or index not found, do not delete the index
	if count > 0 || !found {
		return nil
	}
	if err := es.DeleteIndex(index, j.lc); err != nil {
		log.Error("delete index failed", zap.Error(err))
		return err
	}

	return nil
}

type MigrateOrganizationJob struct {
	UpdatePlan bool
}

type organizationSyncDependencies struct {
	listOrganizations  func(*session.CtxUser) ([]*account.Org, int64, error)
	getOrganization    func(*session.CtxUser, uint64) (*account.Org, error)
	loadUpdatedAt      func([]uint) (map[uint]time.Time, error)
	upsertOrganization func(*db.Organization, bool) error
}

func defaultOrganizationSyncDependencies() organizationSyncDependencies {
	return organizationSyncDependencies{
		listOrganizations:  datasource.RealTimeListOrganizations,
		getOrganization:    datasource.RealtimeGetOrganization,
		loadUpdatedAt:      model.ListCloudOrganizationUpdatedAt,
		upsertOrganization: model.InsertOrUpdateCloudOrg,
	}
}

func (j *MigrateOrganizationJob) Run() {
	defer func() {
		if err := recover(); err != nil {
			log.Error("MigrateOrganizationJob failed", zap.String("err", fmt.Sprintf("%v", err)))
		}
	}()
	j.run(defaultOrganizationSyncDependencies())
}

func (j *MigrateOrganizationJob) run(deps organizationSyncDependencies) {
	startedAt := time.Now()
	var fetched, skipped, processed, failed int
	defer func() {
		log.Info("MigrateOrganizationJob finished",
			zap.Int("fetched", fetched),
			zap.Int("skipped", skipped),
			zap.Int("processed", processed),
			zap.Int("failed", failed),
			zap.Duration("duration", time.Since(startedAt)))
	}()

	adminUserCtx := &session.CtxUser{
		Internal: true,
		Email:    "clinic@pingcap.com",
	}

	cloudOrgs, _, err := deps.listOrganizations(adminUserCtx)
	if err != nil {
		failed++
		log.Error("Get organizations from account server failed", zap.Error(err))
		return
	}
	fetched = len(cloudOrgs)

	updatedAtByOrgID := make(map[uint]time.Time)
	if !j.UpdatePlan {
		dbaasOrgIDs := make([]uint, 0, len(cloudOrgs))
		for _, org := range cloudOrgs {
			dbaasOrgIDs = append(dbaasOrgIDs, uint(org.Id))
		}

		updatedAtByOrgID, err = deps.loadUpdatedAt(dbaasOrgIDs)
		if err != nil {
			failed++
			log.Error("Load existing cloud organizations failed", zap.Error(err))
			return
		}
	}

	for _, org := range cloudOrgs {
		// RealTimeListOrganizations do not return all need data like plan
		if j.UpdatePlan {
			org, err = deps.getOrganization(adminUserCtx, org.Id)
			if err != nil {
				failed++
				log.Error("get org info with plan failed", zap.Any("organization", org), zap.Error(err))
				return
			}
		}

		organization := utils.CloudOrgToModel(org)
		if updatedAt, exists := updatedAtByOrgID[organization.DbaasOrgID]; !j.UpdatePlan && exists && updatedAt.Equal(organization.UpdatedAt) {
			skipped++
			continue
		}

		processed++
		err = deps.upsertOrganization(organization, j.UpdatePlan)
		if err != nil {
			failed++
			log.Error("Migrate cloud organization failed", zap.Any("organization", org), zap.Error(err))
		}
	}
}

type MigrateMgmtClusterJob struct {
	ForceUpdate bool
}

func (j *MigrateMgmtClusterJob) Run() {
	log.Info("corn job running", zap.String("name", "MigrateMgmtClusterJob"))
	defer func() {
		if err := recover(); err != nil {
			log.Error("MigrateMgmtClusterJob failed", zap.String("err", fmt.Sprintf("%v", err)))
		}
	}()

	adminUserCtx := &session.CtxUser{
		Internal: true,
		Email:    "clinic@pingcap.com",
	}
	cloudOrgs, _, err := datasource.RealTimeListOrganizations(adminUserCtx)
	if err != nil {
		log.Error("Get organization from account server failed", zap.Error(err))
	}
	log.Info("RealTimeListOrganizations", zap.Int("count", len(cloudOrgs)))

	for _, org := range cloudOrgs {
		clinicOrg, err := datasource.GetOrganizationByCloudID(adminUserCtx, uint(org.Id))
		if err != nil {
			log.Error("Get organization from clinic db failed", zap.Error(err), zap.Uint64("tenantID", org.Id))
			continue
		}
		mgmtClusters, err := datasource.RealtimeListMgmtClustersByOrg(adminUserCtx, org.Id)
		if err != nil {
			log.Error("Get mgmt clusters failed", zap.Error(err), zap.Uint64("tenantID", org.Id))
			continue
		}
		if len(mgmtClusters) == 0 {
			continue
		}

		for _, cluster := range mgmtClusters {
			clinicC := utils.MgmtClusterToModel(clinicOrg.ID, cluster)
			err = model2.InsertOrUpdateNGCluster(clinicC, j.ForceUpdate)
			if err != nil {
				log.Error("Migrate mgmt cluster failed", zap.Any("clusterID", cluster.ClusterId), zap.Error(err))
				continue
			}
		}
		log.Info("mgmt clusters updated", zap.Int("count", len(mgmtClusters)), zap.Uint64("tenantID", org.Id))
	}
}

type MigrateFullClusterJob struct {
	ForceUpdate bool
}

func (j *MigrateFullClusterJob) Run() {
	log.Info("corn job running", zap.String("name", "MigrateDedicatedClusterJob"))
	defer func() {
		if err := recover(); err != nil {
			log.Error("MigrateDedicatedClusterJob failed", zap.String("err", fmt.Sprintf("%v", err)))
		}
	}()

	adminUserCtx := &session.CtxUser{
		Internal: true,
		Email:    "clinic@pingcap.com",
	}

	recoveringClusters, err := model2.ListRecoveringClusters()
	if err != nil {
		log.Error("Get recovering clusters failed", zap.Error(err))
		return
	}
	log.Info("recovering clusters", zap.Int("count", len(recoveringClusters)))
	for _, c := range recoveringClusters {
		clusterID, err := strconv.ParseUint(c.ClusterID, 10, 64)
		if err != nil {
			log.Error("failed to parse cluster id", zap.String("clusterID", c.ClusterID), zap.Error(err))
			continue
		}
		cluster, err := datasource.RealtimeGetMgmtCluster(adminUserCtx, clusterID)
		if err != nil {
			log.Error("failed to realtime get cloud cluster", zap.Uint64("clusterID", clusterID), zap.Error(err))
			continue
		}
		if err := model2.InsertOrUpdateCloudCluster(cluster, j.ForceUpdate); err != nil {
			log.Error("Migrate dedicated cluster failed", zap.Any("clusterID", cluster.ClusterId), zap.Error(err))
			continue
		}
		log.Info("recovering dedicated cluster updated",
			zap.Uint64("tenantID", cluster.OrganizationId),
			zap.Uint64("clusterID", cluster.ClusterId))
	}

	cloudOrgs, _, err := datasource.RealTimeListOrganizations(adminUserCtx)
	if err != nil {
		log.Error("Get organization from account server failed", zap.Error(err))
	}
	log.Info("RealTimeListOrganizations", zap.Int("count", len(cloudOrgs)))

	for _, org := range cloudOrgs {
		cloudClusters, err := datasource.RealtimeListMgmtClustersByOrg(adminUserCtx, org.Id)
		if err != nil {
			log.Error("Get dedicated clusters failed", zap.Error(err), zap.Uint64("tenantID", org.Id))
			continue
		}
		for _, cluster := range cloudClusters {
			err = model2.InsertOrUpdateCloudCluster(cluster, j.ForceUpdate)
			if err != nil {
				log.Error("Migrate dedicated cluster failed", zap.Any("clusterID", cluster.ClusterId), zap.Error(err))
			}
		}
		if len(cloudClusters) > 0 {
			log.Info("dedicated clusters updated", zap.Int("count", len(cloudClusters)), zap.Uint64("tenantID", org.Id))
		} else {
			log.Info("dedicated org has no cluster", zap.Uint64("tenantID", org.Id))
		}
	}
}

type MigrateFullServerlessClusterJob struct {
	ForceUpdate    bool
	IncludeDeleted bool
}

func (j *MigrateFullServerlessClusterJob) Run() {
	log.Info("cron job running", zap.String("name", "MigrateFullServerlessClusterJob"))
	defer func() {
		if err := recover(); err != nil {
			log.Error("MigrateServerlessClusterJob failed", zap.String("err", fmt.Sprintf("%v", err)))
		}
	}()

	adminUserCtx := &session.CtxUser{
		Internal: true,
		Email:    "clinic@pingcap.com",
	}
	cloudOrgs, _, err := datasource.RealTimeListOrganizations(adminUserCtx)
	if err != nil {
		log.Error("Get organization from account server failed", zap.Error(err))
	}
	log.Info("RealTimeListOrganizations", zap.Int("count", len(cloudOrgs)))

	for _, org := range cloudOrgs {
		cloudClusters, err := datasource.RealtimeListServerlessClusters(adminUserCtx, org.Id, j.IncludeDeleted)
		if err != nil {
			log.Error("Get serverless clusters from core-svc failed", zap.Error(err), zap.Uint64("tenantID", org.Id))
			continue
		}
		log.Info("get serverless cluster", zap.Int("count", len(cloudClusters)), zap.Uint64("tenantID", org.Id))

		for _, cluster := range cloudClusters {
			err = model2.InsertOrUpdateCloudCluster(cluster, j.ForceUpdate)
			if err != nil {
				log.Error("Migrate serverless cluster failed", zap.Any("clusterID", cluster.ClusterId), zap.Error(err))
			}
			// ignore cleared cluster to reduce traffic
			clusterStatus := utils.GetClusterStatus(cluster)
			if clusterStatus == openapiv1.ClusterState_DELETED {
				continue
			}
			branches, err := datasource.RealtimeListServerlessBranches(adminUserCtx, cluster.ClusterId)
			if err != nil {
				log.Error("get branch list failed", zap.Any("clusterID", cluster.ClusterId), zap.Error(err))
				continue
			}
			// log.Info("RealtimeListServerlessBranches", zap.Int("count", len(branches)), zap.Uint64("parentCluster", cluster.Id))

			var branchList []string
			for _, b := range branches {
				err = model2.InsertOrUpdateServerlessBranch(cluster, b, j.ForceUpdate)
				if err != nil {
					log.Error("Migrate serverless branch failed", zap.Any("branchID", b.Id), zap.Error(err))
				}
				branchList = append(branchList, b.Id)
			}
			if model2.UpdateBranches(cluster.ClusterId, branchList) != nil {
				log.Error("Failed to update branches field of cluster table", zap.Any("clusterID", cluster.ClusterId))
			}
		}
	}
	log.Info("corn job finished", zap.String("name", "MigrateFullServerlessClusterJob"))
}

type QpsReportTriggerJob struct {
	ceEndpoint string
	reportCfg  config.ReportConfig
}

func (j *QpsReportTriggerJob) Run() {
	// log.Info("corn job running", zap.String("name", "MigrateOrganizationJob"))
	defer func() {
		if err := recover(); err != nil {
			log.Error("QpsReportTriggerJob failed", zap.String("err", fmt.Sprintf("%v", err)))
		}
	}()

	if j.ceEndpoint == "" {
		log.Error("QpsReportTriggerJob failed", zap.String("err", "invalid compute engine endpoint"))
		return
	}

	m := qps_report.NewQpsReprotManager(j.reportCfg, j.ceEndpoint)
	err := m.Run()
	if err != nil {
		log.Error("run qps report trigger failed", zap.Error(err))
	}
}

type SyncDeployTypeJob struct{}

func (j *SyncDeployTypeJob) Run() {
	log.Info("cron job running", zap.String("name", "SyncDeployTypeJob"))
	defer func() {
		if err := recover(); err != nil {
			log.Error("SyncDeployTypeJob failed", zap.String("err", fmt.Sprintf("%v", err)))
		}
	}()
	adminUserCtx := &session.CtxUser{
		Internal: true,
		Email:    "clinic@pingcap.com",
	}

	total, err := datasource.CountCloudClusters(adminUserCtx)
	if err != nil {
		log.Error("Failed to count all clusters", zap.Error(err))
		return
	}

	var pageSize int64 = 100
	pageCount := int((total + pageSize - 1) / pageSize)
	log.Info("Count clusters", zap.Int64("total", total), zap.Int("pages", pageCount))

	for i := 1; i <= pageCount; i++ {
		clusters, err := datasource.ListCloudClusters(adminUserCtx, i, int(pageSize))
		if err != nil {
			log.Error("Failed to get clusters", zap.Int("page", i), zap.Int("pageCount", pageCount), zap.Error(err))
			return
		}
		clusterIds := make([]uint64, 0, len(clusters))
		clusterMap := make(map[uint64]db.Cluster, len(clusters))
		for _, cluster := range clusters {
			// has already migrated
			if cluster.DeployTypeV2 != "" {
				continue
			}

			clusterId, err := strconv.ParseUint(cluster.ClusterID, 10, 64)
			if err != nil {
				log.Error("Failed to parse clusterID", zap.String("clusterID", cluster.ClusterID))
				continue
			}
			clusterIds = append(clusterIds, clusterId)
			clusterMap[clusterId] = cluster
		}
		if len(clusterIds) == 0 {
			log.Info("No valid cluster ids", zap.Int("page", i), zap.Int("pageCount", pageCount))
			continue
		}

		mgmtClusters, err := datasource.RealtimeGetMgmtClusters(adminUserCtx, clusterIds)
		if err != nil {
			log.Error("Failed to get mgmt clusters", zap.Int("page", i), zap.Int("pageCount", pageCount), zap.Any("clusterIds", clusterIds), zap.Error(err))
			return
		}

		updatedClusterIds := make(map[uint64]struct{}, 0)
		deployTypeV2ClusterMap := make(map[string][]string)
		for _, mgmtCluster := range mgmtClusters {
			deployTypeV2 := utils.GetMgmtDeployTypeV2(mgmtCluster)
			cluster := clusterMap[mgmtCluster.ClusterId]
			if deployTypeV2 == "" {
				log.Warn("Cluster has no deploy_type_v2", zap.String("clusterId", cluster.ClusterID))
			} else if cluster.DeployTypeV2 != deployTypeV2 {
				deployTypeV2ClusterMap[deployTypeV2] = append(deployTypeV2ClusterMap[deployTypeV2], cluster.ClusterID)
				log.Info("Updated cluster deploy_type_v2", zap.String("clusterId", cluster.ClusterID), zap.String("old", cluster.DeployType), zap.String("new", deployTypeV2))

				if cluster.DeployType == o11y.SharedType {
					for _, branch := range cluster.Branches {
						if branch == "" || branch == "null" {
							continue
						}
						deployTypeV2ClusterMap[deployTypeV2] = append(deployTypeV2ClusterMap[deployTypeV2], branch)
					}
				}
			}
			updatedClusterIds[mgmtCluster.ClusterId] = struct{}{}
		}

		for deployTypeV2, clusterIds := range deployTypeV2ClusterMap {
			err = model.UpdateClusterDeployType(deployTypeV2, clusterIds)
			if err != nil {
				log.Error("Failed to update cluster deploy_type", zap.String("deploy_type", deployTypeV2), zap.Strings("clusterIds", clusterIds))
				return
			}
		}

		for _, clusterId := range clusterIds {
			if _, ok := updatedClusterIds[clusterId]; !ok {
				log.Warn("Cluster not found", zap.Uint64("clusterID", clusterId))
			}
		}

		log.Info("Sync cluster deploy type v2", zap.Int("page", i), zap.Int("pageCount", pageCount), zap.Int("count", len(clusterIds)))
	}
}

type SyncParentIdJob struct{}

func (j *SyncParentIdJob) Run() {
	log.Info("cron job running", zap.String("name", "SyncParentIdJob"))
	defer func() {
		if err := recover(); err != nil {
			log.Error("SyncParentIdJob failed", zap.String("err", fmt.Sprintf("%v", err)))
		}
	}()

	adminUserCtx := &session.CtxUser{
		Internal: true,
		Email:    "clinic@pingcap.com",
	}

	pageSize := 100
	finished := false
	for page := 1; !finished; page++ {
		clusters, err := model.ListNextNoParentIdGenClusters(page, pageSize)
		if err != nil {
			log.Error("Failed to get clusters", zap.Int("page", page), zap.Error(err))
			return
		}
		finished = len(clusters) < pageSize

		nextGenClusterIds := make([]uint64, 0, pageSize)
		for _, cluster := range clusters {
			clusterId, err := strconv.ParseUint(cluster.ClusterID, 10, 64)
			if err != nil {
				log.Error("Failed to parse clusterID", zap.String("clusterID", cluster.ClusterID))
				continue
			}
			nextGenClusterIds = append(nextGenClusterIds, clusterId)
		}

		if len(nextGenClusterIds) > 0 {
			mgmtClusters, err := datasource.RealtimeGetMgmtClusters(adminUserCtx, nextGenClusterIds)
			if err != nil {
				log.Error("Failed to get mgmt clusters",
					zap.Any("clusterIds", nextGenClusterIds),
					zap.Error(err))
				return
			}
			for _, mgmtCluster := range mgmtClusters {
				if mgmtCluster.DeployType != mgmtv1.DeployType_DEPLOY_TYPE_NEXTGEN {
					log.Warn("Cluster is not next gen cluster",
						zap.Any("clusterId", mgmtCluster.ClusterId),
						zap.Error(err))
					continue
				}

				clusterId := strconv.FormatUint(mgmtCluster.ClusterId, 10)
				nextGenCluster := mgmtCluster.GetNextGenCluster()
				err = model.UpdateClusterParentId(clusterId, nextGenCluster.GetSharedPoolId())
				if err != nil {
					log.Error("Failed to update parent id",
						zap.Any("clusterId", clusterId),
						zap.Error(err))
					return
				}
			}
		}

		log.Info("Sync parent id",
			zap.Int("page", page),
			zap.Int("nextGenClusterIds", len(nextGenClusterIds)))
	}
}

type MigrateDeployTypeJob struct{}

func (j *MigrateDeployTypeJob) Run() {
	err := model.MigrateDeployTypeV2FromDeployType(o11y.NgHostType, o11y.DeployTypeV2PremiumResourcePool)
	if err != nil {
		log.Error("Failed to migrate nextgen-host deploy_type_v2")
	}
	err = model.MigrateDeployTypeV2FromDeployType(o11y.ByocHostType, o11y.DeployTypeV2ByocResourcePool)
	if err != nil {
		log.Error("Failed to migrate byoc-host deploy_type_v2")
	}
	err = model.MigrateDeployTypeV2FromDeployType(o11y.DedicatedType, o11y.DeployTypeV2Dedicated)
	if err != nil {
		log.Error("Failed to migrate dedicated deploy_type_v2")
	}
	err = model.MigrateDeployTypeV2FromDeployType(o11y.PremiumType, o11y.DeployTypeV2Premium)
	if err != nil {
		log.Error("Failed to migrate premium deploy_type_v2")
	}
	err = model.MigrateDeployTypeV2FromDeployType(o11y.ByocType, o11y.DeployTypeV2Byoc)
	if err != nil {
		log.Error("Failed to migrate premium deploy_type_v2")
	}
	err = model.MigrateDeployTypeV2FromDeployType("tidb-operator", "tidb-operator")
	if err != nil {
		log.Error("Failed to migrate tidb-operator deploy_type_v2")
	}
	err = model.MigrateDeployTypeV2FromDeployType("tiup-cluster", "tiup-cluster")
	if err != nil {
		log.Error("Failed to migrate tiup-cluster deploy_type_v2")
	}
	err = model.MigrateServerlessBranchDeployType()
	if err != nil {
		log.Error("Failed to migrate serverless branch deploy_type_v2")
	}
	err = model.SetDefaultDeployTypeV2(o11y.SharedType, o11y.DeployTypeV2Starter)
	if err != nil {
		log.Error("Failed to migrate shared default deploy_type_v2")
	}
}

type MigrateClusterStatusJob struct{}

func (j *MigrateClusterStatusJob) Run() {
	err := model.MigrateClusterStatus(utils.State2String(openapiv1.ClusterState_DELETED), []string{"cleared", "deletecomplete"})
	if err != nil {
		log.Error("Failed to migrate cluster deleted status")
	}
	err = model.MigrateClusterStatus(utils.State2String(openapiv1.ClusterState_ACTIVE), []string{"normal", "available"})
	if err != nil {
		log.Error("Failed to migrate cluster active status")
	}
	err = model.MigrateClusterStatus(utils.State2String(openapiv1.ClusterState_CREATING), []string{"initializing"})
	if err != nil {
		log.Error("Failed to migrate cluster creating status")
	}
}

type SyncResourcePool struct {
	ForceUpdate bool
}

func (j *SyncResourcePool) Run() {
	log.Info("cron job running", zap.String("name", "SyncResourcePool"))
	defer func() {
		if err := recover(); err != nil {
			log.Error("SyncResourcePool failed", zap.String("err", fmt.Sprintf("%v", err)))
		}
	}()
	adminUserCtx := &session.CtxUser{
		Internal: true,
		Email:    "clinic@pingcap.com",
	}

	total, err := datasource.CountResourcePoolClusters(adminUserCtx)
	if err != nil {
		log.Error("Failed to count all clusters", zap.Error(err))
		return
	}

	var pageSize int64 = 100
	pageCount := int((total + pageSize - 1) / pageSize)
	log.Info("Count resource pool clusters", zap.Int64("total", total), zap.Int("pages", pageCount))
	for i := 1; i <= pageCount; i++ {
		clusters, err := datasource.ListResourcePoolClusters(adminUserCtx, i, int(pageSize))
		if err != nil {
			log.Error("Failed to get clusters", zap.Int("page", i), zap.Int("pageCount", pageCount), zap.Error(err))
			return
		}
		resourcePoolIDs := make([]uint64, 0, len(clusters))
		clusterMap := make(map[uint64]*db.Cluster, len(clusters))
		for i := range clusters {
			cluster := &clusters[i]
			clusterID, err := strconv.ParseUint(cluster.ClusterID, 10, 64)
			if err != nil {
				log.Error("Failed to parse clusterID", zap.String("clusterID", cluster.ClusterID))
				continue
			}
			resourcePoolIDs = append(resourcePoolIDs, clusterID)
			clusterMap[clusterID] = cluster
		}
		if len(clusterMap) == 0 {
			log.Warn("No valid resource pool id", zap.Int("page", i), zap.Int("pageCount", pageCount), zap.Error(err))
			continue
		}

		resourcePools, err := datasource.ListResourcePoolWithProfile(adminUserCtx, resourcePoolIDs)
		if err != nil {
			log.Error("Failed to get resource pool clusters", zap.Int("page", i), zap.Int("pageCount", pageCount), zap.Any("resourcePools", resourcePools), zap.Error(err))
			return
		}
		existedResourcePools := make(map[string]struct{}, len(resourcePools))
		for _, resourcePool := range resourcePools {
			cluster, ok := clusterMap[resourcePool.ResourcePool.Id]
			if !ok {
				log.Warn("Unexpected resource pool id", zap.Int("page", i), zap.Int("pageCount", pageCount), zap.Any("resourcePool", resourcePool))
				continue
			}
			existedResourcePools[cluster.ClusterID] = struct{}{}
			newCluster := utils.ResourcePoolToModel(cluster.OrgID, cluster.TenantID, resourcePool, nil)
			if newCluster.Status != "deleted" {
				newCluster.Status = cluster.Status
			}
			err := model2.InsertOrUpdateNGCluster(newCluster, j.ForceUpdate)
			if err != nil {
				log.Error("Failed to update resource pool", zap.String("id", newCluster.ClusterID), zap.Error(err))
				return
			}
		}

		deletedCount := 0
		for _, cluster := range clusters {
			if _, ok := existedResourcePools[cluster.ClusterID]; !ok && cluster.Status != "deleted" {
				err = model.DeleteResourcePool(cluster.ClusterID)
				if err != nil {
					log.Error("Failed to delete resource pool", zap.String("id", cluster.ClusterID), zap.Error(err))
					return
				}
				deletedCount++
			}
		}

		log.Info("Sync resource pool cluster",
			zap.Int("page", i),
			zap.Int("pageCount", pageCount),
			zap.Int("count", len(resourcePoolIDs)),
			zap.Int("deleted", deletedCount),
			zap.Int("updated", len(existedResourcePools)))
	}
}

type SyncResourcePoolStatus struct{}

func (j *SyncResourcePoolStatus) Run() {
	log.Info("cron job running", zap.String("name", "SyncResourcePoolStatus"))
	defer func() {
		if err := recover(); err != nil {
			log.Error("SyncResourcePoolStatus failed", zap.String("err", fmt.Sprintf("%v", err)))
		}
	}()
	adminUserCtx := &session.CtxUser{
		Internal: true,
		Email:    "clinic@pingcap.com",
	}

	total, err := datasource.CountResourcePoolClusters(adminUserCtx)
	if err != nil {
		log.Error("Failed to count all clusters", zap.Error(err))
		return
	}

	var pageSize int64 = 100
	pageCount := int((total + pageSize - 1) / pageSize)
	log.Info("Count resource pool clusters", zap.Int64("total", total), zap.Int("pages", pageCount))
	for i := 1; i <= pageCount; i++ {
		clusters, err := datasource.ListResourcePoolClusters(adminUserCtx, i, int(pageSize))
		if err != nil {
			log.Error("Failed to get clusters", zap.Int("page", i), zap.Int("pageCount", pageCount), zap.Error(err))
			return
		}

		updatedCount := 0
		for _, cluster := range clusters {
			clusterId, _ := strconv.ParseUint(cluster.ClusterID, 10, 64)
			resourcePool, shardPool, err := datasource.GetResourcePoolWithProfile(adminUserCtx, clusterId)
			if err != nil {
				log.Error("Failed to get resource pool", zap.String("id", cluster.ClusterID), zap.Error(err))
				continue
			}
			status := utils.State2String(utils.GetResourcePoolStatus(resourcePool.ResourcePool, shardPool))
			if status != cluster.Status {
				err = model.UpdateCluster(cluster.ClusterID, cluster.Name, cluster.Desc, status)
				if err != nil {
					log.Error("Failed to update cluster status", zap.String("id", cluster.ClusterID), zap.Error(err))
					continue
				}
				updatedCount++
			}
		}

		log.Info("Sync resource pool cluster status",
			zap.Int("page", i),
			zap.Int("pageCount", pageCount),
			zap.Int("count", len(clusters)),
			zap.Int("updated", updatedCount))
	}
}

type CronJob struct {
	cron *cron.Cron
	Jobs map[cron.EntryID]cron.Job
}

func NewCron(cfg config.Config) *CronJob {
	// the cron should run in 02:00:00 UTC+8
	tzone, err := time.LoadLocation("Asia/Shanghai")
	if err != nil {
		log.Warn("load time location failed", zap.Error(err))
		tzone = time.Local
	}
	log.Info("new cron", zap.String("timezone", tzone.String()))
	c := &CronJob{cron: cron.New(cron.WithLocation(tzone)), Jobs: make(map[cron.EntryID]cron.Job)}
	_ = c.AddJob("0 2 * * ?", &DeleteLogDataJob{lc: cfg.SinkConfig.LogConfig, ExpiredDay: expired})
	_ = c.AddJob("0 2 * * ?", &DeleteMetricDataJob{mc: cfg.SinkConfig.MetricConfig, ExpiredDay: expired})
	_ = c.AddJob("0 2 * * ?", NewDeleteLeakingEBSJob(cfg.FileStorage.AWSStorage.Region, cfg.EBSFilterTags.ToMap()))
	if cfg.LinkTiDBCloud {
		_ = c.addJob(
			"*/2 * * * ?",
			&MigrateOrganizationJob{},
			cron.SkipIfStillRunning(cron.DefaultLogger),
		)
		// _ = c.AddJob("0 8 * * ?", &MigrateFullClusterJob{})
		_ = c.AddJob("30 7 * * ?", &MigrateMgmtClusterJob{})
		_ = c.AddJob("0 7 * * ?", &SyncResourcePool{})
		//_ = c.AddJob("0 7 * * ?", &MigrateFullServerlessClusterJob{})
		//_ = c.AddJob(cfg.ReportConfig.QpsReportCronPlan, &QpsReportTriggerJob{reportCfg: cfg.ReportConfig, ceEndpoint: cfg.GetComputeEngineEndPoint()})
	}
	return c
}

// "0 2 * * ?", DeleteVmDataJob{"dj"}
func (c *CronJob) AddJob(cronplan string, job cron.Job) error {
	return c.addJob(cronplan, job)
}

func (c *CronJob) addJob(cronplan string, job cron.Job, wrappers ...cron.JobWrapper) error {
	chain := append([]cron.JobWrapper{cron.Recover(cron.DefaultLogger)}, wrappers...)
	entryid, err := c.cron.AddJob(cronplan, cron.NewChain(chain...).Then(job))
	if err != nil {
		return err
	}
	c.Jobs[entryid] = job
	return nil
}

func (c *CronJob) Start() {
	c.cron.Start()
}