package compact_test

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

	"github.com/AlekSi/pointer"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	"github.com/stretchr/testify/suite"
	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_mock "github.com/tidbcloud/dataflow-service/app/models/backup/mock"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"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"
	repo_mock "github.com/tidbcloud/dataflow-service/app/repo/mock"
	"github.com/tidbcloud/dataflow-service/app/utils"
	service_config "github.com/tidbcloud/dataflow-service/internal/config"
	grpcli_mock "github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli/mock"
	infraapi "github.com/tidbcloud/dataflow-service/pkg/infra-api"
	infra_mock "github.com/tidbcloud/dataflow-service/pkg/infra-api/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"
)

const (
	TestCompactCycle = 5*time.Hour + 50*time.Minute

	TestTenantID    = uint64(1)
	TestProjectID   = uint64(2)
	TestClusterID   = uint64(300)
	TestLogBackupID = uint64(200)

	InvalidTSO = "invalid_tso"
)

// Helper functions for dynamic TSO generation based on current time
func getValidStartTSO() string {
	return fmt.Sprintf("%d", utils.GoTimeToTSO(time.Now().Add(-12*time.Hour)))
}

func getCurrentTSO() string {
	return fmt.Sprintf("%d", utils.GoTimeToTSO(time.Now()))
}

type compactBackupSchedulerTestSuite struct {
	suite.Suite

	scheduler backup.CompactBackupScheduler

	mockLogBackupRepo      *repo_mock.LogBackupRepo
	mockCompactBackupRepo  *repo_mock.CompactBackupRepo
	mockCompactBackupModel *backup_mock.CompactBackupModel
	mockCompactFacade      *infra_mock.CompactFacade
	mockMgmtGrpcClient     *grpcli_mock.MgmtSvcGrpcClient
	mockFeatureGate        *feature_gate_mock.FeatureGate

	db            *gorm.DB
	transProvider db.TransProvider
}

func TestCompactBackupSchedulerTestSuite(t *testing.T) {
	suite.Run(t, new(compactBackupSchedulerTestSuite))
}

func (s *compactBackupSchedulerTestSuite) SetupSuite() {
	// Initialize global config with proper CompactConfig
	globalConfig := service_config.Global()
	globalConfig.CompactConfig.CompactBackupCycle = "5h50m"

	// Setup database once
	testDB, err := repo.GetFakeDB()
	require.NoError(s.T(), err)
	s.db = testDB
	s.transProvider = db.NewProvider(testDB)
}

func (s *compactBackupSchedulerTestSuite) SetupTest() {
	// Create fresh mocks for each test to ensure isolation
	s.mockLogBackupRepo = repo_mock.NewLogBackupRepo(s.T())
	s.mockCompactBackupRepo = repo_mock.NewCompactBackupRepo(s.T())
	s.mockCompactBackupModel = backup_mock.NewCompactBackupModel(s.T())
	s.mockCompactFacade = infra_mock.NewCompactFacade(s.T())
	s.mockMgmtGrpcClient = grpcli_mock.NewMgmtSvcGrpcClient(s.T())
	s.mockFeatureGate = feature_gate_mock.NewFeatureGate(s.T())

	// Create scheduler instance with fresh mocks
	s.scheduler = impl.NewCompactBackupScheduler(
		s.transProvider,
		s.mockLogBackupRepo,
		s.mockCompactBackupRepo,
		s.mockCompactBackupModel,
		s.mockCompactFacade,
		s.mockMgmtGrpcClient,
		s.mockFeatureGate,
	)
}

func createValidLogBackup() *entity.LogBackup {
	return &entity.LogBackup{
		Updatable: entity.Updatable{
			Base: entity.Base{ID: entity.ID(TestLogBackupID)},
		},
		ClusterID:    entity.ID(TestClusterID),
		TenantID:     entity.ID(TestTenantID),
		ProjectID:    entity.ID(TestProjectID),
		Status:       entity.LogBackupStatusRunning,
		StartTS:      pointer.ToString(getValidStartTSO()),
		CheckpointTS: getCurrentTSO(),
		Storage:      "s3://test-bucket/log-backup-path",
	}
}

func createValidCluster() *mgmt.Cluster {
	return &mgmt.Cluster{
		ClusterId:      TestClusterID,
		OrganizationId: TestTenantID,
		ProjectId:      TestProjectID,
		Region:         "us-west-2",
		CloudProvider:  "AWS",
		ClusterInfo: &mgmt.Cluster_DedicatedCluster{
			DedicatedCluster: &dedicatedv1.Cluster{
				Version: "v7.1.0",
				State:   dedicatedv1.Cluster_AVAILABLE,
			},
		},
	}
}

