package service

import (
	"context"
	"errors"
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/lithammer/shortuuid/v3"
	"github.com/tidbcloud/eventbus-datastruct/event"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy/serverless"
	"github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/pkgv2/logid"
	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/facade"
	"github.com/tidbcloud/serverless-service/internal/domain/repo"
	"github.com/tidbcloud/serverless-service/internal/infra/eventbus"
	"github.com/tidbcloud/serverless-service/internal/utils"
	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"github.com/tidbcloud/tiinf/transport/tieventbus"
	"go.uber.org/multierr"
	"go.uber.org/zap"
)

const (
	BackupJobCleanDays                    = 90
	UserBackupKeepTime                    = 30 * 24 * time.Hour
	UserBackupCleanBatch                  = 5000
	BackupScheduleCleanBatch              = 5000
	backupScheduleSnapshotRepairBatchSize = 500

	ExecuteRestoreRetryTimes = 3

	backupScheduleDefault = "0 0 0 * * *"
)

// UTC time
var regionToCron = map[string]string{
	globalcommonv1.Region_AWS_AP_SOUTHEAST_1.String(): "0 0 18 * * *",
	globalcommonv1.Region_AWS_US_EAST_1.String():      "0 0 7 * * *",
	globalcommonv1.Region_AWS_US_WEST_2.String():      "0 0 10 * * *",
	globalcommonv1.Region_AWS_AP_NORTHEAST_1.String(): "0 0 17 * * *",
	globalcommonv1.Region_AWS_EU_CENTRAL_1.String():   "0 0 3 * * *",
	// 3:00 UTC+8
	globalcommonv1.Region_AWS_AP_EAST_1.String(): "0 0 19 * * *",

	// 3:00 UTC+7
	globalcommonv1.Region_ALICLOUD_AP_SOUTHEAST_5.String(): "0 0 20 * * *",
	// 2:00 UTC+8
	globalcommonv1.Region_ALICLOUD_AP_SOUTHEAST_1.String(): "0 0 18 * * *",
	// 2:00 UTC+9
	globalcommonv1.Region_ALICLOUD_AP_NORTHEAST_1.String(): "0 0 17 * * *",
	// 2:00 UTC-6
	globalcommonv1.Region_ALICLOUD_NA_SOUTH_1.String(): "0 0 8 * * *",
}

type BackupWorker interface {
	CleanBackup(ctx context.Context) (err error)
	SyncRestoreProgress(ctx context.Context) (err error)
	ExecuteRestore(ctx context.Context) (err error)
	CheckBackupSchedule(ctx context.Context, date string) (err error)
}

type backupWorker struct {
	clusterProxyProviderFacade facade.ClusterProxyProviderFacade
	backupJobRepo              repo.BackupJobRepo
	backupSchedule             repo.BackupScheduleRepo
	userBackupRepo             repo.UserBackupRepo
	dbProvider                 db.TransProvider
	restoreJobRepo             repo.RestoreJobRepo
	mgmtSvc                    facade.MgmtSvcFacade
	eventbusService            *eventbus.EventbusService
	billingFacade              facade.BillingFacade
	backupScheduleCheckJobRepo repo.BackupScheduleCheckJobRepo
}

func NewBackupWorker(clusterProxyProviderFacade facade.ClusterProxyProviderFacade, backupJobRepo repo.BackupJobRepo,
	backupSchedule repo.BackupScheduleRepo, dbProvider db.TransProvider,
	userBackupRepo repo.UserBackupRepo, restoreJobRepo repo.RestoreJobRepo,
	mgmtSvc facade.MgmtSvcFacade, eventbusService *eventbus.EventbusService, billingFacade facade.BillingFacade,
	backupScheduleCheckJobRepo repo.BackupScheduleCheckJobRepo,
) BackupWorker {
	return &backupWorker{
		clusterProxyProviderFacade: clusterProxyProviderFacade,
		backupJobRepo:              backupJobRepo,
		backupSchedule:             backupSchedule,
		userBackupRepo:             userBackupRepo,
		dbProvider:                 dbProvider,
		restoreJobRepo:             restoreJobRepo,
		mgmtSvc:                    mgmtSvc,
		eventbusService:            eventbusService,
		billingFacade:              billingFacade,
		backupScheduleCheckJobRepo: backupScheduleCheckJobRepo,
	}
}

