//go:build intergration

package ft

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

	"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"
	"golang.org/x/sync/errgroup"

	xpmeta "github.com/crossplane/crossplane-runtime/pkg/meta"

	regionalpb "github.com/tidbcloud/infra-provider/api/tidbcloud/regional-server/v1"
	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/utils"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"sigs.k8s.io/controller-runtime/pkg/client"

	xpec2v1alpha1 "github.com/crossplane-contrib/provider-aws/apis/ec2/v1alpha1"
	xpgcpvpcpeeringv1beta1 "github.com/crossplane/provider-gcp/apis/vpcpeering/v1beta1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

	. "github.com/onsi/ginkgo/v2"
	. "github.com/onsi/gomega"

	computepb "cloud.google.com/go/compute/apiv1/computepb"
)

func waitVPCPeeringReady(ctx context.Context, peering *v1alpha1.VPCPeering, timeout, interval time.Duration) {
	Eventually(func() error {
		key := client.ObjectKeyFromObject(peering)
		err := kubeClient.Get(ctx, key, peering)
		if err != nil {
			logger.Error(err, "observing vpcpeering status", "vpcpeering", key.String())
			return err
		}
		if peering.Status.Phase != v1alpha1.VPCPeeringPhaseActive {
			logger.Info("wait vpcpeering ready", "vpcpeering", key.String(), "current phase", peering.Status.Phase)
			return errWait
		}
		return nil
	}, timeout, interval).Should(Succeed(), "vpcpeering should be ready")
}

func waitVPCPeeringPending(ctx context.Context, peering *v1alpha1.VPCPeering, timeout, interval time.Duration) {
	Eventually(func() error {
		key := client.ObjectKeyFromObject(peering)
		err := kubeClient.Get(ctx, key, peering)

		if err != nil {
			logger.Error(err, "observing vpcpeering status", "vpcpeering", key.String())
			return err
		}
		if peering.Status.Phase != v1alpha1.VPCPeeringPhasePending {
			logger.Info("wait vpcpeering pending", "vpcpeering", key.String(), "current phase", peering.Status.Phase)
			return errWait
		}
		return nil
	}, timeout, interval).Should(Succeed(), "vpcpeering should be pending")
}

func CreateSourceVPCReq() *regionalpb.CreateVpcRequest {
	return &regionalpb.CreateVpcRequest{
		RequestId: "ft-case-peering-foo",
		Vpc: &regionalpb.Vpc{
			TenantIds:  []string{f.TenantFoo},
			ProjectIds: []string{f.ProjectFoo},
			CidrBlock:  "10.0.0.0/16",
		},
	}
}

func CreateDestinationVPCReq() *regionalpb.CreateVpcRequest {
	return &regionalpb.CreateVpcRequest{
		RequestId: "ft-case-peering-bar",
		Vpc: &regionalpb.Vpc{
			TenantIds:  []string{f.TenantFoo},
			ProjectIds: []string{f.ProjectFoo},
			CidrBlock:  "10.1.0.0/16",
		},
	}
}

func CreateVpcPeeringReq(source *v1alpha1.VPC, dest *v1alpha1.VPC) *regionalpb.CreateVpcPeeringRequest {
	req := &regionalpb.CreateVpcPeeringRequest{
		VpcPeering: &regionalpb.VpcPeering{
			Id:                   "ft-foo",
			TenantId:             f.TenantFoo,
			ProjectId:            f.ProjectFoo,
			SourceCidrBlock:      source.Spec.CIDR,
			DestinationCidrBlock: dest.Spec.CIDR,
		},
	}
	switch source.Spec.Provider {
	case "aws", "eks":
		req.VpcPeering.Info = &regionalpb.VpcPeering_Aws_{
			Aws: &regionalpb.VpcPeering_Aws{
				SourceVpcId:          source.Status.VPC.VPCID,
				SourceAccountId:      source.Spec.AccountRef.AccountID,
				DestinationAccountId: dest.Spec.AccountRef.AccountID,
				DestinationVpcId:     dest.Status.VPC.VPCID,
				DestinationRegion:    dest.Spec.Region,
			},
		}
	case "gcp", "gke":
		req.VpcPeering.Info = &regionalpb.VpcPeering_Gcp_{
			Gcp: &regionalpb.VpcPeering_Gcp{
				SourceNetworkName:      source.Status.VPC.VPCID,
				SourceProjectId:        source.Spec.AccountRef.AccountID,
				DestinationProjectId:   dest.Spec.AccountRef.AccountID,
				DestinationNetworkName: dest.Status.VPC.VPCID,
			},
		}
	}
	return req
}

