package eks

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

	awssdk "github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/ec2"
	ec2v1beta1 "github.com/crossplane-contrib/provider-aws/apis/ec2/v1beta1"
	eksv1beta1 "github.com/crossplane-contrib/provider-aws/apis/eks/v1beta1"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	"github.com/google/go-cmp/cmp"
	"github.com/samber/lo"
	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/deploy"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	"github.com/tidbcloud/infra-provider/pkg/util"
	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/appbox/task"
	"k8s.io/apimachinery/pkg/api/equality"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	subnetTagRolePublicELB   = "kubernetes.io/role/elb"
	subnetTagRoleInternalELB = "kubernetes.io/role/internal-elb"
)

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

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

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

func (t *EKS) buildEKS(ctx context.Context) (*eksv1beta1.Cluster, error) {
	// we need to overwrite some name for migrated appbox
	eksName := namer.UniqueIdentifier(t.APPBox.ObjectMeta)
	if name := t.APPBox.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameEKS]; name != "" {
		eksName = name
	}
	providerConfigName := t.ProviderConfigName()
	if name := t.APPBox.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameProvidreConfig]; name != "" {
		providerConfigName = name
	}
	tags := cloudv1.TagsFromLabels(t.APPBox.Labels)
	tags = util.AddResourceName(tags, eksName)
	tags[consts.TagKeyCrossplaneVendor] = consts.TagValCrossplaneVendor
	version := t.APPBox.Spec.KubernetesConfig.Version
	cluster := &eksv1beta1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:   eksName,
			Labels: t.Namer.Labels(),
		},
		Spec: eksv1beta1.ClusterSpec{
			ForProvider: eksv1beta1.ClusterParameters{
				Region:  pointer.String(t.APPBox.Spec.Region),
				Version: pointer.String(version), // this is the only filed can be update after creation.
				ResourcesVpcConfig: eksv1beta1.VpcConfigRequest{
					EndpointPrivateAccess: pointer.Bool(true),
					EndpointPublicAccess:  pointer.Bool(t.Config.APPBoxConfig.EnableK8sPublicAccess),
				},
				RoleArn: util.GetRoleARN(t.VPC.Spec.AccountRef.AccountID, t.Config.APPBoxConfig.ServiceRoleName),
				Tags:    tags,
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{Name: providerConfigName},
			},
		},
	}

	cluster.Spec.ForProvider.ResourcesVpcConfig.SecurityGroupIDs = append(cluster.Spec.ForProvider.ResourcesVpcConfig.SecurityGroupIDs, t.VPC.Status.SecurityGroup.SecurityGroupID)
	_, subnetRefs, err := t.controlPlaneSubnets()
	if err != nil {
		return nil, err
	}
	cluster.Spec.ForProvider.ResourcesVpcConfig.SubnetIDRefs = subnetRefs

	if t.Config.APPBoxConfig.EnableK8sPublicAccess {
		cluster.Spec.ForProvider.ResourcesVpcConfig.PublicAccessCidrs = t.Config.APPBoxConfig.ApiserverPublicAccessCidrs
	}
	if err := t.ignoreChanges(ctx, cluster); err != nil {
		return nil, err
	}
	return cluster, nil
}

func (t *EKS) ignoreChanges(ctx context.Context, newCluster *eksv1beta1.Cluster) error {
	key := types.NamespacedName{
		Name: newCluster.GetName(),
	}
	existing := &eksv1beta1.Cluster{}
	if err := t.Local.Get(ctx, key, existing); err != nil {
		return client.IgnoreNotFound(err)
	}
	// ignore the changes on roleArn, since it is immutable.
	newCluster.Spec.ForProvider.RoleArn = existing.Spec.ForProvider.RoleArn
	return nil
}

