package impl

import (
	"context"
	"testing"

	"github.com/AlekSi/pointer"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/suite"
	"github.com/tidbcloud/dataflow-service/app/internal/feature_gate"
	feature_gate_mock "github.com/tidbcloud/dataflow-service/app/internal/feature_gate/mock"
	config_manager_mock "github.com/tidbcloud/dataflow-service/app/models/config_manager/mock"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	ticdc_spec_mock "github.com/tidbcloud/dataflow-service/app/models/ticdc_spec/mock"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	repo_mock "github.com/tidbcloud/dataflow-service/app/repo/mock"
	"github.com/tidbcloud/dataflow-service/app/utils"
	"github.com/tidbcloud/dataflow-service/internal/config"
	grpcli_mock "github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli/mock"
	regionalapi "github.com/tidbcloud/dataflow-service/pkg/regional-api"
	regionalapi_mock "github.com/tidbcloud/dataflow-service/pkg/regional-api/mock"
	ticdc_v1 "github.com/tidbcloud/infra-provider/api/tidbcloud/ticdc-service/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db/mock_db"
	global_commonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	tidb_mgmt_servicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/mock/gomock"
	"gorm.io/gorm"
)

type clusterModelRegionalImplTestSuite struct {
	suite.Suite

	mockInstanceTypeModel      *ticdc_spec_mock.CDCInstanceTypeModel
	mockTiCDCClusterCli        *regionalapi_mock.TiCDCClusterClient
	mockTiCDCClusterRepo       *repo_mock.TiCDCClusterRepo
	mockTiCDCCustomStorageRepo *repo_mock.TiCDCCustomStorageRepo
	mockConfigManager          *config_manager_mock.TiCDCClusterConfigManager
	mockRegionCli              *grpcli_mock.RegionGrpcClient
	mockFeatureGate            *feature_gate_mock.FeatureGate

	model *clusterModelRegionalImpl
}

func (s *clusterModelRegionalImplTestSuite) SetupSuite() {
	s.mockInstanceTypeModel = ticdc_spec_mock.NewCDCInstanceTypeModel(s.T())
	s.mockTiCDCClusterCli = regionalapi_mock.NewTiCDCClusterClient(s.T())
	s.mockTiCDCClusterRepo = repo_mock.NewTiCDCClusterRepo(s.T())
	s.mockTiCDCCustomStorageRepo = repo_mock.NewTiCDCCustomStorageRepo(s.T())
	s.mockConfigManager = config_manager_mock.NewTiCDCClusterConfigManager(s.T())
	s.mockRegionCli = grpcli_mock.NewRegionGrpcClient(s.T())
	s.mockFeatureGate = feature_gate_mock.NewFeatureGate(s.T())
	s.model = &clusterModelRegionalImpl{
		instanceTypeModel:      s.mockInstanceTypeModel,
		ticdcClusterCli:        s.mockTiCDCClusterCli,
		ticdcClusterRepo:       s.mockTiCDCClusterRepo,
		ticdcCustomStorageRepo: s.mockTiCDCCustomStorageRepo,
		configManager:          s.mockConfigManager,
		regionCli:              s.mockRegionCli,
		featureGate:            s.mockFeatureGate,
	}
}

func (s *clusterModelRegionalImplTestSuite) TestConvertRegionalTiCDCClusterConfig() {
	testcases := []struct {
		configData   *domain.TiCDCConfigData
		expectedConf *ticdc_v1.TiCDCCluster_Configuration
	}{
		{
			configData: &domain.TiCDCConfigData{
				BaseImage: nil,
				Tag:       nil,
				Config:    nil,
			},
			expectedConf: &ticdc_v1.TiCDCCluster_Configuration{
				BaseImage: "",
				Tag:       "",
				Config:    "",
			},
		},
		{
			configData: &domain.TiCDCConfigData{
				BaseImage: nil,
				Tag:       nil,
				Config:    map[string]interface{}{},
			},
			expectedConf: &ticdc_v1.TiCDCCluster_Configuration{
				BaseImage: "",
				Tag:       "",
				Config:    "",
			},
		},
		{
			configData: &domain.TiCDCConfigData{
				BaseImage: nil,
				Tag:       nil,
				Config: map[string]interface{}{
					"a": 123,
					"b": "abc",
				},
			},
			expectedConf: &ticdc_v1.TiCDCCluster_Configuration{
				BaseImage: "",
				Tag:       "",
				Config:    "a = 123\nb = 'abc'\n",
			},
		},
		{
			configData: &domain.TiCDCConfigData{
				BaseImage: pointer.ToString("ticdc"),
				Tag:       nil,
				Config: map[string]interface{}{
					"a": 123,
					"b": "abc",
				},
			},
			expectedConf: &ticdc_v1.TiCDCCluster_Configuration{
				BaseImage: "ticdc",
				Tag:       "",
				Config:    "a = 123\nb = 'abc'\n",
			},
		},
		{
			configData: &domain.TiCDCConfigData{
				BaseImage: pointer.ToString("ticdc"),
				Tag:       pointer.ToString("v7.5.0"),
				Config: map[string]interface{}{
					"a": 123,
					"b": "abc",
				},
			},
			expectedConf: &ticdc_v1.TiCDCCluster_Configuration{
				BaseImage: "ticdc",
				Tag:       "v7.5.0",
				Config:    "a = 123\nb = 'abc'\n",
			},
		},
	}

	for _, tc := range testcases {
		conf, err := s.model.convertRegionalTiCDCClusterConfig(tc.configData)
		s.NoError(err)
		s.Equal(tc.expectedConf, conf)
	}
}

func (s *clusterModelRegionalImplTestSuite) TestConvertRegionalTiCDCCluster() {
	ticdcClusterEnt := &entity.TiCDCCluster{
		Updatable: entity.Updatable{
			Base: entity.Base{
				ID: 1111,
			},
		},
		TenantID:          1,
		ProjectID:         11,
		ClusterID:         111,
		Name:              "cdc-111",
		BusinessType:      entity.TiCDCClusterBizTypeChangefeed,
		DeployType:        entity.TiCDCClusterDeployTypeShared,
		Version:           "v7.5.0",
		Region:            "us-west-2",
		InfraProviderType: entity.InfraProviderTypeEKS,
		CloudAccountID:    "abc",
		SpecificationCode: "32rcu",
		Status:            entity.TiCDCClusterStatusCreating,
	}
	configData := &domain.TiCDCConfigData{
		BaseImage: pointer.ToString("ticdc"),
		Tag:       pointer.ToString("v7.5.0"),
		Config: map[string]interface{}{
			"a": 123,
			"b": "abc",
		},
	}
	instanceType := &domain.CDCInstanceType{
		Name:             "32rcu",
		RCUNum:           32,
		MachineType:      "m5.4xlarge",
		StorageType:      "gp3",
		Replicas:         2,
		MachineCPUNum:    16,
		MachineMemoryGi:  64,
		MachineStorageGi: 1000,
		IOPS:             4000,
		ThroughputMbs:    400,
	}

	expectedRegionalTiCDCCluster := &ticdc_v1.TiCDCCluster{
		Id:               "cdc-111",
		TenantId:         "1",
		ProjectId:        "11",
		ClusterId:        "111",
		NodeInstanceType: "m5.4xlarge",
		NodeCount:        2,
		Storage: &ticdc_v1.TiCDCCluster_Storage{
			Type:          "gp3",
			SizeGi:        1000,
			Iops:          4000,
			ThroughputMbs: 400,
		},
		Configuration: &ticdc_v1.TiCDCCluster_Configuration{
			BaseImage: "ticdc",
			Tag:       "v7.5.0",
			Config:    "a = 123\nb = 'abc'\n",
		},
	}

	regionalTiCDCCluster, err := s.model.convertRegionalTiCDCCluster(ticdcClusterEnt, configData, instanceType)
	s.NoError(err)
	s.Equal(expectedRegionalTiCDCCluster, regionalTiCDCCluster)
}

