package br

import (
	"context"
	"fmt"
	"time"

	"github.com/aws/smithy-go/ptr"
	clusterngv1beta "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"

	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/br"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/infra/repoimpl/schema"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/pkg/config"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
)

// The sync-missing algorithm is designed to be a standalone algorithm, it didn't coupled with the backup-sync algorithm.
// 1. This algorithm periodically checks the missing backups of every cluster and syncs them.
// 2. It will involve scanning the backup records in the database, compare the expected backups and the actual backups.
// 3. If there is a missing, resync it
// TODO: in order to avoid scanning too many records, we can cache scanning result in memory to lower pressure on database in the future when we have many clusters.

// SyncMissingBackups with rate limit
func (w *worker) SyncMissingBackups(ctx context.Context) error {
	if !w.cfg.SyncMissingConfig.Enable {
		tilog.Info(ctx, "sync missing backups is disabled, skip")
		return nil
	}

	startTime := time.Now().UTC()
	SyncMissingWorkerRunsTotal.Inc()

	defer func() {
		duration := time.Since(startTime).Seconds()
		SyncMissingWorkerDuration.Observe(duration)
	}()

	// SyncMissingBackups has its own distributed lease and may run on a replica
	// whose process-local slots missed the cluster deletion event. Reconcile the
	// cache from the database before taking a snapshot of it.
	if err := w.backupSyncSlots.initIfNot(ctx); err != nil {
		return fmt.Errorf("initialize backup sync slots before missing backup sync: %w", err)
	}
	if err := w.UpdateBackupSyncSlots(ctx); err != nil {
		return fmt.Errorf("reconcile backup sync slots before missing backup sync: %w", err)
	}

	// Convert map to slice for easier batch processing
	// TODO: more efficient way visit backup settings thread-safely
	backupSettings := w.backupSyncSlots.copyBackupSettings()
	exclusionsBySharedPool := make(map[uint64][]br.BackupReconciliationExclusion)
	exclusionsLoaded := make(map[uint64]bool)
	maxRetentionDays := 0
	maxPrePauseLookback := time.Duration(0)
	for _, bs := range backupSettings {
		maxRetentionDays = max(maxRetentionDays, effectiveMissingBackupRetentionDays(bs.Schedule))
		maxPrePauseLookback = max(maxPrePauseLookback, prePauseReconciliationLookback(bs, w.cfg))
	}
	maxBackTraceHours := max(
		w.cfg.SyncMissingConfig.BackTraceStartPerDayDays*24,
		max(
			w.cfg.SyncMissingConfig.BackTraceStartPerHourHours,
			w.cfg.SyncMissingConfig.BackTraceStartPerMinuteHours,
		),
	)
	// A closed Pause interval may be older than the schedule's normal missing
	// backup lookback. Query far enough back to recover any still-retained backup
	// immediately preceding that interval, including the normal reconciliation
	// delay before Pause admission.
	exclusionLookback := max(
		time.Duration(maxBackTraceHours)*time.Hour,
		time.Duration(maxRetentionDays)*24*time.Hour,
	) + maxPrePauseLookback
	exclusionQueryStart := startTime.Add(-exclusionLookback)

	tilog.Info(ctx, "starting missing backup sync",
		zap.Int("total_backup_settings", len(backupSettings)))

	// Process backup settings in batches
	for i := 0; i < len(backupSettings); i += w.cfg.SyncMissingConfig.BatchSize {
		end := min(i+w.cfg.SyncMissingConfig.BatchSize, len(backupSettings))

		batch := backupSettings[i:end]

		tilog.Info(ctx, "processing backup settings batch",
			zap.Int("batch_start", i),
			zap.Int("batch_end", end),
			zap.Int("batch_size", len(batch)))

		// Process each backup setting in the current batch
		for _, bs := range batch {
			inPauseWindow, err := w.isSharedPoolInPauseWindow(ctx, bs.SharedPoolId)
			if err != nil {
				tilog.Error(ctx, "failed to determine shared pool pause state; skip missing backup sync to avoid false positives",
					zap.Error(err), zap.Uint64("shared_pool_id", bs.SharedPoolId), zap.Uint64("cluster_id", bs.ClusterId))
				continue
			}
			if inPauseWindow {
				tilog.Info(ctx, "skip missing backup sync during shared pool pause window",
					zap.Uint64("shared_pool_id", bs.SharedPoolId), zap.Uint64("cluster_id", bs.ClusterId))
				continue
			}
			// The slot cache may predate a schedule change or cluster deletion.
			// Refresh unconditionally and accept only a live setting before deriving
			// expected backups.
			freshSetting, berr := w.backupSettingRepo.Get(ctx, bs.ClusterId)
			if berr != nil {
				if berr.ErrCode() == bizerr.ErrCodeRecordNotFound {
					w.backupSyncSlots.removeCluster(bs.ClusterId)
					tilog.Info(ctx, "skip missing backup sync because backup setting no longer exists",
						zap.Uint64("shared_pool_id", bs.SharedPoolId), zap.Uint64("cluster_id", bs.ClusterId))
					continue
				}
				tilog.Error(ctx, "failed to refresh backup setting before missing backup sync",
					zap.Error(berr), zap.Uint64("shared_pool_id", bs.SharedPoolId), zap.Uint64("cluster_id", bs.ClusterId))
				continue
			}
			if freshSetting.State != schema.BackupSettingStateCreated || freshSetting.DeletedAt.Valid {
				w.backupSyncSlots.removeCluster(bs.ClusterId)
				tilog.Info(ctx, "skip missing backup sync because backup setting is not active",
					zap.Uint64("shared_pool_id", bs.SharedPoolId),
					zap.Uint64("cluster_id", bs.ClusterId),
					zap.String("backup_setting_state", string(freshSetting.State)))
				continue
			}
			bs = freshSetting
			if !exclusionsLoaded[bs.SharedPoolId] {
				exclusions, berr := w.backupSettingRepo.ListBackupReconciliationExclusions(
					ctx,
					bs.SharedPoolId,
					exclusionQueryStart,
					startTime,
				)
				if berr != nil {
					tilog.Error(ctx, "failed to list backup reconciliation exclusions",
						zap.Error(berr), zap.Uint64("shared_pool_id", bs.SharedPoolId))
					continue
				}
				exclusionsBySharedPool[bs.SharedPoolId] = exclusions
				exclusionsLoaded[bs.SharedPoolId] = true
			}
			start, end, needCheck := checkAndCalculateBackTraceDuration(
				ctx,
				bs,
				startTime,
				w.cfg,
				exclusionsBySharedPool[bs.SharedPoolId],
			)
			if !needCheck {
				continue
			}
			tilog.Info(ctx, "calculating back trace duration",
				zap.Time("start", *start),
				zap.Time("end", *end),
				zap.Uint64("cluster_id", bs.ClusterId))

			err = w.handlingBackupSettingIfBackupMissing(
				ctx,
				bs,
				*start,
				*end,
				exclusionsBySharedPool[bs.SharedPoolId],
			)
			if err != nil {
				tilog.Error(ctx, "failed to handling backup settings if backup missing",
					zap.Error(err), zap.Uint64("cluster_id", bs.ClusterId))
			}
		}

		// Add delay between batches (except for the last batch)
		if end < len(backupSettings) {
			tilog.Info(ctx, "waiting between batches", zap.Duration("delay", w.cfg.SyncMissingConfig.BatchInterval))
			time.Sleep(w.cfg.SyncMissingConfig.BatchInterval)
		}
	}

	tilog.Info(ctx, "completed missing backup sync",
		zap.Int("total_backup_settings", len(backupSettings)),
		zap.Duration("total_duration", time.Since(startTime)))

	return nil
}