// Step 2: shouldCompactLogBackup tests - via CompactLogBackup integration testing
func (s *compactBackupSchedulerTestSuite) TestCompactLogBackupBehaviors() {
	testCases := []struct {
		name                    string
		logBackup               *entity.LogBackup
		mockLastSuccessTask     *entity.CompactBackup
		shouldCreateCompactTask bool
		description             string
	}{
		{
			name:                    "should_not_compact_status_paused",
			logBackup:               s.createLogBackupWithStatus(entity.LogBackupStatusPaused),
			shouldCreateCompactTask: false,
			description:             "LogBackup is Paused, should not create compact task",
		},
		{
			name:                    "should_not_compact_empty_checkpoint",
			logBackup:               s.createLogBackupWithEmptyCheckpoint(),
			shouldCreateCompactTask: false,
			description:             "CheckpointTS is empty, should not create compact task",
		},
		{
			name:                    "should_not_compact_runtime_insufficient",
			logBackup:               s.createLogBackupRunning2Hours(),
			shouldCreateCompactTask: false,
			description:             "Run time is insufficient, should not create compact task",
		},
		{
			name:                    "should_not_compact_cycle_not_reached",
			logBackup:               s.createLogBackupRunning7Hours(),
			mockLastSuccessTask:     s.createSuccessTaskFromHoursAgo(3), // Success 3 hours ago
			shouldCreateCompactTask: false,
			description:             "Previous success 3 hours ago, not enough distance, should not create compact task",
		},
		{
			name:                    "should_compact_conditions_met_no_previous_task",
			logBackup:               s.createLogBackupRunning7Hours(),
			mockLastSuccessTask:     nil, // No previous successful tasks
			shouldCreateCompactTask: true,
			description:             "LogBackup running 8 hours, no previous tasks, should reach createCompactTask",
		},
		{
			name:                    "should_compact_cycle_reached",
			logBackup:               s.createLogBackupRunning7Hours(),
			mockLastSuccessTask:     s.createSuccessTaskFromHoursAgo(12), // Success 12 hours ago, well beyond cycle
			shouldCreateCompactTask: true,
			description:             "Previous success 12 hours ago, beyond 5h50m cycle, should reach createCompactTask",
		},
	}

	for _, tc := range testCases {
		testCase := tc
		s.Run(testCase.name, func() {
			s.SetupTest()
			s.runSingleTestCase(testCase)
		})
	}
}

func (s *compactBackupSchedulerTestSuite) runSingleTestCase(tc struct {
	name                    string
	logBackup               *entity.LogBackup
	mockLastSuccessTask     *entity.CompactBackup
	shouldCreateCompactTask bool
	description             string
}) {
	// Set up ALL mock expectations in the right order based on the actual call sequence

	// 1. Mock getRunningLogBackups to return our test LogBackup
	s.mockLogBackupRepo.On("List", mock.Anything, mock.Anything).
		Return([]entity.LogBackup{*tc.logBackup}, nil).Once()

	// 2. Always mock feature check (all cases call this) - Use generous count
	s.mockMgmtGrpcClient.On("GetDedicatedClusterBasicByID", mock.Anything, TestClusterID).
		Return(createValidCluster(), nil).Maybe()
	s.mockFeatureGate.On("IsFeatureEnabled", mock.Anything, mock.Anything).
		Return(true).Maybe()

	// 3. Mock GetAnyClusterByID for getCompactCycle if checkpoint_ts is not empty
	if tc.logBackup.CheckpointTS != "" {
		s.mockMgmtGrpcClient.On("GetAnyDedicatedClusterBasicByID", mock.Anything, TestClusterID).
			Return(createValidCluster(), nil).Maybe()
	}

	// 4. Set up mock for ListByFilter
	s.mockCompactBackupRepo.On("ListByFilter", mock.Anything, mock.MatchedBy(func(filter *backup_repo.CompactBackupFilter) bool {
		if filter == nil || filter.LogBackupID == nil || filter.Status == nil {
			return false
		}

		logBackupID := *filter.LogBackupID

		// Only match calls for our test LogBackupID
		if logBackupID != TestLogBackupID {
			return false
		}

		// Match all status queries for this LogBackupID
		return true
	})).Return(func(ctx context.Context, filter *backup_repo.CompactBackupFilter) []*entity.CompactBackup {
		// Dynamic return based on the status being queried
		status := *filter.Status

		switch status {
		case entity.CompactBackupStatusSuccess:
			// Return the success task from parameter
			if tc.mockLastSuccessTask != nil {
				return []*entity.CompactBackup{tc.mockLastSuccessTask}
			}
			return []*entity.CompactBackup{}
		case entity.CompactBackupStatusRunning, entity.CompactBackupStatusPending:
			// Always return empty for running/pending (no existing running tasks)
			return []*entity.CompactBackup{}
		default:
			return []*entity.CompactBackup{}
		}
	}, nil).Maybe()

	// 5. Set up CompactBackupModel mock expectations
	if tc.shouldCreateCompactTask {
		// Expect AsyncCreate to be called
		s.mockCompactBackupModel.On("AsyncCreate", mock.Anything,
			mock.MatchedBy(func(req *domain.AsyncCreateCompactBackupReq) bool {
				// Verify request parameters
				return req.ClusterID == TestClusterID &&
					req.LogBackupID == TestLogBackupID &&
					req.IsInternal == true &&
					req.UserID == 0 && // Scheduler sets this to 0
					req.ProjectID == 0 // Scheduler sets this to 0
			})).Return(&domain.AsyncCreateCompactBackupResp{
			CompactBackupID: 999,
			WorkflowID:      "test-workflow-id",
		}, nil).Once()
	}

	// Execute CompactLogBackup
	err := s.scheduler.CompactLogBackup(context.Background())

	// Verify results
	if tc.shouldCreateCompactTask {
		s.NoError(err, "Should create compact task successfully for case: %s", tc.name)
	} else {
		s.NoError(err, "Should complete without creating task for case: %s", tc.name)
	}
}

