package backup

import (
	"context"
	"fmt"
	"strings"
	"time"

	"github.com/pingcap/dbaas/pkg/secret/encrypt"
	"github.com/pingcap/errcode"
	"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"
	centralModel "github.com/tidbcloud/central-sdk/model"
	"github.com/tidbcloud/central-sdk/model/util"
	"github.com/tidbcloud/central-sdk/runtime/deployer/gardener"
	runtimek8s "github.com/tidbcloud/central-sdk/runtime/deployer/k8s"
	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"
)

// Restore models a restore, including its job via Operation
type Restore struct {
	ID        entity.ID
	BackupID  entity.ID
	ClusterID entity.ID
	Status    string
	CreatedAt time.Time
}

// Restore saves a Restore record to the database
// In the background a job will be created and monitored as an Operation
func (c UserContext) Restore(orgID entity.ID, projectID entity.ID, clusterID entity.ID, backupID entity.ID, name, password string) (Restore, error) {
	org, err := c.Model.tenantModel.User(c.tenantAndUser).GetOrganization()
	if err != nil {
		c.log.Error("failed to get organization", zap.Error(err))
		return Restore{}, errcode.NewInternalErr(errors.New("failed to get organization"))
	}
	plan := entity.TenantPlan(strings.ToLower(org.Plan))
	if plan == entity.TenantPlanPOC && org.RemainingCredit == 0 {
		return Restore{}, errcode.NewInvalidInputErr(errors.New("you have used up your trial credits"))
	}

	var restore *entity.Restore
	err = db.Transaction(func(tx *gorm.DB) error {
		centralModel.LockTenant(tx, c.tenantAndUser.TenantID)

		err = dbutil.ValidateClusterName(tx, name, projectID)
		if err != nil {
			c.log.Error("validateClusterName", zap.Error(err))
			return err
		}

		if plan == entity.TenantPlanNewUser || plan == entity.TenantPlanFreeTrial {
			count, err := dbutil.GetActiveClusterCount(tx, c.tenantAndUser.TenantID)
			if err != nil {
				c.log.Error("failed to get active cluster count", zap.Error(err))
				return err
			}
			if count > 0 {
				return errcode.NewInvalidInputErr(errors.New("you have reached the maximum number of trial clusters"))
			}
		}

		var tenant entity.Tenant
		err := entity.NewTenantQuerySet(tx.Preload("Quotas")).IDEq(c.tenantAndUser.TenantID).One(&tenant)
		if err != nil {
			c.log.Error("failed to get tenant by ID", zap.Error(err))
			return errcode.NewInternalErr(errors.Annotate(err, "failed to get tenant by ID"))
		}

		var backupCluster entity.Cluster
		backupCluster, err = dbutil.GetClusterByID(tx, orgID, projectID, clusterID)
		if err != nil {
			c.log.Error("failed to get cluster by ID", zap.Int64("clusterID", int64(clusterID)), zap.Error(err))
			return errcode.NewInternalErr(errors.Annotate(err, "failed to get cluster by ID"))
		}

		var backup entity.Backup
		err = entity.NewBackupQuerySet(tx).IDEq(backupID).PreloadBackupSchedule().One(&backup)
		if err != nil {
			c.log.Error("failed to get backup by ID", zap.Uint64("backupID", uint64(backupID)), zap.Error(err))
			return errcode.NewInternalErr(errors.Annotate(err, "failed to get backup by ID"))
		}
		if backup.BackupSchedule.ClusterID != clusterID {
			return errcode.NewInvalidInputErr(fmt.Errorf("backup <%d> is not backuped by cluster<%d> ", backupID, clusterID))
		}

		var tidbQuota entity.Quota
		err = entity.NewQuotaQuerySet(tx).TenantIDEq(tenant.ID).ComponentNameEq(entity.TiDB).One(&tidbQuota)
		if err != nil {
			return errcode.NewInternalErr(err)
		}
		var tikvQuota entity.Quota
		err = entity.NewQuotaQuerySet(tx).TenantIDEq(tenant.ID).ComponentNameEq(entity.TiKV).One(&tikvQuota)
		if err != nil {
			return errcode.NewInternalErr(err)
		}
		var tiflashQuota entity.Quota
		err = entity.NewQuotaQuerySet(tx).TenantIDEq(tenant.ID).ComponentNameEq(entity.TiFlash).One(&tiflashQuota)
		if err != nil {
			return errcode.NewInternalErr(err)
		}

		usedTidbCount, usedTikvCount, usedTiflashCount, err := dbutil.GetUsedComponentsCount(tx, c.tenantAndUser.TenantID)
		if err != nil {
			c.log.Error("failed to get used components count", zap.Error(err))
			return errcode.NewInternalErr(errors.New("failed to get used components count"))
		}

		tidbCount := uint32(len(backup.TidbComponentMachines.IDs))
		tikvCount := uint32(len(backup.TikvComponentMachines.IDs))
		tiflashCount := uint32(0)
		if backup.TiflashComponentMachines.IDs != nil {
			tiflashCount = uint32(len(backup.TiflashComponentMachines.IDs))
		}

		// check tidb quota
		desireUsedTidbCount := usedTidbCount + tidbCount
		if tidbQuota.Quota != 0 && desireUsedTidbCount > tidbQuota.Quota {
			return errcode.NewInvalidInputErr(errors.Errorf("failed to restore cluster, %s", util.ClusterQuotasPrompt))
		}

		// check tikv quota
		desireUsedTikvCount := usedTikvCount + tikvCount
		if tikvQuota.Quota != 0 && desireUsedTikvCount > tikvQuota.Quota {
			return errcode.NewInvalidInputErr(errors.Errorf("failed to restore cluster, %s", util.ClusterQuotasPrompt))
		}

		// check tiflash quota
		desireUsedTiflashCount := usedTiflashCount + tiflashCount
		if tiflashQuota.Quota != 0 && desireUsedTiflashCount > tiflashQuota.Quota {
			return errcode.NewInvalidInputErr(errors.Errorf("failed to restore cluster, %s", util.ClusterQuotasPrompt))
		}

		provider, region, err := dbutil.GetClusterProviderAndRegion(tx, backupCluster)
		if err != nil {
			return errcode.NewInvalidInputErr(err)
		}

		clusterCreateReq := dbutil.ClusterCreate{
			ProjectID:    projectID,
			Name:         name,
			Description:  "_",
			TiKVCount:    tikvCount,
			TiDBCount:    tidbCount,
			TiFlashCount: tiflashCount,
			Provider:     provider.Name,
			Region:       region.Name,
			ProfileName:  backupCluster.ClusterProfile.Name,
			Recovering:   true,
		}
		encryptedPassword, err := encrypt.EncryptBase64(c.encryptor, encrypt.Secret(password))
		if err != nil {
			c.log.Error("failed to encrypt password", zap.Error(err))
			return errcode.NewInternalErr(errors.New("failed to encrypt password"))
		}
		clusterCreateReq.EncryptedRootPassword = encryptedPassword
		// TODO: registry per region #1330
		clusterCreateReq.Registry = ""
		restoreCluster, err := dbutil.CreateCluster(tx, c.tenantAndUser.User.ID, c.tenantAndUser.TenantID, clusterCreateReq, plan)
		if err != nil {
			c.log.Error("failed to create cluster db record", zap.Error(err))
			return errcode.NewInternalErr(errors.New("failed to create cluster db record"))
		}

		restore, err = dbutil.CreateRestore(tx, name, backupID, restoreCluster.ID)
		if err != nil {
			return nil
		}
		return dbutil.AddAccessLogNow(tx, c.tenantAndUser.User.TenantID, c.tenantAndUser.Email, fmt.Sprintf("restore cluster: %d, backup: %d, name: %s", restoreCluster.ID, backup.ID, name))
	})
	if err != nil {
		return Restore{}, err
	}

	return c.Model.convertRestore(restore), nil
}

