package server

import (
	"context"
	"encoding/json"
	"fmt"
	"sort"

	"github.com/google/go-cmp/cmp"
	"github.com/google/go-cmp/cmp/cmpopts"
	"github.com/samber/lo"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/equality"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/yaml"

	internalutil "github.com/tidbcloud/cluster-controller/server/internal/util"
	config2 "github.com/tidbcloud/cluster-controller/server/pkg/config"
	"github.com/tidbcloud/cluster-controller/server/server/cvt"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	apiv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/cluster-server/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/gitops"
	gitopsconsts "github.com/tidbcloud/infra-provider/pkg/gitops-sdk/consts"
)

const logNil = "nil"

type brServer struct {
	kubeClient           client.Client
	clusterReader        internalutil.ClusterReader
	instanceTypeProvider gitops.InstanceTypeProvider
}

func NewBRServer(cli client.Client, instanceTypeProvider gitops.InstanceTypeProvider) apiv1.BRServiceServer {
	s := &brServer{
		kubeClient:           cli,
		clusterReader:        &internalutil.LabelSelectorReader{},
		instanceTypeProvider: instanceTypeProvider,
	}
	return s
}

func (s *brServer) CreateSnapshotBackup(ctx context.Context, req *apiv1.CreateSnapshotBackupRequest) (*apiv1.SnapshotBackup, error) {
	clusterId := req.Backup.ClusterId
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", clusterId))
	cluster, err := s.clusterReader.Get(ctx, s.kubeClient, clusterId)
	if err != nil {
		logger.Error("Failed to get cluster", zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", clusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", clusterId, err)
	}

	backupLabels := buildClusterLabelsForCreating(cluster.GetProviderType(), cluster.GetRegion(), cluster.GetTenantID(), cluster.GetProjectID(), clusterId)
	backupLabels[consts.BackupModeLabelKey] = string(cloudv1.BackupModeSnapshot)

	// Resolve instance type and resource requirements
	instanceType, resources, err := s.resolveInstanceTypeAndResources(
		ctx,
		req.Backup.InstanceType,
		req.Backup.ResourceRequirements,
	)
	if err != nil {
		logger.Error("Failed to resolve instance type and resources", zap.Error(err))
		return nil, status.Errorf(codes.InvalidArgument, "Failed to resolve instance type and resources: %s", err)
	}

	name := getBackupName(req.Backup.Id)
	cfg := config2.GetConfig()
	backup := &cloudv1.Backup{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: cluster.Namespace,
			Annotations: map[string]string{
				gitopsconsts.OverrideHelperImageKey: cfg.OverrideHelperImage,
				gitopsconsts.OverrideRegistryKey:    cfg.OverrideRegistry,
				gitopsconsts.OverrideRepoPrefixKey:  cfg.OverrideRepoPrefix,
			},
			Labels: backupLabels,
		},
		Spec: cloudv1.BackupSpec{
			ClusterName:          cluster.Name,
			Mode:                 ptr.To(cloudv1.BackupModeSnapshot),
			TableFilter:          req.Backup.TableFilters,
			CommitTS:             &req.Backup.CommitTs,
			InstanceType:         ptr.To(instanceType),
			ResourceRequirements: resources,
			Storage:              ptr.To(req.Backup.Bucket),
			Configuration: &cloudv1.DPConfiguration{
				BaseImage: req.Backup.ImageRepo,
				Tag:       req.Backup.ImageTag,
				Config:    req.Backup.Configuration,
			},
			CleanPolicy: ptr.To(cloudv1.CleanPolicyRetain), // default to retain
			Env:         convertEnvMapToEnvVars(req.Backup.Env),
		},
	}

	if err := s.kubeClient.Create(ctx, backup); err != nil {
		logger.Error("failed to create snapshot backup", zap.Error(err))
		if apierrors.IsAlreadyExists(err) {
			return nil, status.Errorf(codes.AlreadyExists, "snapshot backup %s already exists", req.Backup.Id)
		}
		return nil, err
	}

	// newly created backup should be in running status
	return &apiv1.SnapshotBackup{Id: req.Backup.Id, Status: apiv1.SnapshotBackup_RUNNING}, nil
}

func (s *brServer) GetSnapshotBackup(ctx context.Context, req *apiv1.GetSnapshotBackupRequest) (*apiv1.SnapshotBackup, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.ClusterId))
	cluster, err := s.clusterReader.Get(ctx, s.kubeClient, req.ClusterId)
	if err != nil {
		logger.Error("Failed to get cluster", zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", req.ClusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", req.ClusterId, err)
	}
	name := getBackupName(req.BackupId)
	backup := &cloudv1.Backup{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: cluster.Namespace,
		},
	}
	if err := s.kubeClient.Get(ctx, client.ObjectKeyFromObject(backup), backup); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Snapshot backup %s not found", req.BackupId)
		}
		logger.Error("failed to get backup", zap.Error(err))
		return nil, err
	}
	return cvt.SnapshotBackup2PB(req.BackupId, backup), nil
}

func (s *brServer) DeleteSnapshotBackup(ctx context.Context, req *apiv1.DeleteSnapshotBackupRequest) (*apiv1.SnapshotBackup, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.ClusterId))
	cluster, err := s.clusterReader.Get(ctx, s.kubeClient, req.ClusterId)
	if err != nil {
		logger.Error("Failed to get cluster", zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", req.ClusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", req.ClusterId, err)
	}
	name := getBackupName(req.BackupId)
	backup := &cloudv1.Backup{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: cluster.Namespace,
		},
	}
	if err := s.kubeClient.Delete(ctx, backup); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Backup %s not found", name)
		}
		logger.Error("failed to delete snapshot backup", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "failed to delete snapshot backup %s: %s", name, err)
	}
	return &apiv1.SnapshotBackup{Id: req.BackupId}, nil
}