func (w *worker) handlingBackupSettingIfBackupMissing(
	ctx context.Context,
	bs *entity.BackupSetting,
	backTraceStart, backTraceEnd time.Time,
	exclusions []br.BackupReconciliationExclusion,
) (err error) {
	// infos
	tilog.Info(ctx, "handling backup setting if backup missing",
		zap.Time("back_trace_start", backTraceStart),
		zap.Time("back_trace_end", backTraceEnd),
		zap.Any("backup_setting", bs))

	// Calculate expected backups, then remove only the intervals in which Pause
	// intentionally suspended scheduling. Existing Regional backups are never
	// hidden by these intervals from normal synchronization.
	expectedBackups := calculateExpectedBackupsInDuration(bs, backTraceStart, backTraceEnd)
	expectedBackups = excludeBackupTimePoints(expectedBackups, exclusions)
	tilog.Info(ctx, "expected backups in back trace duration",
		zap.Uint64("cluster_id", bs.ClusterId),
		zap.String("schedule_type", bs.Schedule.Type),
		zap.Time("start", backTraceStart),
		zap.Time("end", backTraceEnd),
		zap.Int("expected_backups_count", len(expectedBackups)))

	// query the backups in back trace duration
	actualBackups, err := w.backupRepo.UnscopedQuery(ctx, &br.QueryBackupsFilter{
		ClusterIds:    []uint64{bs.ClusterId},
		TriggerTypes:  []schema.BackupTriggerType{schema.BackupTriggerTypeAuto},
		BackupTimeLte: &backTraceEnd,
		BackupTimeGte: &backTraceStart,
		OrderBy:       ptr.String("backup_time"), // make it ascending order by backup time
	}, nil)
	if err != nil {
		SyncMissingErrorTotal.WithLabelValues(
			fmt.Sprintf("%d", bs.SharedPoolId),
			fmt.Sprintf("%d", bs.ClusterId),
			SyncMissingErrorType_QueryBackups,
		).Inc()
		tilog.Error(ctx, "failed to query backups in back trace duration",
			zap.Error(err), zap.Uint64("cluster_id", bs.ClusterId))
		return err
	}
	tilog.Info(ctx, "actual backups in back trace duration",
		zap.Uint64("cluster_id", bs.ClusterId),
		zap.String("schedule_type", bs.Schedule.Type),
		zap.Time("start", backTraceStart),
		zap.Time("end", backTraceEnd),
		zap.Int("actual_backups_count", len(actualBackups)))

	// compare the missing backups by comparing the expected backups and the actual backups
	missingBackups := calculateMissingBackups(bs, expectedBackups, actualBackups)
	if len(missingBackups) == 0 {
		tilog.Info(ctx, "no missing backups",
			zap.Any("cluster id", bs.ClusterId),
			zap.Time("back_trace_start", backTraceStart), zap.Time("back_trace_end", backTraceEnd))
		return nil
	}
	BackupMissingTotal.WithLabelValues(fmt.Sprintf("%d", bs.SharedPoolId), fmt.Sprintf("%d", bs.ClusterId)).Add(float64(len(missingBackups)))
	tilog.Info(ctx, "missing backups",
		zap.Any("cluster id", bs.ClusterId),
		zap.String("schedule_type", bs.Schedule.Type),
		zap.Time("back_trace_start", backTraceStart),
		zap.Time("back_trace_end", backTraceEnd),
		zap.Int("missing_backups_count", len(missingBackups)),
		zap.Any("missing_backups", missingBackups))

	err = w.resyncMissingBackups(ctx, bs, missingBackups)
	if err != nil {
		tilog.Error(ctx, "failed to resync missing backups",
			zap.Error(err), zap.Any("cluster id", bs.ClusterId))
		SyncMissingErrorTotal.WithLabelValues(
			fmt.Sprintf("%d", bs.SharedPoolId),
			fmt.Sprintf("%d", bs.ClusterId),
			SyncMissingErrorType_Unclassified,
		).Inc()
	}
	tilog.Info(ctx, "resync missing backups done",
		zap.Any("cluster id", bs.ClusterId))

	return nil
}

