package aws

import (
	"context"
	"fmt"
	"strings"

	awssdk "github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/ec2"
	ec2types "github.com/aws/aws-sdk-go-v2/service/ec2/types"
	ec2v1alpha1 "github.com/crossplane-contrib/provider-aws/apis/ec2/v1alpha1"
	"github.com/pingcap/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"

	agentv2 "github.com/tidbcloud/observability/agent-control-server/api/v2"
	"github.com/tidbcloud/observability/agent-control-server/pkg/agentbuild"
	"github.com/tidbcloud/observability/agent-control-server/pkg/cloud/aws"
	"github.com/tidbcloud/observability/agent-control-server/pkg/cloud/aws/awscrossplane"
	"github.com/tidbcloud/observability/agent-control-server/pkg/consts"
	"github.com/tidbcloud/observability/agent-control-server/pkg/deploy"
	"github.com/tidbcloud/observability/agent-control-server/pkg/reconciler"
	"github.com/tidbcloud/observability/agent-control-server/pkg/util"
)

const (
	vpcEpResourceType = "vpc-endpoint"
)

// PrivateLinkTask creates VPC interface endpoint to send o11y data to o11y cluster.
type PrivateLinkTask struct {
	*reconciler.BaseTask[*agentv2.O11yK8sAgent]
}

func NewPrivateLinkTask(ctx *reconciler.Context[*agentv2.O11yK8sAgent]) *PrivateLinkTask {
	return &PrivateLinkTask{
		BaseTask: &reconciler.BaseTask[*agentv2.O11yK8sAgent]{Context: ctx},
	}
}

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

func (t *PrivateLinkTask) Create(ctx context.Context) deploy.Result {
	needUpdate := util.ShouldUpdateO11yVersion(t.Agent.GetAnnotations(), t.Config.O11yConfig.Version)
	endpoint := util.GetAnnotation(t.Agent, consts.AnnotationKeyO11ySvcInterfaceEndpointName)
	if endpoint != "" && !needUpdate {
		return deploy.Done()
	}

	endpointServiceName := util.GetAnnotation(t.Agent, consts.AnnotationKeyO11yEndpointServiceName)
	if endpointServiceName == "" {
		return deploy.Panic(errors.New("endpoint service name not found wait for o11y cluster ready"))
	}
	endpointName, err := awscrossplane.FindReusableVPCEndpointName(ctx, t.Local, string(ec2types.VpcEndpointTypeInterface), endpointServiceName, t.Agent.Spec.EKS.VpcID)
	if err != nil {
		return deploy.Panic(errors.Annotate(err, "failed to find reusable vpc endpoint"))
	}
	if endpointName != "" {
		t.L.Info("found an existed endpoint, do not need to create new one", "name", endpointName)
		// In the case it reuse the one created by aws-provider, we append the sg to the endpoint make sure it can be accessed since eks-provider use the different node sg.
		// We should not update the endpoint's SecurityGroupIDs to contains the eks-provider's node sg only since it will failed the cluster still run at aws-provider, seems it will not now and good lucky for will not in the future(note there are many other "reuse **" code.
		err = ensureSecurityGroup(ctx, t.Local, endpointName, t.Agent.Spec.EKS.SecurityGroupID)
		if err != nil {
			return deploy.Panic(errors.Annotate(err, "failed to ensure security group"))
		}
		util.SetAnnotation(t.Agent, consts.AnnotationKeyO11ySvcInterfaceEndpointName, endpointName)
		return deploy.Done()
	}

	// TODO: find reusable endpoint from AWS API directly.

	dnsName := util.GetAnnotation(t.Agent, consts.AnnotationKeyO11yServiceDNSName)
	if dnsName == "" {
		return deploy.Panic(errors.New("service dns name not found"))
	}
	endpointName = util.InterfaceEndpointName(t.Agent.ObjectMeta, dnsName, consts.O11yEndpointNamePrefixDataplane)

	err = t.createInterfaceEndpoint(ctx, endpointServiceName, endpointName)
	if err != nil {
		return deploy.Panic(errors.Annotate(err, "failed to create interface endpoint"))
	}
	t.L.Info("create interface endpoint success", "name", endpointName)

	// Update interface endpoint name to cluster's annotation.
	util.SetAnnotation(t.Agent, consts.AnnotationKeyO11ySvcInterfaceEndpointName, endpointName)
	return deploy.Done()
}