// CleanBackup clean backup, user backup and backup job
func (b *backupWorker) CleanBackup(ctx context.Context) error {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	logger.Info("start to clean backup")
	var errs error
	if err := b.backupJobRepo.Delete(ctx, BackupJobCleanDays); err != nil {
		logger.Error("clean backup jobs failed", zap.Error(err))
		errs = multierr.Append(errs, err)
	}
	// we will keep the expired user backup for 30 days
	// delete all the expired user backup in batch
	curTime := time.Now().Add(-UserBackupKeepTime)
	deleteTimes := 1
	for {
		logger.Info("start to clean user backups", zap.Int("times", deleteTimes))
		if err := b.userBackupRepo.Delete(ctx, entity.UserBackupDeleteOptions{
			ExpireTimeLt: &curTime,
			Limit:        UserBackupCleanBatch,
		}); err != nil {
			if err.ErrCode() == bizerr.ErrCodeRecordNotFound {
				logger.Info("no more user backups to delete")
				break
			}
			logger.Error("clean user backups failed", zap.Error(err))
			errs = multierr.Append(errs, err)
			break
		}
		time.Sleep(1 * time.Second)
		deleteTimes++
	}
	cleanScheduleTimes := 1
	for {
		logger.Info("start to clean deleted backup schedules", zap.Int("times", cleanScheduleTimes))
		deletedCount, err := b.backupSchedule.CleanDeletedWithoutBackup(ctx, time.Now(), BackupScheduleCleanBatch)
		if err != nil {
			logger.Error("clean deleted backup schedules failed", zap.Error(err))
			errs = multierr.Append(errs, err)
			break
		}
		if deletedCount == 0 {
			logger.Info("no more deleted backup schedules to clean")
			break
		}
		time.Sleep(1 * time.Second)
		cleanScheduleTimes++
	}
	return errs
}

func (b *backupWorker) SyncRestoreProgress(ctx context.Context) (err error) {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	logger.Info("start to sync restore")

	restoreJobs, err := b.restoreJobRepo.List(ctx, entity.ListRestoreJobOptions{State: constant.RestoreJobStateRunning})
	if err != nil {
		logger.Error("list restore jobs failed", zap.Error(err))
		return err
	}

	failedCount := 0
	for _, restoreJob := range restoreJobs {
		if err := b.syncOneRestoreJob(ctx, restoreJob); err != nil {
			logger.Error("sync one restore job failed", zap.String("id", restoreJob.ID), zap.Error(err))
			failedCount++
		}
	}

	logger.Info("sync restore finished", zap.Int("total", len(restoreJobs)), zap.Int("failed count", failedCount))
	if failedCount > 0 {
		return fmt.Errorf("sync restore failed, failed count: %d", failedCount)
	}
	return nil
}

func (b *backupWorker) syncOneRestoreJob(ctx context.Context, restoreJob *entity.RestoreJob) error {
	logger := utils.GetWorkerLoggerFromContext(ctx).
		With(
			zap.String("id", restoreJob.ID),
			zap.String("source_cluster_id", restoreJob.SourceClusterId),
			zap.String("target_cluster_id", restoreJob.TargetClusterId),
			zap.Uint64("restore_id", restoreJob.RestoreId),
		)
	logger.Info("start to sync one restore job")
	// get cluster proxy
	clusterProxyUrl := config.GetClusterProxyUrl(
		restoreJob.GetCloudProvider(),
		restoreJob.Region,
		restoreJob.GetKubernetes(),
	)
	if clusterProxyUrl == "" {
		logger.Error("cluster proxy url not found in config", zap.String("kubernetes", restoreJob.GetKubernetes()))
		return fmt.Errorf("cluster proxy url for %s not found in config", restoreJob.GetKubernetes())
	}
	clusterProxy, err := b.clusterProxyProviderFacade.GetServerlessProxy(ctx, clusterProxyUrl)
	if err != nil {
		logger.Error("get cluster proxy failed", zap.Error(err))
		return err
	}
	// get restore progress
	resp, err := clusterProxy.GetServerlessRestoreProgress(ctx, restoreJob.TargetToken, restoreJob.RestoreId)
	if err != nil {
		logger.Error("get restore progress failed", zap.Error(err))
		return err
	}
	logger.Info("get restore progress", zap.Any("resp", resp))
	// retry restore job when failed
	if resp.Status == serverless.RestoreError {
		restoreJob.SizeBytes = 0
		restoreJob.Progress.Step = ""
		restoreJob.Progress.Percent = 0
		logger.Warn("sync restore failed, will retry", zap.String("error", resp.Error))
		restoreJob.State = constant.RestoreJobSatePending
		if err = b.restoreJobRepo.Save(ctx, restoreJob); err != nil {
			logger.Error("update restore job failed", zap.Error(err))
			return err
		}
		return nil
	}

	// update restore job
	jobChanged := false
	if restoreJob.SizeBytes != resp.RestoreBytes {
		restoreJob.SizeBytes = resp.RestoreBytes
		jobChanged = true
	}
	if restoreJob.Progress.Step != resp.Progress.Step {
		restoreJob.Progress.Step = resp.Progress.Step
		jobChanged = true
	}
	if restoreJob.Progress.Percent != resp.Progress.Progress {
		restoreJob.Progress.Percent = resp.Progress.Progress
		jobChanged = true
	}
	if jobChanged {
		restoreJob.Progress.LastTransitionTS = time.Now().Unix()
	}
	if resp.Status == serverless.RestoreSucceed {
		restoreJob.State = constant.RestoreJobStateSucceeded
		jobChanged = true
		if err := b.recoverStatusAfterRestore(ctx, restoreJob); err != nil {
			logger.Error("recover status after restore failed", zap.Error(err))
			return err
		}
	}
	if jobChanged {
		// update restore job at last so that we can retry when failed to recover status
		if err = b.restoreJobRepo.Save(ctx, restoreJob); err != nil {
			logger.Error("update restore job failed", zap.Error(err))
			return err
		}
	}
	// publish events
	if restoreJob.State == constant.RestoreJobStateSucceeded {
		b.publishRestoreEvent(ctx, restoreJob, event.EventStatusSucceeded)
	}
	return nil
}

