package service

import (
	"context"
	"testing"

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/facade/mock_facade"
	"github.com/tidbcloud/serverless-service/internal/domain/repo/mock_repo"
	"github.com/tidbcloud/serverless-service/internal/domain/vo"
	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	regionalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/regional/serverless/v1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/mock/gomock"
)

func TestRegisterMSKPrivateLinkConnectionSharedByKubernetes(t *testing.T) {
	ctrl := gomock.NewController(t)
	plcRepo := mock_repo.NewMockPrivateLinkConnectionRepo(ctrl)
	plcLinkRepo := mock_repo.NewMockPrivateLinkConnectionLinkRepo(ctrl)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	svc := &privateLinkConnectionService{
		mgmtSvc:     mgmtSvc,
		plcRepo:     plcRepo,
		plcLinkRepo: plcLinkRepo,
	}

	ctx := context.Background()
	plc := newMSKPrivateLinkConnection()
	plcRepo.EXPECT().
		GetById(gomock.Any(), "200", plc.ID).
		Return(nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "not found"))
	plcRepo.EXPECT().GetByID(gomock.Any(), plc.ID).Return(plc, nil)
	mgmtSvc.EXPECT().
		GetCluster(gomock.Any(), uint64(200), tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC).
		Return(newSharedServerlessCluster(), nil)
	plcLinkRepo.EXPECT().
		Create(gomock.Any(), gomock.AssignableToTypeOf(&entity.PrivateLinkConnectionLink{})).
		DoAndReturn(func(_ context.Context, link *entity.PrivateLinkConnectionLink) (*entity.PrivateLinkConnectionLink, bizerr.BizError) {
			require.Equal(t, "200", link.ClusterID)
			require.Equal(t, plc.ID, link.PrivateLinkConnectionID)
			require.Equal(t, "cf-1", link.BizID)
			return link, nil
		})

	endpoints, err := svc.RegisterPrivateLinkConnection(ctx, "200", plc.ID, "cf-1", false)

	require.NoError(t, err)
	require.Equal(t, []string{"b-1.example.com:9098"}, endpoints)
}

func TestListPrivateLinkConnectionsFallsBackToClusterScopeWithoutKubernetes(t *testing.T) {
	ctrl := gomock.NewController(t)
	plcRepo := mock_repo.NewMockPrivateLinkConnectionRepo(ctrl)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	svc := &privateLinkConnectionService{
		mgmtSvc: mgmtSvc,
		plcRepo: plcRepo,
	}

	cluster := newSharedServerlessCluster()
	cluster.GetServerlessCluster().GetCluster().Kubernetes = ""
	mgmtSvc.EXPECT().
		GetCluster(gomock.Any(), uint64(200), tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC).
		Return(cluster, nil)
	assertClusterScopedFilter := func(filter *entity.ListPrivateLinkConnectionFilter) {
		require.Equal(t, "200", filter.ClusterID)
		require.Empty(t, filter.TenantID)
		require.Empty(t, filter.Region)
		require.Empty(t, filter.Kubernetes)
		require.False(t, filter.IncludeSharedMSK)
	}
	plcRepo.EXPECT().
		ListByFilter(gomock.Any(), gomock.AssignableToTypeOf(&entity.ListPrivateLinkConnectionFilter{})).
		DoAndReturn(func(_ context.Context, filter *entity.ListPrivateLinkConnectionFilter) ([]*entity.PrivateLinkConnection, bizerr.BizError) {
			assertClusterScopedFilter(filter)
			return []*entity.PrivateLinkConnection{{ID: "plc-1"}}, nil
		})
	plcRepo.EXPECT().
		CountByFilter(gomock.Any(), gomock.AssignableToTypeOf(&entity.ListPrivateLinkConnectionFilter{})).
		DoAndReturn(func(_ context.Context, filter *entity.ListPrivateLinkConnectionFilter) (int64, bizerr.BizError) {
			assertClusterScopedFilter(filter)
			return 1, nil
		})

	plcs, total, err := svc.ListPrivateLinkConnections(context.Background(), &vo.ListPrivateLinkConnectionOptions{
		ClusterID: "200",
		Limit:     10,
	})

	require.NoError(t, err)
	require.Len(t, plcs, 1)
	require.Equal(t, uint32(1), total)
}

