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

package services

import (
	"context"
	"fmt"
	"path"
	"strings"

	"gitlink.org.cn/cloudream/common/pkgs/ioswitch/exec"
	"gitlink.org.cn/cloudream/common/pkgs/logger"
	"gitlink.org.cn/cloudream/common/pkgs/trie"
	cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
	clitypes "gitlink.org.cn/cloudream/jcs-pub/client/types"
	"gorm.io/gorm"

	"gitlink.org.cn/cloudream/jcs-pub/client/internal/db"
	"gitlink.org.cn/cloudream/jcs-pub/client/internal/downloader/strategy"
	cliapi "gitlink.org.cn/cloudream/jcs-pub/client/sdk/api/v1"
	"gitlink.org.cn/cloudream/jcs-pub/common/ecode"
	stgglb "gitlink.org.cn/cloudream/jcs-pub/common/globals"
	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/ioswitch2"
	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/ioswitch2/parser"
	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/publock/reqbuilder"
	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/storage/factory"
	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/storage/types"
)

type UserSpaceService struct {
	*Service
}

func (svc *Service) UserSpaceSvc() *UserSpaceService {
	return &UserSpaceService{Service: svc}
}

func (svc *UserSpaceService) Get(userspaceID clitypes.UserSpaceID) (clitypes.UserSpace, error) {
	return svc.DB.UserSpace().GetByID(svc.DB.DefCtx(), userspaceID)
}

func (svc *UserSpaceService) GetByName(name string) (clitypes.UserSpace, error) {
	return svc.DB.UserSpace().GetByName(svc.DB.DefCtx(), name)
}

func (svc *UserSpaceService) GetAll() ([]clitypes.UserSpace, error) {
	return svc.DB.UserSpace().GetAll(svc.DB.DefCtx())
}

func (svc *UserSpaceService) Create(req cliapi.UserSpaceCreate) (*cliapi.UserSpaceCreateResp, *ecode.CodeError) {
	db2 := svc.DB
	space, err := db.DoTx01(db2, func(tx db.SQLContext) (clitypes.UserSpace, error) {
		space, err := db2.UserSpace().GetByName(tx, req.Name)
		if err == nil {
			return clitypes.UserSpace{}, gorm.ErrDuplicatedKey
		}
		if err != gorm.ErrRecordNotFound {
			return clitypes.UserSpace{}, err
		}

		space = clitypes.UserSpace{
			Name:       req.Name,
			Storage:    req.Storage,
			Credential: req.Credential,
			ShardStore: req.ShardStore,
			Features:   req.Features,
			WorkingDir: req.WorkingDir,
			Revision:   0,
		}
		err = db2.UserSpace().Create(tx, &space)
		if err != nil {
			return clitypes.UserSpace{}, err
		}
		return space, nil
	})
	if err == gorm.ErrDuplicatedKey {
		return nil, ecode.New(ecode.DataExists, "user space name already exists")
	}
	if err != nil {
		return nil, ecode.Newf(ecode.OperationFailed, "%v", err)
	}
	return &cliapi.UserSpaceCreateResp{UserSpace: space}, nil
}

func (svc *UserSpaceService) Update(req cliapi.UserSpaceUpdate) (*cliapi.UserSpaceUpdateResp, *ecode.CodeError) {
	db2 := svc.DB
	space, err := db.DoTx01(db2, func(tx db.SQLContext) (clitypes.UserSpace, error) {
		space, err := db2.UserSpace().GetByID(tx, req.UserSpaceID)
		if err != nil {
			return clitypes.UserSpace{}, err
		}

		if space.Name != req.Name {
			_, err = db2.UserSpace().GetByName(tx, req.Name)
			if err == nil {
				return clitypes.UserSpace{}, gorm.ErrDuplicatedKey
			}
			if err != gorm.ErrRecordNotFound {
				return clitypes.UserSpace{}, err
			}
		}

		space.Name = req.Name
		space.Credential = req.Credential
		space.Features = req.Features
		space.Revision += 1
		return space, db2.UserSpace().UpdateColumns(tx, space, "Name", "Credential", "Features", "Revision")
	})
	if err == gorm.ErrDuplicatedKey {
		return nil, ecode.New(ecode.DataExists, "user space name already exists")
	}
	if err != nil {
		return nil, ecode.Newf(ecode.OperationFailed, "%v", err)
	}

	// 通知元数据缓存无效
	svc.UserSpaceMeta.Drop([]clitypes.UserSpaceID{req.UserSpaceID})

	// 通知存储服务组件池停止组件。TODO 对于在Hub上运行的组件，需要一个机制去定时清理
	svc.StgPool.Drop(stgglb.UserID, space.UserSpaceID)

	// TODO 考虑加锁再进行操作

	return &cliapi.UserSpaceUpdateResp{UserSpace: space}, nil
}

