// Package dm add by blacknc <blacknc@163.com> at 2024/11/27 17:47:36
package dm

import (
	"fmt"
	"strings"
	"time"

	"github.com/pingcap/errors"
	dmconfig "github.com/pingcap/tiflow/dm/config"
	"github.com/pingcap/tiflow/dm/openapi"
	"github.com/tidbcloud/config-center/proto/config_center"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/internal/config"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
	"github.com/tidbcloud/dataflow-service/internal/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service/internal/handler/agent"
	"github.com/tidbcloud/dataflow-service/internal/utils"
	infraapi "github.com/tidbcloud/dataflow-service/pkg/infra-api"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/eks-provider/api/v1alpha1"
	"github.com/tidbcloud/pkgv2/bizerr"
	"go.uber.org/zap"
)

const (
	DMTaskUintDump = "Dump"
	DMTaskUintLoad = "Load"
	DMTaskUintSync = "Sync"

	DefaultDMImage = "gcr.io/pingcap-public/dbaas/dm:v8.5.0-alpha"
)

type StateBase struct{}

func (s *StateBase) createDMTask(
	c *TaskContext,
	dmJobCR *v1alpha1.DMJob,
	taskMode openapi.TaskTaskMode,
	taskName string,
	dmOpenapiClient *agent.DMOpenapiClient,
	jobSources []entity.DMJobSource,
) (*openapi.OperateTaskResponse, error) {

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

	binlogFilterRuleName := "rule_1"

	cluster, cerr := grpcli.GetClusterWithDataplaneByID(ctx, dmJob.ClusterID.Uint64())
	if cerr != nil {
		return nil, errors.Annotatef(cerr, "failed to get cluster info:%d", dmJob.ClusterID)
	}

	// build dm data source and table migrate rule config
	taskSourceConf, taskMigrateRules, err := m.buildDMDataSource(ctx, dmOpenapiClient, jobSources, cluster, binlogFilterRuleName, taskMode, dmJob.Mode)
	if err != nil {
		return nil, err
	}

	// build dm target db connection
	targetDB, err := m.buildTargetDB(ctx, c.agentClient, dfJob, dmJob, taskMode)
	if err != nil {
		return nil, err
	}

	clientCA, clientCert, clientKey, err := m.getTiDBClientCA(ctx, c.agentClient, dfJob)
	if err != nil {
		logger.Error("failed to get cluster client secret", zap.Error(err))
		return nil, err
	}

	dataDir := s.genDataDir(c, dmJobCR)
	sortDir := s.genSortDir()
	// suggested by jiaqiang.huang@pingcap.com
	// taskShareMode := openapi.TaskShardModePessimistic
	importMode := openapi.TaskFullMigrateConfImportMode(dmJob.ImportMode)
	taskRequest := &openapi.CreateTaskRequest{
		Task: openapi.Task{
			BinlogFilterRule: &openapi.Task_BinlogFilterRule{
				AdditionalProperties: map[string]openapi.TaskBinLogFilterRule{
					binlogFilterRuleName: {
						IgnoreEvent: &[]string{"create database"},
					},
				},
			},
			// whether to enable support for the online ddl plugin
			EnhanceOnlineSchemaChange: dmJob.OnlineDdl,
			// ignore precheck items
			IgnoreCheckingItems: &[]string{dmconfig.AllChecking},
			// downstream database for storing meta information
			MetaSchema: utils.ToPtr("tidbcloud_dm_meta"),
			// task name
			Name: taskName,
			// how to handle conflicted data
			OnDuplicate: openapi.TaskOnDuplicateReplace,
			// the way to coordinate DDL
			// ShardMode: &taskShareMode,
			// source-related configuration
			SourceConfig: openapi.TaskSourceConfig{
				FullMigrateConf: &openapi.TaskFullMigrateConf{
					DataDir:    utils.ToPtr(dataDir),
					SortingDir: utils.ToPtr(sortDir),
					ImportMode: &importMode,
					PdAddr:     utils.ToPtr(fmt.Sprintf("db-pd.tidb%d.svc:2379", dmJob.ClusterID.Uint64())),
					Security: &openapi.Security{
						SslCaContent:   clientCA,
						SslCertContent: clientCert,
						SslKeyContent:  clientKey,
					},
				},
				IncrMigrateConf: &openapi.TaskIncrMigrateConf{},
				SourceConf:      taskSourceConf,
			},
			// table migrate rule
			TableMigrateRule: taskMigrateRules,
			// downstream database configuration
			TargetConfig: *targetDB,
			// migrate mode
			TaskMode: taskMode,
		},
	}

	if c.job.DMJob.ImportMode == entity.DMImportModePhysical {
		// In physical mode, a default disk of 15TiB is mounted,
		// and the maximum limit is set slightly lower than the disk size
		// DiskQuota = 15T - 1G
		taskRequest.Task.SourceConfig.FullMigrateConf.DiskQuota = utils.ToPtr("15359g")
	}

	// speed up by revising threads when user specifies a higher speed
	if dmJob.ImportMode == entity.DMImportModePhysical && dmJob.Mode == entity.DMJobModeAll {
		jobSpec, err := GetJobSpecByID(dmJob.SpecID)
		if err != nil {
			logger.Error("failed to get dm job spec", zap.Error(err))
			return nil, err
		}
		threads := int(jobSpec.CapacityUnit)
		taskRequest.Task.SourceConfig.FullMigrateConf.ExportThreads = utils.ToPtr(2 * threads)
		taskRequest.Task.SourceConfig.FullMigrateConf.ImportThreads = utils.ToPtr(2 * threads)
		taskRequest.Task.SourceConfig.IncrMigrateConf.ReplThreads = utils.ToPtr(2 * threads)
	}

	// set dm task config
	dmTaskConfig := config.Global().DMConfig.DMTaskConfigs.GetMatchedConfig(
		dmJob.TenantID.Uint64(), dmJob.ProjectID.Uint64(), dmJob.ClusterID.Uint64())
	if dmTaskConfig != nil {
		if dmTaskConfig.WorkerCount != nil {
			taskRequest.Task.SourceConfig.IncrMigrateConf.ReplThreads = dmTaskConfig.WorkerCount
		}
		if dmTaskConfig.Batch != nil {
			taskRequest.Task.SourceConfig.IncrMigrateConf.ReplBatch = dmTaskConfig.Batch
		}
		if dmTaskConfig.OnlineDDL != nil {
			taskRequest.Task.EnhanceOnlineSchemaChange = *dmTaskConfig.OnlineDDL
		}
	}

	return dmOpenapiClient.CreateTask(ctx, taskRequest)
}

