package service

import (
	"context"
	"encoding/base64"
	"fmt"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/tidbcloud/pkgv2/bizerr"
	pkgv2 "github.com/tidbcloud/pkgv2/clusterproxy/serverless"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/vo"
	"go.uber.org/zap"
)

const (
	virtualBackupIDPrefix             = constant.BackupIDPrefix + "v1-"
	tombstoneClusterQueryBatchSize    = 500
	physicalBackupSlotQueryBatchSize  = 500
	physicalBackupScopeQueryBatchSize = 500
	virtualBackupNotReadyMessage      = "This backup is not available yet. Please try again later."
)

// noCutover is the zero value used for the cutover parameter when listing
// virtual backups from already-cutover-aware history records. History valid_from
// is computed relative to the real cutover at archive/append time, so an online
// listing path does not need to pass the cutover again.
var noCutover = time.Time{}

type virtualBackupRef struct {
	clusterID    string
	scheduleTime time.Time
}

// VirtualBackup is a scheduled restore point derived from schedule history.
// The restore point itself is never persisted; Physical is an optional shared
// Kubernetes-level slot index row used only after readiness resolution.
type VirtualBackup struct {
	ID            string
	TenantID      string
	ClusterID     string
	Region        string
	ScheduleTime  time.Time
	ExpireTime    time.Time
	Retention     time.Duration
	CloudProvider string
	Kubernetes    string
	Physical      *entity.PhysicalBackupSlot
}

func generateVirtualBackupID(clusterID string, scheduleTime time.Time) string {
	payload := fmt.Sprintf("%s:%d", clusterID, scheduleTime.UTC().Unix())
	return virtualBackupIDPrefix + base64.RawURLEncoding.EncodeToString([]byte(payload))
}

// parseVirtualBackupID parses a virtual backup ID. It returns (nil, false, false)
// for a legacy backup ID, (ref, true, false) for a valid virtual ID, and
// (nil, false, true) for an ID that has the virtual prefix but is malformed.
func parseVirtualBackupID(backupID string) (ref *virtualBackupRef, ok, malformed bool) {
	if !strings.HasPrefix(backupID, virtualBackupIDPrefix) {
		return nil, false, false
	}
	payload, err := base64.RawURLEncoding.DecodeString(strings.TrimPrefix(backupID, virtualBackupIDPrefix))
	if err != nil {
		return nil, false, true
	}
	parts := strings.Split(string(payload), ":")
	if len(parts) != 2 {
		return nil, false, true
	}
	clusterID, err := strconv.ParseUint(parts[0], 10, 64)
	if err != nil || strconv.FormatUint(clusterID, 10) != parts[0] {
		return nil, false, true
	}
	unix, err := strconv.ParseInt(parts[1], 10, 64)
	if err != nil {
		return nil, false, true
	}
	ref = &virtualBackupRef{
		clusterID:    strconv.FormatUint(clusterID, 10),
		scheduleTime: time.Unix(unix, 0).UTC(),
	}
	if generateVirtualBackupID(ref.clusterID, ref.scheduleTime) != backupID {
		return nil, false, true
	}
	return ref, true, false
}

func (b *backupRestoreService) listVirtualBackups(
	ctx context.Context,
	clusterID, tenantID string,
	now time.Time,
) ([]*VirtualBackup, map[string]struct{}, bizerr.BizError) {
	if clusterID == "" {
		return nil, nil, bizerr.New(bizerr.ErrCodeParams, "missing cluster id")
	}
	tombstones, err := b.listTombstones(ctx, []string{clusterID})
	if err != nil {
		return nil, nil, err
	}
	tombstoneRestorePoints := restorePointKeysFromTombstones(tombstones)
	histories, err := b.backupHistoryRepo.List(ctx, &entity.ListBackupScheduleHistoryOptions{
		ClusterId:          clusterID,
		TenantId:           tenantID,
		ValidBeforeOrEqual: &now,
	})
	if err != nil {
		return nil, nil, err
	}
	histories, err = b.appendCurrentScheduleHistory(ctx, clusterID, histories)
	if err != nil {
		return nil, nil, err
	}
	backups, err := b.buildAvailableVirtualBackupsFromHistories(
		ctx, histories, now, noCutover, tombstoneRestorePoints,
	)
	return backups, tombstoneRestorePoints, err
}

func (b *backupRestoreService) appendCurrentScheduleHistory(
	ctx context.Context,
	clusterID string,
	histories []*entity.BackupScheduleHistory,
) ([]*entity.BackupScheduleHistory, bizerr.BizError) {
	schedules, err := b.backupScheduleRepo.List(ctx, entity.ListBackupScheduleOptions{ClusterId: clusterID})
	if err != nil {
		return nil, err
	}
	if len(schedules) == 0 || schedules[0].IsPending() {
		return histories, nil
	}
	if len(schedules) > 1 {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "multiple active backup schedules found")
	}

	cutoverAt, err := b.backupHistoryRepo.GetVirtualizationCutover(ctx)
	if err != nil {
		return nil, err
	}
	history := entity.NewBackupScheduleHistoryFromSchedule(schedules[0], cutoverAt, histories, nil)
	return append(histories, history), nil
}

