package backupv2_test

import (
	"context"
	"errors"
	"testing"
	"time"

	"github.com/AlekSi/pointer"
	dagentity "github.com/shiningrush/fastflow/pkg/entity"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	"github.com/stretchr/testify/suite"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/app/config"
	feature_gate_mock "github.com/tidbcloud/dataflow-service/app/internal/feature_gate/mock"
	"github.com/tidbcloud/dataflow-service/app/models/backup"
	"github.com/tidbcloud/dataflow-service/app/models/backup/impl"
	backup_model_mock "github.com/tidbcloud/dataflow-service/app/models/backup/mock"
	"github.com/tidbcloud/dataflow-service/app/models/br"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/repo"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	repo_mock "github.com/tidbcloud/dataflow-service/app/repo/mock"
	"github.com/tidbcloud/dataflow-service/app/workflow"
	workflow_mock "github.com/tidbcloud/dataflow-service/app/workflow/mock"
	internal_config "github.com/tidbcloud/dataflow-service/internal/config"
	"github.com/tidbcloud/dataflow-service/internal/constants"
	"github.com/tidbcloud/dataflow-service/internal/model/util"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	"github.com/tidbcloud/dataflow-service/pkg/blob"
	blob_mock "github.com/tidbcloud/dataflow-service/pkg/blob/mock"
	grpcli_mock "github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli/mock"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	dedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"gorm.io/gorm"
)

type pitBackupTestSuite struct {
	suite.Suite
	db *gorm.DB
	m  backup.PitBackupModel

	mockWorkflow            *workflow_mock.Workflow
	mockCoreGrpcClient      *grpcli_mock.CoreGrpcClient
	mockMgmtGrpcClient      *grpcli_mock.MgmtSvcGrpcClient
	mockPitBackupSwitchRepo *repo_mock.PitBackupSwitchRepo
	mockBackupRepo          *repo_mock.BackupRepo
	mockLogBackupRepo       *repo_mock.LogBackupRepo
	mockSnapshotBackupModel *backup_model_mock.SnapshotBackupModel
	mockBackupSettingM      *backup_model_mock.BackupSettingModel
	mockBackupScheduleRepo  *repo_mock.BackupScheduleRepo
	mockRestoreRepo         *repo_mock.RestoreRepo
	mockFeatureGate         *feature_gate_mock.FeatureGate
	mockBucketRepo          *repo_mock.BucketRepo

	mockedUserID        uint64
	allMocks            []*mock.Call
	mockedCluster       *mgmt.Cluster
	mockedBackupSetting *entity.BackupSetting
}

func TestPitBackupTestSuite(t *testing.T) {
	suite.Run(t, new(pitBackupTestSuite))
}

func (s *pitBackupTestSuite) SetupSuite() {
	s.mockPitBackupSwitchRepo = repo_mock.NewPitBackupSwitchRepo(s.T())
	s.mockBackupRepo = repo_mock.NewBackupRepo(s.T())
	s.mockLogBackupRepo = repo_mock.NewLogBackupRepo(s.T())
	s.mockCoreGrpcClient = grpcli_mock.NewCoreGrpcClient(s.T())
	s.mockMgmtGrpcClient = grpcli_mock.NewMgmtSvcGrpcClient(s.T())
	s.mockBackupScheduleRepo = repo_mock.NewBackupScheduleRepo(s.T())
	s.mockRestoreRepo = repo_mock.NewRestoreRepo(s.T())
	s.mockBucketRepo = repo_mock.NewBucketRepo(s.T())

	s.mockWorkflow = workflow_mock.NewWorkflow(s.T())
	s.mockedUserID = 1
	s.mockedCluster = &mgmt.Cluster{
		ClusterId: 1, ProjectId: 1, OrganizationId: 1,
		ClusterInfo: &mgmt.Cluster_DedicatedCluster{
			DedicatedCluster: &dedicatedv1.Cluster{
				State:   dedicatedv1.Cluster_AVAILABLE,
				Version: "v7.1.0",
			},
		},
	}
	s.mockedBackupSetting = &entity.BackupSetting{
		Status: entity.BackupSettingStatusNormal,
		Bucket: &entity.BucketSetting{
			Prefix:  pointer.ToString("prefix"),
			Buckets: []entity.BackupSettingBucket{{BucketId: 1, Master: true}},
		},
	}

	s.mockSnapshotBackupModel = backup_model_mock.NewSnapshotBackupModel(s.T())
	s.mockBackupSettingM = backup_model_mock.NewBackupSettingModel(s.T())
	s.mockFeatureGate = feature_gate_mock.NewFeatureGate(s.T())

	testDB, err := repo.GetFakeDB()
	require.NoError(s.T(), err)
	s.db = testDB
	trans := db.NewProvider(s.db)

	s.m = impl.NewPitBackupModelWithStorageFunc(
		trans, s.mockPitBackupSwitchRepo, s.mockLogBackupRepo, s.mockBackupRepo, s.mockRestoreRepo,
		s.mockWorkflow, s.mockCoreGrpcClient, s.mockMgmtGrpcClient, s.mockSnapshotBackupModel, s.mockBackupSettingM, s.mockBackupScheduleRepo, s.mockFeatureGate, s.mockBucketRepo, nil)

	s.NoError(internal_config.DefaultConfig4Test())
	s.NoError(internal_config.Global().BackupConfig.Init())
}

