package eks

import (
	"context"
	"fmt"
	"time"

	elbv2 "github.com/aws/aws-sdk-go-v2/service/elasticloadbalancingv2"
	tagging "github.com/aws/aws-sdk-go-v2/service/resourcegroupstaggingapi"
	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"
	"k8s.io/apimachinery/pkg/api/equality"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/pointer"
	ctrl "sigs.k8s.io/controller-runtime"
	"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"
	"github.com/tidbcloud/eks-provider/controllers/cluster/task/utils"
	"github.com/tidbcloud/eks-provider/utils/tidb-operator/tidbcluster"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/cloud/aws"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/controllers/common"
	route53pause "github.com/tidbcloud/infra-provider/pkg/crossplane/route53"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	"github.com/tidbcloud/infra-provider/pkg/util"
)

// ref https://kubernetes-sigs.github.io/aws-load-balancer-controller/v2.2/guide/service/annotations/#traffic-routing
const awsLBNameAnnKey = "service.beta.kubernetes.io/aws-load-balancer-name"

const (
	legacyPrivateRecordSuffix = "-inpublic"
	publicRecordSuffix        = "-public"
)

type FullService struct {
	*task.BaseTask
	awsCli aws.Client
}

var _ deploy.Task = &FullService{}

func NewFullService(ctx *task.Context, awsCli aws.Client) *FullService {
	return &FullService{BaseTask: &task.BaseTask{Context: ctx}, awsCli: awsCli}
}

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

func (t *FullService) Create(ctx context.Context) deploy.Result {
	if t.TCGetter == nil || t.TCGetter.IsNil() {
		return deploy.Fatal(fmt.Errorf("tidb cluster is not ready yet"))
	}

	ar := t.GetAWSAssumedRoleForSDK()
	elbCli := t.awsCli.ELBV2(ar)
	tagCli := t.awsCli.ResourceGroupsTagging(ar)

	remote := t.EnvProvider.GetClient()

	res, err := t.reconcileEndpoint(
		ctx,
		t.Local,
		t.Cluster,
		remote,
		t.Namer,
		t.EnvProvider.GetRegion(),
		t.Config.FullManage.PublicHostedZoneDomain,
		t.Config.FullManage.PublicHostedZoneID,
		elbCli,
		tagCli,
		t.Config.ControlplaneAWSProviderConfig, // create dns alias record in central aws account
	)
	if err != nil {
		return deploy.Error(err)
	}
	if !res.IsZero() {
		return deploy.Done().WithRequeuAfter(res.RequeueAfter)
	}

	res, err = t.reconcilePrivateEndpoint(
		ctx,
		t.Local,
		t.Cluster,
		remote,
		t.Namer,
		t.EnvProvider.GetRegion(),
		t.Config.FullManage.PublicHostedZoneDomain,
		t.Config.FullManage.PublicHostedZoneID,
		elbCli,
		tagCli,
		t.Config.ControlplaneAWSProviderConfig, // create dns alias record in central aws account
	)
	if err != nil {
		return deploy.Error(err)
	}
	if !res.IsZero() {
		return deploy.Done().WithRequeuAfter(res.RequeueAfter)
	}

	return deploy.Done()
}

func (t *FullService) PostCreate(ctx context.Context) deploy.Result {
	if t.Cluster.IsDNSDisabled(t.Cluster.GetDefaultGroupID()) {
		return deploy.Done()
	}

	recordNames := make([]string, 0, 2)
	if t.Cluster.Spec.SqlEndpoints.Public.IsEnabled() {
		recordNames = append(recordNames, buildPublicRRSName(t.Cluster))
	}
	if t.Cluster.Spec.SqlEndpoints.Private != nil && t.Cluster.Spec.SqlEndpoints.Private.IsEnabled() {
		recordNames = append(recordNames, buildPrivateRRSName(t.Cluster))
	}

	needsRequeue := false
	for _, name := range recordNames {
		recordNeedsRequeue, err := route53pause.ReconcilePause(ctx, t.Local, name)
		if err != nil {
			return deploy.Error(err)
		}
		needsRequeue = needsRequeue || recordNeedsRequeue
	}
	if needsRequeue {
		return deploy.Done().WithRequeuAfter(route53pause.RequeueAfter)
	}
	return deploy.Done()
}