func (s *brServer) CreateLogBackup(ctx context.Context, req *apiv1.CreateLogBackupRequest) (*apiv1.LogBackup, error) {
	clusterId := req.Backup.ClusterId
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", clusterId))
	cluster, err := s.clusterReader.Get(ctx, s.kubeClient, clusterId)
	if err != nil {
		logger.Error("Failed to get cluster", zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", clusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", clusterId, err)
	}

	// Resolve instance type and resource requirements
	instanceType, resources, err := s.resolveInstanceTypeAndResources(
		ctx,
		req.Backup.InstanceType,
		req.Backup.ResourceRequirements,
	)
	if err != nil {
		logger.Error("Failed to resolve instance type and resources", zap.Error(err))
		return nil, status.Errorf(codes.InvalidArgument, "Failed to resolve instance type and resources: %s", err)
	}

	name := getBackupName(req.Backup.Id)
	cfg := config2.GetConfig()
	backupLabels := buildClusterLabelsForCreating(cluster.GetProviderType(), cluster.GetRegion(), cluster.GetTenantID(), cluster.GetProjectID(), clusterId)
	backupLabels[consts.BackupModeLabelKey] = string(cloudv1.BackupModeLog)
	backup := &cloudv1.Backup{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: cluster.Namespace,
			Annotations: map[string]string{
				gitopsconsts.OverrideHelperImageKey: cfg.OverrideHelperImage,
				gitopsconsts.OverrideRegistryKey:    cfg.OverrideRegistry,
				gitopsconsts.OverrideRepoPrefixKey:  cfg.OverrideRepoPrefix,
			},
			Labels: backupLabels,
		},
		Spec: cloudv1.BackupSpec{
			CleanPolicy:          ptr.To(cloudv1.CleanPolicyRetain), // default to retain
			ClusterName:          cluster.Name,
			Mode:                 ptr.To(cloudv1.BackupModeLog),
			CommitTS:             ptr.To(req.Backup.StartTs),
			InstanceType:         ptr.To(instanceType),
			ResourceRequirements: resources,
			Storage:              ptr.To(req.Backup.Bucket),
			Configuration: &cloudv1.DPConfiguration{
				BaseImage: req.Backup.ImageRepo,
				Tag:       req.Backup.ImageTag,
				Config:    req.Backup.Configuration,
			},
			Env: convertEnvMapToEnvVars(req.Backup.Env),
		},
	}

	if err := s.kubeClient.Create(ctx, backup); err != nil {
		logger.Error("failed to create log backup", zap.Error(err))
		if apierrors.IsAlreadyExists(err) {
			return nil, status.Errorf(codes.AlreadyExists, "log backup %s already exists", req.Backup.Id)
		}
		return nil, err
	}

	return &apiv1.LogBackup{Id: req.Backup.Id, LogStatus: apiv1.LogBackup_RUNNING}, nil
}

func (s *brServer) TruncateLogBackup(ctx context.Context, req *apiv1.TruncateLogBackupRequest) (*apiv1.LogBackup, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.ClusterId))
	cluster, err := s.clusterReader.Get(ctx, s.kubeClient, req.ClusterId)
	if err != nil {
		logger.Error("Failed to get cluster", zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", req.ClusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", req.ClusterId, err)
	}

	name := getBackupName(req.BackupId)
	backup := &cloudv1.Backup{}
	if err := s.kubeClient.Get(ctx, client.ObjectKey{Namespace: cluster.Namespace, Name: name}, backup); err != nil {
		logger.Error("failed to get backup", zap.Error(err))
		return nil, err
	}

	backup.Spec.LogTruncateUntil = &req.Until
	stop := false
	backup.Spec.LogStop = &stop
	if err := s.kubeClient.Update(ctx, backup); err != nil {
		logger.Error("failed to update backup for truncation", zap.Error(err))
		return nil, err
	}

	return &apiv1.LogBackup{Id: req.BackupId}, nil
}

func (s *brServer) StopLogBackup(ctx context.Context, req *apiv1.StopLogBackupRequest) (*apiv1.LogBackup, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.ClusterId))
	cluster, err := s.clusterReader.Get(ctx, s.kubeClient, req.ClusterId)
	if err != nil {
		logger.Error("Failed to get cluster", zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", req.ClusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", req.ClusterId, err)
	}

	name := getBackupName(req.BackupId)
	backup := &cloudv1.Backup{}
	if err := s.kubeClient.Get(ctx, client.ObjectKey{Namespace: cluster.Namespace, Name: name}, backup); err != nil {
		logger.Error("failed to get backup", zap.Error(err))
		return nil, err
	}

	backup.Spec.LogStop = ptr.To(true)
	if err := s.kubeClient.Update(ctx, backup); err != nil {
		logger.Error("failed to update backup for stopping", zap.Error(err))
		return nil, err
	}

	return &apiv1.LogBackup{Id: req.BackupId, LogStatus: apiv1.LogBackup_STOPPING}, nil
}

func (s *brServer) GetLogBackup(ctx context.Context, req *apiv1.GetLogBackupRequest) (*apiv1.LogBackup, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.ClusterId))
	cluster, err := s.clusterReader.Get(ctx, s.kubeClient, req.ClusterId)
	if err != nil {
		logger.Error("Failed to get cluster", zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", req.ClusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", req.ClusterId, err)
	}

	name := getBackupName(req.BackupId)
	backup := &cloudv1.Backup{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: cluster.Namespace,
		},
	}
	if err := s.kubeClient.Get(ctx, client.ObjectKeyFromObject(backup), backup); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Log backup %s not found", name)
		}
		logger.Error("failed to get backup", zap.Error(err))
		return nil, err
	}

	return cvt.LogBackup2PB(req.BackupId, backup), nil
}

func (s *brServer) ListAllLogBackups(ctx context.Context, req *apiv1.ListAllLogBackupsRequest) (*apiv1.ListAllLogBackupsResponse, error) {
	var backupList cloudv1.BackupList
	if err := s.kubeClient.List(ctx, &backupList, client.MatchingLabels{consts.BackupModeLabelKey: string(cloudv1.BackupModeLog)}); err != nil {
		return nil, status.Errorf(codes.Internal, "list log backup CRs failed")
	}

	backups := make([]*apiv1.LogBackup, 0, len(backupList.Items))
	for _, backup := range backupList.Items {
		backups = append(backups, cvt.LogBackup2PB(backup.Name, &backup))
	}

	return &apiv1.ListAllLogBackupsResponse{LogBackups: backups}, nil
}