// SetupTest resets mock expectations before each test to prevent state pollution.
func (s *pitBackupTestSuite) SetupTest() {
	// Reset all mock expectations and call history
	s.mockPitBackupSwitchRepo.ExpectedCalls = nil
	s.mockPitBackupSwitchRepo.Calls = nil
	s.mockBackupRepo.ExpectedCalls = nil
	s.mockBackupRepo.Calls = nil
	s.mockLogBackupRepo.ExpectedCalls = nil
	s.mockLogBackupRepo.Calls = nil
	s.mockCoreGrpcClient.ExpectedCalls = nil
	s.mockCoreGrpcClient.Calls = nil
	s.mockMgmtGrpcClient.ExpectedCalls = nil
	s.mockMgmtGrpcClient.Calls = nil
	s.mockBackupScheduleRepo.ExpectedCalls = nil
	s.mockBackupScheduleRepo.Calls = nil
	s.mockRestoreRepo.ExpectedCalls = nil
	s.mockRestoreRepo.Calls = nil
	s.mockBucketRepo.ExpectedCalls = nil
	s.mockBucketRepo.Calls = nil
	s.mockWorkflow.ExpectedCalls = nil
	s.mockWorkflow.Calls = nil
	s.mockSnapshotBackupModel.ExpectedCalls = nil
	s.mockSnapshotBackupModel.Calls = nil
	s.mockBackupSettingM.ExpectedCalls = nil
	s.mockBackupSettingM.Calls = nil
	s.mockFeatureGate.ExpectedCalls = nil
	s.mockFeatureGate.Calls = nil
}

func (s *pitBackupTestSuite) TestAsyncCreateSuccess() {
	ctx := context.Background()
	req := &domain.AsyncCreateBackupReq{
		Name: "test", ClusterID: s.mockedCluster.ClusterId,
		ProjectID: s.mockedCluster.ProjectId, UserID: s.mockedUserID,
	}

	m1 := s.mockMgmtGrpcClient.On("GetAnyDedicatedClusterBasicByID", mock.Anything, mock.Anything).Return(s.mockedCluster, nil)
	dagInstance := &dagentity.DagInstance{BaseInfo: dagentity.BaseInfo{ID: "fake"}}
	m2 := s.mockWorkflow.On("RunDagWithTags", mock.Anything, mock.Anything, mock.Anything).Return(dagInstance, nil)
	pitSwitch := entity.PitBackupSwitch{Status: entity.PitBackupStatusWaitingForEnabling}
	m3 := s.mockPitBackupSwitchRepo.On("List", mock.Anything, mock.Anything).Return([]entity.PitBackupSwitch{
		pitSwitch,
	}, nil)
	m4 := s.mockLogBackupRepo.On("List", mock.Anything, mock.Anything).Return([]entity.LogBackup{}, nil)
	m5 := s.mockLogBackupRepo.On("Save", mock.Anything, mock.Anything).Return(nil)
	backupSettingResp := &domain.QueryBackupSettingRespV2{BackupSetting: s.mockedBackupSetting}
	m6 := s.mockBackupSettingM.On("QueryBackupSetting", mock.Anything, mock.Anything).Return(backupSettingResp, nil)
	m7 := s.mockPitBackupSwitchRepo.On("Save", mock.Anything, mock.Anything).Return(nil)
	m8 := s.mockSnapshotBackupModel.On("AsyncCreate", mock.Anything, mock.Anything).Return(nil, nil)
	m9 := s.mockFeatureGate.On("IsFeatureEnabled", mock.Anything, mock.Anything).Return(false)

	resp, err := s.m.AsyncCreate(ctx, req)
	s.NoError(err)
	s.Equal(resp.WorkflowID, dagInstance.ID)

	m1.Unset()
	m2.Unset()
	m3.Unset()
	m4.Unset()
	m5.Unset()
	m6.Unset()
	m7.Unset()
	m8.Unset()
	m9.Unset()
}

func (s *pitBackupTestSuite) TestAsyncCreateSuccessWithSnapshotBackup() {
	ctx := context.Background()
	req := &domain.AsyncCreateBackupReq{
		Name: "test", ClusterID: s.mockedCluster.ClusterId,
		ProjectID: s.mockedCluster.ProjectId, UserID: s.mockedUserID,
	}

	m1 := s.mockMgmtGrpcClient.On("GetAnyDedicatedClusterBasicByID", mock.Anything, mock.Anything).Return(s.mockedCluster, nil)
	pitSwitch := entity.PitBackupSwitch{Status: entity.PitBackupStatusEnabled}
	m2 := s.mockPitBackupSwitchRepo.On("List", mock.Anything, mock.Anything).Return([]entity.PitBackupSwitch{
		pitSwitch,
	}, nil)
	snapshotResp := &domain.AsyncCreateBackupResp{WorkflowID: "fake"}
	m3 := s.mockSnapshotBackupModel.On("AsyncCreate", mock.Anything, mock.Anything).Return(snapshotResp, nil)

	resp, err := s.m.AsyncCreate(ctx, req)
	s.NoError(err)
	s.Equal(resp.WorkflowID, snapshotResp.WorkflowID)

	m1.Unset()
	m2.Unset()
	m3.Unset()
}