func (t *FullService) Delete(ctx context.Context) deploy.Result {
	err := deleteAWSDNSRecord(ctx, t.Local, t.Cluster.ClusterScopeName()+publicRecordSuffix)
	if err != nil {
		return deploy.Error(err)
	}

	remote := t.EnvProvider.GetClient()

	err = t.deletePrivateEndpoint(ctx, t.Local, t.Cluster, remote)
	if err != nil {
		return deploy.Error(err)
	}

	return deploy.Done()
}

func (t *FullService) reconcileEndpoint(
	ctx context.Context,
	infraClient client.Client,
	cluster *cloudv1.Cluster,
	tcClient client.Client,
	ctxNamer *namer.Namer,
	region string,
	publicHostedZoneDomain string,
	publicHostedZoneID string,
	elbCli *elbv2.Client,
	tagCli *tagging.Client,
	providerConfigForCentralAccount string,
) (res ctrl.Result, err error) {
	waitDuration := time.Second * 10
	logger := log.FromContext(ctx)

	need := cluster.Spec.SqlEndpoints.Public.IsEnabled()

	// ResourceID was introduced on 2024-03-13 by https://github.com/tidbcloud/infra-provider/pull/2880/files.
	// The clusters was created before that PR had already set .Status.SqlEndpoints.PublicEndpoint, so that this code
	// can't set ResourceID forever.
	//
	// After this fix released, we can remove the check for ResourceID.
	ready := cluster.Status.SqlEndpoints.PublicEndpoint != nil &&
		cluster.Status.SqlEndpoints.PublicEndpoint.ResourceID != "" &&
		cluster.Status.SqlEndpoints.PublicEndpoint.Host != ""

	if need && !ready {
		key := types.NamespacedName{Namespace: t.Namer.GetTiDBNamespace(), Name: t.Namer.GetPublicTiDBServiceName()}
		svc := new(corev1.Service)
		err := tcClient.Get(ctx, key, svc)
		if err != nil {
			if apierrors.IsNotFound(err) {
				logger.Info("svc not created yet")
				return ctrl.Result{RequeueAfter: waitDuration}, nil
			}
			return ctrl.Result{}, err
		}

		if len(svc.Status.LoadBalancer.Ingress) == 0 {
			logger.Info("svc.Status.LoadBalancer.Ingress not ready")
			return ctrl.Result{RequeueAfter: waitDuration}, nil
		}

		lbName, err := util.GetAWSLBNameFromSVC(svc)
		if err != nil {
			return ctrl.Result{}, err
		}

		ingress := svc.Status.LoadBalancer.Ingress[0]

		// add an dns alias to the loadbalancer
		dnsName := ctxNamer.GetPublicEndpointAddress()
		alias := DnsAlias{
			Name:         cluster.ClusterScopeName() + publicRecordSuffix,
			Alias:        dnsName,
			HostedZoneId: publicHostedZoneID,
			Address:      ingress.Hostname,
		}

		if !cluster.IsDNSDisabled(t.Cluster.GetDefaultGroupID()) {
			err = ensureAWSDNSRecord(ctx, infraClient, elbCli, lbName, alias, region, providerConfigForCentralAccount, nil)
			if err != nil {
				return ctrl.Result{}, err
			}
		}

		arn, err := t.getLoadBalancerARN(ctx, elbCli, lbName)
		if err != nil {
			return ctrl.Result{}, err
		}

		status := &cloudv1.SqlEndpointStatus{
			Host:       utils.GetEffectiveHostName(dnsName, cluster.IsDNSDisabled(cluster.GetDefaultGroupID())),
			Port:       int(cluster.GetPublicSQLEndpointPort()),
			ResourceID: arn,
		}
		if !equality.Semantic.DeepEqual(cluster.Status.SqlEndpoints.PublicEndpoint, status) {
			cluster.Status.SqlEndpoints.PublicEndpoint = status
			err = infraClient.Status().Update(ctx, cluster)
			if err != nil {
				return ctrl.Result{}, err
			}
		}
	}

	if need {
		if cluster.IsDNSDisabled(cluster.GetDefaultGroupID()) {
			rrsName := buildPublicRRSName(cluster)
			logger.Info("delete public dns record due to disable-dns annotation is set", "rrsName", rrsName)
			err := deleteAWSDNSRecord(ctx, infraClient, rrsName)
			if err != nil {
				return ctrl.Result{}, err
			}

			if cluster.Status.SqlEndpoints.PublicEndpoint != nil && cluster.Status.SqlEndpoints.PublicEndpoint.Host != "" {
				cluster.Status.SqlEndpoints.PublicEndpoint.Host = ""
				err = infraClient.Status().Update(ctx, cluster)
				if err != nil {
					return ctrl.Result{}, err
				}
			}
		}
	}

	if !need && ready {
		err := deleteAWSDNSRecord(ctx, infraClient, cluster.ClusterScopeName()+publicRecordSuffix)
		if err != nil {
			return ctrl.Result{}, err
		}

		if cluster.Status.SqlEndpoints.PublicEndpoint != nil {
			cluster.Status.SqlEndpoints.PublicEndpoint = nil

			err = infraClient.Status().Update(ctx, cluster)
			if err != nil {
				return ctrl.Result{}, err
			}
		}
	}

	return ctrl.Result{}, nil
}

