package task

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

	ec2types "github.com/aws/aws-sdk-go-v2/service/ec2/types"
	ec2v1alpha1 "github.com/crossplane-contrib/provider-aws/apis/ec2/v1alpha1"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	azurenetworkv1beta1 "github.com/upbound/provider-azure/apis/network/v1beta1"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/pointer"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"

	eksv1alpha1 "github.com/tidbcloud/eks-provider/api/v1alpha1"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	compatcst "github.com/tidbcloud/infra-provider/pkg/compat/consts"
	"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"
)

type PLEndpoint struct {
	*BaseTask
}

func NewPLEndpoint(ctx *Context) *PLEndpoint {
	return &PLEndpoint{
		BaseTask: &BaseTask{Context: ctx},
	}
}

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

func (t *PLEndpoint) Create(ctx context.Context) deploy.Result {
	t.Logger.Info("create private link endpoint", "name", t.PLEndpoint.Name)

	var err error
	if t.Config.Provider == consts.ProviderTypeEKS {
		err = t.createAwsVPCEndpointByCrossPlane(ctx)
	} else if t.Config.Provider == consts.ProviderTypeAKS {
		err = t.createAzureVPCEndpointByCrossPlane(ctx)
	} else {
		return deploy.Error(fmt.Errorf("unsupported provider %s", t.Config.Provider))
	}
	return deploy.ErrorOrDone(err)
}

func (t *PLEndpoint) checkAzurePostCreate(ctx context.Context) deploy.Result {
	remoteMySQLResourceID := t.PLEndpoint.Spec.ServiceName

	if remoteMySQLResourceID == "" {
		return deploy.Done()
	}

	endpointName := t.GetEndpointName()

	endpoint := &azurenetworkv1beta1.PrivateEndpoint{
		ObjectMeta: metav1.ObjectMeta{
			Name: endpointName,
		},
	}

	if err := t.Local.Get(ctx, client.ObjectKeyFromObject(endpoint), endpoint); err != nil {
		if apierrors.IsNotFound(err) {
			// resource not ready, retry
			t.Logger.Info("private endpoint not found, will retry", "name", endpointName)
			return deploy.Error(err).WithRequeuAfter(time.Second * 10)
		}
		t.Logger.Info("get private endpoint error", "error", err)
		return deploy.Panic(err)
	}

	var ready bool
	for _, condition := range endpoint.Status.Conditions {
		if condition.Type == xpv1.TypeReady && condition.Status == corev1.ConditionTrue {
			ready = true
			break
		}
	}
	if !ready {
		state := eksv1alpha1.EndpointResourceStatePendingAcceptance
		t.PLEndpoint.Status.AtProvider.State = &state
		t.Logger.Info("private endpoint is not ready")
		return deploy.Error(errors.New("private endpoint is not ready")).WithRequeuAfter(time.Minute * 1)
	}

	// update private endpoint IP address to DMJob status
	if len(endpoint.Status.AtProvider.PrivateServiceConnection) > 0 {
		t.Logger.Info("private endpoint IP address", "ip", *endpoint.Status.AtProvider.PrivateServiceConnection[0].PrivateIPAddress)
		t.PLEndpoint.Status.AtProvider.DNSEntries = []eksv1alpha1.DNSEntry{
			{
				DNSName: endpoint.Status.AtProvider.PrivateServiceConnection[0].PrivateIPAddress,
			},
		}
		state := eksv1alpha1.EndpointResourceStateAvailable
		t.PLEndpoint.Status.AtProvider.State = &state
	}

	return deploy.Done()
}

func (t *PLEndpoint) PostCreate(ctx context.Context) deploy.Result {
	if t.Config.Provider == consts.ProviderTypeAKS {
		return t.checkAzurePostCreate(ctx)
	}

	interfaceEp, err := t.getInterfaceEp(ctx)
	if err != nil {
		return deploy.Error(err)
	}
	return t.watchPLEndpoint(&interfaceEp, time.Second*30)
}

