package cluster

import (
	"context"
	"strconv"
	"testing"

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/fastflow/pkg/entity"
	regionaldedicatedv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/regional-server/v1"
	"github.com/tidbcloud/pkgv2/db/mock_db"
	"go.uber.org/mock/gomock"
	"google.golang.org/protobuf/encoding/protojson"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/adapter/regionserver/mock_regionserver"
	dsentity "github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/repository/cluster/mock_cluster"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/repository/region/mock_region"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/repository/tidbgroup/mock_tidbgroup"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/service/workflow/common"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/featuregate/mock_featuregate"
	"github.com/tidbcloud/tidb-management-service/utilities/grpc/metadata"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow/mock_workflow"
)

func TestDeleteGKEHeterogeneousDeployRequiresTidbGroupForTidb(t *testing.T) {
	t.Parallel()

	svc := &serviceImpl{}

	err := svc.DeleteGKEHeterogeneousDeploy(context.Background(), &globaldedicatedv1.DeleteGKEHeterogeneousDeployRequest{
		TenantId:      1,
		ProjectId:     2,
		ClusterId:     3,
		ComponentType: globaldedicatedv1.ComponentType_TIDB,
	})
	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeParams, err.ErrCode())
}

func TestDeleteGKEHeterogeneousDeployRejectsNonTidbWithTidbGroup(t *testing.T) {
	t.Parallel()

	svc := &serviceImpl{}

	err := svc.DeleteGKEHeterogeneousDeploy(context.Background(), &globaldedicatedv1.DeleteGKEHeterogeneousDeployRequest{
		TenantId:      1,
		ProjectId:     2,
		ClusterId:     3,
		ComponentType: globaldedicatedv1.ComponentType_TIKV,
		TidbGroupId:   4,
	})
	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeParams, err.ErrCode())
}

func TestCheckNoRunningHeterogeneousDagConflict(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	clusterID := uint64(10365974781603411878)
	bizId := gkeHeterogeneousBizId(clusterID, globaldedicatedv1.ComponentType_TIKV, 0)
	concurrencyKey := "10365974781603411878_TIKV_0_n2d-custom-8-16384"

	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return([]*entity.DagInstance{
		{
			Status: entity.DagInstanceStatusRunning,
			Vars: entity.DagInstanceVars{
				common.KeyGKEHeterogeneousConcurrencyKey: entity.DagInstanceVar{
					Value: concurrencyKey,
				},
			},
		},
	}, nil)

	svc := &serviceImpl{workflow: mockWf}
	err := svc.checkNoRunningHeterogeneousDag(context.Background(),
		applyGKEHeterogeneousDagNamePrefix,
		bizId,
		concurrencyKey)
	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeParams, err.ErrCode())
	require.Contains(t, err.Error(), "already in progress")
}

func TestCheckNoRunningHeterogeneousDagPasses(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	clusterID := uint64(10365974781603411878)
	bizId := gkeHeterogeneousBizId(clusterID, globaldedicatedv1.ComponentType_TIKV, 0)

	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return(nil, nil)

	svc := &serviceImpl{workflow: mockWf}
	err := svc.checkNoRunningHeterogeneousDag(context.Background(),
		applyGKEHeterogeneousDagNamePrefix,
		bizId,
		"10365974781603411878_TIKV_0_n2d-standard-8")
	require.NoError(t, err)
}

func TestApplyGKEHeterogeneousDeployRejectsTidbGroupNotFound(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	mockTidbGroupRepo := mock_tidbgroup.NewMockRepo(ctrl)
	mockTidbGroupRepo.EXPECT().Load(gomock.Any(), uint64(12345), uint64(99999)).
		Return(nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "not found"))

	svc := &serviceImpl{tidbGroupRepo: mockTidbGroupRepo}

	_, err := svc.ApplyGKEHeterogeneousDeploy(context.Background(), &globaldedicatedv1.ApplyGKEHeterogeneousDeployRequest{
		TenantId:                1,
		ProjectId:               2,
		ClusterId:               12345,
		ComponentType:           globaldedicatedv1.ComponentType_TIDB,
		CurrentInstanceType:     "n2-standard-8",
		AlternativeInstanceType: "n2d-standard-8",
		TidbGroupId:             99999,
	})
	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeParams, err.ErrCode())
	require.Contains(t, err.Error(), "does not exist")
	require.Contains(t, err.Error(), "12345")
	require.Contains(t, err.Error(), "99999")
}

func TestApplyGKEHeterogeneousDeployPropagatesDBError(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	mockTidbGroupRepo := mock_tidbgroup.NewMockRepo(ctrl)
	mockTidbGroupRepo.EXPECT().Load(gomock.Any(), uint64(12345), uint64(99999)).
		Return(nil, bizerr.New(bizerr.ErrCodeDB, "db error"))

	svc := &serviceImpl{tidbGroupRepo: mockTidbGroupRepo}

	_, err := svc.ApplyGKEHeterogeneousDeploy(context.Background(), &globaldedicatedv1.ApplyGKEHeterogeneousDeployRequest{
		TenantId:                1,
		ProjectId:               2,
		ClusterId:               12345,
		ComponentType:           globaldedicatedv1.ComponentType_TIDB,
		CurrentInstanceType:     "n2-standard-8",
		AlternativeInstanceType: "n2d-standard-8",
		TidbGroupId:             99999,
	})
	require.Error(t, err)
}

