package networkpeer

import (
	"context"
	"errors"
	"fmt"
	"math/rand"
	"net/http"
	"strings"
	"time"

	"github.com/aws/aws-sdk-go-v2/service/ec2"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	awsvpcpeeringv1alpha1 "github.com/crossplane/provider-aws/apis/vpcpeering/v1alpha1"
	gcpvpcpeeringv1beta1 "github.com/crossplane/provider-gcp/apis/vpcpeering/v1beta1"
	"github.com/go-logr/logr"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	"google.golang.org/api/googleapi"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/builder"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/predicate"

	awssdk "github.com/aws/aws-sdk-go/aws"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-api/controllers/common"
	"github.com/tidbcloud/infra-api/controllers/reconciler"
	"github.com/tidbcloud/infra-api/controllers/utils"
	"github.com/tidbcloud/infra-api/pkg/consts"
	"github.com/tidbcloud/infra-api/pkg/gardener/aws/apis/v1alpha1"
	metricsctrl "github.com/tidbcloud/infra-api/pkg/metrics/controller"
	awsapi "github.com/tidbcloud/infra-api/pkg/providers/aws"
	"github.com/tidbcloud/infra-api/pkg/providers/gcp"
	tracerutil "github.com/tidbcloud/infra-api/pkg/tracer"
)

const (
	NetworkpeerFinalizer = "infra.cloud.pingcap.com/finalizer"
	errNotNetworkpeer    = "not a networkpeer resource"
	EnsureSecurityGroup  = "infra.cloud.pingcap.com/ensure-security-group"
	EnsureFireWall       = "infra.cloud.pingcap.com/ensure-firewall"
	// AWS peering reconcile duration, consistent with crossplane provider-aws
	AWSPeeringPollInterval = 5 * time.Minute
	// GCP peering reconcile duration, Consistent with crossplane provider-gcp
	GCPPeeringPollInterval = 1 * time.Minute
	GCPInternalPeerName    = "internal-vpc-peering"
	Charset                = "abcdefghijklmnopqrstuvwxyz0123456789"
)

var (
	scheme     = runtime.NewScheme()
	tracer     = otel.Tracer("NetworkPeerReconciler")
	seededRand = rand.New(rand.NewSource(time.Now().UnixNano()))
)

func init() {
	utilruntime.Must(clientgoscheme.AddToScheme(scheme))
	utilruntime.Must(v1alpha1.AddToScheme(scheme))
}

// NetworkPeerReconciler reconciles a NetworkPeer object
type NetworkPeerReconciler struct {
	reconciler *reconciler.Reconciler
	Log        logr.Logger
}

type networkPeerConnector struct {
	client    client.Client
	config    *cloudv1.InfraAPIConfig
	AWSClient awsapi.Client
	GCPAPI    gcp.GCPAPI
}

type clusterNetworkPeer struct {
	cli                   client.Client
	network               *cloudv1.Network
	providerSpec          *cloudv1.GardenerProviderSpec
	gardenerNetworkStatus *cloudv1.GardenerNetworkStatus
	AWSClient             awsapi.Client
	GCPAPI                gcp.GCPAPI
}

func NewNetworkPeerReconciler(mgr ctrl.Manager, log logr.Logger, config *cloudv1.InfraAPIConfig, awsCli awsapi.Client, gcpCli gcp.GCPAPI) *NetworkPeerReconciler {
	return &NetworkPeerReconciler{
		Log: log,
		reconciler: &reconciler.Reconciler{
			Client: mgr.GetClient(),
			ExternalObj: networkPeerConnector{
				client:    mgr.GetClient(),
				config:    config,
				AWSClient: awsCli,
				GCPAPI:    gcpCli,
			},
			NewResource: func() client.Object {
				return &cloudv1.NetworkPeer{}
			},
		},
	}
}

//+kubebuilder:rbac:groups=cloud.pingcap.com,resources=networkpeers,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=cloud.pingcap.com,resources=networkpeers/status,verbs=get;update;patch
//+kubebuilder:rbac:groups=cloud.pingcap.com,resources=networkpeers/finalizers,verbs=update

