package privatelink

import (
	"context"
	"fmt"

	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/emptypb"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"

	regionalapiv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/dataplane-manager/v1"
	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	cloudaliyun "github.com/tidbcloud/infra-provider/regional-server/pkg/cloud/aliyun"
	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/pkg/ownership"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/repository"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/util"
)

type endpointServiceZoneResolver interface {
	Resolve(
		ctx context.Context,
		roleARN string,
		externalID string,
		serviceRegion string,
		serviceName string,
	) ([]*regionalapiv1.EndpointServiceAvailabilityZone, error)
}

type Service struct {
	regionalapiv1.UnimplementedPrivateLinkConnectionServiceServer

	privatelinkserviceRW  repository.PrivateLinkServiceReaderWriter
	privatelinkendpointRW repository.PrivateLinkEndpointReaderWriter
	dataPlaneR            repository.DataPlaneReader
	vpcR                  repository.VPCReader
	config                config.RootConfig
	endpointZoneResolver  endpointServiceZoneResolver
}

func NewService(
	privatelinkserviceRW repository.PrivateLinkServiceReaderWriter,
	privatelinkendpointRW repository.PrivateLinkEndpointReaderWriter,
	dataPlaneR repository.DataPlaneReader,
	vpcR repository.VPCReader,
	config config.RootConfig,
	awsCli cloudaws.Client,
	aliyunCli cloudaliyun.Client,
) *Service {
	var resolver endpointServiceZoneResolver
	switch config.Provider {
	case consts.ProviderTypeAWS, consts.ProviderTypeEKS:
		resolver = newPrivateLinkEndpointServiceZoneResolver(awsCli)
	case consts.ProviderTypeACK:
		resolver = newAliyunPrivateLinkEndpointServiceZoneResolver(aliyunCli)
	}
	return &Service{
		privatelinkserviceRW:  privatelinkserviceRW,
		privatelinkendpointRW: privatelinkendpointRW,
		dataPlaneR:            dataPlaneR,
		vpcR:                  vpcR,
		config:                config,
		endpointZoneResolver:  resolver,
	}
}

func (s *Service) CreatePrivateLinkService(ctx context.Context, req *regionalapiv1.CreatePrivateLinkServiceRequest) (*regionalapiv1.PrivateLinkService, error) {
	tenantID := req.PrivateLinkService.Tags[consts.TenantLabelKey]
	if tenantID == "" {
		return nil, bizerr.New(bizerr.ParameterInvalid, "missing required tag: tags.tidbcloud.com/tenant")
	}
	dataplaneObj, err := s.getDataPlane(ctx, req.DataPlaneId)
	if err != nil {
		return nil, errors.Trace(err)
	}
	annotations := map[string]string{}
	if s.config.Provider == consts.ProviderTypeEKS {
		if aws := req.PrivateLinkService.GetAws(); aws != nil && aws.ServiceName != "" {
			annotations[v1alpha1.AnnotationKeyPrivateLinkServiceName] = aws.ServiceName
		}
	}
	pls := &v1alpha1.PrivateLinkService{
		ObjectMeta: metav1.ObjectMeta{
			Name:        req.Id,
			Namespace:   s.config.Namespace,
			Labels:      util.MergeMaps(dataplaneObj.Labels, s.config.ResourceLabels, map[string]string{v1alpha1.DataPlaneIDKey: req.DataPlaneId}),
			Annotations: annotations,
		},
		Spec: v1alpha1.PrivateLinkServiceSpec{
			Provider: s.config.Provider,
			Region:   s.config.RegionConfig.Name,
			VPCRef: v1alpha1.Reference{
				Name:      dataplaneObj.Status.Vpc.Name,
				Namespace: dataplaneObj.Status.Vpc.Namespace,
			},
			LoadBalancerName: req.PrivateLinkService.LoadBalancerName,
			Tags:             req.PrivateLinkService.Tags,
			Ownership:        ownership.FromRequest(req.OwnershipContext),
		},
	}
	for _, c := range req.PrivateLinkService.Connections {
		var conn v1alpha1.Connection
		switch s.config.Provider {
		case consts.ProviderTypeEKS:
			aws := c.GetAws()
			if aws == nil {
				return nil, bizerr.New(bizerr.ParameterInvalid, "aws connection required")
			}
			conn.Endpoint = aws.VpcEndpointId
		case consts.ProviderTypeGKE:
			gcp := c.GetGcp()
			if gcp == nil {
				return nil, bizerr.New(bizerr.ParameterInvalid, "gcp connection required")
			}
			conn.Endpoint = gcp.PrivateServiceConnectEndpoint
		case consts.ProviderTypeAKS:
			azure := c.GetAzure()
			if azure == nil {
				return nil, bizerr.New(bizerr.ParameterInvalid, "azure connection required")
			}
			conn.Endpoint = azure.PrivateEndpointResourceId
			if ip := azure.PrivateIpAddress; ip != "" {
				conn.IPs = []string{ip}
			}
		case consts.ProviderTypeACK:
			ali := c.GetAliCloud()
			if ali == nil {
				return nil, bizerr.New(bizerr.ParameterInvalid, "ali cloud connection required")
			}
			conn.Endpoint = ali.VpcEndpointId
			conn.DefaultDomainName = ali.DefaultDomainName
		default:
			return nil, bizerr.New(bizerr.InternalServerError, "provider config invalid")
		}
		pls.Spec.Connections = append(pls.Spec.Connections, conn)
	}

	if err := s.privatelinkserviceRW.Create(ctx, pls); err != nil {
		if apierrors.IsAlreadyExists(err) {
			log.Info("privatelinkservice already exists", zap.String("id", req.Id))
			existing := &v1alpha1.PrivateLinkService{}
			if getErr := s.privatelinkserviceRW.Get(ctx, client.ObjectKeyFromObject(pls), existing); getErr != nil {
				return nil, bizerr.New(bizerr.K8SClientError, "failed to get existing privatelinkservice")
			}
			if matchErr := ownership.Match(req.OwnershipContext, existing.Spec.Ownership); matchErr != nil {
				return nil, matchErr
			}
			return req.PrivateLinkService, bizerr.New(bizerr.ResourceConflict, "privatelinkservice already exists")
		}
		log.Error("failed to create privatelinkservice", zap.Error(err), zap.String("id", req.Id))
		return nil, bizerr.New(bizerr.K8SClientError, "failed to create privatelinkservice")
	}

	req.PrivateLinkService.Id = req.Id
	req.PrivateLinkService.Status = regionalapiv1.PrivateLinkService_CREATING
	return req.PrivateLinkService, nil
}