// IsSafeToApplyEKS check the generated eks cr has the same field as the existed eks cr,
// but only focuse on some fileds.
func (t *EKS) IsSafeToApplyEKS(ctx context.Context, existedCluster *eksv1beta1.Cluster) (bool, error) {
	cluster, err := t.buildEKS(ctx)
	if err != nil {
		return false, err
	}
	if cluster.Spec.ProviderConfigReference.Name != existedCluster.Spec.ProviderConfigReference.Name {
		t.Logger.Info("detect diff on Spec.ProviderConfigReference.Name",
			"key", client.ObjectKeyFromObject(existedCluster).String(),
			"generated", cluster.Spec.ProviderConfigReference.Name,
			"existed", existedCluster.Spec.ProviderConfigReference.Name,
		)
		return false, nil
	}
	generated := cluster.Spec.ForProvider      // the generated cr
	existed := existedCluster.Spec.ForProvider // the existed cr
	if *generated.Region != *existed.Region {
		t.Logger.Info("detect diff on Spec.ForProvider.Region",
			"key", client.ObjectKeyFromObject(existedCluster).String(),
			"generated", *generated.Region,
			"existed", *existed.Region,
		)
		return false, nil
	}
	if *generated.Version != *existed.Version {
		t.Logger.Info("detect diff on Spec.ForProvider.Version",
			"key", client.ObjectKeyFromObject(existedCluster).String(),
			"generated", *generated.Region,
			"existed", *existed.Region,
		)
		return false, nil
	}
	if *generated.ResourcesVpcConfig.EndpointPublicAccess != *existed.ResourcesVpcConfig.EndpointPublicAccess {
		t.Logger.Info("detect diff on Spec.ForProvider.ResourcesVpcConfig.EndpointPublicAccess",
			"key", client.ObjectKeyFromObject(existedCluster).String(),
			"generated", *generated.ResourcesVpcConfig.EndpointPublicAccess,
			"existed", *existed.ResourcesVpcConfig.EndpointPublicAccess,
		)
		return false, nil
	}
	if *generated.ResourcesVpcConfig.EndpointPrivateAccess != *existed.ResourcesVpcConfig.EndpointPrivateAccess {
		t.Logger.Info("detect diff on Spec.ForProvider.ResourcesVpcConfig.EndpointPrivateAccess",
			"key", client.ObjectKeyFromObject(existedCluster).String(),
			"generated", *generated.ResourcesVpcConfig.EndpointPrivateAccess,
			"existed", *existed.ResourcesVpcConfig.EndpointPrivateAccess,
		)
		return false, nil
	}
	if generated.RoleArn != existed.RoleArn {
		t.Logger.Info("detect diff on Spec.ForProvider.RoleArn",
			"key", client.ObjectKeyFromObject(existedCluster).String(),
			"generated", generated.RoleArn,
			"existed", existed.RoleArn,
		)
		return false, nil
	}
	// these three tags should not take into consideration when perform the equality checking.
	// ExternalResourceTagKeyKind     = "crossplane-kind"
	// ExternalResourceTagKeyName     = "crossplane-name"
	// ExternalResourceTagKeyProvider = "crossplane-providerconfig"
	existedTags := map[string]string{}
	for k, v := range existed.Tags {
		switch k {
		case "crossplane-kind", "crossplane-name", "crossplane-providerconfig":
			continue
		}
		existedTags[k] = v
	}
	// if the generated tags contains all existed tags, we consider it safe to apply.
	// thus adding tag on existed eks is consider as a safe operation.
	for k, v := range existedTags {
		if generated.Tags[k] != v {
			t.Logger.Info("detect diff on Spec.ForProvider.Tags",
				"key", client.ObjectKeyFromObject(existedCluster).String(),
				"generated", generated.Tags,
				"existed", existedTags,
			)
			return false, nil
		}
	}
	if fmt.Sprint(generated.ResourcesVpcConfig.SecurityGroupIDs) != fmt.Sprint(existed.ResourcesVpcConfig.SecurityGroupIDs) {
		t.Logger.Info("detect diff on Spec.ForProvider.ResourcesVpcConfig.SecurityGroupIDs",
			"key", client.ObjectKeyFromObject(existedCluster).String(),
			"generated", generated.ResourcesVpcConfig.SecurityGroupIDs,
			"existed", existed.ResourcesVpcConfig.SecurityGroupIDs,
		)
		return false, nil
	}
	generatedCidrsSorted := make([]string, len(generated.ResourcesVpcConfig.PublicAccessCidrs))
	copy(generatedCidrsSorted, generated.ResourcesVpcConfig.PublicAccessCidrs)
	existedCidrsSorted := make([]string, len(existed.ResourcesVpcConfig.PublicAccessCidrs))
	copy(existedCidrsSorted, existed.ResourcesVpcConfig.PublicAccessCidrs)
	sort.Strings(generatedCidrsSorted)
	sort.Strings(existedCidrsSorted)

	if !cmp.Equal(generatedCidrsSorted, existedCidrsSorted) {
		t.Logger.Info("detect diff on Spec.ForProvider.ResourcesVpcConfig.PublicAccessCidrs",
			"key", client.ObjectKeyFromObject(existedCluster).String(),
			"generated", generated.ResourcesVpcConfig.PublicAccessCidrs,
			"existed", existed.ResourcesVpcConfig.PublicAccessCidrs,
		)
		return false, nil
	}
	// this is a workaround comparison.
	if len(existed.ResourcesVpcConfig.SubnetIDRefs) == 0 {
		ids, refs, err := t.controlPlaneSubnets()
		if err != nil {
			return false, err
		}
		if fmt.Sprint(generated.ResourcesVpcConfig.SubnetIDRefs) != fmt.Sprint(refs) {
			t.Logger.Info("detect diff on Spec.ForProvider.ResourcesVpcConfig.SubnetIDRefs",
				"key", client.ObjectKeyFromObject(existedCluster).String(),
				"generated", generated.ResourcesVpcConfig.SubnetIDRefs,
				"existed", refs,
			)
			return false, nil
		}
		if fmt.Sprint(ids) != fmt.Sprint(existed.ResourcesVpcConfig.SubnetIDs) {
			t.Logger.Info("detect diff on Spec.ForProvider.ResourcesVpcConfig.SubnetID",
				"key", client.ObjectKeyFromObject(existedCluster).String(),
				"generated", ids,
				"existed", existed.ResourcesVpcConfig.SubnetIDs,
			)
			return false, nil
		}
	} else {
		if fmt.Sprint(generated.ResourcesVpcConfig.SubnetIDRefs) != fmt.Sprint(existed.ResourcesVpcConfig.SubnetIDRefs) {
			t.Logger.Info("detect diff on Spec.ForProvider.ResourcesVpcConfig.SubnetIDRefs",
				"key", client.ObjectKeyFromObject(existedCluster).String(),
				"generated", generated.ResourcesVpcConfig.SubnetIDRefs,
				"existed", existed.ResourcesVpcConfig.SubnetIDRefs,
			)
			return false, nil
		}
	}
	return true, nil
}