// Reconcile is part of the main kubernetes reconciliation loop which aims to
// move the current state of the cluster closer to the desired state.
// TODO(user): Modify the Reconcile function to compare the state specified by
// the NetworkPeer object against the actual cluster state, and then
// perform operations to make the cluster state reflect the state specified by
// the user.
//
// For more details, check Reconcile and its Result here:
// - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.7.2/pkg/reconcile
func (r *NetworkPeerReconciler) Reconcile(ctx context.Context, req ctrl.Request) (_ ctrl.Result, errResult error) {
	start := time.Now()
	logger := r.Log.WithValues("networkpeer", req.NamespacedName)
	ctx = log.IntoContext(ctx, logger)

	spanCtx, span := tracer.Start(ctx, "NetworkPeer Reconcile")
	defer func() {
		metricsctrl.ReconcileDuration.WithLabelValues("network_peer").Observe(time.Since(start).Seconds())
		tracerutil.EndSpan(span, errResult)
	}()

	span.SetAttributes(
		attribute.String("networkpeer_name", req.Name),
		attribute.String("networkpeer_namespace", req.Namespace),
	)

	return r.reconciler.Reconcile(spanCtx, req, NetworkpeerFinalizer)
}

func (npc networkPeerConnector) Connect(ctx context.Context, obj client.Object) (_ reconciler.ExternalObjClient, errResult error) {
	ctx, span := tracer.Start(ctx, "networkPeerConnector.Connect")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	np, ok := obj.(*cloudv1.NetworkPeer)
	if !ok {
		return nil, errors.New(errNotNetworkpeer)
	}

	span.SetAttributes(
		attribute.String("networkpeer_name", np.Name),
		attribute.String("networkpeer_namespace", np.Namespace),
	)

	var n cloudv1.Network
	networkNamespaceName := types.NamespacedName{
		Name:      np.Spec.Source,
		Namespace: np.ObjectMeta.Namespace,
	}
	err := npc.client.Get(ctx, networkNamespaceName, &n)
	if err != nil {
		return nil, err
	}

	provider := new(cloudv1.Provider)
	err = npc.client.Get(ctx, types.NamespacedName{Name: n.Spec.Provider}, provider)
	if err != nil {
		return nil, err
	}

	providerSpec, err := cloudv1.GardenerProviderSpecForProvider(provider)
	if err != nil {
		return nil, err
	}

	gardenerNetworkStatus, err := cloudv1.GardenerNetworkStatusFromNetwork(&n)
	if err != nil {
		return nil, err
	}
	switch provider.Spec.Type {
	case consts.ProviderTypeGCP:
		if gardenerNetworkStatus.GCP.NetworkName == nil {
			return nil, errors.New("wait the network get ready")
		}
	case consts.ProviderTypeAWS:
		if gardenerNetworkStatus.AWS.VpcID == nil {
			return nil, errors.New("wait the network get ready")
		}
	default:
		return nil, fmt.Errorf("unsupported provider type %s", provider.Spec.Type)
	}

	return &clusterNetworkPeer{
		cli:          npc.client,
		network:      &n,
		providerSpec: providerSpec,

		gardenerNetworkStatus: gardenerNetworkStatus,
		AWSClient:             npc.AWSClient,
		GCPAPI:                npc.GCPAPI,
	}, nil
}

func Ready(conditions []xpv1.Condition) bool {
	if len(conditions) == 0 {
		return false
	}

	for _, c := range conditions {
		if c.Type == xpv1.TypeReady {
			if c.Status == corev1.ConditionTrue {
				return true
			}
		}
	}

	return false
}

