package eks

import (
	"context"
	"fmt"

	awssdk "github.com/aws/aws-sdk-go-v2/aws"
	ec2types "github.com/aws/aws-sdk-go-v2/service/ec2/types"
	"github.com/crossplane-contrib/provider-aws/apis/ec2/v1alpha1"
	"github.com/crossplane-contrib/provider-aws/apis/ec2/v1beta1"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	"github.com/pingcap/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"

	eksv1alpha1 "github.com/tidbcloud/eks-provider/api/v1alpha1"
	"github.com/tidbcloud/eks-provider/controllers/network/task"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"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"
	o11y "github.com/tidbcloud/infra-provider/pkg/o11y/metric_agent"
	"github.com/tidbcloud/infra-provider/pkg/util"
)

// S3GatewayEndpoint deploys the S3 gateway endpoint for all components to access S3.
type S3GatewayEndpoint struct {
	*task.BaseTask
}

var _ deploy.Task = &S3GatewayEndpoint{}

func NewS3GatewayEndpoint(ctx *task.Context) *S3GatewayEndpoint {
	return &S3GatewayEndpoint{
		BaseTask: &task.BaseTask{Context: ctx},
	}
}

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

func (t *S3GatewayEndpoint) Create(ctx context.Context) deploy.Result {
	if eksv1alpha1.IsMigrate(t.Network) {
		return t.createOrUpdateForMigrate(ctx)
	}

	s3ServiceName := fmt.Sprintf("com.amazonaws.%s.s3", t.Network.Spec.Region)
	vpcName := namer.UniqueIdentifier(t.Network.ObjectMeta)
	var vpc v1beta1.VPC
	if err := t.Local.Get(ctx, types.NamespacedName{
		Name: vpcName,
	}, &vpc); err != nil {
		return deploy.Error(errors.Annotate(err, "VPC not found"))
	}

	endpointName := gatewayEndpointName(vpcName)

	var routeTableIDRefs []xpv1.Reference

	refs, err := t.getPrivateRouteTables(ctx, vpcName)
	if err != nil {
		return deploy.Error(errors.Annotate(err, "failed to get private route tables from VPC"))
	}
	routeTableIDRefs = refs

	tags := cloudv1.TagsFromLabels(t.Network.Labels)
	tags = util.AddResourceName(tags, endpointName)

	endpoint := &v1alpha1.VPCEndpoint{
		ObjectMeta: metav1.ObjectMeta{
			Name: endpointName,
		},
		Spec: v1alpha1.VPCEndpointSpec{
			ForProvider: v1alpha1.VPCEndpointParameters{
				Region:          t.Network.Spec.Region,
				ServiceName:     &s3ServiceName,
				VPCEndpointType: awssdk.String(string(ec2types.VpcEndpointTypeGateway)),
				TagSpecifications: []*v1alpha1.TagSpecification{
					{
						ResourceType: awssdk.String(string(ec2types.ResourceTypeVpcEndpoint)),
						Tags:         util.GetEC2V1AlphaTags(tags),
					},
				},
				CustomVPCEndpointParameters: v1alpha1.CustomVPCEndpointParameters{
					VPCIDRef:         &xpv1.Reference{Name: vpcName},
					RouteTableIDRefs: routeTableIDRefs,
				},
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: vpc.GetProviderConfigReference(),
			},
		},
	}
	if err := k8s.Apply(ctx, t.Local, endpoint); err != nil {
		return deploy.Error(errors.Annotate(err, "failed to create vpc s3 gateway endpoint"))
	}

	// update gateway endpoint id to network's annotation
	util.SetAnnotation(t.Network, consts.S3GatewayEndpointNameAnnotationKey, endpointName)
	return deploy.Done()
}

func (t *S3GatewayEndpoint) Delete(ctx context.Context) deploy.Result {
	endpointName := gatewayEndpointName(namer.UniqueIdentifier(t.Network.ObjectMeta))
	endpoint := &v1alpha1.VPCEndpoint{
		ObjectMeta: metav1.ObjectMeta{
			Name: endpointName,
		},
	}
	if err := k8s.DeleteObject(ctx, t.Local, endpoint); err != nil {
		return deploy.Error(errors.Annotate(err, "failed to delete vpc s3 gateway endpoint"))
	}

	// remove gateway endpoint id from network's annotation
	if t.Network.Annotations != nil {
		delete(t.Network.Annotations, consts.S3GatewayEndpointNameAnnotationKey)
	}
	return deploy.Done()
}

// FIXME(liubo02): get route table by selector
func (t *S3GatewayEndpoint) getPrivateRouteTables(ctx context.Context, vpcName string) ([]xpv1.Reference, error) {
	zones, err := t.Config.FindRegionZones(t.Network.Spec.Region)
	if err != nil {
		return nil, err
	}

	var refs []xpv1.Reference
	for _, zone := range zones {
		var route v1alpha1.Route
		if err := t.Local.Get(ctx, types.NamespacedName{
			Name: namer.PrivateSubnetName(vpcName, zone),
		}, &route); err == nil {
			refs = append(refs, xpv1.Reference{Name: route.Name})
		}
	}

	return refs, nil
}

func gatewayEndpointName(vpcName string) string {
	return vpcName + "-gateway-endpoint-s3"
}

// If it's a migration network, there should be a s3 gateway endpoint created in gardener arch.
func (t *S3GatewayEndpoint) createOrUpdateForMigrate(ctx context.Context) deploy.Result {
	s3ServiceName := fmt.Sprintf("com.amazonaws.%s.s3", t.Network.Spec.Region)

	if util.GetAnnotation(t.Network, consts.S3GatewayEndpointNameAnnotationKey) == "" {
		// If there is already an endpoint has the same endpoint service name and vpc id, we can re-use it directly.
		name, err := o11y.FindReusableVPCEndpointName(ctx, t.Local, s3ServiceName, t.EKSNetworkSpec.VpcId, string(ec2types.VpcEndpointTypeGateway))
		if err != nil {
			t.Logger.Error(err, "failed to find reusable vpc endpoint")
		}
		if name != "" {
			// Since there is already a s3 gateway endpoint named with `shoot` prefix, we can re-use it directly.
			t.Logger.Info("found an existed endpoint, do not need to create new one", "name", name)
			util.SetAnnotation(t.Network, consts.S3GatewayEndpointNameAnnotationKey, name)
			return deploy.Done()
		}
	}

	return deploy.Done()
}