// Package backup encapsulates all backup functionality
package backup

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"time"

	"github.com/aws/aws-sdk-go/aws"
	gcpprovider "github.com/pingcap/dbaas/pkg/infra/provider/gcp"
	"github.com/pingcap/errors"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/tidbcloud/central-sdk/config"
	"github.com/tidbcloud/central-sdk/db"
	"github.com/tidbcloud/central-sdk/db/dbutil"
	"github.com/tidbcloud/central-sdk/db/entity"
	infraapi "github.com/tidbcloud/central-sdk/infra-api"
	awsprovider "github.com/tidbcloud/central-sdk/infra/util/aws"
	"github.com/tidbcloud/central-sdk/metrics"
	"github.com/tidbcloud/central-sdk/model/util"
	"github.com/tidbcloud/central-sdk/runtime/deployer/gardener"
	runtimek8s "github.com/tidbcloud/central-sdk/runtime/deployer/k8s"
	runtimemanager "github.com/tidbcloud/central-sdk/runtime/manager"
	runtimeutil "github.com/tidbcloud/central-sdk/runtime/util"
	"github.com/tidbcloud/central-sdk/worker/operator"
	"github.com/tidbcloud/config-service/sdk/api"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"go.uber.org/multierr"
	"go.uber.org/zap"
	"gorm.io/gorm"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/pointer"
)

func (c BgContext) takeClusterBackup(cluster *entity.Cluster, provider config.Cloud, region string) error {
	var (
		err    error
		backup *entity.Backup
	)
	// find the earliest pending backup
	err = db.Transaction(func(tx *gorm.DB) error {
		backup, err = dbutil.NextCanBackup(tx, cluster.ID)
		return err
	})
	if err != nil {
		c.Log().Error("failed to load backup can start from db", zap.Uint64("clusterID", cluster.ID.Uint64()), zap.Error(err))
		return err
	}
	if backup == nil {
		c.Log().Info("No valid next backup in the queue", zap.Uint64("clusterID", cluster.ID.Uint64()), zap.Error(err))
		return nil
	}
	if !cluster.ReadyForBackup() {
		c.Log().Info("cluster is not ready for backup yet", zap.Uint64("clusterID", uint64(cluster.ID)))
		return nil
	}

	return multierr.Append(err, c.takeBackup(cluster, backup, provider, region))
}

func (c BgContext) takeBackup(cluster *entity.Cluster, backup *entity.Backup, provider config.Cloud, region string) error {
	c.Log().Info("taking backup for cluster", zap.Uint64("backupID", uint64(backup.ID)), zap.Uint64("clusterID", uint64(cluster.ID)))
	tidbNamespace := runtimeutil.TiDBNamespace(uint64(cluster.ID))

	backupName := backupCRDName(backup)
	backupVars := operator.TidbBackupVars{
		Bucket:           runtimeutil.GetBucketName(c.getProjectID(), region, cluster.ProjectID.Uint64()),
		BackupName:       backupName,
		BackupNamespace:  tidbNamespace.String(),
		Cloud:            provider,
		Host:             runtimek8s.GetPrivilegedTiDBAddress(config.ClusterPrefix, tidbNamespace.String()),
		SecretName:       runtimek8s.GetExtraAccessSecretName(config.ClusterPrefix),
		Region:           region,
		ProjectID:        cluster.ProjectID,
		ClusterID:        cluster.ID,
		ClusterNamespace: tidbNamespace.String(),
	}

	if c.infraAPI.InfraAPIConfig.EnableBackupRestoreWrite {
		err := c.infraAPI.CreateBackupIgnoreExists(context.TODO(),
			cluster.TenantID.Uint64(), cluster.ProjectID.Uint64(), cluster.ID.Uint64(),
			infraapi.ClusterName(cluster.ID.Uint64()), backupName)
		if err != nil {
			c.Log().Error("create backup failed", zap.Uint64("clusterID", uint64(cluster.ID)), zap.Error(err))
			return err
		}
	} else {
		backupCR, err := c.assembleTiDBBackup(backupVars)
		if err != nil {
			c.Log().Error("create backup failed", zap.Uint64("clusterID", uint64(cluster.ID)), zap.Error(err))
			return err
		}

		err = c.createBackupCRD(backupCR, backupVars)
		if err != nil {
			c.Log().Error("create backup failed", zap.Uint64("clusterID", uint64(cluster.ID)), zap.Error(err))
			return err
		}
	}

	return db.Transaction(func(tx *gorm.DB) error {
		if backup.Status != entity.BackupStatus_RUNNING {
			backup.Status = entity.BackupStatus_RUNNING
			return dbutil.UpdateBackup(tx, backup)
		}
		return nil
	})
}