func (cnp *clusterNetworkPeer) Observe(ctx context.Context, obj client.Object) (_ reconciler.ExternalObservation, errResult error) {
	ctx, span := tracer.Start(ctx, "clusterNetworkPeer.Observe")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	np, ok := obj.(*cloudv1.NetworkPeer)
	if !ok {
		return reconciler.ExternalObservation{}, errors.New(errNotNetworkpeer)
	}

	span.SetAttributes(
		attribute.String("networkpeer_name", np.Name),
		attribute.String("networkpeer_namespace", np.Namespace),
	)

	if cnp.network.Spec.Provider == cloudv1.ProviderAWS {
		var awsvpcpeering awsvpcpeeringv1alpha1.VPCPeeringConnection
		if err := cnp.cli.Get(ctx, types.NamespacedName{
			Name: np.ClusterScopeName(),
		}, &awsvpcpeering); err != nil {
			if reconciler.IgnoreNotFound(err) != nil {
				return reconciler.ExternalObservation{}, err
			}
			np.Status.Phase = cloudv1.NetworkPeerPhasePending
			return reconciler.ExternalObservation{ResourceExists: false}, cnp.cli.Status().Update(ctx, np)
		}

		if awsvpcpeering.Status.AtProvider.VPCPeeringConnectionID != nil {
			np.Status.PeeringID = *awsvpcpeering.Status.AtProvider.VPCPeeringConnectionID
		}

		if awsvpcpeering.Spec.ForProvider.HostZoneID != nil {
			np.Status.HostedZoneID = *awsvpcpeering.Spec.ForProvider.HostZoneID
		}

		if awsvpcpeering.Status.AtProvider.RequesterVPCInfo != nil && awsvpcpeering.Status.AtProvider.RequesterVPCInfo.CIDRBlock != nil {
			np.Status.VpcCidr = *awsvpcpeering.Status.AtProvider.RequesterVPCInfo.CIDRBlock
		}
		// if peering is internal, need add requester vpc cidr to accepter node security group
		securityGroupReady := ensureSecurityGroup(np)
		if Ready(awsvpcpeering.Status.ResourceStatus.ConditionedStatus.Conditions) && securityGroupReady {
			np.Status.Phase = cloudv1.NetworkPeerPhaseActive
			// Sometimes users may delete peering on AWS cloud provder, if not reconcile, the status will always be active
			return reconciler.ExternalObservation{ResourceExists: true, ResourceUpToDate: true, PollInterval: AWSPeeringPollInterval}, cnp.cli.Status().Update(ctx, np)
		}

		np.Status.Phase = cloudv1.NetworkPeerPhasePending
		return reconciler.ExternalObservation{ResourceExists: true}, cnp.cli.Status().Update(ctx, np)
	}

	if cnp.network.Spec.Provider == cloudv1.ProviderGCP {
		var gcpvpcpeering gcpvpcpeeringv1beta1.Peering
		if err := cnp.cli.Get(ctx, types.NamespacedName{
			Name: np.ClusterScopeName(),
		}, &gcpvpcpeering); err != nil {
			if reconciler.IgnoreNotFound(err) != nil {
				return reconciler.ExternalObservation{}, err
			}
			np.Status.Phase = cloudv1.NetworkPeerPhasePending
			return reconciler.ExternalObservation{ResourceExists: false}, cnp.cli.Status().Update(ctx, np)
		}
		fireWallReady := ensureFireWall(np)
		if Ready(gcpvpcpeering.Status.ResourceStatus.ConditionedStatus.Conditions) && fireWallReady {
			np.Status.Phase = cloudv1.NetworkPeerPhaseActive
			// Sometimes users may delete peering on GCP cloud provder, if not reconcile, the status will always be active
			return reconciler.ExternalObservation{ResourceExists: true, ResourceUpToDate: true, PollInterval: GCPPeeringPollInterval}, cnp.cli.Status().Update(ctx, np)
		} else {
			np.Status.Phase = cloudv1.NetworkPeerPhasePending
			return reconciler.ExternalObservation{ResourceExists: true}, cnp.cli.Status().Update(ctx, np)
		}
	}

	return reconciler.ExternalObservation{}, errors.New("unsupported resource")
}

