package server

import (
	"context"
	"testing"
	"time"

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

type mockCassandraProxyTransactioner struct{}

func (m *mockCassandraProxyTransactioner) Transaction(ctx context.Context, fn func(context.Context) error) error {
	return fn(ctx)
}

type cassandraProxyMocks struct {
	tidbRepo     *repomocks.TiDBClusterRepo
	spRepo       *repomocks.SharedPoolRepo
	stateRepo    *repomocks.CassandraProxyStateRepo
	lockRepo     *repomocks.LockRepo
	dagExec      *executormocks.RemoteExecutor
	scaleManager *bractionmocks.ScaleManager
	dpClient     *dpmgrmocks.Client
}

func newCassandraProxyTestServer(t *testing.T) (*serverImpl, *cassandraProxyMocks) {
	t.Helper()
	m := &cassandraProxyMocks{
		tidbRepo:     repomocks.NewTiDBClusterRepo(t),
		spRepo:       repomocks.NewSharedPoolRepo(t),
		stateRepo:    repomocks.NewCassandraProxyStateRepo(t),
		lockRepo:     repomocks.NewLockRepo(t),
		dagExec:      executormocks.NewRemoteExecutor(t),
		scaleManager: bractionmocks.NewScaleManager(t),
		dpClient:     dpmgrmocks.NewClient(t),
	}
	m.dpClient.On("GetDataPlane", mock.Anything, mock.MatchedBy(func(req *dplanemgrv1.GetDataPlaneRequest) bool {
		return req.GetDataPlaneId() == "dp-test"
	})).Return(&dplanemgrv1.DataPlane{Capabilities: []string{"loadbalancer.multi-port"}}, nil).Maybe()
	s := &serverImpl{
		tidbClusterRepo:         m.tidbRepo,
		sharedPoolRepo:          m.spRepo,
		cassandraProxyStateRepo: m.stateRepo,
		lockRepo:                m.lockRepo,
		dagExecutor:             m.dagExec,
		scaleManager:            m.scaleManager,
		dataplaneClient:         m.dpClient,
		transactioner:           &mockCassandraProxyTransactioner{},
		imageConfig:             &config.ImageConfig{CassandraProxyImage: "ghcr.io/pingcap/cassandra-proxy:v0.5.2"},
	}
	return s, m
}

func newCassandraProxyState(phase apiv1.Component_Phase) *entity.CassandraProxyState {
	return &entity.CassandraProxyState{
		ClusterID:          "c1",
		SharedPoolID:       testCassandraProxySharedPoolId,
		HostID:             "stable-host-id",
		NodeSpec:           "8C16G",
		Image:              "ghcr.io/pingcap/cassandra-proxy:v0.5.2",
		Phase:              entity.ComponentPhaseToCassandraProxyPhase(phase),
		PrivatelinkDNSName: "",
		NodeQuotaEntity: &apiv1.NodeQuota{
			NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableMemoryMi: 14000},
			NodeCount: 2,
		},
	}
}

func cassandraProxyImagePtr(image string) *string {
	return &image
}

func newAvailableSharedPool() *entity.SharedPool {
	return &entity.SharedPool{
		SharedPool: apiv1.SharedPool{
			Id:            "9999",
			Phase:         apiv1.SharedPool_AVAILABLE,
			DataplaneName: "dp-test",
		},
	}
}

const testCassandraProxySharedPoolId = "9999"

func newTestTiDBCluster(clusterId string) *entity.TiDBCluster {
	return &entity.TiDBCluster{
		Id:              clusterId,
		SharedPoolId:    testCassandraProxySharedPoolId,
		GatewayIdentity: "gateway-identity",
	}
}

func TestSetCassandraProxy_ValidationErrors(t *testing.T) {
	validQuota := &apiv1.NodeQuota{
		NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableMemoryMi: 14000},
		NodeCount: 2,
	}
	tests := []struct {
		name     string
		req      *apiv1.SetCassandraProxyRequest
		wantCode codes.Code
		wantMsg  string
	}{
		{
			name:     "missing cluster_id",
			req:      &apiv1.SetCassandraProxyRequest{NodeQuota: validQuota},
			wantCode: codes.InvalidArgument,
			wantMsg:  "cluster_id is required",
		},
		{
			name:     "missing node_quota",
			req:      &apiv1.SetCassandraProxyRequest{ClusterId: "1234567890"},
			wantCode: codes.InvalidArgument,
			wantMsg:  "node_quota is required",
		},
		{
			name: "missing node_quota.node_spec",
			req: &apiv1.SetCassandraProxyRequest{
				ClusterId: "1234567890",
				NodeQuota: &apiv1.NodeQuota{NodeCount: 2},
			},
			wantCode: codes.InvalidArgument,
			wantMsg:  "node_quota.node_spec is required",
		},
		{
			name: "cpu below minimum",
			req: &apiv1.SetCassandraProxyRequest{
				ClusterId: "1234567890",
				NodeQuota: &apiv1.NodeQuota{NodeSpec: &apiv1.NodeSpec{Cpu: 1, MemoryGi: 16}, NodeCount: 2},
			},
			wantCode: codes.InvalidArgument,
			wantMsg:  "cpu must be >= 2",
		},
		{
			name: "memory_gi below minimum",
			req: &apiv1.SetCassandraProxyRequest{
				ClusterId: "1234567890",
				NodeQuota: &apiv1.NodeQuota{NodeSpec: &apiv1.NodeSpec{Cpu: 8, MemoryGi: 2}, NodeCount: 2},
			},
			wantCode: codes.InvalidArgument,
			wantMsg:  "memory_gi must be >= 4",
		},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			s, _ := newCassandraProxyTestServer(t)
			_, err := s.SetCassandraProxy(context.Background(), tc.req)
			require.Error(t, err)
			st, ok := status.FromError(err)
			require.True(t, ok)
			assert.Equal(t, tc.wantCode, st.Code())
			assert.Contains(t, st.Message(), tc.wantMsg)
		})
	}
}

func TestSetCassandraProxy_HandleCreate_Success(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)

	clusterId := "1234567890"
	m.tidbRepo.On("Load", mock.Anything, clusterId, false).Return(newTestTiDBCluster(clusterId), nil)

	// pre-check load (no component yet)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, clusterId, false).Return(nil, status.Error(codes.NotFound, "not found")).Once()
	// inside-tx load
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, clusterId, false).Return(nil, status.Error(codes.NotFound, "not found")).Once()

	m.lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(l *entity.Lock) bool { return l.Key == "9999" })).
		Return(func() error { return nil }, nil)

	// mergeSharedPoolStatusFromDagTask
	m.dagExec.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
	m.scaleManager.On("AnyFullfillingDecision", mock.Anything, "9999").Return(false, nil)

	var savedSP *entity.SharedPool
	var savedState *entity.CassandraProxyState
	m.stateRepo.On("Save", mock.Anything, mock.MatchedBy(func(state *entity.CassandraProxyState) bool {
		savedState = state
		return state.ClusterID == clusterId && state.SharedPoolID == "9999"
	})).Return(nil)
	m.spRepo.On("Save", mock.Anything, mock.MatchedBy(func(sp *entity.SharedPool) bool {
		savedSP = sp
		return sp.GetId() == "9999" && sp.GetPhase() == apiv1.SharedPool_MODIFYING
	})).Return(nil)

	var submittedArgs actions.ActionArgs
	m.dagExec.On("Submit", mock.Anything, dag.PlanCreateCassandraProxy, mock.MatchedBy(func(args actions.ActionArgs) bool {
		submittedArgs = args
		clusterID, err := args.GetString(workflow.TiDBClusterIdKey)
		return err == nil && clusterID == clusterId
	}), mock.Anything).Return(nil)

	resp, err := s.SetCassandraProxy(context.Background(), &apiv1.SetCassandraProxyRequest{
		ClusterId: clusterId,
		NodeQuota: &apiv1.NodeQuota{
			NodeSpec:     &apiv1.NodeSpec{Cpu: 4, MemoryGi: 8},
			NodeCount:    1,
			MinNodeCount: 1,
			MaxNodeCount: 5,
		},
	})
	require.NoError(t, err)
	require.NotNil(t, resp)
	assert.NotEmpty(t, resp.GetHostId())
	assert.Equal(t, apiv1.Component_CREATING, resp.GetPhase())
	assert.Equal(t, "cassandra-1234567890.tidb9999.svc.cluster.local:9042", resp.GetInternalEndpoint())
	assert.Empty(t, resp.GetPrivatelinkDnsName())
	assert.Equal(t, "ghcr.io/pingcap/cassandra-proxy:v0.5.2", resp.GetImage())

	require.NotNil(t, savedSP)
	require.Len(t, savedSP.GetComponentGroups(), 1)
	added := savedSP.GetComponentGroups()[0]
	assert.Equal(t, apiv1.ComponentType_CASSANDRA_PROXY, added.GetComponentType())
	assert.Equal(t, clusterId, added.GetId())
	assert.Equal(t, apiv1.Component_CREATING, added.GetPhase())
	assert.Equal(t, resp.GetHostId(), added.GetLabels()[cassandraProxyHostIdLabelKey])
	assert.Equal(t, "4C8G", added.GetLabels()[cassandraProxyNodeSpecLabelKey])
	require.NotNil(t, savedState)
	assert.Equal(t, resp.GetHostId(), savedState.HostID)
	assert.Equal(t, "4C8G", savedState.NodeSpec)
	assert.Equal(t, "ghcr.io/pingcap/cassandra-proxy:v0.5.2", savedState.Image)
	assert.Equal(t, entity.CassandraProxyPhaseCreating, savedState.Phase)
	assert.Equal(t, uint32(1), savedState.NodeQuotaEntity.GetNodeCount())
	assert.Equal(t, uint32(1), savedState.NodeQuotaEntity.GetMinNodeCount())
	assert.Equal(t, uint32(5), savedState.NodeQuotaEntity.GetMaxNodeCount())
	sharedPoolArg, err := submittedArgs.GetString(workflow.SharedPoolKey)
	require.NoError(t, err)
	assert.NotContains(t, sharedPoolArg, "cassandra-proxy/image")
}

func TestSetCassandraProxy_HandleCreate_MissingImageFailsBeforeDurableState(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)
	s.imageConfig = &config.ImageConfig{CassandraProxyImage: "   "}

	clusterId := "1234567890"
	m.tidbRepo.On("Load", mock.Anything, clusterId, false).Return(newTestTiDBCluster(clusterId), nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, clusterId, false).Return(nil, status.Error(codes.NotFound, "not found")).Once()

	_, err := s.SetCassandraProxy(context.Background(), &apiv1.SetCassandraProxyRequest{
		ClusterId: clusterId,
		NodeQuota: &apiv1.NodeQuota{
			NodeSpec:  &apiv1.NodeSpec{Cpu: 4, MemoryGi: 8},
			NodeCount: 1,
		},
	})
	require.Error(t, err)
	assert.Equal(t, codes.FailedPrecondition, status.Code(err))
	assert.Contains(t, status.Convert(err).Message(), "cassandra-proxy image is not configured")
}

func TestSetCassandraProxy_HandleCreate_ExplicitImage(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)

	clusterId := "1234567890"
	image := "ghcr.io/pingcap/cassandra-proxy:v0.6.0"
	m.tidbRepo.On("Load", mock.Anything, clusterId, false).Return(newTestTiDBCluster(clusterId), nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, clusterId, false).Return(nil, status.Error(codes.NotFound, "not found")).Once()
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, clusterId, false).Return(nil, status.Error(codes.NotFound, "not found")).Once()
	m.lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(l *entity.Lock) bool { return l.Key == "9999" })).
		Return(func() error { return nil }, nil).Once()
	m.dagExec.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
	m.scaleManager.On("AnyFullfillingDecision", mock.Anything, "9999").Return(false, nil).Once()

	var savedState *entity.CassandraProxyState
	m.stateRepo.On("Save", mock.Anything, mock.MatchedBy(func(state *entity.CassandraProxyState) bool {
		savedState = state
		return state.ClusterID == clusterId
	})).Return(nil).Once()
	m.spRepo.On("Save", mock.Anything, mock.Anything).Return(nil).Once()
	m.dagExec.On("Submit", mock.Anything, dag.PlanCreateCassandraProxy, mock.Anything, mock.Anything).Return(nil).Once()

	resp, err := s.SetCassandraProxy(context.Background(), &apiv1.SetCassandraProxyRequest{
		ClusterId: clusterId,
		NodeQuota: &apiv1.NodeQuota{
			NodeSpec:  &apiv1.NodeSpec{Cpu: 4, MemoryGi: 8},
			NodeCount: 1,
		},
		Image: cassandraProxyImagePtr(image),
	})
	require.NoError(t, err)
	require.NotNil(t, resp)
	assert.Equal(t, apiv1.Component_CREATING, resp.GetPhase())
	assert.Equal(t, image, resp.GetImage())
	require.NotNil(t, savedState)
	assert.Equal(t, image, savedState.Image)
}

