// Copy from cluster-controller/controllers/networkpeer/task/eks/peer.go
package aws

import (
	"context"
	"fmt"
	"path"
	"strings"
	"time"

	"github.com/aws/aws-sdk-go-v2/service/ec2"
	ec2types "github.com/aws/aws-sdk-go-v2/service/ec2/types"
	"github.com/aws/aws-sdk-go/aws"
	ec2v1alpha1 "github.com/crossplane-contrib/provider-aws/apis/ec2/v1alpha1"
	xpcommon "github.com/crossplane/crossplane-runtime/apis/common/v1"
	"github.com/crossplane/crossplane-runtime/pkg/meta"
	"github.com/go-logr/logr"
	errors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/client"

	eksv1alpha1 "github.com/tidbcloud/eks-provider/api/v1alpha1"
	"github.com/tidbcloud/infra-provider/pkg/applier"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/ec2client"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/utils"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/vpcpeering/task"
)

const (
	LongWaitDuration    = time.Minute * 5
	defaultWaitDuration = time.Second * 30
	quickRetryDuration  = time.Second * 3
	maxResults          = 100
)

type VPCPeeringConnection struct {
	*task.BaseTask
}

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

func NewVCPPeeringConnection(ctx *task.Context) *VPCPeeringConnection {
	return &VPCPeeringConnection{
		BaseTask: &task.BaseTask{
			Context: ctx,
		},
	}
}

func (t *VPCPeeringConnection) Create(ctx context.Context) deploy.Result {
	// we already check the VPCPeering.Spec.Target.AWS in Gardian(), thus VPCPeering.Sepc.Target.AWS cannot be nil here.
	peerRegion := t.VPCPeering.Spec.Target.AWS.PeerRegion
	// peerCIDR := t.VPCPeering.Spec.Target.AWS.PeerCIDR
	peerVPCID := t.VPCPeering.Spec.Target.AWS.VpcID
	peerOwnerID := t.VPCPeering.Spec.Target.AWS.AccountID

	peeringConn := t.NewVPCPeeringConnection(peerRegion, peerVPCID, peerOwnerID)
	// FIXME(shenjun): the Applier is an bad implement since it will create a watch list goroutine on the given resource.
	mutateFn := mutateAWSVPCPeering
	// if the vpcpeering is migrated from infra-apiserver do not mutate anything.
	if islegacy := t.VPCPeering.Labels[v1alpha1.IsLegacyKey]; islegacy == "true" {
		mutateFn = mutateNothing
	}
	peeringConn, err := applier.NewApplier(t.Local, mutateFn).
		Expected(peeringConn).
		Apply(ctx)
	if err != nil {
		return deploy.Error(err)
	}
	if peeringConn.Status.AtProvider.VPCPeeringConnectionID != nil {
		t.VPCPeering.Status.ExternalName = *peeringConn.Status.AtProvider.VPCPeeringConnectionID
	}

	gk := peeringConn.GetObjectKind().GroupVersionKind().GroupKind().String()
	t.VPCPeering.Status.CrossplaneResourceName = path.Join(gk, peeringConn.Name)

	if externalName := meta.GetExternalName(peeringConn); externalName != "" {
		t.VPCPeering.Labels[v1alpha1.LabelKeyExternalName] = externalName
	}
	if t.VPCPeering.Annotations[v1alpha1.AnnotationKeySourceVPCID] == "" {
		t.VPCPeering.Annotations[v1alpha1.AnnotationKeySourceVPCID] = t.VPC.Status.VPC.VPCID
	}

	t.Logger.Info("peeringConn.Status.Conditions", "peeringConn", client.ObjectKeyFromObject(peeringConn))

	ready, synced := utils.GetConditions(peeringConn.Status.Conditions)
	if ready && synced {
		t.VPCPeering.Status.Phase = v1alpha1.VPCPeeringPhaseActive
		t.VPCPeering.Status.Reason = "VPCPeeringConnection is synced and ready"
	} else if !ready && synced {
		t.VPCPeering.Status.Phase = v1alpha1.VPCPeeringPhasePending
		t.VPCPeering.Status.Reason = "VPCPeeringConnection is synced but not ready"
	} else if !synced {
		var statuscode string
		if status := peeringConn.Status.AtProvider.Status; status != nil {
			statuscode = *status.Code
		}
		switch statuscode {
		case "active":
			t.VPCPeering.Status.Phase = v1alpha1.VPCPeeringPhaseActive
			t.VPCPeering.Status.Reason = "VPCPeeringConnection is already in active status"
		case "pending-acceptance":
			t.VPCPeering.Status.Phase = v1alpha1.VPCPeeringPhasePending
			t.VPCPeering.Status.Reason = "VPCPeeringConnection is under pending-acceptance"
		default:
			t.VPCPeering.Status.Reason = "VPCPeeringConnection is under unknown status"
		}
	}
	// always requeue
	return deploy.Done().WithRequeuAfter(LongWaitDuration)
}

