package migrationcluster

import (
	"context"
	"fmt"
	"strconv"

	"github.com/samber/lo"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"golang.org/x/mod/semver"
	"google.golang.org/protobuf/encoding/protojson"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	openapicommonv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/common/v1beta1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/constant"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/service/workflow/dag"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/pkg/apimiddleware/bizerror"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/pkg/common"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/grpc/metadata"
)

func (svc *service) CreateMigrationCluster(
	ctx context.Context,
	req *tidbmgmtservicev1.CreateMigrationClusterRequest,
	projectID uint64,
	sourceCluster *entity.Cluster,
) (*tidbmgmtservicev1.MigrationCluster, error) {
	spanName := "CreateMigrationCluster"
	ctx, span := tracer.Start(ctx, spanName)
	defer span.End()

	migrationCluster := req.GetMigrationCluster()

	sourceGlobalCluster, err := sourceCluster.GetDedicatedCluster()
	if err != nil {
		return migrationCluster, bizerr.WrapErrorWithCode(err, bizerr.GeneralInternalErrorCode, "failed to get dedicated cluster")
	}

	targetRegionID, err := svc.resolveTargetRegionID(ctx, sourceCluster, sourceGlobalCluster, migrationCluster)
	if err != nil {
		return migrationCluster, err
	}
	migrationCluster.TargetRegionId = targetRegionID
	backupRegionID, err := svc.resolveBackupRegionID(ctx, sourceCluster, sourceGlobalCluster, migrationCluster, targetRegionID)
	if err != nil {
		return migrationCluster, err
	}
	migrationCluster.BackupRegionId = backupRegionID

	// transaction
	if err := svc.transaction.Transaction(ctx, func(ctx context.Context) error {
		// Check cluster name is unique under the project
		count, err := svc.clusterRepo.CountDisplayNameUnderProject(ctx, projectID, migrationCluster.ClusterDisplayName)
		if err != nil {
			return err
		}
		if count > 0 {
			return bizerror.NewDisplayNameAlreadyExists(constant.ErrDuplicatedClusterName)
		}

		clusterID := entity.GenClusterID()
		createClusterRequest, err := svc.buildCreateRequest(
			ctx,
			sourceCluster,
			migrationCluster,
			projectID,
			clusterID,
		)
		if err != nil {
			return bizerr.WrapError(err, "build create cluster request")
		}
		createClusterRequest.ValidateOnly = true
		_, err = svc.dedicatedCli.CreateCluster(ctx, createClusterRequest)
		if err != nil {
			tilog.Error(
				ctx,
				"migration cluster preflight validation failed",
				zap.Error(err),
				zap.Uint64("source_cluster_id", sourceCluster.ClusterId),
				zap.Uint64("target_cluster_id", clusterID),
				zap.Uint64("target_project_id", projectID),
				zap.String("target_region", targetRegionID.String()),
				zap.String("backup_region", backupRegionID.String()),
			)
			return err
		}

		additionalNodeGroups, err := buildAdditionalNodeGroups(sourceCluster, migrationCluster)
		if err != nil {
			return bizerr.WrapError(err, "build additional node groups")
		}
		tilog.Info(ctx, "additional node groups for migration cluster", zap.Any("nodeGroups", additionalNodeGroups))

		// return directly if only validate
		if req.ValidateOnly {
			return nil
		}

		migrationCluster.TargetClusterId = clusterID
		migrationCluster.State = tidbmgmtservicev1.MigrationCluster_PENDING
		// create migration cluster record
		if err = svc.migrationClusterRepo.Save(ctx, migrationCluster); err != nil {
			return fmt.Errorf("failed to create migration cluster record: %w", err)
		}

		// create cluster record
		cluster := buildCluster(req, sourceCluster, createClusterRequest.Cluster)
		_, err = svc.clusterRepo.InsertNewCluster(ctx, cluster)
		if err != nil {
			return fmt.Errorf("failed to create cluster record: %w", err)
		}

		createClusterRequest.ValidateOnly = false
		requestJson, err := protojson.Marshal(createClusterRequest)
		if err != nil {
			return fmt.Errorf("failed to marshal create cluster request: %w", err)
		}

		// validate target cluster version
		targetClusterVersion := migrationCluster.TargetClusterVersion
		if targetClusterVersion != "" {
			if !semver.IsValid(targetClusterVersion) {
				return bizerr.New(bizerr.ErrCodeParams, "invalid target cluster version: %s", targetClusterVersion)
			}
			if semver.Compare(targetClusterVersion, sourceGlobalCluster.GetVersion()) <= 0 {
				return bizerr.New(
					bizerr.ErrCodeParams,
					"target cluster version %s is not greater than source cluster version %s",
					targetClusterVersion,
					sourceGlobalCluster.Version,
				)
			}
		}

		err = dag.RunCreateDedicatedMigrationClusterDag(
			ctx,
			svc.workflowSvc,
			&dag.CreateMigrationClusterDagTasks{
				TenantId:             sourceCluster.OrganizationId,
				ProjectId:            projectID,
				ClusterId:            clusterID,
				MigrationClusterId:   migrationCluster.Id,
				CreateClusterRequest: string(requestJson),
				TargetClusterVersion: targetClusterVersion,
				BackupId:             strconv.FormatUint(migrationCluster.GetBackupId(), 10),
				AdditionalNodeGroups: additionalNodeGroups,
				ConfigSyncStrategy: &dag.ConfigSyncStrategy{
					AllowFallbackToBackup: req.GetConfigSyncStrategy().GetAllowFallbackToBackup(),
					SkipSystemVariables:   req.GetConfigSyncStrategy().GetSkipSystemVariables(),
					SkipPdConfig:          req.GetConfigSyncStrategy().GetSkipPdConfig(),
					SkipPdServerConfig:    req.GetConfigSyncStrategy().GetSkipPdServerConfig(),
					SkipPlacementRules:    req.GetConfigSyncStrategy().GetSkipPlacementRules(),
					SkipStoreLimits:       req.GetConfigSyncStrategy().GetSkipStoreLimits(),
				},
			},
		)
		if err != nil {
			return fmt.Errorf("failed to create migration cluster dag: %w", err)
		}

		return nil
	}); err != nil {
		return migrationCluster, err
	}

	return migrationCluster, nil
}

