package backup

import (
	"context"
	"fmt"
	"net/url"
	"reflect"
	"time"

	"github.com/go-logr/logr"
	"github.com/pingcap/errors"
	configsdk "github.com/tidbcloud/config-service/sdk"
	"github.com/tidbcloud/config-service/sdk/api"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	corev1 "k8s.io/api/core/v1"
	kerrors "k8s.io/apimachinery/pkg/api/errors"
	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/pointer"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/builder"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/predicate"

	"github.com/tidbcloud/infra-api/controllers/common"
	tracerutil "github.com/tidbcloud/infra-api/pkg/tracer"

	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"

	"github.com/tidbcloud/infra-api/controllers/reconciler"
	"github.com/tidbcloud/infra-api/controllers/utils"
	"github.com/tidbcloud/infra-api/pkg/consts"
	"github.com/tidbcloud/infra-api/pkg/gardener"
	metricsctrl "github.com/tidbcloud/infra-api/pkg/metrics/controller"
	"github.com/tidbcloud/infra-api/pkg/provider"
)

const (
	BRStatusAddr = "0.0.0.0:8089"
)

const (
	BackupFinalizer = "backup.cloud.pingcap.com/finalizer"
	errNotBackup    = "not backup resource"
)

var tracer = otel.Tracer("BackupRestoreReconciler")

// BackupReconciler reconciles a Backup object
type BackupReconciler struct {
	reconciler *reconciler.Reconciler
	Log        logr.Logger
}

// NewBackupReconciler creates a BackupReconciler instance, clusterCli will be nil for non-shared cluster.
func NewBackupReconciler(mgr ctrl.Manager, log logr.Logger, g *gardener.GardEnvironment, clusterCli client.Client, configService configsdk.Interface, p provider.Backup) *BackupReconciler {
	return &BackupReconciler{
		Log: log,
		reconciler: &reconciler.Reconciler{
			Client: mgr.GetClient(),
			ExternalObj: backupConnector{
				g:          g,
				client:     mgr.GetClient(),
				clusterCli: clusterCli,
				configSvc:  configService,
				provider:   p,
			},
			NewResource: func() client.Object {
				return &cloudv1.Backup{}
			},
		},
	}
}

//+kubebuilder:rbac:groups=cloud.pingcap.com,resources=backups,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=cloud.pingcap.com,resources=backups/status,verbs=get;update;patch
//+kubebuilder:rbac:groups=cloud.pingcap.com,resources=backups/finalizers,verbs=update

func (r *BackupReconciler) Reconcile(ctx context.Context, req ctrl.Request) (_ ctrl.Result, errResult error) {
	start := time.Now()

	ctx, span := tracer.Start(ctx, "BackupReconcile")
	defer func() {
		tracerutil.EndSpan(span, errResult)
		metricsctrl.ReconcileDuration.WithLabelValues("backup").Observe(time.Since(start).Seconds())
	}()

	span.SetAttributes(attribute.String("backup_name", req.Name))

	logger := r.Log.WithValues("backup", req.NamespacedName)
	ctx = log.IntoContext(ctx, logger)
	return r.reconciler.Reconcile(ctx, req, BackupFinalizer)
}

type backupConnector struct {
	g      gardener.GardenerAPI
	client client.Client

	// client to the target cluster, only used for the shared K8s cluster now.
	// if this client is set, no need to try to get a client for a shoot (because we are running for a non-gardener based cluster)
	clusterCli client.Client
	configSvc  configsdk.Interface

	provider provider.Backup
}

