package server

import (
	"context"
	"fmt"

	"github.com/google/uuid"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	workflowaction "github.com/tidbcloud/cluster-service-ng/domain/workflow/action"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow/dag"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/executor"
	"github.com/tidbcloud/cluster-service-ng/pkg/tioperator"
	"github.com/tidbcloud/cluster-service-ng/pkg/utils"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/proto"
)

func (s *serverImpl) SetSharedPoolTiKVRemoteWorkers(
	ctx context.Context,
	sharedPoolID string,
	targetWorkers []entity.ComponentType,
) (*entity.SharedPool, error) {
	if sharedPoolID == "" {
		return nil, status.Errorf(codes.InvalidArgument, "sharedPool id is required")
	}
	// An empty target must never fall back to the legacy/create-time default -
	// entity.NormalizeEnabledTiKVRemoteWorkers treats empty as "default to
	// both workers", which is the right behavior for internal callers but
	// wrong for this RPC's explicit contract.
	if len(targetWorkers) == 0 {
		return nil, status.Errorf(codes.InvalidArgument, "enabled_tikv_remote_workers must not be empty")
	}
	normalizedTarget, err := entity.NormalizeEnabledTiKVRemoteWorkers(targetWorkers)
	if err != nil {
		return nil, status.Errorf(codes.InvalidArgument, "%s", err)
	}
	targetKey, err := entity.EnabledTiKVRemoteWorkersToString(normalizedTarget)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "failed to encode target tikv remote workers: %s", err)
	}
	enableCoprocessor := entity.HasEnabledTiKVRemoteWorker(normalizedTarget, apiv1.ComponentType_COPROCESSOR_WORKER)
	// Keep this token outside the transaction closure so an attempted submission
	// and its rollback share the same exec ID. It intentionally does not depend
	// on the request trace: a client may reuse a trace across independent
	// switches, while completed executor tasks must never suppress a new switch.
	submissionToken := uuid.NewString()

	var result *entity.SharedPool
	var dagExecID string
	var dagSubmitted bool
	if err := s.transactioner.Transaction(ctx, func(ctx context.Context) error {
		unlockFunc, err := s.lock(ctx, sharedPoolID)
		if err != nil {
			return err
		}
		defer func() {
			_ = unlockFunc()
		}()

		sharedPool, err := s.sharedPoolRepo.Load(ctx, sharedPoolID, false)
		if err != nil {
			if status.Code(err) != codes.NotFound {
				return status.Errorf(codes.Internal, "failed to load sharedPool: %s", err)
			}
			return status.Errorf(codes.NotFound, "sharedPool %s not found", sharedPoolID)
		}
		if err := guardSharedPoolMutation(sharedPool, sharedPoolMutationBlockedWhilePaused); err != nil {
			return err
		}

		if err := s.mergeSharedPoolStatusFromDagTask(ctx, sharedPool); err != nil {
			return status.Errorf(codes.Internal, "failed to merge sharedPool status from dag task: %s", err)
		}
		if sharedPool.GetCapacityMode() != apiv1.CapacityMode_RCU_DRIVEN {
			return status.Errorf(codes.FailedPrecondition, "tikv remote worker routing switch is only supported for RCU-driven shared pools")
		}

		runningDags, err := s.getRunningDagIns(ctx, tikvRemoteWorkersDAGPlans(), []string{sharedPoolID}, nil, nil, nil)
		if err != nil {
			return status.Errorf(codes.Internal, "failed to get running dag instance: %s", err)
		}
		if len(runningDags) > 0 {
			sameTarget, err := hasSameTiKVRemoteWorkersTarget(runningDags, targetKey)
			if err != nil {
				return status.Errorf(codes.Internal, "failed to inspect running tikv remote worker dag: %s", err)
			}
			if sameTarget {
				sharedPool.EnabledTikvRemoteWorkers = normalizedTarget
				sharedPool.Phase = apiv1.SharedPool_MODIFYING
				markComponentsPhase(sharedPool, tikvRemoteWorkersAffectedComponentTypes(), apiv1.Component_MODIFYING)
				result = sharedPool
				return nil
			}
			return status.Errorf(codes.FailedPrecondition, "another tikv remote worker routing switch is running for shared pool %s", sharedPoolID)
		}
		if sharedPool.GetPhase() != apiv1.SharedPool_AVAILABLE {
			return status.Errorf(codes.FailedPrecondition, "shared pool is not available, current phase: %s", sharedPool.GetPhase().String())
		}

		defaultComponentMap := sharedPoolDefaultComponentMap(sharedPool)
		tikvComp, ok := defaultComponentMap[apiv1.ComponentType_TIKV]
		if !ok || tikvComp == nil {
			return status.Errorf(codes.FailedPrecondition, "tikv component not found for shared pool %s", sharedPoolID)
		}

		stable, err := s.isTiKVRemoteWorkersStableInTarget(ctx, sharedPool, tikvComp, normalizedTarget, targetKey, enableCoprocessor)
		if err != nil {
			return status.Errorf(codes.Internal, "failed to check tikv remote worker rollout status: %s", err)
		}
		if stable {
			sharedPool.EnabledTikvRemoteWorkers = normalizedTarget
			if err := s.sharedPoolRepo.Save(ctx, sharedPool); err != nil {
				return fmt.Errorf("failed to save sharedPool meta: %w", err)
			}
			sharedPool.Phase = apiv1.SharedPool_AVAILABLE
			markComponentsPhase(sharedPool, tikvRemoteWorkersAffectedComponentTypes(), apiv1.Component_AVAILABLE)
			result = sharedPool
			return nil
		}

		if !enableCoprocessor {
			// Only gate on default worker readiness when a disable is actually
			// about to run - an already-stable idempotent call must not be
			// rejected by a transient readiness flap on the target we're not
			// even touching. missing/zero/unready are distinct failure modes:
			// missing and zero are bookkeeping-level checks against our own
			// NodeQuota (not the live CR), unready is the live-CR check.
			defaultWorkerComp, ok := defaultComponentMap[apiv1.ComponentType_TIKV_WORKER]
			if !ok || defaultWorkerComp == nil {
				return status.Errorf(codes.FailedPrecondition, "default tikv worker component not found for shared pool %s", sharedPoolID)
			}
			if defaultWorkerComp.GetNodeQuota().GetNodeCount() == 0 {
				return status.Errorf(codes.FailedPrecondition, "default tikv worker node count is zero, cannot disable coprocessor worker routing")
			}
			readiness, err := workflowaction.CheckTiKVWorkerGroupReadiness(ctx, s.dataplaneManager, sharedPool, defaultWorkerComp)
			if readiness == workflowaction.TiKVWorkerGroupCheckFailed {
				return status.Errorf(codes.Internal, "failed to check default tikv worker readiness: %s", err)
			}
			if readiness != workflowaction.TiKVWorkerGroupReady {
				return status.Errorf(codes.FailedPrecondition, "default tikv worker is not ready, cannot disable coprocessor worker routing")
			}
		} else if _, ok := defaultComponentMap[apiv1.ComponentType_TIKV_WORKER]; !ok {
			return status.Errorf(codes.FailedPrecondition, "default tikv worker component not found for shared pool %s", sharedPoolID)
		}

		tempSharedPool, err := s.buildTiKVRemoteWorkersTargetSharedPool(ctx, sharedPool, normalizedTarget, enableCoprocessor)
		if err != nil {
			return err
		}

		if enableCoprocessor {
			if err := persistTiKVRemoteWorkersCoprocessorMetadata(sharedPool, tempSharedPool); err != nil {
				return err
			}
		}
		sharedPool.EnabledTikvRemoteWorkers = normalizedTarget
		if err := s.sharedPoolRepo.Save(ctx, sharedPool); err != nil {
			return fmt.Errorf("failed to save sharedPool meta: %w", err)
		}

		args, err := buildTiKVRemoteWorkersArgs(ctx, tempSharedPool, targetKey)
		if err != nil {
			return err
		}
		planName := tikvRemoteWorkersPlanForTarget(enableCoprocessor)
		dagExecID = utils.GeneratedDagExecId(ctx, fmt.Sprintf("set-tikv-remote-workers-%s-%s-%s", sharedPoolID, planName, utils.FnvHash(submissionToken)))
		if err := s.dagExecutor.Submit(ctx, planName, args, dagExecID); err != nil {
			return status.Errorf(codes.Internal, "failed to submit tikv remote workers dag: %s", err)
		}
		dagSubmitted = true

		result = tempSharedPool
		return nil
	}); err != nil {
		if dagSubmitted && dagExecID != "" {
			if abortErr := s.dagExecutor.Abort(ctx, dagExecID); abortErr != nil {
				tilog.Error(ctx, "failed to abort tikv remote workers dag after transaction failure",
					zap.String("dagExecID", dagExecID),
					zap.Error(abortErr))
			}
		}
		return nil, err
	}

	return result, nil
}

