package privatelink

import (
	"context"
	"net"
	"testing"
	"time"

	"buf.build/go/protovalidate"
	"github.com/aws/aws-sdk-go-v2/service/ec2"
	protovalidate_middleware "github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/protovalidate"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	regionalapiv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/dataplane-manager/v1"
	"go.uber.org/mock/gomock"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/grpc/test/bufconn"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	cloudaws "github.com/tidbcloud/infra-provider/regional-server/pkg/cloud/aws"
	bizerr "github.com/tidbcloud/infra-provider/regional-server/pkg/common/errors"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/config"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/consts"
	"github.com/tidbcloud/infra-provider/regional-server/test/mockrepository"
)

type recordingEndpointServiceZoneResolver struct {
	roleARN       string
	externalID    string
	serviceRegion string
	serviceName   string
	zones         []*regionalapiv1.EndpointServiceAvailabilityZone
	err           error
}

func (r *recordingEndpointServiceZoneResolver) Resolve(
	_ context.Context,
	roleARN string,
	externalID string,
	serviceRegion string,
	serviceName string,
) ([]*regionalapiv1.EndpointServiceAvailabilityZone, error) {
	r.roleARN = roleARN
	r.externalID = externalID
	r.serviceRegion = serviceRegion
	r.serviceName = serviceName
	return r.zones, r.err
}

func newTestGRPCServer(t *testing.T) *grpc.Server {
	validator, err := protovalidate.New()
	require.NoError(t, err)

	opts := []grpc.ServerOption{
		grpc.ChainUnaryInterceptor(
			protovalidate_middleware.UnaryServerInterceptor(validator),
		),
	}
	srv := grpc.NewServer(opts...)

	svc := &Service{
		privatelinkserviceRW:  &fakePrivateLinkServiceReaderWriter{},
		privatelinkendpointRW: &fakePrivateLinkEndpointReaderWriter{},
		dataPlaneR:            &fakeDataPlaneReader{},
		vpcR:                  &fakeVPCReader{},
		config: config.RootConfig{
			Namespace: "default",
			ControllerConfig: config.ControllerConfig{
				RegionConfig: config.RegionConfig{
					Name: "us-west-2",
				},
			},
		},
	}

	regionalapiv1.RegisterPrivateLinkConnectionServiceServer(srv, svc)

	return srv
}

func newTestGRPCClient(t *testing.T, srv *grpc.Server) regionalapiv1.PrivateLinkConnectionServiceClient {
	lis := bufconn.Listen(1024 * 1024)
	go func() {
		_ = srv.Serve(lis)
	}()

	dialer := grpc.WithContextDialer(func(ctx context.Context, addr string) (net.Conn, error) {
		return lis.Dial()
	})

	conn, err := grpc.Dial("bufnet", grpc.WithInsecure(), dialer)
	require.NoError(t, err)

	return regionalapiv1.NewPrivateLinkConnectionServiceClient(conn)
}

func TestPrivateLinkService_Protovalidate_InvalidRequests(t *testing.T) {
	srv := newTestGRPCServer(t)
	client := newTestGRPCClient(t, srv)

	t.Run("empty dataplane_id in CreatePrivateLinkServiceRequest", func(t *testing.T) {
		req := &regionalapiv1.CreatePrivateLinkServiceRequest{
			DataPlaneId: "",
			Id:          "pls-abc",
			PrivateLinkService: &regionalapiv1.PrivateLinkService{
				Id:               "pls-abc",
				LoadBalancerName: "lb",
			},
		}
		resp, err := client.CreatePrivateLinkService(context.Background(), req)

		assert.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "data_plane_id")
		assert.Equal(t, codes.InvalidArgument, status.Code(err))
	})

	t.Run("empty Id in CreatePrivateLinkServiceRequest", func(t *testing.T) {
		req := &regionalapiv1.CreatePrivateLinkServiceRequest{
			DataPlaneId: "dp-123",
			Id:          "",
			PrivateLinkService: &regionalapiv1.PrivateLinkService{
				Id:               "pls-abc",
				LoadBalancerName: "lb",
			},
		}
		resp, err := client.CreatePrivateLinkService(context.Background(), req)

		assert.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "id")
		assert.Equal(t, codes.InvalidArgument, status.Code(err))
	})

	t.Run("missing PrivateLinkService in CreatePrivateLinkServiceRequest", func(t *testing.T) {
		req := &regionalapiv1.CreatePrivateLinkServiceRequest{
			DataPlaneId:        "dp-123",
			Id:                 "pls-abc",
			PrivateLinkService: nil,
		}
		resp, err := client.CreatePrivateLinkService(context.Background(), req)

		assert.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "private_link_service")
		assert.Equal(t, codes.InvalidArgument, status.Code(err))
	})

	t.Run("empty DataplaneId in GetPrivateLinkServiceRequest", func(t *testing.T) {
		req := &regionalapiv1.GetPrivateLinkServiceRequest{
			DataPlaneId: "",
			Id:          "pls-abc",
		}
		resp, err := client.GetPrivateLinkService(context.Background(), req)

		assert.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "data_plane_id")
		assert.Equal(t, codes.InvalidArgument, status.Code(err))
	})

	t.Run("empty id in GetPrivateLinkServiceRequest", func(t *testing.T) {
		req := &regionalapiv1.GetPrivateLinkServiceRequest{
			DataPlaneId: "dp-123",
			Id:          "",
		}
		resp, err := client.GetPrivateLinkService(context.Background(), req)

		assert.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "id")
		assert.Equal(t, codes.InvalidArgument, status.Code(err))
	})

	t.Run("empty data_plane_id in DeletePrivateLinkServiceRequest", func(t *testing.T) {
		req := &regionalapiv1.DeletePrivateLinkServiceRequest{
			DataPlaneId: "",
			Id:          "pls-abc",
		}
		resp, err := client.DeletePrivateLinkService(context.Background(), req)

		assert.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "data_plane_id")
		assert.Equal(t, codes.InvalidArgument, status.Code(err))
	})

	t.Run("empty id in DeletePrivateLinkServiceRequest", func(t *testing.T) {
		req := &regionalapiv1.DeletePrivateLinkServiceRequest{
			DataPlaneId: "dp-123",
			Id:          "",
		}
		resp, err := client.DeletePrivateLinkService(context.Background(), req)

		assert.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "id")
		assert.Equal(t, codes.InvalidArgument, status.Code(err))
	})

	t.Run("empty data_plane_id in UpdatePrivateLinkServiceRequest", func(t *testing.T) {
		req := &regionalapiv1.UpdatePrivateLinkServiceRequest{
			DataPlaneId: "",
			Id:          "pls-123",
		}
		resp, err := client.UpdatePrivateLinkService(context.Background(), req)

		assert.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "data_plane_id")
		assert.Equal(t, codes.InvalidArgument, status.Code(err))
	})

	t.Run("empty id in UpdatePrivateLinkServiceRequest", func(t *testing.T) {
		req := &regionalapiv1.UpdatePrivateLinkServiceRequest{
			DataPlaneId: "dp-123",
			Id:          "",
		}
		resp, err := client.UpdatePrivateLinkService(context.Background(), req)

		assert.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "id")
		assert.Equal(t, codes.InvalidArgument, status.Code(err))
	})
}

