package premium_dm

//nolint:gci
import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"maps"
	"net/url"
	"slices"
	"strconv"
	"strings"
	"time"

	"github.com/AlekSi/pointer"
	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/hashicorp/go-multierror"
	"github.com/pingcap/log"
	dmopenapi "github.com/pingcap/tiflow/dm/openapi"
	"github.com/tidbcloud/dataflow-service-ng/app/event_bus/publisher/center"
	"github.com/tidbcloud/dataflow-service-ng/app/internal/feature_gate"
	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
	"github.com/tidbcloud/dataflow-service-ng/app/models/sink_endpoint"
	"github.com/tidbcloud/dataflow-service-ng/app/repo"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	"github.com/tidbcloud/dataflow-service-ng/internal/config"
	"github.com/tidbcloud/dataflow-service-ng/pkg/agent"
	encryptor_wrapper "github.com/tidbcloud/dataflow-service-ng/pkg/encryptor-wrapper"
	"github.com/tidbcloud/dataflow-service-ng/pkg/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service-ng/pkg/id_generator"
	"github.com/tidbcloud/eventbus-datastruct/event"
	"github.com/tidbcloud/pkgv2/bizerr"
	nextgenv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	v1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/common/v1beta1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.opentelemetry.io/otel/trace"
	"go.uber.org/zap"
)

func GetWorkerLoggerFromContext(ctx context.Context) *zap.Logger {
	return log.L().With(zap.String("trace_id", trace.SpanContextFromContext(ctx).TraceID().String()))
}

// buildCreateSourceSecurity prepares the DM OpenAPI TLS payload for source creation.
// DM's CertAllowedCn field is a *[]string with omitempty; passing a non-nil pointer
// to a nil slice marshals as JSON null, which DM rejects for create-source requests.
// When the CN list is empty we intentionally leave the pointer nil so the field is omitted.
func buildCreateSourceSecurity(sec *entity.PremiumSecurity) *dmopenapi.Security {
	if sec == nil {
		return nil
	}

	security := &dmopenapi.Security{
		SslCaContent:   string(sec.SslCaContent),
		SslCertContent: string(sec.SslCertContent),
		SslKeyContent:  string(sec.SslKeyContent),
	}
	if len(sec.CertAllowedCn) > 0 {
		certAllowedCn := slices.Clone(sec.CertAllowedCn)
		security.CertAllowedCn = &certAllowedCn
	}
	return security
}

const MySQLAccessDeniedErrorString = "Error 1045 (28000)"

// errAzureCredentials is returned when neither an account key nor a managed-identity
// client-id is present in the shared-pool storage config for an Azure cluster.
var errAzureCredentials = errors.New("azure internal storage credentials are empty")

func buildSharedPoolDataDir(
	task *entity.PremiumDMTask,
	storage *agent.SharedPool_ObjectStorage,
	provider *agent.Provider,
) (string, error) {
	if storage == nil {
		return "", errors.New("shared pool internal storage is empty")
	}
	bucket := strings.TrimSpace(storage.Bucket)
	if bucket == "" {
		return "", errors.New("shared pool internal storage bucket is empty")
	}
	prefix := strings.Trim(storage.Prefix, "/")
	objectPath := fmt.Sprintf("%s/%s", task.TiDBID, task.ID)
	if prefix != "" {
		objectPath = fmt.Sprintf("%s/%s", prefix, objectPath)
	}

	switch strings.ToLower(strings.TrimSpace(task.CloudProvider)) {
	case string(entity.CloudProviderTypeAzure):
		accountName := strings.TrimSpace(storage.AccountName)
		container := bucket
		if accountName == "" {
			decodedAccountName, decodedContainer, decodeErr := parseAzureSharedPoolBucket(bucket)
			if decodeErr != nil {
				return "", decodeErr
			}
			accountName = decodedAccountName
			container = decodedContainer
		}
		query := url.Values{}
		query.Set("account-name", accountName)
		if provider != nil {
			if clientID := strings.TrimSpace(provider.NodeClientID); clientID != "" {
				query.Set("client-id", clientID)
			}
		}
		if accountKey := strings.TrimSpace(storage.AccountKey); accountKey != "" {
			query.Set("account-key", accountKey)
		} else if query.Get("client-id") == "" {
			return "", errAzureCredentials
		}
		return fmt.Sprintf("azure://%s/%s?%s", container, objectPath, query.Encode()), nil
	case string(entity.CloudProviderTypeGCP):
		return fmt.Sprintf("gcs://%s/%s", bucket, objectPath), nil
	case "", string(entity.CloudProviderTypeAWS), string(entity.CloudProviderTypeAliCloud):
		return fmt.Sprintf("s3://%s/%s", bucket, objectPath), nil
	default:
		return "", fmt.Errorf("unsupported cloud provider %s", task.CloudProvider)
	}
}

const (
	azureSharedPoolBucketParts = 2
)

func parseAzureSharedPoolBucket(bucket string) (string, string, error) {
	parts := strings.SplitN(strings.TrimSpace(bucket), "/", azureSharedPoolBucketParts)
	if len(parts) != 2 || strings.TrimSpace(parts[0]) == "" {
		return "", "", errors.New("azure internal storage account name is empty")
	}
	if strings.TrimSpace(parts[1]) == "" {
		return "", "", errors.New("azure internal storage container is empty")
	}
	return strings.TrimSpace(parts[0]), strings.TrimSpace(parts[1]), nil
}

type importRoleConfigClient interface {
	GetImportRoleConfig(ctx context.Context, clusterID uint64) (*agent.ImportRoleConfig, error)
}

type dmClusterRegionalClient interface {
	GetDMClusterByID(ctx context.Context, sharedPoolId uint64, dmClusterID string) (*agent.DMCluster, error)
	ListDMClusters(ctx context.Context, sharedPoolId uint64) (*agent.ListDMClustersResponse, error)
	CreateDMCluster(ctx context.Context, sharedPoolId uint64, req *agent.CreateDMClusterRequest) (*agent.DMCluster, error)
	ScaleDMCluster(ctx context.Context, sharedPoolId uint64, dmClusterID string, req *agent.ScaleDMClusterRequest) error
	DeleteDMCluster(ctx context.Context, sharedPoolId uint64, dmClusterID string) error
}

func supportsExplicitImportRole(provider string) bool {
	switch strings.ToLower(strings.TrimSpace(provider)) {
	case string(entity.CloudProviderTypeAWS), string(entity.CloudProviderTypeAliCloud):
		return true
	default:
		return false
	}
}

func (w *PremiumDMTaskWorker) isExplicitImportRoleEnabled(task *entity.PremiumDMTask, clusterID uint64) bool {
	if w.FeatureGate == nil {
		return false
	}
	tenantID, _ := strconv.ParseUint(task.TenantID, 10, 64)
	return w.FeatureGate.IsFeatureEnabled(&feature_gate.QueryFeatureGateReq{
		TenantID:      tenantID,
		ClusterID:     clusterID,
		Region:        task.Region,
		InfraProvider: task.CloudProvider,
	}, feature_gate.ExplicitImportRole)
}

func (w *PremiumDMTaskWorker) applyImportRoleConfigToDataDirIfNeeded(
	ctx context.Context,
	task *entity.PremiumDMTask,
	clusterID uint64,
	dataDir string,
	client importRoleConfigClient,
) (string, error) {
	if task.ImportMode != entity.PremiumDMTaskImportModePhysical {
		return dataDir, nil
	}
	if !supportsExplicitImportRole(task.CloudProvider) {
		return dataDir, nil
	}
	if !w.isExplicitImportRoleEnabled(task, clusterID) {
		return dataDir, nil
	}

	importRoleConfig, err := client.GetImportRoleConfig(ctx, clusterID)
	if err != nil {
		return "", fmt.Errorf("get Premium DM import role config: %w", err)
	}
	dataDirWithRole, err := agent.ApplyImportRoleConfigToStorageURI(dataDir, importRoleConfig)
	if err != nil {
		return "", fmt.Errorf("apply Premium DM import role config to data dir: %w", err)
	}
	return dataDirWithRole, nil
}

const (
	// dmClusterMasterCPU is the default CPU quota for DM master nodes.
	dmClusterMasterCPU uint32 = 4
	// dmClusterMasterMemoryGi is the default memory quota (GiB) for DM master nodes.
	dmClusterMasterMemoryGi uint32 = 8
	// dmClusterWorkerCPU is the default CPU quota for DM worker nodes.
	dmClusterWorkerCPU uint32 = 4
	// dmClusterWorkerMemoryGi is the default memory quota (GiB) for DM worker nodes.
	dmClusterWorkerMemoryGi uint32 = 8
)

// PremiumDMTaskWorker is responsible for managing the lifecycle of DM (Data Migration) tasks.
// It handles different stages of task execution including source creation, binding, task creation, and deletion.
type PremiumDMTaskWorker struct {
	PremiumDMTaskDAO    *PremiumDMTaskDAO
	MgmtClient          grpcli.MgmtSvcGrpcClient
	Encryptor           encryptor_wrapper.Encryptor
	AgentConfig         agent.ClientConfig
	SinkEndpointModel   sink_endpoint.SinkEndpointModel
	BizSinkEndpointRepo repo.BizSinkEndpointRepo
	FeatureGate         feature_gate.FeatureGate
	IDGenerator         id_generator.IDGenerator
	EventPublisher      center.EventCenterPublisher
	newProxyFunc        func(ctx context.Context, tiDBID string,
		dmCluster *agent.DMCluster) (*PremiumDMProxy, error)
	queryTaskStatusFunc                            func(context.Context, *PremiumDMProxy, string) (*dmopenapi.GetTaskStatusResponse, error)
	startTaskFunc                                  func(context.Context, *PremiumDMProxy, string, *dmopenapi.StartTaskRequest) error
	stopTaskFunc                                   func(context.Context, *PremiumDMProxy, string, *dmopenapi.StopTaskRequest) error
	newDMClusterRegionalClientByResourcePoolIDFunc func(
		ctx context.Context, resourcePoolID uint64,
	) (dmClusterRegionalClient, error)
}

