package eks

import (
	"context"
	"fmt"
	"strings"

	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/crossplane/crossplane-runtime/pkg/meta"
	"github.com/crossplane/crossplane-runtime/pkg/reconciler/managed"
	"github.com/pingcap/errors"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"

	"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"
	rsv1alpha1 "github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/internet/task"
)

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

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 t.IsHibernated() {
		return deploy.Done()
	}
	s3ServiceName := fmt.Sprintf("com.amazonaws.%s.s3", t.Internet.Spec.Region)
	endpointName := t.gatewayEndpointName()
	vpcRes := &v1beta1.VPC{}
	if err := t.Local.Get(ctx, types.NamespacedName{Name: t.VPCName()}, vpcRes); err != nil {
		return deploy.Error(errors.Annotate(err, "failed to get vpc resource"))
	}

	routeTableIDRefs := t.getPrivateRouteTables()
	if len(routeTableIDRefs) == 0 {
		return deploy.Error(errors.New("no private route tables found"))
	}

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

	endpoint := &v1alpha1.VPCEndpoint{
		ObjectMeta: metav1.ObjectMeta{
			Name: endpointName,
			Labels: map[string]string{
				rsv1alpha1.InternetNameKey: t.InternetName(),
			},
		},
		Spec: v1alpha1.VPCEndpointSpec{
			ForProvider: v1alpha1.VPCEndpointParameters{
				Region:          t.Internet.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: vpcRes.Name},
					RouteTableIDRefs: routeTableIDRefs,
				},
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{Name: t.ProviderConfigName()},
			},
		},
	}
	if err := k8s.Apply(ctx, t.Local, endpoint); err != nil {
		return deploy.Error(errors.Annotate(err, "failed to create vpc s3 gateway endpoint"))
	}

	return deploy.Done()
}

func (t *S3GatewayEndpoint) PostCreate(ctx context.Context) deploy.Result {
	if t.IsHibernated() {
		return deploy.Done()
	}
	endpointName := t.gatewayEndpointName()
	endpoint := &v1alpha1.VPCEndpoint{}
	if err := t.Local.Get(ctx, types.NamespacedName{Name: endpointName}, endpoint); err != nil {
		return deploy.Error(errors.Annotate(err, "failed to get vpc s3 gateway endpoint"))
	}

	if endpoint.Status.GetCondition(xpv1.TypeReady).Status != corev1.ConditionTrue {
		return deploy.Error(errors.New("vpc s3 gateway endpoint is not available"))
	}
	t.Internet.Status.S3GatewayEndpoint = rsv1alpha1.Endpoint{
		Name:       endpointName,
		EndpointID: meta.GetExternalName(endpoint),
	}
	if err := t.InternetRW.UpdateStatus(ctx, t.Internet); err != nil {
		return deploy.Error(errors.Annotate(err, "failed to update internet status"))
	}
	return deploy.Done()
}

func (t *S3GatewayEndpoint) Delete(ctx context.Context) deploy.Result {
	endpointName := t.gatewayEndpointName()
	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"))
	}
	return deploy.Done()
}

func (t *S3GatewayEndpoint) PostDelete(ctx context.Context) deploy.Result {
	endpointName := t.gatewayEndpointName()
	endpoint := &v1alpha1.VPCEndpoint{}
	if err := t.Local.Get(ctx, types.NamespacedName{Name: endpointName}, endpoint); err != nil {
		// If the endpoint is not found, it means the endpoint has been deleted successfully.
		if client.IgnoreNotFound(err) == nil {
			return deploy.Done()
		}
		return deploy.Error(errors.Annotate(err, "failed to get vpc s3 gateway endpoint"))
	}

	// this is a workaround for the issue that the vpcendpoint cr is not deleted successfully.
	// The issue is caused by the vpc endpoint resource is deleted successfully, but crossplane cr is not deleted and reconscile failed.
	// the reconscile error message is "InvalidVpcEndpointId.NotFound: The VpcEndpoint ID 'vpce-0b1b3b3b3b3b3b3b3' does not exist"
	// In this case, we need to remove the finalizer from the endpoint and delete it manually.
	// this issue is caused by the issue in the crossplane aws provider.
	// TODO(cwen0): remove this workaround after the issue is fixed.
	condition := endpoint.Status.GetCondition(xpv1.TypeSynced)
	if strings.Contains(condition.Message, "InvalidVpcEndpointId.NotFound") {
		t.Logger.Info("vpc endpoint id not found, remove finalizer from the vpc s3 gateway endpoint")
		controllerutil.RemoveFinalizer(endpoint, managed.FinalizerName)
		if err := t.Local.Update(ctx, endpoint); err != nil {
			return deploy.Error(errors.Annotate(err, "failed to remove finalizer from vpc s3 gateway endpoint"))
		}
		return deploy.Done()
	}

	return deploy.Error(errors.New("vpc s3 gateway endpoint is not deleted"))
}

func (t *S3GatewayEndpoint) getPrivateRouteTables() []xpv1.Reference {
	var refs []xpv1.Reference
	// after route task is executed successfully, the route tables are stored in the internet status
	// so we can get the route tables from the internet status directly.
	// if the route task is not executed successfully, this task will not be executed.
	for _, rt := range t.Internet.Status.RouteTables {
		if namer.IsPrivateSubnet(rt.Name) {
			refs = append(refs, xpv1.Reference{Name: rt.Name})
		}
	}

	return refs
}

func (t *S3GatewayEndpoint) gatewayEndpointName() string {
	if name := t.Internet.GetAnnotations()[rsv1alpha1.AnnotationKeyOwnCrossplaneNameS3Gateway]; name != "" {
		return name
	}
	name := t.S3GatewayPrefix()
	return name + "-gateway-endpoint-s3"
}