func (t *VPCPeeringConnection) PostCreate(ctx context.Context) deploy.Result {
	peerCIDR := t.VPCPeering.Spec.Target.AWS.PeerCIDR
	sourceCIDR := t.VPC.Spec.CIDR
	peeringConn := new(ec2v1alpha1.VPCPeeringConnection)
	peeringConn.Name = genVpcPeeringConnectionName(t.VPCPeering)
	if err := t.Local.Get(ctx, client.ObjectKeyFromObject(peeringConn), peeringConn); err != nil {
		return deploy.Error(err)
	}
	// if the route table is not ensured, add the route table.
	routeTableReady := t.Internet.GetAnnotations()[v1alpha1.AnnotationKeyRouteTableEnsuredPeeringPrefix+t.VPCPeering.Name] == "true"
	if !routeTableReady {
		if peeringConn.Status.AtProvider.VPCPeeringConnectionID == nil {
			return deploy.Done().WithRequeuAfter(quickRetryDuration)
		}
		err := addRoute(ctx, t.EC2Client, t.Logger, []string{*peeringConn.Spec.ForProvider.VPCID}, peerCIDR, peeringConn.Status.AtProvider.VPCPeeringConnectionID)
		if err != nil {
			return deploy.Panic(err)
		}
		if t.VPC.Spec.AccountRef.AccountID == t.VPCPeering.Spec.Target.AWS.AccountID {
			err := addRoute(ctx, t.EC2Client, t.Logger, []string{*peeringConn.Spec.ForProvider.PeerVPCID}, sourceCIDR, peeringConn.Status.AtProvider.VPCPeeringConnectionID)
			if err != nil {
				return deploy.Panic(err)
			}
		}
		// set the annotation of the internet to true to indicate the route table is ensured.
		if t.Internet.Annotations == nil {
			t.Internet.Annotations = make(map[string]string)
		}
		t.Internet.Annotations[v1alpha1.AnnotationKeyRouteTableEnsuredPeeringPrefix+t.VPCPeering.Name] = "true"
		if err := t.Local.Update(ctx, t.Internet); err != nil {
			return deploy.Panic(err)
		}
	}
	// always requeue
	return deploy.Done().WithRequeuAfter(LongWaitDuration)
}

func (t *VPCPeeringConnection) Delete(ctx context.Context) deploy.Result {
	var name string
	// CrossplaneResourceName has fomat `King.Group/Name`
	if resourceName := t.VPCPeering.Status.CrossplaneResourceName; resourceName != "" {
		name = path.Base(resourceName)
	} else {
		name = genVpcPeeringConnectionName(t.VPCPeering)
	}
	peerCIDR := t.VPCPeering.Spec.Target.AWS.PeerCIDR
	// get then delete
	peeringConn := applier.NewClusterObject[*ec2v1alpha1.VPCPeeringConnection](name)
	if err := t.Local.Get(ctx, client.ObjectKeyFromObject(peeringConn), peeringConn); err != nil {
		if errors.IsNotFound(err) {
			return deploy.Done()
		}
		return deploy.Error(err)
	}
	if peeringConn.Status.AtProvider.VPCPeeringConnectionID != nil {
		err := deleteRoute(ctx, t.EC2Client, t.Logger, []string{*peeringConn.Spec.ForProvider.VPCID}, peerCIDR, peeringConn.Status.AtProvider.VPCPeeringConnectionID)
		if err != nil {
			return deploy.Panic(err)
		}
		sourceVPCAccountID := t.VPCPeering.GetAnnotations()[v1alpha1.AnnotationKeySourceAccountID]
		if sourceVPCAccountID == t.VPCPeering.Spec.Target.AWS.AccountID {
			// in delete task, VPC is not set in Gardian.
			// need to get the sourceCIDR from annotation
			sourceCIDR := t.VPCPeering.GetAnnotations()[v1alpha1.AnnotationKeySourceVPCCIDR]
			err := deleteRoute(ctx, t.EC2Client, t.Logger, []string{*peeringConn.Spec.ForProvider.PeerVPCID}, sourceCIDR, peeringConn.Status.AtProvider.VPCPeeringConnectionID)
			if err != nil {
				return deploy.Panic(err)
			}
		}
	}
	if err := t.Local.Delete(ctx, peeringConn); client.IgnoreNotFound(err) != nil {
		return deploy.Error(err)
	}
	return deploy.Done()
}

