package service

import (
	"context"
	"strconv"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/pkgv2/bizerr"
	pkgv2 "github.com/tidbcloud/pkgv2/clusterproxy/serverless"
	"github.com/tidbcloud/pkgv2/db/mock_db"
	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/facade/mock_facade"
	"github.com/tidbcloud/serverless-service/internal/domain/repo/mock_repo"
	"github.com/tidbcloud/serverless-service/internal/domain/vo"
	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	regionalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/regional/serverless/v1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/mock/gomock"
	"google.golang.org/protobuf/types/known/timestamppb"
)

const (
	testTenantID      = "1001"
	testClusterID     = "2002"
	testProvider      = config.ProviderAWS
	testRegion        = "aws-us-east-1"
	testKubernetes    = "test-us-east-1-f01"
	testProxyURL      = "http://cluster-proxy.example"
	testSourceToken   = "source-token"
	testScheduleCron  = "0 0 0 * * *"
	testRetentionHour = 72
)

func TestListBackupsMergesLegacyAndVirtualBackups(t *testing.T) {
	ctrl := gomock.NewController(t)
	svc, backupRepo, historyRepo, tombstoneRepo := newBackupRestoreServiceForVirtualTest(ctrl)
	withoutCurrentSchedule(svc)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	svc.mgmtSvc = mgmtSvc
	scheduleTime := todayMidnightUTC()
	legacyBackupTime := scheduleTime.Add(-24 * time.Hour)
	legacyExpireTime := scheduleTime.Add(24 * time.Hour)

	backupRepo.EXPECT().
		List(gomock.Any(), gomock.Any()).
		Return([]*entity.UserBackup{{
			ID:         "bak-legacy",
			State:      constant.BackupStateSucceeded,
			ClusterID:  testClusterID,
			TenantID:   testTenantID,
			BackupTime: &legacyBackupTime,
			ExpireTime: &legacyExpireTime,
		}}, nil)
	historyRepo.EXPECT().
		List(gomock.Any(), gomock.Any()).
		Return([]*entity.BackupScheduleHistory{historyForSchedule(scheduleTime)}, nil)
	tombstoneRepo.EXPECT().
		List(gomock.Any(), entity.ListBackupTombstoneOptions{ClusterId: testClusterID}).
		Return(nil, nil)

	backups, total, err := svc.ListBackups(context.Background(), &vo.ListUserBackupsOpt{
		ClusterId: testClusterID,
		PageSize:  10,
		OrderBy:   "backup_time desc",
	})

	require.NoError(t, err)
	require.Equal(t, uint32(2), total)
	require.Len(t, backups, 2)
	require.Equal(t, generateVirtualBackupID(testClusterID, scheduleTime), backups[0].ID)
	require.Equal(t, "bak-legacy", backups[1].ID)
}

func TestListBackupsPrefersVirtualBackupWithPersistedSchedule(t *testing.T) {
	ctrl := gomock.NewController(t)
	svc, backupRepo, historyRepo, tombstoneRepo := newBackupRestoreServiceForVirtualTest(ctrl)
	withoutCurrentSchedule(svc)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	svc.mgmtSvc = mgmtSvc
	scheduleTime := todayMidnightUTC()
	backupTime := scheduleTime.Add(3 * time.Minute)
	expireTime := scheduleTime.Add(24 * time.Hour)
	withReadyPhysicalSlot(t, ctrl, svc, scheduleTime, backupTime)

	backupRepo.EXPECT().
		List(gomock.Any(), gomock.Any()).
		Return([]*entity.UserBackup{{
			ID:           "bak-persisted",
			State:        constant.BackupStateSucceeded,
			Type:         constant.BackupTypeAuto,
			ClusterID:    testClusterID,
			TenantID:     testTenantID,
			BackupTime:   &backupTime,
			ExpireTime:   &expireTime,
			ScheduleTime: &scheduleTime,
		}}, nil)
	historyRepo.EXPECT().
		List(gomock.Any(), gomock.Any()).
		Return([]*entity.BackupScheduleHistory{historyForSchedule(scheduleTime)}, nil)
	tombstoneRepo.EXPECT().
		List(gomock.Any(), entity.ListBackupTombstoneOptions{ClusterId: testClusterID}).
		Return(nil, nil)

	backups, total, err := svc.ListBackups(context.Background(), &vo.ListUserBackupsOpt{
		ClusterId: testClusterID,
		PageSize:  10,
	})

	require.NoError(t, err)
	require.Equal(t, uint32(1), total)
	require.Len(t, backups, 1)
	require.Equal(t, generateVirtualBackupID(testClusterID, scheduleTime), backups[0].ID)
}

func TestListBackupsKeepsPersistedBackupWhenVirtualPhysicalIsNil(t *testing.T) {
	ctrl := gomock.NewController(t)
	svc, backupRepo, historyRepo, tombstoneRepo := newBackupRestoreServiceForVirtualTest(ctrl)
	withoutCurrentSchedule(svc)
	scheduleTime := todayMidnightUTC()
	backupTime := scheduleTime.Add(3 * time.Minute)
	expireTime := scheduleTime.Add(24 * time.Hour)

	backupRepo.EXPECT().
		List(gomock.Any(), gomock.Any()).
		Return([]*entity.UserBackup{{
			ID:           "bak-persisted",
			State:        constant.BackupStateSucceeded,
			Type:         constant.BackupTypeAuto,
			ClusterID:    testClusterID,
			TenantID:     testTenantID,
			BackupTime:   &backupTime,
			ExpireTime:   &expireTime,
			ScheduleTime: &scheduleTime,
		}}, nil)
	historyRepo.EXPECT().
		List(gomock.Any(), gomock.Any()).
		Return([]*entity.BackupScheduleHistory{historyForSchedule(scheduleTime)}, nil)
	tombstoneRepo.EXPECT().
		List(gomock.Any(), entity.ListBackupTombstoneOptions{ClusterId: testClusterID}).
		Return(nil, nil)

	backups, total, err := svc.ListBackups(context.Background(), &vo.ListUserBackupsOpt{
		ClusterId: testClusterID,
		PageSize:  10,
	})

	require.NoError(t, err)
	require.Equal(t, uint32(1), total)
	require.Len(t, backups, 1)
	require.Equal(t, "bak-persisted", backups[0].ID)
}

func TestListBackupsKeepsPersistedBackupUntilVirtualSlotIsReady(t *testing.T) {
	ctrl := gomock.NewController(t)
	svc, backupRepo, historyRepo, tombstoneRepo := newBackupRestoreServiceForVirtualTest(ctrl)
	withoutCurrentSchedule(svc)
	slotRepo := mock_repo.NewMockPhysicalBackupSlotRepo(ctrl)
	svc.physicalBackupSlotRepo = slotRepo
	scheduleTime := todayMidnightUTC()
	backupTime := scheduleTime.Add(3 * time.Minute)
	expireTime := scheduleTime.Add(24 * time.Hour)

	backupRepo.EXPECT().
		List(gomock.Any(), gomock.Any()).
		Return([]*entity.UserBackup{{
			ID:           "bak-persisted",
			State:        constant.BackupStateSucceeded,
			Type:         constant.BackupTypeAuto,
			ClusterID:    testClusterID,
			TenantID:     testTenantID,
			BackupTime:   &backupTime,
			ExpireTime:   &expireTime,
			ScheduleTime: &scheduleTime,
		}}, nil)
	historyRepo.EXPECT().
		List(gomock.Any(), gomock.Any()).
		Return([]*entity.BackupScheduleHistory{historyForSchedule(scheduleTime)}, nil)
	tombstoneRepo.EXPECT().
		List(gomock.Any(), entity.ListBackupTombstoneOptions{ClusterId: testClusterID}).
		Return(nil, nil)
	slotRepo.EXPECT().List(gomock.Any(), gomock.Any()).Return(nil, nil)
	slotRepo.EXPECT().ListByBackupTime(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil)

	backups, total, err := svc.ListBackups(context.Background(), &vo.ListUserBackupsOpt{
		ClusterId: testClusterID,
		PageSize:  10,
	})

	require.NoError(t, err)
	require.Equal(t, uint32(1), total)
	require.Len(t, backups, 1)
	require.Equal(t, "bak-persisted", backups[0].ID)
}

