package cluster

import (
	"math/rand/v2"
	"strconv"
	"testing"

	"github.com/samber/lo"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	regionaldedicatedv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/regional-server/v1"
	"go.uber.org/mock/gomock"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/adapter/account"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/adapter/account/mock_account"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/adapter/billing"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/adapter/billing/mock_billing"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/adapter/clusterconfig/mock_clusterconfig"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/adapter/clustermetric/mock_clustermetric"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/adapter/regionserver/mock_regionserver"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/adapter/remoteencryptor/mock_remoteencryptor"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/service/workflow/common"
)

func TestCreateCluster(t *testing.T) {
	ctrl := gomock.NewController(t, gomock.WithOverridableExpectations())

	encryptor := mock_remoteencryptor.NewMockClient(ctrl)
	encryptor.EXPECT().EncryptBase64(gomock.Any(), gomock.Any()).Return("encrypted", nil).AnyTimes()
	encryptor.EXPECT().DecryptBase64(gomock.Any(), gomock.Any()).Return("password", nil).AnyTimes()

	accountCli := mock_account.NewMockClient(ctrl)
	accountCli.EXPECT().GetProject(gomock.Any(), uint64(ProjectID)).Return(&account.ProjectInfo{
		ID:             ProjectID,
		TenantID:       TenantID,
		Name:           "test",
		AwsCmekEnabled: false,
	}, nil).AnyTimes()
	accountCli.EXPECT().GetTenant(gomock.Any(), uint64(TenantID)).Return(&account.TenantInfo{
		ID:   TenantID,
		Name: "test",
		Plan: account.TenantPlanPOC,
	}, nil).AnyTimes()

	regionalCli := mock_regionserver.NewMockClient(ctrl)
	regionalCli.EXPECT().ListVpcs(gomock.Any(), gomock.Any(), &regionaldedicatedv1.ListVpcsRequest{
		TenantId:  strconv.FormatUint(TenantID, 10),
		ProjectId: strconv.FormatUint(ProjectID, 10),
		Statuses: []regionaldedicatedv1.Vpc_Status{
			regionaldedicatedv1.Vpc_ACTIVE,
			regionaldedicatedv1.Vpc_CREATING,
			regionaldedicatedv1.Vpc_FAILED,
		},
	}).Return(&regionaldedicatedv1.ListVpcsResponse{
		Vpcs: []*regionaldedicatedv1.Vpc{
			{
				VpcId:      "vpc-xxxxxxxxxxxxxxxxx",
				TenantIds:  []string{strconv.FormatUint(TenantID, 10)},
				ProjectIds: []string{strconv.FormatUint(ProjectID, 10)},
				CidrBlock:  "0.0.0.0/0",
			},
		},
	}, nil).AnyTimes()

	metricCli := mock_clustermetric.NewMockClient(ctrl)

	billingCli := mock_billing.NewMockClient(ctrl)
	billingCli.EXPECT().
		BillingActionCheck(gomock.Any(), billing.Action_Create, uint64(TenantID), gomock.Any()).
		Return(&billing.ActionCheckResult{Pass: true}, nil).
		AnyTimes()

	clusterConfigCli := mock_clusterconfig.NewMockClient(ctrl)
	clusterConfigCli.EXPECT().GetClusterVersion(gomock.Any(), gomock.Any()).Return(ClusterVersion, nil).AnyTimes()
	clusterConfigCli.EXPECT().
		ListComponentConfigurations(gomock.Any(), gomock.Any()).
		Return([]*regionaldedicatedv1.ComponentConfiguration{}, nil).
		AnyTimes()

	svc := newTestService(encryptor, accountCli, regionalCli, metricCli, billingCli, clusterConfigCli, true)

	t.Run("tidb/tikv/tiflash components spu-and-sku-related fields should be correct", func(t *testing.T) {
		req := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
							},
						},
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIFLASH,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType:      lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:          555,
								Iops:            555,
								ThroughputMibps: 555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
							},
						},
					},
				},
			},
		}
		resCluster, err := svc.Create(t.Context(), req)
		require.NoError(t, err)

		// check the tidb component.
		tidbComp, ok := lo.Find(
			resCluster.GetClusterComponents(),
			func(comp *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return comp.GetComponentType() == globaldedicatedv1.ComponentType_TIDB
			},
		)
		require.True(t, ok)
		assert.Equal(t, entity.BuildSpuKey(globaldedicatedv1.ComponentType_TIDB, "8c8g", ""), tidbComp.GetSpuKey())
		assert.Equal(t, "tidb-8c8g-arm", tidbComp.GetSkuKey())
		assert.Equal(t, "8c8g", tidbComp.GetSpuName())
		assert.Equal(t, "8c8g", tidbComp.GetSpuDisplayName())
		assert.Equal(t, uint32(1), tidbComp.GetCount())
		assert.NotEmpty(t, tidbComp.GetTidbGroupId())
		assert.NotEmpty(t, tidbComp.GetTidbGroupDisplayName())
		assert.True(t, tidbComp.GetIsDefaultTidbGroup())
		assert.Equal(t, "instance_type_8c8g_arm_6g.", tidbComp.GetInstanceType())
		assert.Equal(t, uint32(8), tidbComp.GetCpuNum())
		assert.Equal(t, uint32(8), tidbComp.GetMemorySizeGi())
		assert.Equal(t, globaldedicatedv1.InstanceMu_ARM64, tidbComp.GetInstanceCpuArch())

		// check the tikv component.
		tikvComp, ok := lo.Find(
			resCluster.GetClusterComponents(),
			func(comp *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return comp.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
			},
		)
		require.True(t, ok)
		assert.Equal(t, entity.BuildSpuKey(globaldedicatedv1.ComponentType_TIKV, "8c8g", ""), tikvComp.GetSpuKey())
		assert.Equal(t, "tikv-8c8g-arm-data:gp3-raft:none-fixed", tikvComp.GetSkuKey())
		assert.Equal(t, "8c8g", tikvComp.GetSpuName())
		assert.Equal(t, "8c8g", tikvComp.GetSpuDisplayName())
		assert.Equal(t, uint32(3), tikvComp.GetCount())
		assert.NotEmpty(t, tidbComp.GetTidbGroupId())
		assert.NotEmpty(t, tidbComp.GetTidbGroupDisplayName())
		assert.True(t, tidbComp.GetIsDefaultTidbGroup())
		assert.Equal(t, "instance_type_8c8g_arm_6g.", tikvComp.GetInstanceType())
		assert.Equal(t, uint32(8), tikvComp.GetCpuNum())
		assert.Equal(t, uint32(8), tikvComp.GetMemorySizeGi())
		assert.Equal(t, globaldedicatedv1.InstanceMu_ARM64, tikvComp.GetInstanceCpuArch())
		// Check storage configuration
		tikvStorage, exists := tikvComp.GetStorages()[globaldedicatedv1.Sku_DATA.String()]
		require.True(t, exists)
		assert.Equal(t, globaldedicatedv1.StorageMu_GP3, tikvStorage.GetVolumeType())
		assert.Equal(t, uint32(555), tikvStorage.GetSizeGi())
		// raft disk is not set, so it should be nil
		assert.Nil(t, tikvComp.GetStorages()[globaldedicatedv1.Sku_RAFT_LOG.String()])

		// check the tiflash component.
		tiflashComp, ok := lo.Find(
			resCluster.GetClusterComponents(),
			func(comp *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return comp.GetComponentType() == globaldedicatedv1.ComponentType_TIFLASH
			},
		)
		require.True(t, ok)
		assert.Equal(
			t,
			entity.BuildSpuKey(globaldedicatedv1.ComponentType_TIFLASH, "8c8g", ""),
			tiflashComp.GetSpuKey(),
		)
		assert.Equal(t, "tiflash-8c8g-arm-data:gp3-dynamic", tiflashComp.GetSkuKey())
		assert.Equal(t, "8c8g", tiflashComp.GetSpuName())
		assert.Equal(t, "8c8g", tiflashComp.GetSpuDisplayName())
		assert.Equal(t, uint32(1), tiflashComp.GetCount())
		assert.NotEmpty(t, tidbComp.GetTidbGroupId())
		assert.NotEmpty(t, tidbComp.GetTidbGroupDisplayName())
		assert.True(t, tidbComp.GetIsDefaultTidbGroup())
		assert.Equal(t, "instance_type_8c8g_arm_6g.", tiflashComp.GetInstanceType())
		assert.Equal(t, uint32(8), tiflashComp.GetCpuNum())
		assert.Equal(t, uint32(8), tiflashComp.GetMemorySizeGi())
		assert.Equal(t, globaldedicatedv1.InstanceMu_ARM64, tiflashComp.GetInstanceCpuArch())
		// Check storage configuration with specific values for IOPS and throughput
		tiflashStorage, exists := tiflashComp.GetStorages()[globaldedicatedv1.Sku_DATA.String()]
		require.True(t, exists)
		assert.Equal(t, globaldedicatedv1.StorageMu_GP3, tiflashStorage.GetVolumeType())
		assert.Equal(t, uint32(555), tiflashStorage.GetSizeGi())
		assert.Equal(t, uint32(999), tiflashStorage.GetIops())
		assert.Equal(t, uint32(278), tiflashStorage.GetThroughputMibps())

		// check the pd component.
		pdComp, ok := lo.Find(
			resCluster.GetClusterComponents(),
			func(comp *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return comp.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, entity.BuildSpuKey(globaldedicatedv1.ComponentType_PD, "8c8g", ""), pdComp.GetSpuKey())
		assert.Equal(t, "pd-8c8g-arm-data:gp3-dynamic", pdComp.GetSkuKey())
		assert.Equal(t, "8c8g", pdComp.GetSpuName())
		assert.Equal(t, "8c8g", pdComp.GetSpuDisplayName())
		assert.Equal(t, uint32(3), pdComp.GetCount())
		assert.NotEmpty(t, tidbComp.GetTidbGroupId())
		assert.NotEmpty(t, tidbComp.GetTidbGroupDisplayName())
		assert.True(t, tidbComp.GetIsDefaultTidbGroup())
		assert.Equal(t, "instance_type_8c8g_arm_6g.", pdComp.GetInstanceType())
		assert.Equal(t, uint32(8), pdComp.GetCpuNum())
		assert.Equal(t, uint32(8), pdComp.GetMemorySizeGi())
		assert.Equal(t, globaldedicatedv1.InstanceMu_ARM64, pdComp.GetInstanceCpuArch())

		// check the dag instance db record.
		dagIns, err_ := svc.workflow.ListRunningDagInstances(
			t.Context(),
			[]string{common.CreateCluster},
			strconv.FormatUint(resCluster.GetClusterId(), 10),
		)
		require.NoError(t, err_)
		assert.Len(t, dagIns, 1)
	})

	t.Run("2c spu special business rule", func(t *testing.T) {
		t.Run("2c tidb 2c tikv", func(t *testing.T) {
			req := &globaldedicatedv1.CreateClusterRequest{
				Cluster: &globaldedicatedv1.Cluster{
					ClusterId:       uint64(rand.Int64()),
					OrganizationId:  TenantID,
					ProjectId:       ProjectID,
					CloudProviderId: globalcommonv1.CloudProvider_AWS,
					RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
					Port:            4000,
					ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
						{
							ComponentType: globaldedicatedv1.ComponentType_TIDB,
							SpuName:       "2c2g",
							Count:         lo.ToPtr(uint32(1)),
						},
						{
							ComponentType: globaldedicatedv1.ComponentType_TIKV,
							SpuName:       "2c2g",
							Count:         lo.ToPtr(uint32(3)),
							Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
								globaldedicatedv1.Sku_DATA.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
									SizeGi:     555,
								},
								globaldedicatedv1.Sku_RAFT_LOG.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
								},
							},
						},
					},
				},
			}
			_, err := svc.Create(t.Context(), req)
			require.NoError(t, err)
		})

		t.Run("2c tidb and 8c tikv not allowed", func(t *testing.T) {
			req := &globaldedicatedv1.CreateClusterRequest{
				Cluster: &globaldedicatedv1.Cluster{
					ClusterId:       uint64(rand.Int64()),
					OrganizationId:  TenantID,
					ProjectId:       ProjectID,
					CloudProviderId: globalcommonv1.CloudProvider_AWS,
					RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
					Port:            4000,
					ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
						{
							ComponentType: globaldedicatedv1.ComponentType_TIDB,
							SpuName:       "2c2g",
							Count:         lo.ToPtr(uint32(1)),
						},
						{
							ComponentType: globaldedicatedv1.ComponentType_TIKV,
							SpuName:       "8c8g",
							Count:         lo.ToPtr(uint32(3)),
							Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
								globaldedicatedv1.Sku_DATA.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
									SizeGi:     555,
								},
								globaldedicatedv1.Sku_RAFT_LOG.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
								},
							},
						},
					},
				},
			}
			_, err := svc.Create(t.Context(), req)
			require.Equal(
				t,
				"if the vCPUs of TiDB or TiKV component is 2 or 4, then their vCPUs need to be the same",
				err.ErrMsg(),
			)
		})

		t.Run("2c tidb 2c tikv, tiflash not allowed", func(t *testing.T) {
			req := &globaldedicatedv1.CreateClusterRequest{
				Cluster: &globaldedicatedv1.Cluster{
					ClusterId:       uint64(rand.Int64()),
					OrganizationId:  TenantID,
					ProjectId:       ProjectID,
					CloudProviderId: globalcommonv1.CloudProvider_AWS,
					RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
					Port:            4000,
					ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
						{
							ComponentType: globaldedicatedv1.ComponentType_TIDB,
							SpuName:       "2c2g",
							Count:         lo.ToPtr(uint32(1)),
						},
						{
							ComponentType: globaldedicatedv1.ComponentType_TIKV,
							SpuName:       "2c2g",
							Count:         lo.ToPtr(uint32(3)),
							Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
								globaldedicatedv1.Sku_DATA.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
									SizeGi:     555,
								},
								globaldedicatedv1.Sku_RAFT_LOG.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
								},
							},
						},
						{
							ComponentType: globaldedicatedv1.ComponentType_TIFLASH,
							SpuName:       "8c8g",
							Count:         lo.ToPtr(uint32(1)),
							Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
								globaldedicatedv1.Sku_DATA.String(): {
									VolumeType:      lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
									SizeGi:          555,
									Iops:            555,
									ThroughputMibps: 555,
								},
								globaldedicatedv1.Sku_RAFT_LOG.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
								},
							},
						},
					},
				},
			}
			_, err := svc.Create(t.Context(), req)
			require.Equal(
				t,
				"if the vCPUs of TiDB or TiKV component is 2 or 4, then the cluster does not support TiFlash",
				err.ErrMsg(),
			)
		})
	})

	t.Run("gcp tidb and tikv", func(t *testing.T) {
		req := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_GCP,
				RegionId:        globalcommonv1.Region_GCP_US_WEST1,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_PD_SSD),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
							},
						},
					},
				},
			},
		}
		resCluster, err := svc.Create(t.Context(), req)
		require.NoError(t, err)

		// check the tidb component.
		tidbComp, ok := lo.Find(
			resCluster.GetClusterComponents(),
			func(comp *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return comp.GetComponentType() == globaldedicatedv1.ComponentType_TIDB
			},
		)
		require.True(t, ok)
		assert.Equal(t, entity.BuildSpuKey(globaldedicatedv1.ComponentType_TIDB, "8c8g", ""), tidbComp.GetSpuKey())
		assert.Equal(t, "tidb-8c8g-gcp", tidbComp.GetSkuKey())
		assert.Equal(t, "8c8g", tidbComp.GetSpuName())
		assert.Equal(t, "8c8g", tidbComp.GetSpuDisplayName())
		assert.Equal(t, uint32(1), tidbComp.GetCount())
		assert.NotEmpty(t, tidbComp.GetTidbGroupId())
		assert.NotEmpty(t, tidbComp.GetTidbGroupDisplayName())
		assert.True(t, tidbComp.GetIsDefaultTidbGroup())
		assert.Equal(t, "instance_type_8c8g_gcp", tidbComp.GetInstanceType())
		assert.Equal(t, uint32(8), tidbComp.GetCpuNum())
		assert.Equal(t, uint32(8), tidbComp.GetMemorySizeGi())
		assert.Equal(t, globaldedicatedv1.InstanceMu_AMD64, tidbComp.GetInstanceCpuArch())

		// check the tikv component.
		tikvComp, ok := lo.Find(
			resCluster.GetClusterComponents(),
			func(comp *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return comp.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
			},
		)
		require.True(t, ok)
		assert.Equal(t, entity.BuildSpuKey(globaldedicatedv1.ComponentType_TIKV, "8c8g", ""), tikvComp.GetSpuKey())
		assert.Equal(t, "tikv-8c8g-gcp-data:pd-ssd-raft:none-fixed", tikvComp.GetSkuKey())
		assert.Equal(t, "8c8g", tikvComp.GetSpuName())
		assert.Equal(t, "8c8g", tikvComp.GetSpuDisplayName())
		assert.Equal(t, uint32(3), tikvComp.GetCount())
		assert.NotEmpty(t, tidbComp.GetTidbGroupId())
		assert.NotEmpty(t, tidbComp.GetTidbGroupDisplayName())
		assert.True(t, tidbComp.GetIsDefaultTidbGroup())
		assert.Equal(t, "instance_type_8c8g_gcp", tikvComp.GetInstanceType())
		assert.Equal(t, uint32(8), tikvComp.GetCpuNum())
		assert.Equal(t, uint32(8), tikvComp.GetMemorySizeGi())
		assert.Equal(t, globaldedicatedv1.InstanceMu_AMD64, tikvComp.GetInstanceCpuArch())
		// Check storage configuration
		tikvStorage, exists := tikvComp.GetStorages()[globaldedicatedv1.Sku_DATA.String()]
		require.True(t, exists)
		assert.Equal(t, globaldedicatedv1.StorageMu_PD_SSD, tikvStorage.GetVolumeType())
		assert.Equal(t, uint32(555), tikvStorage.GetSizeGi())
		// raft disk is not set, so it should be nil
		assert.Nil(t, tikvComp.GetStorages()[globaldedicatedv1.Sku_RAFT_LOG.String()])

		// check the pd component.
		pdComp, ok := lo.Find(
			resCluster.GetClusterComponents(),
			func(comp *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return comp.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, entity.BuildSpuKey(globaldedicatedv1.ComponentType_PD, "8c8g", ""), pdComp.GetSpuKey())
		assert.Equal(t, "pd-8c8g-gcp-data:pd-ssd-fixed", pdComp.GetSkuKey())
		assert.Equal(t, "8c8g", pdComp.GetSpuName())
		assert.Equal(t, "8c8g", pdComp.GetSpuDisplayName())
		assert.Equal(t, uint32(3), pdComp.GetCount())
		assert.NotEmpty(t, tidbComp.GetTidbGroupId())
		assert.NotEmpty(t, tidbComp.GetTidbGroupDisplayName())
		assert.True(t, tidbComp.GetIsDefaultTidbGroup())
		assert.Equal(t, "instance_type_8c8g_gcp", pdComp.GetInstanceType())
		assert.Equal(t, uint32(8), pdComp.GetCpuNum())
		assert.Equal(t, uint32(8), pdComp.GetMemorySizeGi())
		assert.Equal(t, globaldedicatedv1.InstanceMu_AMD64, pdComp.GetInstanceCpuArch())

		// check the dag instance db record.
		dagIns, err_ := svc.workflow.ListRunningDagInstances(
			t.Context(),
			[]string{common.CreateCluster},
			strconv.FormatUint(resCluster.GetClusterId(), 10),
		)
		require.NoError(t, err_)
		assert.Len(t, dagIns, 1)
	})

	t.Run("sku priority should work", func(t *testing.T) {
		req := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
						},
					},
				},
			},
		}
		resCluster, err := svc.Create(t.Context(), req)
		require.NoError(t, err)

		// check the tikv component.
		tikvComp, ok := lo.Find(
			resCluster.GetClusterComponents(),
			func(comp *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return comp.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
			},
		)
		require.True(t, ok)
		assert.Equal(t, "tikv-8c8g-arm-data:gp3-raft:io2-dynamic", tikvComp.GetSkuKey())
	})

	t.Run("tidb with tiproxy", func(t *testing.T) {
		t.Run("2c tidb 2c tiproxy, should failed", func(t *testing.T) {
			req := &globaldedicatedv1.CreateClusterRequest{
				Cluster: &globaldedicatedv1.Cluster{
					ClusterId:       uint64(rand.Int64()),
					OrganizationId:  TenantID,
					ProjectId:       ProjectID,
					CloudProviderId: globalcommonv1.CloudProvider_AWS,
					RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
					Port:            4000,
					ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
						{
							ComponentType: globaldedicatedv1.ComponentType_TIDB,
							SpuName:       "2c2g",
							Count:         lo.ToPtr(uint32(2)),
						},
						{
							ComponentType: globaldedicatedv1.ComponentType_TIPROXY,
							SpuName:       "2c2g",
							Count:         lo.ToPtr(uint32(3)),
						},
						{
							ComponentType: globaldedicatedv1.ComponentType_TIKV,
							SpuName:       "2c2g",
							Count:         lo.ToPtr(uint32(3)),
							Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
								globaldedicatedv1.Sku_DATA.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
									SizeGi:     555,
								},
								globaldedicatedv1.Sku_RAFT_LOG.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
								},
							},
						},
					},
				},
			}
			_, err := svc.Create(t.Context(), req)
			require.ErrorContains(t, err, "When TiDB component has 2 vCPUs, the TiProxy component is not supported")
		})
		t.Run("4c tidb 2c tiproxy should be ok", func(t *testing.T) {
			req := &globaldedicatedv1.CreateClusterRequest{
				Cluster: &globaldedicatedv1.Cluster{
					ClusterId:       uint64(rand.Int64()),
					OrganizationId:  TenantID,
					ProjectId:       ProjectID,
					CloudProviderId: globalcommonv1.CloudProvider_AWS,
					RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
					Port:            4000,
					ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
						{
							ComponentType: globaldedicatedv1.ComponentType_TIDB,
							SpuName:       "8c8g",
							Count:         lo.ToPtr(uint32(2)),
						},
						{
							ComponentType: globaldedicatedv1.ComponentType_TIPROXY,
							SpuName:       "2c2g",
							Count:         lo.ToPtr(uint32(3)),
						},
						{
							ComponentType: globaldedicatedv1.ComponentType_TIKV,
							SpuName:       "8c8g",
							Count:         lo.ToPtr(uint32(3)),
							Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
								globaldedicatedv1.Sku_DATA.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
									SizeGi:     555,
								},
								globaldedicatedv1.Sku_RAFT_LOG.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
								},
							},
						},
					},
				},
			}
			resCluster, err := svc.Create(t.Context(), req)
			require.NoError(t, err)
			// check the tidb component.
			tiproxy, ok := lo.Find(
				resCluster.GetClusterComponents(),
				func(comp *globaldedicatedv1.Cluster_ClusterComponent) bool {
					return comp.GetComponentType() == globaldedicatedv1.ComponentType_TIPROXY
				},
			)
			require.True(t, ok)
			assert.Equal(
				t,
				entity.BuildSpuKey(globaldedicatedv1.ComponentType_TIPROXY, "2c2g", ""),
				tiproxy.GetSpuKey(),
			)
			assert.Equal(t, "tiproxy-2c2g-arm", tiproxy.GetSkuKey())
			assert.Equal(t, "2c2g", tiproxy.GetSpuName())
			assert.Equal(t, "2c2g", tiproxy.GetSpuDisplayName())
			assert.Equal(t, uint32(3), tiproxy.GetCount())
			assert.NotEmpty(t, tiproxy.GetTidbGroupId())
		})
		t.Run("tiproxy on gcp is not supported", func(t *testing.T) {
			req := &globaldedicatedv1.CreateClusterRequest{
				Cluster: &globaldedicatedv1.Cluster{
					ClusterId:       uint64(rand.Int64()),
					OrganizationId:  TenantID,
					ProjectId:       ProjectID,
					CloudProviderId: globalcommonv1.CloudProvider_GCP,
					RegionId:        globalcommonv1.Region_GCP_US_WEST1,
					Port:            4000,
					ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
						{
							ComponentType: globaldedicatedv1.ComponentType_TIDB,
							SpuName:       "8c8g",
							Count:         lo.ToPtr(uint32(2)),
						},
						{
							ComponentType: globaldedicatedv1.ComponentType_TIPROXY,
							SpuName:       "2c2g",
							Count:         lo.ToPtr(uint32(3)),
						},
						{
							ComponentType: globaldedicatedv1.ComponentType_TIKV,
							SpuName:       "8c8g",
							Count:         lo.ToPtr(uint32(3)),
							Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
								globaldedicatedv1.Sku_DATA.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
									SizeGi:     555,
								},
								globaldedicatedv1.Sku_RAFT_LOG.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
								},
							},
						},
					},
				},
			}
			_, err := svc.Create(t.Context(), req)
			require.ErrorContains(t, err, "tiproxy component is not supported in gcp cloud provider")
		})
	})
}