func (bc backupConnector) Connect(ctx context.Context, obj client.Object) (_ reconciler.ExternalObjClient, errResult error) {
	ctx, span := tracer.Start(ctx, "backupConnector.Connect")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	b, ok := obj.(*cloudv1.Backup)
	if !ok {
		return nil, errors.New(errNotBackup)
	}

	var c cloudv1.Cluster
	namespacedName := types.NamespacedName{
		Name:      b.Spec.ClusterName,
		Namespace: b.ObjectMeta.Namespace,
	}
	err := bc.client.Get(ctx, namespacedName, &c)
	if err != nil {
		if kerrors.IsNotFound(err) { // support deletion without cluster
			return bc.partConnectWithoutCluster(ctx, b)
		}
		return nil, err
	}

	span.SetAttributes(
		attribute.String("backup_name", b.Name),
		attribute.String("backup_phase", string(b.Status.Phase)),
	)
	tracerutil.SetCommonAttributes(span, &c)

	// TODO(aylei,fangfang): check cluster phase before connecting
	var k8sStatus *cloudv1.KubernetesStatus
	switch c.Labels[cloudv1.ProviderTypeLabel] {
	case consts.ProviderTypeAWS:
		awsStatus, err := cloudv1.AwsClusterStatusFromCluster(&c)
		if err != nil {
			return nil, err
		}
		k8sStatus = awsStatus.Kubernetes
	case consts.ProviderTypeGCP:
		gcpStatus, err := cloudv1.GcpClusterStatusFromCluster(&c)
		if err != nil {
			return nil, err
		}
		k8sStatus = gcpStatus.Kubernetes
	case consts.ProviderTypeAWSFreeTier:
		awsFreeTierStatus, err := cloudv1.AwsFreeTierClusterStatusFromCluster(&c)
		if err != nil {
			return nil, err
		}
		k8sStatus = awsFreeTierStatus.Kubernetes
	case consts.ProviderTypeAWSSharedTier, consts.ProviderTypeGCPSharedTier:
		sharedTierStatus, err := cloudv1.SharedTierClusterStatusFromCluster(&c)
		if err != nil {
			return nil, err
		}
		k8sStatus = sharedTierStatus.Kubernetes
	case consts.ProviderTypeEKS, consts.ProviderTypeGKE:
		k8sStatus, err = bc.provider.GetKubernetesStatus(&c)
		if err != nil {
			return nil, err
		}
	default:
		return nil, fmt.Errorf("unknown provider %s", c.Labels[cloudv1.ProviderTypeLabel])
	}
	if k8sStatus == nil {
		return nil, fmt.Errorf("k8s status is not yet set by provider %s", c.Labels[cloudv1.ProviderTypeLabel])
	}

	var n cloudv1.Network
	networkNamespaceName := types.NamespacedName{
		Name:      c.Spec.NetworkName,
		Namespace: c.ObjectMeta.Namespace,
	}
	err = bc.client.Get(ctx, networkNamespaceName, &n)
	if err != nil {
		return nil, err
	}

	provider := new(cloudv1.Provider)
	err = bc.client.Get(ctx, types.NamespacedName{Name: n.Spec.Provider}, provider)
	if err != nil {
		return nil, err
	}

	var (
		clusterCli client.Client
		shootName  string
	)
	switch c.Labels[cloudv1.ProviderTypeLabel] {
	case consts.ProviderTypeAWS, consts.ProviderTypeGCP:
		networkStatus, err := cloudv1.GardenerNetworkStatusFromNetwork(&n)
		if err != nil {
			return nil, err
		}
		shootName = networkStatus.ShootName
		shoot, err := bc.g.GetShootContext(ctx, networkStatus.ShootName)
		if err != nil {
			return nil, err
		}
		clusterCli = shoot.Manager.GetClient()
	case consts.ProviderTypeAWSFreeTier, consts.ProviderTypeAWSSharedTier, consts.ProviderTypeGCPSharedTier:
		clusterCli = bc.clusterCli // this should already be set when initialized the `backupConnector`
	case consts.ProviderTypeEKS, consts.ProviderTypeGKE:
		clusterCli, err = bc.provider.GetClient(ctx, &c)
		if err != nil {
			return nil, err
		}
	}

	return &clusterBackup{
		shootName:      shootName, // only set for shoot based cluster.
		clusterCli:     clusterCli,
		cli:            bc.client,
		cluster:        &c,
		network:        &n,
		provider:       provider,
		k8sStatus:      k8sStatus,
		configSvc:      bc.configSvc,
		backupProvider: bc.provider,
	}, nil
}

