package backup

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

	"github.com/go-logr/logr"
	"github.com/pingcap/errors"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	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"
	scheduling_v1 "k8s.io/api/scheduling/v1"
	kerrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	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"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-api/controllers/common"
	"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"
	tracerutil "github.com/tidbcloud/infra-api/pkg/tracer"
)

const (
	BRStatusAddr = "0.0.0.0:8089"
)

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

	ClusterCriticalPriorityClass = "tidb-cluster-backup-critical"
	ClusterCriticalPriorityValue = 900000000
)

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{}
			},
			ReconcileTimeout: 3 * time.Minute,
		},
	}
}

func NewBackupReconcilerCompat(infraCli client.Client, log logr.Logger, g *gardener.GardEnvironment, clusterCli client.Client,
	configService configsdk.Interface, p provider.Backup) *BackupReconciler {
	return &BackupReconciler{
		Log: log,
		reconciler: &reconciler.Reconciler{
			Client: infraCli,
			ExternalObj: backupConnector{
				g:          g,
				client:     infraCli,
				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 := c.GetNetworkRef()
	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
		isDataPlaneK8sCannotConnect  bool
		isDataPlaneNamespaceNotExist bool
	)
	if b.Labels[cloudv1.ProviderTypeLabel] == consts.ProviderTypeEKS || b.Labels[cloudv1.ProviderTypeLabel] == consts.ProviderTypeGKE {
		bns, err := bc.provider.GetBackupNamespace(ctx, b)
		if err != nil {
			if reconciler.IgnoreNotFound(err) != nil {
				return nil, err
			}
		} else {
			namespace = bns
		}
		eksCli, err := bc.provider.GetBackupClient(ctx, b)
		if err != nil {
			// there maybe has no client when network was deleted
			if reconciler.IgnoreNotFound(err) != nil {
				return nil, err
			}
			isDataPlaneK8sCannotConnect = true
		} else {
			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 && reconciler.IgnoreNotFound(err) != nil {
				return nil, err
			}
			isDataPlaneK8sCannotConnect = true
		} else {
			clusterCli = shoot.Manager.GetClient()
		}
	} else {
		status, err := cloudv1.SharedBackupStatusFromBackup(b)
		if err != nil {
			return nil, err
		}
		namespace = status.Namespace
		clusterCli = bc.clusterCli
	}

	// check namespace
	if !isDataPlaneK8sCannotConnect {
		isExist, err := checkNamespaceExistInDataPlane(ctx, clusterCli, namespace)
		if err != nil {
			return nil, err
		}
		isDataPlaneNamespaceNotExist = !isExist
	}

	n := new(cloudv1.Network)
	networkName := fmt.Sprintf("%s-%s", b.Labels[cloudv1.ProviderTypeLabel], b.Labels[cloudv1.RegionLabel])
	networkNamespaceName := types.NamespacedName{
		Namespace: b.Namespace,
		Name:      networkName,
	}
	if err := bc.client.Get(ctx, networkNamespaceName, n); err != nil {
		return nil, fmt.Errorf("get network %s error: %w", networkName, err)
	}

	return &clusterBackup{ // only part of the fields are valid
		network:                      n,
		clusterCli:                   clusterCli,
		cli:                          bc.client,
		shootName:                    shootName, // only set for shoot based cluster.
		k8sStatus:                    &cloudv1.KubernetesStatus{Namespace: namespace},
		backupProvider:               bc.provider,
		isDataPlaneK8sCannotConnect:  isDataPlaneK8sCannotConnect,
		isDataPlaneNamespaceNotExist: isDataPlaneNamespaceNotExist,
	}, 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

	// isDataPlaneK8sCannotConnect and isDataPlaneNamespaceNotExist are caused by tidb cluster deletion, and we will mark the running backup as failed,
	// only when tidb cluster in control plane doesn't exist, we will check them, otherwise, we think they are in normal state.
	isDataPlaneK8sCannotConnect  bool
	isDataPlaneNamespaceNotExist bool
}

func (cb *clusterBackup) inNewArch() bool {
	return networkInNewArch(cb.network)
}

func (cb *clusterBackup) isGKEProvider() bool {
	return cb.network.Spec.Provider == consts.ProviderTypeGKE
}

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)

	// in final state, delete node group
	inFinalState := b.Status.Phase == cloudv1.BackupComplete || b.Status.Phase == cloudv1.BackupFailed || b.LogBackupStopped()
	// if log backup is started, and no log stop or log truncate command, delete node group
	isLogBackupStarted := b.LogBackupStarted() && !b.InLogBackupTruncateCommand() && !b.InLogBackupStopCommand()
	if (inFinalState || isLogBackupStarted) && cb.needUseNodeGroup(b) {
		ngName := backupNodeGroupName(b, cb.inNewArch(), cb.isGKEProvider())
		log.FromContext(ctx).Info(fmt.Sprintf("clean node group for backup ng name: %s", ngName))
		if err := cb.backupProvider.DeleteBRNode(ctx, cb.network.DeepCopy(), ngName); err != nil {
			log.FromContext(ctx).Error(
				err, fmt.Sprintf("clean node group for backup error, ng name: %s", ngName))
			return
		}
	}

	if inFinalState {
		return reconciler.ExternalObservation{ResourceUpToDate: true, ResourceExists: true}, nil
	}

	// fast fail if dataplane k8s can not connect or namespace doesn't exist caused by tidb cluster deletion
	// or cluster cr is not found
	if cb.isDataPlaneK8sCannotConnect || cb.isDataPlaneNamespaceNotExist || cb.cluster == nil {
		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 ||
		v1alpha1.IsLogBackupAlreadyStop(&tidbBackup) {
		return reconciler.ExternalObservation{ResourceExists: true, ResourceUpToDate: true}, cb.syncTiDBBackupStatus2Infra(ctx, &tidbBackup, 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)
	}

	// fast fail if dataplane k8s can not connect or namespace doesn't exist caused by tidb cluster deletion
	// or cluster cr is not found
	if cb.isDataPlaneK8sCannotConnect || cb.isDataPlaneNamespaceNotExist || cb.cluster == nil {
		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)

	providerType := cb.cluster.ObjectMeta.Labels[cloudv1.ProviderTypeLabel]
	var tidbBackup *v1alpha1.Backup
	var err error
	if providerType == consts.ProviderTypeEKS || providerType == consts.ProviderTypeGKE {
		tidbBackup, err = cb.buildTiDBBackupForNewArch(b)
		if err != nil {
			return errors.Annotate(err, "build tidb backup for new arch")
		}
	} else {
		tidbBackup, err = cb.buildTiDBBackupForGardenerArch(b)
		if err != nil {
			return errors.Annotate(err, "build tidb backup for gardener arch")
		}
	}

	// fix log backup dose not need this param, or it will cause log backup failed
	if b.Spec.Mode != nil && cloudv1.BackupModeLog == *b.Spec.Mode {
		tidbBackup.Spec.BR.Concurrency = nil
	}

	tidbBackup.Annotations = commonAnnotationsForBR()
	if b.Spec.CleanPolicy != nil {
		tidbBackup.Spec.CleanPolicy = ConvertCleanPolicy(*b.Spec.CleanPolicy)
	}

	// 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`.
	b.Status.BucketName = getBucketFromSpec(b)
	b.Status.Phase = cloudv1.BackupPrepare
	if err := cb.cli.Status().Update(ctx, b); err != nil {
		return err
	}

	if len(b.Spec.Env) > 0 {
		tidbBackup.Spec.Env = b.Spec.Env
	}
	if b.Spec.ResourceRequirements != nil {
		tidbBackup.Spec.ResourceRequirements = *b.Spec.ResourceRequirements
	}

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

	// add priority class to backup pod to avoid node-pressure eviction
	// ref: https://kubernetes.io/docs/concepts/scheduling-eviction/node-pressure-eviction/#pod-selection-for-kubelet-eviction
	if err := cb.ensurePriorityClass(ctx); err != nil {
		return err
	}
	tidbBackup.Spec.PriorityClassName = ClusterCriticalPriorityClass

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

func (cb *clusterBackup) buildTiDBBackupForGardenerArch(b *cloudv1.Backup) (*v1alpha1.Backup, error) {
	vars := TidbBackupVars{
		BackupName:      b.ObjectMeta.Name,
		BackupNamespace: cb.k8sStatus.Namespace,
		ClusterName:     utils.ClusterPrefix,
		Region:          cb.network.Spec.Region,
		Bucket:          getBucketFromSpec(b),
		BackupPrefix:    b.ObjectMeta.Name, // default is backup name
		TableFilter:     b.Spec.TableFilter,
	}
	if b.Spec.CommitTS != nil {
		vars.CommitTS = *b.Spec.CommitTS
	}
	if b.Spec.Mode != nil {
		vars.Mode = *b.Spec.Mode
	}
	if b.Spec.Storage != nil {
		uri, _ := url.Parse(*b.Spec.Storage)
		vars.BackupPrefix = strings.Trim(fmt.Sprintf("%s/%s", uri.Path, b.Name), "/")
	}

	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 nil, errors.Annotate(err, "get gitops backup config")
	}
	brConfig, err := BRConfig(brConfigOutput)
	if err != nil {
		return nil, errors.Annotate(err, "convert backup config")
	}

	// 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: // gardener gcp
		providerStatus, err := cloudv1.GardenerProviderSpecForProvider(cb.provider)
		if err != nil {
			return nil, errors.Annotate(err, "get gardener provider spec")
		}
		tidbBackup = assembleGcpBackupCRD(vars, providerStatus.GCP.ProjectID, annotations, labels)
		tidbBackup = mergeBackupConfig(tidbBackup, brConfig, brConfigOutput)
	case consts.ProviderTypeAWS: // gardener aws
		tidbBackup = assembleAWSBackup(vars, annotations, labels)
		tidbBackup = mergeBackupConfig(tidbBackup, brConfig, brConfigOutput)
	default:
		return nil, fmt.Errorf("unknown provider %s", providerType)
	}

	err = cloudv1.MutateShootBackupStatus(b, func(status *cloudv1.ShootBackupStatus) error {
		status.ShootName = cb.shootName
		status.Namespace = cb.k8sStatus.Namespace
		return nil
	})
	if err != nil {
		return nil, errors.Annotate(err, "mutate shoot backup status")
	}

	return tidbBackup, nil
}

