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/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	bractionmocks "github.com/tidbcloud/cluster-service-ng/domain/workflow/action/br/mocks"
	"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"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/schema"
	repomnocks "github.com/tidbcloud/cluster-service-ng/pkg/storage/repo/mocks"
	"github.com/tidbcloud/cluster-service-ng/pkg/utils"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/proto"
)

func TestUpdateTiDBCluster_BasicValidation(t *testing.T) {
	testCases := []struct {
		name          string
		tidbCluster   *entity.TiDBCluster
		force         bool
		expectedError string
		expectedCode  codes.Code
	}{
		{
			name:          "nil tidbCluster input",
			tidbCluster:   nil,
			force:         false,
			expectedError: "tidbCluster or tidbCluster id is required",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name: "empty cluster ID",
			tidbCluster: &entity.TiDBCluster{
				Id: "",
			},
			force:         false,
			expectedError: "tidbCluster or tidbCluster id is required",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name: "nil request unit",
			tidbCluster: &entity.TiDBCluster{
				Id:          "testcluster",
				RequestUnit: nil,
			},
			force:         false,
			expectedError: "tidbCluster request unit is required",
			expectedCode:  codes.InvalidArgument,
		},
	}

	for _, tc := range testCases {
		t.Run(tc.name, func(t *testing.T) {
			// Create server instance with minimal dependencies for validation tests
			server := &serverImpl{}

			// Execute test
			result, err := server.UpdateTiDBCluster(context.Background(), tc.tidbCluster, tc.force)

			// Validate results
			require.Error(t, err)
			assert.Contains(t, err.Error(), tc.expectedError)
			statusErr, ok := status.FromError(err)
			require.True(t, ok, "error should be a gRPC status error")
			assert.Equal(t, tc.expectedCode, statusErr.Code())
			assert.Nil(t, result)
		})
	}
}

// setupUpdateTiDBClusterMocks sets up common mocks for UpdateTiDBCluster tests.
// currRuVersion is the existing cluster's ru_version in DB.
// enableScaleTidbGroupByRu controls the feature flag.
// Returns the DAG executor mock so callers can verify Submit args.
func setupUpdateTiDBClusterMocks(
	t *testing.T,
	currRuVersion int32,
	enableScaleTidbGroupByRu bool,
) (*serverImpl, *executormocks.RemoteExecutor) {
	t.Helper()

	tidbClusterRepo := repomnocks.NewTiDBClusterRepo(t)
	sharedPoolRepo := repomnocks.NewSharedPoolRepo(t)
	tidbGroupRepo := repomnocks.NewTiDBGroupRepo(t)
	lockRepo := repomnocks.NewLockRepo(t)
	ruConfigRepo := repomnocks.NewRuConfigRepo(t)
	dagExecutor := executormocks.NewRemoteExecutor(t)
	dataplaneClient := dataplanemgrmocks.NewClient(t)
	scaleManager := bractionmocks.NewScaleManager(t)

	// Current cluster in DB
	currCluster := &entity.TiDBCluster{
		Id:           "test-cluster-id",
		SharedPoolId: "test-sharedpool-id",
		TenantId:     "test-tenant-id",
		Phase:        apiv1.Cluster_AVAILABLE,
		RuVersion:    currRuVersion,
		RequestUnit: &apiv1.Cluster_RequestUnit{
			Min: 100,
			Max: 1000,
		},
	}
	tidbClusterRepo.On("Load", mock.Anything, "test-cluster-id", false).Return(currCluster, nil)

	// Lock
	lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil)

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

	// DAG queries (mergeSharedPoolStatus + getRunningDagIns)
	setupAllDAGMocks(dagExecutor)
	// Also mock the three specific plans for getRunningDagIns in UpdateTiDBCluster
	dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).
		Return(executor.NewExecTaskIdStatus(), nil).Maybe()

	// ScaleManager
	scaleManager.On("AnyFullfillingDecision", mock.Anything, mock.Anything).Return(false, nil).Maybe()

	// RuConfig — when enableScaleTidbGroupByRu is true, we need Components with
	// NodeSpecRuRange so that genDefaultTiProxyComponent can compute node specs.
	ruConfig := &entity.RuConfig{
		RuConfig: apiv1.RuConfig{
			FeatureFlags: &apiv1.RuConfig_FeatureFlags{
				EnableScaleTidbGroupByRu: &enableScaleTidbGroupByRu,
			},
		},
	}
	if enableScaleTidbGroupByRu {
		ruConfig.Components = map[string]*apiv1.RuConfig_ComponentsMap{
			"tiproxy": {
				NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{
					"8c16g": {
						Min: 0,
						Max: 100000000,
					},
				},
			},
			"tidb": {
				NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{
					"8c16g": {
						Min: 0,
						Max: 100000000,
					},
				},
			},
		}
		// TiDBGroup list — return empty so generatedTiDBGroup loop is skipped
		tidbGroupRepo.On("List", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, false).Return([]*entity.TiDBGroup{}, nil)
	}
	ruConfigRepo.On("GetRuConfig", mock.Anything, "test-sharedpool-id", apiv1.RuConfig_SHARED_POOL).Return(ruConfig, nil)

	// Premium preserves the legacy NodePool lookup even when both RU
	// spec-scaling flags are disabled.
	nodePoolResp := &dplanemgrv1.ListNodePoolsResponse{
		NodePools: []*dplanemgrv1.NodePool{
			{
				CpuCore:               8,
				MemoryGi:              16,
				NodeType:              dplanemgrv1.NodeType_Dedicated,
				AvailableCpuMilliCore: 7000,
				AvailableMemoryMi:     15000,
			},
		},
	}
	dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(nodePoolResp, nil)

	// Save
	tidbClusterRepo.On("Save", mock.Anything, mock.Anything).Return(nil)

	// Submit
	dagExecutor.On("Submit", mock.Anything, dag.PlanUpdateTiDBCluster, mock.Anything, mock.Anything).Return(nil)

	server := &serverImpl{
		tidbClusterRepo: tidbClusterRepo,
		sharedPoolRepo:  sharedPoolRepo,
		tidbGroupRepo:   tidbGroupRepo,
		lockRepo:        lockRepo,
		ruConfigRepo:    ruConfigRepo,
		dagExecutor:     dagExecutor,
		transactioner:   &mockTransactioner{},
		dataplaneClient: dataplaneClient,
		scaleManager:    scaleManager,
	}

	return server, dagExecutor
}