func (c BgContext) GetBackupConfigFromConfigService(tenantID, clusterID uint64) (*api.DescribeBRConfigOutput, error) {
	input := api.DescribeBRConfigInput{
		TenantID:  &tenantID,
		ClusterID: &clusterID,
	}
	return c.ConfigService.DescribeBackupConfiguration(&input)
}

func (c BgContext) GetRestoreConfigFromConfigService(tenantID, clusterID uint64) (*api.DescribeBRConfigOutput, error) {
	input := api.DescribeBRConfigInput{
		TenantID:  &tenantID,
		ClusterID: &clusterID,
	}
	return c.ConfigService.DescribeRestoreConfiguration(&input)
}

func (c BgContext) BRConfig(output *api.DescribeBRConfigOutput) (*v1alpha1.BRConfig, error) {
	jsonBody, err := json.Marshal(output.Config)
	if err != nil {
		return nil, err
	}
	var brConfig v1alpha1.BRConfig
	err = json.Unmarshal(jsonBody, &brConfig)
	if err != nil {
		return nil, err
	}
	return &brConfig, nil
}

func (c BgContext) assembleTiDBBackup(vars operator.TidbBackupVars) (*v1alpha1.Backup, error) {
	output, err := c.GetBackupConfigFromConfigService(c.tenantAndUser.TenantID.Uint64(), vars.ClusterID.Uint64())
	if err != nil {
		return nil, err
	}
	brConfig, err := c.BRConfig(output)
	if err != nil {
		return nil, err
	}
	var backupCR *v1alpha1.Backup
	switch vars.Cloud {
	case config.Aws:
		baseRoleArn, err := c.getAwsBaseRoleArn()
		if err != nil {
			return nil, err
		}
		tenantID := c.tenantAndUser.TenantID.Uint64()
		shootName := runtimeutil.ShootName(vars.ProjectID.Uint64(), config.Aws, vars.Region)
		bucketArn := fmt.Sprintf("%s/dbaas-%s-%d-%s-bucket-reader", baseRoleArn, c.getProjectID(), tenantID, shootName)
		annotations := make(map[string]string)
		annotations["iam.amazonaws.com/role"] = bucketArn
		backupCR = c.assembleAwsBackupCRD(vars, annotations)
	case config.Gcp:
		backupCR = c.assembleGcpBackupCRD(vars)
	default:
		return nil, errors.Errorf("unsupported cloud environment to create a CRD backup")
	}

	return c.mergeBackupConfig(backupCR, brConfig, output), nil
}

func (c BgContext) mergeBackupConfig(backup *v1alpha1.Backup, brConfig *v1alpha1.BRConfig, output *api.DescribeBRConfigOutput) *v1alpha1.Backup {
	tag := output.Version
	if output.Image.Tag != nil {
		tag = *output.Image.Tag
	}
	backup.Spec.ToolImage = fmt.Sprintf("%s:%s", output.Image.BaseImage, tag)

	backup.Spec.BR.LogLevel = brConfig.LogLevel
	backup.Spec.BR.StatusAddr = brConfig.StatusAddr
	backup.Spec.BR.TimeAgo = brConfig.TimeAgo
	backup.Spec.BR.Options = brConfig.Options
	if brConfig.Concurrency != nil {
		backup.Spec.BR.Concurrency = brConfig.Concurrency
	}
	if brConfig.RateLimit != nil {
		backup.Spec.BR.RateLimit = brConfig.RateLimit
	}
	if brConfig.Checksum != nil {
		backup.Spec.BR.Checksum = brConfig.Checksum
	}
	if brConfig.OnLine != nil {
		backup.Spec.BR.OnLine = brConfig.OnLine
	}
	return backup
}