func (b *backupWorker) recoverStatusAfterRestore(ctx context.Context, restoreJob *entity.RestoreJob) error {
	logger := utils.GetWorkerLoggerFromContext(ctx).
		With(
			zap.String("id", restoreJob.ID),
			zap.String("source_cluster_id", restoreJob.SourceClusterId),
			zap.String("target_cluster_id", restoreJob.TargetClusterId),
			zap.Uint64("restore_id", restoreJob.RestoreId),
		)
	logger.Info("start to recover status after restore done")
	// 1. update target cluster annotation
	targetClusterId, err := strconv.ParseUint(restoreJob.TargetClusterId, 10, 64)
	if err != nil {
		logger.Error("parse cluster id failed", zap.Error(err))
		return err
	}
	_, bizErr := b.mgmtSvc.GetCluster(ctx, targetClusterId, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_LOCAL)
	if bizErr != nil {
		if bizErr.RawErrCode() == bizerr.ErrCodeRecordNotFound {
			if restoreJob.IsRestoreToNewCluster() {
				logger.Warn("target cluster not found when restore to new, not retry anymore")
				return b.deleteSourceClusterCustomStatus(ctx, restoreJob.SourceClusterId, restoreJob.ClusterStatusId)
			}
			// 1. cluster not find error is impossible for restore in place, because source cluster can not be deleted when restored.
			// 2. even the cluster is deleted, there is no need to delete the cluster custom status for restore in place any more.
			logger.Warn(
				"target cluster not found when restore in place, not retry anymore(it should not happen in theory)",
			)
			return nil
		}
		logger.Error("get cluster failed", zap.Error(bizErr))
		return bizErr
	}
	updated := make(map[string]string)
	if restoreJob.IsRestoreToNewCluster() {
		updated[constant.AnnoKeyRestoreToNewCluster] = constant.AnnoKeyRestoreToNewClusterFinish
	} else {
		updated[constant.AnnoRestoreStatusKey] = constant.AnnoRestoreStatusRestoreFinish
	}
	if _, err := b.mgmtSvc.UpdateServerlessClusterAnnotations(ctx, targetClusterId, updated); err != nil {
		logger.Error("update cluster annotations failed", zap.Error(err))
		return err
	}

	// 2. delete source cluster custom status
	if err = b.deleteSourceClusterCustomStatus(ctx, restoreJob.SourceClusterId, restoreJob.ClusterStatusId); err != nil {
		logger.Error("delete source cluster custom status failed", zap.Error(err))
		return err
	}
	return nil
}

func (b *backupWorker) deleteSourceClusterCustomStatus(
	ctx context.Context,
	sourceClusterId string,
	statusId uint64,
) error {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	if sourceClusterId == "" || statusId == 0 {
		logger.Warn(
			"skip delete source cluster custom status",
			zap.String("cluster_id", sourceClusterId),
			zap.Uint64("status_id", statusId),
		)
		return nil
	}
	clusterId, err := strconv.ParseUint(sourceClusterId, 10, 64)
	if err != nil {
		logger.Error("parse cluster id failed", zap.Error(err))
		return err
	}
	if err = b.mgmtSvc.DeleteClusterCustomStatus(ctx, clusterId, statusId); err != nil {
		logger.Error("delete cluster custom status failed", zap.Error(err))
		return err
	}
	logger.Info("delete cluster custom status success")
	return nil
}

