/*
Copyright 2021.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
    http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package networkconnection

import (
	"context"
	"fmt"
	"time"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/service/ec2"
	ec2v1beta1 "github.com/crossplane-contrib/provider-aws/apis/ec2/v1beta1"
	"github.com/go-logr/logr"
	"go.uber.org/multierr"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/tools/record"
	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/predicate"

	cloudv1alpha1 "github.com/tidbcloud/eks-provider/api/v1alpha1"
	eksv1alpha1 "github.com/tidbcloud/eks-provider/api/v1alpha1"
	"github.com/tidbcloud/eks-provider/controllers"
	"github.com/tidbcloud/eks-provider/controllers/networkconnection/task"
	"github.com/tidbcloud/eks-provider/controllers/utils"
	utilplane "github.com/tidbcloud/eks-provider/controllers/utils/dataplane"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	compatcst "github.com/tidbcloud/infra-provider/pkg/compat/consts"
	compatk8s "github.com/tidbcloud/infra-provider/pkg/compat/k8s"
	"github.com/tidbcloud/infra-provider/pkg/config"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/controllers/common"
	"github.com/tidbcloud/infra-provider/pkg/dataplane"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	metricsctrl "github.com/tidbcloud/infra-provider/pkg/metrics/controller"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	"github.com/tidbcloud/infra-provider/pkg/procedure"
)

// PrivateLinkEndpointReconciler reconciles a PrivateLinkEndpoint object
type PrivateLinkEndpointReconciler struct {
	Logger        logr.Logger
	Cache         client.Reader
	Config        config.RootConfig
	EventRecorder record.EventRecorder
	AWSSVC        procedure.AWSService
	Manager       dataplane.K8sClientManager

	client.Client
	Scheme *runtime.Scheme
}

//+kubebuilder:rbac:groups=eks.infra.pingcap.com,resources=privatelinks,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=eks.infra.pingcap.com,resources=privatelinks/status,verbs=get;update;patch
//+kubebuilder:rbac:groups=eks.infra.pingcap.com,resources=privatelinks/finalizers,verbs=update
//+kubebuilder:rbac:groups=ec2.aws.crossplane.io,resources=vpcendpoints,verbs=list;watch;get;patch;create;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.
//
// For more details, check Reconcile and its Result here:
// - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.12.2/pkg/reconcile
func (r *PrivateLinkEndpointReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	log := r.Logger.WithValues("object", req.NamespacedName)
	log.Info("Reconciling PrivateLink Endpoint")

	start := time.Now()
	defer func() {
		metricsctrl.ReconcileDuration.
			WithLabelValues("private_link_endpoint").Observe(time.Since(start).Seconds())
		log.Info("Finish PrivateLink Endpoint reconcile", "take", time.Since(start))
	}()

	plEndpoint := &eksv1alpha1.PrivateLinkEndpoint{}
	if err := r.Cache.Get(ctx, req.NamespacedName, plEndpoint); err != nil {
		if apierrors.IsNotFound(err) {
			return ctrl.Result{}, nil
		}
		return ctrl.Result{}, fmt.Errorf("unable to get private link endpoint %s: %v", req.NamespacedName, err)
	}

	c, showError, err := r.buildDeployContext(ctx, plEndpoint)
	if err != nil {
		if showError {
			plEndpoint.Status.Phase = eksv1alpha1.PrivateLinkEndpointPhaseError
			plEndpoint.Status.ErrorMessage = err.Error()
		}
		log.Error(err, "build private endpoint context error")
		err := updatePLEndpoint(ctx, r.Client, plEndpoint)
		if err != nil {
			return ctrl.Result{}, err
		}
		return ctrl.Result{RequeueAfter: time.Minute * 1}, nil
	}

	tasks := r.buildTasks(c)
	d := deploy.NewDeployer(tasks, nil, log)

	var res ctrl.Result
	if plEndpoint.DeletionTimestamp.IsZero() {
		res, err = updateCreatePLEndpoint(ctx, c, d.Deploy(ctx))
	} else {
		res, err = updateDeletePLEndpoint(ctx, c, d.Destroy(ctx))
	}

	if err == nil {
		r.EventRecorder.Event(plEndpoint, corev1.EventTypeNormal, "ReconcileSuccess", "")
	} else {
		r.EventRecorder.Event(plEndpoint, corev1.EventTypeWarning, "ReconcileFail", err.Error())
	}
	return res, err
}

func (r *PrivateLinkEndpointReconciler) buildDeployContext(
	ctx context.Context, plEndpoint *eksv1alpha1.PrivateLinkEndpoint) (*task.Context, bool, error) {

	// aws only
	var (
		network     *cloudv1.Network
		networkSpec *eksv1alpha1.EKSNetworkSpec
		subnets     []*ec2v1beta1.Subnet
		err         error
	)

	r.Logger.Info("Reconciling PrivateLink Endpoint")

	// azure only
	var kubernetes *cloudv1.Kubernetes

	if r.Config.Provider == consts.ProviderTypeEKS {
		network, networkSpec, err = r.queryNetworkInfo(ctx, plEndpoint.Namespace, plEndpoint)
		if err != nil {
			return nil, false, err
		}

		uniqueIdentifier := namer.UniqueIdentifier(network.ObjectMeta)
		_, ok := network.Annotations[compatcst.CompatFlagAnnotation]
		subnets, err = r.querySubnets(ctx, uniqueIdentifier, plEndpoint.Spec.Zones, network, ok)
		if err != nil {
			return nil, false, err
		}

		err = r.queryEndpointAvailabilityZones(plEndpoint, networkSpec)
		if err != nil {
			return nil, true, err
		}
	} else if r.Config.Provider == consts.ProviderTypeAKS {
		projectID, err1 := utils.GetProjectIDFromLabel(plEndpoint)
		tenantID, err2 := utils.GetTenantIDFromLabel(plEndpoint)
		err := multierr.Combine(err1, err2)
		if err != nil {
			return nil, false, err
		}
		dataplaneProvider, err := utilplane.NewDataplaneProvider(ctx, r.Logger, tenantID, projectID, "", &r.Config)
		if err != nil {
			return nil, false, fmt.Errorf("failed to get dataplane provider impl, projectID:%s, err:%v", projectID, err)
		}
		kubernetes = dataplaneProvider.GetKubernetes().Kubernetes
	}

	c := (&task.Context{
		Local:       r.Client,
		Config:      r.Config,
		Network:     network,
		NetworkSpec: networkSpec,
		Kubernetes:  kubernetes,
		Subnets:     subnets,
		PLEndpoint:  plEndpoint,
		Logger:      r.Logger,
	}).Default()
	return c, false, nil
}

func (r *PrivateLinkEndpointReconciler) queryEndpointAvailabilityZones(
	plEndpoint *eksv1alpha1.PrivateLinkEndpoint, networkSpec *eksv1alpha1.EKSNetworkSpec) error {
	if len(plEndpoint.Spec.Zones) > 0 {
		return nil
	}
	// Get the availability zone of the endpoint service
	endpointServiceOut, err := r.AWSSVC.EC2API(&procedure.AssumedRole{
		RoleARN:    networkSpec.AssumeRoleArn,
		ExternalID: networkSpec.AssumeExternalId,
	}).DescribeVpcEndpointServices(&ec2.DescribeVpcEndpointServicesInput{
		ServiceNames: aws.StringSlice([]string{plEndpoint.Spec.ServiceName}),
	})
	if err != nil {
		return fmt.Errorf("failed to describe endpoint services %s: %v", plEndpoint.Spec.ServiceName, err)
	}
	if len(endpointServiceOut.ServiceDetails) != 1 || len(endpointServiceOut.ServiceDetails[0].AvailabilityZones) == 0 {
		return fmt.Errorf("service name is %s, no common az between endpoint service and current dataplane", plEndpoint.Spec.ServiceName)
	}
	for _, zone := range endpointServiceOut.ServiceDetails[0].AvailabilityZones {
		plEndpoint.Spec.Zones = append(plEndpoint.Spec.Zones, *zone)
	}
	return nil
}

func (r *PrivateLinkEndpointReconciler) querySubnets(ctx context.Context, vpcName string, zones []string, network *cloudv1.Network, compat bool) ([]*ec2v1beta1.Subnet, error) {
	var subnets []*ec2v1beta1.Subnet
	if compat {
		compatSubnets, err := compatk8s.GetPrivateSubnetCompat(ctx, r.Cache, network, zones)
		if err != nil {
			return nil, fmt.Errorf("unable to get subnet in compat mode: %v", err)
		}
		for _, subnet := range compatSubnets {
			subnets = append(subnets, &ec2v1beta1.Subnet{
				Spec: ec2v1beta1.SubnetSpec{
					ForProvider: ec2v1beta1.SubnetParameters{
						AvailabilityZone: aws.String(subnet.AvailabilityZone),
					},
				},
				Status: ec2v1beta1.SubnetStatus{
					AtProvider: ec2v1beta1.SubnetObservation{
						SubnetID: subnet.SubnetId,
					},
				},
			})
		}
	} else {
		for _, zone := range zones {
			subnetName := namer.PrivateSubnetName(vpcName, zone)
			subnet := &ec2v1beta1.Subnet{}
			err := r.Client.Get(ctx, types.NamespacedName{Name: subnetName}, subnet)
			if err != nil {
				return nil, fmt.Errorf("unable to get subnet %s: %v", subnetName, err)
			}
			subnets = append(subnets, subnet)
		}
	}
	return subnets, nil
}

func (r *PrivateLinkEndpointReconciler) queryNetworkInfo(ctx context.Context, namespace string,
	plEndpoint *eksv1alpha1.PrivateLinkEndpoint) (*cloudv1.Network, *eksv1alpha1.EKSNetworkSpec, error) {
	network := &cloudv1.Network{}
	networkKey := types.NamespacedName{Namespace: namespace, Name: plEndpoint.Spec.NetworkName}
	if err := r.Cache.Get(ctx, networkKey, network); err != nil {
		return nil, nil, fmt.Errorf("unable to get network %s: %v", plEndpoint.Spec.NetworkName, err)
	}
	providerType := network.Spec.Provider
	if providerType != consts.ProviderTypeEKS && providerType != consts.ProviderTypeAKS {
		return nil, nil, fmt.Errorf("the provider type %s of network %s is invaild", plEndpoint.Spec.NetworkName, providerType)
	}
	networkSpec, err := eksv1alpha1.EKSNetworkSpecForNetwork(network)
	if err != nil {
		return nil, nil, fmt.Errorf("unable to get network spec: %v", err)
	}
	return network, networkSpec, nil
}

// SetupWithManager sets up the controller with the Manager.
func (r *PrivateLinkEndpointReconciler) SetupWithManager(mgr ctrl.Manager, p ...predicate.Predicate) error {

	if err := mgr.GetFieldIndexer().IndexField(context.TODO(), &eksv1alpha1.PrivateLinkEndpoint{}, "spec.k8snetwork", func(obj client.Object) []string {
		plObj := obj.(*eksv1alpha1.PrivateLinkEndpoint)
		return []string{plObj.K8sNetworkName()}
	}); err != nil {
		return err
	}

	return ctrl.NewControllerManagedBy(mgr).
		For(&cloudv1alpha1.PrivateLinkEndpoint{}, builder.WithPredicates(p...)).
		WithOptions(controller.Options{RateLimiter: common.DefaultRateLimiter(), MaxConcurrentReconciles: controllers.GetMaxConcurrentReconciles(r.Config, "privateLinkEndpoint")}).
		Complete(r)
}

func (r *PrivateLinkEndpointReconciler) buildTasks(c *task.Context) []deploy.Task {
	switch r.Config.Provider {
	case consts.ProviderTypeAKS:
		return []deploy.Task{
			task.NewGuardian(c),
			task.NewProviderConfig(c),
			task.NewPLEndpoint(c),
		}
	case consts.ProviderTypeEKS:
		return []deploy.Task{
			task.NewGuardian(c),
			task.NewPLEndpoint(c),
		}
	}
	return nil
}