func TestCreatePrivateLinkService(t *testing.T) {
	ctl := gomock.NewController(t)
	defer ctl.Finish()
	dataplaneReader := mockrepository.NewMockDataPlaneReader(ctl)
	plsrw := mockrepository.NewMockPrivateLinkServiceReaderWriter(ctl)
	plerw := mockrepository.NewMockPrivateLinkEndpointReaderWriter(ctl)
	plssvc := NewService(plsrw, plerw, dataplaneReader, nil, config.RootConfig{
		ControllerConfig: config.ControllerConfig{
			Provider: consts.ProviderTypeEKS,
			RegionConfig: config.RegionConfig{
				Name: "us-west-1",
			},
		},
		Namespace: "default",
	}, nil, nil)
	t.Run("get dataplane failed", func(t *testing.T) {
		dataplaneReader.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-dataplane", Namespace: "default"}, gomock.Any()).Return(assert.AnError)
		_, err := plssvc.CreatePrivateLinkService(context.Background(), &regionalapiv1.CreatePrivateLinkServiceRequest{
			DataPlaneId: "test-dataplane",
			PrivateLinkService: &regionalapiv1.PrivateLinkService{
				Id: "test-pls",
				Tags: map[string]string{
					consts.TenantLabelKey: "test-tenant",
				},
			},
		})
		assert.ErrorContains(t, err, "failed get dataplane")
	})
	t.Run("create pls failed", func(t *testing.T) {
		dataplaneObj := &v1alpha1.DataPlane{
			ObjectMeta: metav1.ObjectMeta{
				Name:      "test-dataplane",
				Namespace: "default",
			},
			Spec: v1alpha1.DataPlaneSpec{},
			Status: v1alpha1.DataPlaneStatus{
				Vpc: &v1alpha1.Reference{
					Name:      "test-vpc-name",
					Namespace: "default",
				},
			},
		}
		dataplaneReader.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-dataplane", Namespace: "default"}, gomock.Any()).DoAndReturn(func(ctx context.Context, key client.ObjectKey, dp *v1alpha1.DataPlane) error {
			*dp = *dataplaneObj
			return nil
		})
		plsrw.EXPECT().Create(context.Background(), gomock.Any()).Return(assert.AnError)
		_, err := plssvc.CreatePrivateLinkService(context.Background(), &regionalapiv1.CreatePrivateLinkServiceRequest{
			DataPlaneId: "test-dataplane",
			PrivateLinkService: &regionalapiv1.PrivateLinkService{
				Id: "test-pls",
				Tags: map[string]string{
					consts.TenantLabelKey: "test-tenant",
				},
			},
		})
		assert.ErrorContains(t, err, "failed to create privatelinkservice")
	})
	t.Run("create pls success", func(t *testing.T) {
		req := &regionalapiv1.CreatePrivateLinkServiceRequest{
			DataPlaneId: "test-dataplane",
			Id:          "test-pls",
			PrivateLinkService: &regionalapiv1.PrivateLinkService{
				Id:               "test-pls",
				LoadBalancerName: "test-lb",
				Connections: []*regionalapiv1.PrivateLinkService_Connection{
					{
						Status:  regionalapiv1.PrivateLinkService_Connection_PENDING,
						Message: "pending",
						Info: &regionalapiv1.PrivateLinkService_Connection_Aws_{
							Aws: &regionalapiv1.PrivateLinkService_Connection_Aws{
								VpcEndpointId: "https://test.com",
							},
						},
					},
				},
				Tags: map[string]string{
					consts.TenantLabelKey: "test-tenant",
				},
			},
		}
		dataplaneObj := &v1alpha1.DataPlane{
			ObjectMeta: metav1.ObjectMeta{
				Name:      "test-dataplane",
				Namespace: "default",
			},
			Spec: v1alpha1.DataPlaneSpec{},
			Status: v1alpha1.DataPlaneStatus{
				Vpc: &v1alpha1.Reference{
					Name:      "test-vpc-name",
					Namespace: "default",
				},
			},
		}
		dataplaneReader.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-dataplane", Namespace: "default"}, gomock.Any()).DoAndReturn(func(ctx context.Context, key client.ObjectKey, dp *v1alpha1.DataPlane) error {
			*dp = *dataplaneObj
			return nil
		})
		// plsrw.EXPECT().Create(context.Background(), &v1alpha1.PrivateLinkService{
		// 	ObjectMeta: metav1.ObjectMeta{
		// 		Name:      req.Id,
		// 		Namespace: dataplaneObj.Namespace,
		// 		Labels:    dataplaneObj.Labels,
		// 	},
		// 	Spec: v1alpha1.PrivateLinkServiceSpec{
		// 		Provider: plssvc.config.Provider,
		// 		Region:   plssvc.config.RegionConfig.Name,
		// 		VPCRef: v1alpha1.Reference{
		// 			Name:      dataplaneObj.Status.Vpc.Name,
		// 			Namespace: dataplaneObj.Status.Vpc.Namespace,
		// 		},
		// 		LoadBalancerName: req.PrivateLinkService.LoadBalancerName,
		// 		Tags:             req.PrivateLinkService.Tags,
		// 	},
		// }).Return(nil)
		plsrw.EXPECT().Create(context.Background(), gomock.Any()).Return(nil)
		_, err := plssvc.CreatePrivateLinkService(context.Background(), req)
		assert.NoError(t, err)
	})
}

func TestCreatePrivateLinkService_MissingTenantTag(t *testing.T) {
	ctl := gomock.NewController(t)
	defer ctl.Finish()

	dataplaneReader := mockrepository.NewMockDataPlaneReader(ctl)
	plsrw := mockrepository.NewMockPrivateLinkServiceReaderWriter(ctl)
	plerw := mockrepository.NewMockPrivateLinkEndpointReaderWriter(ctl)

	plssvc := NewService(plsrw, plerw, dataplaneReader, nil, config.RootConfig{
		ControllerConfig: config.ControllerConfig{
			Provider: consts.ProviderTypeEKS,
			RegionConfig: config.RegionConfig{
				Name: "us-west-1",
			},
		},
		Namespace: "default",
	}, nil, nil)

	t.Run("missing tenant tag", func(t *testing.T) {
		req := &regionalapiv1.CreatePrivateLinkServiceRequest{
			DataPlaneId: "test-dataplane",
			Id:          "test-pls",
			PrivateLinkService: &regionalapiv1.PrivateLinkService{
				Id:               "test-pls",
				LoadBalancerName: "test-lb",
				Tags: map[string]string{
					"some-other-tag": "value",
				},
			},
		}

		_, err := plssvc.CreatePrivateLinkService(context.Background(), req)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "missing required tag: tags.tidbcloud.com/tenant")
	})
}

func TestValidatePrivateLinkEndpointZones(t *testing.T) {
	tests := []struct {
		name    string
		vpc     v1alpha1.VPC
		zones   []string
		wantErr string
	}{
		{
			name: "default uses node subnets",
			vpc: v1alpha1.VPC{
				Status: v1alpha1.VPCStatus{
					NodeSubnets: []v1alpha1.Subnet{
						{AvailabilityZoneID: "use1-az1", AvailabilityZone: "us-east-1a"},
						{AvailabilityZoneID: "use1-az2", AvailabilityZone: "us-east-1b"},
					},
				},
			},
			zones: []string{"use1-az1", "use1-az2"},
		},
		{
			name: "default rejects zones outside node subnets",
			vpc: v1alpha1.VPC{
				Status: v1alpha1.VPCStatus{
					NodeSubnets: []v1alpha1.Subnet{
						{AvailabilityZoneID: "use1-az1", AvailabilityZone: "us-east-1a"},
					},
				},
			},
			zones:   []string{"use1-az2"},
			wantErr: "zone use1-az2 does not match with tidb zone",
		},
		{
			name: "endpoint subnets accept requested non node zones",
			vpc: v1alpha1.VPC{
				Spec: v1alpha1.VPCSpec{
					NodeZones:                []string{"us-east-1a"},
					PrivateLinkEndpointZones: []string{"us-east-1a", "us-east-1b", "us-east-1c"},
				},
				Status: v1alpha1.VPCStatus{
					NodeSubnets: []v1alpha1.Subnet{
						{AvailabilityZoneID: "use1-az1", AvailabilityZone: "us-east-1a"},
					},
					PrivateLinkEndpointSubnets: []v1alpha1.Subnet{
						{AvailabilityZoneID: "use1-az2", AvailabilityZone: "us-east-1b"},
						{AvailabilityZoneID: "use1-az3", AvailabilityZone: "us-east-1c"},
					},
				},
			},
			zones: []string{"use1-az2", "use1-az3"},
		},
		{
			name: "endpoint subnets reject zones outside endpoint subnets",
			vpc: v1alpha1.VPC{
				Spec: v1alpha1.VPCSpec{
					NodeZones:                []string{"us-east-1a"},
					PrivateLinkEndpointZones: []string{"us-east-1a", "us-east-1b"},
				},
				Status: v1alpha1.VPCStatus{
					PrivateLinkEndpointSubnets: []v1alpha1.Subnet{
						{AvailabilityZoneID: "use1-az2", AvailabilityZone: "us-east-1b"},
					},
				},
			},
			zones:   []string{"use1-az3"},
			wantErr: "zone use1-az3 does not match with private link endpoint subnets",
		},
		{
			name: "endpoint subnets return not ready before parameter invalid",
			vpc: v1alpha1.VPC{
				Spec: v1alpha1.VPCSpec{
					NodeZones:                []string{"us-east-1a"},
					PrivateLinkEndpointZones: []string{"us-east-1a", "us-east-1b"},
				},
				Status: v1alpha1.VPCStatus{
					NodeSubnets: []v1alpha1.Subnet{
						{AvailabilityZoneID: "use1-az1", AvailabilityZone: "us-east-1a"},
					},
				},
			},
			zones:   []string{"use1-az2"},
			wantErr: "private link endpoint subnets are not ready",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			ctl := gomock.NewController(t)
			dataplaneReader := mockrepository.NewMockDataPlaneReader(ctl)
			vpcReader := mockrepository.NewMockVPCReader(ctl)
			svc := NewService(nil, nil, dataplaneReader, vpcReader, config.RootConfig{
				Namespace: "default",
			}, nil, nil)

			dataplaneReader.EXPECT().Get(context.Background(), client.ObjectKey{Name: "dp-1", Namespace: "default"}, gomock.Any()).DoAndReturn(
				func(ctx context.Context, key client.ObjectKey, dp *v1alpha1.DataPlane) error {
					*dp = v1alpha1.DataPlane{
						ObjectMeta: metav1.ObjectMeta{Name: key.Name, Namespace: key.Namespace},
						Status: v1alpha1.DataPlaneStatus{
							Vpc: &v1alpha1.Reference{Name: "vpc-1", Namespace: key.Namespace},
						},
					}
					return nil
				},
			)
			vpcReader.EXPECT().Get(context.Background(), client.ObjectKey{Name: "vpc-1", Namespace: "default"}, gomock.Any()).DoAndReturn(
				func(ctx context.Context, key client.ObjectKey, vpc *v1alpha1.VPC) error {
					*vpc = tt.vpc
					vpc.Name = key.Name
					vpc.Namespace = key.Namespace
					return nil
				},
			)

			_, err := svc.ValidatePrivateLinkEndpointZones(context.Background(), &regionalapiv1.ValidatePrivateLinkEndpointZonesRequest{
				DataPlaneId: "dp-1",
				Zones:       tt.zones,
			})
			if tt.wantErr != "" {
				require.Error(t, err)
				assert.Contains(t, err.Error(), tt.wantErr)
				return
			}
			require.NoError(t, err)
		})
	}
}