func (s *clusterModelRegionalImplTestSuite) TestCreateRegionalTiCDCCluster() {
	ticdcClusterEnt := &entity.TiCDCCluster{
		TenantID: 1,
		Name:     "cdc-111",
		Region:   "us-west-2",
	}
	configData := &domain.TiCDCConfigData{}
	instanceType := &domain.CDCInstanceType{}

	s.Run("call regional error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockTiCDCClusterCli.
			On("CreateTiCDCCluster", mock.Anything, mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		err := s.model.createRegionalTiCDCClusterIgnoreExists(
			context.Background(), ticdcClusterEnt, configData, instanceType)
		s.Error(err)
		s.Equal(mockErr, err)
	})

	s.Run("ticdc cluster exists", func() {
		mockErr := bizerr.New(bizerr.ErrCodeRecordExist, "ticdc cluster already exists")
		call := s.mockTiCDCClusterCli.
			On("CreateTiCDCCluster", mock.Anything, mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		err := s.model.createRegionalTiCDCClusterIgnoreExists(
			context.Background(), ticdcClusterEnt, configData, instanceType)
		s.NoError(err)
	})

	s.Run("call regional successfully", func() {
		regionMatcher := mock.MatchedBy(func(region *regionalapi.Region) bool {
			return region.TenantID == 1 && region.Name == "us-west-2"
		})
		reqMatcher := mock.MatchedBy(func(req *ticdc_v1.CreateTiCDCClusterRequest) bool {
			return req.TicdcClusterId == "cdc-111"
		})
		call := s.mockTiCDCClusterCli.
			On("CreateTiCDCCluster", mock.Anything, mock.Anything, mock.Anything).
			Return(nil, nil)
		defer call.Unset()
		err := s.model.createRegionalTiCDCClusterIgnoreExists(
			context.Background(), ticdcClusterEnt, configData, instanceType)
		s.NoError(err)
		s.mockTiCDCClusterCli.AssertCalled(
			s.T(), "CreateTiCDCCluster", mock.Anything, regionMatcher, reqMatcher)
	})
}

func (s *clusterModelRegionalImplTestSuite) TestCreateReq2Entity() {
	mockLogicalRegionName := "us-west-2-test-1"
	req := &domain.CreateTiCDCClusterReq{
		TenantID:          1,
		ProjectID:         11,
		ClusterID:         111,
		Name:              "cdc-111",
		InfraProviderType: "eks",
		CloudAccountID:    "abc",
		Region:            "us-west-2",
		Version:           "v7.5.0",
		SpecificationCode: "32rcu",
		BizType:           domain.TiCDCClusterBizTypeChangefeed,
		DeployType:        entity.TiCDCClusterDeployTypeShared,
	}
	expectedTiCDCClusterEnt := &entity.TiCDCCluster{
		TenantID:          1,
		ProjectID:         11,
		ClusterID:         111,
		Name:              "cdc-111",
		InfraProviderType: entity.InfraProviderTypeEKS,
		CloudAccountID:    "abc",
		Region:            "us-west-2",
		LogicalRegion:     "us-west-2-test-1",
		Version:           "v7.5.0",
		SpecificationCode: "32rcu",
		BusinessType:      entity.TiCDCClusterBizTypeChangefeed,
		DeployType:        entity.TiCDCClusterDeployTypeShared,
		Status:            entity.TiCDCClusterStatusCreating,
		IsRegional:        true,
	}

	ticdcClusterEnt := s.model.createReq2Entity(req, mockLogicalRegionName)
	s.Equal(expectedTiCDCClusterEnt, ticdcClusterEnt)
}

func makeTicdcClusterEnt4DoCreate() *entity.TiCDCCluster {
	cdc := &entity.TiCDCCluster{
		TenantID:          1,
		ProjectID:         11,
		ClusterID:         111,
		Name:              "cdc-111",
		InfraProviderType: entity.InfraProviderTypeEKS,
		CloudAccountID:    "abc",
		Region:            "us-west-2",
		Version:           "v7.5.0",
		SpecificationCode: "32rcu",
		BusinessType:      entity.TiCDCClusterBizTypeChangefeed,
		DeployType:        entity.TiCDCClusterDeployTypeShared,
		Status:            entity.TiCDCClusterStatusCreating,
	}
	cdc.ID = 1111
	return cdc
}

func (s *clusterModelRegionalImplTestSuite) TestDoCreate() {
	s.Run("invalid version error", func() {
		ticdcClusterEnt := makeTicdcClusterEnt4DoCreate()
		invalidVersionEnt := *ticdcClusterEnt
		invalidVersionEnt.Version = "invalid-version"
		err := s.model.doCreate(context.Background(), &domain.CreateTiCDCClusterReq{}, &invalidVersionEnt)
		s.Error(err)
		s.Contains(err.Error(), "tidb version is invalid")
	})

	s.Run("new kernel enabled - version override and config update", func() {
		ticdcClusterEnt := makeTicdcClusterEnt4DoCreate()

		origiMinNewkernelVersion := config.Global().ChangefeedConfig.MinNewkernelVersion
		originMinNewkernelCompatibleTiDBVersion := config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion
		defer func() {
			config.Global().ChangefeedConfig.MinNewkernelVersion = origiMinNewkernelVersion
			config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = originMinNewkernelCompatibleTiDBVersion
		}()
		config.Global().ChangefeedConfig.MinNewkernelVersion = "v9.0.0-beta.1"
		config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = "v7.5.0"
		// Mock feature gate to return false for blacklist so the default new kernel path is used.
		allowlistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", &feature_gate.QueryFeatureGateReq{
				TenantID:  ticdcClusterEnt.TenantID.Uint64(),
				ProjectID: ticdcClusterEnt.ProjectID.Uint64(),
				ClusterID: ticdcClusterEnt.ClusterID.Uint64(),
			}, feature_gate.TiCDCNewKernelAllowlist).
			Return(true).Maybe()
		defer allowlistCall.Unset()

		blacklistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", &feature_gate.QueryFeatureGateReq{
				TenantID:  ticdcClusterEnt.TenantID.Uint64(),
				ProjectID: ticdcClusterEnt.ProjectID.Uint64(),
				ClusterID: ticdcClusterEnt.ClusterID.Uint64(),
			}, feature_gate.TiCDCNewKernelBlacklist).
			Return(false)
		defer blacklistCall.Unset()

		// Mock save entity
		saveCall := s.mockTiCDCClusterRepo.
			On("Save", mock.Anything, mock.Anything).
			Run(func(args mock.Arguments) {
				cluster := args.Get(1).(*entity.TiCDCCluster)
				s.True(cluster.IsNewKernel)
				s.Equal("v9.0.0-beta.1", cluster.Version)
			}).
			Return(nil)
		defer saveCall.Unset()

		// Mock create config
		createConfCall := s.mockConfigManager.
			On("Create", mock.Anything, mock.Anything).Run(func(args mock.Arguments) {
			req := args.Get(1).(*domain.CreateTiCDCClusterConfigReq)
			s.Equal(uint64(1), req.TenantID)
			s.Equal(uint64(1111), req.TiCDCClusterID)
			s.Equal("v9.0.0-beta.1", req.TiDBClusterVersion)
		}).Return(nil, nil)
		defer createConfCall.Unset()

		// Mock patch config for new kernel
		patchConfCall := s.mockConfigManager.
			On("PatchConfig", mock.Anything, mock.Anything).Run(func(args mock.Arguments) {
			req := args.Get(1).(*domain.PatchTiCDCClusterConfigReq)
			s.Equal(uint64(1111), req.TiCDCClusterID)
			s.True(req.CustomParams["newarch"].Value.GetBoolValue())
		}).Return(nil, nil)
		defer patchConfCall.Unset()

		// Mock get config
		getConfCall := s.mockConfigManager.
			On("Get", mock.Anything, &domain.GetTiCDCClusterConfigReq{
				TiCDCClusterID: 1111,
			}).Return(&domain.GetTiCDCClusterConfigResp{Configuration: &domain.TiCDCConfigData{}}, nil)
		defer getConfCall.Unset()

		// Mock get instance type
		getInsTypeCall := s.mockInstanceTypeModel.
			On("GetCDCInstanceType", mock.Anything, &domain.GetCDCInstanceTypeReq{
				Name:          "32rcu",
				CloudProvider: domain.CloudProviderTypeAWS,
				Region:        "us-west-2",
			}).Return(&domain.GetCDCInstanceTypeResp{InstanceType: &domain.CDCInstanceType{
			MachineCPUNum: 16,
			IOPS:          4000,
			ThroughputMbs: 400,
		}}, nil)
		defer getInsTypeCall.Unset()

		// Mock regional call
		regionalCall := s.mockTiCDCClusterCli.
			On("CreateTiCDCCluster", mock.Anything, mock.Anything, mock.Anything).
			Return(nil, nil)
		defer regionalCall.Unset()

		err := s.model.doCreate(context.Background(), &domain.CreateTiCDCClusterReq{}, ticdcClusterEnt)
		s.NoError(err)
		s.True(ticdcClusterEnt.IsNewKernel)
		s.Equal(ticdcClusterEnt.Version, "v9.0.0-beta.1")
	})

	s.Run("new kernel enabled - create config error", func() {
		ticdcClusterEnt := makeTicdcClusterEnt4DoCreate()
		origiMinNewkernelVersion := config.Global().ChangefeedConfig.MinNewkernelVersion
		originMinNewkernelCompatibleTiDBVersion := config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion
		defer func() {
			config.Global().ChangefeedConfig.MinNewkernelVersion = origiMinNewkernelVersion
			config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = originMinNewkernelCompatibleTiDBVersion
		}()
		config.Global().ChangefeedConfig.MinNewkernelVersion = "v9.0.0-beta.1"
		config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = "v7.5.0"

		// Mock feature gate to return false for blacklist so the default new kernel path is used.
		allowlistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", mock.Anything, feature_gate.TiCDCNewKernelAllowlist).
			Return(true).Maybe()
		defer allowlistCall.Unset()

		blacklistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", mock.Anything, feature_gate.TiCDCNewKernelBlacklist).
			Return(false)
		defer blacklistCall.Unset()

		// Mock save entity
		saveCall := s.mockTiCDCClusterRepo.
			On("Save", mock.Anything, mock.Anything).
			Run(func(args mock.Arguments) {
				cluster := args.Get(1).(*entity.TiCDCCluster)
				s.True(cluster.IsNewKernel)
				s.Equal("v9.0.0-beta.1", cluster.Version)
			}).
			Return(nil)
		defer saveCall.Unset()

		// Mock create config error
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "create config error")
		createConfCall := s.mockConfigManager.
			On("Create", mock.Anything, mock.Anything).Return(nil, mockErr)
		defer createConfCall.Unset()

		err := s.model.doCreate(context.Background(), &domain.CreateTiCDCClusterReq{}, ticdcClusterEnt)
		s.Error(err)
		s.Contains(err.Error(), "override version and config for new kernel")
	})

	s.Run("new kernel enabled - update config error", func() {
		ticdcClusterEnt := makeTicdcClusterEnt4DoCreate()
		origiMinNewkernelVersion := config.Global().ChangefeedConfig.MinNewkernelVersion
		originMinNewkernelCompatibleTiDBVersion := config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion
		defer func() {
			config.Global().ChangefeedConfig.MinNewkernelVersion = origiMinNewkernelVersion
			config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = originMinNewkernelCompatibleTiDBVersion
		}()
		config.Global().ChangefeedConfig.MinNewkernelVersion = "v9.0.0-beta.1"
		config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = "v7.5.0"

		// Mock feature gate to return false for blacklist so the default new kernel path is used.
		allowlistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", mock.Anything, feature_gate.TiCDCNewKernelAllowlist).
			Return(true).Maybe()
		defer allowlistCall.Unset()

		blacklistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", mock.Anything, feature_gate.TiCDCNewKernelBlacklist).
			Return(false)
		defer blacklistCall.Unset()

		// Mock save entity
		saveCall := s.mockTiCDCClusterRepo.
			On("Save", mock.Anything, mock.Anything).
			Run(func(args mock.Arguments) {
				cluster := args.Get(1).(*entity.TiCDCCluster)
				s.True(cluster.IsNewKernel)
				s.Equal("v9.0.0-beta.1", cluster.Version)
			}).
			Return(nil)
		defer saveCall.Unset()

		// Mock create config success
		createConfCall := s.mockConfigManager.
			On("Create", mock.Anything, mock.Anything).Run(func(args mock.Arguments) {
			req := args.Get(1).(*domain.CreateTiCDCClusterConfigReq)
			s.Equal(uint64(1111), req.TiCDCClusterID)
			s.Equal("v9.0.0-beta.1", req.TiDBClusterVersion)
		}).Return(nil, nil)
		defer createConfCall.Unset()

		// Mock patch config error
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "update config error")
		patchConfCall := s.mockConfigManager.
			On("PatchConfig", mock.Anything, mock.Anything).Run(func(args mock.Arguments) {
			req := args.Get(1).(*domain.PatchTiCDCClusterConfigReq)
			s.Equal(uint64(1111), req.TiCDCClusterID)
			s.True(req.CustomParams["newarch"].Value.GetBoolValue())
		}).Return(mockErr)
		defer patchConfCall.Unset()

		err := s.model.doCreate(context.Background(), &domain.CreateTiCDCClusterReq{}, ticdcClusterEnt)
		s.Error(err)
		s.Contains(err.Error(), "override version and config for new kernel")
	})

	s.Run("new kernel enabled - get config error", func() {
		ticdcClusterEnt := makeTicdcClusterEnt4DoCreate()
		origiMinNewkernelVersion := config.Global().ChangefeedConfig.MinNewkernelVersion
		originMinNewkernelCompatibleTiDBVersion := config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion
		defer func() {
			config.Global().ChangefeedConfig.MinNewkernelVersion = origiMinNewkernelVersion
			config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = originMinNewkernelCompatibleTiDBVersion
		}()
		config.Global().ChangefeedConfig.MinNewkernelVersion = "v9.0.0-beta.1"
		config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = "v7.5.0"

		// Mock feature gate to return false for blacklist so the default new kernel path is used.
		allowlistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", mock.Anything, feature_gate.TiCDCNewKernelAllowlist).
			Return(true).Maybe()
		defer allowlistCall.Unset()

		blacklistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", mock.Anything, feature_gate.TiCDCNewKernelBlacklist).
			Return(false)
		defer blacklistCall.Unset()

		// Mock save entity
		saveCall := s.mockTiCDCClusterRepo.
			On("Save", mock.Anything, mock.Anything).
			Run(func(args mock.Arguments) {
				cluster := args.Get(1).(*entity.TiCDCCluster)
				s.True(cluster.IsNewKernel)
				s.Equal("v9.0.0-beta.1", cluster.Version)
			}).
			Return(nil)
		defer saveCall.Unset()

		// Mock create config success
		createConfCall := s.mockConfigManager.
			On("Create", mock.Anything, mock.Anything).Run(func(args mock.Arguments) {
			req := args.Get(1).(*domain.CreateTiCDCClusterConfigReq)
			s.Equal(uint64(1111), req.TiCDCClusterID)
			s.Equal("v9.0.0-beta.1", req.TiDBClusterVersion)
		}).Return(nil, nil)
		defer createConfCall.Unset()

		// Mock patch config success
		patchConfCall := s.mockConfigManager.
			On("PatchConfig", mock.Anything, mock.Anything).Run(func(args mock.Arguments) {
			req := args.Get(1).(*domain.PatchTiCDCClusterConfigReq)
			s.Equal(uint64(1111), req.TiCDCClusterID)
			s.True(req.CustomParams["newarch"].Value.GetBoolValue())
		}).Return(nil, nil)
		defer patchConfCall.Unset()

		// Mock get config error
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "get config error")
		getConfCall := s.mockConfigManager.
			On("Get", mock.Anything, &domain.GetTiCDCClusterConfigReq{
				TiCDCClusterID: 1111,
			}).Return(nil, mockErr)
		defer getConfCall.Unset()

		err := s.model.doCreate(context.Background(), &domain.CreateTiCDCClusterReq{}, ticdcClusterEnt)
		s.Error(err)
		s.Contains(err.Error(), "get ticdc cluster configuration")
	})

	s.Run("new kernel enabled - override version and config error", func() {
		ticdcClusterEnt := makeTicdcClusterEnt4DoCreate()
		origiMinNewkernelVersion := config.Global().ChangefeedConfig.MinNewkernelVersion
		originMinNewkernelCompatibleTiDBVersion := config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion
		defer func() {
			config.Global().ChangefeedConfig.MinNewkernelVersion = origiMinNewkernelVersion
			config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = originMinNewkernelCompatibleTiDBVersion
		}()
		config.Global().ChangefeedConfig.MinNewkernelVersion = "v9.0.0-beta.1"
		config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = "v7.5.0"

		// Mock feature gate to return false for blacklist so the default new kernel path is used.
		allowlistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", mock.Anything, feature_gate.TiCDCNewKernelAllowlist).
			Return(true).Maybe()
		defer allowlistCall.Unset()

		blacklistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", mock.Anything, feature_gate.TiCDCNewKernelBlacklist).
			Return(false)
		defer blacklistCall.Unset()

		// Mock save entity error
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		saveCall := s.mockTiCDCClusterRepo.
			On("Save", mock.Anything, mock.Anything).
			Run(func(args mock.Arguments) {
				cluster := args.Get(1).(*entity.TiCDCCluster)
				s.True(cluster.IsNewKernel)
				s.Equal("v9.0.0-beta.1", cluster.Version)
			}).
			Return(mockErr)
		defer saveCall.Unset()

		err := s.model.doCreate(context.Background(), &domain.CreateTiCDCClusterReq{}, ticdcClusterEnt)
		s.Error(err)
		s.Contains(err.Error(), "override version and config for new kernel")
	})

	s.Run("old kernel path from blacklist - save and config error", func() {
		ticdcClusterEnt := makeTicdcClusterEnt4DoCreate()
		origiMinNewkernelVersion := config.Global().ChangefeedConfig.MinNewkernelVersion
		originMinNewkernelCompatibleTiDBVersion := config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion
		defer func() {
			config.Global().ChangefeedConfig.MinNewkernelVersion = origiMinNewkernelVersion
			config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = originMinNewkernelCompatibleTiDBVersion
		}()
		config.Global().ChangefeedConfig.MinNewkernelVersion = "v9.0.0-beta.1"
		config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = "v7.5.0"

		// Mock feature gate to return true for blacklist, forcing the old kernel path.
		allowlistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", mock.Anything, feature_gate.TiCDCNewKernelAllowlist).
			Return(false).Maybe()
		defer allowlistCall.Unset()
		blacklistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", mock.Anything, feature_gate.TiCDCNewKernelBlacklist).
			Return(true)
		defer blacklistCall.Unset()

		// Mock save entity error
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		saveCall := s.mockTiCDCClusterRepo.
			On("Save", mock.Anything, mock.Anything).
			Run(func(args mock.Arguments) {
				cluster := args.Get(1).(*entity.TiCDCCluster)
				s.False(cluster.IsNewKernel)
				s.Equal("v7.5.0", cluster.Version)
			}).
			Return(mockErr)
		defer saveCall.Unset()

		err := s.model.doCreate(context.Background(), &domain.CreateTiCDCClusterReq{}, ticdcClusterEnt)
		s.Error(err)
		s.Contains(err.Error(), "save and config for old kernel")
	})

	s.Run("old kernel path from blacklist - create config error", func() {
		ticdcClusterEnt := makeTicdcClusterEnt4DoCreate()
		origiMinNewkernelVersion := config.Global().ChangefeedConfig.MinNewkernelVersion
		originMinNewkernelCompatibleTiDBVersion := config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion
		defer func() {
			config.Global().ChangefeedConfig.MinNewkernelVersion = origiMinNewkernelVersion
			config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = originMinNewkernelCompatibleTiDBVersion
		}()
		config.Global().ChangefeedConfig.MinNewkernelVersion = "v9.0.0-beta.1"
		config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = "v7.5.0"

		// Mock feature gate to return true for blacklist, forcing the old kernel path.
		allowlistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", mock.Anything, feature_gate.TiCDCNewKernelAllowlist).
			Return(false).Maybe()
		defer allowlistCall.Unset()
		blacklistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", mock.Anything, feature_gate.TiCDCNewKernelBlacklist).
			Return(true)
		defer blacklistCall.Unset()

		// Mock save entity
		saveCall := s.mockTiCDCClusterRepo.
			On("Save", mock.Anything, mock.Anything).
			Run(func(args mock.Arguments) {
				cluster := args.Get(1).(*entity.TiCDCCluster)
				s.False(cluster.IsNewKernel)
				s.Equal("v7.5.0", cluster.Version)
			}).
			Return(nil)
		defer saveCall.Unset()

		// Mock create config error
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		createConfCall := s.mockConfigManager.On("Create", mock.Anything, mock.Anything).Return(nil, mockErr)
		defer createConfCall.Unset()

		err := s.model.doCreate(context.Background(), &domain.CreateTiCDCClusterReq{}, ticdcClusterEnt)
		s.Error(err)
		s.Contains(err.Error(), "save and config for old kernel")
	})

	s.Run("old kernel path from blacklist - get config error", func() {
		ticdcClusterEnt := makeTicdcClusterEnt4DoCreate()
		origiMinNewkernelVersion := config.Global().ChangefeedConfig.MinNewkernelVersion
		originMinNewkernelCompatibleTiDBVersion := config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion
		defer func() {
			config.Global().ChangefeedConfig.MinNewkernelVersion = origiMinNewkernelVersion
			config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = originMinNewkernelCompatibleTiDBVersion
		}()
		config.Global().ChangefeedConfig.MinNewkernelVersion = "v9.0.0-beta.1"
		config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = "v7.5.0"

		// Mock feature gate to return true for blacklist, forcing the old kernel path.
		allowlistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", mock.Anything, feature_gate.TiCDCNewKernelAllowlist).
			Return(false).Maybe()
		defer allowlistCall.Unset()
		blacklistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", mock.Anything, feature_gate.TiCDCNewKernelBlacklist).
			Return(true)
		defer blacklistCall.Unset()

		// Mock save entity
		saveCall := s.mockTiCDCClusterRepo.
			On("Save", mock.Anything, mock.Anything).
			Run(func(args mock.Arguments) {
				cluster := args.Get(1).(*entity.TiCDCCluster)
				s.False(cluster.IsNewKernel)
				s.Equal("v7.5.0", cluster.Version)
			}).
			Return(nil)
		defer saveCall.Unset()

		// Mock create config
		createConfCall := s.mockConfigManager.
			On("Create", mock.Anything, mock.Anything).Return(nil, nil)
		defer createConfCall.Unset()

		// Mock get config error
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		getConfCall := s.mockConfigManager.On("Get", mock.Anything, mock.Anything).Return(nil, mockErr)
		defer getConfCall.Unset()

		err := s.model.doCreate(context.Background(), &domain.CreateTiCDCClusterReq{}, ticdcClusterEnt)
		s.Error(err)
		s.Contains(err.Error(), "get ticdc cluster configuration")
	})

	s.Run("old kernel path from blacklist - get cdc instance type error", func() {
		ticdcClusterEnt := makeTicdcClusterEnt4DoCreate()
		origiMinNewkernelVersion := config.Global().ChangefeedConfig.MinNewkernelVersion
		originMinNewkernelCompatibleTiDBVersion := config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion
		defer func() {
			config.Global().ChangefeedConfig.MinNewkernelVersion = origiMinNewkernelVersion
			config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = originMinNewkernelCompatibleTiDBVersion
		}()
		config.Global().ChangefeedConfig.MinNewkernelVersion = "v9.0.0-beta.1"
		config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = "v7.5.0"

		// Mock feature gate to return true for blacklist, forcing the old kernel path.
		allowlistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", mock.Anything, feature_gate.TiCDCNewKernelAllowlist).
			Return(false).Maybe()
		defer allowlistCall.Unset()
		blacklistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", mock.Anything, feature_gate.TiCDCNewKernelBlacklist).
			Return(true)
		defer blacklistCall.Unset()

		// Mock save entity
		saveCall := s.mockTiCDCClusterRepo.
			On("Save", mock.Anything, mock.Anything).
			Run(func(args mock.Arguments) {
				cluster := args.Get(1).(*entity.TiCDCCluster)
				s.False(cluster.IsNewKernel)
				s.Equal("v7.5.0", cluster.Version)
			}).
			Return(nil)
		defer saveCall.Unset()

		// Mock create config
		createConfCall := s.mockConfigManager.
			On("Create", mock.Anything, mock.Anything).Return(nil, nil)
		defer createConfCall.Unset()

		// Mock get config
		getConfCall := s.mockConfigManager.
			On("Get", mock.Anything, &domain.GetTiCDCClusterConfigReq{
				TiCDCClusterID: 1111,
			}).Return(&domain.GetTiCDCClusterConfigResp{Configuration: &domain.TiCDCConfigData{}}, nil)
		defer getConfCall.Unset()

		// Mock get instance type error
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		getInsTypeCall := s.mockInstanceTypeModel.
			On("GetCDCInstanceType", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer getInsTypeCall.Unset()

		err := s.model.doCreate(context.Background(), &domain.CreateTiCDCClusterReq{}, ticdcClusterEnt)
		s.Error(err)
		s.Contains(err.Error(), "get ticdc instance type")
	})

	s.Run("old kernel path from blacklist - call regional error", func() {
		ticdcClusterEnt := makeTicdcClusterEnt4DoCreate()
		origiMinNewkernelVersion := config.Global().ChangefeedConfig.MinNewkernelVersion
		originMinNewkernelCompatibleTiDBVersion := config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion
		defer func() {
			config.Global().ChangefeedConfig.MinNewkernelVersion = origiMinNewkernelVersion
			config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = originMinNewkernelCompatibleTiDBVersion
		}()
		config.Global().ChangefeedConfig.MinNewkernelVersion = "v9.0.0-beta.1"
		config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = "v7.5.0"

		// Mock feature gate to return true for blacklist, forcing the old kernel path.
		allowlistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", mock.Anything, feature_gate.TiCDCNewKernelAllowlist).
			Return(false).Maybe()
		defer allowlistCall.Unset()
		blacklistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", mock.Anything, feature_gate.TiCDCNewKernelBlacklist).
			Return(true)
		defer blacklistCall.Unset()

		// Mock save entity
		saveCall := s.mockTiCDCClusterRepo.
			On("Save", mock.Anything, mock.Anything).
			Run(func(args mock.Arguments) {
				cluster := args.Get(1).(*entity.TiCDCCluster)
				s.False(cluster.IsNewKernel)
				s.Equal("v7.5.0", cluster.Version)
			}).
			Return(nil)
		defer saveCall.Unset()

		// Mock create config
		createConfCall := s.mockConfigManager.
			On("Create", mock.Anything, mock.Anything).Return(nil, nil)
		defer createConfCall.Unset()

		// Mock get config
		getConfCall := s.mockConfigManager.
			On("Get", mock.Anything, &domain.GetTiCDCClusterConfigReq{
				TiCDCClusterID: 1111,
			}).Return(&domain.GetTiCDCClusterConfigResp{Configuration: &domain.TiCDCConfigData{}}, nil)
		defer getConfCall.Unset()

		// Mock get instance type
		getInsTypeCall := s.mockInstanceTypeModel.
			On("GetCDCInstanceType", mock.Anything, &domain.GetCDCInstanceTypeReq{
				Name:          "32rcu",
				CloudProvider: domain.CloudProviderTypeAWS,
				Region:        "us-west-2",
			}).Return(&domain.GetCDCInstanceTypeResp{InstanceType: &domain.CDCInstanceType{}}, nil)
		defer getInsTypeCall.Unset()

		// Mock regional call error
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		regionalCall := s.mockTiCDCClusterCli.
			On("CreateTiCDCCluster", mock.Anything, mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer regionalCall.Unset()

		err := s.model.doCreate(context.Background(), &domain.CreateTiCDCClusterReq{}, ticdcClusterEnt)
		s.Error(err)
		s.Contains(err.Error(), "create regional ticdc cluster")
	})

	s.Run("old kernel path from blacklist - do create successfully", func() {
		ticdcClusterEnt := makeTicdcClusterEnt4DoCreate()
		origiMinNewkernelVersion := config.Global().ChangefeedConfig.MinNewkernelVersion
		originMinNewkernelCompatibleTiDBVersion := config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion
		defer func() {
			config.Global().ChangefeedConfig.MinNewkernelVersion = origiMinNewkernelVersion
			config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = originMinNewkernelCompatibleTiDBVersion
		}()
		config.Global().ChangefeedConfig.MinNewkernelVersion = "v9.0.0-beta.1"
		config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = "v7.5.0"

		// Mock feature gate to return true for blacklist, forcing the old kernel path.
		allowlistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", mock.Anything, feature_gate.TiCDCNewKernelAllowlist).
			Return(false).Maybe()
		defer allowlistCall.Unset()
		blacklistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", mock.Anything, feature_gate.TiCDCNewKernelBlacklist).
			Return(true)
		defer blacklistCall.Unset()

		// Mock save entity
		saveCall := s.mockTiCDCClusterRepo.
			On("Save", mock.Anything, mock.Anything).
			Run(func(args mock.Arguments) {
				cluster := args.Get(1).(*entity.TiCDCCluster)
				s.False(cluster.IsNewKernel)
				s.Equal("v7.5.0", cluster.Version)
			}).
			Return(nil)
		defer saveCall.Unset()

		// Mock create config
		createConfCall := s.mockConfigManager.
			On("Create", mock.Anything, mock.Anything).Return(nil, nil)
		defer createConfCall.Unset()
		// Mock get config
		getConfCall := s.mockConfigManager.
			On("Get", mock.Anything, &domain.GetTiCDCClusterConfigReq{
				TiCDCClusterID: 1111,
			}).Return(&domain.GetTiCDCClusterConfigResp{Configuration: &domain.TiCDCConfigData{}}, nil)
		defer getConfCall.Unset()

		// Mock get instance type
		getInsTypeCall := s.mockInstanceTypeModel.
			On("GetCDCInstanceType", mock.Anything, &domain.GetCDCInstanceTypeReq{
				Name:          "32rcu",
				CloudProvider: domain.CloudProviderTypeAWS,
				Region:        "us-west-2",
			}).Return(&domain.GetCDCInstanceTypeResp{InstanceType: &domain.CDCInstanceType{}}, nil)
		defer getInsTypeCall.Unset()

		// Mock regional call
		regionalCall := s.mockTiCDCClusterCli.
			On("CreateTiCDCCluster", mock.Anything, mock.Anything, mock.Anything).
			Return(nil, nil)
		defer regionalCall.Unset()

		err := s.model.doCreate(context.Background(), &domain.CreateTiCDCClusterReq{}, ticdcClusterEnt)
		s.NoError(err)
		s.False(ticdcClusterEnt.IsNewKernel)
		s.Equal(ticdcClusterEnt.Version, "v7.5.0")
	})

	s.Run("blacklisted regular path - migration enabled new kernel", func() {
		ticdcClusterEnt := makeTicdcClusterEnt4DoCreate()

		origiMinNewkernelVersion := config.Global().ChangefeedConfig.MinNewkernelVersion
		originMinNewkernelCompatibleTiDBVersion := config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion
		defer func() {
			config.Global().ChangefeedConfig.MinNewkernelVersion = origiMinNewkernelVersion
			config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = originMinNewkernelCompatibleTiDBVersion
		}()
		config.Global().ChangefeedConfig.MinNewkernelVersion = "v9.0.0-beta.1"
		config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = "v7.5.0"
		// Migration can force new kernel even when the regular blacklist would opt out.
		allowlistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", &feature_gate.QueryFeatureGateReq{
				TenantID:  ticdcClusterEnt.TenantID.Uint64(),
				ProjectID: ticdcClusterEnt.ProjectID.Uint64(),
				ClusterID: ticdcClusterEnt.ClusterID.Uint64(),
			}, feature_gate.TiCDCNewKernelAllowlist).
			Return(false).Maybe()
		defer allowlistCall.Unset()

		blacklistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", &feature_gate.QueryFeatureGateReq{
				TenantID:  ticdcClusterEnt.TenantID.Uint64(),
				ProjectID: ticdcClusterEnt.ProjectID.Uint64(),
				ClusterID: ticdcClusterEnt.ClusterID.Uint64(),
			}, feature_gate.TiCDCNewKernelBlacklist).
			Return(true).Maybe()
		defer blacklistCall.Unset()

		// Mock save entity
		saveCall := s.mockTiCDCClusterRepo.
			On("Save", mock.Anything, mock.Anything).
			Run(func(args mock.Arguments) {
				cluster := args.Get(1).(*entity.TiCDCCluster)
				s.True(cluster.IsNewKernel)
				s.Equal("v9.0.0-beta.1", cluster.Version)
			}).
			Return(nil)
		defer saveCall.Unset()

		// Mock create config
		createConfCall := s.mockConfigManager.
			On("Create", mock.Anything, mock.Anything).Run(func(args mock.Arguments) {
			req := args.Get(1).(*domain.CreateTiCDCClusterConfigReq)
			s.Equal(uint64(1), req.TenantID)
			s.Equal(uint64(1111), req.TiCDCClusterID)
			s.Equal("v9.0.0-beta.1", req.TiDBClusterVersion)
		}).Return(nil, nil)
		defer createConfCall.Unset()

		// Mock patch config for new kernel
		patchConfCall := s.mockConfigManager.
			On("PatchConfig", mock.Anything, mock.Anything).Run(func(args mock.Arguments) {
			req := args.Get(1).(*domain.PatchTiCDCClusterConfigReq)
			s.Equal(uint64(1111), req.TiCDCClusterID)
			s.True(req.CustomParams["newarch"].Value.GetBoolValue())
		}).Return(nil, nil)
		defer patchConfCall.Unset()

		// Mock get config
		getConfCall := s.mockConfigManager.
			On("Get", mock.Anything, &domain.GetTiCDCClusterConfigReq{
				TiCDCClusterID: 1111,
			}).Return(&domain.GetTiCDCClusterConfigResp{Configuration: &domain.TiCDCConfigData{}}, nil)
		defer getConfCall.Unset()

		// Mock get instance type
		getInsTypeCall := s.mockInstanceTypeModel.
			On("GetCDCInstanceType", mock.Anything, &domain.GetCDCInstanceTypeReq{
				Name:          "32rcu",
				CloudProvider: domain.CloudProviderTypeAWS,
				Region:        "us-west-2",
			}).Return(&domain.GetCDCInstanceTypeResp{InstanceType: &domain.CDCInstanceType{}}, nil)
		defer getInsTypeCall.Unset()

		// Mock regional call
		regionalCall := s.mockTiCDCClusterCli.
			On("CreateTiCDCCluster", mock.Anything, mock.Anything, mock.Anything).
			Return(nil, nil)
		defer regionalCall.Unset()

		err := s.model.doCreate(context.Background(), &domain.CreateTiCDCClusterReq{
			UseNewCdcKernel: true,
		}, ticdcClusterEnt)
		s.NoError(err)
		s.True(ticdcClusterEnt.IsNewKernel)
		s.Equal(ticdcClusterEnt.Version, "v9.0.0-beta.1")
	})
}

