package server

import (
	"context"
	"strings"
	"testing"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	dplanemgrv1 "github.com/tidbcloud/cluster-service-ng/api/dataplane-manager/v1"
	"github.com/tidbcloud/cluster-service-ng/config"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow/dag"
	dataplanemgrmocks "github.com/tidbcloud/cluster-service-ng/pkg/dataplanemgr/mocks"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/executor"
	executormocks "github.com/tidbcloud/cluster-service-ng/pkg/runtime/executor/mocks"
	repomnocks "github.com/tidbcloud/cluster-service-ng/pkg/storage/repo/mocks"
	"github.com/tidbcloud/cluster-service-ng/pkg/utils"
	"go.opentelemetry.io/otel"
	sdktrace "go.opentelemetry.io/otel/sdk/trace"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

func setupDMClusterDAGMocks(dagExecutor *executormocks.RemoteExecutor) {
	dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanApplyDMCluster, mock.AnythingOfType("executor.MatchFilter")).Return(executor.NewExecTaskIdStatus(), nil).Maybe()
}

func setupDMNodePoolMocks(dataplaneClient *dataplanemgrmocks.Client) {
	nodePoolResp := &dplanemgrv1.ListNodePoolsResponse{
		NodePools: []*dplanemgrv1.NodePool{
			{
				CpuCore:               8,
				MemoryGi:              16,
				NodeType:              dplanemgrv1.NodeType_Dedicated,
				AvailableCpuMilliCore: 7000,
				AvailableMemoryMi:     15000,
				WithLocalDisk:         false,
			},
			{
				CpuCore:               4,
				MemoryGi:              8,
				NodeType:              dplanemgrv1.NodeType_Dedicated,
				AvailableCpuMilliCore: 3500,
				AvailableMemoryMi:     7500,
				WithLocalDisk:         false,
			},
		},
	}
	dataplaneClient.On("ListNodePools", mock.Anything, mock.MatchedBy(func(req *dplanemgrv1.ListNodePoolsRequest) bool {
		return req.DataPlaneId == "test-dataplane"
	})).Return(nodePoolResp, nil).Maybe()
}