func TestDescribePrivateLinkEndpointServiceZones(t *testing.T) {
	ctl := gomock.NewController(t)
	dataplaneReader := mockrepository.NewMockDataPlaneReader(ctl)
	fakeEC2 := &fakeEndpointServiceZoneEC2Client{
		describeServices: func(ctx context.Context, params *ec2.DescribeVpcEndpointServicesInput, call int) (*ec2.DescribeVpcEndpointServicesOutput, error) {
			assert.Equal(t, []string{"com.amazonaws.vpce.us-east-1.vpce-svc-123"}, params.ServiceNames)
			return endpointServiceZonesOutput("us-east-1c", "us-east-1a"), nil
		},
		describeZones: func(ctx context.Context, params *ec2.DescribeAvailabilityZonesInput, call int) (*ec2.DescribeAvailabilityZonesOutput, error) {
			assert.ElementsMatch(t, []string{"us-east-1c", "us-east-1a"}, params.ZoneNames)
			return availabilityZonesOutput(
				endpointServiceAvailabilityZone("us-east-1c", "use1-az2"),
				endpointServiceAvailabilityZone("us-east-1a", "use1-az1"),
			), nil
		},
	}
	var gotRole *cloudaws.AssumedRole
	svc := NewService(nil, nil, dataplaneReader, nil, config.RootConfig{
		ControllerConfig: config.ControllerConfig{
			Provider:               consts.ProviderTypeEKS,
			AWSDataPlaneExternalID: "external-id",
		},
		Namespace: "default",
	}, nil, nil)
	svc.endpointZoneResolver = newPrivateLinkEndpointServiceZoneResolverWithFactory(
		func(role *cloudaws.AssumedRole, region string) endpointServiceEC2Client {
			gotRole = role
			assert.Equal(t, "us-east-1", region)
			return fakeEC2
		},
		time.Minute,
		time.Now,
	)
	dataplaneReader.EXPECT().Get(context.Background(), client.ObjectKey{Name: "dp-1", Namespace: "default"}, gomock.Any()).DoAndReturn(
		func(ctx context.Context, key client.ObjectKey, dp *v1alpha1.DataPlane) error {
			*dp = v1alpha1.DataPlane{
				ObjectMeta: metav1.ObjectMeta{Name: key.Name, Namespace: key.Namespace},
				Spec: v1alpha1.DataPlaneSpec{
					AccountRef: v1alpha1.AccountReference{
						AccountID: "123456789012",
						Role:      "dataplane-manager",
					},
				},
			}
			return nil
		},
	)

	resp, err := svc.DescribePrivateLinkEndpointServiceZones(context.Background(), &regionalapiv1.DescribePrivateLinkEndpointServiceZonesRequest{
		DataPlaneId:   "dp-1",
		ServiceName:   "com.amazonaws.vpce.us-east-1.vpce-svc-123",
		ServiceRegion: "us-east-1",
	})
	require.NoError(t, err)
	assertEndpointServiceZones(t, resp.Zones, []string{"us-east-1a:use1-az1", "us-east-1c:use1-az2"})
	require.NotNil(t, gotRole)
	assert.Equal(t, "arn:aws:iam::123456789012:role/dataplane-manager", gotRole.RoleARN)
	require.NotNil(t, gotRole.ExternalID)
	assert.Equal(t, "external-id", *gotRole.ExternalID)
}

func TestDescribePrivateLinkEndpointServiceZonesRejectsUnsupportedProvider(t *testing.T) {
	svc := NewService(nil, nil, nil, nil, config.RootConfig{
		ControllerConfig: config.ControllerConfig{Provider: consts.ProviderTypeAKS},
		Namespace:        "default",
	}, nil, nil)

	_, err := svc.DescribePrivateLinkEndpointServiceZones(context.Background(), &regionalapiv1.DescribePrivateLinkEndpointServiceZonesRequest{
		DataPlaneId:   "dp-1",
		ServiceName:   "service",
		ServiceRegion: "us-east-1",
	})
	require.Error(t, err)
	assert.True(t, bizerr.Is(err, bizerr.ParameterInvalid))
}

func TestNewServiceSelectsEndpointServiceZoneResolver(t *testing.T) {
	tests := []struct {
		name         string
		provider     string
		resolverType any
	}{
		{name: "aws", provider: consts.ProviderTypeAWS, resolverType: &privateLinkEndpointServiceZoneResolver{}},
		{name: "eks", provider: consts.ProviderTypeEKS, resolverType: &privateLinkEndpointServiceZoneResolver{}},
		{name: "ack", provider: consts.ProviderTypeACK, resolverType: &aliyunPrivateLinkEndpointServiceZoneResolver{}},
		{name: "aks", provider: consts.ProviderTypeAKS},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			svc := NewService(nil, nil, nil, nil, config.RootConfig{
				ControllerConfig: config.ControllerConfig{Provider: tt.provider},
			}, nil, nil)
			if tt.resolverType == nil {
				assert.Nil(t, svc.endpointZoneResolver)
				return
			}
			assert.IsType(t, tt.resolverType, svc.endpointZoneResolver)
		})
	}
}

func TestDescribePrivateLinkEndpointServiceZonesForAliyun(t *testing.T) {
	ctl := gomock.NewController(t)
	dataplaneReader := mockrepository.NewMockDataPlaneReader(ctl)
	resolver := &recordingEndpointServiceZoneResolver{
		zones: []*regionalapiv1.EndpointServiceAvailabilityZone{
			{AvailabilityZone: "ap-southeast-1a", AvailabilityZoneId: "ap-southeast-1a"},
			{AvailabilityZone: "ap-southeast-1b", AvailabilityZoneId: "ap-southeast-1b"},
		},
	}
	svc := NewService(nil, nil, dataplaneReader, nil, config.RootConfig{
		ControllerConfig: config.ControllerConfig{Provider: consts.ProviderTypeACK},
		Namespace:        "default",
	}, nil, nil)
	svc.endpointZoneResolver = resolver
	dataplaneReader.EXPECT().Get(
		context.Background(),
		client.ObjectKey{Name: "dp-1", Namespace: "default"},
		gomock.Any(),
	).DoAndReturn(func(_ context.Context, key client.ObjectKey, dp *v1alpha1.DataPlane) error {
		*dp = v1alpha1.DataPlane{
			ObjectMeta: metav1.ObjectMeta{Name: key.Name, Namespace: key.Namespace},
			Spec: v1alpha1.DataPlaneSpec{AccountRef: v1alpha1.AccountReference{
				AccountID: "123456789012",
				Role:      "dataplane-manager",
			}},
		}
		return nil
	})

	resp, err := svc.DescribePrivateLinkEndpointServiceZones(
		context.Background(),
		&regionalapiv1.DescribePrivateLinkEndpointServiceZonesRequest{
			DataPlaneId:   "dp-1",
			ServiceName:   "com.aliyuncs.privatelink.ap-southeast-1.epsrv-123",
			ServiceRegion: "ap-southeast-1",
		},
	)
	require.NoError(t, err)
	assert.Equal(t, resolver.zones, resp.Zones)
	assert.Equal(t, "acs:ram::123456789012:role/dataplane-manager", resolver.roleARN)
	assert.Empty(t, resolver.externalID)
	assert.Equal(t, "ap-southeast-1", resolver.serviceRegion)
	assert.Equal(t, "com.aliyuncs.privatelink.ap-southeast-1.epsrv-123", resolver.serviceName)
}

