package bucket

import (
	"context"
	"fmt"

	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/emptypb"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"

	dmpbv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/dataplane-manager/v1"
	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	bizerr "github.com/tidbcloud/infra-provider/regional-server/pkg/common/errors"
	rsconfig "github.com/tidbcloud/infra-provider/regional-server/pkg/config"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/consts"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/bucket"
	buckettask "github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/bucket/task"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/ownership"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/repository"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/util"
)

type BucketService struct {
	dmpbv1.UnimplementedBucketServiceServer

	dataPlaneRW   repository.DataPlaneReaderWriter
	appBoxRW      repository.APPBoxReaderWriter
	bucketRW      repository.BucketReaderWriter
	bucketQueryer bucket.BucketQueryer
	config        rsconfig.RootConfig
}

func NewBucketService(
	dataPlaneRW repository.DataPlaneReaderWriter,
	appboxRW repository.APPBoxReaderWriter,
	bucketRW repository.BucketReaderWriter,
	bucketQueryer bucket.BucketQueryer,
	config rsconfig.RootConfig,
) dmpbv1.BucketServiceServer {
	return &BucketService{
		dataPlaneRW:   dataPlaneRW,
		appBoxRW:      appboxRW,
		bucketRW:      bucketRW,
		bucketQueryer: bucketQueryer,
		config:        config,
	}
}

func (s *BucketService) CreateBucket(ctx context.Context, req *dmpbv1.CreateBucketRequest) (*dmpbv1.Bucket, error) {
	// Check if the DataPlane exists
	dataplane := &v1alpha1.DataPlane{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: s.config.Namespace,
			Name:      req.DataplaneId,
		},
	}
	if err := s.dataPlaneRW.Get(ctx, client.ObjectKeyFromObject(dataplane), dataplane); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, bizerr.New(bizerr.ResourceNotFound, "dataplane not found")
		}
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get dataplane error", err)
	}
	appbox := &v1alpha1.APPBox{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: s.config.Namespace,
			Name:      dataplane.Status.AppBox.Name,
		},
	}
	if err := s.appBoxRW.Get(ctx, client.ObjectKeyFromObject(appbox), appbox); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, bizerr.New(bizerr.ResourceNotFound, "appbox not found")
		}
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get appbox error", err)
	}
	if appbox.Status.IAM.NodeRoleARN == "" && appbox.Status.IAM.KubeletIdentityObjectID == "" {
		return nil, bizerr.New(bizerr.ResourceNotReady, "node role or kubelet identity object id is not ready")
	}
	tenantID := dataplane.Labels[consts.TenantLabelKey]
	if tenantID == "" {
		return nil, bizerr.New(bizerr.ParameterInvalid, "missing required tag: tags.tidbcloud.com/tenant")
	}
	if req.Bucket.Region == "" {
		req.Bucket.Region = s.config.RegionConfig.Name
	}
	labels := map[string]string{
		v1alpha1.DataPlaneIDKey:             req.DataplaneId,
		consts.DataplaneLabelKey:            req.DataplaneId,
		consts.TenantLabelKey:               tenantID,
		"tags.tidbcloud.com/bucket-id":      req.Bucket.Id,
		"tags.tidbcloud.com/cloud-provider": appbox.Spec.Provider,
	}
	labels = util.MergeMaps(labels, req.Bucket.Tags, s.config.ResourceLabels, dataplane.Spec.CloudTags)

	bucket := v1alpha1.Bucket{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: s.getNamespace(),
			Name:      getBucketName(req.DataplaneId, req.Bucket.Id),
			Labels:    labels,
		},
		Spec: v1alpha1.BucketSpec{
			DataplaneId: req.DataplaneId,
			Id:          req.Bucket.Id,
			Region:      req.Bucket.Region,
			Tags:        req.Bucket.Tags,
			AccountRef:  dataplane.Spec.AccountRef.DeepCopy(),
			NodeRoleARN: appbox.Status.IAM.NodeRoleARN,
			Ownership:   ownership.FromRequest(req.OwnershipContext),
		},
	}
	if err := s.bucketRW.Create(ctx, &bucket); err != nil {
		if apierrors.IsAlreadyExists(err) {
			existing := &v1alpha1.Bucket{}
			if getErr := s.bucketRW.Get(ctx, client.ObjectKeyFromObject(&bucket), existing); getErr != nil {
				return nil, bizerr.Wrap(bizerr.K8SClientError, "get existing bucket error", getErr)
			}
			if matchErr := ownership.Match(req.OwnershipContext, existing.Spec.Ownership); matchErr != nil {
				return nil, matchErr
			}
		}
		return nil, bizerr.Wrap(bizerr.K8SClientError, "create bucket error", err)
	}

	return &dmpbv1.Bucket{
		Id:     req.Bucket.Id,
		Name:   req.Bucket.Name,
		Region: req.Bucket.Region,
		Status: "Pending",
	}, nil
}