func TestDeleteMSKPrivateLinkConnectionChecksLinksAcrossClusters(t *testing.T) {
	ctrl := gomock.NewController(t)
	plcRepo := mock_repo.NewMockPrivateLinkConnectionRepo(ctrl)
	plcLinkRepo := mock_repo.NewMockPrivateLinkConnectionLinkRepo(ctrl)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	svc := &privateLinkConnectionService{
		mgmtSvc:             mgmtSvc,
		plcRepo:             plcRepo,
		plcLinkRepo:         plcLinkRepo,
		plcAttachDomainRepo: mock_repo.NewMockPLCAttachDomainRepo(ctrl),
		transProvider:       fakeTransProvider{},
	}

	ctx := context.Background()
	plc := newMSKPrivateLinkConnection()
	plcRepo.EXPECT().
		GetById(gomock.Any(), "200", plc.ID).
		Return(nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "not found"))
	plcRepo.EXPECT().GetByID(gomock.Any(), plc.ID).Return(plc, nil)
	mgmtSvc.EXPECT().
		GetCluster(gomock.Any(), uint64(200), tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC).
		Return(newSharedServerlessCluster(), nil)
	plcLinkRepo.EXPECT().
		ListByPrivateLinkConnectionID(gomock.Any(), "", plc.ID).
		Return([]*entity.PrivateLinkConnectionLink{
			{ClusterID: "100", PrivateLinkConnectionID: plc.ID, BizID: "cf-owner"},
			{ClusterID: "200", PrivateLinkConnectionID: plc.ID, BizID: "cf-shared"},
		}, nil)

	deleted, err := svc.DeletePrivateLinkConnection(ctx, "200", plc.ID)

	require.Nil(t, deleted)
	require.Error(t, err)
	require.Contains(t, err.Error(), "cf-owner")
	require.Contains(t, err.Error(), "cf-shared")
}

func TestDeleteMSKPrivateLinkConnectionUsesPersistentRoute(t *testing.T) {
	ctrl := gomock.NewController(t)
	plcRepo := mock_repo.NewMockPrivateLinkConnectionRepo(ctrl)
	plcLinkRepo := mock_repo.NewMockPrivateLinkConnectionLinkRepo(ctrl)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	svc := &privateLinkConnectionService{
		mgmtSvc:             mgmtSvc,
		plcRepo:             plcRepo,
		plcLinkRepo:         plcLinkRepo,
		plcAttachDomainRepo: mock_repo.NewMockPLCAttachDomainRepo(ctrl),
		transProvider:       fakeTransProvider{},
	}

	plc := newMSKPrivateLinkConnection()
	plcRepo.EXPECT().GetById(gomock.Any(), "200", plc.ID).Return(plc, nil)
	plcLinkRepo.EXPECT().ListByPrivateLinkConnectionID(gomock.Any(), "", plc.ID).Return(nil, nil)
	mgmtSvc.EXPECT().
		DeleteMSKVPCConnection(gomock.Any(), gomock.AssignableToTypeOf(&globalserverlessv1.DeleteMSKVPCConnectionRequest{})).
		DoAndReturn(func(_ context.Context, req *globalserverlessv1.DeleteMSKVPCConnectionRequest) (*globalserverlessv1.DeleteMSKVPCConnectionResponse, bizerr.BizError) {
			require.Empty(t, req.GetClusterId())
			require.Equal(t, globalcommonv1.CloudProvider_AWS, req.GetCloudProviderId())
			require.Equal(t, uint64(300), req.GetOrganizationId())
			require.Equal(t, globalcommonv1.ServicePlan_Essential, req.GetServicePlan())
			require.Equal(t, plc.Region, req.GetRegion())
			require.Equal(t, plc.Kubernetes, req.GetKubernetes())
			return &globalserverlessv1.DeleteMSKVPCConnectionResponse{}, nil
		})
	plcRepo.EXPECT().Save(gomock.Any(), plc).Return(plc, nil)

	deleted, err := svc.DeletePrivateLinkConnection(context.Background(), "200", plc.ID)

	require.NoError(t, err)
	require.Equal(t, constant.PrivateLinkConnectionStateDeleting, deleted.State)
}