func (t *PrivateLinkTask) Delete(ctx context.Context) deploy.Result {
	// Do not delete interface endpoint since it will be reused by other cluster
	endpointName := util.GetAnnotation(t.Agent, consts.AnnotationKeyO11ySvcInterfaceEndpointName)
	if endpointName == "" {
		return deploy.Done()
	}
	if err := agentbuild.DeleteAwsVPCEndpoint(ctx, t.Local, endpointName); err != nil {
		return deploy.Panic(err)
	}
	return deploy.Done()
}

func (t *PrivateLinkTask) PostDelete(ctx context.Context) deploy.Result {
	endpointName := util.GetAnnotation(t.Agent, consts.AnnotationKeyO11ySvcInterfaceEndpointName)
	if endpointName == "" {
		return deploy.Done()
	}
	if err := agentbuild.CheckAwsVPCEndpointDeleted(ctx, t.Local, endpointName); err != nil {
		return deploy.Panic(err)
	}
	return deploy.Done()
}

func (t *PrivateLinkTask) createInterfaceEndpoint(ctx context.Context, endpointServiceName, interfaceEndpointName string) error {
	t.L.Info("creating interface endpoint",
		"endpointServiceName", endpointServiceName,
		"interfaceEndpointName", interfaceEndpointName,
		"region", t.Config.O11yConfig.Region)
	if t.Config.O11yConfig.Region == "" {
		return errors.New("AWS region is not configured")
	}

	// Get the availability zone of the endpoint service
	pickedSubnets, err := t.pickSubnets(ctx, endpointServiceName, t.Agent.Spec.EKS.NodeSubnetIDs)
	if err != nil {
		return errors.Annotate(err, "failed to pick subnets for interface endpoint")
	}
	// we choose one subnet at least,to save cost
	pickedSubnets = pickedSubnets[:1]
	// Create an interface endpoint with the picked subnet
	endpoint := &ec2v1alpha1.VPCEndpoint{
		ObjectMeta: metav1.ObjectMeta{
			Name: interfaceEndpointName,
		},
	}
	_, err = controllerutil.CreateOrUpdate(ctx, t.Local, endpoint, func() error {
		config := t.Agent.Spec.EKS
		endpoint.Spec.ProviderConfigReference = config.ProviderConfigReference
		endpoint.Spec.ForProvider.Region = t.Config.O11yConfig.Region
		endpoint.Spec.ForProvider.ServiceName = awssdk.String(endpointServiceName)
		endpoint.Spec.ForProvider.VPCID = &config.VpcID
		endpoint.Spec.ForProvider.VPCEndpointType = awssdk.String(string(ec2types.VpcEndpointTypeInterface))
		// Check if private DNS should be disabled (default: disabled unless explicitly set to false)
		disablePrivateDNS := util.GetAnnotation(t.Agent, consts.AnnotationKeyDisablePrivateEndpointPrivateDNS) != "false"
		endpoint.Spec.ForProvider.PrivateDNSEnabled = awssdk.Bool(!disablePrivateDNS)
		endpoint.Spec.ForProvider.SecurityGroupIDs = ensureStringItem(endpoint.Spec.ForProvider.SecurityGroupIDs, config.SecurityGroupID)
		endpoint.Spec.ForProvider.SubnetIDs = awssdk.StringSlice(pickedSubnets)

		// Ensure crossplane vendor tags are set.
		if len(endpoint.Spec.ForProvider.TagSpecifications) != 2 || awssdk.ToString(endpoint.Spec.ForProvider.TagSpecifications[0].ResourceType) != vpcEpResourceType {
			endpoint.Spec.ForProvider.TagSpecifications = []*ec2v1alpha1.TagSpecification{
				{
					ResourceType: awssdk.String(vpcEpResourceType),
					Tags: []*ec2v1alpha1.Tag{
						{
							Key:   awssdk.String(consts.CrossplaneKeyVendor),
							Value: awssdk.String(consts.CrossplaneValueVendor),
						},
						{
							Key:   awssdk.String(consts.CrossplaneKeyName),
							Value: awssdk.String(interfaceEndpointName),
						},
					},
				},
			}
		} else {
			for _, tag := range endpoint.Spec.ForProvider.TagSpecifications[0].Tags {
				if tag == nil {
					continue
				}
				switch awssdk.ToString(tag.Key) {
				case consts.CrossplaneKeyVendor:
					tag.Value = awssdk.String(consts.CrossplaneValueVendor)
				case consts.CrossplaneKeyName:
					tag.Value = awssdk.String(interfaceEndpointName)
				}
			}
		}
		return nil
	})
	if err != nil {
		return errors.Annotate(err, "failed to create/update interface endpoint")
	}
	return nil
}