func (s *StateBase) checkSubTaskStatus(c *TaskContext, dmOpenapiClient *agent.DMOpenapiClient, stage entity.DMStageType, taskStatus []openapi.SubTaskStatus) (bool, error) {

	ctx := c.ctx
	m := c.model
	logger := log.GetLogger(ctx).With(zap.String("stage", string(stage)))

	if len(taskStatus) == 0 {
		logger.Warn("dm job subtask is empty")
		return false, nil
	}

	// check if the dm subtask has been completed
	// wait all job source dump stage finished
	allFinished := true
	var err error
	for _, subTaskStatus := range taskStatus {
		logger := logger.With(zap.String("taskName", subTaskStatus.Name), zap.String("sourceName", subTaskStatus.SourceName))
		switch subTaskStatus.Stage {
		case openapi.TaskStageFinished:
			allFinished = allFinished && true
		case openapi.TaskStageRunning:
			allFinished = false
			logger.Info("dm job task is running, waiting for the next retry.")
		case openapi.TaskStageStopped:
			// TODO check weather is user stop task
			allFinished = false
			logger.Warn("dm job task has been stopped!")
		case openapi.TaskStagePaused:
			allFinished = false
			if subTaskStatus.ErrorMsg != nil {
				logger.Info("dm job task has error!", zap.Stringp("errorMsg", subTaskStatus.ErrorMsg))
				m.recoverTask(ctx, dmOpenapiClient, subTaskStatus.Name, openapi.SourceNameList{subTaskStatus.SourceName})
			} else {
				logger.Info("dm job task has been paused!")
			}
		default:
			allFinished = false
			logger.Error("dm job task return unexpected stage", zap.Any("subTaskStage", subTaskStatus.Stage))
			err = errors.Errorf("unexpect dm job task stage:%s", subTaskStatus.Stage)
		}
	}
	return allFinished, err
}