func TestListBackupsDoesNotGenerateVirtualBackupBeforeLocalHistoryStart(t *testing.T) {
	ctrl := gomock.NewController(t)
	svc, backupRepo, historyRepo, tombstoneRepo := newBackupRestoreServiceForVirtualTest(ctrl)
	withoutCurrentSchedule(svc)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	svc.mgmtSvc = mgmtSvc
	scheduleTime := todayMidnightUTC()
	readyTime := scheduleTime.Add(12 * time.Hour)
	history := historyForSchedule(scheduleTime)
	history.ValidFrom = &readyTime

	backupRepo.EXPECT().List(gomock.Any(), gomock.Any()).Return(nil, nil)
	historyRepo.EXPECT().
		List(gomock.Any(), gomock.Any()).
		Return([]*entity.BackupScheduleHistory{history}, nil)
	tombstoneRepo.EXPECT().
		List(gomock.Any(), entity.ListBackupTombstoneOptions{ClusterId: testClusterID}).
		Return(nil, nil)

	backups, total, err := svc.ListBackups(context.Background(), &vo.ListUserBackupsOpt{
		ClusterId: testClusterID,
		PageSize:  10,
	})

	require.NoError(t, err)
	require.Equal(t, uint32(0), total)
	require.Empty(t, backups)
}

func TestListBackupsExcludesVirtualBackupTombstone(t *testing.T) {
	ctrl := gomock.NewController(t)
	svc, backupRepo, historyRepo, tombstoneRepo := newBackupRestoreServiceForVirtualTest(ctrl)
	withoutCurrentSchedule(svc)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	svc.mgmtSvc = mgmtSvc
	scheduleTime := todayMidnightUTC()
	backupID := generateVirtualBackupID(testClusterID, scheduleTime)

	backupRepo.EXPECT().List(gomock.Any(), gomock.Any()).Return(nil, nil)
	historyRepo.EXPECT().
		List(gomock.Any(), gomock.Any()).
		Return([]*entity.BackupScheduleHistory{historyForSchedule(scheduleTime)}, nil)
	tombstoneRepo.EXPECT().
		List(gomock.Any(), entity.ListBackupTombstoneOptions{ClusterId: testClusterID}).
		Return([]*entity.BackupTombstone{{
			ClusterID:    testClusterID,
			BackupID:     backupID,
			ScheduleTime: &scheduleTime,
		}}, nil)

	backups, total, err := svc.ListBackups(context.Background(), &vo.ListUserBackupsOpt{
		ClusterId: testClusterID,
		PageSize:  10,
	})

	require.NoError(t, err)
	require.Equal(t, uint32(0), total)
	require.Empty(t, backups)
}

func TestBuildVirtualBackupsDelaysRecentSchedules(t *testing.T) {
	ctrl := gomock.NewController(t)
	svc, _, _, tombstoneRepo := newBackupRestoreServiceForVirtualTest(ctrl)
	now := time.Date(2026, 7, 10, 10, 10, 0, 0, time.UTC)
	validFrom := now.Add(-10 * time.Minute)
	history := historyForSchedule(validFrom)
	history.ScheduleCron = "0 * * * * *"
	history.ValidFrom = &validFrom

	tombstoneRepo.EXPECT().
		List(gomock.Any(), entity.ListBackupTombstoneOptions{ClusterId: testClusterID}).
		Return(nil, nil)

	backups, err := svc.buildVirtualBackupsFromHistories(
		context.Background(),
		[]*entity.BackupScheduleHistory{history},
		now,
		time.Time{},
	)

	require.NoError(t, err)
	require.NotEmpty(t, backups)
	for _, backup := range backups {
		require.False(t, backup.ScheduleTime.After(now))
	}
	require.Equal(
		t,
		generateVirtualBackupID(testClusterID, now),
		backups[len(backups)-1].ID,
	)
}

func TestBuildVirtualBackupsSkipsUnresolvedKubernetes(t *testing.T) {
	ctrl := gomock.NewController(t)
	svc, _, _, tombstoneRepo := newBackupRestoreServiceForVirtualTest(ctrl)
	now := todayMidnightUTC()
	history := historyForSchedule(now)
	history.Kubernetes = constant.KubernetesUnknown

	tombstoneRepo.EXPECT().
		List(gomock.Any(), entity.ListBackupTombstoneOptions{ClusterId: testClusterID}).
		Return(nil, nil)

	backups, err := svc.buildVirtualBackupsFromHistories(
		context.Background(),
		[]*entity.BackupScheduleHistory{history},
		now,
		time.Time{},
	)

	require.NoError(t, err)
	require.Empty(t, backups)
}

func TestHydrateVirtualBackupsBatchesDuplicateSlotKeys(t *testing.T) {
	ctrl := gomock.NewController(t)
	svc, _, _, _ := newBackupRestoreServiceForVirtualTest(ctrl)
	slotRepo := mock_repo.NewMockPhysicalBackupSlotRepo(ctrl)
	svc.physicalBackupSlotRepo = slotRepo
	slotTime := todayMidnightUTC()
	slotRepo.EXPECT().List(gomock.Any(), gomock.Any()).DoAndReturn(
		func(_ context.Context, keys []entity.PhysicalBackupSlotKey) ([]*entity.PhysicalBackupSlot, bizerr.BizError) {
			require.Len(t, keys, 1)
			require.Equal(t, slotTime, keys[0].SlotTime)
			return []*entity.PhysicalBackupSlot{{
				CloudProvider: testProvider,
				Region:        testRegion,
				Kubernetes:    testKubernetes,
				SlotTime:      slotTime,
				BackupTime:    slotTime.Add(5 * time.Minute),
			}}, nil
		},
	)

	now := slotTime.Add(24*time.Hour + 2*time.Minute)
	backups, err := svc.hydrateVirtualBackups(context.Background(), []*VirtualBackup{
		{
			ClusterID:     "cluster-a",
			CloudProvider: testProvider,
			Region:        testRegion,
			Kubernetes:    testKubernetes,
			ScheduleTime:  slotTime,
			Retention:     24 * time.Hour,
		},
		{
			ClusterID:     "cluster-b",
			CloudProvider: testProvider,
			Region:        testRegion,
			Kubernetes:    testKubernetes,
			ScheduleTime:  slotTime,
			Retention:     24 * time.Hour,
		},
	}, now)
	require.NoError(t, err)
	require.Len(t, backups, 2)
	require.Same(t, backups[0].Physical, backups[1].Physical)
	require.Equal(t, slotTime.Add(24*time.Hour+5*time.Minute), backups[0].ExpireTime)
}