func (cb *clusterBackup) buildTiDBBackupForNewArch(b *cloudv1.Backup) (*v1alpha1.Backup, error) {
	return cb.backupProvider.BuildBackup(b, cb.cluster)
}

func (cb *clusterBackup) ensurePriorityClass(ctx context.Context) error {
	priorityClass := &scheduling_v1.PriorityClass{
		ObjectMeta: v1.ObjectMeta{
			Name: ClusterCriticalPriorityClass,
		},
		Value:         ClusterCriticalPriorityValue,
		GlobalDefault: false,
		Description:   "Priority class for br",
	}
	_, err := controllerutil.CreateOrUpdate(ctx, cb.clusterCli, priorityClass, func() error { return nil })
	if err != nil {
		return errors.Annotate(err, "fail to create or update priority class")
	}
	return nil
}

func checkNamespaceExistInDataPlane(ctx context.Context, clusterCli client.Client, namespace string) (bool, error) {
	if clusterCli == nil {
		return false, nil
	}

	// namespace may be deleted when tidb cluster is deleted
	var ns corev1.Namespace
	if err := clusterCli.Get(ctx, types.NamespacedName{Name: namespace}, &ns); err != nil {
		if reconciler.IgnoreNotFound(err) != nil {
			return false, err
		}
		return false, nil
	}
	return true, nil
}