func (cnp *clusterNetworkPeer) Create(ctx context.Context, obj client.Object) (errResult error) {
	ctx, span := tracer.Start(ctx, "clusterNetworkPeer.Create")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	np, ok := obj.(*cloudv1.NetworkPeer)
	if !ok {
		return errors.New(errNotNetworkpeer)
	}

	span.SetAttributes(
		attribute.String("networkpeer_name", np.Name),
		attribute.String("networkpeer_namespace", np.Namespace),
	)

	if cnp.network.Spec.Provider != cloudv1.ProviderAWS && cnp.network.Spec.Provider != cloudv1.ProviderGCP {
		return errors.New("unknown type to create")
	}

	// we need to update the status before create the underlying CR, otherwise if the status update failed,
	// then this `Create` will be skipped (without status update again) in the next round of reconcile after `Observe`.
	np.Status.Phase = cloudv1.NetworkPeerPhasePending
	if err := cnp.cli.Status().Update(ctx, np); err != nil {
		return fmt.Errorf("failed update networkpeering status, error: %v", err)
	}

	if cnp.network.Spec.Provider == cloudv1.ProviderAWS {
		var awsnp *awsvpcpeeringv1alpha1.VPCPeeringConnection
		// If target network is not null will process internal vpc peering
		if np.Spec.Target.TargetNetwork != nil {
			tn, targetGardenerStatus, err := cnp.getTargetNetwork(ctx, np, cnp.network.Spec.Provider)
			if err != nil {
				return err
			}
			awsnp = &awsvpcpeeringv1alpha1.VPCPeeringConnection{
				ObjectMeta: metav1.ObjectMeta{
					Name: np.ClusterScopeName(),
				},
				Spec: awsvpcpeeringv1alpha1.VPCPeeringConnectionSpec{
					ResourceSpec: xpv1.ResourceSpec{
						ProviderConfigReference: &xpv1.Reference{
							Name: cnp.providerSpec.AWS.CrossplaneProvider,
						},
					},
					ForProvider: awsvpcpeeringv1alpha1.VPCPeeringConnectionParameters{
						Region:      cnp.network.Spec.Region,
						PeerOwnerID: &cnp.providerSpec.AWS.AccountID,
						PeerRegion:  &tn.Spec.Region,
						PeerVPCID:   targetGardenerStatus.AWS.VpcID,
						PeerCIDR:    &tn.Spec.CIDR,
						VPCID:       cnp.gardenerNetworkStatus.AWS.VpcID,
						HostZoneID:  cnp.gardenerNetworkStatus.AWS.HostedZoneID,
					},
				},
			}
		} else {
			awsnp = &awsvpcpeeringv1alpha1.VPCPeeringConnection{
				ObjectMeta: metav1.ObjectMeta{
					Name: np.ClusterScopeName(),
				},
				Spec: awsvpcpeeringv1alpha1.VPCPeeringConnectionSpec{
					ResourceSpec: xpv1.ResourceSpec{
						ProviderConfigReference: &xpv1.Reference{
							Name: cnp.providerSpec.AWS.CrossplaneProvider,
						},
					},
					ForProvider: awsvpcpeeringv1alpha1.VPCPeeringConnectionParameters{
						Region:      cnp.network.Spec.Region,
						PeerOwnerID: &np.Spec.Target.AWS.AccountID,
						PeerRegion:  &np.Spec.Target.AWS.PeerRegion,
						PeerVPCID:   &np.Spec.Target.AWS.VpcID,
						PeerCIDR:    &np.Spec.Target.AWS.PeerCIDR,
						VPCID:       cnp.gardenerNetworkStatus.AWS.VpcID,
						HostZoneID:  cnp.gardenerNetworkStatus.AWS.HostedZoneID,
					},
				},
			}
		}
		if _, err := controllerutil.CreateOrUpdate(ctx, cnp.cli, awsnp, func() error {
			return nil
		}); err != nil {
			return err
		}
	}

	if cnp.network.Spec.Provider == cloudv1.ProviderGCP {
		var gcpnp *gcpvpcpeeringv1beta1.Peering
		// GCP internal vpc peering
		if np.Spec.Target.TargetNetwork != nil {
			_, targetGardenerStatus, err := cnp.getTargetNetwork(ctx, np, cnp.network.Spec.Provider)
			if err != nil {
				return err
			}
			gcpnp = &gcpvpcpeeringv1beta1.Peering{
				ObjectMeta: metav1.ObjectMeta{
					Name: np.ClusterScopeName(),
				},
				Spec: gcpvpcpeeringv1beta1.PeeringSpec{
					ResourceSpec: xpv1.ResourceSpec{
						ProviderConfigReference: &xpv1.Reference{
							Name: cnp.providerSpec.GCP.CrossplaneProvider,
						},
					},
					ForProvider: gcpvpcpeeringv1beta1.PeeringParameters{
						Project: cnp.providerSpec.GCP.ProjectID,
						Network: *cnp.gardenerNetworkStatus.GCP.NetworkName,
						Name:    np.ObjectMeta.Name,
						PeerNetwork: fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/%s/global/networks/%s",
							*targetGardenerStatus.GCP.ProjectID, *targetGardenerStatus.GCP.NetworkName),
					},
				},
			}
		} else {
			gcpnp = &gcpvpcpeeringv1beta1.Peering{
				ObjectMeta: metav1.ObjectMeta{
					Name: np.ClusterScopeName(),
				},
				Spec: gcpvpcpeeringv1beta1.PeeringSpec{
					ResourceSpec: xpv1.ResourceSpec{
						ProviderConfigReference: &xpv1.Reference{
							Name: cnp.providerSpec.GCP.CrossplaneProvider,
						},
					},
					ForProvider: gcpvpcpeeringv1beta1.PeeringParameters{
						Project: cnp.providerSpec.GCP.ProjectID,
						Network: *cnp.gardenerNetworkStatus.GCP.NetworkName,
						Name:    np.ObjectMeta.Name,
						PeerNetwork: fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/%s/global/networks/%s",
							np.Spec.Target.GCP.ProjectID, np.Spec.Target.GCP.PeerNetwork),
					},
				},
			}
		}

		if _, err := controllerutil.CreateOrUpdate(ctx, cnp.cli, gcpnp, func() error {
			return nil
		}); err != nil {
			return err
		}
	}

	return nil
}

