//nolint:staticcheck // ST1019 TODO(ideascf): is being imported more than once
package impl

import (
	"context"
	"fmt"
	"os"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/AlekSi/pointer"
	"github.com/elliotchance/pie/v2"
	workflowEntity "github.com/shiningrush/fastflow/pkg/entity"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/app/config"
	appconfig "github.com/tidbcloud/dataflow-service/app/config"
	"github.com/tidbcloud/dataflow-service/app/internal/feature_gate"
	"github.com/tidbcloud/dataflow-service/app/internal/status_matrix"
	"github.com/tidbcloud/dataflow-service/app/internal/worflow_helper"
	"github.com/tidbcloud/dataflow-service/app/models/backup"
	"github.com/tidbcloud/dataflow-service/app/models/bucket"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/repo"
	backupRepo "github.com/tidbcloud/dataflow-service/app/repo/backup"
	crrfRepo "github.com/tidbcloud/dataflow-service/app/repo/crrf"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/utils"
	"github.com/tidbcloud/dataflow-service/app/workflow"
	internalconfig "github.com/tidbcloud/dataflow-service/internal/config"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
)

type backupSettingModel struct {
	regionModifierProxy

	dbProvider        db.TransProvider
	backupSettingRepo backupRepo.BackupSettingRepo

	backupScheduleRepo     backupRepo.BackupScheduleRepo
	bucketRepo             repo.BucketRepo
	bucketProxyModel       bucket.BucketModelV2
	BucketReplicationModel bucket.BucketReplicationModel

	tidbClusterInfraClient grpcli.TiDBClusterInfraClient
	coreCli                grpcli.CoreGrpcClient
	mgmtCli                grpcli.MgmtSvcGrpcClient
	accountSvcGrpcClient   grpcli.AccountSvcGrpcClient

	workflowClient workflow.Workflow

	statusMatrix status_matrix.StatusMatrix

	pitBackupSwitcher  backup.PitBackupSwitcher
	featureGate        feature_gate.FeatureGate
	cancelDagInsHelper worflow_helper.DagCanceler

	regionCli grpcli.RegionGrpcClient

	crrfSettingRepo crrfRepo.CRRFSettingRepo

	synchronousRTCModifier synchronousRTCModifier
}

func NewBackupSettingModel(
	dbProvider db.TransProvider,
	backupSettingRepo backupRepo.BackupSettingRepo,
	logBackupRepo backupRepo.LogBackupRepo,
	coreCli grpcli.CoreGrpcClient,
	mgmtCli grpcli.MgmtSvcGrpcClient,
	workflowClient workflow.Workflow,
	bucketProxyModel bucket.BucketModelV2,
	bucketReplicationModel bucket.BucketReplicationModel,
	bucketRepo repo.BucketRepo,
	pitBackupSwitcher backup.PitBackupSwitcher,
	featureGate feature_gate.FeatureGate,
	backupScheduleRepo backupRepo.BackupScheduleRepo,
	accountSvcGrpcClient grpcli.AccountSvcGrpcClient,
	tidbClusterInfraClient grpcli.TiDBClusterInfraClient,
	regionCli grpcli.RegionGrpcClient,
	crrfSettingRepo crrfRepo.CRRFSettingRepo,
	awsBucketManagerModel bucket.AWSBucketManagerModel,
	awsBucketConfigurationRepo repo.AWSBucketConfigurationRepo,
) backup.BackupSettingModel {
	statusMatrix := NewBackupSettingStatusMatrix()
	proxy := newRegionModifierProxy(
		dbProvider,
		backupSettingRepo,
		logBackupRepo,
		workflowClient,
		statusMatrix,
		tidbClusterInfraClient,
		pitBackupSwitcher,
		crrfSettingRepo,
	)
	ins := &backupSettingModel{
		dbProvider:             dbProvider,
		backupSettingRepo:      backupSettingRepo,
		tidbClusterInfraClient: tidbClusterInfraClient,
		coreCli:                coreCli,
		mgmtCli:                mgmtCli,
		workflowClient:         workflowClient,
		statusMatrix:           statusMatrix,
		bucketProxyModel:       bucketProxyModel,
		BucketReplicationModel: bucketReplicationModel,
		bucketRepo:             bucketRepo,
		pitBackupSwitcher:      pitBackupSwitcher,
		featureGate:            featureGate,
		backupScheduleRepo:     backupScheduleRepo,
		accountSvcGrpcClient:   accountSvcGrpcClient,
		cancelDagInsHelper: worflow_helper.NewDagCanceler(
			workflowClient,
		),
		regionCli:       regionCli,
		crrfSettingRepo: crrfSettingRepo,
	}
	ins.regionModifierProxy = *proxy
	ins.synchronousRTCModifier = synchronousRTCModifier{
		dbProvider:                 dbProvider,
		backupSettingRepo:          backupSettingRepo,
		bucketRepo:                 bucketRepo,
		awsBucketManagerModel:      awsBucketManagerModel,
		awsBucketConfigurationRepo: awsBucketConfigurationRepo,
		logBackupRepo:              logBackupRepo,
	}
	return ins
}

func (m *backupSettingModel) CreateBackupSetting(
	ctx context.Context, req *domain.CreateBackupSettingReqV2,
) bizerr.BizError {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx)
	backupSetting, err := m.queryBackupSettingIsAlreadyExisted(ctx, logger, req)
	if err != nil {
		return err
	}
	if backupSetting != nil {
		logger.Info("backup setting already existed, no need to create",
			zap.Any("backup setting", backupSetting))
		return nil
	}

	cluster, err := m.mgmtCli.GetAnyDedicatedClusterBasicByID(ctx, *req.ClusterID)
	if err != nil {
		logger.Error("get cluster failed", zap.Error(err))
		return bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}

	return m.createBackupSettingEntity(ctx, logger, req, cluster)
}