var _ = Describe("vpcpeering create api test", func() {
	It("create vpcpeering", func() {
		ctx := context.TODO()
		interval := 10 * time.Second
		concurrent := 5
		By("prepare resource")
		sourceVPCReq := CreateSourceVPCReq()
		sourceVpc := &v1alpha1.VPC{
			ObjectMeta: metav1.ObjectMeta{
				Name:      utils.MD5(sourceVPCReq.RequestId),
				Namespace: f.Config.Namespace,
			},
		}
		destinationVPCReq := CreateDestinationVPCReq()
		destionationVpc := &v1alpha1.VPC{
			ObjectMeta: metav1.ObjectMeta{
				Name:      utils.MD5(destinationVPCReq.RequestId),
				Namespace: f.Config.Namespace,
			},
		}
		_, err := vpcClient.CreateVpc(ctx, sourceVPCReq)
		Expect(err).To(Or(BeNil(), MatchError(ContainSubstring("HTTPSTATUSCODE:409"))))
		_, err = vpcClient.CreateVpc(ctx, destinationVPCReq)
		Expect(err).To(Or(BeNil(), MatchError(ContainSubstring("HTTPSTATUSCODE:409"))))
		waitVPCReady(ctx, sourceVpc, 10*time.Minute, interval)
		waitVPCReady(ctx, destionationVpc, 10*time.Minute, interval)
		By("create vpcpeering concurrently")
		req := CreateVpcPeeringReq(sourceVpc, destionationVpc)
		errG, _ := errgroup.WithContext(ctx)
		for i := 0; i < concurrent; i++ {
			errG.Go(func() error {
				_, err := vpcpeeringClient.CreateVpcPeering(ctx, req)
				if err != nil {
					if !strings.Contains(err.Error(), "HTTPSTATUSCODE:409") {
						return nil
					}
				}
				return nil
			})
		}
		Expect(errG.Wait()).To(BeNil())
		peering := &v1alpha1.VPCPeering{
			ObjectMeta: metav1.ObjectMeta{
				Name:      req.GetVpcPeering().GetId(),
				Namespace: f.Config.Namespace,
			},
		}
		_, err = vpcpeeringClient.CreateVpcPeering(ctx, req)
		Expect(err).To(BeNil())
		switch f.Config.Provider {
		case "aws":
			waitVPCPeeringPending(ctx, peering, 5*time.Minute, interval)
			By("check crossplane cr, it should be synced but but not ready")
			xppeeringName := path.Base(peering.Status.CrossplaneResourceName)
			peeringConn := new(xpec2v1alpha1.VPCPeeringConnection)
			peeringConn.Name = xppeeringName
			err := kubeClient.Get(ctx, client.ObjectKeyFromObject(peeringConn), peeringConn)
			Expect(err).To(BeNil())
			// external name should match
			Expect(peering.Status.ExternalName).To(Equal(xpmeta.GetExternalName(&peeringConn.ObjectMeta)))
			Expect(peering.GetLabels()[v1alpha1.LabelKeyExternalName]).To(Equal(xpmeta.GetExternalName(&peeringConn.ObjectMeta)))
			ready, synced := utils.GetConditions(peeringConn.Status.Conditions)
			Expect(ready).To(BeFalse())
			Expect(synced).To(BeTrue())
			By("check aws vpc, it should be pending acceptance")
			conntionID := peering.Status.ExternalName
			in := new(ec2.DescribeVpcPeeringConnectionsInput)
			in.VpcPeeringConnectionIds = []string{conntionID}
			in.Filters = append(in.Filters, ec2types.Filter{
				Name:   aws.String("requester-vpc-info.owner-id"),
				Values: []string{f.Config.DefaultAccount},
			})
			out, err := ec2Client.DescribeVpcPeeringConnections(ctx, in)
			Expect(err).To(BeNil())
			Expect(out.VpcPeeringConnections).To(HaveLen(1))
			Expect(out.VpcPeeringConnections[0].Status.Code).To(Equal(ec2types.VpcPeeringConnectionStateReasonCodePendingAcceptance))
		case "gcp":
			waitVPCPeeringReady(ctx, peering, 5*time.Minute, interval)
			By("check crossplane cr, it should be synced but but not ready")
			xppeeringName := path.Base(peering.Status.CrossplaneResourceName)
			xppeering := new(xpgcpvpcpeeringv1beta1.Peering)
			xppeering.Name = xppeeringName
			err := kubeClient.Get(ctx, client.ObjectKeyFromObject(xppeering), xppeering)
			Expect(err).To(BeNil())
			// external name should match
			// the tidbcloud's provider-gcp use externalname annotation incorrectly.
			// use `.Spec.ForPorvider.Name` for the real external name.
			Expect(peering.Status.ExternalName).To(Equal(xppeering.Spec.ForProvider.Name))
			Expect(peering.GetLabels()[v1alpha1.LabelKeyExternalName]).To(Equal(xppeering.Spec.ForProvider.Name))
			ready, synced := utils.GetConditions(xppeering.Status.Conditions)
			Expect(ready).To(BeTrue())
			Expect(synced).To(BeTrue())
			By("check gcp peering, it should be ready")
			in := new(computepb.GetNetworkRequest)
			in.Network = sourceVpc.Status.VPC.VPCID
			in.Project = sourceVpc.Spec.AccountRef.AccountID
			out, err := gcpNetworksClient.Get(ctx, in)
			Expect(err).To(BeNil())
			Expect(len(out.GetPeerings()) > 0).To(BeTrue(), "network %s should have at least one peering", in.Network)
			find := func(peerings []*computepb.NetworkPeering, name string) (*computepb.NetworkPeering, bool) {
				for _, v := range peerings {
					if aws.ToString(v.Name) == name {
						return v, true
					}
				}
				return nil, false
			}
			p, ok := find(out.GetPeerings(), peering.Status.ExternalName)
			Expect(ok).To(BeTrue(), "gcp return peerings: %s, expectd name: %s", out.GetPeerings(), peering.Status.ExternalName)
			Expect(aws.ToString(p.State)).To(Equal("ACTIVE"))
		}
	})
})