func TestGetPrivateLinkService(t *testing.T) {
	ctl := gomock.NewController(t)
	plsrw := mockrepository.NewMockPrivateLinkServiceReaderWriter(ctl)
	plerw := mockrepository.NewMockPrivateLinkEndpointReaderWriter(ctl)
	plssvc := NewService(plsrw, plerw, nil, nil, config.RootConfig{}, nil, nil)
	t.Run("get privatelinkservice failed", func(t *testing.T) {
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: plssvc.config.Namespace}, gomock.Any()).Return(assert.AnError)
		_, err := plssvc.GetPrivateLinkService(context.Background(), &regionalapiv1.GetPrivateLinkServiceRequest{
			DataPlaneId: "test-dataplane",
			Id:          "test-pls",
		})
		assert.ErrorContains(t, err, "failed to get privatelinkservice")
	})

	t.Run("fill privatelinkservice failed", func(t *testing.T) {
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: plssvc.config.Namespace}, gomock.Any()).Return(nil)
		_, err := plssvc.GetPrivateLinkService(context.Background(), &regionalapiv1.GetPrivateLinkServiceRequest{
			DataPlaneId: "test-dataplane",
			Id:          "test-pls",
		})
		assert.ErrorContains(t, err, "failed to fill privatelinkservice")
	})

	t.Run("get privatelinkservice success", func(t *testing.T) {
		plssvc.config = config.RootConfig{
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeEKS,
				RegionConfig: config.RegionConfig{
					Name: "us-west-1",
				},
				Env: "test-env",
			},
		}
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: plssvc.config.Namespace}, gomock.Any()).DoAndReturn(func(ctx context.Context, key client.ObjectKey, pls *v1alpha1.PrivateLinkService) error {
			*pls = v1alpha1.PrivateLinkService{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "test-pls",
					Namespace: plssvc.config.Namespace,
				},
				Spec: v1alpha1.PrivateLinkServiceSpec{
					Provider: consts.ProviderTypeEKS,
					Region:   "us-west-1",
				},
			}
			return nil
		})
		_, err := plssvc.GetPrivateLinkService(context.Background(), &regionalapiv1.GetPrivateLinkServiceRequest{
			DataPlaneId: "test-dataplane",
			Id:          "test-pls",
		})
		assert.NoError(t, err)
	})
}

func TestUpdatePrivateLinkService(t *testing.T) {
	ctl := gomock.NewController(t)
	plsrw := mockrepository.NewMockPrivateLinkServiceReaderWriter(ctl)
	plerw := mockrepository.NewMockPrivateLinkEndpointReaderWriter(ctl)
	plssvc := NewService(plsrw, plerw, nil, nil, config.RootConfig{}, nil, nil)
	t.Run("get privatelinkservice failed", func(t *testing.T) {
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: plssvc.config.Namespace}, gomock.Any()).Return(assert.AnError)
		_, err := plssvc.UpdatePrivateLinkService(context.Background(), &regionalapiv1.UpdatePrivateLinkServiceRequest{
			DataPlaneId: "test-dataplane",
			Id:          "test-pls",
		})
		assert.ErrorContains(t, err, "failed to get privatelinkservice")
	})

	t.Run("update privatelinkservice failed", func(t *testing.T) {
		plssvc.config = config.RootConfig{
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeEKS,
				RegionConfig: config.RegionConfig{
					Name: "us-west-1",
				},
			},
		}
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: plssvc.config.Namespace}, gomock.Any()).DoAndReturn(func(ctx context.Context, key client.ObjectKey, pls *v1alpha1.PrivateLinkService) error {
			*pls = v1alpha1.PrivateLinkService{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "test-pls",
					Namespace: plssvc.config.Namespace,
				},
				Spec: v1alpha1.PrivateLinkServiceSpec{
					Provider: consts.ProviderTypeEKS,
					Region:   "us-west-1",
				},
			}
			return nil
		})
		plsrw.EXPECT().Update(context.Background(), gomock.Any()).Return(assert.AnError)
		_, err := plssvc.UpdatePrivateLinkService(context.Background(), &regionalapiv1.UpdatePrivateLinkServiceRequest{
			DataPlaneId: "test-dataplane",
			Id:          "test-pls",
			PrivateLinkConnections: []*regionalapiv1.PrivateLinkService_Connection{
				{
					Info: &regionalapiv1.PrivateLinkService_Connection_Aws_{
						Aws: &regionalapiv1.PrivateLinkService_Connection_Aws{
							VpcEndpointId: "https://test.com",
						},
					},
				},
			},
		})
		assert.ErrorContains(t, err, "failed to update privatelinkservice")
	})

	t.Run("update privatelinkservice success", func(t *testing.T) {
		plssvc.config = config.RootConfig{
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeEKS,
				RegionConfig: config.RegionConfig{
					Name: "us-west-1",
				},
			},
		}
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: plssvc.config.Namespace}, gomock.Any()).DoAndReturn(func(ctx context.Context, key client.ObjectKey, pls *v1alpha1.PrivateLinkService) error {
			*pls = v1alpha1.PrivateLinkService{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "test-pls",
					Namespace: plssvc.config.Namespace,
				},
				Spec: v1alpha1.PrivateLinkServiceSpec{
					Provider: consts.ProviderTypeEKS,
					Region:   "us-west-1",
				},
			}
			return nil
		})
		plsrw.EXPECT().Update(context.Background(), gomock.Any()).Return(nil)
		_, err := plssvc.UpdatePrivateLinkService(context.Background(), &regionalapiv1.UpdatePrivateLinkServiceRequest{
			DataPlaneId: "test-dataplane",
			Id:          "test-pls",
			PrivateLinkConnections: []*regionalapiv1.PrivateLinkService_Connection{
				{
					Info: &regionalapiv1.PrivateLinkService_Connection_Aws_{
						Aws: &regionalapiv1.PrivateLinkService_Connection_Aws{
							VpcEndpointId: "https://test.com",
						},
					},
				},
			},
		})
		assert.NoError(t, err)
	})
}

func TestCreatePrivateLinkService_ProviderConnectionMismatch(t *testing.T) {
	ctl := gomock.NewController(t)
	dataplaneReader := mockrepository.NewMockDataPlaneReader(ctl)
	plsrw := mockrepository.NewMockPrivateLinkServiceReaderWriter(ctl)
	plerw := mockrepository.NewMockPrivateLinkEndpointReaderWriter(ctl)
	svc := NewService(plsrw, plerw, dataplaneReader, nil, config.RootConfig{
		ControllerConfig: config.ControllerConfig{Provider: consts.ProviderTypeEKS, RegionConfig: config.RegionConfig{Name: "us-west-1"}},
		Namespace:        "default",
	}, nil, nil)

	// DataPlane present
	dataplaneReader.EXPECT().Get(context.Background(), client.ObjectKey{Name: "dp-1", Namespace: "default"}, gomock.Any()).DoAndReturn(
		func(ctx context.Context, key client.ObjectKey, dp *v1alpha1.DataPlane) error {
			*dp = v1alpha1.DataPlane{ObjectMeta: metav1.ObjectMeta{Name: key.Name, Namespace: key.Namespace}, Status: v1alpha1.DataPlaneStatus{Vpc: &v1alpha1.Reference{Name: "vpc-1", Namespace: key.Namespace}}}
			return nil
		},
	)

	// Build request with GCP connection while provider is EKS
	req := &regionalapiv1.CreatePrivateLinkServiceRequest{
		DataPlaneId: "dp-1",
		Id:          "pls-1",
		PrivateLinkService: &regionalapiv1.PrivateLinkService{
			Id:               "pls-1",
			LoadBalancerName: "lb-1",
			Tags:             map[string]string{consts.TenantLabelKey: "t-1"},
			Connections: []*regionalapiv1.PrivateLinkService_Connection{
				{Info: &regionalapiv1.PrivateLinkService_Connection_Gcp_{Gcp: &regionalapiv1.PrivateLinkService_Connection_Gcp{PrivateServiceConnectEndpoint: "psc-1"}}},
			},
		},
	}

	resp, err := svc.CreatePrivateLinkService(context.Background(), req)
	assert.Nil(t, resp)
	assert.Error(t, err)
}