func TestSetCassandraProxy_HandleCreate_ExplicitEmptyImageReturnsInvalidArgument(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)

	clusterId := "1234567890"
	m.tidbRepo.On("Load", mock.Anything, clusterId, false).Return(newTestTiDBCluster(clusterId), nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, clusterId, false).Return(nil, status.Error(codes.NotFound, "not found")).Once()

	_, err := s.SetCassandraProxy(context.Background(), &apiv1.SetCassandraProxyRequest{
		ClusterId: clusterId,
		NodeQuota: &apiv1.NodeQuota{
			NodeSpec:  &apiv1.NodeSpec{Cpu: 4, MemoryGi: 8},
			NodeCount: 1,
		},
		Image: cassandraProxyImagePtr(""),
	})
	require.Error(t, err)
	assert.Equal(t, codes.InvalidArgument, status.Code(err))
	assert.Contains(t, status.Convert(err).Message(), "invalid image")
}

func TestSetCassandraProxy_HandleCreate_DedupesStaleSharedPoolComponent(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)

	clusterId := "1234567890"
	sharedPool := newAvailableSharedPool()
	sharedPool.ComponentGroups = []*entity.Component{
		{
			Id:            clusterId,
			ComponentType: apiv1.ComponentType_CASSANDRA_PROXY,
			Phase:         apiv1.Component_UNAVAILABLE,
			Labels: map[string]string{
				cassandraProxyHostIdLabelKey: "stale-host-id",
			},
		},
		{
			Id:            clusterId,
			ComponentType: apiv1.ComponentType_TIKV,
			Phase:         apiv1.Component_AVAILABLE,
		},
	}

	m.tidbRepo.On("Load", mock.Anything, clusterId, false).Return(newTestTiDBCluster(clusterId), nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, clusterId, false).Return(nil, status.Error(codes.NotFound, "not found")).Once()
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(sharedPool, nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, clusterId, false).Return(nil, status.Error(codes.NotFound, "not found")).Once()
	m.lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(l *entity.Lock) bool { return l.Key == "9999" })).
		Return(func() error { return nil }, nil)
	m.dagExec.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
	m.scaleManager.On("AnyFullfillingDecision", mock.Anything, "9999").Return(false, nil)
	m.stateRepo.On("Save", mock.Anything, mock.Anything).Return(nil).Once()

	var savedSP *entity.SharedPool
	m.spRepo.On("Save", mock.Anything, mock.MatchedBy(func(sp *entity.SharedPool) bool {
		savedSP = sp
		return sp.GetId() == "9999" && sp.GetPhase() == apiv1.SharedPool_MODIFYING
	})).Return(nil).Once()

	var submittedArgs actions.ActionArgs
	m.dagExec.On("Submit", mock.Anything, dag.PlanCreateCassandraProxy, mock.MatchedBy(func(args actions.ActionArgs) bool {
		submittedArgs = args
		return true
	}), mock.Anything).Return(nil).Once()

	resp, err := s.SetCassandraProxy(context.Background(), &apiv1.SetCassandraProxyRequest{
		ClusterId: clusterId,
		NodeQuota: &apiv1.NodeQuota{
			NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableMemoryMi: 14000},
			NodeCount: 2,
		},
	})
	require.NoError(t, err)
	require.NotNil(t, resp)
	assert.NotEqual(t, "stale-host-id", resp.GetHostId())

	require.NotNil(t, savedSP)
	require.Len(t, savedSP.GetComponentGroups(), 2)
	var cassandraComponents []*entity.Component
	for _, component := range savedSP.GetComponentGroups() {
		if component.GetId() == clusterId && component.GetComponentType() == apiv1.ComponentType_CASSANDRA_PROXY {
			cassandraComponents = append(cassandraComponents, component)
		}
	}
	require.Len(t, cassandraComponents, 1)
	assert.Equal(t, resp.GetHostId(), cassandraComponents[0].GetLabels()[cassandraProxyHostIdLabelKey])

	sharedPoolArg, err := submittedArgs.GetString(workflow.SharedPoolKey)
	require.NoError(t, err)
	assert.Contains(t, sharedPoolArg, resp.GetHostId())
	assert.NotContains(t, sharedPoolArg, "stale-host-id")
}

func TestSetCassandraProxy_HandleCreate_SaveAlreadyExistsFallsBackToStaleUpdate(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)

	clusterId := "c1"
	m.tidbRepo.On("Load", mock.Anything, clusterId, false).Return(newTestTiDBCluster(clusterId), nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, clusterId, false).Return(nil, status.Error(codes.NotFound, "not found")).Once()
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, clusterId, false).Return(nil, status.Error(codes.NotFound, "not found")).Once()
	m.lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(l *entity.Lock) bool { return l.Key == "9999" })).
		Return(func() error { return nil }, nil)
	m.dagExec.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
	m.scaleManager.On("AnyFullfillingDecision", mock.Anything, "9999").Return(false, nil)
	m.stateRepo.On("Save", mock.Anything, mock.Anything).Return(status.Error(codes.AlreadyExists, "duplicate")).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, clusterId, false).Return(newCassandraProxyState(apiv1.Component_CREATING), nil).Once()
	m.dagExec.On("QueryRunnings", mock.Anything, dag.PlanCreateCassandraProxy, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil).Maybe()
	m.stateRepo.On("UpdatePhaseIfCurrent", mock.Anything, "c1", "stable-host-id", []entity.ComponentPhase{apiv1.Component_CREATING, apiv1.Component_MODIFYING}, apiv1.Component_UNAVAILABLE).
		Return(newCassandraProxyState(apiv1.Component_UNAVAILABLE), true, nil).Once()

	_, err := s.SetCassandraProxy(context.Background(), &apiv1.SetCassandraProxyRequest{
		ClusterId: clusterId,
		NodeQuota: &apiv1.NodeQuota{
			NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableMemoryMi: 14000},
			NodeCount: 2,
		},
	})
	require.Error(t, err)
	assert.Equal(t, codes.FailedPrecondition, status.Code(err))
	assert.Contains(t, status.Convert(err).Message(), "cannot retry from UNAVAILABLE")
}

func TestSetCassandraProxy_HandleCreate_SaveAlreadyExistsReloadNil(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)

	clusterId := "c1"
	m.tidbRepo.On("Load", mock.Anything, clusterId, false).Return(newTestTiDBCluster(clusterId), nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, clusterId, false).Return(nil, status.Error(codes.NotFound, "not found")).Once()
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, clusterId, false).Return(nil, status.Error(codes.NotFound, "not found")).Once()
	m.lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(l *entity.Lock) bool { return l.Key == "9999" })).
		Return(func() error { return nil }, nil)
	m.dagExec.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
	m.scaleManager.On("AnyFullfillingDecision", mock.Anything, "9999").Return(false, nil)
	m.stateRepo.On("Save", mock.Anything, mock.Anything).Return(status.Error(codes.AlreadyExists, "duplicate")).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, clusterId, false).Return(nil, nil).Once()

	_, err := s.SetCassandraProxy(context.Background(), &apiv1.SetCassandraProxyRequest{
		ClusterId: clusterId,
		NodeQuota: &apiv1.NodeQuota{
			NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableMemoryMi: 14000},
			NodeCount: 2,
		},
	})
	require.Error(t, err)
	assert.Equal(t, codes.Internal, status.Code(err))
	assert.Contains(t, status.Convert(err).Message(), "vanished")
}

func TestSetCassandraProxy_HandleUpdate_AvailableNoOp(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)

	clusterId := "c1"
	m.tidbRepo.On("Load", mock.Anything, clusterId, false).Return(newTestTiDBCluster(clusterId), nil)

	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil)
	m.stateRepo.On("LoadByClusterID", mock.Anything, clusterId, false).Return(newCassandraProxyState(apiv1.Component_AVAILABLE), nil)
	m.dpClient.On("GetPrivateLinkService", mock.Anything, mock.MatchedBy(func(req *dplanemgrv1.GetPrivateLinkServiceRequest) bool {
		return req.GetDataPlaneId() == "dp-test" && req.GetId() == "cassandra-private-c1"
	})).Return(&dplanemgrv1.PrivateLinkService{ServiceDnsName: "privatelink-cassandra.identity.dev.tidbcloud.com"}, nil).Once()

	resp, err := s.SetCassandraProxy(context.Background(), &apiv1.SetCassandraProxyRequest{
		ClusterId: clusterId,
		NodeQuota: &apiv1.NodeQuota{
			NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableMemoryMi: 14000},
			NodeCount: 2,
		},
	})
	require.NoError(t, err)
	require.NotNil(t, resp)
	assert.Equal(t, "stable-host-id", resp.GetHostId())
	assert.Equal(t, apiv1.Component_AVAILABLE, resp.GetPhase())
	assert.Equal(t, "privatelink-cassandra.identity.dev.tidbcloud.com", resp.GetPrivatelinkDnsName())
	assert.Equal(t, "ghcr.io/pingcap/cassandra-proxy:v0.5.2", resp.GetImage())
}

func TestSetCassandraProxy_HandleUpdate_AvailableNoOpLegacyImageReturnsFallback(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)
	s.imageConfig = &config.ImageConfig{CassandraProxyImage: "ghcr.io/pingcap/cassandra-proxy:legacy-backfill"}

	clusterId := "c1"
	legacyState := newCassandraProxyState(apiv1.Component_AVAILABLE)
	legacyState.Image = ""
	m.tidbRepo.On("Load", mock.Anything, clusterId, false).Return(newTestTiDBCluster(clusterId), nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, clusterId, false).Return(legacyState, nil).Once()
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	m.dpClient.On("GetPrivateLinkService", mock.Anything, mock.Anything).
		Return(nil, status.Error(codes.NotFound, "not found")).Maybe()

	resp, err := s.SetCassandraProxy(context.Background(), &apiv1.SetCassandraProxyRequest{
		ClusterId: clusterId,
		NodeQuota: &apiv1.NodeQuota{
			NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableMemoryMi: 14000},
			NodeCount: 2,
		},
	})
	require.NoError(t, err)
	require.NotNil(t, resp)
	assert.Equal(t, apiv1.Component_AVAILABLE, resp.GetPhase())
	assert.Equal(t, "ghcr.io/pingcap/cassandra-proxy:legacy-backfill", resp.GetImage())
}

func TestSetCassandraProxy_HandleUpdate_AvailableNoOpKeepsPersistedDNS(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)

	clusterId := "c1"
	state := newCassandraProxyState(apiv1.Component_AVAILABLE)
	state.PrivatelinkDNSName = "persisted.example.com"
	m.tidbRepo.On("Load", mock.Anything, clusterId, false).Return(newTestTiDBCluster(clusterId), nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, clusterId, false).Return(state, nil).Once()

	resp, err := s.SetCassandraProxy(context.Background(), &apiv1.SetCassandraProxyRequest{
		ClusterId: clusterId,
		NodeQuota: &apiv1.NodeQuota{
			NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableMemoryMi: 14000},
			NodeCount: 2,
		},
	})
	require.NoError(t, err)
	require.NotNil(t, resp)
	assert.Equal(t, apiv1.Component_AVAILABLE, resp.GetPhase())
	assert.Equal(t, "persisted.example.com", resp.GetPrivatelinkDnsName())
	assert.Equal(t, "ghcr.io/pingcap/cassandra-proxy:v0.5.2", resp.GetImage())
}

func TestSetCassandraProxy_HandleUpdate_PhaseBranches(t *testing.T) {
	tests := []struct {
		name     string
		phase    apiv1.Component_Phase
		wantErr  bool
		wantCode codes.Code
	}{
		{name: "creating running no-op", phase: apiv1.Component_CREATING, wantErr: false},
		{name: "modifying running no-op", phase: apiv1.Component_MODIFYING, wantErr: false},
		{name: "deleting -> failed precondition", phase: apiv1.Component_DELETING, wantErr: true, wantCode: codes.FailedPrecondition},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			s, m := newCassandraProxyTestServer(t)
			s.imageConfig = &config.ImageConfig{CassandraProxyImage: "ghcr.io/pingcap/cassandra-proxy:v0.5.9"}
			m.tidbRepo.On("Load", mock.Anything, "c1", false).Return(newTestTiDBCluster("c1"), nil)
			m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil)
			m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(newCassandraProxyState(tc.phase), nil)
			if tc.phase == apiv1.Component_CREATING || tc.phase == apiv1.Component_MODIFYING {
				running := executor.NewExecTaskIdStatus()
				running["exec-1"] = &executor.ExecTaskStatusEx{State: executor.ExecTaskStateTypeRunningOK}
				m.dagExec.On("QueryRunnings", mock.Anything, dag.PlanCreateCassandraProxy, mock.Anything).Return(running, nil).Once()
			}

			resp, err := s.SetCassandraProxy(context.Background(), &apiv1.SetCassandraProxyRequest{
				ClusterId: "c1",
				NodeQuota: &apiv1.NodeQuota{
					NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableMemoryMi: 14000},
					NodeCount: 2,
				},
			})
			if tc.wantErr {
				require.Error(t, err)
				assert.Equal(t, tc.wantCode, status.Code(err))
				return
			}
			require.NoError(t, err)
			assert.Equal(t, tc.phase, resp.GetPhase())
		})
	}
}