func TestUpdateTiDBCluster_RuVersionChange_SendsMeteringMetadata(t *testing.T) {
	tests := []struct {
		name                     string
		currRuVersion            int32
		newRuVersion             int32
		enableScaleTidbGroupByRu bool
		expectMeteringType       bool
	}{
		{
			name:                     "ru_version changed with feature flag off - should send metering",
			currRuVersion:            1,
			newRuVersion:             2,
			enableScaleTidbGroupByRu: false,
			expectMeteringType:       true,
		},
		{
			name:                     "ru_version changed with feature flag on - should send metering",
			currRuVersion:            1,
			newRuVersion:             2,
			enableScaleTidbGroupByRu: true,
			expectMeteringType:       true,
		},
		{
			name:                     "ru_version unchanged with feature flag off - should send metering",
			currRuVersion:            1,
			newRuVersion:             1,
			enableScaleTidbGroupByRu: false,
			expectMeteringType:       true,
		},
		{
			name:                     "ru_version not set (0) with feature flag off - should send metering",
			currRuVersion:            1,
			newRuVersion:             0,
			enableScaleTidbGroupByRu: false,
			expectMeteringType:       true,
		},
		{
			name:                     "ru_version from 0 to 2 with feature flag off - should send metering",
			currRuVersion:            0,
			newRuVersion:             2,
			enableScaleTidbGroupByRu: false,
			expectMeteringType:       true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			server, dagExecutor := setupUpdateTiDBClusterMocks(t, tt.currRuVersion, tt.enableScaleTidbGroupByRu)

			input := &entity.TiDBCluster{
				Id:        "test-cluster-id",
				RuVersion: tt.newRuVersion,
				RequestUnit: &apiv1.Cluster_RequestUnit{
					Min: 100,
					Max: 1000,
				},
			}

			result, err := server.UpdateTiDBCluster(context.Background(), input, false)
			require.NoError(t, err)
			require.NotNil(t, result)

			// Verify the DAG Submit was called and check metering args
			submitCalls := dagExecutor.Calls
			var submitArgs actions.ActionArgs
			for _, call := range submitCalls {
				if call.Method == "Submit" {
					submitArgs = call.Arguments.Get(2).(actions.ActionArgs)
					break
				}
			}
			require.NotNil(t, submitArgs, "Submit should have been called")
			assert.True(t, workflow.IsAllComponent(submitArgs),
				"Premium must preserve the legacy empty-target-means-all behavior")
			_, skipErr := submitArgs.GetString(workflow.SkipComponentHPAApplyKey)
			require.Error(t, skipErr, "Premium DAG args must not opt in to the Essential v2 HPA skip")

			meteringType, err := submitArgs.GetString(workflow.MeteringTypeKey)
			if tt.expectMeteringType {
				require.NoError(t, err)
				assert.Equal(t, workflow.MeteringTypeValueLogicMetaByTiDBCluster, meteringType,
					"metering_type should be set when ru_version changes")
			} else {
				// Either key not present (GetString returns error) or value is empty
				assert.True(t, err != nil || meteringType == "",
					"metering_type should NOT be set when ru_version doesn't change")
			}
		})
	}
}

