package cluster

import (
	"encoding/json"
	"fmt"
	"reflect"
	"strconv"
	"strings"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/ec2"
	ec2types "github.com/aws/aws-sdk-go-v2/service/ec2/types"
	"github.com/aws/aws-sdk-go-v2/service/resourcegroupstaggingapi"
	rgtatypes "github.com/aws/aws-sdk-go-v2/service/resourcegroupstaggingapi/types"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	"github.com/crossplane/crossplane-runtime/pkg/meta"
	ec2v1alpha1 "github.com/crossplane/provider-aws/apis/ec2/v1alpha1"
	route53v1alpha1 "github.com/crossplane/provider-aws/apis/route53/v1alpha1"
	"github.com/pingcap/errors"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/shared-provider/pkg/types"
	"github.com/tidbcloud/shared-provider/pkg/utils"
	"github.com/tidbcloud/shared-provider/pkg/utils/labels"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	annoKeyPrivateLinkSpec   = "alpha.privatelink.tidbcloud.com/spec"
	annoKeyPrivateLinkStatus = "alpha.privatelink.tidbcloud.com/status"
	// NOTE(liubo02): this field can be deleted if https://github.com/crossplane/provider-aws/issues/1179 fixed
	annoKeyPrivateLinkDNSStatus = "alpha.privatelink.tidbcloud.com/dns-status"

	defaultDNSTTLSeconds = 300

	PrivateLinkPhaseReady = "Ready"
)

type PrivateLinkSpec struct {
	AllowedPrincipals   []string `json:"allowedPrincipals"`
	AllowedVPCEndpoints []string `json:"allowedVPCEndpoints"`
	AcceptanceRequired  bool     `json:"acceptanceRequired"`
}

type PrivateLinkStatus struct {
	Phase string `json:"phase"`
	// ServiceName for creating vpc endpoint
	PrivateLinkServiceName string                  `json:"privateLinkServiceName"`
	KubeService            string                  `json:"kubeService"`
	AllowedPrincipals      []string                `json:"allowedPrincipals"`
	Connections            []PrivateLinkConnection `json:"connections"`
}

type PrivateLinkConnection struct {
	VPCEndpointID string `json:"vpcEndpointID"`
	State         string `json:"state"`
	Owner         string `json:"owner"`
}

func GetPrivateLinkSpec(c *cloudv1.Cluster) (*PrivateLinkSpec, error) {
	if c.Annotations == nil {
		return nil, nil
	}
	s, ok := c.Annotations[annoKeyPrivateLinkSpec]
	if !ok {
		return nil, nil
	}

	spec := PrivateLinkSpec{}
	if err := json.Unmarshal([]byte(s), &spec); err != nil {
		return nil, fmt.Errorf("failed to get private link spec: %w", err)
	}

	return &spec, nil
}

func GetPrivateLinkStatus(c *cloudv1.Cluster) (*PrivateLinkStatus, error) {
	if c.Annotations == nil {
		return nil, nil
	}
	s, ok := c.Annotations[annoKeyPrivateLinkStatus]
	if !ok {
		return nil, nil
	}

	status := PrivateLinkStatus{}
	if err := json.Unmarshal([]byte(s), &status); err != nil {
		return nil, fmt.Errorf("failed to get private link status: %w", err)
	}

	return &status, nil
}

func SetPrivateLinkStatus(c *cloudv1.Cluster, s *PrivateLinkStatus) error {
	if c.Annotations == nil {
		c.Annotations = map[string]string{}
	}
	data, err := json.Marshal(s)
	if err != nil {
		return fmt.Errorf("can't set private link status: %w", err)
	}
	c.Annotations[annoKeyPrivateLinkStatus] = string(data)

	return nil
}

const (
	LabelKeyAWSLoadBalancerResource = "service.k8s.aws/resource"
	LabelValAWSLoadBalancerResource = "LoadBalancer"

	LabelKeyAWSLoadBalancerStack   = "service.k8s.aws/stack"
	LabelKeyAWSLoadBalancerCluster = "elbv2.k8s.aws/cluster"
)

