package resourcepool

import (
	"context"
	"fmt"
	"strings"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	clusterngv1beta "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	fastflowentity "github.com/tidbcloud/fastflow/pkg/entity"
	"go.uber.org/mock/gomock"
	"google.golang.org/grpc/codes"
	"google.golang.org/protobuf/types/known/timestamppb"

	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	mockeventbus "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/adapter/eventbus/mock_eventbus"
	mockregionalserver "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/adapter/regionalserver/mock_regionalserver"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/entity"
	mockbrrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/br/mock_br"
	clusterrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/cluster"
	mockcluster "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/cluster/mock_cluster"
	exportrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/export"
	mockexport "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/export/mock_export"
	mockregion "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/region/mock_region"
	mockresourcepool "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/resourcepool/mock_resourcepool"
	"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"
	mockworkflow "github.com/tidbcloud/tidb-management-service/utilities/workflow/mock_workflow"
)

func expectResourcePoolServiceLifecycleEvent(
	t *testing.T,
	eventCommitter *mockeventbus.MockWALCommitter,
	eventType clusterevent.EventType,
	status string,
) {
	t.Helper()
	eventCommitter.EXPECT().Write(gomock.Any(), clusterevent.TopicClusterEvent, gomock.Any()).DoAndReturn(
		func(_ context.Context, _ clusterevent.Topic, event clusterevent.Event) bizerr.BizError {
			clusterEvent, ok := event.(*clusterevent.ClusterEvent)
			require.True(t, ok)
			require.Equal(t, eventType, clusterEvent.EventType)
			require.Equal(t, status, clusterEvent.Status)
			require.Equal(t, clusterevent.ClusterTypeTidbxResourcePool, clusterEvent.ClusterType)
			return nil
		},
	)
}

func TestPauseResumeResourcePoolRejectsOperationIDOver128Bytes(t *testing.T) {
	t.Parallel()

	operationID := strings.Repeat("a", 129)
	service := &serviceImpl{}

	pauseResponse, pauseErr := service.PauseResourcePool(t.Context(), &globalngv1.PauseResourcePoolRequest{
		ResourcePoolId: 100,
		OperationId:    operationID,
	})
	resumeResponse, resumeErr := service.ResumeResourcePool(t.Context(), &globalngv1.ResumeResourcePoolRequest{
		ResourcePoolId: 100,
		OperationId:    operationID,
	})

	require.Nil(t, pauseResponse)
	require.Equal(t, bizerr.ErrCodeParams, pauseErr.ErrCode())
	require.ErrorContains(t, pauseErr, "operation id must not exceed 128 bytes")
	require.Nil(t, resumeResponse)
	require.Equal(t, bizerr.ErrCodeParams, resumeErr.ErrCode())
	require.ErrorContains(t, resumeErr, "operation id must not exceed 128 bytes")
}

func TestAcceptPauseResumeWritesStartedLifecycleEvent(t *testing.T) {
	tests := []struct {
		name          string
		operationType string
		localType     entity.ResourcePoolPauseOperationType
		expectedState globalngv1.ResourcePool_LifecycleState
		acceptedState globalngv1.ResourcePool_LifecycleState
		dagID         string
		eventType     clusterevent.EventType
		status        string
	}{
		{
			name:          "pause",
			operationType: pauseOperationType,
			localType:     entity.ResourcePoolPauseOperationTypePause,
			expectedState: globalngv1.ResourcePool_ACTIVE,
			acceptedState: globalngv1.ResourcePool_PAUSING,
			dagID:         common.PauseResourcePool,
			eventType:     clusterevent.EventTypePausing,
			status:        "CLUSTER_STATUS_PAUSING",
		},
		{
			name:          "resume",
			operationType: resumeOperationType,
			localType:     entity.ResourcePoolPauseOperationTypeResume,
			expectedState: globalngv1.ResourcePool_PAUSED,
			acceptedState: globalngv1.ResourcePool_RESUMING,
			dagID:         common.ResumeResourcePool,
			eventType:     clusterevent.EventTypeResuming,
			status:        "CLUSTER_STATUS_RESUMING",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			t.Parallel()

			ctrl := gomock.NewController(t)
			resourcePoolRepo := mockresourcepool.NewMockRepo(ctrl)
			clusterRepo := mockcluster.NewMockRepo(ctrl)
			pauseOperationRepo := mockresourcepool.NewMockPauseOperationRepo(ctrl)
			flow := mockworkflow.NewMockWorkflow(ctrl)
			eventCommitter := mockeventbus.NewMockWALCommitter(ctrl)
			service := &serviceImpl{
				transaction:        resourcePoolPassthroughTransactioner{},
				workflow:           flow,
				resourcePoolRepo:   resourcePoolRepo,
				clusterRepo:        clusterRepo,
				pauseOperationRepo: pauseOperationRepo,
				eventCommitter:     eventCommitter,
			}
			pool := &entity.ResourcePool{ResourcePool: &globalngv1.ResourcePool{
				Id:             100,
				OwnerId:        42,
				LifecycleState: tt.expectedState,
			}}

			resourcePoolRepo.EXPECT().LoadForUpdate(gomock.Any(), uint64(100), false).Return(pool, nil)
			pauseOperationRepo.EXPECT().Get(gomock.Any(), uint64(100), "operation-1").Return(
				nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "operation not found"),
			)
			if tt.operationType == pauseOperationType {
				clusterRepo.EXPECT().Query(
					gomock.Any(), &clusterrepo.Filter{SharedPoolIds: []uint64{100}},
				).Return(nil, nil)
			}
			pauseOperationRepo.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(
				func(_ context.Context, operation *entity.ResourcePoolPauseOperation) bizerr.BizError {
					require.Equal(t, tt.localType, operation.Type)
					return nil
				},
			)
			resourcePoolRepo.EXPECT().Save(gomock.Any(), pool).Return(nil)
			expectResourcePoolServiceLifecycleEvent(t, eventCommitter, tt.eventType, tt.status)
			flow.EXPECT().RunDagWithBizId(gomock.Any(), tt.dagID, "42", "0", "100", gomock.Any()).Return(nil)

			validationResults, berr := service.acceptPauseResume(
				t.Context(), pool, "operation-1", tt.operationType, tt.expectedState, tt.acceptedState, tt.dagID,
			)

			require.NoError(t, berr)
			require.Empty(t, validationResults)
			require.Equal(t, tt.acceptedState, pool.GetLifecycleState())
		})
	}
}

func TestBeginPauseResourcePoolAdmissionReservesPoolWithoutStartingWorkflow(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	resourcePoolRepo := mockresourcepool.NewMockRepo(ctrl)
	pauseOperationRepo := mockresourcepool.NewMockPauseOperationRepo(ctrl)
	clusterRepo := mockcluster.NewMockRepo(ctrl)
	regionRepo := mockregion.NewMockRepo(ctrl)
	regionalClient := mockregionalserver.NewMockClient(ctrl)
	flow := mockworkflow.NewMockWorkflow(ctrl)
	service := &serviceImpl{
		transaction:        resourcePoolPassthroughTransactioner{},
		workflow:           flow,
		resourcePoolRepo:   resourcePoolRepo,
		pauseOperationRepo: pauseOperationRepo,
		clusterRepo:        clusterRepo,
		regionRepo:         regionRepo,
		regionalCli:        regionalClient,
	}
	pool := &entity.ResourcePool{ResourcePool: &globalngv1.ResourcePool{
		Id:             100,
		OwnerId:        42,
		LifecycleState: globalngv1.ResourcePool_ACTIVE,
	}, RegionId: 10}
	notFound := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
	resourcePoolRepo.EXPECT().Load(gomock.Any(), uint64(100), false).Return(pool, nil)
	pauseOperationRepo.EXPECT().Get(gomock.Any(), uint64(100), "pause-1").Return(nil, notFound).Times(2)
	clusterRepo.EXPECT().Query(gomock.Any(), &clusterrepo.Filter{SharedPoolIds: []uint64{100}}).Return(nil, nil).Times(2)
	regionRepo.EXPECT().Get(gomock.Any(), uint64(10)).Return(&entity.Region{}, nil)
	regionalClient.EXPECT().ValidateSharedPoolPause(gomock.Any(), gomock.Any(), gomock.Any()).Return(
		&clusterngv1beta.ValidateSharedPoolPauseResponse{}, nil,
	)
	resourcePoolRepo.EXPECT().LoadForUpdate(gomock.Any(), uint64(100), false).Return(pool, nil)
	pauseOperationRepo.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(
		func(_ context.Context, operation *entity.ResourcePoolPauseOperation) bizerr.BizError {
			require.Equal(t, entity.ResourcePoolPauseOperationAdmission, operation.Stage)
			require.Equal(t, entity.ResourcePoolPauseOperationRunning, operation.Status)
			return nil
		},
	)
	resourcePoolRepo.EXPECT().Save(gomock.Any(), pool).Return(nil)

	response, berr := service.BeginPauseResourcePoolAdmission(t.Context(),
		&globalngv1.BeginPauseResourcePoolAdmissionRequest{ResourcePoolId: 100, OperationId: "pause-1"})

	require.NoError(t, berr)
	require.Equal(t, "pause-1", response.GetOperationId())
	require.Equal(t, globalngv1.ResourcePool_PAUSE_PRECHECKING, pool.GetLifecycleState())
}

