package service

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"regexp"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/lithammer/shortuuid/v3"
	"github.com/tidbcloud/eventbus-datastruct/event"
	"github.com/tidbcloud/pkgv2/bizerr"
	pkgv2 "github.com/tidbcloud/pkgv2/clusterproxy/serverless"
	"github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/pkgv2/log"
	"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/domain/vo"
	"github.com/tidbcloud/serverless-service/internal/errorcode"
	"github.com/tidbcloud/serverless-service/internal/infra/eventbus"
	"github.com/tidbcloud/serverless-service/internal/metric"
	"github.com/tidbcloud/serverless-service/internal/utils"
	event_constant "github.com/tidbcloud/serverless-service/pkg/event"
	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/zap"
	"google.golang.org/protobuf/proto"
)

type backupRestoreService struct {
	backupRepo             repo.UserBackupRepo
	backupHistoryRepo      repo.BackupScheduleHistoryRepo
	backupTombstoneRepo    repo.BackupTombstoneRepo
	physicalBackupSlotRepo repo.PhysicalBackupSlotRepo
	mgmtSvc                facade.MgmtSvcFacade
	clusterProxyProvider   facade.ClusterProxyProviderFacade
	eventbusService        *eventbus.EventbusService
	restoreJobRepo         repo.RestoreJobRepo
	dbProvider             db.TransProvider
	backupScheduleRepo     repo.BackupScheduleRepo
	accountSvc             facade.AccountFacade
	billingFacade          facade.BillingFacade
	o11yFacade             facade.O11yFacade
}

func NewBackupRestoreService(backupRepo repo.UserBackupRepo, mgmtSvc facade.MgmtSvcFacade,
	clusterProxyProvider facade.ClusterProxyProviderFacade, o11yFacade facade.O11yFacade,
	eventbusService *eventbus.EventbusService, restoreJobRepo repo.RestoreJobRepo, dbProvider db.TransProvider,
	backupScheduleRepo repo.BackupScheduleRepo, backupHistoryRepo repo.BackupScheduleHistoryRepo,
	backupTombstoneRepo repo.BackupTombstoneRepo, accountSvc facade.AccountFacade, billingFacade facade.BillingFacade,
	physicalBackupSlotRepo repo.PhysicalBackupSlotRepo,
) BackupRestoreService {
	return &backupRestoreService{
		backupRepo:             backupRepo,
		backupHistoryRepo:      backupHistoryRepo,
		backupTombstoneRepo:    backupTombstoneRepo,
		physicalBackupSlotRepo: physicalBackupSlotRepo,
		mgmtSvc:                mgmtSvc,
		clusterProxyProvider:   clusterProxyProvider,
		eventbusService:        eventbusService,
		restoreJobRepo:         restoreJobRepo,
		dbProvider:             dbProvider,
		backupScheduleRepo:     backupScheduleRepo,
		accountSvc:             accountSvc,
		billingFacade:          billingFacade,
		o11yFacade:             o11yFacade,
	}
}

func (b *backupRestoreService) ListBackups(
	ctx context.Context,
	opt *vo.ListUserBackupsOpt,
) ([]*vo.Backup, uint32, bizerr.BizError) {
	if opt == nil {
		return nil, 0, bizerr.New(bizerr.ErrCodeUnknown, "opt is required")
	}
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", opt.ClusterId))
	curTime := time.Now()
	options := entity.ListUserBackupOptions{
		ClusterId:     opt.ClusterId,
		ExpireTimeGte: &curTime,
	}
	backups, count, err := b.listAvailableBackups(
		ctx,
		&options,
		curTime,
		opt.OrderBy,
		opt.Offset,
		opt.PageSize,
	)
	if err != nil {
		logger.Error("list backups failed", zap.Error(err))
		return nil, 0, err
	}
	return backups, uint32(count), nil
}

func (b *backupRestoreService) GetBackup(
	ctx context.Context,
	_, clusterId, backupId string,
) (*vo.Backup, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("clusterId", clusterId), zap.String("backupId", backupId))
	resolved, err := b.resolveBackup(ctx, clusterId, backupId, backupPhysical)
	if err != nil {
		logger.Error("resolve backup failed", zap.Error(err))
		return nil, err
	}
	return vo.BackupFromPersisted(resolved.Backup), nil
}

func (b *backupRestoreService) ListRecycledInstances(
	ctx context.Context,
	opt *vo.ListRecycledInstancesOpt,
) ([]*vo.RecycledInstance, uint32, bizerr.BizError) {
	if opt == nil || opt.TenantId == "" {
		return nil, 0, bizerr.New(bizerr.ErrCodeParams, "missing required params")
	}
	if opt.StartTime != nil && opt.EndTime != nil && opt.StartTime.After(*opt.EndTime) {
		return nil, 0, bizerr.New(bizerr.ErrCodeParams, "start time must be earlier than end time")
	}
	logger := log.WithContextL(ctx).With(zap.String("tenant_id", opt.TenantId))
	curTime := time.Now()
	regionPairs := parseRegionPairs(opt.RegionIds)
	if len(opt.RegionIds) > 0 && len(regionPairs) == 0 {
		return []*vo.RecycledInstance{}, 0, nil
	}
	persistedInstances, err := b.backupRepo.ListDeletedClustersWithBackup(ctx, &entity.ListRecycledInstanceOptions{
		TenantId:      opt.TenantId,
		ExpireTimeGte: &curTime,
		RegionPairs:   regionPairs,
	})
	if err != nil {
		logger.Error("list persisted recycled instances failed", zap.Error(err))
		return nil, 0, err
	}
	histories, err := b.backupHistoryRepo.List(ctx, &entity.ListBackupScheduleHistoryOptions{
		TenantId:    opt.TenantId,
		DeletedOnly: true,
		RegionPairs: regionPairs,
	})
	if err != nil {
		logger.Error("list recycled instances failed", zap.Error(err))
		return nil, 0, err
	}
	tenantIDUint64, parseErr := strconv.ParseUint(opt.TenantId, 10, 64)
	if parseErr != nil {
		return nil, 0, bizerr.New(bizerr.ErrCodeParams, "invalid tenant id")
	}
	clusterIDs := mergeClusterIDs(
		clusterIDsFromRecycledInstances(persistedInstances),
		clusterIDsFromHistories(histories, regionPairs),
	)
	deletedClusters, err := b.listDeletedClustersByClusterIDs(ctx, tenantIDUint64, clusterIDs)
	if err != nil {
		logger.Error("list deleted clusters failed", zap.Error(err))
		return nil, 0, err
	}
	virtualInstances, err := b.buildRecycledInstancesFromHistories(
		ctx,
		histories,
		regionPairs,
		curTime,
	)
	if err != nil {
		logger.Error("build recycled instances failed", zap.Error(err))
		return nil, 0, err
	}
	instances := mergeRecycledInstances(persistedInstances, virtualInstances)
	if len(instances) == 0 {
		return []*vo.RecycledInstance{}, 0, nil
	}
	filtered := make([]*entity.RecycledInstance, 0, len(instances))
	for _, item := range instances {
		cluster, ok := deletedClusters[item.ClusterID]
		if !ok {
			continue
		}
		if !matchDeletedClusterFilters(cluster, opt) {
			continue
		}
		fillRecycledInstanceFromCluster(item, cluster)
		filtered = append(filtered, item)
	}
	sortRecycledInstancesByDeleteTime(filtered, opt.OrderBy)
	total := uint32(len(filtered))
	offset := int(opt.Offset)
	if offset > len(filtered) {
		offset = len(filtered)
	}
	limit := len(filtered)
	if opt.PageSize > 0 {
		limit = offset + int(opt.PageSize)
		if limit > len(filtered) {
			limit = len(filtered)
		}
	}
	filtered = filtered[offset:limit]

	recycledInstances := make([]*vo.RecycledInstance, 0, len(filtered))
	for _, item := range filtered {
		recycledInstances = append(recycledInstances, convertRecycledInstanceToVO(item))
	}
	return recycledInstances, total, nil
}