func (w *worker) resyncMissingBackups(
	ctx context.Context,
	bs *entity.BackupSetting,
	timePoints []time.Time,
) error {
	sharedPool, err := w.resourcePoolRepo.Load(ctx, bs.SharedPoolId, false)
	if err != nil {
		tilog.Error(ctx, "load shared pool failed",
			zap.Error(err),
			zap.Uint64("clusterId", bs.ClusterId),
			zap.Uint64("sharedPoolId", bs.SharedPoolId))
		return err.Chain("load shared pool failed")
	}
	region, err := w.regionRepo.Get(ctx, sharedPool.RegionId)
	if err != nil {
		tilog.Error(ctx, "get region failed",
			zap.Error(err),
			zap.Uint64("clusterId", bs.ClusterId),
			zap.Uint64("sharedPoolId", bs.SharedPoolId),
			zap.Uint64("regionId", sharedPool.RegionId))
		return err.Chain("get region failed")
	}

	for _, timePoint := range timePoints {
		tilog.Info(ctx, "resync missing backup ...",
			zap.Any("cluster id", bs.ClusterId),
			zap.Any("timepoint", timePoint))
		resp, err := w.regionalserverCli.ListLightweightBackups(ctx,
			region,
			&clusterngv1beta.ListLightweightBackupsRequest{
				SharedPoolId: fmt.Sprintf("%d", bs.SharedPoolId),
				FromPrefix:   backupListFromPrefix(timePoint),
				LoadDetails:  true,
				MaxCount:     backupListMaxCount,
			})
		if err != nil {
			tilog.Error(
				ctx,
				"list lightweight backups failed in resync missing backups",
				zap.Error(err),
				zap.Uint64("sharedPoolId", bs.SharedPoolId),
				zap.Uint64("clusterId", bs.ClusterId),
				zap.Time("timepoint", timePoint))
			SyncMissingErrorTotal.WithLabelValues(
				fmt.Sprintf("%d", bs.SharedPoolId),
				fmt.Sprintf("%d", bs.ClusterId),
				SyncMissingErrorType_FetchBackups,
			).Inc()
			continue
		}

		records, missingClusters := assembleBackupRecords(
			ctx,
			resp,
			map[uint64]bool{bs.ClusterId: true},
			map[uint64]*entity.BackupSetting{bs.ClusterId: bs},
			timePoint)
		if err := applyCurrentClusterEncryptionInfos(ctx, records, w.clusterRepo, w.backupSyncSlots.backupEncryptionResolver); err != nil {
			tilog.Error(ctx, "apply current cluster encryption infos failed in resync missing backups",
				zap.Error(err),
				zap.Uint64("clusterId", bs.ClusterId),
				zap.Time("timepoint", timePoint))
			return err
		}
		if len(missingClusters) > 0 {
			tilog.Error(ctx, "missing clusters in resync missing backups",
				zap.Any("missing_clusters", missingClusters),
				zap.Uint64("clusterId", bs.ClusterId),
				zap.Time("timepoint", timePoint))
			SyncMissingErrorTotal.WithLabelValues(
				fmt.Sprintf("%d", bs.SharedPoolId),
				fmt.Sprintf("%d", bs.ClusterId),
				SyncMissingErrorType_BackupMissing,
			).Inc()
			continue
		}

		record := records[bs.ClusterId]
		if record == nil {
			tilog.Error(ctx, "record is nil in resync missing backups, there must be a bug",
				zap.Uint64("clusterId", bs.ClusterId),
				zap.Time("timepoint", timePoint))
			SyncMissingErrorTotal.WithLabelValues(
				fmt.Sprintf("%d", bs.SharedPoolId),
				fmt.Sprintf("%d", bs.ClusterId),
				SyncMissingErrorType_Bug,
			).Inc()
			continue
		}

		backupSettingInactive := false
		mutation := func(txCtx context.Context) error {
			// This path only persists a real backup returned by Regional. Schedule
			// and Pause fences apply to synthetic missing records, not real history.
			if _, berr := w.backupSettingRepo.GetActiveForUpdate(txCtx, bs.ClusterId); berr != nil {
				if berr.ErrCode() == bizerr.ErrCodeRecordNotFound {
					backupSettingInactive = true
				}
				return berr
			}
			return w.backupRepo.Create(txCtx, record)
		}
		var mutationErr error
		switch {
		case w.transaction != nil:
			mutationErr = w.transaction.Transaction(ctx, mutation)
		default:
			mutationErr = mutation(ctx)
		}
		if mutationErr != nil {
			if backupSettingInactive {
				w.backupSyncSlots.removeCluster(bs.ClusterId)
				tilog.Info(ctx, "stop resyncing missing backups because backup setting is no longer active",
					zap.Uint64("clusterId", bs.ClusterId),
					zap.Uint64("sharedPoolId", bs.SharedPoolId))
				return nil
			}
			tilog.Error(
				ctx,
				"failed to save backup to database in resync missing backups",
				zap.Error(mutationErr),
				zap.Uint64("clusterId", bs.ClusterId),
				zap.Time("timepoint", timePoint),
				zap.Any("backup", record),
			)
			SyncMissingErrorTotal.WithLabelValues(
				fmt.Sprintf("%d", bs.SharedPoolId),
				fmt.Sprintf("%d", bs.ClusterId),
				SyncMissingErrorType_SaveBackup,
			).Inc()
			continue
		}
		tilog.Info(ctx, "resync missing backup done",
			zap.Any("cluster id", bs.ClusterId),
			zap.Any("timepoint", timePoint))
		SyncMissingSuccessTotal.WithLabelValues(
			fmt.Sprintf("%d", bs.SharedPoolId),
			fmt.Sprintf("%d", bs.ClusterId),
		).Inc()
	}

	return nil
}