func TestSetCassandraProxy_HandleUpdate_NoRunningDagReturnsFailedPrecondition(t *testing.T) {
	tests := []struct {
		name  string
		phase apiv1.Component_Phase
	}{
		{name: "creating", phase: apiv1.Component_CREATING},
		{name: "modifying", phase: apiv1.Component_MODIFYING},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			s, m := newCassandraProxyTestServer(t)
			m.tidbRepo.On("Load", mock.Anything, "c1", false).Return(newTestTiDBCluster("c1"), nil)
			m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil)
			m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(newCassandraProxyState(tc.phase), nil)
			m.dagExec.On("QueryRunnings", mock.Anything, dag.PlanCreateCassandraProxy, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil).Once()
			m.stateRepo.On("UpdatePhaseIfCurrent", mock.Anything, "c1", "stable-host-id", []entity.ComponentPhase{apiv1.Component_CREATING, apiv1.Component_MODIFYING}, apiv1.Component_UNAVAILABLE).
				Return(newCassandraProxyState(apiv1.Component_UNAVAILABLE), true, nil).Once()

			_, err := s.SetCassandraProxy(context.Background(), &apiv1.SetCassandraProxyRequest{
				ClusterId: "c1",
				NodeQuota: &apiv1.NodeQuota{
					NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableMemoryMi: 14000},
					NodeCount: 2,
				},
			})
			require.Error(t, err)
			assert.Equal(t, codes.FailedPrecondition, status.Code(err))
			assert.Contains(t, status.Convert(err).Message(), "cannot retry from UNAVAILABLE")
		})
	}
}

func TestSetCassandraProxy_HandleUpdate_NoRunningDagReloadsFreshAvailableState(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)
	m.tidbRepo.On("Load", mock.Anything, "c1", false).Return(newTestTiDBCluster("c1"), nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil)
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(newCassandraProxyState(apiv1.Component_CREATING), nil).Once()
	m.dagExec.On("QueryRunnings", mock.Anything, dag.PlanCreateCassandraProxy, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil).Once()
	freshState := newCassandraProxyState(apiv1.Component_AVAILABLE)
	m.stateRepo.On("UpdatePhaseIfCurrent", mock.Anything, "c1", "stable-host-id", []entity.ComponentPhase{apiv1.Component_CREATING, apiv1.Component_MODIFYING}, apiv1.Component_UNAVAILABLE).
		Return(freshState, false, nil).Once()

	resp, err := s.SetCassandraProxy(context.Background(), &apiv1.SetCassandraProxyRequest{
		ClusterId: "c1",
		NodeQuota: &apiv1.NodeQuota{
			NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableMemoryMi: 14000},
			NodeCount: 2,
		},
	})
	require.NoError(t, err)
	require.NotNil(t, resp)
	assert.Equal(t, apiv1.Component_AVAILABLE, resp.GetPhase())
	assert.Equal(t, "stable-host-id", resp.GetHostId())
	assert.Equal(t, "ghcr.io/pingcap/cassandra-proxy:v0.5.2", resp.GetImage())
}

func TestSetCassandraProxy_HandleUpdate_UnavailableRetriesSameSpec(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)

	sharedPool := newAvailableSharedPool()
	sharedPool.ComponentGroups = []*entity.Component{
		{
			Id:            "c1",
			ComponentType: apiv1.ComponentType_CASSANDRA_PROXY,
			Phase:         apiv1.Component_UNAVAILABLE,
			Labels: map[string]string{
				cassandraProxyHostIdLabelKey: "stale-host-id",
			},
		},
		{
			Id:            "c1",
			ComponentType: apiv1.ComponentType_TIKV,
			Phase:         apiv1.Component_AVAILABLE,
		},
	}

	m.tidbRepo.On("Load", mock.Anything, "c1", false).Return(newTestTiDBCluster("c1"), nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(newCassandraProxyState(apiv1.Component_UNAVAILABLE), nil).Once()
	m.lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(l *entity.Lock) bool { return l.Key == "9999" })).
		Return(func() error { return nil }, nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(newCassandraProxyState(apiv1.Component_UNAVAILABLE), nil).Once()
	m.dagExec.On("QueryRunnings", mock.Anything, dag.PlanCreateCassandraProxy, mock.MatchedBy(func(filter executor.MatchFilter) bool {
		return assert.ObjectsAreEqual([]string{"9999"}, filter[workflow.SharedPoolIdKey]) &&
			assert.ObjectsAreEqual([]string{"c1"}, filter[workflow.ComponentIdsKey])
	})).Return(executor.NewExecTaskIdStatus(), nil).Once()
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(sharedPool, nil).Once()
	m.dagExec.On("QueryRunnings", mock.Anything, mock.MatchedBy(func(plan string) bool {
		return plan != dag.PlanCreateCassandraProxy
	}), mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
	m.scaleManager.On("AnyFullfillingDecision", mock.Anything, "9999").Return(false, nil).Once()
	m.stateRepo.On("UpdatePhaseIfCurrent", mock.Anything, "c1", "stable-host-id", []entity.ComponentPhase{apiv1.Component_UNAVAILABLE}, apiv1.Component_CREATING).
		Return(newCassandraProxyState(apiv1.Component_CREATING), true, nil).Once()

	var savedSP *entity.SharedPool
	m.spRepo.On("Save", mock.Anything, mock.MatchedBy(func(sp *entity.SharedPool) bool {
		savedSP = sp
		return sp.GetId() == "9999" && sp.GetPhase() == apiv1.SharedPool_MODIFYING
	})).Return(nil).Once()

	var submittedArgs actions.ActionArgs
	m.dagExec.On("Submit", mock.Anything, dag.PlanCreateCassandraProxy, mock.MatchedBy(func(args actions.ActionArgs) bool {
		submittedArgs = args
		clusterID, err := args.GetString(workflow.TiDBClusterIdKey)
		return err == nil && clusterID == "c1"
	}), mock.Anything).Return(nil).Once()

	resp, err := s.SetCassandraProxy(context.Background(), &apiv1.SetCassandraProxyRequest{
		ClusterId: "c1",
		NodeQuota: &apiv1.NodeQuota{
			NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableMemoryMi: 14000},
			NodeCount: 2,
		},
	})
	require.NoError(t, err)
	require.NotNil(t, resp)
	assert.Equal(t, "stable-host-id", resp.GetHostId())
	assert.Equal(t, apiv1.Component_CREATING, resp.GetPhase())
	assert.Equal(t, "cassandra-c1.tidb9999.svc.cluster.local:9042", resp.GetInternalEndpoint())
	assert.Equal(t, "ghcr.io/pingcap/cassandra-proxy:v0.5.2", resp.GetImage())

	require.NotNil(t, savedSP)
	require.Len(t, savedSP.GetComponentGroups(), 2)
	var cassandraComponents []*entity.Component
	for _, component := range savedSP.GetComponentGroups() {
		if component.GetId() == "c1" && component.GetComponentType() == apiv1.ComponentType_CASSANDRA_PROXY {
			cassandraComponents = append(cassandraComponents, component)
		}
	}
	require.Len(t, cassandraComponents, 1)
	assert.Equal(t, apiv1.Component_CREATING, cassandraComponents[0].GetPhase())
	assert.Equal(t, "stable-host-id", cassandraComponents[0].GetLabels()[cassandraProxyHostIdLabelKey])
	assert.Equal(t, "8C16G", cassandraComponents[0].GetLabels()[cassandraProxyNodeSpecLabelKey])

	sharedPoolArg, err := submittedArgs.GetString(workflow.SharedPoolKey)
	require.NoError(t, err)
	assert.Contains(t, sharedPoolArg, "stable-host-id")
	assert.NotContains(t, sharedPoolArg, "stale-host-id")
}

func TestSetCassandraProxy_HandleUpdate_ForceChangesUnavailableSpec(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)
	sharedPool := newAvailableSharedPool()
	sharedPool.Phase = apiv1.SharedPool_UNAVAILABLE
	sharedPool.ComponentGroups = []*entity.Component{{
		Id:            "c1",
		ComponentType: apiv1.ComponentType_CASSANDRA_PROXY,
		Phase:         apiv1.Component_UNAVAILABLE,
	}}

	m.tidbRepo.On("Load", mock.Anything, "c1", false).Return(newTestTiDBCluster("c1"), nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(sharedPool, nil).Twice()
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(newCassandraProxyState(apiv1.Component_UNAVAILABLE), nil).Twice()
	m.lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(l *entity.Lock) bool { return l.Key == "9999" })).
		Return(func() error { return nil }, nil).Once()
	m.dagExec.On("QueryRunnings", mock.Anything, dag.PlanCreateCassandraProxy, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil).Once()
	m.dagExec.On("QueryRunnings", mock.Anything, mock.MatchedBy(func(plan string) bool {
		return plan != dag.PlanCreateCassandraProxy
	}), mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
	m.scaleManager.On("AnyFullfillingDecision", mock.Anything, "9999").Return(false, nil).Once()
	updatedState := newCassandraProxyState(apiv1.Component_CREATING)
	m.stateRepo.On("UpdatePhaseIfCurrent", mock.Anything, "c1", "stable-host-id", []entity.ComponentPhase{apiv1.Component_UNAVAILABLE}, apiv1.Component_CREATING).
		Return(updatedState, true, nil).Once()
	m.stateRepo.On("Save", mock.Anything, mock.MatchedBy(func(state *entity.CassandraProxyState) bool {
		return state.NodeSpec == "16C32G" &&
			state.NodeQuotaEntity.GetNodeCount() == 4 &&
			state.Image == "ghcr.io/pingcap/cassandra-proxy:v0.6.0"
	})).Return(nil).Once()
	m.spRepo.On("Save", mock.Anything, mock.MatchedBy(func(sp *entity.SharedPool) bool {
		for _, component := range sp.GetComponentGroups() {
			if component.GetId() == "c1" && component.GetComponentType() == apiv1.ComponentType_CASSANDRA_PROXY {
				return component.GetPhase() == apiv1.Component_CREATING &&
					component.GetNodeQuota().GetNodeCount() == 4 &&
					component.GetLabels()[cassandraProxyNodeSpecLabelKey] == "16C32G"
			}
		}
		return false
	})).Return(nil).Once()
	m.dagExec.On("Submit", mock.Anything, dag.PlanCreateCassandraProxy, mock.Anything, mock.Anything).Return(nil).Once()

	resp, err := s.SetCassandraProxy(context.Background(), &apiv1.SetCassandraProxyRequest{
		ClusterId: "c1",
		NodeQuota: &apiv1.NodeQuota{
			NodeSpec:  &apiv1.NodeSpec{Cpu: 16, MemoryGi: 32, AvailableMemoryMi: 28000},
			NodeCount: 4,
		},
		Image: cassandraProxyImagePtr("ghcr.io/pingcap/cassandra-proxy:v0.6.0"),
		Force: true,
	})

	require.NoError(t, err)
	assert.Equal(t, apiv1.Component_CREATING, resp.GetPhase())
	assert.Equal(t, "ghcr.io/pingcap/cassandra-proxy:v0.6.0", resp.GetImage())
}

func TestSetCassandraProxy_HandleUpdate_ForceChangeRejectsRunningWorkflow(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)
	running := executor.NewExecTaskIdStatus()
	running["exec-1"] = &executor.ExecTaskStatusEx{}

	m.tidbRepo.On("Load", mock.Anything, "c1", false).Return(newTestTiDBCluster("c1"), nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(newCassandraProxyState(apiv1.Component_UNAVAILABLE), nil).Twice()
	m.lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil).Once()
	m.dagExec.On("QueryRunnings", mock.Anything, dag.PlanCreateCassandraProxy, mock.Anything).Return(running, nil).Once()

	_, err := s.SetCassandraProxy(context.Background(), &apiv1.SetCassandraProxyRequest{
		ClusterId: "c1",
		NodeQuota: &apiv1.NodeQuota{
			NodeSpec:  &apiv1.NodeSpec{Cpu: 16, MemoryGi: 32, AvailableMemoryMi: 28000},
			NodeCount: 2,
		},
		Force: true,
	})

	require.Error(t, err)
	assert.Equal(t, codes.FailedPrecondition, status.Code(err))
	assert.Contains(t, status.Convert(err).Message(), "while a workflow is running")
}

