JCS-pub/client/internal/services/object.go

package services

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/samber/lo"
	"gitlink.org.cn/cloudream/common/pkgs/ioswitch/exec"
	"gitlink.org.cn/cloudream/common/pkgs/logger"
	"gitlink.org.cn/cloudream/common/utils/sort2"
	"gitlink.org.cn/cloudream/jcs-pub/client/internal/db"
	"gitlink.org.cn/cloudream/jcs-pub/client/internal/downloader"
	"gitlink.org.cn/cloudream/jcs-pub/client/sdk/api/v1"
	"gitlink.org.cn/cloudream/jcs-pub/client/types"
	"gitlink.org.cn/cloudream/jcs-pub/common/models/datamap"
	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/distlock/reqbuilder"
	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/ioswitch2/ops2"
	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/ioswitch2/plans"
	"gorm.io/gorm"
)

// ObjectService 定义了对象服务，负责管理对象的上传、下载等操作。
type ObjectService struct {
	*Service
}

// ObjectSvc 返回一个ObjectService的实例。
func (svc *Service) ObjectSvc() *ObjectService {
	return &ObjectService{Service: svc}
}

func (svc *ObjectService) GetByPath(req api.ObjectListByPath) (api.ObjectListByPathResp, error) {
	var resp api.ObjectListByPathResp

	maxKeys := 1000
	if req.MaxKeys > 0 {
		maxKeys = req.MaxKeys
	}

	err := svc.DB.DoTx(func(tx db.SQLContext) error {
		var err error

		_, err = svc.DB.Package().GetByID(tx, req.PackageID)
		if err != nil {
			return fmt.Errorf("getting package by id: %w", err)
		}

		if !req.IsPrefix {
			obj, err := svc.DB.Object().GetByPath(tx, req.PackageID, req.Path)
			if err != nil {
				return fmt.Errorf("getting object by path: %w", err)
			}
			resp.Objects = append(resp.Objects, obj)

			return nil
		}

		if !req.NoRecursive {
			resp.Objects, err = svc.DB.Object().GetWithPathPrefixPaged(tx, req.PackageID, req.Path, req.ContinuationToken, maxKeys)
			if err != nil {
				return fmt.Errorf("getting objects with prefix: %w", err)
			}

			if len(resp.Objects) > 0 {
				resp.NextContinuationToken = resp.Objects[len(resp.Objects)-1].Path
			}

			return nil
		}

		resp.Objects, resp.CommonPrefixes, resp.NextContinuationToken, err = svc.DB.Object().GetByPrefixGroupedPaged(tx, req.PackageID, req.Path, req.ContinuationToken, maxKeys)
		return err
	})
	return resp, err
}

func (svc *ObjectService) GetByIDs(objectIDs []types.ObjectID) ([]*types.Object, error) {
	var ret []*types.Object
	err := svc.DB.DoTx(func(tx db.SQLContext) error {
		objs, err := svc.DB.Object().BatchGet(tx, objectIDs)
		if err != nil {
			return err
		}

		objMp := make(map[types.ObjectID]types.Object)
		for _, obj := range objs {
			objMp[obj.ObjectID] = obj
		}

		for _, objID := range objectIDs {
			o, ok := objMp[objID]
			if ok {
				ret = append(ret, &o)
			} else {
				ret = append(ret, nil)
			}
		}

		return err
	})
	return ret, err
}

func (svc *ObjectService) UpdateInfo(updatings []api.UpdatingObject) ([]types.ObjectID, error) {
	var sucs []types.ObjectID
	err := svc.DB.DoTx(func(tx db.SQLContext) error {
		updatings = sort2.Sort(updatings, func(o1, o2 api.UpdatingObject) int {
			return sort2.Cmp(o1.ObjectID, o2.ObjectID)
		})

		objIDs := make([]types.ObjectID, len(updatings))
		for i, obj := range 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([]types.ObjectID, len(oldObjs))
		for i, obj := range oldObjs {
			oldObjIDs[i] = obj.ObjectID
		}

		avaiUpdatings, notExistsObjs := pickByObjectIDs(updatings, oldObjIDs, func(obj api.UpdatingObject) types.ObjectID { return obj.ObjectID })
		if len(notExistsObjs) > 0 {
			// TODO 部分对象已经不存在
		}

		newObjs := make([]types.Object, len(avaiUpdatings))
		for i := range newObjs {
			newObjs[i] = oldObjs[i]
			avaiUpdatings[i].ApplyTo(&newObjs[i])
		}

		err = svc.DB.Object().BatchUpdate(tx, newObjs)
		if err != nil {
			return fmt.Errorf("batch create or update: %w", err)
		}

		sucs = lo.Map(newObjs, func(obj types.Object, _ int) types.ObjectID { return obj.ObjectID })
		return nil
	})
	return sucs, err
}