func (bc backupConnector) partConnectWithoutCluster(ctx context.Context, b *cloudv1.Backup) (_ reconciler.ExternalObjClient, errResult error) {
	ctx, span := tracer.Start(ctx, "backupConnector.partConnectWithoutCluster")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	var (
		clusterCli client.Client
		namespace  string
		shootName  string
	)
	if b.Labels[cloudv1.ProviderTypeLabel] == consts.ProviderTypeEKS || b.Labels[cloudv1.ProviderTypeLabel] == consts.ProviderTypeGKE {
		eksCli, err := bc.provider.GetBackupClient(ctx, b)
		if err != nil {
			return nil, err
		}
		clusterCli = eksCli
	} else if bc.clusterCli == nil { // if clusterCli not set by the caller, we treat it as gardener based shoot cluster
		status, err := cloudv1.ShootBackupStatusFromBackup(b)
		if err != nil {
			return nil, err
		}
		namespace = status.Namespace
		shootName = status.ShootName
		shoot, err := bc.g.GetShootContext(ctx, status.ShootName)
		if err != nil {
			if err != gardener.ErrShootNotReady {
				return nil, err
			} // nil clusterCli for ErrShootNotReady
		} else {
			clusterCli = shoot.Manager.GetClient()
		}
	} else {
		status, err := cloudv1.SharedBackupStatusFromBackup(b)
		if err != nil {
			return nil, err
		}
		namespace = status.Namespace
		clusterCli = bc.clusterCli
	}

	return &clusterBackup{ // only part of the fields are valid
		clusterCli:     clusterCli,
		cli:            bc.client,
		shootName:      shootName, // only set for shoot based cluster.
		k8sStatus:      &cloudv1.KubernetesStatus{Namespace: namespace},
		backupProvider: bc.provider,
	}, nil
}

type clusterBackup struct {
	shootName  string // only set for shoot based cluster.
	cli        client.Client
	clusterCli client.Client
	cluster    *cloudv1.Cluster
	network    *cloudv1.Network
	provider   *cloudv1.Provider

	k8sStatus *cloudv1.KubernetesStatus
	configSvc configsdk.Interface

	backupProvider provider.Backup
}

func (cb *clusterBackup) Observe(ctx context.Context, obj client.Object) (_ reconciler.ExternalObservation, errResult error) {
	ctx, span := tracer.Start(ctx, "clusterBackup.Observe")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	b, ok := obj.(*cloudv1.Backup)
	if !ok {
		return reconciler.ExternalObservation{}, errors.New(errNotBackup)
	}

	span.SetAttributes(
		attribute.String("backup_name", b.Name),
		attribute.String("backup_phase", string(b.Status.Phase)),
	)
	tracerutil.SetCommonAttributes(span, cb.cluster)

	if b.Status.Phase == cloudv1.BackupComplete || b.Status.Phase == cloudv1.BackupFailed {
		// final state, nothing to do
		return reconciler.ExternalObservation{ResourceUpToDate: true, ResourceExists: true}, nil
	}

	if cb.clusterCli == nil {
		// fast fail if the shoot is not ready (or deleted)
		b.Status.Phase = cloudv1.BackupFailed
		return reconciler.ExternalObservation{ResourceUpToDate: true}, cb.cli.Status().Update(ctx, b)
	}

	var tidbBackup v1alpha1.Backup
	if err := cb.clusterCli.Get(ctx, types.NamespacedName{
		Name:      b.ObjectMeta.Name,
		Namespace: cb.k8sStatus.Namespace,
	}, &tidbBackup); err != nil {
		if reconciler.IgnoreNotFound(err) != nil {
			return reconciler.ExternalObservation{}, err
		}

		b.Status.Phase = cloudv1.BackupPrepare
		return reconciler.ExternalObservation{ResourceExists: false}, cb.cli.Status().Update(ctx, b)
	}

	if tidbBackup.Status.Phase == v1alpha1.BackupComplete || tidbBackup.Status.Phase == v1alpha1.BackupFailed {
		b.Status.BackupSize = tidbBackup.Status.BackupSize
		b.Status.BackupTS = tidbBackup.Status.CommitTs
		b.Status.Phase = convertBackupPhase(tidbBackup.Status.Phase)
		return reconciler.ExternalObservation{ResourceExists: true, ResourceUpToDate: true}, cb.cli.Status().Update(ctx, b)
	}

	return reconciler.ExternalObservation{ResourceExists: true}, nil
}

