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

import (
	"context"
	"fmt"
	"time"

	"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/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/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/predicate"

	eksv1alpha1 "github.com/tidbcloud/eks-provider/api/v1alpha1"
	"github.com/tidbcloud/eks-provider/controllers"
	"github.com/tidbcloud/eks-provider/controllers/dm/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"
	"github.com/tidbcloud/infra-provider/pkg/cloud/gcp"
	"github.com/tidbcloud/infra-provider/pkg/config"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/dataplane"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/metrics"
)

const (
	DMMasterMemberType = "dm-master"
	DMWorkerMemberType = "dm-worker"

	dmMaxConcurrentReconcilesConfigKey = "dm"
)

// DMJobReconciler reconciles a DMJob object
type DMJobReconciler struct {
	client.Client
	Logger        logr.Logger
	EventRecorder record.EventRecorder
	Manager       dataplane.K8sClientManager
	Cache         client.Reader
	Config        config.RootConfig
	GCP           gcp.Client
}

//+kubebuilder:rbac:groups=eks.infra.pingcap.com,resources=tiflowjobs,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=eks.infra.pingcap.com,resources=tiflowjobs/status,verbs=get;update;patch
//+kubebuilder:rbac:groups=eks.infra.pingcap.com,resources=tiflowjobs/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.
//
// 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 *DMJobReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	ctx, cancel := context.WithTimeout(ctx, time.Minute)
	defer cancel()
	logger := log.FromContext(ctx)

	logger.Info("Reconciling DM Job")
	start := time.Now()
	defer func() {
		metrics.ReconcileDuration.
			WithLabelValues("dm_job", r.Config.Provider, r.Config.Network.Region).Observe(float64(time.Since(start).Milliseconds()))
		logger.Info("Finish reconcile", "take", time.Since(start))
	}()

	dmJobCR := &eksv1alpha1.DMJob{}
	if err := r.Cache.Get(ctx, req.NamespacedName, dmJobCR); err != nil {
		if apierrors.IsNotFound(err) {
			return ctrl.Result{}, nil
		}
		metrics.DMFailCounter.WithLabelValues(r.Config.Provider, r.Config.Network.Region, "GetDMJobCR").Inc()
		return ctrl.Result{}, fmt.Errorf("unable to get dm job %s: %v", req.NamespacedName, err)
	}

	providerType := dmJobCR.GetProviderType()
	if !isProviderSupported(providerType) {
		logger.Info("unsupported providerr type", "dm_job", req.NamespacedName, "provider_type", providerType)
		return ctrl.Result{}, nil
	} else if r.Config.Predicate.MatchLabels[cloudv1.ProviderTypeLabel] != providerType {
		logger.Info("provider type not match", "dm_job", req.NamespacedName, "provider_type", providerType)
		return ctrl.Result{}, nil
	}

	c, err := r.buildContext(ctx, dmJobCR, logger, req)
	if err != nil {
		logger.Error(err, "unable to build context for dm job", "name", req.NamespacedName)
		return ctrl.Result{}, err
	}

	tasks := getDMJobTasks(r, c)
	d := deploy.NewDeployer(tasks, nil, logger)
	var res ctrl.Result
	if dmJobCR.DeletionTimestamp.IsZero() && c.IsDataplaneExist() && c.IsClusterExist() {
		res, err = r.updateCreate(ctx, c, d.Deploy(ctx))
	} else {
		res, err = r.updateDelete(ctx, c, d.Destroy(ctx))
	}

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