// persistTiKVRemoteWorkersCoprocessorMetadata copies the enable snapshot's
// generated coprocessor metadata into the persisted shared-pool entity before
// the DAG is submitted. Running-DAG overlays are transient, so leaving this
// component only in the serialized DAG snapshot would make later component
// operations reject the successfully enabled worker as missing.
func persistTiKVRemoteWorkersCoprocessorMetadata(sharedPool, snapshot *entity.SharedPool) error {
	coprocessorComp := sharedPoolDefaultComponentMap(snapshot)[apiv1.ComponentType_COPROCESSOR_WORKER]
	if coprocessorComp == nil {
		return status.Errorf(codes.Internal, "coprocessor worker component missing from enable snapshot for shared pool %s", sharedPool.GetId())
	}

	for i, persistedComp := range sharedPool.ComponentGroups {
		if persistedComp.GetComponentType() == apiv1.ComponentType_COPROCESSOR_WORKER {
			sharedPool.ComponentGroups[i] = proto.Clone(coprocessorComp).(*entity.Component)
			return nil
		}
	}

	sharedPool.ComponentGroups = append(sharedPool.ComponentGroups, proto.Clone(coprocessorComp).(*entity.Component))
	return nil
}

// isTiKVRemoteWorkersStableInTarget reports whether the requested target is a
// genuine no-op: the persisted desired state must already equal targetKey
// AND live dataplane state must already be converged. Persisted state alone
// is not enough - a prior switch may have persisted the target metadata but
// failed partway through rollout. Live state alone is not enough either - an
// unset/legacy persisted field normalizes to "both workers enabled", which
// can coincidentally resolve to the same TiKV-side routing as an explicit
// disable target, and external drift can make live match a target that was
// never actually requested. Requiring both means a persisted-target mismatch
// always goes through a real DAG submission (whose actions are idempotent
// against an already-converged CR), instead of silently rewriting metadata
// over unreconciled state.
func (s *serverImpl) isTiKVRemoteWorkersStableInTarget(
	ctx context.Context,
	sharedPool *entity.SharedPool,
	tikvComp *entity.Component,
	normalizedTarget []entity.ComponentType,
	targetKey string,
	enableCoprocessor bool,
) (bool, error) {
	persistedKey, err := entity.EnabledTiKVRemoteWorkersToString(sharedPool.GetEnabledTikvRemoteWorkers())
	if err != nil {
		return false, err
	}
	if persistedKey != targetKey {
		return false, nil
	}

	expectedTarget, err := tikvRemoteWorkerTargetForNormalized(sharedPool, normalizedTarget)
	if err != nil {
		return false, err
	}

	rolloutStatus, _, err := workflowaction.CheckTiKVGroupRemoteWorkerRollout(ctx, s.dataplaneManager, sharedPool, tikvComp, expectedTarget)
	if rolloutStatus == workflowaction.TiKVGroupRolloutCheckFailed {
		return false, err
	}
	if rolloutStatus != workflowaction.TiKVGroupRolloutComplete {
		return false, nil
	}

	coprocessorComp := sharedPoolDefaultComponentMap(sharedPool)[apiv1.ComponentType_COPROCESSOR_WORKER]
	if coprocessorComp == nil {
		coprocessorComp = &entity.Component{
			Id:            defaultSharedPoolComponentID(sharedPool),
			ComponentType: apiv1.ComponentType_COPROCESSOR_WORKER,
		}
	}
	readiness, err := workflowaction.CheckTiKVWorkerGroupReadiness(ctx, s.dataplaneManager, sharedPool, coprocessorComp)
	if readiness == workflowaction.TiKVWorkerGroupCheckFailed {
		return false, err
	}
	if enableCoprocessor {
		return readiness == workflowaction.TiKVWorkerGroupReady, nil
	}
	// Disable is only stable once the dedicated coprocessor CR is actually
	// gone - if a prior DAG rerouted TiKV but failed before deleting the CR,
	// Ready/NotReady/InvalidReplicas all mean cleanup is still outstanding,
	// so only NotFound counts as converged.
	return readiness == workflowaction.TiKVWorkerGroupNotFound, nil
}