func (r *reconcileContext) getLoadbalancerARNFromService(ns, name string) (string, error) {
	c := r.AwsClient.ResourceGroupsTaggingAPI()
	res, err := c.GetResources(r.ctx, &resourcegroupstaggingapi.GetResourcesInput{
		ResourceTypeFilters: []string{
			"elasticloadbalancing:loadbalancer",
		},
		TagFilters: []rgtatypes.TagFilter{
			{
				Key: aws.String(LabelKeyAWSLoadBalancerResource),
				Values: []string{
					LabelValAWSLoadBalancerResource,
				},
			},
			{
				Key: aws.String(LabelKeyAWSLoadBalancerStack),
				Values: []string{
					ns + "/" + name,
				},
			},
			{
				Key: aws.String(LabelKeyAWSLoadBalancerCluster),
				Values: []string{
					r.LoadBalancerClusterName,
				},
			},
		},
	})
	if err != nil {
		return "", err
	}
	arns := []string{}
	for _, item := range res.ResourceTagMappingList {
		arns = append(arns, *item.ResourceARN)
	}
	if len(arns) == 0 {
		return "", nil
	}
	if len(arns) != 1 {
		return "", fmt.Errorf("only one loadbalancer can be listed, but current has %d: %v", len(arns), arns)
	}

	return arns[0], nil
}

func getPrivateLinkName(ns, clusterName string) string {
	return fmt.Sprintf("%s.%s", ns, clusterName)
}

func (r *reconcileContext) deletePrivateLink() error {
	// delete private svc if not needed by private endpoint
	if !r.needPrivateEndpoint() {
		svc := corev1.Service{
			ObjectMeta: metav1.ObjectMeta{
				Name:      getPrivateServiceName(r.tcKey.Name),
				Namespace: r.tcKey.Namespace,
			},
		}

		if err := r.Client.Delete(r.ctx, &svc); err != nil {
			if !apierrors.IsNotFound(err) {
				return err
			}
		}
	}

	rrs := route53v1alpha1.ResourceRecordSet{
		ObjectMeta: metav1.ObjectMeta{
			Name: r.privateLinkDNSRecordName(),
		},
	}

	if err := r.InfraClient.Delete(r.ctx, &rrs); err != nil {
		if !apierrors.IsNotFound(err) {
			return err
		}
	}

	vesc := ec2v1alpha1.VPCEndpointServiceConfiguration{
		ObjectMeta: metav1.ObjectMeta{
			Name: getPrivateLinkName(r.tcKey.Namespace, r.tcKey.Name),
		},
	}

	if err := r.Client.Get(r.ctx, client.ObjectKeyFromObject(&vesc), &vesc); err != nil {
		if !apierrors.IsNotFound(err) {
			return err
		}
		// mark that vesc has been deleted
		if _, ok := r.cluster.Annotations[annoKeyPrivateLinkStatus]; ok {
			delete(r.cluster.Annotations, annoKeyPrivateLinkStatus)
			return r.InfraClient.Update(r.ctx, r.cluster)
		}
	}

	// vesc is not deleting
	if vesc.DeletionTimestamp.IsZero() {
		if err := r.Client.Delete(r.ctx, &vesc); err != nil {
			if !apierrors.IsNotFound(err) {
				return err
			}

			// mark that vesc has been deleted
			if _, ok := r.cluster.Annotations[annoKeyPrivateLinkStatus]; ok {
				delete(r.cluster.Annotations, annoKeyPrivateLinkStatus)
				return r.InfraClient.Update(r.ctx, r.cluster)
			}
		}
	}

	if vesc.Status.AtProvider.ServiceConfiguration != nil &&
		vesc.Status.AtProvider.ServiceConfiguration.ServiceID != nil {
		svcID := *vesc.Status.AtProvider.ServiceConfiguration.ServiceID

		res, err := r.AwsClient.Ec2().DescribeVpcEndpointConnections(r.ctx, &ec2.DescribeVpcEndpointConnectionsInput{
			Filters: []ec2types.Filter{
				{
					Name: pointer.String("service-id"),
					Values: []string{
						svcID,
					},
				},
			},
		})
		if err != nil {
			return err
		}
		ids := []string{}
		for _, c := range res.VpcEndpointConnections {
			// NOTE(liubo02): reject connection in some states may fail, just return errors and reconcile
			if c.VpcEndpointState != StateRejected {
				ids = append(ids, *c.VpcEndpointId)
			}
		}
		if len(ids) != 0 {
			rejectRes, err := r.AwsClient.Ec2().RejectVpcEndpointConnections(r.ctx, &ec2.RejectVpcEndpointConnectionsInput{
				ServiceId:      &svcID,
				VpcEndpointIds: ids,
			})
			if err != nil {
				return err
			}
			if len(rejectRes.Unsuccessful) != 0 {
				return fmt.Errorf("can't reject all vpc endpoint connections: %v", rejectRes.Unsuccessful)
			}
		}
	}

	return nil
}