// Test data helper functions
func (s *compactBackupSchedulerTestSuite) createLogBackupRunning7Hours() *entity.LogBackup {
	logBackup := createValidLogBackup()
	// Set StartTS to 12 hours ago (well beyond the 5h50m cycle requirement)
	startTime := time.Now().Add(-12 * time.Hour)
	logBackup.StartTS = pointer.ToString(s.timeToTSOString(startTime))
	return logBackup
}

func (s *compactBackupSchedulerTestSuite) createLogBackupRunning2Hours() *entity.LogBackup {
	logBackup := createValidLogBackup()
	// Set StartTS to 2 hours ago (insufficient runtime)
	startTime := time.Now().Add(-2 * time.Hour)
	logBackup.StartTS = pointer.ToString(s.timeToTSOString(startTime))
	return logBackup
}

func (s *compactBackupSchedulerTestSuite) createLogBackupWithStatus(status entity.LogBackupStatus) *entity.LogBackup {
	logBackup := createValidLogBackup()
	logBackup.Status = status
	return logBackup
}

func (s *compactBackupSchedulerTestSuite) createLogBackupWithEmptyCheckpoint() *entity.LogBackup {
	logBackup := createValidLogBackup()
	logBackup.CheckpointTS = ""
	return logBackup
}

func (s *compactBackupSchedulerTestSuite) createSuccessTaskFromHoursAgo(hours int) *entity.CompactBackup {
	return &entity.CompactBackup{
		Updatable: entity.Updatable{
			Base: entity.Base{
				ID:        entity.ID(999),
				CreatedAt: time.Now().Add(-time.Duration(hours) * time.Hour),
			},
		},
		LogBackupID: entity.ID(TestLogBackupID),
		ClusterID:   entity.ID(TestClusterID),
		Status:      entity.CompactBackupStatusSuccess,
	}
}

func (s *compactBackupSchedulerTestSuite) timeToTSOString(t time.Time) string {
	// Use the same TSO conversion logic as the production code
	tso := utils.GoTimeToTSO(t)
	return fmt.Sprintf("%d", tso)
}