func (s *clusterModelRegionalImplTestSuite) TestCreate() {
	req := &domain.CreateTiCDCClusterReq{
		TenantID:          1,
		ProjectID:         11,
		ClusterID:         111,
		Name:              "cdc-111",
		InfraProviderType: "eks",
		CloudAccountID:    "abc",
		Region:            "us-west-2",
		Version:           "v7.5.0",
		SpecificationCode: "32rcu",
		BizType:           domain.TiCDCClusterBizTypeChangefeed,
		DeployType:        entity.TiCDCClusterDeployTypeShared,
	}
	s.Run("get entity error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockTiCDCClusterRepo.On("Get", mock.Anything, mock.Anything).Return(nil, mockErr)
		defer call.Unset()
		_, err := s.model.Create(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "get ticdc cluster entity")
	})

	s.Run("existing ticdc cluster is deleting", func() {
		mockEnt := &entity.TiCDCCluster{
			Status: entity.TiCDCClusterStatusDeleting,
		}
		mockEnt.ID = 1111
		call := s.mockTiCDCClusterRepo.On("Get", mock.Anything, mock.Anything).Return(mockEnt, nil)
		defer call.Unset()
		_, err := s.model.Create(context.Background(), req)
		s.Error(err)
		s.Contains(err.Error(), "is deleting")
	})

	s.Run("ticdc cluster exists", func() {
		mockEnt := &entity.TiCDCCluster{}
		mockEnt.ID = 1111
		call := s.mockTiCDCClusterRepo.On("Get", mock.Anything, mock.Anything).Return(mockEnt, nil)
		defer call.Unset()
		resp, err := s.model.Create(context.Background(), req)
		s.NoError(err)
		s.Equal(mockEnt.ID.Uint64(), resp.ID)
	})

	mockGetErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
	getCall := s.mockTiCDCClusterRepo.On("Get", mock.Anything, mock.Anything).Return(nil, mockGetErr)
	defer getCall.Unset()

	s.Run("get logical region error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeGRPC, "network error")
		call := s.mockRegionCli.
			On("GetRegionByTenantID", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.model.Create(context.Background(), req)
		s.Error(err)
		s.Contains(err.Error(), "get logical region")
	})

	mockRegion := &tidb_mgmt_servicev1.Region{
		Region: &global_commonv1.Region{
			LogicalName: "us-west-2-test-1",
		},
	}
	regionCall := s.mockRegionCli.
		On("GetRegionByTenantID", mock.Anything, uint64(1), "", "us-west-2").
		Return(mockRegion, nil)
	defer regionCall.Unset()

	s.Run("transaction error", func() {
		ctrl := gomock.NewController(s.T())
		defer s.cleanTransProvider()
		mockTransProvider := s.newTransProvider(ctrl)
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		mockTransProvider.EXPECT().Transaction(gomock.Any(), gomock.Any()).Return(mockErr)
		_, err := s.model.Create(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr, err)
	})

	ctrl := gomock.NewController(s.T())
	defer s.cleanTransProvider()
	mockTransProvider := s.newTransProvider(ctrl)
	mockTransProvider.EXPECT().Transaction(gomock.Any(), gomock.Any()).Return(nil)

	s.Run("create successfully", func() {
		_, err := s.model.Create(context.Background(), req)
		s.NoError(err)
	})
}