func (s *BucketService) GetBucket(ctx context.Context, req *dmpbv1.GetBucketRequest) (*dmpbv1.Bucket, error) {
	bucket := &v1alpha1.Bucket{}
	if err := s.bucketRW.Get(ctx, types.NamespacedName{
		Namespace: s.getNamespace(),
		Name:      getBucketName(req.DataplaneId, req.BucketId),
	}, bucket); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, bizerr.New(bizerr.ResourceNotFound, "bucket not found")
		}
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get bucket error", err)
	}
	if err := ownership.Match(req.OwnershipContext, bucket.Spec.Ownership); err != nil {
		return nil, err
	}
	if bucket.Status.Bucket == nil || bucket.Status.Phase == "" {
		return &dmpbv1.Bucket{
			Id:     bucket.Spec.Id,
			Name:   bucket.Name,
			Region: bucket.Spec.Region,
			Status: "Pending",
		}, nil
	}

	accessPath, err := s.bucketQueryer.GetBucketAccessPath(ctx, bucket)
	if err != nil {
		return nil, bizerr.Wrap(bizerr.InternalServerError, "get bucket access path error", err)
	}

	return &dmpbv1.Bucket{
		Id:     bucket.Spec.Id,
		Name:   accessPath,
		Region: bucket.Spec.Region,
		Status: string(bucket.Status.Phase),
	}, nil
}

func (s *BucketService) ListBuckets(ctx context.Context, req *dmpbv1.ListBucketsRequest) (*dmpbv1.ListBucketsResponse, error) {
	bucketList := &v1alpha1.BucketList{}
	listOpts := &client.ListOptions{
		LabelSelector: labels.SelectorFromSet(map[string]string{
			consts.DataplaneLabelKey: req.DataplaneId,
		}),
	}
	err := s.bucketRW.List(ctx, bucketList, listOpts)
	if err != nil {
		return nil, bizerr.Wrap(bizerr.K8SClientError, "list buckets error", err)
	}
	buckets := make([]*dmpbv1.Bucket, 0, len(bucketList.Items))
	for _, bucket := range bucketList.Items {
		if bucket.Status.Bucket == nil || bucket.Status.Phase == "" {
			buckets = append(buckets, &dmpbv1.Bucket{
				Id:     bucket.Spec.Id,
				Name:   bucket.Name,
				Region: bucket.Spec.Region,
				Status: "Creating",
			})
			continue
		}
		accessPath, err := s.bucketQueryer.GetBucketAccessPath(ctx, &bucket)
		if err != nil {
			return nil, bizerr.Wrap(bizerr.InternalServerError, "get bucket access path error", err)
		}
		buckets = append(buckets, &dmpbv1.Bucket{
			Id:     bucket.Spec.Id,
			Name:   accessPath,
			Region: bucket.Spec.Region,
			Status: string(bucket.Status.Phase),
		})
	}
	return &dmpbv1.ListBucketsResponse{Buckets: buckets}, nil
}