func (t *FullService) getLoadBalancerARN(ctx context.Context, elbCli *elbv2.Client, lbName string) (string, error) {
	result, err := elbCli.DescribeLoadBalancers(ctx, &elbv2.DescribeLoadBalancersInput{
		Names: []string{lbName},
	})
	if err != nil {
		return "", err
	}

	if len(result.LoadBalancers) == 0 || result.LoadBalancers[0].LoadBalancerArn == nil {
		return "", fmt.Errorf("failed get loadbalancer arn")
	}
	arn := result.LoadBalancers[0].LoadBalancerArn
	return *arn, nil
}

func (t *FullService) reconcilePrivateEndpoint(
	ctx context.Context,
	infraClient client.Client,
	cluster *cloudv1.Cluster,
	tcClient client.Client,
	ctxNamer *namer.Namer,
	region string,
	publicHostedZoneDomain string,
	publicHostedZoneID string,
	elbCli *elbv2.Client,
	tagCli *tagging.Client,
	providerConfigForCentralAccount string,
) (res ctrl.Result, err error) {
	waitDuration := time.Second * 10
	logger := log.FromContext(ctx)

	need := cluster.Spec.SqlEndpoints.Private != nil && cluster.Spec.SqlEndpoints.Private.IsEnabled()
	ready := cluster.Status.SqlEndpoints.PrivateEndpoint != nil &&
		cluster.Status.SqlEndpoints.PrivateEndpoint.ResourceID != "" && // see `reconcileEndpoint` to get why we need this
		cluster.Status.SqlEndpoints.PrivateEndpoint.Host != ""

	if need {
		svc := &corev1.Service{
			ObjectMeta: metav1.ObjectMeta{
				Name:      getPrivateServiceName(ctxNamer.GetTiDBName()),
				Namespace: ctxNamer.GetTiDBNamespace(),
			},
		}

		_, err = controllerutil.CreateOrUpdate(ctx, tcClient, svc, func() error {
			t.setPrivateService(cluster, t.TCGetter, svc)
			return controllerutil.SetControllerReference(t.TCGetter.Object(), svc, tcClient.Scheme())
		})
		if err != nil {
			return ctrl.Result{}, err
		}

		if !ready {
			if len(svc.Status.LoadBalancer.Ingress) == 0 {
				logger.Info("svc.Status.LoadBalancer.Ingress not ready")
				return ctrl.Result{RequeueAfter: waitDuration}, nil
			}

			lbName, err := util.GetAWSLBNameFromSVC(svc)
			if err != nil {
				return ctrl.Result{}, err
			}

			ingress := svc.Status.LoadBalancer.Ingress[0]

			dnsName := ctxNamer.GetPrivateEndpointAddress()
			alias := DnsAlias{
				Name:         cluster.ClusterScopeName() + "-private" + legacyPrivateRecordSuffix,
				Alias:        dnsName,
				HostedZoneId: publicHostedZoneID,
				Address:      ingress.Hostname,
			}

			if !cluster.IsDNSDisabled(cluster.GetDefaultGroupID()) {
				err = ensureAWSDNSRecord(ctx, infraClient, elbCli, lbName, alias, region, providerConfigForCentralAccount, nil)
				if err != nil {
					return ctrl.Result{}, err
				}
			}

			arn, err := t.getLoadBalancerARN(ctx, elbCli, lbName)
			if err != nil {
				return ctrl.Result{}, err
			}

			status := &cloudv1.SqlEndpointStatus{
				Host:       utils.GetEffectiveHostName(dnsName, cluster.IsDNSDisabled(cluster.GetDefaultGroupID())),
				Port:       int(cluster.GetPrivateSQLEndpointPort()),
				ResourceID: arn,
			}
			if !equality.Semantic.DeepEqual(cluster.Status.SqlEndpoints.PrivateEndpoint, status) {
				cluster.Status.SqlEndpoints.PrivateEndpoint = status
				err = infraClient.Status().Update(ctx, cluster)
				if err != nil {
					return ctrl.Result{}, err
				}
			}
		}

		if cluster.IsDNSDisabled(cluster.GetDefaultGroupID()) {
			rrsName := buildPrivateRRSName(cluster)
			logger.Info("delete private dns record due to disable-dns annotation is set", "rrsName", rrsName)
			err := deleteAWSDNSRecord(ctx, infraClient, rrsName)
			if err != nil {
				return ctrl.Result{}, err
			}

			if cluster.Status.SqlEndpoints.PrivateEndpoint != nil && cluster.Status.SqlEndpoints.PrivateEndpoint.Host != "" {
				cluster.Status.SqlEndpoints.PrivateEndpoint.Host = ""
				err = infraClient.Status().Update(ctx, cluster)
				if err != nil {
					return ctrl.Result{}, err
				}
			}
		}
	}

	if !need && ready {
		err = t.deletePrivateEndpoint(ctx, infraClient, cluster, tcClient)
		if err != nil {
			return ctrl.Result{}, err
		}

		if cluster.Status.SqlEndpoints.PrivateEndpoint != nil {
			cluster.Status.SqlEndpoints.PrivateEndpoint = nil
			err = infraClient.Status().Update(ctx, cluster)
			if err != nil {
				return ctrl.Result{}, err
			}
		}
	}

	return ctrl.Result{}, nil
}

