package impl

import (
	"context"
	"fmt"
	"strconv"
	"sync"
	"time"

	"github.com/AlekSi/pointer"
	"github.com/elliotchance/pie/v2"
	"github.com/gammazero/workerpool"
	"github.com/hashicorp/go-multierror"
	"github.com/samber/lo"
	"github.com/tidbcloud/dataflow-service/app/config"
	backup_model "github.com/tidbcloud/dataflow-service/app/models/backup"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/repo"
	backup_repo "github.com/tidbcloud/dataflow-service/app/repo/backup"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/utils"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	internal_config "github.com/tidbcloud/dataflow-service/internal/config"
	"github.com/tidbcloud/dataflow-service/internal/model/util"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	"github.com/tidbcloud/dataflow-service/pkg/blob"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	dedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
)

type autoBackupSchedulerImpl struct {
	getStorageFunc blob.GetStorageFunc

	transProvider         db.TransProvider
	backupScheduleRepo    backup_repo.BackupScheduleRepo
	backupSettingRepo     backup_repo.BackupSettingRepo
	backupRepo            backup_repo.BackupRepo
	logBackupRepo         backup_repo.LogBackupRepo
	bucketRepo            repo.BucketRepo
	logBackupTruncateRepo backup_repo.LogBackupTruncateRepo

	backupSettingModel backup_model.BackupSettingModel
	backupCreateRouter backup_model.BackupCreateRouter
	snapshotBackupM    backup_model.SnapshotBackupModel
	pitBackupM         backup_model.PitBackupModel

	mgmtCli          grpcli.MgmtSvcGrpcClient
	importGrpcClient grpcli.ImportGrpcClient
}

func NewAutoBackupSchedulerImpl(
	getStorageFunc blob.GetStorageFunc,
	transProvider db.TransProvider,
	backupScheduleRepo backup_repo.BackupScheduleRepo,
	backupSettingRepo backup_repo.BackupSettingRepo,
	backupRepo backup_repo.BackupRepo,
	logBackupRepo backup_repo.LogBackupRepo,
	bucketRepo repo.BucketRepo,
	logBackupTruncateRepo backup_repo.LogBackupTruncateRepo,
	backupSettingModel backup_model.BackupSettingModel,
	mgmtCli grpcli.MgmtSvcGrpcClient,
	importGrpcClient grpcli.ImportGrpcClient,
	backupCreateRouter backup_model.BackupCreateRouter,
	snapshotBackupM backup_model.SnapshotBackupModel,
	pitBackupM backup_model.PitBackupModel,
) backup_model.AutoBackupScheduler {
	w := &autoBackupSchedulerImpl{
		getStorageFunc: getStorageFunc,

		transProvider:         transProvider,
		backupScheduleRepo:    backupScheduleRepo,
		backupSettingRepo:     backupSettingRepo,
		backupRepo:            backupRepo,
		logBackupRepo:         logBackupRepo,
		bucketRepo:            bucketRepo,
		logBackupTruncateRepo: logBackupTruncateRepo,

		mgmtCli:          mgmtCli,
		importGrpcClient: importGrpcClient,

		backupSettingModel: backupSettingModel,
		backupCreateRouter: backupCreateRouter,
		snapshotBackupM:    snapshotBackupM,
		pitBackupM:         pitBackupM,
	}
	return w
}

// filterProtectedBackups filters out protected backups from the list.
// Returns the filtered list and count of protected backups that were removed.
// This ensures protected backups are never auto-deleted.
func filterProtectedBackups(backups []entity.Backup) ([]entity.Backup, int) {
	filtered := make([]entity.Backup, 0, len(backups))
	protectedCount := 0
	for _, b := range backups {
		if b.Protected {
			protectedCount++
			continue
		}
		filtered = append(filtered, b)
	}
	return filtered, protectedCount
}

func (w *autoBackupSchedulerImpl) ScheduleAutoBackup(ctx context.Context) bizerr.BizError {
	ctx = log.WithLogger(ctx, log.GetLogger(ctx).With(zap.Any("worker", "schedule-auto-backups")))
	logger := log.GetLogger(ctx).With(zap.Any("cluster type", entity.BackupClusterTypeDedicated))
	backupSchedules, err := w.getEnabledAutoBackupSchedules(ctx)
	if err != nil {
		logger.Error("get enabled auto backup schedule failed", zap.Error(err))
		return err
	}
	if len(backupSchedules) == 0 {
		return nil
	}

	workerPool := workerpool.New(w.getScheduleAutoBackupConcurrency())
	defer func() {
		workerPool.StopWait()
		logger.Info("schedule auto backup finished")
	}()

	var wg sync.WaitGroup
	chunkSize := 500
	backupScheduleChunks := pie.Chunk(backupSchedules, chunkSize)
	for _, backupScheduleChunk := range backupScheduleChunks {
		wg.Add(len(backupScheduleChunk))
		for _, schedule := range backupScheduleChunk {
			scheduleCopy := schedule
			logger := logger.With(zap.Any("cluster_id", scheduleCopy.ClusterID), zap.Any("schedule_id", scheduleCopy.ID))
			workerPool.Submit(func() {
				defer wg.Done()
				err := w.scheduleAutoBackupForOneSchedule(ctx, logger, scheduleCopy)
				if err != nil {
					if err.ErrCode() == bizerr.ErrCodeOptimisticLockConflict {
						logger.Warn("schedule backup failed since OptimisticLockConflict, will be retry later", zap.Error(err))
					} else if err.ErrCode() == config.ErrCodeAsyncCreateBackupPreCheckClusterNotNormal {
						logger.Warn("schedule backup failed since cluster is not normal", zap.Error(err))
					} else {
						logger.Error("schedule backup", zap.Error(err))
					}
				}
			})
		}
		wg.Wait()
	}
	return nil
}