var _ = Describe("vpcpeering delete api test", func() {
	It("delete not existed vpcpeering", func() {
		in := new(regionalpb.DeleteVpcPeeringRequest)
		in.VpcPeeringId = fmt.Sprintf("notexist-%v", time.Now().Unix())
		_, err := vpcpeeringClient.DeleteVpcPeering(ctx, in)
		Expect(err).To(BeNil())
	})
	It("delete existed vpcpeering", func() {
		interval := 10 * time.Second
		By("prepare resource")
		sourceVPCReq := CreateSourceVPCReq()
		sourceVpc := &v1alpha1.VPC{
			ObjectMeta: metav1.ObjectMeta{
				Name:      utils.MD5(sourceVPCReq.RequestId),
				Namespace: f.Config.Namespace,
			},
		}
		destinationVPCReq := CreateDestinationVPCReq()
		destionationVpc := &v1alpha1.VPC{
			ObjectMeta: metav1.ObjectMeta{
				Name:      utils.MD5(destinationVPCReq.RequestId),
				Namespace: f.Config.Namespace,
			},
		}
		_, err := vpcClient.CreateVpc(ctx, sourceVPCReq)
		Expect(err).To(Or(BeNil(), MatchError(ContainSubstring("HTTPSTATUSCODE:409"))))
		_, err = vpcClient.CreateVpc(ctx, destinationVPCReq)
		Expect(err).To(Or(BeNil(), MatchError(ContainSubstring("HTTPSTATUSCODE:409"))))
		waitVPCReady(ctx, sourceVpc, 10*time.Minute, interval)
		waitVPCReady(ctx, destionationVpc, 10*time.Minute, interval)
		req := CreateVpcPeeringReq(sourceVpc, destionationVpc)
		peering := &v1alpha1.VPCPeering{
			ObjectMeta: metav1.ObjectMeta{
				Name:      req.VpcPeering.Id,
				Namespace: f.Config.Namespace,
			},
		}
		_, err = vpcpeeringClient.CreateVpcPeering(ctx, req)
		Expect(err).To(BeNil())
		switch f.Config.Provider {
		case "aws":
			waitVPCPeeringPending(ctx, peering, 10*time.Minute, interval)
		case "gcp":
			waitVPCPeeringReady(ctx, peering, 10*time.Minute, interval)
		}
		By("delete vpcpeering")
		in := new(regionalpb.DeleteVpcPeeringRequest)
		in.VpcPeeringId = req.GetVpcPeering().GetId()
		_, err = vpcpeeringClient.DeleteVpcPeering(ctx, in)
		Expect(err).To(BeNil())
		Eventually(func() bool {
			key := client.ObjectKeyFromObject(peering)
			err := kubeClient.Get(ctx, client.ObjectKeyFromObject(peering), peering)
			if !apierrors.IsNotFound(err) {
				if err != nil {
					logger.Error(err, "get vpcpeering failed", "vpcpeering", key.String())
				} else {
					logger.Info("wait vpcpeering deleted", "vcppeering", key.String(), "current phase", peering.Status.Phase)
				}
			}
			return apierrors.IsNotFound(err)
		}, 10*time.Minute, time.Minute).Should(BeTrue(), "vpcpeering %s should be deleted", client.ObjectKeyFromObject(peering).String())
		switch f.Config.Provider {
		case "aws":
			By("check vpcpeering cr, crossplane cr and aws/gcp resource should be deleted")
			xppeeringName := path.Base(peering.Status.CrossplaneResourceName)
			peeringConn := new(xpec2v1alpha1.VPCPeeringConnection)
			peeringConn.Name = xppeeringName
			Eventually(func() bool {
				key := client.ObjectKeyFromObject(peeringConn)
				err = kubeClient.Get(ctx, key, peeringConn)
				if !apierrors.IsNotFound(err) {
					if err != nil {
						logger.Error(err, "get vpcpeeringconn failed", "vpcpeeringconn", key.String())
					} else {
						logger.Info("wait vpcpeeringconn deleted", "vpcpeeringconn", key.String(), "current phase", peeringConn.Status.ConditionedStatus)
					}
				}
				return apierrors.IsNotFound(err)
			}, 10*time.Minute, time.Minute).Should(BeTrue(), "crossplane vpcpeeringconnection %s should be deleted", client.ObjectKeyFromObject(peeringConn).String())
			conntionID := peering.Status.ExternalName
			in := new(ec2.DescribeVpcPeeringConnectionsInput)
			in.VpcPeeringConnectionIds = []string{conntionID}
			in.Filters = append(in.Filters, ec2types.Filter{
				Name:   aws.String("requester-vpc-info.owner-id"),
				Values: []string{f.Config.DefaultAccount},
			})
			out, err := ec2Client.DescribeVpcPeeringConnections(ctx, in)
			Expect(err).To(BeNil())
			Expect(out.VpcPeeringConnections).To(HaveLen(1))
			Expect(aws.ToString(out.VpcPeeringConnections[0].VpcPeeringConnectionId)).To(Equal(conntionID))
			Expect(out.VpcPeeringConnections[0].Status.Code).To(Equal(ec2types.VpcPeeringConnectionStateReasonCodeDeleted))
		case "gcp":
			By("check vpcpeering cr, crossplane cr and aws/gcp resource should be deleted")
			xppeeringName := path.Base(peering.Status.CrossplaneResourceName)
			xppeering := new(xpgcpvpcpeeringv1beta1.Peering)
			xppeering.Name = xppeeringName
			Eventually(func() bool {
				key := client.ObjectKeyFromObject(xppeering)
				err = kubeClient.Get(ctx, key, xppeering)
				if !apierrors.IsNotFound(err) {
					if err != nil {
						logger.Error(err, "get xppeering failed", "xppeering", key.String())
					} else {
						logger.Info("wait xppeering deleted", "xppeering", key.String(), "current phase", xppeering.Status.ConditionedStatus)
					}
				}
				return apierrors.IsNotFound(err)
			}, 10*time.Minute, time.Minute).Should(BeTrue(), "crossplane peering %s should be deleted", client.ObjectKeyFromObject(xppeering).String())
			in := new(computepb.GetNetworkRequest)
			in.Network = sourceVpc.Status.VPC.VPCID
			in.Project = sourceVpc.Spec.AccountRef.AccountID
			out, err := gcpNetworksClient.Get(ctx, in)
			Expect(err).To(BeNil())
			find := func(peerings []*computepb.NetworkPeering, name string) (*computepb.NetworkPeering, bool) {
				for _, v := range peerings {
					if aws.ToString(v.Name) == name {
						return v, true
					}
				}
				return nil, false
			}
			_, ok := find(out.GetPeerings(), peering.Status.ExternalName)
			Expect(ok).To(BeFalse())
		}
	})
})