func TestBuildDelayedVirtualBackupsUsesPhysicalBackupTimeWithoutDelayLimit(t *testing.T) {
	ctrl := gomock.NewController(t)
	svc, _, _, _ := newBackupRestoreServiceForVirtualTest(ctrl)
	slotRepo := mock_repo.NewMockPhysicalBackupSlotRepo(ctrl)
	svc.physicalBackupSlotRepo = slotRepo
	now := time.Date(2026, 7, 10, 10, 10, 0, 0, time.UTC)
	scheduleTime := time.Date(2026, 7, 8, 0, 0, 0, 0, time.UTC)
	physicalBackupTime := now.Add(-time.Hour)
	history := historyForSchedule(scheduleTime)
	history.Retention = 24

	slotRepo.EXPECT().ListByBackupTime(
		gomock.Any(),
		[]entity.PhysicalBackupSlotScope{{
			CloudProvider: testProvider,
			Region:        testRegion,
			Kubernetes:    testKubernetes,
		}},
		gomock.Any(),
	).Return([]*entity.PhysicalBackupSlot{{
		CloudProvider: testProvider,
		Region:        testRegion,
		Kubernetes:    testKubernetes,
		SlotTime:      scheduleTime,
		BackupTime:    physicalBackupTime,
	}}, nil)

	backups, err := svc.buildDelayedVirtualBackupsFromHistories(
		context.Background(),
		[]*entity.BackupScheduleHistory{history},
		now,
		noCutover,
		nil,
		false,
	)

	require.NoError(t, err)
	require.Len(t, backups, 1)
	require.Equal(t, scheduleTime, backups[0].ScheduleTime)
	require.Equal(t, physicalBackupTime.Add(24*time.Hour), backups[0].ExpireTime)
}

func TestGetBackupRejectsVirtualBackupClusterMismatch(t *testing.T) {
	ctrl := gomock.NewController(t)
	svc, _, _, _ := newBackupRestoreServiceForVirtualTest(ctrl)
	backupID := generateVirtualBackupID(testClusterID, todayMidnightUTC())

	_, err := svc.GetBackup(context.Background(), testTenantID, "different-cluster", backupID)

	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeRecordNotFound, err.ErrCode())
}

func TestGetBackupResolvesVirtualBackupFromClusterProxy(t *testing.T) {
	ctrl := gomock.NewController(t)
	svc, _, historyRepo, tombstoneRepo := newBackupRestoreServiceForVirtualTest(ctrl)
	withoutCurrentSchedule(svc)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	proxyProvider := mock_facade.NewMockClusterProxyProviderFacade(ctrl)
	proxy := mock_facade.NewMockServerlessClusterProxyFacade(ctrl)
	svc.mgmtSvc = mgmtSvc
	svc.clusterProxyProvider = proxyProvider
	scheduleTime := todayMidnightUTC()
	backupID := generateVirtualBackupID(testClusterID, scheduleTime)
	actualBackupTime := scheduleTime.Add(30 * time.Second)

	withTestClusterProxyConfig(t)
	tombstoneRepo.EXPECT().Exists(gomock.Any(), testClusterID, scheduleTime).Return(false, nil)
	historyRepo.EXPECT().
		List(gomock.Any(), gomock.Any()).
		Return([]*entity.BackupScheduleHistory{historyForSchedule(scheduleTime)}, nil)
	proxyProvider.EXPECT().GetServerlessProxy(gomock.Any(), testProxyURL).Return(proxy, nil)
	proxy.EXPECT().
		ListBackups(gomock.Any(), gomock.Any()).
		Return(&pkgv2.ListBackupResponse{Items: []pkgv2.BackupItem{{
			ID:   888,
			Name: "real-backup",
			Time: pkgv2.JSONTime(actualBackupTime),
		}}}, nil)

	backup, err := svc.GetBackup(context.Background(), testTenantID, testClusterID, backupID)

	require.NoError(t, err)
	require.Equal(t, backupID, backup.ID)
	require.True(t, actualBackupTime.Equal(*backup.BackupTime))
}

func TestGetBackupExplainsWhenVirtualPhysicalBackupIsNotReady(t *testing.T) {
	ctrl := gomock.NewController(t)
	svc, _, historyRepo, tombstoneRepo := newBackupRestoreServiceForVirtualTest(ctrl)
	withoutCurrentSchedule(svc)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	proxyProvider := mock_facade.NewMockClusterProxyProviderFacade(ctrl)
	proxy := mock_facade.NewMockServerlessClusterProxyFacade(ctrl)
	svc.mgmtSvc = mgmtSvc
	svc.clusterProxyProvider = proxyProvider
	scheduleTime := todayMidnightUTC()
	backupID := generateVirtualBackupID(testClusterID, scheduleTime)

	withTestClusterProxyConfig(t)
	tombstoneRepo.EXPECT().Exists(gomock.Any(), testClusterID, scheduleTime).Return(false, nil)
	historyRepo.EXPECT().
		List(gomock.Any(), gomock.Any()).
		Return([]*entity.BackupScheduleHistory{historyForSchedule(scheduleTime)}, nil)
	proxyProvider.EXPECT().GetServerlessProxy(gomock.Any(), testProxyURL).Return(proxy, nil)
	proxy.EXPECT().
		ListBackups(gomock.Any(), gomock.Any()).
		Return(&pkgv2.ListBackupResponse{}, nil)

	_, err := svc.GetBackup(context.Background(), testTenantID, testClusterID, backupID)

	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeRecordNotFound, err.ErrCode())
	require.Contains(t, err.Error(), virtualBackupNotReadyMessage)
}

func TestGetBackupResolvesFirstPhysicalBackupAfterScheduleTime(t *testing.T) {
	ctrl := gomock.NewController(t)
	svc, _, historyRepo, tombstoneRepo := newBackupRestoreServiceForVirtualTest(ctrl)
	withoutCurrentSchedule(svc)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	proxyProvider := mock_facade.NewMockClusterProxyProviderFacade(ctrl)
	proxy := mock_facade.NewMockServerlessClusterProxyFacade(ctrl)
	svc.mgmtSvc = mgmtSvc
	svc.clusterProxyProvider = proxyProvider
	scheduleTime := todayMidnightUTC()
	backupID := generateVirtualBackupID(testClusterID, scheduleTime)

	withTestClusterProxyConfig(t)
	tombstoneRepo.EXPECT().Exists(gomock.Any(), testClusterID, scheduleTime).Return(false, nil)
	historyRepo.EXPECT().List(gomock.Any(), gomock.Any()).
		Return([]*entity.BackupScheduleHistory{historyForSchedule(scheduleTime)}, nil)
	proxyProvider.EXPECT().GetServerlessProxy(gomock.Any(), testProxyURL).Return(proxy, nil)
	proxy.EXPECT().ListBackups(gomock.Any(), gomock.Any()).Return(&pkgv2.ListBackupResponse{
		Items: []pkgv2.BackupItem{{
			ID:   888,
			Name: "next-generation-backup",
			Time: pkgv2.JSONTime(scheduleTime.Add(time.Minute + time.Second)),
		}},
	}, nil)

	backup, err := svc.GetBackup(context.Background(), testTenantID, testClusterID, backupID)

	require.NoError(t, err)
	require.Equal(t, scheduleTime.Add(time.Minute+time.Second), *backup.BackupTime)
}

func TestGetBackupRejectsVirtualBackupBeforeLocalHistoryStart(t *testing.T) {
	ctrl := gomock.NewController(t)
	svc, _, historyRepo, tombstoneRepo := newBackupRestoreServiceForVirtualTest(ctrl)
	withoutCurrentSchedule(svc)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	svc.mgmtSvc = mgmtSvc
	scheduleTime := todayMidnightUTC()
	readyTime := scheduleTime.Add(12 * time.Hour)
	history := historyForSchedule(scheduleTime)
	history.ValidFrom = &readyTime
	backupID := generateVirtualBackupID(testClusterID, scheduleTime)

	tombstoneRepo.EXPECT().Exists(gomock.Any(), testClusterID, scheduleTime).Return(false, nil)
	historyRepo.EXPECT().List(gomock.Any(), gomock.Any()).Return([]*entity.BackupScheduleHistory{history}, nil)

	_, err := svc.GetBackup(context.Background(), testTenantID, testClusterID, backupID)

	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeRecordNotFound, err.ErrCode())
}