// FIXME: we should implement a function directly in BackupSettingModel, the underlying entities should not visible to business logic
func (w *autoBackupSchedulerImpl) getEnabledAutoBackupSchedules(ctx context.Context) (
	[]entity.BackupSchedule, bizerr.BizError,
) {
	// list non delete backup schedules
	backupSchedules, err := w.backupScheduleRepo.List(ctx,
		func(qs entity.BackupScheduleQuerySet) entity.BackupScheduleQuerySet {
			return qs.ClusterTypeEq(entity.BackupClusterTypeDedicated).
				DeletedAtIsNull().
				TypeEq(entity.BackupTypeAuto)
		})
	if err != nil {
		return nil, bizerr.WrapError(err, "get auto backup schedules for dedicated clusters failed")
	}
	// quick exit, avoid backup setting table full scan in next step
	if len(backupSchedules) == 0 {
		return nil, nil
	}

	// get backup setting for schedules
	backupSettings, err := w.backupSettingRepo.List(ctx,
		func(qs entity.BackupSettingQuerySet) entity.BackupSettingQuerySet {
			return qs.ClusterIDIn(pie.Map(backupSchedules, func(s entity.BackupSchedule) entity.ID { return s.ClusterID })...)
		})
	if err != nil {
		return nil, bizerr.WrapError(err, "get backup settings for dedicated clusters failed")
	}
	backupSettingMap := map[entity.ID]*entity.BackupSetting{}
	for _, s := range backupSettings {
		backupSettingMap[s.ClusterID] = &s
	}

	// filter out disabled schedules
	autoEnabled := pie.Filter(backupSchedules, func(s entity.BackupSchedule) bool {
		bs := backupSettingMap[s.ClusterID]
		if bs == nil {
			return false
		}
		return !bs.IsAutoBackupDisabled()
	})
	return autoEnabled, nil
}

func (w *autoBackupSchedulerImpl) scheduleAutoBackupForOneSchedule(
	ctx context.Context, logger *zap.Logger, backupSchedule entity.BackupSchedule,
) bizerr.BizError {
	if backupSchedule.Schedule == "" {
		logger.Error("schedule is nil")
		return bizerr_wrapper.NewInternalErr("schedule is nil")
	}

	curTime := time.Now()
	nextTime := utils.CalculateNextBackupTime(backupSchedule.Schedule)
	// In two circumstances we just update the NextBackupTime field and wait for next schedule.
	// 1. first touch backup schedule record.
	if backupSchedule.NextBackupTime == nil {
		logger.Info("skip this round backup schedule for cluster, because next backup time is nil",
			zap.Any("curTime", curTime.String()))
		return w.updateScheduleNextBackupTime(ctx, &backupSchedule, &nextTime)
	}
	// 2. current time exceed backup time window(30 min) and this could happen
	// when resuming a long time paused cluster.
	if curTime.After(backupSchedule.NextBackupTime.Add(config.BackupMaxPendingTime)) {
		logger.Warn("skip auto backup schedule for dedicated cluster today, "+
			"there may be bugs or some operations conflict with backup, such as import or invalid cluster status. "+
			"Please take a look.",
			zap.Any("curTime", curTime.String()),
			zap.Any("scheduleBackupTime", backupSchedule.NextBackupTime.String()))
		return w.updateScheduleNextBackupTime(ctx, &backupSchedule, &nextTime)
	}

	if curTime.Before(*backupSchedule.NextBackupTime) {
		return nil
	}

	return w.createAutoBackupForOneSchedule(ctx, logger, &backupSchedule, curTime, nextTime)
}

func (w *autoBackupSchedulerImpl) createAutoBackupForOneSchedule(
	ctx context.Context, logger *zap.Logger, backupSchedule *entity.BackupSchedule,
	curTime time.Time, nextTime time.Time,
) bizerr.BizError {
	logger.Info("start this round backup schedule",
		zap.Any("curTime", curTime.String()),
		zap.Any("scheduleBackupTime", backupSchedule.NextBackupTime.String()))

	cluster, err := w.mgmtCli.GetAnyDedicatedClusterBasicByID(ctx, uint64(backupSchedule.ClusterID))
	if err != nil {
		logger.Error("get cluster", zap.Error(err))
		return err
	}

	isCanCreateBackup, err := w.checkCanCreateBackup(ctx, logger, cluster)
	if err != nil {
		return err
	}
	if !isCanCreateBackup {
		return nil
	}

	tErr := w.transProvider.Transaction(ctx, func(ctx context.Context) error {
		err := w.createBackup(ctx, logger, backupSchedule, cluster)
		if err != nil {
			return err
		}
		return w.updateScheduleNextBackupTime(ctx, backupSchedule, &nextTime)
	})
	if tErr != nil {
		logger.Info("create backup failed", zap.Error(tErr))
		return bizerr.WrapError(tErr, "create backup failed")
	}
	return nil
}