func (svc *service) buildCreateRequest(
	ctx context.Context,
	sourceCluster *entity.Cluster,
	migrationCluster *tidbmgmtservicev1.MigrationCluster,
	projectID uint64,
	clusterID uint64,
) (*globaldedicatedv1.CreateClusterRequest, bizerr.BizError) {
	sourceGlobalCluster, err := sourceCluster.GetDedicatedCluster()
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.GeneralInternalErrorCode, "failed to get dedicated cluster")
	}

	targetRegionID := migrationCluster.GetTargetRegionId()

	createClusterRequest := &globaldedicatedv1.CreateClusterRequest{
		Cluster: &globaldedicatedv1.Cluster{
			ClusterId:         clusterID,
			OrganizationId:    sourceCluster.OrganizationId,
			ProjectId:         projectID,
			CloudProviderId:   sourceGlobalCluster.CloudProviderId,
			RegionId:          targetRegionID,
			ClusterComponents: nil, // fill later
			Port:              sourceGlobalCluster.Port,
			Version:           sourceGlobalCluster.Version,
		},
		ValidateOnly:              false,
		ClusterConfiguration:      nil,
		OverwriteTenantInitConfig: true,
	}

	components := make([]*globaldedicatedv1.Cluster_ClusterComponent, 0)
	sourceTidbComponent, ok := lo.Find(sourceGlobalCluster.ClusterComponents, func(c *globaldedicatedv1.Cluster_ClusterComponent) bool {
		return c.ComponentType == globaldedicatedv1.ComponentType_TIDB && c.IsDefaultTidbGroup
	})
	if !ok {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "source cluster missing default TiDB component")
	}
	tidbComponent := &globaldedicatedv1.Cluster_ClusterComponent{
		ComponentType: globaldedicatedv1.ComponentType_TIDB,
		SpuName:       sourceTidbComponent.SpuName,
		Count:         sourceTidbComponent.Count,
	}
	if migrationCluster.TidbNodeSetting != nil {
		if migrationCluster.TidbNodeSetting.NodeSpecKey != "" {
			targetSpuName := migrationCluster.TidbNodeSetting.NodeSpecKey

			err = svc.checkSPU(ctx, sourceCluster, projectID, sourceTidbComponent, targetSpuName, targetRegionID)
			if err != nil {
				return nil, bizerr.WrapError(err, "check TiDB SPU")
			}

			tidbComponent.SpuName = targetSpuName
		}
		for _, group := range migrationCluster.TidbNodeSetting.TidbNodeGroups {
			if group.IsDefaultGroup && group.NodeCount != nil {
				tidbComponent.Count = lo.ToPtr(uint32(group.GetNodeCount()))
			}
		}
	}
	components = append(components, tidbComponent)

	sourceTikvComponent, ok := lo.Find(sourceGlobalCluster.ClusterComponents, func(c *globaldedicatedv1.Cluster_ClusterComponent) bool {
		return c.ComponentType == globaldedicatedv1.ComponentType_TIKV
	})
	if !ok {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "source cluster missing TiKV component")
	}
	tikvComponent := &globaldedicatedv1.Cluster_ClusterComponent{
		ComponentType: globaldedicatedv1.ComponentType_TIKV,
		SpuName:       sourceTikvComponent.SpuName,
		Count:         sourceTikvComponent.Count,
		Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
			globaldedicatedv1.Sku_DATA.String(): {
				VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
			},
			globaldedicatedv1.Sku_RAFT_LOG.String(): {
				VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
			},
		},
	}
	for key, storage := range sourceTikvComponent.Storages {
		if _, ok := tikvComponent.Storages[key]; !ok {
			return nil, bizerr.New(bizerr.ErrCodeUnknown, "unexpected TiKV storage key: %s", key)
		}
		tikvComponent.Storages[key].VolumeType = storage.VolumeType
		tikvComponent.Storages[key].SizeGi = storage.SizeGi
		tikvComponent.Storages[key].Iops = storage.Iops
		tikvComponent.Storages[key].ThroughputMibps = storage.ThroughputMibps
		tikvComponent.Storages[key].AdditionalIops = storage.AdditionalIops
	}
	if migrationCluster.TikvNodeSetting != nil {
		if migrationCluster.TikvNodeSetting.NodeSpecKey != "" {
			targetSpuName := migrationCluster.TikvNodeSetting.NodeSpecKey

			err = svc.checkSPU(ctx, sourceCluster, projectID, sourceTikvComponent, targetSpuName, targetRegionID)
			if err != nil {
				return nil, bizerr.WrapError(err, "check TiKV SPU")
			}

			tikvComponent.SpuName = targetSpuName
		}

		if migrationCluster.TikvNodeSetting.NodeCount != nil {
			tikvComponent.Count = lo.ToPtr(uint32(migrationCluster.TikvNodeSetting.GetNodeCount()))
		}
	}
	components = append(components, tikvComponent)

	sourcePdComponent, ok := lo.Find(sourceGlobalCluster.ClusterComponents, func(c *globaldedicatedv1.Cluster_ClusterComponent) bool {
		return c.ComponentType == globaldedicatedv1.ComponentType_PD
	})
	if !ok {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "source cluster missing PD component")
	}
	pdComponent := &globaldedicatedv1.Cluster_ClusterComponent{
		ComponentType: globaldedicatedv1.ComponentType_PD,
		SpuName:       sourcePdComponent.SpuName,
		Count:         sourcePdComponent.Count,
		Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
			globaldedicatedv1.Sku_DATA.String(): {
				VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
			},
			globaldedicatedv1.Sku_RAFT_LOG.String(): {
				VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
			},
		},
	}
	for key, storage := range sourcePdComponent.Storages {
		pdComponent.Storages[key] = &globaldedicatedv1.Cluster_ClusterComponent_Storage{
			SizeGi:          storage.SizeGi,
			VolumeType:      storage.VolumeType,
			Iops:            storage.Iops,
			ThroughputMibps: storage.ThroughputMibps,
			AdditionalIops:  storage.AdditionalIops,
		}
	}
	if migrationCluster.PdNodeSetting != nil {
		if migrationCluster.PdNodeSetting.NodeSpecKey != "" {
			targetSpuName := migrationCluster.PdNodeSetting.NodeSpecKey

			err = svc.checkSPU(ctx, sourceCluster, projectID, sourcePdComponent, targetSpuName, targetRegionID)
			if err != nil {
				return nil, bizerr.WrapError(err, "check PD SPU")
			}

			pdComponent.SpuName = targetSpuName
		}
	}
	components = append(components, pdComponent)

	sourceTiflashComponent, ok := lo.Find(sourceGlobalCluster.ClusterComponents, func(c *globaldedicatedv1.Cluster_ClusterComponent) bool {
		return c.ComponentType == globaldedicatedv1.ComponentType_TIFLASH
	})
	if ok {
		tiflashComponent := &globaldedicatedv1.Cluster_ClusterComponent{
			ComponentType: globaldedicatedv1.ComponentType_TIFLASH,
			SpuName:       sourceTiflashComponent.SpuName,
			Count:         sourceTiflashComponent.Count,
			Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
				globaldedicatedv1.Sku_DATA.String(): {
					VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
				},
				globaldedicatedv1.Sku_RAFT_LOG.String(): {
					VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
				},
			},
		}
		for key, storage := range sourceTiflashComponent.Storages {
			tiflashComponent.Storages[key] = &globaldedicatedv1.Cluster_ClusterComponent_Storage{
				SizeGi:          storage.SizeGi,
				VolumeType:      storage.VolumeType,
				Iops:            storage.Iops,
				ThroughputMibps: storage.ThroughputMibps,
				AdditionalIops:  storage.AdditionalIops,
			}
		}
		if migrationCluster.TiflashNodeSetting != nil {
			if migrationCluster.TiflashNodeSetting.NodeSpecKey != "" {
				targetSpuName := migrationCluster.TiflashNodeSetting.NodeSpecKey

				err = svc.checkSPU(ctx, sourceCluster, projectID, sourceTiflashComponent, targetSpuName, targetRegionID)
				if err != nil {
					return nil, bizerr.WrapError(err, "check TiFlash SPU")
				}

				tiflashComponent.SpuName = targetSpuName
			}

			if migrationCluster.TiflashNodeSetting.NodeCount != nil {
				tiflashComponent.Count = lo.ToPtr(uint32(migrationCluster.TiflashNodeSetting.GetNodeCount()))
			}
		}

		if tiflashComponent.Count != nil && *tiflashComponent.Count > 0 {
			components = append(components, tiflashComponent)
		}
	}

	createClusterRequest.Cluster.ClusterComponents = components
	return createClusterRequest, nil
}

