package server

import (
	"context"
	"errors"
	"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"
	"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"
	"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"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
	repomnocks "github.com/tidbcloud/cluster-service-ng/pkg/storage/repo/mocks"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

func TestDeleteSharedPool(t *testing.T) {
	tests := []struct {
		name           string
		sharedPoolId   string
		pruneDataplane bool
		setupMocks     func(*repomnocks.SharedPoolRepo, *repomnocks.TiDBClusterRepo, *repomnocks.DMClusterRepo, *repomnocks.RuConfigRepo, *executormocks.RemoteExecutor)
		expectedError  string
		expectedCode   codes.Code
		validate       func(*testing.T, *entity.SharedPool, error)
	}{
		{
			name:           "successful deletion",
			sharedPoolId:   "testsharedpoolid",
			pruneDataplane: true,
			setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, tidbClusterRepo *repomnocks.TiDBClusterRepo, dmClusterRepo *repomnocks.DMClusterRepo, ruConfigRepo *repomnocks.RuConfigRepo, dagExecutor *executormocks.RemoteExecutor) {
				expectedSharedPool := &entity.SharedPool{
					SharedPool: apiv1.SharedPool{
						Id:            "testsharedpoolid",
						DataplaneName: "dp-1",
						Phase:         apiv1.SharedPool_AVAILABLE,
					},
				}
				expectedSharedPool.TenantId = "test-tenant-id"
				sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", false).Return(expectedSharedPool, nil)
				tidbClusterRepo.On("List", mock.Anything, &repo.TiDBClusterListFilter{
					SharedPoolIds: []string{"testsharedpoolid"},
					Deleted:       false,
				}).Return([]*entity.TiDBCluster{}, nil)
				dmClusterRepo.On("List", mock.Anything, &repo.DMClusterListFilter{
					SharedPoolIds: []string{"testsharedpoolid"},
					Deleted:       false,
				}).Return([]*entity.DMCluster{}, nil)

				// Mock no running DAGs
				dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
				sharedPoolRepo.On("Delete", mock.Anything, mock.Anything).Return(nil)
				ruConfigRepo.On("DeleteRuConfig", mock.Anything, "testsharedpoolid", apiv1.RuConfig_SHARED_POOL).Return(nil)
				dagExecutor.On("Submit", mock.Anything, dag.PlanDeleteSharedPool, mock.MatchedBy(func(args actions.ActionArgs) bool {
					meteringType, _ := args.GetString(workflow.MeteringTypeKey)
					deleteFlag, _ := args.GetString(workflow.MeteringDeleteSharedPoolKey)
					tenantId, _ := args.GetString(workflow.TenantIdKey)
					sharedPoolId, _ := args.GetString(workflow.SharedPoolIdKey)
					return meteringType == workflow.MeteringTypeValueSharedPoolMeta &&
						deleteFlag == workflow.MeteringDeleteSharedPoolValue &&
						tenantId == "test-tenant-id" &&
						sharedPoolId == "testsharedpoolid"
				}), mock.Anything).Return(nil)
			},
			validate: func(t *testing.T, result *entity.SharedPool, err error) {
				t.Helper()
				require.NoError(t, err)
				require.NotNil(t, result)
				assert.Equal(t, "testsharedpoolid", result.GetId())
				assert.Equal(t, apiv1.SharedPool_DELETING, result.GetPhase())
			},
		},
		{
			name:           "empty shared pool id",
			sharedPoolId:   "",
			pruneDataplane: false,
			setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, tidbClusterRepo *repomnocks.TiDBClusterRepo, dmClusterRepo *repomnocks.DMClusterRepo, ruConfigRepo *repomnocks.RuConfigRepo, dagExecutor *executormocks.RemoteExecutor) {
				// No mocks needed for this test case
			},
			expectedError: "sharedPool id is required",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name:           "shared pool has tidb clusters",
			sharedPoolId:   "testsharedpoolid",
			pruneDataplane: false,
			setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, tidbClusterRepo *repomnocks.TiDBClusterRepo, dmClusterRepo *repomnocks.DMClusterRepo, ruConfigRepo *repomnocks.RuConfigRepo, dagExecutor *executormocks.RemoteExecutor) {
				expectedSharedPool := &entity.SharedPool{
					SharedPool: apiv1.SharedPool{
						Id:            "testsharedpoolid",
						DataplaneName: "dp-1",
						Phase:         apiv1.SharedPool_AVAILABLE,
					},
				}
				sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", false).Return(expectedSharedPool, nil)

				// Mock existing TiDB clusters
				existingClusters := []*entity.TiDBCluster{
					{
						Id: "existingcluster",
					},
				}
				tidbClusterRepo.On("List", mock.Anything, &repo.TiDBClusterListFilter{
					SharedPoolIds: []string{"testsharedpoolid"},
					Deleted:       false,
				}).Return(existingClusters, nil)
			},
			expectedError: "sharedPoolId has available tidb clusters, please delete them first",
			expectedCode:  codes.FailedPrecondition,
		},
		{
			name:           "shared pool has dm clusters - successful deletion",
			sharedPoolId:   "testsharedpoolid",
			pruneDataplane: false,
			setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, tidbClusterRepo *repomnocks.TiDBClusterRepo, dmClusterRepo *repomnocks.DMClusterRepo, ruConfigRepo *repomnocks.RuConfigRepo, dagExecutor *executormocks.RemoteExecutor) {
				expectedSharedPool := &entity.SharedPool{
					SharedPool: apiv1.SharedPool{
						Id:            "testsharedpoolid",
						DataplaneName: "dp-1",
						Phase:         apiv1.SharedPool_AVAILABLE,
					},
				}
				sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", false).Return(expectedSharedPool, nil)

				// No TiDB clusters
				tidbClusterRepo.On("List", mock.Anything, &repo.TiDBClusterListFilter{
					SharedPoolIds: []string{"testsharedpoolid"},
					Deleted:       false,
				}).Return([]*entity.TiDBCluster{}, nil)

				// Mock existing DM clusters
				existingDMClusters := []*entity.DMCluster{
					{
						Id:           "dm-cluster-1",
						SharedPoolId: "testsharedpoolid",
					},
					{
						Id:           "dm-cluster-2",
						SharedPoolId: "testsharedpoolid",
					},
				}
				dmClusterRepo.On("List", mock.Anything, &repo.DMClusterListFilter{
					SharedPoolIds: []string{"testsharedpoolid"},
					Deleted:       false,
				}).Return(existingDMClusters, nil)

				// Mock dm cluster deletion calls
				dmClusterRepo.On("Load", mock.Anything, "dm-cluster-1", false).Return(existingDMClusters[0], nil)
				dmClusterRepo.On("Load", mock.Anything, "dm-cluster-2", false).Return(existingDMClusters[1], nil)

				// Mock DAG queries for DM cluster deletions
				dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)

				// Mock DM cluster deletions from DB
				dmClusterRepo.On("Delete", mock.Anything, "dm-cluster-1").Return(nil)
				dmClusterRepo.On("Delete", mock.Anything, "dm-cluster-2").Return(nil)

				// Mock DM cluster deletion DAG submissions
				dagExecutor.On("Submit", mock.Anything, dag.PlanApplyDMCluster, mock.Anything, mock.Anything).Return(nil)

				// Mock shared pool deletion
				sharedPoolRepo.On("Delete", mock.Anything, mock.Anything).Return(nil)
				ruConfigRepo.On("DeleteRuConfig", mock.Anything, "testsharedpoolid", apiv1.RuConfig_SHARED_POOL).Return(nil)
				dagExecutor.On("Submit", mock.Anything, dag.PlanDeleteSharedPool, mock.Anything, mock.Anything).Return(nil)
			},
			validate: func(t *testing.T, result *entity.SharedPool, err error) {
				t.Helper()
				require.NoError(t, err)
				require.NotNil(t, result)
				assert.Equal(t, "testsharedpoolid", result.GetId())
				assert.Equal(t, apiv1.SharedPool_DELETING, result.GetPhase())
			},
		},
		{
			name:           "transitional shared pool with dm clusters fails closed",
			sharedPoolId:   "testsharedpoolid",
			pruneDataplane: false,
			setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, tidbClusterRepo *repomnocks.TiDBClusterRepo, dmClusterRepo *repomnocks.DMClusterRepo, ruConfigRepo *repomnocks.RuConfigRepo, dagExecutor *executormocks.RemoteExecutor) {
				expectedSharedPool := &entity.SharedPool{
					SharedPool:       apiv1.SharedPool{Id: "testsharedpoolid", DataplaneName: "dp-1", Phase: apiv1.SharedPool_PAUSING},
					PauseResumeState: entity.PauseResumeStatePausing,
				}
				sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", false).Return(expectedSharedPool, nil)
				tidbClusterRepo.On("List", mock.Anything, mock.Anything).Return([]*entity.TiDBCluster{}, nil)
				dmClusterRepo.On("List", mock.Anything, mock.Anything).Return([]*entity.DMCluster{{Id: "dm-cluster-1", SharedPoolId: "testsharedpoolid"}}, nil)
			},
			expectedError: "sharedPool has DM clusters",
			expectedCode:  codes.FailedPrecondition,
		},
		{
			name:           "shared pool has deleting tidb groups",
			sharedPoolId:   "testsharedpoolid",
			pruneDataplane: false,
			setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, tidbClusterRepo *repomnocks.TiDBClusterRepo, dmClusterRepo *repomnocks.DMClusterRepo, ruConfigRepo *repomnocks.RuConfigRepo, dagExecutor *executormocks.RemoteExecutor) {
				expectedSharedPool := &entity.SharedPool{
					SharedPool: apiv1.SharedPool{
						Id:            "testsharedpoolid",
						DataplaneName: "dp-1",
						Phase:         apiv1.SharedPool_AVAILABLE,
					},
				}
				sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", false).Return(expectedSharedPool, nil)
				tidbClusterRepo.On("List", mock.Anything, &repo.TiDBClusterListFilter{
					SharedPoolIds: []string{"testsharedpoolid"},
					Deleted:       false,
				}).Return([]*entity.TiDBCluster{}, nil)
				dmClusterRepo.On("List", mock.Anything, &repo.DMClusterListFilter{
					SharedPoolIds: []string{"testsharedpoolid"},
					Deleted:       false,
				}).Return([]*entity.DMCluster{}, nil)

				deletingGroup := executor.NewExecTaskIdStatus()
				deletingGroup["delete-tidbgroup-abc"] = &executor.ExecTaskStatusEx{
					ExecTaskStatus: executor.ExecTaskStatus{
						ExecTask: &executor.ExecutorTask{
							ExecId:   "delete-tidbgroup-abc",
							RootName: dag.PlanDeleteTiDBGroup,
						},
					},
					State: executor.ExecTaskStateTypeRunningOK,
				}
				dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(deletingGroup, nil).Once()
			},
			expectedError: "sharedPool has deleting tidb groups, please wait for them to be deleted first",
			expectedCode:  codes.FailedPrecondition,
		},
		{
			name:           "shared pool already deleting",
			sharedPoolId:   "testsharedpoolid",
			pruneDataplane: false,
			setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, tidbClusterRepo *repomnocks.TiDBClusterRepo, dmClusterRepo *repomnocks.DMClusterRepo, ruConfigRepo *repomnocks.RuConfigRepo, dagExecutor *executormocks.RemoteExecutor) {
				expectedSharedPool := &entity.SharedPool{
					SharedPool: apiv1.SharedPool{
						Id:            "testsharedpoolid",
						DataplaneName: "dp-1",
						Phase:         apiv1.SharedPool_DELETING,
					},
				}
				sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", false).Return(expectedSharedPool, nil)
				// When phase is DELETING, the code returns early without calling tidbClusterRepo.List
			},
			validate: func(t *testing.T, result *entity.SharedPool, err error) {
				t.Helper()
				require.NoError(t, err)
				require.NotNil(t, result)
				assert.Equal(t, "testsharedpoolid", result.GetId())
				assert.Equal(t, apiv1.SharedPool_DELETING, result.GetPhase())
			},
		},
		{
			name:           "delete dag already running ensures shared pool meta is deleted",
			sharedPoolId:   "testsharedpoolid",
			pruneDataplane: true,
			setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, tidbClusterRepo *repomnocks.TiDBClusterRepo, dmClusterRepo *repomnocks.DMClusterRepo, ruConfigRepo *repomnocks.RuConfigRepo, dagExecutor *executormocks.RemoteExecutor) {
				expectedSharedPool := &entity.SharedPool{
					SharedPool: apiv1.SharedPool{
						Id:            "testsharedpoolid",
						DataplaneName: "dp-1",
						Phase:         apiv1.SharedPool_AVAILABLE,
					},
				}
				sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", false).Return(expectedSharedPool, nil)
				tidbClusterRepo.On("List", mock.Anything, &repo.TiDBClusterListFilter{
					SharedPoolIds: []string{"testsharedpoolid"},
					Deleted:       false,
				}).Return([]*entity.TiDBCluster{}, nil)
				dmClusterRepo.On("List", mock.Anything, &repo.DMClusterListFilter{
					SharedPoolIds: []string{"testsharedpoolid"},
					Deleted:       false,
				}).Return([]*entity.DMCluster{}, nil)

				deleteDags := executor.NewExecTaskIdStatus()
				deleteDags["delete-sharedpool-testsharedpoolid-abc"] = &executor.ExecTaskStatusEx{
					ExecTaskStatus: executor.ExecTaskStatus{
						ExecTask: &executor.ExecutorTask{
							ExecId:   "delete-sharedpool-testsharedpoolid-abc",
							RootName: dag.PlanDeleteSharedPool,
						},
					},
					State: executor.ExecTaskStateTypeRunningOK,
				}
				dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanDeleteSharedPool, mock.Anything).Return(deleteDags, nil).Once()
				dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
				sharedPoolRepo.On("Delete", mock.Anything, "testsharedpoolid").Return(nil).Once()
				ruConfigRepo.On("DeleteRuConfig", mock.Anything, "testsharedpoolid", apiv1.RuConfig_SHARED_POOL).Return(nil).Once()
			},
			validate: func(t *testing.T, result *entity.SharedPool, err error) {
				t.Helper()
				require.NoError(t, err)
				require.NotNil(t, result)
				assert.Equal(t, "testsharedpoolid", result.GetId())
				assert.Equal(t, apiv1.SharedPool_DELETING, result.GetPhase())
			},
		},
		{
			name:           "shared pool not found in repo",
			sharedPoolId:   "creatingsharedpoolid",
			pruneDataplane: false,
			setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, tidbClusterRepo *repomnocks.TiDBClusterRepo, dmClusterRepo *repomnocks.DMClusterRepo, ruConfigRepo *repomnocks.RuConfigRepo, dagExecutor *executormocks.RemoteExecutor) {
				// Mock repository not found
				sharedPoolRepo.On("Load", mock.Anything, "creatingsharedpoolid", false).Return(nil, status.Error(codes.NotFound, "not found"))
			},
			expectedError: "sharedPool creatingsharedpoolid not found",
			expectedCode:  codes.NotFound,
		},
		{
			name:           "repository load error",
			sharedPoolId:   "testsharedpoolid",
			pruneDataplane: false,
			setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, tidbClusterRepo *repomnocks.TiDBClusterRepo, dmClusterRepo *repomnocks.DMClusterRepo, ruConfigRepo *repomnocks.RuConfigRepo, dagExecutor *executormocks.RemoteExecutor) {
				sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", false).Return(nil, errors.New("database error"))
			},
			expectedError: "failed to load sharedPool: database error",
			expectedCode:  codes.Internal,
		},
		{
			name:           "tidb cluster list error",
			sharedPoolId:   "testsharedpoolid",
			pruneDataplane: false,
			setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, tidbClusterRepo *repomnocks.TiDBClusterRepo, dmClusterRepo *repomnocks.DMClusterRepo, ruConfigRepo *repomnocks.RuConfigRepo, dagExecutor *executormocks.RemoteExecutor) {
				expectedSharedPool := &entity.SharedPool{
					SharedPool: apiv1.SharedPool{
						Id:            "testsharedpoolid",
						DataplaneName: "dp-1",
						Phase:         apiv1.SharedPool_AVAILABLE,
					},
				}
				sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", false).Return(expectedSharedPool, nil)
				tidbClusterRepo.On("List", mock.Anything, &repo.TiDBClusterListFilter{
					SharedPoolIds: []string{"testsharedpoolid"},
					Deleted:       false,
				}).Return(nil, errors.New("list error"))
			},
			expectedError: "failed to list tidb clusters: list error",
			expectedCode:  codes.Internal,
		},
		{
			name:           "dag executor abort error",
			sharedPoolId:   "testsharedpoolid",
			pruneDataplane: false,
			setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, tidbClusterRepo *repomnocks.TiDBClusterRepo, dmClusterRepo *repomnocks.DMClusterRepo, ruConfigRepo *repomnocks.RuConfigRepo, dagExecutor *executormocks.RemoteExecutor) {
				expectedSharedPool := &entity.SharedPool{
					SharedPool: apiv1.SharedPool{
						Id:            "testsharedpoolid",
						DataplaneName: "dp-1",
						Phase:         apiv1.SharedPool_AVAILABLE,
					},
				}
				sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", false).Return(expectedSharedPool, nil)
				tidbClusterRepo.On("List", mock.Anything, &repo.TiDBClusterListFilter{
					SharedPoolIds: []string{"testsharedpoolid"},
					Deleted:       false,
				}).Return([]*entity.TiDBCluster{}, nil)
				dmClusterRepo.On("List", mock.Anything, &repo.DMClusterListFilter{
					SharedPoolIds: []string{"testsharedpoolid"},
					Deleted:       false,
				}).Return([]*entity.DMCluster{}, nil)

				emptyRunning := executor.NewExecTaskIdStatus()
				for _, plan := range []string{
					dag.PlanDeleteTiDBGroup,
					dag.PlanDeleteTiDBCluster,
					dag.PlanEnableCoprocessorWorker,
					dag.PlanDisableCoprocessorWorker,
					dag.PlanCreateSharedPool,
					dag.PlanUpdateComponent,
					dag.PlanCreateTiDBCluster,
					dag.PlanUpdateTiDBCluster,
					dag.PlanCreateTiDBGroup,
					dag.PlanUpdateTiDBGroup,
				} {
					dagExecutor.On("QueryRunnings", mock.Anything, plan, mock.Anything).Return(emptyRunning, nil).Once()
				}

				// The regular shared-pool abort set is only reached after both
				// remote-worker plans are explicitly confirmed idle.
				runningDags := executor.NewExecTaskIdStatus()
				runningDags["test-dag-id"] = &executor.ExecTaskStatusEx{
					ExecTaskStatus: executor.ExecTaskStatus{
						ExecTask: &executor.ExecutorTask{
							ExecId:   "test-exec-id",
							RootName: dag.PlanApplyComponent,
						},
					},
					State: executor.ExecTaskStateTypeRunningOK,
				}
				dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanApplyComponent, mock.Anything).Return(runningDags, nil).Once()
				dagExecutor.On("Abort", mock.Anything, "test-exec-id").Return(errors.New("abort error"))
			},
			expectedError: "failed to abort sharedpool dag",
			expectedCode:  codes.Internal,
		},
		{
			name:           "dag executor submit error",
			sharedPoolId:   "testsharedpoolid",
			pruneDataplane: false,
			setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, tidbClusterRepo *repomnocks.TiDBClusterRepo, dmClusterRepo *repomnocks.DMClusterRepo, ruConfigRepo *repomnocks.RuConfigRepo, dagExecutor *executormocks.RemoteExecutor) {
				expectedSharedPool := &entity.SharedPool{
					SharedPool: apiv1.SharedPool{
						Id:            "testsharedpoolid",
						DataplaneName: "dp-1",
						Phase:         apiv1.SharedPool_AVAILABLE,
					},
				}
				sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", false).Return(expectedSharedPool, nil)
				tidbClusterRepo.On("List", mock.Anything, &repo.TiDBClusterListFilter{
					SharedPoolIds: []string{"testsharedpoolid"},
					Deleted:       false,
				}).Return([]*entity.TiDBCluster{}, nil)
				dmClusterRepo.On("List", mock.Anything, &repo.DMClusterListFilter{
					SharedPoolIds: []string{"testsharedpoolid"},
					Deleted:       false,
				}).Return([]*entity.DMCluster{}, nil)

				// Mock no running DAGs that need abort
				dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
				// Mock no running delete DAGs
				dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
				sharedPoolRepo.On("Delete", mock.Anything, mock.Anything).Return(nil)
				ruConfigRepo.On("DeleteRuConfig", mock.Anything, "testsharedpoolid", apiv1.RuConfig_SHARED_POOL).Return(nil)
				dagExecutor.On("Submit", mock.Anything, dag.PlanDeleteSharedPool, mock.Anything, mock.Anything).Return(errors.New("submit error"))
			},
			expectedError: "failed to submit delete cluster dag: submit error",
			expectedCode:  codes.Internal,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			// Create mocks
			sharedPoolRepo := repomnocks.NewSharedPoolRepo(t)
			tidbClusterRepo := repomnocks.NewTiDBClusterRepo(t)
			dmClusterRepo := repomnocks.NewDMClusterRepo(t)
			dmClusterRepo.On("GetBackend", mock.Anything, mock.Anything, mock.Anything).Return(entity.DMBackendLegacy, nil).Maybe()
			ruConfigRepo := repomnocks.NewRuConfigRepo(t)
			nodeGroupStateRepo := repomnocks.NewDataplaneNodeInstanceGroupStateRepo(t)
			lockRepo := repomnocks.NewLockRepo(t)
			lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil).Maybe()
			dagExecutor := executormocks.NewRemoteExecutor(t)

			// Setup mocks
			tt.setupMocks(sharedPoolRepo, tidbClusterRepo, dmClusterRepo, ruConfigRepo, dagExecutor)
			nodeGroupStateRepo.On("ListByDataplane", mock.Anything, mock.Anything).Return([]*entity.DataplaneNodeInstanceGroupState{}, nil).Maybe()

			// Create server implementation
			server := &serverImpl{
				sharedPoolRepo:     sharedPoolRepo,
				tidbClusterRepo:    tidbClusterRepo,
				dmClusterRepo:      dmClusterRepo,
				ruConfigRepo:       ruConfigRepo,
				nodeGroupStateRepo: nodeGroupStateRepo,
				lockRepo:           lockRepo,
				dagExecutor:        dagExecutor,
				transactioner:      &mockTransactioner{},
			}

			// Execute test
			result, err := server.DeleteSharedPool(context.Background(), tt.sharedPoolId, tt.pruneDataplane)

			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 TestDeleteSharedPoolTakesOverRunningPause(t *testing.T) {
	sharedPoolRepo := repomnocks.NewSharedPoolRepo(t)
	tidbClusterRepo := repomnocks.NewTiDBClusterRepo(t)
	dmClusterRepo := repomnocks.NewDMClusterRepo(t)
	ruConfigRepo := repomnocks.NewRuConfigRepo(t)
	operationRepo := repomnocks.NewPauseResumeOperationRepo(t)
	lockRepo := repomnocks.NewLockRepo(t)
	dagExecutor := executormocks.NewRemoteExecutor(t)
	pool := &entity.SharedPool{
		SharedPool:       apiv1.SharedPool{Id: "pool-1", DataplaneName: "dp-1", Phase: apiv1.SharedPool_PAUSING},
		PauseResumeState: entity.PauseResumeStatePausing,
	}
	sharedPoolRepo.EXPECT().Load(mock.Anything, "pool-1", false).Return(pool, nil).Once()
	tidbClusterRepo.EXPECT().List(mock.Anything, mock.Anything).Return([]*entity.TiDBCluster{}, nil).Once()
	dmClusterRepo.EXPECT().List(mock.Anything, mock.Anything).Return([]*entity.DMCluster{}, nil).Once()
	lockRepo.EXPECT().Lock(mock.Anything, mock.Anything).Return(func() error { return nil }, nil).Once()
	dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
	operation := &entity.PauseResumeOperation{
		SharedPoolId: "pool-1", OperationId: "pause-1", Type: apiv1.PauseResumeOperation_PAUSE, Status: apiv1.PauseResumeOperation_RUNNING,
	}
	operationRepo.EXPECT().LoadLatestBySharedPoolID(mock.Anything, "pool-1").Return(operation, nil).Once()
	cancelled := false
	operationRepo.On("CancelIfRunning", mock.Anything, "pool-1", "pause-1", "cancelled by shared pool deletion", mock.Anything).
		Run(func(mock.Arguments) { cancelled = true }).Return(true, nil).Once()
	dagExecutor.EXPECT().Abort(mock.Anything, workflow.PauseResumeExecID("pool-1", "pause-1", apiv1.PauseResumeOperation_PAUSE.String())).Return(nil).Once()
	sharedPoolRepo.EXPECT().Delete(mock.Anything, "pool-1").Return(nil).Once()
	ruConfigRepo.EXPECT().DeleteRuConfig(mock.Anything, "pool-1", apiv1.RuConfig_SHARED_POOL).Return(nil).Once()
	dagExecutor.On("Submit", mock.Anything, dag.PlanDeleteSharedPool, mock.Anything, mock.Anything).
		Run(func(mock.Arguments) { require.True(t, cancelled) }).Return(nil).Once()

	server := &serverImpl{
		sharedPoolRepo: sharedPoolRepo, pauseResumeOperationRepo: operationRepo, tidbClusterRepo: tidbClusterRepo,
		dmClusterRepo: dmClusterRepo, ruConfigRepo: ruConfigRepo, lockRepo: lockRepo, dagExecutor: dagExecutor, transactioner: &mockTransactioner{},
	}
	result, err := server.DeleteSharedPool(context.Background(), "pool-1", false)
	require.NoError(t, err)
	require.Equal(t, apiv1.SharedPool_DELETING, result.GetPhase())
}