func (s *Service) GetPrivateLinkService(ctx context.Context, req *regionalapiv1.GetPrivateLinkServiceRequest) (*regionalapiv1.PrivateLinkService, error) {
	pls := &v1alpha1.PrivateLinkService{}
	if err := s.privatelinkserviceRW.Get(ctx, client.ObjectKey{Name: req.Id, Namespace: s.config.Namespace}, pls); err != nil {
		if apierrors.IsNotFound(err) {
			log.Info("privatelinkservice not found", zap.String("id", req.Id))
			return nil, bizerr.New(bizerr.ResourceNotFound, "privatelinkservice not found")
		}
		log.Error("failed to get privatelinkservice", zap.Error(err), zap.String("id", req.Id))
		return nil, bizerr.New(bizerr.K8SClientError, "failed to get privatelinkservice")
	}
	if err := ownership.Match(req.OwnershipContext, pls.Spec.Ownership); err != nil {
		return nil, err
	}
	plsPB, err := genPrivateLinkServicePB(pls)
	if err != nil {
		log.Error("failed to fill privatelinkservice", zap.Error(err))
		return nil, bizerr.New(bizerr.InternalServerError, "failed to fill privatelinkservice")
	}

	return plsPB, nil
}

func (s *Service) DeletePrivateLinkService(ctx context.Context, req *regionalapiv1.DeletePrivateLinkServiceRequest) (*emptypb.Empty, error) {
	pls := &v1alpha1.PrivateLinkService{
		ObjectMeta: metav1.ObjectMeta{
			Name:      req.Id,
			Namespace: s.config.Namespace,
		},
	}
	if err := s.privatelinkserviceRW.Get(ctx, client.ObjectKeyFromObject(pls), pls); err != nil {
		if apierrors.IsNotFound(err) {
			log.Info("privatelinkservice not found", zap.String("id", req.Id))
			return new(emptypb.Empty), nil
		}
		return nil, bizerr.New(bizerr.K8SClientError, "failed to get privatelinkservice")
	}
	if err := ownership.Match(req.OwnershipContext, pls.Spec.Ownership); err != nil {
		return nil, err
	}
	if err := s.privatelinkserviceRW.Delete(ctx, pls); err != nil {
		if apierrors.IsNotFound(err) {
			log.Info("privatelinkservice not found", zap.String("id", req.Id))
			return new(emptypb.Empty), nil
		}
		log.Error("failed to delete privatelinkservice", zap.Error(err), zap.String("id", req.Id))
		return nil, bizerr.New(bizerr.K8SClientError, "failed to delete privatelinkservice")
	}
	return new(emptypb.Empty), nil
}

