package resourcepool

import (
	"context"
	"errors"
	"fmt"
	"slices"
	"strconv"
	"strings"
	"time"

	clusterngv1beta "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"

	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/entity"
	clusterrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/cluster"
	exportrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/export"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/service/workflow/common"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/infra/repoimpl/schema"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/clusterevent"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

const (
	pauseOperationType  = "PAUSE"
	resumeOperationType = "RESUME"

	invalidLifecycleStateCheckCode  = "INVALID_LIFECYCLE_STATE"
	dependencyUnavailableCheckCode  = "DEPENDENCY_UNAVAILABLE"
	instanceStateConflictCheckCode  = "INSTANCE_STATE_CONFLICT"
	importExportInProgressCheckCode = "IMPORT_EXPORT_IN_PROGRESS"
	operationInProgressCheckCode    = "OPERATION_IN_PROGRESS"
	pauseAdmissionTimeout           = 5 * time.Minute
)

var pauseBlockingGlobalDagIDs = []string{
	common.UpdatePublicLb,
	common.UpdateVpcPeeringAllowlist,
	common.CreatePrivateLinkConnection,
	common.DeletePrivateLinkConnection,
	common.CreatePrivateLinkService,
	common.DeletePrivateLinkService,
	common.UpdateAuditLogSetting,
	common.CreateAuditLogFilterRule,
	common.UpdateAuditLogFilterRule,
	common.DeleteAuditLogFilterRule,
	common.UpdateLogRedactionPolicy,
}

func (s *serviceImpl) validatePauseResourcePool(
	ctx context.Context,
	resourcePool *entity.ResourcePool,
) []*globalngv1.ResourcePoolPauseValidationResult {
	return s.validatePauseResourcePoolForLifecycle(ctx, resourcePool, globalngv1.ResourcePool_ACTIVE)
}

func (s *serviceImpl) validatePauseResourcePoolForLifecycle(
	ctx context.Context,
	resourcePool *entity.ResourcePool,
	expectedLifecycle globalngv1.ResourcePool_LifecycleState,
) []*globalngv1.ResourcePoolPauseValidationResult {
	results := validatePauseLifecycleStateFor(resourcePool, expectedLifecycle)

	results = append(results, s.validatePauseGlobal(ctx, resourcePool)...)

	regionalResults, berr := s.validatePauseRegional(ctx, resourcePool)
	if berr != nil {
		logPauseValidationDependencyError(ctx, resourcePool.GetId(), "regional", berr)
		results = appendDependencyUnavailable(results, "failed to validate regional resource pool state")
	} else {
		results = append(results, regionalResults...)
	}

	return normalizePauseValidationResults(results)
}

func (s *serviceImpl) ValidatePausePreparation(
	ctx context.Context,
	resourcePool *entity.ResourcePool,
) ([]*globalngv1.ResourcePoolPauseValidationResult, bizerr.BizError) {
	results := s.validatePauseResourcePoolForLifecycle(
		ctx,
		resourcePool,
		globalngv1.ResourcePool_PAUSE_PRECHECKING,
	)
	for _, result := range results {
		if result.GetCheckCode() == dependencyUnavailableCheckCode {
			return results, bizerr.New(bizerr.ErrCodeUnknown, "pause validation dependency is unavailable")
		}
	}
	return results, nil
}

func validatePauseLifecycleState(
	resourcePool *entity.ResourcePool,
) []*globalngv1.ResourcePoolPauseValidationResult {
	return validatePauseLifecycleStateFor(resourcePool, globalngv1.ResourcePool_ACTIVE)
}

func validatePauseLifecycleStateFor(
	resourcePool *entity.ResourcePool,
	expectedLifecycle globalngv1.ResourcePool_LifecycleState,
) []*globalngv1.ResourcePoolPauseValidationResult {
	if resourcePool.GetLifecycleState() != expectedLifecycle {
		return []*globalngv1.ResourcePoolPauseValidationResult{{
			CheckCode: invalidLifecycleStateCheckCode,
			Detail:    fmt.Sprintf("resource pool must be %s to pause", expectedLifecycle.String()),
		}}
	}

	return []*globalngv1.ResourcePoolPauseValidationResult{}
}

func (s *serviceImpl) validatePauseGlobal(
	ctx context.Context,
	resourcePool *entity.ResourcePool,
) []*globalngv1.ResourcePoolPauseValidationResult {
	clusters, berr := s.clusterRepo.Query(ctx, &clusterrepo.Filter{
		SharedPoolIds: []uint64{resourcePool.GetId()},
	})
	if berr != nil {
		logPauseValidationDependencyError(ctx, resourcePool.GetId(), "instances", berr)
		return appendDependencyUnavailable(nil, "failed to query resource pool instances")
	}

	results := validatePauseInstanceStates(clusters)
	results = append(results, s.validatePauseExports(ctx, resourcePool.GetId(), clusters)...)
	results = append(results, s.validatePauseGlobalWorkflows(ctx, resourcePool.GetId(), clusters)...)
	return results
}

func validatePauseInstanceStates(clusters []*entity.Cluster) []*globalngv1.ResourcePoolPauseValidationResult {
	for _, cluster := range clusters {
		if cluster == nil || cluster.GetState() != globalngv1.Cluster_AVAILABLE {
			return []*globalngv1.ResourcePoolPauseValidationResult{{
				CheckCode: instanceStateConflictCheckCode,
				Detail:    "all resource pool instances must be ACTIVE to pause",
			}}
		}
	}

	return []*globalngv1.ResourcePoolPauseValidationResult{}
}

func (s *serviceImpl) validatePauseExports(
	ctx context.Context,
	resourcePoolID uint64,
	clusters []*entity.Cluster,
) []*globalngv1.ResourcePoolPauseValidationResult {
	clusterIDs := pauseClusterIDs(clusters)
	if len(clusterIDs) == 0 {
		return []*globalngv1.ResourcePoolPauseValidationResult{}
	}

	runningCount, berr := s.exportRepo.Count(ctx, &exportrepo.QueryExportsFilter{
		ClusterIds: clusterIDs,
		States:     []schema.ExportState{schema.ExportStateRunning},
	})
	if berr != nil {
		logPauseValidationDependencyError(ctx, resourcePoolID, "exports", berr)
		return []*globalngv1.ResourcePoolPauseValidationResult{{
			CheckCode: dependencyUnavailableCheckCode,
			Detail:    "failed to query running exports",
		}}
	}
	if runningCount == 0 {
		return []*globalngv1.ResourcePoolPauseValidationResult{}
	}

	return []*globalngv1.ResourcePoolPauseValidationResult{{
		CheckCode: importExportInProgressCheckCode,
		Detail:    "resource pool has a running export",
	}}
}

func (s *serviceImpl) validatePauseGlobalWorkflows(
	ctx context.Context,
	resourcePoolID uint64,
	clusters []*entity.Cluster,
) []*globalngv1.ResourcePoolPauseValidationResult {
	clusterIDs := pauseClusterIDs(clusters)
	if len(clusterIDs) == 0 {
		return []*globalngv1.ResourcePoolPauseValidationResult{}
	}

	instances, err := s.workflow.ListRunningDagInstancesByBizIds(
		ctx,
		pauseBlockingGlobalDagIDs,
		clusterIDs,
	)
	if err != nil {
		logPauseValidationDependencyError(ctx, resourcePoolID, "global_workflows", err)
		return []*globalngv1.ResourcePoolPauseValidationResult{{
			CheckCode: dependencyUnavailableCheckCode,
			Detail:    "failed to query running global operations",
		}}
	}
	if len(instances) == 0 {
		return []*globalngv1.ResourcePoolPauseValidationResult{}
	}

	return []*globalngv1.ResourcePoolPauseValidationResult{{
		CheckCode: operationInProgressCheckCode,
		Detail:    "resource pool has a running global operation",
	}}
}

func (s *serviceImpl) validatePauseRegional(
	ctx context.Context,
	resourcePool *entity.ResourcePool,
) ([]*globalngv1.ResourcePoolPauseValidationResult, bizerr.BizError) {
	regionEntity, berr := s.regionRepo.Get(ctx, resourcePool.RegionId)
	if berr != nil {
		return nil, berr
	}
	validation, berr := s.regionalCli.ValidateSharedPoolPause(
		ctx,
		regionEntity,
		&clusterngv1beta.ValidateSharedPoolPauseRequest{
			SharedPoolId: strconv.FormatUint(resourcePool.Id, 10),
		},
	)
	if berr != nil {
		return nil, berr
	}

	results := make([]*globalngv1.ResourcePoolPauseValidationResult, 0, len(validation.GetFailures()))
	for _, failure := range validation.GetFailures() {
		if failure == nil {
			continue
		}
		results = append(results, &globalngv1.ResourcePoolPauseValidationResult{
			CheckCode: failure.GetCode(),
			Detail:    failure.GetMessage(),
		})
	}
	return results, nil
}

func pauseClusterIDs(clusters []*entity.Cluster) []string {
	clusterIDs := make([]string, 0, len(clusters))
	for _, cluster := range clusters {
		if cluster == nil || cluster.GetClusterId() == 0 {
			continue
		}
		clusterIDs = append(clusterIDs, strconv.FormatUint(cluster.GetClusterId(), 10))
	}
	return clusterIDs
}

func appendDependencyUnavailable(
	results []*globalngv1.ResourcePoolPauseValidationResult,
	detail string,
) []*globalngv1.ResourcePoolPauseValidationResult {
	return append(results, &globalngv1.ResourcePoolPauseValidationResult{
		CheckCode: dependencyUnavailableCheckCode,
		Detail:    detail,
	})
}

func logPauseValidationDependencyError(
	ctx context.Context,
	resourcePoolID uint64,
	check string,
	err error,
) {
	tilog.Error(
		ctx,
		"resource pool pause validation dependency failed",
		zap.Uint64("resourcePoolId", resourcePoolID),
		zap.String("check", check),
		zap.Error(err),
	)
}

func normalizePauseValidationResults(
	results []*globalngv1.ResourcePoolPauseValidationResult,
) []*globalngv1.ResourcePoolPauseValidationResult {
	slices.SortStableFunc(results, func(left, right *globalngv1.ResourcePoolPauseValidationResult) int {
		if codeOrder := strings.Compare(left.GetCheckCode(), right.GetCheckCode()); codeOrder != 0 {
			return codeOrder
		}
		return strings.Compare(left.GetDetail(), right.GetDetail())
	})

	normalized := make([]*globalngv1.ResourcePoolPauseValidationResult, 0, len(results))
	for _, result := range results {
		if result == nil {
			continue
		}
		if len(normalized) > 0 && normalized[len(normalized)-1].GetCheckCode() == result.GetCheckCode() &&
			normalized[len(normalized)-1].GetDetail() == result.GetDetail() {
			continue
		}
		normalized = append(normalized, result)
	}
	return normalized
}

func (s *serviceImpl) PauseResourcePool(
	ctx context.Context,
	request *globalngv1.PauseResourcePoolRequest,
) (*globalngv1.PauseResourcePoolResponse, bizerr.BizError) {
	if request == nil || request.GetResourcePoolId() == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "resource pool id is required")
	}
	if !request.GetValidateOnly() && request.GetOperationId() == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "operation id is required")
	}
	if !request.GetValidateOnly() && len(request.GetOperationId()) > 128 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "operation id must not exceed 128 bytes")
	}

	resourcePool, berr := s.resourcePoolRepo.Load(ctx, request.GetResourcePoolId(), false)
	if berr != nil {
		return nil, berr
	}
	if request.GetValidateOnly() {
		results := s.validatePauseResourcePool(ctx, resourcePool)
		return &globalngv1.PauseResourcePoolResponse{ValidationResults: results}, nil
	}
	localReplay, berr := s.checkLocalPauseResumeOperation(
		ctx, resourcePool.Id, request.GetOperationId(), entity.ResourcePoolPauseOperationTypePause,
	)
	if berr != nil {
		return nil, berr
	}
	if localReplay {
		return &globalngv1.PauseResourcePoolResponse{OperationId: request.GetOperationId()}, nil
	}
	if request.GetCreateManualBackupBeforePause() {
		validationResults := s.validatePauseResourcePool(ctx, resourcePool)
		if len(validationResults) > 0 {
			return &globalngv1.PauseResourcePoolResponse{ValidationResults: validationResults}, nil
		}
		transactionValidationResults, berr := s.acceptPauseWithManualBackup(
			ctx,
			resourcePool,
			request.GetOperationId(),
		)
		if berr != nil {
			return nil, berr
		}
		if len(transactionValidationResults) > 0 {
			return &globalngv1.PauseResourcePoolResponse{
				ValidationResults: transactionValidationResults,
			}, nil
		}
		return &globalngv1.PauseResourcePoolResponse{OperationId: request.GetOperationId()}, nil
	}

	validationResults := s.validatePauseResourcePool(ctx, resourcePool)
	if len(validationResults) > 0 {
		return &globalngv1.PauseResourcePoolResponse{ValidationResults: validationResults}, nil
	}

	transactionValidationResults, berr := s.acceptPauseResume(
		ctx,
		resourcePool,
		request.GetOperationId(),
		pauseOperationType,
		globalngv1.ResourcePool_ACTIVE,
		globalngv1.ResourcePool_PAUSING,
		common.PauseResourcePool,
	)
	if berr != nil {
		return nil, berr
	}
	if len(transactionValidationResults) > 0 {
		return &globalngv1.PauseResourcePoolResponse{ValidationResults: transactionValidationResults}, nil
	}
	return &globalngv1.PauseResourcePoolResponse{OperationId: request.GetOperationId()}, nil
}