func (t *PrivateLinkTask) pickSubnets(ctx context.Context, endpointServiceName string, subnetIDs []string) ([]string, error) {
	if len(subnetIDs) == 0 {
		return nil, errors.New("no subnet ids provided")
	}

	ec2cli := t.AWS.EC2(&aws.AssumedRole{
		RoleARN: t.Agent.Spec.EKS.AssumeRoleArn,
	})
	serviceOutput, err := ec2cli.DescribeVpcEndpointServices(ctx, &ec2.DescribeVpcEndpointServicesInput{
		ServiceNames: []string{endpointServiceName},
	})
	if err != nil {
		return nil, errors.Annotate(err, "failed to describe vpc endpoint services")
	}
	if len(serviceOutput.ServiceDetails) == 0 {
		return nil, errors.New("no service details found for the given endpoint service name")
	}

	allowedZones := make(map[string]struct{})
	supportedZones := make([]string, 0)
	for _, detail := range serviceOutput.ServiceDetails {
		for _, zone := range detail.AvailabilityZones {
			if _, exists := allowedZones[zone]; exists {
				continue
			}
			allowedZones[zone] = struct{}{}
			supportedZones = append(supportedZones, zone)
		}
	}
	if len(allowedZones) == 0 {
		return nil, errors.New("endpoint service does not expose availability zones")
	}

	subnetOutput, err := ec2cli.DescribeSubnets(ctx, &ec2.DescribeSubnetsInput{SubnetIds: subnetIDs})
	if err != nil {
		return nil, errors.Annotate(err, "failed to describe subnets")
	}

	selected := make([]string, 0, len(subnetOutput.Subnets))
	providedSubnets := make([]string, 0, len(subnetOutput.Subnets))
	for _, subnet := range subnetOutput.Subnets {
		var (
			subnetID = ""
			zone     = ""
		)
		if subnet.SubnetId != nil {
			subnetID = *subnet.SubnetId
		}
		if subnet.AvailabilityZone != nil {
			zone = *subnet.AvailabilityZone
		}
		providedSubnets = append(providedSubnets, fmt.Sprintf("%s(%s)", subnetID, zone))
		if subnet.SubnetId == nil || subnet.AvailabilityZone == nil {
			continue
		}
		if _, ok := allowedZones[zone]; ok {
			selected = append(selected, subnetID)
		}
	}

	if len(selected) == 0 {
		return nil, errors.Errorf(
			"no subnets match endpoint service availability zones. endpoint supports zones: [%s]; provided subnets: [%s]",
			strings.Join(supportedZones, ","),
			strings.Join(providedSubnets, ","),
		)
	}
	return selected, nil
}

func ensureSecurityGroup(ctx context.Context, cli client.Client, epName string, sgID string) error {
	ep := &ec2v1alpha1.VPCEndpoint{}
	ep.Name = epName
	_, err := controllerutil.CreateOrUpdate(ctx, cli, ep, func() error {
		if ep.CreationTimestamp.Time.IsZero() {
			return fmt.Errorf("endpoint should already created")
		}
		ep.Spec.ForProvider.SecurityGroupIDs = ensureStringItem(ep.Spec.ForProvider.SecurityGroupIDs, sgID)
		return nil
	})
	if err != nil {
		return fmt.Errorf("failed to create or update: %s", err)
	}
	return nil
}

func ensureStringItem(arr []*string, item string) []*string {
	for _, v := range arr {
		if *v == item {
			return arr
		}
	}
	return append(arr, &item)
}