func (b *backupWorker) publishRestoreEvent(
	ctx context.Context,
	restoreJob *entity.RestoreJob,
	status event.EventStatus,
) {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	clusterId, err := strconv.ParseUint(restoreJob.SourceClusterId, 10, 64)
	if err != nil {
		return
	}
	cluster, err := b.mgmtSvc.GetCluster(ctx, clusterId, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_LOCAL)
	if err != nil {
		logger.Error("get cluster failed", zap.String("id", restoreJob.ID), zap.Error(err))
		return
	}
	restoreEvent, err := BuildRestoreEvent(ctx, cluster, restoreJob, status)
	if err != nil {
		logger.Error("build restore event failed", zap.String("id", restoreJob.ID), zap.Error(err))
		return
	}
	// publish restore event
	if err = b.eventbusService.Publish(&tieventbus.Event{
		ID:   shortuuid.New(),
		Type: event.DisplayEventTopic,
		Data: restoreEvent,
	}); err != nil {
		logger.Error("failed to publish events center", zap.String("id", restoreJob.ID), zap.Error(err))
	}
}

func (b *backupWorker) ExecuteRestore(ctx context.Context) (err error) {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	logger.Info("start to execute restore")

	restoreJobs, err := b.restoreJobRepo.List(ctx, entity.ListRestoreJobOptions{State: constant.RestoreJobSatePending})
	if err != nil {
		logger.Error("list restore jobs failed", zap.Error(err))
		return err
	}

	failedCount := 0
	for _, restoreJob := range restoreJobs {
		if err := b.restoreOneJob(ctx, restoreJob); err != nil {
			logger.Error("restore one job failed", zap.String("id", restoreJob.ID), zap.Error(err))
			failedCount++
		}
	}

	logger.Info(
		"execute restore job finished",
		zap.Int("total", len(restoreJobs)),
		zap.Int("failed count", failedCount),
	)
	if failedCount > 0 {
		return fmt.Errorf("restore job failed, failed count: %d", failedCount)
	}
	return nil
}

func (b *backupWorker) restoreOneJob(ctx context.Context, restoreJob *entity.RestoreJob) error {
	logger := utils.GetWorkerLoggerFromContext(ctx).
		With(
			zap.String("id", restoreJob.ID),
			zap.String("source_cluster_id", restoreJob.SourceClusterId),
			zap.String("target_cluster_id", restoreJob.TargetClusterId),
		)
	logger.Info("start to execute restore")
	targetClusterId, err := strconv.ParseUint(restoreJob.TargetClusterId, 10, 64)
	if err != nil {
		logger.Error("parse target cluster id failed", zap.Error(err))
		return err
	}
	// we need to get cluster bypass cache to fetch the changes happened in manager
	cluster, bizErr := b.mgmtSvc.GetClusterBypassCache(
		ctx,
		targetClusterId,
		tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC,
	)
	if bizErr != nil {
		if bizErr.RawErrCode() == bizerr.ErrCodeRecordNotFound {
			logger.Warn(
				"target cluster not found, skip the restore",
				zap.Bool("isRestoreToNewCluster", restoreJob.IsRestoreToNewCluster()),
			)
			// clean the source cluster custom status
			if restoreJob.IsRestoreToNewCluster() {
				if err := b.deleteSourceClusterCustomStatus(ctx, restoreJob.SourceClusterId, restoreJob.ClusterStatusId); err != nil {
					return err
				}
			}
			// update restore job status to failed
			restoreJob.State = constant.RestoreJobStateFailed
			restoreJob.Error = "target cluster not found"
			return b.restoreJobRepo.Save(ctx, restoreJob)
		}
		logger.Error("get cluster failed", zap.Error(bizErr))
		return bizErr
	}
	if restoreJob.IsRestoreToNewCluster() {
		// judge whether we can restore before restore to new cluster
		if cluster.GetServerlessCluster().GetCluster().Annotations[constant.AnnoKeyRestoreToNewCluster] != constant.AnnoKeyRestoreToNewClusterWait {
			logger.Info("skip restore job, target cluster is not wait for restore")
			return nil
		}
		// fill in the target token
		restoreJob.TargetToken = cluster.GetServerlessCluster().GetCluster().GetUserPrefix()
	} else {
		// ensure all connections are killed when restore in place
		switch cluster.GetServerlessCluster().GetCluster().Annotations[constant.AnnoRestoreStatusKey] {
		case constant.AnnoRestoreStatusStart:
			logger.Info("skip restore job, wait cluster to be restarted")
			return nil
		case constant.AnnoRestoreStatusRestartTiDB:
			logger.Info("cluster has restarted, start to restore")
		default:
			logger.Info("set restore status annotation to wait restart")
			updated := map[string]string{
				constant.AnnoRestoreStatusKey: constant.AnnoRestoreStatusStart,
			}
			if _, err := b.mgmtSvc.UpdateServerlessClusterAnnotations(ctx, targetClusterId, updated); err != nil {
				logger.Error("update cluster annotations failed", zap.Error(err))
				return err
			}
			return nil
		}
	}
	// fill in the kubernetes
	kubernetes := cluster.GetServerlessCluster().GetCluster().GetKubernetes()
	if kubernetes == "" {
		logger.Error("kubernetes not found in cluster")
		return errors.New("kubernetes not found in cluster")
	}
	restoreJob.Kubernetes = kubernetes
	restoreRequest, err := buildRestoreRequest(restoreJob)
	if err != nil {
		logger.Error("build restore request failed", zap.Error(err))
		return err
	}
	logger.Info("build restore request success", zap.Any("request", restoreRequest))
	return b.restoreAndRecord(ctx, restoreJob, restoreRequest)
}