func (s *Service) UpdatePrivateLinkService(ctx context.Context, req *regionalapiv1.UpdatePrivateLinkServiceRequest) (*regionalapiv1.PrivateLinkService, error) {
	pls := &v1alpha1.PrivateLinkService{}
	if err := s.privatelinkserviceRW.Get(ctx, client.ObjectKey{Name: req.Id, Namespace: s.config.Namespace}, pls); err != nil {
		if apierrors.IsNotFound(err) {
			log.Info("privatelinkservice not found", zap.String("id", req.Id))
			return nil, bizerr.New(bizerr.ResourceNotFound, "privatelinkservice not found")
		}
		log.Error("failed to get privatelinkservice", zap.Error(err), zap.String("id", req.Id))
		return nil, bizerr.New(bizerr.K8SClientError, "failed to get privatelinkservice")
	}
	if err := ownership.Match(req.OwnershipContext, pls.Spec.Ownership); err != nil {
		return nil, err
	}

	conns := make([]v1alpha1.Connection, 0, len(req.PrivateLinkConnections))
	for _, c := range req.PrivateLinkConnections {
		conn := v1alpha1.Connection{}
		switch s.config.Provider {
		case consts.ProviderTypeEKS:
			if c.GetAws() == nil {
				return nil, bizerr.New(bizerr.ParameterInvalid, "aws connection required")
			}
			conn.Endpoint = c.GetAws().VpcEndpointId
		case consts.ProviderTypeGKE:
			if c.GetGcp() == nil {
				return nil, bizerr.New(bizerr.ParameterInvalid, "gcp connection required")
			}
			conn.Endpoint = c.GetGcp().PrivateServiceConnectEndpoint
		case consts.ProviderTypeAKS:
			if c.GetAzure() == nil {
				return nil, bizerr.New(bizerr.ParameterInvalid, "azure connection required")
			}
			conn = v1alpha1.Connection{
				Endpoint: c.GetAzure().PrivateEndpointResourceId,
				IPs:      []string{c.GetAzure().PrivateIpAddress},
			}
		case consts.ProviderTypeACK:
			ali := c.GetAliCloud()
			if ali == nil {
				return nil, bizerr.New(bizerr.ParameterInvalid, "ali cloud connection required")
			}
			conn.Endpoint = ali.VpcEndpointId
			conn.DefaultDomainName = ali.DefaultDomainName
		default:
			return nil, bizerr.New(bizerr.InternalServerError, "provider config invalid")
		}
		conns = append(conns, conn)
	}
	pls.Spec.Connections = conns

	if err := s.privatelinkserviceRW.Update(ctx, pls); err != nil {
		log.Error("failed to update privatelinkservice", zap.Error(err))
		return nil, bizerr.New(bizerr.K8SClientError, "failed to update privatelinkservice")
	}
	plsPB, err := genPrivateLinkServicePB(pls)
	if err != nil {
		log.Error("failed to fill privatelinkservice", zap.Error(err))
		return nil, bizerr.New(bizerr.InternalServerError, "failed to fill privatelinkservice")
	}
	plsPB.Status = regionalapiv1.PrivateLinkService_CREATING
	return plsPB, nil
}

func (s *Service) getDataPlane(ctx context.Context, dataplaneId string) (*v1alpha1.DataPlane, error) {
	dataPlaneObj := &v1alpha1.DataPlane{}
	if err := s.dataPlaneR.Get(ctx,
		client.ObjectKey{
			Name:      dataplaneId,
			Namespace: s.config.Namespace,
		},
		dataPlaneObj); err != nil {
		log.Error("failed to get dataplane", zap.Error(err))
		return nil, bizerr.New(bizerr.K8SClientError, fmt.Sprintf("failed get dataplane:%s", dataplaneId))
	}
	return dataPlaneObj, nil
}

func genPrivateLinkServicePB(pls *v1alpha1.PrivateLinkService) (*regionalapiv1.PrivateLinkService, error) {
	plsPB := &regionalapiv1.PrivateLinkService{
		Id:               pls.Name,
		LoadBalancerName: pls.Spec.LoadBalancerName,
		Tags:             pls.Spec.Tags,
	}

	switch pls.Spec.Provider {
	case consts.ProviderTypeEKS:
		plsPB.ServiceDnsName = pls.Status.EndpointService.PrivateDNSName
		plsPB.Info = &regionalapiv1.PrivateLinkService_Aws_{
			Aws: &regionalapiv1.PrivateLinkService_Aws{
				ServiceName: pls.Status.EndpointService.ServiceName,
				AzIds:       pls.Status.EndpointService.AvailabilityZones,
			},
		}
	case consts.ProviderTypeGKE:
		plsPB.ServiceDnsName = pls.Status.ServiceAttachment.ConnectDomain
		plsPB.Info = &regionalapiv1.PrivateLinkService_Gcp_{
			Gcp: &regionalapiv1.PrivateLinkService_Gcp{
				ServiceAttachment: pls.Status.ServiceAttachment.ServiceAttachment,
			},
		}
	case consts.ProviderTypeAKS:
		plsPB.ServiceDnsName = pls.Status.AzurePrivateLinkService.ServiceDNSName
		plsPB.Info = &regionalapiv1.PrivateLinkService_Azure_{
			Azure: &regionalapiv1.PrivateLinkService_Azure{
				ServiceResourceId: pls.Status.AzurePrivateLinkService.ID,
			},
		}
	case consts.ProviderTypeACK:
		plsPB.ServiceDnsName = pls.Status.EndpointService.PrivateDNSName
		plsPB.Info = &regionalapiv1.PrivateLinkService_AliCloud_{
			AliCloud: &regionalapiv1.PrivateLinkService_AliCloud{
				ServiceName: pls.Status.EndpointService.ServiceName,
				AzIds:       pls.Status.EndpointService.AvailabilityZones,
			},
		}
	default:
		return nil, bizerr.New(bizerr.InternalServerError, "provider config invalid")
	}
	endpointMap := make(map[string]v1alpha1.Connection)
	for _, c := range pls.Spec.Connections {
		endpointMap[c.Endpoint] = c
		connInfo := &regionalapiv1.PrivateLinkService_Connection{
			Status: regionalapiv1.PrivateLinkService_Connection_STATE_UNSPECIFIED,
		}
		connInfo, err := setConnectionProviderInfo(connInfo, pls.Spec.Provider, c, nil)
		if err != nil {
			return nil, err
		}
		for _, cs := range pls.Status.Connections {
			if c.Endpoint == cs.Endpoint {
				connInfo, err = genConnectionPB(pls.Spec.Provider, c, ptr.To(cs))
				if err != nil {
					return nil, err
				}
				break
			}
		}
		plsPB.Connections = append(plsPB.Connections, connInfo)
	}

	// only AKS need to add pending connections which are not in allowed list
	if pls.Spec.Provider == consts.ProviderTypeAKS {
		for _, connStatus := range pls.Status.Connections {
			if connStatus.Phase != v1alpha1.ConnectionPhasePending {
				continue
			}
			// skip if connection is already in allowed list, because it's already added on above
			if _, ok := endpointMap[connStatus.Endpoint]; ok {
				continue
			}
			connInfo, err := genConnectionPB(pls.Spec.Provider, v1alpha1.Connection{Endpoint: connStatus.Endpoint}, ptr.To(connStatus))
			if err != nil {
				return nil, err
			}
			plsPB.Connections = append(plsPB.Connections, connInfo)
		}
	}

	plsPB.Status = coverToPrivateLinkServicePBStatus(pls.Status.Phase)
	return plsPB, nil
}