func (s *clusterModelRegionalImplTestSuite) newTransProvider(ctrl *gomock.Controller) *mock_db.MockTransProvider {
	mockTransProvider := mock_db.NewMockTransProvider(ctrl)
	s.model.transProvider = mockTransProvider
	return mockTransProvider
}

func (s *clusterModelRegionalImplTestSuite) cleanTransProvider() {
	s.model.transProvider = nil
}

func (s *clusterModelRegionalImplTestSuite) TestScaleRegionalTiCDCCluster() {
	ticdcClusterEnt := &entity.TiCDCCluster{
		Name:              "cdc-111",
		TenantID:          1,
		Region:            "us-west-2",
		InfraProviderType: "eks",
	}
	instanceType := &domain.CDCInstanceType{
		MachineType:      "m5.2xlarge",
		Replicas:         2,
		StorageType:      "gp3",
		MachineStorageGi: 1000,
		IOPS:             4000,
		ThroughputMbs:    400,
	}

	s.Run("call regional error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockTiCDCClusterCli.
			On("ScaleTiCDCCluster", mock.Anything, mock.Anything, mock.Anything).
			Return(mockErr)
		defer call.Unset()
		err := s.model.scaleRegionalTiCDCCluster(context.Background(), ticdcClusterEnt, instanceType)
		s.Error(err)
		s.Equal(mockErr, err)
	})

	s.Run("call regional successfully", func() {
		expectedRegion := &regionalapi.Region{
			TenantID:      1,
			Name:          "us-west-2",
			CloudProvider: "aws",
		}
		expectedScaleReq := &ticdc_v1.ScaleTiCDCClusterRequest{
			TicdcClusterId: "cdc-111",
			InstanceType:   "m5.2xlarge",
			Count:          2,
			Storage: &ticdc_v1.TiCDCCluster_Storage{
				Type:          "gp3",
				SizeGi:        1000,
				Iops:          4000,
				ThroughputMbs: 400,
			},
		}
		call := s.mockTiCDCClusterCli.
			On("ScaleTiCDCCluster", mock.Anything, expectedRegion, expectedScaleReq).
			Return(nil)
		defer call.Unset()
		err := s.model.scaleRegionalTiCDCCluster(context.Background(), ticdcClusterEnt, instanceType)
		s.NoError(err)
	})
}