func TestCheckNoRunningHeterogeneousDagRejectsWhenFailedDag(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	clusterID := uint64(10365974781603411878)
	bizId := gkeHeterogeneousBizId(clusterID, globaldedicatedv1.ComponentType_TIKV, 0)
	concurrencyKey := "10365974781603411878_TIKV_0_n2d-standard-8"

	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return([]*entity.DagInstance{
		{
			Status: entity.DagInstanceStatusFailed,
			Vars: entity.DagInstanceVars{
				common.KeyGKEHeterogeneousConcurrencyKey: entity.DagInstanceVar{
					Value: concurrencyKey,
				},
			},
		},
	}, nil)

	svc := &serviceImpl{workflow: mockWf}
	err := svc.checkNoRunningHeterogeneousDag(context.Background(),
		applyGKEHeterogeneousDagNamePrefix,
		bizId,
		concurrencyKey)
	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeParams, err.ErrCode())
	require.Contains(t, err.Error(), "already in progress")
}

func TestCheckNoRunningHeterogeneousDagSkipsSuccessDag(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	clusterID := uint64(10365974781603411878)
	bizId := gkeHeterogeneousBizId(clusterID, globaldedicatedv1.ComponentType_TIKV, 0)
	concurrencyKey := "10365974781603411878_TIKV_0_n2d-standard-8"

	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return([]*entity.DagInstance{
		{
			Status: entity.DagInstanceStatusSuccess,
			Vars: entity.DagInstanceVars{
				common.KeyGKEHeterogeneousConcurrencyKey: entity.DagInstanceVar{
					Value: concurrencyKey,
				},
			},
		},
	}, nil)

	svc := &serviceImpl{workflow: mockWf}
	err := svc.checkNoRunningHeterogeneousDag(context.Background(),
		applyGKEHeterogeneousDagNamePrefix,
		bizId,
		concurrencyKey)
	require.NoError(t, err)
}

func applyReqJSON() string {
	req := &globaldedicatedv1.ApplyGKEHeterogeneousDeployRequest{
		TenantId:                1,
		ProjectId:               2,
		ClusterId:               12345,
		ComponentType:           globaldedicatedv1.ComponentType_TIKV,
		CurrentInstanceType:     "n2-standard-8",
		AlternativeInstanceType: "n2-standard-16",
	}
	b, _ := protojson.Marshal(req)
	return string(b)
}

func TestQuerySinglePendingHeterogeneousDeployApplyPending(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	bizId := gkeHeterogeneousBizId(12345, globaldedicatedv1.ComponentType_TIKV, 0)

	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return([]*entity.DagInstance{
		{
			Status: entity.DagInstanceStatusRunning,
			Vars: entity.DagInstanceVars{
				common.KeyGKEHeterogeneousApplyReq: entity.DagInstanceVar{Value: applyReqJSON()},
			},
		},
	}, nil)

	svc := &serviceImpl{workflow: mockWf}
	result, err := svc.querySinglePendingHeterogeneousDeploy(t.Context(),
		applyGKEHeterogeneousDagNamePrefix, bizId, globaldedicatedv1.ComponentType_TIKV, 0)
	require.NoError(t, err)
	require.NotNil(t, result)
	require.Equal(t, "APPLYING", result.GetStatus())
	require.Equal(t, "n2-standard-8", result.GetCurrentInstanceType())
	require.Equal(t, "n2-standard-16", result.GetAlternativeInstanceType())
}