func TestCreatePrivateLinkService_AzureEmptyIPFiltered(t *testing.T) {
	ctl := gomock.NewController(t)
	dataplaneReader := mockrepository.NewMockDataPlaneReader(ctl)
	plsrw := mockrepository.NewMockPrivateLinkServiceReaderWriter(ctl)
	plerw := mockrepository.NewMockPrivateLinkEndpointReaderWriter(ctl)
	svc := NewService(plsrw, plerw, dataplaneReader, nil, config.RootConfig{
		ControllerConfig: config.ControllerConfig{Provider: consts.ProviderTypeAKS, RegionConfig: config.RegionConfig{Name: "eastus"}},
		Namespace:        "default",
	}, nil, nil)

	dataplaneReader.EXPECT().Get(context.Background(), client.ObjectKey{Name: "dp-1", Namespace: "default"}, gomock.Any()).DoAndReturn(
		func(ctx context.Context, key client.ObjectKey, dp *v1alpha1.DataPlane) error {
			*dp = v1alpha1.DataPlane{ObjectMeta: metav1.ObjectMeta{Name: key.Name, Namespace: key.Namespace}, Status: v1alpha1.DataPlaneStatus{Vpc: &v1alpha1.Reference{Name: "vpc-1", Namespace: key.Namespace}}}
			return nil
		},
	)

	// Expect Create to receive a Connection without empty IP value
	plsrw.EXPECT().Create(context.Background(), gomock.Any()).DoAndReturn(
		func(ctx context.Context, pls *v1alpha1.PrivateLinkService) error {
			if len(pls.Spec.Connections) != 1 {
				t.Fatalf("expected 1 connection, got %d", len(pls.Spec.Connections))
			}
			if len(pls.Spec.Connections[0].IPs) != 0 {
				t.Fatalf("expected filtered IPs to be empty, got %#v", pls.Spec.Connections[0].IPs)
			}
			return nil
		},
	)

	req := &regionalapiv1.CreatePrivateLinkServiceRequest{
		DataPlaneId: "dp-1",
		Id:          "pls-1",
		PrivateLinkService: &regionalapiv1.PrivateLinkService{
			Id:               "pls-1",
			LoadBalancerName: "lb-aks",
			Tags:             map[string]string{consts.TenantLabelKey: "t-1"},
			Connections: []*regionalapiv1.PrivateLinkService_Connection{
				{Info: &regionalapiv1.PrivateLinkService_Connection_Azure_{Azure: &regionalapiv1.PrivateLinkService_Connection_Azure{PrivateEndpointResourceId: "/pe/1", PrivateIpAddress: ""}}},
			},
		},
	}

	_, err := svc.CreatePrivateLinkService(context.Background(), req)
	require.NoError(t, err)
}

func Test_genConnectionPB_NilStatusDoesNotPanic(t *testing.T) {
	defer func() {
		if r := recover(); r != nil {
			t.Fatalf("genConnectionPB panicked: %v", r)
		}
	}()
	_, err := genConnectionPB(consts.ProviderTypeEKS, v1alpha1.Connection{Endpoint: "e-1"}, nil)
	assert.NoError(t, err)
}

func TestCreatePrivateLinkService_AliCloudHappyPath(t *testing.T) {
	ctl := gomock.NewController(t)
	dataplaneReader := mockrepository.NewMockDataPlaneReader(ctl)
	plsrw := mockrepository.NewMockPrivateLinkServiceReaderWriter(ctl)
	plerw := mockrepository.NewMockPrivateLinkEndpointReaderWriter(ctl)
	svc := NewService(plsrw, plerw, dataplaneReader, nil, config.RootConfig{
		ControllerConfig: config.ControllerConfig{Provider: consts.ProviderTypeACK, RegionConfig: config.RegionConfig{Name: "cn-hangzhou"}},
		Namespace:        "default",
	}, nil, nil)

	// Mock DataPlane
	dataplaneReader.EXPECT().Get(context.Background(), client.ObjectKey{Name: "dp-ali", Namespace: "default"}, gomock.Any()).DoAndReturn(
		func(ctx context.Context, key client.ObjectKey, dp *v1alpha1.DataPlane) error {
			*dp = v1alpha1.DataPlane{ObjectMeta: metav1.ObjectMeta{Name: key.Name, Namespace: key.Namespace}, Status: v1alpha1.DataPlaneStatus{Vpc: &v1alpha1.Reference{Name: "vpc-ali", Namespace: key.Namespace}}}
			return nil
		},
	)

	// Expect Create called with AliCloud connection carrying DefaultDomainName
	plsrw.EXPECT().Create(context.Background(), gomock.Any()).DoAndReturn(
		func(ctx context.Context, pls *v1alpha1.PrivateLinkService) error {
			if pls.Spec.Provider != consts.ProviderTypeACK {
				t.Fatalf("expected provider %s, got %s", consts.ProviderTypeACK, pls.Spec.Provider)
			}
			if len(pls.Spec.Connections) != 1 {
				t.Fatalf("expected 1 connection, got %d", len(pls.Spec.Connections))
			}
			if pls.Spec.Connections[0].Endpoint != "pe-ali-1" {
				t.Fatalf("unexpected endpoint: %s", pls.Spec.Connections[0].Endpoint)
			}
			if pls.Spec.Connections[0].DefaultDomainName != "default.domain.aliyun" {
				t.Fatalf("unexpected default domain: %s", pls.Spec.Connections[0].DefaultDomainName)
			}
			return nil
		},
	)

	req := &regionalapiv1.CreatePrivateLinkServiceRequest{
		DataPlaneId: "dp-ali",
		Id:          "pls-ali-1",
		PrivateLinkService: &regionalapiv1.PrivateLinkService{
			Id:               "pls-ali-1",
			LoadBalancerName: "lb-ali",
			Tags:             map[string]string{consts.TenantLabelKey: "t-1"},
			Connections: []*regionalapiv1.PrivateLinkService_Connection{
				{Info: &regionalapiv1.PrivateLinkService_Connection_AliCloud_{AliCloud: &regionalapiv1.PrivateLinkService_Connection_AliCloud{
					VpcEndpointId:     "pe-ali-1",
					DefaultDomainName: "default.domain.aliyun",
				}}}},
		},
	}

	resp, err := svc.CreatePrivateLinkService(context.Background(), req)
	require.NoError(t, err)
	require.NotNil(t, resp)
	assert.Equal(t, "pls-ali-1", resp.Id)
	assert.Equal(t, regionalapiv1.PrivateLinkService_CREATING, resp.Status)
}

func TestDeletePrivateLinkService(t *testing.T) {
	ctl := gomock.NewController(t)
	plsrw := mockrepository.NewMockPrivateLinkServiceReaderWriter(ctl)
	plerw := mockrepository.NewMockPrivateLinkEndpointReaderWriter(ctl)
	plssvc := NewService(plsrw, plerw, nil, nil, config.RootConfig{}, nil, nil)
	t.Run("delete privatelinkservice failed", func(t *testing.T) {
		plsrw.EXPECT().Get(context.Background(), gomock.Any(), gomock.Any()).Return(nil)
		plsrw.EXPECT().Delete(context.Background(), gomock.Any()).Return(assert.AnError)
		_, err := plssvc.DeletePrivateLinkService(context.Background(), &regionalapiv1.DeletePrivateLinkServiceRequest{
			DataPlaneId: "test-dataplane",
			Id:          "test-pls",
		})
		assert.ErrorContains(t, err, "failed to delete privatelinkservice")
	})

	t.Run("delete privatelinkservice success", func(t *testing.T) {
		plsrw.EXPECT().Get(context.Background(), gomock.Any(), gomock.Any()).Return(nil)
		plsrw.EXPECT().Delete(context.Background(), gomock.Any()).Return(nil)
		_, err := plssvc.DeletePrivateLinkService(context.Background(), &regionalapiv1.DeletePrivateLinkServiceRequest{
			DataPlaneId: "test-dataplane",
			Id:          "test-pls",
		})
		assert.NoError(t, err)
	})
}

func Test_genConnectionPB(t *testing.T) {
	t.Run("aws connection", func(t *testing.T) {
		conn := v1alpha1.Connection{
			Endpoint: "https://test.com",
		}
		connStatus := &v1alpha1.ConnectionStatus{
			Endpoint: "https://test.com",
			Phase:    v1alpha1.ConnectionPhaseActive,
			Message:  "active",
		}
		connPB, err := genConnectionPB(consts.ProviderTypeEKS, conn, connStatus)
		assert.NoError(t, err)
		assert.Equal(t, regionalapiv1.PrivateLinkService_Connection_ACTIVE, connPB.Status)
		assert.Equal(t, "active", connPB.Message)
		assert.NotNil(t, connPB.GetAws())
		assert.Equal(t, "https://test.com", connPB.GetAws().VpcEndpointId)
	})

	t.Run("gcp connection", func(t *testing.T) {
		conn := v1alpha1.Connection{
			Endpoint: "https://test.com",
		}
		connStatus := &v1alpha1.ConnectionStatus{
			Endpoint: "https://test.com",
			Phase:    v1alpha1.ConnectionPhaseActive,
			Message:  "active",
		}
		connPB, err := genConnectionPB(consts.ProviderTypeGKE, conn, connStatus)
		assert.NoError(t, err)
		assert.Equal(t, regionalapiv1.PrivateLinkService_Connection_ACTIVE, connPB.Status)
		assert.Equal(t, "active", connPB.Message)
		assert.NotNil(t, connPB.GetGcp())
		assert.Equal(t, "https://test.com", connPB.GetGcp().PrivateServiceConnectEndpoint)
	})

	t.Run("azure connection", func(t *testing.T) {
		conn := v1alpha1.Connection{
			Endpoint: "https://test.com",
			IPs:      []string{"192.168.1.1"},
		}
		connStatus := &v1alpha1.ConnectionStatus{
			Endpoint: "https://test.com",
			Phase:    v1alpha1.ConnectionPhaseActive,
			Message:  "active",
			IPs:      []string{"192.168.1.1"},
		}
		connPB, err := genConnectionPB(consts.ProviderTypeAKS, conn, connStatus)
		assert.NoError(t, err)
		assert.Equal(t, regionalapiv1.PrivateLinkService_Connection_ACTIVE, connPB.Status)
		assert.Equal(t, "active", connPB.Message)
		assert.NotNil(t, connPB.GetAzure())
		assert.Equal(t, "https://test.com", connPB.GetAzure().PrivateEndpointResourceId)
		assert.Equal(t, "192.168.1.1", connPB.GetAzure().PrivateIpAddress)
	})

	// Alicloud
	t.Run("alicloud connection", func(t *testing.T) {
		conn := v1alpha1.Connection{
			Endpoint:          "pe-ali-1",
			DefaultDomainName: "default.domain.aliyun",
		}
		connStatus := &v1alpha1.ConnectionStatus{
			Endpoint: "pe-ali-1",
			Phase:    v1alpha1.ConnectionPhaseActive,
			Message:  "ok",
			DNSName:  "conn.tidbcloud.com",
		}
		connPB, err := genConnectionPB(consts.ProviderTypeACK, conn, connStatus)
		assert.NoError(t, err)
		assert.Equal(t, regionalapiv1.PrivateLinkService_Connection_ACTIVE, connPB.Status)
		assert.Equal(t, "ok", connPB.Message)
		assert.NotNil(t, connPB.GetAliCloud())
		assert.Equal(t, "pe-ali-1", connPB.GetAliCloud().VpcEndpointId)
		assert.Equal(t, "default.domain.aliyun", connPB.GetAliCloud().DefaultDomainName)
		assert.Equal(t, "conn.tidbcloud.com", connPB.GetAliCloud().Host)
	})
}