// PremiumDMTaskRuntime carries ephemeral runtime-only data derived at worker run time
// that should not be persisted into the database model (e.g. endpoints which
// may change). It is passed down the call chain instead of adding transient
// fields onto entity.PremiumDMTask.
type PremiumDMTaskRuntime struct {
	DMCluster *agent.DMCluster
}

func dmClusterFromRuntime(rt *PremiumDMTaskRuntime) *agent.DMCluster {
	if rt == nil {
		return nil
	}
	return rt.DMCluster
}

func NewPremiumDMTaskWorker(
	dao *PremiumDMTaskDAO,
	mgmtClient grpcli.MgmtSvcGrpcClient,
	encryptor encryptor_wrapper.Encryptor,
	agentCfg agent.ClientConfig,
	sinkEndpointModel sink_endpoint.SinkEndpointModel,
	bizSinkEndpointRepo repo.BizSinkEndpointRepo,
	featureGate feature_gate.FeatureGate,
	idGen id_generator.IDGenerator,
	eventPublisher center.EventCenterPublisher,
) *PremiumDMTaskWorker {
	return &PremiumDMTaskWorker{
		PremiumDMTaskDAO:    dao,
		MgmtClient:          mgmtClient,
		Encryptor:           encryptor,
		AgentConfig:         agentCfg,
		SinkEndpointModel:   sinkEndpointModel,
		BizSinkEndpointRepo: bizSinkEndpointRepo,
		FeatureGate:         featureGate,
		IDGenerator:         idGen,
		EventPublisher:      eventPublisher,
	}
}

func (w *PremiumDMTaskWorker) newProxy(
	ctx context.Context, tiDBID string, dmCluster *agent.DMCluster,
) (*PremiumDMProxy, error) {
	if w.newProxyFunc != nil {
		return w.newProxyFunc(ctx, tiDBID, dmCluster)
	}
	return NewPremiumDMProxy(ctx, tiDBID, w.MgmtClient, w.AgentConfig, dmCluster)
}

func (w *PremiumDMTaskWorker) queryTaskStatus(
	ctx context.Context, proxy *PremiumDMProxy, taskID string,
) (*dmopenapi.GetTaskStatusResponse, error) {
	if w.queryTaskStatusFunc != nil {
		return w.queryTaskStatusFunc(ctx, proxy, taskID)
	}
	return proxy.QueryPremiumDMTaskStatus(ctx, taskID, []string{})
}

func (w *PremiumDMTaskWorker) startTask(
	ctx context.Context, proxy *PremiumDMProxy, taskID string, req *dmopenapi.StartTaskRequest,
) error {
	if w.startTaskFunc != nil {
		return w.startTaskFunc(ctx, proxy, taskID, req)
	}
	return proxy.StartPremiumDMTask(ctx, taskID, req)
}

func (w *PremiumDMTaskWorker) stopTask(
	ctx context.Context, proxy *PremiumDMProxy, taskID string, req *dmopenapi.StopTaskRequest,
) error {
	if w.stopTaskFunc != nil {
		return w.stopTaskFunc(ctx, proxy, taskID, req)
	}
	return proxy.StopPremiumDMTask(ctx, taskID, req)
}

func (w *PremiumDMTaskWorker) newNGRegionalClientByClusterID(
	ctx context.Context, clusterID uint64,
) (*agent.NGRegionalClient, error) {
	return agent.NewNGRegionalClientByClusterID(ctx, clusterID, w.MgmtClient, &w.AgentConfig)
}

func (w *PremiumDMTaskWorker) newNGRegionalClientByResourcePoolID(
	ctx context.Context, resourcePoolID uint64,
) (*agent.NGRegionalClient, error) {
	return agent.NewNGRegionalClientByResourcePoolID(ctx, resourcePoolID, w.MgmtClient, &w.AgentConfig)
}

func (w *PremiumDMTaskWorker) newDMClusterRegionalClientByResourcePoolID(
	ctx context.Context, resourcePoolID uint64,
) (dmClusterRegionalClient, error) {
	if w.newDMClusterRegionalClientByResourcePoolIDFunc != nil {
		return w.newDMClusterRegionalClientByResourcePoolIDFunc(ctx, resourcePoolID)
	}
	return w.newNGRegionalClientByResourcePoolID(ctx, resourcePoolID)
}

func (w *PremiumDMTaskWorker) DeleteDMClusterIfNoWorkers(
	ctx context.Context, sharedPoolID uint64, dmClusterID string,
) error {
	logger := GetWorkerLoggerFromContext(ctx).With(
		zap.Uint64("shared_pool_id", sharedPoolID),
		zap.String("dm_cluster_id", dmClusterID),
	)

	var (
		ngRegionalClient dmClusterRegionalClient
		err              error
	)
	ngRegionalClient, err = w.newDMClusterRegionalClientByResourcePoolID(ctx, sharedPoolID)
	if err != nil {
		logger.Error("failed to create ng regional client by resource pool id", zap.Error(err))
		return err
	}

	dmCluster, err := ngRegionalClient.GetDMClusterByID(ctx, sharedPoolID, dmClusterID)
	if err != nil {
		logger.Error("failed to get dm cluster by id", zap.Error(err))
		return err
	}

	if dmCluster != nil && dmCluster.WorkerNodeQuota != nil && dmCluster.WorkerNodeQuota.NodeCount != 0 {
		return fmt.Errorf("delete premium dm cluster is not allowed when worker node count is %d",
			dmCluster.WorkerNodeQuota.NodeCount)
	}

	if err := ngRegionalClient.DeleteDMCluster(ctx, sharedPoolID, dmClusterID); err != nil {
		logger.Error("failed to delete dm cluster", zap.Error(err))
		return err
	}

	logger.Info("deleted dm cluster")
	return nil
}

func (w *PremiumDMTaskWorker) validateDMClusterID(dmCluster *agent.DMCluster) (string, error) {
	if dmCluster == nil {
		return "", errors.New("dm cluster is nil")
	}
	if dmCluster.Id == "" {
		return "", errors.New("dm cluster id is empty")
	}
	return dmCluster.Id, nil
}

func (w *PremiumDMTaskWorker) resolveDMCluster(
	ctx context.Context,
	ngRegionalClient dmClusterRegionalClient,
	sharedPoolID uint64,
	allowCreate bool,
	desiredWorkerNodeCount uint32,
) (*agent.DMCluster, string, bool, error) {
	listResp, err := ngRegionalClient.ListDMClusters(ctx, sharedPoolID)
	if err != nil {
		return nil, "", false, err
	}
	if len(listResp.DMClusters) > 1 {
		return nil, "", false, fmt.Errorf("shared pool %d has %d dm clusters", sharedPoolID, len(listResp.DMClusters))
	}
	if len(listResp.DMClusters) == 1 {
		dmCluster := listResp.DMClusters[0]
		dmClusterID, err := w.validateDMClusterID(dmCluster)
		if err != nil {
			return nil, "", false, err
		}
		return dmCluster, dmClusterID, false, nil
	}
	if !allowCreate {
		return nil, "", false, agent.DMClusterNotFoundErr
	}
	if w.IDGenerator == nil {
		return nil, "", false, errors.New("dm cluster id generator is nil")
	}

	dmClusterID := strconv.FormatUint(w.IDGenerator.NextID(), 10)
	sharedPoolIDStr := strconv.FormatUint(sharedPoolID, 10)
	dmCluster, err := ngRegionalClient.CreateDMCluster(ctx, sharedPoolID, &agent.CreateDMClusterRequest{
		DmCluster: &agent.DMCluster{
			Id:           dmClusterID,
			SharedPoolId: sharedPoolIDStr,
			MasterNodeQuota: &agent.NodeQuota{
				NodeCount: 1,
				NodeSpec: &agent.NodeSpec{
					Cpu:      dmClusterMasterCPU,
					MemoryGi: dmClusterMasterMemoryGi,
				},
			},
			WorkerNodeQuota: &agent.NodeQuota{
				NodeCount: desiredWorkerNodeCount,
				NodeSpec: &agent.NodeSpec{
					Cpu:      dmClusterWorkerCPU,
					MemoryGi: dmClusterWorkerMemoryGi,
				},
			},
		},
	})
	if err != nil {
		return nil, "", false, err
	}
	return dmCluster, dmClusterID, true, nil
}