func buildRestoreRequest(restoreJob *entity.RestoreJob) (*serverless.RestoreRequest, error) {
	restoreRequest := &serverless.RestoreRequest{
		Token: restoreJob.TargetToken,
	}
	// set source token if restore to new cluster
	if restoreJob.IsRestoreToNewCluster() {
		restoreRequest.SourceToken = &restoreJob.SourceToken
	}
	if restoreJob.BackupInfo != nil {
		restoreRequest.Backup = &serverless.BackupItem{
			ID:   restoreJob.BackupInfo.BackupID,
			Name: restoreJob.BackupInfo.BackupName,
			Time: serverless.JSONTime(*restoreJob.BackupInfo.BackupTime),
		}
	} else if restoreJob.BackupTime != nil {
		pointInTime := serverless.JSONTime(*restoreJob.BackupTime)
		restoreRequest.PointInTime = &pointInTime
	} else {
		return nil, errors.New("missing required restore info")
	}
	return restoreRequest, nil
}

func (b *backupWorker) restoreAndRecord(ctx context.Context, restoreJob *entity.RestoreJob,
	restoreRequest *serverless.RestoreRequest,
) error {
	logger := utils.GetWorkerLoggerFromContext(ctx).
		With(
			zap.String("id", restoreJob.ID),
			zap.String("source_cluster_id", restoreJob.SourceClusterId),
			zap.String("target_cluster_id", restoreJob.TargetClusterId),
		)
	// get cluster proxy url
	clusterProxyUrl := config.GetClusterProxyUrl(
		restoreJob.GetCloudProvider(),
		restoreJob.Region,
		restoreJob.GetKubernetes(),
	)
	if clusterProxyUrl == "" {
		logger.Error("cluster proxy url not found in config", zap.String("kubernetes", restoreJob.GetKubernetes()))
		return fmt.Errorf("cluster proxy url for %s not found in config", restoreJob.GetKubernetes())
	}
	// get cluster proxy
	clusterProxy, err := b.clusterProxyProviderFacade.GetServerlessProxy(ctx, clusterProxyUrl)
	if err != nil {
		logger.Error("get serverless cluster proxy failed", zap.Error(err))
		return errors.New("get serverless cluster proxy failed")
	}
	// restore 3 times in case restoreRequest.ID conflict
	var err1 error
	isRestoreIDExist := restoreJob.RestoreId != 0
	for i := 0; i < ExecuteRestoreRetryTimes; i++ {
		if err1 = b.dbProvider.Transaction(ctx, func(ctx context.Context) error {
			if !isRestoreIDExist {
				restoreJob.RestoreId = GenerateRestoreID()
			}
			restoreRequest.ID = restoreJob.RestoreId
			restoreJob.State = constant.RestoreJobStateRunning
			err := b.restoreJobRepo.Save(ctx, restoreJob)
			if err != nil {
				logger.Error("update restore job failed", zap.Uint64("restore_id", restoreJob.RestoreId), zap.Error(err))
				return err
			}
			logger.Info("update restore job success", zap.Uint64("restore_id", restoreJob.RestoreId))
			err = clusterProxy.RestoreServerlessCluster(ctx, *restoreRequest)
			if err != nil {
				logger.Error("execute restore failed", zap.Uint64("restore_id", restoreJob.RestoreId), zap.Error(err))
				return err
			}
			return nil
		}); err1 == nil {
			break
		}
		logger.Warn("record restore job and execute restore failed, retry", zap.Int("retry_times", i), zap.Error(err1))
	}
	if err1 != nil {
		logger.Error("record restore job and execute restore failed after retry", zap.Error(err1))
		return errors.New("record restore job and execute restore failed")
	}
	return nil
}

