package task

import (
	"context"
	"time"

	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/types"

	"github.com/chaos-mesh/chaos-mesh/api/v1alpha1"
	eksv1alpha1 "github.com/tidbcloud/eks-provider/api/v1alpha1"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/util"

	"github.com/tidbcloud/chaos-provider/controllers/chaos/utils"
)

type ChaosWorkflowTask struct {
	*BaseTask
}

func NewChaosWorkflowTask(ctx *Context) *ChaosWorkflowTask {
	return &ChaosWorkflowTask{
		BaseTask: &BaseTask{Context: ctx},
	}
}

func (t *ChaosWorkflowTask) Name() string {
	return "Chaos Workflow"
}

func (t *ChaosWorkflowTask) Create(ctx context.Context) deploy.Result {
	t.BeginAt = util.Pointer(time.Now())
	cli, err := t.Remote.GetClient(ctx)
	if err != nil {
		return deploy.Error(err)
	}
	if err := cli.Create(ctx, t.ChaosWorkflow.ConvertToWorkflow()); err != nil && !apierrors.IsAlreadyExists(err) {
		return deploy.Error(util.WrapError("create chaos workflow", err))
	}
	return deploy.Done()
}

func (t *ChaosWorkflowTask) PostCreate(ctx context.Context) deploy.Result {
	cli, err := t.Remote.GetClient(ctx)
	if err != nil {
		return deploy.Error(err)
	}
	workflow := &v1alpha1.Workflow{}
	key := types.NamespacedName{Namespace: t.ChaosWorkflow.Spec.RemoteNamespace, Name: t.ChaosWorkflow.GetName()}
	if err := cli.Get(ctx, key, workflow); err != nil {
		return deploy.Error(util.WrapError("update chaos workflow status", err))
	}
	t.ChaosWorkflow.Status = utils.ConvertToChaosWorkflowStatus(workflow.Status)

	if eksv1alpha1.ChaosWorkflowConditionEqualsTo(t.ChaosWorkflow.Status, v1alpha1.WorkflowConditionAccomplished, corev1.ConditionTrue) {
		// Check whether all chaosWorkflowNode has synced before chaosWorkflow accomplished
		chaosWorkflowNodes, err := utils.FetchEntryNode(ctx, cli, *t.ChaosWorkflow)
		if err != nil {
			return deploy.Panic(err).WithRequeuAfter(0)
		}
		for _, node := range chaosWorkflowNodes {
			if !eksv1alpha1.ChaosWorkflowNodeFinished(node.Status) {
				eksv1alpha1.SetChaosWorkflowCondition(&t.ChaosWorkflow.Status, v1alpha1.WorkflowCondition{
					Type:   v1alpha1.WorkflowConditionAccomplished,
					Status: corev1.ConditionFalse,
					Reason: "",
				})
				return deploy.Error(k8s.RequeueOnlyError).WithRequeuAfter(10 * time.Second)
			}
		}
		return deploy.Done()
	}

	return deploy.Error(k8s.RequeueOnlyError).WithRequeuAfter(10 * time.Second)
}

func (t *ChaosWorkflowTask) Delete(ctx context.Context) deploy.Result {
	cli, err := t.Remote.GetClient(ctx)
	if err != nil {
		deploy.Panic(err)
	}
	return deploy.ErrorOrDone(k8s.DeleteObject(ctx, cli, t.getWorkflowObject()))
}

func (t *ChaosWorkflowTask) PostDelete(ctx context.Context) deploy.Result {
	cli, err := t.Remote.GetClient(ctx)
	if err != nil {
		deploy.Panic(err)
	}
	key := types.NamespacedName{Name: t.ChaosWorkflow.GetName(), Namespace: t.ChaosWorkflow.GetNamespace()}
	return deploy.ErrorOrDone(k8s.GetDeletedObject(ctx, cli, key, t.getWorkflowObject()))
}

func (t *ChaosWorkflowTask) getWorkflowObject() *v1alpha1.Workflow {
	return &v1alpha1.Workflow{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: t.ChaosWorkflow.Spec.RemoteNamespace,
			Name:      t.ChaosWorkflow.GetName(),
		},
	}
}