func (t *PLEndpoint) getInterfaceEp(ctx context.Context) (ec2v1alpha1.VPCEndpoint, error) {
	key := types.NamespacedName{
		Name: t.PLEndpoint.Name,
	}
	var interfaceEp ec2v1alpha1.VPCEndpoint
	err := t.Local.Get(ctx, key, &interfaceEp)
	return interfaceEp, err
}

func (t *PLEndpoint) GetEndpointName() string {
	return fmt.Sprintf("pl-%s", t.PLEndpoint.Name)
}

func (t *PLEndpoint) GetEndpointNameUserSide() string {
	return fmt.Sprintf("pl-%s-tidbcloud", t.PLEndpoint.Name)
}

func (t *PLEndpoint) createAzureVPCEndpointByCrossPlane(ctx context.Context) error {
	t.Logger.Info("create azure private link endpoint", "name", t.PLEndpoint.Name)
	remoteMySQLResourceID := t.PLEndpoint.Spec.ServiceName

	endpointName := t.GetEndpointName()
	endpointNameUserSide := t.GetEndpointNameUserSide()

	// Add nil checks to prevent nil pointer dereference
	if t.Kubernetes == nil {
		return fmt.Errorf("invalid kubernetes configuration: Kubernetes is nil")
	}
	if t.Kubernetes.Cloud.Azure == nil {
		return fmt.Errorf("invalid kubernetes configuration: Azure configuration is nil")
	}
	if len(t.Kubernetes.Cloud.Azure.NodeSubnets) == 0 {
		return fmt.Errorf("invalid kubernetes configuration: missing required Azure NodeSubnets")
	}
	subnetID := t.Kubernetes.Cloud.Azure.NodeSubnets[0].Name
	providerConfigName := t.GetAKSProviderConfigName()

	endpoint := &azurenetworkv1beta1.PrivateEndpoint{
		ObjectMeta: metav1.ObjectMeta{
			Name: endpointName,
		},
		Spec: azurenetworkv1beta1.PrivateEndpointSpec{
			ForProvider: azurenetworkv1beta1.PrivateEndpointParameters{
				Location:          ptr.To(t.PLEndpoint.Spec.Region),
				ResourceGroupName: ptr.To(t.Kubernetes.Cloud.Azure.ResourceGroup),
				SubnetID:          ptr.To(subnetID),
				PrivateServiceConnection: []azurenetworkv1beta1.PrivateServiceConnectionParameters{
					{
						IsManualConnection:          ptr.To(true),
						Name:                        ptr.To(endpointNameUserSide),
						PrivateConnectionResourceID: ptr.To(remoteMySQLResourceID),
						RequestMessage:              ptr.To("approve the connection to use DM on tidbcloud"),
						SubresourceNames:            []*string{ptr.To("mysqlServer")},
					},
				},
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: providerConfigName,
				},
			},
		},
	}

	if err := t.Local.Create(ctx, endpoint); err != nil {
		if apierrors.IsAlreadyExists(err) {
			return nil
		}

		t.Logger.Error(err, "create azure private endpoint error")
		return err
	}

	return nil
}

func (t *PLEndpoint) createAwsVPCEndpointByCrossPlane(ctx context.Context) error {
	t.Logger.Info("create aws private link endpoint", "name", t.PLEndpoint.Name)
	interfaceEp := &ec2v1alpha1.VPCEndpoint{
		ObjectMeta: metav1.ObjectMeta{
			Name: t.PLEndpoint.Name,
		},
	}
	_, err := controllerutil.CreateOrUpdate(ctx, t.Local, interfaceEp, func() error {
		t.fillSpecToInterfaceEp(interfaceEp)
		return nil
	})
	return err
}

func (t *PLEndpoint) fillSpecToInterfaceEp(interfaceEp *ec2v1alpha1.VPCEndpoint) {
	uniqueIdentifier := namer.UniqueIdentifier(t.Network.ObjectMeta)

	interfaceEp.Spec.ResourceSpec = xpv1.ResourceSpec{
		ProviderConfigReference: &xpv1.Reference{Name: uniqueIdentifier},
	}

	t.fillForProviderToInterfaceEp(interfaceEp, uniqueIdentifier)
}