// ListClusterRestores lists Restores for a cluster
func (c UserContext) ListClusterRestores(clusterID entity.ID) ([]Restore, error) {
	var restores []entity.Restore
	err := db.Transaction(func(tx *gorm.DB) error {
		return entity.NewRestoreQuerySet(tx).PreloadCluster().ClusterIDEq(clusterID).All(&restores)
	})
	if err != nil {
		return nil, err
	}
	return c.Model.convertListRestores(restores), nil
}

func (m Model) convertListRestores(restores []entity.Restore) []Restore {
	items := make([]Restore, len(restores))
	for i, restore := range restores {
		items[i] = m.convertRestore(&restore)
	}
	return items
}

func (m Model) convertRestore(restore *entity.Restore) Restore {
	return Restore{
		ID:        restore.ID,
		BackupID:  restore.BackupID,
		ClusterID: restore.ClusterID,
		Status:    string(restore.Status),
		CreatedAt: restore.CreatedAt,
	}
}

func (c BgContext) assembleTiDBRestore(vars operator.TidbRestoreVars) (*v1alpha1.Restore, error) {
	output, err := c.GetRestoreConfigFromConfigService(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 restore *v1alpha1.Restore

	switch vars.Cloud {
	case config.Aws:
		baseRoleArn, err := c.getAwsBaseRoleArn()
		if err != nil {
			return nil, errors.Annotate(err, "createTidbRestore")
		}
		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
		restore = c.assembleAwsRestoreCRD(vars, annotations)
	case config.Gcp:
		restore = c.assembleGcpRestoreCRD(vars)
	default:
		return nil, errors.Errorf("unsupported cloud environment to create a tidb CRD restore")
	}

	return c.mergeRestoreConfig(restore, brConfig, output), nil
}

func (c BgContext) createRestoreCRD(restore *v1alpha1.Restore, vars operator.TidbRestoreVars) (*v1alpha1.Restore, error) {
	rx, err := c.runtimes.Context(vars.ProjectID.Uint64(), vars.Cloud, vars.Region)
	if err != nil {
		return nil, errors.Annotate(err, "CreateRestoreCRD")
	}
	restore, err = rx.Runtime.CreateRestore(restore)
	if err != nil && !apierrors.IsAlreadyExists(err) {
		return nil, errors.Annotate(err, "createRestoreCRD")
	}
	return restore, nil
}

func (c BgContext) assembleGcpRestoreCRD(vars operator.TidbRestoreVars) *v1alpha1.Restore {
	namespace := runtimeutil.TiDBNamespace(vars.ClusterID.Uint64()).String()
	return &v1alpha1.Restore{
		ObjectMeta: metav1.ObjectMeta{
			Name:      vars.RestoreName,
			Namespace: namespace,
		},
		Spec: v1alpha1.RestoreSpec{
			Type: v1alpha1.BackupTypeFull,
			StorageProvider: v1alpha1.StorageProvider{
				Gcs: &v1alpha1.GcsStorageProvider{
					ProjectId:  config.Global().InfraConfig.GcpProject,
					Location:   vars.Region,
					Bucket:     vars.Bucket,
					Prefix:     vars.BucketPrefix,
					SecretName: runtimeutil.BRSecretName,
				},
			},
			To: &v1alpha1.TiDBAccessConfig{
				Host:       vars.Host,
				SecretName: vars.SecretName,
				Port:       4000,
				User:       "root",
			},
			BR: &v1alpha1.BRConfig{
				Cluster:          config.ClusterPrefix,
				ClusterNamespace: namespace,
				SendCredToTikv:   pointer.BoolPtr(true),
			},
			UseKMS: false,
		},
	}
}

func (c BgContext) assembleAwsRestoreCRD(vars operator.TidbRestoreVars, annotations map[string]string) *v1alpha1.Restore {
	namespace := runtimeutil.TiDBNamespace(vars.ClusterID.Uint64()).String()
	restore := &v1alpha1.Restore{
		ObjectMeta: metav1.ObjectMeta{
			Name:        vars.RestoreName,
			Namespace:   namespace,
			Annotations: annotations,
		},
		Spec: v1alpha1.RestoreSpec{
			Type: v1alpha1.BackupTypeFull,
			StorageProvider: v1alpha1.StorageProvider{
				S3: &v1alpha1.S3StorageProvider{
					Provider: v1alpha1.S3StorageProviderTypeAWS,
					Region:   vars.Region,
					Bucket:   vars.Bucket,
					Prefix:   vars.BucketPrefix,
				},
			},
			To: &v1alpha1.TiDBAccessConfig{
				Host:       vars.Host,
				SecretName: vars.SecretName,
				Port:       4000,
				User:       "root",
			},
			BR: &v1alpha1.BRConfig{
				Cluster:          config.ClusterPrefix,
				ClusterNamespace: namespace,
				SendCredToTikv:   pointer.BoolPtr(false),
			},
			UseKMS: false,
		},
	}

	return restore
}

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

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

func (c BgContext) restoreClusterBackups(cluster entity.Cluster) error {
	var (
		restores []entity.Restore
		err      error
	)
	err = db.Transaction(func(tx *gorm.DB) error {
		return entity.NewRestoreQuerySet(tx).ClusterIDEq(cluster.ID).StatusEq(entity.RestoreStatus_PENDING).All(&restores)
	})
	if err != nil {
		return err
	}
	if len(restores) == 0 {
		return nil
	}
	if cluster.Status != entity.ClusterStatusRecovering {
		c.Log().Info("cluster not ready for restore yet", zap.Uint64("clusterID", uint64(cluster.ID)))
		return nil
	}
	var errs error
	for _, restore := range restores {
		errs = multierr.Append(errs, c.restoreBackup(cluster, &restore))
	}
	return errs
}

// UpdateRestoreProgress updates the operation when the restore is completed
func (c BgContext) UpdateRestoreProgress(restoreCR *v1alpha1.Restore, restore *entity.Restore) error {
	restore.Status = util.GetRestoreStatus(restoreCR)
	c.Log().Info("get restore status", zap.String("restoreName", restoreCR.GetName()), zap.String("status", string(restore.Status)))

	restoreTracking := dbutil.ConvertRestoreTracking(restore)
	err := db.Transaction(func(tx *gorm.DB) error {
		err := restore.Update(tx, entity.RestoreDBSchema.Status)
		if err != nil {
			return err
		}
		return restoreTracking.Create(tx)
	})
	return err
}

// UpdateRestoreProgressForInfra updates the operation when the restore is completed
func (c BgContext) UpdateRestoreProgressForInfra(restoreCR *cloudv1.Restore, restore *entity.Restore) error {
	restore.Status = util.GetRestoreStatusForInfraAPI(restoreCR)
	c.Log().Info("get restore status", zap.String("restoreName", restoreCR.GetName()), zap.String("status", string(restore.Status)))

	restoreTracking := dbutil.ConvertRestoreTracking(restore)
	err := db.Transaction(func(tx *gorm.DB) error {
		err := restore.Update(tx, entity.RestoreDBSchema.Status)
		if err != nil {
			return err
		}
		return restoreTracking.Create(tx)
	})
	return err
}

func (c BgContext) restoreBackup(cluster entity.Cluster, restore *entity.Restore) error {
	var (
		backup   *entity.Backup
		provider *entity.Provider
		region   *entity.Region
		err      error
	)
	err = db.Transaction(func(tx *gorm.DB) error {
		backup, err = dbutil.GetBackupByIDForWorker(tx, restore.BackupID)
		if err != nil {
			return err
		}
		provider, region, err = dbutil.GetClusterProviderAndRegion(tx, cluster)
		return err
	})
	if err != nil {
		return err
	}

	namespace := runtimeutil.TiDBNamespace(cluster.ID.Uint64())
	c.Log().Debug("restoring from backup to new cluster", zap.String("backupName", backup.BackupSchedule.Name), zap.Uint64("backupID", uint64(backup.ID)), zap.String("restoreName", restore.Name), zap.Uint64("restoreID", uint64(restore.ID)), zap.Uint64("clusterID", uint64(restore.ClusterID)))

	restoreName := restoreCRDName(restore)
	vars := operator.TidbRestoreVars{
		ProjectID:    cluster.ProjectID,
		ClusterID:    cluster.ID,
		BackupID:     backup.ID,
		RestoreName:  restoreName,
		SecretName:   runtimek8s.GetExtraAccessSecretName(config.ClusterPrefix),
		Host:         runtimek8s.GetPrivilegedTiDBAddress(config.ClusterPrefix, namespace.String()),
		Bucket:       runtimeutil.GetBucketName(c.getProjectID(), region.Name, cluster.ProjectID.Uint64()),
		BucketPrefix: backupCRDName(backup),
		Region:       region.Name,
		Cloud:        provider.Name,
	}

	if c.infraAPI.InfraAPIConfig.EnableBackupRestoreWrite {
		err = c.infraAPI.CreateRestoreIgnoreExists(context.TODO(),
			cluster.TenantID.Uint64(), cluster.ProjectID.Uint64(), cluster.ID.Uint64(), infraapi.ClusterName(cluster.ID.Uint64()), backupCRDName(backup), restoreName)
		if err != nil {
			return errors.Annotate(err, "create restore CR failed")
		}
		// hack for UpdateRestoreProgress, it should always be Running when after created
		restoreCR := &cloudv1.Restore{
			Status: cloudv1.RestoreStatus{Phase: cloudv1.RestoreRunning},
		}

		c.Log().Debug("restored k8s backup", zap.String("backupName", backup.BackupSchedule.Name), zap.Uint64("backupID", uint64(backup.ID)))
		return errors.Annotate(c.UpdateRestoreProgressForInfra(restoreCR, restore), "save restore operation")
	}

	restoreCR, err := c.assembleTiDBRestore(vars)
	if err != nil {
		return errors.Annotate(err, "assemble restore CRD failed")
	}

	restoreCR, err = c.createRestoreCRD(restoreCR, vars)
	if err != nil {
		return errors.Annotate(err, "create restore CRD failed")
	}

	c.Log().Debug("restored k8s backup", zap.String("backupName", backup.BackupSchedule.Name), zap.Uint64("backupID", uint64(backup.ID)))
	return errors.Annotate(c.UpdateRestoreProgress(restoreCR, restore), "save restore operation")
}

func (c BgContext) deleteRestoreCRD(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, "deleteRestoreCRD")
	}
	err = rx.Runtime.DeleteRestore(name, namespace)
	if err != nil && !apierrors.IsNotFound(err) {
		return errors.Annotate(err, "deleteRestoreCRD")
	}
	return nil
}

