package impl

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

	"github.com/stretchr/testify/assert"
	"github.com/tidbcloud/dataflow-service/pkg/log"

	"github.com/AlekSi/pointer"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/dataflow-service/app/repo"
	backup_repo "github.com/tidbcloud/dataflow-service/app/repo/backup"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/pkgv2/db"
)

func getFakeDBProvider(t *testing.T) db.Provider {
	testDB, err := repo.GetFakeDB()
	require.NoError(t, err)
	return db.NewProvider(testDB)
}

func newDummyBackupSchedule(clusterID entity.ID,
	deletedAt *time.Time,
	backupType entity.BackupType,
	clusterType entity.BackupClusterType,
) *entity.BackupSchedule {
	return &entity.BackupSchedule{
		ClusterID:   clusterID,
		DeletedAt:   deletedAt,
		Type:        backupType,
		ClusterType: clusterType,
		BackupCycle: entity.BackupCycleDaily,
		Method:      entity.BackupMethodFull,
	}
}

func newDummyBackupSetting(clusterID entity.ID,
	autoBackupDisabled *bool,
) *entity.BackupSetting {
	setting := &entity.BackupSetting{
		ClusterID: clusterID,
		Status:    "normal",
	}
	if autoBackupDisabled != nil {
		setting.AutoBackup = &entity.AutoBackupSetting{Disable: *autoBackupDisabled}
	}
	return setting
}

func Test_getEnabledAutoBackupSchedules_happy(t *testing.T) {
	// setup
	// setup AutoBackupScheduler
	transProvider := getFakeDBProvider(t)
	scheduler := &autoBackupSchedulerImpl{}
	scheduler.backupScheduleRepo = backup_repo.NewBackupScheduleRepo(transProvider)
	scheduler.backupSettingRepo = backup_repo.NewBackupSettingRepo(transProvider)

	// initial deleted and non-deleted schedules
	// serverless schedule
	err := scheduler.backupScheduleRepo.Create(context.Background(), newDummyBackupSchedule(
		entity.ID(12341),
		nil,
		entity.BackupTypeAuto,
		entity.BackupClusterTypeServerless))
	require.Nil(t, err)
	err = scheduler.backupSettingRepo.Save(context.Background(), newDummyBackupSetting(entity.ID(12341), nil))
	require.Nil(t, err)
	// deleted schedule
	err = scheduler.backupScheduleRepo.Create(context.Background(), newDummyBackupSchedule(
		entity.ID(12342),
		pointer.ToTime(time.Now()),
		entity.BackupTypeAuto,
		entity.BackupClusterTypeDedicated))
	require.Nil(t, err)
	err = scheduler.backupSettingRepo.Save(context.Background(), newDummyBackupSetting(
		entity.ID(12342), nil))
	require.Nil(t, err)
	// normal schedule
	err = scheduler.backupScheduleRepo.Create(context.Background(), newDummyBackupSchedule(
		entity.ID(12343),
		nil,
		entity.BackupTypeAuto,
		entity.BackupClusterTypeDedicated))
	require.Nil(t, err)
	err = scheduler.backupSettingRepo.Save(context.Background(), newDummyBackupSetting(
		entity.ID(12343), pointer.ToBool(true)))
	require.Nil(t, err)
	// auo-backup disabled schedule
	err = scheduler.backupScheduleRepo.Create(context.Background(), newDummyBackupSchedule(
		entity.ID(12344),
		nil,
		entity.BackupTypeAuto,
		entity.BackupClusterTypeDedicated))
	require.Nil(t, err)
	err = scheduler.backupSettingRepo.Save(context.Background(), newDummyBackupSetting(
		entity.ID(12344), pointer.ToBool(false)))
	require.Nil(t, err)
	// normal schedule
	err = scheduler.backupScheduleRepo.Create(context.Background(), newDummyBackupSchedule(
		entity.ID(12345),
		nil,
		entity.BackupTypeAuto,
		entity.BackupClusterTypeDedicated))
	require.Nil(t, err)
	err = scheduler.backupSettingRepo.Save(context.Background(), newDummyBackupSetting(
		entity.ID(12345), nil))
	require.Nil(t, err)
	// schedule without backup setting
	err = scheduler.backupScheduleRepo.Create(context.Background(), newDummyBackupSchedule(
		entity.ID(12346),
		nil,
		entity.BackupTypeAuto,
		entity.BackupClusterTypeDedicated))
	require.Nil(t, err)

	// exercise
	schedules, err := scheduler.getEnabledAutoBackupSchedules(context.Background())

	// verify
	require.Nil(t, err)

	// verify only have two normal schedules
	require.Equal(t, 2, len(schedules))
	schedulesMap := map[entity.ID]entity.BackupSchedule{}
	for _, s := range schedules {
		schedulesMap[s.ClusterID] = s
	}
	require.NotNil(t, schedulesMap[entity.ID(12344)])
	require.NotNil(t, schedulesMap[entity.ID(12345)])
}