func TestCreateDMCluster(t *testing.T) {
	tests := []struct {
		name          string
		dmCluster     *entity.DMCluster
		bizCfg        *config.BizConfig
		setupMocks    func(*repomnocks.DMClusterRepo, *repomnocks.SharedPoolRepo, *executormocks.RemoteExecutor, *dataplanemgrmocks.Client)
		expectedError string
		expectedCode  codes.Code
		validate      func(*testing.T, *entity.DMCluster, error)
	}{
		{
			name: "successful create dm cluster",
			dmCluster: &entity.DMCluster{
				Id:           "testdmcluster",
				SharedPoolId: "testsharedpool",
				MasterNodeQuota: &apiv1.NodeQuota{
					NodeCount: 3,
					NodeSpec: &apiv1.NodeSpec{
						Cpu:      8,
						MemoryGi: 16,
					},
				},
				WorkerNodeQuota: &apiv1.NodeQuota{
					NodeCount: 2,
					NodeSpec: &apiv1.NodeSpec{
						Cpu:      4,
						MemoryGi: 8,
					},
				},
			},
			bizCfg: &config.BizConfig{},
			setupMocks: func(dmClusterRepo *repomnocks.DMClusterRepo, sharedPoolRepo *repomnocks.SharedPoolRepo, dagExecutor *executormocks.RemoteExecutor, dataplaneClient *dataplanemgrmocks.Client) {
				setupDMClusterDAGMocks(dagExecutor)
				setupDMNodePoolMocks(dataplaneClient)

				dmClusterRepo.On("Load", mock.Anything, "testdmcluster", true).Return(nil, status.Error(codes.NotFound, "not found"))

				sharedPool := &entity.SharedPool{
					SharedPool: apiv1.SharedPool{
						Id:            "testsharedpool",
						Phase:         apiv1.SharedPool_AVAILABLE,
						DataplaneName: "test-dataplane",
					},
				}
				sharedPoolRepo.On("Load", mock.Anything, "testsharedpool", false).Return(sharedPool, nil)
				dmClusterRepo.On("Save", mock.Anything, mock.Anything).Return(nil)
				dagExecutor.On("Submit", mock.Anything, dag.PlanApplyDMCluster, mock.MatchedBy(func(args actions.ActionArgs) bool {
					v, err := args.GetString(workflow.DMBackendKey)
					return err == nil && v == entity.DMBackendOperatorV2
				}), mock.Anything).Return(nil)
			},
			validate: func(t *testing.T, result *entity.DMCluster, err error) {
				t.Helper()
				require.NoError(t, err)
				require.NotNil(t, result)
				assert.Equal(t, "testdmcluster", result.GetId())
				assert.Equal(t, apiv1.DMCluster_CREATING, result.GetPhase())
			},
		},
		{
			name: "successful create dm cluster with operator v2 backend enabled",
			dmCluster: &entity.DMCluster{
				Id:           "testdmcluster",
				SharedPoolId: "testsharedpool",
				MasterNodeQuota: &apiv1.NodeQuota{
					NodeCount: 3,
					NodeSpec: &apiv1.NodeSpec{
						Cpu:      8,
						MemoryGi: 16,
					},
				},
				WorkerNodeQuota: &apiv1.NodeQuota{
					NodeCount: 2,
					NodeSpec: &apiv1.NodeSpec{
						Cpu:      4,
						MemoryGi: 8,
					},
				},
			},
			setupMocks: func(dmClusterRepo *repomnocks.DMClusterRepo, sharedPoolRepo *repomnocks.SharedPoolRepo, dagExecutor *executormocks.RemoteExecutor, dataplaneClient *dataplanemgrmocks.Client) {
				setupDMClusterDAGMocks(dagExecutor)
				setupDMNodePoolMocks(dataplaneClient)

				dmClusterRepo.On("Load", mock.Anything, "testdmcluster", true).Return(nil, status.Error(codes.NotFound, "not found"))

				sharedPool := &entity.SharedPool{
					SharedPool: apiv1.SharedPool{
						Id:            "testsharedpool",
						Phase:         apiv1.SharedPool_AVAILABLE,
						DataplaneName: "test-dataplane",
					},
				}
				sharedPoolRepo.On("Load", mock.Anything, "testsharedpool", false).Return(sharedPool, nil)
				dmClusterRepo.On("Save", mock.Anything, mock.Anything).Return(nil)
				dagExecutor.On("Submit", mock.Anything, dag.PlanApplyDMCluster, mock.MatchedBy(func(args actions.ActionArgs) bool {
					v, err := args.GetString(workflow.DMBackendKey)
					updateImage, updateImageErr := args.GetBool(workflow.DMUpdateImageKey)
					return err == nil && v == entity.DMBackendOperatorV2 &&
						updateImageErr == nil && updateImage
				}), mock.Anything).Return(nil)
			},
			validate: func(t *testing.T, result *entity.DMCluster, err error) {
				t.Helper()
				require.NoError(t, err)
				require.NotNil(t, result)
				assert.Equal(t, apiv1.DMCluster_CREATING, result.GetPhase())
			},
		},
		{
			name: "empty dm cluster id",
			dmCluster: &entity.DMCluster{
				Id:           "",
				SharedPoolId: "testsharedpool",
			},
			bizCfg: &config.BizConfig{},
			setupMocks: func(dmClusterRepo *repomnocks.DMClusterRepo, sharedPoolRepo *repomnocks.SharedPoolRepo, dagExecutor *executormocks.RemoteExecutor, dataplaneClient *dataplanemgrmocks.Client) {
			},
			expectedError: "dm cluster id is required",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name: "dm cluster id exceeds operator v2 group name limit",
			dmCluster: &entity.DMCluster{
				Id: strings.Repeat("a", 54),
			},
			setupMocks: func(dmClusterRepo *repomnocks.DMClusterRepo, sharedPoolRepo *repomnocks.SharedPoolRepo, dagExecutor *executormocks.RemoteExecutor, dataplaneClient *dataplanemgrmocks.Client) {
			},
			expectedError: "exceeds max length",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name: "empty shared pool id",
			dmCluster: &entity.DMCluster{
				Id:           "testdmcluster",
				SharedPoolId: "",
			},
			setupMocks: func(dmClusterRepo *repomnocks.DMClusterRepo, sharedPoolRepo *repomnocks.SharedPoolRepo, dagExecutor *executormocks.RemoteExecutor, dataplaneClient *dataplanemgrmocks.Client) {
			},
			expectedError: "shared pool id is required",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name: "missing master node quota",
			dmCluster: &entity.DMCluster{
				Id:              "testdmcluster",
				SharedPoolId:    "testsharedpool",
				MasterNodeQuota: nil,
				WorkerNodeQuota: &apiv1.NodeQuota{NodeCount: 2},
			},
			setupMocks: func(dmClusterRepo *repomnocks.DMClusterRepo, sharedPoolRepo *repomnocks.SharedPoolRepo, dagExecutor *executormocks.RemoteExecutor, dataplaneClient *dataplanemgrmocks.Client) {
			},
			expectedError: "master node quota is required",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name: "missing worker node quota",
			dmCluster: &entity.DMCluster{
				Id:              "testdmcluster",
				SharedPoolId:    "testsharedpool",
				MasterNodeQuota: &apiv1.NodeQuota{NodeCount: 3},
				WorkerNodeQuota: nil,
			},
			setupMocks: func(dmClusterRepo *repomnocks.DMClusterRepo, sharedPoolRepo *repomnocks.SharedPoolRepo, dagExecutor *executormocks.RemoteExecutor, dataplaneClient *dataplanemgrmocks.Client) {
			},
			expectedError: "worker node quota is required",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name: "dm cluster already exists",
			dmCluster: &entity.DMCluster{
				Id:              "existingdmcluster",
				SharedPoolId:    "testsharedpool",
				MasterNodeQuota: &apiv1.NodeQuota{NodeCount: 3},
				WorkerNodeQuota: &apiv1.NodeQuota{NodeCount: 2},
			},
			setupMocks: func(dmClusterRepo *repomnocks.DMClusterRepo, sharedPoolRepo *repomnocks.SharedPoolRepo, dagExecutor *executormocks.RemoteExecutor, dataplaneClient *dataplanemgrmocks.Client) {
				existingCluster := &entity.DMCluster{
					Id:           "existingdmcluster",
					SharedPoolId: "testsharedpool",
				}
				dmClusterRepo.On("Load", mock.Anything, "existingdmcluster", true).Return(existingCluster, nil)
			},
			expectedError: "dm cluster existingdmcluster already exists",
			expectedCode:  codes.AlreadyExists,
		},
		{
			name: "shared pool unavailable",
			dmCluster: &entity.DMCluster{
				Id:              "testdmcluster",
				SharedPoolId:    "unavailable-pool",
				MasterNodeQuota: &apiv1.NodeQuota{NodeCount: 3},
				WorkerNodeQuota: &apiv1.NodeQuota{NodeCount: 2},
			},
			setupMocks: func(dmClusterRepo *repomnocks.DMClusterRepo, sharedPoolRepo *repomnocks.SharedPoolRepo, dagExecutor *executormocks.RemoteExecutor, dataplaneClient *dataplanemgrmocks.Client) {
				dmClusterRepo.On("Load", mock.Anything, "testdmcluster", true).Return(nil, status.Error(codes.NotFound, "not found"))
				sharedPool := &entity.SharedPool{
					SharedPool: apiv1.SharedPool{
						Id:    "unavailable-pool",
						Phase: apiv1.SharedPool_UNAVAILABLE,
					},
				}
				sharedPoolRepo.On("Load", mock.Anything, "unavailable-pool", false).Return(sharedPool, nil)
			},
			expectedError: "shared pool is unavailable",
			expectedCode:  codes.FailedPrecondition,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			dmClusterRepo := repomnocks.NewDMClusterRepo(t)
			dmClusterRepo.On("GetBackend", mock.Anything, mock.Anything, mock.Anything).Return(entity.DMBackendLegacy, nil).Maybe()
			sharedPoolRepo := repomnocks.NewSharedPoolRepo(t)
			dagExecutor := executormocks.NewRemoteExecutor(t)
			dataplaneClient := dataplanemgrmocks.NewClient(t)

			tt.setupMocks(dmClusterRepo, sharedPoolRepo, dagExecutor, dataplaneClient)

			server := &serverImpl{
				dmClusterRepo:   dmClusterRepo,
				sharedPoolRepo:  sharedPoolRepo,
				dagExecutor:     dagExecutor,
				dataplaneClient: dataplaneClient,
				transactioner:   &mockTransactioner{},
				bizCfg:          tt.bizCfg,
			}

			result, err := server.CreateDMCluster(context.Background(), tt.dmCluster)

			switch {
			case tt.expectedError != "":
				require.Error(t, err)
				assert.Contains(t, err.Error(), tt.expectedError)
				if tt.expectedCode != codes.OK {
					assert.Equal(t, tt.expectedCode, status.Code(err))
				}
				assert.Nil(t, result)

			case tt.validate != nil:
				tt.validate(t, result, err)

			default:
				require.NoError(t, err)
				require.NotNil(t, result)
			}
		})
	}
}