func (s *clusterModelRegionalImplTestSuite) TestDoScale() {
	req := &domain.ScaleTiCDCClusterReq{
		ID:                1111,
		SpecificationCode: "16rcu",
	}
	ticdcClusterEnt := &entity.TiCDCCluster{}
	instanceType := &domain.CDCInstanceType{}

	s.Run("save entity error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockTiCDCClusterRepo.On("Save", mock.Anything, mock.Anything).Return(mockErr)
		defer call.Unset()
		err := s.model.doScale(context.Background(), req, ticdcClusterEnt, instanceType)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "save ticdc cluster entity")
	})

	saveCall := s.mockTiCDCClusterRepo.On("Save", mock.Anything, mock.Anything).Return(nil)
	defer saveCall.Unset()

	s.Run("call regional error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockTiCDCClusterCli.
			On("ScaleTiCDCCluster", mock.Anything, mock.Anything, mock.Anything).
			Return(mockErr)
		defer call.Unset()
		err := s.model.doScale(context.Background(), req, ticdcClusterEnt, instanceType)
		s.Error(err)
		s.Equal(mockErr, err)
	})

	regionalCall := s.mockTiCDCClusterCli.
		On("ScaleTiCDCCluster", mock.Anything, mock.Anything, mock.Anything).
		Return(nil)
	defer regionalCall.Unset()

	s.Run("do scale successfully", func() {
		err := s.model.doScale(context.Background(), req, ticdcClusterEnt, instanceType)
		s.NoError(err)
		s.Equal("16rcu", ticdcClusterEnt.SpecificationCode)
		s.Equal(entity.TiCDCClusterStatusScaling, ticdcClusterEnt.Status)
	})
}