func matchDeletedClusterFilters(cluster *tidbmgmtservicev1.Cluster, opt *vo.ListRecycledInstancesOpt) bool {
	if cluster == nil || opt == nil {
		return false
	}
	if len(opt.ProjectIds) > 0 &&
		!containsString(opt.ProjectIds, strconv.FormatUint(cluster.GetProjectId(), 10)) {
		return false
	}
	deleteTime := cluster.GetDeleteTime()
	if deleteTime == nil {
		return opt.StartTime == nil && opt.EndTime == nil
	}
	deleteAt := deleteTime.AsTime()
	if opt.StartTime != nil && deleteAt.Before(*opt.StartTime) {
		return false
	}
	if opt.EndTime != nil && deleteAt.After(*opt.EndTime) {
		return false
	}
	return true
}

func parseRegionPairs(regionIDs []string) []entity.ProviderRegion {
	pairs := make([]entity.ProviderRegion, 0, len(regionIDs))
	for _, regionID := range regionIDs {
		provider, region, ok := splitRegionID(regionID)
		if !ok {
			continue
		}
		pairs = append(pairs, entity.ProviderRegion{
			Provider: provider,
			Region:   region,
		})
	}
	return pairs
}

func splitRegionID(regionID string) (provider, region string, ok bool) {
	for _, candidate := range []string{config.ProviderAWS, config.ProviderGCP, config.ProviderAliCloud, config.ProviderAzure} {
		prefix := candidate + "-"
		if strings.HasPrefix(regionID, prefix) {
			region = strings.TrimPrefix(regionID, prefix)
			if region == "" {
				return "", "", false
			}
			return candidate, region, true
		}
	}
	return "", "", false
}

func containsString(values []string, target string) bool {
	for _, value := range values {
		if value == target {
			return true
		}
	}
	return false
}

func (b *backupRestoreService) GetRecycledInstance(
	ctx context.Context,
	tenantId, clusterId string,
) (*vo.RecycledInstance, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("tenant_id", tenantId), zap.String("cluster_id", clusterId))
	if tenantId == "" || clusterId == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "missing required params")
	}
	curTime := time.Now()
	persistedInstances, err := b.backupRepo.ListDeletedClustersWithBackup(ctx, &entity.ListRecycledInstanceOptions{
		TenantId:      tenantId,
		ClusterId:     clusterId,
		ExpireTimeGte: &curTime,
	})
	if err != nil {
		logger.Error("get persisted recycled instance failed", zap.Error(err))
		return nil, err
	}
	histories, err := b.backupHistoryRepo.List(ctx, &entity.ListBackupScheduleHistoryOptions{
		TenantId:    tenantId,
		ClusterId:   clusterId,
		DeletedOnly: true,
	})
	if err != nil {
		logger.Error("get recycled instance failed", zap.Error(err))
		return nil, err
	}
	cluster, err := b.getDeletedClusterMeta(ctx, clusterId)
	if err != nil {
		return nil, err
	}
	virtualInstances, err := b.buildRecycledInstancesFromHistories(
		ctx,
		histories,
		nil,
		curTime,
	)
	if err != nil {
		logger.Error("build recycled instance failed", zap.Error(err))
		return nil, err
	}
	instances := mergeRecycledInstances(persistedInstances, virtualInstances)
	if len(instances) == 0 {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "")
	}
	fillRecycledInstanceFromCluster(instances[0], cluster)
	return convertRecycledInstanceToVO(instances[0]), nil
}

func clusterIDsFromRecycledInstances(instances []*entity.RecycledInstance) []string {
	clusterIDs := make([]string, 0, len(instances))
	for _, instance := range instances {
		if instance == nil || instance.ClusterID == "" {
			continue
		}
		clusterIDs = append(clusterIDs, instance.ClusterID)
	}
	return clusterIDs
}

func mergeClusterIDs(groups ...[]string) []string {
	seen := make(map[string]struct{})
	clusterIDs := make([]string, 0)
	for _, group := range groups {
		for _, clusterID := range group {
			if clusterID == "" {
				continue
			}
			if _, ok := seen[clusterID]; ok {
				continue
			}
			seen[clusterID] = struct{}{}
			clusterIDs = append(clusterIDs, clusterID)
		}
	}
	return clusterIDs
}

func mergeRecycledInstances(
	persistedInstances, virtualInstances []*entity.RecycledInstance,
) []*entity.RecycledInstance {
	instancesByCluster := make(map[string]*entity.RecycledInstance)
	clusterIDs := make([]string, 0, len(persistedInstances)+len(virtualInstances))
	for _, group := range [][]*entity.RecycledInstance{persistedInstances, virtualInstances} {
		for _, instance := range group {
			if instance == nil || instance.ClusterID == "" {
				continue
			}
			if _, ok := instancesByCluster[instance.ClusterID]; !ok {
				clusterIDs = append(clusterIDs, instance.ClusterID)
			}
			// When both sources describe the same cluster, the virtual instance
			// (history-derived snapshot) is preferred for EnhancedEncryptionEnabled
			// and ZoneRedundantEnabled. fillRecycledInstanceFromCluster later
			// overwrites DisplayName, ProjectID, Region, Provider and DeletedAt
			// from cluster metadata. Persisted-only clusters are retained when no
			// matching virtual instance exists.
			instancesByCluster[instance.ClusterID] = instance
		}
	}
	instances := make([]*entity.RecycledInstance, 0, len(clusterIDs))
	for _, clusterID := range clusterIDs {
		instances = append(instances, instancesByCluster[clusterID])
	}
	return instances
}

func (b *backupRestoreService) ListRecycledInstanceBackups(
	ctx context.Context,
	opt *vo.ListUserBackupsOpt,
) ([]*vo.Backup, uint32, bizerr.BizError) {
	if opt == nil || opt.TenantId == "" || opt.ClusterId == "" {
		return nil, 0, bizerr.New(bizerr.ErrCodeParams, "missing required params")
	}
	logger := log.WithContextL(ctx).
		With(zap.String("tenant_id", opt.TenantId), zap.String("cluster_id", opt.ClusterId))
	curTime := time.Now()
	options := entity.ListUserBackupOptions{
		TenantId:      opt.TenantId,
		ClusterId:     opt.ClusterId,
		ExpireTimeGte: &curTime,
		States:        []constant.BackupState{constant.BackupStateSucceeded},
	}
	if _, err := b.getDeletedClusterMeta(ctx, opt.ClusterId); err != nil {
		if err.ErrCode() == bizerr.ErrCodeRecordNotFound {
			return []*vo.Backup{}, 0, nil
		}
		logger.Error("get deleted cluster failed", zap.Error(err))
		return nil, 0, err
	}
	backups, count, err := b.listAvailableBackups(
		ctx,
		&options,
		curTime,
		opt.OrderBy,
		opt.Offset,
		opt.PageSize,
	)
	if err != nil {
		logger.Error("list recycled instance backups failed", zap.Error(err))
		return nil, 0, err
	}
	return backups, uint32(count), nil
}