func (b *backupRestoreService) listAvailableBackups(
	ctx context.Context,
	options *entity.ListUserBackupOptions,
	now time.Time,
	orderBy string,
	offset int64,
	pageSize int32,
) ([]*vo.Backup, int, bizerr.BizError) {
	persistedBackups, err := b.backupRepo.List(ctx, options)
	if err != nil {
		return nil, 0, err
	}
	virtualBackups, tombstoneRestorePoints, err := b.listVirtualBackups(
		ctx, options.ClusterId, options.TenantId, now,
	)
	if err != nil {
		return nil, 0, err
	}
	persistedBackups = suppressPersistedBackupsWithTombstones(
		persistedBackups,
		tombstoneRestorePoints,
	)
	virtualBackups = suppressUnreadyVirtualBackupsWithPersistedSchedule(persistedBackups, virtualBackups)
	persistedBackups = suppressPersistedBackupsWithVirtualSchedule(persistedBackups, virtualBackups)
	items := make([]*vo.Backup, 0, len(persistedBackups)+len(virtualBackups))
	for _, backup := range persistedBackups {
		items = append(items, vo.BackupFromPersisted(backup))
	}
	for _, backup := range virtualBackups {
		items = append(items, backup.ToVO())
	}
	sortBackupViews(items, orderBy)
	total := len(items)
	return paginateBackupViews(items, offset, pageSize), total, nil
}

func (b *backupRestoreService) hydrateVirtualBackups(
	ctx context.Context,
	backups []*VirtualBackup,
	now time.Time,
) ([]*VirtualBackup, bizerr.BizError) {
	if b.physicalBackupSlotRepo == nil {
		available := make([]*VirtualBackup, 0, len(backups))
		for _, backup := range backups {
			if backup != nil && !backup.ExpireTime.Before(now) {
				available = append(available, backup)
			}
		}
		return available, nil
	}
	keysByID := make(map[string]entity.PhysicalBackupSlotKey, len(backups))
	keys := make([]entity.PhysicalBackupSlotKey, 0, len(backups))
	for _, backup := range backups {
		if backup == nil {
			continue
		}
		key := physicalBackupSlotKey(backup.CloudProvider, backup.Region, backup.Kubernetes, backup.ScheduleTime)
		keyID := physicalBackupSlotKeyID(key)
		if _, ok := keysByID[keyID]; ok {
			continue
		}
		keysByID[keyID] = key
		keys = append(keys, key)
	}
	slotsByID := make(map[string]*entity.PhysicalBackupSlot, len(keys))
	for start := 0; start < len(keys); start += physicalBackupSlotQueryBatchSize {
		end := start + physicalBackupSlotQueryBatchSize
		if end > len(keys) {
			end = len(keys)
		}
		slots, err := b.physicalBackupSlotRepo.List(ctx, keys[start:end])
		if err != nil {
			return nil, err
		}
		for _, slot := range slots {
			if slot == nil {
				continue
			}
			slotsByID[physicalBackupSlotKeyID(physicalBackupSlotKey(slot.CloudProvider, slot.Region, slot.Kubernetes, slot.SlotTime))] = slot
		}
	}
	resolved := make([]*VirtualBackup, 0, len(backups))
	for _, backup := range backups {
		if backup == nil {
			continue
		}
		slot, ok := slotsByID[physicalBackupSlotKeyID(physicalBackupSlotKey(backup.CloudProvider, backup.Region, backup.Kubernetes, backup.ScheduleTime))]
		if !ok {
			continue
		}
		backup.Physical = slot
		backup.ExpireTime = slot.BackupTime.UTC().Add(backup.Retention)
		if backup.ExpireTime.Before(now) {
			continue
		}
		resolved = append(resolved, backup)
	}
	return resolved, nil
}

func physicalBackupSlotKey(provider, region, kubernetes string, slotTime time.Time) entity.PhysicalBackupSlotKey {
	return entity.PhysicalBackupSlotKey{
		CloudProvider: provider,
		Region:        region,
		Kubernetes:    kubernetes,
		SlotTime:      slotTime.UTC(),
	}
}

func physicalBackupSlotKeyID(key entity.PhysicalBackupSlotKey) string {
	return key.CloudProvider + "\x00" + key.Region + "\x00" + key.Kubernetes + "\x00" + strconv.FormatInt(
		key.SlotTime.UTC().UnixNano(),
		10,
	)
}

func physicalBackupSlotScope(history *entity.BackupScheduleHistory) entity.PhysicalBackupSlotScope {
	return entity.PhysicalBackupSlotScope{
		CloudProvider: history.GetCloudProvider(),
		Region:        history.Region,
		Kubernetes:    history.GetKubernetes(),
	}
}

func physicalBackupSlotScopeID(scope entity.PhysicalBackupSlotScope) string {
	return scope.CloudProvider + "\x00" + scope.Region + "\x00" + scope.Kubernetes
}

func (b *backupRestoreService) buildVirtualBackupsFromHistories(
	ctx context.Context,
	histories []*entity.BackupScheduleHistory,
	now time.Time,
	cutover time.Time,
) ([]*VirtualBackup, bizerr.BizError) {
	if len(histories) == 0 {
		return []*VirtualBackup{}, nil
	}
	tombstones, err := b.listTombstones(ctx, clusterIDsFromHistories(histories, nil))
	if err != nil {
		return nil, err
	}
	return b.buildAvailableVirtualBackupsFromHistories(
		ctx,
		histories,
		now,
		cutover,
		restorePointKeysFromTombstones(tombstones),
	)
}