func TestSetCassandraProxy_HandleUpdate_ForceChangeRejectsFreshInFlightState(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)
	m.tidbRepo.On("Load", mock.Anything, "c1", false).Return(newTestTiDBCluster("c1"), nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(newCassandraProxyState(apiv1.Component_UNAVAILABLE), nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(newCassandraProxyState(apiv1.Component_CREATING), nil).Once()
	m.lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil).Once()

	_, err := s.SetCassandraProxy(context.Background(), &apiv1.SetCassandraProxyRequest{
		ClusterId: "c1",
		NodeQuota: &apiv1.NodeQuota{
			NodeSpec:  &apiv1.NodeSpec{Cpu: 16, MemoryGi: 32, AvailableMemoryMi: 28000},
			NodeCount: 2,
		},
		Force: true,
	})

	require.Error(t, err)
	assert.Equal(t, codes.FailedPrecondition, status.Code(err))
	assert.Contains(t, status.Convert(err).Message(), "retry after AVAILABLE")
}

func TestSetCassandraProxy_HandleUpdate_UnavailableRetryLegacyImageBackfillsResponse(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)
	s.imageConfig = &config.ImageConfig{CassandraProxyImage: "ghcr.io/pingcap/cassandra-proxy:legacy-backfill"}

	legacyState := newCassandraProxyState(apiv1.Component_UNAVAILABLE)
	legacyState.Image = ""
	freshLegacyState := newCassandraProxyState(apiv1.Component_UNAVAILABLE)
	freshLegacyState.Image = ""
	creatingLegacyState := newCassandraProxyState(apiv1.Component_CREATING)
	creatingLegacyState.Image = ""

	sharedPool := newAvailableSharedPool()
	sharedPool.ComponentGroups = []*entity.Component{{
		Id:            "c1",
		ComponentType: apiv1.ComponentType_CASSANDRA_PROXY,
		Phase:         apiv1.Component_UNAVAILABLE,
		Labels: map[string]string{
			cassandraProxyHostIdLabelKey:   "stable-host-id",
			cassandraProxyNodeSpecLabelKey: "8C16G",
		},
	}}

	m.tidbRepo.On("Load", mock.Anything, "c1", false).Return(newTestTiDBCluster("c1"), nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(legacyState, nil).Once()
	m.lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(l *entity.Lock) bool { return l.Key == "9999" })).
		Return(func() error { return nil }, nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(freshLegacyState, nil).Once()
	m.dagExec.On("QueryRunnings", mock.Anything, dag.PlanCreateCassandraProxy, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil).Once()
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(sharedPool, nil).Once()
	m.dagExec.On("QueryRunnings", mock.Anything, mock.MatchedBy(func(plan string) bool {
		return plan != dag.PlanCreateCassandraProxy
	}), mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
	m.scaleManager.On("AnyFullfillingDecision", mock.Anything, "9999").Return(false, nil).Once()
	m.stateRepo.On("UpdatePhaseIfCurrent", mock.Anything, "c1", "stable-host-id", []entity.ComponentPhase{apiv1.Component_UNAVAILABLE}, apiv1.Component_CREATING).
		Return(creatingLegacyState, true, nil).Once()

	var savedState *entity.CassandraProxyState
	m.stateRepo.On("Save", mock.Anything, mock.MatchedBy(func(st *entity.CassandraProxyState) bool {
		savedState = st
		return st.ClusterID == "c1" && st.Image == "ghcr.io/pingcap/cassandra-proxy:legacy-backfill"
	})).Return(nil).Once()
	m.spRepo.On("Save", mock.Anything, mock.Anything).Return(nil).Once()
	m.dagExec.On("Submit", mock.Anything, dag.PlanCreateCassandraProxy, mock.Anything, mock.Anything).Return(nil).Once()

	resp, err := s.SetCassandraProxy(context.Background(), &apiv1.SetCassandraProxyRequest{
		ClusterId: "c1",
		NodeQuota: &apiv1.NodeQuota{
			NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableMemoryMi: 14000},
			NodeCount: 2,
		},
	})
	require.NoError(t, err)
	require.NotNil(t, resp)
	assert.Equal(t, apiv1.Component_CREATING, resp.GetPhase())
	assert.Equal(t, "ghcr.io/pingcap/cassandra-proxy:legacy-backfill", resp.GetImage())
	require.NotNil(t, savedState)
	assert.Equal(t, "ghcr.io/pingcap/cassandra-proxy:legacy-backfill", savedState.Image)
}

func TestSetCassandraProxy_HandleUpdate_UnavailableRetryCASMissReturnsFreshAvailable(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)

	sharedPool := newAvailableSharedPool()
	sharedPool.ComponentGroups = []*entity.Component{{
		Id:            "c1",
		ComponentType: apiv1.ComponentType_CASSANDRA_PROXY,
		Phase:         apiv1.Component_UNAVAILABLE,
		Labels: map[string]string{
			cassandraProxyHostIdLabelKey: "stable-host-id",
		},
	}}
	availableState := newCassandraProxyState(apiv1.Component_AVAILABLE)

	m.tidbRepo.On("Load", mock.Anything, "c1", false).Return(newTestTiDBCluster("c1"), nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(newCassandraProxyState(apiv1.Component_UNAVAILABLE), nil).Once()
	m.lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(l *entity.Lock) bool { return l.Key == "9999" })).
		Return(func() error { return nil }, nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(newCassandraProxyState(apiv1.Component_UNAVAILABLE), nil).Once()
	m.dagExec.On("QueryRunnings", mock.Anything, dag.PlanCreateCassandraProxy, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil).Once()
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(sharedPool, nil).Once()
	m.dagExec.On("QueryRunnings", mock.Anything, mock.MatchedBy(func(plan string) bool {
		return plan != dag.PlanCreateCassandraProxy
	}), mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
	m.scaleManager.On("AnyFullfillingDecision", mock.Anything, "9999").Return(false, nil).Once()
	m.stateRepo.On("UpdatePhaseIfCurrent", mock.Anything, "c1", "stable-host-id", []entity.ComponentPhase{apiv1.Component_UNAVAILABLE}, apiv1.Component_CREATING).
		Return(availableState, false, nil).Once()

	resp, err := s.SetCassandraProxy(context.Background(), &apiv1.SetCassandraProxyRequest{
		ClusterId: "c1",
		NodeQuota: &apiv1.NodeQuota{
			NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableMemoryMi: 14000},
			NodeCount: 2,
		},
	})
	require.NoError(t, err)
	require.NotNil(t, resp)
	assert.Equal(t, apiv1.Component_AVAILABLE, resp.GetPhase())
	assert.Equal(t, "stable-host-id", resp.GetHostId())
}

func TestSetCassandraProxy_HandleUpdate_ForceChangeCASMissReturnsFailedPrecondition(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)
	sharedPool := newAvailableSharedPool()
	sharedPool.ComponentGroups = []*entity.Component{{
		Id:            "c1",
		ComponentType: apiv1.ComponentType_CASSANDRA_PROXY,
		Phase:         apiv1.Component_UNAVAILABLE,
	}}

	m.tidbRepo.On("Load", mock.Anything, "c1", false).Return(newTestTiDBCluster("c1"), nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(newCassandraProxyState(apiv1.Component_UNAVAILABLE), nil).Twice()
	m.lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil).Once()
	m.dagExec.On("QueryRunnings", mock.Anything, dag.PlanCreateCassandraProxy, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil).Once()
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(sharedPool, nil).Once()
	m.dagExec.On("QueryRunnings", mock.Anything, mock.MatchedBy(func(plan string) bool {
		return plan != dag.PlanCreateCassandraProxy
	}), mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
	m.scaleManager.On("AnyFullfillingDecision", mock.Anything, "9999").Return(false, nil).Once()
	m.stateRepo.On("UpdatePhaseIfCurrent", mock.Anything, "c1", "stable-host-id", []entity.ComponentPhase{apiv1.Component_UNAVAILABLE}, apiv1.Component_CREATING).
		Return(newCassandraProxyState(apiv1.Component_AVAILABLE), false, nil).Once()

	_, err := s.SetCassandraProxy(context.Background(), &apiv1.SetCassandraProxyRequest{
		ClusterId: "c1",
		NodeQuota: &apiv1.NodeQuota{
			NodeSpec:  &apiv1.NodeSpec{Cpu: 16, MemoryGi: 32, AvailableMemoryMi: 28000},
			NodeCount: 2,
		},
		Force: true,
	})

	require.Error(t, err)
	assert.Equal(t, codes.FailedPrecondition, status.Code(err))
	assert.Contains(t, status.Convert(err).Message(), "state changed before applying")
}

func TestSetCassandraProxy_HandleUpdate_UnavailableRetryCASMissLegacyImageReturnsFallback(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)
	s.imageConfig = &config.ImageConfig{CassandraProxyImage: "ghcr.io/pingcap/cassandra-proxy:legacy-backfill"}

	legacyState := newCassandraProxyState(apiv1.Component_UNAVAILABLE)
	legacyState.Image = ""
	freshLegacyState := newCassandraProxyState(apiv1.Component_UNAVAILABLE)
	freshLegacyState.Image = ""
	availableState := newCassandraProxyState(apiv1.Component_AVAILABLE)
	availableState.Image = ""

	sharedPool := newAvailableSharedPool()
	sharedPool.ComponentGroups = []*entity.Component{{
		Id:            "c1",
		ComponentType: apiv1.ComponentType_CASSANDRA_PROXY,
		Phase:         apiv1.Component_UNAVAILABLE,
		Labels: map[string]string{
			cassandraProxyHostIdLabelKey: "stable-host-id",
		},
	}}

	m.tidbRepo.On("Load", mock.Anything, "c1", false).Return(newTestTiDBCluster("c1"), nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(legacyState, nil).Once()
	m.lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(l *entity.Lock) bool { return l.Key == "9999" })).
		Return(func() error { return nil }, nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(freshLegacyState, nil).Once()
	m.dagExec.On("QueryRunnings", mock.Anything, dag.PlanCreateCassandraProxy, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil).Once()
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(sharedPool, nil).Once()
	m.dagExec.On("QueryRunnings", mock.Anything, mock.MatchedBy(func(plan string) bool {
		return plan != dag.PlanCreateCassandraProxy
	}), mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
	m.scaleManager.On("AnyFullfillingDecision", mock.Anything, "9999").Return(false, nil).Once()
	m.stateRepo.On("UpdatePhaseIfCurrent", mock.Anything, "c1", "stable-host-id", []entity.ComponentPhase{apiv1.Component_UNAVAILABLE}, apiv1.Component_CREATING).
		Return(availableState, false, nil).Once()

	resp, err := s.SetCassandraProxy(context.Background(), &apiv1.SetCassandraProxyRequest{
		ClusterId: "c1",
		NodeQuota: &apiv1.NodeQuota{
			NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableMemoryMi: 14000},
			NodeCount: 2,
		},
	})
	require.NoError(t, err)
	require.NotNil(t, resp)
	assert.Equal(t, apiv1.Component_AVAILABLE, resp.GetPhase())
	assert.Equal(t, "stable-host-id", resp.GetHostId())
	assert.Equal(t, "ghcr.io/pingcap/cassandra-proxy:legacy-backfill", resp.GetImage())
}

func TestSetCassandraProxy_HandleUpdate_UnavailableRetryMergesRunningSharedPoolStatus(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)

	baseSharedPool := newAvailableSharedPool()
	runningSharedPool := newAvailableSharedPool()
	runningSharedPool.Phase = apiv1.SharedPool_DELETING
	runningSharedPoolBytes, err := protojson.Marshal(runningSharedPool)
	require.NoError(t, err)

	running := executor.NewExecTaskIdStatus()
	running["delete-sp"] = &executor.ExecTaskStatusEx{
		ExecTaskStatus: executor.ExecTaskStatus{
			ExecTask: &executor.ExecutorTask{
				Args: actions.ActionArgs{
					workflow.SharedPoolKey: actions.ActionArgString(string(runningSharedPoolBytes)),
				},
			},
		},
		State: executor.ExecTaskStateTypeRunningOK,
	}

	m.tidbRepo.On("Load", mock.Anything, "c1", false).Return(newTestTiDBCluster("c1"), nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(newCassandraProxyState(apiv1.Component_UNAVAILABLE), nil).Once()
	m.lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(l *entity.Lock) bool { return l.Key == "9999" })).
		Return(func() error { return nil }, nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(newCassandraProxyState(apiv1.Component_UNAVAILABLE), nil).Once()
	m.dagExec.On("QueryRunnings", mock.Anything, dag.PlanCreateCassandraProxy, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil).Once()
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(baseSharedPool, nil).Once()
	m.dagExec.On("QueryRunnings", mock.Anything, dag.PlanDeleteSharedPool, mock.Anything).Return(running, nil).Once()
	m.dagExec.On("QueryRunnings", mock.Anything, mock.MatchedBy(func(plan string) bool {
		return plan != dag.PlanCreateCassandraProxy
	}), mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
	m.scaleManager.On("AnyFullfillingDecision", mock.Anything, "9999").Return(false, nil).Once()

	_, err = s.SetCassandraProxy(context.Background(), &apiv1.SetCassandraProxyRequest{
		ClusterId: "c1",
		NodeQuota: &apiv1.NodeQuota{
			NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableMemoryMi: 14000},
			NodeCount: 2,
		},
	})
	require.Error(t, err)
	assert.Equal(t, codes.FailedPrecondition, status.Code(err))
	assert.Contains(t, status.Convert(err).Message(), "shared pool is being deleted")
}

