package server

import (
	"context"
	"errors"
	"regexp"
	"sort"

	mysqlDriver "github.com/go-sql-driver/mysql"
	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"
	braction "github.com/tidbcloud/cluster-service-ng/domain/workflow/action/br"
	"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"
	"gorm.io/gorm"
)

var pauseResumeOperationIDPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._~-]{0,127}$`)

var errPauseResumeOperationAlreadyCreated = errors.New("pause/resume operation already created")

const (
	pauseValidationInvalidLifecycleState = "INVALID_LIFECYCLE_STATE"
	pauseValidationDependencyUnavailable = "DEPENDENCY_UNAVAILABLE"
	pauseValidationUpgradeInProgress     = "UPGRADE_IN_PROGRESS"
	pauseValidationBackupInProgress      = "BACKUP_IN_PROGRESS"
	pauseValidationOperationInProgress   = "OPERATION_IN_PROGRESS"
)

var pauseValidationCodeOrder = map[string]int{
	pauseValidationInvalidLifecycleState: 0,
	pauseValidationDependencyUnavailable: 1,
	pauseValidationUpgradeInProgress:     2,
	pauseValidationBackupInProgress:      3,
	pauseValidationOperationInProgress:   4,
}

type pauseValidationFailure struct {
	code    string
	message string
	err     error
}

type pauseValidationCollector struct {
	collectAll bool
	failures   []*pauseValidationFailure
	byCode     map[string]*pauseValidationFailure
}

func newPauseValidationCollector(collectAll bool) *pauseValidationCollector {
	return &pauseValidationCollector{collectAll: collectAll, byCode: make(map[string]*pauseValidationFailure)}
}

func (c *pauseValidationCollector) add(failure *pauseValidationFailure) bool {
	if failure == nil {
		return false
	}
	if existing, ok := c.byCode[failure.code]; ok {
		if failure.message != "" && failure.message != existing.message {
			existing.message += "; " + failure.message
		}
	} else {
		c.byCode[failure.code] = failure
		c.failures = append(c.failures, failure)
	}
	return !c.collectAll
}

func (c *pauseValidationCollector) firstError() error {
	if len(c.failures) == 0 {
		return nil
	}
	return c.failures[0].err
}

func (c *pauseValidationCollector) responseFailures() []*entity.SharedPoolPauseValidationFailure {
	sort.SliceStable(c.failures, func(i, j int) bool {
		return pauseValidationCodeRank(c.failures[i].code) < pauseValidationCodeRank(c.failures[j].code)
	})
	result := make([]*entity.SharedPoolPauseValidationFailure, 0, len(c.failures))
	for _, failure := range c.failures {
		result = append(result, &entity.SharedPoolPauseValidationFailure{Code: failure.code, Message: failure.message})
	}
	return result
}

func pauseValidationCodeRank(code string) int {
	if rank, ok := pauseValidationCodeOrder[code]; ok {
		return rank
	}
	return len(pauseValidationCodeOrder)
}

func newPauseValidationFailure(code, message string, err error) *pauseValidationFailure {
	return &pauseValidationFailure{code: code, message: message, err: err}
}

func (s *serverImpl) ValidateSharedPoolPause(ctx context.Context, sharedPoolID string) (*entity.ValidateSharedPoolPauseResponse, error) {
	if sharedPoolID == "" {
		return nil, status.Error(codes.InvalidArgument, "shared_pool_id is required")
	}
	collector, err := s.evaluatePause(ctx, sharedPoolID, "", true)
	if err != nil {
		return nil, err
	}
	return &entity.ValidateSharedPoolPauseResponse{Failures: collector.responseFailures()}, nil
}

func (s *serverImpl) PauseSharedPool(ctx context.Context, sharedPoolID, operationID string) (*entity.SharedPool, error) {
	return s.acceptPauseResume(ctx, sharedPoolID, operationID, apiv1.PauseResumeOperation_PAUSE)
}

func (s *serverImpl) ResumeSharedPool(ctx context.Context, sharedPoolID, operationID string) (*entity.SharedPool, error) {
	return s.acceptPauseResume(ctx, sharedPoolID, operationID, apiv1.PauseResumeOperation_RESUME)
}

func (s *serverImpl) GetPauseResumeOperation(ctx context.Context, sharedPoolID, operationID string) (*entity.PauseResumeOperation, error) {
	if sharedPoolID == "" || !pauseResumeOperationIDPattern.MatchString(operationID) {
		return nil, status.Error(codes.InvalidArgument, "shared_pool_id and a valid operation_id are required")
	}
	operation, err := s.pauseResumeOperationRepo.LoadByID(ctx, sharedPoolID, operationID)
	if errors.Is(err, gorm.ErrRecordNotFound) {
		return nil, status.Error(codes.NotFound, "pause/resume operation not found")
	}
	if err != nil {
		return nil, resourcePoolStatusError(codes.Unavailable, reasonPauseResumeDependencyUnavailable, "failed to load pause/resume operation")
	}
	return operation, nil
}

func (s *serverImpl) acceptPauseResume(ctx context.Context, sharedPoolID, operationID string, operationType apiv1.PauseResumeOperation_Type) (*entity.SharedPool, error) {
	if sharedPoolID == "" || !pauseResumeOperationIDPattern.MatchString(operationID) {
		return nil, status.Error(codes.InvalidArgument, "shared_pool_id and a valid operation_id are required")
	}
	if replay, found, err := s.loadPauseResumeReplay(ctx, sharedPoolID, operationID, operationType); found || err != nil {
		return replay, err
	}
	if operationType == apiv1.PauseResumeOperation_PAUSE {
		if err := s.validatePause(ctx, sharedPoolID, operationID); err != nil {
			return nil, err
		}
	} else if err := s.validateResume(ctx, sharedPoolID, operationID); err != nil {
		return nil, err
	}

	var accepted *entity.SharedPool
	execID := workflow.PauseResumeExecID(sharedPoolID, operationID, operationType.String())
	args := actions.ActionArgs{
		workflow.SharedPoolIdKey:             actions.ActionArgString(sharedPoolID),
		workflow.PauseResumeOperationIdKey:   actions.ActionArgString(operationID),
		workflow.PauseResumeOperationTypeKey: actions.ActionArgString(operationType.String()),
		utils.TraceIdKey:                     utils.GetTraceId(ctx),
	}
	plan := dag.PlanPauseSharedPool
	if operationType == apiv1.PauseResumeOperation_RESUME {
		plan = dag.PlanResumeSharedPool
	}
	dagSubmitted := false
	// Acquire the advisory lock outside the transaction and release it only
	// after commit. Releasing it from inside the transaction callback would
	// create a window where the next owner observes the pre-commit Pool state.
	unlock, err := s.lock(ctx, sharedPoolID)
	if err != nil {
		return nil, err
	}
	lockHeld := true
	defer func() {
		if lockHeld {
			_ = unlock()
		}
	}()
	err = s.transactioner.Transaction(ctx, func(txCtx context.Context) error {
		if replay, found, replayErr := s.loadPauseResumeReplay(txCtx, sharedPoolID, operationID, operationType); found || replayErr != nil {
			accepted = replay
			return replayErr
		}
		sharedPool, err := s.loadPauseResumeSharedPool(txCtx, sharedPoolID)
		if err != nil {
			return err
		}
		if err := s.validatePauseResumeLifecycle(txCtx, sharedPool, operationType, operationID); err != nil {
			return err
		}
		ignoreCreateConflicts := operationType == apiv1.PauseResumeOperation_RESUME &&
			sharedPool.PauseResumeState == entity.PauseResumeStateResuming
		if err := s.checkPauseResumeConflictingDAGs(txCtx, sharedPoolID, ignoreCreateConflicts); err != nil {
			return err
		}
		stage := apiv1.PauseResumeOperation_PRECHECK
		state := entity.PauseResumeStatePausing
		desired := false
		if operationType == apiv1.PauseResumeOperation_RESUME {
			stage = apiv1.PauseResumeOperation_APPLY_RESUME
			state = entity.PauseResumeStateResuming
			desired = true
		}
		operation := &entity.PauseResumeOperation{SharedPoolId: sharedPoolID, OperationId: operationID, Type: operationType, Status: apiv1.PauseResumeOperation_RUNNING, Stage: stage}
		if err := s.pauseResumeOperationRepo.Create(txCtx, operation); err != nil {
			if isPauseResumeDuplicateKeyError(err) {
				return errPauseResumeOperationAlreadyCreated
			}
			return err
		}
		// A new-ID retry keeps the transition's existing desired value. Resetting
		// it to the initial value could reverse an Operator patch that the failed
		// predecessor had already issued; the replayed DAG converges it instead.
		if sharedPool.PauseResumeState != entity.PauseResumeStatePausing && sharedPool.PauseResumeState != entity.PauseResumeStateResuming {
			if err := s.sharedPoolRepo.UpdatePauseResumeState(txCtx, sharedPoolID, state, desired); err != nil {
				return err
			}
			sharedPool.PauseResumeState = state
			sharedPool.SuspendComputeDesired = desired
		}
		if err := projectSharedPoolPhase(sharedPool); err != nil {
			return err
		}
		accepted = sharedPool
		// Follow the existing operation pattern: persist metadata and submit the
		// DAG in one transaction callback. The executor may start before commit;
		// stage updates remain retryable until the journal is visible, and every
		// external write is additionally fenced by the committed owner record.
		if err := s.dagExecutor.Submit(txCtx, plan, args, execID); err != nil {
			return status.Errorf(codes.Internal, "failed to submit pause/resume DAG: %s", err)
		}
		dagSubmitted = true
		return nil
	})
	if err != nil {
		// Submit and the MySQL commit do not share an atomic commit point. Match
		// existing workflows by best-effort aborting a submitted DAG before
		// releasing the Pool lock. Owner fencing prevents an orphan DAG with no
		// committed owner from issuing external writes.
		if dagSubmitted {
			if abortErr := s.dagExecutor.Abort(ctx, execID); abortErr != nil {
				tilog.Error(ctx, "failed to abort pause/resume DAG after transaction failure", zap.String("execID", execID), zap.Error(abortErr))
			}
		}
	}
	unlockErr := unlock()
	lockHeld = false
	if err != nil {
		if errors.Is(err, errPauseResumeOperationAlreadyCreated) {
			replay, found, replayErr := s.loadPauseResumeReplay(ctx, sharedPoolID, operationID, operationType)
			if !found && replayErr == nil {
				replayErr = resourcePoolStatusError(codes.Unavailable, reasonPauseResumeDependencyUnavailable, "failed to load pause/resume operation after duplicate key")
			}
			return replay, replayErr
		}
		return nil, err
	}
	if unlockErr != nil {
		return nil, unlockErr
	}
	return accepted, nil
}

func isPauseResumeDuplicateKeyError(err error) bool {
	if errors.Is(err, gorm.ErrDuplicatedKey) {
		return true
	}
	var mysqlErr *mysqlDriver.MySQLError
	return errors.As(err, &mysqlErr) && mysqlErr.Number == 1062
}

func (s *serverImpl) loadPauseResumeReplay(ctx context.Context, sharedPoolID, operationID string, operationType apiv1.PauseResumeOperation_Type) (*entity.SharedPool, bool, error) {
	operation, err := s.pauseResumeOperationRepo.LoadByID(ctx, sharedPoolID, operationID)
	if errors.Is(err, gorm.ErrRecordNotFound) {
		return nil, false, nil
	}
	if err != nil {
		return nil, false, resourcePoolStatusError(codes.Unavailable, reasonPauseResumeDependencyUnavailable, "failed to load pause/resume operation")
	}
	if operation.GetType() != operationType {
		return nil, true, status.Error(codes.AlreadyExists, "operation_id was already used for a different operation type")
	}
	sharedPool, err := s.GetSharedPool(ctx, sharedPoolID)
	if err != nil {
		if status.Code(err) == codes.NotFound {
			return nil, true, err
		}
		return nil, true, resourcePoolStatusError(codes.Unavailable, reasonPauseResumeDependencyUnavailable, "failed to load current shared pool state")
	}
	return sharedPool, true, nil
}

func (s *serverImpl) loadPauseResumeSharedPool(ctx context.Context, sharedPoolID string) (*entity.SharedPool, error) {
	sharedPool, err := s.sharedPoolRepo.Load(ctx, sharedPoolID, false)
	if err == nil || status.Code(err) == codes.NotFound {
		return sharedPool, err
	}
	return nil, resourcePoolStatusError(codes.Unavailable, reasonPauseResumeDependencyUnavailable, "failed to load shared pool dependencies")
}

func (s *serverImpl) validatePause(ctx context.Context, sharedPoolID, operationID string) error {
	collector, err := s.evaluatePause(ctx, sharedPoolID, operationID, false)
	if err != nil {
		return err
	}
	return collector.firstError()
}

func (s *serverImpl) evaluatePause(ctx context.Context, sharedPoolID, operationID string, collectAll bool) (*pauseValidationCollector, error) {
	sharedPool, err := s.loadPauseResumeSharedPool(ctx, sharedPoolID)
	if err != nil {
		return nil, err
	}
	collector := newPauseValidationCollector(collectAll)
	if s.collectPauseResumeLifecycle(ctx, sharedPool, apiv1.PauseResumeOperation_PAUSE, operationID, collector) {
		return collector, nil
	}
	if s.collectPauseResumeConflictingDAGs(ctx, sharedPoolID, collector, false) {
		return collector, nil
	}
	skipFlushWAL := false
	remoteCli, err := s.dataplaneManager.GetClient(sharedPool.GetDataplaneName())
	if err != nil {
		remoteCli = nil
		failureErr := resourcePoolStatusError(codes.Unavailable, reasonPauseResumeDependencyUnavailable, "dataplane or Operator cluster is unavailable")
		if collector.add(newPauseValidationFailure(pauseValidationDependencyUnavailable, "dataplane or Operator cluster is unavailable", failureErr)) {
			return collector, nil
		}
	} else {
		operatorStatus, inspectErr := tioperator.InspectClusterSuspendCompute(ctx, remoteCli, sharedPoolID)
		switch {
		case inspectErr != nil:
			failureErr := resourcePoolStatusError(codes.Unavailable, reasonPauseResumeDependencyUnavailable, "dataplane or Operator cluster is unavailable")
			if collector.add(newPauseValidationFailure(pauseValidationDependencyUnavailable, "dataplane or Operator cluster is unavailable", failureErr)) {
				return collector, nil
			}
		case operatorStatus.SpecDesired && !sharedPool.SuspendComputeDesired:
			message := "Operator suspend spec is true while suspend_compute_desired is false"
			failureErr := resourcePoolStatusError(codes.FailedPrecondition, reasonPauseResumeInvalidState, message)
			if collector.add(newPauseValidationFailure(pauseValidationInvalidLifecycleState, message, failureErr)) {
				return collector, nil
			}
			skipFlushWAL = true
		case operatorStatus.SpecDesired && operatorStatus.Observed:
			skipFlushWAL = true
		}
	}
	if !skipFlushWAL {
		if _, err := braction.ValidateFlushWALPrerequisites(ctx, sharedPoolID, s.sharedPoolRepo, s.dataplaneRepo, s.dataplaneManager, remoteCli, s.bizCfg.DefaultTiBRConfig.DisableFlushWAL); err != nil {
			collector.add(pauseFlushWALPrerequisiteValidationFailure(err))
		}
	}
	return collector, nil
}

func pauseFlushWALPrerequisiteValidationFailure(err error) *pauseValidationFailure {
	return newPauseValidationFailure(pauseValidationDependencyUnavailable, err.Error(), pauseFlushWALPrerequisiteStatusError(err))
}

func pauseFlushWALPrerequisiteStatusError(err error) error {
	if braction.IsTerminalFlushWALPrerequisiteError(err) {
		return resourcePoolStatusError(codes.FailedPrecondition, reasonPauseResumeInvalidState, err.Error())
	}
	return resourcePoolStatusError(codes.Unavailable, reasonPauseResumeDependencyUnavailable, "FlushWAL prerequisites are unavailable")
}

func (s *serverImpl) validateResume(ctx context.Context, sharedPoolID, operationID string) error {
	sharedPool, err := s.loadPauseResumeSharedPool(ctx, sharedPoolID)
	if err != nil {
		return err
	}
	if err := s.validatePauseResumeLifecycle(ctx, sharedPool, apiv1.PauseResumeOperation_RESUME, operationID); err != nil {
		return err
	}
	if err := s.checkPauseResumeConflictingDAGs(ctx, sharedPoolID,
		sharedPool.PauseResumeState == entity.PauseResumeStateResuming); err != nil {
		return err
	}
	remoteCli, err := s.dataplaneManager.GetClient(sharedPool.GetDataplaneName())
	if err != nil {
		return resourcePoolStatusError(codes.Unavailable, reasonPauseResumeDependencyUnavailable, "dataplane or Operator cluster is unavailable")
	}
	if _, err := tioperator.IsClusterSuspendComputeObserved(ctx, remoteCli, sharedPoolID, true); err != nil {
		return resourcePoolStatusError(codes.Unavailable, reasonPauseResumeDependencyUnavailable, "dataplane or Operator cluster is unavailable")
	}
	return nil
}

func (s *serverImpl) validatePauseResumeLifecycle(ctx context.Context, sharedPool *entity.SharedPool, operationType apiv1.PauseResumeOperation_Type, operationID string) error {
	collector := newPauseValidationCollector(false)
	s.collectPauseResumeLifecycle(ctx, sharedPool, operationType, operationID, collector)
	return collector.firstError()
}

func (s *serverImpl) collectPauseResumeLifecycle(
	ctx context.Context,
	sharedPool *entity.SharedPool,
	operationType apiv1.PauseResumeOperation_Type,
	operationID string,
	collector *pauseValidationCollector,
) bool {
	if err := entity.ValidatePauseResumeCombination(sharedPool); err != nil {
		failureErr := resourcePoolStatusError(codes.FailedPrecondition, reasonPauseResumeInvalidState, err.Error())
		if collector.add(newPauseValidationFailure(pauseValidationInvalidLifecycleState, err.Error(), failureErr)) {
			return true
		}
	}
	running, err := s.queryPauseValidationRunnings(ctx, dag.PauseResumePlans(), executor.NewMatchFilter().Add(workflow.SharedPoolIdKey, []string{sharedPool.GetId()}), collector.collectAll)
	if err != nil {
		failureErr := resourcePoolStatusError(codes.Unavailable, reasonPauseResumeDependencyUnavailable, "failed to query running pause/resume operations")
		if collector.add(newPauseValidationFailure(pauseValidationDependencyUnavailable, "failed to query running pause/resume operations", failureErr)) {
			return true
		}
	}
	if operationID != "" {
		delete(running, workflow.PauseResumeExecID(sharedPool.GetId(), operationID, apiv1.PauseResumeOperation_PAUSE.String()))
		delete(running, workflow.PauseResumeExecID(sharedPool.GetId(), operationID, apiv1.PauseResumeOperation_RESUME.String()))
	}
	if len(running) > 0 {
		failureErr := resourcePoolStatusError(codes.FailedPrecondition, reasonPauseResumeOperationInProgress, "a pause/resume operation is in progress")
		if collector.add(newPauseValidationFailure(pauseValidationOperationInProgress, "a pause/resume operation is in progress", failureErr)) {
			return true
		}
	}
	if sharedPool.PauseResumeState == entity.PauseResumeStatePausing || sharedPool.PauseResumeState == entity.PauseResumeStateResuming {
		expectedState := entity.PauseResumeStatePausing
		if operationType == apiv1.PauseResumeOperation_RESUME {
			expectedState = entity.PauseResumeStateResuming
		}
		latest, err := s.pauseResumeOperationRepo.LoadLatestBySharedPoolID(ctx, sharedPool.GetId())
		switch {
		case errors.Is(err, gorm.ErrRecordNotFound):
			message := "transitional shared pool has no recoverable pause/resume operation"
			failureErr := resourcePoolStatusError(codes.FailedPrecondition, reasonPauseResumeInvalidState, message)
			if collector.add(newPauseValidationFailure(pauseValidationInvalidLifecycleState, message, failureErr)) {
				return true
			}
		case err != nil:
			message := "failed to load latest pause/resume operation"
			failureErr := resourcePoolStatusError(codes.Unavailable, reasonPauseResumeDependencyUnavailable, message)
			if collector.add(newPauseValidationFailure(pauseValidationDependencyUnavailable, message, failureErr)) {
				return true
			}
		case sharedPool.PauseResumeState != expectedState || latest.GetType() != operationType || latest.GetStatus() != apiv1.PauseResumeOperation_FAILED:
			if latest.GetStatus() == apiv1.PauseResumeOperation_RUNNING {
				message := "a pause/resume operation is in progress"
				failureErr := resourcePoolStatusError(codes.FailedPrecondition, reasonPauseResumeOperationInProgress, message)
				if collector.add(newPauseValidationFailure(pauseValidationOperationInProgress, message, failureErr)) {
					return true
				}
			} else {
				message := "shared pool is not eligible for pause/resume retry"
				failureErr := resourcePoolStatusError(codes.FailedPrecondition, reasonPauseResumeInvalidState, message)
				if collector.add(newPauseValidationFailure(pauseValidationInvalidLifecycleState, message, failureErr)) {
					return true
				}
			}
		}
		return false
	}
	if operationType == apiv1.PauseResumeOperation_PAUSE && (sharedPool.PauseResumeState != entity.PauseResumeStateNone || sharedPool.GetPhase() != apiv1.SharedPool_AVAILABLE) {
		message := "shared pool must be AVAILABLE"
		failureErr := resourcePoolStatusError(codes.FailedPrecondition, reasonPauseResumeInvalidState, message)
		if collector.add(newPauseValidationFailure(pauseValidationInvalidLifecycleState, message, failureErr)) {
			return true
		}
	}
	if operationType == apiv1.PauseResumeOperation_RESUME && sharedPool.PauseResumeState != entity.PauseResumeStatePaused {
		message := "shared pool must be stably PAUSED"
		failureErr := resourcePoolStatusError(codes.FailedPrecondition, reasonPauseResumeInvalidState, message)
		if collector.add(newPauseValidationFailure(pauseValidationInvalidLifecycleState, message, failureErr)) {
			return true
		}
	}
	return false
}

func (s *serverImpl) checkPauseResumeConflictingDAGs(ctx context.Context, sharedPoolID string, ignoreCreateConflicts bool) error {
	collector := newPauseValidationCollector(false)
	s.collectPauseResumeConflictingDAGs(ctx, sharedPoolID, collector, ignoreCreateConflicts)
	return collector.firstError()
}

func (s *serverImpl) collectPauseResumeConflictingDAGs(ctx context.Context, sharedPoolID string, collector *pauseValidationCollector, ignoreCreateConflicts bool) bool {
	targetGroups := []struct {
		plans   []string
		code    string
		message string
	}{
		{plans: dag.PauseResumeResourceMutationPlans(), code: pauseValidationOperationInProgress, message: "a conflicting resource mutation is in progress"},
		{plans: dag.PauseResumeTiBRUpgradePlans(), code: pauseValidationUpgradeInProgress, message: "a conflicting TiBR upgrade is in progress"},
		{plans: dag.PauseResumeDMPlans(), code: pauseValidationOperationInProgress, message: "a conflicting DM operation is in progress"},
		{plans: dag.PauseResumeBackupRestorePlans(), code: pauseValidationBackupInProgress, message: "a conflicting backup/restore operation is in progress"},
	}
	for _, group := range targetGroups {
		plans := pauseResumeConflictPlans(group.plans, ignoreCreateConflicts)
		running, err := s.queryPauseValidationRunnings(ctx, plans, executor.NewMatchFilter().Add(workflow.SharedPoolIdKey, []string{sharedPoolID}), collector.collectAll)
		if err != nil {
			failureErr := resourcePoolStatusError(codes.Unavailable, reasonPauseResumeDependencyUnavailable, "failed to query conflicting operations")
			if collector.add(newPauseValidationFailure(pauseValidationDependencyUnavailable, "failed to query conflicting operations", failureErr)) {
				return true
			}
		}
		if len(running) > 0 {
			failureErr := resourcePoolStatusError(codes.FailedPrecondition, reasonPauseResumeOperationInProgress, "a conflicting operation is in progress")
			if collector.add(newPauseValidationFailure(group.code, group.message, failureErr)) {
				return true
			}
		}
	}
	sourceRunning, err := s.queryPauseValidationRunnings(ctx, dag.PauseResumeBackupRestorePlans(), executor.NewMatchFilter().Add(workflow.SourceSharedPoolIdKey, []string{sharedPoolID}), collector.collectAll)
	if err != nil {
		message := "failed to query conflicting source-pool operations"
		failureErr := resourcePoolStatusError(codes.Unavailable, reasonPauseResumeDependencyUnavailable, message)
		if collector.add(newPauseValidationFailure(pauseValidationDependencyUnavailable, message, failureErr)) {
			return true
		}
	}
	if len(sourceRunning) > 0 {
		message := "a conflicting backup/restore operation is in progress"
		failureErr := resourcePoolStatusError(codes.FailedPrecondition, reasonPauseResumeOperationInProgress, "a conflicting operation is in progress")
		if collector.add(newPauseValidationFailure(pauseValidationBackupInProgress, message, failureErr)) {
			return true
		}
	}
	return false
}

func pauseResumeConflictPlans(plans []string, ignoreCreateConflicts bool) []string {
	if !ignoreCreateConflicts {
		return plans
	}
	filtered := make([]string, 0, len(plans))
	for _, plan := range plans {
		if plan != dag.PlanCreateTiDBCluster {
			filtered = append(filtered, plan)
		}
	}
	return filtered
}

func (s *serverImpl) queryPauseValidationRunnings(ctx context.Context, planNames []string, filter executor.MatchFilter, collectAll bool) (executor.ExecTaskIdStatus, error) {
	if !collectAll {
		return s.queryRunningDagIns(ctx, planNames, filter)
	}
	running := executor.NewExecTaskIdStatus()
	var firstErr error
	for _, planName := range planNames {
		planRunning, err := s.dagExecutor.QueryRunnings(ctx, planName, filter)
		if err != nil {
			if firstErr == nil {
				firstErr = err
			}
			continue
		}
		for execID, taskStatus := range planRunning {
			running[execID] = taskStatus
		}
	}
	return running, firstErr
}