func (b *backupRestoreService) buildAvailableVirtualBackupsFromHistories(
	ctx context.Context,
	histories []*entity.BackupScheduleHistory,
	now time.Time,
	cutover time.Time,
	tombstoneRestorePoints map[string]struct{},
) ([]*VirtualBackup, bizerr.BizError) {
	regular := buildVirtualBackups(
		ctx, histories, now, cutover, tombstoneRestorePoints, 0,
	)
	regular, err := b.hydrateVirtualBackups(ctx, regular, now)
	if err != nil {
		return nil, err
	}
	delayed, err := b.buildDelayedVirtualBackupsFromHistories(
		ctx, histories, now, cutover, tombstoneRestorePoints, false,
	)
	if err != nil {
		return nil, err
	}
	return mergeVirtualBackups(regular, delayed), nil
}

func (b *backupRestoreService) listRecentPhysicalBackupSlots(
	ctx context.Context,
	histories []*entity.BackupScheduleHistory,
	now time.Time,
) (map[string][]*entity.PhysicalBackupSlot, bizerr.BizError) {
	result := make(map[string][]*entity.PhysicalBackupSlot)
	if b.physicalBackupSlotRepo == nil {
		return result, nil
	}
	scopes := make([]entity.PhysicalBackupSlotScope, 0)
	seenScopes := make(map[string]struct{})
	maxRetention := time.Duration(0)
	for _, history := range histories {
		if history == nil || history.ValidFrom == nil || history.Kubernetes == constant.KubernetesUnknown {
			continue
		}
		retention := time.Duration(history.Retention) * time.Hour
		if retention > maxRetention {
			maxRetention = retention
		}
		scope := physicalBackupSlotScope(history)
		scopeID := physicalBackupSlotScopeID(scope)
		if _, ok := seenScopes[scopeID]; ok {
			continue
		}
		seenScopes[scopeID] = struct{}{}
		scopes = append(scopes, scope)
	}
	if len(scopes) == 0 || maxRetention <= 0 {
		return result, nil
	}
	backupTimeGte := now.Add(-maxRetention).UTC()
	for start := 0; start < len(scopes); start += physicalBackupScopeQueryBatchSize {
		end := start + physicalBackupScopeQueryBatchSize
		if end > len(scopes) {
			end = len(scopes)
		}
		slots, err := b.physicalBackupSlotRepo.ListByBackupTime(ctx, scopes[start:end], backupTimeGte)
		if err != nil {
			return nil, err
		}
		for _, slot := range slots {
			if slot == nil {
				continue
			}
			scopeID := physicalBackupSlotScopeID(entity.PhysicalBackupSlotScope{
				CloudProvider: slot.CloudProvider,
				Region:        slot.Region,
				Kubernetes:    slot.Kubernetes,
			})
			result[scopeID] = append(result[scopeID], slot)
		}
	}
	return result, nil
}

func (b *backupRestoreService) buildDelayedVirtualBackupsFromHistories(
	ctx context.Context,
	histories []*entity.BackupScheduleHistory,
	now time.Time,
	cutover time.Time,
	tombstoneRestorePoints map[string]struct{},
	onePerCluster bool,
) ([]*VirtualBackup, bizerr.BizError) {
	if tombstoneRestorePoints == nil {
		tombstoneRestorePoints = map[string]struct{}{}
	}
	slotsByScope, err := b.listRecentPhysicalBackupSlots(ctx, histories, now)
	if err != nil {
		return nil, err
	}
	backups := make([]*VirtualBackup, 0)
	seenBackups := make(map[string]struct{})
	seenClusters := make(map[string]struct{})
	for _, history := range histories {
		if history == nil || history.ValidFrom == nil || history.Kubernetes == constant.KubernetesUnknown {
			continue
		}
		if onePerCluster {
			if _, ok := seenClusters[history.ClusterID]; ok {
				continue
			}
		}
		retention := time.Duration(history.Retention) * time.Hour
		regularStart := now.Add(-retention).UTC()
		effectiveFrom := history.EffectiveFrom(cutover)
		effectiveEnd := history.EffectiveEnd(now)
		for _, slot := range slotsByScope[physicalBackupSlotScopeID(physicalBackupSlotScope(history))] {
			if slot == nil || !slot.SlotTime.UTC().Before(regularStart) {
				continue
			}
			scheduleTime := slot.SlotTime.UTC()
			if scheduleTime.Before(effectiveFrom) || scheduleTime.After(effectiveEnd) {
				continue
			}
			if !historyCoversScheduleTime(history, scheduleTime) {
				continue
			}
			expireTime := slot.BackupTime.UTC().Add(retention)
			if expireTime.Before(now) {
				continue
			}
			if _, deleted := tombstoneRestorePoints[backupScheduleKey(history.ClusterID, scheduleTime)]; deleted {
				continue
			}
			backupID := generateVirtualBackupID(history.ClusterID, scheduleTime)
			if _, ok := seenBackups[backupID]; ok {
				continue
			}
			seenBackups[backupID] = struct{}{}
			seenClusters[history.ClusterID] = struct{}{}
			backups = append(backups, &VirtualBackup{
				ID:            backupID,
				TenantID:      history.TenantID,
				ClusterID:     history.ClusterID,
				Region:        history.Region,
				ScheduleTime:  scheduleTime,
				ExpireTime:    expireTime,
				Retention:     retention,
				CloudProvider: history.GetCloudProvider(),
				Kubernetes:    history.GetKubernetes(),
				Physical:      slot,
			})
			if onePerCluster {
				break
			}
		}
	}
	return backups, nil
}