func (t *VPCPeeringConnection) PostDelete(ctx context.Context) deploy.Result {
	var name string
	// CrossplaneResourceName has fomat `King.Group/Name`
	if resourceName := t.VPCPeering.Status.CrossplaneResourceName; resourceName != "" {
		name = path.Base(resourceName)
	} else {
		name = genVpcPeeringConnectionName(t.VPCPeering)
	}
	obj := applier.NewClusterObject[*ec2v1alpha1.VPCPeeringConnection](name)
	if err := t.Local.Get(ctx, client.ObjectKeyFromObject(obj), obj); err != nil {
		if errors.IsNotFound(err) {
			return deploy.Done()
		}
		return deploy.Error(err)
	}
	return deploy.Error(fmt.Errorf("vpcpeeringconnection cr is not deleted"))
}

// github.com/crossplane-contrib/provider-aws/apis/ec2/v1alpha1
func (t *VPCPeeringConnection) NewVPCPeeringConnection(
	peerRegion string,
	peerVPCID string,
	peerOwnerID string,
) *ec2v1alpha1.VPCPeeringConnection {
	// refName := namer.UniqueIdentifier(t.VPC.ObjectMeta)
	providerConfigName := t.ProviderConfigName()
	name := genVpcPeeringConnectionName(t.VPCPeering)
	region := t.VPC.Spec.Region
	vpcID := t.VPC.Status.VPC.VPCID

	forProvider := ec2v1alpha1.VPCPeeringConnectionParameters{}
	forProvider.Region = region
	forProvider.VPCID = &vpcID
	forProvider.PeerRegion = &peerRegion
	forProvider.PeerVPCID = &peerVPCID
	forProvider.PeerOwnerID = &peerOwnerID
	forProvider.AcceptRequest = false
	// follow the same implementation in
	// https://github.com/tidbcloud/provider-aws/blob/pingcap/provider-aws/pkg/controller/vpcpeering/peering.go#L329
	forProvider.RequesterPeeringOptions = &ec2v1alpha1.VPCPeeringConnectionOptionsDescription{
		AllowDNSResolutionFromRemoteVPC:            pointer.Bool(true),
		AllowEgressFromLocalClassicLinkToRemoteVPC: pointer.Bool(false),
		AllowEgressFromLocalVPCToRemoteClassicLink: pointer.Bool(false),
	}

	return &ec2v1alpha1.VPCPeeringConnection{
		TypeMeta: metav1.TypeMeta{
			APIVersion: "ec2.aws.crossplane.io/v1alpha1",
			Kind:       "VPCPeeringConnection",
		},
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
		},
		Spec: ec2v1alpha1.VPCPeeringConnectionSpec{
			ResourceSpec: xpcommon.ResourceSpec{
				ProviderConfigReference: &xpcommon.Reference{
					Name: providerConfigName,
				},
			},
			ForProvider: forProvider,
		},
	}
}

func genVpcPeeringConnectionName(peering *v1alpha1.VPCPeering) string {
	if v, ok := peering.Annotations[eksv1alpha1.OwnCrossplaneNameAddKey]; ok {
		return v
	}
	return namer.UniqueIdentifier(peering.ObjectMeta, namer.WithLimit(40))
}

func mutateAWSVPCPeering(current, expected *ec2v1alpha1.VPCPeeringConnection) error {
	// we should not replace the label and anno since it is managed by crossplane provider.
	// current.Labels = expected.Labels
	// current.Annotations = expected.Annotations
	current.Spec.ResourceSpec = expected.Spec.ResourceSpec
	current.Spec.ForProvider = expected.Spec.ForProvider
	return nil
}