func (t *EKS) Create(ctx context.Context) deploy.Result {
	t.BeginAt = util.Pointer(time.Now())

	if t.APPBox.Annotations[v1alpha1.AnnotationKeyLegacyAPPBoxSkipEKSTask] == "true" {
		return deploy.Done()
	}
	cluster, err := t.buildEKS(ctx)
	if err != nil {
		return deploy.Panic(err)
	}
	return deploy.PanicOrDone(k8s.Apply(ctx, t.Local, cluster))
}

func (t *EKS) PostCreate(ctx context.Context) (res deploy.Result) {
	key := types.NamespacedName{
		Name: namer.UniqueIdentifier(t.APPBox.ObjectMeta),
	}
	// overwrite Name for migrated data.
	if name := t.APPBox.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameEKS]; name != "" {
		key.Name = name
	}
	var c eksv1beta1.Cluster
	err := t.Local.Get(ctx, key, &c)
	if err != nil {
		return deploy.Panic(err)
	}

	t.Cluster = task.NewEKSClusterIdentifier(&c)
	t.APPBox.Status.KubernetesClusterName = c.Name
	status := c.Status.AtProvider.Status
	if status != eksv1beta1.ClusterStatusActive {
		t.APPBox.Status.Conditions.SetCondition(metav1.Condition{
			Type:               v1alpha1.APPBoxConditionK8SReady,
			Status:             metav1.ConditionFalse,
			Reason:             "Creating",
			Message:            fmt.Sprintf("EKS Cluster has the status of %s", status),
			LastTransitionTime: metav1.Now(),
		})
		return deploy.Error(fmt.Errorf("eks cluster status not ready: %v", status)).WithRequeuAfter(3 * time.Second)
	}

	t.APPBox.Status.Conditions.SetCondition(metav1.Condition{
		Type:               v1alpha1.APPBoxConditionK8SReady,
		Status:             metav1.ConditionTrue,
		Reason:             "EKSReady",
		Message:            "EKS Creation succeeded",
		LastTransitionTime: metav1.Now(),
	})
	eksSecurityGroupID := c.Status.AtProvider.ResourcesVpcConfig.ClusterSecurityGroupID
	if err = t.authorizeSecurityGroupIngress(ctx, eksSecurityGroupID); err != nil {
		return deploy.Error(fmt.Errorf("update eks cluster securitygroup rule failed, %w", err))
	}
	if err = t.addSecureGroupRuleForPrivateLinkEndPoint(ctx, eksSecurityGroupID); err != nil {
		return deploy.Error(fmt.Errorf("update VPC securitygroup rule failed, %w", err))
	}
	if err = t.setSubnetTags(ctx, c.Name); err != nil {
		return deploy.Error(fmt.Errorf("update subNet tags failed: %w", err))
	}
	// remove the skip annotation if it is safe to apply eks for legacy appbox.
	if t.APPBox.Annotations[v1alpha1.AnnotationKeyLegacyAPPBoxSkipEKSTask] == "true" {
		ok, err := t.IsSafeToApplyEKS(ctx, &c)
		if err != nil {
			return deploy.Error(fmt.Errorf("check safe to apply eks for legacy appbox failed: %w", err))
		}
		if ok {
			delete(t.APPBox.Annotations, v1alpha1.AnnotationKeyLegacyAPPBoxSkipEKSTask)
		}
	}

	return deploy.Done()
}