func TestListRecycledInstanceBackupsIncludesVirtualBackups(t *testing.T) {
	ctrl := gomock.NewController(t)
	svc, backupRepo, historyRepo, tombstoneRepo := newBackupRestoreServiceForVirtualTest(ctrl)
	withoutCurrentSchedule(svc)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	svc.mgmtSvc = mgmtSvc
	scheduleTime := todayMidnightUTC()

	mgmtSvc.EXPECT().GetClusterIncludeDeleted(gomock.Any(), uint64(2002)).Return(deletedCluster(), nil)
	backupRepo.EXPECT().List(gomock.Any(), gomock.Any()).Return(nil, nil)
	historyRepo.EXPECT().
		List(gomock.Any(), gomock.Any()).
		Return([]*entity.BackupScheduleHistory{historyForSchedule(scheduleTime)}, nil)
	tombstoneRepo.EXPECT().
		List(gomock.Any(), entity.ListBackupTombstoneOptions{ClusterId: testClusterID}).
		Return(nil, nil)

	backups, total, err := svc.ListRecycledInstanceBackups(context.Background(), &vo.ListUserBackupsOpt{
		TenantId:  testTenantID,
		ClusterId: testClusterID,
		PageSize:  10,
	})

	require.NoError(t, err)
	require.Equal(t, uint32(1), total)
	require.Len(t, backups, 1)
	require.Equal(t, generateVirtualBackupID(testClusterID, scheduleTime), backups[0].ID)
}

func TestListRecycledInstanceBackupsPrefersVirtualBackupWithPersistedSchedule(t *testing.T) {
	ctrl := gomock.NewController(t)
	svc, backupRepo, historyRepo, tombstoneRepo := newBackupRestoreServiceForVirtualTest(ctrl)
	withoutCurrentSchedule(svc)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	svc.mgmtSvc = mgmtSvc
	scheduleTime := todayMidnightUTC()
	backupTime := scheduleTime.Add(3 * time.Minute)
	expireTime := scheduleTime.Add(24 * time.Hour)
	withReadyPhysicalSlot(t, ctrl, svc, scheduleTime, backupTime)

	mgmtSvc.EXPECT().GetClusterIncludeDeleted(gomock.Any(), uint64(2002)).Return(deletedCluster(), nil)
	backupRepo.EXPECT().List(gomock.Any(), gomock.Any()).Return([]*entity.UserBackup{{
		ID:           "bak-recycled-persisted",
		State:        constant.BackupStateSucceeded,
		Type:         constant.BackupTypeAuto,
		ClusterID:    testClusterID,
		TenantID:     testTenantID,
		BackupTime:   &backupTime,
		ExpireTime:   &expireTime,
		ScheduleTime: &scheduleTime,
	}}, nil)
	historyRepo.EXPECT().
		List(gomock.Any(), gomock.Any()).
		Return([]*entity.BackupScheduleHistory{historyForSchedule(scheduleTime)}, nil)
	tombstoneRepo.EXPECT().
		List(gomock.Any(), entity.ListBackupTombstoneOptions{ClusterId: testClusterID}).
		Return(nil, nil)

	backups, total, err := svc.ListRecycledInstanceBackups(context.Background(), &vo.ListUserBackupsOpt{
		TenantId:  testTenantID,
		ClusterId: testClusterID,
		PageSize:  10,
	})

	require.NoError(t, err)
	require.Equal(t, uint32(1), total)
	require.Len(t, backups, 1)
	require.Equal(t, generateVirtualBackupID(testClusterID, scheduleTime), backups[0].ID)
}

func TestListRecycledInstancesUsesVirtualBackups(t *testing.T) {
	ctrl := gomock.NewController(t)
	svc, backupRepo, historyRepo, tombstoneRepo := newBackupRestoreServiceForVirtualTest(ctrl)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	svc.mgmtSvc = mgmtSvc
	scheduleTime := todayMidnightUTC()

	backupRepo.EXPECT().
		ListDeletedClustersWithBackup(gomock.Any(), gomock.Any()).
		Return(nil, nil)
	historyRepo.EXPECT().
		List(gomock.Any(), gomock.Any()).
		Return([]*entity.BackupScheduleHistory{historyForSchedule(scheduleTime)}, nil)
	tombstoneRepo.EXPECT().
		List(gomock.Any(), entity.ListBackupTombstoneOptions{ClusterId: testClusterID}).
		Return(nil, nil)
	mgmtSvc.EXPECT().
		ListClusters(gomock.Any(), gomock.Any()).
		Return(&tidbmgmtservicev1.ListClustersResponse{Clusters: []*tidbmgmtservicev1.Cluster{deletedCluster()}}, nil)

	instances, total, err := svc.ListRecycledInstances(context.Background(), &vo.ListRecycledInstancesOpt{
		TenantId: testTenantID,
		PageSize: 10,
	})

	require.NoError(t, err)
	require.Equal(t, uint32(1), total)
	require.Len(t, instances, 1)
	require.Equal(t, testClusterID, instances[0].ClusterID)
	require.Equal(t, testRegion, instances[0].Region)
	require.Equal(t, testProvider, instances[0].Provider)
}

func TestListRecycledInstancesUsesSameScheduleBoundaryAsActiveCluster(t *testing.T) {
	ctrl := gomock.NewController(t)
	svc, backupRepo, historyRepo, tombstoneRepo := newBackupRestoreServiceForVirtualTest(ctrl)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	svc.mgmtSvc = mgmtSvc
	scheduleTime := todayMidnightUTC().Add(-24 * time.Hour)
	readyTime := scheduleTime.Add(time.Hour)
	readyUnix := readyTime.Unix()
	validTo := scheduleTime.Add(2 * time.Hour)
	history := historyForSchedule(scheduleTime)
	history.ValidTo = &validTo
	cluster := deletedCluster()
	cluster.GetServerlessCluster().CreationFinishedAt = &readyUnix

	backupRepo.EXPECT().
		ListDeletedClustersWithBackup(gomock.Any(), gomock.Any()).
		Return(nil, nil)
	historyRepo.EXPECT().
		List(gomock.Any(), gomock.Any()).
		Return([]*entity.BackupScheduleHistory{history}, nil)
	mgmtSvc.EXPECT().
		ListClusters(gomock.Any(), gomock.Any()).
		Return(&tidbmgmtservicev1.ListClustersResponse{Clusters: []*tidbmgmtservicev1.Cluster{cluster}}, nil)
	tombstoneRepo.EXPECT().
		List(gomock.Any(), entity.ListBackupTombstoneOptions{ClusterId: testClusterID}).
		Return(nil, nil)

	instances, total, err := svc.ListRecycledInstances(context.Background(), &vo.ListRecycledInstancesOpt{
		TenantId: testTenantID,
		PageSize: 10,
	})

	require.NoError(t, err)
	require.Equal(t, uint32(1), total)
	require.Len(t, instances, 1)
	require.Equal(t, testClusterID, instances[0].ClusterID)
}