func (s *clusterModelRegionalImplTestSuite) TestScale() {
	req := &domain.ScaleTiCDCClusterReq{
		ID:                1111,
		SpecificationCode: "16rcu",
	}

	s.Run("load entity error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockTiCDCClusterRepo.On("Load", mock.Anything, mock.Anything).Return(nil, mockErr)
		defer call.Unset()
		_, err := s.model.Scale(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "load ticdc cluster entity")
	})

	mockEntity := &entity.TiCDCCluster{
		InfraProviderType: entity.InfraProviderTypeEKS,
		Region:            "us-west-2",
	}
	loadCall := s.mockTiCDCClusterRepo.On("Load", mock.Anything, mock.Anything).Return(mockEntity, nil)
	defer loadCall.Unset()

	s.Run("get instance type error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockInstanceTypeModel.
			On("GetCDCInstanceType", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.model.Scale(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "get ticdc instance type")
	})

	getInsTypeCall := s.mockInstanceTypeModel.
		On("GetCDCInstanceType", mock.Anything, &domain.GetCDCInstanceTypeReq{
			Name:          "16rcu",
			CloudProvider: domain.CloudProviderTypeAWS,
			Region:        "us-west-2",
		}).Return(&domain.GetCDCInstanceTypeResp{
		InstanceType: &domain.CDCInstanceType{},
	}, nil)
	defer getInsTypeCall.Unset()

	s.Run("load custom storage error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "load ticdc custom storage from db failed")
		call := s.mockTiCDCCustomStorageRepo.On("LoadByTiCDCClusterID", mock.Anything, mock.Anything).Return(nil, mockErr)
		defer call.Unset()

		_, err := s.model.Scale(context.Background(), req)
		s.Error(err)
		s.Contains(err.Error(), "load ticdc custom storage from db failed")
	})

	loadCustomStorageCall := s.mockTiCDCCustomStorageRepo.
		On("LoadByTiCDCClusterID", mock.Anything, mock.Anything).
		Return(nil, bizerr.WrapGormError(gorm.ErrRecordNotFound, "ticdc custom storage not found"))
	defer loadCustomStorageCall.Unset()

	s.Run("transaction error", func() {
		ctrl := gomock.NewController(s.T())
		defer s.cleanTransProvider()
		mockTransProvider := s.newTransProvider(ctrl)
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		mockTransProvider.EXPECT().Transaction(gomock.Any(), gomock.Any()).Return(mockErr)
		_, err := s.model.Scale(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr, err)
	})

	ctrl := gomock.NewController(s.T())
	defer s.cleanTransProvider()
	mockTransProvider := s.newTransProvider(ctrl)
	mockTransProvider.EXPECT().Transaction(gomock.Any(), gomock.Any()).Return(nil)

	s.Run("scale successfully", func() {
		_, err := s.model.Scale(context.Background(), req)
		s.NoError(err)
	})
}

func (s *clusterModelRegionalImplTestSuite) TestScale_WithCustomStorage() {
	req := &domain.ScaleTiCDCClusterReq{
		ID:                1111,
		SpecificationCode: "16rcu",
	}

	// Mocks setup
	mockEntity := &entity.TiCDCCluster{
		InfraProviderType: entity.InfraProviderTypeEKS,
		Region:            "us-west-2",
		Name:              "cdc-111",
		TenantID:          1,
	}
	loadCall := s.mockTiCDCClusterRepo.On("Load", mock.Anything, mock.Anything).Return(mockEntity, nil)
	defer loadCall.Unset()

	getInstanceTypeCall := s.mockInstanceTypeModel.
		On("GetCDCInstanceType", mock.Anything, mock.Anything).
		Return(&domain.GetCDCInstanceTypeResp{
			InstanceType: &domain.CDCInstanceType{
				MachineType:      "m5.xlarge",
				Replicas:         2,
				StorageType:      "gp3",
				MachineStorageGi: 100,
				IOPS:             3000,
				ThroughputMbs:    300,
			},
		}, nil)
	defer getInstanceTypeCall.Unset()

	customStorage := &entity.TiCDCCustomStorage{
		SizeGi:        2000,
		Iops:          8000,
		ThroughputMbs: 800,
	}
	loadCustomStorageCall := s.mockTiCDCCustomStorageRepo.
		On("LoadByTiCDCClusterID", mock.Anything, mock.Anything).
		Return(customStorage, nil)
	defer loadCustomStorageCall.Unset()

	// Mock Transaction to execute the closure
	ctrl := gomock.NewController(s.T())
	defer s.cleanTransProvider()
	mockTransProvider := s.newTransProvider(ctrl)
	mockTransProvider.EXPECT().Transaction(gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, fc func(context.Context) error) error {
		return fc(ctx)
	})

	// Mock Save in doScale
	saveCall := s.mockTiCDCClusterRepo.On("Save", mock.Anything, mock.Anything).Return(nil)
	defer saveCall.Unset()

	// Mock ScaleTiCDCCluster and verify arguments
	scaleCall := s.mockTiCDCClusterCli.
		On("ScaleTiCDCCluster", mock.Anything, mock.Anything, mock.Anything).
		Run(func(args mock.Arguments) {
			req := args.Get(2).(*ticdc_v1.ScaleTiCDCClusterRequest)
			s.Equal(uint32(2000), req.Storage.SizeGi)
			s.Equal(uint32(8000), req.Storage.Iops)
			s.Equal(uint32(800), req.Storage.ThroughputMbs)
		}).
		Return(nil)
	defer scaleCall.Unset()

	_, err := s.model.Scale(context.Background(), req)
	s.NoError(err)
}

func (s *clusterModelRegionalImplTestSuite) TestUpdateStorage() {
	req := &domain.UpdateTiCDCClusterStorageReq{
		TiCDCClusterID: 1111,
		SizeGi:         200,
		Iops:           5000,
		ThroughputMbs:  600,
	}

	s.Run("load entity error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockTiCDCClusterRepo.On("Load", mock.Anything, mock.Anything).Return(nil, mockErr)
		defer call.Unset()
		err := s.model.UpdateStorage(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "load ticdc cluster entity")
	})

	mockEntity := &entity.TiCDCCluster{
		InfraProviderType: entity.InfraProviderTypeEKS,
		Region:            "us-west-2",
		Name:              "cdc-111",
		TenantID:          1,
	}
	mockEntity.ID = 1111
	loadCall := s.mockTiCDCClusterRepo.On("Load", mock.Anything, mock.Anything).Return(mockEntity, nil)
	defer loadCall.Unset()

	s.Run("get instance type error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockInstanceTypeModel.
			On("GetCDCInstanceType", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		err := s.model.UpdateStorage(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "get ticdc instance type")
	})

	getInsTypeCall := s.mockInstanceTypeModel.
		On("GetCDCInstanceType", mock.Anything, &domain.GetCDCInstanceTypeReq{
			Name:          mockEntity.SpecificationCode,
			CloudProvider: domain.CloudProviderTypeAWS,
			Region:        "us-west-2",
		}).Return(&domain.GetCDCInstanceTypeResp{
		InstanceType: &domain.CDCInstanceType{
			StorageType:      "gp3",
			MachineStorageGi: 100,
			IOPS:             3000,
			ThroughputMbs:    300,
		},
	}, nil)
	defer getInsTypeCall.Unset()

	s.Run("transaction error", func() {
		ctrl := gomock.NewController(s.T())
		defer s.cleanTransProvider()
		mockTransProvider := s.newTransProvider(ctrl)
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		mockTransProvider.EXPECT().Transaction(gomock.Any(), gomock.Any()).Return(mockErr)
		err := s.model.UpdateStorage(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr, err)
	})

	ctrl := gomock.NewController(s.T())
	defer s.cleanTransProvider()
	mockTransProvider := s.newTransProvider(ctrl)
	mockTransProvider.EXPECT().Transaction(gomock.Any(), gomock.Any()).Return(nil)

	s.Run("update storage successfully", func() {
		err := s.model.UpdateStorage(context.Background(), req)
		s.NoError(err)
	})
}

func (s *clusterModelRegionalImplTestSuite) TestDoUpdateStorage() {
	req := &domain.UpdateTiCDCClusterStorageReq{
		TiCDCClusterID: 1111,
		SizeGi:         200,
		Iops:           5000,
		ThroughputMbs:  600,
	}
	ticdcClusterEnt := &entity.TiCDCCluster{
		Name:              "cdc-111",
		TenantID:          1,
		Region:            "us-west-2",
		InfraProviderType: entity.InfraProviderTypeEKS,
	}
	ticdcClusterEnt.ID = 1111

	instanceTypeResp := &domain.GetCDCInstanceTypeResp{
		InstanceType: &domain.CDCInstanceType{
			StorageType:      "gp3",
			MachineStorageGi: 100,
			IOPS:             3000,
			ThroughputMbs:    300,
		},
	}

	s.Run("no update needed", func() {
		emptyReq := &domain.UpdateTiCDCClusterStorageReq{TiCDCClusterID: 1111}
		err := s.model.updateStorage(context.Background(), emptyReq, ticdcClusterEnt, instanceTypeResp)
		s.NoError(err)
	})

	s.Run("save custom storage error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockTiCDCCustomStorageRepo.On("Save", mock.Anything, mock.Anything).Return(mockErr)
		defer call.Unset()

		err := s.model.updateStorage(context.Background(), req, ticdcClusterEnt, instanceTypeResp)
		s.Error(err)
		s.Contains(err.Error(), "save ticdc custom storage entity")
	})

	s.Run("call regional error", func() {
		saveCall := s.mockTiCDCCustomStorageRepo.On("Save", mock.Anything, mock.Anything).Return(nil)
		defer saveCall.Unset()

		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockTiCDCClusterCli.On("UpdateTiCDCClusterStorage", mock.Anything, mock.Anything, mock.Anything).Return(mockErr)
		defer call.Unset()

		err := s.model.updateStorage(context.Background(), req, ticdcClusterEnt, instanceTypeResp)
		s.Error(err)
		s.Equal(mockErr, err)
	})

	s.Run("success", func() {
		saveCall := s.mockTiCDCCustomStorageRepo.On("Save", mock.Anything, mock.Anything).Return(nil)
		defer saveCall.Unset()

		call := s.mockTiCDCClusterCli.On("UpdateTiCDCClusterStorage", mock.Anything, mock.Anything, mock.Anything).Return(nil)
		defer call.Unset()

		err := s.model.updateStorage(context.Background(), req, ticdcClusterEnt, instanceTypeResp)
		s.NoError(err)
	})
}

