package eks

import (
	"context"
	"fmt"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/aws/aws-sdk-go-v2/service/ec2"
	ec2types "github.com/aws/aws-sdk-go-v2/service/ec2/types"
	elbv2 "github.com/aws/aws-sdk-go-v2/service/elasticloadbalancingv2"
	tagging "github.com/aws/aws-sdk-go-v2/service/resourcegroupstaggingapi"
	rgtatypes "github.com/aws/aws-sdk-go-v2/service/resourcegroupstaggingapi/types"
	ec2v1alpha1 "github.com/crossplane-contrib/provider-aws/apis/ec2/v1alpha1"
	route53v1alpha1 "github.com/crossplane-contrib/provider-aws/apis/route53/v1alpha1"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	"github.com/crossplane/crossplane-runtime/pkg/meta"
	"github.com/pingcap/errors"
	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"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/log"

	eksv1alpha1 "github.com/tidbcloud/eks-provider/api/v1alpha1"
	"github.com/tidbcloud/eks-provider/controllers/cluster/task"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/cloud/aws"
	"github.com/tidbcloud/infra-provider/pkg/controllers/common"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	"github.com/tidbcloud/infra-provider/pkg/util"
)

const (
	defaultDNSTTLSeconds = 300

	defaultWaitInterval = 10 * time.Second

	AnnotationCrossplaneCreationPending = "crossplane.io/external-create-pending"
	awsEndpointNotFoundErrorCode        = "InvalidVpcEndpoint.NotFound"

	AnnotationAWSEmptyListWaitingStarted      = "privatelink.cloud.pingcap.com/waiting-started"
	AnnotationAWSEmptyListWaitingStartingTime = "privatelink.cloud.pingcap.com/waiting-starting-time"

	AnnotationKeyAllowRotateLBOnce = "privatelink.cloud.pingcap.com/allow-rotate-lb-once"
)

func allowRotateLBOnce(c *cloudv1.Cluster) bool {
	v := c.Annotations[AnnotationKeyAllowRotateLBOnce]
	return v != ""
}

// TODO(zlj): conversion of ec2 types state to lower case string can be removed
// after this fix: https://github.com/aws/aws-sdk-go-v2/issues/1730
// The ec2 API will return state in camelCase with the first letter lowercased,
// e.g., pendingAcceptance, which is not as expected. The consts defined in
// github.com/aws/aws-sdk-go-v2/service/ec2@v1.26.0/types/enums.g are in
// CamelCase but with the first letter capitalized, e.g., PendingAcceptance.
// So we redefined the ec2 state here.
type awsEC2State string

var (
	awsEC2StatePendingAcceptance awsEC2State = "pendingacceptance"
	awsEC2StatePending           awsEC2State = "pending"
	awsEC2StateAvailable         awsEC2State = "available"
	awsEC2StateDeleting          awsEC2State = "deleting"
	awsEC2StateDeleted           awsEC2State = "deleted"
	awsEC2StateRejected          awsEC2State = "rejected"
	awsEC2StateFailed            awsEC2State = "failed"
	awsEC2StateExpired           awsEC2State = "expired"
)

func newAWSEC2State(e ec2types.State) awsEC2State {
	return awsEC2State(strings.ToLower(string(e)))
}

type PrivateLink struct {
	*task.BaseTask

	AWSCli aws.Client
}

type PrivateLinkReconcileCtx struct {
	*PrivateLink
	ec2Cli *ec2.Client
	tagCli *tagging.Client
	elbCli *elbv2.Client
}

// IsPrivateLinkRequired checks if the private link is required by the cluster.
func IsPrivateLinkRequired(cluster cloudv1.Cluster) bool {
	return cluster.Spec.SqlEndpoints.PrivateLinkService != nil ||
		cluster.Status.SqlEndpoints.PrivateLinkService != nil
}

func privateLinkServicePhase(
	cluster *cloudv1.Cluster,
) cloudv1.PrivateLinkServicePhase {
	if cluster.Spec.SqlEndpoints.PrivateLinkService != nil &&
		cluster.Status.SqlEndpoints.PrivateLinkService == nil {
		return cloudv1.PrivateLinkServicePhaseCreating
	}

	if cluster.Spec.SqlEndpoints.PrivateLinkService != nil &&
		cluster.Status.SqlEndpoints.PrivateLinkService != nil {
		return cloudv1.PrivateLinkServicePhaseActive
	}

	// corresponding fields are set in status but not in spec
	return cloudv1.PrivateLinkServicePhaseDeleting
}

func NewPrivateLink(ctx *task.Context, awsCli aws.Client) *PrivateLink {
	return &PrivateLink{BaseTask: &task.BaseTask{Context: ctx}, AWSCli: awsCli}
}

func (t *PrivateLink) Name() string {
	return "PrivateLink"
}

// Create is the main entry for privatelink reconciliation.
func (t *PrivateLink) Create(ctx context.Context) deploy.Result {
	if t.IsRegionalManaged() {
		return deploy.Done()
	}
	if t.Config.DryRun.Enabled {
		return deploy.Done()
	}

	lgr := log.FromContext(ctx)
	role := t.GetAWSAssumedRoleForSDK()
	rCtx := PrivateLinkReconcileCtx{
		PrivateLink: t,
		ec2Cli:      t.AWSCli.EC2(role),
		tagCli:      t.AWSCli.ResourceGroupsTagging(role),
		elbCli:      t.AWSCli.ELBV2(role),
	}

	switch privateLinkServicePhase(t.Cluster) {
	case cloudv1.PrivateLinkServicePhaseCreating:
		lgr.V(8).Info("reconcile private link service",
			"phase", cloudv1.PrivateLinkServicePhaseCreating,
			"cluster", t.Cluster.GetName())
		return rCtx.reconcilePrivateLink(ctx)
	case cloudv1.PrivateLinkServicePhaseActive:
		lgr.V(8).Info("reconcile private link service",
			"phase", cloudv1.PrivateLinkServicePhaseActive,
			"cluster", t.Cluster.GetName())
		return rCtx.reconcilePrivateLink(ctx)
	default:
		lgr.V(8).Info("delete private link service",
			"cluster", t.Cluster.GetName())
		return rCtx.deletePrivateLink(ctx)
	}
}