func (s *BucketService) DeleteBucket(ctx context.Context, req *dmpbv1.DeleteBucketRequest) (*emptypb.Empty, error) {
	bucket := &v1alpha1.Bucket{}
	if err := s.bucketRW.Get(ctx, types.NamespacedName{
		Namespace: s.getNamespace(),
		Name:      getBucketName(req.DataplaneId, req.BucketId),
	}, bucket); err != nil {
		if apierrors.IsNotFound(err) {
			return &emptypb.Empty{}, nil
		}
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get bucket error", err)
	}
	if err := ownership.Match(req.OwnershipContext, bucket.Spec.Ownership); err != nil {
		return nil, err
	}

	if err := s.bucketRW.Delete(ctx, bucket); err != nil {
		return nil, bizerr.Wrap(bizerr.K8SClientError, "delete bucket error", err)
	}

	return &emptypb.Empty{}, nil
}

func (s *BucketService) getNamespace() string {
	return s.config.Namespace
}

func getBucketName(dataplaneID, bucketID string) string {
	return fmt.Sprintf("%s-%s", dataplaneID, bucketID)
}

func (s *BucketService) GrantDataplanePermission(ctx context.Context, req *dmpbv1.GrantDataplanePermissionRequest) (*emptypb.Empty, error) {
	// Get the bucket
	bucket := &v1alpha1.Bucket{}
	if err := s.bucketRW.Get(ctx, types.NamespacedName{
		Namespace: s.getNamespace(),
		Name:      getBucketName(req.DataplaneId, req.BucketId),
	}, bucket); err != nil {
		tilog.Error(ctx, "get bucket error", zap.Any("request", req), zap.Error(err))
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get bucket error", err)
	}
	if err := ownership.Match(req.OwnershipContext, bucket.Spec.Ownership); err != nil {
		return nil, err
	}

	enabled, err := s.isAKSBYOCStaticAuthEnabled()
	if err != nil {
		return nil, bizerr.Wrap(bizerr.InternalServerError, "invalid azure byoc config", err)
	}
	if enabled {
		tilog.Info(ctx, "skip granting dataplane permission for azure byoc static auth", zap.Any("request", req))
		return &emptypb.Empty{}, nil
	}

	// Validate that the permitted dataplane exists
	permittedDataPlane := &v1alpha1.DataPlane{}
	if err := s.dataPlaneRW.Get(ctx, client.ObjectKey{
		Namespace: s.config.Namespace,
		Name:      req.PermittedDataplaneId,
	}, permittedDataPlane); err != nil {
		tilog.Error(ctx, "get permitted dataplane error", zap.Any("request", req), zap.Error(err))
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get permitted dataplane error", err)
	}

	for _, dpId := range bucket.Spec.GrantedDataplaneIds {
		if dpId == req.PermittedDataplaneId {
			tilog.Info(ctx, "permission granting already exists", zap.Any("request", req))
			return nil, nil
		}
	}

	tilog.Info(ctx, "granting permission", zap.Any("request", req))
	bucket.Spec.GrantedDataplaneIds = append(bucket.Spec.GrantedDataplaneIds, req.PermittedDataplaneId)
	err = s.bucketRW.Update(ctx, bucket)
	if err != nil {
		tilog.Error(ctx, "apply bucket error", zap.Any("request", req), zap.Error(err))
		return nil, bizerr.Wrap(bizerr.K8SClientError, "apply bucket error", err)
	}
	tilog.Info(ctx, "permission granted", zap.Any("request", req))

	return &emptypb.Empty{}, nil
}