func TestDeleteSharedPoolBlocksCrossPoolRestoreUsingPoolAsSource(t *testing.T) {
	sharedPoolRepo := repomnocks.NewSharedPoolRepo(t)
	tidbClusterRepo := repomnocks.NewTiDBClusterRepo(t)
	dmClusterRepo := repomnocks.NewDMClusterRepo(t)
	lockRepo := repomnocks.NewLockRepo(t)
	dagExecutor := executormocks.NewRemoteExecutor(t)
	pool := &entity.SharedPool{
		SharedPool:       apiv1.SharedPool{Id: "pool-1", DataplaneName: "dp-1", Phase: apiv1.SharedPool_PAUSING},
		PauseResumeState: entity.PauseResumeStatePausing,
	}
	sharedPoolRepo.EXPECT().Load(mock.Anything, "pool-1", false).Return(pool, nil).Once()
	tidbClusterRepo.EXPECT().List(mock.Anything, mock.Anything).Return([]*entity.TiDBCluster{}, nil).Once()
	dmClusterRepo.EXPECT().List(mock.Anything, mock.Anything).Return([]*entity.DMCluster{}, nil).Once()
	lockRepo.EXPECT().Lock(mock.Anything, mock.Anything).Return(func() error { return nil }, nil).Once()
	running := executor.NewExecTaskIdStatus()
	running["cross-restore"] = nil
	dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanCreateManualBackupXRestore, mock.MatchedBy(matchesSourcePoolFilter("pool-1"))).Return(running, nil).Once()
	dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)

	server := &serverImpl{
		sharedPoolRepo: sharedPoolRepo, tidbClusterRepo: tidbClusterRepo, dmClusterRepo: dmClusterRepo,
		lockRepo: lockRepo, dagExecutor: dagExecutor, transactioner: &mockTransactioner{},
	}
	_, err := server.DeleteSharedPool(context.Background(), "pool-1", false)
	require.Equal(t, codes.FailedPrecondition, status.Code(err))
	require.ErrorContains(t, err, "backup or restore operation is running")
	dagExecutor.AssertNotCalled(t, "Abort", mock.Anything, mock.Anything)
	dagExecutor.AssertNotCalled(t, "Submit", mock.Anything, mock.Anything, mock.Anything, mock.Anything)
}

