/*
Copyright 2022.

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 controllers

import (
	"context"
	"errors"
	"fmt"
	"time"

	cloudv1 "github.com/tidbcloud/operationlet/api/v1"

	v1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/fields"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/runtime"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log"
)

// OperationCommandJobReconciler reconciles a OperationCommandJob object
type OperationCommandJobReconciler struct {
	client.Client
	Scheme *runtime.Scheme
}

const (
	NamespacePrefix = "tidb"
	RunningTimeOut  = 10 * time.Minute
)

//+kubebuilder:rbac:groups=cloud.cloud.pingcap.com,resources=operationcommandjobs,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=cloud.cloud.pingcap.com,resources=operationcommandjobs/status,verbs=get;update;patch
//+kubebuilder:rbac:groups=cloud.cloud.pingcap.com,resources=operationcommandjobs/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 OperationCommandJob 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.8.3/pkg/reconcile
func (r *OperationCommandJobReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	var err error
	logger := log.FromContext(ctx)
	job := &cloudv1.OperationCommandJob{}
	if err := r.Get(ctx, req.NamespacedName, job); err != nil {
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}

	if job.Status.Phase == "" {
		if err := r.transToStatus(ctx, job, cloudv1.JobPhaseWaiting); err != nil {
			return ctrl.Result{}, err
		}
	}
	logger.Info("Watch command job.", "job_type", string(job.Spec.JobType),
		"instance", job.Spec.Instance,
		"cluster_id", job.Spec.ClusterID,
		"tenant_id", job.Spec.TenantID)
	switch job.Status.Phase {
	case cloudv1.JobPhaseWaiting:
		err = r.observeCommandJob(ctx, job)
	case cloudv1.JobPhasePending:
		err = r.runCommandJob(ctx, job)
	case cloudv1.JobPhaseRunning:
		err = r.observeCommandJob(ctx, job)
	}

	return ctrl.Result{Requeue: true}, err
}

// SetupWithManager sets up the controller with the Manager.
func (r *OperationCommandJobReconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&cloudv1.OperationCommandJob{}).
		Complete(r)
}

func (r *OperationCommandJobReconciler) runCommandJob(ctx context.Context, job *cloudv1.OperationCommandJob) error {
	var (
		targetPod v1.PodList
		err       error
	)
	logger := log.FromContext(ctx)
	defer func() {
		if err != nil {
			logger.Error(err, "run job failed", "jobName", job.Name, "cluster_id", job.Spec.ClusterID)
			if err = r.transToStatus(ctx, job, cloudv1.JobPhaseFailed); err != nil {
				logger.Error(err, "update job status failed", "jobName", job.Name, "cluster_id", job.Spec.ClusterID)
			}
		}
	}()
	// When the cluster have job running, we need to pending other job.
	listOptions := &client.ListOptions{
		Namespace:     job.Namespace,
		FieldSelector: fields.OneTermEqualSelector("status.phase", "Running"),
		Limit:         1,
	}
	var lister cloudv1.OperationCommandJobList
	err = r.Client.List(ctx, &lister, listOptions)
	if err != nil {
		logger.Error(err, "Can't list the job!")
		return err
	}
	if len(lister.Items) != 0 {
		logger.Info("This cluster have job running, just pending.",
			"cluster_id", job.Spec.ClusterID,
			"job_type", job.Spec.JobType,
			"job_name", job.Name)
		return nil
	}

	selector, _ := labels.Parse(fmt.Sprintf("app.kubernetes.io/instance=%s", job.Spec.Instance))
	option := &client.ListOptions{
		Namespace:     NamespacePrefix + job.Spec.ClusterID,
		LabelSelector: selector,
		Limit:         1,
	}
	if err = r.Client.List(ctx, &targetPod, option); err != nil || len(targetPod.Items) == 0 {
		logger.Error(err, "Can't find the target instance!")
		return err
	}
	switch job.Spec.JobType {
	case cloudv1.RestartPod:
		r.transToStatus(ctx, job, cloudv1.JobPhaseRunning)
		err = r.Client.Delete(ctx, &targetPod.Items[0])
	}
	if err != nil {
		return err
	}
	return r.transToStatus(ctx, job, cloudv1.JobPhaseSuccess)
}

// observe the newest status of command job
func (r *OperationCommandJobReconciler) observeCommandJob(ctx context.Context, job *cloudv1.OperationCommandJob) error {
	logger := log.FromContext(ctx)
	switch job.Status.Phase {
	case cloudv1.JobPhaseWaiting:
		// observe whether the job have been approved or not.
		logger.Info("job pending and need approve.", "job_name", job.Name,
			"job_type", string(job.Spec.JobType))
	case cloudv1.JobPhaseRunning:
		// observe whether the job have been done or not.
		// and need to consider the case oplet panic when running job.
		// and need to consider overtime.
		logger.Info("Check the job finish or not.", "job_name", job.Name,
			"job_type", string(job.Spec.JobType))
		err := r.checkJobFinishOrNot(ctx, job)
		if err != nil {
			return err
		}
	}
	return nil
}

func (r *OperationCommandJobReconciler) transToStatus(ctx context.Context, job *cloudv1.OperationCommandJob, newStatus cloudv1.CommandJobPhase) error {
	var err error
	logger := log.FromContext(ctx)
	if job.Status.Phase != newStatus {
		logger.Info("command job update status", "jobName", job.Name,
			"oldStatus", string(job.Status.Phase),
			"newStatus", string(newStatus))
		job.Status.Phase = newStatus
		err = r.Status().Update(ctx, job)
	}
	return err
}

// Check the job finish or not.
// This function only be called when job's status = Running
func (r *OperationCommandJobReconciler) checkJobFinishOrNot(ctx context.Context, job *cloudv1.OperationCommandJob) error {
	var err error
	logger := log.FromContext(ctx)
	switch job.Spec.JobType {
	case cloudv1.RestartPod:
		err = r.checkRestartPod(ctx, job)
	}
	if err != nil {
		return err
	}
	if job.Status.Phase == cloudv1.JobPhaseRunning {
		if time.Since(job.CreationTimestamp.Time) > RunningTimeOut {
			logger.Error(errors.New("job is running overtime, failed"), "", "job_name", job.Name,
				"job_type", string(job.Spec.JobType),
				"tenant_id", job.Spec.TenantID,
				"cluster_id", job.Spec.ClusterID)
			return r.transToStatus(ctx, job, cloudv1.JobPhaseFailed)
		}
	}
	return nil
}

func (r *OperationCommandJobReconciler) checkRestartPod(ctx context.Context, job *cloudv1.OperationCommandJob) error {
	var targetPod v1.PodList
	logger := log.FromContext(ctx)
	label, _ := labels.Parse(fmt.Sprintf("app.kubernetes.io/instance=%s", job.Spec.Instance))
	err := r.Client.List(ctx, &targetPod, &client.ListOptions{
		Namespace:     NamespacePrefix + job.Spec.ClusterID,
		LabelSelector: label,
		Limit:         1,
	})
	if err != nil {
		logger.Error(err, "Can't list the pod!")
		return err
	}
	if len(targetPod.Items) == 0 {
		logger.Info("The pod is not exist, job failed.", "job_name", job.Name)
	}

	if targetPod.Items[0].GetCreationTimestamp().After(job.GetCreationTimestamp().Time) {
		logger.Info("job is running successful.", "job_name", job.Name, "job_type", string(job.Spec.JobType))
		return r.transToStatus(ctx, job, cloudv1.JobPhaseSuccess)
	} else {
		logger.Info("job is still running.", "job_name", job.Name, "job_type", string(job.Spec.JobType))
	}
	return nil
}