func TestGetDMCluster(t *testing.T) {
	tests := []struct {
		name          string
		sharedPoolId  string
		dmClusterId   string
		setupMocks    func(*repomnocks.DMClusterRepo, *executormocks.RemoteExecutor)
		expectedError string
		expectedCode  codes.Code
		validate      func(*testing.T, *entity.DMCluster, error)
	}{
		{
			name:         "successful get dm cluster",
			sharedPoolId: "testsharedpool",
			dmClusterId:  "testdmcluster",
			setupMocks: func(dmClusterRepo *repomnocks.DMClusterRepo, dagExecutor *executormocks.RemoteExecutor) {
				dmCluster := &entity.DMCluster{
					Id:           "testdmcluster",
					SharedPoolId: "testsharedpool",
					Phase:        apiv1.DMCluster_AVAILABLE,
				}
				dmClusterRepo.On("Load", mock.Anything, "testdmcluster", false).Return(dmCluster, nil)
				setupDMClusterDAGMocks(dagExecutor)
			},
			validate: func(t *testing.T, result *entity.DMCluster, err error) {
				t.Helper()
				require.NoError(t, err)
				require.NotNil(t, result)
				assert.Equal(t, "testdmcluster", result.GetId())
				assert.Equal(t, apiv1.DMCluster_AVAILABLE, result.GetPhase())
			},
		},
		{
			name:          "empty shared pool id",
			sharedPoolId:  "",
			dmClusterId:   "testdmcluster",
			setupMocks:    func(dmClusterRepo *repomnocks.DMClusterRepo, dagExecutor *executormocks.RemoteExecutor) {},
			expectedError: "shared pool id is required",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name:          "empty dm cluster id",
			sharedPoolId:  "testsharedpool",
			dmClusterId:   "",
			setupMocks:    func(dmClusterRepo *repomnocks.DMClusterRepo, dagExecutor *executormocks.RemoteExecutor) {},
			expectedError: "dm cluster id is required",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name:         "dm cluster not found",
			sharedPoolId: "testsharedpool",
			dmClusterId:  "non-existing",
			setupMocks: func(dmClusterRepo *repomnocks.DMClusterRepo, dagExecutor *executormocks.RemoteExecutor) {
				dmClusterRepo.On("Load", mock.Anything, "non-existing", false).Return(nil, status.Error(codes.NotFound, "not found"))
			},
			expectedError: "dm cluster non-existing not found",
			expectedCode:  codes.NotFound,
		},
		{
			name:         "dm cluster does not belong to shared pool",
			sharedPoolId: "wrong-shared-pool",
			dmClusterId:  "testdmcluster",
			setupMocks: func(dmClusterRepo *repomnocks.DMClusterRepo, dagExecutor *executormocks.RemoteExecutor) {
				dmCluster := &entity.DMCluster{
					Id:           "testdmcluster",
					SharedPoolId: "other-shared-pool",
				}
				dmClusterRepo.On("Load", mock.Anything, "testdmcluster", false).Return(dmCluster, nil)
			},
			expectedError: "dm cluster testdmcluster does not belong to shared pool wrong-shared-pool",
			expectedCode:  codes.PermissionDenied,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			dmClusterRepo := repomnocks.NewDMClusterRepo(t)
			dmClusterRepo.On("GetBackend", mock.Anything, mock.Anything, mock.Anything).Return(entity.DMBackendLegacy, nil).Maybe()
			dagExecutor := executormocks.NewRemoteExecutor(t)

			tt.setupMocks(dmClusterRepo, dagExecutor)

			server := &serverImpl{
				dmClusterRepo:  dmClusterRepo,
				sharedPoolRepo: availableSharedPoolRepoForMutationTest(),
				dagExecutor:    dagExecutor,
			}

			result, err := server.GetDMCluster(context.Background(), tt.sharedPoolId, tt.dmClusterId)

			switch {
			case tt.expectedError != "":
				require.Error(t, err)
				assert.Contains(t, err.Error(), tt.expectedError)
				if tt.expectedCode != codes.OK {
					assert.Equal(t, tt.expectedCode, status.Code(err))
				}
				assert.Nil(t, result)

			case tt.validate != nil:
				tt.validate(t, result, err)

			default:
				require.NoError(t, err)
				require.NotNil(t, result)
			}
		})
	}
}