// DeleteExpiredAutoBackups
// fixme for performance problem as the number of cluster increasing
func (w *autoBackupSchedulerImpl) DeleteExpiredAutoBackups(ctx context.Context) bizerr.BizError {
	ctx = log.WithLogger(ctx, log.GetLogger(ctx).With(zap.String("worker", "delete-expired-auto-backups")))
	logger := log.GetLogger(ctx).With(zap.Any("cluster type", entity.BackupClusterTypeDedicated))
	backupSchedules, err := w.backupScheduleRepo.List(ctx,
		func(qs entity.BackupScheduleQuerySet) entity.BackupScheduleQuerySet {
			return qs.ClusterTypeEq(entity.BackupClusterTypeDedicated).
				TypeEq(entity.BackupTypeAuto).
				CleanedUpEq(false).
				OrderDescByCreatedAt()
		})
	if err != nil {
		logger.Error("get auto backup schedule failed", zap.Error(err))
		return err
	}

	var multiErr error
	for _, backupSchedule := range backupSchedules {
		logger := logger.With(zap.Any("backup schedule id", backupSchedule.ID)).
			With(zap.Any("cluster id", backupSchedule.ClusterID))
		tempBackupSchedule := backupSchedule
		if err := w.deleteExpiredBackupsForOneSchedule(ctx, logger, &tempBackupSchedule); err != nil {
			multiErr = multierror.Append(multiErr, err)
		}
		if err := w.markBackupScheduleIsCleanUp(ctx, logger, &tempBackupSchedule); err != nil {
			multiErr = multierror.Append(multiErr, err)
		}
	}
	if multiErr != nil {
		logger.Error("delete expired auto backup failed", zap.Error(multiErr))
		return bizerr.WrapError(multiErr, "delete expired auto backup of %s failed", entity.BackupClusterTypeDedicated)
	}
	return nil
}

func (w *autoBackupSchedulerImpl) markBackupScheduleIsCleanUp(
	ctx context.Context, logger *zap.Logger, backupSchedule *entity.BackupSchedule,
) bizerr.BizError {
	if backupSchedule.DeletedAt == nil {
		return nil
	}
	deleted, err := w.checkBackupIsDeleted(ctx, backupSchedule)
	if err != nil {
		return err
	}
	if !deleted {
		return nil
	}

	logger.Info("all backups are deleted, mark backup schedule as cleaned up")
	if err := w.cleanupBackupSchedule(ctx, logger, backupSchedule); err != nil {
		return err
	}
	logger.Info("all backups are deleted, mark backup schedule as cleaned up successfully")
	return nil
}

func (w *autoBackupSchedulerImpl) checkBackupIsDeleted(
	ctx context.Context, backupSchedule *entity.BackupSchedule,
) (bool, bizerr.BizError) {
	backupCount, err := w.backupRepo.Count(ctx, func(qs entity.BackupQuerySet) entity.BackupQuerySet {
		return qs.BackupScheduleIDEq(backupSchedule.ID).
			DeletedAtIsNull()
	})
	if err != nil {
		return false, err
	}
	logBackupCount, err2 := w.logBackupRepo.Count(ctx, func(qs entity.LogBackupQuerySet) entity.LogBackupQuerySet {
		return qs.ClusterIDEq(backupSchedule.ClusterID).DeletedAtIsNull()
	})
	if err2 != nil {
		return false, err2
	}
	return backupCount+logBackupCount == 0, nil
}

func (w *autoBackupSchedulerImpl) updateScheduleNextBackupTime(
	ctx context.Context, schedule *entity.BackupSchedule, nextTime *time.Time,
) bizerr.BizError {
	schedule.NextBackupTime = nextTime
	return w.backupScheduleRepo.CASUpdate(ctx, schedule)
}

func (w *autoBackupSchedulerImpl) deleteExpiredBackupsForOneSchedule(
	ctx context.Context, logger *zap.Logger, backupSchedule *entity.BackupSchedule,
) bizerr.BizError {
	err := w.deleteExpiredFailedBackupsForOneSchedule(ctx, logger, backupSchedule)
	if err != nil {
		logger.Error("failed to delete expired failed backups", zap.Error(err))
		return err
	}

	return w.deleteExpiredSuccessBackupsForOneSchedule(ctx, logger, backupSchedule)
}

func (w *autoBackupSchedulerImpl) deleteExpiredSuccessBackupsForOneSchedule(
	ctx context.Context, logger *zap.Logger, backupSchedule *entity.BackupSchedule,
) bizerr.BizError {
	backScheduleType, err := queryBackScheduleType(ctx, logger, w.logBackupRepo, backupSchedule)
	if err != nil {
		return err
	}
	logger = logger.With(zap.Any("backup mode", backScheduleType))
	switch backScheduleType {
	case domain.SnapshotSchedule:
		return w.deleteExpiredForSnapshotSchedule(ctx, logger, backupSchedule)
	case domain.PiTRSchedule:
		return w.deleteExpiredForPiTRSchedule(ctx, logger, backupSchedule)
	default:
		return bizerr_wrapper.NewInternalErr("unknown backup schedule type")
	}
}

func (w *autoBackupSchedulerImpl) deleteExpiredFailedBackupsForOneSchedule(
	ctx context.Context, logger *zap.Logger, backupSchedule *entity.BackupSchedule,
) bizerr.BizError {
	failedBackups, err := w.queryReservedFailedBackups(ctx, backupSchedule)
	if err != nil {
		return err
	}
	expiredTSO := backupSchedule.ExpiredTSOByTSO(util.TransToTSO(time.Now().Unix()))
	deleteBackups := w.calDeleteFailedAutoBackups(failedBackups, expiredTSO)
	if len(deleteBackups) == 0 {
		return nil
	}

	logger.Info("start to delete expired failed backup for one schedule", zap.Int("count", len(deleteBackups)))
	if err := w.deleteBackupsWithDags(ctx, logger, deleteBackups); err != nil {
		return err
	}
	return nil
}

func (w *autoBackupSchedulerImpl) calDeleteFailedAutoBackups(
	backups []entity.Backup, expiredTSO uint64,
) []entity.Backup {
	// Filter out protected backups - protected backups cannot be auto-deleted
	backups, _ = filterProtectedBackups(backups)

	i := 0
	for ; i < len(backups); i++ {
		startTSO := util.TransToTSO(backups[i].CreatedAt.Unix())
		if startTSO >= expiredTSO {
			break
		}
	}
	if i > 0 {
		return backups[:i]
	}
	return nil
}