func (s *serviceImpl) BeginPauseResourcePoolAdmission(
	ctx context.Context,
	request *globalngv1.BeginPauseResourcePoolAdmissionRequest,
) (*globalngv1.BeginPauseResourcePoolAdmissionResponse, bizerr.BizError) {
	if request == nil || request.GetResourcePoolId() == 0 || request.GetOperationId() == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "resource pool id and operation id are required")
	}
	if len(request.GetOperationId()) > 128 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "operation id must not exceed 128 bytes")
	}

	resourcePool, berr := s.resourcePoolRepo.Load(ctx, request.GetResourcePoolId(), false)
	if berr != nil {
		return nil, berr
	}
	localReplay, berr := s.checkLocalPauseResumeOperation(
		ctx, resourcePool.Id, request.GetOperationId(), entity.ResourcePoolPauseOperationTypePause,
	)
	if berr != nil {
		return nil, berr
	}
	if localReplay {
		return &globalngv1.BeginPauseResourcePoolAdmissionResponse{OperationId: request.GetOperationId()}, nil
	}

	validationResults := s.validatePauseResourcePool(ctx, resourcePool)
	if len(validationResults) > 0 {
		return &globalngv1.BeginPauseResourcePoolAdmissionResponse{ValidationResults: validationResults}, nil
	}

	transactionValidationResults := make([]*globalngv1.ResourcePoolPauseValidationResult, 0)
	err := s.transaction.Transaction(ctx, func(ctx context.Context) error {
		locked, berr := s.resourcePoolRepo.LoadForUpdate(ctx, resourcePool.Id, false)
		if berr != nil {
			return berr
		}
		replayed, berr := s.checkLocalPauseResumeOperation(
			ctx, locked.Id, request.GetOperationId(), entity.ResourcePoolPauseOperationTypePause,
		)
		if berr != nil {
			return berr
		}
		if replayed {
			return nil
		}
		if locked.GetLifecycleState() != globalngv1.ResourcePool_ACTIVE {
			transactionValidationResults = validatePauseLifecycleState(locked)
			return nil
		}
		if request.GetCreateManualBackupBeforePause() {
			transactionValidationResults = s.validatePauseManualBackupAcceptance(ctx, locked)
		} else {
			transactionValidationResults = normalizePauseValidationResults(s.validatePauseGlobal(ctx, locked))
		}
		if len(transactionValidationResults) > 0 {
			return nil
		}

		stage := entity.ResourcePoolPauseOperationAdmission
		if request.GetCreateManualBackupBeforePause() {
			stage = entity.ResourcePoolPauseOperationAdmissionWithManualBackup
		}
		if berr := s.pauseOperationRepo.Create(ctx, &entity.ResourcePoolPauseOperation{
			ResourcePoolID: locked.Id,
			OperationID:    request.GetOperationId(),
			Type:           entity.ResourcePoolPauseOperationTypePause,
			Status:         entity.ResourcePoolPauseOperationRunning,
			Stage:          stage,
			DeadlineAt:     time.Now().UTC().Add(pauseAdmissionTimeout),
		}); berr != nil {
			return berr
		}
		locked.LifecycleState = globalngv1.ResourcePool_PAUSE_PRECHECKING
		locked.StateMessageCode = ""
		locked.PausedAt = nil
		locked.AutomaticResumeAt = nil
		return s.resourcePoolRepo.Save(ctx, locked)
	})
	if err != nil {
		var businessErr bizerr.BizError
		if errors.As(err, &businessErr) {
			return nil, businessErr
		}
		return nil, bizerr.WrapGormError(err, "begin pause admission transaction failed")
	}
	if len(transactionValidationResults) > 0 {
		return &globalngv1.BeginPauseResourcePoolAdmissionResponse{ValidationResults: transactionValidationResults}, nil
	}
	return &globalngv1.BeginPauseResourcePoolAdmissionResponse{OperationId: request.GetOperationId()}, nil
}