func (cb *clusterBackup) Create(ctx context.Context, obj client.Object) (errResult error) {
	ctx, span := tracer.Start(ctx, "clusterBackup.Create")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	b, ok := obj.(*cloudv1.Backup)
	if !ok {
		return errors.New(errNotBackup)
	}

	if cb.cluster == nil { // cluster not exists
		b.Status.Phase = cloudv1.BackupFailed
		return cb.cli.Status().Update(ctx, b)
	}

	span.SetAttributes(
		attribute.String("backup_name", b.Name),
		attribute.String("backup_phase", string(b.Status.Phase)),
	)
	tracerutil.SetCommonAttributes(span, cb.cluster)

	if cb.cluster.Status.ObjectStore == nil {
		return errors.Errorf("empty ObjectStore for %s", cb.cluster.Name)
	}
	if cb.cluster.Status.ObjectStore.GCS == nil && cb.cluster.Status.ObjectStore.S3 == nil {
		return errors.Errorf("empty cloud ObjectStore for %s", cb.cluster.Name)
	}

	vars := TidbBackupVars{
		BackupName:      b.ObjectMeta.Name,
		BackupNamespace: cb.k8sStatus.Namespace,
		ClusterName:     utils.ClusterPrefix,
		Region:          cb.network.Spec.Region,
		BackupPrefix:    b.ObjectMeta.Name, // same as BackupName
	}
	if b.Spec.CommitTS != nil {
		vars.CommitTS = *b.Spec.CommitTS
	}
	if b.Spec.Mode != nil {
		vars.Mode = *b.Spec.Mode
	}

	// TODO(aylei): gitops should abandon the concept of specific business
	tenantId := utils.ParsingLegacyId(cb.cluster, consts.TenantTagKey)
	if tenantId == nil {
		dummyId := uint64(0)
		tenantId = &dummyId
	}
	clusterId := utils.ParsingLegacyId(cb.cluster, consts.ClusterTagKey)
	brConfigOutput, err := cb.configSvc.DescribeBackupConfiguration(&api.DescribeBRConfigInput{
		TenantID:  tenantId,
		ClusterID: clusterId,
	})
	if err != nil {
		return errors.Annotate(err, "get gitops br config")
	}
	brConfig, err := BRConfig(brConfigOutput)
	if err != nil {
		return err
	}

	// append custom options
	brConfig.Options = append(brConfig.Options, b.Spec.Options...)

	labels := utils.CopyLabelsForWorkloads(b.Labels)
	annotations := make(map[string]string)
	var tidbBackup *v1alpha1.Backup

	providerType := cb.cluster.ObjectMeta.Labels[cloudv1.ProviderTypeLabel]
	switch providerType {
	case consts.ProviderTypeGCP:
		providerStatus, err := cloudv1.GardenerProviderSpecForProvider(cb.provider)
		if err != nil {
			return err
		}
		vars.Bucket = useCustomBucketIfSpecInBackupObject(b, providerType, cb.cluster.Status.ObjectStore.GCS.Path)

		tidbBackup = assembleGcpBackupCRD(vars, providerStatus.GCP.ProjectID, annotations, labels)
		tidbBackup = mergeBackupConfig(tidbBackup, brConfig, brConfigOutput)
	case consts.ProviderTypeAWS:
		vars.Bucket = cb.cluster.Status.ObjectStore.S3.Path
		annotations := make(map[string]string)
		tidbBackup = assembleAWSBackup(vars, annotations, labels)
		tidbBackup = mergeBackupConfig(tidbBackup, brConfig, brConfigOutput)
	case consts.ProviderTypeAWSFreeTier, consts.ProviderTypeAWSSharedTier:
		// use the `cluster.cloud.pingcap.com` name as the `tidbcluster.pingcap.com` name BUT with a `db` prefix
		vars.ClusterName = "db" + cb.cluster.Name
		vars.Bucket = cb.cluster.Status.ObjectStore.S3.Path
		vars.PodSecurityContext = &corev1.PodSecurityContext{
			RunAsUser:  pointer.Int64Ptr(1000),
			RunAsGroup: pointer.Int64Ptr(2000),
			FSGroup:    pointer.Int64Ptr(2000),
		}
		// do not evit workload pods with cluster-autoscaler, https://github.com/kubernetes/autoscaler/tree/master/cluster-autoscaler
		annotations = map[string]string{
			"cluster-autoscaler.kubernetes.io/safe-to-evict": "false",
		}
		tidbBackup = assembleFreeSharedTierAWSBackup(vars, annotations, labels)
		tidbBackup = mergeBackupConfig(tidbBackup, brConfig, brConfigOutput)
		// for AWS free-tier and shared-tier, add 'skip-check-path' for less S3 permissions.
		// NB: 'skip-check-path' is deprecated in BR and will be removed in the future.
		tidbBackup.Spec.BR.Options = append(tidbBackup.Spec.BR.Options, "--skip-check-path=true")
	case consts.ProviderTypeGCPSharedTier:
		sharedTierClusterStatus, err := cloudv1.SharedTierClusterStatusFromCluster(cb.cluster)
		if err != nil {
			return err
		}
		// use the `cluster.cloud.pingcap.com` name as the `tidbcluster.pingcap.com` name BUT with a `db` prefix
		vars.ClusterName = "db" + cb.cluster.Name
		vars.Bucket = cb.cluster.Status.ObjectStore.GCS.Path
		vars.PodSecurityContext = &corev1.PodSecurityContext{
			RunAsUser:  pointer.Int64Ptr(1000),
			RunAsGroup: pointer.Int64Ptr(2000),
			FSGroup:    pointer.Int64Ptr(2000),
		}
		// do not evit workload pods with cluster-autoscaler, https://github.com/kubernetes/autoscaler/tree/master/cluster-autoscaler
		annotations = map[string]string{
			"cluster-autoscaler.kubernetes.io/safe-to-evict": "false",
		}
		tidbBackup = assembleGcpBackupCRD(vars, *sharedTierClusterStatus.GCPProjectID, annotations, labels)
		tidbBackup = mergeBackupConfig(tidbBackup, brConfig, brConfigOutput)
	case consts.ProviderTypeEKS, consts.ProviderTypeGKE:
		tidbBackup, err = cb.backupProvider.BuildBackup(b, cb.cluster)
		if err != nil {
			return err
		}
	}

	switch providerType {
	case consts.ProviderTypeAWS, consts.ProviderTypeGCP:
		if err := cloudv1.MutateShootBackupStatus(b, func(status *cloudv1.ShootBackupStatus) error {
			status.ShootName = cb.shootName
			status.Namespace = cb.k8sStatus.Namespace
			return nil
		}); err != nil {
			return err
		}
	case consts.ProviderTypeAWSFreeTier, consts.ProviderTypeAWSSharedTier, consts.ProviderTypeGCPSharedTier:
		if err := cloudv1.MutateSharedBackupStatus(b, func(status *cloudv1.SharedBackupStatus) error {
			status.Namespace = cb.k8sStatus.Namespace
			return nil
		}); err != nil {
			return err
		}
	}

	// we need to update the status before create the underlying CRD, otherwise if the status update failed,
	// then this `Create` will be skipped (without status update again) in the next round of reconcile after `Observe`.
	if vars.Bucket != "" {
		b.Status.BucketName = vars.Bucket
	}
	b.Status.Phase = cloudv1.BackupPrepare
	if err := cb.cli.Status().Update(ctx, b); err != nil {
		return err
	}

	_, err = controllerutil.CreateOrUpdate(ctx, cb.clusterCli, tidbBackup, func() error {
		return nil
	})
	return err
}