// updateErrMessage updates the message of the private link service if
// met `err` and return the deploy Error.
func updateErrMessage(t *PrivateLinkReconcileCtx, err error) deploy.Result {
	if err != nil {
		t.Cluster.Status.SqlEndpoints.PrivateLinkService.Message =
			err.Error()
	}
	return deploy.Error(err)
}

// updatePrivateLinkServiceAsCreating updates the private link service
// status as creating.
func updatePrivateLinkServiceAsCreating(t *PrivateLinkReconcileCtx) {
	t.Cluster.Status.SqlEndpoints.PrivateLinkService =
		&cloudv1.PrivateLinkServiceStatus{
			Phase:             cloudv1.PrivateLinkServicePhaseCreating,
			AWS:               &cloudv1.AWSPrivateLinkServiceStatus{},
			CreationTimestamp: metav1.NewTime(time.Now()),
		}
}

// genLatestConnsStatus generates the latest status of all private link
// connections in the correct format.
func genLatestConnsStatus(
	t *PrivateLinkReconcileCtx,
) []*cloudv1.AWSPrivateLinkServiceConnection {
	connsMap := t.getConnsMap()
	connsStatus := []*cloudv1.AWSPrivateLinkServiceConnection{}
	for id, c := range connsMap {
		connsStatus = append(connsStatus,
			&cloudv1.AWSPrivateLinkServiceConnection{
				Endpoint: id,
				Phase:    c.Phase,
				Message:  c.Message,
			})
	}
	// sort the conns by id to ensure it is consistent as shown
	// on the front end
	sort.Slice(connsStatus, func(i, j int) bool {
		return connsStatus[i].Endpoint < connsStatus[j].Endpoint
	})
	return connsStatus
}

// getSubnetAZId returns availability zone ids of given subnets.
func getSubnetAZId(
	ctx context.Context,
	t *PrivateLinkReconcileCtx,
	ids []string,
) ([]string, error) {
	azIds := []string{}
	oups, err := t.ec2Cli.DescribeSubnets(
		ctx,
		&ec2.DescribeSubnetsInput{
			SubnetIds: ids,
		},
	)
	if err != nil {
		return nil, err
	}
	if len(oups.Subnets) != len(ids) {
		return nil, errors.Errorf(
			"expect %d subnets, got %d",
			len(ids), len(oups.Subnets))
	}
	for _, sn := range oups.Subnets {
		id := sn.AvailabilityZoneId
		if id == nil {
			// return error if the Availability Zone Id is not found
			return nil, errors.Errorf("empty AZ Id(%s)", *sn.SubnetId)
		}
		azIds = append(azIds, *id)
	}
	return azIds, nil
}

// getAvailabilityZones returns the availability zone ids of the given NLB.
func getAvailabilityZones(
	ctx context.Context,
	t *PrivateLinkReconcileCtx,
	nlbARN string,
) ([]string, error) {
	ids := []string{}
	oups, err := t.elbCli.DescribeLoadBalancers(ctx,
		&elbv2.DescribeLoadBalancersInput{
			LoadBalancerArns: []string{nlbARN},
		})
	if err != nil {
		return nil, err
	}
	if oups == nil || len(oups.LoadBalancers) < 1 {
		// return error if elb not found
		return nil, errors.Errorf("nlb(%s) not found", nlbARN)
	}
	for _, sn := range oups.LoadBalancers[0].AvailabilityZones {
		ids = append(ids, *sn.SubnetId)
	}
	return getSubnetAZId(ctx, t, ids)
}

// updatePrivateLinkServiceOnAvailable updates the status of the private link
// when it is available. The status will be read by the front-end and shown to
// users who want to connect interface endpoints to this endpoint service.
func updatePrivateLinkServiceOnAvailable(
	ctx context.Context,
	t *PrivateLinkReconcileCtx,
	vesc *ec2v1alpha1.VPCEndpointServiceConfiguration,
) error {
	// PirvateLink Service is available now
	t.Cluster.Status.SqlEndpoints.PrivateLinkService.Phase =
		cloudv1.PrivateLinkServicePhaseActive
	sc := vesc.Status.AtProvider.ServiceConfiguration
	nlbARN := *vesc.Spec.ForProvider.
		CustomVPCEndpointServiceConfigurationParameters.
		NetworkLoadBalancerARNs[0]
	azs, err := getAvailabilityZones(ctx, t, nlbARN)
	if err != nil {
		return err
	}
	t.Cluster.Status.SqlEndpoints.PrivateLinkService.AWS =
		&cloudv1.AWSPrivateLinkServiceStatus{
			NLB:               nlbARN,
			AvailabilityZones: azs,
			ServiceID:         *sc.ServiceID,
			ServiceName:       *sc.ServiceName,
			PrivateDNSName:    *vesc.Spec.ForProvider.PrivateDNSName,
			Port: pointer.Int32(
				t.Cluster.GetPrivateSQLEndpointPort()),
			Connections: t.Cluster.Status.SqlEndpoints.
				PrivateLinkService.AWS.Connections,
		}
	t.Cluster.Status.SqlEndpoints.PrivateLinkService.Message = ""
	return nil
}