func (svc *UserSpaceService) Delete(req cliapi.UserSpaceDelete) (*cliapi.UserSpaceDeleteResp, *ecode.CodeError) {
	db2 := svc.DB
	err := db2.DoTx(func(tx db.SQLContext) error {
		err := db2.UserSpace().DeleteByID(tx, req.UserSpaceID)
		if err != nil {
			return err
		}

		err = db2.ObjectBlock().DeleteByUserSpaceID(tx, req.UserSpaceID)
		if err != nil {
			return err
		}

		err = db2.PinnedObject().DeleteByUserSpaceID(tx, req.UserSpaceID)
		if err != nil {
			return err
		}

		err = db2.ObjectAccessStat().DeleteByUserSpaceID(tx, req.UserSpaceID)
		if err != nil {
			return err
		}

		err = db2.PackageAccessStat().DeleteByUserSpaceID(tx, req.UserSpaceID)
		if err != nil {
			return err
		}

		return nil
	})
	if err != nil {
		return nil, ecode.Newf(ecode.OperationFailed, "%v", err)
	}

	// 通知元数据缓存无效
	svc.UserSpaceMeta.Drop([]clitypes.UserSpaceID{req.UserSpaceID})

	// 通知存储服务组件池停止组件。TODO 对于在Hub上运行的组件，需要一个机制去定时清理
	svc.StgPool.Drop(stgglb.UserID, req.UserSpaceID)

	// TODO 考虑加锁再进行操作，并且增加机制打断已经在进行的操作。

	return &cliapi.UserSpaceDeleteResp{}, nil
}

func (svc *UserSpaceService) Test(req cliapi.UserSpaceTest) (*cliapi.UserSpaceTestResp, *ecode.CodeError) {
	detail := clitypes.UserSpaceDetail{
		UserID: stgglb.UserID,
		UserSpace: clitypes.UserSpace{
			Name:       "test",
			Storage:    req.Storage,
			Credential: req.Credential,
			WorkingDir: req.WorikingDir,
		},
	}
	blder := factory.GetBuilder(&detail)
	baseStore, err := blder.CreateBaseStore(false)
	if err != nil {
		return nil, ecode.Newf(ecode.OperationFailed, "%v", err)
	}

	// TODO 可以考虑增加一个专门用于检查配置的接口F
	_, err = baseStore.ListAll("")
	if err != nil {
		return nil, ecode.Newf(ecode.OperationFailed, "%v", err)
	}

	return &cliapi.UserSpaceTestResp{}, nil
}

func (svc *UserSpaceService) DownloadPackage(packageID clitypes.PackageID, userspaceID clitypes.UserSpaceID, rootPath string) error {
	coorCli := stgglb.CoordinatorRPCPool.Get()
	defer coorCli.Release()

	destStg := svc.UserSpaceMeta.Get(userspaceID)
	if destStg == nil {
		return fmt.Errorf("userspace not found: %d", userspaceID)
	}

	details, err := db.DoTx11(svc.DB, svc.DB.Object().GetPackageObjectDetails, packageID)
	if err != nil {
		return err
	}

	var pinned []clitypes.ObjectID
	plans := exec.NewPlanBuilder()
	for _, obj := range details {
		strg, err := svc.StrategySelector.Select(strategy.Request{
			Detail:       obj,
			DestLocation: destStg.UserSpace.Storage.GetLocation(),
		})
		if err != nil {
			return fmt.Errorf("select download strategy: %w", err)
		}

		ft := ioswitch2.NewFromTo()
		switch strg := strg.(type) {
		case *strategy.DirectStrategy:
			ft.AddFrom(ioswitch2.NewFromShardstore(strg.Detail.Object.FileHash, strg.UserSpace, ioswitch2.RawStream()))

		case *strategy.ECReconstructStrategy:
			for i, b := range strg.Blocks {
				ft.AddFrom(ioswitch2.NewFromShardstore(b.FileHash, strg.UserSpaces[i], ioswitch2.ECStream(b.Index)))
				ft.ECParam = &strg.Redundancy
			}
		default:
			return fmt.Errorf("unsupported download strategy: %T", strg)
		}

		ft.AddTo(ioswitch2.NewToBaseStore(*destStg, path.Join(rootPath, obj.Object.Path)))
		// 顺便保存到同存储服务的分片存储中
		if destStg.UserSpace.ShardStore != nil {
			ft.AddTo(ioswitch2.NewToShardStore(*destStg, ioswitch2.RawStream(), ""))
			pinned = append(pinned, obj.Object.ObjectID)
		}

		err = parser.Parse(ft, plans)
		if err != nil {
			return fmt.Errorf("parse plan: %w", err)
		}
	}

	mutex, err := reqbuilder.NewBuilder().
		UserSpace().Buzy(userspaceID).
		MutexLock(svc.PubLock)
	if err != nil {
		return fmt.Errorf("acquire locks failed, err: %w", err)
	}
	defer mutex.Unlock()

	// 记录访问统计
	for _, obj := range details {
		svc.AccessStat.AddAccessCounter(obj.Object.ObjectID, packageID, userspaceID, 1)
	}
	exeCtx := exec.NewExecContext()
	exec.SetValueByType(exeCtx, svc.StgPool)
	drv := plans.Execute(exeCtx)
	_, err = drv.Wait(context.Background())
	if err != nil {
		return err
	}

	return nil
}