// TestUpdateTiDBCluster_OnlyWhitelistComponentsSubmitted verifies that when RU
// increases and shared pool regeneration runs, only whitelist SharedPool
// components (from RcuScaleComponents) are included in the DAG componentIds.
// Non-whitelist components like PD that may also be in MODIFYING state must
// NOT appear in the submitted action.
func TestUpdateTiDBCluster_OnlyWhitelistComponentsSubmitted(t *testing.T) {
	tidbClusterRepo := repomnocks.NewTiDBClusterRepo(t)
	sharedPoolRepo := repomnocks.NewSharedPoolRepo(t)
	tidbGroupRepo := repomnocks.NewTiDBGroupRepo(t)
	lockRepo := repomnocks.NewLockRepo(t)
	ruConfigRepo := repomnocks.NewRuConfigRepo(t)
	dagExecutor := executormocks.NewRemoteExecutor(t)
	dataplaneClient := dataplanemgrmocks.NewClient(t)
	scaleManager := bractionmocks.NewScaleManager(t)

	currCluster := &entity.TiDBCluster{
		Id:           "test-cluster-id",
		SharedPoolId: "testsharedpoolid",
		TenantId:     "test-tenant-id",
		Phase:        apiv1.Cluster_AVAILABLE,
		Version:      "v7.1.0",
		RequestUnit: &apiv1.Cluster_RequestUnit{
			Min: 100,
			Max: 1000,
		},
	}
	tidbClusterRepo.On("Load", mock.Anything, "test-cluster-id", false).Return(currCluster, nil)
	lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil)

	// Build a shared pool with whitelist and non-whitelist components in MODIFYING.
	// Component IDs for shared-pool-level components must be fnvHash(sharedPoolId).
	defaultCompId := utils.FnvHash("testsharedpoolid")
	sharedPool := &entity.SharedPool{
		SharedPool: apiv1.SharedPool{
			Id:            "testsharedpoolid",
			DataplaneName: "test-dataplane",
			Phase:         apiv1.SharedPool_AVAILABLE,
			Version:       "v7.1.0",
			ComponentGroups: []*apiv1.Component{
				// Whitelist: TIKV (SharedPool scope)
				// Starts AVAILABLE; generatedSharedPoolComponent will set MODIFYING
				// if spec changes (8c16g→16c64g due to RU tier boundary).
				{
					Id: defaultCompId, ComponentType: apiv1.ComponentType_TIKV,
					Phase: apiv1.Component_AVAILABLE,
					NodeQuota: &apiv1.NodeQuota{
						NodeSpec: &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16}, NodeCount: 3, MinNodeCount: 3, MaxNodeCount: 15,
					},
					Storages: []*apiv1.Component_Storage{
						{Name: apiv1.Component_Storage_DATA, Type: "gp3", SizeGi: 500, Iops: 3000},
						{Name: apiv1.Component_Storage_RAFTLOG, Type: "gp3", SizeGi: 50, Iops: 3000},
					},
				},
				// Whitelist: TIDB_WORKER (SharedPool scope)
				{
					Id: defaultCompId, ComponentType: apiv1.ComponentType_TIDB_WORKER,
					Phase: apiv1.Component_AVAILABLE,
					NodeQuota: &apiv1.NodeQuota{
						NodeSpec: &apiv1.NodeSpec{Cpu: 4, MemoryGi: 8}, NodeCount: 1, MinNodeCount: 1, MaxNodeCount: 15,
					},
				},
				// Non-whitelist: PD — will be regenerated and set MODIFYING by
				// generatedSharedPoolComponent, but whitelist filter must exclude it.
				{
					Id: defaultCompId, ComponentType: apiv1.ComponentType_PD,
					Phase: apiv1.Component_AVAILABLE,
					NodeQuota: &apiv1.NodeQuota{
						NodeSpec: &apiv1.NodeSpec{Cpu: 4, MemoryGi: 8}, NodeCount: 3, MinNodeCount: 3, MaxNodeCount: 3,
					},
				},
				// Non-whitelist: PD_TSO
				{
					Id: defaultCompId, ComponentType: apiv1.ComponentType_PD_TSO,
					Phase: apiv1.Component_AVAILABLE,
					NodeQuota: &apiv1.NodeQuota{
						NodeSpec: &apiv1.NodeSpec{Cpu: 4, MemoryGi: 8}, NodeCount: 2, MinNodeCount: 2, MaxNodeCount: 2,
					},
				},
				// Non-whitelist: PD_SCHEDULING
				{
					Id: defaultCompId, ComponentType: apiv1.ComponentType_PD_SCHEDULING,
					Phase: apiv1.Component_AVAILABLE,
					NodeQuota: &apiv1.NodeQuota{
						NodeSpec: &apiv1.NodeSpec{Cpu: 4, MemoryGi: 8}, NodeCount: 2, MinNodeCount: 2, MaxNodeCount: 2,
					},
				},
				// Per-cluster: TIDB — managed separately, should NOT be in shared pool list
				{
					Id: defaultCompId, ComponentType: apiv1.ComponentType_TIDB,
					Phase: apiv1.Component_AVAILABLE,
					NodeQuota: &apiv1.NodeQuota{
						NodeSpec: &apiv1.NodeSpec{Cpu: 4, MemoryGi: 8}, NodeCount: 2, MinNodeCount: 1, MaxNodeCount: 10,
					},
					Storages: []*apiv1.Component_Storage{
						{Name: apiv1.Component_Storage_DATA, Type: "gp3", SizeGi: 100},
					},
				},
				// TiFlash Compute — must pre-exist with storages so generatedSharedPoolComponent
				// doesn't try to create it (which requires initConfig).
				{
					Id: defaultCompId, ComponentType: apiv1.ComponentType_TIFLASH_COMPUTE,
					Phase: apiv1.Component_AVAILABLE,
					NodeQuota: &apiv1.NodeQuota{
						NodeSpec: &apiv1.NodeSpec{Cpu: 4, MemoryGi: 8}, NodeCount: 1, MaxNodeCount: 12,
					},
					Storages: []*apiv1.Component_Storage{
						{Name: apiv1.Component_Storage_DATA, Type: "gp3", SizeGi: 100},
					},
				},
				// TiFlash Write — same reason as compute.
				{
					Id: defaultCompId, ComponentType: apiv1.ComponentType_TIFLASH_WRITE,
					Phase: apiv1.Component_AVAILABLE,
					NodeQuota: &apiv1.NodeQuota{
						NodeSpec: &apiv1.NodeSpec{Cpu: 4, MemoryGi: 8}, NodeCount: 1, MaxNodeCount: 12,
					},
					Storages: []*apiv1.Component_Storage{
						{Name: apiv1.Component_Storage_DATA, Type: "gp3", SizeGi: 100},
					},
				},
			},
		},
	}
	sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", false).Return(sharedPool, nil)
	setupAllDAGMocks(dagExecutor)
	dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil).Maybe()
	scaleManager.On("AnyFullfillingDecision", mock.Anything, mock.Anything).Return(false, nil).Maybe()

	// getRequestUnitTotal calls tidbClusterRepo.List
	tidbClusterRepo.On("List", mock.Anything, mock.Anything).Return([]*entity.TiDBCluster{currCluster}, nil)
	tidbClusterRepo.On("Save", mock.Anything, mock.Anything).Return(nil)
	tidbGroupRepo.On("List", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return([]*entity.TiDBGroup{}, nil).Maybe()

	// nodeSpecMap: return specs that cover all component types
	dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListNodePoolsResponse{
		NodePools: []*dplanemgrv1.NodePool{
			{CpuCore: 4, MemoryGi: 8, NodeType: dplanemgrv1.NodeType_Dedicated, AvailableCpuMilliCore: 3500, AvailableMemoryMi: 7000},
			{CpuCore: 8, MemoryGi: 16, NodeType: dplanemgrv1.NodeType_Dedicated, AvailableCpuMilliCore: 7000, AvailableMemoryMi: 15000},
			{CpuCore: 16, MemoryGi: 64, NodeType: dplanemgrv1.NodeType_Dedicated, AvailableCpuMilliCore: 14000, AvailableMemoryMi: 60000},
		},
	}, nil)

	ruConfigRepo.On("GetRuConfig", mock.Anything, "testsharedpoolid", apiv1.RuConfig_SHARED_POOL).Return(&entity.RuConfig{
		RuConfig: apiv1.RuConfig{
			FeatureFlags: &apiv1.RuConfig_FeatureFlags{
				EnableScaleSharedPoolByRu: proto.Bool(true),
				EnableScaleTidbGroupByRu:  proto.Bool(true),
			},
			Components: map[string]*apiv1.RuConfig_ComponentsMap{
				"tikv":               {MaxNodeCount: 36, NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8C16G": {Min: 0, Max: 1500}, "16C64G": {Min: 1501, Max: 200000}}},
				"tidb":               {MaxNodeCount: 12, NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"4C8G": {Min: 0, Max: 200000}}},
				"tidb_worker":        {MaxNodeCount: 15, NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"4C8G": {Min: 0, Max: 1500}, "8C16G": {Min: 1501, Max: 200000}}},
				"coprocessor_worker": {MaxNodeCount: 15, NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8C16G": {Min: 0, Max: 200000}}},
				"tikv_worker":        {MaxNodeCount: 15, NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8C16G": {Min: 0, Max: 200000}}},
				"pd":                 {MaxNodeCount: 3, NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"4C8G": {Min: 0, Max: 200000}}},
				"pd_tso":             {MaxNodeCount: 2, NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"4C8G": {Min: 0, Max: 200000}}},
				"pd_scheduling":      {MaxNodeCount: 2, NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"4C8G": {Min: 0, Max: 200000}}},
				"tiflash_compute":    {MaxNodeCount: 12, NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"4C8G": {Min: 0, Max: 200000}}},
				"tiflash_write":      {MaxNodeCount: 12, NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"4C8G": {Min: 0, Max: 200000}}},
				"tiproxy":            {MaxNodeCount: 12, NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8C16G": {Min: 0, Max: 200000}}},
			},
		},
	}, nil)

	// Capture the submitted DAG args and verify component filtering.
	var capturedComponentIDs, capturedComponentTypes string
	dagExecutor.On("Submit", mock.Anything, dag.PlanUpdateTiDBCluster, mock.MatchedBy(func(args actions.ActionArgs) bool {
		capturedComponentIDs, _ = args.GetString(workflow.ComponentIdsKey)
		capturedComponentTypes, _ = args.GetString(workflow.ComponentTypesKey)
		return true // always match, we assert below
	}), mock.Anything).Return(nil)

	server := &serverImpl{
		tidbClusterRepo: tidbClusterRepo,
		sharedPoolRepo:  sharedPoolRepo,
		tidbGroupRepo:   tidbGroupRepo,
		lockRepo:        lockRepo,
		ruConfigRepo:    ruConfigRepo,
		dagExecutor:     dagExecutor,
		transactioner:   &mockTransactioner{},
		dataplaneClient: dataplaneClient,
		scaleManager:    scaleManager,
	}

	_, err := server.UpdateTiDBCluster(context.Background(), &entity.TiDBCluster{
		Id: "test-cluster-id",
		RequestUnit: &apiv1.Cluster_RequestUnit{
			Min: 200,
			Max: 2000,
		},
	}, false)

	require.NoError(t, err)
	dagExecutor.AssertCalled(t, "Submit", mock.Anything, dag.PlanUpdateTiDBCluster, mock.Anything, mock.Anything)

	// Parse the submitted component IDs and types.
	ids := splitNonEmpty(capturedComponentIDs)
	types := splitNonEmpty(capturedComponentTypes)

	// Build a set of submitted component types for assertion.
	submittedTypes := make(map[string]bool)
	for _, ct := range types {
		submittedTypes[ct] = true
	}

	// Whitelist SharedPool components should be present.
	// Note: GetComponentTypeName maps TIKV→"tikv", TIDB_WORKER→"worker-tidb",
	// COPROCESSOR_WORKER→"coprocessor-tikv-worker", etc.
	assert.Contains(t, submittedTypes, "tikv", "TIKV (whitelist) should be submitted")
	assert.Contains(t, submittedTypes, "worker-tidb", "TIDB_WORKER (whitelist) should be submitted")

	// Non-whitelist components must NOT be present.
	assert.NotContains(t, submittedTypes, "pd", "PD (non-whitelist) must not be submitted")
	assert.NotContains(t, submittedTypes, "tso", "PD_TSO (non-whitelist) must not be submitted")
	assert.NotContains(t, submittedTypes, "scheduling", "PD_SCHEDULING (non-whitelist) must not be submitted")

	// Per-cluster components should not be in the shared pool list.
	assert.NotContains(t, submittedTypes, "tidb", "TIDB (per-cluster) must not be in shared pool list")
	assert.NotContains(t, submittedTypes, "tiproxy", "TIPROXY must not be submitted")

	// IDs and types must be the same length.
	assert.Len(t, types, len(ids), "componentIds and componentTypes must have same length")
}