func (s *pitBackupTestSuite) TestAsyncCreateRestartsFromCompensatedWaitingState() {
	ctx := context.Background()
	req := &domain.AsyncCreateBackupReq{
		Name: "restart-retry", ClusterID: s.mockedCluster.ClusterId,
		ProjectID: s.mockedCluster.ProjectId, UserID: s.mockedUserID,
	}

	pitSwitch := entity.PitBackupSwitch{
		Status: entity.PitBackupStatusWaitingForRestarting,
		Action: entity.PitBackupSwitchAction{},
	}
	pitSwitch.ID = 303
	oldLogBackup := entity.LogBackup{Status: entity.LogBackupStatusRunning}
	oldLogBackup.ID = 101
	dagInstance := &dagentity.DagInstance{BaseInfo: dagentity.BaseInfo{ID: "restart-retry-dag"}}

	s.mockMgmtGrpcClient.On("GetAnyDedicatedClusterBasicByID", mock.Anything, req.ClusterID).
		Return(s.mockedCluster, nil).Once()
	s.mockPitBackupSwitchRepo.On("List", mock.Anything, mock.Anything).
		Return([]entity.PitBackupSwitch{pitSwitch}, nil).Once()
	s.mockSnapshotBackupModel.On("AsyncCreate", mock.Anything, mock.MatchedBy(func(got *domain.AsyncCreateBackupReq) bool {
		return got.ClusterID == req.ClusterID && got.DryRun
	})).Return(nil, nil).Once()
	s.mockLogBackupRepo.On("List", mock.Anything, mock.Anything).
		Return([]entity.LogBackup{oldLogBackup}, nil).Once()
	s.mockBackupSettingM.On("QueryBackupSetting", mock.Anything, mock.Anything).
		Return(&domain.QueryBackupSettingRespV2{BackupSetting: s.mockedBackupSetting}, nil).Once()
	s.mockLogBackupRepo.On("Save", mock.Anything, mock.AnythingOfType("*entity.LogBackup")).
		Run(func(args mock.Arguments) {
			created := args.Get(1).(*entity.LogBackup)
			created.ID = 202
			created.BackupScheduleID = 404
		}).Return(nil).Once()
	s.mockWorkflow.On("RunDagWithTags", workflow.DagRestartPitBackup,
		mock.MatchedBy(func(vars map[string]string) bool {
			return vars[workflow.KeyDisableLogBackupID] == "101" &&
				vars[workflow.KeyEnableLogBackupID] == "202" &&
				vars[workflow.KeyPitBackupSwitchID] == "303"
		}), mock.Anything).Return(dagInstance, nil).Once()
	var savedSwitch *entity.PitBackupSwitch
	s.mockPitBackupSwitchRepo.On("Save", mock.Anything, mock.AnythingOfType("*entity.PitBackupSwitch")).
		Run(func(args mock.Arguments) {
			savedSwitch = args.Get(1).(*entity.PitBackupSwitch)
		}).Return(nil).Once()

	resp, err := s.m.AsyncCreate(ctx, req)

	s.NoError(err)
	s.Equal(dagInstance.ID, resp.WorkflowID)
	s.Equal(uint64(404), resp.BackupScheduleID)
	s.Require().NotNil(savedSwitch)
	s.Equal(entity.PitBackupStatusRestarting, savedSwitch.Status)
	s.Equal([]string{string(entity.PitBackupSwitchActionDoRestart)}, savedSwitch.Action.CurrentActions)
	s.mockSnapshotBackupModel.AssertNumberOfCalls(s.T(), "AsyncCreate", 1)
}

func (s *pitBackupTestSuite) TestAsyncCreatePreCheck_ClusterVersion() {
	ctx := context.Background()
	req := &domain.AsyncCreateBackupReq{
		Name: "test", ClusterID: s.mockedCluster.ClusterId,
		ProjectID: s.mockedCluster.ProjectId, UserID: s.mockedUserID,
	}
	pitSwitch := entity.PitBackupSwitch{Status: entity.PitBackupStatusWaitingForEnabling}
	m1 := s.mockPitBackupSwitchRepo.On("List", mock.Anything, mock.Anything).Return([]entity.PitBackupSwitch{
		pitSwitch,
	}, nil)

	mockedCluster := &mgmt.Cluster{
		ClusterId: 1, ProjectId: 1, OrganizationId: 1,
		ClusterInfo: &mgmt.Cluster_DedicatedCluster{
			DedicatedCluster: &dedicatedv1.Cluster{
				State:   dedicatedv1.Cluster_AVAILABLE,
				Version: "v1.1.0",
			},
		},
	}
	m2 := s.mockMgmtGrpcClient.On("GetAnyDedicatedClusterBasicByID", mock.Anything, mock.Anything).Return(mockedCluster, nil)
	m3 := s.mockSnapshotBackupModel.On("AsyncCreate", mock.Anything, mock.Anything).Return(nil, nil)

	_, err := s.m.AsyncCreate(ctx, req)
	exceptedErr := bizerr.New(
		config.ErrCodePitBackupPreCheckClusterVersion,
		config.ErrMsgPitBackupPreCheckClusterVersion,
		mockedCluster.ClusterId, mockedCluster.GetDedicatedCluster().GetVersion(), "")
	s.Equal(exceptedErr.ErrCode(), err.ErrCode())

	m1.Unset()
	m2.Unset()
	m3.Unset()
}