func mergeVirtualBackups(groups ...[]*VirtualBackup) []*VirtualBackup {
	seen := make(map[string]struct{})
	result := make([]*VirtualBackup, 0)
	for _, group := range groups {
		for _, backup := range group {
			if backup == nil {
				continue
			}
			if _, ok := seen[backup.ID]; ok {
				continue
			}
			seen[backup.ID] = struct{}{}
			result = append(result, backup)
		}
	}
	return result
}

func (b *backupRestoreService) listTombstones(
	ctx context.Context,
	clusterIDs []string,
) ([]*entity.BackupTombstone, bizerr.BizError) {
	if len(clusterIDs) == 0 {
		return []*entity.BackupTombstone{}, nil
	}
	if len(clusterIDs) == 1 {
		tombstones, err := b.backupTombstoneRepo.List(
			ctx,
			entity.ListBackupTombstoneOptions{ClusterId: clusterIDs[0]},
		)
		if err != nil {
			return nil, err
		}
		return tombstones, nil
	}

	allTombstones := make([]*entity.BackupTombstone, 0)
	for start := 0; start < len(clusterIDs); start += tombstoneClusterQueryBatchSize {
		end := start + tombstoneClusterQueryBatchSize
		if end > len(clusterIDs) {
			end = len(clusterIDs)
		}
		tombstones, err := b.backupTombstoneRepo.List(ctx, entity.ListBackupTombstoneOptions{
			ClustersIn: clusterIDs[start:end],
		})
		if err != nil {
			return nil, err
		}
		allTombstones = append(allTombstones, tombstones...)
	}
	return allTombstones, nil
}

func restorePointKeysFromTombstones(tombstones []*entity.BackupTombstone) map[string]struct{} {
	keys := make(map[string]struct{}, len(tombstones))
	for _, tombstone := range tombstones {
		if tombstone != nil && tombstone.ScheduleTime != nil {
			keys[backupScheduleKey(tombstone.ClusterID, *tombstone.ScheduleTime)] = struct{}{}
		}
	}
	return keys
}

func buildVirtualBackups(
	ctx context.Context,
	histories []*entity.BackupScheduleHistory,
	now time.Time,
	cutover time.Time,
	tombstoneRestorePoints map[string]struct{},
	limit int,
) []*VirtualBackup {
	if tombstoneRestorePoints == nil {
		tombstoneRestorePoints = map[string]struct{}{}
	}
	backups := make([]*VirtualBackup, 0)
	seen := make(map[string]struct{})
	for _, history := range histories {
		if history == nil {
			continue
		}
		if history.ValidFrom == nil || history.Kubernetes == constant.KubernetesUnknown {
			if history.Kubernetes == constant.KubernetesUnknown {
				log.WithContextL(ctx).Warn("skip virtual backup with unresolved kubernetes",
					zap.String("cluster_id", history.ClusterID))
			}
			continue
		}
		retention := time.Duration(history.Retention) * time.Hour
		start := history.EffectiveFrom(cutover)
		if retentionStart := now.Add(-retention).UTC(); retentionStart.After(start) {
			start = retentionStart
		}
		end := history.EffectiveEnd(now)
		if start.After(end) {
			continue
		}
		scheduleTimes, parseErr := enumerateScheduleTimes(history.ScheduleCron, start, end)
		if parseErr != nil {
			log.WithContextL(ctx).Warn(
				"skip invalid backup schedule history cron",
				zap.String("cluster_id", history.ClusterID),
				zap.String("schedule_cron", history.ScheduleCron),
				zap.Error(parseErr),
			)
			continue
		}
		for _, scheduleTime := range scheduleTimes {
			if !history.Contains(scheduleTime) {
				continue
			}
			expireTime := scheduleTime.Add(retention)
			if expireTime.Before(now) {
				continue
			}
			backupID := generateVirtualBackupID(history.ClusterID, scheduleTime)
			if _, ok := tombstoneRestorePoints[backupScheduleKey(history.ClusterID, scheduleTime)]; ok {
				continue
			}
			if _, ok := seen[backupID]; ok {
				continue
			}
			seen[backupID] = struct{}{}
			backups = append(backups, &VirtualBackup{
				ID:            backupID,
				TenantID:      history.TenantID,
				ClusterID:     history.ClusterID,
				Region:        history.Region,
				ScheduleTime:  scheduleTime,
				ExpireTime:    expireTime,
				Retention:     retention,
				CloudProvider: history.GetCloudProvider(),
				Kubernetes:    history.GetKubernetes(),
			})
			if limit > 0 && len(backups) >= limit {
				return backups
			}
		}
	}
	return backups
}