func (t *FullService) deletePrivateEndpoint(ctx context.Context, infraClient client.Client, cluster *cloudv1.Cluster, tcClient client.Client) error {
	svc := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:      t.Namer.GetPrivateTiDBServiceName(),
			Namespace: t.Namer.GetTiDBNamespace(),
		},
	}

	if err := k8s.DeleteObjectCompleted(ctx, tcClient, svc); err != nil {
		return err
	}

	err := deleteAWSDNSRecord(ctx, infraClient, cluster.ClusterScopeName()+"-private"+legacyPrivateRecordSuffix)
	if err != nil {
		return err
	}

	return nil
}

func (t *FullService) setPrivateService(cluster *cloudv1.Cluster, tc tidbcluster.Getter, svc *corev1.Service) {
	tags := common.ConcatenateTags(cloudv1.TagsFromLabels(cluster.Labels),
		map[string]string{
			common.Cluster:   cluster.Name,
			common.Component: common.PrivateLoadbalance,
			common.UsedBy:    common.CustomerTiDB,
		})

	// See https://kubernetes-sigs.github.io/aws-load-balancer-controller/v2.4/guide/service/annotations
	annotation := map[string]string{
		"service.beta.kubernetes.io/aws-load-balancer-type":                     "external",
		"service.beta.kubernetes.io/aws-load-balancer-nlb-target-type":          "instance",
		"service.beta.kubernetes.io/aws-load-balancer-scheme":                   "internal",
		"service.beta.kubernetes.io/aws-load-balancer-additional-resource-tags": cloudv1.TagStringForAWS(tags),
		"service.beta.kubernetes.io/aws-load-balancer-target-group-attributes":  "preserve_client_ip.enabled=false",
	}

	utils.SetNLBAnnoToEnablePPV2(cluster, annotation)
	utils.SetNLBAnnoForTiproxy(annotation, t.TCGetter.IsTLSEnabled(), t.Cluster.Spec.SqlEndpoints.UseTiProxyAsLBTarget)

	// switch to ip mode for clusters enabled dedicated SG
	if t.Cluster.FeatureEnabled(consts.ClusterFeatureDedicatedSG) {
		annotation["service.beta.kubernetes.io/aws-load-balancer-nlb-target-type"] = "ip"
	}

	if v, ok := cluster.Annotations[eksv1alpha1.ClusterPrivateServiceLBNameAnnKey]; ok {
		annotation[awsLBNameAnnKey] = v
	}
	if cluster.Spec.SqlEndpoints.PrivateLinkNLBIPAddressType != "" {
		annotation[utils.AWSLoadBalancerIPAddressTypeAnnotation] = string(cluster.Spec.SqlEndpoints.PrivateLinkNLBIPAddressType)
	}
	// p.NodePort may be set automatically, so set field one by one to avoid diff
	if len(svc.Spec.Ports) != 1 {
		svc.Spec.Ports = []corev1.ServicePort{{}}
	}
	p := &svc.Spec.Ports[0]
	p.Name = "mysql-client"
	p.Port = cluster.GetPrivateSQLEndpointPort()
	p.TargetPort = cluster.Spec.SqlEndpoints.GetTargetPort()
	p.Protocol = corev1.ProtocolTCP

	svc.ObjectMeta.Annotations = annotation
	svc.Spec.Type = corev1.ServiceTypeLoadBalancer
	svc.Spec.SessionAffinity = corev1.ServiceAffinityNone
	svc.Spec.ExternalTrafficPolicy = corev1.ServiceExternalTrafficPolicyTypeLocal
	svc.Spec.LoadBalancerSourceRanges = utils.NLBSourceRanges(cluster.Spec.SqlEndpoints.PrivateLinkNLBIPAddressType)
	svc.Spec.Selector = map[string]string{
		consts.InstanceLabelKey:  tc.Name(),
		consts.ComponentLabelKey: string(cluster.Spec.SqlEndpoints.GetTargetComponent()),
	}
}