func (s *StateBase) genDataDir(c *TaskContext, dmJobCR *v1alpha1.DMJob) string {
	// // TODO just for dev test
	// return fmt.Sprintf("/tmp/dmjob/%d", c.job.DMJob.ID.Uint64())

	container := dmJobCR.Status.BlobContainerName
	accountName := dmJobCR.Status.StorageAccountName
	accountKey := dmJobCR.Status.StorageAccountKey
	switch c.job.DFJob.Provider {
	case infraapi.ProviderTypeAKS:
		return fmt.Sprintf("azure://%s/dmjob-%d?account-name=%s&account-key=%s",
			container, c.job.DMJob.ID.Uint64(), accountName, accountKey)
	case infraapi.ProviderTypeGKE:
		return fmt.Sprintf("gcs://%s/dmjob-%d", container, c.job.DMJob.ID.Uint64())
	case infraapi.ProviderTypeEKS:
		return fmt.Sprintf("s3://%s/dmjob-%d", container, c.job.DMJob.ID.Uint64())
	}
	return ""
}

func (s *StateBase) genSortDir() string {
	// The PV of the worker is mounted in the /var/lib/dm-worker directory，
	// We set the sort directory to this directory for sorting during physical import
	return "/var/lib/dm-worker/sortdata_" + time.Now().Format("20060102150405") + "_" + utils.UniqueID()
}

func (s *StateBase) reconcileDMSpec(c *TaskContext, dmOpenapiClient *agent.DMOpenapiClient, dmJobCR *v1alpha1.DMJob, toSpec, dataSourceName string) (bool, error) {

	ctx := c.ctx
	m := c.model
	logger := log.GetLogger(ctx)

	dmImage := dmJobCR.Spec.Worker.BaseImage
	dmConfig, err := getDMVersionConfig(c)
	if err != nil {
		// Do not make any changes when obtaining DM image failure
		logger.Warn("dm version will not be changed, failed to get dm version config", zap.Error(err))
	} else if dmConfig.UpdateStrategy != config_center.UpdateStrategy_UPDATE_ALL {
		logger.Info("dm version will not be changed, update strategy does not take effect on the running DM", zap.Any("updateStrategy", dmConfig.UpdateStrategy))
	} else {
		dmImage = fmt.Sprintf("%s:%s", dmConfig.BaseImage, dmConfig.Tag)
	}

	switch dmJobCR.Status.Phase {
	case v1alpha1.DMJobPhaseRunning:
		// dmJobCR.Spec.Master.BaseImage = dmImage
		// dmJobCR.Spec.Worker.BaseImage = dmImage
		// dmJobCR.Spec.Worker.SpecCode = toSpec

		var modifyFields []zap.Field
		if toSpec != dmJobCR.Spec.Worker.SpecCode {
			fromSpec := dmJobCR.Spec.Worker.SpecCode
			dmJobCR.Spec.Worker.SpecCode = toSpec
			modifyFields = append(modifyFields, zap.String("fromSpec", fromSpec), zap.String("topSpec", toSpec))
		}
		if dmImage != dmJobCR.Spec.Worker.BaseImage {
			fromImage := dmJobCR.Spec.Worker.BaseImage
			dmJobCR.Spec.Master.BaseImage = dmImage
			dmJobCR.Spec.Worker.BaseImage = dmImage
			modifyFields = append(modifyFields, zap.String("fromImage", fromImage), zap.String("topImage", dmImage))
		}
		if len(modifyFields) > 0 {
			if err := m.resourceAPI.UpdateDMJob(ctx, dmJobCR); err != nil {
				return false, err
			}
			logger.Info("update dm job cr", modifyFields...)
		} else if dmJobCR.Spec.Worker.SpecCode == dmJobCR.Status.WorkerSpecCode &&
			getImageTag(dmJobCR.Spec.Worker.BaseImage) == dmJobCR.Status.DMImageTag {
			// 1. dmJobCR.Phase == running
			// 2. dmJobCR.Spec.Worker.SpecCode == dmJobCR.Status.WorkerSpecCode
			//    (ensure the dmJobCR has been reconciled after
			//    we change the dmJobCR.Spec.Worker.SpecCode)
			// 3. dmJobCR has job detail in dm master
			//    (ensure the dmJobCR has already been scheduled on dm master)
			dataSource, err := dmOpenapiClient.GetDataSource(ctx, dataSourceName)
			if err != nil {
				return false, err
			}
			hasReady := dataSource.StatusList != nil && len(*dataSource.StatusList) > 0
			if !hasReady {
				logger.Info("dm job data source is not ready when scaling, wait next times", zap.String("dataSourceName", dataSourceName))
			} else {
				logger.Info("dm job data source is ready when scaling, scaling completed", zap.String("dataSourceName", dataSourceName))
			}
			return hasReady, nil
		}
	case v1alpha1.DMJobPhaseScaling:
		logger.Info("the dm job is scaling.",
			zap.Any("dmJobID", c.job.DMJob.ID),
			zap.String("fromSpec", dmJobCR.Spec.Worker.SpecCode),
			zap.String("topSpec", dmJobCR.Status.WorkerSpecCode),
			zap.String("fromImage", dmJobCR.Spec.Worker.BaseImage),
			zap.String("topImageTag", dmJobCR.Status.DMImageTag),
		)
	case v1alpha1.DMJobPhaseDeleting:
		// should not happen normally
		// The modification won't be applied and should be marked as finished.
		logger.Info("the dm job has been deleted in other routine", zap.Any("dmJobID", c.job.DMJob.ID))
	}
	return false, nil
}