func (s *pitBackupTestSuite) TestAsyncCreatePreCheck_ClusterNotNormal() {
	ctx := context.Background()
	req := &domain.AsyncCreateBackupReq{
		Name: "test", ClusterID: s.mockedCluster.ClusterId,
		ProjectID: s.mockedCluster.ProjectId, UserID: s.mockedUserID,
	}

	pitSwitch := entity.PitBackupSwitch{Status: entity.PitBackupStatusWaitingForEnabling}
	m1 := s.mockPitBackupSwitchRepo.On("List", mock.Anything, mock.Anything).Return([]entity.PitBackupSwitch{
		pitSwitch,
	}, nil)

	mockedCluster := &mgmt.Cluster{
		ClusterId: 1, ProjectId: 1, OrganizationId: 1,
		ClusterInfo: &mgmt.Cluster_DedicatedCluster{
			DedicatedCluster: &dedicatedv1.Cluster{
				State:   dedicatedv1.Cluster_PAUSED,
				Version: "v7.1.0",
			},
		},
	}
	m2 := s.mockMgmtGrpcClient.On("GetAnyDedicatedClusterBasicByID", mock.Anything, mock.Anything).Return(mockedCluster, nil)
	m3 := s.mockSnapshotBackupModel.On("AsyncCreate", mock.Anything, mock.Anything).Return(nil, nil)

	_, err := s.m.AsyncCreate(ctx, req)
	exceptedErr := bizerr.New(
		config.ErrCodePitBackupPreCheckClusterStatus,
		config.ErrMsgPitBackupPreCheckClusterStatus,
		mockedCluster.ClusterId, mockedCluster.GetDedicatedCluster().GetState())
	s.Equal(exceptedErr.ErrCode(), err.ErrCode())

	m1.Unset()
	m2.Unset()
	m3.Unset()
}

func (s *pitBackupTestSuite) TestAsyncCreatePreCheck_In_Enabling() {
	ctx := context.Background()
	req := &domain.AsyncCreateBackupReq{
		Name: "test", ClusterID: s.mockedCluster.ClusterId,
		ProjectID: s.mockedCluster.ProjectId, UserID: s.mockedUserID,
	}

	pitSwitch := entity.PitBackupSwitch{Status: entity.PitBackupStatusWaitingForEnabling}
	m1 := s.mockPitBackupSwitchRepo.On("List", mock.Anything, mock.Anything).Return([]entity.PitBackupSwitch{
		pitSwitch,
	}, nil)

	mockedCluster := &mgmt.Cluster{
		ClusterId: 1, ProjectId: 1, OrganizationId: 1,
		ClusterInfo: &mgmt.Cluster_DedicatedCluster{
			DedicatedCluster: &dedicatedv1.Cluster{
				State:   dedicatedv1.Cluster_AVAILABLE,
				Version: "v7.1.0",
			},
		},
	}
	m2 := s.mockMgmtGrpcClient.On("GetAnyDedicatedClusterBasicByID", mock.Anything, mock.Anything).Return(mockedCluster, nil)
	logBackup := entity.LogBackup{Status: entity.LogBackupStatusScheduled}
	m3 := s.mockLogBackupRepo.On("List", mock.Anything, mock.Anything).Return([]entity.LogBackup{logBackup}, nil)
	m4 := s.mockSnapshotBackupModel.On("AsyncCreate", mock.Anything, mock.Anything).Return(nil, nil)

	_, err := s.m.AsyncCreate(ctx, req)
	exceptedErr := bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, mockedCluster.ClusterId)
	s.Equal(exceptedErr.ErrCode(), err.ErrCode())

	m1.Unset()
	m2.Unset()
	m3.Unset()
	m4.Unset()
}

func (s *pitBackupTestSuite) stringToTSO(str string) string {
	layout := "2006-01-02 15:04:05"
	ts, _ := time.Parse(layout, str)
	return util.TransToTSOStr(ts.Unix())
}