func buildCluster(
	req *tidbmgmtservicev1.CreateMigrationClusterRequest,
	sourceCluster *entity.Cluster,
	globalTargetCluster *globaldedicatedv1.Cluster,
) *entity.Cluster {
	migrationCluster := req.GetMigrationCluster()
	out := &entity.Cluster{
		ClusterId:      globalTargetCluster.GetClusterId(),
		ProjectId:      globalTargetCluster.GetProjectId(),
		OrganizationId: globalTargetCluster.GetOrganizationId(),
		DisplayName:    migrationCluster.GetClusterDisplayName(),
		DeployType:     entity.DeployTypeDedicated,
		CloudProvider:  sourceCluster.CloudProvider,
		Region:         globalTargetCluster.GetRegionId().String(),
		UserId:         sourceCluster.UserId,
		CreatedBy:      sourceCluster.CreatedBy,
		ServicePlan:    openapicommonv1beta1.ServicePlan_Dedicated,
		Labels: map[string]string{
			common.ProjectIDLabelKey: strconv.FormatUint(globalTargetCluster.GetProjectId(), 10),
			common.OrgIDLabelKey:     strconv.FormatUint(globalTargetCluster.GetOrganizationId(), 10),
			metadata.UserIDHeaderKey: sourceCluster.UserId,
		},
		GlobalCluster: globalTargetCluster,
	}
	return out
}