func (s *serviceImpl) ResolvePauseResourcePoolAdmission(
	ctx context.Context,
	request *globalngv1.ResolvePauseResourcePoolAdmissionRequest,
) (*globalngv1.ResolvePauseResourcePoolAdmissionResponse, bizerr.BizError) {
	if request == nil || request.GetResourcePoolId() == 0 || request.GetOperationId() == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "resource pool id and operation id are required")
	}
	err := s.transaction.Transaction(ctx, func(ctx context.Context) error {
		pool, berr := s.resourcePoolRepo.LoadForUpdate(ctx, request.GetResourcePoolId(), false)
		if berr != nil {
			return berr
		}
		operation, berr := s.pauseOperationRepo.Get(ctx, pool.Id, request.GetOperationId())
		if berr != nil {
			return berr
		}
		if operation.Status != entity.ResourcePoolPauseOperationRunning {
			return bizerr.New(bizerr.ErrCode(codes.FailedPrecondition), "pause admission operation is not running")
		}
		if operation.Stage != entity.ResourcePoolPauseOperationAdmission &&
			operation.Stage != entity.ResourcePoolPauseOperationAdmissionWithManualBackup {
			return nil
		}
		if pool.GetLifecycleState() != globalngv1.ResourcePool_PAUSE_PRECHECKING {
			return bizerr.New(bizerr.ErrCode(codes.FailedPrecondition), "resource pool lost pause admission ownership")
		}

		if !request.GetProceed() {
			pool.LifecycleState = globalngv1.ResourcePool_ACTIVE
			pool.StateMessageCode = ""
			if berr := s.resourcePoolRepo.Save(ctx, pool); berr != nil {
				return berr
			}
			now := time.Now().UTC()
			operation.Status = entity.ResourcePoolPauseOperationFailed
			operation.ResultSummary = pauseAdmissionResultSummary(request.GetValidationResults())
			operation.CompletedAt = &now
			return s.pauseOperationRepo.Save(ctx, operation)
		}

		if operation.Stage == entity.ResourcePoolPauseOperationAdmissionWithManualBackup {
			operation.Stage = entity.ResourcePoolPauseOperationManualBackup
			operation.DeadlineAt = time.Now().UTC().Add(common.PausePreparationTimeout)
			if berr := s.pauseOperationRepo.Save(ctx, operation); berr != nil {
				return berr
			}
			return s.runPauseDag(ctx, pool, request.GetOperationId(), common.PrepareAndPauseResourcePool)
		}

		transactionValidationResults := normalizePauseValidationResults(s.validatePauseGlobal(ctx, pool))
		if len(transactionValidationResults) > 0 {
			return bizerr.New(bizerr.ErrCode(codes.FailedPrecondition), "resource pool pause validation changed during admission")
		}
		pool.LifecycleState = globalngv1.ResourcePool_PAUSING
		if berr := s.resourcePoolRepo.Save(ctx, pool); berr != nil {
			return berr
		}
		operation.Stage = entity.ResourcePoolPauseOperationRegional
		if berr := s.pauseOperationRepo.Save(ctx, operation); berr != nil {
			return berr
		}
		ev, berr := entity.NewResourcePoolLifecycleEvent(ctx, pool, clusterevent.EventTypePausing)
		if berr != nil {
			return berr
		}
		if berr := s.eventCommitter.Write(ctx, clusterevent.TopicClusterEvent, ev); berr != nil {
			return berr
		}
		return s.runPauseDag(ctx, pool, request.GetOperationId(), common.PauseResourcePool)
	})
	if err != nil {
		var businessErr bizerr.BizError
		if errors.As(err, &businessErr) {
			return nil, businessErr
		}
		return nil, bizerr.WrapGormError(err, "resolve pause admission transaction failed")
	}
	return &globalngv1.ResolvePauseResourcePoolAdmissionResponse{
		OperationId:       request.GetOperationId(),
		ValidationResults: request.GetValidationResults(),
	}, nil
}