func (cnp *clusterNetworkPeer) Update(ctx context.Context, obj client.Object) (errResult error) {
	ctx, span := tracer.Start(ctx, "clusterNetworkPeer.Update")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	np, ok := obj.(*cloudv1.NetworkPeer)
	if !ok {
		return errors.New(errNotNetworkpeer)
	}

	span.SetAttributes(
		attribute.String("networkpeer_name", np.Name),
		attribute.String("networkpeer_namespace", np.Namespace),
	)

	if cnp.network.Spec.Provider == cloudv1.ProviderAWS {
		securityGroupReady := ensureSecurityGroup(np)
		if !securityGroupReady {
			if err := cnp.addIngressRuleToTargetNetworkSecurityGroup(ctx, np); err != nil {
				return err
			}
			if err := cnp.addIngressRuleToSourceNetworkSecurityGroup(ctx, np); err != nil {
				return err
			}
			if np.Annotations == nil {
				np.Annotations = make(map[string]string)
			}
			np.Annotations[EnsureSecurityGroup] = "true"
			return cnp.cli.Update(ctx, np)
		}

		var awsvpcpeering awsvpcpeeringv1alpha1.VPCPeeringConnection
		if err := cnp.cli.Get(ctx, types.NamespacedName{
			Name: np.ClusterScopeName(),
		}, &awsvpcpeering); err != nil {
			return err
		}

		if Ready(awsvpcpeering.Status.ResourceStatus.Conditions) && securityGroupReady {
			np.Status.Phase = cloudv1.NetworkPeerPhaseActive
		} else {
			np.Status.Phase = cloudv1.NetworkPeerPhasePending
		}
		return cnp.cli.Status().Update(ctx, np)
	}

	if cnp.network.Spec.Provider == cloudv1.ProviderGCP {
		fireWallReady := ensureFireWall(np)
		if !fireWallReady {
			// add security group for both source and target shoot
			sourceGardenerStatus := cnp.gardenerNetworkStatus
			targetNetwork, targetGardenerStatus, err := cnp.getTargetNetwork(ctx, np, cnp.network.Spec.Provider)
			if err != nil {
				return err
			}
			// add source network cidr to target network firewll rules
			err = cnp.addIngressPolicyToNetworkFirewall(*targetGardenerStatus.GCP.ProjectID, *targetGardenerStatus.GCP.NetworkName, cnp.network.Spec.CIDR)
			if err != nil {
				return err
			}
			// add target network cidr to source network firewll rules
			err = cnp.addIngressPolicyToNetworkFirewall(*sourceGardenerStatus.GCP.ProjectID, *sourceGardenerStatus.GCP.NetworkName, targetNetwork.Spec.CIDR)
			if err != nil {
				return err
			}
			if np.Annotations == nil {
				np.Annotations = make(map[string]string)
			}
			np.Annotations[EnsureFireWall] = "true"
			return cnp.cli.Update(ctx, np)
		}
		var gcpvpcpeering gcpvpcpeeringv1beta1.Peering
		if err := cnp.cli.Get(ctx, types.NamespacedName{
			Name: np.ClusterScopeName(),
		}, &gcpvpcpeering); err != nil {
			return err
		}
		if Ready(gcpvpcpeering.Status.ResourceStatus.Conditions) {
			np.Status.Phase = cloudv1.NetworkPeerPhaseActive
		} else {
			np.Status.Phase = cloudv1.NetworkPeerPhasePending
		}
		return cnp.cli.Status().Update(ctx, np)
	}

	return errors.New("unknown resource to update")
}