func (t *PLEndpoint) fillForProviderToInterfaceEp(interfaceEp *ec2v1alpha1.VPCEndpoint, uniqueIdentifier string) {
	interfaceEp.Spec.ForProvider.Region = t.Network.Spec.Region
	interfaceEp.Spec.ForProvider.ServiceName = pointer.String(t.PLEndpoint.Spec.ServiceName)
	interfaceEp.Spec.ForProvider.VPCEndpointType = pointer.String(string(ec2types.VpcEndpointTypeInterface))
	interfaceEp.Spec.ForProvider.PrivateDNSEnabled = pointer.Bool(false)
	_, ok := t.Network.Annotations[compatcst.CompatFlagAnnotation]
	if !ok {
		interfaceEp.Spec.ForProvider.CustomVPCEndpointParameters.VPCIDRef = &xpv1.Reference{Name: uniqueIdentifier}
		var privateSubnetRefs []xpv1.Reference
		for _, subnet := range t.Subnets {
			privateSubnetRefs = append(privateSubnetRefs, xpv1.Reference{Name: subnet.Name})
		}
		interfaceEp.Spec.ForProvider.CustomVPCEndpointParameters.SubnetIDRefs = privateSubnetRefs
		interfaceEp.Spec.ForProvider.CustomVPCEndpointParameters.SecurityGroupIDRefs = []xpv1.Reference{
			{Name: uniqueIdentifier},
		}
	} else {
		//In the old path, we used the same crossplane to create the VPC,Subnet and PrivateLink.
		//we set the SubnetRef/VPCRef field to let CrossPlane fetch metadata from the manages resources.
		//but in the new path, VPC,Subnet and PrivateLink are no longer managed through the same CrossPlane,
		//so we need to set the VPCID,SubnetID field to let CrossPlane query the  AWS API for metadata

		interfaceEp.Spec.ForProvider.CustomVPCEndpointParameters.VPCID = pointer.String(t.NetworkSpec.VpcId)
		var privateSubnetIDs []*string
		for _, subnet := range t.Subnets {
			privateSubnetIDs = append(privateSubnetIDs, pointer.String(subnet.Status.AtProvider.SubnetID))
		}
		interfaceEp.Spec.ForProvider.CustomVPCEndpointParameters.SubnetIDs = privateSubnetIDs
		interfaceEp.Spec.ForProvider.CustomVPCEndpointParameters.SecurityGroupIDs = []*string{pointer.String(t.NetworkSpec.SecurityGroupId)}
	}

	t.fillTagToInterfaceEp(interfaceEp)
}

func (t *PLEndpoint) fillTagToInterfaceEp(interfaceEp *ec2v1alpha1.VPCEndpoint) {
	if interfaceEp.Spec.ForProvider.TagSpecifications == nil {
		interfaceEp.Spec.ForProvider.TagSpecifications = []*ec2v1alpha1.TagSpecification{}
	}

	index := t.queryVpcEndpointTagSpecificationIndex(interfaceEp)
	if index < 0 {
		t.buildNewVpcEndpointTagSpecification(interfaceEp)
		return
	}
	t.fillExistedVpcEndpointTagSpecification(interfaceEp, index)
}

func (t *PLEndpoint) fillExistedVpcEndpointTagSpecification(interfaceEp *ec2v1alpha1.VPCEndpoint, index int) {
	specification := interfaceEp.Spec.ForProvider.TagSpecifications[index]
	tags := t.buildVpcEndpointTag()
	for _, tag := range specification.Tags {
		if value, ok := tags[*tag.Key]; ok {
			tag.Value = &value
			delete(tags, *tag.Key)
		}
	}
	if len(tags) <= 0 {
		return
	}
	interfaceEpTags := util.GetEC2V1AlphaTags(tags)
	specification.Tags = append(specification.Tags, interfaceEpTags...)
}

func (t *PLEndpoint) buildNewVpcEndpointTagSpecification(interfaceEp *ec2v1alpha1.VPCEndpoint) {
	tags := t.buildVpcEndpointTag()
	interfaceEpTags := util.GetEC2V1AlphaTags(tags)

	interfaceEp.Spec.ForProvider.TagSpecifications = append(interfaceEp.Spec.ForProvider.TagSpecifications,
		&ec2v1alpha1.TagSpecification{
			ResourceType: pointer.String(string(ec2types.ResourceTypeVpcEndpoint)),
			Tags:         interfaceEpTags,
		})
}