func coverToPrivateLinkServicePBStatus(phase v1alpha1.PrivateLinkServicePhase) regionalapiv1.PrivateLinkService_Status {
	status := regionalapiv1.PrivateLinkService_STATE_UNSPECIFIED
	switch phase {
	case "", v1alpha1.PrivateLinkServicePhasePending,
		v1alpha1.PrivateLinkServicePhaseReadyForCreate,
		v1alpha1.PrivateLinkServicePhaseCreating,
		v1alpha1.PrivateLinkServicePhaseFailed:
		status = regionalapiv1.PrivateLinkService_CREATING
	case v1alpha1.PrivateLinkServicePhaseActive:
		status = regionalapiv1.PrivateLinkService_ACTIVE
	case v1alpha1.PrivateLinkServicePhaseDeleting,
		v1alpha1.PrivateLinkServicePhaseDeleted,
		v1alpha1.PrivateLinkServicePhaseReadyForDelete:
		status = regionalapiv1.PrivateLinkService_DELETING
	}
	return status
}

func genConnectionPB(provider string, conn v1alpha1.Connection, connStatus *v1alpha1.ConnectionStatus) (*regionalapiv1.PrivateLinkService_Connection, error) {
	msg := ""
	if connStatus != nil {
		msg = connStatus.Message
	}
	info := &regionalapiv1.PrivateLinkService_Connection{
		Status:  regionalapiv1.PrivateLinkService_Connection_STATE_UNSPECIFIED,
		Message: msg,
	}
	if connStatus != nil {
		switch connStatus.Phase {
		case v1alpha1.ConnectionPhasePending:
			info.Status = regionalapiv1.PrivateLinkService_Connection_PENDING
		case v1alpha1.ConnectionPhaseActive:
			info.Status = regionalapiv1.PrivateLinkService_Connection_ACTIVE
		case v1alpha1.ConnectionPhaseFailed:
			info.Status = regionalapiv1.PrivateLinkService_Connection_FAILED
		case v1alpha1.ConnectionPhaseDeleting:
			info.Status = regionalapiv1.PrivateLinkService_Connection_DELETING
		}
	}

	info, err := setConnectionProviderInfo(info, provider, conn, connStatus)
	if err != nil {
		return nil, err
	}

	return info, nil
}

func setConnectionProviderInfo(info *regionalapiv1.PrivateLinkService_Connection, provider string, conn v1alpha1.Connection, connStatus *v1alpha1.ConnectionStatus) (*regionalapiv1.PrivateLinkService_Connection, error) {
	switch provider {
	case consts.ProviderTypeEKS:
		info.Info = &regionalapiv1.PrivateLinkService_Connection_Aws_{
			Aws: &regionalapiv1.PrivateLinkService_Connection_Aws{
				VpcEndpointId: conn.Endpoint,
			},
		}
	case consts.ProviderTypeGKE:
		host := ""
		if connStatus != nil {
			host = connStatus.DNSName
		}
		info.Info = &regionalapiv1.PrivateLinkService_Connection_Gcp_{
			Gcp: &regionalapiv1.PrivateLinkService_Connection_Gcp{
				PrivateServiceConnectEndpoint: conn.Endpoint,
				Host:                          host,
			},
		}
	case consts.ProviderTypeAKS:
		host, address := "", ""
		if connStatus != nil {
			host = connStatus.DNSName
		}
		if len(conn.IPs) > 0 {
			address = conn.IPs[0]
		}
		info.Info = &regionalapiv1.PrivateLinkService_Connection_Azure_{
			Azure: &regionalapiv1.PrivateLinkService_Connection_Azure{
				PrivateEndpointResourceId: conn.Endpoint,
				Host:                      host,
				PrivateIpAddress:          address,
			},
		}
	case consts.ProviderTypeACK:
		var host string
		if connStatus != nil {
			host = connStatus.DNSName
		}
		info.Info = &regionalapiv1.PrivateLinkService_Connection_AliCloud_{
			AliCloud: &regionalapiv1.PrivateLinkService_Connection_AliCloud{
				VpcEndpointId:     conn.Endpoint,
				DefaultDomainName: conn.DefaultDomainName,
				Host:              host,
			},
		}
		// for ACK, if a connection's host is empty, set status to Pending
		// because we will create a CNAME record and set it to host after it to be ready.
		if info.Status == regionalapiv1.PrivateLinkService_Connection_ACTIVE && host == "" {
			info.Status = regionalapiv1.PrivateLinkService_Connection_PENDING
		}
	default:
		return nil, bizerr.New(bizerr.InternalServerError, "provider config invalid")
	}
	return info, nil
}

