package br

import (
	"cmp"
	"context"
	"errors"
	"fmt"
	"slices"
	"sync"
	"time"

	"github.com/samber/lo"
	clusterngv1beta "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"github.com/tidbcloud/tiinf/sdk/auditlog"
	"go.uber.org/zap"

	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/adapter/regionalserver"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/entity"
	brrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/br"
	clusterrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/cluster"
	regionrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/region"
	brservice "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/service/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 backup-sync algorithm
// 1. Aggregate shared pools and clusters into slots (24 x 60) by minute
// 2. Iterate the slots, sync the backup records
// Problem will be solved in this algorithm:
// 1. How to handle backup schedule change of a cluster? How to avoid loading two much data from database at once?
//    a) We can add a LastScheduleUpdateAt field to determine if the backup schedule has changed recently.
//    b) We should also take recently deleted clusters into account, we should not sync the backup records of deleted clusters.
//    c) We periodically load the changes into memory and update slots.
// 2. How to handle process reboot? Just rebuild the slots.
// 3. How to avoid bugs cause cluster missing in the slots? Rebuild the slots every day.
// 4. How to avoid process crash cause the backup records can not be synced in time? After process start, we can check the slot from 1 hours ago.

// The retry-sync algorithm, for the backup-sync failure caused by others dependent services. In order to keep refresh backups for clusters,
//   the backup-sync algorithm will not stop for endless retry for following situations:
//     a) Some regional cluster service failure
//     b) Some physical cluster failure
//     c) Database temporary failure
//     ...
//   In these cases, some backups will be missing, so we need another retry mechanism to handle these cases. In order to survive from all kinds of failure,
//   the retry mechanism should not couple with the backup-sync algorithm, it should be a standalone mechanism. See sync_missing.go for more details.
//   It can also handle the case that if this process is stopped for a long time, like more than 1 hour.

const (
	MinuteCountOfADay                     = 24 * 60
	backupListLookback                    = 2 * time.Minute
	backupListMaxCount                    = 10
	deletedSettingReconciliationBatchSize = 1000
)

func backupListFromPrefix(slotTime time.Time) string {
	return slotTime.UTC().Add(-backupListLookback).Format("20060102/1504")
}

func (w *worker) SyncBackups(ctx context.Context) error {
	if !w.cfg.SyncConfig.Enable {
		tilog.Info(ctx, "sync backups is disabled, skip")
		return nil
	}

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

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

	tilog.Info(ctx, "sync backups...")

	err := w.backupSyncSlots.initIfNot(ctx)
	if err != nil {
		tilog.Info(ctx, "init backup sync slots failed", zap.Error(err))
		return fmt.Errorf("init backup sync slots failed: %w", err)
	}
	// The lease owner can move between replicas, while backup sync slots are
	// process-local. Reconcile the cache owned by this run before using it so a
	// delete event consumed by another replica cannot leave stale clusters here.
	if err := w.backupSyncSlots.update(ctx); err != nil {
		NormalSyncErrorTotal.WithLabelValues(LabelValue_NA, LabelValue_NA, NormalSyncErrorType_UpdateSlots).Inc()
		tilog.Error(ctx, "update backup sync slots before sync failed", zap.Error(err))
		return fmt.Errorf("update backup sync slots before sync failed: %w", err)
	}

	w.backupSyncSlots.sync(ctx)

	tilog.Info(ctx, "sync backups success")

	return nil
}

func newBackupSyncSlots(
	cfg *config.SyncBackupsConfig,
	backupSettingRepo brrepo.BackupSettingRepo,
	backupRepo brrepo.BackupRepo,
	clusterRepo clusterrepo.Repo,
	backupEncryptionResolver *brservice.Resolver,
	regionRepo regionrepo.Repo,
	transaction db.Transactioner,
	regionalserverCli regionalserver.Client,
	pauseGuards ...pauseWorkerGuard,
) *backupSyncSlots {
	syncSlots := &backupSyncSlots{
		cfg: cfg,
		// dependencies
		backupSettingRepo:        backupSettingRepo,
		backupRepo:               backupRepo,
		clusterRepo:              clusterRepo,
		backupEncryptionResolver: backupEncryptionResolver,
		regionRepo:               regionRepo,
		transaction:              transaction,
		regionalserverCli:        regionalserverCli,

		// internal states
		lock:   &sync.Mutex{},
		isInit: false,
	}
	if len(pauseGuards) > 0 {
		syncSlots.pauseChecker = pauseGuards[0].checker
		syncSlots.pauseFencer = pauseGuards[0].fencer
	}
	return syncSlots
}

type backupSyncSlots struct {
	cfg *config.SyncBackupsConfig

	// dependencies
	backupSettingRepo        brrepo.BackupSettingRepo
	backupRepo               brrepo.BackupRepo
	clusterRepo              clusterrepo.Repo
	backupEncryptionResolver *brservice.Resolver
	regionRepo               regionrepo.Repo
	transaction              db.Transactioner
	regionalserverCli        regionalserver.Client
	pauseChecker             func(context.Context, uint64) (bool, error)
	pauseFencer              pauseFencer

	// internal states
	lock *sync.Mutex
	// config
	isInit bool
	// 24 x 60 slots
	// key: shared pool id
	// value: map[cluster id]bool
	slots []map[uint64]map[uint64]bool
	// used for quickly update the slots
	// key: cluster id
	// value: slot indexes
	clusterSlots map[uint64][]int
	// keep the latest backup settings, use thread-safe map to avoid race condition since it's accessed by sync_missing without protected by lock.
	// key: cluster id
	// value: backup setting
	backupSettings map[uint64]*entity.BackupSetting

	// progress of the sync
	currentSyncedCompletedSlotIndex *int
}

type pauseWorkerGuard struct {
	checker func(context.Context, uint64) (bool, error)
	fencer  pauseFencer
}

type backupClusterEncryptionInfo = brservice.Snapshot

func (s *backupSyncSlots) initIfNot(ctx context.Context) error {
	s.lock.Lock()
	defer s.lock.Unlock()

	if s.isInit {
		return nil
	}

	err := s.rebuildWithoutLock(ctx)
	if err != nil {
		tilog.Error(ctx, "build backup sync slots failed", zap.Error(err))
		return fmt.Errorf("build backup sync slots failed: %w", err)
	}
	tilog.Info(ctx, "first build backup sync slots success")

	// add a ticker to update the slots
	ticker := time.NewTicker(s.cfg.LoadScheduleChangeInterval)
	go func() {
		for range ticker.C {
			tilog.Info(ctx, "update backup sync slots...")
			err := s.update(ctx)
			if err != nil {
				NormalSyncErrorTotal.WithLabelValues(LabelValue_NA, LabelValue_NA, NormalSyncErrorType_UpdateSlots).Inc()
				tilog.Error(ctx, "update backup sync slots failed", zap.Error(err))
				continue
			}
			tilog.Info(ctx, "update backup sync slots success")
		}
	}()

	tilog.Info(ctx, "add ticker to update backup sync slots",
		zap.String("interval", s.cfg.LoadScheduleChangeInterval.String()))

	// add a ticker to rebuild the backup settings
	now := time.Now().UTC()
	currentMinuteOfDay := now.Hour()*60 + now.Minute()
	gap := s.cfg.StartRebuildMinuteAtADay - currentMinuteOfDay
	if gap < 0 {
		gap += MinuteCountOfADay
	}
	next := now.Add(time.Duration(gap) * time.Minute) // next rebuild time
	durationUntilNext := time.Until(next)
	tilog.Info(ctx, "add ticker to rebuild backup sync slots",
		zap.String("currentMinuteOfDay", fmt.Sprintf("%d", currentMinuteOfDay)),
		zap.String("gap_minutes", fmt.Sprintf("%d", gap)),
		zap.String("next_rebuild_time", next.Format(time.RFC3339)),
		zap.String("duration_until_next", durationUntilNext.String()),
		zap.String("interval", fmt.Sprintf("%d", s.cfg.RebuildSlotsIntervalMinutes)),
	)

	time.AfterFunc(durationUntilNext, func() {
		tilog.Info(ctx, "rebuild backup sync slots...")
		err := s.rebuild(ctx) // first next rebuild
		if err != nil {
			NormalSyncErrorTotal.WithLabelValues(LabelValue_NA, LabelValue_NA, NormalSyncErrorType_RebuildSlots).Inc()
			tilog.Error(ctx, "rebuild backup sync slots failed", zap.Error(err))
		} else {
			tilog.Info(ctx, "rebuild backup sync slots success")
		}

		ticker := time.NewTicker(time.Duration(s.cfg.RebuildSlotsIntervalMinutes) * time.Minute)
		for range ticker.C {
			tilog.Info(ctx, "rebuild backup sync slots...")
			err := s.rebuild(ctx)
			if err != nil {
				NormalSyncErrorTotal.WithLabelValues(LabelValue_NA, LabelValue_NA, NormalSyncErrorType_RebuildSlots).Inc()
				tilog.Error(ctx, "rebuild backup sync slots failed", zap.Error(err))
				continue
			}
			tilog.Info(ctx, "rebuild backup sync slots success")
		}
	})

	s.isInit = true

	return nil
}

func (s *backupSyncSlots) rebuildWithoutLock(ctx context.Context) error {
	s.slots = make([]map[uint64]map[uint64]bool, MinuteCountOfADay)
	s.clusterSlots = make(map[uint64][]int)
	s.backupSettings = make(map[uint64]*entity.BackupSetting)

	// get all living backup settings
	// TODO: add pagination when there are too many backup settings
	backupSettings, err := s.backupSettingRepo.UnscopedQuery(ctx, &brrepo.QueryBackupSettingFilter{
		States: []schema.BackupSettingState{schema.BackupSettingStateCreated},
	}, nil)
	if err != nil {
		return fmt.Errorf("list backup settings failed: %w", err)
	}

	for _, backupSetting := range backupSettings {
		// cache the backup settings here to make it easy to query
		s.backupSettings[backupSetting.ClusterId] = backupSetting

		// build the slots indexes
		indexes := calculateSlotsIndexes(backupSetting)
		s.clusterSlots[backupSetting.ClusterId] = indexes
		for _, index := range indexes {
			putSlots(s.slots, index, backupSetting.SharedPoolId, backupSetting.ClusterId)
		}
	}

	return nil
}

func putSlots(slots []map[uint64]map[uint64]bool, index int, sharedPoolId, clusterId uint64) {
	if slots[index] == nil {
		slots[index] = make(map[uint64]map[uint64]bool)
	}
	if slots[index][sharedPoolId] == nil {
		slots[index][sharedPoolId] = make(map[uint64]bool)
	}
	slots[index][sharedPoolId][clusterId] = true
}

func calculateSlotsIndexes(backupSetting *entity.BackupSetting) []int {
	// Flexible retention keeps the regional lightweight backups, but Global
	// exposes only the backup created at the customer-selected daily time.
	// Standard bundles continue to sync their existing hourly backups.
	if backupSetting.Mode == schema.BackupModeFlexibleRetention {
		return []int{backupSetting.Schedule.DailyBackupMinuteOfDay}
	}

	indexes := make([]int, 0)
	switch backupSetting.Schedule.Type {
	case "per-day":
		indexes = append(indexes, backupSetting.Schedule.At)
	case "per-hour":
		for i := range 24 {
			indexes = append(indexes, i*60+backupSetting.Schedule.At)
		}
	case "per-minute":
		for i := range MinuteCountOfADay {
			indexes = append(indexes, i)
		}
	}
	return indexes
}

func (s *backupSyncSlots) rebuild(ctx context.Context) error {
	s.lock.Lock()
	defer s.lock.Unlock()
	return s.rebuildWithoutLock(ctx)
}

func (s *backupSyncSlots) sync(ctx context.Context) {
	s.lock.Lock()
	defer s.lock.Unlock()

	now := time.Now().UTC()
	minutesToBeSynced := calculateToBeSyncedMinutes(ctx, s.currentSyncedCompletedSlotIndex, now, s.cfg)
	tilog.Info(ctx, "calculate minutes to be synced",
		zap.Time("now", now),
		zap.Any("minutesToBeSynced", minutesToBeSynced),
	)
	if len(minutesToBeSynced) == 0 {
		tilog.Info(ctx, "no minutes to be synced, skip",
			zap.Time("now", now),
			zap.Any("currentSyncedCompletedSlotIndex", s.currentSyncedCompletedSlotIndex),
		)
		return
	}

	toBeSyncedUnits := s.calculateToBeSyncedUnit(ctx, minutesToBeSynced, now)
	if len(toBeSyncedUnits) == 0 {
		lastMinute := minutesToBeSynced[len(minutesToBeSynced)-1]
		lastMinuteSlotIndex := mintueToSlotIndex(lastMinute)
		moveFrom := s.currentSyncedCompletedSlotIndex
		s.currentSyncedCompletedSlotIndex = &lastMinuteSlotIndex
		tilog.Info(
			ctx,
			"no units to be synced, move the current synced completed slot index to the last minute slot index",
			zap.Time("now", now),
			zap.Any("moveFrom", moveFrom),
			zap.Any("moveTo", s.currentSyncedCompletedSlotIndex),
			zap.Any("minutesToBeSynced", minutesToBeSynced),
		)
		return
	}

	for _, toBeSyncedUnit := range toBeSyncedUnits {
		tilog.Info(ctx, "sync the unit",
			zap.Int("slotIndex", toBeSyncedUnit.slotIndex),
			zap.Time("slotTime", toBeSyncedUnit.slotTime),
			zap.Any("sharedPoolIds", lo.Keys(toBeSyncedUnit.sharedPoolClusters)),
		)

		s.syncOneUnit(ctx, &toBeSyncedUnit)

		// update the current synced completed slot index any way
		// we can not let the failed unit block the whole sync process
		s.currentSyncedCompletedSlotIndex = &toBeSyncedUnit.slotIndex
		tilog.Info(ctx, "update the current synced completed slot index",
			zap.Int("slotIndex", toBeSyncedUnit.slotIndex),
			zap.Time("slotTime", toBeSyncedUnit.slotTime),
			zap.Int("currentSyncedCompletedSlotIndex", *s.currentSyncedCompletedSlotIndex))
	}
}

func mintueToSlotIndex(minute int) int {
	if minute < 0 {
		return minute + MinuteCountOfADay
	}
	return minute
}

// calculate the minutes to be synced, it's possible
// 1. minutes to be synced is all in one today
// 2. minutes to be synced is part of the minutes in yesterday, and the rest is in today, the return value will be like [..., -2, -1, 0, 1, 2, ...]
// 2.1 process restart, it's in back tracing mode, it may need to back trace to yesterday
// 2.2 normal sync, we need sync a range of minutes, part of the minutes in yesterday, and the rest is in today
func calculateToBeSyncedMinutes(
	ctx context.Context,
	currentSyncedCompletedSlotIndex *int,
	now time.Time,
	cfg *config.SyncBackupsConfig,
) []int {
	currentMinuteOfDay := now.Hour()*60 + now.Minute()
	targetCatchupMinuteOfDay := currentMinuteOfDay - 5

	var startSyncMinute int
	// set the back trace minutes if process restart
	if currentSyncedCompletedSlotIndex == nil {
		backTraceMinutes := cfg.TraceBackHours * 60
		startSyncMinute = targetCatchupMinuteOfDay - backTraceMinutes
		tilog.Info(ctx, "calculate to be synced minutes, currentSyncedCompletedSlotIndex is nil",
			zap.Int("startSyncMinute", startSyncMinute),
			zap.Int("targetCatchupMinuteOfDay", targetCatchupMinuteOfDay),
			zap.Int("backTraceMinutes", backTraceMinutes),
			zap.Time("now", now),
		)
	} else {
		startSyncMinute = *currentSyncedCompletedSlotIndex + 1
		// for cross day case, the currentSyncedCompletedSlotIndex will be greater than targetCatchupMinuteOfDay
		if startSyncMinute > targetCatchupMinuteOfDay {
			startSyncMinute -= MinuteCountOfADay
		}
		tilog.Info(ctx, "calculate to be synced minutes, currentSyncedCompletedSlotIndex is not nil",
			zap.Int("currentSyncedCompletedSlotIndex", *currentSyncedCompletedSlotIndex),
			zap.Int("startSyncMinute", startSyncMinute),
			zap.Int("targetCatchupMinuteOfDay", targetCatchupMinuteOfDay),
			zap.Time("now", now),
		)
	}

	minutesToBeSynced := make([]int, 0)
	for i := startSyncMinute; i <= targetCatchupMinuteOfDay; i++ {
		minutesToBeSynced = append(minutesToBeSynced, i)
	}
	return minutesToBeSynced
}

type toBeSyncedUnit struct {
	slotIndex          int
	slotTime           time.Time
	sharedPoolClusters map[uint64]map[uint64]bool // key: shared pool id, value: map[cluster id]bool
}

func (s *backupSyncSlots) calculateToBeSyncedUnit(
	ctx context.Context,
	minutesToBeSynced []int,
	now time.Time,
) []toBeSyncedUnit {
	toBeSyncedUnits := make([]toBeSyncedUnit, 0)
	zeroTimeOfToday := now.Truncate(time.Hour * 24)
	for _, minute := range minutesToBeSynced {
		slotIndex := mintueToSlotIndex(minute)
		slotTime := zeroTimeOfToday.Add(time.Duration(minute) * time.Minute)
		slot := s.slots[slotIndex]

		needSync := false
		sharedPoolClusters := make(map[uint64]map[uint64]bool)
		for sharedPoolId, clusterIds := range slot {
			for clusterId := range clusterIds {
				backupSetting := s.backupSettings[clusterId]
				if backupSetting == nil {
					tilog.Warn(
						ctx,
						"backup setting not found, there should be a bug or cluster already deleted",
						zap.Uint64("clusterId", clusterId),
					)
					continue
				}
				if checkIfBackupSettingIsSynced(backupSetting, slotTime) {
					continue
				}
				needSync = true
				if sharedPoolClusters[sharedPoolId] == nil {
					sharedPoolClusters[sharedPoolId] = make(map[uint64]bool)
				}
				sharedPoolClusters[sharedPoolId][clusterId] = true
			}
		}

		if !needSync {
			continue
		}

		toBeSyncedUnits = append(toBeSyncedUnits, toBeSyncedUnit{
			slotIndex:          slotIndex,
			slotTime:           slotTime,
			sharedPoolClusters: sharedPoolClusters,
		})
	}
	return toBeSyncedUnits
}

func checkIfBackupSettingIsSynced(backupSetting *entity.BackupSetting, slotTime time.Time) bool {
	// create time is after the slot time, there should not be any backup at that time, so the backup setting is synced
	// this avoid uneccessary sync for new created cluster when syncer is in back trace mode
	if backupSetting.CreatedAt.After(slotTime) {
		return true
	}
	if backupSetting.LastAutoBackupAt == nil {
		return false
	}
	return backupSetting.LastAutoBackupAt.After(slotTime) || backupSetting.LastAutoBackupAt.Equal(slotTime)
}

func (s *backupSyncSlots) syncOneUnit(ctx context.Context, toBeSyncedUnit *toBeSyncedUnit) {
	for sharedPoolId, clusterIds := range toBeSyncedUnit.sharedPoolClusters {
		// ListLightweightBackups is proxied to TiBR through cluster-gateway. TiBR
		// is unavailable while the pool is paused, so do not issue requests that
		// can only create error logs and missing metrics. After Resume, the missing
		// worker reconciles any pre-Pause slot because only the persisted Pause
		// interval itself is excluded.
		if s.pauseChecker != nil {
			inPauseWindow, err := s.pauseChecker(ctx, sharedPoolId)
			if err != nil {
				tilog.Error(ctx, "failed to determine shared pool pause state; skip backup sync",
					zap.Error(err), zap.Uint64("sharedPoolId", sharedPoolId))
				continue
			}
			if inPauseWindow {
				tilog.Info(ctx, "skip backup sync during shared pool pause window", zap.Uint64("sharedPoolId", sharedPoolId))
				continue
			}
		}
		// if the schedule changed after the slot time, it's hard to know what exactly the schedule is before changed.
		// so we only sync the clusters that schedule changed before the slot time.
		okClusterIds := map[uint64]bool{}
		for clusterId := range clusterIds {
			bs := s.backupSettings[clusterId]
			if bs != nil && bs.LastScheduleUpdatedAt.Before(toBeSyncedUnit.slotTime) {
				okClusterIds[clusterId] = true
				continue
			}
			tilog.Info(
				ctx,
				"cluster schedule changed after the slot time, will not sync cluster in this slot",
				zap.Uint64("clusterId", clusterId),
				zap.Time("lastScheduleUpdatedAt", bs.LastScheduleUpdatedAt),
				zap.Time("slotTime", toBeSyncedUnit.slotTime),
				zap.Int("slotIndex", toBeSyncedUnit.slotIndex),
			)
		}
		if len(okClusterIds) == 0 {
			tilog.Info(ctx, "no cluster to sync in shared pool", zap.Uint64("sharedPoolId", sharedPoolId),
				zap.Time("slotTime", toBeSyncedUnit.slotTime),
				zap.Int("slotIndex", toBeSyncedUnit.slotIndex),
			)
			continue
		}

		tilog.Info(ctx, "sync the shared pool", zap.Uint64("sharedPoolId", sharedPoolId),
			zap.Any("okClusterIds", lo.Keys(okClusterIds)),
			zap.Time("slotTime", toBeSyncedUnit.slotTime),
			zap.Int("slotIndex", toBeSyncedUnit.slotIndex),
		)

		resp, berr := s.fetchSharedPoolBackups(ctx, sharedPoolId, okClusterIds, toBeSyncedUnit)
		if berr != nil {
			NormalSyncErrorTotal.WithLabelValues(fmt.Sprintf("%d", sharedPoolId), LabelValue_NA, NormalSyncErrorType_FetchBackups).Inc()
			tilog.Error(ctx, "fetch shared pool backups failed", zap.Error(berr), zap.Uint64("sharedPoolId", sharedPoolId),
				zap.Time("slotTime", toBeSyncedUnit.slotTime),
				zap.Int("slotIndex", toBeSyncedUnit.slotIndex),
				zap.Any("okClusterIds", okClusterIds))
			continue
		}

		tilog.Info(ctx, "fetch backups success", zap.Uint64("sharedPoolId", sharedPoolId),
			zap.Time("slotTime", toBeSyncedUnit.slotTime),
			zap.Int("slotIndex", toBeSyncedUnit.slotIndex))

		serr := s.mapBackupsToSharedPoolClusters(ctx, resp, sharedPoolId, okClusterIds, toBeSyncedUnit.slotTime)
		if serr != nil {
			tilog.Error(ctx, "map backups to shared pool clusters failed",
				zap.Error(serr),
				zap.Uint64("sharedPoolId", sharedPoolId),
				zap.Time("slotTime", toBeSyncedUnit.slotTime),
				zap.Int("slotIndex", toBeSyncedUnit.slotIndex))
			continue
		}
	}
}

func (s *backupSyncSlots) fetchSharedPoolBackups(
	ctx context.Context,
	sharedPoolId uint64,
	clusterIds map[uint64]bool,
	toBeSyncedUnit *toBeSyncedUnit,
) (*clusterngv1beta.ListLightweightBackupsResponse, bizerr.BizError) {
	clusterId, _ := lo.First(lo.Keys(clusterIds))
	cluster, err := s.clusterRepo.Get(ctx, clusterId, true)
	if err != nil {
		tilog.Error(ctx, "get cluster failed", zap.Error(err), zap.Uint64("clusterId", clusterId))
		return nil, err.Chain("get cluster failed")
	}
	region, err := s.regionRepo.Get(ctx, cluster.RegionId)
	if err != nil {
		tilog.Error(ctx, "get region failed", zap.Error(err), zap.Uint64("clusterId", clusterId))
		return nil, err.Chain("get region failed")
	}

	tilog.Info(ctx, "sync the shared pool in unit",
		zap.Uint64("sharedPoolId", sharedPoolId),
		zap.Time("slotTime", toBeSyncedUnit.slotTime),
		zap.Int("slotIndex", toBeSyncedUnit.slotIndex))

	// make query to the regional cluster service to get the backups, retry failed request
	resp, err := s.regionalserverCli.ListLightweightBackups(ctx,
		region,
		&clusterngv1beta.ListLightweightBackupsRequest{
			SharedPoolId: fmt.Sprintf("%d", sharedPoolId),
			FromPrefix:   backupListFromPrefix(toBeSyncedUnit.slotTime),
			LoadDetails:  true,
			MaxCount:     backupListMaxCount,
		})
	if err != nil {
		tilog.Error(
			ctx,
			"list lightweight backups failed",
			zap.Error(err),
			zap.Uint64("sharedPoolId", sharedPoolId),
			zap.Time("slotTime", toBeSyncedUnit.slotTime),
			zap.Int("slotIndex", toBeSyncedUnit.slotIndex),
		)
		return nil, err.Chain("list lightweight backups failed")
	}
	return resp, nil
}

func (s *backupSyncSlots) mapBackupsToSharedPoolClusters(
	ctx context.Context,
	backupResp *clusterngv1beta.ListLightweightBackupsResponse,
	sharedPoolId uint64,
	clusterIds map[uint64]bool,
	slotTime time.Time,
) error {
	// calculate the backup and cluster mapping
	records, missingClusters := assembleBackupRecords(
		ctx,
		backupResp,
		clusterIds,
		s.backupSettings,
		slotTime,
	)
	if err := applyCurrentClusterEncryptionInfos(ctx, records, s.clusterRepo, s.backupEncryptionResolver); err != nil {
		return fmt.Errorf("apply current cluster encryption infos failed: %w", err)
	}

	if len(missingClusters) > 0 {
		for _, clusterId := range missingClusters {
			NormalSyncErrorTotal.WithLabelValues(
				fmt.Sprintf("%d", sharedPoolId),
				fmt.Sprintf("%d", clusterId),
				NormalSyncErrorType_BackupMissing,
			).Inc()
		}
		tilog.Error(
			ctx,
			"missing clusters",
			zap.Uint64("sharedPoolId", sharedPoolId),
			zap.Time("slotTime", slotTime),
			zap.Any("missingClusters", missingClusters),
		)
	}

	if len(records) == 0 {
		tilog.Info(ctx, "no backup to sync", zap.Uint64("sharedPoolId", sharedPoolId),
			zap.Time("slotTime", slotTime),
		)
		return nil
	}

	errs := s.saveBackupSyncResult(ctx, records)
	recordSaveBackupSyncResultMetrics(sharedPoolId, records, errs)
	if len(errs) > 0 {
		tilog.Error(ctx, "save backup sync result failed", zap.Any("errs", errs))
		return fmt.Errorf("save backup sync result failed: %v", errs)
	}

	return nil
}

func recordSaveBackupSyncResultMetrics(sharedPoolId uint64, toSaveRecords map[uint64]*entity.Backup, errs map[uint64]error) {
	for clusterId := range errs {
		NormalSyncErrorTotal.WithLabelValues(fmt.Sprintf("%d", sharedPoolId), fmt.Sprintf("%d", clusterId), NormalSyncErrorType_SaveBackup).Inc()
	}

	if errs == nil {
		errs = map[uint64]error{}
	}
	for clusterId := range toSaveRecords {
		if errs[clusterId] == nil {
			NormalSyncSuccessTotal.WithLabelValues(fmt.Sprintf("%d", sharedPoolId), fmt.Sprintf("%d", clusterId)).Inc()
		}
	}
}

func applyCurrentClusterEncryptionInfos(
	ctx context.Context,
	records map[uint64]*entity.Backup,
	clusterRepo clusterrepo.Repo,
	backupEncryptionResolver *brservice.Resolver,
) error {
	if len(records) == 0 {
		return nil
	}

	infos, err := resolveClusterEncryptionInfos(ctx, lo.Keys(records), clusterRepo, backupEncryptionResolver)
	if err != nil {
		return err
	}
	applyClusterEncryptionInfos(records, infos)
	return nil
}

func resolveClusterEncryptionInfos(
	ctx context.Context,
	clusterIds []uint64,
	clusterRepo clusterrepo.Repo,
	backupEncryptionResolver *brservice.Resolver,
) (map[uint64]backupClusterEncryptionInfo, error) {
	if len(clusterIds) == 0 || clusterRepo == nil || backupEncryptionResolver == nil {
		return map[uint64]backupClusterEncryptionInfo{}, nil
	}

	clusters, berr := clusterRepo.Query(ctx, &clusterrepo.Filter{
		ClusterIds:  clusterIds,
		ShowDeleted: true,
	})
	if berr != nil {
		return nil, fmt.Errorf("query clusters for backup encryption info failed: %w", berr)
	}

	clusterMap := make(map[uint64]*entity.Cluster, len(clusters))
	for _, cluster := range clusters {
		clusterMap[cluster.GetClusterId()] = cluster
	}

	snapshots, berr := backupEncryptionResolver.ResolveByClusters(ctx, clusterMap)
	if berr != nil {
		return nil, fmt.Errorf("resolve backup encryption info failed: %w", berr)
	}
	return snapshots, nil
}

func applyClusterEncryptionInfos(records map[uint64]*entity.Backup, infos map[uint64]backupClusterEncryptionInfo) {
	for clusterId, record := range records {
		if record == nil {
			continue
		}
		info, ok := infos[clusterId]
		if !ok {
			continue
		}
		record.CustomerManaged = info.CustomerManaged
		record.EncryptionKeyId = info.EncryptionKeyId
	}
}

func assembleBackupRecords(
	ctx context.Context,
	backupResp *clusterngv1beta.ListLightweightBackupsResponse,
	clusterIds map[uint64]bool,
	backupSettings map[uint64]*entity.BackupSetting,
	slotTime time.Time,
) (records map[uint64]*entity.Backup, missingClusters []uint64) {
	records = make(map[uint64]*entity.Backup)
	missingClusters = make([]uint64, 0)

	// sort kernel backup by time ASC
	slices.SortFunc(backupResp.Backups, func(a, b *clusterngv1beta.LightweightBackup) int {
		return cmp.Compare(a.BackupTs, b.BackupTs)
	})

	for _, backup := range backupResp.Backups {
		// make a map for quick lookup
		clusterBackupInfoMap := make(map[string]*clusterngv1beta.ClusterBackupInfo)
		for _, clusterBackupInfo := range backup.ClusterBackupInfo {
			clusterBackupInfoMap[clusterBackupInfo.ClusterId] = clusterBackupInfo
		}

		for clusterId := range clusterIds {
			// skip the cluster if the backup record already exists
			if records[clusterId] != nil {
				continue
			}

			clusterBackupInfo, ok := clusterBackupInfoMap[fmt.Sprintf("%d", clusterId)]
			if !ok {
				continue
			}
			backupSetting := backupSettings[clusterId]
			if backupSetting == nil {
				NormalSyncErrorTotal.WithLabelValues(LabelValue_NA, fmt.Sprintf("%d", clusterId), NormalSyncErrorType_BackupSettingNotFound).Inc()
				tilog.Error(ctx, "backup setting not found, there should be a bug", zap.Uint64("clusterId", clusterId))
				continue
			}

			// parse the backup time
			backupTime, err := parseBackupTime(backup.BackupTime)
			if err != nil {
				NormalSyncErrorTotal.WithLabelValues(LabelValue_NA, fmt.Sprintf("%d", clusterId), NormalSyncErrorType_InvalidBackupTime).Inc()
				tilog.Error(
					ctx,
					"parse backup time failed",
					zap.Error(err),
					zap.String("backupTime", backup.BackupTime),
				)
				continue
			}

			if !isBackupTimeMatchSlotTime(ctx, backupTime, slotTime, backupSetting) {
				continue
			}

			tilog.Info(ctx, "found matching backup for cluster",
				zap.Uint64("clusterId", clusterId),
				zap.String("backupTime", backup.BackupTime),
				zap.String("slotTime", slotTime.Format(time.RFC3339)),
				zap.String("backupName", backup.Name),
				zap.String("backupId", fmt.Sprintf("%d", backup.Id)))

			records[clusterId] = &entity.Backup{
				Backup: &schema.Backup{
					State:          schema.BackupStateSuccess,
					ExpirationTime: getBackupExpirationTime(backupTime, slotTime, backupSetting),

					Type:           schema.BackupTypeSnapshot,
					TriggerType:    schema.BackupTriggerTypeAuto,
					AutoBackupType: getAutoBackupType(slotTime, backupSetting),
					DisplayName:    fmt.Sprintf("auto-backup-%s", backupTime.Format("200601021504")),
					Description:    "",

					KernelBackupId:   backup.Id,
					KernelBackupName: backup.Name,
					BackupTime:       backupTime,
					BackupTs:         backup.BackupTs,
					BackupSizeBytes:  clusterBackupInfo.BackupSizeBytes,
					PdClusterId:      backupResp.PdClusterId,
					DfsPrefix:        backupResp.DfsPrefix,
					DfsBucket:        backupResp.DfsBucket,
					DfsRegion:        backupResp.DfsRegion,

					ClusterId:      clusterId,
					ClusterVersion: backupSetting.ClusterVersion,
					SharedPoolId:   backupSetting.SharedPoolId,
					TenantId:       backupSetting.TenantId,
					RegionId:       backupSetting.RegionId,
					CloudProvider:  backupSetting.CloudProvider,
					ServicePlan:    backupSetting.ServicePlan,
				},
			}
		}
	}

	// findout missing clusters
	for clusterId := range clusterIds {
		if records[clusterId] == nil {
			missingClusters = append(missingClusters, clusterId)
		}
	}

	// assemble missing backups as failed backups
	for _, clusterId := range missingClusters {
		backupSetting := backupSettings[clusterId]
		if backupSetting == nil {
			continue
		}
		records[clusterId] = &entity.Backup{
			Backup: &schema.Backup{
				State:          schema.BackupStateFailed,
				ExpirationTime: getBackupExpirationTime(slotTime, slotTime, backupSetting),

				Type:           schema.BackupTypeSnapshot,
				TriggerType:    schema.BackupTriggerTypeAuto,
				AutoBackupType: getAutoBackupType(slotTime, backupSetting),
				DisplayName:    fmt.Sprintf("auto-backup-%s", slotTime.Format("200601021504")),
				Description:    "backup failed",
				BackupTime:     slotTime,
				BackupTs:       uint64(slotTime.UnixMilli() << 18),

				ClusterId:      clusterId,
				ClusterVersion: backupSetting.ClusterVersion,
				SharedPoolId:   backupSetting.SharedPoolId,
				TenantId:       backupSetting.TenantId,
				RegionId:       backupSetting.RegionId,
				CloudProvider:  backupSetting.CloudProvider,
				ServicePlan:    backupSetting.ServicePlan,
			},
		}
	}

	return records, missingClusters
}

func getAutoBackupType(slotTime time.Time, backupSetting *entity.BackupSetting) schema.AutoBackupType {
	if backupSetting.Mode == schema.BackupModeFlexibleRetention || backupSetting.Schedule.Type == "per-day" {
		return schema.AutoBackupTypeDaily
	}
	if backupSetting.Schedule.Type != "per-hour" {
		return schema.AutoBackupTypeUnknown
	}
	if brservice.IsDailyBackupTime(slotTime, backupSetting.Schedule) {
		return schema.AutoBackupTypeDaily
	}
	return schema.AutoBackupTypeHourly
}

func parseBackupTime(value string) (time.Time, error) {
	// RFC3339Nano accepts both "2006-01-02T15:04:05Z" and "2006-01-02T15:04:05.000Z".
	return time.Parse(time.RFC3339Nano, value)
}

func getBackupExpirationTime(backupTime, slotTime time.Time, backupSetting *entity.BackupSetting) time.Time {
	if backupSetting.BackupPolicy == "" {
		return backupTime.Add(time.Duration(backupSetting.Schedule.RetentionDays) * 24 * time.Hour)
	}
	if getAutoBackupType(slotTime, backupSetting) == schema.AutoBackupTypeDaily {
		return backupTime.Add(time.Duration(backupSetting.Schedule.ZeroClockRetentionDays) * 24 * time.Hour)
	}
	return backupTime.Add(time.Duration(backupSetting.Schedule.NonZeroClockRetentionDays) * 24 * time.Hour)
}

func isBackupTimeMatchSlotTime(
	ctx context.Context,
	backupTime time.Time,
	slotTime time.Time,
	backupSetting *entity.BackupSetting,
) bool {
	slotTime = slotTime.UTC()
	backupTime = backupTime.UTC()
	switch backupSetting.Schedule.Type {
	case "per-day":
		return !backupTime.Before(slotTime) && backupTime.Before(slotTime.Add(24*time.Hour))
	case "per-hour":
		return !backupTime.Before(slotTime) && backupTime.Before(slotTime.Add(time.Hour))
	case "per-minute":
		if !backupTime.Truncate(time.Minute).Equal(slotTime.Truncate(time.Minute)) {
			return false
		}
	default:
		tilog.Error(
			ctx,
			"unknown backup schedule type",
			zap.String("scheduleType", backupSetting.Schedule.Type),
			zap.Uint64("clusterId", backupSetting.ClusterId),
		)
		return false
	}
	return true
}

func (s *backupSyncSlots) saveBackupSyncResult(ctx context.Context, records map[uint64]*entity.Backup) map[uint64]error /* cluster id -> error */ {
	errs := make(map[uint64]error)
	// insert the backup records into the database
	// update the backup settings into the database & memory,
	// make test stable
	clusterIds := lo.Keys(records)
	slices.Sort(clusterIds)
	for _, clusterId := range clusterIds {
		backup := records[clusterId]
		mutationSkipped := false
		backupSettingDeleted := false
		var finalWatermark *time.Time
		mutation := func(txCtx context.Context) error {
			// Only synthetic failures are derived from the scheduled slot. A real
			// Regional backup must remain ingestible even if its schedule changes
			// after the fetch completes.
			if backup.State == schema.BackupStateFailed && s.pauseFencer != nil {
				freshSetting, berr := s.backupSettingRepo.Get(txCtx, clusterId)
				if berr != nil {
					return berr.Chain("load latest backup setting before backup sync")
				}
				if !freshSetting.LastScheduleUpdatedAt.Before(backup.BackupTime) {
					mutationSkipped = true
					return nil
				}
			}
			// insert the backup record into the database
			err := s.backupRepo.Create(txCtx, backup)
			if err != nil {
				tilog.Error(ctx, "create backup record failed", zap.Error(err),
					zap.Uint64("clusterId", clusterId),
					zap.Any("backup", backup))
				return err.Chain("create backup record failed")
			}

			backupSetting := s.backupSettings[clusterId]
			if backupSetting == nil {
				tilog.Error(ctx, "backup setting not found", zap.Uint64("clusterId", clusterId))
				return errors.New("backup setting not found")
			}

			// Always reconcile through the database's atomic max operation. The
			// in-process cache is not authoritative and may lag after worker
			// handoff or a failed transaction.
			updatedWatermark, updateErr := s.backupSettingRepo.UpdateLastAutoBackupAt(
				txCtx,
				clusterId,
				&backup.BackupTime,
			)
			if updateErr != nil {
				if updateErr.ErrCode() == bizerr.ErrCodeRecordNotFound {
					// The backup setting can be deleted after this process built its
					// local slots. Return the error to roll back backupRepo.Create;
					// the caller handles this expected race without emitting a save
					// failure metric.
					backupSettingDeleted = true
					return updateErr
				}
				tilog.Error(
					ctx,
					"update LastAutoBackupAt in database failed",
					zap.Error(updateErr),
					zap.Uint64("clusterId", clusterId),
				)
				return updateErr.Chain("update LastAutoBackupAt in database failed")
			}
			finalWatermark = &updatedWatermark

			s.submitBackupCompleteAuditLog(ctx, backup)
			return nil
		}
		var err error
		switch {
		// A successful record represents a real backup that already exists in
		// Regional. Keep ingesting that immutable fact during Pause preparation;
		// it is unrelated to the operation-owned manual backups selected by
		// pause_operation_id. Only synthetic failed-slot records need the Pause
		// fence because the worker creates those records locally.
		case backup.State == schema.BackupStateFailed && s.pauseFencer != nil:
			var pauseSkipped bool
			pauseSkipped, err = s.pauseFencer(ctx, backup.SharedPoolId, mutation)
			mutationSkipped = mutationSkipped || pauseSkipped
		case s.transaction != nil:
			err = s.transaction.Transaction(ctx, mutation)
		default:
			err = mutation(ctx)
		}
		if err != nil {
			if backupSettingDeleted {
				s.removeClusterWithoutLock(clusterId)
				delete(records, clusterId)
				tilog.Info(ctx, "skip backup sync because backup setting was deleted",
					zap.Uint64("clusterId", clusterId),
					zap.Uint64("sharedPoolId", backup.SharedPoolId),
					zap.Time("backupTime", backup.BackupTime))
				continue
			}
			tilog.Error(
				ctx,
				"insert backup and update backup setting failed",
				zap.Error(err),
				zap.Uint64("clusterId", clusterId),
			)
			errs[clusterId] = err
			continue
		}
		if !mutationSkipped && finalWatermark != nil {
			s.backupSettings[clusterId].LastAutoBackupAt = finalWatermark
		}
		if mutationSkipped {
			tilog.Info(ctx, "skip synthetic backup result after schedule change or Pause admission",
				zap.Uint64("clusterId", clusterId),
				zap.Uint64("sharedPoolId", backup.SharedPoolId),
				zap.Time("backupTime", backup.BackupTime))
		}
	}
	if len(errs) > 0 {
		return errs
	}
	return nil
}

func (s *backupSyncSlots) submitBackupCompleteAuditLog(ctx context.Context, backup *entity.Backup) {
	// should not let panic to affect the main logic
	defer func() {
		if r := recover(); r != nil {
			tilog.Error(ctx, "panic when submit audit log", zap.Any("panic", r), zap.Any("backup", backup))
			NormalSyncErrorTotal.WithLabelValues(LabelValue_NA, fmt.Sprintf("%d", backup.ClusterId), NormalSyncErrorType_SubmitAuditLog).Inc()
		}
	}()

	logDetails := auditlog.Details{
		auditlog.KeyTierType: "premium",
		"backup_id":          backup.ID,
		"backup_name":        backup.DisplayName,
		"trigger_type":       string(backup.TriggerType),
		"starts_at":          backup.BackupTime,
		"ends_at":            backup.BackupTime,
	}
	success := backup.State == schema.BackupStateSuccess
	if !success {
		logDetails[auditlog.KeyError] = backup.Description
	}
	event := auditlog.NewEvent(ctx, auditlog.EventBackupCompleted).
		SetResult(success).
		SetOrgID(backup.TenantId).
		SetClusterID(backup.ClusterId).
		SetDetails(logDetails.Build()).
		SetOperatorIP("-").
		SetOperatorID(1).
		SetOperatorEmail("system").
		SetOperatorLoginMethod("-").
		SetOperatorType(auditlog.OperatorTypeSystem)
	if err := event.Submit(); err != nil {
		NormalSyncErrorTotal.WithLabelValues(LabelValue_NA, fmt.Sprintf("%d", backup.ClusterId), NormalSyncErrorType_SubmitAuditLog).Inc()
		tilog.Error(ctx, "failed to submit audit log", zap.Error(err), zap.Any("backup", backup))
	}
}

func (s *backupSyncSlots) update(ctx context.Context) error {
	// Notice: keep this line here, don't let it after s.lock.Lock(),  to avoid waiting for lock too long
	//  cause some backup setting changes may be missed
	recentTime := time.Now().UTC().Add(-2 * s.cfg.LoadScheduleChangeInterval)

	s.lock.Lock()
	defer s.lock.Unlock()

	if !s.isInit {
		tilog.Info(ctx, "skip update backup sync slots, backup sync slots is not initialized")
		return nil
	}

	recentlyScheduleChanged, err := s.backupSettingRepo.UnscopedQuery(ctx, &brrepo.QueryBackupSettingFilter{
		States:                   []schema.BackupSettingState{schema.BackupSettingStateCreated},
		LastScheduleUpdatedAtGte: &recentTime,
	}, nil)
	if err != nil {
		tilog.Error(ctx, "list recently schedule changed backup settings failed", zap.Error(err))
		return fmt.Errorf("list recently schedule changed backup settings failed: %w", err)
	}

	if len(recentlyScheduleChanged) > 0 {
		tilog.Info(
			ctx,
			"there are recently schedule changed backup settings, will update the slots indexes",
			zap.Any("backupSettings", recentlyScheduleChanged),
		)
		for _, backupSetting := range recentlyScheduleChanged {
			tilog.Info(ctx, "one backup setting is schedule changed recently, update the slots indexes",
				zap.Uint64("clusterId", backupSetting.ClusterId),
				zap.Any("backupSetting", backupSetting))
			s.backupSettings[backupSetting.ClusterId] = backupSetting
			// remove old indexes if exists
			oldIndexes := s.clusterSlots[backupSetting.ClusterId]
			if len(oldIndexes) != 0 {
				tilog.Info(ctx, "there is a old backup setting, remove old indexes",
					zap.Uint64("clusterId", backupSetting.ClusterId),
					zap.Any("oldIndexes", oldIndexes))
				for _, index := range oldIndexes {
					if s.slots[index] != nil || s.slots[index][backupSetting.SharedPoolId] != nil {
						delete(s.slots[index][backupSetting.SharedPoolId], backupSetting.ClusterId)
					}
				}
			}

			// add new indexes
			indexes := calculateSlotsIndexes(backupSetting)
			s.clusterSlots[backupSetting.ClusterId] = indexes
			for _, index := range indexes {
				putSlots(s.slots, index, backupSetting.SharedPoolId, backupSetting.ClusterId)
			}
			tilog.Info(ctx, "one backup setting is schedule changed recently, update the slots indexes success",
				zap.Uint64("clusterId", backupSetting.ClusterId),
				zap.Any("backupSetting", backupSetting),
				zap.Any("newIndexes", indexes))
		}
	}

	// Query every cluster held by this process's cache instead of only a recent
	// deleted_at window. Lease handoff and transient refresh failures can leave a
	// replica unused for longer than that window.
	cachedClusterIds := lo.Keys(s.backupSettings)
	deletedCachedSettings := make([]*entity.BackupSetting, 0)
	for _, clusterIds := range lo.Chunk(cachedClusterIds, deletedSettingReconciliationBatchSize) {
		deletedInBatch, queryErr := s.backupSettingRepo.UnscopedQuery(ctx, &brrepo.QueryBackupSettingFilter{
			ClusterIds: clusterIds,
			States:     schema.BackupSettingStatesForDeletedCluster,
		}, nil)
		if queryErr != nil {
			tilog.Error(ctx, "list deleted cached backup settings failed", zap.Error(queryErr))
			return fmt.Errorf("list deleted cached backup settings failed: %w", queryErr)
		}
		deletedCachedSettings = append(deletedCachedSettings, deletedInBatch...)
	}

	if len(deletedCachedSettings) > 0 {
		tilog.Info(
			ctx,
			"there are deleted cached backup settings, will remove them from backup sync slots",
			zap.Any("backupSettings", deletedCachedSettings),
		)
		for _, backupSetting := range deletedCachedSettings {
			tilog.Info(
				ctx,
				"remove deleted backup setting from backup sync slots",
				zap.Uint64("clusterId", backupSetting.ClusterId),
				zap.Any("backupSetting", backupSetting),
			)

			oldIndexes := s.clusterSlots[backupSetting.ClusterId]
			s.removeClusterWithoutLock(backupSetting.ClusterId)
			tilog.Info(
				ctx,
				"removed deleted backup setting from backup sync slots",
				zap.Uint64("clusterId", backupSetting.ClusterId),
				zap.Any("backupSetting", backupSetting),
				zap.Any("oldIndexes", oldIndexes),
			)
		}
	}

	return nil
}

// removeClusterWithoutLock removes a cluster from every process-local backup
// sync index. The caller must hold s.lock when the worker is running.
func (s *backupSyncSlots) removeClusterWithoutLock(clusterId uint64) {
	for _, index := range s.clusterSlots[clusterId] {
		for _, clusterIds := range s.slots[index] {
			delete(clusterIds, clusterId)
		}
	}
	delete(s.backupSettings, clusterId)
	delete(s.clusterSlots, clusterId)
}

func (s *backupSyncSlots) removeCluster(clusterId uint64) {
	s.lock.Lock()
	defer s.lock.Unlock()
	s.removeClusterWithoutLock(clusterId)
}

func (s *backupSyncSlots) copyBackupSettings() []*entity.BackupSetting {
	s.lock.Lock()
	defer s.lock.Unlock()
	backupSettings := make([]*entity.BackupSetting, 0, len(s.backupSettings))
	for _, backupSetting := range s.backupSettings {
		backupSettings = append(backupSettings, backupSetting)
	}
	return backupSettings
}