// UpdateBackupProgress updates the operation when the backup is completed
func (c BgContext) UpdateBackupProgress(backupCR *v1alpha1.Backup, backupCRInfra *cloudv1.Backup, backup *entity.Backup) error {
	var (
		backupName string
		status     entity.BackupStatus
		backupSize float64
	)
	if backupCRInfra != nil {
		backupName = backupCRInfra.GetName()
		status = entity.BackupStatus(util.GetBackupStatusForInfraAPI(backupCRInfra))
		backupSize = float64(backupCRInfra.Status.BackupSize)
	} else {
		backupName = backupCR.GetName()
		status = entity.BackupStatus(util.GetBackupStatus(backupCR))
		backupSize = float64(backupCR.Status.BackupSize)
	}
	if backup.Status != status {
		if status == entity.BackupStatus_SUCCESS {
			if usedTime := time.Since(backup.CreatedAt).Seconds(); usedTime > time.Minute.Seconds() {
				tikvNum := float64(len(backup.TikvComponentMachines.IDs))
				metrics.BackupExecutionSpeed.Observe((backupSize / tikvNum) / usedTime)
			}
			now := time.Now()
			backup.FinishedAt = &now
		}
		c.Log().Debug("get backup status", zap.String("backupName", backupName), zap.String("status", string(backup.Status)))
		backup.Status = status
		return db.Transaction(func(tx *gorm.DB) error {
			return dbutil.UpdateBackup(tx, backup)
		})
	}
	return nil
}

// TakeBackups creates a k8s job for all backup records
func (c BgContext) TakeBackups() error {
	var (
		clusters []entity.Cluster
		err      error
	)
	// TODO: put more logic into one transaction.
	err = db.Transaction(func(tx *gorm.DB) error {
		clusters, err = dbutil.GetClustersReadyForBackup(tx, c.tenantAndUser.TenantID)
		return err
	})
	if err != nil {
		return errors.Annotate(err, "get clusters ready for backup")
	}
	var errs error
	for _, cluster := range clusters {
		var (
			provider *entity.Provider
			region   *entity.Region
		)
		err = db.Transaction(func(tx *gorm.DB) error {
			provider, region, err = dbutil.GetClusterProviderAndRegion(tx, cluster)
			return err
		})
		if err != nil {
			c.Log().Error("failed to get region and provider", zap.Error(err))
			continue
		}
		errs = multierr.Append(errs, c.takeClusterBackup(&cluster, provider.Name, region.Name))
	}
	for _, cluster := range clusters {
		var (
			provider *entity.Provider
			region   *entity.Region
		)
		err = db.Transaction(func(tx *gorm.DB) error {
			provider, region, err = dbutil.GetClusterProviderAndRegion(tx, cluster)
			return err
		})
		if err != nil {
			c.Log().Error("failed to get region and provider", zap.Error(err))
			continue
		}
		var backups []entity.Backup
		err = db.Transaction(func(tx *gorm.DB) error {
			backups, err = dbutil.BackupsInProgress(tx, cluster.ID)
			return err
		})
		if err != nil {
			errs = multierr.Append(errs, err)
			continue
		}

		for _, backup := range backups {
			var (
				backupCR      *v1alpha1.Backup
				backupCRInfra *cloudv1.Backup
			)
			if c.infraAPI.InfraAPIConfig.EnableBackupRestoreRead {
				backupCRInfra, err = c.infraAPI.GetBackup(context.TODO(), cluster.ProjectID.Uint64(), backupCRDName(&backup))
				if err != nil {
					errs = multierr.Append(errs, err)
					continue
				}
			} else {
				rx, err := c.runtimes.Context(cluster.ProjectID.Uint64(), provider.Name, region.Name)
				if err != nil {
					if gardener.IsUnavailable(err) {
						c.log.Warn("runtime context for cluster is not ready", zap.Uint64("clusterID", uint64(cluster.ID)))
					} else {
						errs = multierr.Append(errs, err)
					}
					continue
				}
				namespace := runtimeutil.TiDBNamespace(backup.BackupSchedule.ClusterID.Uint64())
				backupCR, err = rx.Runtime.GetBackup(backupCRDName(&backup), namespace.String())
				if err != nil {
					errs = multierr.Append(errs, err)
					continue
				}
			}
			errs = multierr.Append(errs, c.UpdateBackupProgress(backupCR, backupCRInfra, &backup))
		}
	}
	return errs
}