func (s *serviceImpl) runPauseDag(ctx context.Context, pool *entity.ResourcePool, operationID, dagID string) error {
	if err := s.workflow.RunDagWithBizId(
		ctx,
		dagID,
		strconv.FormatUint(pool.OwnerId, 10),
		"0",
		strconv.FormatUint(pool.Id, 10),
		map[string]string{
			workflow.KeyTraceId:      workflow.GetTraceId(ctx),
			common.KeyResourcePoolId: strconv.FormatUint(pool.Id, 10),
			common.KeyOperationId:    operationID,
		},
	); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "create pause workflow failed")
	}
	return nil
}

func pauseAdmissionResultSummary(results []*globalngv1.ResourcePoolPauseValidationResult) string {
	parts := make([]string, 0, len(results))
	for _, result := range results {
		if result == nil {
			continue
		}
		parts = append(parts, fmt.Sprintf("%s: %s", result.GetCheckCode(), result.GetDetail()))
	}
	return strings.Join(parts, "; ")
}

func (s *serviceImpl) ExpirePauseAdmissions(ctx context.Context) bizerr.BizError {
	now := time.Now().UTC()
	operations, berr := s.pauseOperationRepo.ListExpiredAdmissions(ctx, now)
	if berr != nil {
		return berr
	}
	for _, candidate := range operations {
		if candidate == nil {
			continue
		}
		err := s.transaction.Transaction(ctx, func(ctx context.Context) error {
			pool, berr := s.resourcePoolRepo.LoadForUpdate(ctx, candidate.ResourcePoolID, false)
			if berr != nil {
				return berr
			}
			operation, berr := s.pauseOperationRepo.Get(ctx, candidate.ResourcePoolID, candidate.OperationID)
			if berr != nil {
				return berr
			}
			if operation.Status != entity.ResourcePoolPauseOperationRunning || operation.DeadlineAt.After(now) ||
				(operation.Stage != entity.ResourcePoolPauseOperationAdmission &&
					operation.Stage != entity.ResourcePoolPauseOperationAdmissionWithManualBackup) {
				return nil
			}
			if pool.GetLifecycleState() == globalngv1.ResourcePool_PAUSE_PRECHECKING {
				pool.LifecycleState = globalngv1.ResourcePool_ACTIVE
				pool.StateMessageCode = ""
				if berr := s.resourcePoolRepo.Save(ctx, pool); berr != nil {
					return berr
				}
			}
			operation.Status = entity.ResourcePoolPauseOperationFailed
			operation.ResultSummary = "pause admission timed out"
			operation.CompletedAt = &now
			return s.pauseOperationRepo.Save(ctx, operation)
		})
		if err != nil {
			var businessErr bizerr.BizError
			if errors.As(err, &businessErr) {
				return businessErr
			}
			return bizerr.WrapGormError(err, "expire pause admission transaction failed")
		}
	}
	return nil
}