func TestUpdateTiDBClusterDoesNotResizeEssentialV2ComponentsByRU(t *testing.T) {
	tidbClusterRepo := repomnocks.NewTiDBClusterRepo(t)
	sharedPoolRepo := repomnocks.NewSharedPoolRepo(t)
	tidbGroupRepo := repomnocks.NewTiDBGroupRepo(t)
	lockRepo := repomnocks.NewLockRepo(t)
	ruConfigRepo := repomnocks.NewRuConfigRepo(t)
	dagExecutor := executormocks.NewRemoteExecutor(t)
	dataplaneClient := dataplanemgrmocks.NewClient(t)
	scaleManager := bractionmocks.NewScaleManager(t)

	currCluster := &entity.TiDBCluster{
		Id:           "test-cluster-id",
		SharedPoolId: "essential-v2-pool",
		TenantId:     "test-tenant-id",
		Phase:        apiv1.Cluster_AVAILABLE,
		RequestUnit:  &apiv1.Cluster_RequestUnit{Min: 100, Max: 100},
	}
	tidbClusterRepo.On("Load", mock.Anything, currCluster.GetId(), false).Return(currCluster, nil)
	tidbClusterRepo.On("Save", mock.Anything, mock.Anything).Return(nil)
	lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil)

	worker := &entity.Component{
		Id:            utils.FnvHash(currCluster.GetSharedPoolId()),
		ComponentType: apiv1.ComponentType_TIKV_WORKER,
		Phase:         apiv1.Component_AVAILABLE,
		NodeQuota: &entity.NodeQuota{
			NodeSpec:     &entity.NodeSpec{Cpu: 4, MemoryGi: 16},
			NodeCount:    1,
			MinNodeCount: 1,
			MaxNodeCount: 15,
		},
	}
	tidb := &entity.Component{
		Id:            "tidb-group-id",
		ComponentType: apiv1.ComponentType_TIDB,
		Phase:         apiv1.Component_AVAILABLE,
		NodeQuota: &entity.NodeQuota{
			NodeSpec:     &entity.NodeSpec{Cpu: 4, MemoryGi: 8},
			NodeCount:    1,
			MinNodeCount: 1,
			MaxNodeCount: 10,
		},
		Storages: []*entity.Storage{{Name: apiv1.Component_Storage_DATA, Type: "gp3", SizeGi: 100}},
	}
	sharedPool := &entity.SharedPool{
		SharedPool: apiv1.SharedPool{
			Id:              currCluster.GetSharedPoolId(),
			DataplaneName:   "test-dataplane",
			Phase:           apiv1.SharedPool_AVAILABLE,
			Version:         "v7.1.0",
			Provider:        &apiv1.Provider{Type: apiv1.ProviderType_AWS},
			ComponentGroups: []*entity.Component{worker, tidb},
		},
		DeployType: schema.SharedPoolDeployTypeSharedTiProxy,
	}
	sharedPoolRepo.On("Load", mock.Anything, sharedPool.GetId(), false).Return(sharedPool, nil)

	setupAllDAGMocks(dagExecutor)
	dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).
		Return(executor.NewExecTaskIdStatus(), nil).Maybe()
	scaleManager.On("AnyFullfillingDecision", mock.Anything, mock.Anything).Return(false, nil).Maybe()
	ruConfigRepo.On("GetRuConfig", mock.Anything, sharedPool.GetId(), apiv1.RuConfig_SHARED_POOL).Return(&entity.RuConfig{
		RuConfig: apiv1.RuConfig{
			FeatureFlags: &apiv1.RuConfig_FeatureFlags{
				EnableScaleSharedPoolByRu: proto.Bool(true),
				EnableScaleTidbGroupByRu:  proto.Bool(true),
			},
			Components: map[string]*apiv1.RuConfig_ComponentsMap{
				"tikv_worker": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{
					"8c16g": {Min: 0, Max: 0},
				}},
				"tidb": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{
					"8c16g": {Min: 0, Max: 0},
				}},
			},
		},
	}, nil)

	var capturedArgs actions.ActionArgs
	dagExecutor.On("Submit", mock.Anything, dag.PlanUpdateTiDBCluster, mock.Anything, mock.Anything).
		Run(func(args mock.Arguments) { capturedArgs = args.Get(2).(actions.ActionArgs) }).
		Return(nil)

	server := &serverImpl{
		tidbClusterRepo: tidbClusterRepo,
		sharedPoolRepo:  sharedPoolRepo,
		tidbGroupRepo:   tidbGroupRepo,
		lockRepo:        lockRepo,
		ruConfigRepo:    ruConfigRepo,
		dagExecutor:     dagExecutor,
		transactioner:   &mockTransactioner{},
		dataplaneClient: dataplaneClient,
		scaleManager:    scaleManager,
	}

	result, err := server.UpdateTiDBCluster(context.Background(), &entity.TiDBCluster{
		Id:          currCluster.GetId(),
		RequestUnit: &apiv1.Cluster_RequestUnit{Min: 2000, Max: 2100},
	}, false)
	require.NoError(t, err)
	assert.Equal(t, apiv1.Cluster_AVAILABLE, result.GetPhase())
	require.NotNil(t, capturedArgs)

	sharedPoolJSON, err := capturedArgs.GetString(workflow.SharedPoolKey)
	require.NoError(t, err)
	var submittedSharedPool entity.SharedPool
	require.NoError(t, protojson.Unmarshal([]byte(sharedPoolJSON), &submittedSharedPool))
	require.Len(t, submittedSharedPool.GetComponentGroups(), 2)
	assert.Equal(t, uint32(4), submittedSharedPool.GetComponentGroups()[0].GetNodeQuota().GetNodeSpec().GetCpu())
	assert.Equal(t, uint32(4), submittedSharedPool.GetComponentGroups()[1].GetNodeQuota().GetNodeSpec().GetCpu())
	assert.Empty(t, mustGetActionArgString(t, capturedArgs, workflow.ComponentIdsKey))
	assert.Empty(t, mustGetActionArgString(t, capturedArgs, workflow.ComponentTypesKey))
	assert.Equal(t, workflow.SkipComponentHPAApplyValue, mustGetActionArgString(t, capturedArgs, workflow.SkipComponentHPAApplyKey))
	dataplaneClient.AssertNotCalled(t, "ListNodePools", mock.Anything, mock.Anything)
	tidbClusterRepo.AssertNotCalled(t, "List", mock.Anything, mock.Anything)
	tidbGroupRepo.AssertNotCalled(t, "List", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything)
}