func backupCRDName(backup *entity.Backup) string {
	return util.BackupCRDName(backup)
}

func (c BgContext) getProjectID() string {
	gardenerRuntime := c.runtimes.(*runtimemanager.GardenerRuntimeManager)
	return gardenerRuntime.GardenProject
}

func (c BgContext) assembleAwsBackupCRD(vars operator.TidbBackupVars, annotations map[string]string) *v1alpha1.Backup {
	concurrency := uint32(32)
	return &v1alpha1.Backup{
		ObjectMeta: metav1.ObjectMeta{
			Name:        vars.BackupName,
			Namespace:   vars.BackupNamespace,
			Annotations: annotations,
		},
		Spec: v1alpha1.BackupSpec{
			Type:        v1alpha1.BackupTypeFull,
			CleanPolicy: v1alpha1.CleanPolicyTypeDelete,
			StorageProvider: v1alpha1.StorageProvider{
				S3: &v1alpha1.S3StorageProvider{
					Provider: v1alpha1.S3StorageProviderTypeAWS,
					Region:   vars.Region,
					Bucket:   vars.Bucket,
					Prefix:   vars.BackupName,
				},
			},
			From: &v1alpha1.TiDBAccessConfig{
				Host:       vars.Host,
				SecretName: vars.SecretName,
				Port:       4000,
				User:       "root",
			},
			BR: &v1alpha1.BRConfig{
				Cluster:          config.ClusterPrefix,
				ClusterNamespace: vars.ClusterNamespace,
				SendCredToTikv:   pointer.BoolPtr(false),
				Concurrency:      &concurrency,
				StatusAddr:       config.BRStatusAddr,
			},
			UseKMS: false,
		},
	}
}

func (c BgContext) assembleGcpBackupCRD(vars operator.TidbBackupVars) *v1alpha1.Backup {
	concurrency := uint32(32)
	return &v1alpha1.Backup{
		ObjectMeta: metav1.ObjectMeta{
			Name:      vars.BackupName,
			Namespace: vars.BackupNamespace,
		},
		Spec: v1alpha1.BackupSpec{
			Type:        v1alpha1.BackupTypeFull,
			CleanPolicy: v1alpha1.CleanPolicyTypeDelete,
			StorageProvider: v1alpha1.StorageProvider{
				Gcs: &v1alpha1.GcsStorageProvider{
					ProjectId:  config.Global().InfraConfig.GcpProject,
					Location:   vars.Region,
					Bucket:     vars.Bucket,
					Prefix:     vars.BackupName,
					SecretName: runtimeutil.BRSecretName,
				},
			},
			From: &v1alpha1.TiDBAccessConfig{
				Host:       vars.Host,
				SecretName: vars.SecretName,
				Port:       4000,
				User:       "root",
			},
			BR: &v1alpha1.BRConfig{
				Cluster:          config.ClusterPrefix,
				ClusterNamespace: vars.ClusterNamespace,
				SendCredToTikv:   pointer.BoolPtr(true),
				Concurrency:      &concurrency,
				StatusAddr:       config.BRStatusAddr,
			},
			UseKMS: false,
		},
	}
}