// GenerateRestoreID generates unique restore id in one instance
func GenerateRestoreID() uint64 {
	// add lock and sleep 1ms to avoid restore id generated in the same millisecond
	generateIdMutex.Lock()
	defer generateIdMutex.Unlock()
	time.Sleep(1 * time.Millisecond)
	// time.Now().UnixMilli() is 41 bit in 2024, and will be 43 bits after 100 years
	milliTs := time.Now().UnixMilli()
	// use lower 44 bits of milliTs to distinguish restore id
	// backup restore use 0x01FFF
	return uint64(0x01FFF00000000000 | (milliTs & 0xFFFFFFFFFFF)) // mask
}

func GetDefaultBackupScheduleCron(regionName string) string {
	for region, cron := range regionToCron {
		if strings.EqualFold(regionName, region) {
			return cron
		}
	}
	return backupScheduleDefault
}

func (b *backupWorker) CheckBackupSchedule(ctx context.Context, targetDate string) (err error) {
	logger := utils.GetWorkerLoggerFromContext(ctx)

	// check whether we have checked this date
	checkJob, err := b.getOrSaveBackupScheduleCheckJob(ctx, targetDate)
	if err != nil {
		return err
	}
	if err := b.repairBackupScheduleSnapshotFields(ctx); err != nil {
		return err
	}
	if checkJob.State == constant.BackupScheduleCheckJobStateSucceeded {
		logger.Info("check backup schedule already done", zap.String("date", targetDate))
		return nil
	}

	logger.Info("start to check backup schedule", zap.String("date", targetDate))

	targetTotal := 0
	inconsistentTotal := 0
	fixedTotal := 0
	defer func() {
		if err != nil {
			return
		}
		if inconsistentTotal-fixedTotal == 0 {
			checkJob.State = constant.BackupScheduleCheckJobStateSucceeded
		} else {
			checkJob.State = constant.BackupScheduleCheckJobStateFailed
		}
		_, err1 := b.backupScheduleCheckJobRepo.Save(ctx, checkJob)
		if err1 != nil {
			logger.Error("save backup schedule check job failed", zap.Error(err1))
		}
	}()

	var pageSize uint32 = 500
	var page uint32 = 1
	for {
		logger := logger.With(zap.Uint32("page", page))
		// list clusters from mgmt
		req := &tidbmgmtservicev1.ListClustersRequest{
			PageSize: pageSize,
			Page:     page,
			Filter: &tidbmgmtservicev1.ClusterFilter{
				View:          tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_LOCAL,
				ExcludeStates: facade.DeletedClusterStatus,
				ClusterType:   tidbmgmtservicev1.DeployType_DEPLOY_TYPE_SERVERLESS.Enum(),
			},
		}
		resp, err := b.mgmtSvc.ListClusters(ctx, req)
		if err != nil {
			logger.Error("list clusters failed", zap.Error(err))
			return err
		}
		// get yesterday's backup clusters
		targetClusters, allChecked := listTargetClusters(targetDate, resp.Clusters)
		targetTotal += len(targetClusters)
		logger.Info(
			"list target date clusters done",
			zap.Int("total", len(resp.Clusters)),
			zap.Int("target_date_total", len(targetClusters)),
		)
		// get inconsistent clusters
		inconsistentClusters, err := b.checkInconsistentClusters(ctx, targetClusters)
		if err != nil {
			logger.Error("check inconsistent clusters failed", zap.Error(err))
			return err
		}
		inconsistentTotal += len(inconsistentClusters)
		// fix inconsistent clusters
		fixedClustersCount := b.fixInconsistentClusters(ctx, inconsistentClusters)
		logger.Info("fix inconsistent clusters done", zap.Int("inconsistent_total", len(inconsistentClusters)),
			zap.Int("success", fixedClustersCount), zap.Int("failed", len(inconsistentClusters)-fixedClustersCount))
		fixedTotal += fixedClustersCount
		// break when there are no more clusters or all yesterday clusters are checked
		if allChecked || resp.Total <= pageSize*page {
			break
		}
		page++
	}
	logger.Info(
		"check backup schedule done",
		zap.Int("target_total", targetTotal),
		zap.Int("inconsistent_total", inconsistentTotal),
		zap.Int("fixed_total", fixedTotal),
		zap.Int("unfixed_total", inconsistentTotal-fixedTotal),
	)
	return nil
}

// repairBackupScheduleSnapshotFields repairs the metadata required to restore
// backups after their source cluster has been deleted. It is intentionally
// bounded and does not create user_backups or advance next_schedule_time.
func (b *backupWorker) repairBackupScheduleSnapshotFields(ctx context.Context) error {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	schedules, err := b.backupSchedule.List(ctx, entity.ListBackupScheduleOptions{
		IncludeDeleted:              true,
		IncompleteSnapshot:          true,
		IncompleteSnapshotBatchSize: backupScheduleSnapshotRepairBatchSize,
	})
	if err != nil {
		logger.Error("list incomplete backup schedule snapshots failed", zap.Error(err))
		return err
	}
	if len(schedules) == 0 {
		return nil
	}

	failedCount := 0
	for _, schedule := range schedules {
		if err := b.repairBackupScheduleSnapshot(ctx, schedule); err != nil {
			failedCount++
			logger.Error(
				"repair backup schedule snapshot failed",
				zap.String("cluster_id", schedule.ClusterID),
				zap.Int64("schedule_id", schedule.ID),
				zap.Error(err),
			)
		}
	}
	if failedCount > 0 {
		return fmt.Errorf("repair backup schedule snapshots failed, failed count: %d", failedCount)
	}
	return nil
}