func (b *backupRestoreService) DeleteRecycledInstanceBackup(
	ctx context.Context,
	tenantId, clusterId, backupId string,
) bizerr.BizError {
	logger := log.WithContextL(ctx).With(
		zap.String("tenant_id", tenantId),
		zap.String("cluster_id", clusterId),
		zap.String("backup_id", backupId),
	)
	resolved, err := b.resolveBackupForDelete(ctx, clusterId, backupId)
	if err != nil {
		logger.Error("resolve recycled instance backup failed", zap.Error(err))
		return err
	}
	if resolved.hasVirtualRestorePoint() {
		backup := resolved.Virtual
		if backup.TenantID != tenantId {
			return bizerr.New(bizerr.ErrCodeRecordNotFound, "")
		}
		if _, err := b.getDeletedClusterMeta(ctx, backup.ClusterID); err != nil {
			if err.ErrCode() == bizerr.ErrCodeRecordNotFound {
				return bizerr.New(bizerr.ErrCodeRecordNotFound, "")
			}
			logger.Error("get deleted cluster failed", zap.Error(err))
			return err
		}
		deleteTime := time.Now()
		tombstone := &entity.BackupTombstone{
			TenantID:     tenantId,
			ClusterID:    backup.ClusterID,
			BackupID:     backup.ID,
			ScheduleTime: ptrTime(backup.ScheduleTime),
			DeletedAt:    &deleteTime,
			ExpireTime:   ptrTime(backup.ExpireTime),
		}
		if err := b.dbProvider.Transaction(ctx, func(txCtx context.Context) error {
			return b.backupTombstoneRepo.Create(txCtx, tombstone)
		}); err != nil {
			logger.Error("create backup tombstone failed", zap.Error(err))
			return utils.ToBizError(err)
		}
		b.publishDeleteBackupEvent(ctx, backup.ClusterID, backup.ID)
		return nil
	}
	backup := resolved.Backup
	if backup.State != constant.BackupStateSucceeded {
		return bizerr.New(bizerr.ErrCodeRecordNotFound, "")
	}
	if _, err := b.getDeletedClusterMeta(ctx, backup.ClusterID); err != nil {
		if err.ErrCode() == bizerr.ErrCodeRecordNotFound {
			return bizerr.New(bizerr.ErrCodeRecordNotFound, "")
		}
		logger.Error("get deleted cluster failed", zap.Error(err))
		return err
	}
	if clusterId != "" && backup.ClusterID != clusterId {
		return bizerr.New(bizerr.ErrCodeRecordNotFound, "")
	}
	deletePersisted := func(txCtx context.Context) error {
		if backup.ScheduleTime != nil {
			deleteTime := time.Now()
			tombstone := &entity.BackupTombstone{
				TenantID:     tenantId,
				ClusterID:    backup.ClusterID,
				BackupID:     generateVirtualBackupID(backup.ClusterID, *backup.ScheduleTime),
				ScheduleTime: ptrTime(backup.ScheduleTime.UTC()),
				DeletedAt:    &deleteTime,
				ExpireTime:   backup.ExpireTime,
			}
			if err := b.backupTombstoneRepo.Create(txCtx, tombstone); err != nil {
				return err
			}
		}
		return b.backupRepo.DeleteByBackupId(txCtx, backupId)
	}
	if err := b.dbProvider.Transaction(ctx, deletePersisted); err != nil {
		logger.Error("delete recycled instance backup failed", zap.Error(err))
		return utils.ToBizError(err)
	}
	b.publishDeleteBackupEvent(ctx, backup.ClusterID, backupId)
	return nil
}

func sortRecycledInstancesByDeleteTime(instances []*entity.RecycledInstance, orderBy string) {
	desc := !strings.Contains(strings.ToLower(orderBy), "asc")
	sort.SliceStable(instances, func(i, j int) bool {
		iTime := time.Time{}
		jTime := time.Time{}
		if instances[i] != nil && instances[i].DeletedAt != nil {
			iTime = instances[i].DeletedAt.UTC()
		}
		if instances[j] != nil && instances[j].DeletedAt != nil {
			jTime = instances[j].DeletedAt.UTC()
		}
		if iTime.Equal(jTime) {
			iClusterID := ""
			jClusterID := ""
			if instances[i] != nil {
				iClusterID = instances[i].ClusterID
			}
			if instances[j] != nil {
				jClusterID = instances[j].ClusterID
			}
			return iClusterID > jClusterID
		}
		if desc {
			return iTime.After(jTime)
		}
		return iTime.Before(jTime)
	})
}

func (b *backupRestoreService) listDeletedClustersByClusterIDs(
	ctx context.Context,
	tenantID uint64,
	clusterIDs []string,
) (map[string]*tidbmgmtservicev1.Cluster, bizerr.BizError) {
	clusterIDList := make([]uint64, 0, len(clusterIDs))
	for _, id := range clusterIDs {
		parsed, err := strconv.ParseUint(id, 10, 64)
		if err != nil {
			continue
		}
		clusterIDList = append(clusterIDList, parsed)
	}
	if len(clusterIDList) == 0 {
		return map[string]*tidbmgmtservicev1.Cluster{}, nil
	}

	result := make(map[string]*tidbmgmtservicev1.Cluster, len(clusterIDList))
	for start := 0; start < len(clusterIDList); start += deletedClusterQueryBatchSize {
		end := start + deletedClusterQueryBatchSize
		if end > len(clusterIDList) {
			end = len(clusterIDList)
		}
		req := &tidbmgmtservicev1.ListClustersRequest{
			PageSize: uint32(end - start),
			Page:     1,
			Filter: &tidbmgmtservicev1.ClusterFilter{
				ClusterIds:  clusterIDList[start:end],
				TenantIds:   []uint64{tenantID},
				View:        tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_LOCAL,
				ClusterType: tidbmgmtservicev1.DeployType_DEPLOY_TYPE_SERVERLESS.Enum(),
				Deleted:     proto.Bool(true),
			},
		}
		resp, err := b.mgmtSvc.ListClusters(ctx, req)
		if err != nil {
			return nil, err
		}
		for _, cluster := range resp.GetClusters() {
			if cluster.GetDeleteTime() == nil {
				continue
			}
			result[strconv.FormatUint(cluster.GetClusterId(), 10)] = cluster
		}
	}
	return result, nil
}

func fillRecycledInstanceFromCluster(instance *entity.RecycledInstance, cluster *tidbmgmtservicev1.Cluster) {
	if instance == nil || cluster == nil {
		return
	}
	instance.DisplayName = cluster.GetClusterName()
	instance.ProjectID = strconv.FormatUint(cluster.GetProjectId(), 10)
	instance.Region = cluster.GetRegion()
	instance.Provider = cluster.GetCloudProvider()
	deleteTime := cluster.GetDeleteTime().AsTime().UTC()
	instance.DeletedAt = &deleteTime
}

func (b *backupRestoreService) getDeletedClusterMeta(
	ctx context.Context,
	clusterID string,
) (*tidbmgmtservicev1.Cluster, bizerr.BizError) {
	clusterIDUint64, parseErr := strconv.ParseUint(clusterID, 10, 64)
	if parseErr != nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid cluster id")
	}
	cluster, bizErr := b.mgmtSvc.GetClusterIncludeDeleted(
		ctx,
		clusterIDUint64,
	)
	if bizErr != nil {
		return nil, bizErr
	}
	if cluster.GetDeleteTime() == nil {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "")
	}
	return cluster, nil
}

func (b *backupRestoreService) Restore(
	ctx context.Context,
	restore *vo.Restore,
) (clusterId string, err bizerr.BizError) {
	logger := log.WithContextL(ctx)
	logger.Info("start to restore", zap.Any("request", *restore))
	// build restore request
	logger.Info("start to build restore request")
	var sourceCluster *tidbmgmtservicev1.Cluster
	var backup *entity.UserBackup
	var restoreRequest *pkgv2.RestoreRequest
	if restore.BackupId != nil {
		restoreRequest, sourceCluster, backup, err = b.buildRestoreRequestWithBackup(
			ctx,
			restore.ProjectId,
			*restore.BackupId,
		)
		if err != nil {
			return "", err
		}
	} else if restore.PointInTime != nil {
		restoreRequest, sourceCluster, err = b.buildRestoreRequestWithPITR(ctx, restore.ProjectId, restore.PointInTime)
		if err != nil {
			return "", err
		}
	}
	if backup != nil && sourceCluster.GetDeleteTime() != nil && restore.RestoreType != constant.RestoreTypeNew {
		return "", bizerr.New(errorcode.ErrCodeRecycleBinRestoreOnlyToNew, "")
	}
	// start to restore
	switch restore.RestoreType {
	case constant.RestoreTypeInPlace:
		// we will set custom status when restore in place
		return b.restoreInPlace(ctx, sourceCluster, backup, restoreRequest)
	case constant.RestoreTypeNew:
		// mgmt will alter underlying cluster status to recover when restore to new cluster
		return b.restoreToNewCluster(ctx, sourceCluster, backup, restoreRequest, restore.RestoreCluster)
	default:
		return "", bizerr.New(errorcode.ErrCodeUnsupportedRestoreType, "")
	}
}

