// Package dm add by blacknc <blacknc@163.com> at 2024/08/28 00:28:23
package dm

import (
	"context"
	"fmt"
	"runtime/debug"
	"strconv"
	"time"

	"github.com/tidbcloud/eks-provider/controllers/dm/task"
	akstask "github.com/tidbcloud/eks-provider/controllers/dm/task/aks"
	"github.com/tidbcloud/eks-provider/controllers/dm/task/common"
	ekstask "github.com/tidbcloud/eks-provider/controllers/dm/task/eks"
	gketask "github.com/tidbcloud/eks-provider/controllers/dm/task/gke"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/metrics"
)

func getDMJobTasks(r *DMJobReconciler, c *task.Context) []deploy.Task {
	switch c.Config.Provider {
	case consts.ProviderTypeAKS:
		return GetAKSClusterTasks(r, c)
	case consts.ProviderTypeGKE:
		return GetGKEClusterTasks(r, c)
	case consts.ProviderTypeEKS:
		return GetEKSClusterTasks(r, c)
	}

	return nil
}

func GetAKSClusterTasks(r *DMJobReconciler, c *task.Context) (tasks []deploy.Task) {
	tasks = []deploy.Task{
		akstask.NewGuardian(c),
		akstask.NewProviderConfig(c), // provider config
		akstask.NewStorageAccount(c), // storage account create/update/delete
		akstask.NewBlobContainer(c),  // create a new bucket for dm job
		akstask.NewMasterNodePool(c), // node pool create/update/delete
		akstask.NewWorkerNodePool(c), // node pool create/update/delete
		common.NewDMPhase(c),         // status check: scaling/running phase
		common.NewDMNamespace(c),     // dm cluster namespace
		common.NewDMCluster(c),       // dm cluster create/update/delete
		akstask.NewNodePoolGC(c),     // node pool gc
		common.NewPVCGC(c),           // pvc gc
	}
	for i := range tasks {
		tasks[i] = metricWrap(c, tasks[i])
	}
	return tasks
}

func GetGKEClusterTasks(r *DMJobReconciler, c *task.Context) (tasks []deploy.Task) {
	tasks = []deploy.Task{
		gketask.NewGuardian(c),
		gketask.NewProviderConfig(c), // provider config
		gketask.NewBucket(c),         // create a new bucket for dm job
		gketask.NewMasterNodePool(c), // node pool create/update/delete
		gketask.NewWorkerNodePool(c), // node pool create/update/delete
		common.NewDMPhase(c),         // status check: scaling/running phase
		common.NewDMNamespace(c),     // dm cluster namespace
		common.NewDMCluster(c),       // dm cluster create/update/delete
		gketask.NewNodePoolGC(c),     // node pool gc
		common.NewPVCGC(c),           // pvc gc
	}
	for i := range tasks {
		tasks[i] = metricWrap(c, tasks[i])
	}
	return tasks
}

func GetEKSClusterTasks(r *DMJobReconciler, c *task.Context) (tasks []deploy.Task) {
	tasks = []deploy.Task{
		ekstask.NewGuardian(c),
		ekstask.NewProviderConfig(c), // provider config
		ekstask.NewNodeRole(c),       // create a new iam role
		ekstask.NewBucket(c),         // create a new bucket for dm job
		ekstask.NewMasterNodePool(c), // node pool create/update/delete
		ekstask.NewWorkerNodePool(c), // node pool create/update/delete
		common.NewDMPhase(c),         // status check: scaling/running phase
		common.NewDMNamespace(c),     // dm cluster namespace
		common.NewDMCluster(c),       // dm cluster create/update/delete
		ekstask.NewNodePoolGC(c),     // node pool gc
		common.NewPVCGC(c),           // pvc gc
	}
	for i := range tasks {
		tasks[i] = metricWrap(c, tasks[i])
	}
	return tasks
}

func metricWrap(c *task.Context, t deploy.Task) deploy.Task {
	return &metricTask{c, t}
}

type metricTask struct {
	ctx *task.Context
	deploy.Task
}

func (t *metricTask) Create(ctx context.Context) (result deploy.Result) {
	return t.do(ctx, t.Task.Name(), "create", t.Task.Create)
}
func (t *metricTask) PostCreate(ctx context.Context) deploy.Result {
	return t.do(ctx, t.Task.Name(), "postCreate", t.Task.PostCreate)
}
func (t *metricTask) Delete(ctx context.Context) deploy.Result {
	return t.do(ctx, t.Task.Name(), "delete", t.Task.Delete)
}
func (t *metricTask) PostDelete(ctx context.Context) deploy.Result {
	return t.do(ctx, t.Task.Name(), "postDelete", t.Task.PostDelete)
}

func (t *metricTask) do(ctx context.Context, taskName, step string, f func(context.Context) deploy.Result) (result deploy.Result) {
	startTime := time.Now()
	t.ctx.Logger.Info("metricTask begin", "name", taskName, "step", step)
	defer func() {
		if err := recover(); err != nil {
			stack := debug.Stack()
			result = deploy.Panic(fmt.Errorf("task panic:%v\n%s", err, stack))
		}
		var ok bool
		// Done or Other requeue result is considered normal, for example:
		// OK: deploy.Done(), deploy.Panic(nil), , deploy.Panic(err).WithRequeuAfter(xx)
		// Not OK: deploy.Panic(err), deploy.Error(err)
		if result.IsDoneState() || result.RawError() == nil {
			ok = true
		} else if res := k8s.ShouldRequeue(result.RawError()); !res.IsZero() {
			ok = true
		}
		cost := time.Since(startTime).Milliseconds()
		metrics.ControllerTaskDuration.WithLabelValues(
			"dm-job",
			t.ctx.Config.Provider,
			t.ctx.GetRegion(),
			t.Task.Name(), step, strconv.FormatBool(ok)).Observe(float64(cost))
		t.ctx.Logger.Info("metricTask completed", "name", taskName, "step", step, "cost", cost, "ok", ok)
	}()
	result = f(ctx)
	return
}