// reconcilePrivateLink is the main reconciliation loop for the private link
// service, it will:
//  1. create the endpoint service if not exist;
//  2. register private DNS for the endpoint service if not exist;
//  3. accept all newly registered endpoint connection.
//
// This function will be invoked periodically, as we need to poll the
// real connection status from AWS.
func (t *PrivateLinkReconcileCtx) reconcilePrivateLink(
	ctx context.Context,
) deploy.Result {
	lgr := log.FromContext(ctx)
	lgr.V(8).Info("reconcile private link", "cluster", t.Cluster.Name)
	if t.Cluster.Status.SqlEndpoints.PrivateLinkService == nil {
		lgr.V(8).Info("update private link status to creating",
			"cluster", t.Cluster.Name)
		updatePrivateLinkServiceAsCreating(t)
	}

	// TODO(zlj): delete when endpoints ready in metadb and worker in central
	// Show endpoints in status early
	connsStatus := genLatestConnsStatus(t)
	lgr.V(8).Info("get the latest private linke connections",
		"cluster", t.Cluster.Name)
	t.Cluster.Status.
		SqlEndpoints.PrivateLinkService.AWS.Connections = connsStatus

	// 1. create the endpoint service on the TiDB service side
	vesc, err := t.reconcileVPCEndpointServiceConfiguration(ctx)
	if err != nil {
		return updateErrMessage(t, err)
	}
	lgr.V(8).Info("endpoint service is created",
		"endpointService", vesc.GetName(),
		"cluster", t.Cluster.Name)

	sc := vesc.Status.AtProvider.ServiceConfiguration
	if sc == nil || sc.ServiceID == nil {
		return deploy.Error(errors.New("wait vpc endpoint service sync"))
	}
	lgr.V(8).Info("endpoint service is synced",
		"endpointService", vesc.GetName(),
		"cluster", t.Cluster.Name)

	// 2. enable the private DNS on the TiDB service side
	if err := t.reconcilePrivateDNSName(ctx, sc); err != nil {
		return updateErrMessage(t, err)
	}
	lgr.V(8).Info("private DNS is ready",
		"dnsRecord", t.privatelinkDnsRecordName(),
		"cluster", t.Cluster.Name)

	if sc.ServiceState == nil ||
		*sc.ServiceState != string(ec2types.ServiceStateAvailable) {
		return deploy.Error(errors.New("wait vpc endpoint service sync"))
	}
	lgr.V(8).Info("endpoint service is ready",
		"service", sc.ServiceName,
		"cluster", t.Cluster.Name)

	if err := t.reconcileAllowedPrincipals(ctx, sc); err != nil {
		return updateErrMessage(t, err)
	}

	if err := updatePrivateLinkServiceOnAvailable(ctx, t, vesc); err != nil {
		return updateErrMessage(t, err)
	}

	// 3. accept the connection from users
	connsStatus, err = t.reconcilePrivateLinkServiceConnections(ctx, sc)
	if err != nil {
		return updateErrMessage(t, err)
	}
	t.Cluster.Status.SqlEndpoints.PrivateLinkService.AWS.Connections =
		connsStatus

	// Check endpoint connections periodically
	// TODO(charleszheng44): is the requeuing really necessary?
	return deploy.Done().WithRequeuAfter(2 * defaultWaitInterval)
}

// registerPrivateDNS register the domain name through the route53.
func registerPrivateDNS(ctx context.Context,
	t *PrivateLinkReconcileCtx,
	rrs *route53v1alpha1.ResourceRecordSet,
	dnsConfig *ec2v1alpha1.PrivateDNSNameConfiguration,
	providerConfigForCentralAccount string,
) error {
	rrs.ObjectMeta.Annotations = map[string]string{
		meta.AnnotationKeyExternalName: *dnsConfig.Name + "." +
			t.Config.FullManage.PublicHostedZoneDomain,
	}
	rrs.Spec = route53v1alpha1.ResourceRecordSetSpec{
		ResourceSpec: xpv1.ResourceSpec{
			ProviderConfigReference: &xpv1.Reference{
				Name: providerConfigForCentralAccount,
			},
		},
		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.String(t.Config.FullManage.PublicHostedZoneID),
		},
	}
	return t.Local.Create(ctx, rrs)
}

// checkRecordSetCondition checks the route53 resource record set condition.
// The check will pass only when:
//  2. The condition is ready and available.
func checkRecordSetCondition(
	ctx context.Context,
	rrs *route53v1alpha1.ResourceRecordSet,
) error {
	lgr := log.FromContext(ctx)
	if rrs.Status.GetCondition(xpv1.TypeReady).Reason != xpv1.ReasonAvailable {
		return errors.Errorf("wait route53 ResourceRecordSet "+
			"condition to %s, condition: %v", xpv1.ReasonAvailable,
			rrs.Status.GetCondition(xpv1.TypeReady))
	}
	lgr.V(8).Info("resource record set is ready", "rrs", rrs.GetName())
	return nil
}

// verifyOwnership will verify the ownership of the newly registered domain
// name. The domain name can only be used by the endpoint service only when
// its ownership is verified.
func verifyOwnership(
	ctx context.Context,
	t *PrivateLinkReconcileCtx,
	vescID string,
	dnsConfig ec2v1alpha1.PrivateDNSNameConfiguration,
	dryRun bool,
) error {
	verifyOpRes, err := t.ec2Cli.
		StartVpcEndpointServicePrivateDnsVerification(
			ctx, &ec2.StartVpcEndpointServicePrivateDnsVerificationInput{
				ServiceId: &vescID,
				DryRun:    &dryRun,
			}, func(opts *ec2.Options) {
				opts.Region = t.EnvProvider.GetRegion()
			})
	if err != nil {
		return err
	}
	if *verifyOpRes.ReturnValue == false {
		return errors.Errorf("verify domain ownership failed, "+
			"dns config: %v", dnsConfig)
	}

	if *dnsConfig.State != string(ec2types.DnsNameStateVerified) {
		return errors.Errorf("wait vpc endpoint service dns verified, "+
			"current state is %v", *dnsConfig.State)
	}

	return nil
}