func TestSetCassandraProxy_HandleUpdate_UnavailableWithRunningDagReturnsCreating(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)
	sharedPool := newAvailableSharedPool()
	sharedPool.ComponentGroups = []*entity.Component{{
		Id:            "c1",
		ComponentType: apiv1.ComponentType_CASSANDRA_PROXY,
		Phase:         apiv1.Component_CREATING,
		Labels: map[string]string{
			cassandraProxyHostIdLabelKey: "stable-host-id",
		},
	}}
	sharedPoolBytes, err := protojson.Marshal(sharedPool)
	require.NoError(t, err)

	running := executor.NewExecTaskIdStatus()
	running["exec-1"] = &executor.ExecTaskStatusEx{
		ExecTaskStatus: executor.ExecTaskStatus{
			ExecTask: &executor.ExecutorTask{
				Args: actions.ActionArgs{workflow.SharedPoolKey: actions.ActionArgString(string(sharedPoolBytes))},
			},
		},
		State: executor.ExecTaskStateTypeRunningOK,
	}

	m.tidbRepo.On("Load", mock.Anything, "c1", false).Return(newTestTiDBCluster("c1"), nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(newCassandraProxyState(apiv1.Component_UNAVAILABLE), nil).Once()
	m.lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(l *entity.Lock) bool { return l.Key == "9999" })).
		Return(func() error { return nil }, nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(newCassandraProxyState(apiv1.Component_UNAVAILABLE), nil).Once()
	m.dagExec.On("QueryRunnings", mock.Anything, dag.PlanCreateCassandraProxy, mock.Anything).Return(running, nil).Once()
	m.stateRepo.On("UpdatePhaseIfCurrent", mock.Anything, "c1", "stable-host-id", []entity.ComponentPhase{apiv1.Component_UNAVAILABLE}, apiv1.Component_CREATING).
		Return(newCassandraProxyState(apiv1.Component_CREATING), true, nil).Once()

	resp, err := s.SetCassandraProxy(context.Background(), &apiv1.SetCassandraProxyRequest{
		ClusterId: "c1",
		NodeQuota: &apiv1.NodeQuota{
			NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableMemoryMi: 14000},
			NodeCount: 2,
		},
	})
	require.NoError(t, err)
	require.NotNil(t, resp)
	assert.Equal(t, "stable-host-id", resp.GetHostId())
	assert.Equal(t, apiv1.Component_CREATING, resp.GetPhase())
}

func TestSetCassandraProxy_HandleUpdate_UnavailableWithRunningDagCASMissReturnsFreshAvailable(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)
	sharedPool := newAvailableSharedPool()
	sharedPool.ComponentGroups = []*entity.Component{{
		Id:            "c1",
		ComponentType: apiv1.ComponentType_CASSANDRA_PROXY,
		Phase:         apiv1.Component_CREATING,
		Labels: map[string]string{
			cassandraProxyHostIdLabelKey: "stable-host-id",
		},
	}}
	sharedPoolBytes, err := protojson.Marshal(sharedPool)
	require.NoError(t, err)

	running := executor.NewExecTaskIdStatus()
	running["exec-1"] = &executor.ExecTaskStatusEx{
		ExecTaskStatus: executor.ExecTaskStatus{
			ExecTask: &executor.ExecutorTask{
				Args: actions.ActionArgs{workflow.SharedPoolKey: actions.ActionArgString(string(sharedPoolBytes))},
			},
		},
		State: executor.ExecTaskStateTypeRunningOK,
	}
	availableState := newCassandraProxyState(apiv1.Component_AVAILABLE)

	m.tidbRepo.On("Load", mock.Anything, "c1", false).Return(newTestTiDBCluster("c1"), nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(newCassandraProxyState(apiv1.Component_UNAVAILABLE), nil).Once()
	m.lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(l *entity.Lock) bool { return l.Key == "9999" })).
		Return(func() error { return nil }, nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(newCassandraProxyState(apiv1.Component_UNAVAILABLE), nil).Once()
	m.dagExec.On("QueryRunnings", mock.Anything, dag.PlanCreateCassandraProxy, mock.Anything).Return(running, nil).Once()
	m.stateRepo.On("UpdatePhaseIfCurrent", mock.Anything, "c1", "stable-host-id", []entity.ComponentPhase{apiv1.Component_UNAVAILABLE}, apiv1.Component_CREATING).
		Return(availableState, false, nil).Once()

	resp, err := s.SetCassandraProxy(context.Background(), &apiv1.SetCassandraProxyRequest{
		ClusterId: "c1",
		NodeQuota: &apiv1.NodeQuota{
			NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableMemoryMi: 14000},
			NodeCount: 2,
		},
	})
	require.NoError(t, err)
	require.NotNil(t, resp)
	assert.Equal(t, apiv1.Component_AVAILABLE, resp.GetPhase())
	assert.Equal(t, "stable-host-id", resp.GetHostId())
}

func TestSetCassandraProxy_HandleUpdate_UnavailableIgnoresStaleRunningDag(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)
	staleSharedPool := newAvailableSharedPool()
	staleSharedPool.ComponentGroups = []*entity.Component{{
		Id:            "c1",
		ComponentType: apiv1.ComponentType_CASSANDRA_PROXY,
		Phase:         apiv1.Component_UNAVAILABLE,
		Labels: map[string]string{
			cassandraProxyHostIdLabelKey: "stale-host-id",
		},
	}}
	staleSharedPoolBytes, err := protojson.Marshal(staleSharedPool)
	require.NoError(t, err)

	running := executor.NewExecTaskIdStatus()
	running["exec-1"] = &executor.ExecTaskStatusEx{
		ExecTaskStatus: executor.ExecTaskStatus{
			ExecTask: &executor.ExecutorTask{
				ExecId: "create-cassandra-proxy-9999-c1-old",
				Args:   actions.ActionArgs{workflow.SharedPoolKey: actions.ActionArgString(string(staleSharedPoolBytes))},
			},
		},
		State: executor.ExecTaskStateTypeRunningOK,
	}
	sharedPool := newAvailableSharedPool()
	sharedPool.ComponentGroups = []*entity.Component{{
		Id:            "c1",
		ComponentType: apiv1.ComponentType_CASSANDRA_PROXY,
		Phase:         apiv1.Component_UNAVAILABLE,
		Labels: map[string]string{
			cassandraProxyHostIdLabelKey: "stale-host-id",
		},
	}}

	m.tidbRepo.On("Load", mock.Anything, "c1", false).Return(newTestTiDBCluster("c1"), nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(newCassandraProxyState(apiv1.Component_UNAVAILABLE), nil).Once()
	m.lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(l *entity.Lock) bool { return l.Key == "9999" })).
		Return(func() error { return nil }, nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(newCassandraProxyState(apiv1.Component_UNAVAILABLE), nil).Once()
	m.dagExec.On("QueryRunnings", mock.Anything, dag.PlanCreateCassandraProxy, mock.Anything).Return(running, nil).Once()
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(sharedPool, nil).Once()
	m.dagExec.On("QueryRunnings", mock.Anything, mock.MatchedBy(func(plan string) bool {
		return plan != dag.PlanCreateCassandraProxy
	}), mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
	m.scaleManager.On("AnyFullfillingDecision", mock.Anything, "9999").Return(false, nil).Once()
	m.stateRepo.On("UpdatePhaseIfCurrent", mock.Anything, "c1", "stable-host-id", []entity.ComponentPhase{apiv1.Component_UNAVAILABLE}, apiv1.Component_CREATING).
		Return(newCassandraProxyState(apiv1.Component_CREATING), true, nil).Once()
	m.spRepo.On("Save", mock.Anything, mock.Anything).Return(nil).Once()

	var submittedExecID string
	var submittedArgs actions.ActionArgs
	m.dagExec.On("Submit", mock.Anything, dag.PlanCreateCassandraProxy, mock.MatchedBy(func(args actions.ActionArgs) bool {
		submittedArgs = args
		return true
	}), mock.MatchedBy(func(execID string) bool {
		submittedExecID = execID
		return execID != "create-cassandra-proxy-9999-c1-old"
	})).Return(nil).Once()

	resp, err := s.SetCassandraProxy(context.Background(), &apiv1.SetCassandraProxyRequest{
		ClusterId: "c1",
		NodeQuota: &apiv1.NodeQuota{
			NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableMemoryMi: 14000},
			NodeCount: 2,
		},
	})
	require.NoError(t, err)
	require.NotNil(t, resp)
	assert.Equal(t, apiv1.Component_CREATING, resp.GetPhase())
	assert.NotEmpty(t, submittedExecID)

	sharedPoolArg, err := submittedArgs.GetString(workflow.SharedPoolKey)
	require.NoError(t, err)
	assert.Contains(t, sharedPoolArg, "stable-host-id")
	assert.NotContains(t, sharedPoolArg, "stale-host-id")
}

func TestSetCassandraProxy_HandleUpdate_UnavailableSpecDiffReturnsFailedPrecondition(t *testing.T) {
	tests := []struct {
		name string
		req  *apiv1.SetCassandraProxyRequest
	}{
		{
			name: "node_spec changed",
			req: &apiv1.SetCassandraProxyRequest{
				ClusterId: "c1",
				NodeQuota: &apiv1.NodeQuota{
					NodeSpec:  &apiv1.NodeSpec{Cpu: 16, MemoryGi: 32, AvailableMemoryMi: 28000},
					NodeCount: 2,
				},
			},
		},
		{
			name: "node_quota count changed",
			req: &apiv1.SetCassandraProxyRequest{
				ClusterId: "c1",
				NodeQuota: &apiv1.NodeQuota{
					NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableMemoryMi: 14000},
					NodeCount: 4,
				},
			},
		},
		{
			name: "image changed",
			req: &apiv1.SetCassandraProxyRequest{
				ClusterId: "c1",
				NodeQuota: &apiv1.NodeQuota{
					NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableMemoryMi: 14000},
					NodeCount: 2,
				},
				Image: cassandraProxyImagePtr("ghcr.io/pingcap/cassandra-proxy:v0.6.0"),
			},
		},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			s, m := newCassandraProxyTestServer(t)
			m.tidbRepo.On("Load", mock.Anything, "c1", false).Return(newTestTiDBCluster("c1"), nil)
			m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil)
			m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(newCassandraProxyState(apiv1.Component_UNAVAILABLE), nil)

			_, err := s.SetCassandraProxy(context.Background(), tc.req)
			require.Error(t, err)
			assert.Equal(t, codes.FailedPrecondition, status.Code(err))
			assert.Contains(t, status.Convert(err).Message(), "cannot change cassandra-proxy desired state while component is UNAVAILABLE")
		})
	}
}