func TestBeginPauseResourcePoolAdmissionRejectsOperationIDOwnedByResume(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	resourcePoolRepo := mockresourcepool.NewMockRepo(ctrl)
	pauseOperationRepo := mockresourcepool.NewMockPauseOperationRepo(ctrl)
	service := &serviceImpl{
		resourcePoolRepo:   resourcePoolRepo,
		pauseOperationRepo: pauseOperationRepo,
	}
	pool := &entity.ResourcePool{ResourcePool: &globalngv1.ResourcePool{Id: 100}}
	resourcePoolRepo.EXPECT().Load(gomock.Any(), uint64(100), false).Return(pool, nil)
	pauseOperationRepo.EXPECT().Get(gomock.Any(), uint64(100), "resume-1").Return(
		&entity.ResourcePoolPauseOperation{
			ResourcePoolID: 100,
			OperationID:    "resume-1",
			Type:           entity.ResourcePoolPauseOperationTypeResume,
		}, nil,
	)

	response, berr := service.BeginPauseResourcePoolAdmission(t.Context(),
		&globalngv1.BeginPauseResourcePoolAdmissionRequest{ResourcePoolId: 100, OperationId: "resume-1"})

	require.Nil(t, response)
	require.Equal(t, bizerr.ErrCode(codes.FailedPrecondition), berr.ErrCode())
	require.ErrorContains(t, berr, "already owned by RESUME")
}

func TestBeginPauseResourcePoolAdmissionRejectsRacedOperationIDOwnedByResume(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	resourcePoolRepo := mockresourcepool.NewMockRepo(ctrl)
	pauseOperationRepo := mockresourcepool.NewMockPauseOperationRepo(ctrl)
	clusterRepo := mockcluster.NewMockRepo(ctrl)
	regionRepo := mockregion.NewMockRepo(ctrl)
	regionalClient := mockregionalserver.NewMockClient(ctrl)
	service := &serviceImpl{
		transaction:        resourcePoolPassthroughTransactioner{},
		resourcePoolRepo:   resourcePoolRepo,
		pauseOperationRepo: pauseOperationRepo,
		clusterRepo:        clusterRepo,
		regionRepo:         regionRepo,
		regionalCli:        regionalClient,
	}
	pool := &entity.ResourcePool{ResourcePool: &globalngv1.ResourcePool{
		Id:             100,
		LifecycleState: globalngv1.ResourcePool_ACTIVE,
	}, RegionId: 10}
	resumeOperation := &entity.ResourcePoolPauseOperation{
		ResourcePoolID: 100,
		OperationID:    "resume-1",
		Type:           entity.ResourcePoolPauseOperationTypeResume,
	}
	notFound := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
	resourcePoolRepo.EXPECT().Load(gomock.Any(), uint64(100), false).Return(pool, nil)
	pauseOperationRepo.EXPECT().Get(gomock.Any(), uint64(100), "resume-1").Return(nil, notFound)
	clusterRepo.EXPECT().Query(gomock.Any(), &clusterrepo.Filter{SharedPoolIds: []uint64{100}}).Return(nil, nil)
	regionRepo.EXPECT().Get(gomock.Any(), uint64(10)).Return(&entity.Region{}, nil)
	regionalClient.EXPECT().ValidateSharedPoolPause(gomock.Any(), gomock.Any(), gomock.Any()).Return(
		&clusterngv1beta.ValidateSharedPoolPauseResponse{}, nil,
	)
	resourcePoolRepo.EXPECT().LoadForUpdate(gomock.Any(), uint64(100), false).Return(pool, nil)
	pauseOperationRepo.EXPECT().Get(gomock.Any(), uint64(100), "resume-1").Return(resumeOperation, nil)

	response, berr := service.BeginPauseResourcePoolAdmission(t.Context(),
		&globalngv1.BeginPauseResourcePoolAdmissionRequest{ResourcePoolId: 100, OperationId: "resume-1"})

	require.Nil(t, response)
	require.Equal(t, bizerr.ErrCode(codes.FailedPrecondition), berr.ErrCode())
	require.ErrorContains(t, berr, "already owned by RESUME")
}

func TestResolvePauseResourcePoolAdmissionAbortRestoresActive(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	resourcePoolRepo := mockresourcepool.NewMockRepo(ctrl)
	pauseOperationRepo := mockresourcepool.NewMockPauseOperationRepo(ctrl)
	service := &serviceImpl{
		transaction:        resourcePoolPassthroughTransactioner{},
		resourcePoolRepo:   resourcePoolRepo,
		pauseOperationRepo: pauseOperationRepo,
	}
	pool := &entity.ResourcePool{ResourcePool: &globalngv1.ResourcePool{
		Id:             100,
		LifecycleState: globalngv1.ResourcePool_PAUSE_PRECHECKING,
	}}
	operation := &entity.ResourcePoolPauseOperation{
		ResourcePoolID: 100,
		OperationID:    "pause-1",
		Status:         entity.ResourcePoolPauseOperationRunning,
		Stage:          entity.ResourcePoolPauseOperationAdmission,
	}
	resourcePoolRepo.EXPECT().LoadForUpdate(gomock.Any(), uint64(100), false).Return(pool, nil)
	pauseOperationRepo.EXPECT().Get(gomock.Any(), uint64(100), "pause-1").Return(operation, nil)
	resourcePoolRepo.EXPECT().Save(gomock.Any(), pool).Return(nil)
	pauseOperationRepo.EXPECT().Save(gomock.Any(), operation).DoAndReturn(
		func(_ context.Context, saved *entity.ResourcePoolPauseOperation) bizerr.BizError {
			require.Equal(t, entity.ResourcePoolPauseOperationFailed, saved.Status)
			require.Equal(t, "CHANGEFEED_IN_PROGRESS: resource pool has a running changefeed", saved.ResultSummary)
			require.NotNil(t, saved.CompletedAt)
			return nil
		},
	)

	_, berr := service.ResolvePauseResourcePoolAdmission(t.Context(),
		&globalngv1.ResolvePauseResourcePoolAdmissionRequest{
			ResourcePoolId: 100,
			OperationId:    "pause-1",
			Proceed:        false,
			ValidationResults: []*globalngv1.ResourcePoolPauseValidationResult{{
				CheckCode: "CHANGEFEED_IN_PROGRESS",
				Detail:    "resource pool has a running changefeed",
			}},
		})

	require.NoError(t, berr)
	require.Equal(t, globalngv1.ResourcePool_ACTIVE, pool.GetLifecycleState())
}

func TestResolvePauseResourcePoolAdmissionRejectsFailedAdmission(t *testing.T) {
	t.Parallel()

	for _, proceed := range []bool{false, true} {
		t.Run(fmt.Sprintf("proceed=%t", proceed), func(t *testing.T) {
			t.Parallel()

			ctrl := gomock.NewController(t)
			resourcePoolRepo := mockresourcepool.NewMockRepo(ctrl)
			pauseOperationRepo := mockresourcepool.NewMockPauseOperationRepo(ctrl)
			clusterRepo := mockcluster.NewMockRepo(ctrl)
			service := &serviceImpl{
				transaction:        resourcePoolPassthroughTransactioner{},
				resourcePoolRepo:   resourcePoolRepo,
				pauseOperationRepo: pauseOperationRepo,
				clusterRepo:        clusterRepo,
			}
			pool := &entity.ResourcePool{ResourcePool: &globalngv1.ResourcePool{
				Id:             100,
				LifecycleState: globalngv1.ResourcePool_PAUSE_PRECHECKING,
			}}
			failedAdmission := &entity.ResourcePoolPauseOperation{
				ResourcePoolID: 100,
				OperationID:    "pause-a",
				Status:         entity.ResourcePoolPauseOperationFailed,
				Stage:          entity.ResourcePoolPauseOperationAdmission,
			}
			resourcePoolRepo.EXPECT().LoadForUpdate(gomock.Any(), uint64(100), false).Return(pool, nil)
			pauseOperationRepo.EXPECT().Get(gomock.Any(), uint64(100), "pause-a").Return(failedAdmission, nil)

			_, berr := service.ResolvePauseResourcePoolAdmission(t.Context(),
				&globalngv1.ResolvePauseResourcePoolAdmissionRequest{
					ResourcePoolId: 100,
					OperationId:    "pause-a",
					Proceed:        proceed,
				})

			require.Error(t, berr)
			require.Equal(t, bizerr.ErrCode(codes.FailedPrecondition), berr.ErrCode())
			require.Equal(t, globalngv1.ResourcePool_PAUSE_PRECHECKING, pool.GetLifecycleState())
			require.Equal(t, entity.ResourcePoolPauseOperationFailed, failedAdmission.Status)
			require.Equal(t, entity.ResourcePoolPauseOperationAdmission, failedAdmission.Stage)
		})
	}
}