func TestQuerySinglePendingHeterogeneousDeployDeletePending(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	bizId := gkeHeterogeneousBizId(12345, globaldedicatedv1.ComponentType_TIKV, 0)

	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(deleteGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return([]*entity.DagInstance{
		{
			Status: entity.DagInstanceStatusRunning,
		},
	}, nil)

	svc := &serviceImpl{workflow: mockWf}
	result, err := svc.querySinglePendingHeterogeneousDeploy(t.Context(),
		deleteGKEHeterogeneousDagNamePrefix, bizId, globaldedicatedv1.ComponentType_TIKV, 0)
	require.NoError(t, err)
	require.NotNil(t, result)
	require.Equal(t, "DELETING", result.GetStatus())
	require.Equal(t, uint64(0), result.GetTidbGroupId())
}

func TestQuerySinglePendingHeterogeneousDeployNoPending(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	bizId := gkeHeterogeneousBizId(12345, globaldedicatedv1.ComponentType_TIKV, 0)

	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return(nil, nil)

	svc := &serviceImpl{workflow: mockWf}
	result, err := svc.querySinglePendingHeterogeneousDeploy(t.Context(),
		applyGKEHeterogeneousDagNamePrefix, bizId, globaldedicatedv1.ComponentType_TIKV, 0)
	require.NoError(t, err)
	require.Nil(t, result)
}

func TestQuerySinglePendingHeterogeneousDeploySkipsFinished(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	bizId := gkeHeterogeneousBizId(12345, globaldedicatedv1.ComponentType_TIKV, 0)

	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return(nil, nil)

	svc := &serviceImpl{workflow: mockWf}
	result, err := svc.querySinglePendingHeterogeneousDeploy(t.Context(),
		applyGKEHeterogeneousDagNamePrefix, bizId, globaldedicatedv1.ComponentType_TIKV, 0)
	require.NoError(t, err)
	require.Nil(t, result)
}

func TestApplyGKEHeterogeneousDeployFullHappyPathTidbGroupExists(t *testing.T) {
	// Verifies: when TIDB group exists, the flow reaches RegisterDynamicDagThenRun
	// with correct prefix, BizId, and tasks.

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	clusterID := uint64(12345)
	orgID := uint64(1)
	projectID := uint64(2)
	tidbGroupID := uint64(99999)

	mockTidbGroupRepo := mock_tidbgroup.NewMockRepo(ctrl)
	mockTidbGroupRepo.EXPECT().Load(gomock.Any(), clusterID, tidbGroupID).
		Return(&dsentity.TidbNodeGroup{Id: tidbGroupID, ClusterId: clusterID}, nil)

	clusterRepo := mock_cluster.NewMockRepo(ctrl)
	clusterRepo.EXPECT().Get(gomock.Any(), clusterID, false).Return(&dsentity.Cluster{
		Cluster: &globaldedicatedv1.Cluster{
			ClusterId:       clusterID,
			OrganizationId:  orgID,
			ProjectId:       projectID,
			CloudProviderId: globalcommonv1.CloudProvider_GCP,
			RegionId:        globalcommonv1.Region_GCP_US_WEST2,
			State:           globaldedicatedv1.Cluster_AVAILABLE,
		},
	}, nil)

	regionRepo := mock_region.NewMockRepo(ctrl)
	regionRepo.EXPECT().SelectRegionById(
		gomock.Any(), globalcommonv1.Region_GCP_US_WEST2, orgID,
	).Return(&globalcommonv1.Region{
		Id:    globalcommonv1.Region_GCP_US_WEST2,
		State: globalcommonv1.Region_ACTIVE,
	}, nil)

	featGateCli := mock_featuregate.NewMockService(ctrl)
	featGateCli.EXPECT().CheckFeatureEnabled(
		gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
	).Return(true)

	regionServerCli := mock_regionserver.NewMockClient(ctrl)
	regionServerCli.EXPECT().GetGKEHeterogeneousDeploy(
		gomock.Any(), gomock.Any(), gomock.Any(),
	).Return(nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "not found"))

	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	bizId := gkeHeterogeneousBizId(clusterID, globaldedicatedv1.ComponentType_TIDB, tidbGroupID)

	// checkNoPendingHeterogeneousDeploy: two NamePrefix queries
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return(nil, nil)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(deleteGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return(nil, nil)

	// checkNoRunningHeterogeneousDag: one NamePrefix query for apply
	concurrencyKey := gkeHeterogeneousConcurrencyKey(
		clusterID, globaldedicatedv1.ComponentType_TIDB, tidbGroupID, "n2d-standard-8")
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return(nil, nil)

	// RegisterDynamicDagThenRun: called inside transaction
	mockWf.EXPECT().RegisterDynamicDagThenRun(
		gomock.Any(),
		applyGKEHeterogeneousDagNamePrefix,
		gomock.Any(),
	).DoAndReturn(func(
		_ context.Context,
		prefix string,
		opt *workflow.CreateDynamicDagOption,
	) error {
		require.Equal(t, applyGKEHeterogeneousDagNamePrefix, prefix)
		require.Equal(t, bizId, opt.BizId)
		require.Equal(t, strconv.FormatUint(orgID, 10), opt.OrgId)
		require.Equal(t, strconv.FormatUint(projectID, 10), opt.ProjectId)
		require.NotNil(t, opt.DagTasks)
		require.NotEmpty(t, opt.DagVars[common.KeyGKEHeterogeneousApplyReq])
		require.Equal(t, concurrencyKey, opt.DagVars[common.KeyGKEHeterogeneousConcurrencyKey])
		return nil
	})

	transaction := mock_db.NewFakeTransProvider()

	svc := &serviceImpl{
		tidbGroupRepo:   mockTidbGroupRepo,
		clusterRepo:     clusterRepo,
		regionRepo:      regionRepo,
		featGateCli:     featGateCli,
		regionServerCli: regionServerCli,
		workflow:        mockWf,
		transaction:     transaction,
	}

	result, err := svc.ApplyGKEHeterogeneousDeploy(
		metadata.AppendToIncomingMeta(context.Background(), &metadata.RequestMeta{
			OrgID:            orgID,
			DefaultProjectID: projectID,
			ClusterID:        clusterID,
		}), &globaldedicatedv1.ApplyGKEHeterogeneousDeployRequest{
			TenantId:                orgID,
			ProjectId:               projectID,
			ClusterId:               clusterID,
			ComponentType:           globaldedicatedv1.ComponentType_TIDB,
			CurrentInstanceType:     "n2-standard-8",
			AlternativeInstanceType: "n2d-standard-8",
			TidbGroupId:             tidbGroupID,
		})
	require.NoError(t, err)
	require.NotNil(t, result)
	require.True(t, result.GetEnabled())
}

func TestApplyGKEHeterogeneousDeployFullHappyPathTikvNoGroup(t *testing.T) {
	// Verifies: non-TIDB component without group → flow reaches RegisterDynamicDagThenRun.

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	clusterID := uint64(12345)
	orgID := uint64(1)
	projectID := uint64(2)

	clusterRepo := mock_cluster.NewMockRepo(ctrl)
	clusterRepo.EXPECT().Get(gomock.Any(), clusterID, false).Return(&dsentity.Cluster{
		Cluster: &globaldedicatedv1.Cluster{
			ClusterId:       clusterID,
			OrganizationId:  orgID,
			ProjectId:       projectID,
			CloudProviderId: globalcommonv1.CloudProvider_GCP,
			RegionId:        globalcommonv1.Region_GCP_US_WEST2,
			State:           globaldedicatedv1.Cluster_AVAILABLE,
		},
	}, nil)

	regionRepo := mock_region.NewMockRepo(ctrl)
	regionRepo.EXPECT().SelectRegionById(
		gomock.Any(), globalcommonv1.Region_GCP_US_WEST2, orgID,
	).Return(&globalcommonv1.Region{
		Id:    globalcommonv1.Region_GCP_US_WEST2,
		State: globalcommonv1.Region_ACTIVE,
	}, nil)

	featGateCli := mock_featuregate.NewMockService(ctrl)
	featGateCli.EXPECT().CheckFeatureEnabled(
		gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
	).Return(true)

	regionServerCli := mock_regionserver.NewMockClient(ctrl)
	regionServerCli.EXPECT().GetGKEHeterogeneousDeploy(
		gomock.Any(), gomock.Any(), gomock.Any(),
	).Return(nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "not found"))

	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	bizId := gkeHeterogeneousBizId(clusterID, globaldedicatedv1.ComponentType_TIKV, 0)

	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return(nil, nil)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(deleteGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return(nil, nil)

	concurrencyKey := gkeHeterogeneousConcurrencyKey(
		clusterID, globaldedicatedv1.ComponentType_TIKV, 0, "n2d-standard-8")
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return(nil, nil)

	mockWf.EXPECT().RegisterDynamicDagThenRun(
		gomock.Any(),
		applyGKEHeterogeneousDagNamePrefix,
		gomock.Any(),
	).DoAndReturn(func(
		_ context.Context,
		prefix string,
		opt *workflow.CreateDynamicDagOption,
	) error {
		require.Equal(t, applyGKEHeterogeneousDagNamePrefix, prefix)
		require.Equal(t, bizId, opt.BizId)
		require.NotNil(t, opt.DagTasks)
		require.NotEmpty(t, opt.DagVars[common.KeyGKEHeterogeneousApplyReq])
		require.Equal(t, concurrencyKey, opt.DagVars[common.KeyGKEHeterogeneousConcurrencyKey])
		return nil
	})

	transaction := mock_db.NewFakeTransProvider()

	svc := &serviceImpl{
		clusterRepo:     clusterRepo,
		regionRepo:      regionRepo,
		featGateCli:     featGateCli,
		regionServerCli: regionServerCli,
		workflow:        mockWf,
		transaction:     transaction,
	}

	result, err := svc.ApplyGKEHeterogeneousDeploy(
		metadata.AppendToIncomingMeta(context.Background(), &metadata.RequestMeta{
			OrgID:            orgID,
			DefaultProjectID: projectID,
			ClusterID:        clusterID,
		}), &globaldedicatedv1.ApplyGKEHeterogeneousDeployRequest{
			TenantId:                orgID,
			ProjectId:               projectID,
			ClusterId:               clusterID,
			ComponentType:           globaldedicatedv1.ComponentType_TIKV,
			CurrentInstanceType:     "n2-standard-8",
			AlternativeInstanceType: "n2d-standard-8",
		})
	require.NoError(t, err)
	require.NotNil(t, result)
	require.True(t, result.GetEnabled())
}

func setupGetRegionalMocks(
	ctrl *gomock.Controller,
	clusterID, orgID, projectID uint64,
) (*mock_cluster.MockRepo, *mock_region.MockRepo, *mock_regionserver.MockClient) {
	clusterRepo := mock_cluster.NewMockRepo(ctrl)
	clusterRepo.EXPECT().Get(gomock.Any(), clusterID, false).Return(&dsentity.Cluster{
		Cluster: &globaldedicatedv1.Cluster{
			ClusterId:       clusterID,
			OrganizationId:  orgID,
			ProjectId:       projectID,
			CloudProviderId: globalcommonv1.CloudProvider_GCP,
			RegionId:        globalcommonv1.Region_GCP_US_WEST2,
			State:           globaldedicatedv1.Cluster_AVAILABLE,
		},
	}, nil)

	regionRepo := mock_region.NewMockRepo(ctrl)
	regionRepo.EXPECT().SelectRegionById(
		gomock.Any(), globalcommonv1.Region_GCP_US_WEST2, orgID,
	).Return(&globalcommonv1.Region{
		Id:    globalcommonv1.Region_GCP_US_WEST2,
		State: globalcommonv1.Region_ACTIVE,
	}, nil)

	regionServerCli := mock_regionserver.NewMockClient(ctrl)
	regionServerCli.EXPECT().GetGKEHeterogeneousDeploy(
		gomock.Any(), gomock.Any(), gomock.Any(),
	).Return(nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "not found"))

	return clusterRepo, regionRepo, regionServerCli
}