func Test_setConnectionProviderInfo(t *testing.T) {
	t.Run("aws connection", func(t *testing.T) {
		conn := v1alpha1.Connection{
			Endpoint: "https://test.com",
		}
		connInfo := &regionalapiv1.PrivateLinkService_Connection{}
		connInfo, err := setConnectionProviderInfo(connInfo, consts.ProviderTypeEKS, conn, nil)
		assert.NoError(t, err)
		assert.NotNil(t, connInfo.GetAws())
		assert.Equal(t, "https://test.com", connInfo.GetAws().VpcEndpointId)
	})

	t.Run("gcp connection", func(t *testing.T) {
		conn := v1alpha1.Connection{
			Endpoint: "https://test.com",
		}
		connInfo := &regionalapiv1.PrivateLinkService_Connection{}
		connInfo, err := setConnectionProviderInfo(connInfo, consts.ProviderTypeGKE, conn, nil)
		assert.NoError(t, err)
		assert.NotNil(t, connInfo.GetGcp())
		assert.Equal(t, "https://test.com", connInfo.GetGcp().PrivateServiceConnectEndpoint)
	})

	t.Run("azure connection", func(t *testing.T) {
		conn := v1alpha1.Connection{
			Endpoint: "https://test.com",
			IPs:      []string{"192.168.1.1"},
		}
		connInfo := &regionalapiv1.PrivateLinkService_Connection{}
		connInfo, err := setConnectionProviderInfo(connInfo, consts.ProviderTypeAKS, conn, nil)
		assert.NoError(t, err)
		assert.NotNil(t, connInfo.GetAzure())
		assert.Equal(t, "https://test.com", connInfo.GetAzure().PrivateEndpointResourceId)
		assert.Equal(t, "192.168.1.1", connInfo.GetAzure().PrivateIpAddress)
	})

	// Alicloud
	t.Run("alicloud connection", func(t *testing.T) {
		conn := v1alpha1.Connection{
			Endpoint:          "pe-ali-1",
			DefaultDomainName: "default.domain.aliyun",
		}
		connInfo := &regionalapiv1.PrivateLinkService_Connection{}
		connInfo, err := setConnectionProviderInfo(connInfo, consts.ProviderTypeACK, conn, nil)
		assert.NoError(t, err)
		assert.NotNil(t, connInfo.GetAliCloud())
		assert.Equal(t, "pe-ali-1", connInfo.GetAliCloud().VpcEndpointId)
		assert.Equal(t, "default.domain.aliyun", connInfo.GetAliCloud().DefaultDomainName)
		assert.Equal(t, "", connInfo.GetAliCloud().Host)
	})

	// Alicloud connection with host populated
	t.Run("alicloud connection with host", func(t *testing.T) {
		conn := v1alpha1.Connection{
			Endpoint:          "pe-ali-1",
			DefaultDomainName: "default.domain.aliyun",
		}
		connStatus := &v1alpha1.ConnectionStatus{
			Endpoint: "pe-ali-1",
			DNSName:  "conn.tidbcloud.com",
		}
		connInfo := &regionalapiv1.PrivateLinkService_Connection{
			Status: regionalapiv1.PrivateLinkService_Connection_ACTIVE,
		}
		connInfo, err := setConnectionProviderInfo(connInfo, consts.ProviderTypeACK, conn, connStatus)
		assert.NoError(t, err)
		assert.NotNil(t, connInfo.GetAliCloud())
		assert.Equal(t, "pe-ali-1", connInfo.GetAliCloud().VpcEndpointId)
		assert.Equal(t, "default.domain.aliyun", connInfo.GetAliCloud().DefaultDomainName)
		assert.Equal(t, "conn.tidbcloud.com", connInfo.GetAliCloud().Host)
		// Status should remain ACTIVE when host is populated
		assert.Equal(t, regionalapiv1.PrivateLinkService_Connection_ACTIVE, connInfo.Status)
	})

	// Alicloud connection with empty host but status is ACTIVE - should override to PENDING
	t.Run("alicloud connection with empty host status active", func(t *testing.T) {
		conn := v1alpha1.Connection{
			Endpoint:          "pe-ali-2",
			DefaultDomainName: "default.domain.aliyun",
		}
		connStatus := &v1alpha1.ConnectionStatus{
			Endpoint: "pe-ali-2",
			DNSName:  "", // Host is empty
		}
		connInfo := &regionalapiv1.PrivateLinkService_Connection{
			Status: regionalapiv1.PrivateLinkService_Connection_ACTIVE, // Status is ACTIVE
		}
		connInfo, err := setConnectionProviderInfo(connInfo, consts.ProviderTypeACK, conn, connStatus)
		assert.NoError(t, err)
		assert.NotNil(t, connInfo.GetAliCloud())
		assert.Equal(t, "pe-ali-2", connInfo.GetAliCloud().VpcEndpointId)
		assert.Equal(t, "", connInfo.GetAliCloud().Host) // Host is empty
		// Status should be overridden to PENDING when ACTIVE but host is empty
		assert.Equal(t, regionalapiv1.PrivateLinkService_Connection_PENDING, connInfo.Status)
	})

	// Alicloud connection with empty host but status is not ACTIVE - should not override
	t.Run("alicloud connection with empty host status pending", func(t *testing.T) {
		conn := v1alpha1.Connection{
			Endpoint:          "pe-ali-3",
			DefaultDomainName: "default.domain.aliyun",
		}
		connStatus := &v1alpha1.ConnectionStatus{
			Endpoint: "pe-ali-3",
			DNSName:  "", // Host is empty
		}
		connInfo := &regionalapiv1.PrivateLinkService_Connection{
			Status: regionalapiv1.PrivateLinkService_Connection_PENDING, // Status is already PENDING
		}
		connInfo, err := setConnectionProviderInfo(connInfo, consts.ProviderTypeACK, conn, connStatus)
		assert.NoError(t, err)
		assert.NotNil(t, connInfo.GetAliCloud())
		assert.Equal(t, "", connInfo.GetAliCloud().Host) // Host is empty
		// Status should remain PENDING (no change needed)
		assert.Equal(t, regionalapiv1.PrivateLinkService_Connection_PENDING, connInfo.Status)
	})
}