func (m *backupSettingModel) queryBackupSettingIsAlreadyExisted(
	ctx context.Context, logger *zap.Logger, req *domain.CreateBackupSettingReqV2,
) (*entity.BackupSetting, bizerr.BizError) {
	backupSetting, err := m.backupSettingRepo.Get(ctx,
		func(qs entity.BackupSettingQuerySet) entity.BackupSettingQuerySet {
			return qs.ClusterIDEq(entity.ID(*req.ClusterID)).
				TenantIDEq(entity.ID(*req.TenantID)).
				ProjectIDEq(entity.ID(*req.ProjectID))
		})
	if err == nil {
		return backupSetting, nil
	}
	if err.ErrCode() != bizerr.ErrCodeRecordNotFound {
		logger.Error("get backup setting failed", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	return nil, nil
}

func (m *backupSettingModel) createBackupSettingEntity(
	ctx context.Context, logger *zap.Logger, req *domain.CreateBackupSettingReqV2, cluster *mgmt.Cluster,
) bizerr.BizError {
	backupSetting, err := m.genBackupSettingEntity(ctx, logger, req, cluster)
	if err != nil {
		return err
	}
	var dag *workflowEntity.DagInstance
	dbErr := m.dbProvider.Transaction(ctx, func(ctx context.Context) error {
		if err := m.CreateDefaultBackupSchedule(ctx, cluster); err != nil {
			return err
		}
		if err := m.backupSettingRepo.Save(ctx, backupSetting); err != nil {
			if err.ErrCode() == bizerr.ErrCodeRecordExist {
				logger.Info("the backup setting is already created by another goroutine",
					zap.Any("backup setting", backupSetting))
				return nil
			}
			return err
		}
		if err := m.createPitBackupSwitch(ctx, backupSetting); err != nil {
			return err
		}
		dagInsVars, err := m.buildCreateBackupSettingDagVars(ctx, logger, cluster, backupSetting)
		if err != nil {
			logger.Warn("build create backup setting dag vars failed", zap.Error(err))
			return err
		}
		var dagErr error
		dag, dagErr = m.workflowClient.RunDagWithTags(
			workflow.DagConfigBucketForDefaultBackupSetting, dagInsVars,
			map[string]string{
				workflow.BusinessTypeKey:   string(workflow.BusinessTypeBackupSetting),
				workflow.BusinessActionKey: string(workflow.BusinessTypeBackupSettingCreate),
				workflow.BusinessID:        strconv.FormatUint(uint64(backupSetting.ID), 10),
				workflow.ClusterID:         strconv.FormatUint(uint64(backupSetting.ClusterID), 10),
			})
		if dagErr != nil {
			return dagErr
		}
		logger.Info("create backup setting dag started",
			zap.Any("request", req), zap.Any("dag id", dag.ID))
		return nil
	})
	if dbErr != nil {
		logger.Error("create backup setting failed", zap.Any("request", req), zap.Error(dbErr))
		return bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	logger.Info("backup setting is creating...",
		zap.Any("backup setting", backupSetting), zap.Any("dag", dag))
	return nil
}

func (m *backupSettingModel) CreateDefaultBackupSchedule(
	ctx context.Context, cluster *mgmt.Cluster,
) bizerr.BizError {
	schedules, err := m.backupScheduleRepo.List(ctx,
		func(qs entity.BackupScheduleQuerySet) entity.BackupScheduleQuerySet {
			return qs.TypeEq(entity.BackupTypeAuto).ClusterIDEq(entity.ID(cluster.GetClusterId()))
		})
	if err != nil {
		log.GetLogger(ctx).Error("list backup schedule failed", zap.Error(err))
		return bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}

	if len(schedules) > 0 {
		log.GetLogger(ctx).Info("default auto backup schedule already exists")
		return nil
	}

	schedule, err := m.getDefaultBackupScheduleCron(ctx, cluster)
	if err != nil {
		return err
	}

	creatorID, parseErr := strconv.ParseUint(cluster.CreatedBy, 10, 64)
	if parseErr != nil {
		log.GetLogger(ctx).Error("failed to parse cluster created by, use default '0'", zap.Error(parseErr))
	}
	scheduleBackup := &entity.BackupSchedule{
		Name:            fmt.Sprintf("ab-%d", cluster.GetClusterId()),
		Type:            entity.BackupTypeAuto,
		ClusterID:       entity.ID(cluster.GetClusterId()),
		BackupCycle:     entity.BackupCycleDaily,
		RetentionPeriod: internalconfig.DefaultRetentionPeriod,
		Schedule:        schedule,
		Targets:         "",
		UserID:          entity.ID(creatorID),
		Method:          entity.BackupMethodFull,
		Description:     "",
		DeletedAt:       nil,
		NextBackupTime:  nil,
		ClusterType:     entity.BackupClusterTypeDedicated,
	}

	return m.backupScheduleRepo.Create(ctx, scheduleBackup)
}

func (m *backupSettingModel) getDefaultBackupScheduleCron(
	ctx context.Context, cluster *mgmt.Cluster,
) (string, bizerr.BizError) {
	var defaultBackupTime string

	log.GetLogger(ctx).Info(
		"getDefaultBackupScheduleCron",
		zap.Uint64("tenantID", cluster.GetOrganizationId()),
		zap.String("regionName", cluster.GetRegion()),
		zap.String("cloudProviderName", cluster.GetCloudProvider()),
	)

	cloudRegion, err := m.regionCli.GetRegionByTenantID(ctx, cluster.GetOrganizationId(),
		cluster.GetCloudProvider(), cluster.GetRegion())
	if err != nil {
		return "", err
	}
	defaultBackupTime = cloudRegion.GetRegion().GetDefaultBackupTime()

	log.GetLogger(ctx).Info("the default backup time",
		zap.Any("cloudProviderName", cluster.GetCloudProvider()),
		zap.Any("regionName", cluster.GetRegion()),
		zap.Any("default backup Time", defaultBackupTime))

	t, parseErr := time.Parse(internalconfig.RegionDefaultBackupTimeFormat, defaultBackupTime)
	if parseErr != nil {
		return "", bizerr.New(bizerr.ErrCodeUnknown, "failed to parse default backup time of region")
	}
	schedule := fmt.Sprintf("0 %02d %02d 1/1 * ? *", t.UTC().Minute(), t.UTC().Hour())
	return schedule, nil
}

func (m *backupSettingModel) createPitBackupSwitch(ctx context.Context, setting *entity.BackupSetting) bizerr.BizError {
	switchReq := &domain.SwitchPitBackupReq{ClusterID: setting.ClusterID.Uint64()}
	_, err := m.pitBackupSwitcher.GetOrCreatePitBackupSwitch(ctx, switchReq)
	return err
}

func (m *backupSettingModel) genBackupSettingEntity(
	ctx context.Context, logger *zap.Logger, req *domain.CreateBackupSettingReqV2, cluster *mgmt.Cluster,
) (*entity.BackupSetting, bizerr.BizError) {
	backupSetting := &entity.BackupSetting{
		ClusterID:         entity.ID(*req.ClusterID),
		TenantID:          entity.ID(*req.TenantID),
		ProjectID:         entity.ID(*req.ProjectID),
		Status:            entity.BackupSettingStatusCreating,
		InfraProviderType: string(*req.CloudProvider),
		VolumeBackup: &entity.VolumeBackupSetting{
			Enable: false,
		},
		AutoBackup: &entity.AutoBackupSetting{
			Disable: false,
		},
		PiTR: &entity.PiTRSetting{
			Enable: false,
		},
		Bucket: &entity.BucketSetting{
			DualRegionEnable: false,
			Buckets: []entity.BackupSettingBucket{
				{
					Regions: []string{cluster.GetRegion()},
					Master:  true,
				},
			},
		},
		Action: entity.BackupSettingAction{
			CurrentActions: []string{string(entity.BackupSettingActionCreate)},
		},
	}
	err := backupSetting.RefreshBucketPrefix()
	if err != nil {
		logger.Error("refresh bucket prefix failed", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	cloudAccountID, gErr := m.getAccountID(ctx, logger, cluster)
	if gErr != nil {
		return nil, gErr
	}
	backupSetting.CloudAccountID = cloudAccountID
	return backupSetting, nil
}

func (m *backupSettingModel) getAccountID(
	ctx context.Context, logger *zap.Logger, cluster *mgmt.Cluster,
) (*string, bizerr.BizError) {
	pt := domain.CloudProviderToInfraType(cluster.GetCloudProvider())
	req := &grpcli.RegionalNetworkReq{
		ProjectID:    cluster.GetProjectId(),
		ProviderType: string(pt),
		Region:       cluster.GetRegion(),
	}
	if cluster.GetCloudProvider() == string(domain.CloudProviderTypeAzure) {
		// Use environment variable for test
		// TODO: We need to get the cloud account of a cluster from tidb-mgmt in 3-layer arch
		// Once tidb-mgmt provides such API
		accountID := os.Getenv("AZURE_DATAPLANE_SUBSCRIPTION")
		return &accountID, nil
	}
	accountID, err := m.tidbClusterInfraClient.GetCloudAccountID(ctx, req)
	if err != nil {
		logger.Error("get cloud account id failed", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	return &accountID, nil
}

func (m *backupSettingModel) buildCreateBackupSettingDagVars(ctx context.Context, logger *zap.Logger,
	cluster *mgmt.Cluster, backupSetting *entity.BackupSetting,
) (map[string]string, error) {
	dagInsVars := map[string]string{}
	dagInsVars[workflow.KeyBackupSettingId] = fmt.Sprintf("%d", backupSetting.ID)
	dagInsVars[workflow.KeyRegion] = cluster.GetRegion()

	pt := domain.CloudProviderToInfraType(cluster.GetCloudProvider())
	// on Azure, we use SAS token instead of node role for authentication
	if cluster.GetCloudProvider() == string(domain.CloudProviderTypeAzure) {
		return dagInsVars, nil
	}

	req := &grpcli.RegionalNetworkReq{
		ProjectID:    cluster.GetProjectId(),
		ProviderType: string(pt),
		Region:       cluster.GetRegion(),
	}
	nodeRole, err := m.tidbClusterInfraClient.GetNetworkNodeRole(ctx, req)
	if err != nil {
		logger.Warn("get node role error", zap.Error(err))
		return nil, err
	}
	dagInsVars[workflow.KeyNetworkNodeRole] = nodeRole

	return dagInsVars, nil
}

func (m *backupSettingModel) DeleteBackupSetting(
	ctx context.Context, req *domain.DeleteBackupSettingReqV2,
) bizerr.BizError {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx)

	backupSetting, err := m.backupSettingRepo.Get(ctx,
		func(qs entity.BackupSettingQuerySet) entity.BackupSettingQuerySet {
			return qs.ClusterIDEq(entity.ID(*req.ClusterID))
		})
	if err != nil {
		if err.ErrCode() == bizerr.ErrCodeRecordNotFound {
			return nil
		}
		logger.Error("load backup setting failed", zap.Any("request", req), zap.Error(err))
		return bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}

	if backupSetting.Status == entity.BackupSettingStatusDeleted {
		logger.Info("backup setting already deleted, no need to delete again",
			zap.Any("backup setting", backupSetting))
		return nil
	}

	isFriendly := m.statusMatrix.IsActionFriendly2Status(status_matrix.CurrentStatus{
		Status:  string(backupSetting.Status),
		Actions: backupSetting.Action.CurrentActions,
	}, string(entity.BackupSettingActionDelete))

	if !isFriendly {
		logger.Warn("delete backup setting failed, status is not friendly",
			zap.Any("backup setting", backupSetting))
		return bizerr_wrapper.NewInvalidInputError(
			appconfig.ErrActionParam, backupSetting.Status, entity.BackupSettingActionDelete)
	}

	dbErr := m.dbProvider.Transaction(ctx, func(ctx context.Context) error {
		err := m.cancelAllDagInstances(ctx, backupSetting)
		if err != nil {
			return err
		}
		err = m.deleteBucketReplication(ctx, backupSetting)
		if err != nil {
			return err
		}
		backupSetting.Status = entity.BackupSettingStatusDeleted
		backupSetting.Action = entity.BackupSettingAction{}
		err = m.backupSettingRepo.Save(ctx, backupSetting)
		if err != nil {
			return err
		}
		return nil
	})
	if dbErr != nil {
		logger.Error("delete bucket failed", zap.Any("request", req), zap.Error(dbErr))
		return bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	return nil
}

func (m *backupSettingModel) deleteBucketReplication(
	ctx context.Context, backupSetting *entity.BackupSetting,
) bizerr.BizError {
	if !m.needDeleteBucketReplication(backupSetting) {
		return nil
	}

	var sourceBucketID uint64
	var targetBucketID uint64
	for _, b := range backupSetting.Bucket.Buckets {
		if b.Master && b.BucketId != 0 {
			sourceBucketID = b.BucketId
		}
		if !b.Master && b.BucketId != 0 {
			targetBucketID = b.BucketId
		}
	}
	if sourceBucketID == 0 || targetBucketID == 0 {
		return nil
	}

	_, err := m.BucketReplicationModel.DeleteBucketReplication(ctx, &domain.DeleteBucketReplicationReq{
		SourceBucketID: &sourceBucketID,
		TargetBucketID: &targetBucketID,

		ReplicationPrefix:  backupSetting.Bucket.Prefix,
		InfraCloudProvider: pointer.To(domain.InfraProviderType(backupSetting.InfraProviderType)),
	})
	return err
}

func (m *backupSettingModel) needDeleteBucketReplication(backupSetting *entity.BackupSetting,
) bool {
	if backupSetting.InfraProviderType == string(domain.InfraProviderTypeGCP) ||
		backupSetting.InfraProviderType == string(domain.InfraProviderTypeGKE) {
		return false
	}
	if !backupSetting.Bucket.DualRegionEnable {
		return false
	}
	return true
}

func (m *backupSettingModel) cancelAllDagInstances(ctx context.Context, backupSetting *entity.BackupSetting) error {
	return m.cancelDagInsHelper.SendCancellations(ctx, map[string]string{
		workflow.BusinessTypeKey: string(workflow.BusinessTypeBackupSetting),
		workflow.BusinessID:      strconv.FormatUint(uint64(backupSetting.ID), 10),
	})
}

// QueryBackupSetting just for internal use
func (m *backupSettingModel) QueryBackupSetting(
	ctx context.Context, clusterId *uint64,
) (*domain.QueryBackupSettingRespV2, bizerr.BizError) {
	backupSetting, err := m.backupSettingRepo.Get(ctx,
		func(qs entity.BackupSettingQuerySet) entity.BackupSettingQuerySet {
			return qs.ClusterIDEq(entity.ID(*clusterId))
		})
	if err != nil {
		log.GetLogger(ctx).Warn("get backup setting failed", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	return &domain.QueryBackupSettingRespV2{
		BackupSetting: backupSetting,
	}, nil
}

func (m *backupSettingModel) QueryBackupSettingForAPI(
	ctx context.Context, clusterId *uint64,
) (*domain.QueryBackupSettingForAPIResp, bizerr.BizError) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx)
	backupSetting, err := m.backupSettingRepo.Get(ctx,
		func(qs entity.BackupSettingQuerySet) entity.BackupSettingQuerySet {
			return qs.ClusterIDEq(entity.ID(*clusterId))
		})
	if err != nil {
		log.GetLogger(ctx).Warn("get backup setting failed", zap.Error(err))
		return nil, bizerr.New(config.ErrCodeBackupSettingIsNotReady, config.ErrMsgBackupSettingIsNotReady)
	}
	pitBackupSwitch, err := m.pitBackupSwitcher.GetOrCreatePitBackupSwitch(
		ctx, &domain.SwitchPitBackupReq{ClusterID: *clusterId})
	if err != nil {
		log.GetLogger(ctx).Warn("get pit backup switch failed", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	cluster, err := m.mgmtCli.GetAnyDedicatedClusterBasicByID(ctx, uint64(backupSetting.ClusterID))
	if err != nil {
		logger.Error("get cluster failed", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	cloudRegionMap, err := m.queryCloudRegions(ctx, cluster.GetProjectId(), cluster.GetCloudProvider())
	if err != nil {
		return nil, err
	}

	pitrSetting := m.convertPiTRSettingToResp(backupSetting)
	snapshotBackupSchedule, err := m.convertAutoBackupSettingToResp(ctx, backupSetting)
	if err != nil {
		return nil, err
	}
	bucketSetting := m.convertBucketSettingToResp(backupSetting, cluster, cloudRegionMap)
	actions := m.convertBackupSettingAllowActions(backupSetting)
	crrfStatus, err := m.queryCRRFStatusForAPI(ctx, *clusterId)
	if err != nil {
		return nil, err
	}

	return &domain.QueryBackupSettingForAPIResp{
		BackupSetting:   backupSetting,
		PitBackupSwitch: pitBackupSwitch.PitBackupSwitch,
		Status:          backupSetting.Status,
		AllowActions:    actions,

		CloudProvider:          cluster.GetCloudProvider(),
		SnapshotBackupSchedule: snapshotBackupSchedule,
		BucketSetting:          bucketSetting,
		PiTRSetting:            pitrSetting,
		CRRFStatus:             crrfStatus,
	}, nil
}

func (m *backupSettingModel) queryCRRFStatusForAPI(
	ctx context.Context, clusterID uint64,
) (entity.CRRFSettingStatus, bizerr.BizError) {
	setting, err := m.crrfSettingRepo.GetByClusterID(ctx, clusterID)
	if err != nil {
		if err.ErrCode() == bizerr.ErrCodeRecordNotFound {
			return entity.CRRFSettingStatusDisabled, nil
		}
		return "", bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	if setting == nil {
		return entity.CRRFSettingStatusDisabled, nil
	}
	return setting.Status, nil
}

func (m *backupSettingModel) MarkBackupScheduleAsDeleted(
	ctx context.Context, req *domain.MarkBackupScheduleAsDeletedReq,
) bizerr.BizError {
	return m.backupScheduleRepo.SoftDeleteAuto(ctx, entity.ID(*req.ClusterId))
}

func (m *backupSettingModel) convertBackupSettingAllowActions(
	backupSetting *entity.BackupSetting,
) []entity.BackupSettingActionType {
	actions := m.statusMatrix.QueryAllowActions(status_matrix.CurrentStatus{
		Status:  string(backupSetting.Status),
		Actions: backupSetting.Action.CurrentActions,
	})
	return pie.Map(actions, func(t string) entity.BackupSettingActionType {
		return entity.BackupSettingActionType(t)
	})
}

func (m *backupSettingModel) convertBucketSettingToResp(
	backupSetting *entity.BackupSetting, cluster *mgmt.Cluster, cloudRegionMap map[string]*core.CloudRegion,
) *domain.BucketSetting {
	bucketSetting := &domain.BucketSetting{
		DualRegionEnable: false,
	}
	if backupSetting.Bucket != nil {
		bucketSetting.DualRegionEnable = backupSetting.Bucket.DualRegionEnable
		bucketSetting.Prefix = backupSetting.Bucket.Prefix
		pie.Of(backupSetting.Bucket.Buckets).Each(func(b entity.BackupSettingBucket) {
			backupSettingBucket := domain.BackupSettingBucket{
				BucketId: b.BucketId,
				Master:   b.Master,
			}
			pie.Of(b.Regions).Each(func(r string) {
				regionInfo := m.convertDataflowRegion(r, cloudRegionMap)
				backupSettingBucket.Regions = append(backupSettingBucket.Regions, *regionInfo)
			})
			bucketSetting.Buckets = append(bucketSetting.Buckets, backupSettingBucket)
		})
		return bucketSetting
	}

	regionInfo := m.convertDataflowRegion(cluster.GetRegion(), cloudRegionMap)
	bucketSetting.Buckets = []domain.BackupSettingBucket{
		{
			BucketId: 0,
			Master:   true,
			Regions:  []domain.RegionInfo{*regionInfo},
		},
	}
	return bucketSetting
}

func (m *backupSettingModel) convertPiTRSettingToResp(backupSetting *entity.BackupSetting) *domain.PiTRSetting {
	return &domain.PiTRSetting{
		Enable: backupSetting.IsPiTREnabled(),
	}
}

func (m *backupSettingModel) convertAutoBackupSettingToResp(
	ctx context.Context, backupSetting *entity.BackupSetting,
) (*domain.SnapshotBackupSchedule, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	snapshotBackupSchedule := &domain.SnapshotBackupSchedule{
		Disable: backupSetting.IsAutoBackupDisabled(),
	}
	backupSchedule, err := m.queryAutoBackupSchedule(ctx, uint64(backupSetting.ClusterID))
	if err != nil {
		return nil, err
	}
	snapshotBackupSchedule.ScheduleType = domain.BackupCycleEntity2Domain(backupSchedule.BackupCycle)
	backupTime, weekdays, pErr := utils.ParseCron(backupSchedule.Schedule)
	if pErr != nil {
		logger.Error("failed to parse backup schedule", zap.Error(pErr))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	if snapshotBackupSchedule.ScheduleType == domain.SnapshotBackupScheduleTypeDaily {
		snapshotBackupSchedule.DailyScheduleTime = backupTime
	} else {
		snapshotBackupSchedule.WeeklySchedule = &domain.WeeklySchedule{
			ScheduleTime: backupTime,
			Days:         domain.WeekDayEntityList2Domain(weekdays),
		}
	}
	snapshotBackupSchedule.RetentionPolicy = &domain.RetentionPolicy{
		RetentionPeriod: int(backupSchedule.RetentionPeriod),
	}
	if !snapshotBackupSchedule.Disable && backupSchedule.NextBackupTime != nil {
		// convert local process timezone to utc
		snapshotBackupSchedule.NextBackupTime = pointer.ToTime(backupSchedule.NextBackupTime.UTC())
	}
	return snapshotBackupSchedule, nil
}

func (m *backupSettingModel) ListBackupSetting(
	ctx context.Context, req *domain.ListBackupSettingReq,
) (*domain.ListBackupSettingResp, bizerr.BizError) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx)
	backupSettings, err := m.backupSettingRepo.List(ctx,
		m.listBackupSettingFilterOp(req.Filter),
	)
	if err != nil {
		logger.Error("list backup setting failed", zap.Any("request", req), zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrInternalError)
	}
	count, err := m.backupSettingRepo.Count(ctx,
		m.listBackupSettingFilterOp(req.Filter),
	)
	if err != nil {
		return nil, err
	}
	resp := &domain.ListBackupSettingResp{Total: count}
	if len(backupSettings) == 0 {
		return resp, nil
	}
	pie.Of(backupSettings).Each(func(backupSetting entity.BackupSetting) {
		resp.Items = append(resp.Items, domain.ListBackupSettingRespItem{
			ID:            uint64(backupSetting.ID),
			BackupSetting: &backupSetting,
		})
	})
	return resp, nil
}

func (m *backupSettingModel) listBackupSettingFilterOp(
	filter *domain.ListBackupSettingFilter,
) backupRepo.BackupSettingFilterOp {
	return func(qs entity.BackupSettingQuerySet) entity.BackupSettingQuerySet {
		if len(filter.ClusterIDs) != 0 {
			qs = qs.ClusterIDIn(pie.Map(filter.ClusterIDs, func(t uint64) entity.ID { return entity.ID(t) })...)
		}
		return qs
	}
}

func (m *backupSettingModel) ModifyRegionSetting(
	ctx context.Context, req *domain.RegionSettingModificationReqV2,
) (*domain.RegionSettingModificationRespV2, bizerr.BizError) {
	ctx = log.WithTraceID(ctx)
	backupSetting, err := m.backupSettingRepo.Get(ctx,
		func(qs entity.BackupSettingQuerySet) entity.BackupSettingQuerySet {
			return qs.ClusterIDEq(entity.ID(*req.ClusterID))
		})
	if err != nil {
		log.GetLogger(ctx).Warn("get backup setting failed", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	return m.regionModifierProxy.get(req, domain.InfraProviderType(backupSetting.InfraProviderType)).execute(ctx, req)
}

func (m *backupSettingModel) QueryMasterBucket(
	ctx context.Context, req *domain.QueryMasterBucketReqV2,
) (*domain.QueryMasterBucketRespV2, bizerr.BizError) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx)
	backupSetting, err := m.QueryBackupSetting(ctx, req.ClusterID)
	if err != nil {
		return nil, err
	}
	setting := backupSetting.BackupSetting
	masterBucket := setting.QueryMasterBucket()
	if masterBucket == nil || masterBucket.BucketId == 0 {
		logger.Error("master bucket not found")
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	bucket, getBucketErr := m.bucketRepo.Load(ctx, masterBucket.BucketId)
	if getBucketErr != nil {
		logger.Error("get bucket failed", zap.Error(getBucketErr))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	return &domain.QueryMasterBucketRespV2{
		BucketID:  &masterBucket.BucketId,
		BucketURI: pointer.ToString(bucket.URI()),
	}, nil
}

func (m *backupSettingModel) ModifyPitBackupSetting(
	ctx context.Context, req *domain.PitSettingModificationReq,
) bizerr.BizError {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx)
	backupSettingResp, err := m.QueryBackupSetting(ctx, &req.ClusterID)
	if err != nil {
		logger.Error("query backup setting failed", zap.Error(err))
		return bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	backupSetting := backupSettingResp.BackupSetting
	if err := m.preCheckModifyPitBackupSetting(ctx, req, backupSetting); err != nil {
		return err
	}

	if req.DryRun {
		logger.Info("dry run modify pit backup setting success")
		return nil
	}

	if err := m.dbProvider.Transaction(ctx, func(ctx context.Context) error {
		return m.executeModifyPiTRBackupSetting(ctx, req, backupSetting)
	}); err != nil {
		logger.Error("ModifyPitBackupSetting", zap.Error(err))
		return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterID)
	}
	return nil
}

func (m *backupSettingModel) executeModifyPiTRBackupSetting(
	ctx context.Context, req *domain.PitSettingModificationReq, backupSetting *entity.BackupSetting,
) error {
	logger := log.GetLogger(ctx)
	switchReq := domain.SwitchPitBackupReq{
		ClusterID: req.ClusterID,
	}
	nextStatus := m.statusMatrix.CalculateNextStatusForAction(status_matrix.CurrentStatus{
		Status:  string(backupSetting.Status),
		Actions: backupSetting.Action.CurrentActions,
	}, string(entity.BackupSettingActionModifyPitR))
	backupSetting.Action.CurrentActions = nextStatus.Actions
	backupSetting.Status = entity.BackupSettingStatus(nextStatus.Status)
	if req.Enable {
		_, err := m.pitBackupSwitcher.Enable(ctx, &domain.EnablePitBackupReq{SwitchPitBackupReq: switchReq})
		if err != nil {
			logger.Error("enable pit backup failed", zap.Error(err))
			return err
		}
		backupSetting.SetPiTREnable(true)
		return m.backupSettingRepo.Save(ctx, backupSetting)
	} else {
		resp, err := m.pitBackupSwitcher.Disable(
			ctx, &domain.DisablePitBackupReq{SwitchPitBackupReq: switchReq})
		if err != nil {
			logger.Error("disable pit backup failed", zap.Error(err))
			return err
		}
		if resp.WorkflowID == "" {
			backupSetting.Action.CurrentActions = pie.Of(backupSetting.Action.CurrentActions).
				Filter(func(s string) bool {
					return s != string(entity.BackupSettingActionModifyPitR)
				}).Result
		}
		backupSetting.SetPiTREnable(false)
		return m.backupSettingRepo.Save(ctx, backupSetting)
	}
}

func (m *backupSettingModel) preCheckModifyPitBackupSetting(
	ctx context.Context, req *domain.PitSettingModificationReq, backupSetting *entity.BackupSetting,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	isFriendly := m.statusMatrix.IsActionFriendly2Status(status_matrix.CurrentStatus{
		Status:  string(backupSetting.Status),
		Actions: backupSetting.Action.CurrentActions,
	}, string(entity.BackupSettingActionModifyPitR))
	if !isFriendly {
		logger.Warn("modify pit setting failed, status is not friendly",
			zap.Any("backup setting", backupSetting))
		return bizerr_wrapper.NewInvalidInputError(
			appconfig.ErrActionParam, backupSetting.Status, entity.BackupSettingActionModifyPitR)
	}
	switchReq := domain.SwitchPitBackupReq{
		ClusterID: req.ClusterID,
		DryRun:    true,
	}
	if req.Enable {
		_, err := m.pitBackupSwitcher.Enable(ctx, &domain.EnablePitBackupReq{SwitchPitBackupReq: switchReq})
		if err != nil {
			logger.Warn("enable pit backup failed", zap.Error(err))
			return err
		}
	} else {
		_, err := m.pitBackupSwitcher.Disable(
			ctx, &domain.DisablePitBackupReq{SwitchPitBackupReq: switchReq})
		if err != nil {
			logger.Warn("disable pit backup failed", zap.Error(err))
			return err
		}
	}
	return nil
}

func (m *backupSettingModel) QueryBackupRegionOptions(
	ctx context.Context, req *domain.QueryBackupRegionOptionsReq,
) (*domain.QueryBackupRegionOptionsResp, bizerr.BizError) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx)
	cluster, err := m.mgmtCli.GetAnyDedicatedClusterBasicByID(ctx, req.ClusterID)
	if err != nil {
		logger.Error("get cluster failed", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}

	cloudRegionMap, err := m.queryCloudRegions(ctx, cluster.GetProjectId(), cluster.GetCloudProvider())
	if err != nil {
		return nil, err
	}

	switch cp := strings.ToLower(cluster.GetCloudProvider()); cp {
	case "gcp":
		return m.queryGCPBackupRegionOptions(ctx, cluster, cloudRegionMap), nil
	case "aws":
		return m.genAWSBackupRegionOptions(ctx, cluster, cloudRegionMap), nil
	case "azure":
		return m.queryAzureBackupRegionOptions(ctx, cluster, cloudRegionMap), nil
	default:
		return nil, bizerr_wrapper.NewInternalErr("not supported cloud provider %s", cp)
	}
}

func (m *backupSettingModel) queryAzureBackupRegionOptions(
	ctx context.Context, cluster *mgmt.Cluster, cloudRegionMap map[string]*core.CloudRegion,
) *domain.QueryBackupRegionOptionsResp {
	dc := cluster.GetDedicatedCluster()
	// Azure does not support dual region backup, so use only the cluster region
	regionOpts := []domain.BackupRegionOption{
		{
			Region: &domain.RegionInfo{
				Name:        cluster.GetRegion(),
				DisplayName: dc.GetRegionDisplayName(),
			},
		},
	}
	resp := &domain.QueryBackupRegionOptionsResp{
		ClusterID: cluster.GetClusterId(),
		ClusterRegion: &domain.RegionInfo{
			Name:        cluster.GetRegion(),
			DisplayName: dc.GetRegionDisplayName(),
		},
		BackupRegionOptions: regionOpts,
	}
	return resp
}

func (m *backupSettingModel) queryGCPBackupRegionOptions(
	ctx context.Context, cluster *mgmt.Cluster, cloudRegionMap map[string]*core.CloudRegion,
) *domain.QueryBackupRegionOptionsResp {
	dc := cluster.GetDedicatedCluster()
	resp := &domain.QueryBackupRegionOptionsResp{
		ClusterID: cluster.GetClusterId(),
		ClusterRegion: &domain.RegionInfo{
			Name:        cluster.GetRegion(),
			DisplayName: dc.GetRegionDisplayName(),
		},
		BackupRegionOptions: make([]domain.BackupRegionOption, 0, len(cloudRegionMap)),
	}
	bucketOption := internalconfig.Global().ServerConfig.BucketOptions
	dualRegionCandidateSet := bucketOption.GetGCPDualRegionCandidateSet(cluster.GetRegion())
	if len(dualRegionCandidateSet) == 0 {
		// cloud region doesn't support dual region bucket
		return resp
	}

	for _, regionInfo := range cloudRegionMap {
		if regionInfo.Name == cluster.GetRegion() {
			continue
		}
		if !isRegionEnabled(regionInfo, cluster.GetOrganizationId()) {
			continue
		}

		if _, ok := dualRegionCandidateSet[regionInfo.Name]; !ok {
			continue
		}

		resp.BackupRegionOptions = append(resp.BackupRegionOptions, domain.BackupRegionOption{
			Region: &domain.RegionInfo{
				Name:        regionInfo.Name,
				DisplayName: regionInfo.DisplayName,
			},
		})
	}

	sort.Slice(resp.BackupRegionOptions, func(i, j int) bool {
		return resp.BackupRegionOptions[i].Region.Name < resp.BackupRegionOptions[j].Region.Name
	})
	return resp
}

func (m *backupSettingModel) genAWSBackupRegionOptions(
	ctx context.Context, cluster *mgmt.Cluster, cloudRegionMap map[string]*core.CloudRegion,
) *domain.QueryBackupRegionOptionsResp {
	dc := cluster.GetDedicatedCluster()
	regionOptions := make([]domain.BackupRegionOption, 0, len(cloudRegionMap))
	for _, regionInfo := range cloudRegionMap {
		if regionInfo.Name == cluster.GetRegion() {
			continue
		}
		if !isRegionEnabled(regionInfo, cluster.GetOrganizationId()) {
			continue
		}

		regionOptions = append(regionOptions, domain.BackupRegionOption{
			Region: &domain.RegionInfo{
				Name:        regionInfo.Name,
				DisplayName: regionInfo.DisplayName,
			},
		})
	}

	sort.Slice(regionOptions, func(i, j int) bool {
		return regionOptions[i].Region.Name < regionOptions[j].Region.Name
	})
	return &domain.QueryBackupRegionOptionsResp{
		ClusterID: cluster.GetClusterId(),
		ClusterRegion: &domain.RegionInfo{
			Name:        cluster.GetRegion(),
			DisplayName: dc.GetRegionDisplayName(),
		},
		BackupRegionOptions: regionOptions,
	}
}

func (m *backupSettingModel) queryCloudRegions(
	ctx context.Context, projectID uint64, cloudProviderName string,
) (map[string]*core.CloudRegion, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	cloudRegions, err := m.coreCli.ListCloudRegions(ctx, &core.ListCloudRegionsReq{
		ArchRouting: &core.ArchRouting{
			ProjectId: projectID,
		},
		Filter: &core.CloudRegionFilter{
			CloudProviderNames: []string{cloudProviderName},
		},
	})
	if err != nil {
		logger.Error("list cloud regions failed", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	if m, ok := cloudRegions[cloudv1.CloudProviderAWS]; ok {
		if _, ok = m[config.AWSRegionNameOsaka]; !ok {
			m[config.AWSRegionNameOsaka] = &core.CloudRegion{
				Name:              config.AWSRegionNameOsaka,
				CloudProviderName: cloudv1.CloudProviderAWS,
				DisplayName:       config.AWSRegionDisplayNameOsaka,
				Status:            core.CloudRegion_STATUS_ACTIVE,
				IsPublic:          true,
			}
		}
	} else if m, ok := cloudRegions[cloudv1.ProviderTypeGCP]; ok {
		if _, ok = m[config.GCPRegionNameOsaka]; !ok {
			m[config.GCPRegionNameOsaka] = &core.CloudRegion{
				Name:              config.GCPRegionNameOsaka,
				CloudProviderName: cloudv1.CloudProviderGCP,
				DisplayName:       config.GCPRegionDisplayNameOsaka,
				Status:            core.CloudRegion_STATUS_ACTIVE,
				IsPublic:          true,
			}
		}
	}
	return cloudRegions[cloudProviderName], nil
}

func (m *backupSettingModel) convertDataflowRegion(
	region string, regionMap map[string]*core.CloudRegion,
) *domain.RegionInfo {
	regionInfo := &domain.RegionInfo{
		Name:        region,
		DisplayName: region,
	}
	if region, ok := regionMap[region]; ok {
		regionInfo.DisplayName = region.DisplayName
	}
	return regionInfo
}

func (m *backupSettingModel) ModifyAutoBackupSetting(
	ctx context.Context, req *domain.ModifyAutoBackupSettingReq,
) (*domain.ModifyAutoBackupSettingResp, bizerr.BizError) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx)
	backupSettingResp, err := m.QueryBackupSetting(ctx, &req.ClusterID)
	if err != nil {
		return nil, err
	}
	setting := backupSettingResp.BackupSetting
	setting.SetAutoBackupDisable(req.Disable)

	if req.DryRun {
		logger.Info("dry run modify auto backup setting success")
		return nil, nil
	}

	if req.Disable {
		if err := m.dbProvider.Transaction(ctx, func(ctx context.Context) error {
			return m.backupSettingRepo.Save(ctx, setting)
		}); err != nil {
			logger.Error("ModifyAutoBackupSetting", zap.Error(err))
			return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterID)
		}
		return &domain.ModifyAutoBackupSettingResp{}, nil
	}

	backupSchedule, err := m.queryAutoBackupSchedule(ctx, req.ClusterID)
	if err != nil {
		return nil, err
	}

	cronSchedule, err := m.calculateScheduleCron(ctx, req)
	if err != nil {
		return nil, err
	}

	if err := m.dbProvider.Transaction(ctx, func(ctx context.Context) error {
		if err := m.backupSettingRepo.Save(ctx, setting); err != nil {
			return err
		}
		err = m.backupScheduleRepo.PatchAuto(ctx,
			backupSchedule.ClusterID,
			map[string]interface{}{
				"schedule":         cronSchedule,
				"retention_period": uint64(req.RetentionPolicy.RetentionPeriod),
				"backup_cycle":     domain.SnapshotBackupScheduleDomain2Entity(req.ScheduleType),
				"next_backup_time": utils.CalculateNextBackupTime(cronSchedule),
			})
		if err != nil {
			return err
		}
		return nil
	}); err != nil {
		logger.Error("ModifyAutoBackupSetting", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterID)
	}
	return &domain.ModifyAutoBackupSettingResp{}, nil
}

func (m *backupSettingModel) calculateScheduleCron(
	ctx context.Context, req *domain.ModifyAutoBackupSettingReq,
) (string, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	if req.ScheduleType == domain.SnapshotBackupScheduleTypeDaily {
		if cronSchedule, gErr := entity.GetCron(
			entity.BackupCycleDaily, req.DailyScheduleTime, nil); gErr != nil {
			logger.Warn("failed to get cron schedule", zap.Error(gErr))
			return "", bizerr_wrapper.NewInvalidInputError(appconfig.ErrInvalidParam,
				"dailyScheduleTime", req.DailyScheduleTime)
		} else {
			return cronSchedule, nil
		}
	}
	if req.WeeklySchedule == nil {
		logger.Warn("weekly schedule is required")
		return "", bizerr_wrapper.NewInvalidInputError(appconfig.ErrInvalidParam,
			"weeklySchedule", req.WeeklySchedule)
	}
	if cronSchedule, gErr := entity.GetCron(entity.BackupCycleWeekly, req.WeeklySchedule.ScheduleTime,
		domain.WeekdayDomainList2Entity(req.WeeklySchedule.Days)); gErr != nil {
		logger.Warn("failed to get cron schedule", zap.Error(gErr))
		return "", bizerr_wrapper.NewInvalidInputError(appconfig.ErrInvalidParam,
			"weeklySchedule", req.WeeklySchedule)
	} else {
		return cronSchedule, nil
	}
}

func (m *backupSettingModel) queryAutoBackupSchedule(
	ctx context.Context, clusterID uint64,
) (*entity.BackupSchedule, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	backupSchedules, err := m.backupScheduleRepo.List(ctx,
		func(qs entity.BackupScheduleQuerySet) entity.BackupScheduleQuerySet {
			return qs.ClusterIDEq(entity.ID(clusterID)).TypeEq(entity.BackupTypeAuto)
		})
	if err != nil {
		logger.Error("list backup schedule failed", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	if len(backupSchedules) == 0 {
		logger.Error("no backup schedule found")
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	return &backupSchedules[0], nil
}

func isRegionEnabled(region *core.CloudRegion, tenantID uint64) bool {
	// status of inactive means the region is not open to any tenant
	if region.Status == core.CloudRegion_STATUS_INACTIVE {
		return false
	}
	if region.IsPublic {
		return true
	}
	// IsPublic false means the region is only open to limited tenants
	return pie.Contains(region.OpenedTenantIds, tenantID)
}