func buildAdditionalNodeGroups(
	sourceCluster *entity.Cluster,
	migrationCluster *tidbmgmtservicev1.MigrationCluster,
) ([]dag.NodeGroup, bizerr.BizError) {
	sourceGlobalCluster, err := sourceCluster.GetDedicatedCluster()
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.GeneralInternalErrorCode, "failed to get dedicated cluster")
	}

	additionalTidbComponents := lo.Filter(sourceGlobalCluster.ClusterComponents, func(c *globaldedicatedv1.Cluster_ClusterComponent, _ int) bool {
		return c.ComponentType == globaldedicatedv1.ComponentType_TIDB && !c.IsDefaultTidbGroup && lo.FromPtr(c.Count) > 0
	})
	if len(additionalTidbComponents) == 0 {
		return nil, nil
	}

	additionalNodeGroups := lo.Map(additionalTidbComponents, func(c *globaldedicatedv1.Cluster_ClusterComponent, _ int) dag.NodeGroup {
		return dag.NodeGroup{
			DisplayName: c.TidbGroupDisplayName,
			NodeCount:   int32(lo.FromPtr(c.Count)),
		}
	})

	if migrationCluster.TidbNodeSetting != nil {
		nodeGroupMap := lo.Associate(
			migrationCluster.TidbNodeSetting.TidbNodeGroups,
			func(group *tidbmgmtservicev1.MigrationCluster_TidbNodeGroup) (string, *tidbmgmtservicev1.MigrationCluster_TidbNodeGroup) {
				return group.DisplayName, group
			},
		)

		for i, group := range additionalNodeGroups {
			if ng, ok := nodeGroupMap[group.DisplayName]; ok && !ng.IsDefaultGroup {
				if ng.NodeCount != nil {
					additionalNodeGroups[i].NodeCount = ng.GetNodeCount()
				}
			}
		}
	}

	return additionalNodeGroups, nil
}