func (s *brServer) DeleteLogBackup(ctx context.Context, req *apiv1.DeleteLogBackupRequest) (*apiv1.LogBackup, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.ClusterId))
	name := getBackupName(req.BackupId)
	var backup *cloudv1.Backup
	cluster, err := s.clusterReader.Get(ctx, s.kubeClient, req.ClusterId)
	if err != nil {
		if apierrors.IsNotFound(err) {
			// The parent Cluster can be removed before dataflow-service deletes a
			// log backup. The Backup CR is labelled with the cluster ID, so it is
			// still safe to find and delete that exact log backup without the
			// parent Cluster.
			logger.Info("parent cluster not found; deleting matching log backup by labels", zap.String("backup", name))
			backup, err = s.getLogBackupForDeletedCluster(ctx, req.ClusterId, name)
			if err != nil {
				return nil, err
			}
		} else {
			logger.Error("Failed to get cluster", zap.Error(err))
			return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", req.ClusterId, err)
		}
	} else {
		backup = &cloudv1.Backup{
			ObjectMeta: metav1.ObjectMeta{
				Name:      name,
				Namespace: cluster.Namespace,
			},
		}
	}

	if err := s.kubeClient.Delete(ctx, backup); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Backup %s not found", name)
		}
		logger.Error("failed to delete log backup", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "failed to delete log backup %s: %s", name, err)
	}

	return &apiv1.LogBackup{Id: req.BackupId}, nil
}

func (s *brServer) getLogBackupForDeletedCluster(ctx context.Context, clusterID, backupName string) (*cloudv1.Backup, error) {
	var backups cloudv1.BackupList
	if err := s.kubeClient.List(ctx, &backups, client.MatchingLabels{
		cloudv1.ClusterTagKey:     clusterID,
		consts.BackupModeLabelKey: string(cloudv1.BackupModeLog),
	}); err != nil {
		return nil, status.Errorf(codes.Internal, "list log backups for deleted cluster %s: %s", clusterID, err)
	}

	for i := range backups.Items {
		if backups.Items[i].Name == backupName {
			return &backups.Items[i], nil
		}
	}

	return nil, status.Errorf(codes.NotFound, "Log backup %s not found", backupName)
}

func (s *brServer) CreateRestore(ctx context.Context, req *apiv1.CreateRestoreRequest) (*apiv1.Restore, error) {
	clusterId := req.Restore.ClusterId
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", clusterId))
	logger.Info("CreateRestore request received",
		zap.String("id", req.Restore.Id),
		zap.String("cluster", req.Restore.ClusterId),
		zap.String("compactbackupName", req.Restore.CompactBackupName),
	)

	cluster, err := s.clusterReader.Get(ctx, s.kubeClient, clusterId)
	if err != nil {
		logger.Error("Failed to get cluster", zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", clusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", clusterId, err)
	}

	// Resolve instance type and resource requirements
	instanceType, resources, err := s.resolveInstanceTypeAndResources(
		ctx,
		req.Restore.InstanceType,
		req.Restore.ResourceRequirements,
	)
	if err != nil {
		logger.Error("Failed to resolve instance type and resources", zap.Error(err))
		return nil, status.Errorf(codes.InvalidArgument, "Failed to resolve instance type and resources: %s", err)
	}

	cfg := config2.GetConfig()
	name := getRestoreName(req.Restore.Id)
	restore := &cloudv1.Restore{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: cluster.Namespace,
			Annotations: map[string]string{
				gitopsconsts.OverrideHelperImageKey: cfg.OverrideHelperImage,
				gitopsconsts.OverrideRegistryKey:    cfg.OverrideRegistry,
				gitopsconsts.OverrideRepoPrefixKey:  cfg.OverrideRepoPrefix,
			},
			Labels: buildBackupLabelsForCreating(cluster.GetProviderType(), cluster.GetRegion(), cluster.GetTenantID(), cluster.GetProjectID(), clusterId),
		},
		Spec: cloudv1.RestoreSpec{
			ClusterName: cluster.Name,
			From: cloudv1.From{
				Region:                ptr.To(bucketRegion(req.Restore, cluster.GetRegion())),
				SnapshotBackupStorage: ptr.To(req.Restore.SnapshotBucket),
			},
			InstanceType:         ptr.To(instanceType),
			ResourceRequirements: resources,
			Configuration: &cloudv1.DPConfiguration{
				BaseImage: req.Restore.ImageRepo,
				Tag:       req.Restore.ImageTag,
				Config:    req.Restore.Configuration,
			},
			Env: convertEnvMapToEnvVars(req.Restore.Env),
		},
	}

	if req.Restore.PitrTs != "" {
		restore.Spec.From.PiTRRestoredTS = ptr.To(req.Restore.PitrTs)
		restore.Spec.From.LogBackupStorage = ptr.To(req.Restore.LogBucket)
	}

	if name := req.Restore.GetCompactBackupName(); name != "" {
		if restore.Spec.From.PiTRRestoredTS == nil {
			return nil, status.Errorf(codes.InvalidArgument, "compact_backup_name requires a PiTR restore (pitr_ts must be set)")
		}
		restore.Spec.CompactBackupName = name
	}

	if err := s.kubeClient.Create(ctx, restore); err != nil {
		logger.Error("failed to create restore", zap.Error(err))
		if apierrors.IsAlreadyExists(err) {
			return nil, status.Errorf(codes.AlreadyExists, "restore %s already exists", req.Restore.Id)
		}
		return nil, err
	}

	// newly created restore job should be in running status
	return &apiv1.Restore{Id: req.Restore.Id, Status: apiv1.Restore_RUNNING}, nil
}

