feat(fs): add cross-storage move support (#211)

* feat(fs): add cross-storage move support

* fix(fs): add check before moving files

* fix(fs): changed error detect method

---------

Co-authored-by: ShenLin <773933146@qq.com>

internal/bootstrap/data/task.go

@@ -22,6 +22,7 @@ func initTasks() {
 func InitialTasks() []model.TaskItem {
 	initialTaskItems = []model.TaskItem{
 		{Key: "copy", PersistData: "[]"},
+		{Key: "move", PersistData: "[]"},
 		{Key: "download", PersistData: "[]"},
 		{Key: "transfer", PersistData: "[]"},
 	}

internal/bootstrap/task.go

@@ -26,6 +26,10 @@ func InitTaskManager() {
 	op.RegisterSettingChangingCallback(func() {
 		fs.CopyTaskManager.SetWorkersNumActive(taskFilterNegative(setting.GetInt(conf.TaskCopyThreadsNum, conf.Conf.Tasks.Copy.Workers)))
 	})
+	fs.MoveTaskManager = tache.NewManager[*fs.MoveTask](tache.WithWorks(setting.GetInt(conf.TaskMoveThreadsNum, conf.Conf.Tasks.Move.Workers)), tache.WithPersistFunction(db.GetTaskDataFunc("move", conf.Conf.Tasks.Move.TaskPersistant), db.UpdateTaskDataFunc("move", conf.Conf.Tasks.Move.TaskPersistant)), tache.WithMaxRetry(conf.Conf.Tasks.Move.MaxRetry))
+	op.RegisterSettingChangingCallback(func() {
+		fs.MoveTaskManager.SetWorkersNumActive(taskFilterNegative(setting.GetInt(conf.TaskMoveThreadsNum, conf.Conf.Tasks.Move.Workers)))
+	})
 	tool.DownloadTaskManager = tache.NewManager[*tool.DownloadTask](tache.WithWorks(setting.GetInt(conf.TaskOfflineDownloadThreadsNum, conf.Conf.Tasks.Download.Workers)), tache.WithPersistFunction(db.GetTaskDataFunc("download", conf.Conf.Tasks.Download.TaskPersistant), db.UpdateTaskDataFunc("download", conf.Conf.Tasks.Download.TaskPersistant)), tache.WithMaxRetry(conf.Conf.Tasks.Download.MaxRetry))
 	op.RegisterSettingChangingCallback(func() {
 		tool.DownloadTaskManager.SetWorkersNumActive(taskFilterNegative(setting.GetInt(conf.TaskOfflineDownloadThreadsNum, conf.Conf.Tasks.Download.Workers)))

internal/conf/config.go

@@ -58,6 +58,7 @@ type TasksConfig struct {
 	Transfer           TaskConfig `json:"transfer" envPrefix:"TRANSFER_"`
 	Upload             TaskConfig `json:"upload" envPrefix:"UPLOAD_"`
 	Copy               TaskConfig `json:"copy" envPrefix:"COPY_"`
+	Move               TaskConfig `json:"move" envPrefix:"MOVE_"`
 	Decompress         TaskConfig `json:"decompress" envPrefix:"DECOMPRESS_"`
 	DecompressUpload   TaskConfig `json:"decompress_upload" envPrefix:"DECOMPRESS_UPLOAD_"`
 	AllowRetryCanceled bool       `json:"allow_retry_canceled" env:"ALLOW_RETRY_CANCELED"`
@@ -175,6 +176,11 @@ func DefaultConfig() *Config {
 				MaxRetry: 2,
 				// TaskPersistant: true,
 			},
+			Move: TaskConfig{
+				Workers:  5,
+				MaxRetry: 2,
+				// TaskPersistant: true,
+			},
 			Decompress: TaskConfig{
 				Workers:  5,
 				MaxRetry: 2,

internal/conf/const.go

@@ -123,6 +123,7 @@ const (
 	TaskOfflineDownloadTransferThreadsNum = "offline_download_transfer_task_threads_num"
 	TaskUploadThreadsNum                  = "upload_task_threads_num"
 	TaskCopyThreadsNum                    = "copy_task_threads_num"
+	TaskMoveThreadsNum                    = "move_task_threads_num"
 	TaskDecompressDownloadThreadsNum      = "decompress_download_task_threads_num"
 	TaskDecompressUploadThreadsNum        = "decompress_upload_task_threads_num"
 	StreamMaxClientDownloadSpeed          = "max_client_download_speed"

internal/fs/fs.go

@@ -74,6 +74,14 @@ func Move(ctx context.Context, srcPath, dstDirPath string, lazyCache ...bool) er
 	return err
 }
 
+func MoveWithTask(ctx context.Context, srcPath, dstDirPath string, lazyCache ...bool) (task.TaskExtensionInfo, error) {
+	res, err := _move(ctx, srcPath, dstDirPath, lazyCache...)
+	if err != nil {
+		log.Errorf("failed move %s to %s: %+v", srcPath, dstDirPath, err)
+	}
+	return res, err
+}
+
 func Copy(ctx context.Context, srcObjPath, dstDirPath string, lazyCache ...bool) (task.TaskExtensionInfo, error) {
 	res, err := _copy(ctx, srcObjPath, dstDirPath, lazyCache...)
 	if err != nil {

internal/fs/move.go

@@ -0,0 +1,190 @@
+package fs
+
+import (
+	"context"
+	"fmt"
+	stdpath "path"
+	"time"
+
+	"github.com/OpenListTeam/OpenList/internal/driver"
+	"github.com/OpenListTeam/OpenList/internal/errs"
+	"github.com/OpenListTeam/OpenList/internal/model"
+	"github.com/OpenListTeam/OpenList/internal/op"
+	"github.com/OpenListTeam/OpenList/internal/task"
+	"github.com/OpenListTeam/OpenList/pkg/utils"
+	"github.com/pkg/errors"
+	"github.com/xhofe/tache"
+)
+
+type MoveTask struct {
+	task.TaskExtension
+	Status       string        `json:"-"`
+	SrcObjPath   string        `json:"src_path"`
+	DstDirPath   string        `json:"dst_path"`
+	srcStorage   driver.Driver `json:"-"`
+	dstStorage   driver.Driver `json:"-"`
+	SrcStorageMp string        `json:"src_storage_mp"`
+	DstStorageMp string        `json:"dst_storage_mp"`
+}
+
+func (t *MoveTask) GetName() string {
+	return fmt.Sprintf("move [%s](%s) to [%s](%s)", t.SrcStorageMp, t.SrcObjPath, t.DstStorageMp, t.DstDirPath)
+}
+
+func (t *MoveTask) GetStatus() string {
+	return t.Status
+}
+
+func (t *MoveTask) Run() error {
+	t.ReinitCtx()
+	t.ClearEndTime()
+	t.SetStartTime(time.Now())
+	defer func() { t.SetEndTime(time.Now()) }()
+	var err error
+	if t.srcStorage == nil {
+		t.srcStorage, err = op.GetStorageByMountPath(t.SrcStorageMp)
+	}
+	if t.dstStorage == nil {
+		t.dstStorage, err = op.GetStorageByMountPath(t.DstStorageMp)
+	}
+	if err != nil {
+		return errors.WithMessage(err, "failed get storage")
+	}
+
+	return moveBetween2Storages(t, t.srcStorage, t.dstStorage, t.SrcObjPath, t.DstDirPath)
+}
+
+var MoveTaskManager *tache.Manager[*MoveTask]
+
+
+func moveBetween2Storages(t *MoveTask, srcStorage, dstStorage driver.Driver, srcObjPath, dstDirPath string) error {
+	t.Status = "getting src object"
+	srcObj, err := op.Get(t.Ctx(), srcStorage, srcObjPath)
+	if err != nil {
+		return errors.WithMessagef(err, "failed get src [%s] file", srcObjPath)
+	}
+	
+	if srcObj.IsDir() {
+		t.Status = "src object is dir, listing objs"
+		objs, err := op.List(t.Ctx(), srcStorage, srcObjPath, model.ListArgs{})
+		if err != nil {
+			return errors.WithMessagef(err, "failed list src [%s] objs", srcObjPath)
+		}
+		
+		dstObjPath := stdpath.Join(dstDirPath, srcObj.GetName())
+		t.Status = "creating destination directory"
+		err = op.MakeDir(t.Ctx(), dstStorage, dstObjPath)
+		if err != nil {
+			// Check if this is an upload-related error and provide a clearer message
+			if errors.Is(err, errs.UploadNotSupported) {
+				return errors.WithMessagef(err, "destination storage [%s] does not support creating directories", dstStorage.GetStorage().MountPath)
+			}
+			return errors.WithMessagef(err, "failed to create destination directory [%s] in storage [%s]", dstObjPath, dstStorage.GetStorage().MountPath)
+		}
+		
+		for _, obj := range objs {
+			if utils.IsCanceled(t.Ctx()) {
+				return nil
+			}
+			srcSubObjPath := stdpath.Join(srcObjPath, obj.GetName())
+			MoveTaskManager.Add(&MoveTask{
+				TaskExtension: task.TaskExtension{
+					Creator: t.GetCreator(),
+				},
+				srcStorage:   srcStorage,
+				dstStorage:   dstStorage,
+				SrcObjPath:   srcSubObjPath,
+				DstDirPath:   dstObjPath,
+				SrcStorageMp: srcStorage.GetStorage().MountPath,
+				DstStorageMp: dstStorage.GetStorage().MountPath,
+			})
+		}
+
+		t.Status = "cleaning up source directory"
+		err = op.Remove(t.Ctx(), srcStorage, srcObjPath)
+		if err != nil {
+			t.Status = "completed (source directory cleanup pending)"
+		} else {
+			t.Status = "completed"
+		}
+		return nil
+	} else {
+		return moveFileBetween2Storages(t, srcStorage, dstStorage, srcObjPath, dstDirPath)
+	}
+}
+
+
+func moveFileBetween2Storages(tsk *MoveTask, srcStorage, dstStorage driver.Driver, srcFilePath, dstDirPath string) error {
+	tsk.Status = "copying file to destination"
+	
+	copyTask := &CopyTask{
+		TaskExtension: task.TaskExtension{
+			Creator: tsk.GetCreator(),
+		},
+		srcStorage:   srcStorage,
+		dstStorage:   dstStorage,
+		SrcObjPath:   srcFilePath,
+		DstDirPath:   dstDirPath,
+		SrcStorageMp: srcStorage.GetStorage().MountPath,
+		DstStorageMp: dstStorage.GetStorage().MountPath,
+	}
+	
+
+	copyTask.SetCtx(tsk.Ctx())
+	
+
+	err := copyBetween2Storages(copyTask, srcStorage, dstStorage, srcFilePath, dstDirPath)
+	if err != nil {
+		// Check if this is an upload-related error and provide a clearer message
+		if errors.Is(err, errs.UploadNotSupported) {
+			return errors.WithMessagef(err, "destination storage [%s] does not support file uploads", dstStorage.GetStorage().MountPath)
+		}
+		return errors.WithMessagef(err, "failed to copy [%s] to destination storage [%s]", srcFilePath, dstStorage.GetStorage().MountPath)
+	}
+	
+	tsk.SetProgress(50)
+	
+	tsk.Status = "deleting source file"
+	err = op.Remove(tsk.Ctx(), srcStorage, srcFilePath)
+	if err != nil {
+		return errors.WithMessagef(err, "failed to delete src [%s] file from storage [%s] after successful copy", srcFilePath, srcStorage.GetStorage().MountPath)
+	}
+	
+	tsk.SetProgress(100)
+	tsk.Status = "completed"
+	return nil
+}
+
+
+func _move(ctx context.Context, srcObjPath, dstDirPath string, lazyCache ...bool) (task.TaskExtensionInfo, error) {
+	srcStorage, srcObjActualPath, err := op.GetStorageAndActualPath(srcObjPath)
+	if err != nil {
+		return nil, errors.WithMessage(err, "failed get src storage")
+	}
+	dstStorage, dstDirActualPath, err := op.GetStorageAndActualPath(dstDirPath)
+	if err != nil {
+		return nil, errors.WithMessage(err, "failed get dst storage")
+	}
+
+	if srcStorage.GetStorage() == dstStorage.GetStorage() {
+		err = op.Move(ctx, srcStorage, srcObjActualPath, dstDirActualPath, lazyCache...)
+		if !errors.Is(err, errs.NotImplement) && !errors.Is(err, errs.NotSupport) {
+			return nil, err
+		}
+	}
+
+	taskCreator, _ := ctx.Value("user").(*model.User)
+	t := &MoveTask{
+		TaskExtension: task.TaskExtension{
+			Creator: taskCreator,
+		},
+		srcStorage:   srcStorage,
+		dstStorage:   dstStorage,
+		SrcObjPath:   srcObjActualPath,
+		DstDirPath:   dstDirActualPath,
+		SrcStorageMp: srcStorage.GetStorage().MountPath,
+		DstStorageMp: dstStorage.GetStorage().MountPath,
+	}
+	MoveTaskManager.Add(t)
+	return t, nil
+}

server/handles/fsmanage.go

@@ -96,14 +96,24 @@ func FsMove(c *gin.Context) {
 			}
 		}
 	}
+	var addedTasks []task.TaskExtensionInfo
 	for i, name := range req.Names {
-		err := fs.Move(c, stdpath.Join(srcDir, name), dstDir, len(req.Names) > i+1)
+		t, err := fs.MoveWithTask(c, stdpath.Join(srcDir, name), dstDir, len(req.Names) > i+1)
+		if t != nil {
+			addedTasks = append(addedTasks, t)
+		}
 		if err != nil {
 			common.ErrorResp(c, err, 500)
 			return
 		}
 	}
+	if len(addedTasks) > 0 {
+		common.SuccessResp(c, gin.H{
+			"tasks": getTaskInfos(addedTasks),
+		})
+	} else {
 		common.SuccessResp(c)
+	}
 }
 
 func FsCopy(c *gin.Context) {

server/handles/task.go

@@ -219,6 +219,7 @@ func taskRoute[T task.TaskExtensionInfo](g *gin.RouterGroup, manager task.Manage
 func SetupTaskRoute(g *gin.RouterGroup) {
 	taskRoute(g.Group("/upload"), fs.UploadTaskManager)
 	taskRoute(g.Group("/copy"), fs.CopyTaskManager)
+	taskRoute(g.Group("/move"), fs.MoveTaskManager)
 	taskRoute(g.Group("/offline_download"), tool.DownloadTaskManager)
 	taskRoute(g.Group("/offline_download_transfer"), tool.TransferTaskManager)
 	taskRoute(g.Group("/decompress"), fs.ArchiveDownloadTaskManager)