func (r *DMJobReconciler) buildContext(ctx context.Context, dmJobCR *eksv1alpha1.DMJob, logger logr.Logger, req ctrl.Request) (*task.Context, error) {

	clusterID, err1 := utils.GetClusterIDFromLabel(dmJobCR)
	projectID, err2 := utils.GetProjectIDFromLabel(dmJobCR)
	tenantID, err3 := utils.GetTenantIDFromLabel(dmJobCR)
	err := multierr.Combine(err1, err2, err3)
	if err != nil {
		metrics.DMFailCounter.WithLabelValues(r.Config.Provider, r.Config.Network.Region, "ParseTenantClusterID").Inc()
		return nil, err
	}

	dataplaneProvider, err := utilplane.NewDataplaneProvider(ctx, logger, tenantID, projectID, clusterID, &r.Config)
	if err != nil {
		metrics.DMFailCounter.WithLabelValues(r.Config.Provider, r.Config.Network.Region, "NewDataplaneProvider").Inc()
		return nil, fmt.Errorf("failed to get dataplane provider impl, dm_job:%s, cluster:%s, err:%v", req.NamespacedName, dmJobCR.Spec.ClusterName, err)
	}
	kubernetesInfo := dataplaneProvider.GetKubernetes()
	cluster := dataplaneProvider.GetCluster()

	var ep dataplane.EnvProvider
	if dataplaneProvider.IsDataplaneExist() {
		var dataplaneKubeData []byte
		if kubernetesInfo.KubeConfigContent != "" {
			if r.Config.Provider == consts.ProviderTypeAKS {
				var err error
				dataplaneKubeData, err = dataplane.OverrideAksKubeconfigWithBase64([]byte(kubernetesInfo.KubeConfigContent))
				if err != nil {
					r.EventRecorder.Eventf(dmJobCR, corev1.EventTypeWarning, "CannotDecodeKubeConfig", "cannot base64 decode kubeconfig, dm_job:%s, err:%v", req.NamespacedName, err)
					metrics.DMFailCounter.WithLabelValues(r.Config.Provider, r.Config.Network.Region, "OverrideAksKubeconfig").Inc()
					return nil, fmt.Errorf("cannot base64 decode kubeconfig: %w", err)
				}
				logger.Info("get dataplane kubeconfig from appbox KubeConfigContent", "kubeconfig_len", len(dataplaneKubeData))
			} else {
				dataplaneKubeData = []byte(kubernetesInfo.KubeConfigContent)
			}
		} else if ref := kubernetesInfo.KubeConfigSecretRef; ref != nil {
			var err error
			if dataplaneKubeData, err = dataplane.GetDataplaneKubeConfigFromMetaSecret(ctx, r.Client, ref); err != nil {
				r.EventRecorder.Eventf(dmJobCR, corev1.EventTypeWarning, "CannotGetKubeConfig", "cannot get kubeconfig, dm_job:%s, err:%v", req.NamespacedName, err)
				metrics.DMFailCounter.WithLabelValues(r.Config.Provider, r.Config.Network.Region, "GetKubeconfigFromSecret").Inc()
				return nil, fmt.Errorf("cannot get kubeconfig: %w", err)
			}
			logger.Info("get dataplane kubeconfig from appbox KubeConfigSecretRef", "kubeconfig_len", len(dataplaneKubeData))
		}

		ep, err = dataplane.NewWithKubenets(r.Manager,
			clusterID,
			kubernetesInfo.Kubernetes,
			r.Config.Network.Region,
			r.Config.FullManage.AssumeRoleName,
			r.Config.FullManage.AssumeRoleExternalID,
			dataplaneKubeData)
		if err != nil {
			r.EventRecorder.Eventf(dmJobCR, corev1.EventTypeWarning, "CannotNewEnvProvider", "cannot new env provier: %v", err)
			metrics.DMFailCounter.WithLabelValues(r.Config.Provider, r.Config.Network.Region, "NewEnvProvider").Inc()
			return nil, fmt.Errorf("cannot new env provider: %w", err)
		}
	} else {
		ep = &dataplane.NopEnvProvider{}
	}

	masterSpecCode, workerSpecCode := dmJobCR.Spec.Master.SpecCode, dmJobCR.Spec.Worker.SpecCode
	if masterSpecCode == "" {
		masterSpecCode = r.Config.DMConfig.Master.SpecCode
	}
	if workerSpecCode == "" {
		workerSpecCode = r.Config.DMConfig.Worker.SpecCode
	}

	dmMasterSpec, err := r.getDMSpecByCode(ctx, DMMasterMemberType, masterSpecCode)
	if err != nil {
		return nil, fmt.Errorf("check master spec code failed, specCode:%s, err:%v", masterSpecCode, err)
	}

	dmWorkerSpec, err := r.getDMSpecByCode(ctx, DMWorkerMemberType, workerSpecCode)
	if err != nil {
		return nil, fmt.Errorf("check worker spec code failed, specCode:%s, err:%v", workerSpecCode, err)
	}

	c, err := (&task.Context{
		EnvProvider:   ep,
		Local:         r.Client,
		EventRecorder: r.EventRecorder,
		Config:        r.Config,
		TenantID:      tenantID,
		ProjectID:     projectID,
		ClusterID:     clusterID,
		Cluster:       cluster,
		ClusterRole:   dataplaneProvider.GetClusterRole(),
		Kubernetes:    kubernetesInfo.Kubernetes,
		DMJob:         dmJobCR,
		DMMasterSpec:  dmMasterSpec,
		DMWorkerSpec:  dmWorkerSpec,
		Logger:        logger,
		GCP:           r.GCP,
	}).Default()
	if err != nil {
		metrics.DMFailCounter.WithLabelValues(r.Config.Provider, r.Config.Network.Region, "InitContext").Inc()
		return nil, err
	}

	logger.Info("dmjob buildContext completed", "masterSpec", dmMasterSpec, "workerSpec", dmWorkerSpec)
	return c, nil
}