func useCustomBucketIfSpecInBackupObject(b *cloudv1.Backup, providerType string, defaultBucket string) string {
	// storage uri for gcp was specified by dataflow-service
	// validate code in api/v1/backup_helpers.go
	if b.Spec.Storage != nil && *b.Spec.Storage != "" {
		uri, _ := url.Parse(*b.Spec.Storage)
		return uri.Host
	} else {
		return defaultBucket
	}
}

func 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
}

func (cb *clusterBackup) Update(ctx context.Context, obj client.Object) (errResult error) {
	ctx, span := tracer.Start(ctx, "clusterBackup.Update")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	b, ok := obj.(*cloudv1.Backup)
	if !ok {
		return errors.New(errNotBackup)
	}

	if cb.cluster == nil { // cluster not exists any more
		b.Status.Phase = cloudv1.BackupFailed
		return cb.cli.Status().Update(ctx, b)
	}

	span.SetAttributes(
		attribute.String("backup_name", b.Name),
		attribute.String("backup_phase", string(b.Status.Phase)),
	)
	tracerutil.SetCommonAttributes(span, cb.cluster)

	var tidbBackup v1alpha1.Backup
	if err := cb.clusterCli.Get(ctx, types.NamespacedName{
		Name:      b.ObjectMeta.Name,
		Namespace: cb.k8sStatus.Namespace,
	}, &tidbBackup); err != nil {
		return err
	}

	if err := cb.syncInfraBackupSpec2TiDB(ctx, b, tidbBackup.DeepCopy()); err != nil {
		return err
	}

	if err := cb.syncTiDBBackupStatus2Infra(ctx, &tidbBackup, b); err != nil {
		return err
	}
	return nil
}