func (b *backupRestoreService) restoreInPlace(ctx context.Context, sourceCluster *tidbmgmtservicev1.Cluster,
	backup *entity.UserBackup, restoreRequest *pkgv2.RestoreRequest,
) (_ string, err bizerr.BizError) {
	logger := log.WithContextL(ctx)
	targetCluster := sourceCluster
	clusterId := targetCluster.ClusterId
	restoreRequest.Token = targetCluster.GetServerlessCluster().GetCluster().UserPrefix

	restoreId := entity.GenerateIDWithPrefix(constant.RestoreIDPrefix)
	// set cluster to restore_in_place status. this rpc will return 49900003 error code when
	// 1. cluster is not in normal status
	// 2. cluster has set a mutex custom status, for example, restore_to_new, import, etc.
	var statusId uint64
	if statusId, err = b.mgmtSvc.SetClusterRestoreInPlaceStatus(ctx, clusterId, restoreId, RestoreTimeout); err != nil {
		if err.ErrCode() == CustomStatusMutexErrorCode {
			return "", err
		}
		logger.Error("set cluster restore status failed", zap.Error(err))
		return "", bizerr.New(bizerr.ErrCodeUnknown, "")
	}
	logger.Info("set cluster restore status success", zap.Uint64("statusId", statusId))

	defer func() {
		if err != nil {
			b.rollbackRestoreStatus(ctx, statusId, clusterId)
		}
	}()

	// record restore job
	sourceClusterId := strconv.FormatUint(sourceCluster.ClusterId, 10)
	restoreJob, err := b.recordRestoreJob(
		ctx,
		sourceClusterId,
		targetCluster,
		backup,
		restoreRequest,
		statusId,
		restoreId,
	)
	if err != nil {
		logger.Error("record restore job failed", zap.Error(err))
		return "", err
	}

	logger.Info("restore inplace success", zap.Uint64("target_cluster_id", clusterId))
	go b.publishRestoreEvent(ctx, sourceCluster, restoreJob, event.EventStatusStarted)
	return strconv.FormatUint(clusterId, 10), nil
}

func (b *backupRestoreService) restoreToNewCluster(
	ctx context.Context,
	sourceCluster *tidbmgmtservicev1.Cluster,
	backup *entity.UserBackup,
	restoreRequest *pkgv2.RestoreRequest,
	restoreCluster *vo.RestoreCluster,
) (_ string, err bizerr.BizError) {
	logger := log.WithContextL(ctx)
	backupID := ""
	backupState := ""
	sourceToken := ""
	if backup != nil {
		backupID = backup.ID
		backupState = string(backup.State)
	}
	if restoreRequest != nil && restoreRequest.SourceToken != nil {
		sourceToken = *restoreRequest.SourceToken
	}
	logger.Info(
		"restore to new cluster start",
		zap.Uint64("source_cluster_id", sourceCluster.GetClusterId()),
		zap.String("source_cluster_name", sourceCluster.GetClusterName()),
		zap.String("backup_id", backupID),
		zap.String("backup_state", backupState),
		zap.String("source_token", sourceToken),
	)

	sourceClusterDeleted := sourceCluster.GetDeleteTime() != nil
	// create new cluster
	param, err := b.buildRestoreNewClusterParams(ctx, sourceCluster, backup, restoreCluster)
	if err != nil {
		logger.Error(
			"build restore new cluster params failed",
			zap.Uint64("source_cluster_id", sourceCluster.GetClusterId()),
			zap.String("source_cluster_name", sourceCluster.GetClusterName()),
			zap.String("backup_id", backupID),
			zap.String("backup_state", backupState),
			zap.String("source_token", sourceToken),
			zap.Error(err),
		)
		return "", err
	}

	clusterId := sourceCluster.ClusterId
	restoreId := entity.GenerateIDWithPrefix(constant.RestoreIDPrefix)
	var statusId uint64
	// For deleted source clusters, skip setting source custom status.
	if !sourceClusterDeleted {
		// set cluster to restore_to_new status. this rpc will return 49900003 error code when
		// 1. cluster is not in normal status
		// 2. cluster has set a mutex custom status, for example, restore_inplace, import, etc.
		if statusId, err = b.mgmtSvc.SetClusterRestoreToNewStatus(ctx, clusterId, restoreId); err != nil {
			if err.ErrCode() == CustomStatusMutexErrorCode {
				return "", err
			}
			logger.Error("set cluster restore status failed", zap.Error(err))
			return "", bizerr.New(bizerr.ErrCodeUnknown, "")
		}
		logger.Info("set cluster restore status success", zap.Uint64("status_id", statusId))

		defer func() {
			if err != nil {
				b.rollbackRestoreStatus(ctx, statusId, clusterId)
			}
		}()
	} else {
		logger.Info("skip setting source cluster restore status for deleted source cluster", zap.String("backup_id", backupID))
	}

	logger.Info(
		"restore to new cluster request params",
		zap.Uint64("source_cluster_id", sourceCluster.GetClusterId()),
		zap.String("source_cluster_name", sourceCluster.GetClusterName()),
		zap.String("backup_id", backupID),
		zap.String("backup_state", backupState),
		zap.String("source_token", sourceToken),
	)

	cluster, err := b.mgmtSvc.RestoreNewCluster(ctx, param)
	if err != nil {
		logger.Error(
			"failed to create cluster when restore to new cluster",
			zap.Uint64("source_cluster_id", sourceCluster.GetClusterId()),
			zap.Any("param", param),
			zap.Error(err),
		)
		return "", err
	}
	logger.Info("create new cluster success")

	// record restore job with pending status
	sourceClusterId := strconv.FormatUint(sourceCluster.ClusterId, 10)
	restoreJob, err := b.recordRestoreJob(ctx, sourceClusterId, cluster, backup, restoreRequest, statusId, restoreId)
	if err != nil {
		logger.Error("record restore job failed", zap.Error(err))
		return "", err
	}
	logger.Info("restore to new cluster success", zap.Uint64("target_cluster_id", cluster.ClusterId))
	go b.publishRestoreEvent(ctx, sourceCluster, restoreJob, event.EventStatusStarted)
	return strconv.FormatUint(cluster.ClusterId, 10), nil
}

func (b *backupRestoreService) buildRestoreNewClusterParams(
	ctx context.Context,
	sourceCluster *tidbmgmtservicev1.Cluster,
	backup *entity.UserBackup,
	restoreCluster *vo.RestoreCluster,
) (*tidbmgmtservicev1.RestoreServerlessClusterRequest, bizerr.BizError) {
	logger := log.WithContextL(ctx)
	if sourceCluster == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "source cluster is empty")
	}
	if restoreCluster == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "restore cluster is empty")
	}

	serverlessCluster := sourceCluster.GetServerlessCluster().GetCluster()
	sourceKubernetes := serverlessCluster.GetKubernetes()
	sourceEnhancedEncryptionEnabled := serverlessCluster.GetEnhancedEncryptionEnabled()
	sourceZoneRedundantEnabled := serverlessCluster.GetZoneRedundantEnabled()
	sourceClusterPlan := ""
	sourceClusterDeleted := sourceCluster.GetDeleteTime() != nil
	if sourceClusterDeleted {
		sourceClusterID := strconv.FormatUint(sourceCluster.GetClusterId(), 10)
		if backup != nil && backup.ClusterID != "" {
			sourceClusterID = backup.ClusterID
		}
		schedules, err := b.backupScheduleRepo.List(ctx, entity.ListBackupScheduleOptions{
			ClusterId:      sourceClusterID,
			IncludeDeleted: true,
		})
		if err != nil {
			return nil, err
		}
		if len(schedules) == 0 {
			return nil, bizerr.New(
				bizerr.ErrCodeRecordNotFound,
				"backup schedule for deleted cluster %s is not found",
				sourceClusterID,
			)
		}
		schedule := schedules[0]
		if schedule.Kubernetes != "" {
			sourceKubernetes = schedule.Kubernetes
		}
		if schedule.EnhancedEncryptionEnabled != nil {
			sourceEnhancedEncryptionEnabled = *schedule.EnhancedEncryptionEnabled
		}
		if schedule.ZoneRedundantEnabled != nil {
			sourceZoneRedundantEnabled = *schedule.ZoneRedundantEnabled
		}
		if schedule.ServicePlan != "" {
			sourceClusterPlan = schedule.ServicePlan
		}
	}
	req := &tidbmgmtservicev1.RestoreServerlessClusterRequest{
		ClusterName:          restoreCluster.DisplayName,
		ProjectId:            sourceCluster.ProjectId,
		OrganizationId:       sourceCluster.OrganizationId,
		CreatedBy:            utils.BuildCreatorID(ctx),
		CloudProvider:        sourceCluster.CloudProvider,
		Region:               sourceCluster.Region,
		Kubernetes:           sourceKubernetes,
		ZoneRedundantEnabled: sourceZoneRedundantEnabled,
	}

	req.EncryptionConfig = &globalserverlessv1.EncryptionConfig{
		EnhancedEncryptionEnabled: sourceEnhancedEncryptionEnabled,
	}

	var minRcu, maxRcu, spendingLimit uint32
	if restoreCluster.AutoScaling != nil {
		minRcu = restoreCluster.AutoScaling.MinRcu
		maxRcu = restoreCluster.AutoScaling.MaxRcu
	}
	spendingLimit = uint32(restoreCluster.SpendingLimitMonthly)
	restoreClusterPlan := GetClusterPlanWithSpendingLimit(minRcu, spendingLimit)
	logger.Info(
		"start to build restore request",
		zap.Any("source cluster", serverlessCluster),
		zap.Any("restore cluster", restoreCluster),
	)

	if sourceClusterPlan == "" {
		if sourceClusterDeleted {
			return nil, bizerr.New(
				bizerr.ErrCodeRecordNotFound,
				"source service plan is empty for deleted cluster %d",
				sourceCluster.GetClusterId(),
			)
		}
		var err bizerr.BizError
		sourceClusterPlan, err = GetClusterPlan(ctx, sourceCluster, b.billingFacade)
		if err != nil {
			return nil, err
		}
	}
	sourcePlanLevel, sourcePlanOK := GetServicePlanLevel(sourceClusterPlan)
	targetPlanLevel, targetPlanOK := GetServicePlanLevel(restoreClusterPlan)
	if !sourcePlanOK || !targetPlanOK {
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"invalid service plan, source=%s, target=%s",
			sourceClusterPlan,
			restoreClusterPlan,
		)
	}
	if targetPlanLevel < sourcePlanLevel {
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"Not allowed to restore to a lower plan, source=%s, target=%s",
			sourceClusterPlan,
			restoreClusterPlan,
		)
	}

	if restoreClusterPlan == constant.SERVERLESS_PLAN_ESSENTIAL {
		req.AutoScaling = &tidbmgmtservicev1.AutoScaling{
			MinRcu: minRcu,
			MaxRcu: maxRcu,
		}
	} else {
		req.SpendingLimit = &globalserverlessv1.SpendLimit{
			Monthly: spendingLimit,
		}
	}
	logger.Info(
		"build restore request done",
		zap.Any("restore_request", req),
	)

	return req, nil
}