func (c BgContext) createBackupCRD(backup *v1alpha1.Backup, vars operator.TidbBackupVars) error {
	rx, err := c.runtimes.Context(vars.ProjectID.Uint64(), vars.Cloud, vars.Region)
	if err != nil {
		return errors.Annotate(err, "CreateBackupCRD")
	}
	_, err = rx.Runtime.CreateBackup(backup)
	if err != nil && !apierrors.IsAlreadyExists(err) {
		return errors.Annotate(err, "createBackupCRD")
	}
	return nil
}

func (c BgContext) deleteBackupCRD(projectID entity.ID, cloud config.Cloud, region, name, namespace string) error {
	rx, err := c.runtimes.Context(projectID.Uint64(), cloud, region)
	if err != nil {
		return errors.Annotate(err, "deleteBackupCRD")
	}
	err = rx.Runtime.DeleteBackup(name, namespace)
	if err != nil && !apierrors.IsNotFound(err) {
		return errors.Annotate(err, "deleteBackupCRD")
	}
	return nil
}

func (c BgContext) CleanupBackupRestoreJobs() error {
	org, err := c.Model.tenantModel.User(c.tenantAndUser).GetOrganization()
	if err != nil {
		c.Log().Error("failed to GetOrganization", zap.Error(err))
		return err
	}
	plan := entity.TenantPlan(strings.ToLower(org.Plan))
	orgID := c.tenantAndUser.TenantID

	var clusters []entity.Cluster
	err = db.Transaction(func(tx *gorm.DB) error {
		clusters, err = dbutil.GetClustersForCleanupBackupRestore(tx, orgID)
		return err
	})
	if err != nil {
		return errors.Annotate(err, "fetch deleted no cleaned up")
	}
	var errs error
	for _, cluster := range clusters {
		var (
			provider *entity.Provider
			region   *entity.Region
		)
		err = db.Transaction(func(tx *gorm.DB) error {
			provider, region, err = dbutil.GetClusterProviderAndRegion(tx, cluster)
			return err
		})
		if err != nil {
			c.Log().Error("failed to get cluster region and provider", zap.Uint64("clusterID", cluster.ID.Uint64()), zap.Error(err))
			continue
		}
		namespace := runtimeutil.TiDBNamespace(cluster.ID.Uint64()).String()

		var backups []entity.Backup
		// TODO: put more logic into one transaction.
		err = db.Transaction(func(tx *gorm.DB) error {
			backups, err = dbutil.BackupsFetchNotCleanedUp(tx, cluster.ID)
			return err
		})
		if err != nil {
			errs = multierr.Append(errs, errors.Annotatef(err, "get backup in progress for cluster %d", cluster.ID))
			continue
		}

		if cluster.Status == entity.ClusterStatusDeletecomplete {
			if c.infraAPI.InfraAPIConfig.EnableBackupRestoreWrite {
				infraCluster, err := c.infraAPI.GetCluster(context.TODO(), cluster.ProjectID.Uint64(), cluster.ID.Uint64())
				if err != nil {
					if !apierrors.IsNotFound(err) {
						errs = multierr.Append(errs, errors.Annotatef(err, "error occurs when get infra cluster %d", cluster.ID))
						continue
					}
				} else {
					if infraCluster.DeletionTimestamp == nil {
						errs = multierr.Append(errs, errors.Errorf("infra cluster is not deleted cluster %d", cluster.ID))
						continue
					}
				}
			} else {
				rx, err := c.runtimes.Context(cluster.ProjectID.Uint64(), provider.Name, region.Name)
				if err != nil {
					if !errors.IsNotFound(err) {
						if gardener.IsUnavailable(err) {
							c.log.Info("runtime context for cluster is not ready", zap.Uint64("clusterID", uint64(cluster.ID)))
						} else {
							errs = multierr.Append(errs, errors.Annotatef(err, "get runtime context for cluster %d", cluster.ID))
						}
					}
					continue
				}
				_, err = rx.Runtime.GetTiDBCluster(namespace, true)
				if err == nil {
					errs = multierr.Append(errs, errors.Errorf("tc is not deleted cluster %d", cluster.ID))
					continue
				} else if !apierrors.IsNotFound(err) {
					errs = multierr.Append(errs, errors.Annotatef(err, "error occurs when get tc %d", cluster.ID))
					continue
				}
			}
			if len(backups) == 0 {
				errUpdate := db.Transaction(func(tx *gorm.DB) error {
					c.Log().Info("change cluster status to cleared", zap.Uint64("cluster", cluster.ID.Uint64()))
					return dbutil.ClusterUpdateStatus(tx, cluster.ID, entity.ClusterStatusCleared)
				})
				if errUpdate != nil {
					errs = multierr.Append(errs, errors.Annotatef(errUpdate, "failed to update status to cleared for cluster %d", cluster.ID))
					continue
				}
			}

			if plan == entity.TenantPlanPOC {
				// support config by SRE team
				// poc manual backup reserved time default to 7 days
				// poc auto backup reserved time defatult to 7 days
				pocManualBackupReservedDays := time.Duration(util.GetPocManualBackupReservedDays())
				pocAutoBackupReservedDays := time.Duration(util.GetPocAutoBackupReservedDays())

				if org.RemainingCredit == 0 && org.ActiveClustersCount == 0 {
					var pocEndEmailReminder entity.EmailReminder
					err = db.Transaction(func(tx *gorm.DB) error {
						return entity.NewEmailReminderQuerySet(tx).TenantIDEq(org.ID).TypeEq(entity.EmailReminderTypePocEnd).One(&pocEndEmailReminder)
					})
					if err != nil {
						if nil == errors.Find(err, func(e error) bool { return e == gorm.ErrRecordNotFound }) {
							return err
						}
					} else {
						pocEndBackupDeletionReminderTime := pocEndEmailReminder.CreatedAt.Add((pocManualBackupReservedDays - 2) * 24 * time.Hour)
						if pocEndBackupDeletionReminderTime.Before(time.Now()) {
							err = db.Transaction(func(tx *gorm.DB) error {
								var PocEndBackupDeletionReminder entity.EmailReminder
								return entity.NewEmailReminderQuerySet(tx).TenantIDEq(org.ID).TypeEq(entity.EmailReminderTypePocEndBackupDeletion).One(&PocEndBackupDeletionReminder)
							})
							if err != nil {
								if nil != errors.Find(err, func(e error) bool { return e == gorm.ErrRecordNotFound }) {
									var user entity.User
									err = db.Transaction(func(tx *gorm.DB) error {
										user, err = dbutil.GetUserByEmail(tx, org.Email)
										return err
									})
									if err != nil {
										c.Log().Error("failed to get admin user of tenant", zap.Error(err))
										return err
									}

									consoleURL := fmt.Sprintf("https://%s", config.Global().ServerConfig.FrontendDNS)
									endTime := util.ConvertUtcTime(pocEndEmailReminder.CreatedAt.Add(pocManualBackupReservedDays*24*time.Hour), org.Timezone)
									err := c.mail.SendPocEndBackupDeletionEmail(org.Email, consoleURL, user.FirstName, endTime)
									if err != nil {
										c.Log().Error("trial worker: failed to SendPocEndBackupDeletionEmail", zap.Error(err))
										return err
									}
									c.Log().Info("trial worker: SendPocEndBackupDeletionEmail", zap.Time("time", time.Now().UTC()))

									err = db.Transaction(func(tx *gorm.DB) error {
										reminder := entity.EmailReminder{
											TenantID: org.ID,
											Type:     entity.EmailReminderTypePocEndBackupDeletion,
											Amount:   1,
										}
										return reminder.Create(tx)
									})
									if err != nil {
										c.Log().Error("failed to create EmailReminder", zap.String("type", string(entity.EmailReminderTypePocEndBackupDeletion)), zap.Error(err))
										return err
									}
									c.Log().Info("create EmailReminder", zap.String("type", string(entity.EmailReminderTypePocEndBackupDeletion)))
								} else {
									c.Log().Error("failed to get EmailReminder", zap.String("type", string(entity.EmailReminderTypePocEndBackupDeletion)), zap.Error(err))
									return err
								}
							}
						}
					}
				}

				if cluster.DeletedTime.Add(pocManualBackupReservedDays * 24 * time.Hour).Before(time.Now()) {
					errDel := db.Transaction(func(tx *gorm.DB) error {
						return dbutil.DeleteDeletedClusterBackups(tx, orgID, cluster.ProjectID, cluster.ID, entity.BackupType_Manual)
					})
					if errDel != nil {
						errs = multierr.Append(errs, errors.Annotatef(errDel, "delete all expired manual backups of cluster %d", cluster.ID))
						continue
					}
				}
				if cluster.DeletedTime.Add(pocAutoBackupReservedDays * 24 * time.Hour).Before(time.Now()) {
					errDel := db.Transaction(func(tx *gorm.DB) error {
						return dbutil.DeleteDeletedClusterBackups(tx, orgID, cluster.ProjectID, cluster.ID, entity.BackupType_Auto)
					})
					if errDel != nil {
						errs = multierr.Append(errs, errors.Annotatef(errDel, "delete all expired auto backups of cluster %d", cluster.ID))
						continue
					}
				}
			}

			if plan == entity.TenantPlanOnDemand {
				// support config by SRE team
				// onDemand auto backup reserved time defatult to 7 days
				onDemandAutoBackupReservedDays := time.Duration(util.GetOnDemandAutoBackupReservedDays())
				if cluster.DeletedTime.Add(onDemandAutoBackupReservedDays * 24 * time.Hour).Before(time.Now()) {
					errDel := db.Transaction(func(tx *gorm.DB) error {
						return dbutil.DeleteDeletedClusterBackups(tx, orgID, cluster.ProjectID, cluster.ID, entity.BackupType_Auto)
					})
					if errDel != nil {
						errs = multierr.Append(errs, errors.Annotatef(errDel, "delete all expired auto backups of cluster %d", cluster.ID))
						continue
					}
				}
			}

			var restores []entity.Restore
			errRestore := db.Transaction(func(tx *gorm.DB) error {
				restores, err = dbutil.RestoresFetchAll(tx, cluster.ID)
				return err
			})
			if errRestore != nil {
				errs = multierr.Append(errs, errors.Annotatef(errRestore, "get restore in progress for cluster %d", cluster.ID))
				continue
			}
			for _, restore := range restores {
				var errDel error
				if c.infraAPI.InfraAPIConfig.EnableBackupRestoreWrite {
					errDel = c.infraAPI.DeleteRestoreIgnoreNotFound(context.TODO(), cluster.ProjectID.Uint64(), restoreCRDName(&restore))
				} else {
					errDel = c.deleteRestoreCRD(cluster.ProjectID, provider.Name, region.Name, restoreCRDName(&restore), namespace)
				}
				if errDel != nil {
					errs = multierr.Append(errs, errors.Annotatef(errDel, "delete restore CRD %d for cluster %d", restore.ID, cluster.ID))
				}
			}
		}

		for _, backup := range backups {
			if backup.DeletedAt != nil {
				var errDel error
				if c.infraAPI.InfraAPIConfig.EnableBackupRestoreWrite {
					errDel = c.infraAPI.DeleteBackupIgnoreNotFound(context.TODO(), cluster.ProjectID.Uint64(), backupCRDName(&backup))
				} else {
					errDel = c.deleteBackupCRD(cluster.ProjectID, provider.Name, region.Name, backupCRDName(&backup), namespace)
				}
				if errDel != nil {
					c.Log().Info("delete backup CRD", zap.String("CRD name", backupCRDName(&backup)), zap.Error(errDel))
					errs = multierr.Append(errs, errors.Annotatef(errDel, "delete backup CRD %d for cluster %d", backup.ID, cluster.ID))
				}
				errUpdate := db.Transaction(func(tx *gorm.DB) error {
					return dbutil.UpdateBackupCleanedUp(tx, backup, true)
				})
				if errUpdate != nil {
					errs = multierr.Append(errs, errors.Annotatef(errUpdate, "failed to update backup CleanedUp %d for cluster %d", backup.ID, cluster.ID))
				}
			} else if backup.Status == entity.BackupStatus_FAILED {
				if c.infraAPI.InfraAPIConfig.EnableBackupRestoreWrite {
					errDel := c.infraAPI.DeleteBackupIgnoreNotFound(context.TODO(), cluster.ProjectID.Uint64(), backupCRDName(&backup))
					if errDel != nil {
						c.Log().Info("delete backup CRD", zap.String("CRD name", backupCRDName(&backup)), zap.Error(errDel))
						errs = multierr.Append(errs, errors.Annotatef(errDel, "delete backup CRD %d for cluster %d", backup.ID, cluster.ID))
					}
				} else {
					bucket, prefix := c.GetRemotePath(backup, &cluster, region.Name)
					if provider.Name == config.Aws {
						awsConfig := aws.NewConfig().
							WithMaxRetries(3).
							WithRegion(region.Name)
						err = awsprovider.DeleteBucketFolder(awsConfig, bucket, prefix)
						if err != nil {
							c.Log().Info("clean up backup failed", zap.String("bucket", bucket), zap.String("prefix", prefix), zap.Error(err))
							continue
						}
					} else if provider.Name == config.Gcp {
						err = gcpprovider.DeleteBucketFolder(bucket, prefix)
						if err != nil {
							c.Log().Info("clean up backup failed", zap.String("bucket", bucket), zap.String("prefix", prefix), zap.Error(err))
							continue
						}
					} else {
						c.Log().Info("unsupport cloud provider", zap.Stringer("provider", provider.Name), zap.Uint64("clusterID", uint64(cluster.ID)), zap.Uint64("backupID", uint64(backup.ID)))
						continue
					}
				}

				errUpdate := db.Transaction(func(tx *gorm.DB) error {
					return dbutil.UpdateBackupCleanedUp(tx, backup, true)
				})
				if errUpdate != nil {
					errs = multierr.Append(errs, errors.Annotatef(errUpdate, "failed to update backup CleanedUp %d for cluster %d", backup.ID, cluster.ID))
				}
			}
		}
	}
	return errs
}