func TestResolvePauseResourcePoolAdmissionExtendsManualBackupDeadline(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	resourcePoolRepo := mockresourcepool.NewMockRepo(ctrl)
	pauseOperationRepo := mockresourcepool.NewMockPauseOperationRepo(ctrl)
	flow := mockworkflow.NewMockWorkflow(ctrl)
	eventCommitter := mockeventbus.NewMockWALCommitter(ctrl)
	service := &serviceImpl{
		transaction:        resourcePoolPassthroughTransactioner{},
		workflow:           flow,
		resourcePoolRepo:   resourcePoolRepo,
		pauseOperationRepo: pauseOperationRepo,
		eventCommitter:     eventCommitter,
	}
	pool := &entity.ResourcePool{ResourcePool: &globalngv1.ResourcePool{
		Id:             100,
		OwnerId:        42,
		LifecycleState: globalngv1.ResourcePool_PAUSE_PRECHECKING,
	}}
	operation := &entity.ResourcePoolPauseOperation{
		ResourcePoolID: 100,
		OperationID:    "pause-1",
		Status:         entity.ResourcePoolPauseOperationRunning,
		Stage:          entity.ResourcePoolPauseOperationAdmissionWithManualBackup,
		DeadlineAt:     time.Now().UTC().Add(pauseAdmissionTimeout),
	}
	resourcePoolRepo.EXPECT().LoadForUpdate(gomock.Any(), uint64(100), false).Return(pool, nil)
	pauseOperationRepo.EXPECT().Get(gomock.Any(), uint64(100), "pause-1").Return(operation, nil)
	pauseOperationRepo.EXPECT().Save(gomock.Any(), operation).DoAndReturn(
		func(_ context.Context, saved *entity.ResourcePoolPauseOperation) bizerr.BizError {
			require.Equal(t, entity.ResourcePoolPauseOperationManualBackup, saved.Stage)
			require.WithinDuration(t, time.Now().UTC().Add(common.PausePreparationTimeout), saved.DeadlineAt, time.Second)
			return nil
		},
	)
	flow.EXPECT().RunDagWithBizId(
		gomock.Any(),
		common.PrepareAndPauseResourcePool,
		"42",
		"0",
		"100",
		gomock.Any(),
	).Return(nil)

	response, berr := service.ResolvePauseResourcePoolAdmission(t.Context(),
		&globalngv1.ResolvePauseResourcePoolAdmissionRequest{
			ResourcePoolId: 100,
			OperationId:    "pause-1",
			Proceed:        true,
		})

	require.NoError(t, berr)
	require.Equal(t, "pause-1", response.GetOperationId())
}

func TestExpirePauseAdmissionsRestoresActive(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	resourcePoolRepo := mockresourcepool.NewMockRepo(ctrl)
	pauseOperationRepo := mockresourcepool.NewMockPauseOperationRepo(ctrl)
	service := &serviceImpl{
		transaction:        resourcePoolPassthroughTransactioner{},
		resourcePoolRepo:   resourcePoolRepo,
		pauseOperationRepo: pauseOperationRepo,
	}
	pool := &entity.ResourcePool{ResourcePool: &globalngv1.ResourcePool{
		Id:             100,
		LifecycleState: globalngv1.ResourcePool_PAUSE_PRECHECKING,
	}}
	operation := &entity.ResourcePoolPauseOperation{
		ResourcePoolID: 100,
		OperationID:    "pause-1",
		Status:         entity.ResourcePoolPauseOperationRunning,
		Stage:          entity.ResourcePoolPauseOperationAdmission,
		DeadlineAt:     time.Now().Add(-time.Minute),
	}
	pauseOperationRepo.EXPECT().ListExpiredAdmissions(gomock.Any(), gomock.Any()).Return(
		[]*entity.ResourcePoolPauseOperation{operation}, nil,
	)
	resourcePoolRepo.EXPECT().LoadForUpdate(gomock.Any(), uint64(100), false).Return(pool, nil)
	pauseOperationRepo.EXPECT().Get(gomock.Any(), uint64(100), "pause-1").Return(operation, nil)
	resourcePoolRepo.EXPECT().Save(gomock.Any(), pool).Return(nil)
	pauseOperationRepo.EXPECT().Save(gomock.Any(), operation).DoAndReturn(
		func(_ context.Context, saved *entity.ResourcePoolPauseOperation) bizerr.BizError {
			require.Equal(t, entity.ResourcePoolPauseOperationFailed, saved.Status)
			require.Equal(t, "pause admission timed out", saved.ResultSummary)
			return nil
		},
	)

	berr := service.ExpirePauseAdmissions(t.Context())

	require.NoError(t, berr)
	require.Equal(t, globalngv1.ResourcePool_ACTIVE, pool.GetLifecycleState())
}

func TestPauseResourcePoolReplaysHistoricalRegionalOperation(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	resourcePoolRepo := mockresourcepool.NewMockRepo(ctrl)
	pauseOperationRepo := mockresourcepool.NewMockPauseOperationRepo(ctrl)
	service := &serviceImpl{
		resourcePoolRepo:   resourcePoolRepo,
		pauseOperationRepo: pauseOperationRepo,
	}
	resourcePool := &entity.ResourcePool{
		ResourcePool: &globalngv1.ResourcePool{
			Id:             100,
			LifecycleState: globalngv1.ResourcePool_ACTIVE,
		},
		RegionId: 10,
	}
	resourcePoolRepo.EXPECT().Load(gomock.Any(), uint64(100), false).Return(resourcePool, nil)
	pauseOperationRepo.EXPECT().Get(gomock.Any(), uint64(100), "pause-1").Return(
		&entity.ResourcePoolPauseOperation{
			ResourcePoolID: 100,
			OperationID:    "pause-1",
			Type:           entity.ResourcePoolPauseOperationTypePause,
			Status:         entity.ResourcePoolPauseOperationSucceeded,
		}, nil,
	)

	response, berr := service.PauseResourcePool(t.Context(), &globalngv1.PauseResourcePoolRequest{
		ResourcePoolId: 100,
		OperationId:    "pause-1",
	})

	require.NoError(t, berr)
	require.Equal(t, "pause-1", response.GetOperationId())
	require.Equal(t, globalngv1.ResourcePool_ACTIVE, resourcePool.GetLifecycleState())
}

func TestPauseResourcePoolValidateOnlyReturnsLifecycleBlockerWithoutOperationID(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	resourcePoolRepo := mockresourcepool.NewMockRepo(ctrl)
	clusterRepo := mockcluster.NewMockRepo(ctrl)
	regionRepo := mockregion.NewMockRepo(ctrl)
	regionalClient := mockregionalserver.NewMockClient(ctrl)
	service := &serviceImpl{
		resourcePoolRepo: resourcePoolRepo,
		clusterRepo:      clusterRepo,
		regionRepo:       regionRepo,
		regionalCli:      regionalClient,
	}
	region := &entity.Region{}
	resourcePoolRepo.EXPECT().Load(gomock.Any(), uint64(100), false).Return(&entity.ResourcePool{
		ResourcePool: &globalngv1.ResourcePool{
			Id:             100,
			LifecycleState: globalngv1.ResourcePool_PAUSED,
		},
		RegionId: 10,
	}, nil)
	clusterRepo.EXPECT().Query(gomock.Any(), &clusterrepo.Filter{SharedPoolIds: []uint64{100}}).Return(nil, nil)
	regionRepo.EXPECT().Get(gomock.Any(), uint64(10)).Return(region, nil)
	regionalClient.EXPECT().ValidateSharedPoolPause(
		gomock.Any(),
		region,
		&clusterngv1beta.ValidateSharedPoolPauseRequest{SharedPoolId: "100"},
	).Return(&clusterngv1beta.ValidateSharedPoolPauseResponse{}, nil)

	response, berr := service.PauseResourcePool(t.Context(), &globalngv1.PauseResourcePoolRequest{
		ResourcePoolId: 100,
		ValidateOnly:   true,
	})

	require.NoError(t, berr)
	require.Empty(t, response.GetOperationId())
	require.Equal(t, []*globalngv1.ResourcePoolPauseValidationResult{{
		CheckCode: "INVALID_LIFECYCLE_STATE",
		Detail:    "resource pool must be ACTIVE to pause",
	}}, response.GetValidationResults())
}