func matchesSourcePoolFilter(sharedPoolID string) func(executor.MatchFilter) bool {
	return func(filter executor.MatchFilter) bool {
		values, ok := filter[workflow.SourceSharedPoolIdKey]
		return ok && len(filter) == 1 && len(values) == 1 && values[0] == sharedPoolID
	}
}

func TestDeleteSharedPool_BlocksWhileTiKVRemoteWorkerSwitchRuns(t *testing.T) {
	for _, runningPlan := range tikvRemoteWorkersDAGPlans() {
		t.Run(runningPlan, func(t *testing.T) {
			sharedPoolRepo := repomnocks.NewSharedPoolRepo(t)
			tidbClusterRepo := repomnocks.NewTiDBClusterRepo(t)
			dmClusterRepo := repomnocks.NewDMClusterRepo(t)
			ruConfigRepo := repomnocks.NewRuConfigRepo(t)
			lockRepo := repomnocks.NewLockRepo(t)
			dagExecutor := executormocks.NewRemoteExecutor(t)
			sharedPoolRepo.On("Load", mock.Anything, testTKRWSharedPoolID, false).Return(newTKRWSharedPool(tkrwEnabledBoth(t), 3), nil)
			tidbClusterRepo.On("List", mock.Anything, mock.Anything).Return([]*entity.TiDBCluster{}, nil)
			dmClusterRepo.On("List", mock.Anything, mock.Anything).Return([]*entity.DMCluster{}, nil)
			lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil)

			running := executor.NewExecTaskIdStatus()
			running["remote-workers"] = &executor.ExecTaskStatusEx{ExecTaskStatus: executor.ExecTaskStatus{ExecTask: &executor.ExecutorTask{ExecId: "remote-workers", RootName: runningPlan}}, State: executor.ExecTaskStateTypeRunningOK}
			dagExecutor.On("QueryRunnings", mock.Anything, runningPlan, mock.Anything).Return(running, nil)
			dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)

			server := &serverImpl{sharedPoolRepo: sharedPoolRepo, tidbClusterRepo: tidbClusterRepo, dmClusterRepo: dmClusterRepo, ruConfigRepo: ruConfigRepo, lockRepo: lockRepo, dagExecutor: dagExecutor, transactioner: &mockTransactioner{}}
			_, err := server.DeleteSharedPool(context.Background(), testTKRWSharedPoolID, false)
			require.Error(t, err)
			assert.Equal(t, codes.FailedPrecondition, status.Code(err))
			assert.Contains(t, err.Error(), "tikv remote worker routing switch is running")
			dagExecutor.AssertNotCalled(t, "Abort", mock.Anything, mock.Anything)
			dagExecutor.AssertNotCalled(t, "Submit", mock.Anything, dag.PlanDeleteSharedPool, mock.Anything, mock.Anything)
			sharedPoolRepo.AssertNotCalled(t, "Delete", mock.Anything, mock.Anything)
		})
	}
}