func TestCreateMSKPrivateLinkConnectionReusesSameKubernetesEndpoint(t *testing.T) {
	ctrl := gomock.NewController(t)
	plcRepo := mock_repo.NewMockPrivateLinkConnectionRepo(ctrl)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	svc := &privateLinkConnectionService{
		mgmtSvc:       mgmtSvc,
		plcRepo:       plcRepo,
		transProvider: fakeTransProvider{},
	}

	ctx := context.Background()
	existing := newMSKPrivateLinkConnection()
	mgmtSvc.EXPECT().
		GetCluster(gomock.Any(), uint64(200), tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC).
		Return(newSharedServerlessCluster(), nil)
	plcRepo.EXPECT().
		ListByFilter(gomock.Any(), gomock.AssignableToTypeOf(&entity.ListPrivateLinkConnectionFilter{})).
		DoAndReturn(func(_ context.Context, filter *entity.ListPrivateLinkConnectionFilter) ([]*entity.PrivateLinkConnection, bizerr.BizError) {
			require.Equal(t, "300", filter.TenantID)
			require.Equal(t, globalcommonv1.Region_AWS_US_EAST_1.String(), filter.Region)
			require.Equal(t, "shared-k8s", filter.Kubernetes)
			require.Equal(t, constant.PrivateLinkConnectionAmazonMSKProvisioned, filter.Type)
			require.Equal(t, existing.AmazonMSKProvisioned.MSKArn, filter.AmazonMSKProvisionedArn)
			require.ElementsMatch(t, []constant.PrivateLinkConnectionState{
				constant.PrivateLinkConnectionStateCreating,
				constant.PrivateLinkConnectionStatePendingAcceptance,
				constant.PrivateLinkConnectionStateAvailable,
			}, filter.States)
			return []*entity.PrivateLinkConnection{existing}, nil
		})

	created, err := svc.CreatePrivateLinkConnection(ctx, "200", &entity.PrivateLinkConnection{
		DisplayName:             "msk",
		CreatedBy:               "user",
		Type:                    constant.PrivateLinkConnectionAmazonMSKProvisioned,
		AmazonMSKProvisionedArn: existing.AmazonMSKProvisioned.MSKArn,
		AmazonMSKProvisioned: &entity.AmazonMSKProvisioned{
			MSKArn: existing.AmazonMSKProvisioned.MSKArn,
		},
	})

	require.NoError(t, err)
	require.Equal(t, existing, created)
}