func (cb *clusterBackup) syncInfraBackupSpec2TiDB(ctx context.Context, infraBackup *cloudv1.Backup, tidbBackup *v1alpha1.Backup) error {
	updated := false
	if infraBackup.Spec.LogTruncateUntil != nil && tidbBackup.Spec.LogTruncateUntil != *infraBackup.Spec.LogTruncateUntil {
		tidbBackup.Spec.LogTruncateUntil = *infraBackup.Spec.LogTruncateUntil
		updated = true
	}
	if infraBackup.Spec.LogStop != nil && tidbBackup.Spec.LogStop != *infraBackup.Spec.LogStop {
		tidbBackup.Spec.LogStop = *infraBackup.Spec.LogStop
		updated = true
	}

	if !updated {
		return nil
	}

	return cb.clusterCli.Update(ctx, tidbBackup)
}

func (cb *clusterBackup) syncTiDBBackupStatus2Infra(ctx context.Context, tidbBackup *v1alpha1.Backup, infraBackup *cloudv1.Backup) error {
	copiedInfraBackup := infraBackup.DeepCopy()
	copiedInfraBackup.Status.BackupSize = tidbBackup.Status.BackupSize
	copiedInfraBackup.Status.BackupTS = tidbBackup.Status.CommitTs
	copiedInfraBackup.Status.BackupPath = tidbBackup.Status.BackupPath
	copiedInfraBackup.Status.Phase = convertBackupPhase(tidbBackup.Status.Phase)
	copiedInfraBackup.Status.LogCheckpointTs = tidbBackup.Status.LogCheckpointTs
	copiedInfraBackup.Status.LogTruncateUntil = tidbBackup.Status.LogSuccessTruncateUntil

	if reflect.DeepEqual(infraBackup.Status, copiedInfraBackup.Status) {
		return nil
	}
	return cb.cli.Status().Update(ctx, copiedInfraBackup)
}

func convertBackupPhase(conditionType v1alpha1.BackupConditionType) cloudv1.BackupConditionType {
	switch conditionType {
	case v1alpha1.BackupScheduled:
		return cloudv1.BackupScheduled
	case v1alpha1.BackupPrepare:
		return cloudv1.BackupPrepare
	case v1alpha1.BackupRunning:
		return cloudv1.BackupRunning
	case v1alpha1.BackupComplete:
		return cloudv1.BackupComplete
	case v1alpha1.BackupClean:
		return cloudv1.BackupClean
	case v1alpha1.BackupFailed, v1alpha1.BackupInvalid:
		return cloudv1.BackupFailed
	case v1alpha1.BackupRetryFailed:
		return cloudv1.BackupPrepare // retryable error as prepare
	default:
		return cloudv1.BackupPrepare // empty phase as prepare
	}
}