func (s *clusterModelRegionalImplTestSuite) TestQuery() {
	req := &domain.QueryTiCDCClusterReq{
		ID: 1111,
	}
	s.Run("get entity error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockTiCDCClusterRepo.On("Get", mock.Anything, mock.Anything).Return(nil, mockErr)
		defer call.Unset()
		_, err := s.model.Query(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "get ticdc cluster entity")
	})

	mockEntity := &entity.TiCDCCluster{}
	mockEntity.ID = 1111
	loadCall := s.mockTiCDCClusterRepo.On("Get", mock.Anything, mock.Anything).Return(mockEntity, nil)
	defer loadCall.Unset()

	s.Run("query successfully", func() {
		resp, err := s.model.Query(context.Background(), req)
		s.NoError(err)
		s.Equal(mockEntity, resp.TiCDCClusterEntity)
	})
}

func (s *clusterModelRegionalImplTestSuite) TestDeleteRegionalTiCDCCluster() {
	ticdcClusterEnt := &entity.TiCDCCluster{
		Name:              "cdc-111",
		TenantID:          1,
		Region:            "us-west-2",
		InfraProviderType: entity.InfraProviderTypeEKS,
	}

	s.Run("call regional error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockTiCDCClusterCli.
			On("DeleteTiCDCCluster", mock.Anything, mock.Anything, mock.Anything).
			Return(mockErr)
		defer call.Unset()
		err := s.model.deleteRegionalTiCDCCluster(context.Background(), ticdcClusterEnt)
		s.Error(err)
		s.Equal(mockErr, err)
	})

	s.Run("call regional successfully", func() {
		expectedRegion := &regionalapi.Region{
			TenantID:      1,
			Name:          "us-west-2",
			CloudProvider: "aws",
		}
		expectedScaleReq := &ticdc_v1.DeleteTiCDCClusterRequest{
			TicdcClusterId: "cdc-111",
		}
		call := s.mockTiCDCClusterCli.
			On("DeleteTiCDCCluster", mock.Anything, expectedRegion, expectedScaleReq).
			Return(nil)
		defer call.Unset()
		err := s.model.deleteRegionalTiCDCCluster(context.Background(), ticdcClusterEnt)
		s.NoError(err)
	})
}

func (s *clusterModelRegionalImplTestSuite) TestDoDelete() {
	ticdcClusterEnt := &entity.TiCDCCluster{}
	ticdcClusterEnt.ID = 1111

	s.Run("save entity error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockTiCDCClusterRepo.On("Save", mock.Anything, mock.Anything).Return(mockErr)
		defer call.Unset()
		err := s.model.doDelete(context.Background(), ticdcClusterEnt)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "save ticdc cluster entity")
	})

	saveCall := s.mockTiCDCClusterRepo.On("Save", mock.Anything, mock.Anything).Return(nil)
	defer saveCall.Unset()

	s.Run("call regional error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockTiCDCClusterCli.
			On("DeleteTiCDCCluster", mock.Anything, mock.Anything, mock.Anything).
			Return(mockErr)
		defer call.Unset()
		err := s.model.doDelete(context.Background(), ticdcClusterEnt)
		s.Error(err)
		s.Equal(mockErr, err)
	})

	regionalCall := s.mockTiCDCClusterCli.
		On("DeleteTiCDCCluster", mock.Anything, mock.Anything, mock.Anything).
		Return(nil)
	defer regionalCall.Unset()

	s.Run("do delete successfully", func() {
		err := s.model.doDelete(context.Background(), ticdcClusterEnt)
		s.NoError(err)
		s.Equal(entity.TiCDCClusterStatusDeleting, ticdcClusterEnt.Status)
	})
}

func (s *clusterModelRegionalImplTestSuite) TestDelete() {
	req := &domain.DeleteTiCDCClusterReq{
		ID: 1111,
	}
	s.Run("load entity error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockTiCDCClusterRepo.On("Load", mock.Anything, mock.Anything).Return(nil, mockErr)
		defer call.Unset()
		_, err := s.model.Delete(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "load ticdc cluster entity")
	})

	s.Run("ticdc cluster not exists", func() {
		mockErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
		call := s.mockTiCDCClusterRepo.On("Load", mock.Anything, mock.Anything).Return(nil, mockErr)
		defer call.Unset()
		_, err := s.model.Delete(context.Background(), req)
		s.NoError(err)
	})

	mockEnt := &entity.TiCDCCluster{}
	mockEnt.ID = 1111
	loadCall := s.mockTiCDCClusterRepo.On("Load", mock.Anything, mock.Anything).Return(mockEnt, nil)
	defer loadCall.Unset()

	s.Run("ticdc cluster is deleting", func() {
		mockEnt.Status = entity.TiCDCClusterStatusDeleting
		_, err := s.model.Delete(context.Background(), req)
		s.NoError(err)
	})

	s.Run("ticdc cluster is deleted", func() {
		mockEnt.Status = entity.TiCDCClusterStatusDeleted
		_, err := s.model.Delete(context.Background(), req)
		s.NoError(err)
	})

	mockEnt.Status = entity.TiCDCClusterStatusRunning
	s.Run("transaction error", func() {
		ctrl := gomock.NewController(s.T())
		defer s.cleanTransProvider()
		mockTransProvider := s.newTransProvider(ctrl)
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		mockTransProvider.EXPECT().Transaction(gomock.Any(), gomock.Any()).Return(mockErr)
		_, err := s.model.Delete(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr, err)
	})

	ctrl := gomock.NewController(s.T())
	defer s.cleanTransProvider()
	mockTransProvider := s.newTransProvider(ctrl)
	mockTransProvider.EXPECT().Transaction(gomock.Any(), gomock.Any()).Return(nil)

	s.Run("delete successfully", func() {
		_, err := s.model.Delete(context.Background(), req)
		s.NoError(err)
	})
}

func (s *clusterModelRegionalImplTestSuite) TestUpdateConfigRegionalTiCDCCluster() {
	ticdcClusterEnt := &entity.TiCDCCluster{
		Name:              "cdc-111",
		TenantID:          1,
		Region:            "us-west-2",
		InfraProviderType: "eks",
	}
	configData := &domain.TiCDCConfigData{
		BaseImage: pointer.ToString("ticdc"),
		Tag:       pointer.ToString("v7.5.0"),
		Config: map[string]interface{}{
			"a": 123,
			"b": "abc",
		},
	}

	s.Run("update ticdc cluster entity error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockTiCDCClusterRepo.On("Save", mock.Anything, mock.Anything).Return(mockErr)
		defer call.Unset()
		err := s.model.updateConfigRegionalTiCDCCluster(context.Background(), ticdcClusterEnt, configData)
		s.Error(err)
		s.Contains(err.Error(), "save ticdc cluster entity")
	})

	saveCall := s.mockTiCDCClusterRepo.On("Save", mock.Anything, mock.Anything).Return(nil)
	defer saveCall.Unset()

	s.Run("call regional error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockTiCDCClusterCli.
			On("UpdateTiCDCClusterConfig", mock.Anything, mock.Anything, mock.Anything).
			Return(mockErr)
		defer call.Unset()
		err := s.model.updateConfigRegionalTiCDCCluster(context.Background(), ticdcClusterEnt, configData)
		s.Error(err)
		s.Equal(mockErr, err)
	})

	s.Run("call regional successfully", func() {
		expectedRegion := &regionalapi.Region{
			TenantID:      1,
			Name:          "us-west-2",
			CloudProvider: "aws",
		}
		expectedUpdateReq := &ticdc_v1.UpdateTiCDCClusterConfigRequest{
			TicdcClusterId: "cdc-111",
			Configuration: &ticdc_v1.TiCDCCluster_Configuration{
				BaseImage: "ticdc",
				Tag:       "v7.5.0",
				Config:    "a = 123\nb = 'abc'\n",
			},
		}
		call := s.mockTiCDCClusterCli.
			On("UpdateTiCDCClusterConfig", mock.Anything, expectedRegion, expectedUpdateReq).
			Return(nil)
		defer call.Unset()
		err := s.model.updateConfigRegionalTiCDCCluster(context.Background(), ticdcClusterEnt, configData)
		s.NoError(err)
		s.Equal(entity.TiCDCClusterStatusUpgrading, ticdcClusterEnt.Status)
	})
}

func (s *clusterModelRegionalImplTestSuite) TestApplyConfig() {
	req := &domain.ApplyTiCDCClusterConfigReq{
		TiCDCClusterID: 1111,
	}
	s.Run("load entity error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockTiCDCClusterRepo.On("Load", mock.Anything, mock.Anything).Return(nil, mockErr)
		defer call.Unset()
		err := s.model.ApplyConfig(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "load ticdc cluster entity")
	})

	mockEntity := &entity.TiCDCCluster{}
	mockEntity.ID = 1111
	loadCall := s.mockTiCDCClusterRepo.On("Load", mock.Anything, mock.Anything).Return(mockEntity, nil)
	defer loadCall.Unset()

	s.Run("get config error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockConfigManager.On("Get", mock.Anything, mock.Anything).Return(nil, mockErr)
		defer call.Unset()
		err := s.model.ApplyConfig(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "get ticdc cluster configuration")
	})

	getConfCall := s.mockConfigManager.
		On("Get", mock.Anything, &domain.GetTiCDCClusterConfigReq{
			TiCDCClusterID: 1111,
		}).Return(&domain.GetTiCDCClusterConfigResp{Configuration: &domain.TiCDCConfigData{}}, nil)
	defer getConfCall.Unset()

	s.Run("transaction error", func() {
		ctrl := gomock.NewController(s.T())
		defer s.cleanTransProvider()
		mockTransProvider := s.newTransProvider(ctrl)
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		mockTransProvider.EXPECT().Transaction(gomock.Any(), gomock.Any()).Return(mockErr)
		err := s.model.ApplyConfig(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr, err)
	})

	ctrl := gomock.NewController(s.T())
	defer s.cleanTransProvider()
	mockTransProvider := s.newTransProvider(ctrl)
	mockTransProvider.EXPECT().Transaction(gomock.Any(), gomock.Any()).Return(nil)

	s.Run("apply config successfully", func() {
		err := s.model.ApplyConfig(context.Background(), req)
		s.NoError(err)
	})
}