// Step 3: hasRunningCompactTask tests - test scenarios where existing tasks prevent new task creation
func (s *compactBackupSchedulerTestSuite) TestCompactLogBackupWithRunningTasks() {
	testCases := []struct {
		name                    string
		existingLocalTasks      []*entity.CompactBackup
		downstreamResponses     map[string]*infraapi.CompactBackupResponse
		shouldCreateCompactTask bool
		description             string
	}{
		{
			name: "should_not_compact_has_running_task",
			existingLocalTasks: []*entity.CompactBackup{
				{
					Updatable: entity.Updatable{
						Base: entity.Base{ID: entity.ID(800)},
					},
					LogBackupID:         entity.ID(TestLogBackupID),
					ClusterID:           entity.ID(TestClusterID),
					Status:              entity.CompactBackupStatusRunning,
					DownstreamCompactID: pointer.ToString("downstream-800"),
				},
			},
			downstreamResponses: map[string]*infraapi.CompactBackupResponse{
				"downstream-800": {
					Status: infraapi.CompactBackupStatusRunning,
				},
			},
			shouldCreateCompactTask: false,
			description:             "Has running compact task, should not create new task",
		},
		{
			name: "should_not_compact_has_pending_task",
			existingLocalTasks: []*entity.CompactBackup{
				{
					Updatable: entity.Updatable{
						Base: entity.Base{ID: entity.ID(801)},
					},
					LogBackupID:         entity.ID(TestLogBackupID),
					ClusterID:           entity.ID(TestClusterID),
					Status:              entity.CompactBackupStatusPending,
					DownstreamCompactID: pointer.ToString("downstream-801"),
				},
			},
			downstreamResponses: map[string]*infraapi.CompactBackupResponse{
				"downstream-801": {
					Status: infraapi.CompactBackupStatusPending,
				},
			},
			shouldCreateCompactTask: false,
			description:             "Has pending compact task, should not create new task",
		},
		{
			name: "should_compact_local_task_downstream_completed",
			existingLocalTasks: []*entity.CompactBackup{
				{
					Updatable: entity.Updatable{
						Base: entity.Base{ID: entity.ID(802)},
					},
					LogBackupID:         entity.ID(TestLogBackupID),
					ClusterID:           entity.ID(TestClusterID),
					Status:              entity.CompactBackupStatusRunning, // Local shows running
					DownstreamCompactID: pointer.ToString("downstream-802"),
				},
			},
			downstreamResponses: map[string]*infraapi.CompactBackupResponse{
				"downstream-802": {
					Status: infraapi.CompactBackupStatusComplete, // But downstream is complete
				},
			},
			shouldCreateCompactTask: true,
			description:             "Local task shows running but downstream completed, should create new task",
		},
		{
			name: "should_not_compact_task_without_downstream_id",
			existingLocalTasks: []*entity.CompactBackup{
				{
					Updatable: entity.Updatable{
						Base: entity.Base{ID: entity.ID(803)},
					},
					LogBackupID:         entity.ID(TestLogBackupID),
					ClusterID:           entity.ID(TestClusterID),
					Status:              entity.CompactBackupStatusRunning,
					DownstreamCompactID: nil, // No downstream ID
				},
			},
			downstreamResponses:     map[string]*infraapi.CompactBackupResponse{},
			shouldCreateCompactTask: true,
			description:             "Task without downstream ID gets marked failed, should create new task",
		},
	}

	for _, tc := range testCases {
		testCase := tc
		s.Run(testCase.name, func() {
			s.SetupTest()
			s.runRunningTaskTestCase(testCase)
		})
	}
}