func TestCreateMSKPrivateLinkConnectionReplacesFailedEndpoint(t *testing.T) {
	ctrl := gomock.NewController(t)
	plcRepo := mock_repo.NewMockPrivateLinkConnectionRepo(ctrl)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	svc := &privateLinkConnectionService{
		mgmtSvc:       mgmtSvc,
		plcRepo:       plcRepo,
		transProvider: fakeTransProvider{},
	}

	ctx := context.Background()
	mskArn := "arn:aws:kafka:us-east-1:999:cluster/msk/abc"
	mgmtSvc.EXPECT().
		GetCluster(gomock.Any(), uint64(200), tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC).
		Return(newSharedServerlessCluster(), nil)
	plcRepo.EXPECT().
		ListByFilter(gomock.Any(), gomock.AssignableToTypeOf(&entity.ListPrivateLinkConnectionFilter{})).
		DoAndReturn(func(_ context.Context, filter *entity.ListPrivateLinkConnectionFilter) ([]*entity.PrivateLinkConnection, bizerr.BizError) {
			require.Equal(t, mskArn, filter.AmazonMSKProvisionedArn)
			require.ElementsMatch(t, []constant.PrivateLinkConnectionState{
				constant.PrivateLinkConnectionStateCreating,
				constant.PrivateLinkConnectionStatePendingAcceptance,
				constant.PrivateLinkConnectionStateAvailable,
			}, filter.States)
			return nil, nil
		})
	plcRepo.EXPECT().
		CountByFilter(gomock.Any(), gomock.AssignableToTypeOf(&entity.ListPrivateLinkConnectionFilter{})).
		DoAndReturn(func(_ context.Context, filter *entity.ListPrivateLinkConnectionFilter) (int64, bizerr.BizError) {
			require.Equal(
				t,
				[]constant.PrivateLinkConnectionState{constant.PrivateLinkConnectionStateDeleting},
				filter.States,
			)
			return 0, nil
		})
	plcRepo.EXPECT().
		CountByFilter(gomock.Any(), gomock.AssignableToTypeOf(&entity.ListPrivateLinkConnectionFilter{})).
		DoAndReturn(func(_ context.Context, filter *entity.ListPrivateLinkConnectionFilter) (int64, bizerr.BizError) {
			require.ElementsMatch(t, []constant.PrivateLinkConnectionState{
				constant.PrivateLinkConnectionStateCreating,
				constant.PrivateLinkConnectionStatePendingAcceptance,
				constant.PrivateLinkConnectionStateAvailable,
			}, filter.States)
			return 0, nil
		})
	// Shared MSK connections must not pin a cluster custom status, otherwise the
	// creating cluster can never be deleted while other clusters reuse them.
	mgmtSvc.EXPECT().
		SetClusterPrivateLinkConnectionStatus(gomock.Any(), gomock.Any(), gomock.Any()).
		Times(0)
	mgmtSvc.EXPECT().
		CreateMSKVPCConnection(gomock.Any(), gomock.AssignableToTypeOf(&globalserverlessv1.CreateMSKVPCConnectionRequest{})).
		DoAndReturn(func(_ context.Context, req *globalserverlessv1.CreateMSKVPCConnectionRequest) (*globalserverlessv1.CreateMSKVPCConnectionResponse, bizerr.BizError) {
			require.Empty(t, req.GetClusterId())
			require.Equal(t, globalcommonv1.Region_AWS_US_EAST_1.String(), req.GetRegion())
			require.Equal(t, "shared-k8s", req.GetKubernetes())
			require.Equal(t, uint64(300), req.GetOrganizationId())
			require.Equal(t, globalcommonv1.ServicePlan_Essential, req.GetServicePlan())
			require.Equal(t, mskArn, req.GetMskArn())
			return &globalserverlessv1.CreateMSKVPCConnectionResponse{
				MskVpcConnection: &regionalserverlessv1.MSKVpcConnection{
					MskVpcConnectionArn: "arn:aws:kafka:us-east-1:300:vpc-connection/msk/new",
					BootstrapBroker:     "b-2.example.com:9098",
					State:               regionalserverlessv1.MSKVpcConnectionState_CREATING,
				},
			}, nil
		})
	plcRepo.EXPECT().
		Create(gomock.Any(), gomock.AssignableToTypeOf(&entity.PrivateLinkConnection{})).
		DoAndReturn(func(_ context.Context, plc *entity.PrivateLinkConnection) (*entity.PrivateLinkConnection, bizerr.BizError) {
			require.Equal(t, globalcommonv1.Region_AWS_US_EAST_1.String(), plc.Region)
			require.Equal(t, "shared-k8s", plc.Kubernetes)
			require.Equal(t, constant.PrivateLinkConnectionStateCreating, plc.State)
			require.Zero(t, plc.ClusterStatusId)
			return plc, nil
		})

	created, err := svc.CreatePrivateLinkConnection(ctx, "200", &entity.PrivateLinkConnection{
		DisplayName:             "msk",
		CreatedBy:               "user",
		Type:                    constant.PrivateLinkConnectionAmazonMSKProvisioned,
		AmazonMSKProvisionedArn: mskArn,
		AmazonMSKProvisioned: &entity.AmazonMSKProvisioned{
			MSKArn: mskArn,
		},
	})

	require.NoError(t, err)
	require.NotNil(t, created)
	require.Equal(
		t,
		"arn:aws:kafka:us-east-1:300:vpc-connection/msk/new",
		created.AmazonMSKProvisioned.MSKVPCConnectionArn,
	)
}

func TestCreateMSKPrivateLinkConnectionRejectsMissingKubernetes(t *testing.T) {
	ctrl := gomock.NewController(t)
	plcRepo := mock_repo.NewMockPrivateLinkConnectionRepo(ctrl)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	svc := &privateLinkConnectionService{
		mgmtSvc:       mgmtSvc,
		plcRepo:       plcRepo,
		transProvider: fakeTransProvider{},
	}

	mgmtSvc.EXPECT().
		GetCluster(gomock.Any(), uint64(200), tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC).
		Return(&tidbmgmtservicev1.Cluster{
			ClusterId:      200,
			OrganizationId: 300,
			CloudProvider:  config.ProviderAWS,
			Region:         globalcommonv1.Region_AWS_US_EAST_1.String(),
			ClusterInfo: &tidbmgmtservicev1.Cluster_ServerlessCluster{
				ServerlessCluster: &globalserverlessv1.ServerlessServiceCluster{
					UserStatus: globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_NORMAL,
					Cluster:    &regionalserverlessv1.ServerlessRegionalServiceClusterInfo{},
				},
			},
		}, nil)

	_, err := svc.CreatePrivateLinkConnection(context.Background(), "200", &entity.PrivateLinkConnection{
		DisplayName: "msk",
		CreatedBy:   "user",
		Type:        constant.PrivateLinkConnectionAmazonMSKProvisioned,
		AmazonMSKProvisioned: &entity.AmazonMSKProvisioned{
			MSKArn: "arn:aws:kafka:us-east-1:999:cluster/msk/abc",
		},
	})

	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeParams, err.ErrCode())
	require.Contains(t, err.Error(), "kubernetes is required")
}