func (s *serviceImpl) acceptPauseWithManualBackup(
	ctx context.Context,
	resourcePool *entity.ResourcePool,
	operationID string,
) ([]*globalngv1.ResourcePoolPauseValidationResult, bizerr.BizError) {
	transactionValidationResults := make([]*globalngv1.ResourcePoolPauseValidationResult, 0)
	err := s.transaction.Transaction(ctx, func(ctx context.Context) error {
		locked, berr := s.resourcePoolRepo.LoadForUpdate(ctx, resourcePool.Id, false)
		if berr != nil {
			return berr
		}
		if locked.GetLifecycleState() != globalngv1.ResourcePool_ACTIVE {
			existing, operationErr := s.pauseOperationRepo.Get(ctx, locked.Id, operationID)
			if operationErr == nil && existing != nil {
				return nil
			}
			if operationErr != nil && operationErr.ErrCode() != bizerr.ErrCodeRecordNotFound {
				return operationErr
			}
			transactionValidationResults = validatePauseLifecycleState(locked)
			return nil
		}

		transactionValidationResults = s.validatePauseManualBackupAcceptance(ctx, locked)
		if len(transactionValidationResults) > 0 {
			return nil
		}

		operation := &entity.ResourcePoolPauseOperation{
			ResourcePoolID: locked.Id,
			OperationID:    operationID,
			Type:           entity.ResourcePoolPauseOperationTypePause,
			Status:         entity.ResourcePoolPauseOperationRunning,
			Stage:          entity.ResourcePoolPauseOperationManualBackup,
			DeadlineAt:     time.Now().UTC().Add(common.PausePreparationTimeout),
		}
		if berr := s.pauseOperationRepo.Create(ctx, operation); berr != nil {
			return berr
		}
		locked.LifecycleState = globalngv1.ResourcePool_PAUSE_PRECHECKING
		locked.StateMessageCode = ""
		locked.PausedAt = nil
		locked.AutomaticResumeAt = nil
		if berr := s.resourcePoolRepo.Save(ctx, locked); berr != nil {
			return berr
		}
		if err := s.workflow.RunDagWithBizId(
			ctx,
			common.PrepareAndPauseResourcePool,
			strconv.FormatUint(locked.OwnerId, 10),
			"0",
			strconv.FormatUint(locked.Id, 10),
			map[string]string{
				workflow.KeyTraceId:      workflow.GetTraceId(ctx),
				common.KeyResourcePoolId: strconv.FormatUint(locked.Id, 10),
				common.KeyOperationId:    operationID,
			},
		); err != nil {
			return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "create pause preparation workflow failed")
		}
		return nil
	})
	if err == nil {
		return transactionValidationResults, nil
	}
	var businessErr bizerr.BizError
	if errors.As(err, &businessErr) {
		return nil, businessErr
	}
	return nil, bizerr.WrapGormError(err, "pause preparation workflow transaction failed")
}