func (w *PremiumDMTaskWorker) Run(ctx context.Context) error {
	logger := GetWorkerLoggerFromContext(ctx)
	startTime := time.Now()
	logger.Info("start to run dm task worker")

	tasks, clusterMap, err := w.fetchRunnableTasks(ctx, logger)
	if err != nil {
		return err
	}
	logger.Info("collected tasks", zap.Int("taskCount", len(tasks)))

	// create or scale out dm cluster for each resource pool
	dmClusterMap, err := w.createOrScaleOutDMClustersIfNeeded(ctx, logger, tasks, slices.Collect(maps.Values(clusterMap)))
	if err != nil {
		logger.Error("failed to create or scale dm clusters if needed", zap.Error(err))
		return err
	}
	logger.Info("finished to create or scale dm clusters if needed")

	var errs error
	for _, task := range tasks {
		runtime, ok := w.buildRuntimeForTask(logger, task, clusterMap, dmClusterMap)
		if !ok {
			continue
		}
		if err := w.handleTask(ctx, task, runtime); err != nil {
			errs = multierror.Append(errs, err)
		}
	}

	logger.Info("finished to run dm task worker", zap.Int("taskCount", len(tasks)),
		zap.Any("costSeconds", time.Since(startTime).Seconds()))
	return errs
}

// fetchRunnableTasks returns tasks in runnable stages and a map of clusterId->Cluster
func (w *PremiumDMTaskWorker) fetchRunnableTasks(
	ctx context.Context,
	logger *zap.Logger,
) ([]*entity.PremiumDMTask, map[string]*tidbmgmtservicev1.Cluster, error) {
	tasks, bizErr := w.PremiumDMTaskDAO.ListTasks(ctx, &entity.PremiumDMTaskFilter{
		Stages: []entity.PremiumDMTaskStage{
			entity.PremiumDMTaskStageSourceCreating,
			entity.PremiumDMTaskStageDeleting,
			entity.PremiumDMTaskStageTaskCreating,
			entity.PremiumDMTaskStageSourceBounding,
			entity.PremiumDMTaskStageRunning,
			entity.PremiumDMTaskStagePausing,
			entity.PremiumDMTaskStageStopped,
			entity.PremiumDMTaskStageResuming,
			entity.PremiumDMTaskStageFailed,
		},
	})
	if bizErr != nil {
		logger.Error("failed to list dm tasks", zap.Error(bizErr))
		return nil, nil, bizErr
	}
	if len(tasks) == 0 {
		return tasks, map[string]*tidbmgmtservicev1.Cluster{}, nil
	}

	clusterIDs := make([]uint64, len(tasks))
	for i, t := range tasks {
		cID, err := strconv.ParseUint(t.TiDBID, 10, 64)
		if err != nil {
			return nil, nil, err
		}
		clusterIDs[i] = cID
	}
	listResp, err := w.MgmtClient.ListClusters(
		ctx, &tidbmgmtservicev1.ListClustersRequest{
			Filter: &tidbmgmtservicev1.ClusterFilter{
				ClusterIds:    clusterIDs,
				ExcludeStates: []v1beta1.ClusterState{v1beta1.ClusterState_DELETED},
				View:          tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC,
				ClusterType:   tidbmgmtservicev1.DeployType_DEPLOY_TYPE_NEXTGEN.Enum(),
			},
		})
	if err != nil {
		logger.Error("failed to list clusters from mgmt svc", zap.Error(err))
		return nil, nil, err
	}
	clusters := listResp.Clusters
	clusterMap := make(map[string]*tidbmgmtservicev1.Cluster, len(clusters))
	for i := range clusters {
		clusterMap[strconv.FormatUint(clusters[i].ClusterId, 10)] = clusters[i]
	}
	return tasks, clusterMap, nil
}

// buildRuntimeForTask derives runtime info for a task. Returns ok=false when task should be skipped.
func (w *PremiumDMTaskWorker) buildRuntimeForTask(
	logger *zap.Logger,
	task *entity.PremiumDMTask,
	clusterMap map[string]*tidbmgmtservicev1.Cluster,
	dmClusterMap map[uint64]*agent.DMCluster,
) (*PremiumDMTaskRuntime, bool) {
	if task == nil {
		return nil, false
	}
	cluster, ok := clusterMap[task.TiDBID]
	if !ok {
		return nil, false
	}
	if cluster.GetNextGenCluster().GetState() == nextgenv1.Cluster_DELETED {
		logger.With(zap.String("clusterId", task.TiDBID)).Warn("cluster has been deleted")
		return nil, false
	}
	runtime := &PremiumDMTaskRuntime{}
	if cluster.GetNextGenCluster() != nil && cluster.GetNextGenCluster().GetSharedPoolId() != "" {
		sharedPoolID, err := strconv.ParseUint(cluster.GetNextGenCluster().GetSharedPoolId(), 10, 64)
		if err != nil {
			logger.With(zap.String("clusterId", task.TiDBID)).Warn("failed to parse shared pool id", zap.Error(err))
		} else {
			runtime.DMCluster = dmClusterMap[sharedPoolID]
		}
	}
	return runtime, true
}

func (w *PremiumDMTaskWorker) handleTask(
	ctx context.Context, task *entity.PremiumDMTask, rt *PremiumDMTaskRuntime,
) error {
	logger := GetWorkerLoggerFromContext(ctx).With(
		zap.String("task_id", task.ID),
		zap.String("task_name", task.Name),
	)

	switch task.Stage {
	case entity.PremiumDMTaskStageSourceCreating:
		err := w.OnTaskSourceCreating(ctx, task, rt)
		if err != nil {
			return err
		}
	case entity.PremiumDMTaskStageSourceBounding:
		err := w.OnTaskSourceBounding(ctx, task, rt)
		if err != nil {
			return err
		}
	case entity.PremiumDMTaskStageTaskCreating:
		err := w.OnTaskCreating(ctx, task, rt)
		if err != nil {
			return err
		}
	case entity.PremiumDMTaskStageDeleting:
		err := w.OnTaskDeleting(ctx, task, rt)
		if err != nil {
			return err
		}
	case entity.PremiumDMTaskStagePausing:
		if err := w.OnTaskPausing(ctx, task, rt); err != nil {
			return err
		}
	case entity.PremiumDMTaskStageResuming:
		if err := w.OnTaskResuming(ctx, task, rt); err != nil {
			return err
		}
	case entity.PremiumDMTaskStageRunning, entity.PremiumDMTaskStageStopped, entity.PremiumDMTaskStageFailed:
		err := w.OnTaskPolling(ctx, task, rt)
		if err != nil {
			return err
		}
	default:
		logger.Warn("unknown dm task stage", zap.String("stage", string(task.Stage)))
		return errors.New("unknown dm task stage")
	}
	return nil
}

// OnTaskSourceCreating handles the source creation stage of a DM task.
// It creates DM sources in the cluster and updates their status to bounding stage.
//
//nolint:funlen // DM source creation has many sequential steps
func (w *PremiumDMTaskWorker) OnTaskSourceCreating(
	ctx context.Context, task *entity.PremiumDMTask, rt *PremiumDMTaskRuntime,
) error {
	logger := GetWorkerLoggerFromContext(ctx).With(
		zap.String("task_id", task.ID),
		zap.String("task_name", task.Name),
		zap.String("stage", string(task.Stage)),
	)
	logger.Info("OnTaskSourceCreating", zap.Int("total_sources", len(task.DMSources)))
	for i := range task.DMSources {
		s := &task.DMSources[i]
		if s.Stage != entity.PremiumDMSourceStageCreating {
			continue
		}

		req := &dmopenapi.CreateSourceRequest{
			Source: dmopenapi.Source{
				Enable: true,
				// For mode "all", users do not need to manually provide the GTID; it will be handled automatically in the code.
				EnableGtid: s.BinlogGTID != "" && task.Mode == entity.PremiumDMTaskModeIncremental,
				Port:       int(s.ConnProfile.Port),
				SourceName: s.ID,
				User:       s.ConnProfile.User,
			},
		}

		req.Source.Host = s.ConnProfile.Target
		password, decErr := w.Encryptor.DecryptBase64(s.ConnProfile.Password)
		if decErr != nil {
			logger.Error("failed to decrypt source password", zap.Error(decErr))
			return decErr
		}
		req.Source.Password = &password
		if s.ConnProfile.Security != nil {
			req.Source.Security = buildCreateSourceSecurity(s.ConnProfile.Security)
		}
		proxy, bizErr := w.newProxy(ctx, task.TiDBID, dmClusterFromRuntime(rt))
		if bizErr != nil {
			logger.Error("failed to get dm proxy from cluster",
				zap.Error(bizErr), zap.String("task_id", task.ID), zap.String("tidb_id", task.TiDBID))
			return bizErr
		}
		_, bizErr = proxy.CreatePremiumDMSource(ctx, req)
		if bizErr != nil {
			logger.Error("failed to create dm source", zap.Error(bizErr))
			return bizErr
		}
		err := w.PremiumDMTaskDAO.UpdateSource(ctx, s.ID, map[string]any{
			"stage": entity.PremiumDMSourceStageBounding,
		})
		if err != nil {
			logger.Error("failed to update dm source stage to bounding", zap.Error(err))
			return err
		}
	}

	// Update task stage to source bounding when all sources are created
	err := w.PremiumDMTaskDAO.UpdateTask(
		ctx,
		task.ID,
		[]entity.PremiumDMTaskStage{entity.PremiumDMTaskStageSourceCreating},
		map[string]any{"stage": entity.PremiumDMTaskStageSourceBounding},
	)
	if err != nil {
		logger.Error("failed to update dm task stage to sourceBounding", zap.Error(err))
		return err
	}

	return nil
}