func ConvertBackupMode(mode cloudv1.BackupMode) v1alpha1.BackupMode {
	switch mode {
	case cloudv1.BackupModeSnapshot:
		return v1alpha1.BackupModeSnapshot
	case cloudv1.BackupModeLog:
		return v1alpha1.BackupModeLog
	default:
		return ""
	}
}

func (cb *clusterBackup) Delete(ctx context.Context, obj client.Object) (_ reconciler.ExternalDeletion, errResult error) {
	ctx, span := tracer.Start(ctx, "clusterBackup.Delete")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	b, ok := obj.(*cloudv1.Backup)
	if !ok {
		return reconciler.ExternalDeletion{}, errors.New(errNotBackup)
	}

	span.SetAttributes(
		attribute.String("backup_name", b.Name),
		attribute.String("backup_phase", string(b.Status.Phase)),
	)
	tracerutil.SetCommonAttributes(span, cb.cluster)

	// wait for all restores depends on this backup to be complete or failed
	var restoreList cloudv1.RestoreList
	err := cb.cli.List(ctx, &restoreList, client.InNamespace(b.Namespace), client.MatchingFields(map[string]string{
		"spec.from.backup": b.Name,
	}))
	if err != nil {
		return reconciler.ExternalDeletion{}, err
	}
	var restoreNames []string
	for _, restore := range restoreList.Items {
		if restore.Status.Phase != cloudv1.RestoreComplete && restore.Status.Phase != cloudv1.RestoreFailed {
			restoreNames = append(restoreNames, restore.Name)
		}
	}
	if len(restoreNames) > 0 {
		return reconciler.ExternalDeletion{}, errors.Errorf("wait for restores %v to be complete or failed", restoreNames)
	}

	// backupNS is the cluster namespace in tenant's k8s
	var backupNS string
	switch b.Labels[cloudv1.ProviderTypeLabel] {
	case consts.ProviderTypeEKS, consts.ProviderTypeGKE:
		ns, err := cb.backupProvider.GetBackupNamespace(ctx, b)
		if err != nil {
			return reconciler.ExternalDeletion{}, err
		}
		backupNS = ns
	case consts.ProviderTypeAWS, consts.ProviderTypeGCP, consts.ProviderTypeAWSSharedTier, consts.ProviderTypeAWSFreeTier:
		backupNS = cb.k8sStatus.Namespace
	default:
		return reconciler.ExternalDeletion{}, fmt.Errorf("unrecognized provider '%s'", b.Labels[cloudv1.ProviderTypeLabel])
	}

	// support to delete the backup even if the cluster already deleted (but the shoot/clusterCli is still valid)
	var tidbBackup v1alpha1.Backup
	if err := cb.clusterCli.Get(ctx, types.NamespacedName{
		Name:      b.ObjectMeta.Name,
		Namespace: backupNS,
	}, &tidbBackup); err != nil {
		if reconciler.IgnoreNotFound(err) != nil {
			return reconciler.ExternalDeletion{}, err
		}

		return reconciler.ExternalDeletion{ResourceDeleted: true}, nil
	}

	if !tidbBackup.GetDeletionTimestamp().IsZero() {
		return reconciler.ExternalDeletion{ResourceDeleted: false}, nil
	}

	return reconciler.ExternalDeletion{ResourceDeleted: false}, cb.clusterCli.Delete(ctx, &tidbBackup)
}

type TidbBackupVars struct {
	BackupName      string
	BackupNamespace string
	ClusterName     string
	Region          string
	Bucket          string
	BackupPrefix    string
	Mode            cloudv1.BackupMode
	CommitTS        string

	PodSecurityContext *corev1.PodSecurityContext
}