func (s *brServer) GetRestore(ctx context.Context, req *apiv1.GetRestoreRequest) (*apiv1.Restore, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.ClusterId))
	cluster, err := s.clusterReader.Get(ctx, s.kubeClient, req.ClusterId)
	if err != nil {
		logger.Error("Failed to get cluster", zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", req.ClusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", req.ClusterId, err)
	}
	name := getRestoreName(req.RestoreId)
	restore := &cloudv1.Restore{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: cluster.Namespace,
		},
	}
	if err := s.kubeClient.Get(ctx, client.ObjectKeyFromObject(restore), restore); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Restore %s not found", req.RestoreId)
		}
		logger.Error("failed to get restore", zap.Error(err))
		return nil, err
	}
	return cvt.Restore2PB(req.RestoreId, restore), nil
}

func (s *brServer) DeleteRestore(ctx context.Context, req *apiv1.DeleteRestoreRequest) (*apiv1.Restore, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.ClusterId))
	cluster, err := s.clusterReader.Get(ctx, s.kubeClient, req.ClusterId)
	if err != nil {
		logger.Error("Failed to get cluster", zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", req.ClusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", req.ClusterId, err)
	}
	name := getRestoreName(req.RestoreId)
	restore := &cloudv1.Restore{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: cluster.Namespace,
		},
	}
	if err := s.kubeClient.Delete(ctx, restore); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Restore %s not found", name)
		}
		logger.Error("failed to delete restore", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "failed to delete restore %s: %s", name, err)
	}
	return &apiv1.Restore{Id: req.RestoreId}, nil
}

// We use backup/restore name to generate infra resource name, such as providerconfig, nodepool
// If backup and restore have the same name, it would affect infra resource management
// In dataflow-service Backup and Restore are different tables, they can have same ID
// However, dataflow-service generates id based on the id of the entity, so no conflict happens
// This method is reserved here for clarification
func getBackupName(id string) string {
	return id
}

func getRestoreName(id string) string {
	return id
}

// bucketRegion returns the bucket region from the restore request, falling back
// to the cluster region when bucket_region is not explicitly specified.
func bucketRegion(restore *apiv1.Restore, clusterRegion string) string {
	region, _ := lo.Coalesce(restore.GetBucketRegion(), clusterRegion)
	return region
}

func getCompactBackupName(id string) string {
	return id
}

// convertProtoResourceRequirements converts proto ResourceRequirements to k8s ResourceRequirements
func convertProtoResourceRequirements(protoRR *apiv1.ResourceRequirements) *corev1.ResourceRequirements {
	if protoRR == nil {
		return nil
	}

	k8sRR := &corev1.ResourceRequirements{}

	// Convert limits
	if len(protoRR.Limits) > 0 {
		k8sRR.Limits = make(corev1.ResourceList)
		for key, value := range protoRR.Limits {
			k8sRR.Limits[corev1.ResourceName(key)] = resource.MustParse(value)
		}
	}

	// Convert requests
	if len(protoRR.Requests) > 0 {
		k8sRR.Requests = make(corev1.ResourceList)
		for key, value := range protoRR.Requests {
			k8sRR.Requests[corev1.ResourceName(key)] = resource.MustParse(value)
		}
	}

	return k8sRR
}

func convertEnvMapToEnvVars(envMap map[string]string) []corev1.EnvVar {
	keys := lo.Keys(envMap)
	sort.Strings(keys) // Sort keys for consistent ordering
	var envVars []corev1.EnvVar
	for _, key := range keys {
		envVars = append(envVars, corev1.EnvVar{
			Name:  key,
			Value: envMap[key],
		})
	}

	return envVars
}