// TODO(ideascf): check whether PiTR is enabled in entity.BackupSetting
func queryBackScheduleType(
	ctx context.Context, logger *zap.Logger,
	logBackupRepo backup_repo.LogBackupRepo,
	backupSchedule *entity.BackupSchedule,
) (domain.BackupScheduleType, bizerr.BizError) {
	logBackupCount, err := logBackupRepo.Count(ctx, func(qs entity.LogBackupQuerySet) entity.LogBackupQuerySet {
		return qs.ClusterIDEq(backupSchedule.ClusterID).
			DeletedAtIsNull()
	})
	if err != nil {
		logger.Error("get log backup count failed", zap.Error(err))
		return "", err
	}
	if logBackupCount > 0 {
		return domain.PiTRSchedule, nil
	}
	return domain.SnapshotSchedule, nil
}

func (w *autoBackupSchedulerImpl) deleteExpiredForSnapshotSchedule(
	ctx context.Context, logger *zap.Logger, backupSchedule *entity.BackupSchedule,
) bizerr.BizError {
	backups, err := w.queryReservedBackups(ctx, backupSchedule)
	if err != nil {
		logger.Error("query backups failed", zap.Error(err))
		return err
	}

	if !w.checkWhetherReachDeleteExpiredCycle(backupSchedule) {
		return nil
	}

	expiredTSO := backupSchedule.ExpiredTSOByTSO(util.TransToTSO(time.Now().Unix()))
	deleteBackups, err := calDeleteAutoBackups(logger, backups, expiredTSO, false)
	if err != nil {
		return err
	}
	if len(deleteBackups) == 0 {
		return nil
	}

	return w.executeDeleteExpiredForSnapshotSchedule(ctx, logger, backupSchedule, deleteBackups)
}

func (w *autoBackupSchedulerImpl) executeDeleteExpiredForSnapshotSchedule(
	ctx context.Context, logger *zap.Logger, backupSchedule *entity.BackupSchedule, deleteBackups []entity.Backup,
) bizerr.BizError {
	if err := w.deleteBackupsWithDags(ctx, logger, deleteBackups); err != nil {
		return err
	}
	return w.markBackupScheduleLastCleanAt(ctx, logger, backupSchedule)
}

func (w *autoBackupSchedulerImpl) deleteBackupsWithDags(
	ctx context.Context, logger *zap.Logger, backups []entity.Backup,
) bizerr.BizError {
	for _, backup := range backups {
		_, delErr := w.snapshotBackupM.AsyncDelete(ctx, &domain.AsyncDeleteBackupReq{BackupID: backup.ID.Uint64()})
		if delErr != nil {
			logger.Error("failed to delete backup of cluster",
				zap.Uint64("clusterID", backup.ClusterID.Uint64()),
				zap.Uint64("backupID", backup.ID.Uint64()), zap.Error(delErr))
			return bizerr_wrapper.NewInternalErr(
				"delete backup of cluster with dag %d failed", backup.ClusterID.Uint64())
		}
	}
	return nil
}

func (w *autoBackupSchedulerImpl) deleteLogBackupsWithDags(
	ctx context.Context, logger *zap.Logger, logBackups []entity.LogBackup,
) bizerr.BizError {
	for _, logBackup := range logBackups {
		_, delErr := w.pitBackupM.AsyncDeleteLogBackup(ctx,
			&domain.DeleteLogBackupReq{LogBackupID: logBackup.ID.Uint64()})
		if delErr != nil {
			logger.Error("failed to delete log backup of cluster",
				zap.Uint64("clusterID", logBackup.ClusterID.Uint64()),
				zap.Uint64("logBackupID", logBackup.ID.Uint64()), zap.Error(delErr))
			return bizerr_wrapper.NewInternalErr(
				"delete backup of cluster with dag %d failed", logBackup.ClusterID.Uint64())
		}
	}
	return nil
}

func (w *autoBackupSchedulerImpl) markScheduleCleanupAsNeeded(
	ctx context.Context, logger *zap.Logger, backupSchedule *entity.BackupSchedule,
) bizerr.BizError {
	if backupSchedule.DeletedAt == nil {
		return nil
	}
	return w.cleanupBackupSchedule(ctx, logger, backupSchedule)
}

func (w *autoBackupSchedulerImpl) queryReservedBackups(
	ctx context.Context, backupSchedule *entity.BackupSchedule,
) ([]entity.Backup, bizerr.BizError) {
	backups, err := w.backupRepo.List(ctx, func(qs entity.BackupQuerySet) entity.BackupQuerySet {
		return qs.BackupScheduleIDEq(backupSchedule.ID).
			StatusEq(entity.BackupStatusSuccess).
			DeletedAtIsNull().
			OrderAscByStartTimeStamp()
	})
	return backups, err
}

func (w *autoBackupSchedulerImpl) queryReservedFailedBackups(
	ctx context.Context, backupSchedule *entity.BackupSchedule,
) ([]entity.Backup, bizerr.BizError) {
	backups, err := w.backupRepo.List(ctx, func(qs entity.BackupQuerySet) entity.BackupQuerySet {
		return qs.BackupScheduleIDEq(backupSchedule.ID).
			StatusNe(entity.BackupStatusSuccess).
			DeletedAtIsNull().
			OrderAscByCreatedAt()
	})
	return backups, err
}

func (w *autoBackupSchedulerImpl) cleanupBackupSchedule(
	ctx context.Context, logger *zap.Logger, backupSchedule *entity.BackupSchedule,
) bizerr.BizError {
	logger.Info("all backups are deleted, mark backup schedule as cleaned up")
	return w.backupScheduleRepo.PatchAuto(ctx, backupSchedule.ClusterID, map[string]interface{}{"cleaned_up": true})
}