// reconcilePrivateDNSName register the private domain name for the endpoint
// service through route53 if not exist and complete the ownership verification.
func (t *PrivateLinkReconcileCtx) reconcilePrivateDNSName(
	ctx context.Context,
	sc *ec2v1alpha1.ServiceConfiguration,
) error {
	lgr := log.FromContext(ctx)
	vescID := *sc.ServiceID
	dnsConfig := sc.PrivateDNSNameConfiguration

	if *dnsConfig.State == string(ec2types.DnsNameStateVerified) {
		return nil
	}

	rrs := &route53v1alpha1.ResourceRecordSet{
		ObjectMeta: metav1.ObjectMeta{
			Name: t.privatelinkDnsRecordName(),
		},
	}
	if err := t.Local.Get(ctx, client.ObjectKeyFromObject(rrs), rrs); err != nil {
		if !apierrors.IsNotFound(err) {
			return err
		}
		lgr.V(8).Info("resource record set is not exist",
			"rrs", t.privatelinkDnsRecordName(),
			"endpointService", sc.ServiceName)
		// register the domain name if haven't done so.
		if err := registerPrivateDNS(ctx, t, rrs, dnsConfig, t.Config.ControlplaneAWSProviderConfig); err != nil {
			return err
		}
		lgr.V(8).Info("resource record set is registered",
			"rrs", t.privatelinkDnsRecordName(),
			"endpointService", sc.ServiceName)
	}
	if err := checkRecordSetCondition(ctx, rrs); err != nil {
		return err
	}
	lgr.V(8).Info("resource record set condition is valid",
		"rrs", t.privatelinkDnsRecordName(),
		"endpointService", sc.ServiceName)

	if err := verifyOwnership(ctx, t, vescID, *dnsConfig, t.Config.DryRun.Enabled); err != nil {
		return err
	}
	lgr.V(8).Info("DNS name has been verified",
		"rrs", t.privatelinkDnsRecordName(),
		"dnsName", dnsConfig.Name,
		"endpointService", sc.ServiceName)
	return nil
}

// markRemovedEp update the phase of existing connections that have been
// recently removed from the front-end as deleting.
func markRemovedEp(
	acceptedEp map[string]struct{},
	connsMap map[string]*cloudv1.AWSPrivateLinkServiceConnection,
) {
	for _, c := range connsMap {
		if _, exist := acceptedEp[c.Endpoint]; !exist &&
			// delete the endpoint if it is not in the accepted endpoint list
			c.Phase != cloudv1.PrivateLinkServiceConnectionPhaseDeleting {
			c.Phase = cloudv1.PrivateLinkServiceConnectionPhaseDeleting
			c.Message = ""
		}
	}
}

// getConnsMap generates the latest private link connection map containing
// both newly registered connection that have not been accepted and existing
// private link connections. This function will also update the status of the
// existing connections, if some of them have just been removed by the users,
// i.e., removed from the `spec.AWS.AcceptedEndpoints`.
func (t *PrivateLinkReconcileCtx) getConnsMap() map[string]*cloudv1.
	AWSPrivateLinkServiceConnection {
	spec := t.Cluster.Spec.SqlEndpoints.PrivateLinkService.DeepCopy()
	status := t.Cluster.Status.SqlEndpoints.PrivateLinkService.DeepCopy()
	connsMap := make(map[string]*cloudv1.AWSPrivateLinkServiceConnection)

	// Include existing connections from the status.
	for _, c := range status.AWS.Connections {
		connsMap[c.Endpoint] = c
	}

	// acceptedIDs contains all endpoints registered by users from the
	// front-end.
	acceptedIDs := make(map[string]struct{})
	for _, c := range spec.AWS.AcceptedEndpoints {
		acceptedIDs[c] = struct{}{}
		// Add new endpoints to the connections map and set them as Pending.
		if _, ok := connsMap[c]; !ok {
			connsMap[c] = &cloudv1.AWSPrivateLinkServiceConnection{
				Endpoint: c,
				Phase:    cloudv1.PrivateLinkServiceConnectionPhasePending,
			}
		}
	}

	markRemovedEp(acceptedIDs, connsMap)

	return connsMap
}

// getAWSConnectionsMap gets the up-to-date endpoint connections status
// by calling the aws-sdk.
func (t *PrivateLinkReconcileCtx) getAWSConnectionsMap(
	ctx context.Context,
	serviceID string,
) (map[string]awsEC2State, error) {
	connectionsMap := make(map[string]awsEC2State, 0)
	res, err := t.ec2Cli.DescribeVpcEndpointConnections(ctx,
		&ec2.DescribeVpcEndpointConnectionsInput{
			Filters: []ec2types.Filter{
				{
					Name: pointer.String("service-id"),
					Values: []string{
						serviceID,
					},
				},
			},
		}, func(opts *ec2.Options) {
			opts.Region = t.EnvProvider.GetRegion()
		})
	if err != nil {
		return nil, err
	}

	for _, c := range res.VpcEndpointConnections {
		connectionsMap[*c.VpcEndpointId] =
			newAWSEC2State(c.VpcEndpointState)
	}

	return connectionsMap, nil
}