// Helper function to run test cases with existing running tasks
func (s *compactBackupSchedulerTestSuite) runRunningTaskTestCase(tc struct {
	name                    string
	existingLocalTasks      []*entity.CompactBackup
	downstreamResponses     map[string]*infraapi.CompactBackupResponse
	shouldCreateCompactTask bool
	description             string
}) {
	// Create a valid log backup that should normally trigger compact
	logBackup := s.createLogBackupRunning7Hours() // 12 hours runtime, sufficient

	// 1. Mock getRunningLogBackups to return our test LogBackup
	s.mockLogBackupRepo.On("List", mock.Anything, mock.Anything).
		Return([]entity.LogBackup{*logBackup}, nil).Once()

	// 2. Mock feature gate and cluster checks (all should pass)
	s.mockMgmtGrpcClient.On("GetDedicatedClusterBasicByID", mock.Anything, TestClusterID).
		Return(createValidCluster(), nil).Maybe()
	s.mockFeatureGate.On("IsFeatureEnabled", mock.Anything, mock.Anything).
		Return(true).Maybe()

	// 3. Mock ShouldCompactLogBackup checks (should return true)
	s.mockCompactBackupRepo.On("ListByFilter", mock.Anything, mock.MatchedBy(func(filter *backup_repo.CompactBackupFilter) bool {
		return filter != nil && filter.LogBackupID != nil && *filter.LogBackupID == TestLogBackupID &&
			filter.Status != nil && *filter.Status == entity.CompactBackupStatusSuccess
	})).Return([]*entity.CompactBackup{}, nil).Maybe() // No previous success tasks

	// Mock GetAnyClusterByID for shouldCompactLogBackup
	s.mockMgmtGrpcClient.On("GetAnyDedicatedClusterBasicByID", mock.Anything, TestClusterID).
		Return(createValidCluster(), nil).Maybe()

	// 4. Mock getLocalRunningTasks - return existing tasks based on status
	runningTasks := []*entity.CompactBackup{}
	pendingTasks := []*entity.CompactBackup{}

	for _, task := range tc.existingLocalTasks {
		switch task.Status {
		case entity.CompactBackupStatusRunning:
			runningTasks = append(runningTasks, task)
		case entity.CompactBackupStatusPending:
			pendingTasks = append(pendingTasks, task)
		}
	}

	// Mock running tasks query
	s.mockCompactBackupRepo.On("ListByFilter", mock.Anything, mock.MatchedBy(func(filter *backup_repo.CompactBackupFilter) bool {
		return filter != nil && filter.LogBackupID != nil && *filter.LogBackupID == TestLogBackupID &&
			filter.Status != nil && *filter.Status == entity.CompactBackupStatusRunning
	})).Return(runningTasks, nil).Maybe()

	// Mock pending tasks query
	s.mockCompactBackupRepo.On("ListByFilter", mock.Anything, mock.MatchedBy(func(filter *backup_repo.CompactBackupFilter) bool {
		return filter != nil && filter.LogBackupID != nil && *filter.LogBackupID == TestLogBackupID &&
			filter.Status != nil && *filter.Status == entity.CompactBackupStatusPending
	})).Return(pendingTasks, nil).Maybe()

	// 5. Mock downstream status checks if tasks have DownstreamCompactID
	for _, task := range tc.existingLocalTasks {
		if task.DownstreamCompactID != nil {
			downstreamID := *task.DownstreamCompactID
			if response, exists := tc.downstreamResponses[downstreamID]; exists {
				s.mockCompactFacade.On("GetCompactBackup", mock.Anything, mock.Anything, mock.Anything,
					TestTenantID, TestProjectID, TestClusterID, downstreamID).
					Return(response, nil).Maybe()

				// If downstream is complete/failed, expect Save call to update local status
				if response.Status == infraapi.CompactBackupStatusComplete || response.Status == infraapi.CompactBackupStatusFailed {
					s.mockCompactBackupRepo.On("Save", mock.Anything, mock.MatchedBy(func(task *entity.CompactBackup) bool {
						return task.ID == entity.ID(802) // Match the task ID
					})).Return(nil).Maybe()
				}
			}
		} else {
			// Task without downstream ID - expect Save call to mark as failed
			s.mockCompactBackupRepo.On("Save", mock.Anything, mock.MatchedBy(func(task *entity.CompactBackup) bool {
				return task.Status == entity.CompactBackupStatusFailed
			})).Return(nil).Maybe()
		}
	}

	// 6. Mock cluster region fetch for downstream status checks
	if len(tc.existingLocalTasks) > 0 {
		s.mockMgmtGrpcClient.On("GetDedicatedClusterBasicByID", mock.Anything, TestClusterID).
			Return(createValidCluster(), nil).Maybe()
	}

	// 7. Set up CompactBackupModel mock expectations if task should be created
	if tc.shouldCreateCompactTask {
		s.mockCompactBackupModel.On("AsyncCreate", mock.Anything,
			mock.MatchedBy(func(req *domain.AsyncCreateCompactBackupReq) bool {
				return req.ClusterID == TestClusterID && req.LogBackupID == TestLogBackupID
			})).Return(&domain.AsyncCreateCompactBackupResp{
			CompactBackupID: 999,
			WorkflowID:      "test-workflow-id",
		}, nil).Once()
	}

	// Execute CompactLogBackup
	err := s.scheduler.CompactLogBackup(context.Background())

	// Verify results
	if tc.shouldCreateCompactTask {
		s.NoError(err, "Should create compact task successfully for case: %s", tc.name)
	} else {
		s.NoError(err, "Should complete without creating task for case: %s", tc.name)
	}
}