func TestDynamicIopsThroughput(t *testing.T) {
	ctrl := gomock.NewController(t, gomock.WithOverridableExpectations())

	encryptor := mock_remoteencryptor.NewMockClient(ctrl)
	encryptor.EXPECT().EncryptBase64(gomock.Any(), gomock.Any()).Return("encrypted", nil).AnyTimes()
	encryptor.EXPECT().DecryptBase64(gomock.Any(), gomock.Any()).Return("password", nil).AnyTimes()

	accountCli := mock_account.NewMockClient(ctrl)
	accountCli.EXPECT().GetProject(gomock.Any(), uint64(ProjectID)).Return(&account.ProjectInfo{
		ID:             ProjectID,
		TenantID:       TenantID,
		Name:           "test",
		AwsCmekEnabled: false,
	}, nil).AnyTimes()
	accountCli.EXPECT().GetTenant(gomock.Any(), uint64(TenantID)).Return(&account.TenantInfo{
		ID:   TenantID,
		Name: "test",
		Plan: account.TenantPlanPOC,
	}, nil).AnyTimes()

	regionalCli := mock_regionserver.NewMockClient(ctrl)
	regionalCli.EXPECT().ListVpcs(gomock.Any(), gomock.Any(), &regionaldedicatedv1.ListVpcsRequest{
		TenantId:  strconv.FormatUint(TenantID, 10),
		ProjectId: strconv.FormatUint(ProjectID, 10),
		Statuses: []regionaldedicatedv1.Vpc_Status{
			regionaldedicatedv1.Vpc_ACTIVE,
			regionaldedicatedv1.Vpc_CREATING,
			regionaldedicatedv1.Vpc_FAILED,
		},
	}).Return(&regionaldedicatedv1.ListVpcsResponse{
		Vpcs: []*regionaldedicatedv1.Vpc{
			{
				VpcId:      "vpc-xxxxxxxxxxxxxxxxx",
				TenantIds:  []string{strconv.FormatUint(TenantID, 10)},
				ProjectIds: []string{strconv.FormatUint(ProjectID, 10)},
				CidrBlock:  "0.0.0.0/0",
			},
		},
	}, nil).AnyTimes()

	metricCli := mock_clustermetric.NewMockClient(ctrl)

	billingCli := mock_billing.NewMockClient(ctrl)
	billingCli.EXPECT().
		BillingActionCheck(gomock.Any(), billing.Action_Create, uint64(TenantID), gomock.Any()).
		Return(&billing.ActionCheckResult{Pass: true}, nil).
		AnyTimes()

	clusterConfigCli := mock_clusterconfig.NewMockClient(ctrl)
	clusterConfigCli.EXPECT().GetClusterVersion(gomock.Any(), gomock.Any()).Return(ClusterVersion, nil).AnyTimes()
	clusterConfigCli.EXPECT().
		ListComponentConfigurations(gomock.Any(), gomock.Any()).
		Return([]*regionaldedicatedv1.ComponentConfiguration{}, nil).
		AnyTimes()

	svc := newTestService(encryptor, accountCli, regionalCli, metricCli, billingCli, clusterConfigCli, true)

	t.Run("data:gp3 raft:none fixed-iops fixed-throughput", func(t *testing.T) {
		req := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
							},
						},
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIFLASH,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType:      lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:          555,
								Iops:            555,
								ThroughputMibps: 555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
							},
						},
					},
				},
			},
		}
		resCluster, err := svc.Create(t.Context(), req)
		require.NoError(t, err)

		// Check the iops and throughput.
		tikvComp, _ := lo.Find(
			resCluster.GetClusterComponents(),
			func(comp *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return comp.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
			},
		)
		dataStore := tikvComp.GetStorages()[globaldedicatedv1.Sku_DATA.String()]
		assert.Equal(t, uint32(2), dataStore.GetIops())
		assert.Equal(t, uint32(1), dataStore.GetThroughputMibps())
	})

	t.Run("dynamic-iops dynamic-throughput", func(t *testing.T) {
		t.Run("in the range of [min, max]", func(t *testing.T) {
			req := &globaldedicatedv1.CreateClusterRequest{
				Cluster: &globaldedicatedv1.Cluster{
					ClusterId:       uint64(rand.Int64()),
					OrganizationId:  TenantID,
					ProjectId:       ProjectID,
					CloudProviderId: globalcommonv1.CloudProvider_AWS,
					RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
					Port:            4000,
					ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
						{
							ComponentType: globaldedicatedv1.ComponentType_TIDB,
							SpuName:       "8c8g",
							Count:         lo.ToPtr(uint32(1)),
						},
						{
							ComponentType: globaldedicatedv1.ComponentType_TIKV,
							SpuName:       "8c8g",
							Count:         lo.ToPtr(uint32(3)),
							Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
								globaldedicatedv1.Sku_DATA.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
									SizeGi:     555,
								},
								globaldedicatedv1.Sku_RAFT_LOG.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_IO2),
									SizeGi:     555,
								},
							},
						},
						{
							ComponentType: globaldedicatedv1.ComponentType_TIFLASH,
							SpuName:       "8c8g",
							Count:         lo.ToPtr(uint32(1)),
							Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
								globaldedicatedv1.Sku_DATA.String(): {
									VolumeType:      lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
									SizeGi:          555,
									Iops:            555,
									ThroughputMibps: 555,
								},
								globaldedicatedv1.Sku_RAFT_LOG.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
								},
							},
						},
					},
				},
			}
			resCluster, err := svc.Create(t.Context(), req)
			require.NoError(t, err)

			// Check the iops and throughput.
			tikvComp, _ := lo.Find(
				resCluster.GetClusterComponents(),
				func(comp *globaldedicatedv1.Cluster_ClusterComponent) bool {
					return comp.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
				},
			)
			assert.Equal(t, "tikv-8c8g-arm-data:gp3-raft:io2-dynamic", tikvComp.GetSkuKey())
			dataStore := tikvComp.GetStorages()[globaldedicatedv1.Sku_DATA.String()]
			assert.Equal(t, uint32(1110), dataStore.GetIops())
			assert.Equal(t, uint32(278), dataStore.GetThroughputMibps())
			raftStore := tikvComp.GetStorages()[globaldedicatedv1.Sku_RAFT_LOG.String()]
			assert.Equal(t, uint32(11100), raftStore.GetIops())
			assert.Equal(t, uint32(0), raftStore.GetThroughputMibps())
		})

		t.Run("out of the range of [min, max]", func(t *testing.T) {
			req := &globaldedicatedv1.CreateClusterRequest{
				Cluster: &globaldedicatedv1.Cluster{
					ClusterId:       uint64(rand.Int64()),
					OrganizationId:  TenantID,
					ProjectId:       ProjectID,
					CloudProviderId: globalcommonv1.CloudProvider_AWS,
					RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
					Port:            4000,
					ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
						{
							ComponentType: globaldedicatedv1.ComponentType_TIDB,
							SpuName:       "8c8g",
							Count:         lo.ToPtr(uint32(1)),
						},
						{
							ComponentType: globaldedicatedv1.ComponentType_TIKV,
							SpuName:       "8c8g",
							Count:         lo.ToPtr(uint32(3)),
							Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
								globaldedicatedv1.Sku_DATA.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
									SizeGi:     99999,
								},
								globaldedicatedv1.Sku_RAFT_LOG.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_IO2),
									SizeGi:     55555,
								},
							},
						},
						{
							ComponentType: globaldedicatedv1.ComponentType_TIFLASH,
							SpuName:       "8c8g",
							Count:         lo.ToPtr(uint32(1)),
							Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
								globaldedicatedv1.Sku_DATA.String(): {
									VolumeType:      lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
									SizeGi:          555,
									Iops:            555,
									ThroughputMibps: 555,
								},
								globaldedicatedv1.Sku_RAFT_LOG.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
								},
							},
						},
					},
				},
			}
			resCluster, err := svc.Create(t.Context(), req)
			require.NoError(t, err)

			// Check the iops and throughput.
			tikvComp, _ := lo.Find(
				resCluster.GetClusterComponents(),
				func(comp *globaldedicatedv1.Cluster_ClusterComponent) bool {
					return comp.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
				},
			)
			assert.Equal(t, "tikv-8c8g-arm-data:gp3-raft:io2-dynamic", tikvComp.GetSkuKey())
			dataStore := tikvComp.GetStorages()[globaldedicatedv1.Sku_DATA.String()]
			assert.Equal(t, uint32(99999), dataStore.GetIops())
			assert.Equal(t, uint32(999), dataStore.GetThroughputMibps())
			raftStore := tikvComp.GetStorages()[globaldedicatedv1.Sku_RAFT_LOG.String()]
			assert.Equal(t, uint32(99999), raftStore.GetIops())
		})

		t.Run("max is 0", func(t *testing.T) {
			req := &globaldedicatedv1.CreateClusterRequest{
				Cluster: &globaldedicatedv1.Cluster{
					ClusterId:       uint64(rand.Int64()),
					OrganizationId:  TenantID,
					ProjectId:       ProjectID,
					CloudProviderId: globalcommonv1.CloudProvider_AWS,
					RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
					Port:            4000,
					ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
						{
							ComponentType: globaldedicatedv1.ComponentType_TIDB,
							SpuName:       "8c8g",
							Count:         lo.ToPtr(uint32(1)),
						},
						{
							ComponentType: globaldedicatedv1.ComponentType_TIKV,
							SpuName:       "7c7g",
							Count:         lo.ToPtr(uint32(3)),
							Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
								globaldedicatedv1.Sku_DATA.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
									SizeGi:     99999,
									Iops:       99999,
								},
								globaldedicatedv1.Sku_RAFT_LOG.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
								},
							},
						},
						{
							ComponentType: globaldedicatedv1.ComponentType_TIFLASH,
							SpuName:       "8c8g",
							Count:         lo.ToPtr(uint32(1)),
							Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
								globaldedicatedv1.Sku_DATA.String(): {
									VolumeType:      lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
									SizeGi:          555,
									Iops:            555,
									ThroughputMibps: 555,
								},
								globaldedicatedv1.Sku_RAFT_LOG.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
								},
							},
						},
					},
				},
			}
			resCluster, err := svc.Create(t.Context(), req)
			require.NoError(t, err)
			tikvComp, _ := lo.Find(
				resCluster.GetClusterComponents(),
				func(comp *globaldedicatedv1.Cluster_ClusterComponent) bool {
					return comp.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
				},
			)
			assert.Equal(t, "tikv-7c7g-arm-data:gp3-raft:none-fixed", tikvComp.GetSkuKey())
			dataStore := tikvComp.GetStorages()[globaldedicatedv1.Sku_DATA.String()]
			assert.Equal(t, uint32(99999), dataStore.GetIops())
		})

		t.Run("with additional iops", func(t *testing.T) {
			req := &globaldedicatedv1.CreateClusterRequest{
				Cluster: &globaldedicatedv1.Cluster{
					ClusterId:       uint64(rand.Int64()),
					OrganizationId:  TenantID,
					ProjectId:       ProjectID,
					CloudProviderId: globalcommonv1.CloudProvider_AWS,
					RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
					Port:            4000,
					ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
						{
							ComponentType: globaldedicatedv1.ComponentType_TIDB,
							SpuName:       "8c8g",
							Count:         lo.ToPtr(uint32(1)),
						},
						{
							ComponentType: globaldedicatedv1.ComponentType_TIKV,
							SpuName:       "8c8g",
							Count:         lo.ToPtr(uint32(3)),
							Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
								globaldedicatedv1.Sku_DATA.String(): {
									VolumeType:     lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
									SizeGi:         555,
									AdditionalIops: lo.ToPtr(uint32(1)),
								},
								globaldedicatedv1.Sku_RAFT_LOG.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_IO2),
									SizeGi:     555,
								},
							},
						},
						{
							ComponentType: globaldedicatedv1.ComponentType_TIFLASH,
							SpuName:       "8c8g",
							Count:         lo.ToPtr(uint32(1)),
							Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
								globaldedicatedv1.Sku_DATA.String(): {
									VolumeType:      lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
									SizeGi:          555,
									Iops:            555,
									ThroughputMibps: 555,
								},
								globaldedicatedv1.Sku_RAFT_LOG.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
								},
							},
						},
					},
				},
			}
			resCluster, err := svc.Create(t.Context(), req)
			require.NoError(t, err)

			// Check the iops and throughput.
			tikvComp, _ := lo.Find(
				resCluster.GetClusterComponents(),
				func(comp *globaldedicatedv1.Cluster_ClusterComponent) bool {
					return comp.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
				},
			)
			dataStore := tikvComp.GetStorages()[globaldedicatedv1.Sku_DATA.String()]
			assert.Equal(t, uint32(1111), dataStore.GetIops())
			assert.Equal(t, uint32(278), dataStore.GetThroughputMibps())
			raftStore := tikvComp.GetStorages()[globaldedicatedv1.Sku_RAFT_LOG.String()]
			assert.Equal(t, uint32(11100), raftStore.GetIops())
			assert.Equal(t, uint32(0), raftStore.GetThroughputMibps())
		})
	})
}