// OnTaskSourceBounding waits for all DM sources to be bound to workers in the DM cluster.
// Once all sources are bound, it transitions the task to the task creation stage.
func (w *PremiumDMTaskWorker) OnTaskSourceBounding(
	ctx context.Context, task *entity.PremiumDMTask, rt *PremiumDMTaskRuntime,
) error {
	logger := GetWorkerLoggerFromContext(ctx).With(
		zap.String("task_id", task.ID),
		zap.String("task_name", task.Name),
		zap.String("stage", string(task.Stage)))
	logger.Info("OnTaskSourceBounding", zap.Int("total_sources", len(task.DMSources)))
	allBounded := true
	proxy, bizErr := w.newProxy(ctx, task.TiDBID, dmClusterFromRuntime(rt))
	if bizErr != nil {
		logger.Error("failed to get dm proxy from cluster",
			zap.Error(bizErr), zap.String("task_id", task.ID), zap.String("tidb_id", task.TiDBID))
		return bizErr
	}
	for i := range task.DMSources {
		s := &task.DMSources[i]
		if s.Stage == entity.PremiumDMSourceStageBounded {
			continue
		}

		source, err := proxy.GetPremiumDMSource(ctx, s.ID)
		if err != nil {
			logger.Error("failed to get dm source from proxy", zap.Error(err))
			return err
		}
		if source.StatusList != nil && len(*source.StatusList) > 0 && (*source.StatusList)[0].WorkerName != "" {
			err = w.PremiumDMTaskDAO.UpdateSource(ctx, s.ID, map[string]any{
				"stage":        entity.PremiumDMSourceStageBounded,
				"conn_profile": s.ConnProfile,
			})
			if err != nil {
				logger.Error("failed to update dm source stage to bounded", zap.Error(err))
				return err
			}
			continue
		}
		allBounded = false
	}

	if allBounded {
		err := w.PremiumDMTaskDAO.UpdateTask(
			ctx,
			task.ID,
			[]entity.PremiumDMTaskStage{entity.PremiumDMTaskStageSourceBounding},
			map[string]any{"stage": entity.PremiumDMTaskStageTaskCreating},
		)
		if err != nil {
			logger.Error("failed to update dm task stage to taskCreating", zap.Error(err))
			return err
		}
	}

	return nil
}

// OnTaskCreating creates the actual DM task in the cluster after all sources are bound.
// It configures the migration rules, source configurations, and target database settings.
//
//nolint:funlen,gocyclo // DM task creation has many configuration steps
func (w *PremiumDMTaskWorker) OnTaskCreating(
	ctx context.Context, task *entity.PremiumDMTask, rt *PremiumDMTaskRuntime,
) error {
	logger := GetWorkerLoggerFromContext(ctx).With(
		zap.String("task_id", task.ID),
		zap.String("task_name", task.Name),
		zap.String("stage", string(task.Stage)),
	)
	logger.Info("OnTaskCreating", zap.Int("total_sources", len(task.DMSources)))
	proxy, bizErr := w.newProxy(ctx, task.TiDBID, dmClusterFromRuntime(rt))
	if bizErr != nil {
		logger.Error("failed to get dm proxy from cluster",
			zap.Error(bizErr), zap.String("task_id", task.ID), zap.String("tidb_id", task.TiDBID))
		return bizErr
	}

	password, decErr := w.Encryptor.DecryptBase64(task.TargetEncryptedPassword)
	if decErr != nil {
		logger.Error("failed to decrypt target password", zap.Error(decErr))
		return decErr
	}
	clusterID, parseErr := strconv.ParseUint(task.TiDBID, 10, 64)
	if parseErr != nil {
		return parseErr
	}
	agentClient, agentErr := w.newNGRegionalClientByClusterID(ctx, clusterID)
	if agentErr != nil {
		return agentErr
	}
	shardPool, agentErr := agentClient.GetSharePoolByClusterID(ctx, clusterID)
	if agentErr != nil {
		return agentErr
	}
	dataDir, dataDirErr := buildSharedPoolDataDir(task, shardPool.InternalStorage, shardPool.Provider)
	if dataDirErr != nil {
		logger.Error(
			"failed to build shared pool data dir",
			zap.Error(dataDirErr),
			zap.String("cloudProvider", task.CloudProvider),
		)
		return dataDirErr
	}
	dataDir, dataDirErr = w.applyImportRoleConfigToDataDirIfNeeded(ctx, task, clusterID, dataDir, agentClient)
	if dataDirErr != nil {
		logger.Error("failed to apply import role config to data dir", zap.Error(dataDirErr))
		return dataDirErr
	}
	uri := aws.String(dataDir)

	// Modify URI for AliCloud OSS if needed
	uri, parseErr = w.modifyURIForAliCloudOSS(task, uri, logger)
	if parseErr != nil {
		logger.Error("failed to modify uri", zap.Error(parseErr))
		return parseErr
	}

	dmConf := config.Global().PremiumDMConfig
	taskFullMigrateConf := &dmopenapi.TaskFullMigrateConf{
		Consistency:   aws.String("auto"),
		DataDir:       uri,
		ImportThreads: pointer.To(dmConf.ImportThreads),
		ExportThreads: pointer.To(dmConf.ExportThreads),
	}
	switch task.ImportMode {
	case entity.PremiumDMTaskImportModeLogical:
		taskFullMigrateConf.ImportMode = pointer.To(dmopenapi.TaskFullMigrateConfImportModeLogical)
		taskFullMigrateConf.OnDuplicateLogical = pointer.To(dmopenapi.TaskFullMigrateConfOnDuplicateLogicalReplace)
	case entity.PremiumDMTaskImportModePhysical:
		taskFullMigrateConf.ImportMode = pointer.To(dmopenapi.TaskFullMigrateConfImportMode("import-into"))
	}
	req := &dmopenapi.CreateTaskRequest{
		Task: dmopenapi.Task{
			Name:                      task.ID,
			OnDuplicate:               dmopenapi.TaskOnDuplicateReplace,
			EnhanceOnlineSchemaChange: true,
			// do precheck using separate api before creating task
			IgnoreCheckingItems: &[]string{"all"},
			SourceConfig: dmopenapi.TaskSourceConfig{
				FullMigrateConf: taskFullMigrateConf,
				IncrMigrateConf: &dmopenapi.TaskIncrMigrateConf{
					ReplThreads: pointer.To(dmConf.ReplThreads),
					ReplBatch:   pointer.To(dmConf.ReplBatch),
				},
			},
			TargetConfig: dmopenapi.TaskTargetDataBase{
				Password: password,
				Port:     defaultTiDBPort,
				User:     task.TargetUser,
			},
			TaskMode: buildTaskMode(task.Mode),
		},
	}

	// Configure source configurations and migration rules
	sourceConfs := make([]dmopenapi.TaskSourceConf, 0, len(task.DMSources))
	migrateRules := make([]dmopenapi.TaskTableMigrateRule, 0, len(task.DMSources))
	for i := range task.DMSources {
		s := &task.DMSources[i]
		sourceConfs = append(sourceConfs, dmopenapi.TaskSourceConf{
			BinlogGtid: aws.String(s.BinlogGTID),
			BinlogName: aws.String(s.BinlogName),
			BinlogPos:  aws.Int(int(s.BinlogPos)),
			SourceName: s.ID,
		})
		if len(s.MigrationRules) > 0 {
			migrateRules = append(migrateRules, buildTaskMigrateRulesFromMigrationRules(s)...)
		} else {
			migrateRules = append(migrateRules, buildTaskMigrateRulesFromBlockAllowRules(s)...)
		}
	}
	req.Task.SourceConfig.SourceConf = sourceConfs
	req.Task.TableMigrateRule = migrateRules

	_, bizErr = createPremiumDMTaskWithMetricLabels(ctx, proxy, req, task.TiDBID)
	var openapiErr *DMOpenapiError
	if errors.As(bizErr, &openapiErr) && openapiErr.ErrorCode == CodeSchedulerSubTaskExist {
		logger.Warn("dm task already created", zap.Error(bizErr))
	} else if bizErr != nil {
		logger.Error("failed to create dm task", zap.Error(bizErr))
		return bizErr
	}

	sources := make(dmopenapi.SourceNameList, 0, len(task.DMSources))
	for i := range task.DMSources {
		sources = append(sources, task.DMSources[i].ID)
	}
	bizErr = proxy.StartPremiumDMTask(ctx, task.ID, &dmopenapi.StartTaskRequest{
		SourceNameList: &sources,
	})
	if bizErr != nil {
		logger.Error("failed to start dm task", zap.Error(bizErr))
		return bizErr
	}

	err := w.PremiumDMTaskDAO.UpdateTask(
		ctx,
		task.ID,
		[]entity.PremiumDMTaskStage{entity.PremiumDMTaskStageTaskCreating},
		map[string]any{"stage": entity.PremiumDMTaskStageRunning},
	)
	if err != nil {
		logger.Error("failed to update dm task stage to running", zap.Error(err))
		return err
	}
	task.Stage = entity.PremiumDMTaskStageRunning
	PublishPremiumDMEvent(
		ctx, w.EventPublisher, task, PremiumDMOperationCreate, event.EventStatusRunning,
	)

	return nil
}

// generateOssInternalEndpoint generates the internal OSS endpoint for AliCloud.
func generateOssInternalEndpoint(ossRegion string) string {
	return fmt.Sprintf("https://oss-%s-internal.aliyuncs.com", ossRegion)
}