func calculateExpectedBackupsInDuration(bs *entity.BackupSetting, start, end time.Time) []time.Time {
	if bs.Mode == schema.BackupModeFlexibleRetention {
		return calculateExpectedBackupsPerDay(bs.Schedule.DailyBackupMinuteOfDay, start, end)
	}

	switch bs.Schedule.Type {
	case "per-day":
		return calculateExpectedBackupsPerDay(bs.Schedule.At, start, end)
	case "per-hour":
		return calculateExpectedBackupsPerHour(bs.Schedule.At, start, end)
	case "per-minute":
		return calculateExpectedBackupsPerMinute(start, end)
	}
	return nil
}

func excludeBackupTimePoints(
	timePoints []time.Time,
	exclusions []br.BackupReconciliationExclusion,
) []time.Time {
	result := make([]time.Time, 0, len(timePoints))
	for _, timePoint := range timePoints {
		excluded := false
		for _, exclusion := range exclusions {
			if timePoint.Before(exclusion.StartAt) {
				continue
			}
			if exclusion.EndAt == nil || !timePoint.After(*exclusion.EndAt) {
				excluded = true
				break
			}
		}
		if !excluded {
			result = append(result, timePoint)
		}
	}
	return result
}

func calculateExpectedBackupsPerDay(scheduleAtMinute int, start, end time.Time) []time.Time {
	firstScheduleTime := start.Truncate(24 * time.Hour).Add(time.Duration(scheduleAtMinute) * time.Minute)
	if firstScheduleTime.Before(start) {
		firstScheduleTime = firstScheduleTime.Add(24 * time.Hour)
	}
	endScheduleTime := end.Truncate(24 * time.Hour).Add(time.Duration(scheduleAtMinute) * time.Minute)
	if endScheduleTime.After(end) {
		endScheduleTime = endScheduleTime.Add(-24 * time.Hour)
	}

	timePoints := []time.Time{}
	for p := firstScheduleTime; p.Before(endScheduleTime) || p.Equal(endScheduleTime); p = p.Add(24 * time.Hour) {
		timePoints = append(timePoints, p)
	}
	return timePoints
}