func TestUpdateTiDBClusterScalesUpExistingRCUTiDBGroup(t *testing.T) {
	for _, tt := range []struct {
		name                     string
		currentRequestUnit       *apiv1.Cluster_RequestUnit
		requestUnit              *apiv1.Cluster_RequestUnit
		existingPhase            apiv1.Component_Phase
		existingNodeCPU          uint32
		existingNodeMemoryGi     uint32
		existingNodeCount        uint32
		expectUpdate             bool
		expectFailedPrecondition bool
		expectedNodeCPU          uint32
		expectedCount            uint32
	}{
		{
			name:                 "crosses RU tier and submits changed TiDB group",
			requestUnit:          &apiv1.Cluster_RequestUnit{Min: 200, Max: 200},
			existingPhase:        apiv1.Component_AVAILABLE,
			existingNodeCPU:      4,
			existingNodeMemoryGi: 8,
			existingNodeCount:    1,
			expectUpdate:         true,
			expectedNodeCPU:      8,
			expectedCount:        1,
		},
		{
			name:                 "same RU tier increase does not reconcile drift",
			currentRequestUnit:   &apiv1.Cluster_RequestUnit{Min: 50, Max: 50},
			requestUnit:          &apiv1.Cluster_RequestUnit{Min: 100, Max: 100},
			existingPhase:        apiv1.Component_AVAILABLE,
			existingNodeCPU:      2,
			existingNodeMemoryGi: 4,
			existingNodeCount:    1,
			expectUpdate:         false,
			expectedNodeCPU:      2,
			expectedCount:        1,
		},
		{
			name:                 "RU decrease remains owned by scale down job",
			currentRequestUnit:   &apiv1.Cluster_RequestUnit{Min: 200, Max: 200},
			requestUnit:          &apiv1.Cluster_RequestUnit{Min: 100, Max: 100},
			existingPhase:        apiv1.Component_AVAILABLE,
			existingNodeCPU:      8,
			existingNodeMemoryGi: 16,
			existingNodeCount:    2,
			expectUpdate:         false,
			expectedNodeCPU:      8,
			expectedCount:        2,
		},
		{
			name:                 "pre-existing modifying TiDB in same tier is not resubmitted",
			currentRequestUnit:   &apiv1.Cluster_RequestUnit{Min: 50, Max: 50},
			requestUnit:          &apiv1.Cluster_RequestUnit{Min: 100, Max: 100},
			existingPhase:        apiv1.Component_MODIFYING,
			existingNodeCPU:      4,
			existingNodeMemoryGi: 8,
			existingNodeCount:    1,
			expectUpdate:         false,
			expectedNodeCPU:      4,
			expectedCount:        1,
		},
		{
			name:                 "upward RU tier does not reduce an overprovisioned spec",
			requestUnit:          &apiv1.Cluster_RequestUnit{Min: 200, Max: 200},
			existingPhase:        apiv1.Component_AVAILABLE,
			existingNodeCPU:      16,
			existingNodeMemoryGi: 32,
			existingNodeCount:    2,
			expectUpdate:         false,
			expectedNodeCPU:      16,
			expectedCount:        2,
		},
		{
			name:                     "modifying TiDB with changed quota fails before save and submit",
			requestUnit:              &apiv1.Cluster_RequestUnit{Min: 200, Max: 200},
			existingPhase:            apiv1.Component_MODIFYING,
			existingNodeCPU:          4,
			existingNodeMemoryGi:     8,
			existingNodeCount:        1,
			expectFailedPrecondition: true,
		},
		{
			name:                     "creating TiDB fails before save and submit",
			requestUnit:              &apiv1.Cluster_RequestUnit{Min: 200, Max: 200},
			existingPhase:            apiv1.Component_CREATING,
			existingNodeCPU:          4,
			existingNodeMemoryGi:     8,
			existingNodeCount:        1,
			expectFailedPrecondition: true,
		},
		{
			name:                 "target spec already satisfied does not raise count to tier baseline",
			requestUnit:          &apiv1.Cluster_RequestUnit{Min: 200, Max: 200},
			existingPhase:        apiv1.Component_AVAILABLE,
			existingNodeCPU:      8,
			existingNodeMemoryGi: 16,
			existingNodeCount:    1,
			expectUpdate:         false,
			expectedNodeCPU:      8,
			expectedCount:        1,
		},
	} {
		t.Run(tt.name, func(t *testing.T) {
			tidbClusterRepo := repomnocks.NewTiDBClusterRepo(t)
			sharedPoolRepo := repomnocks.NewSharedPoolRepo(t)
			tidbGroupRepo := repomnocks.NewTiDBGroupRepo(t)
			lockRepo := repomnocks.NewLockRepo(t)
			ruConfigRepo := repomnocks.NewRuConfigRepo(t)
			dagExecutor := executormocks.NewRemoteExecutor(t)
			dataplaneClient := dataplanemgrmocks.NewClient(t)
			scaleManager := bractionmocks.NewScaleManager(t)

			currentRequestUnit := tt.currentRequestUnit
			if currentRequestUnit == nil {
				currentRequestUnit = &apiv1.Cluster_RequestUnit{Min: 100, Max: 100}
			}
			currCluster := &entity.TiDBCluster{
				Id:           "test-cluster-id",
				SharedPoolId: "test-sharedpool-id",
				TenantId:     "test-tenant-id",
				Phase:        apiv1.Cluster_AVAILABLE,
				RequestUnit:  currentRequestUnit,
			}
			tidbClusterRepo.On("Load", mock.Anything, currCluster.GetId(), false).Return(currCluster, nil)
			if !tt.expectFailedPrecondition {
				tidbClusterRepo.On("Save", mock.Anything, mock.Anything).Return(nil)
			}
			lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil)

			existingTiDB := &entity.Component{
				Id:            "tidb-group-id",
				ComponentType: apiv1.ComponentType_TIDB,
				Phase:         tt.existingPhase,
				NodeQuota: &entity.NodeQuota{
					NodeCount:    tt.existingNodeCount,
					MinNodeCount: 1,
					MaxNodeCount: 10,
					NodeSpec:     &entity.NodeSpec{Cpu: tt.existingNodeCPU, MemoryGi: tt.existingNodeMemoryGi, AvailableCpu: tt.existingNodeCPU, AvailableMemoryMi: tt.existingNodeMemoryGi * 1024},
				},
				Storages: []*entity.Storage{{Name: apiv1.Component_Storage_DATA, Type: "gp3", SizeGi: 100}},
			}
			unrelatedPD := &entity.Component{
				Id:            "unrelated-pd-id",
				ComponentType: apiv1.ComponentType_PD,
				Phase:         apiv1.Component_MODIFYING,
				NodeQuota:     &entity.NodeQuota{NodeCount: 3, MinNodeCount: 3, MaxNodeCount: 3, NodeSpec: &entity.NodeSpec{Cpu: 4, MemoryGi: 8}},
			}
			sharedPool := &entity.SharedPool{SharedPool: apiv1.SharedPool{
				Id:              currCluster.GetSharedPoolId(),
				DataplaneName:   "test-dataplane",
				Phase:           apiv1.SharedPool_AVAILABLE,
				Provider:        &apiv1.Provider{Type: apiv1.ProviderType_AWS},
				ComponentGroups: []*apiv1.Component{existingTiDB, unrelatedPD},
			}}
			sharedPoolRepo.On("Load", mock.Anything, currCluster.GetSharedPoolId(), false).Return(sharedPool, nil)
			tidbGroupRepo.On("List", mock.Anything, []string{currCluster.GetSharedPoolId()}, mock.Anything, []string{currCluster.GetId()}, mock.Anything, false).
				Return([]*entity.TiDBGroup{{TiDBGroup: apiv1.TiDBGroup{Id: existingTiDB.GetId()}}}, nil)

			setupAllDAGMocks(dagExecutor)
			dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).
				Return(executor.NewExecTaskIdStatus(), nil).Maybe()
			scaleManager.On("AnyFullfillingDecision", mock.Anything, mock.Anything).Return(false, nil).Maybe()
			dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListNodePoolsResponse{
				NodePools: []*dplanemgrv1.NodePool{
					{CpuCore: 4, MemoryGi: 8, NodeType: dplanemgrv1.NodeType_Dedicated, AvailableCpuMilliCore: 4000, AvailableMemoryMi: 8192},
					{CpuCore: 8, MemoryGi: 16, NodeType: dplanemgrv1.NodeType_Dedicated, AvailableCpuMilliCore: 8000, AvailableMemoryMi: 16384},
				},
			}, nil)
			ruConfigRepo.On("GetRuConfig", mock.Anything, currCluster.GetSharedPoolId(), apiv1.RuConfig_SHARED_POOL).Return(&entity.RuConfig{
				RuConfig: apiv1.RuConfig{
					FeatureFlags: &apiv1.RuConfig_FeatureFlags{EnableScaleTidbGroupByRu: proto.Bool(true)},
					Components: map[string]*apiv1.RuConfig_ComponentsMap{
						"tidb": {MaxNodeCount: 10, NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{
							"4c8g":  {Min: 0, Max: 100, DefaultNodeCount: uint32Ptr(1)},
							"8c16g": {Min: 101, Max: 0, DefaultNodeCount: uint32Ptr(3)},
						}},
					},
				},
			}, nil)

			var capturedArgs actions.ActionArgs
			if !tt.expectFailedPrecondition {
				dagExecutor.On("Submit", mock.Anything, dag.PlanUpdateTiDBCluster, mock.Anything, mock.Anything).Run(func(args mock.Arguments) {
					capturedArgs = args.Get(2).(actions.ActionArgs)
				}).Return(nil)
			}

			server := &serverImpl{
				tidbClusterRepo: tidbClusterRepo,
				sharedPoolRepo:  sharedPoolRepo,
				tidbGroupRepo:   tidbGroupRepo,
				lockRepo:        lockRepo,
				ruConfigRepo:    ruConfigRepo,
				dagExecutor:     dagExecutor,
				transactioner:   &mockTransactioner{},
				dataplaneClient: dataplaneClient,
				scaleManager:    scaleManager,
			}

			_, err := server.UpdateTiDBCluster(context.Background(), &entity.TiDBCluster{
				Id:          currCluster.GetId(),
				RequestUnit: tt.requestUnit,
			}, false)
			if tt.expectFailedPrecondition {
				require.Error(t, err)
				assert.Equal(t, codes.FailedPrecondition, status.Code(err))
				tidbClusterRepo.AssertNotCalled(t, "Save", mock.Anything, mock.Anything)
				dagExecutor.AssertNotCalled(t, "Submit", mock.Anything, dag.PlanUpdateTiDBCluster, mock.Anything, mock.Anything)
				return
			}
			require.NoError(t, err)
			require.NotNil(t, capturedArgs)

			sharedPoolJSON, err := capturedArgs.GetString(workflow.SharedPoolKey)
			require.NoError(t, err)
			var submittedSharedPool entity.SharedPool
			require.NoError(t, protojson.Unmarshal([]byte(sharedPoolJSON), &submittedSharedPool))
			var submittedTiDB *entity.Component
			for _, component := range submittedSharedPool.GetComponentGroups() {
				if component.GetComponentType() == apiv1.ComponentType_TIDB && component.GetId() == existingTiDB.GetId() {
					submittedTiDB = component
					break
				}
			}
			require.NotNil(t, submittedTiDB)
			require.Equal(t, tt.expectedNodeCPU, submittedTiDB.GetNodeQuota().GetNodeSpec().GetCpu())
			require.Equal(t, tt.expectedCount, submittedTiDB.GetNodeQuota().GetNodeCount())
			require.Equal(t, uint32(1), submittedTiDB.GetNodeQuota().GetMinNodeCount())
			require.Equal(t, uint32(10), submittedTiDB.GetNodeQuota().GetMaxNodeCount())

			optionsKey := entity.GetUpdateComponentKey(existingTiDB)
			options, optionsErr := capturedArgs.GetString(optionsKey)
			if tt.expectUpdate {
				require.Equal(t, apiv1.Component_MODIFYING, submittedTiDB.GetPhase())
				require.NoError(t, optionsErr)
				assert.Equal(t, []string{"update-node-spec"}, splitNonEmpty(options))
				assert.Contains(t, splitNonEmpty(mustGetActionArgString(t, capturedArgs, workflow.ComponentIdsKey)), existingTiDB.GetId())
				assert.Contains(t, splitNonEmpty(mustGetActionArgString(t, capturedArgs, workflow.ComponentTypesKey)), "tidb")
				assert.NotContains(t, splitNonEmpty(mustGetActionArgString(t, capturedArgs, workflow.ComponentIdsKey)), unrelatedPD.GetId())
				assert.NotContains(t, splitNonEmpty(mustGetActionArgString(t, capturedArgs, workflow.ComponentTypesKey)), "pd")
				_, pdOptionsErr := capturedArgs.GetString(entity.GetUpdateComponentKey(unrelatedPD))
				require.Error(t, pdOptionsErr)
				return
			}

			require.Equal(t, tt.existingPhase, submittedTiDB.GetPhase())
			require.Error(t, optionsErr)
			assert.NotContains(t, splitNonEmpty(mustGetActionArgString(t, capturedArgs, workflow.ComponentIdsKey)), existingTiDB.GetId())
			assert.NotContains(t, splitNonEmpty(mustGetActionArgString(t, capturedArgs, workflow.ComponentTypesKey)), "tidb")
		})
	}
}