func Test_genPrivateLinkServicePB(t *testing.T) {
	t.Run("aws privatelinkservice", func(t *testing.T) {
		pls := &v1alpha1.PrivateLinkService{
			ObjectMeta: metav1.ObjectMeta{
				Name:      "test-pls",
				Namespace: "test-namespace",
			},
			Spec: v1alpha1.PrivateLinkServiceSpec{
				Provider:         consts.ProviderTypeEKS,
				Region:           "us-west-1",
				LoadBalancerName: "test-lb",
				Connections: []v1alpha1.Connection{
					{Endpoint: "https://test.com"},
				},
			},
			Status: v1alpha1.PrivateLinkServiceStatus{
				EndpointService: v1alpha1.EndpointServiceStatus{
					PrivateDNSName:    "test.dns.com",
					ServiceName:       "test-service",
					AvailabilityZones: []string{"us-west-1a"},
				},
				Connections: []v1alpha1.ConnectionStatus{
					{
						Endpoint: "https://test.com",
						Phase:    v1alpha1.ConnectionPhaseActive,
						Message:  "active",
					},
				},
			},
		}
		plsPB, err := genPrivateLinkServicePB(pls)
		assert.NoError(t, err)
		assert.Equal(t, "test-pls", plsPB.Id)
		assert.Equal(t, "test-lb", plsPB.LoadBalancerName)
		assert.Equal(t, "test.dns.com", plsPB.ServiceDnsName)
		assert.NotNil(t, plsPB.GetAws())
		assert.Equal(t, "test-service", plsPB.GetAws().ServiceName)
		assert.Equal(t, []string{"us-west-1a"}, plsPB.GetAws().AzIds)
		assert.Len(t, plsPB.Connections, 1)
		assert.Equal(t, regionalapiv1.PrivateLinkService_Connection_ACTIVE, plsPB.Connections[0].Status)
	})

	t.Run("gcp privatelinkservice", func(t *testing.T) {
		pls := &v1alpha1.PrivateLinkService{
			ObjectMeta: metav1.ObjectMeta{
				Name:      "test-pls",
				Namespace: "test-namespace",
			},
			Spec: v1alpha1.PrivateLinkServiceSpec{
				Provider:         consts.ProviderTypeGKE,
				Region:           "us-west-1",
				LoadBalancerName: "test-lb",
				Connections: []v1alpha1.Connection{
					{Endpoint: "https://test.com"},
				},
			},
			Status: v1alpha1.PrivateLinkServiceStatus{
				ServiceAttachment: v1alpha1.ServiceAttachmentStatus{
					ConnectDomain:     "test.dns.com",
					ServiceAttachment: "test-service",
				},
				Connections: []v1alpha1.ConnectionStatus{
					{
						Endpoint: "https://test.com",
						Phase:    v1alpha1.ConnectionPhaseActive,
						Message:  "active",
					},
				},
			},
		}
		plsPB, err := genPrivateLinkServicePB(pls)
		assert.NoError(t, err)
		assert.Equal(t, "test-pls", plsPB.Id)
		assert.Equal(t, "test-lb", plsPB.LoadBalancerName)
		assert.Equal(t, "test.dns.com", plsPB.ServiceDnsName)
		assert.NotNil(t, plsPB.GetGcp())
		assert.Equal(t, "test-service", plsPB.GetGcp().ServiceAttachment)
		assert.Len(t, plsPB.Connections, 1)
		assert.Equal(t, regionalapiv1.PrivateLinkService_Connection_ACTIVE, plsPB.Connections[0].Status)
	})

	t.Run("azure privatelinkservice", func(t *testing.T) {
		pls := &v1alpha1.PrivateLinkService{
			ObjectMeta: metav1.ObjectMeta{
				Name:      "test-pls",
				Namespace: "test-namespace",
			},
			Spec: v1alpha1.PrivateLinkServiceSpec{
				Provider:         consts.ProviderTypeAKS,
				Region:           "us-west-1",
				LoadBalancerName: "test-lb",
				Connections: []v1alpha1.Connection{
					{Endpoint: "https://test.com"},
				},
			},
			Status: v1alpha1.PrivateLinkServiceStatus{
				AzurePrivateLinkService: v1alpha1.AzurePrivateLinkServiceStatus{
					ServiceDNSName: "test.dns.com",
					ID:             "test-id",
				},
				Connections: []v1alpha1.ConnectionStatus{
					{
						Endpoint: "https://test.com",
						Phase:    v1alpha1.ConnectionPhaseActive,
						Message:  "active",
					},
				},
			},
		}
		plsPB, err := genPrivateLinkServicePB(pls)
		assert.NoError(t, err)
		assert.Equal(t, "test-pls", plsPB.Id)
		assert.Equal(t, "test-lb", plsPB.LoadBalancerName)
		assert.Equal(t, "test.dns.com", plsPB.ServiceDnsName)
		assert.NotNil(t, plsPB.GetAzure())
		assert.Equal(t, "test-id", plsPB.GetAzure().ServiceResourceId)
		assert.Len(t, plsPB.Connections, 1)
		assert.Equal(t, regionalapiv1.PrivateLinkService_Connection_ACTIVE, plsPB.Connections[0].Status)
	})

	// Alicloud provider
	t.Run("alicloud privatelinkservice", func(t *testing.T) {
		pls := &v1alpha1.PrivateLinkService{
			ObjectMeta: metav1.ObjectMeta{
				Name:      "pls-ali-1",
				Namespace: "test-namespace",
			},
			Spec: v1alpha1.PrivateLinkServiceSpec{
				Provider:         consts.ProviderTypeACK,
				Region:           "cn-hangzhou",
				LoadBalancerName: "lb-ali",
				Connections: []v1alpha1.Connection{
					{Endpoint: "pe-ali-1", DefaultDomainName: "default.domain.aliyun"},
				},
			},
			Status: v1alpha1.PrivateLinkServiceStatus{
				EndpointService: v1alpha1.EndpointServiceStatus{
					PrivateDNSName:    "service.dns.aliyun",
					ServiceName:       "svc-ali",
					AvailabilityZones: []string{"cn-hangzhou-i"},
				},
				Connections: []v1alpha1.ConnectionStatus{
					{Endpoint: "pe-ali-1", Phase: v1alpha1.ConnectionPhaseActive, Message: "ok", DNSName: "conn.tidbcloud.com"},
				},
			},
		}
		plsPB, err := genPrivateLinkServicePB(pls)
		assert.NoError(t, err)
		assert.Equal(t, "pls-ali-1", plsPB.Id)
		assert.Equal(t, "lb-ali", plsPB.LoadBalancerName)
		assert.Equal(t, "service.dns.aliyun", plsPB.ServiceDnsName)
		assert.NotNil(t, plsPB.GetAliCloud())
		assert.Equal(t, "svc-ali", plsPB.GetAliCloud().ServiceName)
		assert.Equal(t, []string{"cn-hangzhou-i"}, plsPB.GetAliCloud().AzIds)
		assert.Len(t, plsPB.Connections, 1)
		connPB := plsPB.Connections[0]
		assert.Equal(t, regionalapiv1.PrivateLinkService_Connection_ACTIVE, connPB.Status)
		assert.NotNil(t, connPB.GetAliCloud())
		assert.Equal(t, "pe-ali-1", connPB.GetAliCloud().VpcEndpointId)
		assert.Equal(t, "default.domain.aliyun", connPB.GetAliCloud().DefaultDomainName)
		assert.Equal(t, "conn.tidbcloud.com", connPB.GetAliCloud().Host)
	})
}

func Test_coverToPrivateLinkServicePBStatus(t *testing.T) {
	t.Run("pending status", func(t *testing.T) {
		status := coverToPrivateLinkServicePBStatus(v1alpha1.PrivateLinkServicePhasePending)
		assert.Equal(t, regionalapiv1.PrivateLinkService_CREATING, status)
	})

	t.Run("active status", func(t *testing.T) {
		status := coverToPrivateLinkServicePBStatus(v1alpha1.PrivateLinkServicePhaseActive)
		assert.Equal(t, regionalapiv1.PrivateLinkService_ACTIVE, status)
	})

	t.Run("deleting status", func(t *testing.T) {
		status := coverToPrivateLinkServicePBStatus(v1alpha1.PrivateLinkServicePhaseDeleting)
		assert.Equal(t, regionalapiv1.PrivateLinkService_DELETING, status)
	})

	t.Run("failed status", func(t *testing.T) {
		status := coverToPrivateLinkServicePBStatus(v1alpha1.PrivateLinkServicePhaseFailed)
		assert.Equal(t, regionalapiv1.PrivateLinkService_CREATING, status)
	})
}

type fakePrivateLinkServiceReaderWriter struct{}

func (f *fakePrivateLinkServiceReaderWriter) CreatePrivateLinkService(ctx context.Context, pls *v1alpha1.PrivateLinkService) error {
	return nil
}

func (f *fakePrivateLinkServiceReaderWriter) Get(ctx context.Context, key client.ObjectKey, pls *v1alpha1.PrivateLinkService) error {
	*pls = v1alpha1.PrivateLinkService{
		Spec: v1alpha1.PrivateLinkServiceSpec{
			LoadBalancerName: "test-pls",
		},
	}
	return nil
}

func (f *fakePrivateLinkServiceReaderWriter) List(ctx context.Context, list *v1alpha1.PrivateLinkServiceList, opts ...client.ListOption) error {
	list.Items = []v1alpha1.PrivateLinkService{}
	return nil
}

func (f *fakePrivateLinkServiceReaderWriter) Create(ctx context.Context, pls *v1alpha1.PrivateLinkService) error {
	return nil
}

func (f *fakePrivateLinkServiceReaderWriter) Update(ctx context.Context, pls *v1alpha1.PrivateLinkService) error {
	return nil
}

func (f *fakePrivateLinkServiceReaderWriter) Delete(ctx context.Context, pls *v1alpha1.PrivateLinkService) error {
	return nil
}

func (f *fakePrivateLinkServiceReaderWriter) Patch(ctx context.Context, pls *v1alpha1.PrivateLinkService, patch client.Patch) error {
	return nil
}