func (t *EKS) Delete(ctx context.Context) deploy.Result {
	cluster := &eksv1beta1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name: namer.UniqueIdentifier(t.APPBox.ObjectMeta),
		},
	}
	// overwrite Name for migrated data.
	if t.APPBox.Labels[v1alpha1.IsLegacyKey] == "true" {
		name := t.APPBox.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameEKS]
		if name == "" {
			return deploy.Panic(fmt.Errorf("appbox is legacy but own-crossplane-name-eks is empty"))
		}
		cluster.Name = name
	}
	err := t.Local.Get(ctx, client.ObjectKeyFromObject(cluster), cluster)
	if err != nil {
		if client.IgnoreNotFound(err) != nil {
			return deploy.Error(err)
		}
		// cluster is deleted
		return deploy.Done()
	}
	if err := t.removeSecureGroupRuleDependency(ctx, cluster.Status.AtProvider.ResourcesVpcConfig.ClusterSecurityGroupID); err != nil {
		return deploy.Error(err)
	}
	if err := k8s.DeleteObject(ctx, t.Local, cluster); err != nil {
		return deploy.Error(err)
	}
	if err := k8s.PatchObject(ctx, t.Local, cluster, k8s.RemoveCrossplaneExternalCreatePending, k8s.DeletionTrigger); err != nil {
		return deploy.Error(err)
	}
	return deploy.Done()
}