// guardTiKVRemoteWorkersTopologyStable blocks other shared-pool mutations
// (quota, scale, HPA, logical-cluster create/update) from racing an
// in-flight tikv-remote-worker routing switch. It fails if a
// tikv-remote-worker enable/disable DAG is currently running for sharedPool,
// or if the persisted EnabledTikvRemoteWorkers target has not yet converged
// in the live topology - reusing the same running-dag and
// isTiKVRemoteWorkersStableInTarget checks SetSharedPoolTiKVRemoteWorkers
// itself relies on, so the two entry points can never disagree about what
// "stable" means.
func (s *serverImpl) guardTiKVRemoteWorkersTopologyStable(ctx context.Context, sharedPool *entity.SharedPool) error {
	if sharedPool.GetCapacityMode() != apiv1.CapacityMode_RCU_DRIVEN {
		// tikv-remote-worker desired/live semantics only exist for RCU-driven
		// shared pools (SetSharedPoolTiKVRemoteWorkers itself rejects any other
		// capacity mode) - topology-driven/dedicated pools have no such
		// routing switch to guard against.
		return nil
	}

	runningDags, err := s.getRunningDagIns(ctx, tikvRemoteWorkersDAGPlans(), []string{sharedPool.GetId()}, nil, nil, nil)
	if err != nil {
		return status.Errorf(codes.Internal, "failed to get running dag instance: %s", err)
	}
	if len(runningDags) > 0 {
		return status.Errorf(codes.FailedPrecondition, "a tikv remote worker routing switch is running for shared pool %s", sharedPool.GetId())
	}

	if sharedPool.GetPhase() == apiv1.SharedPool_CREATING {
		// Shared pool creation legitimately populates TIKV component metadata
		// (via generatedSharedPoolComponent(force=true)) before the live CR
		// exists. There is no established topology to have drifted from yet,
		// so only the running-plan race check above applies during CREATING;
		// the live-drift check below only makes sense once the shared pool
		// has actually reached a terminal (non-CREATING) phase.
		return nil
	}

	tikvComp, ok := sharedPoolDefaultComponentMap(sharedPool)[apiv1.ComponentType_TIKV]
	if !ok || tikvComp == nil {
		// No TIKV component yet (e.g. shared pool still being created) - there
		// is no tikv-remote-worker topology to drift-check.
		return nil
	}

	normalizedTarget, err := entity.NormalizeEnabledTiKVRemoteWorkers(sharedPool.GetEnabledTikvRemoteWorkers())
	if err != nil {
		return status.Errorf(codes.Internal, "failed to normalize enabled tikv remote workers: %s", err)
	}
	targetKey, err := entity.EnabledTiKVRemoteWorkersToString(normalizedTarget)
	if err != nil {
		return status.Errorf(codes.Internal, "failed to encode enabled tikv remote workers: %s", err)
	}
	enableCoprocessor := entity.HasEnabledTiKVRemoteWorker(normalizedTarget, apiv1.ComponentType_COPROCESSOR_WORKER)

	stable, err := s.isTiKVRemoteWorkersStableInTarget(ctx, sharedPool, tikvComp, normalizedTarget, targetKey, enableCoprocessor)
	if err != nil {
		return status.Errorf(codes.Internal, "failed to check tikv remote worker rollout status: %s", err)
	}
	if !stable {
		return status.Errorf(codes.FailedPrecondition, "tikv remote worker topology has not converged for shared pool %s", sharedPool.GetId())
	}
	return nil
}

