Merge pull request 'cds支持rclone挂载' (#23) from feature_gxh into master

agent/internal/task/create_package.go

@@ -63,7 +63,7 @@ func (t *CreatePackage) Execute(task *task.Task[TaskContext], ctx TaskContext, c
 		return
 	}
 
-	uploadRet, err := cmd.NewUploadObjects(t.userID, createResp.PackageID, t.objIter, t.nodeAffinity).Execute(&cmd.UploadObjectsContext{
+	uploadRet, err := cmd.NewUploadObjects(t.userID, createResp.Package.PackageID, t.objIter, t.nodeAffinity).Execute(&cmd.UploadObjectsContext{
 		Distlock:     ctx.distlock,
 		Connectivity: ctx.connectivity,
 	})
@@ -76,7 +76,7 @@ func (t *CreatePackage) Execute(task *task.Task[TaskContext], ctx TaskContext, c
 		return
 	}
 
-	t.Result.PackageID = createResp.PackageID
+	t.Result.PackageID = createResp.Package.PackageID
 	t.Result.Objects = uploadRet.Objects
 
 	complete(nil, CompleteOption{

client/internal/http/bucket.go

@@ -19,17 +19,39 @@ func (s *Server) Bucket() *BucketService {
 	}
 }
 
+func (s *BucketService) GetByName(ctx *gin.Context) {
+	log := logger.WithField("HTTP", "Bucket.GetByName")
+
+	var req cdssdk.BucketGetByName
+	if err := ctx.ShouldBindQuery(&req); err != nil {
+		log.Warnf("binding query: %s", err.Error())
+		ctx.JSON(http.StatusBadRequest, Failed(errorcode.BadArgument, "missing argument or invalid argument"))
+		return
+	}
+
+	bucket, err := s.svc.BucketSvc().GetBucketByName(req.UserID, req.Name)
+	if err != nil {
+		log.Warnf("getting bucket by name: %s", err.Error())
+		ctx.JSON(http.StatusOK, Failed(errorcode.OperationFailed, "get bucket by name failed"))
+		return
+	}
+
+	ctx.JSON(http.StatusOK, OK(cdssdk.BucketGetByNameResp{
+		Bucket: bucket,
+	}))
+}
+
 func (s *BucketService) Create(ctx *gin.Context) {
 	log := logger.WithField("HTTP", "Bucket.Create")
 
-	var req cdssdk.BucketCreateReq
+	var req cdssdk.BucketCreate
 	if err := ctx.ShouldBindJSON(&req); err != nil {
 		log.Warnf("binding body: %s", err.Error())
 		ctx.JSON(http.StatusBadRequest, Failed(errorcode.BadArgument, "missing argument or invalid argument"))
 		return
 	}
 
-	bucketID, err := s.svc.BucketSvc().CreateBucket(req.UserID, req.BucketName)
+	bucket, err := s.svc.BucketSvc().CreateBucket(req.UserID, req.Name)
 	if err != nil {
 		log.Warnf("creating bucket: %s", err.Error())
 		ctx.JSON(http.StatusOK, Failed(errorcode.OperationFailed, "create bucket failed"))
@@ -37,14 +59,14 @@ func (s *BucketService) Create(ctx *gin.Context) {
 	}
 
 	ctx.JSON(http.StatusOK, OK(cdssdk.BucketCreateResp{
-		BucketID: bucketID,
+		Bucket: bucket,
 	}))
 }
 
 func (s *BucketService) Delete(ctx *gin.Context) {
 	log := logger.WithField("HTTP", "Bucket.Delete")
 
-	var req cdssdk.BucketDeleteReq
+	var req cdssdk.BucketDelete
 	if err := ctx.ShouldBindJSON(&req); err != nil {
 		log.Warnf("binding body: %s", err.Error())
 		ctx.JSON(http.StatusBadRequest, Failed(errorcode.BadArgument, "missing argument or invalid argument"))
@@ -59,3 +81,25 @@ func (s *BucketService) Delete(ctx *gin.Context) {
 
 	ctx.JSON(http.StatusOK, OK(nil))
 }
+
+func (s *BucketService) ListUserBuckets(ctx *gin.Context) {
+	log := logger.WithField("HTTP", "Bucket.ListUserBuckets")
+
+	var req cdssdk.BucketListUserBucketsReq
+	if err := ctx.ShouldBindQuery(&req); err != nil {
+		log.Warnf("binding query: %s", err.Error())
+		ctx.JSON(http.StatusBadRequest, Failed(errorcode.BadArgument, "missing argument or invalid argument"))
+		return
+	}
+
+	buckets, err := s.svc.BucketSvc().GetUserBuckets(req.UserID)
+	if err != nil {
+		log.Warnf("getting user buckets: %s", err.Error())
+		ctx.JSON(http.StatusOK, Failed(errorcode.OperationFailed, "get user buckets failed"))
+		return
+	}
+
+	ctx.JSON(http.StatusOK, OK(cdssdk.BucketListUserBucketsResp{
+		Buckets: buckets,
+	}))
+}

client/internal/http/object.go

@@ -1,16 +1,18 @@
 package http
 
 import (
+	"fmt"
 	"io"
 	"mime/multipart"
 	"net/http"
+	"path"
 	"time"
 
 	"github.com/gin-gonic/gin"
 	"gitlink.org.cn/cloudream/common/consts/errorcode"
 	"gitlink.org.cn/cloudream/common/pkgs/logger"
 	cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
-	myio "gitlink.org.cn/cloudream/common/utils/io"
+	myhttp "gitlink.org.cn/cloudream/common/utils/http"
 )
 
 type ObjectService struct {
@@ -51,7 +53,7 @@ func (s *ObjectService) Upload(ctx *gin.Context) {
 	}
 
 	for {
-		complete, _, err := s.svc.ObjectSvc().WaitUploading(taskID, time.Second*5)
+		complete, objs, err := s.svc.ObjectSvc().WaitUploading(taskID, time.Second*5)
 		if complete {
 			if err != nil {
 				log.Warnf("uploading object: %s", err.Error())
@@ -59,7 +61,20 @@ func (s *ObjectService) Upload(ctx *gin.Context) {
 				return
 			}
 
-			ctx.JSON(http.StatusOK, OK(nil))
+			uploadeds := make([]cdssdk.UploadedObject, len(objs.Objects))
+			for i, obj := range objs.Objects {
+				err := ""
+				if obj.Error != nil {
+					err = obj.Error.Error()
+				}
+				o := obj.Object
+				uploadeds[i] = cdssdk.UploadedObject{
+					Object: &o,
+					Error:  err,
+				}
+			}
+
+			ctx.JSON(http.StatusOK, OK(cdssdk.ObjectUploadResp{Uploadeds: uploadeds}))
 			return
 		}
 
@@ -71,81 +86,114 @@ func (s *ObjectService) Upload(ctx *gin.Context) {
 	}
 }
 
-type ObjectDownloadReq struct {
-	UserID   *cdssdk.UserID   `form:"userID" binding:"required"`
-	ObjectID *cdssdk.ObjectID `form:"objectID" binding:"required"`
-}
-
 func (s *ObjectService) Download(ctx *gin.Context) {
 	log := logger.WithField("HTTP", "Object.Download")
 
-	var req ObjectDownloadReq
+	var req cdssdk.ObjectDownload
 	if err := ctx.ShouldBindQuery(&req); err != nil {
 		log.Warnf("binding body: %s", err.Error())
 		ctx.JSON(http.StatusBadRequest, Failed(errorcode.BadArgument, "missing argument or invalid argument"))
 		return
 	}
 
-	file, err := s.svc.ObjectSvc().Download(*req.UserID, *req.ObjectID)
+	file, err := s.svc.ObjectSvc().Download(req.UserID, req.ObjectID)
 	if err != nil {
 		log.Warnf("downloading object: %s", err.Error())
 		ctx.JSON(http.StatusOK, Failed(errorcode.OperationFailed, "download object failed"))
 		return
 	}
 
+	mw := multipart.NewWriter(ctx.Writer)
+	defer mw.Close()
+
+	ctx.Writer.Header().Set("Content-Type", fmt.Sprintf("%s;boundary=%s", myhttp.ContentTypeMultiPart, mw.Boundary()))
 	ctx.Writer.WriteHeader(http.StatusOK)
-	// TODO 需要设置FileName
-	ctx.Header("Content-Disposition", "attachment; filename=filename")
-	ctx.Header("Content-Type", "application/octet-stream")
-
-	buf := make([]byte, 4096)
-	ctx.Stream(func(w io.Writer) bool {
-		rd, err := file.Read(buf)
-		if err == io.EOF {
-			err = myio.WriteAll(w, buf[:rd])
-			if err != nil {
-				log.Warnf("writing data to response: %s", err.Error())
-			}
-			return false
-		}
 
-		if err != nil {
-			log.Warnf("reading file data: %s", err.Error())
-			return false
-		}
+	fw, err := mw.CreateFormFile("file", path.Base(file.Object.Path))
+	if err != nil {
+		log.Warnf("creating form file: %s", err.Error())
+		return
+	}
 
-		err = myio.WriteAll(w, buf[:rd])
-		if err != nil {
-			log.Warnf("writing data to response: %s", err.Error())
-			return false
-		}
+	io.Copy(fw, file.File)
+}
+
+func (s *ObjectService) UpdateInfo(ctx *gin.Context) {
+	log := logger.WithField("HTTP", "Object.UpdateInfo")
+
+	var req cdssdk.ObjectUpdateInfo
+	if err := ctx.ShouldBindJSON(&req); err != nil {
+		log.Warnf("binding body: %s", err.Error())
+		ctx.JSON(http.StatusBadRequest, Failed(errorcode.BadArgument, "missing argument or invalid argument"))
+		return
+	}
+
+	sucs, err := s.svc.ObjectSvc().UpdateInfo(req.UserID, req.Updatings)
+	if err != nil {
+		log.Warnf("updating objects: %s", err.Error())
+		ctx.JSON(http.StatusOK, Failed(errorcode.OperationFailed, "update objects failed"))
+		return
+	}
+
+	ctx.JSON(http.StatusOK, OK(cdssdk.ObjectUpdateInfoResp{Successes: sucs}))
+}
+
+func (s *ObjectService) Move(ctx *gin.Context) {
+	log := logger.WithField("HTTP", "Object.Move")
+
+	var req cdssdk.ObjectMove
+	if err := ctx.ShouldBindJSON(&req); err != nil {
+		log.Warnf("binding body: %s", err.Error())
+		ctx.JSON(http.StatusBadRequest, Failed(errorcode.BadArgument, "missing argument or invalid argument"))
+		return
+	}
+
+	sucs, err := s.svc.ObjectSvc().Move(req.UserID, req.Movings)
+	if err != nil {
+		log.Warnf("moving objects: %s", err.Error())
+		ctx.JSON(http.StatusOK, Failed(errorcode.OperationFailed, "move objects failed"))
+		return
+	}
 
-		return true
-	})
+	ctx.JSON(http.StatusOK, OK(cdssdk.ObjectMoveResp{Successes: sucs}))
 }
 
-type GetPackageObjectsReq struct {
-	UserID    *cdssdk.UserID    `form:"userID" binding:"required"`
-	PackageID *cdssdk.PackageID `form:"packageID" binding:"required"`
+func (s *ObjectService) Delete(ctx *gin.Context) {
+	log := logger.WithField("HTTP", "Object.Delete")
+
+	var req cdssdk.ObjectDelete
+	if err := ctx.ShouldBindJSON(&req); err != nil {
+		log.Warnf("binding body: %s", err.Error())
+		ctx.JSON(http.StatusBadRequest, Failed(errorcode.BadArgument, "missing argument or invalid argument"))
+		return
+	}
+
+	err := s.svc.ObjectSvc().Delete(req.UserID, req.ObjectIDs)
+	if err != nil {
+		log.Warnf("deleting objects: %s", err.Error())
+		ctx.JSON(http.StatusOK, Failed(errorcode.OperationFailed, "delete objects failed"))
+		return
+	}
+
+	ctx.JSON(http.StatusOK, OK(nil))
 }
-type GetPackageObjectsResp = cdssdk.ObjectGetPackageObjectsResp
 
 func (s *ObjectService) GetPackageObjects(ctx *gin.Context) {
 	log := logger.WithField("HTTP", "Object.GetPackageObjects")
 
-	var req GetPackageObjectsReq
+	var req cdssdk.ObjectGetPackageObjects
 	if err := ctx.ShouldBindQuery(&req); err != nil {
 		log.Warnf("binding body: %s", err.Error())
 		ctx.JSON(http.StatusBadRequest, Failed(errorcode.BadArgument, "missing argument or invalid argument"))
 		return
 	}
 
-	objs, err := s.svc.ObjectSvc().GetPackageObjects(*req.UserID, *req.PackageID)
+	objs, err := s.svc.ObjectSvc().GetPackageObjects(req.UserID, req.PackageID)
 	if err != nil {
 		log.Warnf("getting package objects: %s", err.Error())
 		ctx.JSON(http.StatusOK, Failed(errorcode.OperationFailed, "get package object failed"))
 		return
 	}
 
-	ctx.JSON(http.StatusOK, OK(GetPackageObjectsResp{Objects: objs}))
+	ctx.JSON(http.StatusOK, OK(cdssdk.ObjectGetPackageObjectsResp{Objects: objs}))
 }

client/internal/http/package.go

@@ -3,6 +3,7 @@ package http
 import (
 	"mime/multipart"
 	"net/http"
+	"net/url"
 
 	"github.com/gin-gonic/gin"
 	"gitlink.org.cn/cloudream/common/consts/errorcode"
@@ -10,7 +11,6 @@ import (
 	"gitlink.org.cn/cloudream/common/pkgs/logger"
 	cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
 
-	"gitlink.org.cn/cloudream/storage/common/pkgs/db/model"
 	stgiter "gitlink.org.cn/cloudream/storage/common/pkgs/iterator"
 )
 
@@ -24,44 +24,56 @@ func (s *Server) Package() *PackageService {
 	}
 }
 
-type PackageGetReq struct {
-	UserID    *cdssdk.UserID    `form:"userID" binding:"required"`
-	PackageID *cdssdk.PackageID `form:"packageID" binding:"required"`
-}
-type PackageGetResp struct {
-	model.Package
-}
-
 func (s *PackageService) Get(ctx *gin.Context) {
 	log := logger.WithField("HTTP", "Package.Get")
 
-	var req PackageGetReq
+	var req cdssdk.PackageGetReq
 	if err := ctx.ShouldBindQuery(&req); err != nil {
 		log.Warnf("binding body: %s", err.Error())
 		ctx.JSON(http.StatusBadRequest, Failed(errorcode.BadArgument, "missing argument or invalid argument"))
 		return
 	}
 
-	pkg, err := s.svc.PackageSvc().Get(*req.UserID, *req.PackageID)
+	pkg, err := s.svc.PackageSvc().Get(req.UserID, req.PackageID)
 	if err != nil {
 		log.Warnf("getting package: %s", err.Error())
 		ctx.JSON(http.StatusOK, Failed(errorcode.OperationFailed, "get package failed"))
 		return
 	}
 
-	ctx.JSON(http.StatusOK, OK(PackageGetResp{Package: *pkg}))
+	ctx.JSON(http.StatusOK, OK(cdssdk.PackageGetResp{Package: *pkg}))
+}
+
+func (s *PackageService) GetByName(ctx *gin.Context) {
+	log := logger.WithField("HTTP", "Package.GetByName")
+
+	var req cdssdk.PackageGetByName
+	if err := ctx.ShouldBindQuery(&req); err != nil {
+		log.Warnf("binding query: %s", err.Error())
+		ctx.JSON(http.StatusBadRequest, Failed(errorcode.BadArgument, "missing argument or invalid argument"))
+		return
+	}
+
+	pkg, err := s.svc.PackageSvc().GetByName(req.UserID, req.BucketName, req.PackageName)
+	if err != nil {
+		log.Warnf("getting package by name: %s", err.Error())
+		ctx.JSON(http.StatusOK, Failed(errorcode.OperationFailed, "get package by name failed"))
+		return
+	}
+
+	ctx.JSON(http.StatusOK, OK(cdssdk.PackageGetByNameResp{Package: *pkg}))
 }
 
 func (s *PackageService) Create(ctx *gin.Context) {
 	log := logger.WithField("HTTP", "Package.Create")
-	var req cdssdk.PackageCreateReq
+	var req cdssdk.PackageCreate
 	if err := ctx.ShouldBindJSON(&req); err != nil {
 		log.Warnf("binding body: %s", err.Error())
 		ctx.JSON(http.StatusBadRequest, Failed(errorcode.BadArgument, "missing argument or invalid argument"))
 		return
 	}
 
-	pkgID, err := s.svc.PackageSvc().Create(req.UserID, req.BucketID, req.Name)
+	pkg, err := s.svc.PackageSvc().Create(req.UserID, req.BucketID, req.Name)
 	if err != nil {
 		log.Warnf("creating package: %s", err.Error())
 		ctx.JSON(http.StatusOK, Failed(errorcode.OperationFailed, "create package failed"))
@@ -69,26 +81,21 @@ func (s *PackageService) Create(ctx *gin.Context) {
 	}
 
 	ctx.JSON(http.StatusOK, OK(cdssdk.PackageCreateResp{
-		PackageID: pkgID,
+		Package: pkg,
 	}))
 }
 
-type PackageDeleteReq struct {
-	UserID    *cdssdk.UserID    `json:"userID" binding:"required"`
-	PackageID *cdssdk.PackageID `json:"packageID" binding:"required"`
-}
-
 func (s *PackageService) Delete(ctx *gin.Context) {
 	log := logger.WithField("HTTP", "Package.Delete")
 
-	var req PackageDeleteReq
+	var req cdssdk.PackageDelete
 	if err := ctx.ShouldBindJSON(&req); err != nil {
 		log.Warnf("binding body: %s", err.Error())
 		ctx.JSON(http.StatusBadRequest, Failed(errorcode.BadArgument, "missing argument or invalid argument"))
 		return
 	}
 
-	err := s.svc.PackageSvc().DeletePackage(*req.UserID, *req.PackageID)
+	err := s.svc.PackageSvc().DeletePackage(req.UserID, req.PackageID)
 	if err != nil {
 		log.Warnf("deleting package: %s", err.Error())
 		ctx.JSON(http.StatusOK, Failed(errorcode.OperationFailed, "delete package failed"))
@@ -98,61 +105,66 @@ func (s *PackageService) Delete(ctx *gin.Context) {
 	ctx.JSON(http.StatusOK, OK(nil))
 }
 
-type GetCachedNodesReq struct {
-	UserID    *cdssdk.UserID    `json:"userID" binding:"required"`
-	PackageID *cdssdk.PackageID `json:"packageID" binding:"required"`
-}
-type GetCachedNodesResp struct {
-	cdssdk.PackageCachingInfo
+func (s *PackageService) ListBucketPackages(ctx *gin.Context) {
+	log := logger.WithField("HTTP", "Package.ListBucketPackages")
+
+	var req cdssdk.PackageListBucketPackages
+	if err := ctx.ShouldBindQuery(&req); err != nil {
+		log.Warnf("binding query: %s", err.Error())
+		ctx.JSON(http.StatusBadRequest, Failed(errorcode.BadArgument, "missing argument or invalid argument"))
+		return
+	}
+
+	pkgs, err := s.svc.PackageSvc().GetBucketPackages(req.UserID, req.BucketID)
+	if err != nil {
+		log.Warnf("getting bucket packages: %s", err.Error())
+		ctx.JSON(http.StatusOK, Failed(errorcode.OperationFailed, "get bucket packages failed"))
+		return
+	}
+
+	ctx.JSON(http.StatusOK, OK(cdssdk.PackageListBucketPackagesResp{
+		Packages: pkgs,
+	}))
 }
 
 func (s *PackageService) GetCachedNodes(ctx *gin.Context) {
 	log := logger.WithField("HTTP", "Package.GetCachedNodes")
 
-	var req GetCachedNodesReq
-	if err := ctx.ShouldBindJSON(&req); err != nil {
-		log.Warnf("binding body: %s", err.Error())
+	var req cdssdk.PackageGetCachedNodesReq
+	if err := ctx.ShouldBindQuery(&req); err != nil {
+		log.Warnf("binding query: %s", err.Error())
 		ctx.JSON(http.StatusBadRequest, Failed(errorcode.BadArgument, "missing argument or invalid argument"))
 		return
 	}
 
-	resp, err := s.svc.PackageSvc().GetCachedNodes(*req.UserID, *req.PackageID)
+	resp, err := s.svc.PackageSvc().GetCachedNodes(req.UserID, req.PackageID)
 	if err != nil {
 		log.Warnf("get package cached nodes failed: %s", err.Error())
 		ctx.JSON(http.StatusOK, Failed(errorcode.OperationFailed, "get package cached nodes failed"))
 		return
 	}
 
-	ctx.JSON(http.StatusOK, OK(GetCachedNodesResp{resp}))
-}
-
-type GetLoadedNodesReq struct {
-	UserID    *cdssdk.UserID    `json:"userID" binding:"required"`
-	PackageID *cdssdk.PackageID `json:"packageID" binding:"required"`
-}
-
-type GetLoadedNodesResp struct {
-	NodeIDs []cdssdk.NodeID `json:"nodeIDs"`
+	ctx.JSON(http.StatusOK, OK(cdssdk.PackageGetCachedNodesResp{PackageCachingInfo: resp}))
 }
 
 func (s *PackageService) GetLoadedNodes(ctx *gin.Context) {
 	log := logger.WithField("HTTP", "Package.GetLoadedNodes")
 
-	var req GetLoadedNodesReq
-	if err := ctx.ShouldBindJSON(&req); err != nil {
-		log.Warnf("binding body: %s", err.Error())
+	var req cdssdk.PackageGetLoadedNodesReq
+	if err := ctx.ShouldBindQuery(&req); err != nil {
+		log.Warnf("binding query: %s", err.Error())
 		ctx.JSON(http.StatusBadRequest, Failed(errorcode.BadArgument, "missing argument or invalid argument"))
 		return
 	}
 
-	nodeIDs, err := s.svc.PackageSvc().GetLoadedNodes(*req.UserID, *req.PackageID)
+	nodeIDs, err := s.svc.PackageSvc().GetLoadedNodes(req.UserID, req.PackageID)
 	if err != nil {
 		log.Warnf("get package loaded nodes failed: %s", err.Error())
 		ctx.JSON(http.StatusOK, Failed(errorcode.OperationFailed, "get package loaded nodes failed"))
 		return
 	}
 
-	ctx.JSON(http.StatusOK, OK(GetLoadedNodesResp{
+	ctx.JSON(http.StatusOK, OK(cdssdk.PackageGetLoadedNodesResp{
 		NodeIDs: nodeIDs,
 	}))
 }
@@ -166,8 +178,13 @@ func mapMultiPartFileToUploadingObject(files []*multipart.FileHeader) stgiter.Up
 				return nil, err
 			}
 
+			fileName, err := url.PathUnescape(file.Filename)
+			if err != nil {
+				return nil, err
+			}
+
 			return &stgiter.IterUploadingObject{
-				Path: file.Filename,
+				Path: fileName,
 				Size: file.Size,
 				File: stream,
 			}, nil

client/internal/http/server.go

@@ -42,12 +42,17 @@ func (s *Server) initRouters() {
 	s.engine.GET(cdssdk.ObjectDownloadPath, s.Object().Download)
 	s.engine.POST(cdssdk.ObjectUploadPath, s.Object().Upload)
 	s.engine.GET(cdssdk.ObjectGetPackageObjectsPath, s.Object().GetPackageObjects)
+	s.engine.POST(cdssdk.ObjectUpdateInfoPath, s.Object().UpdateInfo)
+	s.engine.POST(cdssdk.ObjectMovePath, s.Object().Move)
+	s.engine.POST(cdssdk.ObjectDeletePath, s.Object().Delete)
 
 	s.engine.GET(cdssdk.PackageGetPath, s.Package().Get)
+	s.engine.GET(cdssdk.PackageGetByNamePath, s.Package().GetByName)
 	s.engine.POST(cdssdk.PackageCreatePath, s.Package().Create)
-	s.engine.POST("/package/delete", s.Package().Delete)
-	s.engine.GET("/package/getCachedNodes", s.Package().GetCachedNodes)
-	s.engine.GET("/package/getLoadedNodes", s.Package().GetLoadedNodes)
+	s.engine.POST(cdssdk.PackageDeletePath, s.Package().Delete)
+	s.engine.GET(cdssdk.PackageListBucketPackagesPath, s.Package().ListBucketPackages)
+	s.engine.GET(cdssdk.PackageGetCachedNodesPath, s.Package().GetCachedNodes)
+	s.engine.GET(cdssdk.PackageGetLoadedNodesPath, s.Package().GetLoadedNodes)
 
 	s.engine.POST("/storage/loadPackage", s.Storage().LoadPackage)
 	s.engine.POST("/storage/createPackage", s.Storage().CreatePackage)
@@ -55,6 +60,8 @@ func (s *Server) initRouters() {
 
 	s.engine.POST(cdssdk.CacheMovePackagePath, s.Cache().MovePackage)
 
+	s.engine.GET(cdssdk.BucketGetByNamePath, s.Bucket().GetByName)
 	s.engine.POST(cdssdk.BucketCreatePath, s.Bucket().Create)
 	s.engine.POST(cdssdk.BucketDeletePath, s.Bucket().Delete)
+	s.engine.GET(cdssdk.BucketListUserBucketsPath, s.Bucket().ListUserBuckets)
 }

client/internal/services/bucket.go

@@ -22,6 +22,21 @@ func (svc *BucketService) GetBucket(userID cdssdk.UserID, bucketID cdssdk.Bucket
 	panic("not implement yet")
 }
 
+func (svc *BucketService) GetBucketByName(userID cdssdk.UserID, bucketName string) (model.Bucket, error) {
+	coorCli, err := stgglb.CoordinatorMQPool.Acquire()
+	if err != nil {
+		return model.Bucket{}, fmt.Errorf("new coordinator client: %w", err)
+	}
+	defer stgglb.CoordinatorMQPool.Release(coorCli)
+
+	resp, err := coorCli.GetBucketByName(coormq.ReqGetBucketByName(userID, bucketName))
+	if err != nil {
+		return model.Bucket{}, fmt.Errorf("get bucket by name failed, err: %w", err)
+	}
+
+	return resp.Bucket, nil
+}
+
 func (svc *BucketService) GetUserBuckets(userID cdssdk.UserID) ([]model.Bucket, error) {
 	coorCli, err := stgglb.CoordinatorMQPool.Acquire()
 	if err != nil {
@@ -52,19 +67,19 @@ func (svc *BucketService) GetBucketPackages(userID cdssdk.UserID, bucketID cdssd
 	return resp.Packages, nil
 }
 
-func (svc *BucketService) CreateBucket(userID cdssdk.UserID, bucketName string) (cdssdk.BucketID, error) {
+func (svc *BucketService) CreateBucket(userID cdssdk.UserID, bucketName string) (cdssdk.Bucket, error) {
 	coorCli, err := stgglb.CoordinatorMQPool.Acquire()
 	if err != nil {
-		return 0, fmt.Errorf("new coordinator client: %w", err)
+		return cdssdk.Bucket{}, fmt.Errorf("new coordinator client: %w", err)
 	}
 	defer stgglb.CoordinatorMQPool.Release(coorCli)
 
 	resp, err := coorCli.CreateBucket(coormq.NewCreateBucket(userID, bucketName))
 	if err != nil {
-		return 0, fmt.Errorf("creating bucket: %w", err)
+		return cdssdk.Bucket{}, fmt.Errorf("creating bucket: %w", err)
 	}
 
-	return resp.BucketID, nil
+	return resp.Bucket, nil
 }
 
 func (svc *BucketService) DeleteBucket(userID cdssdk.UserID, bucketID cdssdk.BucketID) error {
@@ -74,8 +89,6 @@ func (svc *BucketService) DeleteBucket(userID cdssdk.UserID, bucketID cdssdk.Buc
 	}
 	defer stgglb.CoordinatorMQPool.Release(coorCli)
 
-	// TODO 检查用户是否有删除这个Bucket的权限。检查的时候可以只上UserBucket的Read锁
-
 	_, err = coorCli.DeleteBucket(coormq.NewDeleteBucket(userID, bucketID))
 	if err != nil {
 		return fmt.Errorf("request to coordinator failed, err: %w", err)

client/internal/services/object.go

@@ -2,12 +2,12 @@ package services
 
 import (
 	"fmt"
-	"io"
 	"time"
 
 	cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
 	mytask "gitlink.org.cn/cloudream/storage/client/internal/task"
 	stgglb "gitlink.org.cn/cloudream/storage/common/globals"
+	stgmod "gitlink.org.cn/cloudream/storage/common/models"
 	"gitlink.org.cn/cloudream/storage/common/pkgs/db/model"
 	"gitlink.org.cn/cloudream/storage/common/pkgs/iterator"
 	coormq "gitlink.org.cn/cloudream/storage/common/pkgs/mq/coordinator"
@@ -35,8 +35,78 @@ func (svc *ObjectService) WaitUploading(taskID string, waitTimeout time.Duration
 	return false, nil, nil
 }
 
-func (svc *ObjectService) Download(userID cdssdk.UserID, objectID cdssdk.ObjectID) (io.ReadCloser, error) {
-	panic("not implement yet!")
+func (svc *ObjectService) UpdateInfo(userID cdssdk.UserID, updatings []cdssdk.UpdatingObject) ([]cdssdk.ObjectID, error) {
+	coorCli, err := stgglb.CoordinatorMQPool.Acquire()
+	if err != nil {
+		return nil, fmt.Errorf("new coordinator client: %w", err)
+	}
+	defer stgglb.CoordinatorMQPool.Release(coorCli)
+
+	resp, err := coorCli.UpdateObjectInfos(coormq.ReqUpdateObjectInfos(userID, updatings))
+	if err != nil {
+		return nil, fmt.Errorf("requsting to coodinator: %w", err)
+	}
+
+	return resp.Successes, nil
+}
+
+func (svc *ObjectService) Move(userID cdssdk.UserID, movings []cdssdk.MovingObject) ([]cdssdk.ObjectID, error) {
+	coorCli, err := stgglb.CoordinatorMQPool.Acquire()
+	if err != nil {
+		return nil, fmt.Errorf("new coordinator client: %w", err)
+	}
+	defer stgglb.CoordinatorMQPool.Release(coorCli)
+
+	resp, err := coorCli.MoveObjects(coormq.ReqMoveObjects(userID, movings))
+	if err != nil {
+		return nil, fmt.Errorf("requsting to coodinator: %w", err)
+	}
+
+	return resp.Successes, nil
+}
+
+func (svc *ObjectService) Download(userID cdssdk.UserID, objectID cdssdk.ObjectID) (*iterator.IterDownloadingObject, error) {
+	coorCli, err := stgglb.CoordinatorMQPool.Acquire()
+	if err != nil {
+		return nil, fmt.Errorf("new coordinator client: %w", err)
+	}
+	defer stgglb.CoordinatorMQPool.Release(coorCli)
+
+	resp, err := coorCli.GetObjectDetails(coormq.ReqGetObjectDetails([]cdssdk.ObjectID{objectID}))
+	if err != nil {
+		return nil, fmt.Errorf("requesting to coordinator")
+	}
+
+	if resp.Objects[0] == nil {
+		return nil, fmt.Errorf("object not found")
+	}
+
+	iter := iterator.NewDownloadObjectIterator([]stgmod.ObjectDetail{*resp.Objects[0]}, &iterator.DownloadContext{
+		Distlock: svc.DistLock,
+	})
+	defer iter.Close()
+
+	downloading, err := iter.MoveNext()
+	if err != nil {
+		return nil, err
+	}
+
+	return downloading, nil
+}
+
+func (svc *ObjectService) Delete(userID cdssdk.UserID, objectIDs []cdssdk.ObjectID) error {
+	coorCli, err := stgglb.CoordinatorMQPool.Acquire()
+	if err != nil {
+		return fmt.Errorf("new coordinator client: %w", err)
+	}
+	defer stgglb.CoordinatorMQPool.Release(coorCli)
+
+	_, err = coorCli.DeleteObjects(coormq.ReqDeleteObjects(userID, objectIDs))
+	if err != nil {
+		return fmt.Errorf("requsting to coodinator: %w", err)
+	}
+
+	return nil
 }
 
 func (svc *ObjectService) GetPackageObjects(userID cdssdk.UserID, packageID cdssdk.PackageID) ([]model.Object, error) {

client/internal/services/package.go

@@ -6,7 +6,6 @@ import (
 	cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
 
 	stgglb "gitlink.org.cn/cloudream/storage/common/globals"
-	"gitlink.org.cn/cloudream/storage/common/pkgs/db/model"
 	"gitlink.org.cn/cloudream/storage/common/pkgs/iterator"
 	coormq "gitlink.org.cn/cloudream/storage/common/pkgs/mq/coordinator"
 )
@@ -19,7 +18,7 @@ func (svc *Service) PackageSvc() *PackageService {
 	return &PackageService{Service: svc}
 }
 
-func (svc *PackageService) Get(userID cdssdk.UserID, packageID cdssdk.PackageID) (*model.Package, error) {
+func (svc *PackageService) Get(userID cdssdk.UserID, packageID cdssdk.PackageID) (*cdssdk.Package, error) {
 	coorCli, err := stgglb.CoordinatorMQPool.Acquire()
 	if err != nil {
 		return nil, fmt.Errorf("new coordinator client: %w", err)
@@ -34,19 +33,49 @@ func (svc *PackageService) Get(userID cdssdk.UserID, packageID cdssdk.PackageID)
 	return &getResp.Package, nil
 }
 
-func (svc *PackageService) Create(userID cdssdk.UserID, bucketID cdssdk.BucketID, name string) (cdssdk.PackageID, error) {
+func (svc *PackageService) GetByName(userID cdssdk.UserID, bucketName string, packageName string) (*cdssdk.Package, error) {
 	coorCli, err := stgglb.CoordinatorMQPool.Acquire()
 	if err != nil {
-		return 0, fmt.Errorf("new coordinator client: %w", err)
+		return nil, fmt.Errorf("new coordinator client: %w", err)
+	}
+	defer stgglb.CoordinatorMQPool.Release(coorCli)
+
+	getResp, err := coorCli.GetPackageByName(coormq.ReqGetPackageByName(userID, bucketName, packageName))
+	if err != nil {
+		return nil, fmt.Errorf("requsting to coodinator: %w", err)
+	}
+
+	return &getResp.Package, nil
+}
+
+func (svc *PackageService) GetBucketPackages(userID cdssdk.UserID, bucketID cdssdk.BucketID) ([]cdssdk.Package, error) {
+	coorCli, err := stgglb.CoordinatorMQPool.Acquire()
+	if err != nil {
+		return nil, fmt.Errorf("new coordinator client: %w", err)
+	}
+	defer stgglb.CoordinatorMQPool.Release(coorCli)
+
+	getResp, err := coorCli.GetBucketPackages(coormq.NewGetBucketPackages(userID, bucketID))
+	if err != nil {
+		return nil, fmt.Errorf("requsting to coodinator: %w", err)
+	}
+
+	return getResp.Packages, nil
+}
+
+func (svc *PackageService) Create(userID cdssdk.UserID, bucketID cdssdk.BucketID, name string) (cdssdk.Package, error) {
+	coorCli, err := stgglb.CoordinatorMQPool.Acquire()
+	if err != nil {
+		return cdssdk.Package{}, fmt.Errorf("new coordinator client: %w", err)
 	}
 	defer stgglb.CoordinatorMQPool.Release(coorCli)
 
 	resp, err := coorCli.CreatePackage(coormq.NewCreatePackage(userID, bucketID, name))
 	if err != nil {
-		return 0, fmt.Errorf("creating package: %w", err)
+		return cdssdk.Package{}, fmt.Errorf("creating package: %w", err)
 	}
 
-	return resp.PackageID, nil
+	return resp.Package, nil
 }
 
 func (svc *PackageService) DownloadPackage(userID cdssdk.UserID, packageID cdssdk.PackageID) (iterator.DownloadingObjectIterator, error) {

common/pkgs/cmd/upload_objects.go

@@ -34,10 +34,9 @@ type UploadObjectsResult struct {
 }
 
 type ObjectUploadResult struct {
-	Info  *iterator.IterUploadingObject
-	Error error
-	// TODO 这个字段没有被赋值
-	ObjectID cdssdk.ObjectID
+	Info   *iterator.IterUploadingObject
+	Error  error
+	Object cdssdk.Object
 }
 
 type UploadNodeInfo struct {
@@ -189,11 +188,26 @@ func uploadAndUpdatePackage(packageID cdssdk.PackageID, objectIter iterator.Uplo
 		}
 	}
 
-	_, err = coorCli.UpdatePackage(coormq.NewUpdatePackage(packageID, adds, nil))
+	updateResp, err := coorCli.UpdatePackage(coormq.NewUpdatePackage(packageID, adds, nil))
 	if err != nil {
 		return nil, fmt.Errorf("updating package: %w", err)
 	}
 
+	updatedObjs := make(map[string]*cdssdk.Object)
+	for _, obj := range updateResp.Added {
+		o := obj
+		updatedObjs[obj.Path] = &o
+	}
+
+	for i := range uploadRets {
+		obj := updatedObjs[uploadRets[i].Info.Path]
+		if obj == nil {
+			uploadRets[i].Error = fmt.Errorf("object %s not found in package", uploadRets[i].Info.Path)
+			continue
+		}
+		uploadRets[i].Object = *obj
+	}
+
 	return uploadRets, nil
 }
 

common/pkgs/db/bucket.go

@@ -18,6 +18,12 @@ func (db *DB) Bucket() *BucketDB {
 	return &BucketDB{DB: db}
 }
 
+func (db *BucketDB) GetByID(ctx SQLContext, bucketID cdssdk.BucketID) (cdssdk.Bucket, error) {
+	var ret cdssdk.Bucket
+	err := sqlx.Get(ctx, &ret, "select * from Bucket where BucketID = ?", bucketID)
+	return ret, err
+}
+
 // GetIDByName 根据BucketName查询BucketID
 func (db *BucketDB) GetIDByName(bucketName string) (int64, error) {
 	//桶结构体
@@ -57,6 +63,15 @@ func (*BucketDB) GetUserBucket(ctx SQLContext, userID cdssdk.UserID, bucketID cd
 	return ret, err
 }
 
+func (*BucketDB) GetUserBucketByName(ctx SQLContext, userID cdssdk.UserID, bucketName string) (model.Bucket, error) {
+	var ret model.Bucket
+	err := sqlx.Get(ctx, &ret,
+		"select Bucket.* from UserBucket, Bucket where UserID = ? and"+
+			" UserBucket.BucketID = Bucket.BucketID and"+
+			" Bucket.Name = ?", userID, bucketName)
+	return ret, err
+}
+
 func (*BucketDB) GetUserBuckets(ctx SQLContext, userID cdssdk.UserID) ([]model.Bucket, error) {
 	var ret []model.Bucket
 	err := sqlx.Select(ctx, &ret, "select Bucket.* from UserBucket, Bucket where UserID = ? and UserBucket.BucketID = Bucket.BucketID", userID)

common/pkgs/db/model/model.go

@@ -22,7 +22,7 @@ type Storage struct {
 
 type User struct {
 	UserID   cdssdk.UserID `db:"UserID" json:"userID"`
-	Password string        `db:"PassWord" json:"password"`
+	Password string        `db:"Password" json:"password"`
 }
 
 type UserBucket struct {

common/pkgs/db/object.go

@@ -26,25 +26,46 @@ func (db *ObjectDB) GetByID(ctx SQLContext, objectID cdssdk.ObjectID) (model.Obj
 	return ret.ToObject(), err
 }
 
-func (db *ObjectDB) BatchGetPackageObjectIDs(ctx SQLContext, pkgID cdssdk.PackageID, pathes []string) ([]cdssdk.ObjectID, error) {
+func (db *ObjectDB) BatchGet(ctx SQLContext, objectIDs []cdssdk.ObjectID) ([]model.Object, error) {
+	if len(objectIDs) == 0 {
+		return nil, nil
+	}
+
+	// TODO In语句
+	stmt, args, err := sqlx.In("select * from Object where ObjectID in (?) order by ObjectID asc", objectIDs)
+	if err != nil {
+		return nil, err
+	}
+	stmt = ctx.Rebind(stmt)
+
+	objs := make([]model.TempObject, 0, len(objectIDs))
+	err = sqlx.Select(ctx, &objs, stmt, args...)
+	if err != nil {
+		return nil, err
+	}
+
+	return lo.Map(objs, func(o model.TempObject, idx int) cdssdk.Object { return o.ToObject() }), nil
+}
+
+func (db *ObjectDB) BatchGetByPackagePath(ctx SQLContext, pkgID cdssdk.PackageID, pathes []string) ([]cdssdk.Object, error) {
 	if len(pathes) == 0 {
 		return nil, nil
 	}
 
 	// TODO In语句
-	stmt, args, err := sqlx.In("select ObjectID from Object force index(PackagePath) where PackageID=? and Path in (?)", pkgID, pathes)
+	stmt, args, err := sqlx.In("select * from Object force index(PackagePath) where PackageID=? and Path in (?)", pkgID, pathes)
 	if err != nil {
 		return nil, err
 	}
 	stmt = ctx.Rebind(stmt)
 
-	objIDs := make([]cdssdk.ObjectID, 0, len(pathes))
-	err = sqlx.Select(ctx, &objIDs, stmt, args...)
+	objs := make([]model.TempObject, 0, len(pathes))
+	err = sqlx.Select(ctx, &objs, stmt, args...)
 	if err != nil {
 		return nil, err
 	}
 
-	return objIDs, nil
+	return lo.Map(objs, func(o model.TempObject, idx int) cdssdk.Object { return o.ToObject() }), nil
 }
 
 func (db *ObjectDB) Create(ctx SQLContext, obj cdssdk.Object) (cdssdk.ObjectID, error) {
@@ -63,10 +84,10 @@ func (db *ObjectDB) Create(ctx SQLContext, obj cdssdk.Object) (cdssdk.ObjectID,
 	return cdssdk.ObjectID(objectID), nil
 }
 
-// 可以用于批量创建或者更新记录
-// 用于创建时，需要额外检查PackageID+Path的唯一性
+// 可以用于批量创建或者更新记录。
+// 用于创建时，需要额外检查PackageID+Path的唯一性。
 // 用于更新时，需要额外检查现存的PackageID+Path对应的ObjectID是否与待更新的ObjectID相同。不会更新CreateTime。
-func (db *ObjectDB) BatchCreateOrUpdate(ctx SQLContext, objs []cdssdk.Object) error {
+func (db *ObjectDB) BatchUpsertByPackagePath(ctx SQLContext, objs []cdssdk.Object) error {
 	if len(objs) == 0 {
 		return nil
 	}
@@ -78,6 +99,18 @@ func (db *ObjectDB) BatchCreateOrUpdate(ctx SQLContext, objs []cdssdk.Object) er
 	return BatchNamedExec(ctx, sql, 7, objs, nil)
 }
 
+func (db *ObjectDB) BatchUpert(ctx SQLContext, objs []cdssdk.Object) error {
+	if len(objs) == 0 {
+		return nil
+	}
+
+	sql := "insert into Object(ObjectID, PackageID, Path, Size, FileHash, Redundancy, CreateTime ,UpdateTime)" +
+		" values(:ObjectID, :PackageID,:Path,:Size,:FileHash,:Redundancy, :CreateTime, :UpdateTime) as new" +
+		" on duplicate key update PackageID = new.PackageID, Path = new.Path, Size = new.Size, FileHash = new.FileHash, Redundancy = new.Redundancy, UpdateTime = new.UpdateTime"
+
+	return BatchNamedExec(ctx, sql, 8, objs, nil)
+}
+
 func (*ObjectDB) GetPackageObjects(ctx SQLContext, packageID cdssdk.PackageID) ([]model.Object, error) {
 	var ret []model.TempObject
 	err := sqlx.Select(ctx, &ret, "select * from Object where PackageID = ? order by ObjectID asc", packageID)
@@ -136,7 +169,7 @@ func (db *ObjectDB) GetPackageObjectDetails(ctx SQLContext, packageID cdssdk.Pac
 	return rets, nil
 }
 
-func (db *ObjectDB) BatchAdd(ctx SQLContext, packageID cdssdk.PackageID, adds []coormq.AddObjectEntry) ([]cdssdk.ObjectID, error) {
+func (db *ObjectDB) BatchAdd(ctx SQLContext, packageID cdssdk.PackageID, adds []coormq.AddObjectEntry) ([]cdssdk.Object, error) {
 	if len(adds) == 0 {
 		return nil, nil
 	}
@@ -154,7 +187,7 @@ func (db *ObjectDB) BatchAdd(ctx SQLContext, packageID cdssdk.PackageID, adds []
 		})
 	}
 
-	err := db.BatchCreateOrUpdate(ctx, objs)
+	err := db.BatchUpsertByPackagePath(ctx, objs)
 	if err != nil {
 		return nil, fmt.Errorf("batch create or update objects: %w", err)
 	}
@@ -163,17 +196,22 @@ func (db *ObjectDB) BatchAdd(ctx SQLContext, packageID cdssdk.PackageID, adds []
 	for _, add := range adds {
 		pathes = append(pathes, add.Path)
 	}
-	objIDs, err := db.BatchGetPackageObjectIDs(ctx, packageID, pathes)
+	// 这里可以不用检查查询结果是否与pathes的数量相同
+	addedObjs, err := db.BatchGetByPackagePath(ctx, packageID, pathes)
 	if err != nil {
 		return nil, fmt.Errorf("batch get object ids: %w", err)
 	}
+	addedObjIDs := make([]cdssdk.ObjectID, len(addedObjs))
+	for i := range addedObjs {
+		addedObjIDs[i] = addedObjs[i].ObjectID
+	}
 
-	err = db.ObjectBlock().BatchDeleteByObjectID(ctx, objIDs)
+	err = db.ObjectBlock().BatchDeleteByObjectID(ctx, addedObjIDs)
 	if err != nil {
 		return nil, fmt.Errorf("batch delete object blocks: %w", err)
 	}
 
-	err = db.PinnedObject().BatchDeleteByObjectID(ctx, objIDs)
+	err = db.PinnedObject().BatchDeleteByObjectID(ctx, addedObjIDs)
 	if err != nil {
 		return nil, fmt.Errorf("batch delete pinned objects: %w", err)
 	}
@@ -181,7 +219,7 @@ func (db *ObjectDB) BatchAdd(ctx SQLContext, packageID cdssdk.PackageID, adds []
 	objBlocks := make([]stgmod.ObjectBlock, 0, len(adds))
 	for i, add := range adds {
 		objBlocks = append(objBlocks, stgmod.ObjectBlock{
-			ObjectID: objIDs[i],
+			ObjectID: addedObjIDs[i],
 			Index:    0,
 			NodeID:   add.NodeID,
 			FileHash: add.FileHash,
@@ -206,10 +244,10 @@ func (db *ObjectDB) BatchAdd(ctx SQLContext, packageID cdssdk.PackageID, adds []
 		return nil, fmt.Errorf("batch create caches: %w", err)
 	}
 
-	return objIDs, nil
+	return addedObjs, nil
 }
 
-func (db *ObjectDB) BatchUpdateRedundancy(ctx SQLContext, objs []coormq.ChangeObjectRedundancyEntry) error {
+func (db *ObjectDB) BatchUpdateRedundancy(ctx SQLContext, objs []coormq.UpdatingObjectRedundancy) error {
 	if len(objs) == 0 {
 		return nil
 	}

common/pkgs/db/object_block.go

@@ -24,6 +24,26 @@ func (db *ObjectBlockDB) GetByNodeID(ctx SQLContext, nodeID cdssdk.NodeID) ([]st
 	return rets, err
 }
 
+func (db *ObjectBlockDB) BatchGetByObjectID(ctx SQLContext, objectIDs []cdssdk.ObjectID) ([]stgmod.ObjectBlock, error) {
+	if len(objectIDs) == 0 {
+		return nil, nil
+	}
+
+	stmt, args, err := sqlx.In("select * from ObjectBlock where ObjectID in (?) order by ObjectID, `Index` asc", objectIDs)
+	if err != nil {
+		return nil, err
+	}
+	stmt = ctx.Rebind(stmt)
+
+	var blocks []stgmod.ObjectBlock
+	err = sqlx.Select(ctx, &blocks, stmt, args...)
+	if err != nil {
+		return nil, err
+	}
+
+	return blocks, nil
+}
+
 func (db *ObjectBlockDB) Create(ctx SQLContext, objectID cdssdk.ObjectID, index int, nodeID cdssdk.NodeID, fileHash string) error {
 	_, err := ctx.Exec("insert into ObjectBlock values(?,?,?,?)", objectID, index, nodeID, fileHash)
 	return err

common/pkgs/db/package.go

@@ -77,6 +77,20 @@ func (db *PackageDB) GetUserPackage(ctx SQLContext, userID cdssdk.UserID, packag
 	return ret, err
 }
 
+// 在指定名称的Bucket中查找指定名称的Package
+func (*PackageDB) GetUserPackageByName(ctx SQLContext, userID cdssdk.UserID, bucketName string, packageName string) (cdssdk.Package, error) {
+	var ret model.Package
+	err := sqlx.Get(ctx, &ret,
+		"select Package.* from Package, Bucket, UserBucket where"+
+			" Package.Name = ? and"+
+			" Package.BucketID = Bucket.BucketID and"+
+			" Bucket.Name = ? and"+
+			" UserBucket.UserID = ? and"+
+			" UserBucket.BucketID = Bucket.BucketID",
+		packageName, bucketName, userID)
+	return ret, err
+}
+
 func (db *PackageDB) Create(ctx SQLContext, bucketID cdssdk.BucketID, name string) (cdssdk.PackageID, error) {
 	// 根据packagename和bucketid查询，若不存在则插入，若存在则返回错误
 	var packageID int64

common/pkgs/db/pinned_object.go

@@ -34,6 +34,26 @@ func (*PinnedObjectDB) Create(ctx SQLContext, nodeID cdssdk.NodeID, objectID cds
 	return err
 }
 
+func (*PinnedObjectDB) BatchGetByObjectID(ctx SQLContext, objectIDs []cdssdk.ObjectID) ([]cdssdk.PinnedObject, error) {
+	if len(objectIDs) == 0 {
+		return nil, nil
+	}
+
+	stmt, args, err := sqlx.In("select * from PinnedObject where ObjectID in (?) order by ObjectID asc", objectIDs)
+	if err != nil {
+		return nil, err
+	}
+	stmt = ctx.Rebind(stmt)
+
+	var pinneds []cdssdk.PinnedObject
+	err = sqlx.Select(ctx, &pinneds, stmt, args...)
+	if err != nil {
+		return nil, err
+	}
+
+	return pinneds, nil
+}
+
 func (*PinnedObjectDB) TryCreate(ctx SQLContext, nodeID cdssdk.NodeID, objectID cdssdk.ObjectID, createTime time.Time) error {
 	_, err := ctx.Exec("insert ignore into PinnedObject values(?,?,?)", nodeID, objectID, createTime)
 	return err

common/pkgs/iterator/download_object_iterator.go

@@ -18,7 +18,6 @@ import (
 	stgglb "gitlink.org.cn/cloudream/storage/common/globals"
 	stgmod "gitlink.org.cn/cloudream/storage/common/models"
 	stgmodels "gitlink.org.cn/cloudream/storage/common/models"
-	"gitlink.org.cn/cloudream/storage/common/pkgs/db/model"
 	"gitlink.org.cn/cloudream/storage/common/pkgs/distlock"
 	"gitlink.org.cn/cloudream/storage/common/pkgs/ec"
 	coormq "gitlink.org.cn/cloudream/storage/common/pkgs/mq/coordinator"
@@ -27,7 +26,7 @@ import (
 type DownloadingObjectIterator = Iterator[*IterDownloadingObject]
 
 type IterDownloadingObject struct {
-	Object model.Object
+	Object cdssdk.Object
 	File   io.ReadCloser
 }
 

common/pkgs/mq/coordinator/bucket.go

@@ -7,6 +7,8 @@ import (
 )
 
 type BucketService interface {
+	GetBucketByName(msg *GetBucketByName) (*GetBucketByNameResp, *mq.CodeMessage)
+
 	GetUserBuckets(msg *GetUserBuckets) (*GetUserBucketsResp, *mq.CodeMessage)
 
 	GetBucketPackages(msg *GetBucketPackages) (*GetBucketPackagesResp, *mq.CodeMessage)
@@ -16,6 +18,34 @@ type BucketService interface {
 	DeleteBucket(msg *DeleteBucket) (*DeleteBucketResp, *mq.CodeMessage)
 }
 
+// 根据桶名获取桶
+var _ = Register(Service.GetBucketByName)
+
+type GetBucketByName struct {
+	mq.MessageBodyBase
+	UserID cdssdk.UserID `json:"userID"`
+	Name   string        `json:"name"`
+}
+type GetBucketByNameResp struct {
+	mq.MessageBodyBase
+	Bucket cdssdk.Bucket `json:"bucket"`
+}
+
+func ReqGetBucketByName(userID cdssdk.UserID, name string) *GetBucketByName {
+	return &GetBucketByName{
+		UserID: userID,
+		Name:   name,
+	}
+}
+func RespGetBucketByName(bucket cdssdk.Bucket) *GetBucketByNameResp {
+	return &GetBucketByNameResp{
+		Bucket: bucket,
+	}
+}
+func (client *Client) GetBucketByName(msg *GetBucketByName) (*GetBucketByNameResp, error) {
+	return mq.Request(Service.GetBucketByName, client.rabbitCli, msg)
+}
+
 // 获取用户所有的桶
 var _ = Register(Service.GetUserBuckets)
 
@@ -80,7 +110,7 @@ type CreateBucket struct {
 }
 type CreateBucketResp struct {
 	mq.MessageBodyBase
-	BucketID cdssdk.BucketID `json:"bucketID"`
+	Bucket cdssdk.Bucket `json:"bucket"`
 }
 
 func NewCreateBucket(userID cdssdk.UserID, bucketName string) *CreateBucket {
@@ -89,9 +119,9 @@ func NewCreateBucket(userID cdssdk.UserID, bucketName string) *CreateBucket {
 		BucketName: bucketName,
 	}
 }
-func NewCreateBucketResp(bucketID cdssdk.BucketID) *CreateBucketResp {
+func NewCreateBucketResp(bucket cdssdk.Bucket) *CreateBucketResp {
 	return &CreateBucketResp{
-		BucketID: bucketID,
+		Bucket: bucket,
 	}
 }
 func (client *Client) CreateBucket(msg *CreateBucket) (*CreateBucketResp, error) {

common/pkgs/mq/coordinator/object.go

@@ -13,7 +13,15 @@ type ObjectService interface {
 
 	GetPackageObjectDetails(msg *GetPackageObjectDetails) (*GetPackageObjectDetailsResp, *mq.CodeMessage)
 
-	ChangeObjectRedundancy(msg *ChangeObjectRedundancy) (*ChangeObjectRedundancyResp, *mq.CodeMessage)
+	GetObjectDetails(msg *GetObjectDetails) (*GetObjectDetailsResp, *mq.CodeMessage)
+
+	UpdateObjectRedundancy(msg *UpdateObjectRedundancy) (*UpdateObjectRedundancyResp, *mq.CodeMessage)
+
+	UpdateObjectInfos(msg *UpdateObjectInfos) (*UpdateObjectInfosResp, *mq.CodeMessage)
+
+	MoveObjects(msg *MoveObjects) (*MoveObjectsResp, *mq.CodeMessage)
+
+	DeleteObjects(msg *DeleteObjects) (*DeleteObjectsResp, *mq.CodeMessage)
 }
 
 // 查询Package中的所有Object，返回的Objects会按照ObjectID升序
@@ -70,31 +78,141 @@ func (client *Client) GetPackageObjectDetails(msg *GetPackageObjectDetails) (*Ge
 	return mq.Request(Service.GetPackageObjectDetails, client.rabbitCli, msg)
 }
 
+// 获取多个Object以及它们的分块详细信息，返回的Objects会按照ObjectID升序。
+var _ = Register(Service.GetObjectDetails)
+
+type GetObjectDetails struct {
+	mq.MessageBodyBase
+	ObjectIDs []cdssdk.ObjectID `json:"objectIDs"`
+}
+type GetObjectDetailsResp struct {
+	mq.MessageBodyBase
+	Objects []*stgmod.ObjectDetail `json:"objects"` // 如果没有查询到某个ID对应的信息，则此数组对应位置为nil
+}
+
+func ReqGetObjectDetails(objectIDs []cdssdk.ObjectID) *GetObjectDetails {
+	return &GetObjectDetails{
+		ObjectIDs: objectIDs,
+	}
+}
+func RespGetObjectDetails(objects []*stgmod.ObjectDetail) *GetObjectDetailsResp {
+	return &GetObjectDetailsResp{
+		Objects: objects,
+	}
+}
+func (client *Client) GetObjectDetails(msg *GetObjectDetails) (*GetObjectDetailsResp, error) {
+	return mq.Request(Service.GetObjectDetails, client.rabbitCli, msg)
+}
+
 // 更新Object的冗余方式
-var _ = Register(Service.ChangeObjectRedundancy)
+var _ = Register(Service.UpdateObjectRedundancy)
 
-type ChangeObjectRedundancy struct {
+type UpdateObjectRedundancy struct {
 	mq.MessageBodyBase
-	Entries []ChangeObjectRedundancyEntry `json:"entries"`
+	Updatings []UpdatingObjectRedundancy `json:"updatings"`
 }
-type ChangeObjectRedundancyResp struct {
+type UpdateObjectRedundancyResp struct {
 	mq.MessageBodyBase
 }
-type ChangeObjectRedundancyEntry struct {
+type UpdatingObjectRedundancy struct {
 	ObjectID   cdssdk.ObjectID      `json:"objectID" db:"ObjectID"`
 	Redundancy cdssdk.Redundancy    `json:"redundancy" db:"Redundancy"`
 	PinnedAt   []cdssdk.NodeID      `json:"pinnedAt"`
 	Blocks     []stgmod.ObjectBlock `json:"blocks"`
 }
 
-func ReqChangeObjectRedundancy(entries []ChangeObjectRedundancyEntry) *ChangeObjectRedundancy {
-	return &ChangeObjectRedundancy{
-		Entries: entries,
+func ReqUpdateObjectRedundancy(updatings []UpdatingObjectRedundancy) *UpdateObjectRedundancy {
+	return &UpdateObjectRedundancy{
+		Updatings: updatings,
+	}
+}
+func RespUpdateObjectRedundancy() *UpdateObjectRedundancyResp {
+	return &UpdateObjectRedundancyResp{}
+}
+func (client *Client) UpdateObjectRedundancy(msg *UpdateObjectRedundancy) (*UpdateObjectRedundancyResp, error) {
+	return mq.Request(Service.UpdateObjectRedundancy, client.rabbitCli, msg)
+}
+
+// 更新Object元数据
+var _ = Register(Service.UpdateObjectInfos)
+
+type UpdateObjectInfos struct {
+	mq.MessageBodyBase
+	UserID    cdssdk.UserID           `json:"userID"`
+	Updatings []cdssdk.UpdatingObject `json:"updatings"`
+}
+
+type UpdateObjectInfosResp struct {
+	mq.MessageBodyBase
+	Successes []cdssdk.ObjectID `json:"successes"`
+}
+
+func ReqUpdateObjectInfos(userID cdssdk.UserID, updatings []cdssdk.UpdatingObject) *UpdateObjectInfos {
+	return &UpdateObjectInfos{
+		UserID:    userID,
+		Updatings: updatings,
+	}
+}
+func RespUpdateObjectInfos(successes []cdssdk.ObjectID) *UpdateObjectInfosResp {
+	return &UpdateObjectInfosResp{
+		Successes: successes,
+	}
+}
+func (client *Client) UpdateObjectInfos(msg *UpdateObjectInfos) (*UpdateObjectInfosResp, error) {
+	return mq.Request(Service.UpdateObjectInfos, client.rabbitCli, msg)
+}
+
+// 移动Object
+var _ = Register(Service.MoveObjects)
+
+type MoveObjects struct {
+	mq.MessageBodyBase
+	UserID  cdssdk.UserID         `json:"userID"`
+	Movings []cdssdk.MovingObject `json:"movings"`
+}
+
+type MoveObjectsResp struct {
+	mq.MessageBodyBase
+	Successes []cdssdk.ObjectID `json:"successes"`
+}
+
+func ReqMoveObjects(userID cdssdk.UserID, movings []cdssdk.MovingObject) *MoveObjects {
+	return &MoveObjects{
+		UserID:  userID,
+		Movings: movings,
+	}
+}
+func RespMoveObjects(successes []cdssdk.ObjectID) *MoveObjectsResp {
+	return &MoveObjectsResp{
+		Successes: successes,
+	}
+}
+func (client *Client) MoveObjects(msg *MoveObjects) (*MoveObjectsResp, error) {
+	return mq.Request(Service.MoveObjects, client.rabbitCli, msg)
+}
+
+// 删除Object
+var _ = Register(Service.DeleteObjects)
+
+type DeleteObjects struct {
+	mq.MessageBodyBase
+	UserID    cdssdk.UserID     `json:"userID"`
+	ObjectIDs []cdssdk.ObjectID `json:"objectIDs"`
+}
+
+type DeleteObjectsResp struct {
+	mq.MessageBodyBase
+}
+
+func ReqDeleteObjects(userID cdssdk.UserID, objectIDs []cdssdk.ObjectID) *DeleteObjects {
+	return &DeleteObjects{
+		UserID:    userID,
+		ObjectIDs: objectIDs,
 	}
 }
-func RespChangeObjectRedundancy() *ChangeObjectRedundancyResp {
-	return &ChangeObjectRedundancyResp{}
+func RespDeleteObjects() *DeleteObjectsResp {
+	return &DeleteObjectsResp{}
 }
-func (client *Client) ChangeObjectRedundancy(msg *ChangeObjectRedundancy) (*ChangeObjectRedundancyResp, error) {
-	return mq.Request(Service.ChangeObjectRedundancy, client.rabbitCli, msg)
+func (client *Client) DeleteObjects(msg *DeleteObjects) (*DeleteObjectsResp, error) {
+	return mq.Request(Service.DeleteObjects, client.rabbitCli, msg)
 }

common/pkgs/mq/coordinator/package.go

@@ -12,6 +12,8 @@ import (
 type PackageService interface {
 	GetPackage(msg *GetPackage) (*GetPackageResp, *mq.CodeMessage)
 
+	GetPackageByName(msg *GetPackageByName) (*GetPackageByNameResp, *mq.CodeMessage)
+
 	CreatePackage(msg *CreatePackage) (*CreatePackageResp, *mq.CodeMessage)
 
 	UpdatePackage(msg *UpdatePackage) (*UpdatePackageResp, *mq.CodeMessage)
@@ -51,6 +53,36 @@ func (client *Client) GetPackage(msg *GetPackage) (*GetPackageResp, error) {
 	return mq.Request(Service.GetPackage, client.rabbitCli, msg)
 }
 
+// 根据名称获取Package
+var _ = Register(Service.GetPackageByName)
+
+type GetPackageByName struct {
+	mq.MessageBodyBase
+	UserID      cdssdk.UserID `json:"userID"`
+	BucketName  string        `json:"bucketName"`
+	PackageName string        `json:"packageName"`
+}
+type GetPackageByNameResp struct {
+	mq.MessageBodyBase
+	Package cdssdk.Package `json:"package"`
+}
+
+func ReqGetPackageByName(userID cdssdk.UserID, bucketName string, packageName string) *GetPackageByName {
+	return &GetPackageByName{
+		UserID:      userID,
+		BucketName:  bucketName,
+		PackageName: packageName,
+	}
+}
+func NewGetPackageByNameResp(pkg cdssdk.Package) *GetPackageByNameResp {
+	return &GetPackageByNameResp{
+		Package: pkg,
+	}
+}
+func (client *Client) GetPackageByName(msg *GetPackageByName) (*GetPackageByNameResp, error) {
+	return mq.Request(Service.GetPackageByName, client.rabbitCli, msg)
+}
+
 // 创建一个Package
 var _ = Register(Service.CreatePackage)
 
@@ -62,7 +94,7 @@ type CreatePackage struct {
 }
 type CreatePackageResp struct {
 	mq.MessageBodyBase
-	PackageID cdssdk.PackageID `json:"packageID"`
+	Package cdssdk.Package `json:"package"`
 }
 
 func NewCreatePackage(userID cdssdk.UserID, bucketID cdssdk.BucketID, name string) *CreatePackage {
@@ -72,9 +104,9 @@ func NewCreatePackage(userID cdssdk.UserID, bucketID cdssdk.BucketID, name strin
 		Name:     name,
 	}
 }
-func NewCreatePackageResp(packageID cdssdk.PackageID) *CreatePackageResp {
+func NewCreatePackageResp(pkg cdssdk.Package) *CreatePackageResp {
 	return &CreatePackageResp{
-		PackageID: packageID,
+		Package: pkg,
 	}
 }
 func (client *Client) CreatePackage(msg *CreatePackage) (*CreatePackageResp, error) {
@@ -92,6 +124,7 @@ type UpdatePackage struct {
 }
 type UpdatePackageResp struct {
 	mq.MessageBodyBase
+	Added []cdssdk.Object `json:"added"`
 }
 type AddObjectEntry struct {
 	Path       string        `json:"path"`
@@ -108,8 +141,10 @@ func NewUpdatePackage(packageID cdssdk.PackageID, adds []AddObjectEntry, deletes
 		Deletes:   deletes,
 	}
 }
-func NewUpdatePackageResp() *UpdatePackageResp {
-	return &UpdatePackageResp{}
+func NewUpdatePackageResp(added []cdssdk.Object) *UpdatePackageResp {
+	return &UpdatePackageResp{
+		Added: added,
+	}
 }
 func NewAddObjectEntry(path string, size int64, fileHash string, uploadTime time.Time, nodeID cdssdk.NodeID) AddObjectEntry {
 	return AddObjectEntry{

coordinator/internal/mq/bucket.go

@@ -18,6 +18,18 @@ func (svc *Service) GetBucket(userID cdssdk.UserID, bucketID cdssdk.BucketID) (m
 	panic("not implement yet")
 }
 
+func (svc *Service) GetBucketByName(msg *coormq.GetBucketByName) (*coormq.GetBucketByNameResp, *mq.CodeMessage) {
+	bucket, err := svc.db.Bucket().GetUserBucketByName(svc.db.SQLCtx(), msg.UserID, msg.Name)
+	if err != nil {
+		logger.WithField("UserID", msg.UserID).
+			WithField("Name", msg.Name).
+			Warnf("getting bucket by name: %s", err.Error())
+		return nil, mq.Failed(errorcode.OperationFailed, "get bucket by name failed")
+	}
+
+	return mq.ReplyOK(coormq.RespGetBucketByName(bucket))
+}
+
 func (svc *Service) GetUserBuckets(msg *coormq.GetUserBuckets) (*coormq.GetUserBucketsResp, *mq.CodeMessage) {
 	buckets, err := svc.db.Bucket().GetUserBuckets(svc.db.SQLCtx(), msg.UserID)
 
@@ -44,18 +56,22 @@ func (svc *Service) GetBucketPackages(msg *coormq.GetBucketPackages) (*coormq.Ge
 }
 
 func (svc *Service) CreateBucket(msg *coormq.CreateBucket) (*coormq.CreateBucketResp, *mq.CodeMessage) {
-	var bucketID cdssdk.BucketID
+	var bucket cdssdk.Bucket
 	err := svc.db.DoTx(sql.LevelSerializable, func(tx *sqlx.Tx) error {
 		_, err := svc.db.User().GetByID(tx, msg.UserID)
 		if err != nil {
 			return fmt.Errorf("getting user by id: %w", err)
 		}
 
-		bucketID, err = svc.db.Bucket().Create(tx, msg.UserID, msg.BucketName)
+		bucketID, err := svc.db.Bucket().Create(tx, msg.UserID, msg.BucketName)
 		if err != nil {
 			return fmt.Errorf("creating bucket: %w", err)
 		}
 
+		bucket, err = svc.db.Bucket().GetByID(tx, bucketID)
+		if err != nil {
+			return fmt.Errorf("getting bucket by id: %w", err)
+		}
 		return nil
 	})
 	if err != nil {
@@ -65,7 +81,7 @@ func (svc *Service) CreateBucket(msg *coormq.CreateBucket) (*coormq.CreateBucket
 		return nil, mq.Failed(errorcode.OperationFailed, "create bucket failed")
 	}
 
-	return mq.ReplyOK(coormq.NewCreateBucketResp(bucketID))
+	return mq.ReplyOK(coormq.NewCreateBucketResp(bucket))
 }
 
 func (svc *Service) DeleteBucket(msg *coormq.DeleteBucket) (*coormq.DeleteBucketResp, *mq.CodeMessage) {

coordinator/internal/mq/object.go

@@ -5,19 +5,34 @@ import (
 	"fmt"
 
 	"github.com/jmoiron/sqlx"
+	"github.com/samber/lo"
 	"gitlink.org.cn/cloudream/common/consts/errorcode"
 	"gitlink.org.cn/cloudream/common/pkgs/logger"
 	"gitlink.org.cn/cloudream/common/pkgs/mq"
+	cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
+	"gitlink.org.cn/cloudream/common/utils/sort2"
 	stgmod "gitlink.org.cn/cloudream/storage/common/models"
 	coormq "gitlink.org.cn/cloudream/storage/common/pkgs/mq/coordinator"
 )
 
 func (svc *Service) GetPackageObjects(msg *coormq.GetPackageObjects) (*coormq.GetPackageObjectsResp, *mq.CodeMessage) {
-	// TODO 检查用户是否有权限
-	objs, err := svc.db.Object().GetPackageObjects(svc.db.SQLCtx(), msg.PackageID)
+	var objs []cdssdk.Object
+	err := svc.db.DoTx(sql.LevelSerializable, func(tx *sqlx.Tx) error {
+		_, err := svc.db.Package().GetUserPackage(tx, msg.UserID, msg.PackageID)
+		if err != nil {
+			return fmt.Errorf("getting package by id: %w", err)
+		}
+
+		objs, err = svc.db.Object().GetPackageObjects(svc.db.SQLCtx(), msg.PackageID)
+		if err != nil {
+			return fmt.Errorf("getting package objects: %w", err)
+		}
+
+		return nil
+	})
 	if err != nil {
-		logger.WithField("PackageID", msg.PackageID).
-			Warnf("get package objects: %s", err.Error())
+		logger.WithField("UserID", msg.UserID).WithField("PackageID", msg.PackageID).
+			Warn(err.Error())
 
 		return nil, mq.Failed(errorcode.OperationFailed, "get package objects failed")
 	}
@@ -51,14 +66,375 @@ func (svc *Service) GetPackageObjectDetails(msg *coormq.GetPackageObjectDetails)
 	return mq.ReplyOK(coormq.NewGetPackageObjectDetailsResp(details))
 }
 
-func (svc *Service) ChangeObjectRedundancy(msg *coormq.ChangeObjectRedundancy) (*coormq.ChangeObjectRedundancyResp, *mq.CodeMessage) {
+func (svc *Service) GetObjectDetails(msg *coormq.GetObjectDetails) (*coormq.GetObjectDetailsResp, *mq.CodeMessage) {
+	details := make([]*stgmod.ObjectDetail, len(msg.ObjectIDs))
+	err := svc.db.DoTx(sql.LevelSerializable, func(tx *sqlx.Tx) error {
+		var err error
+
+		msg.ObjectIDs = sort2.SortAsc(msg.ObjectIDs)
+
+		// 根据ID依次查询Object，ObjectBlock，PinnedObject，并根据升序的特点进行合并
+		objs, err := svc.db.Object().BatchGet(tx, msg.ObjectIDs)
+		if err != nil {
+			return fmt.Errorf("batch get objects: %w", err)
+		}
+
+		objIDIdx := 0
+		objIdx := 0
+		for objIDIdx < len(msg.ObjectIDs) && objIdx < len(objs) {
+			if msg.ObjectIDs[objIDIdx] < objs[objIdx].ObjectID {
+				objIDIdx++
+				continue
+			}
+
+			// 由于是使用msg.ObjectIDs去查询Object，因此不存在msg.ObjectIDs > Object.ObjectID的情况，
+			// 下面同理
+			obj := stgmod.ObjectDetail{
+				Object: objs[objIDIdx],
+			}
+			details[objIDIdx] = &obj
+			objIdx++
+		}
+
+		// 查询合并
+		blocks, err := svc.db.ObjectBlock().BatchGetByObjectID(tx, msg.ObjectIDs)
+		if err != nil {
+			return fmt.Errorf("batch get object blocks: %w", err)
+		}
+
+		objIDIdx = 0
+		blkIdx := 0
+		for objIDIdx < len(msg.ObjectIDs) && blkIdx < len(blocks) {
+			if details[objIDIdx] == nil {
+				objIDIdx++
+				continue
+			}
+
+			if msg.ObjectIDs[objIDIdx] < blocks[blkIdx].ObjectID {
+				objIDIdx++
+				continue
+			}
+
+			details[objIDIdx].Blocks = append(details[objIDIdx].Blocks, blocks[blkIdx])
+			blkIdx++
+		}
+
+		// 查询合并
+		pinneds, err := svc.db.PinnedObject().BatchGetByObjectID(tx, msg.ObjectIDs)
+		if err != nil {
+			return fmt.Errorf("batch get pinned objects: %w", err)
+		}
+
+		objIDIdx = 0
+		pinIdx := 0
+		for objIDIdx < len(msg.ObjectIDs) && pinIdx < len(pinneds) {
+			if details[objIDIdx] == nil {
+				objIDIdx++
+				continue
+			}
+
+			if msg.ObjectIDs[objIDIdx] < pinneds[pinIdx].ObjectID {
+				objIDIdx++
+				continue
+			}
+
+			details[objIDIdx].PinnedAt = append(details[objIDIdx].PinnedAt, pinneds[pinIdx].NodeID)
+			pinIdx++
+		}
+		return nil
+	})
+
+	if err != nil {
+		logger.Warn(err.Error())
+		return nil, mq.Failed(errorcode.OperationFailed, "get object details failed")
+	}
+
+	return mq.ReplyOK(coormq.RespGetObjectDetails(details))
+}
+
+func (svc *Service) UpdateObjectRedundancy(msg *coormq.UpdateObjectRedundancy) (*coormq.UpdateObjectRedundancyResp, *mq.CodeMessage) {
 	err := svc.db.DoTx(sql.LevelSerializable, func(tx *sqlx.Tx) error {
-		return svc.db.Object().BatchUpdateRedundancy(tx, msg.Entries)
+		return svc.db.Object().BatchUpdateRedundancy(tx, msg.Updatings)
 	})
 	if err != nil {
 		logger.Warnf("batch updating redundancy: %s", err.Error())
 		return nil, mq.Failed(errorcode.OperationFailed, "batch update redundancy failed")
 	}
 
-	return mq.ReplyOK(coormq.RespChangeObjectRedundancy())
+	return mq.ReplyOK(coormq.RespUpdateObjectRedundancy())
+}
+
+func (svc *Service) UpdateObjectInfos(msg *coormq.UpdateObjectInfos) (*coormq.UpdateObjectInfosResp, *mq.CodeMessage) {
+	var sucs []cdssdk.ObjectID
+	err := svc.db.DoTx(sql.LevelSerializable, func(tx *sqlx.Tx) error {
+		msg.Updatings = sort2.Sort(msg.Updatings, func(o1, o2 cdssdk.UpdatingObject) int {
+			return sort2.Cmp(o1.ObjectID, o2.ObjectID)
+		})
+
+		objIDs := make([]cdssdk.ObjectID, len(msg.Updatings))
+		for i, obj := range msg.Updatings {
+			objIDs[i] = obj.ObjectID
+		}
+
+		oldObjs, err := svc.db.Object().BatchGet(tx, objIDs)
+		if err != nil {
+			return fmt.Errorf("batch getting objects: %w", err)
+		}
+		oldObjIDs := make([]cdssdk.ObjectID, len(oldObjs))
+		for i, obj := range oldObjs {
+			oldObjIDs[i] = obj.ObjectID
+		}
+
+		avaiUpdatings, notExistsObjs := pickByObjectIDs(msg.Updatings, oldObjIDs, func(obj cdssdk.UpdatingObject) cdssdk.ObjectID { return obj.ObjectID })
+		if len(notExistsObjs) > 0 {
+			// TODO 部分对象已经不存在
+		}
+
+		newObjs := make([]cdssdk.Object, len(avaiUpdatings))
+		for i := range newObjs {
+			newObjs[i] = oldObjs[i]
+			avaiUpdatings[i].ApplyTo(&newObjs[i])
+		}
+
+		err = svc.db.Object().BatchUpsertByPackagePath(tx, newObjs)
+		if err != nil {
+			return fmt.Errorf("batch create or update: %w", err)
+		}
+
+		sucs = lo.Map(newObjs, func(obj cdssdk.Object, _ int) cdssdk.ObjectID { return obj.ObjectID })
+		return nil
+	})
+
+	if err != nil {
+		logger.Warnf("batch updating objects: %s", err.Error())
+		return nil, mq.Failed(errorcode.OperationFailed, "batch update objects failed")
+	}
+
+	return mq.ReplyOK(coormq.RespUpdateObjectInfos(sucs))
+}
+
+// 根据objIDs从objs中挑选Object。
+// len(objs) >= len(objIDs)
+func pickByObjectIDs[T any](objs []T, objIDs []cdssdk.ObjectID, getID func(T) cdssdk.ObjectID) (picked []T, notFound []T) {
+	objIdx := 0
+	idIdx := 0
+
+	for idIdx < len(objIDs) && objIdx < len(objs) {
+		if getID(objs[objIdx]) < objIDs[idIdx] {
+			notFound = append(notFound, objs[objIdx])
+			objIdx++
+			continue
+		}
+
+		picked = append(picked, objs[objIdx])
+		objIdx++
+		idIdx++
+	}
+
+	return
+}
+
+func (svc *Service) MoveObjects(msg *coormq.MoveObjects) (*coormq.MoveObjectsResp, *mq.CodeMessage) {
+	var sucs []cdssdk.ObjectID
+	err := svc.db.DoTx(sql.LevelSerializable, func(tx *sqlx.Tx) error {
+		msg.Movings = sort2.Sort(msg.Movings, func(o1, o2 cdssdk.MovingObject) int {
+			return sort2.Cmp(o1.ObjectID, o2.ObjectID)
+		})
+
+		objIDs := make([]cdssdk.ObjectID, len(msg.Movings))
+		for i, obj := range msg.Movings {
+			objIDs[i] = obj.ObjectID
+		}
+
+		oldObjs, err := svc.db.Object().BatchGet(tx, objIDs)
+		if err != nil {
+			return fmt.Errorf("batch getting objects: %w", err)
+		}
+		oldObjIDs := make([]cdssdk.ObjectID, len(oldObjs))
+		for i, obj := range oldObjs {
+			oldObjIDs[i] = obj.ObjectID
+		}
+
+		avaiMovings, notExistsObjs := pickByObjectIDs(msg.Movings, oldObjIDs, func(obj cdssdk.MovingObject) cdssdk.ObjectID { return obj.ObjectID })
+		if len(notExistsObjs) > 0 {
+			// TODO 部分对象已经不存在
+		}
+
+		// 筛选出PackageID变化、Path变化的对象，这两种对象要检测改变后是否有冲突
+		var pkgIDChangedObjs []cdssdk.Object
+		var pathChangedObjs []cdssdk.Object
+		for i := range avaiMovings {
+			if avaiMovings[i].PackageID != oldObjs[i].PackageID {
+				newObj := oldObjs[i]
+				avaiMovings[i].ApplyTo(&newObj)
+				pkgIDChangedObjs = append(pkgIDChangedObjs, newObj)
+			} else if avaiMovings[i].Path != oldObjs[i].Path {
+				newObj := oldObjs[i]
+				avaiMovings[i].ApplyTo(&newObj)
+				pathChangedObjs = append(pathChangedObjs, newObj)
+			}
+		}
+
+		var newObjs []cdssdk.Object
+		// 对于PackageID发生变化的对象，需要检查目标Package内是否存在同Path的对象
+		ensuredObjs, err := svc.ensurePackageChangedObjects(tx, msg.UserID, pkgIDChangedObjs)
+		if err != nil {
+			return err
+		}
+		newObjs = append(newObjs, ensuredObjs...)
+
+		// 对于只有Path发生变化的对象，则检查同Package内有没有同Path的对象
+		ensuredObjs, err = svc.ensurePathChangedObjects(tx, msg.UserID, pathChangedObjs)
+		if err != nil {
+			return err
+		}
+		newObjs = append(newObjs, ensuredObjs...)
+
+		err = svc.db.Object().BatchUpert(tx, newObjs)
+		if err != nil {
+			return fmt.Errorf("batch create or update: %w", err)
+		}
+
+		sucs = lo.Map(newObjs, func(obj cdssdk.Object, _ int) cdssdk.ObjectID { return obj.ObjectID })
+		return nil
+	})
+	if err != nil {
+		logger.Warn(err.Error())
+		return nil, mq.Failed(errorcode.OperationFailed, "move objects failed")
+	}
+
+	return mq.ReplyOK(coormq.RespMoveObjects(sucs))
+}
+
+func (svc *Service) ensurePackageChangedObjects(tx *sqlx.Tx, userID cdssdk.UserID, objs []cdssdk.Object) ([]cdssdk.Object, error) {
+	if len(objs) == 0 {
+		return nil, nil
+	}
+
+	type PackageObjects struct {
+		PackageID    cdssdk.PackageID
+		ObjectByPath map[string]*cdssdk.Object
+	}
+
+	packages := make(map[cdssdk.PackageID]*PackageObjects)
+	for _, obj := range objs {
+		pkg, ok := packages[obj.PackageID]
+		if !ok {
+			pkg = &PackageObjects{
+				PackageID:    obj.PackageID,
+				ObjectByPath: make(map[string]*cdssdk.Object),
+			}
+			packages[obj.PackageID] = pkg
+		}
+
+		if pkg.ObjectByPath[obj.Path] == nil {
+			o := obj
+			pkg.ObjectByPath[obj.Path] = &o
+		} else {
+			// TODO 有两个对象移动到同一个路径，有冲突
+		}
+	}
+
+	var willUpdateObjs []cdssdk.Object
+	for _, pkg := range packages {
+		_, err := svc.db.Package().GetUserPackage(tx, userID, pkg.PackageID)
+		if err == sql.ErrNoRows {
+			continue
+		}
+		if err != nil {
+			return nil, fmt.Errorf("getting user package by id: %w", err)
+		}
+
+		existsObjs, err := svc.db.Object().BatchGetByPackagePath(tx, pkg.PackageID, lo.Keys(pkg.ObjectByPath))
+		if err != nil {
+			return nil, fmt.Errorf("batch getting objects by package path: %w", err)
+		}
+
+		// 标记冲突的对象
+		for _, obj := range existsObjs {
+			pkg.ObjectByPath[obj.Path] = nil
+			// TODO 目标Package内有冲突的对象
+		}
+
+		for _, obj := range pkg.ObjectByPath {
+			if obj == nil {
+				continue
+			}
+			willUpdateObjs = append(willUpdateObjs, *obj)
+		}
+	}
+
+	return willUpdateObjs, nil
+}
+
+func (svc *Service) ensurePathChangedObjects(tx *sqlx.Tx, userID cdssdk.UserID, objs []cdssdk.Object) ([]cdssdk.Object, error) {
+	if len(objs) == 0 {
+		return nil, nil
+	}
+
+	objByPath := make(map[string]*cdssdk.Object)
+	for _, obj := range objs {
+		if objByPath[obj.Path] == nil {
+			o := obj
+			objByPath[obj.Path] = &o
+		} else {
+			// TODO 有两个对象移动到同一个路径，有冲突
+		}
+
+	}
+
+	_, err := svc.db.Package().GetUserPackage(tx, userID, objs[0].PackageID)
+	if err == sql.ErrNoRows {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, fmt.Errorf("getting user package by id: %w", err)
+	}
+
+	existsObjs, err := svc.db.Object().BatchGetByPackagePath(tx, objs[0].PackageID, lo.Map(objs, func(obj cdssdk.Object, idx int) string { return obj.Path }))
+	if err != nil {
+		return nil, fmt.Errorf("batch getting objects by package path: %w", err)
+	}
+
+	// 不支持两个对象交换位置的情况，因为数据库不支持
+	for _, obj := range existsObjs {
+		objByPath[obj.Path] = nil
+	}
+
+	var willMoveObjs []cdssdk.Object
+	for _, obj := range objByPath {
+		if obj == nil {
+			continue
+		}
+		willMoveObjs = append(willMoveObjs, *obj)
+	}
+
+	return willMoveObjs, nil
+}
+
+func (svc *Service) DeleteObjects(msg *coormq.DeleteObjects) (*coormq.DeleteObjectsResp, *mq.CodeMessage) {
+	err := svc.db.DoTx(sql.LevelSerializable, func(tx *sqlx.Tx) error {
+		err := svc.db.Object().BatchDelete(tx, msg.ObjectIDs)
+		if err != nil {
+			return fmt.Errorf("batch deleting objects: %w", err)
+		}
+
+		err = svc.db.ObjectBlock().BatchDeleteByObjectID(tx, msg.ObjectIDs)
+		if err != nil {
+			return fmt.Errorf("batch deleting object blocks: %w", err)
+		}
+
+		err = svc.db.PinnedObject().BatchDeleteByObjectID(tx, msg.ObjectIDs)
+		if err != nil {
+			return fmt.Errorf("batch deleting pinned objects: %w", err)
+		}
+
+		return nil
+	})
+	if err != nil {
+		logger.Warnf("batch deleting objects: %s", err.Error())
+		return nil, mq.Failed(errorcode.OperationFailed, "batch delete objects failed")
+	}
+
+	return mq.ReplyOK(coormq.RespDeleteObjects())
 }

coordinator/internal/mq/package.go

@@ -26,8 +26,22 @@ func (svc *Service) GetPackage(msg *coormq.GetPackage) (*coormq.GetPackageResp,
 	return mq.ReplyOK(coormq.NewGetPackageResp(pkg))
 }
 
+func (svc *Service) GetPackageByName(msg *coormq.GetPackageByName) (*coormq.GetPackageByNameResp, *mq.CodeMessage) {
+	pkg, err := svc.db.Package().GetUserPackageByName(svc.db.SQLCtx(), msg.UserID, msg.BucketName, msg.PackageName)
+	if err != nil {
+		logger.WithField("UserID", msg.UserID).
+			WithField("BucketName", msg.BucketName).
+			WithField("PackageName", msg.PackageName).
+			Warnf("get package by name: %s", err.Error())
+
+		return nil, mq.Failed(errorcode.OperationFailed, "get package by name failed")
+	}
+
+	return mq.ReplyOK(coormq.NewGetPackageByNameResp(pkg))
+}
+
 func (svc *Service) CreatePackage(msg *coormq.CreatePackage) (*coormq.CreatePackageResp, *mq.CodeMessage) {
-	var pkgID cdssdk.PackageID
+	var pkg cdssdk.Package
 	err := svc.db.DoTx(sql.LevelSerializable, func(tx *sqlx.Tx) error {
 		var err error
 
@@ -36,11 +50,16 @@ func (svc *Service) CreatePackage(msg *coormq.CreatePackage) (*coormq.CreatePack
 			return fmt.Errorf("bucket is not avaiable to the user")
 		}
 
-		pkgID, err = svc.db.Package().Create(tx, msg.BucketID, msg.Name)
+		pkgID, err := svc.db.Package().Create(tx, msg.BucketID, msg.Name)
 		if err != nil {
 			return fmt.Errorf("creating package: %w", err)
 		}
 
+		pkg, err = svc.db.Package().GetByID(tx, pkgID)
+		if err != nil {
+			return fmt.Errorf("getting package by id: %w", err)
+		}
+
 		return nil
 	})
 	if err != nil {
@@ -50,10 +69,11 @@ func (svc *Service) CreatePackage(msg *coormq.CreatePackage) (*coormq.CreatePack
 		return nil, mq.Failed(errorcode.OperationFailed, "creating package failed")
 	}
 
-	return mq.ReplyOK(coormq.NewCreatePackageResp(pkgID))
+	return mq.ReplyOK(coormq.NewCreatePackageResp(pkg))
 }
 
 func (svc *Service) UpdatePackage(msg *coormq.UpdatePackage) (*coormq.UpdatePackageResp, *mq.CodeMessage) {
+	var added []cdssdk.Object
 	err := svc.db.DoTx(sql.LevelSerializable, func(tx *sqlx.Tx) error {
 		_, err := svc.db.Package().GetByID(tx, msg.PackageID)
 		if err != nil {
@@ -69,9 +89,11 @@ func (svc *Service) UpdatePackage(msg *coormq.UpdatePackage) (*coormq.UpdatePack
 
 		// 再执行添加操作
 		if len(msg.Adds) > 0 {
-			if _, err := svc.db.Object().BatchAdd(tx, msg.PackageID, msg.Adds); err != nil {
+			ad, err := svc.db.Object().BatchAdd(tx, msg.PackageID, msg.Adds)
+			if err != nil {
 				return fmt.Errorf("adding objects: %w", err)
 			}
+			added = ad
 		}
 
 		return nil
@@ -81,7 +103,7 @@ func (svc *Service) UpdatePackage(msg *coormq.UpdatePackage) (*coormq.UpdatePack
 		return nil, mq.Failed(errorcode.OperationFailed, "update package failed")
 	}
 
-	return mq.ReplyOK(coormq.NewUpdatePackageResp())
+	return mq.ReplyOK(coormq.NewUpdatePackageResp(added))
 }
 
 func (svc *Service) DeletePackage(msg *coormq.DeletePackage) (*coormq.DeletePackageResp, *mq.CodeMessage) {

scanner/internal/event/check_package_redundancy.go

@@ -109,7 +109,7 @@ func (t *CheckPackageRedundancy) Execute(execCtx ExecuteContext) {
 		}
 	}
 
-	var changedObjects []coormq.ChangeObjectRedundancyEntry
+	var changedObjects []coormq.UpdatingObjectRedundancy
 
 	defRep := cdssdk.DefaultRepRedundancy
 	defEC := cdssdk.DefaultECRedundancy
@@ -136,7 +136,7 @@ func (t *CheckPackageRedundancy) Execute(execCtx ExecuteContext) {
 	defer mutex.Unlock()
 
 	for _, obj := range getObjs.Objects {
-		var entry *coormq.ChangeObjectRedundancyEntry
+		var updating *coormq.UpdatingObjectRedundancy
 		var err error
 
 		shouldUseEC := obj.Object.Size > config.Cfg().ECFileSizeThreshold
@@ -145,32 +145,32 @@ func (t *CheckPackageRedundancy) Execute(execCtx ExecuteContext) {
 		case *cdssdk.NoneRedundancy:
 			if shouldUseEC {
 				log.WithField("ObjectID", obj.Object.ObjectID).Debugf("redundancy: none -> ec")
-				entry, err = t.noneToEC(obj, &defEC, newECNodes)
+				updating, err = t.noneToEC(obj, &defEC, newECNodes)
 			} else {
 				log.WithField("ObjectID", obj.Object.ObjectID).Debugf("redundancy: none -> rep")
-				entry, err = t.noneToRep(obj, &defRep, newRepNodes)
+				updating, err = t.noneToRep(obj, &defRep, newRepNodes)
 			}
 
 		case *cdssdk.RepRedundancy:
 			if shouldUseEC {
 				log.WithField("ObjectID", obj.Object.ObjectID).Debugf("redundancy: rep -> ec")
-				entry, err = t.repToEC(obj, &defEC, newECNodes)
+				updating, err = t.repToEC(obj, &defEC, newECNodes)
 			} else {
-				entry, err = t.repToRep(obj, &defRep, rechoosedRepNodes)
+				updating, err = t.repToRep(obj, &defRep, rechoosedRepNodes)
 			}
 
 		case *cdssdk.ECRedundancy:
 			if shouldUseEC {
 				uploadNodes := t.rechooseNodesForEC(obj, red, allNodes)
-				entry, err = t.ecToEC(obj, red, &defEC, uploadNodes)
+				updating, err = t.ecToEC(obj, red, &defEC, uploadNodes)
 			} else {
 				log.WithField("ObjectID", obj.Object.ObjectID).Debugf("redundancy: ec -> rep")
-				entry, err = t.ecToRep(obj, red, &defRep, newRepNodes)
+				updating, err = t.ecToRep(obj, red, &defRep, newRepNodes)
 			}
 		}
 
-		if entry != nil {
-			changedObjects = append(changedObjects, *entry)
+		if updating != nil {
+			changedObjects = append(changedObjects, *updating)
 		}
 
 		if err != nil {
@@ -182,7 +182,7 @@ func (t *CheckPackageRedundancy) Execute(execCtx ExecuteContext) {
 		return
 	}
 
-	_, err = coorCli.ChangeObjectRedundancy(coormq.ReqChangeObjectRedundancy(changedObjects))
+	_, err = coorCli.UpdateObjectRedundancy(coormq.ReqUpdateObjectRedundancy(changedObjects))
 	if err != nil {
 		log.Warnf("requesting to change object redundancy: %s", err.Error())
 		return
@@ -367,7 +367,7 @@ func (t *CheckPackageRedundancy) chooseSoManyNodes(count int, nodes []*NodeLoadI
 	return chosen
 }
 
-func (t *CheckPackageRedundancy) noneToRep(obj stgmod.ObjectDetail, red *cdssdk.RepRedundancy, uploadNodes []*NodeLoadInfo) (*coormq.ChangeObjectRedundancyEntry, error) {
+func (t *CheckPackageRedundancy) noneToRep(obj stgmod.ObjectDetail, red *cdssdk.RepRedundancy, uploadNodes []*NodeLoadInfo) (*coormq.UpdatingObjectRedundancy, error) {
 	if len(obj.Blocks) == 0 {
 		return nil, fmt.Errorf("object is not cached on any nodes, cannot change its redundancy to rep")
 	}
@@ -389,14 +389,14 @@ func (t *CheckPackageRedundancy) noneToRep(obj stgmod.ObjectDetail, red *cdssdk.
 		})
 	}
 
-	return &coormq.ChangeObjectRedundancyEntry{
+	return &coormq.UpdatingObjectRedundancy{
 		ObjectID:   obj.Object.ObjectID,
 		Redundancy: red,
 		Blocks:     blocks,
 	}, nil
 }
 
-func (t *CheckPackageRedundancy) noneToEC(obj stgmod.ObjectDetail, red *cdssdk.ECRedundancy, uploadNodes []*NodeLoadInfo) (*coormq.ChangeObjectRedundancyEntry, error) {
+func (t *CheckPackageRedundancy) noneToEC(obj stgmod.ObjectDetail, red *cdssdk.ECRedundancy, uploadNodes []*NodeLoadInfo) (*coormq.UpdatingObjectRedundancy, error) {
 	coorCli, err := stgglb.CoordinatorMQPool.Acquire()
 	if err != nil {
 		return nil, fmt.Errorf("new coordinator client: %w", err)
@@ -443,14 +443,14 @@ func (t *CheckPackageRedundancy) noneToEC(obj stgmod.ObjectDetail, red *cdssdk.E
 		})
 	}
 
-	return &coormq.ChangeObjectRedundancyEntry{
+	return &coormq.UpdatingObjectRedundancy{
 		ObjectID:   obj.Object.ObjectID,
 		Redundancy: red,
 		Blocks:     blocks,
 	}, nil
 }
 
-func (t *CheckPackageRedundancy) repToRep(obj stgmod.ObjectDetail, red *cdssdk.RepRedundancy, uploadNodes []*NodeLoadInfo) (*coormq.ChangeObjectRedundancyEntry, error) {
+func (t *CheckPackageRedundancy) repToRep(obj stgmod.ObjectDetail, red *cdssdk.RepRedundancy, uploadNodes []*NodeLoadInfo) (*coormq.UpdatingObjectRedundancy, error) {
 	if len(obj.Blocks) == 0 {
 		return nil, fmt.Errorf("object is not cached on any nodes, cannot change its redundancy to rep")
 	}
@@ -479,18 +479,18 @@ func (t *CheckPackageRedundancy) repToRep(obj stgmod.ObjectDetail, red *cdssdk.R
 		})
 	}
 
-	return &coormq.ChangeObjectRedundancyEntry{
+	return &coormq.UpdatingObjectRedundancy{
 		ObjectID:   obj.Object.ObjectID,
 		Redundancy: red,
 		Blocks:     blocks,
 	}, nil
 }
 
-func (t *CheckPackageRedundancy) repToEC(obj stgmod.ObjectDetail, red *cdssdk.ECRedundancy, uploadNodes []*NodeLoadInfo) (*coormq.ChangeObjectRedundancyEntry, error) {
+func (t *CheckPackageRedundancy) repToEC(obj stgmod.ObjectDetail, red *cdssdk.ECRedundancy, uploadNodes []*NodeLoadInfo) (*coormq.UpdatingObjectRedundancy, error) {
 	return t.noneToEC(obj, red, uploadNodes)
 }
 
-func (t *CheckPackageRedundancy) ecToRep(obj stgmod.ObjectDetail, srcRed *cdssdk.ECRedundancy, tarRed *cdssdk.RepRedundancy, uploadNodes []*NodeLoadInfo) (*coormq.ChangeObjectRedundancyEntry, error) {
+func (t *CheckPackageRedundancy) ecToRep(obj stgmod.ObjectDetail, srcRed *cdssdk.ECRedundancy, tarRed *cdssdk.RepRedundancy, uploadNodes []*NodeLoadInfo) (*coormq.UpdatingObjectRedundancy, error) {
 	coorCli, err := stgglb.CoordinatorMQPool.Acquire()
 	if err != nil {
 		return nil, fmt.Errorf("new coordinator client: %w", err)
@@ -556,14 +556,14 @@ func (t *CheckPackageRedundancy) ecToRep(obj stgmod.ObjectDetail, srcRed *cdssdk
 		})
 	}
 
-	return &coormq.ChangeObjectRedundancyEntry{
+	return &coormq.UpdatingObjectRedundancy{
 		ObjectID:   obj.Object.ObjectID,
 		Redundancy: tarRed,
 		Blocks:     blocks,
 	}, nil
 }
 
-func (t *CheckPackageRedundancy) ecToEC(obj stgmod.ObjectDetail, srcRed *cdssdk.ECRedundancy, tarRed *cdssdk.ECRedundancy, uploadNodes []*NodeLoadInfo) (*coormq.ChangeObjectRedundancyEntry, error) {
+func (t *CheckPackageRedundancy) ecToEC(obj stgmod.ObjectDetail, srcRed *cdssdk.ECRedundancy, tarRed *cdssdk.ECRedundancy, uploadNodes []*NodeLoadInfo) (*coormq.UpdatingObjectRedundancy, error) {
 	coorCli, err := stgglb.CoordinatorMQPool.Acquire()
 	if err != nil {
 		return nil, fmt.Errorf("new coordinator client: %w", err)
@@ -654,7 +654,7 @@ func (t *CheckPackageRedundancy) ecToEC(obj stgmod.ObjectDetail, srcRed *cdssdk.
 		newBlocks[idx].FileHash = v.(string)
 	}
 
-	return &coormq.ChangeObjectRedundancyEntry{
+	return &coormq.UpdatingObjectRedundancy{
 		ObjectID:   obj.Object.ObjectID,
 		Redundancy: tarRed,
 		Blocks:     newBlocks,

scanner/internal/event/clean_pinned.go

@@ -109,7 +109,7 @@ func (t *CleanPinned) Execute(execCtx ExecuteContext) {
 	pinPlans := make(map[cdssdk.NodeID]*[]string)
 
 	// 对于rep对象，统计出所有对象块分布最多的两个节点，用这两个节点代表所有rep对象块的分布，去进行退火算法
-	var repObjectsUpdateEntries []coormq.ChangeObjectRedundancyEntry
+	var repObjectsUpdating []coormq.UpdatingObjectRedundancy
 	repMostNodeIDs := t.summaryRepObjectBlockNodes(repObjects)
 	solu := t.startAnnealing(allNodeInfos, readerNodeIDs, annealingObject{
 		totalBlockCount: 1,
@@ -118,11 +118,11 @@ func (t *CleanPinned) Execute(execCtx ExecuteContext) {
 		blocks:          nil,
 	})
 	for _, obj := range repObjects {
-		repObjectsUpdateEntries = append(repObjectsUpdateEntries, t.makePlansForRepObject(solu, obj, pinPlans))
+		repObjectsUpdating = append(repObjectsUpdating, t.makePlansForRepObject(solu, obj, pinPlans))
 	}
 
 	// 对于ec对象，则每个对象单独进行退火算法
-	var ecObjectsUpdateEntries []coormq.ChangeObjectRedundancyEntry
+	var ecObjectsUpdating []coormq.UpdatingObjectRedundancy
 	for _, obj := range ecObjects {
 		ecRed := obj.Object.Redundancy.(*cdssdk.ECRedundancy)
 		solu := t.startAnnealing(allNodeInfos, readerNodeIDs, annealingObject{
@@ -131,7 +131,7 @@ func (t *CleanPinned) Execute(execCtx ExecuteContext) {
 			pinnedAt:        obj.PinnedAt,
 			blocks:          obj.Blocks,
 		})
-		ecObjectsUpdateEntries = append(ecObjectsUpdateEntries, t.makePlansForECObject(allNodeInfos, solu, obj, &planBld))
+		ecObjectsUpdating = append(ecObjectsUpdating, t.makePlansForECObject(allNodeInfos, solu, obj, &planBld))
 	}
 
 	ioSwRets, err := t.executePlans(execCtx, pinPlans, &planBld)
@@ -141,13 +141,13 @@ func (t *CleanPinned) Execute(execCtx ExecuteContext) {
 	}
 
 	// 根据按照方案进行调整的结果，填充更新元数据的命令
-	for i := range ecObjectsUpdateEntries {
-		t.populateECObjectEntry(&ecObjectsUpdateEntries[i], ecObjects[i], ioSwRets)
+	for i := range ecObjectsUpdating {
+		t.populateECObjectEntry(&ecObjectsUpdating[i], ecObjects[i], ioSwRets)
 	}
 
-	finalEntries := append(repObjectsUpdateEntries, ecObjectsUpdateEntries...)
+	finalEntries := append(repObjectsUpdating, ecObjectsUpdating...)
 	if len(finalEntries) > 0 {
-		_, err = coorCli.ChangeObjectRedundancy(coormq.ReqChangeObjectRedundancy(finalEntries))
+		_, err = coorCli.UpdateObjectRedundancy(coormq.ReqUpdateObjectRedundancy(finalEntries))
 		if err != nil {
 			log.Warnf("changing object redundancy: %s", err.Error())
 			return
@@ -715,8 +715,8 @@ func (t *CleanPinned) alwaysAccept(curTemp float64, dScore float64, coolingRate
 	return v > rand.Float64()
 }
 
-func (t *CleanPinned) makePlansForRepObject(solu annealingSolution, obj stgmod.ObjectDetail, pinPlans map[cdssdk.NodeID]*[]string) coormq.ChangeObjectRedundancyEntry {
-	entry := coormq.ChangeObjectRedundancyEntry{
+func (t *CleanPinned) makePlansForRepObject(solu annealingSolution, obj stgmod.ObjectDetail, pinPlans map[cdssdk.NodeID]*[]string) coormq.UpdatingObjectRedundancy {
+	entry := coormq.UpdatingObjectRedundancy{
 		ObjectID:   obj.Object.ObjectID,
 		Redundancy: obj.Object.Redundancy,
 	}
@@ -748,8 +748,8 @@ func (t *CleanPinned) makePlansForRepObject(solu annealingSolution, obj stgmod.O
 	return entry
 }
 
-func (t *CleanPinned) makePlansForECObject(allNodeInfos map[cdssdk.NodeID]*cdssdk.Node, solu annealingSolution, obj stgmod.ObjectDetail, planBld *plans.PlanBuilder) coormq.ChangeObjectRedundancyEntry {
-	entry := coormq.ChangeObjectRedundancyEntry{
+func (t *CleanPinned) makePlansForECObject(allNodeInfos map[cdssdk.NodeID]*cdssdk.Node, solu annealingSolution, obj stgmod.ObjectDetail, planBld *plans.PlanBuilder) coormq.UpdatingObjectRedundancy {
+	entry := coormq.UpdatingObjectRedundancy{
 		ObjectID:   obj.Object.ObjectID,
 		Redundancy: obj.Object.Redundancy,
 	}
@@ -871,7 +871,7 @@ func (t *CleanPinned) executePlans(execCtx ExecuteContext, pinPlans map[cdssdk.N
 	return ioSwRets, nil
 }
 
-func (t *CleanPinned) populateECObjectEntry(entry *coormq.ChangeObjectRedundancyEntry, obj stgmod.ObjectDetail, ioRets map[string]any) {
+func (t *CleanPinned) populateECObjectEntry(entry *coormq.UpdatingObjectRedundancy, obj stgmod.ObjectDetail, ioRets map[string]any) {
 	for i := range entry.Blocks {
 		if entry.Blocks[i].FileHash != "" {
 			continue