func enumerateScheduleTimes(scheduleCron string, start, end time.Time) ([]time.Time, error) {
	expr, err := constant.BackupCronParse.Parse(scheduleCron)
	if err != nil {
		return nil, err
	}
	times := make([]time.Time, 0)
	next := expr.Next(start.Add(-time.Second).UTC()).UTC()
	for !next.After(end.UTC()) {
		times = append(times, next)
		next = expr.Next(next).UTC()
	}
	return times, nil
}

func (b *VirtualBackup) ToVO() *vo.Backup {
	createdAt := b.ScheduleTime
	backupTime := b.ScheduleTime
	if b.Physical != nil {
		backupTime = b.Physical.BackupTime
	}
	expireTime := b.ExpireTime
	view := &vo.Backup{
		ID:         b.ID,
		CreatedAt:  &createdAt,
		ClusterID:  b.ClusterID,
		ExpireTime: &expireTime,
		BackupTime: &backupTime,
		State:      constant.BackupStateSucceeded,
		Type:       string(constant.BackupTypeAuto),
	}
	view.GenerateBackupName()
	return view
}

func sortBackupViews(items []*vo.Backup, orderBy string) {
	desc := strings.Contains(strings.ToLower(orderBy), "desc")
	sort.SliceStable(items, func(i, j int) bool {
		iTime := time.Time{}
		jTime := time.Time{}
		if items[i] != nil && items[i].BackupTime != nil {
			iTime = items[i].BackupTime.UTC()
		}
		if items[j] != nil && items[j].BackupTime != nil {
			jTime = items[j].BackupTime.UTC()
		}
		if desc {
			return iTime.After(jTime)
		}
		return iTime.Before(jTime)
	})
}

func paginateBackupViews(items []*vo.Backup, offset int64, pageSize int32) []*vo.Backup {
	if offset >= int64(len(items)) {
		return []*vo.Backup{}
	}
	end := offset + int64(pageSize)
	if pageSize <= 0 || end > int64(len(items)) {
		end = int64(len(items))
	}
	return items[offset:end]
}

func suppressPersistedBackupsWithVirtualSchedule(
	persistedBackups []*entity.UserBackup,
	virtualBackups []*VirtualBackup,
) []*entity.UserBackup {
	if len(persistedBackups) == 0 || len(virtualBackups) == 0 {
		return persistedBackups
	}
	virtualSchedules := make(map[string]struct{}, len(virtualBackups))
	for _, backup := range virtualBackups {
		if backup == nil || backup.Physical == nil {
			continue
		}
		virtualSchedules[backupScheduleKey(backup.ClusterID, backup.ScheduleTime)] = struct{}{}
	}
	if len(virtualSchedules) == 0 {
		return persistedBackups
	}
	filtered := make([]*entity.UserBackup, 0, len(persistedBackups))
	for _, backup := range persistedBackups {
		if backup == nil || backup.Type != constant.BackupTypeAuto || backup.ScheduleTime == nil {
			filtered = append(filtered, backup)
			continue
		}
		if _, ok := virtualSchedules[backupScheduleKey(backup.ClusterID, *backup.ScheduleTime)]; ok {
			continue
		}
		filtered = append(filtered, backup)
	}
	return filtered
}

func suppressUnreadyVirtualBackupsWithPersistedSchedule(
	persistedBackups []*entity.UserBackup,
	virtualBackups []*VirtualBackup,
) []*VirtualBackup {
	if len(persistedBackups) == 0 || len(virtualBackups) == 0 {
		return virtualBackups
	}
	persistedSchedules := make(map[string]struct{}, len(persistedBackups))
	for _, backup := range persistedBackups {
		if backup == nil || backup.Type != constant.BackupTypeAuto || backup.ScheduleTime == nil {
			continue
		}
		persistedSchedules[backupScheduleKey(backup.ClusterID, *backup.ScheduleTime)] = struct{}{}
	}
	if len(persistedSchedules) == 0 {
		return virtualBackups
	}
	filtered := make([]*VirtualBackup, 0, len(virtualBackups))
	for _, backup := range virtualBackups {
		if backup == nil || backup.Physical != nil {
			filtered = append(filtered, backup)
			continue
		}
		if _, ok := persistedSchedules[backupScheduleKey(backup.ClusterID, backup.ScheduleTime)]; ok {
			continue
		}
		filtered = append(filtered, backup)
	}
	return filtered
}

func suppressPersistedBackupsWithTombstones(
	backups []*entity.UserBackup,
	tombstoneRestorePoints map[string]struct{},
) []*entity.UserBackup {
	filtered := make([]*entity.UserBackup, 0, len(backups))
	for _, backup := range backups {
		if backup == nil || backup.ScheduleTime == nil {
			filtered = append(filtered, backup)
			continue
		}
		key := backupScheduleKey(backup.ClusterID, *backup.ScheduleTime)
		if _, deleted := tombstoneRestorePoints[key]; !deleted {
			filtered = append(filtered, backup)
		}
	}
	return filtered
}

func backupScheduleKey(clusterID string, scheduleTime time.Time) string {
	return fmt.Sprintf("%s:%d", clusterID, scheduleTime.UTC().Unix())
}

func ptrTime(t time.Time) *time.Time {
	return &t
}