func (t *EKS) PostDelete(ctx context.Context) deploy.Result {
	key := types.NamespacedName{
		Name: namer.UniqueIdentifier(t.APPBox.ObjectMeta),
	}
	// overwrite Name for legacy data.
	if t.APPBox.Labels[v1alpha1.IsLegacyKey] == "true" {
		name := t.APPBox.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameEKS]
		if name == "" {
			return deploy.Panic(fmt.Errorf("appbox is legacy but own-crossplane-name-eks is empty"))
		}
		key.Name = name
	}
	var c eksv1beta1.Cluster
	return deploy.ErrorOrDone(k8s.GetDeletedObject(ctx, t.Local, key, &c))
}

func (t *EKS) controlPlaneSubnets() ([]string, []xpv1.Reference, error) {
	subnetRefs := make([]xpv1.Reference, 0, len(t.VPC.Status.NodeSubnets))
	subnetIDs := make([]string, 0, len(t.VPC.Status.NodeSubnets))
	for _, sn := range t.VPC.Status.NodeSubnets {
		subnetRefs = append(subnetRefs, xpv1.Reference{
			Name: sn.Name,
		})
		subnetIDs = append(subnetIDs, sn.SubnetID)
	}
	return subnetIDs, subnetRefs, nil
}

// TODO(liubo02): Change it to avoid too many api calls.
// authorizeSecurityGroupIngress adds inbound rules to the security group of EKS control plane.
// it is safe to authorize again for migrated resource.
func (t *EKS) authorizeSecurityGroupIngress(ctx context.Context, eskClusterSgId string) error {
	cidrs := []string{"10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16"}
	ipProtocol := "-1"
	ec2Cli := t.awsCli.EC2(&aws.AssumedRole{
		RoleARN: t.RoleARN(),
	})
	for _, cidr := range cidrs {
		if _, err := ec2Cli.AuthorizeSecurityGroupIngress(ctx, &ec2.AuthorizeSecurityGroupIngressInput{
			CidrIp:  pointer.String(cidr),
			GroupId: pointer.String(eskClusterSgId),
			// all protocol
			IpProtocol: pointer.String(ipProtocol),
		}); err != nil {
			if !strings.Contains(fmt.Sprintf("%s", err), "InvalidPermission.Duplicate") {
				return fmt.Errorf("failed authorize security group ingress: %v", err)
			}
		}
	}

	return nil
}

// it is safe to call this func again for migrated resource, since it has something compare an update logic.
func (t *EKS) addSecureGroupRuleForPrivateLinkEndPoint(ctx context.Context, eksSecurityGroupID string) error {
	sg := &ec2v1beta1.SecurityGroup{}
	sgName := namer.UniqueIdentifier(t.VPC.ObjectMeta)
	if name := t.VPC.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameSecurityGroup]; name != "" {
		sgName = name
	}

	// TODO: replace with securitygroup name in anno on vpc.
	// fix after migration logic for vpc is merged.
	if err := t.Local.Get(ctx, types.NamespacedName{Name: sgName}, sg); err != nil {
		return fmt.Errorf("failed to get VPC securityGroup for %s: %v", sgName, err)
	}
	ingRule := ec2v1beta1.IPPermission{
		IPProtocol: "tcp",
		FromPort:   pointer.Int32(0),
		ToPort:     pointer.Int32(65535),
		UserIDGroupPairs: []ec2v1beta1.UserIDGroupPair{
			{
				GroupID: awssdk.String(eksSecurityGroupID),
			},
		},
	}
	for _, ing := range sg.Spec.ForProvider.Ingress {
		if equality.Semantic.DeepEqual(ing, ingRule) {
			return nil
		}
	}
	sg.Spec.ForProvider.Ingress = append(sg.Spec.ForProvider.Ingress, ingRule)
	return t.Local.Update(ctx, sg)
}