func (svc *service) checkSPU(
	ctx context.Context,
	sourceCluster *entity.Cluster,
	projectID uint64,
	sourceComponent *globaldedicatedv1.Cluster_ClusterComponent,
	targetSpuName string,
	targetRegionID globalcommonv1.Region_Name,
) bizerr.BizError {
	sourceGlobalCluster, err := sourceCluster.GetDedicatedCluster()
	if err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.GeneralInternalErrorCode, "failed to get dedicated cluster")
	}

	var targetSpu *globaldedicatedv1.Spu

	// Same-region path keeps the original logic unchanged.
	// PD must also use the without_sku path for cross-region migration because
	// ListSpus(with_sku) only returns components backed by ComponentsUserEditable
	// in dedicated-global-service, and PD is excluded there.
	if targetRegionID == sourceGlobalCluster.GetRegionId() || sourceComponent.GetComponentType() == globaldedicatedv1.ComponentType_PD {
		response, err := svc.dedicatedCli.ListSpus(
			ctx,
			&globaldedicatedv1.ListSpusRequest{
				Request: &globaldedicatedv1.ListSpusRequest_WithoutSku{
					WithoutSku: &globaldedicatedv1.ListSpusRequest_FilterWithoutSku{
						ComponentTypes: []globaldedicatedv1.ComponentType{sourceComponent.ComponentType},
						Name:           targetSpuName,
						Versions:       nil,
					},
				},
			},
		)
		if err != nil {
			return bizerr.WrapError(err, "list spus")
		}
		if len(response.Spus) == 0 {
			return bizerr.New(
				bizerr.ErrCodeParams,
				"component type: %s: target SPU %s not found",
				sourceComponent.ComponentType.String(),
				targetSpuName,
			)
		}
		targetSpu = response.Spus[0]
	} else {
		response, err := svc.dedicatedCli.ListSpus(
			ctx,
			&globaldedicatedv1.ListSpusRequest{
				Request: &globaldedicatedv1.ListSpusRequest_WithSku{
					WithSku: &globaldedicatedv1.ListSpusRequest_FilterWithSku{
						CloudProvider: sourceGlobalCluster.GetCloudProviderId(),
						Region:        targetRegionID,
						TenantId:      sourceCluster.OrganizationId,
						ProjectId:     projectID,
					},
				},
			},
		)
		if err != nil {
			return bizerr.WrapError(err, "list spus with sku")
		}
		var ok bool
		targetSpu, ok = lo.Find(response.GetSpus(), func(spu *globaldedicatedv1.Spu) bool {
			return spu.GetComponentType() == sourceComponent.GetComponentType() &&
				spu.GetName() == targetSpuName
		})
		if !ok {
			return bizerr.New(
				bizerr.ErrCodeParams,
				"component type: %s: target SPU %s is not available in target region %s",
				sourceComponent.ComponentType.String(),
				targetSpuName,
				targetRegionID.String(),
			)
		}
	}

	if targetSpu.GetCpuCore() != int32(sourceComponent.GetCpuNum()) || targetSpu.GetMemorySizeGi() != int32(sourceComponent.GetMemorySizeGi()) {
		return bizerr.New(
			bizerr.ErrCodeParams,
			"%s: target SPU (%s) CPU/Memory size does not match source cluster SPU (%s)",
			sourceComponent.ComponentType.String(), targetSpuName, sourceComponent.SpuName,
		)
	}

	return nil
}