func (s *BucketService) RevokeDataplanePermission(ctx context.Context, req *dmpbv1.RevokeDataplanePermissionRequest) (*emptypb.Empty, error) {
	// Get the bucket
	bucket := &v1alpha1.Bucket{}
	if err := s.bucketRW.Get(ctx, types.NamespacedName{
		Namespace: s.getNamespace(),
		Name:      getBucketName(req.DataplaneId, req.BucketId),
	}, bucket); err != nil {
		tilog.Error(ctx, "get bucket error", zap.Any("request", req), zap.Error(err))
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get bucket error", err)
	}
	if err := ownership.Match(req.OwnershipContext, bucket.Spec.Ownership); err != nil {
		return nil, err
	}

	enabled, err := s.isAKSBYOCStaticAuthEnabled()
	if err != nil {
		return nil, bizerr.Wrap(bizerr.InternalServerError, "invalid azure byoc config", err)
	}
	if enabled {
		tilog.Info(ctx, "skip revoking dataplane permission for azure byoc static auth", zap.Any("request", req))
		return &emptypb.Empty{}, nil
	}

	// Find and remove the auth
	dpIdFound := false
	var newGrantedDataplaneIds []string
	for _, dpId := range bucket.Spec.GrantedDataplaneIds {
		if dpId != req.PermittedDataplaneId {
			newGrantedDataplaneIds = append(newGrantedDataplaneIds, dpId)
		} else {
			dpIdFound = true
		}
	}
	if !dpIdFound {
		tilog.Info(ctx, "permission not found, no need to delete", zap.Any("request", req))
		return &emptypb.Empty{}, nil
	}

	// Update the bucket with the new permission list
	bucket.Spec.GrantedDataplaneIds = newGrantedDataplaneIds
	err = s.bucketRW.Update(ctx, bucket)
	if err != nil {
		tilog.Error(ctx, "apply bucket error", zap.Any("request", req), zap.Error(err))
		return nil, bizerr.Wrap(bizerr.K8SClientError, "apply bucket error", err)
	}
	tilog.Info(ctx, "permission revoked", zap.Any("request", req))

	return &emptypb.Empty{}, nil
}

func (s *BucketService) GetDataplanePermission(ctx context.Context, req *dmpbv1.GetDataplanePermissionRequest) (*dmpbv1.GetDataplanePermissionResponse, error) {
	// Get the bucket
	bucket := &v1alpha1.Bucket{}
	if err := s.bucketRW.Get(ctx, types.NamespacedName{
		Namespace: s.getNamespace(),
		Name:      getBucketName(req.DataplaneId, req.BucketId),
	}, bucket); err != nil {
		tilog.Error(ctx, "get bucket error", zap.Any("request", req), zap.Error(err))
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get bucket error", err)
	}
	if err := ownership.Match(req.OwnershipContext, bucket.Spec.Ownership); err != nil {
		return nil, err
	}

	enabled, err := s.isAKSBYOCStaticAuthEnabled()
	if err != nil {
		return nil, bizerr.Wrap(bizerr.InternalServerError, "invalid azure byoc config", err)
	}
	if enabled {
		return &dmpbv1.GetDataplanePermissionResponse{
			Status: dmpbv1.GetDataplanePermissionResponse_Synced,
		}, nil
	}

	status, err := s.bucketQueryer.GetDataplanePermissionStatus(ctx, s.getNamespace(), getBucketName(req.DataplaneId, req.BucketId), req.PermittedDataplaneId)
	if err != nil {
		tilog.Error(ctx, "get bucket auth error", zap.Any("request", req), zap.Error(err))
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get bucket auth error", err)
	}
	return convertDataplanePermissionStatus(status), nil
}

func (s *BucketService) isAKSBYOCStaticAuthEnabled() (bool, error) {
	return s.config.IsAKSBYOCStaticAuthEnabled()
}

func convertDataplanePermissionStatus(status *buckettask.BucketDataplanePermissionStatus) *dmpbv1.GetDataplanePermissionResponse {
	switch *status {
	case buckettask.BucketDataplanePermissionStatusNotExists:
		return &dmpbv1.GetDataplanePermissionResponse{
			Status: dmpbv1.GetDataplanePermissionResponse_NotExists,
		}
	case buckettask.BucketDataplanePermissionStatusPending, buckettask.BucketDataplanePermissionStatusAccepted:
		return &dmpbv1.GetDataplanePermissionResponse{
			Status: dmpbv1.GetDataplanePermissionResponse_Accepted,
		}
	case buckettask.BucketDataplanePermissionStatusSynced:
		return &dmpbv1.GetDataplanePermissionResponse{
			Status: dmpbv1.GetDataplanePermissionResponse_Synced,
		}
	default:
		return &dmpbv1.GetDataplanePermissionResponse{
			Status: dmpbv1.GetDataplanePermissionResponse_DataplanePermission_Unspecified,
		}
	}
}