func (s *Service) CreatePrivateLinkEndpoint(ctx context.Context, req *regionalapiv1.CreatePrivateLinkEndpointRequest) (*regionalapiv1.PrivateLinkEndpoint, error) {
	tenantID := req.PrivateLinkEndpoint.Tags[consts.TenantLabelKey]
	if tenantID == "" {
		return nil, bizerr.New(bizerr.ParameterInvalid, "missing required tag: tags.tidbcloud.com/tenant")
	}
	dataplaneObj, err := s.getDataPlane(ctx, req.DataPlaneId)
	if err != nil {
		return nil, errors.Trace(err)
	}
	ple := &v1alpha1.PrivateLinkEndpoint{
		ObjectMeta: metav1.ObjectMeta{
			Name:      req.Id,
			Namespace: s.config.Namespace,
			Labels:    util.MergeMaps(dataplaneObj.Labels, s.config.ResourceLabels, map[string]string{v1alpha1.DataPlaneIDKey: req.DataPlaneId}),
		},
		Spec: v1alpha1.PrivateLinkEndpointSpec{
			Provider: s.config.Provider,
			Region:   s.config.RegionConfig.Name,
			VPCRef: v1alpha1.Reference{
				Name:      dataplaneObj.Status.Vpc.Name,
				Namespace: dataplaneObj.Status.Vpc.Namespace,
			},
			RootDomainName: req.PrivateLinkEndpoint.RootDomainName,
			Tags:           req.PrivateLinkEndpoint.Tags,
			Ownership:      ownership.FromRequest(req.OwnershipContext),
		},
	}

	// Amazon MSK multi-VPC connectivity: only the MSK ARN + auth scheme are needed,
	// the customer endpoint service / DNS records do not apply.
	if msk := req.PrivateLinkEndpoint.GetAwsMsk(); msk != nil {
		if s.config.Provider != consts.ProviderTypeEKS {
			return nil, bizerr.New(bizerr.ParameterInvalid, "amazon msk private link is only supported on AWS")
		}
		authType := mskAuthTypeFromPB(msk.AuthType)
		if authType != v1alpha1.PrivateLinkEndpointAuthTypeSASLScram {
			return nil, bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("amazon msk auth type %s is not supported yet", msk.AuthType.String()))
		}
		ple.Spec.Type = v1alpha1.PrivateLinkEndpointTypeAmazonMSK
		ple.Spec.MSKArn = msk.MskArn
		ple.Spec.AuthType = authType
		if err := s.privatelinkendpointRW.Create(ctx, ple); err != nil {
			if apierrors.IsAlreadyExists(err) {
				log.Info("privatelinkendpoint already exists", zap.String("id", req.Id))
				if matchErr := s.matchPrivateLinkEndpointOwnership(ctx, req, ple); matchErr != nil {
					return nil, matchErr
				}
				return req.PrivateLinkEndpoint, bizerr.New(bizerr.ResourceConflict, "privatelinkendpoint already exists")
			}
			log.Error("failed to create privatelinkendpoint", zap.Error(err), zap.String("id", req.Id))
			return nil, bizerr.New(bizerr.K8SClientError, "failed to create privatelinkendpoint")
		}
		req.PrivateLinkEndpoint.Id = req.Id
		req.PrivateLinkEndpoint.Status = regionalapiv1.PrivateLinkEndpoint_CREATING
		return req.PrivateLinkEndpoint, nil
	}

	ple.Spec.Type = v1alpha1.PrivateLinkEndpointTypeVPCEndpointService
	switch s.config.Provider {
	case consts.ProviderTypeEKS:
		info := req.PrivateLinkEndpoint.GetAws()
		if info == nil {
			return nil, bizerr.New(bizerr.ParameterInvalid, "aws parameter required")
		}
		ple.Spec.ServiceRegion = info.ServiceRegion
		ple.Spec.ServiceName = info.EndpointServiceName
		ple.Spec.DNSRecords = map[string]string{}
		for _, record := range info.DnsRecords {
			ple.Spec.DNSRecords[record.AvailabilityZone] = record.RecordName
		}
		ple.Spec.Zones = info.AzIds
	case consts.ProviderTypeAKS:
		info := req.PrivateLinkEndpoint.GetAzure()
		if info == nil {
			return nil, bizerr.New(bizerr.ParameterInvalid, "azure parameter required")
		}
		ple.Spec.ServiceName = info.EndpointServiceAlias
	case consts.ProviderTypeACK:
		info := req.PrivateLinkEndpoint.GetAliCloud()
		if info == nil {
			return nil, bizerr.New(bizerr.ParameterInvalid, "alicloud parameter required")
		}
		ple.Spec.ServiceName = info.EndpointServiceName
		ple.Spec.DNSRecords = map[string]string{}
		for _, record := range info.DnsRecords {
			ple.Spec.DNSRecords[record.AvailabilityZone] = record.RecordName
		}
		ple.Spec.Zones = info.AzIds
	default:
		return nil, bizerr.New(bizerr.InternalServerError, "provider config invalid")
	}
	if err := s.privatelinkendpointRW.Create(ctx, ple); err != nil {
		if apierrors.IsAlreadyExists(err) {
			log.Info("privatelinkendpoint already exists", zap.String("id", req.Id))
			if matchErr := s.matchPrivateLinkEndpointOwnership(ctx, req, ple); matchErr != nil {
				return nil, matchErr
			}
			return req.PrivateLinkEndpoint, bizerr.New(bizerr.ResourceConflict, "privatelinkendpoint already exists")
		}
		log.Error("failed to create privatelinkendpoint", zap.Error(err), zap.String("id", req.Id))
		return nil, bizerr.New(bizerr.K8SClientError, "failed to create privatelinkendpoint")
	}
	req.PrivateLinkEndpoint.Id = req.Id
	req.PrivateLinkEndpoint.Status = regionalapiv1.PrivateLinkEndpoint_CREATING
	return req.PrivateLinkEndpoint, nil
}