func (w *autoBackupSchedulerImpl) markBackupScheduleLastCleanAt(
	ctx context.Context, logger *zap.Logger, backupSchedule *entity.BackupSchedule,
) bizerr.BizError {
	logger.Info("mark backup schedule last clean at")
	return w.backupScheduleRepo.PatchAuto(ctx,
		backupSchedule.ClusterID,
		map[string]interface{}{"last_clean_at": time.Now()})
}

func (w *autoBackupSchedulerImpl) markLogBackupTruncateAsDeleted(
	ctx context.Context, logger *zap.Logger, deleteLogBackups []entity.LogBackup,
) bizerr.BizError {
	if len(deleteLogBackups) == 0 {
		return nil
	}
	logger.Info("start to delete expired log backup truncates for one schedule")
	now := time.Now()
	for _, b := range deleteLogBackups {
		truncates, err := w.logBackupTruncateRepo.List(
			ctx, func(qs entity.LogBackupTruncateQuerySet) entity.LogBackupTruncateQuerySet {
				return qs.LogBackupIDEq(b.ID).
					DeletedAtIsNull().
					// fixme: should we add status filter here?
					StatusNotIn(entity.LogBackupTruncateStatusSuccess, entity.LogBackupTruncateStatusFailed).
					OrderAscByCreatedAt()
			})
		if err != nil {
			logger.Error("failed to list log backup truncates", zap.Any("log backup id", b.ID), zap.Error(err))
			return err
		}

		for _, truncate := range truncates {
			truncate.DeletedAt = &now
			err := w.logBackupTruncateRepo.Save(ctx, &truncate)
			if err != nil {
				logger.Error("failed to delete log backup truncate",
					zap.Any("log backup truncate id", truncate.ID), zap.Error(err))
				return err
			}
		}
	}
	return nil
}

func (w *autoBackupSchedulerImpl) truncateLogBackup(
	ctx context.Context, logger *zap.Logger, logBackup *entity.LogBackup, truncateUntil uint64,
) bizerr.BizError {
	if logBackup == nil {
		return nil
	}
	logger.Info("start to truncate log backup", zap.Any("log backup id", logBackup.ID))
	truncate := &entity.LogBackupTruncate{
		LogBackupID: logBackup.ID,
		Until:       strconv.FormatUint(truncateUntil, 10),
		Status:      entity.LogBackupTruncateStatusPending,
	}
	return w.logBackupTruncateRepo.Save(ctx, truncate)
}

// deleteExpiredForPiTRSchedule calculate what backups and log backup we can truncate or delete.
// snapshot1----snapshot2----snapshot3----snapshot...----snapshot n----log backup---->
//
//	t1-----------t2------------t3----------t...-----------tn--------checkpointTS--->
//
// supposed that expiredTS = checkpointTS - retentionPeriod, and expiredTS is between t2 and t3.
// we should promise user can restore to any time between expiredTS and checkpointTS,
// so we should reserve snapshots from snapshot2 to snapshot n.
// then we can delete snapshot1 and truncate log backup to t2 and delete log backups before t2.
// so the returned value is snapshot1, t2 and the log backup, stopped log backups before t2.
func (w *autoBackupSchedulerImpl) deleteExpiredForPiTRSchedule(
	ctx context.Context, logger *zap.Logger, backupSchedule *entity.BackupSchedule,
) bizerr.BizError {
	backups, logBackups, err := w.queryReservedBackupsAndLogBackups(ctx, logger, backupSchedule)
	if err != nil {
		return err
	}

	if backupSchedule.DeletedAt != nil {
		err := w.postClusterDeletedCleanup4PiTRSchedule(ctx, logger, logBackups)
		if err != nil {
			return err
		}
	}

	if !w.checkWhetherReachDeleteExpiredCycle(backupSchedule) {
		return nil
	}
	deleteBackups, err := w.calDeleteAutoBackupsForPiTR(ctx, logger, backupSchedule, backups, logBackups)
	if err != nil {
		return err
	}
	if len(deleteBackups) == 0 {
		return nil
	}

	truncateTSO, cErr := util.ParseTSOFromTSString(deleteBackups[len(deleteBackups)-1].StartTimeStamp)
	if cErr != nil {
		return bizerr.WrapError(err, "parse start timestamp of backup %d", deleteBackups[len(deleteBackups)-1].ID)
	}

	// calculate delete or truncate log backups
	truncateLogBackup, deleteLogBackups, err := w.calDeleteLogBackupsWithTruncateTSO(logBackups, truncateTSO)
	if err != nil {
		logger.Error("calculate log backups which should be truncated or stopped", zap.Error(err))
		return err
	}

	return w.markPiTRBackupAsDeleted(ctx, logger, backupSchedule, truncateLogBackup,
		deleteBackups, truncateTSO, deleteLogBackups)
}

func (w *autoBackupSchedulerImpl) postClusterDeletedCleanup4PiTRSchedule(
	ctx context.Context, logger *zap.Logger, logBackups []entity.LogBackup,
) bizerr.BizError {
	logger.Info("start to delete expired log backup for one schedule")
	if err := w.deleteLogBackupsWithDags(ctx, logger, logBackups); err != nil {
		return err
	}

	if err := w.markLogBackupTruncateAsDeleted(ctx, logger, logBackups); err != nil {
		return err
	}

	return nil
}

func (w *autoBackupSchedulerImpl) markPiTRBackupAsDeleted(
	ctx context.Context, logger *zap.Logger, backupSchedule *entity.BackupSchedule, truncateLogBackup *entity.LogBackup,
	deleteBackups []entity.Backup, truncateTSO uint64, deleteLogBackups []entity.LogBackup,
) bizerr.BizError {
	logger.Info("start to delete expired auto backup for one schedule")
	if err := w.deleteBackupsWithDags(ctx, logger, deleteBackups); err != nil {
		return err
	}
	if err := w.deleteLogBackupsWithDags(ctx, logger, deleteLogBackups); err != nil {
		return err
	}
	if err := w.transProvider.Transaction(ctx, func(ctx context.Context) error {
		if iErr := w.truncateLogBackup(ctx, logger, truncateLogBackup, truncateTSO); iErr != nil {
			return iErr
		}
		return w.markBackupScheduleLastCleanAt(ctx, logger, backupSchedule)
	}); err != nil {
		logger.Error("delete auto PiTR Backup transaction failed", zap.Error(err))
		return bizerr.WrapError(err, "delete auto PiTR Backup transaction failed")
	}
	return nil
}