func getBucketFromSpec(b *cloudv1.Backup) string {
	// storage uri is 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 ""
	}
}

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

	// fast fail if dataplane k8s can not connect or namespace doesn't exist caused by tidb cluster deletion
	// or cluster cr is not found
	if cb.isDataPlaneK8sCannotConnect || cb.isDataPlaneNamespaceNotExist || cb.cluster == nil {
		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
	}

	cb.tryToDeleteBRNodeAfterLogTruncated(ctx, b, &tidbBackup)

	return nil
}

func (cb *clusterBackup) tryToDeleteBRNodeAfterLogTruncated(
	ctx context.Context, infraBackup *cloudv1.Backup, tidbBackup *v1alpha1.Backup,
) {
	// if set log stop, it means we may need node to execute log stop command, we can't delete node here
	logStop := infraBackup.InLogBackupStopCommand()
	if cb.checkLogBackupTruncated(infraBackup, tidbBackup) && !logStop && cb.needUseNodeGroup(infraBackup) {
		ngName := backupNodeGroupName(infraBackup, cb.inNewArch(), cb.isGKEProvider())
		if err := cb.backupProvider.DeleteBRNode(ctx, cb.network.DeepCopy(), ngName); err != nil {
			log.FromContext(ctx).Error(
				err, fmt.Sprintf("clean node group for backup error, ng name: %s", ngName))
		}
	}
}