// modifyURIForAliCloudOSS modifies the S3 URI for AliCloud OSS compatibility.
func (w *PremiumDMTaskWorker) modifyURIForAliCloudOSS(task *entity.PremiumDMTask,
	uri *string, logger *zap.Logger,
) (*string, error) {
	if !strings.EqualFold(strings.TrimSpace(task.CloudProvider), string(entity.CloudProviderTypeAliCloud)) {
		return uri, nil
	}

	sourceUrl, err := url.Parse(*uri)
	if err != nil {
		logger.Error("failed to parse source url", zap.Error(err), zap.String("uri", *uri))
		return nil, err
	}

	query := sourceUrl.Query()
	query.Set("provider", "alibaba")
	query.Set("region", task.Region)
	query.Set("endpoint", generateOssInternalEndpoint(task.Region))
	sourceUrl.RawQuery = query.Encode()

	return aws.String(sourceUrl.String()), nil
}

// OnTaskDeleting handles the deletion of DM tasks and their associated sources.
// It gracefully handles cases where resources may have already been deleted.
func (w *PremiumDMTaskWorker) OnTaskDeleting(
	ctx context.Context, task *entity.PremiumDMTask, rt *PremiumDMTaskRuntime,
) error {
	logger := GetWorkerLoggerFromContext(ctx).With(
		zap.String("task_id", task.ID),
		zap.String("task_name", task.Name),
		zap.String("stage", string(task.Stage)),
	)
	logger.Info("OnTaskDeleting")
	proxy, bizErr := w.newProxy(ctx, task.TiDBID, dmClusterFromRuntime(rt))
	if bizErr != nil {
		if bizerr.ParseGormErr2code(bizErr) != bizerr.ErrCodeRecordNotFound {
			logger.Error("failed to get dm proxy from cluster",
				zap.Error(bizErr), zap.String("task_id", task.ID), zap.String("tidb_id", task.TiDBID))
			return bizErr
		}
		logger.Warn("dm proxy not found as cluster may be deleted, skip deleting dm task and sources",
			zap.Error(bizErr), zap.String("task_id", task.ID), zap.String("tidb_id", task.TiDBID))
	}

	uint64ClusterId, err := strconv.ParseUint(task.TiDBID, 10, 64)
	if err != nil {
		logger.Error("failed to parse cluster id", zap.Error(err))
		return err
	}

	if proxy != nil {
		if err = w.deletePremiumDMResources(ctx, logger, proxy, task, uint64ClusterId); err != nil {
			return err
		}
	}

	if err = w.deletePremiumDMClusterStatus(ctx, logger, task, uint64ClusterId); err != nil {
		return err
	}

	return w.finishPremiumDMTaskDeletion(ctx, logger, task)
}

func (w *PremiumDMTaskWorker) deletePremiumDMResources(
	ctx context.Context,
	logger *zap.Logger,
	proxy *PremiumDMProxy,
	task *entity.PremiumDMTask,
	clusterID uint64,
) error {
	if err := stopPremiumDMTaskForDeletion(ctx, logger, proxy, task); err != nil {
		return err
	}
	if err := deletePremiumDMTaskMetadata(ctx, logger, proxy, task.ID); err != nil {
		return err
	}
	if err := w.deletePremiumDMSources(ctx, logger, proxy, task); err != nil {
		return err
	}

	w.scaleInPremiumDMCluster(ctx, logger, clusterID)
	return nil
}

func stopPremiumDMTaskForDeletion(
	ctx context.Context,
	logger *zap.Logger,
	proxy *PremiumDMProxy,
	task *entity.PremiumDMTask,
) error {
	sourceNames := make(dmopenapi.SourceNameList, 0, len(task.DMSources))
	for i := range task.DMSources {
		sourceNames = append(sourceNames, task.DMSources[i].ID)
	}
	err := proxy.StopPremiumDMTask(ctx, task.ID, &dmopenapi.StopTaskRequest{
		SourceNameList: &sourceNames,
	})
	var openapiErr *DMOpenapiError
	if errors.As(err, &openapiErr) && openapiErr.ErrorCode == CodeSchedulerSubTaskOpTaskNotExist {
		logger.Warn("dm task not found, may be already delete or not started", zap.Error(err))
		return nil
	}
	if err != nil {
		logger.Error("delete dm task failed", zap.Error(err))
	}
	return err
}

func deletePremiumDMTaskMetadata(
	ctx context.Context,
	logger *zap.Logger,
	proxy *PremiumDMProxy,
	taskID string,
) error {
	// Deleting a task also removes metadata from the downstream database. Allow
	// force deletion to continue if changed credentials make that cleanup fail.
	err := proxy.DeletePremiumDMTask(ctx, taskID)
	var openapiErr *DMOpenapiError
	if errors.As(err, &openapiErr) && openapiErr.ErrorCode == CodeSchedulerTaskNotExist {
		logger.Warn("dm task not found, may be already deleted or not started", zap.Error(err))
		return nil
	}
	if err != nil && !strings.Contains(err.Error(), MySQLAccessDeniedErrorString) {
		logger.Error("delete dm task failed", zap.Error(err))
		return err
	}
	return nil
}

func (w *PremiumDMTaskWorker) deletePremiumDMSources(
	ctx context.Context,
	logger *zap.Logger,
	proxy *PremiumDMProxy,
	task *entity.PremiumDMTask,
) error {
	for i := range task.DMSources {
		source := &task.DMSources[i]
		if source.Stage == entity.PremiumDMSourceStageDeleted {
			continue
		}
		if err := w.deletePremiumDMSource(ctx, logger, proxy, task.ID, source); err != nil {
			return err
		}
	}
	return nil
}

func (w *PremiumDMTaskWorker) deletePremiumDMSource(
	ctx context.Context,
	logger *zap.Logger,
	proxy *PremiumDMProxy,
	taskID string,
	source *entity.PremiumDMSource,
) error {
	logger = logger.With(zap.String("dm_source_id", source.ID))
	err := proxy.DeletePremiumDMSource(ctx, source.ID)
	var openapiErr *DMOpenapiError
	if errors.As(err, &openapiErr) && openapiErr.ErrorCode == CodeSchedulerSourceCfgNotExist {
		logger.Warn(
			"dm source not found in sharedpool cluster, may be already deleted or not created",
			zap.Error(err),
		)
	} else if err != nil {
		logger.Error("delete dm source failed", zap.Error(err), zap.String("dm_source_id", source.ID))
		return err
	}

	if source.ConnProfile.Type == entity.PremiumConnTypePrivateLink {
		delErr := w.deleteSinkEndpointUsageForSource(ctx, taskID, source.ConnProfile.SinkEndpointID)
		if delErr != nil {
			logger.Error("failed to delete sink endpoint usage",
				zap.Error(delErr), zap.String("source_id", source.ID))
			// Non-fatal: continue deletion even if usage cleanup fails.
		}
	}

	source.ConnProfile.Password = ""
	source.ConnProfile.Security = nil
	err = w.PremiumDMTaskDAO.UpdateSource(ctx, source.ID, map[string]any{
		"stage":        entity.PremiumDMSourceStageDeleted,
		"deleted_time": time.Now(),
		"conn_profile": source.ConnProfile,
	})
	if err != nil {
		logger.Error(
			"failed to update dm source stage to deleted",
			zap.Error(err),
			zap.String("dm_source_id", source.ID),
		)
		return err
	}
	logger.Info("successfully deleted dm source", zap.String("source_id", source.ID))
	return nil
}

func (w *PremiumDMTaskWorker) scaleInPremiumDMCluster(
	ctx context.Context,
	logger *zap.Logger,
	clusterID uint64,
) {
	// TODO: remove synchronized scale in operation, use async scale in operation instead.
	// Scale in DM cluster worker nodes after task deletion, regardless of cluster state.
	clusters, listErr := w.MgmtClient.ListClusters(ctx, &tidbmgmtservicev1.ListClustersRequest{
		Filter: &tidbmgmtservicev1.ClusterFilter{
			ClusterIds:  []uint64{clusterID},
			View:        tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC,
			ClusterType: tidbmgmtservicev1.DeployType_DEPLOY_TYPE_NEXTGEN.Enum(),
		},
	})
	if listErr != nil {
		logger.Warn("failed to get cluster by id", zap.Uint64("cluster_id", clusterID), zap.Error(listErr))
	}
	if clusters == nil || len(clusters.Clusters) == 0 {
		logger.Info("cluster not found, skip scaling in dm cluster worker nodes", zap.Uint64("cluster_id", clusterID))
		return
	}

	cluster := clusters.Clusters[0]
	if cluster == nil {
		logger.Warn("cluster info or its shared pool id is nil or empty", zap.Uint64("cluster_id", clusterID))
		return
	}
	ngCluster := cluster.GetNextGenCluster()
	if ngCluster == nil || ngCluster.GetSharedPoolId() == "" {
		logger.Warn("cluster info or its shared pool id is nil or empty", zap.Uint64("cluster_id", clusterID))
		return
	}

	sharedPoolID := ngCluster.GetSharedPoolId()
	if err := w.decrementDMClusterWorkerCount(ctx, logger, sharedPoolID); err != nil {
		logger.Warn("failed to scale in dm cluster worker nodes by 1",
			zap.String("shared_pool_id", sharedPoolID), zap.Error(err))
		return
	}
	logger.Info("successfully scaled in dm cluster worker nodes by 1", zap.String("shared_pool_id", sharedPoolID))
}