func TestCreateMSKPrivateLinkConnectionRejectsStarterCluster(t *testing.T) {
	ctrl := gomock.NewController(t)
	plcRepo := mock_repo.NewMockPrivateLinkConnectionRepo(ctrl)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	svc := &privateLinkConnectionService{
		mgmtSvc:       mgmtSvc,
		plcRepo:       plcRepo,
		transProvider: fakeTransProvider{},
	}

	cluster := newSharedServerlessCluster()
	cluster.GetServerlessCluster().GetCluster().AutoScaling = nil
	mgmtSvc.EXPECT().
		GetCluster(gomock.Any(), uint64(200), tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC).
		Return(cluster, nil)

	_, err := svc.CreatePrivateLinkConnection(context.Background(), "200", &entity.PrivateLinkConnection{
		DisplayName: "msk",
		CreatedBy:   "user",
		Type:        constant.PrivateLinkConnectionAmazonMSKProvisioned,
		AmazonMSKProvisioned: &entity.AmazonMSKProvisioned{
			MSKArn: "arn:aws:kafka:us-east-1:999:cluster/msk/abc",
		},
	})

	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeParams, err.ErrCode())
	require.Contains(t, err.Error(), "only supports Essential")
}

func TestMSKPrivateLinkConnectionWorkerUsesPersistentRoute(t *testing.T) {
	ctrl := gomock.NewController(t)
	plcRepo := mock_repo.NewMockPrivateLinkConnectionRepo(ctrl)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	worker := &privateLinkConnectionWorker{
		mgmtSvc: mgmtSvc,
		plcRepo: plcRepo,
	}
	plc := newMSKPrivateLinkConnection()

	mgmtSvc.EXPECT().
		GetMSKVPCConnection(gomock.Any(), gomock.AssignableToTypeOf(&globalserverlessv1.GetMSKVPCConnectionRequest{})).
		DoAndReturn(func(_ context.Context, req *globalserverlessv1.GetMSKVPCConnectionRequest) (*globalserverlessv1.GetMSKVPCConnectionResponse, bizerr.BizError) {
			require.Empty(t, req.GetClusterId())
			require.Equal(t, globalcommonv1.CloudProvider_AWS, req.GetCloudProviderId())
			require.Equal(t, uint64(300), req.GetOrganizationId())
			require.Equal(t, globalcommonv1.ServicePlan_Essential, req.GetServicePlan())
			require.Equal(t, plc.Region, req.GetRegion())
			require.Equal(t, plc.Kubernetes, req.GetKubernetes())
			return &globalserverlessv1.GetMSKVPCConnectionResponse{
				MskVpcConnection: &regionalserverlessv1.MSKVpcConnection{
					State: regionalserverlessv1.MSKVpcConnectionState_AVAILABLE,
				},
			}, nil
		})
	plcRepo.EXPECT().Save(gomock.Any(), plc).Return(plc, nil)

	require.NoError(t, worker.syncAmazonMSKProvisionedPLC(context.Background(), plc))
}