func getPrivateServiceName(tcName string) string {
	return fmt.Sprintf("%s-tidb-private", tcName)
}

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

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

type DnsAlias struct {
	// name is a readable name of the record
	Name string
	// alias is the alias address
	Alias string
	// address is the source address
	Address string
	// hostedZoneId is the dns hosted zone id of the source address
	HostedZoneId string
}

func ensureAWSDNSRecord(
	ctx context.Context,
	cli client.Client,
	elbCli *elbv2.Client,
	lbName string,
	alias DnsAlias,
	region string,
	providerConfigName string,
	labels map[string]string,
) error {
	result, err := elbCli.DescribeLoadBalancers(ctx, &elbv2.DescribeLoadBalancersInput{
		Names: []string{lbName},
	})
	if err != nil {
		return errors.Annotate(err, "get hosted zone of AWS ELB")
	}
	if len(result.LoadBalancers) == 0 || result.LoadBalancers[0].CanonicalHostedZoneId == nil {
		return errors.Annotatef(err, "AWS ELB not found for address %s", alias.Address)
	}
	elbHostedZoneId := *result.LoadBalancers[0].CanonicalHostedZoneId

	// first check if the ResourceRecordSet is being deleting, if so, return err, to avoid race condition
	// ensure the ResourceRecordSet is not being deleted when we're trying to create or update it
	existingRRSet := &route53v1alpha1.ResourceRecordSet{
		ObjectMeta: metav1.ObjectMeta{
			Name: alias.Name,
		},
	}
	err = cli.Get(ctx, types.NamespacedName{
		Name: alias.Name,
	}, existingRRSet)
	// failed to get existing ResourceRecordSet, but not because it's not found, return err
	if err != nil && !apierrors.IsNotFound(err) {
		return errors.Annotate(err, "get existing ResourceRecordSet")
	}
	// ResourceRecordSet is being deleted, return err
	if existingRRSet.DeletionTimestamp != nil {
		return errors.Annotate(err, "ResourceRecordSet is being deleted")
	}

	// create or update ResourceRecordSet if it doesn't exist
	rrs := &route53v1alpha1.ResourceRecordSet{
		ObjectMeta: metav1.ObjectMeta{
			Name:   alias.Name,
			Labels: labels,
			Annotations: map[string]string{
				// external name will be used as the A record alias address
				meta.AnnotationKeyExternalName: alias.Alias,
			},
		},
	}

	_, err = k8s.CreateOrUpdate(ctx, cli, rrs, func() error {
		setRRSSpec(rrs, providerConfigName, elbHostedZoneId, alias)
		return nil
	})
	return err
}