// TODO(liubo02): delete it after this bug fixed
// See https://github.com/aws/aws-sdk-go-v2/issues/1730
var (
	StatePendingAcceptance ec2types.State = transVPCEndpointState(ec2types.StatePendingAcceptance)
	StatePending           ec2types.State = transVPCEndpointState(ec2types.StatePending)
	StateAvailable         ec2types.State = transVPCEndpointState(ec2types.StateAvailable)
	StateDeleting          ec2types.State = transVPCEndpointState(ec2types.StateDeleting)
	StateDeleted           ec2types.State = transVPCEndpointState(ec2types.StateDeleted)
	StateRejected          ec2types.State = transVPCEndpointState(ec2types.StateRejected)
	StateFailed            ec2types.State = transVPCEndpointState(ec2types.StateFailed)
	StateExpired           ec2types.State = transVPCEndpointState(ec2types.StateExpired)
)

func transVPCEndpointState(in ec2types.State) ec2types.State {
	s := string(in)
	var b strings.Builder
	b.Grow(len(s))
	c := s[0]
	if 'A' <= c && c <= 'Z' {
		c += 'a' - 'A'
	}
	b.WriteByte(c)
	b.WriteString(s[1:])

	return ec2types.State(b.String())
}

func (r *reconcileContext) getOrCreatePrivateService(_ *PrivateLinkSpec) (string, error) {
	svcName := getPrivateServiceName(r.tcKey.Name)

	svc := &corev1.Service{}
	if err := r.Cache.Get(r.ctx,
		client.ObjectKey{
			Name:      svcName,
			Namespace: r.tcKey.Namespace,
		}, svc); err != nil {
		if !apierrors.IsNotFound(err) {
			return "", fmt.Errorf("can't get service %s: %v", svcName, err)
		}

		svc = r.newServiceInAWS(false)
		if err := r.createWithOwnerLabel(svc); err != nil {
			return "", err
		}

		return "", &utils.RequeueAfterError{
			RequeueAfter: defaultWaitInterval,
			Cause:        errors.New("wait for private loadbalancer ready"),
		}
	}

	return svcName, nil
}