func (w *PremiumDMTaskWorker) deletePremiumDMClusterStatus(
	ctx context.Context,
	logger *zap.Logger,
	task *entity.PremiumDMTask,
	clusterID uint64,
) error {
	if task.ClusterStatusID == 0 {
		return nil
	}
	err := w.MgmtClient.DeleteClusterCustomStatus(ctx, &tidbmgmtservicev1.DeleteClusterCustomStatusRequest{
		ClusterId: clusterID,
		StatusId:  task.ClusterStatusID,
	})
	if err != nil {
		logger.Error("failed to delete cluster custom status for dm task", zap.Error(err))
	}
	return err
}

func (w *PremiumDMTaskWorker) finishPremiumDMTaskDeletion(
	ctx context.Context,
	logger *zap.Logger,
	task *entity.PremiumDMTask,
) error {
	err := w.PremiumDMTaskDAO.UpdateTask(ctx, task.ID, entity.NonDeletedPremiumDMTaskStages, map[string]any{
		"stage":                     entity.PremiumDMTaskStageDeleted,
		"deleted_time":              time.Now(),
		"target_encrypted_password": "",
	})
	if err != nil {
		logger.Error("failed to update dm task stage to deleted", zap.Error(err))
		return err
	}
	task.Stage = entity.PremiumDMTaskStageDeleted
	PublishPremiumDMEvent(
		ctx, w.EventPublisher, task, PremiumDMOperationDelete, event.EventStatusSucceeded,
	)

	logger.Info("successfully deleted dm task and all sources")

	return nil
}

// deleteSinkEndpointUsageForSource deletes the BizSinkEndpointUsage for a
// premium DM task source identified by taskID and sinkEndpointID.
func (w *PremiumDMTaskWorker) deleteSinkEndpointUsageForSource(
	ctx context.Context,
	taskID string,
	sinkEndpointID uint64,
) error {
	return w.SinkEndpointModel.DeleteSinkEndpointUsage(ctx, &domain.DeleteSinkEndpointUsageReq{
		SinkEndpointID: sinkEndpointID,
		OwnerID:        taskID,
		UsageType:      entity.BizSinkEndpointUsageTypePremiumDMTask,
	})
}

func (w *PremiumDMTaskWorker) OnTaskPausing(
	ctx context.Context, task *entity.PremiumDMTask, rt *PremiumDMTaskRuntime,
) error {
	return w.onTaskPauseOrResume(ctx, task, rt, true)
}

func (w *PremiumDMTaskWorker) OnTaskResuming(
	ctx context.Context, task *entity.PremiumDMTask, rt *PremiumDMTaskRuntime,
) error {
	return w.onTaskPauseOrResume(ctx, task, rt, false)
}

// onTaskPauseOrResume dispatches a durable task command once, then leaves the
// task in its transitional stage until all remote subtasks have converged.
func (w *PremiumDMTaskWorker) onTaskPauseOrResume(
	ctx context.Context, task *entity.PremiumDMTask, rt *PremiumDMTaskRuntime, pausing bool,
) error {
	logger := GetWorkerLoggerFromContext(ctx).With(
		zap.String("task_id", task.ID),
		zap.String("task_name", task.Name),
		zap.String("stage", string(task.Stage)),
	)
	proxy, err := w.newProxy(ctx, task.TiDBID, dmClusterFromRuntime(rt))
	if err != nil {
		logger.Error("failed to get dm proxy for task transition", zap.Error(err))
		return err
	}

	status, err := w.queryTaskStatus(ctx, proxy, task.ID)
	var openapiErr *DMOpenapiError
	if errors.As(err, &openapiErr) && openapiErr.ErrorCode == CodeSchedulerTaskNotExist {
		if pausing {
			logger.Warn("dm task not found while pausing; treating remote task as stopped", zap.Error(err))
			return w.finishPremiumDMTransition(ctx, task, true, "")
		}
		logger.Warn("dm task not found while resuming; marking transition failed", zap.Error(err))
		return w.finishPremiumDMTransition(ctx, task, false, openapiErr.Error())
	}
	if err != nil {
		logger.Error("query dm task transition status failed", zap.Error(err))
		return err
	}
	if err := w.persistPremiumDMSubTaskStatuses(ctx, task, status); err != nil {
		logger.Error("persist dm task transition status failed", zap.Error(err))
		return err
	}
	converged, remoteErr := premiumDMTransitionState(task, status, pausing)
	if converged {
		return w.finishPremiumDMTransition(ctx, task, pausing, "")
	}
	if remoteErr != "" && pausing {
		return w.finishPremiumDMTransition(ctx, task, true, remoteErr)
	}
	if task.OperationDispatchedAt != nil {
		if remoteErr != "" {
			// DM exposes no operation version or request ID. Once Start/Stop has
			// been accepted, an explicit remote subtask error is therefore treated
			// as the current operation failure instead of waiting indefinitely on
			// an error string that may happen to match a historical failure.
			return w.finishPremiumDMTransition(ctx, task, pausing, remoteErr)
		}
		return nil
	}

	sourceNames := make(dmopenapi.SourceNameList, 0, len(task.DMSources))
	for i := range task.DMSources {
		sourceNames = append(sourceNames, task.DMSources[i].ID)
	}
	if pausing {
		err = w.stopTask(ctx, proxy, task.ID, &dmopenapi.StopTaskRequest{SourceNameList: &sourceNames})
	} else {
		err = w.startTask(ctx, proxy, task.ID, &dmopenapi.StartTaskRequest{SourceNameList: &sourceNames})
	}
	if err != nil {
		logger.Error("dispatch dm task transition failed", zap.Error(err))
		return err
	}

	dispatchedAt := time.Now()
	rowsAffected, bizErr := w.PremiumDMTaskDAO.UpdateTaskWithRowsAffected(
		ctx,
		task.ID,
		[]entity.PremiumDMTaskStage{task.Stage},
		map[string]any{"operation_dispatched_at": dispatchedAt},
	)
	if bizErr != nil {
		logger.Error("persist dm task transition dispatch failed", zap.Error(bizErr))
		return bizErr
	}
	if rowsAffected > 0 {
		task.OperationDispatchedAt = &dispatchedAt
	}
	return nil
}

func premiumDMTransitionState(
	task *entity.PremiumDMTask,
	status *dmopenapi.GetTaskStatusResponse,
	pausing bool,
) (converged bool, remoteErr string) {
	if status == nil || len(task.DMSources) == 0 {
		return false, ""
	}
	currentStatus := make(map[string]*dmopenapi.SubTaskStatus, len(status.Data))
	for i := range status.Data {
		currentStatus[status.Data[i].SourceName] = &status.Data[i]
	}

	converged = true
	errMsgs := make([]string, 0)
	for i := range task.DMSources {
		st, ok := currentStatus[task.DMSources[i].ID]
		if !ok {
			converged = false
			continue
		}
		isStopped := st.Stage == dmopenapi.TaskStageStopped || st.Stage == dmopenapi.TaskStagePaused
		if isStopped && st.ErrorMsg != nil && *st.ErrorMsg != "" {
			errMsgs = append(errMsgs, *st.ErrorMsg)
			converged = false
		}
		if pausing {
			if st.Stage != dmopenapi.TaskStageStopped && st.Stage != dmopenapi.TaskStagePaused {
				converged = false
			}
		} else if st.Stage != dmopenapi.TaskStageRunning {
			converged = false
		}
	}
	return converged, strings.Join(errMsgs, "\n")
}

func (w *PremiumDMTaskWorker) persistPremiumDMSubTaskStatuses(
	ctx context.Context,
	task *entity.PremiumDMTask,
	status *dmopenapi.GetTaskStatusResponse,
) error {
	if status == nil {
		return nil
	}
	knownSources := make(map[string]struct{}, len(task.DMSources))
	for i := range task.DMSources {
		knownSources[task.DMSources[i].ID] = struct{}{}
	}
	for i := range status.Data {
		st := &status.Data[i]
		if _, ok := knownSources[st.SourceName]; !ok {
			continue
		}
		payload, err := json.Marshal(st)
		if err != nil {
			return fmt.Errorf("json marshal sub_task_status for source %s failed: %w", st.SourceName, err)
		}
		if err := w.PremiumDMTaskDAO.UpdateSource(ctx, st.SourceName, map[string]any{
			"sub_task_status": payload,
		}); err != nil {
			return fmt.Errorf("update sub_task_status for source %s failed: %w", st.SourceName, err)
		}
	}
	return nil
}

func (w *PremiumDMTaskWorker) finishPremiumDMTransition(
	ctx context.Context,
	task *entity.PremiumDMTask,
	pausing bool,
	remoteErr string,
) error {
	targetStage := entity.PremiumDMTaskStageRunning
	operation := PremiumDMOperationResume
	displayStatus := event.EventStatusSucceeded
	if pausing {
		targetStage = entity.PremiumDMTaskStageStopped
		operation = PremiumDMOperationPause
	}
	updates := map[string]any{
		"stage":                   targetStage,
		"operation_dispatched_at": nil,
		"err_msg":                 "",
		"failed_time":             nil,
	}
	if remoteErr != "" {
		targetStage = entity.PremiumDMTaskStageFailed
		displayStatus = event.EventStatusFailed
		updates["stage"] = targetStage
		updates["err_msg"] = remoteErr
		updates["failed_time"] = time.Now()
	}

	rowsAffected, bizErr := w.PremiumDMTaskDAO.UpdateTaskWithRowsAffected(
		ctx,
		task.ID,
		[]entity.PremiumDMTaskStage{task.Stage},
		updates,
	)
	if bizErr != nil {
		return bizErr
	}
	if rowsAffected == 0 {
		return nil
	}
	task.Stage = targetStage
	task.OperationDispatchedAt = nil
	task.ErrMsg = remoteErr
	PublishPremiumDMEvent(ctx, w.EventPublisher, task, operation, displayStatus)
	return nil
}