func TestSetCassandraProxy_HandleUpdate_Scale_Success(t *testing.T) {
	tests := []struct {
		name      string
		req       *apiv1.SetCassandraProxyRequest
		wantSpec  string
		wantImage string
	}{
		{
			name: "node_spec_changed",
			req: &apiv1.SetCassandraProxyRequest{
				ClusterId: "c1",
				NodeQuota: &apiv1.NodeQuota{
					NodeSpec:  &apiv1.NodeSpec{Cpu: 16, MemoryGi: 32, AvailableMemoryMi: 28000},
					NodeCount: 2,
				},
			},
			wantSpec:  "16C32G",
			wantImage: "ghcr.io/pingcap/cassandra-proxy:v0.5.2",
		},
		{
			name: "node_spec_and_image_changed",
			req: &apiv1.SetCassandraProxyRequest{
				ClusterId: "c1",
				NodeQuota: &apiv1.NodeQuota{
					NodeSpec:  &apiv1.NodeSpec{Cpu: 16, MemoryGi: 32, AvailableMemoryMi: 28000},
					NodeCount: 2,
				},
				Image: cassandraProxyImagePtr("ghcr.io/pingcap/cassandra-proxy:v0.6.0"),
			},
			wantSpec:  "16C32G",
			wantImage: "ghcr.io/pingcap/cassandra-proxy:v0.6.0",
		},
		{
			name: "image_only_changed",
			req: &apiv1.SetCassandraProxyRequest{
				ClusterId: "c1",
				NodeQuota: &apiv1.NodeQuota{
					NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableMemoryMi: 14000},
					NodeCount: 2,
				},
				Image: cassandraProxyImagePtr("ghcr.io/pingcap/cassandra-proxy:v0.6.1"),
			},
			wantSpec:  "8C16G",
			wantImage: "ghcr.io/pingcap/cassandra-proxy:v0.6.1",
		},
		{
			name: "node_count_changed",
			req: &apiv1.SetCassandraProxyRequest{
				ClusterId: "c1",
				NodeQuota: &apiv1.NodeQuota{
					NodeSpec:     &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableMemoryMi: 14000},
					NodeCount:    3,
					MinNodeCount: 1,
					MaxNodeCount: 5,
				},
			},
			wantSpec:  "8C16G",
			wantImage: "ghcr.io/pingcap/cassandra-proxy:v0.5.2",
		},
		{
			name: "node_quota_min_max_changed",
			req: &apiv1.SetCassandraProxyRequest{
				ClusterId: "c1",
				NodeQuota: &apiv1.NodeQuota{
					NodeSpec:     &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableMemoryMi: 14000},
					NodeCount:    2,
					MinNodeCount: 2,
					MaxNodeCount: 8,
				},
			},
			wantSpec:  "8C16G",
			wantImage: "ghcr.io/pingcap/cassandra-proxy:v0.5.2",
		},
	}

	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			s, m := newCassandraProxyTestServer(t)

			// outer loads (before lock)
			m.tidbRepo.On("Load", mock.Anything, "c1", false).Return(newTestTiDBCluster("c1"), nil)
			m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
			m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(newCassandraProxyState(apiv1.Component_AVAILABLE), nil).Once()

			// inside transaction: lock, sharedPool with component, mergeStatus, freshState reload
			m.lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(l *entity.Lock) bool { return l.Key == "9999" })).
				Return(func() error { return nil }, nil).Once()

			spWithComp := newAvailableSharedPool()
			spWithComp.ComponentGroups = []*entity.Component{{
				Id:            "c1",
				ComponentType: apiv1.ComponentType_CASSANDRA_PROXY,
				Phase:         apiv1.Component_AVAILABLE,
				NodeQuota: &apiv1.NodeQuota{
					NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableMemoryMi: 14000},
					NodeCount: 2,
				},
				Labels: map[string]string{
					cassandraProxyHostIdLabelKey:   "stable-host-id",
					cassandraProxyNodeSpecLabelKey: "8C16G",
				},
			}}
			m.spRepo.On("Load", mock.Anything, "9999", false).Return(spWithComp, nil).Once()
			m.dagExec.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
			m.scaleManager.On("AnyFullfillingDecision", mock.Anything, "9999").Return(false, nil).Once()
			m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(newCassandraProxyState(apiv1.Component_AVAILABLE), nil).Once()

			// save durable state
			var savedState *entity.CassandraProxyState
			m.stateRepo.On("Save", mock.Anything, mock.MatchedBy(func(st *entity.CassandraProxyState) bool {
				savedState = st
				return st.ClusterID == "c1" && st.Phase == entity.ComponentPhaseToCassandraProxyPhase(apiv1.Component_MODIFYING)
			})).Return(nil).Once()

			// save sharedPool
			var savedSP *entity.SharedPool
			m.spRepo.On("Save", mock.Anything, mock.MatchedBy(func(sp *entity.SharedPool) bool {
				savedSP = sp
				return sp.GetId() == "9999" && sp.GetPhase() == apiv1.SharedPool_MODIFYING
			})).Return(nil).Once()

			// submit DAG
			m.dagExec.On("Submit", mock.Anything, dag.PlanCreateCassandraProxy, mock.Anything, mock.MatchedBy(func(id string) bool {
				return id != ""
			})).Return(nil).Once()

			resp, err := s.SetCassandraProxy(context.Background(), tc.req)
			require.NoError(t, err)
			require.NotNil(t, resp)
			assert.Equal(t, apiv1.Component_MODIFYING, resp.GetPhase())
			assert.Equal(t, "stable-host-id", resp.GetHostId())
			assert.Equal(t, "cassandra-c1.tidb9999.svc.cluster.local:9042", resp.GetInternalEndpoint())
			assert.Equal(t, tc.wantImage, resp.GetImage())

			require.NotNil(t, savedState)
			assert.Equal(t, tc.wantSpec, savedState.NodeSpec)
			assert.Equal(t, tc.wantImage, savedState.Image)
			assert.Equal(t, entity.ComponentPhaseToCassandraProxyPhase(apiv1.Component_MODIFYING), savedState.Phase)
			assert.Equal(t, "stable-host-id", savedState.HostID)
			assert.Equal(t, tc.req.GetNodeQuota().GetNodeCount(), savedState.NodeQuotaEntity.GetNodeCount())
			assert.Equal(t, tc.req.GetNodeQuota().GetMinNodeCount(), savedState.NodeQuotaEntity.GetMinNodeCount())
			assert.Equal(t, tc.req.GetNodeQuota().GetMaxNodeCount(), savedState.NodeQuotaEntity.GetMaxNodeCount())

			require.NotNil(t, savedSP)
			var cassComp *entity.Component
			for _, c := range savedSP.GetComponentGroups() {
				if c.GetId() == "c1" && c.GetComponentType() == apiv1.ComponentType_CASSANDRA_PROXY {
					cassComp = c
				}
			}
			require.NotNil(t, cassComp)
			assert.Equal(t, apiv1.Component_MODIFYING, cassComp.GetPhase())
			assert.Equal(t, tc.wantSpec, cassComp.GetLabels()[cassandraProxyNodeSpecLabelKey])
			assert.Equal(t, tc.req.GetNodeQuota().GetNodeCount(), cassComp.GetNodeQuota().GetNodeCount())
			assert.Equal(t, tc.req.GetNodeQuota().GetMinNodeCount(), cassComp.GetNodeQuota().GetMinNodeCount())
			assert.Equal(t, tc.req.GetNodeQuota().GetMaxNodeCount(), cassComp.GetNodeQuota().GetMaxNodeCount())
		})
	}
}

func TestSetCassandraProxy_HandleUpdate_Scale_NoOp(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)
	s.imageConfig = &config.ImageConfig{CassandraProxyImage: "ghcr.io/pingcap/cassandra-proxy:legacy-backfill"}

	legacyState := newCassandraProxyState(apiv1.Component_AVAILABLE)
	legacyState.Image = ""
	m.tidbRepo.On("Load", mock.Anything, "c1", false).Return(newTestTiDBCluster("c1"), nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(legacyState, nil).Once()

	// outer spec check sees diff, enters scale; under lock fresh state matches req (race)
	m.lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(l *entity.Lock) bool { return l.Key == "9999" })).
		Return(func() error { return nil }, nil).Once()
	spWithComp := newAvailableSharedPool()
	spWithComp.ComponentGroups = []*entity.Component{{
		Id:            "c1",
		ComponentType: apiv1.ComponentType_CASSANDRA_PROXY,
		Phase:         apiv1.Component_AVAILABLE,
		Labels: map[string]string{
			cassandraProxyHostIdLabelKey:   "stable-host-id",
			cassandraProxyNodeSpecLabelKey: "8C16G",
		},
	}}
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(spWithComp, nil).Once()
	m.dagExec.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
	m.scaleManager.On("AnyFullfillingDecision", mock.Anything, "9999").Return(false, nil).Once()

	// fresh state reload: spec now matches req (concurrent scale resolved it)
	sameSpecState := newCassandraProxyState(apiv1.Component_AVAILABLE)
	sameSpecState.NodeSpec = "16C32G"
	sameSpecState.NodeQuotaEntity = &apiv1.NodeQuota{
		NodeSpec:  &apiv1.NodeSpec{Cpu: 16, MemoryGi: 32, AvailableMemoryMi: 28000},
		NodeCount: 2,
	}
	sameSpecState.Image = ""
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(sameSpecState, nil).Once()
	// buildCassandraProxySetResponse (no-op path) triggers PLS DNS lookup which needs sharedPool
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	m.dpClient.On("GetPrivateLinkService", mock.Anything, mock.Anything).
		Return(nil, status.Error(codes.NotFound, "not found")).Maybe()

	resp, err := s.SetCassandraProxy(context.Background(), &apiv1.SetCassandraProxyRequest{
		ClusterId: "c1",
		NodeQuota: &apiv1.NodeQuota{
			NodeSpec:  &apiv1.NodeSpec{Cpu: 16, MemoryGi: 32, AvailableMemoryMi: 28000},
			NodeCount: 2,
		},
	})
	require.NoError(t, err)
	require.NotNil(t, resp)
	assert.Equal(t, apiv1.Component_AVAILABLE, resp.GetPhase())
	assert.Equal(t, "stable-host-id", resp.GetHostId())
	assert.Equal(t, "ghcr.io/pingcap/cassandra-proxy:legacy-backfill", resp.GetImage())
}

func TestSetCassandraProxy_HandleUpdate_InFlightDiffReturnsFailedPrecondition(t *testing.T) {
	tests := []struct {
		name  string
		phase apiv1.Component_Phase
		req   *apiv1.SetCassandraProxyRequest
	}{
		{
			name:  "creating_spec_diff",
			phase: apiv1.Component_CREATING,
			req: &apiv1.SetCassandraProxyRequest{
				ClusterId: "c1",
				NodeQuota: &apiv1.NodeQuota{
					NodeSpec:  &apiv1.NodeSpec{Cpu: 16, MemoryGi: 32, AvailableMemoryMi: 28000},
					NodeCount: 2,
				},
			},
		},
		{
			name:  "modifying_spec_diff",
			phase: apiv1.Component_MODIFYING,
			req: &apiv1.SetCassandraProxyRequest{
				ClusterId: "c1",
				NodeQuota: &apiv1.NodeQuota{
					NodeSpec:  &apiv1.NodeSpec{Cpu: 16, MemoryGi: 32, AvailableMemoryMi: 28000},
					NodeCount: 2,
				},
			},
		},
		{
			name:  "creating_image_diff",
			phase: apiv1.Component_CREATING,
			req: &apiv1.SetCassandraProxyRequest{
				ClusterId: "c1",
				NodeQuota: &apiv1.NodeQuota{
					NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableMemoryMi: 14000},
					NodeCount: 2,
				},
				Image: cassandraProxyImagePtr("ghcr.io/pingcap/cassandra-proxy:v0.6.0"),
			},
		},
		{
			name:  "modifying_image_diff",
			phase: apiv1.Component_MODIFYING,
			req: &apiv1.SetCassandraProxyRequest{
				ClusterId: "c1",
				NodeQuota: &apiv1.NodeQuota{
					NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableMemoryMi: 14000},
					NodeCount: 2,
				},
				Image: cassandraProxyImagePtr("ghcr.io/pingcap/cassandra-proxy:v0.6.0"),
			},
		},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			s, m := newCassandraProxyTestServer(t)
			m.tidbRepo.On("Load", mock.Anything, "c1", false).Return(newTestTiDBCluster("c1"), nil)
			m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil)
			m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(newCassandraProxyState(tc.phase), nil)

			_, err := s.SetCassandraProxy(context.Background(), tc.req)
			require.Error(t, err)
			assert.Equal(t, codes.FailedPrecondition, status.Code(err))
		})
	}
}

func TestSetCassandraProxy_HandleUpdate_Scale_TOCTOURaceUnavailableReturnsFailedPrecondition(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)

	// outer load sees AVAILABLE; under lock state raced to UNAVAILABLE
	m.tidbRepo.On("Load", mock.Anything, "c1", false).Return(newTestTiDBCluster("c1"), nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(newCassandraProxyState(apiv1.Component_AVAILABLE), nil).Once()

	m.lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(l *entity.Lock) bool { return l.Key == "9999" })).
		Return(func() error { return nil }, nil).Once()
	spWithComp := newAvailableSharedPool()
	spWithComp.ComponentGroups = []*entity.Component{{
		Id:            "c1",
		ComponentType: apiv1.ComponentType_CASSANDRA_PROXY,
		Phase:         apiv1.Component_AVAILABLE,
		Labels: map[string]string{
			cassandraProxyHostIdLabelKey:   "stable-host-id",
			cassandraProxyNodeSpecLabelKey: "8C16G",
		},
	}}
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(spWithComp, nil).Once()
	m.dagExec.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
	m.scaleManager.On("AnyFullfillingDecision", mock.Anything, "9999").Return(false, nil).Once()
	// fresh reload under lock shows UNAVAILABLE (race)
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(newCassandraProxyState(apiv1.Component_UNAVAILABLE), nil).Once()

	_, err := s.SetCassandraProxy(context.Background(), &apiv1.SetCassandraProxyRequest{
		ClusterId: "c1",
		NodeQuota: &apiv1.NodeQuota{
			NodeSpec:  &apiv1.NodeSpec{Cpu: 16, MemoryGi: 32, AvailableMemoryMi: 28000},
			NodeCount: 2,
		},
	})
	require.Error(t, err)
	assert.Equal(t, codes.FailedPrecondition, status.Code(err))
	assert.Contains(t, status.Convert(err).Message(), "cannot change cassandra-proxy desired state while component is UNAVAILABLE")
}