//nolint:funlen
func (s *pitBackupTestSuite) TestListPitrTimeRange_BasicFunc() {
	ctx := context.Background()
	req := &domain.PiTRTimeRangeListReq{
		OrgId:     100,
		ProjectId: 1000,
		ClusterId: 999,
	}
	backup1 := entity.Backup{
		Status:         entity.BackupStatusSuccess,
		BucketID:       100,
		StartTimeStamp: s.stringToTSO("2022-09-25 15:04:05"),
	}
	backup2 := entity.Backup{
		Status:         entity.BackupStatusSuccess,
		BucketID:       200,
		StartTimeStamp: s.stringToTSO("2023-09-25 15:04:05"),
	}
	backup3 := entity.Backup{
		Status:         entity.BackupStatusSuccess,
		BucketID:       100,
		StartTimeStamp: s.stringToTSO("2022-09-10 14:06:07"),
	}
	backup4 := entity.Backup{
		Status:         entity.BackupStatusSuccess,
		BucketID:       100,
		StartTimeStamp: s.stringToTSO("2022-09-01 12:09:10"),
	}
	backup5 := entity.Backup{
		Status:         entity.BackupStatusSuccess,
		BucketID:       400,
		StartTimeStamp: s.stringToTSO("2023-09-25 15:04:05"),
	}
	m1 := s.mockBackupRepo.On("ListExternal", mock.Anything, mock.Anything).
		Return([]entity.Backup{backup1, backup2, backup3, backup4, backup5}, nil)

	// 1, 3, 4 are valid time range
	// 1. 3 should be merged
	logBackup1 := entity.LogBackup{
		Status:       entity.LogBackupStatusScheduled,
		BucketID:     100,
		StartTS:      pointer.To(s.stringToTSO("2022-09-25 15:04:05")),
		CheckpointTS: s.stringToTSO("2022-09-30 18:08:09"),
		Regions:      "eus",
	}
	logBackup2 := entity.LogBackup{
		Status:       entity.LogBackupStatusScheduled,
		BucketID:     300,
		StartTS:      pointer.To(s.stringToTSO("2023-09-30 18:08:09")),
		CheckpointTS: s.stringToTSO("2023-10-30 13:23:56"),
		Regions:      "eus",
	}
	logBackup3 := entity.LogBackup{
		Status:       entity.LogBackupStatusScheduled,
		BucketID:     100,
		StartTS:      pointer.To(s.stringToTSO("2022-09-10 14:06:07")),
		CheckpointTS: s.stringToTSO("2022-09-25 15:04:06"),
		Regions:      "eus",
	}
	logBackup4 := entity.LogBackup{
		Status:       entity.LogBackupStatusScheduled,
		BucketID:     100,
		StartTS:      pointer.To(s.stringToTSO("2022-09-01 12:09:10")),
		CheckpointTS: s.stringToTSO("2022-09-09 13:23:56"),
		Regions:      "eus",
	}
	logBackup5 := entity.LogBackup{
		Status:       entity.LogBackupStatusScheduled,
		BucketID:     400,
		StartTS:      pointer.To(s.stringToTSO("2023-09-25 15:04:05")),
		CheckpointTS: s.stringToTSO("2023-09-30 18:08:09"),
		Regions:      "wus",
		BucketSettingSnapshot: &entity.BucketSetting{
			Buckets: []entity.BackupSettingBucket{
				{Regions: []string{"eus", "wus"}},
			},
		},
	}
	m2 := s.mockLogBackupRepo.On("List", mock.Anything, mock.Anything).
		Return([]entity.LogBackup{logBackup1, logBackup2, logBackup3, logBackup4, logBackup5}, nil)
	mockedCluster := &mgmt.Cluster{
		ClusterId: 1, ProjectId: 1, OrganizationId: 1,
		CloudProvider: constants.CloudProviderAWS,
		Region:        "eus",
		ClusterInfo: &mgmt.Cluster_DedicatedCluster{
			DedicatedCluster: &dedicatedv1.Cluster{
				State:             dedicatedv1.Cluster_AVAILABLE,
				Version:           "v7.1.0",
				RegionDisplayName: "eus",
			},
		},
	}
	m3 := s.mockMgmtGrpcClient.On("GetAnyDedicatedClusterBasicByID", mock.Anything, mock.Anything).
		Return(mockedCluster, nil)
	regions := map[string]*core.CloudRegion{
		"eus": {Name: "eus", DisplayName: "eus"},
		"wus": {Name: "wus", DisplayName: "wus"},
	}
	cloudRegions := map[string]map[string]*core.CloudRegion{
		constants.CloudProviderAWS: regions, constants.CloudProviderGCP: nil,
	}
	m4 := s.mockCoreGrpcClient.On("ListCloudRegions", mock.Anything, mock.Anything).
		Return(cloudRegions, nil)

	// Mock bucketRepo.Load to return not-found bizerr for CRPiTR checkpoint fallback
	notFoundErr := bizerr_wrapper.NewInternalErr("bucket not found")
	m5 := s.mockBucketRepo.On("Load", mock.Anything, mock.Anything).
		Return(nil, notFoundErr)

	resp, err := s.m.ListPiTRTimeRanges(ctx, req)
	s.Equal(nil, err)
	s.Equal(2, len(resp.Items))
	s.Equal(3, len(resp.Items[0].TimeRanges))
	s.Equal(int64(1662818768), resp.Items[0].TimeRanges[0].StartTS.Seconds)
	s.Equal(int64(1664561289), resp.Items[0].TimeRanges[0].EndTS.Seconds)

	m1.Unset()
	m2.Unset()
	m3.Unset()
	m4.Unset()
	m5.Unset()
}

func (s *pitBackupTestSuite) TestListPitrTimeRange_NoLogBackupTip() {
	ctx := context.Background()
	req := &domain.PiTRTimeRangeListReq{
		OrgId:     100,
		ProjectId: 1000,
		ClusterId: 999,
	}
	backup1 := entity.Backup{
		Status:         entity.BackupStatusSuccess,
		BucketID:       100,
		StartTimeStamp: s.stringToTSO("2022-09-25 15:04:05"),
	}
	m1 := s.mockBackupRepo.On("ListExternal", mock.Anything, mock.Anything).Return([]entity.Backup{backup1}, nil)

	m2 := s.mockLogBackupRepo.On("List", mock.Anything, mock.Anything).Return([]entity.LogBackup{}, nil)
	mockedCluster := &mgmt.Cluster{
		ClusterId: 1, ProjectId: 1, OrganizationId: 1,
		CloudProvider: constants.CloudProviderAWS,
		Region:        "eus",
		ClusterInfo: &mgmt.Cluster_DedicatedCluster{
			DedicatedCluster: &dedicatedv1.Cluster{
				State:             dedicatedv1.Cluster_AVAILABLE,
				Version:           "v7.1.0",
				RegionDisplayName: "eus",
			},
		},
	}
	m3 := s.mockMgmtGrpcClient.On("GetAnyDedicatedClusterBasicByID", mock.Anything, mock.Anything).
		Return(mockedCluster, nil)
	regions := map[string]*core.CloudRegion{
		"eus": {Name: "eus", DisplayName: "eus"},
		"wus": {Name: "wus", DisplayName: "wus"},
	}
	cloudRegions := map[string]map[string]*core.CloudRegion{
		constants.CloudProviderAWS: regions, constants.CloudProviderGCP: nil,
	}
	m4 := s.mockCoreGrpcClient.On("ListCloudRegions", mock.Anything, mock.Anything).
		Return(cloudRegions, nil)

	resp, err := s.m.ListPiTRTimeRanges(ctx, req)
	s.Equal(nil, err)
	s.Equal("PiTR is off, you can turn it on in Backup Setting.", resp.Msg)

	m1.Unset()
	m2.Unset()
	m3.Unset()
	m4.Unset()
}