func (cnp *clusterNetworkPeer) Delete(ctx context.Context, obj client.Object) (_ reconciler.ExternalDeletion, errResult error) {
	ctx, span := tracer.Start(ctx, "clusterNetworkPeer.Delete")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	np, ok := obj.(*cloudv1.NetworkPeer)
	if !ok {
		return reconciler.ExternalDeletion{}, errors.New(errNotNetworkpeer)
	}

	span.SetAttributes(
		attribute.String("networkpeer_name", np.Name),
		attribute.String("networkpeer_namespace", np.Namespace),
	)

	if cnp.network.Spec.Provider == cloudv1.ProviderAWS {
		var awsvpcpeering awsvpcpeeringv1alpha1.VPCPeeringConnection
		awsvpcpeering.Name = np.ClusterScopeName()
		err := cnp.cli.Delete(ctx, &awsvpcpeering)
		if err != nil {
			if apierrors.IsNotFound(err) {
				return reconciler.ExternalDeletion{ResourceDeleted: true}, nil
			} else {
				return reconciler.ExternalDeletion{ResourceDeleted: false}, err
			}
		} else {
			return reconciler.ExternalDeletion{ResourceDeleted: false}, fmt.Errorf("Wait aws vpc peering %s deleted", awsvpcpeering.Name)
		}
	}

	if cnp.network.Spec.Provider == cloudv1.ProviderGCP {
		var gcpvpcpeering gcpvpcpeeringv1beta1.Peering
		gcpvpcpeering.Name = np.ClusterScopeName()
		err := cnp.cli.Delete(ctx, &gcpvpcpeering)
		if err != nil {
			if err != nil {
				if apierrors.IsNotFound(err) {
					return reconciler.ExternalDeletion{ResourceDeleted: true}, nil
				} else {
					return reconciler.ExternalDeletion{ResourceDeleted: false}, err
				}
			} else {
				return reconciler.ExternalDeletion{ResourceDeleted: false}, fmt.Errorf("Wait gcp vpc peering %s deleted", gcpvpcpeering.Name)
			}
		}
	}

	return reconciler.ExternalDeletion{ResourceDeleted: false}, errors.New("unknown resource to delete")
}

func (cnp *clusterNetworkPeer) addIngressRuleToTargetNetworkSecurityGroup(ctx context.Context, np *cloudv1.NetworkPeer) error {
	targetNetwork, targetNetworkStatus, err := cnp.getTargetNetwork(ctx, np, cnp.network.Spec.Provider)
	if err != nil {
		return err
	}
	nodeSg := findAWSNodeSecurityGroup(targetNetworkStatus.AWS.SecurityGroups)
	if nodeSg == "" {
		return fmt.Errorf("can not find aws node security group")
	}
	sourceNetwork := cnp.network

	return cnp.addIngressRuleToSecurityGroup(ctx, nodeSg, sourceNetwork.Spec.CIDR, targetNetwork.Spec.Region)
}

func (cnp *clusterNetworkPeer) addIngressRuleToSourceNetworkSecurityGroup(ctx context.Context, np *cloudv1.NetworkPeer) error {
	targetNetwork, _, err := cnp.getTargetNetwork(ctx, np, cnp.network.Spec.Provider)
	if err != nil {
		return err
	}

	sourceNetwork := cnp.network
	sourceNetworkStatus := cnp.gardenerNetworkStatus

	nodeSg := findAWSNodeSecurityGroup(sourceNetworkStatus.AWS.SecurityGroups)
	if nodeSg == "" {
		return fmt.Errorf("can not find aws node security group")
	}

	return cnp.addIngressRuleToSecurityGroup(ctx, nodeSg, targetNetwork.Spec.CIDR, sourceNetwork.Spec.Region)
}