func (s *serviceImpl) validatePauseManualBackupAcceptance(
	ctx context.Context,
	resourcePool *entity.ResourcePool,
) []*globalngv1.ResourcePoolPauseValidationResult {
	clusters, berr := s.clusterRepo.Query(ctx, &clusterrepo.Filter{
		SharedPoolIds: []uint64{resourcePool.GetId()},
	})
	if berr != nil {
		logPauseValidationDependencyError(ctx, resourcePool.GetId(), "instances", berr)
		return appendDependencyUnavailable(nil, "failed to query resource pool instances")
	}

	results := validatePauseInstanceStates(clusters)
	results = append(results, s.validatePauseExports(ctx, resourcePool.GetId(), clusters)...)
	results = append(results, s.validatePauseGlobalWorkflows(ctx, resourcePool.GetId(), clusters)...)

	clusterIDs := make([]uint64, 0, len(clusters))
	for _, cluster := range clusters {
		if cluster == nil || cluster.GetClusterId() == 0 {
			continue
		}
		clusterIDs = append(clusterIDs, cluster.GetClusterId())
	}
	if len(clusterIDs) == 0 {
		return normalizePauseValidationResults(results)
	}

	hasActiveRestore, berr := s.restoreRepo.HasActiveByTargetClusterIDs(ctx, clusterIDs)
	if berr != nil {
		logPauseValidationDependencyError(ctx, resourcePool.GetId(), "restores", berr)
		results = appendDependencyUnavailable(results, "failed to query active restores")
	} else if hasActiveRestore {
		results = append(results, &globalngv1.ResourcePoolPauseValidationResult{
			CheckCode: operationInProgressCheckCode,
			Detail:    "resource pool has an active restore",
		})
	}

	return normalizePauseValidationResults(results)
}