func (b *backupRestoreService) buildRecycledInstancesFromHistories(
	ctx context.Context,
	histories []*entity.BackupScheduleHistory,
	regionPairs []entity.ProviderRegion,
	now time.Time,
) ([]*entity.RecycledInstance, bizerr.BizError) {
	historiesByCluster := make(map[string][]*entity.BackupScheduleHistory)
	eligibleHistories := make([]*entity.BackupScheduleHistory, 0, len(histories))
	for _, history := range histories {
		if len(regionPairs) > 0 && !matchHistoryRegion(history, regionPairs) {
			continue
		}
		historiesByCluster[history.ClusterID] = append(historiesByCluster[history.ClusterID], history)
		eligibleHistories = append(eligibleHistories, history)
	}
	instances := make([]*entity.RecycledInstance, 0, len(historiesByCluster))
	tombstones, err := b.listTombstones(ctx, clusterIDsFromHistories(histories, regionPairs))
	if err != nil {
		return nil, err
	}
	tombstoneRestorePoints := restorePointKeysFromTombstones(tombstones)
	delayed, err := b.buildDelayedVirtualBackupsFromHistories(
		ctx, eligibleHistories, now, noCutover, tombstoneRestorePoints, true,
	)
	if err != nil {
		return nil, err
	}
	availableClusters := make(map[string]struct{}, len(historiesByCluster))
	for _, backup := range delayed {
		availableClusters[backup.ClusterID] = struct{}{}
	}
	cursors := make(map[string]*virtualBackupClusterCursor, len(historiesByCluster))
	pending := make(map[string]*VirtualBackup, len(historiesByCluster))
	for clusterID, clusterHistories := range historiesByCluster {
		if _, ok := availableClusters[clusterID]; ok {
			continue
		}
		cursor := newVirtualBackupClusterCursor(
			ctx, clusterHistories, now, tombstoneRestorePoints,
		)
		if candidate := cursor.next(ctx); candidate != nil {
			cursors[clusterID] = cursor
			pending[clusterID] = candidate
		}
	}
	for len(pending) > 0 {
		candidates := make([]*VirtualBackup, 0, len(pending))
		for _, candidate := range pending {
			candidates = append(candidates, candidate)
		}
		available, hydrateErr := b.hydrateVirtualBackups(ctx, candidates, now)
		if hydrateErr != nil {
			return nil, hydrateErr
		}
		for _, backup := range available {
			availableClusters[backup.ClusterID] = struct{}{}
		}
		for clusterID := range pending {
			if _, ok := availableClusters[clusterID]; ok {
				delete(pending, clusterID)
				continue
			}
			next := cursors[clusterID].next(ctx)
			if next == nil {
				delete(pending, clusterID)
				continue
			}
			pending[clusterID] = next
		}
	}
	for clusterID, clusterHistories := range historiesByCluster {
		if _, ok := availableClusters[clusterID]; !ok {
			continue
		}
		history := latestBackupScheduleHistory(clusterHistories)
		if history == nil {
			continue
		}
		instances = append(instances, &entity.RecycledInstance{
			ClusterID:                 clusterID,
			Region:                    history.Region,
			EnhancedEncryptionEnabled: history.EnhancedEncryptionEnabled != nil && *history.EnhancedEncryptionEnabled,
			ZoneRedundantEnabled:      history.ZoneRedundantEnabled != nil && *history.ZoneRedundantEnabled,
			Provider:                  history.GetCloudProvider(),
		})
	}
	return instances, nil
}

type virtualBackupClusterCursor struct {
	cursors    []*virtualBackupHistoryCursor
	candidates []*VirtualBackup
	seen       map[string]struct{}
}

func newVirtualBackupClusterCursor(
	ctx context.Context,
	histories []*entity.BackupScheduleHistory,
	now time.Time,
	tombstoneRestorePoints map[string]struct{},
) *virtualBackupClusterCursor {
	result := &virtualBackupClusterCursor{
		cursors:    make([]*virtualBackupHistoryCursor, 0, len(histories)),
		candidates: make([]*VirtualBackup, 0, len(histories)),
		seen:       make(map[string]struct{}),
	}
	for _, history := range histories {
		cursor := newVirtualBackupHistoryCursor(history, now, tombstoneRestorePoints)
		if cursor == nil {
			continue
		}
		candidate := cursor.previous(ctx)
		if candidate == nil {
			continue
		}
		result.cursors = append(result.cursors, cursor)
		result.candidates = append(result.candidates, candidate)
	}
	return result
}

func (c *virtualBackupClusterCursor) next(ctx context.Context) *VirtualBackup {
	if c == nil {
		return nil
	}
	for {
		latest := -1
		for i, candidate := range c.candidates {
			if candidate != nil && (latest < 0 || candidate.ScheduleTime.After(c.candidates[latest].ScheduleTime)) {
				latest = i
			}
		}
		if latest < 0 {
			return nil
		}
		candidate := c.candidates[latest]
		c.candidates[latest] = c.cursors[latest].previous(ctx)
		if _, ok := c.seen[candidate.ID]; ok {
			continue
		}
		c.seen[candidate.ID] = struct{}{}
		return candidate
	}
}

type virtualBackupHistoryCursor struct {
	history                   *entity.BackupScheduleHistory
	lower                     time.Time
	upper                     time.Time
	retention                 time.Duration
	tombstoneRestorePoints    map[string]struct{}
	invalidScheduleExpression bool
}