func TestCalDeleteAutoBackups(t *testing.T) {
	tests := []struct {
		name           string
		backups        []entity.Backup
		expiredTSO     uint64
		isPitrEnable   bool
		mockSetup      func()
		expectedResult []entity.Backup
		expectedError  error
	}{
		{
			name:           "No backups",
			backups:        []entity.Backup{},
			expiredTSO:     100,
			isPitrEnable:   false,
			expectedResult: nil,
			expectedError:  nil,
		},
		{
			name: "No expired backups",
			backups: []entity.Backup{
				{StartTimeStamp: "110", ClusterID: 1},
				{StartTimeStamp: "120", ClusterID: 1},
			},
			expiredTSO:     100,
			isPitrEnable:   false,
			expectedResult: nil,
			expectedError:  nil,
		},
		{
			name: "Two backups expired with PiTR",
			backups: []entity.Backup{
				{StartTimeStamp: "90", ClusterID: 1},
				{StartTimeStamp: "95", ClusterID: 1},
			},
			expiredTSO:   100,
			isPitrEnable: true,
			expectedResult: []entity.Backup{
				{StartTimeStamp: "90", ClusterID: 1},
			},
			expectedError: nil,
		},
		{
			name: "One backups expired with PiTR",
			backups: []entity.Backup{
				{StartTimeStamp: "90", ClusterID: 1},
			},
			expiredTSO:     100,
			isPitrEnable:   true,
			mockSetup:      func() {},
			expectedResult: []entity.Backup{},
			expectedError:  nil,
		},
		{
			name: "Trigger latest backup protection(more than one expired)",
			backups: []entity.Backup{
				{StartTimeStamp: "90", ClusterID: 1},
				{StartTimeStamp: "95", ClusterID: 1},
			},
			expiredTSO:   100,
			isPitrEnable: false,
			expectedResult: []entity.Backup{
				{StartTimeStamp: "90", ClusterID: 1},
			},
			expectedError: nil,
		},
		{
			name: "Trigger latest backup protection(one expired)",
			backups: []entity.Backup{
				{StartTimeStamp: "90", ClusterID: 1},
			},
			expiredTSO:     100,
			isPitrEnable:   false,
			expectedResult: []entity.Backup{},
			expectedError:  nil,
		},
		{
			name: "Not all expired",
			backups: []entity.Backup{
				{StartTimeStamp: "90", ClusterID: 1},
				{StartTimeStamp: "110", ClusterID: 1},
			},
			expiredTSO:   100,
			isPitrEnable: false,
			mockSetup:    func() {},
			expectedResult: []entity.Backup{
				{StartTimeStamp: "90", ClusterID: 1},
			},
			expectedError: nil,
		},
		{
			name: "Backups are not sorted by .startTimeStamp in ascending order",
			backups: []entity.Backup{
				{StartTimeStamp: "110", ClusterID: 1},
				{StartTimeStamp: "90", ClusterID: 1},
			},
			expiredTSO:     100,
			isPitrEnable:   false,
			expectedResult: nil,
			expectedError:  fmt.Errorf("backups are not sorted by .startTimeStamp in ascending order"),
		},
		{
			name: "contain invalid TSO",
			backups: []entity.Backup{
				{StartTimeStamp: "invalid tso", ClusterID: 1},
			},
			expiredTSO:     100,
			isPitrEnable:   false,
			expectedResult: nil,
			expectedError:  fmt.Errorf("parsing \"invalid tso\""),
		},
		{
			name: "Protected backups are filtered out before deletion",
			backups: []entity.Backup{
				{StartTimeStamp: "90", ClusterID: 1, Protected: true},
				{StartTimeStamp: "95", ClusterID: 1, Protected: false},
			},
			expiredTSO:   100,
			isPitrEnable: false,
			// The protected backup is filtered out first, leaving only one backup (95)
			// Then latest backup protection kicks in, keeping that one backup
			expectedResult: []entity.Backup{},
			expectedError:  nil,
		},
		{
			name: "Protected backups with PiTR - last protected backup is treated as deleted",
			backups: []entity.Backup{
				{StartTimeStamp: "90", ClusterID: 1, Protected: false},
				{StartTimeStamp: "95", ClusterID: 1, Protected: true},
			},
			expiredTSO:   100,
			isPitrEnable: true,
			// The protected backup (95) is filtered out first, leaving only one backup (90)
			// With PiTR enabled, we keep the last backup, so nothing is deleted
			expectedResult: []entity.Backup{},
			expectedError:  nil,
		},
		{
			name: "All expired backups are protected - none should be deleted",
			backups: []entity.Backup{
				{StartTimeStamp: "90", ClusterID: 1, Protected: true},
				{StartTimeStamp: "95", ClusterID: 1, Protected: true},
			},
			expiredTSO:   100,
			isPitrEnable: false,
			// All protected backups are filtered out, so nothing to delete
			expectedResult: nil,
			expectedError:  nil,
		},
	}

	for _, tt := range tests {
		logger := log.GetLogger(context.Background())
		t.Run(tt.name, func(t *testing.T) {
			result, err := calDeleteAutoBackups(logger, tt.backups, tt.expiredTSO, tt.isPitrEnable)
			if tt.expectedError != nil {
				assert.Error(t, err)
				assert.Contains(t, err.Error(), tt.expectedError.Error())
			} else {
				assert.NoError(t, err)
			}
			assert.Equal(t, tt.expectedResult, result)
		})
	}
}