func (s *brServer) CreateCompactBackup(ctx context.Context, req *apiv1.CreateCompactBackupRequest) (*apiv1.CompactBackup, error) {
	clusterId := req.ClusterId
	logger := tilog.WithContextL(ctx).With(
		zap.String("cluster", clusterId),
		zap.String("compact_backup_id", req.CompactBackup.Id))

	logger.Info("CreateCompactBackup request received",
		zap.String("compact_backup_id", req.CompactBackup.Id),
		zap.Int32("shardCount", lo.FromPtr(req.CompactBackup.ShardCount)),
		zap.String("start_ts", func() string {
			if req.CompactBackup.StartTs != nil {
				return *req.CompactBackup.StartTs
			}
			return logNil
		}()),
		zap.String("end_ts", func() string {
			if req.CompactBackup.EndTs != nil {
				return *req.CompactBackup.EndTs
			}
			return logNil
		}()))

	// Validate required parameters
	if req.CompactBackup.Storage == nil || *req.CompactBackup.Storage == "" {
		logger.Warn("Storage parameter is missing")
	}
	if req.CompactBackup.Configuration == nil {
		logger.Warn("Configuration is missing")
	}

	cluster, err := s.clusterReader.Get(ctx, s.kubeClient, clusterId)
	if err != nil {
		logger.Error("Failed to get cluster", zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", clusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", clusterId, err)
	}

	// Use the ID from request - consistent with other backup types
	compactBackupId := req.CompactBackup.Id
	name := getCompactBackupName(compactBackupId)
	cfg := config2.GetConfig()

	// Build labels using existing helper function
	compactBackupLabels := buildBackupLabelsForCreating(cluster.GetProviderType(), cluster.GetRegion(), cluster.GetTenantID(), cluster.GetProjectID(), clusterId)

	// Resolve instance type and resource requirements
	instanceType, resources, err := s.resolveInstanceTypeAndResources(
		ctx,
		lo.FromPtr(req.CompactBackup.InstanceType),
		req.CompactBackup.ResourceRequirements,
	)
	if err != nil {
		logger.Error("Failed to resolve instance type and resources", zap.Error(err))
		return nil, status.Errorf(codes.InvalidArgument, "Failed to resolve instance type and resources: %s", err)
	}
	compactBackup := &cloudv1.CompactBackup{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: cluster.Namespace,
			Annotations: map[string]string{
				gitopsconsts.OverrideHelperImageKey: cfg.OverrideHelperImage,
				gitopsconsts.OverrideRegistryKey:    cfg.OverrideRegistry,
				gitopsconsts.OverrideRepoPrefixKey:  cfg.OverrideRepoPrefix,
			},
			Labels: compactBackupLabels,
		},
		Spec: cloudv1.CompactBackupSpec{
			ClusterName:          cluster.Name,
			CleanPolicy:          ptr.To(cloudv1.CleanPolicyRetain), // default to retain
			InstanceType:         ptr.To(instanceType),
			ResourceRequirements: resources,
		},
	}

	// Set fields from the request
	if req.CompactBackup.StartTs != nil {
		compactBackup.Spec.StartTs = req.CompactBackup.StartTs
	}
	if req.CompactBackup.EndTs != nil {
		compactBackup.Spec.EndTs = req.CompactBackup.EndTs
	}
	if req.CompactBackup.Concurrency != nil {
		compactBackup.Spec.Concurrency = req.CompactBackup.Concurrency
	}
	if req.CompactBackup.CleanPolicy != nil {
		compactBackup.Spec.CleanPolicy = (*cloudv1.CleanPolicy)(req.CompactBackup.CleanPolicy)
	}
	if req.CompactBackup.ShardCount != nil && *req.CompactBackup.ShardCount > 0 {
		compactBackup.Spec.ShardCount = req.CompactBackup.ShardCount
	}

	// Set storage field - this is critical for the backup process
	if req.CompactBackup.Storage != nil {
		logger.Info("[STORAGE_TRACE] Setting storage field in CompactBackup CR spec",
			zap.String("storage_value", *req.CompactBackup.Storage))
		compactBackup.Spec.Storage = req.CompactBackup.Storage
	} else {
		logger.Error("[STORAGE_TRACE] Storage field is missing in request - this will cause backup to fail")
	}

	// Set configuration with TiKV image support
	if req.CompactBackup.Configuration != nil {
		config := &cloudv1.DPConfiguration{}
		if req.CompactBackup.Configuration.BaseImage != nil {
			config.BaseImage = *req.CompactBackup.Configuration.BaseImage
		}
		if req.CompactBackup.Configuration.Tag != nil {
			config.Tag = *req.CompactBackup.Configuration.Tag
		}
		if req.CompactBackup.Configuration.Config != nil {
			config.Config = *req.CompactBackup.Configuration.Config
		}
		if req.CompactBackup.Configuration.TikvImage != nil {
			config.TiKVImage = *req.CompactBackup.Configuration.TikvImage // 新增
		}
		if req.CompactBackup.Configuration.TikvVersion != nil {
			config.TiKVVersion = *req.CompactBackup.Configuration.TikvVersion // 新增
		}
		compactBackup.Spec.Configuration = config
	}

	// Log the complete spec before creating
	logger.Info("DEBUG: About to create CompactBackup with complete spec",
		zap.Any("resourceRequirements", compactBackup.Spec.ResourceRequirements),
		zap.String("clusterName", compactBackup.Spec.ClusterName),
		zap.Any("concurrency", compactBackup.Spec.Concurrency),
		zap.Any("instanceType", compactBackup.Spec.InstanceType))

	// Create CompactBackup CR
	if err := s.kubeClient.Create(ctx, compactBackup); err != nil {
		logger.Error("Failed to create CompactBackup CR", zap.Error(err))
		if apierrors.IsAlreadyExists(err) {
			return nil, status.Errorf(codes.AlreadyExists, "compact backup %s already exists", compactBackupId)
		}
		return nil, err
	}

	logger.Info("CompactBackup CR created successfully", zap.String("cr_name", compactBackup.Name))

	// Immediately verify what was actually created
	createdCR := &cloudv1.CompactBackup{}
	if err := s.kubeClient.Get(ctx, client.ObjectKeyFromObject(compactBackup), createdCR); err == nil {
		logger.Info("DEBUG: Verified created CompactBackup CR",
			zap.Any("actual_resourceRequirements", createdCR.Spec.ResourceRequirements),
			zap.String("actual_clusterName", createdCR.Spec.ClusterName))
	} else {
		logger.Error("DEBUG: Failed to verify created CompactBackup CR", zap.Error(err))
	}

	// Return the full CompactBackup object with all fields including resource_requirements
	return cvt.CompactBackup2PB(compactBackupId, compactBackup), nil
}

func (s *brServer) GetCompactBackup(ctx context.Context, req *apiv1.GetCompactBackupRequest) (*apiv1.CompactBackup, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.ClusterId))
	cluster, err := s.clusterReader.Get(ctx, s.kubeClient, req.ClusterId)
	if err != nil {
		logger.Error("Failed to get cluster", zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", req.ClusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", req.ClusterId, err)
	}

	name := getCompactBackupName(req.CompactBackupId)
	compactBackup := &cloudv1.CompactBackup{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: cluster.Namespace,
		},
	}

	if err := s.kubeClient.Get(ctx, client.ObjectKeyFromObject(compactBackup), compactBackup); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "CompactBackup %s not found", req.CompactBackupId)
		}
		logger.Error("failed to get compact backup", zap.Error(err))
		return nil, err
	}

	return cvt.CompactBackup2PB(req.CompactBackupId, compactBackup), nil
}

func (s *brServer) DeleteCompactBackup(ctx context.Context, req *apiv1.DeleteCompactBackupRequest) (*apiv1.CompactBackup, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.ClusterId))
	cluster, err := s.clusterReader.Get(ctx, s.kubeClient, req.ClusterId)
	if err != nil {
		logger.Error("Failed to get cluster", zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", req.ClusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", req.ClusterId, err)
	}

	name := getCompactBackupName(req.CompactBackupId)
	compactBackup := &cloudv1.CompactBackup{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: cluster.Namespace,
		},
	}

	if err := s.kubeClient.Delete(ctx, compactBackup); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "CompactBackup %s not found", name)
		}
		logger.Error("failed to delete compact backup", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "failed to delete compact backup %s: %s", name, err)
	}

	return &apiv1.CompactBackup{Id: req.CompactBackupId, ClusterId: req.ClusterId}, nil
}