func TestListRecycledInstancesKeepsPersistedOnlyInstance(t *testing.T) {
	ctrl := gomock.NewController(t)
	svc, backupRepo, historyRepo, _ := newBackupRestoreServiceForVirtualTest(ctrl)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	svc.mgmtSvc = mgmtSvc

	backupRepo.EXPECT().
		ListDeletedClustersWithBackup(gomock.Any(), gomock.Any()).
		Return([]*entity.RecycledInstance{{
			ClusterID: testClusterID,
			Region:    testRegion,
			Provider:  testProvider,
		}}, nil)
	historyRepo.EXPECT().List(gomock.Any(), gomock.Any()).Return(nil, nil)
	mgmtSvc.EXPECT().
		ListClusters(gomock.Any(), gomock.Any()).
		Return(&tidbmgmtservicev1.ListClustersResponse{Clusters: []*tidbmgmtservicev1.Cluster{deletedCluster()}}, nil)

	instances, total, err := svc.ListRecycledInstances(context.Background(), &vo.ListRecycledInstancesOpt{
		TenantId: testTenantID,
		PageSize: 10,
	})

	require.NoError(t, err)
	require.Equal(t, uint32(1), total)
	require.Len(t, instances, 1)
	require.Equal(t, testClusterID, instances[0].ClusterID)
}

func TestGetRecycledInstanceKeepsPersistedOnlyInstance(t *testing.T) {
	ctrl := gomock.NewController(t)
	svc, backupRepo, historyRepo, _ := newBackupRestoreServiceForVirtualTest(ctrl)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	svc.mgmtSvc = mgmtSvc

	backupRepo.EXPECT().
		ListDeletedClustersWithBackup(gomock.Any(), gomock.Any()).
		Return([]*entity.RecycledInstance{{ClusterID: testClusterID}}, nil)
	historyRepo.EXPECT().List(gomock.Any(), gomock.Any()).Return(nil, nil)
	mgmtSvc.EXPECT().
		GetClusterIncludeDeleted(gomock.Any(), uint64(2002)).
		Return(deletedCluster(), nil)

	instance, err := svc.GetRecycledInstance(context.Background(), testTenantID, testClusterID)

	require.NoError(t, err)
	require.Equal(t, testClusterID, instance.ClusterID)
}

func TestBuildRecycledInstancesLoadsTombstonesForAllClustersOnce(t *testing.T) {
	ctrl := gomock.NewController(t)
	svc, _, _, tombstoneRepo := newBackupRestoreServiceForVirtualTest(ctrl)
	slotRepo := mock_repo.NewMockPhysicalBackupSlotRepo(ctrl)
	svc.physicalBackupSlotRepo = slotRepo
	now := time.Date(2026, 7, 10, 10, 10, 0, 0, time.UTC)
	scheduleTime := now.Truncate(24 * time.Hour)
	firstHistory := historyForSchedule(scheduleTime)
	secondHistory := historyForSchedule(scheduleTime)
	secondHistory.ClusterID = "3003"

	tombstoneRepo.EXPECT().List(gomock.Any(), entity.ListBackupTombstoneOptions{
		ClustersIn: []string{testClusterID, secondHistory.ClusterID},
	}).Return(nil, nil).Times(1)
	slotRepo.EXPECT().ListByBackupTime(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil)
	slotRepo.EXPECT().List(gomock.Any(), gomock.Any()).DoAndReturn(
		func(_ context.Context, keys []entity.PhysicalBackupSlotKey) ([]*entity.PhysicalBackupSlot, bizerr.BizError) {
			require.Len(t, keys, 1)
			require.Equal(t, scheduleTime, keys[0].SlotTime)
			return []*entity.PhysicalBackupSlot{{
				CloudProvider: testProvider,
				Region:        testRegion,
				Kubernetes:    testKubernetes,
				SlotTime:      scheduleTime,
				BackupTime:    scheduleTime,
			}}, nil
		},
	).Times(1)

	instances, err := svc.buildRecycledInstancesFromHistories(
		context.Background(),
		[]*entity.BackupScheduleHistory{firstHistory, secondHistory},
		nil,
		now,
	)

	require.NoError(t, err)
	require.Len(t, instances, 2)
}

func TestBuildRecycledInstancesAdvancesMissingSlotsByBatchRound(t *testing.T) {
	ctrl := gomock.NewController(t)
	svc, _, _, tombstoneRepo := newBackupRestoreServiceForVirtualTest(ctrl)
	slotRepo := mock_repo.NewMockPhysicalBackupSlotRepo(ctrl)
	svc.physicalBackupSlotRepo = slotRepo
	now := time.Date(2026, 7, 10, 10, 10, 0, 0, time.UTC)
	latestScheduleTime := now.Truncate(24 * time.Hour)
	previousScheduleTime := latestScheduleTime.Add(-24 * time.Hour)
	history := historyForSchedule(previousScheduleTime)

	tombstoneRepo.EXPECT().List(gomock.Any(), entity.ListBackupTombstoneOptions{
		ClusterId: testClusterID,
	}).Return(nil, nil)
	slotRepo.EXPECT().ListByBackupTime(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil)
	firstRound := slotRepo.EXPECT().List(gomock.Any(), gomock.Any()).DoAndReturn(
		func(_ context.Context, keys []entity.PhysicalBackupSlotKey) ([]*entity.PhysicalBackupSlot, bizerr.BizError) {
			require.Equal(t, []entity.PhysicalBackupSlotKey{{
				CloudProvider: testProvider,
				Region:        testRegion,
				Kubernetes:    testKubernetes,
				SlotTime:      latestScheduleTime,
			}}, keys)
			return nil, nil
		},
	)
	slotRepo.EXPECT().List(gomock.Any(), gomock.Any()).After(firstRound).DoAndReturn(
		func(_ context.Context, keys []entity.PhysicalBackupSlotKey) ([]*entity.PhysicalBackupSlot, bizerr.BizError) {
			require.Equal(t, previousScheduleTime, keys[0].SlotTime)
			return []*entity.PhysicalBackupSlot{{
				CloudProvider: testProvider,
				Region:        testRegion,
				Kubernetes:    testKubernetes,
				SlotTime:      previousScheduleTime,
				BackupTime:    previousScheduleTime,
			}}, nil
		},
	)

	instances, err := svc.buildRecycledInstancesFromHistories(
		context.Background(),
		[]*entity.BackupScheduleHistory{history},
		nil,
		now,
	)

	require.NoError(t, err)
	require.Len(t, instances, 1)
}

func TestListTombstonesBatchesLargeClusterSets(t *testing.T) {
	ctrl := gomock.NewController(t)
	svc, _, _, tombstoneRepo := newBackupRestoreServiceForVirtualTest(ctrl)
	clusterIDs := make([]string, tombstoneClusterQueryBatchSize+1)
	for i := range clusterIDs {
		clusterIDs[i] = strconv.Itoa(i + 1)
	}

	tombstoneRepo.EXPECT().List(gomock.Any(), entity.ListBackupTombstoneOptions{
		ClustersIn: clusterIDs[:tombstoneClusterQueryBatchSize],
	}).Return([]*entity.BackupTombstone{{BackupID: "first-batch"}}, nil)
	tombstoneRepo.EXPECT().List(gomock.Any(), entity.ListBackupTombstoneOptions{
		ClustersIn: clusterIDs[tombstoneClusterQueryBatchSize:],
	}).Return([]*entity.BackupTombstone{{BackupID: "second-batch"}}, nil)

	tombstones, err := svc.listTombstones(context.Background(), clusterIDs)

	require.NoError(t, err)
	require.Equal(t, []string{"first-batch", "second-batch"}, []string{
		tombstones[0].BackupID,
		tombstones[1].BackupID,
	})
}

