package br

import (
	"context"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	clusterngv1beta "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"go.uber.org/mock/gomock"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	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"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/br"
	mockbrrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/br/mock_br"
	mockclusterrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/cluster/mock_cluster"
	mockencryptionkeyrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/encryptionkey/mock_encryptionkey"
	mockregionrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/region/mock_region"
	resourcepoolrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/resourcepool"
	mockresourcepoolrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/resourcepool/mock_resourcepool"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/infra/repoimpl/schema"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	grpcmetadata "github.com/tidbcloud/tidb-management-service/utilities/grpc/metadata"
	"github.com/tidbcloud/tidb-management-service/utilities/pagination"
	mockworkflow "github.com/tidbcloud/tidb-management-service/utilities/workflow/mock_workflow"
)

func TestCreateManualBackup(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	ctx := grpcmetadata.AppendToOutgoingMeta(t.Context(), &grpcmetadata.RequestMeta{OrgID: 456})

	t.Run("success - create manual backup", func(t *testing.T) {
		mockBackupRepo := mockbrrepo.NewMockBackupRepo(ctrl)
		mockClusterRepo := mockclusterrepo.NewMockRepo(ctrl)
		mockEncryptionKeyRepo := mockencryptionkeyrepo.NewMockRepo(ctrl)
		mockRegionRepo := mockregionrepo.NewMockRepo(ctrl)
		mockResourcePoolRepo := mockresourcepoolrepo.NewMockRepo(ctrl)
		mockWorkflow := mockworkflow.NewMockWorkflow(ctrl)
		mockTransaction := newMockTransactioner()

		service := &serviceImpl{
			backupRepo:               mockBackupRepo,
			clusterRepo:              mockClusterRepo,
			regionRepo:               mockRegionRepo,
			resourcePoolRepo:         mockResourcePoolRepo,
			workflow:                 mockWorkflow,
			transaction:              mockTransaction,
			backupEncryptionResolver: NewResolver(mockEncryptionKeyRepo),
		}

		req := &globalngv1.CreateManualBackupRequest{
			ClusterId:      "123",
			OrganizationId: "456",
			Name:           "test-backup",
		}

		cluster := &entity.Cluster{
			Cluster: &globalngv1.Cluster{
				ClusterId:       123,
				OrganizationId:  456,
				SharedPoolId:    "789",
				RegionId:        globalcommonv1.Region_AWS_US_EAST_1,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				ServicePlan:     globalcommonv1.ServicePlan_Premium,
				Version:         "v1.0.0",
			},
			EncryptionKeyId: 987,
			RegionId:        1,
			CloudProviderId: 1,
		}

		region := &entity.Region{
			RegionID: 1,
		}

		mockClusterRepo.EXPECT().
			Get(gomock.Any(), uint64(123), false).
			Return(cluster, nil)

		mockRegionRepo.EXPECT().
			Get(gomock.Any(), uint64(1)).
			Return(region, nil)
		expectBRMutationAllowed(mockResourcePoolRepo, uint64(789))

		mockBackupRepo.EXPECT().
			Create(gomock.Any(), gomock.Any()).
			DoAndReturn(func(ctx context.Context, backup *entity.Backup) bizerr.BizError {
				assert.Equal(t, schema.BackupStatePending, backup.State)
				assert.Equal(t, schema.BackupTriggerTypeManual, backup.TriggerType)
				assert.Equal(t, schema.BackupTypeSnapshot, backup.Type)
				assert.Equal(t, "test-backup", backup.DisplayName)
				assert.Equal(t, uint64(123), backup.ClusterId)
				assert.Equal(t, uint64(789), backup.SharedPoolId)
				assert.Equal(t, uint64(456), backup.TenantId)
				assert.Equal(t, "AWS_US_EAST_1", backup.RegionId)
				assert.Equal(t, "AWS", backup.CloudProvider)
				assert.True(t, backup.CustomerManaged)
				assert.Equal(t, uint64(987), backup.EncryptionKeyId)
				backup.ID = 1000
				return nil
			})

		mockEncryptionKeyRepo.EXPECT().
			ListEncryptionKeys(gomock.Any(), gomock.Any()).
			Return([]*entity.EncryptionKey{
				{
					Id:              987,
					CustomerManaged: true,
				},
			}, nil)

		mockWorkflow.EXPECT().
			RunDagWithBizId(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
			DoAndReturn(func(ctx context.Context, dagId, orgId, projectId, clusterId string, vars map[string]string) error {
				assert.Equal(t, "create_manual_backup", dagId)
				assert.Equal(t, "456", orgId)
				assert.Equal(t, "0", projectId)
				assert.Equal(t, "123", clusterId)
				return nil
			})

		resp, err := service.CreateManualBackup(ctx, req)
		require.NoError(t, err)
		assert.NotNil(t, resp)
		assert.Equal(t, "1000", resp.BackupId)
	})

	t.Run("error - resource pool is paused", func(t *testing.T) {
		mockClusterRepo := mockclusterrepo.NewMockRepo(ctrl)
		mockRegionRepo := mockregionrepo.NewMockRepo(ctrl)
		mockResourcePoolRepo := mockresourcepoolrepo.NewMockRepo(ctrl)
		service := &serviceImpl{
			clusterRepo:      mockClusterRepo,
			regionRepo:       mockRegionRepo,
			resourcePoolRepo: mockResourcePoolRepo,
			transaction:      newMockTransactioner(),
		}
		cluster := &entity.Cluster{
			Cluster: &globalngv1.Cluster{
				ClusterId:      123,
				OrganizationId: 456,
				SharedPoolId:   "789",
			},
			RegionId: 1,
		}
		mockClusterRepo.EXPECT().Get(gomock.Any(), uint64(123), false).Return(cluster, nil)
		mockRegionRepo.EXPECT().Get(gomock.Any(), uint64(1)).Return(&entity.Region{RegionID: 1}, nil)
		mockResourcePoolRepo.EXPECT().Load(gomock.Any(), uint64(789), false).Return(&entity.ResourcePool{
			ResourcePool: &globalngv1.ResourcePool{Id: 789, LifecycleState: globalngv1.ResourcePool_PAUSED},
		}, nil)

		resp, err := service.CreateManualBackup(ctx, &globalngv1.CreateManualBackupRequest{
			ClusterId:      "123",
			OrganizationId: "456",
			Name:           "blocked-backup",
		})

		require.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), globalngv1.ResourcePool_PAUSED.String())
	})

	t.Run("error - invalid cluster id", func(t *testing.T) {
		service := &serviceImpl{}

		req := &globalngv1.CreateManualBackupRequest{
			ClusterId: "invalid",
		}

		resp, err := service.CreateManualBackup(ctx, req)
		require.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "failed to parse cluster id")
	})

	t.Run("error - cluster not found", func(t *testing.T) {
		mockClusterRepo := mockclusterrepo.NewMockRepo(ctrl)

		service := &serviceImpl{
			clusterRepo: mockClusterRepo,
		}

		req := &globalngv1.CreateManualBackupRequest{
			ClusterId:      "123",
			OrganizationId: "456",
			Name:           "test-backup",
		}

		mockClusterRepo.EXPECT().
			Get(gomock.Any(), uint64(123), false).
			Return(nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "cluster not found"))

		resp, err := service.CreateManualBackup(ctx, req)
		require.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "failed to get cluster")
	})

	t.Run("error - region not found", func(t *testing.T) {
		mockClusterRepo := mockclusterrepo.NewMockRepo(ctrl)
		mockRegionRepo := mockregionrepo.NewMockRepo(ctrl)

		service := &serviceImpl{
			clusterRepo: mockClusterRepo,
			regionRepo:  mockRegionRepo,
		}

		req := &globalngv1.CreateManualBackupRequest{
			ClusterId:      "123",
			OrganizationId: "456",
			Name:           "test-backup",
		}

		cluster := &entity.Cluster{
			Cluster: &globalngv1.Cluster{
				ClusterId:      123,
				OrganizationId: 456,
				SharedPoolId:   "789",
			},
			RegionId: 1,
		}

		mockClusterRepo.EXPECT().
			Get(gomock.Any(), uint64(123), false).
			Return(cluster, nil)

		mockRegionRepo.EXPECT().
			Get(gomock.Any(), uint64(1)).
			Return(nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "region not found"))

		resp, err := service.CreateManualBackup(ctx, req)
		require.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "failed to get region")
	})

	t.Run("error - invalid shared pool id", func(t *testing.T) {
		mockClusterRepo := mockclusterrepo.NewMockRepo(ctrl)
		mockRegionRepo := mockregionrepo.NewMockRepo(ctrl)

		service := &serviceImpl{
			clusterRepo: mockClusterRepo,
			regionRepo:  mockRegionRepo,
		}

		req := &globalngv1.CreateManualBackupRequest{
			ClusterId:      "123",
			OrganizationId: "456",
			Name:           "test-backup",
		}

		cluster := &entity.Cluster{
			Cluster: &globalngv1.Cluster{
				ClusterId:      123,
				OrganizationId: 456,
				SharedPoolId:   "invalid",
			},
			RegionId: 1,
		}

		region := &entity.Region{
			RegionID: 1,
		}

		mockClusterRepo.EXPECT().
			Get(gomock.Any(), uint64(123), false).
			Return(cluster, nil)

		mockRegionRepo.EXPECT().
			Get(gomock.Any(), uint64(1)).
			Return(region, nil)

		resp, err := service.CreateManualBackup(ctx, req)
		require.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "failed to parse shared pool id")
	})

	t.Run("error - backup create fails", func(t *testing.T) {
		mockBackupRepo := mockbrrepo.NewMockBackupRepo(ctrl)
		mockClusterRepo := mockclusterrepo.NewMockRepo(ctrl)
		mockRegionRepo := mockregionrepo.NewMockRepo(ctrl)
		mockResourcePoolRepo := mockresourcepoolrepo.NewMockRepo(ctrl)
		mockTransaction := newMockTransactioner()

		service := &serviceImpl{
			backupRepo:       mockBackupRepo,
			clusterRepo:      mockClusterRepo,
			regionRepo:       mockRegionRepo,
			resourcePoolRepo: mockResourcePoolRepo,
			transaction:      mockTransaction,
		}

		req := &globalngv1.CreateManualBackupRequest{
			ClusterId:      "123",
			OrganizationId: "456",
			Name:           "test-backup",
		}

		cluster := &entity.Cluster{
			Cluster: &globalngv1.Cluster{
				ClusterId:      123,
				OrganizationId: 456,
				SharedPoolId:   "789",
			},
			RegionId: 1,
		}

		region := &entity.Region{
			RegionID: 1,
		}

		mockClusterRepo.EXPECT().
			Get(gomock.Any(), uint64(123), false).
			Return(cluster, nil)

		mockRegionRepo.EXPECT().
			Get(gomock.Any(), uint64(1)).
			Return(region, nil)
		expectBRMutationAllowed(mockResourcePoolRepo, uint64(789))

		mockBackupRepo.EXPECT().
			Create(gomock.Any(), gomock.Any()).
			Return(bizerr.New(bizerr.ErrCodeDB, "create failed"))

		resp, err := service.CreateManualBackup(ctx, req)
		require.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "failed to create backup record")
	})

	t.Run("error - workflow run fails", func(t *testing.T) {
		mockBackupRepo := mockbrrepo.NewMockBackupRepo(ctrl)
		mockClusterRepo := mockclusterrepo.NewMockRepo(ctrl)
		mockRegionRepo := mockregionrepo.NewMockRepo(ctrl)
		mockResourcePoolRepo := mockresourcepoolrepo.NewMockRepo(ctrl)
		mockWorkflow := mockworkflow.NewMockWorkflow(ctrl)
		mockTransaction := newMockTransactioner()

		service := &serviceImpl{
			backupRepo:       mockBackupRepo,
			clusterRepo:      mockClusterRepo,
			regionRepo:       mockRegionRepo,
			resourcePoolRepo: mockResourcePoolRepo,
			workflow:         mockWorkflow,
			transaction:      mockTransaction,
		}

		req := &globalngv1.CreateManualBackupRequest{
			ClusterId:      "123",
			OrganizationId: "456",
			Name:           "test-backup",
		}

		cluster := &entity.Cluster{
			Cluster: &globalngv1.Cluster{
				ClusterId:      123,
				OrganizationId: 456,
				SharedPoolId:   "789",
			},
			RegionId: 1,
		}

		region := &entity.Region{
			RegionID: 1,
		}

		mockClusterRepo.EXPECT().
			Get(gomock.Any(), uint64(123), false).
			Return(cluster, nil)

		mockRegionRepo.EXPECT().
			Get(gomock.Any(), uint64(1)).
			Return(region, nil)
		expectBRMutationAllowed(mockResourcePoolRepo, uint64(789))

		mockBackupRepo.EXPECT().
			Create(gomock.Any(), gomock.Any()).
			DoAndReturn(func(ctx context.Context, backup *entity.Backup) bizerr.BizError {
				backup.ID = 1000
				return nil
			})

		mockWorkflow.EXPECT().
			RunDagWithBizId(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
			Return(assert.AnError)

		resp, err := service.CreateManualBackup(ctx, req)
		require.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "create async workflow failed")
	})
}