func (w *autoBackupSchedulerImpl) calDeleteAutoBackupsForPiTR(
	ctx context.Context, logger *zap.Logger, backupSchedule *entity.BackupSchedule,
	backups []entity.Backup, logBackups []entity.LogBackup,
) ([]entity.Backup, bizerr.BizError) {
	if len(backups) == 0 {
		return nil, nil
	}
	// calculate latestTSO
	latestTSO, err := w.calLatestTSOFromBackups(backups, logBackups)
	if err != nil {
		logger.Error("calculate latest tso failed", zap.Error(err))
		return nil, err
	}
	// By default the cleanup window is anchored on latestTSO (the newest recoverable point, i.e. max
	// of the latest full backup TS and the latest log backup checkpoint TS). This is correct for
	// active clusters: PiTR must guarantee restore to any point within the retention window measured
	// from the real data timeline, not from wall-clock.
	anchorExpiredTSO := backupSchedule.ExpiredTSOByTSO(latestTSO)
	anchorDeletes, err := calDeleteAutoBackups(logger, backups, anchorExpiredTSO, true)
	if err != nil {
		logger.Error("calculate full backups which should be deleted and truncate tso", zap.Error(err))
		return nil, err
	}

	// When a cluster is paused no new data arrives, so latestTSO freezes and the anchor window stops
	// advancing. Expired backups are then never removed and the wall-clock-based
	// BackupFailedToBeCleanedUp alert keeps firing forever. Per product decision, retention is measured
	// by wall-clock while a cluster is paused, so once we confirm the cluster is actually PAUSED we
	// fall back to the wall-clock window.
	wallClockExpiredTSO := backupSchedule.ExpiredTSO(time.Now())
	if wallClockExpiredTSO <= anchorExpiredTSO {
		return anchorDeletes, nil
	}
	wallClockDeletes, err := calDeleteAutoBackups(logger, backups, wallClockExpiredTSO, true)
	if err != nil {
		logger.Error("calculate wall-clock based delete backups for paused PiTR cluster", zap.Error(err))
		return nil, err
	}
	// Only consult the cluster-status RPC when the wall-clock window would actually delete more than
	// the anchor window. Healthy active clusters select the same set (their checkpoint keeps up), so
	// they short-circuit here without any RPC.
	if len(wallClockDeletes) <= len(anchorDeletes) {
		return anchorDeletes, nil
	}

	paused, bizErr := w.isClusterPaused(ctx, backupSchedule.ClusterID)
	if bizErr != nil {
		// fail safe: keep the conservative anchor window rather than risk deleting snapshots that are
		// still needed (e.g. resume catch-up, or a temporarily stuck log backup) when we cannot
		// confirm the cluster is paused.
		logger.Warn("failed to get cluster status for PiTR retention, keep anchor-based cleanup",
			zap.Uint64("clusterID", backupSchedule.ClusterID.Uint64()), zap.Error(bizErr))
		return anchorDeletes, nil
	}
	if !paused {
		return anchorDeletes, nil
	}

	logger.Info("cluster is paused, use wall-clock based retention for PiTR cleanup",
		zap.Uint64("clusterID", backupSchedule.ClusterID.Uint64()),
		zap.Int("anchorDeleteCount", len(anchorDeletes)),
		zap.Int("wallClockDeleteCount", len(wallClockDeletes)))
	return wallClockDeletes, nil
}

func (w *autoBackupSchedulerImpl) isClusterPaused(
	ctx context.Context, clusterID entity.ID,
) (bool, bizerr.BizError) {
	// Basic view is served from the tidb-management-service DB (no regional K8s call) and stays
	// available during CRRF / disaster-recovery, which suits this background cleanup worker.
	cluster, err := w.mgmtCli.GetAnyDedicatedClusterBasicByID(ctx, clusterID.Uint64())
	if err != nil {
		return false, err
	}
	return cluster.GetDedicatedCluster().GetState() == dedicatedv1.Cluster_PAUSED, nil
}

func (w *autoBackupSchedulerImpl) queryReservedBackupsAndLogBackups(
	ctx context.Context, logger *zap.Logger, backupSchedule *entity.BackupSchedule,
) ([]entity.Backup, []entity.LogBackup, bizerr.BizError) {
	backups, err := w.queryReservedBackups(ctx, backupSchedule)
	if err != nil {
		logger.Error("get backups failed", zap.Error(err))
		return nil, nil, err
	}
	logBackups, err := w.queryReservedLogBackups(ctx, backupSchedule)
	if err != nil {
		logger.Error("get logBackups failed", zap.Error(err))
		return nil, nil, err
	}
	return backups, logBackups, err
}

func (w *autoBackupSchedulerImpl) queryReservedLogBackups(
	ctx context.Context, backupSchedule *entity.BackupSchedule,
) ([]entity.LogBackup, bizerr.BizError) {
	return w.logBackupRepo.List(ctx, func(qs entity.LogBackupQuerySet) entity.LogBackupQuerySet {
		return qs.ClusterIDEq(backupSchedule.ClusterID).
			DeletedAtIsNull().
			OrderAscByStartTS()
	})
}