func (f *fakePrivateLinkServiceReaderWriter) UpdateStatus(ctx context.Context, pls *v1alpha1.PrivateLinkService) error {
	return nil
}

type fakePrivateLinkEndpointReaderWriter struct{}

func (f *fakePrivateLinkEndpointReaderWriter) CreatePrivateLinkService(ctx context.Context, ple *v1alpha1.PrivateLinkEndpoint) error {
	return nil
}

func (f *fakePrivateLinkEndpointReaderWriter) Get(ctx context.Context, key client.ObjectKey, ple *v1alpha1.PrivateLinkEndpoint) error {
	*ple = v1alpha1.PrivateLinkEndpoint{
		Spec: v1alpha1.PrivateLinkEndpointSpec{
			//LoadBalancerName: "test-ple",
		},
	}
	return nil
}

func (f *fakePrivateLinkEndpointReaderWriter) List(ctx context.Context, list *v1alpha1.PrivateLinkEndpointList, opts ...client.ListOption) error {
	list.Items = []v1alpha1.PrivateLinkEndpoint{}
	return nil
}

func (f *fakePrivateLinkEndpointReaderWriter) Create(ctx context.Context, ple *v1alpha1.PrivateLinkEndpoint) error {
	return nil
}

func (f *fakePrivateLinkEndpointReaderWriter) Update(ctx context.Context, ple *v1alpha1.PrivateLinkEndpoint) error {
	return nil
}

func (f *fakePrivateLinkEndpointReaderWriter) Delete(ctx context.Context, ple *v1alpha1.PrivateLinkEndpoint) error {
	return nil
}

func (f *fakePrivateLinkEndpointReaderWriter) Patch(ctx context.Context, ple *v1alpha1.PrivateLinkEndpoint, patch client.Patch) error {
	return nil
}

func (f *fakePrivateLinkEndpointReaderWriter) UpdateStatus(ctx context.Context, ple *v1alpha1.PrivateLinkEndpoint) error {
	return nil
}

type fakeVPCReader struct{}

func (f *fakeVPCReader) GetVPC(ctx context.Context, id string) (*v1alpha1.VPC, error) {
	return &v1alpha1.VPC{}, nil
}

func (f *fakeVPCReader) Get(ctx context.Context, key client.ObjectKey, vpc *v1alpha1.VPC) error {
	*vpc = v1alpha1.VPC{
		ObjectMeta: metav1.ObjectMeta{
			Name:      key.Name,
			Namespace: key.Namespace,
		},
		Spec: v1alpha1.VPCSpec{
			CIDR: "10.0.0.0/16",
		},
	}
	return nil
}

func (f *fakeVPCReader) List(ctx context.Context, list *v1alpha1.VPCList, opts ...client.ListOption) error {
	list.Items = []v1alpha1.VPC{}
	return nil
}

type fakeDataPlaneReader struct{}

func (f *fakeDataPlaneReader) GetDataPlane(ctx context.Context, id string) (*v1alpha1.DataPlane, error) {
	return &v1alpha1.DataPlane{
		Spec: v1alpha1.DataPlaneSpec{
			VpcCidr: "10.0.0.0/16",
			AccountRef: v1alpha1.AccountReference{
				AccountID: "test-account",
			},
			AvailabilityZones: []string{"us-west-2a", "us-west-2b"},
		},
	}, nil
}

func (f *fakeDataPlaneReader) Get(ctx context.Context, key client.ObjectKey, dataPlane *v1alpha1.DataPlane) error {
	*dataPlane = v1alpha1.DataPlane{
		ObjectMeta: metav1.ObjectMeta{
			Name:      key.Name,
			Namespace: key.Namespace,
			Labels:    map[string]string{"app": "test"},
		},
		Status: v1alpha1.DataPlaneStatus{
			Vpc: &v1alpha1.Reference{
				Name:      "test-vpc-name",
				Namespace: key.Namespace,
			},
		},
	}
	return nil
}

func (f *fakeDataPlaneReader) List(ctx context.Context, list *v1alpha1.DataPlaneList, opts ...client.ListOption) error {
	list.Items = []v1alpha1.DataPlane{}
	return nil
}

func newMSKEndpointService(t *testing.T, plerw *mockrepository.MockPrivateLinkEndpointReaderWriter) *Service {
	t.Helper()
	return &Service{
		privatelinkendpointRW: plerw,
		dataPlaneR:            &fakeDataPlaneReader{},
		config: config.RootConfig{
			Namespace: "default",
			ControllerConfig: config.ControllerConfig{
				Provider:     consts.ProviderTypeEKS,
				RegionConfig: config.RegionConfig{Name: "us-west-2"},
			},
		},
	}
}

func mskEndpointRequest(auth regionalapiv1.PrivateLinkEndpoint_AuthType) *regionalapiv1.CreatePrivateLinkEndpointRequest {
	return &regionalapiv1.CreatePrivateLinkEndpointRequest{
		DataPlaneId: "dp-1",
		Id:          "ep-msk-1",
		PrivateLinkEndpoint: &regionalapiv1.PrivateLinkEndpoint{
			Tags: map[string]string{consts.TenantLabelKey: "111"},
			Info: &regionalapiv1.PrivateLinkEndpoint_AwsMsk_{
				AwsMsk: &regionalapiv1.PrivateLinkEndpoint_AwsMsk{
					MskArn:   "arn:aws:kafka:us-west-2:111:cluster/demo/abc",
					AuthType: auth,
				},
			},
		},
	}
}

func TestCreatePrivateLinkEndpoint_AmazonMSK(t *testing.T) {
	ctrl := gomock.NewController(t)
	plerw := mockrepository.NewMockPrivateLinkEndpointReaderWriter(ctrl)
	var created *v1alpha1.PrivateLinkEndpoint
	plerw.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(
		func(_ context.Context, ple *v1alpha1.PrivateLinkEndpoint) error {
			created = ple
			return nil
		})
	svc := newMSKEndpointService(t, plerw)

	resp, err := svc.CreatePrivateLinkEndpoint(context.Background(), mskEndpointRequest(regionalapiv1.PrivateLinkEndpoint_SASL_SCRAM))
	require.NoError(t, err)
	require.NotNil(t, created)
	assert.Equal(t, v1alpha1.PrivateLinkEndpointTypeAmazonMSK, created.Spec.Type)
	assert.Equal(t, "arn:aws:kafka:us-west-2:111:cluster/demo/abc", created.Spec.MSKArn)
	assert.Equal(t, v1alpha1.PrivateLinkEndpointAuthTypeSASLScram, created.Spec.AuthType)
	assert.Empty(t, created.Spec.ServiceName, "msk endpoint must not carry an endpoint service name")
	assert.Equal(t, regionalapiv1.PrivateLinkEndpoint_CREATING, resp.Status)
}

func TestCreatePrivateLinkEndpoint_AmazonMSK_RejectsIAM(t *testing.T) {
	ctrl := gomock.NewController(t)
	plerw := mockrepository.NewMockPrivateLinkEndpointReaderWriter(ctrl)
	// Create must never be called when auth is unsupported.
	svc := newMSKEndpointService(t, plerw)

	_, err := svc.CreatePrivateLinkEndpoint(context.Background(), mskEndpointRequest(regionalapiv1.PrivateLinkEndpoint_SASL_IAM))
	require.Error(t, err)
	assert.Contains(t, err.Error(), "not supported")
}

func TestGenPrivateLinkEndpointPB_AmazonMSK(t *testing.T) {
	ple := &v1alpha1.PrivateLinkEndpoint{
		ObjectMeta: metav1.ObjectMeta{Name: "ep-msk-1"},
		Spec: v1alpha1.PrivateLinkEndpointSpec{
			Type:     v1alpha1.PrivateLinkEndpointTypeAmazonMSK,
			MSKArn:   "arn:msk",
			AuthType: v1alpha1.PrivateLinkEndpointAuthTypeSASLScram,
		},
		Status: v1alpha1.PrivateLinkEndpointStatus{
			Phase:            v1alpha1.PrivateLinkEndpointPhaseAvailable,
			VPCConnectionArn: "arn:vpc-conn",
			BootstrapBroker:  "b-1.demo:9196,b-2.demo:9196",
		},
	}
	pb, err := genPrivateLinkEndpointPB(ple)
	require.NoError(t, err)
	require.NotNil(t, pb.GetAwsMsk())
	assert.Equal(t, "arn:msk", pb.GetAwsMsk().MskArn)
	assert.Equal(t, "arn:vpc-conn", pb.GetAwsMsk().VpcConnectionArn)
	assert.Equal(t, "b-1.demo:9196,b-2.demo:9196", pb.GetAwsMsk().BootstrapBroker)
	assert.Equal(t, regionalapiv1.PrivateLinkEndpoint_SASL_SCRAM, pb.GetAwsMsk().AuthType)
	assert.Equal(t, regionalapiv1.PrivateLinkEndpoint_AVAILABLE, pb.Status)
}