func calculateExpectedBackupsPerHour(scheduleAtMinute int, start, end time.Time) []time.Time {
	firstScheduleTime := start.Truncate(time.Hour).Add(time.Duration(scheduleAtMinute) * time.Minute)
	if firstScheduleTime.Before(start) {
		firstScheduleTime = firstScheduleTime.Add(time.Hour)
	}
	endScheduleTime := end.Truncate(time.Hour).Add(time.Duration(scheduleAtMinute) * time.Minute)
	if endScheduleTime.After(end) {
		endScheduleTime = endScheduleTime.Add(-time.Hour)
	}

	timePoints := []time.Time{}
	for p := firstScheduleTime; p.Before(endScheduleTime) || p.Equal(endScheduleTime); p = p.Add(time.Hour) {
		timePoints = append(timePoints, p)
	}
	return timePoints
}

func calculateExpectedBackupsPerMinute(start, end time.Time) []time.Time {
	startMinute := roundUp(start, time.Minute)
	endMinute := end.Truncate(time.Minute)

	timePoints := []time.Time{}
	for p := startMinute; p.Before(endMinute) || p.Equal(endMinute); p = p.Add(time.Minute) {
		timePoints = append(timePoints, p)
	}
	return timePoints
}

func roundUp(t time.Time, d time.Duration) time.Time {
	if d <= 0 {
		return t
	}
	truncated := t.Truncate(d)
	if truncated.Equal(t) {
		return t
	}
	return truncated.Add(d)
}

func calculateMissingBackups(
	bs *entity.BackupSetting,
	expectedBackups []time.Time,
	actualBackups []*entity.Backup,
) []time.Time {
	if bs.Mode == schema.BackupModeFlexibleRetention {
		return calculateMissingBackupsPerDay(bs.Schedule.DailyBackupMinuteOfDay, expectedBackups, actualBackups)
	}

	switch bs.Schedule.Type {
	case "per-day":
		return calculateMissingBackupsPerDay(bs.Schedule.At, expectedBackups, actualBackups)
	case "per-hour":
		return calculateMissingBackupsPerHour(bs.Schedule.At, expectedBackups, actualBackups)
	case "per-minute":
		return calculateMissingBackupsPerMinute(expectedBackups, actualBackups)
	}
	return nil
}