func (b *backupRestoreService) rollbackRestoreStatus(ctx context.Context, statusId, clusterId uint64) {
	logger := log.WithContextL(ctx)
	success := true
	defer func() {
		if !success {
			metric.RestoreRecoverFailedGauge.WithLabelValues(strconv.FormatUint(clusterId, 10),
				metric.RestoreJobRecoverStatusFail).Add(1)
		}
	}()
	if err := b.mgmtSvc.DeleteClusterCustomStatus(ctx, clusterId, statusId); err != nil {
		logger.Error("rollback cluster status failed", zap.Error(err))
		success = false
	}
}

// recordRestoreJob will record restore job with pending status
// ClusterStatusId and TargetToken will be empty when restore to new cluster
func (b *backupRestoreService) recordRestoreJob(
	ctx context.Context,
	sourceClusterId string,
	cluster *tidbmgmtservicev1.Cluster,
	backup *entity.UserBackup,
	restoreRequest *pkgv2.RestoreRequest,
	statusId uint64,
	restoreId string,
) (*entity.RestoreJob, bizerr.BizError) {
	logger := log.WithContextL(ctx)
	sourceToken := ""
	if restoreRequest.SourceToken != nil {
		sourceToken = *restoreRequest.SourceToken
	}
	restoreJob := &entity.RestoreJob{
		ID:              restoreId,
		TenantId:        cluster.OrganizationId,
		SourceClusterId: sourceClusterId,
		TargetClusterId: strconv.FormatUint(cluster.ClusterId, 10),
		SourceToken:     sourceToken,
		TargetToken:     restoreRequest.Token,
		CreatedBy:       utils.BuildCreatorID(ctx),
		Region:          cluster.Region,
		CloudProvider:   cluster.CloudProvider,
		State:           constant.RestoreJobSatePending,
		ClusterStatusId: statusId,
	}
	// backup will not be nil when restore with backup
	if backup != nil {
		restoreJob.BackupId = backup.ID
		restoreJob.BackupInfo = backup
		restoreJob.BackupTime = backup.BackupTime
	} else {
		pointInTime := time.Time(*restoreRequest.PointInTime)
		restoreJob.BackupTime = &pointInTime
	}
	logger.Info("start to create restore job", zap.Any("restore_job", restoreJob))
	err := b.restoreJobRepo.Create(ctx, restoreJob)
	if err != nil {
		return restoreJob, err
	}
	return restoreJob, nil
}

// publishRestoreEvent will publish restore event to target cluster
func (b *backupRestoreService) publishRestoreEvent(
	ctx context.Context,
	cluster *tidbmgmtservicev1.Cluster,
	restore *entity.RestoreJob,
	status event.EventStatus,
) {
	logger := log.WithContextL(ctx)
	if restore == nil || status == "" {
		logger.Error("skip publish restore event, restore or status is nil")
		return
	}
	restoreEvent, err := BuildRestoreEvent(ctx, cluster, restore, status)
	if err != nil {
		logger.Error("build restore event failed", 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.Error(err))
	}
}

func BuildRestoreEvent(
	ctx context.Context,
	cluster *tidbmgmtservicev1.Cluster,
	restore *entity.RestoreJob,
	status event.EventStatus,
) (*event.DisplayEvent, error) {
	logger := log.WithContextL(ctx)
	// build restore event
	clusterId, err := strconv.ParseUint(restore.TargetClusterId, 10, 64)
	if err != nil {
		logger.Error("parse cluster id failed", zap.Error(err))
		return nil, err
	}
	restoreEvent := &event.DisplayEvent{
		HappenedAtMs:     uint64(time.Now().UnixMilli()),
		TenantID:         restore.TenantId,
		ProjectID:        cluster.ProjectId,
		ClusterID:        clusterId,
		Author:           restore.CreatedBy,
		DisplayStatus:    status,
		DisplayEventType: event_constant.EventTypeRestore,
	}
	if status == event.EventStatusCanceled {
		restoreEvent.Author = "system"
	}
	// build summary
	msgType := event_constant.PiTRRestore
	if restore.BackupId != "" {
		msgType = event_constant.SnapshotRestore
	}
	summary := fmt.Sprintf("The %s %s.\n", string(msgType), string(status))
	summary += fmt.Sprintf("Backup Time: %s.\n", restore.BackupTime.UTC().Format(time.RFC3339))
	summary += fmt.Sprintf("Source Cluster ID: %s.\n", restore.SourceClusterId)
	if msgType == event_constant.SnapshotRestore {
		summary += fmt.Sprintf("Backup ID: %s.\n", restore.BackupId)
	}
	restoreEvent.Summary = summary
	// build detail
	detail, err := json.Marshal(restore)
	if err != nil {
		logger.Error("failed to marshal restore event", zap.Error(err))
		return nil, err
	}
	restoreEvent.Detail = string(detail)
	return restoreEvent, nil
}

func (b *backupRestoreService) publishDeleteBackupEvent(
	ctx context.Context,
	clusterID, backupID string,
) {
	publishDisplayEventAsync(
		ctx,
		b.eventbusService,
		func(ctx context.Context) *tidbmgmtservicev1.Cluster {
			return loadClusterForEvent(ctx, b.mgmtSvc, clusterID)
		},
		event_constant.EventTypeDeleteBackup,
		fmt.Sprintf("Backup %s deleted.", backupID),
		event_constant.DeleteBackupDetail{BackupID: backupID},
	)
}

