package service

import (
	"context"
	"fmt"
	"sort"
	"time"

	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy/serverless"
	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/facade"
	"github.com/tidbcloud/serverless-service/internal/domain/repo"
	"go.uber.org/multierr"
)

const (
	virtualizedSlotSyncBatchSize      = 48
	virtualizedBackupListPageMaxCount = 100
)

type VirtualizedBackupSlotWorker struct {
	proxyProvider facade.ClusterProxyProviderFacade
	slotRepo      repo.PhysicalBackupSlotRepo
	historyRepo   repo.BackupScheduleHistoryRepo
}

func NewVirtualizedBackupSlotWorker(
	proxyProvider facade.ClusterProxyProviderFacade,
	slotRepo repo.PhysicalBackupSlotRepo,
	historyRepo repo.BackupScheduleHistoryRepo,
) *VirtualizedBackupSlotWorker {
	return &VirtualizedBackupSlotWorker{
		proxyProvider: proxyProvider,
		slotRepo:      slotRepo,
		historyRepo:   historyRepo,
	}
}

func (w *VirtualizedBackupSlotWorker) Sync(ctx context.Context) error {
	due := virtualizedSlotAtOrBefore(time.Now().UTC().Add(-5 * time.Minute))
	var errs error
	var bootstrapAt *time.Time
	for _, regional := range config.Global().WorkerConfig.RegionalConfig {
		for _, k8s := range regional.K8sConfig {
			latest, getErr := w.slotRepo.GetLatest(ctx, regional.Provider, regional.Region, k8s.Kubernetes)
			if getErr != nil && getErr.ErrCode() != bizerr.ErrCodeRecordNotFound {
				errs = multierr.Append(errs, getErr)
				continue
			}
			if latest == nil && bootstrapAt == nil {
				cutover, cutoverErr := w.historyRepo.GetVirtualizationCutover(ctx)
				if cutoverErr != nil {
					errs = multierr.Append(errs, cutoverErr)
					continue
				}
				bootstrapAt = &cutover
			}
			slots := virtualizedNextSlots(latest, due, bootstrapAt, virtualizedSlotSyncBatchSize)
			if len(slots) == 0 {
				continue
			}
			if k8s.ClusterProxyUrl == "" {
				errs = multierr.Append(errs, fmt.Errorf("cluster proxy url for %s not found", k8s.Kubernetes))
				continue
			}
			proxy, proxyErr := w.proxyProvider.GetServerlessProxy(ctx, k8s.ClusterProxyUrl)
			if proxyErr != nil {
				errs = multierr.Append(errs, proxyErr)
				continue
			}
			backups, listErr := listPhysicalBackupsForSlotBatch(ctx, proxy, slots[0], slots[len(slots)-1])
			if listErr != nil {
				errs = multierr.Append(errs, listErr)
				continue
			}
			resolved := buildPhysicalBackupSlots(
				regional.Provider,
				regional.Region,
				k8s.Kubernetes,
				slots,
				backups,
			)
			if len(resolved) == 0 {
				continue
			}
			errs = multierr.Append(errs, w.slotRepo.BatchUpsert(ctx, resolved))
		}
	}
	return errs
}

func listPhysicalBackupsForSlotBatch(
	ctx context.Context,
	proxy facade.ServerlessClusterProxyFacade,
	firstSlot, lastSlot time.Time,
) ([]serverless.BackupItem, error) {
	cursor := firstSlot.UTC().Add(-time.Second)
	backups := make([]serverless.BackupItem, 0)
	for {
		resp, err := proxy.ListBackups(ctx, serverless.ListBackupRequest{
			LastBackupTime: serverless.JSONTime(cursor),
			MaxCount:       virtualizedBackupListPageMaxCount,
		})
		if err != nil {
			return nil, err
		}
		if resp == nil || len(resp.Items) == 0 {
			if resp != nil && resp.HasMore {
				return nil, fmt.Errorf(
					"physical backup pagination did not advance after %s",
					cursor.Format(time.RFC3339),
				)
			}
			break
		}
		sort.Slice(resp.Items, func(i, j int) bool {
			return time.Time(resp.Items[i].Time).Before(time.Time(resp.Items[j].Time))
		})
		backups = append(backups, resp.Items...)
		latestBackupTime := time.Time(resp.Items[len(resp.Items)-1].Time).UTC()
		if !latestBackupTime.Before(lastSlot.UTC()) || !resp.HasMore {
			break
		}
		if !latestBackupTime.After(cursor) {
			return nil, fmt.Errorf("physical backup pagination did not advance after %s", cursor.Format(time.RFC3339))
		}
		cursor = latestBackupTime
	}
	return backups, nil
}

func buildPhysicalBackupSlots(
	provider, region, kubernetes string,
	slotTimes []time.Time,
	backups []serverless.BackupItem,
) []*entity.PhysicalBackupSlot {
	sort.Slice(backups, func(i, j int) bool {
		return time.Time(backups[i].Time).Before(time.Time(backups[j].Time))
	})
	result := make([]*entity.PhysicalBackupSlot, 0, len(slotTimes))
	backupIndex := 0
	for _, slotTime := range slotTimes {
		slotTime = slotTime.UTC()
		for backupIndex < len(backups) && time.Time(backups[backupIndex].Time).UTC().Before(slotTime) {
			backupIndex++
		}
		if backupIndex >= len(backups) {
			break
		}
		backup := backups[backupIndex]
		result = append(result, &entity.PhysicalBackupSlot{
			CloudProvider: provider,
			Region:        region,
			Kubernetes:    kubernetes,
			SlotTime:      slotTime,
			BackupID:      backup.ID,
			BackupName:    backup.Name,
			BackupTime:    time.Time(backup.Time).UTC(),
		})
	}
	return result
}

func virtualizedSlotAtOrBefore(t time.Time) time.Time {
	t = t.UTC().Truncate(time.Minute)
	return t.Add(-time.Duration(t.Minute()%30) * time.Minute)
}

func virtualizedSlotAtOrAfter(t time.Time) time.Time {
	before := virtualizedSlotAtOrBefore(t)
	if before.Before(t.UTC()) {
		return before.Add(30 * time.Minute)
	}
	return before
}

func virtualizedNextSlot(
	latest *entity.PhysicalBackupSlot,
	due time.Time,
	bootstrapAt *time.Time,
) time.Time {
	if latest == nil {
		if bootstrapAt == nil {
			return due
		}
		return virtualizedSlotAtOrAfter(*bootstrapAt)
	}
	return latest.SlotTime.UTC().Add(30 * time.Minute)
}

func virtualizedNextSlots(
	latest *entity.PhysicalBackupSlot,
	due time.Time,
	bootstrapAt *time.Time,
	limit int,
) []time.Time {
	if limit <= 0 {
		return []time.Time{}
	}
	first := virtualizedNextSlot(latest, due, bootstrapAt)
	slots := make([]time.Time, 0, limit)
	for slot := first; !slot.After(due.UTC()) && len(slots) < limit; slot = slot.Add(30 * time.Minute) {
		slots = append(slots, slot)
	}
	return slots
}