// Package common add by blacknc <blacknc@163.com> at 2024/08/29 10:21:38
package common

import (
	"context"
	"fmt"
	"time"

	"github.com/juju/errors"
	"go.opentelemetry.io/otel/attribute"
	batchv1 "k8s.io/api/batch/v1"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/kubernetes"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/tidbcloud/eks-provider/controllers/imports/task"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	tracerutil "github.com/tidbcloud/infra-provider/pkg/tracer"
	"github.com/tidbcloud/infra-provider/util/config"
)

const (
	defaultLoopInterval = 15 * time.Second
)

type Progress struct {
	*task.BaseTask
}

func (t *Progress) Name() string {
	return "Progress"
}

func NewProgress(ctx *task.Context) *Progress {
	return &Progress{
		BaseTask: &task.BaseTask{Context: ctx},
	}
}

func (t *Progress) Create(ctx context.Context) (res deploy.Result) {
	ctx, span := task.Tracer.Start(ctx, "progress")
	defer func() { tracerutil.EndSpan(span, res.Error()) }()

	span.SetAttributes(
		attribute.String("cluster_name", t.Import.Spec.ClusterName),
		attribute.String("import_name", t.Import.Name),
	)
	name := t.Import.Name
	ns := t.GetClusterNameSpace()

	logger := t.Logger.WithName(name).WithValues("namespace", ns, "import name", t.Import.Name)

	clusterCli := t.EnvProvider.GetClient()
	lightningPods := &corev1.PodList{}
	if err := clusterCli.List(ctx, lightningPods,
		client.MatchingLabels{"job-name": t.Import.Name},
		client.InNamespace(ns)); err != nil {
		return deploy.Panic(fmt.Errorf("faild to get lightning pods, requeue..., err:%v", err))
	}
	if len(lightningPods.Items) == 0 {
		return deploy.Panic(errors.New("lightning pod not up, requeue after 15 seconds...")).WithRequeuAfter(defaultLoopInterval)
	}
	if lightningPods.Items[0].Status.Phase == "Pending" {
		return deploy.Panic(errors.New("lightning pod phase is pending, requeue after 15 seconds...")).WithRequeuAfter(defaultLoopInterval)
	}
	logger.Info("lightning pod is up", "pod name", lightningPods.Items[0].Name)

	cliSet, err := kubernetes.NewForConfig(t.EnvProvider.GetRestConfig())
	if err != nil {
		return deploy.Panic(errors.Annotate(err, "failed to dataplane k8s rest config"))
	}
	link := k8s.NewHttpLinkByKubeProxy(ctx, cliSet, t.GetClusterNameSpace(), t.Import.Name, "8289")

	lightningClient := newLightningClient(link)
	logger.Info("start request to lightning")
	task, err := lightningClient.GetTaskProgress()
	if err != nil {
		// todo:error count
		logger.Error(err, "failed to get import task progress, setStatusFailedOrCompleteByJob and requeue...")
		// This is expected to fail when the job just started.
		// return nil would also requeue but without exponential backoff

		// but when job failed, we should update import's status by job status
		if setStatusErr := t.setStatusFailedOrCompleteByJob(ctx); setStatusErr != nil {
			logger.Error(setStatusErr, "failed setStatusFailedOrCompleteByJob requeue...")
			return deploy.Panic(setStatusErr)
		}
		logger.Info("setStatusFailedOrCompleteByJob complete, requeue", "status", t.Import.Status.Phase, "message", t.Import.Status.Message)
		return deploy.Panic(errors.Annotate(err, "failed to get task progress"))
	}

	// todo:lightning is online,reset
	logger.Info("successfully get task progress", "task status", task.Status, "task message", task.Message, "task table count", len(task.Tables))
	// if task is not started, create task
	if task.Status == taskStatusNotStarted && !t.Import.Status.IsTaskCreated {
		logger.Info("task not started, create task")
		taskId, err := lightningClient.CreateTask(string(t.LightningConfig))
		if err != nil {
			// if create task failed,termination
			t.Import.Status.Phase = cloudv1.ImportFailed
			t.Import.Status.Message = err.Error()
			logger.Error(err, "task create failed")
			return deploy.Panic(err)
		}
		logger.Info("task created", "task id", taskId)
		t.Import.Status.IsTaskCreated = true
		t.Import.Status.TaskId = taskId
	}

	lightningBackend := t.getLightningBackend()
	saveImportStatus(&(t.Import.Status), task, lightningBackend, logger)
	logger.Info("import detail info",
		"import phase", t.Import.Status.Phase,
		"import message", t.Import.Status.Message,
		"task id", t.Import.Status.TaskId,
		"task phase", t.Import.Status.TaskPhase,
		"total tables", t.Import.Status.TotalTables,
		"pendding tables", t.Import.Status.PendingTables,
		"current tables", len(t.Import.Status.CurrentTables),
		"completed tables", t.Import.Status.CompletedTables,
		"completed percent", t.Import.Status.CompletedPercent,
		"post import completed percent", t.Import.Status.PostImportCompletedPercent,
		"total data size", t.Import.Status.TotalSize,
		"process data size", t.Import.Status.ProcessedDataSize,
		"total files", t.Import.Status.TotalFiles,
	)
	if t.Import.Status.Phase == cloudv1.ImportFailed {
		logger.Error(errors.New("task failed"), "task failed", "task status", task.Status, "task message", task.Message)
	}

	if err := t.setStatusFailedOrCompleteByJob(ctx); err != nil {
		return deploy.Panic(err)
	}
	logger.Info("import status update complete, reconcile againe latter.", "import phase", t.Import.Status.Phase, "import message", t.Import.Status.Message)
	return deploy.Done().WithRequeuAfter(defaultLoopInterval)
}

func (t *Progress) PostCreate(ctx context.Context) deploy.Result {
	return deploy.Done()
}

func (t *Progress) setStatusFailedOrCompleteByJob(ctx context.Context) error {
	var job batchv1.Job
	i := t.Import
	if err := t.EnvProvider.GetClient().Get(ctx, types.NamespacedName{
		Name:      i.ObjectMeta.Name,
		Namespace: t.GetClusterNameSpace(),
	}, &job); err != nil {
		return err
	}

	for _, cond := range job.Status.Conditions {
		if cond.Status != corev1.ConditionTrue {
			continue
		}

		if cond.Type == batchv1.JobFailed {
			t.Logger.Info("lightning job failed", "job name", job.Name, "job condition message", cond.Message, "job condition reason", cond.Reason)
			i.Status.Phase = cloudv1.ImportFailed
			if i.Status.Message == "" {
				i.Status.Message = "abnormal termination, please seek support"
			}
		} else if cond.Type == batchv1.JobComplete {
			i.Status.Phase = cloudv1.ImportCompleted
		}
	}
	return nil
}

func (t *Progress) getLightningBackend() string {
	defaultBackend := "local"
	cfg := config.New(make(map[string]interface{}))
	if err := cfg.UnmarshalTOML(t.LightningConfig); err != nil {
		t.Logger.Error(err, "Unmarshal config string into TOML failed")
		return defaultBackend
	}
	v := cfg.Get("tikv-importer.backend")
	if v == nil {
		return defaultBackend
	}
	lightningBackendStr, err := v.AsString()
	if err != nil {
		return defaultBackend
	}
	return lightningBackendStr
}