func TestPauseResourcePoolWithManualBackupEntersPrecheckingAtomically(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	resourcePoolRepo := mockresourcepool.NewMockRepo(ctrl)
	pauseOperationRepo := mockresourcepool.NewMockPauseOperationRepo(ctrl)
	clusterRepo := mockcluster.NewMockRepo(ctrl)
	regionRepo := mockregion.NewMockRepo(ctrl)
	regionalClient := mockregionalserver.NewMockClient(ctrl)
	flow := mockworkflow.NewMockWorkflow(ctrl)
	service := &serviceImpl{
		transaction:        resourcePoolPassthroughTransactioner{},
		workflow:           flow,
		resourcePoolRepo:   resourcePoolRepo,
		pauseOperationRepo: pauseOperationRepo,
		clusterRepo:        clusterRepo,
		regionRepo:         regionRepo,
		regionalCli:        regionalClient,
	}
	pool := &entity.ResourcePool{
		ResourcePool: &globalngv1.ResourcePool{
			Id:             100,
			OwnerId:        42,
			LifecycleState: globalngv1.ResourcePool_ACTIVE,
		},
		RegionId: 10,
	}
	region := &entity.Region{}
	resourcePoolRepo.EXPECT().Load(gomock.Any(), uint64(100), false).Return(pool, nil)
	pauseOperationRepo.EXPECT().Get(gomock.Any(), uint64(100), "pause-1").Return(
		nil,
		bizerr.New(bizerr.ErrCodeRecordNotFound, "not found"),
	)
	clusterRepo.EXPECT().Query(gomock.Any(), &clusterrepo.Filter{SharedPoolIds: []uint64{100}}).Return(nil, nil).Times(2)
	regionRepo.EXPECT().Get(gomock.Any(), uint64(10)).Return(region, nil)
	regionalClient.EXPECT().ValidateSharedPoolPause(
		gomock.Any(),
		region,
		&clusterngv1beta.ValidateSharedPoolPauseRequest{SharedPoolId: "100"},
	).Return(&clusterngv1beta.ValidateSharedPoolPauseResponse{}, nil)
	resourcePoolRepo.EXPECT().LoadForUpdate(gomock.Any(), uint64(100), false).Return(pool, nil)
	pauseOperationRepo.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(
		func(_ context.Context, operation *entity.ResourcePoolPauseOperation) bizerr.BizError {
			require.Equal(t, entity.ResourcePoolPauseOperationRunning, operation.Status)
			require.Equal(t, entity.ResourcePoolPauseOperationManualBackup, operation.Stage)
			require.WithinDuration(t, time.Now().Add(common.PausePreparationTimeout), operation.DeadlineAt, time.Second)
			return nil
		},
	)
	resourcePoolRepo.EXPECT().Save(gomock.Any(), pool).Return(nil)
	flow.EXPECT().RunDagWithBizId(
		gomock.Any(),
		common.PrepareAndPauseResourcePool,
		"42",
		"0",
		"100",
		gomock.Any(),
	).Return(nil)

	response, berr := service.PauseResourcePool(t.Context(), &globalngv1.PauseResourcePoolRequest{
		ResourcePoolId:                100,
		OperationId:                   "pause-1",
		CreateManualBackupBeforePause: true,
	})

	require.NoError(t, berr)
	require.Equal(t, "pause-1", response.GetOperationId())
	require.Equal(t, globalngv1.ResourcePool_PAUSE_PRECHECKING, pool.GetLifecycleState())
}

func TestPauseResourcePoolWithManualBackupRejectsInvalidInstanceBeforeCreatingOperation(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	resourcePoolRepo := mockresourcepool.NewMockRepo(ctrl)
	pauseOperationRepo := mockresourcepool.NewMockPauseOperationRepo(ctrl)
	clusterRepo := mockcluster.NewMockRepo(ctrl)
	exportRepo := mockexport.NewMockExportRepo(ctrl)
	regionRepo := mockregion.NewMockRepo(ctrl)
	regionalClient := mockregionalserver.NewMockClient(ctrl)
	flow := mockworkflow.NewMockWorkflow(ctrl)
	service := &serviceImpl{
		resourcePoolRepo:   resourcePoolRepo,
		pauseOperationRepo: pauseOperationRepo,
		clusterRepo:        clusterRepo,
		exportRepo:         exportRepo,
		regionRepo:         regionRepo,
		regionalCli:        regionalClient,
		workflow:           flow,
	}
	pool := &entity.ResourcePool{
		ResourcePool: &globalngv1.ResourcePool{
			Id:             100,
			LifecycleState: globalngv1.ResourcePool_ACTIVE,
		},
		RegionId: 10,
	}
	clusters := []*entity.Cluster{{Cluster: &globalngv1.Cluster{
		ClusterId: 200,
		State:     globalngv1.Cluster_CREATING,
	}}}
	region := &entity.Region{}

	resourcePoolRepo.EXPECT().Load(gomock.Any(), uint64(100), false).Return(pool, nil)
	pauseOperationRepo.EXPECT().Get(gomock.Any(), uint64(100), "pause-1").Return(
		nil,
		bizerr.New(bizerr.ErrCodeRecordNotFound, "not found"),
	)
	clusterRepo.EXPECT().Query(gomock.Any(), &clusterrepo.Filter{SharedPoolIds: []uint64{100}}).Return(clusters, nil)
	exportRepo.EXPECT().Count(gomock.Any(), &exportrepo.QueryExportsFilter{
		ClusterIds: []string{"200"},
		States:     []schema.ExportState{schema.ExportStateRunning},
	}).Return(int64(0), nil)
	flow.EXPECT().ListRunningDagInstancesByBizIds(
		gomock.Any(),
		pauseBlockingGlobalDagIDs,
		[]string{"200"},
	).Return(nil, nil)
	regionRepo.EXPECT().Get(gomock.Any(), uint64(10)).Return(region, nil)
	regionalClient.EXPECT().ValidateSharedPoolPause(
		gomock.Any(),
		region,
		&clusterngv1beta.ValidateSharedPoolPauseRequest{SharedPoolId: "100"},
	).Return(&clusterngv1beta.ValidateSharedPoolPauseResponse{}, nil)

	response, berr := service.PauseResourcePool(t.Context(), &globalngv1.PauseResourcePoolRequest{
		ResourcePoolId:                100,
		OperationId:                   "pause-1",
		CreateManualBackupBeforePause: true,
	})

	require.NoError(t, berr)
	require.Empty(t, response.GetOperationId())
	require.Equal(t, []*globalngv1.ResourcePoolPauseValidationResult{{
		CheckCode: instanceStateConflictCheckCode,
		Detail:    "all resource pool instances must be ACTIVE to pause",
	}}, response.GetValidationResults())
	require.Equal(t, globalngv1.ResourcePool_ACTIVE, pool.GetLifecycleState())
}

func TestPauseResourcePoolWithManualBackupRejectsInstanceCreatedAfterAdmissionValidation(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	resourcePoolRepo := mockresourcepool.NewMockRepo(ctrl)
	pauseOperationRepo := mockresourcepool.NewMockPauseOperationRepo(ctrl)
	clusterRepo := mockcluster.NewMockRepo(ctrl)
	restoreRepo := mockbrrepo.NewMockRestoreRepo(ctrl)
	exportRepo := mockexport.NewMockExportRepo(ctrl)
	regionRepo := mockregion.NewMockRepo(ctrl)
	regionalClient := mockregionalserver.NewMockClient(ctrl)
	flow := mockworkflow.NewMockWorkflow(ctrl)
	service := &serviceImpl{
		transaction:        resourcePoolPassthroughTransactioner{},
		resourcePoolRepo:   resourcePoolRepo,
		pauseOperationRepo: pauseOperationRepo,
		clusterRepo:        clusterRepo,
		restoreRepo:        restoreRepo,
		exportRepo:         exportRepo,
		regionRepo:         regionRepo,
		regionalCli:        regionalClient,
		workflow:           flow,
	}
	pool := &entity.ResourcePool{
		ResourcePool: &globalngv1.ResourcePool{
			Id:             100,
			LifecycleState: globalngv1.ResourcePool_ACTIVE,
		},
		RegionId: 10,
	}
	availableCluster := []*entity.Cluster{{Cluster: &globalngv1.Cluster{
		ClusterId: 200,
		State:     globalngv1.Cluster_AVAILABLE,
	}}}
	creatingCluster := []*entity.Cluster{{Cluster: &globalngv1.Cluster{
		ClusterId: 200,
		State:     globalngv1.Cluster_CREATING,
	}}}
	region := &entity.Region{}

	resourcePoolRepo.EXPECT().Load(gomock.Any(), uint64(100), false).Return(pool, nil)
	pauseOperationRepo.EXPECT().Get(gomock.Any(), uint64(100), "pause-1").Return(
		nil,
		bizerr.New(bizerr.ErrCodeRecordNotFound, "not found"),
	)
	clusterRepo.EXPECT().Query(
		gomock.Any(),
		&clusterrepo.Filter{SharedPoolIds: []uint64{100}},
	).Return(availableCluster, nil)
	clusterRepo.EXPECT().Query(
		gomock.Any(),
		&clusterrepo.Filter{SharedPoolIds: []uint64{100}},
	).Return(creatingCluster, nil)
	exportRepo.EXPECT().Count(gomock.Any(), gomock.Any()).Return(int64(0), nil).Times(2)
	flow.EXPECT().ListRunningDagInstancesByBizIds(
		gomock.Any(),
		pauseBlockingGlobalDagIDs,
		[]string{"200"},
	).Return(nil, nil).Times(2)
	regionRepo.EXPECT().Get(gomock.Any(), uint64(10)).Return(region, nil)
	regionalClient.EXPECT().ValidateSharedPoolPause(
		gomock.Any(),
		region,
		&clusterngv1beta.ValidateSharedPoolPauseRequest{SharedPoolId: "100"},
	).Return(&clusterngv1beta.ValidateSharedPoolPauseResponse{}, nil)
	resourcePoolRepo.EXPECT().LoadForUpdate(gomock.Any(), uint64(100), false).Return(pool, nil)
	restoreRepo.EXPECT().HasActiveByTargetClusterIDs(gomock.Any(), []uint64{200}).Return(false, nil)

	response, berr := service.PauseResourcePool(t.Context(), &globalngv1.PauseResourcePoolRequest{
		ResourcePoolId:                100,
		OperationId:                   "pause-1",
		CreateManualBackupBeforePause: true,
	})

	require.NoError(t, berr)
	require.Empty(t, response.GetOperationId())
	require.Equal(t, []*globalngv1.ResourcePoolPauseValidationResult{{
		CheckCode: instanceStateConflictCheckCode,
		Detail:    "all resource pool instances must be ACTIVE to pause",
	}}, response.GetValidationResults())
	require.Equal(t, globalngv1.ResourcePool_ACTIVE, pool.GetLifecycleState())
}