func (s *Service) ValidatePrivateLinkEndpointZones(ctx context.Context, req *regionalapiv1.ValidatePrivateLinkEndpointZonesRequest) (*emptypb.Empty, error) {
	dataplaneObj, err := s.getDataPlane(ctx, req.DataPlaneId)
	if err != nil {
		return nil, errors.Trace(err)
	}
	vpcObj := &v1alpha1.VPC{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: dataplaneObj.Status.Vpc.Namespace,
			Name:      dataplaneObj.Status.Vpc.Name,
		},
	}
	if err := s.vpcR.Get(ctx, client.ObjectKeyFromObject(vpcObj), vpcObj); err != nil {
		return nil, errors.Trace(err)
	}

	subnets := vpcObj.Status.NodeSubnets
	useEndpointSubnets := util.ShouldUsePrivateLinkEndpointSubnets(vpcObj.Spec.PrivateLinkEndpointZones, vpcObj.Spec.NodeZones)
	if useEndpointSubnets {
		if len(vpcObj.Status.PrivateLinkEndpointSubnets) == 0 {
			return nil, bizerr.New(bizerr.ResourceNotReady, "private link endpoint subnets are not ready")
		}
		subnets = vpcObj.Status.PrivateLinkEndpointSubnets
	}
	availableZones := make(map[string]string, len(subnets))
	for _, subnet := range subnets {
		availableZones[subnet.AvailabilityZoneID] = subnet.AvailabilityZone
	}
	for _, zone := range req.Zones {
		if _, ok := availableZones[zone]; !ok {
			if useEndpointSubnets {
				return nil, bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("zone %s does not match with private link endpoint subnets", zone))
			}
			return nil, bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("zone %s does not match with tidb zone", zone))
		}
	}
	// TODO: Check customer provided zones match endpoint service
	return &emptypb.Empty{}, nil
}

func (s *Service) DescribePrivateLinkEndpointServiceZones(ctx context.Context, req *regionalapiv1.DescribePrivateLinkEndpointServiceZonesRequest) (*regionalapiv1.DescribePrivateLinkEndpointServiceZonesResponse, error) {
	if s.config.Provider != consts.ProviderTypeAWS &&
		s.config.Provider != consts.ProviderTypeEKS &&
		s.config.Provider != consts.ProviderTypeACK {
		return nil, bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("provider %s does not support private link endpoint service zone discovery", s.config.Provider))
	}

	dataplaneObj, err := s.getDataPlane(ctx, req.DataPlaneId)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if dataplaneObj.Spec.AccountRef.AccountID == "" || dataplaneObj.Spec.AccountRef.Role == "" {
		return nil, bizerr.New(bizerr.ParameterInvalid, "dataplane account id and role are required")
	}

	resolver := s.endpointZoneResolver
	if resolver == nil {
		return nil, bizerr.New(bizerr.InternalServerError, "private link endpoint service zone resolver is not configured")
	}
	var roleARN string
	var externalID string
	switch s.config.Provider {
	case consts.ProviderTypeAWS, consts.ProviderTypeEKS:
		roleARN = util.GetRoleARN(dataplaneObj.Spec.AccountRef.AccountID, dataplaneObj.Spec.AccountRef.Role)
		externalID = s.config.AWSDataPlaneExternalID
	case consts.ProviderTypeACK:
		roleARN = util.GetRAMRoleARN(dataplaneObj.Spec.AccountRef.AccountID, dataplaneObj.Spec.AccountRef.Role)
	}
	zones, err := resolver.Resolve(ctx, roleARN, externalID, req.ServiceRegion, req.ServiceName)
	if err != nil {
		return nil, errors.Trace(err)
	}
	return &regionalapiv1.DescribePrivateLinkEndpointServiceZonesResponse{Zones: zones}, nil
}

