// Package dm add by blacknc <blacknc@163.com> at 2024/10/31 00:19:09
package dm

import (
	"github.com/pingcap/tiflow/dm/openapi"
	"github.com/tidbcloud/dataflow-service/internal/cluster_job"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
	"github.com/tidbcloud/dataflow-service/internal/handler/agent"
	"github.com/tidbcloud/dataflow-service/internal/utils"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"go.uber.org/zap"
)

type StateFinished struct{}

func (s *StateFinished) String() string {
	return "StateFinished"
}

func (s *StateFinished) Exec(c *TaskContext) (next TaskState, err error) {

	dfJob := c.job.DFJob
	dmJob := c.job.DMJob
	ctx := c.ctx
	logger := log.GetLogger(ctx)
	m := c.model

	jobSources, err := GetDMJobSources(dmJob.ID)
	if err != nil {
		logger.Error("failed get dm job source", zap.Error(err))
		return nil, err
	} else if len(jobSources) == 0 {
		// should not happend
		logger.Error("dm job source is empty, this should not happend!")
		return nil, nil
	}

	dmJobCR, err := m.resourceAPI.GetDMJob(ctx, dmJob.ProjectID.Uint64(), dmJob.ID.Uint64())
	if err != nil && !m.resourceAPI.IsNotFound(err) {
		if dmJobCR != nil && !dmJobCR.DeletionTimestamp.IsZero() {
			logger.Info("dm job CR is executing deletion, waiting for next retry", zap.Time("deletionTime", dmJobCR.DeletionTimestamp.Time))
			err = nil
			return
		}
		logger.Error("failed to get dm job cr", zap.Error(err))
		return
	}

	if dmJobCR != nil {

		// stop job
		dmOpenapiClient := c.agentClient.GetDMOpenapiClient(dmJob.ClusterID.Uint64(), dmJobCR.Status.MasterServerName)
		for _, jobSource := range jobSources {
			stage := m.getDMStage(dmJob, &jobSource)
			sourceName := generateDMDataSourceName(&jobSource)
			taskName := generateDMTaskName(dmJob.ID, stage)
			logger := logger.With(zap.String("taskName", taskName), zap.String("sourceName", sourceName))

			_, err = dmOpenapiClient.QueryTaskStatus(ctx, taskName, openapi.SourceNameList{sourceName})
			if err == agent.ErrorDMResourceNotFound {
				continue
			} else if err != nil {
				logger.Error("failed to query task status", zap.Error(err))
				return
			}

			if err = dmOpenapiClient.StopTask(ctx, taskName, &openapi.StopTaskRequest{
				TimeoutDuration: utils.ToPtr(taskStopTimeout),
				SourceNameList:  &openapi.SourceNameList{sourceName},
			}); err != nil {
				logger.Error("failed to stop dm job task", zap.Error(err))
				return
			}
			logger.Info("successfully stoped dm job task")
		}

		// delte dm job cr
		if err = m.resourceAPI.DeleteDMJob(ctx, dmJob.ProjectID.Uint64(), dmJob.ID.Uint64()); err != nil {
			logger.Error("failed to delete dm job CR", zap.Error(err))
			return
		}

		// we wait until the DM-job CR is deleted completely, so return and enter this method again.
		logger.Info("dm job CR marked for deletion, waiting for next retry")
		return
	}

	if err = cluster_job.DeregisterDMPauseLock(ctx, dmJob.ClusterID.Uint64(), dmJob.ID.Uint64()); err != nil {
		return
	}

	if err = UpdateDFJobPhase(dfJob, entity.DFJobPhaseFinished); err != nil {
		logger.Error("failed to update dm job to finished phase", zap.Error(err))
	} else {
		// success
		logger.Info("successfully update dm job to finished phase")
		dfJob.Phase = entity.DFJobPhaseFinished
		PublishEvent(ctx, &PublishEventReq{Job: c.job})
	}
	return
}