func expectBRMutationAllowed(resourcePoolRepo *mockresourcepoolrepo.MockRepo, sharedPoolID uint64) {
	pool := &entity.ResourcePool{ResourcePool: &globalngv1.ResourcePool{
		Id:             sharedPoolID,
		LifecycleState: globalngv1.ResourcePool_ACTIVE,
	}}
	resourcePoolRepo.EXPECT().
		Load(gomock.Any(), sharedPoolID, false).
		Return(pool, nil)
	resourcePoolRepo.EXPECT().
		LoadLifecycleForUpdate(gomock.Any(), sharedPoolID, false).
		Return(&resourcepoolrepo.Lifecycle{State: globalngv1.ResourcePool_ACTIVE}, nil)
}

func TestDeleteBackupWithManualBackup(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	ctx := t.Context()

	t.Run("success - delete auto backup", func(t *testing.T) {
		mockBackupRepo := mockbrrepo.NewMockBackupRepo(ctrl)
		mockBackupSettingRepo := mockbrrepo.NewMockBackupSettingRepo(ctrl)
		mockTransaction := newMockTransactioner()

		service := &serviceImpl{
			backupRepo:        mockBackupRepo,
			backupSettingRepo: mockBackupSettingRepo,
			transaction:       mockTransaction,
		}

		backup := &globalngv1.Backup{
			Id:             "100",
			ClusterId:      "123",
			OrganizationId: "456",
		}

		mockBackupRepo.EXPECT().
			UnscopedQuery(gomock.Any(), gomock.Any(), gomock.Any()).
			DoAndReturn(func(ctx context.Context, filter *br.QueryBackupsFilter, page any) ([]*entity.Backup, bizerr.BizError) {
				return []*entity.Backup{{
					Backup: &schema.Backup{
						ID:          100,
						ClusterId:   123,
						State:       schema.BackupStateSuccess,
						TriggerType: schema.BackupTriggerTypeAuto,
					},
				}}, nil
			})

		mockBackupRepo.EXPECT().
			UnscopedQuery(gomock.Any(), gomock.Any(), gomock.Any()).
			DoAndReturn(func(ctx context.Context, filter *br.QueryBackupsFilter, page *pagination.Page) ([]*entity.Backup, bizerr.BizError) {
				assert.Equal(t, []schema.BackupState{schema.BackupStateSuccess}, filter.States)
				assert.Equal(t, []uint64{123}, filter.ClusterIds)
				assert.Equal(t, []schema.BackupTriggerType{schema.BackupTriggerTypeAuto}, filter.TriggerTypes)
				assert.NotNil(t, filter.Deleted)
				assert.False(t, *filter.Deleted)
				assert.NotNil(t, page)
				assert.Equal(t, int64(1), page.PageSize)
				return []*entity.Backup{{
					Backup: &schema.Backup{
						ID:          101,
						ClusterId:   123,
						State:       schema.BackupStateSuccess,
						TriggerType: schema.BackupTriggerTypeAuto,
					},
				}}, nil
			})

		mockBackupRepo.EXPECT().
			Delete(gomock.Any(), gomock.Any()).
			Return(nil)

		mockBackupSettingRepo.EXPECT().
			TransitStateAfterBackupDeletedIfNeeded(gomock.Any(), uint64(123)).
			Return(nil)

		err := service.DeleteBackup(ctx, backup)
		require.NoError(t, err)
	})

	t.Run("success - delete manual backup", func(t *testing.T) {
		mockBackupRepo := mockbrrepo.NewMockBackupRepo(ctrl)
		mockBackupSettingRepo := mockbrrepo.NewMockBackupSettingRepo(ctrl)
		mockClusterRepo := mockclusterrepo.NewMockRepo(ctrl)
		mockRegionRepo := mockregionrepo.NewMockRepo(ctrl)
		mockResourcePoolRepo := mockresourcepoolrepo.NewMockRepo(ctrl)
		mockRegionalClient := mockregionalserver.NewMockClient(ctrl)
		mockTransaction := newMockTransactioner()
		region := &entity.Region{RegionID: 1}

		service := &serviceImpl{
			backupRepo:        mockBackupRepo,
			backupSettingRepo: mockBackupSettingRepo,
			clusterRepo:       mockClusterRepo,
			regionRepo:        mockRegionRepo,
			resourcePoolRepo:  mockResourcePoolRepo,
			regionalCli:       mockRegionalClient,
			transaction:       mockTransaction,
		}

		mockBackupRepo.EXPECT().
			UnscopedQuery(gomock.Any(), gomock.Any(), gomock.Any()).
			Return([]*entity.Backup{{Backup: &schema.Backup{
				ID:           100,
				ClusterId:    123,
				SharedPoolId: 789,
				TriggerType:  schema.BackupTriggerTypeManual,
			}}}, nil)
		mockClusterRepo.EXPECT().Get(gomock.Any(), uint64(123), true).
			Return(&entity.Cluster{RegionId: 1}, nil)
		mockRegionRepo.EXPECT().Get(gomock.Any(), uint64(1)).Return(region, nil)
		mockResourcePoolRepo.EXPECT().LoadLifecycleForUpdate(gomock.Any(), uint64(789), true).
			Return(&resourcepoolrepo.Lifecycle{State: globalngv1.ResourcePool_ACTIVE}, nil)
		mockRegionalClient.EXPECT().DeleteManualBackup(
			gomock.Any(),
			region,
			&clusterngv1beta.DeleteManualBackupRequest{ClusterId: "123", BackupId: "100"},
		).Return(nil)
		mockBackupRepo.EXPECT().Delete(gomock.Any(), gomock.Any()).Return(nil)
		mockBackupSettingRepo.EXPECT().TransitStateAfterBackupDeletedIfNeeded(gomock.Any(), uint64(123)).Return(nil)

		err := service.DeleteBackup(ctx, &globalngv1.Backup{
			Id:             "100",
			ClusterId:      "123",
			OrganizationId: "456",
		})
		require.NoError(t, err)
	})

	t.Run("error - latest successful auto backup is protected before expiration", func(t *testing.T) {
		mockBackupRepo := mockbrrepo.NewMockBackupRepo(ctrl)
		mockBackupSettingRepo := mockbrrepo.NewMockBackupSettingRepo(ctrl)
		mockTransaction := newMockTransactioner()

		service := &serviceImpl{
			backupRepo:        mockBackupRepo,
			backupSettingRepo: mockBackupSettingRepo,
			transaction:       mockTransaction,
		}

		backup := &globalngv1.Backup{
			Id:             "100",
			ClusterId:      "123",
			OrganizationId: "456",
		}

		mockBackupRepo.EXPECT().
			UnscopedQuery(gomock.Any(), gomock.Any(), gomock.Any()).
			DoAndReturn(func(ctx context.Context, filter *br.QueryBackupsFilter, page *pagination.Page) ([]*entity.Backup, bizerr.BizError) {
				assert.Equal(t, []uint64{100}, filter.Ids)
				assert.Nil(t, page)
				return []*entity.Backup{{
					Backup: &schema.Backup{
						ID:             100,
						ClusterId:      123,
						State:          schema.BackupStateSuccess,
						TriggerType:    schema.BackupTriggerTypeAuto,
						BackupTime:     time.Now().UTC(),
						ExpirationTime: time.Now().UTC().Add(time.Hour),
					},
				}}, nil
			})

		mockBackupRepo.EXPECT().
			UnscopedQuery(gomock.Any(), gomock.Any(), gomock.Any()).
			DoAndReturn(func(ctx context.Context, filter *br.QueryBackupsFilter, page *pagination.Page) ([]*entity.Backup, bizerr.BizError) {
				assert.Equal(t, []schema.BackupState{schema.BackupStateSuccess}, filter.States)
				assert.Equal(t, []uint64{123}, filter.ClusterIds)
				assert.Equal(t, []schema.BackupTriggerType{schema.BackupTriggerTypeAuto}, filter.TriggerTypes)
				assert.NotNil(t, filter.Deleted)
				assert.False(t, *filter.Deleted)
				assert.NotNil(t, page)
				assert.Equal(t, int64(1), page.PageSize)
				return []*entity.Backup{{
					Backup: &schema.Backup{
						ID:          100,
						ClusterId:   123,
						State:       schema.BackupStateSuccess,
						TriggerType: schema.BackupTriggerTypeAuto,
						BackupTime:  time.Now().UTC(),
					},
				}}, nil
			})

		err := service.DeleteBackup(ctx, backup)
		require.Error(t, err)
		assert.Contains(t, err.Error(), "Data Safety Policy")
	})

	t.Run("success - latest successful auto backup can be deleted after expiration", func(t *testing.T) {
		mockBackupRepo := mockbrrepo.NewMockBackupRepo(ctrl)
		mockBackupSettingRepo := mockbrrepo.NewMockBackupSettingRepo(ctrl)
		mockTransaction := newMockTransactioner()

		service := &serviceImpl{
			backupRepo:        mockBackupRepo,
			backupSettingRepo: mockBackupSettingRepo,
			transaction:       mockTransaction,
		}

		backup := &globalngv1.Backup{
			Id:             "100",
			ClusterId:      "123",
			OrganizationId: "456",
		}

		mockBackupRepo.EXPECT().
			UnscopedQuery(gomock.Any(), gomock.Any(), gomock.Any()).
			DoAndReturn(func(ctx context.Context, filter *br.QueryBackupsFilter, page *pagination.Page) ([]*entity.Backup, bizerr.BizError) {
				return []*entity.Backup{{
					Backup: &schema.Backup{
						ID:             100,
						ClusterId:      123,
						State:          schema.BackupStateSuccess,
						TriggerType:    schema.BackupTriggerTypeAuto,
						ExpirationTime: time.Now().UTC().Add(-time.Hour),
					},
				}}, nil
			})

		mockBackupRepo.EXPECT().
			Delete(gomock.Any(), gomock.Any()).
			Return(nil)

		mockBackupSettingRepo.EXPECT().
			TransitStateAfterBackupDeletedIfNeeded(gomock.Any(), uint64(123)).
			Return(nil)

		err := service.DeleteBackup(ctx, backup)
		require.NoError(t, err)
	})

	t.Run("error - manual backup regional delete fails", func(t *testing.T) {
		mockBackupRepo := mockbrrepo.NewMockBackupRepo(ctrl)
		mockBackupSettingRepo := mockbrrepo.NewMockBackupSettingRepo(ctrl)
		mockClusterRepo := mockclusterrepo.NewMockRepo(ctrl)
		mockRegionRepo := mockregionrepo.NewMockRepo(ctrl)
		mockTransaction := newMockTransactioner()

		service := &serviceImpl{
			backupRepo:        mockBackupRepo,
			backupSettingRepo: mockBackupSettingRepo,
			clusterRepo:       mockClusterRepo,
			regionRepo:        mockRegionRepo,
			transaction:       mockTransaction,
		}

		backup := &globalngv1.Backup{
			Id:             "100",
			ClusterId:      "123",
			OrganizationId: "456",
		}

		mockBackupRepo.EXPECT().
			UnscopedQuery(gomock.Any(), gomock.Any(), gomock.Any()).
			DoAndReturn(func(ctx context.Context, filter *br.QueryBackupsFilter, page any) ([]*entity.Backup, bizerr.BizError) {
				return []*entity.Backup{{
					Backup: &schema.Backup{
						ID:          100,
						ClusterId:   123,
						RegionId:    "AWS_US_EAST_1",
						TriggerType: schema.BackupTriggerTypeManual,
					},
				}}, nil
			})

		mockClusterRepo.EXPECT().
			Get(gomock.Any(), uint64(123), true).
			Return(&entity.Cluster{RegionId: 1}, nil)

		mockRegionRepo.EXPECT().
			Get(gomock.Any(), uint64(1)).
			Return(nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "region not found"))

		err := service.DeleteBackup(ctx, backup)
		require.Error(t, err)
		assert.Contains(t, err.Error(), "failed to get region")
	})

	t.Run("error - manual backup deletion is fenced during pause", func(t *testing.T) {
		mockBackupRepo := mockbrrepo.NewMockBackupRepo(ctrl)
		mockBackupSettingRepo := mockbrrepo.NewMockBackupSettingRepo(ctrl)
		mockClusterRepo := mockclusterrepo.NewMockRepo(ctrl)
		mockRegionRepo := mockregionrepo.NewMockRepo(ctrl)
		mockResourcePoolRepo := mockresourcepoolrepo.NewMockRepo(ctrl)
		mockTransaction := newMockTransactioner()

		service := &serviceImpl{
			backupRepo:        mockBackupRepo,
			backupSettingRepo: mockBackupSettingRepo,
			clusterRepo:       mockClusterRepo,
			regionRepo:        mockRegionRepo,
			resourcePoolRepo:  mockResourcePoolRepo,
			transaction:       mockTransaction,
		}

		mockBackupRepo.EXPECT().
			UnscopedQuery(gomock.Any(), gomock.Any(), gomock.Any()).
			Return([]*entity.Backup{{Backup: &schema.Backup{
				ID:           100,
				ClusterId:    123,
				SharedPoolId: 789,
				TriggerType:  schema.BackupTriggerTypeManual,
			}}}, nil)
		mockClusterRepo.EXPECT().
			Get(gomock.Any(), uint64(123), true).
			Return(&entity.Cluster{RegionId: 1}, nil)
		mockRegionRepo.EXPECT().
			Get(gomock.Any(), uint64(1)).
			Return(&entity.Region{RegionID: 1}, nil)
		mockResourcePoolRepo.EXPECT().
			LoadLifecycleForUpdate(gomock.Any(), uint64(789), true).
			Return(&resourcepoolrepo.Lifecycle{
				State: globalngv1.ResourcePool_PAUSED,
			}, nil)

		err := service.DeleteBackup(ctx, &globalngv1.Backup{
			Id:             "100",
			ClusterId:      "123",
			OrganizationId: "456",
		})
		require.Error(t, err)
		assert.Contains(t, err.Error(), "operation is not allowed")
	})
}