func calculateMissingBackupsPerDay(
	scheduleAtMinute int,
	expectedBackups []time.Time,
	actualBackups []*entity.Backup,
) []time.Time {
	// convert actualBackups to expected schedule time
	normalizedActualBackups := []time.Time{}
	for _, actualBackup := range actualBackups {
		backupTime := actualBackup.BackupTime
		todayExpectedScheduleTime := backupTime.Truncate(24 * time.Hour).
			Add(time.Duration(scheduleAtMinute) * time.Minute)

		if todayExpectedScheduleTime.Before(backupTime) || todayExpectedScheduleTime.Equal(backupTime) {
			normalizedActualBackups = append(normalizedActualBackups, todayExpectedScheduleTime)
		} else {
			normalizedActualBackups = append(normalizedActualBackups, todayExpectedScheduleTime.Add(-24*time.Hour))
		}
	}

	return calculateMissingTimePoints(expectedBackups, normalizedActualBackups)
}

func calculateMissingBackupsPerHour(
	scheduleAtMinute int,
	expectedBackups []time.Time,
	actualBackups []*entity.Backup,
) []time.Time {
	// convert actualBackups to expected schedule time
	normalizedActualBackups := []time.Time{}
	for _, actualBackup := range actualBackups {
		backupTime := actualBackup.BackupTime
		thisHourExpectedScheduleTime := backupTime.Truncate(time.Hour).
			Add(time.Duration(scheduleAtMinute) * time.Minute)

		if thisHourExpectedScheduleTime.Before(backupTime) || thisHourExpectedScheduleTime.Equal(backupTime) {
			normalizedActualBackups = append(normalizedActualBackups, thisHourExpectedScheduleTime)
		} else {
			normalizedActualBackups = append(normalizedActualBackups, thisHourExpectedScheduleTime.Add(-time.Hour))
		}
	}

	return calculateMissingTimePoints(expectedBackups, normalizedActualBackups)
}

func calculateMissingBackupsPerMinute(expectedBackups []time.Time, actualBackups []*entity.Backup) []time.Time {
	// convert actualBackups to expected schedule time
	normalizedActualBackups := []time.Time{}
	for _, actualBackup := range actualBackups {
		backupTime := actualBackup.BackupTime
		normalizedActualBackups = append(normalizedActualBackups, backupTime.Truncate(time.Minute))
	}
	return calculateMissingTimePoints(expectedBackups, normalizedActualBackups)
}

func calculateMissingTimePoints(expectedTimePoints, actualTimePoints []time.Time) []time.Time {
	if len(expectedTimePoints) == 0 {
		return nil
	}
	samllestTimePoint := expectedTimePoints[0]
	largestTimePoint := expectedTimePoints[len(expectedTimePoints)-1]
	removeOutOfRange := []time.Time{}
	for _, p := range actualTimePoints {
		if p.Before(samllestTimePoint) || p.After(largestTimePoint) {
			continue
		}
		removeOutOfRange = append(removeOutOfRange, p)
	}
	if len(removeOutOfRange) == 0 {
		return expectedTimePoints
	}

	current := removeOutOfRange[0]
	removeDuplicated := []time.Time{current}
	for i := 1; i < len(removeOutOfRange); i++ {
		if removeOutOfRange[i].Equal(current) {
			continue
		}
		current = removeOutOfRange[i]
		removeDuplicated = append(removeDuplicated, current)
	}

	missing := []time.Time{}
	i := 0
	for _, expectedTimePoint := range expectedTimePoints {
		if i >= len(removeDuplicated) {
			missing = append(missing, expectedTimePoint)
			continue
		}
		if removeDuplicated[i].Equal(expectedTimePoint) {
			i++
			continue
		}
		if removeDuplicated[i].After(expectedTimePoint) {
			missing = append(missing, expectedTimePoint)
		}
	}
	return missing
}