func (s *BucketService) ApplyLifecycleRules(ctx context.Context, req *dmpbv1.ApplyLifecycleRulesRequest) (*emptypb.Empty, error) {
	bucket := &v1alpha1.Bucket{}
	if err := s.bucketRW.Get(ctx, types.NamespacedName{
		Namespace: s.getNamespace(),
		Name:      getBucketName(req.DataplaneId, req.BucketId),
	}, bucket); err != nil {
		tilog.Error(ctx, "get bucket error", zap.Any("request", req), zap.Error(err))
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get bucket error", err)
	}
	if err := ownership.Match(req.OwnershipContext, bucket.Spec.Ownership); err != nil {
		return nil, err
	}

	bucket.Spec.LifecycleRules = convertLifecycleRules2CR(req.Rules)

	err := s.bucketRW.Update(ctx, bucket)
	if err != nil {
		tilog.Error(ctx, "apply bucket error", zap.Any("request", req), zap.Error(err))
		return nil, bizerr.Wrap(bizerr.K8SClientError, "apply bucket error", err)
	}
	return &emptypb.Empty{}, nil
}

func convertLifecycleRules2CR(reqRules []*dmpbv1.LifecycleRule) []v1alpha1.BucketLifecycleRule {
	rules := make([]v1alpha1.BucketLifecycleRule, 0, len(reqRules))
	for _, r := range reqRules {
		tagsFilter := make([]v1alpha1.BucketTagFilter, 0, len(r.TagsFilter))
		for _, tag := range r.TagsFilter {
			tagsFilter = append(tagsFilter, v1alpha1.BucketTagFilter{
				Key:   tag.Key,
				Value: tag.Value,
			})
		}
		rules = append(rules, v1alpha1.BucketLifecycleRule{
			Id:                             r.Id,
			Enabled:                        r.Enabled,
			PrefixFilter:                   r.PrefixFilter,
			TagsFilter:                     tagsFilter,
			ExpirationDays:                 r.ExpirationDays,
			IncompleteUploadExpirationDays: r.IncompleteUploadExpirationDays,
		})
	}
	return rules
}

func (s *BucketService) GetLifecycleRules(ctx context.Context, req *dmpbv1.GetLifecycleRulesRequest) (*dmpbv1.GetLifecycleRulesResponse, error) {
	bucket := &v1alpha1.Bucket{}
	if err := s.bucketRW.Get(ctx, types.NamespacedName{
		Namespace: s.getNamespace(),
		Name:      getBucketName(req.DataplaneId, req.BucketId),
	}, bucket); err != nil {
		tilog.Error(ctx, "get bucket error", zap.Any("request", req), zap.Error(err))
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get bucket error", err)
	}
	if err := ownership.Match(req.OwnershipContext, bucket.Spec.Ownership); err != nil {
		return nil, err
	}

	status, err := s.bucketQueryer.GetLifecycleRulesStatus(ctx, bucket)
	if err != nil {
		tilog.Error(ctx, "get bucket lifecycle rules status error", zap.Any("request", req), zap.Error(err))
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get bucket lifecycle rules status error", err)
	}

	allRules := bucket.Spec.GetAllLifecycleRules()
	resp := &dmpbv1.GetLifecycleRulesResponse{
		Status: convertLifecycleRulesStatus(status),
		Rules:  convertLifecycleRules2Resp(allRules),
	}

	return resp, nil
}

func convertLifecycleRulesStatus(status *buckettask.BucketLifecycleRulesStatus) dmpbv1.GetLifecycleRulesResponse_Status {
	switch *status {
	case buckettask.BucketLifecycleRulesStatusUnsynced:
		return dmpbv1.GetLifecycleRulesResponse_Unsynced
	case buckettask.BucketLifecycleRulesStatusSynced:
		return dmpbv1.GetLifecycleRulesResponse_Synced
	}
	return dmpbv1.GetLifecycleRulesResponse_LifecycleRules_Unspecified
}

