feat: add get object and get bucket with payment apis

feat: modify long name

feat: sort import correctly

fix: typos

fix: typo

proto/service/metadata/types/metadata.proto

@@ -147,6 +147,22 @@ message GetPaymentByBucketIDRequest {
   bool is_full_list = 2;
 }
 
+// GetBucketWithPaymentRequest is request type for the GetBucketWithPayment RPC method
+message GetBucketWithPaymentRequest {
+  // bucket_name is the name of the bucket
+  string bucket_name = 1;
+  // is_full_list indicates whether this request can get the private buckets information
+  bool is_full_list = 2;
+}
+
+// GetBucketWithPaymentResponse is response type for the GetBucketWithPayment RPC method
+message GetBucketWithPaymentResponse {
+  // bucket defines the information of a bucket
+  Bucket bucket = 1;
+  // stream_record defines stream payment record of a stream account
+  bnbchain.greenfield.payment.StreamRecord stream_record = 2;
+}
+
 // GetUserBucketsResponse is response type for the GetUserBuckets RPC method.
 message GetUserBucketsResponse {
   // buckets defines the list of bucket
@@ -251,4 +267,6 @@ service MetadataService {
   rpc GetPaymentByBucketID(GetPaymentByBucketIDRequest) returns (GetPaymentByBucketIDResponse) {};
   // VerifyPermission Verify the input items permission.
   rpc VerifyPermission(bnbchain.greenfield.storage.QueryVerifyPermissionRequest) returns (bnbchain.greenfield.storage.QueryVerifyPermissionResponse) {};
+  // GetBucketWithPayment get bucket info with payment info by a bucket name
+  rpc GetBucketWithPayment(GetBucketWithPaymentRequest) returns (GetBucketWithPaymentResponse) {};
 }

service/gateway/metadata_handler.go

@@ -200,3 +200,124 @@ func (gateway *Gateway) listObjectsByBucketNameHandler(w http.ResponseWriter, r
 	w.Header().Set(model.ContentTypeHeader, model.ContentTypeJSONHeaderValue)
 	w.Write(b.Bytes())
 }
+
+// getObjectMetaByNameHandler handle get object info by name request
+func (gateway *Gateway) getObjectByNameHandler(w http.ResponseWriter, r *http.Request) {
+	var (
+		err            error
+		b              bytes.Buffer
+		errDescription *errorDescription
+		reqContext     *requestContext
+	)
+
+	reqContext = newRequestContext(r)
+	defer func() {
+		if errDescription != nil {
+			_ = errDescription.errorJSONResponse(w, reqContext)
+		}
+		if errDescription != nil && errDescription.statusCode != http.StatusOK {
+			log.Errorf("action(%v) statusCode(%v) %v", getObjectByNameRouterName, errDescription.statusCode, reqContext.generateRequestDetail())
+		} else {
+			log.Infof("action(%v) statusCode(200) %v", getObjectByNameRouterName, reqContext.generateRequestDetail())
+		}
+	}()
+
+	if gateway.metadata == nil {
+		log.Error("failed to get object by name due to not config metadata")
+		errDescription = NotExistComponentError
+		return
+	}
+
+	if err = s3util.CheckValidBucketName(reqContext.bucketName); err != nil {
+		log.Errorw("failed to check bucket name", "bucket_name", reqContext.bucketName, "error", err)
+		errDescription = InvalidBucketName
+		return
+	}
+
+	if err = s3util.CheckValidObjectName(reqContext.objectName); err != nil {
+		log.Errorw("failed to check object name", "object_name", reqContext.objectName, "error", err)
+		errDescription = InvalidKey
+		return
+	}
+
+	req := &metatypes.GetObjectByObjectNameAndBucketNameRequest{
+		BucketName: reqContext.bucketName,
+		ObjectName: reqContext.objectName,
+		IsFullList: true,
+	}
+
+	ctx := log.Context(context.Background(), req)
+	resp, err := gateway.metadata.GetObjectByObjectNameAndBucketName(ctx, req)
+	if err != nil {
+		log.Errorf("failed to get object by object name and bucket name", "error", err)
+		errDescription = makeErrorDescription(err)
+		return
+	}
+
+	m := jsonpb.Marshaler{EmitDefaults: true, OrigName: true, EnumsAsInts: true}
+	if err = m.Marshal(&b, resp); err != nil {
+		log.Errorf("failed to get object by object name and bucket name", "error", err)
+		errDescription = makeErrorDescription(err)
+		return
+	}
+
+	w.Header().Set(model.ContentTypeHeader, model.ContentTypeJSONHeaderValue)
+	w.Write(b.Bytes())
+}
+
+// getBucketWithPaymentHandler handle get bucket info with payment request
+func (gateway *Gateway) getBucketWithPaymentHandler(w http.ResponseWriter, r *http.Request) {
+	var (
+		err            error
+		b              bytes.Buffer
+		errDescription *errorDescription
+		reqContext     *requestContext
+	)
+
+	reqContext = newRequestContext(r)
+	defer func() {
+		if errDescription != nil {
+			_ = errDescription.errorJSONResponse(w, reqContext)
+		}
+		if errDescription != nil && errDescription.statusCode != http.StatusOK {
+			log.Errorf("action(%v) statusCode(%v) %v", getBucketAndPaymentRouterName, errDescription.statusCode, reqContext.generateRequestDetail())
+		} else {
+			log.Infof("action(%v) statusCode(200) %v", getBucketAndPaymentRouterName, reqContext.generateRequestDetail())
+		}
+	}()
+
+	if gateway.metadata == nil {
+		log.Error("failed to get bucket and payment info due to not config metadata")
+		errDescription = NotExistComponentError
+		return
+	}
+
+	if err = s3util.CheckValidBucketName(reqContext.bucketName); err != nil {
+		log.Errorw("failed to check bucket name", "bucket_name", reqContext.bucketName, "error", err)
+		errDescription = InvalidBucketName
+		return
+	}
+
+	req := &metatypes.GetBucketWithPaymentRequest{
+		BucketName: reqContext.bucketName,
+		IsFullList: true,
+	}
+
+	ctx := log.Context(context.Background(), req)
+	resp, err := gateway.metadata.GetBucketWithPayment(ctx, req)
+	if err != nil {
+		log.Errorf("failed to get bucket with payment", "error", err)
+		errDescription = makeErrorDescription(err)
+		return
+	}
+
+	m := jsonpb.Marshaler{EmitDefaults: true, OrigName: true, EnumsAsInts: true}
+	if err = m.Marshal(&b, resp); err != nil {
+		log.Errorf("failed to get bucket with payment", "error", err)
+		errDescription = makeErrorDescription(err)
+		return
+	}
+
+	w.Header().Set(model.ContentTypeHeader, model.ContentTypeJSONHeaderValue)
+	w.Write(b.Bytes())
+}

service/gateway/router.go

@@ -26,6 +26,8 @@ const (
 	queryUploadProgressRouterName         = "queryUploadProgress"
 	downloadObjectByUniversalEndpointName = "DownloadObjectByUniversalEndpoint"
 	viewObjectByUniversalEndpointName     = "ViewObjectByUniversalEndpoint"
+	getObjectByNameRouterName             = "getObjectByName"
+	getBucketAndPaymentRouterName         = "getBucketAndPayment"
 )
 
 const (
@@ -59,6 +61,16 @@ func (g *Gateway) registerHandler(r *mux.Router) {
 		Path("/{object:.+}").
 		Queries(model.UploadProgressQuery, "").
 		HandlerFunc(g.queryUploadProgressHandler)
+	hostBucketRouter.NewRoute().
+		Name(getBucketAndPaymentRouterName).
+		Methods(http.MethodGet).
+		Path("/bucket/payment").
+		HandlerFunc(g.getBucketWithPaymentHandler)
+	hostBucketRouter.NewRoute().
+		Name(getObjectByNameRouterName).
+		Methods(http.MethodGet).
+		Path("/{object:.+}/metadata").
+		HandlerFunc(g.getObjectByNameHandler)
 	hostBucketRouter.NewRoute().
 		Name(getObjectRouterName).
 		Methods(http.MethodGet).
@@ -85,7 +97,7 @@ func (g *Gateway) registerHandler(r *mux.Router) {
 		HandlerFunc(g.listObjectsByBucketNameHandler)
 	hostBucketRouter.NotFoundHandler = http.HandlerFunc(g.notFoundHandler)
 
-	// bucket list router, virtual-hosted style
+	// bucket list router, path style
 	bucketListRouter := r.Host(g.config.Domain).Subrouter()
 	bucketListRouter.NewRoute().
 		Name(getUserBucketsRouterName).
@@ -144,6 +156,16 @@ func (g *Gateway) registerHandler(r *mux.Router) {
 		Path("/{object:.+}").
 		Queries(model.UploadProgressQuery, "").
 		HandlerFunc(g.queryUploadProgressHandler)
+	pathBucketRouter.NewRoute().
+		Name(getBucketAndPaymentRouterName).
+		Methods(http.MethodGet).
+		Path("/bucket/payment").
+		HandlerFunc(g.getBucketWithPaymentHandler)
+	pathBucketRouter.NewRoute().
+		Name(getObjectByNameRouterName).
+		Methods(http.MethodGet).
+		Path("/{object:.+}/metadata").
+		HandlerFunc(g.getObjectByNameHandler)
 	pathBucketRouter.NewRoute().
 		Name(getObjectRouterName).
 		Methods(http.MethodGet).

service/metadata/client/metadata_client.go

@@ -164,3 +164,14 @@ func (client *MetadataClient) VerifyPermission(ctx context.Context, in *storaget
 	}
 	return resp, nil
 }
+
+// GetBucketWithPayment get bucket info with its related payment info
+func (client *MetadataClient) GetBucketWithPayment(ctx context.Context, in *metatypes.GetBucketWithPaymentRequest, opts ...grpc.CallOption) (*metatypes.GetBucketWithPaymentResponse, error) {
+	resp, err := client.metadata.GetBucketWithPayment(ctx, in, opts...)
+	ctx = log.Context(ctx, resp)
+	if err != nil {
+		log.CtxErrorw(ctx, "failed to send get bucket with payment rpc by bucket name", "error", err)
+		return nil, err
+	}
+	return resp, nil
+}

service/metadata/service/bucket.go

@@ -7,10 +7,12 @@ import (
 	"github.com/bnb-chain/greenfield/types/s3util"
 	"github.com/bnb-chain/greenfield/x/storage/types"
 	"github.com/forbole/juno/v4/common"
+	jsoniter "github.com/json-iterator/go"
 
 	"github.com/bnb-chain/greenfield-storage-provider/pkg/log"
 	metatypes "github.com/bnb-chain/greenfield-storage-provider/service/metadata/types"
 	model "github.com/bnb-chain/greenfield-storage-provider/store/bsdb"
+	paymenttypes "github.com/bnb-chain/greenfield/x/payment/types"
 )
 
 // GetUserBuckets get buckets info by a user address
@@ -210,3 +212,73 @@ func (metadata *Metadata) ListExpiredBucketsBySp(ctx context.Context, req *metat
 	log.CtxInfow(ctx, "succeed to get user buckets")
 	return resp, nil
 }
+
+func (metadata *Metadata) GetBucketWithPayment(ctx context.Context, req *metatypes.GetBucketWithPaymentRequest) (resp *metatypes.GetBucketWithPaymentResponse, err error) {
+	var (
+		bucket          *model.Bucket
+		bucketRes       *metatypes.Bucket
+		streamRecord    *model.StreamRecord
+		streamRecordRes *paymenttypes.StreamRecord
+		outflows        []paymenttypes.OutFlow
+	)
+
+	ctx = log.Context(ctx, req)
+	bucket, streamRecord, err = metadata.bsDB.GetBucketWithPayment(req.GetBucketName(), req.GetIsFullList())
+	if err != nil {
+		log.CtxErrorw(ctx, "failed to get bucket info with payment info", "error", err)
+		return
+	}
+
+	if bucket != nil {
+		bucketRes = &metatypes.Bucket{
+			BucketInfo: &types.BucketInfo{
+				Owner:            bucket.Owner.String(),
+				BucketName:       bucket.BucketName,
+				Id:               math.NewUintFromBigInt(bucket.BucketID.Big()),
+				SourceType:       types.SourceType(types.SourceType_value[bucket.SourceType]),
+				CreateAt:         bucket.CreateTime,
+				PaymentAddress:   bucket.PaymentAddress.String(),
+				PrimarySpAddress: bucket.PrimarySpAddress.String(),
+				ChargedReadQuota: bucket.ChargedReadQuota,
+				Visibility:       types.VisibilityType(types.VisibilityType_value[bucket.Visibility]),
+				BillingInfo: types.BillingInfo{
+					PriceTime:              0,
+					TotalChargeSize:        0,
+					SecondarySpObjectsSize: nil,
+				},
+				BucketStatus: types.BucketStatus(types.BucketStatus_value[bucket.Status]),
+			},
+			Removed:      bucket.Removed,
+			DeleteAt:     bucket.DeleteAt,
+			DeleteReason: bucket.DeleteReason,
+			Operator:     bucket.Operator.String(),
+			CreateTxHash: bucket.CreateTxHash.String(),
+			UpdateTxHash: bucket.UpdateTxHash.String(),
+			UpdateAt:     bucket.UpdateAt,
+			UpdateTime:   bucket.UpdateTime,
+		}
+	}
+
+	if streamRecord != nil {
+		err = jsoniter.Unmarshal(streamRecord.OutFlows, &outflows)
+		if err != nil {
+			log.CtxErrorw(ctx, "failed to unmarshal out flows", "error", err)
+			return
+		}
+		streamRecordRes = &paymenttypes.StreamRecord{
+			Account:         streamRecord.Account.String(),
+			CrudTimestamp:   streamRecord.CrudTimestamp,
+			NetflowRate:     math.NewIntFromBigInt(streamRecord.NetflowRate.Raw()),
+			StaticBalance:   math.NewIntFromBigInt(streamRecord.StaticBalance.Raw()),
+			BufferBalance:   math.NewIntFromBigInt(streamRecord.BufferBalance.Raw()),
+			LockBalance:     math.NewIntFromBigInt(streamRecord.LockBalance.Raw()),
+			Status:          paymenttypes.StreamAccountStatus(paymenttypes.StreamAccountStatus_value[streamRecord.Status]),
+			SettleTimestamp: streamRecord.SettleTimestamp,
+			OutFlows:        outflows,
+		}
+	}
+
+	resp = &metatypes.GetBucketWithPaymentResponse{Bucket: bucketRes, StreamRecord: streamRecordRes}
+	log.CtxInfow(ctx, "succeed to get bucket info with payment info")
+	return resp, nil
+}

service/metadata/service/object.go

@@ -164,7 +164,7 @@ func (metadata *Metadata) ListDeletedObjectsByBlockNumberRange(ctx context.Conte
 	return resp, nil
 }
 
-// GetObjectByObjectNameAndBucketName get object info by an object name
+// GetObjectByObjectNameAndBucketName get object info by an object name and bucket name
 func (metadata *Metadata) GetObjectByObjectNameAndBucketName(ctx context.Context, req *metatypes.GetObjectByObjectNameAndBucketNameRequest) (resp *metatypes.GetObjectByObjectNameAndBucketNameResponse, err error) {
 	var (
 		object *model.Object

store/bsdb/bucket.go

@@ -102,3 +102,27 @@ func (b *BsDBImpl) ListExpiredBucketsBySp(createAt int64, primarySpAddress strin
 
 	return buckets, err
 }
+
+func (b *BsDBImpl) GetBucketWithPayment(bucketName string, isFullList bool) (*Bucket, *StreamRecord, error) {
+	var (
+		bucketWithPayment *BucketWithPayment
+		err               error
+	)
+
+	if isFullList {
+		err = b.db.Table((&Bucket{}).TableName()).
+			Select("*").
+			Joins("left join stream_records on buckets.payment_address = stream_records.account").
+			Where("buckets.bucket_name = ?", bucketName).
+			Take(&bucketWithPayment).Error
+	} else {
+		err = b.db.Table((&Bucket{}).TableName()).
+			Select("*").
+			Joins("left join stream_records on buckets.payment_address = stream_records.account").
+			Where("buckets.bucket_name = ? and "+
+				"buckets.visibility='VISIBILITY_TYPE_PUBLIC_READ'", bucketName).
+			Take(&bucketWithPayment).Error
+	}
+
+	return &bucketWithPayment.Bucket, &bucketWithPayment.StreamRecord, err
+}

store/bsdb/bucket_schema.go

@@ -55,3 +55,9 @@ type Bucket struct {
 func (b *Bucket) TableName() string {
 	return BucketTableName
 }
+
+// BucketWithPayment is the structure for user bucket with its related payment info
+type BucketWithPayment struct {
+	Bucket
+	StreamRecord
+}

store/bsdb/database.go

@@ -40,6 +40,8 @@ type Metadata interface {
 	GetObjectByName(objectName string, bucketName string, isFullList bool) (*Object, error)
 	// GetSwitchDBSignal check if there is a signal to switch the database
 	GetSwitchDBSignal() (*MasterDB, error)
+	// GetBucketWithPayment get bucket info with its related payment info
+	GetBucketWithPayment(bucketName string, isFullList bool) (*Bucket, *StreamRecord, error)
 }
 
 // BSDB contains all the methods required by block syncer database