func mustGetActionArgString(t *testing.T, args actions.ActionArgs, key string) string {
	t.Helper()
	value, err := args.GetString(key)
	require.NoError(t, err)
	return value
}

// tkrwUpdateRuConfig returns a ru config with EnableScaleSharedPoolByRu
// enabled (driving UpdateTiDBCluster into the branch that calls
// guardTiKVRemoteWorkersTopologyStable) and EnableScaleTidbGroupByRu disabled
// (so the test does not also need to wire tidbGroupRepo.List).
func tkrwUpdateRuConfig() *entity.RuConfig {
	return &entity.RuConfig{
		RuConfig: apiv1.RuConfig{
			FeatureFlags: &apiv1.RuConfig_FeatureFlags{
				EnableScaleSharedPoolByRu: proto.Bool(true),
				EnableScaleTidbGroupByRu:  proto.Bool(false),
			},
			Components: map[string]*apiv1.RuConfig_ComponentsMap{
				"tikv": {MaxNodeCount: 36, NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8C16G": {Min: 0, Max: 200000}}},
			},
		},
	}
}

func TestUpdateTiDBCluster_GuardBlocksOnRunningTiKVRemoteWorkerDAG(t *testing.T) {
	tidbClusterRepo := repomnocks.NewTiDBClusterRepo(t)
	sharedPoolRepo := repomnocks.NewSharedPoolRepo(t)
	lockRepo := repomnocks.NewLockRepo(t)
	ruConfigRepo := repomnocks.NewRuConfigRepo(t)
	dagExecutor := executormocks.NewRemoteExecutor(t)
	dataplaneClient := dataplanemgrmocks.NewClient(t)
	scaleManager := bractionmocks.NewScaleManager(t)

	currCluster := &entity.TiDBCluster{
		Id:           "test-cluster-id",
		SharedPoolId: testTKRWSharedPoolID,
		TenantId:     "test-tenant-id",
		Phase:        apiv1.Cluster_AVAILABLE,
		Version:      "v7.1.0",
		RequestUnit: &apiv1.Cluster_RequestUnit{
			Min: 100,
			Max: 1000,
		},
	}
	tidbClusterRepo.On("Load", mock.Anything, "test-cluster-id", false).Return(currCluster, nil)
	lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil)

	// Uses PlanDisableCoprocessorWorker here (paired with the Enable-direction
	// running-plan case in TestCreateTiDBCluster_GuardBlocksOnRunningTiKVRemoteWorkerDAG)
	// to prove the guard's running-plan check queries both routing-switch
	// directions, not just enable.
	setupAllDAGMocksWithConflict(dagExecutor, dag.PlanDisableCoprocessorWorker, runningDagStatus(tkrwKey(t, tkrwEnabledBoth(t))))
	scaleManager.On("AnyFullfillingDecision", mock.Anything, mock.Anything).Return(false, nil).Maybe()

	sharedPool := newTKRWSharedPool(tkrwEnabledBoth(t), 3)
	sharedPoolRepo.On("Load", mock.Anything, testTKRWSharedPoolID, false).Return(sharedPool, nil)
	dataplaneClient.On("ListNodePoolConfigs", mock.Anything, mock.Anything).Return(tkrwNodePoolConfigsResponse(), nil)
	ruConfigRepo.On("GetRuConfig", mock.Anything, testTKRWSharedPoolID, apiv1.RuConfig_SHARED_POOL).Return(tkrwUpdateRuConfig(), nil)
	tidbClusterRepo.On("List", mock.Anything, mock.Anything).Return([]*entity.TiDBCluster{currCluster}, nil)

	server := &serverImpl{
		tidbClusterRepo: tidbClusterRepo,
		sharedPoolRepo:  sharedPoolRepo,
		lockRepo:        lockRepo,
		ruConfigRepo:    ruConfigRepo,
		dagExecutor:     dagExecutor,
		transactioner:   &mockTransactioner{},
		dataplaneClient: dataplaneClient,
		scaleManager:    scaleManager,
	}

	result, err := server.UpdateTiDBCluster(context.Background(), &entity.TiDBCluster{
		Id: "test-cluster-id",
		RequestUnit: &apiv1.Cluster_RequestUnit{
			Min: 200,
			Max: 2000,
		},
	}, false)

	require.Error(t, err)
	require.Nil(t, result)
	statusErr, ok := status.FromError(err)
	require.True(t, ok)
	assert.Equal(t, codes.FailedPrecondition, statusErr.Code())
	dagExecutor.AssertNotCalled(t, "Submit", mock.Anything, dag.PlanUpdateTiDBCluster, mock.Anything, mock.Anything)
}