func restoreCRDName(restore *entity.Restore) string {
	return fmt.Sprintf("%s-%d", strings.ToLower(restore.Name), restore.ID)
}

// RestoreBackups creates a k8s job for all restore records
func (c BgContext) RestoreBackups() error {
	var (
		clusters      []entity.Cluster
		err           error
		restoreStatus entity.ClusterStatus = entity.ClusterStatusRecovering
	)
	err = db.Transaction(func(tx *gorm.DB) error {
		clusters, err = dbutil.FindClusters(tx, dbutil.ClusterFilter{
			TenantID: &c.tenantAndUser.TenantID,
			Status:   &restoreStatus,
		})
		return err
	})
	if err != nil {
		c.Log().Error("failed to get deployed clusters", zap.Error(err))
		return errors.Annotate(err, "get deployed clusters")
	}
	var errs error
	for _, cluster := range clusters {
		errs = multierr.Append(errs, c.restoreClusterBackups(cluster))
	}
	for _, cluster := range clusters {
		var (
			provider *entity.Provider
			region   *entity.Region
			restores []entity.Restore
		)
		err = db.Transaction(func(tx *gorm.DB) error {
			provider, region, err = dbutil.GetClusterProviderAndRegion(tx, cluster)
			if err != nil {
				return err
			}
			// Fetch the running restores from database.
			return entity.NewRestoreQuerySet(tx).
				ClusterIDEq(cluster.ID).
				All(&restores)
		})
		if err != nil {
			c.Log().Error("operate db failed", zap.Uint64("clusterID", uint64(cluster.ID)), zap.Error(err))
			errs = multierr.Append(errs, err)
			continue
		}

		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
		}
		for _, restore := range restores {
			ns := runtimeutil.TiDBNamespace(restore.ClusterID.Uint64())
			if restore.Status == entity.RestoreStatus_RUNNING || restore.Status == entity.RestoreStatus_PENDING {
				if c.infraAPI.InfraAPIConfig.EnableBackupRestoreRead {
					restoreCR, err := c.infraAPI.GetRestore(context.TODO(), cluster.ProjectID.Uint64(), restoreCRDName(&restore))
					if err != nil {
						errs = multierr.Append(errs, err)
					}
					errs = multierr.Append(errs, c.UpdateRestoreProgressForInfra(restoreCR, &restore))
				} else {
					restoreCR, err := rx.Runtime.GetRestore(restoreCRDName(&restore), ns.String())
					if err != nil {
						errs = multierr.Append(errs, err)
					}
					errs = multierr.Append(errs, c.UpdateRestoreProgress(restoreCR, &restore))
				}
			}
			if restore.Status == entity.RestoreStatus_SUCCESS {
				errs = multierr.Append(errs, db.Transaction(func(tx *gorm.DB) error {
					return dbutil.ClusterUpdateStatus(tx, cluster.ID, entity.ClusterStatusNormal)
				}))
				if c.infraAPI.InfraAPIConfig.EnableBackupRestoreWrite {
					errs = multierr.Append(errs, c.infraAPI.DeleteRestoreIgnoreNotFound(context.TODO(), cluster.ProjectID.Uint64(), restoreCRDName(&restore)))
				} else {
					errs = multierr.Append(errs, c.deleteRestoreCRD(cluster.ProjectID, provider.Name, region.Name, restoreCRDName(&restore), ns.String()))
				}
			} else {
				errs = multierr.Append(errs, db.Transaction(func(tx *gorm.DB) error {
					return dbutil.ClusterUpdateStatus(tx, cluster.ID, entity.ClusterStatusRecovering)
				}))
			}
		}
	}
	if errs != nil {
		c.Log().Error("restore failed", zap.Error(errs))
	}
	return errs
}