func TestSetCassandraProxy_HandleUpdate_Scale_StateOnlyComponentAppended(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)
	s.imageConfig = &config.ImageConfig{CassandraProxyImage: "ghcr.io/pingcap/cassandra-proxy:legacy-backfill"}
	legacyState := newCassandraProxyState(apiv1.Component_AVAILABLE)
	legacyState.Image = ""
	freshLegacyState := newCassandraProxyState(apiv1.Component_AVAILABLE)
	freshLegacyState.Image = ""

	// sharedPool snapshot does not contain the CASSANDRA_PROXY component (state-only)
	m.tidbRepo.On("Load", mock.Anything, "c1", false).Return(newTestTiDBCluster("c1"), nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(legacyState, nil).Once()

	m.lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(l *entity.Lock) bool { return l.Key == "9999" })).
		Return(func() error { return nil }, nil).Once()
	// sharedPool has no CASSANDRA_PROXY component in ComponentGroups
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	m.dagExec.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
	m.scaleManager.On("AnyFullfillingDecision", mock.Anything, "9999").Return(false, nil).Once()
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(freshLegacyState, nil).Once()

	var savedSP *entity.SharedPool
	var savedState *entity.CassandraProxyState
	m.stateRepo.On("Save", mock.Anything, mock.MatchedBy(func(st *entity.CassandraProxyState) bool {
		savedState = st
		return true
	})).Return(nil).Once()
	m.spRepo.On("Save", mock.Anything, mock.MatchedBy(func(sp *entity.SharedPool) bool {
		savedSP = sp
		return sp.GetId() == "9999"
	})).Return(nil).Once()
	m.dagExec.On("Submit", mock.Anything, dag.PlanCreateCassandraProxy, mock.Anything, mock.Anything).Return(nil).Once()

	resp, err := s.SetCassandraProxy(context.Background(), &apiv1.SetCassandraProxyRequest{
		ClusterId: "c1",
		NodeQuota: &apiv1.NodeQuota{
			NodeSpec:  &apiv1.NodeSpec{Cpu: 16, MemoryGi: 32, AvailableMemoryMi: 28000},
			NodeCount: 2,
		},
	})
	require.NoError(t, err)
	assert.Equal(t, apiv1.Component_MODIFYING, resp.GetPhase())
	assert.Equal(t, "ghcr.io/pingcap/cassandra-proxy:legacy-backfill", resp.GetImage())
	require.NotNil(t, savedState)
	assert.Equal(t, "ghcr.io/pingcap/cassandra-proxy:legacy-backfill", savedState.Image)

	// component was appended (not replaced) since it was missing from snapshot
	require.NotNil(t, savedSP)
	var cassComp *entity.Component
	for _, c := range savedSP.GetComponentGroups() {
		if c.GetId() == "c1" && c.GetComponentType() == apiv1.ComponentType_CASSANDRA_PROXY {
			cassComp = c
		}
	}
	require.NotNil(t, cassComp, "CASSANDRA_PROXY component should be appended to snapshot")
	assert.Equal(t, apiv1.Component_MODIFYING, cassComp.GetPhase())
	assert.Equal(t, "16C32G", cassComp.GetLabels()[cassandraProxyNodeSpecLabelKey])
}

func TestDeleteTiDBCluster_CassandraProxyDeleteStart(t *testing.T) {
	tidbClusterRepo := repomocks.NewTiDBClusterRepo(t)
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	tidbGroupRepo := repomocks.NewTiDBGroupRepo(t)
	cassandraStateRepo := repomocks.NewCassandraProxyStateRepo(t)
	dagExecutor := executormocks.NewRemoteExecutor(t)

	tidbCluster := &entity.TiDBCluster{Id: "c1", SharedPoolId: "9999", Phase: apiv1.Cluster_AVAILABLE}
	sharedPool := newAvailableSharedPool()
	sharedPool.ComponentGroups = []*entity.Component{{
		Id:            "c1",
		ComponentType: apiv1.ComponentType_CASSANDRA_PROXY,
		Phase:         apiv1.Component_CREATING,
		Labels: map[string]string{
			cassandraProxyHostIdLabelKey: "stale-host-id",
		},
	}}
	tidbClusterRepo.On("Load", mock.Anything, "c1", false).Return(tidbCluster, nil).Once()
	tidbGroupRepo.On("List", mock.Anything, mock.Anything, mock.Anything, []string{"c1"}, mock.Anything, false).Return([]*entity.TiDBGroup{}, nil).Once()
	dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(func(_ context.Context, planName string, filter executor.MatchFilter) executor.ExecTaskIdStatus {
		switch planName {
		case dag.PlanUpdateTiDBCluster:
			if len(filter[workflow.TiDBClusterIdKey]) == 0 {
				return executor.NewExecTaskIdStatus()
			}
			assert.Equal(t, []string{"c1"}, filter[workflow.TiDBClusterIdKey])
			assert.Empty(t, filter[workflow.ComponentIdsKey])
			running := executor.NewExecTaskIdStatus()
			running["update-cluster-exec"] = &executor.ExecTaskStatusEx{
				ExecTaskStatus: executor.ExecTaskStatus{ExecTask: &executor.ExecutorTask{ExecId: "update-cluster-exec"}},
			}
			return running
		case dag.PlanCreateCassandraProxy:
			assert.Equal(t, []string{"c1"}, filter[workflow.TiDBClusterIdKey])
			assert.Equal(t, []string{"c1"}, filter[workflow.ComponentIdsKey])
			running := executor.NewExecTaskIdStatus()
			running["create-cassandra-exec"] = &executor.ExecTaskStatusEx{
				ExecTaskStatus: executor.ExecTaskStatus{ExecTask: &executor.ExecutorTask{ExecId: "create-cassandra-exec"}},
			}
			return running
		default:
			return executor.NewExecTaskIdStatus()
		}
	}, nil)
	dagExecutor.On("Abort", mock.Anything, "update-cluster-exec").Return(nil).Once()
	dagExecutor.On("Abort", mock.Anything, "create-cassandra-exec").Return(nil).Once()
	scaleManager := bractionmocks.NewScaleManager(t)
	scaleManager.On("AnyFullfillingDecision", mock.Anything, "9999").Return(false, nil).Once()
	sharedPoolRepo.On("Load", mock.Anything, "9999", false).Return(sharedPool, nil).Once()
	cassandraStateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(newCassandraProxyState(apiv1.Component_AVAILABLE), nil).Once()
	cassandraStateRepo.On("UpdatePhase", mock.Anything, "c1", "stable-host-id", apiv1.Component_DELETING).Return(nil).Once()
	tidbClusterRepo.On("Delete", mock.Anything, "c1").Return(nil).Once()

	var submittedArgs actions.ActionArgs
	dagExecutor.On("Submit", mock.Anything, dag.PlanDeleteTiDBCluster, mock.MatchedBy(func(args actions.ActionArgs) bool {
		submittedArgs = args
		return true
	}), mock.Anything).Return(nil).Once()

	server := &serverImpl{
		tidbClusterRepo:         tidbClusterRepo,
		sharedPoolRepo:          sharedPoolRepo,
		tidbGroupRepo:           tidbGroupRepo,
		cassandraProxyStateRepo: cassandraStateRepo,
		dagExecutor:             dagExecutor,
		scaleManager:            scaleManager,
		transactioner:           &mockTransactioner{},
	}

	result, err := server.DeleteTiDBCluster(context.Background(), "c1")
	require.NoError(t, err)
	require.NotNil(t, result)

	componentIds, err := submittedArgs.GetString(workflow.ComponentIdsKey)
	require.NoError(t, err)
	assert.Equal(t, "c1,c1", componentIds)
	componentTypes, err := submittedArgs.GetString(workflow.ComponentTypesKey)
	require.NoError(t, err)
	assert.Equal(t, "tiproxy,cassandra-proxy", componentTypes)
	sharedPoolStr, err := submittedArgs.GetString(workflow.SharedPoolKey)
	require.NoError(t, err)
	assert.Contains(t, sharedPoolStr, "CASSANDRA_PROXY")
	assert.Contains(t, sharedPoolStr, "DELETING")
	assert.Contains(t, sharedPoolStr, "stable-host-id")
	assert.NotContains(t, sharedPoolStr, "stale-host-id")
}

func TestDeleteTiDBCluster_NoCassandraProxyStateSubmitsOnlyTiProxy(t *testing.T) {
	tidbClusterRepo := repomocks.NewTiDBClusterRepo(t)
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	tidbGroupRepo := repomocks.NewTiDBGroupRepo(t)
	cassandraStateRepo := repomocks.NewCassandraProxyStateRepo(t)
	dagExecutor := executormocks.NewRemoteExecutor(t)

	tidbCluster := &entity.TiDBCluster{Id: "c1", SharedPoolId: "9999", Phase: apiv1.Cluster_AVAILABLE}
	sharedPool := newAvailableSharedPool()
	sharedPool.ComponentGroups = []*entity.Component{{
		Id:            "c1",
		ComponentType: apiv1.ComponentType_TIPROXY,
		Phase:         apiv1.Component_AVAILABLE,
		NodeQuota:     &entity.NodeQuota{NodeCount: 1},
	}}
	tidbClusterRepo.On("Load", mock.Anything, "c1", false).Return(tidbCluster, nil).Once()
	tidbGroupRepo.On("List", mock.Anything, mock.Anything, mock.Anything, []string{"c1"}, mock.Anything, false).Return([]*entity.TiDBGroup{}, nil).Once()
	dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
	scaleManager := bractionmocks.NewScaleManager(t)
	scaleManager.On("AnyFullfillingDecision", mock.Anything, "9999").Return(false, nil).Once()
	sharedPoolRepo.On("Load", mock.Anything, "9999", false).Return(sharedPool, nil).Once()
	cassandraStateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(nil, status.Error(codes.NotFound, "not found")).Once()
	tidbClusterRepo.On("Delete", mock.Anything, "c1").Return(nil).Once()

	var submittedArgs actions.ActionArgs
	dagExecutor.On("Submit", mock.Anything, dag.PlanDeleteTiDBCluster, mock.MatchedBy(func(args actions.ActionArgs) bool {
		submittedArgs = args
		return true
	}), mock.Anything).Return(nil).Once()

	server := &serverImpl{
		tidbClusterRepo:         tidbClusterRepo,
		sharedPoolRepo:          sharedPoolRepo,
		tidbGroupRepo:           tidbGroupRepo,
		cassandraProxyStateRepo: cassandraStateRepo,
		dagExecutor:             dagExecutor,
		scaleManager:            scaleManager,
		transactioner:           &mockTransactioner{},
	}

	result, err := server.DeleteTiDBCluster(context.Background(), "c1")
	require.NoError(t, err)
	require.NotNil(t, result)

	componentIds, err := submittedArgs.GetString(workflow.ComponentIdsKey)
	require.NoError(t, err)
	assert.Equal(t, "c1", componentIds)
	componentTypes, err := submittedArgs.GetString(workflow.ComponentTypesKey)
	require.NoError(t, err)
	assert.Equal(t, "tiproxy", componentTypes)
	sharedPoolStr, err := submittedArgs.GetString(workflow.SharedPoolKey)
	require.NoError(t, err)
	assert.NotContains(t, sharedPoolStr, "CASSANDRA_PROXY")
}

func TestGetCassandraProxy_NotFound(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)
	m.tidbRepo.On("Load", mock.Anything, "c1", true).Return(newTestTiDBCluster("c1"), nil)
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(nil, status.Error(codes.NotFound, "not found"))

	_, err := s.GetCassandraProxy(context.Background(), &apiv1.GetCassandraProxyRequest{ClusterId: "c1"})
	require.Error(t, err)
	assert.Equal(t, codes.NotFound, status.Code(err))
}

func TestGetCassandraProxy_HappyPath(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)
	m.tidbRepo.On("Load", mock.Anything, "c1", true).Return(newTestTiDBCluster("c1"), nil)
	state := newCassandraProxyState(apiv1.Component_AVAILABLE)
	state.HostID = "host-uuid"
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(state, nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil)
	m.dpClient.On("GetPrivateLinkService", mock.Anything, mock.MatchedBy(func(req *dplanemgrv1.GetPrivateLinkServiceRequest) bool {
		return req.GetDataPlaneId() == "dp-test" && req.GetId() == "cassandra-private-c1"
	})).Return(&dplanemgrv1.PrivateLinkService{ServiceDnsName: "privatelink-cassandra.identity.dev.tidbcloud.com"}, nil)

	resp, err := s.GetCassandraProxy(context.Background(), &apiv1.GetCassandraProxyRequest{ClusterId: "c1"})
	require.NoError(t, err)
	assert.Equal(t, "host-uuid", resp.GetHostId())
	assert.Equal(t, apiv1.Component_AVAILABLE, resp.GetPhase())
	assert.Equal(t, "cassandra-c1.tidb9999.svc.cluster.local:9042", resp.GetInternalEndpoint())
	assert.Equal(t, "privatelink-cassandra.identity.dev.tidbcloud.com", resp.GetPrivatelinkDnsName())
	require.NotNil(t, resp.GetNodeQuota())
	assert.Equal(t, uint32(8), resp.GetNodeQuota().GetNodeSpec().GetCpu())
	assert.Equal(t, uint32(16), resp.GetNodeQuota().GetNodeSpec().GetMemoryGi())
	assert.Equal(t, uint32(2), resp.GetNodeQuota().GetNodeCount())
	assert.Equal(t, "ghcr.io/pingcap/cassandra-proxy:v0.5.2", resp.GetImage())
}