func (cb *clusterBackup) checkLogBackupTruncated(infraBackup *cloudv1.Backup, tidbBackup *v1alpha1.Backup) bool {
	if !infraBackup.InLogBackupTruncateCommand() {
		return false
	}
	truncateStatus, ok := tidbBackup.Status.LogSubCommandStatuses[v1alpha1.LogTruncateCommand]
	if !ok {
		return false
	}

	truncateFinished := truncateStatus.Phase == v1alpha1.BackupComplete || truncateStatus.Phase == v1alpha1.BackupFailed
	return *infraBackup.Spec.LogTruncateUntil == truncateStatus.LogTruncatingUntil && truncateFinished
}

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 tidbBackup.Spec.LogTruncateUntil != "" && cb.needUseNodeGroup(infraBackup) {
			if err := cb.createNodeForBackup(ctx, infraBackup, tidbBackup); err != nil {
				return err
			}
		}
	}
	if infraBackup.Spec.LogStop != nil && tidbBackup.Spec.LogStop != *infraBackup.Spec.LogStop {
		tidbBackup.Spec.LogStop = *infraBackup.Spec.LogStop
		updated = true

		// create node for log backup stop pod if it needs
		if tidbBackup.Spec.LogStop && cb.needUseNodeGroup(infraBackup) {
			if err := cb.createNodeForBackup(ctx, infraBackup, tidbBackup); err != nil {
				return err
			}
		}
	}

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

// create node group for backup see more in https://pingcap.feishu.cn/wiki/wikcnTC7NOM6Gavdhzq7wmg5fMc
func (cb *clusterBackup) needUseNodeGroup(backup *cloudv1.Backup) bool {
	if cb.network == nil {
		return false
	}
	// the default memory request and the value is 100Mi
	// https://github.com/tidbcloud/dataflow-service/blob/461fa309267d25f2d45fc6987dd7f3409db817d1/internal/core_ctx/infra/facade_imp/infra_imp.go#LL45C24-L45C24
	hasMemoryRequest := backup.Spec.ResourceRequirements != nil && backup.Spec.ResourceRequirements.Requests.Memory() != nil
	needMoreMemory := (hasMemoryRequest &&
		backup.Spec.ResourceRequirements.Requests.Memory().Cmp(resource.MustParse("100Mi")) > 0)
	isSnapshotBackup := backup.Spec.Mode != nil && *backup.Spec.Mode == cloudv1.BackupModeSnapshot
	isLogBackupStart := backup.Spec.Mode != nil && *backup.Spec.Mode == cloudv1.BackupModeLog
	return (needMoreMemory && (isSnapshotBackup || isLogBackupStart))
}

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.BackupStopped:
		return cloudv1.BackupStopped
	case v1alpha1.BackupFailed, v1alpha1.BackupInvalid:
		return cloudv1.BackupFailed
	case v1alpha1.BackupRetryTheFailed:
		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
	case cloudv1.BackupModeVolumeSnapshot:
		return v1alpha1.BackupModeVolumeSnapshot
	default:
		return ""
	}
}