func TestGetGKEHeterogeneousDeployApplyPendingWhenRegionalNotEnabled(t *testing.T) {
	// regional returns Enabled=false → check apply pending → return APPLYING

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	clusterID := uint64(12345)
	orgID := uint64(1)
	projectID := uint64(2)
	bizId := gkeHeterogeneousBizId(clusterID, globaldedicatedv1.ComponentType_TIKV, 0)

	clusterRepo, regionRepo, regionServerCli := setupGetRegionalMocks(ctrl, clusterID, orgID, projectID)

	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return([]*entity.DagInstance{
		{
			Status: entity.DagInstanceStatusRunning,
			Vars: entity.DagInstanceVars{
				common.KeyGKEHeterogeneousApplyReq: entity.DagInstanceVar{Value: applyReqJSON()},
			},
		},
	}, nil)

	svc := &serviceImpl{
		clusterRepo:     clusterRepo,
		regionRepo:      regionRepo,
		regionServerCli: regionServerCli,
		workflow:        mockWf,
	}

	result, err := svc.GetGKEHeterogeneousDeploy(context.Background(), &globaldedicatedv1.GetGKEHeterogeneousDeployRequest{
		TenantId:      orgID,
		ProjectId:     projectID,
		ClusterId:     clusterID,
		ComponentType: globaldedicatedv1.ComponentType_TIKV,
	})
	require.NoError(t, err)
	require.NotNil(t, result)
	require.Equal(t, "APPLYING", result.GetStatus())
	require.Equal(t, "n2-standard-8", result.GetCurrentInstanceType())
	require.Equal(t, "n2-standard-16", result.GetAlternativeInstanceType())
}