func (r *reconcileContext) createPrivateLink(spec *PrivateLinkSpec) error {
	svcName, err := r.getOrCreatePrivateService(spec)
	if err != nil {
		return err
	}
	arn, err := r.getLoadbalancerARNFromService(r.tcKey.Namespace, svcName)
	if err != nil {
		return err
	}
	if arn == "" {
		return fmt.Errorf("can't get loadbalancer for cluster %s", r.tcKey)
	}

	dnsName := r.endpointDNS(EndpointTypePrivateLink)

	region := r.network.Spec.Region
	vesc := ec2v1alpha1.VPCEndpointServiceConfiguration{
		ObjectMeta: metav1.ObjectMeta{
			Name:   getPrivateLinkName(r.tcKey.Namespace, r.tcKey.Name),
			Labels: labels.NewOwnerLabels(r.cluster, nil),
		},
		Spec: ec2v1alpha1.VPCEndpointServiceConfigurationSpec{
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: types.CrossplaneProviderConfigReferenceName,
				},
			},
			ForProvider: ec2v1alpha1.VPCEndpointServiceConfigurationParameters{
				Region:             region,
				AcceptanceRequired: &spec.AcceptanceRequired,
				PrivateDNSName:     pointer.String(dnsName),
				CustomVPCEndpointServiceConfigurationParameters: ec2v1alpha1.CustomVPCEndpointServiceConfigurationParameters{
					NetworkLoadBalancerARNs: []*string{
						&arn,
					},
				},
				TagSpecifications: r.getAWSPrivateLinkTagSpecification(),
			},
		},
	}

	if err := r.createWithOwnerLabel(&vesc); err != nil {
		return err
	}

	return nil
}

func (r *reconcileContext) reconcilePrivateLinkPrincipals(
	spec *PrivateLinkSpec,
	status *PrivateLinkStatus,
	vescId string,
) error {
	// if current status is as expected as spec, just return
	if status != nil && reflect.DeepEqual(spec.AllowedPrincipals, status.AllowedPrincipals) {
		return nil
	}
	c := r.AwsClient.Ec2()
	res, err := c.DescribeVpcEndpointServicePermissions(r.ctx, &ec2.DescribeVpcEndpointServicePermissionsInput{
		ServiceId: &vescId,
	})
	if err != nil {
		return err
	}

	currentMap := map[string]bool{}
	for _, p := range res.AllowedPrincipals {
		if p.Principal == nil {
			continue
		}
		currentMap[*p.Principal] = false
	}

	added, removed := []string{}, []string{}
	for _, p := range spec.AllowedPrincipals {
		if isAllowed, ok := currentMap[p]; !ok {
			// not in current, add it
			added = append(added, p)
			currentMap[p] = true
		} else if !isAllowed {
			// in current, mark it
			currentMap[p] = true
		}
	}
	for p, isAllowed := range currentMap {
		if !isAllowed {
			removed = append(removed, p)
		}
	}

	if len(added) == 0 && len(removed) == 0 {
		return nil
	}

	mRes, err := c.ModifyVpcEndpointServicePermissions(r.ctx, &ec2.ModifyVpcEndpointServicePermissionsInput{
		ServiceId:               &vescId,
		AddAllowedPrincipals:    added,
		RemoveAllowedPrincipals: removed,
	})
	if err != nil {
		return err
	}
	if mRes.ReturnValue == nil || !*mRes.ReturnValue {
		return fmt.Errorf("failed to modify vpc endpoint service permission")
	}

	return nil
}

func (r *reconcileContext) updatePrivateLink(
	spec *PrivateLinkSpec,
	vesc *ec2v1alpha1.VPCEndpointServiceConfiguration,
) error {
	svcName, err := r.getOrCreatePrivateService(spec)
	if err != nil {
		return err
	}
	arn, err := r.getLoadbalancerARNFromService(r.tcKey.Namespace, svcName)
	if err != nil {
		return err
	}
	if arn == "" {
		return fmt.Errorf("can't get loadbalancer for cluster %s", r.tcKey)
	}

	arns := vesc.Spec.ForProvider.NetworkLoadBalancerARNs
	if len(arns) != 1 || *arns[0] != arn {
		vesc.Spec.ForProvider.NetworkLoadBalancerARNs = []*string{
			pointer.StringPtr(arn),
		}
	}

	if err := r.Client.Update(r.ctx, vesc); err != nil {
		return err
	}

	return nil
}

func (r *reconcileContext) privateLinkDNSRecordName() string {
	return fmt.Sprintf("%s-privatelink", r.cluster.ClusterScopeName())
}