func (s *Service) GetPrivateLinkEndpoint(ctx context.Context, req *regionalapiv1.GetPrivateLinkEndpointRequest) (*regionalapiv1.PrivateLinkEndpoint, error) {
	ple := &v1alpha1.PrivateLinkEndpoint{}
	if err := s.privatelinkendpointRW.Get(ctx, client.ObjectKey{Name: req.Id, Namespace: s.config.Namespace}, ple); err != nil {
		if apierrors.IsNotFound(err) {
			log.Info("privatelinkendpoint not found", zap.String("id", req.Id))
			return nil, bizerr.New(bizerr.ResourceNotFound, "privatelinkendpoint not found")
		}
		log.Error("failed to get privatelinkendpoint", zap.Error(err), zap.String("id", req.Id))
		return nil, bizerr.New(bizerr.K8SClientError, "failed to get privatelinkendpoint")
	}
	if err := ownership.Match(req.OwnershipContext, ple.Spec.Ownership); err != nil {
		return nil, err
	}
	plePB, err := genPrivateLinkEndpointPB(ple)
	if err != nil {
		log.Error("failed to fill privatelinkendpoint", zap.Error(err))
		return nil, bizerr.New(bizerr.InternalServerError, "failed to fill privatelinkendpoint")
	}
	return plePB, nil
}

func (s *Service) ListPrivateLinkEndpoints(ctx context.Context, req *regionalapiv1.ListPrivateLinkEndpointsRequest) (*regionalapiv1.ListPrivateLinkEndpointsResponse, error) {
	endpointList := &v1alpha1.PrivateLinkEndpointList{}
	selector := client.MatchingLabels{
		v1alpha1.DataPlaneIDKey: req.DataPlaneId,
	}
	if err := s.privatelinkendpointRW.List(ctx, endpointList, selector); err != nil {
		return nil, bizerr.New(bizerr.K8SClientError, "failed to list privatelinkendpoint")
	}
	ples := make([]*regionalapiv1.PrivateLinkEndpoint, len(endpointList.Items))
	for idx, ep := range endpointList.Items {
		ple, err := genPrivateLinkEndpointPB(&ep)
		if err != nil {
			log.Error("failed to build privatelinkendpoint", zap.Error(err))
			continue
		}
		ples[idx] = ple
	}
	return &regionalapiv1.ListPrivateLinkEndpointsResponse{PrivateLinkEndpoints: ples}, nil
}

func (s *Service) DeletePrivateLinkEndpoint(ctx context.Context, req *regionalapiv1.DeletePrivateLinkEndpointRequest) (*emptypb.Empty, error) {
	ple := &v1alpha1.PrivateLinkEndpoint{
		ObjectMeta: metav1.ObjectMeta{
			Name:      req.Id,
			Namespace: s.config.Namespace,
		},
	}
	if err := s.privatelinkendpointRW.Get(ctx, client.ObjectKeyFromObject(ple), ple); err != nil {
		if apierrors.IsNotFound(err) {
			log.Info("privatelinkendpoint not found", zap.String("id", req.Id))
			return new(emptypb.Empty), nil
		}
		return nil, bizerr.New(bizerr.K8SClientError, "failed to get privatelinkendpoint")
	}
	if err := ownership.Match(req.OwnershipContext, ple.Spec.Ownership); err != nil {
		return nil, err
	}
	if err := s.privatelinkendpointRW.Delete(ctx, ple); err != nil {
		if apierrors.IsNotFound(err) {
			log.Info("privatelinkendpoint not found", zap.String("id", req.Id))
			return new(emptypb.Empty), nil
		}
		log.Error("failed to delete privatelinkendpoint", zap.Error(err), zap.String("id", req.Id))
		return nil, bizerr.New(bizerr.K8SClientError, "failed to delete privatelinkendpoint")
	}
	return new(emptypb.Empty), nil
}

func (s *Service) matchPrivateLinkEndpointOwnership(
	ctx context.Context,
	req *regionalapiv1.CreatePrivateLinkEndpointRequest,
	desired *v1alpha1.PrivateLinkEndpoint,
) error {
	existing := &v1alpha1.PrivateLinkEndpoint{}
	if err := s.privatelinkendpointRW.Get(ctx, client.ObjectKeyFromObject(desired), existing); err != nil {
		return bizerr.New(bizerr.K8SClientError, "failed to get existing privatelinkendpoint")
	}
	return ownership.Match(req.OwnershipContext, existing.Spec.Ownership)
}