func (svc *service) resolveTargetRegionID(
	ctx context.Context,
	sourceCluster *entity.Cluster,
	sourceGlobalCluster *globaldedicatedv1.Cluster,
	migrationCluster *tidbmgmtservicev1.MigrationCluster,
) (globalcommonv1.Region_Name, bizerr.BizError) {
	targetRegionID := migrationCluster.GetTargetRegionId()
	if targetRegionID == globalcommonv1.Region_NAME_UNSPECIFIED {
		return sourceGlobalCluster.GetRegionId(), nil
	}
	if err := svc.validateRegionIDUnderSourceCloudProvider(ctx, sourceCluster, sourceGlobalCluster, targetRegionID, "target_region_id"); err != nil {
		return globalcommonv1.Region_NAME_UNSPECIFIED, err
	}
	return targetRegionID, nil
}

func (svc *service) resolveBackupRegionID(
	ctx context.Context,
	sourceCluster *entity.Cluster,
	sourceGlobalCluster *globaldedicatedv1.Cluster,
	migrationCluster *tidbmgmtservicev1.MigrationCluster,
	targetRegionID globalcommonv1.Region_Name,
) (globalcommonv1.Region_Name, bizerr.BizError) {
	backupRegionID := migrationCluster.GetBackupRegionId()
	if backupRegionID == globalcommonv1.Region_NAME_UNSPECIFIED {
		return targetRegionID, nil
	}
	if err := svc.validateRegionIDUnderSourceCloudProvider(ctx, sourceCluster, sourceGlobalCluster, backupRegionID, "backup_region_id"); err != nil {
		return globalcommonv1.Region_NAME_UNSPECIFIED, err
	}
	return backupRegionID, nil
}

func (svc *service) validateRegionIDUnderSourceCloudProvider(
	ctx context.Context,
	sourceCluster *entity.Cluster,
	sourceGlobalCluster *globaldedicatedv1.Cluster,
	regionID globalcommonv1.Region_Name,
	fieldName string,
) bizerr.BizError {
	if regionID == sourceGlobalCluster.GetRegionId() {
		return nil
	}
	if regionID == globalcommonv1.Region_NAME_UNSPECIFIED {
		return bizerr.New(bizerr.ErrCodeParams, "%s is unspecified", fieldName)
	}

	_, err := svc.dedicatedCli.SelectRegionByOrgId(
		ctx,
		&globaldedicatedv1.SelectRegionByOrgIdRequest{
			OrganizationId:  sourceCluster.OrganizationId,
			CloudProviderId: sourceGlobalCluster.CloudProviderId,
			RegionId:        regionID,
		},
	)
	if err != nil {
		return bizerr.New(
			bizerr.ErrCodeParams,
			"%s %s is invalid or not under the same cloud provider as source cluster",
			fieldName,
			regionID.String(),
		)
	}
	return nil
}