func TestAppendCurrentScheduleHistory(t *testing.T) {
	ctrl := gomock.NewController(t)
	scheduleRepo := mock_repo.NewMockBackupScheduleRepo(ctrl)
	historyRepo := mock_repo.NewMockBackupScheduleHistoryRepo(ctrl)
	createdAt := time.Date(2026, 7, 1, 0, 0, 0, 0, time.UTC)
	cutoverAt := createdAt.Add(24 * time.Hour)
	scheduleRepo.EXPECT().List(gomock.Any(), entity.ListBackupScheduleOptions{ClusterId: testClusterID}).
		Return([]*entity.BackupSchedule{{
			TenantID:      testTenantID,
			ClusterID:     testClusterID,
			Region:        testRegion,
			ScheduleCron:  testScheduleCron,
			Retention:     testRetentionHour,
			Kubernetes:    testKubernetes,
			CloudProvider: testProvider,
			SourceToken:   testSourceToken,
			CreatedAt:     &createdAt,
		}}, nil)
	historyRepo.EXPECT().GetVirtualizationCutover(gomock.Any()).Return(cutoverAt, nil)
	svc := &backupRestoreService{
		backupScheduleRepo: scheduleRepo,
		backupHistoryRepo:  historyRepo,
	}

	histories, err := svc.appendCurrentScheduleHistory(context.Background(), testClusterID, nil)

	require.NoError(t, err)
	require.Len(t, histories, 1)
	require.Equal(t, testClusterID, histories[0].ClusterID)
	require.True(t, cutoverAt.Equal(*histories[0].ValidFrom))
	require.Nil(t, histories[0].ValidTo)
}

func TestListBackupsUsesCurrentScheduleWithoutHistory(t *testing.T) {
	ctrl := gomock.NewController(t)
	backupRepo := mock_repo.NewMockUserBackupRepo(ctrl)
	scheduleRepo := mock_repo.NewMockBackupScheduleRepo(ctrl)
	historyRepo := mock_repo.NewMockBackupScheduleHistoryRepo(ctrl)
	tombstoneRepo := mock_repo.NewMockBackupTombstoneRepo(ctrl)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	scheduleTime := todayMidnightUTC()
	validFrom := scheduleTime.Add(-time.Hour)
	svc := &backupRestoreService{
		backupRepo:          backupRepo,
		backupScheduleRepo:  scheduleRepo,
		backupHistoryRepo:   historyRepo,
		backupTombstoneRepo: tombstoneRepo,
		mgmtSvc:             mgmtSvc,
	}

	backupRepo.EXPECT().List(gomock.Any(), gomock.Any()).Return(nil, nil)
	historyRepo.EXPECT().List(gomock.Any(), gomock.Any()).Return(nil, nil)
	historyRepo.EXPECT().GetVirtualizationCutover(gomock.Any()).Return(validFrom, nil)
	scheduleRepo.EXPECT().List(gomock.Any(), entity.ListBackupScheduleOptions{ClusterId: testClusterID}).
		Return([]*entity.BackupSchedule{{
			TenantID:      testTenantID,
			ClusterID:     testClusterID,
			Region:        testRegion,
			ScheduleCron:  testScheduleCron,
			Retention:     testRetentionHour,
			Kubernetes:    testKubernetes,
			CloudProvider: testProvider,
			CreatedAt:     &validFrom,
		}}, nil)
	tombstoneRepo.EXPECT().
		List(gomock.Any(), entity.ListBackupTombstoneOptions{ClusterId: testClusterID}).
		Return(nil, nil)

	backups, total, err := svc.ListBackups(context.Background(), &vo.ListUserBackupsOpt{
		ClusterId: testClusterID,
		PageSize:  10,
	})

	require.NoError(t, err)
	require.Equal(t, uint32(1), total)
	require.Len(t, backups, 1)
	require.Equal(t, generateVirtualBackupID(testClusterID, scheduleTime), backups[0].ID)
}

func TestListBackupsDoesNotProjectCurrentScheduleBeforeCutover(t *testing.T) {
	ctrl := gomock.NewController(t)
	backupRepo := mock_repo.NewMockUserBackupRepo(ctrl)
	scheduleRepo := mock_repo.NewMockBackupScheduleRepo(ctrl)
	historyRepo := mock_repo.NewMockBackupScheduleHistoryRepo(ctrl)
	tombstoneRepo := mock_repo.NewMockBackupTombstoneRepo(ctrl)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	scheduleTime := todayMidnightUTC()
	createdAt := scheduleTime.Add(-7 * 24 * time.Hour)
	cutoverAt := scheduleTime.Add(time.Hour)
	expireTime := scheduleTime.Add(24 * time.Hour)
	svc := &backupRestoreService{
		backupRepo:          backupRepo,
		backupScheduleRepo:  scheduleRepo,
		backupHistoryRepo:   historyRepo,
		backupTombstoneRepo: tombstoneRepo,
		mgmtSvc:             mgmtSvc,
	}

	backupRepo.EXPECT().List(gomock.Any(), gomock.Any()).Return([]*entity.UserBackup{{
		ID:         "bak-persisted-before-cutover",
		State:      constant.BackupStateSucceeded,
		ClusterID:  testClusterID,
		TenantID:   testTenantID,
		BackupTime: &scheduleTime,
		ExpireTime: &expireTime,
	}}, nil)
	historyRepo.EXPECT().List(gomock.Any(), gomock.Any()).Return(nil, nil)
	historyRepo.EXPECT().GetVirtualizationCutover(gomock.Any()).Return(cutoverAt, nil)
	scheduleRepo.EXPECT().List(gomock.Any(), entity.ListBackupScheduleOptions{ClusterId: testClusterID}).
		Return([]*entity.BackupSchedule{{
			TenantID:      testTenantID,
			ClusterID:     testClusterID,
			Region:        testRegion,
			ScheduleCron:  testScheduleCron,
			Retention:     testRetentionHour,
			Kubernetes:    testKubernetes,
			CloudProvider: testProvider,
			CreatedAt:     &createdAt,
		}}, nil)
	tombstoneRepo.EXPECT().
		List(gomock.Any(), entity.ListBackupTombstoneOptions{ClusterId: testClusterID}).
		Return(nil, nil)

	backups, total, err := svc.ListBackups(context.Background(), &vo.ListUserBackupsOpt{
		ClusterId: testClusterID,
		PageSize:  10,
	})

	require.NoError(t, err)
	require.Equal(t, uint32(1), total)
	require.Len(t, backups, 1)
	require.Equal(t, "bak-persisted-before-cutover", backups[0].ID)
}

func TestBuildRestoreRequestWithVirtualBackupResolvesRealBackup(t *testing.T) {
	ctrl := gomock.NewController(t)
	svc, _, historyRepo, tombstoneRepo := newBackupRestoreServiceForVirtualTest(ctrl)
	withoutCurrentSchedule(svc)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	proxyProvider := mock_facade.NewMockClusterProxyProviderFacade(ctrl)
	proxy := mock_facade.NewMockServerlessClusterProxyFacade(ctrl)
	svc.mgmtSvc = mgmtSvc
	svc.clusterProxyProvider = proxyProvider
	scheduleTime := todayMidnightUTC()
	backupID := generateVirtualBackupID(testClusterID, scheduleTime)
	actualBackupTime := scheduleTime.Add(30 * time.Second)

	withTestClusterProxyConfig(t)
	tombstoneRepo.EXPECT().Exists(gomock.Any(), testClusterID, scheduleTime).Return(false, nil)
	historyRepo.EXPECT().
		List(gomock.Any(), gomock.Any()).
		Return([]*entity.BackupScheduleHistory{historyForSchedule(scheduleTime)}, nil)
	proxyProvider.EXPECT().GetServerlessProxy(gomock.Any(), testProxyURL).Return(proxy, nil)
	proxy.EXPECT().
		ListBackups(gomock.Any(), gomock.Any()).
		Return(&pkgv2.ListBackupResponse{Items: []pkgv2.BackupItem{{
			ID:   999,
			Name: "real-backup-for-restore",
			Time: pkgv2.JSONTime(actualBackupTime),
		}}}, nil)
	mgmtSvc.EXPECT().GetClusterIncludeDeleted(gomock.Any(), uint64(2002)).Return(activeCluster(), nil)

	request, cluster, backup, err := svc.buildRestoreRequestWithBackup(context.Background(), nil, backupID)

	require.NoError(t, err)
	require.Equal(t, uint64(2002), cluster.GetClusterId())
	require.Equal(t, backupID, backup.ID)
	require.Equal(t, uint64(999), backup.BackupID)
	require.Equal(t, "real-backup-for-restore", backup.BackupName)
	require.Equal(t, constant.BackupStateSucceeded, backup.State)
	require.Equal(t, constant.BackupTypeAuto, backup.Type)
	require.True(t, scheduleTime.Equal(*backup.ScheduleTime))
	require.True(t, actualBackupTime.Equal(*backup.BackupTime))
	require.NotNil(t, backup.ExpireTime)
	require.NotNil(t, request.Backup)
	require.Equal(t, uint64(999), request.Backup.ID)
	require.Equal(t, testSourceToken, *request.SourceToken)
	require.True(t, actualBackupTime.Equal(time.Time(request.Backup.Time)))
}