//nolint:funlen
func (s *pitBackupTestSuite) TestListPitrTimeRange_CRPiTR_CheckpointFound() {
	ctx := context.Background()
	req := &domain.PiTRTimeRangeListReq{
		OrgId:     100,
		ProjectId: 1000,
		ClusterId: 999,
	}

	// Enable the V2 PiTR time range algorithm for this test.
	internal_config.Global().BackupConfig.CRRFOverrides = internal_config.CRRFOverrides{
		{Gate: internal_config.FeatureGate{All: true}, CalculatePiTRTimeRangeV2: true},
	}
	defer func() { internal_config.Global().BackupConfig.CRRFOverrides = nil }()

	// Setup: latest logbackup has 2 regions (main + 1 secondary)
	// Add a corresponding snapshot backup
	backup1 := entity.Backup{
		Status:         entity.BackupStatusSuccess,
		BucketID:       100,
		StartTimeStamp: s.stringToTSO("2023-09-25 15:04:05"),
	}

	logBackup1 := entity.LogBackup{
		Status:       entity.LogBackupStatusScheduled,
		BucketID:     100,
		StartTS:      pointer.To(s.stringToTSO("2023-09-25 15:04:05")),
		CheckpointTS: s.stringToTSO("2023-09-30 18:08:09"),
		Regions:      "eus",
		Storage:      "s3://test-bucket/logbackup1",
		BucketSettingSnapshot: &entity.BucketSetting{
			Buckets: []entity.BackupSettingBucket{
				{Regions: []string{"eus"}, Master: true},
				{Regions: []string{"wus"}, Master: false, BucketId: 1},
			},
		},
	}

	m1 := s.mockBackupRepo.On("ListExternal", mock.Anything, mock.Anything).Return([]entity.Backup{backup1}, nil)
	m2 := s.mockLogBackupRepo.On("List", mock.Anything, mock.Anything).Return([]entity.LogBackup{logBackup1}, nil)

	mockedCluster := &mgmt.Cluster{
		ClusterId: 1, ProjectId: 1, OrganizationId: 1,
		CloudProvider: constants.CloudProviderAWS,
		Region:        "eus",
		ClusterInfo: &mgmt.Cluster_DedicatedCluster{
			DedicatedCluster: &dedicatedv1.Cluster{
				State:             dedicatedv1.Cluster_AVAILABLE,
				Version:           "v7.1.0",
				RegionDisplayName: "eus",
			},
		},
	}
	m3 := s.mockMgmtGrpcClient.On("GetAnyDedicatedClusterBasicByID", mock.Anything, mock.Anything).Return(mockedCluster, nil)

	regions := map[string]*core.CloudRegion{
		"eus": {Name: "eus", DisplayName: "eus"},
		"wus": {Name: "wus", DisplayName: "wus"},
	}
	cloudRegions := map[string]map[string]*core.CloudRegion{
		constants.CloudProviderAWS: regions, constants.CloudProviderGCP: nil,
	}
	m4 := s.mockCoreGrpcClient.On("ListCloudRegions", mock.Anything, mock.Anything).Return(cloudRegions, nil)

	// Mock bucket repo to return a valid AWS replica bucket for the secondary region.
	testBucket := &entity.Bucket{
		InfraProviderType: entity.InfraProviderTypeAWS,
		Provider:          string(entity.CloudProviderTypeAWS),
		Name:              "test-replica-bucket",
		CloudAccountID:    pointer.ToString("123456789012"),
		ProjectID:         "1000",
	}
	m5 := s.mockBucketRepo.On("Load", mock.Anything, mock.Anything).Return(testBucket, nil)

	// Mock storage func to return checkpoint from resume-state.json.
	// checkpointTSO = TransToTSO(1695895200) = 2023-09-28 12:00:00 UTC
	// (within the log backup range: 2023-09-25 -> 2023-09-30)
	checkpointTSO := uint64(444568751308800000)
	resumeStateJSON := []byte(`{"last_checkpoint": 444568751308800000}`)

	mockStorage := blob_mock.NewStorage(s.T())
	mockStorage.On("ReadFile", mock.Anything, br.CRPiTRResumeStateFile).Return(resumeStateJSON, nil)

	mockGetStorageFunc := func(ctx context.Context, uri string, opt *blob.StorageOption) (blob.Storage, error) {
		return mockStorage, nil
	}

	// Recreate the model with the mock storage func
	trans := db.NewProvider(s.db)
	m := impl.NewPitBackupModelWithStorageFunc(
		trans, s.mockPitBackupSwitchRepo, s.mockLogBackupRepo, s.mockBackupRepo, s.mockRestoreRepo,
		s.mockWorkflow, s.mockCoreGrpcClient, s.mockMgmtGrpcClient, s.mockSnapshotBackupModel, s.mockBackupSettingM,
		s.mockBackupScheduleRepo, s.mockFeatureGate, s.mockBucketRepo, mockGetStorageFunc)

	resp, err := m.ListPiTRTimeRanges(ctx, req)
	s.NoError(err)
	s.Equal(2, len(resp.Items), "should have 2 regions")

	// Find the wus region (non-main region)
	var wusRegion *domain.PiTRRegionTimeRange
	for _, item := range resp.Items {
		if item.Region.Name == "wus" {
			wusRegion = item
			break
		}
	}
	s.NotNil(wusRegion, "wus region should exist")
	s.Equal(1, len(wusRegion.TimeRanges), "wus should have 1 time range")

	// Assert: secondary region's end = checkpoint from resume-state.json
	expectedEndTS := util.TransToTS(checkpointTSO)
	s.Equal(expectedEndTS, wusRegion.TimeRanges[0].EndTS.Seconds, "wus endTS should match checkpoint")

	m1.Unset()
	m2.Unset()
	m3.Unset()
	m4.Unset()
	m5.Unset()
}