// ScheduleBackups looks at cluster backup settings and creates backup records accordingly
func (c BgContext) ScheduleBackups() error {
	var (
		clusters []entity.Cluster
		err      error
	)
	err = db.Transaction(func(tx *gorm.DB) error {
		clusters, err = dbutil.GetDeployedCluster(tx, c.tenantAndUser.TenantID)
		return err
	})
	if err != nil {
		return errors.Annotate(err, "Deployed clusters")
	}

	var errs error
	for _, cluster := range clusters {
		err = db.Transaction(func(tx *gorm.DB) error {
			return dbutil.ScheduleBackup(tx, cluster)
		})
		if err != nil {
			errs = multierr.Append(errs, err)
		}
		errs = multierr.Append(errs, c.deleteOldScheduledBackups(&cluster))
	}
	return errs
}

func (c BgContext) deleteOldScheduledBackups(cluster *entity.Cluster) error {
	var (
		backups []entity.Backup
		err     error
	)
	err = db.Transaction(func(tx *gorm.DB) error {
		backups, err = dbutil.OldScheduledBackup(tx, cluster.ID)
		return err
	})
	if err != nil {
		return err
	}
	if len(backups) == 0 {
		return nil
	}

	var errs error
	namespace := runtimeutil.TiDBNamespace(uint64(cluster.ID)).String()
	for _, backup := range backups {
		c.Log().Info("deleting old scheduled backup for cluster", zap.Uint64("backupID", uint64(backup.ID)), zap.String("namespace", namespace))
		if err = db.Transaction(func(tx *gorm.DB) error {
			return dbutil.DeleteBackup(tx, c.tenantAndUser.TenantID, cluster.ProjectID, cluster.ID, backup.ID)
		}); err != nil {
			errs = multierr.Append(errs, err)
		}
	}
	return errs
}

func (c BgContext) GetRemotePath(backup entity.Backup, cluster *entity.Cluster, region string) (string, string) {
	return runtimeutil.GetBucketName(c.getProjectID(), region, cluster.ProjectID.Uint64()), backupCRDName(&backup)
}