func TestPauseResourcePoolWithManualBackupRejectsActiveRestoreUnderLock(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	resourcePoolRepo := mockresourcepool.NewMockRepo(ctrl)
	pauseOperationRepo := mockresourcepool.NewMockPauseOperationRepo(ctrl)
	clusterRepo := mockcluster.NewMockRepo(ctrl)
	restoreRepo := mockbrrepo.NewMockRestoreRepo(ctrl)
	exportRepo := mockexport.NewMockExportRepo(ctrl)
	regionRepo := mockregion.NewMockRepo(ctrl)
	regionalClient := mockregionalserver.NewMockClient(ctrl)
	flow := mockworkflow.NewMockWorkflow(ctrl)
	service := &serviceImpl{
		transaction:        resourcePoolPassthroughTransactioner{},
		resourcePoolRepo:   resourcePoolRepo,
		pauseOperationRepo: pauseOperationRepo,
		clusterRepo:        clusterRepo,
		restoreRepo:        restoreRepo,
		exportRepo:         exportRepo,
		regionRepo:         regionRepo,
		regionalCli:        regionalClient,
		workflow:           flow,
	}
	pool := &entity.ResourcePool{
		ResourcePool: &globalngv1.ResourcePool{
			Id:             100,
			LifecycleState: globalngv1.ResourcePool_ACTIVE,
		},
		RegionId: 10,
	}
	clusters := []*entity.Cluster{{Cluster: &globalngv1.Cluster{
		ClusterId: 200,
		State:     globalngv1.Cluster_AVAILABLE,
	}}}
	region := &entity.Region{}

	resourcePoolRepo.EXPECT().Load(gomock.Any(), uint64(100), false).Return(pool, nil)
	pauseOperationRepo.EXPECT().Get(gomock.Any(), uint64(100), "pause-1").Return(
		nil,
		bizerr.New(bizerr.ErrCodeRecordNotFound, "not found"),
	)
	clusterRepo.EXPECT().Query(
		gomock.Any(),
		&clusterrepo.Filter{SharedPoolIds: []uint64{100}},
	).Return(clusters, nil).Times(2)
	exportRepo.EXPECT().Count(gomock.Any(), gomock.Any()).Return(int64(0), nil).Times(2)
	flow.EXPECT().ListRunningDagInstancesByBizIds(
		gomock.Any(),
		pauseBlockingGlobalDagIDs,
		[]string{"200"},
	).Return(nil, nil).Times(2)
	regionRepo.EXPECT().Get(gomock.Any(), uint64(10)).Return(region, nil)
	regionalClient.EXPECT().ValidateSharedPoolPause(
		gomock.Any(),
		region,
		&clusterngv1beta.ValidateSharedPoolPauseRequest{SharedPoolId: "100"},
	).Return(&clusterngv1beta.ValidateSharedPoolPauseResponse{}, nil)
	resourcePoolRepo.EXPECT().LoadForUpdate(gomock.Any(), uint64(100), false).Return(pool, nil)
	restoreRepo.EXPECT().HasActiveByTargetClusterIDs(gomock.Any(), []uint64{200}).Return(true, nil)

	response, berr := service.PauseResourcePool(t.Context(), &globalngv1.PauseResourcePoolRequest{
		ResourcePoolId:                100,
		OperationId:                   "pause-1",
		CreateManualBackupBeforePause: true,
	})

	require.NoError(t, berr)
	require.Empty(t, response.GetOperationId())
	require.Equal(t, []*globalngv1.ResourcePoolPauseValidationResult{{
		CheckCode: operationInProgressCheckCode,
		Detail:    "resource pool has an active restore",
	}}, response.GetValidationResults())
	require.Equal(t, globalngv1.ResourcePool_ACTIVE, pool.GetLifecycleState())
}

func TestPauseResourcePoolWithManualBackupReplaysLocalOperation(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	resourcePoolRepo := mockresourcepool.NewMockRepo(ctrl)
	pauseOperationRepo := mockresourcepool.NewMockPauseOperationRepo(ctrl)
	service := &serviceImpl{
		resourcePoolRepo:   resourcePoolRepo,
		pauseOperationRepo: pauseOperationRepo,
	}
	pool := &entity.ResourcePool{ResourcePool: &globalngv1.ResourcePool{
		Id:             100,
		LifecycleState: globalngv1.ResourcePool_PAUSE_PRECHECKING,
	}}
	resourcePoolRepo.EXPECT().Load(gomock.Any(), uint64(100), false).Return(pool, nil)
	pauseOperationRepo.EXPECT().Get(gomock.Any(), uint64(100), "pause-1").Return(
		&entity.ResourcePoolPauseOperation{
			ResourcePoolID: 100,
			OperationID:    "pause-1",
			Type:           entity.ResourcePoolPauseOperationTypePause,
		},
		nil,
	)

	response, berr := service.PauseResourcePool(t.Context(), &globalngv1.PauseResourcePoolRequest{
		ResourcePoolId:                100,
		OperationId:                   "pause-1",
		CreateManualBackupBeforePause: true,
	})

	require.NoError(t, berr)
	require.Equal(t, "pause-1", response.GetOperationId())
}

func TestPauseResourcePoolValidateOnlyReturnsRegionalBlockers(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	resourcePoolRepo := mockresourcepool.NewMockRepo(ctrl)
	clusterRepo := mockcluster.NewMockRepo(ctrl)
	regionRepo := mockregion.NewMockRepo(ctrl)
	regionalClient := mockregionalserver.NewMockClient(ctrl)
	service := &serviceImpl{
		resourcePoolRepo: resourcePoolRepo,
		clusterRepo:      clusterRepo,
		regionRepo:       regionRepo,
		regionalCli:      regionalClient,
	}
	resourcePool := &entity.ResourcePool{
		ResourcePool: &globalngv1.ResourcePool{
			Id:             100,
			LifecycleState: globalngv1.ResourcePool_ACTIVE,
		},
		RegionId: 10,
	}
	region := &entity.Region{}
	resourcePoolRepo.EXPECT().Load(gomock.Any(), uint64(100), false).Return(resourcePool, nil)
	clusterRepo.EXPECT().Query(gomock.Any(), &clusterrepo.Filter{SharedPoolIds: []uint64{100}}).Return(nil, nil)
	regionRepo.EXPECT().Get(gomock.Any(), uint64(10)).Return(region, nil)
	regionalClient.EXPECT().ValidateSharedPoolPause(
		gomock.Any(),
		region,
		&clusterngv1beta.ValidateSharedPoolPauseRequest{SharedPoolId: "100"},
	).Return(&clusterngv1beta.ValidateSharedPoolPauseResponse{
		Failures: []*clusterngv1beta.SharedPoolPauseValidationFailure{
			{Code: "RESOURCE_MUTATION_IN_PROGRESS", Message: "resource mutation is running"},
			{Code: "TIBR_UPGRADE_IN_PROGRESS", Message: "TiBR upgrade is running"},
		},
	}, nil)

	response, berr := service.PauseResourcePool(t.Context(), &globalngv1.PauseResourcePoolRequest{
		ResourcePoolId: 100,
		ValidateOnly:   true,
	})

	require.NoError(t, berr)
	require.Equal(t, []*globalngv1.ResourcePoolPauseValidationResult{
		{CheckCode: "RESOURCE_MUTATION_IN_PROGRESS", Detail: "resource mutation is running"},
		{CheckCode: "TIBR_UPGRADE_IN_PROGRESS", Detail: "TiBR upgrade is running"},
	}, response.GetValidationResults())
}