// 根据objIDs从objs中挑选Object。
// len(objs) >= len(objIDs)
func pickByObjectIDs[T any](objs []T, objIDs []types.ObjectID, getID func(T) types.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 *ObjectService) Move(movings []api.MovingObject) ([]types.ObjectID, error) {
	var sucs []types.ObjectID
	var evt []*datamap.BodyObjectInfoUpdated

	err := svc.DB.DoTx(func(tx db.SQLContext) error {
		movings = sort2.Sort(movings, func(o1, o2 api.MovingObject) int {
			return sort2.Cmp(o1.ObjectID, o2.ObjectID)
		})

		objIDs := make([]types.ObjectID, len(movings))
		for i, obj := range 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([]types.ObjectID, len(oldObjs))
		for i, obj := range oldObjs {
			oldObjIDs[i] = obj.ObjectID
		}

		// 找出仍在数据库的Object
		avaiMovings, notExistsObjs := pickByObjectIDs(movings, oldObjIDs, func(obj api.MovingObject) types.ObjectID { return obj.ObjectID })
		if len(notExistsObjs) > 0 {
			// TODO 部分对象已经不存在
		}

		// 筛选出PackageID变化、Path变化的对象，这两种对象要检测改变后是否有冲突
		var pkgIDChangedObjs []types.Object
		var pathChangedObjs []types.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 []types.Object
		// 对于PackageID发生变化的对象，需要检查目标Package内是否存在同Path的对象
		checkedObjs, err := svc.checkPackageChangedObjects(tx, pkgIDChangedObjs)
		if err != nil {
			return err
		}
		newObjs = append(newObjs, checkedObjs...)

		// 对于只有Path发生变化的对象，则检查同Package内有没有同Path的对象
		checkedObjs, err = svc.checkPathChangedObjects(tx, pathChangedObjs)
		if err != nil {
			return err
		}
		newObjs = append(newObjs, checkedObjs...)

		err = svc.DB.Object().BatchUpdate(tx, newObjs)
		if err != nil {
			return fmt.Errorf("batch create or update: %w", err)
		}

		sucs = lo.Map(newObjs, func(obj types.Object, _ int) types.ObjectID { return obj.ObjectID })
		evt = lo.Map(newObjs, func(obj types.Object, _ int) *datamap.BodyObjectInfoUpdated {
			return &datamap.BodyObjectInfoUpdated{
				Object: obj,
			}
		})
		return nil
	})
	if err != nil {
		logger.Warn(err.Error())
		return nil, err
	}

	for _, e := range evt {
		svc.EvtPub.Publish(e)
	}

	return sucs, nil
}

func (svc *ObjectService) Download(req downloader.DownloadReqeust) (*downloader.Downloading, error) {
	iter := svc.Downloader.DownloadObjects([]downloader.DownloadReqeust{req})

	// 初始化下载过程
	downloading, err := iter.MoveNext()
	if err != nil {
		return nil, err
	}
	if downloading.Object == nil {
		return nil, fmt.Errorf("object %v not found", req.ObjectID)
	}

	return downloading, nil
}