// isTiKVRemoteWorkerGuardedComponentType reports whether componentType is one
// of the components a tikv-remote-worker routing switch can affect - guard
// callers should only gate on these, not on unrelated component types (e.g.
// PD or TIDB quota changes must not be blocked by this guard).
func isTiKVRemoteWorkerGuardedComponentType(componentType entity.ComponentType) bool {
	switch componentType {
	case apiv1.ComponentType_TIKV, apiv1.ComponentType_TIKV_WORKER, apiv1.ComponentType_COPROCESSOR_WORKER:
		return true
	default:
		return false
	}
}

// lockTiKVRemoteWorkerGuardedMutation keeps the stability check and the
// following direct HPA patch in the same shared-pool critical section. The
// remote-worker switch holds this lock while it observes/submits its DAG, so a
// caller cannot pass a stale topology check and patch an HPA concurrently.
func (s *serverImpl) lockTiKVRemoteWorkerGuardedMutation(
	ctx context.Context,
	sharedPoolID string,
	componentType entity.ComponentType,
) (func(), error) {
	if !isTiKVRemoteWorkerGuardedComponentType(componentType) {
		return func() {}, nil
	}

	unlockFunc, err := s.lock(ctx, sharedPoolID)
	if err != nil {
		return nil, err
	}
	return func() {
		if err := unlockFunc(); err != nil {
			tilog.Error(ctx, "failed to unlock sharedPool after HPA mutation",
				zap.String("sharedPoolId", sharedPoolID), zap.Error(err))
		}
	}, nil
}