func TestPauseResourcePoolRevalidatesBeforeExecution(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	resourcePoolRepo := mockresourcepool.NewMockRepo(ctrl)
	clusterRepo := mockcluster.NewMockRepo(ctrl)
	regionRepo := mockregion.NewMockRepo(ctrl)
	regionalClient := mockregionalserver.NewMockClient(ctrl)
	service := &serviceImpl{
		resourcePoolRepo: resourcePoolRepo,
		clusterRepo:      clusterRepo,
		regionRepo:       regionRepo,
		regionalCli:      regionalClient,
	}
	resourcePool := &entity.ResourcePool{
		ResourcePool: &globalngv1.ResourcePool{
			Id:             100,
			LifecycleState: globalngv1.ResourcePool_ACTIVE,
		},
		RegionId: 10,
	}
	region := &entity.Region{}
	resourcePoolRepo.EXPECT().Load(gomock.Any(), uint64(100), false).Return(resourcePool, nil)
	regionRepo.EXPECT().Get(gomock.Any(), uint64(10)).Return(region, nil)
	clusterRepo.EXPECT().Query(gomock.Any(), &clusterrepo.Filter{SharedPoolIds: []uint64{100}}).Return(nil, nil)
	regionalClient.EXPECT().ValidateSharedPoolPause(
		gomock.Any(),
		region,
		&clusterngv1beta.ValidateSharedPoolPauseRequest{SharedPoolId: "100"},
	).Return(&clusterngv1beta.ValidateSharedPoolPauseResponse{
		Failures: []*clusterngv1beta.SharedPoolPauseValidationFailure{{
			Code:    "RESOURCE_MUTATION_IN_PROGRESS",
			Message: "resource mutation is running",
		}},
	}, nil)

	response, berr := service.PauseResourcePool(t.Context(), &globalngv1.PauseResourcePoolRequest{
		ResourcePoolId: 100,
		OperationId:    "pause-1",
	})

	require.NoError(t, berr)
	require.Empty(t, response.GetOperationId())
	require.Equal(t, []*globalngv1.ResourcePoolPauseValidationResult{{
		CheckCode: "RESOURCE_MUTATION_IN_PROGRESS",
		Detail:    "resource mutation is running",
	}}, response.GetValidationResults())
	require.Equal(t, globalngv1.ResourcePool_ACTIVE, resourcePool.GetLifecycleState())
}

func TestPauseResourcePoolRechecksGlobalBlockersAfterAcquiringLock(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	resourcePoolRepo := mockresourcepool.NewMockRepo(ctrl)
	clusterRepo := mockcluster.NewMockRepo(ctrl)
	exportRepo := mockexport.NewMockExportRepo(ctrl)
	regionRepo := mockregion.NewMockRepo(ctrl)
	regionalClient := mockregionalserver.NewMockClient(ctrl)
	flow := mockworkflow.NewMockWorkflow(ctrl)
	service := &serviceImpl{
		transaction:      resourcePoolPassthroughTransactioner{},
		resourcePoolRepo: resourcePoolRepo,
		clusterRepo:      clusterRepo,
		exportRepo:       exportRepo,
		regionRepo:       regionRepo,
		regionalCli:      regionalClient,
		workflow:         flow,
	}
	resourcePool := &entity.ResourcePool{
		ResourcePool: &globalngv1.ResourcePool{
			Id:             100,
			OwnerId:        42,
			LifecycleState: globalngv1.ResourcePool_ACTIVE,
		},
		RegionId: 10,
	}
	clusters := []*entity.Cluster{{
		Cluster: &globalngv1.Cluster{
			ClusterId: 1,
			State:     globalngv1.Cluster_AVAILABLE,
		},
	}}
	region := &entity.Region{}

	gomock.InOrder(
		resourcePoolRepo.EXPECT().Load(gomock.Any(), uint64(100), false).Return(resourcePool, nil),
		clusterRepo.EXPECT().Query(
			gomock.Any(),
			&clusterrepo.Filter{SharedPoolIds: []uint64{100}},
		).Return(clusters, nil),
		exportRepo.EXPECT().Count(gomock.Any(), &exportrepo.QueryExportsFilter{
			ClusterIds: []string{"1"},
			States:     []schema.ExportState{schema.ExportStateRunning},
		}).Return(int64(0), nil),
		flow.EXPECT().ListRunningDagInstancesByBizIds(
			gomock.Any(),
			pauseBlockingGlobalDagIDs,
			[]string{"1"},
		).Return(nil, nil),
		regionRepo.EXPECT().Get(gomock.Any(), uint64(10)).Return(region, nil),
		regionalClient.EXPECT().ValidateSharedPoolPause(
			gomock.Any(),
			region,
			&clusterngv1beta.ValidateSharedPoolPauseRequest{SharedPoolId: "100"},
		).Return(&clusterngv1beta.ValidateSharedPoolPauseResponse{}, nil),
		resourcePoolRepo.EXPECT().LoadForUpdate(gomock.Any(), uint64(100), false).Return(resourcePool, nil),
		clusterRepo.EXPECT().Query(
			gomock.Any(),
			&clusterrepo.Filter{SharedPoolIds: []uint64{100}},
		).Return(clusters, nil),
		exportRepo.EXPECT().Count(gomock.Any(), &exportrepo.QueryExportsFilter{
			ClusterIds: []string{"1"},
			States:     []schema.ExportState{schema.ExportStateRunning},
		}).Return(int64(0), nil),
		flow.EXPECT().ListRunningDagInstancesByBizIds(
			gomock.Any(),
			pauseBlockingGlobalDagIDs,
			[]string{"1"},
		).Return([]*fastflowentity.DagInstance{{}}, nil),
	)

	response, berr := service.PauseResourcePool(t.Context(), &globalngv1.PauseResourcePoolRequest{
		ResourcePoolId: 100,
		OperationId:    "pause-1",
	})

	require.NoError(t, berr)
	require.Empty(t, response.GetOperationId())
	require.Equal(t, []*globalngv1.ResourcePoolPauseValidationResult{{
		CheckCode: operationInProgressCheckCode,
		Detail:    "resource pool has a running global operation",
	}}, response.GetValidationResults())
	require.Equal(t, globalngv1.ResourcePool_ACTIVE, resourcePool.GetLifecycleState())
}

func TestPauseResourcePoolReturnsLifecycleBlockerWhenStateChangesAfterPrecheck(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	resourcePoolRepo := mockresourcepool.NewMockRepo(ctrl)
	clusterRepo := mockcluster.NewMockRepo(ctrl)
	regionRepo := mockregion.NewMockRepo(ctrl)
	regionalClient := mockregionalserver.NewMockClient(ctrl)
	service := &serviceImpl{
		transaction:      resourcePoolPassthroughTransactioner{},
		resourcePoolRepo: resourcePoolRepo,
		clusterRepo:      clusterRepo,
		regionRepo:       regionRepo,
		regionalCli:      regionalClient,
	}
	resourcePool := &entity.ResourcePool{
		ResourcePool: &globalngv1.ResourcePool{
			Id:             100,
			LifecycleState: globalngv1.ResourcePool_ACTIVE,
		},
		RegionId: 10,
	}
	lockedResourcePool := &entity.ResourcePool{
		ResourcePool: &globalngv1.ResourcePool{
			Id:             100,
			LifecycleState: globalngv1.ResourcePool_PAUSING,
		},
		RegionId: 10,
	}
	region := &entity.Region{}

	gomock.InOrder(
		resourcePoolRepo.EXPECT().Load(gomock.Any(), uint64(100), false).Return(resourcePool, nil),
		clusterRepo.EXPECT().Query(
			gomock.Any(),
			&clusterrepo.Filter{SharedPoolIds: []uint64{100}},
		).Return(nil, nil),
		regionRepo.EXPECT().Get(gomock.Any(), uint64(10)).Return(region, nil),
		regionalClient.EXPECT().ValidateSharedPoolPause(
			gomock.Any(),
			region,
			&clusterngv1beta.ValidateSharedPoolPauseRequest{SharedPoolId: "100"},
		).Return(&clusterngv1beta.ValidateSharedPoolPauseResponse{}, nil),
		resourcePoolRepo.EXPECT().LoadForUpdate(gomock.Any(), uint64(100), false).Return(lockedResourcePool, nil),
	)

	response, berr := service.PauseResourcePool(t.Context(), &globalngv1.PauseResourcePoolRequest{
		ResourcePoolId: 100,
		OperationId:    "pause-1",
	})

	require.NoError(t, berr)
	require.Empty(t, response.GetOperationId())
	require.Equal(t, []*globalngv1.ResourcePoolPauseValidationResult{{
		CheckCode: invalidLifecycleStateCheckCode,
		Detail:    "resource pool must be ACTIVE to pause",
	}}, response.GetValidationResults())
	require.Equal(t, globalngv1.ResourcePool_PAUSING, lockedResourcePool.GetLifecycleState())
}