// categorizeEndpoints divides the endpoint connections into two
// categories, i.e., endpoints to be rejected and endpoints to be accepted.
func categorizeEndpoints(
	conns map[string]*cloudv1.AWSPrivateLinkServiceConnection,
) (
	toReject []string,
	toAccept []string,
) {
	for _, c := range conns {
		if c.Phase == cloudv1.PrivateLinkServiceConnectionPhaseDeleting {
			toReject = append(toReject, c.Endpoint)
			continue
		}
		if c.Phase != cloudv1.PrivateLinkServiceConnectionPhaseActive {
			// ignore if this endpoint is not found
			if c.Phase == cloudv1.PrivateLinkServiceConnectionPhaseFailed && c.Message == awsEndpointNotFoundErrorCode {
				continue
			}

			toAccept = append(toAccept, c.Endpoint)
		}
	}
	return
}

// acceptEndpoints accepts all the specified endpoints.
func acceptEndpoints(
	ctx context.Context,
	t *PrivateLinkReconcileCtx,
	toAccept []string,
	sc *ec2v1alpha1.ServiceConfiguration,
	conns map[string]*cloudv1.AWSPrivateLinkServiceConnection,
	dryRun bool,
) error {
	acceptOpRes, err := t.ec2Cli.AcceptVpcEndpointConnections(ctx,
		&ec2.AcceptVpcEndpointConnectionsInput{
			ServiceId:      sc.ServiceID,
			VpcEndpointIds: toAccept,
			DryRun:         &dryRun,
		}, func(opts *ec2.Options) {
			opts.Region = t.EnvProvider.GetRegion()
		})
	if err != nil {
		return err
	}

	if acceptOpRes.Unsuccessful != nil {
		// Modify connections information
		for _, c := range acceptOpRes.Unsuccessful {
			endpointID := *c.ResourceId
			msg := *c.Error.Message
			// mark this endpoint as NotFound to prevent useless retries which cause rate limit issue
			if pointer.StringDeref(c.Error.Code, "") == awsEndpointNotFoundErrorCode {
				msg = awsEndpointNotFoundErrorCode
			}

			conns[endpointID].Phase = cloudv1.PrivateLinkServiceConnectionPhaseFailed
			conns[endpointID].Message = msg
		}
	}
	return nil
}

// rejectEndpoints rejects all the specified endpoints.
func rejectEndpoints(
	ctx context.Context,
	t *PrivateLinkReconcileCtx,
	toReject []string,
	sc *ec2v1alpha1.ServiceConfiguration,
	conns map[string]*cloudv1.AWSPrivateLinkServiceConnection,
	dryRun bool,
) error {
	rejectOpRes, err := t.ec2Cli.RejectVpcEndpointConnections(ctx,
		&ec2.RejectVpcEndpointConnectionsInput{
			ServiceId:      sc.ServiceID,
			VpcEndpointIds: toReject,
			DryRun:         &dryRun,
		}, func(opts *ec2.Options) {
			opts.Region = t.EnvProvider.GetRegion()
		})
	if err != nil {
		return err
	}
	if rejectOpRes.Unsuccessful != nil {
		// Modify connections information
		for _, c := range rejectOpRes.Unsuccessful {
			endpointID := *c.ResourceId

			// Skip not found endpoint
			if *c.Error.Code == awsEndpointNotFoundErrorCode {
				continue
			}

			// Keep in deleting
			conns[endpointID].Phase =
				cloudv1.PrivateLinkServiceConnectionPhaseDeleting
			conns[endpointID].Message = *c.Error.Message
		}
	}
	return nil
}

func genLatestConnsStatusAWS(
	awsConns map[string]awsEC2State,
	conns map[string]*cloudv1.AWSPrivateLinkServiceConnection,
) []*cloudv1.AWSPrivateLinkServiceConnection {
	connsStatus := []*cloudv1.AWSPrivateLinkServiceConnection{}
	for id, c := range conns {
		c = refreshStatusConnsWithAWS(c, awsConns)
		if c == nil {
			continue
		}

		connsStatus = append(connsStatus,
			&cloudv1.AWSPrivateLinkServiceConnection{
				Endpoint: id,
				Phase:    c.Phase,
				Message:  c.Message,
			})
	}

	// Sort with endpoint ID
	sort.Slice(connsStatus, func(i, j int) bool {
		return connsStatus[i].Endpoint < connsStatus[j].Endpoint
	})
	return connsStatus
}

func (t *PrivateLinkReconcileCtx) reconcilePrivateLinkServiceConnections(
	ctx context.Context,
	sc *ec2v1alpha1.ServiceConfiguration,
) ([]*cloudv1.AWSPrivateLinkServiceConnection, error) {
	conns := t.getConnsMap()
	toReject, toAccept := categorizeEndpoints(conns)
	lgr := log.FromContext(ctx)

	if len(toAccept) > 0 {
		if err := acceptEndpoints(ctx, t, toAccept, sc, conns, t.Config.DryRun.Enabled); err != nil {
			return nil, err
		}
		lgr.V(8).Info("endpoints accepted", "endpoints", toAccept)
	}

	if len(toReject) > 0 {
		if err := rejectEndpoints(ctx, t, toReject, sc, conns, t.Config.DryRun.Enabled); err != nil {
			return nil, err
		}
		lgr.V(8).Info("endpoints rejected", "endpoints", toReject)
	}

	// currentAWSConnectionsMap is got through the aws ec2 API directly
	awsConns, err := t.getAWSConnectionsMap(ctx, *sc.ServiceID)
	if err != nil {
		return nil, err
	}
	lgr.V(8).Info("connection status updated")

	// Construct connections with connection map
	return genLatestConnsStatusAWS(awsConns, conns), nil
}