func TestListDMClusters(t *testing.T) {
	tests := []struct {
		name          string
		sharedPoolId  string
		setupMocks    func(*repomnocks.DMClusterRepo, *executormocks.RemoteExecutor)
		expectedError string
		expectedCode  codes.Code
		validate      func(*testing.T, []*entity.DMCluster, error)
	}{
		{
			name:         "successful list dm clusters",
			sharedPoolId: "testsharedpool",
			setupMocks: func(dmClusterRepo *repomnocks.DMClusterRepo, dagExecutor *executormocks.RemoteExecutor) {
				dmClusters := []*entity.DMCluster{
					{
						Id:           "dm-cluster-1",
						SharedPoolId: "testsharedpool",
						Phase:        apiv1.DMCluster_AVAILABLE,
					},
					{
						Id:           "dm-cluster-2",
						SharedPoolId: "testsharedpool",
						Phase:        apiv1.DMCluster_CREATING,
					},
				}
				dmClusterRepo.On("List", mock.Anything, mock.Anything).Return(dmClusters, nil)
				setupDMClusterDAGMocks(dagExecutor)
			},
			validate: func(t *testing.T, result []*entity.DMCluster, err error) {
				t.Helper()
				require.NoError(t, err)
				require.NotNil(t, result)
				assert.Len(t, result, 2)
			},
		},
		{
			name:          "empty shared pool id",
			sharedPoolId:  "",
			setupMocks:    func(dmClusterRepo *repomnocks.DMClusterRepo, dagExecutor *executormocks.RemoteExecutor) {},
			expectedError: "shared pool id is required",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name:         "list returns empty",
			sharedPoolId: "empty-shared-pool",
			setupMocks: func(dmClusterRepo *repomnocks.DMClusterRepo, dagExecutor *executormocks.RemoteExecutor) {
				dmClusterRepo.On("List", mock.Anything, mock.Anything).Return([]*entity.DMCluster{}, nil)
				setupDMClusterDAGMocks(dagExecutor)
			},
			validate: func(t *testing.T, result []*entity.DMCluster, err error) {
				t.Helper()
				require.NoError(t, err)
				assert.Empty(t, result)
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			dmClusterRepo := repomnocks.NewDMClusterRepo(t)
			dmClusterRepo.On("GetBackend", mock.Anything, mock.Anything, mock.Anything).Return(entity.DMBackendLegacy, nil).Maybe()
			dagExecutor := executormocks.NewRemoteExecutor(t)

			tt.setupMocks(dmClusterRepo, dagExecutor)

			server := &serverImpl{
				dmClusterRepo: dmClusterRepo,
				dagExecutor:   dagExecutor,
			}

			result, err := server.ListDMClusters(context.Background(), tt.sharedPoolId)

			switch {
			case tt.expectedError != "":
				require.Error(t, err)
				assert.Contains(t, err.Error(), tt.expectedError)
				if tt.expectedCode != codes.OK {
					assert.Equal(t, tt.expectedCode, status.Code(err))
				}
				assert.Nil(t, result)

			case tt.validate != nil:
				tt.validate(t, result, err)

			default:
				require.NoError(t, err)
				require.NotNil(t, result)
			}
		})
	}
}