func (svc *Service) checkPackageChangedObjects(tx db.SQLContext, objs []types.Object) ([]types.Object, error) {
	if len(objs) == 0 {
		return nil, nil
	}

	type PackageObjects struct {
		PackageID    types.PackageID
		ObjectByPath map[string]*types.Object
	}

	packages := make(map[types.PackageID]*PackageObjects)
	for _, obj := range objs {
		pkg, ok := packages[obj.PackageID]
		if !ok {
			pkg = &PackageObjects{
				PackageID:    obj.PackageID,
				ObjectByPath: make(map[string]*types.Object),
			}
			packages[obj.PackageID] = pkg
		}

		if pkg.ObjectByPath[obj.Path] == nil {
			o := obj
			pkg.ObjectByPath[obj.Path] = &o
		} else {
			// TODO 有两个对象移动到同一个路径，有冲突
		}
	}

	var willUpdateObjs []types.Object
	for _, pkg := range packages {
		_, err := svc.DB.Package().GetByID(tx, pkg.PackageID)
		if errors.Is(err, gorm.ErrRecordNotFound) {
			continue
		}
		if err != nil {
			return nil, fmt.Errorf("getting 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) checkPathChangedObjects(tx db.SQLContext, objs []types.Object) ([]types.Object, error) {
	if len(objs) == 0 {
		return nil, nil
	}

	objByPath := make(map[string]*types.Object)
	for _, obj := range objs {
		if objByPath[obj.Path] == nil {
			o := obj
			objByPath[obj.Path] = &o
		} else {
			// TODO 有两个对象移动到同一个路径，有冲突
		}

	}

	_, err := svc.DB.Package().GetByID(tx, objs[0].PackageID)
	if errors.Is(err, gorm.ErrRecordNotFound) {
		return nil, nil
	}
	if err != nil {
		return nil, fmt.Errorf("getting package by id: %w", err)
	}

	existsObjs, err := svc.DB.Object().BatchGetByPackagePath(tx, objs[0].PackageID, lo.Map(objs, func(obj types.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 []types.Object
	for _, obj := range objByPath {
		if obj == nil {
			continue
		}
		willMoveObjs = append(willMoveObjs, *obj)
	}

	return willMoveObjs, nil
}

func (svc *ObjectService) Delete(objectIDs []types.ObjectID) error {
	var sucs []types.ObjectID
	err := svc.DB.DoTx(func(tx db.SQLContext) error {
		avaiIDs, err := svc.DB.Object().BatchTestObjectID(tx, objectIDs)
		if err != nil {
			return fmt.Errorf("batch testing object id: %w", err)
		}
		sucs = lo.Keys(avaiIDs)

		return svc.DB.Object().BatchDeleteComplete(tx, sucs)
	})
	if err != nil {
		return err
	}

	for _, objID := range sucs {
		svc.EvtPub.Publish(&datamap.BodyObjectDeleted{
			ObjectID: objID,
		})
	}
	return nil
}

func (svc *ObjectService) Clone(clonings []api.CloningObject) ([]*types.Object, error) {
	type CloningObject struct {
		Cloning  api.CloningObject
		OrgIndex int
	}
	type PackageClonings struct {
		PackageID types.PackageID
		Clonings  map[string]CloningObject
	}

	var evt []*datamap.BodyNewOrUpdateObject

	cloningMap := make(map[types.PackageID]*PackageClonings)
	for i, cloning := range clonings {
		pkg, ok := cloningMap[cloning.NewPackageID]
		if !ok {
			pkg = &PackageClonings{
				PackageID: cloning.NewPackageID,
				Clonings:  make(map[string]CloningObject),
			}
			cloningMap[cloning.NewPackageID] = pkg
		}
		pkg.Clonings[cloning.NewPath] = CloningObject{
			Cloning:  cloning,
			OrgIndex: i,
		}
	}

	ret := make([]*types.Object, len(cloningMap))
	err := svc.DB.DoTx(func(tx db.SQLContext) error {
		// 剔除掉新路径已经存在的对象
		for _, pkg := range cloningMap {
			exists, err := svc.DB.Object().BatchGetByPackagePath(tx, pkg.PackageID, lo.Keys(pkg.Clonings))
			if err != nil {
				return fmt.Errorf("batch getting objects by package path: %w", err)
			}

			for _, obj := range exists {
				delete(pkg.Clonings, obj.Path)
			}
		}

		// 删除目的Package不存在的对象
		newPkg, err := svc.DB.Package().BatchTestPackageID(tx, lo.Keys(cloningMap))
		if err != nil {
			return fmt.Errorf("batch testing package id: %w", err)
		}
		for _, pkg := range cloningMap {
			if !newPkg[pkg.PackageID] {
				delete(cloningMap, pkg.PackageID)
			}
		}

		var avaiClonings []CloningObject
		var avaiObjIDs []types.ObjectID
		for _, pkg := range cloningMap {
			for _, cloning := range pkg.Clonings {
				avaiClonings = append(avaiClonings, cloning)
				avaiObjIDs = append(avaiObjIDs, cloning.Cloning.ObjectID)
			}
		}

		avaiDetails, err := svc.DB.Object().BatchGetDetails(tx, avaiObjIDs)
		if err != nil {
			return fmt.Errorf("batch getting object details: %w", err)
		}

		avaiDetailsMap := make(map[types.ObjectID]types.ObjectDetail)
		for _, detail := range avaiDetails {
			avaiDetailsMap[detail.Object.ObjectID] = detail
		}

		oldAvaiClonings := avaiClonings
		avaiClonings = nil

		var newObjs []types.Object
		for _, cloning := range oldAvaiClonings {
			// 进一步剔除原始对象不存在的情况
			detail, ok := avaiDetailsMap[cloning.Cloning.ObjectID]
			if !ok {
				continue
			}

			avaiClonings = append(avaiClonings, cloning)

			newObj := detail.Object
			newObj.ObjectID = 0
			newObj.Path = cloning.Cloning.NewPath
			newObj.PackageID = cloning.Cloning.NewPackageID
			newObjs = append(newObjs, newObj)
		}

		// 先创建出新对象
		err = svc.DB.Object().BatchCreate(tx, &newObjs)
		if err != nil {
			return fmt.Errorf("batch creating objects: %w", err)
		}

		// 创建了新对象就能拿到新对象ID，再创建新对象块
		var newBlks []types.ObjectBlock
		for i, cloning := range avaiClonings {
			oldBlks := avaiDetailsMap[cloning.Cloning.ObjectID].Blocks
			for _, blk := range oldBlks {
				newBlk := blk
				newBlk.ObjectID = newObjs[i].ObjectID
				newBlks = append(newBlks, newBlk)
			}
		}

		err = svc.DB.ObjectBlock().BatchCreate(tx, newBlks)
		if err != nil {
			return fmt.Errorf("batch creating object blocks: %w", err)
		}

		for i, cloning := range avaiClonings {
			ret[cloning.OrgIndex] = &newObjs[i]
		}

		for i, cloning := range avaiClonings {
			var evtBlks []datamap.BlockDistributionObjectInfo
			blkType := getBlockTypeFromRed(newObjs[i].Redundancy)

			oldBlks := avaiDetailsMap[cloning.Cloning.ObjectID].Blocks
			for _, blk := range oldBlks {
				evtBlks = append(evtBlks, datamap.BlockDistributionObjectInfo{
					BlockType:   blkType,
					Index:       blk.Index,
					UserSpaceID: blk.UserSpaceID,
				})
			}

			evt = append(evt, &datamap.BodyNewOrUpdateObject{
				Info:              newObjs[i],
				BlockDistribution: evtBlks,
			})
		}
		return nil
	})

	if err != nil {
		logger.Warnf("cloning objects: %s", err.Error())
		return nil, err
	}

	for _, e := range evt {
		svc.EvtPub.Publish(e)
	}

	return ret, nil
}

// GetPackageObjects 获取包中的对象列表。
// userID: 用户ID。
// packageID: 包ID。
// 返回值: 对象列表和错误信息。
func (svc *ObjectService) GetPackageObjects(packageID types.PackageID) ([]types.Object, error) {
	return svc.DB.Object().GetPackageObjects(svc.DB.DefCtx(), packageID)
}

func (svc *ObjectService) GetObjectDetails(objectIDs []types.ObjectID) ([]*types.ObjectDetail, error) {
	detailsMp := make(map[types.ObjectID]*types.ObjectDetail)

	err := svc.DB.DoTx(func(tx db.SQLContext) error {
		var err error

		objectIDs = sort2.SortAsc(objectIDs)

		// 根据ID依次查询Object，ObjectBlock，PinnedObject，并根据升序的特点进行合并
		objs, err := svc.DB.Object().BatchGet(tx, objectIDs)
		if err != nil {
			return fmt.Errorf("batch get objects: %w", err)
		}
		for _, obj := range objs {
			detailsMp[obj.ObjectID] = &types.ObjectDetail{
				Object: obj,
			}
		}

		// 查询合并
		blocks, err := svc.DB.ObjectBlock().BatchGetByObjectID(tx, objectIDs)
		if err != nil {
			return fmt.Errorf("batch get object blocks: %w", err)
		}
		for _, block := range blocks {
			d := detailsMp[block.ObjectID]
			d.Blocks = append(d.Blocks, block)
		}

		// 查询合并
		pinneds, err := svc.DB.PinnedObject().BatchGetByObjectID(tx, objectIDs)
		if err != nil {
			return fmt.Errorf("batch get pinned objects: %w", err)
		}
		for _, pinned := range pinneds {
			d := detailsMp[pinned.ObjectID]
			d.PinnedAt = append(d.PinnedAt, pinned.UserSpaceID)
		}

		return nil
	})

	if err != nil {
		logger.Warn(err.Error())
		return nil, err
	}

	details := make([]*types.ObjectDetail, len(objectIDs))
	for i, objID := range objectIDs {
		details[i] = detailsMp[objID]
	}

	return details, nil
}

func (svc *ObjectService) NewMultipartUploadObject(packageID types.PackageID, path string) (types.Object, error) {
	var obj types.Object
	err := svc.DB.DoTx(func(tx db.SQLContext) error {
		oldObj, err := svc.DB.Object().GetByPath(tx, packageID, path)
		if err == nil {
			obj = oldObj
			err := svc.DB.ObjectBlock().DeleteByObjectID(tx, obj.ObjectID)
			if err != nil {
				return fmt.Errorf("delete object blocks: %w", err)
			}

			obj.FileHash = types.EmptyHash
			obj.Size = 0
			obj.Redundancy = types.NewMultipartUploadRedundancy()
			obj.UpdateTime = time.Now()

			err = svc.DB.Object().BatchUpdate(tx, []types.Object{obj})
			if err != nil {
				return fmt.Errorf("update object: %w", err)
			}

			return nil
		}

		obj = types.Object{
			PackageID:  packageID,
			Path:       path,
			FileHash:   types.EmptyHash,
			Size:       0,
			Redundancy: types.NewMultipartUploadRedundancy(),
			CreateTime: time.Now(),
			UpdateTime: time.Now(),
		}
		objID, err := svc.DB.Object().Create(tx, obj)
		if err != nil {
			return fmt.Errorf("create object: %w", err)
		}

		obj.ObjectID = objID
		return nil
	})
	if err != nil {
		logger.Warnf("new multipart upload object: %s", err.Error())
		return types.Object{}, err
	}

	return obj, nil
}

func (svc *ObjectService) CompleteMultipartUpload(objectID types.ObjectID, indexes []int) (types.Object, error) {
	if len(indexes) == 0 {
		return types.Object{}, fmt.Errorf("no block indexes specified")
	}

	objDe, err := db.DoTx11(svc.DB, svc.DB.Object().GetDetail, objectID)
	if err != nil {
		return types.Object{}, err
	}

	_, ok := objDe.Object.Redundancy.(*types.MultipartUploadRedundancy)
	if !ok {
		return types.Object{}, fmt.Errorf("object %v is not a multipart upload", objectID)
	}

	if len(objDe.Blocks) == 0 {
		return types.Object{}, fmt.Errorf("object %v has no blocks", objectID)
	}

	objBlkMap := make(map[int]types.ObjectBlock)
	for _, blk := range objDe.Blocks {
		objBlkMap[blk.Index] = blk
	}

	lockBld := reqbuilder.NewBuilder()

	var compBlks []types.ObjectBlock
	var compBlkSpaces []types.UserSpaceDetail
	var targetSpace types.UserSpaceDetail
	for i, idx := range indexes {
		blk, ok := objBlkMap[idx]
		if !ok {
			return types.Object{}, fmt.Errorf("block %d not found in object %v", idx, objectID)
		}

		stg := svc.UserSpaceMeta.Get(blk.UserSpaceID)
		if stg == nil {
			return types.Object{}, fmt.Errorf("storage of user space %d not found", blk.UserSpaceID)
		}

		compBlks = append(compBlks, blk)
		compBlkSpaces = append(compBlkSpaces, *stg)
		if i == 0 {
			targetSpace = *stg
		}
		lockBld.UserSpace().Buzy(stg.UserSpace.UserSpaceID)
	}

	mutex, err := lockBld.MutexLock(svc.PubLock)
	if err != nil {
		return types.Object{}, fmt.Errorf("acquire lock: %w", err)
	}
	defer mutex.Unlock()

	bld := exec.NewPlanBuilder()
	err = plans.CompleteMultipart(compBlks, compBlkSpaces, targetSpace, "shard", bld)
	if err != nil {
		return types.Object{}, err
	}

	exeCtx := exec.NewExecContext()
	exec.SetValueByType(exeCtx, svc.StgPool)
	ret, err := bld.Execute(exeCtx).Wait(context.Background())
	if err != nil {
		return types.Object{}, err
	}

	shardInfo := ret["shard"].(*ops2.FileInfoValue)

	err = db.DoTx10(svc.DB, svc.DB.Object().BatchUpdateRedundancy, []db.UpdatingObjectRedundancy{
		{
			ObjectID:   objectID,
			FileHash:   shardInfo.Hash,
			Size:       shardInfo.Size,
			Redundancy: types.NewNoneRedundancy(),
			Blocks: []types.ObjectBlock{{
				ObjectID:    objectID,
				Index:       0,
				UserSpaceID: targetSpace.UserSpace.UserSpaceID,
				FileHash:    shardInfo.Hash,
				Size:        shardInfo.Size,
			}},
		},
	})

	if err != nil {
		return types.Object{}, err
	}

	obj, err := svc.DB.Object().GetByID(svc.DB.DefCtx(), objectID)
	if err != nil {
		return types.Object{}, err
	}

	return obj, nil
}