mirror of
https://github.com/OpenListTeam/OpenList.git
synced 2025-07-18 17:38:07 +08:00
refactor(fs):Refactor the delete function and fix known issues (#353)
* fix(move):Fix file move logic * fix(move):fixed move logic * fix(move):Fixed move logic * Fixed errs * fix(move):fixed movetask * fix(move):fix movetask * fixed * fix(move):Refactoring the move structure * fix(move):Fixed move system * Fixed * Fixed * Fixed * Fixed * Fixed * Rollback * Fixed and Refactor * fix(move):fixed * fix(move):Solve related performance issues * refacter(move):fixed build bugs
This commit is contained in:
@ -82,6 +82,14 @@ func MoveWithTask(ctx context.Context, srcPath, dstDirPath string, lazyCache ...
|
||||
return res, err
|
||||
}
|
||||
|
||||
func MoveWithTaskAndValidation(ctx context.Context, srcPath, dstDirPath string, validateExistence bool, lazyCache ...bool) (task.TaskExtensionInfo, error) {
|
||||
res, err := _moveWithValidation(ctx, srcPath, dstDirPath, validateExistence, 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 {
|
||||
|
@ -3,13 +3,16 @@ package fs
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
stdpath "path"
|
||||
"sync"
|
||||
"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/stream"
|
||||
"github.com/OpenListTeam/OpenList/internal/task"
|
||||
"github.com/OpenListTeam/OpenList/pkg/utils"
|
||||
"github.com/pkg/errors"
|
||||
@ -18,28 +21,101 @@ import (
|
||||
|
||||
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"`
|
||||
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"`
|
||||
IsRootTask bool `json:"is_root_task"`
|
||||
RootTaskID string `json:"root_task_id"`
|
||||
TotalFiles int `json:"total_files"`
|
||||
CompletedFiles int `json:"completed_files"`
|
||||
Phase string `json:"phase"` // "copying", "verifying", "deleting", "completed"
|
||||
ValidateExistence bool `json:"validate_existence"`
|
||||
mu sync.RWMutex `json:"-"`
|
||||
}
|
||||
|
||||
type MoveProgress struct {
|
||||
TaskID string `json:"task_id"`
|
||||
Phase string `json:"phase"`
|
||||
TotalFiles int `json:"total_files"`
|
||||
CompletedFiles int `json:"completed_files"`
|
||||
CurrentFile string `json:"current_file"`
|
||||
Status string `json:"status"`
|
||||
Progress int `json:"progress"`
|
||||
}
|
||||
|
||||
var moveProgressMap = sync.Map{}
|
||||
|
||||
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 {
|
||||
t.mu.RLock()
|
||||
defer t.mu.RUnlock()
|
||||
return t.Status
|
||||
}
|
||||
|
||||
func (t *MoveTask) GetProgress() float64 {
|
||||
t.mu.RLock()
|
||||
defer t.mu.RUnlock()
|
||||
|
||||
if t.TotalFiles == 0 {
|
||||
return 0
|
||||
}
|
||||
|
||||
switch t.Phase {
|
||||
case "copying":
|
||||
return float64(t.CompletedFiles*60) / float64(t.TotalFiles)
|
||||
case "verifying":
|
||||
return 60 + float64(t.CompletedFiles*20)/float64(t.TotalFiles)
|
||||
case "deleting":
|
||||
return 80 + float64(t.CompletedFiles*20)/float64(t.TotalFiles)
|
||||
case "completed":
|
||||
return 100
|
||||
default:
|
||||
return 0
|
||||
}
|
||||
}
|
||||
|
||||
func (t *MoveTask) GetMoveProgress() *MoveProgress {
|
||||
t.mu.RLock()
|
||||
defer t.mu.RUnlock()
|
||||
|
||||
progress := int(t.GetProgress())
|
||||
|
||||
return &MoveProgress{
|
||||
TaskID: t.GetID(),
|
||||
Phase: t.Phase,
|
||||
TotalFiles: t.TotalFiles,
|
||||
CompletedFiles: t.CompletedFiles,
|
||||
CurrentFile: t.SrcObjPath,
|
||||
Status: t.Status,
|
||||
Progress: progress,
|
||||
}
|
||||
}
|
||||
|
||||
func (t *MoveTask) updateProgress() {
|
||||
if t.IsRootTask {
|
||||
progress := t.GetMoveProgress()
|
||||
moveProgressMap.Store(t.GetID(), progress)
|
||||
}
|
||||
}
|
||||
|
||||
func (t *MoveTask) Run() error {
|
||||
t.ReinitCtx()
|
||||
t.ClearEndTime()
|
||||
t.SetStartTime(time.Now())
|
||||
defer func() { t.SetEndTime(time.Now()) }()
|
||||
defer func() {
|
||||
t.SetEndTime(time.Now())
|
||||
if t.IsRootTask {
|
||||
moveProgressMap.Delete(t.GetID())
|
||||
}
|
||||
}()
|
||||
|
||||
var err error
|
||||
if t.srcStorage == nil {
|
||||
t.srcStorage, err = op.GetStorageByMountPath(t.SrcStorageMp)
|
||||
@ -51,11 +127,337 @@ func (t *MoveTask) Run() error {
|
||||
return errors.WithMessage(err, "failed get storage")
|
||||
}
|
||||
|
||||
return moveBetween2Storages(t, t.srcStorage, t.dstStorage, t.SrcObjPath, t.DstDirPath)
|
||||
// Phase 1: Async validation (all validation happens in background)
|
||||
t.mu.Lock()
|
||||
t.Status = "validating source and destination"
|
||||
t.mu.Unlock()
|
||||
|
||||
// Check if source exists
|
||||
srcObj, err := op.Get(t.Ctx(), t.srcStorage, t.SrcObjPath)
|
||||
if err != nil {
|
||||
return errors.WithMessagef(err, "source file [%s] not found", stdpath.Base(t.SrcObjPath))
|
||||
}
|
||||
|
||||
// Check if destination already exists (if validation is required)
|
||||
if t.ValidateExistence {
|
||||
dstFilePath := stdpath.Join(t.DstDirPath, srcObj.GetName())
|
||||
if res, _ := op.Get(t.Ctx(), t.dstStorage, dstFilePath); res != nil {
|
||||
return errors.Errorf("destination file [%s] already exists", srcObj.GetName())
|
||||
}
|
||||
}
|
||||
|
||||
// Phase 2: Execute move operation with proper sequencing
|
||||
// Determine if we should use batch optimization for directories
|
||||
if srcObj.IsDir() {
|
||||
t.mu.Lock()
|
||||
t.IsRootTask = true
|
||||
t.RootTaskID = t.GetID()
|
||||
t.mu.Unlock()
|
||||
return t.runRootMoveTask()
|
||||
}
|
||||
|
||||
// Use safe move logic for files
|
||||
return t.safeMoveOperation(srcObj)
|
||||
}
|
||||
|
||||
func (t *MoveTask) runRootMoveTask() error {
|
||||
// First check if source is actually a directory
|
||||
// If not, fall back to regular move logic
|
||||
srcObj, err := op.Get(t.Ctx(), t.srcStorage, t.SrcObjPath)
|
||||
if err != nil {
|
||||
return errors.WithMessagef(err, "failed get src [%s] object", t.SrcObjPath)
|
||||
}
|
||||
|
||||
if !srcObj.IsDir() {
|
||||
// Source is not a directory, use regular move logic
|
||||
t.mu.Lock()
|
||||
t.IsRootTask = false
|
||||
t.mu.Unlock()
|
||||
return t.safeMoveOperation(srcObj)
|
||||
}
|
||||
|
||||
// Phase 1: Count total files and create directory structure
|
||||
t.mu.Lock()
|
||||
t.Phase = "preparing"
|
||||
t.Status = "counting files and preparing directory structure"
|
||||
t.mu.Unlock()
|
||||
t.updateProgress()
|
||||
|
||||
totalFiles, err := t.countFilesAndCreateDirs(t.srcStorage, t.dstStorage, t.SrcObjPath, t.DstDirPath)
|
||||
if err != nil {
|
||||
return errors.WithMessage(err, "failed to prepare directory structure")
|
||||
}
|
||||
|
||||
t.mu.Lock()
|
||||
t.TotalFiles = totalFiles
|
||||
t.Phase = "copying"
|
||||
t.Status = "copying files"
|
||||
t.mu.Unlock()
|
||||
t.updateProgress()
|
||||
|
||||
// Phase 2: Copy all files
|
||||
err = t.copyAllFiles(t.srcStorage, t.dstStorage, t.SrcObjPath, t.DstDirPath)
|
||||
if err != nil {
|
||||
return errors.WithMessage(err, "failed to copy files")
|
||||
}
|
||||
|
||||
// Phase 3: Verify directory structure
|
||||
t.mu.Lock()
|
||||
t.Phase = "verifying"
|
||||
t.Status = "verifying copied files"
|
||||
t.CompletedFiles = 0
|
||||
t.mu.Unlock()
|
||||
t.updateProgress()
|
||||
|
||||
err = t.verifyDirectoryStructure(t.srcStorage, t.dstStorage, t.SrcObjPath, t.DstDirPath)
|
||||
if err != nil {
|
||||
return errors.WithMessage(err, "verification failed")
|
||||
}
|
||||
|
||||
// Phase 4: Delete source files and directories
|
||||
t.mu.Lock()
|
||||
t.Phase = "deleting"
|
||||
t.Status = "deleting source files"
|
||||
t.CompletedFiles = 0
|
||||
t.mu.Unlock()
|
||||
t.updateProgress()
|
||||
|
||||
err = t.deleteSourceRecursively(t.srcStorage, t.SrcObjPath)
|
||||
if err != nil {
|
||||
return errors.WithMessage(err, "failed to delete source files")
|
||||
}
|
||||
|
||||
t.mu.Lock()
|
||||
t.Phase = "completed"
|
||||
t.Status = "completed"
|
||||
t.mu.Unlock()
|
||||
t.updateProgress()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
var MoveTaskManager *tache.Manager[*MoveTask]
|
||||
|
||||
// GetMoveProgress returns the progress of a move task by task ID
|
||||
func GetMoveProgress(taskID string) (*MoveProgress, bool) {
|
||||
if progress, ok := moveProgressMap.Load(taskID); ok {
|
||||
return progress.(*MoveProgress), true
|
||||
}
|
||||
return nil, false
|
||||
}
|
||||
|
||||
// GetMoveTaskProgress returns the progress of a specific move task
|
||||
func GetMoveTaskProgress(task *MoveTask) *MoveProgress {
|
||||
return task.GetMoveProgress()
|
||||
}
|
||||
|
||||
// countFilesAndCreateDirs recursively counts files and creates directory structure
|
||||
func (t *MoveTask) countFilesAndCreateDirs(srcStorage, dstStorage driver.Driver, srcPath, dstPath string) (int, error) {
|
||||
srcObj, err := op.Get(t.Ctx(), srcStorage, srcPath)
|
||||
if err != nil {
|
||||
return 0, errors.WithMessagef(err, "failed get src [%s] object", srcPath)
|
||||
}
|
||||
|
||||
if !srcObj.IsDir() {
|
||||
return 1, nil
|
||||
}
|
||||
|
||||
// Create destination directory
|
||||
dstObjPath := stdpath.Join(dstPath, srcObj.GetName())
|
||||
err = op.MakeDir(t.Ctx(), dstStorage, dstObjPath)
|
||||
if err != nil {
|
||||
if errors.Is(err, errs.UploadNotSupported) {
|
||||
return 0, errors.WithMessagef(err, "destination storage [%s] does not support creating directories", dstStorage.GetStorage().MountPath)
|
||||
}
|
||||
return 0, errors.WithMessagef(err, "failed to create destination directory [%s] in storage [%s]", dstObjPath, dstStorage.GetStorage().MountPath)
|
||||
}
|
||||
|
||||
// List and count files recursively
|
||||
objs, err := op.List(t.Ctx(), srcStorage, srcPath, model.ListArgs{})
|
||||
if err != nil {
|
||||
return 0, errors.WithMessagef(err, "failed list src [%s] objs", srcPath)
|
||||
}
|
||||
|
||||
totalFiles := 0
|
||||
for _, obj := range objs {
|
||||
if utils.IsCanceled(t.Ctx()) {
|
||||
return 0, nil
|
||||
}
|
||||
srcSubPath := stdpath.Join(srcPath, obj.GetName())
|
||||
subCount, err := t.countFilesAndCreateDirs(srcStorage, dstStorage, srcSubPath, dstObjPath)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
totalFiles += subCount
|
||||
}
|
||||
|
||||
return totalFiles, nil
|
||||
}
|
||||
|
||||
// copyAllFiles recursively copies all files
|
||||
func (t *MoveTask) copyAllFiles(srcStorage, dstStorage driver.Driver, srcPath, dstPath string) error {
|
||||
srcObj, err := op.Get(t.Ctx(), srcStorage, srcPath)
|
||||
if err != nil {
|
||||
return errors.WithMessagef(err, "failed get src [%s] object", srcPath)
|
||||
}
|
||||
|
||||
if !srcObj.IsDir() {
|
||||
// Copy single file
|
||||
err := t.copyFile(srcStorage, dstStorage, srcPath, dstPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
t.mu.Lock()
|
||||
t.CompletedFiles++
|
||||
t.mu.Unlock()
|
||||
t.updateProgress()
|
||||
return nil
|
||||
}
|
||||
|
||||
// Copy directory contents
|
||||
objs, err := op.List(t.Ctx(), srcStorage, srcPath, model.ListArgs{})
|
||||
if err != nil {
|
||||
return errors.WithMessagef(err, "failed list src [%s] objs", srcPath)
|
||||
}
|
||||
|
||||
dstObjPath := stdpath.Join(dstPath, srcObj.GetName())
|
||||
for _, obj := range objs {
|
||||
if utils.IsCanceled(t.Ctx()) {
|
||||
return nil
|
||||
}
|
||||
srcSubPath := stdpath.Join(srcPath, obj.GetName())
|
||||
err := t.copyAllFiles(srcStorage, dstStorage, srcSubPath, dstObjPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// copyFile copies a single file between storages
|
||||
func (t *MoveTask) copyFile(srcStorage, dstStorage driver.Driver, srcFilePath, dstDirPath string) error {
|
||||
srcFile, err := op.Get(t.Ctx(), srcStorage, srcFilePath)
|
||||
if err != nil {
|
||||
return errors.WithMessagef(err, "failed get src [%s] file", srcFilePath)
|
||||
}
|
||||
|
||||
link, _, err := op.Link(t.Ctx(), srcStorage, srcFilePath, model.LinkArgs{
|
||||
Header: http.Header{},
|
||||
})
|
||||
if err != nil {
|
||||
return errors.WithMessagef(err, "failed get [%s] link", srcFilePath)
|
||||
}
|
||||
|
||||
fs := stream.FileStream{
|
||||
Obj: srcFile,
|
||||
Ctx: t.Ctx(),
|
||||
}
|
||||
|
||||
ss, err := stream.NewSeekableStream(fs, link)
|
||||
if err != nil {
|
||||
return errors.WithMessagef(err, "failed get [%s] stream", srcFilePath)
|
||||
}
|
||||
|
||||
return op.Put(t.Ctx(), dstStorage, dstDirPath, ss, nil, true)
|
||||
}
|
||||
|
||||
// verifyDirectoryStructure compares source and destination directory structures
|
||||
func (t *MoveTask) verifyDirectoryStructure(srcStorage, dstStorage driver.Driver, srcPath, dstPath string) error {
|
||||
srcObj, err := op.Get(t.Ctx(), srcStorage, srcPath)
|
||||
if err != nil {
|
||||
return errors.WithMessagef(err, "failed get src [%s] object", srcPath)
|
||||
}
|
||||
|
||||
if !srcObj.IsDir() {
|
||||
// Verify single file
|
||||
dstFilePath := stdpath.Join(dstPath, srcObj.GetName())
|
||||
_, err := op.Get(t.Ctx(), dstStorage, dstFilePath)
|
||||
if err != nil {
|
||||
return errors.WithMessagef(err, "verification failed: destination file [%s] not found", dstFilePath)
|
||||
}
|
||||
|
||||
t.mu.Lock()
|
||||
t.CompletedFiles++
|
||||
t.mu.Unlock()
|
||||
t.updateProgress()
|
||||
return nil
|
||||
}
|
||||
|
||||
// Verify directory
|
||||
dstObjPath := stdpath.Join(dstPath, srcObj.GetName())
|
||||
_, err = op.Get(t.Ctx(), dstStorage, dstObjPath)
|
||||
if err != nil {
|
||||
return errors.WithMessagef(err, "verification failed: destination directory [%s] not found", dstObjPath)
|
||||
}
|
||||
|
||||
// Verify directory contents
|
||||
srcObjs, err := op.List(t.Ctx(), srcStorage, srcPath, model.ListArgs{})
|
||||
if err != nil {
|
||||
return errors.WithMessagef(err, "failed list src [%s] objs for verification", srcPath)
|
||||
}
|
||||
|
||||
for _, obj := range srcObjs {
|
||||
if utils.IsCanceled(t.Ctx()) {
|
||||
return nil
|
||||
}
|
||||
srcSubPath := stdpath.Join(srcPath, obj.GetName())
|
||||
err := t.verifyDirectoryStructure(srcStorage, dstStorage, srcSubPath, dstObjPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// deleteSourceRecursively deletes source files and directories recursively
|
||||
func (t *MoveTask) deleteSourceRecursively(srcStorage driver.Driver, srcPath string) error {
|
||||
srcObj, err := op.Get(t.Ctx(), srcStorage, srcPath)
|
||||
if err != nil {
|
||||
return errors.WithMessagef(err, "failed get src [%s] object for deletion", srcPath)
|
||||
}
|
||||
|
||||
if !srcObj.IsDir() {
|
||||
// Delete single file
|
||||
err := op.Remove(t.Ctx(), srcStorage, srcPath)
|
||||
if err != nil {
|
||||
return errors.WithMessagef(err, "failed to delete src [%s] file", srcPath)
|
||||
}
|
||||
|
||||
t.mu.Lock()
|
||||
t.CompletedFiles++
|
||||
t.mu.Unlock()
|
||||
t.updateProgress()
|
||||
return nil
|
||||
}
|
||||
|
||||
// Delete directory contents first
|
||||
objs, err := op.List(t.Ctx(), srcStorage, srcPath, model.ListArgs{})
|
||||
if err != nil {
|
||||
return errors.WithMessagef(err, "failed list src [%s] objs for deletion", srcPath)
|
||||
}
|
||||
|
||||
for _, obj := range objs {
|
||||
if utils.IsCanceled(t.Ctx()) {
|
||||
return nil
|
||||
}
|
||||
srcSubPath := stdpath.Join(srcPath, obj.GetName())
|
||||
err := t.deleteSourceRecursively(srcStorage, srcSubPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// Delete the directory itself
|
||||
err = op.Remove(t.Ctx(), srcStorage, srcPath)
|
||||
if err != nil {
|
||||
return errors.WithMessagef(err, "failed to delete src [%s] directory", srcPath)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func moveBetween2Storages(t *MoveTask, srcStorage, dstStorage driver.Driver, srcObjPath, dstDirPath string) error {
|
||||
t.Status = "getting src object"
|
||||
@ -116,7 +518,7 @@ func moveBetween2Storages(t *MoveTask, srcStorage, dstStorage driver.Driver, src
|
||||
|
||||
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(),
|
||||
@ -128,47 +530,50 @@ func moveFileBetween2Storages(tsk *MoveTask, srcStorage, dstStorage driver.Drive
|
||||
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 = "verifying file in destination"
|
||||
|
||||
// check target files
|
||||
dstFilePath := stdpath.Join(dstDirPath, stdpath.Base(srcFilePath))
|
||||
const maxRetries = 3
|
||||
const retryInterval = time.Second
|
||||
var checkErr error
|
||||
for i := 0; i < maxRetries; i++ {
|
||||
_, checkErr = op.Get(tsk.Ctx(), dstStorage, dstFilePath)
|
||||
if checkErr == nil {
|
||||
break
|
||||
}
|
||||
time.Sleep(retryInterval)
|
||||
}
|
||||
if checkErr != nil {
|
||||
return errors.WithMessagef(checkErr, "file not found in destination [%s] after copy", dstFilePath)
|
||||
}
|
||||
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
|
||||
// safeMoveOperation ensures copy-then-delete sequence for safe move operations
|
||||
func (t *MoveTask) safeMoveOperation(srcObj model.Obj) error {
|
||||
if srcObj.IsDir() {
|
||||
// For directories, use the original logic but ensure proper sequencing
|
||||
return moveBetween2Storages(t, t.srcStorage, t.dstStorage, t.SrcObjPath, t.DstDirPath)
|
||||
} else {
|
||||
// For files, use the safe file move logic
|
||||
return moveFileBetween2Storages(t, t.srcStorage, t.dstStorage, t.SrcObjPath, t.DstDirPath)
|
||||
}
|
||||
}
|
||||
|
||||
func _move(ctx context.Context, srcObjPath, dstDirPath string, lazyCache ...bool) (task.TaskExtensionInfo, error) {
|
||||
return _moveWithValidation(ctx, srcObjPath, dstDirPath, false, lazyCache...)
|
||||
}
|
||||
|
||||
func _moveWithValidation(ctx context.Context, srcObjPath, dstDirPath string, validateExistence bool, lazyCache ...bool) (task.TaskExtensionInfo, error) {
|
||||
srcStorage, srcObjActualPath, err := op.GetStorageAndActualPath(srcObjPath)
|
||||
if err != nil {
|
||||
return nil, errors.WithMessage(err, "failed get src storage")
|
||||
@ -178,6 +583,7 @@ func _move(ctx context.Context, srcObjPath, dstDirPath string, lazyCache ...bool
|
||||
return nil, errors.WithMessage(err, "failed get dst storage")
|
||||
}
|
||||
|
||||
// Try native move first if in the same 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) {
|
||||
@ -186,17 +592,23 @@ func _move(ctx context.Context, srcObjPath, dstDirPath string, lazyCache ...bool
|
||||
}
|
||||
|
||||
taskCreator, _ := ctx.Value("user").(*model.User)
|
||||
|
||||
// Create task immediately without any synchronous checks to avoid blocking frontend
|
||||
// All validation and type checking will be done asynchronously in the Run method
|
||||
t := &MoveTask{
|
||||
TaskExtension: task.TaskExtension{
|
||||
Creator: taskCreator,
|
||||
},
|
||||
srcStorage: srcStorage,
|
||||
dstStorage: dstStorage,
|
||||
SrcObjPath: srcObjActualPath,
|
||||
DstDirPath: dstDirActualPath,
|
||||
SrcStorageMp: srcStorage.GetStorage().MountPath,
|
||||
DstStorageMp: dstStorage.GetStorage().MountPath,
|
||||
srcStorage: srcStorage,
|
||||
dstStorage: dstStorage,
|
||||
SrcObjPath: srcObjActualPath,
|
||||
DstDirPath: dstDirActualPath,
|
||||
SrcStorageMp: srcStorage.GetStorage().MountPath,
|
||||
DstStorageMp: dstStorage.GetStorage().MountPath,
|
||||
ValidateExistence: validateExistence,
|
||||
Phase: "initializing",
|
||||
}
|
||||
|
||||
MoveTaskManager.Add(t)
|
||||
return t, nil
|
||||
}
|
||||
}
|
@ -88,17 +88,12 @@ func FsMove(c *gin.Context) {
|
||||
common.ErrorResp(c, err, 403)
|
||||
return
|
||||
}
|
||||
if !req.Overwrite {
|
||||
for _, name := range req.Names {
|
||||
if res, _ := fs.Get(c, stdpath.Join(dstDir, name), &fs.GetArgs{NoLog: true}); res != nil {
|
||||
common.ErrorStrResp(c, fmt.Sprintf("file [%s] exists", name), 403)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Create all tasks immediately without any synchronous validation
|
||||
// All validation will be done asynchronously in the background
|
||||
var addedTasks []task.TaskExtensionInfo
|
||||
for i, name := range req.Names {
|
||||
t, err := fs.MoveWithTask(c, stdpath.Join(srcDir, name), dstDir, len(req.Names) > i+1)
|
||||
t, err := fs.MoveWithTaskAndValidation(c, stdpath.Join(srcDir, name), dstDir, !req.Overwrite, len(req.Names) > i+1)
|
||||
if t != nil {
|
||||
addedTasks = append(addedTasks, t)
|
||||
}
|
||||
@ -107,12 +102,17 @@ func FsMove(c *gin.Context) {
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// Return immediately with task information
|
||||
if len(addedTasks) > 0 {
|
||||
common.SuccessResp(c, gin.H{
|
||||
"message": fmt.Sprintf("Successfully created %d move task(s)", len(addedTasks)),
|
||||
"tasks": getTaskInfos(addedTasks),
|
||||
})
|
||||
} else {
|
||||
common.SuccessResp(c)
|
||||
common.SuccessResp(c, gin.H{
|
||||
"message": "Move operations completed immediately",
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@ -141,14 +141,9 @@ func FsCopy(c *gin.Context) {
|
||||
common.ErrorResp(c, err, 403)
|
||||
return
|
||||
}
|
||||
if !req.Overwrite {
|
||||
for _, name := range req.Names {
|
||||
if res, _ := fs.Get(c, stdpath.Join(dstDir, name), &fs.GetArgs{NoLog: true}); res != nil {
|
||||
common.ErrorStrResp(c, fmt.Sprintf("file [%s] exists", name), 403)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Create all tasks immediately without any synchronous validation
|
||||
// All validation will be done asynchronously in the background
|
||||
var addedTasks []task.TaskExtensionInfo
|
||||
for i, name := range req.Names {
|
||||
t, err := fs.Copy(c, stdpath.Join(srcDir, name), dstDir, len(req.Names) > i+1)
|
||||
@ -160,9 +155,18 @@ func FsCopy(c *gin.Context) {
|
||||
return
|
||||
}
|
||||
}
|
||||
common.SuccessResp(c, gin.H{
|
||||
"tasks": getTaskInfos(addedTasks),
|
||||
})
|
||||
|
||||
// Return immediately with task information
|
||||
if len(addedTasks) > 0 {
|
||||
common.SuccessResp(c, gin.H{
|
||||
"message": fmt.Sprintf("Successfully created %d copy task(s)", len(addedTasks)),
|
||||
"tasks": getTaskInfos(addedTasks),
|
||||
})
|
||||
} else {
|
||||
common.SuccessResp(c, gin.H{
|
||||
"message": "Copy operations completed immediately",
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
type RenameReq struct {
|
||||
|
Reference in New Issue
Block a user