func TestPauseResourcePoolReturnsAllGlobalAndRegionalBlockers(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	resourcePoolRepo := mockresourcepool.NewMockRepo(ctrl)
	clusterRepo := mockcluster.NewMockRepo(ctrl)
	exportRepo := mockexport.NewMockExportRepo(ctrl)
	regionRepo := mockregion.NewMockRepo(ctrl)
	regionalClient := mockregionalserver.NewMockClient(ctrl)
	flow := mockworkflow.NewMockWorkflow(ctrl)
	service := &serviceImpl{
		resourcePoolRepo: resourcePoolRepo,
		clusterRepo:      clusterRepo,
		exportRepo:       exportRepo,
		regionRepo:       regionRepo,
		regionalCli:      regionalClient,
		workflow:         flow,
	}
	resourcePool := &entity.ResourcePool{
		ResourcePool: &globalngv1.ResourcePool{
			Id:             100,
			LifecycleState: globalngv1.ResourcePool_ACTIVE,
		},
		RegionId: 10,
	}
	clusters := []*entity.Cluster{
		{Cluster: &globalngv1.Cluster{ClusterId: 1, State: globalngv1.Cluster_AVAILABLE}},
		{Cluster: &globalngv1.Cluster{ClusterId: 2, State: globalngv1.Cluster_MODIFYING}},
	}
	region := &entity.Region{}

	resourcePoolRepo.EXPECT().Load(gomock.Any(), uint64(100), false).Return(resourcePool, nil)
	clusterRepo.EXPECT().Query(
		gomock.Any(),
		&clusterrepo.Filter{SharedPoolIds: []uint64{100}},
	).Return(clusters, nil)
	exportRepo.EXPECT().Count(gomock.Any(), &exportrepo.QueryExportsFilter{
		ClusterIds: []string{"1", "2"},
		States:     []schema.ExportState{schema.ExportStateRunning},
	}).Return(int64(1), nil)
	flow.EXPECT().ListRunningDagInstancesByBizIds(
		gomock.Any(),
		pauseBlockingGlobalDagIDs,
		[]string{"1", "2"},
	).Return([]*fastflowentity.DagInstance{{}}, nil)
	regionRepo.EXPECT().Get(gomock.Any(), uint64(10)).Return(region, nil)
	regionalClient.EXPECT().ValidateSharedPoolPause(
		gomock.Any(),
		region,
		&clusterngv1beta.ValidateSharedPoolPauseRequest{SharedPoolId: "100"},
	).Return(&clusterngv1beta.ValidateSharedPoolPauseResponse{
		Failures: []*clusterngv1beta.SharedPoolPauseValidationFailure{
			{Code: "TIBR_UPGRADE_IN_PROGRESS", Message: "TiBR upgrade is running"},
		},
	}, nil)

	response, berr := service.PauseResourcePool(t.Context(), &globalngv1.PauseResourcePoolRequest{
		ResourcePoolId: 100,
		ValidateOnly:   true,
	})

	require.NoError(t, berr)
	require.Equal(t, []*globalngv1.ResourcePoolPauseValidationResult{
		{CheckCode: importExportInProgressCheckCode, Detail: "resource pool has a running export"},
		{CheckCode: instanceStateConflictCheckCode, Detail: "all resource pool instances must be ACTIVE to pause"},
		{CheckCode: operationInProgressCheckCode, Detail: "resource pool has a running global operation"},
		{CheckCode: "TIBR_UPGRADE_IN_PROGRESS", Detail: "TiBR upgrade is running"},
	}, response.GetValidationResults())
}

func TestPauseResourcePoolConvertsValidationDependencyFailuresToBlockers(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	resourcePoolRepo := mockresourcepool.NewMockRepo(ctrl)
	clusterRepo := mockcluster.NewMockRepo(ctrl)
	regionRepo := mockregion.NewMockRepo(ctrl)
	service := &serviceImpl{
		resourcePoolRepo: resourcePoolRepo,
		clusterRepo:      clusterRepo,
		regionRepo:       regionRepo,
	}
	resourcePool := &entity.ResourcePool{
		ResourcePool: &globalngv1.ResourcePool{
			Id:             100,
			LifecycleState: globalngv1.ResourcePool_ACTIVE,
		},
		RegionId: 10,
	}

	resourcePoolRepo.EXPECT().Load(gomock.Any(), uint64(100), false).Return(resourcePool, nil)
	clusterRepo.EXPECT().Query(
		gomock.Any(),
		&clusterrepo.Filter{SharedPoolIds: []uint64{100}},
	).Return(nil, bizerr.New(bizerr.ErrCodeUnknown, "cluster database unavailable"))
	regionRepo.EXPECT().Get(gomock.Any(), uint64(10)).Return(
		nil,
		bizerr.New(bizerr.ErrCodeUnknown, "region database unavailable"),
	)

	response, berr := service.PauseResourcePool(t.Context(), &globalngv1.PauseResourcePoolRequest{
		ResourcePoolId: 100,
		ValidateOnly:   true,
	})

	require.NoError(t, berr)
	require.Equal(t, []*globalngv1.ResourcePoolPauseValidationResult{
		{CheckCode: dependencyUnavailableCheckCode, Detail: "failed to query resource pool instances"},
		{CheckCode: dependencyUnavailableCheckCode, Detail: "failed to validate regional resource pool state"},
	}, response.GetValidationResults())
}

func TestResumeResourcePoolRejectsHistoricalOperationIDOwnedByPause(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	resourcePoolRepo := mockresourcepool.NewMockRepo(ctrl)
	pauseOperationRepo := mockresourcepool.NewMockPauseOperationRepo(ctrl)
	service := &serviceImpl{
		resourcePoolRepo:   resourcePoolRepo,
		pauseOperationRepo: pauseOperationRepo,
	}
	resourcePool := &entity.ResourcePool{
		ResourcePool: &globalngv1.ResourcePool{
			Id:             100,
			LifecycleState: globalngv1.ResourcePool_PAUSED,
		},
		RegionId: 10,
	}
	resourcePoolRepo.EXPECT().Load(gomock.Any(), uint64(100), false).Return(resourcePool, nil)
	pauseOperationRepo.EXPECT().Get(gomock.Any(), uint64(100), "pause-1").Return(
		&entity.ResourcePoolPauseOperation{
			ResourcePoolID: 100,
			OperationID:    "pause-1",
			Type:           entity.ResourcePoolPauseOperationTypePause,
			Status:         entity.ResourcePoolPauseOperationSucceeded,
		}, nil,
	)

	response, berr := service.ResumeResourcePool(t.Context(), &globalngv1.ResumeResourcePoolRequest{
		ResourcePoolId: 100,
		OperationId:    "pause-1",
	})

	require.Nil(t, response)
	require.Error(t, berr)
	require.Equal(t, bizerr.ErrCode(codes.FailedPrecondition), berr.ErrCode())
	require.ErrorContains(t, berr, "already owned by PAUSE")
	require.Equal(t, globalngv1.ResourcePool_PAUSED, resourcePool.GetLifecycleState())
}