func refreshStatusConnsWithAWS(
	c *cloudv1.AWSPrivateLinkServiceConnection,
	m map[string]awsEC2State,
) *cloudv1.AWSPrivateLinkServiceConnection {
	id := c.Endpoint
	currentState, ok := m[id]
	if ok {
		switch currentState {
		case awsEC2StateAvailable:
			c.Phase = cloudv1.PrivateLinkServiceConnectionPhaseActive
			c.Message = ""
		case awsEC2StateDeleted, awsEC2StateRejected:
			// Deleted
			if c.Phase == cloudv1.PrivateLinkServiceConnectionPhaseDeleting {
				return nil
			}
		case awsEC2StatePending, awsEC2StatePendingAcceptance:
			c.Phase = cloudv1.PrivateLinkServiceConnectionPhasePending
			c.Message = ""
		case awsEC2StateDeleting, awsEC2StateFailed, awsEC2StateExpired:
			// Ignore
		default:
			// Ignore
		}
		return c
	}

	// Deleted
	if c.Phase == cloudv1.PrivateLinkServiceConnectionPhaseDeleting {
		return nil
	}
	if c.Phase == cloudv1.PrivateLinkServiceConnectionPhaseActive ||
		c.Phase == cloudv1.PrivateLinkServiceConnectionPhasePending {
		// The customer deletes the endpoint by himself
		c.Phase = cloudv1.PrivateLinkServiceConnectionPhaseFailed
		c.Message = fmt.Sprintf("The Vpc Endpoint Id '%s' does not exist", id)
	}
	return c
}

func (t *PrivateLinkReconcileCtx) reconcileAllowedPrincipals(
	ctx context.Context,
	sc *ec2v1alpha1.ServiceConfiguration,
) error {
	mOpRes, err := t.ec2Cli.ModifyVpcEndpointServicePermissions(ctx,
		&ec2.ModifyVpcEndpointServicePermissionsInput{
			ServiceId:            sc.ServiceID,
			AddAllowedPrincipals: []string{"*"},
			DryRun:               &t.Config.DryRun.Enabled,
		}, func(opts *ec2.Options) {
			opts.Region = t.EnvProvider.GetRegion()
		})
	if err != nil {
		return err
	}
	if mOpRes.ReturnValue == nil ||
		!*mOpRes.ReturnValue {
		return errors.Errorf("modify vpc endpoint service permission " +
			"to '*' failed")
	}

	return nil
}