func TestGetGKEHeterogeneousDeployNoPendingWhenRegionalNotEnabled(t *testing.T) {
	// regional returns Enabled=false + no apply pending → return regional result

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	clusterID := uint64(12345)
	orgID := uint64(1)
	projectID := uint64(2)
	bizId := gkeHeterogeneousBizId(clusterID, globaldedicatedv1.ComponentType_TIKV, 0)

	clusterRepo, regionRepo, regionServerCli := setupGetRegionalMocks(ctrl, clusterID, orgID, projectID)

	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return(nil, nil)

	svc := &serviceImpl{
		clusterRepo:     clusterRepo,
		regionRepo:      regionRepo,
		regionServerCli: regionServerCli,
		workflow:        mockWf,
	}

	result, err := svc.GetGKEHeterogeneousDeploy(context.Background(), &globaldedicatedv1.GetGKEHeterogeneousDeployRequest{
		TenantId:      orgID,
		ProjectId:     projectID,
		ClusterId:     clusterID,
		ComponentType: globaldedicatedv1.ComponentType_TIKV,
	})
	require.NoError(t, err)
	require.NotNil(t, result)
	require.False(t, result.GetEnabled())
}

func TestGetGKEHeterogeneousDeployPropagatesRegionalError(t *testing.T) {
	// regional returns real error + no pending on either side → propagate error

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	clusterID := uint64(12345)
	orgID := uint64(1)
	projectID := uint64(2)

	clusterRepo := mock_cluster.NewMockRepo(ctrl)
	clusterRepo.EXPECT().Get(gomock.Any(), clusterID, false).
		Return(nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "cluster not found"))

	bizId := gkeHeterogeneousBizId(clusterID, globaldedicatedv1.ComponentType_TIKV, 0)
	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return(nil, nil)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(deleteGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return(nil, nil)

	svc := &serviceImpl{
		clusterRepo: clusterRepo,
		workflow:    mockWf,
	}

	result, err := svc.GetGKEHeterogeneousDeploy(context.Background(), &globaldedicatedv1.GetGKEHeterogeneousDeployRequest{
		TenantId:      orgID,
		ProjectId:     projectID,
		ClusterId:     clusterID,
		ComponentType: globaldedicatedv1.ComponentType_TIKV,
	})
	require.Error(t, err)
	require.Nil(t, result)
	require.Contains(t, err.Error(), "cluster not found")
}