func (s *BucketService) AddManualBackupLifecycleRules(ctx context.Context, req *dmpbv1.AddManualBackupLifecycleRulesRequest) (*emptypb.Empty, error) {
	bucket := &v1alpha1.Bucket{}
	if err := s.bucketRW.Get(ctx, types.NamespacedName{
		Namespace: s.getNamespace(),
		Name:      getBucketName(req.DataplaneId, req.BucketId),
	}, bucket); err != nil {
		tilog.Error(ctx, "get bucket error", zap.Any("request", req), zap.Error(err))
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get bucket error", err)
	}
	if err := ownership.Match(req.OwnershipContext, bucket.Spec.Ownership); err != nil {
		return nil, err
	}

	existingByID := make(map[string]int, len(bucket.Spec.ManualBackupLifecycleRules))
	for i, rule := range bucket.Spec.ManualBackupLifecycleRules {
		existingByID[rule.Id] = i
	}
	for _, newRule := range convertLifecycleRules2CR(req.Rules) {
		if idx, exists := existingByID[newRule.Id]; exists {
			bucket.Spec.ManualBackupLifecycleRules[idx] = newRule
		} else {
			bucket.Spec.ManualBackupLifecycleRules = append(bucket.Spec.ManualBackupLifecycleRules, newRule)
		}
	}

	if err := s.bucketRW.Update(ctx, bucket); err != nil {
		tilog.Error(ctx, "update bucket error", zap.Any("request", req), zap.Error(err))
		return nil, bizerr.Wrap(bizerr.K8SClientError, "update bucket error", err)
	}
	return &emptypb.Empty{}, nil
}

func (s *BucketService) DeleteManualBackupLifecycleRules(ctx context.Context, req *dmpbv1.DeleteManualBackupLifecycleRulesRequest) (*emptypb.Empty, error) {
	bucket := &v1alpha1.Bucket{}
	if err := s.bucketRW.Get(ctx, types.NamespacedName{
		Namespace: s.getNamespace(),
		Name:      getBucketName(req.DataplaneId, req.BucketId),
	}, bucket); err != nil {
		tilog.Error(ctx, "get bucket error", zap.Any("request", req), zap.Error(err))
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get bucket error", err)
	}
	if err := ownership.Match(req.OwnershipContext, bucket.Spec.Ownership); err != nil {
		return nil, err
	}

	toDelete := make(map[string]bool, len(req.RuleIds))
	for _, id := range req.RuleIds {
		toDelete[id] = true
	}

	filtered := make([]v1alpha1.BucketLifecycleRule, 0, len(bucket.Spec.ManualBackupLifecycleRules))
	for _, rule := range bucket.Spec.ManualBackupLifecycleRules {
		if !toDelete[rule.Id] {
			filtered = append(filtered, rule)
		}
	}
	bucket.Spec.ManualBackupLifecycleRules = filtered

	if err := s.bucketRW.Update(ctx, bucket); err != nil {
		tilog.Error(ctx, "update bucket error", zap.Any("request", req), zap.Error(err))
		return nil, bizerr.Wrap(bizerr.K8SClientError, "update bucket error", err)
	}
	return &emptypb.Empty{}, nil
}

func convertLifecycleRules2Resp(rules []v1alpha1.BucketLifecycleRule) []*dmpbv1.LifecycleRule {
	respRules := make([]*dmpbv1.LifecycleRule, 0, len(rules))
	for _, rule := range rules {
		tagsFilter := make([]*dmpbv1.LifecycleRule_Tag, 0, len(rule.TagsFilter))
		for _, tag := range rule.TagsFilter {
			tagsFilter = append(tagsFilter, &dmpbv1.LifecycleRule_Tag{
				Key:   tag.Key,
				Value: tag.Value,
			})
		}
		respRules = append(respRules, &dmpbv1.LifecycleRule{
			Id:                             rule.Id,
			Enabled:                        rule.Enabled,
			PrefixFilter:                   rule.PrefixFilter,
			TagsFilter:                     tagsFilter,
			ExpirationDays:                 rule.ExpirationDays,
			IncompleteUploadExpirationDays: rule.IncompleteUploadExpirationDays,
		})
	}
	return respRules
}