func (t *PLEndpoint) buildVpcEndpointTag() map[string]string {
	labels := map[string]string{}
	labels = util.ConcatenateTags(labels, cloudv1.TagsFromLabels(t.Network.Labels))
	labels = util.ConcatenateTags(labels, cloudv1.TagsFromLabels(t.PLEndpoint.Labels))
	labels = util.ConcatenateTags(labels, map[string]string{
		"Name": t.PLEndpoint.Name,
	})
	return labels
}

func (t *PLEndpoint) queryVpcEndpointTagSpecificationIndex(interfaceEp *ec2v1alpha1.VPCEndpoint) int {
	for i, tag := range interfaceEp.Spec.ForProvider.TagSpecifications {
		if *tag.ResourceType == string(ec2types.ResourceTypeVpcEndpoint) {
			return i
		}
	}
	return -1
}

func (t *PLEndpoint) watchPLEndpoint(interfaceEp *ec2v1alpha1.VPCEndpoint, interval time.Duration) deploy.Result {
	t.observe(interfaceEp)

	if interfaceEp.Status.AtProvider.State == nil {
		return deploy.Error(fmt.Errorf("wait private link endpoint ready")).WithRequeuAfter(interval)
	}
	switch phase := t.PLEndpoint.Status.Phase; phase {
	case eksv1alpha1.PrivateLinkEndpointPhaseCreating:
		if t.PLEndpoint.Status.AtProvider.State == nil ||
			*t.PLEndpoint.Status.AtProvider.State == eksv1alpha1.EndpointResourceStatePendingAcceptance ||
			*t.PLEndpoint.Status.AtProvider.State == eksv1alpha1.EndpointResourceStatePending ||
			*t.PLEndpoint.Status.AtProvider.State == eksv1alpha1.EndpointResourceStateDeleting {
			t.Logger.Info("wait private link endpoint ready", "name", interfaceEp.Name, "state", interfaceEp.Status.AtProvider.State)
			return deploy.Error(fmt.Errorf("wait private link endpoint ready")).WithRequeuAfter(interval)
		} else {
			return deploy.Error(k8s.RequeueOnlyError).WithRequeuAfter(interval)
		}
	default:
		// watch private link status, sync status to CR status
		return deploy.Error(k8s.RequeueOnlyError).WithRequeuAfter(interval)
	}
}

func (t *PLEndpoint) observe(interfaceEp *ec2v1alpha1.VPCEndpoint) {
	if interfaceEp.Status.AtProvider.State == nil {
		return
	}

	resourceState := convertToPrivateLinkEndpointPhase(*interfaceEp.Status.AtProvider.State)
	t.PLEndpoint.Status.AtProvider = eksv1alpha1.EndpointResource{
		State: &resourceState,
	}

	if interfaceEp.Annotations != nil {
		if vpceId, ok := interfaceEp.Annotations["crossplane.io/external-name"]; ok {
			t.PLEndpoint.Status.AtProvider.EndpointID = &vpceId
		}
	}

	t.fillDNSEntriesToPLEndpoint(interfaceEp, resourceState)
}

func (t *PLEndpoint) fillDNSEntriesToPLEndpoint(interfaceEp *ec2v1alpha1.VPCEndpoint, resourceState eksv1alpha1.EndpointResourceState) {
	if resourceState != eksv1alpha1.EndpointResourceStateAvailable {
		t.PLEndpoint.Status.AtProvider.DNSEntries = nil
		return
	}

	var dnsEntries []eksv1alpha1.DNSEntry
	for _, entry := range interfaceEp.Status.AtProvider.DNSEntries {
		matchZone := ""
		for _, zone := range t.PLEndpoint.Spec.Zones {
			if strings.Contains(*entry.DNSName, zone) {
				matchZone = zone
				break
			}
		}
		if matchZone != "" {
			dnsEntries = append(dnsEntries, eksv1alpha1.DNSEntry{
				Zone:    &matchZone,
				DNSName: entry.DNSName,
			})
		} else {
			dnsEntries = append(dnsEntries, eksv1alpha1.DNSEntry{
				DNSName: entry.DNSName,
			})
		}
	}
	sort.SliceStable(dnsEntries, func(i, j int) bool {
		return *dnsEntries[i].DNSName < *dnsEntries[j].DNSName
	})
	t.PLEndpoint.Status.AtProvider.DNSEntries = dnsEntries
}