func (w *autoBackupSchedulerImpl) calLatestTSOFromBackups(
	backups []entity.Backup, logBackups []entity.LogBackup,
) (uint64, bizerr.BizError) {
	latestTSOFromFullBackups, err := w.calLatestTSOFromFullBackups(backups)
	if err != nil {
		return 0, err
	}
	if len(logBackups) == 0 {
		return latestTSOFromFullBackups, nil
	}

	// get latestCheckPointTs
	calLatestTSOFromLogBackups, err := w.calLatestTSOFromLogBackups(logBackups)
	if err != nil {
		return 0, err
	}

	// use the larger of backup and logBackup ts as the latestTS
	if calLatestTSOFromLogBackups > latestTSOFromFullBackups {
		return calLatestTSOFromLogBackups, nil
	}
	return latestTSOFromFullBackups, nil
}

func (w *autoBackupSchedulerImpl) calLatestTSOFromLogBackups(
	logBackups []entity.LogBackup,
) (uint64, bizerr.BizError) {
	var latestCheckPointTSO uint64
	var err error
	for i := len(logBackups) - 1; i >= 0; i-- {
		// checkpoint ts may be nil when log backup just started and checkpoint ts was not synced for the first time
		if len(logBackups[i].CheckpointTS) <= 0 {
			continue
		}

		latestCheckPointTSO, err = util.ParseTSOFromTSString(logBackups[i].CheckpointTS)
		if err != nil {
			return 0, bizerr.WrapError(err, "parse checkpoint ts of log backup %d", logBackups[i].ID)
		}
		break
	}
	return latestCheckPointTSO, nil
}

func (w *autoBackupSchedulerImpl) calLatestTSOFromFullBackups(backups []entity.Backup,
) (uint64, bizerr.BizError) {
	latestBackup := backups[len(backups)-1]
	latestBackupTSO, err := util.ParseTSOFromTSString(latestBackup.StartTimeStamp)
	if err != nil {
		return 0, bizerr.WrapError(err, "parse start timestamp of backup %d", latestBackup.ID)
	}
	return latestBackupTSO, nil
}

// calDeleteAutoBackups calculates the expired backups and return the backups that should be deleted.
// Note:
//  1. the backups must be sorted by .startTimeStamp in ascending order.
//  2. When the `isPitrEnable` is true, the latest expired backup will be kept.
//
//nolint:funlen
func calDeleteAutoBackups(
	logger *zap.Logger, backups []entity.Backup, expiredTSO uint64, isPitrEnable bool,
) ([]entity.Backup, bizerr.BizError) {
	logger = logger.With(zap.Any("isPitrEnable", isPitrEnable), zap.Any("function", "calDeleteAutoBackups"))

	// Filter out protected backups first - protected backups cannot be auto-deleted
	backups, protectedCount := filterProtectedBackups(backups)
	if protectedCount > 0 {
		logger.Info("skipped protected backups from auto cleanup", zap.Int("protectedCount", protectedCount))
	}

	// check whether the backups are sorted by .startTimeStamp in ascending order.
	for i := 1; i < len(backups); i++ {
		compare, err := util.CompareTSOStr(backups[i].StartTimeStamp, backups[i-1].StartTimeStamp)
		if err != nil {
			return nil, bizerr.WrapError(err, "compare start timestamp of backup %d and %d", backups[i].ID, backups[i-1].ID)
		}
		if compare < 0 {
			return nil, bizerr.New(bizerr.ErrCodeUnknown, "backups are not sorted by .startTimeStamp in ascending order")
		}
	}

	backupNum := len(backups)
	if backupNum == 0 {
		logger.Info("no backup found")
		return nil, nil
	}

	i := 0
	for ; i < backupNum; i++ {
		startTSO, err := util.ParseTSOFromTSString(backups[i].StartTimeStamp)
		if err != nil {
			return nil, bizerr.WrapError(err, "parse start timestamp of backup %d", backups[i].ID)
		}
		if startTSO >= expiredTSO {
			break
		}
	}

	hasExpiredSnapshotBackup := i > 0
	if !hasExpiredSnapshotBackup {
		logger.Info(fmt.Sprintf("there are %d backups, no backups expired", backupNum))
		return nil, nil
	}

	logger.Info(fmt.Sprintf("there are %d backups, %d expired", backupNum, i))

	// 1. When calculate expired snapshot backups in PiTR, we keep the last expired snapshot backup to ensure that any
	//    point in time within backup retention period can be restored.
	// 2. After cluster deleted, log backups will be deleted if expired, then last expired snapshot backup will be
	//    cleaned too.
	//
	// Timeline for PiTR with expired backups:
	//
	// Time --->
	//                  expiredTSO
	//                     ↓
	// |-------|-------|-------|-------|-------|-------|
	// B1      B2      B3      B4      B5      B6      Now
	// |<---  Expired  --->|   |<-- Not Expired -->|
	//                  ^
	//                  |
	// We keep this backup (B3) when PiTR is enabled to ensure
	// continuous recovery within retention period ( expiredTSO - Now).
	//
	if isPitrEnable {
		logger.Info(fmt.Sprintf("there are %d backups, %d expired, will delete %d expired,"+
			" keep one expired since it's PiTR enabled", backupNum, i, i-1))
		return backups[:i-1], nil
	}

	// Notice: if all expired we should not auto delete the latest one, prevent the cluster is mistakenly deleted and
	// no data to restore.
	if len(backups) == i {
		logger.Info(fmt.Sprintf("LatestBackupProtection is triggered, there are %d backups, %d expired, "+
			"will delete %d expired, keep the latest expired", backupNum, i, i-1))
		return backups[:backupNum-1], nil
	}

	logger.Info(fmt.Sprintf("All expired backups will be deleted, there are %d backups, %d expired", backupNum, i))
	return backups[:i], nil
}