func TestDeleteRecycledInstanceBackupCreatesTombstoneForVirtualBackup(t *testing.T) {
	ctrl := gomock.NewController(t)
	svc, _, historyRepo, tombstoneRepo := newBackupRestoreServiceForVirtualTest(ctrl)
	withoutCurrentSchedule(svc)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	dbProvider := mock_db.NewMockTransProvider(ctrl)
	svc.mgmtSvc = mgmtSvc
	svc.dbProvider = dbProvider
	scheduleTime := todayMidnightUTC()
	backupID := generateVirtualBackupID(testClusterID, scheduleTime)

	tombstoneRepo.EXPECT().Exists(gomock.Any(), testClusterID, scheduleTime).Return(false, nil)
	historyRepo.EXPECT().
		List(gomock.Any(), gomock.Any()).
		Return([]*entity.BackupScheduleHistory{historyForSchedule(scheduleTime)}, nil)
	mgmtSvc.EXPECT().GetClusterIncludeDeleted(gomock.Any(), uint64(2002)).Return(deletedCluster(), nil)
	dbProvider.EXPECT().Transaction(gomock.Any(), gomock.Any()).DoAndReturn(
		func(ctx context.Context, callback func(context.Context) error) error { return callback(ctx) },
	)
	tombstoneRepo.EXPECT().
		Create(gomock.Any(), gomock.AssignableToTypeOf(&entity.BackupTombstone{})).
		DoAndReturn(func(_ context.Context, tombstone *entity.BackupTombstone) error {
			require.Equal(t, testTenantID, tombstone.TenantID)
			require.Equal(t, testClusterID, tombstone.ClusterID)
			require.Equal(t, backupID, tombstone.BackupID)
			require.True(t, scheduleTime.Equal(*tombstone.ScheduleTime))
			require.NotNil(t, tombstone.DeletedAt)
			return nil
		})

	err := svc.DeleteRecycledInstanceBackup(context.Background(), testTenantID, testClusterID, backupID)

	require.NoError(t, err)
}

func TestDeleteVirtualBackupRejectsActiveCluster(t *testing.T) {
	ctrl := gomock.NewController(t)
	svc, _, historyRepo, tombstoneRepo := newBackupRestoreServiceForVirtualTest(ctrl)
	withoutCurrentSchedule(svc)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	svc.mgmtSvc = mgmtSvc
	scheduleTime := todayMidnightUTC()
	backupID := generateVirtualBackupID(testClusterID, scheduleTime)

	tombstoneRepo.EXPECT().Exists(gomock.Any(), testClusterID, scheduleTime).Return(false, nil)
	historyRepo.EXPECT().List(gomock.Any(), gomock.Any()).Return(
		[]*entity.BackupScheduleHistory{historyForSchedule(scheduleTime)}, nil,
	)
	mgmtSvc.EXPECT().GetClusterIncludeDeleted(gomock.Any(), uint64(2002)).Return(activeCluster(), nil)

	err := svc.DeleteRecycledInstanceBackup(context.Background(), testTenantID, testClusterID, backupID)

	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeRecordNotFound, err.ErrCode())
}

func TestDeleteVirtualBackupWorksWithoutWhitelistedOrg(t *testing.T) {
	ctrl := gomock.NewController(t)
	svc, _, historyRepo, tombstoneRepo := newBackupRestoreServiceForVirtualTest(ctrl)
	withoutCurrentSchedule(svc)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	dbProvider := mock_db.NewMockTransProvider(ctrl)
	svc.mgmtSvc = mgmtSvc
	svc.dbProvider = dbProvider
	scheduleTime := todayMidnightUTC()
	backupID := generateVirtualBackupID(testClusterID, scheduleTime)

	tombstoneRepo.EXPECT().Exists(gomock.Any(), testClusterID, scheduleTime).Return(false, nil)
	historyRepo.EXPECT().List(gomock.Any(), gomock.Any()).Return(
		[]*entity.BackupScheduleHistory{historyForSchedule(scheduleTime)}, nil,
	)
	mgmtSvc.EXPECT().GetClusterIncludeDeleted(gomock.Any(), uint64(2002)).Return(deletedCluster(), nil)
	dbProvider.EXPECT().Transaction(gomock.Any(), gomock.Any()).DoAndReturn(
		func(ctx context.Context, callback func(context.Context) error) error { return callback(ctx) },
	)
	tombstoneRepo.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil)

	err := svc.DeleteRecycledInstanceBackup(context.Background(), testTenantID, testClusterID, backupID)

	require.NoError(t, err)
}

func TestDeleteRecycledInstancePersistedBackupCreatesTombstoneAtomically(t *testing.T) {
	ctrl := gomock.NewController(t)
	backupRepo := mock_repo.NewMockUserBackupRepo(ctrl)
	tombstoneRepo := mock_repo.NewMockBackupTombstoneRepo(ctrl)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	dbProvider := mock_db.NewMockTransProvider(ctrl)
	scheduleTime := todayMidnightUTC()
	backup := &entity.UserBackup{
		ID:           "legacy-backup",
		TenantID:     testTenantID,
		ClusterID:    testClusterID,
		State:        constant.BackupStateSucceeded,
		ScheduleTime: &scheduleTime,
	}
	svc := &backupRestoreService{
		backupRepo:          backupRepo,
		backupTombstoneRepo: tombstoneRepo,
		mgmtSvc:             mgmtSvc,
		dbProvider:          dbProvider,
	}

	backupRepo.EXPECT().GetByClusterIdAndBackupId(gomock.Any(), testClusterID, backup.ID).Return(backup, nil)
	tombstoneRepo.EXPECT().Exists(gomock.Any(), testClusterID, scheduleTime).Return(false, nil)
	mgmtSvc.EXPECT().GetClusterIncludeDeleted(gomock.Any(), uint64(2002)).Return(deletedCluster(), nil)
	dbProvider.EXPECT().Transaction(gomock.Any(), gomock.Any()).DoAndReturn(
		func(ctx context.Context, callback func(context.Context) error) error { return callback(ctx) },
	)
	tombstoneRepo.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(
		func(_ context.Context, tombstone *entity.BackupTombstone) error {
			require.Equal(t, generateVirtualBackupID(testClusterID, scheduleTime), tombstone.BackupID)
			require.Equal(t, testClusterID, tombstone.ClusterID)
			require.Equal(t, scheduleTime, *tombstone.ScheduleTime)
			return nil
		},
	)
	backupRepo.EXPECT().DeleteByBackupId(gomock.Any(), backup.ID).Return(nil)

	err := svc.DeleteRecycledInstanceBackup(context.Background(), testTenantID, testClusterID, backup.ID)

	require.NoError(t, err)
}

