// Package dm add by blacknc <blacknc@163.com> at 2024/10/30 22:56:42
package dm

import (
	"context"
	"runtime"
	"time"

	"github.com/pingcap/errors"
	"github.com/tidbcloud/dataflow-service/internal/common"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	pkglog "github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/logid"
	"github.com/tidbcloud/pkgv2/tracing"
	"go.uber.org/zap"
)

type Job struct {
	DFJob *entity.DFJob
	DMJob *entity.DMJob
}

func runDMJob(ctx context.Context, logger *zap.Logger, dfJob *entity.DFJob, model Model) {

	var err error
	ctx, span := Tracer.Start(ctx, "dm.RunJob")
	defer func() { tracing.EndSpan(span, err) }()

	logger = logger.With(pkglog.FieldsFromContext(ctx)...)
	logger = logger.With(
		zap.Any("runID", logid.GenLogid()),
		zap.Any("dfJobName", dfJob.Name),
		zap.Any("dfJobID", dfJob.ID),
		zap.Any("clusterID", dfJob.ClusterID),
		zap.Any("projectID", dfJob.ProjectID),
		zap.Any("providerType", dfJob.Provider),
		zap.String("initPhase", string(dfJob.Phase)),
	)
	ctx = log.WithLogger(ctx, logger)
	logger.Info("[worker_v2]begin run dm job")

	if dfJob.WorkerVersion != entity.DMWorkerVersionV2 {
		logger.Error("[worker_v2]unexpected dm worker version, skip running dm job", zap.String("workerVersion", dfJob.WorkerVersion))
		return
	}

	resourceAPI, err := NewRegionalResourceAPI(ctx, dfJob.Provider, dfJob.Region, dfJob.TenantID.Uint64())
	if err != nil {
		logger.Error("failed to get resource api", zap.Error(err))
		return
	}
	model.resourceAPI = resourceAPI

	agentClient, err := common.GetAgentClientByNetwork(ctx, dfJob.ClusterID.Uint64(), nil)
	if err != nil {
		logger.Error("failed to get agent client", zap.Error(err))
		return
	}

	dmJob, err := GetDMJobByID(dfJob.ID)
	if err != nil {
		logger.Error("failed to get dm job", zap.Error(err))
		return
	}

	taskContext := &TaskContext{
		ctx: ctx,
		job: &Job{
			DFJob: dfJob,
			DMJob: dmJob,
		},
		model:       model,
		agentClient: agentClient,
	}
	startTime := time.Now()

	defer func() {
		cost := time.Since(startTime)
		if rerr := recover(); rerr != nil {
			buf := make([]byte, 1<<16)
			runtime.Stack(buf, true)
			err = errors.New("run dm job encountered a panic: " + string(buf))
		}
		if err != nil {
			logger.Error("failed to run dm job", zap.Duration("cost", cost), zap.Any("stateStack", taskContext.stateStack), zap.Error(err))
		} else {
			logger.Info("succeed to run dm job", zap.Duration("cost", cost), zap.Any("stateStack", taskContext.stateStack), zap.Error(err))
		}
	}()

	var state TaskState
	switch dfJob.Phase {
	case entity.DFJobPhaseCreating:
		state = &StateDeploy{}
	case entity.DFJobPhaseCreated:
		var jobSources []entity.DMJobSource
		jobSources, err = GetDMJobSources(dmJob.ID)
		if err != nil {
			logger.Error("failed get dm job source", zap.Error(err))
			return
		} else if len(jobSources) == 0 {
			// should not happend
			logger.Error("dm job source is empty, this should not happend!")
			return
		}
		for _, jobSource := range jobSources {
			stage := model.getDMStage(dmJob, &jobSource)
			switch stage {
			case entity.DMStageDump:
				state = &StateDump{}
			case entity.DMStageLoad:
				state = &StateLoad{}
			case entity.DMStageSync:
				state = &StateSync{}
			}
		}
	case entity.DFJobPhaseModifying:
		state = &StateScale{}
	case entity.DFJobPhaseFinishing:
		state = &StateFinished{}
	case entity.DFJobPhaseDeleting:
		state = &StateDelete{}
	}

	if state == nil {
		err = errors.Errorf("nothing to do for dm job with phase:%s", dfJob.Phase)
		return
	}

	err = walk(taskContext, state)
}

func walk(c *TaskContext, s TaskState) error {

	var prevState, curState TaskState
	logger := log.GetLogger(c.ctx)
	prevState, curState = new(StateBing), s
	for curState != nil {
		logger.Info("dm job trans to new state begin", zap.String("fromState", prevState.String()), zap.String("toState", curState.String()))

		insights := &TaskStateInsights{
			state:     curState,
			beginTime: time.Now(),
		}
		c.stateStack = append(c.stateStack, insights)
		nextState, err := curState.Exec(c)
		insights.cost = time.Since(insights.beginTime)
		insights.err = err

		if err != nil {
			logger.Error("dm job trans to new state completed with error", zap.String("state", curState.String()), zap.String("insights", insights.String()), zap.Error(err))
			if uerr := UpdateDFJobErrorMessage(c.job.DFJob, err.Error()); uerr != nil {
				logger.Error("failed to set df job error message", zap.Error(uerr))
			}
			return err
		} else if c.job.DFJob.ErrorMessage != "" {
			if uerr := UpdateDFJobErrorMessage(c.job.DFJob, ""); uerr != nil {
				logger.Error("failed to clear df job error message", zap.Error(uerr))
			}
			return err
		}
		logger.Info("dm job trans to new state completed with succeed", zap.String("state", curState.String()), zap.String("insights", insights.String()))
		prevState, curState = curState, nextState
	}
	return nil
}