func TestScaleDMCluster(t *testing.T) {
	prevTracerProvider := otel.GetTracerProvider()
	tracerProvider := sdktrace.NewTracerProvider()
	otel.SetTracerProvider(tracerProvider)
	t.Cleanup(func() {
		_ = tracerProvider.Shutdown(context.Background())
		otel.SetTracerProvider(prevTracerProvider)
	})

	tests := []struct {
		name            string
		sharedPoolId    string
		dmClusterId     string
		masterNodeQuota *entity.NodeQuota
		workerNodeQuota *entity.NodeQuota
		force           bool
		setupMocks      func(*repomnocks.DMClusterRepo, *repomnocks.SharedPoolRepo, *executormocks.RemoteExecutor, *dataplanemgrmocks.Client)
		expectedError   string
		expectedCode    codes.Code
		validate        func(*testing.T, *entity.DMCluster, error)
	}{
		{
			name:         "successful scale dm cluster",
			sharedPoolId: "testsharedpool",
			dmClusterId:  "testdmcluster",
			masterNodeQuota: &apiv1.NodeQuota{
				NodeCount: 5,
				NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16},
			},
			workerNodeQuota: &apiv1.NodeQuota{
				NodeCount: 3,
				NodeSpec:  &apiv1.NodeSpec{Cpu: 4, MemoryGi: 8},
			},
			force: false,
			setupMocks: func(dmClusterRepo *repomnocks.DMClusterRepo, sharedPoolRepo *repomnocks.SharedPoolRepo, dagExecutor *executormocks.RemoteExecutor, dataplaneClient *dataplanemgrmocks.Client) {
				setupDMClusterDAGMocks(dagExecutor)
				setupDMNodePoolMocks(dataplaneClient)

				dmCluster := &entity.DMCluster{
					Id:            "testdmcluster",
					SharedPoolId:  "testsharedpool",
					DataplaneName: "test-dataplane",
					Phase:         apiv1.DMCluster_AVAILABLE,
					MasterNodeQuota: &apiv1.NodeQuota{
						NodeCount: 3,
						NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16},
					},
					WorkerNodeQuota: &apiv1.NodeQuota{
						NodeCount: 2,
						NodeSpec:  &apiv1.NodeSpec{Cpu: 4, MemoryGi: 8},
					},
				}
				dmClusterRepo.On("Load", mock.Anything, "testdmcluster", false).Return(dmCluster, nil)

				sharedPool := &entity.SharedPool{
					SharedPool: apiv1.SharedPool{
						Id:            "testsharedpool",
						Phase:         apiv1.SharedPool_AVAILABLE,
						DataplaneName: "test-dataplane",
					},
				}
				sharedPoolRepo.On("Load", mock.Anything, "testsharedpool", false).Return(sharedPool, nil)

				dmClusterRepo.On("Save", mock.Anything, mock.Anything).Return(nil)
				dagExecutor.On("Submit", mock.Anything, dag.PlanApplyDMCluster, mock.MatchedBy(func(args actions.ActionArgs) bool {
					traceId, err := args.GetString(utils.TraceIdKey)
					return err == nil && traceId != ""
				}), mock.MatchedBy(func(execID string) bool {
					emptyTraceExecID := "scale-dmcluster-testdmcluster-" + utils.FnvHash("")
					return strings.HasPrefix(execID, "scale-dmcluster-testdmcluster-") && execID != emptyTraceExecID
				})).Return(nil)
			},
			validate: func(t *testing.T, result *entity.DMCluster, err error) {
				t.Helper()
				require.NoError(t, err)
				require.NotNil(t, result)
				assert.Equal(t, uint32(5), result.GetMasterNodeQuota().GetNodeCount())
				assert.Equal(t, uint32(3), result.GetWorkerNodeQuota().GetNodeCount())
			},
		},
		{
			name:            "empty shared pool id",
			sharedPoolId:    "",
			dmClusterId:     "testdmcluster",
			masterNodeQuota: nil,
			workerNodeQuota: nil,
			force:           false,
			setupMocks: func(dmClusterRepo *repomnocks.DMClusterRepo, sharedPoolRepo *repomnocks.SharedPoolRepo, dagExecutor *executormocks.RemoteExecutor, dataplaneClient *dataplanemgrmocks.Client) {
			},
			expectedError: "shared pool id is required",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name:            "empty dm cluster id",
			sharedPoolId:    "testsharedpool",
			dmClusterId:     "",
			masterNodeQuota: nil,
			workerNodeQuota: nil,
			force:           false,
			setupMocks: func(dmClusterRepo *repomnocks.DMClusterRepo, sharedPoolRepo *repomnocks.SharedPoolRepo, dagExecutor *executormocks.RemoteExecutor, dataplaneClient *dataplanemgrmocks.Client) {
			},
			expectedError: "dm cluster id is required",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name:         "dm cluster not available",
			sharedPoolId: "testsharedpool",
			dmClusterId:  "testdmcluster",
			masterNodeQuota: &apiv1.NodeQuota{
				NodeCount: 5,
			},
			workerNodeQuota: nil,
			force:           false,
			setupMocks: func(dmClusterRepo *repomnocks.DMClusterRepo, sharedPoolRepo *repomnocks.SharedPoolRepo, dagExecutor *executormocks.RemoteExecutor, dataplaneClient *dataplanemgrmocks.Client) {
				dmCluster := &entity.DMCluster{
					Id:           "testdmcluster",
					SharedPoolId: "testsharedpool",
					Phase:        apiv1.DMCluster_MODIFYING,
				}
				dmClusterRepo.On("Load", mock.Anything, "testdmcluster", false).Return(dmCluster, nil)
			},
			expectedError: "dm cluster is not available",
			expectedCode:  codes.FailedPrecondition,
		},
		{
			name:         "master node count is zero",
			sharedPoolId: "testsharedpool",
			dmClusterId:  "testdmcluster",
			masterNodeQuota: &apiv1.NodeQuota{
				NodeCount: 0,
			},
			workerNodeQuota: nil,
			force:           false,
			setupMocks: func(dmClusterRepo *repomnocks.DMClusterRepo, sharedPoolRepo *repomnocks.SharedPoolRepo, dagExecutor *executormocks.RemoteExecutor, dataplaneClient *dataplanemgrmocks.Client) {
				setupDMClusterDAGMocks(dagExecutor)
				dmCluster := &entity.DMCluster{
					Id:            "testdmcluster",
					SharedPoolId:  "testsharedpool",
					DataplaneName: "test-dataplane",
					MasterNodeQuota: &apiv1.NodeQuota{
						NodeCount: 3,
					},
					WorkerNodeQuota: &apiv1.NodeQuota{
						NodeCount: 2,
					},
					Phase: apiv1.DMCluster_AVAILABLE,
				}
				dmClusterRepo.On("Load", mock.Anything, "testdmcluster", false).Return(dmCluster, nil)

				sharedPool := &entity.SharedPool{
					SharedPool: apiv1.SharedPool{
						Id:            "testsharedpool",
						Phase:         apiv1.SharedPool_AVAILABLE,
						DataplaneName: "test-dataplane",
					},
				}
				sharedPoolRepo.On("Load", mock.Anything, "testsharedpool", false).Return(sharedPool, nil)
				setupDMNodePoolMocks(dataplaneClient)
			},
			expectedError: "only when the current worker node count is 0, the master node count can be scaled to 0",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name:         "force scale on modifying cluster",
			sharedPoolId: "testsharedpool",
			dmClusterId:  "testdmcluster",
			masterNodeQuota: &apiv1.NodeQuota{
				NodeCount: 5,
				NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16},
			},
			workerNodeQuota: nil,
			force:           true,
			setupMocks: func(dmClusterRepo *repomnocks.DMClusterRepo, sharedPoolRepo *repomnocks.SharedPoolRepo, dagExecutor *executormocks.RemoteExecutor, dataplaneClient *dataplanemgrmocks.Client) {
				setupDMClusterDAGMocks(dagExecutor)
				setupDMNodePoolMocks(dataplaneClient)

				dmCluster := &entity.DMCluster{
					Id:            "testdmcluster",
					SharedPoolId:  "testsharedpool",
					DataplaneName: "test-dataplane",
					Phase:         apiv1.DMCluster_MODIFYING,
					MasterNodeQuota: &apiv1.NodeQuota{
						NodeCount: 3,
						NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16},
					},
					WorkerNodeQuota: &apiv1.NodeQuota{
						NodeCount: 2,
						NodeSpec:  &apiv1.NodeSpec{Cpu: 4, MemoryGi: 8},
					},
				}
				dmClusterRepo.On("Load", mock.Anything, "testdmcluster", false).Return(dmCluster, nil)

				sharedPool := &entity.SharedPool{
					SharedPool: apiv1.SharedPool{
						Id:            "testsharedpool",
						Phase:         apiv1.SharedPool_AVAILABLE,
						DataplaneName: "test-dataplane",
					},
				}
				sharedPoolRepo.On("Load", mock.Anything, "testsharedpool", false).Return(sharedPool, nil)

				dmClusterRepo.On("Save", mock.Anything, mock.Anything).Return(nil)
				dagExecutor.On("Submit", mock.Anything, dag.PlanApplyDMCluster, mock.Anything, mock.Anything).Return(nil)
			},
			validate: func(t *testing.T, result *entity.DMCluster, err error) {
				t.Helper()
				require.NoError(t, err)
				require.NotNil(t, result)
				assert.Equal(t, uint32(5), result.GetMasterNodeQuota().GetNodeCount())
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			dmClusterRepo := repomnocks.NewDMClusterRepo(t)
			dmClusterRepo.On("GetBackend", mock.Anything, mock.Anything, mock.Anything).Return(entity.DMBackendLegacy, nil).Maybe()
			sharedPoolRepo := repomnocks.NewSharedPoolRepo(t)
			dagExecutor := executormocks.NewRemoteExecutor(t)
			dataplaneClient := dataplanemgrmocks.NewClient(t)

			tt.setupMocks(dmClusterRepo, sharedPoolRepo, dagExecutor, dataplaneClient)

			server := &serverImpl{
				dmClusterRepo:   dmClusterRepo,
				sharedPoolRepo:  sharedPoolRepo,
				dagExecutor:     dagExecutor,
				dataplaneClient: dataplaneClient,
				transactioner:   &mockTransactioner{},
			}

			result, err := server.ScaleDMCluster(context.Background(), tt.sharedPoolId, tt.dmClusterId, tt.masterNodeQuota, tt.workerNodeQuota, tt.force)

			switch {
			case tt.expectedError != "":
				require.Error(t, err)
				assert.Contains(t, err.Error(), tt.expectedError)
				if tt.expectedCode != codes.OK {
					assert.Equal(t, tt.expectedCode, status.Code(err))
				}
				assert.Nil(t, result)

			case tt.validate != nil:
				tt.validate(t, result, err)

			default:
				require.NoError(t, err)
				require.NotNil(t, result)
			}
		})
	}
}