//nolint:funlen,gocyclo // polling logic has many sequential checks
func (w *PremiumDMTaskWorker) OnTaskPolling(
	ctx context.Context, task *entity.PremiumDMTask, rt *PremiumDMTaskRuntime,
) error {
	logger := GetWorkerLoggerFromContext(ctx).With(
		zap.String("task_id", task.ID),
		zap.String("task_name", task.Name),
		zap.String("stage", string(task.Stage)),
	)
	logger.Debug("OnTaskPolling: querying dm task status")

	proxy, bizErr := w.newProxy(ctx, task.TiDBID, dmClusterFromRuntime(rt))
	if bizErr != nil {
		logger.Error("failed to get dm proxy from cluster",
			zap.Error(bizErr), zap.String("task_id", task.ID), zap.String("tidb_id", task.TiDBID))
		return bizErr
	}

	status, err := w.queryTaskStatus(ctx, proxy, task.ID)
	var openapiErr *DMOpenapiError
	if errors.As(err, &openapiErr) && openapiErr.ErrorCode == CodeSchedulerTaskNotExist {
		logger.Warn("dm task not found, may be already deleted or not started", zap.Error(err))
		return nil
	} else if err != nil {
		logger.Error("query dm task status failed", zap.Error(err))
		return err
	}
	if err := w.persistPremiumDMSubTaskStatuses(ctx, task, status); err != nil {
		logger.Error("update sub_task_status failed", zap.Error(err))
		return err
	}

	// Build map sourceName -> *dmopenapi.SubTaskStatus from API
	currentStatus := make(map[string]*dmopenapi.SubTaskStatus, len(status.Data))
	for _, st := range status.Data {
		// we will fetch/merge existing stored status per source below
		currentStatus[st.SourceName] = &st
	}

	stage := entity.PremiumDMTaskStageRunning
	var errMsgs []string
	prevStage := task.Stage
	for i := range task.DMSources {
		s := &task.DMSources[i]
		st, ok := currentStatus[s.ID]
		if !ok {
			continue
		}

		if st.Stage == dmopenapi.TaskStageStopped || st.Stage == dmopenapi.TaskStagePaused {
			isFailed := st.ErrorMsg != nil && *st.ErrorMsg != ""
			if isFailed {
				stage = entity.PremiumDMTaskStageFailed
				errMsgs = append(errMsgs, *st.ErrorMsg)
				continue
			}

			if stage != entity.PremiumDMTaskStageFailed {
				stage = entity.PremiumDMTaskStageStopped
			}
		}
	}

	updates := map[string]any{
		"stage": stage,
	}

	if len(errMsgs) > 0 {
		errMsg := strings.Join(errMsgs, "\n")
		updates["err_msg"] = errMsg
		updates["failed_time"] = time.Now()
	} else {
		updates["err_msg"] = ""
		updates["failed_time"] = nil
	}

	rowsAffected, bizErr := w.PremiumDMTaskDAO.UpdateTaskWithRowsAffected(
		ctx, task.ID, entity.PollingPremiumDMTaskStages, updates,
	)
	if bizErr != nil {
		logger.Error("failed to update dm task stage during polling", zap.Error(bizErr))
		return bizErr
	}
	if rowsAffected == 0 {
		return nil
	}
	if prevStage != stage && stage == entity.PremiumDMTaskStageFailed {
		task.Stage = stage
		if len(errMsgs) > 0 {
			task.ErrMsg = strings.Join(errMsgs, "\n")
		}
		PublishPremiumDMEvent(
			ctx, w.EventPublisher, task, PremiumDMOperationCreate, event.EventStatusFailed,
		)
	}

	return nil
}

// decrementDMClusterWorkerCount attempts to scale in the DM cluster worker nodes by 1
// when a DM task is deleted. It will not scale below 0 and is a no-op if there are no nodes.
//
//nolint:funlen // cluster scaling has many sequential steps
func (w *PremiumDMTaskWorker) decrementDMClusterWorkerCount(
	ctx context.Context,
	logger *zap.Logger,
	sharedPoolID string,
) error {
	sharedPoolIDUint, err := strconv.ParseUint(sharedPoolID, 10, 64)
	if err != nil {
		logger.Error("failed to parse shared pool id to uint64",
			zap.String("shared_pool_id", sharedPoolID), zap.Error(err))
		return err
	}

	ngRegionalClient, err := w.newDMClusterRegionalClientByResourcePoolID(ctx, sharedPoolIDUint)
	if err != nil {
		if bizerr.ParseGormErr2code(err) == bizerr.ErrCodeRecordNotFound {
			logger.Info("resource pool not found, may already be deleted, skip scaling in",
				zap.Uint64("shared_pool_id", sharedPoolIDUint))
			return nil
		}
		logger.Error("failed to create ng regional client by resource pool id",
			zap.Uint64("shared_pool_id", sharedPoolIDUint), zap.Error(err))
		return err
	}

	dmCluster, dmClusterID, _, err := w.resolveDMCluster(ctx, ngRegionalClient, sharedPoolIDUint, false, 0)
	if err != nil {
		if errors.Is(err, agent.DMClusterNotFoundErr) {
			logger.Warn("dm cluster not found, skip scaling in",
				zap.Uint64("shared_pool_id", sharedPoolIDUint))
			return nil
		}
		logger.Error("failed to resolve dm cluster", zap.Uint64("shared_pool_id", sharedPoolIDUint), zap.Error(err))
		return err
	}
	// check dm cluster node quota/spec
	if dmCluster == nil || dmCluster.WorkerNodeQuota == nil ||
		dmCluster.WorkerNodeQuota.NodeSpec == nil ||
		dmCluster.WorkerNodeQuota.NodeSpec.Cpu == 0 ||
		dmCluster.WorkerNodeQuota.NodeSpec.MemoryGi == 0 {
		logger.Error("dm cluster or its node quota/spec is nil or invalid", zap.Uint64("shared_pool_id", sharedPoolIDUint))
		return errors.New("dm cluster or its worker node quota/spec is nil or invalid")
	}

	if dmCluster.WorkerNodeQuota.NodeCount == 0 {
		logger.Info("dm cluster worker node count is already 0, no need to scale in",
			zap.Uint64("shared_pool_id", sharedPoolIDUint))
		return nil
	}

	// scale dm cluster in by 1
	newNodeCount := dmCluster.WorkerNodeQuota.NodeCount - 1
	logger.Info("scaling in dm cluster worker nodes by 1", zap.Uint64("shared_pool_id", sharedPoolIDUint),
		zap.Int("current_worker_nodes", int(dmCluster.WorkerNodeQuota.NodeCount)),
		zap.Int("new_worker_nodes", int(newNodeCount)))
	err = ngRegionalClient.ScaleDMCluster(ctx, sharedPoolIDUint, dmClusterID, &agent.ScaleDMClusterRequest{
		DmClusterId: dmClusterID,
		WorkerNodeQuota: &agent.NodeQuota{
			NodeCount: newNodeCount,
			// keep same spec as before
			NodeSpec: &agent.NodeSpec{
				Cpu:      dmCluster.WorkerNodeQuota.NodeSpec.Cpu,
				MemoryGi: dmCluster.WorkerNodeQuota.NodeSpec.MemoryGi,
			},
		},
	})
	if err != nil {
		logger.Error("failed to scale in dm cluster worker nodes",
			zap.Uint64("shared_pool_id", sharedPoolIDUint), zap.Error(err))
		return err
	}

	return nil
}