func (b *backupWorker) repairBackupScheduleSnapshot(
	ctx context.Context,
	schedule *entity.BackupSchedule,
) error {
	if schedule == nil {
		return nil
	}
	logger := utils.GetWorkerLoggerFromContext(ctx)
	clusterID, err := strconv.ParseUint(schedule.ClusterID, 10, 64)
	if err != nil {
		return err
	}
	cluster, bizErr := b.mgmtSvc.GetClusterIncludeDeleted(ctx, clusterID)
	if bizErr != nil {
		if bizErr.ErrCode() == bizerr.ErrCodeRecordNotFound {
			logger.Warn(
				"skip backup schedule snapshot repair because cluster is not found",
				zap.String("cluster_id", schedule.ClusterID),
				zap.Int64("schedule_id", schedule.ID),
			)
			return nil
		}
		return bizErr
	}
	serverlessCluster := cluster.GetServerlessCluster().GetCluster()
	if serverlessCluster == nil {
		return errors.New("serverless cluster is nil")
	}

	changed := false
	if schedule.SourceToken == "" {
		sourceToken := serverlessCluster.GetUserPrefix()
		if sourceToken == "" {
			logger.Warn(
				"skip backup schedule snapshot repair because source token is empty",
				zap.String("cluster_id", schedule.ClusterID),
				zap.Int64("schedule_id", schedule.ID),
			)
			return nil
		}
		schedule.SourceToken = sourceToken
		changed = true
	}
	if schedule.EnhancedEncryptionEnabled == nil {
		enhancedEncryptionEnabled := serverlessCluster.GetEnhancedEncryptionEnabled()
		schedule.EnhancedEncryptionEnabled = &enhancedEncryptionEnabled
		changed = true
	}
	if schedule.ZoneRedundantEnabled == nil {
		zoneRedundantEnabled := serverlessCluster.GetZoneRedundantEnabled()
		schedule.ZoneRedundantEnabled = &zoneRedundantEnabled
		changed = true
	}
	if schedule.ServicePlan == "" {
		servicePlan, planErr := GetClusterPlan(ctx, cluster, b.billingFacade)
		if planErr != nil {
			return planErr
		}
		schedule.ServicePlan = servicePlan
		changed = true
	}
	if !changed {
		return nil
	}
	_, saveErr := b.backupSchedule.Save(ctx, schedule)
	return saveErr
}

func (b *backupWorker) getOrSaveBackupScheduleCheckJob(
	ctx context.Context,
	date string,
) (*entity.BackupScheduleCheckJob, bizerr.BizError) {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	checkJob, err := b.backupScheduleCheckJobRepo.Get(ctx, date)
	if err != nil {
		logger.Error("get backup schedule check job failed", zap.String("date", date), zap.Error(err))
		return nil, err
	}
	if checkJob == nil {
		checkJob = &entity.BackupScheduleCheckJob{
			CheckDate:    date,
			ExecuteTimes: 1,
			LogID:        logid.GetLogid(ctx),
		}
		err = b.backupScheduleCheckJobRepo.Create(ctx, checkJob)
		if err != nil {
			logger.Error("create backup schedule check job failed", zap.String("date", date), zap.Error(err))
			return nil, err
		}
	} else if checkJob.State != constant.BackupScheduleCheckJobStateSucceeded {
		checkJob.ExecuteTimes++
		checkJob.LogID = logid.GetLogid(ctx)
		_, err = b.backupScheduleCheckJobRepo.Save(ctx, checkJob)
		if err != nil {
			logger.Error("update backup schedule check job failed", zap.String("date", date), zap.Error(err))
			return nil, err
		}
	}
	return checkJob, nil
}

func listTargetClusters(targetDate string, clusters []*tidbmgmtservicev1.Cluster) ([]*tidbmgmtservicev1.Cluster, bool) {
	allChecked := false
	targetClusters := make([]*tidbmgmtservicev1.Cluster, 0)
	// clusters from mgmt will select from the database order by create time desc first,
	// then return the whole page order by create time asc, so we need to check every page from the last one
	for i := len(clusters) - 1; i >= 0; i-- {
		cluster := clusters[i]
		createDate := cluster.CreateTime.AsTime().UTC().Format(time.DateOnly)
		if createDate > targetDate {
			continue
		}
		if createDate == targetDate {
			targetClusters = append(targetClusters, cluster)
		}
		if createDate < targetDate {
			allChecked = true
			break
		}
	}
	return targetClusters, allChecked
}