func TestDeleteDMCluster(t *testing.T) {
	tests := []struct {
		name          string
		sharedPoolId  string
		dmClusterId   string
		setupMocks    func(*repomnocks.DMClusterRepo, *executormocks.RemoteExecutor)
		expectedError string
		expectedCode  codes.Code
	}{
		{
			name:         "successful delete dm cluster",
			sharedPoolId: "testsharedpool",
			dmClusterId:  "testdmcluster",
			setupMocks: func(dmClusterRepo *repomnocks.DMClusterRepo, dagExecutor *executormocks.RemoteExecutor) {
				dmCluster := &entity.DMCluster{
					Id:            "testdmcluster",
					SharedPoolId:  "testsharedpool",
					DataplaneName: "test-dataplane",
					Phase:         apiv1.DMCluster_AVAILABLE,
					Backend:       entity.DMBackendLegacyProto,
				}
				dmClusterRepo.On("Load", mock.Anything, "testdmcluster", false).Return(dmCluster, nil)
				setupDMClusterDAGMocks(dagExecutor)
				dmClusterRepo.On("Delete", mock.Anything, "testdmcluster").Return(nil)
				dagExecutor.On("Submit", mock.Anything, dag.PlanApplyDMCluster, mock.Anything, mock.Anything).Return(nil)
			},
		},
		{
			name:          "empty shared pool id",
			sharedPoolId:  "",
			dmClusterId:   "testdmcluster",
			setupMocks:    func(dmClusterRepo *repomnocks.DMClusterRepo, dagExecutor *executormocks.RemoteExecutor) {},
			expectedError: "shared pool id is required",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name:          "empty dm cluster id",
			sharedPoolId:  "testsharedpool",
			dmClusterId:   "",
			setupMocks:    func(dmClusterRepo *repomnocks.DMClusterRepo, dagExecutor *executormocks.RemoteExecutor) {},
			expectedError: "dm cluster id is required",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name:         "dm cluster not found",
			sharedPoolId: "testsharedpool",
			dmClusterId:  "non-existing",
			setupMocks: func(dmClusterRepo *repomnocks.DMClusterRepo, dagExecutor *executormocks.RemoteExecutor) {
				dmClusterRepo.On("Load", mock.Anything, "non-existing", false).Return(nil, status.Error(codes.NotFound, "not found"))
			},
			expectedError: "dm cluster non-existing not found",
			expectedCode:  codes.NotFound,
		},
		{
			name:         "dm cluster does not belong to shared pool",
			sharedPoolId: "wrong-shared-pool",
			dmClusterId:  "testdmcluster",
			setupMocks: func(dmClusterRepo *repomnocks.DMClusterRepo, dagExecutor *executormocks.RemoteExecutor) {
				dmCluster := &entity.DMCluster{
					Id:           "testdmcluster",
					SharedPoolId: "other-shared-pool",
				}
				dmClusterRepo.On("Load", mock.Anything, "testdmcluster", false).Return(dmCluster, nil)
			},
			expectedError: "dm cluster testdmcluster does not belong to shared pool wrong-shared-pool",
			expectedCode:  codes.PermissionDenied,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			dmClusterRepo := repomnocks.NewDMClusterRepo(t)
			dagExecutor := executormocks.NewRemoteExecutor(t)

			tt.setupMocks(dmClusterRepo, dagExecutor)

			server := &serverImpl{
				dmClusterRepo:  dmClusterRepo,
				sharedPoolRepo: availableSharedPoolRepoForMutationTest(),
				dagExecutor:    dagExecutor,
				transactioner:  &mockTransactioner{},
			}

			err := server.DeleteDMCluster(context.Background(), tt.sharedPoolId, tt.dmClusterId)

			if tt.expectedError != "" {
				require.Error(t, err)
				assert.Contains(t, err.Error(), tt.expectedError)
				if tt.expectedCode != codes.OK {
					assert.Equal(t, tt.expectedCode, status.Code(err))
				}
			} else {
				require.NoError(t, err)
			}
		})
	}
}