func (b *backupRestoreService) buildRestoreRequestWithBackup(
	ctx context.Context,
	projectId *string,
	backupId string,
) (*pkgv2.RestoreRequest, *tidbmgmtservicev1.Cluster, *entity.UserBackup, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("backup_id", backupId))
	resolved, err := b.resolveBackup(ctx, "", backupId, backupPhysical)
	if err != nil {
		logger.Error("failed to resolve backup", zap.Error(err))
		return nil, nil, nil, err
	}
	backup := resolved.Backup
	history := resolved.History
	if backup.State != constant.BackupStateSucceeded {
		logger.Error("backup is not available", zap.String("state", string(backup.State)))
		return nil, nil, backup, bizerr.New(bizerr.ErrCodeRecordNotFound, "")
	}
	clusterId, err1 := strconv.ParseUint(backup.ClusterID, 10, 64)
	if err1 != nil {
		logger.Error("failed to parse cluster id", zap.Error(err1))
		return nil, nil, backup, bizerr.New(bizerr.ErrCodeUnknown, "")
	}
	// check expire
	if backup.ExpireTime == nil || time.Now().After(*backup.ExpireTime) {
		logger.Error("backup is expired")
		return nil, nil, backup, bizerr.New(bizerr.ErrCodeRecordNotFound, "")
	}
	if backup.BackupTime == nil {
		logger.Error("backup time is empty")
		return nil, nil, backup, bizerr.New(bizerr.ErrCodeUnknown, "")
	}

	cluster, err := b.mgmtSvc.GetClusterIncludeDeleted(
		ctx,
		clusterId,
	)
	if err != nil {
		logger.Error("failed to get source cluster", zap.Error(err))
		return nil, nil, backup, err
	}

	// check rbac, the projectId in request not matching the backup's current projectId
	if projectId != nil && *projectId != strconv.FormatUint(cluster.GetProjectId(), 10) {
		logger.Error("mismatch from project id", zap.Error(err))
		return nil, nil, backup, bizerr.New(bizerr.ErrCodeRecordNotFound, "")
	}

	sourceToken, err := b.resolveSourceToken(ctx, backup, history, cluster)
	if err != nil {
		return nil, nil, backup, err
	}

	request := &pkgv2.RestoreRequest{
		Backup: &pkgv2.BackupItem{
			ID:   backup.BackupID,
			Time: pkgv2.JSONTime(*backup.BackupTime),
			Name: backup.BackupName,
		},
		SourceToken: &sourceToken,
	}
	return request, cluster, backup, nil
}

func (b *backupRestoreService) resolveSourceToken(
	ctx context.Context,
	backup *entity.UserBackup,
	history *entity.BackupScheduleHistory,
	cluster *tidbmgmtservicev1.Cluster,
) (string, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("backup_id", backup.ID))
	if history != nil && history.SourceToken != "" {
		return history.SourceToken, nil
	}
	if cluster.GetDeleteTime() == nil {
		return cluster.GetServerlessCluster().GetCluster().GetUserPrefix(), nil
	}
	schedules, scheduleErr := b.backupScheduleRepo.List(ctx, entity.ListBackupScheduleOptions{
		ClusterId:      backup.ClusterID,
		IncludeDeleted: true,
	})
	if scheduleErr != nil {
		logger.Error("failed to list backup schedule for source token", zap.Error(scheduleErr))
		return "", scheduleErr
	}
	if len(schedules) > 0 && schedules[0].SourceToken != "" {
		return schedules[0].SourceToken, nil
	}
	logger.Error(
		"source token is empty",
		zap.String("cluster_id", backup.ClusterID),
		zap.String("backup_id", backup.ID),
		zap.Bool("is_deleted_cluster", true),
	)
	return "", bizerr.New(bizerr.ErrCodeParams, "source token is empty")
}

func (b *backupRestoreService) buildRestoreRequestWithPITR(
	ctx context.Context,
	projectId *string,
	pitr *vo.PointInTime,
) (*pkgv2.RestoreRequest, *tidbmgmtservicev1.Cluster, bizerr.BizError) {
	logger := log.WithContextL(ctx).
		With(zap.String("clusterId", pitr.ClusterId)).
		With(zap.Time("backup_time", pitr.BackupTime))

	clusterId, err1 := strconv.ParseUint(pitr.ClusterId, 10, 64)
	if err1 != nil {
		logger.Error("failed to parse cluster id", zap.Error(err1))
		return nil, nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}
	cluster, err := b.checkPitrCluster(ctx, clusterId)
	if err != nil {
		return nil, nil, err
	}
	// check rbac, it may not check for console api
	if projectId != nil && strconv.FormatUint(cluster.ProjectId, 10) != *projectId {
		logger.Error("mismatch from project id", zap.Error(err))
		return nil, nil, bizerr.New(errorcode.ErrCodeClusterNotFind, "")
	}
	// check point in time range
	retention, err := b.getBackupRetention(ctx, pitr.ClusterId)
	if err != nil {
		logger.Error("failed to get backup retention", zap.Error(err))
		return nil, nil, err
	}
	if cluster.GetServerlessCluster().CreationFinishedAt == nil {
		logger.Error("cluster creation finished time is empty, it should not happen")
		return nil, nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}
	if err := CheckPointInTimeRange(*cluster.GetServerlessCluster().CreationFinishedAt, pitr.BackupTime, retention, logger); err != nil {
		return nil, nil, err
	}
	// build restore request
	sourceToken := cluster.GetServerlessCluster().GetCluster().GetUserPrefix()
	backupTime := pkgv2.JSONTime(pitr.BackupTime)
	request := &pkgv2.RestoreRequest{
		PointInTime: &backupTime,
		SourceToken: &sourceToken,
	}
	return request, cluster, nil
}

func (b *backupRestoreService) GetBackupRetention(
	ctx context.Context, _, clusterId string,
) (time.Duration, bizerr.BizError) {
	return b.getBackupRetention(ctx, clusterId)
}

func (b *backupRestoreService) getBackupRetention(
	ctx context.Context, clusterId string,
) (time.Duration, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", clusterId))
	schedules, err := b.backupScheduleRepo.List(ctx, entity.ListBackupScheduleOptions{ClusterId: clusterId})
	if err != nil {
		logger.Error("failed to list backup schedules", zap.Error(err))
		return 0, err
	}
	if len(schedules) > 0 {
		return time.Duration(schedules[0].Retention) * time.Hour, nil
	}

	// get cluster plan
	clusterIdUint64, err1 := strconv.ParseUint(clusterId, 10, 64)
	if err1 != nil {
		logger.Error("failed to parse cluster id", zap.Error(err1))
		return 0, bizerr.New(errorcode.ErrCodeInvalidClusterID, "")
	}
	cluster, err := b.mgmtSvc.GetCluster(ctx, clusterIdUint64, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC)
	if err != nil {
		logger.Error("failed to get cluster", zap.Error(err))
		return 0, err
	}
	clusterPlan, err := GetClusterPlan(ctx, cluster, b.billingFacade)
	if err != nil {
		logger.Error("failed to get cluster plan", zap.Error(err))
		return 0, err
	}

	return GetDefaultBackupRetentionByClusterPlan(clusterPlan), nil
}

func CheckPointInTimeRange(
	creationFinished int64,
	backupTime time.Time,
	retention time.Duration,
	logger *zap.Logger,
) bizerr.BizError {
	// get start time
	startTs := creationFinished
	backupRetentionTS := time.Now().Add(-retention).Unix()
	if startTs < backupRetentionTS {
		startTs = backupRetentionTS
	}
	// get end time: set to 1 minute before, to help TiKV to implement rate limit.
	endTs := time.Now().Add(-PiTREndTimeLimit).Unix()
	if startTs >= endTs {
		logger.Error("start time is greater than end time, restore within one minute after creation is not allowed")
		return bizerr.New(errorcode.ErrCodeRestorePITRWithinOneMinute, "")
	}
	// check range
	backupTs := backupTime.Unix()
	if backupTs <= startTs || backupTs > endTs {
		logger.Error("backup time is not in range", zap.Int64("start_ts", startTs),
			zap.Int64("end_ts", endTs), zap.Int64("backup_ts", backupTs))
		errMsg := fmt.Sprintf("%s Backup time %s should be in range (%s, %s]",
			errorcode.BackupErrorMessages[errorcode.ErrCodeRestorePITROutOfRange],
			time.Unix(backupTs, 0).UTC().Format(time.RFC3339),
			time.Unix(startTs, 0).UTC().Format(time.RFC3339),
			time.Unix(endTs, 0).UTC().Format(time.RFC3339))
		return bizerr.New(errorcode.ErrCodeRestorePITROutOfRange, "%s", errMsg)
	}
	return nil
}

