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

import (
	"context"
	"time"

	"github.com/pingcap/tiflow/dm/openapi"
	cluster_job "github.com/tidbcloud/dataflow-service/internal/cluster_job"
	"github.com/tidbcloud/dataflow-service/internal/db"
	"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"
	"github.com/tidbcloud/eks-provider/api/v1alpha1"
	"go.uber.org/zap"
	"gorm.io/gorm"
)

const (
	taskStopTimeout = "30s"
)

type StateDelete struct{}

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

func (s *StateDelete) 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

	m.addPhaseMetric(dfJob, entity.DFJobPhaseDeleting, *dfJob.UpdatedAt)

	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 err = s.finishAllJobModification(ctx, dmJob); err != nil {
		logger.Error("failed to finish all job modification", zap.Error(err))
		return
	}

	// stop dm stask && delte dm job CR
	if dmJobCR != nil {
		switch dmJobCR.Status.Phase {
		case v1alpha1.DMJobPhaseRunning, v1alpha1.DMJobPhaseScaling:
			// just break
		default:
			// delete dmjob CR
			logger.Info("dm job cr is not running, delete it", zap.Reflect("job-cr-phase", dmJobCR.Status.Phase))
			err = m.resourceAPI.DeleteDMJob(ctx, dmJob.ProjectID.Uint64(), dmJob.ID.Uint64())
			if err != nil {
				logger.Error("failed to delete dm job cr", zap.Error(err))
			} else {
				logger.Info("dm job CR marked for deletion, waiting for next retry")
			}
			return
		}

		for _, jobSource := range jobSources {
			// stop job
			stage := m.getDMStage(dmJob, &jobSource)
			dmOpenapiClient := c.agentClient.GetDMOpenapiClient(dmJob.ClusterID.Uint64(), dmJobCR.Status.MasterServerName)
			sourceName := generateDMDataSourceName(&jobSource)
			taskName := generateDMTaskName(dmJob.ID, stage)
			_, err = dmOpenapiClient.QueryTaskStatus(ctx, taskName, openapi.SourceNameList{sourceName})
			if err == nil {
				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.String("taskName", taskName), zap.String("sourceName", sourceName), zap.Error(err))
					// return
				} else {
					logger.Info("succeed to stop dm job task", zap.String("taskName", taskName), zap.String("sourceName", sourceName))
				}
			} else if err != agent.ErrorDMResourceNotFound {
				logger.Error("failed to query task status", zap.String("taskName", taskName), zap.String("sourceName", sourceName), zap.Error(err))
				// return
			}
		}

		// delte dm job cr
		if err = m.resourceAPI.DeleteDMJob(ctx, dmJob.ProjectID.Uint64(), dmJob.ID.Uint64()); err != nil {
			logger.Error("failed to delte 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
	}

	// change dm job phase
	if err = deleteDMJob(dfJob); err != nil {
		logger.Error("failed to delete the dm job in metadb", zap.Error(err))
	} else {
		logger.Info("delete job success")
		m.deletePhaseMetric(logger, entity.DFJobPhaseCreating, dfJob)
		m.deletePhaseMetric(logger, entity.DFJobPhaseDeleting, dfJob)
		m.deletePhaseMetric(logger, entity.DFJobPhaseModifying, dfJob)
		PublishEvent(ctx, &PublishEventReq{Job: c.job})
	}

	return
}

func (s *StateDelete) finishAllJobModification(ctx context.Context, dmJob *entity.DMJob) error {
	logger := log.GetLogger(ctx)
	mods, err := GetUnfinishedJobModification(dmJob.ID)
	if err != nil {
		logger.Error("failed to get unfinished job modification", zap.Error(err))
		return err
	}
	if len(mods) > 1 {
		// should not happen
		logger.Warn("too many modification for a job")
	} else if len(mods) > 0 {
		curTs := time.Now()
		err = db.Transaction(func(tx *gorm.DB) error {
			for _, mod := range mods {
				mod.FinishedAt = &curTs
				if err1 := mod.Update(tx, entity.DMJobModificationsDBSchema.FinishedAt); err != nil {
					logger.Error("failed to finish job modification", zap.Any("modID", mod.ID), zap.Error(err))
					return err1
				}
			}
			return nil
		})
		if err != nil {
			logger.Error("failed to finish job modification", zap.Error(err))
			return err
		}
	}
	return nil
}