func TestMSKPrivateLinkConnectionWorkerReleasesLegacyClusterStatus(t *testing.T) {
	ctrl := gomock.NewController(t)
	plcRepo := mock_repo.NewMockPrivateLinkConnectionRepo(ctrl)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	worker := &privateLinkConnectionWorker{
		mgmtSvc:       mgmtSvc,
		plcRepo:       plcRepo,
		transProvider: fakeTransProvider{},
	}
	// An MSK connection created before they became shared still pins a custom
	// status on its creating cluster, which would block deleting that cluster.
	plc := newMSKPrivateLinkConnection()
	plc.ClusterStatusId = 42

	plcRepo.EXPECT().
		Save(gomock.Any(), plc).
		DoAndReturn(func(_ context.Context, p *entity.PrivateLinkConnection) (*entity.PrivateLinkConnection, bizerr.BizError) {
			require.Zero(t, p.ClusterStatusId)
			return p, nil
		})
	mgmtSvc.EXPECT().DeleteClusterCustomStatus(gomock.Any(), uint64(100), uint64(42)).Return(nil)

	require.NoError(t, worker.releaseMSKClusterStatus(context.Background(), plc))
	require.Zero(t, plc.ClusterStatusId)

	// Already healed: nothing further to do, and no extra mgmtsvc calls.
	require.NoError(t, worker.releaseMSKClusterStatus(context.Background(), plc))
}

func TestDeletingMSKPrivateLinkConnectionWorkerUsesPersistentRoute(t *testing.T) {
	ctrl := gomock.NewController(t)
	plcRepo := mock_repo.NewMockPrivateLinkConnectionRepo(ctrl)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	worker := &privateLinkConnectionWorker{
		mgmtSvc: mgmtSvc,
		plcRepo: plcRepo,
	}
	plc := newMSKPrivateLinkConnection()

	mgmtSvc.EXPECT().
		GetMSKVPCConnection(gomock.Any(), gomock.AssignableToTypeOf(&globalserverlessv1.GetMSKVPCConnectionRequest{})).
		DoAndReturn(func(_ context.Context, req *globalserverlessv1.GetMSKVPCConnectionRequest) (*globalserverlessv1.GetMSKVPCConnectionResponse, bizerr.BizError) {
			require.Empty(t, req.GetClusterId())
			require.Equal(t, globalcommonv1.CloudProvider_AWS, req.GetCloudProviderId())
			require.Equal(t, uint64(300), req.GetOrganizationId())
			require.Equal(t, globalcommonv1.ServicePlan_Essential, req.GetServicePlan())
			return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
		})
	plcRepo.EXPECT().DeleteById(gomock.Any(), "100", plc.ID).Return(nil)

	require.NoError(t, worker.deleteAmazonMSKProvisionedPLC(context.Background(), plc))
}

func newMSKPrivateLinkConnection() *entity.PrivateLinkConnection {
	return &entity.PrivateLinkConnection{
		ID:                      "plc-1",
		ClusterID:               "100",
		TenantID:                "300",
		CloudProvider:           config.ProviderAWS,
		Region:                  globalcommonv1.Region_AWS_US_EAST_1.String(),
		Kubernetes:              "shared-k8s",
		DisplayName:             "msk",
		State:                   constant.PrivateLinkConnectionStateAvailable,
		Type:                    constant.PrivateLinkConnectionAmazonMSKProvisioned,
		AmazonMSKProvisionedArn: "arn:aws:kafka:us-east-1:999:cluster/msk/abc",
		AmazonMSKProvisioned: &entity.AmazonMSKProvisioned{
			MSKArn:                "arn:aws:kafka:us-east-1:999:cluster/msk/abc",
			MSKVPCConnectionArn:   "arn:aws:kafka:us-east-1:300:vpc-connection/msk/abc",
			MSKVPCConnectionState: "AVAILABLE",
			BootstrapBroker:       "b-1.example.com:9098",
		},
	}
}

func newSharedServerlessCluster() *tidbmgmtservicev1.Cluster {
	return &tidbmgmtservicev1.Cluster{
		ClusterId:      200,
		OrganizationId: 300,
		CloudProvider:  config.ProviderAWS,
		Region:         globalcommonv1.Region_AWS_US_EAST_1.String(),
		ClusterInfo: &tidbmgmtservicev1.Cluster_ServerlessCluster{
			ServerlessCluster: &globalserverlessv1.ServerlessServiceCluster{
				UserStatus: globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_NORMAL,
				Cluster: &regionalserverlessv1.ServerlessRegionalServiceClusterInfo{
					Kubernetes: "shared-k8s",
					AutoScaling: &regionalserverlessv1.ServerlessRegionalServiceClusterInfo_AutoScaling{
						MinRcu: 1,
						MaxRcu: 1,
					},
				},
			},
		},
	}
}