// resolveInstanceTypeAndResources resolves instance type and resource requirements based on the request.
func (s *brServer) resolveInstanceTypeAndResources(
	ctx context.Context,
	instanceType string,
	resourceReqs *apiv1.ResourceRequirements,
) (string, *corev1.ResourceRequirements, error) {
	logger := tilog.WithContextL(ctx)

	hc := gitops.NewHardcode()
	// If instance_type is not provided, try to get instance_type by resource_requirements
	if instanceType == "" {
		ins, err := getMachineSpecByResourceRequirements(s.instanceTypeProvider, convertProtoResourceRequirements(resourceReqs))
		if err != nil {
			return "", nil, fmt.Errorf("failed to get instance type by resource requirements: %w", err)
		}
		instanceType = ins.Name
		logger.Info("Resolved instance type from resource requirements", zap.String("instance_type", instanceType))
	}
	requirements, err := hc.GetCPUMemResourceRequirements(s.instanceTypeProvider, instanceType)
	return instanceType, requirements, err

}

func getMachineSpecByResourceRequirements(
	provider gitops.InstanceTypeProvider, resourceRequirements *corev1.ResourceRequirements) (instanceType cloudv1.InstanceType, _ error) {
	// empty resourceRequirements means no requirement, so we can return the smallest instance type
	if resourceRequirements == nil {
		resourceRequirements = &corev1.ResourceRequirements{}
	}

	cpuLimit := resourceRequirements.Limits.Cpu()
	cpuRequest := resourceRequirements.Requests.Cpu()
	memoryLimit := resourceRequirements.Limits.Memory()
	memoryRequest := resourceRequirements.Requests.Memory()

	instanceTypes, err := provider.GetInstanceTypes()
	if err != nil {
		return cloudv1.InstanceType{}, err
	}
	// find the instance that meets the minimum requirements of cpu and memory in provider.Spec.InstanceTypes
	// and matches the required architecture
	for _, ins := range sortedInstanceTypes(instanceTypes) {
		if ins.CPU.Cmp(cpuRequest.DeepCopy()) > 0 && ins.CPU.Cmp(cpuLimit.DeepCopy()) >= 0 &&
			ins.Memory.Cmp(memoryRequest.DeepCopy()) > 0 && ins.Memory.Cmp(memoryLimit.DeepCopy()) >= 0 &&
			ins.Arch == consts.AMD64ArchName {
			return ins, nil
		}
	}
	return cloudv1.InstanceType{}, fmt.Errorf("no matching instance type found")
}

func sortedInstanceTypes(in []cloudv1.InstanceType) []cloudv1.InstanceType {
	sort.Slice(in, func(i, j int) bool {
		// when memory is equal, sort by cpu
		if in[i].Memory.Cmp(in[j].Memory) == 0 {
			return in[i].CPU.Cmp(in[j].CPU) < 0
		}
		// sort instance types by memory
		return in[i].Memory.Cmp(in[j].Memory) < 0
	})
	return in
}

// RawCreateBackupCR creates a raw backup CR in Kubernetes without the API translation layer.
// To ensure the created CR is as expected, it creates a paused CR first, retrieves it to compare
// the spec, and then unpauses it if the spec matches. Otherwise, it deletes the CR and returns an error.
func (s *brServer) RawCreateBackupCR(ctx context.Context, req *apiv1.RawCreateBackupCRRequest) (*apiv1.RawCreateBackupCRResponse, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.ClusterId))

	cluster, err := s.clusterReader.Get(ctx, s.kubeClient, req.ClusterId)
	if err != nil {
		logger.Error("Failed to get cluster", zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", req.ClusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", req.ClusterId, err)
	}

	// Unmarshal the raw CR content
	var backup cloudv1.Backup
	if err := yaml.Unmarshal([]byte(req.Content), &backup); err != nil {
		logger.Error("Failed to unmarshal backup CR content", zap.Error(err))
		return nil, status.Errorf(codes.InvalidArgument, "Failed to unmarshal backup CR content: %s", err)
	}

	// Ensure the backup is created in the cluster's namespace
	// Clear fields
	backup.CreationTimestamp = metav1.Time{}
	backup.Generation = 0
	backup.ResourceVersion = ""
	backup.UID = ""
	backup.Namespace = cluster.Namespace
	if backup.Labels == nil {
		backup.Labels = make(map[string]string)
	}
	if ptr.Deref(backup.Spec.InstanceType, "") == "" {
		ins, err := getMachineSpecByResourceRequirements(s.instanceTypeProvider, backup.Spec.ResourceRequirements)
		if err != nil {
			logger.Error("Failed to get instance type by resource requirements", zap.Error(err))
			return nil, status.Errorf(codes.Internal, "failed to get instance type by resource requirements: %s", err)
		}
		backup.Spec.InstanceType = ptr.To(ins.Name)
		logger.Info("Resolved instance type from resource requirements for raw backup CR", zap.String("instance_type", ins.Name))
	}
	backup.Labels[cloudv1.ReconcilerLabelKey] = "br-3-layer-migration" // add reconciler label for preventing reconciliation
	backup.Spec.Paused = ptr.To(true)                                  // pause the backup for later verification

	// Make a deep copy of the original backup spec for later comparison
	original := backup.DeepCopy()

	// Create the paused backup CR
	if err := s.kubeClient.Create(ctx, &backup); err != nil {
		if !apierrors.IsAlreadyExists(err) {
			logger.Error("Failed to create raw backup CR", zap.Error(err))
			return nil, status.Errorf(codes.Internal, "Failed to create raw backup CR: %s", err)
		}
		logger.Info("Raw backup CR already exists, proceeding to verification", zap.String("backup_name", backup.Name))
	}

	// Retrieve the created CR to verify the spec
	created := &cloudv1.Backup{}
	if err := s.kubeClient.Get(ctx, client.ObjectKeyFromObject(&backup), created); err != nil {
		logger.Error("Failed to get created backup CR for verification", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to verify created backup CR: %s", err)
	}

	original.Spec.Paused = created.Spec.Paused // ignore paused field for comparison
	// Compare the created spec with the original (ignoring the paused field)
	if !equality.Semantic.DeepEqual(original.Spec, created.Spec) {
		// Spec mismatch: delete the CR and return an error
		diff := cmp.Diff(original.Spec, created.Spec, cmpopts.EquateEmpty())
		logger.Error("Created backup CR spec does not match expected spec", zap.String("diff", diff))
		return nil, status.Errorf(codes.Internal, "Created backup CR spec does not match expected spec")
	}

	// unpause CR for reconciliation
	created.Spec.Paused = nil
	delete(created.Labels, cloudv1.ReconcilerLabelKey) // remove reconciler label to allow reconciliation
	if err := s.kubeClient.Update(ctx, created); err != nil {
		logger.Error("Failed to unpause backup CR", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to unpause backup CR: %s", err)
	}

	result, err := yaml.Marshal(created)
	if err != nil {
		logger.Error("Failed to marshal backup CR response", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to marshal backup CR response: %s", err)
	}

	return &apiv1.RawCreateBackupCRResponse{Content: string(result)}, nil
}

// RawGetBackupCR gets a raw backup CR from Kubernetes without the API translation layer.
func (s *brServer) RawGetBackupCR(ctx context.Context, req *apiv1.RawGetBackupCRRequest) (*apiv1.RawGetBackupCRResponse, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.ClusterId), zap.String("name", req.Name))

	cluster, err := s.clusterReader.Get(ctx, s.kubeClient, req.ClusterId)
	if err != nil {
		logger.Error("Failed to get cluster", zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", req.ClusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", req.ClusterId, err)
	}

	backup := &cloudv1.Backup{}
	if err := s.kubeClient.Get(ctx, client.ObjectKey{Namespace: cluster.Namespace, Name: req.Name}, backup); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Backup CR %s not found", req.Name)
		}
		logger.Error("Failed to get backup CR", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to get backup CR %s: %s", req.Name, err)
	}

	result, err := yaml.Marshal(backup)
	if err != nil {
		logger.Error("Failed to marshal backup CR", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to marshal backup CR: %s", err)
	}

	return &apiv1.RawGetBackupCRResponse{Content: string(result)}, nil
}