var _ = Describe("vpcpeering get api test", func() {
	It("get not existed vpcpeering", func() {
		in := new(regionalpb.GetVpcPeeringRequest)
		in.VpcPeeringId = fmt.Sprintf("notexist-%v", time.Now().Unix())
		_, err := vpcpeeringClient.GetVpcPeering(ctx, in)
		Expect(err).To(HaveOccurred())
		Expect(err.Error()).Should(ContainSubstring("HTTPSTATUSCODE:404"))
	})
	It("get existed vpcpeering", func() {
		interval := 10 * time.Second
		By("prepare resource")
		sourceVPCReq := CreateSourceVPCReq()
		sourceVpc := &v1alpha1.VPC{
			ObjectMeta: metav1.ObjectMeta{
				Name:      utils.MD5(sourceVPCReq.RequestId),
				Namespace: f.Config.Namespace,
			},
		}
		destinationVPCReq := CreateDestinationVPCReq()
		destionationVpc := &v1alpha1.VPC{
			ObjectMeta: metav1.ObjectMeta{
				Name:      utils.MD5(destinationVPCReq.RequestId),
				Namespace: f.Config.Namespace,
			},
		}
		_, err := vpcClient.CreateVpc(ctx, sourceVPCReq)
		Expect(err).To(Or(BeNil(), MatchError(ContainSubstring("HTTPSTATUSCODE:409"))))
		_, err = vpcClient.CreateVpc(ctx, destinationVPCReq)
		Expect(err).To(Or(BeNil(), MatchError(ContainSubstring("HTTPSTATUSCODE:409"))))
		waitVPCReady(ctx, sourceVpc, 10*time.Minute, interval)
		waitVPCReady(ctx, destionationVpc, 10*time.Minute, interval)
		req := CreateVpcPeeringReq(sourceVpc, destionationVpc)
		peering := &v1alpha1.VPCPeering{
			ObjectMeta: metav1.ObjectMeta{
				Name:      req.VpcPeering.Id,
				Namespace: f.Config.Namespace,
			},
		}
		_, err = vpcpeeringClient.CreateVpcPeering(ctx, req)
		Expect(err).To(BeNil(), "request %v", req)
		switch f.Config.Provider {
		case "aws":
			waitVPCPeeringPending(ctx, peering, 10*time.Minute, interval)
		case "gcp":
			waitVPCPeeringReady(ctx, peering, 10*time.Minute, interval)
		}
		By("get existed vpcpeering")
		in := new(regionalpb.GetVpcPeeringRequest)
		in.VpcPeeringId = req.VpcPeering.Id
		out, err := vpcpeeringClient.GetVpcPeering(ctx, in)
		Expect(err).To(BeNil())
		Expect(out.GetId()).To(Equal(req.VpcPeering.Id))
		Expect(out.GetTenantId()).To(Equal(req.VpcPeering.TenantId))
		Expect(out.GetProjectId()).To(Equal(req.VpcPeering.ProjectId))
	})
})