func (s *serviceImpl) ResumeResourcePool(
	ctx context.Context,
	request *globalngv1.ResumeResourcePoolRequest,
) (*globalngv1.ResumeResourcePoolResponse, bizerr.BizError) {
	if request == nil || request.GetResourcePoolId() == 0 || request.GetOperationId() == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "resource pool id and operation id are required")
	}
	if len(request.GetOperationId()) > 128 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "operation id must not exceed 128 bytes")
	}

	resourcePool, berr := s.resourcePoolRepo.Load(ctx, request.GetResourcePoolId(), false)
	if berr != nil {
		return nil, berr
	}
	localReplay, berr := s.checkLocalPauseResumeOperation(
		ctx, resourcePool.Id, request.GetOperationId(), entity.ResourcePoolPauseOperationTypeResume,
	)
	if berr != nil {
		return nil, berr
	}
	if localReplay {
		return &globalngv1.ResumeResourcePoolResponse{OperationId: request.GetOperationId()}, nil
	}

	_, berr = s.acceptPauseResume(
		ctx,
		resourcePool,
		request.GetOperationId(),
		resumeOperationType,
		globalngv1.ResourcePool_PAUSED,
		globalngv1.ResourcePool_RESUMING,
		common.ResumeResourcePool,
	)
	if berr != nil {
		return nil, berr
	}
	return &globalngv1.ResumeResourcePoolResponse{OperationId: request.GetOperationId()}, nil
}