func setRRSSpec(rrs *route53v1alpha1.ResourceRecordSet, providerConfigName, elbHostedZoneId string, alias DnsAlias) {
	if rrs.Spec.ResourceSpec.ProviderConfigReference == nil {
		rrs.Spec.ResourceSpec.ProviderConfigReference = &xpv1.Reference{}
	}
	rrs.Spec.ResourceSpec.ProviderConfigReference.Name = providerConfigName
	if rrs.Spec.ForProvider.AliasTarget == nil {
		rrs.Spec.ForProvider.AliasTarget = &route53v1alpha1.AliasTarget{}
	}
	rrs.Spec.ForProvider.AliasTarget.DNSName = fmt.Sprintf("dualstack.%s", alias.Address)
	rrs.Spec.ForProvider.AliasTarget.EvaluateTargetHealth = false
	rrs.Spec.ForProvider.AliasTarget.HostedZoneID = elbHostedZoneId
	rrs.Spec.ForProvider.Type = "A"
	rrs.Spec.ForProvider.ZoneID = pointer.String(alias.HostedZoneId)
}

func deleteAWSDNSRecord(ctx context.Context, cli client.Client, name string) error {
	rrs := &route53v1alpha1.ResourceRecordSet{
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
		},
	}
	if err := k8s.DeleteObjectCompletedWithHooks(ctx, cli, rrs, k8s.HookAutoUnpauseCrossplaneCR); err != nil {
		return err
	}

	return nil
}

func buildPublicRRSName(cluster *cloudv1.Cluster) string {
	return cluster.ClusterScopeName() + publicRecordSuffix
}

func buildPrivateRRSName(cluster *cloudv1.Cluster) string {
	return fmt.Sprintf("%s-private%s", cluster.ClusterScopeName(), legacyPrivateRecordSuffix)
}