// NOTE(liubo02): this function can be deleted if https://github.com/crossplane/provider-aws/issues/1179 fixed
func (r *reconcileContext) getPrivateDNSNameConfiguration(vescID string) (
	*ec2v1alpha1.PrivateDNSNameConfiguration,
	error,
) {
	res, err := r.AwsClient.Ec2().DescribeVpcEndpointServiceConfigurations(
		r.ctx,
		&ec2.DescribeVpcEndpointServiceConfigurationsInput{
			ServiceIds: []string{vescID},
		},
	)
	if err != nil {
		return nil, err
	}
	if len(res.ServiceConfigurations) != 1 {
		return nil, fmt.Errorf(
			"can't get vpc endpoint service configuration with id %s, get %d items",
			vescID,
			len(res.ServiceConfigurations),
		)
	}
	svcConfig := res.ServiceConfigurations[0]

	dnsConfig := svcConfig.PrivateDnsNameConfiguration

	return &ec2v1alpha1.PrivateDNSNameConfiguration{
		Name:  dnsConfig.Name,
		Type:  dnsConfig.Type,
		Value: dnsConfig.Value,
		State: (*string)(&dnsConfig.State),
	}, nil
}

func (r *reconcileContext) reconcilePrivateLinkDNSRecord(
	vescID string,
	dnsConfig *ec2v1alpha1.PrivateDNSNameConfiguration,
) error {
	rrs := &route53v1alpha1.ResourceRecordSet{
		ObjectMeta: metav1.ObjectMeta{
			Name: r.privateLinkDNSRecordName(),
		},
	}

	if err := r.InfraCache.Get(r.ctx, client.ObjectKeyFromObject(rrs), rrs); err != nil {
		if !apierrors.IsNotFound(err) {
			return err
		}
		if dnsConfig == nil {
			cfg, err := r.getPrivateDNSNameConfiguration(vescID)
			if err != nil {
				return err
			}
			dnsConfig = cfg
		}

		rrs = &route53v1alpha1.ResourceRecordSet{
			ObjectMeta: metav1.ObjectMeta{
				Name: r.privateLinkDNSRecordName(),
				Annotations: map[string]string{
					meta.AnnotationKeyExternalName: *dnsConfig.Name + "." + r.PublicHostedZoneDomain,
					annoKeyPrivateLinkDNSStatus:    *dnsConfig.State,
				},
				OwnerReferences: []metav1.OwnerReference{
					*metav1.NewControllerRef(r.cluster, cloudv1.GroupVersion.WithKind("Cluster")),
				},
			},
			Spec: route53v1alpha1.ResourceRecordSetSpec{
				ResourceSpec: xpv1.ResourceSpec{
					ProviderConfigReference: &xpv1.Reference{
						Name: types.CrossplaneProviderConfigReferenceName,
					},
				},
				ForProvider: route53v1alpha1.ResourceRecordSetParameters{
					Type: *dnsConfig.Type,
					TTL:  pointer.Int64(defaultDNSTTLSeconds),
					ResourceRecords: []route53v1alpha1.ResourceRecord{
						{
							// Txt record value should be enclosed in quotation marks, for example, "Sample text entry".
							Value: strconv.Quote(*dnsConfig.Value),
						},
					},
					ZoneID: pointer.StringPtr(r.PublicHostedZoneId),
				},
			},
		}

		if err := r.InfraClient.Create(r.ctx, rrs); err != nil {
			return err
		}
	}

	status, ok := rrs.Annotations[annoKeyPrivateLinkDNSStatus]
	if !ok || status != string(ec2types.DnsNameStateVerified) {
		if dnsConfig == nil {
			cfg, err := r.getPrivateDNSNameConfiguration(vescID)
			if err != nil {
				return err
			}
			dnsConfig = cfg
		}
		if status == *dnsConfig.State {
			return &utils.RequeueAfterError{
				RequeueAfter: defaultWaitInterval,
				Cause:        fmt.Errorf("wait vpc endpoint service dns verified, current state is %v", *dnsConfig.State),
			}
		}

		rrs.Annotations[annoKeyPrivateLinkDNSStatus] = *dnsConfig.State
		if err := r.InfraClient.Update(r.ctx, rrs); err != nil {
			return err
		}
	}

	return nil
}