func TestGetGKEHeterogeneousDeployDeletePendingWhenRegionalEnabled(t *testing.T) {
	// regional returns Enabled=true → check delete pending → return DELETING

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	clusterID := uint64(12345)
	orgID := uint64(1)
	projectID := uint64(2)
	bizId := gkeHeterogeneousBizId(clusterID, globaldedicatedv1.ComponentType_TIKV, 0)

	clusterRepo := mock_cluster.NewMockRepo(ctrl)
	clusterRepo.EXPECT().Get(gomock.Any(), clusterID, false).Return(&dsentity.Cluster{
		Cluster: &globaldedicatedv1.Cluster{
			ClusterId:       clusterID,
			OrganizationId:  orgID,
			ProjectId:       projectID,
			CloudProviderId: globalcommonv1.CloudProvider_GCP,
			RegionId:        globalcommonv1.Region_GCP_US_WEST2,
			State:           globaldedicatedv1.Cluster_AVAILABLE,
		},
	}, nil)

	regionRepo := mock_region.NewMockRepo(ctrl)
	regionRepo.EXPECT().SelectRegionById(
		gomock.Any(), globalcommonv1.Region_GCP_US_WEST2, orgID,
	).Return(&globalcommonv1.Region{
		Id:    globalcommonv1.Region_GCP_US_WEST2,
		State: globalcommonv1.Region_ACTIVE,
	}, nil)

	deploy := &regionaldedicatedv1.GKEHeterogeneousDeploy{
		Enabled:       true,
		ComponentType: regionaldedicatedv1.ComponentType_TIKV,
		InstanceTypes: []string{"n2-standard-8", "n2d-standard-8"},
		NodePoolNames: []string{"np-tikv-alt"},
	}

	regionServerCli := mock_regionserver.NewMockClient(ctrl)
	regionServerCli.EXPECT().GetGKEHeterogeneousDeploy(
		gomock.Any(), gomock.Any(), gomock.Any(),
	).Return(deploy, nil)
	regionServerCli.EXPECT().GetGKEHeterogeneousNodePools(
		gomock.Any(), gomock.Any(), gomock.Any(),
	).Return(&regionaldedicatedv1.ListGKEHeterogeneousNodePoolsResponse{
		NodePools: []*regionaldedicatedv1.GKEHeterogeneousNodePool{
			{NodePoolName: "np-tikv-alt", InstanceType: "n2d-standard-8"},
		},
	}, nil)

	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(deleteGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return([]*entity.DagInstance{
		{Status: entity.DagInstanceStatusRunning},
	}, nil)

	svc := &serviceImpl{
		clusterRepo:     clusterRepo,
		regionRepo:      regionRepo,
		regionServerCli: regionServerCli,
		workflow:        mockWf,
	}

	result, err := svc.GetGKEHeterogeneousDeploy(context.Background(), &globaldedicatedv1.GetGKEHeterogeneousDeployRequest{
		TenantId:      orgID,
		ProjectId:     projectID,
		ClusterId:     clusterID,
		ComponentType: globaldedicatedv1.ComponentType_TIKV,
	})
	require.NoError(t, err)
	require.NotNil(t, result)
	require.Equal(t, "DELETING", result.GetStatus())
}

func TestQuerySinglePendingHeterogeneousDeployFailedIsPending(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	bizId := gkeHeterogeneousBizId(12345, globaldedicatedv1.ComponentType_TIKV, 0)

	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return([]*entity.DagInstance{
		{
			Status: entity.DagInstanceStatusFailed,
			Vars: entity.DagInstanceVars{
				common.KeyGKEHeterogeneousApplyReq: entity.DagInstanceVar{Value: applyReqJSON()},
			},
		},
	}, nil)

	svc := &serviceImpl{workflow: mockWf}
	result, err := svc.querySinglePendingHeterogeneousDeploy(t.Context(),
		applyGKEHeterogeneousDagNamePrefix, bizId, globaldedicatedv1.ComponentType_TIKV, 0)
	require.NoError(t, err)
	require.NotNil(t, result)
	require.Equal(t, "APPLYING", result.GetStatus())
}

func TestGetGKEHeterogeneousDeployFailedDagReturnsApplying(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	clusterID := uint64(12345)
	orgID := uint64(1)
	projectID := uint64(2)
	bizId := gkeHeterogeneousBizId(clusterID, globaldedicatedv1.ComponentType_TIKV, 0)

	clusterRepo, regionRepo, regionServerCli := setupGetRegionalMocks(ctrl, clusterID, orgID, projectID)

	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return([]*entity.DagInstance{
		{
			Status: entity.DagInstanceStatusFailed,
			Vars: entity.DagInstanceVars{
				common.KeyGKEHeterogeneousApplyReq: entity.DagInstanceVar{Value: applyReqJSON()},
			},
		},
	}, nil)

	svc := &serviceImpl{
		clusterRepo:     clusterRepo,
		regionRepo:      regionRepo,
		regionServerCli: regionServerCli,
		workflow:        mockWf,
	}

	result, err := svc.GetGKEHeterogeneousDeploy(context.Background(), &globaldedicatedv1.GetGKEHeterogeneousDeployRequest{
		TenantId:      orgID,
		ProjectId:     projectID,
		ClusterId:     clusterID,
		ComponentType: globaldedicatedv1.ComponentType_TIKV,
	})
	require.NoError(t, err)
	require.NotNil(t, result)
	require.Equal(t, "APPLYING", result.GetStatus())
}

func TestGetGKEHeterogeneousDeployRegionalErrorWithApplyPending(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	clusterID := uint64(12345)
	orgID := uint64(1)
	projectID := uint64(2)

	clusterRepo := mock_cluster.NewMockRepo(ctrl)
	clusterRepo.EXPECT().Get(gomock.Any(), clusterID, false).
		Return(nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "cluster not found"))

	bizId := gkeHeterogeneousBizId(clusterID, globaldedicatedv1.ComponentType_TIKV, 0)
	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return([]*entity.DagInstance{
		{
			Status: entity.DagInstanceStatusRunning,
			Vars: entity.DagInstanceVars{
				common.KeyGKEHeterogeneousApplyReq: entity.DagInstanceVar{Value: applyReqJSON()},
			},
		},
	}, nil)

	svc := &serviceImpl{
		clusterRepo: clusterRepo,
		workflow:    mockWf,
	}

	result, err := svc.GetGKEHeterogeneousDeploy(context.Background(), &globaldedicatedv1.GetGKEHeterogeneousDeployRequest{
		TenantId:      orgID,
		ProjectId:     projectID,
		ClusterId:     clusterID,
		ComponentType: globaldedicatedv1.ComponentType_TIKV,
	})
	require.NoError(t, err)
	require.NotNil(t, result)
	require.Equal(t, "APPLYING", result.GetStatus())
	require.Equal(t, "n2-standard-8", result.GetCurrentInstanceType())
	require.Equal(t, "n2-standard-16", result.GetAlternativeInstanceType())
}

func TestGetGKEHeterogeneousDeployRegionalErrorWithDeletePending(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	clusterID := uint64(12345)
	orgID := uint64(1)
	projectID := uint64(2)

	clusterRepo := mock_cluster.NewMockRepo(ctrl)
	clusterRepo.EXPECT().Get(gomock.Any(), clusterID, false).
		Return(nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "cluster not found"))

	bizId := gkeHeterogeneousBizId(clusterID, globaldedicatedv1.ComponentType_TIKV, 0)
	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return(nil, nil)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(deleteGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return([]*entity.DagInstance{
		{Status: entity.DagInstanceStatusRunning},
	}, nil)

	svc := &serviceImpl{
		clusterRepo: clusterRepo,
		workflow:    mockWf,
	}

	result, err := svc.GetGKEHeterogeneousDeploy(context.Background(), &globaldedicatedv1.GetGKEHeterogeneousDeployRequest{
		TenantId:      orgID,
		ProjectId:     projectID,
		ClusterId:     clusterID,
		ComponentType: globaldedicatedv1.ComponentType_TIKV,
	})
	require.NoError(t, err)
	require.NotNil(t, result)
	require.Equal(t, "DELETING", result.GetStatus())
}

