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

import (
	"strconv"

	"github.com/pingcap/tiflow/pkg/errors"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
	infraapi "github.com/tidbcloud/dataflow-service/pkg/infra-api"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/eks-provider/api/v1alpha1"
	v1 "github.com/tidbcloud/infra-api/api/v1"
	"go.uber.org/zap"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

type StateDeploy struct{}

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

func (s *StateDeploy) 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.DFJobPhaseCreating, dfJob.CreatedAt)

	jobSpec, err := GetJobSpecByID(dmJob.SpecID)
	if err != nil {
		logger.Error("failed to get job spec", zap.Int64("spec-id", int64(dmJob.SpecID)), zap.Error(err))
		return
	}

	dmJobCR, err := m.resourceAPI.GetDMJob(ctx, dmJob.ProjectID.Uint64(), dmJob.ID.Uint64())
	if err != nil {
		if !m.resourceAPI.IsNotFound(err) {
			logger.Error("failed to get dm job cr", zap.Error(err))
			return
		}
		dmJobCR = generateDMJobCR(c, jobSpec.Name)
		err = m.resourceAPI.CreateDMJobIgnoreExists(ctx, dmJobCR)
		if err != nil {
			logger.Error("failed to create dm job cr", zap.Error(err))
		} else if cerr := UpdateDMJobComponentVersionWithImage(dmJob, dmJobCR.Spec.Worker.BaseImage); cerr != nil {
			logger.Error("failed to update dm job component version", zap.Error(cerr))
		}
		return
	}

	switch dmJobCR.Status.Phase {
	case v1alpha1.DMJobPhaseRunning:
		break
	case "", v1alpha1.DMJobPhaseCreating, v1alpha1.DMJobPhasePending:
		logger.Info("dm job CR is still creating", zap.Reflect("dmJobCRPhase", dmJobCR.Status.Phase))
		return
	default:
		// if the job is being scaling, whose previous phase is creating,
		// then there must be something wrong with the code.
		// Jobs can only be scaled after being create.
		// should not reach here normally
		logger.Error("dm job CR in unexpected phase", zap.Reflect("dmJobCR", dmJobCR))
		err = errors.Errorf("dm job CR in unexpected phase: %s", dmJobCR.Status.Phase)
		return
	}

	if dmJobCR.Status.MasterServerName == "" {
		logger.Error("dm job CR MasterServerName can not be empty", zap.Reflect("dmJobCR", dmJobCR))
		err = errors.New("dm job CR MasterServerName can not be empty")
		return
	}

	if err = UpdateDFJobPhase(dfJob, entity.DFJobPhaseCreated); err != nil {
		logger.Error("failed to update job phase", zap.Error(err))
		return
	}
	dfJob.Phase = entity.DFJobPhaseCreated

	logger.Info("dm job CR is ready")
	m.deletePhaseMetric(logger, entity.DFJobPhaseCreating, dfJob)
	next = new(StateDump)
	PublishEvent(ctx, &PublishEventReq{Job: c.job})
	return
}

func generateDMJobCR(c *TaskContext, specCode string) *v1alpha1.DMJob {

	dmImage := getDMImage(c)

	return &v1alpha1.DMJob{
		ObjectMeta: metav1.ObjectMeta{
			Name:      getDMJobName(c.job.DFJob.ID.Uint64()),
			Namespace: c.model.resourceAPI.GetObjectNamespace(c.job.DFJob.ProjectID.Uint64()),
			Labels: map[string]string{
				v1.ProviderTypeLabel: c.job.DFJob.Provider,
				v1.RegionLabel:       c.job.DFJob.Region,
				v1.TenantTagKey:      strconv.FormatUint(c.job.DFJob.TenantID.Uint64(), 10),
				v1.ProjectTagKey:     strconv.FormatUint(c.job.DFJob.ProjectID.Uint64(), 10),
				v1.ClusterTagKey:     strconv.FormatUint(c.job.DFJob.ClusterID.Uint64(), 10),
			},
			Annotations: map[string]string{
				"tags.tidbcloud.com/dm-import-mode":          string(c.job.DMJob.ImportMode),
				// AnnoKeyEnableClusterIDLabel enables cluster ID label on DM master/worker pods.
				// Matches dm/task/common.AnnoKeyEnableClusterIDLabel in infra-provider.
				"dm.cloud.pingcap.com/enable-cluster-id-label": "true",
			},
		},
		Spec: v1alpha1.DMJobSpec{
			ClusterName: infraapi.ClusterName(c.job.DFJob.ClusterID.Uint64()),
			Master: v1alpha1.DMClusterConf{
				BaseImage: dmImage,
			},
			Worker: v1alpha1.DMClusterConf{
				SpecCode:  specCode,
				BaseImage: dmImage,
			},
		},
	}
}