// TODO: in order to avoid scanning too many records, we can cache scanning result in memory to lower pressure on database,
// then we can increase the scanning frequency to make up missing backups quickly
func checkAndCalculateBackTraceDuration(
	ctx context.Context,
	bs *entity.BackupSetting,
	now time.Time,
	cfg *config.BackupConfig,
	exclusions []br.BackupReconciliationExclusion,
) (startTime, endTime *time.Time, needCheck bool) {
	// avoid overlay with normal sync
	end := now.Add(-time.Duration(cfg.SyncMissingConfig.BackTraceEndHoursAgo) * time.Hour)
	watermark := bs.LastScheduleUpdatedAt
	if end.Before(watermark) {
		// if schedule is changed, it's hard to know the expected backup history, so skip check
		tilog.Info(
			ctx,
			"backup reconciliation watermark is after the end of the back trace duration, skip check",
			zap.Time("reconciliation_watermark", watermark),
			zap.Time("end", end),
		)
		return nil, nil, false
	}

	var start time.Time
	if bs.Mode == schema.BackupModeFlexibleRetention {
		start = now.Add(-time.Duration(cfg.SyncMissingConfig.BackTraceStartPerDayDays) * 24 * time.Hour)
	} else {
		switch bs.Schedule.Type {
		case "per-day":
			start = now.Add(-time.Duration(cfg.SyncMissingConfig.BackTraceStartPerDayDays) * 24 * time.Hour)
		case "per-hour":
			start = now.Add(-time.Duration(cfg.SyncMissingConfig.BackTraceStartPerHourHours) * time.Hour)
		case "per-minute":
			start = now.Add(-time.Duration(cfg.SyncMissingConfig.BackTraceStartPerMinuteHours) * time.Hour)
		default:
			tilog.Error(ctx, "invalid backup schedule type",
				zap.String("schedule_type", bs.Schedule.Type),
				zap.Uint64("cluster_id", bs.ClusterId))
			return nil, nil, false
		}
	}

	// Normal sync intentionally does not call TiBR during Pause. Extend the
	// missing-backup scan before each Pause interval by the normal reconciliation
	// delay so a still-retained pre-Pause backup cannot age out of the ordinary
	// lookback while scheduling is suspended.
	if lookback := prePauseReconciliationLookback(bs, cfg); lookback > 0 {
		for _, exclusion := range exclusions {
			candidate := exclusion.StartAt.Add(-lookback)
			if candidate.Before(start) {
				start = candidate
			}
		}
	}

	if start.Before(watermark) {
		start = watermark.Add(
			cfg.SyncConfig.LoadScheduleChangeInterval,
		) // add interval since change won't take effect immediately
	}

	retentionDays := effectiveMissingBackupRetentionDays(bs.Schedule)
	expireTime := now.Add(-time.Duration(retentionDays) * 24 * time.Hour)

	if start.Before(expireTime) {
		start = expireTime // it's meaningless to check backups before the expire time
	}

	if start.After(end) {
		tilog.Info(ctx, "start time is after end time, skip check",
			zap.Time("start", start),
			zap.Time("end", end),
			zap.Uint64("cluster_id", bs.ClusterId))
		return nil, nil, false
	}

	backupTimePoints := calculateExpectedBackupsInDuration(bs, start, end)
	if len(backupTimePoints) == 0 {
		tilog.Info(ctx, "no backup in back trace duration expected, skip check",
			zap.Time("start", start),
			zap.Time("end", end),
			zap.Uint64("cluster_id", bs.ClusterId))
		return nil, nil, false
	}

	return &start, &end, true
}

func prePauseReconciliationLookback(bs *entity.BackupSetting, cfg *config.BackupConfig) time.Duration {
	if cfg == nil {
		return backupScheduleCadence(bs)
	}
	return max(
		backupScheduleCadence(bs),
		max(
			time.Duration(cfg.SyncConfig.TraceBackHours)*time.Hour,
			time.Duration(cfg.SyncMissingConfig.BackTraceEndHoursAgo)*time.Hour,
		),
	)
}

func backupScheduleCadence(bs *entity.BackupSetting) time.Duration {
	if bs == nil || bs.BackupSetting == nil {
		return 0
	}
	if bs.Mode == schema.BackupModeFlexibleRetention {
		return 24 * time.Hour
	}
	switch bs.Schedule.Type {
	case "per-day":
		return 24 * time.Hour
	case "per-hour":
		return time.Hour
	case "per-minute":
		return time.Minute
	default:
		return 0
	}
}

// effectiveMissingBackupRetentionDays keeps legacy schedules working while
// accounting for the separate hourly and daily retention of newer schedules.
// Missing-backup scanning must cover the longest-lived backup category.
func effectiveMissingBackupRetentionDays(schedule schema.AutoSchedule) int {
	retentionDays := max(schedule.ZeroClockRetentionDays, max(schedule.NonZeroClockRetentionDays, schedule.RetentionDays))
	return retentionDays
}