// TODO(zc): what are all these tags for?
func getAWSPrivateLinkTagSpecification(
	cluster *cloudv1.Cluster,
) []*ec2v1alpha1.TagSpecification {
	tags := common.ConcatenateTags(cloudv1.TagsFromLabels(cluster.Labels),
		map[string]string{
			common.Cluster:   cluster.Name,
			common.Component: common.Loadbalance,
			common.UsedBy:    common.CustomerTiDB,
		})

	ntags := []*ec2v1alpha1.Tag{}

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

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

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

// nolint: unused
func (t *PrivateLinkReconcileCtx) getPrivateNetworkLoadBalancerARN2(ctx context.Context) (string, error) {
	remote := t.EnvProvider.GetClient()

	svc := &corev1.Service{}

	key := client.ObjectKey{
		Namespace: t.TCGetter.Namespace(),
		Name:      getPrivateServiceName(t.TCGetter.Name()),
	}

	err := remote.Get(ctx, key, svc)
	if err != nil {
		return "", fmt.Errorf("failed to get private service: %w", err)
	}

	lbName, err := util.GetAWSLBNameFromSVC(svc)
	if err != nil {
		return "", fmt.Errorf("failed to get load balancer name from service: %w", err)
	}

	ar := t.GetAWSAssumedRoleForSDK()
	elbCli := t.AWSCli.ELBV2(ar)
	result, err := elbCli.DescribeLoadBalancers(ctx, &elbv2.DescribeLoadBalancersInput{
		Names: []string{lbName},
	})
	if err != nil {
		return "", fmt.Errorf("failed to describe load balancers: %w", err)
	}

	if len(result.LoadBalancers) == 0 {
		return "", fmt.Errorf("load balancer %s not found", lbName)
	}

	lb := result.LoadBalancers[0]
	return *lb.LoadBalancerArn, nil
}

func (t *PrivateLinkReconcileCtx) getPrivateNetworkLoadBalancerARN(
	ctx context.Context,
) (string, error) {
	svcKey := client.ObjectKey{
		Namespace: t.Namer.GetTiDBNamespace(),
		Name:      t.Namer.GetPrivateTiDBServiceName(),
	}
	c := t.tagCli
	res, err := c.GetResources(ctx, &tagging.GetResourcesInput{
		ResourceTypeFilters: []string{
			"elasticloadbalancing:loadbalancer",
		},
		TagFilters: []rgtatypes.TagFilter{
			// TODO: remove this filter
			{
				Key: pointer.String("component"),
				Values: []string{
					common.PrivateLoadbalance,
				},
			},
			// TODO(charleszheng44): only one of the
			// following tags is required?
			// TODO: remove this filter
			{
				Key: pointer.String("cluster"),
				Values: []string{
					t.Cluster.Name,
				},
			},
			{
				Key: pointer.String(LabelKeyAWSLoadBalancerResource),
				Values: []string{
					LabelValAWSLoadBalancerResource,
				},
			},
			{
				Key: pointer.String(LabelKeyAWSLoadBalancerCluster),
				Values: []string{
					t.EnvProvider.GetClusterName(),
				},
			},
			{
				Key: pointer.String(LabelKeyAWSLoadBalancerStack),
				Values: []string{
					svcKey.String(),
				},
			},
		},
	}, func(opts *tagging.Options) {
		opts.Region = t.EnvProvider.GetRegion()
	})
	if err != nil {
		return "", err
	}
	arns := []string{}
	for _, item := range res.ResourceTagMappingList {
		arns = append(arns, *item.ResourceARN)
	}
	if len(arns) == 0 {
		return "", fmt.Errorf("no loadbalancer")
	}
	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
}

// buildEndpointService build the VPCEndpointServiceConfiguration.
func buildEndpointService(
	ctx context.Context,
	t *PrivateLinkReconcileCtx,
	providerConfigName string,
) (*ec2v1alpha1.VPCEndpointServiceConfiguration, error) {
	lgr := log.FromContext(ctx)
	vesc := &ec2v1alpha1.VPCEndpointServiceConfiguration{
		ObjectMeta: metav1.ObjectMeta{
			Name: getPrivateLinkServiceName(
				t.Namer.GetTiDBNamespace(),
				t.Namer.GetTiDBName(),
			),
		},
	}

	nlbARN, err := t.getPrivateNetworkLoadBalancerARN(ctx)
	if err != nil {
		return nil, err
	}
	lgr.V(8).Info("got NLB ARN for endpoint service",
		"ARN", nlbARN, "endpointService", vesc.GetName())

	privateDNSName := t.Namer.GetPrivateLinkEndpointAddress()
	lgr.V(8).Info("got private DNS Name",
		"DNSName", privateDNSName, "endpointService", vesc.GetName())
	vesc.ObjectMeta.Labels = namer.NewOwnerLabels(t.Cluster, nil)
	vesc.Spec = ec2v1alpha1.VPCEndpointServiceConfigurationSpec{
		ResourceSpec: xpv1.ResourceSpec{
			ProviderConfigReference: &xpv1.Reference{
				Name: providerConfigName,
			},
		},
		ForProvider: ec2v1alpha1.VPCEndpointServiceConfigurationParameters{
			Region:             t.EnvProvider.GetRegion(),
			AcceptanceRequired: pointer.Bool(true),
			PrivateDNSName:     &privateDNSName,
			CustomVPCEndpointServiceConfigurationParameters: ec2v1alpha1.
				CustomVPCEndpointServiceConfigurationParameters{
				NetworkLoadBalancerARNs: []*string{
					&nlbARN,
				},
			},
			TagSpecifications: getAWSPrivateLinkTagSpecification(t.Cluster),
		},
	}
	if v, ok := t.Cluster.Annotations[eksv1alpha1.ClusterPrivateLinkServiceIDAnnKey]; ok {
		if vesc.Annotations == nil {
			vesc.Annotations = make(map[string]string)
		}
		vesc.Annotations[meta.AnnotationKeyExternalName] = v
	}

	return vesc, nil
}

// reconcileVPCEndpointServiceConfiguration reconciles the
// VPCEndpointServiceConfiguration which is the crossplane objects that used
// to control the endpoint service.
func (t *PrivateLinkReconcileCtx) reconcileVPCEndpointServiceConfiguration(
	ctx context.Context,
) (
	*ec2v1alpha1.VPCEndpointServiceConfiguration,
	error,
) {
	lgr := log.FromContext(ctx)

	build, err := buildEndpointService(ctx, t, t.GetAWSProviderConfigName())
	if err != nil {
		return nil, err
	}

	vesc := &ec2v1alpha1.VPCEndpointServiceConfiguration{
		ObjectMeta: metav1.ObjectMeta{
			Name: getPrivateLinkServiceName(
				t.Namer.GetTiDBNamespace(),
				t.Namer.GetTiDBName(),
			),
		},
	}

	op, err := controllerutil.CreateOrUpdate(ctx, t.Local, vesc, func() error {
		if vesc.CreationTimestamp.IsZero() {
			vesc.Spec = build.Spec
			vesc.Labels = build.Labels
			vesc.Annotations = build.Annotations
		} else {
			if allowRotateLBOnce(t.Cluster) {
				vesc.Spec.ForProvider.NetworkLoadBalancerARNs = build.Spec.ForProvider.NetworkLoadBalancerARNs
			}
		}
		return nil
	})
	if err != nil {
		return nil, fmt.Errorf("failed to create or update VPCEndpointServiceConfiguration: %v", err)
	}
	if op == controllerutil.OperationResultUpdated {
		lgr.Info("VPCEndpointServiceConfiguration updated", "name", vesc.GetName())
	}

	sc := vesc.Status.AtProvider.ServiceConfiguration
	if sc == nil || sc.ServiceID == nil {
		lgr.V(8).Info("endpoint service is not synced",
			"endpointService", vesc.GetName(), "cluster", t.Cluster.GetName())
		return nil, errors.New("wait crossplane vpc endpoint service sync")
	}

	if allowRotateLBOnce(t.Cluster) {
		if len(sc.NetworkLoadBalancerARNs) == 0 ||
			*sc.NetworkLoadBalancerARNs[0] != *build.Spec.ForProvider.NetworkLoadBalancerARNs[0] {
			lgr.Info("endpoint lb is not synced")
			// To remove the old lb, we need to reject the connections already associate with the old LB.
			// After update the NLB success, the connections will be accepted again and re-associated with the new LB.
			err := rejectAllConnections(ctx, t, vesc, t.Config.DryRun.Enabled)
			if err != nil {
				return nil, errors.New("failed to reject all connection")
			}
			return nil, errors.New("wait crossplane vpc endpoint service LB sync")
		}
		delete(t.Cluster.Annotations, AnnotationKeyAllowRotateLBOnce)
	}

	lgr.V(8).Info("endpoint service is synced",
		"endpointService", vesc.GetName(), "cluster", t.Cluster.GetName())
	return vesc, nil
}

func updateStatusToNil(t *PrivateLinkReconcileCtx) {
	t.Cluster.Status.SqlEndpoints.PrivateLinkService = nil
}

func updateStatusWithMessage(t *PrivateLinkReconcileCtx, msg string) {
	t.Cluster.Status.SqlEndpoints.PrivateLinkService.Message = msg
}

func updateStatusWithPhase(
	t *PrivateLinkReconcileCtx,
	phase cloudv1.PrivateLinkServicePhase,
) {
	t.Cluster.Status.SqlEndpoints.PrivateLinkService.Phase = phase
}

func rejectAllConnections(
	ctx context.Context,
	t *PrivateLinkReconcileCtx,
	vesc *ec2v1alpha1.VPCEndpointServiceConfiguration,
	dryRun bool,
) error {
	if vesc.Status.AtProvider.ServiceConfiguration == nil ||
		vesc.Status.AtProvider.ServiceConfiguration.ServiceID == nil {
		return nil
	}
	serviceID := *vesc.Status.AtProvider.ServiceConfiguration.ServiceID
	updateStatusWithMessage(t, "reject all connections")

	connectionsMap, err := t.getAWSConnectionsMap(ctx, serviceID)
	if err != nil {
		return err
	}
	ids := []string{}
	for c, state := range connectionsMap {
		if state != awsEC2StateRejected &&
			state != awsEC2StateDeleted {
			ids = append(ids, c)
		}
	}

	if len(ids) != 0 {
		rejectRes, err := t.ec2Cli.RejectVpcEndpointConnections(ctx,
			&ec2.RejectVpcEndpointConnectionsInput{
				ServiceId:      &serviceID,
				VpcEndpointIds: ids,
				DryRun:         &dryRun,
			}, func(opts *ec2.Options) {
				opts.Region = t.EnvProvider.GetRegion()
			})
		if err != nil {
			return err
		}
		if len(rejectRes.Unsuccessful) != 0 {
			return errors.Errorf("failed to reject "+
				"vpc endpoint connections: %v",
				rejectRes.Unsuccessful)
		}
	}
	return nil
}

func unregisterPrivateDNS(
	ctx context.Context,
	t *PrivateLinkReconcileCtx,
) error {
	updateStatusWithMessage(t, "remove private DNS name")
	rrs := route53v1alpha1.ResourceRecordSet{
		ObjectMeta: metav1.ObjectMeta{
			Name: t.privatelinkDnsRecordName(),
		},
	}
	if err := t.Local.Delete(ctx, &rrs); err != nil {
		if !apierrors.IsNotFound(err) {
			return err
		}
	}
	return nil
}

func deleteVpcEndpointServiceConfiguration(
	ctx context.Context,
	t *PrivateLinkReconcileCtx,
	vesc *ec2v1alpha1.VPCEndpointServiceConfiguration,
) error {
	if !vesc.DeletionTimestamp.IsZero() {
		return nil
	}
	updateStatusWithMessage(t, "delete crossplane ServiceConfiguration")
	if err := t.Local.Delete(ctx, vesc); err != nil {
		if !apierrors.IsNotFound(err) {
			return err
		}
	}
	// ServiceConfiguration has been deleted
	// Update status
	updateStatusToNil(t)
	return nil
}

func (t *PrivateLinkReconcileCtx) deletePrivateLink(
	ctx context.Context,
) deploy.Result {
	if t.Cluster.Status.SqlEndpoints.PrivateLinkService == nil {
		// the status will be nil, only after privatelink has been deleted
		return deploy.Done()
	}
	updateStatusWithPhase(t, cloudv1.PrivateLinkServicePhaseDeleting)

	vesc := &ec2v1alpha1.VPCEndpointServiceConfiguration{
		ObjectMeta: metav1.ObjectMeta{
			Name: getPrivateLinkServiceName(
				t.Namer.GetTiDBNamespace(),
				t.Namer.GetTiDBName()),
		},
	}

	if err := t.Local.Get(ctx, client.ObjectKeyFromObject(vesc),
		vesc); err != nil {
		if !apierrors.IsNotFound(err) {
			return deploy.ErrorOrDone(err)
		}
		// ServiceConfiguration has been deleted
		// Update status
		updateStatusToNil(t)
		return deploy.Done()
	}

	// Reject all connections before deleting ServiceConfiguration
	if err := rejectAllConnections(ctx, t, vesc, t.Config.DryRun.Enabled); err != nil {
		return deploy.Error(err)
	}

	// Remove private DNS record
	if err := unregisterPrivateDNS(ctx, t); err != nil {
		return deploy.Error(err)
	}

	// Remove creation-pending annotation if exist
	// Otherwise it will block deletion of crossplane object
	if vesc.Annotations != nil &&
		vesc.Annotations[AnnotationCrossplaneCreationPending] != "" {
		delete(vesc.Annotations, AnnotationCrossplaneCreationPending)
	}

	// ServiceConfiguration is not deleting, delete it
	if err := deleteVpcEndpointServiceConfiguration(
		ctx, t, vesc); err != nil {
		return deploy.Error(err)
	}

	return deploy.Done()
}

func (t *PrivateLinkReconcileCtx) privatelinkDnsRecordName() string {
	return fmt.Sprintf("%s-privatelink", t.Cluster.ClusterScopeName())
}

func (t *PrivateLink) Delete(ctx context.Context) deploy.Result {
	if t.IsRegionalManaged() {
		return deploy.Done()
	}
	if t.Config.DryRun.Enabled {
		return deploy.Done()
	}

	rCtx := PrivateLinkReconcileCtx{
		PrivateLink: t,
		ec2Cli:      t.AWSCli.EC2(t.GetAWSAssumedRoleForSDK()),
	}
	return rCtx.deletePrivateLink(ctx)
}