func convertToPrivateLinkEndpointPhase(str string) eksv1alpha1.EndpointResourceState {
	switch state := strings.ToLower(str); state {
	case strings.ToLower(string(ec2v1alpha1.State_PendingAcceptance)):
		return eksv1alpha1.EndpointResourceStatePendingAcceptance
	case strings.ToLower(string(ec2v1alpha1.State_Pending)):
		return eksv1alpha1.EndpointResourceStatePending
	case strings.ToLower(string(ec2v1alpha1.State_Available)):
		return eksv1alpha1.EndpointResourceStateAvailable
	case strings.ToLower(string(ec2v1alpha1.State_Deleting)):
		return eksv1alpha1.EndpointResourceStateDeleting
	case strings.ToLower(string(ec2v1alpha1.State_Deleted)):
		return eksv1alpha1.EndpointResourceStateDeleted
	case strings.ToLower(string(ec2v1alpha1.State_Rejected)):
		return eksv1alpha1.EndpointResourceStateRejected
	case strings.ToLower(string(ec2v1alpha1.State_Failed)):
		return eksv1alpha1.EndpointResourceStateFailed
	case strings.ToLower(string(ec2v1alpha1.State_Expired)):
		return eksv1alpha1.EndpointResourceStateExpired
	default:
		return eksv1alpha1.EndpointResourceStatePending
	}
}

func (t *PLEndpoint) Delete(ctx context.Context) deploy.Result {
	t.Logger.Info("delete private link endpoint", "name", t.PLEndpoint.Name)
	if t.Config.Provider == cloudv1.ProviderAzure || t.Config.Provider == consts.ProviderTypeAKS {
		return t.deleteAzureVpcEndpoint(ctx)
	}

	if err := deleteVpcEndpoint(ctx, t.Local, t.PLEndpoint.Name); err != nil {
		return deploy.Error(err)
	}
	return deploy.Done()
}

func (t *PLEndpoint) deleteAzureVpcEndpoint(ctx context.Context) deploy.Result {
	endpointName := t.GetEndpointName()
	endpoint := &azurenetworkv1beta1.PrivateEndpoint{
		ObjectMeta: metav1.ObjectMeta{
			Name: endpointName,
		},
	}

	if err := t.Local.Delete(ctx, endpoint); err != nil && !apierrors.IsNotFound(err) {
		return deploy.Error(err)
	}
	return deploy.Done()
}

func deleteVpcEndpoint(ctx context.Context, cli client.Client, epName string) error {
	ep := &ec2v1alpha1.VPCEndpoint{
		ObjectMeta: metav1.ObjectMeta{
			Name: epName,
		},
	}
	return k8s.DeleteObject(ctx, cli, ep)
}

func (t *PLEndpoint) PostDelete(ctx context.Context) deploy.Result {
	if t.Config.Provider == cloudv1.ProviderAzure || t.Config.Provider == consts.ProviderTypeAKS {
		return t.checkAzurePostDelete(ctx)
	}

	return deploy.ErrorOrDone(k8s.GetDeletedObject(ctx, t.Local, types.NamespacedName{Name: t.PLEndpoint.Name},
		&ec2v1alpha1.VPCEndpoint{
			ObjectMeta: metav1.ObjectMeta{
				Name: t.PLEndpoint.Name,
			},
		}))
}

func (t *PLEndpoint) checkAzurePostDelete(ctx context.Context) deploy.Result {

	endpointName := t.GetEndpointName()

	return deploy.ErrorOrDone(k8s.GetDeletedObject(ctx, t.Local, types.NamespacedName{Name: endpointName},
		&azurenetworkv1beta1.PrivateEndpoint{
			ObjectMeta: metav1.ObjectMeta{
				Name: endpointName,
			},
		}))
}