func (s *brServer) EnableCRRF(ctx context.Context, req *apiv1.EnableCRRFRequest) (*apiv1.CRRFInfo, error) {
	logger := tilog.WithContextL(ctx).With(
		zap.String("cluster", req.ClusterId),
		zap.String("log_backup_name", req.LogBackupName),
		zap.String("crrf_id", req.CrrfId),
	)

	cluster, err := s.clusterReader.Get(ctx, s.kubeClient, req.ClusterId)
	if err != nil {
		logger.Error("Failed to get cluster", zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", req.ClusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", req.ClusterId, err)
	}

	name := getBackupName(req.LogBackupName)
	backup := &cloudv1.Backup{}
	if err := s.kubeClient.Get(ctx, client.ObjectKey{Namespace: cluster.Namespace, Name: name}, backup); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Log backup %s not found", req.LogBackupName)
		}
		logger.Error("Failed to get backup", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to get backup %s: %s", req.LogBackupName, err)
	}

	// Validate this is a log backup
	if backup.Spec.Mode == nil || *backup.Spec.Mode != cloudv1.BackupModeLog {
		return nil, status.Errorf(codes.InvalidArgument, "Backup %s is not a log backup", req.LogBackupName)
	}

	// Validate the backup belongs to the requested cluster
	if backup.Spec.ClusterName != cluster.Name {
		return nil, status.Errorf(codes.InvalidArgument, "Backup %s does not belong to cluster %s", req.LogBackupName, req.ClusterId)
	}

	// Check for existing crrf-spec annotation (idempotency check)
	if existing, ok := backup.Annotations[cloudv1.BackupAnnotationKeyCRRFSpec]; ok {
		var existingSpec cloudv1.CRRFSpec
		if err := json.Unmarshal([]byte(existing), &existingSpec); err == nil {
			if existingSpec.CRRFID == req.CrrfId {
				return cvt.CRRFStatus2PB(req.ClusterId, req.LogBackupName, backup)
			}
			return nil, status.Errorf(codes.AlreadyExists, "CRRF with different id %s already exists for backup %s", existingSpec.CRRFID, req.LogBackupName)
		}
	}

	// Build the CRRF spec
	crrfSpec := cloudv1.CRRFSpec{
		CRRFID:            req.CrrfId,
		DownstreamStorage: req.DownstreamStorage,
		Image:             req.Image,
	}

	specJSON, err := json.Marshal(crrfSpec)
	if err != nil {
		logger.Error("Failed to marshal CRRF spec", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to marshal CRRF spec: %s", err)
	}

	if backup.Annotations == nil {
		backup.Annotations = make(map[string]string)
	}
	backup.Annotations[cloudv1.BackupAnnotationKeyCRRFSpec] = string(specJSON)

	if err := s.kubeClient.Update(ctx, backup); err != nil {
		logger.Error("Failed to update backup with CRRF spec", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to update backup: %s", err)
	}

	logger.Info("CRRF enabled successfully")
	return &apiv1.CRRFInfo{
		CrrfId:        req.CrrfId,
		ClusterId:     req.ClusterId,
		LogBackupName: req.LogBackupName,
		Phase:         apiv1.CRRFInfo_PENDING,
	}, nil
}

func (s *brServer) DisableCRRF(ctx context.Context, req *apiv1.DisableCRRFRequest) (*apiv1.CRRFInfo, error) {
	logger := tilog.WithContextL(ctx).With(
		zap.String("cluster", req.ClusterId),
		zap.String("log_backup_name", req.LogBackupName),
		zap.String("crrf_id", req.CrrfId),
	)

	cluster, err := s.clusterReader.Get(ctx, s.kubeClient, req.ClusterId)
	if err != nil {
		logger.Error("Failed to get cluster", zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", req.ClusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", req.ClusterId, err)
	}

	name := getBackupName(req.LogBackupName)
	backup := &cloudv1.Backup{}
	if err := s.kubeClient.Get(ctx, client.ObjectKey{Namespace: cluster.Namespace, Name: name}, backup); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Log backup %s not found", req.LogBackupName)
		}
		logger.Error("Failed to get backup", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to get backup %s: %s", req.LogBackupName, err)
	}

	// Validate crrf-spec annotation exists
	existing, ok := backup.Annotations[cloudv1.BackupAnnotationKeyCRRFSpec]
	if !ok {
		return nil, status.Errorf(codes.NotFound, "CRRF is not enabled for backup %s", req.LogBackupName)
	}

	// Validate crrf_id matches
	var existingSpec cloudv1.CRRFSpec
	if err := json.Unmarshal([]byte(existing), &existingSpec); err != nil {
		logger.Error("Failed to unmarshal existing CRRF spec", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to parse existing CRRF spec: %s", err)
	}
	if existingSpec.CRRFID != req.CrrfId {
		return nil, status.Errorf(codes.InvalidArgument, "CRRF id mismatch: expected %s, got %s", existingSpec.CRRFID, req.CrrfId)
	}

	// Remove crrf-spec annotation to trigger controller cleanup
	delete(backup.Annotations, cloudv1.BackupAnnotationKeyCRRFSpec)

	if err := s.kubeClient.Update(ctx, backup); err != nil {
		logger.Error("Failed to update backup to remove CRRF spec", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to update backup: %s", err)
	}

	logger.Info("CRRF disabled successfully")
	return &apiv1.CRRFInfo{
		CrrfId:        req.CrrfId,
		ClusterId:     req.ClusterId,
		LogBackupName: req.LogBackupName,
		Phase:         apiv1.CRRFInfo_DESTROYING,
	}, nil
}

func (s *brServer) GetCRRF(ctx context.Context, req *apiv1.GetCRRFRequest) (*apiv1.CRRFInfo, error) {
	logger := tilog.WithContextL(ctx).With(
		zap.String("cluster", req.ClusterId),
		zap.String("log_backup_name", req.LogBackupName),
	)

	cluster, err := s.clusterReader.Get(ctx, s.kubeClient, req.ClusterId)
	if err != nil {
		logger.Error("Failed to get cluster", zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", req.ClusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", req.ClusterId, err)
	}

	name := getBackupName(req.LogBackupName)
	backup := &cloudv1.Backup{}
	if err := s.kubeClient.Get(ctx, client.ObjectKey{Namespace: cluster.Namespace, Name: name}, backup); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Log backup %s not found", req.LogBackupName)
		}
		logger.Error("Failed to get backup", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to get backup %s: %s", req.LogBackupName, err)
	}

	// Check if CRRF exists (either spec or status annotation)
	_, hasSpec := backup.Annotations[cloudv1.BackupAnnotationKeyCRRFSpec]
	_, hasStatus := backup.Annotations[cloudv1.BackupAnnotationKeyCRRFStatus]
	if !hasSpec && !hasStatus {
		return nil, status.Errorf(codes.NotFound, "CRRF not enabled on backup %s", req.LogBackupName)
	}

	return cvt.CRRFStatus2PB(req.ClusterId, req.LogBackupName, backup)
}

func (s *brServer) UpdateCRRF(ctx context.Context, req *apiv1.UpdateCRRFRequest) (*apiv1.CRRFInfo, error) {
	logger := tilog.WithContextL(ctx).With(
		zap.String("cluster", req.ClusterId),
		zap.String("log_backup_name", req.LogBackupName),
		zap.String("crrf_id", req.CrrfId),
	)

	cluster, err := s.clusterReader.Get(ctx, s.kubeClient, req.ClusterId)
	if err != nil {
		logger.Error("Failed to get cluster", zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", req.ClusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", req.ClusterId, err)
	}

	name := getBackupName(req.LogBackupName)
	backup := &cloudv1.Backup{}
	if err := s.kubeClient.Get(ctx, client.ObjectKey{Namespace: cluster.Namespace, Name: name}, backup); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Log backup %s not found", req.LogBackupName)
		}
		logger.Error("Failed to get backup", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to get backup %s: %s", req.LogBackupName, err)
	}

	if err := updateCrrfSpec(backup, func(spec *cloudv1.CRRFSpec) error {
		// Validate crrf_id matches
		if spec.CRRFID != req.CrrfId {
			return status.Errorf(codes.InvalidArgument, "CRRF id mismatch: expected %s, got %s", spec.CRRFID, req.CrrfId)
		}

		if req.Image != nil {
			logger.Info("update image", zap.String("before", spec.Image), zap.String("after", *req.Image))
			spec.Image = *req.Image
		}
		return nil
	}); err != nil {
		logger.Error("Failed to update CRRF spec on backup", zap.Error(err))
		return nil, err
	}

	if err := s.kubeClient.Update(ctx, backup); err != nil {
		logger.Error("Failed to update backup with new CRRF spec", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to update backup: %s", err)
	}

	logger.Info("CRRF updated successfully")
	return cvt.CRRFStatus2PB(req.ClusterId, req.LogBackupName, backup)
}

func updateCrrfSpec(backup *cloudv1.Backup, updateFunc func(*cloudv1.CRRFSpec) error) error {
	existing, ok := backup.Annotations[cloudv1.BackupAnnotationKeyCRRFSpec]
	if !ok {
		return status.Errorf(codes.NotFound, "CRRF is not enabled for backup %s", backup.Name)
	}

	var existingSpec cloudv1.CRRFSpec
	if err := json.Unmarshal([]byte(existing), &existingSpec); err != nil {
		return status.Errorf(codes.Internal, "Failed to parse existing CRRF spec: %s", err)
	}

	if err := updateFunc(&existingSpec); err != nil {
		return err
	}

	specJSON, err := json.Marshal(existingSpec)
	if err != nil {
		return status.Errorf(codes.Internal, "Failed to marshal updated CRRF spec: %s", err)
	}
	backup.Annotations[cloudv1.BackupAnnotationKeyCRRFSpec] = string(specJSON)
	return nil
}