func (b *backupRestoreService) checkPitrCluster(
	ctx context.Context,
	clusterId uint64,
) (*tidbmgmtservicev1.Cluster, bizerr.BizError) {
	logger := log.WithContextL(ctx)
	cluster, err := b.mgmtSvc.GetClusterIncludeDedicated(
		ctx,
		clusterId,
		tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC,
	)
	if err != nil {
		logger.Error("failed to get cluster", zap.Error(err))
		return nil, err
	}
	if cluster.DeployType != tidbmgmtservicev1.DeployType_DEPLOY_TYPE_SERVERLESS {
		logger.Error("cluster is not serverless")
		return nil, bizerr.New(errorcode.ErrCodeClusterNotServerless, "")
	}
	if cluster.GetServerlessCluster().GetCluster().UserPrefix == "" {
		logger.Error("serverless cluster user prefix not find")
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}
	if cluster.GetServerlessCluster().UserStatus != globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_NORMAL {
		logger.Error("cluster is not normal")
		return nil, bizerr.New(errorcode.ErrCodeClusterNotRunning, "")
	}

	clusterPlan, err := GetClusterPlan(ctx, cluster, b.billingFacade)
	if err != nil {
		logger.Error("failed to get cluster plan", zap.Error(err))
		return nil, err
	}
	if cluster.CloudProvider == config.ProviderAliCloud && clusterPlan != constant.SERVERLESS_PLAN_ESSENTIAL {
		logger.Error("starter plan is not allowed to use pitr")
		return nil, bizerr.New(errorcode.ErrCodeRestorePITRNotAllowed, "starter plan is not allowed to use pitr")
	}
	if (cluster.CloudProvider == config.ProviderAWS || cluster.CloudProvider == config.ProviderGCP) &&
		clusterPlan == constant.SERVERLESS_PLAN_STARTER_FREE {
		logger.Error("starter plan without spending limit is not allowed to use pitr")
		return nil, bizerr.New(errorcode.ErrCodeRestorePITRNotAllowed, "")
	}
	return cluster, nil
}

func convertRecycledInstanceToVO(e *entity.RecycledInstance) *vo.RecycledInstance {
	if e == nil {
		return nil
	}
	return &vo.RecycledInstance{
		ClusterID:                 e.ClusterID,
		DisplayName:               e.DisplayName,
		ProjectID:                 e.ProjectID,
		Region:                    e.Region,
		DeleteTime:                e.DeletedAt,
		EnhancedEncryptionEnabled: e.EnhancedEncryptionEnabled,
		ZoneRedundantEnabled:      e.ZoneRedundantEnabled,
		Provider:                  e.Provider,
	}
}

func (b *backupRestoreService) ListRestores(
	ctx context.Context,
	opt *vo.ListRestoresOpt,
) ([]*vo.RestoreJob, uint32, bizerr.BizError) {
	if opt == nil || opt.ClusterId == "" {
		return nil, 0, bizerr.New(bizerr.ErrCodeParams, "missing required params")
	}
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", opt.ClusterId))
	options := entity.ListRestoreJobOptions{
		ClusterId: opt.ClusterId,
		Offset:    &opt.Offset,
		PageSize:  &opt.PageSize,
		OrderBy:   opt.OrderBy,
	}
	restoreJobs, err := b.restoreJobRepo.List(ctx, options)
	if err != nil {
		logger.Error("list restore jobs failed", zap.Error(err))
		return nil, 0, err
	}
	// count backups
	count, err := b.restoreJobRepo.Count(ctx, options)
	if err != nil {
		logger.Error("count restore jobs failed", zap.Error(err))
		return nil, 0, err
	}

	restoreJobVos := make([]*vo.RestoreJob, 0, len(restoreJobs))
	for _, restoreJob := range restoreJobs {
		restoreJobVos = append(restoreJobVos, convertRestoreJobToVO(restoreJob))
	}
	return restoreJobVos, uint32(count), nil
}

func convertRestoreJobToVO(e *entity.RestoreJob) *vo.RestoreJob {
	restore := &vo.RestoreJob{
		Id:        e.ID,
		CreatedAt: e.CreatedAt,
		TenantId:  strconv.FormatUint(e.TenantId, 10),
		ClusterId: e.TargetClusterId,
		Region:    e.Region,
		CreatedBy: e.CreatedBy,
		State:     e.State,
		Step:      e.Progress.Step,
		SizeBytes: e.SizeBytes,
		Percent:   e.Progress.Percent,
		Error:     e.Error,
	}
	if e.BackupId != "" {
		restore.BackupId = &e.BackupId
	} else {
		restore.PointInTime = &vo.PointInTime{
			BackupTime: *e.BackupTime,
			ClusterId:  e.SourceClusterId,
		}
	}
	// TiKV may return 90 percent when restore job is done, set it to 100 for better user experience
	if e.State == constant.RestoreJobStateSucceeded {
		restore.Percent = 100
	}
	return restore
}

func (b *backupRestoreService) ListBackupSchedules(
	ctx context.Context,
	opt *vo.ListBackupSchedulesOpt,
) ([]*vo.BackupSchedule, bizerr.BizError) {
	if opt == nil {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "opt is required")
	}
	logger := log.WithContextL(ctx).With(zap.String("clusterId", opt.ClusterId)).With(zap.Uint64("OrgId", opt.OrgId))
	logger.Info("list backup schedules")

	schedules, err := b.backupScheduleRepo.List(ctx, entity.ListBackupScheduleOptions{
		ClusterId: opt.ClusterId,
	})
	if err != nil {
		logger.Error("list backup schedules failed", zap.Error(err))
		return nil, err
	}
	backupScheduleVos := make([]*vo.BackupSchedule, 0, len(schedules))
	for _, schedule := range schedules {
		scheduleVo, err1 := convertBackupSchedulesToVO(schedule)
		if err1 != nil {
			logger.Error("convert BackupSchedules to vo failed", zap.Error(err1))
			return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
		}
		backupScheduleVos = append(backupScheduleVos, scheduleVo)
	}
	return backupScheduleVos, nil
}