func (t *EKS) removeSecureGroupRuleDependency(ctx context.Context, eksSecurityGroupID string) error {
	sg := &ec2v1beta1.SecurityGroup{}
	sgName := namer.UniqueIdentifier(t.VPC.ObjectMeta)
	if name := t.VPC.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameSecurityGroup]; name != "" {
		sgName = name
	}
	if err := t.Local.Get(ctx, types.NamespacedName{Name: sgName}, sg); client.IgnoreNotFound(err) != nil {
		return fmt.Errorf("failed to get VPC securityGroup for %s: %v", sgName, err)
	}
	var dependencyFound bool
	updatedIngress := make([]ec2v1beta1.IPPermission, 0)
outter:
	for _, ing := range sg.Spec.ForProvider.Ingress {
		for _, pair := range ing.UserIDGroupPairs {
			if *pair.GroupID == eksSecurityGroupID {
				dependencyFound = true
				continue outter
			}
			updatedIngress = append(updatedIngress, ing)
		}
	}
	if dependencyFound {
		sg.Spec.ForProvider.Ingress = updatedIngress
		return t.Local.Update(ctx, sg)
	}
	return nil
}

// it is safe to rerun this func again for migrated resource, since it hase compare and update logic.
func (t *EKS) setSubnetTags(ctx context.Context, clusterName string) error {
	setTagF := func(subnets []v1alpha1.Subnet, elbRoleTag string) error {
		clusterTag := eksClusterNameTagKey(clusterName)
		for _, subnet := range subnets {
			sn := &ec2v1beta1.Subnet{
				ObjectMeta: metav1.ObjectMeta{
					Name:      subnet.Name,
					Namespace: t.VPC.Namespace,
				},
			}
			err := t.Local.Get(ctx, client.ObjectKeyFromObject(sn), sn)
			if err != nil {
				return err
			}

			var clusterTagSet, elbRoleTagSet bool
			for _, tag := range sn.Spec.ForProvider.Tags {
				// cluster tag already set
				if !clusterTagSet && tag.Key == clusterTag {
					clusterTagSet = true
				}
				if !elbRoleTagSet && tag.Key == elbRoleTag {
					elbRoleTagSet = true
				}
			}

			// https://kubernetes-sigs.github.io/aws-load-balancer-controller/v2.1/deploy/subnet_discovery/#public-subnets
			if !clusterTagSet {
				sn.Spec.ForProvider.Tags = append(sn.Spec.ForProvider.Tags, ec2v1beta1.Tag{
					Key:   clusterTag,
					Value: "owned",
				})
			}
			if !elbRoleTagSet {
				sn.Spec.ForProvider.Tags = append(sn.Spec.ForProvider.Tags, ec2v1beta1.Tag{
					Key:   elbRoleTag,
					Value: "1",
				})
			}
			// Due to bug https://github.com/tidbcloud/infra-provider/pull/3464, leading to duplicate tags.
			// we need to remove the duplicate tags.
			sn.Spec.ForProvider.Tags = duplicateTags(sn.Spec.ForProvider.Tags)
			if err := t.Local.Update(ctx, sn); err != nil {
				return err
			}
		}
		return nil
	}

	if err := setTagF(t.VPC.Status.PublicSubnets, subnetTagRolePublicELB); err != nil {
		return err
	}

	if err := setTagF(t.VPC.Status.PrivateSubnets, subnetTagRoleInternalELB); err != nil {
		return err
	}

	return nil
}

func eksClusterNameTagKey(clusterName string) string {
	return fmt.Sprintf("kubernetes.io/cluster/%s", clusterName)
}

func duplicateTags(tags []ec2v1beta1.Tag) []ec2v1beta1.Tag {
	tagMap := make(map[string]ec2v1beta1.Tag)
	for _, tag := range tags {
		tagMap[tag.Key] = tag
	}
	return lo.Values(tagMap)
}