func TestFilterProtectedBackups(t *testing.T) {
	testCases := map[string]struct {
		backups              []entity.Backup
		expectedFiltered     []entity.Backup
		expectedProtectedCnt int
	}{
		"no backups": {
			backups:              []entity.Backup{},
			expectedFiltered:     []entity.Backup{},
			expectedProtectedCnt: 0,
		},
		"all backups are protected": {
			backups: []entity.Backup{
				{Updatable: entity.Updatable{Base: entity.Base{ID: entity.ID(1)}}, Protected: true},
				{Updatable: entity.Updatable{Base: entity.Base{ID: entity.ID(2)}}, Protected: true},
				{Updatable: entity.Updatable{Base: entity.Base{ID: entity.ID(3)}}, Protected: true},
			},
			expectedFiltered:     []entity.Backup{},
			expectedProtectedCnt: 3,
		},
		"no backups are protected": {
			backups: []entity.Backup{
				{Updatable: entity.Updatable{Base: entity.Base{ID: entity.ID(1)}}, Protected: false},
				{Updatable: entity.Updatable{Base: entity.Base{ID: entity.ID(2)}}, Protected: false},
			},
			expectedFiltered: []entity.Backup{
				{Updatable: entity.Updatable{Base: entity.Base{ID: entity.ID(1)}}, Protected: false},
				{Updatable: entity.Updatable{Base: entity.Base{ID: entity.ID(2)}}, Protected: false},
			},
			expectedProtectedCnt: 0,
		},
		"mixed protected and unprotected backups": {
			backups: []entity.Backup{
				{Updatable: entity.Updatable{Base: entity.Base{ID: entity.ID(1)}}, Protected: true},
				{Updatable: entity.Updatable{Base: entity.Base{ID: entity.ID(2)}}, Protected: false},
				{Updatable: entity.Updatable{Base: entity.Base{ID: entity.ID(3)}}, Protected: true},
				{Updatable: entity.Updatable{Base: entity.Base{ID: entity.ID(4)}}, Protected: false},
			},
			expectedFiltered: []entity.Backup{
				{Updatable: entity.Updatable{Base: entity.Base{ID: entity.ID(2)}}, Protected: false},
				{Updatable: entity.Updatable{Base: entity.Base{ID: entity.ID(4)}}, Protected: false},
			},
			expectedProtectedCnt: 2,
		},
	}

	for name, tc := range testCases {
		t.Run(name, func(t *testing.T) {
			filtered, protectedCount := filterProtectedBackups(tc.backups)
			assert.Equal(t, tc.expectedProtectedCnt, protectedCount)
			assert.Equal(t, len(tc.expectedFiltered), len(filtered))

			// Verify that all returned backups are not protected
			for _, backup := range filtered {
				assert.False(t, backup.Protected, "filtered backups should not be protected")
			}
		})
	}
}