func ConvertCleanPolicy(cleanPolicy cloudv1.CleanPolicy) v1alpha1.CleanPolicyType {
	switch cleanPolicy {
	case cloudv1.CleanPolicyDelete:
		return v1alpha1.CleanPolicyTypeDelete
	case cloudv1.CleanPolicyRetain:
		return v1alpha1.CleanPolicyTypeRetain
	case cloudv1.CleanPolicyOnFailure:
		return v1alpha1.CleanPolicyTypeOnFailure
	default:
		return v1alpha1.CleanPolicyTypeDelete
	}
}

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)

	if cb.needUseNodeGroup(b) {
		ngName := backupNodeGroupName(b, cb.inNewArch(), cb.isGKEProvider())
		log.FromContext(ctx).Info(fmt.Sprintf("clean node group for backup ng name: %s", ngName))
		if err := cb.backupProvider.DeleteBRNode(ctx, cb.network.DeepCopy(), ngName); err != nil {
			log.FromContext(ctx).Error(
				err, fmt.Sprintf("clean node group for backup error, ng name: %s", ngName))
			return reconciler.ExternalDeletion{}, err
		}
	}

	// 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])
	}

	// we treat as already deleted when dataplane k8s can not connect or namespace doesn't exist caused by tidb cluster deletion
	// or cluster is nil
	if cb.isDataPlaneK8sCannotConnect || cb.isDataPlaneNamespaceNotExist || cb.cluster == nil {
		return reconciler.ExternalDeletion{ResourceDeleted: true}, nil
	}

	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
	}

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

func (cb *clusterBackup) createNodeForBackup(ctx context.Context, infraBackup *cloudv1.Backup, tidbBackup *v1alpha1.Backup) error {
	logger := log.FromContext(ctx).WithName("createNodeForBackup")
	region, err := cb.provider.FindRegion(cb.network.Spec.Region)
	if err != nil {
		logger.Error(err, "find region failed")
		return err
	}
	ngSpec, err := buildBackupNodeGroupSpecWithTaints(backupNodeGroupName(infraBackup, cb.inNewArch(), cb.isGKEProvider()), infraBackup, cb.provider, &region)
	if err != nil {
		logger.Error(err, "build backup node group spec failed")
		return err
	}
	logger.Info(fmt.Sprintf("find ngSpec success the machine: %s", ngSpec.MachineSpec.MachineType))
	if err := cb.backupProvider.CreateBRNode(ctx, cb.network.DeepCopy(), ngSpec); err != nil {
		logger.Error(err, fmt.Sprintf("create backup node failed, node group spec name %s", ngSpec.Name))
		return err
	}
	tidbBackup.Spec.Affinity = utils.NodeAffinityWithUK(backupNodeAffinityKey(infraBackup), backupNodeAffinityValue(infraBackup))
	tidbBackup.Spec.Tolerations = utils.TolerationWithUK(cloudv1.ComponentBackup, infraBackup.Name)
	logger.Info(fmt.Sprintf("create node for backup with taints success taints %s", ngSpec.Taints))
	return nil
}

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

	PodSecurityContext *corev1.PodSecurityContext
}

// assembleFreeSharedTierAWSBackup assembles the backup command both for ProviderTypeAWSFreeTier and ProviderTypeAWSSharedTier.
// nolint:unused
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)
	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,
			TableFilter: vars.TableFilter,
			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.Bool(false),
				Concurrency:      &concurrency,
				StatusAddr:       BRStatusAddr,
			},
			UseKMS:             false,
			PodSecurityContext: vars.PodSecurityContext,
		},
	}
}

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,
			TableFilter: vars.TableFilter,
			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.Bool(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(), MaxConcurrentReconciles: 10}).
		WithEventFilter(utils.GetMigrationAnnotationSkipPredicator()).
		Complete(r)
}