// tikvRemoteWorkerTargetForNormalized resolves the TiKVGroup remoteWorkers
// target for an arbitrary candidate EnabledTikvRemoteWorkers set, without
// mutating sharedPool.
func tikvRemoteWorkerTargetForNormalized(sharedPool *entity.SharedPool, normalizedTarget []entity.ComponentType) (tioperator.TiKVRemoteWorkerTarget, error) {
	tempSharedPool := &entity.SharedPool{}
	proto.Merge(&tempSharedPool.SharedPool, &sharedPool.SharedPool)
	tempSharedPool.EnabledTikvRemoteWorkers = normalizedTarget

	resolved, err := entity.ResolveTiKVRemoteWorkerGroups(tempSharedPool)
	if err != nil {
		return tioperator.TiKVRemoteWorkerTarget{}, err
	}
	return tioperator.TiKVRemoteWorkerTarget{
		Worker:      resolved.Worker,
		Coprocessor: resolved.Coprocessor,
	}, nil
}

// buildTiKVRemoteWorkersTargetSharedPool builds the frozen DAG-args snapshot
// for a real enable/disable rollout. TIKV is always flipped to MODIFYING so
// UpdateComponentGroupAction does not skip it; COPROCESSOR_WORKER is either
// (re)generated via the shared default-worker generator (enable) or marked
// DELETING (disable, matching CleanUpCertAction's and
// WaitCoprocessorWorkerGoneAction's Phase/presence contract).
func (s *serverImpl) buildTiKVRemoteWorkersTargetSharedPool(
	ctx context.Context,
	sharedPool *entity.SharedPool,
	normalizedTarget []entity.ComponentType,
	enableCoprocessor bool,
) (*entity.SharedPool, error) {
	tempSharedPool := &entity.SharedPool{
		TemplateVersion:           sharedPool.TemplateVersion,
		TenantId:                  sharedPool.TenantId,
		DeployType:                sharedPool.DeployType,
		ConfigCenterMetaProfileId: sharedPool.ConfigCenterMetaProfileId,
		CreatedAt:                 sharedPool.CreatedAt,
		DeletedAt:                 sharedPool.DeletedAt,
	}
	proto.Merge(&tempSharedPool.SharedPool, &sharedPool.SharedPool)
	tempSharedPool.EnabledTikvRemoteWorkers = normalizedTarget
	tempSharedPool.Phase = apiv1.SharedPool_MODIFYING

	defaultComponentMap := sharedPoolDefaultComponentMap(tempSharedPool)
	tikvComp, ok := defaultComponentMap[apiv1.ComponentType_TIKV]
	if !ok || tikvComp == nil {
		return nil, status.Errorf(codes.FailedPrecondition, "tikv component not found for shared pool %s", sharedPool.GetId())
	}
	tikvComp.Phase = apiv1.Component_MODIFYING

	if !enableCoprocessor {
		coprocessorComp, ok := defaultComponentMap[apiv1.ComponentType_COPROCESSOR_WORKER]
		if !ok || coprocessorComp == nil {
			// The live CR is already gone, so the repository has rebuilt its snapshot
			// without the component metadata. Keep a deleting snapshot entry so the
			// repair DAG can still render the deletion action and wait idempotently.
			coprocessorComp = &entity.Component{
				Id:            defaultSharedPoolComponentID(tempSharedPool),
				ComponentType: apiv1.ComponentType_COPROCESSOR_WORKER,
				NodeQuota:     &apiv1.NodeQuota{},
			}
			tempSharedPool.ComponentGroups = append(tempSharedPool.ComponentGroups, coprocessorComp)
		}
		coprocessorComp.Phase = apiv1.Component_DELETING
		return tempSharedPool, nil
	}

	ruConfig, err := s.ruConfigRepo.GetRuConfig(ctx, sharedPool.GetId(), apiv1.RuConfig_SHARED_POOL)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "failed to get shared pool ru config: %s", err)
	}
	nodeSpecMap, err := s.genNodeSpecMap(ctx, sharedPool)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "failed to get node spec map: %s", err)
	}
	ruTotal, err := s.getSharedPoolRequestUnitTotal(ctx, sharedPool.GetId())
	if err != nil {
		return nil, err
	}

	coprocessorComp, ok := defaultComponentMap[apiv1.ComponentType_COPROCESSOR_WORKER]
	if !ok || coprocessorComp == nil {
		coprocessorComp = &entity.Component{
			Id:            defaultSharedPoolComponentID(tempSharedPool),
			ComponentType: apiv1.ComponentType_COPROCESSOR_WORKER,
		}
		tempSharedPool.ComponentGroups = append(tempSharedPool.ComponentGroups, coprocessorComp)
	}
	// Always mark CREATING on enable, whether coprocessorComp is newly minted or
	// a retained-metadata reuse from a prior disable (task doc allows metadata to
	// survive disable). genDefaultTiKVWorkerComponent only flips Phase to
	// MODIFYING when the computed quota differs from the existing one, so an
	// unchanged-quota re-enable of a retained AVAILABLE component would
	// otherwise leave Phase untouched and violate the enable snapshot contract
	// (COPROCESSOR_WORKER=CREATING), causing ApplyComponentGroupAction to skip
	// recreating the already-deleted coprocessor CR.
	coprocessorComp.Phase = apiv1.Component_CREATING
	if err := s.genDefaultTiKVWorkerComponent(ctx, tempSharedPool, coprocessorComp, ruTotal, nodeSpecMap, ruConfig, nil, false); err != nil {
		return nil, err
	}
	return tempSharedPool, nil
}