func (r *DMJobReconciler) updateCreate(ctx context.Context, t *task.Context, results []deploy.Result) (res ctrl.Result, err error) {
	res, err, retriableErrs := deploy.MinRequeue(results)
	switch phase := t.DMJob.Status.Phase; phase {
	case eksv1alpha1.DMJobPhaseCreating:
		if err == nil && retriableErrs == nil {
			log.FromContext(ctx).Info("mark dm job become running when creating job",
				"sinceCreate", time.Since(t.DMJob.CreationTimestamp.Time))
			t.DMJob.Status.Phase = eksv1alpha1.DMJobPhaseRunning
		}
	default:
	}
	if err := r.update(ctx, t.Local, t.DMJob); err != nil {
		return res, err
	}
	return
}

func (r *DMJobReconciler) updateDelete(ctx context.Context, t *task.Context, results []deploy.Result) (res ctrl.Result, err error) {
	// If the cluster is pending delete, then the deletion tasks are not actually performed
	res, err, retriableErrs := deploy.MinRequeue(results)
	switch phase := t.DMJob.Status.Phase; phase {
	case eksv1alpha1.DMJobPhaseDeleting:
		if err == nil && retriableErrs == nil {
			controllerutil.RemoveFinalizer(t.DMJob, task.Finalizer)
		}
	default:
	}
	if err := r.update(ctx, t.Local, t.DMJob); err != nil {
		return res, err
	}
	return
}

func (r *DMJobReconciler) update(ctx context.Context, cli client.Client, dmJob *eksv1alpha1.DMJob) error {
	dmJobStatus := dmJob.Status.DeepCopy()
	if err := cli.Update(ctx, dmJob); err != nil {
		return err
	}
	if err := cli.Get(ctx, types.NamespacedName{Name: dmJob.Name, Namespace: dmJob.Namespace}, dmJob); err != nil {
		return err
	}
	dmJob.Status = *dmJobStatus
	return cli.Status().Update(ctx, dmJob)
}

// SetupWithManager sets up the controller with the Manager.
func (r *DMJobReconciler) SetupWithManager(mgr ctrl.Manager, p ...predicate.Predicate) error {
	if err := mgr.GetFieldIndexer().IndexField(context.Background(), &eksv1alpha1.DMJob{}, "spec.clusterName", func(rawObj client.Object) []string {
		dmJob := rawObj.(*eksv1alpha1.DMJob)
		return []string{dmJob.Spec.ClusterName}
	}); err != nil {
		return err
	}

	return ctrl.NewControllerManagedBy(mgr).
		For(&eksv1alpha1.DMJob{}, builder.WithPredicates(p...)).
		WithOptions(controller.Options{RateLimiter: k8s.RetryLimiter, MaxConcurrentReconciles: getDMMaxConcurrentReconciles(r.Config)}).
		WithEventFilter(utils.GetMigrationAnnotationSkipPredicator()).
		Complete(r)

}

func getDMMaxConcurrentReconciles(cfg config.RootConfig) int {
	return controllers.GetMaxConcurrentReconciles(cfg, dmMaxConcurrentReconcilesConfigKey)
}

func isProviderSupported(provider string) bool {
	switch provider {
	case
		consts.ProviderTypeEKS,
		consts.ProviderTypeGKE,
		consts.ProviderTypeAKS:
		return true
	default:
		return false
	}
}

func (r *DMJobReconciler) queryDMProvider(ctx context.Context) (*eksv1alpha1.DMProvider, error) {
	provider, region := r.Config.Provider, r.Config.Network.Region
	dmProvider := &eksv1alpha1.DMProvider{}
	dmProviderName := fmt.Sprintf("dm-provider-%s-%s", provider, region)
	dmProviderKey := types.NamespacedName{Name: dmProviderName}
	if err := r.Cache.Get(ctx, dmProviderKey, dmProvider); err != nil {
		return nil, fmt.Errorf("unable to get dm provider %s: %v", dmProviderName, err)
	}
	return dmProvider, nil
}

func (r *DMJobReconciler) getDMSpecByCode(ctx context.Context, memberType string, specCode string) (*eksv1alpha1.DMSpec, error) {
	dmProvider, err := r.queryDMProvider(ctx)
	if err != nil {
		return nil, err
	}

	if memberType == DMMasterMemberType {
		if spec, ok := dmProvider.Spec.MasterSpec[specCode]; ok {
			return &spec, nil
		}
	} else if memberType == DMWorkerMemberType {
		if spec, ok := dmProvider.Spec.WorkerSpec[specCode]; ok {
			return &spec, nil
		}
	}
	return nil, fmt.Errorf("can't find dm provider specification for specCode:%s", specCode)
}