func (b *backupRestoreService) UpdateBackupSchedules(
	ctx context.Context,
	opt *vo.UpdateBackupSchedulesOpt,
) (*vo.BackupSchedule, bizerr.BizError) {
	if opt == nil {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "opt is required")
	}
	logger := log.WithContextL(ctx).
		With(zap.String("clusterId", opt.ClusterId)).
		With(zap.String("scheduleTime", opt.ScheduleTime))
	logger.Info("update backup schedules", zap.Any("request", opt))
	// check cluster
	clusterId, err1 := strconv.ParseUint(opt.ClusterId, 10, 64)
	if err1 != nil {
		logger.Error("parse clusterId failed", zap.Error(err1))
		return nil, bizerr.New(errorcode.ErrCodeInvalidClusterID, "")
	}
	cluster, err := b.mgmtSvc.GetCluster(ctx, clusterId, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC)
	if err != nil {
		logger.Error("get cluster failed", zap.Error(err))
		return nil, err
	}
	kubernetes := cluster.GetServerlessCluster().GetCluster().GetKubernetes()
	// check cluster plan
	clusterPlan, err := GetClusterPlan(ctx, cluster, b.billingFacade)
	if err != nil {
		logger.Error("get cluster plan failed", zap.Error(err))
		return nil, err
	}
	if clusterPlan == constant.SERVERLESS_PLAN_STARTER_FREE {
		logger.Error("can not set or update backup schedule for starter plan cluster without Spending Limit")
	}
	// check retention is valid
	if opt.RetentionDays < 0 || opt.RetentionDays > constant.MaxBackupRetentionDays {
		logger.Error("invalid retention", zap.Int("retention", opt.RetentionDays))
		return nil, bizerr.New(bizerr.ErrCodeParams, "Invalid retention, max to 30 days")
	}
	// check schedule time is valid: HH:mm (every half hour), assume is in UTC time
	schedulesCron, err := generateScheduleCron(ctx, opt.ScheduleTime)
	if err != nil {
		return nil, err
	}
	nextScheduleTime, cronErr := calculateNextScheduleTime(time.Now().UTC(), schedulesCron)
	if cronErr != nil {
		logger.Error("calculate next schedule time failed", zap.Error(cronErr))
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}
	// check schedule
	schedules, err := b.backupScheduleRepo.List(ctx, entity.ListBackupScheduleOptions{ClusterId: opt.ClusterId})
	if err != nil {
		logger.Error("list backup schedules failed", zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}
	if len(schedules) > 1 {
		logger.Error("backup schedule is more than one, it should not happen now")
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}
	// create new backup schedule, cluster create also call UpdateBackupSchedules
	if len(schedules) == 0 {
		retentionHours := int(GetDefaultBackupRetentionByClusterPlan(clusterPlan).Hours())
		if opt.RetentionDays > 0 {
			retentionHours = opt.RetentionDays * 24
		}
		logger.Info("backup schedule not found, create new one")
		// mgmt normally calls this API before regional creation has selected a
		// Kubernetes cluster. Prefer the value when it is already available: this
		// endpoint can also be called later for an existing ready cluster.
		if kubernetes == "" {
			kubernetes = constant.KubernetesUnknown
		}
		enhancedEncryptionEnabled := cluster.GetServerlessCluster().GetCluster().GetEnhancedEncryptionEnabled()
		zoneRedundantEnabled := cluster.GetServerlessCluster().GetCluster().GetZoneRedundantEnabled()
		backupSchedule := &entity.BackupSchedule{
			TenantID:                  strconv.FormatUint(cluster.OrganizationId, 10),
			ClusterID:                 strconv.FormatUint(cluster.ClusterId, 10),
			Region:                    cluster.Region,
			Retention:                 retentionHours,
			ScheduleCron:              schedulesCron,
			NextScheduleTime:          &nextScheduleTime,
			Kubernetes:                kubernetes,
			CloudProvider:             cluster.CloudProvider,
			SourceToken:               cluster.GetServerlessCluster().GetCluster().GetUserPrefix(),
			ServicePlan:               clusterPlan,
			EnhancedEncryptionEnabled: &enhancedEncryptionEnabled,
			ZoneRedundantEnabled:      &zoneRedundantEnabled,
		}
		err := b.backupScheduleRepo.Create(ctx, backupSchedule)
		if err != nil && err.ErrCode() != bizerr.ErrCodeRecordExist {
			logger.Error("create backup schedule failed", zap.Error(err))
			return nil, err
		}
		if err == nil {
			resp, err1 := convertBackupSchedulesToVO(backupSchedule)
			if err1 != nil {
				logger.Error("convert BackupSchedules to vo failed", zap.Error(err1))
				return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
			}
			return resp, nil
		}
		// double check if backup schedule is created
		logger.Warn("backup schedule already exists, skip creating backup schedule and update it")
		schedules, err = b.backupScheduleRepo.List(ctx, entity.ListBackupScheduleOptions{ClusterId: opt.ClusterId})
		if err != nil {
			logger.Error("list backup schedules failed", zap.Error(err))
			return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
		}
		if len(schedules) != 1 {
			logger.Error("backup schedule wrong number", zap.Int("count", len(schedules)))
			return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
		}
	}
	// update backup schedule
	wasPending := schedules[0].IsPending()
	retentionHours := schedules[0].Retention
	if opt.RetentionDays > 0 {
		retentionHours = opt.RetentionDays * 24
	}
	historyChanged := backupScheduleHistoryConfigChanged(schedules[0], schedulesCron, retentionHours)
	previousSchedule := *schedules[0]
	schedules[0].ScheduleCron = schedulesCron
	schedules[0].NextScheduleTime = &nextScheduleTime
	if wasPending && kubernetes != "" {
		schedules[0].Kubernetes = kubernetes
	}
	schedules[0].Retention = retentionHours
	var backupSchedule *entity.BackupSchedule
	now := time.Now().UTC()
	txErr := b.dbProvider.Transaction(ctx, func(ctx context.Context) error {
		if historyChanged && !wasPending {
			if err := b.backupHistoryRepo.ArchiveCurrentSchedule(ctx, &previousSchedule, now, nil); err != nil {
				return err
			}
		}
		var saveErr bizerr.BizError
		backupSchedule, saveErr = b.backupScheduleRepo.Save(ctx, schedules[0])
		return saveErr
	})
	err = utils.ToBizError(txErr)
	if err != nil {
		logger.Error("update backup schedule failed", zap.Error(err))
		return nil, err
	}
	resp, err1 := convertBackupSchedulesToVO(backupSchedule)
	if err1 != nil {
		logger.Error("convert BackupSchedules to vo failed", zap.Error(err1))
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}
	return resp, nil
}

func backupScheduleHistoryConfigChanged(
	schedule *entity.BackupSchedule,
	scheduleCron string,
	retentionHours int,
) bool {
	return schedule == nil || schedule.ScheduleCron != scheduleCron || schedule.Retention != retentionHours
}

func calculateNextScheduleTime(curTime time.Time, scheduleCron string) (time.Time, error) {
	expr, err := constant.BackupCronParse.Parse(scheduleCron)
	if err != nil {
		return time.Time{}, err
	}
	return expr.Next(curTime.UTC()).UTC(), nil
}

func generateScheduleCron(ctx context.Context, scheduleTime string) (string, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("scheduleTime", scheduleTime))
	// scheduleTime: HH:mm
	matched, err1 := regexp.MatchString(`^([0-1][0-9]|2[0-3]):[03]0$`, scheduleTime)
	if err1 != nil {
		logger.Error("match schedule time failed", zap.Error(err1))
		return "", bizerr.New(bizerr.ErrCodeUnknown, "")
	}
	if !matched {
		logger.Error("invalid schedule time", zap.Error(err1))
		return "", bizerr.New(errorcode.ErrCodeInvalidScheduleTime, "")
	}
	// scheduleCron: 0 mm HH * * *
	hour, _ := strconv.Atoi(scheduleTime[:2])
	minute := scheduleTime[3:]
	return fmt.Sprintf("0 %s %d * * *", minute, hour), nil
}

func convertBackupSchedulesToVO(e *entity.BackupSchedule) (*vo.BackupSchedule, error) {
	backupSchedule := &vo.BackupSchedule{
		Id:        strconv.FormatInt(e.ID, 10),
		ClusterId: e.ClusterID,
		// convert to days
		Retention:    e.Retention / 24,
		ScheduleCron: e.ScheduleCron,
	}
	// ScheduleCron: 0 0 10 * * *
	crons := strings.Split(e.ScheduleCron, " ")
	if len(crons) != 6 {
		return nil, errors.New("invalid schedule cron: " + e.ScheduleCron)
	}
	minute, err := strconv.Atoi(crons[1])
	if err != nil {
		return nil, errors.New("invalid schedule cron: " + e.ScheduleCron)
	}
	hour, err := strconv.Atoi(crons[2])
	if err != nil {
		return nil, errors.New("invalid schedule cron: " + e.ScheduleCron)
	}
	backupSchedule.ScheduleTime = fmt.Sprintf("%02d:%02d", hour, minute)
	return backupSchedule, nil
}

func GetDefaultBackupRetentionByClusterPlan(clusterPlan string) time.Duration {
	switch clusterPlan {
	case constant.SERVERLESS_PLAN_STARTER_FREE:
		return constant.StarterFreePlanBackupRetention
	case constant.SERVERLESS_PLAN_STARTER_SCALABLE:
		return constant.StarterPaidPlanBackupRetention
	case constant.SERVERLESS_PLAN_ESSENTIAL:
		return constant.EssentialPlanBackupRetention
	default:
		return constant.StarterFreePlanBackupRetention
	}
}

func GetClusterPlan(ctx context.Context,
	cluster *tidbmgmtservicev1.Cluster,
	billingFacade facade.BillingFacade,
) (string, bizerr.BizError) {
	if cluster.GetServerlessCluster().GetCluster().GetAutoScaling().GetMinRcu() > 0 ||
		cluster.GetServerlessCluster().GetCluster().GetAutoScaling().GetMaxRcu() > 0 {
		return constant.SERVERLESS_PLAN_ESSENTIAL, nil
	}
	spendingLimit, err := billingFacade.GetInvoiceLimit(ctx, cluster.OrganizationId, cluster.ClusterId)
	if err != nil {
		return "", err
	}
	if spendingLimit > 0 {
		return constant.SERVERLESS_PLAN_STARTER_SCALABLE, nil
	}
	return constant.SERVERLESS_PLAN_STARTER_FREE, nil
}

func GetClusterPlanWithSpendingLimit(rcu, spendingLimit uint32) string {
	if rcu > 0 {
		return constant.SERVERLESS_PLAN_ESSENTIAL
	}
	if spendingLimit > 0 {
		return constant.SERVERLESS_PLAN_STARTER_SCALABLE
	}
	return constant.SERVERLESS_PLAN_STARTER_FREE
}

func GetServicePlanLevel(plan string) (int, bool) {
	switch plan {
	case constant.SERVERLESS_PLAN_STARTER_FREE:
		return 1, true
	case constant.SERVERLESS_PLAN_STARTER_SCALABLE:
		return 2, true
	case constant.SERVERLESS_PLAN_ESSENTIAL:
		return 3, true
	default:
		return 0, false
	}
}