/*
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 dwworker

import (
	"context"
	"fmt"
	"time"

	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/kubernetes"
	"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/log"
	"sigs.k8s.io/controller-runtime/pkg/predicate"

	"github.com/tidbcloud/eks-provider/controllers"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	compatk8s "github.com/tidbcloud/infra-provider/pkg/compat/k8s/dataplane"
	"github.com/tidbcloud/infra-provider/pkg/config"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	metricsctrl "github.com/tidbcloud/infra-provider/pkg/metrics/controller"
	"github.com/tidbcloud/infra-provider/pkg/namer"
)

// DWWorkerReconciler reconciles a DWWorker object
type DWWorkerReconciler struct {
	LocalClient client.Client
	LocalCache  client.Reader

	Config        config.RootConfig
	EventRecorder record.EventRecorder

	ClusterProxyAuthDecorator RequestDecorator
}

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

func (r *DWWorkerReconciler) buildReconcileContext(ctx context.Context, req ctrl.Request) (*ReconcileContext, error) {
	rCtx := &ReconcileContext{
		Ctx:                       ctx,
		LocalClient:               r.LocalClient,
		LocalCache:                r.LocalCache,
		Obj:                       &cloudv1.DWWorker{},
		Cluster:                   &cloudv1.Cluster{},
		Config:                    r.Config,
		ClusterProxyAuthDecorator: r.ClusterProxyAuthDecorator,
		Logger:                    log.FromContext(ctx),
	}

	if err := r.LocalCache.Get(ctx, req.NamespacedName, rCtx.Obj); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, nil
		}
		return nil, fmt.Errorf("unable to get dwworker object %s: %v", req.NamespacedName, err)
	}

	// Get referenced cluster
	clusterName := types.NamespacedName{
		Name:      rCtx.Obj.Spec.ClusterName,
		Namespace: rCtx.Obj.Namespace,
	}
	if err := r.LocalCache.Get(ctx, clusterName, rCtx.Cluster); err != nil {
		if apierrors.IsNotFound(err) {
			// Cluster is deleted.
			rCtx.Cluster = nil
		} else {
			return nil, fmt.Errorf("unable to get cluster object %s: %v", clusterName, err)
		}
	}

	// Remote controller client and kube client
	network := &cloudv1.Network{}
	networkKey := types.NamespacedName{}

	if rCtx.Cluster != nil {
		networkKey = rCtx.Cluster.GetNetworkRef()
	} else {
		// These labels are recorded in the webhook. This allows us to access the network
		// when cluster is destroyed.
		networkKey.Name = rCtx.Obj.Labels[cloudv1.DWWorkerLabelNetworkName]
		networkKey.Namespace = rCtx.Obj.Labels[cloudv1.DWWorkerLabelNetworkNamespace]
	}
	if err := r.LocalCache.Get(ctx, networkKey, network); err != nil {
		return nil, fmt.Errorf("unable to get network object %s/%s: %v", networkKey.Namespace, networkKey.Name, err)
	}
	rCtx.Network = network

	providerType := network.Spec.Provider
	if providerType != consts.ProviderTypeEKS && providerType != consts.ProviderTypeGKE {
		return nil, fmt.Errorf("unsupported provider type %s", providerType)
	}

	scheme, err := k8s.TiDBOperatorScheme()
	if err != nil {
		return nil, err
	}

	// Remote Controller Client
	{
		g, err := compatk8s.NetworkClientGetter(rCtx.Ctx, rCtx.LocalClient, network, scheme)
		if err != nil {
			return nil, err
		}

		client := k8s.NewCacheClientGetter(g)
		remoteClient, err := client.GetClient(ctx)
		if err != nil {
			return nil, fmt.Errorf("get data plane client failed: %w", err)
		}

		rCtx.RemoteClient = remoteClient
	}
	// Remote Rest Client
	{
		opts := client.Options{
			Scheme: scheme,
		}
		var g k8s.ClientManager
		switch providerType {
		case consts.ProviderTypeEKS:
			g = compatk8s.NewEKSClientManagerCompat(rCtx.LocalClient, *network, opts)
		case consts.ProviderTypeGKE:
			g = compatk8s.NewGKEClientManagerCompat(rCtx.LocalClient, *network, opts)
		}
		key := types.NamespacedName{
			Name: namer.LegacyBuildClusterNameByNetwork(network),
		}
		cfg, err := g.GetConfig(ctx, key)
		if err != nil {
			return nil, fmt.Errorf("get data plane kubeconfig failed: %w", err)
		}
		restClient, err := kubernetes.NewForConfig(cfg)
		if err != nil {
			return nil, err
		}
		rCtx.RemoteRestClient = restClient
	}

	return rCtx, nil
}

// Reconcile is part of the main kubernetes reconciliation loop which aims to
// move the current state of the cluster closer to the desired state.
func (r *DWWorkerReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	logger := log.FromContext(ctx)

	logger.Info("Reconciling DWWorker")
	start := time.Now()
	defer func() {
		metricsctrl.ReconcileDuration.
			WithLabelValues("dwworker").Observe(time.Since(start).Seconds())
		logger.Info("Finish reconcile", "take", time.Since(start))
	}()

	reconcileCtx, err := r.buildReconcileContext(ctx, req)
	if err != nil {
		return ctrl.Result{}, err
	}
	if reconcileCtx == nil {
		return ctrl.Result{}, nil
	}

	res, err := reconcileCtx.Reconcile()
	if err != nil {
		r.EventRecorder.Event(reconcileCtx.Obj, corev1.EventTypeWarning, "ReconcileFail", err.Error())
	}

	return res, err
}

// SetupWithManager sets up the controller with the Manager.
func (r *DWWorkerReconciler) SetupWithManager(mgr ctrl.Manager, pds ...predicate.Predicate) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&cloudv1.DWWorker{}, builder.WithPredicates(pds...)).
		WithEventFilter(predicate.Or(predicate.GenerationChangedPredicate{}, predicate.AnnotationChangedPredicate{})).
		WithOptions(controller.Options{RateLimiter: k8s.RetryLimiter, MaxConcurrentReconciles: controllers.GetMaxConcurrentReconciles(r.Config, "dwworker")}).
		Complete(r)
}