func (r *reconcileContext) reconcilePrivateLink() error {
	if !r.needPrivateLink() {
		return r.deletePrivateLink()
	}

	spec, err := GetPrivateLinkSpec(r.cluster)
	if err != nil {
		return err
	}
	oldStatus, err := GetPrivateLinkStatus(r.cluster)
	if err != nil {
		return err
	}

	vesc := ec2v1alpha1.VPCEndpointServiceConfiguration{
		ObjectMeta: metav1.ObjectMeta{
			Name: getPrivateLinkName(r.tcKey.Namespace, r.tcKey.Name),
		},
	}

	if err := r.Cache.Get(r.ctx, client.ObjectKeyFromObject(&vesc), &vesc); err != nil {
		if !apierrors.IsNotFound(err) {
			return err
		}

		if err := r.createPrivateLink(spec); err != nil {
			return err
		}
	} else if oldStatus == nil || oldStatus.KubeService != getPrivateServiceName(r.tcKey.Name) {
		if err := r.updatePrivateLink(spec, &vesc); err != nil {
			return err
		}
	}

	sc := vesc.Status.AtProvider.ServiceConfiguration
	if sc == nil || sc.ServiceID == nil {
		return &utils.RequeueAfterError{
			RequeueAfter: defaultWaitInterval,
			Cause:        errors.New("wait vpc endpoint service sync"),
		}
	}

	vescId := sc.ServiceID

	dnsConfig := sc.PrivateDNSNameConfiguration
	if err := r.reconcilePrivateLinkDNSRecord(*vescId, dnsConfig); err != nil {
		return err
	}

	if sc.ServiceState == nil ||
		*sc.ServiceState != string(ec2types.ServiceStateAvailable) {
		return &utils.RequeueAfterError{
			RequeueAfter: defaultWaitInterval,
			Cause:        errors.New("wait vpc endpoint service available"),
		}
	}

	if err := r.reconcilePrivateLinkPrincipals(spec, oldStatus, *vescId); err != nil {
		return err
	}

	conns, err := r.reconcilePrivateLinkConnections(*vescId, spec.AllowedVPCEndpoints)
	if err != nil {
		return err
	}

	status := PrivateLinkStatus{
		Phase:                  PrivateLinkPhaseReady,
		PrivateLinkServiceName: *vesc.Status.AtProvider.ServiceConfiguration.ServiceName,
		KubeService:            getPrivateServiceName(r.tcKey.Name),
		AllowedPrincipals:      spec.AllowedPrincipals,
		Connections:            conns,
	}

	if err := SetPrivateLinkStatus(r.cluster, &status); err != nil {
		return err
	}

	return r.InfraClient.Update(r.ctx, r.cluster)
}

func (r *reconcileContext) getCurrentConnections(svcID string) ([]PrivateLinkConnection, error) {
	res, err := r.AwsClient.Ec2().DescribeVpcEndpointConnections(r.ctx, &ec2.DescribeVpcEndpointConnectionsInput{
		Filters: []ec2types.Filter{
			{
				Name: pointer.String("service-id"),
				Values: []string{
					svcID,
				},
			},
		},
	})
	if err != nil {
		return nil, err
	}

	conns := []PrivateLinkConnection{}

	for _, c := range res.VpcEndpointConnections {
		conns = append(conns, PrivateLinkConnection{
			VPCEndpointID: *c.VpcEndpointId,
			State:         string(c.VpcEndpointState),
			Owner:         *c.VpcEndpointOwner,
		})
	}

	return conns, nil
}