func TestCreateX86Cluster(t *testing.T) {
	ctrl := gomock.NewController(t, gomock.WithOverridableExpectations())

	encryptor := mock_remoteencryptor.NewMockClient(ctrl)
	encryptor.EXPECT().EncryptBase64(gomock.Any(), gomock.Any()).Return("encrypted", nil).AnyTimes()
	encryptor.EXPECT().DecryptBase64(gomock.Any(), gomock.Any()).Return("password", nil).AnyTimes()

	accountCli := mock_account.NewMockClient(ctrl)
	accountCli.EXPECT().GetProject(gomock.Any(), uint64(ProjectID)).Return(&account.ProjectInfo{
		ID:             ProjectID,
		TenantID:       TenantID,
		Name:           "test",
		AwsCmekEnabled: false,
	}, nil).AnyTimes()
	accountCli.EXPECT().GetTenant(gomock.Any(), uint64(TenantID)).Return(&account.TenantInfo{
		ID:   TenantID,
		Name: "test",
		Plan: account.TenantPlanPOC,
	}, nil).AnyTimes()

	regionalCli := mock_regionserver.NewMockClient(ctrl)
	regionalCli.EXPECT().ListVpcs(gomock.Any(), gomock.Any(), &regionaldedicatedv1.ListVpcsRequest{
		TenantId:  strconv.FormatUint(TenantID, 10),
		ProjectId: strconv.FormatUint(ProjectID, 10),
		Statuses: []regionaldedicatedv1.Vpc_Status{
			regionaldedicatedv1.Vpc_ACTIVE,
			regionaldedicatedv1.Vpc_CREATING,
			regionaldedicatedv1.Vpc_FAILED,
		},
	}).Return(&regionaldedicatedv1.ListVpcsResponse{
		Vpcs: []*regionaldedicatedv1.Vpc{
			{
				VpcId:      "vpc-xxxxxxxxxxxxxxxxx",
				TenantIds:  []string{strconv.FormatUint(TenantID, 10)},
				ProjectIds: []string{strconv.FormatUint(ProjectID, 10)},
				CidrBlock:  "0.0.0.0/0",
			},
		},
	}, nil).AnyTimes()

	metricCli := mock_clustermetric.NewMockClient(ctrl)

	billingCli := mock_billing.NewMockClient(ctrl)
	billingCli.EXPECT().
		BillingActionCheck(gomock.Any(), billing.Action_Create, uint64(TenantID), gomock.Any()).
		Return(&billing.ActionCheckResult{Pass: true}, nil).
		AnyTimes()

	clusterConfigCli := mock_clusterconfig.NewMockClient(ctrl)
	clusterConfigCli.EXPECT().GetClusterVersion(gomock.Any(), gomock.Any()).Return(ClusterVersion, nil).AnyTimes()
	clusterConfigCli.EXPECT().
		ListComponentConfigurations(gomock.Any(), gomock.Any()).
		Return([]*regionaldedicatedv1.ComponentConfiguration{}, nil).
		AnyTimes()

	svc := newTestService(encryptor, accountCli, regionalCli, metricCli, billingCli, clusterConfigCli, false)

	t.Run("create x86 cluster", func(t *testing.T) {
		req := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
							},
						},
					},
				},
			},
		}
		resCluster, err := svc.Create(t.Context(), req)
		require.NoError(t, err)

		// check the tikv component.
		tikvComp, ok := lo.Find(
			resCluster.GetClusterComponents(),
			func(comp *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return comp.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
			},
		)
		require.True(t, ok)
		assert.Equal(t, "tikv-8c8g-x86-data:gp3-raft:none-fixed", tikvComp.GetSkuKey())

		tidbComp, ok := lo.Find(
			resCluster.GetClusterComponents(),
			func(comp *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return comp.GetComponentType() == globaldedicatedv1.ComponentType_TIDB
			},
		)
		require.True(t, ok)
		assert.Equal(t, "tidb-8c8g-x86", tidbComp.GetSkuKey())
	})
}