var _ = Describe("vpcpeering list api test", func() {
	It("list not existed vpcpeering", func() {
		{
			in := new(regionalpb.ListVpcPeeringsRequest)
			in.TenantId = fmt.Sprintf("notexist-%v", time.Now().Unix())
			out, err := vpcpeeringClient.ListVpcPeerings(ctx, in)
			Expect(err).To(BeNil())
			Expect(out.GetVpcPeerings()).To(HaveLen(0))
		}
		{
			in := new(regionalpb.ListVpcPeeringsRequest)
			in.TenantId = fmt.Sprintf("notexist-%v", time.Now().Unix())
			in.ProjectId = fmt.Sprintf("notexist-%v", time.Now().Unix())
			out, err := vpcpeeringClient.ListVpcPeerings(ctx, in)
			Expect(err).To(BeNil())
			Expect(out.GetVpcPeerings()).To(HaveLen(0))
		}
	})
	It("list existed vpcpeering", func() {
		interval := 10 * time.Second
		By("prepare resource")
		sourceVPCReq := CreateSourceVPCReq()
		sourceVpc := &v1alpha1.VPC{
			ObjectMeta: metav1.ObjectMeta{
				Name:      utils.MD5(sourceVPCReq.RequestId),
				Namespace: f.Config.Namespace,
			},
		}
		destinationVPCReq := CreateDestinationVPCReq()
		destionationVpc := &v1alpha1.VPC{
			ObjectMeta: metav1.ObjectMeta{
				Name:      utils.MD5(destinationVPCReq.RequestId),
				Namespace: f.Config.Namespace,
			},
		}
		_, err := vpcClient.CreateVpc(ctx, sourceVPCReq)
		Expect(err).To(Or(BeNil(), MatchError(ContainSubstring("HTTPSTATUSCODE:409"))))
		_, err = vpcClient.CreateVpc(ctx, destinationVPCReq)
		Expect(err).To(Or(BeNil(), MatchError(ContainSubstring("HTTPSTATUSCODE:409"))))
		waitVPCReady(ctx, sourceVpc, 10*time.Minute, interval)
		waitVPCReady(ctx, destionationVpc, 10*time.Minute, interval)
		req := CreateVpcPeeringReq(sourceVpc, destionationVpc)
		peering := &v1alpha1.VPCPeering{
			ObjectMeta: metav1.ObjectMeta{
				Name:      req.VpcPeering.Id,
				Namespace: f.Config.Namespace,
			},
		}
		_, err = vpcpeeringClient.CreateVpcPeering(ctx, req)
		Expect(err).To(BeNil())
		switch f.Config.Provider {
		case "aws":
			waitVPCPeeringPending(ctx, peering, 10*time.Minute, interval)
		case "gcp":
			waitVPCPeeringReady(ctx, peering, 10*time.Minute, interval)
		}
		By("list existed vpcpeering")
		in := new(regionalpb.ListVpcPeeringsRequest)
		in.TenantId = req.VpcPeering.TenantId
		in.ProjectId = req.VpcPeering.ProjectId
		out, err := vpcpeeringClient.ListVpcPeerings(ctx, in)
		Expect(err).To(BeNil())
		Expect(len(out.GetVpcPeerings()) > 0).To(BeTrue())
		for _, item := range out.GetVpcPeerings() {
			Expect(item.GetTenantId()).To(Equal(in.TenantId))
			Expect(item.GetProjectId()).To(Equal(in.ProjectId))
		}
	})
})