func newVirtualBackupHistoryCursor(
	history *entity.BackupScheduleHistory,
	now time.Time,
	tombstoneRestorePoints map[string]struct{},
) *virtualBackupHistoryCursor {
	if history == nil || history.ValidFrom == nil || history.Kubernetes == constant.KubernetesUnknown {
		return nil
	}
	retention := time.Duration(history.Retention) * time.Hour
	lower := history.EffectiveFrom(noCutover)
	if retentionStart := now.Add(-retention).UTC(); retentionStart.After(lower) {
		lower = retentionStart
	}
	upper := history.EffectiveEnd(now)
	if lower.After(upper) {
		return nil
	}
	return &virtualBackupHistoryCursor{
		history:                history,
		lower:                  lower,
		upper:                  upper,
		retention:              retention,
		tombstoneRestorePoints: tombstoneRestorePoints,
	}
}

func (c *virtualBackupHistoryCursor) previous(ctx context.Context) *VirtualBackup {
	if c == nil || c.invalidScheduleExpression {
		return nil
	}
	for !c.upper.Before(c.lower) {
		windowStart := c.upper.Add(-UserBackupInterval)
		if windowStart.Before(c.lower) {
			windowStart = c.lower
		}
		scheduleTimes, err := enumerateScheduleTimes(c.history.ScheduleCron, windowStart, c.upper)
		if err != nil {
			log.WithContextL(ctx).Warn(
				"skip invalid backup schedule history cron",
				zap.String("cluster_id", c.history.ClusterID),
				zap.String("schedule_cron", c.history.ScheduleCron),
				zap.Error(err),
			)
			c.invalidScheduleExpression = true
			return nil
		}
		for i := len(scheduleTimes) - 1; i >= 0; i-- {
			scheduleTime := scheduleTimes[i]
			if !c.history.Contains(scheduleTime) {
				continue
			}
			c.upper = scheduleTime.Add(-time.Nanosecond)
			if _, deleted := c.tombstoneRestorePoints[backupScheduleKey(c.history.ClusterID, scheduleTime)]; deleted {
				continue
			}
			return &VirtualBackup{
				ID:            generateVirtualBackupID(c.history.ClusterID, scheduleTime),
				TenantID:      c.history.TenantID,
				ClusterID:     c.history.ClusterID,
				Region:        c.history.Region,
				ScheduleTime:  scheduleTime,
				ExpireTime:    scheduleTime.Add(c.retention),
				Retention:     c.retention,
				CloudProvider: c.history.GetCloudProvider(),
				Kubernetes:    c.history.GetKubernetes(),
			}
		}
		if windowStart.Equal(c.lower) {
			return nil
		}
		c.upper = windowStart.Add(-time.Nanosecond)
	}
	return nil
}

func clusterIDsFromHistories(
	histories []*entity.BackupScheduleHistory,
	regionPairs []entity.ProviderRegion,
) []string {
	seen := make(map[string]struct{}, len(histories))
	clusterIDs := make([]string, 0, len(histories))
	for _, history := range histories {
		if history == nil {
			continue
		}
		if len(regionPairs) > 0 && !matchHistoryRegion(history, regionPairs) {
			continue
		}
		if _, ok := seen[history.ClusterID]; ok {
			continue
		}
		seen[history.ClusterID] = struct{}{}
		clusterIDs = append(clusterIDs, history.ClusterID)
	}
	return clusterIDs
}

func matchHistoryRegion(history *entity.BackupScheduleHistory, regionPairs []entity.ProviderRegion) bool {
	for _, pair := range regionPairs {
		if history.GetCloudProvider() == pair.Provider && history.Region == pair.Region {
			return true
		}
	}
	return false
}

func latestBackupScheduleHistory(histories []*entity.BackupScheduleHistory) *entity.BackupScheduleHistory {
	var latest *entity.BackupScheduleHistory
	for _, history := range histories {
		if latest == nil {
			latest = history
			continue
		}
		if history.ValidFrom != nil && latest.ValidFrom != nil && history.ValidFrom.After(*latest.ValidFrom) {
			latest = history
		}
	}
	return latest
}

func (b *backupRestoreService) getVirtualBackup(
	ctx context.Context,
	clusterID, backupID string,
	ref *virtualBackupRef,
	now time.Time,
) (*VirtualBackup, *entity.BackupScheduleHistory, bizerr.BizError) {
	if ref == nil {
		return nil, nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "")
	}
	if clusterID != "" && ref.clusterID != clusterID {
		return nil, nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "")
	}
	exists, err := b.backupTombstoneRepo.Exists(ctx, ref.clusterID, ref.scheduleTime)
	if err != nil {
		return nil, nil, err
	}
	if exists {
		return nil, nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "")
	}
	histories, err := b.backupHistoryRepo.List(ctx, &entity.ListBackupScheduleHistoryOptions{
		ClusterId:          ref.clusterID,
		ValidBeforeOrEqual: &ref.scheduleTime,
		ValidAfter:         &ref.scheduleTime,
	})
	if err != nil {
		return nil, nil, err
	}
	histories, err = b.appendCurrentScheduleHistory(ctx, ref.clusterID, histories)
	if err != nil {
		return nil, nil, err
	}
	for _, history := range histories {
		if history.Kubernetes == constant.KubernetesUnknown {
			continue
		}
		if !historyCoversScheduleTime(history, ref.scheduleTime) {
			continue
		}
		retention := time.Duration(history.Retention) * time.Hour
		expireTime := ref.scheduleTime.Add(retention)
		backup := &VirtualBackup{
			ID:            backupID,
			TenantID:      history.TenantID,
			ClusterID:     history.ClusterID,
			Region:        history.Region,
			ExpireTime:    expireTime,
			ScheduleTime:  ref.scheduleTime,
			Retention:     retention,
			CloudProvider: history.GetCloudProvider(),
			Kubernetes:    history.GetKubernetes(),
		}
		backups, hydrateErr := b.hydrateVirtualBackups(ctx, []*VirtualBackup{backup}, now)
		if hydrateErr != nil {
			return nil, nil, hydrateErr
		}
		if len(backups) == 0 {
			return nil, nil, bizerr.New(bizerr.ErrCodeRecordNotFound, virtualBackupNotReadyMessage)
		}
		return backups[0], history, nil
	}
	return nil, nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "")
}