func (s *pitBackupTestSuite) TestListPitrTimeRange_CRPiTR_FileNotFound() {
	ctx := context.Background()
	req := &domain.PiTRTimeRangeListReq{
		OrgId:     100,
		ProjectId: 1000,
		ClusterId: 999,
	}

	// Enable the V2 PiTR time range algorithm for this test.
	internal_config.Global().BackupConfig.CRRFOverrides = internal_config.CRRFOverrides{
		{Gate: internal_config.FeatureGate{All: true}, CalculatePiTRTimeRangeV2: true},
	}
	defer func() { internal_config.Global().BackupConfig.CRRFOverrides = nil }()

	// Setup: snapshot backup + latest logbackup has 2 regions
	snapshotBackup1 := entity.Backup{
		Status:         entity.BackupStatusSuccess,
		BucketID:       100,
		StartTimeStamp: s.stringToTSO("2023-09-25 15:04:05"),
	}
	logBackup1 := entity.LogBackup{
		Status:       entity.LogBackupStatusScheduled,
		BucketID:     100,
		StartTS:      pointer.To(s.stringToTSO("2023-09-25 15:04:05")),
		CheckpointTS: s.stringToTSO("2023-09-30 18:08:09"),
		Regions:      "eus",
		Storage:      "s3://test-bucket/logbackup1",
		BucketSettingSnapshot: &entity.BucketSetting{
			Buckets: []entity.BackupSettingBucket{
				{Regions: []string{"eus"}, Master: true},
				{Regions: []string{"wus"}, Master: false, BucketId: 1},
			},
		},
	}

	m1 := s.mockBackupRepo.On("ListExternal", mock.Anything, mock.Anything).Return([]entity.Backup{snapshotBackup1}, nil)
	m2 := s.mockLogBackupRepo.On("List", mock.Anything, mock.Anything).Return([]entity.LogBackup{logBackup1}, nil)

	mockedCluster := &mgmt.Cluster{
		ClusterId: 1, ProjectId: 1, OrganizationId: 1,
		CloudProvider: constants.CloudProviderAWS,
		Region:        "eus",
		ClusterInfo: &mgmt.Cluster_DedicatedCluster{
			DedicatedCluster: &dedicatedv1.Cluster{
				State:             dedicatedv1.Cluster_AVAILABLE,
				Version:           "v7.1.0",
				RegionDisplayName: "eus",
			},
		},
	}
	m3 := s.mockMgmtGrpcClient.On("GetAnyDedicatedClusterBasicByID", mock.Anything, mock.Anything).Return(mockedCluster, nil)

	regions := map[string]*core.CloudRegion{
		"eus": {Name: "eus", DisplayName: "eus"},
		"wus": {Name: "wus", DisplayName: "wus"},
	}
	cloudRegions := map[string]map[string]*core.CloudRegion{
		constants.CloudProviderAWS: regions, constants.CloudProviderGCP: nil,
	}
	m4 := s.mockCoreGrpcClient.On("ListCloudRegions", mock.Anything, mock.Anything).Return(cloudRegions, nil)

	// Mock bucket repo to return a valid AWS replica bucket for the secondary region.
	testBucket := &entity.Bucket{
		InfraProviderType: entity.InfraProviderTypeAWS,
		Provider:          string(entity.CloudProviderTypeAWS),
		Name:              "test-replica-bucket",
		CloudAccountID:    pointer.ToString("123456789012"),
		ProjectID:         "1000",
	}
	m5 := s.mockBucketRepo.On("Load", mock.Anything, mock.Anything).Return(testBucket, nil)

	// Mock storage to return not found error
	mockStorage := blob_mock.NewStorage(s.T())
	mockStorage.On("ReadFile", mock.Anything, br.CRPiTRResumeStateFile).Return(nil, errors.New("Not Found"))

	mockGetStorageFunc := func(ctx context.Context, uri string, opt *blob.StorageOption) (blob.Storage, error) {
		return mockStorage, nil
	}

	// Recreate the model with the mock storage func
	trans := db.NewProvider(s.db)
	m := impl.NewPitBackupModelWithStorageFunc(
		trans, s.mockPitBackupSwitchRepo, s.mockLogBackupRepo, s.mockBackupRepo, s.mockRestoreRepo,
		s.mockWorkflow, s.mockCoreGrpcClient, s.mockMgmtGrpcClient, s.mockSnapshotBackupModel, s.mockBackupSettingM,
		s.mockBackupScheduleRepo, s.mockFeatureGate, s.mockBucketRepo, mockGetStorageFunc)

	resp, err := m.ListPiTRTimeRanges(ctx, req)
	s.NoError(err)
	s.Equal(2, len(resp.Items), "should have 2 regions")

	// Find the wus region (non-main region)
	var wusRegion *domain.PiTRRegionTimeRange
	var eusRegion *domain.PiTRRegionTimeRange
	for _, item := range resp.Items {
		if item.Region.Name == "wus" {
			wusRegion = item
		}
		if item.Region.Name == "eus" {
			eusRegion = item
		}
	}
	s.NotNil(wusRegion, "wus region should exist")
	s.NotNil(eusRegion, "eus region should exist")

	// Assert: both regions should have same endTS (fallback to main region checkpoint)
	s.Equal(eusRegion.TimeRanges[0].EndTS.Seconds, wusRegion.TimeRanges[0].EndTS.Seconds,
		"wus should fallback to main region endTS")

	m1.Unset()
	m2.Unset()
	m3.Unset()
	m4.Unset()
	m5.Unset()
}