func TestGetGKEHeterogeneousDeployReturnRegionalWhenEnabledNoPending(t *testing.T) {
	// regional returns Enabled=true + no pending → return regional result

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	clusterID := uint64(12345)
	orgID := uint64(1)
	projectID := uint64(2)
	bizId := gkeHeterogeneousBizId(clusterID, globaldedicatedv1.ComponentType_TIKV, 0)

	clusterRepo := mock_cluster.NewMockRepo(ctrl)
	clusterRepo.EXPECT().Get(gomock.Any(), clusterID, false).Return(&dsentity.Cluster{
		Cluster: &globaldedicatedv1.Cluster{
			ClusterId:       clusterID,
			OrganizationId:  orgID,
			ProjectId:       projectID,
			CloudProviderId: globalcommonv1.CloudProvider_GCP,
			RegionId:        globalcommonv1.Region_GCP_US_WEST2,
			State:           globaldedicatedv1.Cluster_AVAILABLE,
		},
	}, nil)

	regionRepo := mock_region.NewMockRepo(ctrl)
	regionRepo.EXPECT().SelectRegionById(
		gomock.Any(), globalcommonv1.Region_GCP_US_WEST2, orgID,
	).Return(&globalcommonv1.Region{
		Id:    globalcommonv1.Region_GCP_US_WEST2,
		State: globalcommonv1.Region_ACTIVE,
	}, nil)

	deploy := &regionaldedicatedv1.GKEHeterogeneousDeploy{
		Enabled:       true,
		ComponentType: regionaldedicatedv1.ComponentType_TIKV,
		InstanceTypes: []string{"n2-standard-8", "n2d-standard-8"},
		NodePoolNames: []string{"np-tikv-alt"},
		PodCounts: []*regionaldedicatedv1.GKEHeterogeneousDeployPodCount{
			{MachineType: "n2d-standard-8", PodCount: 4},
		},
	}

	regionServerCli := mock_regionserver.NewMockClient(ctrl)
	regionServerCli.EXPECT().GetGKEHeterogeneousDeploy(
		gomock.Any(), gomock.Any(), gomock.Any(),
	).Return(deploy, nil)
	regionServerCli.EXPECT().GetGKEHeterogeneousNodePools(
		gomock.Any(), gomock.Any(), gomock.Any(),
	).Return(&regionaldedicatedv1.ListGKEHeterogeneousNodePoolsResponse{
		NodePools: []*regionaldedicatedv1.GKEHeterogeneousNodePool{
			{NodePoolName: "np-tikv-alt", InstanceType: "n2d-standard-8"},
		},
	}, nil)

	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(deleteGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return(nil, nil)

	svc := &serviceImpl{
		clusterRepo:     clusterRepo,
		regionRepo:      regionRepo,
		regionServerCli: regionServerCli,
		workflow:        mockWf,
	}

	result, err := svc.GetGKEHeterogeneousDeploy(context.Background(), &globaldedicatedv1.GetGKEHeterogeneousDeployRequest{
		TenantId:      orgID,
		ProjectId:     projectID,
		ClusterId:     clusterID,
		ComponentType: globaldedicatedv1.ComponentType_TIKV,
	})
	require.NoError(t, err)
	require.NotNil(t, result)
	require.True(t, result.GetEnabled())
	require.Len(t, result.GetPodCounts(), 1)
	require.Equal(t, "n2d-standard-8", result.GetPodCounts()[0].GetMachineType())
	require.Equal(t, int32(4), result.GetPodCounts()[0].GetPodCount())
}

func TestDeleteGKEHeterogeneousDeployRejectsWhenApplyDAGPending(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	clusterID := uint64(12345)
	componentType := globaldedicatedv1.ComponentType_TIKV

	clusterRepo := mock_cluster.NewMockRepo(ctrl)
	clusterRepo.EXPECT().Get(gomock.Any(), clusterID, false).Return(&dsentity.Cluster{
		Cluster: &globaldedicatedv1.Cluster{
			ClusterId:       clusterID,
			OrganizationId:  1,
			ProjectId:       2,
			CloudProviderId: globalcommonv1.CloudProvider_GCP,
			RegionId:        globalcommonv1.Region_GCP_US_WEST2,
			State:           globaldedicatedv1.Cluster_AVAILABLE,
		},
	}, nil)

	regionRepo := mock_region.NewMockRepo(ctrl)
	regionRepo.EXPECT().SelectRegionById(
		gomock.Any(), globalcommonv1.Region_GCP_US_WEST2, uint64(1),
	).Return(&globalcommonv1.Region{
		Id:    globalcommonv1.Region_GCP_US_WEST2,
		State: globalcommonv1.Region_ACTIVE,
	}, nil)

	bizId := gkeHeterogeneousBizId(clusterID, componentType, 0)
	concurrencyVal := gkeHeterogeneousConcurrencyKey(clusterID, componentType, 0, "n2-standard-8")
	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return([]*entity.DagInstance{{
		Status: entity.DagInstanceStatusRunning,
		Vars: entity.DagInstanceVars{
			common.KeyGKEHeterogeneousConcurrencyKey: {Value: concurrencyVal},
		},
	}}, nil)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(deleteGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return(nil, nil)

	svc := &serviceImpl{
		clusterRepo:     clusterRepo,
		regionRepo:      regionRepo,
		workflow:        mockWf,
		regionServerCli: mock_regionserver.NewMockClient(ctrl),
	}

	err := svc.DeleteGKEHeterogeneousDeploy(context.Background(), &globaldedicatedv1.DeleteGKEHeterogeneousDeployRequest{
		TenantId:      1,
		ProjectId:     2,
		ClusterId:     clusterID,
		ComponentType: componentType,
	})
	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeParams, err.ErrCode())
	require.Contains(t, err.Error(), "already in progress")
}

func TestDeleteGKEHeterogeneousDeployRejectedWhenDeleteDAGPending(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	clusterID := uint64(12345)
	componentType := globaldedicatedv1.ComponentType_TIKV

	clusterRepo := mock_cluster.NewMockRepo(ctrl)
	clusterRepo.EXPECT().Get(gomock.Any(), clusterID, false).Return(&dsentity.Cluster{
		Cluster: &globaldedicatedv1.Cluster{
			ClusterId:       clusterID,
			OrganizationId:  1,
			ProjectId:       2,
			CloudProviderId: globalcommonv1.CloudProvider_GCP,
			RegionId:        globalcommonv1.Region_GCP_US_WEST2,
			State:           globaldedicatedv1.Cluster_AVAILABLE,
		},
	}, nil)

	regionRepo := mock_region.NewMockRepo(ctrl)
	regionRepo.EXPECT().SelectRegionById(
		gomock.Any(), globalcommonv1.Region_GCP_US_WEST2, uint64(1),
	).Return(&globalcommonv1.Region{
		Id:    globalcommonv1.Region_GCP_US_WEST2,
		State: globalcommonv1.Region_ACTIVE,
	}, nil)

	bizId := gkeHeterogeneousBizId(clusterID, componentType, 0)
	concurrencyVal := gkeHeterogeneousConcurrencyKey(clusterID, componentType, 0, "n2-standard-8")
	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return(nil, nil)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(deleteGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return([]*entity.DagInstance{{
		Status: entity.DagInstanceStatusRunning,
		Vars: entity.DagInstanceVars{
			common.KeyGKEHeterogeneousConcurrencyKey: {Value: concurrencyVal},
		},
	}}, nil)

	svc := &serviceImpl{
		clusterRepo:     clusterRepo,
		regionRepo:      regionRepo,
		workflow:        mockWf,
		regionServerCli: mock_regionserver.NewMockClient(ctrl),
	}

	err := svc.DeleteGKEHeterogeneousDeploy(context.Background(), &globaldedicatedv1.DeleteGKEHeterogeneousDeployRequest{
		TenantId:      1,
		ProjectId:     2,
		ClusterId:     clusterID,
		ComponentType: componentType,
	})
	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeParams, err.ErrCode())
	require.Contains(t, err.Error(), "already in progress")
}

func TestDeleteGKEHeterogeneousDeploySucceedsWhenNoPending(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	clusterID := uint64(90001)
	componentType := globaldedicatedv1.ComponentType_TIKV
	orgID := uint64(1)
	projectID := uint64(2)

	clusterRepo := mock_cluster.NewMockRepo(ctrl)
	clusterRepo.EXPECT().Get(gomock.Any(), clusterID, false).Return(&dsentity.Cluster{
		Cluster: &globaldedicatedv1.Cluster{
			ClusterId:       clusterID,
			OrganizationId:  orgID,
			ProjectId:       projectID,
			CloudProviderId: globalcommonv1.CloudProvider_GCP,
			RegionId:        globalcommonv1.Region_GCP_US_WEST2,
			State:           globaldedicatedv1.Cluster_AVAILABLE,
		},
	}, nil)

	regionRepo := mock_region.NewMockRepo(ctrl)
	regionRepo.EXPECT().SelectRegionById(
		gomock.Any(), globalcommonv1.Region_GCP_US_WEST2, orgID,
	).Return(&globalcommonv1.Region{
		Id:    globalcommonv1.Region_GCP_US_WEST2,
		State: globalcommonv1.Region_ACTIVE,
	}, nil)

	bizId := gkeHeterogeneousBizId(clusterID, componentType, 0)
	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return(nil, nil)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(deleteGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return(nil, nil)

	regionServerCli := mock_regionserver.NewMockClient(ctrl)
	regionServerCli.EXPECT().GetGKEHeterogeneousDeploy(
		gomock.Any(), gomock.Any(), gomock.Any(),
	).Return(&regionaldedicatedv1.GKEHeterogeneousDeploy{
		Enabled:       true,
		ComponentType: regionaldedicatedv1.ComponentType_TIKV,
		InstanceTypes: []string{"n2-standard-8", "n2d-standard-8"},
	}, nil)

	cKey := gkeHeterogeneousConcurrencyKey(clusterID, componentType, 0, "n2d-standard-8")
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(deleteGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return(nil, nil)
	mockWf.EXPECT().RegisterDynamicDagThenRun(
		gomock.Any(),
		deleteGKEHeterogeneousDagNamePrefix,
		gomock.Any(),
	).DoAndReturn(func(_ context.Context, p string, o *workflow.CreateDynamicDagOption) error {
		require.Equal(t, deleteGKEHeterogeneousDagNamePrefix, p)
		require.Equal(t, bizId, o.BizId)
		require.NotEmpty(t, o.DagVars[common.KeyGKEHeterogeneousDeleteReq])
		require.Equal(t, cKey, o.DagVars[common.KeyGKEHeterogeneousConcurrencyKey])
		return nil
	})

	svc := &serviceImpl{
		clusterRepo:     clusterRepo,
		regionRepo:      regionRepo,
		regionServerCli: regionServerCli,
		workflow:        mockWf,
		transaction:     mock_db.NewFakeTransProvider(),
	}

	ctx := metadata.AppendToIncomingMeta(context.Background(), &metadata.RequestMeta{
		OrgID:            orgID,
		DefaultProjectID: projectID,
		ClusterID:        clusterID,
	})
	err := svc.DeleteGKEHeterogeneousDeploy(ctx, &globaldedicatedv1.DeleteGKEHeterogeneousDeployRequest{
		TenantId:      orgID,
		ProjectId:     projectID,
		ClusterId:     clusterID,
		ComponentType: componentType,
	})
	require.NoError(t, err)
}