func historyCoversScheduleTime(history *entity.BackupScheduleHistory, scheduleTime time.Time) bool {
	if history == nil || !history.Contains(scheduleTime) {
		return false
	}
	// Virtual backup IDs contain a user-controlled timestamp. Ensure it is an
	// actual occurrence of the schedule rather than merely a time inside this
	// history's validity window.
	expr, err := constant.BackupCronParse.Parse(history.ScheduleCron)
	if err != nil {
		return false
	}
	next := expr.Next(scheduleTime.Add(-time.Second).UTC()).UTC()
	return next.Equal(scheduleTime.UTC())
}

func (b *backupRestoreService) resolveVirtualBackup(
	ctx context.Context,
	backup *VirtualBackup,
	history *entity.BackupScheduleHistory,
) (*entity.UserBackup, bizerr.BizError) {
	if backup == nil || history == nil {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "")
	}
	var backupItem *pkgv2.BackupItem
	var err error
	if b.physicalBackupSlotRepo != nil {
		slot := backup.Physical
		if slot == nil {
			var slotErr bizerr.BizError
			slot, slotErr = b.physicalBackupSlotRepo.Get(
				ctx,
				history.GetCloudProvider(),
				history.Region,
				history.GetKubernetes(),
				backup.ScheduleTime,
			)
			if slotErr != nil {
				if slotErr.ErrCode() == bizerr.ErrCodeRecordNotFound {
					return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, virtualBackupNotReadyMessage)
				}
				return nil, slotErr
			}
		}
		backupItem = &pkgv2.BackupItem{ID: slot.BackupID, Name: slot.BackupName, Time: pkgv2.JSONTime(slot.BackupTime)}
	} else {
		backupItem, err = b.searchServerlessBackupForHistory(ctx, backup.ScheduleTime, history)
	}
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "fail to resolve backup")
	}
	if backupItem == nil {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, virtualBackupNotReadyMessage)
	}
	backupTime := time.Time(backupItem.Time)
	createdAt := backup.ScheduleTime
	expireTime := backupTime.UTC().Add(backup.Retention)
	if expireTime.Before(time.Now().UTC()) {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "")
	}
	backup.ExpireTime = expireTime
	scheduleTime := backup.ScheduleTime
	return &entity.UserBackup{
		ID:           backup.ID,
		State:        constant.BackupStateSucceeded,
		CreatedAt:    &createdAt,
		TenantID:     backup.TenantID,
		ClusterID:    backup.ClusterID,
		Region:       backup.Region,
		Type:         constant.BackupTypeAuto,
		BackupTime:   &backupTime,
		ExpireTime:   &expireTime,
		ScheduleTime: &scheduleTime,
		BackupID:     backupItem.ID,
		BackupName:   backupItem.Name,
	}, nil
}

func (b *backupRestoreService) searchServerlessBackupForHistory(
	ctx context.Context,
	scheduleTime time.Time,
	history *entity.BackupScheduleHistory,
) (*pkgv2.BackupItem, error) {
	logger := log.WithContextL(ctx).
		With(zap.String("cluster_id", history.ClusterID)).
		With(zap.Time("schedule_time", scheduleTime))
	clusterProxyURL := config.GetClusterProxyUrl(history.GetCloudProvider(), history.Region, history.GetKubernetes())
	if clusterProxyURL == "" {
		return nil, fmt.Errorf("cluster proxy url for %s not found", history.GetKubernetes())
	}
	proxy, err := b.clusterProxyProvider.GetServerlessProxy(ctx, clusterProxyURL)
	if err != nil {
		return nil, err
	}
	resp, err := proxy.ListBackups(ctx, pkgv2.ListBackupRequest{
		LastBackupTime: pkgv2.JSONTime(scheduleTime.Add(-time.Second)),
		MaxCount:       100,
	})
	if err != nil {
		return nil, err
	}
	sort.Slice(resp.Items, func(i, j int) bool {
		return time.Time(resp.Items[i].Time).Before(time.Time(resp.Items[j].Time))
	})
	for _, item := range resp.Items {
		backupTime := time.Time(item.Time)
		if backupTime.Before(scheduleTime) {
			continue
		}
		logger.Info("resolved virtual backup", zap.Uint64("backup_id", item.ID))
		return &item, nil
	}
	return nil, nil
}