func (s *serviceImpl) checkLocalPauseResumeOperation(
	ctx context.Context,
	resourcePoolID uint64,
	operationID string,
	operationType entity.ResourcePoolPauseOperationType,
) (bool, bizerr.BizError) {
	if s.pauseOperationRepo == nil {
		return false, nil
	}
	operation, berr := s.pauseOperationRepo.Get(ctx, resourcePoolID, operationID)
	if berr != nil {
		if berr.ErrCode() == bizerr.ErrCodeRecordNotFound {
			return false, nil
		}
		return false, berr
	}
	if operation.Type != operationType {
		return false, bizerr.New(
			bizerr.ErrCode(codes.FailedPrecondition),
			"resource pool operation %q is already owned by %s",
			operationID,
			strings.ToUpper(string(operation.Type)),
		)
	}
	return true, nil
}

func (s *serviceImpl) acceptPauseResume(
	ctx context.Context,
	resourcePool *entity.ResourcePool,
	operationID string,
	operationType string,
	expectedState globalngv1.ResourcePool_LifecycleState,
	acceptedState globalngv1.ResourcePool_LifecycleState,
	dagID string,
) ([]*globalngv1.ResourcePoolPauseValidationResult, bizerr.BizError) {
	transactionValidationResults := make([]*globalngv1.ResourcePoolPauseValidationResult, 0)
	err := s.transaction.Transaction(ctx, func(ctx context.Context) error {
		locked, berr := s.resourcePoolRepo.LoadForUpdate(ctx, resourcePool.Id, false)
		if berr != nil {
			return berr
		}
		localType := entity.ResourcePoolPauseOperationTypePause
		if operationType == resumeOperationType {
			localType = entity.ResourcePoolPauseOperationTypeResume
		}
		replayed, berr := s.checkLocalPauseResumeOperation(ctx, locked.Id, operationID, localType)
		if berr != nil {
			return berr
		}
		if replayed {
			return nil
		}
		if locked.GetLifecycleState() != expectedState {
			if operationType == pauseOperationType {
				transactionValidationResults = validatePauseLifecycleState(locked)
				return nil
			}
			return bizerr.New(
				bizerr.ErrCode(codes.FailedPrecondition),
				"resource pool lifecycle state is %s, expected %s",
				locked.GetLifecycleState().String(),
				expectedState.String(),
			)
		}
		if operationType == pauseOperationType {
			transactionValidationResults = normalizePauseValidationResults(s.validatePauseGlobal(ctx, locked))
			if len(transactionValidationResults) > 0 {
				return nil
			}
		}
		if s.pauseOperationRepo != nil {
			if berr := s.pauseOperationRepo.Create(ctx, &entity.ResourcePoolPauseOperation{
				ResourcePoolID: locked.Id,
				OperationID:    operationID,
				Type:           localType,
				Status:         entity.ResourcePoolPauseOperationRunning,
				Stage:          entity.ResourcePoolPauseOperationRegional,
				DeadlineAt:     time.Now().UTC(),
			}); berr != nil {
				return berr
			}
		}

		locked.LifecycleState = acceptedState
		locked.StateMessageCode = ""
		if operationType == pauseOperationType {
			locked.PausedAt = nil
		}
		locked.AutomaticResumeAt = nil
		if berr := s.resourcePoolRepo.Save(ctx, locked); berr != nil {
			return berr
		}
		startedEventType := clusterevent.EventTypePausing
		if operationType == resumeOperationType {
			startedEventType = clusterevent.EventTypeResuming
		}
		ev, berr := entity.NewResourcePoolLifecycleEvent(ctx, locked, startedEventType)
		if berr != nil {
			return berr
		}
		if berr := s.eventCommitter.Write(ctx, clusterevent.TopicClusterEvent, ev); berr != nil {
			return berr
		}

		if err := s.workflow.RunDagWithBizId(
			ctx,
			dagID,
			strconv.FormatUint(locked.OwnerId, 10),
			"0",
			strconv.FormatUint(locked.Id, 10),
			map[string]string{
				workflow.KeyTraceId:      workflow.GetTraceId(ctx),
				common.KeyResourcePoolId: strconv.FormatUint(locked.Id, 10),
				common.KeyOperationId:    operationID,
			},
		); err != nil {
			return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "create pause/resume workflow failed")
		}
		return nil
	})
	if err == nil {
		return transactionValidationResults, nil
	}
	var businessErr bizerr.BizError
	if errors.As(err, &businessErr) {
		return nil, businessErr
	}
	return nil, bizerr.WrapGormError(err, "pause/resume workflow transaction failed")
}