func buildTiKVRemoteWorkersArgs(
	ctx context.Context,
	sharedPool *entity.SharedPool,
	targetKey string,
) (actions.ActionArgs, error) {
	tikvComp, ok := sharedPoolDefaultComponentMap(sharedPool)[apiv1.ComponentType_TIKV]
	if !ok || tikvComp == nil {
		return nil, status.Errorf(codes.Internal, "tikv component not found for shared pool %s", sharedPool.GetId())
	}

	sharedPoolStr, err := protojson.Marshal(sharedPool)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "failed to marshal sharedPool: %s", err)
	}

	return actions.ActionArgs{
		workflow.SharedPoolIdKey: actions.ActionArgString(sharedPool.GetId()),
		workflow.SharedPoolKey:   actions.ActionArgString(string(sharedPoolStr)),
		workflow.ComponentIdsKey: actions.ActionArgString(defaultSharedPoolComponentID(sharedPool)),
		// ComponentTypesKey/AffectedComponentTypesKey are matched by
		// CheckIsTargetComponent/CheckIsAffectedComponent against
		// entity.GetComponentTypeName(comp), i.e. runtime type names such as
		// "tikv"/"coprocessor-tikv-worker" - not proto enum .String() names
		// like "TIKV"/"COPROCESSOR_WORKER". Using enum names here silently
		// filters out every component and the DAG never mutates anything.
		workflow.ComponentTypesKey:             actions.ActionArgString(entity.CoprocessorWorkerType),
		workflow.AffectedComponentTypesKey:     actions.ActionArgString(entity.GetComponentTypeName(tikvComp) + "," + entity.CoprocessorWorkerType),
		workflow.EnabledTiKVRemoteWorkersKey:   actions.ActionArgString(targetKey),
		workflow.MeteringTypeKey:               actions.ActionArgString(workflow.MeteringTypeValueLogicMetaBySharedPool),
		entity.GetUpdateComponentKey(tikvComp): actions.ActionArgString(string(tioperator.OpsUpdateTiKVRemoteWorkers)),
		utils.TraceIdKey:                       utils.GetTraceId(ctx),
	}, nil
}