func (s *clusterModelRegionalImplTestSuite) TestList() {
	req := &domain.ListTiCDCClustersReq{
		TiDBClusterID: 111,
	}
	s.Run("list ticdc clusters error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockTiCDCClusterRepo.On("List", mock.Anything, mock.Anything).Return(nil, mockErr)
		defer call.Unset()
		_, err := s.model.List(context.Background(), req)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "list ticdc clusters")
	})

	mockEntities := []entity.TiCDCCluster{
		{
			Name:      "ticdc-1",
			ClusterID: entity.ID(111),
		},
		{
			Name:      "ticdc-2",
			ClusterID: entity.ID(111),
		},
	}
	loadCall := s.mockTiCDCClusterRepo.On("List", mock.Anything, mock.Anything).Return(mockEntities, nil)
	defer loadCall.Unset()

	s.Run("list ticdc clusters successfully", func() {
		resp, err := s.model.List(context.Background(), req)
		s.NoError(err)
		s.Equal(mockEntities, resp.Items)
	})
}

func (s *clusterModelRegionalImplTestSuite) TestIsNewKernelEnabled() {
	s.Run("tidb version too old - should return false", func() {
		ticdcClusterEnt := &entity.TiCDCCluster{
			TenantID:  entity.ID(111),
			ProjectID: entity.ID(111),
			ClusterID: entity.ID(111),
		}
		tidbVersion, _ := utils.NewVersion("v7.0.0") // Older than MinNewkernelCompatibleTiDBVersion

		// Set MinNewkernelCompatibleTiDBVersion to v7.5.0
		origiMinNewkernelCompatibleTiDBVersion := config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion
		defer func() {
			config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = origiMinNewkernelCompatibleTiDBVersion
		}()
		config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = "v7.5.0"

		result := s.model.isNewKernelEnabled(ticdcClusterEnt.TenantID.Uint64(), ticdcClusterEnt.ProjectID.Uint64(), ticdcClusterEnt.ClusterID.Uint64(), tidbVersion, false)
		s.False(result)
	})

	s.Run("tidb version newer than minimum compatible version - not in blacklist - should return true", func() {
		ticdcClusterEnt := &entity.TiCDCCluster{
			TenantID:  entity.ID(111),
			ProjectID: entity.ID(111),
			ClusterID: entity.ID(111),
		}
		tidbVersion, _ := utils.NewVersion("v8.0.0") // Newer than minimum version

		// Set MinNewkernelCompatibleTiDBVersion to v7.5.0
		origiMinNewkernelCompatibleTiDBVersion := config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion
		defer func() {
			config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = origiMinNewkernelCompatibleTiDBVersion
		}()
		config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = "v7.5.0"

		// Mock feature gate to return false for blacklist so new kernel is enabled by default.
		allowlistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", &feature_gate.QueryFeatureGateReq{
				TenantID:  ticdcClusterEnt.TenantID.Uint64(),
				ProjectID: ticdcClusterEnt.ProjectID.Uint64(),
				ClusterID: ticdcClusterEnt.ClusterID.Uint64(),
			}, feature_gate.TiCDCNewKernelAllowlist).
			Return(false).Maybe()
		defer allowlistCall.Unset()

		blacklistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", &feature_gate.QueryFeatureGateReq{
				TenantID:  ticdcClusterEnt.TenantID.Uint64(),
				ProjectID: ticdcClusterEnt.ProjectID.Uint64(),
				ClusterID: ticdcClusterEnt.ClusterID.Uint64(),
			}, feature_gate.TiCDCNewKernelBlacklist).
			Return(false)
		defer blacklistCall.Unset()

		result := s.model.isNewKernelEnabled(ticdcClusterEnt.TenantID.Uint64(), ticdcClusterEnt.ProjectID.Uint64(), ticdcClusterEnt.ClusterID.Uint64(), tidbVersion, false)
		s.True(result)
	})

	s.Run("tidb version newer than minimum compatible version - in blacklist but migrate to new kernel - should return true", func() {
		ticdcClusterEnt := &entity.TiCDCCluster{
			TenantID:  entity.ID(111),
			ProjectID: entity.ID(111),
			ClusterID: entity.ID(111),
		}
		tidbVersion, _ := utils.NewVersion("v8.0.0") // Newer than minimum version

		// Set MinNewkernelCompatibleTiDBVersion to v7.5.0
		origiMinNewkernelCompatibleTiDBVersion := config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion
		defer func() {
			config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = origiMinNewkernelCompatibleTiDBVersion
		}()
		config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = "v7.5.0"

		// Migration returns before regular feature gate checks.
		allowlistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", &feature_gate.QueryFeatureGateReq{
				TenantID:  ticdcClusterEnt.TenantID.Uint64(),
				ProjectID: ticdcClusterEnt.ProjectID.Uint64(),
				ClusterID: ticdcClusterEnt.ClusterID.Uint64(),
			}, feature_gate.TiCDCNewKernelAllowlist).
			Return(false).Maybe()
		defer allowlistCall.Unset()

		blacklistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", &feature_gate.QueryFeatureGateReq{
				TenantID:  ticdcClusterEnt.TenantID.Uint64(),
				ProjectID: ticdcClusterEnt.ProjectID.Uint64(),
				ClusterID: ticdcClusterEnt.ClusterID.Uint64(),
			}, feature_gate.TiCDCNewKernelBlacklist).
			Return(true).Maybe()
		defer blacklistCall.Unset()

		result := s.model.isNewKernelEnabled(ticdcClusterEnt.TenantID.Uint64(), ticdcClusterEnt.ProjectID.Uint64(), ticdcClusterEnt.ClusterID.Uint64(), tidbVersion, true)
		s.True(result)
	})

	s.Run("tidb version newer than minimum compatible version - legacy allowlist ignored and not in blacklist - should return true", func() {
		ticdcClusterEnt := &entity.TiCDCCluster{
			TenantID:  entity.ID(111),
			ProjectID: entity.ID(111),
			ClusterID: entity.ID(111),
		}
		tidbVersion, _ := utils.NewVersion("v8.0.0") // Newer than minimum version

		// Set MinNewkernelCompatibleTiDBVersion to v7.5.0
		origiMinNewkernelCompatibleTiDBVersion := config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion
		defer func() {
			config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = origiMinNewkernelCompatibleTiDBVersion
		}()
		config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = "v7.5.0"

		// Mock legacy allowlist as false to ensure blacklist is the only regular opt-out.
		allowlistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", &feature_gate.QueryFeatureGateReq{
				TenantID:  ticdcClusterEnt.TenantID.Uint64(),
				ProjectID: ticdcClusterEnt.ProjectID.Uint64(),
				ClusterID: ticdcClusterEnt.ClusterID.Uint64(),
			}, feature_gate.TiCDCNewKernelAllowlist).
			Return(false).Maybe()
		defer allowlistCall.Unset()

		blacklistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", &feature_gate.QueryFeatureGateReq{
				TenantID:  ticdcClusterEnt.TenantID.Uint64(),
				ProjectID: ticdcClusterEnt.ProjectID.Uint64(),
				ClusterID: ticdcClusterEnt.ClusterID.Uint64(),
			}, feature_gate.TiCDCNewKernelBlacklist).
			Return(false)
		defer blacklistCall.Unset()

		result := s.model.isNewKernelEnabled(ticdcClusterEnt.TenantID.Uint64(), ticdcClusterEnt.ProjectID.Uint64(), ticdcClusterEnt.ClusterID.Uint64(), tidbVersion, false)
		s.True(result)
	})

	s.Run("feature gate calls with correct parameters - in blacklist - should return false", func() {
		ticdcClusterEnt := &entity.TiCDCCluster{
			TenantID:  entity.ID(111),
			ProjectID: entity.ID(111),
			ClusterID: entity.ID(111),
		}
		tidbVersion, _ := utils.NewVersion("v7.5.0")

		// Set MinNewkernelCompatibleTiDBVersion to v7.5.0
		origiMinNewkernelCompatibleTiDBVersion := config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion
		defer func() {
			config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = origiMinNewkernelCompatibleTiDBVersion
		}()
		config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion = "v7.5.0"

		allowlistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", &feature_gate.QueryFeatureGateReq{
				TenantID:  ticdcClusterEnt.TenantID.Uint64(),
				ProjectID: ticdcClusterEnt.ProjectID.Uint64(),
				ClusterID: ticdcClusterEnt.ClusterID.Uint64(),
			}, feature_gate.TiCDCNewKernelAllowlist).
			Return(true).Maybe()
		defer allowlistCall.Unset()

		blacklistCall := s.mockFeatureGate.
			On("IsFeatureEnabled", &feature_gate.QueryFeatureGateReq{
				TenantID:  ticdcClusterEnt.TenantID.Uint64(),
				ProjectID: ticdcClusterEnt.ProjectID.Uint64(),
				ClusterID: ticdcClusterEnt.ClusterID.Uint64(),
			}, feature_gate.TiCDCNewKernelBlacklist).
			Return(true)
		defer blacklistCall.Unset()

		result := s.model.isNewKernelEnabled(ticdcClusterEnt.TenantID.Uint64(), ticdcClusterEnt.ProjectID.Uint64(), ticdcClusterEnt.ClusterID.Uint64(), tidbVersion, false)
		s.False(result)
	})
}

func (s *clusterModelRegionalImplTestSuite) TestFillInstanceTypeIOPSAndThroughput() {
	testCases := []struct {
		cpu                int
		expectedIOPS       uint32
		expectedThroughput uint32
	}{
		{2, 4000, 400},
		{4, 4000, 400},
		{8, 6000, 500},
		{16, 12000, 1000},
		{32, 12000, 1000},
		{64, 12000, 1000},
		{1, 0, 0},
	}

	for _, tc := range testCases {
		instanceType := &domain.CDCInstanceType{
			MachineCPUNum: uint32(tc.cpu),
		}
		s.model.overwriteInstanceTypeIOPSAndThroughput(instanceType)
		s.Equal(tc.expectedIOPS, instanceType.IOPS, "CPU: %d", tc.cpu)
		s.Equal(tc.expectedThroughput, instanceType.ThroughputMbs, "CPU: %d", tc.cpu)
	}
}

func TestClusterModelRegionalImplTestSuite(t *testing.T) {
	suite.Run(t, new(clusterModelRegionalImplTestSuite))
}