func (r *reconcileContext) reconcilePrivateLinkConnections(
	svcID string,
	allowedEndpoints []string,
) ([]PrivateLinkConnection, error) {
	conns, err := r.getCurrentConnections(svcID)
	if err != nil {
		return nil, err
	}

	r.Info("allowed vpc endpoints", "serviceID", svcID, "endpoints", allowedEndpoints)

	allowedEndpointSet := map[string]struct{}{}
	for _, id := range allowedEndpoints {
		allowedEndpointSet[id] = struct{}{}
	}

	isAllowed := func(id string) bool {
		_, ok := allowedEndpointSet[id]
		return ok
	}

	accept, reject := []string{}, []string{}
	// nolint: exhaustive // FIXME(liubo02): it should be changed to use enum in ec2types directly
	for _, c := range conns {
		switch ec2types.State(c.State) {
		case StatePendingAcceptance, StatePending:
			if isAllowed(c.VPCEndpointID) {
				accept = append(accept, c.VPCEndpointID)
			}
		case StateAvailable:
			if !isAllowed(c.VPCEndpointID) {
				reject = append(reject, c.VPCEndpointID)
			}

		case StateDeleting,
			StateDeleted,
			StateRejected,
			StateFailed,
			StateExpired:
			// ignored
		default:
			// ignored
		}
	}

	// no need to reconcile connections
	if len(accept) == 0 && len(reject) == 0 {
		return conns, nil
	}

	errs := []error{}
	if err := r.rejectVPCEndpointConnections(svcID, reject); err != nil {
		errs = append(errs, err)
	}
	if err := r.acceptVPCEndpointConnections(svcID, accept); err != nil {
		errs = append(errs, err)
	}
	if len(errs) != 0 {
		return nil, fmt.Errorf("can't reconcile vpc endpoint connections: %v", errs)
	}

	return r.getCurrentConnections(svcID)
}

func (r *reconcileContext) acceptVPCEndpointConnections(svcID string, endpointIDs []string) error {
	if len(endpointIDs) == 0 {
		return nil
	}

	r.Info("try to accept vpc endpoints", "serviceID", svcID, "endpoints", endpointIDs)

	res, err := r.AwsClient.Ec2().AcceptVpcEndpointConnections(r.ctx, &ec2.AcceptVpcEndpointConnectionsInput{
		ServiceId:      &svcID,
		VpcEndpointIds: endpointIDs,
	})
	if err != nil {
		return err
	}
	if len(res.Unsuccessful) != 0 {
		return fmt.Errorf("can't accept endpoints %v", res.Unsuccessful)
	}

	return nil
}

func (r *reconcileContext) rejectVPCEndpointConnections(svcID string, endpointIDs []string) error {
	if len(endpointIDs) == 0 {
		return nil
	}

	r.Info("try to reject vpc endpoints", "serviceID", svcID, "endpoints", endpointIDs)

	res, err := r.AwsClient.Ec2().RejectVpcEndpointConnections(r.ctx, &ec2.RejectVpcEndpointConnectionsInput{
		ServiceId:      &svcID,
		VpcEndpointIds: endpointIDs,
	})
	if err != nil {
		return err
	}
	if len(res.Unsuccessful) != 0 {
		return fmt.Errorf("can't reject endpoints %v", res.Unsuccessful)
	}

	return nil
}

func (r *reconcileContext) getAWSPrivateLinkTagSpecification() []*ec2v1alpha1.TagSpecification {
	tags := r.getTags(types.BillingTagValComponentPrivateLink, types.BillingTagValUsedByCustomerTiDB)

	ntags := []*ec2v1alpha1.Tag{}

	for _, item := range tags {
		ntags = append(ntags, &ec2v1alpha1.Tag{
			Key:   pointer.String(item.Key),
			Value: pointer.String(item.Value),
		})
	}

	return []*ec2v1alpha1.TagSpecification{
		{
			ResourceType: pointer.String("vpc-endpoint-service"),
			Tags:         ntags,
		},
	}
}