// hasSameTiKVRemoteWorkersTarget reports whether every running remote-worker
// DAG already targets targetKey. A mixed state (e.g. enable and disable both
// in flight, which should never happen but must not be trusted blindly) has
// to be treated as a conflict even if one of them happens to match - so this
// requires unanimous agreement, not merely one match.
func hasSameTiKVRemoteWorkersTarget(runningDags executor.ExecTaskIdStatus, targetKey string) (bool, error) {
	if len(runningDags) == 0 {
		return false, nil
	}
	for _, runningDag := range runningDags {
		key, err := runningDag.ExecTask.Args.GetString(workflow.EnabledTiKVRemoteWorkersKey)
		if err != nil {
			return false, err
		}
		if key != targetKey {
			return false, nil
		}
	}
	return true, nil
}

func tikvRemoteWorkersDAGPlans() []string {
	return []string{
		dag.PlanEnableCoprocessorWorker,
		dag.PlanDisableCoprocessorWorker,
	}
}

func tikvRemoteWorkersPlanForTarget(enableCoprocessor bool) string {
	if enableCoprocessor {
		return dag.PlanEnableCoprocessorWorker
	}
	return dag.PlanDisableCoprocessorWorker
}

func tikvRemoteWorkersAffectedComponentTypes() []entity.ComponentType {
	return []entity.ComponentType{apiv1.ComponentType_TIKV, apiv1.ComponentType_COPROCESSOR_WORKER}
}