// calDeleteLogBackupsWithTruncateTSO calculate truncate log backup and delete log backups
func (w *autoBackupSchedulerImpl) calDeleteLogBackupsWithTruncateTSO(
	logBackups []entity.LogBackup, truncateTSO uint64,
) (*entity.LogBackup, []entity.LogBackup, bizerr.BizError) {
	var truncate *entity.LogBackup
	deletes := make([]entity.LogBackup, 0, len(logBackups))
	for i := 0; i < len(logBackups); i++ {
		hasCatch, err := w.checkLogBackupCatchTruncateTSO(&logBackups[i], truncateTSO)
		if err != nil {
			return nil, nil, bizerr.WrapError(err, "check whether log backup catch truncate ts or not")
		}

		if !hasCatch {
			// not catch and stopped, we can delete
			if logBackups[i].Status == entity.LogBackupStatusStopped {
				deletes = append(deletes, logBackups[i])
			}
			continue
		}

		if logBackups[i].Status == entity.LogBackupStatusStopped {
			// if log backup is stopped, tidb-operator doesn't support truncate log backup.
			// so we should skip this log backup and wait it's expected truncate ts > checkpoint ts to delete it.
			continue
		}

		isBefore, err := w.checkLogBackupStartBeforeTruncateTSO(&logBackups[i], truncateTSO)
		if err != nil {
			return nil, nil, bizerr.WrapError(err, "check whether log backup start before truncate ts or not")
		}

		if isBefore {
			truncate = &logBackups[i]
		}
	}

	return truncate, deletes, nil
}

// checkLogBackupCatchTruncateTSO  whether log backup catch truncate ts or not
func (w *autoBackupSchedulerImpl) checkLogBackupCatchTruncateTSO(
	logBackup *entity.LogBackup, truncateTSO uint64,
) (bool, error) {
	if logBackup == nil {
		return false, nil
	}

	// no checkpoint ts, we treat as not catch truncate tso
	if len(logBackup.CheckpointTS) == 0 {
		return false, nil
	}

	checkpointTSO, err := util.ParseTSOFromTSString(logBackup.CheckpointTS)
	if err != nil {
		return false, bizerr.WrapError(err, "parse checkpoint ts")
	}

	if checkpointTSO >= truncateTSO {
		return true, nil
	}

	return false, nil
}

// checkLogBackupStartBeforeTruncateTSO  whether log backup start before truncate ts or not
func (w *autoBackupSchedulerImpl) checkLogBackupStartBeforeTruncateTSO(
	logBackup *entity.LogBackup, truncateTSO uint64,
) (bool, error) {
	if logBackup == nil {
		return false, nil
	}

	startTSO, err := util.ParseTSOFromTSString(*logBackup.StartTS)
	if err != nil {
		return false, bizerr.WrapError(err, "parse start ts %s", *logBackup.StartTS)
	}

	if startTSO < truncateTSO {
		return true, nil
	}

	return false, nil
}

func (w *autoBackupSchedulerImpl) getAvailableRetentionPeriod(backupSchedule *entity.BackupSchedule) uint64 {
	return backupSchedule.RetentionPeriod
}

func (w *autoBackupSchedulerImpl) checkWhetherReachDeleteExpiredCycle(backupSchedule *entity.BackupSchedule) bool {
	if backupSchedule.LastCleanAt != nil && time.Since(*backupSchedule.LastCleanAt) < config.AutoDeleteBackupsCycle {
		return false
	}
	return true
}

func (w *autoBackupSchedulerImpl) getScheduleAutoBackupConcurrency() int {
	return internal_config.Global().BackupConfig.DedicatedBackupScheduleConcurrency
}

func (w *autoBackupSchedulerImpl) checkCanCreateBackup(
	ctx context.Context, logger *zap.Logger, cluster *mgmt.Cluster,
) (bool, bizerr.BizError) {
	dc := cluster.GetDedicatedCluster()
	if !lo.Contains(validClusterStatesForBackup, dc.GetState()) {
		logger.Info("cluster is not in valid states for backup, can not create backup",
			zap.Any("valid states", validClusterStatesForBackup),
			zap.Any("cluster state", dc.GetState()))
		return false, nil
	}

	// fixme: Imports are checked during scheduling but not during actual execution.
	listImportResp, err := w.importGrpcClient.List(ctx, &dataflow.ListImportReq{
		OrgId:     cluster.GetOrganizationId(),
		ProjectId: cluster.GetProjectId(),
		ClusterId: cluster.GetClusterId(),
		Statuses: []dataflow.ImportGetResp_Status{
			dataflow.ImportGetResp_PREPARING,
			dataflow.ImportGetResp_IMPORTING,
			dataflow.ImportGetResp_CANCELING,
		},
	})
	if err != nil {
		logger.Error("list import grpc", zap.Error(err))
		return false, bizerr.WrapError(err, "list import grpc")
	}
	if listImportResp != nil && len(listImportResp.Imports) > 0 {
		logger.Info("cluster is importing, can not create backup")
		return false, nil
	}
	return true, nil
}

func (w *autoBackupSchedulerImpl) createBackup(
	ctx context.Context, logger *zap.Logger, backupSchedule *entity.BackupSchedule, cluster *mgmt.Cluster,
) bizerr.BizError {
	createReq := &domain.AsyncCreateBackupReq{
		Name:                 backupSchedule.Name,
		ProjectID:            cluster.GetProjectId(),
		ClusterID:            cluster.GetClusterId(),
		BackupType:           dataflow.BackupType_AUTO,
		UserID:               uint64(backupSchedule.UserID),
		AutoBackupScheduleID: pointer.ToUint64(backupSchedule.ID.Uint64()),
		IsWeekly:             backupSchedule.BackupCycle == entity.BackupCycleWeekly,
	}
	ctx = log.WithLogger(ctx, logger)
	_, createErr := w.backupCreateRouter.AsyncCreate(ctx, createReq)
	return createErr
}