func (b *backupWorker) checkInconsistentClusters(
	ctx context.Context,
	clusters []*tidbmgmtservicev1.Cluster,
) ([]*tidbmgmtservicev1.Cluster, bizerr.BizError) {
	inconsistentClusters := make([]*tidbmgmtservicev1.Cluster, 0)
	if len(clusters) == 0 {
		return inconsistentClusters, nil
	}
	// get backup schedules
	clusterIds := make([]string, 0, len(clusters))
	for _, cluster := range clusters {
		clusterIds = append(clusterIds, strconv.FormatUint(cluster.ClusterId, 10))
	}
	schedules, err := b.backupSchedule.List(ctx, entity.ListBackupScheduleOptions{
		ClustersIn: clusterIds,
	})
	if err != nil {
		return nil, err
	}
	// check backup schedules
	schedulesMap := make(map[string]*entity.BackupSchedule)
	for _, schedule := range schedules {
		schedulesMap[schedule.ClusterID] = schedule
	}
	for _, cluster := range clusters {
		if _, ok := schedulesMap[strconv.FormatUint(cluster.ClusterId, 10)]; !ok {
			inconsistentClusters = append(inconsistentClusters, cluster)
			continue
		}
	}
	return inconsistentClusters, nil
}

func (b *backupWorker) fixInconsistentClusters(ctx context.Context, clusters []*tidbmgmtservicev1.Cluster) int {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	backupSchedules := make([]*entity.BackupSchedule, 0, len(clusters))
	curTime := time.Now()
	logger.Info("try to fix the following clusters", zap.Any("clusters", clusters))
	for _, cluster := range clusters {
		basicCluster, err := b.mgmtSvc.GetCluster(
			ctx,
			cluster.ClusterId,
			tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC,
		)
		if err != nil {
			logger.Error("get cluster failed", zap.Uint64("cluster_id", cluster.ClusterId), zap.Error(err))
			continue
		}
		// skip creating cluster as the backup schedule will be created when receive cluster created event.
		if basicCluster.GetServerlessCluster().
			GetUserStatus() ==
			globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_CREATING {
			logger.Info("cluster is still creating, skip", zap.Uint64("cluster_id", cluster.ClusterId))
			continue
		}
		clusterPlan, err := GetClusterPlan(ctx, cluster, b.billingFacade)
		if err != nil {
			logger.Error("get cluster plan failed", zap.Uint64("cluster_id", cluster.ClusterId), zap.Error(err))
			continue
		}
		retention := GetDefaultBackupRetentionByClusterPlan(clusterPlan)
		scheduleCron := GetDefaultBackupScheduleCron(cluster.Region)
		nextScheduleTime, scheduleErr := calculateNextScheduleTime(curTime, scheduleCron)
		if scheduleErr != nil {
			logger.Error("calculate next schedule time failed", zap.Error(scheduleErr))
			continue
		}
		enhancedEncryptionEnabled := basicCluster.GetServerlessCluster().GetCluster().GetEnhancedEncryptionEnabled()
		zoneRedundantEnabled := basicCluster.GetServerlessCluster().GetCluster().GetZoneRedundantEnabled()

		backupSchedule := &entity.BackupSchedule{
			TenantID:                  strconv.FormatUint(cluster.OrganizationId, 10),
			ClusterID:                 strconv.FormatUint(cluster.ClusterId, 10),
			Region:                    cluster.Region,
			Retention:                 int(retention.Hours()),
			ScheduleCron:              scheduleCron,
			NextScheduleTime:          &nextScheduleTime,
			Kubernetes:                basicCluster.GetServerlessCluster().GetCluster().GetKubernetes(),
			CloudProvider:             basicCluster.GetCloudProvider(),
			SourceToken:               basicCluster.GetServerlessCluster().GetCluster().GetUserPrefix(),
			ServicePlan:               clusterPlan,
			EnhancedEncryptionEnabled: &enhancedEncryptionEnabled,
			ZoneRedundantEnabled:      &zoneRedundantEnabled,
		}
		backupSchedules = append(backupSchedules, backupSchedule)
	}
	if len(backupSchedules) > 0 {
		err := b.backupSchedule.BatchAdd(ctx, backupSchedules)
		if err != nil {
			logger.Error("batch add backup schedules failed", zap.Error(err))
			return 0
		}
	}
	return len(backupSchedules)
}