func (s *pitBackupTestSuite) TestListPitrTimeRange_CRPiTR_BucketLoadFailed() {
	ctx := context.Background()
	req := &domain.PiTRTimeRangeListReq{
		OrgId:     100,
		ProjectId: 1000,
		ClusterId: 999,
	}

	// Enable the V2 PiTR time range algorithm for this test.
	internal_config.Global().BackupConfig.CRRFOverrides = internal_config.CRRFOverrides{
		{Gate: internal_config.FeatureGate{All: true}, CalculatePiTRTimeRangeV2: true},
	}
	defer func() { internal_config.Global().BackupConfig.CRRFOverrides = nil }()

	// Setup: snapshot backup + latest logbackup has 2 regions
	snapshotBackup1 := entity.Backup{
		Status:         entity.BackupStatusSuccess,
		BucketID:       100,
		StartTimeStamp: s.stringToTSO("2023-09-25 15:04:05"),
	}
	logBackup1 := entity.LogBackup{
		Status:       entity.LogBackupStatusScheduled,
		BucketID:     100,
		StartTS:      pointer.To(s.stringToTSO("2023-09-25 15:04:05")),
		CheckpointTS: s.stringToTSO("2023-09-30 18:08:09"),
		Regions:      "eus",
		Storage:      "s3://test-bucket/logbackup1",
		BucketSettingSnapshot: &entity.BucketSetting{
			Buckets: []entity.BackupSettingBucket{
				{Regions: []string{"eus"}, Master: true},
				{Regions: []string{"wus"}, Master: false, BucketId: 1},
			},
		},
	}

	m1 := s.mockBackupRepo.On("ListExternal", mock.Anything, mock.Anything).Return([]entity.Backup{snapshotBackup1}, nil)
	m2 := s.mockLogBackupRepo.On("List", mock.Anything, mock.Anything).Return([]entity.LogBackup{logBackup1}, nil)

	mockedCluster := &mgmt.Cluster{
		ClusterId: 1, ProjectId: 1, OrganizationId: 1,
		CloudProvider: constants.CloudProviderAWS,
		Region:        "eus",
		ClusterInfo: &mgmt.Cluster_DedicatedCluster{
			DedicatedCluster: &dedicatedv1.Cluster{
				State:             dedicatedv1.Cluster_AVAILABLE,
				Version:           "v7.1.0",
				RegionDisplayName: "eus",
			},
		},
	}
	m3 := s.mockMgmtGrpcClient.On("GetAnyDedicatedClusterBasicByID", mock.Anything, mock.Anything).Return(mockedCluster, nil)

	regions := map[string]*core.CloudRegion{
		"eus": {Name: "eus", DisplayName: "eus"},
		"wus": {Name: "wus", DisplayName: "wus"},
	}
	cloudRegions := map[string]map[string]*core.CloudRegion{
		constants.CloudProviderAWS: regions, constants.CloudProviderGCP: nil,
	}
	m4 := s.mockCoreGrpcClient.On("ListCloudRegions", mock.Anything, mock.Anything).Return(cloudRegions, nil)

	// Mock bucket repo to return error
	mockErr := bizerr_wrapper.NewInternalErr("internal error")
	m5 := s.mockBucketRepo.On("Load", mock.Anything, mock.Anything).Return(nil, mockErr)

	// Recreate the model with nil storage func (won't be called due to bucket load failure)
	trans := db.NewProvider(s.db)
	m := impl.NewPitBackupModelWithStorageFunc(
		trans, s.mockPitBackupSwitchRepo, s.mockLogBackupRepo, s.mockBackupRepo, s.mockRestoreRepo,
		s.mockWorkflow, s.mockCoreGrpcClient, s.mockMgmtGrpcClient, s.mockSnapshotBackupModel, s.mockBackupSettingM,
		s.mockBackupScheduleRepo, s.mockFeatureGate, s.mockBucketRepo, nil)

	resp, err := m.ListPiTRTimeRanges(ctx, req)
	s.NoError(err)
	s.Equal(2, len(resp.Items), "should have 2 regions")

	// Find the wus region (non-main region)
	var wusRegion *domain.PiTRRegionTimeRange
	var eusRegion *domain.PiTRRegionTimeRange
	for _, item := range resp.Items {
		if item.Region.Name == "wus" {
			wusRegion = item
		}
		if item.Region.Name == "eus" {
			eusRegion = item
		}
	}
	s.NotNil(wusRegion, "wus region should exist")
	s.NotNil(eusRegion, "eus region should exist")

	// Assert: both regions should have same endTS (fallback to main region checkpoint)
	s.Equal(eusRegion.TimeRanges[0].EndTS.Seconds, wusRegion.TimeRanges[0].EndTS.Seconds,
		"wus should fallback to main region endTS when bucket load fails")

	m1.Unset()
	m2.Unset()
	m3.Unset()
	m4.Unset()
	m5.Unset()
}