func mutateNothing(current, expected *ec2v1alpha1.VPCPeeringConnection) error {
	return nil
}

func addRoute(ctx context.Context, ec2cli ec2client.Client, logger logr.Logger, vpcIDs []string, peerCIDR string, pcx *string) error {
	filter := ec2types.Filter{
		Name:   aws.String("vpc-id"),
		Values: vpcIDs,
	}
	describeRouteTablesInput := &ec2.DescribeRouteTablesInput{
		Filters:    []ec2types.Filter{filter},
		MaxResults: aws.Int32(maxResults),
	}
	routeTablesRes, err := ec2cli.DescribeRouteTables(ctx, describeRouteTablesInput)
	if err != nil {
		logger.Error(err, "call DescribeRouteTables failed", "input", describeRouteTablesInput)
		return err
	}

	for _, rt := range routeTablesRes.RouteTables {
		createRouteInput := &ec2.CreateRouteInput{
			RouteTableId:           rt.RouteTableId,
			DestinationCidrBlock:   aws.String(peerCIDR),
			VpcPeeringConnectionId: pcx,
		}

		// If the route table exists, and the state is blackhole, remove it.
		for _, route := range rt.Routes {
			if route.DestinationCidrBlock != nil && *route.DestinationCidrBlock == peerCIDR && route.State == ec2types.RouteStateBlackhole {
				input := &ec2.DeleteRouteInput{
					DestinationCidrBlock: aws.String(peerCIDR),
					RouteTableId:         rt.RouteTableId,
				}
				_, err := ec2cli.DeleteRoute(ctx, input)
				if err != nil && !strings.Contains(err.Error(), "InvalidRoute.NotFound") {
					logger.Error(err, "call DeleteRoute failed", "input", input)
					return err
				}
			}
		}

		if _, err := ec2cli.CreateRoute(ctx, createRouteInput); err != nil {
			// FIXME: The error is not aws.Err type?
			if !strings.Contains(err.Error(), "RouteAlreadyExists") {
				logger.Error(err, "call CreateRoute failed", "input", createRouteInput)
				return err
			} else {
				for _, route := range rt.Routes {
					// The route identified by DestinationCidrBlock, if route table already have DestinationCidrBlock point to other vpc peering connetion ID, should be return error
					if route.DestinationCidrBlock != nil && *route.DestinationCidrBlock == peerCIDR {
						if route.VpcPeeringConnectionId != nil && *route.VpcPeeringConnectionId == *pcx {
							continue
						} else {
							return fmt.Errorf("failed add route for vpc peering connection: %s, routeID: %s, err: %w", *pcx, *rt.RouteTableId, err)
						}
					}
				}
			}
		}
	}
	return nil
}

func deleteRoute(ctx context.Context, ec2cli ec2client.Client, logger logr.Logger, vpcIDs []string, peerCIDR string, pcx *string) error {
	filter := ec2types.Filter{
		Name:   aws.String("vpc-id"),
		Values: vpcIDs,
	}
	describeRouteTablesInput := &ec2.DescribeRouteTablesInput{
		Filters:    []ec2types.Filter{filter},
		MaxResults: aws.Int32(maxResults),
	}
	routeTablesRes, err := ec2cli.DescribeRouteTables(ctx, describeRouteTablesInput)
	if err != nil {
		logger.Error(err, "call DescribeRouteTables failed", "input", describeRouteTablesInput)
		return err
	}

	for _, rt := range routeTablesRes.RouteTables {
		for _, r := range rt.Routes {
			if r.VpcPeeringConnectionId != nil && pcx != nil && *r.VpcPeeringConnectionId == *pcx {
				input := &ec2.DeleteRouteInput{
					DestinationCidrBlock: aws.String(peerCIDR),
					RouteTableId:         rt.RouteTableId,
				}
				_, err := ec2cli.DeleteRoute(ctx, input)
				if err != nil {
					if !strings.Contains(err.Error(), "InvalidRoute.NotFound") {
						logger.Error(err, "call DeleteRoute failed", "input", input)
						return err
					}
				}
			}
		}
	}
	return nil
}