func (cnp *clusterNetworkPeer) addIngressRuleToSecurityGroup(ctx context.Context, sg string, cidr string, region string) error {
	_, err := cnp.AWSClient.Ec2().AuthorizeSecurityGroupIngress(ctx, &ec2.AuthorizeSecurityGroupIngressInput{
		CidrIp:  awssdk.String(cidr),
		GroupId: awssdk.String(sg),
		// all protocol
		IpProtocol: awssdk.String("-1"),
	}, awsapi.WithEC2Region(region))
	if err != nil {
		if !strings.Contains(fmt.Sprintf("%s", err), "InvalidPermission.Duplicate") {
			return fmt.Errorf("failed authorize security group ingress: %v", err)
		}
	}
	return nil
}

// SetupWithManager sets up the controller with the Manager.
func (r *NetworkPeerReconciler) SetupWithManager(mgr ctrl.Manager, p predicate.Predicate) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&cloudv1.NetworkPeer{}, builder.WithPredicates(p)).
		WithOptions(controller.Options{RateLimiter: common.DefaultRateLimiter()}).
		WithEventFilter(utils.GetMigrationAnnotationSkipPredicator()).
		Complete(r)
}

func findAWSNodeSecurityGroup(sgs []cloudv1.SecurityGroup) string {
	for _, sg := range sgs {
		if sg.Purpose == cloudv1.NodePurposeSecurityGroup {
			return sg.ID
		}
	}
	return ""
}

func isInternalPeering(np *cloudv1.NetworkPeer) bool {
	if np.Spec.Target.TargetNetwork != nil {
		return true
	}
	return false
}

func ensureSecurityGroup(np *cloudv1.NetworkPeer) bool {
	securityGroupReady := true
	if isInternalPeering(np) {
		ensureSecurityGroup := np.Annotations[EnsureSecurityGroup]
		if ensureSecurityGroup != "true" {
			securityGroupReady = false
		}
	}
	return securityGroupReady
}

func ensureFireWall(np *cloudv1.NetworkPeer) bool {
	securityFireWall := true
	if isInternalPeering(np) {
		ensureSecurityGroup := np.Annotations[EnsureFireWall]
		if ensureSecurityGroup != "true" {
			securityFireWall = false
		}
	}
	return securityFireWall
}

func (cnp *clusterNetworkPeer) getTargetNetwork(ctx context.Context, np *cloudv1.NetworkPeer, provider string) (*cloudv1.Network, *cloudv1.GardenerNetworkStatus, error) {
	var tn cloudv1.Network
	if err := cnp.cli.Get(ctx, types.NamespacedName{
		Name:      np.Spec.Target.TargetNetwork.NetworkName,
		Namespace: np.ObjectMeta.Namespace,
	}, &tn); err != nil {
		return nil, nil, err
	}
	targetGardenerStatus, err := cloudv1.GardenerNetworkStatusFromNetwork(&tn)
	if err != nil {
		return nil, nil, err
	}
	if provider == cloudv1.CloudProviderGCP {
		if targetGardenerStatus == nil || targetGardenerStatus.GCP.ProjectID == nil || targetGardenerStatus.GCP.NetworkName == nil {
			return nil, nil, fmt.Errorf("Wait target network ready")
		}
	} else if provider == cloudv1.CloudProviderAWS {
		if targetGardenerStatus == nil || targetGardenerStatus.AWS.VpcID == nil {
			return nil, nil, fmt.Errorf("Wait target network ready")
		}
	}

	return &tn, targetGardenerStatus, nil
}

func (cnp *clusterNetworkPeer) addIngressPolicyToNetworkFirewall(projectID string, networkName string, cidr string) error {
	networkURL := networkUrl(projectID, networkName)
	err := cnp.GCPAPI.AttachIngressPolicy(projectID, generateName(6), networkURL, cidr)
	if err != nil {
		if err, ok := err.(*googleapi.Error); ok {
			if err.Code == http.StatusConflict {
				return nil
			}
		}
		return err
	}
	return nil
}

func generateName(length int) string {
	b := make([]byte, length)
	for i := range b {
		b[i] = Charset[seededRand.Intn(len(Charset))]
	}
	return fmt.Sprintf("%s-%s", GCPInternalPeerName, string(b))
}

func networkUrl(projectID, networkName string) string {
	return fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/%s/global/networks/%s", projectID, networkName)
}