func TestGetCassandraProxy_LegacyImageReturnsFallback(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)
	s.imageConfig = &config.ImageConfig{CassandraProxyImage: "ghcr.io/pingcap/cassandra-proxy:legacy-backfill"}

	m.tidbRepo.On("Load", mock.Anything, "c1", true).Return(newTestTiDBCluster("c1"), nil)
	state := newCassandraProxyState(apiv1.Component_UNAVAILABLE)
	state.Image = ""
	state.PrivatelinkDNSName = "persisted.example.com"
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(state, nil)

	resp, err := s.GetCassandraProxy(context.Background(), &apiv1.GetCassandraProxyRequest{ClusterId: "c1"})
	require.NoError(t, err)
	require.NotNil(t, resp)
	assert.Equal(t, apiv1.Component_UNAVAILABLE, resp.GetPhase())
	assert.Equal(t, "ghcr.io/pingcap/cassandra-proxy:legacy-backfill", resp.GetImage())
}

func TestGetCassandraProxy_DeletedClusterReturnsDeletingState(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)
	m.tidbRepo.On("Load", mock.Anything, "c1", true).Return(newTestTiDBCluster("c1"), nil)
	state := newCassandraProxyState(apiv1.Component_DELETING)
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(state, nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil)
	m.dpClient.On("GetPrivateLinkService", mock.Anything, mock.Anything).
		Return(nil, status.Error(codes.NotFound, "not found"))

	resp, err := s.GetCassandraProxy(context.Background(), &apiv1.GetCassandraProxyRequest{ClusterId: "c1"})
	require.NoError(t, err)
	assert.Equal(t, "stable-host-id", resp.GetHostId())
	assert.Equal(t, apiv1.Component_DELETING, resp.GetPhase())
	assert.Equal(t, "cassandra-c1.tidb9999.svc.cluster.local:9042", resp.GetInternalEndpoint())
	assert.Empty(t, resp.GetPrivatelinkDnsName())
}

func TestGetCassandraProxy_PLSReadThroughErrorReturnsPersistedDNS(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)
	m.tidbRepo.On("Load", mock.Anything, "c1", true).Return(newTestTiDBCluster("c1"), nil)
	state := newCassandraProxyState(apiv1.Component_AVAILABLE)
	state.PrivatelinkDNSName = "persisted.example.com"
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(state, nil)
	m.spRepo.On("Load", mock.MatchedBy(contextHasDeadline), "9999", false).Return(newAvailableSharedPool(), nil)
	m.dpClient.On("GetPrivateLinkService", mock.MatchedBy(contextHasDeadline), mock.Anything).
		Return(nil, status.Error(codes.Unavailable, "transport"))

	resp, err := s.GetCassandraProxy(context.Background(), &apiv1.GetCassandraProxyRequest{ClusterId: "c1"})
	require.NoError(t, err)
	assert.Equal(t, "persisted.example.com", resp.GetPrivatelinkDnsName())
	assert.Nil(t, resp.GetPrivateLinkService())
}

func TestGetCassandraProxy_PersistedDNSInUnavailableStateSkipsPLSReadThrough(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)
	m.tidbRepo.On("Load", mock.Anything, "c1", true).Return(newTestTiDBCluster("c1"), nil)
	state := newCassandraProxyState(apiv1.Component_UNAVAILABLE)
	state.PrivatelinkDNSName = "persisted.example.com"
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(state, nil)

	resp, err := s.GetCassandraProxy(context.Background(), &apiv1.GetCassandraProxyRequest{ClusterId: "c1"})
	require.NoError(t, err)
	assert.Equal(t, "persisted.example.com", resp.GetPrivatelinkDnsName())
	assert.Equal(t, apiv1.Component_UNAVAILABLE, resp.GetPhase())
	assert.Nil(t, resp.GetPrivateLinkService())
	require.NotNil(t, resp.GetNodeQuota())
	assert.Equal(t, uint32(8), resp.GetNodeQuota().GetNodeSpec().GetCpu())
	assert.Equal(t, uint32(16), resp.GetNodeQuota().GetNodeSpec().GetMemoryGi())
	assert.Equal(t, uint32(2), resp.GetNodeQuota().GetNodeCount())
}

func TestGetCassandraProxy_EmptyPersistedDNSFallsBackOnPLSReadThroughError(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)
	m.tidbRepo.On("Load", mock.Anything, "c1", true).Return(newTestTiDBCluster("c1"), nil)
	state := newCassandraProxyState(apiv1.Component_AVAILABLE)
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(state, nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil)
	m.dpClient.On("GetPrivateLinkService", mock.Anything, mock.Anything).
		Return(nil, status.Error(codes.Unavailable, "transport"))

	resp, err := s.GetCassandraProxy(context.Background(), &apiv1.GetCassandraProxyRequest{ClusterId: "c1"})
	require.NoError(t, err)
	assert.Empty(t, resp.GetPrivatelinkDnsName())
}

func TestSetCassandraProxy_CapabilityMissing(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)
	clusterId := "c1"
	m.tidbRepo.On("Load", mock.Anything, clusterId, false).Return(newTestTiDBCluster(clusterId), nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil).Once()
	dpClient := dpmgrmocks.NewClient(t)
	s.dataplaneClient = dpClient
	dpClient.On("GetDataPlane", mock.Anything, mock.MatchedBy(func(req *dplanemgrv1.GetDataPlaneRequest) bool {
		return req.GetDataPlaneId() == "dp-test"
	})).Return(&dplanemgrv1.DataPlane{Capabilities: []string{"other.capability"}}, nil)

	_, err := s.SetCassandraProxy(context.Background(), &apiv1.SetCassandraProxyRequest{
		ClusterId: clusterId,
		NodeQuota: &apiv1.NodeQuota{
			NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableMemoryMi: 14000},
			NodeCount: 2,
		},
	})
	require.Error(t, err)
	assert.Equal(t, codes.FailedPrecondition, status.Code(err))
}

func TestSetCassandraProxy_MissingGatewayIdentity(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)
	clusterId := "c1"
	tidbCluster := newTestTiDBCluster(clusterId)
	tidbCluster.GatewayIdentity = ""
	m.tidbRepo.On("Load", mock.Anything, clusterId, false).Return(tidbCluster, nil)

	_, err := s.SetCassandraProxy(context.Background(), &apiv1.SetCassandraProxyRequest{
		ClusterId: clusterId,
		NodeQuota: &apiv1.NodeQuota{
			NodeSpec:  &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableMemoryMi: 14000},
			NodeCount: 2,
		},
	})
	require.Error(t, err)
	assert.Equal(t, codes.FailedPrecondition, status.Code(err))
}

func TestGetCassandraProxy_PLSActivePopulatesServiceInfo(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)
	m.tidbRepo.On("Load", mock.Anything, "c1", true).Return(newTestTiDBCluster("c1"), nil)
	state := newCassandraProxyState(apiv1.Component_AVAILABLE)
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(state, nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil)

	activePLS := &dplanemgrv1.PrivateLinkService{
		ServiceDnsName: "privatelink-cassandra.identity.dev.tidbcloud.com",
		Status:         dplanemgrv1.PrivateLinkService_ACTIVE,
		Info: &dplanemgrv1.PrivateLinkService_Aws_{
			Aws: &dplanemgrv1.PrivateLinkService_Aws{
				ServiceName: "com.amazonaws.vpce.us-east-1.vpce-svc-0abc123",
				AzIds:       []string{"use1-az1", "use1-az2"},
			},
		},
	}
	m.dpClient.On("GetPrivateLinkService", mock.Anything, mock.MatchedBy(func(req *dplanemgrv1.GetPrivateLinkServiceRequest) bool {
		return req.GetId() == "cassandra-private-c1"
	})).Return(activePLS, nil)

	resp, err := s.GetCassandraProxy(context.Background(), &apiv1.GetCassandraProxyRequest{ClusterId: "c1"})
	require.NoError(t, err)
	assert.Equal(t, "privatelink-cassandra.identity.dev.tidbcloud.com", resp.GetPrivatelinkDnsName())
	require.NotNil(t, resp.GetPrivateLinkService())
	assert.Equal(t, "com.amazonaws.vpce.us-east-1.vpce-svc-0abc123", resp.GetPrivateLinkService().GetServiceName())
	assert.Equal(t, []string{"use1-az1", "use1-az2"}, resp.GetPrivateLinkService().GetAvailabilityZoneIds())
	assert.Equal(t, apiv1.PrivateLinkService_ACTIVE, resp.GetPrivateLinkService().GetPhase())
}

func TestGetCassandraProxy_PLSCreatingReturnsPhaseNoServiceName(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)
	m.tidbRepo.On("Load", mock.Anything, "c1", true).Return(newTestTiDBCluster("c1"), nil)
	state := newCassandraProxyState(apiv1.Component_AVAILABLE)
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(state, nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil)
	m.dpClient.On("GetPrivateLinkService", mock.Anything, mock.Anything).Return(&dplanemgrv1.PrivateLinkService{
		Status: dplanemgrv1.PrivateLinkService_CREATING,
	}, nil)

	resp, err := s.GetCassandraProxy(context.Background(), &apiv1.GetCassandraProxyRequest{ClusterId: "c1"})
	require.NoError(t, err)
	require.NotNil(t, resp.GetPrivateLinkService())
	assert.Empty(t, resp.GetPrivateLinkService().GetServiceName())
	assert.Equal(t, apiv1.PrivateLinkService_CREATING, resp.GetPrivateLinkService().GetPhase())
}

func TestGetCassandraProxy_PLSUnspecifiedOmitsServiceInfo(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)
	m.tidbRepo.On("Load", mock.Anything, "c1", true).Return(newTestTiDBCluster("c1"), nil)
	state := newCassandraProxyState(apiv1.Component_AVAILABLE)
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(state, nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(newAvailableSharedPool(), nil)
	m.dpClient.On("GetPrivateLinkService", mock.Anything, mock.Anything).Return(&dplanemgrv1.PrivateLinkService{
		Status: dplanemgrv1.PrivateLinkService_STATE_UNSPECIFIED,
	}, nil)

	resp, err := s.GetCassandraProxy(context.Background(), &apiv1.GetCassandraProxyRequest{ClusterId: "c1"})
	require.NoError(t, err)
	assert.Nil(t, resp.GetPrivateLinkService())
}

func TestGetCassandraProxy_DPMUnavailableReturnsEmptyServiceInfo(t *testing.T) {
	s, m := newCassandraProxyTestServer(t)
	m.tidbRepo.On("Load", mock.Anything, "c1", true).Return(newTestTiDBCluster("c1"), nil)
	state := newCassandraProxyState(apiv1.Component_AVAILABLE)
	state.HostID = "host-uuid"
	m.stateRepo.On("LoadByClusterID", mock.Anything, "c1", false).Return(state, nil)
	m.spRepo.On("Load", mock.Anything, "9999", false).Return(nil, status.Error(codes.Unavailable, "transport"))

	resp, err := s.GetCassandraProxy(context.Background(), &apiv1.GetCassandraProxyRequest{ClusterId: "c1"})
	require.NoError(t, err)
	assert.Equal(t, "host-uuid", resp.GetHostId())
	assert.Equal(t, apiv1.Component_AVAILABLE, resp.GetPhase())
	assert.Nil(t, resp.GetPrivateLinkService())
}

func contextHasDeadline(ctx context.Context) bool {
	deadline, ok := ctx.Deadline()
	return ok && time.Until(deadline) <= cassandraProxyPLSReadThroughTimeout
}

func TestNodeQuotaEqual(t *testing.T) {
	a := &apiv1.NodeQuota{
		NodeSpec:     &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16},
		NodeCount:    2,
		MinNodeCount: 2,
		MaxNodeCount: 4,
	}
	b := &apiv1.NodeQuota{
		NodeSpec:     &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16},
		NodeCount:    2,
		MinNodeCount: 2,
		MaxNodeCount: 4,
	}
	assert.True(t, nodeQuotaEqual(a, b))

	c := &apiv1.NodeQuota{
		NodeSpec:     &apiv1.NodeSpec{Cpu: 16, MemoryGi: 16},
		NodeCount:    2,
		MinNodeCount: 2,
		MaxNodeCount: 4,
	}
	assert.False(t, nodeQuotaEqual(a, c))

	d := &apiv1.NodeQuota{
		NodeSpec:     &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableMemoryMi: 14000},
		NodeCount:    2,
		MinNodeCount: 2,
		MaxNodeCount: 4,
	}
	assert.True(t, nodeQuotaEqual(a, d))

	d.NodeCount = 3
	assert.False(t, nodeQuotaEqual(a, d))
	assert.False(t, nodeQuotaEqual(nil, a))
	assert.True(t, nodeQuotaEqual(nil, nil))
}