func TestUpdateTiDBCluster_GuardBlocksOnUnconvergedTiKVRemoteWorkerTopology(t *testing.T) {
	tidbClusterRepo := repomnocks.NewTiDBClusterRepo(t)
	sharedPoolRepo := repomnocks.NewSharedPoolRepo(t)
	lockRepo := repomnocks.NewLockRepo(t)
	ruConfigRepo := repomnocks.NewRuConfigRepo(t)
	dagExecutor := executormocks.NewRemoteExecutor(t)
	dataplaneClient := dataplanemgrmocks.NewClient(t)
	scaleManager := bractionmocks.NewScaleManager(t)

	currCluster := &entity.TiDBCluster{
		Id:           "test-cluster-id",
		SharedPoolId: testTKRWSharedPoolID,
		TenantId:     "test-tenant-id",
		Phase:        apiv1.Cluster_AVAILABLE,
		Version:      "v7.1.0",
		RequestUnit: &apiv1.Cluster_RequestUnit{
			Min: 100,
			Max: 1000,
		},
	}
	tidbClusterRepo.On("Load", mock.Anything, "test-cluster-id", false).Return(currCluster, nil)
	lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil)

	setupAllDAGMocks(dagExecutor)
	scaleManager.On("AnyFullfillingDecision", mock.Anything, mock.Anything).Return(false, nil).Maybe()

	// Persisted target already equals the guard's normalized target, so the
	// guard reaches the live-CR check - which sees no CR at all (zero objects
	// seeded), i.e. genuine drift rather than a legacy-mismatch shortcut.
	sharedPool := newTKRWSharedPool(tkrwEnabledBoth(t), 3)
	sharedPoolRepo.On("Load", mock.Anything, testTKRWSharedPoolID, false).Return(sharedPool, nil)
	dataplaneClient.On("ListNodePoolConfigs", mock.Anything, mock.Anything).Return(tkrwNodePoolConfigsResponse(), nil)
	ruConfigRepo.On("GetRuConfig", mock.Anything, testTKRWSharedPoolID, apiv1.RuConfig_SHARED_POOL).Return(tkrwUpdateRuConfig(), nil)
	tidbClusterRepo.On("List", mock.Anything, mock.Anything).Return([]*entity.TiDBCluster{currCluster}, nil)

	server := &serverImpl{
		tidbClusterRepo:  tidbClusterRepo,
		sharedPoolRepo:   sharedPoolRepo,
		lockRepo:         lockRepo,
		ruConfigRepo:     ruConfigRepo,
		dagExecutor:      dagExecutor,
		transactioner:    &mockTransactioner{},
		dataplaneClient:  dataplaneClient,
		scaleManager:     scaleManager,
		dataplaneManager: newTKRWDataplaneManager(t),
	}

	result, err := server.UpdateTiDBCluster(context.Background(), &entity.TiDBCluster{
		Id: "test-cluster-id",
		RequestUnit: &apiv1.Cluster_RequestUnit{
			Min: 200,
			Max: 2000,
		},
	}, false)

	require.Error(t, err)
	require.Nil(t, result)
	statusErr, ok := status.FromError(err)
	require.True(t, ok)
	assert.Equal(t, codes.FailedPrecondition, statusErr.Code())
	dagExecutor.AssertNotCalled(t, "Submit", mock.Anything, dag.PlanUpdateTiDBCluster, mock.Anything, mock.Anything)
}

// splitNonEmpty splits a comma-separated string, filtering out empty strings.
func splitNonEmpty(s string) []string {
	if s == "" {
		return nil
	}
	var out []string
	for _, part := range strings.Split(s, ",") {
		if part != "" {
			out = append(out, part)
		}
	}
	return out
}