// assembleFreeSharedTierAWSBackup assembles the backup command both for ProviderTypeAWSFreeTier and ProviderTypeAWSSharedTier.
func assembleFreeSharedTierAWSBackup(vars TidbBackupVars, annotations map[string]string, labels map[string]string) *v1alpha1.Backup {
	backup := assembleAWSBackup(vars, annotations, labels)
	backup.Spec.CleanPolicy = v1alpha1.CleanPolicyTypeRetain

	return backup
}

func assembleAWSBackup(vars TidbBackupVars, annotations map[string]string, labels map[string]string) *v1alpha1.Backup {
	concurrency := uint32(32)
	backup := &v1alpha1.Backup{
		ObjectMeta: v1.ObjectMeta{
			Name:        vars.BackupName,
			Namespace:   vars.BackupNamespace,
			Annotations: annotations,
			Labels:      labels,
		},
		Spec: v1alpha1.BackupSpec{
			Type:        v1alpha1.BackupTypeFull,
			Mode:        ConvertBackupMode(vars.Mode),
			CommitTs:    vars.CommitTS,
			CleanPolicy: v1alpha1.CleanPolicyTypeDelete,
			StorageProvider: v1alpha1.StorageProvider{
				S3: &v1alpha1.S3StorageProvider{
					Provider: v1alpha1.S3StorageProviderTypeAWS,
					Region:   vars.Region,
					Bucket:   vars.Bucket,
					Prefix:   vars.BackupPrefix,
				},
			},
			BR: &v1alpha1.BRConfig{
				Cluster:          vars.ClusterName,
				ClusterNamespace: vars.BackupNamespace,
				SendCredToTikv:   pointer.BoolPtr(false),
				Concurrency:      &concurrency,
				StatusAddr:       BRStatusAddr,
			},
			UseKMS:             false,
			PodSecurityContext: vars.PodSecurityContext,
		},
	}

	return backup
}

func assembleGcpBackupCRD(vars TidbBackupVars, gcpProject string, annotations map[string]string, labels map[string]string) *v1alpha1.Backup {
	concurrency := uint32(32)
	return &v1alpha1.Backup{
		ObjectMeta: v1.ObjectMeta{
			Name:        vars.BackupName,
			Namespace:   vars.BackupNamespace,
			Annotations: annotations,
			Labels:      labels,
		},
		Spec: v1alpha1.BackupSpec{
			Type:        v1alpha1.BackupTypeFull,
			Mode:        ConvertBackupMode(vars.Mode),
			CommitTs:    vars.CommitTS,
			CleanPolicy: v1alpha1.CleanPolicyTypeDelete,
			StorageProvider: v1alpha1.StorageProvider{
				Gcs: &v1alpha1.GcsStorageProvider{
					ProjectId:  gcpProject,
					Location:   vars.Region,
					Bucket:     vars.Bucket,
					Prefix:     vars.BackupPrefix,
					SecretName: consts.BRGCPSecretName,
				},
			},
			BR: &v1alpha1.BRConfig{
				Cluster:          vars.ClusterName,
				ClusterNamespace: vars.BackupNamespace,
				SendCredToTikv:   pointer.BoolPtr(true),
				Concurrency:      &concurrency,
				StatusAddr:       BRStatusAddr,
			},
			UseKMS:             false,
			PodSecurityContext: vars.PodSecurityContext,
		},
	}
}

// SetupWithManager sets up the controller with the Manager.
func (r *BackupReconciler) SetupWithManager(mgr ctrl.Manager, p ...predicate.Predicate) error {
	// restore should be indexed by .spec.from.backup
	if err := mgr.GetFieldIndexer().IndexField(context.TODO(), &cloudv1.Restore{}, "spec.from.backup", func(obj client.Object) []string {
		restore := obj.(*cloudv1.Restore)
		return []string{restore.Spec.From.Backup}
	}); err != nil {
		return err
	}

	return ctrl.NewControllerManagedBy(mgr).
		For(&cloudv1.Backup{}, builder.WithPredicates(p...)).
		WithOptions(controller.Options{RateLimiter: common.DefaultRateLimiter()}).
		WithEventFilter(utils.GetMigrationAnnotationSkipPredicator()).
		Complete(r)
}