func TestResumeResourcePoolCreatesGlobalOperationAtomically(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	resourcePoolRepo := mockresourcepool.NewMockRepo(ctrl)
	pauseOperationRepo := mockresourcepool.NewMockPauseOperationRepo(ctrl)
	flow := mockworkflow.NewMockWorkflow(ctrl)
	eventCommitter := mockeventbus.NewMockWALCommitter(ctrl)
	service := &serviceImpl{
		transaction:        resourcePoolPassthroughTransactioner{},
		workflow:           flow,
		resourcePoolRepo:   resourcePoolRepo,
		pauseOperationRepo: pauseOperationRepo,
		eventCommitter:     eventCommitter,
	}
	pausedAt := timestamppb.Now()
	resourcePool := &entity.ResourcePool{
		ResourcePool: &globalngv1.ResourcePool{
			Id:                100,
			OwnerId:           42,
			LifecycleState:    globalngv1.ResourcePool_PAUSED,
			PausedAt:          pausedAt,
			AutomaticResumeAt: timestamppb.Now(),
		},
		RegionId: 10,
	}
	resourcePoolRepo.EXPECT().Load(gomock.Any(), uint64(100), false).Return(resourcePool, nil)
	pauseOperationRepo.EXPECT().Get(gomock.Any(), uint64(100), "resume-1").Return(
		nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "operation not found"),
	).Times(2)
	resourcePoolRepo.EXPECT().LoadForUpdate(gomock.Any(), uint64(100), false).Return(resourcePool, nil)
	pauseOperationRepo.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(
		func(_ context.Context, operation *entity.ResourcePoolPauseOperation) bizerr.BizError {
			require.Equal(t, entity.ResourcePoolPauseOperationTypeResume, operation.Type)
			require.Equal(t, entity.ResourcePoolPauseOperationRunning, operation.Status)
			return nil
		},
	)
	resourcePoolRepo.EXPECT().Save(gomock.Any(), resourcePool).Return(nil)
	expectResourcePoolServiceLifecycleEvent(t, eventCommitter, clusterevent.EventTypeResuming, "CLUSTER_STATUS_RESUMING")
	flow.EXPECT().RunDagWithBizId(
		gomock.Any(),
		common.ResumeResourcePool,
		"42",
		"0",
		"100",
		gomock.Any(),
	).Return(nil)

	response, berr := service.ResumeResourcePool(t.Context(), &globalngv1.ResumeResourcePoolRequest{
		ResourcePoolId: 100,
		OperationId:    "resume-1",
	})

	require.NoError(t, berr)
	require.Equal(t, "resume-1", response.GetOperationId())
	require.Equal(t, globalngv1.ResourcePool_RESUMING, resourcePool.GetLifecycleState())
	require.Equal(t, pausedAt, resourcePool.GetPausedAt())
	require.Nil(t, resourcePool.GetAutomaticResumeAt())
}

func TestResumeResourcePoolRetryReplaysGlobalOperation(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	resourcePoolRepo := mockresourcepool.NewMockRepo(ctrl)
	pauseOperationRepo := mockresourcepool.NewMockPauseOperationRepo(ctrl)
	service := &serviceImpl{
		transaction:        resourcePoolPassthroughTransactioner{},
		resourcePoolRepo:   resourcePoolRepo,
		pauseOperationRepo: pauseOperationRepo,
	}
	resourcePool := &entity.ResourcePool{
		ResourcePool: &globalngv1.ResourcePool{
			Id:             100,
			LifecycleState: globalngv1.ResourcePool_RESUMING,
		},
		RegionId: 10,
	}
	resourcePoolRepo.EXPECT().Load(gomock.Any(), uint64(100), false).Return(resourcePool, nil)
	pauseOperationRepo.EXPECT().Get(gomock.Any(), uint64(100), "resume-1").Return(
		&entity.ResourcePoolPauseOperation{
			ResourcePoolID: 100,
			OperationID:    "resume-1",
			Type:           entity.ResourcePoolPauseOperationTypeResume,
			Status:         entity.ResourcePoolPauseOperationRunning,
		}, nil,
	)

	response, berr := service.ResumeResourcePool(t.Context(), &globalngv1.ResumeResourcePoolRequest{
		ResourcePoolId: 100,
		OperationId:    "resume-1",
	})

	require.NoError(t, berr)
	require.Equal(t, "resume-1", response.GetOperationId())
}

func TestGetResourcePoolPauseResumeOperationReturnsGlobalOperation(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	pauseOperationRepo := mockresourcepool.NewMockPauseOperationRepo(ctrl)
	service := &serviceImpl{pauseOperationRepo: pauseOperationRepo}
	now := time.Now().UTC()
	pauseOperationRepo.EXPECT().Get(gomock.Any(), uint64(100), "operation-1").Return(
		&entity.ResourcePoolPauseOperation{
			ResourcePoolID: 100,
			OperationID:    "operation-1",
			Type:           entity.ResourcePoolPauseOperationTypeResume,
			Status:         entity.ResourcePoolPauseOperationSucceeded,
			ResultSummary:  "resume completed",
			CreatedAt:      now.Add(-time.Minute),
			UpdatedAt:      now,
			CompletedAt:    &now,
		}, nil,
	)

	operation, berr := service.GetResourcePoolPauseResumeOperation(t.Context(), &globalngv1.GetResourcePoolPauseResumeOperationRequest{
		ResourcePoolId: 100,
		OperationId:    "operation-1",
	})

	require.Nil(t, berr)
	require.Equal(t, "operation-1", operation.GetOperationId())
	require.Equal(t, uint64(100), operation.GetResourcePoolId())
	require.Equal(t, globalngv1.ResourcePoolPauseResumeOperation_OPERATION_RESUME, operation.GetType())
	require.Equal(t, globalngv1.ResourcePoolPauseResumeOperation_SUCCEEDED, operation.GetStatus())
	require.Equal(t, "resume completed", operation.GetResultSummary())
	require.Equal(t, timestamppb.New(now), operation.GetCompletedAt())
}

func TestGetResourcePoolPauseResumeOperationKeepsHistoricalSuccessTerminal(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	pauseOperationRepo := mockresourcepool.NewMockPauseOperationRepo(ctrl)
	service := &serviceImpl{pauseOperationRepo: pauseOperationRepo}
	completedAt := time.Now().UTC()
	pauseOperationRepo.EXPECT().Get(gomock.Any(), uint64(100), "pause-1").Return(
		&entity.ResourcePoolPauseOperation{
			ResourcePoolID: 100,
			OperationID:    "pause-1",
			Type:           entity.ResourcePoolPauseOperationTypePause,
			Status:         entity.ResourcePoolPauseOperationSucceeded,
			CreatedAt:      completedAt.Add(-time.Minute),
			UpdatedAt:      completedAt,
			CompletedAt:    &completedAt,
		}, nil,
	)

	operation, berr := service.GetResourcePoolPauseResumeOperation(t.Context(),
		&globalngv1.GetResourcePoolPauseResumeOperationRequest{ResourcePoolId: 100, OperationId: "pause-1"})

	require.NoError(t, berr)
	require.Equal(t, globalngv1.ResourcePoolPauseResumeOperation_SUCCEEDED, operation.GetStatus())
	require.Equal(t, timestamppb.New(completedAt), operation.GetCompletedAt())
}

func TestGetResourcePoolPauseResumeOperationReturnsLocalPreparationFailure(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	pauseOperationRepo := mockresourcepool.NewMockPauseOperationRepo(ctrl)
	service := &serviceImpl{pauseOperationRepo: pauseOperationRepo}
	completedAt := time.Now().UTC()
	pauseOperationRepo.EXPECT().Get(gomock.Any(), uint64(100), "pause-1").Return(
		&entity.ResourcePoolPauseOperation{
			ResourcePoolID: 100,
			OperationID:    "pause-1",
			Status:         entity.ResourcePoolPauseOperationFailed,
			Stage:          entity.ResourcePoolPauseOperationPrecheck,
			ResultSummary:  "pause precheck failed: BUSY",
			CreatedAt:      completedAt.Add(-time.Minute),
			UpdatedAt:      completedAt,
			CompletedAt:    &completedAt,
		},
		nil,
	)

	operation, berr := service.GetResourcePoolPauseResumeOperation(t.Context(), &globalngv1.GetResourcePoolPauseResumeOperationRequest{
		ResourcePoolId: 100,
		OperationId:    "pause-1",
	})

	require.NoError(t, berr)
	require.Equal(t, globalngv1.ResourcePoolPauseResumeOperation_FAILED, operation.GetStatus())
	require.Equal(t, "pause precheck failed: BUSY", operation.GetResultSummary())
	require.NotNil(t, operation.GetCompletedAt())
}

func TestGetResourcePoolPauseResumeOperationReturnsGlobalRunningState(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	pauseOperationRepo := mockresourcepool.NewMockPauseOperationRepo(ctrl)
	service := &serviceImpl{pauseOperationRepo: pauseOperationRepo}
	now := time.Now().UTC()
	local := &entity.ResourcePoolPauseOperation{
		ResourcePoolID: 100,
		OperationID:    "pause-1",
		Type:           entity.ResourcePoolPauseOperationTypePause,
		Status:         entity.ResourcePoolPauseOperationRunning,
		Stage:          entity.ResourcePoolPauseOperationRegional,
		CreatedAt:      now,
		UpdatedAt:      now,
	}
	pauseOperationRepo.EXPECT().Get(gomock.Any(), uint64(100), "pause-1").Return(local, nil)

	operation, berr := service.GetResourcePoolPauseResumeOperation(t.Context(), &globalngv1.GetResourcePoolPauseResumeOperationRequest{
		ResourcePoolId: 100,
		OperationId:    "pause-1",
	})

	require.NoError(t, berr)
	require.Equal(t, "pause-1", operation.GetOperationId())
	require.Equal(t, globalngv1.ResourcePoolPauseResumeOperation_RUNNING, operation.GetStatus())
}