func getDMImage(c *TaskContext) string {
	var dmImage string
	dmConfig, err := getDMVersionConfig(c)
	// the error log has been printed inside the function
	if err == nil {
		dmImage = fmt.Sprintf("%s:%s", dmConfig.BaseImage, dmConfig.Tag)
	} else {
		dmImage = DefaultDMImage
	}
	return dmImage
}

func getDMVersionConfig(c *TaskContext) (_ *domain.ConfigVersionInfo, err error) {

	ctx := c.ctx
	configCenter := c.model.configCenter
	logger := log.GetLogger(c.ctx)

	defer func() {
		if err != nil {
			logger.Warn("failed to get dm version config", zap.Error(err))
		}
	}()

	cluster, err := grpcli.GetClusterByID(ctx, c.job.DMJob.ClusterID.Uint64())
	if err != nil {
		return nil, err
	}

	getConfigReq := &domain.GetConfigVersionReq{
		ClusterID: cluster.Id,
		Component: domain.ConfigComponentDataMigration,
	}
	conf, bizErr := configCenter.GetConfigVersion(ctx, getConfigReq)
	if bizErr != nil && bizErr.ErrCode() != bizerr.ErrCodeRecordNotFound {
		return nil, errors.Annotate(bizErr, "failed to get exists dm version config")
	}
	if bizErr != nil {

		// get default version
		defaultVersionReq := &domain.GetDefaultVersionReq{
			Component:       domain.ConfigComponentDataMigration,
			ClusterVersion:  cluster.Version,
			ClusterProvider: cluster.InfraProviderType, // TODO(ideascf): should use cloud provider ???
			Region:          cluster.RegionName,
			TenantID:        cluster.TenantId,
		}
		defaultVersion, bizErr := configCenter.GetComponentDefaultConfigVersion(ctx, defaultVersionReq)
		if bizErr != nil {
			logger.Error("failed to get default dm version config", zap.Any("defaultVersionReq", defaultVersionReq), zap.Error(bizErr))
			return nil, errors.Annotate(bizErr, "failed to get default dm version config")
		}
		logger.Info("get default dm version config", zap.Any("defaultVersionReq", defaultVersionReq), zap.Any("defaultVersionResp", defaultVersion))

		initConfigReq := &domain.InitConfigVersionReq{
			ClusterID:     cluster.Id,
			TenantID:      cluster.TenantId,
			Component:     domain.ConfigComponentDataMigration,
			Version:       defaultVersion.Version,
			CloudProvider: domain.CloudProviderType(cluster.CloudProviderName),
		}
		conf, bizErr = configCenter.GetAndInitConfigVersion(ctx, initConfigReq)
		if bizErr != nil {
			logger.Error("failed to init and get dm version config", zap.Any("initConfigReq", initConfigReq), zap.Error(bizErr))
			return nil, errors.Annotate(bizErr, "failed init and get dm version config")
		}
		logger.Info("init and get dm version config", zap.Any("initConfigReq", initConfigReq), zap.Any("configVersion", conf))
	}
	return conf, nil
}

func getImageTag(image string) string {
	var tag string
	if pos := strings.Index(image, ":"); pos != -1 {
		tag = image[pos+1:]
	} else {
		tag = image
	}
	return tag
}