// Step 4: checkCompactFeatureEnabled tests - test feature gate scenarios
func (s *compactBackupSchedulerTestSuite) TestCompactLogBackupFeatureGate() {
	testCases := []struct {
		name                    string
		featureEnabled          bool
		getClusterError         error
		shouldCreateCompactTask bool
		description             string
	}{
		{
			name:                    "should_not_compact_feature_disabled",
			featureEnabled:          false,
			getClusterError:         nil,
			shouldCreateCompactTask: false,
			description:             "Feature gate disabled, should not create compact task",
		},
		{
			name:                    "should_not_compact_cluster_fetch_failed",
			featureEnabled:          true,
			getClusterError:         bizerr.NewWithStack(bizerr.ErrCodeUnknown, "failed to get cluster"),
			shouldCreateCompactTask: false,
			description:             "Cluster fetch failed, should not create compact task",
		},
		{
			name:                    "should_compact_feature_enabled",
			featureEnabled:          true,
			getClusterError:         nil,
			shouldCreateCompactTask: true,
			description:             "Feature enabled and cluster accessible, should create compact task",
		},
	}

	for _, tc := range testCases {
		testCase := tc
		s.Run(testCase.name, func() {
			s.SetupTest()
			s.runFeatureGateTestCase(testCase)
		})
	}
}

// Helper function to run feature gate test cases
func (s *compactBackupSchedulerTestSuite) runFeatureGateTestCase(tc struct {
	name                    string
	featureEnabled          bool
	getClusterError         error
	shouldCreateCompactTask bool
	description             string
}) {
	// Create a valid log backup that should normally trigger compact
	logBackup := s.createLogBackupRunning7Hours()

	// 1. Mock getRunningLogBackups to return our test LogBackup
	s.mockLogBackupRepo.On("List", mock.Anything, mock.Anything).
		Return([]entity.LogBackup{*logBackup}, nil).Once()

	// 2. Mock cluster fetch for feature gate check
	if tc.getClusterError != nil {
		s.mockMgmtGrpcClient.On("GetDedicatedClusterBasicByID", mock.Anything, TestClusterID).
			Return(nil, tc.getClusterError).Once()
	} else {
		s.mockMgmtGrpcClient.On("GetDedicatedClusterBasicByID", mock.Anything, TestClusterID).
			Return(createValidCluster(), nil).Maybe()
	}

	// 3. Mock feature gate check
	s.mockFeatureGate.On("IsFeatureEnabled", mock.Anything, mock.Anything).
		Return(tc.featureEnabled).Maybe()

	// 4. If feature is enabled and no cluster error, mock the rest of the flow
	if tc.shouldCreateCompactTask {
		// Mock GetAnyClusterByID for getCompactCycle
		s.mockMgmtGrpcClient.On("GetAnyDedicatedClusterBasicByID", mock.Anything, TestClusterID).
			Return(createValidCluster(), nil).Maybe()

		// Mock ShouldCompactLogBackup flow
		s.mockCompactBackupRepo.On("ListByFilter", mock.Anything, mock.MatchedBy(func(filter *backup_repo.CompactBackupFilter) bool {
			return filter != nil && filter.LogBackupID != nil && *filter.LogBackupID == TestLogBackupID &&
				filter.Status != nil && *filter.Status == entity.CompactBackupStatusSuccess
		})).Return([]*entity.CompactBackup{}, nil).Maybe() // No previous success tasks

		// Mock hasRunningCompactTask flow (no existing tasks)
		s.mockCompactBackupRepo.On("ListByFilter", mock.Anything, mock.MatchedBy(func(filter *backup_repo.CompactBackupFilter) bool {
			return filter != nil && filter.LogBackupID != nil && *filter.LogBackupID == TestLogBackupID &&
				(filter.Status != nil && (*filter.Status == entity.CompactBackupStatusRunning || *filter.Status == entity.CompactBackupStatusPending))
		})).Return([]*entity.CompactBackup{}, nil).Maybe() // No running/pending tasks

		// Mock CompactBackupModel.AsyncCreate
		s.mockCompactBackupModel.On("AsyncCreate", mock.Anything,
			mock.MatchedBy(func(req *domain.AsyncCreateCompactBackupReq) bool {
				return req.ClusterID == TestClusterID && req.LogBackupID == TestLogBackupID
			})).Return(&domain.AsyncCreateCompactBackupResp{
			CompactBackupID: 999,
			WorkflowID:      "test-workflow-id",
		}, nil).Once()
	}

	// Execute CompactLogBackup
	err := s.scheduler.CompactLogBackup(context.Background())

	// Verify results
	if tc.getClusterError != nil {
		// Should fail if cluster fetch fails
		s.Error(err, "Should fail when cluster fetch fails for case: %s", tc.name)
	} else {
		// Should succeed regardless of whether task is created
		s.NoError(err, "Should complete without error for case: %s", tc.name)
	}
}