func (svc *UserSpaceService) SpaceToSpace(srcSpaceID clitypes.UserSpaceID, srcPath string, dstSpaceID clitypes.UserSpaceID, dstPath string) (clitypes.SpaceToSpaceResult, error) {
	srcSpace := svc.UserSpaceMeta.Get(srcSpaceID)
	if srcSpace == nil {
		return clitypes.SpaceToSpaceResult{}, fmt.Errorf("source userspace not found: %d", srcSpaceID)
	}

	srcStore, err := svc.StgPool.GetBaseStore(srcSpace)
	if err != nil {
		return clitypes.SpaceToSpaceResult{}, fmt.Errorf("get source userspace store: %w", err)
	}

	dstSpace := svc.UserSpaceMeta.Get(dstSpaceID)
	if dstSpace == nil {
		return clitypes.SpaceToSpaceResult{}, fmt.Errorf("destination userspace not found: %d", dstSpaceID)
	}

	dstStore, err := svc.StgPool.GetBaseStore(dstSpace)
	if err != nil {
		return clitypes.SpaceToSpaceResult{}, fmt.Errorf("get destination userspace store: %w", err)
	}

	srcPath = strings.Trim(srcPath, cdssdk.ObjectPathSeparator)
	dstPath = strings.Trim(dstPath, cdssdk.ObjectPathSeparator)

	if srcPath == "" {
		return clitypes.SpaceToSpaceResult{}, fmt.Errorf("source path is empty")
	}

	if dstPath == "" {
		return clitypes.SpaceToSpaceResult{}, fmt.Errorf("destination path is empty")
	}

	entries, cerr := srcStore.ListAll(srcPath)
	if cerr != nil {
		return clitypes.SpaceToSpaceResult{}, fmt.Errorf("list all from source userspace: %w", cerr)
	}

	srcPathComps := clitypes.SplitObjectPath(srcPath)
	srcDirCompLen := len(srcPathComps) - 1

	entryTree := trie.NewTrie[*types.ListEntry]()
	for _, e := range entries {
		pa, ok := strings.CutSuffix(e.Path, clitypes.ObjectPathSeparator)
		comps := clitypes.SplitObjectPath(pa)
		e.Path = pa

		e2 := e
		entryTree.CreateWords(comps[srcDirCompLen:]).Value = &e2
		e2.IsDir = e2.IsDir || ok
	}

	entryTree.Iterate(func(path []string, node *trie.Node[*types.ListEntry], isWordNode bool) trie.VisitCtrl {
		if node.Value == nil {
			return trie.VisitContinue
		}

		if node.Value.IsDir && len(node.WordNexts) > 0 {
			node.Value = nil
			return trie.VisitContinue
		}

		if !node.Value.IsDir && len(node.WordNexts) == 0 {
			node.WordNexts = nil
		}

		return trie.VisitContinue
	})

	var filePathes []string
	var dirPathes []string
	entryTree.Iterate(func(path []string, node *trie.Node[*types.ListEntry], isWordNode bool) trie.VisitCtrl {
		if node.Value == nil {
			return trie.VisitContinue
		}

		if node.Value.IsDir {
			dirPathes = append(dirPathes, node.Value.Path)
		} else {
			filePathes = append(filePathes, node.Value.Path)
		}

		return trie.VisitContinue
	})

	mutex, err := reqbuilder.NewBuilder().UserSpace().Buzy(srcSpaceID).Buzy(dstSpaceID).MutexLock(svc.PubLock)
	if err != nil {
		return clitypes.SpaceToSpaceResult{}, fmt.Errorf("acquire lock: %w", err)
	}
	defer mutex.Unlock()

	var success []string
	var failed []string

	for _, f := range filePathes {
		newPath := strings.Replace(f, srcPath, dstPath, 1)

		ft := ioswitch2.NewFromTo()
		ft.AddFrom(ioswitch2.NewFromBaseStore(*srcSpace, f))
		ft.AddTo(ioswitch2.NewToBaseStore(*dstSpace, newPath))

		plans := exec.NewPlanBuilder()
		err := parser.Parse(ft, plans)
		if err != nil {
			failed = append(failed, f)
			logger.Warnf("s2s: parse plan of file %v: %v", f, err)
			continue
		}
		exeCtx := exec.NewExecContext()
		exec.SetValueByType(exeCtx, svc.StgPool)
		_, cerr := plans.Execute(exeCtx).Wait(context.Background())
		if cerr != nil {
			failed = append(failed, f)
			logger.Warnf("s2s: execute plan of file %v: %v", f, cerr)
			continue
		}

		success = append(success, f)
	}

	newDirPathes := make([]string, 0, len(dirPathes))
	for i := range dirPathes {
		newDirPathes = append(newDirPathes, strings.Replace(dirPathes[i], srcPath, dstPath, 1))
	}

	for _, d := range newDirPathes {
		err := dstStore.Mkdir(d)
		if err != nil {
			failed = append(failed, d)
		} else {
			success = append(success, d)
		}
	}

	return clitypes.SpaceToSpaceResult{
		Success: success,
		Failed:  failed,
	}, nil
}