func genPrivateLinkEndpointPB(ple *v1alpha1.PrivateLinkEndpoint) (*regionalapiv1.PrivateLinkEndpoint, error) {
	plePB := &regionalapiv1.PrivateLinkEndpoint{
		Id:   ple.Name,
		Tags: ple.Spec.Tags,
	}
	if ple.Spec.Type == v1alpha1.PrivateLinkEndpointTypeAmazonMSK {
		plePB.Info = &regionalapiv1.PrivateLinkEndpoint_AwsMsk_{
			AwsMsk: &regionalapiv1.PrivateLinkEndpoint_AwsMsk{
				MskArn:           ple.Spec.MSKArn,
				AuthType:         mskAuthTypeToPB(ple.Spec.AuthType),
				VpcConnectionArn: ple.Status.VPCConnectionArn,
				BootstrapBroker:  ple.Status.BootstrapBroker,
			},
		}
		plePB.Status = endpointPhaseToPB(ple.Status.Phase)
		return plePB, nil
	}
	switch ple.Spec.Provider {
	case consts.ProviderTypeEKS:
		records := make([]*regionalapiv1.PrivateLinkEndpoint_DNSRecord, 0, len(ple.Spec.DNSRecords))
		for az, name := range ple.Spec.DNSRecords {
			records = append(records, &regionalapiv1.PrivateLinkEndpoint_DNSRecord{
				AvailabilityZone: az,
				RecordName:       name,
			})
		}
		plePB.Info = &regionalapiv1.PrivateLinkEndpoint_Aws_{
			Aws: &regionalapiv1.PrivateLinkEndpoint_Aws{
				EndpointId:          ple.Status.ID,
				EndpointServiceName: ple.Spec.ServiceName,
				ServiceRegion:       ple.Spec.ServiceRegion,
				DnsRecords:          records,
				AzIds:               ple.Spec.Zones,
				EndpointAddress:     ple.Status.EndpointAddress,
			},
		}
	case consts.ProviderTypeAKS:
		plePB.Info = &regionalapiv1.PrivateLinkEndpoint_Azure_{
			Azure: &regionalapiv1.PrivateLinkEndpoint_Azure{
				EndpointId:           ple.Status.ID,
				EndpointServiceAlias: ple.Spec.ServiceName,
				EndpointAddress:      ple.Status.EndpointAddress,
			},
		}
	case consts.ProviderTypeACK:
		records := make([]*regionalapiv1.PrivateLinkEndpoint_DNSRecord, 0, len(ple.Spec.DNSRecords))
		for az, name := range ple.Spec.DNSRecords {
			records = append(records, &regionalapiv1.PrivateLinkEndpoint_DNSRecord{
				AvailabilityZone: az,
				RecordName:       name,
			})
		}
		plePB.Info = &regionalapiv1.PrivateLinkEndpoint_AliCloud_{
			AliCloud: &regionalapiv1.PrivateLinkEndpoint_AliCloud{
				EndpointId:          ple.Status.ID,
				EndpointServiceName: ple.Spec.ServiceName,
				DnsRecords:          records,
				AzIds:               ple.Spec.Zones,
				EndpointAddress:     ple.Status.EndpointAddress,
			},
		}
	default:
		return nil, bizerr.New(bizerr.InternalServerError, "unsupported provider type")
	}
	plePB.Status = endpointPhaseToPB(ple.Status.Phase)
	return plePB, nil
}

func endpointPhaseToPB(phase v1alpha1.PrivateLinkEndpointPhase) regionalapiv1.PrivateLinkEndpoint_Status {
	switch phase {
	case v1alpha1.PrivateLinkEndpointPhaseCreating:
		return regionalapiv1.PrivateLinkEndpoint_CREATING
	case v1alpha1.PrivateLinkEndpointPhasePendingAcceptance:
		return regionalapiv1.PrivateLinkEndpoint_PENDING
	case v1alpha1.PrivateLinkEndpointPhasePending, v1alpha1.PrivateLinkEndpointPhaseEndpointAvailable:
		return regionalapiv1.PrivateLinkEndpoint_PENDING
	case v1alpha1.PrivateLinkEndpointPhaseAvailable:
		return regionalapiv1.PrivateLinkEndpoint_AVAILABLE
	case v1alpha1.PrivateLinkEndpointPhaseDeleting:
		return regionalapiv1.PrivateLinkEndpoint_DELETING
	case v1alpha1.PrivateLinkEndpointPhaseFailed:
		return regionalapiv1.PrivateLinkEndpoint_FAILED
	case v1alpha1.PrivateLinkEndpointPhaseRejected:
		return regionalapiv1.PrivateLinkEndpoint_REJECTED
	default:
		return regionalapiv1.PrivateLinkEndpoint_STATE_UNSPECIFIED
	}
}

func mskAuthTypeToPB(t v1alpha1.PrivateLinkEndpointAuthType) regionalapiv1.PrivateLinkEndpoint_AuthType {
	switch t {
	case v1alpha1.PrivateLinkEndpointAuthTypeSASLScram:
		return regionalapiv1.PrivateLinkEndpoint_SASL_SCRAM
	case v1alpha1.PrivateLinkEndpointAuthTypeSASLIAM:
		return regionalapiv1.PrivateLinkEndpoint_SASL_IAM
	default:
		return regionalapiv1.PrivateLinkEndpoint_AUTH_TYPE_UNSPECIFIED
	}
}

func mskAuthTypeFromPB(t regionalapiv1.PrivateLinkEndpoint_AuthType) v1alpha1.PrivateLinkEndpointAuthType {
	switch t {
	case regionalapiv1.PrivateLinkEndpoint_SASL_SCRAM:
		return v1alpha1.PrivateLinkEndpointAuthTypeSASLScram
	case regionalapiv1.PrivateLinkEndpoint_SASL_IAM:
		return v1alpha1.PrivateLinkEndpointAuthTypeSASLIAM
	default:
		return ""
	}
}