func TestUpdateBackupSchedulesArchivesHistoryWithoutWhitelistedOrg(t *testing.T) {
	ctrl := gomock.NewController(t)
	scheduleRepo := mock_repo.NewMockBackupScheduleRepo(ctrl)
	historyRepo := mock_repo.NewMockBackupScheduleHistoryRepo(ctrl)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	dbProvider := mock_db.NewMockTransProvider(ctrl)
	svc := &backupRestoreService{
		backupScheduleRepo: scheduleRepo,
		backupHistoryRepo:  historyRepo,
		mgmtSvc:            mgmtSvc,
		dbProvider:         dbProvider,
	}
	cluster := activeCluster()
	cluster.GetServerlessCluster().GetCluster().AutoScaling = &regionalserverlessv1.ServerlessRegionalServiceClusterInfo_AutoScaling{
		MinRcu: 1,
		MaxRcu: 1,
	}
	createdAt := time.Now().Add(-24 * time.Hour).UTC()
	oldSchedule := &entity.BackupSchedule{
		ID:            1,
		CreatedAt:     &createdAt,
		TenantID:      testTenantID,
		ClusterID:     testClusterID,
		Region:        testRegion,
		ScheduleCron:  testScheduleCron,
		Retention:     testRetentionHour,
		Kubernetes:    testKubernetes,
		CloudProvider: testProvider,
	}

	mgmtSvc.EXPECT().GetCluster(gomock.Any(), uint64(2002), gomock.Any()).Return(cluster, nil)
	scheduleRepo.EXPECT().List(gomock.Any(), entity.ListBackupScheduleOptions{ClusterId: testClusterID}).
		Return([]*entity.BackupSchedule{oldSchedule}, nil)
	dbProvider.EXPECT().Transaction(gomock.Any(), gomock.Any()).DoAndReturn(
		func(ctx context.Context, callback func(context.Context) error) error { return callback(ctx) },
	)
	historyRepo.EXPECT().ArchiveCurrentSchedule(gomock.Any(), gomock.Any(), gomock.Any(), nil).
		DoAndReturn(func(_ context.Context, schedule *entity.BackupSchedule, _ time.Time, _ *time.Time) error {
			require.Equal(t, testScheduleCron, schedule.ScheduleCron)
			require.Equal(t, testRetentionHour, schedule.Retention)
			return nil
		})
	scheduleRepo.EXPECT().Save(gomock.Any(), gomock.Any()).
		DoAndReturn(func(_ context.Context, schedule *entity.BackupSchedule) (*entity.BackupSchedule, error) {
			require.Equal(t, "0 00 12 * * *", schedule.ScheduleCron)
			require.Equal(t, 7*24, schedule.Retention)
			return schedule, nil
		})

	_, err := svc.UpdateBackupSchedules(context.Background(), &vo.UpdateBackupSchedulesOpt{
		ClusterId:     testClusterID,
		ScheduleTime:  "12:00",
		RetentionDays: 7,
	})

	require.NoError(t, err)
}

func newBackupRestoreServiceForVirtualTest(
	ctrl *gomock.Controller,
) (*backupRestoreService, *mock_repo.MockUserBackupRepo, *mock_repo.MockBackupScheduleHistoryRepo, *mock_repo.MockBackupTombstoneRepo) {
	backupRepo := mock_repo.NewMockUserBackupRepo(ctrl)
	scheduleRepo := mock_repo.NewMockBackupScheduleRepo(ctrl)
	historyRepo := mock_repo.NewMockBackupScheduleHistoryRepo(ctrl)
	tombstoneRepo := mock_repo.NewMockBackupTombstoneRepo(ctrl)
	return &backupRestoreService{
		backupRepo:          backupRepo,
		backupScheduleRepo:  scheduleRepo,
		backupHistoryRepo:   historyRepo,
		backupTombstoneRepo: tombstoneRepo,
	}, backupRepo, historyRepo, tombstoneRepo
}

// withoutCurrentSchedule tells the schedule repo to return no active schedule.
// Use this when a test exercises virtual backup listing for clusters that only
// have history records.
func withoutCurrentSchedule(svc *backupRestoreService) {
	svc.backupScheduleRepo.(*mock_repo.MockBackupScheduleRepo).
		EXPECT().List(gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes()
}

func withReadyPhysicalSlot(
	t *testing.T,
	ctrl *gomock.Controller,
	svc *backupRestoreService,
	scheduleTime, backupTime time.Time,
) {
	t.Helper()
	slotRepo := mock_repo.NewMockPhysicalBackupSlotRepo(ctrl)
	svc.physicalBackupSlotRepo = slotRepo
	slotRepo.EXPECT().List(gomock.Any(), gomock.Any()).Return([]*entity.PhysicalBackupSlot{{
		CloudProvider: testProvider,
		Region:        testRegion,
		Kubernetes:    testKubernetes,
		SlotTime:      scheduleTime,
		BackupTime:    backupTime,
	}}, nil)
	slotRepo.EXPECT().ListByBackupTime(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil)
}

func todayMidnightUTC() time.Time {
	return time.Now().UTC().Truncate(24 * time.Hour)
}

func historyForSchedule(scheduleTime time.Time) *entity.BackupScheduleHistory {
	validFrom := scheduleTime.Add(-time.Hour)
	return &entity.BackupScheduleHistory{
		TenantID:      testTenantID,
		ClusterID:     testClusterID,
		Region:        testRegion,
		ScheduleCron:  testScheduleCron,
		Retention:     testRetentionHour,
		Kubernetes:    testKubernetes,
		CloudProvider: testProvider,
		SourceToken:   testSourceToken,
		ValidFrom:     &validFrom,
	}
}

func withTestClusterProxyConfig(t *testing.T) {
	t.Helper()
	originalRegionalConfig := config.Global().WorkerConfig.RegionalConfig
	config.Global().WorkerConfig.RegionalConfig = []config.RegionalConfig{{
		Provider: testProvider,
		Region:   testRegion,
		K8sConfig: []config.K8sConfig{{
			Kubernetes:      testKubernetes,
			ClusterProxyUrl: testProxyURL,
		}},
	}}
	t.Cleanup(func() {
		config.Global().WorkerConfig.RegionalConfig = originalRegionalConfig
	})
}

func deletedCluster() *tidbmgmtservicev1.Cluster {
	deleteTime := timestamppb.New(time.Now().Add(-time.Hour))
	cluster := activeCluster()
	cluster.DeleteTime = deleteTime
	return cluster
}

func activeCluster() *tidbmgmtservicev1.Cluster {
	creationFinishedAt := time.Now().Add(-24 * time.Hour).Unix()
	return &tidbmgmtservicev1.Cluster{
		ClusterId:      2002,
		ProjectId:      3003,
		OrganizationId: 1001,
		CloudProvider:  testProvider,
		Region:         testRegion,
		ClusterInfo: &tidbmgmtservicev1.Cluster_ServerlessCluster{
			ServerlessCluster: &globalserverlessv1.ServerlessServiceCluster{
				CreationFinishedAt: &creationFinishedAt,
				Cluster: &regionalserverlessv1.ServerlessRegionalServiceClusterInfo{
					UserPrefix: testSourceToken,
					Kubernetes: testKubernetes,
				},
			},
		},
	}
}