//nolint:funlen // cluster provisioning has many sequential steps
func (w *PremiumDMTaskWorker) createOrScaleOutDMClustersIfNeeded(
	ctx context.Context,
	logger *zap.Logger,
	tasks []*entity.PremiumDMTask,
	clusters []*tidbmgmtservicev1.Cluster,
) (map[uint64]*agent.DMCluster, error) {
	poolToTasks, err := buildPoolToTasksMap(logger, tasks, clusters)
	if err != nil {
		logger.Error("failed to build pool to tasks map", zap.Error(err))
		return nil, err
	}

	dmClusterMap := make(map[uint64]*agent.DMCluster, len(poolToTasks))

	// create or scale clusters per resource pool
	for sharedPoolID, tasks := range poolToTasks {
		ngRegionalClient, err := w.newDMClusterRegionalClientByResourcePoolID(ctx, sharedPoolID)
		if err != nil {
			logger.Error("failed to create ng regional client", zap.Uint64("shared_pool_id", sharedPoolID), zap.Error(err))
			continue
		}

		if len(tasks) == 0 {
			logger.Warn("no tasks found for resource pool; skip create or scale out", zap.Uint64("shared_pool_id", sharedPoolID))
			continue
		}

		dmCluster, dmClusterID, created, err := w.resolveDMCluster(
			ctx, ngRegionalClient, sharedPoolID, true, uint32(len(tasks)),
		)
		if err != nil {
			logger.Error("failed to resolve dm cluster", zap.Uint64("shared_pool_id", sharedPoolID), zap.Error(err))
			continue
		}
		if created {
			logger.Info("successfully created dm cluster",
				zap.Uint64("shared_pool_id", sharedPoolID),
				zap.String("dm_cluster_id", dmClusterID))
			dmClusterMap[sharedPoolID] = dmCluster
			continue
		}
		dmClusterMap[sharedPoolID] = dmCluster

		// check dm cluster node quota/spec
		if dmCluster == nil || dmCluster.WorkerNodeQuota == nil ||
			dmCluster.WorkerNodeQuota.NodeSpec == nil ||
			dmCluster.WorkerNodeQuota.NodeSpec.Cpu == 0 ||
			dmCluster.WorkerNodeQuota.NodeSpec.MemoryGi == 0 {
			logger.Error("dm cluster or its node quota/spec is nil or invalid", zap.Uint64("shared_pool_id", sharedPoolID))
			continue
		}

		if dmCluster.WorkerNodeQuota.NodeCount < uint32(len(tasks)) {
			// scale dm cluster worker nodes up as needed
			logger.Info("scaling dm cluster worker nodes", zap.Uint64("shared_pool_id", sharedPoolID),
				zap.String("dm_cluster_id", dmClusterID),
				zap.Int("current_worker_nodes", int(dmCluster.WorkerNodeQuota.NodeCount)),
				zap.Int("desired_worker_nodes", len(tasks)))
			err := ngRegionalClient.ScaleDMCluster(ctx, sharedPoolID, dmClusterID, &agent.ScaleDMClusterRequest{
				DmClusterId: dmClusterID,
				WorkerNodeQuota: &agent.NodeQuota{
					NodeCount: uint32(len(tasks)),
					// keep same spec as before
					NodeSpec: &agent.NodeSpec{
						Cpu:      dmCluster.WorkerNodeQuota.NodeSpec.Cpu,
						MemoryGi: dmCluster.WorkerNodeQuota.NodeSpec.MemoryGi,
					},
				},
			})
			if err != nil {
				logger.Error("failed to scale dm cluster worker nodes", zap.Uint64("shared_pool_id", sharedPoolID), zap.Error(err))
				continue
			}
		} else {
			logger.Info("dm cluster worker nodes are sufficient, no scaling needed", zap.Uint64("shared_pool_id", sharedPoolID),
				zap.Int("current_worker_nodes", int(dmCluster.WorkerNodeQuota.NodeCount)),
				zap.Int("desired_worker_nodes", len(tasks)))
		}
	}

	return dmClusterMap, nil
}

func buildPoolToTasksMap(
	logger *zap.Logger,
	tasks []*entity.PremiumDMTask,
	clusters []*tidbmgmtservicev1.Cluster,
) (map[uint64][]*entity.PremiumDMTask, error) {
	// Group tasks by resource pool
	poolToTasks := make(map[uint64][]*entity.PremiumDMTask)
	clusterToSharedPool := make(map[uint64]uint64) // clusterID -> sharedPoolID
	for _, cluster := range clusters {
		if cluster.GetNextGenCluster() == nil || cluster.GetNextGenCluster().GetSharedPoolId() == "" {
			return nil, fmt.Errorf("cluster %d does not have a valid shared pool", cluster.GetClusterId())
		}

		sharedPoolId, err := strconv.ParseUint(cluster.GetNextGenCluster().GetSharedPoolId(), 10, 64)
		if err != nil {
			return nil, fmt.Errorf("failed to parse shared pool id %s: %w", cluster.GetNextGenCluster().GetSharedPoolId(), err)
		}
		clusterToSharedPool[cluster.GetClusterId()] = sharedPoolId
	}

	// build poolToTasks map
	for _, task := range tasks {
		clusterID, err := strconv.ParseUint(task.TiDBID, 10, 64)
		if err != nil {
			return nil, fmt.Errorf("failed to parse cluster id %s: %w", task.TiDBID, err)
		}
		sharedPoolID, ok := clusterToSharedPool[clusterID]
		if !ok {
			logger.Warn("cluster not found in clusterToSharedPool map", zap.Uint64("cluster_id", clusterID))
			continue
		}
		if _, ok := poolToTasks[sharedPoolID]; !ok {
			poolToTasks[sharedPoolID] = make([]*entity.PremiumDMTask, 0)
		}
		poolToTasks[sharedPoolID] = append(poolToTasks[sharedPoolID], task)
	}

	return poolToTasks, nil
}

// buildTaskMigrateRulesFromBlockAllowRules generates DM TaskTableMigrateRule(s) from BlockAllowRules for a given DMSource.
// For DM, BlockAllowRules must be set first. If needed, MigrateRule can be configured additionally.
// The DM OpenAPI does not expose BlockAllowRules directly; instead, BlockAllowRules are generated from MigrateRule.
// If you only configure table_migrate_rule like:
//
//	[
//	  {
//	    "source": {"schema": "db", "source_name": "mysql-01", "table": "tb"},
//	    "target": {"schema": "db", "table": "tb"}
//	  }
//	]
//
// The generated BlockAllowRules will be: do-dbs: [], do-tables: ["db.tb"].
// This means all databases are included by default, but only the specified tables (db.tb) are allowed.
// However, this does not meet the requirement, because all databases will be synchronized except for the tables listed in do-tables.
// To ensure only the intended tables are synchronized, you must also add a rule like:
//
//	{
//	  "source": {"schema": "db", "source_name": "mysql-01"},
//	  "target": {"schema": "db"}
//	}
//
// This ensures the correct BlockAllowRules are generated, so only the specified tables in the specified database are synchronized.
func buildTaskMigrateRulesFromBlockAllowRules(s *entity.PremiumDMSource) []dmopenapi.TaskTableMigrateRule {
	taskMigrateRules := make([]dmopenapi.TaskTableMigrateRule, 0)
	schemaExists := make(map[string]struct{})
	for _, schema := range s.BlockAllowRules.DoDBs {
		migrateRule := dmopenapi.TaskTableMigrateRule{}
		migrateRule.Source.Schema = schema
		migrateRule.Source.SourceName = s.ID
		migrateRule.Target = &dmopenapi.TaskTableMigrateRuleTarget{
			Schema: pointer.To(schema),
		}
		taskMigrateRules = append(taskMigrateRules, migrateRule)
		schemaExists[schema] = struct{}{}
	}
	for _, table := range s.BlockAllowRules.DoTables {
		migrateRule := dmopenapi.TaskTableMigrateRule{}
		migrateRule.Source.SourceName = s.ID
		migrateRule.Source.Schema = table.Schema
		migrateRule.Source.Table = table.Table
		migrateRule.Target = &dmopenapi.TaskTableMigrateRuleTarget{
			Schema: pointer.To(table.Schema),
		}
		if migrateRule.Source.Table != "*" && migrateRule.Source.Table != "~.*" {
			migrateRule.Target.Table = pointer.To(table.Table)
		}
		taskMigrateRules = append(taskMigrateRules, migrateRule)
		if _, ok := schemaExists[table.Schema]; !ok {
			migrateRule := dmopenapi.TaskTableMigrateRule{}
			migrateRule.Source.Schema = table.Schema
			migrateRule.Source.SourceName = s.ID
			migrateRule.Target = &dmopenapi.TaskTableMigrateRuleTarget{
				Schema: pointer.To(table.Schema),
			}
			taskMigrateRules = append(taskMigrateRules, migrateRule)
			schemaExists[table.Schema] = struct{}{}
		}
	}
	return taskMigrateRules
}

// buildTaskMigrateRulesFromMigrationRules translates MigrationRules into DM TaskTableMigrateRule(s) for a given DMSource.
// Each MigrationRule already specifies how a source schema/table should be rewritten, so we convert it into a table-level migrate rule.
// Similar to BlockAllowRules handling, we also add a schema-level migrate rule per unique source schema to ensure DM generates the correct
// block-allow lists and only migrates the intended schemas/tables.
func buildTaskMigrateRulesFromMigrationRules(s *entity.PremiumDMSource) []dmopenapi.TaskTableMigrateRule {
	taskMigrateRules := make([]dmopenapi.TaskTableMigrateRule, 0)
	schemaExists := make(map[string]struct{})
	for _, rule := range s.MigrationRules {
		migrateRule := dmopenapi.TaskTableMigrateRule{}
		migrateRule.Source.SourceName = s.ID
		migrateRule.Source.Schema = rule.Source.SchemaPattern
		migrateRule.Source.Table = rule.Source.TablePattern
		migrateRule.Target = &dmopenapi.TaskTableMigrateRuleTarget{
			Schema: pointer.To(rule.Target.Schema),
			Table:  pointer.To(rule.Target.Table),
		}
		taskMigrateRules = append(taskMigrateRules, migrateRule)

		if _, ok := schemaExists[rule.Source.SchemaPattern]; !ok {
			migrateRule := dmopenapi.TaskTableMigrateRule{}
			migrateRule.Source.Schema = rule.Source.SchemaPattern
			migrateRule.Source.SourceName = s.ID
			migrateRule.Target = &dmopenapi.TaskTableMigrateRuleTarget{
				Schema: pointer.To(rule.Target.Schema),
			}
			taskMigrateRules = append(taskMigrateRules, migrateRule)
			schemaExists[rule.Source.SchemaPattern] = struct{}{}
		}
	}

	return taskMigrateRules
}

// buildTaskMode converts the entity.PremiumDMTaskMode to the corresponding dmopenapi.TaskMode for the DM API.
func buildTaskMode(mode entity.PremiumDMTaskMode) dmopenapi.TaskTaskMode {
	switch mode {
	case entity.PremiumDMTaskModeAll:
		return dmopenapi.TaskTaskModeAll
	case entity.PremiumDMTaskModeIncremental:
		return dmopenapi.TaskTaskModeIncremental
	default:
		return ""
	}
}